🤖 fix: preserve queued follow-ups during dispatch handoff

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

@@ -514,6 +514,59 @@ describe("AgentSession queued message tool-call dispatch", () => {
     }
   });
 
+  test("keeps a dequeued user message visible until its durable row is emitted", async () => {
+    const workspaceId = "queue-dispatch-visible-handoff";
+    const { session, cleanup, historyService, events } = await createAgentSessionHarness({
+      workspaceId,
+      captureEvents: true,
+    });
+    const originalAppend = historyService.appendToHistory.bind(historyService);
+    const appendStarted = Promise.withResolvers<void>();
+    const appendRelease = Promise.withResolvers<void>();
+    const appendSpy = spyOn(historyService, "appendToHistory").mockImplementation(
+      async (...args) => {
+        appendStarted.resolve();
+        await appendRelease.promise;
+        return originalAppend(...args);
+      }
+    );
+    const followUp = "Follow up after compaction";
+    const isFollowUpUserMessage = (event: (typeof events)[number]) =>
+      event.type === "message" &&
+      event.role === "user" &&
+      event.parts.some((part) => part.type === "text" && part.text === followUp);
+    const latestQueuedMessages = () =>
+      events.filter((event) => event.type === "queued-message-changed").at(-1)?.queuedMessages;
+
+    try {
+      session.queueMessage(followUp, { model: TEST_MODEL, agentId: "exec" });
+      session.sendQueuedMessages();
+      await appendStarted.promise;
+
+      expect(latestQueuedMessages()).toEqual([followUp]);
+      expect(events.some(isFollowUpUserMessage)).toBe(false);
+
+      appendRelease.resolve();
+      expect(await waitForCondition(() => events.some(isFollowUpUserMessage))).toBe(true);
+      expect(await waitForCondition(() => latestQueuedMessages()?.length === 0)).toBe(true);
+
+      const userMessageIndex = events.findIndex(isFollowUpUserMessage);
+      const clearedQueueIndex = events.findIndex(
+        (event, index) =>
+          index > userMessageIndex &&
+          event.type === "queued-message-changed" &&
+          event.queuedMessages.length === 0
+      );
+      expect(userMessageIndex).toBeGreaterThanOrEqual(0);
+      expect(clearedQueueIndex).toBeGreaterThan(userMessageIndex);
+    } finally {
+      appendRelease.resolve();
+      appendSpy.mockRestore();
+      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

@@ -5909,7 +5909,19 @@ export class AgentSession {
       const { message, options, internal } = this.messageQueue.dequeueNext();
       this.dispatchingQueuedEntry = true;
       this.dispatchingQueuedEntryMuxMetadata = options?.muxMetadata;
-      this.emitQueuedMessageChanged();
+
+      // Keep the dequeued user entry visible until sendMessage emits its durable user row.
+      // Otherwise compaction completion clears the queue card before the transcript replacement exists.
+      const finishDispatchWithoutStream = (): void => {
+        this.emitQueuedMessageChanged();
+        this.dispatchingQueuedEntry = false;
+        this.dispatchingQueuedEntryMuxMetadata = undefined;
+        if (this.turnPhase === TurnPhase.PREPARING) {
+          this.setTurnPhase(TurnPhase.IDLE);
+        }
+        // No stream will drain later entries, so continue now (each attempt pops one entry).
+        this.sendQueuedMessages();
+      };
 
       // 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).
@@ -5925,41 +5937,26 @@ export class AgentSession {
       this.preparingWorkspaceTurnMetadata = getWorkspaceTurnMuxMetadata(options?.muxMetadata);
       this.setTurnPhase(TurnPhase.PREPARING);
 
-      void this.sendMessage(message, options, internal)
+      void this.sendMessage(message, options, {
+        ...internal,
+        onAccepted: async () => {
+          this.emitQueuedMessageChanged();
+          await internal?.onAccepted?.();
+        },
+      })
         .then(async (result) => {
           // Keep the dispatch marker through the dequeue-to-stream-start window. A background
           // send can resolve before startup emits stream-start, and later reports must not claim
           // that window as the next continuation.
-          // If sendMessage fails before it can start streaming, ensure we don't
-          // leave the session stuck in PREPARING and notify correlated internal callers.
           if (!result.success) {
             await internal?.onAcceptedPreStreamFailure?.(result.error);
-            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();
-            return;
-          }
-          if (internal?.cancelState?.canceledBeforeAcceptance === true) {
-            // Cancellation can arrive after dequeue while sendMessage is validating or writing
-            // history. No stream will start, so release PREPARING and continue with later entries.
-            if (this.turnPhase === TurnPhase.PREPARING) {
-              this.setTurnPhase(TurnPhase.IDLE);
-            }
-            this.sendQueuedMessages();
+            finishDispatchWithoutStream();
+          } else if (internal?.cancelState?.canceledBeforeAcceptance === true) {
+            // Cancellation can arrive after dequeue while sendMessage is validating or writing.
+            finishDispatchWithoutStream();
           }
         })
-        .catch(() => {
-          this.dispatchingQueuedEntry = false;
-          this.dispatchingQueuedEntryMuxMetadata = undefined;
-          if (this.turnPhase === TurnPhase.PREPARING) {
-            this.setTurnPhase(TurnPhase.IDLE);
-          }
-          this.sendQueuedMessages();
-        });
+        .catch(finishDispatchWithoutStream);
     }
   }