From ed00c8c4893a9e7362988fcdee7d0d1025ffe268 Mon Sep 17 00:00:00 2001 From: Markus Hartung Date: Wed, 2 Sep 2026 01:56:21 -0300 Subject: [PATCH] feat(providers): bridge chatgpt-session adapter events to OpenAI payloads --- open-sse/executors/chatgpt-session/bridge.ts | 199 +++++++++++++++++++ tests/unit/chatgpt-session-bridge.test.ts | 142 +++++++++++++ 2 files changed, 341 insertions(+) create mode 100644 open-sse/executors/chatgpt-session/bridge.ts create mode 100644 tests/unit/chatgpt-session-bridge.test.ts diff --git a/open-sse/executors/chatgpt-session/bridge.ts b/open-sse/executors/chatgpt-session/bridge.ts new file mode 100644 index 0000000000..217e42df57 --- /dev/null +++ b/open-sse/executors/chatgpt-session/bridge.ts @@ -0,0 +1,199 @@ +/** + * 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. + */ + +import { buildErrorBody, sanitizeErrorMessage } from "../../utils/error.ts"; +import type { AdapterEvent, CodexUsage } from "../../vendor/codex-chatgpt-web/types.ts"; +import { classifyChatGptSessionError } from "./errors.ts"; + +export interface ChatGptSessionResponseMeta { + cid: string; + created: number; + model: string; +} + +export type ChatGptSessionStreamOpen = + | { kind: "error"; status: number; code: string; message: string } + | { kind: "stream"; stream: ReadableStream }; + +interface OpenAiUsage { + prompt_tokens: number; + completion_tokens: number; + total_tokens: number; + completion_tokens_details?: { reasoning_tokens: number }; +} + +function mapUsage(usage: CodexUsage | undefined): OpenAiUsage | undefined { + if (!usage) return undefined; + const prompt = usage.inputTokens ?? 0; + const completion = usage.outputTokens ?? 0; + return { + prompt_tokens: prompt, + completion_tokens: completion, + total_tokens: usage.totalTokens ?? prompt + completion, + ...(typeof usage.reasoningOutputTokens === "number" + ? { completion_tokens_details: { reasoning_tokens: usage.reasoningOutputTokens } } + : {}), + }; +} + +function chunk( + meta: ChatGptSessionResponseMeta, + delta: Record, + finishReason: string | null, + usage?: OpenAiUsage +): string { + const payload: Record = { + id: meta.cid, + object: "chat.completion.chunk", + created: meta.created, + model: meta.model, + choices: [{ index: 0, delta, finish_reason: finishReason }], + }; + if (usage) payload.usage = usage; + return `data: ${JSON.stringify(payload)}\n\n`; +} + +function finishReasonFor(event: AdapterEvent): string { + if (event.type === "incomplete") return event.endTurn ? "stop" : "length"; + return "stop"; +} + +export async function openChatGptSessionStream( + events: AsyncIterable, + meta: ChatGptSessionResponseMeta +): Promise { + const iterator = events[Symbol.asyncIterator](); + let first: AdapterEvent | null = null; + + for (;;) { + const next = await iterator.next(); + if (next.done) break; + if (next.value.type === "heartbeat") continue; + first = next.value; + break; + } + + if (first && first.type === "error") { + const classified = classifyChatGptSessionError(first); + return { + kind: "error", + status: classified.status, + code: classified.code, + message: first.message, + }; + } + + const pending = first; + const encoder = new TextEncoder(); + const stream = new ReadableStream( + { + async start(controller) { + const emit = (text: string) => controller.enqueue(encoder.encode(text)); + emit(chunk(meta, { role: "assistant" }, null)); + + const handle = (event: AdapterEvent): boolean => { + switch (event.type) { + case "heartbeat": + emit(": keepalive\n\n"); + return true; + case "text_delta": + 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 "done": + emit(chunk(meta, {}, "stop", mapUsage(event.usage))); + return false; + case "incomplete": + emit(chunk(meta, {}, finishReasonFor(event), mapUsage(event.usage))); + return false; + case "error": + emit(chunk(meta, {}, "stop", mapUsage(event.usage))); + return false; + default: + return true; + } + }; + + try { + let open = true; + if (pending) open = handle(pending); + while (open) { + const next = await iterator.next(); + if (next.done) { + emit(chunk(meta, {}, "stop")); + break; + } + open = handle(next.value); + } + } finally { + emit("data: [DONE]\n\n"); + controller.close(); + } + }, + cancel() { + void iterator.return?.(); + }, + }, + { highWaterMark: 16384 } + ); + + return { kind: "stream", stream }; +} + +export function buildChatGptSessionCompletion( + events: readonly AdapterEvent[], + meta: ChatGptSessionResponseMeta +): { status: number; body: Record } { + let content = ""; + let reasoning = ""; + let finishReason = "stop"; + let usage: CodexUsage | undefined; + 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; + else if (event.type === "done") usage = event.usage; + else if (event.type === "incomplete") { + usage = event.usage; + finishReason = finishReasonFor(event); + } else if (event.type === "error") { + usage = event.usage ?? usage; + failure = event; + } + } + + if (failure && failure.type === "error" && !content) { + const classified = classifyChatGptSessionError(failure); + return { + status: classified.status, + body: buildErrorBody(classified.status, sanitizeErrorMessage(failure.message), undefined, { + type: classified.status >= 500 ? "provider_error" : "invalid_request_error", + code: classified.code, + }) as unknown as Record, + }; + } + + const message: Record = { role: "assistant", content }; + if (reasoning) message.reasoning_content = reasoning; + + return { + status: 200, + body: { + id: meta.cid, + object: "chat.completion", + created: meta.created, + model: meta.model, + choices: [{ index: 0, message, finish_reason: finishReason, logprobs: null }], + ...(mapUsage(usage) ? { usage: mapUsage(usage) } : {}), + }, + }; +} diff --git a/tests/unit/chatgpt-session-bridge.test.ts b/tests/unit/chatgpt-session-bridge.test.ts new file mode 100644 index 0000000000..1bbf246cc1 --- /dev/null +++ b/tests/unit/chatgpt-session-bridge.test.ts @@ -0,0 +1,142 @@ +import { test } from "node:test"; +import assert from "node:assert/strict"; + +import { + buildChatGptSessionCompletion, + openChatGptSessionStream, +} from "../../open-sse/executors/chatgpt-session/bridge.ts"; +import type { AdapterEvent } from "../../open-sse/vendor/codex-chatgpt-web/types.ts"; + +const META = { cid: "chatcmpl-test", created: 1_700_000_000, model: "chatgpt-session/high" }; + +async function* iterate(events: AdapterEvent[]): AsyncGenerator { + for (const event of events) yield event; +} + +async function readAll(stream: ReadableStream): Promise { + const decoder = new TextDecoder(); + let out = ""; + for await (const chunk of stream as unknown as AsyncIterable) { + out += decoder.decode(chunk, { stream: true }); + } + return out + decoder.decode(); +} + +test("streams role, content and a terminal stop chunk", async () => { + const opened = await openChatGptSessionStream( + iterate([ + { type: "text_delta", text: "Hel" }, + { type: "text_delta", text: "lo" }, + { type: "done", usage: { inputTokens: 10, outputTokens: 2 } }, + ]), + META + ); + assert.equal(opened.kind, "stream"); + const text = await readAll((opened as { stream: ReadableStream }).stream); + assert.match(text, /"delta":\{"role":"assistant"\}/); + assert.match(text, /"content":"Hel"/); + assert.match(text, /"content":"lo"/); + assert.match(text, /"finish_reason":"stop"/); + assert.match(text, /"prompt_tokens":10/); + assert.ok(text.trimEnd().endsWith("data: [DONE]")); +}); + +test("thinking deltas surface as reasoning_content", async () => { + const opened = await openChatGptSessionStream( + iterate([ + { type: "thinking_delta", thinking: "hmm" }, + { type: "text_delta", text: "ok" }, + { type: "done" }, + ]), + META + ); + const text = await readAll((opened as { stream: ReadableStream }).stream); + assert.match(text, /"reasoning_content":"hmm"/); +}); + +test("heartbeats become SSE comments, never data chunks", async () => { + const opened = await openChatGptSessionStream( + iterate([{ type: "text_delta", text: "x" }, { type: "heartbeat" }, { type: "done" }]), + META + ); + const text = await readAll((opened as { stream: ReadableStream }).stream); + assert.match(text, /^: keepalive$/m); + assert.doesNotMatch(text, /"heartbeat"/); +}); + +test("an error before any output returns an error verdict instead of a stream", async () => { + const opened = await openChatGptSessionStream( + iterate([{ type: "error", message: "ChatGPT page is not authenticated" }]), + META + ); + assert.equal(opened.kind, "error"); + assert.equal((opened as { status: number }).status, 401); + assert.equal((opened as { code: string }).code, "session_expired"); +}); + +test("an error mid-stream terminates the stream after the emitted text", async () => { + const opened = await openChatGptSessionStream( + iterate([ + { type: "text_delta", text: "partial" }, + { type: "error", message: "boom", status: 502 }, + ]), + META + ); + assert.equal(opened.kind, "stream"); + const text = await readAll((opened as { stream: ReadableStream }).stream); + assert.match(text, /"content":"partial"/); + assert.match(text, /"finish_reason":"stop"/); + assert.ok(text.trimEnd().endsWith("data: [DONE]")); +}); + +test("incomplete maps to a length finish reason", async () => { + const opened = await openChatGptSessionStream( + iterate([ + { type: "text_delta", text: "x" }, + { type: "incomplete", reason: "max_output" }, + ]), + META + ); + const text = await readAll((opened as { stream: ReadableStream }).stream); + assert.match(text, /"finish_reason":"length"/); +}); + +test("buffered completion collects content, reasoning and usage", () => { + const result = buildChatGptSessionCompletion( + [ + { type: "thinking_delta", thinking: "think" }, + { type: "text_delta", text: "Hello" }, + { type: "text_delta", text: " world" }, + { type: "done", usage: { inputTokens: 7, outputTokens: 3, totalTokens: 10 } }, + ], + META + ); + assert.equal(result.status, 200); + const choice = (result.body.choices as Array>)[0]; + const message = choice.message as Record; + assert.equal(message.content, "Hello world"); + assert.equal(message.reasoning_content, "think"); + assert.equal(choice.finish_reason, "stop"); + assert.deepEqual(result.body.usage, { + prompt_tokens: 7, + completion_tokens: 3, + total_tokens: 10, + }); +}); + +test("buffered completion surfaces an error event as a classified status", () => { + const result = buildChatGptSessionCompletion( + [{ type: "error", message: "ChatGPT reported a usage limit" }], + META + ); + assert.equal(result.status, 429); +}); + +test("buffered error bodies never leak a stack trace", () => { + const result = buildChatGptSessionCompletion( + [{ type: "error", message: "failure\n at /app/open-sse/x.ts:1:1" }], + META + ); + const error = result.body.error as Record; + assert.doesNotMatch(String(error.message), /at \//); +});