fix(auth): rotate through quota-limited accounts

Keep ordinary 401 retries bounded while replay-safe quota failures
walk every distinct eligible credential. Anchor blocks to the failed
credential and stop on cycles, aborts, or 64 attempts.
This commit is contained in:
Jeff Scott Ward
2026-07-11 20:34:57 -04:00
parent 20c0a2e410
commit 0ab90f63e5
14 changed files with 1009 additions and 164 deletions
+1
View File
@@ -12,6 +12,7 @@
### Fixed
- Fixed OAuth credential resolution returning "No API key found" when every plan-eligible OpenAI Codex account was rate-limit blocked and the only unblocked account failed the model's plan gate: resolution now runs a last-resort ladder that first yields a plan-fitting account regardless of usage blocks (so callers get real usage-limit retry semantics), then tries every account with the plan filter dropped before reporting no credential
- Fixed provider-agnostic replay-safe usage/account-quota failures to rotate through every distinct eligible credential instead of stopping after the fixed a/b/c ladder, while preserving transient-429 backoff, cycle/abort guards, exact failed-credential targeting despite stale session stickiness, and a finite safety ceiling.
## [16.4.5] - 2026-07-11
+191 -53
View File
@@ -2,12 +2,13 @@ import type { OAuthAccess } from "./auth-storage";
import * as AIError from "./error";
import { isAuthRetryableError } from "./error/auth-classify";
import { isUsageLimit } from "./error/flags";
import { isUsageLimitOutcome } from "./error/rate-limit";
/**
* Context passed to an {@link ApiKeyResolver} on each resolution attempt.
*
* The `error`/`lastChance` pair drives the central a/b/c retry policy shared by
* the streaming ({@link streamSimple}) and non-streaming ({@link withAuth})
* The `error`/`lastChance` pair preserves the legacy a/b/c resolver contract
* shared by streaming ({@link streamSimple}) and non-streaming ({@link withAuth})
* drivers:
* - `error === undefined` → **initial resolve** (no force-refresh; cheap, may
* return a locally-cached not-yet-expired token).
@@ -16,11 +17,13 @@ import { isUsageLimit } from "./error/flags";
* - `error !== undefined && lastChance` → **step (c): switch account**
* (invalidate/usage-limit the current credential and rotate to a sibling).
*
* The resolver returns the bearer to send, or `undefined` to stop retrying and
* surface the last error to the caller.
* Current drivers preserve that bounded a/b/c sequence for ordinary 401/auth
* failures. Usage/account-limit failures skip refresh and may repeat step (c)
* until the resolver returns `undefined`, cycles, or hits
* {@link AUTH_RETRY_MAX_ATTEMPTS}.
*/
export interface ApiKeyResolveContext {
/** True on the final retry step — the resolver should rotate to a sibling credential. */
/** True when the resolver should rotate to a sibling credential. */
lastChance: boolean;
/** The auth error that triggered this re-resolution, or `undefined` on the initial resolve. */
error: unknown;
@@ -77,13 +80,22 @@ export function seedApiKeyResolver(seed: string | undefined, resolver: ApiKeyRes
export { isAuthRetryableError };
/**
* The ordered `lastChance` values for the retry steps after the initial
* attempt fails: `false` → step (b) refresh-same, `true` → step (c) switch.
* Shared by {@link withAuth} and the streaming retry driver so both run the
* same policy.
* Legacy bounded a/b/c retry sequence retained for public compatibility:
* `false` → refresh-same, `true` → rotate/switch. Current drivers consume it
* once for ordinary 401/auth failures; usage/account-limit failures may repeat
* sibling rotation until a termination guard fires.
*/
export const AUTH_RETRY_STEPS: readonly boolean[] = [false, true];
export const AUTH_RETRY_MAX_ATTEMPTS = 64;
function isDirectCredentialRotationError(error: unknown): boolean {
if (isUsageLimit(error)) return true;
const status = AIError.status(error);
const message = error instanceof Error ? error.message : typeof error === "string" ? error : undefined;
return isUsageLimitOutcome(status, message);
}
/** Resolve a single retry step, swallowing resolver failures into `undefined`. */
export async function resolveRetryKey(
resolver: ApiKeyResolver,
@@ -93,22 +105,102 @@ export async function resolveRetryKey(
previousKey?: string,
): Promise<string | undefined> {
try {
const rotateSibling = lastChance || (!lastChance && isUsageLimit(error));
const rotateSibling = lastChance || (!lastChance && isDirectCredentialRotationError(error));
return (await resolver({ lastChance: rotateSibling, error, signal, previousKey })) || undefined;
} catch {
return undefined;
}
}
export interface AuthRetryKeyState {
/** Bearer strings already sent during this logical operation. */
attemptedKeys: Set<string>;
/** Bearer used by the most recent failed attempt. */
lastKey: string;
/** Whether the current credential already consumed its 401 refresh-same retry. */
refreshedCurrent: boolean;
/** Whether the legacy non-usage auth path already switched to one sibling. */
legacyAuthSwitchUsed: boolean;
/** Total outbound attempts accepted for this logical operation, including the initial request. */
attempts: number;
}
export function createAuthRetryKeyState(initialKey: string): AuthRetryKeyState {
return {
attemptedKeys: new Set([initialKey]),
lastKey: initialKey,
refreshedCurrent: false,
legacyAuthSwitchUsed: false,
attempts: 1,
};
}
function acceptRetryKey(state: AuthRetryKeyState, key: string, refreshedCurrent: boolean): string | undefined {
if (state.attemptedKeys.has(key) || state.attempts >= AUTH_RETRY_MAX_ATTEMPTS) return undefined;
state.attemptedKeys.add(key);
state.attempts += 1;
state.lastKey = key;
state.refreshedCurrent = refreshedCurrent;
return key;
}
export async function resolveNextAuthRetryKey(
state: AuthRetryKeyState,
resolver: ApiKeyResolver,
error: unknown,
signal?: AbortSignal,
): Promise<string | undefined> {
if (signal?.aborted) return undefined;
if (state.attempts >= AUTH_RETRY_MAX_ATTEMPTS) return undefined;
const directRotation = isDirectCredentialRotationError(error);
if (!directRotation) {
if (state.legacyAuthSwitchUsed) return undefined;
if (!state.refreshedCurrent) {
const refreshed = await resolveRetryKey(resolver, false, error, signal, state.lastKey);
state.refreshedCurrent = true;
if (signal?.aborted) return undefined;
if (refreshed !== undefined) {
const accepted = acceptRetryKey(state, refreshed, true);
if (accepted !== undefined) return accepted;
}
}
}
if (signal?.aborted) return undefined;
const rotated = await resolveRetryKey(resolver, true, error, signal, state.lastKey);
if (signal?.aborted || rotated === undefined) return undefined;
const accepted = acceptRetryKey(state, rotated, !directRotation);
if (accepted !== undefined && !directRotation) state.legacyAuthSwitchUsed = true;
return accepted;
}
function oauthCredentialIdentity(access: OAuthAccess): string {
return access.credentialId !== undefined ? `credential:${access.credentialId}` : `bearer:${access.accessToken}`;
}
async function runOAuthAttempt<T>(
access: OAuthAccess,
attempt: (access: OAuthAccess) => Promise<T>,
isAuthError: (error: unknown) => boolean,
): Promise<{ ok: true; result: T } | { ok: false; error: unknown }> {
try {
return { ok: true, result: await attempt(access) };
} catch (error) {
if (!isAuthError(error)) throw error;
return { ok: false, error };
}
}
/**
* Runs an auth-protected operation through the central a/b/c retry policy.
*
* - A static string key (or any non-resolver) → a single `attempt` with no
* retry (identical to the legacy static-key path).
* - A resolver → initial `attempt`, then on a retryable auth error up to two
* more attempts (refresh-same, then switch). A step is skipped when the
* resolver returns the same key it just tried or `undefined`; non-auth errors
* propagate immediately.
* - A resolver → initial `attempt`, then resolver-driven retries until the
* applicable policy is exhausted, the resolver declines or cycles, or the
* operation reaches {@link AUTH_RETRY_MAX_ATTEMPTS}. Ordinary 401/auth
* failures retain one refresh-same plus one sibling switch; usage/account
* limits rotate directly through distinct siblings.
*
* Used by non-streaming consumers (image generation, web search, completion
* helpers). The streaming driver in `stream.ts` implements the same policy with
@@ -129,21 +221,21 @@ export async function withAuth<T>(
const resolver = key;
const signal = opts?.signal;
let lastKey = await resolveRetryKey(resolver, false, undefined, signal);
if (lastKey === undefined) throw missingKey();
const initialKey = await resolveRetryKey(resolver, false, undefined, signal);
if (initialKey === undefined) throw missingKey();
const state = createAuthRetryKeyState(initialKey);
let lastError: unknown;
try {
return await attempt(lastKey);
return await attempt(initialKey);
} catch (error) {
if (!isAuthError(error)) throw error;
lastError = error;
}
for (let i = 0; i < AUTH_RETRY_STEPS.length; i++) {
const nextKey = await resolveRetryKey(resolver, AUTH_RETRY_STEPS[i]!, lastError, signal, lastKey);
if (nextKey === undefined || nextKey === lastKey) continue;
lastKey = nextKey;
while (true) {
const nextKey = await resolveNextAuthRetryKey(state, resolver, lastError, signal);
if (nextKey === undefined) break;
try {
return await attempt(nextKey);
} catch (error) {
@@ -169,7 +261,7 @@ export interface OAuthAccessSource {
rotateSessionCredential(
provider: string,
sessionId: string | undefined,
options?: { error?: unknown; signal?: AbortSignal },
options?: { error?: unknown; signal?: AbortSignal; apiKey?: string; credentialId?: number },
): Promise<boolean>;
}
@@ -195,14 +287,18 @@ export interface WithOAuthAccessOptions {
* `projectId`, `enterpriseUrl`) instead of bare API-key bytes.
*
* - initial → `getOAuthAccess` (or `opts.seed`).
* - step (b) → `getOAuthAccess` with `forceRefresh: true` (re-mint the SAME
* account; picks up peer/broker rotations).
* - step (c) → `rotateSessionCredential` then re-resolve (switch to a sibling).
* - 401/auth failure → one `getOAuthAccess` with `forceRefresh: true` for the
* current account, then sibling rotation.
* - usage-limit failure → `rotateSessionCredential` directly, without a
* force-refresh detour.
*
* A step is skipped when it yields no access or the same `accessToken` that
* just failed; non-auth errors propagate immediately. Use this instead of
* hand-rolled `getOAuthAccess` + fetch flows so 401s and usage-limits rotate
* credentials instead of failing the call.
* A refresh-same step may retry a new bearer for the same credential identity;
* sibling rotation stops when it yields a credential identity
* (`credentialId ?? accessToken`) or bearer already attempted in this turn.
* All OAuth attempts share the {@link AUTH_RETRY_MAX_ATTEMPTS} ceiling.
* Non-auth errors propagate immediately. Use this instead of hand-rolled
* `getOAuthAccess` + fetch flows so 401s and usage-limits rotate credentials
* instead of failing the call.
*/
export async function withOAuthAccess<T>(
storage: OAuthAccessSource,
@@ -221,34 +317,76 @@ export async function withOAuthAccess<T>(
);
}
const resolveStep = async (lastChance: boolean, error: unknown): Promise<OAuthAccess | undefined> => {
const attemptedBearers = new Set([lastAccess.accessToken]);
const attemptedCredentialIdentities = new Set([oauthCredentialIdentity(lastAccess)]);
let attemptCount = 1;
let legacyAuthSwitchUsed = false;
let refreshedCurrent = false;
let attemptResult = await runOAuthAttempt(lastAccess, attempt, isAuthError);
if (attemptResult.ok) return attemptResult.result;
let lastError = attemptResult.error;
while (true) {
let next: OAuthAccess | undefined;
if (signal?.aborted || attemptCount >= AUTH_RETRY_MAX_ATTEMPTS) break;
const directRotation = isDirectCredentialRotationError(lastError);
if (!directRotation) {
if (legacyAuthSwitchUsed) break;
if (!refreshedCurrent) {
refreshedCurrent = true;
try {
next = await storage.getOAuthAccess(provider, sessionId, { forceRefresh: true, signal });
} catch {
next = undefined;
}
if (signal?.aborted) break;
if (next) {
const bearer = next.accessToken;
if (!attemptedBearers.has(bearer) && attemptCount < AUTH_RETRY_MAX_ATTEMPTS) {
attemptedCredentialIdentities.add(oauthCredentialIdentity(next));
attemptedBearers.add(bearer);
attemptCount += 1;
lastAccess = next;
attemptResult = await runOAuthAttempt(next, attempt, isAuthError);
if (attemptResult.ok) return attemptResult.result;
lastError = attemptResult.error;
continue;
}
}
}
}
if (signal?.aborted || attemptCount >= AUTH_RETRY_MAX_ATTEMPTS) break;
try {
if (!lastChance) return await storage.getOAuthAccess(provider, sessionId, { forceRefresh: true, signal });
await storage.rotateSessionCredential(provider, sessionId, { error, signal });
return await storage.getOAuthAccess(provider, sessionId, { signal });
const rotated = await storage.rotateSessionCredential(provider, sessionId, {
error: lastError,
signal,
apiKey: lastAccess.accessToken,
credentialId: lastAccess.credentialId,
});
if (!rotated) break;
next = await storage.getOAuthAccess(provider, sessionId, { signal });
} catch {
return undefined;
next = undefined;
}
};
let lastError: unknown;
try {
return await attempt(lastAccess);
} catch (error) {
if (!isAuthError(error)) throw error;
lastError = error;
}
for (const lastChance of AUTH_RETRY_STEPS) {
const next = await resolveStep(lastChance, lastError);
if (!next || next.accessToken === lastAccess.accessToken) continue;
if (signal?.aborted || !next) break;
const credentialIdentity = oauthCredentialIdentity(next);
if (
attemptedCredentialIdentities.has(credentialIdentity) ||
attemptedBearers.has(next.accessToken) ||
attemptCount >= AUTH_RETRY_MAX_ATTEMPTS
) {
break;
}
attemptedCredentialIdentities.add(credentialIdentity);
attemptedBearers.add(next.accessToken);
attemptCount += 1;
lastAccess = next;
try {
return await attempt(next);
} catch (error) {
if (!isAuthError(error)) throw error;
lastError = error;
}
refreshedCurrent = !directRotation;
if (!directRotation) legacyAuthSwitchUsed = true;
attemptResult = await runOAuthAttempt(next, attempt, isAuthError);
if (attemptResult.ok) return attemptResult.result;
lastError = attemptResult.error;
}
throw lastError;
+82 -28
View File
@@ -682,7 +682,7 @@ type AuthApiKeyOptions = {
*/
forceRefresh?: boolean;
};
type OAuthResolutionResult = { apiKey: string; credential: OAuthCredential };
type OAuthResolutionResult = { apiKey: string; credential: OAuthCredential; credentialId?: number };
/**
* Refreshed OAuth access plus identity metadata returned by
@@ -3401,6 +3401,39 @@ export class AuthStorage {
return results;
}
async #resolveCredentialTarget(
provider: string,
sessionId: string | undefined,
options?: { credentialId?: number; apiKey?: string },
): Promise<{ type: AuthCredential["type"]; index: number; explicit: boolean } | undefined> {
const explicit = options?.credentialId !== undefined || options?.apiKey !== undefined;
if (explicit) {
const latestRows = this.#store.listAuthCredentials(provider);
this.#setStoredCredentials(
provider,
latestRows.map(row => ({ id: row.id, credential: row.credential })),
);
}
if (options?.credentialId !== undefined) {
const stored = this.#getStoredCredentials(provider);
const index = stored.findIndex(entry => entry.id === options.credentialId);
const entry = index === -1 ? undefined : stored[index];
if (entry) return { type: entry.credential.type, index, explicit: true };
}
if (options?.apiKey !== undefined) {
const stored = this.#getStoredCredentials(provider);
for (let index = 0; index < stored.length; index++) {
const entry = stored[index];
if (entry && (await this.#credentialMatchesApiKey(entry.credential, options.apiKey))) {
return { type: entry.credential.type, index, explicit: true };
}
}
}
if (explicit) return undefined;
const sessionCredential = this.#getSessionCredential(provider, sessionId);
return sessionCredential ? { ...sessionCredential, explicit: false } : undefined;
}
/**
* Marks the current session's credential as temporarily blocked due to usage limits.
* Uses usage reports to determine accurate reset time when available.
@@ -3411,20 +3444,19 @@ export class AuthStorage {
async markUsageLimitReached(
provider: string,
sessionId: string | undefined,
options?: { retryAfterMs?: number; baseUrl?: string; modelId?: string; apiKey?: string; signal?: AbortSignal },
options?: {
retryAfterMs?: number;
baseUrl?: string;
modelId?: string;
apiKey?: string;
credentialId?: number;
signal?: AbortSignal;
},
): Promise<UsageLimitMarkResult> {
let sessionCredential: { type: AuthCredential["type"]; index: number } | undefined;
if (options?.apiKey) {
const stored = this.#getStoredCredentials(provider);
for (let index = 0; index < stored.length; index++) {
const entry = stored[index];
if (entry && (await this.#credentialMatchesApiKey(entry.credential, options.apiKey))) {
sessionCredential = { type: entry.credential.type, index };
break;
}
}
}
sessionCredential ??= this.#getSessionCredential(provider, sessionId);
const sessionCredential = await this.#resolveCredentialTarget(provider, sessionId, {
credentialId: options?.credentialId,
apiKey: options?.apiKey,
});
if (!sessionCredential) return { switched: false };
const providerKey = this.#getProviderTypeKey(provider, sessionCredential.type);
@@ -4209,7 +4241,7 @@ export class AuthStorage {
}
}
this.#recordSessionCredential(provider, sessionId, "oauth", selection.index);
return { apiKey: result.apiKey, credential: updated };
return { apiKey: result.apiKey, credential: updated, credentialId };
} catch (error) {
const errorMsg = String(error);
// Only remove credentials for definitive auth failures
@@ -4434,9 +4466,10 @@ export class AuthStorage {
}
const resolved = await this.#resolveOAuthSelection(provider, sessionId, options);
if (!resolved) return undefined;
const { credential } = resolved;
const { credential, credentialId } = resolved;
return {
accessToken: credential.access,
credentialId,
accountId: credential.accountId,
email: credential.email,
projectId: credential.projectId,
@@ -4922,36 +4955,45 @@ export class AuthStorage {
}
/**
* Rotate away from the session's current credential after a retryable auth
* error — step (c) of the auth-retry policy. Stateless: looks up the
* session-sticky credential (no API-key matching needed), applies the
* storage action for the error class, then clears the sticky so the next
* {@link AuthStorage.getApiKey} for this session picks a sibling.
* Rotate away from the credential that failed after a retryable auth error —
* step (c) of the auth-retry policy. Prefer the failed stored row id supplied
* in `options.credentialId`, then the failed bearer supplied in
* `options.apiKey`, so overlapping requests cannot redirect rotation through
* stale session stickiness. Fall back to the session-sticky credential only
* when neither explicit target is available. If an explicit target is supplied
* but no longer matches storage, return `false` without mutating sticky state.
* Apply the storage action for the error class, then let the next resolve
* select a sibling.
*
* - usage-limit / account-rate-limit error → {@link AuthStorage.markUsageLimitReached}
* (temporary block via its own backoff — default plus server usage-report
* reset; sticky left intact so the next resolve re-ranks around the block).
* - otherwise (hard 401 / auth failure) → mark the credential suspect (or
* reload when no broker hook is wired) and block it, then drop the sticky.
* reload when no broker hook is wired) and block it, then drop matching
* sticky state.
*
* Returns whether another usable credential of the same type remains.
*/
async rotateSessionCredential(
provider: string,
sessionId: string | undefined,
options?: { error?: unknown; modelId?: string; apiKey?: string; signal?: AbortSignal },
options?: { error?: unknown; modelId?: string; apiKey?: string; credentialId?: number; signal?: AbortSignal },
): Promise<boolean> {
const sessionCredential = this.#getSessionCredential(provider, sessionId);
const sessionCredential = await this.#resolveCredentialTarget(provider, sessionId, {
credentialId: options?.credentialId,
apiKey: options?.apiKey,
});
if (!sessionCredential) return false;
const error = options?.error;
const status = AIError.status(error);
const message = error instanceof Error ? error.message : typeof error === "string" ? error : undefined;
if (isUsageLimitOutcome(status, message)) {
if (AIError.isUsageLimit(error) || isUsageLimitOutcome(status, message)) {
return (
await this.markUsageLimitReached(provider, sessionId, {
modelId: options?.modelId,
apiKey: options?.apiKey,
credentialId: options?.credentialId,
signal: options?.signal,
})
).switched;
@@ -4967,7 +5009,13 @@ export class AuthStorage {
!this.#isCredentialBlocked(provider, providerKey, index),
);
const target = this.#getStoredCredentials(provider)[sessionCredential.index];
this.#clearSessionCredential(provider, sessionId);
const sticky = this.#getSessionCredential(provider, sessionId);
if (
!sessionCredential.explicit ||
(sticky?.type === sessionCredential.type && sticky.index === sessionCredential.index)
) {
this.#clearSessionCredential(provider, sessionId);
}
this.#markCredentialBlocked(
provider,
providerKey,
@@ -5005,12 +5053,18 @@ export class AuthStorage {
*/
resolver(provider: string, options?: { sessionId?: string; baseUrl?: string; modelId?: string }): ApiKeyResolver {
const { sessionId, baseUrl, modelId } = options ?? {};
return async ({ lastChance, error, signal }) => {
return async ({ lastChance, error, signal, previousKey }) => {
if (error === undefined) {
return this.getApiKey(provider, sessionId, { baseUrl, modelId, signal });
}
if (lastChance) {
await this.rotateSessionCredential(provider, sessionId, { error, modelId, signal });
const rotated = await this.rotateSessionCredential(provider, sessionId, {
error,
modelId,
signal,
apiKey: previousKey,
});
if (!rotated) return undefined;
return this.getApiKey(provider, sessionId, { baseUrl, modelId, signal });
}
return this.getApiKey(provider, sessionId, { baseUrl, modelId, forceRefresh: true, signal });
+2 -1
View File
@@ -1,5 +1,5 @@
import { extractHttpStatusFromError } from "@oh-my-pi/pi-utils";
import { isOAuthExpiry } from "./flags";
import { isOAuthExpiry, isUsageLimit } from "./flags";
import { isUsageLimitOutcome } from "./rate-limit";
/**
@@ -21,6 +21,7 @@ export function isDefinitiveOAuthFailure(errorMsg: string): boolean {
* upstream-backoff lane.
*/
export function isAuthRetryableError(error: unknown): boolean {
if (isUsageLimit(error)) return true;
const httpStatus = extractHttpStatusFromError(error);
if (httpStatus === 401) return true;
const message = error instanceof Error ? error.message : typeof error === "string" ? error : undefined;
+19 -30
View File
@@ -17,7 +17,7 @@ import { CATALOG_PROVIDERS, type ProviderCatalogEntry } from "@oh-my-pi/pi-catal
import { CODEX_BASE_URL } from "@oh-my-pi/pi-catalog/wire/codex";
import { $env, $pickenv, getConfigRootDir, isEnoent, logger, withExtraCaFetch } from "@oh-my-pi/pi-utils";
import { getCustomApi } from "./api-registry";
import { AUTH_RETRY_STEPS, isApiKeyResolver, resolveRetryKey } from "./auth-retry";
import { createAuthRetryKeyState, isApiKeyResolver, resolveNextAuthRetryKey } from "./auth-retry";
import * as AIError from "./error";
import { ProviderHttpError } from "./error";
import { isUsageLimitOutcome } from "./error/rate-limit";
@@ -978,17 +978,19 @@ function isRetryableUpstreamError(error: unknown, status: number | undefined, me
// per-minute caps) classify as RATE_LIMIT_EXCEEDED in
// `parseRateLimitReason` and stay in the provider's own backoff layer
// instead of burning siblings.
if (AIError.isUsageLimit(error)) return true;
if (status === 401) return true;
void error;
return isUsageLimitOutcome(status, message);
}
function createAssistantAuthError(message: AssistantMessage): Error {
const text = message.errorMessage ?? "Provider authentication failed";
const status = extractStatusFromAssistantError(message);
return status === undefined
? new AIError.ProviderResponseError(text, { kind: "runtime" })
: new ProviderHttpError(text, status);
const error =
status === undefined
? new AIError.ProviderResponseError(text, { kind: "runtime" })
: new ProviderHttpError(text, status);
return typeof message.errorId === "number" ? AIError.attach(error, message.errorId) : error;
}
function emitBufferedEvents(stream: AssistantMessageEventStream, events: AssistantMessageEvent[]): void {
@@ -1012,12 +1014,12 @@ export function streamSimple<TApi extends Api>(
if (apiKeyResolver) {
const outer = new AssistantMessageEventStream();
const signal = requestOptions?.signal;
// One inner attempt against a resolved string key. When
// `captureAuthFailure` is set, a retryable auth error that arrives before
// any replay-unsafe event is buffered and returned (so the caller can
// retry with a fresh key) instead of surfaced. The terminal attempt
// clears the flag and emits whatever it gets.
const runAttempt = async (apiKey: string, captureAuthFailure: boolean): Promise<AuthRetryFailure | undefined> => {
// One inner attempt against a resolved string key. A retryable auth error
// that arrives before any replay-unsafe event is buffered and returned
// (so the caller can retry with a fresh key) instead of surfaced. Once any
// non-start event escapes, retry is no longer safe and the failure is
// emitted directly.
const runAttempt = async (apiKey: string): Promise<AuthRetryFailure | undefined> => {
const bufferedEvents: AssistantMessageEvent[] = [];
let emittedReplayUnsafeEvent = false;
const flushBuffered = (): void => {
@@ -1034,7 +1036,6 @@ export function streamSimple<TApi extends Api>(
}
if (
!emittedReplayUnsafeEvent &&
captureAuthFailure &&
event.type === "error" &&
isRetryableUpstreamError(
event.error,
@@ -1054,7 +1055,6 @@ export function streamSimple<TApi extends Api>(
} catch (error) {
if (
!emittedReplayUnsafeEvent &&
captureAuthFailure &&
isRetryableUpstreamError(
error,
AIError.status(error),
@@ -1096,27 +1096,16 @@ export function streamSimple<TApi extends Api>(
outer.fail(new AIError.MissingApiKeyError(model.provider));
return;
}
let failure = await runAttempt(lastKey, true);
const retryState = createAuthRetryKeyState(lastKey);
let failure = await runAttempt(lastKey);
if (!failure) return;
// a/b/c policy: refresh the same account (lastChance=false), then
// switch to a sibling (lastChance=true). A step is skipped when the
// resolver yields the same key it just tried or `undefined`; the
// final step's attempt clears the capture flag so it emits directly.
for (let step = 0; step < AUTH_RETRY_STEPS.length; step++) {
while (true) {
// Caller aborted between attempts: don't mint a fresh token or fire
// another doomed request — emit the captured failure instead.
if (signal?.aborted) break;
const nextKey = await resolveRetryKey(
apiKeyResolver,
AUTH_RETRY_STEPS[step]!,
failure.error,
signal,
lastKey,
);
if (nextKey === undefined || nextKey === lastKey) continue;
lastKey = nextKey;
const isLastStep = step === AUTH_RETRY_STEPS.length - 1;
const next = await runAttempt(nextKey, !isLastStep);
const nextKey = await resolveNextAuthRetryKey(retryState, apiKeyResolver, failure.error, signal);
if (nextKey === undefined) break;
const next = await runAttempt(nextKey);
if (!next) return;
failure = next;
}
+299 -10
View File
@@ -1,6 +1,14 @@
import { describe, expect, it } from "bun:test";
import type { ApiKeyResolveContext, OAuthAccess, OAuthAccessSource } from "@oh-my-pi/pi-ai";
import { isApiKeyResolver, isAuthRetryableError, resolveApiKeyOnce, withAuth, withOAuthAccess } from "@oh-my-pi/pi-ai";
import {
AUTH_RETRY_MAX_ATTEMPTS,
isApiKeyResolver,
isAuthRetryableError,
resolveApiKeyOnce,
withAuth,
withOAuthAccess,
} from "@oh-my-pi/pi-ai";
import { ProviderHttpError } from "@oh-my-pi/pi-ai/error";
function authError(status = 401): Error & { status: number } {
return Object.assign(new Error(`${status} authentication_error`), { status });
@@ -12,6 +20,10 @@ function usageLimitError(): Error & { status: number } {
});
}
function opaque429Error(): Error & { status: number } {
return Object.assign(new Error(""), { status: 429 });
}
describe("isApiKeyResolver / resolveApiKeyOnce", () => {
it("narrows resolver vs static key and resolves the initial value", async () => {
expect(isApiKeyResolver("static")).toBe(false);
@@ -36,6 +48,12 @@ describe("isAuthRetryableError", () => {
it("treats 401 and usage-limit phrasing as retryable, everything else as not", () => {
expect(isAuthRetryableError(authError(401))).toBe(true);
expect(isAuthRetryableError(usageLimitError())).toBe(true);
expect(
isAuthRetryableError(new ProviderHttpError("Generic provider failure", 429, { code: "insufficient_quota" })),
).toBe(true);
expect(
isAuthRetryableError(new ProviderHttpError("Generic provider failure", 429, { code: "rate_limit_error" })),
).toBe(false);
// A 429 whose body names the *account's* rate limit is rotatable (switch
// account), even though it isn't a 401 and isn't phrased "usage limit".
expect(
@@ -111,6 +129,58 @@ describe("withAuth", () => {
]);
});
it("does not exhaust every sibling on pure 401 auth failures", async () => {
const keys: string[] = [];
const contexts: ApiKeyResolveContext[] = [];
const pool = ["k0", "k1", "k2", "k3"];
let resolveIndex = 0;
let lastError: unknown;
let caught: unknown;
try {
await withAuth(
ctx => {
contexts.push(ctx);
return ctx.error === undefined ? pool[0] : pool[++resolveIndex];
},
async key => {
keys.push(key);
lastError = authError();
throw lastError;
},
);
} catch (error) {
caught = error;
}
expect(caught).toBe(lastError);
expect(keys).toEqual(["k0", "k1", "k2"]);
expect(contexts.map(ctx => ctx.lastChance)).toEqual([false, false, true]);
});
it("continues quota rotation when a refreshed 401 retry becomes a usage limit", async () => {
const keys: string[] = [];
const contexts: ApiKeyResolveContext[] = [];
const pool = ["k0", "k1", "k2", "k3"];
let resolveIndex = 0;
const result = await withAuth(
ctx => {
contexts.push(ctx);
return ctx.error === undefined ? pool[0] : pool[++resolveIndex];
},
async key => {
keys.push(key);
if (key === "k3") return "success";
if (key === "k0") throw authError();
throw usageLimitError();
},
);
expect(result).toBe("success");
expect(keys).toEqual(pool);
expect(contexts.map(ctx => ctx.lastChance)).toEqual([false, false, true, true]);
});
it("switches accounts before refreshing the same account on usage limits", async () => {
const keys: string[] = [];
const contexts: ApiKeyResolveContext[] = [];
@@ -133,6 +203,115 @@ describe("withAuth", () => {
]);
});
it("switches accounts before refreshing on opaque 429 usage outcomes", async () => {
const keys: string[] = [];
const contexts: ApiKeyResolveContext[] = [];
const result = await withAuth(
ctx => {
contexts.push(ctx);
return ctx.error === undefined ? "k0" : ctx.lastChance ? "k2" : "k1";
},
async key => {
keys.push(key);
if (key === "k2") return "success";
throw opaque429Error();
},
);
expect(result).toBe("success");
expect(keys).toEqual(["k0", "k2"]);
expect(contexts.map(ctx => ({ lastChance: ctx.lastChance, hasError: ctx.error !== undefined }))).toEqual([
{ lastChance: false, hasError: false },
{ lastChance: true, hasError: true },
]);
});
it("rotates through every distinct sibling after consecutive usage limits", async () => {
const keys: string[] = [];
const contexts: ApiKeyResolveContext[] = [];
const pool = ["k0", "k1", "k2", "k3"];
let nextSibling = 0;
const result = await withAuth(
ctx => {
contexts.push(ctx);
return ctx.error === undefined ? pool[0] : pool[++nextSibling];
},
async key => {
keys.push(key);
if (key === "k3") return "success";
throw usageLimitError();
},
);
expect(result).toBe("success");
expect(keys).toEqual(pool);
expect(contexts.map(ctx => ctx.lastChance)).toEqual([false, true, true, true]);
});
it("stops usage-limit rotation before retrying an already-attempted credential", async () => {
const keys: string[] = [];
const errors = [usageLimitError(), usageLimitError()];
const resolved = ["k0", "k1", "k0"];
let resolveIndex = 0;
let attemptIndex = 0;
await expect(
withAuth(
() => resolved[resolveIndex++],
async key => {
keys.push(key);
throw errors[Math.min(attemptIndex++, errors.length - 1)]!;
},
),
).rejects.toBe(errors[1]);
expect(keys).toEqual(["k0", "k1"]);
});
it("caps endlessly unique resolver retries", async () => {
const keys: string[] = [];
let resolveIndex = 0;
let lastError: unknown;
let caught: unknown;
try {
await withAuth(
() => `k${resolveIndex++}`,
async key => {
keys.push(key);
lastError = usageLimitError();
throw lastError;
},
);
} catch (error) {
caught = error;
}
expect(caught).toBe(lastError);
expect(keys).toHaveLength(AUTH_RETRY_MAX_ATTEMPTS);
expect(resolveIndex).toBe(AUTH_RETRY_MAX_ATTEMPTS);
});
it("does not attempt a retry key resolved after abort", async () => {
const controller = new AbortController();
const keys: string[] = [];
const original = usageLimitError();
await expect(
withAuth(
ctx => {
if (ctx.error === undefined) return "k0";
controller.abort();
return "k1";
},
async key => {
keys.push(key);
throw original;
},
{ signal: controller.signal },
),
).rejects.toBe(original);
expect(keys).toEqual(["k0"]);
});
it("stops retrying when the resolver returns undefined", async () => {
const keys: string[] = [];
const original = authError();
@@ -255,10 +434,67 @@ describe("withOAuthAccess", () => {
expect(storage.calls).toEqual([{ forceRefresh: undefined }, { forceRefresh: true }]);
});
it("skips an unchanged force-refresh token and rotates to a sibling", async () => {
it("tries a refreshed bearer for the same credential id on 401 before rotating", async () => {
const storage = fakeStorage({
initial: access("stale", { credentialId: 7 }),
forced: access("fresh", { credentialId: 7 }),
rotated: access("sibling", { credentialId: 8 }),
});
const attempts: string[] = [];
const result = await withOAuthAccess(storage, "prov", async a => {
attempts.push(a.accessToken);
if (a.accessToken === "stale") throw authError();
return "ok";
});
expect(result).toBe("ok");
expect(attempts).toEqual(["stale", "fresh"]);
expect(storage.calls).toEqual([{ forceRefresh: undefined }, { forceRefresh: true }]);
});
it("does not exhaust every OAuth sibling on pure 401 auth failures", async () => {
const attempts: string[] = [];
const calls: Array<{ forceRefresh: boolean | undefined } | "rotate"> = [];
const rotated = [access("sibling-1", { credentialId: 2 }), access("sibling-2", { credentialId: 3 })];
let rotateIndex = 0;
let lastError: unknown;
let caught: unknown;
const storage: OAuthAccessSource = {
async getOAuthAccess(_provider, _sessionId, options) {
calls.push({ forceRefresh: options?.forceRefresh });
if (options?.forceRefresh) return access("fresh", { credentialId: 1 });
if (rotateIndex > 0) return rotated[rotateIndex - 1];
return access("stale", { credentialId: 1 });
},
async rotateSessionCredential() {
calls.push("rotate");
rotateIndex += 1;
return true;
},
};
try {
await withOAuthAccess(storage, "prov", async a => {
attempts.push(a.accessToken);
lastError = authError();
throw lastError;
});
} catch (error) {
caught = error;
}
expect(caught).toBe(lastError);
expect(attempts).toEqual(["stale", "fresh", "sibling-1"]);
expect(calls).toEqual([
{ forceRefresh: undefined },
{ forceRefresh: true },
"rotate",
{ forceRefresh: undefined },
]);
});
it("rotates directly to a sibling on usage limits", async () => {
const storage = fakeStorage({
initial: access("dead"),
forced: access("dead"),
rotated: access("sibling"),
});
const attempts: string[] = [];
@@ -268,14 +504,67 @@ describe("withOAuthAccess", () => {
return "ok";
});
expect(result).toBe("ok");
// "dead" must not be re-attempted after the no-op force refresh.
// Usage-limit failures burn/rotate the exhausted account directly; a
// force-refresh of the same account would duplicate the failed side effect.
expect(attempts).toEqual(["dead", "sibling"]);
expect(storage.calls).toEqual([
{ forceRefresh: undefined },
{ forceRefresh: true },
"rotate",
{ forceRefresh: undefined },
]);
expect(storage.calls).toEqual([{ forceRefresh: undefined }, "rotate", { forceRefresh: undefined }]);
});
it("passes the failed OAuth bearer to rotation", async () => {
const rotationTargets: Array<{ apiKey: string | undefined; credentialId: number | undefined }> = [];
const storage: OAuthAccessSource = {
async getOAuthAccess() {
return rotationTargets.length === 0
? access("dead", { credentialId: 17 })
: access("sibling", { credentialId: 18 });
},
async rotateSessionCredential(_provider, _sessionId, options) {
rotationTargets.push({ apiKey: options?.apiKey, credentialId: options?.credentialId });
return true;
},
};
const attempts: string[] = [];
const result = await withOAuthAccess(storage, "prov", async a => {
attempts.push(a.accessToken);
if (a.accessToken === "dead") throw usageLimitError();
return "ok";
});
expect(result).toBe("ok");
expect(attempts).toEqual(["dead", "sibling"]);
expect(rotationTargets).toEqual([{ apiKey: "dead", credentialId: 17 }]);
});
it("caps endlessly unique OAuth rotation attempts", async () => {
const attempts: string[] = [];
let nextCredential = 0;
let rotateCalls = 0;
let lastError: unknown;
let caught: unknown;
const storage: OAuthAccessSource = {
async getOAuthAccess() {
const credentialId = nextCredential++;
return access(`token-${credentialId}`, { credentialId });
},
async rotateSessionCredential() {
rotateCalls += 1;
return true;
},
};
try {
await withOAuthAccess(storage, "prov", async a => {
attempts.push(a.accessToken);
lastError = usageLimitError();
throw lastError;
});
} catch (error) {
caught = error;
}
expect(caught).toBe(lastError);
expect(attempts).toHaveLength(AUTH_RETRY_MAX_ATTEMPTS);
expect(rotateCalls).toBe(AUTH_RETRY_MAX_ATTEMPTS - 1);
});
it("propagates non-auth errors immediately and surfaces the last auth error when exhausted", async () => {
@@ -70,6 +70,7 @@ describe("AuthStorage broker sentinel refresh", () => {
expect(access).toEqual({
accessToken: "broker-access-rotated",
credentialId: expect.any(Number),
accountId: "broker-account",
email: "broker@example.com",
projectId: "broker-project",
@@ -3,6 +3,7 @@ import * as fs from "node:fs/promises";
import * as os from "node:os";
import * as path from "node:path";
import { type AuthCredentialStore, AuthStorage, SqliteAuthCredentialStore } from "@oh-my-pi/pi-ai/auth-storage";
import { ProviderHttpError } from "@oh-my-pi/pi-ai/error";
import { registerOAuthProvider, unregisterOAuthProviders } from "@oh-my-pi/pi-ai/registry/oauth";
import { removeWithRetries } from "../../utils/src/temp";
@@ -100,6 +101,28 @@ describe("AuthStorage forceRefresh + rotateSessionCredential", () => {
expect(after).toBe("minted-access");
});
test("getOAuthAccess includes a stable credentialId across cached and forced refresh resolves", async () => {
if (!authStorage) throw new Error("test setup failed");
registerProvider();
await authStorage.set(PROVIDER, [
{ type: "oauth", access: "cached-access", refresh: "cached-refresh", expires: farExpiry() },
]);
const cached = await authStorage.getOAuthAccess(PROVIDER, "oauth-identity");
expect(cached?.accessToken).toBe("cached-access");
expect(typeof cached?.credentialId).toBe("number");
const credentialId = cached?.credentialId;
if (credentialId === undefined) throw new Error("expected OAuth credential id");
const forced = await authStorage.getOAuthAccess(PROVIDER, "oauth-identity", { forceRefresh: true });
expect(forced?.accessToken).toBe("minted-access");
expect(forced?.credentialId).toBe(credentialId);
const after = await authStorage.getOAuthAccess(PROVIDER, "oauth-identity");
expect(after?.accessToken).toBe("minted-access");
expect(after?.credentialId).toBe(credentialId);
});
test("rotateSessionCredential(401) blocks + clears the sticky and rotates to a sibling", async () => {
if (!authStorage) throw new Error("test setup failed");
registerProvider();
@@ -123,6 +146,117 @@ describe("AuthStorage forceRefresh + rotateSessionCredential", () => {
expect(second).not.toBe(first);
});
test("resolver rotates the credential matching previousKey instead of a stale sticky", async () => {
if (!authStorage) throw new Error("test setup failed");
await authStorage.set(PROVIDER, [
{ type: "api_key", key: "sticky-key" },
{ type: "api_key", key: "failed-key" },
{ type: "api_key", key: "survivor-key" },
]);
const sessionId = "resolver-previous-key";
const sticky = await authStorage.getApiKey(PROVIDER, sessionId);
if (!sticky) throw new Error("expected initial sticky credential");
const failed = sticky === "failed-key" ? "sticky-key" : "failed-key";
const resolver = authStorage.resolver(PROVIDER, { sessionId });
const retry = await resolver({
lastChance: true,
error: authError(),
previousKey: failed,
});
expect(retry).toBe(sticky);
expect(retry).not.toBe(failed);
const laterSelections = new Set<string>();
for (let index = 0; index < 6; index += 1) {
const selected = await authStorage.getApiKey(PROVIDER);
if (selected) laterSelections.add(selected);
}
expect(laterSelections.has(failed)).toBe(false);
expect(laterSelections.has(sticky)).toBe(true);
});
test("explicit missing rotation targets do not fall back to stale stickiness", async () => {
if (!authStorage || !store) throw new Error("test setup failed");
await authStorage.set(PROVIDER, [
{ type: "api_key", key: "acc-A" },
{ type: "api_key", key: "acc-B" },
{ type: "api_key", key: "acc-C" },
]);
const sessionId = "explicit-missing-target";
const sticky = await authStorage.getApiKey(PROVIDER, sessionId);
if (!sticky) throw new Error("expected sticky credential");
const maxCredentialId = Math.max(...store.listAuthCredentials(PROVIDER).map(row => row.id));
const missingCredentialId = maxCredentialId + 1000;
const rotated = await authStorage.rotateSessionCredential(PROVIDER, sessionId, {
error: authError(),
apiKey: "missing-or-changed-failed-bearer",
});
expect(rotated).toBe(false);
expect(await authStorage.getApiKey(PROVIDER, sessionId)).toBe(sticky);
const rotatedByMissingId = await authStorage.rotateSessionCredential(PROVIDER, sessionId, {
error: authError(),
credentialId: missingCredentialId,
});
expect(rotatedByMissingId).toBe(false);
expect(await authStorage.getApiKey(PROVIDER, sessionId)).toBe(sticky);
const marked = await authStorage.markUsageLimitReached(PROVIDER, sessionId, {
apiKey: "missing-or-changed-failed-bearer",
});
expect(marked.switched).toBe(false);
expect(await authStorage.getApiKey(PROVIDER, sessionId)).toBe(sticky);
const markedByMissingId = await authStorage.markUsageLimitReached(PROVIDER, sessionId, {
credentialId: missingCredentialId,
});
expect(markedByMissingId.switched).toBe(false);
expect(await authStorage.getApiKey(PROVIDER, sessionId)).toBe(sticky);
});
test("credentialId rotation targets the failed row after bearer changes without clearing stale sticky", async () => {
if (!authStorage || !store) throw new Error("test setup failed");
await authStorage.set(PROVIDER, [
{ type: "api_key", key: "acc-A" },
{ type: "api_key", key: "acc-B" },
{ type: "api_key", key: "acc-C" },
]);
const sessionId = "credential-id-target";
const sticky = await authStorage.getApiKey(PROVIDER, sessionId);
if (!sticky) throw new Error("expected sticky credential");
const targetRow = store.listAuthCredentials(PROVIDER).find(row => {
const credential = row.credential;
return credential.type === "api_key" && credential.key !== sticky;
});
if (targetRow?.credential.type !== "api_key") throw new Error("expected non-sticky target row");
const oldKey = targetRow.credential.key;
const changedKey = `${oldKey}-rotated`;
store.updateAuthCredential(targetRow.id, { type: "api_key", key: changedKey });
await authStorage.reload();
const rotated = await authStorage.rotateSessionCredential(PROVIDER, sessionId, {
error: authError(),
apiKey: oldKey,
credentialId: targetRow.id,
});
expect(rotated).toBe(true);
expect(await authStorage.getApiKey(PROVIDER, sessionId)).toBe(sticky);
const laterSelections = new Set<string>();
for (let index = 0; index < 6; index += 1) {
const selected = await authStorage.getApiKey(PROVIDER);
if (selected) laterSelections.add(selected);
}
expect(laterSelections.has(changedKey)).toBe(false);
expect(laterSelections.has(sticky)).toBe(true);
});
test("rotateSessionCredential(usage-limit) delegates to markUsageLimitReached", async () => {
if (!authStorage) throw new Error("test setup failed");
registerProvider();
@@ -150,6 +284,27 @@ describe("AuthStorage forceRefresh + rotateSessionCredential", () => {
expect(second).not.toBe(first);
});
test("rotateSessionCredential treats structured usage codes as quota blocks despite generic messages", async () => {
if (!authStorage) throw new Error("test setup failed");
registerProvider();
await authStorage.set(PROVIDER, [
{ type: "oauth", access: "acc-A", refresh: "ref-A", expires: farExpiry() },
{ type: "oauth", access: "acc-B", refresh: "ref-B", expires: farExpiry() },
]);
const first = await authStorage.getApiKey(PROVIDER, "machine-code-quota");
const usageLimitSpy = vi.spyOn(authStorage, "markUsageLimitReached");
const rotated = await authStorage.rotateSessionCredential(PROVIDER, "machine-code-quota", {
error: new ProviderHttpError("Generic provider failure", 401, { code: "insufficient_quota" }),
});
expect(rotated).toBe(true);
expect(usageLimitSpy).toHaveBeenCalledTimes(1);
expect(usageLimitSpy.mock.calls[0]?.[0]).toBe(PROVIDER);
expect(usageLimitSpy.mock.calls[0]?.[1]).toBe("machine-code-quota");
expect(await authStorage.getApiKey(PROVIDER, "machine-code-quota")).not.toBe(first);
});
test("rotateSessionCredential(xAI credits 403) blocks the exhausted account and rotates", async () => {
if (!authStorage) throw new Error("test setup failed");
registerProvider();
+10
View File
@@ -1,4 +1,5 @@
import { describe, expect, it } from "bun:test";
import { ProviderHttpError } from "@oh-my-pi/pi-ai/error";
import { isUsageLimit } from "@oh-my-pi/pi-ai/error/flags";
import {
calculateRateLimitBackoffMs,
@@ -149,6 +150,15 @@ describe("isUsageLimit", () => {
expect(isUsageLimitStatus(429)).toBe(true);
expect(isUsageLimitStatus(400)).toBe(false);
});
it("detects structured provider usage codes without quota wording", () => {
expect(isUsageLimit(new ProviderHttpError("Generic provider failure", 429, { code: "insufficient_quota" }))).toBe(
true,
);
expect(isUsageLimit(new ProviderHttpError("Generic provider failure", 429, { code: "rate_limit_error" }))).toBe(
false,
);
});
});
describe("isUsageLimitOutcome", () => {
+122 -2
View File
@@ -1,6 +1,8 @@
import { afterEach, describe, expect, it } from "bun:test";
import type { ApiKeyResolveContext } from "@oh-my-pi/pi-ai";
import { registerCustomApi, unregisterCustomApis } from "@oh-my-pi/pi-ai";
import { ProviderHttpError } from "@oh-my-pi/pi-ai/error";
import { classify } from "@oh-my-pi/pi-ai/error/flags";
import { streamSimple } from "@oh-my-pi/pi-ai/stream";
import type { Api, AssistantMessage, Context, Model, SimpleStreamOptions, Usage } from "@oh-my-pi/pi-ai/types";
import { AssistantMessageEventStream } from "@oh-my-pi/pi-ai/utils/event-stream";
@@ -32,8 +34,8 @@ function assistant(content: string[] = []): AssistantMessage {
};
}
function assistantError(errorMessage: string, errorStatus?: number): AssistantMessage {
return { ...assistant(), stopReason: "error", errorMessage, errorStatus };
function assistantError(errorMessage: string, errorStatus?: number, errorId?: number): AssistantMessage {
return { ...assistant(), stopReason: "error", errorMessage, errorStatus, errorId };
}
function authError(): Error & { status: number } {
@@ -357,6 +359,124 @@ describe("streamSimple resolver auth retry", () => {
expect(keys).toEqual(["old-key", "new-key"]);
});
it("rotates on a machine-code-only usage error event before content", async () => {
const keys: unknown[] = [];
const contexts: ApiKeyResolveContext[] = [];
const errorId = classify(new ProviderHttpError("Generic provider failure", 429, { code: "insufficient_quota" }));
registerCustomApi(
API,
(_model: Model<Api>, _context: Context, options?: SimpleStreamOptions) => {
pushKey(keys, options);
const stream = new AssistantMessageEventStream();
queueMicrotask(() => {
if (keys.length === 1) {
stream.push({ type: "start", partial: assistant() });
stream.push({
type: "error",
reason: "error",
error: assistantError("Generic provider failure", 429, errorId),
});
return;
}
ok(stream);
});
return stream;
},
SOURCE_ID,
);
const stream = streamSimple(model(), context, {
apiKey: async ctx => {
contexts.push(ctx);
return ctx.error === undefined ? "old-key" : "new-key";
},
});
for await (const _event of stream) {
// drain
}
expect((await stream.result()).content).toEqual([{ type: "text", text: "ok" }]);
expect(keys).toEqual(["old-key", "new-key"]);
expect(contexts.map(ctx => ctx.lastChance)).toEqual([false, true]);
});
it("rotates through every distinct sibling while usage failures remain replay-safe", async () => {
const keys: unknown[] = [];
const eventTypes: string[] = [];
const contexts: ApiKeyResolveContext[] = [];
const pool = ["credential-A", "credential-B", "credential-C", "credential-D"];
let nextSibling = 0;
registerCustomApi(
API,
(_model: Model<Api>, _context: Context, options?: SimpleStreamOptions) => {
pushKey(keys, options);
const stream = new AssistantMessageEventStream();
queueMicrotask(() => {
if (options?.apiKey === "credential-D") {
ok(stream);
return;
}
stream.push({ type: "start", partial: assistant() });
stream.push({
type: "error",
reason: "error",
error: assistantError("You have hit your ChatGPT usage limit (pro plan). Try again later.", 429),
});
});
return stream;
},
SOURCE_ID,
);
const stream = streamSimple(model(), context, {
apiKey: async ctx => {
contexts.push(ctx);
return ctx.error === undefined ? pool[0] : pool[++nextSibling];
},
});
for await (const event of stream) {
eventTypes.push(event.type);
}
expect((await stream.result()).content).toEqual([{ type: "text", text: "ok" }]);
expect(keys).toEqual(pool);
expect(contexts.map(ctx => ctx.lastChance)).toEqual([false, true, true, true]);
expect(eventTypes).toEqual(["start", "text_start", "text_delta", "text_end", "done"]);
});
it("stops replay-safe usage rotation when the resolver cycles to an attempted credential", async () => {
const keys: unknown[] = [];
const resolved = ["credential-A", "credential-B", "credential-A"];
let resolveIndex = 0;
registerCustomApi(
API,
(_model: Model<Api>, _context: Context, options?: SimpleStreamOptions) => {
pushKey(keys, options);
const stream = new AssistantMessageEventStream();
queueMicrotask(() => {
stream.push({ type: "start", partial: assistant() });
stream.push({
type: "error",
reason: "error",
error: assistantError("You have hit your ChatGPT usage limit (pro plan). Try again later.", 429),
});
});
return stream;
},
SOURCE_ID,
);
const stream = streamSimple(model(), context, {
apiKey: async () => resolved[resolveIndex++],
});
for await (const _event of stream) {
// drain
}
expect((await stream.result()).stopReason).toBe("error");
expect(keys).toEqual(["credential-A", "credential-B"]);
});
it("rotates before emitting content for Codex quota payloads", async () => {
const payloads: Array<{ message: string; status?: number }> = [
{ message: "429", status: 429 },
+1
View File
@@ -30,6 +30,7 @@
- Fixed the Model Hub role-assignment strip hiding the selected chip once the row overflowed; the strip now scrolls horizontally, truncating passed chips behind a leading ellipsis so the selection (plus one chip of lookahead) stays visible.
- Fixed mouse hover and clicks in the /models Roles view landing one row above the pointer (the row mapping subtracted the status row twice).
- Fixed model search keeping the most-recently-used model on top of the results: match quality now ranks first (an exact `gpt-5.5` beats the active `gpt-5.6-sol`), with MRU order only breaking ties between equally good matches.
- Fixed ModelRegistry/AuthStorage resolvers to continue replay-safe usage/account-quota turns across every distinct eligible credential and return exhaustion when no sibling switches, instead of re-resolving the same failed credential.
## [16.4.5] - 2026-07-11
@@ -59,12 +59,13 @@ export function createApiKeyResolver(
// sibling exists we switch immediately; the precise no-sibling backoff
// is owned by `markUsageLimitReached` (default + server usage-report
// reset) and the outer whole-turn retry layer.
await registry.authStorage.rotateSessionCredential(provider, sessionId, {
const rotated = await registry.authStorage.rotateSessionCredential(provider, sessionId, {
error,
modelId,
signal,
apiKey: previousKey,
});
if (!rotated) return undefined;
return registry.getApiKeyForProvider(provider, sessionId, { baseUrl, modelId });
}
return registry.getApiKeyForProvider(provider, sessionId, { baseUrl, modelId, forceRefresh: true, signal });
@@ -3,7 +3,8 @@ import * as path from "node:path";
import { scheduler } from "node:timers/promises";
import { Agent } from "@oh-my-pi/pi-agent-core";
import type { ApiKeyResolveContext, AssistantMessage, ToolCall } from "@oh-my-pi/pi-ai";
import { createMockModel } from "@oh-my-pi/pi-ai/providers/mock";
import { unregisterCustomApis } from "@oh-my-pi/pi-ai/api-registry";
import { createMockModel, registerMockApi } from "@oh-my-pi/pi-ai/providers/mock";
import * as aiStream from "@oh-my-pi/pi-ai/stream";
import { AssistantMessageEventStream } from "@oh-my-pi/pi-ai/utils/event-stream";
import { getBundledModel } from "@oh-my-pi/pi-catalog/models";
@@ -17,6 +18,8 @@ import { TempDir } from "@oh-my-pi/pi-utils";
type AutoRetryEndEvent = Extract<AgentSessionEvent, { type: "auto_retry_end" }>;
type AutoRetryStartEvent = Extract<AgentSessionEvent, { type: "auto_retry_start" }>;
const RETRY_CAP_MOCK_API_SOURCE = "agent-session-retry-cap-test";
function lastAssistant(session: AgentSession): AssistantMessage {
const message = session.agent.state.messages.at(-1);
if (message?.role !== "assistant") {
@@ -67,9 +70,10 @@ describe("AgentSession retry delay cap", () => {
await session.dispose();
session = undefined;
}
unregisterCustomApis(RETRY_CAP_MOCK_API_SOURCE);
vi.restoreAllMocks();
authStorage.close();
tempDir.removeSync();
vi.restoreAllMocks();
});
it("bails immediately when retry-after exceeds retry.maxDelayMs", async () => {
@@ -205,25 +209,45 @@ describe("AgentSession retry delay cap", () => {
expect(last.content).toContainEqual({ type: "text", text: "recovered after stream read retry" });
});
it("switches credentials instead of failing the delay cap for account rate limits", async () => {
it("rolls through four sibling credentials inside one AgentSession prompt before delay-cap retry", async () => {
const model = getBundledModel("anthropic", "claude-sonnet-4-5");
if (!model) {
throw new Error("Expected bundled Anthropic test model to exist");
const fallbackModel = getBundledModel("openai", "gpt-5");
if (!model || !fallbackModel) {
throw new Error("Expected bundled primary and fallback test models to exist");
}
const providerSessionId = "retry-four-credential-session-3";
registerMockApi(RETRY_CAP_MOCK_API_SOURCE);
authStorage.removeRuntimeApiKey("anthropic");
authStorage.setRuntimeApiKey("openai", "openai-fallback-key");
await authStorage.set("anthropic", [
{ type: "api_key", key: "anthropic-key-1" },
{ type: "api_key", key: "anthropic-key-2" },
{ type: "api_key", key: "anthropic-key-A" },
{ type: "api_key", key: "anthropic-key-B" },
{ type: "api_key", key: "anthropic-key-C" },
{ type: "api_key", key: "anthropic-key-D" },
]);
const rateLimitError =
'429 {"type":"error","error":{"type":"rate_limit_error","message":"This request would exceed your account\'s rate limit. Please try again later."}} retry-after-ms=11180000';
const mock = createMockModel();
const requestedKeys: string[] = [];
let agent!: Agent;
agent = new Agent({
getApiKey: model => modelRegistry.resolver(model, agent.sessionId),
const mock = createMockModel({
id: model.id,
provider: model.provider,
handler: (_context, options) => {
const apiKey = typeof options?.apiKey === "string" ? options.apiKey : undefined;
if (!apiKey) {
throw new Error("Expected streamSimple to pass a resolved string API key");
}
requestedKeys.push(apiKey);
return apiKey === "anthropic-key-D"
? { content: ["recovered on fourth credential"], stopReason: "stop" }
: { throw: rateLimitError };
},
});
const requestedModels: string[] = [];
const agent = new Agent({
getApiKey: model => modelRegistry.resolver(model, providerSessionId),
sessionId: providerSessionId,
initialState: {
model,
systemPrompt: ["Test"],
@@ -231,22 +255,20 @@ describe("AgentSession retry delay cap", () => {
messages: [],
},
streamFn: (requestedModel, context, options) => {
const apiKey = resolveInitialApiKey(options?.apiKey);
requestedKeys.push(apiKey);
if (requestedKeys.length === 1) {
mock.push({ throw: rateLimitError });
} else {
mock.push({ content: ["recovered after credential switch"] });
}
return mock.stream(requestedModel, context, options);
requestedModels.push(`${requestedModel.provider}/${requestedModel.id}`);
return aiStream.streamSimple(mock.model, context, options);
},
});
const settings = Settings.isolated({
"compaction.enabled": false,
"retry.baseDelayMs": 5,
"retry.maxDelayMs": 100,
"retry.maxRetries": 1,
"retry.baseDelayMs": 1,
"retry.maxDelayMs": 1,
"retry.maxRetries": 0,
"retry.modelFallback": true,
"retry.fallbackChains": {
default: [`${fallbackModel.provider}/${fallbackModel.id}`],
},
});
settings.setModelRole("default", `${model.provider}/${model.id}`);
@@ -255,9 +277,9 @@ describe("AgentSession retry delay cap", () => {
sessionManager: SessionManager.inMemory(),
settings,
modelRegistry,
providerSessionId,
});
const waitSpy = vi.spyOn(scheduler, "wait").mockResolvedValue(undefined);
const retryStartEvents: AutoRetryStartEvent[] = [];
const retryEndEvents: AutoRetryEndEvent[] = [];
session.subscribe(event => {
@@ -268,18 +290,26 @@ describe("AgentSession retry delay cap", () => {
await session.prompt("Trigger account rate limit with long retry-after");
await session.waitForIdle();
expect(requestedKeys).toHaveLength(2);
expect(new Set(requestedKeys).size).toBe(2);
expect(retryStartEvents).toHaveLength(1);
expect(retryStartEvents[0]).toMatchObject({ delayMs: 0 });
expect(retryEndEvents).toHaveLength(1);
expect(retryEndEvents[0]).toMatchObject({ success: true, attempt: 1 });
for (const call of waitSpy.mock.calls) {
expect(call[0]).toBeLessThanOrEqual(100);
expect(requestedModels).toEqual([`${model.provider}/${model.id}`]);
expect(requestedKeys).toEqual(["anthropic-key-A", "anthropic-key-B", "anthropic-key-C", "anthropic-key-D"]);
expect(new Set(requestedKeys).size).toBe(4);
expect(mock.calls).toHaveLength(4);
expect(retryStartEvents).toHaveLength(0);
expect(retryEndEvents).toHaveLength(0);
expect(session.model?.provider).toBe(model.provider);
expect(session.model?.id).toBe(model.id);
for (const call of mock.calls) {
expect(call.context.messages.filter(message => message.role === "user")).toHaveLength(1);
}
expect(session.agent.state.messages.filter(message => message.role === "user")).toHaveLength(1);
expect(
session.agent.state.messages.some(
message => message.role === "custom" && "customType" in message && message.customType === "irc:incoming",
),
).toBe(false);
const last = lastAssistant(session);
expect(last.stopReason).toBe("stop");
expect(last.content).toContainEqual({ type: "text", text: "recovered after credential switch" });
expect(last.content).toContainEqual({ type: "text", text: "recovered on fourth credential" });
});
it("switches same-provider credentials before model fallback on ChatGPT usage limits", async () => {
@@ -2,9 +2,11 @@ import { afterEach, beforeEach, describe, expect, test, vi } from "bun:test";
import * as fs from "node:fs";
import * as os from "node:os";
import * as path from "node:path";
import type { UsageProvider } from "@oh-my-pi/pi-ai";
import { type UsageProvider, withAuth } from "@oh-my-pi/pi-ai";
import * as oauth from "@oh-my-pi/pi-ai/oauth";
import type { OAuthCredentials } from "@oh-my-pi/pi-ai/oauth/types";
import { getBundledModel } from "@oh-my-pi/pi-catalog/models";
import { ModelRegistry } from "@oh-my-pi/pi-coding-agent/config/model-registry";
import { AuthStorage } from "@oh-my-pi/pi-coding-agent/session/auth-storage";
import { removeSyncWithRetries, Snowflake } from "@oh-my-pi/pi-utils";
@@ -44,7 +46,7 @@ describe("AuthStorage account rotation", () => {
// Stub the refresh path so AuthStorage doesn't hit a real OAuth endpoint
// when the credential lands inside the 60s skew. Returning the credential
// unchanged preserves the test's deterministic accountId routing.
// unchanged preserves deterministic access-token routing.
vi.spyOn(oauth, "refreshOAuthToken").mockImplementation(async (_provider, credential) => {
return credential;
});
@@ -52,7 +54,7 @@ describe("AuthStorage account rotation", () => {
const credential = credentials["openai-codex"] as OAuthCredentials | undefined;
if (!credential) return null;
return {
apiKey: `api-${credential.accountId ?? "unknown"}`,
apiKey: credential.access,
newCredentials: credential,
};
});
@@ -86,14 +88,14 @@ describe("AuthStorage account rotation", () => {
const sessionId = "issue-55-session";
const firstKey = await authStorage.getApiKey("openai-codex", sessionId);
expect(firstKey).toMatch(/^api-acct-/);
expect(firstKey).toMatch(/^access-/);
usageExhausted = true;
const { switched } = await authStorage.markUsageLimitReached("openai-codex", sessionId);
expect(switched).toBe(true);
const exhaustedFallbackKey = await authStorage.getApiKey("openai-codex", sessionId);
expect(exhaustedFallbackKey).toMatch(/^api-acct-/);
expect(exhaustedFallbackKey).toMatch(/^access-/);
});
test("usage-limit rotation can match the failed bearer when session stickiness is missing", async () => {
@@ -117,7 +119,7 @@ describe("AuthStorage account rotation", () => {
const sessionId = "missing-sticky-session";
const result = await authStorage.markUsageLimitReached("openai-codex", sessionId, { apiKey: "access-1" });
expect(result.switched).toBe(true);
expect(await authStorage.getApiKey("openai-codex", sessionId)).toBe("api-acct-2");
expect(await authStorage.getApiKey("openai-codex", sessionId)).toBe("access-2");
});
test("usage-limit rotation trusts the failed bearer over stale session stickiness", async () => {
@@ -140,9 +142,62 @@ describe("AuthStorage account rotation", () => {
const sessionId = "stale-sticky-session";
const stickyKey = await authStorage.getApiKey("openai-codex", sessionId);
const failedKey = stickyKey === "api-plus-acct" ? "k12-access" : "plus-access";
const failedKey = stickyKey === "plus-access" ? "k12-access" : "plus-access";
const result = await authStorage.markUsageLimitReached("openai-codex", sessionId, { apiKey: failedKey });
expect(result.switched).toBe(true);
expect(await authStorage.getApiKey("openai-codex", sessionId)).toBe(stickyKey);
});
test("withAuth reaches a fourth healthy Codex OAuth sibling through ModelRegistry", async () => {
await authStorage.set("openai-codex", [
{
type: "oauth",
access: "access-a",
refresh: "refresh-a",
expires: Date.now() + 60_000,
accountId: "acct-a",
},
{
type: "oauth",
access: "access-b",
refresh: "refresh-b",
expires: Date.now() + 60_000,
accountId: "acct-b",
},
{
type: "oauth",
access: "access-c",
refresh: "refresh-c",
expires: Date.now() + 60_000,
accountId: "acct-c",
},
{
type: "oauth",
access: "access-d",
refresh: "refresh-d",
expires: Date.now() + 60_000,
accountId: "acct-d",
},
]);
const model = getBundledModel("openai-codex", "gpt-5.5");
if (!model) {
throw new Error("Expected bundled Codex test model to exist");
}
const modelRegistry = new ModelRegistry(authStorage, path.join(tempDir, "models.yml"));
const attemptedKeys: string[] = [];
const result = await withAuth(modelRegistry.resolver(model, "codex-four-oauth-session"), async key => {
attemptedKeys.push(key);
if (key !== "access-d") {
throw new Error("You have hit your ChatGPT usage limit (pro plan). Try again later.");
}
return key;
});
expect(result).toBe("access-d");
expect(attemptedKeys.at(-1)).toBe("access-d");
expect([...attemptedKeys].sort()).toEqual(["access-a", "access-b", "access-c", "access-d"]);
expect(new Set(attemptedKeys).size).toBe(4);
});
});