feat: add worker execution state

Author: devkade

README.md

@@ -148,15 +148,15 @@ Kapi does not own subagent orchestration. Use `pi-subagents` for agent delegatio
 
 ### Graph Execution Runtime Boundary
 
-The `graph-execution` preset treats `TaskGraph` as the concrete execute-phase runtime primitive for single-agent, sequential, DAG-parallel, and team-parallel work. Policy selection can provide advisory `PolicyGraphSketch` inputs, but the execute phase owns concrete task ids, titles/descriptions, dependencies, topological ordering, readiness reasons, blocked/downstream status projection, claim/lease ownership, stale-claim recovery, evidence expectations, and task-graph/readiness/claim event records.
+The `graph-execution` preset treats `TaskGraph` as the concrete execute-phase runtime primitive for single-agent, sequential, DAG-parallel, and team-parallel work. Policy selection can provide advisory `PolicyGraphSketch` inputs, but the execute phase owns concrete task ids, titles/descriptions, dependencies, topological ordering, readiness reasons, blocked/downstream status projection, claim/lease ownership, worker dispatch state, heartbeat/staleness transitions, structured worker reports, stale-claim recovery, evidence expectations, and task-graph/readiness/claim event records.
 
 Phase presets serialize as a versioned `schemaVersion: 1` catalog. Legacy arrays migrate explicitly; unsupported versions, malformed catalogs, and gate evaluation with missing top-level required evidence refs or artifact ids fail closed.
 
 Agent execution stays adapter-neutral. The `AgentAdapterContract` describes required launch/send/capture/health/readiness/report/substrate behavior for Pi, Codex, and Claude Code compatible workers without coupling the domain layer to any one CLI. Readiness requires a nonce-equivalent proof, worker reports require `taskId`, `status`, `evidenceRefs`, and `summary`, and health can be supported or best-effort but not absent.
 
 The runtime state schema is separately versioned as `RuntimeState.schemaVersion: 1`. It defines additive boundaries for RunObjective, PolicySelection, TaskGraph refs, WorkerState, EvidenceRef, EvaluationResult, RewardRecord, and IntegrationCandidate data; unknown newer versions fail closed, and RunContract-facing artifact refs expose only objective, policy-selection, and evaluation artifacts.
 
-The domain boundary is explicit: `Decomposer` creates the concrete graph, `Scheduler` computes ready tasks, `WorkerRuntime` dispatches and heartbeats work, `Verifier` validates evidence refs, and `GateEngine` decides transitions.
+The domain boundary is explicit: `Decomposer` creates the concrete graph, `Scheduler` computes ready tasks, `WorkerRuntime` dispatches and heartbeats work, `WorkerExecutionState` records dispatch/report progress for claimed tasks, `Verifier` validates evidence refs, and `GateEngine` decides transitions.
 
 Runtime events are append-only `schemaVersion: 1` envelopes with monotonic `seq`, stable `eventId`, `idempotencyKey`, `type`, `category`, timestamp, run id, and a typed payload. The event taxonomy covers run lifecycle, objective/policy decisions, graph readiness, claim/lease transitions, worker readiness/heartbeat/retention/safe-close, evidence, evaluation, repair, integration, and reward records. Replay applies events by ascending sequence into derived state; duplicate event ids must be byte-equivalent, stale leases and missing workers recover only through explicit events, and `run.sealed` is terminal.
 

src/domain/runtime-state.ts

@@ -2,7 +2,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 RuntimeWorkerStatus = "ready" | "busy" | "unhealthy" | "completed-retained" | "safe-to-close" | "stale-registry";
+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";
 export type IntegrationCandidateStatus = "pending" | "accepted" | "rejected" | "repair_required";
@@ -36,7 +36,7 @@ export type RuntimeStateParseResult = { ok: true; state: RuntimeState } | { ok:
 
 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 workerStatuses = new Set<RuntimeWorkerStatus>(["ready", "busy", "unhealthy", "completed-retained", "safe-to-close", "stale-registry"]);
+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"]);
 const integrationStatuses = new Set<IntegrationCandidateStatus>(["pending", "accepted", "rejected", "repair_required"]);

src/domain/task-graph.ts

@@ -167,6 +167,34 @@ export interface WorkerRuntimeRef {
   id: string;
   state: WorkerRuntimeState;
   lastHeartbeatAt?: string;
+  readinessNonce?: string;
+}
+
+export type WorkerTaskReportStatus = "completed" | "failed" | "repair_required" | "blocked";
+
+export interface WorkerTaskDispatch {
+  taskId: string;
+  workerId: string;
+  claimToken: string;
+  dispatchedAt: string;
+  readinessNonce?: string;
+}
+
+export interface WorkerTaskReport {
+  taskId: string;
+  workerId: string;
+  claimToken: string;
+  status: WorkerTaskReportStatus;
+  summary: string;
+  evidenceRefs: string[];
+  reportedAt: string;
+}
+
+export interface WorkerExecutionState {
+  graph: TaskGraph;
+  workers: WorkerRuntimeRef[];
+  dispatches: WorkerTaskDispatch[];
+  reports: WorkerTaskReport[];
 }
 
 export function validateTaskGraph(graph: TaskGraph): TaskGraphValidationIssue[] {
@@ -362,6 +390,66 @@ export function getStaleWorkerIds(workers: WorkerRuntimeRef[], options: { now: s
     .map((worker) => worker.id);
 }
 
+export function recordWorkerReady(workers: WorkerRuntimeRef[], input: { workerId: string; now: string; readinessNonce?: string }): WorkerRuntimeRef[] {
+  parseTimestamp(input.now);
+  return upsertWorker(workers, {
+    id: input.workerId,
+    state: "ready",
+    lastHeartbeatAt: input.now,
+    ...(input.readinessNonce ? { readinessNonce: input.readinessNonce } : {}),
+  });
+}
+
+export function recordWorkerHeartbeat(workers: WorkerRuntimeRef[], input: { workerId: string; now: string }): WorkerRuntimeRef[] {
+  parseTimestamp(input.now);
+  return updateWorker(workers, input.workerId, (worker) => ({ ...worker, lastHeartbeatAt: input.now }));
+}
+
+export function markMissingHeartbeats(workers: WorkerRuntimeRef[], options: { now: string; heartbeatTimeoutSeconds: number }): WorkerRuntimeRef[] {
+  const stale = new Set(getStaleWorkerIds(workers, options));
+  return workers.map((worker) => (stale.has(worker.id) ? { ...worker, state: "unhealthy" } : worker));
+}
+
+export function dispatchClaimedTask(state: WorkerExecutionState, request: { taskId: string; workerId: string; claimToken: string; now: string; readinessNonce?: string }): WorkerExecutionState {
+  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`);
+  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}`);
+  const graph = updateTask(state.graph, task.id, (current) => ({ ...current, status: "in_progress" }));
+  return {
+    ...state,
+    graph,
+    workers: updateWorker(state.workers, worker.id, (current) => ({ ...current, state: "busy", lastHeartbeatAt: request.now })),
+    dispatches: [...state.dispatches, { taskId: task.id, workerId: worker.id, claimToken: request.claimToken, dispatchedAt: request.now, ...(request.readinessNonce ? { readinessNonce: request.readinessNonce } : {}) }],
+  };
+}
+
+export function captureWorkerReport(state: WorkerExecutionState, report: WorkerTaskReport): WorkerExecutionState {
+  validateWorkerReport(state, report);
+  const nextStatus = report.status === "completed" ? "verifying" : report.status;
+  return {
+    ...state,
+    graph: updateTask(state.graph, report.taskId, (task) => ({ ...task, status: nextStatus, evidenceRefs: [...report.evidenceRefs] })),
+    workers: updateWorker(state.workers, report.workerId, (worker) => ({ ...worker, state: report.status === "completed" ? "completed-retained" : "unhealthy", lastHeartbeatAt: report.reportedAt })),
+    reports: [...state.reports, { ...report, evidenceRefs: [...report.evidenceRefs] }],
+  };
+}
+
+export function completeTaskFromWorkerReport(state: WorkerExecutionState, request: { taskId: string; workerId: string; now: string; releaseWorker?: boolean }): WorkerExecutionState {
+  const report = [...state.reports].reverse().find((item) => item.taskId === request.taskId && item.workerId === request.workerId);
+  if (!report) throw new Error(`task ${request.taskId} has no worker report`);
+  if (report.status !== "completed") throw new Error(`task ${request.taskId} report is not completed`);
+  if (report.evidenceRefs.length === 0) throw new Error(`task ${request.taskId} report requires evidence refs`);
+  return {
+    ...state,
+    graph: completeTask(state.graph, { taskId: request.taskId, claimToken: report.claimToken, evidenceRefs: report.evidenceRefs, now: request.now }),
+    workers: updateWorker(state.workers, request.workerId, (worker) => markWorkerTerminal(worker, { retained: !request.releaseWorker })),
+  };
+}
+
 export function markWorkerStale(worker: WorkerRuntimeRef): WorkerRuntimeRef {
   return { ...worker, state: "stale-registry" };
 }
@@ -375,6 +463,41 @@ function assertValidGraph(graph: TaskGraph): void {
   if (issues.length > 0) throw new Error(`invalid task graph: ${issues.map((issue) => issue.code).join(", ")}`);
 }
 
+function validateWorkerReport(state: WorkerExecutionState, report: WorkerTaskReport): void {
+  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 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}`);
+}
+
+function findTask(graph: TaskGraph, taskId: string): TaskNode {
+  const task = graph.tasks.find((item) => item.id === taskId);
+  if (!task) throw new Error(`unknown task ${taskId}`);
+  return task;
+}
+
+function findWorker(workers: WorkerRuntimeRef[], workerId: string): WorkerRuntimeRef {
+  const worker = workers.find((item) => item.id === workerId);
+  if (!worker) throw new Error(`unknown worker ${workerId}`);
+  return worker;
+}
+
+function upsertWorker(workers: WorkerRuntimeRef[], worker: WorkerRuntimeRef): WorkerRuntimeRef[] {
+  return workers.some((item) => item.id === worker.id) ? updateWorker(workers, worker.id, () => worker) : [...workers, worker];
+}
+
+function updateWorker(workers: WorkerRuntimeRef[], workerId: string, update: (worker: WorkerRuntimeRef) => WorkerRuntimeRef): WorkerRuntimeRef[] {
+  let found = false;
+  const next = workers.map((worker) => {
+    if (worker.id !== workerId) return worker;
+    found = true;
+    return update(worker);
+  });
+  if (!found) throw new Error(`unknown worker ${workerId}`);
+  return next;
+}
+
 function computeFailedBlockersByTask(graph: TaskGraph, failed: Set<string>): Map<string, string[]> {
   const tasksById = new Map(graph.tasks.map((task) => [task.id, task]));
   const memo = new Map<string, string[]>();

test/runtime-state.test.ts

@@ -48,8 +48,8 @@ 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: [] }] };
-  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" }] }], integrationCandidates: [{ id: "candidate", taskIds: ["build"], status: "repair_required", evidenceRefs: [] }] };
+  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 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

@@ -1,25 +1,32 @@
 import assert from "node:assert/strict";
 import { test } from "node:test";
 import {
+  captureWorkerReport,
   claimTask,
   completeTask,
+  completeTaskFromWorkerReport,
   computeBlockedTaskIds,
   computeReadyTaskIds,
   computeTaskReadiness,
   createClaimLease,
   createClaimLeaseRuntimeEvent,
   createClaimLeaseToken,
   createTaskGraphRuntimeEvents,
+  dispatchClaimedTask,
   getStaleWorkerIds,
+  markMissingHeartbeats,
   markWorkerStale,
   markWorkerTerminal,
   projectTaskGraphStatus,
   recoverExpiredClaim,
+  recordWorkerHeartbeat,
+  recordWorkerReady,
   releaseClaim,
   renewClaimLease,
   topologicalTaskIds,
   transitionReadyTasks,
   validateTaskGraph,
+  type WorkerExecutionState,
 } from "../src/domain/task-graph.js";
 
 test("task graph readiness waits for completed dependencies", () => {
@@ -215,6 +222,42 @@ 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 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/);
+  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"]);
+  assert.deepEqual(state.workers.map((worker) => worker.state), ["busy", "busy"]);
+
+  state = { ...state, workers: recordWorkerHeartbeat(state.workers, { workerId: "w1", now: "2026-01-01T00:01:00.000Z" }) };
+  assert.equal(state.workers[0]?.lastHeartbeatAt, "2026-01-01T00:01:00.000Z");
+  assert.throws(
+    () => captureWorkerReport(state, { taskId: "branch-a", workerId: "w1", claimToken: "tok-a", status: "completed", summary: "done", evidenceRefs: [], reportedAt: "2026-01-01T00:02:00.000Z" }),
+    /evidence refs/,
+  );
+
+  state = captureWorkerReport(state, { taskId: "branch-a", workerId: "w1", claimToken: "tok-a", status: "completed", summary: "done", evidenceRefs: ["logs/a.txt", "verify/a.md"], reportedAt: "2026-01-01T00:02:00.000Z" });
+  assert.equal(state.graph.tasks[0]?.status, "verifying");
+  state = completeTaskFromWorkerReport(state, { taskId: "branch-a", workerId: "w1", now: "2026-01-01T00:03:00.000Z", releaseWorker: true });
+  assert.equal(state.graph.tasks[0]?.status, "completed");
+  assert.deepEqual(state.graph.tasks[0]?.evidenceRefs, ["logs/a.txt", "verify/a.md"]);
+  assert.equal(state.workers[0]?.state, "safe-to-close");
+
+  const stale = markMissingHeartbeats(state.workers, { now: "2026-01-01T00:30:00.000Z", heartbeatTimeoutSeconds: 60 });
+  assert.equal(stale.find((worker) => worker.id === "w2")?.state, "unhealthy");
+});
+
 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" },