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

src/node/services/agentWorkflowRunReferences.test.ts

@@ -31,4 +31,36 @@ describe("agent workflow run references", () => {
       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

@@ -30,6 +30,7 @@ function parseReferences(value: unknown): AgentWorkflowRunReference[] {
   }
 
   const parsed: AgentWorkflowRunReference[] = [];
+  const now = Date.now();
   for (const reference of references) {
     if (reference == null || typeof reference !== "object") {
       continue;
@@ -41,7 +42,10 @@ function parseReferences(value: unknown): AgentWorkflowRunReference[] {
     if (typeof record.createdAtMs !== "number" || !Number.isFinite(record.createdAtMs)) {
       continue;
     }
-    parsed.push({ runId: record.runId, createdAtMs: record.createdAtMs });
+    // Self-heal implausible future timestamps (clock correction, corruption): a future-dated
+    // reference would otherwise outrank every later user/reset boundary in supersession
+    // comparisons until wall time catches up.
+    parsed.push({ runId: record.runId, createdAtMs: Math.min(record.createdAtMs, now) });
   }
   return parsed;
 }
@@ -71,11 +75,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,
         {

src/node/services/workspaceService.test.ts

@@ -54,9 +54,11 @@ import type { SendMessageOptions, WorkspaceChatMessage } from "@/common/orpc/typ
 import { createMuxMessage } from "@/common/types/message";
 import { buildStagedAttachmentNotice } from "@/browser/features/ChatInput/stagedAttachments";
 import {
+  WORKFLOW_RESULT_METADATA_TYPE,
   WORKFLOW_RUN_CARD_DISPLAY_METADATA_TYPE,
   WORKFLOW_TRIGGER_DISPLAY_METADATA_TYPE,
 } from "@/common/utils/workflowRunMessages";
+import { recordAgentWorkflowRunReference } from "@/node/services/agentWorkflowRunReferences";
 import { getPlanFilePath } from "@/common/utils/planStorage";
 import * as todoStorageModule from "@/node/services/todos/todoStorage";
 import * as runtimeFactory from "@/node/runtime/runtimeFactory";
@@ -5924,6 +5926,111 @@ describe("WorkspaceService workflow invocation events", () => {
     }
   });
 
+  test("counts a kernel-launched run recorded in the sidecar as the current invocation", async () => {
+    const { config, historyService, cleanup } = await createTestHistoryService();
+    const workspaceId = "workflow-currentness-kernel";
+    const runId = "wfr_currentness_kernel";
+    const projectPath = path.join(config.rootDir, "project");
+    try {
+      await config.addWorkspace(projectPath, {
+        id: workspaceId,
+        name: "workflow-currentness-kernel",
+        projectName: "project",
+        projectPath,
+        runtimeConfig: { type: "local" },
+      });
+      const workspaceService = createWorkspaceServiceForTest({
+        config,
+        historyService,
+        aiService: createMockAIService({
+          stopStream: mock(() => Promise.resolve(Ok(undefined))),
+        }),
+        extensionMetadata: new ExtensionMetadataService(
+          path.join(config.rootDir, "extensionMetadata.json")
+        ),
+        initStateManager: {
+          ...mockInitStateManager,
+          off: mock(() => undefined as unknown as InitStateManager),
+        } as unknown as InitStateManager,
+      });
+
+      // mux.workflow_run inside code_execution leaves no workflow_run tool part in history; the
+      // agent-workflow-runs sidecar reference is the only durable invocation evidence.
+      await historyService.appendToHistory(
+        workspaceId,
+        createMuxMessage("manual-user", "user", "run the audit workflow", { timestamp: 1_000 })
+      );
+      await historyService.appendToHistory(
+        workspaceId,
+        createMuxMessage("assistant-kernel-launch", "assistant", "", { timestamp: 1_100 }, [
+          {
+            type: "dynamic-tool",
+            toolCallId: "code-exec-1",
+            toolName: "code_execution",
+            state: "output-available",
+            input: { code: "return xum.workflow_run({ script_path: './workflows/demo.js' })" },
+            output: { success: true, result: { status: "running", runId } },
+          },
+        ])
+      );
+
+      // The nested runId in the code_execution output alone is not invocation evidence.
+      expect(await workspaceService.isWorkflowInvocationCurrent(workspaceId, runId)).toBe(false);
+
+      await recordAgentWorkflowRunReference({
+        workspaceSessionDir: config.getSessionDir(workspaceId),
+        runId,
+        createdAtMs: 1_150,
+      });
+      expect(await workspaceService.isWorkflowInvocationCurrent(workspaceId, runId)).toBe(true);
+
+      // A newer manual user message supersedes the sidecar reference.
+      await historyService.appendToHistory(
+        workspaceId,
+        createMuxMessage("manual-user-2", "user", "never mind, answer something else", {
+          timestamp: 1_200,
+        })
+      );
+      expect(await workspaceService.isWorkflowInvocationCurrent(workspaceId, runId)).toBe(false);
+
+      // A kernel workflow_resume re-records the reference after the supersession and
+      // re-establishes provenance (latest record wins).
+      await recordAgentWorkflowRunReference({
+        workspaceSessionDir: config.getSessionDir(workspaceId),
+        runId,
+        createdAtMs: 1_250,
+      });
+      expect(await workspaceService.isWorkflowInvocationCurrent(workspaceId, runId)).toBe(true);
+
+      // Once the terminal result was delivered, the sidecar must not resurrect the invocation.
+      await historyService.appendToHistory(
+        workspaceId,
+        createMuxMessage("workflow-result", "user", "The workflow below has finished.", {
+          timestamp: 1_300,
+          synthetic: true,
+          muxMetadata: {
+            type: WORKFLOW_RESULT_METADATA_TYPE,
+            rawCommand: "workflow_run ./workflows/demo.js",
+            runId,
+          },
+        })
+      );
+      expect(await workspaceService.isWorkflowInvocationCurrent(workspaceId, runId)).toBe(false);
+
+      // A kernel background resume issued after the delivered result re-records the reference,
+      // so the retried run's next terminal wake must count as current again.
+      await recordAgentWorkflowRunReference({
+        workspaceSessionDir: config.getSessionDir(workspaceId),
+        runId,
+        createdAtMs: 1_350,
+      });
+      expect(await workspaceService.isWorkflowInvocationCurrent(workspaceId, runId)).toBe(true);
+      workspaceService.disposeSession(workspaceId);
+    } finally {
+      await cleanup();
+    }
+  });
+
   test.each(["workflow_run", "workflow_resume"] as const)(
     "treats terminal %s output as a consumed workflow result",
     async (toolName) => {