Phase 4c: Migrate thread history from JSONL to ocas variable store #18
@@ -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,
|
||||||
|
|||||||
@@ -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;
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -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 };
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -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",
|
||||||
|
},
|
||||||
|
});
|
||||||
}
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user