🤖 feat: queue messages behind special sends instead of erroring (FIFO message queue)

src/browser/features/ChatInput/index.tsx

@@ -1741,8 +1741,16 @@ const ChatInputInner: React.FC<ChatInputProps> = (props) => {
         mode?: "append" | "replace";
         fileParts?: FilePart[];
         reviews?: ReviewNoteDataForDisplay[];
+        workspaceId?: string;
       }>;
 
+      if (
+        customEvent.detail.workspaceId != null &&
+        workspaceIdForComposerClear !== customEvent.detail.workspaceId
+      ) {
+        return;
+      }
+
       const { text, mode = "append", fileParts, reviews } = customEvent.detail;
       const restoredIdPrefix = `restored-${Date.now()}`;
       const restoredPending = buildPendingFromRestoredInput({
@@ -1781,7 +1789,15 @@ const ChatInputInner: React.FC<ChatInputProps> = (props) => {
     window.addEventListener(CUSTOM_EVENTS.UPDATE_CHAT_INPUT, handler as EventListener);
     return () =>
       window.removeEventListener(CUSTOM_EVENTS.UPDATE_CHAT_INPUT, handler as EventListener);
-  }, [appendText, restoreText, restoreDraft, applyDraftFromPending, getDraft, editingMessageForUi]);
+  }, [
+    appendText,
+    restoreText,
+    restoreDraft,
+    applyDraftFromPending,
+    getDraft,
+    editingMessageForUi,
+    workspaceIdForComposerClear,
+  ]);
 
   useEffect(() => {
     const handler = (event: CustomEvent<{ workspaceId: string }>) => {

src/browser/stores/WorkspaceStore.test.ts

@@ -4405,6 +4405,42 @@ describe("WorkspaceStore", () => {
     });
   });
 
+  it("queued message identity changes when a partial FIFO drain changes visible content", async () => {
+    const workspaceId = "queued-message-identity";
+    mockChatStreamFor(workspaceId, async function* () {
+      yield {
+        type: "queued-message-changed",
+        workspaceId,
+        hasQueuedMessages: true,
+        queuedMessages: ["first", "second"],
+        displayText: "first\nsecond",
+      };
+      yield { type: "caught-up", hasOlderHistory: false };
+      await tick(25);
+      yield {
+        type: "queued-message-changed",
+        workspaceId,
+        hasQueuedMessages: true,
+        queuedMessages: ["second"],
+        displayText: "second",
+      };
+    });
+
+    createAndAddWorkspace(store, workspaceId);
+    expect(await waitUntil(() => store.getWorkspaceState(workspaceId).queuedMessage != null)).toBe(
+      true
+    );
+    const firstId = store.getWorkspaceState(workspaceId).queuedMessage?.id;
+
+    expect(
+      await waitUntil(
+        () =>
+          store.getWorkspaceState(workspaceId).queuedMessage?.content === "second" &&
+          store.getWorkspaceState(workspaceId).queuedMessage?.id !== firstId
+      )
+    ).toBe(true);
+  });
+
   describe("bash-output events", () => {
     it("retains live output when bash tool result has no output", async () => {
       const workspaceId = "bash-output-workspace-1";

src/browser/stores/WorkspaceStore.ts

@@ -1025,7 +1025,15 @@ export class WorkspaceStore {
         (data.reviews?.length ?? 0) > 0;
       const queuedMessage: QueuedMessage | null = hasContent
         ? {
-            id: `queued-${workspaceId}`,
+            // Change identity whenever the visible queue projection changes so
+            // per-card actions (notably Send now) reset after a partial FIFO drain.
+            id: `queued-${workspaceId}-${JSON.stringify([
+              data.displayText,
+              data.fileParts?.map((part) => [part.mediaType, part.filename, part.url.length]) ?? [],
+              data.reviews?.map((review) => [review.filePath, review.lineRange]) ?? [],
+              data.queueDispatchMode,
+              data.hasCompactionRequest,
+            ])}`,
             content: data.displayText,
             fileParts: data.fileParts,
             reviews: data.reviews,
@@ -1036,7 +1044,7 @@ export class WorkspaceStore {
 
       // Mirror the queue signal onto active streams so response notifications follow
       // user-visible terminal turns instead of every intermediate handoff.
-      aggregator.setActiveQueuedFollowUp(queuedMessage !== null);
+      aggregator.setActiveQueuedFollowUp(data.hasQueuedMessages ?? queuedMessage !== null);
       this.assertChatTransientState(workspaceId).queuedMessage = queuedMessage;
       this.states.bump(workspaceId);
     },
@@ -1050,6 +1058,9 @@ export class WorkspaceStore {
           mode: "replace",
           fileParts: data.fileParts,
           reviews: data.reviews,
+          // Restore events can arrive for a background workspace; never let them
+          // overwrite the composer currently mounted for another workspace.
+          workspaceId: data.workspaceId,
         })
       );
     },

src/common/constants/events.ts

@@ -150,6 +150,8 @@ export interface CustomEventPayloads {
     mode?: "replace" | "append";
     fileParts?: FilePart[];
     reviews?: ReviewNoteDataForDisplay[];
+    /** When set, only the matching workspace composer may apply this update. */
+    workspaceId?: string;
   };
   [CUSTOM_EVENTS.CLEAR_CHAT_COMPOSER]: {
     workspaceId: string;

src/common/orpc/schemas/stream.ts

@@ -602,6 +602,8 @@ export const GoalBudgetLimitedEventSchema = z.object({
 export const QueuedMessageChangedEventSchema = z.object({
   type: z.literal("queued-message-changed"),
   workspaceId: z.string(),
+  /** True when any entry is queued, including hidden synthetic/background entries. */
+  hasQueuedMessages: z.boolean().optional(),
   queuedMessages: z.array(z.string()),
   displayText: z.string(),
   fileParts: z.array(FilePartSchema).optional(),

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

@@ -319,6 +319,58 @@ describe("AgentSession queued message tool-call dispatch", () => {
     }
   });
 
+  test("synthetic background entries neither surface in queue UI nor restore over user input", async () => {
+    const workspaceId = "queue-dispatch-hide-synthetic";
+    const { session, cleanup } = await createAgentSessionHarness({ workspaceId });
+
+    try {
+      const queuedSnapshots: string[][] = [];
+      const hasQueuedSnapshots: boolean[] = [];
+      const restoredTexts: string[] = [];
+      const canceledReasons: string[] = [];
+      const unsubscribe = session.onChatEvent((event) => {
+        if (event.message.type === "queued-message-changed") {
+          queuedSnapshots.push(event.message.queuedMessages);
+          hasQueuedSnapshots.push(event.message.hasQueuedMessages ?? false);
+        }
+        if (event.message.type === "restore-to-input") {
+          restoredTexts.push(event.message.text);
+        }
+      });
+
+      session.queueMessage(
+        "Background monitor wake",
+        { model: TEST_MODEL, agentId: "exec" },
+        {
+          synthetic: true,
+          agentInitiated: true,
+          onCanceled: (reason) => {
+            canceledReasons.push(reason);
+          },
+        }
+      );
+      expect(hasQueuedSnapshots.at(-1)).toBe(true);
+      expect(queuedSnapshots.at(-1)).toEqual([]);
+
+      session.queueMessage("my own words", {
+        model: TEST_MODEL,
+        agentId: "exec",
+        queueDispatchMode: "turn-end",
+      });
+      expect(queuedSnapshots.at(-1)).toEqual(["my own words"]);
+
+      session.restoreQueueToInput();
+      unsubscribe();
+
+      expect(restoredTexts).toEqual(["my own words"]);
+      expect(canceledReasons).toHaveLength(1);
+      expect(session.hasQueuedMessages()).toBe(false);
+    } finally {
+      session.dispose();
+      await cleanup();
+    }
+  });
+
   test("does not send the queued message after a user abort", async () => {
     const workspaceId = "queue-dispatch-user-abort";
     const { session, cleanup, aiEmitter, aiService } = await createAgentSessionHarness({

src/node/services/agentSession.ts

@@ -26,7 +26,6 @@ import type {
   StreamErrorMessage,
 } from "@/common/orpc/types";
 import { WORKSPACE_DEFAULTS } from "@/constants/workspaceDefaults";
-import { HEARTBEAT_QUEUE_DEDUPE_KEY } from "@/constants/heartbeat";
 import {
   GOAL_BUDGET_LIMIT_KIND,
   GOAL_CONTINUATION_KIND,
@@ -2177,12 +2176,13 @@ export class AgentSession {
         message: {
           type: "queued-message-changed",
           workspaceId: this.workspaceId,
-          queuedMessages: this.messageQueue.getMessages(),
-          displayText: this.messageQueue.getDisplayText(),
-          fileParts: this.messageQueue.getFileParts(),
-          reviews: this.messageQueue.getReviews(),
-          queueDispatchMode: this.messageQueue.getQueueDispatchMode(),
-          hasCompactionRequest: this.messageQueue.hasCompactionRequest(),
+          hasQueuedMessages: !this.messageQueue.isEmpty(),
+          queuedMessages: this.messageQueue.getVisibleMessages(),
+          displayText: this.messageQueue.getVisibleDisplayText(),
+          fileParts: this.messageQueue.getVisibleFileParts(),
+          reviews: this.messageQueue.getVisibleReviews(),
+          queueDispatchMode: this.messageQueue.getVisibleQueueDispatchMode(),
+          hasCompactionRequest: this.messageQueue.hasVisibleCompactionRequest(),
         },
       });
 
@@ -5082,11 +5082,13 @@ export class AgentSession {
 
   clearQueue(cancelReason = "Queued message cleared before dispatch."): void {
     this.assertNotDisposed("clearQueue");
-    const callbacks = this.messageQueue.getClearCallbacks();
+    const callbackSets = this.messageQueue.getClearCallbacks();
     this.messageQueue.clear();
     this.emitQueuedMessageChanged();
     this.backgroundProcessManager.setMessageQueued(this.workspaceId, false);
-    this.notifyQueuedMessageCleared(callbacks, cancelReason);
+    for (const callbacks of callbackSets) {
+      this.notifyQueuedMessageCleared(callbacks, cancelReason);
+    }
   }
 
   private notifyQueuedMessageCleared(
@@ -5116,6 +5118,27 @@ export class AgentSession {
     return this.messageQueue.hasWorkspaceTurn(handleId);
   }
 
+  /**
+   * Remove only the queued workspace-turn entry for this handle, keeping any
+   * unrelated queued messages (interrupting a queued turn must not drop user
+   * input queued before/behind it). Returns true when an entry was removed.
+   */
+  removeQueuedWorkspaceTurn(handleId: string, cancelReason: string): boolean {
+    this.assertNotDisposed("removeQueuedWorkspaceTurn");
+    assert(handleId.length > 0, "removeQueuedWorkspaceTurn requires handleId");
+    const callbacks = this.messageQueue.removeWorkspaceTurn(handleId);
+    if (callbacks == null) {
+      return false;
+    }
+    this.emitQueuedMessageChanged();
+    this.backgroundProcessManager.setMessageQueued(
+      this.workspaceId,
+      !this.messageQueue.isEmpty() && this.messageQueue.getQueueDispatchMode() === "tool-end"
+    );
+    this.notifyQueuedMessageCleared(callbacks, cancelReason);
+    return true;
+  }
+
   hasQueuedMessages(dispatchMode?: "tool-end" | "turn-end"): boolean {
     return (
       !this.messageQueue.isEmpty() &&
@@ -5200,32 +5223,34 @@ export class AgentSession {
   }
 
   /**
-   * Restore queued messages to input box.
-   * Called by IPC handler on user-initiated interrupt.
+   * Restore queued user input to the composer after a user-initiated interrupt.
+   * Fully synthetic background work is canceled with the queue but never surfaced
+   * as editable text, so monitor wakes cannot replace or pollute the user's draft.
    */
   restoreQueueToInput(): void {
     this.assertNotDisposed("restoreQueueToInput");
-    // Restore-to-input exists to give the user their own words back (interrupt / edit).
-    // A queued scheduled heartbeat is backend-initiated maintenance, not user input —
-    // surfacing it as editable composer text would be confusing, so discard it instead
-    // (the check-in is periodic; its next slot fires anyway). It can only ever be the
-    // queue's sole content: it is enqueued exclusively into an empty queue, and any
-    // later user message supersedes it before queueing.
-    if (this.dropQueuedMessageWithOnlyDedupeKey(HEARTBEAT_QUEUE_DEDUPE_KEY)) {
+    if (this.messageQueue.isEmpty()) {
       return;
     }
-    if (!this.messageQueue.isEmpty()) {
-      const displayText = this.messageQueue.getDisplayText();
-      const fileParts = this.messageQueue.getFileParts();
-      const reviews = this.messageQueue.getReviews();
-      this.clearQueue();
 
+    const queuedMessages = this.messageQueue.getVisibleMessages();
+    const displayText = this.messageQueue.getVisibleDisplayText();
+    const fileParts = this.messageQueue.getVisibleFileParts();
+    const reviews = this.messageQueue.getVisibleReviews();
+    const hasVisibleContent =
+      queuedMessages.length > 0 || fileParts.length > 0 || (reviews?.length ?? 0) > 0;
+
+    // Clear everything: synthetic wake callbacks need cancellation so their durable
+    // records do not retry after the user explicitly interrupted the workspace.
+    this.clearQueue();
+
+    if (hasVisibleContent) {
       this.emitChatEvent({
         type: "restore-to-input",
         workspaceId: this.workspaceId,
         text: displayText,
-        fileParts: fileParts,
-        reviews: reviews,
+        fileParts,
+        reviews,
       });
     }
   }
@@ -5234,15 +5259,29 @@ export class AgentSession {
     this.emitChatEvent({
       type: "queued-message-changed",
       workspaceId: this.workspaceId,
-      queuedMessages: this.messageQueue.getMessages(),
-      displayText: this.messageQueue.getDisplayText(),
-      fileParts: this.messageQueue.getFileParts(),
-      reviews: this.messageQueue.getReviews(),
-      queueDispatchMode: this.messageQueue.getQueueDispatchMode(),
-      hasCompactionRequest: this.messageQueue.hasCompactionRequest(),
+      hasQueuedMessages: !this.messageQueue.isEmpty(),
+      queuedMessages: this.messageQueue.getVisibleMessages(),
+      displayText: this.messageQueue.getVisibleDisplayText(),
+      fileParts: this.messageQueue.getVisibleFileParts(),
+      reviews: this.messageQueue.getVisibleReviews(),
+      queueDispatchMode: this.messageQueue.getVisibleQueueDispatchMode(),
+      hasCompactionRequest: this.messageQueue.hasVisibleCompactionRequest(),
     });
   }
 
+  /**
+   * Dispatch the next user-authored queued entry immediately. Hidden synthetic
+   * entries remain queued behind it and resume through the normal drain lifecycle.
+   */
+  sendNextUserQueuedMessage(): boolean {
+    this.assertNotDisposed("sendNextUserQueuedMessage");
+    if (!this.messageQueue.prioritizeNextUserEntry()) {
+      return false;
+    }
+    this.sendQueuedMessages();
+    return true;
+  }
+
   /**
    * Send queued messages if any exist.
    * Called when tool execution completes, stream ends, or user clicks send immediately.
@@ -5260,10 +5299,21 @@ export class AgentSession {
     this.backgroundProcessManager.setMessageQueued(this.workspaceId, false);
 
     if (!this.messageQueue.isEmpty()) {
-      const { message, options, internal } = this.messageQueue.produceMessage();
-      this.messageQueue.clear();
+      // Entries dispatch one at a time (FIFO): special sends (compaction, agent
+      // skills, workspace-turn follow-ups) own their turn, and anything queued
+      // behind them dispatches on a later drain instead of batching into them.
+      const { message, options, internal } = this.messageQueue.dequeueNext();
       this.emitQueuedMessageChanged();
 
+      // Re-arm dispatch signals for the remaining entries so the stream we are
+      // about to start drains them at its next tool end (or stream end).
+      if (!this.messageQueue.isEmpty()) {
+        this.backgroundProcessManager.setMessageQueued(
+          this.workspaceId,
+          this.messageQueue.getQueueDispatchMode() === "tool-end"
+        );
+      }
+
       // Set PREPARING synchronously before the async sendMessage to prevent
       // incoming messages from bypassing the queue during the await gap.
       this.setTurnPhase(TurnPhase.PREPARING);
@@ -5277,12 +5327,17 @@ export class AgentSession {
             if (this.turnPhase === TurnPhase.PREPARING) {
               this.setTurnPhase(TurnPhase.IDLE);
             }
+            // No stream started, so no stream-end drain will fire for the
+            // remaining entries — try the next one now (each attempt pops an
+            // entry, so this terminates).
+            this.sendQueuedMessages();
           }
         })
         .catch(() => {
           if (this.turnPhase === TurnPhase.PREPARING) {
             this.setTurnPhase(TurnPhase.IDLE);
           }
+          this.sendQueuedMessages();
         });
     }
   }