From 30eabbef3bab516b51334f131ea119e607b24968 Mon Sep 17 00:00:00 2001 From: Markus Hartung Date: Wed, 2 Sep 2026 08:30:33 -0300 Subject: [PATCH] fix(sse): drop the Codex read-only banner and align chatgpt-session failover MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The vendored adapter emits its "local Codex computer is unavailable" warning as a commentary-phase text_delta on every fresh turn, because this provider pins localToolsEnabled: false. The bridge treated every text_delta alike and let assistant_boundary fall through to `default`, so 100% of answers were prefixed with the banner. Commentary-phase deltas are now dropped on both the streaming and the buffered path (not rerouted to reasoning_content — it is transport chatter, not model reasoning), and assistant_boundary gets an explicit ignoring case. Same commit aligns the streaming gate with the buffered failover: the stream no longer commits a 200 on a heartbeat, an assistant boundary, a commentary delta, an empty text delta or a thinking delta, so `thinking_delta` followed by `error{429}` now returns HTTP 429 on both paths instead of a 200 stream carrying an in-band error. Reasoning that arrives while the gate is closed is buffered and replayed once the gate opens. --- open-sse/executors/chatgpt-session/bridge.ts | 67 +++++++++-- tests/unit/chatgpt-session-bridge.test.ts | 116 +++++++++++++++++++ 2 files changed, 171 insertions(+), 12 deletions(-) diff --git a/open-sse/executors/chatgpt-session/bridge.ts b/open-sse/executors/chatgpt-session/bridge.ts index 7c766ab9b0..772dd09e8f 100644 --- a/open-sse/executors/chatgpt-session/bridge.ts +++ b/open-sse/executors/chatgpt-session/bridge.ts @@ -1,15 +1,22 @@ /** * Bridges the vendored adapter's event stream into OpenAI chat-completions payloads. * - * Stream opening is gated on the first meaningful event so a turn that fails before producing - * any output can still be answered with a real HTTP status instead of a 200 stream carrying an - * error chunk. Once any text has been emitted the status line is already committed, so a later - * failure just closes the stream cleanly. + * Stream opening is gated on the first event that would ALSO count as committed output on the + * buffered path, so a turn that fails before producing any assistant content can still be + * answered with a real HTTP status instead of a 200 stream carrying an error chunk. The gate and + * `buildChatGptSessionCompletion` must agree on what "output" means — transport framing + * (heartbeats, assistant boundaries), commentary-phase text, empty text deltas and reasoning are + * all non-committing on both paths. Once real content has been emitted the status line is + * already committed, so a later failure just closes the stream cleanly. */ import { buildErrorBody, sanitizeErrorMessage } from "../../utils/error.ts"; import { formatTranslatedStreamError } from "../../utils/streamErrorFormat.ts"; -import type { AdapterEvent, CodexUsage } from "../../vendor/codex-chatgpt-web/types.ts"; +import type { + AdapterEvent, + CodexMessagePhase, + CodexUsage, +} from "../../vendor/codex-chatgpt-web/types.ts"; import { classifyChatGptSessionError } from "./errors.ts"; export interface ChatGptSessionResponseMeta { @@ -19,9 +26,23 @@ export interface ChatGptSessionResponseMeta { } export type ChatGptSessionStreamOpen = - | { kind: "error"; status: number; code: string; message: string } + | { + kind: "error"; + status: number; + code: string; + message: string; + fallbackHint?: "connection_cooldown"; + } | { kind: "stream"; stream: ReadableStream }; +/** + * The adapter tags its own transport chatter with the commentary phase — most visibly the + * "local Codex computer is unavailable" banner it emits on every fresh turn while + * `localToolsEnabled` is false, which is this provider's permanent configuration. It is not + * model output and it is not model reasoning, so it never reaches the client on either path. + */ +const COMMENTARY_PHASE: CodexMessagePhase = "commentary"; + interface OpenAiUsage { prompt_tokens: number; completion_tokens: number; @@ -70,13 +91,23 @@ export async function openChatGptSessionStream( meta: ChatGptSessionResponseMeta ): Promise { const iterator = events[Symbol.asyncIterator](); + // Reasoning that arrives while the gate is still closed is real output the client must + // receive; it just may not commit the status line, because the buffered path fails over on + // absent CONTENT regardless of how much reasoning preceded it. + const bufferedReasoning: AdapterEvent[] = []; let first: AdapterEvent | null = null; for (;;) { const next = await iterator.next(); if (next.done) break; - if (next.value.type === "heartbeat") continue; - first = next.value; + const event = next.value; + if (event.type === "heartbeat" || event.type === "assistant_boundary") continue; + if (event.type === "text_delta" && (!event.text || event.phase === COMMENTARY_PHASE)) continue; + if (event.type === "thinking_delta") { + bufferedReasoning.push(event); + continue; + } + first = event; break; } @@ -87,10 +118,11 @@ export async function openChatGptSessionStream( status: classified.status, code: classified.code, message: sanitizeErrorMessage(first.message), + ...(classified.fallbackHint ? { fallbackHint: classified.fallbackHint } : {}), }; } - const pending = first; + const pending: AdapterEvent[] = first ? [...bufferedReasoning, first] : bufferedReasoning; const encoder = new TextEncoder(); const stream = new ReadableStream( { @@ -106,11 +138,18 @@ export async function openChatGptSessionStream( emit(": keepalive\n\n"); return true; case "text_delta": + if (event.phase === COMMENTARY_PHASE) return true; if (event.text) emit(chunk(meta, { content: event.text }, null)); return true; case "thinking_delta": if (event.thinking) emit(chunk(meta, { reasoning_content: event.thinking }, null)); return true; + case "assistant_boundary": + // Internal framing between the adapter's guarded first pass and its one-shot + // continuation (the vendor's Responses bridge only closes the open item here). + // There is no chat-completions delta for it, and letting it reach `default` would + // be indistinguishable from a real event we forgot to handle. + return true; case "done": emit(chunk(meta, {}, "stop", mapUsage(event.usage))); return false; @@ -137,7 +176,10 @@ export async function openChatGptSessionStream( try { let open = true; - if (pending) open = handle(pending); + for (const event of pending) { + open = handle(event); + if (!open) break; + } while (open) { const next = await iterator.next(); if (next.done) { @@ -172,8 +214,9 @@ export function buildChatGptSessionCompletion( let failure: AdapterEvent | null = null; for (const event of events) { - if (event.type === "text_delta") content += event.text; - else if (event.type === "thinking_delta") reasoning += event.thinking; + if (event.type === "text_delta") { + if (event.phase !== COMMENTARY_PHASE) content += event.text; + } else if (event.type === "thinking_delta") reasoning += event.thinking; else if (event.type === "done") usage = event.usage; else if (event.type === "incomplete") { usage = event.usage; diff --git a/tests/unit/chatgpt-session-bridge.test.ts b/tests/unit/chatgpt-session-bridge.test.ts index f4a6b8763e..8cadc7f073 100644 --- a/tests/unit/chatgpt-session-bridge.test.ts +++ b/tests/unit/chatgpt-session-bridge.test.ts @@ -215,3 +215,119 @@ test("buffered error bodies never leak a stack trace", () => { assert.doesNotMatch(String(error.message), /at \//); assert.match(String(error.message), /failure/); }); + +test("the commentary read-only banner never reaches streamed content", async () => { + const opened = await openChatGptSessionStream( + iterate([ + { type: "assistant_boundary" }, + { + type: "text_delta", + text: "⚠️ The local Codex computer is unavailable, so this turn is read-only.", + phase: "commentary", + }, + { type: "assistant_boundary" }, + { type: "text_delta", text: "real answer" }, + { type: "done" }, + ]), + META + ); + assert.equal(opened.kind, "stream"); + const text = await readAll((opened as { stream: ReadableStream }).stream); + const content = [...text.matchAll(/"content":"((?:[^"\\]|\\.)*)"/g)] + .map((match) => JSON.parse(`"${match[1]}"`) as string) + .join(""); + assert.equal(content, "real answer"); + assert.doesNotMatch(text, /read-only/); + assert.doesNotMatch(text, /Codex computer/); + // Commentary must not be laundered into reasoning either — it is transport chatter. + assert.doesNotMatch(text, /"reasoning_content"/); +}); + +test("the commentary read-only banner never reaches buffered content", () => { + const result = buildChatGptSessionCompletion( + [ + { type: "assistant_boundary" }, + { + type: "text_delta", + text: "⚠️ The local Codex computer is unavailable, so this turn is read-only.", + phase: "commentary", + }, + { type: "assistant_boundary" }, + { type: "text_delta", text: "real answer" }, + { type: "done" }, + ], + META + ); + assert.equal(result.status, 200); + const message = (result.body.choices as Array>)[0].message as Record< + string, + unknown + >; + assert.equal(message.content, "real answer"); + assert.equal("reasoning_content" in message, false); + assert.doesNotMatch(JSON.stringify(result.body), /read-only/); +}); + +// I1 — the streaming gate and the buffered failover must agree on what counts as committed +// output. Reasoning alone commits neither, so a failure right after a thinking delta is a real +// HTTP status on BOTH paths instead of a 200 stream carrying an in-band error chunk. +test("reasoning before an error does not commit a 200 on the streaming path", async () => { + const opened = await openChatGptSessionStream( + iterate([ + { type: "thinking_delta", thinking: "weighing options" }, + { type: "error", message: "ChatGPT reported a usage limit", status: 429 }, + ]), + META + ); + assert.equal(opened.kind, "error"); + assert.equal((opened as { status: number }).status, 429); +}); + +test("the buffered path returns the same status for reasoning followed by an error", () => { + const result = buildChatGptSessionCompletion( + [ + { type: "thinking_delta", thinking: "weighing options" }, + { type: "error", message: "ChatGPT reported a usage limit", status: 429 }, + ], + META + ); + assert.equal(result.status, 429); +}); + +test("reasoning buffered behind the gate is still streamed once the gate opens", async () => { + const opened = await openChatGptSessionStream( + iterate([ + { type: "thinking_delta", thinking: "first" }, + { type: "thinking_delta", thinking: "second" }, + { type: "text_delta", text: "answer" }, + { type: "done" }, + ]), + META + ); + assert.equal(opened.kind, "stream"); + const text = await readAll((opened as { stream: ReadableStream }).stream); + assert.match(text, /"reasoning_content":"first"/); + assert.match(text, /"reasoning_content":"second"/); + assert.match(text, /"content":"answer"/); +}); + +test("an empty text delta does not commit a 200 ahead of an error", async () => { + const opened = await openChatGptSessionStream( + iterate([ + { type: "text_delta", text: "" }, + { type: "error", message: "ChatGPT reported a usage limit" }, + ]), + META + ); + assert.equal(opened.kind, "error"); + assert.equal((opened as { status: number }).status, 429); +}); + +test("a cooldown-hinted classification reaches the stream-open error verdict", async () => { + const opened = await openChatGptSessionStream( + iterate([{ type: "error", message: "upstream unavailable", status: 503 }]), + META + ); + assert.equal(opened.kind, "error"); + assert.equal((opened as { fallbackHint?: string }).fallbackHint, "connection_cooldown"); +});