Compare commits

..

3 Commits

Author SHA1 Message Date
xiaoju 3dc835e1de refactor(store): rename LogEntry/WorkflowRun/ThreadRoundRow ts → timestamp
- 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
2026-04-25 02:24:39 +00:00
xiaoju 4da2c87a77 refactor(store): rename LogEntry.ts → LogEntry.timestamp
- Rename logs table column ts → timestamp (no migration, breaking)
- Update all SQL, type definitions, and consumers
- Integration tests use mkdtempSync to avoid stale DB conflicts

Fixes #113
2026-04-25 02:08:57 +00:00
xiaoju 529cceba06 Merge pull request 'refactor(core): remove unnecessary | null, unify timestamp naming' (#112) from refactor/108-remove-null-unify-ts into main 2026-04-25 01:57:48 +00:00
20 changed files with 361 additions and 216 deletions
+23 -23
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, 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,8 +251,8 @@ describe("buildInspectOutput", () => {
expect(eventLines.join("")).toContain("no events recorded");
});
it("shows event lines with type and ts", () => {
const logs = [{ ts: 1_700_000_001_000, type: "started", payload: null }];
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("");
expect(text).toContain("type=started");
@@ -260,7 +260,7 @@ describe("buildInspectOutput", () => {
it("truncates long payloads to 200 chars with ellipsis", () => {
const longPayload = "x".repeat(250);
const logs = [{ ts: 1000, type: "step_complete", payload: longPayload }];
const logs = [{ timestamp: 1000, type: "step_complete", payload: longPayload }];
const { eventLines } = buildInspectOutput(baseRun, logs, 0, 20);
const text = eventLines.join("");
expect(text).toContain("…");
@@ -268,14 +268,14 @@ describe("buildInspectOutput", () => {
});
it("shows short payloads in full", () => {
const logs = [{ ts: 1000, type: "step_complete", payload: '{"count":5}' }];
const logs = [{ timestamp: 1000, type: "step_complete", payload: '{"count":5}' }];
const { eventLines } = buildInspectOutput(baseRun, logs, 0, 20);
expect(eventLines.join("")).toContain('{"count":5}');
});
it("paginates events with a hint", () => {
const logs = Array.from({ length: 5 }, (_, i) => ({
ts: 1000 + i,
timestamp: 1000 + i,
type: "step_complete",
payload: null,
}));
@@ -287,7 +287,7 @@ describe("buildInspectOutput", () => {
});
it("no pagination hint when all events fit on one page", () => {
const logs = [{ ts: 1000, type: "started", payload: null }];
const logs = [{ timestamp: 1000, type: "started", payload: null }];
const { paginationHint } = buildInspectOutput(baseRun, logs, 0, 20);
expect(paginationHint).toBeNull();
});
@@ -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", () => {
+9 -9
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`;
}
/**
@@ -139,7 +139,7 @@ export type InspectOutput = {
export function buildInspectOutput(
run: WorkflowRun,
allLogs: Array<{ ts: number; type: string; payload: string | null }>,
allLogs: Array<{ timestamp: number; type: string; payload: string | null }>,
offset: number,
limit: number,
): InspectOutput {
@@ -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`,
];
@@ -167,7 +167,7 @@ export function buildInspectOutput(
: entry.payload.length <= 200
? ` payload=${entry.payload}`
: ` payload=${entry.payload.slice(0, 200)}`;
eventLines.push(` [${formatTs(entry.ts)}] type=${entry.type}${payloadStr}\n`);
eventLines.push(` [${formatTs(entry.timestamp)}] type=${entry.type}${payloadStr}\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
@@ -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([]);
@@ -10,6 +10,9 @@
*/
import { EventEmitter } from "node:events";
import { mkdtempSync, rmSync } from "node:fs";
import { tmpdir } from "node:os";
import { join } from "node:path";
import type { NerveConfig, WorkflowConfig } from "@uncaged/nerve-core";
import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
@@ -236,14 +239,18 @@ describe("WorkflowManager — drainAndRespawn (Phase 3 hot reload)", () => {
});
describe("Kernel — workflow hot reload via file-watcher (Phase 3)", () => {
let nerveRoot: string;
beforeEach(() => {
mockChildren.length = 0;
vi.useFakeTimers({ shouldAdvanceTime: true });
nerveRoot = mkdtempSync(join(tmpdir(), "nerve-hot-reload-"));
});
afterEach(async () => {
vi.useRealTimers();
vi.clearAllMocks();
rmSync(nerveRoot, { recursive: true, force: true });
});
it("handleWorkflowFileChange logs workflow_reload system event", async () => {
@@ -255,7 +262,7 @@ describe("Kernel — workflow hot reload via file-watcher (Phase 3)", () => {
maxRounds: 10,
};
const kernel = createKernel(config, "/tmp/nerve-hot-reload-test", {
const kernel = createKernel(config, nerveRoot, {
workerScript: "fake-worker.js",
logStore,
});
@@ -290,7 +297,7 @@ describe("Kernel — workflow hot reload via file-watcher (Phase 3)", () => {
maxRounds: 10,
};
const kernel = createKernel(initialConfig, "/tmp/nerve-hot-reload-test", {
const kernel = createKernel(initialConfig, nerveRoot, {
workerScript: "fake-worker.js",
logStore,
});
@@ -333,7 +340,7 @@ describe("Kernel — workflow hot reload via file-watcher (Phase 3)", () => {
maxRounds: 10,
};
const kernel = createKernel(initialConfig, "/tmp/nerve-hot-reload-test", {
const kernel = createKernel(initialConfig, nerveRoot, {
workerScript: "fake-worker.js",
logStore,
});
@@ -6,12 +6,14 @@
* artifacts are required.
*/
import { mkdtempSync, rmSync } from "node:fs";
import { tmpdir } from "node:os";
import { dirname, join } from "node:path";
import { fileURLToPath } from "node:url";
import type { Signal } from "@uncaged/nerve-core";
import type { NerveConfig } from "@uncaged/nerve-core";
import { afterEach, describe, expect, it } from "vitest";
import { afterEach, beforeEach, describe, expect, it } from "vitest";
import { createKernel } from "../kernel.js";
import type { Kernel } from "../kernel.js";
@@ -55,12 +57,18 @@ async function pollUntil(
describe("kernel integration — real child processes", () => {
let kernel: Kernel | null = null;
let nerveRoot: string;
beforeEach(() => {
nerveRoot = mkdtempSync(join(tmpdir(), "nerve-integration-"));
});
afterEach(async () => {
if (kernel !== null) {
await kernel.stop();
kernel = null;
}
rmSync(nerveRoot, { recursive: true, force: true });
});
it("returns correct groups and senseCount", () => {
@@ -71,7 +79,7 @@ describe("kernel integration — real child processes", () => {
"net-rx": { group: "network", throttle: null, timeout: null, gracePeriod: null },
},
});
kernel = createKernel(config, "/tmp/nerve-integration-test", {
kernel = createKernel(config, nerveRoot, {
workerScript: MOCK_WORKER,
});
@@ -83,7 +91,7 @@ describe("kernel integration — real child processes", () => {
it("workers start and respond to compute messages with signals", async () => {
const config = makeConfig();
kernel = createKernel(config, "/tmp/nerve-integration-test", {
kernel = createKernel(config, nerveRoot, {
workerScript: MOCK_WORKER,
});
@@ -115,7 +123,7 @@ describe("kernel integration — real child processes", () => {
"net-rx": { group: "network", throttle: null, timeout: null, gracePeriod: null },
},
});
kernel = createKernel(config, "/tmp/nerve-integration-test", {
kernel = createKernel(config, nerveRoot, {
workerScript: MOCK_WORKER,
});
@@ -131,7 +139,7 @@ describe("kernel integration — real child processes", () => {
it("compute round-trip: worker receives compute and sends signal back through bus", async () => {
const config = makeConfig();
kernel = createKernel(config, "/tmp/nerve-integration-test", {
kernel = createKernel(config, nerveRoot, {
workerScript: MOCK_WORKER,
});
@@ -158,7 +166,7 @@ describe("kernel integration — real child processes", () => {
it("crash recovery: kernel respawns worker after unexpected exit and new worker is functional", async () => {
const config = makeConfig();
kernel = createKernel(config, "/tmp/nerve-integration-test", {
kernel = createKernel(config, nerveRoot, {
workerScript: MOCK_WORKER,
});
@@ -4,6 +4,9 @@
*/
import { EventEmitter } from "node:events";
import { mkdtempSync, rmSync } from "node:fs";
import { tmpdir } from "node:os";
import { join } from "node:path";
import type { NerveConfig } from "@uncaged/nerve-core";
import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
@@ -84,13 +87,17 @@ function makeConfig(overrides: Partial<NerveConfig> = {}): NerveConfig {
// ---------------------------------------------------------------------------
describe("kernel — getHealth", () => {
let nerveRoot: string;
beforeEach(() => {
mockChildren.length = 0;
vi.useFakeTimers({ shouldAdvanceTime: true });
nerveRoot = mkdtempSync(join(tmpdir(), "nerve-kernel-p6-health-"));
});
afterEach(() => {
vi.useRealTimers();
rmSync(nerveRoot, { recursive: true, force: true });
});
it("returns correct health shape", async () => {
@@ -101,7 +108,7 @@ describe("kernel — getHealth", () => {
"net-rx": { group: "network", throttle: null, timeout: null, gracePeriod: null },
},
});
const kernel = createKernel(config, "/tmp/nerve-test");
const kernel = createKernel(config, nerveRoot);
const health = kernel.getHealth();
expect(health.activeSenses).toBe(3);
@@ -115,18 +122,22 @@ describe("kernel — getHealth", () => {
});
describe("kernel — restartGroup", () => {
let nerveRoot: string;
beforeEach(() => {
mockChildren.length = 0;
vi.useFakeTimers({ shouldAdvanceTime: true });
nerveRoot = mkdtempSync(join(tmpdir(), "nerve-kernel-p6-restart-"));
});
afterEach(() => {
vi.useRealTimers();
rmSync(nerveRoot, { recursive: true, force: true });
});
it("sends shutdown to old worker and spawns new one", async () => {
const config = makeConfig();
const kernel = createKernel(config, "/tmp/nerve-test");
const kernel = createKernel(config, nerveRoot);
expect(mockChildren.length).toBe(1);
const oldChild = mockChildren[0];
@@ -146,7 +157,7 @@ describe("kernel — restartGroup", () => {
it("restartGroup on unknown group does nothing", async () => {
const config = makeConfig();
const kernel = createKernel(config, "/tmp/nerve-test");
const kernel = createKernel(config, nerveRoot);
expect(mockChildren.length).toBe(1);
await kernel.restartGroup("nonexistent");
@@ -158,18 +169,22 @@ describe("kernel — restartGroup", () => {
});
describe("kernel — reloadConfig", () => {
let nerveRoot: string;
beforeEach(() => {
mockChildren.length = 0;
vi.useFakeTimers({ shouldAdvanceTime: true });
nerveRoot = mkdtempSync(join(tmpdir(), "nerve-kernel-p6-reload-"));
});
afterEach(() => {
vi.useRealTimers();
rmSync(nerveRoot, { recursive: true, force: true });
});
it("adds new group worker when new sense group appears", async () => {
const config = makeConfig();
const kernel = createKernel(config, "/tmp/nerve-test");
const kernel = createKernel(config, nerveRoot);
expect(mockChildren.length).toBe(1); // only system group
expect(kernel.groups.has("network")).toBe(false);
@@ -200,7 +215,7 @@ describe("kernel — reloadConfig", () => {
workflows: {},
maxRounds: 10,
};
const kernel = createKernel(config, "/tmp/nerve-test");
const kernel = createKernel(config, nerveRoot);
expect(mockChildren.length).toBe(2);
expect(kernel.groups.has("network")).toBe(true);
@@ -225,7 +240,7 @@ describe("kernel — reloadConfig", () => {
it("health reflects updated sense count after reloadConfig", async () => {
const config = makeConfig();
const kernel = createKernel(config, "/tmp/nerve-test");
const kernel = createKernel(config, nerveRoot);
expect(kernel.getHealth().activeSenses).toBe(1);
@@ -6,6 +6,9 @@
*/
import { EventEmitter } from "node:events";
import { mkdtempSync, rmSync } from "node:fs";
import { tmpdir } from "node:os";
import { join } from "node:path";
import type { NerveConfig } from "@uncaged/nerve-core";
import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
@@ -103,18 +106,22 @@ function makeConfig(overrides: Partial<NerveConfig> = {}): NerveConfig {
// ---------------------------------------------------------------------------
describe("kernel.triggerSense()", () => {
let nerveRoot: string;
beforeEach(() => {
mockChildren.length = 0;
vi.useFakeTimers({ shouldAdvanceTime: true });
nerveRoot = mkdtempSync(join(tmpdir(), "nerve-trigger-sense-"));
});
afterEach(() => {
vi.useRealTimers();
rmSync(nerveRoot, { recursive: true, force: true });
});
it("throws for an unknown sense name", async () => {
const config = makeConfig();
const kernel = createKernel(config, "/tmp/nerve-test", {
const kernel = createKernel(config, nerveRoot, {
workerScript: null,
logStore: makeMockLogStore() as never,
});
@@ -132,7 +139,7 @@ describe("kernel.triggerSense()", () => {
},
reflexes: [],
});
const kernel = createKernel(config, "/tmp/nerve-test", {
const kernel = createKernel(config, nerveRoot, {
workerScript: null,
logStore: makeMockLogStore() as never,
});
@@ -162,7 +169,7 @@ describe("kernel.triggerSense()", () => {
},
reflexes: [],
});
const kernel = createKernel(config, "/tmp/nerve-test", {
const kernel = createKernel(config, nerveRoot, {
workerScript: null,
logStore: makeMockLogStore() as never,
});
@@ -185,7 +192,7 @@ describe("kernel.triggerSense()", () => {
vi.useRealTimers();
const config = makeConfig();
const kernel = createKernel(config, "/tmp/nerve-test", {
const kernel = createKernel(config, nerveRoot, {
workerScript: null,
logStore: makeMockLogStore() as never,
});
@@ -9,6 +9,9 @@
*/
import { EventEmitter } from "node:events";
import { mkdtempSync, rmSync } from "node:fs";
import { tmpdir } from "node:os";
import { join } from "node:path";
import type { NerveConfig } from "@uncaged/nerve-core";
import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
@@ -106,14 +109,18 @@ function makeConfig(overrides: Partial<NerveConfig> = {}): NerveConfig {
// ---------------------------------------------------------------------------
describe("kernel + workflowManager integration", () => {
let nerveRoot: string;
beforeEach(() => {
mockChildren.length = 0;
vi.useFakeTimers({ shouldAdvanceTime: true });
nerveRoot = mkdtempSync(join(tmpdir(), "nerve-kernel-wf-"));
});
afterEach(async () => {
vi.useRealTimers();
vi.clearAllMocks();
rmSync(nerveRoot, { recursive: true, force: true });
});
describe("sense compute triggers workflow via return value", () => {
@@ -127,7 +134,7 @@ describe("kernel + workflowManager integration", () => {
workflows: { "my-workflow": { concurrency: 2, overflow: "drop" } },
});
const kernel = createKernel(config, "/tmp/nerve-test", {
const kernel = createKernel(config, nerveRoot, {
workerScript: "fake-worker.js",
logStore,
});
@@ -171,7 +178,7 @@ describe("kernel + workflowManager integration", () => {
workflows: { "alert-workflow": { concurrency: 1, overflow: "drop" } },
});
const kernel = createKernel(config, "/tmp/nerve-test", {
const kernel = createKernel(config, nerveRoot, {
workerScript: "fake-worker.js",
logStore,
});
@@ -221,7 +228,7 @@ describe("kernel + workflowManager integration", () => {
workflows: { "my-workflow": { concurrency: 1, overflow: "drop" } },
});
const kernel = createKernel(config, "/tmp/nerve-test", {
const kernel = createKernel(config, nerveRoot, {
workerScript: "fake-worker.js",
logStore,
});
@@ -264,7 +271,7 @@ describe("kernel + workflowManager integration", () => {
workflows: { "log-test-workflow": { concurrency: 2, overflow: "drop" } },
});
const kernel = createKernel(config, "/tmp/nerve-test", {
const kernel = createKernel(config, nerveRoot, {
workerScript: "fake-worker.js",
logStore,
});
@@ -302,7 +309,7 @@ describe("kernel + workflowManager integration", () => {
maxRounds: 10,
});
const kernel = createKernel(initialConfig, "/tmp/nerve-test", {
const kernel = createKernel(initialConfig, nerveRoot, {
workerScript: "fake-worker.js",
logStore,
});
@@ -355,7 +362,7 @@ describe("kernel + workflowManager integration", () => {
workflows: { "old-workflow": { concurrency: 1, overflow: "drop" } },
});
const kernel = createKernel(initialConfig, "/tmp/nerve-test", {
const kernel = createKernel(initialConfig, nerveRoot, {
workerScript: "fake-worker.js",
logStore,
});
@@ -414,7 +421,7 @@ describe("kernel + workflowManager integration", () => {
workflows: { "shutdown-test": { concurrency: 1, overflow: "drop" } },
});
const kernel = createKernel(config, "/tmp/nerve-test", {
const kernel = createKernel(config, nerveRoot, {
workerScript: "fake-worker.js",
logStore,
});
@@ -440,7 +447,7 @@ describe("kernel + workflowManager integration", () => {
workflows: { "my-wf": { concurrency: 1, overflow: "drop" } },
});
const kernel = createKernel(config, "/tmp/nerve-test", {
const kernel = createKernel(config, nerveRoot, {
workerScript: "fake-worker.js",
logStore,
});
@@ -463,7 +470,7 @@ describe("kernel + workflowManager integration", () => {
workflows: { "health-wf": { concurrency: 2, overflow: "drop" } },
});
const kernel = createKernel(config, "/tmp/nerve-test", {
const kernel = createKernel(config, nerveRoot, {
workerScript: "fake-worker.js",
logStore,
});
+16 -6
View File
@@ -70,13 +70,17 @@ function makeConfig(overrides: Partial<NerveConfig> = {}): NerveConfig {
// ---------------------------------------------------------------------------
describe("kernel — message routing", () => {
let nerveRoot: string;
beforeEach(() => {
mockChildren.length = 0;
vi.useFakeTimers({ shouldAdvanceTime: true });
nerveRoot = mkdtempSync(join(tmpdir(), "nerve-kernel-msg-"));
});
afterEach(() => {
vi.useRealTimers();
rmSync(nerveRoot, { recursive: true, force: true });
});
it("routes signal message to bus without throwing", async () => {
@@ -86,7 +90,7 @@ describe("kernel — message routing", () => {
},
reflexes: [],
});
const kernel = createKernel(config, "/tmp/nerve-test");
const kernel = createKernel(config, nerveRoot);
expect(mockChildren.length).toBe(1);
const child = mockChildren[0];
@@ -129,7 +133,7 @@ describe("kernel — message routing", () => {
},
reflexes: [],
});
const kernel = createKernel(config, "/tmp/nerve-test");
const kernel = createKernel(config, nerveRoot);
const child = mockChildren[0];
child.emit("message", { type: "error", sense: "cpu-usage", error: "compute failed" });
@@ -150,7 +154,7 @@ describe("kernel — message routing", () => {
},
reflexes: [],
});
const kernel = createKernel(config, "/tmp/nerve-test");
const kernel = createKernel(config, nerveRoot);
const child = mockChildren[0];
const callsBefore = stderrSpy.mock.calls.length;
@@ -170,7 +174,7 @@ describe("kernel — message routing", () => {
},
reflexes: [],
});
const kernel = createKernel(config, "/tmp/nerve-test");
const kernel = createKernel(config, nerveRoot);
const child = mockChildren[0];
expect(() => child.emit("message", { type: "unknown-type" })).not.toThrow();
@@ -183,13 +187,17 @@ describe("kernel — message routing", () => {
});
describe("kernel — groupForSense mapping", () => {
let nerveRoot: string;
beforeEach(() => {
mockChildren.length = 0;
vi.useFakeTimers({ shouldAdvanceTime: true });
nerveRoot = mkdtempSync(join(tmpdir(), "nerve-kernel-groups-"));
});
afterEach(() => {
vi.useRealTimers();
rmSync(nerveRoot, { recursive: true, force: true });
});
it("spawns one worker per unique group", async () => {
@@ -203,7 +211,7 @@ describe("kernel — groupForSense mapping", () => {
workflows: {},
maxRounds: 10,
};
const kernel = createKernel(config, "/tmp/nerve-test");
const kernel = createKernel(config, nerveRoot);
// system and network = 2 unique groups
expect(mockChildren.length).toBe(2);
@@ -217,7 +225,7 @@ describe("kernel — groupForSense mapping", () => {
},
reflexes: [{ kind: "sense", sense: "cpu-usage", interval: 500, on: [] }],
});
createKernel(config, "/tmp/nerve-test");
const kernel = createKernel(config, nerveRoot);
const child = mockChildren[0];
vi.advanceTimersByTime(500);
@@ -225,5 +233,7 @@ describe("kernel — groupForSense mapping", () => {
expect(child.send).toHaveBeenCalledWith(
expect.objectContaining({ type: "compute", sense: "cpu-usage" }),
);
await kernel.stop();
});
});
@@ -108,7 +108,7 @@ describe("LogStore + ReflexScheduler integration", () => {
type: "run_complete",
refId: "cpu-usage",
payload: '{"v":99}',
ts: Date.now(),
timestamp: Date.now(),
});
// Writing to the log store should NOT trigger any reflex.
@@ -2,12 +2,14 @@
* Phase 6 integration tests — hot reload, error isolation, grace period, health.
*/
import { mkdtempSync, rmSync } from "node:fs";
import { tmpdir } from "node:os";
import { dirname, join } from "node:path";
import { fileURLToPath } from "node:url";
import type { Signal } from "@uncaged/nerve-core";
import type { NerveConfig } from "@uncaged/nerve-core";
import { afterEach, describe, expect, it } from "vitest";
import { afterEach, beforeEach, describe, expect, it } from "vitest";
import { createKernel } from "../kernel.js";
import type { Kernel } from "../kernel.js";
@@ -55,17 +57,23 @@ async function pollUntil(
describe("phase6 — restartGroup", () => {
let kernel: Kernel | null = null;
let nerveRoot: string;
beforeEach(() => {
nerveRoot = mkdtempSync(join(tmpdir(), "nerve-phase6-restart-"));
});
afterEach(async () => {
if (kernel !== null) {
await kernel.stop();
kernel = null;
}
rmSync(nerveRoot, { recursive: true, force: true });
});
it("restartGroup stops old worker and spawns a new one", async () => {
const config = makeConfig();
kernel = createKernel(config, "/tmp/nerve-phase6-test", {
kernel = createKernel(config, nerveRoot, {
workerScript: MOCK_WORKER,
});
@@ -97,7 +105,7 @@ describe("phase6 — restartGroup", () => {
it("restartGroup on nonexistent group does nothing", async () => {
const config = makeConfig();
kernel = createKernel(config, "/tmp/nerve-phase6-test", {
kernel = createKernel(config, nerveRoot, {
workerScript: MOCK_WORKER,
});
await kernel.ready;
@@ -113,17 +121,23 @@ describe("phase6 — restartGroup", () => {
describe("phase6 — reloadConfig", () => {
let kernel: Kernel | null = null;
let nerveRoot: string;
beforeEach(() => {
nerveRoot = mkdtempSync(join(tmpdir(), "nerve-phase6-reload-"));
});
afterEach(async () => {
if (kernel !== null) {
await kernel.stop();
kernel = null;
}
rmSync(nerveRoot, { recursive: true, force: true });
});
it("adds new group when new sense group is introduced", async () => {
const config = makeConfig();
kernel = createKernel(config, "/tmp/nerve-phase6-test", {
kernel = createKernel(config, nerveRoot, {
workerScript: MOCK_WORKER,
});
await kernel.ready;
@@ -159,7 +173,7 @@ describe("phase6 — reloadConfig", () => {
workflows: {},
maxRounds: 10,
};
kernel = createKernel(config, "/tmp/nerve-phase6-test", {
kernel = createKernel(config, nerveRoot, {
workerScript: MOCK_WORKER,
});
await kernel.ready;
@@ -187,12 +201,18 @@ describe("phase6 — reloadConfig", () => {
describe("phase6 — error isolation", () => {
let kernel: Kernel | null = null;
let nerveRoot: string;
beforeEach(() => {
nerveRoot = mkdtempSync(join(tmpdir(), "nerve-phase6-err-"));
});
afterEach(async () => {
if (kernel !== null) {
await kernel.stop();
kernel = null;
}
rmSync(nerveRoot, { recursive: true, force: true });
});
it("error from one sense does not crash the worker — other senses still work", async () => {
@@ -206,7 +226,7 @@ describe("phase6 — error isolation", () => {
maxRounds: 10,
};
kernel = createKernel(config, "/tmp/nerve-phase6-test", {
kernel = createKernel(config, nerveRoot, {
workerScript: MOCK_WORKER,
});
await kernel.ready;
@@ -238,7 +258,7 @@ describe("phase6 — error isolation", () => {
process.stderr.write = stderrSpy as typeof process.stderr.write;
const config = makeConfig();
kernel = createKernel(config, "/tmp/nerve-phase6-test", {
kernel = createKernel(config, nerveRoot, {
workerScript: ERROR_WORKER,
});
await kernel.ready;
@@ -261,12 +281,18 @@ describe("phase6 — error isolation", () => {
describe("phase6 — getHealth", () => {
let kernel: Kernel | null = null;
let nerveRoot: string;
beforeEach(() => {
nerveRoot = mkdtempSync(join(tmpdir(), "nerve-phase6-health-"));
});
afterEach(async () => {
if (kernel !== null) {
await kernel.stop();
kernel = null;
}
rmSync(nerveRoot, { recursive: true, force: true });
});
it("returns health snapshot with correct shape", async () => {
@@ -277,7 +303,7 @@ describe("phase6 — getHealth", () => {
"net-rx": { group: "network", throttle: null, timeout: null, gracePeriod: null },
},
});
kernel = createKernel(config, "/tmp/nerve-phase6-test", {
kernel = createKernel(config, nerveRoot, {
workerScript: MOCK_WORKER,
});
await kernel.ready;
@@ -293,7 +319,7 @@ describe("phase6 — getHealth", () => {
it("health reflects config changes after reloadConfig", async () => {
const config = makeConfig();
kernel = createKernel(config, "/tmp/nerve-phase6-test", {
kernel = createKernel(config, nerveRoot, {
workerScript: MOCK_WORKER,
});
await kernel.ready;
@@ -322,17 +348,23 @@ describe("phase6 — getHealth", () => {
describe("phase6 — auto-respawn on worker crash", () => {
let kernel: Kernel | null = null;
let nerveRoot: string;
beforeEach(() => {
nerveRoot = mkdtempSync(join(tmpdir(), "nerve-phase6-crash-"));
});
afterEach(async () => {
if (kernel !== null) {
await kernel.stop();
kernel = null;
}
rmSync(nerveRoot, { recursive: true, force: true });
});
it("kernel auto-respawns worker and new worker is functional", async () => {
const config = makeConfig();
kernel = createKernel(config, "/tmp/nerve-phase6-test", {
kernel = createKernel(config, nerveRoot, {
workerScript: MOCK_WORKER,
});
await kernel.ready;
+3 -3
View File
@@ -38,7 +38,7 @@ export function createKernelFileWatchHandlers(deps: KernelFileWatchDeps): Kernel
type: "sense_reload",
refId: senseName,
payload: null,
ts: Date.now(),
timestamp: Date.now(),
});
deps.restartGroup(sc.group).catch((e) => {
const msg = e instanceof Error ? e.message : String(e);
@@ -55,7 +55,7 @@ export function createKernelFileWatchHandlers(deps: KernelFileWatchDeps): Kernel
type: "workflow_reload",
refId: workflowName,
payload: null,
ts: Date.now(),
timestamp: Date.now(),
});
deps.workflowManager.drainAndRespawn(workflowName).catch((e) => {
const msg = e instanceof Error ? e.message : String(e);
@@ -70,7 +70,7 @@ export function createKernelFileWatchHandlers(deps: KernelFileWatchDeps): Kernel
type: "config_reload",
refId: null,
payload: null,
ts: Date.now(),
timestamp: Date.now(),
});
try {
const raw = readFileSync(join(deps.nerveRoot, "nerve.yaml"), "utf8");
+6 -6
View File
@@ -81,7 +81,7 @@ export function createKernel(
type: "start",
refId: null,
payload: null,
ts: startTime,
timestamp: startTime,
});
let config = initialConfig;
@@ -138,7 +138,7 @@ export function createKernel(
type: "error",
refId: msg.sense,
payload: JSON.stringify({ error: msg.error }),
ts: Date.now(),
timestamp: Date.now(),
});
scheduler.onComputeComplete(msg.sense);
return;
@@ -154,7 +154,7 @@ export function createKernel(
type: "workflow-launch",
refId: msg.sense,
payload: JSON.stringify(route.launch),
ts: Date.now(),
timestamp: Date.now(),
});
} else {
const signal: Signal = {
@@ -168,7 +168,7 @@ export function createKernel(
type: "signal",
refId: msg.sense,
payload: JSON.stringify(route.payload),
ts: signal.timestamp,
timestamp: signal.timestamp,
});
bus.emit(signal);
}
@@ -327,7 +327,7 @@ export function createKernel(
group: senseConfig.group,
throttle: senseConfig.throttle,
timeout: senseConfig.timeout,
lastSignalTimestamp: lastEntry !== null ? lastEntry.ts : null,
lastSignalTimestamp: lastEntry !== null ? lastEntry.timestamp : null,
};
});
},
@@ -352,7 +352,7 @@ export function createKernel(
type: "stop",
refId: null,
payload: null,
ts: Date.now(),
timestamp: Date.now(),
});
logStore.close();
}
+1 -1
View File
@@ -78,7 +78,7 @@ export function createReflexScheduler(
type: "run_start",
refId: senseName,
payload: null,
ts: Date.now(),
timestamp: Date.now(),
});
triggerFn(senseName);
}
+11 -5
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, 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,
ts,
timestamp,
});
}
}
@@ -440,7 +446,7 @@ export function createWorkflowManager(
type: "thread_workflow_message",
refId: msg.runId,
payload: JSON.stringify(msg.message),
ts: Date.now(),
timestamp: Date.now(),
});
return;
}
@@ -30,8 +30,8 @@ describe("LogStore — cold archive (RFC-001 §5.4)", () => {
it("exports one UTC day to JSONL, deletes rows, advances archived_up_to", () => {
const ts = Date.UTC(2026, 1, 1, 10, 0, 0);
store.append({ source: "system", type: "x", refId: null, payload: '{"a":1}', ts });
store.append({ source: "reflex", type: "y", refId: "z", payload: null, ts: ts + 1 });
store.append({ source: "system", type: "x", refId: null, payload: '{"a":1}', timestamp: ts });
store.append({ source: "reflex", type: "y", refId: "z", payload: null, timestamp: ts + 1 });
const now = nowForLastArchivableFeb1();
const result = store.archiveLogs({ now, retentionMs: 30 * DAY_MS });
@@ -61,7 +61,7 @@ describe("LogStore — cold archive (RFC-001 §5.4)", () => {
it("does nothing when all logs are inside the hot window", () => {
const now = Date.UTC(2026, 3, 23, 12, 0, 0);
const ts = now - 5 * DAY_MS;
store.append({ source: "system", type: "warm", refId: null, payload: null, ts });
store.append({ source: "system", type: "warm", refId: null, payload: null, timestamp: ts });
const r = store.archiveLogs({ now, retentionMs: 30 * DAY_MS });
expect(r.days).toHaveLength(0);
expect(store.query()).toHaveLength(1);
@@ -69,7 +69,7 @@ describe("LogStore — cold archive (RFC-001 §5.4)", () => {
it("second archive with same clock is a no-op (watermark already caught up)", () => {
const ts = Date.UTC(2026, 1, 1, 10, 0, 0);
store.append({ source: "system", type: "x", refId: null, payload: null, ts });
store.append({ source: "system", type: "x", refId: null, payload: null, timestamp: ts });
const now = nowForLastArchivableFeb1();
store.archiveLogs({ now, retentionMs: 30 * DAY_MS });
const path = join(tmpDir, "data", "archive", "logs", "2026-02-01.jsonl");
@@ -82,11 +82,11 @@ describe("LogStore — cold archive (RFC-001 §5.4)", () => {
it("overwrites JSONL when the same UTC day is archived again after watermark rewind", () => {
const ts = Date.UTC(2026, 1, 1, 10, 0, 0);
store.append({ source: "a", type: "1", refId: null, payload: null, ts });
store.append({ source: "a", type: "1", refId: null, payload: null, timestamp: ts });
const now = nowForLastArchivableFeb1();
store.archiveLogs({ now, retentionMs: 30 * DAY_MS });
store.setMeta(LOG_ARCHIVE_META_KEY, "2026-01-31");
store.append({ source: "b", type: "2", refId: null, payload: null, ts: ts + 100 });
store.append({ source: "b", type: "2", refId: null, payload: null, timestamp: ts + 100 });
store.archiveLogs({ now, retentionMs: 30 * DAY_MS });
const path = join(tmpDir, "data", "archive", "logs", "2026-02-01.jsonl");
@@ -98,8 +98,8 @@ describe("LogStore — cold archive (RFC-001 §5.4)", () => {
it("respects maxDays across invocations", () => {
const t1 = Date.UTC(2026, 1, 1, 10, 0, 0);
const t2 = Date.UTC(2026, 1, 2, 10, 0, 0);
store.append({ source: "system", type: "a", refId: null, payload: null, ts: t1 });
store.append({ source: "system", type: "b", refId: null, payload: null, ts: t2 });
store.append({ source: "system", type: "a", refId: null, payload: null, timestamp: t1 });
store.append({ source: "system", type: "b", refId: null, payload: null, timestamp: t2 });
const now = Date.UTC(2027, 0, 1, 12, 0, 0);
const r1 = store.archiveLogs({ now, retentionMs: 30 * DAY_MS, maxDays: 1 });
@@ -116,7 +116,7 @@ describe("LogStore — cold archive (RFC-001 §5.4)", () => {
it("starts from earliest log day when it is before watermark+1", () => {
store.setMeta(LOG_ARCHIVE_META_KEY, "2026-01-10");
const ts = Date.UTC(2026, 1, 1, 10, 0, 0);
store.append({ source: "x", type: "p", refId: null, payload: null, ts });
store.append({ source: "x", type: "p", refId: null, payload: null, timestamp: ts });
const result = store.archiveLogs({ now: nowForLastArchivableFeb1(), retentionMs: 30 * DAY_MS });
expect(result.days.map((d) => d.day)).toContain("2026-02-01");
});
@@ -128,7 +128,7 @@ describe("LogStore — cold archive (RFC-001 §5.4)", () => {
it("runs VACUUM when vacuum: true", () => {
const ts = Date.UTC(2026, 1, 1, 10, 0, 0);
store.append({ source: "system", type: "x", refId: null, payload: null, ts });
store.append({ source: "system", type: "x", refId: null, payload: null, timestamp: ts });
const r = store.archiveLogs({
now: nowForLastArchivableFeb1(),
retentionMs: 30 * DAY_MS,
@@ -39,9 +39,9 @@ describe("LogStore — crash recovery helpers (Phase 3)", () => {
type: "started",
refId: "run-1",
payload: JSON.stringify({ triggerPayload: payload }),
ts: 1000,
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");
@@ -55,9 +55,9 @@ describe("LogStore — crash recovery helpers (Phase 3)", () => {
type: "started",
refId: "run-2",
payload: null,
ts: 1000,
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();
@@ -72,14 +72,14 @@ describe("LogStore — crash recovery helpers (Phase 3)", () => {
type: "started",
refId: "run-3",
payload: JSON.stringify({ triggerPayload: payloadA }),
ts: 100,
timestamp: 100,
});
store.append({
source: "workflow",
type: "started",
refId: "run-3",
payload: JSON.stringify({ triggerPayload: payloadB }),
ts: 200,
timestamp: 200,
});
const result = store.getTriggerPayload("run-3");
@@ -106,7 +106,7 @@ describe("LogStore — crash recovery helpers (Phase 3)", () => {
type: "thread_command_event",
refId: "run-4",
payload: JSON.stringify(event),
ts: Date.now(),
timestamp: Date.now(),
});
}
@@ -123,14 +123,14 @@ describe("LogStore — crash recovery helpers (Phase 3)", () => {
type: "thread_command_event",
refId: "run-5",
payload: null,
ts: 1000,
timestamp: 1000,
});
store.append({
source: "workflow",
type: "thread_command_event",
refId: "run-5",
payload: JSON.stringify({ type: "valid_event" }),
ts: 1001,
timestamp: 1001,
});
const result = store.getThreadEvents("run-5");
@@ -146,23 +146,23 @@ describe("LogStore — crash recovery helpers (Phase 3)", () => {
type: "started",
refId: "run-6",
payload: JSON.stringify({ triggerPayload: {} }),
ts: 1000,
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",
type: "thread_command_event",
refId: "run-6",
payload: JSON.stringify({ type: "step_one" }),
ts: 1001,
timestamp: 1001,
});
store.append({
source: "workflow",
type: "step_complete",
refId: "run-6",
payload: JSON.stringify({ message: "done step" }),
ts: 1002,
timestamp: 1002,
});
const result = store.getThreadEvents("run-6");
@@ -176,14 +176,14 @@ describe("LogStore — crash recovery helpers (Phase 3)", () => {
type: "thread_command_event",
refId: "run-7",
payload: JSON.stringify({ type: "event_for_7" }),
ts: 1000,
timestamp: 1000,
});
store.append({
source: "workflow",
type: "thread_command_event",
refId: "run-8",
payload: JSON.stringify({ type: "event_for_8" }),
ts: 1001,
timestamp: 1001,
});
const result7 = store.getThreadEvents("run-7");
@@ -203,7 +203,7 @@ describe("LogStore — crash recovery helpers (Phase 3)", () => {
type: "thread_command_event",
refId: "run-tr",
payload: JSON.stringify({ type: "thread_start", triggerPayload: { x: 1 } }),
ts: 100,
timestamp: 100,
});
store.append({
source: "workflow",
@@ -215,14 +215,14 @@ describe("LogStore — crash recovery helpers (Phase 3)", () => {
content: "hello",
meta: 1,
}),
ts: 101,
timestamp: 101,
});
store.append({
source: "workflow",
type: "thread_command_event",
refId: "run-tr",
payload: JSON.stringify({ type: "step_b", role: "beta", content: "world" }),
ts: 102,
timestamp: 102,
});
expect(store.getThreadRoundCount("run-tr")).toBe(2);
@@ -241,7 +241,7 @@ describe("LogStore — crash recovery helpers (Phase 3)", () => {
type: "thread_command_event",
refId: "run-b4",
payload: JSON.stringify({ type: `ev_${i}`, role: "r", content: String(i) }),
ts: 200 + i,
timestamp: 200 + i,
});
}
@@ -30,11 +30,11 @@ describe("LogStore — workflow_runs", () => {
runId: "run-1",
workflow: "cleanup",
status: "started",
ts: 1000,
timestamp: 1000,
};
const entry = store.upsertWorkflowRun(
{ source: "workflow", type: "started", refId: "run-1", payload: null, ts: 1000 },
{ source: "workflow", type: "started", refId: "run-1", payload: null, timestamp: 1000 },
run,
);
@@ -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, ts: 1000 },
{ runId: "run-2", workflow: "cleanup", status: "started", ts: 1000 },
{ source: "workflow", type: "started", refId: "run-2", payload: null, timestamp: 1000 },
{ runId: "run-2", workflow: "cleanup", status: "started", timestamp: 1000 },
);
store.upsertWorkflowRun(
{ source: "workflow", type: "completed", refId: "run-2", payload: null, ts: 2000 },
{ runId: "run-2", workflow: "cleanup", status: "completed", ts: 2000 },
{ source: "workflow", type: "completed", refId: "run-2", payload: null, timestamp: 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, ts },
{ runId: "run-3", workflow: "cleanup", status, ts },
{ source: "workflow", type, refId: "run-3", payload: null, timestamp },
{ runId: "run-3", workflow: "cleanup", status, timestamp },
);
}
@@ -97,37 +97,37 @@ describe("LogStore — workflow_runs", () => {
it("returns the latest state after multiple upserts", () => {
store.upsertWorkflowRun(
{ source: "workflow", type: "queued", refId: "run-4", payload: null, ts: 100 },
{ runId: "run-4", workflow: "code-review", status: "queued", ts: 100 },
{ source: "workflow", type: "queued", refId: "run-4", payload: null, timestamp: 100 },
{ runId: "run-4", workflow: "code-review", status: "queued", timestamp: 100 },
);
store.upsertWorkflowRun(
{ source: "workflow", type: "started", refId: "run-4", payload: null, ts: 200 },
{ runId: "run-4", workflow: "code-review", status: "started", ts: 200 },
{ source: "workflow", type: "started", refId: "run-4", payload: null, timestamp: 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);
});
});
describe("getActiveWorkflowRuns", () => {
beforeEach(() => {
store.upsertWorkflowRun(
{ source: "workflow", type: "queued", refId: "r1", payload: null, ts: 100 },
{ runId: "r1", workflow: "cleanup", status: "queued", ts: 100 },
{ source: "workflow", type: "queued", refId: "r1", payload: null, timestamp: 100 },
{ runId: "r1", workflow: "cleanup", status: "queued", timestamp: 100 },
);
store.upsertWorkflowRun(
{ source: "workflow", type: "started", refId: "r2", payload: null, ts: 200 },
{ runId: "r2", workflow: "cleanup", status: "started", ts: 200 },
{ source: "workflow", type: "started", refId: "r2", payload: null, timestamp: 200 },
{ runId: "r2", workflow: "cleanup", status: "started", timestamp: 200 },
);
store.upsertWorkflowRun(
{ source: "workflow", type: "completed", refId: "r3", payload: null, ts: 300 },
{ runId: "r3", workflow: "cleanup", status: "completed", ts: 300 },
{ source: "workflow", type: "completed", refId: "r3", payload: null, timestamp: 300 },
{ runId: "r3", workflow: "cleanup", status: "completed", timestamp: 300 },
);
store.upsertWorkflowRun(
{ source: "workflow", type: "failed", refId: "r4", payload: null, ts: 400 },
{ runId: "r4", workflow: "deploy", status: "queued", ts: 400 },
{ source: "workflow", type: "failed", refId: "r4", payload: null, timestamp: 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);
});
});
@@ -176,8 +176,8 @@ describe("LogStore — workflow_runs", () => {
(status) => {
const runId = `run-${status}`;
store.upsertWorkflowRun(
{ source: "workflow", type: status, refId: runId, payload: null, ts: 1 },
{ runId, workflow: "test", status, ts: 1 },
{ source: "workflow", type: status, refId: runId, payload: null, timestamp: 1 },
{ runId, workflow: "test", status, timestamp: 1 },
);
expect(store.getWorkflowRun(runId)?.status).toBe(status);
},
+67 -19
View File
@@ -27,7 +27,7 @@ describe("LogStore", () => {
type: "start",
refId: null,
payload: null,
ts: 1000,
timestamp: 1000,
});
expect(entry.id).toBe(1);
@@ -41,28 +41,40 @@ describe("LogStore", () => {
type: "start",
refId: null,
payload: null,
ts: 1000,
timestamp: 1000,
});
const e2 = store.append({
source: "system",
type: "stop",
refId: null,
payload: null,
ts: 2000,
timestamp: 2000,
});
expect(e2.id).toBe((e1.id ?? 0) + 1);
});
it("returns all entries when queried with no filter", () => {
store.append({ source: "system", type: "start", refId: null, payload: null, ts: 1000 });
store.append({ source: "reflex", type: "run_start", refId: "cpu", payload: null, ts: 2000 });
store.append({
source: "system",
type: "start",
refId: null,
payload: null,
timestamp: 1000,
});
store.append({
source: "reflex",
type: "run_start",
refId: "cpu",
payload: null,
timestamp: 2000,
});
store.append({
source: "reflex",
type: "run_complete",
refId: "cpu",
payload: '{"v":42}',
ts: 3000,
timestamp: 3000,
});
const all = store.query();
@@ -72,23 +84,35 @@ describe("LogStore", () => {
describe("query filters", () => {
beforeEach(() => {
store.append({ source: "system", type: "start", refId: null, payload: null, ts: 1000 });
store.append({ source: "reflex", type: "run_start", refId: "cpu", payload: null, ts: 2000 });
store.append({
source: "system",
type: "start",
refId: null,
payload: null,
timestamp: 1000,
});
store.append({
source: "reflex",
type: "run_start",
refId: "cpu",
payload: null,
timestamp: 2000,
});
store.append({
source: "reflex",
type: "run_complete",
refId: "cpu",
payload: '{"v":42}',
ts: 3000,
timestamp: 3000,
});
store.append({
source: "system",
type: "error",
refId: "disk",
payload: '{"error":"fail"}',
ts: 4000,
timestamp: 4000,
});
store.append({ source: "system", type: "stop", refId: null, payload: null, ts: 5000 });
store.append({ source: "system", type: "stop", refId: null, payload: null, timestamp: 5000 });
});
it("filters by source", () => {
@@ -111,7 +135,7 @@ describe("LogStore", () => {
it("filters by since (inclusive)", () => {
const results = store.query({ since: 3000 });
expect(results).toHaveLength(3);
expect(results[0].ts).toBe(3000);
expect(results[0].timestamp).toBe(3000);
});
it("filters by until (inclusive)", () => {
@@ -146,12 +170,24 @@ describe("LogStore", () => {
describe("query ordering", () => {
it("returns entries in insertion order (ascending id)", () => {
store.append({ source: "system", type: "start", refId: null, payload: null, ts: 5000 });
store.append({ source: "reflex", type: "run_start", refId: "a", payload: null, ts: 1000 });
store.append({
source: "system",
type: "start",
refId: null,
payload: null,
timestamp: 5000,
});
store.append({
source: "reflex",
type: "run_start",
refId: "a",
payload: null,
timestamp: 1000,
});
const all = store.query();
expect(all[0].ts).toBe(5000);
expect(all[1].ts).toBe(1000);
expect(all[0].timestamp).toBe(5000);
expect(all[1].timestamp).toBe(1000);
});
});
@@ -182,7 +218,7 @@ describe("LogStore", () => {
describe("append-only semantics", () => {
it("ids are always increasing", () => {
const entries = Array.from({ length: 10 }, (_, i) =>
store.append({ source: "system", type: "test", refId: null, payload: null, ts: i }),
store.append({ source: "system", type: "test", refId: null, payload: null, timestamp: i }),
);
for (let i = 1; i < entries.length; i++) {
@@ -194,7 +230,13 @@ describe("LogStore", () => {
describe("payload JSON round-trip", () => {
it("preserves JSON payload", () => {
const payload = JSON.stringify({ cpu: 95, host: "node-1" });
store.append({ source: "reflex", type: "run_complete", refId: "cpu", payload, ts: 1000 });
store.append({
source: "reflex",
type: "run_complete",
refId: "cpu",
payload,
timestamp: 1000,
});
const results = store.query({ refId: "cpu" });
expect(results).toHaveLength(1);
@@ -206,7 +248,13 @@ describe("LogStore", () => {
it("creates nested directory structure for db path", () => {
const deepPath = join(tmpDir, "a", "b", "c", "test.db");
const deepStore = createLogStore(deepPath);
deepStore.append({ source: "system", type: "start", refId: null, payload: null, ts: 1000 });
deepStore.append({
source: "system",
type: "start",
refId: null,
payload: null,
timestamp: 1000,
});
expect(deepStore.query()).toHaveLength(1);
deepStore.close();
});
+43 -43
View File
@@ -35,7 +35,7 @@ export type LogEntry = {
type: string;
refId: string | null;
payload: string | null;
ts: number;
timestamp: number;
};
export type LogQuery = {
@@ -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[];
@@ -174,16 +174,16 @@ export type LogStore = {
const SCHEMA_SQL = `
CREATE TABLE IF NOT EXISTS logs (
id INTEGER PRIMARY KEY AUTOINCREMENT,
source TEXT NOT NULL,
type TEXT NOT NULL,
ref_id TEXT,
payload TEXT,
ts INTEGER NOT NULL
id INTEGER PRIMARY KEY AUTOINCREMENT,
source TEXT NOT NULL,
type TEXT NOT NULL,
ref_id TEXT,
payload TEXT,
timestamp INTEGER NOT NULL
);
CREATE INDEX IF NOT EXISTS idx_logs_source_type ON logs(source, type);
CREATE INDEX IF NOT EXISTS idx_logs_ts ON logs(ts);
CREATE INDEX IF NOT EXISTS idx_logs_timestamp ON logs(timestamp);
CREATE INDEX IF NOT EXISTS idx_logs_ref_id ON logs(ref_id);
CREATE TABLE IF NOT EXISTS meta (
@@ -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);
@@ -208,7 +208,7 @@ type SqlLogRow = {
type: string;
ref_id: string | null;
payload: string | null;
ts: number;
timestamp: number;
};
function buildJsonlBody(rows: SqlLogRow[]): string {
@@ -220,7 +220,7 @@ function buildJsonlBody(rows: SqlLogRow[]): string {
type: r.type,
refId: r.ref_id,
payload: r.payload,
ts: r.ts,
timestamp: r.timestamp,
}),
);
return `${lines.join("\n")}\n`;
@@ -333,7 +333,7 @@ export function createLogStore(dbPath: string): LogStore {
sqlite.exec(SCHEMA_SQL);
const insertStmt = sqlite.prepare(
"INSERT INTO logs (source, type, ref_id, payload, ts) VALUES (@source, @type, @refId, @payload, @ts)",
"INSERT INTO logs (source, type, ref_id, payload, timestamp) VALUES (@source, @type, @refId, @payload, @timestamp)",
);
const getMetaStmt = sqlite.prepare("SELECT value FROM meta WHERE key = ?");
@@ -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(
@@ -371,7 +371,7 @@ export function createLogStore(dbPath: string): LogStore {
const getThreadRoundsStmt = sqlite.prepare(
`WITH numbered AS (
SELECT id, ts, payload,
SELECT id, timestamp, payload,
ROW_NUMBER() OVER (ORDER BY id ASC) AS rn
FROM logs
WHERE source = 'workflow' AND type IN ('thread_command_event', 'thread_workflow_message') AND ref_id = @runId
@@ -379,34 +379,34 @@ export function createLogStore(dbPath: string): LogStore {
AND COALESCE(json_extract(payload, '$.type'), '') != 'thread_start'
AND COALESCE(json_extract(payload, '$.role'), '') != '__start__'
)
SELECT id, ts, payload, rn FROM numbered
SELECT id, timestamp, payload, rn FROM numbered
WHERE (@before = 0 OR rn < @before)
ORDER BY rn DESC
LIMIT @lim`,
);
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(ts) AS m FROM logs");
const minLogTsStmt = sqlite.prepare("SELECT MIN(timestamp) AS m FROM logs");
const selectLogsForDayStmt = sqlite.prepare(
"SELECT id, source, type, ref_id, payload, ts FROM logs WHERE ts >= @start AND ts < @endExclusive ORDER BY id ASC",
"SELECT id, source, type, ref_id, payload, timestamp FROM logs WHERE timestamp >= @start AND timestamp < @endExclusive ORDER BY id ASC",
);
const deleteLogsForDayStmt = sqlite.prepare(
"DELETE FROM logs WHERE ts >= @start AND ts < @endExclusive",
"DELETE FROM logs WHERE timestamp >= @start AND timestamp < @endExclusive",
);
function upsertWorkflowRunTx(entry: Omit<LogEntry, "id">, run: WorkflowRun): LogEntry {
@@ -416,13 +416,13 @@ export function createLogStore(dbPath: string): LogStore {
type: entry.type,
refId: entry.refId,
payload: entry.payload,
ts: entry.ts,
timestamp: entry.timestamp,
});
upsertWorkflowRunStmt.run({
runId: run.runId,
workflow: run.workflow,
status: run.status,
ts: run.ts,
timestamp: run.timestamp,
});
return { ...entry, id: Number(info.lastInsertRowid) };
});
@@ -434,7 +434,7 @@ export function createLogStore(dbPath: string): LogStore {
type: entry.type,
refId: entry.refId,
payload: entry.payload,
ts: entry.ts,
timestamp: entry.timestamp,
});
return { ...entry, id: Number(info.lastInsertRowid) };
}
@@ -456,17 +456,17 @@ export function createLogStore(dbPath: string): LogStore {
params.refId = filter.refId;
}
if (filter.since !== undefined) {
conditions.push("ts >= @since");
conditions.push("timestamp >= @since");
params.since = filter.since;
}
if (filter.until !== undefined) {
conditions.push("ts <= @until");
conditions.push("timestamp <= @until");
params.until = filter.until;
}
const where = conditions.length > 0 ? `WHERE ${conditions.join(" AND ")}` : "";
const limit = filter.limit !== undefined ? `LIMIT ${filter.limit}` : "";
const sql = `SELECT id, source, type, ref_id, payload, ts FROM logs ${where} ORDER BY id ASC ${limit}`;
const sql = `SELECT id, source, type, ref_id, payload, timestamp FROM logs ${where} ORDER BY id ASC ${limit}`;
const rows = sqlite.prepare(sql).all(params) as Array<{
id: number;
@@ -474,7 +474,7 @@ export function createLogStore(dbPath: string): LogStore {
type: string;
ref_id: string | null;
payload: string | null;
ts: number;
timestamp: number;
}>;
return rows.map((r) => ({
@@ -483,7 +483,7 @@ export function createLogStore(dbPath: string): LogStore {
type: r.type,
refId: r.ref_id,
payload: r.payload,
ts: r.ts,
timestamp: r.timestamp,
}));
}
@@ -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,
}));
}
@@ -650,14 +650,14 @@ export function createLogStore(dbPath: string): LogStore {
runId,
before: params.before,
lim: params.limit,
}) as Array<{ id: number; ts: number; payload: string | null; rn: number }>;
}) as Array<{ id: number; timestamp: number; payload: string | null; rn: number }>;
const out: ThreadRoundRow[] = [];
for (const row of rows) {
if (row.payload === null) continue;
const message = parseRoundPayload(row.payload, row.ts);
const message = parseRoundPayload(row.payload, row.timestamp);
if (message !== null) {
out.push({ round: row.rn, logId: row.id, ts: row.ts, message });
out.push({ round: row.rn, logId: row.id, timestamp: row.timestamp, message });
}
}
return out;