diff --git a/src/browser/features/ChatInput/index.tsx b/src/browser/features/ChatInput/index.tsx index d1c8e6b112..9d699b40d9 100644 --- a/src/browser/features/ChatInput/index.tsx +++ b/src/browser/features/ChatInput/index.tsx @@ -1741,8 +1741,16 @@ const ChatInputInner: React.FC = (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 = (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 }>) => { diff --git a/src/browser/stores/WorkspaceStore.test.ts b/src/browser/stores/WorkspaceStore.test.ts index 60f929876d..f8d9fc6857 100644 --- a/src/browser/stores/WorkspaceStore.test.ts +++ b/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"; diff --git a/src/browser/stores/WorkspaceStore.ts b/src/browser/stores/WorkspaceStore.ts index 85b1fbb567..b303720103 100644 --- a/src/browser/stores/WorkspaceStore.ts +++ b/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, }) ); }, diff --git a/src/common/constants/events.ts b/src/common/constants/events.ts index df2b5ffbac..7da8be0e4f 100644 --- a/src/common/constants/events.ts +++ b/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; diff --git a/src/common/orpc/schemas/stream.ts b/src/common/orpc/schemas/stream.ts index 45497ae6bb..167cede0ea 100644 --- a/src/common/orpc/schemas/stream.ts +++ b/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(), diff --git a/src/node/services/agentSession.queueDispatch.test.ts b/src/node/services/agentSession.queueDispatch.test.ts index 24abb43bec..5567ca92eb 100644 --- a/src/node/services/agentSession.queueDispatch.test.ts +++ b/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({ diff --git a/src/node/services/agentSession.ts b/src/node/services/agentSession.ts index 7104cdc907..da8bd9dfc8 100644 --- a/src/node/services/agentSession.ts +++ b/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(); }); } } diff --git a/src/node/services/messageQueue.test.ts b/src/node/services/messageQueue.test.ts index 7bf8103751..fd7f02ac49 100644 --- a/src/node/services/messageQueue.test.ts +++ b/src/node/services/messageQueue.test.ts @@ -18,6 +18,31 @@ describe("MessageQueue", () => { expect(queue.getDisplayText()).toBe("First message\nSecond message"); }); + it("should hide synthetic background entries from the user-visible queue snapshot", () => { + queue.add( + "Background monitor wake", + { model: "gpt-4", agentId: "exec", queueDispatchMode: "tool-end" }, + { synthetic: true, agentInitiated: true } + ); + queue.add("User follow-up", { + model: "gpt-4", + agentId: "exec", + queueDispatchMode: "turn-end", + }); + + // Dispatch state still includes both FIFO entries, but the composer/queue card + // only exposes the user's own input and its chosen dispatch boundary. + 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.getQueueDispatchMode()).toBe("tool-end"); + expect(queue.getVisibleQueueDispatchMode()).toBe("turn-end"); + const background = queue.dequeueNext(); + expect(background.message).toBe("Background monitor wake"); + expect(background.internal).toMatchObject({ synthetic: true, agentInitiated: true }); + expect(queue.dequeueNext().message).toBe("User follow-up"); + }); + it("should return rawCommand for compaction request", () => { const metadata: MuxMessageMetadata = { type: "compaction-request", @@ -36,7 +61,7 @@ describe("MessageQueue", () => { expect(queue.getDisplayText()).toBe("/compact -t 3000"); }); - it("should throw when adding compaction after normal message", () => { + it("should queue compaction after normal message as its own entry", () => { queue.add("First message"); const metadata: MuxMessageMetadata = { @@ -51,11 +76,19 @@ describe("MessageQueue", () => { muxMetadata: metadata, }; - // Compaction requests cannot be mixed with other messages to prevent - // silent failures where compaction metadata would be lost - expect(() => queue.add("Summarize this conversation...", options)).toThrow( - /Cannot queue compaction request/ - ); + // Compaction must not adopt earlier batched texts; it starts a new entry + // and dispatches after the pending messages instead of erroring. + expect(queue.add("Summarize this conversation...", options)).toBe(true); + expect(queue.getDisplayText()).toBe("First message\n/compact"); + + const first = queue.dequeueNext(); + expect(first.message).toBe("First message"); + expect(first.options?.muxMetadata).toBeUndefined(); + + const second = queue.dequeueNext(); + expect(second.message).toBe("Summarize this conversation..."); + expect((second.options?.muxMetadata as MuxMessageMetadata).type).toBe("compaction-request"); + expect(queue.isEmpty()).toBe(true); }); it("should return joined messages when metadata type is not compaction-request", () => { @@ -198,7 +231,7 @@ describe("MessageQueue", () => { expect(queue.getQueueDispatchMode()).toBe("tool-end"); }); - it("should preserve turn-end mode when compaction enqueue is rejected", () => { + it("should report tool-end when any pending entry wants tool-end", () => { const validOptions: SendMessageOptions = { model: "gpt-4", agentId: "exec", @@ -211,52 +244,41 @@ describe("MessageQueue", () => { expect(queue.getQueueDispatchMode()).toBe("turn-end"); const metadata: MuxMessageMetadata = { - type: "compaction-request", - rawCommand: "/compact", - parsed: {}, + type: "agent-skill", + rawCommand: "/init", + skillName: "init", + scope: "built-in", }; - expect(() => - queue.add("summarize", { - ...validOptions, - queueDispatchMode: "tool-end", - muxMetadata: metadata, - }) - ).toThrow(/Cannot queue compaction request/); + // A later entry queued for tool-end makes the whole queue drain at tool-end + // (sticky, matching pre-entry batching semantics). + queue.add("run skill", { + ...validOptions, + queueDispatchMode: "tool-end", + muxMetadata: metadata, + }); + expect(queue.getQueueDispatchMode()).toBe("tool-end"); - expect(queue.getQueueDispatchMode()).toBe("turn-end"); - expect(queue.getMessages()).toHaveLength(1); + // Once the tool-end entry dispatches, the remaining entries' mode wins again. + queue.dequeueNext(); // "wait for turn end" (FIFO head) + expect(queue.getQueueDispatchMode()).toBe("tool-end"); + queue.dequeueNext(); // the tool-end skill entry + expect(queue.getQueueDispatchMode()).toBe("tool-end"); // empty queue default }); - it("should preserve turn-end mode when agent-skill enqueue is rejected", () => { - const validOptions: SendMessageOptions = { + it("should keep per-entry modes so a turn-end tail does not downgrade the queue", () => { + queue.add("interrupt soon", { + model: "gpt-4", + agentId: "exec", + queueDispatchMode: "tool-end", + }); + queue.add("later is fine", { model: "gpt-4", agentId: "exec", - }; - - queue.add("wait for turn end", { - ...validOptions, queueDispatchMode: "turn-end", }); - expect(queue.getQueueDispatchMode()).toBe("turn-end"); - const metadata: MuxMessageMetadata = { - type: "agent-skill", - rawCommand: "/init", - skillName: "init", - scope: "built-in", - }; - - expect(() => - queue.add("run skill", { - ...validOptions, - queueDispatchMode: "tool-end", - muxMetadata: metadata, - }) - ).toThrow(/Cannot queue agent skill/); - - expect(queue.getQueueDispatchMode()).toBe("turn-end"); - expect(queue.getMessages()).toHaveLength(1); + expect(queue.getQueueDispatchMode()).toBe("tool-end"); }); it("should reset mode to tool-end when cleared", () => { @@ -280,17 +302,48 @@ describe("MessageQueue", () => { turnId: "turn-1", }; - it("should not batch workspace-turn follow-ups with other queued messages", () => { - queue.add("Follow up", { model: "gpt-4", agentId: "exec", muxMetadata: metadata }); + it("should queue user messages behind a workspace-turn follow-up instead of erroring", () => { + // Regression: sending a message while an internal workspace-turn follow-up + // was queued used to fail with "Cannot queue additional messages". + const onAccepted = () => undefined; + queue.add( + "Follow up", + { model: "gpt-4", agentId: "exec", muxMetadata: metadata }, + { agentInitiated: true, onAccepted } + ); - expect(() => queue.add("Second message")).toThrow(/workspace turn/); + expect(queue.add("Second message")).toBe(true); + expect(queue.getMessages()).toEqual(["Follow up", "Second message"]); - queue.clear(); - queue.add("Normal message"); + // FIFO: the workspace turn dispatches first with its metadata + callbacks... + const first = queue.dequeueNext(); + expect(first.message).toBe("Follow up"); + expect((first.options?.muxMetadata as MuxMessageMetadata).type).toBe("workspace-turn-task"); + expect(first.internal?.onAccepted).toBe(onAccepted); + + // ...and the user message dispatches after it, without adopting either. + const second = queue.dequeueNext(); + expect(second.message).toBe("Second message"); + expect(second.options?.muxMetadata).toBeUndefined(); + expect(second.internal).toBeUndefined(); + expect(queue.isEmpty()).toBe(true); + }); - expect(() => + it("should queue a workspace-turn follow-up behind pending messages", () => { + queue.add("Normal message"); + expect( queue.add("Follow up", { model: "gpt-4", agentId: "exec", muxMetadata: metadata }) - ).toThrow(/workspace turn/); + ).toBe(true); + expect(queue.hasWorkspaceTurn("wst_followup")).toBe(true); + + const first = queue.dequeueNext(); + expect(first.message).toBe("Normal message"); + expect(first.options?.muxMetadata).toBeUndefined(); + expect(queue.hasWorkspaceTurn("wst_followup")).toBe(true); + + const second = queue.dequeueNext(); + expect((second.options?.muxMetadata as MuxMessageMetadata).type).toBe("workspace-turn-task"); + expect(queue.hasWorkspaceTurn("wst_followup")).toBe(false); }); it("should preserve internal workspace-turn callbacks", () => { @@ -304,12 +357,61 @@ describe("MessageQueue", () => { { agentInitiated: true, onAccepted, onAcceptedPreStreamFailure, onCanceled } ); - const { internal } = queue.produceMessage(); + const clearCallbacks = queue.getClearCallbacks(); + expect(clearCallbacks).toHaveLength(1); + expect(clearCallbacks[0].onCanceled).toBe(onCanceled); + + const { internal } = queue.dequeueNext(); expect(internal?.agentInitiated).toBe(true); expect(internal?.onAccepted).toBe(onAccepted); expect(internal?.onAcceptedPreStreamFailure).toBe(onAcceptedPreStreamFailure); expect(internal?.onCanceled).toBe(onCanceled); - expect(queue.getClearCallbacks().onCanceled).toBe(onCanceled); + }); + + it("removeWorkspaceTurn drops only the matching entry and keeps user messages", () => { + const onCanceled = () => undefined; + queue.add("User message before"); + queue.add( + "Follow up", + { model: "gpt-4", agentId: "exec", muxMetadata: metadata }, + { agentInitiated: true, onCanceled } + ); + queue.add("User message after"); + + expect(queue.removeWorkspaceTurn("wst_other")).toBeNull(); + + const callbacks = queue.removeWorkspaceTurn("wst_followup"); + expect(callbacks?.onCanceled).toBe(onCanceled); + expect(queue.hasWorkspaceTurn("wst_followup")).toBe(false); + // Unrelated queued input survives the targeted cancel. + expect(queue.getMessages()).toEqual(["User message before", "User message after"]); + }); + + it("should report clear callbacks for every pending entry", () => { + const onCanceledFirst = () => undefined; + const onCanceledSecond = () => undefined; + + queue.add( + "First follow up", + { model: "gpt-4", agentId: "exec" }, + { + onCanceled: onCanceledFirst, + } + ); + queue.add("User message in between"); + queue.add( + "Second follow up", + { model: "gpt-4", agentId: "exec" }, + { + onCanceled: onCanceledSecond, + } + ); + + const clearCallbacks = queue.getClearCallbacks(); + expect(clearCallbacks.map((callbacks) => callbacks.onCanceled)).toEqual([ + onCanceledFirst, + onCanceledSecond, + ]); }); }); @@ -321,7 +423,7 @@ describe("MessageQueue", () => { goalInterventionPolicy: "steer", }); - const { options } = queue.produceMessage(); + const { options } = queue.dequeueNext(); expect(options?.goalInterventionPolicy).toBe("steer"); }); @@ -338,7 +440,7 @@ describe("MessageQueue", () => { goalInterventionPolicy: "steer", }); - const { options } = queue.produceMessage(); + const { options } = queue.dequeueNext(); expect(options?.goalInterventionPolicy).toBe("pause"); }); @@ -353,7 +455,7 @@ describe("MessageQueue", () => { queue.clear(); queue.add("Plain follow-up", { model: "gpt-4", agentId: "exec" }); - const { options } = queue.produceMessage(); + const { options } = queue.dequeueNext(); expect(options?.goalInterventionPolicy).toBeUndefined(); }); }); @@ -417,6 +519,38 @@ describe("MessageQueue", () => { ).toBe(false); expect(queue.getMessages()).toEqual(["User follow-up", "Heartbeat"]); }); + + it("prioritizeNextUserEntry moves user input ahead of hidden background work", () => { + queue.add("Background wake", { model: "gpt-4", agentId: "exec" }, { synthetic: true }); + queue.add("User send now", { model: "gpt-4", agentId: "exec" }); + queue.add( + "Later background wake", + { model: "gpt-4", agentId: "exec" }, + { + synthetic: true, + } + ); + + expect(queue.prioritizeNextUserEntry()).toBe(true); + expect(queue.dequeueNext().message).toBe("User send now"); + expect(queue.dequeueNext().message).toBe("Background wake"); + expect(queue.dequeueNext().message).toBe("Later background wake"); + expect(queue.prioritizeNextUserEntry()).toBe(false); + }); + + it("should release a dedupe key when its entry dispatches", () => { + queue.addOnce("Heartbeat", { model: "gpt-4", agentId: "exec" }, "heartbeat-request"); + expect(queue.hasDedupeKey("heartbeat-request")).toBe(true); + + queue.dequeueNext(); + + // The key belongs to the dispatched entry, so the next scheduled message + // can enqueue again even if other entries were still pending. + expect(queue.hasDedupeKey("heartbeat-request")).toBe(false); + expect( + queue.addOnce("Heartbeat", { model: "gpt-4", agentId: "exec" }, "heartbeat-request") + ).toBe(true); + }); }); describe("multi-message batching", () => { @@ -449,8 +583,8 @@ describe("MessageQueue", () => { // getMessages includes both expect(queue.getMessages()).toEqual(["Summarize...", "And then do this follow-up task"]); - // produceMessage preserves compaction metadata from first message - const { message, options } = queue.produceMessage(); + // dequeueNext preserves compaction metadata from the entry's first message + const { message, options } = queue.dequeueNext(); expect(message).toBe("Summarize...\nAnd then do this follow-up task"); const muxMeta = options?.muxMetadata as MuxMessageMetadata; expect(muxMeta.type).toBe("compaction-request"); @@ -459,7 +593,7 @@ describe("MessageQueue", () => { } }); - it("should throw when adding agent-skill invocation after normal message", () => { + it("should queue an agent-skill invocation after a normal message as its own entry", () => { queue.add("First message"); const metadata: MuxMessageMetadata = { @@ -475,12 +609,20 @@ describe("MessageQueue", () => { muxMetadata: metadata, }; - expect(() => queue.add("Using skill init", options)).toThrow( - /Cannot queue agent skill invocation/ - ); + // Skill metadata must not adopt earlier batched texts; the invocation + // dispatches after the pending messages instead of erroring. + expect(queue.add("Using skill init", options)).toBe(true); + expect(queue.getDisplayText()).toBe("First message\n/init"); + + const first = queue.dequeueNext(); + expect(first.message).toBe("First message"); + expect(first.options?.muxMetadata).toBeUndefined(); + + const second = queue.dequeueNext(); + expect((second.options?.muxMetadata as MuxMessageMetadata).type).toBe("agent-skill"); }); - it("should throw when adding normal message after agent-skill invocation", () => { + it("should queue a normal message behind an agent-skill invocation without leaking metadata", () => { const metadata: MuxMessageMetadata = { type: "agent-skill", rawCommand: "/init", @@ -496,16 +638,25 @@ describe("MessageQueue", () => { expect(queue.getDisplayText()).toBe("/init"); - expect(() => queue.add("Follow-up message")).toThrow( - /agent skill invocation is already queued/ - ); + // Skill entries are sealed: the follow-up starts a new entry and dispatches + // after the skill turn instead of adopting its metadata (or erroring). + expect(queue.add("Follow-up message")).toBe(true); + expect(queue.getDisplayText()).toBe("/init\nFollow-up message"); + + const first = queue.dequeueNext(); + expect(first.message).toBe("Use skill init"); + expect((first.options?.muxMetadata as MuxMessageMetadata).type).toBe("agent-skill"); + + const second = queue.dequeueNext(); + expect(second.message).toBe("Follow-up message"); + expect(second.options?.muxMetadata).toBeUndefined(); }); it("should produce combined message for API call", () => { queue.add("First message", { model: "gpt-4", agentId: "exec" }); queue.add("Second message"); - const { message, options } = queue.produceMessage(); + const { message, options } = queue.dequeueNext(); // Messages are joined with newlines expect(message).toBe("First message\nSecond message"); @@ -549,16 +700,21 @@ describe("MessageQueue", () => { { synthetic: true } ); - const { internal } = queue.produceMessage(); + const { internal } = queue.dequeueNext(); expect(internal).toEqual({ synthetic: true }); }); - it("should not mark mixed synthetic + user batches as synthetic", () => { + it("should keep synthetic and user messages in separate entries", () => { queue.add("Idle compaction", { model: "gpt-4", agentId: "compact" }, { synthetic: true }); queue.add("User follow-up", { model: "gpt-4", agentId: "exec" }); - const { internal } = queue.produceMessage(); - expect(internal).toBeUndefined(); + const background = queue.dequeueNext(); + expect(background.message).toBe("Idle compaction"); + expect(background.internal).toEqual({ synthetic: true }); + + const user = queue.dequeueNext(); + expect(user.message).toBe("User follow-up"); + expect(user.internal).toBeUndefined(); }); it("should clear synthetic flag when queue is cleared", () => { @@ -566,7 +722,7 @@ describe("MessageQueue", () => { queue.clear(); queue.add("User message", { model: "gpt-4", agentId: "exec" }); - const { internal } = queue.produceMessage(); + const { internal } = queue.dequeueNext(); expect(internal).toBeUndefined(); }); }); @@ -676,7 +832,7 @@ describe("MessageQueue", () => { const image = { url: "data:image/png;base64,abc", mediaType: "image/png" }; queue.add("", { model: "gpt-4", agentId: "exec", fileParts: [image] }); - const { message, options } = queue.produceMessage(); + const { message, options } = queue.dequeueNext(); expect(message).toBe(""); expect(options?.fileParts).toEqual([image]); diff --git a/src/node/services/messageQueue.ts b/src/node/services/messageQueue.ts index 76991cb068..bfcb2e754f 100644 --- a/src/node/services/messageQueue.ts +++ b/src/node/services/messageQueue.ts @@ -32,7 +32,7 @@ function isCompactionMetadata(meta: unknown): meta is CompactionMetadata { return obj.type === "compaction-request" && typeof obj.rawCommand === "string"; } -// Workspace-turn task metadata must stay attached to exactly one queued message; +// Workspace-turn task metadata must stay attached to exactly one queued entry; // otherwise a batched follow-up would leave one durable task handle with no matching stream-end. interface WorkspaceTurnMetadata { type: "workspace-turn-task"; @@ -68,25 +68,6 @@ type GoalInterventionPolicy = NonNullable; -/** - * Queue for messages sent during active streaming. - * - * Stores: - * - Message texts (accumulated) - * - First muxMetadata (preserved - never overwritten by subsequent adds) - * - Latest options (model, etc. - updated on each add) - * - File parts (accumulated across all messages) - * - * IMPORTANT: - * - Compaction requests must preserve their muxMetadata even when follow-up messages are queued. - * - Agent-skill invocations cannot be batched with other messages; otherwise the skill metadata would - * “leak” onto later queued sends. - * - * Display logic: - * - Single compaction request → shows rawCommand (/compact) - * - Single agent-skill invocation → shows rawCommand (/{skill}) - * - Multiple messages → shows all actual message texts - */ interface QueuedMessageInternalOptions { synthetic?: boolean; agentInitiated?: boolean; @@ -95,46 +76,111 @@ interface QueuedMessageInternalOptions { onCanceled?: (reason: string) => Promise | void; } +type QueueClearCallbacks = Pick< + QueuedMessageInternalOptions, + "onCanceled" | "onAcceptedPreStreamFailure" +>; + +/** + * One dispatchable unit in the queue. Plain follow-up messages batch into a single + * entry (joined text, accumulated file parts); "special" sends (compaction requests, + * agent-skill invocations, workspace-turn follow-ups, callback-carrying internal + * sends) always start their own entry so their metadata/callbacks stay attached to + * exactly one dispatch. + */ +interface QueueEntry { + messages: string[]; + /** First muxMetadata added to this entry (never overwritten by later batched adds). */ + muxMetadata?: unknown; + latestOptions?: SendMessageOptions; + fileParts: FilePart[]; + /** Dedupe keys registered by addOnce for adds that landed in this entry. */ + dedupeKeys: Set; + goalInterventionPolicy?: GoalInterventionPolicy; + dispatchMode: QueueDispatchMode; + /** + * Sealed entries never accept later batched messages: their callbacks/metadata + * correlate to exactly one turn (workspace-turn follow-ups, agent skills). + * Later messages queue as a new entry behind them instead. + */ + sealed: boolean; + /** User-originated entries are the only ones exposed to/restored into the composer. */ + userAuthored: boolean; + addCount: number; + syntheticCount: number; + agentInitiatedCount: number; + onCanceled?: (reason: string) => Promise | void; + onAccepted?: () => Promise | void; + onAcceptedPreStreamFailure?: (error: SendMessageError) => Promise | void; +} + +/** + * FIFO queue of messages sent during active streaming. + * + * The queue holds ordered entries that dispatch one at a time (see dequeueNext): + * - Plain messages batch into the newest open entry (texts joined, file parts + * accumulated, first muxMetadata preserved, latest options win). + * - Compaction requests, agent-skill invocations, workspace-turn follow-ups, and + * callback-carrying internal sends each start their own entry, so queueing one + * never blocks later sends — they simply dispatch after it (no enqueue errors). + * - Agent-skill / workspace-turn / callback entries are sealed: later messages + * start a new entry instead of adopting their metadata or callbacks. + * - User-authored and background/agent-initiated messages never share an entry, + * so renderer/restoration projections can omit background work precisely. + * - Compaction entries stay open: a follow-up typed behind a pending /compact + * batches under the compaction request (long-standing behavior). + * + * Display logic: + * - A single-message compaction or agent-skill entry shows its rawCommand + * (e.g. /compact, /{skill}); otherwise entries show their actual message texts. + */ export class MessageQueue { - private messages: string[] = []; - private firstMuxMetadata?: unknown; - private latestOptions?: SendMessageOptions; - private accumulatedFileParts: FilePart[] = []; - private dedupeKeys: Set = new Set(); - private goalInterventionPolicy?: GoalInterventionPolicy; - private queueDispatchMode: QueueDispatchMode = "tool-end"; - private queuedEntryCount = 0; - private queuedSyntheticCount = 0; - private queuedAgentInitiatedCount = 0; - private onCanceled?: (reason: string) => Promise | void; - private onAccepted?: () => Promise | void; - private onAcceptedPreStreamFailure?: (error: SendMessageError) => Promise | void; + private entries: QueueEntry[] = []; /** * Check if the queue currently contains a compaction request. */ hasCompactionRequest(): boolean { - return isCompactionMetadata(this.firstMuxMetadata); + return this.entries.some((entry) => isCompactionMetadata(entry.muxMetadata)); } hasWorkspaceTurn(handleId: string): boolean { return ( handleId.length > 0 && - isWorkspaceTurnMetadata(this.firstMuxMetadata) && - this.firstMuxMetadata.taskHandleId === handleId + this.entries.some( + (entry) => + isWorkspaceTurnMetadata(entry.muxMetadata) && entry.muxMetadata.taskHandleId === handleId + ) ); } + private getDispatchMode(entries: readonly QueueEntry[]): QueueDispatchMode { + if (entries.length === 0) { + return "tool-end"; + } + return entries.some((entry) => entry.dispatchMode === "tool-end") ? "tool-end" : "turn-end"; + } + + /** + * Effective dispatch mode across pending entries: any entry queued for tool-end + * makes the whole queue dispatch at tool-end (sticky, matching pre-entry behavior), + * otherwise turn-end. Empty queue reports the tool-end default. + */ getQueueDispatchMode(): QueueDispatchMode { - return this.queueDispatchMode; + return this.getDispatchMode(this.entries); } /** - * Add a message to the queue. - * Preserves muxMetadata from first message, updates other options. - * Accumulates file parts. - * - * @throws Error if trying to add a compaction request when queue already has messages + * Dispatch mode for user-visible entries only. Backend-initiated maintenance/wake + * messages should not change the queue badge shown beside the user's own follow-up. + */ + getVisibleQueueDispatchMode(): QueueDispatchMode { + return this.getDispatchMode(this.getVisibleEntries()); + } + + /** + * Add a message to the queue. Plain messages batch into the newest open entry; + * special sends start their own entry (see class docblock). Never throws. */ add( message: string, @@ -146,19 +192,23 @@ export class MessageQueue { /** * Whether a message queued via {@link addOnce} with this dedupe key is still pending. - * Keys reset when the queue is cleared (drain or user clear). + * Keys release when their entry dispatches or the queue is cleared. */ hasDedupeKey(dedupeKey: string): boolean { - return this.dedupeKeys.has(dedupeKey); + return this.entries.some((entry) => entry.dedupeKeys.has(dedupeKey)); } /** - * Whether the queue's only content is the single entry queued under this dedupe key. + * Whether the queue's only content is the single message queued under this dedupe key. * Used to supersede low-value scheduled entries (heartbeats): a later real message must * not batch behind them, because batching would adopt the first entry's muxMetadata. */ holdsOnlyDedupeKey(dedupeKey: string): boolean { - return this.queuedEntryCount === 1 && this.dedupeKeys.has(dedupeKey); + return ( + this.entries.length === 1 && + this.entries[0].addCount === 1 && + this.entries[0].dedupeKeys.has(dedupeKey) + ); } /** @@ -171,13 +221,13 @@ export class MessageQueue { dedupeKey?: string, internal?: QueuedMessageInternalOptions ): boolean { - if (dedupeKey !== undefined && this.dedupeKeys.has(dedupeKey)) { + if (dedupeKey !== undefined && this.hasDedupeKey(dedupeKey)) { return false; } const didAdd = this.addInternal(message, options, internal); if (didAdd && dedupeKey !== undefined) { - this.dedupeKeys.add(dedupeKey); + this.entries[this.entries.length - 1].dedupeKeys.add(dedupeKey); } return didAdd; } @@ -195,231 +245,293 @@ export class MessageQueue { return false; } - const incomingIsCompaction = isCompactionMetadata(options?.muxMetadata); - const incomingIsAgentSkill = isAgentSkillMetadata(options?.muxMetadata); - const incomingIsWorkspaceTurn = isWorkspaceTurnMetadata(options?.muxMetadata); const incomingHasAcceptedCallbacks = internal?.onAccepted != null || internal?.onAcceptedPreStreamFailure != null || internal?.onCanceled != null; - const queueHasMessages = !this.isEmpty(); + const incomingIsUserAuthored = + internal?.synthetic !== true && internal?.agentInitiated !== true; + // Sealed entries must own their turn end-to-end: workspace-turn metadata and + // internal callbacks correlate to exactly one dispatch, and agent-skill metadata + // must not leak onto batched follow-ups. + const incomingIsSealed = + isAgentSkillMetadata(options?.muxMetadata) || + isWorkspaceTurnMetadata(options?.muxMetadata) || + incomingHasAcceptedCallbacks; + // Compaction starts its own entry (its metadata must not adopt earlier batched + // texts), but stays open so a follow-up typed behind a pending /compact batches + // under the compaction request, preserving long-standing behavior. + const incomingStartsNewEntry = incomingIsSealed || isCompactionMetadata(options?.muxMetadata); const incomingMode = options?.queueDispatchMode ?? "tool-end"; - const nextQueueDispatchMode = !queueHasMessages - ? incomingMode - : incomingMode === "tool-end" - ? "tool-end" - : this.queueDispatchMode; - - const queueHasAgentSkill = isAgentSkillMetadata(this.firstMuxMetadata); - const queueHasWorkspaceTurn = isWorkspaceTurnMetadata(this.firstMuxMetadata); - const queueHasAcceptedCallbacks = - this.onAccepted != null || this.onAcceptedPreStreamFailure != null || this.onCanceled != null; - - // Avoid leaking agent-skill metadata to later queued messages. - // A skill invocation must be sent alone (or the user should restore/edit the queued message). - if (queueHasAgentSkill) { - throw new Error( - "Cannot queue additional messages: an agent skill invocation is already queued. " + - "Wait for the current stream to complete before sending another message." - ); - } - if (queueHasWorkspaceTurn) { - throw new Error( - "Cannot queue additional messages: a workspace turn follow-up is already queued. " + - "Wait for it to dispatch before sending another message." - ); - } - if (queueHasAcceptedCallbacks) { - throw new Error( - "Cannot queue additional messages: an internal workspace turn follow-up is already queued. " + - "Wait for it to dispatch before sending another message." - ); - } - - if (incomingHasAcceptedCallbacks && queueHasMessages) { - throw new Error( - "Cannot queue workspace turn follow-up: queue already has messages. " + - "Wait for the current stream to complete before sending another workspace turn." - ); - } - - // Cannot add compaction to a queue that already has messages - // (user should wait for those messages to send first) - if (incomingIsCompaction && queueHasMessages) { - throw new Error( - "Cannot queue compaction request: queue already has messages. " + - "Wait for current stream to complete before compacting." - ); - } - - // Cannot batch agent-skill metadata with other messages (it would apply to the whole batch). - if (incomingIsAgentSkill && queueHasMessages) { - throw new Error( - "Cannot queue agent skill invocation: queue already has messages. " + - "Wait for the current stream to complete before running a skill." - ); - } - if (incomingIsWorkspaceTurn && queueHasMessages) { - throw new Error( - "Cannot queue workspace turn follow-up: queue already has messages. " + - "Wait for the current stream to complete before sending another workspace turn." - ); + const tail = this.entries[this.entries.length - 1]; + let entry: QueueEntry; + if ( + tail !== undefined && + !tail.sealed && + !incomingStartsNewEntry && + tail.userAuthored === incomingIsUserAuthored + ) { + entry = tail; + // tool-end is sticky within an entry; turn-end never downgrades an entry + // that something already queued for tool-end dispatch. + if (incomingMode === "tool-end") { + entry.dispatchMode = "tool-end"; + } + } else { + entry = { + messages: [], + fileParts: [], + dedupeKeys: new Set(), + dispatchMode: incomingMode, + sealed: incomingIsSealed, + userAuthored: incomingIsUserAuthored, + addCount: 0, + syntheticCount: 0, + agentInitiatedCount: 0, + }; + this.entries.push(entry); } - const nextGoalInterventionPolicy = - this.goalInterventionPolicy === "pause" || options?.goalInterventionPolicy === "pause" + // Explicit pause is sticky within an entry (a batched steer must not unpause). + entry.goalInterventionPolicy = + entry.goalInterventionPolicy === "pause" || options?.goalInterventionPolicy === "pause" ? "pause" - : (options?.goalInterventionPolicy ?? this.goalInterventionPolicy); - - // Commit dispatch mode only after validation checks pass - this.queueDispatchMode = nextQueueDispatchMode; - - this.goalInterventionPolicy = nextGoalInterventionPolicy; + : (options?.goalInterventionPolicy ?? entry.goalInterventionPolicy); // Add text message if non-empty if (trimmedMessage.length > 0) { - this.messages.push(trimmedMessage); + entry.messages.push(trimmedMessage); } if (options) { const { fileParts, ...restOptions } = options; - // Preserve first muxMetadata (see class docblock for rationale) - if (options.muxMetadata !== undefined && this.firstMuxMetadata === undefined) { - this.firstMuxMetadata = options.muxMetadata; + // Preserve first muxMetadata per entry (see class docblock for rationale) + if (options.muxMetadata !== undefined && entry.muxMetadata === undefined) { + entry.muxMetadata = options.muxMetadata; } - this.latestOptions = restOptions; + entry.latestOptions = restOptions; if (fileParts && fileParts.length > 0) { - this.accumulatedFileParts.push(...fileParts); + entry.fileParts.push(...fileParts); } } if (internal?.onCanceled != null) { - this.onCanceled = internal.onCanceled; + entry.onCanceled = internal.onCanceled; } if (internal?.onAccepted != null) { - this.onAccepted = internal.onAccepted; + entry.onAccepted = internal.onAccepted; } if (internal?.onAcceptedPreStreamFailure != null) { - this.onAcceptedPreStreamFailure = internal.onAcceptedPreStreamFailure; + entry.onAcceptedPreStreamFailure = internal.onAcceptedPreStreamFailure; } - this.queuedEntryCount += 1; + entry.addCount += 1; if (internal?.synthetic === true) { - this.queuedSyntheticCount += 1; + entry.syntheticCount += 1; } if (internal?.agentInitiated === true) { - this.queuedAgentInitiatedCount += 1; + entry.agentInitiatedCount += 1; } return true; } /** - * Get all queued message texts (for editing/restoration). + * Entries containing user-originated input. Fully synthetic entries (background + * monitor wakes, scheduled maintenance, internal follow-ups) remain dispatchable + * but must not appear in or restore over the user's composer. */ + private getVisibleEntries(): QueueEntry[] { + return this.entries.filter((entry) => entry.userAuthored); + } + + private getMessagesForEntries(entries: readonly QueueEntry[]): string[] { + return entries.flatMap((entry) => entry.messages); + } + + private getDisplayTextForEntries(entries: readonly QueueEntry[]): string { + return entries + .map((entry) => { + if ( + entry.messages.length <= 1 && + (isCompactionMetadata(entry.muxMetadata) || isAgentSkillMetadata(entry.muxMetadata)) + ) { + return entry.muxMetadata.rawCommand; + } + return entry.messages.join("\n"); + }) + .filter((text) => text.length > 0) + .join("\n"); + } + + private getFilePartsForEntries(entries: readonly QueueEntry[]): FilePart[] { + return entries.flatMap((entry) => entry.fileParts); + } + + private getReviewsForEntries(entries: readonly QueueEntry[]): ReviewNoteData[] | undefined { + const reviews = entries.flatMap((entry) => + hasReviews(entry.muxMetadata) ? (entry.muxMetadata.reviews ?? []) : [] + ); + return reviews.length > 0 ? reviews : undefined; + } + + /** Get all queued message texts across entries (including synthetic entries). */ getMessages(): string[] { - return [...this.messages]; + 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. - * - Single compaction request shows rawCommand (/compact) - * - Single agent-skill invocation shows rawCommand (/{skill}) - * - Multiple messages show all actual message texts + * - A single-message compaction/agent-skill entry shows its rawCommand (/compact, /{skill}) + * - Otherwise entries show their actual message texts, joined with newlines */ getDisplayText(): string { - // Only show rawCommand for single compaction request - if (this.messages.length === 1 && isCompactionMetadata(this.firstMuxMetadata)) { - return this.firstMuxMetadata.rawCommand; - } + return this.getDisplayTextForEntries(this.entries); + } - // Only show rawCommand for a single agent-skill invocation. - // (Batching agent-skill with other messages is disallowed.) - if (this.messages.length <= 1 && isAgentSkillMetadata(this.firstMuxMetadata)) { - return this.firstMuxMetadata.rawCommand; - } + /** 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()); + } - return this.messages.join("\n"); + /** Whether a user-visible queued entry is a compaction request. */ + hasVisibleCompactionRequest(): boolean { + return this.getVisibleEntries().some((entry) => isCompactionMetadata(entry.muxMetadata)); } /** - * Get accumulated file parts for display. + * Cancellation callbacks for every pending entry, in queue order. + * Callers must notify each one when clearing the queue. */ - getFileParts(): FilePart[] { - return [...this.accumulatedFileParts]; + 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 } + : {}), + })); } /** - * Get reviews from metadata for display. + * Remove only the entry pinned to this workspace-turn handle, leaving unrelated + * queued messages intact (interrupting a queued turn must not drop user input). + * Returns the removed entry's cancellation callbacks, or null when no entry matches. */ - getReviews(): ReviewNoteData[] | undefined { - if (hasReviews(this.firstMuxMetadata) && this.firstMuxMetadata.reviews?.length) { - return this.firstMuxMetadata.reviews; + removeWorkspaceTurn(handleId: string): QueueClearCallbacks | null { + if (handleId.length === 0) { + return null; } - return undefined; - } - - getClearCallbacks(): Pick< - QueuedMessageInternalOptions, - "onCanceled" | "onAcceptedPreStreamFailure" - > { + const index = this.entries.findIndex( + (entry) => + isWorkspaceTurnMetadata(entry.muxMetadata) && entry.muxMetadata.taskHandleId === handleId + ); + if (index === -1) { + return null; + } + const [entry] = this.entries.splice(index, 1); return { - ...(this.onCanceled != null ? { onCanceled: this.onCanceled } : {}), - ...(this.onAcceptedPreStreamFailure != null - ? { onAcceptedPreStreamFailure: this.onAcceptedPreStreamFailure } + ...(entry.onCanceled != null ? { onCanceled: entry.onCanceled } : {}), + ...(entry.onAcceptedPreStreamFailure != null + ? { onAcceptedPreStreamFailure: entry.onAcceptedPreStreamFailure } : {}), }; } /** - * Get combined message and options for sending. + * Move the oldest user-authored entry to the head so an explicit user "Send now" + * action cannot be blocked by hidden synthetic/background work queued before it. + * Returns false when no user-authored entry is pending. */ - produceMessage(): { + prioritizeNextUserEntry(): boolean { + const index = this.entries.findIndex((entry) => entry.userAuthored); + if (index === -1) { + return false; + } + if (index > 0) { + const [entry] = this.entries.splice(index, 1); + this.entries.unshift(entry); + } + return true; + } + + /** + * Remove the first entry and return its combined message and options for sending. + * Later entries stay queued and dispatch on subsequent drains (FIFO). + * Caller must check {@link isEmpty} first. + */ + dequeueNext(): { message: string; options?: SendMessageOptions & { fileParts?: FilePart[] }; internal?: QueuedMessageInternalOptions; } { - const joinedMessages = this.messages.join("\n"); - // First metadata takes precedence (preserves compaction + agent-skill invocations) - const muxMetadata = - this.firstMuxMetadata !== undefined - ? this.firstMuxMetadata - : (this.latestOptions?.muxMetadata as unknown); - const options = this.latestOptions + const entry = this.entries.shift(); + if (entry === undefined) { + return { message: "" }; + } + + const joinedMessages = entry.messages.join("\n"); + const options = entry.latestOptions ? (() => { - const restOptions: SendMessageOptions = { ...this.latestOptions }; + const restOptions: SendMessageOptions = { ...entry.latestOptions }; delete restOptions.queueDispatchMode; - if (this.goalInterventionPolicy != null) { - restOptions.goalInterventionPolicy = this.goalInterventionPolicy; + if (entry.goalInterventionPolicy != null) { + restOptions.goalInterventionPolicy = entry.goalInterventionPolicy; } return { ...restOptions, - muxMetadata, - fileParts: this.accumulatedFileParts.length > 0 ? this.accumulatedFileParts : undefined, + // First metadata takes precedence (preserves compaction + agent-skill invocations) + muxMetadata: entry.muxMetadata, + fileParts: entry.fileParts.length > 0 ? entry.fileParts : undefined, }; })() : undefined; - const allQueuedEntriesAreSynthetic = - this.queuedEntryCount > 0 && this.queuedSyntheticCount === this.queuedEntryCount; - const allQueuedEntriesAreAgentInitiated = - this.queuedEntryCount > 0 && this.queuedAgentInitiatedCount === this.queuedEntryCount; + const allAddsAreSynthetic = entry.addCount > 0 && entry.syntheticCount === entry.addCount; + const allAddsAreAgentInitiated = + entry.addCount > 0 && entry.agentInitiatedCount === entry.addCount; const hasInternalOptions = - allQueuedEntriesAreSynthetic || - allQueuedEntriesAreAgentInitiated || - this.onAccepted != null || - this.onAcceptedPreStreamFailure != null || - this.onCanceled != null; + allAddsAreSynthetic || + allAddsAreAgentInitiated || + entry.onAccepted != null || + entry.onAcceptedPreStreamFailure != null || + entry.onCanceled != null; const internal = hasInternalOptions ? { - ...(allQueuedEntriesAreSynthetic ? { synthetic: true } : {}), - ...(allQueuedEntriesAreAgentInitiated ? { agentInitiated: true } : {}), - ...(this.onCanceled != null ? { onCanceled: this.onCanceled } : {}), - ...(this.onAccepted != null ? { onAccepted: this.onAccepted } : {}), - ...(this.onAcceptedPreStreamFailure != null - ? { onAcceptedPreStreamFailure: this.onAcceptedPreStreamFailure } + ...(allAddsAreSynthetic ? { synthetic: true } : {}), + ...(allAddsAreAgentInitiated ? { agentInitiated: true } : {}), + ...(entry.onCanceled != null ? { onCanceled: entry.onCanceled } : {}), + ...(entry.onAccepted != null ? { onAccepted: entry.onAccepted } : {}), + ...(entry.onAcceptedPreStreamFailure != null + ? { onAcceptedPreStreamFailure: entry.onAcceptedPreStreamFailure } : {}), } : undefined; @@ -428,28 +540,17 @@ export class MessageQueue { } /** - * Clear all queued messages, options, and images. + * Clear all queued entries. Callers that need to notify canceled entries must + * capture {@link getClearCallbacks} beforehand. */ clear(): void { - this.messages = []; - this.firstMuxMetadata = undefined; - this.latestOptions = undefined; - this.accumulatedFileParts = []; - this.dedupeKeys.clear(); - this.goalInterventionPolicy = undefined; - this.queueDispatchMode = "tool-end"; - this.onCanceled = undefined; - this.onAccepted = undefined; - this.onAcceptedPreStreamFailure = undefined; - this.queuedEntryCount = 0; - this.queuedSyntheticCount = 0; - this.queuedAgentInitiatedCount = 0; + this.entries = []; } /** - * Check if queue is empty (no messages AND no images). + * Check if queue is empty (no pending entries). */ isEmpty(): boolean { - return this.messages.length === 0 && this.accumulatedFileParts.length === 0; + return this.entries.length === 0; } } diff --git a/src/node/services/taskService.test.ts b/src/node/services/taskService.test.ts index 2cc33a9be9..72c48aaae3 100644 --- a/src/node/services/taskService.test.ts +++ b/src/node/services/taskService.test.ts @@ -366,6 +366,7 @@ function createWorkspaceServiceMocks( sendMessage: ReturnType; resumeStream: ReturnType; clearQueue: ReturnType; + removeQueuedWorkspaceTurn: ReturnType; hasQueuedWorkspaceTurn: ReturnType; hasQueuedMessages: ReturnType; isBusyForMessage: ReturnType; @@ -390,6 +391,7 @@ function createWorkspaceServiceMocks( sendMessage: ReturnType; resumeStream: ReturnType; clearQueue: ReturnType; + removeQueuedWorkspaceTurn: ReturnType; hasQueuedWorkspaceTurn: ReturnType; hasQueuedMessages: ReturnType; isBusyForMessage: ReturnType; @@ -415,6 +417,8 @@ function createWorkspaceServiceMocks( overrides?.resumeStream ?? mock((): Promise> => Promise.resolve(Ok({ started: true }))); const clearQueue = overrides?.clearQueue ?? mock((): Result => Ok(undefined)); + const removeQueuedWorkspaceTurn = + overrides?.removeQueuedWorkspaceTurn ?? mock((): Result => Ok(true)); const hasQueuedWorkspaceTurn = overrides?.hasQueuedWorkspaceTurn ?? mock(() => false); const hasQueuedMessages = overrides?.hasQueuedMessages ?? mock(() => false); const isBusyForMessage = overrides?.isBusyForMessage ?? mock(() => false); @@ -457,6 +461,7 @@ function createWorkspaceServiceMocks( sendMessage, resumeStream, clearQueue, + removeQueuedWorkspaceTurn, isBusyForMessage, hasQueuedWorkspaceTurn, hasQueuedMessages, @@ -479,6 +484,7 @@ function createWorkspaceServiceMocks( sendMessage, resumeStream, clearQueue, + removeQueuedWorkspaceTurn, hasQueuedWorkspaceTurn, hasQueuedMessages, isBusyForMessage, @@ -1490,9 +1496,11 @@ describe("TaskService", () => { const interrupted = await taskService.interruptWorkspaceTurn(parentId, "wst_secondhandle"); expect(interrupted.success).toBe(true); - expect(workspaceMocks.clearQueue).toHaveBeenCalledWith("childworkspace", { - cancelReason: "Workspace turn interrupted", - }); + expect(workspaceMocks.removeQueuedWorkspaceTurn).toHaveBeenCalledWith( + "childworkspace", + "wst_secondhandle", + { cancelReason: "Workspace turn interrupted" } + ); const sendInternal = secondSend[3] as { onAccepted: () => Promise }; let acceptedAfterInterruptError: unknown; try { diff --git a/src/node/services/taskService.ts b/src/node/services/taskService.ts index 10f681d81d..881ad1c79c 100644 --- a/src/node/services/taskService.ts +++ b/src/node/services/taskService.ts @@ -6008,11 +6008,13 @@ export class TaskService { } if (shouldClearQueuedPrompt && workspaceId != null) { - const clearQueueResult = this.workspaceService.clearQueue(workspaceId, { + // Targeted removal: the queue can hold unrelated user messages before/behind + // this turn's entry, so clearing the whole queue would drop real input. + const removeResult = this.workspaceService.removeQueuedWorkspaceTurn(workspaceId, handleId, { cancelReason: "Workspace turn interrupted", }); - if (!clearQueueResult.success) { - return Err(`Failed to clear queued workspace turn: ${clearQueueResult.error}`); + if (!removeResult.success) { + return Err(`Failed to clear queued workspace turn: ${removeResult.error}`); } } if (shouldStopStream && workspaceId != null) { diff --git a/src/node/services/workspaceService.test.ts b/src/node/services/workspaceService.test.ts index 1e9eb7212a..e27d798928 100644 --- a/src/node/services/workspaceService.test.ts +++ b/src/node/services/workspaceService.test.ts @@ -992,8 +992,12 @@ describe("WorkspaceService bash monitor wakes", () => { runtimeConfig: { type: "local" }, }); + const getForegroundToolCallIds = mock(() => ["tool-call-1"]); + const sendToBackground = mock(() => ({ success: true as const })); const backgroundProcessManager = Object.assign(new EventEmitter(), { cleanup: mock(() => Promise.resolve()), + getForegroundToolCallIds, + sendToBackground, }) as unknown as BackgroundProcessManager & EventEmitter; const workspaceService = createWorkspaceServiceForTest({ config, @@ -1025,9 +1029,11 @@ describe("WorkspaceService bash monitor wakes", () => { timestamp: Date.now(), }); - await waitForCondition(() => sendSpy.mock.calls.length === 1); + await waitForCondition(() => sendToBackground.mock.calls.length === 1); expect(sendSpy.mock.calls[0][2]).toMatchObject({ queueDispatchMode: "tool-end" }); expect(sendSpy.mock.calls[0][3]?.requireIdle).toBeUndefined(); + expect(getForegroundToolCallIds).toHaveBeenCalledWith(workspaceId); + expect(sendToBackground).toHaveBeenCalledWith("tool-call-1"); expect(waitForIdleSpy).not.toHaveBeenCalled(); } finally { await cleanup(); @@ -11455,12 +11461,12 @@ describe("WorkspaceService interruptStream", () => { terminateAllDescendantAgentTasks, } as unknown as TaskService); - const sendQueuedMessages = mock(() => undefined); + const sendNextUserQueuedMessage = mock(() => true); const restoreQueueToInput = mock(() => undefined); const interruptStream = mock(() => Promise.resolve(Ok(undefined))); const fakeSession = { interruptStream, - sendQueuedMessages, + sendNextUserQueuedMessage, restoreQueueToInput, }; const getOrCreateSessionSpy = spyOn(workspaceService, "getOrCreateSession").mockReturnValue( @@ -11476,7 +11482,7 @@ describe("WorkspaceService interruptStream", () => { expect(markParentWorkspaceInterrupted).toHaveBeenCalledWith(workspaceId); expect(terminateAllDescendantAgentTasks).toHaveBeenCalledWith(workspaceId); expect(resetAutoResumeCount).toHaveBeenCalledTimes(2); - expect(sendQueuedMessages).toHaveBeenCalledTimes(1); + expect(sendNextUserQueuedMessage).toHaveBeenCalledTimes(1); expect(restoreQueueToInput).not.toHaveBeenCalled(); } finally { getOrCreateSessionSpy.mockRestore(); diff --git a/src/node/services/workspaceService.ts b/src/node/services/workspaceService.ts index 18bb8d7447..d5ca944dd3 100644 --- a/src/node/services/workspaceService.ts +++ b/src/node/services/workspaceService.ts @@ -1942,6 +1942,28 @@ export class WorkspaceService extends EventEmitter { return `${ownerWorkspaceId}:${wakeId}`; } + /** + * A queued monitor wake has the same semantics as the user's "send after step": + * foreground bashes keep running, but detach into background tracking before the + * current stream is soft-stopped. Otherwise the stream abort can kill the process + * and discard work that the monitor wake was meant to observe. + */ + private backgroundForegroundBashesForMonitorWake(ownerWorkspaceId: string): void { + for (const toolCallId of this.backgroundProcessManager.getForegroundToolCallIds( + ownerWorkspaceId + )) { + const result = this.backgroundProcessManager.sendToBackground(toolCallId); + if (!result.success) { + // The bash may have completed between the snapshot and the request. + log.debug("Failed to background foreground bash for monitor wake", { + ownerWorkspaceId, + toolCallId, + error: result.error, + }); + } + } + } + private async drainBashMonitorWakes(ownerWorkspaceId: string): Promise { const pending = (await this.bashMonitorWakeStore.listPending(ownerWorkspaceId)).filter( (record) => @@ -2115,6 +2137,10 @@ export class WorkspaceService extends EventEmitter { return; } + if (ownerHasAiServiceStream) { + this.backgroundForegroundBashesForMonitorWake(ownerWorkspaceId); + } + if (accepted) { await markDeliveredOnce(); } @@ -8279,8 +8305,9 @@ export class WorkspaceService extends EventEmitter { // `sendQueuedMessages()` routes through AgentSession directly, so explicitly // clear hard-interrupt suppression first (it won't flow through sendMessage()). this.taskService?.resetAutoResumeCount(workspaceId); - // Send queued messages immediately instead of restoring to input - session.sendQueuedMessages(); + // The card represents only user-authored queue content. Prioritize that + // entry over hidden synthetic/background work before dispatching. + session.sendNextUserQueuedMessage(); } else { // Restore queued messages to input box for user-initiated interrupts session.restoreQueueToInput(); @@ -8512,6 +8539,28 @@ export class WorkspaceService extends EventEmitter { return this.sessions.get(workspaceId.trim())?.hasQueuedWorkspaceTurn(handleId) ?? false; } + /** + * Remove only the queued workspace-turn entry for this handle (targeted cancel); + * unrelated queued messages stay pending. Returns whether an entry was removed. + */ + removeQueuedWorkspaceTurn( + workspaceId: string, + handleId: string, + options: { cancelReason: string } + ): Result { + try { + const session = this.sessions.get(workspaceId.trim()); + if (session == null) { + return Ok(false); + } + return Ok(session.removeQueuedWorkspaceTurn(handleId, options.cancelReason)); + } catch (error) { + const errorMessage = getErrorMessage(error); + log.error("Unexpected error in removeQueuedWorkspaceTurn handler:", error); + return Err(`Failed to remove queued workspace turn: ${errorMessage}`); + } + } + hasQueuedMessages(workspaceId: string, dispatchMode?: "tool-end" | "turn-end"): boolean { return this.sessions.get(workspaceId.trim())?.hasQueuedMessages(dispatchMode) ?? false; }