diff --git a/open-sse/executors/codex.ts b/open-sse/executors/codex.ts index e1674c30fb..6e39bb81ca 100644 --- a/open-sse/executors/codex.ts +++ b/open-sse/executors/codex.ts @@ -517,10 +517,12 @@ export function filterNonstandardCodexSse(response: Response): Response { const transform = new TransformStream({ 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)); } }, diff --git a/open-sse/utils/streamHandler.ts b/open-sse/utils/streamHandler.ts index 4d8993c767..9adb5dbf2f 100644 --- a/open-sse/utils/streamHandler.ts +++ b/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?.(); } diff --git a/open-sse/utils/streamReadiness.ts b/open-sse/utils/streamReadiness.ts index 2d06659b80..1696a7a5c5 100644 --- a/open-sse/utils/streamReadiness.ts +++ b/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", diff --git a/tests/unit/codex-drop-nonstandard-events.test.ts b/tests/unit/codex-drop-nonstandard-events.test.ts index d2aff54814..dec24970ca 100644 --- a/tests/unit/codex-drop-nonstandard-events.test.ts +++ b/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 { 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")); diff --git a/tests/unit/empty-stream-no-content-8649.test.ts b/tests/unit/empty-stream-no-content-8649.test.ts index e46157fbe1..918ae7c4c0 100644 --- a/tests/unit/empty-stream-no-content-8649.test.ts +++ b/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" + ); +}); diff --git a/tests/unit/stream-handler.test.ts b/tests/unit/stream-handler.test.ts index d19a13f351..8c19c08802 100644 --- a/tests/unit/stream-handler.test.ts +++ b/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.";