From f993030ed24cd6defde9b685bdb06ecdae0f1b61 Mon Sep 17 00:00:00 2001 From: PlaygrndLabs Date: Wed, 28 Oct 2026 13:55:23 +0000 Subject: [PATCH] Route across logins and keys with CPA's strategies, retries and cooldowns Each Claude and ChatGPT login is now a candidate of its own, so a request moves on to the next login when one is rate limited or failing, as it already did for API keys. The new free/src/routing.ts follows CPA's auth conductor: round-robin, weighted-round-robin and fill-first strategies, request-retry rounds with max-retry-credentials and max-retry-interval, and cooldowns (Retry-After and provider reset hints, the 1 s to 30 min quota backoff, 30 min for 401/402/403 keys, 12 h for 404, 60 s for transient errors), with disable-cooling and request-retry overrides per key and per login. Isolates share cooldowns through the Durable Object, the Auth Files list shows them in CPA's shape, and the dashboard's reset button clears them. --- free/README.md | 5 +- free/src/account.ts | 86 +++++++-- free/src/app.ts | 278 +++++++++++++++++++---------- free/src/management.ts | 57 +++++- free/src/providers.ts | 88 ++++++++-- free/src/routing.ts | 329 +++++++++++++++++++++++++++++++++++ free/src/store.ts | 12 +- free/src/vault.ts | 39 +++++ free/test/app.test.ts | 191 ++++++++++++++++++-- free/test/management.test.ts | 61 +++++++ free/test/routing.test.ts | 73 ++++++++ 11 files changed, 1068 insertions(+), 151 deletions(-) create mode 100644 free/src/routing.ts create mode 100644 free/test/routing.test.ts diff --git a/free/README.md b/free/README.md index e6729d7..ba932ab 100644 --- a/free/README.md +++ b/free/README.md @@ -6,10 +6,10 @@ It covers **Claude and ChatGPT (Codex) subscriptions** and **API keys** for Clau - CPA's dashboard, downloaded at build time from the latest panel release (pin one with `PANEL_VERSION`, and its checksum with `PANEL_SHA256`). - Claude and ChatGPT (Codex) OAuth logins from the dashboard's **OAuth Login** page (the same PKCE flows as CPA, Claude Code and the Codex CLI), with tokens refreshed automatically. -- Several logins per provider, used in turn. Upload, download, disable, edit and delete them under **Auth Files**, in CPA's auth-file format, so files move between the two installs. +- Several logins per provider, used in turn, with a failing or rate-limited one skipped for the next. Upload, download, disable, edit and delete them under **Auth Files**, in CPA's auth-file format, so files move between the two installs. - CPA's YAML config, edited in the **Config Panel**. Client API keys live in `access.api-keys`, as in CPA. - `POST /v1/messages`, `POST /v1/messages/count_tokens` and `GET /v1/models`, streamed straight through to Anthropic. -- API-key providers added under **AI Providers** (CPA's `api-keys` config): Claude, Gemini, Codex and OpenAI-compatible endpoints such as OpenRouter. Model aliases, `prefix/model` routing, priorities, key weights and excluded models work as in CPA, and a request moves on to the next key when one is rate limited or failing. +- API-key providers added under **AI Providers** (CPA's `api-keys` config): Claude, Gemini, Codex and OpenAI-compatible endpoints such as OpenRouter. Model aliases, `prefix/model` routing, priorities, key weights (with the `weighted-round-robin` strategy) and excluded models work as in CPA, and a request moves on to the next key when one is rate limited or failing. - Codex CLI through a ChatGPT login, on `/backend-api/codex/responses` or `/v1/responses`. Other OpenAI Responses clients work too: their requests are adapted to what the ChatGPT endpoint takes, and a non-streaming request gets the final response as JSON. - Claude Code on an OpenAI-compatible key (OpenRouter and the like): ask for one of the provider's model aliases and the request goes out as Chat Completions, with the answer streamed back as Anthropic events, following CPA's Claude-to-OpenAI translator. - Claude Code on a ChatGPT login or a Codex API key: ask for a ChatGPT model (for example `ANTHROPIC_MODEL=gpt-5.6-sol`) and the request is translated to OpenAI Responses and the answer streamed back as Anthropic events, with thinking, tool calls and images, following CPA's Claude-to-Codex translator. @@ -17,6 +17,7 @@ It covers **Claude and ChatGPT (Codex) subscriptions** and **API keys** for Clau - OpenAI Chat clients (Cursor, Cline, the OpenAI SDKs and the like) on a Claude API key: ask `/v1/chat/completions` for a Claude model and the request goes out as Anthropic Messages, with reasoning effort, tool calls, images, files and prompt caching, and the answer comes back as Chat Completions, following CPA's OpenAI-to-Claude translator. Claude logins aren't offered to these clients, since without CPA's cloaking Anthropic accepts a subscription token from Claude Code only. - Each client talks in its own format: Anthropic (`/v1/messages`), OpenAI Chat (`/v1/chat/completions`), OpenAI Responses (`/v1/responses`) and Gemini (`/v1beta/models/{model}:generateContent`). A request goes to a provider that takes the same format, apart from the Claude translations above. - Usage and logs as in CPA: each login and API key shows its success and failure counts and the last 200 minutes in 10-minute buckets, and with **Logging to file** on, the **Logs** page shows a line per API request (the last 2,000). Lines hold the time, status, latency, client IP, path, provider and model, never request bodies or keys. Records are sent to the Durable Object in batches a few seconds apart. +- CPA's routing across logins and keys, set in the `routing` config section: the `round-robin` (default), `weighted-round-robin` and `fill-first` strategies within the highest priority, failover to the next login or key, extra rounds with `retry.request-retry`, `max-retry-credentials` and `max-retry-interval`, and CPA's cooldowns: a rate-limited credential rests until the provider's stated reset (Retry-After, Anthropic's rejected-window reset, ChatGPT's usage-limit reset, Gemini's retry delay) or on CPA's 1 s to 30 min backoff, a refused or unpaid one for 30 minutes, a missing model for 12 hours, and a transient error for 60 seconds (`cooldown.transient-error-cooldown-seconds`). `cooldown.disable-cooling`, and per-key or per-login `disable-cooling`, `request-retry`, `priority`, `weight` and `prefix`, work as in CPA. **Auth Files** shows each login's cooldowns, and the reset button clears them. Each isolate shares its cooldowns through the Durable Object, so other isolates see them within a minute. A request makes at most 10 upstream attempts and waits at most 2 minutes in all, to stay inside the free plan's limits. With a single Claude login and nothing else, Claude Code requests still stream straight through. - Everything is stored in one SQLite-backed Durable Object, which the free plan includes. Claude Code and Codex CLI request bodies are never parsed, so a Worker spends well under 1 ms of its 10 ms CPU budget even on a 600 KB Claude Code request (see the CPU test in `test/app.test.ts`). Requests that need a model rewritten or a body adapted are parsed once. Translated requests cost the most, about 4 to 6 ms for a 600 KB conversation; very long conversations translated can run past the free plan's 10 ms, which the Container install doesn't have. diff --git a/free/src/account.ts b/free/src/account.ts index f133345..52ace15 100644 --- a/free/src/account.ts +++ b/free/src/account.ts @@ -1,7 +1,8 @@ import { OAuthError, authorizeUrl, exchangeCode, parseCallback, randomToken, refresh, type Fetcher } from "./claude-oauth.ts"; import * as codex from "./codex-oauth.ts"; import { Config } from "./config.ts"; -import { upstreamCredentials, type RoutingTable, type UpstreamCredential } from "./providers.ts"; +import { flagOf, retryOf, upstreamCredentials, weightOf, type LoginEntry, type RoutingTable, type UpstreamCredential } from "./providers.ts"; +import { routingSettings, type CooldownEvent, type RoutingSettings } from "./routing.ts"; import { storedKey, type UsageRecord } from "./usage.ts"; import { claudeCredential, claudeFileData, type CredentialFile, type LoginSession, type Vault } from "./vault.ts"; @@ -17,13 +18,15 @@ const LABEL: Record = { claude: "Claude", codex: "ChatGPT // state below (config, token refreshes, login sessions) is never raced by another copy. export interface Backend { keyValid(hash: string): Promise; - token(): Promise; - codexToken(): Promise; + // A login's token by its file name, or, with no name, the next login's in turn. + token(name?: string): Promise; + codexToken(name?: string): Promise; invalidate(accessToken: string): Promise; attemptAllowed(client: string): Promise; manage(request: Request): Promise; routing(): Promise; record(records: UsageRecord[]): Promise; + cool(events: CooldownEvent[]): Promise; } export const REFRESH_MARGIN_SECONDS = 5 * 60; @@ -36,6 +39,29 @@ interface Login { readonly accountId: string; } +function textOf(value: unknown): string { + return typeof value === "string" ? value.trim() : ""; +} + +function numberOf(value: unknown): number { + const parsed = typeof value === "string" && value.trim() !== "" ? Number(value) : value; + return typeof parsed === "number" && Number.isFinite(parsed) ? parsed : 0; +} + +// CPA reads a login's routing fields from the top level of its auth file. +function loginEntry(file: CredentialFile): LoginEntry { + const data = file.data; + return { + name: file.name, + provider: file.provider === "codex" ? "codex" : "claude", + priority: numberOf(data.priority), + weight: weightOf(data.weight), + prefix: textOf(data.prefix), + disableCooling: flagOf(data.disable_cooling ?? data["disable-cooling"]), + requestRetry: retryOf(data.request_retry ?? data["request-retry"]), + }; +} + function loginOf(file: CredentialFile): Login | null { if (file.provider === "codex") return codex.codexCredential(file.data); const credential = claudeCredential(file.data); @@ -50,6 +76,7 @@ export async function hashKey(key: string): Promise { export class Account { private config: Config | null = null; private upstreams: UpstreamCredential[] | null = null; + private settings: RoutingSettings | null = null; private keyHashes: Set | null = null; private readonly refreshing = new Map>(); private readonly lastForcedRefresh = new Map(); @@ -71,13 +98,40 @@ export class Account { this.config = config; this.keyHashes = null; this.upstreams = null; + this.settings = null; + // API-key ids follow their place in the config, so their cooldowns don't outlive an edit. + this.vault.clearCooldowns("keys"); } - // What the Worker needs to pick an upstream for a request: the configured API-key providers - // and whether any Claude login is on. + // What the Worker needs to pick an upstream for a request: the configured API-key providers, + // the enabled logins, the routing settings and the cooldowns every isolate has reported. async routing(): Promise { - this.upstreams ??= upstreamCredentials(this.currentConfig().read([])); - return { credentials: this.upstreams, claudeOAuth: this.loginFiles("claude").length > 0, codexOAuth: this.loginFiles("codex").length > 0 }; + const tree = this.currentConfig().read([]); + this.upstreams ??= upstreamCredentials(tree); + this.settings ??= routingSettings(tree); + const logins = LOGIN_PROVIDERS.flatMap((provider) => this.loginFiles(provider)).map(loginEntry); + return { + credentials: this.upstreams, + claudeOAuth: logins.some((login) => login.provider === "claude"), + codexOAuth: logins.some((login) => login.provider === "codex"), + logins, + settings: this.settings, + cooldowns: this.vault.cooldowns(this.nowSeconds()), + }; + } + + // Cooldowns and successes reported by the Worker's isolates. + async cool(events: CooldownEvent[]): Promise { + for (const event of events.slice(0, 200)) { + if (typeof event?.id !== "string" || event.id === "" || event.id.length > 512) continue; + if ("clear" in event) { + this.vault.clearCooldowns(event.id); + continue; + } + const until = Math.min(Number(event.until), this.nowSeconds() + 24 * 60 * 60); + if (typeof event.model !== "string" || event.model.length > 256 || !Number.isFinite(until)) continue; + this.vault.saveCooldown({ id: event.id, model: event.model, until, reason: String(event.reason).slice(0, 64), status: Number(event.status) || 0, level: Math.max(0, Math.min(32, Number(event.level) || 0)) }); + } } async keyValid(hash: string): Promise { @@ -96,18 +150,24 @@ export class Account { return this.loginFiles("claude"); } - token(): Promise { - return this.loginToken("claude"); + token(name?: string): Promise { + return this.loginToken("claude", name); } - codexToken(): Promise { - return this.loginToken("codex"); + codexToken(name?: string): Promise { + return this.loginToken("codex", name); } - // Picks the provider's next enabled login in turn (round-robin), refreshing it when near expiry. - async loginToken(provider: LoginProvider): Promise { + // The named login's token, or the provider's next enabled login in turn (round-robin), + // refreshed when near expiry. + async loginToken(provider: LoginProvider, name?: string): Promise { const files = this.loginFiles(provider); if (files.length === 0) return { ok: false, status: 503, message: `No ${LABEL[provider]} account is connected yet: add one under OAuth login in the dashboard.` }; + if (name !== undefined) { + const file = files.find((candidate) => candidate.name === name); + if (file === undefined) return { ok: false, status: 503, message: `That ${LABEL[provider]} login is disabled or was removed.` }; + return this.tokenFor(file); + } const turn = this.next.get(provider) ?? 0; this.next.set(provider, turn + 1); return this.tokenFor(files[turn % files.length] as CredentialFile); diff --git a/free/src/app.ts b/free/src/app.ts index 9890113..94c6b12 100644 --- a/free/src/app.ts +++ b/free/src/app.ts @@ -6,8 +6,9 @@ import { claudeEventStream, claudeMessage, ClaudeStream, claudeToResponses } fro import { chatCompletion, ChatStream, chatToClaude } from "./openai-claude.ts"; import { MANAGEMENT_PREFIX } from "./management.ts"; import { loginPage, messagePage } from "./pages.ts"; -import { candidates, inDefaultFamily, listedModels, NATIVE_FORMAT, ordered, type Candidate, type Format, type RoutingTable } from "./providers.ts"; +import { candidates, inDefaultFamily, listedModels, NATIVE_FORMAT, splitModel, type Candidate, type Format, type LoginEntry, type RoutingTable } from "./providers.ts"; import { CODEX_UPSTREAM, codexRequest, downstreamResponse, formatError, presentedKey, providerRequest, routeOf, upstreamRequest, UPSTREAM, type Route } from "./proxy.ts"; +import { CooldownBook, retryHint, route as routed, type CooldownEvent } from "./routing.ts"; import { apiKeyKey, loginKey, requestLine, type Attempt, type UsageRecord } from "./usage.ts"; import { SESSION_COOKIE, SESSION_SECONDS, issueSession, passwordMatches, readCookie, readSmallForm, sessionIsValid } from "./session.ts"; @@ -26,22 +27,33 @@ const USAGE_FLUSH_MS = 5000; const USAGE_FLUSH_COUNT = 50; const OAUTH_CALLBACK_PATHS = new Set(["/callback", "/auth/callback", "/oauth-callback", "/anthropic/callback", "/codex/callback"]); +export interface CachedToken { + readonly value: string; + // The login file the token belongs to. + readonly name: string; + readonly accountId: string; + readonly expiresAt: number; + readonly cachedUntil: number; +} + // Per-isolate memory, so most API calls skip the Durable Object round trip entirely. export class EdgeCache { readonly keys = new Map(); - token: { value: string; name: string; expiresAt: number; cachedUntil: number } | null = null; - codexToken: { value: string; name: string; accountId: string; expiresAt: number; cachedUntil: number } | null = null; + // Login tokens by ":"; the name is empty for "the next login in turn". + readonly tokens = new Map(); routing: { table: RoutingTable; until: number } | null = null; + readonly cooldowns = new CooldownBook(); turn = 0; usage: UsageRecord[] = []; + cooling: CooldownEvent[] = []; flushing = false; // A config or login change reaches this isolate at once and others within a minute. forgetConfig(): void { this.keys.clear(); this.routing = null; - this.token = null; - this.codexToken = null; + this.tokens.clear(); + this.cooldowns.clear(); } } @@ -56,6 +68,8 @@ export interface AppConfig { // Keeps the isolate alive for work after the response, such as sending usage records. readonly waitUntil?: (promise: Promise) => void; readonly usageFlushMs?: number; + // Waits between retry rounds; tests pass their own. + readonly sleep?: (ms: number) => Promise; } function apiError(status: number, type: string, message: string): Response { @@ -102,38 +116,45 @@ async function routingTable(config: AppConfig): Promise { return table; } -async function accessToken(config: AppConfig): Promise<{ token: string } | Response> { - const cached = config.cache.token; +function tokenKey(provider: LoginEntry["provider"], name: string): string { + return `${provider}:${name}`; +} + +// A login's token from this isolate's cache, or from the Durable Object, which keeps it fresh. +// An empty name asks for the next login in turn. +async function loginToken(config: AppConfig, provider: LoginEntry["provider"], name: string): Promise { + const key = tokenKey(provider, name); + const cached = config.cache.tokens.get(key); const now = config.nowSeconds(); - if (cached !== null && cached.cachedUntil > now && cached.expiresAt - now > REFRESH_MARGIN_SECONDS) return { token: cached.value }; - const result = await config.backend.token(); - if (!result.ok) return apiError(result.status, "api_error", result.message); - config.cache.token = { value: result.accessToken, name: result.name ?? "", expiresAt: result.expiresAt, cachedUntil: now + TOKEN_CACHE_SECONDS }; - return { token: result.accessToken }; + if (cached !== undefined && cached.cachedUntil > now && cached.expiresAt - now > REFRESH_MARGIN_SECONDS) return cached; + const login = name === "" ? undefined : name; + const result = await (provider === "claude" ? config.backend.token(login) : config.backend.codexToken(login)); + if (!result.ok) return { status: result.status, message: result.message }; + const token = { value: result.accessToken, name: result.name ?? name, accountId: result.accountId ?? "", expiresAt: result.expiresAt, cachedUntil: now + TOKEN_CACHE_SECONDS }; + config.cache.tokens.set(key, token); + return token; +} + +// The provider refused the token: drop it here and have the Durable Object refresh it. +async function refused(config: AppConfig, provider: LoginEntry["provider"], name: string, token: string): Promise { + config.cache.tokens.delete(tokenKey(provider, name)); + await config.backend.invalidate(token); } // A Claude subscription login, through the OAuth token the Durable Object keeps fresh. -async function viaClaudeLogin(request: Request, url: URL, config: AppConfig, body?: string): Promise { - const token = await accessToken(config); - if (token instanceof Response) return token; - const upstream = await config.upstream(upstreamRequest(request, url, token.token, body)).catch(() => null); +async function viaClaudeLogin(request: Request, url: URL, config: AppConfig, body?: string, name = ""): Promise { + const token = await loginToken(config, "claude", name); + if (!("value" in token)) return apiError(token.status, "api_error", token.message); + const upstream = await config.upstream(upstreamRequest(request, url, token.value, body)).catch(() => null); if (upstream === null) return apiError(502, "api_error", "Anthropic couldn't be reached."); - if (upstream.status === 401) { - config.cache.token = null; - await config.backend.invalidate(token.token); - } + if (upstream.status === 401) await refused(config, "claude", name, token.value); return downstreamResponse(upstream); } -async function codexAccess(config: AppConfig): Promise<{ token: string; accountId: string } | Response> { - const cached = config.cache.codexToken; - const now = config.nowSeconds(); - if (cached !== null && cached.cachedUntil > now && cached.expiresAt - now > REFRESH_MARGIN_SECONDS) return { token: cached.value, accountId: cached.accountId }; - const result = await config.backend.codexToken(); - if (!result.ok) return formatError("openai-response", result.status, "api_error", result.message); - const accountId = result.accountId ?? ""; - config.cache.codexToken = { value: result.accessToken, name: result.name ?? "", accountId, expiresAt: result.expiresAt, cachedUntil: now + TOKEN_CACHE_SECONDS }; - return { token: result.accessToken, accountId }; +async function codexAccess(config: AppConfig, name = ""): Promise<{ token: string; accountId: string } | Response> { + const token = await loginToken(config, "codex", name); + if (!("value" in token)) return formatError("openai-response", token.status, "api_error", token.message); + return { token: token.value, accountId: token.accountId }; } // Fields the ChatGPT Codex endpoint refuses, which other Responses clients may send (as CPA strips). @@ -182,18 +203,15 @@ async function collected(upstream: Response): Promise { } // A ChatGPT (Codex) subscription login, through the OAuth token the Durable Object keeps fresh. -async function viaCodexLogin(request: Request, url: URL, config: AppConfig, text: string): Promise { +async function viaCodexLogin(request: Request, url: URL, config: AppConfig, text: string, name = ""): Promise { const native = url.pathname === "/backend-api/codex/responses" || request.headers.has("originator"); const prepared = codexBody(text, native); if (prepared === null) return formatError("openai-response", 400, "invalid_request_error", "The request body must be a JSON object."); - const access = await codexAccess(config); + const access = await codexAccess(config, name); if (access instanceof Response) return access; const upstream = await config.upstream(codexRequest(request, access.token, access.accountId, prepared.body)).catch(() => null); if (upstream === null) return formatError("openai-response", 502, "api_error", "ChatGPT couldn't be reached."); - if (upstream.status === 401) { - config.cache.codexToken = null; - await config.backend.invalidate(access.token); - } + if (upstream.status === 401) await refused(config, "codex", name, access.token); if (!upstream.ok || prepared.stream) return downstreamResponse(upstream); return collected(upstream); } @@ -240,8 +258,9 @@ async function viaTranslation(request: Request, url: URL, route: Route, config: const bare = new Request(request.url, { method: "POST" }); let upstream: Response | null; let accessToken = ""; - if (candidate.credential.id === CODEX_LOGIN_ID) { - const access = await codexAccess(config); + const login = loginName(candidate); + if (login !== null) { + const access = await codexAccess(config, login); if (access instanceof Response) return formatError("claude", access.status, "api_error", await upstreamMessage(access)); accessToken = access.token; upstream = await config.upstream(codexRequest(bare, access.token, access.accountId, body)).catch(() => null); @@ -256,10 +275,7 @@ async function viaTranslation(request: Request, url: URL, route: Route, config: } if (upstream === null) return formatError("claude", 502, "api_error", "The upstream couldn't be reached."); if (!upstream.ok) { - if (upstream.status === 401 && accessToken !== "") { - config.cache.codexToken = null; - await config.backend.invalidate(accessToken); - } + if (upstream.status === 401 && accessToken !== "") await refused(config, "codex", login ?? "", accessToken); return formatError("claude", upstream.status, upstream.status === 429 ? "rate_limit_error" : "api_error", await upstreamMessage(upstream)); } if (translated.stream && upstream.body !== null) { @@ -298,16 +314,45 @@ async function viaClaudeKey(request: Request, config: AppConfig, candidate: Cand return Response.json(chatCompletion(final, requested, translated.toolNames), { headers: { "cache-control": "no-store" } }); } -// Statuses after which another credential may well succeed. -const RETRY_STATUSES = new Set([401, 403, 408, 429, 500, 502, 503, 504, 529]); -const MAX_ATTEMPTS = 3; -const CLAUDE_LOGIN_ID = "claude-login"; -const CODEX_LOGIN_ID = "codex-login"; +// A login is a candidate of its own, with id "file:" as in the usage counts. +const LOGIN_GROUP = "oauth"; + +function loginCandidate(login: LoginEntry, model: string): Candidate { + const baseUrl = login.provider === "claude" ? UPSTREAM : CODEX_UPSTREAM; + return { + model, + credential: { + id: loginKey(login.name), + kind: login.provider, + group: LOGIN_GROUP, + baseUrl, + apiKey: "", + weight: login.weight, + priority: login.priority, + prefix: login.prefix, + headers: {}, + models: [], + excluded: [], + disableCooling: login.disableCooling, + requestRetry: login.requestRetry, + }, + }; +} + +// The login file a candidate uses ("" for the next login in turn), or null for an API key. +function loginName(candidate: Candidate): string | null { + return candidate.credential.group === LOGIN_GROUP ? candidate.credential.id.slice("file:".length) : null; +} -function loginCandidate(kind: "claude" | "codex", model: string): Candidate { - const id = kind === "claude" ? CLAUDE_LOGIN_ID : CODEX_LOGIN_ID; - const baseUrl = kind === "claude" ? UPSTREAM : CODEX_UPSTREAM; - return { model, credential: { id, kind, group: "oauth", baseUrl, apiKey: "", weight: 1, priority: 0, prefix: "", headers: {}, models: [], excluded: [] } }; +// Each enabled login. A table without the list (an older Durable Object) stands for one login per +// provider, which the Durable Object then picks in turn. +function anyLogin(provider: LoginEntry["provider"]): LoginEntry { + return { name: "", provider, priority: 0, weight: 1, prefix: "", disableCooling: null, requestRetry: null }; +} + +function loginsOf(table: RoutingTable): readonly LoginEntry[] { + if (table.logins !== undefined) return table.logins; + return [...(table.claudeOAuth ? [anyLogin("claude")] : []), ...(table.codexOAuth ? [anyLogin("codex")] : [])]; } // Most clients put "model" first. Then it is read and replaced in place, without parsing the @@ -349,18 +394,32 @@ interface Trace { function noteAttempt(trace: Trace, config: AppConfig, candidate: Candidate, status: number): void { trace.provider = candidate.credential.kind; - let key: string | null = apiKeyKey(candidate.credential); - if (candidate.credential.id === CLAUDE_LOGIN_ID) key = config.cache.token?.name ? loginKey(config.cache.token.name) : null; - else if (candidate.credential.id === CODEX_LOGIN_ID) key = config.cache.codexToken?.name ? loginKey(config.cache.codexToken.name) : null; + const login = loginName(candidate); + let key: string | null = login === null ? apiKeyKey(candidate.credential) : loginKey(login); + if (login === "") { + const name = config.cache.tokens.get(tokenKey(candidate.credential.kind === "codex" ? "codex" : "claude", ""))?.name; + key = name ? loginKey(name) : null; + } if (key !== null) trace.attempts.push({ key, ok: status < 400 }); } +function sleep(ms: number): Promise { + return new Promise((resolve) => setTimeout(resolve, ms)); +} + +// Error bodies are read for a wait hint only while small. +const HINT_BODY_LIMIT = 64 * 1024; + async function proxied(request: Request, url: URL, route: Route & { format: Format }, config: AppConfig, trace: Trace): Promise { const table = await routingTable(config); - // Claude clients with only Claude logins configured: stream the body through unread. - if (route.format === "claude" && !table.codexOAuth && !table.credentials.some((credential) => servable("claude", credential.kind))) { - const response = await viaClaudeLogin(request, url, config); - noteAttempt(trace, config, loginCandidate("claude", ""), response.status); + const logins = loginsOf(table); + const claudeLogins = logins.filter((login) => login.provider === "claude"); + const codexLogins = logins.filter((login) => login.provider === "codex"); + // Claude clients with one Claude login and nothing else: stream the body through unread. + if (route.format === "claude" && claudeLogins.length === 1 && codexLogins.length === 0 && !table.credentials.some((credential) => servable("claude", credential.kind))) { + const only = loginCandidate(claudeLogins[0] as LoginEntry, ""); + const response = await viaClaudeLogin(request, url, config, undefined, loginName(only) ?? ""); + noteAttempt(trace, config, only, response.status); return response; } const length = Number(request.headers.get("content-length") ?? "0"); @@ -372,15 +431,18 @@ async function proxied(request: Request, url: URL, route: Route & { format: Form trace.model = model; const found = candidates(table, model); const usable = found.filter((candidate) => servable(route.format, candidate.credential.kind)); - if (route.format === "claude" && table.claudeOAuth && /^claude-/i.test(model)) usable.push(loginCandidate("claude", model)); + // "prefix/model" picks the logins with that prefix; other names may use any login. + const { prefix, model: bare } = splitModel(table, model); + const loginsFor = (provider: LoginEntry["provider"]) => (provider === "claude" ? claudeLogins : codexLogins).filter((login) => prefix === "" || login.prefix === prefix).map((login) => loginCandidate(login, bare)); + if (route.format === "claude" && /^claude-/i.test(bare)) usable.push(...loginsFor("claude")); // ChatGPT logins take Codex's own models, and any Responses model nothing else serves. - if (route.format === "openai-response" && table.codexOAuth && (inDefaultFamily("codex", model) || usable.length === 0)) usable.push(loginCandidate("codex", model)); + if (route.format === "openai-response" && (inDefaultFamily("codex", bare) || usable.length === 0)) usable.push(...loginsFor("codex")); // Claude Code asking for a ChatGPT model by name, e.g. ANTHROPIC_MODEL=gpt-5.6-sol. - if (route.format === "claude" && table.codexOAuth && inDefaultFamily("codex", model)) usable.push(loginCandidate("codex", model)); + if (route.format === "claude" && inDefaultFamily("codex", bare)) usable.push(...loginsFor("codex")); if (usable.length === 0) { if (route.format === "claude" && table.claudeOAuth) { const response = await viaClaudeLogin(request, url, config, text); - noteAttempt(trace, config, loginCandidate("claude", model), response.status); + noteAttempt(trace, config, loginCandidate(anyLogin("claude"), model), response.status); return response; } // Without CPA's cloaking a Claude login serves Claude Code only, so it isn't offered here. @@ -392,34 +454,55 @@ async function proxied(request: Request, url: URL, route: Route & { format: Form : `No provider is configured for model ${model}. Add one under AI Providers in the dashboard.`; return formatError(route.format, 400, "invalid_request_error", reason); } - config.cache.turn += 1; - const attempts = ordered(usable, config.cache.turn).slice(0, MAX_ATTEMPTS); - let last: Response | null = null; - for (const [index, candidate] of attempts.entries()) { - const isLast = index === attempts.length - 1; + // One attempt with one candidate, through the path its provider and the request format need. + const serveWith = async (candidate: Candidate, attemptConfig: AppConfig): Promise => { const body = parsed === null || candidate.model === model ? text : parsed.withModel(candidate.model); const translated = route.format !== NATIVE_FORMAT[candidate.credential.kind]; - if (translated || candidate.credential.id === CLAUDE_LOGIN_ID || candidate.credential.id === CODEX_LOGIN_ID) { - let response: Response; - if (translated && route.format === "openai") response = await viaClaudeKey(request, config, candidate, text); - else if (translated) response = await viaTranslation(request, url, route, config, candidate, text); - else if (candidate.credential.id === CLAUDE_LOGIN_ID) response = await viaClaudeLogin(request, url, config, body); - else response = await viaCodexLogin(request, url, config, body); + const login = loginName(candidate); + if (translated && route.format === "openai") return viaClaudeKey(request, attemptConfig, candidate, text); + if (translated) return viaTranslation(request, url, route, attemptConfig, candidate, text); + if (login !== null && candidate.credential.kind === "claude") return viaClaudeLogin(request, url, attemptConfig, body, login); + if (login !== null) return viaCodexLogin(request, url, attemptConfig, body, login); + const upstream = await attemptConfig.upstream(providerRequest(request, url, route, candidate.credential, candidate.model, body)).catch(() => null); + return upstream === null ? formatError(route.format, 502, "api_error", "The upstream couldn't be reached.") : downstreamResponse(upstream); + }; + // The first request in an isolate starts at the first credential, as CPA's rotation does. + const turn = config.cache.turn; + config.cache.turn += 1; + return routed({ + table, + usable, + turn, + book: config.cache.cooldowns, + nowSeconds: config.nowSeconds, + sleep: config.sleep ?? sleep, + isLogin: (candidate) => loginName(candidate) !== null, + attempt: async (candidate) => { + // The upstream's own answer carries the wait hint, whatever shape the client gets back. + let hint: Promise = Promise.resolve(null); + const watched: AppConfig = { + ...config, + upstream: async (upstreamRequest) => { + const upstream = await config.upstream(upstreamRequest); + if (upstream.status === 429 || upstream.status === 404 || upstream.status >= 500) { + const small = Number(upstream.headers.get("content-length") ?? "0") <= HINT_BODY_LIMIT; + const body = upstream.status === 429 && small ? upstream.clone().text().catch(() => "") : Promise.resolve(""); + hint = body.then((text) => retryHint(upstream.headers, text.slice(0, HINT_BODY_LIMIT), config.nowSeconds())); + } + return upstream; + }, + }; + const response = await serveWith(candidate, watched); noteAttempt(trace, config, candidate, response.status); - if (isLast || !RETRY_STATUSES.has(response.status)) return response; - await response.body?.cancel(); - last = new Response(null, { status: response.status }); - continue; - } - const upstream = await config.upstream(providerRequest(request, url, route, candidate.credential, candidate.model, body)).catch(() => null); - noteAttempt(trace, config, candidate, upstream?.status ?? 502); - if (upstream !== null && (isLast || !RETRY_STATUSES.has(upstream.status))) return downstreamResponse(upstream); - if (upstream !== null) { - await upstream.body?.cancel(); - last = new Response(null, { status: upstream.status }); - } - } - return formatError(route.format, last?.status ?? 502, "api_error", "No provider could serve the request right now."); + return { response, hint: response.ok ? null : await hint }; + }, + report: (event) => share(config, event), + cooling: (seconds) => { + const response = formatError(route.format, 429, "rate_limit_error", `Every credential for model ${model} is cooling down after errors or rate limits. Try again in ${Math.max(1, seconds)} s, or reset the cooldowns in the dashboard.`); + response.headers.set("retry-after", String(Math.max(1, seconds))); + return response; + }, + }); } async function modelList(request: Request, url: URL, config: AppConfig): Promise { @@ -444,17 +527,18 @@ async function modelList(request: Request, url: URL, config: AppConfig): Promise return Response.json({ ...body, data: [...(body.data ?? []), ...extra] }, { headers }); } -// Sends the isolate's usage records to the Durable Object, a few seconds after the first one so -// that requests close together share one call. -function remember(config: AppConfig, record: UsageRecord): void { +// Sends the isolate's usage records and cooldowns to the Durable Object, a few seconds after the +// first one so that requests close together share one call. +function flushSoon(config: AppConfig): void { const cache = config.cache; if (config.waitUntil === undefined) return; - cache.usage.push(record); const flush = async () => { const batch = cache.usage.splice(0); + const cooling = cache.cooling.splice(0); if (batch.length > 0) await config.backend.record(batch).catch(() => undefined); + if (cooling.length > 0) await config.backend.cool(cooling).catch(() => undefined); }; - if (cache.usage.length >= USAGE_FLUSH_COUNT) { + if (cache.usage.length + cache.cooling.length >= USAGE_FLUSH_COUNT) { config.waitUntil(flush()); return; } @@ -469,6 +553,18 @@ function remember(config: AppConfig, record: UsageRecord): void { ); } +function remember(config: AppConfig, record: UsageRecord): void { + if (config.waitUntil === undefined) return; + config.cache.usage.push(record); + flushSoon(config); +} + +function share(config: AppConfig, event: CooldownEvent): void { + if (config.waitUntil === undefined) return; + config.cache.cooling.push(event); + flushSoon(config); +} + function requestId(): string { return crypto.randomUUID().replaceAll("-", "").slice(0, 8); } diff --git a/free/src/management.ts b/free/src/management.ts index 5d9cc1d..584314f 100644 --- a/free/src/management.ts +++ b/free/src/management.ts @@ -1,6 +1,7 @@ import { LOGIN_PROVIDERS, type Account, type LoginProvider } from "./account.ts"; import { Config, ConfigError, configTag, ifMatchSatisfied } from "./config.ts"; import { apiKeyKey, apiKeyProvider, loginKey, storedKey, type UsageStats } from "./usage.ts"; +import type { Cooldown } from "./routing.ts"; import type { CredentialFile } from "./vault.ts"; // CPA's v8 management API (the subset the official dashboard uses), answered from the Durable @@ -64,13 +65,41 @@ async function configRoute(request: Request, path: string[], yaml: boolean, acco return json({ status: "ok", "config-version": 8 }, 200, { etag: await configTag(next.text) }); } +// The status messages CPA gives a credential while it cools down. +const COOLDOWN_MESSAGES: Record = { + unauthorized: "unauthorized", + payment_required: "payment_required", + not_found: "not_found", + quota: "quota exhausted", + transient_error: "transient upstream error", +}; + function iso(seconds: number): string { return new Date(seconds * 1000).toISOString(); } -function credentialEntry(file: CredentialFile, nowSeconds: number, usage: UsageStats): Record { +// CPA's cooldown view: the active ones, the credential-wide one first, then models by name. +function cooldownViews(cooldowns: readonly Cooldown[], nowSeconds: number): Record[] { + return cooldowns + .filter((cooldown) => cooldown.until > nowSeconds) + .sort((a, b) => (a.model === "" ? -1 : b.model === "" ? 1 : a.model.localeCompare(b.model))) + .map((cooldown) => { + const view: Record = { scope: cooldown.model === "" ? "credential" : "model" }; + if (cooldown.model !== "") view.model_key = cooldown.model; + view.reason = cooldown.reason; + view.retry_at = iso(cooldown.until); + view.remaining_seconds = cooldown.until - nowSeconds; + if (cooldown.reason === "quota") view.backoff_level = cooldown.level; + if (cooldown.status >= 400 && cooldown.status <= 599) view.http_status = cooldown.status; + return view; + }); +} + +function credentialEntry(file: CredentialFile, nowSeconds: number, usage: UsageStats, cooldowns: readonly Cooldown[] = []): Record { const expires = typeof file.data.expired === "string" ? Date.parse(file.data.expired) / 1000 : Number.NaN; const expired = Number.isFinite(expires) && expires <= nowSeconds && file.statusMessage === "token expired"; + const views = cooldownViews(cooldowns, nowSeconds); + const cooling = views[0]?.scope === "credential" ? cooldowns.find((cooldown) => cooldown.model === "" && cooldown.until > nowSeconds) : undefined; const entry: Record = { id: file.name, auth_index: file.name, @@ -78,10 +107,10 @@ function credentialEntry(file: CredentialFile, nowSeconds: number, usage: UsageS type: file.provider, provider: file.provider, label: typeof file.data.email === "string" ? file.data.email : file.name, - status: file.disabled ? "disabled" : expired ? "error" : "active", - status_message: file.statusMessage, + status: file.disabled ? "disabled" : expired || cooling !== undefined ? "error" : "active", + status_message: file.statusMessage || (cooling === undefined ? "" : Object.hasOwn(COOLDOWN_MESSAGES, cooling.reason) ? COOLDOWN_MESSAGES[cooling.reason] : "request failed"), disabled: file.disabled, - unavailable: expired, + unavailable: expired || cooling !== undefined, runtime_only: false, source: "file", path: file.name, @@ -90,11 +119,12 @@ function credentialEntry(file: CredentialFile, nowSeconds: number, usage: UsageS failed: usage.failed, recent_requests: usage.recent_requests, quota: null, - cooldowns: null, + cooldowns: views, created_at: iso(file.createdAt), modtime: iso(file.updatedAt), updated_at: iso(file.updatedAt), }; + if (cooling !== undefined) entry.next_retry_after = iso(cooling.until); for (const field of ["email", "project_id", "note", "priority", "prefix", "proxy_url", "last_refresh"]) { if (file.data[field] !== undefined && file.data[field] !== "") entry[field] = file.data[field]; } @@ -105,7 +135,8 @@ function listCredentials(url: URL, account: Account): Response { const name = url.searchParams.get("name")?.trim() ?? ""; const files = account.vault.credentialFiles().filter((file) => name === "" || file.name === name); const now = account.nowSeconds(); - const entries = files.map((file) => credentialEntry(file, now, account.vault.usage.stats(loginKey(file.name), now))); + const cooldowns = account.vault.cooldowns(now); + const entries = files.map((file) => credentialEntry(file, now, account.vault.usage.stats(loginKey(file.name), now), cooldowns.filter((cooldown) => cooldown.id === loginKey(file.name)))); const observedAt = new Date().toISOString(); if (!url.searchParams.has("page") && !url.searchParams.has("page_size")) return json({ observed_at: observedAt, files: entries }); const page = Math.max(1, Number(url.searchParams.get("page") ?? "1") || 1); @@ -248,6 +279,18 @@ async function patchCredential(request: Request, account: Account, fields: boole return json({ status: "ok" }); } +// CPA's reset of one credential's cooldowns, by its auth index (here the login's file name). +async function resetCooldowns(request: Request, account: Account): Promise { + const body = await readJson(request); + if (!body.ok || typeof body.value !== "object" || body.value === null) return failure(400, "invalid request body"); + const index = (body.value as { auth_index?: unknown }).auth_index; + const name = typeof index === "string" ? index.trim() : ""; + if (name === "") return failure(400, "auth_index is required"); + if (account.vault.credentialFile(name) === null) return failure(404, "auth not found"); + const models = account.vault.clearCooldowns(loginKey(name)).filter((model) => model !== ""); + return json({ status: "ok", auth_index: name, models }); +} + async function anthropicModels(account: Account): Promise<{ id: string; display_name: string }[]> { const token = await account.token(); if (!token.ok) return []; @@ -385,7 +428,7 @@ export async function handleManagement(request: Request, account: Account): Prom case "GET server/latest-version": return json({ "latest-version": GATEWAY_VERSION }); case "POST routing/cooldown/reset": - return json({ status: "ok" }); + return resetCooldowns(request, account); default: if (request.method === "GET" && head === "routing" && rest[0] === "model-definitions" && rest.length === 2) { const channel = (rest[1] as string).toLowerCase(); diff --git a/free/src/providers.ts b/free/src/providers.ts index f8ffbaf..6055731 100644 --- a/free/src/providers.ts +++ b/free/src/providers.ts @@ -1,3 +1,5 @@ +import type { Cooldown, RoutingSettings } from "./routing.ts"; + // The API-key upstreams configured under CPA's `api-keys` config section, and how a requested // model is matched to them. Each provider holds named groups (one endpoint, shared settings) // and a list of keys; a key may override the group's settings, as in CPA. @@ -50,6 +52,20 @@ export interface UpstreamCredential { readonly headers: Readonly>; readonly models: readonly ModelMapping[]; readonly excluded: readonly string[]; + // CPA's per-key overrides; null inherits the routing section's value. + readonly disableCooling: boolean | null; + readonly requestRetry: number | null; +} + +// One enabled subscription login, with the routing fields CPA reads from its auth file. +export interface LoginEntry { + readonly name: string; + readonly provider: "claude" | "codex"; + readonly priority: number; + readonly weight: number; + readonly prefix: string; + readonly disableCooling: boolean | null; + readonly requestRetry: number | null; } export interface RoutingTable { @@ -58,6 +74,12 @@ export interface RoutingTable { readonly claudeOAuth: boolean; // Whether at least one ChatGPT (Codex) OAuth login is enabled. readonly codexOAuth: boolean; + // Each enabled login. Left out, the Durable Object picks a login for each token in turn. + readonly logins?: readonly LoginEntry[]; + // CPA's routing section (strategy, retry rounds, cooldowns); defaults when left out. + readonly settings?: RoutingSettings; + // Cooldowns the Durable Object has heard of from every isolate. + readonly cooldowns?: readonly Cooldown[]; } export interface Candidate { @@ -103,6 +125,26 @@ function number(value: unknown, fallback: number): number { return typeof value === "number" && Number.isFinite(value) ? value : fallback; } +// CPA reads weight from config as a number and from auth files as a number or a numeric string. +// Non-positive weights take the credential out of weighted round-robin. +export function weightOf(value: unknown): number { + const parsed = typeof value === "string" && value.trim() !== "" ? Number(value) : value; + return Math.max(0, Math.min(1_000_000, Math.floor(number(parsed, 1)))); +} + +export function flagOf(value: unknown): boolean | null { + if (typeof value === "boolean") return value; + if (value === "true") return true; + if (value === "false") return false; + return null; +} + +// A negative or missing request-retry inherits the routing section's value. +export function retryOf(value: unknown): number | null { + const parsed = typeof value === "string" && value.trim() !== "" ? Number(value) : value; + return typeof parsed === "number" && Number.isFinite(parsed) && parsed >= 0 ? Math.floor(parsed) : null; +} + // Reads `api-keys` from the config tree into one flat list of usable credentials. export function upstreamCredentials(config: unknown): UpstreamCredential[] { const root = isTree(config) ? config["api-keys"] : undefined; @@ -127,12 +169,14 @@ export function upstreamCredentials(config: unknown): UpstreamCredential[] { group: groupName, baseUrl, apiKey, - weight: Math.max(1, Math.min(1_000_000, Math.floor(number(key.weight, 1)))), + weight: weightOf(key.weight), priority: number(inherit(key, group, "priority"), 0), prefix: text(inherit(key, group, "prefix")), headers: headersOf(inherit(key, group, "headers")), models: modelsOf(inherit(key, group, "models")), excluded: listOf(inherit(key, group, "excluded-models")), + disableCooling: flagOf(inherit(key, group, "disable-cooling")), + requestRetry: retryOf(inherit(key, group, "request-retry")), }); }); }); @@ -171,39 +215,51 @@ function upstreamModel(credential: UpstreamCredential, model: string): string | return target !== null && !excluded(target, credential.excluded) ? target : null; } -// Credentials that can serve the model, highest priority first. "prefix/model" picks the -// credentials carrying that prefix, as in CPA. -export function candidates(table: RoutingTable, requested: string): Candidate[] { +// "prefix/model" targets the credentials and logins carrying that prefix, as in CPA; any other +// model name is matched as it is. +export function splitModel(table: RoutingTable, requested: string): { prefix: string; model: string } { const slash = requested.indexOf("/"); const prefix = slash > 0 ? requested.slice(0, slash) : ""; - const prefixed = prefix !== "" && table.credentials.some((credential) => credential.prefix === prefix); - const model = prefixed ? requested.slice(slash + 1) : requested; + if (prefix === "") return { prefix: "", model: requested }; + const known = table.credentials.some((credential) => credential.prefix === prefix) || (table.logins ?? []).some((login) => login.prefix === prefix); + return known ? { prefix, model: requested.slice(slash + 1) } : { prefix: "", model: requested }; +} + +// Credentials that can serve the model, highest priority first. +export function candidates(table: RoutingTable, requested: string): Candidate[] { + const { prefix, model } = splitModel(table, requested); const found: Candidate[] = []; for (const credential of table.credentials) { - if (prefixed && credential.prefix !== prefix) continue; + if (prefix !== "" && credential.prefix !== prefix) continue; const target = upstreamModel(credential, model); if (target !== null) found.push({ credential, model: target }); } return found.sort((a, b) => b.credential.priority - a.credential.priority); } -// Weighted round-robin within the highest priority, then the rest in priority order as -// fallbacks. `turn` is a counter the caller keeps per isolate. -export function ordered(found: readonly Candidate[], turn: number): Candidate[] { - if (found.length <= 1) return [...found]; - const top = found.filter((candidate) => candidate.credential.priority === found[0]?.credential.priority); - const total = top.reduce((sum, candidate) => sum + candidate.credential.weight, 0); +export type Strategy = "round-robin" | "weighted-round-robin" | "fill-first"; + +// CPA's strategies, within the highest priority: round-robin gives each credential a turn, +// weighted-round-robin gives turns by weight (leaving out weight 0), and fill-first keeps to the +// first until it fails. The lower priorities follow in order, as fallbacks. `found` must be +// sorted by priority, and `turn` is a counter the caller keeps per isolate. +export function ordered(found: readonly Candidate[], turn: number, strategy: Strategy = "round-robin"): Candidate[] { + const pool = strategy === "weighted-round-robin" ? found.filter((candidate) => candidate.credential.weight > 0) : [...found]; + if (pool.length <= 1 || strategy === "fill-first") return pool; + const top = pool.filter((candidate) => candidate.credential.priority === pool[0]?.credential.priority); + const share = (candidate: Candidate) => (strategy === "weighted-round-robin" ? candidate.credential.weight : 1); + const total = top.reduce((sum, candidate) => sum + share(candidate), 0); let slot = turn % total; let first = 0; for (const [index, candidate] of top.entries()) { - if (slot < candidate.credential.weight) { + if (slot < share(candidate)) { first = index; break; } - slot -= candidate.credential.weight; + slot -= share(candidate); } const rotated = [...top.slice(first), ...top.slice(0, first)]; - return [...rotated, ...found.slice(top.length)]; + return [...rotated, ...pool.slice(top.length)]; } // Every model name clients can ask for by name, for the model lists. diff --git a/free/src/routing.ts b/free/src/routing.ts new file mode 100644 index 0000000..56d53a5 --- /dev/null +++ b/free/src/routing.ts @@ -0,0 +1,329 @@ +import { ordered, type Candidate, type RoutingTable, type Strategy } from "./providers.ts"; + +// CPA's request routing across credentials: the strategy that picks the next credential, the +// failover to the next one when a credential fails, retry rounds, and the cooldowns that keep a +// failing or rate-limited credential out of rotation for a while. Durations follow CPA's auth +// conductor; cooldowns are kept per isolate and shared through the Durable Object. + +export interface RoutingSettings { + readonly strategy: Strategy; + // Extra rounds after the first has tried every eligible credential. + readonly requestRetry: number; + // How many credentials one round may try; 0 for all. + readonly maxRetryCredentials: number; + // The longest wait, in seconds, for a cooldown to end before another round; 0 never waits. + readonly maxRetryInterval: number; + readonly disableCooling: boolean; + // Seconds a transient upstream error cools a credential: 0 for CPA's 60, negative for none. + readonly transientCooldownSeconds: number; +} + +// CPA's values when the config leaves a field out. +export const DEFAULT_ROUTING: RoutingSettings = { + strategy: "round-robin", + requestRetry: 0, + maxRetryCredentials: 0, + maxRetryInterval: 0, + disableCooling: false, + transientCooldownSeconds: 0, +}; + +// Upstream calls one request may make in all, well inside the 50 subrequests a Worker gets on +// the Free plan (the Durable Object and token calls count too). +export const MAX_ATTEMPTS = 10; +// Rounds wait at most this long in all, whatever max-retry-interval says. +const MAX_TOTAL_WAIT_SECONDS = 120; + +// A credential (`model` "") or one of its models, kept out of rotation until `until`. +export interface Cooldown { + readonly id: string; + readonly model: string; + readonly until: number; + // CPA's reason codes: unauthorized, payment_required, not_found, quota, transient_error. + readonly reason: string; + readonly status: number; + // Steps up the quota backoff (1 s, 2 s, 4 s ... 30 min) until the credential succeeds again. + readonly level: number; +} + +// What an isolate tells the Durable Object: a new cooldown, or that a credential succeeded. +export type CooldownEvent = Cooldown | { readonly id: string; readonly clear: true }; + +const MINUTE = 60; +const HOUR = 60 * MINUTE; +const AUTH_COOLDOWN = 30 * MINUTE; +const NOT_FOUND_COOLDOWN = 12 * HOUR; +const TRANSIENT_COOLDOWN = MINUTE; +const QUOTA_MAX = 30 * MINUTE; +const QUOTA_FLOOR = 10; +const TRANSIENT_STATUSES = new Set([408, 500, 502, 503, 504, 520, 521, 522, 523, 524, 525, 526, 529]); +// After these, the next credential is tried in the same round. +const FAILOVER_STATUSES = new Set([401, 402, 403, 404, 429, ...TRANSIENT_STATUSES]); +// After these, a further round may start once the first has run out of credentials (as in CPA). +const ROUND_STATUSES = new Set([403, 408, 429, 500, 502, 503, 504, 529]); +// Local entries stay this long so a request soon after doesn't depend on the shared copy. +const LOCAL_SECONDS = 2 * MINUTE; +const LOCAL_LIMIT = 1000; + +type Tree = Record; + +function isTree(value: unknown): value is Tree { + return typeof value === "object" && value !== null && !Array.isArray(value); +} + +function integer(value: unknown, fallback: number): number { + return typeof value === "number" && Number.isFinite(value) ? Math.trunc(value) : fallback; +} + +function strategyOf(value: unknown): Strategy { + const name = typeof value === "string" ? value.trim().toLowerCase() : ""; + if (["weighted-round-robin", "weightedroundrobin", "wrr"].includes(name)) return "weighted-round-robin"; + if (["fill-first", "fillfirst", "ff"].includes(name)) return "fill-first"; + return "round-robin"; +} + +// Reads CPA's v8 `routing` section, and the old top-level spellings CPA still accepts. +export function routingSettings(config: unknown): RoutingSettings { + const root = isTree(config) ? config : {}; + const routing = isTree(root.routing) ? root.routing : {}; + const retry = isTree(routing.retry) ? routing.retry : {}; + const cooldown = isTree(routing.cooldown) ? routing.cooldown : {}; + return { + strategy: strategyOf(routing.strategy), + requestRetry: Math.max(0, Math.min(10, integer(retry["request-retry"] ?? root["request-retry"], DEFAULT_ROUTING.requestRetry))), + maxRetryCredentials: Math.max(0, integer(retry["max-retry-credentials"] ?? root["max-retry-credentials"], 0)), + maxRetryInterval: Math.max(0, integer(retry["max-retry-interval"] ?? root["max-retry-interval"], 0)), + disableCooling: (cooldown["disable-cooling"] ?? root["disable-cooling"]) === true, + transientCooldownSeconds: integer(cooldown["transient-error-cooldown-seconds"] ?? root["transient-error-cooldown-seconds"], 0), + }; +} + +export function cooldownKey(id: string, model: string): string { + return `${id}\n${model}`; +} + +function parseSeconds(value: string, now: number): number | null { + const text = value.trim(); + if (/^\d+(\.\d+)?$/.test(text)) return Math.ceil(Number(text)); + const date = Date.parse(text); + return Number.isFinite(date) ? Math.max(0, Math.ceil(date / 1000 - now)) : null; +} + +// How long an upstream asked to wait, in seconds, from what CPA reads: Retry-After, Anthropic's +// unified rate-limit resets for a rejected window, Codex's usage_limit_reached reset, and +// Gemini's RetryInfo delay. +export function retryHint(headers: Headers, body: string, now: number): number | null { + const hints: number[] = []; + const retryAfter = headers.get("retry-after"); + if (retryAfter !== null) { + const seconds = parseSeconds(retryAfter, now); + if (seconds !== null) hints.push(seconds); + } + for (const window of ["", "5h-", "7d-"]) { + if (headers.get(`anthropic-ratelimit-unified-${window}status`)?.trim().toLowerCase() !== "rejected") continue; + const reset = Number(headers.get(`anthropic-ratelimit-unified-${window}reset`) ?? ""); + if (Number.isFinite(reset) && reset > now) hints.push(Math.ceil(reset - now)); + } + if (body !== "") { + let parsed: unknown; + try { + parsed = JSON.parse(body); + } catch { + parsed = null; + } + const root = Array.isArray(parsed) ? parsed[0] : parsed; + const error = isTree(root) && isTree(root.error) ? root.error : isTree(root) ? root : {}; + if (error.type === "usage_limit_reached") { + const resetsAt = integer(error.resets_at, 0); + const resetsIn = integer(error.resets_in_seconds, 0); + if (resetsAt > now) hints.push(resetsAt - now); + else if (resetsIn > 0) hints.push(resetsIn); + } + const details = Array.isArray(error.details) ? error.details : []; + for (const detail of details) { + if (!isTree(detail) || typeof detail.retryDelay !== "string") continue; + const delay = /^(\d+(?:\.\d+)?)s$/.exec(detail.retryDelay.trim()); + if (delay !== null) hints.push(Math.ceil(Number(delay[1]))); + } + } + return hints.length === 0 ? null : Math.max(...hints); +} + +export interface Failure { + readonly status: number; + // Seconds the upstream asked to wait, if it said. + readonly hint: number | null; +} + +interface Cooled { + readonly kind: Candidate["credential"]["kind"]; + readonly login: boolean; + readonly disableCooling: boolean; +} + +// The cooldown CPA sets after a failure, or null when the credential stays in rotation. +// `recorded` finds the last cooldown of the credential ("") or a model, for the quota backoff. +export function cooldownAfter(id: string, model: string, cooled: Cooled, failure: Failure, recorded: (model: string) => Cooldown | null, settings: RoutingSettings, now: number): Cooldown | null { + if (cooled.disableCooling) return null; + const { status, hint } = failure; + const transient = settings.transientCooldownSeconds < 0 ? 0 : settings.transientCooldownSeconds || TRANSIENT_COOLDOWN; + const make = (scope: "credential" | "model", seconds: number, reason: string, level = 0): Cooldown | null => + seconds > 0 ? { id, model: scope === "credential" ? "" : model, until: now + seconds, reason, status, level } : null; + // A login's token is refreshed after a 401, so it is only rested like a transient error. + if (status === 401) return cooled.login ? make("credential", transient, "unauthorized") : make("credential", AUTH_COOLDOWN, "unauthorized"); + if (status === 402 || status === 403) return make("credential", AUTH_COOLDOWN, "payment_required"); + if (status === 404) return make("model", hint !== null && hint > 0 ? hint : NOT_FOUND_COOLDOWN, "not_found"); + if (status === 429) { + // Claude and ChatGPT limits are per account; Gemini and OpenAI-compatible ones per model. + const scope = cooled.kind === "claude" || cooled.kind === "codex" ? "credential" : "model"; + const last = recorded(scope === "credential" ? "" : model); + const previous = last?.reason === "quota" ? last : null; + const level = previous?.level ?? 0; + if (hint !== null) return make(scope, Math.max(QUOTA_FLOOR, hint), "quota", level); + // Failures while an earlier quota window is still open share it, so a burst steps once. + if (previous !== null && previous.until > now) return make(scope, previous.until - now, "quota", level); + const seconds = Math.min(QUOTA_MAX, 2 ** level); + return make(scope, seconds, "quota", seconds >= QUOTA_MAX ? level : level + 1); + } + if (TRANSIENT_STATUSES.has(status)) return make("model", hint !== null && hint > 0 ? hint : transient, "transient_error"); + return null; +} + +// This isolate's cooldowns, on top of the Durable Object's copy in the routing table. +export class CooldownBook { + private readonly local = new Map(); + // Credentials that succeeded here since the table was read, whose shared entries are stale. + private cleared = new Set(); + private clearedFor: RoutingTable | null = null; + + private shared(table: RoutingTable, id: string): readonly Cooldown[] { + if (this.clearedFor !== table) { + this.clearedFor = table; + this.cleared = new Set(); + } + return this.cleared.has(id) ? [] : (table.cooldowns ?? []).filter((cooldown) => cooldown.id === id); + } + + // The latest cooldown recorded for a credential ("") or one of its models, active or not. + find(table: RoutingTable, id: string, model: string): Cooldown | null { + const shared = this.shared(table, id).find((cooldown) => cooldown.model === model) ?? null; + const local = this.local.get(cooldownKey(id, model))?.cooldown ?? null; + if (shared === null || local === null) return local ?? shared; + return local.until >= shared.until ? local : shared; + } + + // When the candidate is free again; at or before `now` means it is free. + freeAt(table: RoutingTable, candidate: Candidate, now: number): number { + const id = candidate.credential.id; + return Math.max(this.find(table, id, "")?.until ?? 0, this.find(table, id, candidate.model)?.until ?? 0, now); + } + + note(cooldown: Cooldown, now: number): void { + this.prune(now); + if (this.local.size >= LOCAL_LIMIT) this.local.clear(); + this.local.set(cooldownKey(cooldown.id, cooldown.model), { cooldown, seenAt: now }); + } + + // A success ends the credential's backoff. Reports whether anything was recorded for it, so the + // Durable Object is told once. + succeed(table: RoutingTable, id: string): boolean { + let found = this.shared(table, id).length > 0; + this.cleared.add(id); + for (const key of this.local.keys()) { + if (!key.startsWith(`${id}\n`)) continue; + this.local.delete(key); + found = true; + } + return found; + } + + clear(): void { + this.local.clear(); + this.cleared = new Set(); + this.clearedFor = null; + } + + private prune(now: number): void { + for (const [key, entry] of this.local) if (entry.cooldown.until <= now && entry.seenAt + LOCAL_SECONDS <= now) this.local.delete(key); + } +} + +export interface Attempted { + readonly response: Response; + // Seconds the upstream asked to wait, read from the upstream's own answer. + readonly hint: number | null; +} + +export interface RouteOptions { + readonly table: RoutingTable; + readonly usable: readonly Candidate[]; + readonly turn: number; + readonly book: CooldownBook; + readonly nowSeconds: () => number; + readonly sleep: (ms: number) => Promise; + readonly isLogin: (candidate: Candidate) => boolean; + readonly attempt: (candidate: Candidate) => Promise; + readonly report: (event: CooldownEvent) => void; + // The answer when every credential is cooling down, given the seconds until one is free. + readonly cooling: (seconds: number) => Response; +} + +// A failed answer kept whole, so the client gets the last upstream error if nothing succeeds. +async function kept(response: Response): Promise { + const body = await response.arrayBuffer().catch(() => new ArrayBuffer(0)); + return new Response(body, { status: response.status, statusText: response.statusText, headers: response.headers }); +} + +// Tries the candidates as CPA does: in strategy order within each round, skipping cooling ones, +// moving on after a credential failure, and starting another round while request-retry allows. +export async function route(options: RouteOptions): Promise { + const { table, book } = options; + const settings = table.settings ?? DEFAULT_ROUTING; + const sorted = [...options.usable].sort((a, b) => b.credential.priority - a.credential.priority); + const retries = (candidate: Candidate) => candidate.credential.requestRetry ?? settings.requestRetry; + const rounds = Math.max(0, ...sorted.map(retries)); + let budget = MAX_ATTEMPTS; + let waited = 0; + let last: Response | null = null; + for (let round = 0; round <= rounds && budget > 0; round += 1) { + const admitted = round === 0 ? sorted : sorted.filter((candidate) => retries(candidate) >= round); + if (admitted.length === 0) break; + const free = () => admitted.filter((candidate) => book.freeAt(table, candidate, options.nowSeconds()) <= options.nowSeconds()); + let open = free(); + if (open.length === 0) { + const now = options.nowSeconds(); + const wait = Math.min(...admitted.map((candidate) => book.freeAt(table, candidate, now))) - now; + if (settings.maxRetryInterval <= 0 || wait > settings.maxRetryInterval || waited + wait > MAX_TOTAL_WAIT_SECONDS) return last ?? options.cooling(wait); + waited += wait; + await options.sleep(wait * 1000); + open = free(); + if (open.length === 0) return last ?? options.cooling(wait); + } + let list = ordered(open, options.turn + round, settings.strategy); + if (list.length === 0) list = open; + if (settings.maxRetryCredentials > 0) list = list.slice(0, settings.maxRetryCredentials); + let status = 0; + for (const candidate of list) { + if (budget === 0) break; + budget -= 1; + const { response, hint } = await options.attempt(candidate); + status = response.status; + const id = candidate.credential.id; + if (status < 400) { + if (book.succeed(table, id)) options.report({ id, clear: true }); + return response; + } + if (!FAILOVER_STATUSES.has(status)) return response; + const now = options.nowSeconds(); + const cooled = { kind: candidate.credential.kind, login: options.isLogin(candidate), disableCooling: candidate.credential.disableCooling ?? settings.disableCooling }; + const cooldown = cooldownAfter(id, candidate.model, cooled, { status, hint }, (model) => book.find(table, id, model), settings, now); + if (cooldown !== null) { + book.note(cooldown, now); + options.report(cooldown); + } + last = await kept(response); + } + if (!ROUND_STATUSES.has(status)) break; + } + return last ?? options.cooling(0); +} diff --git a/free/src/store.ts b/free/src/store.ts index 5280ffa..a6a9ed9 100644 --- a/free/src/store.ts +++ b/free/src/store.ts @@ -3,6 +3,7 @@ import { Account, type Backend, type TokenResult } from "./account.ts"; import type { Env } from "./env.ts"; import { handleManagement } from "./management.ts"; import type { RoutingTable } from "./providers.ts"; +import type { CooldownEvent } from "./routing.ts"; import type { UsageRecord } from "./usage.ts"; import { Vault } from "./vault.ts"; @@ -21,11 +22,11 @@ export class Store extends DurableObject implements Backend { keyValid(hash: string): Promise { return this.account.keyValid(hash); } - token(): Promise { - return this.account.token(); + token(name?: string): Promise { + return this.account.token(name); } - codexToken(): Promise { - return this.account.codexToken(); + codexToken(name?: string): Promise { + return this.account.codexToken(name); } invalidate(accessToken: string): Promise { return this.account.invalidate(accessToken); @@ -42,4 +43,7 @@ export class Store extends DurableObject implements Backend { record(records: UsageRecord[]): Promise { return this.account.record(records); } + cool(events: CooldownEvent[]): Promise { + return this.account.cool(events); + } } diff --git a/free/src/vault.ts b/free/src/vault.ts index e62e086..c0cdf5f 100644 --- a/free/src/vault.ts +++ b/free/src/vault.ts @@ -1,5 +1,6 @@ import type { Credential } from "./claude-oauth.ts"; import { DEFAULT_CONFIG } from "./config.ts"; +import type { Cooldown } from "./routing.ts"; import { UsageStore } from "./usage.ts"; // The slice of Durable Object SQLite storage this file uses, so tests can back it with node:sqlite. @@ -37,6 +38,9 @@ export interface KeyRow { } const SESSION_SECONDS = 15 * 60; +// An ended cooldown is kept this long for its quota backoff step, unless the credential succeeds. +const COOLDOWN_KEPT_SECONDS = 24 * 60 * 60; +const COOLDOWN_LIMIT = 2000; const ATTEMPT_WINDOW = 60; const ATTEMPT_LIMIT = 5; @@ -58,6 +62,9 @@ export class Vault { sql.exec(`CREATE TABLE IF NOT EXISTS oauth_sessions ( state TEXT PRIMARY KEY, provider TEXT NOT NULL, verifier TEXT NOT NULL, created_at INTEGER NOT NULL, status TEXT NOT NULL DEFAULT 'wait', error TEXT NOT NULL DEFAULT '')`); + sql.exec(`CREATE TABLE IF NOT EXISTS cooldowns ( + id TEXT NOT NULL, model TEXT NOT NULL, until INTEGER NOT NULL, reason TEXT NOT NULL, status INTEGER NOT NULL, + level INTEGER NOT NULL, PRIMARY KEY (id, model))`); this.usage = new UsageStore(sql); this.migrateSingleLogin(); } @@ -193,6 +200,38 @@ export class Vault { return found; } + cooldowns(nowSeconds: number): Cooldown[] { + this.sql.exec("DELETE FROM cooldowns WHERE until < ?", nowSeconds - COOLDOWN_KEPT_SECONDS); + return this.sql + .exec("SELECT id, model, until, reason, status, level FROM cooldowns ORDER BY until DESC LIMIT ?", COOLDOWN_LIMIT) + .toArray() + .map((row) => ({ id: String(row.id), model: String(row.model), until: Number(row.until), reason: String(row.reason), status: Number(row.status), level: Number(row.level) })); + } + + // A later report never shortens a cooldown another isolate set. + saveCooldown(cooldown: Cooldown): void { + this.sql.exec( + `INSERT INTO cooldowns (id, model, until, reason, status, level) VALUES (?, ?, ?, ?, ?, ?) + ON CONFLICT (id, model) DO UPDATE SET until = max(until, excluded.until), reason = excluded.reason, status = excluded.status, + level = max(level, excluded.level)`, + cooldown.id, + cooldown.model, + cooldown.until, + cooldown.reason, + cooldown.status, + cooldown.level, + ); + } + + // Removes a credential's cooldowns, or every API key's (ids not starting "file:") with "keys". + clearCooldowns(id: string | "keys"): string[] { + const where = id === "keys" ? "id NOT LIKE 'file:%'" : "id = ?"; + const bindings = id === "keys" ? [] : [id]; + const models = this.sql.exec(`SELECT model FROM cooldowns WHERE ${where}`, ...bindings).toArray().map((row) => String(row.model)); + this.sql.exec(`DELETE FROM cooldowns WHERE ${where}`, ...bindings); + return models; + } + // Sign-in throttle: at most ATTEMPT_LIMIT tries per client per minute. attemptAllowed(client: string, nowSeconds: number): boolean { this.sql.exec("DELETE FROM login_attempts WHERE at <= ?", nowSeconds - ATTEMPT_WINDOW); diff --git a/free/test/app.test.ts b/free/test/app.test.ts index 1bad918..f2fde6b 100644 --- a/free/test/app.test.ts +++ b/free/test/app.test.ts @@ -1,9 +1,10 @@ import { describe, expect, it } from "vitest"; import { hashKey, type Backend, type TokenResult } from "../src/account.ts"; import { EdgeCache, handle, type AppConfig } from "../src/app.ts"; -import { upstreamCredentials, type RoutingTable } from "../src/providers.ts"; +import { upstreamCredentials, type LoginEntry, type RoutingTable } from "../src/providers.ts"; import { OAUTH_BETA } from "../src/proxy.ts"; import { SESSION_COOKIE, issueSession } from "../src/session.ts"; +import { routingSettings, type Cooldown, type CooldownEvent } from "../src/routing.ts"; import type { UsageRecord } from "../src/usage.ts"; const PASSWORD = "correct-horse-battery-staple-0123456789"; @@ -24,14 +25,18 @@ class FakeBackend implements Backend { this.calls.push("keyValid"); return this.hashes.has(hash); } - async token() { - this.calls.push("token"); - return this.tokenResult; + // Named logins get a token of their own, unless a test sets one. + named: Record = {}; + async token(name?: string) { + this.calls.push(name === undefined ? "token" : `token:${name}`); + if (name === undefined) return this.tokenResult; + return this.named[name] ?? { ok: true, accessToken: `access-${name}`, expiresAt: NOW + 3600, name }; } codexResult: TokenResult = { ok: true, accessToken: "chatgpt-1", expiresAt: NOW + 3600, accountId: "acct-1" }; - async codexToken() { - this.calls.push("codexToken"); - return this.codexResult; + async codexToken(name?: string) { + this.calls.push(name === undefined ? "codexToken" : `codexToken:${name}`); + if (name === undefined) return this.codexResult; + return this.named[name] ?? { ok: true, accessToken: `chatgpt-${name}`, expiresAt: NOW + 3600, accountId: `acct-${name}`, name }; } async invalidate(accessToken: string) { this.invalidated.push(accessToken); @@ -52,6 +57,10 @@ class FakeBackend implements Backend { async record(records: UsageRecord[]) { this.recorded.push(...records); } + cooled: CooldownEvent[] = []; + async cool(events: CooldownEvent[]) { + this.cooled.push(...events); + } } async function setup(upstream: (request: Request) => Promise = async () => new Response("{}")) { @@ -174,7 +183,7 @@ describe("API proxy", () => { const later = { ...config, nowSeconds: () => NOW + 61 }; await handle(messages({ "x-api-key": KEY }), later); expect(backend.calls.filter((call) => call === "token")).toHaveLength(2); - expect(config.cache.token?.value).toBe("access-2"); + expect(config.cache.tokens.get("claude:")?.value).toBe("access-2"); }); it("drops the token and tells the store when Anthropic refuses it", async () => { @@ -182,7 +191,7 @@ describe("API proxy", () => { const response = await handle(messages({ "x-api-key": KEY }), config); expect(response.status).toBe(401); expect(backend.invalidated).toEqual(["access-1"]); - expect(config.cache.token).toBeNull(); + expect(config.cache.tokens.size).toBe(0); }); it("explains a missing Claude login in Anthropic's error shape", async () => { @@ -262,10 +271,18 @@ describe("API-key providers", () => { expect(await seen[1]?.text()).toBe(' { "model" : "vendor/kimi-k2:free", "x": "\\"" }'); }); - it("shares requests between keys by weight", async () => { - const { config, seen } = await providersSetup(); - for (let turn = 0; turn < 4; turn += 1) await handle(post("/v1/chat/completions", { model: "deepseek-v3" }), config); - expect(seen.map((request) => request.headers.get("authorization"))).toEqual(["Bearer sk-router-1", "Bearer sk-router-1", "Bearer sk-router-2", "Bearer sk-router-1"]); + it("takes keys in turn, and by weight under weighted-round-robin, as CPA's strategies do", async () => { + const plain = await providersSetup(); + for (let turn = 0; turn < 4; turn += 1) await handle(post("/v1/chat/completions", { model: "deepseek-v3" }), plain.config); + expect(plain.seen.map((request) => request.headers.get("authorization"))).toEqual(["Bearer sk-router-1", "Bearer sk-router-2", "Bearer sk-router-1", "Bearer sk-router-2"]); + const weighted = await providersSetup(); + weighted.backend.table = { ...weighted.backend.table, settings: routingSettings({ routing: { strategy: "weighted-round-robin" } }) }; + for (let turn = 0; turn < 4; turn += 1) await handle(post("/v1/chat/completions", { model: "deepseek-v3" }), weighted.config); + expect(weighted.seen.map((request) => request.headers.get("authorization"))).toEqual(["Bearer sk-router-1", "Bearer sk-router-1", "Bearer sk-router-1", "Bearer sk-router-2"]); + const filling = await providersSetup(); + filling.backend.table = { ...filling.backend.table, settings: routingSettings({ routing: { strategy: "fill-first" } }) }; + for (let turn = 0; turn < 3; turn += 1) await handle(post("/v1/chat/completions", { model: "deepseek-v3" }), filling.config); + expect(filling.seen.map((request) => request.headers.get("authorization"))).toEqual(["Bearer sk-router-1", "Bearer sk-router-1", "Bearer sk-router-1"]); }); it("tries the next key when one is rate limited", async () => { @@ -417,9 +434,9 @@ describe("management API", () => { it("forgets the cached token after a credential change", async () => { const { config } = await setup(); - config.cache.token = { value: "old", name: "claude-a.json", expiresAt: NOW + 3600, cachedUntil: NOW + 60 }; + config.cache.tokens.set("claude:", { value: "old", name: "claude-a.json", accountId: "", expiresAt: NOW + 3600, cachedUntil: NOW + 60 }); await handle(await managementCall("/v8/management/credentials?name=claude-a.json", { authorization: `Bearer ${MANAGEMENT_KEY}` }, { method: "DELETE" }), config); - expect(config.cache.token).toBeNull(); + expect(config.cache.tokens.size).toBe(0); }); }); @@ -457,11 +474,11 @@ describe("OAuth callback", () => { it("needs sign-in, then hands the code to the store", async () => { const { config, backend } = await setup(); expect((await handle(at("/callback?code=c&state=s"), config)).status).toBe(401); - config.cache.token = { value: "old", name: "claude-a.json", expiresAt: NOW + 3600, cachedUntil: NOW + 60 }; + config.cache.tokens.set("claude:", { value: "old", name: "claude-a.json", accountId: "", expiresAt: NOW + 3600, cachedUntil: NOW + 60 }); const done = await handle(at("/callback?code=c&state=s", { headers: { cookie: await cookie() } }), config); expect(await done.text()).toContain("Login received"); expect(backend.managed[0]?.url).toBe(`${ORIGIN}/v8/management/oauth/callback?code=c&state=s`); - expect(config.cache.token).toBeNull(); + expect(config.cache.tokens.size).toBe(0); }); it("shows the store's error, escaped", async () => { @@ -528,6 +545,20 @@ describe("CPU budget", () => { console.log(`CPU per 600 KB rewritten request: ${gateway.toFixed(2)} ms in the gateway`); expect(gateway).toBeLessThan(8); }); + + it("reads a 600 KB Claude Code request once for routing across two logins, within budget", async () => { + const history = Array.from({ length: 400 }, (_, index) => ({ role: index % 2 ? "assistant" : "user", content: "x".repeat(1500) })); + const body = JSON.stringify({ model: "claude-sonnet-5-5", max_tokens: 32000, stream: true, messages: history }); + const { config, backend } = await setup(async (request) => { + await request.arrayBuffer(); + return new Response("event: message_stop\ndata: {}\n\n", { headers: { "content-type": "text/event-stream" } }); + }); + const login = (name: string): LoginEntry => ({ name, provider: "claude", priority: 0, weight: 1, prefix: "", disableCooling: null, requestRetry: null }); + backend.table = { credentials: [], claudeOAuth: true, codexOAuth: false, logins: [login("a.json"), login("b.json")] }; + const gateway = await gatewayCpuMs(() => messages({ "x-api-key": KEY }, body), config); + console.log(`CPU per 600 KB request across two logins: ${gateway.toFixed(2)} ms in the gateway`); + expect(gateway).toBeLessThan(5); + }); }); describe("ChatGPT (Codex) logins", () => { @@ -574,7 +605,7 @@ describe("ChatGPT (Codex) logins", () => { const refused = await codexSetup(async () => new Response("no", { status: 401 })); expect((await handle(post("/v1/responses", { model: "gpt-5.5", input: "hi", stream: true }), refused.config)).status).toBe(401); expect(refused.backend.invalidated).toEqual(["chatgpt-1"]); - expect(refused.config.cache.codexToken).toBeNull(); + expect(refused.config.cache.tokens.size).toBe(0); }); it("leaves Responses requests alone when no ChatGPT login is on", async () => { @@ -753,3 +784,127 @@ describe("Claude Code on ChatGPT logins and Codex keys", () => { expect(gateway).toBeLessThan(9); }); }); + +describe("routing across logins and keys", () => { + const managementCall = async (path: string, headers: Record, init: RequestInit = {}) => at(path, { ...init, headers: { cookie: await cookie(), ...headers } }); + const sse = (events: unknown[]) => new Response(events.map((event) => `data: ${JSON.stringify(event)}\n\n`).join(""), { headers: { "content-type": "text/event-stream" } }); + const login = (name: string, provider: LoginEntry["provider"] = "claude", fields: Partial = {}): LoginEntry => ({ name, provider, priority: 0, weight: 1, prefix: "", disableCooling: null, requestRetry: null, ...fields }); + + async function loginsSetup(upstream: (request: Request) => Promise, logins = [login("a.json"), login("b.json")], settings = {}) { + const fixture = await setup(upstream); + fixture.backend.table = { credentials: [], claudeOAuth: true, codexOAuth: false, logins, settings: routingSettings(settings) }; + return fixture; + } + + const tokenOf = (request: Request | undefined) => request?.headers.get("authorization"); + + it("moves on to the next Claude login when one is rate limited, and keeps it out until its cooldown ends", async () => { + const limited = async (request: Request) => (tokenOf(request) === "Bearer access-a.json" ? new Response('{"type":"error"}', { status: 429, headers: { "retry-after": "120" } }) : new Response("{}")); + const { config, backend, seen, settle } = await loginsSetup(limited); + expect((await handle(messages({ "x-api-key": KEY }), config)).status).toBe(200); + expect(seen.map(tokenOf)).toEqual(["Bearer access-a.json", "Bearer access-b.json"]); + // The next turn would be a.json's, but it is cooling down. + expect((await handle(messages({ "x-api-key": KEY }), config)).status).toBe(200); + expect(seen.map(tokenOf).slice(2)).toEqual(["Bearer access-b.json"]); + await settle(); + expect(backend.cooled).toEqual([{ id: "file:a.json", model: "", until: NOW + 120, reason: "quota", status: 429, level: 0 }]); + expect(backend.recorded[0]?.attempts).toEqual([ + { key: "file:a.json", ok: false }, + { key: "file:b.json", ok: true }, + ]); + // Another isolate learns of it from the routing table. + const other = { ...config, cache: new EdgeCache() }; + backend.table = { ...backend.table, cooldowns: backend.cooled.filter((event): event is Cooldown => !("clear" in event)) }; + await handle(messages({ "x-api-key": KEY }), other); + expect(seen.map(tokenOf).slice(3)).toEqual(["Bearer access-b.json"]); + }); + + it("answers 429 with Retry-After without calling upstream while every credential cools down", async () => { + const { config, seen } = await loginsSetup(async () => new Response("{}", { status: 503 })); + expect((await handle(messages({ "x-api-key": KEY }), config)).status).toBe(503); + expect(seen).toHaveLength(2); + const cooling = await handle(messages({ "x-api-key": KEY }), config); + expect(cooling.status).toBe(429); + expect(cooling.headers.get("retry-after")).toBe("60"); + expect(await cooling.json()).toMatchObject({ type: "error", error: { type: "rate_limit_error" } }); + expect(seen).toHaveLength(2); + }); + + it("returns the last upstream error whole when every credential fails", async () => { + const { config } = await loginsSetup(async (request) => Response.json({ type: "error", error: { message: `down for ${tokenOf(request)}` } }, { status: 529 })); + const response = await handle(messages({ "x-api-key": KEY }), config); + expect(response.status).toBe(529); + expect(await response.json()).toMatchObject({ error: { message: "down for Bearer access-b.json" } }); + }); + + it("hands a request error back without trying other credentials or cooling down", async () => { + const { config, backend, seen, settle } = await loginsSetup(async () => Response.json({ type: "error" }, { status: 400 })); + expect((await handle(messages({ "x-api-key": KEY }), config)).status).toBe(400); + expect(seen).toHaveLength(1); + await settle(); + expect(backend.cooled).toEqual([]); + }); + + it("waits for a short cooldown and runs another round when request-retry allows", async () => { + let clock = NOW; + const naps: number[] = []; + let calls = 0; + const { config, backend, seen } = await loginsSetup(async () => (calls++ === 0 ? new Response("{}", { status: 429, headers: { "retry-after": "1" } }) : new Response("{}")), [login("a.json")], { + routing: { retry: { "request-retry": 1, "max-retry-interval": 30 } }, + }); + const timed = { ...config, nowSeconds: () => clock, sleep: async (ms: number) => { naps.push(ms); clock += ms / 1000; } }; + // Two logins, so the body is read and the request can be sent again. + backend.table = { ...backend.table, logins: [login("a.json"), login("b.json", "claude", { requestRetry: 0 })] }; + backend.named["b.json"] = { ok: false, status: 503, message: "The login b.json has expired." }; + const response = await handle(messages({ "x-api-key": KEY }), timed); + expect(response.status).toBe(200); + // a.json is rate limited, b.json can't get a token, then a.json again after its 10 s floor. + expect(naps).toEqual([10_000]); + expect(seen.map(tokenOf)).toEqual(["Bearer access-a.json", "Bearer access-a.json"]); + }); + + it("doesn't wait longer than max-retry-interval", async () => { + const naps: number[] = []; + const { config, seen } = await loginsSetup(async () => new Response("{}", { status: 429, headers: { "retry-after": "600" } }), undefined, { routing: { retry: { "request-retry": 3, "max-retry-interval": 30 } } }); + const response = await handle(messages({ "x-api-key": KEY }), { ...config, sleep: async (ms: number) => void naps.push(ms) }); + expect(response.status).toBe(429); + expect(naps).toEqual([]); + expect(seen).toHaveLength(2); + }); + + it("keeps to higher priorities first and leaves disable-cooling logins in rotation", async () => { + const { config, seen } = await loginsSetup(async () => new Response("{}", { status: 500 }), [login("low.json"), login("high.json", "claude", { priority: 5, disableCooling: true })]); + await handle(messages({ "x-api-key": KEY }), config); + expect(seen.map(tokenOf)).toEqual(["Bearer access-high.json", "Bearer access-low.json"]); + await handle(messages({ "x-api-key": KEY }), config); + expect(seen.map(tokenOf).slice(2)).toEqual(["Bearer access-high.json"]); + }); + + it("sends prefix/model only to logins with that prefix, with the prefix taken off", async () => { + const { config, seen } = await loginsSetup(async () => new Response("{}"), [login("home.json"), login("work.json", "claude", { prefix: "team" })]); + const body = '{"model":"team/claude-sonnet-5-5","max_tokens":8,"messages":[]}'; + expect((await handle(messages({ "x-api-key": KEY }, body), config)).status).toBe(200); + expect(tokenOf(seen[0])).toBe("Bearer access-work.json"); + expect(await seen[0]?.json()).toMatchObject({ model: "claude-sonnet-5-5" }); + }); + + it("fails over between ChatGPT logins too", async () => { + const { config, backend, seen } = await setup(async (request) => + tokenOf(request) === "Bearer chatgpt-one.json" ? Response.json({ error: { type: "usage_limit_reached", resets_in_seconds: 3600 } }, { status: 429 }) : sse([{ type: "response.completed", response: { id: "r" } }]), + ); + backend.table = { credentials: [], claudeOAuth: false, codexOAuth: true, logins: [login("one.json", "codex"), login("two.json", "codex")] }; + expect((await handle(post("/v1/responses", { model: "gpt-5.5", input: "hi" }), config)).status).toBe(200); + expect(seen.map((request) => request.headers.get("chatgpt-account-id"))).toEqual(["acct-one.json", "acct-two.json"]); + expect(config.cache.cooldowns.find(backend.table, "file:one.json", "")).toMatchObject({ until: NOW + 3600, reason: "quota" }); + }); + + it("forgets this isolate's cooldowns after a dashboard change, such as a cooldown reset", async () => { + let refusals = 1; + const { config, seen } = await loginsSetup(async (request) => new Response("{}", { status: tokenOf(request) === "Bearer access-a.json" && refusals-- > 0 ? 403 : 200 })); + await handle(messages({ "x-api-key": KEY }), config); + await handle(await managementCall("/v8/management/routing/cooldown/reset", { authorization: `Bearer ${MANAGEMENT_KEY}` }, { method: "POST", body: '{"auth_index":"a.json"}' }), config); + await handle(messages({ "x-api-key": KEY }), config); + await handle(messages({ "x-api-key": KEY }), config); + expect(seen.map(tokenOf)).toEqual(["Bearer access-a.json", "Bearer access-b.json", "Bearer access-b.json", "Bearer access-a.json"]); + }); +}); diff --git a/free/test/management.test.ts b/free/test/management.test.ts index 71e0e8d..62398c4 100644 --- a/free/test/management.test.ts +++ b/free/test/management.test.ts @@ -143,3 +143,64 @@ describe("pages the free install doesn't have", () => { expect((await call("GET", "nothing/here")).status).toBe(404); }); }); + +describe("routing state", () => { + it("hands the Worker each login's routing fields, the routing settings and the cooldowns", async () => { + const { vault, account, call } = setup(); + saveLogin(vault, "claude-a.json", "a@example.com"); + vault.saveCredentialFile("claude-b.json", "claude", { ...claudeFileData({ accessToken: "at-b", refreshToken: "rt-b", expiresAt: NOW + 3600, email: "b@example.com" }), priority: "3", weight: 2, prefix: "work", disable_cooling: true, "request-retry": 2 }, NOW); + await call("PUT", "config.yaml", "routing:\n strategy: fill-first\n retry:\n request-retry: 2\n max-retry-interval: 30\n cooldown:\n transient-error-cooldown-seconds: -1\n"); + await account.cool([{ id: "file:claude-a.json", model: "", until: NOW + 90, reason: "quota", status: 429, level: 1 }]); + const table = await account.routing(); + expect(table.logins).toEqual([ + { name: "claude-a.json", provider: "claude", priority: 0, weight: 1, prefix: "", disableCooling: null, requestRetry: null }, + { name: "claude-b.json", provider: "claude", priority: 3, weight: 2, prefix: "work", disableCooling: true, requestRetry: 2 }, + ]); + expect(table.settings).toEqual({ strategy: "fill-first", requestRetry: 2, maxRetryCredentials: 0, maxRetryInterval: 30, disableCooling: false, transientCooldownSeconds: -1 }); + expect(table.cooldowns).toEqual([{ id: "file:claude-a.json", model: "", until: NOW + 90, reason: "quota", status: 429, level: 1 }]); + expect(await account.token("claude-b.json")).toMatchObject({ ok: true, accessToken: "at-b", name: "claude-b.json" }); + expect(await account.token("missing.json")).toMatchObject({ ok: false, status: 503 }); + }); + + it("keeps the longer of two reports, clears on success, and drops API-key cooldowns on a config edit", async () => { + const { account, call } = setup(); + await account.cool([ + { id: "file:x.json", model: "m", until: NOW + 600, reason: "not_found", status: 404, level: 0 }, + { id: "file:x.json", model: "m", until: NOW + 60, reason: "not_found", status: 404, level: 0 }, + { id: "claude/c/0", model: "", until: NOW + 60, reason: "unauthorized", status: 401, level: 0 }, + ]); + expect((await account.routing()).cooldowns?.map((cooldown) => [cooldown.id, cooldown.until])).toEqual([ + ["file:x.json", NOW + 600], + ["claude/c/0", NOW + 60], + ]); + await call("PATCH", "config/routing", '{"strategy":"round-robin"}'); + expect((await account.routing()).cooldowns?.map((cooldown) => cooldown.id)).toEqual(["file:x.json"]); + await account.cool([{ id: "file:x.json", clear: true }]); + expect((await account.routing()).cooldowns).toEqual([]); + }); + + it("shows cooldowns on the auth files list and resets them", async () => { + const { vault, account, call } = setup(); + saveLogin(vault, "claude-a.json"); + await account.cool([ + { id: "file:claude-a.json", model: "", until: NOW + 90, reason: "quota", status: 429, level: 2 }, + { id: "file:claude-a.json", model: "claude-x", until: NOW + 30, reason: "not_found", status: 404, level: 0 }, + ]); + const listed = (await (await call("GET", "credentials")).json()) as { files: Record[] }; + expect(listed.files[0]).toMatchObject({ + status: "error", + status_message: "quota exhausted", + unavailable: true, + next_retry_after: new Date((NOW + 90) * 1000).toISOString(), + cooldowns: [ + { scope: "credential", reason: "quota", retry_at: new Date((NOW + 90) * 1000).toISOString(), remaining_seconds: 90, backoff_level: 2, http_status: 429 }, + { scope: "model", model_key: "claude-x", reason: "not_found", remaining_seconds: 30, http_status: 404 }, + ], + }); + expect((await call("POST", "routing/cooldown/reset", "{}")).status).toBe(400); + expect((await call("POST", "routing/cooldown/reset", '{"auth_index":"nope.json"}')).status).toBe(404); + expect(await (await call("POST", "routing/cooldown/reset", '{"auth_index":"claude-a.json"}')).json()).toEqual({ status: "ok", auth_index: "claude-a.json", models: ["claude-x"] }); + const after = (await (await call("GET", "credentials")).json()) as { files: Record[] }; + expect(after.files[0]).toMatchObject({ status: "active", unavailable: false, cooldowns: [] }); + }); +}); diff --git a/free/test/routing.test.ts b/free/test/routing.test.ts new file mode 100644 index 0000000..f6d47eb --- /dev/null +++ b/free/test/routing.test.ts @@ -0,0 +1,73 @@ +import { describe, expect, it } from "vitest"; +import { cooldownAfter, DEFAULT_ROUTING, retryHint, routingSettings, type Cooldown } from "../src/routing.ts"; + +const NOW = 1_800_000_000; +const key = { kind: "openai-compatibility", login: false, disableCooling: false } as const; +const claudeKey = { kind: "claude", login: false, disableCooling: false } as const; +const none = () => null; + +function after(status: number, cooled: Parameters[2] = key, hint: number | null = null, previous: Cooldown | null = null, settings = DEFAULT_ROUTING) { + return cooldownAfter("k", "m", cooled, { status, hint }, () => previous, settings, NOW); +} + +describe("routing settings", () => { + it("reads CPA's v8 routing section and the old top-level spellings, with CPA's defaults", () => { + expect(routingSettings({})).toEqual(DEFAULT_ROUTING); + expect(routingSettings({ routing: { strategy: "WRR", retry: { "request-retry": 3, "max-retry-credentials": 2, "max-retry-interval": 30 }, cooldown: { "disable-cooling": true, "transient-error-cooldown-seconds": 5 } } })).toEqual({ + strategy: "weighted-round-robin", + requestRetry: 3, + maxRetryCredentials: 2, + maxRetryInterval: 30, + disableCooling: true, + transientCooldownSeconds: 5, + }); + expect(routingSettings({ "request-retry": 2, routing: { strategy: "ff" } })).toMatchObject({ strategy: "fill-first", requestRetry: 2 }); + expect(routingSettings({ routing: { strategy: "nonsense", retry: { "request-retry": 99 } } })).toMatchObject({ strategy: "round-robin", requestRetry: 10 }); + }); +}); + +describe("cooldowns", () => { + it("follows CPA's durations for each failure", () => { + expect(after(401)).toMatchObject({ model: "", until: NOW + 1800, reason: "unauthorized" }); + expect(after(401, { ...claudeKey, login: true })).toMatchObject({ model: "", until: NOW + 60 }); + expect(after(402)).toMatchObject({ model: "", until: NOW + 1800, reason: "payment_required" }); + expect(after(403)).toMatchObject({ model: "", until: NOW + 1800, reason: "payment_required" }); + expect(after(404)).toMatchObject({ model: "m", until: NOW + 12 * 3600, reason: "not_found" }); + expect(after(503)).toMatchObject({ model: "m", until: NOW + 60, reason: "transient_error" }); + expect(after(503, key, null, null, { ...DEFAULT_ROUTING, transientCooldownSeconds: 5 })).toMatchObject({ until: NOW + 5 }); + expect(after(503, key, null, null, { ...DEFAULT_ROUTING, transientCooldownSeconds: -1 })).toBeNull(); + expect(after(400)).toBeNull(); + expect(after(429, { ...key, disableCooling: true })).toBeNull(); + }); + + it("backs off on repeated quota errors, sharing an open window, and honours a stated wait", () => { + const first = after(429, claudeKey); + expect(first).toMatchObject({ model: "", until: NOW + 1, reason: "quota", level: 1 }); + const later = { ...(first as Cooldown), until: NOW - 1 }; + expect(after(429, claudeKey, null, later)).toMatchObject({ until: NOW + 2, level: 2 }); + expect(after(429, claudeKey, null, { ...later, until: NOW + 50 })).toMatchObject({ until: NOW + 50, level: 1 }); + expect(after(429, claudeKey, null, { ...later, level: 20 })).toMatchObject({ until: NOW + 1800, level: 20 }); + expect(after(429, claudeKey, 3)).toMatchObject({ until: NOW + 10 }); + expect(after(429, claudeKey, 600)).toMatchObject({ until: NOW + 600 }); + // Gemini and OpenAI-compatible limits are per model. + expect(after(429, key)).toMatchObject({ model: "m" }); + expect(cooldownAfter("k", "m", claudeKey, { status: 429, hint: null }, none, DEFAULT_ROUTING, NOW)).toMatchObject({ level: 1 }); + }); +}); + +describe("wait hints", () => { + it("reads Retry-After in seconds or as a date", () => { + expect(retryHint(new Headers({ "retry-after": "17" }), "", NOW)).toBe(17); + expect(retryHint(new Headers({ "retry-after": new Date((NOW + 40) * 1000).toUTCString() }), "", NOW)).toBe(40); + expect(retryHint(new Headers(), "", NOW)).toBeNull(); + }); + + it("reads Anthropic's rejected windows, Codex's usage limit and Gemini's retry delay", () => { + const anthropic = new Headers({ "anthropic-ratelimit-unified-status": "rejected", "anthropic-ratelimit-unified-reset": String(NOW + 3600), "anthropic-ratelimit-unified-7d-reset": String(NOW + 99_999) }); + expect(retryHint(anthropic, "", NOW)).toBe(3600); + expect(retryHint(new Headers(), JSON.stringify({ error: { type: "usage_limit_reached", resets_at: NOW + 500 } }), NOW)).toBe(500); + expect(retryHint(new Headers(), JSON.stringify({ error: { type: "usage_limit_reached", resets_in_seconds: 70 } }), NOW)).toBe(70); + expect(retryHint(new Headers(), JSON.stringify([{ error: { code: 429, details: [{ "@type": "type.googleapis.com/google.rpc.RetryInfo", retryDelay: "12.5s" }] } }]), NOW)).toBe(13); + expect(retryHint(new Headers(), "not json", NOW)).toBeNull(); + }); +});