From d17d323e6505c0f37bc4cee99f21dede1efae0ac Mon Sep 17 00:00:00 2001 From: t Date: Mon, 7 Sep 2026 01:44:57 +0900 Subject: [PATCH 1/2] fix(responses): classify incomplete quota terminals [skip ci] Carry the quota-attribution slice of #3769. Keep canonical compact 404 fallback deferred pending identity and history preservation. Local checks deferred to final hosted CI by maintainer instruction. Co-authored-by: Siddarth Reddy <221318067+ideabib@users.noreply.github.com> --- .../content/docs/reference/architecture.md | 10 + scripts/test-layout/layout.json | 1 + src/server/request-log.ts | 33 +- src/server/responses/core.ts | 48 +-- structure/04_transports-and-sidecars.md | 10 + tests/fixtures/test-layout-expected.json | 1 + ...responses-forward-incomplete-quota.test.ts | 337 ++++++++++++++++++ .../server/server-combo-failover-e2e.test.ts | 173 ++++++++- tests/usage/request-log.test.ts | 36 ++ 9 files changed, 622 insertions(+), 27 deletions(-) create mode 100644 tests/responses/responses-forward-incomplete-quota.test.ts diff --git a/docs-site/src/content/docs/reference/architecture.md b/docs-site/src/content/docs/reference/architecture.md index 96a3aec6dc..eb856811e1 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 0fbe7cf746..67093dcd07 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 a4c942bd88..eea04c684d 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 debba5c707..526010f31e 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 4716c447bb..5196684d21 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 db2583b00b..0823ba202c 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 0000000000..0d93e3d5e4 --- /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 14cdc0ceab..e4523aabb3 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 1a3b4e3264..511780cde0 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" }; From 17b1899a6acfaf63720237be702eb7489625837b Mon Sep 17 00:00:00 2001 From: t Date: Mon, 7 Sep 2026 01:45:20 +0900 Subject: [PATCH 2/2] fix(compact): preserve progress and accepted request lifetime [skip ci] Carry the progress portion of #3736 and adapt #3744 to release listener idle protection only after a complete valid request body. Preserve the 300-second default and existing encrypted output. Local validation deferred to the final hosted CI by maintainer instruction. Co-authored-by: mashfromband Co-authored-by: Hylouis233 <88263959+Hylouis233@users.noreply.github.com> --- .../content/docs/reference/architecture.md | 15 ++ scripts/test-layout/layout.json | 1 + src/adapters/openai-responses.ts | 14 ++ src/server/index.ts | 4 +- src/server/responses/compact.ts | 72 +++--- structure/04_transports-and-sidecars.md | 16 ++ tests/fixtures/test-layout-expected.json | 1 + tests/responses/compaction-progress.test.ts | 205 ++++++++++++++++++ .../responses-compaction-routing.test.ts | 53 ++++- tests/responses/responses-compaction.test.ts | 177 ++++++++++++++- tests/server/server-auth.test.ts | 37 +++- 11 files changed, 558 insertions(+), 37 deletions(-) create mode 100644 tests/responses/compaction-progress.test.ts diff --git a/docs-site/src/content/docs/reference/architecture.md b/docs-site/src/content/docs/reference/architecture.md index eb856811e1..b5fe8db9ac 100644 --- a/docs-site/src/content/docs/reference/architecture.md +++ b/docs-site/src/content/docs/reference/architecture.md @@ -210,3 +210,18 @@ 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. + +Remote compact requests can buffer their response for longer than the server's +request-idle timeout. That listener timeout is disabled after the request body is +accepted; client cancellation and upstream operation deadlines still apply. + +Buffered routed compaction treats nonempty text and reasoning deltas as progress +without exposing partial summary text. Comments, empty deltas and gateway +keepalives do not reset the adapter-event stall watchdog. The default stall +timeout stays 300 seconds; encrypted compaction content is preserved unchanged. + +Native compact response buffering also enforces a body-byte inactivity deadline +using `stallTimeoutSec` (300 seconds by default). Nonempty chunks reset that +deadline; a stalled body returns HTTP 504, client cancellation retains HTTP 499, +and cleanup does not wait for a stuck upstream cancellation promise. The 32 MiB +response ceiling and the original body bytes are preserved. diff --git a/scripts/test-layout/layout.json b/scripts/test-layout/layout.json index 67093dcd07..707518fc31 100644 --- a/scripts/test-layout/layout.json +++ b/scripts/test-layout/layout.json @@ -492,6 +492,7 @@ "command-code-quota.test.ts": "providers", "command-code-workspace-cache.test.ts": "providers", "commandcode-provider.test.ts": "providers", + "compaction-progress.test.ts": "responses", "compatibility-manifest.test.ts": "codex-integration", "compatibility-provider-equivalence.test.ts": "routing", "compatibility-version.test.ts": "ci-workflows", diff --git a/src/adapters/openai-responses.ts b/src/adapters/openai-responses.ts index 1faa9c0cbb..07c0556d5e 100644 --- a/src/adapters/openai-responses.ts +++ b/src/adapters/openai-responses.ts @@ -2546,6 +2546,7 @@ export function createResponsesPassthroughAdapter(provider: OcxProviderConfig): let snapshot = ""; let usage: OcxUsage | undefined; let compactionEncryptedContent: string | undefined; + let completedSeen = false; for await (const event of decodeServerSentEvents(response.body, { translatorBudget: budget })) { let payload: unknown; try { payload = JSON.parse(event.data); } catch { continue; } @@ -2580,6 +2581,7 @@ export function createResponsesPassthroughAdapter(provider: OcxProviderConfig): return; case "response.completed": { + completedSeen = true; const responsePayload = isPlainObject(payload.response) ? payload.response : undefined; const output = Array.isArray(responsePayload?.output) ? responsePayload.output : []; const compaction = output.find(item => isPlainObject(item) && item.type === "compaction"); @@ -2620,6 +2622,18 @@ export function createResponsesPassthroughAdapter(provider: OcxProviderConfig): } break; } + // Buffered text is still upstream progress, but gateway keepalives are not. + // Yield after accounting, directly to the consumer: no progress queue or content leak. + if ( + !completedSeen + && (payload.type === "response.output_text.delta" + || payload.type === "response.reasoning_summary_text.delta" + || payload.type === "response.reasoning_text.delta") + && typeof payload.delta === "string" + && payload.delta.length > 0 + ) { + yield { type: "heartbeat" }; + } } // Gateways differ in which of these they emit; prefer the authoritative // completed snapshot so text is never double-counted. diff --git a/src/server/index.ts b/src/server/index.ts index 0391ff031d..8d46c30ab6 100644 --- a/src/server/index.ts +++ b/src/server/index.ts @@ -1765,7 +1765,9 @@ export function startServer(port?: number, deps: StartServerDeps = {}): Server { let response: Response; try { - response = await handleResponsesCompact(req, config, logCtx, turnAdmissionLease, admission); + response = await handleResponsesCompact(req, config, logCtx, turnAdmissionLease, admission, { + onRequestBodyRead: () => disableResponsesRequestTimeout(req, requestServer), + }); } catch { response = formatErrorResponse(500, "server_error", "Unexpected compact request failure"); } diff --git a/src/server/responses/compact.ts b/src/server/responses/compact.ts index a742fad98d..c914dcb4b2 100644 --- a/src/server/responses/compact.ts +++ b/src/server/responses/compact.ts @@ -112,7 +112,8 @@ import type { WsData } from "../ws-bridge"; import { codexAccountSelectionForTurn, registerTurn, trackStreamLifetime, unregisterTurn } from "../lifecycle"; import type { AdmissionLease } from "../../lib/admission"; import { redactSecretString } from "../../lib/redact"; -import { readBoundedResponseBody } from "../../lib/bounded-body"; +import { readBoundedResponseBytes } from "../../lib/bounded-body"; +import { resolveStallTimeoutSec } from "../../stall-timeout"; import { isRateLimitOrQuotaFailureMessage } from "../../lib/errors"; import { supportedLadderFor } from "../effort-policy"; import { @@ -212,6 +213,8 @@ function compactHandoffRoute(req: Request, previousModel: string, now = Date.now export interface HandleResponsesCompactOptions { nativeMainRefreshDependencies?: NativeMainRefreshDependencies; + /** Release the listener's idle guard only after the complete request body is accepted. */ + onRequestBodyRead?: () => void; } export function compactResponseTooLargeError(): Response { @@ -464,43 +467,45 @@ function compactResponseHeaders(upstream: Response): Headers { return headers; } -export async function bufferCompactResponse(upstream: Response, signal: AbortSignal): Promise { - const reader = upstream.body?.getReader(); +export async function bufferCompactResponse( + upstream: Response, + signal: AbortSignal, + stallTimeoutSec?: number, +): Promise { 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); - return compactResponseTooLargeError(); - } - const chunks: Uint8Array[] = []; - let total = 0; try { - while (true) { - if (signal.aborted) { - await reader.cancel(signal.reason).catch(() => undefined); - return formatErrorResponse(499, "client_cancelled", "Client cancelled compact request"); - } - const { done, value } = await reader.read(); - if (done) break; - total += value.byteLength; - if (total > COMPACT_RESPONSE_MAX_BYTES) { - await reader.cancel("compact_response_too_large").catch(() => undefined); - return compactResponseTooLargeError(); - } - chunks.push(value); + if (signal.aborted) { + // No reader is attached yet. Cancellation must not wait for a broken source's cleanup. + void upstream.body?.cancel(signal.reason).catch(() => undefined); + return formatErrorResponse(499, "client_cancelled", "Client cancelled compact request"); } - } catch { + if (!upstream.body) 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) { + void upstream.body.cancel("compact_response_too_large").catch(() => undefined); + return compactResponseTooLargeError(); + } + // Header admission has finished; only non-empty body chunks re-arm this deadline. + // The raw reader preserves bytes and cancels/releases without awaiting source cleanup. + const result = await readBoundedResponseBytes(upstream, { + signal, + maxBytes: COMPACT_RESPONSE_MAX_BYTES, + inactivityTimeoutMs: resolveStallTimeoutSec(stallTimeoutSec) * 1_000, + }); + if (signal.aborted) return formatErrorResponse(499, "client_cancelled", "Client cancelled compact request"); + if (result.oversized) return compactResponseTooLargeError(); + return new Response(result.bytes, { status: upstream.status, statusText: upstream.statusText, headers }); + } catch (error) { if (signal.aborted) return formatErrorResponse(499, "client_cancelled", "Client cancelled compact request"); + if (error instanceof DOMException && error.name === "TimeoutError") { + return Response.json({ error: { + message: "Compact response body stalled", + type: "upstream_stall_timeout", + code: "upstream_stall_timeout", + } }, { status: 504 }); + } return formatErrorResponse(502, "upstream_error", "Failed to read compact response"); } - const body = new Uint8Array(total); - let offset = 0; - for (const chunk of chunks) { - body.set(chunk, offset); - offset += chunk.byteLength; - } - return new Response(body, { status: upstream.status, statusText: upstream.statusText, headers }); } @@ -526,6 +531,7 @@ export async function handleResponsesCompact( if (typeof raw.model !== "string" || raw.model.length === 0) { return formatErrorResponse(400, "invalid_request_error", "compaction request requires a model"); } + options.onRequestBodyRead?.(); // Correct the IDENTITY before routing, or the synthetic id does not route at all. Held in // a local rather than written back to `raw.model`: assigning to the property widens it out // of the `string` narrowing the guard above just established. @@ -1037,7 +1043,7 @@ export async function handleResponsesCompact( upstream.headers.get("x-codex-secondary-reset-at"), upstream.headers.get("x-codex-tertiary-reset-at"), ].filter(Boolean); - const buffered = await bufferCompactResponse(upstream, req.signal); + const buffered = await bufferCompactResponse(upstream, req.signal, config.stallTimeoutSec); const bufferedErrorText = buffered.ok ? "" : await buffered.clone().text().catch(() => ""); diff --git a/structure/04_transports-and-sidecars.md b/structure/04_transports-and-sidecars.md index 5196684d21..a20eda8ca6 100644 --- a/structure/04_transports-and-sidecars.md +++ b/structure/04_transports-and-sidecars.md @@ -1655,3 +1655,19 @@ 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. + +Remote compact requests release the server request-idle timeout only after a complete +JSON object with a valid model has been read. Partial or invalid uploads retain +the listener guard; admitted compaction then uses the upstream operation's own +deadlines and client cancellation. + +Buffered routed compaction treats nonempty text and reasoning deltas as progress +without exposing partial summary text. Comments, empty deltas and gateway +keepalives do not reset the adapter-event stall watchdog. The default stall +timeout stays 300 seconds; encrypted compaction content is preserved unchanged. + +Native compact response buffering also enforces a body-byte inactivity deadline +using `stallTimeoutSec` (300 seconds by default). Nonempty chunks reset that +deadline; a stalled body returns HTTP 504, client cancellation retains HTTP 499, +and cleanup does not wait for a stuck upstream cancellation promise. The 32 MiB +response ceiling and the original body bytes are preserved. diff --git a/tests/fixtures/test-layout-expected.json b/tests/fixtures/test-layout-expected.json index 0823ba202c..f9df7f508f 100644 --- a/tests/fixtures/test-layout-expected.json +++ b/tests/fixtures/test-layout-expected.json @@ -327,6 +327,7 @@ "command-code-quota.test.ts": "providers", "command-code-workspace-cache.test.ts": "providers", "commandcode-provider.test.ts": "providers", + "compaction-progress.test.ts": "responses", "compatibility-manifest.test.ts": "codex-integration", "compatibility-provider-equivalence.test.ts": "routing", "compatibility-version.test.ts": "ci-workflows", diff --git a/tests/responses/compaction-progress.test.ts b/tests/responses/compaction-progress.test.ts new file mode 100644 index 0000000000..3e1a1556ea --- /dev/null +++ b/tests/responses/compaction-progress.test.ts @@ -0,0 +1,205 @@ +import { describe, expect, test } from "bun:test"; +import { createResponsesPassthroughAdapter } from "../../src/adapters/openai-responses"; +import { bridgeToResponsesSSE, buildResponseJSON } from "../../src/bridge"; +import type { AdapterEvent } from "../../src/types"; +import { createTestTranslatorBudget } from "../helpers/translator-budget"; + +const encoder = new TextEncoder(); +const provider = { adapter: "openai-responses", baseUrl: "https://gateway.example/v1", authMode: "key" as const }; +const frame = (payload: unknown) => `data: ${JSON.stringify(payload)}\n\n`; +const completed = { + type: "response.completed", + response: { + id: "resp_compaction", + status: "completed", + output: [{ type: "message", role: "assistant", content: [{ type: "output_text", text: "Final summary" }] }], + }, +}; + +function upstream() { + let controller!: ReadableStreamDefaultController; + let nextRead = Promise.withResolvers(); + let ended = false; + let pulls = 0; + let cancelled = false; + const body = new ReadableStream({ + start(value) { controller = value; }, + pull() { pulls++; nextRead.resolve(); }, + cancel() { ended = true; cancelled = true; }, + }, { highWaterMark: 0 }); + return { + body, + get pulls() { return pulls; }, + get cancelled() { return cancelled; }, + waitingForRead: () => nextRead.promise, + send(text: string) { + nextRead = Promise.withResolvers(); + controller.enqueue(encoder.encode(text)); + }, + close() { if (!ended) { ended = true; controller.close(); } }, + }; +} + +function bridged() { + const source = upstream(); + const budget = createTestTranslatorBudget(); + let beat = () => {}; + let cleanupCalls = 0; + const stream = bridgeToResponsesSSE( + createResponsesPassthroughAdapter(provider).parseStream(new Response(source.body), budget), + "example-model", undefined, undefined, undefined, + () => { cleanupCalls++; source.close(); }, 500, + { + translatorBudget: budget, compaction: true, stallTimeoutSec: 1, + timers: { + setInterval(callback) { beat = callback; return 1; }, + clearInterval() { beat = () => {}; }, + }, + }, + ); + const text = new Response(stream).text(); + return { + source, text, + get cleanupCalls() { return cleanupCalls; }, + tick: () => beat(), + async send(text: string) { + await source.waitingForRead(); + source.send(text); + // The next upstream read occurs after the bridge consumes any adapter heartbeat. + await source.waitingForRead(); + }, + }; +} + +describe("buffered Responses compaction progress", () => { + // Codex oracle: openai/codex d2d5b702, codex-api/src/sse/responses.rs:367-408. + // Indices make these canonical reasoning fixtures; progress itself carries no content. + for (const delta of [ + { type: "response.output_text.delta", delta: "Buffered progress" }, + { type: "response.reasoning_summary_text.delta", delta: "Buffered progress", summary_index: 0 }, + { type: "response.reasoning_text.delta", delta: "Buffered progress", content_index: 0 }, + ]) { + test(`${delta.type} prevents stall before terminal without exposing partial content`, async () => { + const h = bridged(); + try { + for (let i = 0; i < 6; i++) { + await h.send(frame(delta)); + h.tick(); + expect(h.cleanupCalls).toBe(0); + } + await h.send(frame(completed)); + h.source.close(); + const wire = await h.text; + expect(wire.match(/event: response.completed\n/g)).toHaveLength(1); + expect(wire.match(/event: response.output_item.done\n/g)).toHaveLength(1); + expect(wire).toContain('"type":"compaction"'); + expect(wire).not.toContain("Buffered progress"); + expect(wire).not.toContain("event: response.output_text.delta"); + expect(wire).not.toContain("upstream_stall_timeout"); + // The bridge invokes its upstream cleanup callback on normal terminal events too. + expect(h.cleanupCalls).toBe(1); + } finally { h.source.close(); await h.text; } + }); + } + + test("comments, typed keepalives and empty or malformed deltas do not reset stall", async () => { + const h = bridged(); + try { + const noise = ": keep-alive\n\ndata: invalid-json\n\n" + + frame({ type: "response.heartbeat" }) + + frame({ type: "response.output_text.delta", delta: "" }) + + frame({ type: "response.reasoning_summary_text.delta", delta: null }) + + frame({ type: "response.reasoning_text.delta", delta: 42 }) + + frame({ type: "response.unknown.delta", delta: "not recognized progress" }); + await h.send(noise); + h.tick(); + await h.send(noise); + h.tick(); + const wire = await h.text; + expect(wire).toContain("upstream_stall_timeout"); + expect(wire).not.toContain("event: response.completed"); + expect(wire).not.toContain('"type":"compaction"'); + expect(h.cleanupCalls).toBe(1); + } finally { h.source.close(); await h.text; } + }); + + test("progress preserves snapshot precedence, usage and native ciphertext", async () => { + const budget = createTestTranslatorBudget(); + const ciphertext = "gAAAAABm-native-compaction-ciphertext"; + const usage = { input_tokens: 12, output_tokens: 4, total_tokens: 16, gateway_metadata: { cached: true } }; + const terminal = { + ...completed, + response: { ...completed.response, usage, output: [ + ...completed.response.output, { type: "compaction", encrypted_content: ciphertext }, + ] }, + }; + const input = frame({ type: "response.output_text.delta", delta: "Partial text" }) + + frame({ type: "response.output_text.done", text: "Done text" }) + + frame(terminal) + + frame({ type: "response.output_text.delta", delta: "Late text" }); + const events: AdapterEvent[] = []; + for await (const event of createResponsesPassthroughAdapter(provider).parseStream(new Response(input), budget)) { + events.push(event); + } + expect(events).toEqual([ + { type: "heartbeat" }, + { type: "text_delta", text: "Final summary" }, + { type: "done", usage: { inputTokens: 12, outputTokens: 4, totalTokens: 16, rawUsage: usage }, compactionEncryptedContent: ciphertext }, + ]); + const result = buildResponseJSON(events, "example-model", { compaction: true, translatorBudget: budget }); + expect(result.output).toEqual([expect.objectContaining({ type: "compaction", encrypted_content: ciphertext })]); + }); + + test("reasoning progress with ciphertext-only completion does not manufacture summary text", async () => { + const budget = createTestTranslatorBudget(); + const ciphertext = "gAAAAABm-ciphertext-only"; + const input = frame({ type: "response.reasoning_text.delta", content_index: 0, delta: "Hidden reasoning" }) + + frame({ ...completed, response: { + ...completed.response, output: [{ type: "compaction", encrypted_content: ciphertext }], + } }); + const events: AdapterEvent[] = []; + for await (const event of createResponsesPassthroughAdapter(provider).parseStream(new Response(input), budget)) { + events.push(event); + } + expect(events).toEqual([{ type: "heartbeat" }, { type: "done", compactionEncryptedContent: ciphertext }]); + expect(budget.snapshot().currentBytes).toBe(encoder.encode(ciphertext).byteLength); + const result = buildResponseJSON(events, "example-model", { compaction: true, translatorBudget: budget }); + expect(result.output).toEqual([expect.objectContaining({ type: "compaction", encrypted_content: ciphertext })]); + }); + + test("a suspended heartbeat does not read ahead and return cancels the reader", async () => { + const source = upstream(); + const budget = createTestTranslatorBudget(); + const iterator = createResponsesPassthroughAdapter(provider).parseStream(new Response(source.body), budget); + try { + source.send(frame({ type: "response.reasoning_text.delta", content_index: 0, delta: "Hidden reasoning" }).repeat(64)); + expect(await iterator.next()).toEqual({ done: false, value: { type: "heartbeat" } }); + expect(source.pulls).toBe(0); // Only the already-enqueued chunk was consumed (HWM 0). + for (let i = 1; i < 64; i++) { + expect(await iterator.next()).toEqual({ done: false, value: { type: "heartbeat" } }); + expect(source.pulls).toBe(0); + } + await iterator.return(undefined); + expect(source.cancelled).toBe(true); + expect(budget.snapshot().currentBytes).toBe(0); + } finally { source.close(); await iterator.return(undefined); } + }); + + for (const type of ["response.failed", "response.incomplete"]) { + test(`${type} after progress never flushes a successful summary`, async () => { + const budget = createTestTranslatorBudget(); + const events: AdapterEvent[] = []; + const input = frame({ type: "response.output_text.delta", delta: "Unfinished summary" }) + + frame({ type, response: type === "response.failed" + ? { error: { message: "stopped" } } + : { incomplete_details: { reason: "stopped" } } }); + for await (const event of createResponsesPassthroughAdapter(provider).parseStream(new Response(input), budget)) { + events.push(event); + } + expect(events).toEqual([ + { type: "heartbeat" }, + type === "response.failed" ? { type: "error", message: "stopped" } : { type: "incomplete", reason: "stopped" }, + ]); + }); + } +}); diff --git a/tests/responses/responses-compaction-routing.test.ts b/tests/responses/responses-compaction-routing.test.ts index 51f76ab12c..8e92f28715 100644 --- a/tests/responses/responses-compaction-routing.test.ts +++ b/tests/responses/responses-compaction-routing.test.ts @@ -4,7 +4,7 @@ * contract; every other gateway has to be driven as a plain summarizer, or Codex * fatals on a compaction turn that came back as an ordinary message. */ -import { afterEach, describe, expect, spyOn, test } from "bun:test"; +import { afterEach, describe, expect, jest, spyOn, test } from "bun:test"; import { mkdtempSync, writeFileSync } from "node:fs"; import { tmpdir } from "node:os"; import { join } from "node:path"; @@ -948,6 +948,57 @@ describe("compact alternate-account attempt (#913)", () => { }); } + test("native compact headers followed by a stalled body return 504 without retry and release account cleanup", async () => { + await withPoolEnv("ocx-compact-body-deadline-", async config => { + config.stallTimeoutSec = 2; + const readStarted = Promise.withResolvers(); + let sends = 0; + let cancelled = 0; + let acceptedBody = false; + const body = new ReadableStream({ + pull() { readStarted.resolve(); }, + cancel() { cancelled++; return new Promise(() => {}); }, + }, { highWaterMark: 0 }); + globalThis.fetch = (async () => { + sends++; + return new Response(body, { headers: { "content-type": "application/json" } }); + }) as typeof fetch; + const releaseSpy = spyOn(authContextModule, "releaseCodexAuthContextProbeLease"); + const client = new AbortController(); + // Same scoped, non-concurrent Bun timer control as responses/ws-upstream.test.ts. + jest.useFakeTimers(); + const pending = handleResponsesCompact( + compactionRequest({ model: "gpt-5.5", input: [ + { type: "message", role: "user", content: [{ type: "input_text", text: "earlier turn" }] }, + ] }, client.signal), config, { model: "", provider: "" }, + undefined, undefined, { onRequestBodyRead: () => { acceptedBody = true; } }, + ); + try { + await Promise.race([ + readStarted.promise, + pending.then(response => { throw new Error(`compact returned ${response.status} before reading its body`); }), + ]); + expect(acceptedBody).toBe(true); + expect(sends).toBe(1); + jest.advanceTimersByTime(2_000); + const response = await pending; + expect(response.status).toBe(504); + expect(await response.json()).toMatchObject({ error: { code: "upstream_stall_timeout" } }); + expect(sends).toBe(1); + expect(cancelled).toBe(1); + expect(body.locked).toBe(false); + expect(releaseSpy).toHaveBeenCalledWith(expect.objectContaining({ kind: "pool", accountId: "pool-a" })); + } finally { + client.abort(); + try { await pending; } finally { + jest.clearAllTimers(); + jest.useRealTimers(); + releaseSpy.mockRestore(); + } + } + }); + }); + test("canonical trailing slashes are pinned before native compact sends pool credentials", async () => { await withPoolEnv("ocx-compact-canonical-url-", async config => { config.providers.openai!.baseUrl = "https://chatgpt.com/backend-api/codex///"; diff --git a/tests/responses/responses-compaction.test.ts b/tests/responses/responses-compaction.test.ts index 631ff48e2b..edf6fec1bb 100644 --- a/tests/responses/responses-compaction.test.ts +++ b/tests/responses/responses-compaction.test.ts @@ -1,4 +1,4 @@ -import { describe, expect, test } from "bun:test"; +import { describe, expect, jest, test } from "bun:test"; import { bridgeToResponsesSSE, buildResponseJSON } from "../../src/bridge"; import { createResponsesPassthroughAdapter as createResponsesPassthroughAdapterProduction } from "../../src/adapters/openai-responses"; import { createTranslatorBudget } from "../../src/lib/translator-budget"; @@ -15,10 +15,185 @@ import { } from "../../src/responses/compaction"; import type { AdapterEvent } from "../../src/types"; import { withTestTranslatorBudget } from "../helpers/translator-budget"; +import { bufferCompactResponse, COMPACT_RESPONSE_MAX_BYTES } from "../../src/server/responses/compact"; const createResponsesPassthroughAdapter = (...args: Parameters) => withTestTranslatorBudget(createResponsesPassthroughAdapterProduction(...args)); +// These non-concurrent tests scope Bun's fake timers like responses/ws-upstream.test.ts. +// The real bounded-body reader and idleDeadline run; upstream pull acknowledgements +// synchronize chunk consumption before advancing time, without sleeps or mocking either helper. +async function withCompactBodyClock(run: () => Promise): Promise { + jest.useFakeTimers(); + try { + await run(); + expect(jest.getTimerCount()).toBe(0); + } finally { + jest.clearAllTimers(); + jest.useRealTimers(); + } +} + +function compactBodySource(onCancel?: () => void) { + let controller!: ReadableStreamDefaultController; + let nextRead = Promise.withResolvers(); + const cancellationReasons: unknown[] = []; + let ended = false; + const body = new ReadableStream({ + start(value) { controller = value; }, + pull() { nextRead.resolve(); }, + cancel(reason) { + ended = true; + cancellationReasons.push(reason); + onCancel?.(); + // A deadline must return even if the upstream's cancellation cleanup never finishes. + return new Promise(() => {}); + }, + }, { highWaterMark: 0 }); + return { + body, cancellationReasons, + waitingForRead: () => nextRead.promise, + async send(bytes: Uint8Array) { + await nextRead.promise; + nextRead = Promise.withResolvers(); + controller.enqueue(bytes); + await nextRead.promise; + }, + close() { if (!ended) { ended = true; controller.close(); } }, + }; +} + +describe("native compact response body deadline", () => { + test("headers followed by silence expire at the default 300 seconds without waiting for cancel", () => withCompactBodyClock(async () => { + const source = compactBodySource(); + const pending = bufferCompactResponse(new Response(source.body), new AbortController().signal); + try { + await source.waitingForRead(); + jest.advanceTimersByTime(299_999); + expect(jest.getTimerCount()).toBe(1); + expect(source.cancellationReasons).toHaveLength(0); + jest.advanceTimersByTime(1); + const response = await pending; + expect(response.status).toBe(504); + expect(await response.json()).toMatchObject({ error: { type: "upstream_stall_timeout", code: "upstream_stall_timeout" } }); + expect(source.cancellationReasons).toHaveLength(1); + expect(source.cancellationReasons[0]).toBeInstanceOf(DOMException); + expect((source.cancellationReasons[0] as DOMException).name).toBe("TimeoutError"); + expect(source.body.locked).toBe(false); + } finally { source.close(); await pending; } + })); + + test("nonempty chunks rearm the deadline and success preserves exact bytes and header hints", () => withCompactBodyClock(async () => { + const source = compactBodySource(); + const expected = new Uint8Array([0, 255, 128, 195, 40]); + const pending = bufferCompactResponse(new Response(source.body, { + status: 201, statusText: "Compact ready", + headers: { + "content-type": "application/octet-stream", "content-length": "999", + "retry-after": "42", "x-codex-primary-reset-at": "1900000000", + "x-codex-secondary-reset-at": "1900000001", "x-codex-tertiary-reset-at": "1900000002", + location: "/compact-result", "set-cookie": "ignored=1", "transfer-encoding": "chunked", + }, + }), new AbortController().signal, 2); + try { + await source.waitingForRead(); + for (let i = 0; i < expected.length; i++) { + jest.advanceTimersByTime(1_500); + await source.send(expected.subarray(i, i + 1)); + } + source.close(); + const response = await pending; + expect(response.status).toBe(201); + expect(response.statusText).toBe("Compact ready"); + expect(new Uint8Array(await response.arrayBuffer())).toEqual(expected); + expect(Object.fromEntries(response.headers)).toEqual({ + "content-type": "application/octet-stream", "retry-after": "42", + "x-codex-primary-reset-at": "1900000000", "x-codex-secondary-reset-at": "1900000001", + "x-codex-tertiary-reset-at": "1900000002", location: "/compact-result", + }); + expect(source.cancellationReasons).toHaveLength(0); + expect(source.body.locked).toBe(false); + } finally { source.close(); await pending; } + })); + + test("empty chunks do not rearm the byte inactivity deadline", () => withCompactBodyClock(async () => { + const source = compactBodySource(); + const pending = bufferCompactResponse(new Response(source.body), new AbortController().signal, 2); + try { + await source.waitingForRead(); + jest.advanceTimersByTime(1_000); + await source.send(new Uint8Array(0)); + jest.advanceTimersByTime(999); + expect(source.cancellationReasons).toHaveLength(0); + jest.advanceTimersByTime(1); + expect((await pending).status).toBe(504); + expect(source.cancellationReasons).toHaveLength(1); + expect(source.body.locked).toBe(false); + } finally { source.close(); await pending; } + })); + + for (const idleAlsoFires of [false, true]) { + test(`client cancellation unblocks a pending read and wins over idle expiry (${idleAlsoFires})`, () => withCompactBodyClock(async () => { + const client = new AbortController(); + // Abort during the timeout's source-cleanup callback, before the wrapper + // classifies its result. Advancing fake time can already flush promises. + const source = compactBodySource(idleAlsoFires ? () => client.abort(new Error("client stopped")) : undefined); + const pending = bufferCompactResponse(new Response(source.body), client.signal, 2); + try { + await source.waitingForRead(); + if (idleAlsoFires) jest.advanceTimersByTime(2_000); + else client.abort(new Error("client stopped")); + const response = await pending; + expect(client.signal.aborted).toBe(true); + expect(response.status).toBe(499); + expect(await response.json()).toMatchObject({ error: { code: "client_cancelled" } }); + expect(source.cancellationReasons).toHaveLength(1); + expect(source.body.locked).toBe(false); + } finally { source.close(); await pending; } + })); + } + + test("cancellation after a completed timeout does not retroactively replace its 504", () => withCompactBodyClock(async () => { + const source = compactBodySource(); + const client = new AbortController(); + const pending = bufferCompactResponse(new Response(source.body), client.signal, 2); + try { + await source.waitingForRead(); + jest.advanceTimersByTime(2_000); + const response = await pending; + expect(response.status).toBe(504); + client.abort(new Error("late cancellation")); + expect(response.status).toBe(504); + expect(source.cancellationReasons).toHaveLength(1); + } finally { source.close(); await pending; } + })); + + test("declared and observed oversize bodies retain the 32 MiB limit without waiting for cancel", () => withCompactBodyClock(async () => { + for (const declared of [true, false]) { + let cancelled = 0; + const body = new ReadableStream({ + pull(controller) { controller.enqueue(new Uint8Array(COMPACT_RESPONSE_MAX_BYTES + 1)); }, + cancel() { cancelled++; return new Promise(() => {}); }, + }, { highWaterMark: 0 }); + const response = await bufferCompactResponse(new Response(body, { + headers: declared ? { "content-length": String(COMPACT_RESPONSE_MAX_BYTES + 1) } : {}, + }), new AbortController().signal, 2); + expect(response.status).toBe(502); + expect(await response.json()).toMatchObject({ error: { code: "compact_response_too_large" } }); + expect(cancelled).toBe(1); + expect(body.locked).toBe(false); + } + const atLimit = new Uint8Array(COMPACT_RESPONSE_MAX_BYTES); + atLimit[atLimit.length - 1] = 255; + const response = await bufferCompactResponse(new Response(atLimit), new AbortController().signal, 2); + expect(response.status).toBe(200); + const bytes = new Uint8Array(await response.arrayBuffer()); + expect(bytes.byteLength).toBe(COMPACT_RESPONSE_MAX_BYTES); + expect(bytes[0]).toBe(0); + expect(bytes[bytes.length - 1]).toBe(255); + })); +}); + async function* replay(events: AdapterEvent[]): AsyncGenerator { for (const event of events) yield event; } diff --git a/tests/server/server-auth.test.ts b/tests/server/server-auth.test.ts index 11fddb9772..72a1297f37 100644 --- a/tests/server/server-auth.test.ts +++ b/tests/server/server-auth.test.ts @@ -38,7 +38,7 @@ import { import { clearRequestLogsForTests, getRequestLogEntries } from "../../src/server/request-log"; import { readUsageEntries } from "../../src/usage/log"; import { handleManagementAPI } from "../../src/server/management-api"; -import { handleResponses } from "../../src/server/responses"; +import { handleResponses, handleResponsesCompact } from "../../src/server/responses"; import type { OcxConfig } from "../../src/types"; import { fakeChatGptJwt } from "../helpers/fake-chatgpt-jwt"; import { installIsolatedCodexHome, type IsolatedCodexHome } from "../helpers/isolated-codex-home"; @@ -602,6 +602,41 @@ describe("server local API auth", () => { })).toBe(false); }); + test("compact keeps the idle guard until a valid request body is complete", async () => { + let bodyController!: ReadableStreamDefaultController; + const body = new ReadableStream({ start(controller) { bodyController = controller; } }); + const cfg = config(); + cfg.defaultProvider = "fixture"; + cfg.providers = { fixture: { ...cfg.providers.openai!, disabled: true } }; + const request = new Request("http://localhost/v1/responses/compact", { + method: "POST", headers: { "content-type": "application/json" }, body, + }); + let accepted = 0; + const result = handleResponsesCompact(request, cfg, { model: "unknown", provider: "unknown" }, undefined, undefined, { + onRequestBodyRead: () => { accepted++; }, + }); + bodyController.enqueue(new TextEncoder().encode('{"model":"fixture/gpt-test","input":[')); + expect(accepted).toBe(0); + bodyController.enqueue(new TextEncoder().encode(']}')); + bodyController.close(); + expect((await result).status).toBe(404); + expect(accepted).toBe(1); + }); + + for (const body of ["{", "[]", "{}", '{"model":0}', '{"model":""}']) { + test(`compact does not release idle protection for rejected body ${body}`, async () => { + let accepted = false; + const request = new Request("http://localhost/v1/responses/compact", { + method: "POST", headers: { "content-type": "application/json" }, body, + }); + const response = await handleResponsesCompact(request, config(), { model: "unknown", provider: "unknown" }, undefined, undefined, { + onRequestBodyRead: () => { accepted = true; }, + }); + expect(response.status).toBe(400); + expect(accepted).toBe(false); + }); + } + test("responses handler keeps the request timeout until the body is fully accepted", async () => { let controller!: ReadableStreamDefaultController; const body = new ReadableStream({