From 6b84dedb6054639e99ee17241835f456ca289ef3 Mon Sep 17 00:00:00 2001 From: gg Date: Sun, 6 Sep 2026 05:13:40 -0700 Subject: [PATCH] fix(responses): fallback to routed compaction on 404 and enable quota failover on incomplete terminal --- src/server/request-log.ts | 12 +- src/server/responses/compact.ts | 9 +- src/server/responses/core.ts | 32 +++- ...responses-forward-incomplete-quota.test.ts | 172 ++++++++++++++++++ 4 files changed, 216 insertions(+), 9 deletions(-) create mode 100644 tests/responses/responses-forward-incomplete-quota.test.ts diff --git a/src/server/request-log.ts b/src/server/request-log.ts index a4c942bd88..c444f675e6 100644 --- a/src/server/request-log.ts +++ b/src/server/request-log.ts @@ -8,6 +8,7 @@ import { isClientClosedMessage, isCyberPolicyCode, isCyberPolicyMessage, + isRateLimitOrQuotaFailureMessage, upstreamErrorMessageFromPayload, } from "../lib/errors"; import { CODEX_CONFIG_PATH, readRootTomlString } from "../codex/paths"; @@ -842,7 +843,7 @@ function incompleteReasonLabel(reason: string): string { } } -function captureTerminalHttpStatus( +export function captureTerminalHttpStatus( logCtx: RequestLogContext, json: { type?: unknown; @@ -875,6 +876,15 @@ function captureTerminalHttpStatus( logCtx.terminalHttpStatus = 400; return; } + const quota = candidates.some(candidate => ( + typeof candidate?.message === "string" + && candidate.message.trim().length > 0 + && isRateLimitOrQuotaFailureMessage(candidate.message) + )); + if (quota) { + logCtx.terminalHttpStatus = 429; + return; + } if (type !== "response.failed" || !responseError || typeof responseError !== "object") return; const responseCode = responseError.code === null || typeof responseError.code === "string" ? responseError.code diff --git a/src/server/responses/compact.ts b/src/server/responses/compact.ts index a742fad98d..da03acbae3 100644 --- a/src/server/responses/compact.ts +++ b/src/server/responses/compact.ts @@ -1088,7 +1088,12 @@ export async function handleResponsesCompact( } } } - return buffered; + if (buffered.status !== 404) { + return buffered; + } + // Upstream returned 404 on /responses/compact (e.g. canonical ChatGPT Codex forward backend + // does not serve /responses/compact; it only supports compaction via POST /responses turns). + // Fall through to the routed synthetic-compaction turn below. } finally { releaseUpstreamHostAdmission(compactHostAdmissionLease); releaseCodexAuthContextProbeLease(authCtx); @@ -1105,7 +1110,7 @@ export async function handleResponsesCompact( // the completed event back into the v1 compact JSON contract below. Combo-dispatched // turns also go out as SSE: failover can land on a canonical child that rejects a // non-streaming turn, and every combo-capable provider already serves streaming traffic. - stream: accountGatedCompactWireModel || route.combo ? true : false, + stream: isCanonicalOpenAiForwardProvider(route.provider) || accountGatedCompactWireModel || route.combo ? true : false, input: [...inputItems, { type: "compaction_trigger" }], }; const internalHeaders = new Headers({ "content-type": "application/json" }); diff --git a/src/server/responses/core.ts b/src/server/responses/core.ts index b48438deb5..44e4a340e7 100644 --- a/src/server/responses/core.ts +++ b/src/server/responses/core.ts @@ -1533,6 +1533,23 @@ export function codexForwardTerminalOutcomeRecorder( if (!usesCodexForwardPoolAuth(authCtx, provider)) return undefined; return (status, httpStatusOverride) => { if (status === "incomplete") { + const isQuotaOrRateLimit = Boolean( + (logCtx?.upstreamError && isRateLimitOrQuotaFailureMessage(logCtx.upstreamError)) + || logCtx?.terminalHttpStatus === 429 + || httpStatusOverride === 429 + ); + if (isQuotaOrRateLimit) { + recordCodexUpstreamOutcome(config, authCtx.accountId, 429, { + threadId: authCtx.affinityKey, + fixedAccount: authCtx.fixedAccount, + modelId, + probeLeaseId: codexProbeLeaseId(authCtx), + probeQuotaScope: codexProbeQuotaScope(authCtx), + writerGeneration: authCtx.writerGeneration, + ...(authCtx.kind === "pool" ? { credentialGeneration: authCtx.generation } : {}), + }); + return; + } // Normal limit/content-filter/stall terminal — the account served the // request. Don't penalize account health; record success to clear any // prior soft-avoid so a healthy account isn't stuck avoided. @@ -5143,11 +5160,12 @@ async function handleResponsesInner( if (terminalBodyWillRecord) { options.setTerminalOutcomeRecorder?.((status, httpStatusOverride) => { terminalRecorder(status, httpStatusOverride); - if (status === "failed") { + if (status === "failed" || status === "incomplete") { const quotaFailureMessage = httpStatusOverride === 429 || httpStatusOverride === 402 || logCtx.terminalHttpStatus === 429 || logCtx.terminalHttpStatus === 402 - ? (httpStatusOverride ?? logCtx.terminalHttpStatus) + || (logCtx.upstreamError && isRateLimitOrQuotaFailureMessage(logCtx.upstreamError)) + ? (httpStatusOverride ?? logCtx.terminalHttpStatus ?? 429) : undefined; if (!isFixedCodexAccount(authCtx) && quotaFailureMessage !== undefined) { recordSubagentQuotaFailureForThreadSpawn( @@ -5354,11 +5372,12 @@ async function handleResponsesInner( const reportNativeTerminal = recordTerminalOutcomes ? (status: ResponsesTerminalStatus, httpStatusOverride?: number) => { terminalRecorder?.(status, httpStatusOverride); - if (status === "failed") { + if (status === "failed" || status === "incomplete") { const quotaFailureMessage = httpStatusOverride === 429 || httpStatusOverride === 402 || logCtx.terminalHttpStatus === 429 || logCtx.terminalHttpStatus === 402 - ? (httpStatusOverride ?? logCtx.terminalHttpStatus) + || (logCtx.upstreamError && isRateLimitOrQuotaFailureMessage(logCtx.upstreamError)) + ? (httpStatusOverride ?? logCtx.terminalHttpStatus ?? 429) : undefined; if (!isFixedCodexAccount(authCtx) && quotaFailureMessage !== undefined) { recordSubagentQuotaFailureForThreadSpawn( @@ -5447,11 +5466,12 @@ async function handleResponsesInner( // client-cancel (no terminal seen) is finalized separately via consumeForInspection's onCancel. const reportNativeTerminal = (status: ResponsesTerminalStatus, httpStatusOverride?: number) => { terminalRecorder?.(status, httpStatusOverride); - if (status === "failed") { + if (status === "failed" || status === "incomplete") { const quotaFailureMessage = httpStatusOverride === 429 || httpStatusOverride === 402 || logCtx.terminalHttpStatus === 429 || logCtx.terminalHttpStatus === 402 - ? (httpStatusOverride ?? logCtx.terminalHttpStatus) + || (logCtx.upstreamError && isRateLimitOrQuotaFailureMessage(logCtx.upstreamError)) + ? (httpStatusOverride ?? logCtx.terminalHttpStatus ?? 429) : undefined; if (!isFixedCodexAccount(authCtx) && quotaFailureMessage !== undefined) { recordSubagentQuotaFailureForThreadSpawn( diff --git a/tests/responses/responses-forward-incomplete-quota.test.ts b/tests/responses/responses-forward-incomplete-quota.test.ts new file mode 100644 index 0000000000..454790f1b0 --- /dev/null +++ b/tests/responses/responses-forward-incomplete-quota.test.ts @@ -0,0 +1,172 @@ +import { describe, expect, test, beforeEach } from "bun:test"; +import { captureTerminalHttpStatus } from "../../src/server/request-log"; +import { codexForwardTerminalOutcomeRecorder } from "../../src/server/responses/core"; +import { + clearCodexUpstreamHealth, + getCodexAccountCooldownUntil, +} from "../../src/codex/routing"; +import type { CodexAuthContext } from "../../src/codex/auth-context"; +import type { OcxConfig, OcxProviderConfig } from "../../src/types"; + +describe("forward incomplete quota failover handling", () => { + beforeEach(() => { + clearCodexUpstreamHealth(); + }); + + test("captureTerminalHttpStatus records 429 when response.incomplete has quota error message", () => { + const logCtx: Record = {}; + captureTerminalHttpStatus(logCtx as any, { + type: "response.incomplete", + response: { + incomplete_details: { + reason: "usage_limit_reached", + message: "The usage limit has been reached", + }, + }, + }); + expect(logCtx.terminalHttpStatus).toBe(429); + }); + + test("codexForwardTerminalOutcomeRecorder trips cooldown on incomplete quota terminal", () => { + const config = { + codexAccounts: [ + { id: "pool-a", email: "pool-a@example.com", isMain: false }, + { id: "pool-b", email: "pool-b@example.com", isMain: false }, + ], + activeCodexAccountId: "pool-a", + } as unknown as OcxConfig; + + const authCtx: CodexAuthContext = { + kind: "pool", + accountId: "pool-a", + generation: 1, + affinityKey: "thread_123", + fixedAccount: false, + } as unknown as CodexAuthContext; + + const provider: OcxProviderConfig = { + adapter: "openai-responses", + baseUrl: "https://chatgpt.com/backend-api/codex", + authMode: "forward", + }; + + const logCtx = { + upstreamError: "The usage limit has been reached", + }; + + const recorder = codexForwardTerminalOutcomeRecorder( + config, + authCtx, + provider, + "gpt-5.6", + logCtx as any, + ); + expect(recorder).toBeDefined(); + + recorder!("incomplete"); + + // The account should now be on cooldown due to 429 + const cooldownUntil = getCodexAccountCooldownUntil("pool-a"); + expect(cooldownUntil).toBeGreaterThan(Date.now()); + }); + + test("codexForwardTerminalOutcomeRecorder records 200 on standard incomplete (e.g. max tokens)", () => { + const config = { + codexAccounts: [ + { id: "pool-a", email: "pool-a@example.com", isMain: false }, + ], + activeCodexAccountId: "pool-a", + } as unknown as OcxConfig; + + const authCtx: CodexAuthContext = { + kind: "pool", + accountId: "pool-a", + generation: 1, + affinityKey: "thread_123", + fixedAccount: false, + } as unknown as CodexAuthContext; + + const provider: OcxProviderConfig = { + adapter: "openai-responses", + baseUrl: "https://chatgpt.com/backend-api/codex", + authMode: "forward", + }; + + const logCtx = { + terminalIncompleteReason: "max_output_tokens", + }; + + const recorder = codexForwardTerminalOutcomeRecorder( + config, + authCtx, + provider, + "gpt-5.6", + logCtx as any, + ); + expect(recorder).toBeDefined(); + + recorder!("incomplete"); + + // Normal incomplete terminal does not penalize account health + const cooldownUntil = getCodexAccountCooldownUntil("pool-a"); + expect(cooldownUntil).toBeNull(); + }); + + test("handleResponsesCompact falls through to routed synthetic compaction when upstream returns 404 on /responses/compact", async () => { + const { handleResponsesCompact } = await import("../../src/server/responses/compact"); + const originalFetch = globalThis.fetch; + const requestedUrls: string[] = []; + globalThis.fetch = (async (url: string | URL | Request) => { + const urlStr = typeof url === "string" ? url : url instanceof URL ? url.toString() : (url as Request).url; + requestedUrls.push(urlStr); + if (urlStr.includes("/responses/compact")) { + return new Response(JSON.stringify({ detail: "Not Found" }), { + status: 404, + headers: { "content-type": "application/json" }, + }); + } + const payload = { + id: "resp_1", + status: "completed", + output: [{ type: "compaction", encrypted_content: "opaque_blob_xyz" }], + }; + return new Response(JSON.stringify(payload), { + status: 200, + headers: { "content-type": "application/json" }, + }); + }) as typeof fetch; + + try { + const config = { + defaultProvider: "openai-apikey", + providers: { + "openai-apikey": { + adapter: "openai-responses", + baseUrl: "https://api.openai.com/v1", + authMode: "key", + apiKey: "test-key", + }, + }, + } as unknown as OcxConfig; + + const req = new Request("http://localhost/v1/responses/compact", { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ + model: "openai-apikey/gpt-5.6", + input: [{ type: "message", role: "user", content: [{ type: "input_text", text: "hi" }] }], + }), + }); + + const res = await handleResponsesCompact(req, config, {} as any); + expect(res.status).toBe(200); + expect(requestedUrls.some(u => u.includes("/responses/compact"))).toBe(true); + expect(requestedUrls.some(u => u.endsWith("/responses") || u.includes("/v1/responses"))).toBe(true); + const json = await res.json() as any; + expect(json.output).toBeDefined(); + expect(json.output[0].type).toBe("compaction"); + } finally { + globalThis.fetch = originalFetch; + } + }); +});