🤖 fix: deliver terminal wakes for kernel-launched background workflow runs

src/node/services/agentWorkflowRunReferences.test.ts

@@ -31,4 +31,118 @@ describe("agent workflow run references", () => {
       await fs.rm(workspaceSessionDir, { recursive: true, force: true });
     }
   });
+
+  test("collapses persisted duplicate entries to the newest sane timestamp", async () => {
+    const workspaceSessionDir = await fs.mkdtemp(path.join(os.tmpdir(), "agent-workflow-runs-"));
+    try {
+      // Corrupted files can carry duplicates in either order; order-sensitive consumers must
+      // never observe a stale duplicate ahead of a legitimate re-record.
+      await fs.writeFile(
+        path.join(workspaceSessionDir, "agent-workflow-runs.json"),
+        JSON.stringify({
+          references: [
+            { runId: "wfr_dup", createdAtMs: 1_000 },
+            { runId: "wfr_dup", createdAtMs: 2_000 },
+            { runId: "wfr_dup_reversed", createdAtMs: 2_000 },
+            { runId: "wfr_dup_reversed", createdAtMs: 1_000 },
+          ],
+        })
+      );
+
+      const references = await readAgentWorkflowRunReferences(workspaceSessionDir);
+      expect(references).toHaveLength(2);
+      expect(references).toContainEqual({ runId: "wfr_dup", createdAtMs: 2_000 });
+      expect(references).toContainEqual({ runId: "wfr_dup_reversed", createdAtMs: 2_000 });
+    } finally {
+      await fs.rm(workspaceSessionDir, { recursive: true, force: true });
+    }
+  });
+
+  test("drops persisted future-dated references and repairs them on the next record", async () => {
+    const workspaceSessionDir = await fs.mkdtemp(path.join(os.tmpdir(), "agent-workflow-runs-"));
+    try {
+      const futureMs = Date.now() + 86_400_000;
+      await fs.writeFile(
+        path.join(workspaceSessionDir, "agent-workflow-runs.json"),
+        JSON.stringify({
+          references: [
+            { runId: "wfr_corrupt_future", createdAtMs: futureMs },
+            { runId: "wfr_sane", createdAtMs: 1_000 },
+          ],
+        })
+      );
+
+      // A per-read clamp would re-evaluate to "now" on every read and outrank every later
+      // user/reset boundary; the corrupted entry must be dropped instead.
+      expect(await readAgentWorkflowRunReferences(workspaceSessionDir)).toEqual([
+        { runId: "wfr_sane", createdAtMs: 1_000 },
+      ]);
+
+      await recordAgentWorkflowRunReference({
+        workspaceSessionDir,
+        runId: "wfr_corrupt_future",
+        createdAtMs: 2_000,
+      });
+      const repaired = await readAgentWorkflowRunReferences(workspaceSessionDir);
+      expect(repaired).toContainEqual({ runId: "wfr_corrupt_future", createdAtMs: 2_000 });
+      const raw = await fs.readFile(
+        path.join(workspaceSessionDir, "agent-workflow-runs.json"),
+        "utf-8"
+      );
+      expect(raw).not.toContain(String(futureMs));
+    } finally {
+      await fs.rm(workspaceSessionDir, { recursive: true, force: true });
+    }
+  });
+
+  test("keeps references within the backward-clock skew tolerance", async () => {
+    const workspaceSessionDir = await fs.mkdtemp(path.join(os.tmpdir(), "agent-workflow-runs-"));
+    try {
+      // A backward clock correction makes a legitimately recorded reference look slightly
+      // future-dated; dropping it would strand the run's terminal wake.
+      const slightlyFutureMs = Date.now() + 5 * 60_000;
+      await fs.writeFile(
+        path.join(workspaceSessionDir, "agent-workflow-runs.json"),
+        JSON.stringify({ references: [{ runId: "wfr_clock_skew", createdAtMs: slightlyFutureMs }] })
+      );
+
+      expect(await readAgentWorkflowRunReferences(workspaceSessionDir)).toEqual([
+        { runId: "wfr_clock_skew", createdAtMs: slightlyFutureMs },
+      ]);
+    } finally {
+      await fs.rm(workspaceSessionDir, { recursive: true, force: true });
+    }
+  });
+
+  test("clamps future-dated createdAtMs to the current time", async () => {
+    const workspaceSessionDir = await fs.mkdtemp(path.join(os.tmpdir(), "agent-workflow-runs-"));
+    try {
+      const runId = "wfr_future";
+      await recordAgentWorkflowRunReference({
+        workspaceSessionDir,
+        runId,
+        createdAtMs: Date.now() + 86_400_000,
+      });
+
+      const references = await readAgentWorkflowRunReferences(workspaceSessionDir);
+      expect(references).toHaveLength(1);
+      expect(references[0]?.createdAtMs).toBeLessThanOrEqual(Date.now());
+    } finally {
+      await fs.rm(workspaceSessionDir, { recursive: true, force: true });
+    }
+  });
+
+  test("keeps the newest createdAtMs across re-records", async () => {
+    const workspaceSessionDir = await fs.mkdtemp(path.join(os.tmpdir(), "agent-workflow-runs-"));
+    try {
+      const runId = "wfr_re_recorded";
+      await recordAgentWorkflowRunReference({ workspaceSessionDir, runId, createdAtMs: 2_000 });
+      await recordAgentWorkflowRunReference({ workspaceSessionDir, runId, createdAtMs: 1_000 });
+
+      const references = await readAgentWorkflowRunReferences(workspaceSessionDir);
+      expect(references).toEqual([{ runId, createdAtMs: 2_000 }]);
+    } finally {
+      await fs.rm(workspaceSessionDir, { recursive: true, force: true });
+    }
+  });
 });

src/node/services/agentWorkflowRunReferences.ts

@@ -9,10 +9,23 @@ import { MutexMap } from "@/node/utils/concurrency/mutexMap";
 export interface AgentWorkflowRunReference {
   runId: string;
   createdAtMs: number;
+  /**
+   * Message ID of the newest invocation-decision row (manual user/reset supersession, consumed
+   * terminal result for this run, or direct invocation part) at record time; null when history
+   * had none. Row identity, not wall clock: currentness compares this against the row the
+   * history walk stops at, so a backward clock correction cannot reorder the comparison.
+   * Absent on legacy entries, which fail safe to not-current.
+   */
+  afterBoundaryMessageId?: string | null;
 }
 
 const AGENT_WORKFLOW_RUN_REFERENCES_FILE = "agent-workflow-runs.json";
 
+// Backward clock corrections (e.g. an NTP step after booting with a fast clock) can make a
+// legitimately recorded reference look future-dated. Tolerate that bounded skew so the run's
+// terminal wake is not dropped; only implausibly future values are treated as corruption.
+const MAX_FUTURE_SKEW_MS = 60 * 60_000;
+
 const referenceFileLocks = new MutexMap<string>();
 
 function referencesPath(workspaceSessionDir: string): string {
@@ -29,7 +42,8 @@ function parseReferences(value: unknown): AgentWorkflowRunReference[] {
     return [];
   }
 
-  const parsed: AgentWorkflowRunReference[] = [];
+  const parsedByRunId = new Map<string, AgentWorkflowRunReference>();
+  const now = Date.now();
   for (const reference of references) {
     if (reference == null || typeof reference !== "object") {
       continue;
@@ -41,9 +55,34 @@ function parseReferences(value: unknown): AgentWorkflowRunReference[] {
     if (typeof record.createdAtMs !== "number" || !Number.isFinite(record.createdAtMs)) {
       continue;
     }
-    parsed.push({ runId: record.runId, createdAtMs: record.createdAtMs });
+    // Reject implausibly future-dated references (corruption) instead of clamping at read
+    // time: a per-read clamp re-evaluates to "now" on every read, so the entry would outrank
+    // every later user/reset boundary until wall time catches up. Values within
+    // MAX_FUTURE_SKEW_MS are kept as-is (backward clock correction, not corruption). Rejected
+    // entries are replaced with a sane timestamp by the next legitimate record.
+    if (record.createdAtMs > now + MAX_FUTURE_SKEW_MS) {
+      continue;
+    }
+    const boundaryRaw = record.afterBoundaryMessageId;
+    const afterBoundaryMessageId =
+      typeof boundaryRaw === "string" && boundaryRaw.length > 0
+        ? boundaryRaw
+        : boundaryRaw === null
+          ? null
+          : undefined;
+    // Collapse corrupted duplicate entries to the newest sane timestamp so order-sensitive
+    // consumers cannot pick a stale duplicate and declare a legitimately re-recorded run
+    // superseded. The chosen record is kept wholesale, including its boundary snapshot.
+    const existing = parsedByRunId.get(record.runId);
+    if (existing == null || record.createdAtMs > existing.createdAtMs) {
+      parsedByRunId.set(record.runId, {
+        runId: record.runId,
+        createdAtMs: record.createdAtMs,
+        ...(afterBoundaryMessageId !== undefined ? { afterBoundaryMessageId } : {}),
+      });
+    }
   }
-  return parsed;
+  return Array.from(parsedByRunId.values());
 }
 
 export async function readAgentWorkflowRunReferences(
@@ -64,18 +103,28 @@ export async function recordAgentWorkflowRunReference(input: {
   workspaceSessionDir: string;
   runId: string;
   createdAtMs?: number;
+  afterBoundaryMessageId?: string | null;
 }): Promise<void> {
   assert(input.runId.length > 0, "agent workflow reference requires runId");
   const filePath = referencesPath(input.workspaceSessionDir);
 
   await referenceFileLocks.withLock(filePath, async () => {
     const existing = await readAgentWorkflowRunReferences(input.workspaceSessionDir);
     const byRunId = new Map(existing.map((reference) => [reference.runId, reference]));
-    const createdAtMs = input.createdAtMs ?? Date.now();
+    // Clamp like parseReferences: never persist a future-dated timestamp.
+    const createdAtMs = Math.min(input.createdAtMs ?? Date.now(), Date.now());
     const previous = byRunId.get(input.runId);
     byRunId.set(input.runId, {
       runId: input.runId,
-      createdAtMs: previous ? Math.min(previous.createdAtMs, createdAtMs) : createdAtMs,
+      // Latest record wins: workflow_resume re-records the reference, and a resume issued after
+      // a manual user message must re-establish provenance for supersession-timestamp
+      // comparisons (listAgentReferencedWorkflowRunIds).
+      createdAtMs: previous ? Math.max(previous.createdAtMs, createdAtMs) : createdAtMs,
+      // The new record event defines currentness provenance wholesale; a caller without
+      // boundary knowledge produces a legacy-style entry that fails safe to not-current.
+      ...(input.afterBoundaryMessageId !== undefined
+        ? { afterBoundaryMessageId: input.afterBoundaryMessageId }
+        : {}),
     });
 
     await fs.mkdir(path.dirname(filePath), { recursive: true });

src/node/services/taskService.test.ts

@@ -665,6 +665,14 @@ function createWorkspaceServiceMocks(
     mock((_workspaceId: string, _message: WorkspaceChatMessage) => undefined);
   const isWorkflowInvocationCurrent =
     overrides?.isWorkflowInvocationCurrent ?? mock(() => Promise.resolve(true));
+  // Derived from the boolean mock so tests that override isWorkflowInvocationCurrent keep
+  // steering the drain's three-state check.
+  const getWorkflowInvocationCurrentness = mock(
+    async (workspaceId: string, runId: string) =>
+      ((await isWorkflowInvocationCurrent(workspaceId, runId)) === true
+        ? "current"
+        : "not_current") as "current" | "not_current" | "indeterminate"
+  );
   const countQueuedAgentPeerMessages = overrides?.countQueuedAgentPeerMessages ?? mock(() => 0);
   // Granted by default (no live user activity): interrupt_active tests exercise the
   // interruption/archive flow; the hold's own refusal logic lives in workspaceService.test.ts.
@@ -728,6 +736,7 @@ function createWorkspaceServiceMocks(
       isExperimentEnabled,
       emitChatEvent,
       isWorkflowInvocationCurrent,
+      getWorkflowInvocationCurrentness,
       countQueuedAgentPeerMessages,
     } as unknown as WorkspaceService,
     create,
@@ -6239,6 +6248,50 @@ describe("TaskService", () => {
     expect(await terminalAttentionStore.listPending(parentId)).toHaveLength(0);
   });
 
+  test("terminal workflow wake-up defers when history is unreadable", async () => {
+    const config = await createTestConfig(rootDir);
+    const { parentId } = await saveLocalParentWorkspace(config, rootDir);
+    const runId = "wfr_terminal_defer";
+    const runStore = new WorkflowRunStore({ sessionDir: config.getSessionDir(parentId) });
+    await runStore.createRun({
+      id: runId,
+      workspaceId: parentId,
+      workflow: {
+        name: "research",
+        description: "Research workflow",
+        scope: "built-in",
+        executable: true,
+      },
+      source: "export default function workflow() { return { reportMarkdown: 'done' }; }\n",
+      args: {},
+      attentionPolicy: "notify_on_terminal",
+      now: "2026-06-19T00:00:00.000Z",
+    });
+    await runStore.appendStatus(runId, "running", "2026-06-19T00:00:01.000Z");
+    await runStore.appendStatus(runId, "completed", "2026-06-19T00:00:03.000Z");
+
+    const terminalAttentionStore = new TerminalAttentionStore(config);
+    const sendMessage = mock(
+      (..._args: unknown[]): Promise<Result<void>> => Promise.resolve(Ok(undefined))
+    );
+    const { workspaceService } = createWorkspaceServiceMocks({ sendMessage });
+    // History unreadable at drain time: currentness is indeterminate, so the notification must
+    // stay pending for a later drain instead of being tombstoned as superseded.
+    (workspaceService as unknown as Record<string, unknown>).getWorkflowInvocationCurrentness =
+      mock(() => Promise.resolve("indeterminate"));
+    const { taskService } = createTaskServiceHarness(config, { workspaceService });
+
+    await taskService.enqueueWorkflowRunTerminalAttention({
+      ownerWorkspaceId: parentId,
+      runId,
+      status: "completed",
+    });
+    await flushTerminalAttentionDrains(taskService);
+
+    expect(sendMessage).not.toHaveBeenCalled();
+    expect(await terminalAttentionStore.listPending(parentId)).toHaveLength(1);
+  });
+
   test("initialize replays and clears persisted pending task guidance", async () => {
     const config = await createTestConfig(rootDir);
     const projectPath = path.join(rootDir, "repo");