fix(responses): refuse oversized input and reframe routed compaction to SSE (#1128)

src/server/responses/core.ts

@@ -1318,6 +1318,25 @@ async function handleResponsesInner(
   logCtx: RequestLogContext,
   options: HandleResponsesOptions & { translatorBudget: TranslatorBudget },
 ): Promise<Response> {
+  /**
+   * #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<string, unknown>): 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" } });
   }
 

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<string, unknown>): Promise<Response> {
+  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"');
+  });
+});