diff --git a/src/server/responses/compact.ts b/src/server/responses/compact.ts index 25bfe5a4d..17e9ff93d 100644 --- a/src/server/responses/compact.ts +++ b/src/server/responses/compact.ts @@ -27,7 +27,7 @@ import { import { isInjectionDebugEnabled } from "../../lib/debug-settings"; import { injectionDebugLog } from "../../lib/injection-debug-log"; import { modelInList, namespacedToolName } from "../../types"; -import type { AdapterEvent, OcxConfig, OcxParsedRequest, OcxProviderConfig, OcxProviderContinuationState, OcxUsage } from "../../types"; +import type { AdapterEvent, CodexAccountMode, OcxConfig, OcxParsedRequest, OcxProviderConfig, OcxProviderContinuationState, OcxUsage } from "../../types"; import { forceRefreshOAuthAccessSnapshot, getOAuthCredentialApiBaseUrl, @@ -52,6 +52,7 @@ import { resolveCodexAuthContext, codexProbeLeaseId, codexProbeQuotaScope, + releaseCodexAuthContextProbeLease, type CodexAuthContext, } from "../../codex/auth-context"; import { @@ -59,7 +60,12 @@ import { recordCodexUpstreamOutcome, type CodexUpstreamOutcome, } from "../../codex/routing"; -import { fetchWithResetRetry, fetchWithTransientRetry, applyUpstreamRecoveryInit } from "../../lib/upstream-retry"; +import { + fetchWithResetRetry, + fetchWithTransientRetry, + applyUpstreamRecoveryInit, + type UpstreamSendRecovery, +} from "../../lib/upstream-retry"; import { ForwardAdmissionCredentialError, validateForwardAdmissionCredential } from "../auth-cors"; import { listOpenAiForwardSidecarCandidates, resolveFirstUsableOpenAiSidecar, type ResolvedOpenAiForwardSidecar } from "../../providers/openai-sidecar"; import { isCanonicalOpenAiForwardProvider, supportsNativeResponsesCompactEndpoint } from "../../providers/openai-tiers"; @@ -117,10 +123,78 @@ export function compactResponseTooLargeError(): Response { +/** + * Resolve one eligible pool account other than `excludeAccountId`, and build everything + * the alternate send needs. Returns null when no alternate exists or construction fails, + * in which case the caller keeps the first account's rejection intact. + * + * Mirrors the auth resolution the native compact branch already does for the first + * account, so the alternate is built the same way rather than through a second, + * divergent path. + */ +async function resolveAlternateCompactContext(args: { + req: Request; + config: OcxConfig; + route: { provider: OcxProviderConfig; codexAccountMode?: CodexAccountMode }; + selectedModelId: string | undefined; + excludeAccountId: string | null; +}): Promise<{ authCtx: CodexAuthContext; provider: OcxProviderConfig; headers: Headers } | null> { + const { req, config, route, selectedModelId, excludeAccountId } = args; + if (!route.codexAccountMode || !excludeAccountId) return null; + try { + const authCtx = await resolveCodexAuthContext(req.headers, config, route.codexAccountMode, { + ...(selectedModelId ? { modelId: selectedModelId } : {}), + excludeAccountId, + }); + if (!authCtx.accountId || authCtx.accountId === excludeAccountId) return null; + const provider = applyCodexAuthContextToProvider(route.provider, authCtx, route.codexAccountMode); + const headers = new Headers({ "content-type": "application/json" }); + const selected = headersForCodexAuthContext(req.headers, authCtx); + for (const name of FORWARD_HEADERS) { + const value = selected.get(name); + if (value) headers.set(name, value); + } + const override = (provider as { _codexAccountOverride?: { accessToken: string; chatgptAccountId: string } })._codexAccountOverride; + if (override) { + headers.set("authorization", `Bearer ${override.accessToken}`); + headers.set("chatgpt-account-id", override.chatgptAccountId); + } + if (provider.apiKey) headers.set("authorization", `Bearer ${resolveEnvValue(provider.apiKey)}`); + return { authCtx, provider, headers }; + } catch { + // No eligible alternate (all cooled, affinity expired, reauth needed) — the caller + // returns the first account's rejection unchanged, which is today's behavior. + return null; + } +} + +/** + * Headers a client needs to back off correctly after a pool rejection. The buffered + * response is rebuilt from scratch, so anything not listed here is dropped — which is + * what silently discarded `Retry-After` and the reset hints from a 429 before. + * Deliberately narrow: no hop-by-hop headers, no content-length (the buffered body sets + * its own), no cookies. + */ +const COMPACT_PASSTHROUGH_HEADERS = [ + "retry-after", + "x-codex-primary-reset-at", + "x-codex-secondary-reset-at", + "x-codex-tertiary-reset-at", +]; + +function compactResponseHeaders(upstream: Response): Headers { + const headers = new Headers({ "Content-Type": upstream.headers.get("content-type") ?? "application/json" }); + for (const name of COMPACT_PASSTHROUGH_HEADERS) { + const value = upstream.headers.get(name); + if (value) headers.set(name, value); + } + return headers; +} + export async function bufferCompactResponse(upstream: Response, signal: AbortSignal): Promise { const reader = upstream.body?.getReader(); - const contentType = upstream.headers.get("content-type") ?? "application/json"; - if (!reader) return new Response(null, { status: upstream.status, headers: { "Content-Type": contentType } }); + const headers = compactResponseHeaders(upstream); + if (!reader) return new Response(null, { status: upstream.status, statusText: upstream.statusText, headers }); const declaredLength = Number(upstream.headers.get("content-length")); if (Number.isFinite(declaredLength) && declaredLength > COMPACT_RESPONSE_MAX_BYTES) { await reader.cancel("compact_response_too_large").catch(() => undefined); @@ -153,7 +227,7 @@ export async function bufferCompactResponse(upstream: Response, signal: AbortSig body.set(chunk, offset); offset += chunk.byteLength; } - return new Response(body, { status: upstream.status, headers: { "Content-Type": contentType } }); + return new Response(body, { status: upstream.status, statusText: upstream.statusText, headers }); } @@ -256,50 +330,142 @@ export async function handleResponsesCompact( const compactUrl = `${base}/responses/compact`; const compactThreadId = req.headers.get("x-codex-parent-thread-id"); const connectMs = config.connectTimeoutMs ?? 200_000; + // Takes its context explicitly: the alternate-account flow below records a rejection + // against A while promoting B, then records B's own outcome. A closure over a single + // `authCtx` cannot express either. const recordCompactPoolOutcome = ( + ctx: CodexAuthContext, outcome: CodexUpstreamOutcome, - meta: { retryAfter?: string | null; resetAt?: unknown | unknown[] } = {}, + meta: { + retryAfter?: string | null; + resetAt?: unknown | unknown[]; + promoteAccountId?: string; + } = {}, ) => { - if (!usesCodexForwardPoolAuth(authCtx, route.provider)) return; - recordCodexUpstreamOutcome(config, authCtx.accountId, outcome, { + if (!usesCodexForwardPoolAuth(ctx, route.provider)) return; + recordCodexUpstreamOutcome(config, ctx.accountId, outcome, { ...meta, threadId: compactThreadId, modelId: selectedModelId, - probeLeaseId: codexProbeLeaseId(authCtx), - probeQuotaScope: codexProbeQuotaScope(authCtx), - writerGeneration: authCtx.kind === "pool" || authCtx.kind === "main-pool" - ? authCtx.writerGeneration + probeLeaseId: codexProbeLeaseId(ctx), + probeQuotaScope: codexProbeQuotaScope(ctx), + writerGeneration: ctx.kind === "pool" || ctx.kind === "main-pool" + ? ctx.writerGeneration : undefined, }); }; + // Two recovery modes, mirroring retryCodexPoolOnAlternateAccount() on the regular + // path (core.ts:396). The first account keeps the full ladder — transient-5xx retry + // wrapping reset retry — because those retries happen before any alternate is even + // considered. The alternate is one bounded send: a second ladder would multiply the + // work an already-rejecting pool is doing. + const sendCompactAttempt = ( + sendProvider: OcxProviderConfig, + sendHeaders: Headers, + recovery: "normal" | "single", + ): Promise => { + const doFetch = (upstreamRecovery?: UpstreamSendRecovery) => fetchWithHeaderTimeout( + compactUrl, + applyUpstreamRecoveryInit({ + method: "POST", + headers: sendHeaders, + body: JSON.stringify({ ...compactBody, model: route.modelId }), + }, upstreamRecovery), + req.signal, + connectMs, + false, + providerFetch(sendProvider), + ); + return recovery === "single" + ? doFetch() + : fetchWithTransientRetry(doFetch, { abortSignal: req.signal, label: safeHostLabel(compactUrl) }); + }; + + // The account each outcome belongs to. Reassigned only when the alternate send below + // actually happens, so every recorder call names the context that produced it. + let outcomeCtx = authCtx; let upstream: Response; 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). - upstream = await fetchWithTransientRetry( - recovery => fetchWithHeaderTimeout( - compactUrl, - applyUpstreamRecoveryInit({ - method: "POST", - headers, - body: JSON.stringify({ ...compactBody, model: route.modelId }), - }, recovery), - req.signal, - connectMs, - false, - providerFetch(compactProvider), - ), - { abortSignal: req.signal, label: safeHostLabel(compactUrl) }, - ); + upstream = await sendCompactAttempt(compactProvider, headers, "normal"); } catch (err) { if (req.signal.aborted) { - recordCompactPoolOutcome(499); + recordCompactPoolOutcome(outcomeCtx, 499); return formatErrorResponse(499, "client_cancelled", "Client cancelled compact request"); } const outcome = err instanceof Error && err.name === "TimeoutError" ? "timeout" : "connect_error"; - recordCompactPoolOutcome(outcome); + recordCompactPoolOutcome(outcomeCtx, outcome); return formatErrorResponse(502, "upstream_error", "Failed to connect to compact upstream"); } + + // Bounded same-request alternate: the regular /v1/responses path already does this + // (core.ts:319-423) and recognizes exactly 429/402. Without it a pool rejection + // surfaces to the client, which retries the compact task OUTSIDE the logical request + // — reporting exhausted retries while another pool account sat idle (#913). + if ( + (upstream.status === 429 || upstream.status === 402) + && usesCodexForwardPoolAuth(authCtx, route.provider) + && route.codexAccountMode + && !req.signal.aborted + ) { + const firstRetryAfter = upstream.headers.get("retry-after"); + const firstResetAt = [ + upstream.headers.get("x-codex-primary-reset-at"), + upstream.headers.get("x-codex-secondary-reset-at"), + upstream.headers.get("x-codex-tertiary-reset-at"), + ].filter(Boolean); + // Build the alternate COMPLETELY before cancelling the first body: if construction + // throws, the first rejection is still intact and can be returned to the client. + const alternate = await resolveAlternateCompactContext({ + req, + config, + route, + selectedModelId, + excludeAccountId: authCtx.accountId, + }); + // Resolution can await a credential refresh, so the client may have gone away + // while we were choosing B. Re-check before spending anything: recording A, + // cancelling its body, and sending B are all observable side effects, and B's + // quota is not ours to spend on a request nobody is waiting for. + if (alternate && req.signal.aborted) { + releaseCodexAuthContextProbeLease(alternate.authCtx); + recordCompactPoolOutcome(outcomeCtx, 499); + return formatErrorResponse(499, "client_cancelled", "Client cancelled compact request"); + } + if (alternate) { + // Same order the regular path uses (core.ts:349-357): a 429/402 carries the + // quota snapshot that produced it, so refresh A's cache before recording its + // rejection. Skipping this leaves quota-strategy routing and the dashboard + // reading numbers from before the account ran out. + if (authCtx.kind === "pool" || authCtx.kind === "main-pool") { + const { applyAccountQuotaFromUpstreamHeaders } = await import("../../codex/auth-api"); + applyAccountQuotaFromUpstreamHeaders( + authCtx.accountId, + upstream.headers, + authCtx.writerGeneration, + ); + } + recordCompactPoolOutcome(authCtx, upstream.status, { + retryAfter: firstRetryAfter, + resetAt: firstResetAt, + ...(alternate.authCtx.accountId ? { promoteAccountId: alternate.authCtx.accountId } : {}), + }); + await upstream.body?.cancel().catch(() => undefined); + outcomeCtx = alternate.authCtx; + try { + upstream = await sendCompactAttempt(alternate.provider, alternate.headers, "single"); + } catch (err) { + if (req.signal.aborted) { + recordCompactPoolOutcome(outcomeCtx, 499); + return formatErrorResponse(499, "client_cancelled", "Client cancelled compact request"); + } + const outcome = err instanceof Error && err.name === "TimeoutError" ? "timeout" : "connect_error"; + recordCompactPoolOutcome(outcomeCtx, outcome); + return formatErrorResponse(502, "upstream_error", "Failed to connect to compact upstream"); + } + } + } const retryAfter = upstream.headers.get("retry-after"); const resetAt = [ upstream.headers.get("x-codex-primary-reset-at"), @@ -310,12 +476,12 @@ export async function handleResponsesCompact( // Record pool health only after the body is fully delivered (or definitively failed). // A premature 200 would clear soft-avoid while the client still sees a buffer 502. if (buffered.status === 499) { - recordCompactPoolOutcome(499); + recordCompactPoolOutcome(outcomeCtx, 499); return buffered; } // Always record the real upstream status: a local buffering failure after a // 200 upstream response must not soft-avoid a healthy account or rotate a thread. - recordCompactPoolOutcome(upstream.status, { retryAfter, resetAt }); + recordCompactPoolOutcome(outcomeCtx, upstream.status, { retryAfter, resetAt }); return buffered; } diff --git a/tests/responses-compaction-routing.test.ts b/tests/responses-compaction-routing.test.ts index cb50a15fd..33b1b50fb 100644 --- a/tests/responses-compaction-routing.test.ts +++ b/tests/responses-compaction-routing.test.ts @@ -13,8 +13,11 @@ import { saveCodexAccountCredential } from "../src/codex/account-store"; import { CODEX_QUOTA_PROBE_INTERVAL_MS, clearCodexUpstreamHealth, + getCodexUpstreamHealth, recordCodexUpstreamOutcome, + resolveCodexAccountForThread, } from "../src/codex/routing"; +import { updateAccountQuota } from "../src/codex/auth-api"; import { releaseCodexAuthContextProbeLease, resolveCodexAuthContext, @@ -64,10 +67,24 @@ function nativePoolConfig(): OcxConfig { } as OcxConfig; } -function compactionRequest(body: Record, signal?: AbortSignal): Request { +/** Two-account pool: the alternate-attempt tests need somewhere for the retry to go. */ +function twoAccountPoolConfig(): OcxConfig { + const config = nativePoolConfig(); + config.codexAccounts = [ + { id: "pool-a", email: "a@example.test", isMain: false, chatgptAccountId: "pool_acc_a" }, + { id: "pool-b", email: "b@example.test", isMain: false, chatgptAccountId: "pool_acc_b" }, + ] as OcxConfig["codexAccounts"]; + return config; +} + +function compactionRequest( + body: Record, + signal?: AbortSignal, + extraHeaders: Record = {}, +): Request { return new Request("http://localhost/v1/responses", { method: "POST", - headers: { "content-type": "application/json" }, + headers: { "content-type": "application/json", ...extraHeaders }, body: JSON.stringify(body), signal, }); @@ -506,3 +523,292 @@ describe("compaction terminal handling (#422)", () => { expect(await res.text()).toContain("\"type\":\"compaction\""); }); }); + +/** + * #913: `/v1/responses` already tries one eligible alternate account inside the same + * logical request after a pre-stream 429/402. Compact did not, so a pool rejection + * reached the client, which retried the compact task OUTSIDE the logical request and + * could report exhausted retries while another account sat idle. + * + * The send count is the activation proof throughout: one send means the branch never + * fired, three means it recursed. + */ +describe("compact alternate-account attempt (#913)", () => { + function withPoolEnv(name: string, run: (config: OcxConfig) => Promise): Promise { + const testDir = mkdtempSync(join(tmpdir(), name)); + const previousOpencodexHome = process.env.OPENCODEX_HOME; + const previousCodexHome = process.env.CODEX_HOME; + process.env.OPENCODEX_HOME = testDir; + process.env.CODEX_HOME = testDir; + clearCodexUpstreamHealth(); + for (const id of ["pool-a", "pool-b"]) { + saveCodexAccountCredential(id, { + accessToken: `${id}-access-token`, + refreshToken: `${id}-refresh-token`, + expiresAt: Date.now() + 300_000, + chatgptAccountId: id === "pool-a" ? "pool_acc_a" : "pool_acc_b", + }); + } + return run(twoAccountPoolConfig()).finally(() => { + globalThis.fetch = originalFetch; + clearCodexUpstreamHealth(); + rmSync(testDir, { recursive: true, force: true }); + if (previousOpencodexHome === undefined) delete process.env.OPENCODEX_HOME; + else process.env.OPENCODEX_HOME = previousOpencodexHome; + if (previousCodexHome === undefined) delete process.env.CODEX_HOME; + else process.env.CODEX_HOME = previousCodexHome; + }); + } + + for (const rejection of [429, 402] as const) { + test(`a pre-body ${rejection} tries exactly one alternate account`, async () => { + await withPoolEnv(`ocx-compact-alt-${rejection}-`, async config => { + const bearers: string[] = []; + globalThis.fetch = (async (_url: string, init?: RequestInit) => { + const auth = new Headers(init?.headers).get("authorization") ?? ""; + bearers.push(auth); + if (bearers.length === 1) { + return Response.json({ error: { message: "pool exhausted" } }, { + status: rejection, + headers: { "retry-after": "42" }, + }); + } + return jsonResponse(completedPayload("alternate compact response")); + }) as typeof fetch; + + const res = await handleResponsesCompact( + compactionRequest(baseCompactionBody({})), + config, + { model: "", provider: "" }, + ); + + // Two sends, not one and not three: the alternate ran once and did not recurse. + expect(bearers).toHaveLength(2); + expect(bearers[0]).not.toBe(bearers[1]); + expect(res.status).toBe(200); + }); + }); + + test(`the alternate after a ${rejection} gets one send even when it returns a transient 5xx`, async () => { + // Activation proof for the two recovery modes. The first account keeps + // fetchWithTransientRetry (up to three status attempts); the alternate must run + // as a single direct send. Without `recovery: "single"` the 503 below would be + // retried and the alternate's share of the send count would be three. + await withPoolEnv(`ocx-compact-alt-${rejection}-5xx-`, async config => { + const bearers: string[] = []; + globalThis.fetch = (async (_url: string, init?: RequestInit) => { + const auth = new Headers(init?.headers).get("authorization") ?? ""; + bearers.push(auth); + if (bearers.length === 1) { + return Response.json({ error: { message: "pool exhausted" } }, { status: rejection }); + } + return Response.json({ error: { message: "upstream busy" } }, { status: 503 }); + }) as typeof fetch; + + const res = await handleResponsesCompact( + compactionRequest(baseCompactionBody({})), + config, + { model: "", provider: "" }, + ); + + expect(bearers).toHaveLength(2); + expect(bearers[0]).not.toBe(bearers[1]); + expect(res.status).toBe(503); + }); + }); + } + + test("a bound thread at 100% local quota still sends once, with no alternate attempt", async () => { + // The scope guard. The alternate path must trigger on an actual upstream 429/402, + // never on a local quota reading: a cached 100% is what the affined account looked + // like last time, not a rejection. If the gate ever widened to consult quota, this + // request would resolve an alternate and send twice. + await withPoolEnv("ocx-compact-quota-100-", async config => { + const affined = resolveCodexAccountForThread("compact-quota-thread", config); + updateAccountQuota(affined, 100); + const bearers: string[] = []; + globalThis.fetch = (async (_url: string, init?: RequestInit) => { + bearers.push(new Headers(init?.headers).get("authorization") ?? ""); + return jsonResponse(completedPayload("compact response at full quota")); + }) as typeof fetch; + + const res = await handleResponsesCompact( + compactionRequest(baseCompactionBody({}), undefined, { + "x-codex-parent-thread-id": "compact-quota-thread", + }), + config, + { model: "", provider: "" }, + ); + + // Exactly one send, and the upstream succeeded, so nothing rotated. + expect(bearers).toHaveLength(1); + expect(res.status).toBe(200); + }); + }); + + test("with no eligible alternate the first rejection is returned with its backoff headers", async () => { + await withPoolEnv("ocx-compact-alt-none-", async config => { + // Single-account pool: nothing to fail over to. + config.codexAccounts = [config.codexAccounts![0]]; + let sends = 0; + globalThis.fetch = (async () => { + sends += 1; + return Response.json({ error: { message: "pool exhausted" } }, { + status: 429, + headers: { "retry-after": "77", "x-codex-primary-reset-at": "1900000000" }, + }); + }) as typeof fetch; + + const res = await handleResponsesCompact( + compactionRequest(baseCompactionBody({})), + config, + { model: "", provider: "" }, + ); + + expect(sends).toBe(1); + expect(res.status).toBe(429); + // The buffered response is rebuilt from scratch, so these have to be carried + // deliberately. Dropping them left the client with no basis to back off. + expect(res.headers.get("retry-after")).toBe("77"); + expect(res.headers.get("x-codex-primary-reset-at")).toBe("1900000000"); + }); + }); + + test("when the alternate also rejects, both sends happen and its rejection is returned", async () => { + await withPoolEnv("ocx-compact-alt-both-", async config => { + let sends = 0; + globalThis.fetch = (async () => { + sends += 1; + return Response.json({ error: { message: `rejection ${sends}` } }, { + status: 429, + headers: { "retry-after": String(sends) }, + }); + }) as typeof fetch; + + const res = await handleResponsesCompact( + compactionRequest(baseCompactionBody({})), + config, + { model: "", provider: "" }, + ); + + expect(sends).toBe(2); + expect(res.status).toBe(429); + // The SECOND rejection is what the client sees, headers included. + expect(res.headers.get("retry-after")).toBe("2"); + }); + }); + + test("a non-quota rejection does not trigger an alternate", async () => { + await withPoolEnv("ocx-compact-alt-400-", async config => { + let sends = 0; + globalThis.fetch = (async () => { + sends += 1; + return Response.json({ error: { message: "bad request" } }, { status: 400 }); + }) as typeof fetch; + + const res = await handleResponsesCompact( + compactionRequest(baseCompactionBody({})), + config, + { model: "", provider: "" }, + ); + + // Recognition is 429/402 only, matching the regular path's deliberate narrowness. + expect(sends).toBe(1); + expect(res.status).toBe(400); + }); + }); + + test("an abort between attempts prevents the alternate send", async () => { + await withPoolEnv("ocx-compact-alt-abort-", async config => { + const abort = new AbortController(); + let sends = 0; + globalThis.fetch = (async () => { + sends += 1; + abort.abort(); + return Response.json({ error: { message: "pool exhausted" } }, { status: 429 }); + }) as typeof fetch; + + await handleResponsesCompact( + compactionRequest(baseCompactionBody({}), abort.signal), + config, + { model: "", provider: "" }, + ); + + expect(sends).toBe(1); + }); + }); + + test("the alternate sends once even against a transient 5xx", async () => { + // The two-mode crux. Compact's normal send wraps fetchWithTransientRetry, which + // retries a 5xx up to three times. The alternate must NOT inherit that ladder: + // it is a last bounded try, not a second retry stack. Without the mode split this + // reads four sends (one from A, three from B's ladder). + await withPoolEnv("ocx-compact-alt-single-", async config => { + let sends = 0; + globalThis.fetch = (async () => { + sends += 1; + if (sends === 1) { + return Response.json({ error: { message: "pool exhausted" } }, { status: 429 }); + } + return Response.json({ error: { message: "upstream flaked" } }, { status: 503 }); + }) as typeof fetch; + + const res = await handleResponsesCompact( + compactionRequest(baseCompactionBody({})), + config, + { model: "", provider: "" }, + ); + + expect(sends).toBe(2); + expect(res.status).toBe(503); + }); + }); + + test("the first account keeps its transient-retry ladder", async () => { + // The control for the test above: A's recovery is unchanged, so a transient 5xx + // on A is still retried in place rather than treated as a reason to fail over. + await withPoolEnv("ocx-compact-alt-ladder-", async config => { + let sends = 0; + globalThis.fetch = (async () => { + sends += 1; + if (sends === 1) return Response.json({ error: { message: "flake" } }, { status: 503 }); + return jsonResponse(completedPayload("recovered on retry")); + }) as typeof fetch; + + const res = await handleResponsesCompact( + compactionRequest(baseCompactionBody({})), + config, + { model: "", provider: "" }, + ); + + // Two sends, both to A — a 503 is not 429/402, so no alternate is involved. + expect(sends).toBe(2); + expect(res.status).toBe(200); + }); + }); + + test("each account's health records its own outcome", async () => { + // Attribution: A's rejection belongs to A and B's belongs to B. Recording B's + // outcome against A would soft-avoid the wrong account and defeat the failover. + await withPoolEnv("ocx-compact-alt-attrib-", async config => { + let sends = 0; + globalThis.fetch = (async () => { + sends += 1; + return sends === 1 + ? Response.json({ error: { message: "a exhausted" } }, { status: 429 }) + : Response.json({ error: { message: "b rejected" } }, { status: 402 }); + }) as typeof fetch; + + await handleResponsesCompact( + compactionRequest(baseCompactionBody({})), + config, + { model: "", provider: "" }, + ); + + const health = (id: string) => getCodexUpstreamHealth(id) as { lastFailureStatus?: number } | null; + // Whichever account routing picked first carries the 429; the other carries B's 402. + const statuses = ["pool-a", "pool-b"].map(id => health(id)?.lastFailureStatus).sort(); + expect(statuses).toEqual([402, 429]); + }); + }); +});