mirror of
https://github.com/diegosouzapw/OmniRoute.git
synced 2026-09-14 10:52:17 +03:00
fix(sse): drop the Codex read-only banner and align chatgpt-session failover
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.
This commit is contained in:
@@ -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<Uint8Array> };
|
||||
|
||||
/**
|
||||
* 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<ChatGptSessionStreamOpen> {
|
||||
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<Uint8Array>(
|
||||
{
|
||||
@@ -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;
|
||||
|
||||
@@ -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<Uint8Array> }).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<Record<string, unknown>>)[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<Uint8Array> }).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");
|
||||
});
|
||||
|
||||
Reference in New Issue
Block a user