fix(codex): make remote compaction V2 complete reliably (#11041)

Compaction-V2 output now counts as real model output (no synthetic response.failed after response.completed), the Codex SSE filter handles CRLF framing, and terminal detection runs before scan-state bounding. 88/88 stream/readiness suites on the board. Thank you @jackjinke!

open-sse/executors/codex.ts

@@ -517,10 +517,12 @@ export function filterNonstandardCodexSse(response: Response): Response {
   const transform = new TransformStream<Uint8Array, Uint8Array>({
     transform(chunk, controller) {
       buffer += decoder.decode(chunk, { stream: true });
-      let sep: number;
-      while ((sep = buffer.indexOf("\n\n")) !== -1) {
-        const block = buffer.slice(0, sep + 2);
-        buffer = buffer.slice(sep + 2);
+      while (true) {
+        const separator = /\r?\n\r?\n/.exec(buffer);
+        if (!separator) break;
+        const blockEnd = separator.index + separator[0].length;
+        const block = buffer.slice(0, blockEnd);
+        buffer = buffer.slice(blockEnd);
         if (!dropBlock(block)) controller.enqueue(encoder.encode(block));
       }
     },

open-sse/utils/streamHandler.ts

@@ -683,13 +683,15 @@ export function createDisconnectAwareStream(transformStream, streamController) {
     if (clientTerminalSeen) return;
 
     terminalTail += terminalDecoder.decode(chunk, { stream: true });
-    if (terminalTail.length > 4096) {
-      terminalTail = terminalTail.slice(-4096);
-    }
+    // Scan before bounding retained state: a compaction terminal frame can
+    // exceed the tail budget because encrypted_content is carried inline.
     clientTerminalSeen = hasClientTerminalSseMarker(
       terminalTail,
       streamController.clientResponseFormat
     );
+    if (terminalTail.length > 4096) {
+      terminalTail = terminalTail.slice(-4096);
+    }
     if (clientTerminalSeen) {
       streamController.markClientTerminalSeen?.();
     }

open-sse/utils/streamReadiness.ts

@@ -34,6 +34,14 @@ function hasUsefulValue(value: unknown): boolean {
   if (Array.isArray(value)) return value.some(hasUsefulValue);
   if (!isRecord(value)) return false;
 
+  // A Responses compaction item IS the turn's output: remote compaction
+  // completes with output = [{type:"compaction", encrypted_content}] and no
+  // assistant text. Deliberately NOT a blanket encrypted_content key — an
+  // encrypted reasoning item alone is not user-visible output and must keep
+  // tripping the #8649 empty-content guard.
+  // This shape is specific to Responses streams; chat-completion frames do not produce it.
+  if (value.type === "compaction" && hasNonEmptyString(value.encrypted_content)) return true;
+
   for (const key of [
     "content",
     "text",

tests/unit/codex-drop-nonstandard-events.test.ts

@@ -17,6 +17,19 @@ function sseResponse(body: string): Response {
   });
 }
 
+function chunkedSseResponse(chunks: string[]): Response {
+  const encoder = new TextEncoder();
+  return new Response(
+    new ReadableStream({
+      start(controller) {
+        for (const chunk of chunks) controller.enqueue(encoder.encode(chunk));
+        controller.close();
+      },
+    }),
+    { status: 200, headers: { "content-type": "text/event-stream" } }
+  );
+}
+
 async function readAll(res: Response): Promise<string> {
   return await res.text();
 }
@@ -61,10 +74,10 @@ describe("codexDropNonstandardEvents (#11014)", () => {
 describe("filterNonstandardCodexSse (#4715)", () => {
   it("drops codex.* event blocks but keeps standard response.* events", async () => {
     const stream =
-      "event: response.created\ndata: {\"type\":\"response.created\"}\n\n" +
+      'event: response.created\ndata: {"type":"response.created"}\n\n' +
       "event: codex.rate_limits\n\n" +
-      "event: response.output_text.delta\ndata: {\"delta\":\"hi\"}\n\n" +
-      "event: response.completed\ndata: {\"type\":\"response.completed\"}\n\n";
+      'event: response.output_text.delta\ndata: {"delta":"hi"}\n\n' +
+      'event: response.completed\ndata: {"type":"response.completed"}\n\n';
     const out = await readAll(filterNonstandardCodexSse(sseResponse(stream)));
     assert.ok(!out.includes("codex.rate_limits"), "codex.* frame must be stripped");
     assert.ok(out.includes("response.created"), "standard events preserved");
@@ -72,18 +85,31 @@ describe("filterNonstandardCodexSse (#4715)", () => {
     assert.ok(out.includes("response.completed"), "terminal event preserved");
   });
 
+  it("filters CRLF-framed events split across transport chunks", async () => {
+    const response = chunkedSseResponse([
+      'event: response.created\r\ndata: {"type":"response.created"}\r\n\r',
+      "\nevent: codex.rate_limits\r\n\r\n",
+      'event: response.completed\r\ndata: {"type":"response.completed"}\r\n\r\n',
+    ]);
+
+    const out = await readAll(filterNonstandardCodexSse(response));
+
+    assert.ok(!out.includes("codex.rate_limits"), "codex.* frame must be stripped");
+    assert.ok(out.includes("response.created"), "standard events preserved");
+    assert.ok(out.includes("response.completed"), "terminal event preserved");
+  });
+
   it("passes through non-SSE responses untouched", async () => {
-    const json = new Response("{\"ok\":true}", {
+    const json = new Response('{"ok":true}', {
       status: 200,
       headers: { "content-type": "application/json" },
     });
     const out = filterNonstandardCodexSse(json);
-    assert.equal(await out.text(), "{\"ok\":true}");
+    assert.equal(await out.text(), '{"ok":true}');
   });
 
   it("drops a trailing codex.* block with no double-newline terminator (flush path)", async () => {
-    const stream =
-      "event: response.created\ndata: {}\n\n" + "event: codex.token_count\ndata: {}";
+    const stream = "event: response.created\ndata: {}\n\n" + "event: codex.token_count\ndata: {}";
     const out = await readAll(filterNonstandardCodexSse(sseResponse(stream)));
     assert.ok(out.includes("response.created"));
     assert.ok(!out.includes("codex.token_count"));

tests/unit/empty-stream-no-content-8649.test.ts

@@ -252,3 +252,68 @@ test("#8649 buildStreamErrorChunks-shaped error must not be rewritten as empty c
   assert.match(text, /AI Model Not Found/);
   assert.doesNotMatch(text, /Provider returned empty content/);
 });
+
+test("#8649 a Responses compaction-only stream is real output, not empty content", async () => {
+  // Codex remote compaction V2: POST /v1/responses with a compaction_trigger
+  // input item completes with output = [{type:"compaction", encrypted_content}]
+  // and no assistant text. The watcher's content keys do not include
+  // encrypted_content, so the healthy stream was followed by a synthetic
+  // response.failed ("Provider returned empty content") — which strict
+  // Responses clients reject even after response.completed.
+  const text = await runClientStream(
+    [
+      `data: {"type":"response.in_progress"}\n\n`,
+      `event: response.created\ndata: ${JSON.stringify({
+        type: "response.created",
+        response: { id: "resp_cmp", status: "in_progress", output: [] },
+      })}\n\n`,
+      `event: response.completed\ndata: ${JSON.stringify({
+        type: "response.completed",
+        response: {
+          id: "resp_cmp",
+          status: "completed",
+          output: [
+            { id: "cmp_1", type: "compaction", encrypted_content: "gAAAAABencryptedpayload" },
+          ],
+        },
+      })}\n\n`,
+    ],
+    FORMATS.OPENAI_RESPONSES
+  );
+
+  assert.match(text, /"type":"compaction"/);
+  assert.doesNotMatch(
+    text,
+    /Provider returned empty content|response\.failed/,
+    "a completed compaction response must not be followed by a synthetic failure frame"
+  );
+});
+
+test("#8649 an encrypted-reasoning-only stream is still empty content", async () => {
+  // Inverse of the compaction carve-out: an encrypted reasoning item is not
+  // user-visible output. A turn that produces only a reasoning trace and no
+  // message/tool call is the fake-success shape this guard exists to catch.
+  const text = await runClientStream(
+    [
+      `event: response.created\ndata: ${JSON.stringify({
+        type: "response.created",
+        response: { id: "resp_r", status: "in_progress", output: [] },
+      })}\n\n`,
+      `event: response.completed\ndata: ${JSON.stringify({
+        type: "response.completed",
+        response: {
+          id: "resp_r",
+          status: "completed",
+          output: [{ id: "rs_1", type: "reasoning", encrypted_content: "gAAAAABencryptedtrace" }],
+        },
+      })}\n\n`,
+    ],
+    FORMATS.OPENAI_RESPONSES
+  );
+
+  assert.match(
+    text,
+    /response\.failed|Provider returned empty content/,
+    "a reasoning-only turn must keep tripping the empty-content guard"
+  );
+});

tests/unit/stream-handler.test.ts

@@ -167,6 +167,41 @@ test("createDisconnectAwareStream treats cancel after Responses completed as suc
   assert.equal(disconnectHandled, false);
 });
 
+test("createDisconnectAwareStream recognizes a large Responses compaction completion", async () => {
+  let errorHandled = false;
+  const completed = `event: response.completed\ndata: ${JSON.stringify({
+    type: "response.completed",
+    response: {
+      status: "completed",
+      output: [{ type: "compaction", encrypted_content: "x".repeat(5000) }],
+    },
+  })}\n\n`;
+  const transformStream = {
+    readable: new ReadableStream({
+      start(controller) {
+        controller.enqueue(encoder.encode(completed));
+        controller.close();
+      },
+    }),
+    writable: createNoopAbortWritable(),
+  };
+
+  const stream = createDisconnectAwareStream(
+    transformStream,
+    createStreamController({
+      clientResponseFormat: FORMATS.OPENAI_RESPONSES,
+      onError() {
+        errorHandled = true;
+      },
+    })
+  );
+  const text = await readStreamText(stream);
+
+  assert.equal(text, completed);
+  assert.equal(errorHandled, false);
+  assert.doesNotMatch(text, /response\.failed/);
+});
+
 test("createDisconnectAwareStream: Gemini 503 high-demand error becomes SSE error chunk with message preserved", async () => {
   const geminiMsg =
     "[503]: This model is currently experiencing high demand. Spikes in demand are usually temporary. Please try again later.";