From 36f9e1eaa8639dc84408f4a72b90ad664a2780de Mon Sep 17 00:00:00 2001 From: orientpine Date: Sun, 20 Sep 2026 13:11:46 -0400 Subject: [PATCH 1/6] fix(credential-pool): reuse prose subscription-limit failover Incorporate the combined implementation and regressions from senpi PR #1769 as a separately reviewable prerequisite for quota-aware internal routing. Preserve newer upstream tests and document the dependency. Constraint: Reuse https://github.com/code-yeongyu/senpi/pull/1769 rather than maintain a competing classifier Directive: Drop this prerequisite when PR #1769 is integrated Confidence: high Scope-risk: narrow --- packages/coding-agent/src/core/changes.md | 23 ++++++++++++++++++ .../src/core/credential-pool/classify.ts | 9 ++++++- .../test/credential-error-taxonomy.test.ts | 24 +++++++++++++++++++ .../test/credential-failover.test.ts | 24 +++++++++++++++++++ 4 files changed, 79 insertions(+), 1 deletion(-) diff --git a/packages/coding-agent/src/core/changes.md b/packages/coding-agent/src/core/changes.md index 6b4824273..5ee0c5596 100644 --- a/packages/coding-agent/src/core/changes.md +++ b/packages/coding-agent/src/core/changes.md @@ -391,6 +391,29 @@ The provider id is resolved inside the package before any extension loads, and t - LOW: the `editAssistantMessage` / `_navigateTree` heads in `agent-session.ts` (one added method and one widened parameter type); the fork-only `edited-user-message.ts`. - Coverage: `test/suite/tree-edit-user-message.test.ts`. +## 2026-09-21 - Reuse prose subscription-limit classification from PR #1769 + +### What changed + +- `packages/coding-agent/src/core/credential-pool/classify.ts`: recognizes the + narrow usage/session/weekly/daily/hourly limit vocabulary from PR #1769. +- The prerequisite retains that PR's taxonomy and failover regressions. + +### Why + +- The quota-routing feature needs the same classification when a subscription + limit is reported after preflight. Reusing the existing implementation avoids + a competing expression and keeps its dependency separately reviewable. + +### Why an extension could not handle it + +- Classification controls the native credential failover runner before an + extension could select a healthy replacement account. + +### Expected merge conflict zones + +- LOW: the rate-limit expression. This prerequisite incorporates PR #1769's + combined implementation/tests and should be dropped when that PR is integrated. ## 2026-09-20 - A fallback rung too small for the transcript is repaired, not rejected (senpi#1873) diff --git a/packages/coding-agent/src/core/credential-pool/classify.ts b/packages/coding-agent/src/core/credential-pool/classify.ts index cd4723eca..aa38fe8f7 100644 --- a/packages/coding-agent/src/core/credential-pool/classify.ts +++ b/packages/coding-agent/src/core/credential-pool/classify.ts @@ -40,7 +40,14 @@ export function rateLimitCooldown( const INVALID_KEY_TEXT = /invalid[ _-]?(?:api[ _-]?)?key|authentication[_ ]?error|invalid x-api-key|unauthorized/i; const ACCOUNT_SCOPED_403_TEXT = /account|credential|token|api[ _-]?key|organization|subscription/i; -const RATE_LIMIT_TEXT = /rate[ _-]?limit|too many requests|resource_exhausted/i; +// Subscription exhaustion reaches the pool as prose with no HTTP status: Codex reports +// `The usage limit has been reached`, Claude reports `You've hit your session limit`. +// The alternation stays narrow because this branch runs BEFORE the overflow branch - a +// broad ` limit` match would read the provider overflow prose catalogued in +// `utils/overflow.ts` (`token limit`, `context limit`, `byte limit`) as a rate limit and +// block a healthy credential for a request that was merely too large. +const RATE_LIMIT_TEXT = + /rate[ _-]?limit|too many requests|resource_exhausted|\b(?:usage|session|weekly|daily|hourly)[ _-]?limit\b/i; const BILLING_TEXT = /billing|credits?[ _-]?(?:required|exhausted|balance)|insufficient[ _-]?(?:funds|quota|credit)|payment[ _-]?required|quota[ _-]?exhausted/i; const OVERLOAD_TEXT = /overloaded/i; diff --git a/packages/coding-agent/test/credential-error-taxonomy.test.ts b/packages/coding-agent/test/credential-error-taxonomy.test.ts index 3e5b64b74..eb7b98863 100644 --- a/packages/coding-agent/test/credential-error-taxonomy.test.ts +++ b/packages/coding-agent/test/credential-error-taxonomy.test.ts @@ -38,6 +38,30 @@ describe("credential error taxonomy", () => { expect(action.block.retryAfterWasCapped).toBe(false); }); + test.each([ + ["codex usage limit", new Error("Codex error: The usage limit has been reached")], + ["claude session limit", new Error("You've hit your session limit \u00b7 resets 12am (Asia/Seoul)")], + ["weekly limit", new Error("You've hit your weekly limit \u00b7 resets 5am (Asia/Seoul)")], + ] as const)("%s fails over with a rate-limit cooldown", (_label, error) => { + // Subscription exhaustion reaches the pool as prose: no HTTP status, no + // `rate limit` wording. Without this vocabulary the failure default-denies + // to `fail_request`, so the exhausted credential is never blocked and the + // healthy sibling is never attempted. + const action = classifyCredentialFailure(error); + expect(action.kind).toBe("failover"); + if (action.kind !== "failover" || action.block.reason !== "rate_limit") throw new Error("expected rate_limit"); + expect(action.block.cooldownMs).toBe(COOLDOWN_BASE_MS); + }); + + test("overflow prose that names a limit still fails the request", () => { + // The rate-limit branch runs BEFORE the overflow branch, so the subscription + // vocabulary stays narrow: a provider that says `token limit` is reporting a + // context overflow, and blocking a healthy credential for it is wrong. + expect(classifyCredentialFailure(new Error("Your request exceeded model token limit: 262144")).kind).toBe( + "fail_request", + ); + }); + test.each([ ["529", status(529, "overloaded")], ["500", status(500, "Internal Server Error")], diff --git a/packages/coding-agent/test/credential-failover.test.ts b/packages/coding-agent/test/credential-failover.test.ts index 1f6f31f22..474b51d01 100644 --- a/packages/coding-agent/test/credential-failover.test.ts +++ b/packages/coding-agent/test/credential-failover.test.ts @@ -68,6 +68,30 @@ describe("generic credential failover runner", () => { ]); }); + test("a prose subscription limit rotates to the sibling credential", async () => { + const attempts: string[] = []; + const persisted: string[] = []; + const stream = runCredentialFailover({ + listSlots: () => [{ name: "alpha" }, { name: "beta" }], + select: firstAvailable, + runAttempt: (slot) => { + attempts.push(slot.name); + return slot.name === "alpha" + ? failWith(new Error("Codex error: The usage limit has been reached")) + : events({ type: "text_delta" }); + }, + isCommittedOutput: committedUnlessBookkeeping, + persistBlock: (slot, block) => { + persisted.push(`${slot.name}:${block.reason}`); + }, + }); + // A provider that reports exhaustion as prose must cost the pool exactly one + // slot, not the whole request. + expect(await collect(stream)).toEqual([{ type: "text_delta" }]); + expect(attempts).toEqual(["alpha", "beta"]); + expect(persisted).toEqual(["alpha:rate_limit"]); + }); + test("an UNKNOWN event type sets the committed-output barrier: no rotation, provider text kept verbatim", async () => { const attempts: string[] = []; const persisted: string[] = []; From 7a3d5513f191c6cd3c5fb5d9d50cba7d4559118c Mon Sep 17 00:00:00 2001 From: Daniel Choi Date: Sun, 20 Sep 2026 13:37:54 -0400 Subject: [PATCH 2/6] fix(credential-pool): honor Codex quota across internal requests Prefer confirmed included quota before paid credits, including cooling and leased accounts, and isolate account-switch observers from request execution. Route auxiliary model calls through session-scoped fallback while retaining explicit credential overrides and remote compaction's local fallback. Add portable regressions, safe session notices, API documentation, and fork change tracking. Preserve current-main model-switch and fallback behavior. Constraint: Unknown quota must never authorize paid usage Constraint: Explicit request keys retain upstream bypass semantics Rejected: Rotate requests carrying explicit external keys | violates caller intent Directive: Keep notification failures observational and credential-free Confidence: high Scope-risk: moderate Not-tested: Paid credits were simulated; no deliberate live credit consumption --- packages/coding-agent/CHANGELOG.md | 5 + packages/coding-agent/docs/extensions.md | 14 + packages/coding-agent/src/changes.md | 20 + .../coding-agent/src/core/agent-session.ts | 67 ++- packages/coding-agent/src/core/changes.md | 112 +++++ .../core/credential-pool/account-notices.ts | 57 +++ .../src/core/credential-pool/codex-quota.ts | 120 ++++++ .../core/credential-pool/rotation-stream.ts | 101 ++++- .../core/extensions/builtin/btw/changes.md | 14 + .../src/core/extensions/builtin/btw/index.ts | 16 +- .../extensions/builtin/compaction/changes.md | 23 ++ .../extensions/builtin/compaction/index.ts | 11 +- .../builtin/compaction/openai-remote.ts | 113 +++++- .../builtin/compaction/speculative-summary.ts | 22 +- .../builtin/compaction/speculative.ts | 9 +- .../core/extensions/builtin/look-at/runner.ts | 16 +- .../src/core/extensions/changes.md | 14 + .../src/core/extensions/runner.ts | 4 + .../coding-agent/src/core/extensions/types.ts | 2 + .../src/core/internal-model-request.ts | 193 +++++++++ .../coding-agent/src/core/model-runtime.ts | 74 +++- .../src/modes/interactive/changes.md | 22 + .../src/modes/interactive/interactive-mode.ts | 19 + packages/coding-agent/src/modes/print-mode.ts | 6 + .../before-compact-error-surfacing.test.ts | 1 + .../test/suite/account-notices.test.ts | 146 +++++++ .../suite/codex-auxiliary-routing.test.mjs | 382 ++++++++++++++++++ .../test/suite/codex-quota.test.ts | 125 ++++++ .../test/suite/codex-runtime-routing.test.ts | 158 ++++++++ .../test/suite/internal-model-request.test.ts | 210 ++++++++++ 30 files changed, 2041 insertions(+), 35 deletions(-) create mode 100644 packages/coding-agent/src/core/credential-pool/account-notices.ts create mode 100644 packages/coding-agent/src/core/credential-pool/codex-quota.ts create mode 100644 packages/coding-agent/src/core/internal-model-request.ts create mode 100644 packages/coding-agent/test/suite/account-notices.test.ts create mode 100644 packages/coding-agent/test/suite/codex-auxiliary-routing.test.mjs create mode 100644 packages/coding-agent/test/suite/codex-quota.test.ts create mode 100644 packages/coding-agent/test/suite/codex-runtime-routing.test.ts create mode 100644 packages/coding-agent/test/suite/internal-model-request.test.ts diff --git a/packages/coding-agent/CHANGELOG.md b/packages/coding-agent/CHANGELOG.md index 6979538de..4b0742fbd 100644 --- a/packages/coding-agent/CHANGELOG.md +++ b/packages/coding-agent/CHANGELOG.md @@ -10,6 +10,11 @@ ### Fixed +- Automatic Codex OAuth account selection now checks included quota before using paid credits, including cooling and temporarily leased accounts. Unknown quota cannot authorize paid usage. +- Title generation, compaction, branch summaries, side queries, and vision requests retain account rotation and configured model fallback without changing the active chat model. Explicit request keys remain pinned, and remote compaction keeps its local-summary fallback. +- Account switches are reported in the terminal and session events without credential material. Failing or self-removing notice listeners cannot interrupt routing or other listeners. +- Subscription-limit prose triggers credential failover rather than ending the request ([#1769](https://github.com/code-yeongyu/senpi/pull/1769) by [@orientpine](https://github.com/orientpine)). + ### Removed ## [2026.9.22-4] - 2026-09-22 diff --git a/packages/coding-agent/docs/extensions.md b/packages/coding-agent/docs/extensions.md index 61d84a006..28984ac1d 100644 --- a/packages/coding-agent/docs/extensions.md +++ b/packages/coding-agent/docs/extensions.md @@ -1233,6 +1233,20 @@ if (usage && usage.tokens > 100_000) { } ``` +### ctx.getRetryFallbackSettings() + +Returns the resolved retry/fallback policy for the current session, including +session overrides. Auxiliary model requests should use this policy rather than +loading global settings independently. The method is optional for contexts +supplied by older integrations. + +```typescript +const policy = ctx.getRetryFallbackSettings?.(); +if (policy?.modelFallback) { + // Apply the session's configured fallback policy. +} +``` + ### ctx.compact() Trigger compaction without awaiting completion. Use `onComplete` and `onError` for follow-up actions. diff --git a/packages/coding-agent/src/changes.md b/packages/coding-agent/src/changes.md index 270d5331e..20e5b7dce 100644 --- a/packages/coding-agent/src/changes.md +++ b/packages/coding-agent/src/changes.md @@ -137,6 +137,26 @@ ### Expected merge conflict zones - `packages/coding-agent/src/modes/print-mode.ts`: navigation and assistant-edit neighbours inside `commandContextActions`. +## 2026-09-20 - Print mode renders account-switch notices + +### What changed + +- `packages/coding-agent/src/modes/print-mode.ts`: the session subscription adds + `account_failover` (formatted via `formatAccountSwitchNotice`) and + `internal_model_fallback` to the stderr notice handling. + +### Why + +- Print/json mode is non-interactive; account switches still need a stderr line consistent + with the other model-fallback notices. + +### Why an extension could not handle it + +- One-shot mode prints engine events outside any extension surface. + +### Expected merge conflict zones + +- LOW: the two new `else if` branches in the subscription callback. ## 2026-09-20 - Name what the startup timing table measures before the stdin read (senpi#1868 follow-up) diff --git a/packages/coding-agent/src/core/agent-session.ts b/packages/coding-agent/src/core/agent-session.ts index 40d4ae263..213f13cf8 100644 --- a/packages/coding-agent/src/core/agent-session.ts +++ b/packages/coding-agent/src/core/agent-session.ts @@ -42,6 +42,7 @@ import { import type { Api, AssistantMessage, + AssistantMessageEventStream, AuthResult, Context, ImageContent, @@ -109,6 +110,7 @@ import { CompactionLifecycleCoordinator, type CompactionLifecycleState } from ". import { isTurnStuckOnContextOverflow } from "./compaction/stuck-overflow.ts"; import { isWarmSummaryAnchorValid } from "./compaction/warm-anchor.ts"; import type { CompactionModelSelector } from "./compaction-settings-access.ts"; +import { emitAccountSwitch, subscribeAccountSwitch } from "./credential-pool/account-notices.ts"; import { admitCursorHistory, cursorAdmissionBudgetBytes } from "./cursor-history-admission.ts"; import { DEFAULT_THINKING_LEVEL } from "./defaults.ts"; import { resolveDiscoveredResourcePaths } from "./discovered-resource-scope.ts"; @@ -203,6 +205,7 @@ import type { } from "./extensions/types.ts"; import { normalizeToolExposure, RUNTIME_EXTENSION_PATH } from "./extensions/types.ts"; import { shouldWarnHighReasoning } from "./high-reasoning-warning.ts"; +import { streamInternalModel } from "./internal-model-request.ts"; import { isManualContinueSubmission, MANUAL_CONTINUE_CUSTOM_TYPE, @@ -447,6 +450,23 @@ export type AgentSessionEvent = to: string; chainConfigured: boolean; } + | { + type: "account_failover"; + provider: string; + from: string; + to: string; + reason: string; + sessionId?: string; + source?: string; + } + | { + type: "internal_model_fallback"; + source: string; + from: string; + to: string; + reason: string; + chainKey: string; + } // Auth login flow (task 13) is additive with event-only completion. The // login_start command responds immediately, then the OAuth URL and the // terminal result arrive here, because an interactive browser round-trip @@ -876,6 +896,7 @@ export class AgentSession { // Event subscription state private _unsubscribeAgent?: () => void; private _unsubscribeSettingsSource?: () => void; + private _unsubscribeAccountSwitch?: () => void; private _eventListeners: AgentSessionEventListener[] = []; private _agentEventQueue: Promise = Promise.resolve(); /** @@ -1073,6 +1094,18 @@ export class AgentSession { this._unsubscribeSettingsSource = this.settingsManager.subscribeToSourceSelection((source) => { this._emit({ type: "settings_source_selected", ...source }); }); + this._unsubscribeAccountSwitch = subscribeAccountSwitch((event) => { + if (event.sessionId !== this.sessionId) return; + this._emit({ + type: "account_failover", + provider: event.provider, + from: event.from, + to: event.to, + reason: event.reason, + ...(event.sessionId === undefined ? {} : { sessionId: event.sessionId }), + ...(event.source === undefined ? {} : { source: event.source }), + }); + }); const noModelFallback = config.resourceLoader.getExtensions().runtime.flagValues.get("no-model-fallback") === true || envValue("NO_FALLBACK") === "1"; @@ -1590,6 +1623,10 @@ export class AgentSession { private _emit(event: AgentSessionEvent): void { this._logSessionEvent(event); + if (event.type === "account_failover") { + emitAccountSwitch(event, this._eventListeners); + return; + } for (const l of this._eventListeners) { l(event); } @@ -2930,6 +2967,8 @@ export class AgentSession { this._disconnectFromAgent(); this._unsubscribeSettingsSource?.(); this._unsubscribeSettingsSource = undefined; + this._unsubscribeAccountSwitch?.(); + this._unsubscribeAccountSwitch = undefined; this._unsubscribeWakeSources?.(); this._unsubscribeWakeSources = undefined; this._eventListeners = []; @@ -4286,7 +4325,7 @@ export class AgentSession { baseOptions: this._buildSessionTitleBaseOptions(), retry: sessionTitleRetryPolicy(this.settingsManager.getRetrySettings()), signal: abortController.signal, - streamFn: this.agent.streamFunction, + streamFn: (model, context, options) => this._streamInternalModel(model, context, options, "title"), }); if (abortController.signal.aborted) { return; @@ -4330,6 +4369,27 @@ export class AgentSession { }; } + private _streamInternalModel( + model: Model, + context: Context, + options: SimpleStreamOptions = {}, + purpose: string, + ): AssistantMessageEventStream | Promise { + return streamInternalModel( + this._modelRuntime, + model, + context, + { ...options, purpose, affinitySessionId: this.sessionId }, + { + settings: this.settingsManager, + cooldowns: this._selectorCooldowns, + streamFn: this.agent.streamFunction, + notify: (event) => this._emit(event), + agentDir: this._agentDir, + }, + ); + } + /** * Internal: Queue a steering message (already expanded, no extension command check). */ @@ -5853,7 +5913,7 @@ export class AgentSession { signal, extraBody, this.thinkingLevel, - this.agent.streamFunction, + (model, context, options) => this._streamInternalModel(model, context, options, "compaction"), env, this.agent.transformContext, this.settingsManager.getRetrySettings(), @@ -9210,7 +9270,8 @@ export class AgentSession { customInstructions, replaceInstructions, reserveTokens: branchSummarySettings.reserveTokens, - streamFn: this.agent.streamFunction, + streamFn: (model, context, options) => + this._streamInternalModel(model, context, options, "branch summary"), retry: this.settingsManager.getRetrySettings(), callbacks: this._summarizationRetryCallbacks({ source: "branchSummary", diff --git a/packages/coding-agent/src/core/changes.md b/packages/coding-agent/src/core/changes.md index 5ee0c5596..c706c22b6 100644 --- a/packages/coding-agent/src/core/changes.md +++ b/packages/coding-agent/src/core/changes.md @@ -391,6 +391,118 @@ The provider id is resolved inside the package before any extension loads, and t - LOW: the `editAssistantMessage` / `_navigateTree` heads in `agent-session.ts` (one added method and one widened parameter type); the fork-only `edited-user-message.ts`. - Coverage: `test/suite/tree-edit-user-message.test.ts`. +## 2026-09-20 - Auxiliary requests stream through one internal fallback seam + +### What changed + +- `packages/coding-agent/src/core/internal-model-request.ts` (new): `streamInternalModel` gives title, compaction, branch summary, /btw, and look_at a per-request fallback lane that never mutates the agent's selected chat model. It strips a pre-resolved Codex OAuth `apiKey` only at the internal seam, drops `apiKey`/`headers`/`extraBody`/`env`/`reasoningEffort` on a cross-provider fallback, and excludes text-only fallback models while the context contains images. +- `packages/coding-agent/src/core/agent-session.ts`: new `_streamInternalModel` helper and the title, blocking compaction, and branch-summary `streamFn` call sites now route through it. +- `packages/coding-agent/src/core/extensions/types.ts` and `runner.ts`: the extension context gains an optional `getRetryFallbackSettings()` so builtins pass session settings into the helper instead of deriving a global `SettingsManager`. +- `compaction/speculative-summary.ts`, `compaction/openai-remote.ts`, `compaction/index.ts`, `btw/index.ts`, and `look-at/runner.ts`: auxiliary lanes use the helper; remote Codex compaction uses the public `ModelRuntime.requestWithCredentialRotation` seam. +- `packages/coding-agent/test/suite/internal-model-request.test.ts` and + `packages/coding-agent/test/suite/codex-auxiliary-routing.test.mjs`: portable + helper and native title/summary/remote-compaction/side-query/vision regressions. +- `packages/coding-agent/test/compaction/before-compact-error-surfacing.test.ts`: + the session fixture now supplies its required `getSessionId` contract. +- When no session fallback policy is provided or it is disabled, the internal + helper delegates the original stream directly, preserving rejection identity + and cancellation rather than creating another error-conversion layer. + +### Why + +Auxiliary lanes previously called the runtime directly. Routing them through this seam while keeping their selector local means a resolved credential or provider-specific body can never cross into a different provider, and the failures never re-select the whole session. + +### Why an extension could not handle it + +Title and branch-summary generation originate inside AgentSession, and native +remote compaction requires the selected credential before its HTTP request. +The shared internal seam covers these core requests and builtins together. + +### Expected merge conflict zones + +- LOW: the additive import plus `_streamInternalModel` and the three `streamFn` call sites in `agent-session.ts`, the new `internal-model-request.ts`, and the `getRetryFallbackSettings` addition in `extensions/types.ts`/`runner.ts`. + +## 2026-09-20 - Codex quota admission precedes paid usage + +### What changed + +- `packages/coding-agent/src/core/credential-pool/codex-quota.ts`: validates fresh + read-only usage responses and distinguishes included quota, confirmed + exhaustion, and unknown availability. Credits are admitted only after every + normal bucket is confirmed exhausted. +- `packages/coding-agent/src/core/credential-pool/rotation-stream.ts`: checks + the full account inventory, including cooling accounts and accounts leased + by another recovery probe, before filtering generation candidates. Streams + and non-streaming requests share selection, health persistence, and notices. +- `packages/coding-agent/src/core/model-runtime.ts`: lazily loads quota readers, + resolves OAuth through native account-specific refresh, preserves explicit + request-key bypass, carries affinity/purpose, and provides the typed + `requestWithCredentialRotation` HTTP seam. Payload provenance uses the + selected Codex credential, not a stale caller snapshot. +- `packages/coding-agent/src/core/credential-pool/classify.ts`: recognizes + subscription usage-limit prose as an account rate limit. This overlaps the + narrow classification repair proposed in upstream PR #1769. +- Portable regressions: + `packages/coding-agent/test/suite/codex-quota.test.ts`, + `packages/coding-agent/test/suite/codex-runtime-routing.test.ts`, and + `packages/coding-agent/test/suite/account-notices.test.ts`. + +### Why + +- Generation cooldown and probe ownership are not evidence of exhausted + quota. Omitting those accounts could spend credits while normal quota + remained. Unknown or denied quota must not authorize spending either. +- Explicit keys are a caller contract; only internal request adapters may + discard a previously resolved OAuth snapshot to delegate account selection. + +### Why an extension could not handle it + +- Account admission, credential refresh, payload provenance, and pre-output + failover belong to ModelRuntime and must also cover noninteractive workers. + +### Expected merge conflict zones + +- MEDIUM: ModelRuntime request preparation and credential-pool selection. +- LOW: new quota module and portable regressions. Coordinate the classifier + line with PR #1769 rather than submitting two independent copies. + +## 2026-09-20 - Account-switch notices are fault-isolated + +### What changed + +- `packages/coding-agent/src/core/credential-pool/account-notices.ts` (new): the + `emitAccountSwitch` / `subscribeAccountSwitch` / `formatAccountSwitchNotice` surface. + `emitAccountSwitch` wraps each subscriber in its own try/catch and attaches a rejection + handler to async observers. Failures produce a fixed, credential-free diagnostic, + never stop provider selection, never drop later observers, and never surface exception text. + Delivery snapshots the observer collection so a failing observer that removes + itself cannot shift a session listener array and skip its next peer. +- `packages/coding-agent/src/core/agent-session.ts`: each session subscribes at construction + and converts a module-level `account_failover` whose `sessionId` matches into a session + event, forwarding `provider`/`from`/`to`/`reason` plus the optional `sessionId`/`source`. + The `AgentSessionEvent` union gains `account_failover` and `internal_model_fallback`. + `dispose()` unsubscribes the account-switch listener alongside the settings-source listener. + +### Why + +- The credential-pool selector notifies these listeners from its synchronous path. A throwing + listener previously aborted selection with zero provider attempts and silenced every later + observer; a rejected async observer became an unhandled rejection. The session-scoped + subscription keeps one account switch from leaking across sessions, and dispose keeps the + module-level set from retaining dead sessions. + +### Why an extension could not handle it + +- The emit runs inside the credential-pool selector before any provider attempt, and the + constructor/dispose subscription in `agent-session.ts` sits below every extension of the + session. Neither seam is reachable from extension code. + +### Expected merge conflict zones + +- LOW: the new `credential-pool/account-notices.ts` (fork-only) and the additive seams in + `agent-session.ts` - the import block, the `AgentSessionEvent` union entries, the + constructor subscription, and the dispose cleanup. + ## 2026-09-21 - Reuse prose subscription-limit classification from PR #1769 ### What changed diff --git a/packages/coding-agent/src/core/credential-pool/account-notices.ts b/packages/coding-agent/src/core/credential-pool/account-notices.ts new file mode 100644 index 000000000..dbda36a80 --- /dev/null +++ b/packages/coding-agent/src/core/credential-pool/account-notices.ts @@ -0,0 +1,57 @@ +export interface AccountSwitchNotice { + type: "account_failover"; + provider: string; + from: string; + to: string; + reason: string; + sessionId?: string; + source?: string; +} + +export type AccountSwitchListener = (event: AccountSwitchNotice) => void | PromiseLike; + +const listeners = new Set(); + +function reportListenerFailure(): void { + console.error("Account-switch notice observer failed"); +} + +/** + * Emit an account switch to every subscriber in isolation. A synchronous throw + * or async rejection must never stop provider selection, prevent later + * subscribers from observing the switch, or surface exception text that could + * carry credential material. + */ +export function emitAccountSwitch( + event: AccountSwitchNotice, + observers: Iterable = listeners, +): void { + for (const listener of [...observers]) { + try { + const result = listener(event); + if (result) { + void Promise.resolve(result).catch(reportListenerFailure); + } + } catch { + reportListenerFailure(); + } + } +} + +export function subscribeAccountSwitch(listener: AccountSwitchListener): () => void { + listeners.add(listener); + return () => { + listeners.delete(listener); + }; +} + +export function formatAccountSwitchNotice({ provider, from, to, reason, source }: AccountSwitchNotice): { + title: string; + why: string; +} { + const why = [`Reason: ${reason}`, source === undefined ? "" : `Source: ${source}`].filter(Boolean).join(" · "); + return { + title: `Account fallback: ${provider}/${from} -> ${to}`, + why, + }; +} diff --git a/packages/coding-agent/src/core/credential-pool/codex-quota.ts b/packages/coding-agent/src/core/credential-pool/codex-quota.ts new file mode 100644 index 000000000..1df5399fc --- /dev/null +++ b/packages/coding-agent/src/core/credential-pool/codex-quota.ts @@ -0,0 +1,120 @@ +import { z } from "zod"; +import type { RotationSlot, RotationSources } from "./rotation-stream.ts"; + +const windowSchema = z.object({ + used_percent: z.number().nonnegative(), + reset_at: z.number().nonnegative().optional(), +}); +const limitSchema = z.object({ + allowed: z.boolean(), + limit_reached: z.boolean(), + primary_window: windowSchema.nullable(), + secondary_window: windowSchema.nullable().optional(), +}); +const usageSchema = z.object({ + rate_limit: limitSchema, + additional_rate_limits: z + .array( + z.object({ + normal_model_slug: z.string().optional(), + rate_limit: limitSchema, + }), + ) + .nullable() + .optional(), + credits: z + .object({ + has_credits: z.boolean(), + unlimited: z.boolean().optional(), + overage_limit_reached: z.boolean().optional(), + balance: z.union([z.string(), z.number()]).nullable().optional(), + }) + .nullable() + .optional(), +}); + +/** + * Read-only quota admission. Token resolution remains owned by ModelRuntime, + * including its file-locked OAuth refresh. Never persist a token or WHAM body. + */ +export async function fetchCodexUsage(token: string | undefined, signal?: AbortSignal): Promise { + if (!token) throw new Error("Codex quota admission unavailable: missing OAuth credential"); + const claims = z + .object({ "https://api.openai.com/auth": z.object({ chatgpt_account_id: z.string().optional() }).optional() }) + .parse(JSON.parse(Buffer.from(token.split(".")[1] ?? "", "base64url").toString())); + const accountId = claims["https://api.openai.com/auth"]?.chatgpt_account_id; + const response = await fetch("https://chatgpt.com/backend-api/wham/usage", { + headers: { + Authorization: `Bearer ${token}`, + ...(accountId ? { "chatgpt-account-id": accountId } : {}), + }, + signal: signal ? AbortSignal.any([signal, AbortSignal.timeout(15_000)]) : AbortSignal.timeout(15_000), + }); + if (!response.ok) throw new Error(`Codex quota admission unavailable (HTTP ${response.status})`); + return response.json(); +} + +/** Annotate each candidate; selection must rank only after native health filtering. */ +export async function admitCodexQuota( + sources: Pick, + slots: RotationSlot[], +): Promise { + if (sources.providerId !== "openai-codex") return slots; + const getUsage = sources.getCodexUsage; + if (!getUsage) throw new Error("Codex quota admission unavailable: missing quota reader"); + const assessed = await Promise.all( + slots.map(async (slot): Promise => { + if (slot.blockReason === "auth_error" || slot.blockReason === "account_disabled") + return { ...slot, quotaUnavailable: true }; + let parsed: ReturnType; + try { + parsed = usageSchema.safeParse(await getUsage(slot)); + } catch (error) { + if (sources.signal?.aborted) throw sources.signal.reason; + if (error instanceof Error && error.name === "AbortError") throw error; + return { ...slot, blockReason: "account_disabled", quotaUnavailable: true }; + } + if (!parsed.success) return { ...slot, blockReason: "account_disabled", quotaUnavailable: true }; + const usage = parsed.data; + const limit = usage.rate_limit; + const limits = [ + limit, + ...(usage.additional_rate_limits ?? []) + .filter((entry) => sources.modelId && entry.normal_model_slug === sources.modelId) + .map((entry) => entry.rate_limit), + ]; + const included = limits.some( + (entry) => + entry.allowed && + !entry.limit_reached && + [entry.primary_window, entry.secondary_window].every((window) => !window || window.used_percent < 100), + ); + const exhausted = limits.every( + (entry) => + entry.limit_reached || + [entry.primary_window, entry.secondary_window].some((window) => window && window.used_percent >= 100), + ); + const credits = + usage.credits?.has_credits === true && + usage.credits?.overage_limit_reached !== true && + (usage.credits?.unlimited === true || Number(usage.credits?.balance) > 0); + // A credit-enabled account can still have included quota. Preserve pins + // and affinity within the included tier; paid-only accounts come last. + if (included) return { ...slot, quotaTier: 0 }; + if (credits && exhausted) return { ...slot, quotaTier: 1 }; + // This is an admission result, not a permanent credential-store block. + // The next request rechecks WHAM, so a reset/replenishment is seen immediately. + return { ...slot, blockReason: "account_disabled", quotaUnavailable: !exhausted }; + }), + ); + // Unknown is not exhausted: healthy included quota may proceed, but an + // unavailable quota response must never authorize spending paid credits. + if (assessed.some((slot) => slot.quotaUnavailable || slot.quotaTier === 0)) + return assessed.map((slot) => (slot.quotaTier === 1 ? { ...slot, blockReason: "account_disabled" } : slot)); + return assessed; +} + +export function preferredCodexQuotaTier(candidates: readonly RotationSlot[]): RotationSlot[] { + const tier = Math.min(...candidates.map((slot) => slot.quotaTier ?? 0)); + return candidates.filter((slot) => (slot.quotaTier ?? 0) === tier); +} diff --git a/packages/coding-agent/src/core/credential-pool/rotation-stream.ts b/packages/coding-agent/src/core/credential-pool/rotation-stream.ts index 00ee72524..063174b09 100644 --- a/packages/coding-agent/src/core/credential-pool/rotation-stream.ts +++ b/packages/coding-agent/src/core/credential-pool/rotation-stream.ts @@ -3,10 +3,13 @@ import type { AssistantMessageEvent, Credential } from "@earendil-works/pi-ai"; import { rendezvousOrder, type SlotHasher } from "@earendil-works/pi-ai/auth/pool/select"; import { listSlots as listCredentialSlots, type PooledCredential } from "@earendil-works/pi-ai/auth/pool/slots"; import { resolveConfigValue } from "../resolve-config-value.ts"; +import { emitAccountSwitch } from "./account-notices.ts"; import { type CredentialBlock, classifyCredentialFailure } from "./classify.ts"; +import { admitCodexQuota, preferredCodexQuotaTier } from "./codex-quota.ts"; import { discoverEnvSlots } from "./env-slots.ts"; import { type RunSlot, runCredentialFailover } from "./failover.ts"; import { isCommittedRotationOutput, isRotationStreamStart, rotationErrorFromEvent } from "./rotation-events.ts"; +import { type RunCredentialFailoverOptions, type RunSlot, runCredentialFailover } from "./failover.ts"; import { acquireHalfOpenLease, type CredentialSlotRepository, type CredentialSlotState } from "./state-store.ts"; /** The exact hash the claude-sdk-oauth affinity oracle uses, so pools never remap. */ @@ -16,6 +19,8 @@ export type RotationLane = "stored" | "env"; export type RotationSlot = RunSlot & { lane: RotationLane; + quotaTier?: 0 | 1; + quotaUnavailable?: boolean; /** Env-lane key material for the attempt; never serialized or persisted. */ envKey?: string; envVarName?: string; @@ -25,6 +30,12 @@ export type RotationSlot = RunSlot & { export type RotationSources = { providerId: string; + modelId?: string; + signal?: AbortSignal; + sessionId?: string; + source?: string; + selectionState?: Map; + getCodexUsage?: (slot: RotationSlot) => Promise; credential: Credential | undefined; env: (name: string) => string | undefined; repository: CredentialSlotRepository; @@ -217,27 +228,103 @@ export type CredentialRotationOptions = { export function streamWithCredentialRotation( options: CredentialRotationOptions, ): AsyncGenerator { + return runRotation({ + ...options, + isCommittedOutput: isCommittedRotationOutput, + isStreamStart: isRotationStreamStart, + errorFromEvent: rotationErrorFromEvent, + }); +} + +/** Non-streaming provider requests share the same admission, health, and notices. */ +export async function requestWithCredentialRotation( + options: Omit & { runAttempt: (slot: RotationSlot) => Promise }, +): Promise { + let result: { value: T } | undefined; + for await (const event of runRotation({ + ...options, + runAttempt: async function* (slot) { + yield { value: await options.runAttempt(slot) }; + }, + isCommittedOutput: () => true, + isStreamStart: () => false, + })) { + result = event; + } + if (!result) throw new Error("Credential request completed without a result"); + return result.value; +} + +function runRotation( + options: Omit & + Pick, "runAttempt" | "isCommittedOutput" | "errorFromEvent">, +): AsyncGenerator { const { sources, runAttempt } = options; const hasher = options.hasher ?? sha256SlotHasher; const affinityKey = options.affinityKey ?? randomUUID(); const useAffinity = sources.policy?.affinity !== false; const now = sources.now ?? Date.now; + const selectionKey = `${sources.providerId}\0${sources.sessionId ?? affinityKey}`; + let previous = sources.selectionState?.get(selectionKey); + let previousFailure: string | undefined; + let admitted: RotationSlot[] = []; - return runCredentialFailover({ - listSlots: () => listRotationSlots(sources), + return runCredentialFailover({ + listSlots: async () => { + const slots = await listRotationSlots(sources); + // A different caller's probe lease hides a generation candidate, not + // its remaining normal quota. Assess the full inventory before filtering. + const inventory = + sources.providerId === "openai-codex" ? await listRotationSlots(sources, { acquireLeases: false }) : slots; + previous ??= + inventory.find((slot) => slot.pinned)?.name ?? + (useAffinity ? rendezvousOrder(affinityKey, inventory, hasher) : inventory)[0]?.name; + admitted = await admitCodexQuota(sources, inventory); + return admitted.filter((slot) => + slots.some((candidate) => candidate.name === slot.name && candidate.lane === slot.lane), + ); + }, select: (candidates) => { + candidates = preferredCodexQuotaTier(candidates); const pinned = candidates.find((candidate) => candidate.pinned === true); - if (pinned) return pinned; const ordered = useAffinity ? rendezvousOrder(affinityKey, candidates, hasher) : candidates; - const winner = ordered[0]; + const winner = pinned ?? ordered[0]; if (!winner) throw new Error("credential rotation selected from an empty candidate set"); + if (sources.providerId === "openai-codex" && previous && previous !== winner.name) { + const prior = admitted.find((slot) => slot.name === previous); + const reason = + previousFailure ?? + (prior?.quotaUnavailable + ? "quota unavailable" + : prior?.blockReason === "rate_limit" + ? "cooldown" + : prior?.blockReason === "account_disabled" + ? "quota exhausted" + : (prior?.blockReason ?? "account selection")); + emitAccountSwitch({ + type: "account_failover", + provider: sources.providerId, + from: previous, + to: winner.name, + reason: winner.quotaTier === 1 ? `${reason}; using extra usage` : reason, + sessionId: sources.sessionId, + source: sources.source, + }); + } + previous = winner.name; + previousFailure = undefined; + sources.selectionState?.set(selectionKey, winner.name); return winner; }, + onRotate: ({ slot, block }) => { + previous = slot.name; + previousFailure = block.reason; + }, runAttempt, - isCommittedOutput: isCommittedRotationOutput, - isStreamStart: isRotationStreamStart, - errorFromEvent: rotationErrorFromEvent, + isCommittedOutput: options.isCommittedOutput, + isStreamStart: options.isStreamStart, + errorFromEvent: options.errorFromEvent, classify: (error, context) => classifyCredentialFailure(error, { ...context, diff --git a/packages/coding-agent/src/core/extensions/builtin/btw/changes.md b/packages/coding-agent/src/core/extensions/builtin/btw/changes.md index 264702a16..0dbd03ec5 100644 --- a/packages/coding-agent/src/core/extensions/builtin/btw/changes.md +++ b/packages/coding-agent/src/core/extensions/builtin/btw/changes.md @@ -1,5 +1,19 @@ # changes — btw +## 2026-09-20 - Side query routes through the internal fallback seam + +### What changed + +- `index.ts`: the `/btw` stream function now calls `streamInternalModel` instead of `ctx.modelRegistry.modelRuntime.streamSimple`, passing `purpose`/session id and the session's settings/agent directory. A fallback notice shows the from/to transition without leaking credentials. + +### Why + +Side queries are auxiliary requests and must not handle their own selection sharing the main chat fallback; the internal seam keeps the primary model untouched and strips resolved credentials on cross-provider fallback. + +### Expected merge conflict zones + +- LOW: the `streamFn` in the command handler. + ## 2026-09-13 - Explicit off switch: bare /btw and kitty-safe Escape ### What changed diff --git a/packages/coding-agent/src/core/extensions/builtin/btw/index.ts b/packages/coding-agent/src/core/extensions/builtin/btw/index.ts index 212df6c75..ae2d3601b 100644 --- a/packages/coding-agent/src/core/extensions/builtin/btw/index.ts +++ b/packages/coding-agent/src/core/extensions/builtin/btw/index.ts @@ -1,4 +1,5 @@ import { isKeyRelease, matchesKey } from "@earendil-works/pi-tui"; +import { streamInternalModel } from "../../../internal-model-request.ts"; import { convertToLlm, filterContextExcludedMessages } from "../../../messages.ts"; import { buildSessionContext } from "../../../session-manager.ts"; import type { ExtensionAPI, ExtensionContext } from "../../types.ts"; @@ -117,7 +118,20 @@ export default function btwExtension(pi: ExtensionAPI) { sessionId, thinkingLevel: thinkingLevel === "off" ? undefined : thinkingLevel, streamFn: (streamModel, streamContext, options) => - ctx.modelRegistry.modelRuntime.streamSimple(streamModel, streamContext, options), + streamInternalModel( + ctx.modelRegistry.modelRuntime, + streamModel, + streamContext, + { ...options, purpose: "side query", affinitySessionId: sessionId }, + { + agentDir: ctx.agentDir, + settings: ctx.getRetryFallbackSettings + ? { getRetryFallbackSettings: ctx.getRetryFallbackSettings } + : undefined, + notify: (event) => + ctx.ui.notify(`Side query model fallback: ${event.from} -> ${event.to}`, "warning"), + }, + ), }, context, { diff --git a/packages/coding-agent/src/core/extensions/builtin/compaction/changes.md b/packages/coding-agent/src/core/extensions/builtin/compaction/changes.md index 3872e01e5..bd83522f5 100644 --- a/packages/coding-agent/src/core/extensions/builtin/compaction/changes.md +++ b/packages/coding-agent/src/core/extensions/builtin/compaction/changes.md @@ -36,6 +36,29 @@ The provider id is resolved inside the package before any extension loads, and t ### Expected merge conflict zones - These three files, against any other change to the remote-compaction provider gate. +## Auxiliary summarizer/remote compaction route through the internal fallback seam (2026-09-20) + +### What changed + +- `packages/coding-agent/src/core/extensions/builtin/compaction/speculative-summary.ts`: `summarizationStream` routes through `streamInternalModel`, carrying the purpose, session ID, injected fallback policy, agent directory, and runtime stream primitive. +- `packages/coding-agent/src/core/extensions/builtin/compaction/openai-remote.ts`: Codex compact-endpoint requests use `ModelRuntime.requestWithCredentialRotation` and selected-slot headers. Pool exhaustion yields to local summarization. Only prepared rotation attempts convert credential HTTP failures into errors, retaining the numeric status for classification; legacy and non-Codex callers retain their undefined/local-fallback result. +- `packages/coding-agent/src/core/extensions/builtin/compaction/index.ts`: the request rewrite uses resolved authorization metadata for selected-account provenance. +- `packages/coding-agent/src/core/extensions/builtin/compaction/speculative.ts`: the summarizer context accepts the injected agent directory and retry/fallback policy. + +### Why + +Compaction lanes previously streamed through `stream`/`runtime.streamSimple` directly. The shared seam avoids a pre-resolved credential crossing providers and lets account rotation remain on the public runtime primitive instead of private `prepareRequest` access. + +### Why an extension could not handle it + +These builtin consumers need the shared core account-selection and internal +request primitives; an external extension cannot replace the native title, +summary, and HTTP credential-selection paths consistently. + +### Expected merge conflict zones + +- LOW: the summarization dispatch, compact-endpoint request block, authorization + recovery, and optional context fields in the four paths listed above. ## Hold a model switch until the next send can compact for it (2026-09-20) diff --git a/packages/coding-agent/src/core/extensions/builtin/compaction/index.ts b/packages/coding-agent/src/core/extensions/builtin/compaction/index.ts index 70c368f3b..c3092a994 100644 --- a/packages/coding-agent/src/core/extensions/builtin/compaction/index.ts +++ b/packages/coding-agent/src/core/extensions/builtin/compaction/index.ts @@ -978,9 +978,18 @@ export default function compactionExtension( ...(auth.baseUrl ? { baseUrl: auth.baseUrl } : {}), } : model; + const selectedAuthorization = Object.entries(event.headers ?? {}).find( + ([key]) => key.toLowerCase() === "authorization", + )?.[1]; const headers = createOpenAiRemoteCompactionHeaders( effectiveModel, - { ...auth, headers: event.headers ?? auth.headers }, + { + ...auth, + ...(model.provider === "openai-codex" && selectedAuthorization?.startsWith("Bearer ") + ? { apiKey: selectedAuthorization.slice(7) } + : {}), + headers: event.headers ?? auth.headers, + }, ctx.sessionManager.getSessionId(), ); if (!headers) return undefined; diff --git a/packages/coding-agent/src/core/extensions/builtin/compaction/openai-remote.ts b/packages/coding-agent/src/core/extensions/builtin/compaction/openai-remote.ts index 30c8d44a9..0f7b766f8 100644 --- a/packages/coding-agent/src/core/extensions/builtin/compaction/openai-remote.ts +++ b/packages/coding-agent/src/core/extensions/builtin/compaction/openai-remote.ts @@ -11,6 +11,7 @@ import { import { streamSimple } from "@earendil-works/pi-ai/compat"; import type { CompactionResult } from "../../../compaction/index.ts"; import { convertToLlm } from "../../../messages.ts"; +import type { ModelRuntime } from "../../../model-runtime.ts"; import { buildContextEntries, buildSessionContext, @@ -113,7 +114,10 @@ type OpenAiRemoteCompactionContext = { error: string; } >; - modelRuntime?: { streamSimple: OpenAiResponsesStreamRunner }; + modelRuntime?: { + streamSimple: OpenAiResponsesStreamRunner; + requestWithCredentialRotation?: ModelRuntime["requestWithCredentialRotation"]; + }; }; serviceTier: ServiceTier | undefined; sessionManager: { @@ -779,19 +783,100 @@ export async function runOpenAiRemoteCompaction( reason: REMOTE_COMPACTION_TIMEOUT_REASON, transport: "compact-endpoint", }), - run: (signal) => - runOpenAiCompactEndpointCompaction({ - fetchImpl: dependencies.fetch ?? fetch, - headers: requestHeaders, - model: requestModel, - request: transformedRequest, - requestId: event.requestId, - signal, - firstKeptEntryId: event.preparation.firstKeptEntryId, - now: dependencies.now ?? Date.now, - emit, - origin, - }), + run: async (signal) => { + const execute = async (prepared?: { + model: Model; + options: { apiKey?: string; headers?: ProviderHeaders }; + }) => { + const headers = prepared + ? createOpenAiRemoteCompactionHeaders( + requestModel, + { apiKey: prepared.options.apiKey, headers: prepared.options.headers }, + request.body.prompt_cache_key, + ) + : requestHeaders; + if (!headers) { + emit?.({ + version: 1, + action: "remote_fallback", + route: "builtin.compaction.openai_remote", + requestId: event.requestId, + modelId: requestModel.id, + reason: "missing-openai-auth", + transport: "compact-endpoint", + }); + return undefined; + } + const resolvedOrigin = prepared ? openAiRemoteCompactionOrigin(requestModel, headers) : origin; + if (!resolvedOrigin) { + emit?.({ + version: 1, + action: "remote_fallback", + route: "builtin.compaction.openai_remote", + requestId: event.requestId, + modelId: requestModel.id, + reason: MISSING_REMOTE_REPLAY_ORIGIN_REASON, + transport: "compact-endpoint", + }); + return undefined; + } + let credentialFailure: Error | undefined; + const result = await runOpenAiCompactEndpointCompaction({ + fetchImpl: async (...args) => { + const response = await (dependencies.fetch ?? fetch)(...args); + if (prepared && [401, 402, 403, 429].includes(response.status)) { + const hint = response.headers.get("retry-after"); + credentialFailure = Object.assign( + new Error( + `${response.status}: ${await response.clone().text()}${hint ? ` (retry-after: ${hint})` : ""}`, + ), + { status: response.status }, + ); + } + return response; + }, + headers, + model: requestModel, + request: transformedRequest, + requestId: event.requestId, + signal, + firstKeptEntryId: event.preparation.firstKeptEntryId, + now: dependencies.now ?? Date.now, + emit, + origin: resolvedOrigin, + }); + if (credentialFailure) throw credentialFailure; + return result; + }; + + const runtime = ctx.modelRegistry?.modelRuntime; + if (requestModel.provider === "openai-codex" && runtime?.requestWithCredentialRotation) { + try { + return await runtime.requestWithCredentialRotation( + requestModel, + { + signal, + sessionId: ctx.sessionManager.getSessionId(), + purpose: "remote compaction", + headers: transformedHeaders, + }, + (prepared) => execute(prepared), + ); + } catch (error) { + if (signal.aborted) throw error; + emit?.({ + version: 1, + action: "remote_fallback", + route: "builtin.compaction.openai_remote", + requestId: event.requestId, + modelId: requestModel.id, + reason: "Codex account pool unavailable", + }); + return undefined; + } + } + return execute(); + }, }); } diff --git a/packages/coding-agent/src/core/extensions/builtin/compaction/speculative-summary.ts b/packages/coding-agent/src/core/extensions/builtin/compaction/speculative-summary.ts index 114af8c4b..b73125cd3 100644 --- a/packages/coding-agent/src/core/extensions/builtin/compaction/speculative-summary.ts +++ b/packages/coding-agent/src/core/extensions/builtin/compaction/speculative-summary.ts @@ -17,6 +17,7 @@ import { DEFAULT_SUMMARIZATION_IDLE_TIMEOUT_MS, summarizationMaxDurationMs, } from "../../../compaction/stream-watchdog.ts"; +import { streamInternalModel } from "../../../internal-model-request.ts"; import { convertToLlm } from "../../../messages.ts"; import type { buildPrompt } from "./prompts.ts"; import { repairOrphanedToolResults } from "./repair-tool-pairs.ts"; @@ -99,7 +100,26 @@ function summarizationStream( options: StreamOptions & Record, ): AssistantMessageEventStream { const runtime = context.modelRegistry?.modelRuntime; - return runtime ? runtime.stream(model, requestContext, options) : stream(model, requestContext, options); + return runtime + ? streamInternalModel( + runtime, + model, + requestContext, + { + ...options, + purpose: "speculative compaction", + affinitySessionId: context.sessionManager?.getSessionId(), + }, + { + agentDir: context.agentDir, + settings: context.getRetryFallbackSettings + ? { getRetryFallbackSettings: context.getRetryFallbackSettings } + : undefined, + streamFn: (streamModel, streamContext, streamOptions) => + runtime.stream(streamModel, streamContext, streamOptions), + }, + ) + : stream(model, requestContext, options); } export async function generateSummaryMessage(options: { diff --git a/packages/coding-agent/src/core/extensions/builtin/compaction/speculative.ts b/packages/coding-agent/src/core/extensions/builtin/compaction/speculative.ts index d91d57e7b..23486ddf6 100644 --- a/packages/coding-agent/src/core/extensions/builtin/compaction/speculative.ts +++ b/packages/coding-agent/src/core/extensions/builtin/compaction/speculative.ts @@ -35,7 +35,12 @@ import { CredentialFailoverError, TURN_RETRY_SUPPRESSION_PREFIX } from "../../.. import { convertToLlm } from "../../../messages.ts"; import type { ModelRegistry } from "../../../model-registry.ts"; import type { ReadonlySessionManager } from "../../../session-manager.ts"; -import type { ApplyCompactionResult, ContextUsage, ProviderRequestPreparation } from "../../types.ts"; +import type { + ApplyCompactionResult, + ContextUsage, + ProviderRequestPreparation, + RetryFallbackSettings, +} from "../../types.ts"; import { pruneToolResults } from "./emergency-prune.ts"; import { allowOverflowRetry, @@ -79,6 +84,8 @@ export interface SpeculativeCompactionContext { model: Model | undefined; sessionManager: ReadonlySessionManager; modelRegistry?: ModelRegistry; + agentDir?: string; + getRetryFallbackSettings?(): RetryFallbackSettings; getContextUsage(): ContextUsage | undefined; getCompactionSettings?(): CompactionPreparation["settings"]; getMessageRevision(): number; diff --git a/packages/coding-agent/src/core/extensions/builtin/look-at/runner.ts b/packages/coding-agent/src/core/extensions/builtin/look-at/runner.ts index a52494a94..44bc5f504 100644 --- a/packages/coding-agent/src/core/extensions/builtin/look-at/runner.ts +++ b/packages/coding-agent/src/core/extensions/builtin/look-at/runner.ts @@ -10,6 +10,7 @@ import type { TextContent, UserMessage, } from "@earendil-works/pi-ai"; +import { streamInternalModel } from "../../../internal-model-request.ts"; import type { ExtensionContext } from "../../types.ts"; import type { NormalizedLookAtArgs } from "./arguments.ts"; import { loadLookAtInputs } from "./image-input.ts"; @@ -79,7 +80,20 @@ export async function runLookAt( }; const streamRunner = dependencies.streamRunner ?? - ((model, context, options) => ctx.modelRegistry.modelRuntime.streamSimple(model, context, options)); + ((model, context, options) => + streamInternalModel( + ctx.modelRegistry.modelRuntime, + model, + context, + { ...options, purpose: "vision analysis", affinitySessionId: ctx.sessionManager.getSessionId() }, + { + agentDir: ctx.agentDir, + settings: ctx.getRetryFallbackSettings + ? { getRetryFallbackSettings: ctx.getRetryFallbackSettings } + : undefined, + notify: (event) => ctx.ui.notify(`Vision model fallback: ${event.from} -> ${event.to}`, "warning"), + }, + )); const reasoning = toStreamReasoning(resolved.thinkingLevel); const response = await streamRunner( resolved.model, diff --git a/packages/coding-agent/src/core/extensions/changes.md b/packages/coding-agent/src/core/extensions/changes.md index bfa230edf..0c20eb867 100644 --- a/packages/coding-agent/src/core/extensions/changes.md +++ b/packages/coding-agent/src/core/extensions/changes.md @@ -90,6 +90,20 @@ - `packages/coding-agent/src/core/extensions/types.ts`: command context and action declarations beside `navigateTree` and `editAssistantMessage`. - `packages/coding-agent/src/core/extensions/runner.ts`: handler types, default fields, command binding/reset, and context injection. +## 2026-09-20 - Session retry-fallback settings are readable from the extension context + +### What changed + +- `src/core/extensions/types.ts`: `ExtensionContext` gains an optional `getRetryFallbackSettings(): RetryFallbackSettings` getter. +- `src/core/extensions/runner.ts`: `createContext()` exposes that getter from the already-bound `sessionSettingsFn`, so an extension auxiliary request reads the active session's resolved fallback configuration rather than reconstructing one from the global agent directory. + +### Why + +Title/compaction/btw/vision auxiliary requests need the session's configured fallback chains, including SDK in-memory settings, without each builtin importing `SettingsManager` and reaching for `getAgentDir()`. + +### Expected merge conflict zones + +- LOW: the new optional member in `ExtensionContext` and the one getter in `createContext()`. ## 2026-09-20 - Import attributes survive the CommonJS rewrite (#1864) diff --git a/packages/coding-agent/src/core/extensions/runner.ts b/packages/coding-agent/src/core/extensions/runner.ts index d17454601..816e2618b 100644 --- a/packages/coding-agent/src/core/extensions/runner.ts +++ b/packages/coding-agent/src/core/extensions/runner.ts @@ -1209,6 +1209,10 @@ export class ExtensionRunner { runner.assertActive(); return runner.getCompactionSettingsFn(); }, + getRetryFallbackSettings: () => { + runner.assertActive(); + return runner.sessionSettingsFn.getRetryFallbackSettings(); + }, getPromptCacheSafeWaitSeconds: () => { runner.assertActive(); return runner.getPromptCacheSafeWaitSecondsFn(); diff --git a/packages/coding-agent/src/core/extensions/types.ts b/packages/coding-agent/src/core/extensions/types.ts index c042be687..fa183db09 100644 --- a/packages/coding-agent/src/core/extensions/types.ts +++ b/packages/coding-agent/src/core/extensions/types.ts @@ -516,6 +516,8 @@ export interface ExtensionContext { getContextUsage(): ContextUsage | undefined; /** Get resolved compaction settings from global/project/user overrides. */ getCompactionSettings(): CompactionPreparation["settings"]; + /** Resolved retry-fallback settings for this session; injected so auxiliary requests use session config over globals. */ + getRetryFallbackSettings?(): RetryFallbackSettings; /** * Longest a tool may block in the foreground before the active model's prompt * cache expires, or `undefined` when no cache-derived budget applies. Reads the diff --git a/packages/coding-agent/src/core/internal-model-request.ts b/packages/coding-agent/src/core/internal-model-request.ts new file mode 100644 index 000000000..a78980d0e --- /dev/null +++ b/packages/coding-agent/src/core/internal-model-request.ts @@ -0,0 +1,193 @@ +import type { ThinkingLevel } from "@earendil-works/pi-agent-core"; +import { + type Api, + type AssistantMessageEvent, + type AssistantMessageEventStream, + type Context, + isContextOverflow, + lazyStream, + type Model, + type ModelsSimpleStreamOptions, +} from "@earendil-works/pi-ai"; +import { getAgentDir } from "../config.ts"; +import { RetryFallbackController } from "./retry-fallback/controller.ts"; +import { SelectorCooldowns } from "./retry-fallback/cooldown.ts"; +import { createFallbackLogger } from "./retry-fallback/log.ts"; +import type { ResolvedRetryFallbackSettings } from "./retry-fallback/settings.ts"; + +/** The subset of `ModelRuntime` an auxiliary request needs. `ModelRuntime` satisfies this shape. */ +export interface InternalStreamRuntime { + streamSimple(model: Model, context: Context, options?: ModelsSimpleStreamOptions): AssistantMessageEventStream; + getModel(providerId: string, modelId: string): Model | undefined; + getModels(): readonly Model[]; + isUsingOAuth(providerId: string): boolean; + isFallbackEligible(providerId: string): boolean; + hasConfiguredAuth(providerId: string): boolean; +} + +/** Simple-stream options plus the auxiliary fields and API-specific passthrough the request needs. */ +export type InternalStreamOptions = ModelsSimpleStreamOptions & { + reasoningEffort?: ThinkingLevel; + purpose?: string; + [key: string]: unknown; +}; + +export interface InternalModelFallbackEvent { + type: "internal_model_fallback"; + source: string; + from: string; + to: string; + reason: string; + chainKey: string; +} + +export type InternalStreamFunction = ( + model: Model, + context: Context, + options?: InternalStreamOptions, +) => AssistantMessageEventStream | Promise; + +export interface InternalModelSettings { + getRetryFallbackSettings(): ResolvedRetryFallbackSettings; +} + +export interface StreamInternalModelConfig { + settings?: InternalModelSettings; + cooldowns?: SelectorCooldowns; + streamFn?: InternalStreamFunction; + notify?: (event: InternalModelFallbackEvent) => void; + /** Injected agent directory; replaces the private `ModelRuntime.poolStatePath` read the installed build used. */ + agentDir?: string; +} + +const cooldownsByRuntime = new WeakMap(); + +/** + * Streams an auxiliary request through the account-fallback lane without changing the + * caller's configured chat model. Credential material is stripped exactly at the + * internal seams: pre-resolved Codex OAuth before dispatch, and all resolved + * provider-specific material on a cross-provider fallback. Explicit external + * `ModelRuntime` `apiKey` calls are unaffected. + */ +export function streamInternalModel( + runtime: InternalStreamRuntime, + model: Model, + context: Context, + options?: InternalStreamOptions, + config?: Omit & { + streamFn?: (...args: Parameters) => AssistantMessageEventStream; + }, +): AssistantMessageEventStream; +export function streamInternalModel( + runtime: InternalStreamRuntime, + model: Model, + context: Context, + options: InternalStreamOptions | undefined, + config: StreamInternalModelConfig, +): AssistantMessageEventStream | Promise; +export function streamInternalModel( + runtime: InternalStreamRuntime, + model: Model, + context: Context, + options: InternalStreamOptions = {}, + config: StreamInternalModelConfig = {}, +): AssistantMessageEventStream | Promise { + let effectiveOptions = options; + if (model.provider === "openai-codex" && runtime.isUsingOAuth(model.provider)) { + const { apiKey, ...neutral } = effectiveOptions; + effectiveOptions = neutral; + } + const settings = config.settings; + // Without a session fallback policy, preserve the native stream and its + // original rejection/cancellation semantics after removing the OAuth snapshot. + if (!settings?.getRetryFallbackSettings().modelFallback) { + return config.streamFn + ? config.streamFn(model, context, effectiveOptions) + : runtime.streamSimple(model, context, effectiveOptions); + } + const agentDir = config.agentDir ?? getAgentDir(); + let cooldowns = config.cooldowns ?? cooldownsByRuntime.get(runtime); + if (!cooldowns) { + cooldowns = new SelectorCooldowns(Date.now); + cooldownsByRuntime.set(runtime, cooldowns); + } + const initialThinking: ThinkingLevel = options.reasoning ?? options.reasoningEffort ?? "off"; + let current: { model: Model; thinkingLevel: ThinkingLevel } = { model, thinkingLevel: initialThinking }; + const needsImages = context.messages.some( + (message) => Array.isArray(message.content) && message.content.some((part) => part.type === "image"), + ); + const controller = new RetryFallbackController({ + getSettings: () => settings.getRetryFallbackSettings(), + registry: { + find: (provider, id) => { + const candidate = runtime.getModel(provider, id); + return needsImages && !candidate?.input.includes("image") ? undefined : candidate; + }, + getAll: () => [...runtime.getModels()], + isUsingOAuth: (candidate) => runtime.isUsingOAuth(candidate.provider), + isFallbackEligible: (candidate) => runtime.isFallbackEligible(candidate.provider), + }, + cooldowns, + logger: createFallbackLogger(agentDir), + getCurrentSelector: () => current, + isAuthAvailable: (provider) => runtime.hasConfiguredAuth(provider), + switchModel: async (next, thinkingLevel) => { + current = { model: next, thinkingLevel }; + }, + emit: (event) => { + if (event.type !== "retry_fallback_applied") return; + config.notify?.({ + ...event, + type: "internal_model_fallback", + source: effectiveOptions.purpose ?? "internal", + }); + }, + }); + return lazyStream(model, async () => + (async function* () { + if (cooldowns.isSuppressed(`${model.provider}/${model.id}`)) { + if (!(await controller.tryFallback("hard-error", { errorMessage: "Model is in cooldown" }))) { + throw new Error("Internal model and its fallback chain are unavailable"); + } + } + while (true) { + effectiveOptions.signal?.throwIfAborted(); + const changed = current.model.provider !== model.provider || current.model.id !== model.id; + const { apiKey, headers, extraBody, env, reasoningEffort, reasoning, ...neutral } = effectiveOptions; + const requestOptions: InternalStreamOptions = changed + ? { + ...neutral, + ...(current.thinkingLevel === "off" ? {} : { reasoning: current.thinkingLevel }), + maxRetries: 0, + } + : effectiveOptions; + const stream = + changed || !config.streamFn + ? runtime.streamSimple(current.model, context, requestOptions) + : await config.streamFn(current.model, context, requestOptions); + let committed = false; + let failure: Extract | undefined; + for await (const event of stream) { + if (event.type === "error") { + failure = event; + break; + } + committed ||= event.type !== "start"; + yield event; + } + if (!failure) return; + const error = failure.error; + if ( + committed || + effectiveOptions.signal?.aborted || + error.errorMessage?.startsWith("senpi:no-turn-retry:") || + isContextOverflow(error, current.model.contextWindow) || + !(await controller.tryFallback("hard-error", { errorMessage: error.errorMessage })) + ) { + yield failure; + return; + } + } + })(), + ); +} diff --git a/packages/coding-agent/src/core/model-runtime.ts b/packages/coding-agent/src/core/model-runtime.ts index 85e977cc3..cd7501a61 100644 --- a/packages/coding-agent/src/core/model-runtime.ts +++ b/packages/coding-agent/src/core/model-runtime.ts @@ -17,6 +17,7 @@ import { type DeferredCancelOptions, type DeferredFetchOptions, type DeferredHandle, + extractOpenAiCodexAccountId, lazyStream, type Model, type Models, @@ -112,7 +113,11 @@ export interface ModelRuntimeAuthOverrides extends AuthOperationOptions { * id) keeps one conversation on one credential slot, and its absence simply * distributes requests instead of concentrating them. */ -export type CredentialRotationStreamOptions = StreamOptions & ModelsRequestTransforms & { affinityKey?: string }; +export type CredentialRotationStreamOptions = StreamOptions & + ModelsRequestTransforms & { + affinityKey?: string; + purpose?: string; + }; function mightHoldCredentialPool( providerId: string, @@ -168,12 +173,21 @@ function mergeHeaders( function withPayloadRequestMetadata(options: StreamOptions, model: Model): StreamOptions { if (!options.onPayload) return options; const onPayload = options.onPayload; + const headers = { ...options.headers }; + if (model.api === "openai-codex-responses" && options.apiKey) { + for (const name of Object.keys(headers)) { + if (["authorization", "chatgpt-account-id"].includes(name.toLowerCase())) delete headers[name]; + } + headers.authorization = `Bearer ${options.apiKey}`; + const accountId = extractOpenAiCodexAccountId(options.apiKey); + if (accountId) headers["chatgpt-account-id"] = accountId; + } return { ...options, onPayload: async (payload, providerModel) => await onPayload(payload, providerModel, { model, - headers: options.headers ?? {}, + headers, }), }; } @@ -230,9 +244,11 @@ export class ModelRuntime implements Models { * ModelRuntime, which the import-graph guard forbids. */ private readonly poolStatePath: string | undefined; + private readonly credentialSelectionState = new Map(); private credentialPoolModules: | Promise<{ rotation: typeof import("./credential-pool/rotation-stream.ts"); + quota: typeof import("./credential-pool/codex-quota.ts"); repository: InstanceType; }> | undefined; @@ -802,12 +818,14 @@ export class ModelRuntime implements Models { */ private loadCredentialPool(): NonNullable { this.credentialPoolModules ??= (async () => { - const [rotation, stateStore] = await Promise.all([ + const [rotation, stateStore, quota] = await Promise.all([ import("./credential-pool/rotation-stream.ts"), import("./credential-pool/state-store.ts"), + import("./credential-pool/codex-quota.ts"), ]); return { rotation, + quota, repository: new stateStore.CredentialSlotRepository(this.poolStatePath), }; })(); @@ -843,6 +861,23 @@ export class ModelRuntime implements Models { const pool = await this.loadCredentialPool(); const sources: RotationSources = { providerId: model.provider, + modelId: model.id, + signal: options?.signal, + sessionId: options?.affinitySessionId ?? options?.sessionId, + source: + options?.purpose ?? + (process.env.DREAM_TARGET_PATH + ? "memory dream" + : process.env.SENPI_MEMORY_FACTS === "1" + ? "memory facts" + : process.env.SENPI_MEMORY_REFLECTION === "1" + ? "memory reflection" + : undefined), + selectionState: this.credentialSelectionState, + getCodexUsage: async (slot) => { + const auth = await this.getAuth(model, { slotName: slot.name, signal: options?.signal }); + return pool.quota.fetchCodexUsage(auth?.auth?.apiKey, options?.signal); + }, credential, env, repository: pool.repository, @@ -852,6 +887,31 @@ export class ModelRuntime implements Models { return slots.length > 1 ? sources : undefined; } + /** Run an auxiliary HTTP request with the same credential policy as model streams. */ + async requestWithCredentialRotation( + model: Model, + options: CredentialRotationStreamOptions | undefined, + runAttempt: (prepared: { model: Model; options: ProviderRequestOptions }) => Promise, + ): Promise { + const sources = this.couldRotateCredentials(model, options) + ? await this.credentialRotationSources(model, options) + : undefined; + if (!sources) return runAttempt(await this.prepareRequest(model, options)); + const { rotation } = await this.loadCredentialPool(); + return rotation.requestWithCredentialRotation({ + sources, + affinityKey: options?.affinityKey ?? options?.affinitySessionId ?? options?.sessionId, + runAttempt: async (slot) => + runAttempt( + await this.prepareRequest( + model, + options, + slot.lane === "env" ? { apiKey: slot.envKey } : { slotName: slot.name }, + ), + ), + }); + } + stream( model: Model, context: Context, @@ -868,8 +928,8 @@ export class ModelRuntime implements Models { sources, ...(streamOptions?.affinityKey !== undefined ? { affinityKey: streamOptions.affinityKey } - : streamOptions?.sessionId !== undefined - ? { affinityKey: streamOptions.sessionId } + : (streamOptions?.affinitySessionId ?? streamOptions?.sessionId) !== undefined + ? { affinityKey: streamOptions?.affinitySessionId ?? streamOptions?.sessionId } : {}), runAttempt: async (slot) => { const prepared = await this.prepareRequest( @@ -923,7 +983,9 @@ export class ModelRuntime implements Models { const { rotation } = await this.loadCredentialPool(); return rotation.streamWithCredentialRotation({ sources, - ...(streamOptions?.sessionId === undefined ? {} : { affinityKey: streamOptions.sessionId }), + ...((streamOptions?.affinitySessionId ?? streamOptions?.sessionId) === undefined + ? {} + : { affinityKey: streamOptions?.affinitySessionId ?? streamOptions?.sessionId }), runAttempt: async (slot) => { const prepared = await this.prepareRequest( model, diff --git a/packages/coding-agent/src/modes/interactive/changes.md b/packages/coding-agent/src/modes/interactive/changes.md index d872392c8..32513dd7e 100644 --- a/packages/coding-agent/src/modes/interactive/changes.md +++ b/packages/coding-agent/src/modes/interactive/changes.md @@ -73,6 +73,28 @@ The login command is interactive mode's own command handler; an extension cannot - `packages/coding-agent/src/modes/interactive/interactive-mode.ts`: `commandContextActions` navigation and assistant-edit neighbours. - `packages/coding-agent/src/modes/interactive/interactive-host-runtime.ts`: edit-related imports and proxy navigation/edit property cases. +## 2026-09-20 - Account-switch and internal-model fallback notices render in the TUI + +### What changed + +- `packages/coding-agent/src/modes/interactive/interactive-mode.ts`: the session event switch + gains `account_failover` (rendered via `formatAccountSwitchNotice`) and + `internal_model_fallback` (a notice stating the internal request switched models while the + active session model is unchanged). Neither branch mutates main model or footer state. + +### Why + +- Both events report account/model routing changes the user should see without the host + treating an internal fallback as a session-model switch. + +### Why an extension could not handle it + +- They are engine-emitted session events rendered by the host's own event dispatcher, before + any extension widget can intercept them. + +### Expected merge conflict zones + +- LOW: the two new case arms after `server_fallback_aborted` in the session event switch. ## 2026-09-20 - Surface a held model switch (senpi#1873) diff --git a/packages/coding-agent/src/modes/interactive/interactive-mode.ts b/packages/coding-agent/src/modes/interactive/interactive-mode.ts index 18962385a..c59eedc1c 100644 --- a/packages/coding-agent/src/modes/interactive/interactive-mode.ts +++ b/packages/coding-agent/src/modes/interactive/interactive-mode.ts @@ -86,6 +86,7 @@ import { } from "../../core/cache-stats.ts"; import { resolveChangelogSource } from "../../core/changelog-source.ts"; import { collectEntriesForBranchSummary } from "../../core/compaction/branch-summarization.ts"; +import { formatAccountSwitchNotice } from "../../core/credential-pool/account-notices.ts"; import { AssistantEditError, assistantTextEquals } from "../../core/edited-assistant-message.ts"; import { formatUserMessage } from "../../core/extensions/builtin/ask-user/format.ts"; import { askUserRenderers } from "../../core/extensions/builtin/ask-user/render.ts"; @@ -5607,6 +5608,24 @@ export class InteractiveMode { }); break; + case "account_failover": { + const notice = formatAccountSwitchNotice(event); + this.showNoticeBox({ + title: notice.title, + tone: "warning", + why: notice.why, + }); + break; + } + + case "internal_model_fallback": + this.showNoticeBox({ + title: `⇆ Model fallback · ${event.source} · ${event.from} → ${event.to}`, + tone: "warning", + why: `Internal ${event.source} request switched models (${event.reason}); the active session model is unchanged.`, + }); + break; + case "auto_retry_start": { if (isNetworkProviderError(event.errorMessage)) { this.getProviderErrors().retrying(event.errorMessage, this.toolOutputExpanded); diff --git a/packages/coding-agent/src/modes/print-mode.ts b/packages/coding-agent/src/modes/print-mode.ts index b1e1a604f..eb479712f 100644 --- a/packages/coding-agent/src/modes/print-mode.ts +++ b/packages/coding-agent/src/modes/print-mode.ts @@ -12,6 +12,7 @@ import { stripTurnRetrySuppressionPrefix, } from "@earendil-works/pi-ai"; import type { AgentSessionRuntime } from "../core/agent-session-runtime.ts"; +import { formatAccountSwitchNotice } from "../core/credential-pool/account-notices.ts"; import { flushRawStdout, waitForRawStdoutBackpressure, writeRawStdout } from "../core/output-guard.ts"; import { killTrackedDetachedChildren } from "../utils/shell.ts"; import { toJsonEvent } from "./json-event.ts"; @@ -134,6 +135,11 @@ export async function runPrintMode(runtimeHost: AgentSessionRuntime, options: Pr console.error(`Model fallback exhausted: ${event.chainKey} (${event.lastError})`); } else if (event.type === "retry_fallback_reverted") { console.error(`Model fallback reverted: ${event.from} -> ${event.to}`); + } else if (event.type === "account_failover") { + const notice = formatAccountSwitchNotice(event); + console.error(notice.why ? `${notice.title} — ${notice.why}` : notice.title); + } else if (event.type === "internal_model_fallback") { + console.error(`Model fallback (${event.source}): ${event.from} -> ${event.to} (${event.reason})`); } if (mode === "json") { writeRawStdout(`${JSON.stringify(toJsonEvent(event))}\n`); diff --git a/packages/coding-agent/test/compaction/before-compact-error-surfacing.test.ts b/packages/coding-agent/test/compaction/before-compact-error-surfacing.test.ts index de3c7d8fb..6e4ffa00c 100644 --- a/packages/coding-agent/test/compaction/before-compact-error-surfacing.test.ts +++ b/packages/coding-agent/test/compaction/before-compact-error-surfacing.test.ts @@ -90,6 +90,7 @@ function createHarness(options?: { withAuth?: boolean }): Harness { cwd: process.cwd(), isProjectTrusted: () => true, sessionManager: { + getSessionId: () => "before-compact-test", getEntries: () => [], getBranch: () => [], } as unknown as ExtensionContext["sessionManager"], diff --git a/packages/coding-agent/test/suite/account-notices.test.ts b/packages/coding-agent/test/suite/account-notices.test.ts new file mode 100644 index 000000000..62b7fa58a --- /dev/null +++ b/packages/coding-agent/test/suite/account-notices.test.ts @@ -0,0 +1,146 @@ +import { rmSync } from "node:fs"; +import { afterEach, describe, expect, it, vi } from "vitest"; +import { + type AccountSwitchNotice, + emitAccountSwitch, + formatAccountSwitchNotice, + subscribeAccountSwitch, +} from "../../src/core/credential-pool/account-notices.ts"; +import { createHarness } from "./harness.ts"; + +const event: AccountSwitchNotice = { + type: "account_failover", + provider: "openai-codex", + from: "slot-a", + to: "slot-b", + reason: "quota exhausted", + sessionId: "affinity-session", + source: "rotation", +}; + +describe("account switch notices", () => { + afterEach(() => vi.restoreAllMocks()); + it("subscribes, emits, and unsubscribes", () => { + const seen: AccountSwitchNotice[] = []; + const unsubscribe = subscribeAccountSwitch((notice) => { + seen.push(notice); + }); + emitAccountSwitch(event); + unsubscribe(); + emitAccountSwitch(event); + expect(seen).toEqual([event]); + }); + + it("isolates a synchronous throwing listener without dropping later observers", () => { + const diagnostic = vi.spyOn(console, "error").mockImplementation(() => {}); + const seen: string[] = []; + const unsubscribeThrowing = subscribeAccountSwitch(() => { + throw new Error("listener boom"); + }); + const unsubscribeLater = subscribeAccountSwitch((notice) => { + seen.push(notice.to); + }); + expect(() => emitAccountSwitch(event)).not.toThrow(); + expect(seen).toEqual(["slot-b"]); + expect(diagnostic).toHaveBeenCalledTimes(1); + expect(JSON.stringify(diagnostic.mock.calls)).not.toContain("listener boom"); + unsubscribeThrowing(); + unsubscribeLater(); + }); + + it("isolates a rejected async observer without an unhandled rejection", async () => { + const reported = Promise.withResolvers(); + const diagnostic = vi.spyOn(console, "error").mockImplementation(() => reported.resolve()); + const seen: string[] = []; + const unsubscribeRejected = subscribeAccountSwitch(async () => { + throw new Error("async listener boom"); + }); + const unsubscribeLater = subscribeAccountSwitch((notice) => { + seen.push(notice.to); + }); + try { + emitAccountSwitch(event); + expect(seen).toEqual(["slot-b"]); + await reported.promise; + expect(diagnostic).toHaveBeenCalledTimes(1); + expect(JSON.stringify(diagnostic.mock.calls)).not.toContain("async listener boom"); + } finally { + unsubscribeRejected(); + unsubscribeLater(); + } + }, 1000); + + it("renders routing labels without serializing extra credential fields", () => { + const input = { ...event, token: "private-credential-marker" }; + const formatted = formatAccountSwitchNotice(input); + for (const label of [event.provider, event.from, event.to]) { + expect(formatted.title).toContain(label); + } + expect(formatted.why).toContain(event.reason); + expect(formatted.why).toContain(event.source); + expect(JSON.stringify(formatted)).not.toContain(input.token); + expect(formatAccountSwitchNotice({ ...event, source: undefined }).why).not.toContain("undefined"); + }); +}); + +describe("agent session account-switch isolation", () => { + it.each(["sync", "async", "self-unsubscribe"])( + "isolates a %s session observer and still delivers to its peers", + async (mode) => { + const harness = await createHarness(); + const reported = Promise.withResolvers(); + const diagnostic = vi.spyOn(console, "error").mockImplementation(() => reported.resolve()); + const seen: string[] = []; + const unsubscribe = harness.session.subscribe((notice) => { + if (notice.type !== "account_failover") return; + if (mode === "self-unsubscribe") unsubscribe(); + if (mode !== "async") throw new Error("private session observer error"); + return Promise.reject(new Error("private session observer error")); + }); + harness.session.subscribe((notice) => { + if (notice.type === "account_failover") seen.push(notice.to); + }); + try { + emitAccountSwitch({ ...event, sessionId: harness.session.sessionId }); + expect(seen).toEqual(["slot-b"]); + await reported.promise; + expect(diagnostic).toHaveBeenCalledTimes(1); + expect(JSON.stringify(diagnostic.mock.calls)).not.toContain("private session observer error"); + } finally { + diagnostic.mockRestore(); + harness.cleanup(); + } + }, + ); + + it("surfaces only its own session and stops after disposal", async () => { + const harness = await createHarness(); + const matching = harness.session.sessionId; + try { + emitAccountSwitch({ ...event, sessionId: matching }); + expect(harness.eventsOfType("account_failover")).toEqual([ + { + type: "account_failover", + provider: event.provider, + from: event.from, + to: event.to, + reason: event.reason, + sessionId: matching, + source: event.source, + }, + ]); + + emitAccountSwitch({ ...event, sessionId: "other-session" }); + expect(harness.eventsOfType("account_failover")).toHaveLength(1); + + harness.session.dispose(); + emitAccountSwitch({ ...event, sessionId: matching }); + expect(harness.eventsOfType("account_failover")).toHaveLength(1); + } finally { + // dispose() is intentionally run once inside the test; clean up the + // harness's non-session resources here instead of calling cleanup(). + harness.faux.unregister(); + rmSync(harness.tempDir, { recursive: true, force: true }); + } + }); +}); diff --git a/packages/coding-agent/test/suite/codex-auxiliary-routing.test.mjs b/packages/coding-agent/test/suite/codex-auxiliary-routing.test.mjs new file mode 100644 index 000000000..6b5ed48a7 --- /dev/null +++ b/packages/coding-agent/test/suite/codex-auxiliary-routing.test.mjs @@ -0,0 +1,382 @@ +import assert from "node:assert/strict"; +import { test } from "vitest"; +import { mkdtempSync, rmSync, writeFileSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { AuthStorage } from "../../src/core/auth-storage.ts"; +import { ModelRuntime } from "../../src/core/model-runtime.ts"; +import { streamInternalModel } from "../../src/core/internal-model-request.ts"; +import { lazyStream } from "@earendil-works/pi-ai"; +import { generateSessionTitle } from "../../src/core/session-title-generator.ts"; +import { completeSummarization } from "../../src/core/compaction/compaction.ts"; +import { generateSummaryMessage } from "../../src/core/extensions/builtin/compaction/speculative-summary.ts"; +import { runOpenAiRemoteCompaction } from "../../src/core/extensions/builtin/compaction/openai-remote.ts"; +import { createAgentSession, DefaultResourceLoader, SessionManager, SettingsManager } from "../../src/index.ts"; +import { subscribeAccountSwitch } from "../../src/core/credential-pool/account-notices.ts"; +import btwExtension from "../../src/core/extensions/builtin/btw/index.ts"; +import { runLookAt } from "../../src/core/extensions/builtin/look-at/runner.ts"; +import { generateBranchSummary } from "../../src/core/compaction/branch-summarization.ts"; + +const fallbackSettings = { + retry: { enabled: true, maxRetries: 0, modelFallback: true, fallbackRevertPolicy: "never", + fallbackChains: { "openai-codex/gpt-6-astra": ["openrouter/deepseek/deepseek-v4-pro-0813:max"] } }, +}; + +const token = (name) => `test.${Buffer.from(JSON.stringify({ + "https://api.openai.com/auth": { chatgpt_account_id: name }, +})).toString("base64url")}.test`; +const usage = (used) => ({ + rate_limit: { allowed: used < 100, limit_reached: used >= 100, primary_window: { used_percent: used } }, + credits: { has_credits: false, balance: "0" }, +}); +const context = { + systemPrompt: "Return a short summary.", + messages: [{ role: "user", content: "Repair the account selector.", timestamp: 1 }], +}; + +async function fixture(t, allExhausted = false) { + const dir = mkdtempSync(join(tmpdir(), "omo-auxiliary-")); + t.onTestFinished(() => rmSync(dir, { recursive: true, force: true })); + writeFileSync(join(dir, "settings.json"), JSON.stringify(fallbackSettings)); + const accounts = ["exhausted", "ready"].map((name) => ({ + name, access: token(name), refresh: `test-refresh-${name}`, expires: 9e12, + })); + const credentials = AuthStorage.inMemory({ + "openai-codex": { type: "oauth", ...accounts[0], pinned: "exhausted", accounts }, + openrouter: { type: "api_key", key: "test-openrouter-key" }, + }); + const runtime = await ModelRuntime.create({ credentials, agentDir: dir, modelsPath: null, allowModelNetwork: false }); + const attempts = []; + const attemptedModels = []; + const provider = runtime.getProvider("openai-codex"); + const produce = (model, _context, options) => { + attempts.push(options.apiKey); + attemptedModels.push(`${model.provider}/${model.id}`); + const message = { + role: "assistant", provider: model.provider, api: model.api, model: model.id, + content: [{ type: "text", text: "Account Routing Repaired" }], + stopReason: "stop", timestamp: 1, + usage: { input: 1, output: 1, cacheRead: 0, cacheWrite: 0, totalTokens: 2, + cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 } }, + }; + return lazyStream(model, async () => { + await options.onPayload?.({}, model); + return (async function* () { + yield { type: "text_delta", contentIndex: 0, delta: message.content[0].text, partial: message }; + yield { type: "done", reason: "stop", message }; + })(); + }); + }; + await runtime.registerNativeProvider({ ...provider, stream: produce, streamSimple: produce }, { refresh: false }); + await runtime.registerNativeProvider({ + ...runtime.getProvider("openrouter"), stream: produce, streamSimple: produce, + }, { refresh: false }); + const resolveSources = runtime.credentialRotationSources.bind(runtime); + runtime.credentialRotationSources = async (model, options) => { + const sources = await resolveSources(model, options); + if (sources) sources.getCodexUsage = async (slot) => usage(allExhausted || slot.name === "exhausted" ? 100 : 0); + return sources; + }; + return { runtime, model: runtime.getModel("openai-codex", "gpt-6-astra"), attempts, attemptedModels, dir, credentials }; +} + +async function makeSession(t, f) { + const settingsManager = SettingsManager.inMemory(fallbackSettings); + const resourceLoader = new DefaultResourceLoader({ + cwd: f.dir, agentDir: f.dir, settingsManager, + noExtensions: true, noSkills: true, noPromptTemplates: true, noThemes: true, noContextFiles: true, + systemPrompt: "Reply briefly.", + }); + await resourceLoader.reload(); + const { session } = await createAgentSession({ + cwd: f.dir, agentDir: f.dir, modelRuntime: f.runtime, authStorage: f.credentials, + settingsManager, resourceLoader, model: f.model, + sessionManager: SessionManager.inMemory(), noTools: "all", + }); + t.onTestFinished(() => session.dispose()); + return session; +} + +test("title generation rotates away from its pre-resolved exhausted OAuth token", async (t) => { + const f = await fixture(t); + const title = await generateSessionTitle({ + model: f.model, firstPrompt: "Repair the credential account selector", + auth: { apiKey: token("exhausted") }, sessionId: "title-test", + streamFn: (model, context, options) => streamInternalModel(f.runtime, model, context, options, { settings: SettingsManager.inMemory(fallbackSettings), agentDir: f.dir }), + retry: { enabled: true, maxRetries: 0, baseDelayMs: 0 }, + }); + assert.ok(title); + assert.deepEqual(f.attempts, [token("ready")]); +}); + +test("compaction rotates even when summary options contain a resolved OAuth token", async (t) => { + const f = await fixture(t); + await completeSummarization(f.model, context, + { apiKey: token("exhausted"), affinitySessionId: "compaction-test" }, + (model, context, options) => streamInternalModel(f.runtime, model, context, options, { settings: SettingsManager.inMemory(fallbackSettings), agentDir: f.dir }), + { enabled: true, maxRetries: 0, baseDelayMs: 0 }); + assert.deepEqual(f.attempts, [token("ready")]); +}); + +test("speculative compaction uses account admission on the typed stream path", async (t) => { + const f = await fixture(t); + const result = await generateSummaryMessage({ + context: { agentDir: f.dir, cwd: f.dir, getRetryFallbackSettings: () => SettingsManager.inMemory(fallbackSettings).getRetryFallbackSettings(), modelRegistry: { modelRuntime: f.runtime } }, + snapshot: { model: f.model, contextWindow: f.model.contextWindow, systemPrompt: "Summarize." }, + auth: { apiKey: token("exhausted") }, + messages: context.messages, + prompt: { system: "Summarize.", user: "Summarize the request." }, + }); + assert.ok(result); + assert.deepEqual(f.attempts, [token("ready")]); +}); + +test("an unrelated explicit credential still bypasses the stored account pool", async (t) => { + const f = await fixture(t); + await f.runtime.completeSimple(f.model, context, { apiKey: token("external") }); + assert.deepEqual(f.attempts, [token("external")]); +}); + +test("an explicit stored credential remains pinned instead of silently rotating", async (t) => { + const f = await fixture(t); + await f.runtime.completeSimple(f.model, context, { apiKey: token("exhausted") }); + assert.deepEqual(f.attempts, [token("exhausted")]); +}); + +test("native title generation uses the configured model fallback after account exhaustion", async (t) => { + const f = await fixture(t, true); + const session = await makeSession(t, f); + await session._generateSessionTitle("Repair credential account selection", f.model, new AbortController()); + assert.ok(session.sessionManager.getSessionName()); + assert.deepEqual(f.attemptedModels, ["openrouter/deepseek/deepseek-v4-pro-0813"]); + assert.equal(session.model.provider, "openai-codex"); +}); + +test("speculative summary uses the configured model fallback without Codex generation", async (t) => { + const f = await fixture(t, true); + const result = await generateSummaryMessage({ + context: { agentDir: f.dir, cwd: f.dir, getRetryFallbackSettings: () => SettingsManager.inMemory(fallbackSettings).getRetryFallbackSettings(), modelRegistry: { modelRuntime: f.runtime } }, + snapshot: { model: f.model, contextWindow: f.model.contextWindow, systemPrompt: "Summarize." }, + auth: { apiKey: token("exhausted") }, messages: context.messages, + prompt: { system: "Summarize.", user: "Summarize the request." }, + }); + assert.equal(result.stopReason, "stop"); + assert.deepEqual(f.attemptedModels, ["openrouter/deepseek/deepseek-v4-pro-0813"]); +}); + +test("remote Codex compaction uses a quota-eligible account on the HTTP endpoint", async (t) => { + const f = await fixture(t); + const seenAccounts = []; + const result = await runOpenAiRemoteCompaction({ + model: f.model, + modelRegistry: { + modelRuntime: f.runtime, + getApiKeyAndHeaders: async () => ({ ok: true, apiKey: token("exhausted") }), + }, + getSystemPrompt: () => "Summarize.", + sessionManager: { getSessionId: () => "remote-test" }, + }, { + reason: "manual", requestId: "remote-request", signal: new AbortController().signal, + branchEntries: [{ type: "message", id: "u", message: context.messages[0] }], + preparation: { tokensBefore: 100, firstKeptEntryId: "u" }, + }, undefined, { + fetch: async (_url, options) => { + seenAccounts.push(new Headers(options.headers).get("chatgpt-account-id")); + return Response.json({ output: [{ type: "compaction", encrypted_content: "test-checkpoint" }] }); + }, + }); + assert.ok(result); + assert.deepEqual(seenAccounts, ["ready"]); +}); + +async function closeAfterTerminal(stream) { + for await (const event of stream) { + if (event.type === "error") throw new Error(event.error.errorMessage); + if (event.type === "done") break; + } +} + +test("payload provenance uses the selected account rather than the stale auth snapshot", async (t) => { + const f = await fixture(t); + let headers; + await closeAfterTerminal(streamInternalModel(f.runtime, f.model, context, { + apiKey: token("exhausted"), + onPayload: (_payload, _model, metadata) => { headers = new Headers(metadata.headers); }, + }, { settings: SettingsManager.inMemory(fallbackSettings), agentDir: f.dir })); + assert.equal(headers.get("chatgpt-account-id"), "ready"); + assert.equal(headers.get("authorization"), `Bearer ${token("ready")}`); +}); + +test("remote compaction returns control to local fallback when every account is exhausted", async (t) => { + const f = await fixture(t, true); + const events = []; + let requests = 0; + const result = await runOpenAiRemoteCompaction({ + model: f.model, + modelRegistry: { + modelRuntime: f.runtime, + getApiKeyAndHeaders: async () => ({ ok: true, apiKey: token("exhausted") }), + }, + getSystemPrompt: () => "Summarize.", + sessionManager: { getSessionId: () => "remote-exhausted-test" }, + }, { + reason: "manual", requestId: "remote-exhausted", signal: new AbortController().signal, + branchEntries: [{ type: "message", id: "u", message: context.messages[0] }], + preparation: { tokensBefore: 100, firstKeptEntryId: "u" }, + }, (event) => events.push(event), { + fetch: async () => { requests += 1; throw new Error("Unexpected compact endpoint request"); }, + }); + assert.equal(result, undefined); + assert.equal(requests, 0); + assert.ok(events.some((event) => event.action === "remote_fallback")); +}); + +function remoteRequest(f, model, runtime, fetchImpl) { + return runOpenAiRemoteCompaction({ + model, + modelRegistry: { + modelRuntime: runtime, + getApiKeyAndHeaders: async () => ({ ok: true, apiKey: token("exhausted") }), + }, + getSystemPrompt: () => "Summarize.", + sessionManager: { getSessionId: () => "remote-errors-test" }, + }, { + reason: "manual", requestId: "remote-errors", signal: new AbortController().signal, + branchEntries: [{ type: "message", id: "u", message: context.messages[0] }], + preparation: { tokensBefore: 100, firstKeptEntryId: "u" }, + }, undefined, { fetch: fetchImpl }); +} + +for (const provider of ["openai-codex", "openai"]) { + for (const status of [401, 402, 403, 429]) { + test(`legacy ${provider} remote HTTP ${status} preserves local fallback`, async (t) => { + const f = await fixture(t); + const model = provider === "openai-codex" ? f.model : { + ...f.model, provider, api: "openai-responses", baseUrl: "https://example.invalid/v1", + compat: { supportsWebSocket: false, supportsRemoteCompactionV2: false }, + }; + let requests = 0; + const result = await remoteRequest(f, model, { + streamSimple() { throw new Error("Unexpected streaming call"); }, + }, async () => { + requests += 1; + return new Response("synthetic account rejection", { status }); + }); + assert.equal(result, undefined); + assert.equal(requests, 1); + }); + } +} + +for (const status of [401, 429]) { + test(`rotating remote HTTP ${status} still tries the healthy peer`, async (t) => { + const f = await fixture(t); + const resolveSources = f.runtime.credentialRotationSources.bind(f.runtime); + f.runtime.credentialRotationSources = async (model, options) => { + const sources = await resolveSources(model, options); + if (sources) sources.getCodexUsage = async () => usage(20); + return sources; + }; + const attempts = []; + const result = await remoteRequest(f, f.model, f.runtime, async (_url, options) => { + const account = new Headers(options.headers).get("chatgpt-account-id"); + attempts.push(account); + return account === "exhausted" + ? new Response("synthetic account rejection", { status }) + : Response.json({ output: [{ type: "compaction", encrypted_content: "test-checkpoint" }] }); + }); + assert.ok(result); + assert.deepEqual(attempts, ["exhausted", "ready"]); + }); +} + +test("account fallback emits one sanitized notice, not another notice each request", async (t) => { + const f = await fixture(t); + const events = []; + const unsubscribe = subscribeAccountSwitch((event) => events.push(event)); + t.onTestFinished(unsubscribe); + const options = { sessionId: "account-notice-test", purpose: "title", apiKey: token("exhausted") }; + await closeAfterTerminal(streamInternalModel(f.runtime, f.model, context, options, { settings: SettingsManager.inMemory(fallbackSettings), agentDir: f.dir })); + await closeAfterTerminal(streamInternalModel(f.runtime, f.model, context, options, { settings: SettingsManager.inMemory(fallbackSettings), agentDir: f.dir })); + assert.equal(events.length, 1); + assert.equal(events[0].from, "exhausted"); + assert.equal(events[0].to, "ready"); + assert.equal(events[0].source, "title"); + assert.equal(events[0].sessionId, "account-notice-test"); + assert.match(events[0].reason, /quota/); + assert.equal(JSON.stringify(events).includes(token("exhausted")), false); + assert.equal(JSON.stringify(events).includes(token("ready")), false); +}); + +test("the native side-query command falls back without changing the main model", async (t) => { + const f = await fixture(t, true); + let command; + const notices = []; + btwExtension({ + on() {}, + getThinkingLevel: () => "low", + registerCommand: (_name, definition) => { command = definition; }, + }); + const ctx = { + getRetryFallbackSettings: () => SettingsManager.inMemory(fallbackSettings).getRetryFallbackSettings(), + model: f.model, agentDir: f.dir, cwd: f.dir, mode: "print", hasUI: false, + getSystemPrompt: () => "Reply briefly.", + sessionManager: { getEntries: () => [], getLeafId: () => null, getSessionId: () => "side-query-test" }, + modelRegistry: { + modelRuntime: f.runtime, + getApiKeyAndHeaders: async () => ({ ok: true, apiKey: token("exhausted") }), + }, + ui: { notify: (message, kind) => notices.push({ message, kind }) }, + }; + await command.handler("Which account can serve this request?", ctx); + assert.deepEqual(f.attemptedModels, ["openrouter/deepseek/deepseek-v4-pro-0813"]); + assert.equal(ctx.model.provider, "openai-codex"); + assert.equal(notices.some((notice) => notice.kind === "error"), false); + assert.ok(notices.some((notice) => notice.kind === "info" && notice.message.length > 0)); +}); + +const pixel = "data:image/png;base64,iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAQAAAC1HAwCAAAAC0lEQVR42mP8/x8AAwMCAO+/l9sAAAAASUVORK5CYII="; +function visionContext(f) { + return { + agentDir: f.dir, cwd: f.dir, + getRetryFallbackSettings: () => SettingsManager.inMemory(fallbackSettings).getRetryFallbackSettings(), + getImageSettings: () => ({ blockImages: false, autoResize: false }), + getLookAtSettings: () => ({ models: ["openai-codex/gpt-6-astra:low"] }), + sessionManager: { getSessionId: () => "vision-test" }, + modelRegistry: { + modelRuntime: f.runtime, getAvailable: () => [f.model], + getApiKeyAndHeaders: async () => ({ ok: true, apiKey: token("exhausted") }), + }, + ui: { notify() {} }, + }; +} + +test("native vision analysis skips an exhausted account", async (t) => { + const f = await fixture(t); + const result = await runLookAt({ image_data: pixel, goal: "Describe the image" }, + undefined, visionContext(f), { getOverride: () => ({}) }); + assert.ok(result.text); + assert.deepEqual(f.attempts, [token("ready")]); +}); + +test("vision fallback never sends image input to a text-only fallback model", async (t) => { + const f = await fixture(t, true); + await assert.rejects(runLookAt({ image_data: pixel, goal: "Describe the image" }, + undefined, visionContext(f), { getOverride: () => ({}) }), /No credential slots available/); + assert.deepEqual(f.attemptedModels, []); +}); + +test("native branch summaries use the internal model fallback after account exhaustion", async (t) => { + const f = await fixture(t, true); + const session = await makeSession(t, f); + const result = await generateBranchSummary([ + { type: "message", id: "u", parentId: null, message: context.messages[0] }, + ], { + model: f.model, apiKey: token("exhausted"), reserveTokens: 512, + streamFn: (model, request, options) => session._streamInternalModel(model, request, options, "branch summary"), + retry: { enabled: true, maxRetries: 0, baseDelayMs: 0 }, + }); + assert.ok(result.summary); + assert.equal(result.error, undefined); + assert.deepEqual(f.attemptedModels, ["openrouter/deepseek/deepseek-v4-pro-0813"]); +}); diff --git a/packages/coding-agent/test/suite/codex-quota.test.ts b/packages/coding-agent/test/suite/codex-quota.test.ts new file mode 100644 index 000000000..2761bd2aa --- /dev/null +++ b/packages/coding-agent/test/suite/codex-quota.test.ts @@ -0,0 +1,125 @@ +import { describe, expect, it } from "vitest"; +import { classifyCredentialFailure } from "../../src/core/credential-pool/classify.ts"; +import { admitCodexQuota } from "../../src/core/credential-pool/codex-quota.ts"; +import type { RotationSlot } from "../../src/core/credential-pool/rotation-stream.ts"; + +const now = 1_000_000; +const quota = (used: number, credits = false, allowed = used < 100) => ({ + rate_limit: { allowed, limit_reached: used >= 100, primary_window: { used_percent: used } }, + credits: { has_credits: credits, balance: credits ? "100" : "0" }, +}); +const available = (slots: RotationSlot[]) => + slots + .filter( + (slot) => + slot.blockReason !== "auth_error" && + slot.blockReason !== "account_disabled" && + !(slot.blockedUntil !== undefined && slot.blockedUntil > now), + ) + .map((slot) => slot.name); + +describe("Codex paid quota admission", () => { + it.each(["You have hit your ChatGPT usage limit.", "You have hit your usage limit."])( + "classifies provider quota-limit prose without an HTTP status: %s", + (message) => { + expect(classifyCredentialFailure(new Error(message))).toMatchObject({ + kind: "failover", + block: { reason: "rate_limit" }, + }); + }, + ); + it.each([20, 100])("reads cooling accounts without unblocking generation (%i percent)", async (used) => { + const lookedUp: string[] = []; + const slots = await admitCodexQuota( + { + providerId: "openai-codex", + getCodexUsage: async (slot) => { + lookedUp.push(slot.name); + return quota(slot.name === "cooling" ? used : 100, slot.name === "paid"); + }, + }, + [ + { name: "cooling", lane: "stored", blockReason: "rate_limit", blockedUntil: now + 60_000 }, + { name: "paid", lane: "stored" }, + ], + ); + expect(lookedUp).toContain("cooling"); + expect(available(slots)).toEqual(used === 100 ? ["paid"] : []); + expect(slots.find((slot) => slot.name === "cooling")?.blockedUntil).toBe(now + 60_000); + }); + + it.each(["auth_error", "account_disabled"] as const)("treats %s quota as unknown", async (blockReason) => { + const slots = await admitCodexQuota( + { + providerId: "openai-codex", + getCodexUsage: async () => quota(100, true), + }, + [ + { name: "unknown", lane: "stored", blockReason }, + { name: "paid", lane: "stored" }, + ], + ); + expect(available(slots)).toEqual([]); + }); + + it("does not confuse denial with quota exhaustion", async () => { + const slots = await admitCodexQuota( + { + providerId: "openai-codex", + getCodexUsage: async () => quota(20, true, false), + }, + [{ name: "denied", lane: "stored" }], + ); + expect(available(slots)).toEqual([]); + }); + + it.each(["error", "malformed"])("preserves healthy included quota when another account is %s", async (mode) => { + const slots = await admitCodexQuota( + { + providerId: "openai-codex", + getCodexUsage: async (slot) => { + if (slot.name === "unknown") { + if (mode === "error") throw new Error("test transport failure"); + return {}; + } + return quota(slot.name === "included" ? 20 : 100, slot.name === "paid"); + }, + }, + ["unknown", "included", "paid"].map((name) => ({ name, lane: "stored" })), + ); + expect(available(slots)).toEqual(["included"]); + }); + + it("requires exhaustion of matching model-specific quota as well", async () => { + const slots = await admitCodexQuota( + { + providerId: "openai-codex", + modelId: "test-model", + getCodexUsage: async () => ({ + ...quota(100, true), + additional_rate_limits: [{ normal_model_slug: "test-model", rate_limit: quota(10).rate_limit }], + }), + }, + [{ name: "included", lane: "stored" }], + ); + expect(slots[0]?.quotaTier).toBe(0); + }); + + it("propagates cancellation instead of making an admission decision", async () => { + const controller = new AbortController(); + const reason = new Error("cancelled test"); + controller.abort(reason); + await expect( + admitCodexQuota( + { + providerId: "openai-codex", + signal: controller.signal, + getCodexUsage: async () => { + throw reason; + }, + }, + [{ name: "cancelled", lane: "stored" }], + ), + ).rejects.toBe(reason); + }); +}); diff --git a/packages/coding-agent/test/suite/codex-runtime-routing.test.ts b/packages/coding-agent/test/suite/codex-runtime-routing.test.ts new file mode 100644 index 000000000..581cbdb0b --- /dev/null +++ b/packages/coding-agent/test/suite/codex-runtime-routing.test.ts @@ -0,0 +1,158 @@ +import { mkdtempSync, rmSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { fauxAssistantMessage, fauxProvider } from "@earendil-works/pi-ai"; +import { afterEach, expect, it, vi } from "vitest"; +import { AuthStorage } from "../../src/core/auth-storage.ts"; +import { subscribeAccountSwitch } from "../../src/core/credential-pool/account-notices.ts"; +import { CredentialSlotRepository } from "../../src/core/credential-pool/state-store.ts"; +import { ModelRuntime } from "../../src/core/model-runtime.ts"; + +afterEach(() => vi.restoreAllMocks()); +afterEach(() => vi.unstubAllGlobals()); + +const token = (name: string) => + `test.${Buffer.from( + JSON.stringify({ + "https://api.openai.com/auth": { chatgpt_account_id: name }, + }), + ).toString("base64url")}.test`; + +async function fixture() { + const dir = mkdtempSync(join(tmpdir(), "codex-runtime-")); + const accounts = ["exhausted", "ready"].map((name) => ({ + name, + access: token(name), + refresh: `test-refresh-${name}`, + expires: 9e12, + })); + const credentials = AuthStorage.inMemory({ + "openai-codex": { + type: "oauth", + access: token("exhausted"), + refresh: "test-refresh-exhausted", + expires: 9e12, + pinned: "exhausted", + accounts, + }, + }); + const runtime = await ModelRuntime.create({ + credentials, + agentDir: dir, + modelsPath: null, + allowModelNetwork: false, + }); + const provider = runtime.getProvider("openai-codex"); + const model = runtime.getModels().find((candidate) => candidate.provider === "openai-codex"); + if (!provider || !model) throw new Error("Codex builtin is required"); + const fetchUsage = vi.fn(async (_url: unknown, init?: RequestInit) => { + const exhausted = new Headers(init?.headers).get("authorization") === `Bearer ${token("exhausted")}`; + return Response.json({ + rate_limit: { + allowed: !exhausted, + limit_reached: exhausted, + primary_window: { used_percent: exhausted ? 100 : 0 }, + }, + credits: { has_credits: false }, + }); + }); + vi.stubGlobal("fetch", fetchUsage); + return { runtime, provider, model, fetchUsage, dir, cleanup: () => rmSync(dir, { recursive: true, force: true }) }; +} + +it.each(["exhausted", "external"])("explicit %s credentials bypass Codex rotation", async (name) => { + const f = await fixture(); + const faux = fauxProvider({ provider: "openai-codex" }); + try { + await f.runtime.registerNativeProvider( + { + ...f.provider, + stream: faux.provider.stream, + streamSimple: faux.provider.streamSimple, + }, + { refresh: false }, + ); + faux.setResponses([fauxAssistantMessage("pinned")]); + for await (const event of f.runtime.streamSimple(f.model, { messages: [] }, { apiKey: token(name) })) { + if (event.type === "error") throw new Error(event.error.errorMessage); + if (event.type === "done") break; + } + expect(faux.getCallLog().map((call) => call.options?.apiKey)).toEqual([token(name)]); + expect(f.fetchUsage).not.toHaveBeenCalled(); + } finally { + f.cleanup(); + } +}); + +it("HTTP requests use the same quota selection and fault-isolated notices", async () => { + const f = await fixture(); + const diagnostics = vi.spyOn(console, "error").mockImplementation(() => {}); + const notices: string[] = []; + const removeBad = subscribeAccountSwitch(() => { + throw new Error("private observer error"); + }); + const removeGood = subscribeAccountSwitch((event) => { + notices.push(event.to); + }); + try { + const key = await f.runtime.requestWithCredentialRotation( + f.model, + { sessionId: "request-test" }, + async (prepared) => prepared.options.apiKey, + ); + expect(key).toBe(token("ready")); + expect(notices).toEqual(["ready"]); + expect(diagnostics).toHaveBeenCalledTimes(1); + expect(JSON.stringify(diagnostics.mock.calls)).not.toContain("private observer error"); + } finally { + removeBad(); + removeGood(); + f.cleanup(); + } +}); + +it("an account reserved by another half-open probe cannot authorize paid usage", async () => { + const f = await fixture(); + try { + const repository = new CredentialSlotRepository(join(f.dir, "credential-pool-state.json")); + const credentialRevision = await repository.storedCredentialRevision("openai-codex", "exhausted", { + access: token("exhausted"), + refresh: "test-refresh-exhausted", + }); + await repository.mutateSlotState("openai-codex", "stored", "exhausted", () => ({ + credentialRevision, + blockedUntil: 1, + blockReason: "rate_limit", + lease: { id: "another-request", expiresAt: 9e12 }, + })); + expect((await repository.listSlots("openai-codex", "stored")).exhausted?.lease?.id).toBe("another-request"); + f.fetchUsage.mockImplementation(async (_url, init) => { + const included = new Headers(init?.headers).get("authorization") === `Bearer ${token("exhausted")}`; + return Response.json({ + rate_limit: { + allowed: included, + limit_reached: !included, + primary_window: { used_percent: included ? 20 : 100 }, + }, + credits: { has_credits: !included, balance: "100" }, + }); + }); + const attempt = vi.fn(async () => "must not spend"); + let failure: unknown; + try { + await f.runtime.requestWithCredentialRotation(f.model, undefined, attempt); + } catch (error) { + failure = error; + } + expect( + f.fetchUsage.mock.calls.map( + ([, init]) => new Headers(init?.headers).get("authorization") === `Bearer ${token("exhausted")}`, + ), + ).toContain(true); + expect(failure).toBeInstanceOf(Error); + expect(String(failure)).toContain("No credential slots available"); + expect(attempt).not.toHaveBeenCalled(); + } finally { + f.cleanup(); + } +}); diff --git a/packages/coding-agent/test/suite/internal-model-request.test.ts b/packages/coding-agent/test/suite/internal-model-request.test.ts new file mode 100644 index 000000000..3399e63d0 --- /dev/null +++ b/packages/coding-agent/test/suite/internal-model-request.test.ts @@ -0,0 +1,210 @@ +import { mkdtempSync, rmSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { + type Api, + type AssistantMessageEvent, + type AssistantMessageEventStream, + type Context, + fauxAssistantMessage, + lazyStream, + type Model, +} from "@earendil-works/pi-ai"; +import { afterEach, beforeEach, describe, expect, it } from "vitest"; +import { + type InternalModelSettings, + type InternalStreamRuntime, + streamInternalModel, +} from "../../src/core/internal-model-request.ts"; + +let agentDir: string; +beforeEach(() => { + agentDir = mkdtempSync(join(tmpdir(), "internal-request-")); +}); +afterEach(() => { + rmSync(agentDir, { recursive: true, force: true }); +}); + +function model(provider: string, id: string, inputs: Model["input"] = ["text"]): Model { + return { + id, + provider, + api: "openai-completions", + name: id, + baseUrl: "https://example.invalid", + maxTokens: 2048, + contextWindow: 128_000, + reasoning: false, + thinkingLevelMap: {}, + input: inputs, + cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0 }, + }; +} + +function imageContext(): Context { + return { + systemPrompt: "system", + messages: [ + { role: "user", content: [{ type: "image" as const, data: "data", mimeType: "image/png" }], timestamp: 0 }, + ], + }; +} + +function textContext(): Context { + return { systemPrompt: "system", messages: [{ role: "user", content: "hi", timestamp: 0 }] }; +} + +function okEvent(): AssistantMessageEvent { + return { + type: "done", + reason: "stop", + message: fauxAssistantMessage("ok", { timestamp: 0 }), + }; +} + +function errorEvent(message: string): AssistantMessageEvent { + return { + type: "error", + reason: "error", + error: fauxAssistantMessage("", { stopReason: "error", errorMessage: message, timestamp: 0 }), + }; +} + +function eventStream(events: AssistantMessageEvent[]): AssistantMessageEventStream { + return lazyStream(model("test", "test"), async () => ({ + async *[Symbol.asyncIterator](): AsyncGenerator { + yield* events; + }, + })); +} + +function settingsFor(sets: Record): InternalModelSettings { + return { + getRetryFallbackSettings: () => ({ modelFallback: true, chains: sets, revertPolicy: "never" }), + }; +} + +interface CapturedAttempt { + model: Model; + options: Record; +} + +function makeRuntime( + primary: RuntimeModel, + answers: ((model: RuntimeModel) => AssistantMessageEventStream)[], + overrides: Partial = {}, +): { runtime: InternalStreamRuntime; captured: CapturedAttempt[] } { + const captured: CapturedAttempt[] = []; + let attempt = 0; + const defaultModels = () => [primary]; + return { + captured, + runtime: { + streamSimple: (next, _context, options) => { + const answer = answers[Math.min(attempt, answers.length - 1)]; + captured.push({ model: next, options: { ...options } }); + attempt += 1; + return answer(next); + }, + getModels: () => defaultModels(), + getModel: (provider, id) => (provider === primary.provider && id === primary.id ? primary : undefined), + isUsingOAuth: (provider) => provider === "openai-codex", + isFallbackEligible: () => true, + hasConfiguredAuth: () => true, + ...overrides, + }, + }; +} + +type RuntimeModel = Model; + +describe("streamInternalModel", () => { + it("strips a pre-resolved Codex OAuth apiKey before dispatch", async () => { + const primary = model("openai-codex", "model"); + const { runtime, captured } = makeRuntime(primary, [() => eventStream([okEvent()])]); + + const events: AssistantMessageEvent[] = []; + for await (const event of await streamInternalModel( + runtime, + primary, + textContext(), + { apiKey: "codex-token" }, + { + settings: settingsFor({}), + agentDir, + }, + )) { + events.push(event); + } + + expect(events.some((event) => event.type === "done")).toBe(true); + expect(captured).toHaveLength(1); + expect(captured[0]?.options.apiKey).toBeUndefined(); + }); + + it("drops credential material on a cross-provider fallback", async () => { + const primarySource = model("anthropic", "primary"); + const fallbackSource = model("openai", "fallback"); + const { runtime, captured } = makeRuntime( + primarySource, + [() => eventStream([errorEvent("primary failed")]), () => eventStream([okEvent()])], + { + getModels: () => [primarySource, fallbackSource], + getModel: (provider, id) => (provider === "openai" && id === "fallback" ? fallbackSource : undefined), + isUsingOAuth: () => false, + }, + ); + + const events: AssistantMessageEvent[] = []; + for await (const event of await streamInternalModel( + runtime, + primarySource, + textContext(), + { + apiKey: "primary-key", + headers: { authorization: "Bearer primary-key" }, + extraBody: { x: 1 }, + env: { FOO: "bar" }, + reasoningEffort: "high", + }, + { settings: settingsFor({ "anthropic/primary": ["openai/fallback"] }), agentDir }, + )) { + events.push(event); + } + + expect(events.some((event) => event.type === "done")).toBe(true); + expect(captured).toHaveLength(2); + expect(captured[1]?.model.id).toBe("fallback"); + expect(captured[1]?.options.apiKey).toBeUndefined(); + expect(captured[1]?.options.headers).toBeUndefined(); + expect(captured[1]?.options.extraBody).toBeUndefined(); + expect(captured[1]?.options.env).toBeUndefined(); + expect(captured[1]?.options.reasoningEffort).toBeUndefined(); + expect(captured[1]?.options.maxRetries).toBe(0); + }); + + it("does not fall back to a text-only model when the conversation has images", async () => { + const primarySource = model("anthropic", "primary"); + const textOnlySource = model("openai", "fallback", ["text"]); + const { runtime, captured } = makeRuntime(primarySource, [() => eventStream([errorEvent("primary failed")])], { + getModels: () => [primarySource, textOnlySource], + getModel: (provider, id) => (provider === "openai" && id === "fallback" ? textOnlySource : undefined), + isUsingOAuth: () => false, + }); + + const events: AssistantMessageEvent[] = []; + for await (const event of await streamInternalModel( + runtime, + primarySource, + imageContext(), + {}, + { settings: settingsFor({ "anthropic/primary": ["openai/fallback"] }), agentDir }, + )) { + events.push(event); + } + + expect(events.some((event) => event.type === "error")).toBe(true); + expect(captured).toHaveLength(1); + expect(captured[0]?.model.id).toBe("primary"); + }); +}); From 5c0832a27625c76255305adca2dd9a6863b6267b Mon Sep 17 00:00:00 2001 From: Daniel Choi Date: Sun, 20 Sep 2026 13:44:54 -0400 Subject: [PATCH 3/6] docs(extensions): complete fallback policy tracker coverage --- packages/coding-agent/src/core/extensions/changes.md | 10 ++++++++-- 1 file changed, 8 insertions(+), 2 deletions(-) diff --git a/packages/coding-agent/src/core/extensions/changes.md b/packages/coding-agent/src/core/extensions/changes.md index 0c20eb867..d2f04eca9 100644 --- a/packages/coding-agent/src/core/extensions/changes.md +++ b/packages/coding-agent/src/core/extensions/changes.md @@ -94,13 +94,19 @@ ### What changed -- `src/core/extensions/types.ts`: `ExtensionContext` gains an optional `getRetryFallbackSettings(): RetryFallbackSettings` getter. -- `src/core/extensions/runner.ts`: `createContext()` exposes that getter from the already-bound `sessionSettingsFn`, so an extension auxiliary request reads the active session's resolved fallback configuration rather than reconstructing one from the global agent directory. +- `packages/coding-agent/src/core/extensions/types.ts`: `ExtensionContext` gains an optional `getRetryFallbackSettings(): RetryFallbackSettings` getter. +- `packages/coding-agent/src/core/extensions/runner.ts`: `createContext()` exposes that getter from the already-bound `sessionSettingsFn`, so an extension auxiliary request reads the active session's resolved fallback configuration rather than reconstructing one from the global agent directory. ### Why Title/compaction/btw/vision auxiliary requests need the session's configured fallback chains, including SDK in-memory settings, without each builtin importing `SettingsManager` and reaching for `getAgentDir()`. +### Why an extension could not handle it + +Only the host binds the active session settings into the extension context. +An external extension cannot add a typed context capability or access SDK +in-memory overrides through the global settings file. + ### Expected merge conflict zones - LOW: the new optional member in `ExtensionContext` and the one getter in `createContext()`. From fb3c5e4b3ba39614996f078661fc33f5334f22c7 Mon Sep 17 00:00:00 2001 From: Daniel Choi Date: Sun, 20 Sep 2026 15:05:01 -0400 Subject: [PATCH 4/6] fix(internal): preserve native error identity without a fallback chain --- packages/coding-agent/src/core/changes.md | 8 ++-- .../src/core/internal-model-request.ts | 48 ++++++++++++++----- 2 files changed, 41 insertions(+), 15 deletions(-) diff --git a/packages/coding-agent/src/core/changes.md b/packages/coding-agent/src/core/changes.md index c706c22b6..ea3fbc82b 100644 --- a/packages/coding-agent/src/core/changes.md +++ b/packages/coding-agent/src/core/changes.md @@ -404,9 +404,11 @@ The provider id is resolved inside the package before any extension loads, and t helper and native title/summary/remote-compaction/side-query/vision regressions. - `packages/coding-agent/test/compaction/before-compact-error-surfacing.test.ts`: the session fixture now supplies its required `getSessionId` contract. -- When no session fallback policy is provided or it is disabled, the internal - helper delegates the original stream directly, preserving rejection identity - and cancellation rather than creating another error-conversion layer. +- When no session fallback policy is provided, it is disabled, or the active + model has no configured fallback chain, the internal helper delegates the + original stream directly, preserving rejection identity, synchronous-throw + semantics, and cancellation rather than relabeling failures through another + lazy error-event layer. ### Why diff --git a/packages/coding-agent/src/core/internal-model-request.ts b/packages/coding-agent/src/core/internal-model-request.ts index a78980d0e..bcb746007 100644 --- a/packages/coding-agent/src/core/internal-model-request.ts +++ b/packages/coding-agent/src/core/internal-model-request.ts @@ -97,14 +97,13 @@ export function streamInternalModel( const { apiKey, ...neutral } = effectiveOptions; effectiveOptions = neutral; } - const settings = config.settings; - // Without a session fallback policy, preserve the native stream and its - // original rejection/cancellation semantics after removing the OAuth snapshot. - if (!settings?.getRetryFallbackSettings().modelFallback) { - return config.streamFn - ? config.streamFn(model, context, effectiveOptions) - : runtime.streamSimple(model, context, effectiveOptions); - } + const settings: InternalModelSettings = config.settings ?? { + getRetryFallbackSettings: (): ResolvedRetryFallbackSettings => ({ + modelFallback: false, + chains: {}, + revertPolicy: "cooldown-expiry", + }), + }; const agentDir = config.agentDir ?? getAgentDir(); let cooldowns = config.cooldowns ?? cooldownsByRuntime.get(runtime); if (!cooldowns) { @@ -143,6 +142,15 @@ export function streamInternalModel( }); }, }); + // A session without model fallback, or a model without a configured fallback + // chain, must keep the native stream exactly as the caller supplied it. This + // preserves rejection identity and synchronous-throw semantics; wrapping the + // native stream here would relabel its failure into an error event. + if (!settings.getRetryFallbackSettings().modelFallback || !controller.hasConfiguredChain()) { + return config.streamFn + ? config.streamFn(model, context, effectiveOptions) + : runtime.streamSimple(model, context, effectiveOptions); + } return lazyStream(model, async () => (async function* () { if (cooldowns.isSuppressed(`${model.provider}/${model.id}`)) { @@ -161,10 +169,26 @@ export function streamInternalModel( maxRetries: 0, } : effectiveOptions; - const stream = - changed || !config.streamFn - ? runtime.streamSimple(current.model, context, requestOptions) - : await config.streamFn(current.model, context, requestOptions); + let stream: AssistantMessageEventStream; + try { + stream = + changed || !config.streamFn + ? runtime.streamSimple(current.model, context, requestOptions) + : await config.streamFn(current.model, context, requestOptions); + } catch (streamError) { + // A synchronous throw from the native stream must reach the caller + // as the original error, not as a lazy-wrapped error event: summary + // consumers otherwise relabel it and change their reported message. + if ( + effectiveOptions.signal?.aborted || + !(await controller.tryFallback("hard-error", { + errorMessage: streamError instanceof Error ? streamError.message : String(streamError), + })) + ) { + throw streamError; + } + continue; + } let committed = false; let failure: Extract | undefined; for await (const event of stream) { From ddc9666038a4b9833c7b6bd49889e76498b9133c Mon Sep 17 00:00:00 2001 From: Daniel Choi Date: Tue, 22 Sep 2026 20:09:57 -0400 Subject: [PATCH 5/6] fix(credential-pool): follow the chatgpt-subscription provider rename The subscription provider id was renamed from `openai-codex` to `chatgpt-subscription` (senpi#1989) after this branch's base, so the quota admission, the account-switch notices and the internal-request OAuth strip now compare through pi-ai's `normalizeProviderId`, and the account-id header rebuild uses the renamed `extractChatGptSubscriptionAccountId` helper. Legacy ids keep resolving through the same helper, so the existing fixtures still exercise the alias path. Constraint: Upstream renamed the provider id after this branch's base Rejected: Comparing the canonical id directly | legacy ids in fixtures and stored configs would stop routing Confidence: high Scope-risk: narrow Not-tested: Windows-only CI jobs --- .../src/core/credential-pool/codex-quota.ts | 3 ++- .../core/credential-pool/rotation-stream.ts | 11 +++++------ .../extensions/builtin/compaction/index.ts | 5 +++-- .../builtin/compaction/openai-remote.ts | 6 +++++- .../src/core/internal-model-request.ts | 3 ++- .../coding-agent/src/core/model-runtime.ts | 4 ++-- .../test/suite/account-notices.test.ts | 2 +- .../suite/codex-auxiliary-routing.test.mjs | 18 +++++++++--------- .../test/suite/codex-quota.test.ts | 12 ++++++------ .../test/suite/codex-runtime-routing.test.ts | 16 +++++++++------- .../test/suite/internal-model-request.test.ts | 4 ++-- 11 files changed, 46 insertions(+), 38 deletions(-) diff --git a/packages/coding-agent/src/core/credential-pool/codex-quota.ts b/packages/coding-agent/src/core/credential-pool/codex-quota.ts index 1df5399fc..a1e8082f1 100644 --- a/packages/coding-agent/src/core/credential-pool/codex-quota.ts +++ b/packages/coding-agent/src/core/credential-pool/codex-quota.ts @@ -1,3 +1,4 @@ +import { normalizeProviderId } from "@earendil-works/pi-ai"; import { z } from "zod"; import type { RotationSlot, RotationSources } from "./rotation-stream.ts"; @@ -59,7 +60,7 @@ export async function admitCodexQuota( sources: Pick, slots: RotationSlot[], ): Promise { - if (sources.providerId !== "openai-codex") return slots; + if (normalizeProviderId(sources.providerId) !== "chatgpt-subscription") return slots; const getUsage = sources.getCodexUsage; if (!getUsage) throw new Error("Codex quota admission unavailable: missing quota reader"); const assessed = await Promise.all( diff --git a/packages/coding-agent/src/core/credential-pool/rotation-stream.ts b/packages/coding-agent/src/core/credential-pool/rotation-stream.ts index 063174b09..3d97af1cf 100644 --- a/packages/coding-agent/src/core/credential-pool/rotation-stream.ts +++ b/packages/coding-agent/src/core/credential-pool/rotation-stream.ts @@ -1,5 +1,5 @@ import { createHash, randomUUID } from "node:crypto"; -import type { AssistantMessageEvent, Credential } from "@earendil-works/pi-ai"; +import { type AssistantMessageEvent, type Credential, normalizeProviderId } from "@earendil-works/pi-ai"; import { rendezvousOrder, type SlotHasher } from "@earendil-works/pi-ai/auth/pool/select"; import { listSlots as listCredentialSlots, type PooledCredential } from "@earendil-works/pi-ai/auth/pool/slots"; import { resolveConfigValue } from "../resolve-config-value.ts"; @@ -7,9 +7,8 @@ import { emitAccountSwitch } from "./account-notices.ts"; import { type CredentialBlock, classifyCredentialFailure } from "./classify.ts"; import { admitCodexQuota, preferredCodexQuotaTier } from "./codex-quota.ts"; import { discoverEnvSlots } from "./env-slots.ts"; -import { type RunSlot, runCredentialFailover } from "./failover.ts"; -import { isCommittedRotationOutput, isRotationStreamStart, rotationErrorFromEvent } from "./rotation-events.ts"; import { type RunCredentialFailoverOptions, type RunSlot, runCredentialFailover } from "./failover.ts"; +import { isCommittedRotationOutput, isRotationStreamStart, rotationErrorFromEvent } from "./rotation-events.ts"; import { acquireHalfOpenLease, type CredentialSlotRepository, type CredentialSlotState } from "./state-store.ts"; /** The exact hash the claude-sdk-oauth affinity oracle uses, so pools never remap. */ @@ -265,6 +264,7 @@ function runRotation( const useAffinity = sources.policy?.affinity !== false; const now = sources.now ?? Date.now; const selectionKey = `${sources.providerId}\0${sources.sessionId ?? affinityKey}`; + const codexProvider = normalizeProviderId(sources.providerId) === "chatgpt-subscription"; let previous = sources.selectionState?.get(selectionKey); let previousFailure: string | undefined; let admitted: RotationSlot[] = []; @@ -274,8 +274,7 @@ function runRotation( const slots = await listRotationSlots(sources); // A different caller's probe lease hides a generation candidate, not // its remaining normal quota. Assess the full inventory before filtering. - const inventory = - sources.providerId === "openai-codex" ? await listRotationSlots(sources, { acquireLeases: false }) : slots; + const inventory = codexProvider ? await listRotationSlots(sources, { acquireLeases: false }) : slots; previous ??= inventory.find((slot) => slot.pinned)?.name ?? (useAffinity ? rendezvousOrder(affinityKey, inventory, hasher) : inventory)[0]?.name; @@ -291,7 +290,7 @@ function runRotation( const winner = pinned ?? ordered[0]; if (!winner) throw new Error("credential rotation selected from an empty candidate set"); - if (sources.providerId === "openai-codex" && previous && previous !== winner.name) { + if (codexProvider && previous && previous !== winner.name) { const prior = admitted.find((slot) => slot.name === previous); const reason = previousFailure ?? diff --git a/packages/coding-agent/src/core/extensions/builtin/compaction/index.ts b/packages/coding-agent/src/core/extensions/builtin/compaction/index.ts index c3092a994..3aec405d7 100644 --- a/packages/coding-agent/src/core/extensions/builtin/compaction/index.ts +++ b/packages/coding-agent/src/core/extensions/builtin/compaction/index.ts @@ -1,4 +1,4 @@ -import type { Tool } from "@earendil-works/pi-ai"; +import { normalizeProviderId, type Tool } from "@earendil-works/pi-ai"; import type { CompactionResult } from "../../../compaction/index.ts"; import { createWarmAnchorSnapshot, isWarmSummaryAnchorValid } from "../../../compaction/warm-anchor.ts"; import type { ExtensionAPI, ExtensionContext, SessionBeforeCompactEvent, SessionCompactEvent } from "../../types.ts"; @@ -985,7 +985,8 @@ export default function compactionExtension( effectiveModel, { ...auth, - ...(model.provider === "openai-codex" && selectedAuthorization?.startsWith("Bearer ") + ...(normalizeProviderId(model.provider) === "chatgpt-subscription" && + selectedAuthorization?.startsWith("Bearer ") ? { apiKey: selectedAuthorization.slice(7) } : {}), headers: event.headers ?? auth.headers, diff --git a/packages/coding-agent/src/core/extensions/builtin/compaction/openai-remote.ts b/packages/coding-agent/src/core/extensions/builtin/compaction/openai-remote.ts index 0f7b766f8..a5d8e8664 100644 --- a/packages/coding-agent/src/core/extensions/builtin/compaction/openai-remote.ts +++ b/packages/coding-agent/src/core/extensions/builtin/compaction/openai-remote.ts @@ -6,6 +6,7 @@ import { convertResponsesMessages, getContextProvenance, type Model, + normalizeProviderId, type ProviderHeaders, } from "@earendil-works/pi-ai"; import { streamSimple } from "@earendil-works/pi-ai/compat"; @@ -850,7 +851,10 @@ export async function runOpenAiRemoteCompaction( }; const runtime = ctx.modelRegistry?.modelRuntime; - if (requestModel.provider === "openai-codex" && runtime?.requestWithCredentialRotation) { + if ( + normalizeProviderId(requestModel.provider) === "chatgpt-subscription" && + runtime?.requestWithCredentialRotation + ) { try { return await runtime.requestWithCredentialRotation( requestModel, diff --git a/packages/coding-agent/src/core/internal-model-request.ts b/packages/coding-agent/src/core/internal-model-request.ts index bcb746007..e4c94c027 100644 --- a/packages/coding-agent/src/core/internal-model-request.ts +++ b/packages/coding-agent/src/core/internal-model-request.ts @@ -8,6 +8,7 @@ import { lazyStream, type Model, type ModelsSimpleStreamOptions, + normalizeProviderId, } from "@earendil-works/pi-ai"; import { getAgentDir } from "../config.ts"; import { RetryFallbackController } from "./retry-fallback/controller.ts"; @@ -93,7 +94,7 @@ export function streamInternalModel( config: StreamInternalModelConfig = {}, ): AssistantMessageEventStream | Promise { let effectiveOptions = options; - if (model.provider === "openai-codex" && runtime.isUsingOAuth(model.provider)) { + if (normalizeProviderId(model.provider) === "chatgpt-subscription" && runtime.isUsingOAuth(model.provider)) { const { apiKey, ...neutral } = effectiveOptions; effectiveOptions = neutral; } diff --git a/packages/coding-agent/src/core/model-runtime.ts b/packages/coding-agent/src/core/model-runtime.ts index cd7501a61..4c9ee4f3a 100644 --- a/packages/coding-agent/src/core/model-runtime.ts +++ b/packages/coding-agent/src/core/model-runtime.ts @@ -17,7 +17,7 @@ import { type DeferredCancelOptions, type DeferredFetchOptions, type DeferredHandle, - extractOpenAiCodexAccountId, + extractChatGptSubscriptionAccountId, lazyStream, type Model, type Models, @@ -179,7 +179,7 @@ function withPayloadRequestMetadata(options: StreamOptions, model: Model): if (["authorization", "chatgpt-account-id"].includes(name.toLowerCase())) delete headers[name]; } headers.authorization = `Bearer ${options.apiKey}`; - const accountId = extractOpenAiCodexAccountId(options.apiKey); + const accountId = extractChatGptSubscriptionAccountId(options.apiKey); if (accountId) headers["chatgpt-account-id"] = accountId; } return { diff --git a/packages/coding-agent/test/suite/account-notices.test.ts b/packages/coding-agent/test/suite/account-notices.test.ts index 62b7fa58a..23d2a7048 100644 --- a/packages/coding-agent/test/suite/account-notices.test.ts +++ b/packages/coding-agent/test/suite/account-notices.test.ts @@ -10,7 +10,7 @@ import { createHarness } from "./harness.ts"; const event: AccountSwitchNotice = { type: "account_failover", - provider: "openai-codex", + provider: "chatgpt-subscription", from: "slot-a", to: "slot-b", reason: "quota exhausted", diff --git a/packages/coding-agent/test/suite/codex-auxiliary-routing.test.mjs b/packages/coding-agent/test/suite/codex-auxiliary-routing.test.mjs index 6b5ed48a7..e4245f8d5 100644 --- a/packages/coding-agent/test/suite/codex-auxiliary-routing.test.mjs +++ b/packages/coding-agent/test/suite/codex-auxiliary-routing.test.mjs @@ -19,7 +19,7 @@ import { generateBranchSummary } from "../../src/core/compaction/branch-summariz const fallbackSettings = { retry: { enabled: true, maxRetries: 0, modelFallback: true, fallbackRevertPolicy: "never", - fallbackChains: { "openai-codex/gpt-6-astra": ["openrouter/deepseek/deepseek-v4-pro-0813:max"] } }, + fallbackChains: { "chatgpt-subscription/gpt-6-astra": ["openrouter/deepseek/deepseek-v4-pro-0813:max"] } }, }; const token = (name) => `test.${Buffer.from(JSON.stringify({ @@ -42,13 +42,13 @@ async function fixture(t, allExhausted = false) { name, access: token(name), refresh: `test-refresh-${name}`, expires: 9e12, })); const credentials = AuthStorage.inMemory({ - "openai-codex": { type: "oauth", ...accounts[0], pinned: "exhausted", accounts }, + "chatgpt-subscription": { type: "oauth", ...accounts[0], pinned: "exhausted", accounts }, openrouter: { type: "api_key", key: "test-openrouter-key" }, }); const runtime = await ModelRuntime.create({ credentials, agentDir: dir, modelsPath: null, allowModelNetwork: false }); const attempts = []; const attemptedModels = []; - const provider = runtime.getProvider("openai-codex"); + const provider = runtime.getProvider("chatgpt-subscription"); const produce = (model, _context, options) => { attempts.push(options.apiKey); attemptedModels.push(`${model.provider}/${model.id}`); @@ -77,7 +77,7 @@ async function fixture(t, allExhausted = false) { if (sources) sources.getCodexUsage = async (slot) => usage(allExhausted || slot.name === "exhausted" ? 100 : 0); return sources; }; - return { runtime, model: runtime.getModel("openai-codex", "gpt-6-astra"), attempts, attemptedModels, dir, credentials }; + return { runtime, model: runtime.getModel("chatgpt-subscription", "gpt-6-astra"), attempts, attemptedModels, dir, credentials }; } async function makeSession(t, f) { @@ -149,7 +149,7 @@ test("native title generation uses the configured model fallback after account e await session._generateSessionTitle("Repair credential account selection", f.model, new AbortController()); assert.ok(session.sessionManager.getSessionName()); assert.deepEqual(f.attemptedModels, ["openrouter/deepseek/deepseek-v4-pro-0813"]); - assert.equal(session.model.provider, "openai-codex"); + assert.equal(session.model.provider, "chatgpt-subscription"); }); test("speculative summary uses the configured model fallback without Codex generation", async (t) => { @@ -247,11 +247,11 @@ function remoteRequest(f, model, runtime, fetchImpl) { }, undefined, { fetch: fetchImpl }); } -for (const provider of ["openai-codex", "openai"]) { +for (const provider of ["chatgpt-subscription", "openai"]) { for (const status of [401, 402, 403, 429]) { test(`legacy ${provider} remote HTTP ${status} preserves local fallback`, async (t) => { const f = await fixture(t); - const model = provider === "openai-codex" ? f.model : { + const model = provider === "chatgpt-subscription" ? f.model : { ...f.model, provider, api: "openai-responses", baseUrl: "https://example.invalid/v1", compat: { supportsWebSocket: false, supportsRemoteCompactionV2: false }, }; @@ -330,7 +330,7 @@ test("the native side-query command falls back without changing the main model", }; await command.handler("Which account can serve this request?", ctx); assert.deepEqual(f.attemptedModels, ["openrouter/deepseek/deepseek-v4-pro-0813"]); - assert.equal(ctx.model.provider, "openai-codex"); + assert.equal(ctx.model.provider, "chatgpt-subscription"); assert.equal(notices.some((notice) => notice.kind === "error"), false); assert.ok(notices.some((notice) => notice.kind === "info" && notice.message.length > 0)); }); @@ -341,7 +341,7 @@ function visionContext(f) { agentDir: f.dir, cwd: f.dir, getRetryFallbackSettings: () => SettingsManager.inMemory(fallbackSettings).getRetryFallbackSettings(), getImageSettings: () => ({ blockImages: false, autoResize: false }), - getLookAtSettings: () => ({ models: ["openai-codex/gpt-6-astra:low"] }), + getLookAtSettings: () => ({ models: ["chatgpt-subscription/gpt-6-astra:low"] }), sessionManager: { getSessionId: () => "vision-test" }, modelRegistry: { modelRuntime: f.runtime, getAvailable: () => [f.model], diff --git a/packages/coding-agent/test/suite/codex-quota.test.ts b/packages/coding-agent/test/suite/codex-quota.test.ts index 2761bd2aa..61b54a68d 100644 --- a/packages/coding-agent/test/suite/codex-quota.test.ts +++ b/packages/coding-agent/test/suite/codex-quota.test.ts @@ -32,7 +32,7 @@ describe("Codex paid quota admission", () => { const lookedUp: string[] = []; const slots = await admitCodexQuota( { - providerId: "openai-codex", + providerId: "chatgpt-subscription", getCodexUsage: async (slot) => { lookedUp.push(slot.name); return quota(slot.name === "cooling" ? used : 100, slot.name === "paid"); @@ -51,7 +51,7 @@ describe("Codex paid quota admission", () => { it.each(["auth_error", "account_disabled"] as const)("treats %s quota as unknown", async (blockReason) => { const slots = await admitCodexQuota( { - providerId: "openai-codex", + providerId: "chatgpt-subscription", getCodexUsage: async () => quota(100, true), }, [ @@ -65,7 +65,7 @@ describe("Codex paid quota admission", () => { it("does not confuse denial with quota exhaustion", async () => { const slots = await admitCodexQuota( { - providerId: "openai-codex", + providerId: "chatgpt-subscription", getCodexUsage: async () => quota(20, true, false), }, [{ name: "denied", lane: "stored" }], @@ -76,7 +76,7 @@ describe("Codex paid quota admission", () => { it.each(["error", "malformed"])("preserves healthy included quota when another account is %s", async (mode) => { const slots = await admitCodexQuota( { - providerId: "openai-codex", + providerId: "chatgpt-subscription", getCodexUsage: async (slot) => { if (slot.name === "unknown") { if (mode === "error") throw new Error("test transport failure"); @@ -93,7 +93,7 @@ describe("Codex paid quota admission", () => { it("requires exhaustion of matching model-specific quota as well", async () => { const slots = await admitCodexQuota( { - providerId: "openai-codex", + providerId: "chatgpt-subscription", modelId: "test-model", getCodexUsage: async () => ({ ...quota(100, true), @@ -112,7 +112,7 @@ describe("Codex paid quota admission", () => { await expect( admitCodexQuota( { - providerId: "openai-codex", + providerId: "chatgpt-subscription", signal: controller.signal, getCodexUsage: async () => { throw reason; diff --git a/packages/coding-agent/test/suite/codex-runtime-routing.test.ts b/packages/coding-agent/test/suite/codex-runtime-routing.test.ts index 581cbdb0b..f6c896778 100644 --- a/packages/coding-agent/test/suite/codex-runtime-routing.test.ts +++ b/packages/coding-agent/test/suite/codex-runtime-routing.test.ts @@ -27,7 +27,7 @@ async function fixture() { expires: 9e12, })); const credentials = AuthStorage.inMemory({ - "openai-codex": { + "chatgpt-subscription": { type: "oauth", access: token("exhausted"), refresh: "test-refresh-exhausted", @@ -42,8 +42,8 @@ async function fixture() { modelsPath: null, allowModelNetwork: false, }); - const provider = runtime.getProvider("openai-codex"); - const model = runtime.getModels().find((candidate) => candidate.provider === "openai-codex"); + const provider = runtime.getProvider("chatgpt-subscription"); + const model = runtime.getModels().find((candidate) => candidate.provider === "chatgpt-subscription"); if (!provider || !model) throw new Error("Codex builtin is required"); const fetchUsage = vi.fn(async (_url: unknown, init?: RequestInit) => { const exhausted = new Headers(init?.headers).get("authorization") === `Bearer ${token("exhausted")}`; @@ -62,7 +62,7 @@ async function fixture() { it.each(["exhausted", "external"])("explicit %s credentials bypass Codex rotation", async (name) => { const f = await fixture(); - const faux = fauxProvider({ provider: "openai-codex" }); + const faux = fauxProvider({ provider: "chatgpt-subscription" }); try { await f.runtime.registerNativeProvider( { @@ -115,17 +115,19 @@ it("an account reserved by another half-open probe cannot authorize paid usage", const f = await fixture(); try { const repository = new CredentialSlotRepository(join(f.dir, "credential-pool-state.json")); - const credentialRevision = await repository.storedCredentialRevision("openai-codex", "exhausted", { + const credentialRevision = await repository.storedCredentialRevision("chatgpt-subscription", "exhausted", { access: token("exhausted"), refresh: "test-refresh-exhausted", }); - await repository.mutateSlotState("openai-codex", "stored", "exhausted", () => ({ + await repository.mutateSlotState("chatgpt-subscription", "stored", "exhausted", () => ({ credentialRevision, blockedUntil: 1, blockReason: "rate_limit", lease: { id: "another-request", expiresAt: 9e12 }, })); - expect((await repository.listSlots("openai-codex", "stored")).exhausted?.lease?.id).toBe("another-request"); + expect((await repository.listSlots("chatgpt-subscription", "stored")).exhausted?.lease?.id).toBe( + "another-request", + ); f.fetchUsage.mockImplementation(async (_url, init) => { const included = new Headers(init?.headers).get("authorization") === `Bearer ${token("exhausted")}`; return Response.json({ diff --git a/packages/coding-agent/test/suite/internal-model-request.test.ts b/packages/coding-agent/test/suite/internal-model-request.test.ts index 3399e63d0..09901597d 100644 --- a/packages/coding-agent/test/suite/internal-model-request.test.ts +++ b/packages/coding-agent/test/suite/internal-model-request.test.ts @@ -108,7 +108,7 @@ function makeRuntime( }, getModels: () => defaultModels(), getModel: (provider, id) => (provider === primary.provider && id === primary.id ? primary : undefined), - isUsingOAuth: (provider) => provider === "openai-codex", + isUsingOAuth: (provider) => provider === "chatgpt-subscription", isFallbackEligible: () => true, hasConfiguredAuth: () => true, ...overrides, @@ -120,7 +120,7 @@ type RuntimeModel = Model; describe("streamInternalModel", () => { it("strips a pre-resolved Codex OAuth apiKey before dispatch", async () => { - const primary = model("openai-codex", "model"); + const primary = model("chatgpt-subscription", "model"); const { runtime, captured } = makeRuntime(primary, [() => eventStream([okEvent()])]); const events: AssistantMessageEvent[] = []; From d7bcb8473658417aa4ad37fbb3084994c693fe99 Mon Sep 17 00:00:00 2001 From: Daniel Choi Date: Tue, 22 Sep 2026 20:21:09 -0400 Subject: [PATCH 6/6] fix(credential-pool): keep the rotation hooks type-complete runRotation's option Pick omitted isStreamStart, so the rotation-events threading and the non-streaming wrapper failed to typecheck on the rebased base. The Pick now names all three hooks and the non-streaming call states its event type explicitly. Constraint: Upstream's RunCredentialFailoverOptions gained isStreamStart Confidence: high Scope-risk: narrow --- .../src/core/credential-pool/rotation-stream.ts | 7 +++++-- 1 file changed, 5 insertions(+), 2 deletions(-) diff --git a/packages/coding-agent/src/core/credential-pool/rotation-stream.ts b/packages/coding-agent/src/core/credential-pool/rotation-stream.ts index 3d97af1cf..326c8976b 100644 --- a/packages/coding-agent/src/core/credential-pool/rotation-stream.ts +++ b/packages/coding-agent/src/core/credential-pool/rotation-stream.ts @@ -240,7 +240,7 @@ export async function requestWithCredentialRotation( options: Omit & { runAttempt: (slot: RotationSlot) => Promise }, ): Promise { let result: { value: T } | undefined; - for await (const event of runRotation({ + for await (const event of runRotation<{ value: T }>({ ...options, runAttempt: async function* (slot) { yield { value: await options.runAttempt(slot) }; @@ -256,7 +256,10 @@ export async function requestWithCredentialRotation( function runRotation( options: Omit & - Pick, "runAttempt" | "isCommittedOutput" | "errorFromEvent">, + Pick< + RunCredentialFailoverOptions, + "runAttempt" | "isCommittedOutput" | "isStreamStart" | "errorFromEvent" + >, ): AsyncGenerator { const { sources, runAttempt } = options; const hasher = options.hasher ?? sha256SlotHasher;