From 0f7a3da68bc45ebb817135fc6fc2f1f13f75d3dd Mon Sep 17 00:00:00 2001 From: SSU-WEI HUANG Date: Thu, 20 Aug 2026 08:53:41 +0800 Subject: [PATCH] feat(cursor): mid-turn Steer via concurrent ACP session/prompt (#888) (#1609) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * feat(shared): steer capability gates and live steered signal schemas - STEERING_SUPPORTED_FLAVORS / isSteeringSupportedForSession gate which agents can deliver queued messages into the active turn (pi, codex, cursor ACP; legacy stream-json cursor excluded) - AgentState.steeringActive, DecryptedMessage.steered and messages-consumed live signal (never persisted by the hub) * feat(cli): queue reservations and steered messages-consumed option - MessageQueue2 gains takeByLocalId/restoreReservation/ beginReservationDispatch/commitReservation so an async steer can reserve a queued row without racing the main loop's turn/start drain - emitMessagesConsumed accepts steered: true to mark mid-turn delivery * feat(codex): mid-turn steer via app-server turn/steer (#888) - CodexAppServerClient.steerTurn + TurnSteerParams/Response types - CodexRemoteLauncher registers the steer-queued-message RPC handler: reserves the queued row, validates it against the active turn (no control commands, matching mode hash), injects via turn/steer with an epoch guard that invalidates in-flight steers on abort/cleanup - steeringActive agent state tracks the active-turn window - hub syncEngine gate opens to codex; messages-consumed relays steered * feat(web): Steered badge and steer gating for codex sessions - HappyUserMessage shows a ↳ Steered badge fed by the live messages-consumed steered signal, preserved across server echoes and refetches (mergeMessages carries the optimistic marker) - SessionChat gates canSteer via isSteeringSupportedForSession instead of the pi-only check - clearStaleQueuedStatus normalizes a queued status on an invoked message - fix(web): drop duplicate showSessionSummaryInChat in markdown test (upstream typecheck breakage) * feat(acp): split request dispatch from completion and add soft steer - AcpStdioTransport.sendRequestWithDispatch separates stdin-accepted dispatch from the JSON-RPC response, keeping sendRequest behavior unchanged - AcpSdkBackend tracks concurrent session/prompt requests with an activePromptRequests counter (main prompt + soft steers); response completion stays pending until every concurrent prompt settles - beginSoftSteerPrompt kicks off a concurrent session/prompt (Cursor GUI Send semantics — no cancel, no handler swap) returning {dispatched, completed}; softSteerPrompt awaits the full response for direct callers * feat(cursor): mid-turn soft steer via concurrent session/prompt (#888) - CursorAcpRemoteLauncher registers the steer-queued-message RPC handler: reserves the queued row, rejects control commands and mode mismatches, then soft-injects via beginSoftSteerPrompt without canceling the in-flight turn - Acks the hub once stdin accepts the inject (not on turn completion) to stay inside the 30s RPC window; the launcher stays busy until the concurrent prompt settles so handlers are not swapped mid-inject - steeringActive agent state mirrors the active-turn window; abort and cleanup reset it and invalidate pending steers - Legacy stream-json Cursor sessions register a steer handler that reports unsupported * fix(codex,shared): address bot findings on steer gate and ambiguous turn/steer - STEERING_SUPPORTED_FLAVORS / isSteeringSupportedForSession advertise codex and pi only; cursor joins when its soft-steer handler lands (#1609) - turn/steer now splits dispatch (stdin accepted) from completion (turn finished): the hub RPC acks once dispatch succeeds — never on the concurrent turn's completion, which can exceed the 30s RPC window - queue row commits only after the turn settles; a rejected/aborted steer restores the row so the message still delivers via turn/start, and a dispatched steer is never restored (no duplicate delivery) - steer carries clientUserMessageId (echoed as userMessage.clientId) so ambiguous transport failures can reconcile the thread later - client tests cover dispatch/complete split and stdin-write failure * feat(shared): advertise cursor in the steer gate now that its handler lands Cursor ACP sessions pass the web and hub steer gates; legacy stream-json cursor sessions stay excluded. * fix(codex): reconcile dispatched steers before restoring; align error copy - A dispatched turn/steer whose completion fails (disconnect / protocol error) is now reconciled via thread/read by clientUserMessageId before the queued row is restored — the instruction is only re-delivered by turn/start when the thread never received it - Reconcile targets the pinned steer thread, not whichever turn is current when completion fails - syncEngine unsupported-flavor error now matches the capability gate (Pi and Codex only until the cursor handler lands) - launcher tests cover steer success (ack on dispatch), reconcile-accepted and reconcile-rejected outcomes * fix(codex): consume the row at dispatch; drop background reconcile - The hub RPC acks and the queue row is consumed as soon as stdin accepts turn/steer; completion is background-only logging. A dispatched steer is never restored, so the same localId cannot be re-delivered via turn/start after the caller was told the steer succeeded - Dispatch failure (stdin write error) still restores the row and reports failure - steer.completed rejection is always handled (no unhandled rejection on the dispatch-failure path) - tests updated: completion failure after dispatch keeps the row consumed; dispatch failure restores it * fix(cursor): consume the row at dispatch; keep waiters for prompt gating - The hub RPC acks and the queue row is consumed as soon as stdin accepts the concurrent session/prompt; completion is background-only. A dispatched steer is never restored (no duplicate via the next prompt) - softSteerWaiters are registered before awaiting dispatch so the main loop's finally cannot start the next prompt mid-inject; they still gate prompt handover on completion - tests updated: post-dispatch ACP rejection keeps the row consumed * fix(cursor,hub): never hang teardown on unresolved soft steer; align diagnostics - Prompt-finally waits for soft-steer completion only when not exiting; the outer finally no longer waits at all — cleanup() disconnects the ACP transport, which rejects pending requests and settles the waiters - syncEngine gate diagnostics and JSDoc name all supported flavors (Pi, Codex, Cursor ACP) - regression test: Switch with an unresolved soft-steer completion still reaches teardown * fix(codex): distinguish definite rejection from indeterminate completion - Transport-level failures (timeout, abort, disconnect, spawn, protocol) carry an indeterminate marker; explicit JSON-RPC error responses do not - After a dispatched steer, turn completion resolves → commit + consumed; a definite app-server rejection restores the row (instruction was never accepted, so turn/start cannot duplicate it); an indeterminate outcome leaves the row reserved so it can never be delivered twice - Completion handling registers before awaiting dispatch so the dispatch-failure path cannot leak an unhandled rejection - client/launcher tests cover explicit rejection (restore), indeterminate outcome (row stays reserved) and dispatch failure * fix(codex): reconcile indeterminate steers instead of a permanent reservation - After an indeterminate completion (disconnect/protocol), reconcile the thread by clientUserMessageId immediately: accepted → commit + consumed, provably rejected → restore, still unreadable → keep the reservation and retry from the main-loop top on later passes (post-reconnect) - A row never sits in dispatching forever: the hub cannot stamp it invoked while the instruction may never have been accepted - tests: indeterminate keeps reserved while thread unreadable; accepted reconciliation consumes; rejected path restores * fix(cursor): abort drops soft-steer waiters so the next prompt never blocks - Ordinary Abort (shouldExit false) now clears softSteerWaiters: the prompt finally cannot wait forever on a soft steer whose completion is unbounded; the ACP cancel rejects in-flight requests, and cleanup() settles leftovers on session end - regression test: unresolved soft-steer completion after Abort no longer blocks the next prompt * fix(codex): accept all thread item shapes; retry reconcile; ack through abort - Reconcile matcher accepts userMessage/user_message with clientId/ client_id, matching the shapes the thread parser supports — an accepted steer can no longer be misclassified as rejected - A pending reconciliation schedules a wakeLoop retry, so a temporary app-server outage cannot strand the reservation behind waitForTurnOrRecovery - The success-path ACK no longer checks the steer epoch: the hub already reported steered on dispatch, so commit + messages-consumed must reach it even when an abort resets the queue in between * fix(cursor,acp): abort force-settles soft-steer bookkeeping - AcpSdkBackend.abortSoftSteers() drops the concurrent-prompt counter and notifies response-complete so the next turn's waitForResponseComplete() cannot block on a soft steer that will never settle after abort - handleAbort calls it before clearing the waiters; the main prompt's own finishPromptRequest stays guarded by Math.max(0, ...) - unit tests cover counter release and no-op when idle * fix(codex): reinit reconnected app-server; keep reconcile retries alive - thread/read after a disconnect auto-connects a fresh app-server, which must be initialized before any request — reconcile now ensures connect + initialize (isConnected getter added to the client) - every still-unknown loop-top reconciliation schedules the next retry, so recovery without external traffic is eventually observed - launcher mock gains isConnected * test(acp): match finishPromptRequest epoch signature in whitebox test * fix(codex): timer-driven reconciliation; init tracking; abort-safe ACK - Reconciliation runs on a self-rescheduling 1s timer independent of the main loop (wakes it too), so idle loops and waitForTurnOrRecovery still observe app-server recovery; abort clears nothing implicitly — the ACK path commits and consumes even when the reservation was cancelled - Absence of a durable client id is ambiguous: unmatched reads stay 'unknown' and keep retrying instead of restoring the row - CodexAppServerClient tracks initialized state (reset on disconnect/exit) so ensureAppServerInitialized re-initializes a fresh process before thread/read; initialize failures leave the flag false for the next retry - tests: accepted reconciliation via scheduled timer, indeterminate keeps reserved, explicit rejection restores * fix(codex): bind reconciliation to the launcher lifecycle - runSteerReconciliation clears any armed retry timer on entry and never installs a second one, so loop-top and timer-driven passes cannot multiply - shuttingDown is set when the main loop ends: timers are cleared and the pending map is dropped, so an unresolved steer can never respawn an app-server after cleanup (remote-to-local switch included) * fix(cursor): abort releases an in-progress soft-steer wait - The prompt-finally wait races Promise.allSettled against the abort signal: an Abort that clears the waiters now also releases a wait that already started, so the launcher always reaches the next queued prompt * fix(codex): report steered only after app-server acceptance - The handler now awaits steer.completed (the inject-acceptance response): an explicit JSON-RPC rejection surfaces as failed and restores the row for the normal turn/start path instead of a false steered - Transport failure after dispatch reports 'Steer outcome is being reconciled' and keeps the row reserved while the timer-driven thread reconciliation runs - dispatch-failure path also swallows the paired completion rejection * fix(cursor,acp): commit on ACP acceptance; distinguish transport failures - AcpStdioTransport marks transport-level failures (timeout, closed, stdin write) as indeterminate; explicit JSON-RPC error responses are not - The steer handler commits + consumes on completion (ACP acceptance) and restores the row on an explicit rejection; an indeterminate transport failure keeps the row reserved so a delivered instruction is never re-sent, and the ACK reaches the hub even when abort reset the queue - launcher/transport tests updated for the three outcomes * fix(steer): tri-state cancel, clear-safe reservations, bounded acceptance wait - MessageQueue2.cancelByLocalId returns 'in-flight' for a dispatching steer reservation: the hub neither deletes the row nor stamps invoked_at (new CancelMessageResponse 'busy' status; web restores the optimistic row); pushIsolateAndClear and reset/close share cancelReservations so /clear-style commands cannot have a rejected steer resurrect a discarded prompt - turn/steer acceptance wait bounded at 25s (< hub 30s RPC timeout): a lost response is indeterminate and funnels into thread reconciliation instead of stranding the reservation - tests updated for the tri-state cancel contract * test(cursor): match tri-state cancel contract for dispatching steers * fix(cursor): drop duplicate promptInFlight declaration after upstream merge * fix(codex,web): busy-aware edit flow; bound reconciliation reads - QueuedMessagesBar edit flow treats a 'busy' cancel as unsuccessful: it never prefills the composer when the row is inside an async steer, so a second client cannot send a duplicate - reconcileSteerByClientId bounds thread/read with a 5s timeout so a connected-but-silent app-server cannot hold the reservation in-flight indefinitely * fix(steer): inFlight-dominated cancel acks; bounded reconciliation - hub cancel-queued-message acks check inFlight before removed: a stale duplicate socket reporting removed can no longer delete the durable row while another socket is dispatching the steer - reconciliation entries expire after 60s and mark delivered: after the rejection window, a dispatched steer that the app-server never proved (client ids dropped on restart) is committed instead of polling thread/read forever - pre-dispatch failures (abort before write included) never enter reconciliation — they restore the row and report failure * fix(cursor,acp,web): indeterminate close marks, steer gating precision - AcpStdioTransport.rejectAllPending marks close/protocol failures indeterminate, so an accepted-but-close-interrupted soft steer restores nothing (no duplicate delivery) - the abort race in the soft-steer wait removes its listener in finally (no accumulation across repeated waits) - SessionChat gates the Steer button on agentState.steeringActive for codex/cursor instead of the queued-grace thinking flag, so Steer is not exposed before the launcher can accept it - codex pre-dispatch abort never enters reconciliation (merged from #1606) * fix(steer): persist indeterminate outcomes without replay * fix(cursor): hold ambiguous steers for explicit resolution * fix(steer): make ambiguous delivery restart-safe * fix(cursor): make ambiguous delivery restart-safe * fix(steer): recover crash-held rows and preserve retry dedup * fix(steer): ack retries and bound stdin dispatch * fix(cursor): reject steers when prompt generation changes * fix(steer): reconcile indeterminate dispatches and serialize retries * fix(cursor): preserve soft-steer reservations across abort * fix(codex): classify stdin callback failures as indeterminate * fix(steer): recheck indeterminate cancels after ACK * fix(steer): close retry and abort races * fix(cursor): hold ambiguous dispatch failures * fix(steer): serialize live retries and abort admission * fix(steer): distinguish live dispatching from unknown * fix(cursor): bound ACP dispatch acknowledgements * fix(steer): keep ACK failures held and reconcile busy cancel * fix(cursor): preserve state when dispatch ACK is uncertain * fix(steer): distinguish held cancel from removal * fix(store): combine schema v24 migrations * fix(cursor): distinguish held cancel from removal * fix(store): reserve schema v25 for steer delivery state * fix(cursor): suppress late ACP updates after abort * fix(steer): keep held cancel state and notify requeue * fix(cursor): isolate late updates after abort * fix(steer): release explicitly cancelled unknown reservations * test(cursor): cover explicit held cancellation * fix(codex): reject cancelled reservations before native steer * fix(cursor): reject cancelled reservations before ACP steer * fix(codex): make reservation restore atomic with state * fix(cursor): make reservation restore atomic with state * fix(codex): terminate abandoned transport writes * fix(cursor): hard-stop abandoned writes and update native queue state * fix(steer): own abandoned app-server lifecycle and consume races * fix(cursor): isolate aborts and add native retry resolution * fix(codex): confirm dispatch and recover abandoned turns * test(codex): mock abandoned transport callback * fix(native): reconcile retry responses * fix(codex): clear visible turn state on transport loss * fix(steer): claim retries and cover native delivery state * fix(native): resync busy cancel outcomes * fix(native): preserve indeterminate state on Android hydration * fix(steer): make retry claims single-winner * fix(cursor): hold restore failures for explicit resolution * fix(steer): serialize concurrent retry claims * fix(socket): tolerate missing steer-state ACK callbacks * fix(native): serialize retry operations * docs(web): document unknown steer delivery and retry controls * fix(steer): handle retry failures and abort-before-connect * fix(cursor): drain foreground prompt after soft-steer abort * fix(steer): reinitialize after transport loss and finish iOS retry errors * fix(steer): preserve indeterminate rows across reconnect gaps * test(web): mock indeterminate queued recovery state * fix(steer): recover consumed ACK tombstones * fix(steer): expose consumed cancel tombstones * fix(cursor): drain soft steers before handler replacement * fix(cursor): preserve buffered output on abort --- .../agent/backends/acp/AcpMessageHandler.ts | 16 +- .../agent/backends/acp/AcpSdkBackend.test.ts | 243 ++++++++++++++ cli/src/agent/backends/acp/AcpSdkBackend.ts | 145 ++++++++- .../backends/acp/AcpStdioTransport.test.ts | 25 +- .../agent/backends/acp/AcpStdioTransport.ts | 157 +++++++-- .../cursor/cursorAcpRemoteLauncher.test.ts | 305 +++++++++++++++++- cli/src/cursor/cursorAcpRemoteLauncher.ts | 241 +++++++++++++- cli/src/cursor/cursorLegacyRemoteLauncher.ts | 9 + cli/src/cursor/runCursor.test.ts | 1 + cli/src/cursor/runCursor.ts | 1 + hub/src/sync/steerQueuedMessage.test.ts | 2 +- hub/src/sync/syncEngine.ts | 10 +- shared/src/modes.test.ts | 20 +- shared/src/modes.ts | 16 +- web/src/components/SessionChat.tsx | 4 +- 15 files changed, 1138 insertions(+), 57 deletions(-) diff --git a/cli/src/agent/backends/acp/AcpMessageHandler.ts b/cli/src/agent/backends/acp/AcpMessageHandler.ts index a08fdb70..69aaec32 100644 --- a/cli/src/agent/backends/acp/AcpMessageHandler.ts +++ b/cli/src/agent/backends/acp/AcpMessageHandler.ts @@ -398,6 +398,7 @@ function getSuffixPrefixOverlap(base: string, next: string): number { export class AcpMessageHandler { private readonly toolCalls = new Map(); + private acceptingUpdates = true; private bufferedText = ''; // Array buffer avoids the O(N²) string concatenation that per-token // ACP streams (OpenCode/Zen emits one chunk per generated token) would @@ -410,13 +411,26 @@ export class AcpMessageHandler { private reasoningSnapshotEmitted = false; private readonly textChunkMode: AcpTextChunkMode; + private readonly onMessage: (message: AgentMessage) => void; + constructor( - private readonly onMessage: (message: AgentMessage) => void, + onMessage: (message: AgentMessage) => void, private readonly options: { textChunkMode?: AcpTextChunkMode; flavor?: string } = {} ) { + this.onMessage = (message) => { + if (this.acceptingUpdates) onMessage(message); + }; this.textChunkMode = this.options.textChunkMode ?? 'dedupe'; } + /** Drop late updates from a cancelled prompt before the next handler exists. */ + deactivate(): void { + this.acceptingUpdates = false; + this.bufferedText = ''; + this.bufferedReasoning = []; + this.resetReasoningState(); + } + /** * Emits any buffered assistant text as a single message and clears the * buffer. Callers must treat this as a text-segment boundary: it is diff --git a/cli/src/agent/backends/acp/AcpSdkBackend.test.ts b/cli/src/agent/backends/acp/AcpSdkBackend.test.ts index 4d17fa7b..4d903601 100644 --- a/cli/src/agent/backends/acp/AcpSdkBackend.test.ts +++ b/cli/src/agent/backends/acp/AcpSdkBackend.test.ts @@ -1,6 +1,7 @@ import { afterEach, describe, expect, it, vi } from 'vitest'; import type { AgentMessage } from '@/agent/types'; import { AcpSdkBackend } from './AcpSdkBackend'; +import { AcpMessageHandler } from './AcpMessageHandler'; import { buildAcpStdioSpawnOptions } from './AcpStdioTransport'; import { ACP_SESSION_UPDATE_TYPES } from './constants'; @@ -1142,6 +1143,194 @@ describe('AcpSdkBackend', () => { expect(registered.get('cursor/ask_question')).toBe(handler); }); + it('beginSoftSteerPrompt sends concurrent session/prompt without cancel', () => { + const backend = new AcpSdkBackend({ command: 'agent' }); + const calls: Array<{ method: string; params: unknown }> = []; + const backendInternal = backend as unknown as { + transport: { + sendRequestWithDispatch: (method: string, params: unknown, options?: { timeoutMs?: number }) => { dispatched: Promise; completed: Promise }; + sendNotification: (method: string, params: unknown) => void; + close: () => Promise; + } | null; + isProcessingMessage: boolean; + }; + backendInternal.isProcessingMessage = true; + backendInternal.transport = { + sendRequestWithDispatch: (method, params) => { + calls.push({ method, params }); + return { dispatched: Promise.resolve(), completed: Promise.resolve({ stopReason: 'end_turn' }) }; + }, + sendNotification: () => {}, + close: async () => {} + }; + + backend.beginSoftSteerPrompt('session-1', [{ type: 'text', text: 'pivot now' }]); + + expect(calls).toEqual([{ + method: 'session/prompt', + params: { + sessionId: 'session-1', + prompt: [{ type: 'text', text: 'pivot now' }] + } + }]); + }); + + it('beginSoftSteerPrompt drains buffered output after completion', async () => { + const backend = new AcpSdkBackend({ command: 'agent' }); + const backendInternal = backend as unknown as { + transport: { + sendRequestWithDispatch: (method: string, params: unknown, options?: { timeoutMs?: number }) => { dispatched: Promise; completed: Promise }; + close: () => Promise; + } | null; + isProcessingMessage: boolean; + waitForSessionUpdateQuiet: (quietMs: number, timeoutMs: number) => Promise; + drainLateBuffers: () => Promise; + messageHandler: { drainBuffers: () => void } | null; + }; + const events: string[] = []; + backendInternal.isProcessingMessage = true; + backendInternal.transport = { + sendRequestWithDispatch: () => { + events.push('request'); + return { dispatched: Promise.resolve(), completed: Promise.resolve({ stopReason: 'end_turn' }) }; + }, + close: async () => {} + }; + backendInternal.waitForSessionUpdateQuiet = async () => { + events.push('quiet'); + }; + backendInternal.messageHandler = { + drainBuffers: () => events.push('drain') + }; + backendInternal.drainLateBuffers = async () => { + events.push('late'); + }; + + await backend.beginSoftSteerPrompt('session-1', [{ type: 'text', text: 'pivot now' }]).completed; + + expect(events).toEqual(['request', 'quiet', 'drain', 'late', 'drain']); + }); + + it('beginSoftSteerPrompt returns a pending promise without blocking the caller', async () => { + const backend = new AcpSdkBackend({ command: 'agent' }); + let resolvePrompt: ((value: unknown) => void) | null = null; + const backendInternal = backend as unknown as { + transport: { + sendRequestWithDispatch: (method: string, params: unknown, options?: { timeoutMs?: number }) => { dispatched: Promise; completed: Promise }; + sendNotification: (method: string, params: unknown) => void; + close: () => Promise; + } | null; + isProcessingMessage: boolean; + }; + backendInternal.isProcessingMessage = true; + backendInternal.transport = { + sendRequestWithDispatch: () => ({ + dispatched: Promise.resolve(), + completed: new Promise((resolve) => { resolvePrompt = resolve; }) + }), + sendNotification: () => {}, + close: async () => {} + }; + + // Must not hang waiting for the ACP prompt response (hub RPC is 30s). + let settled = false; + const pending = backend.beginSoftSteerPrompt('session-1', [{ type: 'text', text: 'pivot now' }]).completed.then(() => { + settled = true; + }); + expect(resolvePrompt).not.toBeNull(); + expect(settled).toBe(false); + resolvePrompt!({ stopReason: 'end_turn' }); + await pending; + expect(settled).toBe(true); + }); + + it('keeps response completion pending until a concurrent soft steer settles', async () => { + const backend = new AcpSdkBackend({ command: 'agent' }); + let resolvePrompt: ((value: unknown) => void) | null = null; + const backendInternal = backend as unknown as { + transport: { sendRequestWithDispatch: () => { dispatched: Promise; completed: Promise }; close: () => Promise } | null; + isProcessingMessage: boolean; + activePromptRequests: number; + finishPromptRequest: (epoch: number) => void; + waitForSessionUpdateQuiet: () => Promise; + drainLateBuffers: () => Promise; + }; + backendInternal.isProcessingMessage = true; + backendInternal.activePromptRequests = 1; + backendInternal.transport = { + sendRequestWithDispatch: () => ({ + dispatched: Promise.resolve(), + completed: new Promise((resolve) => { resolvePrompt = resolve; }) + }), + close: async () => {} + }; + backendInternal.waitForSessionUpdateQuiet = async () => {}; + backendInternal.drainLateBuffers = async () => {}; + + const pendingSteer = backend.beginSoftSteerPrompt('session-1', [{ type: 'text', text: 'pivot now' }]); + let responseComplete = false; + const responseWait = backend.waitForResponseComplete().then(() => { responseComplete = true; }); + + backendInternal.finishPromptRequest(0); + await Promise.resolve(); + expect(backend.processingMessage).toBe(true); + expect(responseComplete).toBe(false); + + resolvePrompt!({ stopReason: 'end_turn' }); + await pendingSteer.completed; + await responseWait; + expect(backend.processingMessage).toBe(false); + expect(responseComplete).toBe(true); + }); + + it('softSteerPrompt awaits session/prompt completion', async () => { + const backend = new AcpSdkBackend({ command: 'agent' }); + let resolvePrompt: ((value: unknown) => void) | null = null; + const backendInternal = backend as unknown as { + transport: { + sendRequest: (method: string, params: unknown, options?: { timeoutMs?: number }) => Promise; + sendNotification: (method: string, params: unknown) => void; + close: () => Promise; + } | null; + isProcessingMessage: boolean; + }; + backendInternal.isProcessingMessage = true; + backendInternal.transport = { + sendRequest: () => new Promise((resolve) => { + resolvePrompt = resolve; + }), + sendNotification: () => {}, + close: async () => {} + }; + + let done = false; + const pending = backend.softSteerPrompt('session-1', [{ type: 'text', text: 'pivot now' }]).then(() => { + done = true; + }); + await Promise.resolve(); + expect(done).toBe(false); + resolvePrompt!({ stopReason: 'end_turn' }); + await pending; + expect(done).toBe(true); + }); + + it('beginSoftSteerPrompt rejects when no prompt is in flight', () => { + const backend = new AcpSdkBackend({ command: 'agent' }); + const backendInternal = backend as unknown as { + transport: { sendRequest: () => Promise; close: () => Promise } | null; + isProcessingMessage: boolean; + }; + backendInternal.isProcessingMessage = false; + backendInternal.transport = { + sendRequest: async () => null, + close: async () => {} + }; + + expect(() => backend.beginSoftSteerPrompt('session-1', [{ type: 'text', text: 'x' }])) + .toThrow(/No active ACP prompt/); + }); + + it('suppressUpdatesDuring drops session/update notifications that would otherwise leak into the previous turn\'s onUpdate, then restores normal forwarding', async () => { // Reproduces the real /compact duplicate-summary bug: OpenCode keeps // streaming session/update notifications (over the same ACP @@ -1540,3 +1729,57 @@ describe('AcpSdkBackend', () => { await pending; }); }); + +describe('AcpSdkBackend abortSoftSteers', () => { + it('drains buffered foreground output before suppressing late updates', async () => { + const backend = new AcpSdkBackend({ command: 'agent' }); + const updates: AgentMessage[] = []; + const handler = new AcpMessageHandler((message) => updates.push(message)); + const backendInternal = backend as unknown as { + messageHandler: AcpMessageHandler | null; + }; + + await handler.handleUpdate({ + sessionUpdate: ACP_SESSION_UPDATE_TYPES.agentMessageChunk, + content: { type: 'text', text: 'partial answer' } + }); + await handler.handleUpdate({ + sessionUpdate: ACP_SESSION_UPDATE_TYPES.agentThoughtChunk, + content: { type: 'text', text: 'partial thought' } + }); + backendInternal.messageHandler = handler; + + backend.abortSoftSteers(); + + expect(updates).toEqual([ + { type: 'reasoning', text: 'partial thought' }, + { type: 'text', text: 'partial answer' } + ]); + }); + + it('releases processingMessage without waiting for the concurrent prompt', () => { + const backend = new AcpSdkBackend({ command: 'agent' }); + const backendInternal = backend as unknown as { + isProcessingMessage: boolean; + activePromptRequests: number; + }; + backendInternal.isProcessingMessage = true; + backendInternal.activePromptRequests = 2; + + backend.abortSoftSteers(); + + expect(backend.processingMessage).toBe(false); + expect(backendInternal.activePromptRequests).toBe(0); + }); + + it('is a no-op when nothing is in flight', () => { + const backend = new AcpSdkBackend({ command: 'agent' }); + const backendInternal = backend as unknown as { + activePromptRequests: number; + }; + backendInternal.activePromptRequests = 0; + + expect(() => backend.abortSoftSteers()).not.toThrow(); + expect(backend.processingMessage).toBe(false); + }); +}); diff --git a/cli/src/agent/backends/acp/AcpSdkBackend.ts b/cli/src/agent/backends/acp/AcpSdkBackend.ts index f686a9dd..dca43137 100644 --- a/cli/src/agent/backends/acp/AcpSdkBackend.ts +++ b/cli/src/agent/backends/acp/AcpSdkBackend.ts @@ -78,6 +78,14 @@ export class AcpSdkBackend implements AgentBackend { private setModeSupported: boolean | undefined = undefined; private isProcessingMessage = false; private promptRequestInFlight = false; + /** Concurrent session/prompt requests (main prompt + soft steers). */ + private activePromptRequests = 0; + /** Foreground prompt only; soft steers are excluded after Abort. */ + private foregroundPromptRequests = 0; + /** Bumped by abortSoftSteers; stale finishes from cancelled requests are dropped. */ + private promptRequestEpoch = 0; + /** Incremented for each foreground prompt turn, including retry-wrapped turns. */ + private promptGeneration = 0; private responseCompleteResolvers: Array<() => void> = []; private lastSessionUpdateAt = 0; private latestUsageUpdate: AcpUsageUpdate | null = null; @@ -557,7 +565,9 @@ export class AcpSdkBackend implements AgentBackend { textChunkMode: this.options.textChunkMode, flavor: this.options.flavor, }); - this.isProcessingMessage = true; + this.promptGeneration++; + this.foregroundPromptRequests++; + const promptRequestEpoch = this.beginPromptRequest(); this.lastSessionUpdateAt = Date.now(); this.latestUsageUpdate = null; this.lastForwardedUsageUpdate = null; @@ -635,8 +645,14 @@ export class AcpSdkBackend implements AgentBackend { } } finally { this.promptUsageCallback = null; - this.isProcessingMessage = false; - this.notifyResponseComplete(); + this.foregroundPromptRequests = Math.max(0, this.foregroundPromptRequests - 1); + if (promptRequestEpoch !== this.promptRequestEpoch) { + this.activePromptRequests = Math.max(0, this.activePromptRequests - 1); + this.isProcessingMessage = this.activePromptRequests > 0; + if (!this.isProcessingMessage) this.notifyResponseComplete(); + } else { + this.finishPromptRequest(promptRequestEpoch); + } } } } @@ -649,6 +665,83 @@ export class AcpSdkBackend implements AgentBackend { this.transport.sendNotification('session/cancel', { sessionId }); } + /** + * Soft-inject a follow-up `session/prompt` while another prompt is in flight. + * + * Used for Cursor mid-turn steer (GUI "Send" / next-opportune soft send). + * Does **not** cancel the active prompt and does **not** swap message handlers — + * `session/update` notifications keep flowing to the in-flight turn's handler. + * + * Awaits the full concurrent `session/prompt` JSON-RPC response (turn completion + * for that inject). Do **not** call this from the hub `SteerQueuedMessage` handler — + * that RPC uses a 30s Socket.IO timeout. Use {@link beginSoftSteerPrompt} there. + */ + async softSteerPrompt(sessionId: string, content: PromptContent[]): Promise { + if (!this.transport) { + throw new Error('ACP transport not initialized'); + } + if (!this.isProcessingMessage) { + throw new Error('No active ACP prompt to soft-steer into'); + } + + const promptRequestEpoch = this.beginPromptRequest(); + try { + await this.transport.sendRequest('session/prompt', { + sessionId, + prompt: content + }, { timeoutMs: Infinity }); + } finally { + this.finishPromptRequest(promptRequestEpoch); + } + } + + /** + * Kick off a soft steer without blocking the hub RPC on turn completion. + * Separates transport dispatch from prompt completion so callers can commit + * queue state only after stdin accepted the request without waiting for the turn. + */ + beginSoftSteerPrompt(sessionId: string, content: PromptContent[]): { + dispatched: Promise; + completed: Promise; + } { + if (!this.transport) { + throw new Error('ACP transport not initialized'); + } + if (!this.isProcessingMessage) { + throw new Error('No active ACP prompt to soft-steer into'); + } + + const transport = this.transport; + const promptRequestEpoch = this.beginPromptRequest(); + const request = transport.sendRequestWithDispatch('session/prompt', { + sessionId, + prompt: content + }, { timeoutMs: Infinity, dispatchTimeoutMs: 20_000 }); + const completed = (async () => { + try { + await request.completed; + } finally { + try { + await this.waitForSessionUpdateQuiet( + AcpSdkBackend.UPDATE_QUIET_PERIOD_MS, + AcpSdkBackend.UPDATE_DRAIN_TIMEOUT_MS + ); + this.messageHandler?.drainBuffers(); + await this.drainLateBuffers(); + this.messageHandler?.drainBuffers(); + } finally { + this.finishPromptRequest(promptRequestEpoch); + } + } + })(); + + void completed.catch((error) => { + logger.warn('[ACP] soft-steer session/prompt failed', error); + }); + + return { dispatched: request.dispatched, completed }; + } + async respondToPermission( _sessionId: string, request: PermissionRequest, @@ -750,13 +843,17 @@ export class AcpSdkBackend implements AgentBackend { * Useful for checking if it's safe to perform session operations. */ get processingMessage(): boolean { - return this.isProcessingMessage; + return this.activePromptRequests > 0; } isPromptRequestInFlight(): boolean { return this.promptRequestInFlight; } + getPromptGeneration(): number { + return this.promptGeneration; + } + getLastSessionUpdateAt(): number { return this.lastSessionUpdateAt; } @@ -768,7 +865,7 @@ export class AcpSdkBackend implements AgentBackend { * like session swap or sending task_complete. */ async waitForResponseComplete(): Promise { - if (!this.isProcessingMessage) { + if (this.activePromptRequests === 0) { return; } return new Promise((resolve) => { @@ -787,6 +884,8 @@ export class AcpSdkBackend implements AgentBackend { this.messageHandler?.drainBuffers(); this.messageHandler = null; this.activeSessionId = null; + this.activePromptRequests = 0; + this.foregroundPromptRequests = 0; this.isProcessingMessage = false; this.sessionModelsMetadata.clear(); this.initialAvailableCommands.clear(); @@ -1067,6 +1166,42 @@ export class AcpSdkBackend implements AgentBackend { return await responsePromise; } + private beginPromptRequest(): number { + this.activePromptRequests++; + this.isProcessingMessage = true; + return this.promptRequestEpoch; + } + + /** + * Force-settle soft-steer bookkeeping without waiting for the concurrent + * `session/prompt` to finish. Called on abort: the in-flight turn is + * cancelled anyway, so a pending soft steer may never complete; dropping + * its counter keeps {@link waitForResponseComplete} from blocking the next + * turn. Bumps the epoch so a stale finish from a cancelled request cannot + * decrement a newer prompt's counter. + */ + abortSoftSteers(): void { + this.messageHandler?.drainBuffers(); + this.messageHandler?.deactivate?.(); + this.promptRequestEpoch++; + this.activePromptRequests = this.foregroundPromptRequests; + this.isProcessingMessage = this.activePromptRequests > 0; + if (!this.isProcessingMessage) { + this.notifyResponseComplete(); + } + } + + private finishPromptRequest(epoch: number): void { + if (epoch !== this.promptRequestEpoch) { + return; + } + this.activePromptRequests = Math.max(0, this.activePromptRequests - 1); + this.isProcessingMessage = this.activePromptRequests > 0; + if (!this.isProcessingMessage) { + this.notifyResponseComplete(); + } + } + private notifyResponseComplete(): void { const resolvers = this.responseCompleteResolvers; this.responseCompleteResolvers = []; diff --git a/cli/src/agent/backends/acp/AcpStdioTransport.test.ts b/cli/src/agent/backends/acp/AcpStdioTransport.test.ts index 33a94d54..5013618a 100644 --- a/cli/src/agent/backends/acp/AcpStdioTransport.test.ts +++ b/cli/src/agent/backends/acp/AcpStdioTransport.test.ts @@ -91,7 +91,7 @@ vi.mock('node:child_process', () => ({ }) })); -import { AcpStdioTransport } from './AcpStdioTransport'; +import { AcpStdioTransport, isAcpIndeterminateError } from './AcpStdioTransport'; import { killProcessByChildProcess } from '@/utils/process'; function emitStdout(chunk: string): void { @@ -594,6 +594,29 @@ describe('AcpStdioTransport closed stdin writes', () => { expect(buffer.length).toBeLessThanOrEqual(8_000); }); + test('bounds a stalled stdin dispatch and rejects both promises', async () => { + vi.useFakeTimers(); + try { + const transport = await AcpStdioTransport.create({ command: 'gemini' }); + const request = transport.sendRequestWithDispatch('session/prompt', {}, { + timeoutMs: Infinity, + dispatchTimeoutMs: 10 + }); + let dispatchError: unknown; + let completedError: unknown; + const dispatch = request.dispatched.catch((error) => { dispatchError = error; }); + const completed = request.completed.catch((error) => { completedError = error; }); + await vi.advanceTimersByTimeAsync(10); + await dispatch; + await completed; + expect(isAcpIndeterminateError(dispatchError)).toBe(true); + expect(isAcpIndeterminateError(completedError)).toBe(true); + await transport.close(); + } finally { + vi.useRealTimers(); + } + }); + test('rejects pending requests when stdin.write throws', async () => { spawnState.stdinWrite.mockImplementation(() => { throw new Error('WritableIterable is closed'); diff --git a/cli/src/agent/backends/acp/AcpStdioTransport.ts b/cli/src/agent/backends/acp/AcpStdioTransport.ts index d5bee14c..e94cf840 100644 --- a/cli/src/agent/backends/acp/AcpStdioTransport.ts +++ b/cli/src/agent/backends/acp/AcpStdioTransport.ts @@ -16,6 +16,20 @@ import { } from './agentCliGuard'; import { matchesAcpHttp2Cancel, matchesAcpRetryBackoff } from './acpStderrErrors'; +/** Marks transport-level failures whose request outcome is unknown (unlike an + * explicit JSON-RPC error response). */ +export const ACP_INDETERMINATE_SYMBOL = Symbol('acp-indeterminate'); + +function markAcpIndeterminate(error: Error): Error { + Object.defineProperty(error, ACP_INDETERMINATE_SYMBOL, { value: true }); + return error; +} + +export function isAcpIndeterminateError(error: unknown): boolean { + return typeof error === 'object' && error !== null + && (error as Record)[ACP_INDETERMINATE_SYMBOL] === true; +} + interface JsonRpcRequest { jsonrpc: '2.0'; id: string | number | null; @@ -68,6 +82,7 @@ export class AcpStdioTransport { private readonly pending = new Map void; reject: (error: Error) => void; + rejectDispatched: (error: Error) => void; }>(); private readonly requestHandlers = new Map(); private notificationHandler: ((method: string, params: unknown) => void) | null = null; @@ -243,11 +258,25 @@ export class AcpStdioTransport { /** Default timeout for requests in milliseconds (2 minutes) */ static readonly DEFAULT_TIMEOUT_MS = 120_000; - async sendRequest(method: string, params?: unknown, options?: { timeoutMs?: number }): Promise { + async sendRequest(method: string, params?: unknown, options?: { timeoutMs?: number; dispatchTimeoutMs?: number }): Promise { + const request = this.sendRequestWithDispatch(method, params, options); + void request.dispatched.catch(() => {}); + return request.completed; + } + + /** + * Split a request into transport dispatch (stdin accepted) and completion + * (JSON-RPC response). Lets callers commit state once stdin accepted the + * request without waiting for the (possibly long-running) response. + */ + sendRequestWithDispatch( + method: string, + params?: unknown, + options?: { timeoutMs?: number; dispatchTimeoutMs?: number } + ): { dispatched: Promise; completed: Promise } { if (this.closed || this.exited) { - return Promise.reject( - this.closeError ?? this.exitError ?? new Error('ACP transport is closed') - ); + const error = markAcpIndeterminate(this.closeError ?? this.exitError ?? new Error('ACP transport is closed')); + return { dispatched: Promise.reject(error), completed: Promise.reject(error) }; } const id = this.nextId++; @@ -259,37 +288,107 @@ export class AcpStdioTransport { }; const timeoutMs = options?.timeoutMs ?? AcpStdioTransport.DEFAULT_TIMEOUT_MS; + const dispatchTimeoutMs = options?.dispatchTimeoutMs ?? timeoutMs; - // Skip timeout for infinite/no-timeout requests (e.g., long-running prompts) - if (!Number.isFinite(timeoutMs)) { - return new Promise((resolve, reject) => { - this.pending.set(id, { resolve, reject }); - this.writePayload(payload); - }); - } + let timer: ReturnType | null = null; + let dispatchTimer: ReturnType | null = null; + let resolveDispatched!: () => void; + let rejectDispatched!: (error: Error) => void; + let resolveCompleted!: (value: unknown) => void; + let rejectCompleted!: (error: Error) => void; + const dispatched = new Promise((resolve, reject) => { + resolveDispatched = resolve; + rejectDispatched = reject; + }); + const completed = new Promise((resolve, reject) => { + resolveCompleted = resolve; + rejectCompleted = reject; + }); + let dispatchSettled = false; - return new Promise((resolve, reject) => { - const timer = setTimeout(() => { + const clearTimers = () => { + if (timer) clearTimeout(timer); + if (dispatchTimer) clearTimeout(dispatchTimer); + }; + const failRequest = (error: Error) => { + this.pending.delete(id); + clearTimers(); + if (!dispatchSettled) { + dispatchSettled = true; + rejectDispatched(error); + } + rejectCompleted(error); + }; + if (Number.isFinite(timeoutMs)) { + timer = setTimeout(() => { if (this.pending.has(id)) { - this.pending.delete(id); - reject(new Error(`ACP request '${method}' timed out after ${timeoutMs}ms`)); + failRequest(markAcpIndeterminate(new Error(`ACP request '${method}' timed out after ${timeoutMs}ms`))); } }, timeoutMs); - // Don't let timer keep Node alive if process wants to exit timer.unref(); + } + if (Number.isFinite(dispatchTimeoutMs)) { + dispatchTimer = setTimeout(() => { + if (this.pending.has(id) && !dispatchSettled) { + const error = markAcpIndeterminate(new Error(`ACP request '${method}' dispatch timed out after ${dispatchTimeoutMs}ms`)); + try { + this.process.stdin.destroy(); + } catch (destroyError) { + logger.debug('[ACP] Error destroying stalled stdin', destroyError); + } + this.markClosed(error); + } + }, dispatchTimeoutMs); + dispatchTimer.unref(); + } - this.pending.set(id, { - resolve: (value) => { - clearTimeout(timer); - resolve(value); - }, - reject: (error) => { - clearTimeout(timer); - reject(error); + this.pending.set(id, { + resolve: (value) => { + clearTimers(); + if (!dispatchSettled) { + dispatchSettled = true; + resolveDispatched(); + } + resolveCompleted(value); + }, + reject: (error) => { + clearTimers(); + if (!dispatchSettled) { + dispatchSettled = true; + resolveDispatched(); + } + rejectCompleted(error); + }, + rejectDispatched: (error) => { + if (!dispatchSettled) { + dispatchSettled = true; + rejectDispatched(error); + } + } + }); + + try { + const serialized = JSON.stringify(payload); + this.process.stdin.write(`${serialized}\n`, (error) => { + if (error) { + const writeError = markAcpIndeterminate(error instanceof Error ? error : new Error(String(error))); + this.markClosed(writeError); + failRequest(writeError); + return; + } + if (!dispatchSettled) { + dispatchSettled = true; + if (dispatchTimer) clearTimeout(dispatchTimer); + resolveDispatched(); } }); - this.writePayload(payload); - }); + } catch (error) { + const writeError = error instanceof Error ? error : new Error(String(error)); + this.markClosed(writeError); + failRequest(writeError); + } + + return { dispatched, completed }; } sendNotification(method: string, params?: unknown): void { @@ -475,8 +574,10 @@ export class AcpStdioTransport { } private rejectAllPending(error: Error): void { - for (const { reject } of this.pending.values()) { - reject(error); + const indeterminate = markAcpIndeterminate(error); + for (const { reject, rejectDispatched } of this.pending.values()) { + rejectDispatched(indeterminate); + reject(indeterminate); } this.pending.clear(); } diff --git a/cli/src/cursor/cursorAcpRemoteLauncher.test.ts b/cli/src/cursor/cursorAcpRemoteLauncher.test.ts index ba869896..010b4bd0 100644 --- a/cli/src/cursor/cursorAcpRemoteLauncher.test.ts +++ b/cli/src/cursor/cursorAcpRemoteLauncher.test.ts @@ -2,6 +2,7 @@ import { afterEach, describe, expect, it, vi, beforeEach } from 'vitest'; import { MessageQueue2 } from '@/utils/MessageQueue2'; import type { EnhancedMode } from './loop'; import type { AgentMessage } from '@/agent/types'; +import { ACP_INDETERMINATE_SYMBOL } from '@/agent/backends/acp/AcpStdioTransport'; const harness = vi.hoisted(() => ({ initializeError: null as Error | null, @@ -15,11 +16,14 @@ const harness = vi.hoisted(() => ({ newSessionAttempts: 0, promptCalls: 0, prompts: [] as unknown[][], + deferPrompt: null as Promise | null, + deferSoftSteer: null as Promise | null, + softSteerDispatchError: null as Error | null, + deferSoftSteerDispatch: null as Promise | null, promptErrors: [] as Error[], promptMessages: [] as AgentMessage[], promptMessageBatches: [] as AgentMessage[][], promptStderrErrors: [] as Array<{ type: string; message: string; raw: string }>, - deferPrompt: null as Promise | null, releasePrompt: null as (() => void) | null, backendArgs: null as { command: string; args?: string[] } | null, setConfigOptionCalls: [] as Array<{ sessionId: string; configId: string; value: string }>, @@ -145,6 +149,16 @@ vi.mock('./utils/cursorAcpBackend', () => ({ if (error) throw error; }), cancelPrompt: vi.fn(async () => {}), + getPromptGeneration: vi.fn(() => 1), + beginSoftSteerPrompt: vi.fn(() => ({ + dispatched: harness.softSteerDispatchError + ? Promise.reject(harness.softSteerDispatchError) + : (harness.deferSoftSteerDispatch ?? Promise.resolve()), + completed: harness.deferSoftSteer ?? Promise.resolve() + })), + softSteerPrompt: vi.fn(async () => {}), + abortSoftSteers: vi.fn(), + waitForResponseComplete: vi.fn(async () => {}), respondToPermission: vi.fn(async () => {}), onStderrError: vi.fn((handler) => { harness.stderrErrorHandler = handler ?? null; @@ -206,6 +220,7 @@ vi.mock('@/ui/logger', () => ({ import { classifyCursorAcpLoadError, cursorAcpRemoteLauncher } from './cursorAcpRemoteLauncher'; import { createCursorAcpBackend } from './utils/cursorAcpBackend'; +import { RPC_METHODS } from '@hapi/protocol/rpcMethods'; import { CursorSession } from './session'; import { ApiSessionClient } from '@/api/apiSession'; import { @@ -213,7 +228,7 @@ import { writeSharedCursorModelsCache } from '@/modules/common/cursorModelsSharedCache'; -function makeSession(sessionId: string | null): CursorSession { +function makeSession(sessionId: string | null, closeQueue = true): CursorSession { const queue = new MessageQueue2(() => 'mode'); const client = makeClient(); @@ -232,22 +247,31 @@ function makeSession(sessionId: string | null): CursorSession { }); session.onSessionFoundWithProtocol = vi.fn(); - queue.close(); + if (closeQueue) { + queue.close(); + } return session; } function makeClient() { + const handlers = new Map Promise>(); return { sessionId: 'test-session-id', rpcHandlerManager: { - registerHandler: vi.fn(), + handlers, + registerHandler: vi.fn((method: string, handler: (payload?: unknown) => Promise) => { + handlers.set(method, handler); + }), unregisterHandler: vi.fn() }, updateMetadata: vi.fn(), flushMetadata: vi.fn(async () => true), sendSessionEvent: vi.fn(), sendAgentMessage: vi.fn(), + emitMessagesConsumed: vi.fn(), + emitSteerIndeterminate: vi.fn(), + setSteerDeliveryState: vi.fn(async () => true), sendClaudeSessionMessage: vi.fn(), keepAlive: vi.fn(), emitSessionReady: vi.fn() @@ -267,11 +291,14 @@ describe('cursorAcpRemoteLauncher', () => { harness.newSessionAttempts = 0; harness.promptCalls = 0; harness.prompts = []; + harness.deferPrompt = null; + harness.deferSoftSteer = null; + harness.softSteerDispatchError = null; + harness.deferSoftSteerDispatch = null; harness.promptErrors = []; harness.promptMessages = []; harness.promptMessageBatches = []; harness.promptStderrErrors = []; - harness.deferPrompt = null; harness.releasePrompt = null; harness.setConfigOptionCalls = []; harness.deferSetConfigOption = null; @@ -292,6 +319,274 @@ describe('cursorAcpRemoteLauncher', () => { _resetSharedCursorModelsCacheForTests(); }); + it('ends the launcher when a soft steer outlives an abort', async () => { + let releasePrompt!: () => void; + harness.deferPrompt = new Promise((resolve) => { releasePrompt = resolve; }); + // Soft-steer completion never settles — simulates Cursor keeping the + // concurrent request open after an ordinary Abort. + harness.deferSoftSteer = new Promise(() => {}); + const session = makeSession(null, false); + const mode = { permissionMode: 'default' } as EnhancedMode; + session.queue.push('first', mode, 'first'); + + const runPromise = cursorAcpRemoteLauncher(session); + await vi.waitFor(() => expect(harness.promptCalls).toBe(1)); + const handlers = (session.client as unknown as { + rpcHandlerManager: { handlers: Map Promise> }; + }).rpcHandlerManager.handlers; + + session.queue.push('soft steer', mode, 'steer'); + await handlers.get(RPC_METHODS.SteerQueuedMessage)!({ localId: 'steer' }); + harness.deferPrompt = null; + releasePrompt(); + await handlers.get(RPC_METHODS.Abort)!(); + + // The old soft steer never settled, so the launcher must not install + // another prompt handler over it; the bounded drain ends the session. + expect(harness.promptCalls).toBe(1); + session.queue.close(); + await runPromise; + }, 10_000); + + it('restores a queued steer when ACP dispatch fails', async () => { + let releasePrompt!: () => void; + harness.deferPrompt = new Promise((resolve) => { releasePrompt = resolve; }); + harness.softSteerDispatchError = new Error('stdin closed'); + const session = makeSession(null, false); + const mode = { permissionMode: 'default' } as EnhancedMode; + session.queue.push('first', mode, 'first'); + + const runPromise = cursorAcpRemoteLauncher(session); + await vi.waitFor(() => expect(harness.promptCalls).toBe(1)); + const handlers = (session.client as unknown as { + rpcHandlerManager: { handlers: Map Promise> }; + }).rpcHandlerManager.handlers; + + session.queue.push('soft steer', mode, 'steer'); + await expect(handlers.get(RPC_METHODS.SteerQueuedMessage)!({ localId: 'steer' })) + .resolves.toEqual({ steered: false, error: 'Failed to soft-steer into active turn' }); + expect(session.client.emitMessagesConsumed).not.toHaveBeenCalledWith(['steer'], { steered: true }); + + harness.softSteerDispatchError = null; + harness.deferPrompt = null; + releasePrompt(); + session.queue.close(); + await runPromise; + }); + + it('restores the row when ACP explicitly rejects after dispatch', async () => { + let releasePrompt!: () => void; + let rejectSoftSteer!: (error: Error) => void; + harness.deferPrompt = new Promise((resolve) => { releasePrompt = resolve; }); + harness.deferSoftSteer = new Promise((_, reject) => { rejectSoftSteer = reject; }); + const session = makeSession(null, false); + const mode = { permissionMode: 'default' } as EnhancedMode; + session.queue.push('first', mode, 'first'); + + const runPromise = cursorAcpRemoteLauncher(session); + await vi.waitFor(() => expect(harness.promptCalls).toBe(1)); + const handlers = (session.client as unknown as { + rpcHandlerManager: { handlers: Map Promise> }; + }).rpcHandlerManager.handlers; + + session.queue.push('soft steer', mode, 'steer'); + await expect(handlers.get(RPC_METHODS.SteerQueuedMessage)!({ localId: 'steer' })) + .resolves.toEqual({ steered: true }); + // An explicit JSON-RPC rejection means ACP never accepted the + // instruction — the row is restored for the next prompt (no data loss). + rejectSoftSteer(new Error('request rejected')); + await vi.waitFor(() => expect(session.queue.cancelByLocalId('steer')).toBe(true)); + expect(session.client.emitMessagesConsumed).not.toHaveBeenCalledWith(['steer'], { steered: true }); + + harness.deferPrompt = null; + releasePrompt(); + session.queue.close(); + await runPromise; + }); + + it('holds a transport-ambiguous steer for explicit retry or cancel', async () => { + let releasePrompt!: () => void; + let rejectSoftSteer!: (error: Error) => void; + harness.deferPrompt = new Promise((resolve) => { releasePrompt = resolve; }); + harness.deferSoftSteer = new Promise((_, reject) => { rejectSoftSteer = reject; }); + const indeterminate = new Error('ACP transport closed'); + Object.defineProperty(indeterminate, ACP_INDETERMINATE_SYMBOL, { value: true }); + const session = makeSession(null, false); + const mode = { permissionMode: 'default' } as EnhancedMode; + session.queue.push('first', mode, 'first'); + + const runPromise = cursorAcpRemoteLauncher(session); + await vi.waitFor(() => expect(harness.promptCalls).toBe(1)); + const handlers = (session.client as unknown as { + rpcHandlerManager: { handlers: Map Promise> }; + }).rpcHandlerManager.handlers; + + session.queue.push('soft steer', mode, 'steer'); + await expect(handlers.get(RPC_METHODS.SteerQueuedMessage)!({ localId: 'steer' })) + .resolves.toEqual({ steered: true }); + rejectSoftSteer(indeterminate); + await vi.waitFor(() => expect(session.queue.cancelByLocalId('steer')).toBe(true)); + expect(session.client.emitSteerIndeterminate).toHaveBeenCalledWith(['steer']); + expect(session.client.emitMessagesConsumed).not.toHaveBeenCalledWith(['steer'], { steered: true }); + + harness.deferPrompt = null; + releasePrompt(); + session.queue.close(); + await runPromise; + }); + + it('holds an indeterminate dispatch failure instead of restoring it', async () => { + let releasePrompt!: () => void; + harness.deferPrompt = new Promise((resolve) => { releasePrompt = resolve; }); + const indeterminate = new Error('ACP write callback failed'); + Object.defineProperty(indeterminate, ACP_INDETERMINATE_SYMBOL, { value: true }); + harness.softSteerDispatchError = indeterminate; + const session = makeSession(null, false); + const mode = { permissionMode: 'default' } as EnhancedMode; + session.queue.push('first', mode, 'first'); + + const runPromise = cursorAcpRemoteLauncher(session); + await vi.waitFor(() => expect(harness.promptCalls).toBe(1)); + const handlers = (session.client as unknown as { + rpcHandlerManager: { handlers: Map Promise> }; + }).rpcHandlerManager.handlers; + + session.queue.push('soft steer', mode, 'steer'); + await expect(handlers.get(RPC_METHODS.SteerQueuedMessage)!({ localId: 'steer' })) + .resolves.toEqual({ steered: false, error: 'Steer outcome is being reconciled' }); + expect(session.client.emitSteerIndeterminate).toHaveBeenCalledWith(['steer']); + expect(session.queue.cancelByLocalId('steer')).toBe(true); + + harness.deferPrompt = null; + releasePrompt(); + session.queue.close(); + await runPromise; + }); + + it('prevents cancellation once ACP steer dispatch starts', async () => { + let releasePrompt!: () => void; + let releaseDispatch!: () => void; + harness.deferPrompt = new Promise((resolve) => { releasePrompt = resolve; }); + harness.deferSoftSteerDispatch = new Promise((resolve) => { releaseDispatch = resolve; }); + const session = makeSession(null, false); + const mode = { permissionMode: 'default' } as EnhancedMode; + session.queue.push('first', mode, 'first'); + + const runPromise = cursorAcpRemoteLauncher(session); + await vi.waitFor(() => expect(harness.promptCalls).toBe(1)); + const handlers = (session.client as unknown as { + rpcHandlerManager: { handlers: Map Promise> }; + }).rpcHandlerManager.handlers; + + session.queue.push('soft steer', mode, 'steer'); + const steerResult = handlers.get(RPC_METHODS.SteerQueuedMessage)!({ localId: 'steer' }); + await Promise.resolve(); + expect(session.queue.cancelByLocalId('steer')).toBe('in-flight'); + releaseDispatch(); + await expect(steerResult).resolves.toEqual({ steered: true }); + expect(session.client.emitMessagesConsumed).toHaveBeenCalledWith(['steer'], { steered: true }); + + harness.deferPrompt = null; + releasePrompt(); + session.queue.close(); + await runPromise; + }); + + it('blocks the next prompt while soft-steer dispatch is pending', async () => { + let releasePrompt!: () => void; + let releaseDispatch!: () => void; + let releaseSoftSteer!: () => void; + harness.deferPrompt = new Promise((resolve) => { releasePrompt = resolve; }); + harness.deferSoftSteerDispatch = new Promise((resolve) => { releaseDispatch = resolve; }); + harness.deferSoftSteer = new Promise((resolve) => { releaseSoftSteer = resolve; }); + const session = makeSession(null, false); + const mode = { permissionMode: 'default' } as EnhancedMode; + session.queue.push('first', mode, 'first'); + + const runPromise = cursorAcpRemoteLauncher(session); + await vi.waitFor(() => expect(harness.promptCalls).toBe(1)); + const handlers = (session.client as unknown as { + rpcHandlerManager: { handlers: Map Promise> }; + }).rpcHandlerManager.handlers; + + session.queue.push('soft steer', mode, 'steer'); + const steerResult = handlers.get(RPC_METHODS.SteerQueuedMessage)!({ localId: 'steer' }); + await Promise.resolve(); + session.queue.push('next', mode, 'next'); + harness.deferPrompt = null; + releasePrompt(); + await Promise.resolve(); + expect(harness.promptCalls).toBe(1); + + releaseDispatch(); + await expect(steerResult).resolves.toEqual({ steered: true }); + expect(harness.promptCalls).toBe(1); + releaseSoftSteer(); + await vi.waitFor(() => expect(harness.promptCalls).toBe(2)); + session.queue.close(); + await runPromise; + }); + + it('acknowledges a steer dispatched before an overlapping abort', async () => { + let releasePrompt!: () => void; + let releaseDispatch!: () => void; + harness.deferPrompt = new Promise((resolve) => { releasePrompt = resolve; }); + harness.deferSoftSteerDispatch = new Promise((resolve) => { releaseDispatch = resolve; }); + const session = makeSession(null, false); + const mode = { permissionMode: 'default' } as EnhancedMode; + session.queue.push('first', mode, 'first'); + + const runPromise = cursorAcpRemoteLauncher(session); + await vi.waitFor(() => expect(harness.promptCalls).toBe(1)); + const handlers = (session.client as unknown as { + rpcHandlerManager: { handlers: Map Promise> }; + }).rpcHandlerManager.handlers; + + session.queue.push('soft steer', mode, 'steer'); + const steerResult = handlers.get(RPC_METHODS.SteerQueuedMessage)!({ localId: 'steer' }); + await Promise.resolve(); + releaseDispatch(); + await expect(steerResult).resolves.toEqual({ steered: true }); + await handlers.get(RPC_METHODS.Abort)!(); + + await vi.waitFor(() => expect(session.client.emitMessagesConsumed).toHaveBeenCalledWith(['steer'], { steered: true })); + expect(vi.mocked(session.client.emitMessagesConsumed).mock.calls + .filter(([ids]) => ids.includes('steer'))).toHaveLength(1); + + harness.deferPrompt = null; + releasePrompt(); + session.queue.close(); + await runPromise; + }); + + it('does not hang teardown while a soft steer completion is unresolved', async () => { + let releasePrompt!: () => void; + harness.deferPrompt = new Promise((resolve) => { releasePrompt = resolve; }); + // Completion never resolves — simulates Cursor keeping the concurrent + // request open past Exit/Switch. + harness.deferSoftSteer = new Promise(() => {}); + const session = makeSession(null, false); + const mode = { permissionMode: 'default' } as EnhancedMode; + session.queue.push('first', mode, 'first'); + + const runPromise = cursorAcpRemoteLauncher(session); + await vi.waitFor(() => expect(harness.promptCalls).toBe(1)); + const handlers = (session.client as unknown as { + rpcHandlerManager: { handlers: Map Promise> }; + }).rpcHandlerManager.handlers; + + session.queue.push('soft steer', mode, 'steer'); + await expect(handlers.get(RPC_METHODS.SteerQueuedMessage)!({ localId: 'steer' })) + .resolves.toEqual({ steered: true }); + + // Exit/Switch must reach cleanup (which disconnects the transport and + // rejects pending ACP requests) without waiting on the soft steer. + harness.deferPrompt = null; + releasePrompt(); + await handlers.get(RPC_METHODS.Switch)!(); + await vi.waitFor(() => expect(runPromise).resolves.toBeDefined()); + }); + it('spawns agent acp backend, not stream-json', async () => { const session = makeSession(null); await cursorAcpRemoteLauncher(session); diff --git a/cli/src/cursor/cursorAcpRemoteLauncher.ts b/cli/src/cursor/cursorAcpRemoteLauncher.ts index cfa3539d..5130a17c 100644 --- a/cli/src/cursor/cursorAcpRemoteLauncher.ts +++ b/cli/src/cursor/cursorAcpRemoteLauncher.ts @@ -34,7 +34,9 @@ import { buildCursorModelsSeedPayload, seedCursorModelsCache } from '@/modules/c import { readSharedCursorModelsCache } from '@/modules/common/cursorModelsSharedCache'; import type { AcpSdkBackend } from '@/agent/backends/acp'; import type { AcpStderrError } from '@/agent/backends/acp/AcpStdioTransport'; +import { isAcpIndeterminateError } from '@/agent/backends/acp/AcpStdioTransport'; import { registerAcpSessionTitleSync } from '@/agent/acpSessionTitle'; +import { RPC_METHODS } from '@hapi/protocol/rpcMethods'; import { cursorHapiMcpServerId, installCursorMcpOverlay, @@ -50,6 +52,8 @@ import { stripRetryableCursorError } from './cursorAutoRetry'; +const CURSOR_ABORT_DRAIN_TIMEOUT_MS = 5_000; + class CursorAcpRemoteLauncher extends RemoteLauncherBase { private readonly session: CursorSession; private backend: ReturnType | null = null; @@ -63,6 +67,11 @@ class CursorAcpRemoteLauncher extends RemoteLauncherBase { private defaultBackendModel: string | null = null; private unregisterModelApplyHandler: (() => void) | null = null; private modelApplySeq = 0; + private activePromptModeHash: string | null = null; + /** True while a backend.prompt turn is in flight. */ + private promptInFlight = false; + /** Concurrent soft-steer session/prompt RPCs still running after kickoff. */ + private softSteerWaiters: Promise[] = []; /** True when ACP process was spawned with `--auto-review`. */ private spawnedWithAutoReview = false; /** Avoid re-queueing `/auto-review` on every mid-session mode sync. */ @@ -72,7 +81,6 @@ class CursorAcpRemoteLauncher extends RemoteLauncherBase { private pendingRetryableFromStderr = false; private pendingInlineRetryableError = false; private attemptProducedToolActivity = false; - private promptInFlight = false; private userAbortRequested = false; constructor(session: CursorSession) { @@ -94,6 +102,7 @@ class CursorAcpRemoteLauncher extends RemoteLauncherBase { protected async runMainLoop(): Promise { const session = this.session; const messageBuffer = this.messageBuffer; + session.client.updateAgentState?.((state) => ({ ...state, steeringActive: false })); const { server: happyServer, mcpServers } = await buildHapiMcpBridge(session.client, { enableChangeTitle: false, @@ -414,13 +423,165 @@ class CursorAcpRemoteLauncher extends RemoteLauncherBase { onSwitch: () => this.handleSwitchRequest() }); + // Soft steer = Cursor GUI "Send" (next-opportune / soft inject): fire a + // concurrent session/prompt without canceling the in-flight turn. Abort + // remains the hard stop path (GUI "Stop & send"). + session.client.rpcHandlerManager.registerHandler( + RPC_METHODS.SteerQueuedMessage, + async (payload: unknown) => { + const localId = typeof (payload as { localId?: unknown } | null)?.localId === 'string' + ? (payload as { localId: string }).localId + : ''; + if (!localId) { + return { steered: false, error: 'Missing localId' }; + } + const backend = this.backend; + const acpSessionId = this.acpSessionId; + if (!this.promptInFlight || !acpSessionId || !backend) { + return { steered: false, error: 'No active steerable turn' }; + } + const targetPromptGeneration = backend.getPromptGeneration(); + const taken = session.queue.takeByLocalId(localId); + if (!taken) { + return { steered: false, error: 'Message not in queue' }; + } + const isControlCommand = Boolean(taken.item.isolate) + || parseCursorSpecialCommand(taken.item.message).type !== null; + if (isControlCommand) { + session.queue.restoreReservation(taken); + return { steered: false, error: 'Control commands cannot be steered' }; + } + if (this.activePromptModeHash !== taken.item.modeHash) { + session.queue.restoreReservation(taken); + return { steered: false, error: 'Queued message mode differs from the active turn' }; + } + + // Ack the hub once the soft-steer request is kicked off — not when + // the concurrent session/prompt finishes. ACP treats that response as + // turn completion, which can exceed the hub's 30s Socket.IO RPC timeout + // and report a false failure after the inject already started. + // Keep the launcher busy until that background prompt settles so we + // do not emit ready / start the next backend.prompt() while it runs. + if (!session.queue.beginReservationDispatch(taken)) { + return { steered: false, error: 'Steer cancelled' }; + } + const dispatchStatePersisted = await session.client.setSteerDeliveryState([localId], 'dispatching'); + if (!dispatchStatePersisted) { + session.queue.markReservationIndeterminate(taken); + session.client.emitSteerIndeterminate([localId]); + return { steered: false, error: 'Steer state is indeterminate' }; + } + const restoreQueuedReservation = async (): Promise => { + if (!taken.originIndeterminate) { + const persisted = await session.client.setSteerDeliveryState([localId], 'queued'); + if (!persisted) { + session.queue.markReservationIndeterminate(taken); + session.client.emitSteerIndeterminate([localId]); + return false; + } + } + if (taken.state !== 'dispatching' || !session.queue.restoreReservation(taken)) { + session.client.emitSteerIndeterminate([localId]); + return false; + } + return true; + }; + if (taken.state !== 'dispatching') { + session.client.emitSteerIndeterminate([localId]); + return { steered: false, error: 'Steer cancelled' }; + } + if (!this.promptInFlight + || this.backend !== backend + || this.acpSessionId !== acpSessionId + || backend.getPromptGeneration() !== targetPromptGeneration) { + await restoreQueuedReservation(); + return { steered: false, error: 'Active turn changed' }; + } + let steer: { dispatched: Promise; completed: Promise }; + try { + steer = backend.beginSoftSteerPrompt(acpSessionId, [{ + type: 'text', + text: taken.item.message + }]); + } catch (error) { + if (isAcpIndeterminateError(error)) { + if (session.queue.markReservationIndeterminate(taken)) { + session.client.emitSteerIndeterminate([localId]); + } + logger.debug('[cursor-acp] soft-steer dispatch outcome unknown', error); + return { steered: false, error: 'Steer outcome is being reconciled' }; + } + logger.debug('[cursor-acp] soft-steer failed to start', error); + await restoreQueuedReservation(); + return { steered: false, error: 'Failed to soft-steer into active turn' }; + } + // Completion still gates the next prompt (handler swap safety); + // register the waiter before awaiting dispatch so the main loop's + // finally cannot slip a prompt in between. + const steerDone = Promise.all([steer.dispatched, steer.completed]).then(() => {}, (error) => { + logger.debug('[cursor-acp] soft-steer completion failed after dispatch', error); + }); + this.softSteerWaiters.push(steerDone); + const removeWaiter = () => { + this.softSteerWaiters = this.softSteerWaiters.filter((p) => p !== steerDone); + }; + void steerDone.then(removeWaiter); + try { + await steer.dispatched; + } catch (error) { + if (isAcpIndeterminateError(error)) { + if (session.queue.markReservationIndeterminate(taken)) { + session.client.emitSteerIndeterminate([localId]); + } + logger.debug('[cursor-acp] soft-steer dispatch outcome unknown', error); + return { steered: false, error: 'Steer outcome is being reconciled' }; + } + await restoreQueuedReservation(); + logger.debug('[cursor-acp] soft-steer failed to start', error); + return { steered: false, error: 'Failed to soft-steer into active turn' }; + } + // The RPC acks once stdin accepted the inject. The queue row is + // committed only when the concurrent prompt settles: an explicit + // JSON-RPC rejection means ACP never accepted the instruction + // (restore it for the next prompt), while a transport failure + // (abort/disconnect) keeps the row reserved — never re-delivered. + void steer.completed.then(() => { + // Completion means ACP accepted the inject: the ACK must + // reach the hub even when an abort reset the queue and + // cancelled the reservation in between. + session.queue.commitReservation(taken); + messageBuffer.addMessage(taken.item.message, 'user'); + session.client.emitMessagesConsumed([localId], { steered: true }); + }, (error) => { + if (isAcpIndeterminateError(error)) { + // Do not leave the reservation dispatching forever. Hold + // it outside automatic replay and persist the ambiguous + // outcome; a later explicit Steer retries this same row. + if (session.queue.markReservationIndeterminate(taken)) { + session.client.emitSteerIndeterminate([localId]); + } + logger.debug('[cursor-acp] soft-steer outcome unknown after dispatch; row held for explicit resolution', error); + return; + } + void restoreQueuedReservation().then((restored) => { + if (restored) { + logger.debug('[cursor-acp] soft-steer rejected by ACP; row restored', error); + } + }); + }); + return { steered: true }; + } + ); + const sendReady = () => { session.sendSessionEvent({ type: 'ready' }); }; + try { while (!this.shouldExit) { const waitSignal = this.abortController.signal; const batch = await session.queue.waitForMessagesAndGetAsString(waitSignal); + if (!batch) { if (waitSignal.aborted && !this.shouldExit) { continue; @@ -463,6 +624,9 @@ class CursorAcpRemoteLauncher extends RemoteLauncherBase { }]; session.onThinkingChange(true); + this.promptInFlight = true; + session.client.updateAgentState?.((state) => ({ ...state, steeringActive: true })); + this.activePromptModeHash = batch.hash; try { this.promptInFlight = true; @@ -508,6 +672,31 @@ class CursorAcpRemoteLauncher extends RemoteLauncherBase { } } finally { this.promptInFlight = false; + session.client.updateAgentState?.((state) => ({ ...state, steeringActive: false })); + // Soft-steers share the ACP session; wait for them before ready / + // the next prompt so message handlers are not swapped mid-inject. + // An Abort (which clears the waiters) must release this wait too: + // race the settle against the abort signal so the launcher never + // blocks on a soft steer whose completion is unbounded. + if (this.softSteerWaiters.length > 0 && !this.shouldExit) { + const waitSignal = this.abortController.signal; + let releaseWait!: () => void; + const abortListener = () => releaseWait(); + if (!waitSignal.aborted) { + waitSignal.addEventListener('abort', abortListener, { once: true }); + } + try { + await Promise.race([ + Promise.allSettled([...this.softSteerWaiters]), + new Promise((resolve) => { releaseWait = resolve; }) + ]); + } finally { + // Repeated waits must not accumulate abort listeners. + waitSignal.removeEventListener('abort', abortListener); + } + this.softSteerWaiters = []; + } + this.activePromptModeHash = null; this.pendingRetryableError = null; this.pendingRetryableFromStderr = false; this.pendingInlineRetryableError = false; @@ -520,6 +709,11 @@ class CursorAcpRemoteLauncher extends RemoteLauncherBase { } } } + } finally { + // No wait here: Exit/Switch must reach cleanup() promptly; it + // disconnects the ACP transport, rejecting pending soft-steer + // requests and settling any waiters. + } } protected async cleanup(): Promise { @@ -530,6 +724,13 @@ class CursorAcpRemoteLauncher extends RemoteLauncherBase { try { this.clearAbortHandlers(this.session.client.rpcHandlerManager); + this.session.client.rpcHandlerManager.registerHandler(RPC_METHODS.SteerQueuedMessage, async () => ({ + steered: false, + error: 'Session ending' + })); + this.promptInFlight = false; + this.session.client.updateAgentState?.((state) => ({ ...state, steeringActive: false })); + this.softSteerWaiters = []; this.unregisterModelApplyHandler?.(); this.unregisterModelApplyHandler = null; @@ -887,13 +1088,47 @@ class CursorAcpRemoteLauncher extends RemoteLauncherBase { private async handleAbort(): Promise { this.userAbortRequested = true; const backend = this.backend; - const sessionId = this.session.sessionId; + const sessionId = this.acpSessionId ?? this.session.sessionId; if (backend && sessionId) { + const pendingSoftSteers = [...this.softSteerWaiters]; await backend.cancelPrompt(sessionId); + // Drop soft-steer bookkeeping first; retain the foreground prompt + // count and wait for both boundaries before a new handler is used. + backend.abortSoftSteers(); + if (!this.shouldExit) { + let timeout: ReturnType | null = null; + const drained = await Promise.race([ + Promise.all([ + backend.waitForResponseComplete(), + Promise.allSettled(pendingSoftSteers) + ]).then(() => true), + new Promise((resolve) => { + timeout = setTimeout(() => resolve(false), CURSOR_ABORT_DRAIN_TIMEOUT_MS); + timeout.unref?.(); + }) + ]); + if (timeout) clearTimeout(timeout); + if (!drained) { + // An ACP request that ignores cancel cannot safely share a + // handler with the next prompt. End the launcher instead + // of allowing late updates to cross the turn boundary. + logger.warn('[cursor-acp] abort drain timed out; ending session to isolate late ACP updates'); + this.shouldExit = true; + } + } } await this.permissionAdapter?.cancelAll('User aborted'); await this.extensionAdapter?.cancelAll('User aborted'); - this.session.queue.reset(); + // A soft steer may settle after Abort; preserve its reservation until + // the completion callback records accepted or indeterminate. + this.session.queue.reset({ preserveDispatchingReservations: true }); + this.promptInFlight = false; + // Abort is the hard-stop path: drop soft-steer waiters so the prompt + // finally cannot block the next prompt on a soft steer whose completion + // is unbounded and may never settle. Soft counters were already reset + // above; only the foreground prompt was drained before continuing. + this.softSteerWaiters = []; + this.session.client.updateAgentState?.((state) => ({ ...state, steeringActive: false })); this.session.onThinkingChange(false); this.abortController.abort(); this.abortController = new AbortController(); diff --git a/cli/src/cursor/cursorLegacyRemoteLauncher.ts b/cli/src/cursor/cursorLegacyRemoteLauncher.ts index 3d4437f1..a273423f 100644 --- a/cli/src/cursor/cursorLegacyRemoteLauncher.ts +++ b/cli/src/cursor/cursorLegacyRemoteLauncher.ts @@ -12,6 +12,7 @@ import { } from '@/modules/common/remote/RemoteLauncherBase'; import type { CursorSession } from './session'; import type { EnhancedMode } from './loop'; +import { RPC_METHODS } from '@hapi/protocol/rpcMethods'; // TODO(cursor-acp): remove legacy stream-json resume path after migration window. // New Cursor sessions use ACP only. This path exists because pre-ACP Cursor // session_id values are not loadable via ACP session/load. @@ -138,6 +139,14 @@ class CursorRemoteLauncher extends RemoteLauncherBase { onSwitch: () => this.handleSwitchRequest() }); + session.client.rpcHandlerManager.registerHandler( + RPC_METHODS.SteerQueuedMessage, + async () => ({ + steered: false, + error: 'Mid-turn steering requires a Cursor ACP session' + }) + ); + const sendReady = () => { session.sendSessionEvent({ type: 'ready' }); }; diff --git a/cli/src/cursor/runCursor.test.ts b/cli/src/cursor/runCursor.test.ts index d4580887..227b7028 100644 --- a/cli/src/cursor/runCursor.test.ts +++ b/cli/src/cursor/runCursor.test.ts @@ -16,6 +16,7 @@ const harness = vi.hoisted(() => ({ session: { onUserMessage: vi.fn(), onCancelQueuedMessage: vi.fn(), + onRetryQueuedMessage: vi.fn(), sendSessionEvent: vi.fn(), rpcHandlerManager: { registerHandler: vi.fn() diff --git a/cli/src/cursor/runCursor.ts b/cli/src/cursor/runCursor.ts index fec43e80..0cf46367 100644 --- a/cli/src/cursor/runCursor.ts +++ b/cli/src/cursor/runCursor.ts @@ -112,6 +112,7 @@ export async function runCursor(opts: { logger.debug(`[cursor] cancelByLocalId(${localId}): ${removed ? 'removed' : 'not found (best-effort)'}`); return removed; }); + session.onRetryQueuedMessage((localId) => messageQueue.releaseIndeterminateReservation(localId)); session.rpcHandlerManager.registerHandler(RPC_METHODS.SetSessionConfig, async (payload: unknown) => { if (!payload || typeof payload !== 'object') { diff --git a/hub/src/sync/steerQueuedMessage.test.ts b/hub/src/sync/steerQueuedMessage.test.ts index 8515054a..1704a959 100644 --- a/hub/src/sync/steerQueuedMessage.test.ts +++ b/hub/src/sync/steerQueuedMessage.test.ts @@ -63,7 +63,7 @@ describe('SyncEngine.steerQueuedMessage', () => { expect(result).toEqual({ status: 'failed', - error: 'Steering is only supported for Pi and Codex sessions', + error: 'Steering is only supported for Pi, Codex, and Cursor ACP sessions', localId: null }) } finally { diff --git a/hub/src/sync/syncEngine.ts b/hub/src/sync/syncEngine.ts index 28212f2b..049398f4 100644 --- a/hub/src/sync/syncEngine.ts +++ b/hub/src/sync/syncEngine.ts @@ -1035,9 +1035,11 @@ export class SyncEngine { } /** - * Ask the CLI to deliver one waiting-queue message into the active Pi turn - * (Pi native steer). Only pi sessions support this today; the CLI's - * `steer-queued-message` handler is registered by the pi runner alone. + * Ask the CLI to deliver one waiting-queue message into the active turn + * (native steer). Supported for Pi, Codex, and Cursor ACP sessions; the + * CLI's `steer-queued-message` handler is registered per flavor. Legacy + * stream-json Cursor sessions and other flavors are rejected by the + * capability gate. */ async steerQueuedMessage( sessionId: string, @@ -1048,7 +1050,7 @@ export class SyncEngine { return { status: 'failed', error: 'Session not found', localId: null } } if (!isSteeringSupportedForSession(session.metadata)) { - return { status: 'failed', error: 'Steering is only supported for Pi and Codex sessions', localId: null } + return { status: 'failed', error: 'Steering is only supported for Pi, Codex, and Cursor ACP sessions', localId: null } } if (session.agentState?.controlledByUser === true) { return { status: 'failed', error: 'Steering is only available for remote sessions', localId: null } diff --git a/shared/src/modes.test.ts b/shared/src/modes.test.ts index 0e60af3b..69ab759f 100644 --- a/shared/src/modes.test.ts +++ b/shared/src/modes.test.ts @@ -133,10 +133,10 @@ describe('claude auto permission mode', () => { }) describe('isSteeringSupportedForFlavor', () => { - it('supports codex and pi only', () => { + it('supports codex, cursor and pi', () => { expect(isSteeringSupportedForFlavor('codex')).toBe(true) + expect(isSteeringSupportedForFlavor('cursor')).toBe(true) expect(isSteeringSupportedForFlavor('pi')).toBe(true) - expect(isSteeringSupportedForFlavor('cursor')).toBe(false) expect(isSteeringSupportedForFlavor('claude')).toBe(false) expect(isSteeringSupportedForFlavor('opencode')).toBe(false) expect(isSteeringSupportedForFlavor(undefined)).toBe(false) @@ -150,13 +150,25 @@ describe('isSteeringSupportedForSession', () => { expect(isSteeringSupportedForSession({ flavor: 'pi' })).toBe(true) }) - it('rejects cursor until its steer handler lands', () => { + it('supports Cursor ACP sessions', () => { expect(isSteeringSupportedForSession({ flavor: 'cursor', cursorSessionProtocol: 'acp', cursorSessionId: 'sess-1', + })).toBe(true) + expect(isSteeringSupportedForSession({ flavor: 'cursor' })).toBe(true) + }) + + it('rejects legacy Cursor stream-json sessions', () => { + expect(isSteeringSupportedForSession({ + flavor: 'cursor', + cursorSessionProtocol: 'stream-json', + cursorSessionId: 'legacy-1', + })).toBe(false) + expect(isSteeringSupportedForSession({ + flavor: 'cursor', + cursorSessionId: 'legacy-without-protocol', })).toBe(false) - expect(isSteeringSupportedForSession({ flavor: 'cursor' })).toBe(false) }) it('rejects non-steerable flavors', () => { diff --git a/shared/src/modes.ts b/shared/src/modes.ts index 8f18e63f..73162e49 100644 --- a/shared/src/modes.ts +++ b/shared/src/modes.ts @@ -190,12 +190,11 @@ export function getCodexCollaborationModeOptions(): CodexCollaborationModeOption * - Codex: app-server `turn/steer` (true mid-turn inject) * - Cursor ACP: concurrent `session/prompt` soft-send (no cancel). Legacy * stream-json Cursor sessions are NOT steerable — gate with - * {@link isSteeringSupportedForSession}. (Cursor joins this list once its - * soft-steer handler lands.) + * {@link isSteeringSupportedForSession}. * * Claude / others: not supported (no reachable soft-steer path) — UI hides Steer. */ -export const STEERING_SUPPORTED_FLAVORS = ['codex', 'pi'] as const +export const STEERING_SUPPORTED_FLAVORS = ['codex', 'cursor', 'pi'] as const export function isSteeringSupportedForFlavor(flavor?: string | null): boolean { return (STEERING_SUPPORTED_FLAVORS as readonly string[]).includes(flavor ?? '') @@ -217,5 +216,14 @@ export function isSteeringSupportedForSession(metadata?: { if (metadata?.flavor === 'codex' || metadata?.flavor === 'pi') { return true } - return false + if (metadata?.flavor !== 'cursor') { + return false + } + if (metadata.cursorSessionProtocol === 'stream-json') { + return false + } + if (!metadata.cursorSessionProtocol && metadata.cursorSessionId) { + return false + } + return true } diff --git a/web/src/components/SessionChat.tsx b/web/src/components/SessionChat.tsx index 425ab6e5..d57f18e7 100644 --- a/web/src/components/SessionChat.tsx +++ b/web/src/components/SessionChat.tsx @@ -1866,7 +1866,9 @@ function SessionChatInner(props: SessionChatProps) { updatePendingSchedule(restored) }} canSteer={isSteeringSupportedForSession(props.session.metadata) - && props.session.thinking + && (agentFlavor === 'pi' + ? props.session.thinking + : props.session.agentState?.steeringActive === true) && !controlledByUser} />