diff --git a/open-sse/executors/chatgpt-session.ts b/open-sse/executors/chatgpt-session.ts new file mode 100644 index 0000000000..4633406c0f --- /dev/null +++ b/open-sse/executors/chatgpt-session.ts @@ -0,0 +1,333 @@ +/** + * ChatGptSessionExecutor — OpenAI chat completions over an authenticated ChatGPT browser + * session. + * + * The vendored MIT browser adapter owns every anti-bot interaction (sentinel, turnstile, + * proof-of-work) because a real signed-in browser performs them; this executor only translates + * request and response shapes around it. + */ + +import { CHATGPT_WEB_CODEX_CONNECTOR_NAME } from "@/shared/constants/chatgptWebCodex"; + +import { FORMATS } from "../translator/formats.ts"; +import { prepareToolMessages } from "../translator/webTools.ts"; +import { buildErrorBody, sanitizeErrorMessage } from "../utils/error.ts"; +import { AsyncEventQueue } from "../vendor/codex-chatgpt-web/event-queue.ts"; +import type { AdapterEvent, CodexProviderConfig } from "../vendor/codex-chatgpt-web/types.ts"; +import { BaseExecutor, type ExecuteInput, type ExecutorExecuteResult } from "./base.ts"; +import { buildToolModeResponse } from "./chatgptWebTools.ts"; +import { + decodeChatGptWebCodexSecrets, + encodeChatGptWebCodexSecrets, +} from "./chatgpt-web-codex/credentials.ts"; +import { connectionRuntimePaths } from "./chatgpt-web-codex/storageState.ts"; +import { + buildChatGptSessionCompletion, + openChatGptSessionStream, + type ChatGptSessionResponseMeta, +} from "./chatgpt-session/bridge.ts"; +import { classifyChatGptSessionError } from "./chatgpt-session/errors.ts"; +import { buildParsedRequest } from "./chatgpt-session/messages.ts"; +import { requireChatGptSessionRoute, type ChatGptSessionRoute } from "./chatgpt-session/models.ts"; +import { + chatGptSessionRuntime, + type ChatGptSessionLoginConfig, +} from "./chatgpt-session/runtime.ts"; + +const BASE_URL = "https://chatgpt.com"; +const JSON_HEADERS = { "Content-Type": "application/json" }; +const SSE_HEADERS = { + "Cache-Control": "no-cache, no-transform", + Connection: "keep-alive", + "Content-Type": "text/event-stream; charset=utf-8", +}; + +function record(value: unknown): Record { + return value && typeof value === "object" && !Array.isArray(value) + ? (value as Record) + : {}; +} + +function configuredString(data: Record, ...keys: string[]): string | undefined { + for (const key of keys) { + const value = data[key]; + if (typeof value === "string" && value.trim()) return value.trim(); + } + return undefined; +} + +function wrapped(response: Response, body: unknown): ExecutorExecuteResult { + return { + response, + url: BASE_URL, + headers: {}, + transformedBody: body, + transport: "chatgpt-session-browser", + }; +} + +function errorResponse( + status: number, + message: unknown, + code: string, + fallbackHint?: "connection_cooldown" +): Response { + return new Response( + JSON.stringify( + buildErrorBody(status, sanitizeErrorMessage(message), undefined, { + type: status >= 500 ? "provider_error" : "invalid_request_error", + code, + }) + ), + { + status, + headers: fallbackHint + ? { ...JSON_HEADERS, "X-Omni-Fallback-Hint": fallbackHint } + : JSON_HEADERS, + } + ); +} + +export function buildChatGptSessionProviderConfig(args: { + route: ChatGptSessionRoute; + connectionId: string; + storageStatePath: string; + connectorName: string; + chromeExecutablePath?: string; + cdpEndpoint?: string; + solAvailable: boolean; + proAvailable: boolean; +}): CodexProviderConfig { + const paths = connectionRuntimePaths(args.connectionId); + return { + adapter: "chatgpt-web", + baseUrl: BASE_URL, + defaultModel: args.route.backendModel, + models: [args.route.backendModel], + chatgptWeb: { + appName: args.connectorName, + storageStatePath: args.storageStatePath, + ...(args.chromeExecutablePath ? { chromeExecutablePath: args.chromeExecutablePath } : {}), + ...(args.cdpEndpoint ? { cdpEndpoint: args.cdpEndpoint } : {}), + brokerSocketPath: paths.brokerSocketPath, + threadEnvironmentStatePath: paths.threadEnvironmentStatePath, + lunaCheckpointStatePath: paths.lunaCheckpointStatePath, + headed: true, + // Prompt-emulated tools only: never attach the turn-bound Codex connector capability. + localToolsEnabled: false, + solAvailable: args.solAvailable, + proAvailable: args.proAvailable, + autoApproveToolCalls: false, + }, + }; +} + +export class ChatGptSessionExecutor extends BaseExecutor { + constructor() { + super("chatgpt-session", { + id: "chatgpt-session", + baseUrl: BASE_URL, + format: FORMATS.OPENAI, + }); + } + + override async execute(input: ExecuteInput): Promise { + const runtime = chatGptSessionRuntime(); + const requestBody = record(input.body); + + try { + const route = requireChatGptSessionRoute(input.model); + + const connectionId = input.credentials.connectionId?.trim(); + const encodedCredentials = input.credentials.apiKey?.trim(); + if (!connectionId || !encodedCredentials) { + throw new Error("ChatGPT browser credentials are missing"); + } + const secrets = decodeChatGptWebCodexSecrets(encodedCredentials); + + const providerData = record(input.credentials.providerSpecificData); + const cdpEndpoint = + configuredString(providerData, "browserCdpEndpoint") ?? + process.env.CHATGPT_WEB_CODEX_CDP_URL?.trim(); + const chromeExecutablePath = runtime.detectChrome( + configuredString(providerData, "chromeExecutablePath") + ); + if (!chromeExecutablePath && !cdpEndpoint) { + throw new Error("No supported Chrome or Chromium executable was found"); + } + + const storageStatePath = runtime.ensureStorageState(connectionId, secrets); + const connectorName = + configuredString(providerData, "connectorName", "appName") ?? + CHATGPT_WEB_CODEX_CONNECTOR_NAME; + + const loginConfig: ChatGptSessionLoginConfig = { + appName: connectorName, + storageStatePath, + headed: true, + proAvailable: providerData.proAvailable === true, + autoApproveToolCalls: false, + ...(chromeExecutablePath ? { chromeExecutablePath } : {}), + ...(cdpEndpoint ? { cdpEndpoint } : {}), + }; + + let solAvailable = providerData.solAvailable !== false; + let proAvailable = providerData.proAvailable === true; + if (!runtime.loginStateExists(loginConfig)) { + const capabilities = await runtime.inspectLogin(loginConfig); + solAvailable = capabilities.solAvailable; + proAvailable = capabilities.proAvailable; + await input.onCredentialsRefreshed?.({ + providerSpecificData: { + ...providerData, + solAvailable, + proAvailable, + browserVerified: true, + ...(chromeExecutablePath ? { chromeExecutablePath } : {}), + ...(cdpEndpoint ? { browserCdpEndpoint: cdpEndpoint } : {}), + }, + }); + } + + if (route.sol !== solAvailable) { + throw new Error( + route.sol + ? `${route.id} is not available for this Luna-only connection` + : `${route.id} is not available while the account exposes the Sol model selector` + ); + } + if (route.pro && !proAvailable) { + throw new Error(`${route.id} is not available for this non-Pro connection`); + } + + const messages = Array.isArray(requestBody.messages) + ? (requestBody.messages as Array<{ role: string; content: unknown }>) + : []; + const { hasTools, requestedTools, effectiveMessages } = prepareToolMessages( + requestBody, + messages + ); + + const parsed = buildParsedRequest({ + route, + messages: effectiveMessages, + stream: Boolean(input.stream) && !hasTools, + rawBody: input.body, + }); + + const provider = buildChatGptSessionProviderConfig({ + route, + connectionId, + storageStatePath, + connectorName, + solAvailable, + proAvailable, + ...(chromeExecutablePath ? { chromeExecutablePath } : {}), + ...(cdpEndpoint ? { cdpEndpoint } : {}), + }); + + const events = new AsyncEventQueue(); + const incoming = { + headers: new Headers(), + ...(input.signal ? { abortSignal: input.signal } : {}), + }; + + const persistRotatedState = async () => { + try { + const storageState = runtime.readStorageState(storageStatePath); + await input.onCredentialsRefreshed?.({ + apiKey: encodeChatGptWebCodexSecrets({ + storageState, + ...(secrets.runtimeKey ? { runtimeKey: secrets.runtimeKey } : {}), + }), + }); + } catch (refreshError) { + input.log?.warn?.( + "CHATGPT_SESSION", + sanitizeErrorMessage( + refreshError instanceof Error ? refreshError.message : refreshError + ) + ); + } + }; + + const run = async () => { + try { + await runtime.runTurn(parsed, incoming, (event) => events.push(event), provider); + } catch (error) { + const classified = classifyChatGptSessionError(error); + events.push({ + type: "error", + message: sanitizeErrorMessage(error instanceof Error ? error.message : error), + status: classified.status, + code: classified.code, + }); + } finally { + await persistRotatedState(); + events.close(); + } + }; + + const meta: ChatGptSessionResponseMeta = { + cid: `chatcmpl-cgpts-${crypto.randomUUID().slice(0, 12)}`, + created: Math.floor(Date.now() / 1000), + model: input.model, + }; + + if (!parsed.stream) { + const running = run(); + const collected = await events.collect(); + await running; + const built = buildChatGptSessionCompletion(collected, meta); + const jsonResponse = new Response(JSON.stringify(built.body), { + status: built.status, + headers: JSON_HEADERS, + }); + if (!hasTools || built.status !== 200) return wrapped(jsonResponse, input.body); + const toolResponse = await buildToolModeResponse( + jsonResponse, + requestedTools, + Boolean(input.stream), + { cid: meta.cid, created: meta.created, model: meta.model, idSeed: "cgpts" } + ); + return wrapped(toolResponse, input.body); + } + + void run(); + const opened = await openChatGptSessionStream(events, meta); + if (opened.kind === "error") { + // Re-classify from the raw message so a message-derived hint (e.g. the + // connection-cooldown hint on a browser failure) is not lost. + const classified = classifyChatGptSessionError(new Error(opened.message)); + return wrapped( + errorResponse( + classified.status, + opened.message, + classified.code, + classified.fallbackHint + ), + input.body + ); + } + return wrapped( + new Response(opened.stream, { status: 200, headers: SSE_HEADERS }), + input.body + ); + } catch (error) { + const classified = classifyChatGptSessionError(error); + input.log?.warn?.( + "CHATGPT_SESSION", + sanitizeErrorMessage(error instanceof Error ? error.message : error) + ); + return wrapped( + errorResponse( + classified.status, + error instanceof Error ? error.message : error, + classified.code, + classified.fallbackHint + ), + input.body + ); + } + } +} diff --git a/open-sse/executors/chatgpt-session/runtime.ts b/open-sse/executors/chatgpt-session/runtime.ts new file mode 100644 index 0000000000..fc107d8f39 --- /dev/null +++ b/open-sse/executors/chatgpt-session/runtime.ts @@ -0,0 +1,73 @@ +/** + * Indirection layer over every side-effecting dependency of the executor (browser detection, + * storage-state IO, login probe, adapter turn). Tests swap the whole record so no unit test + * ever launches Chrome; production resolves to the vendored implementations. + */ + +import { createChatGptWebAdapter } from "../../vendor/codex-chatgpt-web/adapters/chatgpt-web/index.ts"; +import { + browserLoginStateExists, + inspectBrowserLoginCapabilities, +} from "../../vendor/codex-chatgpt-web/browser-login.ts"; +import type { + AdapterEvent, + CodexParsedRequest, + CodexProviderConfig, +} from "../../vendor/codex-chatgpt-web/types.ts"; +import { detectChromeExecutable } from "../chatgpt-web-codex.ts"; +import { + ensureConnectionStorageStateFromCredential, + readConnectionStorageState, +} from "../chatgpt-web-codex/storageState.ts"; + +export interface ChatGptSessionLoginConfig { + appName: string; + storageStatePath: string; + headed: boolean; + proAvailable: boolean; + autoApproveToolCalls: boolean; + chromeExecutablePath?: string; + cdpEndpoint?: string; +} + +export interface ChatGptSessionRuntime { + detectChrome(explicit?: string): string | undefined; + ensureStorageState( + connectionId: string, + credential: { cookie?: string; storageState?: Record } + ): string; + readStorageState(path: string): Record; + loginStateExists(config: ChatGptSessionLoginConfig): boolean; + inspectLogin( + config: ChatGptSessionLoginConfig + ): Promise<{ solAvailable: boolean; proAvailable: boolean }>; + runTurn( + parsed: CodexParsedRequest, + incoming: { headers: Headers; abortSignal?: AbortSignal }, + emit: (event: AdapterEvent) => void, + provider: CodexProviderConfig + ): Promise; +} + +const productionRuntime: ChatGptSessionRuntime = { + detectChrome: (explicit) => detectChromeExecutable(explicit), + ensureStorageState: (connectionId, credential) => + ensureConnectionStorageStateFromCredential(connectionId, credential), + readStorageState: (path) => readConnectionStorageState(path), + loginStateExists: (config) => browserLoginStateExists(config), + inspectLogin: (config) => inspectBrowserLoginCapabilities(config), + runTurn: (parsed, incoming, emit, provider) => + createChatGptWebAdapter(provider).runTurn(parsed, incoming, emit), +}; + +let override: Partial | null = null; + +export function __setChatGptSessionRuntimeForTesting( + next: Partial | null +): void { + override = next; +} + +export function chatGptSessionRuntime(): ChatGptSessionRuntime { + return override ? { ...productionRuntime, ...override } : productionRuntime; +} diff --git a/tests/unit/chatgpt-session-executor.test.ts b/tests/unit/chatgpt-session-executor.test.ts new file mode 100644 index 0000000000..90d9b59cf5 --- /dev/null +++ b/tests/unit/chatgpt-session-executor.test.ts @@ -0,0 +1,194 @@ +import { test } from "node:test"; +import assert from "node:assert/strict"; + +import { ChatGptSessionExecutor } from "../../open-sse/executors/chatgpt-session.ts"; +import { __setChatGptSessionRuntimeForTesting } from "../../open-sse/executors/chatgpt-session/runtime.ts"; +import type { AdapterEvent } from "../../open-sse/vendor/codex-chatgpt-web/types.ts"; + +const CREDENTIALS = { + connectionId: "conn-1", + apiKey: JSON.stringify({ version: 2, storageState: { cookies: [], origins: [] } }), + providerSpecificData: { solAvailable: true, proAvailable: false, browserVerified: true }, +}; + +function stubRuntime(events: AdapterEvent[], overrides: Record = {}) { + __setChatGptSessionRuntimeForTesting({ + detectChrome: () => "/usr/bin/chromium", + ensureStorageState: () => "/tmp/state.json", + readStorageState: () => ({ cookies: [], origins: [] }), + loginStateExists: () => true, + inspectLogin: async () => ({ solAvailable: true, proAvailable: false }), + runTurn: async (_parsed, _incoming, emit) => { + for (const event of events) emit(event); + }, + ...overrides, + }); +} + +function body(extra: Record = {}) { + return { messages: [{ role: "user", content: "Hi" }], ...extra }; +} + +test.afterEach(() => { + __setChatGptSessionRuntimeForTesting(null); +}); + +test("returns a streaming completion for a normal turn", async () => { + stubRuntime([{ type: "text_delta", text: "Hello" }, { type: "done" }]); + const result = await new ChatGptSessionExecutor().execute({ + model: "high", + body: body(), + stream: true, + credentials: CREDENTIALS, + }); + const response = "response" in result ? result.response : result; + assert.equal(response.status, 200); + assert.equal(response.headers.get("Content-Type"), "text/event-stream; charset=utf-8"); + const text = await response.text(); + assert.match(text, /"content":"Hello"/); +}); + +test("returns a buffered completion when stream is false", async () => { + stubRuntime([{ type: "text_delta", text: "Hello" }, { type: "done" }]); + const result = await new ChatGptSessionExecutor().execute({ + model: "high", + body: body(), + stream: false, + credentials: CREDENTIALS, + }); + const response = "response" in result ? result.response : result; + const json = (await response.json()) as Record; + assert.equal(json.object, "chat.completion"); + assert.equal( + ((json.choices as Array>)[0].message as Record) + .content, + "Hello" + ); +}); + +test("answers 503 with a cooldown hint when no browser is installed", async () => { + stubRuntime([], { detectChrome: () => undefined }); + const result = await new ChatGptSessionExecutor().execute({ + model: "high", + body: body(), + stream: true, + credentials: CREDENTIALS, + }); + const response = "response" in result ? result.response : result; + assert.equal(response.status, 503); + assert.equal(response.headers.get("X-Omni-Fallback-Hint"), "connection_cooldown"); +}); + +test("answers 401 when the connection has no credentials", async () => { + stubRuntime([]); + const result = await new ChatGptSessionExecutor().execute({ + model: "high", + body: body(), + stream: true, + credentials: { connectionId: "conn-1" }, + }); + const response = "response" in result ? result.response : result; + assert.equal(response.status, 401); +}); + +test("rejects a Pro route on a non-Pro account without touching the browser", async () => { + let ran = false; + stubRuntime([], { + runTurn: async () => { + ran = true; + }, + }); + const result = await new ChatGptSessionExecutor().execute({ + model: "pro", + body: body(), + stream: true, + credentials: CREDENTIALS, + }); + const response = "response" in result ? result.response : result; + assert.equal(response.status, 400); + assert.equal(ran, false); +}); + +test("an expired session before output becomes a 401, not a 200 stream", async () => { + stubRuntime([{ type: "error", message: "ChatGPT page is not authenticated" }]); + const result = await new ChatGptSessionExecutor().execute({ + model: "high", + body: body(), + stream: true, + credentials: CREDENTIALS, + }); + const response = "response" in result ? result.response : result; + assert.equal(response.status, 401); + const json = (await response.json()) as { error: { message: string } }; + assert.doesNotMatch(json.error.message, /at \//); +}); + +test("persists rotated storage state through onCredentialsRefreshed", async () => { + stubRuntime([{ type: "text_delta", text: "x" }, { type: "done" }], { + readStorageState: () => ({ cookies: [{ name: "rotated" }], origins: [] }), + }); + const patches: Array> = []; + const result = await new ChatGptSessionExecutor().execute({ + model: "high", + body: body(), + stream: false, + credentials: CREDENTIALS, + onCredentialsRefreshed: async (patch) => { + patches.push(patch); + }, + }); + const response = "response" in result ? result.response : result; + await response.text(); + const persisted = patches.find((patch) => typeof patch.apiKey === "string"); + assert.ok(persisted, "expected an apiKey patch"); + assert.match(String(persisted.apiKey), /rotated/); +}); + +test("emulated tool calls go through the shared web-tools contract", async () => { + stubRuntime([ + { type: "text_delta", text: '{"name": "get_time", "arguments": {}}' }, + { type: "done" }, + ]); + const result = await new ChatGptSessionExecutor().execute({ + model: "high", + body: body({ + tools: [ + { + type: "function", + function: { name: "get_time", description: "time", parameters: { type: "object" } }, + }, + ], + }), + stream: false, + credentials: CREDENTIALS, + }); + const response = "response" in result ? result.response : result; + const json = (await response.json()) as Record; + const choice = (json.choices as Array>)[0]; + assert.equal(choice.finish_reason, "tool_calls"); +}); + +test("forwards the abort signal to the adapter", async () => { + let seenSignal: AbortSignal | undefined; + stubRuntime([{ type: "done" }], { + runTurn: async ( + _parsed: unknown, + incoming: { abortSignal?: AbortSignal }, + emit: (e: AdapterEvent) => void + ) => { + seenSignal = incoming.abortSignal; + emit({ type: "done" }); + }, + }); + const controller = new AbortController(); + const result = await new ChatGptSessionExecutor().execute({ + model: "high", + body: body(), + stream: false, + credentials: CREDENTIALS, + signal: controller.signal, + }); + const response = "response" in result ? result.response : result; + await response.text(); + assert.equal(seenSignal, controller.signal); +});