fix(coding-agent): single-source the interactive queue state (#1842)
* fix(coding-agent): single-source the interactive queue state * fix(coding-agent): close queue-state races around rebind, moves, and failed edits - rebindCurrentSession: re-sync sessionActions after resubscribing; a session_action_update fired in the unsubscribed gap was lost forever. - browseQueueSelection: block browsing while a queue move RPC is pending, matching the pendingQueueEdit guard, so the move response cannot reselect the original item under newly browsed editor text. - moveQueueSelection: when the applied move's session_action_update has not landed yet (sessionActions reference unchanged), mirror the swap locally so the moved selection is not dropped against the pre-move snapshot; reconcile the selection in the error path. - applyQueueSelection: reconcile the selection after the pending-edit flag clears, so a rejected/failed edit no longer leaves Enter targeting a row that was removed while events were suppressed.
1- Fixed queued-message editing so duplicate prompts always target the selected queue entry.24 unmodified lines25262728292930313263 unmodified lines9697989910110010110210365 unmodified lines169170171172175173174175176177178179180183181182183184185186187188189204 unmodified linesExpand all24 unmodified lines isBashRunning: boolean; retryAttempt: number; sessionActions: { queuedCount: number; steering: readonly string[]; followUps: readonly string[] }; }; connectionQueue: { steering: string[]; followUp: string[] }; agentConnection: { abort: Mock; clearQueue: Mock; abortAndClearQueue: Mock;63 unmodified lines isBashRunning: options.bashRunning ?? false, retryAttempt: options.retryAttempt ?? 0, sessionActions: { queuedCount: 0, steering: [], followUps: [] }, }, connectionQueue: { steering: [], followUp: [] }, agentConnection: { abort: vi.fn().mockResolvedValue(undefined), clearQueue: vi.fn().mockResolvedValue({ steering: [], followUp: [] }), abortAndClearQueue: vi.fn().mockResolvedValue({ steering: [], followUp: [] }),65 unmodified lines });
it("preserves the queue and the draft when interrupting streaming", () => { const mode = createInteractiveFake({ editorText: "draft", streaming: true }); mode.connectionQueue = { steering: ["steer"], followUp: ["follow"] }; mode.connectionState.sessionActions = { queuedCount: 2, steering: ["steer"], followUps: ["follow"] };
Reflect.get(InteractiveMode.prototype, "handleCtrlC").call(mode);
expect(mode.agentConnection.abort).toHaveBeenCalledTimes(1); expect(mode.agentConnection.abortAndClearQueue).not.toHaveBeenCalled(); expect(mode.agentConnection.clearQueue).not.toHaveBeenCalled(); expect(mode.editor.getText()).toBe("draft"); expect(mode.connectionQueue).toEqual({ steering: ["steer"], followUp: ["follow"] }); expect(mode.connectionState.sessionActions).toEqual({ queuedCount: 2, steering: ["steer"], followUps: ["follow"], }); });
it("exits on the second Ctrl+C while the hint is visible", () => { const mode = createInteractiveFake({ streaming: true });204 unmodified linesExpand all31 unmodified lines323334353636373839404142431 unmodified line454647484649505152153 unmodified linesExpand all31 unmodified lines statusContainer, featureHintContainer, loadingAnimation: loader, workingVisible: true, connectionState: { isStreaming: true }, connectionState: { isStreaming: true, sessionActions: { queuedCount: 0, steering: [], followUps: [] }, }, workingTimer: undefined, workingStartedAt: 0, featureHintDeck, currentFeatureHint: undefined,1 unmodified line featureHintTimer: undefined, featureHintAnimationTimer: undefined, featureHintComponent: undefined, featureHintRunPending: false, connectionQueue: { steering: [], followUp: [] }, compactionQueuedMessages: [], options: { returnToAgentsView: true }, ui: { requestRender }, };153 unmodified linesExpand all42 unmodified lines434445464747484950515253575 unmodified linesExpand all62963063163263163363463563663763863763964064164264318 unmodified lines6626636646656646666676686693 unmodified lines673674675676676677678679680681136 unmodified linesExpand all8188198208218218228238248258262 unmodified lines42 unmodified lines clearShortcutGuide: Mock;};
type PromptStashLiveMarkerHarness = PromptStashHarness & { connectionQueue: { steering: string[]; followUp: string[] }; connectionState: { sessionActions: { queuedCount: number; steering: readonly string[]; followUps: readonly string[] }; };};
type SharedPromptStashHarness = PromptStashHarness & { promptStashStore: ClientPromptStashStore;575 unmodified linesExpand all expect(mode.promptStash).toBeUndefined(); expect(mode.editor.getText()).toBe("half-written draft"); });
it("drops queued image references from old sessions while keeping stashed images", () => { it("drops old-session images while keeping stashed images", () => { const base = createPromptStashHarness({ stash: "keep [image #1]" }); const mode: ResetHarness = { ...base, defaultEditor: base.editor, queueSelection: new QueueSelection(), connectionQueue: { steering: ["old [image #2]"], followUp: [] }, connectionState: { sessionActions: { queuedCount: 0, steering: [], followUps: [] } }, chatContainer: { clear: vi.fn() }, shortcutGuideContainer: { clear: vi.fn() }, pendingMessagesContainer: { clear: vi.fn() }, queuedMessagesContainer: { clear: vi.fn() },18 unmodified lines Object.setPrototypeOf(mode, InteractiveMode.prototype);
interactiveModeMethods.resetCurrentSessionRenderState.call(mode);
expect(mode.connectionQueue).toEqual({ steering: [], followUp: [] }); expect(mode.promptStash?.text).toBe("keep [image #1]"); expect(mode.pastedImages.has(1)).toBe(true); expect(mode.pastedImages.has(2)).toBe(false); });3 unmodified lines const mode: ResetHarness = { ...base, defaultEditor: base.editor, queueSelection: new QueueSelection(), connectionQueue: { steering: [], followUp: [] }, connectionState: { sessionActions: { queuedCount: 0, steering: [], followUps: [] } }, chatContainer: { clear: vi.fn() }, shortcutGuideContainer: { clear: vi.fn() }, pendingMessagesContainer: { clear: vi.fn() }, queuedMessagesContainer: { clear: vi.fn() },136 unmodified linesExpand all
it("keeps image markers in a stashed prompt live", () => { const mode: PromptStashLiveMarkerHarness = { ...createPromptStashHarness({ stash: "look at [image #7]" }), connectionQueue: { steering: [], followUp: [] }, connectionState: { sessionActions: { queuedCount: 0, steering: [], followUps: [] } }, }; Object.setPrototypeOf(mode, InteractiveMode.prototype);
expect(interactiveModeMethods.liveImageMarkerIds.call(mode)).toEqual(new Set([7]));2 unmodified lines1262 unmodified linesExpand all1263126412651266126712671268126912701271161 unmodified linesExpand all143314341435143614371437143814391440144114421443144114421443144414471448144914501451145214531445144614471448144914501451145214531457145414551456145714611462146314641465146614671468146914701471147214731474147514761477147814791480148114821458145914601461146214631464146514661467146814691470147114721473147414751476147714781479148014811482148314841484148514861487148814891490149114921493149414951496149714981499150015011502150315041485148614871488148914901491149214931494184 unmodified linesExpand all16791680168116821683168416851686168716881689169016911692169316941695169616971698169917001701170217031704170517061707170817091710171117121713171417151716171717181719172017211722172317241725172617271728172917301731173217331734173517361737173817391740174117421743174425 unmodified lines1770177117721773177417751776177717781 unmodified line178017811782178317391784178517861787178838 unmodified lines1827182818291830183118321833183418351 unmodified line183718381839184017951841184218431844184529 unmodified lines18751876187718781833187918801881188218831 unmodified line18851886188718881843188918901891189218931270 unmodified linesExpand all316431653166316731223168316931703171317222 unmodified lines31953196319731983199320032013202320332043158320532063207320832092330 unmodified linesExpand all1262 unmodified linesExpand all });
const bashComponent = () => ({ render: () => [], invalidate: () => {} });
test("keeps pending bash components visible across queue refreshes", () => { test("keeps pending bash components visible across queue display updates", () => { const pendingMessagesContainer = new Container(); const component = bashComponent(); const fakeThis = { pendingMessagesContainer,161 unmodified linesExpand all });});
describe("InteractiveMode connection events", () => { test("rendering a switched session tolerates a transient queue refresh failure", async () => { test("rendering a switched session updates the pending display from its snapshot", async () => { const harness = { resetCurrentSessionRenderState: vi.fn(), renderInitialMessages: vi.fn(async () => {}), refreshConnectionQueue: vi.fn(async () => { throw new Error("queue unavailable"); }), updatePendingMessagesDisplay: vi.fn(), syncWorkingLoader: vi.fn(), };
await expect( ( InteractiveMode.prototype as unknown as { renderCurrentSessionState(this: typeof harness): Promise<void>; } ).renderCurrentSessionState.call(harness), ).resolves.toBeUndefined(); await ( InteractiveMode.prototype as unknown as { renderCurrentSessionState(this: typeof harness): Promise<void>; } ).renderCurrentSessionState.call(harness); expect(harness.updatePendingMessagesDisplay).toHaveBeenCalledOnce(); expect(harness.syncWorkingLoader).toHaveBeenCalledOnce(); });
test("degrades heartbeat refresh failures without hiding queue refresh failures during rebind", async () => { test("degrades heartbeat refresh failures while updating the pending display during rebind", async () => { const rebindCurrentSession = ( InteractiveMode.prototype as unknown as { rebindCurrentSession(this: InteractiveMode): Promise<void> } ).rebindCurrentSession; const createHarness = ( refreshConnectionQueue: () => Promise<void>, refreshHeartbeatCatalog: () => Promise<void>, ) => ({ unsubscribe: undefined, localSessionHost: undefined, toolDefinitionCache: { clear: vi.fn() }, applyRuntimeSettings: vi.fn(), bindLocalSessionExtensions: true, bindCurrentSessionExtensions: vi.fn(async () => {}), subscribeToAgent: vi.fn(), refreshConnectionQueue, refreshHeartbeatCatalog, updateAvailableProviderCount: vi.fn(async () => {}), updateEditorBorderColor: vi.fn(), updateTerminalTitle: vi.fn(), setGoalAnnouncementBaseline: vi.fn(), syncGoalTray: vi.fn(), syncWorkingLoader: vi.fn(), getGoalState: () => emptyGoalState(), }) as unknown as InteractiveMode; const updatePendingMessagesDisplay = vi.fn(); const subscribeToAgent = vi.fn(); const getState = vi.fn(async () => createConnectionState()); const harness = { unsubscribe: undefined, localSessionHost: undefined, toolDefinitionCache: { clear: vi.fn() }, applyRuntimeSettings: vi.fn(), bindLocalSessionExtensions: true, bindCurrentSessionExtensions: vi.fn(async () => {}), subscribeToAgent, agentConnection: { getState }, patchConnectionState: vi.fn(), refreshQueueSelectionFromState: vi.fn(), updatePendingMessagesDisplay, refreshHeartbeatCatalog: vi.fn(async () => { throw new Error("heartbeat unavailable"); }), updateAvailableProviderCount: vi.fn(async () => {}), updateEditorBorderColor: vi.fn(), updateTerminalTitle: vi.fn(), setGoalAnnouncementBaseline: vi.fn(), syncGoalTray: vi.fn(), syncWorkingLoader: vi.fn(), getGoalState: () => emptyGoalState(), } as unknown as InteractiveMode;
await expect( rebindCurrentSession.call( createHarness( vi.fn(async () => {}), vi.fn(async () => { throw new Error("heartbeat unavailable"); }), ), ), ).resolves.toBeUndefined();
await expect( rebindCurrentSession.call( createHarness( vi.fn(async () => { throw new Error("queue unavailable"); }), vi.fn(async () => {}), ), ), ).rejects.toThrow("queue unavailable"); await expect(rebindCurrentSession.call(harness)).resolves.toBeUndefined(); expect(updatePendingMessagesDisplay).toHaveBeenCalledOnce(); // The queue re-sync must run after subscribing, or updates in the gap are lost. expect(getState.mock.invocationCallOrder[0]).toBeGreaterThan( subscribeToAgent.mock.invocationCallOrder[0] as number, ); });
test("restores in-flight assistant state on every session render", async () => { const streamingMessage = {184 unmodified linesExpand all expect(fakeThis.renderResyncedSession).not.toHaveBeenCalled(); expect(fakeThis.renderInitialMessages).toHaveBeenCalledOnce(); });
test("exits stale queue browsing when a resync replaces the queue snapshot", async () => { const queueSelection = new QueueSelection(); let editorText = "draft"; queueSelection.move({ steering: [], followUp: ["queued"] }, editorText, -1); editorText = "queued"; const snapshot: AgentConnectionSnapshot = { state: createConnectionState({ sessionActions: { queuedCount: 0, steering: [], followUps: [] }, }), messages: [], }; const fakeThis = { connectionState: createConnectionState({ sessionActions: { queuedCount: 1, steering: [], followUps: ["queued"] }, }), queueSelection, pendingQueueEdit: undefined, pendingQueueMove: false, isApplyingQueueSelectionText: false, editor: { getText: () => editorText, setText: (text: string) => { editorText = text; }, }, isBashRunning: () => false, applyConnectionStateSnapshot: vi.fn(), restoreTurnStartFromMessages: vi.fn(), replaceSubagentSummary: vi.fn(), getSessionContextFromConnectionSnapshot: vi.fn(() => ({ messages: [], thinkingLevel: "medium", model: null, })), renderSessionContext: vi.fn(async () => {}), restoreStreamingMessageFromSnapshot: vi.fn(), updatePendingMessagesDisplay: vi.fn(), updateTerminalTitle: vi.fn(), setGoalAnnouncementBaseline: vi.fn(), syncGoalTray: vi.fn(), syncWorkingLoader: vi.fn(), getGoalState: () => emptyGoalState(), }; fakeThis.applyConnectionStateSnapshot.mockImplementation((state: AgentConnectionState) => { fakeThis.connectionState = state; }); Object.setPrototypeOf(fakeThis, InteractiveMode.prototype);
await ( InteractiveMode.prototype as unknown as { renderResyncedSession(this: unknown, value: AgentConnectionSnapshot): Promise<void>; } ).renderResyncedSession.call(fakeThis, snapshot);
expect(queueSelection.isBrowsing).toBe(false); expect(editorText).toBe("draft"); });
test("preserves client-local work while rendering a resynchronized snapshot", async () => { const sideQuestion = { id: "side-1", status: "running" }; const extensionRequests = new Map([["request-1", { cancelLocal: vi.fn() }]]); const activeBashComponent = {};25 unmodified lines isAgentStreaming: () => true, streamingComponent: {}, streamingMessage: {}, applyConnectionStateSnapshot: vi.fn(), refreshQueueSelectionFromState: vi.fn(), updateWorkingLoaderMessage: vi.fn(), replaceSubagentSummary: vi.fn(), getSessionContextFromConnectionSnapshot: vi.fn(() => ({ messages: snapshot.messages,1 unmodified line model: null, })), renderSessionContext: vi.fn(async () => {}), restoreStreamingMessageFromSnapshot, refreshConnectionQueue: vi.fn(async () => {}), updatePendingMessagesDisplay: vi.fn(), flushPendingBashComponents: vi.fn(), updateTerminalTitle: vi.fn(), setGoalAnnouncementBaseline: vi.fn(), syncGoalTray: vi.fn(),38 unmodified lines streamingMessage: {}, isAgentCompacting: () => true, isBashRunning: () => true, applyConnectionStateSnapshot: vi.fn(), refreshQueueSelectionFromState: vi.fn(), restoreTurnStartFromMessages: vi.fn(), replaceSubagentSummary: vi.fn(), getSessionContextFromConnectionSnapshot: vi.fn(() => ({ messages: [],1 unmodified line model: null, })), renderSessionContext: vi.fn(async () => {}), restoreStreamingMessageFromSnapshot: vi.fn(), refreshConnectionQueue: vi.fn(async () => {}), updatePendingMessagesDisplay: vi.fn(), flushPendingBashComponents, updateTerminalTitle: vi.fn(), setGoalAnnouncementBaseline: vi.fn(), syncGoalTray: vi.fn(),29 unmodified lines calls.push("replacement"); }), resetCurrentSessionRenderState: () => calls.push("reset"), renderInitialMessages: async () => calls.push("messages"), refreshConnectionQueue: async () => calls.push("queue"), updatePendingMessagesDisplay: () => calls.push("display"), syncWorkingLoader: () => calls.push("loader"), };
await (1 unmodified line renderCurrentSessionState(this: typeof fakeThis): Promise<void>; } ).renderCurrentSessionState.call(fakeThis);
expect(calls).toEqual(["replacement", "reset", "messages", "queue", "loader"]); expect(calls).toEqual(["replacement", "reset", "messages", "display", "loader"]); });
test("drops a queued source event after the session is replaced", async () => { type Event =1270 unmodified linesExpand all}
describe("InteractiveMode session switch command catalog", () => { test.each(["switchSession", "newSession", "fork"] as const)( "refreshes an event-emitting in-process %s replacement exactly once before replay", "refreshes the command catalog for an event-emitting in-process %s replacement exactly once before replay", async (operation) => { const sourceSession = createFakeConnectionSession("source-command"); const targetSession = createFakeConnectionSession("target-command"); const runtime = new EventEmittingReplacementRuntime(sourceSession, targetSession);22 unmodified lines resetSideQuestion: vi.fn(), resetExtensionUI: vi.fn(), applyConnectionStateSnapshot: vi.fn(), resetCurrentSessionRenderState: vi.fn(() => calls.push("reset")), queueSelection: { selected: undefined }, setupAutocompleteProvider: vi.fn(() => calls.push("catalog")), renderInitialMessages: vi.fn(async () => { calls.push("render"); expect(fakeThis.connectionCommands.map((command) => command.name)).toEqual(["target-command"]); }), refreshConnectionQueue: vi.fn(async () => {}), updatePendingMessagesDisplay: vi.fn(), syncWorkingLoader: vi.fn(), ui: { requestRender: vi.fn() }, handleEvent: vi.fn(), handleConnectionExtensionUiRequest: vi.fn(),2330 unmodified linesExpand all