[task-service] 🤖 fix: prevent false workspace turn interruptions

src/node/services/agentSession.queueDispatch.test.ts

@@ -514,6 +514,41 @@ describe("AgentSession queued message tool-call dispatch", () => {
     }
   });
 
+  test("intentional dedupe cleanup can remove a synthetic entry without canceling its owner", async () => {
+    const workspaceId = "queue-dispatch-silent-dedupe-removal";
+    const { session, cleanup } = await createAgentSessionHarness({ workspaceId });
+
+    try {
+      const canceledReasons: string[] = [];
+      session.queueMessage(
+        "Incremental report",
+        { model: TEST_MODEL, agentId: "exec" },
+        {
+          synthetic: true,
+          agentInitiated: true,
+          dedupeKey: "agent-report:child:progress",
+          removableDedupeKey: true,
+          onCanceled: (reason) => {
+            canceledReasons.push(reason);
+          },
+        }
+      );
+
+      expect(
+        session.removeQueuedMessagesByDedupeKeyPrefix(
+          "agent-report:child:",
+          "Terminal report replaced progress.",
+          { notifyCancellation: false }
+        )
+      ).toBe(1);
+      expect(canceledReasons).toEqual([]);
+      expect(session.hasQueuedMessages()).toBe(false);
+    } finally {
+      session.dispose();
+      await cleanup();
+    }
+  });
+
   test("cancel signal retracts a synthetic entry after dequeue while history append is preparing", async () => {
     const workspaceId = "queue-dispatch-cancel-preparing";
     const { session, cleanup, historyService, events } = await createAgentSessionHarness({

src/node/services/agentSession.ts

@@ -49,6 +49,7 @@ import {
   createFileSnapshotMessageId,
   createAgentSkillSnapshotMessageId,
   createMcpPromptSnapshotMessageId,
+  FILE_CHANGE_NOTIFICATION_MESSAGE_ID_PREFIX,
 } from "@/node/services/utils/messageIds";
 import {
   FileChangeTracker,
@@ -346,12 +347,11 @@ function hasSameWorkspaceTurnCorrelation(
  * and the turn's real outcome can never settle the task handle (see
  * TaskService.finalizeWorkspaceTurnFromStreamEnd).
  *
- * Scans newest→oldest: interleaved monitor wakes keep the chain open; any
- * other user input (manual prompt, new workspace-turn prompt) supersedes the
- * turn, and only a correlated assistant message that ended with "tool-calls"
- * (a queue-dispatch cut) leaves the turn open. The inherited metadata is
- * persisted on each continuation's assistant message, so chains survive
- * restarts.
+ * Scans newest→oldest. Monitor wakes, file-change notices, and correlated
+ * nested reports keep the chain open. Manual prompts supersede the turn.
+ * Only a correlated assistant message that ended with "tool-calls" leaves the
+ * older chain open. Each continuation persists the inherited metadata, so the
+ * chain survives restarts.
  */
 export function inheritOpenWorkspaceTurnMetadata(
   messages: readonly MuxMessage[]
@@ -384,6 +384,19 @@ export function inheritOpenWorkspaceTurnMetadata(
       if (muxMetadata?.type === "bash-monitor-wake") {
         continue;
       }
+      // File-change rows are machine context for the pending continuation.
+      // They do not replace the delegated request that caused the stream.
+      if (
+        message.metadata?.synthetic === true &&
+        message.id.startsWith(FILE_CHANGE_NOTIFICATION_MESSAGE_ID_PREFIX)
+      ) {
+        continue;
+      }
+      // A correlated nested report can queue before a monitor wake. It continues
+      // the same delegated turn and is stronger evidence than the older cut.
+      if (muxMetadata?.type === "workspace-turn-task") {
+        return muxMetadata;
+      }
       return undefined;
     }
   }
@@ -5517,6 +5530,8 @@ export class AgentSession {
       onAccepted?: () => Promise<void> | void;
       onAcceptedPreStreamFailure?: (error: SendMessageError) => Promise<void> | void;
       onCanceled?: (reason: string) => Promise<void> | void;
+      onDeliveryAcceptedPreStreamFailure?: (error: SendMessageError) => Promise<void> | void;
+      onDeliveryCanceled?: (reason: string) => Promise<void> | void;
       cancelState?: { canceledBeforeAcceptance: boolean };
       cancelSignal?: AbortSignal;
     }
@@ -5590,7 +5605,11 @@ export class AgentSession {
     });
   }
 
-  removeQueuedMessagesByDedupeKeyPrefix(prefix: string, cancelReason: string): number {
+  removeQueuedMessagesByDedupeKeyPrefix(
+    prefix: string,
+    cancelReason: string,
+    options?: { notifyCancellation?: boolean }
+  ): number {
     this.assertNotDisposed("removeQueuedMessagesByDedupeKeyPrefix");
     assert(prefix.length > 0, "removeQueuedMessagesByDedupeKeyPrefix requires prefix");
     const removal = this.messageQueue.removeByDedupeKeyPrefix(prefix);
@@ -5602,8 +5621,10 @@ export class AgentSession {
       this.workspaceId,
       !this.messageQueue.isEmpty() && this.messageQueue.getNextQueueDispatchMode() === "tool-end"
     );
-    for (const callbacks of removal.callbacks) {
-      this.notifyQueuedMessageCleared(callbacks, cancelReason);
+    if (options?.notifyCancellation !== false) {
+      for (const callbacks of removal.callbacks) {
+        this.notifyQueuedMessageCleared(callbacks, cancelReason);
+      }
     }
     return removal.removedCount;
   }

src/node/services/agentSession.workspaceTurnInheritance.test.ts

@@ -48,6 +48,30 @@ describe("inheritOpenWorkspaceTurnMetadata", () => {
     expect(inheritOpenWorkspaceTurnMetadata(messages)).toEqual(correlation);
   });
 
+  test("a file-change row after a monitor wake keeps the turn open", () => {
+    const messages = [
+      turnPrompt("prompt"),
+      cutAssistant("cut"),
+      wake("wake"),
+      createMuxMessage("file-change-1", "user", "<system-file-update />", {
+        synthetic: true,
+      }),
+    ];
+    expect(inheritOpenWorkspaceTurnMetadata(messages)).toEqual(correlation);
+  });
+
+  test("a correlated nested report before a monitor wake keeps the turn open", () => {
+    const messages = [
+      turnPrompt("prompt"),
+      cutAssistant("cut"),
+      createMuxMessage("nested-report", "user", "Nested task completed", {
+        muxMetadata: correlation,
+      }),
+      wake("wake"),
+    ];
+    expect(inheritOpenWorkspaceTurnMetadata(messages)).toEqual(correlation);
+  });
+
   test("a correlated assistant that finished with stop closes the turn", () => {
     const messages = [
       turnPrompt("prompt"),

src/node/services/messageQueue.test.ts

@@ -1,6 +1,7 @@
-import { describe, it, expect, beforeEach } from "bun:test";
+import { describe, it, expect, beforeEach, mock } from "bun:test";
 import { MessageQueue } from "./messageQueue";
 import type { MuxMessageMetadata } from "@/common/types/message";
+import type { SendMessageError } from "@/common/types/errors";
 import type { SendMessageOptions } from "@/common/orpc/types";
 
 describe("MessageQueue", () => {
@@ -504,6 +505,40 @@ describe("MessageQueue", () => {
       expect(second.internal?.onAcceptedPreStreamFailure).toBeUndefined();
     });
 
+    it("should preserve delivery callbacks when reordering strips correlation", async () => {
+      const onCanceled = mock(() => undefined);
+      const onAcceptedPreStreamFailure = mock(() => undefined);
+      const onDeliveryCanceled = mock(() => undefined);
+      const onDeliveryAcceptedPreStreamFailure = mock(() => undefined);
+      queue.add(
+        "Background report",
+        { model: "gpt-4", agentId: "exec", muxMetadata: metadata },
+        {
+          synthetic: true,
+          agentInitiated: true,
+          workspaceTurnContinuation: true,
+          onCanceled,
+          onAcceptedPreStreamFailure,
+          onDeliveryCanceled,
+          onDeliveryAcceptedPreStreamFailure,
+        }
+      );
+      queue.add("User send now", { model: "gpt-4", agentId: "exec" });
+
+      expect(queue.setVisibleQueueDispatchMode("tool-end")).toBe(true);
+      queue.dequeueNext();
+      const reordered = queue.dequeueNext();
+      const error: SendMessageError = { type: "unknown", raw: "startup failed" };
+      await reordered.internal?.onCanceled?.("cleared");
+      await reordered.internal?.onAcceptedPreStreamFailure?.(error);
+
+      expect(reordered.options?.muxMetadata).toBeUndefined();
+      expect(onCanceled).not.toHaveBeenCalled();
+      expect(onAcceptedPreStreamFailure).not.toHaveBeenCalled();
+      expect(onDeliveryCanceled).toHaveBeenCalledWith("cleared");
+      expect(onDeliveryAcceptedPreStreamFailure).toHaveBeenCalledWith(error);
+    });
+
     it("should preserve an original queued workspace-turn prompt during reordering", () => {
       const onAccepted = () => undefined;
       const onCanceled = () => undefined;
@@ -567,6 +602,17 @@ describe("MessageQueue", () => {
       expect(queue.getMessages()).toEqual(["User message before", "User message after"]);
     });
 
+    it("removeWorkspaceTurn reports removal when the entry has no callbacks", () => {
+      queue.add(
+        "Follow up without callbacks",
+        { model: "gpt-4", agentId: "exec", muxMetadata: metadata },
+        { agentInitiated: true, workspaceTurnContinuation: true }
+      );
+
+      expect(queue.removeWorkspaceTurn("wst_followup")).toEqual({});
+      expect(queue.hasWorkspaceTurn("wst_followup")).toBe(false);
+    });
+
     it("should report clear callbacks for every pending entry", () => {
       const onCanceledFirst = () => undefined;
       const onCanceledSecond = () => undefined;

src/node/services/messageQueue.ts

@@ -90,8 +90,14 @@ interface QueuedMessageInternalOptions {
   /** Dedupe-keyed maintenance sends are removable by prefix without changing global queue rules. */
   removableDedupeKey?: boolean;
   onAccepted?: () => Promise<void> | void;
+  /** Correlation callback. Queue reordering can remove it with workspace-turn metadata. */
   onAcceptedPreStreamFailure?: (error: SendMessageError) => Promise<void> | void;
+  /** Correlation callback. Queue reordering can remove it with workspace-turn metadata. */
   onCanceled?: (reason: string) => Promise<void> | void;
+  /** Delivery callback. It survives workspace-turn correlation removal. */
+  onDeliveryAcceptedPreStreamFailure?: (error: SendMessageError) => Promise<void> | void;
+  /** Delivery callback. It survives workspace-turn correlation removal. */
+  onDeliveryCanceled?: (reason: string) => Promise<void> | void;
   /** Mutable dispatch outcome shared with sendQueuedMessages. */
   cancelState?: { canceledBeforeAcceptance: boolean };
   /** Cancels a queued entry even after it has been dequeued into PREPARING. */
@@ -136,10 +142,46 @@ interface QueueEntry {
   onCanceled?: (reason: string) => Promise<void> | void;
   onAccepted?: () => Promise<void> | void;
   onAcceptedPreStreamFailure?: (error: SendMessageError) => Promise<void> | void;
+  onDeliveryCanceled?: (reason: string) => Promise<void> | void;
+  onDeliveryAcceptedPreStreamFailure?: (error: SendMessageError) => Promise<void> | void;
   cancelState?: { canceledBeforeAcceptance: boolean };
   cancelSignal?: AbortSignal;
 }
 
+function combineCallbacks<T>(
+  correlationCallback: ((value: T) => Promise<void> | void) | undefined,
+  deliveryCallback: ((value: T) => Promise<void> | void) | undefined
+): ((value: T) => Promise<void> | void) | undefined {
+  if (correlationCallback == null) {
+    return deliveryCallback;
+  }
+  if (deliveryCallback == null) {
+    return correlationCallback;
+  }
+  return async (value: T) => {
+    try {
+      await correlationCallback?.(value);
+    } finally {
+      await deliveryCallback?.(value);
+    }
+  };
+}
+
+function getQueueClearCallbacks(entry: QueueEntry): QueueClearCallbacks | null {
+  const onCanceled = combineCallbacks(entry.onCanceled, entry.onDeliveryCanceled);
+  const onAcceptedPreStreamFailure = combineCallbacks(
+    entry.onAcceptedPreStreamFailure,
+    entry.onDeliveryAcceptedPreStreamFailure
+  );
+  if (onCanceled == null && onAcceptedPreStreamFailure == null) {
+    return null;
+  }
+  return {
+    ...(onCanceled != null ? { onCanceled } : {}),
+    ...(onAcceptedPreStreamFailure != null ? { onAcceptedPreStreamFailure } : {}),
+  };
+}
+
 /**
  * FIFO queue of messages sent during active streaming.
  *
@@ -396,6 +438,8 @@ export class MessageQueue {
       internal?.onAccepted != null ||
       internal?.onAcceptedPreStreamFailure != null ||
       internal?.onCanceled != null ||
+      internal?.onDeliveryAcceptedPreStreamFailure != null ||
+      internal?.onDeliveryCanceled != null ||
       internal?.cancelSignal != null;
     const incomingIsUserAuthored =
       internal?.synthetic !== true && internal?.agentInitiated !== true;
@@ -479,6 +523,13 @@ export class MessageQueue {
       entry.onAcceptedPreStreamFailure = internal.onAcceptedPreStreamFailure;
     }
 
+    if (internal?.onDeliveryCanceled != null) {
+      entry.onDeliveryCanceled = internal.onDeliveryCanceled;
+    }
+    if (internal?.onDeliveryAcceptedPreStreamFailure != null) {
+      entry.onDeliveryAcceptedPreStreamFailure = internal.onDeliveryAcceptedPreStreamFailure;
+    }
+
     if (internal?.cancelState != null) {
       entry.cancelState = internal.cancelState;
     }
@@ -590,14 +641,10 @@ export class MessageQueue {
    * Callers must notify each one when clearing the queue.
    */
   getClearCallbacks(): QueueClearCallbacks[] {
-    return this.entries
-      .filter((entry) => entry.onCanceled != null || entry.onAcceptedPreStreamFailure != null)
-      .map((entry) => ({
-        ...(entry.onCanceled != null ? { onCanceled: entry.onCanceled } : {}),
-        ...(entry.onAcceptedPreStreamFailure != null
-          ? { onAcceptedPreStreamFailure: entry.onAcceptedPreStreamFailure }
-          : {}),
-      }));
+    return this.entries.flatMap((entry) => {
+      const callbacks = getQueueClearCallbacks(entry);
+      return callbacks == null ? [] : [callbacks];
+    });
   }
 
   /**
@@ -617,12 +664,9 @@ export class MessageQueue {
       return null;
     }
     const [entry] = this.entries.splice(index, 1);
-    return {
-      ...(entry.onCanceled != null ? { onCanceled: entry.onCanceled } : {}),
-      ...(entry.onAcceptedPreStreamFailure != null
-        ? { onAcceptedPreStreamFailure: entry.onAcceptedPreStreamFailure }
-        : {}),
-    };
+    // An empty object still means the entry was removed. Callers must not confuse
+    // callback absence with a missing queue entry.
+    return getQueueClearCallbacks(entry) ?? {};
   }
 
   /** Remove queued entries carrying a dedupe key with the given prefix. */
@@ -661,13 +705,9 @@ export class MessageQueue {
         entry.agentInitiatedCount = Math.min(entry.agentInitiatedCount, entry.addCount);
         return [entry];
       }
-      if (entry.onCanceled != null || entry.onAcceptedPreStreamFailure != null) {
-        removedCallbacks.push({
-          ...(entry.onCanceled != null ? { onCanceled: entry.onCanceled } : {}),
-          ...(entry.onAcceptedPreStreamFailure != null
-            ? { onAcceptedPreStreamFailure: entry.onAcceptedPreStreamFailure }
-            : {}),
-        });
+      const callbacks = getQueueClearCallbacks(entry);
+      if (callbacks != null) {
+        removedCallbacks.push(callbacks);
       }
       return [];
     });
@@ -727,24 +767,27 @@ export class MessageQueue {
     const allAddsAreSynthetic = entry.addCount > 0 && entry.syntheticCount === entry.addCount;
     const allAddsAreAgentInitiated =
       entry.addCount > 0 && entry.agentInitiatedCount === entry.addCount;
+    const onCanceled = combineCallbacks(entry.onCanceled, entry.onDeliveryCanceled);
+    const onAcceptedPreStreamFailure = combineCallbacks(
+      entry.onAcceptedPreStreamFailure,
+      entry.onDeliveryAcceptedPreStreamFailure
+    );
     const hasInternalOptions =
       allAddsAreSynthetic ||
       allAddsAreAgentInitiated ||
       entry.onAccepted != null ||
-      entry.onAcceptedPreStreamFailure != null ||
-      entry.onCanceled != null ||
+      onAcceptedPreStreamFailure != null ||
+      onCanceled != null ||
       entry.cancelSignal != null;
     const internal = hasInternalOptions
       ? {
           ...(allAddsAreSynthetic ? { synthetic: true } : {}),
           ...(allAddsAreAgentInitiated ? { agentInitiated: true } : {}),
-          ...(entry.onCanceled != null ? { onCanceled: entry.onCanceled } : {}),
+          ...(onCanceled != null ? { onCanceled } : {}),
           ...(entry.cancelState != null ? { cancelState: entry.cancelState } : {}),
           ...(entry.cancelSignal != null ? { cancelSignal: entry.cancelSignal } : {}),
           ...(entry.onAccepted != null ? { onAccepted: entry.onAccepted } : {}),
-          ...(entry.onAcceptedPreStreamFailure != null
-            ? { onAcceptedPreStreamFailure: entry.onAcceptedPreStreamFailure }
-            : {}),
+          ...(onAcceptedPreStreamFailure != null ? { onAcceptedPreStreamFailure } : {}),
         }
       : undefined;