From 4d570a81b9ba0eb0a830bb2a201d42e99fa0d704 Mon Sep 17 00:00:00 2001 From: Ammar Date: Fri, 21 Aug 2026 22:25:04 -0500 Subject: [PATCH 1/4] =?UTF-8?q?=F0=9F=A4=96=20fix:=20preserve=20queued=20f?= =?UTF-8?q?ollow-ups=20during=20dispatch=20handoff?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Keep a dequeued user follow-up visible until its durable transcript row is emitted, and consolidate no-stream queue cleanup. Add a regression test that verifies the queue card-to-transcript event ordering. --- _Generated with `xum` • Model: `openai:gpt-5.6-sol` • Thinking: `xhigh` • Cost: `$25.07`_ --- .../agentSession.queueDispatch.test.ts | 53 +++++++++++++++++++ src/node/services/agentSession.ts | 53 +++++++++---------- 2 files changed, 78 insertions(+), 28 deletions(-) diff --git a/src/node/services/agentSession.queueDispatch.test.ts b/src/node/services/agentSession.queueDispatch.test.ts index fa4a827d7bf..2d0b0a0975f 100644 --- a/src/node/services/agentSession.queueDispatch.test.ts +++ b/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(); + const appendRelease = Promise.withResolvers(); + 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({ diff --git a/src/node/services/agentSession.ts b/src/node/services/agentSession.ts index 2359feab9db..9e23386701e 100644 --- a/src/node/services/agentSession.ts +++ b/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); } } From 0baeab9ec33739e93f023db717ebe30d6f2adfc0 Mon Sep 17 00:00:00 2001 From: Ammar Date: Fri, 21 Aug 2026 22:45:53 -0500 Subject: [PATCH 2/4] =?UTF-8?q?=F0=9F=A4=96=20fix:=20make=20queued=20dispa?= =?UTF-8?q?tch=20projection=20authoritative?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Retain the in-flight user entry in the server queue projection through durable acceptance, include it in reconnect and mutation snapshots, and disable stale queue actions during the handoff. --- _Generated with `xum` • Model: `openai:gpt-5.6-sol` • Thinking: `xhigh` • Cost: `$64.30`_ --- src/browser/components/ChatPane/ChatPane.tsx | 9 +- src/browser/features/ChatInput/index.tsx | 1 + .../features/Messages/QueuedMessage.tsx | 25 +++-- .../agentSession.queueDispatch.test.ts | 28 +++-- src/node/services/agentSession.ts | 106 +++++++++--------- src/node/services/messageQueue.test.ts | 8 +- src/node/services/messageQueue.ts | 59 +++++----- tests/ui/chat/queuedMessageBanner.test.tsx | 17 +++ 8 files changed, 150 insertions(+), 103 deletions(-) diff --git a/src/browser/components/ChatPane/ChatPane.tsx b/src/browser/components/ChatPane/ChatPane.tsx index 4b43cd913f6..93db1a74181 100644 --- a/src/browser/components/ChatPane/ChatPane.tsx +++ b/src/browser/components/ChatPane/ChatPane.tsx @@ -860,10 +860,10 @@ const ChatPaneContent: React.FC = (props) => { const handleEditQueuedMessage = useCallback(async () => { const queuedMessage = workspaceState?.queuedMessage; - if (!queuedMessage) return; + if (!queuedMessage || workspaceState?.isStreamStarting) return; await restoreQueuedDraft(queuedMessage); - }, [restoreQueuedDraft, workspaceState?.queuedMessage]); + }, [restoreQueuedDraft, workspaceState?.isStreamStarting, workspaceState?.queuedMessage]); const sendQueuedImmediatelyInFlightRef = useRef(null); @@ -958,7 +958,9 @@ const ChatPaneContent: React.FC = (props) => { if (!current) return; if (current.queuedMessage) { - await restoreQueuedDraft(current.queuedMessage); + if (!current.isStreamStarting) { + await restoreQueuedDraft(current.queuedMessage); + } return; } @@ -1818,6 +1820,7 @@ const ChatInputPane: React.FC = (props) => { node: ( void props.onEditQueuedMessage()} onChangeDispatchMode={props.onQueuedDispatchModeChange} onActionError={props.onQueuedActionError} diff --git a/src/browser/features/ChatInput/index.tsx b/src/browser/features/ChatInput/index.tsx index ca05974833a..3ecba19becb 100644 --- a/src/browser/features/ChatInput/index.tsx +++ b/src/browser/features/ChatInput/index.tsx @@ -3356,6 +3356,7 @@ const ChatInputInner: React.FC = (props) => { variant === "workspace" && !editingMessageForUi && props.queuedMessage != null && + !isStreamStarting && input.trim() === "" && attachments.length === 0 && reviewPanelItems.length === 0; diff --git a/src/browser/features/Messages/QueuedMessage.tsx b/src/browser/features/Messages/QueuedMessage.tsx index 7608c6d0914..e0238c73fbe 100644 --- a/src/browser/features/Messages/QueuedMessage.tsx +++ b/src/browser/features/Messages/QueuedMessage.tsx @@ -11,6 +11,7 @@ import { cn } from "@/common/lib/utils"; interface QueuedMessageProps { message: QueuedMessageType; + isDispatching?: boolean; className?: string; onEdit?: () => void; onChangeDispatchMode?: (mode: QueueDispatchMode) => Promise; @@ -44,6 +45,7 @@ export const QueuedMessage: React.FC = (props) => { const queueDispatchMode = props.message.queueDispatchMode ?? "tool-end"; const queueStatusLabel = queueDispatchMode === "turn-end" ? "Sends after this turn" : "Sends after this step"; + const isDispatching = props.isDispatching === true; const isActionPending = pendingAction != null; const handleDispatchModeChange = (mode: QueueDispatchMode) => { @@ -114,7 +116,7 @@ export const QueuedMessage: React.FC = (props) => { className="mt-1.5 flex max-w-full flex-wrap items-center justify-end gap-1 text-[11px]" data-component="QueuedMessageActions" > - {props.onEdit && ( + {!isDispatching && props.onEdit && ( - {isMenuOpen && ( + {!isDispatching && isMenuOpen && (
{ } ); const followUp = "Follow up after compaction"; + const nextFollowUp = "A later queued message"; 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; + const latestQueueEvent = () => + events.filter((event) => event.type === "queued-message-changed").at(-1); try { session.queueMessage(followUp, { model: TEST_MODEL, agentId: "exec" }); + const queueEventCountBeforeDispatch = events.filter( + (event) => event.type === "queued-message-changed" + ).length; session.sendQueuedMessages(); await appendStarted.promise; - expect(latestQueuedMessages()).toEqual([followUp]); + expect(events.filter((event) => event.type === "queued-message-changed").length).toBe( + queueEventCountBeforeDispatch + 1 + ); + expect(latestQueueEvent()?.queuedMessages).toEqual([followUp]); expect(events.some(isFollowUpUserMessage)).toBe(false); + // Any queue mutation during persistence must retain the in-flight entry in the + // authoritative projection instead of falling back to a stale renderer snapshot. + session.queueMessage(nextFollowUp, { model: TEST_MODEL, agentId: "exec" }); + expect(latestQueueEvent()?.queuedMessages).toEqual([followUp, nextFollowUp]); + appendRelease.resolve(); expect(await waitForCondition(() => events.some(isFollowUpUserMessage))).toBe(true); - expect(await waitForCondition(() => latestQueuedMessages()?.length === 0)).toBe(true); + expect( + await waitForCondition(() => latestQueueEvent()?.queuedMessages.join() === nextFollowUp) + ).toBe(true); const userMessageIndex = events.findIndex(isFollowUpUserMessage); - const clearedQueueIndex = events.findIndex( + const handoffIndex = events.findIndex( (event, index) => index > userMessageIndex && event.type === "queued-message-changed" && - event.queuedMessages.length === 0 + event.queuedMessages.join() === nextFollowUp ); expect(userMessageIndex).toBeGreaterThanOrEqual(0); - expect(clearedQueueIndex).toBeGreaterThan(userMessageIndex); + expect(handoffIndex).toBeGreaterThan(userMessageIndex); } finally { appendRelease.resolve(); appendSpy.mockRestore(); diff --git a/src/node/services/agentSession.ts b/src/node/services/agentSession.ts index 9e23386701e..962f2f39fdd 100644 --- a/src/node/services/agentSession.ts +++ b/src/node/services/agentSession.ts @@ -99,7 +99,7 @@ import { createRuntimeContextForWorkspace, createRuntimeForWorkspace, } from "@/node/runtime/runtimeHelpers"; -import { MessageQueue } from "./messageQueue"; +import { MessageQueue, type MessageQueueVisibleProjection } from "./messageQueue"; import { copyStreamLifecycleSnapshot, type RuntimeStatusEvent, @@ -706,14 +706,11 @@ export class AgentSession { /** Tracks whether the current stream included post-compaction attachments. */ private activeStreamHadPostCompactionInjection = false; - /** - * muxMetadata of the queued entry currently being dispatched, held from - * dequeue until its sendMessage settles (the stream has started or failed). - * Lets hasPendingBashMonitorWakeContinuation see a wake continuation during - * the dequeue→stream-start window without consulting stale stream context. - */ - private dispatchingQueuedEntry = false; - private dispatchingQueuedEntryMuxMetadata?: unknown; + /** Queued entry held through the dequeue→acceptance handoff for projection and correlation. */ + private dispatchingQueuedEntry?: { + muxMetadata?: unknown; + visibleProjection?: MessageQueueVisibleProjection; + }; /** Correlation of the direct send currently in the PREPARING phase, if any. */ private preparingWorkspaceTurnMetadata?: WorkspaceTurnMuxMetadata; @@ -2504,21 +2501,11 @@ export class AgentSession { emitCurrentReplayTerminalState(); } - // Replay queued-message snapshot before caught-up so reconnect clients can - // rebuild queue UI state even when history replay errored mid-flight. + // Replay the authoritative queue projection before caught-up so reconnect clients + // retain a user entry that is between dequeue and durable transcript emission. listener({ workspaceId: this.workspaceId, - message: { - type: "queued-message-changed", - workspaceId: this.workspaceId, - 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(), - }, + message: this.getQueuedMessageChangedEvent(), }); // Rehydrate pending auto-retry countdown state on reconnect/reload so @@ -4890,8 +4877,7 @@ export class AgentSession { forward("stream-start", (payload) => { if (payload.type === "stream-start") { - this.dispatchingQueuedEntry = false; - this.dispatchingQueuedEntryMuxMetadata = undefined; + this.dispatchingQueuedEntry = undefined; this.preparingWorkspaceTurnMetadata = undefined; this.activeStreamStartedAtMs = payload.startTime; this.queuedProviderToolEndAbortInFlight = false; @@ -5389,8 +5375,7 @@ export class AgentSession { this.emitStreamLifecycleIfChanged(); if (next === TurnPhase.IDLE) { - this.dispatchingQueuedEntry = false; - this.dispatchingQueuedEntryMuxMetadata = undefined; + this.dispatchingQueuedEntry = undefined; this.preparingWorkspaceTurnMetadata = undefined; // Turn ended: expire any mid-turn thinking override. Safe unconditionally // because a replacement turn (e.g. an edit) only creates its holder after @@ -5657,7 +5642,7 @@ export class AgentSession { if (this.dispatchingQueuedEntry) { const dispatchingMetadata = getWorkspaceTurnMuxMetadata( - this.dispatchingQueuedEntryMuxMetadata + this.dispatchingQueuedEntry.muxMetadata ); if (!hasSameWorkspaceTurnCorrelation(dispatchingMetadata, continuationMetadata)) { return true; @@ -5692,7 +5677,7 @@ export class AgentSession { if (this.messageQueue.isNextEntryBashMonitorWake()) { return true; } - const dispatching = this.dispatchingQueuedEntryMuxMetadata as MuxMessageMetadata | undefined; + const dispatching = this.dispatchingQueuedEntry?.muxMetadata as MuxMessageMetadata | undefined; return dispatching?.type === "bash-monitor-wake"; } @@ -5716,7 +5701,7 @@ export class AgentSession { return true; } - const dispatching = this.dispatchingQueuedEntryMuxMetadata as MuxMessageMetadata | undefined; + const dispatching = this.dispatchingQueuedEntry?.muxMetadata as MuxMessageMetadata | undefined; return ( dispatching?.type === "workspace-turn-task" && dispatching.taskHandleId === metadata.taskHandleId && @@ -5837,10 +5822,8 @@ export class AgentSession { return; } - const queuedMessages = this.messageQueue.getVisibleMessages(); - const displayText = this.messageQueue.getVisibleDisplayText(); - const fileParts = this.messageQueue.getVisibleFileParts(); - const reviews = this.messageQueue.getVisibleReviews(); + const { queuedMessages, displayText, fileParts, reviews } = + this.messageQueue.getVisibleProjection(); const hasVisibleContent = queuedMessages.length > 0 || fileParts.length > 0 || (reviews?.length ?? 0) > 0; @@ -5859,18 +5842,30 @@ export class AgentSession { } } - private emitQueuedMessageChanged(): void { - this.emitChatEvent({ + private getQueuedMessageChangedEvent(): Extract< + WorkspaceChatMessage, + { type: "queued-message-changed" } + > { + const pending = this.messageQueue.getVisibleProjection(); + const dispatching = this.dispatchingQueuedEntry?.visibleProjection; + const reviews = [...(dispatching?.reviews ?? []), ...(pending.reviews ?? [])]; + + return { type: "queued-message-changed", workspaceId: this.workspaceId, - 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(), - }); + hasQueuedMessages: this.dispatchingQueuedEntry != null || !this.messageQueue.isEmpty(), + queuedMessages: [...(dispatching?.queuedMessages ?? []), ...pending.queuedMessages], + displayText: [dispatching?.displayText, pending.displayText].filter(Boolean).join("\n"), + fileParts: [...(dispatching?.fileParts ?? []), ...pending.fileParts], + reviews: reviews.length > 0 ? reviews : undefined, + queueDispatchMode: dispatching?.queueDispatchMode ?? pending.queueDispatchMode, + hasCompactionRequest: + dispatching?.hasCompactionRequest === true || pending.hasCompactionRequest, + }; + } + + private emitQueuedMessageChanged(): void { + this.emitChatEvent(this.getQueuedMessageChangedEvent()); } /** @@ -5906,16 +5901,20 @@ export class AgentSession { // 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.dispatchingQueuedEntry = true; - this.dispatchingQueuedEntryMuxMetadata = options?.muxMetadata; + const { message, options, internal, visibleProjection } = this.messageQueue.dequeueNext(); + this.dispatchingQueuedEntry = { + muxMetadata: options?.muxMetadata, + visibleProjection, + }; + // PREPARING disables queue actions while the authoritative projection includes + // the in-flight entry, preventing stale Edit/Send-now operations during persistence. + this.preparingWorkspaceTurnMetadata = getWorkspaceTurnMuxMetadata(options?.muxMetadata); + this.setTurnPhase(TurnPhase.PREPARING); + 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.dispatchingQueuedEntry = undefined; this.emitQueuedMessageChanged(); - this.dispatchingQueuedEntry = false; - this.dispatchingQueuedEntryMuxMetadata = undefined; if (this.turnPhase === TurnPhase.PREPARING) { this.setTurnPhase(TurnPhase.IDLE); } @@ -5932,14 +5931,11 @@ export class AgentSession { ); } - // Set PREPARING synchronously before the async sendMessage to prevent - // incoming messages from bypassing the queue during the await gap. - this.preparingWorkspaceTurnMetadata = getWorkspaceTurnMuxMetadata(options?.muxMetadata); - this.setTurnPhase(TurnPhase.PREPARING); - void this.sendMessage(message, options, { ...internal, onAccepted: async () => { + // sendMessage has emitted the durable user row, so the transient queue projection can hand off. + this.dispatchingQueuedEntry = undefined; this.emitQueuedMessageChanged(); await internal?.onAccepted?.(); }, diff --git a/src/node/services/messageQueue.test.ts b/src/node/services/messageQueue.test.ts index 0373f37ca53..bac8ff62735 100644 --- a/src/node/services/messageQueue.test.ts +++ b/src/node/services/messageQueue.test.ts @@ -33,10 +33,12 @@ describe("MessageQueue", () => { // Content projection hides backend work, while the visible card reflects the FIFO head's // effective boundary until the user explicitly reprioritizes their queued follow-up. expect(queue.getMessages()).toEqual(["Background monitor wake", "User follow-up"]); - expect(queue.getVisibleMessages()).toEqual(["User follow-up"]); - expect(queue.getVisibleDisplayText()).toBe("User follow-up"); + expect(queue.getVisibleProjection()).toMatchObject({ + queuedMessages: ["User follow-up"], + displayText: "User follow-up", + queueDispatchMode: "tool-end", + }); expect(queue.getQueueDispatchMode()).toBe("tool-end"); - expect(queue.getVisibleQueueDispatchMode()).toBe("tool-end"); const background = queue.dequeueNext(); expect(background.message).toBe("Background monitor wake"); expect(background.internal).toMatchObject({ synthetic: true, agentInitiated: true }); diff --git a/src/node/services/messageQueue.ts b/src/node/services/messageQueue.ts index 83091339cf5..3eac131a8e4 100644 --- a/src/node/services/messageQueue.ts +++ b/src/node/services/messageQueue.ts @@ -103,6 +103,15 @@ type QueueClearCallbacks = Pick< "onCanceled" | "onAcceptedPreStreamFailure" >; +export interface MessageQueueVisibleProjection { + queuedMessages: string[]; + displayText: string; + fileParts: FilePart[]; + reviews?: ReviewNoteData[]; + queueDispatchMode: QueueDispatchMode; + hasCompactionRequest: boolean; +} + /** * One dispatchable unit in the queue. Plain follow-up messages batch into a single * entry (joined text, accumulated file parts); "special" sends (compaction requests, @@ -536,16 +545,30 @@ export class MessageQueue { return reviews.length > 0 ? reviews : undefined; } + private getVisibleProjectionForEntries( + entries: readonly QueueEntry[], + queueDispatchMode: QueueDispatchMode + ): MessageQueueVisibleProjection { + return { + queuedMessages: this.getMessagesForEntries(entries), + displayText: this.getDisplayTextForEntries(entries), + fileParts: this.getFilePartsForEntries(entries), + reviews: this.getReviewsForEntries(entries), + queueDispatchMode, + hasCompactionRequest: entries.some((entry) => isCompactionMetadata(entry.muxMetadata)), + }; + } + + getVisibleProjection(): MessageQueueVisibleProjection { + const entries = this.getVisibleEntries(); + return this.getVisibleProjectionForEntries(entries, this.getVisibleQueueDispatchMode()); + } + /** Get all queued message texts across entries (including synthetic entries). */ getMessages(): string[] { return this.getMessagesForEntries(this.entries); } - /** Get user-visible queued message texts for the renderer/composer. */ - getVisibleMessages(): string[] { - return this.getMessagesForEntries(this.getVisibleEntries()); - } - /** * Get display text for queued messages. * - A single-message compaction/agent-skill entry shows its rawCommand (/compact, /{skill}) @@ -555,36 +578,16 @@ export class MessageQueue { return this.getDisplayTextForEntries(this.entries); } - /** Get display text for user-visible entries only. */ - getVisibleDisplayText(): string { - return this.getDisplayTextForEntries(this.getVisibleEntries()); - } - /** Get accumulated file parts across all entries. */ getFileParts(): FilePart[] { return this.getFilePartsForEntries(this.entries); } - /** Get accumulated file parts for user-visible entries only. */ - getVisibleFileParts(): FilePart[] { - return this.getFilePartsForEntries(this.getVisibleEntries()); - } - /** Get reviews across all entries' metadata. */ getReviews(): ReviewNoteData[] | undefined { return this.getReviewsForEntries(this.entries); } - /** Get reviews across user-visible entries' metadata only. */ - getVisibleReviews(): ReviewNoteData[] | undefined { - return this.getReviewsForEntries(this.getVisibleEntries()); - } - - /** Whether a user-visible queued entry is a compaction request. */ - hasVisibleCompactionRequest(): boolean { - return this.getVisibleEntries().some((entry) => isCompactionMetadata(entry.muxMetadata)); - } - /** * Cancellation callbacks for every pending entry, in queue order. * Callers must notify each one when clearing the queue. @@ -701,12 +704,16 @@ export class MessageQueue { message: string; options?: SendMessageOptions & { fileParts?: FilePart[] }; internal?: QueuedMessageInternalOptions; + visibleProjection?: MessageQueueVisibleProjection; } { const entry = this.entries.shift(); if (entry === undefined) { return { message: "" }; } + const visibleProjection = entry.userAuthored + ? this.getVisibleProjectionForEntries([entry], entry.dispatchMode) + : undefined; const joinedMessages = entry.messages.join("\n"); const options = entry.latestOptions ? (() => { @@ -748,7 +755,7 @@ export class MessageQueue { } : undefined; - return { message: joinedMessages, options, internal }; + return { message: joinedMessages, options, internal, visibleProjection }; } /** diff --git a/tests/ui/chat/queuedMessageBanner.test.tsx b/tests/ui/chat/queuedMessageBanner.test.tsx index 5d19e592b85..b3ce8e57b75 100644 --- a/tests/ui/chat/queuedMessageBanner.test.tsx +++ b/tests/ui/chat/queuedMessageBanner.test.tsx @@ -65,6 +65,23 @@ describe("QueuedMessage banner", () => { expect(view.getAllByRole("menuitem")).toHaveLength(3); }); + test("keeps a dispatching message visible without editable queue actions", () => { + const view = render( + {})} + onChangeDispatchMode={mock(async () => {})} + onSendImmediately={mock(async () => {})} + /> + ); + + expect(view.getByText("Review this change before sending")).toBeTruthy(); + expect(view.queryByRole("button", { name: "Edit" })).toBeNull(); + expect(view.getByRole("button", { name: "Sending" }).hasAttribute("disabled")).toBe(true); + expect(view.queryByRole("menu")).toBeNull(); + }); + test("renders queued preview text and step-dispatch label", () => { const view = render(); From 6414887b99589ef6c22371c9b2086bf9803cc59a Mon Sep 17 00:00:00 2001 From: Ammar Date: Fri, 21 Aug 2026 23:03:40 -0500 Subject: [PATCH 3/4] =?UTF-8?q?=F0=9F=A4=96=20tests:=20cover=20queued=20ac?= =?UTF-8?q?ceptance=20callback=20wrapping?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Update the disposal-race assertion to verify synthetic attribution remains intact while allowing the queue handoff to add its durable-acceptance callback. --- _Generated with `xum` • Model: `openai:gpt-5.6-sol` • Thinking: `xhigh` • Cost: `$64.30`_ --- src/node/services/agentSession.disposeRace.test.ts | 13 +++++++------ 1 file changed, 7 insertions(+), 6 deletions(-) diff --git a/src/node/services/agentSession.disposeRace.test.ts b/src/node/services/agentSession.disposeRace.test.ts index 0b848e02180..570f39da7c3 100644 --- a/src/node/services/agentSession.disposeRace.test.ts +++ b/src/node/services/agentSession.disposeRace.test.ts @@ -426,7 +426,7 @@ describe("AgentSession disposal race conditions", () => { ( _message: string, _options?: { model: string; agentId: string }, - _internal?: { synthetic?: boolean } + _internal?: { synthetic?: boolean; onAccepted?: () => Promise } ) => Promise.resolve(Ok(undefined)) ); @@ -440,10 +440,11 @@ describe("AgentSession disposal race conditions", () => { session.sendQueuedMessages(); expect(sendMessage).toHaveBeenCalledTimes(1); - expect(sendMessage).toHaveBeenCalledWith( - "Background compaction request", - expect.objectContaining({ model: "anthropic:claude-sonnet-4-5", agentId: "compact" }), - { synthetic: true } - ); + const call = sendMessage.mock.calls[0]; + if (!call) throw new Error("Expected queued message dispatch"); + expect(call[0]).toBe("Background compaction request"); + expect(call[1]).toMatchObject({ model: "anthropic:claude-sonnet-4-5", agentId: "compact" }); + expect(call[2]?.synthetic).toBe(true); + expect(typeof call[2]?.onAccepted).toBe("function"); }); }); From ecd686cfd5bba5021c305c8582ed3a264a6479dd Mon Sep 17 00:00:00 2001 From: Ammar Date: Fri, 21 Aug 2026 23:09:30 -0500 Subject: [PATCH 4/4] =?UTF-8?q?=F0=9F=A4=96=20fix:=20make=20dispatch=20sta?= =?UTF-8?q?te=20replay-aware?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Expose dispatching state from the authoritative queue projection instead of general stream startup, and omit an in-flight card from reconnect snapshots once its user row is already durable in replayed history. --- _Generated with `xum` • Model: `openai:gpt-5.6-sol` • Thinking: `xhigh` • Cost: `$64.30`_ --- src/browser/components/ChatPane/ChatPane.tsx | 7 ++--- src/browser/features/ChatInput/index.tsx | 2 +- .../features/Messages/QueuedMessage.tsx | 3 +- src/browser/stores/WorkspaceStore.ts | 2 ++ src/common/orpc/schemas/stream.ts | 2 ++ src/common/types/message.ts | 2 ++ .../agentSession.queueDispatch.test.ts | 28 +++++++++++++++++++ src/node/services/agentSession.ts | 21 ++++++++++---- tests/ui/chat/queuedMessageBanner.test.tsx | 3 +- 9 files changed, 55 insertions(+), 15 deletions(-) diff --git a/src/browser/components/ChatPane/ChatPane.tsx b/src/browser/components/ChatPane/ChatPane.tsx index 93db1a74181..1f5b5d710d5 100644 --- a/src/browser/components/ChatPane/ChatPane.tsx +++ b/src/browser/components/ChatPane/ChatPane.tsx @@ -860,10 +860,10 @@ const ChatPaneContent: React.FC = (props) => { const handleEditQueuedMessage = useCallback(async () => { const queuedMessage = workspaceState?.queuedMessage; - if (!queuedMessage || workspaceState?.isStreamStarting) return; + if (!queuedMessage || queuedMessage.isDispatching) return; await restoreQueuedDraft(queuedMessage); - }, [restoreQueuedDraft, workspaceState?.isStreamStarting, workspaceState?.queuedMessage]); + }, [restoreQueuedDraft, workspaceState?.queuedMessage]); const sendQueuedImmediatelyInFlightRef = useRef(null); @@ -958,7 +958,7 @@ const ChatPaneContent: React.FC = (props) => { if (!current) return; if (current.queuedMessage) { - if (!current.isStreamStarting) { + if (!current.queuedMessage.isDispatching) { await restoreQueuedDraft(current.queuedMessage); } return; @@ -1820,7 +1820,6 @@ const ChatInputPane: React.FC = (props) => { node: ( void props.onEditQueuedMessage()} onChangeDispatchMode={props.onQueuedDispatchModeChange} onActionError={props.onQueuedActionError} diff --git a/src/browser/features/ChatInput/index.tsx b/src/browser/features/ChatInput/index.tsx index 3ecba19becb..41f25e8e5ba 100644 --- a/src/browser/features/ChatInput/index.tsx +++ b/src/browser/features/ChatInput/index.tsx @@ -3356,7 +3356,7 @@ const ChatInputInner: React.FC = (props) => { variant === "workspace" && !editingMessageForUi && props.queuedMessage != null && - !isStreamStarting && + props.queuedMessage.isDispatching !== true && input.trim() === "" && attachments.length === 0 && reviewPanelItems.length === 0; diff --git a/src/browser/features/Messages/QueuedMessage.tsx b/src/browser/features/Messages/QueuedMessage.tsx index e0238c73fbe..c2a85322625 100644 --- a/src/browser/features/Messages/QueuedMessage.tsx +++ b/src/browser/features/Messages/QueuedMessage.tsx @@ -11,7 +11,6 @@ import { cn } from "@/common/lib/utils"; interface QueuedMessageProps { message: QueuedMessageType; - isDispatching?: boolean; className?: string; onEdit?: () => void; onChangeDispatchMode?: (mode: QueueDispatchMode) => Promise; @@ -45,7 +44,7 @@ export const QueuedMessage: React.FC = (props) => { const queueDispatchMode = props.message.queueDispatchMode ?? "tool-end"; const queueStatusLabel = queueDispatchMode === "turn-end" ? "Sends after this turn" : "Sends after this step"; - const isDispatching = props.isDispatching === true; + const isDispatching = props.message.isDispatching === true; const isActionPending = pendingAction != null; const handleDispatchModeChange = (mode: QueueDispatchMode) => { diff --git a/src/browser/stores/WorkspaceStore.ts b/src/browser/stores/WorkspaceStore.ts index 527855f91e6..32d6d50791c 100644 --- a/src/browser/stores/WorkspaceStore.ts +++ b/src/browser/stores/WorkspaceStore.ts @@ -1132,12 +1132,14 @@ export class WorkspaceStore { data.reviews?.map((review) => [review.filePath, review.lineRange]) ?? [], data.queueDispatchMode, data.hasCompactionRequest, + data.isDispatching, ])}`, content: data.displayText, fileParts: data.fileParts, reviews: data.reviews, queueDispatchMode: data.queueDispatchMode, hasCompactionRequest: data.hasCompactionRequest, + isDispatching: data.isDispatching, } : null; diff --git a/src/common/orpc/schemas/stream.ts b/src/common/orpc/schemas/stream.ts index 4b328676ac9..87cb8de1282 100644 --- a/src/common/orpc/schemas/stream.ts +++ b/src/common/orpc/schemas/stream.ts @@ -642,6 +642,8 @@ export const QueuedMessageChangedEventSchema = z.object({ queueDispatchMode: z.enum(["tool-end", "turn-end"]).optional(), /** True when the queued message is a compaction request (/compact) */ hasCompactionRequest: z.boolean().optional(), + /** True when the visible projection includes an entry being durably accepted. */ + isDispatching: z.boolean().optional(), }); export const RestoreToInputEventSchema = z.object({ diff --git a/src/common/types/message.ts b/src/common/types/message.ts index fd709781b6d..bcd23b63c74 100644 --- a/src/common/types/message.ts +++ b/src/common/types/message.ts @@ -1079,6 +1079,8 @@ export interface QueuedMessage { queueDispatchMode?: QueueDispatchMode; /** True when the queued message is a compaction request (/compact) */ hasCompactionRequest?: boolean; + /** True when this projection includes an entry being durably accepted. */ + isDispatching?: boolean; } /** Keep every snapshot kind here so history scans and edits retain it with its user message. */ diff --git a/src/node/services/agentSession.queueDispatch.test.ts b/src/node/services/agentSession.queueDispatch.test.ts index a67069aaeb9..8b02b4f6fab 100644 --- a/src/node/services/agentSession.queueDispatch.test.ts +++ b/src/node/services/agentSession.queueDispatch.test.ts @@ -516,8 +516,21 @@ 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 goalSyncStarted = Promise.withResolvers(); + const goalSyncRelease = Promise.withResolvers(); + const workspaceGoalService = { + assertPricedModelForBudgetedGoal: mock(() => Promise.resolve(Ok(undefined))), + syncGoalModeWithChatTail: mock(async () => { + goalSyncStarted.resolve(); + await goalSyncRelease.promise; + }), + clearPendingContinuationForManualUserMessage: mock(() => undefined), + acknowledgeUser: mock(() => Promise.resolve(null)), + } as unknown as WorkspaceGoalService; const { session, cleanup, historyService, events } = await createAgentSessionHarness({ workspaceId, + workspaceGoalService, + initStateManagerOverrides: { replayInit: mock(() => Promise.resolve()) }, captureEvents: true, }); const originalAppend = historyService.appendToHistory.bind(historyService); @@ -559,6 +572,20 @@ describe("AgentSession queued message tool-call dispatch", () => { expect(latestQueueEvent()?.queuedMessages).toEqual([followUp, nextFollowUp]); appendRelease.resolve(); + await goalSyncStarted.promise; + + // Reconnect replay already includes the persisted user row, so its queue snapshot must + // omit that dispatch while retaining later queued input. + const replayEvents: Array<(typeof events)[number]> = []; + await session.replayHistory(({ message }) => replayEvents.push(message)); + expect(replayEvents.some(isFollowUpUserMessage)).toBe(true); + const replayQueueEvent = replayEvents.find( + (event) => event.type === "queued-message-changed" + ); + expect(replayQueueEvent?.queuedMessages).toEqual([nextFollowUp]); + expect(replayQueueEvent?.isDispatching).toBe(false); + + goalSyncRelease.resolve(); expect(await waitForCondition(() => events.some(isFollowUpUserMessage))).toBe(true); expect( await waitForCondition(() => latestQueueEvent()?.queuedMessages.join() === nextFollowUp) @@ -575,6 +602,7 @@ describe("AgentSession queued message tool-call dispatch", () => { expect(handoffIndex).toBeGreaterThan(userMessageIndex); } finally { appendRelease.resolve(); + goalSyncRelease.resolve(); appendSpy.mockRestore(); session.dispose(); await cleanup(); diff --git a/src/node/services/agentSession.ts b/src/node/services/agentSession.ts index 962f2f39fdd..d2cfa38e1a1 100644 --- a/src/node/services/agentSession.ts +++ b/src/node/services/agentSession.ts @@ -710,6 +710,7 @@ export class AgentSession { private dispatchingQueuedEntry?: { muxMetadata?: unknown; visibleProjection?: MessageQueueVisibleProjection; + persisted: boolean; }; /** Correlation of the direct send currently in the PREPARING phase, if any. */ @@ -2505,7 +2506,7 @@ export class AgentSession { // retain a user entry that is between dequeue and durable transcript emission. listener({ workspaceId: this.workspaceId, - message: this.getQueuedMessageChangedEvent(), + message: this.getQueuedMessageChangedEvent(false), }); // Rehydrate pending auto-retry countdown state on reconnect/reload so @@ -3249,6 +3250,9 @@ export class AgentSession { if (await cancelBeforeAcceptance()) { return Ok(undefined); } + if (this.dispatchingQueuedEntry) { + this.dispatchingQueuedEntry.persisted = true; + } } // Goal synchronization can mutate goal.json based on this durable user row. Once it begins, the @@ -5842,12 +5846,15 @@ export class AgentSession { } } - private getQueuedMessageChangedEvent(): Extract< - WorkspaceChatMessage, - { type: "queued-message-changed" } - > { + private getQueuedMessageChangedEvent( + includePersistedDispatching = true + ): Extract { const pending = this.messageQueue.getVisibleProjection(); - const dispatching = this.dispatchingQueuedEntry?.visibleProjection; + const dispatchingState = this.dispatchingQueuedEntry; + const dispatching = + !includePersistedDispatching && dispatchingState?.persisted + ? undefined + : dispatchingState?.visibleProjection; const reviews = [...(dispatching?.reviews ?? []), ...(pending.reviews ?? [])]; return { @@ -5861,6 +5868,7 @@ export class AgentSession { queueDispatchMode: dispatching?.queueDispatchMode ?? pending.queueDispatchMode, hasCompactionRequest: dispatching?.hasCompactionRequest === true || pending.hasCompactionRequest, + isDispatching: dispatching != null, }; } @@ -5905,6 +5913,7 @@ export class AgentSession { this.dispatchingQueuedEntry = { muxMetadata: options?.muxMetadata, visibleProjection, + persisted: false, }; // PREPARING disables queue actions while the authoritative projection includes // the in-flight entry, preventing stale Edit/Send-now operations during persistence. diff --git a/tests/ui/chat/queuedMessageBanner.test.tsx b/tests/ui/chat/queuedMessageBanner.test.tsx index b3ce8e57b75..6b819911c73 100644 --- a/tests/ui/chat/queuedMessageBanner.test.tsx +++ b/tests/ui/chat/queuedMessageBanner.test.tsx @@ -68,8 +68,7 @@ describe("QueuedMessage banner", () => { test("keeps a dispatching message visible without editable queue actions", () => { const view = render( {})} onChangeDispatchMode={mock(async () => {})} onSendImmediately={mock(async () => {})}