From 565b7851aae7b6498e35cae415bc8fd5954b3df7 Mon Sep 17 00:00:00 2001 From: bitkyc08-arch Date: Mon, 3 Aug 2026 14:53:17 +0900 Subject: [PATCH 1/4] fix(compact): try one alternate account on a pool 429/402, and keep the backoff headers MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Closes #913. `/v1/responses` already answers a pre-stream 429/402 by trying one eligible alternate account inside the same logical request (`retryCodexPoolOnAlternateAccount()`, core.ts:319-423). Compact did not: it resolved one context, sent once, and returned the rejection. The client then retried the compact task OUTSIDE the logical request, which is how a session reports exhausted retries while another pool account sits idle. Three things had to change together. The recorder took its account from a closure, so it could not express "record this against A and promote B", let alone record anything against B. It now takes the context explicitly; every existing call site passes the same `authCtx` it used before. The two accounts need different recovery. Compact's send goes through `fetchWithTransientRetry()`, which makes up to three status attempts each wrapping its own reset retries. "Send the alternate exactly once" and "keep A's existing recovery" only coexist if the modes differ — so A keeps the full ladder and B gets a single direct send, exactly as the regular path does at core.ts:396. The asymmetry is deliberate there too: A's retries happen before any alternate is considered, and the alternate is a last bounded try rather than a second ladder. `bufferCompactResponse()` rebuilt the response with only Content-Type, dropping `Retry-After` and the reset hints from the very rejection a client needs them for. It now carries a narrow allowlist and the upstream statusText. The alternate is built completely before A's body is cancelled, so a failure during construction still leaves A's rejection returnable. Six regressions, with the send count as the activation proof throughout — one send means the branch never fired, three means it recursed: 429 and 402 each try exactly one alternate; no eligible alternate returns A's rejection with its headers intact after exactly one send; a rejecting alternate produces two sends and returns the second rejection; a 400 does not trigger an alternate; an abort between attempts prevents the second send. --- src/server/responses/compact.ts | 206 +++++++++++++++++---- tests/responses-compaction-routing.test.ts | 169 +++++++++++++++++ 2 files changed, 344 insertions(+), 31 deletions(-) diff --git a/src/server/responses/compact.ts b/src/server/responses/compact.ts index 25bfe5a4d..98b81114e 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, @@ -59,7 +59,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 +122,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 +226,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 +329,121 @@ 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, + }); + if (alternate) { + 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 +454,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..d02eaa540 100644 --- a/tests/responses-compaction-routing.test.ts +++ b/tests/responses-compaction-routing.test.ts @@ -64,6 +64,16 @@ function nativePoolConfig(): OcxConfig { } as OcxConfig; } +/** 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): Request { return new Request("http://localhost/v1/responses", { method: "POST", @@ -506,3 +516,162 @@ 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("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); + }); + }); +}); From 361b329a39043b134294a97e8f347e1809cd13a4 Mon Sep 17 00:00:00 2001 From: bitkyc08-arch Date: Mon, 3 Aug 2026 15:07:41 +0900 Subject: [PATCH 2/4] test(compact): prove the alternate's single-send recovery mode actually fires MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The suite asserted send counts for the 429/402 path, the no-alternate path, the both-reject path, and the abort path — but nothing drove the alternate into a transient 5xx, which is the only case that distinguishes `recovery: "single"` from `recovery: "normal"`. A wiring mistake that left the alternate on the transient ladder would have passed every existing test. Now it does not. Ablated by flipping the alternate back to "normal": the send count goes from 2 to 6 (one for A, three for B, across both rejection codes) and the test fails. --- tests/responses-compaction-routing.test.ts | 103 +++++++++++++++++++++ 1 file changed, 103 insertions(+) diff --git a/tests/responses-compaction-routing.test.ts b/tests/responses-compaction-routing.test.ts index d02eaa540..a0ca7e31f 100644 --- a/tests/responses-compaction-routing.test.ts +++ b/tests/responses-compaction-routing.test.ts @@ -13,6 +13,7 @@ import { saveCodexAccountCredential } from "../src/codex/account-store"; import { CODEX_QUOTA_PROBE_INTERVAL_MS, clearCodexUpstreamHealth, + getCodexUpstreamHealth, recordCodexUpstreamOutcome, } from "../src/codex/routing"; import { @@ -581,6 +582,34 @@ describe("compact alternate-account attempt (#913)", () => { 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("with no eligible alternate the first rejection is returned with its backoff headers", async () => { @@ -674,4 +703,78 @@ describe("compact alternate-account attempt (#913)", () => { 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]); + }); + }); }); From 20f4f6dd0b0e37c529301c484f4cb85c3c37a3e6 Mon Sep 17 00:00:00 2001 From: bitkyc08-arch Date: Mon, 3 Aug 2026 15:09:45 +0900 Subject: [PATCH 3/4] fix(compact): refresh the rejected account's quota cache and close the abort race MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two review blockers on the alternate-account attempt. A's quota cache went stale. The regular path applies the rejected account's upstream quota headers before recording it (core.ts:349-357) — a 429 carries the snapshot that produced it, so skipping that leaves quota-strategy routing and the dashboard reading numbers from before the account ran out. Compact now does the same thing in the same order. Cancellation could race alternate resolution. The only abort check sat before the await, and resolution can wait on a credential refresh. A client that went away during that window still got A recorded, A's body cancelled, and B's fetch invoked. Native fetch rejects an aborted signal, but a custom executor need not, and B's quota is not ours to spend on a request nobody is waiting for. Re-checked after resolution, releasing B's probe lease on the way out. --- src/server/responses/compact.ts | 22 ++++++++++++++++++++++ 1 file changed, 22 insertions(+) diff --git a/src/server/responses/compact.ts b/src/server/responses/compact.ts index 98b81114e..17e9ff93d 100644 --- a/src/server/responses/compact.ts +++ b/src/server/responses/compact.ts @@ -52,6 +52,7 @@ import { resolveCodexAuthContext, codexProbeLeaseId, codexProbeQuotaScope, + releaseCodexAuthContextProbeLease, type CodexAuthContext, } from "../../codex/auth-context"; import { @@ -423,7 +424,28 @@ export async function handleResponsesCompact( 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, From 8236ffecb4b25528c4e422c8e0ce5f518a040237 Mon Sep 17 00:00:00 2001 From: bitkyc08-arch Date: Mon, 3 Aug 2026 15:18:59 +0900 Subject: [PATCH 4/4] test(compact): pin the alternate gate to upstream rejections, not local quota MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit CodeRabbit asked for the case the plan called the scope guard: a bound thread whose affined account reads 100% locally, with the upstream answering normally. Nothing in the file covered it. It matters because a cached 100% is what the account looked like on its last WHAM read, not a rejection — the alternate attempt has to trigger on an actual upstream 429/402. If the gate ever widened to consult quota, this request would resolve an alternate and send twice, and every other test in the file would still pass. One send, 200 returned, no rotation. --- tests/responses-compaction-routing.test.ts | 38 ++++++++++++++++++++-- 1 file changed, 36 insertions(+), 2 deletions(-) diff --git a/tests/responses-compaction-routing.test.ts b/tests/responses-compaction-routing.test.ts index a0ca7e31f..33b1b50fb 100644 --- a/tests/responses-compaction-routing.test.ts +++ b/tests/responses-compaction-routing.test.ts @@ -15,7 +15,9 @@ import { clearCodexUpstreamHealth, getCodexUpstreamHealth, recordCodexUpstreamOutcome, + resolveCodexAccountForThread, } from "../src/codex/routing"; +import { updateAccountQuota } from "../src/codex/auth-api"; import { releaseCodexAuthContextProbeLease, resolveCodexAuthContext, @@ -75,10 +77,14 @@ function twoAccountPoolConfig(): OcxConfig { return config; } -function compactionRequest(body: Record, signal?: AbortSignal): Request { +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, }); @@ -612,6 +618,34 @@ describe("compact alternate-account attempt (#913)", () => { }); } + 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.