uncaged/nerve
Archived
10
0
Fork 0
Code Issues 5 Pull Requests Actions Packages Projects Releases Wiki Activity

refactor(store): rename LogEntry/WorkflowRun/ThreadRoundRow ts → timestamp

Browse Source 
- Rename logs & workflow_runs table column ts → timestamp (breaking, no migration)
- Update all SQL, types, mocks, CLI output, and tests
- Integration tests use mkdtempSync to avoid stale DB conflicts

Fixes #113
This commit is contained in:
小橘
2026-04-25 02:20:18 +00:00
parent 4da2c87a77
commit 3dc835e1de
7 changed files with 84 additions and 73 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/cli/src/__tests__/workflow.test.ts
+18 -18
View File
@@ -41,11 +41,11 @@ function upsertRun(
runId: string,
workflow: string,
status: WorkflowRun["status"],
ts: number,
timestampMs: number,
): void {
store.upsertWorkflowRun(
{ source: "workflow", type: status, refId: runId, payload: null, timestamp: ts },
{ runId, workflow, status, ts },
{ source: "workflow", type: status, refId: runId, payload: null, timestamp: timestampMs },
{ runId, workflow, status, timestamp: timestampMs },
);
}
@@ -65,8 +65,8 @@ afterEach(() => {
describe("formatTs", () => {
it("returns ISO 8601 string", () => {
const ts = new Date("2026-01-01T00:00:00.000Z").getTime();
expect(formatTs(ts)).toBe("2026-01-01T00:00:00.000Z");
const timestampMs = new Date("2026-01-01T00:00:00.000Z").getTime();
expect(formatTs(timestampMs)).toBe("2026-01-01T00:00:00.000Z");
});
});
@@ -127,14 +127,14 @@ describe("getAllWorkflowRuns", () => {
}
});
it("sorts by ts descending (newest first)", () => {
it("sorts by timestamp descending (newest first)", () => {
upsertRun("r1", "cleanup", "completed", 1000);
upsertRun("r2", "cleanup", "started", 3000);
upsertRun("r3", "cleanup", "failed", 2000);
const runs = getAllWorkflowRuns(store, null);
expect(runs[0].ts).toBeGreaterThan(runs[1].ts);
expect(runs[1].ts).toBeGreaterThan(runs[2].ts);
expect(runs[0].timestamp).toBeGreaterThan(runs[1].timestamp);
expect(runs[1].timestamp).toBeGreaterThan(runs[2].timestamp);
});
});
@@ -147,9 +147,9 @@ describe("buildListOutput", () => {
runId: string,
workflow: string,
status: WorkflowRun["status"],
ts: number,
timestampMs: number,
): WorkflowRun {
return { runId, workflow, status, ts };
return { runId, workflow, status, timestamp: timestampMs };
}
it("returns empty message when no runs and --all=false", () => {
@@ -235,7 +235,7 @@ describe("buildInspectOutput", () => {
runId: "run-xyz",
workflow: "cleanup",
status: "completed",
ts: 1_700_000_000_000,
timestamp: 1_700_000_000_000,
};
it("shows header with run details", () => {
@@ -251,7 +251,7 @@ describe("buildInspectOutput", () => {
expect(eventLines.join("")).toContain("no events recorded");
});
it("shows event lines with type and ts", () => {
it("shows event lines with type and timestamp", () => {
const logs = [{ timestamp: 1_700_000_001_000, type: "started", payload: null }];
const { eventLines } = buildInspectOutput(baseRun, logs, 0, 20);
const text = eventLines.join("");
@@ -358,7 +358,7 @@ describe("formatThreadRoundBlock", () => {
const row: ThreadRoundRow = {
round: 2,
logId: 99,
ts: new Date("2026-01-02T03:04:05.006Z").getTime(),
timestamp: new Date("2026-01-02T03:04:05.006Z").getTime(),
message: { role: "bot", content: "hi", meta: { score: 0.5 }, timestamp: 1735783445006 },
};
@@ -376,7 +376,7 @@ describe("buildThreadCommandOutput", () => {
return {
round: n,
logId: 10 + n,
ts: 1000 + n,
timestamp: 1000 + n,
message: { role: "r", content, meta: { extra: n }, timestamp: 1000 + n },
};
}
@@ -462,15 +462,15 @@ describe("getAllWorkflowRuns — uses store.getAllWorkflowRuns SQL path", () =>
expect(runs).toHaveLength(7);
});
it("returns runs sorted by ts descending (newest first)", () => {
it("returns runs sorted by timestamp descending (newest first)", () => {
upsertRun("r1", "deploy", "completed", 1000);
upsertRun("r2", "deploy", "completed", 3000);
upsertRun("r3", "deploy", "completed", 2000);
const runs = getAllWorkflowRuns(store, null);
expect(runs[0].ts).toBe(3000);
expect(runs[1].ts).toBe(2000);
expect(runs[2].ts).toBe(1000);
expect(runs[0].timestamp).toBe(3000);
expect(runs[1].timestamp).toBe(2000);
expect(runs[2].timestamp).toBe(1000);
});
it("filters by workflow name", () => {
packages/cli/src/commands/workflow.ts
+7 -7
View File
@@ -28,8 +28,8 @@ export function getDbPath(): string {
return join(getNerveRoot(), "data", "logs.db");
}
export function formatTs(ts: number): string {
return new Date(ts).toISOString();
export function formatTs(timestampMs: number): string {
return new Date(timestampMs).toISOString();
}
async function openStore(): Promise<LogStore> {
@@ -67,7 +67,7 @@ export function statusIcon(status: WorkflowRun["status"]): string {
}
/**
* Retrieve all workflow runs from the store, sorted by ts descending (newest first).
* Retrieve all workflow runs from the store, sorted by timestamp descending (newest first).
* Delegates to the store's efficient SQL query on the workflow_runs table.
*/
export function getAllWorkflowRuns(store: LogStore, filterWorkflow: string | null): WorkflowRun[] {
@@ -79,7 +79,7 @@ export function getAllWorkflowRuns(store: LogStore, filterWorkflow: string | nul
*/
export function formatRunLine(run: WorkflowRun): string {
const icon = statusIcon(run.status);
return ` ${icon} ${run.runId} workflow=${run.workflow} status=${run.status} ts=${formatTs(run.ts)}\n`;
return ` ${icon} ${run.runId} workflow=${run.workflow} status=${run.status} timestamp=${formatTs(run.timestamp)}\n`;
}
/**
@@ -152,7 +152,7 @@ export function buildInspectOutput(
`🔍 Workflow run: ${run.runId}\n`,
` workflow: ${run.workflow}\n`,
` status: ${run.status}\n`,
` ts: ${formatTs(run.ts)}\n`,
` timestamp: ${formatTs(run.timestamp)}\n`,
`\n📜 Thread events (${shown} of ${total}):\n`,
];
@@ -219,7 +219,7 @@ export function formatThreadRoundBlock(row: ThreadRoundRow): string {
const { roleStr, contentBody, meta } = partitionWorkflowMessage(row.message);
const yamlBlock =
Object.keys(meta).length === 0 ? "{}\n" : `${stringify(meta, { lineWidth: 100 })}\n`;
return `[#${row.round} ${roleStr}] ${formatTs(row.ts)}\n---\n${yamlBlock}---\n${contentBody}\n\n`;
return `[#${row.round} ${roleStr}] ${formatTs(row.timestamp)}\n---\n${yamlBlock}---\n${contentBody}\n\n`;
}
export type ThreadCommandOutput = {
@@ -237,7 +237,7 @@ function buildTruncatedSingleRound(
const { roleStr, contentBody, meta } = partitionWorkflowMessage(row.message);
const yamlBlock =
Object.keys(meta).length === 0 ? "{}\n" : `${stringify(meta, { lineWidth: 100 })}\n`;
const header = `[#${row.round} ${roleStr}] ${formatTs(row.ts)}\n---\n${yamlBlock}---\n`;
const header = `[#${row.round} ${roleStr}] ${formatTs(row.timestamp)}\n---\n${yamlBlock}---\n`;
const maxBody = Math.max(0, remaining - header.length - "[truncated]\n".length);
const truncated =
maxBody > 0 && contentBody.length > maxBody
packages/daemon/src/__tests__/crash-recovery.test.ts
+10 -5
View File
@@ -73,7 +73,7 @@ function makeLogStore(
runId: string;
workflow: string;
status: "queued" | "started";
ts: number;
timestamp: number;
}> = [],
) {
const store = {
@@ -199,7 +199,7 @@ describe("WorkflowManager — crash recovery (Phase 3)", () => {
it("sends resume-thread for 'started' runs from DB after respawn", async () => {
const activeRuns = [
{ runId: "run-started-1", workflow: "my-wf", status: "started" as const, ts: 1000 },
{ runId: "run-started-1", workflow: "my-wf", status: "started" as const, timestamp: 1000 },
];
const logStore = makeLogStore(activeRuns);
logStore.getThreadMessages.mockReturnValue([
@@ -245,7 +245,7 @@ describe("WorkflowManager — crash recovery (Phase 3)", () => {
it("re-queues 'queued' runs from DB after respawn", async () => {
const activeRuns = [
{ runId: "run-queued-1", workflow: "my-wf", status: "queued" as const, ts: 900 },
{ runId: "run-queued-1", workflow: "my-wf", status: "queued" as const, timestamp: 900 },
];
const logStore = makeLogStore(activeRuns);
logStore.getTriggerPayload.mockReturnValue({ queued: "payload" });
@@ -342,7 +342,7 @@ describe("WorkflowManager — crash recovery (Phase 3)", () => {
describe("runId deduplication in crash recovery", () => {
it("does not push duplicate runIds into the queue during crash recovery", async () => {
const activeRuns = [
{ runId: "run-queued-dup", workflow: "my-wf", status: "queued" as const, ts: 900 },
{ runId: "run-queued-dup", workflow: "my-wf", status: "queued" as const, timestamp: 900 },
];
const logStore = makeLogStore(activeRuns);
logStore.getTriggerPayload.mockReturnValue({ q: 1 });
@@ -378,7 +378,12 @@ describe("WorkflowManager — crash recovery (Phase 3)", () => {
it("does not add duplicate active runIds during crash recovery", async () => {
const activeRuns = [
{ runId: "run-started-dup", workflow: "my-wf", status: "started" as const, ts: 1000 },
{
runId: "run-started-dup",
workflow: "my-wf",
status: "started" as const,
timestamp: 1000,
},
];
const logStore = makeLogStore(activeRuns);
logStore.getThreadMessages.mockReturnValue([]);
packages/daemon/src/workflow-manager.ts
+10 -4
View File
@@ -239,14 +239,20 @@ export function createWorkflowManager(
eventType: string,
payload?: unknown,
): void {
const ts = Date.now();
const timestamp = Date.now();
const serialised = payload !== undefined ? JSON.stringify(payload) : null;
const status = toWorkflowRunStatus(eventType);
if (status !== null) {
logStore.upsertWorkflowRun(
{ source: "workflow", type: eventType, refId: runId, payload: serialised, timestamp: ts },
{ runId, workflow: workflowName, status, ts },
{
source: "workflow",
type: eventType,
refId: runId,
payload: serialised,
timestamp,
},
{ runId, workflow: workflowName, status, timestamp },
);
} else {
logStore.append({
@@ -254,7 +260,7 @@ export function createWorkflowManager(
type: eventType,
refId: runId,
payload: serialised,
timestamp: ts,
timestamp,
});
}
}
packages/store/src/__tests__/log-store-crash-recovery.test.ts
+3 -3
View File
@@ -41,7 +41,7 @@ describe("LogStore — crash recovery helpers (Phase 3)", () => {
payload: JSON.stringify({ triggerPayload: payload }),
timestamp: 1000,
},
{ runId: "run-1", workflow: "my-wf", status: "started", ts: 1000 },
{ runId: "run-1", workflow: "my-wf", status: "started", timestamp: 1000 },
);
const result = store.getTriggerPayload("run-1");
@@ -57,7 +57,7 @@ describe("LogStore — crash recovery helpers (Phase 3)", () => {
payload: null,
timestamp: 1000,
},
{ runId: "run-2", workflow: "my-wf", status: "started", ts: 1000 },
{ runId: "run-2", workflow: "my-wf", status: "started", timestamp: 1000 },
);
expect(store.getTriggerPayload("run-2")).toBeNull();
@@ -148,7 +148,7 @@ describe("LogStore — crash recovery helpers (Phase 3)", () => {
payload: JSON.stringify({ triggerPayload: {} }),
timestamp: 1000,
},
{ runId: "run-6", workflow: "my-wf", status: "started", ts: 1000 },
{ runId: "run-6", workflow: "my-wf", status: "started", timestamp: 1000 },
);
store.append({
source: "workflow",
packages/store/src/__tests__/log-store-workflow.test.ts
+18 -18
View File
@@ -30,7 +30,7 @@ describe("LogStore — workflow_runs", () => {
runId: "run-1",
workflow: "cleanup",
status: "started",
ts: 1000,
timestamp: 1000,
};
const entry = store.upsertWorkflowRun(
@@ -47,23 +47,23 @@ describe("LogStore — workflow_runs", () => {
expect(stored?.runId).toBe("run-1");
expect(stored?.workflow).toBe("cleanup");
expect(stored?.status).toBe("started");
expect(stored?.ts).toBe(1000);
expect(stored?.timestamp).toBe(1000);
});
it("updates existing workflow_runs row on upsert (status transition)", () => {
store.upsertWorkflowRun(
{ source: "workflow", type: "started", refId: "run-2", payload: null, timestamp: 1000 },
{ runId: "run-2", workflow: "cleanup", status: "started", ts: 1000 },
{ runId: "run-2", workflow: "cleanup", status: "started", timestamp: 1000 },
);
store.upsertWorkflowRun(
{ source: "workflow", type: "completed", refId: "run-2", payload: null, timestamp: 2000 },
{ runId: "run-2", workflow: "cleanup", status: "completed", ts: 2000 },
{ runId: "run-2", workflow: "cleanup", status: "completed", timestamp: 2000 },
);
const stored = store.getWorkflowRun("run-2");
expect(stored?.status).toBe("completed");
expect(stored?.ts).toBe(2000);
expect(stored?.timestamp).toBe(2000);
// Both log entries should be present (event sourcing)
const logs = store.query({ refId: "run-2" });
@@ -71,15 +71,15 @@ describe("LogStore — workflow_runs", () => {
});
it("the log entries act as source of truth for event history", () => {
for (const [type, status, ts] of [
for (const [type, status, timestamp] of [
["queued", "queued", 1000],
["started", "started", 1001],
["step_complete", "started", 1002],
["completed", "completed", 1005],
] as const) {
store.upsertWorkflowRun(
{ source: "workflow", type, refId: "run-3", payload: null, timestamp: ts },
{ runId: "run-3", workflow: "cleanup", status, ts },
{ source: "workflow", type, refId: "run-3", payload: null, timestamp },
{ runId: "run-3", workflow: "cleanup", status, timestamp },
);
}
@@ -98,16 +98,16 @@ describe("LogStore — workflow_runs", () => {
it("returns the latest state after multiple upserts", () => {
store.upsertWorkflowRun(
{ source: "workflow", type: "queued", refId: "run-4", payload: null, timestamp: 100 },
{ runId: "run-4", workflow: "code-review", status: "queued", ts: 100 },
{ runId: "run-4", workflow: "code-review", status: "queued", timestamp: 100 },
);
store.upsertWorkflowRun(
{ source: "workflow", type: "started", refId: "run-4", payload: null, timestamp: 200 },
{ runId: "run-4", workflow: "code-review", status: "started", ts: 200 },
{ runId: "run-4", workflow: "code-review", status: "started", timestamp: 200 },
);
const run = store.getWorkflowRun("run-4");
expect(run?.status).toBe("started");
expect(run?.ts).toBe(200);
expect(run?.timestamp).toBe(200);
});
});
@@ -115,19 +115,19 @@ describe("LogStore — workflow_runs", () => {
beforeEach(() => {
store.upsertWorkflowRun(
{ source: "workflow", type: "queued", refId: "r1", payload: null, timestamp: 100 },
{ runId: "r1", workflow: "cleanup", status: "queued", ts: 100 },
{ runId: "r1", workflow: "cleanup", status: "queued", timestamp: 100 },
);
store.upsertWorkflowRun(
{ source: "workflow", type: "started", refId: "r2", payload: null, timestamp: 200 },
{ runId: "r2", workflow: "cleanup", status: "started", ts: 200 },
{ runId: "r2", workflow: "cleanup", status: "started", timestamp: 200 },
);
store.upsertWorkflowRun(
{ source: "workflow", type: "completed", refId: "r3", payload: null, timestamp: 300 },
{ runId: "r3", workflow: "cleanup", status: "completed", ts: 300 },
{ runId: "r3", workflow: "cleanup", status: "completed", timestamp: 300 },
);
store.upsertWorkflowRun(
{ source: "workflow", type: "failed", refId: "r4", payload: null, timestamp: 400 },
{ runId: "r4", workflow: "deploy", status: "queued", ts: 400 },
{ runId: "r4", workflow: "deploy", status: "queued", timestamp: 400 },
);
});
@@ -164,9 +164,9 @@ describe("LogStore — workflow_runs", () => {
expect(store.getActiveWorkflowRuns("nonexistent")).toHaveLength(0);
});
it("returns runs ordered by ts ascending", () => {
it("returns runs ordered by timestamp ascending", () => {
const active = store.getActiveWorkflowRuns();
expect(active[0].ts).toBeLessThan(active[1].ts);
expect(active[0].timestamp).toBeLessThan(active[1].timestamp);
});
});
@@ -177,7 +177,7 @@ describe("LogStore — workflow_runs", () => {
const runId = `run-${status}`;
store.upsertWorkflowRun(
{ source: "workflow", type: status, refId: runId, payload: null, timestamp: 1 },
{ runId, workflow: "test", status, ts: 1 },
{ runId, workflow: "test", status, timestamp: 1 },
);
expect(store.getWorkflowRun(runId)?.status).toBe(status);
},
packages/store/src/log-store.ts
+18 -18
View File
@@ -86,14 +86,14 @@ export type WorkflowRun = {
runId: string;
workflow: string;
status: WorkflowRunStatus;
ts: number;
timestamp: number;
};
/** One role-produced workflow-message row with 1-based round index (ROW_NUMBER over role messages only). */
export type ThreadRoundRow = {
round: number;
logId: number;
ts: number;
timestamp: number;
message: { role: string; content: string; meta: unknown; timestamp: number };
};
@@ -131,7 +131,7 @@ export type LogStore = {
*/
getActiveWorkflowRuns: (workflowName?: string) => WorkflowRun[];
/**
* Get all workflow runs regardless of status, sorted by ts descending.
* Get all workflow runs regardless of status, sorted by timestamp descending.
* Optionally filter by workflow name.
*/
getAllWorkflowRuns: (workflowName: string | null) => WorkflowRun[];
@@ -195,7 +195,7 @@ CREATE TABLE IF NOT EXISTS workflow_runs (
run_id TEXT PRIMARY KEY,
workflow TEXT NOT NULL,
status TEXT NOT NULL,
ts INTEGER NOT NULL
timestamp INTEGER NOT NULL
);
CREATE INDEX IF NOT EXISTS idx_workflow_runs_status ON workflow_runs(status);
@@ -342,11 +342,11 @@ export function createLogStore(dbPath: string): LogStore {
);
const upsertWorkflowRunStmt = sqlite.prepare(
"INSERT OR REPLACE INTO workflow_runs (run_id, workflow, status, ts) VALUES (@runId, @workflow, @status, @ts)",
"INSERT OR REPLACE INTO workflow_runs (run_id, workflow, status, timestamp) VALUES (@runId, @workflow, @status, @timestamp)",
);
const getWorkflowRunStmt = sqlite.prepare(
"SELECT run_id, workflow, status, ts FROM workflow_runs WHERE run_id = ?",
"SELECT run_id, workflow, status, timestamp FROM workflow_runs WHERE run_id = ?",
);
const getTriggerPayloadStmt = sqlite.prepare(
@@ -386,19 +386,19 @@ export function createLogStore(dbPath: string): LogStore {
);
const getActiveWorkflowRunsStmt = sqlite.prepare(
"SELECT run_id, workflow, status, ts FROM workflow_runs WHERE status IN ('queued', 'started') ORDER BY ts ASC",
"SELECT run_id, workflow, status, timestamp FROM workflow_runs WHERE status IN ('queued', 'started') ORDER BY timestamp ASC",
);
const getActiveWorkflowRunsByNameStmt = sqlite.prepare(
"SELECT run_id, workflow, status, ts FROM workflow_runs WHERE status IN ('queued', 'started') AND workflow = ? ORDER BY ts ASC",
"SELECT run_id, workflow, status, timestamp FROM workflow_runs WHERE status IN ('queued', 'started') AND workflow = ? ORDER BY timestamp ASC",
);
const getAllWorkflowRunsStmt = sqlite.prepare(
"SELECT run_id, workflow, status, ts FROM workflow_runs ORDER BY ts DESC",
"SELECT run_id, workflow, status, timestamp FROM workflow_runs ORDER BY timestamp DESC",
);
const getAllWorkflowRunsByNameStmt = sqlite.prepare(
"SELECT run_id, workflow, status, ts FROM workflow_runs WHERE workflow = ? ORDER BY ts DESC",
"SELECT run_id, workflow, status, timestamp FROM workflow_runs WHERE workflow = ? ORDER BY timestamp DESC",
);
const minLogTsStmt = sqlite.prepare("SELECT MIN(timestamp) AS m FROM logs");
@@ -422,7 +422,7 @@ export function createLogStore(dbPath: string): LogStore {
runId: run.runId,
workflow: run.workflow,
status: run.status,
ts: run.ts,
timestamp: run.timestamp,
});
return { ...entry, id: Number(info.lastInsertRowid) };
});
@@ -506,14 +506,14 @@ export function createLogStore(dbPath: string): LogStore {
function getWorkflowRun(runId: string): WorkflowRun | null {
const row = getWorkflowRunStmt.get(runId) as
| { run_id: string; workflow: string; status: string; ts: number }
| { run_id: string; workflow: string; status: string; timestamp: number }
| undefined;
if (row === undefined) return null;
return {
runId: row.run_id,
workflow: row.workflow,
status: validateWorkflowRunStatus(row.status),
ts: row.ts,
timestamp: row.timestamp,
};
}
@@ -522,12 +522,12 @@ export function createLogStore(dbPath: string): LogStore {
workflowName !== undefined
? getActiveWorkflowRunsByNameStmt.all(workflowName)
: getActiveWorkflowRunsStmt.all()
) as Array<{ run_id: string; workflow: string; status: string; ts: number }>;
) as Array<{ run_id: string; workflow: string; status: string; timestamp: number }>;
return rows.map((r) => ({
runId: r.run_id,
workflow: r.workflow,
status: validateWorkflowRunStatus(r.status),
ts: r.ts,
timestamp: r.timestamp,
}));
}
@@ -536,12 +536,12 @@ export function createLogStore(dbPath: string): LogStore {
workflowName !== null
? getAllWorkflowRunsByNameStmt.all(workflowName)
: getAllWorkflowRunsStmt.all()
) as Array<{ run_id: string; workflow: string; status: string; ts: number }>;
) as Array<{ run_id: string; workflow: string; status: string; timestamp: number }>;
return rows.map((r) => ({
runId: r.run_id,
workflow: r.workflow,
status: validateWorkflowRunStatus(r.status),
ts: r.ts,
timestamp: r.timestamp,
}));
}
@@ -657,7 +657,7 @@ export function createLogStore(dbPath: string): LogStore {
if (row.payload === null) continue;
const message = parseRoundPayload(row.payload, row.timestamp);
if (message !== null) {
out.push({ round: row.rn, logId: row.id, ts: row.timestamp, message });
out.push({ round: row.rn, logId: row.id, timestamp: row.timestamp, message });
}
}
return out;