fix: reject stale worker reports

Author: devkade

src/domain/runtime-state.ts

@@ -1,7 +1,7 @@
 export const RUNTIME_STATE_SCHEMA_VERSION = 1;
 
 export type RuntimeStatus = "draft" | "active" | "blocked" | "failed" | "sealed";
-export type RuntimeTaskStatus = "pending" | "ready" | "claimed" | "verifying" | "completed" | "blocked" | "failed" | "repair_required";
+export type RuntimeTaskStatus = "pending" | "ready" | "claimed" | "in_progress" | "verifying" | "completed" | "blocked" | "failed" | "repair_required";
 export type RuntimeWorkerStatus = "ready" | "busy" | "unhealthy" | "completed-retained" | "safe-to-close" | "stale-registry" | "cleanup-released" | "closed";
 export type RuntimeArtifactKind = "run-objective" | "policy-selection" | "task-graph" | "worker-state" | "evidence" | "evaluation" | "reward" | "integration-candidate" | "final-report";
 export type EvaluationVerdict = "pass" | "fail" | "repair" | "blocked";
@@ -35,7 +35,7 @@ export interface RuntimeState {
 export type RuntimeStateParseResult = { ok: true; state: RuntimeState } | { ok: false; reason: "malformed" | "unsupported-newer-schema"; issues: string[] };
 
 const runtimeStatuses = new Set<RuntimeStatus>(["draft", "active", "blocked", "failed", "sealed"]);
-const taskStatuses = new Set<RuntimeTaskStatus>(["pending", "ready", "claimed", "verifying", "completed", "blocked", "failed", "repair_required"]);
+const taskStatuses = new Set<RuntimeTaskStatus>(["pending", "ready", "claimed", "in_progress", "verifying", "completed", "blocked", "failed", "repair_required"]);
 const workerStatuses = new Set<RuntimeWorkerStatus>(["ready", "busy", "unhealthy", "completed-retained", "safe-to-close", "stale-registry", "cleanup-released", "closed"]);
 const artifactKinds = new Set<RuntimeArtifactKind>(["run-objective", "policy-selection", "task-graph", "worker-state", "evidence", "evaluation", "reward", "integration-candidate", "final-report"]);
 const evaluationVerdicts = new Set<EvaluationVerdict>(["pass", "fail", "repair", "blocked"]);

src/domain/task-graph.ts

@@ -391,6 +391,8 @@ export function getStaleWorkerIds(workers: WorkerRuntimeRef[], options: { now: s
 }
 
 export function recordWorkerReady(workers: WorkerRuntimeRef[], input: { workerId: string; now: string; readinessNonce?: string }): WorkerRuntimeRef[] {
+  requireText(input.workerId, "workerId");
+  if (input.readinessNonce !== undefined) requireText(input.readinessNonce, "readinessNonce");
   parseTimestamp(input.now);
   return upsertWorker(workers, {
     id: input.workerId,
@@ -414,10 +416,11 @@ export function dispatchClaimedTask(state: WorkerExecutionState, request: { task
   const now = parseTimestamp(request.now);
   const worker = findWorker(state.workers, request.workerId);
   if (worker.state !== "ready") throw new Error(`worker ${worker.id} is not ready`);
-  if (request.readinessNonce && worker.readinessNonce !== request.readinessNonce) throw new Error(`worker ${worker.id} readiness nonce mismatch`);
+  if (worker.readinessNonce && request.readinessNonce !== worker.readinessNonce) throw new Error(`worker ${worker.id} readiness nonce mismatch`);
   const task = findTask(state.graph, request.taskId);
   const claim = requireActiveClaim(task, request.claimToken, now);
   if (claim.workerId !== request.workerId) throw new Error(`task ${task.id} is claimed by ${claim.workerId}`);
+  if (state.dispatches.some((dispatch) => dispatch.taskId === task.id && dispatch.claimToken === request.claimToken)) throw new Error(`task ${task.id} claim is already dispatched`);
   const graph = updateTask(state.graph, task.id, (current) => ({ ...current, status: "in_progress" }));
   return {
     ...state,
@@ -464,9 +467,12 @@ function assertValidGraph(graph: TaskGraph): void {
 }
 
 function validateWorkerReport(state: WorkerExecutionState, report: WorkerTaskReport): void {
-  parseTimestamp(report.reportedAt);
+  const reportedAt = parseTimestamp(report.reportedAt);
   if (!report.summary.trim()) throw new Error(`task ${report.taskId} report requires summary`);
   if (report.status === "completed" && report.evidenceRefs.length === 0) throw new Error(`task ${report.taskId} report requires evidence refs`);
+  const task = findTask(state.graph, report.taskId);
+  const claim = requireActiveClaim(task, report.claimToken, reportedAt);
+  if (claim.workerId !== report.workerId) throw new Error(`task ${report.taskId} is claimed by ${claim.workerId}`);
   const dispatch = state.dispatches.find((item) => item.taskId === report.taskId && item.workerId === report.workerId && item.claimToken === report.claimToken);
   if (!dispatch) throw new Error(`task ${report.taskId} was not dispatched to worker ${report.workerId}`);
 }
@@ -588,6 +594,10 @@ function findCycleTaskId(graph: TaskGraph): string | undefined {
   return undefined;
 }
 
+function requireText(value: string, label: string): void {
+  if (!value.trim()) throw new Error(`${label} is required`);
+}
+
 function parseTimestamp(value: string): number {
   const parsed = Date.parse(value);
   if (!Number.isFinite(parsed)) throw new Error(`invalid timestamp: ${value}`);

test/runtime-state.test.ts

@@ -48,7 +48,7 @@ test("run contract runtime refs expose objective, policy, and evaluation artifac
 });
 
 test("runtime state can represent successful and repair-required examples", () => {
-  const success: RuntimeState = { ...state(), status: "sealed", objective: { id: "objective", goal: "ship MVP", successCriteria: ["tests pass"], constraints: ["no destructive cleanup"] }, tasks: [{ id: "build", title: "Build slice", status: "completed", dependsOn: [], evidenceRefs: [{ kind: "evidence", artifactId: "verify", path: "verify.md" }], evaluationRefs: [] }], workers: [{ id: "w1", adapterId: "local", status: "closed", readinessNonce: "nonce", lastHeartbeatAt: now }] };
+  const success: RuntimeState = { ...state(), status: "sealed", objective: { id: "objective", goal: "ship MVP", successCriteria: ["tests pass"], constraints: ["no destructive cleanup"] }, tasks: [{ id: "build", title: "Build slice", status: "completed", dependsOn: [], evidenceRefs: [{ kind: "evidence", artifactId: "verify", path: "verify.md" }], evaluationRefs: [] }, { id: "parallel", title: "Parallel slice", status: "in_progress", dependsOn: [], evidenceRefs: [], evaluationRefs: [] }], workers: [{ id: "w1", adapterId: "local", status: "closed", readinessNonce: "nonce", lastHeartbeatAt: now }] };
   const repair: RuntimeState = { ...state(), status: "blocked", tasks: [{ id: "build", title: "Build slice", status: "repair_required", dependsOn: [], evidenceRefs: [], evaluationRefs: [{ kind: "evaluation", artifactId: "eval", path: "evaluation.json" }] }], workers: [{ id: "w2", adapterId: "local", status: "cleanup-released" }], integrationCandidates: [{ id: "candidate", taskIds: ["build"], status: "repair_required", evidenceRefs: [] }] };
   assert.equal(parseRuntimeState(success).ok, true);
   assert.equal(parseRuntimeState(repair).ok, true);

test/task-graph.test.ts

@@ -223,18 +223,14 @@ test("failed dependencies block downstream tasks unless a repair supersedes them
 });
 
 test("worker execution dispatches claimed parallel tasks and completes only from evidence reports", () => {
-  const graph = {
-    tasks: [
-      { id: "branch-a", status: "ready" as const, dependencies: [] },
-      { id: "branch-b", status: "ready" as const, dependencies: [] },
-    ],
-  };
+  const graph = { tasks: [{ id: "branch-a", status: "ready" as const, dependencies: [] }, { id: "branch-b", status: "ready" as const, dependencies: [] }] };
   const claimedA = claimTask(graph, { taskId: "branch-a", workerId: "w1", token: "tok-a", now: "2026-01-01T00:00:00.000Z", leaseExpiresAt: "2026-01-01T00:20:00.000Z" });
   const claimed = claimTask(claimedA, { taskId: "branch-b", workerId: "w2", token: "tok-b", now: "2026-01-01T00:00:00.000Z", leaseExpiresAt: "2026-01-01T00:20:00.000Z" });
   const workers = recordWorkerReady(recordWorkerReady([], { workerId: "w1", now: "2026-01-01T00:00:01.000Z", readinessNonce: "n1" }), { workerId: "w2", now: "2026-01-01T00:00:01.000Z", readinessNonce: "n2" });
   let state: WorkerExecutionState = { graph: claimed, workers, dispatches: [], reports: [] };
 
   assert.throws(() => dispatchClaimedTask(state, { taskId: "branch-a", workerId: "w1", claimToken: "tok-a", now: "2026-01-01T00:00:02.000Z", readinessNonce: "wrong" }), /nonce/);
+  assert.throws(() => dispatchClaimedTask(state, { taskId: "branch-a", workerId: "w1", claimToken: "tok-a", now: "2026-01-01T00:00:02.000Z" }), /nonce/);
   state = dispatchClaimedTask(state, { taskId: "branch-a", workerId: "w1", claimToken: "tok-a", now: "2026-01-01T00:00:02.000Z", readinessNonce: "n1" });
   state = dispatchClaimedTask(state, { taskId: "branch-b", workerId: "w2", claimToken: "tok-b", now: "2026-01-01T00:00:02.000Z", readinessNonce: "n2" });
   assert.deepEqual(state.graph.tasks.map((task) => task.status), ["in_progress", "in_progress"]);
@@ -258,6 +254,24 @@ test("worker execution dispatches claimed parallel tasks and completes only from
   assert.equal(stale.find((worker) => worker.id === "w2")?.state, "unhealthy");
 });
 
+test("stale worker reports cannot mutate expired or recovered claims", () => {
+  const state: WorkerExecutionState = {
+    graph: { tasks: [{ id: "build", status: "in_progress" as const, dependencies: [], assignedWorker: "w1", claim: { token: "old", workerId: "w1", claimedAt: "2026-01-01T00:00:00.000Z", expiresAt: "2026-01-01T00:10:00.000Z" } }] },
+    workers: [{ id: "w1", state: "busy", lastHeartbeatAt: "2026-01-01T00:01:00.000Z" }],
+    dispatches: [{ taskId: "build", workerId: "w1", claimToken: "old", dispatchedAt: "2026-01-01T00:01:00.000Z" }],
+    reports: [],
+  };
+  const report = { taskId: "build", workerId: "w1", claimToken: "old", status: "completed" as const, summary: "late", evidenceRefs: ["verify.md#late"], reportedAt: "2026-01-01T00:11:00.000Z" };
+
+  assert.throws(() => captureWorkerReport(state, report), /expired/);
+  assert.equal(state.graph.tasks[0]?.status, "in_progress");
+  assert.equal(state.workers[0]?.state, "busy");
+  assert.deepEqual(state.reports, []);
+
+  const recovered = recoverExpiredClaim(state.graph, { taskId: "build", workerId: "w2", token: "new", now: "2026-01-01T00:11:00.000Z", leaseExpiresAt: "2026-01-01T00:20:00.000Z" });
+  assert.throws(() => captureWorkerReport({ ...state, graph: recovered }, { ...report, reportedAt: "2026-01-01T00:12:00.000Z" }), /matching claim token/);
+});
+
 test("worker heartbeat stale detection and terminal retention are modeled", () => {
   const workers = [
     { id: "fresh", state: "busy" as const, lastHeartbeatAt: "2026-01-01T00:04:30.000Z" },