diff --git a/src/server/responses/core.ts b/src/server/responses/core.ts index 539617208..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"; @@ -1537,6 +1556,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 @@ -2692,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" } }); } @@ -3395,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 new file mode 100644 index 000000000..f856e167d --- /dev/null +++ b/tests/responses-input-guard.test.ts @@ -0,0 +1,200 @@ +/** + * 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 { 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"; + +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); + }); +}); + +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"'); + }); +});