feat(providers): bridge chatgpt-session adapter events to OpenAI payloads

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<Uint8Array> };
+
+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<string, unknown>,
+  finishReason: string | null,
+  usage?: OpenAiUsage
+): string {
+  const payload: Record<string, unknown> = {
+    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<AdapterEvent>,
+  meta: ChatGptSessionResponseMeta
+): Promise<ChatGptSessionStreamOpen> {
+  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<Uint8Array>(
+    {
+      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<string, unknown> } {
+  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<string, unknown>,
+    };
+  }
+
+  const message: Record<string, unknown> = { 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) } : {}),
+    },
+  };
+}

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<AdapterEvent> {
+  for (const event of events) yield event;
+}
+
+async function readAll(stream: ReadableStream<Uint8Array>): Promise<string> {
+  const decoder = new TextDecoder();
+  let out = "";
+  for await (const chunk of stream as unknown as AsyncIterable<Uint8Array>) {
+    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<Uint8Array> }).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<Uint8Array> }).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<Uint8Array> }).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<Uint8Array> }).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<Uint8Array> }).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<Record<string, unknown>>)[0];
+  const message = choice.message as Record<string, unknown>;
+  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<string, unknown>;
+  assert.doesNotMatch(String(error.message), /at \//);
+});