diff --git a/packages/coding-agent/CHANGELOG.md b/packages/coding-agent/CHANGELOG.md index 0ebe5e722c..f877cfbd3d 100644 --- a/packages/coding-agent/CHANGELOG.md +++ b/packages/coding-agent/CHANGELOG.md @@ -2,6 +2,7 @@ ## [Unreleased] - Conventional MCP autoload now reads the user scope from the agent directory instead of a home-relative `/.gjc/agent` path (#4767). Every writer and denylist reader already resolved user scope through `getMCPConfigPath("user")` (the agent directory), while native discovery derived it from the load context's home, so the two disagreed the moment an agent-directory profile was in play: `GJC_CODING_AGENT_DIR= gjc mcp add ` wrote `/mcp.json` and reported the server as loaded by ordinary sessions at startup, but startup read `~/.gjc/agent/mcp.json` — the profile's own registrations never loaded and the default profile's servers loaded into the profile instead. `loadAllMCPConfigs` accepts an `agentDir`, sessions created with their own `agentDir` pass it, and the `disabledServers` denylist follows the same scope, so discovery, `gjc mcp add`, the `/mcp` wizard, and `gjc customize doctor` all name one file. This also restores isolation for the MCP autoload suites, which established their temp user scope with `setAgentDir()`: after the trusted-home provenance rework their `os.homedir()` mock no longer reached discovery, so on a developer machine the red-team suite read the real `~/.gjc/agent/mcp.json` and in CI it found nothing. +- Immediate `turn.prompt` after `turn.abort` acknowledgement now starts exactly one successor turn instead of being silently consumed by aborted-turn teardown. Abort unwind no longer classifies the successor as steering into the dying loop, delayed `agent_end` is paired with the aborted invocation rather than the successor, and abort acknowledgement waits for that serialized terminal transition — no client-side sleep (#4749). Retention is bounded in the other direction too: abort stays terminal across overlapping aborts. Every abort request advances an abort-admission epoch synchronously, including one that shares an already in-flight unwind, so a prompt admitted between two aborts is refused as a cancelled preflight instead of refreshing its generation and starting work the user already aborted twice. An abort sharing another abort's unwind also applies its own request-scoped effects (preflight cancellation, and abort visibility so a later real abort is not silenced by an earlier `silent: true` abort). A prompt admitted after an abort settles still starts a normal successor turn. - Esc/Ctrl+C now recover a WSL/basic-terminal session whose busy indicator outlived its turn (#4741). Both the global interrupt/clear listener and the editor escape handler treated any mounted working loader as unconditionally cancellable work: after a completed turn left the loader mounted, every press ran a no-op abort, consumed the key, and reset the escape gestures, so the composer stayed `working [esc]`/`Fetching … [esc]` and Esc, Ctrl+C, and Ctrl+X never did anything (Ctrl+Z suspend/resume was the only escape). The loading branch now stops the stale indicator via the shared activity-indicator stop and lets the key fall through to idle semantics when there is no pending submission (including a started one still inside prompt preflight, exposed through the new `hasPendingSubmission()` context query), no queued steering/follow-up/compaction messages, and the session is neither streaming nor compacting; active work (streaming, queued messages, pending optimistic or started-preflight submission) still aborts exactly as before. Recovery gates on drainable queues only: the new `AgentSession.drainableQueuedMessageCount` counts exactly the steering and follow-up entries that `clearQueue()`/`popLastQueuedMessage()`/`getQueuedMessageEntries()` return, whereas the aggregate `queuedMessageCount` also counts hidden next-turn context that no key press can drain and that deliberately survives turn completion (a `todo_write` failure reminder queued with `deliverAs: "nextTurn"` and no `triggerTurn`) — gating on the aggregate left a permanently nonzero count that reproduced the same lockout while idle. Hidden next-turn ordering and delivery are unchanged; recovery neither clears nor delivers those entries. The same correction applies to the two adjacent gates whose handlers are also visible-queue-only: `app.message.sendNow` no longer advertises itself when only hidden context is queued (its only outcome was "No visible queued message to send"), and an empty submit while streaming no longer aborts the live turn to flush a queue that holds nothing drainable. - Bare-default Codex and Anthropic provider-overload retries now honor the configured retry ceiling instead of entering the unbounded transient path, and every replay still requires a clean retry scope after extension lifecycle handlers participate. - `gjc accounts` command errors no longer escape as uncaught exceptions in text mode. `accounts pin` resolves its target through `resolveOAuthPinTarget`, which throws a typed `OAuthCredentialSelectorError` for user-correctable selector problems (API-key rows, active overrides, disabled or missing accounts, ambiguity), and `runAccountsCommand` rendered `AccountsCommandError` only in `--json` mode — text mode rethrew everything, so even the command's own "Provider … is not configured; no pin was written" surfaced as a stack trace plus a `gjc-crash.log` entry. Selector failures now map to `AccountsCommandError` with the message preserved (so `--json` reports `accounts-error` instead of `internal-error`), and text mode prints one clean stderr line with exit code 1. The framework's `CliParseError` handling and the JSON machine contract (exactly one document, never stacks or secrets) are unchanged. diff --git a/packages/coding-agent/src/sdk/host/session-runtime.test.ts b/packages/coding-agent/src/sdk/host/session-runtime.test.ts index ef0bcfb307..da7f09e5d3 100644 --- a/packages/coding-agent/src/sdk/host/session-runtime.test.ts +++ b/packages/coding-agent/src/sdk/host/session-runtime.test.ts @@ -2534,6 +2534,96 @@ describe("post-acceptance invocation terminalization", () => { await rm(cwd, { recursive: true, force: true }); } }); + test("immediate prompt after abort ack is not terminalized by the aborted turn's delayed agent_end", async () => { + const cwd = await mkdtemp(path.join(os.tmpdir(), "gjc-abort-immediate-prompt-")); + try { + const firstInflight = Promise.withResolvers(); + const successorStarted = Promise.withResolvers(); + let prompts = 0; + const harness = await invocationHarness("abort-immediate-prompt", cwd, { + sendUserMessage: async (_content, options) => { + prompts += 1; + await options?.onPreflightAcceptCommit?.(); + if (prompts === 1) { + await firstInflight.promise; + return; + } + successorStarted.resolve(); + await Promise.withResolvers().promise; + }, + }); + const first = await harness.control("turn.prompt", { text: "first" }); + expect(first.ok).toBe(true); + const firstIds = { commandId: first.result?.commandId, turnId: first.result?.turnId }; + await harness.emit("agent_start"); + expect(await harness.control("turn.abort", {})).toMatchObject({ ok: true }); + firstInflight.reject(Object.assign(new Error("turn aborted"), { code: "aborted" })); + const second = await harness.control("turn.prompt", { text: "successor" }); + expect(second.ok).toBe(true); + const secondIds = { commandId: second.result?.commandId, turnId: second.result?.turnId }; + expect(secondIds.commandId).not.toBe(firstIds.commandId); + expect(secondIds.turnId).not.toBe(firstIds.turnId); + await harness.emit("agent_start"); + await successorStarted.promise; + await harness.emit("agent_end"); + expect(await harness.query("turn.prompt_status", secondIds)).toMatchObject({ + result: { status: expect.stringMatching(/accepted|in_flight/) }, + }); + expect(await settledStatus(harness, "turn.prompt_status", firstIds)).toMatchObject({ + status: "failed", + error: { code: "aborted" }, + }); + await harness.emit("agent_end"); + expect(await settledStatus(harness, "turn.prompt_status", secondIds)).toMatchObject({ + status: "terminal_ok", + }); + expect(prompts).toBe(2); + await harness.stop(); + } finally { + await Bun.sleep(50); + await rm(cwd, { recursive: true, force: true }); + } + }); + + test("abort_and_prompt starts exactly one successor and is not duplicated by delayed abort teardown", async () => { + const cwd = await mkdtemp(path.join(os.tmpdir(), "gjc-abort-and-prompt-once-")); + try { + const firstInflight = Promise.withResolvers(); + const abortReleased = Promise.withResolvers(); + let prompts = 0; + const harness = await invocationHarness("abort-and-prompt-once", cwd, { + sendUserMessage: async (_content, options) => { + prompts += 1; + await options?.onPreflightAcceptCommit?.(); + if (prompts === 1) { + await firstInflight.promise; + return; + } + }, + abort: () => abortReleased.promise, + }); + const first = await harness.control("turn.prompt", { text: "first" }); + expect(first.ok).toBe(true); + await harness.emit("agent_start"); + const replacement = harness.control("turn.abort_and_prompt", { text: "replacement" }); + firstInflight.reject(Object.assign(new Error("turn aborted"), { code: "aborted" })); + await harness.emit("agent_end"); + abortReleased.resolve(); + const accepted = await replacement; + expect(accepted.ok).toBe(true); + const successorIds = { commandId: accepted.result?.commandId, turnId: accepted.result?.turnId }; + await harness.emit("agent_start"); + await harness.emit("agent_end"); + expect(await settledStatus(harness, "turn.prompt_status", successorIds)).toMatchObject({ + status: "terminal_ok", + }); + expect(prompts).toBe(2); + await harness.stop(); + } finally { + await Bun.sleep(50); + await rm(cwd, { recursive: true, force: true }); + } + }); test("a failed skill invocation still reports a terminal failed status", async () => { const cwd = await mkdtemp(path.join(os.tmpdir(), "gjc-terminalize-skill-")); diff --git a/packages/coding-agent/src/sdk/host/session-runtime.ts b/packages/coding-agent/src/sdk/host/session-runtime.ts index 60f66456ec..3dac331157 100644 --- a/packages/coding-agent/src/sdk/host/session-runtime.ts +++ b/packages/coding-agent/src/sdk/host/session-runtime.ts @@ -2351,8 +2351,8 @@ function createControlSurface( // Follow-ups never start inline; ownership correlates at promotion. true, ), - abort: () => { - ctx.abort(); + abort: async () => { + await Promise.resolve(ctx.abort()).catch(() => undefined); return { aborted: true }; }, abortTerminal: terminalAbort, @@ -2573,6 +2573,13 @@ export function createSdkSessionRuntimeExtension(api: ExtensionAPI, options: Cre waitForGateResolutionQuiescence: () => Promise; activeInvocation?: { kind: InvocationKind; correlation: InvocationCorrelation }; drainedInvocations?: Array<{ kind: InvocationKind; correlation: InvocationCorrelation }>; + openLifecycleBatches: Array<{ + invocations: Array<{ + kind: InvocationKind; + correlation: InvocationCorrelation; + connectionId: string | undefined; + }>; + }>; disposeGate?: () => void; } | undefined; @@ -2594,6 +2601,28 @@ export function createSdkSessionRuntimeExtension(api: ExtensionAPI, options: Cre const emitLifecycle = async (type: "agent_start" | "agent_end", ctx: ExtensionContext): Promise => { const current = active; if (!current) return; + const adoptLifecycleBatch = ( + batch: + | Array<{ + kind: InvocationKind; + correlation: InvocationCorrelation; + connectionId: string | undefined; + }> + | undefined, + ): void => { + if (!batch || batch.length === 0) { + current.activeInvocation = undefined; + current.drainedInvocations = undefined; + activePromptOwnerHolder.connectionIds = undefined; + return; + } + current.activeInvocation = batch[0]; + current.drainedInvocations = batch.map(({ kind, correlation }) => ({ kind, correlation })); + const owners = new Set(); + for (const entry of batch) if (entry.connectionId !== undefined) owners.add(entry.connectionId); + activePromptOwnerHolder.connectionIds = owners; + }; + let transitions: Array<{ kind: InvocationKind; correlation: InvocationCorrelation }> = []; if (type === "agent_start") { // Drain EVERY entry admitted for this run: a continuation may promote // several follow-ups (each with its own requester correlation) into one @@ -2603,19 +2632,24 @@ export function createSdkSessionRuntimeExtension(api: ExtensionAPI, options: Cre // continuation agent_start with an empty queue leaves the current owner // untouched (review thread P1). const drained = current.pending.splice(0); - current.activeInvocation = drained[0]; if (drained.length > 0) { - const owners = new Set(); - for (const entry of drained) if (entry.connectionId !== undefined) owners.add(entry.connectionId); - activePromptOwnerHolder.connectionIds = owners; - // A single run may drain several follow-ups promoted together; each - // has its own durable record that must reach terminal state, so keep - // the full batch for the transition pass below (review thread P1). - current.drainedInvocations = drained.map(({ kind, correlation }) => ({ kind, correlation })); - } - if (current.activeInvocation?.kind === "prompt") { - current.deadlineManager.onAccepted(current.activeInvocation.correlation); + current.openLifecycleBatches.push({ invocations: drained }); + adoptLifecycleBatch(drained); + if (current.activeInvocation?.kind === "prompt") { + current.deadlineManager.onAccepted(current.activeInvocation.correlation); + } } + transitions = drained.map(({ kind, correlation }) => ({ kind, correlation })); + } else { + // Pair this agent_end with the oldest unmatched start. A delayed + // aborted-turn end that lands after a successor agent_start must + // terminalize the aborted invocation, never the successor. + const ended = current.openLifecycleBatches[0]; + transitions = ended + ? ended.invocations.map(({ kind, correlation }) => ({ kind, correlation })) + : current.activeInvocation + ? [current.activeInvocation] + : []; } // Observe whether the lifecycle publication actually landed: a terminal // abort awaits this result so its durable row only claims @@ -2624,12 +2658,6 @@ export function createSdkSessionRuntimeExtension(api: ExtensionAPI, options: Cre // recorded as observed=false, never rethrown into the api handler. let observed = true; try { - const transitions = - current.drainedInvocations && current.drainedInvocations.length > 0 - ? current.drainedInvocations - : current.activeInvocation - ? [current.activeInvocation] - : []; for (const invocation of transitions) { await current.reconciliation.noteTransition(invocation.kind, invocation.correlation, { type } as never); if ((type as string) === "agent_end" || (type as string) === "agent_failed") { @@ -2641,20 +2669,10 @@ export function createSdkSessionRuntimeExtension(api: ExtensionAPI, options: Cre observed = false; } if (type === "agent_end") { - if (current.activeInvocation?.kind === "prompt") { - current.deadlineManager.clear(current.activeInvocation.correlation); - } else if (current.drainedInvocations) { - for (const inv of current.drainedInvocations) - if (inv.kind === "prompt") current.deadlineManager.clear(inv.correlation); - } - current.activeInvocation = undefined; - current.drainedInvocations = undefined; - // The turn is over: no connection owns it anymore. Clearing here means - // an abort against a later agent-initiated turn (monitor/cron - // follow-up) finds no owner and fails closed, instead of letting the - // previous prompt's owner stop a turn it did not submit (review - // thread P1). - activePromptOwnerHolder.connectionIds = undefined; + if (current.openLifecycleBatches.length > 0) current.openLifecycleBatches.shift(); + for (const invocation of transitions) + if (invocation.kind === "prompt") current.deadlineManager.clear(invocation.correlation); + adoptLifecycleBatch(current.openLifecycleBatches[0]?.invocations); // Resolve EVERY concurrent waiter for the aborted turn: the turn emits // exactly one agent_end, and each admitted abort of it must observe the // same publication result rather than a single latest-wins slot (review @@ -2728,6 +2746,13 @@ export function createSdkSessionRuntimeExtension(api: ExtensionAPI, options: Cre correlation: InvocationCorrelation; connectionId: string | undefined; }> = []; + const openLifecycleBatches: Array<{ + invocations: Array<{ + kind: InvocationKind; + correlation: InvocationCorrelation; + connectionId: string | undefined; + }>; + }> = []; const configRevision = { current: 0 }; let acceptingGateResolutions = true; const inFlightGateResolutions = new Set>(); @@ -3003,6 +3028,7 @@ export function createSdkSessionRuntimeExtension(api: ExtensionAPI, options: Cre steerReconciliation, deadlineManager, pending, + openLifecycleBatches, registerBroker, fenceGateResolutions: () => { acceptingGateResolutions = false; @@ -3031,6 +3057,7 @@ export function createSdkSessionRuntimeExtension(api: ExtensionAPI, options: Cre steerReconciliation, deadlineManager, pending, + openLifecycleBatches, registerBroker, fenceGateResolutions: () => { acceptingGateResolutions = false; diff --git a/packages/coding-agent/src/sdk/session.ts b/packages/coding-agent/src/sdk/session.ts index b573a52c95..cae40dd32a 100644 --- a/packages/coding-agent/src/sdk/session.ts +++ b/packages/coding-agent/src/sdk/session.ts @@ -2849,9 +2849,7 @@ export async function createAgentSession(options: CreateAgentSessionOptions = {} model: agent.state.model, isIdle: () => !session.isStreaming, hasQueuedMessages: () => session.queuedMessageCount > 0, - abort: () => { - session.abort(); - }, + abort: () => session.abort(), settings, }); const toolContextStore = new ToolContextStore(getSessionContext); diff --git a/packages/coding-agent/src/session/agent-session.ts b/packages/coding-agent/src/session/agent-session.ts index 9ea8dc177a..87697b9494 100644 --- a/packages/coding-agent/src/session/agent-session.ts +++ b/packages/coding-agent/src/session/agent-session.ts @@ -2461,6 +2461,18 @@ export class AgentSession { * surface a red "Operation aborted" line; cleared by a later non-silent abort * or by `abort`'s safety net when no aborted message_end is produced. */ #silentAbortPending = false; + /** In-flight `abort()` unwind. Fresh prompts wait so they cannot steer into the dying turn. */ + #abortUnwind: Promise | undefined; + #abortEpoch = 0; + /** + * Monotonic count of ADMITTED abort requests, advanced synchronously on every + * `#abortWithOutcome` entry — including an abort that shares an already + * in-flight unwind rather than starting its own. A prompt that waits for an + * unwind captures this value and refuses to resume if it advanced while it + * waited: abort is terminal, so work admitted before the user's latest abort + * must never start just because the physical unwind was shared. + */ + #abortAdmissionEpoch = 0; /** Monotonic counter for `enqueueCustomMessageDisplay` tag generation; * combined with `Date.now()` so tags stay unique even across rapid * same-tick enqueues. */ @@ -2512,6 +2524,7 @@ export class AgentSession { nonEditDeterminations: 0, }; #promptInFlightCount = 0; + #inFlightGenerations: number[] = []; #agentEventHandlersInFlight = 0; #queuedExtensionEventCount = 0; #extensionTurnGeneration = 0; @@ -2822,10 +2835,18 @@ export class AgentSession { #beginInFlight(): void { this.#promptInFlightCount++; + this.#inFlightGenerations.push(this.#abortEpoch); if (this.#promptInFlightCount === 1) { this.#acquirePowerAssertion(); } } + + /** True while a live agent loop or a non-aborted in-flight prompt owns the session. */ + #isLiveTurnBusy(): boolean { + if (this.agent.state.isStreaming) return true; + if (!this.isStreaming) return false; + return this.#inFlightGenerations.every(generation => generation === this.#abortEpoch); + } /** * Allocate a FRESH prompt attempt/lineage for an allowed owned-completion * delivery (corrected turn semantics). The new turn gets a new attempt epoch @@ -2875,6 +2896,27 @@ export class AgentSession { return signal.aborted || this.#promptGeneration !== generation; } + /** + * Wait for the in-flight abort unwind, then prove no LATER abort was admitted + * while waiting. Overlapping aborts share one physical unwind, so awaiting + * `#abortUnwind` alone cannot distinguish "the abort I raced" from "a second + * abort the user issued after me". Returns false when the caller must not + * resume; the caller then rejects as a cancelled preflight rather than + * starting a successor the user already aborted. + * + * Loops because an abort admitted during the wait can install its own unwind: + * the fence only opens once no unwind is in flight and the admission epoch has + * stopped advancing. + */ + async #awaitAbortUnwindFence(): Promise { + const admissionEpoch = this.#abortAdmissionEpoch; + while (this.#abortUnwind !== undefined) { + await this.#abortUnwind; + if (this.#abortAdmissionEpoch !== admissionEpoch) return false; + } + return this.#abortAdmissionEpoch === admissionEpoch; + } + #throwIfPromptPreflightCancelled(generation: number, signal: AbortSignal): void { if (this.#isPromptPreflightCancelled(generation, signal)) { throw promptPreflightCancelledError(); @@ -3090,6 +3132,41 @@ export class AgentSession { #wakeScopedSettlementWaiters(): void { for (const check of [...this.#scopedSettlementWaiters]) check(); } + #abortedTurnTerminalPending(): boolean { + return ( + this.#promptInFlightCount > 0 || + this.#agentEventHandlersInFlight > 0 || + this.#pendingAgentEndEmit !== undefined || + this.#agentEndPublicationInFlight > 0 + ); + } + + /** Wait for the aborted turn's agent_end handler and publication, not just agent idle. */ + async #awaitAbortedTurnTerminal(): Promise { + await Promise.resolve(); + await this.#agentEndHandlingPromise; + this.#flushPendingAgentEnd(); + await this.#agentEndPublicationPromise; + while (this.#abortedTurnTerminalPending()) { + const wake = Promise.withResolvers(); + const check = () => { + if (!this.#abortedTurnTerminalPending()) wake.resolve(); + }; + this.#scopedSettlementWaiters.add(check); + check(); + try { + const waiters: Array> = [wake.promise]; + if (this.#agentEventHandlersInFlight > 0) waiters.push(this.#agentEndHandlingPromise); + if (this.#agentEndPublicationInFlight > 0) waiters.push(this.#agentEndPublicationPromise); + await Promise.race(waiters); + } finally { + this.#scopedSettlementWaiters.delete(check); + } + this.#flushPendingAgentEnd(); + await this.#agentEndHandlingPromise; + await this.#agentEndPublicationPromise; + } + } /** * Wait for session settlement. `ignoreSelectionFenceGeneration` carries the @@ -3129,6 +3206,7 @@ export class AgentSession { #endInFlight(): unknown { this.#promptInFlightCount = Math.max(0, this.#promptInFlightCount - 1); + if (this.#inFlightGenerations.length > 0) this.#inFlightGenerations.shift(); if (this.#promptInFlightCount !== 0) return undefined; this.#releasePowerAssertion(); @@ -9849,18 +9927,31 @@ export class AgentSession { assertImagePlaceholdersHavePayload(expandedText, options?.images); const workflowIntentDiff = options?.synthetic ? null : buildWorkflowIntentDiff(expandedText); const claimsGenuineUserIntent = !options?.synthetic && options?.attribution !== "agent"; - const admissionGeneration = this.#promptGeneration; - const admissionSignal = options?.preflightSignal + let admissionGeneration = this.#promptGeneration; + let admissionSignal = options?.preflightSignal ? AbortSignal.any([this.#promptPreflightAbortController.signal, options.preflightSignal]) : this.#promptPreflightAbortController.signal; if (this.#pendingSelectionFences > 0) { await awaitPromptInvocationPreflight(this.#selectionFenceTail, admissionSignal); } + let waitedForAbortUnwind = false; + if (this.#abortUnwind) { + // A LATER abort admitted while this prompt waited keeps abort terminal: + // the successor is refused instead of refreshing its generation and + // starting work the user already aborted twice. + if (!(await this.#awaitAbortUnwindFence())) throw promptPreflightCancelledError(); + admissionGeneration = this.#promptGeneration; + admissionSignal = options?.preflightSignal + ? AbortSignal.any([this.#promptPreflightAbortController.signal, options.preflightSignal]) + : this.#promptPreflightAbortController.signal; + waitedForAbortUnwind = true; + } const deepInterviewUserIntentEpoch = claimsGenuineUserIntent && !this.isStreaming ? this.#claimDeepInterviewUserIntent() : undefined; - // If streaming, queue via steer() or followUp() based on option - if (this.isStreaming) { + // If streaming, queue via steer() or followUp() based on option. + // Abort unwind is awaited above so a successor is not busy against leftover in-flight. + if (this.#isLiveTurnBusy() && !waitedForAbortUnwind) { if (!options?.streamingBehavior) { throw new AgentBusyError(); } @@ -11422,10 +11513,21 @@ export class AgentSession { if (images.length === 0) images = undefined; } - const admissionSignal = options?.preflightSignal + let admissionSignal = options?.preflightSignal ? AbortSignal.any([this.#promptPreflightAbortController.signal, options.preflightSignal]) : this.#promptPreflightAbortController.signal; - const preflightCancellationGeneration = this.#promptPreflightCancellationGeneration; + let preflightCancellationGeneration = this.#promptPreflightCancellationGeneration; + let waitedForAbortUnwind = false; + if (this.#abortUnwind && options?.deliverAs === undefined) { + // Same terminal-abort fence as `prompt()`: an overlapping second abort + // refuses this retained submission instead of letting it start late. + if (!(await this.#awaitAbortUnwindFence())) throw promptPreflightCancelledError(); + preflightCancellationGeneration = this.#promptPreflightCancellationGeneration; + admissionSignal = options?.preflightSignal + ? AbortSignal.any([this.#promptPreflightAbortController.signal, options.preflightSignal]) + : this.#promptPreflightAbortController.signal; + waitedForAbortUnwind = true; + } // Classify and reserve follow-up order synchronously, before any await // (selection fence or durable acceptance): a follow-up dispatch that is // not yet durably enqueued must already count as ahead, or a later plain @@ -11433,7 +11535,14 @@ export class AgentSession { // delivery/steering and overtake it. The reservation is released once // the durable enqueue settles either way, so a rejected or cancelled // acceptance leaves no phantom ordering behind. - const queuedPlainPrompt = options?.queuedAtDispatch === true && options.deliverAs === undefined; + // A busy SDK dispatch is only queued while the agent loop is live. + // Abort unwind is not a live loop: queuedAtDispatch from that window + // must not divert a fresh prompt into the dying turn's steer queue. + // Keep the original busy-dispatch bit for promotion/ownership: after + // unwind the successor starts its own turn and must fire onQueuedPromoted + // so SDK pending correlation is not dropped. + const dispatchedWhileBusy = options?.queuedAtDispatch === true && options.deliverAs === undefined; + const queuedPlainPrompt = dispatchedWhileBusy && this.agent.state.isStreaming; const hasFollowUpAhead = (): boolean => this.#activeFollowUpReservationEpochs.size > 0 || this.agent.snapshotFollowUp().length > 0 || @@ -11444,6 +11553,7 @@ export class AgentSession { !followUpAheadAtReservation && !this.agent.state.isStreaming && !this.#canAutoContinueForSteer(); + const promoteAfterAbortUnwind = waitedForAbortUnwind && dispatchedWhileBusy && !followUpAheadAtReservation; const deliverAs = options?.deliverAs ?? (queuedPlainPrompt @@ -11521,7 +11631,9 @@ export class AgentSession { // Compaction is intentionally NOT diverted here: prompt() handles an // in-flight compaction internally, and #queueSteer would otherwise park // the message in the steering queue with no turn to consume it. - if (this.isStreaming) { + // Abort unwind is awaited before classification so a successor is not + // parked in the dying turn's steer queue. + if (this.#isLiveTurnBusy() && !waitedForAbortUnwind) { if (options?.onPreflightAcceptCommit) await options.onPreflightAcceptCommit(); assertPreflightStillOpen(); await this.#queueSteer(text, images, { @@ -11536,7 +11648,7 @@ export class AgentSession { // Use prompt() with expandPromptTemplates: false to skip command handling and template expansion let queuedPromotionFired = false; const fireQueuedPromotion = () => { - if (!freshAtReservation || queuedPromotionFired) return; + if ((!freshAtReservation && !promoteAfterAbortUnwind) || queuedPromotionFired) return; queuedPromotionFired = true; options?.onQueuedPromoted?.(); }; @@ -11548,7 +11660,7 @@ export class AgentSession { fireQueuedPromotion(); }, onPreflightAcceptCommit: - options?.onPreflightAcceptCommit || freshAtReservation + options?.onPreflightAcceptCommit || freshAtReservation || promoteAfterAbortUnwind ? async () => { if (options?.onPreflightAcceptCommit) await options.onPreflightAcceptCommit(); else options?.onPreflightAccepted?.(); @@ -11954,6 +12066,7 @@ export class AgentSession { this.#markRetryReplayUnsafe(); this.abortRetry(); this.#promptGeneration++; + this.#abortEpoch++; this.#promptPreflightCancellationGeneration++; this.#promptPreflightAbortController.abort(); this.#promptPreflightAbortController = new AbortController(); @@ -11977,66 +12090,94 @@ export class AgentSession { | "internal"; silent?: boolean; }): Promise { - this.#abortOptions(options); - const postPromptDrain = this.#cancelPostPromptTasks(); - const managedLogicalRunId = - this.#defaultFallbackChain().chain.entries.length > 1 ? this.agent.currentManagedLogicalRunId : undefined; - this.agent.abort(); - const cleanup = Promise.all([postPromptDrain, this.agent.waitForIdle()]).then( - () => ({ kind: "settled" as const }), - (cause: unknown) => ({ kind: "error" as const, cause }), - ); - cleanup.catch(() => {}); - let outcome: AbortOutcome; - if (options?.timeoutMs !== undefined && options.timeoutMs > 0) { - outcome = await Promise.race([ - cleanup, - Bun.sleep(options.timeoutMs).then(() => ({ kind: "timeout" as const })), - ]); - if (outcome.kind === "timeout") { - this.#abandonPostPromptTasks(); - const forceAbortLogicalRunId = this.agent.currentManagedLogicalRunId ?? this.#activeLogicalRunId; - try { - if (forceAbortLogicalRunId !== undefined) - this.agent.forceAbort("Abort cleanup timed out", forceAbortLogicalRunId); - else this.agent.forceAbort("Abort cleanup timed out"); - } catch { - this.agent.forceAbort("Abort cleanup timed out"); + // Advance the admission epoch SYNCHRONOUSLY, before any await and before the + // shared-unwind branch. A second abort that piggybacks on the first unwind + // still creates a cancellation fence, so a prompt admitted between the two + // aborts is refused rather than started after the shared unwind resolves. + this.#abortAdmissionEpoch++; + if (this.#abortUnwind) { + // The shared unwind already ran `#abortOptions` for the FIRST abort, so this + // abort must still apply the effects that are specific to its own request. + // + // Cancellation fence: a prompt admitted before this abort keeps a live + // generation and preflight signal otherwise, and would start after the + // shared unwind resolves — work the user has already aborted twice. + this.#promptPreflightCancellationGeneration++; + this.#promptPreflightAbortController.abort(); + this.#promptPreflightAbortController = new AbortController(); + // Abort visibility is per-request: a later real abort must not inherit an + // earlier silent abort's suppression and swallow the user-visible notice. + if (options?.silent !== true) this.#silentAbortPending = false; + await this.#abortUnwind; + return { kind: "settled" }; + } + const unwind = Promise.withResolvers(); + this.#abortUnwind = unwind.promise; + try { + this.#abortOptions(options); + const postPromptDrain = this.#cancelPostPromptTasks(); + const managedLogicalRunId = + this.#defaultFallbackChain().chain.entries.length > 1 ? this.agent.currentManagedLogicalRunId : undefined; + this.agent.abort(); + const cleanup = Promise.all([postPromptDrain, this.agent.waitForIdle()]).then( + () => ({ kind: "settled" as const }), + (cause: unknown) => ({ kind: "error" as const, cause }), + ); + cleanup.catch(() => {}); + let outcome: AbortOutcome; + if (options?.timeoutMs !== undefined && options.timeoutMs > 0) { + outcome = await Promise.race([ + cleanup, + Bun.sleep(options.timeoutMs).then(() => ({ kind: "timeout" as const })), + ]); + if (outcome.kind === "timeout") { + this.#abandonPostPromptTasks(); + const forceAbortLogicalRunId = this.agent.currentManagedLogicalRunId ?? this.#activeLogicalRunId; + try { + if (forceAbortLogicalRunId !== undefined) + this.agent.forceAbort("Abort cleanup timed out", forceAbortLogicalRunId); + else this.agent.forceAbort("Abort cleanup timed out"); + } catch { + this.agent.forceAbort("Abort cleanup timed out"); + } + this.emitNotice( + "warning", + "Abort cleanup timed out; forced session recovery. The previous provider stream or tool may still be unwinding in the background.", + "abort", + ); } - this.emitNotice( - "warning", - "Abort cleanup timed out; forced session recovery. The previous provider stream or tool may still be unwinding in the background.", - "abort", - ); + } else { + outcome = await cleanup; } - } else { - outcome = await cleanup; - } - try { - await this.#goalRuntime.onTaskAborted({ reason: options?.goalReason ?? "interrupted" }); - if (managedLogicalRunId !== undefined) - this.agent.requestRunTerminal(managedLogicalRunId, { stopReason: "cancelled" }); - this.#flushPendingBackgroundExchanges(); - this.#flushPendingAgentEnd(); - if ( - !this.#cancelAndSubmitInProgress && - (options?.cause ?? "internal") === "user_interrupt" && - this.agent.hasQueuedSteering() - ) { - this.#scheduleAgentContinue({ - delayMs: 1, - generation: this.#promptGeneration, - shouldContinue: () => this.agent.hasQueuedSteering(), - rescheduleOnBusy: true, - continueQueuedOnly: true, - }); + try { + await this.#goalRuntime.onTaskAborted({ reason: options?.goalReason ?? "interrupted" }); + if (managedLogicalRunId !== undefined) + this.agent.requestRunTerminal(managedLogicalRunId, { stopReason: "cancelled" }); + this.#flushPendingBackgroundExchanges(); + await this.#awaitAbortedTurnTerminal(); + if ( + !this.#cancelAndSubmitInProgress && + (options?.cause ?? "internal") === "user_interrupt" && + this.agent.hasQueuedSteering() + ) { + this.#scheduleAgentContinue({ + delayMs: 1, + generation: this.#promptGeneration, + shouldContinue: () => this.agent.hasQueuedSteering(), + rescheduleOnBusy: true, + continueQueuedOnly: true, + }); + } + return outcome; + } catch (cause) { + return { kind: "error", cause }; + } finally { + this.#silentAbortPending = false; + if (this.#toolChoiceQueue.hasInFlight) this.#toolChoiceQueue.reject("aborted"); } - return outcome; - } catch (cause) { - return { kind: "error", cause }; } finally { - this.#silentAbortPending = false; - if (this.#toolChoiceQueue.hasInFlight) this.#toolChoiceQueue.reject("aborted"); + this.#abortUnwind = undefined; + unwind.resolve(); } } diff --git a/packages/coding-agent/test/agent-session-concurrent.test.ts b/packages/coding-agent/test/agent-session-concurrent.test.ts index b1afdce5be..749b59d1a4 100644 --- a/packages/coding-agent/test/agent-session-concurrent.test.ts +++ b/packages/coding-agent/test/agent-session-concurrent.test.ts @@ -267,6 +267,297 @@ describe("AgentSession concurrent prompt guard", () => { await session.abort(); await send.catch(() => {}); }); + it("immediate sendUserMessage after abort starts a successor turn instead of parking in steer", async () => { + const model = getBundledModel("anthropic", "claude-sonnet-4-5")!; + const agent = new Agent({ + getApiKey: () => "test-key", + initialState: { model, systemPrompt: ["Test"], tools: [] }, + // Branch on the newest user message, NOT a stream-call counter: the + // session reports `isStreaming` before `streamFn` runs, so the aborted + // turn's dispatch races the abort. Keyed on a counter, the SUCCESSOR turn + // could take the blocking branch and hang the test whenever the first + // turn's dispatch had not landed yet. + streamFn: (_model, context, options) => { + const lastUser = [...(context?.messages ?? [])].reverse().find(message => message.role === "user"); + const text = + lastUser && Array.isArray(lastUser.content) + ? lastUser.content + .map(part => (typeof part === "object" && part.type === "text" ? part.text : "")) + .join("") + : ""; + const signal = options?.signal; + const stream = new AssistantMessageEventStream(); + queueMicrotask(() => { + if (text === "First message") { + stream.push({ type: "start", partial: createAssistantMessage("") }); + const abortStream = () => { + stream.push({ + type: "error", + reason: "aborted", + error: createAssistantMessage("Aborted"), + }); + }; + if (signal?.aborted) { + abortStream(); + return; + } + signal?.addEventListener("abort", abortStream, { once: true }); + return; + } + const message = createAssistantMessage("successor ok"); + stream.push({ type: "start", partial: message }); + stream.push({ type: "done", reason: "stop", message }); + }); + return stream; + }, + }); + const sessionManager = SessionManager.inMemory(); + const authStorage = await AuthStorage.create(path.join(tempDir, "testauth-abort-immediate.db")); + authStorages.push(authStorage); + const modelRegistry = new ModelRegistry(authStorage, path.join(tempDir, "models.yml")); + authStorage.setRuntimeApiKey("anthropic", "test-key"); + session = new AgentSession({ agent, sessionManager, settings: Settings.isolated(), modelRegistry }); + + const firstPrompt = session.prompt("First message"); + await waitFor(() => session.agent.state.isStreaming); + const aborting = session.abort(); + let promoted = 0; + const successor = session.sendUserMessage("successor after abort", { + queuedAtDispatch: true, + onQueuedPromoted: () => { + promoted += 1; + }, + }); + await aborting; + await waitFor(() => { + const lastUser = [...agent.state.messages].reverse().find(message => message.role === "user"); + const content = lastUser?.content[0]; + return Boolean( + content && typeof content === "object" && "text" in content && content.text === "successor after abort", + ); + }, 3_000); + expect(session.getQueuedMessages()).toEqual({ steering: [], followUp: [] }); + expect(promoted).toBe(1); + await successor.catch(() => {}); + await firstPrompt.catch(() => {}); + }, 15_000); + + /** + * Session for abort-lifecycle ordering. The turn whose newest user message is + * `BLOCKING_PROMPT` streams until aborted; any other turn terminates on its + * own. Dispatch is keyed on message CONTENT rather than a call counter: the + * provider dispatch for the aborted turn races the abort itself, so a counter + * cannot reliably identify which turn a stream belongs to. `dispatched` + * therefore doubles as the exact list of turns that actually reached the + * provider, which is what "no successor started" has to assert. + * `successorOutcome` selects whether a successor turn completes or errors. + */ + const BLOCKING_PROMPT = "First message"; + async function createAbortLifecycleSession( + dbName: string, + successorOutcome: "complete" | "error" = "complete", + ): Promise<{ + agent: Agent; + dispatched: () => string[]; + userTexts: () => string[]; + }> { + const model = getBundledModel("anthropic", "claude-sonnet-4-5")!; + const dispatched: string[] = []; + const textOf = (message: Message | undefined): string => + message && Array.isArray(message.content) + ? message.content.map(part => (typeof part === "object" && part.type === "text" ? part.text : "")).join("") + : ""; + const agent = new Agent({ + getApiKey: () => "test-key", + initialState: { model, systemPrompt: ["Test"], tools: [] }, + streamFn: (_model, context, options) => { + const text = textOf([...(context?.messages ?? [])].reverse().find(message => message.role === "user")); + dispatched.push(text); + const signal = options?.signal; + const stream = new AssistantMessageEventStream(); + queueMicrotask(() => { + if (text === BLOCKING_PROMPT) { + stream.push({ type: "start", partial: createAssistantMessage("") }); + const abortStream = () => { + stream.push({ type: "error", reason: "aborted", error: createAssistantMessage("Aborted") }); + }; + if (signal?.aborted) abortStream(); + else signal?.addEventListener("abort", abortStream, { once: true }); + return; + } + const message = createAssistantMessage("successor ok"); + stream.push({ type: "start", partial: message }); + if (successorOutcome === "error") { + stream.push({ type: "error", reason: "error", error: createAssistantMessage("successor failed") }); + return; + } + stream.push({ type: "done", reason: "stop", message }); + }); + return stream; + }, + }); + const authStorage = await AuthStorage.create(path.join(tempDir, dbName)); + authStorages.push(authStorage); + const modelRegistry = new ModelRegistry(authStorage, path.join(tempDir, "models.yml")); + authStorage.setRuntimeApiKey("anthropic", "test-key"); + session = new AgentSession({ + agent, + sessionManager: SessionManager.inMemory(), + settings: Settings.isolated(), + modelRegistry, + }); + return { + agent, + dispatched: () => [...dispatched], + userTexts: () => agent.state.messages.filter(message => message.role === "user").map(textOf), + }; + } + + /** + * The #4753 overlapping-abort regression. Abort A installs the unwind, prompt P + * is admitted and parks on it, then abort B shares that same physical unwind. + * All three enter synchronously in one tick, so the interleaving is exact and + * does not depend on timers. Before the abort-admission fence, B acknowledged + * success and P then refreshed its generation and started a successor turn: + * the user aborted twice and work began anyway. + */ + it("a second abort fences a prompt retained by the first abort's unwind (#4753 overlapping abort)", async () => { + const { dispatched, userTexts } = await createAbortLifecycleSession("testauth-overlap-abort.db"); + + const firstPrompt = session.prompt(BLOCKING_PROMPT); + await waitFor(() => session.agent.state.isStreaming); + + const abortA = session.abort({ cause: "user_interrupt" }); + const retained = session.sendUserMessage("retained prompt", { queuedAtDispatch: true }); + const abortB = session.abort({ cause: "user_interrupt" }); + + // Both aborts acknowledge; abort stays terminal for the prompt between them. + await abortA; + await abortB; + await expect(retained).rejects.toMatchObject({ name: "PromptPreflightCancelledError" }); + + // No successor: the retained prompt never reached the provider and never + // entered the transcript. + await session.waitForIdle(); + await Bun.sleep(50); + expect(dispatched()).not.toContain("retained prompt"); + expect(userTexts()).not.toContain("retained prompt"); + expect(session.getQueuedMessages()).toEqual({ steering: [], followUp: [] }); + expect(session.isStreaming).toBe(false); + + await firstPrompt.catch(() => {}); + }, 15_000); + + it("a retained prompt survives a single abort and is delivered exactly once (#4753)", async () => { + const { dispatched, userTexts } = await createAbortLifecycleSession("testauth-retained-once.db"); + + const firstPrompt = session.prompt(BLOCKING_PROMPT); + await waitFor(() => session.agent.state.isStreaming); + + const aborting = session.abort({ cause: "user_interrupt" }); + let promoted = 0; + const retained = session.sendUserMessage("retained prompt", { + queuedAtDispatch: true, + onQueuedPromoted: () => { + promoted += 1; + }, + }); + + await aborting; + await retained; + await session.waitForIdle(); + await Bun.sleep(50); + + // Delivered once, as exactly one successor turn that reached the provider once. + expect(userTexts().filter(text => text === "retained prompt")).toEqual(["retained prompt"]); + expect(dispatched().filter(text => text === "retained prompt")).toEqual(["retained prompt"]); + expect(promoted).toBe(1); + expect(session.getQueuedMessages()).toEqual({ steering: [], followUp: [] }); + + await firstPrompt.catch(() => {}); + }, 20_000); + + it("queued steering is still resumed after an abort unwind (#4753)", async () => { + const { userTexts } = await createAbortLifecycleSession("testauth-abort-steering.db"); + + const firstPrompt = session.prompt(BLOCKING_PROMPT); + await waitFor(() => session.agent.state.isStreaming); + + // Explicit steering is delivered into the aborted turn's queue, then the + // abort's user_interrupt resume path promotes it to its own turn. + await session.sendUserMessage("queued steer", { deliverAs: "steer" }); + await session.abort({ cause: "user_interrupt" }); + await firstPrompt.catch(() => {}); + await waitFor(() => userTexts().includes("queued steer"), 5_000); + await session.waitForIdle(); + + expect(userTexts().filter(text => text === "queued steer")).toEqual(["queued steer"]); + expect(session.getQueuedMessages()).toEqual({ steering: [], followUp: [] }); + }, 20_000); + + it("rapid repeated aborts all settle and leave the session idle (#4753)", async () => { + const { dispatched } = await createAbortLifecycleSession("testauth-rapid-abort.db"); + + const firstPrompt = session.prompt(BLOCKING_PROMPT); + await waitFor(() => session.agent.state.isStreaming); + + // Four same-tick aborts share one physical unwind; each must still settle. + const aborts = [ + session.abort({ cause: "user_interrupt" }), + session.abort({ cause: "user_interrupt" }), + session.abort({ cause: "user_interrupt" }), + session.abort({ cause: "user_interrupt" }), + ]; + for (const settled of await Promise.allSettled(aborts)) expect(settled.status).toBe("fulfilled"); + // A later abort, after the shared unwind completed, is still terminal. + await session.abort({ cause: "user_interrupt" }); + + await session.waitForIdle(); + expect(session.isStreaming).toBe(false); + // No turn other than the aborted one ever reached the provider. + expect(dispatched().filter(text => text !== BLOCKING_PROMPT)).toEqual([]); + expect(session.getQueuedMessages()).toEqual({ steering: [], followUp: [] }); + + await firstPrompt.catch(() => {}); + }, 15_000); + + it("a prompt admitted after the abort settled runs to execution completion (#4753)", async () => { + const { dispatched, userTexts } = await createAbortLifecycleSession("testauth-post-abort.db"); + + const firstPrompt = session.prompt(BLOCKING_PROMPT); + await waitFor(() => session.agent.state.isStreaming); + await session.abort({ cause: "user_interrupt" }); + await firstPrompt.catch(() => {}); + + // The fence is terminal only for prompts admitted BEFORE an abort; a prompt + // admitted after it settled runs a normal turn to completion. + await session.sendUserMessage("after abort settled"); + await session.waitForIdle(); + + expect(userTexts()).toContain("after abort settled"); + expect(dispatched().filter(text => text !== BLOCKING_PROMPT)).toEqual(["after abort settled"]); + expect(session.isStreaming).toBe(false); + }, 15_000); + + it("a successor turn that errors leaves the session idle and abortable (#4753)", async () => { + const { dispatched, userTexts } = await createAbortLifecycleSession("testauth-successor-error.db", "error"); + + const firstPrompt = session.prompt(BLOCKING_PROMPT); + await waitFor(() => session.agent.state.isStreaming); + await session.abort({ cause: "user_interrupt" }); + await firstPrompt.catch(() => {}); + + // The successor is admitted and started, then its execution fails. + await session.sendUserMessage("failing successor").catch(() => {}); + await session.waitForIdle(); + + expect(userTexts()).toContain("failing successor"); + expect(dispatched().filter(text => text !== BLOCKING_PROMPT)).toEqual(["failing successor"]); + expect(session.isStreaming).toBe(false); + // An execution error is not an abort: a later abort still settles cleanly. + await session.abort({ cause: "user_interrupt" }); + expect(session.getQueuedMessages()).toEqual({ steering: [], followUp: [] }); + }, 15_000); it("sendUserMessage rejects absent content with a typed invalid_input error without crashing", async () => { await createSession();