diff --git a/src/server/responses/compact.ts b/src/server/responses/compact.ts index 6eb6b7754e..b14800fc7b 100644 --- a/src/server/responses/compact.ts +++ b/src/server/responses/compact.ts @@ -163,6 +163,10 @@ interface CompactHandoffRoute { */ const compactHandoffRoutes = new Map(); +export function clearCompactHandoffRoutesForTests(): void { + compactHandoffRoutes.clear(); +} + function pruneCompactHandoffRoutes(now: number): void { for (const [key, entry] of compactHandoffRoutes) { if (now - entry.lastUsedAt > COMPACT_HANDOFF_ROUTE_TTL_MS) compactHandoffRoutes.delete(key); @@ -740,6 +744,7 @@ export async function handleResponsesCompact( // actually happens, so every recorder call names the context that produced it. let outcomeCtx = authCtx; let upstream: Response; + let storedPool401ReplayAttempted = false; try { // Same connect timeout + keep-alive reset + transient-5xx recovery as /v1/responses — // compact hits the same ChatGPT host and must soft-avoid / clear affinity (#186). @@ -777,6 +782,7 @@ export async function handleResponsesCompact( ) { await upstream.body?.cancel().catch(() => undefined); const poolAuthCtx = authCtx.kind === "pool" ? authCtx : undefined; + storedPool401ReplayAttempted = poolAuthCtx !== undefined; const poolReplay = poolAuthCtx ? await refreshPoolCompactContext({ req, @@ -837,6 +843,7 @@ export async function handleResponsesCompact( // — reporting exhausted retries while another pool account sat idle (#913). if ( (upstream.status === 429 || upstream.status === 402) + && !storedPool401ReplayAttempted && usesCodexForwardPoolAuth(authCtx, route.provider) && !authCtx.fixedAccount && route.codexAccountMode @@ -947,7 +954,7 @@ export async function handleResponsesCompact( if (buffered.ok) { inspectResponseLogJson(logCtx, await buffered.clone().text()); forgetCompactHandoffRoute(req); - } else if (quotaFailure) { + } else if (quotaFailure && !storedPool401ReplayAttempted) { const fallbackModel = compactHandoffRoute(req, raw.model); if (fallbackModel && !req.signal.aborted) { const fallbackReq = new Request(req.url, { diff --git a/src/server/responses/core.ts b/src/server/responses/core.ts index f0c1291c2a..1c0fcd87f9 100644 --- a/src/server/responses/core.ts +++ b/src/server/responses/core.ts @@ -1433,6 +1433,8 @@ export interface HandleResponsesOptions { deferCodexResetDerivedCooldown?: boolean; /** 030-owned handoff when a child consumed the original failure under bounds. */ onConsumedComboFailure?: (failure: ConsumedComboFailure) => void; + /** A stored Pool credential was refreshed and its one allowed same-account replay was sent. */ + onStoredPool401ReplayDispatched?: () => void; /** Caller-owned for Chat/Claude replay; omitted only at genuine Responses ingress. */ translatorBudget?: TranslatorBudget; /** @@ -2289,6 +2291,7 @@ export async function handleComboResponses( attemptRetained = true; }; let consumedChildFailure: ConsumedComboFailure | undefined; + let storedPool401ReplayDispatched = false; const callbackGate = createChildPassthroughCallbackGate(options); let response: Response; try { @@ -2315,6 +2318,7 @@ export async function handleComboResponses( onCodexAuthContextResolved: value => { resolvedAuth = value; }, setTerminalOutcomeRecorder: value => { terminalRecorder = value; }, onConsumedComboFailure: value => { consumedChildFailure = value; }, + onStoredPool401ReplayDispatched: () => { storedPool401ReplayDispatched = true; }, onNativePassthroughTerminal: callbackGate.onTerminal, onNativePassthroughCancel: callbackGate.onCancel, }); @@ -2420,6 +2424,10 @@ export async function handleComboResponses( (logCtx.attempts ??= []).push(attempt); attemptRetained = true; lastFailure = failure.response; + if (storedPool401ReplayDispatched) { + adoptFailedChildLog(childLog); + return lastFailure; + } if (comboFailureDecision(failure.response.status, failure.classificationText, { code: failure.upstreamCode, }) === "stop") { @@ -3839,7 +3847,7 @@ async function handleResponsesInner( const opaqueBlobRecoveryGuard: OpaqueBlobRecoveryGuard = { attempted: false }; let oauth401ReplayAttempted = false; - let codexMain401ReplayAttempted = false; + let codex401ReplayKind: "main" | "stored" | null = null; const rateLimitPolicy = rateLimitRetryPolicyFor(route.provider); let rateLimitRetries = 0; const rebuildAndRefetch = async ( @@ -3914,9 +3922,9 @@ async function handleResponsesInner( upstreamResponse.status === 401 && (authCtx.kind === "main-pool" || authCtx.kind === "pool") && usesCodexForwardPoolAuth(authCtx, route.provider) - && !codexMain401ReplayAttempted + && codex401ReplayKind === null ) { - codexMain401ReplayAttempted = true; + codex401ReplayKind = authCtx.kind === "pool" ? "stored" : "main"; try { void upstreamResponse.body?.cancel().catch(() => {}); } catch { /* already consumed */ } const poolAuthCtx = authCtx.kind === "pool" ? authCtx : undefined; const poolReplay = poolAuthCtx @@ -3972,6 +3980,7 @@ async function handleResponsesInner( recordAdapterTier(logCtx, request); refreshUndeclaredToolGuard(request); noteAttemptSend(logCtx.activeAttempt, passthroughEstimate, "oauth-401"); + if (codex401ReplayKind === "stored") options.onStoredPool401ReplayDispatched?.(); upstreamResponse = await fetchWithHeaderTimeout( request.url, { method: request.method, headers: request.headers, body: request.body }, @@ -3995,7 +4004,11 @@ async function handleResponsesInner( continue passthroughRecovery; } - if (codexMain401ReplayAttempted && upstreamResponse.status === 401) break; + if (codex401ReplayKind !== null && upstreamResponse.status === 401) break; + // A stored Pool 401 owns one refresh and one same-account replay. The replay + // result is authoritative for this logical request and cannot enter a later + // account, model, or combo recovery ladder. + if (codex401ReplayKind === "stored" && upstreamResponse.status >= 400) break; // Native Responses providers return before the generic adapter recovery loop below. Keep // their OAuth contract identical: one pre-stream 401 forces a credential refresh and one diff --git a/src/server/responses/policy-fallback.ts b/src/server/responses/policy-fallback.ts index eabb70cf25..a4f06d0fa6 100644 --- a/src/server/responses/policy-fallback.ts +++ b/src/server/responses/policy-fallback.ts @@ -116,16 +116,21 @@ export async function handleResponsesWithPolicyFallback( ): Promise { const runCore = deps.runCore ?? handleResponsesCore; let requestBodyReadNotified = false; - const coreOptions: CoreOptions = options.onRequestBodyRead - ? { - ...options, + let storedPool401ReplayDispatched = false; + const coreOptions: CoreOptions = { + ...options, + ...(options.onRequestBodyRead ? { onRequestBodyRead: () => { if (requestBodyReadNotified) return; requestBodyReadNotified = true; options.onRequestBodyRead?.(); }, - } - : options; + } : {}), + onStoredPool401ReplayDispatched: () => { + storedPool401ReplayDispatched = true; + options.onStoredPool401ReplayDispatched?.(); + }, + }; let rawBody: Record | null = null; try { const parsed = await readJsonRequestBody(req.clone()); @@ -150,7 +155,7 @@ export async function handleResponsesWithPolicyFallback( candidateKey({ provider: initialTrace.selected.provider, model: initialTrace.selected.model }), ]); - while (await shouldHopPolicyCandidate(response, req.signal)) { + while (!storedPool401ReplayDispatched && await shouldHopPolicyCandidate(response, req.signal)) { if (req.signal.aborted) return response; const next = rankPolicyFallbackCandidates(initialTrace, tried)[0]; if (!next) return response; diff --git a/tests/responses-native-main-refresh.test.ts b/tests/responses-native-main-refresh.test.ts index f700d7cae6..ab59bebe1f 100644 --- a/tests/responses-native-main-refresh.test.ts +++ b/tests/responses-native-main-refresh.test.ts @@ -3,6 +3,7 @@ import { mkdtempSync, readFileSync, rmSync, writeFileSync } from "node:fs"; import { tmpdir } from "node:os"; import { join } from "node:path"; import { clearAccountNeedsReauth } from "../src/codex/auth-api"; +import { saveCodexAccountCredential } from "../src/codex/account-store"; import { MAIN_CODEX_ACCOUNT_ID } from "../src/codex/main-account"; import { clearCodexUpstreamHealth, clearThreadAccountMap } from "../src/codex/routing"; import { handleResponses, handleResponsesCompact } from "../src/server/responses"; @@ -13,8 +14,9 @@ const originalFetch = globalThis.fetch; let home = ""; let previousOcxHome: string | undefined; let previousCodexHome: string | undefined; +const OTHER_ACCOUNT_ID = "other"; -function config(): OcxConfig { +function config(options: { secondAccount?: boolean } = {}): OcxConfig { return { defaultProvider: "openai", activeCodexAccountId: MAIN_CODEX_ACCOUNT_ID, @@ -27,7 +29,8 @@ function config(): OcxConfig { codexAccountMode: "pool", }, }, - codexAccounts: [], + codexAccounts: options.secondAccount ? [{ id: OTHER_ACCOUNT_ID, label: "other" }] : [], + ...(options.secondAccount ? { accountPoolStrategy: "fill-first" } : {}), } as OcxConfig; } @@ -48,6 +51,7 @@ beforeEach(() => { process.env.OPENCODEX_HOME = home; process.env.CODEX_HOME = home; clearAccountNeedsReauth(MAIN_CODEX_ACCOUNT_ID); + clearAccountNeedsReauth(OTHER_ACCOUNT_ID); clearCodexUpstreamHealth(); clearThreadAccountMap(); writeFileSync(join(home, "auth.json"), JSON.stringify({ @@ -62,6 +66,7 @@ beforeEach(() => { afterEach(() => { globalThis.fetch = originalFetch; clearAccountNeedsReauth(MAIN_CODEX_ACCOUNT_ID); + clearAccountNeedsReauth(OTHER_ACCOUNT_ID); clearCodexUpstreamHealth(); clearThreadAccountMap(); if (previousOcxHome === undefined) delete process.env.OPENCODEX_HOME; @@ -159,4 +164,52 @@ describe("native main 401 refresh and replay", () => { expect(harness.sends).toEqual(["Bearer rejected-access", "Bearer refreshed-access"]); expect(harness.refreshes).toEqual(["refresh-grant"]); }); + + for (const path of ["/v1/responses", "/v1/responses/compact"] as const) { + test(`${path} keeps main-pool recovery eligible for a later Pool account`, async () => { + saveCodexAccountCredential(OTHER_ACCOUNT_ID, { + accessToken: "other-access", + refreshToken: "other-refresh", + expiresAt: Date.now() + 3_600_000, + chatgptAccountId: "account-other", + }); + const sends: string[] = []; + const refreshes: string[] = []; + globalThis.fetch = (async (input: RequestInfo | URL, init?: RequestInit) => { + const url = new URL(input instanceof Request ? input.url : String(input)); + if (url.hostname === "auth.openai.com") { + refreshes.push(new URLSearchParams(String(init?.body)).get("refresh_token") ?? ""); + return Response.json({ + access_token: "refreshed-access", + refresh_token: "rotated-refresh", + expires_in: 3600, + }); + } + if (!url.pathname.endsWith("/responses") && !url.pathname.endsWith("/responses/compact")) { + return Response.json({ rate_limit: { primary_window: { used_percent: 10 } } }); + } + const authorization = new Headers(init?.headers).get("authorization") ?? ""; + sends.push(authorization); + if (authorization === "Bearer rejected-access") { + return Response.json({ error: { message: "expired bearer" } }, { status: 401 }); + } + if (authorization === "Bearer refreshed-access") { + return Response.json({ error: { message: "main quota exhausted" } }, { status: 429 }); + } + if (authorization === "Bearer other-access") { + return Response.json({ id: "resp_other", object: "response", status: "completed", output: [] }); + } + return Response.json({ error: { message: "unexpected bearer" } }, { status: 500 }); + }) as typeof fetch; + + const cfg = config({ secondAccount: true }); + const response = path.endsWith("compact") + ? await handleResponsesCompact(request(path), cfg, { model: "", provider: "" } as RequestLogContext) + : await handleResponses(request(path), cfg, { model: "", provider: "" } as RequestLogContext); + + expect(response.status).toBe(200); + expect(sends).toEqual(["Bearer rejected-access", "Bearer refreshed-access", "Bearer other-access"]); + expect(refreshes).toEqual(["refresh-grant"]); + }); + } }); diff --git a/tests/responses-pool-401-refresh.test.ts b/tests/responses-pool-401-refresh.test.ts index 73c78a8d13..a733953064 100644 --- a/tests/responses-pool-401-refresh.test.ts +++ b/tests/responses-pool-401-refresh.test.ts @@ -10,6 +10,7 @@ import { resolveCodexAccountForThreadDetailed, } from "../src/codex/routing"; import { handleResponses, handleResponsesCompact } from "../src/server/responses"; +import { clearCompactHandoffRoutesForTests } from "../src/server/responses/compact"; import type { RequestLogContext } from "../src/server/request-log"; import type { OcxConfig } from "../src/types"; @@ -57,19 +58,17 @@ const THREAD_ID = "thread-2887"; function request( path: "/v1/responses" | "/v1/responses/compact", - options: { affined?: boolean } = {}, + options: { affined?: boolean; model?: string; headers?: HeadersInit; stream?: boolean } = {}, ): Request { + const headers = new Headers(options.headers); + headers.set("content-type", "application/json"); + if (options.affined) headers.set("x-codex-parent-thread-id", THREAD_ID); return new Request(`http://localhost${path}`, { method: "POST", - headers: { - "content-type": "application/json", - // A bound thread is what makes affinity exist at all; without it there is no - // entry to carry across the refresh and the handoff cannot be observed. - ...(options.affined ? { "x-codex-parent-thread-id": THREAD_ID } : {}), - }, + headers, body: JSON.stringify(path.endsWith("compact") - ? { model: "gpt-5.5", input: [] } - : { model: "gpt-5.5", input: "hello", stream: false }), + ? { model: options.model ?? "gpt-5.5", input: [] } + : { model: options.model ?? "gpt-5.5", input: "hello", stream: options.stream ?? false }), }); } @@ -118,7 +117,10 @@ type Harness = { sends: string[]; refreshes: string[] }; * Upstream rejects the old bearer once, the token endpoint rotates, and the replay with the * new bearer succeeds — the reporter's deterministic harness. */ -function installHarness(options: { refresh?: () => Response } = {}): Harness { +function installHarness(options: { + refresh?: () => Response; + responseForSend?: (authorization: string, sendNumber: number, url: URL) => Response | undefined; +} = {}): Harness { const sends: string[] = []; const refreshes: string[] = []; globalThis.fetch = (async (input: RequestInfo | URL, init?: RequestInit) => { @@ -137,6 +139,8 @@ function installHarness(options: { refresh?: () => Response } = {}): Harness { } const authorization = new Headers(init?.headers).get("authorization") ?? ""; sends.push(authorization); + const customResponse = options.responseForSend?.(authorization, sends.length, url); + if (customResponse) return customResponse; if (authorization === "Bearer rejected-access") { return Response.json({ error: { message: "expired bearer" } }, { status: 401 }); } @@ -145,6 +149,26 @@ function installHarness(options: { refresh?: () => Response } = {}): Harness { return { sends, refreshes }; } +function recoveryComboConfig(): OcxConfig { + const cfg = config(); + cfg.providers.backup = { + adapter: "openai-responses", + baseUrl: "https://backup.example/v1", + authMode: "key", + apiKey: "backup-test-key", + }; + cfg.combos = { + recovery: { + strategy: "failover", + targets: [ + { provider: "openai", model: "gpt-5.5" }, + { provider: "backup", model: "m2" }, + ], + }, + }; + return cfg; +} + beforeEach(() => { home = mkdtempSync(join(tmpdir(), "ocx-responses-pool-401-")); previousOcxHome = process.env.OPENCODEX_HOME; @@ -152,6 +176,7 @@ beforeEach(() => { process.env.OPENCODEX_HOME = home; process.env.CODEX_HOME = home; clearAccountNeedsReauth(ACCOUNT_ID); + clearAccountNeedsReauth(OTHER_ACCOUNT_ID); clearCodexUpstreamHealth(); clearThreadAccountMap(); writeStoredAccount(); @@ -159,7 +184,9 @@ beforeEach(() => { afterEach(() => { globalThis.fetch = originalFetch; + clearCompactHandoffRoutesForTests(); clearAccountNeedsReauth(ACCOUNT_ID); + clearAccountNeedsReauth(OTHER_ACCOUNT_ID); clearCodexUpstreamHealth(); clearThreadAccountMap(); if (previousOcxHome === undefined) delete process.env.OPENCODEX_HOME; @@ -201,6 +228,215 @@ describe("ordinary pool 401 refresh and replay (#2887)", () => { expect(isAccountNeedsReauth(ACCOUNT_ID)).toBe(false); }); + test("Responses does not compose a stored-account replay 429 with another Pool account", async () => { + writeStoredAccount({ + [OTHER_ACCOUNT_ID]: storedRecord({ + accessToken: "other-access", + refreshToken: "other-grant", + generation: 1, + chatgptAccountId: "acc-other", + }), + }); + const harness = installHarness({ + responseForSend: authorization => { + if (authorization === "Bearer rejected-access") { + return Response.json({ error: { message: "rejected bearer" } }, { status: 401 }); + } + if (authorization === "Bearer refreshed-access") { + return Response.json({ error: { message: "pool exhausted" } }, { status: 429 }); + } + if (authorization === "Bearer other-access") { + return Response.json({ id: "must-not-run", object: "response", status: "completed", output: [] }); + } + return undefined; + }, + }); + + const cfg = config({ secondAccount: true }); + // This test needs one eligible alternate but must not advance the process-wide + // round-robin cursor used by the existing next-request affinity regression. + cfg.accountPoolStrategy = "fill-first"; + const response = await handleResponses( + request("/v1/responses"), + cfg, + { model: "", provider: "" } as RequestLogContext, + ); + + expect(response.status).toBe(429); + expect(harness.sends).toEqual(["Bearer rejected-access", "Bearer refreshed-access"]); + expect(harness.refreshes).toEqual(["refresh-grant"]); + }); + + test("a combo stops after a stored-account replay consumes the recovery budget", async () => { + const cfg = recoveryComboConfig(); + const harness = installHarness({ + responseForSend: (authorization, _sendNumber, url) => { + if (url.hostname === "backup.example") { + return Response.json({ id: "must-not-run", object: "response", status: "completed", output: [] }); + } + if (authorization === "Bearer rejected-access") { + return Response.json({ error: { message: "rejected bearer" } }, { status: 401 }); + } + if (authorization === "Bearer refreshed-access") { + return Response.json({ error: { message: "pool exhausted" } }, { status: 429 }); + } + return undefined; + }, + }); + + const response = await handleResponses( + request("/v1/responses", { model: "combo/recovery" }), + cfg, + { model: "", provider: "" } as RequestLogContext, + ); + + expect(response.status).toBe(429); + expect(harness.sends).toEqual(["Bearer rejected-access", "Bearer refreshed-access"]); + expect(harness.refreshes).toEqual(["refresh-grant"]); + }); + + test("a combo stops when the stored-account replay hits a transport error", async () => { + const cfg = recoveryComboConfig(); + const harness = installHarness({ + responseForSend: (authorization, _sendNumber, url) => { + if (url.hostname === "backup.example") { + return Response.json({ id: "must-not-run", object: "response", status: "completed", output: [] }); + } + if (authorization === "Bearer rejected-access") { + return Response.json({ error: { message: "rejected bearer" } }, { status: 401 }); + } + if (authorization === "Bearer refreshed-access") { + throw new TypeError("stored replay transport failure"); + } + return undefined; + }, + }); + + const response = await handleResponses( + request("/v1/responses", { model: "combo/recovery" }), + cfg, + { model: "", provider: "" } as RequestLogContext, + ); + + expect(response.status).toBe(502); + expect(harness.sends).toEqual(["Bearer rejected-access", "Bearer refreshed-access"]); + expect(harness.refreshes).toEqual(["refresh-grant"]); + }); + + test("a combo stops on a zero-output failure from the stored-account replay stream", async () => { + const cfg = recoveryComboConfig(); + const harness = installHarness({ + responseForSend: (authorization, _sendNumber, url) => { + if (url.hostname === "backup.example") { + return Response.json({ id: "must-not-run", object: "response", status: "completed", output: [] }); + } + if (authorization === "Bearer rejected-access") { + return Response.json({ error: { message: "rejected bearer" } }, { status: 401 }); + } + if (authorization === "Bearer refreshed-access") { + const events = [ + { type: "response.created", response: { id: "replay", status: "in_progress" } }, + { + type: "response.failed", + response: { + id: "replay", + status: "failed", + error: { type: "server_error", code: "upstream_server_error", message: "busy" }, + }, + }, + ]; + return new Response( + events.map(event => `data: ${JSON.stringify(event)}\n\n`).join(""), + { headers: { "content-type": "text/event-stream" } }, + ); + } + return undefined; + }, + }); + + const response = await handleResponses( + request("/v1/responses", { model: "combo/recovery", stream: true }), + cfg, + { model: "", provider: "" } as RequestLogContext, + ); + + expect(response.status).toBe(502); + const failure = await response.clone().json() as { + error?: { code?: string; message?: string }; + }; + expect(failure.error?.code).toBe("upstream_server_error"); + expect(failure.error?.message).toContain("busy"); + expect(harness.sends).toEqual(["Bearer rejected-access", "Bearer refreshed-access"]); + expect(harness.refreshes).toEqual(["refresh-grant"]); + }); + + test("compact does not compose a stored-account replay 429 with a remembered model", async () => { + const headers = { "x-codex-parent-thread-id": "compact-refresh-budget" }; + writeStoredAccount({ + [OTHER_ACCOUNT_ID]: storedRecord({ + accessToken: "other-access", + refreshToken: "other-grant", + generation: 1, + chatgptAccountId: "acc-other", + }), + }); + const cfg = config({ secondAccount: true }); + cfg.accountPoolStrategy = "fill-first"; + cfg.providers.seed = { + adapter: "openai-responses", + baseUrl: "https://seed.example/v1", + authMode: "key", + apiKey: "seed-test-key", + }; + const harness = installHarness({ + responseForSend: (authorization, _sendNumber, url) => { + if (url.hostname === "seed.example") { + return Response.json({ + id: "seed", + object: "response", + status: "completed", + output: [{ + type: "message", + role: "assistant", + content: [{ type: "output_text", text: "seed summary", annotations: [] }], + }], + }); + } + if (authorization === "Bearer rejected-access") { + return Response.json({ error: { message: "rejected bearer" } }, { status: 401 }); + } + if (authorization === "Bearer refreshed-access") { + return Response.json({ error: { message: "pool exhausted" } }, { status: 429 }); + } + if (authorization === "Bearer other-access") { + return Response.json({ id: "must-not-run", object: "response", status: "completed", output: [] }); + } + return undefined; + }, + }); + + const seed = await handleResponsesCompact( + request("/v1/responses/compact", { model: "seed/seed-model", headers }), + cfg, + { model: "", provider: "" } as RequestLogContext, + ); + expect(seed.status).toBe(200); + + const response = await handleResponsesCompact( + request("/v1/responses/compact", { headers }), + cfg, + { model: "", provider: "" } as RequestLogContext, + ); + + expect(response.status).toBe(429); + expect(harness.sends).toEqual([ + "Bearer seed-test-key", + "Bearer rejected-access", + "Bearer refreshed-access", + ]); + expect(harness.refreshes).toEqual(["refresh-grant"]); + }); + test("the replayed account is still selectable on the NEXT request, not just this one", async () => { // The affinity entry is bound under generation G; the forced refresh CAS-writes G+1 and // isThreadAffinityGenerationLive demands exact equality. Without the same-lineage handoff diff --git a/tests/routing-policy-fallback.test.ts b/tests/routing-policy-fallback.test.ts index ac2fc34f28..acfadb3744 100644 --- a/tests/routing-policy-fallback.test.ts +++ b/tests/routing-policy-fallback.test.ts @@ -179,6 +179,33 @@ describe("policy candidate fallback", () => { expect(logCtx.activeAttempt).toBe(logCtx.attempts?.[1]); }); + test("a stored Pool 401 replay dispatch stops policy candidate fallback", async () => { + const trace = policyTrace(); + const logCtx = { requestedModel: "policy/daily", routeDecision: trace, attempts: [] } as unknown as RequestLogContext; + const seenModels: string[] = []; + let replaySignals = 0; + + const response = await handleResponsesWithPolicyFallback(request(), {} as OcxConfig, logCtx, { + onStoredPool401ReplayDispatched: () => { replaySignals += 1; }, + }, { + runCore: async (req, _config, childLog, options) => { + const body = await req.json() as { model: string }; + seenModels.push(body.model); + childLog.routeDecision = trace; + seedAttempt(childLog, "provider-a", "model-a"); + options.onStoredPool401ReplayDispatched?.(); + return Response.json( + { error: { message: "stored replay exhausted", type: "rate_limit_error" } }, + { status: 429 }, + ); + }, + }); + + expect(response.status).toBe(429); + expect(seenModels).toEqual(["policy/daily"]); + expect(replaySignals).toBe(1); + }); + test("returns local pacing overload without switching policy candidates", async () => { const trace = policyTrace(); const logCtx = { requestedModel: "policy/daily", routeDecision: trace, attempts: [] } as unknown as RequestLogContext;