refactor: migrate thread history from JSONL to ocas variable store (Phase 4c)
CI / check (pull_request) Failing after 10m39s

This commit is contained in:
2026-06-02 22:53:24 +08:00
parent e881e0772b
commit 3e12e6ebc0
10 changed files with 252 additions and 169 deletions
@@ -7,7 +7,7 @@ import type { CasRef, ThreadId } from "@united-workforce/protocol";
import { createMarker, deleteMarker } from "../background/index.js"; import { createMarker, deleteMarker } from "../background/index.js";
import { cmdThreadList, cmdThreadShow, cmdThreadStart } from "../commands/thread.js"; import { cmdThreadList, cmdThreadShow, cmdThreadStart } from "../commands/thread.js";
import { import {
appendThreadHistory, addHistoryEntry,
createUwfStore, createUwfStore,
deleteThread, deleteThread,
loadAllThreads, loadAllThreads,
@@ -288,7 +288,7 @@ describe("currentRole field", () => {
const uwfForIndex = await createUwfStore(storageRoot); const uwfForIndex = await createUwfStore(storageRoot);
const head = loadAllThreads(uwfForIndex.varStore)[tid]!.head; const head = loadAllThreads(uwfForIndex.varStore)[tid]!.head;
deleteThread(uwfForIndex.varStore, tid); deleteThread(uwfForIndex.varStore, tid);
await appendThreadHistory(storageRoot, { addHistoryEntry(uwfForIndex.varStore, {
thread: tid, thread: tid,
workflow, workflow,
head, head,
@@ -316,7 +316,7 @@ describe("currentRole field", () => {
const uwfForIndex = await createUwfStore(storageRoot); const uwfForIndex = await createUwfStore(storageRoot);
const head = loadAllThreads(uwfForIndex.varStore)[tid]!.head; const head = loadAllThreads(uwfForIndex.varStore)[tid]!.head;
deleteThread(uwfForIndex.varStore, tid); deleteThread(uwfForIndex.varStore, tid);
await appendThreadHistory(storageRoot, { addHistoryEntry(uwfForIndex.varStore, {
thread: tid, thread: tid,
workflow, workflow,
head, head,
@@ -377,7 +377,7 @@ describe("currentRole field", () => {
const uwfForIndex = await createUwfStore(storageRoot); const uwfForIndex = await createUwfStore(storageRoot);
const compHead = loadAllThreads(uwfForIndex.varStore)[compId]!.head; const compHead = loadAllThreads(uwfForIndex.varStore)[compId]!.head;
deleteThread(uwfForIndex.varStore, compId); deleteThread(uwfForIndex.varStore, compId);
await appendThreadHistory(storageRoot, { addHistoryEntry(uwfForIndex.varStore, {
thread: compId, thread: compId,
workflow: comp.workflow, workflow: comp.workflow,
head: compHead, head: compHead,
@@ -1,15 +1,18 @@
import { afterEach, beforeEach, describe, expect, test } from "bun:test"; import { afterEach, beforeEach, describe, expect, test } from "bun:test";
import { mkdtemp, rm } from "node:fs/promises"; import { mkdir, mkdtemp, rm } from "node:fs/promises";
import { tmpdir } from "node:os"; import { tmpdir } from "node:os";
import { join } from "node:path"; import { join } from "node:path";
import { type CasRef, createThreadIndexEntry, type ThreadId } from "@united-workforce/protocol"; import { type CasRef, createThreadIndexEntry, type ThreadId } from "@united-workforce/protocol";
import { resolveHeadHash } from "../commands/shared.js"; import { resolveHeadHash } from "../commands/shared.js";
import { appendThreadHistory } from "../store.js"; import { addHistoryEntry, createUwfStore, setThread } from "../store.js";
let tmpDir: string; let tmpDir: string;
beforeEach(async () => { beforeEach(async () => {
tmpDir = await mkdtemp(join(tmpdir(), "cli-uwf-resolve-head-")); tmpDir = await mkdtemp(join(tmpdir(), "cli-uwf-resolve-head-"));
const casDir = join(tmpDir, "cas");
await mkdir(casDir, { recursive: true });
process.env.UNCAGED_CAS_DIR = casDir;
}); });
afterEach(async () => { afterEach(async () => {
@@ -19,7 +22,6 @@ afterEach(async () => {
describe("resolveHeadHash", () => { describe("resolveHeadHash", () => {
test("returns head hash from variable store for active thread", async () => { test("returns head hash from variable store for active thread", async () => {
const threadId = "01JTEST0000000000000000001" as ThreadId; const threadId = "01JTEST0000000000000000001" as ThreadId;
const { createUwfStore, setThread } = await import("../store.js");
const uwf = await createUwfStore(tmpDir); const uwf = await createUwfStore(tmpDir);
const headHash = (await uwf.store.put(uwf.schemas.text, "active")) as CasRef; const headHash = (await uwf.store.put(uwf.schemas.text, "active")) as CasRef;
setThread(uwf.varStore, threadId, createThreadIndexEntry(headHash as CasRef)); setThread(uwf.varStore, threadId, createThreadIndexEntry(headHash as CasRef));
@@ -29,13 +31,13 @@ describe("resolveHeadHash", () => {
expect(result).toBe(headHash); expect(result).toBe(headHash);
}); });
test("falls back to history.jsonl when thread not in threads.yaml", async () => { test("falls back to history variable when thread not in active index", async () => {
const threadId = "01JTEST0000000000000000002" as ThreadId; const threadId = "01JTEST0000000000000000002" as ThreadId;
const headHash = "completed_hash_456" as CasRef;
const workflowHash = "workflow_hash_789" as CasRef; const workflowHash = "workflow_hash_789" as CasRef;
// No entry in threads.yaml, only in history.jsonl const uwf = await createUwfStore(tmpDir);
await appendThreadHistory(tmpDir, { const headHash = (await uwf.store.put(uwf.schemas.text, "completed-head")) as CasRef;
addHistoryEntry(uwf.varStore, {
thread: threadId, thread: threadId,
workflow: workflowHash, workflow: workflowHash,
head: headHash, head: headHash,
@@ -54,15 +56,13 @@ describe("resolveHeadHash", () => {
test("prioritizes active thread over history when thread exists in both", async () => { test("prioritizes active thread over history when thread exists in both", async () => {
const threadId = "01JTEST0000000000000000004" as ThreadId; const threadId = "01JTEST0000000000000000004" as ThreadId;
const historicalHash = "historical_hash_v1" as CasRef;
const workflowHash = "workflow_hash_xyz" as CasRef; const workflowHash = "workflow_hash_xyz" as CasRef;
const { createUwfStore, setThread } = await import("../store.js");
const { createThreadIndexEntry } = await import("@united-workforce/protocol");
const uwf = await createUwfStore(tmpDir); const uwf = await createUwfStore(tmpDir);
const activeHead = (await uwf.store.put(uwf.schemas.text, "active-v2")) as CasRef; const activeHead = (await uwf.store.put(uwf.schemas.text, "active-v2")) as CasRef;
const historicalHash = (await uwf.store.put(uwf.schemas.text, "historical-v1")) as CasRef;
setThread(uwf.varStore, threadId, createThreadIndexEntry(activeHead)); setThread(uwf.varStore, threadId, createThreadIndexEntry(activeHead));
await appendThreadHistory(tmpDir, { addHistoryEntry(uwf.varStore, {
thread: threadId, thread: threadId,
workflow: workflowHash, workflow: workflowHash,
head: historicalHash, head: historicalHash,
@@ -80,25 +80,26 @@ describe("resolveHeadHash", () => {
const threadId1 = "01JTEST0000000000000000005" as ThreadId; const threadId1 = "01JTEST0000000000000000005" as ThreadId;
const threadId2 = "01JTEST0000000000000000006" as ThreadId; const threadId2 = "01JTEST0000000000000000006" as ThreadId;
const threadId3 = "01JTEST0000000000000000007" as ThreadId; const threadId3 = "01JTEST0000000000000000007" as ThreadId;
const hash1 = "hash_thread1" as CasRef;
const hash2 = "hash_thread2" as CasRef;
const hash3 = "hash_thread3" as CasRef;
const workflowHash = "workflow_hash_abc" as CasRef; const workflowHash = "workflow_hash_abc" as CasRef;
await appendThreadHistory(tmpDir, { const uwf = await createUwfStore(tmpDir);
const hash1 = (await uwf.store.put(uwf.schemas.text, "hash-thread1")) as CasRef;
const hash2 = (await uwf.store.put(uwf.schemas.text, "hash-thread2")) as CasRef;
const hash3 = (await uwf.store.put(uwf.schemas.text, "hash-thread3")) as CasRef;
addHistoryEntry(uwf.varStore, {
thread: threadId1, thread: threadId1,
workflow: workflowHash, workflow: workflowHash,
head: hash1, head: hash1,
completedAt: Date.now() - 2000, completedAt: Date.now() - 2000,
reason: null, reason: null,
}); });
await appendThreadHistory(tmpDir, { addHistoryEntry(uwf.varStore, {
thread: threadId2, thread: threadId2,
workflow: workflowHash, workflow: workflowHash,
head: hash2, head: hash2,
completedAt: Date.now() - 1000, completedAt: Date.now() - 1000,
reason: null, reason: null,
}); });
await appendThreadHistory(tmpDir, { addHistoryEntry(uwf.varStore, {
thread: threadId3, thread: threadId3,
workflow: workflowHash, workflow: workflowHash,
head: hash3, head: hash3,
@@ -238,35 +238,78 @@ describe("Global CAS directory", () => {
await expect(readFile(threadsPath, "utf8")).rejects.toThrow(); await expect(readFile(threadsPath, "utf8")).rejects.toThrow();
}); });
test("history remains in storageRoot", async () => { test("history is stored in global CAS variable store", async () => {
const globalCasDir = join(tmpDir, "global-cas"); const globalCasDir = join(tmpDir, "global-cas");
process.env.UNCAGED_CAS_DIR = globalCasDir; process.env.UNCAGED_CAS_DIR = globalCasDir;
const storageRoot = join(tmpDir, "storage"); const storageRoot = join(tmpDir, "storage");
await mkdir(storageRoot, { recursive: true }); await mkdir(storageRoot, { recursive: true });
await createUwfStore(storageRoot); const uwf = await createUwfStore(storageRoot);
const threadId = "thread-123" as ThreadId;
// Write history const headHash = await uwf.store.put(uwf.schemas.text, "history-head");
const { appendThreadHistory } = await import("../store.js"); const { addHistoryEntry, findHistoryEntry } = await import("../store.js");
await appendThreadHistory(storageRoot, { addHistoryEntry(uwf.varStore, {
thread: "thread-123" as any, thread: threadId,
workflow: "workflow-456", workflow: "workflow-456",
head: "hash-789", head: headHash,
completedAt: Date.now(), completedAt: Date.now(),
reason: "completed", reason: "completed",
}); });
// Verify history.jsonl is in storageRoot, not global CAS const entry = findHistoryEntry(uwf.varStore, threadId);
const { readFile } = await import("node:fs/promises"); expect(entry?.thread).toBe(threadId);
const historyPath = join(storageRoot, "history.jsonl"); expect(entry?.workflow).toBe("workflow-456");
const content = await readFile(historyPath, "utf8"); expect(entry?.head).toBe(headHash);
expect(content).toContain("thread-123");
expect(content).toContain("workflow-456");
// Verify history.jsonl is NOT in global CAS directory const { access } = await import("node:fs/promises");
const globalHistoryPath = join(globalCasDir, "history.jsonl"); await access(join(globalCasDir, "variables.db"));
await expect(readFile(globalHistoryPath, "utf8")).rejects.toThrow();
const historyPath = join(storageRoot, "history.jsonl");
await expect(access(historyPath)).rejects.toThrow();
});
test("migrates history.jsonl to variable store and renames file", async () => {
const globalCasDir = join(tmpDir, "global-cas-history");
process.env.UNCAGED_CAS_DIR = globalCasDir;
const storageRoot = join(tmpDir, "storage-history-migrate");
await mkdir(storageRoot, { recursive: true });
const threadId = "01JTEST0000000000000000CD" as ThreadId;
const uwfSeed = await createUwfStore(storageRoot);
const workflowHash = await uwfSeed.store.put(uwfSeed.schemas.text, "migrated-workflow");
const headHash = await uwfSeed.store.put(uwfSeed.schemas.text, "migrated-head");
const completedAt = 1780410000000;
const { writeFile, access, readFile } = await import("node:fs/promises");
const historyPath = join(storageRoot, "history.jsonl");
await writeFile(
historyPath,
`${JSON.stringify({
thread: threadId,
workflow: workflowHash,
head: headHash,
completedAt,
reason: "cancelled",
})}\n`,
"utf8",
);
const uwf = await createUwfStore(storageRoot);
const { findHistoryEntry } = await import("../store.js");
const entry = findHistoryEntry(uwf.varStore, threadId);
expect(entry).toEqual({
thread: threadId,
workflow: workflowHash,
head: headHash,
completedAt,
reason: "cancelled",
});
await expect(access(historyPath)).rejects.toThrow();
const migratedContent = await readFile(`${historyPath}.migrated`, "utf8");
expect(migratedContent).toContain(threadId);
expect(migratedContent).toContain(workflowHash);
}); });
test("CAS nodes are stored in global directory", async () => { test("CAS nodes are stored in global directory", async () => {
@@ -1,24 +1,40 @@
import { describe, expect, test } from "bun:test"; import { describe, expect, test } from "bun:test";
import { mkdtemp } from "node:fs/promises"; import { mkdir, mkdtemp } from "node:fs/promises";
import { tmpdir } from "node:os"; import { tmpdir } from "node:os";
import { join } from "node:path"; import { join } from "node:path";
import type { CasRef, ThreadId } from "@united-workforce/protocol"; import type { CasRef, ThreadId } from "@united-workforce/protocol";
import { appendThreadHistory, loadThreadHistory } from "../store.js"; import { addHistoryEntry, createUwfStore, loadAllHistory } from "../store.js";
async function makeUwfStore(storageRoot: string) {
const casDir = join(storageRoot, "cas");
await mkdir(casDir, { recursive: true });
process.env.UNCAGED_CAS_DIR = casDir;
return createUwfStore(storageRoot);
}
async function seedHistoryHead(
uwf: Awaited<ReturnType<typeof createUwfStore>>,
label: string,
): Promise<CasRef> {
return (await uwf.store.put(uwf.schemas.text, label)) as CasRef;
}
describe("thread cancel status", () => { describe("thread cancel status", () => {
test("cancelled history entry has reason 'cancelled'", async () => { test("cancelled history entry has reason 'cancelled'", async () => {
const tmpDir = await mkdtemp(join(tmpdir(), "uwf-cancel-test-")); const tmpDir = await mkdtemp(join(tmpdir(), "uwf-cancel-test-"));
const threadId = "01JTEST000000000000CANCEL1" as ThreadId; const threadId = "01JTEST000000000000CANCEL1" as ThreadId;
const uwf = await makeUwfStore(tmpDir);
const head = await seedHistoryHead(uwf, "cancelled-head");
await appendThreadHistory(tmpDir, { addHistoryEntry(uwf.varStore, {
thread: threadId, thread: threadId,
workflow: "test-workflow", workflow: "test-workflow",
head: "test-head-hash" as CasRef, head,
completedAt: Date.now(), completedAt: Date.now(),
reason: "cancelled", reason: "cancelled",
}); });
const history = await loadThreadHistory(tmpDir); const history = loadAllHistory(uwf.varStore);
expect(history).toHaveLength(1); expect(history).toHaveLength(1);
expect(history[0]?.reason).toBe("cancelled"); expect(history[0]?.reason).toBe("cancelled");
}); });
@@ -26,60 +42,66 @@ describe("thread cancel status", () => {
test("completed history entry has reason 'completed'", async () => { test("completed history entry has reason 'completed'", async () => {
const tmpDir = await mkdtemp(join(tmpdir(), "uwf-cancel-test-")); const tmpDir = await mkdtemp(join(tmpdir(), "uwf-cancel-test-"));
const threadId = "01JTEST000000000000CANCEL2" as ThreadId; const threadId = "01JTEST000000000000CANCEL2" as ThreadId;
const uwf = await makeUwfStore(tmpDir);
const head = await seedHistoryHead(uwf, "completed-head");
await appendThreadHistory(tmpDir, { addHistoryEntry(uwf.varStore, {
thread: threadId, thread: threadId,
workflow: "test-workflow", workflow: "test-workflow",
head: "test-head-hash" as CasRef, head,
completedAt: Date.now(), completedAt: Date.now(),
reason: "completed", reason: "completed",
}); });
const history = await loadThreadHistory(tmpDir); const history = loadAllHistory(uwf.varStore);
expect(history).toHaveLength(1); expect(history).toHaveLength(1);
expect(history[0]?.reason).toBe("completed"); expect(history[0]?.reason).toBe("completed");
}); });
test("legacy history entry without reason parses as null", async () => { test("history entry with null reason is stored as completed", async () => {
const tmpDir = await mkdtemp(join(tmpdir(), "uwf-cancel-test-")); const tmpDir = await mkdtemp(join(tmpdir(), "uwf-cancel-test-"));
const threadId = "01JTEST000000000000CANCEL3" as ThreadId; const threadId = "01JTEST000000000000CANCEL3" as ThreadId;
const uwf = await makeUwfStore(tmpDir);
const head = await seedHistoryHead(uwf, "legacy-head");
// Simulate legacy entry without reason field addHistoryEntry(uwf.varStore, {
await appendThreadHistory(tmpDir, {
thread: threadId, thread: threadId,
workflow: "test-workflow", workflow: "test-workflow",
head: "test-head-hash" as CasRef, head,
completedAt: Date.now(), completedAt: Date.now(),
reason: null, reason: null,
}); });
const history = await loadThreadHistory(tmpDir); const history = loadAllHistory(uwf.varStore);
expect(history).toHaveLength(1); expect(history).toHaveLength(1);
expect(history[0]?.reason).toBeNull(); expect(history[0]?.reason).toBe("completed");
}); });
test("mixed completed and cancelled entries preserve distinct reasons", async () => { test("mixed completed and cancelled entries preserve distinct reasons", async () => {
const tmpDir = await mkdtemp(join(tmpdir(), "uwf-cancel-test-")); const tmpDir = await mkdtemp(join(tmpdir(), "uwf-cancel-test-"));
const uwf = await makeUwfStore(tmpDir);
const head1 = await seedHistoryHead(uwf, "head1");
const head2 = await seedHistoryHead(uwf, "head2");
await appendThreadHistory(tmpDir, { addHistoryEntry(uwf.varStore, {
thread: "01JTEST000000000000CANCEL4" as ThreadId, thread: "01JTEST000000000000CANCEL4" as ThreadId,
workflow: "test-workflow", workflow: "test-workflow",
head: "head1" as CasRef, head: head1,
completedAt: Date.now(), completedAt: Date.now(),
reason: "completed", reason: "completed",
}); });
await appendThreadHistory(tmpDir, { addHistoryEntry(uwf.varStore, {
thread: "01JTEST000000000000CANCEL5" as ThreadId, thread: "01JTEST000000000000CANCEL5" as ThreadId,
workflow: "test-workflow", workflow: "test-workflow",
head: "head2" as CasRef, head: head2,
completedAt: Date.now(), completedAt: Date.now(),
reason: "cancelled", reason: "cancelled",
}); });
const history = await loadThreadHistory(tmpDir); const history = loadAllHistory(uwf.varStore);
expect(history).toHaveLength(2); expect(history).toHaveLength(2);
expect(history[0]?.reason).toBe("completed"); const reasons = history.map((entry) => entry.reason).sort();
expect(history[1]?.reason).toBe("cancelled"); expect(reasons).toEqual(["cancelled", "completed"]);
}); });
}); });
@@ -10,7 +10,7 @@ import { cmdThreadList } from "../commands/thread.js";
import { parseTimeInput } from "../commands/thread-time-parser.js"; import { parseTimeInput } from "../commands/thread-time-parser.js";
import type { UwfStore } from "../store.js"; import type { UwfStore } from "../store.js";
import { import {
appendThreadHistory, addHistoryEntry,
createUwfStore, createUwfStore,
deleteThread, deleteThread,
loadAllThreads, loadAllThreads,
@@ -78,7 +78,7 @@ async function completeThread(
) { ) {
const uwfIdx = await createUwfStore(storageRoot); const uwfIdx = await createUwfStore(storageRoot);
deleteThread(uwfIdx.varStore, threadId); deleteThread(uwfIdx.varStore, threadId);
await appendThreadHistory(storageRoot, { addHistoryEntry(uwfIdx.varStore, {
thread: threadId, thread: threadId,
workflow: workflowHash, workflow: workflowHash,
head: headHash, head: headHash,
@@ -7,7 +7,7 @@ import type { CasRef, ThreadId } from "@united-workforce/protocol";
import { createMarker, deleteMarker } from "../background/index.js"; import { createMarker, deleteMarker } from "../background/index.js";
import { cmdThreadShow, cmdThreadStart } from "../commands/thread.js"; import { cmdThreadShow, cmdThreadStart } from "../commands/thread.js";
import { import {
appendThreadHistory, addHistoryEntry,
createUwfStore, createUwfStore,
deleteThread, deleteThread,
loadAllThreads, loadAllThreads,
@@ -210,7 +210,7 @@ describe("thread show status field", () => {
deleteThread(uwfForIndex.varStore, threadId); deleteThread(uwfForIndex.varStore, threadId);
await appendThreadHistory(storageRoot, { addHistoryEntry(uwfForIndex.varStore, {
thread: threadId, thread: threadId,
workflow, workflow,
head, head,
@@ -247,7 +247,7 @@ describe("thread show status field", () => {
deleteThread(uwfForIndex.varStore, threadId); deleteThread(uwfForIndex.varStore, threadId);
await appendThreadHistory(storageRoot, { addHistoryEntry(uwfForIndex.varStore, {
thread: threadId, thread: threadId,
workflow, workflow,
head, head,
@@ -284,7 +284,7 @@ describe("thread show status field", () => {
deleteThread(uwfForIndex.varStore, threadId); deleteThread(uwfForIndex.varStore, threadId);
await appendThreadHistory(storageRoot, { addHistoryEntry(uwfForIndex.varStore, {
thread: threadId, thread: threadId,
workflow, workflow,
head, head,
@@ -12,7 +12,7 @@ import {
THREAD_READ_DEFAULT_QUOTA, THREAD_READ_DEFAULT_QUOTA,
} from "../commands/thread.js"; } from "../commands/thread.js";
import type { UwfStore } from "../store.js"; import type { UwfStore } from "../store.js";
import { appendThreadHistory, createUwfStore } from "../store.js"; import { addHistoryEntry, createUwfStore } from "../store.js";
import { seedThreads } from "./thread-test-helpers.js"; import { seedThreads } from "./thread-test-helpers.js";
// ── schemas used in tests ──────────────────────────────────────────────────── // ── schemas used in tests ────────────────────────────────────────────────────
@@ -744,9 +744,9 @@ describe("cmdStepList with completed threads", () => {
}); });
const threadId = "01JTEST0000000000000000A2" as ThreadId; const threadId = "01JTEST0000000000000000A2" as ThreadId;
// Thread is NOT in threads.yaml (simulating completed thread) // Thread is NOT in active index (simulating completed thread)
// But it IS in history.jsonl // But it IS in history variable store
await appendThreadHistory(tmpDir, { addHistoryEntry(uwf.varStore, {
thread: threadId, thread: threadId,
workflow: workflowHash, workflow: workflowHash,
head: step2Hash, head: step2Hash,
@@ -872,9 +872,9 @@ describe("cmdStepShow with completed threads", () => {
}); });
const threadId = "01JTEST0000000000000000B2" as ThreadId; const threadId = "01JTEST0000000000000000B2" as ThreadId;
// Thread is NOT in threads.yaml // Thread is NOT in active index
// But it IS in history.jsonl // But it IS in history variable store
await appendThreadHistory(tmpDir, { addHistoryEntry(uwf.varStore, {
thread: threadId, thread: threadId,
workflow: workflowHash, workflow: workflowHash,
head: stepHash, head: stepHash,
@@ -935,9 +935,9 @@ describe("cmdThreadRead with completed threads", () => {
}); });
const threadId = "01JTEST0000000000000000C1" as ThreadId; const threadId = "01JTEST0000000000000000C1" as ThreadId;
// Thread is NOT in threads.yaml // Thread is NOT in active index
// But it IS in history.jsonl // But it IS in history variable store
await appendThreadHistory(tmpDir, { addHistoryEntry(uwf.varStore, {
thread: threadId, thread: threadId,
workflow: workflowHash, workflow: workflowHash,
head: stepHash, head: stepHash,
@@ -999,7 +999,7 @@ describe("cmdThreadRead with completed threads", () => {
}); });
const threadId = "01JTEST0000000000000000C2" as ThreadId; const threadId = "01JTEST0000000000000000C2" as ThreadId;
await appendThreadHistory(tmpDir, { addHistoryEntry(uwf.varStore, {
thread: threadId, thread: threadId,
workflow: workflowHash, workflow: workflowHash,
head: step3Hash, head: step3Hash,
+2 -2
View File
@@ -6,7 +6,7 @@ import type {
StepNodePayload, StepNodePayload,
ThreadId, ThreadId,
} from "@united-workforce/protocol"; } from "@united-workforce/protocol";
import { createUwfStore, findThreadInHistory, getThread, type UwfStore } from "../store.js"; import { createUwfStore, findHistoryEntry, getThread, type UwfStore } from "../store.js";
type ChainState = { type ChainState = {
startHash: CasRef; startHash: CasRef;
@@ -207,7 +207,7 @@ async function resolveHeadHash(storageRoot: string, threadId: ThreadId): Promise
if (entry !== null) { if (entry !== null) {
return entry.head; return entry.head;
} }
const hist = await findThreadInHistory(storageRoot, threadId); const hist = findHistoryEntry(uwf.varStore, threadId);
if (hist !== null) { if (hist !== null) {
return hist.head; return hist.head;
} }
+15 -21
View File
@@ -38,13 +38,13 @@ import { createMarker, deleteMarker, isThreadRunning } from "../background/index
import { createIncludeTag } from "../include.js"; import { createIncludeTag } from "../include.js";
import { evaluate, isSuspendResult } from "../moderator/index.js"; import { evaluate, isSuspendResult } from "../moderator/index.js";
import { import {
appendThreadHistory, addHistoryEntry,
createUwfStore, createUwfStore,
deleteThread, deleteThread,
findThreadInHistory, findHistoryEntry,
getThread, getThread,
loadAllHistory,
loadAllThreads, loadAllThreads,
loadThreadHistory,
loadWorkflowRegistry, loadWorkflowRegistry,
resolveWorkflowHash, resolveWorkflowHash,
setThread, setThread,
@@ -521,7 +521,7 @@ export async function cmdThreadShow(
}; };
} }
const hist = await findThreadInHistory(storageRoot, threadId); const hist = findHistoryEntry(uwf.varStore, threadId);
if (hist !== null) { if (hist !== null) {
const status: ThreadStatus = hist.reason === "cancelled" ? "cancelled" : "completed"; const status: ThreadStatus = hist.reason === "cancelled" ? "cancelled" : "completed";
@@ -593,12 +593,12 @@ async function collectActiveThreads(
return items; return items;
} }
async function collectCompletedThreads( function collectCompletedThreads(
storageRoot: string, varStore: VariableStore,
activeIds: Set<ThreadId>, activeIds: Set<ThreadId>,
): Promise<ThreadListItemWithStatus[]> { ): ThreadListItemWithStatus[] {
const items: ThreadListItemWithStatus[] = []; const items: ThreadListItemWithStatus[] = [];
const history = await loadThreadHistory(storageRoot); const history = loadAllHistory(varStore);
const seen = new Set<ThreadId>(); // Deduplication (issue #470) const seen = new Set<ThreadId>(); // Deduplication (issue #470)
for (const entry of history) { for (const entry of history) {
if (!activeIds.has(entry.thread) && !seen.has(entry.thread)) { if (!activeIds.has(entry.thread) && !seen.has(entry.thread)) {
@@ -671,7 +671,7 @@ export async function cmdThreadList(
statusFilter.includes("cancelled"); statusFilter.includes("cancelled");
if (includeCompleted) { if (includeCompleted) {
const activeIds = new Set(items.map((i) => i.thread)); const activeIds = new Set(items.map((i) => i.thread));
const completedItems = await collectCompletedThreads(storageRoot, activeIds); const completedItems = collectCompletedThreads(uwf.varStore, activeIds);
items = items.concat(completedItems); items = items.concat(completedItems);
} }
@@ -1035,15 +1035,9 @@ function spawnAgent(
return obj as unknown as AdapterOutput; return obj as unknown as AdapterOutput;
} }
async function archiveThread( function archiveThread(uwf: UwfStore, threadId: ThreadId, workflow: CasRef, head: CasRef): void {
uwf: UwfStore,
storageRoot: string,
threadId: ThreadId,
workflow: CasRef,
head: CasRef,
): Promise<void> {
deleteThread(uwf.varStore, threadId); deleteThread(uwf.varStore, threadId);
await appendThreadHistory(storageRoot, { addHistoryEntry(uwf.varStore, {
thread: threadId, thread: threadId,
workflow, workflow,
head, head,
@@ -1302,7 +1296,7 @@ async function resolveModeratorStepTarget(
if (nextResult.value.role === END_ROLE) { if (nextResult.value.role === END_ROLE) {
plog.log(PL_THREAD_ARCHIVED, `thread archived head=${headHash}`, null); plog.log(PL_THREAD_ARCHIVED, `thread archived head=${headHash}`, null);
await archiveThread(uwf, storageRoot, threadId, workflowHash, headHash); archiveThread(uwf, threadId, workflowHash, headHash);
return { return {
workflow: workflowHash, workflow: workflowHash,
thread: threadId, thread: threadId,
@@ -1368,7 +1362,7 @@ async function finalizeAgentStep(
const done = afterResult.value.role === END_ROLE; const done = afterResult.value.role === END_ROLE;
if (done) { if (done) {
plog.log(PL_THREAD_ARCHIVED, `thread archived head=${newHead}`, null); plog.log(PL_THREAD_ARCHIVED, `thread archived head=${newHead}`, null);
await archiveThread(uwfAfter, storageRoot, threadId, workflowHash, newHead); archiveThread(uwfAfter, threadId, workflowHash, newHead);
} }
const status: ThreadStatus = done ? "completed" : "idle"; const status: ThreadStatus = done ? "completed" : "idle";
@@ -1456,7 +1450,7 @@ async function resolveHeadHash(storageRoot: string, threadId: ThreadId): Promise
if (entry !== null) { if (entry !== null) {
return entry.head; return entry.head;
} }
const hist = await findThreadInHistory(storageRoot, threadId); const hist = findHistoryEntry(uwf.varStore, threadId);
if (hist !== null) { if (hist !== null) {
return hist.head; return hist.head;
} }
@@ -1566,7 +1560,7 @@ export async function cmdThreadCancel(
completedAt: Date.now(), completedAt: Date.now(),
reason: "cancelled", reason: "cancelled",
}; };
await appendThreadHistory(storageRoot, historyEntry); addHistoryEntry(uwf.varStore, historyEntry);
return { thread: threadId, cancelled: true }; return { thread: threadId, cancelled: true };
} }
+70 -47
View File
@@ -1,6 +1,6 @@
import type { Dirent } from "node:fs"; import type { Dirent } from "node:fs";
import { existsSync, symlinkSync } from "node:fs"; import { existsSync, symlinkSync } from "node:fs";
import { access, appendFile, mkdir, readdir, readFile, rename } from "node:fs/promises"; import { access, mkdir, readdir, readFile, rename } from "node:fs/promises";
import { homedir } from "node:os"; import { homedir } from "node:os";
import { join } from "node:path"; import { join } from "node:path";
@@ -27,6 +27,9 @@ export const REGISTRY_VAR_PREFIX = "@uwf/registry/";
/** Variable name prefix for active thread entries (`@uwf/thread/<thread-id>`). */ /** Variable name prefix for active thread entries (`@uwf/thread/<thread-id>`). */
export const THREAD_VAR_PREFIX = "@uwf/thread/"; export const THREAD_VAR_PREFIX = "@uwf/thread/";
/** Variable name prefix for completed/cancelled thread history (`@uwf/history/<thread-id>`). */
export const HISTORY_VAR_PREFIX = "@uwf/history/";
/** A workflow entry discovered from the project-local .workflows/ directory. */ /** A workflow entry discovered from the project-local .workflows/ directory. */
export type ProjectWorkflowEntry = { export type ProjectWorkflowEntry = {
/** Workflow name (from YAML `name` field, equals filename stem). */ /** Workflow name (from YAML `name` field, equals filename stem). */
@@ -190,10 +193,6 @@ export function getThreadsPath(storageRoot: string): string {
return join(storageRoot, "threads.yaml"); return join(storageRoot, "threads.yaml");
} }
export function getHistoryPath(storageRoot: string): string {
return join(storageRoot, "history.jsonl");
}
export type ThreadHistoryLine = ThreadListItem & { export type ThreadHistoryLine = ThreadListItem & {
completedAt: number; completedAt: number;
reason: "completed" | "cancelled" | null; reason: "completed" | "cancelled" | null;
@@ -214,6 +213,7 @@ export async function createUwfStore(storageRoot: string): Promise<UwfStore> {
const varStore = createVariableStore(join(casDir, "variables.db"), store); const varStore = createVariableStore(join(casDir, "variables.db"), store);
await migrateWorkflowRegistryIfNeeded(storageRoot, varStore); await migrateWorkflowRegistryIfNeeded(storageRoot, varStore);
await migrateThreadsIndexIfNeeded(storageRoot, varStore); await migrateThreadsIndexIfNeeded(storageRoot, varStore);
await migrateHistoryIfNeeded(storageRoot, varStore);
return { storageRoot, store, schemas, varStore }; return { storageRoot, store, schemas, varStore };
} }
@@ -384,24 +384,15 @@ export function deleteThread(varStore: VariableStore, threadId: ThreadId): void
varStore.remove(threadVarName(threadId)); varStore.remove(threadVarName(threadId));
} }
export async function loadThreadHistory(storageRoot: string): Promise<ThreadHistoryLine[]> { function parseHistoryJsonlLine(trimmed: string): ThreadHistoryLine | null {
const path = getHistoryPath(storageRoot);
try {
const text = await readFile(path, "utf8");
const lines: ThreadHistoryLine[] = [];
for (const line of text.split("\n")) {
const trimmed = line.trim();
if (trimmed === "") {
continue;
}
let raw: unknown; let raw: unknown;
try { try {
raw = JSON.parse(trimmed) as unknown; raw = JSON.parse(trimmed) as unknown;
} catch { } catch {
continue; return null;
} }
if (raw === null || typeof raw !== "object" || Array.isArray(raw)) { if (raw === null || typeof raw !== "object" || Array.isArray(raw)) {
continue; return null;
} }
const rec = raw as Record<string, unknown>; const rec = raw as Record<string, unknown>;
const thread = rec.thread; const thread = rec.thread;
@@ -416,45 +407,77 @@ export async function loadThreadHistory(storageRoot: string): Promise<ThreadHist
) { ) {
const reason = rec.reason; const reason = rec.reason;
const parsedReason = reason === "completed" || reason === "cancelled" ? reason : null; const parsedReason = reason === "completed" || reason === "cancelled" ? reason : null;
lines.push({ return {
thread: thread as ThreadId, thread: thread as ThreadId,
workflow, workflow,
head, head,
completedAt, completedAt,
reason: parsedReason, reason: parsedReason,
}); };
}
}
return lines;
} catch (e) {
const err = e as NodeJS.ErrnoException;
if (err.code === "ENOENT") {
return [];
}
throw e;
}
}
export async function findThreadInHistory(
storageRoot: string,
threadId: ThreadId,
): Promise<ThreadHistoryLine | null> {
const history = await loadThreadHistory(storageRoot);
for (let i = history.length - 1; i >= 0; i--) {
const entry = history[i];
if (entry !== undefined && entry.thread === threadId) {
return entry;
}
} }
return null; return null;
} }
export async function appendThreadHistory( /** One-time migration: `~/.uwf/history.jsonl` → `@uwf/history/*` variables. */
export async function migrateHistoryIfNeeded(
storageRoot: string, storageRoot: string,
entry: ThreadHistoryLine, varStore: VariableStore,
): Promise<void> { ): Promise<void> {
const path = getHistoryPath(storageRoot); const path = join(storageRoot, "history.jsonl");
await mkdir(storageRoot, { recursive: true }); if (!existsSync(path)) {
const line = `${JSON.stringify(entry)}\n`; return;
await appendFile(path, line, "utf8"); }
const text = await readFile(path, "utf8");
for (const line of text.split("\n")) {
const trimmed = line.trim();
if (trimmed === "") {
continue;
}
const entry = parseHistoryJsonlLine(trimmed);
if (entry !== null) {
addHistoryEntry(varStore, entry);
}
}
await rename(path, `${path}.migrated`);
}
export function loadAllHistory(varStore: VariableStore): ThreadHistoryLine[] {
const vars = varStore.list({ namePrefix: HISTORY_VAR_PREFIX });
return vars.map((v) => ({
thread: v.name.slice(HISTORY_VAR_PREFIX.length) as ThreadId,
workflow: v.tags.workflow ?? "",
head: v.value as CasRef,
completedAt: Number(v.tags.completedAt ?? "0"),
reason: v.tags.reason === "completed" || v.tags.reason === "cancelled" ? v.tags.reason : null,
}));
}
export function findHistoryEntry(
varStore: VariableStore,
threadId: ThreadId,
): ThreadHistoryLine | null {
const vars = varStore.list({ namePrefix: `${HISTORY_VAR_PREFIX}${threadId}` });
const v = vars.find((entry) => entry.name === `${HISTORY_VAR_PREFIX}${threadId}`);
if (v === undefined) {
return null;
}
return {
thread: threadId,
workflow: v.tags.workflow ?? "",
head: v.value as CasRef,
completedAt: Number(v.tags.completedAt ?? "0"),
reason: v.tags.reason === "completed" || v.tags.reason === "cancelled" ? v.tags.reason : null,
};
}
export function addHistoryEntry(varStore: VariableStore, entry: ThreadHistoryLine): void {
varStore.set(`${HISTORY_VAR_PREFIX}${entry.thread}`, entry.head, {
tags: {
workflow: entry.workflow,
completedAt: String(entry.completedAt),
reason: entry.reason ?? "completed",
},
});
} }