workflow: isolate decisions and resync agent tasks

Author: reoring

src/agents/coordinator.ts

@@ -29,6 +29,8 @@ import {
   inboxDead,
 } from "./paths.js";
 import { ResidentAgent } from "./resident-agent.js";
+import { syncWorkflowStatusToCurrentState } from "./status-store.js";
+import { syncTasksToCurrentState } from "./task-store.js";
 import {
   DEFAULT_RECONCILE_INTERVAL_MS,
   DEFAULT_MAX_TEAMMATES,
@@ -162,10 +164,15 @@ export class AgentCoordinator {
     // Spawn teammate worker tasks for non-lead members
     await this.spawnTeammates();
 
+    await syncTasksToCurrentState(this.contextDir).catch(() => {});
+    await syncWorkflowStatusToCurrentState(this.contextDir, this.leadMemberId ?? "lead").catch(() => {});
+
     // Start periodic housekeeping (mailbox + tasks)
     this.housekeepTimer = setInterval(async () => {
       if (this.stopped) return;
       try {
+        await syncTasksToCurrentState(this.contextDir);
+        await syncWorkflowStatusToCurrentState(this.contextDir, this.leadMemberId ?? "lead");
         const mailboxResult = await housekeepMailbox({ contextDir: this.contextDir });
         const taskResult = await housekeepTasksInProgress({ contextDir: this.contextDir });
         // Emit warnings for dead-lettered messages and requeued tasks

src/agents/current-state-phase.ts

@@ -20,6 +20,15 @@ export interface CurrentStatePhaseSnapshot {
   mtimeMs: number;
 }
 
+export interface CurrentStateRoutingSnapshot extends CurrentStatePhaseSnapshot {
+  runId: string | null;
+  issueId: string | null;
+  contractId: string | null;
+  workspaceStatusFingerprint: string | null;
+  proofPacketId: string | null;
+  issuePath: string | null;
+}
+
 function assertRecord(value: unknown, label: string): Record<string, unknown> {
   if (!value || typeof value !== "object" || Array.isArray(value)) {
     throw new Error(`${label} must be an object`);
@@ -54,6 +63,57 @@ export async function readCurrentStatePhaseSnapshot(
   };
 }
 
+function optionalString(record: Record<string, unknown>, ...keys: string[]): string | null {
+  for (const key of keys) {
+    const value = record[key];
+    if (typeof value === "string" && value.trim()) {
+      return value.trim();
+    }
+  }
+  return null;
+}
+
+export async function readCurrentStateRoutingSnapshot(
+  contextDir: string,
+  sourcePath: string,
+): Promise<CurrentStateRoutingSnapshot> {
+  const resolvedPath = resolveCurrentStatePhaseSourcePath(contextDir, sourcePath);
+  const [raw, sourceStat] = await Promise.all([
+    readFile(resolvedPath, "utf-8"),
+    stat(resolvedPath),
+  ]);
+  const data = assertRecord(JSON.parse(raw), "current-state.json");
+  const phase = data.phase;
+  if (typeof phase !== "string" || !phase.trim()) {
+    throw new Error(`current-state phase source missing string phase: ${resolvedPath}`);
+  }
+  const phaseReason = typeof data.phase_reason === "string" ? data.phase_reason.trim() : "";
+  return {
+    phase: phase.trim(),
+    phaseReason,
+    sourcePath: resolvedPath,
+    mtimeMs: sourceStat.mtimeMs,
+    runId: optionalString(data, "run_id"),
+    issueId: optionalString(data, "canonical_issue_id", "issue_id"),
+    contractId: optionalString(data, "contract_id"),
+    workspaceStatusFingerprint: optionalString(data, "workspace_status_fingerprint"),
+    proofPacketId: optionalString(data, "proof_packet_id"),
+    issuePath: optionalString(data, "issue_path"),
+  };
+}
+
+export function currentStateRoutingKey(snapshot: CurrentStateRoutingSnapshot): string {
+  return JSON.stringify({
+    source_path: snapshot.sourcePath,
+    run_id: snapshot.runId,
+    issue_id: snapshot.issueId,
+    contract_id: snapshot.contractId,
+    workspace_status_fingerprint: snapshot.workspaceStatusFingerprint,
+    proof_packet_id: snapshot.proofPacketId,
+    issue_path: snapshot.issuePath,
+  });
+}
+
 export function workflowStatusSummaryForPhase(phase: string): string {
   switch (phase) {
     case "awaiting-reviewer-fast-gates":

src/agents/housekeeping.ts

@@ -20,7 +20,7 @@ import {
 import { DEFAULT_PROCESSING_TTL_MS, MAX_DELIVERY_ATTEMPTS } from "./constants.js";
 import { readMembers } from "./store.js";
 import type { AgentMessage, AgentTask } from "./types.js";
-import { reconcilePhaseGuardedTasks } from "./task-store.js";
+import { syncTasksToCurrentState } from "./task-store.js";
 
 export interface HousekeepResult {
   requeued: number;
@@ -194,10 +194,16 @@ export async function housekeepTasksInProgress(
   const result: TaskHousekeepResult = { requeued: 0, warnings: [] };
   const now = Date.now();
 
-  const phaseTransitions = await reconcilePhaseGuardedTasks(opts.contextDir);
+  const routingSync = await syncTasksToCurrentState(opts.contextDir);
+  const phaseTransitions = routingSync.phaseTransitions;
   if (phaseTransitions > 0) {
     result.warnings.push(`Reconciled ${phaseTransitions} phase-guarded task transition(s)`);
   }
+  if (routingSync.added > 0 || routingSync.superseded > 0) {
+    result.warnings.push(
+      `Synced ${routingSync.added} routed task(s), superseded ${routingSync.superseded} stale task(s)`,
+    );
+  }
 
   const inProgressDir = tasksStatusDir(opts.contextDir, "in_progress");
   const pendingDir = tasksStatusDir(opts.contextDir, "pending");

src/agents/paths.ts

@@ -103,6 +103,10 @@ export function tasksEventsPath(contextDir: string): string {
   return resolve(tasksRoot(contextDir), "_events.jsonl");
 }
 
+export function taskTemplatesPath(contextDir: string): string {
+  return resolve(agentsRoot(contextDir), "task-templates.json");
+}
+
 // -- Inbox summary --------------------------------------------------------
 
 export function inboxSummaryPath(contextDir: string): string {

src/agents/status-store.ts

@@ -9,6 +9,7 @@ import {
   workflowStatusNextActionsForPhase,
   workflowStatusSummaryForPhase,
 } from "./current-state-phase.js";
+import { listTasks, readTaskTemplates } from "./task-store.js";
 
 export interface WorkflowStatusSummary {
   version: "1";
@@ -73,41 +74,73 @@ function buildDerivedWorkflowStatus(
   };
 }
 
-export async function readWorkflowStatus(
+async function detectWorkflowStatusSourcePath(
   contextDir: string,
+  current: WorkflowStatusSummary | null,
+): Promise<string | null> {
+  if (current?.source?.kind === "current_state_phase_v1" && current.source.path.trim()) {
+    return current.source.path;
+  }
+  const templates = await readTaskTemplates(contextDir);
+  const templateSource = templates.find((template) => template.phase_guard?.source_path)?.phase_guard?.source_path;
+  if (templateSource) return templateSource;
+
+  const tasks = await listTasks(contextDir);
+  const taskSource = tasks.find((task) => task.phase_guard?.source_path)?.phase_guard?.source_path;
+  return taskSource ?? null;
+}
+
+export async function syncWorkflowStatusToCurrentState(
+  contextDir: string,
+  ownerMemberId = "lead",
+  sourcePath?: string | null,
 ): Promise<WorkflowStatusSummary | null> {
+  let current: WorkflowStatusSummary | null = null;
   try {
     const raw = await readFile(workflowStatusPath(contextDir), "utf-8");
     const data = JSON.parse(raw);
     assertWorkflowStatusSummary(data);
-    if (data.source?.kind === "current_state_phase_v1") {
-      try {
-        const snapshot = await readCurrentStatePhaseSnapshot(contextDir, data.source.path);
-        const derived = buildDerivedWorkflowStatus(data.owner_member_id, snapshot);
-        const sourceMatches = data.source.path === derived.source?.path
-          && data.source.mtime_ms === derived.source?.mtime_ms;
-        if (
-          !sourceMatches
-          || data.summary !== derived.summary
-          || !arraysEqual(data.blockers, derived.blockers)
-          || !arraysEqual(data.next_actions, derived.next_actions)
-        ) {
-          const root = agentsRoot(contextDir);
-          const tmpDir = resolve(root, "tmp");
-          await mkdir(dirname(workflowStatusPath(contextDir)), { recursive: true });
-          await atomicJsonWrite(tmpDir, workflowStatusPath(contextDir), derived);
-          return derived;
-        }
-      } catch {
-        // Fall back to the persisted summary if the source is temporarily unreadable.
-      }
+    current = data;
+  } catch {
+    current = null;
+  }
+
+  const resolvedSourcePath = sourcePath ?? await detectWorkflowStatusSourcePath(contextDir, current);
+  if (!resolvedSourcePath) {
+    return current;
+  }
+
+  try {
+    const snapshot = await readCurrentStatePhaseSnapshot(contextDir, resolvedSourcePath);
+    const derived = buildDerivedWorkflowStatus(current?.owner_member_id ?? ownerMemberId, snapshot);
+    if (
+      current
+      && current.source?.kind === "current_state_phase_v1"
+      && current.source.path === derived.source?.path
+      && current.source.mtime_ms === derived.source?.mtime_ms
+      && current.summary === derived.summary
+      && arraysEqual(current.blockers, derived.blockers)
+      && arraysEqual(current.next_actions, derived.next_actions)
+    ) {
+      return current;
     }
-    return data;
+
+    const root = agentsRoot(contextDir);
+    const tmpDir = resolve(root, "tmp");
+    await mkdir(dirname(workflowStatusPath(contextDir)), { recursive: true });
+    await atomicJsonWrite(tmpDir, workflowStatusPath(contextDir), derived);
+    return derived;
   } catch {
-    return null;
+    return current;
   }
 }
 
+export async function readWorkflowStatus(
+  contextDir: string,
+): Promise<WorkflowStatusSummary | null> {
+  return await syncWorkflowStatusToCurrentState(contextDir);
+}
+
 export async function writeWorkflowStatus(
   opts: WriteWorkflowStatusOptions,
 ): Promise<WorkflowStatusSummary> {