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

src/node/services/agentWorkflowRunReferences.test.ts

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

@@ -75,7 +75,10 @@ export async function recordAgentWorkflowRunReference(input: {
     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/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,102 @@ 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);
+      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) => {

src/node/services/workspaceService.ts

@@ -3,6 +3,7 @@ import { raceWithAbortAndTimeout } from "@/node/utils/concurrency/withTimeout";
 import { EventEmitter } from "events";
 import * as path from "path";
 import { acquireCrossProcessLock } from "@/node/utils/main/crossProcessLock";
+import { readAgentWorkflowRunReferences } from "@/node/services/agentWorkflowRunReferences";
 import * as fsPromises from "fs/promises";
 import assert from "@/common/utils/assert";
 import { DEFAULT_WORKTREE_ARCHIVE_BEHAVIOR } from "@/common/config/worktreeArchiveBehavior";
@@ -10935,35 +10936,29 @@ export class WorkspaceService extends EventEmitter {
     assert(workspaceId.length > 0, "isWorkflowInvocationCurrent requires workspaceId");
     assert(runId.length > 0, "isWorkflowInvocationCurrent requires runId");
 
-    let current = false;
-    let foundDecision = false;
+    let outcome: "invocation" | "consumed" | "superseded" | null = null;
+    let supersededAtMs: number | null = null;
     const historyResult = await this.historyService.iterateFullHistory(
       workspaceId,
       "backward",
       (messages) => {
         for (const message of messages) {
-          if (isManualUserSupersessionMessage(message)) {
-            current = false;
-            foundDecision = true;
-            return false;
-          }
-          if (isResetBoundaryMessage(message)) {
-            current = false;
-            foundDecision = true;
+          if (isManualUserSupersessionMessage(message) || isResetBoundaryMessage(message)) {
+            outcome = "superseded";
+            const timestamp = message.metadata?.timestamp;
+            supersededAtMs = typeof timestamp === "number" ? timestamp : null;
             return false;
           }
           if (
             isWorkflowResultContinuationMessage(message, runId) ||
             isTerminalWorkflowTaskAwaitResultMessage(message, runId) ||
             isTerminalWorkflowToolResultMessage(message, runId)
           ) {
-            current = false;
-            foundDecision = true;
+            outcome = "consumed";
             return false;
           }
           if (isWorkflowInvocationMessage(message, runId)) {
-            current = true;
-            foundDecision = true;
+            outcome = "invocation";
             return false;
           }
         }
@@ -10979,7 +10974,29 @@ export class WorkspaceService extends EventEmitter {
       return false;
     }
 
-    return foundDecision && current;
+    if (outcome === "invocation") {
+      return true;
+    }
+    if (outcome === "consumed") {
+      return false;
+    }
+
+    // Kernel-launched runs (mux.workflow_run / mux.workflow_resume inside code_execution) leave
+    // no recognizable invocation part in history, so the backward walk above stops at the prior
+    // real user message and would wrongly treat the run as superseded, silently dropping its
+    // notify_on_terminal wake. Their durable provenance is the agent-workflow-runs sidecar: a
+    // reference recorded after the latest supersession boundary counts as the current
+    // invocation. A boundary without a durable timestamp fails safe to superseded, mirroring
+    // TaskService.listAgentReferencedWorkflowRunIds.
+    if (outcome === "superseded" && supersededAtMs === null) {
+      return false;
+    }
+    const references = await readAgentWorkflowRunReferences(this.config.getSessionDir(workspaceId));
+    const reference = references.find((candidate) => candidate.runId === runId);
+    if (reference == null) {
+      return false;
+    }
+    return supersededAtMs === null || reference.createdAtMs > supersededAtMs;
   }
 
   /**