diff --git a/packages/coding-agent/CHANGELOG.md b/packages/coding-agent/CHANGELOG.md index 8d8d63722a..b756c9d24a 100644 --- a/packages/coding-agent/CHANGELOG.md +++ b/packages/coding-agent/CHANGELOG.md @@ -26,6 +26,7 @@ - Post-merge repair for #4542: `CHAT_DAEMON_GENERATIONS.discord` 64→65 and `.slack` 67→68 so the `SessionRouter` initial attachment replay change is generation-fenced for already-running Discord and Slack daemons. The semantic guard manifest is regenerated with the corrected generation and declaration digests. - Cursor split responses now preserve canonical provider order across pre-admission artifact spilling, so a server-side tool result cannot be persisted after the assistant continuation that follows it. The canonical admission reservation is owned by each emission's own handler, so a host bridge or replay emitting the same `message_end` event object twice can no longer overwrite the reservation and deadlock every later canonical admission (#4536). - Uncontended canonical `message_end` admissions no longer cost a microtask: the admission lane keeps a synchronous fast path when its predecessor slot is already released, restoring synchronous visibility of persisted appends for external emitters (successor-finalization and deep-interview continuation flows) while FIFO admission still holds under a genuinely in-flight predecessor (#4536). +- SDK `session.prompt`/`steer` dispatches that arrive while a default-model selection already owns the admission fence are now preserved as queued work instead of failing: the shared ingress awaits the selection fence, keeps same-session prompt entry fail-fast, and only private scheduled continuations (auto-compaction/queue continuation) bypass the fence with an explicit reentry capability. Queued dispatches reserve follow-up order before durable acceptance so a later plain prompt cannot overtake an earlier follow-up behind the same fence, and continuations parked behind a pending selection fence are tracked as settlement work so `waitForIdle` cannot report completion while the continuation is still waiting (#4519). - Default-model selection now reserves a causal fence before credential probing, so an already accepted prompt preflight cannot be overtaken and a selection accepted first blocks later prompt preflight through durable publication. The fence does not hold session admission across `waitForIdle`, allowing inherited auto-compaction continuations to obtain prompt admission; same-session reentrancy still fails fast, successor sessions remain protected, and disposal deterministically drains accepted selections while rejecting queued prompts without unhandled rejections (#4519). - Ordinary sessions no longer import or execute Claude Code and Codex directory hooks as competing runtime authorities. Runtime hook discovery is fail-closed to canonical native `.gjc/hooks/` providers while explicit configured paths, constrained plugin hooks, and foreign-provider import/diagnostic discovery remain available (#4516). - Telegram/Slack/Discord outbound publications no longer freeze after a session-host rehost. A rehosted fleet re-attaches every session in one reconcile pass, and each attachment's initial `event_replay` was awaited inside the serialized `#reconcileTail`, so one slow replay (up to its full retry budget) wedged all later reconciles and the sends funneling through them; leases and inbound polling stayed green while delivery died until daemon restart. Reconcile-driven attachments now publish immediately and run initial replay on the attachment's ready tail (matching the reconnect path); `start()` still drains those tails so bootstrap callers observe replay completion. Replay ordering, generation fences, cross-session isolation, and provider hooks are unchanged (#4527). diff --git a/packages/coding-agent/src/extensibility/extensions/loader.ts b/packages/coding-agent/src/extensibility/extensions/loader.ts index 20b24cdb8e..af512db88e 100644 --- a/packages/coding-agent/src/extensibility/extensions/loader.ts +++ b/packages/coding-agent/src/extensibility/extensions/loader.ts @@ -242,9 +242,12 @@ class ConcreteExtensionAPI implements ExtensionAPI, IExtensionRuntime { content: string | (TextContent | ImageContent)[], options?: { deliverAs?: "steer" | "followUp"; + queuedAtDispatch?: boolean; onPreflightAccepted?: () => void; onPreflightAcceptCommit?: () => void | Promise; + onQueuedPromoted?: () => void; preflightSignal?: AbortSignal; + sdkRunToken?: string; }, ): Promise { return Promise.resolve(this.runtime.sendUserMessage(content, options)); diff --git a/packages/coding-agent/src/extensibility/extensions/runner.ts b/packages/coding-agent/src/extensibility/extensions/runner.ts index cb499ddcd7..a0da33a565 100644 --- a/packages/coding-agent/src/extensibility/extensions/runner.ts +++ b/packages/coding-agent/src/extensibility/extensions/runner.ts @@ -207,7 +207,7 @@ export class ExtensionRunner { #isIdleFn: () => boolean = () => true; #getActivePromptHandleFn: () => string | undefined = () => undefined; #waitForIdleFn: () => Promise = async () => {}; - #abortFn: () => void = () => {}; + #abortFn: () => void | Promise = () => {}; #abortPromptAndWaitFn: NonNullable = async () => { throw new Error("abortPromptAndWait binding is unavailable"); }; diff --git a/packages/coding-agent/src/extensibility/extensions/types.ts b/packages/coding-agent/src/extensibility/extensions/types.ts index 1ae3be8d62..1c5b65eb04 100644 --- a/packages/coding-agent/src/extensibility/extensions/types.ts +++ b/packages/coding-agent/src/extensibility/extensions/types.ts @@ -369,7 +369,7 @@ export interface ExtensionContext { /** Stable resource ownership identifier for the active prompt run. */ getActivePromptHandle(): string | undefined; /** Abort the current agent operation */ - abort(): void; + abort(): void | Promise; /** Abort and prove whether resources for a specific prompt settled. */ abortPromptAndWait?(handle: string, options: { graceMs: number }): Promise; /** Whether there are queued messages waiting */ @@ -1186,8 +1186,11 @@ export interface ExtensionAPI { content: string | (TextContent | ImageContent)[], options?: { deliverAs?: "steer" | "followUp"; + /** Internal SDK signal preserving a busy dispatch across async admission fences. */ + queuedAtDispatch?: boolean; onPreflightAccepted?: () => void; onPreflightAcceptCommit?: () => void | Promise; + onQueuedPromoted?: () => void; preflightSignal?: AbortSignal; /** Internal SDK correlation owner for an exact queued follow-up. */ sdkRunToken?: string; @@ -1505,7 +1508,7 @@ export interface ExtensionContextActions { isIdle: () => boolean; /** Stable resource ownership identifier for the active prompt run. */ getActivePromptHandle?: () => string | undefined; - abort: () => void; + abort: () => void | Promise; abortPromptAndWait?: (handle: string, options: { graceMs: number }) => Promise; hasPendingMessages: () => boolean; 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 1509cda2ca..ef0bcfb307 100644 --- a/packages/coding-agent/src/sdk/host/session-runtime.test.ts +++ b/packages/coding-agent/src/sdk/host/session-runtime.test.ts @@ -434,6 +434,7 @@ describe("SessionSdkSessionRuntime", () => { const cwd = await mkdtemp(path.join(os.tmpdir(), "gjc-sdk-unwind-promoted-")); let idle = true; let promoted: (() => void) | undefined; + const queuedDispositions: boolean[] = []; const handlers = new Map Promise | void>(); const api = { on(event: string, handler: (event: unknown, ctx: ExtensionContext) => Promise | void) { @@ -446,10 +447,12 @@ describe("SessionSdkSessionRuntime", () => { onPreflightAccepted?: () => void; onPreflightAcceptCommit?: () => void; onQueuedPromoted?: () => void; + queuedAtDispatch?: boolean; } | undefined, ) => Promise.resolve(options?.onPreflightAcceptCommit?.()).then(() => { + queuedDispositions.push(options?.queuedAtDispatch === true); promoted = options?.onQueuedPromoted; options?.onPreflightAccepted?.(); return {}; @@ -509,6 +512,7 @@ describe("SessionSdkSessionRuntime", () => { idle = false; prompt("conn-b", "unwind-b"); await waitResponse("unwind-b"); + expect(queuedDispositions).toEqual([false, true]); expect(promoted).toBeDefined(); // The unwind continuation promotes the queued steer to its own run: the // correlation hook fires before agent_start... diff --git a/packages/coding-agent/src/sdk/host/session-runtime.ts b/packages/coding-agent/src/sdk/host/session-runtime.ts index 268a6bc61a..60f66456ec 100644 --- a/packages/coding-agent/src/sdk/host/session-runtime.ts +++ b/packages/coding-agent/src/sdk/host/session-runtime.ts @@ -1317,6 +1317,7 @@ function createControlSurface( onPreflightAcceptCommit: () => Promise; /** Fired when a queued submission (steering or follow-up) is promoted to its own run (SDK ownership correlation). */ onQueuedPromoted: () => void; + queuedAtDispatch: boolean; }) => Promise, acceptedFields?: () => Record, allowCompletionFallback = false, @@ -1392,6 +1393,7 @@ function createControlSurface( // created at promotion so the submitting connection can // terminal-abort that turn (review threads P1/P2). onQueuedPromoted: () => onPromotedTurn?.(kind, correlation, requesterConnectionId), + queuedAtDispatch, }), ); void submission.then( @@ -2316,10 +2318,10 @@ function createControlSurface( }; return { prompt: async (text, images, clientRef) => - submit("prompt", clientRef, options => + submit("prompt", clientRef, ({ queuedAtDispatch, ...options }) => api.sendUserMessage( typeof images === "undefined" ? text : ([{ type: "text", text }, ...(images as never[])] as never), - options, + queuedAtDispatch ? { ...options, queuedAtDispatch: true } : options, ), ), steer: async (text, clientRef) => { @@ -2355,7 +2357,7 @@ function createControlSurface( }, abortTerminal: terminalAbort, abortAndPrompt: async text => { - ctx.abort(); + await ctx.abort(); return await submit("prompt", undefined, options => api.sendUserMessage(text, options)); }, answerAsk: unavailable("ask.answer"), diff --git a/packages/coding-agent/src/session/agent-session.ts b/packages/coding-agent/src/session/agent-session.ts index aadb5147fb..14588f1fad 100644 --- a/packages/coding-agent/src/session/agent-session.ts +++ b/packages/coding-agent/src/session/agent-session.ts @@ -1886,6 +1886,7 @@ type SessionAdmissionEntry = { ready: PromiseWithResolvers; settled: PromiseWithResolvers; released: boolean; + selectionFenceGeneration: number; continuationCapability?: symbol; }; @@ -1987,7 +1988,17 @@ export class AgentSession { #sessionAdmissionClosing = false; #sessionAdmissionClosed = false; #sessionAdmissionContext = new AsyncLocalStorage(); + #selectionFenceGenerationContext = new AsyncLocalStorage(); #selectionFenceTail: Promise = Promise.resolve(); + #pendingSelectionFences = 0; + #selectionFenceDeferredContinuations = new Map(); + #scopedSettlementWaiters = new Set<() => void>(); + #oldestPendingSelectionFenceGeneration = 0; + #followUpReservationEpoch = 0; + /** Epochs of follow-up reservations still between reservation and durable enqueue. */ + #activeFollowUpReservationEpochs = new Set(); + #followUpReservationDrainWaiters = new Set<() => void>(); + #selectionFenceGeneration = 0; #defaultModelSelectionMutationRevision = 0; #thinkingLevelMutationRevision = 0; #thinkingVisibilityMutationRevision = 0; @@ -2431,6 +2442,7 @@ export class AgentSession { /** Test-only abort outcome override for cancel-and-submit rollback coverage. */ #cancelAndSubmitAbortOutcomeProviderForTests: (() => Promise) | undefined = undefined; #postPromptTasks = new Set>(); + #postPromptTaskSelectionFenceGenerations = new Map, number>(); #postPromptTasksPromise: Promise | undefined = undefined; #postPromptTasksResolve: (() => void) | undefined = undefined; #postPromptTasksAbortController = new AbortController(); @@ -2505,6 +2517,7 @@ export class AgentSession { #terminalLineageSecret = crypto.randomUUID(); #promptGeneration = 0; #promptPreflightAbortController = new AbortController(); + #promptPreflightCancellationGeneration = 0; #providerSessionState = new Map(); #temporaryProviderSessionScopes: TemporaryProviderSessionScopeRecord[] = []; @@ -2570,6 +2583,12 @@ export class AgentSession { }); } + #assertSessionAdmissionOpen(): void { + if (this.#sessionAdmissionClosing || this.#sessionAdmissionClosed || this.#isDisposed) { + throw this.#sessionAdmissionBusyError(); + } + } + /** * Reject a turn start while a handoff transition owns the session. Handoff never * routes its own generation/injection through these turn-start chokepoints, and @@ -2647,12 +2666,12 @@ export class AgentSession { body: (lease: SessionAdmissionLease) => Promise, signal?: AbortSignal, continuationAdmission?: ScheduledContinuationAdmission, - options?: { allowDuringClosing?: boolean }, + options?: { + allowDuringClosing?: boolean; + bypassSelectionFenceGeneration?: number; + allowPromptContinuationReentry?: boolean; + }, ): Promise { - if (kind === "prompt") await awaitPromptInvocationPreflight(this.#awaitStartupTurnBarrier(), signal); - if (kind === "prompt" && continuationAdmission === undefined) { - await awaitPromptInvocationPreflight(this.#selectionFenceTail, signal); - } const owner = this.#sessionAdmissionContext.getStore(); if (owner && !owner.released) { if ( @@ -2660,8 +2679,23 @@ export class AgentSession { continuationAdmission.capability === owner.continuationCapability ) return await body({ release: () => {} }); + if (options?.allowPromptContinuationReentry === true && owner.kind === "prompt") { + return await body({ release: () => {} }); + } throw this.#sessionAdmissionBusyError(); } + if (kind === "prompt") await awaitPromptInvocationPreflight(this.#awaitStartupTurnBarrier(), signal); + const bypassesSelectionFence = + options?.bypassSelectionFenceGeneration !== undefined && + options.bypassSelectionFenceGeneration < this.#selectionFenceGeneration; + if ( + kind === "prompt" && + continuationAdmission === undefined && + !bypassesSelectionFence && + this.#pendingSelectionFences > 0 + ) { + await awaitPromptInvocationPreflight(this.#selectionFenceTail, signal); + } if ( this.#sessionAdmissionClosed || ((this.#sessionAdmissionClosing || this.#isDisposed) && options?.allowDuringClosing !== true) @@ -2683,6 +2717,7 @@ export class AgentSession { ready: Promise.withResolvers(), settled: Promise.withResolvers(), released: false, + selectionFenceGeneration: this.#selectionFenceGeneration, ...(kind === "prompt" ? { continuationCapability: Symbol("scheduled-continuation") } : {}), }; const releaseEntry = () => { @@ -2739,6 +2774,7 @@ export class AgentSession { const active = this.#activeSessionAdmission; if (active?.kind === "prompt") { this.#promptGeneration++; + this.#promptPreflightCancellationGeneration++; this.#promptPreflightAbortController.abort(); } if (active) await active.settled.promise; @@ -2837,6 +2873,17 @@ export class AgentSession { return hold; } + #restoreAndReserveDeferredAgentEndForContinuation(pending: AgentSessionEvent | undefined): symbol | undefined { + if (!pending) return undefined; + if (this.#pendingAgentEndEmit && this.#pendingAgentEndEmit !== pending) { + throw new Error("Cannot restore a deferred agent_end over a different pending terminal event"); + } + this.#pendingAgentEndEmit = pending; + const hold = Symbol("deferred-agent-end-continuation"); + this.#pendingAgentEndContinuationHolds.set(hold, pending); + return hold; + } + #claimDeferredAgentEndForContinuation(hold: symbol | undefined): AgentSessionEvent | undefined { if (!hold) return undefined; const pending = this.#pendingAgentEndContinuationHolds.get(hold); @@ -2894,33 +2941,159 @@ export class AgentSession { } lease.closeDiscovery(); } + /** + * Keep a continuation parked behind a pending selection fence observable to + * external idle waits. The deferral window is bounded: it ends when the + * deferred invocation synchronously re-reserves its settlement markers + * inside the scheduler (`#endSelectionFenceDeferralTracking`); the + * promise-settled release is only the safety net for a deferral whose + * re-entry never ran (disposed/aborted session). + * + * The counter is keyed by the fence generation it parks behind. The + * selection that owns that fence must not wait on its own parked + * continuations: its mid-selection `waitForIdle` is exactly what lets an + * inherited continuation settle before the durable mutation, and waiting on + * a continuation parked behind the fence the selection itself holds would + * reintroduce the selection self-deadlock (#4519). + */ + #trackSelectionFenceDeferredContinuation(fenceGeneration: number, deferred: Promise): void { + this.#selectionFenceDeferredContinuations.set( + fenceGeneration, + (this.#selectionFenceDeferredContinuations.get(fenceGeneration) ?? 0) + 1, + ); + void deferred + .catch(() => {}) + .finally(() => { + this.#endSelectionFenceDeferralTracking(fenceGeneration); + }); + } + + /** + * End the deferral-limbo tracking window: called synchronously by the + * deferred scheduling path once it has re-reserved its own settlement + * markers (the predecessor hold and post-prompt task), so the counter no + * longer covers the continuation's own execution. + */ + #endSelectionFenceDeferralTracking(fenceGeneration: number): void { + const pending = this.#selectionFenceDeferredContinuations.get(fenceGeneration) ?? 0; + if (pending <= 0) return; + if (pending === 1) this.#selectionFenceDeferredContinuations.delete(fenceGeneration); + else this.#selectionFenceDeferredContinuations.set(fenceGeneration, pending - 1); + this.#resolveSessionSettlement(); + } + + /** + * Continuations parked behind selection fences that idle waits must observe. + * + * Only deferrals whose fence has already settled are counted: their + * deferred scheduling tail is (re-)entering within microtask time and the + * session is not idle until it does. A deferral behind a still-pending + * fence is invisible by construction — the fence owner's own completion may + * causally depend on the very settlement the waiter would block on, so + * counting it would reintroduce the selection self-deadlock (#4519). It + * becomes observable the moment its fence settles, and either the + * synchronous re-entry (`#endSelectionFenceDeferralTracking`) or the + * promise-settled safety net clears it. + */ + #pendingSelectionFenceDeferredContinuations(ignoreGenerationFrom?: number): number { + let total = 0; + for (const [generation, count] of this.#selectionFenceDeferredContinuations) { + if (ignoreGenerationFrom !== undefined && generation >= ignoreGenerationFrom) continue; + total += count; + } + return total; + } - #isSessionSettlementPending(): boolean { + /** + * Deferral generations that causally-upstream settlement work must ignore: + * a deferral parked behind a still-pending selection fence cannot settle + * before that fence resolves, and the fence owner's completion may itself + * depend on this very settlement. Returns the oldest still-pending fence + * generation, or undefined when no fence is pending. + */ + #turnSettlementDeferralFloor(): number | undefined { + return this.#pendingSelectionFences > 0 ? this.#oldestPendingSelectionFenceGeneration : undefined; + } + + /** + * Wait until every follow-up reservation minted before `ownEpoch` has been + * released, so a later dispatch can never durably enqueue ahead of an + * earlier reserved follow-up. Resolves immediately when no earlier + * reservation remains. + */ + async #waitForEarlierFollowUpReservations( + ownEpoch: number, + requesterSignal?: AbortSignal, + admissionSignal?: AbortSignal, + ): Promise { + while ([...this.#activeFollowUpReservationEpochs].some(epoch => epoch < ownEpoch)) { + if (requesterSignal?.aborted || admissionSignal?.aborted) throw promptPreflightCancelledError(); + const drained = Promise.withResolvers(); + this.#followUpReservationDrainWaiters.add(drained.resolve); + await drained.promise; + } + } + + #isSessionSettlementPending(ignoreSelectionFenceGeneration?: number): boolean { return ( this.#promptInFlightCount > 0 || this.#agentEventHandlersInFlight > 0 || this.#agentEndPublicationInFlight > 0 || this.#pendingAgentEndContinuationHolds.size > 0 || + this.#pendingSelectionFenceDeferredContinuations(ignoreSelectionFenceGeneration) > 0 || this.#pendingAgentEndEmit !== undefined ); } #resolveSessionSettlement(): void { - if (this.#isSessionSettlementPending() || !this.#sessionSettlementResolve) return; + if (this.#isSessionSettlementPending() || !this.#sessionSettlementResolve) { + this.#wakeScopedSettlementWaiters(); + return; + } const resolve = this.#sessionSettlementResolve; this.#sessionSettlementResolve = undefined; this.#sessionSettlementPromise = undefined; resolve(); + this.#wakeScopedSettlementWaiters(); + } + /** Re-evaluate every scoped settlement waiter (selection-internal drains). */ + #wakeScopedSettlementWaiters(): void { + for (const check of [...this.#scopedSettlementWaiters]) check(); } - async #waitForSessionSettlement(): Promise { - while (this.#isSessionSettlementPending()) { - if (!this.#sessionSettlementPromise) { - const { promise, resolve } = Promise.withResolvers(); - this.#sessionSettlementPromise = promise; - this.#sessionSettlementResolve = resolve; + /** + * Wait for session settlement. `ignoreSelectionFenceGeneration` carries the + * selection's own fence generation for its internal drain only (see + * {@link waitForIdle}); external callers observe every unresolved + * fence-deferred continuation. + * + * A scoped waiter registers a wake callback instead of sharing the strict + * settlement promise: the strict promise may stay pending on a + * fence-deferred continuation (correctly), and the selection that owns that + * fence must not block behind it (#4519). + */ + async #waitForSessionSettlement(ignoreSelectionFenceGeneration?: number): Promise { + while (this.#isSessionSettlementPending(ignoreSelectionFenceGeneration)) { + if (ignoreSelectionFenceGeneration === undefined) { + if (!this.#sessionSettlementPromise) { + const { promise, resolve } = Promise.withResolvers(); + this.#sessionSettlementPromise = promise; + this.#sessionSettlementResolve = resolve; + } + await this.#sessionSettlementPromise; + continue; + } + const wake = Promise.withResolvers(); + const check = () => { + if (!this.#isSessionSettlementPending(ignoreSelectionFenceGeneration)) wake.resolve(); + }; + this.#scopedSettlementWaiters.add(check); + check(); + try { + await wake.promise; + } finally { + this.#scopedSettlementWaiters.delete(check); } - await this.#sessionSettlementPromise; } } @@ -2957,7 +3130,12 @@ export class AgentSession { // A post-prompt continuation runs before its predecessor prompt returns. Its // publication must settle, but session-wide settlement can still be owned by // that predecessor; waiting here would make each prompt wait on the other. - if (!predecessorPromptStillInFlight) await this.#waitForSessionSettlement(); + // Deferrals behind still-pending selection fences are excluded here: they + // cannot settle before their fence resolves, and the fence owner may be + // waiting on this very settlement (#4519). + if (!predecessorPromptStillInFlight) { + await this.#waitForSessionSettlement(this.#turnSettlementDeferralFloor()); + } if (flushError) throw flushError; } @@ -5446,8 +5624,14 @@ export class AgentSession { this.#postPromptTasksPromise = undefined; } - #trackPostPromptTask(task: Promise, lease?: RunResourceProducerLease, leaseTask: Promise = task): void { + #trackPostPromptTask( + task: Promise, + selectionFenceGeneration: number, + lease?: RunResourceProducerLease, + leaseTask: Promise = task, + ): void { this.#postPromptTasks.add(task); + this.#postPromptTaskSelectionFenceGenerations.set(task, selectionFenceGeneration); this.#ensurePostPromptTasksPromise(); if (lease) { lease.track("post_prompt", "agent-session", leaseTask); @@ -5457,6 +5641,7 @@ export class AgentSession { .catch(() => {}) .finally(() => { this.#postPromptTasks.delete(task); + this.#postPromptTaskSelectionFenceGenerations.delete(task); if (this.#postPromptTasks.size === 0) this.#resolvePostPromptTasks(); }); } @@ -5469,8 +5654,14 @@ export class AgentSession { onSkip?: () => void; resourceRunId?: string; leaseTask?: Promise; + selectionFenceGeneration?: number; }, ): Promise { + const selectionFenceGeneration = + options?.selectionFenceGeneration ?? + this.#selectionFenceGenerationContext.getStore() ?? + this.#sessionAdmissionContext.getStore()?.selectionFenceGeneration ?? + this.#selectionFenceGeneration; const delayMs = options?.delayMs ?? 0; const resourceRunId = options?.resourceRunId; const contextualLease = this.#runResourceLeaseContext.getStore(); @@ -5508,12 +5699,17 @@ export class AgentSession { options.onSkip?.(); return; } - await task(signal); + await this.#selectionFenceGenerationContext.run(selectionFenceGeneration, () => task(signal)); }; const scheduled = reservation?.ok ? this.#runResourceLeaseContext.run(reservation.lease, runScheduled) : runScheduled(); - this.#trackPostPromptTask(scheduled, reservation?.ok ? reservation.lease : undefined, options?.leaseTask); + this.#trackPostPromptTask( + scheduled, + selectionFenceGeneration, + reservation?.ok ? reservation.lease : undefined, + options?.leaseTask, + ); return scheduled; } @@ -5543,10 +5739,53 @@ export class AgentSession { rescheduleOnBusy?: boolean; /** Called when the scheduled continuation accepts its run (before agent_start). */ onRunAccepted?: (handle: AttemptRunHandle) => void; + /** Internal causal generation retained when a later continuation is deferred behind selection. */ + selectionFenceGeneration?: number; + /** Internal predecessor publication hold retained across selection-fence deferral. */ + predecessorAgentEndHold?: symbol; + /** Internal predecessor terminal event sequestered while waiting behind selection. */ + deferredPredecessorAgentEnd?: AgentSessionEvent; }): Promise { - const predecessorAgentEndHold = options?.suppressPredecessorAgentEnd - ? this.#reserveDeferredAgentEndForContinuation() - : undefined; + const continuationAdmission = this.#captureScheduledContinuationAdmission(); + const selectionFenceGeneration = + options?.selectionFenceGeneration ?? + this.#selectionFenceGenerationContext.getStore() ?? + this.#sessionAdmissionContext.getStore()?.selectionFenceGeneration ?? + continuationAdmission?.entry.selectionFenceGeneration ?? + this.#selectionFenceGeneration; + if (this.#pendingSelectionFences > 0 && selectionFenceGeneration === this.#selectionFenceGeneration) { + const deferredPredecessorAgentEnd = + options?.deferredPredecessorAgentEnd ?? + (options?.suppressPredecessorAgentEnd + ? this.#claimDeferredAgentEndForContinuation(this.#reserveDeferredAgentEndForContinuation()) + : undefined); + const precedingSelectionFence = this.#selectionFenceTail; + const deferredPromptGeneration = options?.generation ?? this.#promptGeneration; + const deferredScheduling = precedingSelectionFence.then(() => { + try { + this.#scheduleAgentContinue({ + ...options, + generation: deferredPromptGeneration, + selectionFenceGeneration, + deferredPredecessorAgentEnd, + }); + } finally { + // The recursive call synchronously re-reserved its settlement + // markers (predecessor hold / post-prompt task); end the limbo + // window so the counter never spans the continuation run itself. + this.#endSelectionFenceDeferralTracking(selectionFenceGeneration); + } + }); + this.#trackSelectionFenceDeferredContinuation(selectionFenceGeneration, deferredScheduling); + return Promise.resolve(); + } + const predecessorAgentEndHold = + options?.predecessorAgentEndHold ?? + (options?.deferredPredecessorAgentEnd + ? this.#restoreAndReserveDeferredAgentEndForContinuation(options.deferredPredecessorAgentEnd) + : options?.suppressPredecessorAgentEnd + ? this.#reserveDeferredAgentEndForContinuation() + : undefined); let terminalized = false; const skip = ( reason: "generation_changed" | "aborted_signal" | "queue_drained" | "handoff_in_progress" | "terminal_turn", @@ -5566,7 +5805,6 @@ export class AgentSession { options?.onError?.(error); }; const scheduledGeneration = options?.generation; - const continuationAdmission = this.#captureScheduledContinuationAdmission(); let busyReschedules = 0; const scheduleAttempt = (delayMs = options?.delayMs): Promise => { const leaseSettlement = Promise.withResolvers(); @@ -5720,6 +5958,10 @@ export class AgentSession { }, scheduledSignal, continuationAdmission, + { + bypassSelectionFenceGeneration: selectionFenceGeneration, + allowPromptContinuationReentry: true, + }, ); } catch (error) { if (scheduledSignal.aborted) skip("aborted_signal"); @@ -5734,6 +5976,7 @@ export class AgentSession { }, resourceRunId: options?.resourceRunId, leaseTask: leaseSettlement.promise, + selectionFenceGeneration, }, ); }; @@ -5815,8 +6058,13 @@ export class AgentSession { return false; } - #scheduleAutoContinuePrompt(generation: number, requireUnfinishedWork = true, resourceRunId?: string): void { - const predecessorAgentEndHold = this.#reserveDeferredAgentEndForContinuation(); + #scheduleAutoContinuePrompt( + generation: number, + requireUnfinishedWork = true, + resourceRunId?: string, + deferredSelectionFenceGeneration?: number, + deferredPredecessorAgentEnd?: AgentSessionEvent, + ): void { const scheduledGeneration = generation; const continuationAuthorized = async ( signal: AbortSignal, @@ -5851,6 +6099,36 @@ export class AgentSession { return authorized; }; const continuationAdmission = this.#captureScheduledContinuationAdmission(); + const selectionFenceGeneration = + deferredSelectionFenceGeneration ?? + this.#selectionFenceGenerationContext.getStore() ?? + this.#sessionAdmissionContext.getStore()?.selectionFenceGeneration ?? + continuationAdmission?.entry.selectionFenceGeneration ?? + this.#selectionFenceGeneration; + if (this.#pendingSelectionFences > 0 && selectionFenceGeneration === this.#selectionFenceGeneration) { + const predecessorAgentEnd = + deferredPredecessorAgentEnd ?? + this.#claimDeferredAgentEndForContinuation(this.#reserveDeferredAgentEndForContinuation()); + const precedingSelectionFence = this.#selectionFenceTail; + const deferredScheduling = precedingSelectionFence.then(() => { + try { + this.#scheduleAutoContinuePrompt( + generation, + requireUnfinishedWork, + resourceRunId, + selectionFenceGeneration, + predecessorAgentEnd, + ); + } finally { + this.#endSelectionFenceDeferralTracking(selectionFenceGeneration); + } + }); + this.#trackSelectionFenceDeferredContinuation(selectionFenceGeneration, deferredScheduling); + return; + } + const predecessorAgentEndHold = deferredPredecessorAgentEnd + ? this.#restoreAndReserveDeferredAgentEndForContinuation(deferredPredecessorAgentEnd) + : this.#reserveDeferredAgentEndForContinuation(); void this.#schedulePostPromptTask( async signal => { try { @@ -5886,6 +6164,7 @@ export class AgentSession { }, signal, continuationAdmission, + { bypassSelectionFenceGeneration: selectionFenceGeneration }, ); } catch (error) { if (signal.aborted) { @@ -5901,6 +6180,7 @@ export class AgentSession { delayMs: 0, generation: scheduledGeneration, resourceRunId, + selectionFenceGeneration, onSkip: () => { this.#logCompactionContinuationSkipped("auto_continue_prompt", "aborted_signal"); this.#releaseDeferredAgentEndContinuation(predecessorAgentEndHold); @@ -5932,6 +6212,7 @@ export class AgentSession { this.#postPromptTasksAbortController.abort(); this.#postPromptTasksAbortController = new AbortController(); this.#postPromptTasks.clear(); + this.#postPromptTaskSelectionFenceGenerations.clear(); this.#releaseDeferredAgentEndContinuations(); this.#resolveTtsrResume(); this.#resolvePostPromptTasks(); @@ -5968,6 +6249,19 @@ export class AgentSession { } } + /** + * A selection's scoped idle drain must not wait on work parked behind its + * own fence, but it still needs to yield to post-prompt work that predates + * that fence. Otherwise the drain can repeatedly await already-settled + * session work while a timer-backed predecessor task is starved. + */ + async #waitForPostPromptTasksBeforeSelectionFence(selectionFenceGeneration: number): Promise { + const precedingTasks = [...this.#postPromptTasks].filter( + task => (this.#postPromptTaskSelectionFenceGenerations.get(task) ?? 0) < selectionFenceGeneration, + ); + if (precedingTasks.length > 0) await Promise.allSettled(precedingTasks); + } + /** Get TTSR injection payload and clear pending injections. */ #getTtsrInjectionContent(): { content: string; rules: Rule[] } | undefined { if (this.#pendingTtsrInjections.length === 0) return undefined; @@ -7272,17 +7566,30 @@ export class AgentSession { } /** Wait until streaming and session settlement work are fully settled. */ - async waitForIdle(): Promise { + /** + * Wait until streaming and session settlement work are fully settled. + * + * The internal-only `ignoreSelectionFenceGeneration` is used exclusively by + * `setDefaultModelSelection`'s mid-selection drain: the selection must not + * wait on work parked behind its own (or a later) fence — its agent run, + * recovery, and deferrals — because that work is waiting on the selection + * itself (#4519). External callers observe everything. + */ + async waitForIdle(ignoreSelectionFenceGeneration?: number): Promise { while (true) { - await this.agent.waitForIdle(); - await this.#waitForPostPromptRecovery(); - await this.#waitForSessionSettlement(); + if (ignoreSelectionFenceGeneration === undefined) { + await this.agent.waitForIdle(); + await this.#waitForPostPromptRecovery(); + } else { + await this.#waitForPostPromptTasksBeforeSelectionFence(ignoreSelectionFenceGeneration); + } + await this.#waitForSessionSettlement(ignoreSelectionFenceGeneration); if ( !this.agent.state.isStreaming && !this.#retryPromise && !this.#ttsrResumePromise && !this.#postPromptTasksPromise && - !this.#isSessionSettlementPending() + !this.#isSessionSettlementPending(ignoreSelectionFenceGeneration) ) return; } @@ -9443,6 +9750,8 @@ export class AgentSession { */ async prompt(text: string, options?: PromptOptions): Promise { this.#assertRecoveryHydrationPromoted(); + const owner = this.#sessionAdmissionContext.getStore(); + if (owner && !owner.released) throw this.#sessionAdmissionBusyError(); const expandPromptTemplates = options?.expandPromptTemplates ?? true; if (expandPromptTemplates && text.startsWith("/skill:") && !options?.images?.length) { @@ -9497,13 +9806,13 @@ export class AgentSession { assertImagePlaceholdersHavePayload(expandedText, options?.images); const workflowIntentDiff = options?.synthetic ? null : buildWorkflowIntentDiff(expandedText); const claimsGenuineUserIntent = !options?.synthetic && options?.attribution !== "agent"; - const owner = this.#sessionAdmissionContext.getStore(); - if (owner && !owner.released) throw this.#sessionAdmissionBusyError(); const admissionGeneration = this.#promptGeneration; const admissionSignal = options?.preflightSignal ? AbortSignal.any([this.#promptPreflightAbortController.signal, options.preflightSignal]) : this.#promptPreflightAbortController.signal; - await awaitPromptInvocationPreflight(this.#selectionFenceTail, admissionSignal); + if (this.#pendingSelectionFences > 0) { + await awaitPromptInvocationPreflight(this.#selectionFenceTail, admissionSignal); + } const deepInterviewUserIntentEpoch = claimsGenuineUserIntent && !this.isStreaming ? this.#claimDeepInterviewUserIntent() : undefined; @@ -11030,6 +11339,8 @@ export class AgentSession { content: string | (TextContent | ImageContent)[], options?: { deliverAs?: "steer" | "followUp"; + /** Preserve a busy SDK dispatch as queued work across an admission fence. */ + queuedAtDispatch?: boolean; onPreflightAccepted?: () => void; onPreflightAcceptCommit?: () => void | Promise; /** Fired when a queued submission (steering or follow-up) is promoted to its own run (SDK ownership correlation). */ @@ -11039,6 +11350,9 @@ export class AgentSession { }, ): Promise { this.#assertRecoveryHydrationPromoted(); + const owner = this.#sessionAdmissionContext.getStore(); + if (owner && !owner.released) throw this.#sessionAdmissionBusyError(); + this.#assertSessionAdmissionOpen(); if (options?.preflightSignal?.aborted) throw promptPreflightCancelledError(); if (typeof content !== "string" && !Array.isArray(content)) { throw Object.assign(new Error("sendUserMessage requires string or content-array content."), { @@ -11065,55 +11379,145 @@ export class AgentSession { if (images.length === 0) images = undefined; } - if (options?.deliverAs === "followUp") { - if (options.onPreflightAcceptCommit) await options.onPreflightAcceptCommit(); - const queuedFollowUp = await this.#queueFollowUp(text, images, { - claimsGenuineUserIntent: true, - forceOneAtATime: Boolean(options.preflightSignal), - onPromoted: options?.onQueuedPromoted, - sdkRunToken: options.sdkRunToken, - }); - const cancelQueuedFollowUp = () => queuedFollowUp.cancel(); - options.preflightSignal?.addEventListener("abort", cancelQueuedFollowUp, { once: true }); - if (options.preflightSignal?.aborted) cancelQueuedFollowUp(); - options.onPreflightAccepted?.(); - return; - } - if (options?.deliverAs === "steer") { - if (options.onPreflightAcceptCommit) await options.onPreflightAcceptCommit(); - await this.#queueSteer(text, images, { - claimsGenuineUserIntent: true, - onPromoted: options?.onQueuedPromoted, - external: true, - }); - options.onPreflightAccepted?.(); - return; - } + const admissionSignal = options?.preflightSignal + ? AbortSignal.any([this.#promptPreflightAbortController.signal, options.preflightSignal]) + : this.#promptPreflightAbortController.signal; + const preflightCancellationGeneration = this.#promptPreflightCancellationGeneration; + // 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 + // prompt admitted through the same window would classify itself as fresh + // 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; + const hasFollowUpAhead = (): boolean => + this.#activeFollowUpReservationEpochs.size > 0 || + this.agent.snapshotFollowUp().length > 0 || + this.#deferredSdkFollowUps.length > 0; + const followUpAheadAtReservation = queuedPlainPrompt && hasFollowUpAhead(); + const freshAtReservation = + queuedPlainPrompt && + !followUpAheadAtReservation && + !this.agent.state.isStreaming && + !this.#canAutoContinueForSteer(); + const deliverAs = + options?.deliverAs ?? + (queuedPlainPrompt + ? followUpAheadAtReservation + ? "followUp" + : freshAtReservation + ? undefined + : "steer" + : undefined); + const followUpReservationEpoch = deliverAs === "followUp" ? ++this.#followUpReservationEpoch : undefined; + if (followUpReservationEpoch !== undefined) { + this.#activeFollowUpReservationEpochs.add(followUpReservationEpoch); + } + const releaseFollowUpReservation = () => { + if (followUpReservationEpoch === undefined) return; + this.#activeFollowUpReservationEpochs.delete(followUpReservationEpoch); + const waiters = [...this.#followUpReservationDrainWaiters]; + this.#followUpReservationDrainWaiters.clear(); + for (const waiter of waiters) waiter(); + }; + try { + if (this.#pendingSelectionFences > 0) { + await awaitPromptInvocationPreflight(this.#selectionFenceTail, admissionSignal); + } + const assertPreflightStillOpen = () => { + this.#assertSessionAdmissionOpen(); + if ( + options?.preflightSignal?.aborted || + this.#promptPreflightCancellationGeneration !== preflightCancellationGeneration + ) { + throw promptPreflightCancelledError(); + } + }; + assertPreflightStillOpen(); + if (deliverAs === "followUp") { + // Durable enqueue preserves reservation order: while an earlier + // follow-up dispatch is still between its reservation and its own + // durable enqueue, wait for those earlier reservations so this + // dispatch can never enqueue ahead of them. + if (followUpReservationEpoch !== undefined) { + await this.#waitForEarlierFollowUpReservations( + followUpReservationEpoch, + options?.preflightSignal, + admissionSignal, + ); + } + if (options?.onPreflightAcceptCommit) await options.onPreflightAcceptCommit(); + assertPreflightStillOpen(); + const queuedFollowUp = await this.#queueFollowUp(text, images, { + claimsGenuineUserIntent: true, + forceOneAtATime: Boolean(options?.preflightSignal || options?.queuedAtDispatch), + onPromoted: options?.onQueuedPromoted, + sdkRunToken: options?.sdkRunToken, + }); + const cancelQueuedFollowUp = () => queuedFollowUp.cancel(); + options?.preflightSignal?.addEventListener("abort", cancelQueuedFollowUp, { once: true }); + if (options?.preflightSignal?.aborted) cancelQueuedFollowUp(); + options?.onPreflightAccepted?.(); + return; + } + if (deliverAs === "steer") { + if (options?.onPreflightAcceptCommit) await options.onPreflightAcceptCommit(); + assertPreflightStillOpen(); + await this.#queueSteer(text, images, { + claimsGenuineUserIntent: true, + onPromoted: options?.onQueuedPromoted, + external: true, + }); + options?.onPreflightAccepted?.(); + return; + } - // No explicit delivery mode: only a live stream makes prompt() throw - // AgentBusyError, so queue the message as steering while streaming. - // 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) { - if (options?.onPreflightAcceptCommit) await options.onPreflightAcceptCommit(); - await this.#queueSteer(text, images, { - claimsGenuineUserIntent: true, - onPromoted: options?.onQueuedPromoted, - external: true, + // No explicit delivery mode: only a live stream makes prompt() throw + // AgentBusyError, so queue the message as steering while streaming. + // 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) { + if (options?.onPreflightAcceptCommit) await options.onPreflightAcceptCommit(); + assertPreflightStillOpen(); + await this.#queueSteer(text, images, { + claimsGenuineUserIntent: true, + onPromoted: options?.onQueuedPromoted, + external: true, + }); + options?.onPreflightAccepted?.(); + return; + } + + // Use prompt() with expandPromptTemplates: false to skip command handling and template expansion + let queuedPromotionFired = false; + const fireQueuedPromotion = () => { + if (!freshAtReservation || queuedPromotionFired) return; + queuedPromotionFired = true; + options?.onQueuedPromoted?.(); + }; + await this.prompt(text, { + expandPromptTemplates: false, + images, + onPreflightAccepted: () => { + options?.onPreflightAccepted?.(); + fireQueuedPromotion(); + }, + onPreflightAcceptCommit: + options?.onPreflightAcceptCommit || freshAtReservation + ? async () => { + if (options?.onPreflightAcceptCommit) await options.onPreflightAcceptCommit(); + else options?.onPreflightAccepted?.(); + assertPreflightStillOpen(); + fireQueuedPromotion(); + } + : undefined, + preflightSignal: options?.preflightSignal, }); - options?.onPreflightAccepted?.(); - return; + } finally { + releaseFollowUpReservation(); } - - // Use prompt() with expandPromptTemplates: false to skip command handling and template expansion - await this.prompt(text, { - expandPromptTemplates: false, - images, - onPreflightAccepted: options?.onPreflightAccepted, - onPreflightAcceptCommit: options?.onPreflightAcceptCommit, - preflightSignal: options?.preflightSignal, - }); } /** @@ -11492,6 +11896,7 @@ export class AgentSession { this.#markRetryReplayUnsafe(); this.abortRetry(); this.#promptGeneration++; + this.#promptPreflightCancellationGeneration++; this.#promptPreflightAbortController.abort(); this.#promptPreflightAbortController = new AbortController(); this.#scheduledHiddenNextTurnGeneration = undefined; @@ -11609,6 +12014,7 @@ export class AgentSession { * reset for the next admission. No run handle exists for a preflight prompt. */ cancelPendingPreflightForTerminalAbort(): void { + this.#promptPreflightCancellationGeneration++; this.#promptPreflightAbortController.abort(); this.#promptPreflightAbortController = new AbortController(); } @@ -11847,6 +12253,7 @@ export class AgentSession { // resetRetryReplaySafety) advances the epoch before minting, so the // next turn after the abort gets a distinct (lineage, epoch) and is // never captured by this scope (review thread P2). + this.#promptPreflightCancellationGeneration++; this.#promptPreflightAbortController.abort(); this.#promptPreflightAbortController = new AbortController(); } @@ -13293,6 +13700,12 @@ export class AgentSession { const expectedSessionId = this.sessionId; const priorSelectionFence = this.#selectionFenceTail; const selectionFence = Promise.withResolvers(); + this.#selectionFenceGeneration += 1; + this.#pendingSelectionFences += 1; + const selectionFenceGeneration = this.#selectionFenceGeneration; + if (this.#oldestPendingSelectionFenceGeneration === 0) { + this.#oldestPendingSelectionFenceGeneration = selectionFenceGeneration; + } this.#selectionFenceTail = priorSelectionFence.then(() => selectionFence.promise); void this.#selectionFenceTail.catch(() => {}); try { @@ -13322,7 +13735,7 @@ export class AgentSession { undefined, { allowDuringClosing: true }, ); - await this.waitForIdle(); + await this.waitForIdle(selectionFenceGeneration); return await this.#withSessionAdmission( "selection", async () => { @@ -13408,6 +13821,11 @@ export class AgentSession { ); } finally { selectionFence.resolve(); + this.#pendingSelectionFences -= 1; + if (this.#oldestPendingSelectionFenceGeneration === selectionFenceGeneration) { + this.#oldestPendingSelectionFenceGeneration = 0; + } + this.#resolveSessionSettlement(); } } /** @@ -18168,6 +18586,7 @@ export class AgentSession { } if (!preflightAccepted) { await seam?.onPreflightAccepted?.(); + if (seam?.signal?.aborted) throw promptPreflightCancelledError(); preflightAccepted = true; } await this.agent.prompt(messages, options); diff --git a/packages/coding-agent/test/agent-session-auto-compaction-continue.test.ts b/packages/coding-agent/test/agent-session-auto-compaction-continue.test.ts index a26d6107e2..6b3d62e040 100644 --- a/packages/coding-agent/test/agent-session-auto-compaction-continue.test.ts +++ b/packages/coding-agent/test/agent-session-auto-compaction-continue.test.ts @@ -730,6 +730,179 @@ describe("AgentSession auto-compaction continuation", () => { expect(session.model).toBe(selectionModel); }); + it("does not deadlock default selection waiting for an ownerless emergency continuation", async () => { + const releaseStartupBarrier = Promise.withResolvers(); + session.extendStartupTurnBarrier(releaseStartupBarrier.promise); + const compactionEnded = Promise.withResolvers(); + const runtimeSignals = getRuntimeSignals(); + const pushRuntimeSignal = runtimeSignals.push.bind(runtimeSignals); + vi.spyOn(runtimeSignals, "push").mockImplementation((...signals) => { + const length = pushRuntimeSignal(...signals); + if (signals.includes("compaction:end:ok")) compactionEnded.resolve(); + return length; + }); + const order: string[] = []; + const promptSpy = vi.spyOn(session.agent, "prompt").mockImplementation(async () => { + order.push("continuation"); + }); + const currentModel = session.model; + if (!currentModel) throw new Error("Expected session model"); + const selectionModel = { ...currentModel, id: "ownerless-selection-model" }; + const selectionValidated = Promise.withResolvers(); + const originalGetApiKey = modelRegistry.getApiKey.bind(modelRegistry); + vi.spyOn(modelRegistry, "getApiKey").mockImplementation(async (model, ...args) => { + const apiKey = await originalGetApiKey(model, ...args); + if (model === selectionModel) selectionValidated.resolve(); + return apiKey; + }); + const message = assistantMessage(); + sessionManager.appendMessage(message); + session.agent.emitExternalEvent({ type: "message_end", message }); + session.agent.emitExternalEvent({ type: "agent_end", messages: [message] }); + await compactionEnded.promise; + + const selection = session.setDefaultModelSelection(selectionModel, undefined, { + onAfterMutation: () => order.push("selection"), + }); + await selectionValidated.promise; + for (let i = 0; i < 10; i++) await Promise.resolve(); + releaseStartupBarrier.resolve(); + await selection; + await session.waitForIdle(); + + expect(promptSpy).toHaveBeenCalledTimes(1); + expect(order).toEqual(["continuation", "selection"]); + expect(session.model).toBe(selectionModel); + }); + + it("keeps an ownerless emergency continuation scheduled later behind selection", async () => { + const currentModel = session.model; + if (!currentModel) throw new Error("Expected session model"); + const selectionModel = { ...currentModel, id: "selection-before-ownerless-model" }; + const selectionValidationStarted = Promise.withResolvers(); + const releaseSelectionValidation = Promise.withResolvers(); + const originalGetApiKey = modelRegistry.getApiKey.bind(modelRegistry); + vi.spyOn(modelRegistry, "getApiKey").mockImplementation(async (model, ...args) => { + if (model === selectionModel) { + selectionValidationStarted.resolve(); + await releaseSelectionValidation.promise; + } + return originalGetApiKey(model, ...args); + }); + const order: string[] = []; + const promptSpy = vi.spyOn(session.agent, "prompt").mockImplementation(async () => { + order.push("continuation"); + }); + const selection = session.setDefaultModelSelection(selectionModel, undefined, { + onAfterMutation: () => order.push("selection"), + }); + await selectionValidationStarted.promise; + const message = assistantMessage(); + sessionManager.appendMessage(message); + session.agent.emitExternalEvent({ type: "message_end", message }); + session.agent.emitExternalEvent({ type: "agent_end", messages: [message] }); + releaseSelectionValidation.resolve(); + + await selection; + await session.waitForIdle(); + + expect(promptSpy).toHaveBeenCalledTimes(1); + expect(order).toEqual(["selection", "continuation"]); + expect(session.model).toBe(selectionModel); + }); + it("keeps waitForIdle pending while a deferred continuation waits behind selection", async () => { + const currentModel = session.model; + if (!currentModel) throw new Error("Expected session model"); + const selectionModel = { ...currentModel, id: "selection-deferred-idle-model" }; + const selectionValidationStarted = Promise.withResolvers(); + const releaseSelectionValidation = Promise.withResolvers(); + const originalGetApiKey = modelRegistry.getApiKey.bind(modelRegistry); + vi.spyOn(modelRegistry, "getApiKey").mockImplementation(async (model, ...args) => { + if (model === selectionModel) { + selectionValidationStarted.resolve(); + await releaseSelectionValidation.promise; + } + return originalGetApiKey(model, ...args); + }); + const order: string[] = []; + const promptSpy = vi.spyOn(session.agent, "prompt").mockImplementation(async () => { + order.push("continuation"); + }); + const selection = session.setDefaultModelSelection(selectionModel, undefined, { + onAfterMutation: () => order.push("selection"), + }); + await selectionValidationStarted.promise; + const message = assistantMessage(); + sessionManager.appendMessage(message); + session.agent.emitExternalEvent({ type: "message_end", message }); + session.agent.emitExternalEvent({ type: "agent_end", messages: [message] }); + // The continuation is now parked behind the pending selection fence, + // having claimed the predecessor agent_end: external waitForIdle must + // not report the session idle while that continuation is still waiting + // on the fence, even once every other in-flight work settles. + let idleReported = false; + const idle = session.waitForIdle().then(() => { + idleReported = true; + }); + for (let i = 0; i < 20; i++) await Promise.resolve(); + expect(idleReported).toBe(false); + releaseSelectionValidation.resolve(); + + await selection; + await idle; + await session.waitForIdle(); + + expect(promptSpy).toHaveBeenCalledTimes(1); + expect(order).toEqual(["selection", "continuation"]); + expect(session.model).toBe(selectionModel); + }); + it("does not deadlock when a second selection reserves while a continuation waits behind the first", async () => { + const currentModel = session.model; + if (!currentModel) throw new Error("Expected session model"); + const firstModel = { ...currentModel, id: "overlapping-selection-one" }; + const secondModel = { ...currentModel, id: "overlapping-selection-two" }; + const firstValidationStarted = Promise.withResolvers(); + const releaseFirstValidation = Promise.withResolvers(); + const originalGetApiKey = modelRegistry.getApiKey.bind(modelRegistry); + vi.spyOn(modelRegistry, "getApiKey").mockImplementation(async (model, ...args) => { + if (model === firstModel) { + firstValidationStarted.resolve(); + await releaseFirstValidation.promise; + } + return originalGetApiKey(model, ...args); + }); + const order: string[] = []; + const promptSpy = vi.spyOn(session.agent, "prompt").mockImplementation(async () => { + order.push("continuation"); + }); + // Selection A reserves generation 1 and parks inside credential probing. + const first = session.setDefaultModelSelection(firstModel, undefined, { + onAfterMutation: () => order.push("selection-one"), + }); + await firstValidationStarted.promise; + // A continuation defers behind A's fence. + const message = assistantMessage(); + sessionManager.appendMessage(message); + session.agent.emitExternalEvent({ type: "message_end", message }); + session.agent.emitExternalEvent({ type: "agent_end", messages: [message] }); + for (let i = 0; i < 10; i++) await Promise.resolve(); + // Selection B reserves generation 2 while A is still parked and the + // generation-1 continuation is still deferred behind A's fence. + const second = session.setDefaultModelSelection(secondModel, undefined, { + onAfterMutation: () => order.push("selection-two"), + }); + releaseFirstValidation.resolve(); + + await Promise.all([first, second]); + await session.waitForIdle(); + + expect(promptSpy).toHaveBeenCalledTimes(1); + expect(order).toContain("selection-one"); + expect(order).toContain("selection-two"); + expect(order).toContain("continuation"); + expect(session.model).toBe(secondModel); + }); + it("reschedules an AgentBusyError racing the queued-followup continue until delivery", async () => { session.agent.followUp({ role: "custom", diff --git a/packages/coding-agent/test/agent-session-before-agent-start-attribution.test.ts b/packages/coding-agent/test/agent-session-before-agent-start-attribution.test.ts index 2504377b58..0b8a3486d7 100644 --- a/packages/coding-agent/test/agent-session-before-agent-start-attribution.test.ts +++ b/packages/coding-agent/test/agent-session-before-agent-start-attribution.test.ts @@ -341,6 +341,512 @@ describe("AgentSession before_agent_start attribution fallback", () => { await successor; expect(session.model).toBe(selectionModel); }); + it("fences SDK prompt, steer, and follow-up ingress behind an earlier selection", async () => { + const { emitBeforeAgentStart } = createSession(); + const currentModel = session.model; + if (!currentModel) throw new Error("Expected session model"); + const selectionModel = { ...currentModel, provider: "selection-provider", id: "selection-model" }; + authStorage?.setRuntimeApiKey(selectionModel.provider, "selection-key"); + const preflightStarted = Promise.withResolvers(); + const releasePreflight = Promise.withResolvers(); + emitBeforeAgentStart.mockImplementationOnce(async () => { + preflightStarted.resolve(); + await releasePreflight.promise; + return undefined; + }); + const order: string[] = []; + + const activePrompt = session.prompt("active prompt"); + await preflightStarted.promise; + const selection = session.setDefaultModelSelection(selectionModel, undefined, { + onAfterMutation: () => order.push("selection"), + }); + const sdkPrompt = session.sendUserMessage("SDK prompt", { + onPreflightAccepted: () => order.push("prompt"), + }); + const sdkSteer = session.sendUserMessage("SDK steer", { + deliverAs: "steer", + onPreflightAccepted: () => order.push("steer"), + }); + const sdkFollowUp = session.sendUserMessage("SDK follow-up", { + deliverAs: "followUp", + onPreflightAccepted: () => order.push("followUp"), + }); + await Promise.resolve(); + await Promise.resolve(); + + expect(order).toEqual([]); + expect(session.agent.hasQueuedMessages()).toBe(false); + releasePreflight.resolve(); + await Promise.all([activePrompt, selection, sdkPrompt, sdkSteer, sdkFollowUp]); + + expect(order[0]).toBe("selection"); + expect(order).toEqual(expect.arrayContaining(["selection", "prompt", "steer", "followUp"])); + expect(session.model).toBe(selectionModel); + }); + it("promotes an SDK prompt queued at dispatch when no predecessor remains after selection", async () => { + createSession(); + const currentModel = session.model; + if (!currentModel) throw new Error("Expected session model"); + const selectionModel = { ...currentModel, provider: "selection-provider", id: "selection-model" }; + authStorage?.setRuntimeApiKey(selectionModel.provider, "selection-key"); + const selectionValidationStarted = Promise.withResolvers(); + const releaseSelectionValidation = Promise.withResolvers(); + const originalGetApiKey = modelRegistry.getApiKey.bind(modelRegistry); + vi.spyOn(modelRegistry, "getApiKey").mockImplementation(async (model, ...args) => { + if (model === selectionModel) { + selectionValidationStarted.resolve(); + await releaseSelectionValidation.promise; + } + return originalGetApiKey(model, ...args); + }); + const order: string[] = []; + const promoted = Promise.withResolvers(); + const selection = session.setDefaultModelSelection(selectionModel, undefined, { + onAfterMutation: () => order.push("selection"), + }); + await selectionValidationStarted.promise; + const sdkPrompt = session.sendUserMessage("queued SDK prompt", { + queuedAtDispatch: true, + onPreflightAccepted: () => order.push("accepted"), + onPreflightAcceptCommit: () => { + order.push("committed"); + }, + onQueuedPromoted: () => { + order.push("promoted"); + promoted.resolve(); + }, + }); + releaseSelectionValidation.resolve(); + + await selection; + await sdkPrompt; + await promoted.promise; + await session.waitForIdle(); + + expect(order).toEqual(["selection", "committed", "promoted"]); + expect(session.agent.hasQueuedMessages()).toBe(false); + expect(session.model).toBe(selectionModel); + }); + it("keeps an explicit SDK follow-up queued across a selection fence", async () => { + createSession(); + const currentModel = session.model; + if (!currentModel) throw new Error("Expected session model"); + const selectionModel = { ...currentModel, provider: "selection-provider", id: "selection-model" }; + authStorage?.setRuntimeApiKey(selectionModel.provider, "selection-key"); + const selectionValidationStarted = Promise.withResolvers(); + const releaseSelectionValidation = Promise.withResolvers(); + const originalGetApiKey = modelRegistry.getApiKey.bind(modelRegistry); + vi.spyOn(modelRegistry, "getApiKey").mockImplementation(async (model, ...args) => { + if (model === selectionModel) { + selectionValidationStarted.resolve(); + await releaseSelectionValidation.promise; + } + return originalGetApiKey(model, ...args); + }); + const accepted = Promise.withResolvers(); + const selection = session.setDefaultModelSelection(selectionModel, undefined); + await selectionValidationStarted.promise; + const sdkFollowUp = session.sendUserMessage("queued SDK follow-up", { + queuedAtDispatch: true, + deliverAs: "followUp", + onPreflightAccepted: () => accepted.resolve(), + }); + releaseSelectionValidation.resolve(); + + await selection; + await sdkFollowUp; + await accepted.promise; + + expect(session.pendingMessageCounts).toEqual({ steering: 0, followUp: 1, nextTurn: 0 }); + expect(session.model).toBe(selectionModel); + }); + it("keeps a later queued SDK prompt behind an earlier follow-up", async () => { + createSession(); + await session.prompt("seed assistant tail"); + await session.waitForIdle(); + const releaseStartupBarrier = Promise.withResolvers(); + session.extendStartupTurnBarrier(releaseStartupBarrier.promise); + const currentModel = session.model; + if (!currentModel) throw new Error("Expected session model"); + const selectionModel = { ...currentModel, provider: "selection-provider", id: "selection-model" }; + authStorage?.setRuntimeApiKey(selectionModel.provider, "selection-key"); + const selectionValidationStarted = Promise.withResolvers(); + const releaseSelectionValidation = Promise.withResolvers(); + const originalGetApiKey = modelRegistry.getApiKey.bind(modelRegistry); + vi.spyOn(modelRegistry, "getApiKey").mockImplementation(async (model, ...args) => { + if (model === selectionModel) { + selectionValidationStarted.resolve(); + await releaseSelectionValidation.promise; + } + return originalGetApiKey(model, ...args); + }); + const promoted: string[] = []; + const selection = session.setDefaultModelSelection(selectionModel, undefined); + await selectionValidationStarted.promise; + const first = session.sendUserMessage("first follow-up", { + queuedAtDispatch: true, + deliverAs: "followUp", + onQueuedPromoted: () => promoted.push("follow-up"), + }); + const second = session.sendUserMessage("later plain prompt", { + queuedAtDispatch: true, + onQueuedPromoted: () => promoted.push("plain"), + }); + releaseSelectionValidation.resolve(); + + await Promise.all([selection, first, second]); + expect(session.pendingMessageCounts).toEqual({ steering: 0, followUp: 2, nextTurn: 0 }); + const queuedFollowUps = session.agent.snapshotFollowUp(); + expect(queuedFollowUps[0]).toMatchObject({ + role: "user", + content: expect.arrayContaining([{ type: "text", text: "first follow-up" }]), + }); + expect(queuedFollowUps[1]).toMatchObject({ + role: "user", + content: expect.arrayContaining([{ type: "text", text: "later plain prompt" }]), + }); + releaseStartupBarrier.resolve(); + await session.waitForIdle(); + + expect(promoted[0]).toBe("follow-up"); + }); + it("reserves an earlier follow-up before classifying a later prompt behind the same fence", async () => { + createSession(); + await session.prompt("seed assistant tail"); + await session.waitForIdle(); + const releaseStartupBarrier = Promise.withResolvers(); + session.extendStartupTurnBarrier(releaseStartupBarrier.promise); + const currentModel = session.model; + if (!currentModel) throw new Error("Expected session model"); + const selectionModel = { ...currentModel, provider: "selection-provider", id: "selection-model" }; + authStorage?.setRuntimeApiKey(selectionModel.provider, "selection-key"); + const selectionValidationStarted = Promise.withResolvers(); + const releaseSelectionValidation = Promise.withResolvers(); + const followUpCommitStarted = Promise.withResolvers(); + const releaseFollowUpCommit = Promise.withResolvers(); + const originalGetApiKey = modelRegistry.getApiKey.bind(modelRegistry); + vi.spyOn(modelRegistry, "getApiKey").mockImplementation(async (model, ...args) => { + if (model === selectionModel) { + selectionValidationStarted.resolve(); + await releaseSelectionValidation.promise; + } + return originalGetApiKey(model, ...args); + }); + const promoted: string[] = []; + const selection = session.setDefaultModelSelection(selectionModel, undefined); + await selectionValidationStarted.promise; + // The explicit follow-up parks inside its asynchronous durable acceptance + // (onPreflightAcceptCommit) before #queueFollowUp records anything. + const first = session.sendUserMessage("first follow-up", { + queuedAtDispatch: true, + deliverAs: "followUp", + onPreflightAcceptCommit: async () => { + followUpCommitStarted.resolve(); + await releaseFollowUpCommit.promise; + }, + onQueuedPromoted: () => promoted.push("follow-up"), + }); + // Let the selection fence settle so the follow-up dispatch reaches (and + // parks inside) its durable commit before the later prompt classifies. + releaseSelectionValidation.resolve(); + await selection; + await followUpCommitStarted.promise; + // The later plain prompt classifies while the earlier follow-up is still + // awaiting its commit: it must observe the reserved follow-up ahead and + // queue as a follow-up behind it rather than starting a fresh run. + const second = session.sendUserMessage("later plain prompt", { + queuedAtDispatch: true, + onQueuedPromoted: () => promoted.push("plain"), + }); + releaseFollowUpCommit.resolve(); + + await Promise.all([first, second]); + expect(session.pendingMessageCounts).toEqual({ steering: 0, followUp: 2, nextTurn: 0 }); + const queuedFollowUps = session.agent.snapshotFollowUp(); + expect(queuedFollowUps[0]).toMatchObject({ + role: "user", + content: expect.arrayContaining([{ type: "text", text: "first follow-up" }]), + }); + expect(queuedFollowUps[1]).toMatchObject({ + role: "user", + content: expect.arrayContaining([{ type: "text", text: "later plain prompt" }]), + }); + releaseStartupBarrier.resolve(); + await session.waitForIdle(); + + expect(promoted[0]).toBe("follow-up"); + }); + it("releases a follow-up reservation when terminal abort cancels the selection-fence wait", async () => { + createSession(); + const currentModel = session.model; + if (!currentModel) throw new Error("Expected session model"); + const selectionModel = { ...currentModel, provider: "selection-provider", id: "selection-model" }; + authStorage?.setRuntimeApiKey(selectionModel.provider, "selection-key"); + const selectionValidationStarted = Promise.withResolvers(); + const releaseSelectionValidation = Promise.withResolvers(); + const originalGetApiKey = modelRegistry.getApiKey.bind(modelRegistry); + vi.spyOn(modelRegistry, "getApiKey").mockImplementation(async (model, ...args) => { + if (model === selectionModel) { + selectionValidationStarted.resolve(); + await releaseSelectionValidation.promise; + } + return originalGetApiKey(model, ...args); + }); + // Park a selection fence so the follow-up dispatch parks inside its + // selection-fence wait holding an un-enqueued reservation. + const selection = session.setDefaultModelSelection(selectionModel, undefined); + await selectionValidationStarted.promise; + const cancelled = session.sendUserMessage("cancelled follow-up", { + queuedAtDispatch: true, + deliverAs: "followUp", + }); + for (let i = 0; i < 10; i++) await Promise.resolve(); + // Cancel the preflight while the dispatch still waits on the fence: the + // reservation must not outlive the rejection. + session.cancelPendingPreflightForTerminalAbort(); + + await expect(cancelled).rejects.toMatchObject({ + code: "busy", + message: "Prompt preflight was cancelled before execution.", + }); + releaseSelectionValidation.resolve(); + await selection; + + // A later plain queued prompt must classify as fresh (no leaked + // follow-up reservation ahead of it) and start its own run. + let promoted = false; + await session.sendUserMessage("later plain prompt", { + queuedAtDispatch: true, + onQueuedPromoted: () => { + promoted = true; + }, + }); + await session.waitForIdle(); + + expect(promoted).toBe(true); + expect(session.pendingMessageCounts.followUp).toBe(0); + }); + it("rejects fresh queued SDK promotion when disposal starts during durable acceptance", async () => { + createSession(); + const providerPrompt = vi.spyOn(session.agent, "prompt"); + const currentModel = session.model; + if (!currentModel) throw new Error("Expected session model"); + const selectionModel = { ...currentModel, provider: "selection-provider", id: "selection-model" }; + authStorage?.setRuntimeApiKey(selectionModel.provider, "selection-key"); + const selectionValidationStarted = Promise.withResolvers(); + const releaseSelectionValidation = Promise.withResolvers(); + const durableAcceptanceStarted = Promise.withResolvers(); + const releaseDurableAcceptance = Promise.withResolvers(); + const originalGetApiKey = modelRegistry.getApiKey.bind(modelRegistry); + vi.spyOn(modelRegistry, "getApiKey").mockImplementation(async (model, ...args) => { + if (model === selectionModel) { + selectionValidationStarted.resolve(); + await releaseSelectionValidation.promise; + } + return originalGetApiKey(model, ...args); + }); + let promoted = false; + const selection = session.setDefaultModelSelection(selectionModel, undefined); + await selectionValidationStarted.promise; + const sdkPrompt = session.sendUserMessage("queued SDK prompt", { + queuedAtDispatch: true, + onPreflightAcceptCommit: async () => { + durableAcceptanceStarted.resolve(); + await releaseDurableAcceptance.promise; + }, + onQueuedPromoted: () => { + promoted = true; + }, + }); + releaseSelectionValidation.resolve(); + await selection; + await durableAcceptanceStarted.promise; + const disposal = session.dispose(); + releaseDurableAcceptance.resolve(); + + await expect(sdkPrompt).rejects.toMatchObject({ + code: "busy", + message: "Prompt preflight was cancelled before execution.", + }); + await disposal; + + expect(promoted).toBe(false); + expect(providerPrompt).not.toHaveBeenCalled(); + }); + it("rejects fresh queued SDK promotion when terminal abort cancels durable acceptance", async () => { + createSession(); + const providerPrompt = vi.spyOn(session.agent, "prompt"); + const currentModel = session.model; + if (!currentModel) throw new Error("Expected session model"); + const selectionModel = { ...currentModel, provider: "selection-provider", id: "selection-model" }; + authStorage?.setRuntimeApiKey(selectionModel.provider, "selection-key"); + const selectionValidationStarted = Promise.withResolvers(); + const releaseSelectionValidation = Promise.withResolvers(); + const durableAcceptanceStarted = Promise.withResolvers(); + const releaseDurableAcceptance = Promise.withResolvers(); + const originalGetApiKey = modelRegistry.getApiKey.bind(modelRegistry); + vi.spyOn(modelRegistry, "getApiKey").mockImplementation(async (model, ...args) => { + if (model === selectionModel) { + selectionValidationStarted.resolve(); + await releaseSelectionValidation.promise; + } + return originalGetApiKey(model, ...args); + }); + let promoted = false; + const selection = session.setDefaultModelSelection(selectionModel, undefined); + await selectionValidationStarted.promise; + const sdkPrompt = session.sendUserMessage("queued SDK prompt", { + queuedAtDispatch: true, + onPreflightAcceptCommit: async () => { + durableAcceptanceStarted.resolve(); + await releaseDurableAcceptance.promise; + }, + onQueuedPromoted: () => { + promoted = true; + }, + }); + releaseSelectionValidation.resolve(); + await selection; + await durableAcceptanceStarted.promise; + session.cancelPendingPreflightForTerminalAbort(); + releaseDurableAcceptance.resolve(); + + await expect(sdkPrompt).rejects.toMatchObject({ + code: "busy", + message: "Prompt preflight was cancelled before execution.", + }); + expect(promoted).toBe(false); + expect(providerPrompt).not.toHaveBeenCalled(); + }); + it("does not start a provider when terminal abort races after requester promotion", async () => { + createSession(); + const providerPrompt = vi.spyOn(session.agent, "prompt"); + let promotions = 0; + const sdkPrompt = session.sendUserMessage("queued SDK prompt", { + queuedAtDispatch: true, + onPreflightAcceptCommit: async () => {}, + onQueuedPromoted: () => { + promotions += 1; + session.cancelPendingPreflightForTerminalAbort(); + }, + }); + + await expect(sdkPrompt).rejects.toMatchObject({ + code: "busy", + message: "Prompt preflight was cancelled before execution.", + }); + expect(promotions).toBe(1); + expect(providerPrompt).not.toHaveBeenCalled(); + }); + for (const deliverAs of ["steer", "followUp"] as const) { + it(`rejects SDK ${deliverAs} when terminal abort cancels durable acceptance`, async () => { + createSession(); + const durableAcceptanceStarted = Promise.withResolvers(); + const releaseDurableAcceptance = Promise.withResolvers(); + let accepted = false; + const sdkMessage = session.sendUserMessage(`queued SDK ${deliverAs}`, { + deliverAs, + onPreflightAcceptCommit: async () => { + durableAcceptanceStarted.resolve(); + await releaseDurableAcceptance.promise; + }, + onPreflightAccepted: () => { + accepted = true; + }, + }); + await durableAcceptanceStarted.promise; + session.cancelPendingPreflightForTerminalAbort(); + releaseDurableAcceptance.resolve(); + + await expect(sdkMessage).rejects.toMatchObject({ + code: "busy", + message: "Prompt preflight was cancelled before execution.", + }); + expect(accepted).toBe(false); + expect(session.agent.hasQueuedMessages()).toBe(false); + }); + } + it("rejects fenced SDK ingress when disposal drains the earlier selection", async () => { + createSession(); + const currentModel = session.model; + if (!currentModel) throw new Error("Expected session model"); + const selectionModel = { ...currentModel, provider: "selection-provider", id: "selection-model" }; + authStorage?.setRuntimeApiKey(selectionModel.provider, "selection-key"); + const selectionValidationStarted = Promise.withResolvers(); + const releaseSelectionValidation = Promise.withResolvers(); + const originalGetApiKey = modelRegistry.getApiKey.bind(modelRegistry); + vi.spyOn(modelRegistry, "getApiKey").mockImplementation(async (model, ...args) => { + if (model === selectionModel) { + selectionValidationStarted.resolve(); + await releaseSelectionValidation.promise; + } + return originalGetApiKey(model, ...args); + }); + let accepted = false; + + const selection = session.setDefaultModelSelection(selectionModel, undefined); + await selectionValidationStarted.promise; + const sdkFollowUp = session.sendUserMessage("reject during disposal", { + deliverAs: "followUp", + onPreflightAccepted: () => { + accepted = true; + }, + }); + const disposal = session.dispose(); + releaseSelectionValidation.resolve(); + + await selection; + await expect(sdkFollowUp).rejects.toMatchObject({ name: "AgentBusyError", code: "busy" }); + await disposal; + expect(accepted).toBe(false); + expect(session.agent.hasQueuedMessages()).toBe(false); + }); + for (const deliverAs of ["steer", "followUp"] as const) { + it(`rejects SDK ${deliverAs} when disposal starts during durable acceptance`, async () => { + createSession(); + const currentModel = session.model; + if (!currentModel) throw new Error("Expected session model"); + const selectionModel = { ...currentModel, provider: "selection-provider", id: "selection-model" }; + authStorage?.setRuntimeApiKey(selectionModel.provider, "selection-key"); + const selectionValidationStarted = Promise.withResolvers(); + const releaseSelectionValidation = Promise.withResolvers(); + const originalGetApiKey = modelRegistry.getApiKey.bind(modelRegistry); + vi.spyOn(modelRegistry, "getApiKey").mockImplementation(async (model, ...args) => { + if (model === selectionModel) { + selectionValidationStarted.resolve(); + await releaseSelectionValidation.promise; + } + return originalGetApiKey(model, ...args); + }); + const acceptanceStarted = Promise.withResolvers(); + const releaseAcceptance = Promise.withResolvers(); + let accepted = false; + + const selection = session.setDefaultModelSelection(selectionModel, undefined); + await selectionValidationStarted.promise; + const sdkMessage = session.sendUserMessage(`reject ${deliverAs} during acceptance`, { + deliverAs, + onPreflightAcceptCommit: async () => { + acceptanceStarted.resolve(); + await releaseAcceptance.promise; + }, + onPreflightAccepted: () => { + accepted = true; + }, + }); + releaseSelectionValidation.resolve(); + await acceptanceStarted.promise; + const disposal = session.dispose(); + releaseAcceptance.resolve(); + + await selection; + await expect(sdkMessage).rejects.toMatchObject({ name: "AgentBusyError", code: "busy" }); + await disposal; + expect(accepted).toBe(false); + expect(session.agent.hasQueuedMessages()).toBe(false); + }); + } it("cancels a queued prompt without starving later admission", async () => { createSession(); const currentModel = session.model; diff --git a/packages/coding-agent/test/agent-session-manual-retry.test.ts b/packages/coding-agent/test/agent-session-manual-retry.test.ts index 1454720c02..e97147983c 100644 --- a/packages/coding-agent/test/agent-session-manual-retry.test.ts +++ b/packages/coding-agent/test/agent-session-manual-retry.test.ts @@ -1,4 +1,4 @@ -import { afterEach, beforeEach, describe, expect, it } from "bun:test"; +import { afterEach, beforeEach, describe, expect, it, spyOn } from "bun:test"; import * as fs from "node:fs/promises"; import * as path from "node:path"; import { Agent, type AgentMessage } from "@gajae-code/agent-core"; @@ -121,6 +121,48 @@ describe("AgentSession manual retry", () => { expect(lastAgentMessage(session).content).toContainEqual({ type: "text", text: "recovered after manual retry" }); }); + it("does not start a retry deferred behind selection after abort", async () => { + const model = getTestModel(); + const mock = createMockModel({ responses: [{ throw: "manual retry test failure" }] }); + const agent = new Agent({ + getApiKey: provider => `${provider}-test-key`, + initialState: { model, systemPrompt: ["Test"], tools: [], messages: [] }, + streamFn: mock.stream, + }); + const modelRegistry = new ModelRegistry(authStorage); + session = new AgentSession({ + agent, + sessionManager: SessionManager.inMemory(), + settings: Settings.isolated({ "compaction.enabled": false, "retry.enabled": false }), + modelRegistry, + }); + session.subscribe(() => {}); + await session.prompt("fail once"); + await session.waitForIdle(); + const selectionModel = { ...model, id: "selection-holds-retry" }; + const selectionValidationStarted = Promise.withResolvers(); + const releaseSelectionValidation = Promise.withResolvers(); + const originalGetApiKey = modelRegistry.getApiKey.bind(modelRegistry); + spyOn(modelRegistry, "getApiKey").mockImplementation(async (selectedModel, ...args) => { + if (selectedModel === selectionModel) { + selectionValidationStarted.resolve(); + await releaseSelectionValidation.promise; + } + return originalGetApiKey(selectedModel, ...args); + }); + + const selection = session.setDefaultModelSelection(selectionModel, undefined); + await selectionValidationStarted.promise; + await expect(session.retry()).resolves.toBe(true); + await session.abort(); + releaseSelectionValidation.resolve(); + await selection; + await session.waitForIdle(); + + expect(mock.calls.length).toBe(1); + expect(session.model).toBe(selectionModel); + }); + it("continues from a persisted user tail left by a process crash", async () => { const model = getTestModel(); const mock = createMockModel({