From 07cbc25c653303293b76701caf3399b48297447b Mon Sep 17 00:00:00 2001 From: HoshimiRox1 <166687527+HoshimiRox1@users.noreply.github.com> Date: Fri, 7 Aug 2026 00:08:16 +0800 Subject: [PATCH 1/2] fix(codex): refuse responses input beyond the advertised context window A chained-turn replay can balloon a request far past the model's context window (observed: a 4x expansion pushed a ~400k-token conversation to 1.6M input tokens). The proxy forwarded it verbatim; processing it on Windows ballooned bun RSS and native-crashed the whole service (upstream Bun memory bug, #314), taking every active thread down until restart. Reject the request with a clean 413 before any upstream I/O when the parsed input exceeds the model's configured modelContextWindows value. The client compacts well before the window, so the guard only fires on abnormal duplication. --- src/server/responses/core.ts | 34 +++++++++ tests/responses-input-guard.test.ts | 110 ++++++++++++++++++++++++++++ 2 files changed, 144 insertions(+) create mode 100644 tests/responses-input-guard.test.ts diff --git a/src/server/responses/core.ts b/src/server/responses/core.ts index 539617208..99f17357b 100644 --- a/src/server/responses/core.ts +++ b/src/server/responses/core.ts @@ -1537,6 +1537,40 @@ async function handleResponsesInner( ); } + // Input-size guard: refuse to forward an input that exceeds the model's advertised context + // window. The client compacts well before this limit, so an oversized body means abnormal + // duplication (observed: a 4x replay expansion pushed a ~400k-token conversation to 1.6M). + // Forwarding it on Windows balloons bun RSS and can native-crash the whole proxy (upstream + // Bun memory bug, issue #314), taking every active thread down at once. Fail one request + // cleanly instead. Token estimate = UTF-8 bytes / 4 (Chinese ~3B/token, ASCII/code ~4B/token): + // the estimate undercounts near-limit traffic and over-counts nothing, so it only fires on + // genuine blowups. # ponytail: bytes/4 heuristic, replace with a real tokenizer if a + // 2x-class duplication ever needs catching at the margin. + const advertisedWindow = route.provider.modelContextWindows?.[route.modelId]; + if (typeof advertisedWindow === "number" && advertisedWindow > 0) { + let inputBytes = 0; + for (const msg of parsed.context.messages) { + const content = msg.content; + if (typeof content === "string") { + inputBytes += Buffer.byteLength(content); + } else if (Array.isArray(content)) { + for (const part of content) { + if (part && typeof part === "object" && typeof (part as { text?: unknown }).text === "string") { + inputBytes += Buffer.byteLength((part as { text: string }).text); + } + } + } + } + if (inputBytes / 4 > advertisedWindow) { + return formatErrorResponse( + 413, + "request_too_large", + `input (≈${Math.round(inputBytes / 4)} tokens) exceeds ${route.modelId} context window (${advertisedWindow} tokens); refusing to forward`, + { code: "input_context_window_exceeded" }, + ); + } + } + // Captured before normalization: whether the CLIENT asked for SSE. The // transport-neutral upstream-streaming policy below may force a bounded JSON // upstream for reliability (#875); the answer must then be reframed to SSE diff --git a/tests/responses-input-guard.test.ts b/tests/responses-input-guard.test.ts new file mode 100644 index 000000000..ce4f32d02 --- /dev/null +++ b/tests/responses-input-guard.test.ts @@ -0,0 +1,110 @@ +/** + * Regression coverage for the responses input-size guard: a request whose input + * exceeds the model's advertised context window must be rejected with a clean 413 + * instead of being forwarded (forwarding a ~1.6M-token duplication on Windows + * ballooned bun RSS and native-crashed the whole proxy, issue #314). + */ +import { afterEach, beforeEach, describe, expect, setDefaultTimeout, test } from "bun:test"; +import { mkdtempSync, rmSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { handleResponses } from "../src/server/responses"; +import type { OcxConfig } from "../src/types"; +import type { RequestLogContext } from "../src/server/request-log"; + +setDefaultTimeout(30_000); + +const originalFetch = globalThis.fetch; +let testDir: string; +let previousOpencodexHome: string | undefined; +let previousCodexHome: string | undefined; + +beforeEach(() => { + testDir = mkdtempSync(join(tmpdir(), "ocx-input-guard-")); + previousOpencodexHome = process.env.OPENCODEX_HOME; + previousCodexHome = process.env.CODEX_HOME; + process.env.OPENCODEX_HOME = testDir; + process.env.CODEX_HOME = testDir; +}); + +afterEach(() => { + globalThis.fetch = originalFetch; + rmSync(testDir, { recursive: true, force: true }); + if (previousOpencodexHome === undefined) delete process.env.OPENCODEX_HOME; + else process.env.OPENCODEX_HOME = previousOpencodexHome; + if (previousCodexHome === undefined) delete process.env.CODEX_HOME; + else process.env.CODEX_HOME = previousCodexHome; +}); + +function deepseekConfig(): OcxConfig { + return { + port: 0, + defaultProvider: "deepseek", + providers: { + deepseek: { + adapter: "openai-responses", + baseUrl: "https://api.deepseek.com", + responsesPath: "/responses", + authMode: "key", + apiKey: "sk-test", + models: ["deepseek-v4-flash"], + modelContextWindows: { "deepseek-v4-flash": 1_000_000 }, + }, + }, + } as OcxConfig; +} + +async function postResponses(config: OcxConfig, body: Record): Promise { + return handleResponses( + new Request("http://localhost/v1/responses", { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify(body), + }), + config, + { model: "", provider: "" } as RequestLogContext, + ); +} + +describe("responses input-size guard", () => { + test("rejects an input above the advertised context window without calling upstream", async () => { + let upstreamCalls = 0; + globalThis.fetch = (async () => { + upstreamCalls += 1; + return Response.json({ + id: "resp_x", + object: "response", + status: "completed", + output: [], + usage: { input_tokens: 1, output_tokens: 1, total_tokens: 2 }, + }); + }) as typeof fetch; + // 4.2MB of text ≈ 1.05M tokens at 4 bytes/token — above the 1M window. + const bigText = "a".repeat(4_200_000); + const res = await postResponses(deepseekConfig(), { + model: "deepseek/deepseek-v4-flash", + input: [{ role: "user", content: [{ type: "input_text", text: bigText }] }], + }); + expect(res.status).toBe(413); + expect(upstreamCalls).toBe(0); + }); + + test("forwards an input within the window", async () => { + let upstreamCalls = 0; + globalThis.fetch = (async () => { + upstreamCalls += 1; + return Response.json({ + id: "resp_x", + object: "response", + status: "completed", + output: [], + usage: { input_tokens: 1, output_tokens: 1, total_tokens: 2 }, + }); + }) as typeof fetch; + const res = await postResponses(deepseekConfig(), { + model: "deepseek/deepseek-v4-flash", + input: [{ role: "user", content: [{ type: "input_text", text: "hello" }] }], + }); + expect(upstreamCalls).toBe(1); + }); +}); From e51f614f2b739e21640bf18191c68e4576e64774 Mon Sep 17 00:00:00 2001 From: HoshimiRox1 <166687527+HoshimiRox1@users.noreply.github.com> Date: Fri, 7 Aug 2026 00:30:43 +0800 Subject: [PATCH 2/2] fix(codex): reframe routed compaction JSON to SSE for streaming clients The #875 transport policy forces a bounded JSON upstream for providers like deepseek (modelResponsesUpstreamStreaming: false), even when the client asked for SSE. The passthrough branch already reframes that JSON back to the canonical terminal SSE sequence, but routed-compaction turns skip the passthrough branch: they returned application/json, so Codex SSE parser hit EOF at the first byte and the remote compact task failed with "stream closed before response.completed", retrying 6x and leaving the thread unusable. Mirror the passthrough reframe in both routed branches (runTurn and parseStream): when the client requested stream and the upstream policy forced JSON, emit response.created, one output_item.done per output item (including the synthetic compaction item), response.completed, then [DONE]. --- src/server/responses/core.ts | 23 ++++++++ tests/responses-input-guard.test.ts | 90 +++++++++++++++++++++++++++++ 2 files changed, 113 insertions(+) diff --git a/src/server/responses/core.ts b/src/server/responses/core.ts index 99f17357b..eb84cac26 100644 --- a/src/server/responses/core.ts +++ b/src/server/responses/core.ts @@ -1318,6 +1318,25 @@ async function handleResponsesInner( logCtx: RequestLogContext, options: HandleResponsesOptions & { translatorBudget: TranslatorBudget }, ): Promise { + /** + * #875 mirror: the transport policy forced a bounded JSON upstream even though the client + * asked for SSE. Reframe the completed JSON as the canonical terminal SSE sequence — + * otherwise Codex's SSE parser hits EOF at the first byte ("stream closed before + * response.completed") and fails the turn. The passthrough branch already does this; the + * routed branches (runTurn and parseStream) need the same reframe (observed on deepseek + * compaction turns, where routedCompaction skips the passthrough reframe). + */ + const reframeRoutedJsonToSseIfRequested = (json: Record): Response | null => { + if (clientRequestedStream === true + && options.inboundTransport !== "websocket" + && providerModelResponsesUpstreamStreaming(route.providerName, route.provider, route.modelId) === false + && route.provider.adapter === "openai-responses") { + const sseHeaders = new Headers({ "content-type": "text/event-stream", "cache-control": "no-store" }); + return new Response(responsesJsonToSseBody(json), { status: 200, headers: sseHeaders }); + } + return null; + }; + // The Chat and Anthropic surfaces replay through here with a Responses-shaped body, // so an omitted value means a genuine Responses inbound. const inboundWire = options.inboundWire ?? "responses"; @@ -2726,6 +2745,8 @@ async function handleResponsesInner( adapterNeedsForcedContinuation(adapter.name) ? { force: true } : undefined, ); } + const reframed = reframeRoutedJsonToSseIfRequested(json); + if (reframed) return reframed; return new Response(JSON.stringify(json), { headers: { "Content-Type": "application/json" } }); } @@ -3429,6 +3450,8 @@ async function handleResponsesInner( activeAdapter.name === "kiro" ? { force: true } : undefined, ); } + const reframed = reframeRoutedJsonToSseIfRequested(json); + if (reframed) return reframed; return new Response(JSON.stringify(json), { headers: { "Content-Type": "application/json" } }); } diff --git a/tests/responses-input-guard.test.ts b/tests/responses-input-guard.test.ts index ce4f32d02..f856e167d 100644 --- a/tests/responses-input-guard.test.ts +++ b/tests/responses-input-guard.test.ts @@ -9,6 +9,8 @@ import { mkdtempSync, rmSync } from "node:fs"; import { tmpdir } from "node:os"; import { join } from "node:path"; import { handleResponses } from "../src/server/responses"; +import { enrichProviderFromRegistry, providerConfigSeed } from "../src/providers/derive"; +import { getProviderRegistryEntry } from "../src/providers/registry"; import type { OcxConfig } from "../src/types"; import type { RequestLogContext } from "../src/server/request-log"; @@ -108,3 +110,91 @@ describe("responses input-size guard", () => { expect(upstreamCalls).toBe(1); }); }); + +describe("routed compaction turn reframes bounded JSON to SSE", () => { + function seededDeepseekConfig(): OcxConfig { + const provider = { ...providerConfigSeed(getProviderRegistryEntry("deepseek")!), apiKey: "sk-test" }; + enrichProviderFromRegistry("deepseek", provider); + return { + port: 0, + defaultProvider: "deepseek", + providers: { deepseek: provider }, + } as OcxConfig; + } + + test("compaction_trigger with stream:true returns SSE with the compaction item and a terminal", async () => { + let upstreamCalls = 0; + globalThis.fetch = (async () => { + upstreamCalls += 1; + return Response.json({ + id: "resp_up", + object: "response", + status: "completed", + model: "deepseek-v4-flash", + output: [{ + id: "msg_1", + type: "message", + status: "completed", + role: "assistant", + content: [{ type: "output_text", text: "handoff summary text", annotations: [] }], + }], + usage: { + input_tokens: 3, + output_tokens: 2, + total_tokens: 5, + input_tokens_details: { cached_tokens: 0 }, + output_tokens_details: { reasoning_tokens: 0 }, + }, + }); + }) as typeof fetch; + const res = await postResponses(seededDeepseekConfig(), { + model: "deepseek/deepseek-v4-flash", + stream: true, + input: [ + { role: "user", content: [{ type: "input_text", text: "hello" }] }, + { type: "compaction_trigger" }, + ], + }); + expect(upstreamCalls).toBe(1); + expect(res.headers.get("content-type")).toContain("text/event-stream"); + const body = await res.text(); + expect(body).toContain("response.completed"); + expect(body).toContain('"type":"compaction"'); + expect(body).toContain("[DONE]"); + expect(body.indexOf('"type":"compaction"')).toBeLessThan(body.indexOf("response.completed")); + }); + + test("same compaction turn without stream:true stays JSON", async () => { + globalThis.fetch = (async () => Response.json({ + id: "resp_up", + object: "response", + status: "completed", + model: "deepseek-v4-flash", + output: [{ + id: "msg_1", + type: "message", + status: "completed", + role: "assistant", + content: [{ type: "output_text", text: "handoff summary text", annotations: [] }], + }], + usage: { + input_tokens: 3, + output_tokens: 2, + total_tokens: 5, + input_tokens_details: { cached_tokens: 0 }, + output_tokens_details: { reasoning_tokens: 0 }, + }, + })) as typeof fetch; + const res = await postResponses(seededDeepseekConfig(), { + model: "deepseek/deepseek-v4-flash", + stream: false, + input: [ + { role: "user", content: [{ type: "input_text", text: "hello" }] }, + { type: "compaction_trigger" }, + ], + }); + expect(res.headers.get("content-type")).toContain("application/json"); + const body = await res.text(); + expect(body).toContain('"type":"compaction"'); + }); +});