diff --git a/docs-site/src/content/docs/reference/architecture.md b/docs-site/src/content/docs/reference/architecture.md index 96a3aec6dc6..eb856811e19 100644 --- a/docs-site/src/content/docs/reference/architecture.md +++ b/docs-site/src/content/docs/reference/architecture.md @@ -200,3 +200,13 @@ The internal model lives in `types.ts`: `OcxParsedRequest`, `OcxContext`, the `O `OcxContentPart` (text / image), `OcxToolCall`, `OcxTool`, `AdapterEvent`, and the config types (`OcxConfig`, `OcxProviderConfig`). Two helpers are widely used: `namespacedToolName()` and `modelInList()` (tolerant `:size`-tag matching for `noVisionModels` / `noReasoningModels`). + + +### Incomplete quota terminals + +A native forward response that ends with quota or rate-limit evidence in an +`incomplete` terminal records account quota failure and spawn-fallback health. +Structured `incomplete_details.reason` and error codes are accepted without a +message; ordinary output-limit, filtering, steering and stall incompletes do not +cool an account. Cyber-policy classification retains precedence. The terminal is +not replayed after output, and fixed-account request selection remains fixed. diff --git a/scripts/test-layout/layout.json b/scripts/test-layout/layout.json index 0fbe7cf746b..67093dcd07a 100644 --- a/scripts/test-layout/layout.json +++ b/scripts/test-layout/layout.json @@ -1021,6 +1021,7 @@ "responses-context-overflow.test.ts": "responses", "responses-custom-tool-guidance.test.ts": "responses", "responses-custom-tool-repair.test.ts": "responses", + "responses-forward-incomplete-quota.test.ts": "responses", "responses-function-tool-repair.test.ts": "responses", "responses-fetch-helpers-boundary.test.ts": "responses", "responses-field-backfill.test.ts": "responses", diff --git a/src/server/request-log.ts b/src/server/request-log.ts index a4c942bd883..eea04c684dd 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"; @@ -852,7 +853,7 @@ function captureTerminalHttpStatus( last_error?: { type?: unknown; code?: unknown; message?: unknown }; response?: { error?: { type?: unknown; code?: unknown; message?: unknown }; - incomplete_details?: { code?: unknown; message?: unknown }; + incomplete_details?: { code?: unknown; message?: unknown; reason?: unknown }; }; }, ): void { @@ -861,7 +862,9 @@ function captureTerminalHttpStatus( if (type !== "response.failed" && type !== "response.incomplete" && type !== "error") return; const responseError = json.response?.error; const responseDetails = json.response?.incomplete_details; - const candidates = [json.error, json.last_error, responseError, responseDetails, json]; + const candidates: Array<{ type?: unknown; code?: unknown; message?: unknown } | undefined> = [ + json.error, json.last_error, responseError, responseDetails, json, + ]; const policy = candidates.some(candidate => ( candidate?.code === null || typeof candidate?.code === "string" ) && isCyberPolicyCode(candidate.code as string | null | undefined)) @@ -875,6 +878,29 @@ function captureTerminalHttpStatus( logCtx.terminalHttpStatus = 400; return; } + // A quota terminal can carry only a structured reason, without an error message. + // Keep this separate from normal output limits and from the policy precedence above. + const quotaTag = (value: unknown): boolean => value === "usage_limit_reached" + || value === "rate_limit_exceeded" || value === "insufficient_quota"; + const structuredRefusal = candidates.some(candidate => [400, 401, 403, 499].includes( + httpStatusFromTerminalError({ + type: typeof candidate?.type === "string" ? candidate.type : undefined, + code: typeof candidate?.code === "string" ? candidate.code : undefined, + }), + )); + const ordinaryIncompleteReason = typeof responseDetails?.reason === "string" + && ["max_output_tokens", "content_filter", "steered", "upstream_stall_timeout", "adapter_eof"].includes(responseDetails.reason); + if (type === "response.incomplete" && !structuredRefusal && (quotaTag(responseDetails?.reason) || candidates.some(candidate => + quotaTag(candidate?.code) + || quotaTag(candidate?.type) || candidate?.type === "rate_limit_error" + || (!ordinaryIncompleteReason && typeof candidate?.message === "string" && isRateLimitOrQuotaFailureMessage(candidate.message)) + ))) { + // The shared quota classifier also accepts a numeric HTTP status as its message. + // Preserve explicit payment-required evidence rather than relabeling it as 429. + logCtx.terminalHttpStatus = candidates.some(candidate => typeof candidate?.message === "string" + && Number(candidate.message.trim()) === 402) ? 402 : 429; + return; + } if (type !== "response.failed" || !responseError || typeof responseError !== "object") return; const responseCode = responseError.code === null || typeof responseError.code === "string" ? responseError.code @@ -903,6 +929,9 @@ export function httpStatusForRequestLogTerminal( status: ResponsesTerminalStatus, logCtx?: RequestLogContext, ): number { + if (status === "incomplete" && (logCtx?.terminalHttpStatus === 429 || logCtx?.terminalHttpStatus === 402)) { + return logCtx.terminalHttpStatus; + } /** * [Decision Log] * - 목적과 의도: Keep request logs aligned with the successful HTTP/SSE contract. diff --git a/src/server/responses/core.ts b/src/server/responses/core.ts index debba5c707e..526010f31ed 100644 --- a/src/server/responses/core.ts +++ b/src/server/responses/core.ts @@ -1534,7 +1534,9 @@ export function codexForwardTerminalOutcomeRecorder( ): ((status: ResponsesTerminalStatus, httpStatusOverride?: number) => void) | undefined { if (!usesCodexForwardPoolAuth(authCtx, provider)) return undefined; return (status, httpStatusOverride) => { - if (status === "incomplete") { + const quotaStatus = [httpStatusOverride, logCtx?.terminalHttpStatus] + .find(value => value === 429 || value === 402); + if (status === "incomplete" && quotaStatus === undefined) { // 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. @@ -1559,7 +1561,7 @@ export function codexForwardTerminalOutcomeRecorder( // the parent's terminalHttpStatus so the semantic status is not lost. const outcome = status === "completed" ? 200 - : (httpStatusOverride ?? logCtx?.terminalHttpStatus ?? 502); + : (quotaStatus ?? httpStatusOverride ?? logCtx?.terminalHttpStatus ?? 502); recordCodexUpstreamOutcome(config, authCtx.accountId, outcome, { threadId: authCtx.affinityKey, fixedAccount: authCtx.fixedAccount, @@ -2668,7 +2670,20 @@ export async function handleComboResponses( attemptRetained = true; }; let consumedChildFailure: ConsumedComboFailure | undefined; - const callbackGate = createChildPassthroughCallbackGate(options); + const callbackGate = createChildPassthroughCallbackGate({ + ...options, + onNativePassthroughTerminal: status => { + // A committed stream can acquire terminal metadata after preflight copied + // the child log. Publish it before the outer logger finalizes, but only + // through the gate: discarded attempts must never affect the parent. + // Undefined child fields must preserve metadata already inspected by WS. + if (childLog.terminalHttpStatus !== undefined) logCtx.terminalHttpStatus = childLog.terminalHttpStatus; + if (childLog.terminalIncompleteReason !== undefined) logCtx.terminalIncompleteReason = childLog.terminalIncompleteReason; + if (childLog.terminalErrorCode !== undefined) logCtx.terminalErrorCode = childLog.terminalErrorCode; + if (childLog.upstreamError !== undefined) logCtx.upstreamError = childLog.upstreamError; + options.onNativePassthroughTerminal?.(status); + }, + }); let response: Response; try { const currentTargetProvider = pick.target.provider; @@ -5359,12 +5374,9 @@ async function handleResponsesInner( if (terminalBodyWillRecord) { options.setTerminalOutcomeRecorder?.((status, httpStatusOverride) => { terminalRecorder(status, httpStatusOverride); - if (status === "failed") { - const quotaFailureMessage = httpStatusOverride === 429 || httpStatusOverride === 402 - || logCtx.terminalHttpStatus === 429 - || logCtx.terminalHttpStatus === 402 - ? (httpStatusOverride ?? logCtx.terminalHttpStatus) - : undefined; + if (status === "failed" || status === "incomplete") { + const quotaFailureMessage = [httpStatusOverride, logCtx.terminalHttpStatus] + .find(value => value === 429 || value === 402); if (!isFixedCodexAccount(authCtx) && quotaFailureMessage !== undefined) { recordSubagentQuotaFailureForThreadSpawn( req.headers, @@ -5570,12 +5582,9 @@ async function handleResponsesInner( const reportNativeTerminal = recordTerminalOutcomes ? (status: ResponsesTerminalStatus, httpStatusOverride?: number) => { terminalRecorder?.(status, httpStatusOverride); - if (status === "failed") { - const quotaFailureMessage = httpStatusOverride === 429 || httpStatusOverride === 402 - || logCtx.terminalHttpStatus === 429 - || logCtx.terminalHttpStatus === 402 - ? (httpStatusOverride ?? logCtx.terminalHttpStatus) - : undefined; + if (status === "failed" || status === "incomplete") { + const quotaFailureMessage = [httpStatusOverride, logCtx.terminalHttpStatus] + .find(value => value === 429 || value === 402); if (!isFixedCodexAccount(authCtx) && quotaFailureMessage !== undefined) { recordSubagentQuotaFailureForThreadSpawn( req.headers, @@ -5663,12 +5672,9 @@ 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") { - const quotaFailureMessage = httpStatusOverride === 429 || httpStatusOverride === 402 - || logCtx.terminalHttpStatus === 429 - || logCtx.terminalHttpStatus === 402 - ? (httpStatusOverride ?? logCtx.terminalHttpStatus) - : undefined; + if (status === "failed" || status === "incomplete") { + const quotaFailureMessage = [httpStatusOverride, logCtx.terminalHttpStatus] + .find(value => value === 429 || value === 402); if (!isFixedCodexAccount(authCtx) && quotaFailureMessage !== undefined) { recordSubagentQuotaFailureForThreadSpawn( req.headers, diff --git a/structure/04_transports-and-sidecars.md b/structure/04_transports-and-sidecars.md index 4716c447bb6..5196684d210 100644 --- a/structure/04_transports-and-sidecars.md +++ b/structure/04_transports-and-sidecars.md @@ -1645,3 +1645,13 @@ dispatch. Selection revisions fence stale retries and reselection; request ident actual committed account/key. Generic proactive selection is opt-in and preserves a healthy active account, while reactive429 recovery remains enabled even with the pool off. Post-commit selection events immediately invalidate dashboard roster state; see`05_gui-and-management-api.md`. + + +### Incomplete quota terminals + +A native forward response that ends with quota or rate-limit evidence in an +`incomplete` terminal records account quota failure and spawn-fallback health. +Structured `incomplete_details.reason` and error codes are accepted without a +message; ordinary output-limit, filtering, steering and stall incompletes do not +cool an account. Cyber-policy classification retains precedence. The terminal is +not replayed after output, and fixed-account request selection remains fixed. diff --git a/tests/fixtures/test-layout-expected.json b/tests/fixtures/test-layout-expected.json index db2583b00b7..0823ba202c0 100644 --- a/tests/fixtures/test-layout-expected.json +++ b/tests/fixtures/test-layout-expected.json @@ -856,6 +856,7 @@ "responses-context-overflow.test.ts": "responses", "responses-custom-tool-guidance.test.ts": "responses", "responses-custom-tool-repair.test.ts": "responses", + "responses-forward-incomplete-quota.test.ts": "responses", "responses-function-tool-repair.test.ts": "responses", "responses-fetch-helpers-boundary.test.ts": "responses", "responses-field-backfill.test.ts": "responses", 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 00000000000..0d93e3d5e44 --- /dev/null +++ b/tests/responses/responses-forward-incomplete-quota.test.ts @@ -0,0 +1,337 @@ +import { afterEach, describe, expect, test } from "bun:test"; +import { mkdtempSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { getDefaultConfig } from "../../src/config"; +import { captureConfigGeneration } from "../../src/lib/state-store-sweeper"; +import { clearCodexUpstreamHealth, clearThreadAccountMap, getCodexAccountCooldownUntil } from "../../src/codex/routing"; +import type { CodexAuthContext } from "../../src/codex/auth-context"; +import { codexForwardTerminalOutcomeRecorder } from "../../src/server/responses/core"; +import { + httpStatusForRequestLogTerminal, + inspectResponseLogSsePayload, + type RequestLogContext, +} from "../../src/server/request-log"; +import type { OcxConfig, OcxProviderConfig } from "../../src/types"; +import { saveCodexAccountCredential } from "../../src/codex/account-store"; +import { clearAccountQuota, updateAccountQuota } from "../../src/codex/quota"; +import { + isModelHealthBlocked, + resetSubagentModelFallbackStateForTests, + setSubagentQuotaPrimeForTests, +} from "../../src/codex/subagent-model-fallback"; +import { handleResponses } from "../../src/server/responses"; +import type { HandleResponsesOptions } from "../../src/server/responses/core"; +import { isEagerRelaySseResponse } from "../../src/server/relay"; +import { sendResponseToWebSocket, type WsData } from "../../src/server/ws-bridge"; +import { installIsolatedCodexHome } from "../helpers/isolated-codex-home"; +import { removeTreeWithRetry } from "../helpers/remove-tree"; +import { INTERNAL_DEADLINE_MS, SERVER_BUDGET_MS } from "../helpers/test-budget"; + +const provider: OcxProviderConfig = { + adapter: "openai-responses", + baseUrl: "https://chatgpt.com/backend-api/codex", + authMode: "forward", +}; + +function auth(fixedAccount = false): CodexAuthContext { + return { + kind: "pool", accountId: "incomplete-quota-fixture", accessToken: "test-token", + chatgptAccountId: "test-account", generation: 1, + writerGeneration: captureConfigGeneration(), fixedAccount, + }; +} + +function inspect(response: Record): RequestLogContext { + const log: RequestLogContext = { model: "gpt-test", provider: "openai" }; + inspectResponseLogSsePayload(log, JSON.stringify({ type: "response.incomplete", response })); + return log; +} + +afterEach(() => clearCodexUpstreamHealth()); + +describe("incomplete quota terminal attribution", () => { + for (const response of [ + { incomplete_details: { reason: "usage_limit_reached" } }, + { incomplete_details: { reason: "rate_limit_exceeded" } }, + { incomplete_details: { reason: "insufficient_quota" } }, + { error: { code: "usage_limit_reached" } }, + { error: { type: "rate_limit_error" } }, + { incomplete_details: { message: "The usage limit has been reached" } }, + ]) { + test(`SSE inspection records quota health for ${JSON.stringify(response)}`, () => { + const log = inspect(response); + expect(log.terminalHttpStatus).toBe(429); + expect(httpStatusForRequestLogTerminal("incomplete", log)).toBe(429); + const record = codexForwardTerminalOutcomeRecorder(getDefaultConfig(), auth(), provider, "gpt-test", log); + expect(record).toBeDefined(); + record!("incomplete"); + expect(getCodexAccountCooldownUntil("incomplete-quota-fixture")).toBeGreaterThan(Date.now()); + }); + } + + for (const reason of ["max_output_tokens", "content_filter", "steered", "upstream_stall_timeout", "unknown"]) { + test(`ordinary ${reason} incomplete does not cool the account`, () => { + const log = inspect({ incomplete_details: { reason } }); + expect(log.terminalHttpStatus).toBeUndefined(); + codexForwardTerminalOutcomeRecorder(getDefaultConfig(), auth(), provider, "gpt-test", log)!("incomplete"); + expect(getCodexAccountCooldownUntil("incomplete-quota-fixture")).toBeNull(); + }); + } + + test("policy refusal takes precedence over conflicting quota details", () => { + const log = inspect({ + error: { code: "cyber_policy", message: "blocked" }, + incomplete_details: { reason: "usage_limit_reached" }, + }); + expect(log.terminalHttpStatus).toBe(400); + codexForwardTerminalOutcomeRecorder(getDefaultConfig(), auth(), provider, "gpt-test", log)!("incomplete"); + expect(getCodexAccountCooldownUntil("incomplete-quota-fixture")).toBeNull(); + }); + + test("a generic transport override does not erase captured quota evidence", () => { + const log = inspect({ incomplete_details: { reason: "usage_limit_reached" } }); + codexForwardTerminalOutcomeRecorder(getDefaultConfig(), auth(), provider, "gpt-test", log)!("incomplete", 502); + expect(getCodexAccountCooldownUntil("incomplete-quota-fixture")).toBeGreaterThan(Date.now()); + }); + + for (const status of [402, 429]) { + test(`parent terminal override ${status} reaches the child recorder`, () => { + // Combo/WS inspection owns the parent log, while this recorder closes over a child log. + const child: RequestLogContext = { model: "gpt-test", provider: "openai" }; + codexForwardTerminalOutcomeRecorder(getDefaultConfig(), auth(true), provider, "gpt-test", child)!("incomplete", status); + expect(getCodexAccountCooldownUntil("incomplete-quota-fixture")).toBeGreaterThan(Date.now()); + }); + } +}); + +type ReporterPath = "parent-recorder" | "guarded-ws" | "native-sse"; + +// Drive the endpoint and its real transport/inspection owners. Only the external +// Codex destination is redirected; the recorder and spawn health store stay real. +async function exerciseSpawnReporter(path: ReporterPath): Promise { + const realFetch = globalThis.fetch; + const RealWebSocket = globalThis.WebSocket; + const previousHome = process.env.OPENCODEX_HOME; + const home = mkdtempSync(join(tmpdir(), "ocx-incomplete-quota-")); + const codexHome = installIsolatedCodexHome("ocx-incomplete-quota-codex-"); + process.env.OPENCODEX_HOME = home; + const accountId = "incomplete-quota-endpoint"; + const model = "gpt-test"; + const config: OcxConfig = { + ...getDefaultConfig(), + port: 0, + defaultProvider: "openai", + openaiProviderTierVersion: 2, + streamMode: "legacy-tee", + providers: { openai: { ...provider, codexAccountMode: "pool" } }, + codexAccounts: [{ + id: accountId, email: "quota@example.test", isMain: false, + chatgptAccountId: "acct-quota-endpoint", + }], + activeCodexAccountId: accountId, + }; + let reason = "max_output_tokens"; + let httpDispatches = 0; + let wsDispatches = 0; + const terminal = () => ({ + type: "response.incomplete", + response: { + id: `resp-${path}-${reason}`, object: "response", status: "incomplete", + model, output: [], incomplete_details: { reason }, + }, + }); + const upstream = Bun.serve({ + hostname: "127.0.0.1", + port: 0, + fetch(req, server) { + if (req.headers.get("upgrade") === "websocket" && server.upgrade(req)) return; + httpDispatches++; + return new Response(`event: response.incomplete\ndata: ${JSON.stringify(terminal())}\n\n`, { + headers: { "content-type": "text/event-stream" }, + }); + }, + websocket: { + message(ws, message) { + const request = JSON.parse(String(message)); + expect(request.type).toBe("response.create"); + expect(request.model).toBe(model); + wsDispatches++; + ws.send(JSON.stringify(terminal())); + }, + }, + }); + let logCtx: RequestLogContext = { model: "", provider: "" }; + const resolved: { auth?: CodexAuthContext } = {}; + let parentTerminal: string | undefined; + let eager: boolean | undefined; + let registered: Parameters>[0]; + let reportTerminal: (status: string) => void = () => {}; + let rejectTerminal: (error: unknown) => void = () => {}; + const options = (): HandleResponsesOptions => ({ + // Use the existing runtime seam: HTTP fixtures must not accidentally select + // WS on a newer Bun, and the WS fixture must exercise the guarded relay. + codexWsRuntimeIdentity: path === "guarded-ws" ? "1.4.0" : "1.3.14", + recordTerminalOutcomes: path !== "parent-recorder", + onCodexAuthContextResolved: context => { resolved.auth = context; }, + setTerminalOutcomeRecorder: recorder => { registered = recorder; }, + onNativePassthroughTerminal: status => { + if (path === "parent-recorder") parentTerminal = status; + else reportTerminal(status); + }, + }); + const endpoint = Bun.serve({ + hostname: "127.0.0.1", + port: 0, + async fetch(req, server) { + if (path === "parent-recorder" && server.upgrade(req, { data: { headers: req.headers } })) return; + const response = await handleResponses(req, config, logCtx, options()); + eager = isEagerRelaySseResponse(response); + return response; + }, + websocket: { + async message(ws, message) { + try { + const payload = JSON.parse(String(message)); + const response = await handleResponses(new Request("http://localhost/v1/responses", { + method: "POST", headers: ws.data.headers, + body: JSON.stringify({ ...payload, stream: true }), + }), config, logCtx, { ...options(), inboundTransport: "websocket" }); + expect(response.status).toBe(200); + expect(registered).toBeDefined(); + // Same ownership as server/index.ts: the bridge inspects first, then + // calls the recorder registered by core. Inject 502 only at this + // existing override seam to prove it cannot erase captured typed 429. + await sendResponseToWebSocket(ws, response, () => true, { + onSsePayload: payload => inspectResponseLogSsePayload(logCtx, payload), + onTerminal: status => registered!(status, 502), + }); + expect(parentTerminal).toBe("incomplete"); + reportTerminal(parentTerminal!); + } catch (error) { + rejectTerminal(error); + } + }, + }, + }); + let client: WebSocket | undefined; + try { + clearCodexUpstreamHealth(); + clearThreadAccountMap(); + clearAccountQuota(); + resetSubagentModelFallbackStateForTests(); + setSubagentQuotaPrimeForTests(async () => {}); + saveCodexAccountCredential(accountId, { + accessToken: "endpoint-token", refreshToken: "endpoint-refresh", + expiresAt: Date.now() + 60 * 60_000, chatgptAccountId: "acct-quota-endpoint", + }); + updateAccountQuota(accountId, 10); + globalThis.fetch = ((input: RequestInfo | URL, init?: RequestInit) => { + const url = new URL(input instanceof Request ? input.url : String(input)); + if (url.origin === "https://chatgpt.com" && url.pathname === "/backend-api/codex/responses") { + return realFetch(new URL("/responses", upstream.url), init); + } + if (url.origin === endpoint.url.origin) return realFetch(input, init); + throw new Error(`Unexpected quota fixture fetch: ${url.origin}${url.pathname}`); + }) as typeof fetch; + globalThis.WebSocket = new Proxy(RealWebSocket, { + construct(target, args) { + const url = new URL(String(args[0])); + if (url.origin === "wss://chatgpt.com" && url.pathname === "/backend-api/codex/responses") { + return Reflect.construct(target, [upstream.url.toString().replace("http:", "ws:"), ...args.slice(1)]); + } + throw new Error(`Unexpected quota fixture WebSocket: ${url.origin}${url.pathname}`); + }, + }); + + // Ordinary incomplete comes first, so its negative assertion cannot be + // masked by clearing health produced by the quota terminal. + for (const quota of [false, true]) { + reason = quota ? "usage_limit_reached" : "max_output_tokens"; + logCtx = { model: "", provider: "" }; + resolved.auth = undefined; + parentTerminal = undefined; + registered = undefined; + expect(isModelHealthBlocked(model, config, accountId)).toBe(false); + let timer: ReturnType | undefined; + const reported = new Promise((resolve, reject) => { + reportTerminal = resolve; + rejectTerminal = reject; + timer = setTimeout(() => reject(new Error(`${path}: terminal reporter did not run`)), INTERNAL_DEADLINE_MS); + }); + const headers = { + "content-type": "application/json", authorization: "Bearer inbound-fixture", + "x-openai-subagent": "collab_spawn", + }; + const body = { model, input: "hello", stream: true }; + try { + const deliver = async () => { + if (path === "parent-recorder") { + // Real downstream WS; construction bypasses only our upstream redirect. + client = new RealWebSocket(endpoint.url.toString().replace("http:", "ws:") + "v1/responses", { + headers, + } as unknown as string[]); + client.addEventListener("open", () => client!.send(JSON.stringify({ type: "response.create", ...body }))); + client.addEventListener("error", () => rejectTerminal(new Error("endpoint WebSocket failed"))); + } else { + const response = await realFetch(new URL("/v1/responses", endpoint.url), { + method: "POST", headers, body: JSON.stringify(body), + signal: AbortSignal.timeout(INTERNAL_DEADLINE_MS), + }); + expect(response.status).toBe(200); + expect(await response.text()).toContain('"status":"incomplete"'); + expect(eager).toBe(path === "guarded-ws"); + } + }; + const [status] = await Promise.all([reported, deliver()]); + expect(status).toBe("incomplete"); + expect(resolved).toMatchObject({ auth: { kind: "pool", accountId } }); + expect(resolved).not.toMatchObject({ auth: { fixedAccount: true } }); + expect(logCtx.terminalHttpStatus).toBe(quota ? 429 : undefined); + // This is the actual store read by selectAvailableSubagentModel, separate + // from pool cooldown: removing any one reporter's spawn write fails here. + expect(isModelHealthBlocked(model, config, accountId)).toBe(quota); + expect(isModelHealthBlocked(model, config, "another-account")).toBe(false); + if (quota) expect(getCodexAccountCooldownUntil(accountId)).toBeGreaterThan(Date.now()); + else expect(getCodexAccountCooldownUntil(accountId)).toBeNull(); + } finally { + clearTimeout(timer); + client?.close(); + client = undefined; + } + } + expect(wsDispatches).toBe(path === "guarded-ws" ? 2 : 0); + expect(httpDispatches).toBe(path === "guarded-ws" ? 0 : 2); + } finally { + client?.close(); + await endpoint.stop(true); + await upstream.stop(true); + globalThis.fetch = realFetch; + globalThis.WebSocket = RealWebSocket; + clearCodexUpstreamHealth(); + clearThreadAccountMap(); + clearAccountQuota(); + resetSubagentModelFallbackStateForTests(); + codexHome.restore(); + if (previousHome === undefined) delete process.env.OPENCODEX_HOME; + else process.env.OPENCODEX_HOME = previousHome; + removeTreeWithRetry(home); + } +} + +describe("incomplete quota endpoint reporter wiring", () => { + test("registered parent reporter preserves typed quota over 502 and updates spawn health", async () => { + await exerciseSpawnReporter("parent-recorder"); + }, { timeout: SERVER_BUDGET_MS }); + + test("guarded native WS reporter updates spawn health only for quota incomplete", async () => { + await exerciseSpawnReporter("guarded-ws"); + }, { timeout: SERVER_BUDGET_MS }); + + // core always applies a field-backfill rewrite; win32 therefore forces eager + // before the tee reporter regardless of streamMode (Bun#32111). Do not label + // that eager path as native-SSE reporter coverage on Windows. + test.skipIf(process.platform === "win32")("regular native SSE reporter updates spawn health only for quota incomplete", async () => { + await exerciseSpawnReporter("native-sse"); + }, { timeout: SERVER_BUDGET_MS }); +}); diff --git a/tests/server/server-combo-failover-e2e.test.ts b/tests/server/server-combo-failover-e2e.test.ts index 14cdc0ceab3..e4523aabb3b 100644 --- a/tests/server/server-combo-failover-e2e.test.ts +++ b/tests/server/server-combo-failover-e2e.test.ts @@ -17,7 +17,7 @@ import { XAI_OAUTH_DISCOVERY_URL } from "../../src/oauth/xai"; import { XAI_GROK_CLI_BASE_URL } from "../../src/providers/xai-transport"; import type { AdapterEvent, OcxConfig, OcxProviderConfig, OcxProviderContinuationState } from "../../src/types"; import { installIsolatedCodexHome, type IsolatedCodexHome } from "../helpers/isolated-codex-home"; -import { clearRequestLogsForTests, hydrateRequestLogsFromDisk, type RequestLogContext } from "../../src/server/request-log"; +import { clearRequestLogsForTests, hydrateRequestLogsFromDisk, httpStatusForRequestLogTerminal, inspectResponseLogSsePayload, type RequestLogContext } from "../../src/server/request-log"; import { responseWithDeferredRequestLog } from "../../src/server/relay"; import { readUsageEntries } from "../../src/usage/log"; import { saveCodexAccountCredential } from "../../src/codex/account-store"; @@ -414,6 +414,23 @@ async function within(promise: Promise, ms = 2_000): Promise { } } +function heldNativeTerminal(payload: Record) { + const release = deferred(); + const encoder = new TextEncoder(); + const upstream = serve(() => new Response(new ReadableStream({ + async start(controller) { + controller.enqueue(encoder.encode(`event: response.output_text.delta\ndata: ${JSON.stringify({ + type: "response.output_text.delta", item_id: "msg_late", output_index: 0, + content_index: 0, delta: "already visible", + })}\n\n`)); + await release.promise; + controller.enqueue(encoder.encode(`event: ${payload.type}\ndata: ${JSON.stringify(payload)}\n\n`)); + controller.close(); + }, + }), { headers: { "content-type": "text/event-stream" } })); + return { upstream, release: release.resolve }; +} + describe("server combo failover 030 activation matrix", () => { test("dispatches a selected concrete target despite a shadowing combo alias", async () => { const hits: string[] = []; @@ -581,6 +598,79 @@ describe("server combo failover 030 activation matrix", () => { } }); + for (const scenario of [ + { + name: "quota incomplete", status: "incomplete", logStatus: 429, + details: { incomplete_details: { reason: "usage_limit_reached" }, error: { message: "quota exhausted after output" } }, + message: "quota exhausted after output", + }, + { + name: "normal output limit", status: "incomplete", logStatus: 200, + details: { incomplete_details: { reason: "max_output_tokens" }, error: { message: "output limit reached" } }, + message: "output limit reached", + }, + { + name: "policy refusal", status: "failed", logStatus: 400, + details: { error: { code: "cyber_policy", message: "blocked by cyber policy" } }, + message: "blocked by cyber policy", + }, + ]) { + test(`late committed native ${scenario.name} reaches the HTTP combo log`, async () => { + const held = heldNativeTerminal({ + type: `response.${scenario.status}`, + response: { ...responsesSuccess("already visible", "m1"), status: scenario.status, ...scenario.details }, + }); + let backupHits = 0; + const backup = serve(() => { backupHits++; return chatStream("must not replay"); }); + const config = comboConfig({ + a: provider("openai-responses", baseUrl(held.upstream), "key-a"), + b: provider("openai-chat", baseUrl(backup), "key-b"), + }); + config.streamMode = "legacy-tee"; + saveConfig(config); + const server = startServer(0); + try { + const response = await within(fetch(new URL("/v1/responses", server.url), { + method: "POST", headers: { "content-type": "application/json" }, + body: JSON.stringify({ model: "combo/free", input: "hello", stream: true }), + })); + expect(response.status).toBe(200); + const reader = response.body!.getReader(); + const decoder = new TextDecoder(); + let text = ""; + while (!text.includes("already visible")) { + const chunk = await within(reader.read()); + expect(chunk.done).toBe(false); + text += decoder.decode(chunk.value, { stream: true }); + } + // Client-visible content proves preflight committed and copied childLog. + // There is still no terminal to inspect, so no finalized parent receipt. + expect(logsFromApiBody(await (await fetch(new URL("/api/logs?tail=1", server.url))).json())).toHaveLength(0); + held.release(); + for (;;) { + const chunk = await within(reader.read()); + if (chunk.done) break; + text += decoder.decode(chunk.value, { stream: true }); + } + expect(text).toContain(`response.${scenario.status}`); + expect(backupHits).toBe(0); + const logs = logsFromApiBody(await (await fetch(new URL("/api/logs?tail=1", server.url))).json()); + expect(logs).toHaveLength(1); + expect(logs[0]).toMatchObject({ + provider: "combo", model: "combo/free", resolvedModel: "m1", + status: scenario.logStatus, terminalStatus: scenario.status, + closeReason: "terminal", upstreamError: scenario.message, + }); + expect(logs[0]!.attempts).toMatchObject([{ provider: "a", model: "m1", status: scenario.logStatus }]); + expect(logs[0]!.attempts).toHaveLength(1); + if (scenario.status === "failed") expect(logs[0]!.errorCode).toBe("cyber_policy"); + } finally { + held.release(); + await server.stop(true); + } + }); + } + test("terminal SSE failure after output stays on the first target and never replays", async () => { const hits: string[] = []; const a = serve(() => { @@ -2790,12 +2880,15 @@ describe("server combo failover 030 activation matrix", () => { test("failed passthrough child callbacks stay buffered and only B finalizes", async () => { const terminalFrame = (status: "failed" | "completed") => [ `event: response.${status}`, - `data: ${JSON.stringify({ type: `response.${status}`, response: { id: `resp_${status}`, status, output: [] } })}`, + `data: ${JSON.stringify({ type: `response.${status}`, response: { + id: `resp_${status}`, status, output: [], + ...(status === "failed" ? { error: { code: "rate_limit_exceeded", message: "discarded quota failure" } } : {}), + } })}`, "", "", ].join("\n"); const a = serve(() => new Response(terminalFrame("failed"), { - status: 503, + status: 200, headers: { "content-type": "text/event-stream" }, })); const b = serve(() => new Response(terminalFrame("completed"), { @@ -2808,9 +2901,15 @@ describe("server combo failover 030 activation matrix", () => { const finalized = deferred(); const statuses: string[] = []; let cancels = 0; - const response = await post(config, { stream: true }, { + const parent: RequestLogContext = { model: "", provider: "" }; + const snapshots: RequestLogContext[] = []; + const response = await handleResponses(new Request("http://localhost/v1/responses", { + method: "POST", headers: { "content-type": "application/json" }, + body: JSON.stringify({ model: "combo/free", input: "hello", stream: true }), + }), config, parent, { onNativePassthroughTerminal: status => { statuses.push(status); + snapshots.push({ ...parent }); finalized.resolve(); }, onNativePassthroughCancel: () => { cancels += 1; }, @@ -2820,6 +2919,72 @@ describe("server combo failover 030 activation matrix", () => { await within(finalized.promise); expect(statuses).toEqual(["completed"]); expect(cancels).toBe(0); + expect(snapshots).toHaveLength(1); + expect(snapshots[0]).toMatchObject({ provider: "combo", model: "combo/free", resolvedModel: "m2" }); + for (const field of ["terminalHttpStatus", "terminalIncompleteReason", "terminalErrorCode", "upstreamError"] as const) { + expect(snapshots[0]![field]).toBeUndefined(); + } + expect(parent.attempts).toMatchObject([ + { provider: "a", model: "m1", status: 429 }, + { provider: "b", model: "m2" }, + ]); + }); + + test("a metadata-less committed child preserves independently inspected parent metadata and scope", async () => { + const held = heldNativeTerminal({ + type: "response.incomplete", + response: { ...responsesSuccess("already visible", "m1"), status: "incomplete" }, + }); + const config = comboConfig({ a: provider("openai-responses", baseUrl(held.upstream), "key-a") }); + config.streamMode = "legacy-tee"; + const parent: RequestLogContext = { model: "", provider: "" }; + const finalized = deferred(); + const observed: Array<{ status: number; log: RequestLogContext }> = []; + try { + const response = await within(handleResponses(new Request("http://localhost/v1/responses", { + method: "POST", headers: { "content-type": "application/json" }, + body: JSON.stringify({ model: "combo/free", input: "hello", stream: true }), + }), config, parent, { + onNativePassthroughTerminal: status => { + observed.push({ status: httpStatusForRequestLogTerminal(status, parent), log: { ...parent } }); + finalized.resolve(); + }, + })); + expect(response.status).toBe(200); + expect(observed).toHaveLength(0); + const parentTrace = parent.routeDecision; + const parentAttempts = parent.attempts; + parent.firstOutputMs = 17; + // Scope-boundary regression, not a claim about WS scheduling: the WS + // bridge can inspect into its parent log independently of child inspection. + // Populate that state through the real inspector after preflight committed; + // the held child terminal deliberately defines none of these four fields. + inspectResponseLogSsePayload(parent, JSON.stringify({ + type: "response.incomplete", + response: { + incomplete_details: { reason: "usage_limit_reached" }, + error: { message: "parent-observed quota" }, + }, + })); + expect(parent.terminalHttpStatus).toBe(429); + held.release(); + expect(await within(response.text())).toContain("response.incomplete"); + await within(finalized.promise); + expect(observed).toHaveLength(1); + expect(observed[0]).toMatchObject({ + status: 429, + log: { + provider: "combo", model: "combo/free", requestedModel: "combo/free", + resolvedModel: "m1", comboId: "free", firstOutputMs: 17, + terminalHttpStatus: 429, terminalIncompleteReason: "usage_limit_reached", + upstreamError: "parent-observed quota", + }, + }); + expect(observed[0]!.log.routeDecision).toBe(parentTrace); + expect(observed[0]!.log.attempts).toBe(parentAttempts); + } finally { + held.release(); + } }); test("connect cancellation wins with 499, no backup, warning, or cooldown", async () => { diff --git a/tests/usage/request-log.test.ts b/tests/usage/request-log.test.ts index 1a3b4e32640..511780cde07 100644 --- a/tests/usage/request-log.test.ts +++ b/tests/usage/request-log.test.ts @@ -23,6 +23,8 @@ import { requestLogEntryFromPersistedUsage, sealRequestAttemptIdentity, recordAttemptCredentialSource, + inspectResponseLogSsePayload, + httpStatusForRequestLogTerminal, type RequestLogContext, } from "../../src/server/request-log"; import { handleResponses } from "../../src/server/responses"; @@ -58,6 +60,40 @@ function log(overrides: Partial): RequestLogEntry { } describe("request log metadata", () => { + test("incomplete quota evidence preserves an explicit HTTP 402 message", () => { + const log: RequestLogContext = { model: "gpt-test", provider: "openai" }; + inspectResponseLogSsePayload(log, JSON.stringify({ + type: "response.incomplete", + response: { incomplete_details: { message: "402" } }, + })); + expect(log.terminalHttpStatus).toBe(402); + expect(httpStatusForRequestLogTerminal("incomplete", log)).toBe(402); + }); + + test("normal structured incomplete reason wins over quota-like display text", () => { + const log: RequestLogContext = { model: "gpt-test", provider: "openai" }; + inspectResponseLogSsePayload(log, JSON.stringify({ + type: "response.incomplete", + response: { incomplete_details: { reason: "max_output_tokens", message: "Token usage limit reached" } }, + })); + expect(log.terminalHttpStatus).toBeUndefined(); + expect(log.terminalIncompleteReason).toBe("max_output_tokens"); + }); + + for (const error of [ + { type: "authentication_error", message: "Usage limit lookup requires renewed authentication" }, + { code: "invalid_api_key", message: "Usage limit unavailable for this credential" }, + ]) { + test(`structured auth failure wins over quota wording: ${JSON.stringify(error)}`, () => { + const failed: RequestLogContext = { model: "gpt-test", provider: "openai" }; + inspectResponseLogSsePayload(failed, JSON.stringify({ type: "response.failed", response: { error } })); + expect(failed.terminalHttpStatus).toBe(401); + const incomplete: RequestLogContext = { model: "gpt-test", provider: "openai" }; + inspectResponseLogSsePayload(incomplete, JSON.stringify({ type: "response.incomplete", response: { error } })); + expect(incomplete.terminalHttpStatus).toBeUndefined(); + }); + } + test("upstream credential attribution requires the resolved canonical xAI transport", () => { const attempt = beginRequestAttempt(1, "xai", "grok-test", "openai-chat"); const oauth = { adapter: "openai-chat", authMode: "oauth" as const, baseUrl: "https://cli-chat-proxy.grok.com/v1" };