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} />