From b441398ffee9e60784937f5a0623e7d1fc067bae Mon Sep 17 00:00:00 2001 From: Yeachan Heo Date: Fri, 14 Aug 2026 03:53:32 +0000 Subject: [PATCH 1/5] fix(session): fence SDK queued prompt ingress The first Ultragoal boundary cohort found that SDK turn.prompt, steer, and follow-up could queue directly through sendUserMessage while a previously accepted default selection still owned the causal fence. Apply the same abortable selection fence at the shared external SDK ingress, check same-session ownership before prompt waits, and give only private scheduled continuations an explicit fence/reentry capability so #4496 ownerless continuations remain deadlock-free without admitting later external successors. Lore-id: issue-4519-sdk-ingress-fence Constraint: only internal scheduled continuations bypass selection fences Constraint: waitForIdle remains outside selection admission Confidence: high Scope-risk: medium Reversibility: clean Tested: attribution 12/12, default selection 45/45, auto-compaction 23 pass + 1 skip, SDK host wiring 104/104, coding-agent check/types --- .../src/extensibility/extensions/loader.ts | 3 + .../src/extensibility/extensions/runner.ts | 2 +- .../src/extensibility/extensions/types.ts | 7 +- .../src/sdk/host/session-runtime.test.ts | 4 + .../src/sdk/host/session-runtime.ts | 8 +- .../coding-agent/src/session/agent-session.ts | 236 +++++++++-- ...t-session-auto-compaction-continue.test.ts | 81 ++++ ...ion-before-agent-start-attribution.test.ts | 390 ++++++++++++++++++ .../test/agent-session-manual-retry.test.ts | 44 +- 9 files changed, 741 insertions(+), 34 deletions(-) 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..f38d35516f 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,10 @@ export class AgentSession { #sessionAdmissionClosing = false; #sessionAdmissionClosed = false; #sessionAdmissionContext = new AsyncLocalStorage(); + #selectionFenceGenerationContext = new AsyncLocalStorage(); #selectionFenceTail: Promise = Promise.resolve(); + #pendingSelectionFences = 0; + #selectionFenceGeneration = 0; #defaultModelSelectionMutationRevision = 0; #thinkingLevelMutationRevision = 0; #thinkingVisibilityMutationRevision = 0; @@ -2505,6 +2509,7 @@ export class AgentSession { #terminalLineageSecret = crypto.randomUUID(); #promptGeneration = 0; #promptPreflightAbortController = new AbortController(); + #promptPreflightCancellationGeneration = 0; #providerSessionState = new Map(); #temporaryProviderSessionScopes: TemporaryProviderSessionScopeRecord[] = []; @@ -2570,6 +2575,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 +2658,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 +2671,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 +2709,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 +2766,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 +2865,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); @@ -5469,8 +5508,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,7 +5553,7 @@ 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) @@ -5543,10 +5588,45 @@ 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; + void precedingSelectionFence.then(() => + this.#scheduleAgentContinue({ + ...options, + generation: deferredPromptGeneration, + selectionFenceGeneration, + deferredPredecessorAgentEnd, + }), + ); + 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 +5646,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 +5799,10 @@ export class AgentSession { }, scheduledSignal, continuationAdmission, + { + bypassSelectionFenceGeneration: selectionFenceGeneration, + allowPromptContinuationReentry: true, + }, ); } catch (error) { if (scheduledSignal.aborted) skip("aborted_signal"); @@ -5734,6 +5817,7 @@ export class AgentSession { }, resourceRunId: options?.resourceRunId, leaseTask: leaseSettlement.promise, + selectionFenceGeneration, }, ); }; @@ -5815,8 +5899,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 +5940,31 @@ 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; + void precedingSelectionFence.then(() => + this.#scheduleAutoContinuePrompt( + generation, + requireUnfinishedWork, + resourceRunId, + selectionFenceGeneration, + predecessorAgentEnd, + ), + ); + return; + } + const predecessorAgentEndHold = deferredPredecessorAgentEnd + ? this.#restoreAndReserveDeferredAgentEndForContinuation(deferredPredecessorAgentEnd) + : this.#reserveDeferredAgentEndForContinuation(); void this.#schedulePostPromptTask( async signal => { try { @@ -5886,6 +6000,7 @@ export class AgentSession { }, signal, continuationAdmission, + { bypassSelectionFenceGeneration: selectionFenceGeneration }, ); } catch (error) { if (signal.aborted) { @@ -5901,6 +6016,7 @@ export class AgentSession { delayMs: 0, generation: scheduledGeneration, resourceRunId, + selectionFenceGeneration, onSkip: () => { this.#logCompactionContinuationSkipped("auto_continue_prompt", "aborted_signal"); this.#releaseDeferredAgentEndContinuation(predecessorAgentEndHold); @@ -9443,6 +9559,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 +9615,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 +11148,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 +11159,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,28 +11188,62 @@ export class AgentSession { if (images.length === 0) images = undefined; } - if (options?.deliverAs === "followUp") { - if (options.onPreflightAcceptCommit) await options.onPreflightAcceptCommit(); + const admissionSignal = options?.preflightSignal + ? AbortSignal.any([this.#promptPreflightAbortController.signal, options.preflightSignal]) + : this.#promptPreflightAbortController.signal; + const preflightCancellationGeneration = this.#promptPreflightCancellationGeneration; + if (this.#pendingSelectionFences > 0) { + await awaitPromptInvocationPreflight(this.#selectionFenceTail, admissionSignal); + } + const assertPreflightStillOpen = () => { + this.#assertSessionAdmissionOpen(); + if ( + options?.preflightSignal?.aborted || + this.#promptPreflightCancellationGeneration !== preflightCancellationGeneration + ) { + throw promptPreflightCancelledError(); + } + }; + assertPreflightStillOpen(); + const queuedPlainPrompt = options?.queuedAtDispatch === true && options.deliverAs === undefined; + const queuedFollowUpAhead = + queuedPlainPrompt && (this.agent.snapshotFollowUp().length > 0 || this.#deferredSdkFollowUps.length > 0); + const queuedAtDispatchStartsFresh = + queuedPlainPrompt && !queuedFollowUpAhead && !this.agent.state.isStreaming && !this.#canAutoContinueForSteer(); + const deliverAs = + options?.deliverAs ?? + (queuedPlainPrompt + ? queuedFollowUpAhead + ? "followUp" + : queuedAtDispatchStartsFresh + ? undefined + : "steer" + : undefined); + + if (deliverAs === "followUp") { + if (options?.onPreflightAcceptCommit) await options.onPreflightAcceptCommit(); + assertPreflightStillOpen(); const queuedFollowUp = await this.#queueFollowUp(text, images, { claimsGenuineUserIntent: true, - forceOneAtATime: Boolean(options.preflightSignal), + forceOneAtATime: Boolean(options?.preflightSignal || options?.queuedAtDispatch), onPromoted: options?.onQueuedPromoted, - sdkRunToken: options.sdkRunToken, + sdkRunToken: options?.sdkRunToken, }); const cancelQueuedFollowUp = () => queuedFollowUp.cancel(); - options.preflightSignal?.addEventListener("abort", cancelQueuedFollowUp, { once: true }); - if (options.preflightSignal?.aborted) cancelQueuedFollowUp(); - options.onPreflightAccepted?.(); + 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(); + if (deliverAs === "steer") { + if (options?.onPreflightAcceptCommit) await options.onPreflightAcceptCommit(); + assertPreflightStillOpen(); await this.#queueSteer(text, images, { claimsGenuineUserIntent: true, onPromoted: options?.onQueuedPromoted, external: true, }); - options.onPreflightAccepted?.(); + options?.onPreflightAccepted?.(); return; } @@ -11097,6 +11254,7 @@ export class AgentSession { // 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, @@ -11107,11 +11265,28 @@ export class AgentSession { } // Use prompt() with expandPromptTemplates: false to skip command handling and template expansion + let queuedPromotionFired = false; + const fireQueuedPromotion = () => { + if (!queuedAtDispatchStartsFresh || queuedPromotionFired) return; + queuedPromotionFired = true; + options?.onQueuedPromoted?.(); + }; await this.prompt(text, { expandPromptTemplates: false, images, - onPreflightAccepted: options?.onPreflightAccepted, - onPreflightAcceptCommit: options?.onPreflightAcceptCommit, + onPreflightAccepted: () => { + options?.onPreflightAccepted?.(); + fireQueuedPromotion(); + }, + onPreflightAcceptCommit: + options?.onPreflightAcceptCommit || queuedAtDispatchStartsFresh + ? async () => { + if (options?.onPreflightAcceptCommit) await options.onPreflightAcceptCommit(); + else options?.onPreflightAccepted?.(); + assertPreflightStillOpen(); + fireQueuedPromotion(); + } + : undefined, preflightSignal: options?.preflightSignal, }); } @@ -11492,6 +11667,7 @@ export class AgentSession { this.#markRetryReplayUnsafe(); this.abortRetry(); this.#promptGeneration++; + this.#promptPreflightCancellationGeneration++; this.#promptPreflightAbortController.abort(); this.#promptPreflightAbortController = new AbortController(); this.#scheduledHiddenNextTurnGeneration = undefined; @@ -11609,6 +11785,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 +12024,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 +13471,8 @@ export class AgentSession { const expectedSessionId = this.sessionId; const priorSelectionFence = this.#selectionFenceTail; const selectionFence = Promise.withResolvers(); + this.#selectionFenceGeneration += 1; + this.#pendingSelectionFences += 1; this.#selectionFenceTail = priorSelectionFence.then(() => selectionFence.promise); void this.#selectionFenceTail.catch(() => {}); try { @@ -13408,6 +13588,7 @@ export class AgentSession { ); } finally { selectionFence.resolve(); + this.#pendingSelectionFences -= 1; } } /** @@ -18168,6 +18349,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..3eec27e227 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,87 @@ 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("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..bc3832f6b1 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,396 @@ 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("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({ From 162d2c80f8d5992c1a49a743a5448010831bdf03 Mon Sep 17 00:00:00 2001 From: Yeachan Heo Date: Fri, 14 Aug 2026 15:06:00 +0000 Subject: [PATCH 2/5] fix(session): reserve follow-up order and track fence-deferred continuations Codex review of exact head 6907359f495c flagged two ordering gaps plus the missing changelog entry. A later plain prompt could overtake an earlier follow-up dispatch that was still awaiting its durable onPreflightAcceptCommit, because classification only observed already enqueued messages; and a continuation parked behind a pending selection fence cleared the settlement markers without registering anything waitForIdle could observe, so an external idle wait could report the session settled while the continuation was still waiting on the fence. Follow-up ingress now classifies and reserves order synchronously before any await (selection fence or durable acceptance) using counted epoch reservations, and a later dispatch drains earlier reservations before its own durable enqueue so enqueue order can never invert. Fence-deferred continuations are tracked as settlement work keyed by fence generation; the tracking window ends at the deferred invocation's synchronous re-entry, and the selection owning the fence never waits on its own parked continuations, preserving the #4519 no-self-deadlock guarantee. Lore-id: issue-4519-selection-order-review Constraint: selection must never wait on continuations parked behind its own fence Constraint: reservations must not survive a rejected or cancelled acceptance Tested: attribution 24/24 incl. new reservation ordering regression, auto-compaction 26 pass + 1 skip incl. new waitForIdle fence regression, manual-retry 5/5, SDK host runtime 48/48, coding-agent check/types, biome, verify-gjc-state-writers fast gate Not-tested: live multi-connection SDK broker soak beyond the wired host runtime suite Confidence: high Scope-risk: medium Reversibility: clean --- packages/coding-agent/CHANGELOG.md | 1 + .../coding-agent/src/session/agent-session.ts | 228 ++++++++++++++---- ...t-session-auto-compaction-continue.test.ts | 45 ++++ ...ion-before-agent-start-attribution.test.ts | 66 +++++ 4 files changed, 294 insertions(+), 46 deletions(-) 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/session/agent-session.ts b/packages/coding-agent/src/session/agent-session.ts index f38d35516f..b523773a4e 100644 --- a/packages/coding-agent/src/session/agent-session.ts +++ b/packages/coding-agent/src/session/agent-session.ts @@ -1991,6 +1991,12 @@ export class AgentSession { #selectionFenceGenerationContext = new AsyncLocalStorage(); #selectionFenceTail: Promise = Promise.resolve(); #pendingSelectionFences = 0; + #selectionFenceDeferredContinuations = new Map(); + #followUpReservationEpoch = 0; + /** Epochs of follow-up reservations still between reservation and durable enqueue. */ + #activeFollowUpReservationEpochs = new Set(); + #followUpReservationDrainWaiters = new Set<() => void>(); + #activeSelectionFenceGeneration: number | undefined; #selectionFenceGeneration = 0; #defaultModelSelectionMutationRevision = 0; #thinkingLevelMutationRevision = 0; @@ -2933,6 +2939,79 @@ 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 this waiter must observe. */ + #pendingSelectionFenceDeferredContinuations(): number { + const ownerGeneration = this.#activeSelectionFenceGeneration; + let total = 0; + for (const [generation, count] of this.#selectionFenceDeferredContinuations) { + // An in-flight selection never waits on continuations parked behind its + // own fence (or a later one it will itself sequence behind): that is + // the self-deadlock this fence exists to avoid. + if (ownerGeneration !== undefined && generation >= ownerGeneration) continue; + total += count; + } + return total; + } + + /** + * 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(): boolean { return ( @@ -2940,6 +3019,7 @@ export class AgentSession { this.#agentEventHandlersInFlight > 0 || this.#agentEndPublicationInFlight > 0 || this.#pendingAgentEndContinuationHolds.size > 0 || + this.#pendingSelectionFenceDeferredContinuations() > 0 || this.#pendingAgentEndEmit !== undefined ); } @@ -5610,14 +5690,22 @@ export class AgentSession { : undefined); const precedingSelectionFence = this.#selectionFenceTail; const deferredPromptGeneration = options?.generation ?? this.#promptGeneration; - void precedingSelectionFence.then(() => - this.#scheduleAgentContinue({ - ...options, - generation: deferredPromptGeneration, - selectionFenceGeneration, - deferredPredecessorAgentEnd, - }), - ); + 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 = @@ -5951,15 +6039,20 @@ export class AgentSession { deferredPredecessorAgentEnd ?? this.#claimDeferredAgentEndForContinuation(this.#reserveDeferredAgentEndForContinuation()); const precedingSelectionFence = this.#selectionFenceTail; - void precedingSelectionFence.then(() => - this.#scheduleAutoContinuePrompt( - generation, - requireUnfinishedWork, - resourceRunId, - selectionFenceGeneration, - predecessorAgentEnd, - ), - ); + const deferredScheduling = precedingSelectionFence.then(() => { + try { + this.#scheduleAutoContinuePrompt( + generation, + requireUnfinishedWork, + resourceRunId, + selectionFenceGeneration, + predecessorAgentEnd, + ); + } finally { + this.#endSelectionFenceDeferralTracking(selectionFenceGeneration); + } + }); + this.#trackSelectionFenceDeferredContinuation(selectionFenceGeneration, deferredScheduling); return; } const predecessorAgentEndHold = deferredPredecessorAgentEnd @@ -11192,6 +11285,44 @@ export class AgentSession { ? 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(); + }; if (this.#pendingSelectionFences > 0) { await awaitPromptInvocationPreflight(this.#selectionFenceTail, admissionSignal); } @@ -11205,34 +11336,34 @@ export class AgentSession { } }; assertPreflightStillOpen(); - const queuedPlainPrompt = options?.queuedAtDispatch === true && options.deliverAs === undefined; - const queuedFollowUpAhead = - queuedPlainPrompt && (this.agent.snapshotFollowUp().length > 0 || this.#deferredSdkFollowUps.length > 0); - const queuedAtDispatchStartsFresh = - queuedPlainPrompt && !queuedFollowUpAhead && !this.agent.state.isStreaming && !this.#canAutoContinueForSteer(); - const deliverAs = - options?.deliverAs ?? - (queuedPlainPrompt - ? queuedFollowUpAhead - ? "followUp" - : queuedAtDispatchStartsFresh - ? undefined - : "steer" - : undefined); - if (deliverAs === "followUp") { - 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?.(); + try { + // 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?.(); + } finally { + releaseFollowUpReservation(); + } return; } if (deliverAs === "steer") { @@ -11267,7 +11398,7 @@ export class AgentSession { // Use prompt() with expandPromptTemplates: false to skip command handling and template expansion let queuedPromotionFired = false; const fireQueuedPromotion = () => { - if (!queuedAtDispatchStartsFresh || queuedPromotionFired) return; + if (!freshAtReservation || queuedPromotionFired) return; queuedPromotionFired = true; options?.onQueuedPromoted?.(); }; @@ -11279,7 +11410,7 @@ export class AgentSession { fireQueuedPromotion(); }, onPreflightAcceptCommit: - options?.onPreflightAcceptCommit || queuedAtDispatchStartsFresh + options?.onPreflightAcceptCommit || freshAtReservation ? async () => { if (options?.onPreflightAcceptCommit) await options.onPreflightAcceptCommit(); else options?.onPreflightAccepted?.(); @@ -13473,6 +13604,8 @@ export class AgentSession { const selectionFence = Promise.withResolvers(); this.#selectionFenceGeneration += 1; this.#pendingSelectionFences += 1; + const selectionFenceGeneration = this.#selectionFenceGeneration; + this.#activeSelectionFenceGeneration = selectionFenceGeneration; this.#selectionFenceTail = priorSelectionFence.then(() => selectionFence.promise); void this.#selectionFenceTail.catch(() => {}); try { @@ -13589,6 +13722,9 @@ export class AgentSession { } finally { selectionFence.resolve(); this.#pendingSelectionFences -= 1; + if (this.#activeSelectionFenceGeneration === selectionFenceGeneration) { + this.#activeSelectionFenceGeneration = undefined; + } } } /** 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 3eec27e227..2a7b5a580a 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 @@ -810,6 +810,51 @@ describe("AgentSession auto-compaction continuation", () => { 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 with + // no other in-flight work: waitForIdle must not report the session idle + // while that continuation is still waiting on the fence. + 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("reschedules an AgentBusyError racing the queued-followup continue until delivery", async () => { session.agent.followUp({ 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 bc3832f6b1..f713558d19 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 @@ -511,6 +511,72 @@ describe("AgentSession before_agent_start attribution fallback", () => { 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("rejects fresh queued SDK promotion when disposal starts during durable acceptance", async () => { createSession(); const providerPrompt = vi.spyOn(session.agent, "prompt"); From bd98e24687e888bd00307c11d7c44de9774c7b29 Mon Sep 17 00:00:00 2001 From: Yeachan Heo Date: Fri, 14 Aug 2026 16:05:20 +0000 Subject: [PATCH 3/5] fix(session): settle-fence deferral scoping and reservation leak MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Codex review of head 69b98ef30a flagged two P1 gaps in the fix-forward. The idle-observable deferral counter keyed off a single active-generation field, so when selection B reserved generation 2 while a continuation was still deferred behind selection A's generation-1 fence, A observed generation 2 as its own and could wait on the generation-1 continuation that was itself waiting on A — deadlock. And a follow-up dispatch that reserved an epoch and then had its selection-fence wait rejected by terminal cancellation exited before the cleanup scope, leaking the epoch so later prompts misclassified as having a follow-up ahead and later follow-ups waited forever. Deferral observability is now scoped structurally instead of by caller identity: only deferrals whose fence has already settled count toward pending settlement, because their deferred tails re-enter within microtask time. A deferral behind a still-pending fence is invisible by construction — the fence owner's completion may causally depend on the settlement a waiter would block on — so counting it could always reintroduce the #4519 self-deadlock regardless of which selection observes it. The selection finally records the settled generation and resolves settlement waiters. The follow-up reservation now covers the fence wait and post-wait assertion inside its try/finally, so any exit path releases the epoch and wakes drain waiters. Lore-id: issue-4519-selection-order-review-2 Constraint: no caller-identity-based fence exclusion; observability is settled-fence-scoped only Constraint: every reservation exit path must release the epoch Tested: attribution 25/25 incl. new reservation-leak regression, auto-compaction 27 pass + 1 skip incl. new overlapping-selections no-deadlock regression, manual-retry 5/5, SDK host runtime 48/48, coding-agent check/types, fast gate 0 violations Confidence: high Scope-risk: medium Reversibility: clean --- .../coding-agent/src/session/agent-session.ts | 167 +++++++++--------- ...t-session-auto-compaction-continue.test.ts | 46 +++++ ...ion-before-agent-start-attribution.test.ts | 50 ++++++ 3 files changed, 184 insertions(+), 79 deletions(-) diff --git a/packages/coding-agent/src/session/agent-session.ts b/packages/coding-agent/src/session/agent-session.ts index b523773a4e..6425217738 100644 --- a/packages/coding-agent/src/session/agent-session.ts +++ b/packages/coding-agent/src/session/agent-session.ts @@ -1991,12 +1991,12 @@ export class AgentSession { #selectionFenceGenerationContext = new AsyncLocalStorage(); #selectionFenceTail: Promise = Promise.resolve(); #pendingSelectionFences = 0; + #settledSelectionFenceGeneration = 0; #selectionFenceDeferredContinuations = new Map(); #followUpReservationEpoch = 0; /** Epochs of follow-up reservations still between reservation and durable enqueue. */ #activeFollowUpReservationEpochs = new Set(); #followUpReservationDrainWaiters = new Set<() => void>(); - #activeSelectionFenceGeneration: number | undefined; #selectionFenceGeneration = 0; #defaultModelSelectionMutationRevision = 0; #thinkingLevelMutationRevision = 0; @@ -2980,15 +2980,24 @@ export class AgentSession { this.#resolveSessionSettlement(); } - /** Continuations parked behind selection fences this waiter must observe. */ + /** + * 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(): number { - const ownerGeneration = this.#activeSelectionFenceGeneration; + const settledGeneration = this.#settledSelectionFenceGeneration; let total = 0; for (const [generation, count] of this.#selectionFenceDeferredContinuations) { - // An in-flight selection never waits on continuations parked behind its - // own fence (or a later one it will itself sequence behind): that is - // the self-deadlock this fence exists to avoid. - if (ownerGeneration !== undefined && generation >= ownerGeneration) continue; + if (generation > settledGeneration) continue; total += count; } return total; @@ -11323,21 +11332,21 @@ export class AgentSession { this.#followUpReservationDrainWaiters.clear(); for (const waiter of waiters) waiter(); }; - if (this.#pendingSelectionFences > 0) { - await awaitPromptInvocationPreflight(this.#selectionFenceTail, admissionSignal); - } - const assertPreflightStillOpen = () => { - this.#assertSessionAdmissionOpen(); - if ( - options?.preflightSignal?.aborted || - this.#promptPreflightCancellationGeneration !== preflightCancellationGeneration - ) { - throw promptPreflightCancelledError(); + try { + if (this.#pendingSelectionFences > 0) { + await awaitPromptInvocationPreflight(this.#selectionFenceTail, admissionSignal); } - }; - assertPreflightStillOpen(); - if (deliverAs === "followUp") { - try { + 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 @@ -11361,65 +11370,65 @@ export class AgentSession { options?.preflightSignal?.addEventListener("abort", cancelQueuedFollowUp, { once: true }); if (options?.preflightSignal?.aborted) cancelQueuedFollowUp(); options?.onPreflightAccepted?.(); - } finally { - releaseFollowUpReservation(); + 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; } - 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(); - assertPreflightStillOpen(); - 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 - 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, - }); } /** @@ -13605,7 +13614,6 @@ export class AgentSession { this.#selectionFenceGeneration += 1; this.#pendingSelectionFences += 1; const selectionFenceGeneration = this.#selectionFenceGeneration; - this.#activeSelectionFenceGeneration = selectionFenceGeneration; this.#selectionFenceTail = priorSelectionFence.then(() => selectionFence.promise); void this.#selectionFenceTail.catch(() => {}); try { @@ -13722,9 +13730,10 @@ export class AgentSession { } finally { selectionFence.resolve(); this.#pendingSelectionFences -= 1; - if (this.#activeSelectionFenceGeneration === selectionFenceGeneration) { - this.#activeSelectionFenceGeneration = undefined; + if (selectionFenceGeneration > this.#settledSelectionFenceGeneration) { + this.#settledSelectionFenceGeneration = selectionFenceGeneration; } + this.#resolveSessionSettlement(); } } /** 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 2a7b5a580a..02f64b222b 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 @@ -855,6 +855,52 @@ describe("AgentSession auto-compaction continuation", () => { 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({ 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 f713558d19..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 @@ -577,6 +577,56 @@ describe("AgentSession before_agent_start attribution fallback", () => { 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"); From cec2d55c564565e451d41880682d53a2c19f376f Mon Sep 17 00:00:00 2001 From: Yeachan Heo Date: Fri, 14 Aug 2026 19:52:15 +0000 Subject: [PATCH 4/5] fix(session): keep fence-deferred continuations visible to external idle waits MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Codex review of head 88deb97c72 flagged that the settled-generation filter hid unresolved fence-deferred continuations from every idle wait, not just the selection's own: once the predecessor event handler settled, an external waitForIdle could report the session idle while the parked continuation was still waiting on its fence, letting SDK completion reporting race the continuation. External idle waits now count every unresolved deferral. Only two callers narrow their view, both by explicit internal-only parameter rather than caller inference: setDefaultModelSelection's mid-selection drain passes its own fence generation (work parked behind its own fence is waiting on the selection itself), and the causally-upstream turn-settle path excludes deferrals behind still-pending fences via the oldest pending fence generation. Scoped waiters no longer share the strict settlement promise — they register a wake callback re-evaluated on settlement transitions — because that promise may correctly stay pending on a fence-deferred continuation the fence owner must not block behind. Lore-id: issue-4519-selection-order-review-3 Constraint: external waitForIdle observes all unresolved fence-deferred continuations Constraint: only explicit internal parameters narrow deferral visibility; no caller inference Constraint: scoped settlement waits never block behind the strict settlement promise Tested: attribution 25/25, auto-compaction 27 pass + 1 skip incl. waitForIdle-pending regression, manual-retry 5/5, SDK host 48/48, sdk-session-router-authority 40/40, coding-agent check/types, fast gate 0 Confidence: high Scope-risk: medium Reversibility: clean --- .../coding-agent/src/session/agent-session.ts | 109 ++++++++++++++---- ...t-session-auto-compaction-continue.test.ts | 7 +- 2 files changed, 90 insertions(+), 26 deletions(-) diff --git a/packages/coding-agent/src/session/agent-session.ts b/packages/coding-agent/src/session/agent-session.ts index 6425217738..856f392af7 100644 --- a/packages/coding-agent/src/session/agent-session.ts +++ b/packages/coding-agent/src/session/agent-session.ts @@ -1991,8 +1991,9 @@ export class AgentSession { #selectionFenceGenerationContext = new AsyncLocalStorage(); #selectionFenceTail: Promise = Promise.resolve(); #pendingSelectionFences = 0; - #settledSelectionFenceGeneration = 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(); @@ -2993,16 +2994,26 @@ export class AgentSession { * synchronous re-entry (`#endSelectionFenceDeferralTracking`) or the * promise-settled safety net clears it. */ - #pendingSelectionFenceDeferredContinuations(): number { - const settledGeneration = this.#settledSelectionFenceGeneration; + #pendingSelectionFenceDeferredContinuations(ignoreGenerationFrom?: number): number { let total = 0; for (const [generation, count] of this.#selectionFenceDeferredContinuations) { - if (generation > settledGeneration) continue; + if (ignoreGenerationFrom !== undefined && generation >= ignoreGenerationFrom) continue; total += count; } return total; } + /** + * 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 @@ -3022,33 +3033,66 @@ export class AgentSession { } } - #isSessionSettlementPending(): boolean { + #isSessionSettlementPending(ignoreSelectionFenceGeneration?: number): boolean { return ( this.#promptInFlightCount > 0 || this.#agentEventHandlersInFlight > 0 || this.#agentEndPublicationInFlight > 0 || this.#pendingAgentEndContinuationHolds.size > 0 || - this.#pendingSelectionFenceDeferredContinuations() > 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; } } @@ -3085,7 +3129,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; } @@ -7490,17 +7539,28 @@ 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(); + } + await this.#waitForSessionSettlement(ignoreSelectionFenceGeneration); if ( !this.agent.state.isStreaming && !this.#retryPromise && !this.#ttsrResumePromise && !this.#postPromptTasksPromise && - !this.#isSessionSettlementPending() + !this.#isSessionSettlementPending(ignoreSelectionFenceGeneration) ) return; } @@ -13614,6 +13674,9 @@ export class AgentSession { 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 { @@ -13643,7 +13706,7 @@ export class AgentSession { undefined, { allowDuringClosing: true }, ); - await this.waitForIdle(); + await this.waitForIdle(selectionFenceGeneration); return await this.#withSessionAdmission( "selection", async () => { @@ -13730,8 +13793,8 @@ export class AgentSession { } finally { selectionFence.resolve(); this.#pendingSelectionFences -= 1; - if (selectionFenceGeneration > this.#settledSelectionFenceGeneration) { - this.#settledSelectionFenceGeneration = selectionFenceGeneration; + if (this.#oldestPendingSelectionFenceGeneration === selectionFenceGeneration) { + this.#oldestPendingSelectionFenceGeneration = 0; } this.#resolveSessionSettlement(); } 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 02f64b222b..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 @@ -836,9 +836,10 @@ describe("AgentSession auto-compaction continuation", () => { 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 with - // no other in-flight work: waitForIdle must not report the session idle - // while that continuation is still waiting on the fence. + // 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; From 33dc4d61e8b8dd1c578900b04fa1e5c544c70271 Mon Sep 17 00:00:00 2001 From: Yeachan Heo Date: Fri, 14 Aug 2026 22:45:38 +0000 Subject: [PATCH 5/5] fix(session): await pre-fence recovery in scoped idle drains Selection drains skipped all post-prompt recovery, which could starve a timer-backed continuation that predates the selection fence. Track each task fence generation so scoped drains await only causally earlier work. Lore-id: 4540-scoped-idle-recovery Constraint: selection must not await continuations parked behind its own fence Confidence: high Scope-risk: narrow Reversibility: simple Tested: bun --cwd=packages/coding-agent run check; bun test packages/coding-agent/test/agent-session-before-agent-start-attribution.test.ts packages/coding-agent/test/agent-session-auto-compaction-continue.test.ts packages/coding-agent/test/agent-session-manual-retry.test.ts --- .../coding-agent/src/session/agent-session.ts | 33 +++++++++++++++++-- 1 file changed, 31 insertions(+), 2 deletions(-) diff --git a/packages/coding-agent/src/session/agent-session.ts b/packages/coding-agent/src/session/agent-session.ts index 856f392af7..14588f1fad 100644 --- a/packages/coding-agent/src/session/agent-session.ts +++ b/packages/coding-agent/src/session/agent-session.ts @@ -2442,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(); @@ -5623,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); @@ -5634,6 +5641,7 @@ export class AgentSession { .catch(() => {}) .finally(() => { this.#postPromptTasks.delete(task); + this.#postPromptTaskSelectionFenceGenerations.delete(task); if (this.#postPromptTasks.size === 0) this.#resolvePostPromptTasks(); }); } @@ -5696,7 +5704,12 @@ export class AgentSession { 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; } @@ -6199,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(); @@ -6235,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; @@ -7553,6 +7580,8 @@ export class AgentSession { if (ignoreSelectionFenceGeneration === undefined) { await this.agent.waitForIdle(); await this.#waitForPostPromptRecovery(); + } else { + await this.#waitForPostPromptTasksBeforeSelectionFence(ignoreSelectionFenceGeneration); } await this.#waitForSessionSettlement(ignoreSelectionFenceGeneration); if (