🤖 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

@@ -13,6 +13,11 @@ export interface AgentWorkflowRunReference {
 
 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 +34,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 +47,23 @@ 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;
+    }
+    // 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.
+    const existing = parsedByRunId.get(record.runId);
+    if (existing == null || record.createdAtMs > existing.createdAtMs) {
+      parsedByRunId.set(record.runId, { runId: record.runId, createdAtMs: record.createdAtMs });
+    }
   }
-  return parsed;
+  return Array.from(parsedByRunId.values());
 }
 
 export async function readAgentWorkflowRunReferences(
@@ -71,11 +91,15 @@ export async function recordAgentWorkflowRunReference(input: {
   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 (isWorkflowInvocationCurrent, listAgentReferencedWorkflowRunIds).
+      createdAtMs: previous ? Math.max(previous.createdAtMs, createdAtMs) : createdAtMs,
     });
 
     await fs.mkdir(path.dirname(filePath), { recursive: true });

src/node/services/taskService.ts

@@ -8232,6 +8232,11 @@ export class TaskService {
         notification.sourceId
       );
       if (workflowPrompt == null) {
+        // Dropping a notify_on_terminal wake strands the run's owner; keep the drop diagnosable.
+        log.warn("Dropping superseded workflow terminal attention", {
+          ownerWorkspaceId,
+          runId: notification.sourceId,
+        });
         await this.terminalAttentionStore.markSuperseded(ownerWorkspaceId, notification.id);
         continue;
       }

src/node/services/tools/workflow_resume.test.ts

@@ -6,6 +6,7 @@ import { TestTempDir, createTestToolConfig } from "./testHelpers";
 import { readAgentWorkflowRunReferences } from "@/node/services/agentWorkflowRunReferences";
 import { WORKFLOW_CHECKPOINT_RETRY_ERROR_MESSAGE } from "@/common/utils/workflowRetryEligibility";
 import type { WorkflowRunRecord } from "@/common/types/workflow";
+import type { TaskService } from "@/node/services/taskService";
 import type { WorkflowRunAttachedEvent } from "@/common/types/stream";
 
 const mockToolCallOptions: ToolExecutionOptions<unknown> = {
@@ -229,6 +230,105 @@ describe("workflow_resume tool", () => {
     });
   });
 
+  test("marks terminal attention consumed when returning an already-completed run's result", async () => {
+    using tempDir = new TestTempDir("test-workflow-resume-consumed");
+    const completedRun = buildRun({
+      status: "completed",
+      events: [
+        { sequence: 1, type: "status", at: "2026-05-29T00:00:00.000Z", status: "running" },
+        {
+          sequence: 2,
+          type: "result",
+          at: "2026-05-29T00:00:01.000Z",
+          result: { reportMarkdown: "already done" },
+        },
+        { sequence: 3, type: "status", at: "2026-05-29T00:00:01.000Z", status: "completed" },
+      ],
+    });
+    const workflowService = buildWorkflowService({ getRun: mock(async () => completedRun) });
+    const markWorkflowRunTerminalAttentionConsumed = mock(() => Promise.resolve());
+    const tool = createWorkflowResumeTool({
+      ...createTestToolConfig(tempDir.path, { workspaceId: "workspace-1" }),
+      trusted: true,
+      workflowService,
+      taskService: { markWorkflowRunTerminalAttentionConsumed } as unknown as TaskService,
+    });
+
+    await tool.execute!(
+      { run_id: "wfr_resume_me", run_in_background: false, mode: null },
+      mockToolCallOptions
+    );
+
+    expect(markWorkflowRunTerminalAttentionConsumed).toHaveBeenCalledWith({
+      ownerWorkspaceId: "workspace-1",
+      runId: "wfr_resume_me",
+      status: "completed",
+    });
+  });
+
+  test("does not mark terminal attention consumed for background dispatches", async () => {
+    using tempDir = new TestTempDir("test-workflow-resume-background-no-consume");
+    // The refresh after a background dispatch can still observe the stale pre-dispatch failed
+    // status; consuming it would tombstone the retried run's future terminal wake.
+    const workflowService = buildWorkflowService({ getRun: mock(async () => buildFailedRun()) });
+    const markWorkflowRunTerminalAttentionConsumed = mock(() => Promise.resolve());
+    const tool = createWorkflowResumeTool({
+      ...createTestToolConfig(tempDir.path, { workspaceId: "workspace-1" }),
+      trusted: true,
+      workflowService,
+      taskService: { markWorkflowRunTerminalAttentionConsumed } as unknown as TaskService,
+    });
+
+    await tool.execute!(
+      { run_id: "wfr_resume_me", run_in_background: true, mode: "retry_from_checkpoint" },
+      mockToolCallOptions
+    );
+
+    expect(markWorkflowRunTerminalAttentionConsumed).not.toHaveBeenCalled();
+  });
+
+  test("marks terminal attention consumed when a foreground retry finishes terminal", async () => {
+    using tempDir = new TestTempDir("test-workflow-resume-foreground-consume");
+    const failedRun = buildFailedRun();
+    const completedRun = buildRun({
+      status: "completed",
+      events: [
+        { sequence: 1, type: "status", at: "2026-05-29T00:00:00.000Z", status: "running" },
+        { sequence: 2, type: "status", at: "2026-05-29T00:00:02.000Z", status: "completed" },
+      ],
+    });
+    let getRunCalls = 0;
+    const workflowService = buildWorkflowService({
+      getRun: mock(async () => {
+        getRunCalls += 1;
+        return getRunCalls === 1 ? failedRun : completedRun;
+      }),
+      retryRunFromCheckpoint: mock(async () => ({
+        runId: "wfr_resume_me",
+        status: "completed" as const,
+        result: { reportMarkdown: "retried" },
+      })),
+    });
+    const markWorkflowRunTerminalAttentionConsumed = mock(() => Promise.resolve());
+    const tool = createWorkflowResumeTool({
+      ...createTestToolConfig(tempDir.path, { workspaceId: "workspace-1" }),
+      trusted: true,
+      workflowService,
+      taskService: { markWorkflowRunTerminalAttentionConsumed } as unknown as TaskService,
+    });
+
+    await tool.execute!(
+      { run_id: "wfr_resume_me", run_in_background: false, mode: "retry_from_checkpoint" },
+      mockToolCallOptions
+    );
+
+    expect(markWorkflowRunTerminalAttentionConsumed).toHaveBeenCalledWith({
+      ownerWorkspaceId: "workspace-1",
+      runId: "wfr_resume_me",
+      status: "completed",
+    });
+  });
+
   test("rejects default resume of a failed run with checkpoint retry guidance", async () => {
     using tempDir = new TestTempDir("test-workflow-resume-failed");
     const workflowService = buildWorkflowService({ getRun: mock(async () => buildFailedRun()) });

src/node/services/tools/workflow_resume.ts

@@ -2,7 +2,7 @@ import { tool } from "ai";
 
 import { getErrorMessage } from "@/common/utils/errors";
 import type { ToolConfiguration, ToolFactory } from "@/common/utils/tools/tools";
-import type { WorkflowRunRecord } from "@/common/types/workflow";
+import { isTerminalWorkflowRunStatus, type WorkflowRunRecord } from "@/common/types/workflow";
 import { getWorkflowCheckpointRetryEligibility } from "@/common/utils/workflowRetryEligibility";
 import { WorkflowRunRecordSchema } from "@/common/orpc/schemas";
 import {
@@ -154,9 +154,25 @@ export const createWorkflowResumeTool: ToolFactory = (config: ToolConfiguration)
       const mode: WorkflowResumeMode = args.mode ?? "resume";
       const invocationStartedAtMs = Date.now();
 
+      // A kernel-nested resume (mux.workflow_resume inside code_execution) leaves no top-level
+      // workflow_resume part in history, so the history-walk consumption predicates cannot see
+      // that this turn already received the terminal result. Persist consumption durably so the
+      // terminal-attention drain never re-delivers it.
+      const markTerminalAttentionConsumed = async (terminalRun: WorkflowRunRecord) => {
+        if (!isTerminalWorkflowRunStatus(terminalRun.status)) {
+          return;
+        }
+        await config.taskService?.markWorkflowRunTerminalAttentionConsumed?.({
+          ownerWorkspaceId: workspaceId,
+          runId: terminalRun.id,
+          status: terminalRun.status,
+        });
+      };
+
       // Idempotent success: the work is already done, so hand back the durable result instead
       // of failing the agent's recovery loop (e.g. resuming after a crash that actually finished).
       if (run.status === "completed" && mode === "resume") {
+        await markTerminalAttentionConsumed(run);
         return parseToolResult(
           WorkflowResumeToolResultSchema,
           {
@@ -231,6 +247,12 @@ export const createWorkflowResumeTool: ToolFactory = (config: ToolConfiguration)
       const refreshedRunIsStale =
         isBackgroundDispatch && refreshedRun != null && refreshedRun.status === run.status;
 
+      // Foreground only: a background dispatch can still observe the stale pre-dispatch terminal
+      // status, and consuming it would tombstone the retried run's future terminal wake.
+      if (!isBackgroundDispatch && refreshedRun != null) {
+        await markTerminalAttentionConsumed(refreshedRun);
+      }
+
       return parseToolResult(
         WorkflowResumeToolResultSchema,
         {