diff --git a/src/agent/providers/anthropic-direct/query.ts b/src/agent/providers/anthropic-direct/query.ts index 586664b3..d55e683d 100644 --- a/src/agent/providers/anthropic-direct/query.ts +++ b/src/agent/providers/anthropic-direct/query.ts @@ -69,7 +69,7 @@ import type { } from './types.js'; import { repairOrphanToolUses } from './query/repair-orphan-tool-uses.js'; import { type SessionState, createSessionState } from './query/session-state.js'; -import { AbortCoordinator } from './query/abort-coordinator.js'; +import { AbortCoordinator } from '../shared/abort-coordinator.js'; import { RetryLayer } from './query/retry-layer.js'; import { compactHistory } from './query/compact-handler.js'; import { listUserTurns, rewindConversationHistory } from './query/rewind-conversation.js'; diff --git a/src/agent/providers/anthropic-direct/query/compact-handler.microcompact.test.ts b/src/agent/providers/anthropic-direct/query/compact-handler.microcompact.test.ts index 1c2f77c7..99245f52 100644 --- a/src/agent/providers/anthropic-direct/query/compact-handler.microcompact.test.ts +++ b/src/agent/providers/anthropic-direct/query/compact-handler.microcompact.test.ts @@ -16,7 +16,7 @@ import { describe, it, expect, beforeEach, afterEach } from 'vitest'; import type { MessageParam } from '@anthropic-ai/sdk/resources'; import { compactHistory } from './compact-handler.js'; import { createSessionState, type SessionState } from './session-state.js'; -import { AbortCoordinator } from './abort-coordinator.js'; +import { AbortCoordinator } from '../../shared/abort-coordinator.js'; import type { RetryLayer } from './retry-layer.js'; import type { ToolDispatcher } from '../tool-dispatcher.js'; import { findCompactionBoundary } from '../compact.js'; diff --git a/src/agent/providers/anthropic-direct/query/compact-handler.ts b/src/agent/providers/anthropic-direct/query/compact-handler.ts index 50031905..483cce5d 100644 --- a/src/agent/providers/anthropic-direct/query/compact-handler.ts +++ b/src/agent/providers/anthropic-direct/query/compact-handler.ts @@ -56,7 +56,7 @@ import { emitCompaction } from '../../../trace/emit.js'; import { resolveModelId } from '../../../session/model-resolution.js'; import type { AnthropicClientLike } from '../types.js'; import type { SessionState } from './session-state.js'; -import type { AbortCoordinator } from './abort-coordinator.js'; +import type { AbortCoordinator } from '../../shared/abort-coordinator.js'; import type { RetryLayer } from './retry-layer.js'; import { env } from '../../../../config/env.js'; diff --git a/src/agent/providers/anthropic-direct/query/rewind-conversation.test.ts b/src/agent/providers/anthropic-direct/query/rewind-conversation.test.ts index 8c021e22..c640c8cd 100644 --- a/src/agent/providers/anthropic-direct/query/rewind-conversation.test.ts +++ b/src/agent/providers/anthropic-direct/query/rewind-conversation.test.ts @@ -12,7 +12,7 @@ import { describe, it, expect } from 'vitest'; import type { ContentBlockParam, MessageParam } from '@anthropic-ai/sdk/resources'; import { listUserTurns, rewindConversationHistory } from './rewind-conversation.js'; import type { SessionState } from './session-state.js'; -import type { AbortCoordinator } from './abort-coordinator.js'; +import type { AbortCoordinator } from '../../shared/abort-coordinator.js'; function makeState(messages: MessageParam[], closed = false): SessionState { // The handler only reads `messages` + `closed`; the rest of SessionState is diff --git a/src/agent/providers/anthropic-direct/query/rewind-conversation.ts b/src/agent/providers/anthropic-direct/query/rewind-conversation.ts index 58b2e601..72a1b726 100644 --- a/src/agent/providers/anthropic-direct/query/rewind-conversation.ts +++ b/src/agent/providers/anthropic-direct/query/rewind-conversation.ts @@ -35,7 +35,7 @@ import type { } from '../../../provider.js'; import { repairOrphanToolUses } from './repair-orphan-tool-uses.js'; import type { SessionState } from './session-state.js'; -import type { AbortCoordinator } from './abort-coordinator.js'; +import type { AbortCoordinator } from '../../shared/abort-coordinator.js'; const PREVIEW_MAX_CHARS = 72; diff --git a/src/agent/providers/openai-compatible/query.ts b/src/agent/providers/openai-compatible/query.ts index c37ef45f..09631405 100644 --- a/src/agent/providers/openai-compatible/query.ts +++ b/src/agent/providers/openai-compatible/query.ts @@ -11,8 +11,10 @@ * Lifecycle: * - constructor is synchronous; emits `session.init` on first iterator pull * - main loop awaits user turns from `promptStream`, races against - * `closedPromise` so `close()` unblocks a "waiting for next turn" state - * - each user turn opens a new AbortController; `interrupt()` aborts it + * `abort.closedPromise` so `close()` unblocks a "waiting for next turn" + * state + * - each user turn opens a new abort scope via `abort.begin()`; + * `interrupt()` aborts it — see shared/abort-coordinator.ts * - `runTurn` iterates model→tools→model until the model stops calling tools * (or the tool-round cap fires — see shared/tool-loop-cap.ts — after which * one tools-stripped wind-down round runs, matching anthropic-direct/loop.ts) @@ -88,6 +90,7 @@ import { shouldAutoCompact, resolveAutoCompactThreshold, } from '../shared/auto-compact.js'; +import { AbortCoordinator, CLOSED_SENTINEL } from '../shared/abort-coordinator.js'; import { HookBlockedError, DenialCircuitBreakerError } from '../../../utils/errors.js'; import { COMPACT_SYSTEM_PROMPT, wrapTranscriptForSummary } from '../shared/compaction.js'; import { compactOpenAIHistory, readShrinkFraction } from './compact.js'; @@ -193,11 +196,23 @@ export class OpenAICompatibleQuery implements ProviderQuery { private currentModel: string; private currentPermissionMode: string; - private abortController: AbortController | null = null; - private pendingAbortReason: 'interrupted' | 'closed' | null = null; + /** + * Per-session abort coordination — see {@link AbortCoordinator}. Owns the + * in-flight controller slot, the pending-abort drain (an `interrupt()` / + * `close()` that lands between turns), and the close promise the outer + * loop races against `promptStream.next()`. + * + * Shared with anthropic-direct so the two providers cannot drift apart on + * the compare-and-clear invariant this encapsulates. + */ + private readonly abort = new AbortCoordinator(); + + /** + * Stays here rather than in the coordinator: sub-generators read it on + * every loop iteration, and `close()` updates the flag and the promise + * together. See the coordinator's "what this module does NOT own" note. + */ private closed = false; - private closeResolve: (() => void) | null = null; - private readonly closedPromise: Promise<'__closed__'>; /** * Last completed turn's accumulated usage — drives `getContextUsage()`. @@ -260,10 +275,6 @@ export class OpenAICompatibleQuery implements ProviderQuery { if (wire.headers !== undefined) clientOpts.defaultHeaders = wire.headers; this.client = ctor(clientOpts); } - - this.closedPromise = new Promise<'__closed__'>((resolve) => { - this.closeResolve = () => resolve('__closed__'); - }); } /** @@ -311,17 +322,17 @@ export class OpenAICompatibleQuery implements ProviderQuery { const promptIterator = this.opts.promptStream[Symbol.asyncIterator](); try { while (!this.closed) { - const nextOrClose = await Promise.race([promptIterator.next(), this.closedPromise]); - if (nextOrClose === '__closed__') break; + const nextOrClose = await Promise.race([promptIterator.next(), this.abort.closedPromise]); + if (nextOrClose === CLOSED_SENTINEL) break; const turnResult = nextOrClose as IteratorResult; if (turnResult.done) break; yield* this.runTurn(turnResult.value.content); // Auto-compaction fires at the natural turn boundary — runTurn has - // returned and its `finally` nulled `abortController`, so the handler's - // idle guard passes and compaction never runs mid-tool-call. Mirrors - // anthropic-direct/query.ts. `compactHistory` itself never throws (every + // returned and every exit path called `abort.clear(controller)`, so + // `abort.isIdle()` is true and compaction never runs mid-tool-call. + // Mirrors anthropic-direct/query.ts. `compactHistory` never throws (every // summarize failure is a typed no-op leaving history byte-for-byte // unchanged); only a PreCompact `block` decision throws HookBlockedError, // caught here to skip this turn's compaction without surfacing an error. @@ -373,12 +384,10 @@ export class OpenAICompatibleQuery implements ProviderQuery { * across the tool-call loop into one final event). */ private async *runTurn(content: ProviderUserTurn['content']): AsyncGenerator { - const controller = new AbortController(); - this.abortController = controller; - if (this.pendingAbortReason !== null && !controller.signal.aborted) { - controller.abort(this.pendingAbortReason); - this.pendingAbortReason = null; - } + // begin() mints the controller, installs it as the current slot, and + // drains any abort that arrived between turns onto it — so a pre-aborted + // turn is caught by the guard below before any work starts. + const controller = this.abort.begin(); if (controller.signal.aborted) return; // Wall-clock anchor for the REPL footer's `◦ Xs · $cost · N tok` line. @@ -493,7 +502,7 @@ export class OpenAICompatibleQuery implements ProviderQuery { for (;;) { if (controller.signal.aborted) { - if (this.abortController === controller) this.abortController = null; + this.abort.clear(controller); yield* this.finishTurn(accumulatedUsage, turnStartTime); return; } @@ -507,7 +516,7 @@ export class OpenAICompatibleQuery implements ProviderQuery { // (agent-session.ts:sendMessageStreamInternal) unblocks its turn loop; // on a real stream error the already-yielded `error` event is itself // terminal, so skip turn.completed to avoid double-advancing turn state. - if (this.abortController === controller) this.abortController = null; + this.abort.clear(controller); if (controller.signal.aborted || this.closed) { yield* this.finishTurn(accumulatedUsage, turnStartTime); } @@ -553,7 +562,7 @@ export class OpenAICompatibleQuery implements ProviderQuery { // finishTurn: the error event is itself terminal (mirrors the stream-error // path above at `result === null`). if (denialTrip) { - if (this.abortController === controller) this.abortController = null; + this.abort.clear(controller); yield { type: 'error', error: new DenialCircuitBreakerError(denialTrip.content) }; return; } @@ -603,7 +612,7 @@ export class OpenAICompatibleQuery implements ProviderQuery { } if (controller.signal.aborted) { - if (this.abortController === controller) this.abortController = null; + this.abort.clear(controller); yield* this.finishTurn(accumulatedUsage, turnStartTime); return; } @@ -618,7 +627,7 @@ export class OpenAICompatibleQuery implements ProviderQuery { } } - if (this.abortController === controller) this.abortController = null; + this.abort.clear(controller); if (finalAssistantText.length > 0) { const assistantTurn: OpenAIMessage = { @@ -863,12 +872,9 @@ export class OpenAICompatibleQuery implements ProviderQuery { // ---- ProviderQuery surface ------------------------------------------------ async interrupt(): Promise { - const c = this.abortController; - if (c && !c.signal.aborted) { - c.abort('interrupted'); - return; - } - this.pendingAbortReason = 'interrupted'; + // Aborts the in-flight turn, or parks the reason for the next begin() + // when the interrupt lands between turns. + this.abort.requestAbort('interrupted'); } /** @@ -943,15 +949,17 @@ export class OpenAICompatibleQuery implements ProviderQuery { signal, }), isClosed: this.closed, - isIdle: this.abortController === null, - beginAbort: () => { - const controller = new AbortController(); - this.abortController = controller; - return controller; - }, - clearAbort: (controller) => { - if (this.abortController === controller) this.abortController = null; - }, + isIdle: this.abort.isIdle(), + // Invariant: compaction opens a real abort scope through the same + // coordinator the turn loop uses, so `interrupt()` cancels an + // in-flight summarize. Note `begin()` also DRAINS a reason parked + // between turns — an ESC that lands at the turn boundary now + // pre-aborts the auto-compaction that fires there (and consumes the + // reason) instead of letting it run. This matches + // anthropic-direct/query/compact-handler.ts:148; the previous + // openai-only behaviour installed a controller without draining. + beginAbort: () => this.abort.begin(), + clearAbort: (controller) => this.abort.clear(controller), trigger, traceWriter: this.traceWriter, }); @@ -1076,14 +1084,13 @@ export class OpenAICompatibleQuery implements ProviderQuery { } close(): void { + // Invariant: the `closed` flag and the close promise must be updated + // together — sub-generators poll the flag each loop iteration while the + // outer loop is parked on the promise, so resolving one without the + // other leaves a reader stuck. this.closed = true; - const c = this.abortController; - if (c && !c.signal.aborted) { - c.abort('closed'); - } else { - this.pendingAbortReason = 'closed'; - } - this.closeResolve?.(); + this.abort.requestAbort('closed'); + this.abort.markClosed(); debugLog(`🟢 ${PROVIDER_NAME}: closed`); } } diff --git a/src/agent/providers/shared/abort-coordinator.test.ts b/src/agent/providers/shared/abort-coordinator.test.ts new file mode 100644 index 00000000..622a4eb3 --- /dev/null +++ b/src/agent/providers/shared/abort-coordinator.test.ts @@ -0,0 +1,157 @@ +import { describe, it, expect } from 'vitest'; +import { AbortCoordinator, CLOSED_SENTINEL } from './abort-coordinator.js'; + +describe('shared/abort-coordinator', () => { + describe('begin / clear scope lifecycle', () => { + it('starts idle, is busy inside a scope, and is idle again after clear', () => { + const abort = new AbortCoordinator(); + expect(abort.isIdle()).toBe(true); + + const controller = abort.begin(); + expect(abort.isIdle()).toBe(false); + expect(controller.signal.aborted).toBe(false); + + abort.clear(controller); + expect(abort.isIdle()).toBe(true); + }); + + it('mints a distinct controller per begin()', () => { + const abort = new AbortCoordinator(); + const first = abort.begin(); + const second = abort.begin(); + expect(second).not.toBe(first); + }); + + it('clear() is idempotent for the same controller', () => { + const abort = new AbortCoordinator(); + const controller = abort.begin(); + abort.clear(controller); + abort.clear(controller); + expect(abort.isIdle()).toBe(true); + }); + }); + + // The highest-risk invariant this class exists to centralize: a late clear() + // from an older scope must NOT null a slot a newer begin() already claimed, + // or compact()'s idle guard would misread "turn in flight" as idle and run + // summarization mid-turn. Previously hand-rolled at 5 sites in + // anthropic-direct and 6 in openai-compatible. + describe('compare-and-clear (stale-scope protection)', () => { + it('a stale clear() does not release a newer scope', () => { + const abort = new AbortCoordinator(); + const stale = abort.begin(); + const current = abort.begin(); + + abort.clear(stale); + + expect(abort.isIdle()).toBe(false); + abort.clear(current); + expect(abort.isIdle()).toBe(true); + }); + + it('clearing an entirely foreign controller is a no-op', () => { + const abort = new AbortCoordinator(); + const current = abort.begin(); + abort.clear(new AbortController()); + expect(abort.isIdle()).toBe(false); + abort.clear(current); + expect(abort.isIdle()).toBe(true); + }); + }); + + describe('requestAbort with a scope in flight', () => { + it('aborts the current controller with the given reason', () => { + const abort = new AbortCoordinator(); + const controller = abort.begin(); + + abort.requestAbort('interrupted'); + + expect(controller.signal.aborted).toBe(true); + expect(controller.signal.reason).toBe('interrupted'); + }); + + it('does not overwrite the reason of an already-aborted scope', () => { + const abort = new AbortCoordinator(); + const controller = abort.begin(); + + abort.requestAbort('interrupted'); + abort.requestAbort('closed'); + + expect(controller.signal.reason).toBe('interrupted'); + }); + }); + + describe('pending-abort drain (abort arrives between turns)', () => { + it('parks the reason and fires it on the next begin()', () => { + const abort = new AbortCoordinator(); + // No scope in flight — this is the interrupt()-between-turns path. + abort.requestAbort('interrupted'); + expect(abort.isIdle()).toBe(true); + + const controller = abort.begin(); + expect(controller.signal.aborted).toBe(true); + expect(controller.signal.reason).toBe('interrupted'); + }); + + it('drains the parked reason exactly once', () => { + const abort = new AbortCoordinator(); + abort.requestAbort('closed'); + + const first = abort.begin(); + expect(first.signal.aborted).toBe(true); + abort.clear(first); + + const second = abort.begin(); + expect(second.signal.aborted).toBe(false); + }); + + it('parks a reason when the current scope was already aborted', () => { + const abort = new AbortCoordinator(); + const controller = abort.begin(); + abort.requestAbort('interrupted'); + expect(controller.signal.aborted).toBe(true); + + // Slot still holds an aborted controller; the next request must park + // rather than silently no-op, so the following turn still halts. + abort.requestAbort('closed'); + abort.clear(controller); + + const next = abort.begin(); + expect(next.signal.aborted).toBe(true); + expect(next.signal.reason).toBe('closed'); + }); + }); + + describe('close promise', () => { + it('resolves with the sentinel once markClosed() fires', async () => { + const abort = new AbortCoordinator(); + abort.markClosed(); + await expect(abort.closedPromise).resolves.toBe(CLOSED_SENTINEL); + }); + + it('markClosed() is idempotent', async () => { + const abort = new AbortCoordinator(); + abort.markClosed(); + abort.markClosed(); + await expect(abort.closedPromise).resolves.toBe(CLOSED_SENTINEL); + }); + + it('stays pending until markClosed() is called', async () => { + const abort = new AbortCoordinator(); + const raced = await Promise.race([ + abort.closedPromise, + Promise.resolve('still-open' as const), + ]); + expect(raced).toBe('still-open'); + }); + + it('unblocks a parked race the way the provider main loop uses it', async () => { + const abort = new AbortCoordinator(); + // Mirrors query.ts: race the next user turn against close(). + const neverResolves = new Promise<'turn'>(() => {}); + setTimeout(() => abort.markClosed(), 0); + const winner = await Promise.race([neverResolves, abort.closedPromise]); + expect(winner).toBe(CLOSED_SENTINEL); + }); + }); +}); diff --git a/src/agent/providers/anthropic-direct/query/abort-coordinator.ts b/src/agent/providers/shared/abort-coordinator.ts similarity index 74% rename from src/agent/providers/anthropic-direct/query/abort-coordinator.ts rename to src/agent/providers/shared/abort-coordinator.ts index 15537adc..06e91625 100644 --- a/src/agent/providers/anthropic-direct/query/abort-coordinator.ts +++ b/src/agent/providers/shared/abort-coordinator.ts @@ -1,5 +1,6 @@ /** - * Per-session abort coordination for {@link AnthropicDirectQuery}. + * Per-session abort coordination, shared by every provider's query + * orchestrator (`anthropic-direct/query.ts`, `openai-compatible/query.ts`). * * Owns the single `AbortController | null` slot that represents "is a turn * (or compact) currently in flight?" and the deferred abort reason for the @@ -7,21 +8,28 @@ * to fire yet). Also owns the `closedPromise` that the outer loop races * against `promptStream.next()` so `close()` unblocks a pending pull. * + * Lives in `providers/shared/` for the same reason `tool-loop-cap.ts` does: + * the abort contract is provider-neutral (pure `AbortController` plumbing, + * no SDK types), and keeping one copy is what stops the two providers from + * drifting apart on a safety-critical invariant. + * * # Highest-risk invariant * * **The current-controller slot must be `null` between turns** so that * `compact()`'s `isIdle()` guard does not misclassify a leaked stale - * controller as `turn-in-flight`. Five sites in the old `query.ts` enforced - * this with the compare-and-clear pattern: + * controller as `turn-in-flight`. Before this was extracted, each provider + * hand-rolled the compare-and-clear pattern at every end-of-scope site + * (five in anthropic-direct, six in openai-compatible): * * if (this.abortController === controller) this.abortController = null; * - * After this extraction, **{@link AbortCoordinator.clear} is the only write - * path to null** — every "end of scope" site calls `clear(controller)` with - * the controller it received from {@link begin}. The compare-and-clear - * semantics are preserved: if a parallel scope already replaced the slot, - * we leave that newer controller in place. The canary test for this - * invariant is `concurrent-session-isolation.test.ts`. + * Now **{@link AbortCoordinator.clear} is the only write path to null** — + * every "end of scope" site calls `clear(controller)` with the controller + * it received from {@link begin}. The compare-and-clear semantics are + * preserved: if a parallel scope already replaced the slot, we leave that + * newer controller in place. The canary tests for this invariant are + * `anthropic-direct/concurrent-session-isolation.test.ts` and + * `shared/abort-coordinator.test.ts`. * * # Pending abort drain * @@ -34,14 +42,15 @@ * * # What this module does NOT own * - * - `state.closed` (the boolean) lives in `SessionState` because sub- - * generators read it on every loop iteration. The coordinator only - * owns the **promise side** of close — the field and the promise are - * updated together by the orchestrator's `close()` method. - * - Whether to *check* `state.closed` after `await coordinator.()` - * is the caller's job. The coordinator only coordinates abort. + * - The `closed` **boolean** stays on the caller (anthropic-direct keeps it + * in `SessionState`, openai-compatible on the query object) because sub- + * generators read it on every loop iteration. The coordinator only owns + * the **promise side** of close — the field and the promise are updated + * together by the orchestrator's `close()` method. + * - Whether to *check* `closed` after `await coordinator.()` is the + * caller's job. The coordinator only coordinates abort. * - * @module agent/providers/anthropic-direct/query/abort-coordinator + * @module agent/providers/shared/abort-coordinator */ export type AbortReason = 'interrupted' | 'closed';