518 lines
17 KiB
TypeScript
518 lines
17 KiB
TypeScript
import assert from "node:assert/strict";
|
|
import {
|
|
mkdir,
|
|
mkdtemp,
|
|
readFile,
|
|
rm,
|
|
stat,
|
|
writeFile,
|
|
} from "node:fs/promises";
|
|
import { tmpdir } from "node:os";
|
|
import * as path from "node:path";
|
|
import test from "node:test";
|
|
import { DEFAULT_WORKFLOW_CONFIG, type WorkflowRun } from "./domain.ts";
|
|
import { readVerifiedReportArtifact, WorkflowStore } from "./store.ts";
|
|
|
|
async function withTempDir(run: (directory: string) => Promise<void>) {
|
|
const directory = await mkdtemp(path.join(tmpdir(), "workflow-store-"));
|
|
try {
|
|
await run(directory);
|
|
} finally {
|
|
await rm(directory, { recursive: true, force: true });
|
|
}
|
|
}
|
|
|
|
function workflow(runId = "run-1"): WorkflowRun {
|
|
return {
|
|
runId,
|
|
revision: 0,
|
|
mode: "feature",
|
|
status: "running",
|
|
background: false,
|
|
request: "Build the feature",
|
|
featureName: "build-feature",
|
|
parentSessionIds: ["session-1"],
|
|
lease: {
|
|
processId: 100,
|
|
activeParentSessionId: "session-1",
|
|
heartbeatAt: 1,
|
|
},
|
|
workspace: {
|
|
state: "creating",
|
|
sourceRoot: "/repository",
|
|
root: `/repository/.worktrees/${runId}`,
|
|
branch: "feature/build-feature",
|
|
baseHead: "abc123",
|
|
},
|
|
effectiveConfig: structuredClone(DEFAULT_WORKFLOW_CONFIG),
|
|
pinnedProfiles: [],
|
|
preflightEligibility: [],
|
|
facts: {
|
|
stage: "discover",
|
|
taskKind: "feature",
|
|
requestTerms: ["build", "feature"],
|
|
candidatePaths: [],
|
|
languages: [],
|
|
tags: [],
|
|
riskTags: [],
|
|
},
|
|
stages: [],
|
|
gates: [],
|
|
decisions: [],
|
|
assignments: [],
|
|
prototypeCandidates: [],
|
|
prototypeCleanupCompleted: false,
|
|
deliveries: [],
|
|
commandReceipts: [],
|
|
mutatorProfileNames: [],
|
|
uiImpactEvidence: [],
|
|
affectedUiSurfaces: [],
|
|
reviewCycles: 0,
|
|
verificationCycles: 0,
|
|
};
|
|
}
|
|
|
|
test("plan revisions use immutable artifact paths while updating plan.md convenience", async () => {
|
|
await withTempDir(async (directory) => {
|
|
const store = new WorkflowStore(directory);
|
|
await store.createRun(workflow());
|
|
const first = await store.savePlan("run-1", 1, "# First plan");
|
|
const second = await store.savePlan("run-1", 2, "# Second plan");
|
|
assert.notEqual(first.path, second.path);
|
|
assert.equal(await readFile(first.path, "utf8"), "# First plan\n");
|
|
assert.equal(await readFile(second.path, "utf8"), "# Second plan\n");
|
|
});
|
|
});
|
|
|
|
test("prototype evidence is retained after throwaway source cleanup", async () => {
|
|
await withTempDir(async (directory) => {
|
|
const store = new WorkflowStore(path.join(directory, "state"));
|
|
await store.createRun(workflow());
|
|
const candidate = path.join(directory, "worktree", "prototypes", "candidate-1");
|
|
await mkdir(candidate, { recursive: true });
|
|
const reportPath = path.join(candidate, "mobile.json");
|
|
const screenshotPath = path.join(candidate, "mobile.png");
|
|
await writeFile(reportPath, '{"result":"passed"}', "utf8");
|
|
await writeFile(screenshotPath, Buffer.from([137, 80, 78, 71, 13, 10, 26, 10]));
|
|
const artifact = await store.savePrototypeArtifact(
|
|
"run-1",
|
|
"candidate-1",
|
|
{ path: "prototypes/candidate-1" },
|
|
path.join(directory, "worktree"),
|
|
["prototypes/candidate-1/mobile.json", "prototypes/candidate-1/mobile.png"],
|
|
);
|
|
await rm(candidate, { recursive: true });
|
|
assert.equal(artifact.retainedFiles.length, 2);
|
|
assert.equal(await readFile(artifact.retainedFiles[0]!, "utf8"), '{"result":"passed"}');
|
|
await stat(artifact.path);
|
|
});
|
|
});
|
|
|
|
test("terminal reports bound large persisted Decisions instead of blocking completion", async () => {
|
|
await withTempDir(async (directory) => {
|
|
const store = new WorkflowStore(path.join(directory, "state"));
|
|
const run = workflow();
|
|
run.status = "completed";
|
|
run.decisions.push({
|
|
gateId: "gate-large",
|
|
revision: 1,
|
|
idempotencyKey: "large-decision",
|
|
decidedAt: 1,
|
|
value: {
|
|
kind: "answer-questions",
|
|
answers: { large: "x".repeat(900_000) },
|
|
},
|
|
});
|
|
run.verificationEvidence = {
|
|
focusedBehavioralTests: true,
|
|
verifierSucceeded: true,
|
|
commands: Array.from({ length: 1_024 }, (_, index) => ({
|
|
id: `command-${index}`,
|
|
executable: "node",
|
|
args: ["--test", "x".repeat(2_000)],
|
|
ownedPaths: [],
|
|
kind: "test" as const,
|
|
status: "succeeded" as const,
|
|
})),
|
|
};
|
|
await store.createRun(run);
|
|
const report = await store.saveReport(run);
|
|
const contents = await readFile(report.path, "utf8");
|
|
assert.match(contents, /^# workflow build-feature/m);
|
|
assert.match(contents, /- Run ID: run-1/);
|
|
assert.match(contents, /\[truncated\]/);
|
|
assert.match(contents, /report truncated to 1 MiB/);
|
|
assert.ok(Buffer.byteLength(contents, "utf8") <= 1024 * 1024);
|
|
});
|
|
});
|
|
|
|
test("supported version-zero snapshots migrate missing lifecycle collections", async () => {
|
|
await withTempDir(async (directory) => {
|
|
const store = new WorkflowStore(path.join(directory, "state"));
|
|
const legacy = workflow();
|
|
const record = legacy as unknown as Record<string, unknown>;
|
|
delete record.background;
|
|
delete record.assignments;
|
|
delete record.deliveries;
|
|
delete record.prototypeCandidates;
|
|
delete record.prototypeCleanupCompleted;
|
|
delete record.commandReceipts;
|
|
delete record.uiImpactEvidence;
|
|
delete record.affectedUiSurfaces;
|
|
delete record.reviewCycles;
|
|
delete record.verificationCycles;
|
|
delete (legacy.effectiveConfig as Partial<typeof legacy.effectiveConfig>)
|
|
.standardAgents;
|
|
const runDirectory = path.join(
|
|
directory,
|
|
"state",
|
|
"runs",
|
|
legacy.runId,
|
|
"revisions",
|
|
);
|
|
await mkdir(runDirectory, { recursive: true });
|
|
await writeFile(
|
|
path.join(runDirectory, "0.json"),
|
|
`${JSON.stringify({ schemaVersion: 0, run: legacy })}\n`,
|
|
"utf8",
|
|
);
|
|
const loaded = await store.loadRun(legacy.runId);
|
|
assert.equal(loaded.background, false);
|
|
assert.deepEqual(loaded.assignments, []);
|
|
assert.deepEqual(loaded.deliveries, []);
|
|
assert.deepEqual(loaded.prototypeCandidates, []);
|
|
assert.deepEqual(loaded.commandReceipts, []);
|
|
assert.equal(loaded.prototypeCleanupCompleted, false);
|
|
assert.deepEqual(
|
|
loaded.effectiveConfig.standardAgents,
|
|
DEFAULT_WORKFLOW_CONFIG.standardAgents,
|
|
);
|
|
});
|
|
});
|
|
|
|
test("report reads reject tampering before dashboard display", async () => {
|
|
await withTempDir(async (directory) => {
|
|
const store = new WorkflowStore(path.join(directory, "state"));
|
|
const run = workflow();
|
|
run.status = "completed";
|
|
await store.createRun(run);
|
|
const report = await store.saveReport(run);
|
|
await writeFile(report.path, "tampered\n", "utf8");
|
|
await assert.rejects(
|
|
readVerifiedReportArtifact(report),
|
|
/digest mismatch/,
|
|
);
|
|
});
|
|
});
|
|
|
|
test("delivery claims admit one process and recover only after expiry", async () => {
|
|
await withTempDir(async (directory) => {
|
|
const store = new WorkflowStore(path.join(directory, "state"));
|
|
await store.createRun(workflow());
|
|
const claims = await Promise.all([
|
|
store.claimDelivery("run-1", "gate:key", "owner-a", 1_000),
|
|
store.claimDelivery("run-1", "gate:key", "owner-b", 1_000),
|
|
]);
|
|
assert.equal(claims.filter(Boolean).length, 1);
|
|
assert.equal(
|
|
await store.claimDelivery("run-1", "gate:key", "owner-c", 30_999),
|
|
false,
|
|
);
|
|
assert.equal(
|
|
await store.claimDelivery("run-1", "gate:key", "owner-c", 31_000),
|
|
true,
|
|
);
|
|
});
|
|
});
|
|
|
|
test("current snapshots default newly introduced lifecycle collections", async () => {
|
|
await withTempDir(async (directory) => {
|
|
const store = new WorkflowStore(directory);
|
|
const legacy = workflow();
|
|
const legacyRecord = legacy as unknown as Record<string, unknown>;
|
|
delete legacyRecord.mutatorProfileNames;
|
|
delete legacyRecord.prototypeCleanupCompleted;
|
|
delete legacyRecord.uiImpactEvidence;
|
|
delete legacyRecord.affectedUiSurfaces;
|
|
delete legacyRecord.reviewCycles;
|
|
delete legacyRecord.verificationCycles;
|
|
await store.createRun(legacy);
|
|
const loaded = await store.loadRun(legacy.runId);
|
|
assert.deepEqual(loaded.mutatorProfileNames, []);
|
|
assert.equal(loaded.prototypeCleanupCompleted, false);
|
|
assert.deepEqual(loaded.uiImpactEvidence, []);
|
|
assert.deepEqual(loaded.affectedUiSurfaces, []);
|
|
assert.equal(loaded.reviewCycles, 0);
|
|
assert.equal(loaded.verificationCycles, 0);
|
|
});
|
|
});
|
|
|
|
test("atomic snapshots leave the complete previous value when replacement is interrupted", async () => {
|
|
await withTempDir(async (directory) => {
|
|
let interrupt = false;
|
|
const store = new WorkflowStore(directory, {
|
|
beforeRename(target) {
|
|
if (interrupt && target.endsWith(path.join("revisions", "1.json"))) {
|
|
throw new Error("simulated interruption");
|
|
}
|
|
},
|
|
});
|
|
const initial = workflow();
|
|
await store.createRun(initial);
|
|
|
|
interrupt = true;
|
|
const changed = structuredClone(initial);
|
|
changed.revision = 1;
|
|
changed.status = "waiting";
|
|
await assert.rejects(store.saveRun(changed), /simulated interruption/);
|
|
assert.deepEqual(await store.loadRun(initial.runId), initial);
|
|
|
|
interrupt = false;
|
|
await store.saveRun(changed);
|
|
assert.deepEqual(await store.loadRun(initial.runId), changed);
|
|
const info = await stat(
|
|
path.join(directory, "runs", initial.runId, "workflow.json"),
|
|
);
|
|
assert.equal(info.mode & 0o777, 0o600);
|
|
});
|
|
});
|
|
|
|
test("revision-checked saves serialize concurrent writers", async () => {
|
|
await withTempDir(async (directory) => {
|
|
const store = new WorkflowStore(directory);
|
|
const initial = workflow();
|
|
await store.createRun(initial);
|
|
const left = structuredClone(initial);
|
|
left.revision = 1;
|
|
left.status = "waiting";
|
|
const right = structuredClone(initial);
|
|
right.revision = 1;
|
|
right.status = "aborted";
|
|
|
|
const results = await Promise.allSettled([
|
|
store.saveRun(left, 0),
|
|
store.saveRun(right, 0),
|
|
]);
|
|
assert.equal(
|
|
results.filter((result) => result.status === "fulfilled").length,
|
|
1,
|
|
);
|
|
const rejected = results.find((result) => result.status === "rejected");
|
|
assert.match(
|
|
String(rejected && rejected.status === "rejected" && rejected.reason),
|
|
/Stale workflow revision/,
|
|
);
|
|
assert.equal((await store.loadRun(initial.runId)).revision, 1);
|
|
});
|
|
});
|
|
|
|
test("a published revision survives interruption before the convenience snapshot", async () => {
|
|
await withTempDir(async (directory) => {
|
|
let interrupt = false;
|
|
const store = new WorkflowStore(directory, {
|
|
beforeRename(target) {
|
|
if (interrupt && target.endsWith("workflow.json")) {
|
|
throw new Error("simulated pointer interruption");
|
|
}
|
|
},
|
|
});
|
|
const initial = workflow();
|
|
await store.createRun(initial);
|
|
const changed = structuredClone(initial);
|
|
changed.revision = 1;
|
|
changed.status = "waiting";
|
|
interrupt = true;
|
|
await assert.rejects(store.saveRun(changed, 0), /pointer interruption/);
|
|
assert.deepEqual(await store.loadRun(initial.runId), changed);
|
|
});
|
|
});
|
|
|
|
test("run creation is atomically published and a pending start can retry", async () => {
|
|
await withTempDir(async (directory) => {
|
|
let interrupt = true;
|
|
const interruptedStore = new WorkflowStore(directory, {
|
|
beforeRename(target) {
|
|
if (interrupt && target.endsWith("request.md")) {
|
|
throw new Error("simulated creation crash");
|
|
}
|
|
},
|
|
});
|
|
const initial = workflow();
|
|
await interruptedStore.reserveStart(
|
|
"launch-key",
|
|
"fingerprint",
|
|
initial.runId,
|
|
);
|
|
await assert.rejects(
|
|
interruptedStore.createRun(initial),
|
|
/simulated creation crash/,
|
|
);
|
|
assert.deepEqual(await interruptedStore.listRuns(), []);
|
|
|
|
interrupt = false;
|
|
const reservation = await interruptedStore.reserveStart(
|
|
"launch-key",
|
|
"fingerprint",
|
|
initial.runId,
|
|
);
|
|
assert.equal(reservation.created, false);
|
|
await interruptedStore.createRun(initial);
|
|
assert.deepEqual(
|
|
(await interruptedStore.listRuns()).map((run) => run.runId),
|
|
[initial.runId],
|
|
);
|
|
});
|
|
});
|
|
|
|
test("a concurrent publisher is reloaded instead of quarantined", async () => {
|
|
await withTempDir(async (directory) => {
|
|
let markMiss: (() => void) | undefined;
|
|
let releaseMiss: (() => void) | undefined;
|
|
const missed = new Promise<void>((resolve) => {
|
|
markMiss = resolve;
|
|
});
|
|
const wait = new Promise<void>((resolve) => {
|
|
releaseMiss = resolve;
|
|
});
|
|
const slow = new WorkflowStore(directory, {
|
|
async afterRunLoadMiss() {
|
|
markMiss?.();
|
|
await wait;
|
|
},
|
|
});
|
|
const fast = new WorkflowStore(directory);
|
|
const slowRun = workflow("run-race");
|
|
slowRun.request = "Slow publisher";
|
|
const fastRun = workflow("run-race");
|
|
fastRun.request = "Fast publisher";
|
|
|
|
const slowPublication = slow.createRun(slowRun);
|
|
await missed;
|
|
const fastPublication = await fast.createRun(fastRun);
|
|
assert.equal(fastPublication.created, true);
|
|
releaseMiss?.();
|
|
const slowResult = await slowPublication;
|
|
assert.equal(slowResult.created, false);
|
|
assert.equal(slowResult.run.request, "Fast publisher");
|
|
assert.equal((await fast.loadRun("run-race")).request, "Fast publisher");
|
|
});
|
|
});
|
|
|
|
test("incomplete legacy run directories are skipped and fail closed", async () => {
|
|
await withTempDir(async (directory) => {
|
|
const store = new WorkflowStore(directory);
|
|
await store.initialize();
|
|
const initial = workflow();
|
|
await mkdir(path.join(directory, "runs", initial.runId), { mode: 0o700 });
|
|
assert.deepEqual(await store.listRuns(), []);
|
|
await assert.rejects(
|
|
store.createRun(initial),
|
|
/incomplete published directory/,
|
|
);
|
|
});
|
|
});
|
|
|
|
test("start reservations are idempotent and recover an incomplete record", async () => {
|
|
await withTempDir(async (directory) => {
|
|
let interrupt = true;
|
|
const store = new WorkflowStore(directory, {
|
|
beforeRename(target) {
|
|
if (interrupt && target.endsWith("record.json")) {
|
|
throw new Error("simulated reservation crash");
|
|
}
|
|
},
|
|
});
|
|
await assert.rejects(
|
|
store.reserveStart("launch-key", "fingerprint-a", "run-a"),
|
|
/simulated reservation crash/,
|
|
);
|
|
interrupt = false;
|
|
const recovered = await store.reserveStart(
|
|
"launch-key",
|
|
"fingerprint-a",
|
|
"run-a",
|
|
);
|
|
assert.equal(recovered.created, true);
|
|
const duplicate = await store.reserveStart(
|
|
"launch-key",
|
|
"fingerprint-a",
|
|
"run-b",
|
|
);
|
|
assert.equal(duplicate.created, false);
|
|
assert.equal(duplicate.record.runId, "run-a");
|
|
await assert.rejects(
|
|
store.reserveStart("launch-key", "fingerprint-b", "run-c"),
|
|
/another start command/,
|
|
);
|
|
|
|
const result = { runId: "run-a", revision: 3 };
|
|
await store.completeStart("launch-key", result);
|
|
assert.deepEqual(await store.getStart("launch-key"), {
|
|
key: "launch-key",
|
|
fingerprint: "fingerprint-a",
|
|
runId: "run-a",
|
|
status: "completed",
|
|
result,
|
|
});
|
|
});
|
|
});
|
|
|
|
test("start completion keeps the highest durable run revision", async () => {
|
|
await withTempDir(async (directory) => {
|
|
const store = new WorkflowStore(directory);
|
|
await store.reserveStart("completion-key", "fingerprint", "run-a");
|
|
await store.completeStart("completion-key", {
|
|
runId: "run-a",
|
|
revision: 5,
|
|
status: "waiting",
|
|
});
|
|
await store.completeStart("completion-key", {
|
|
runId: "run-a",
|
|
revision: 2,
|
|
status: "running",
|
|
});
|
|
const completed = await store.getStart("completion-key");
|
|
assert.equal(completed?.status, "completed");
|
|
assert.equal(
|
|
(completed?.result as { revision?: number } | undefined)?.revision,
|
|
5,
|
|
);
|
|
});
|
|
});
|
|
|
|
test("concurrent start intents publish one authoritative run ID", async () => {
|
|
await withTempDir(async (directory) => {
|
|
const store = new WorkflowStore(directory);
|
|
const [left, right] = await Promise.all([
|
|
store.reserveStart("claimed-key", "fingerprint", "run-left"),
|
|
store.reserveStart("claimed-key", "fingerprint", "run-right"),
|
|
]);
|
|
assert.equal([left, right].filter((result) => result.created).length, 1);
|
|
assert.equal(left.record.runId, right.record.runId);
|
|
assert.ok(
|
|
left.record.runId === "run-left" || left.record.runId === "run-right",
|
|
);
|
|
});
|
|
});
|
|
|
|
test("run artifacts and plans are bounded, pinned, and discoverable", async () => {
|
|
await withTempDir(async (directory) => {
|
|
const store = new WorkflowStore(directory);
|
|
const first = workflow("run-a");
|
|
const second = workflow("run-b");
|
|
await store.createRun(first);
|
|
await store.createRun(second);
|
|
const plan = await store.savePlan(
|
|
first.runId,
|
|
4,
|
|
"# Plan\n\nImplement it.",
|
|
);
|
|
assert.equal(plan.revision, 4);
|
|
assert.match(plan.digest, /^[a-f0-9]{64}$/);
|
|
assert.deepEqual(
|
|
(await store.listRuns()).map((run) => run.runId),
|
|
["run-a", "run-b"],
|
|
);
|
|
});
|
|
});
|