mirror of
https://github.com/wu736139669/hapi.git
synced 2026-10-25 22:28:04 +00:00
* 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
This commit is contained in:
SSU-WEI HUANG
@@ -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<typeof createCursorAcpBackend> | 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<void>[] = [];
|
||||
/** 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<void> {
|
||||
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<boolean> => {
|
||||
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<void>; completed: Promise<void> };
|
||||
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<void>((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<void> {
|
||||
@@ -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<void> {
|
||||
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<typeof setTimeout> | null = null;
|
||||
const drained = await Promise.race([
|
||||
Promise.all([
|
||||
backend.waitForResponseComplete(),
|
||||
Promise.allSettled(pendingSoftSteers)
|
||||
]).then(() => true),
|
||||
new Promise<boolean>((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();
|
||||
|
||||
Reference in New Issue
Block a user