diff --git a/scripts/dev/responses-ws-proxy.mjs b/scripts/dev/responses-ws-proxy.mjs index 0a8201dc7a..abbb435e8f 100644 --- a/scripts/dev/responses-ws-proxy.mjs +++ b/scripts/dev/responses-ws-proxy.mjs @@ -50,6 +50,10 @@ function isText(value) { return typeof value === "string" && value.length > 0; } +function isRecord(value) { + return value && typeof value === "object" && !Array.isArray(value); +} + function jsonStringifySafe(value) { try { return JSON.stringify(value); @@ -67,6 +71,92 @@ function jsonStringifySafe(value) { } } +function parseJsonRecord(value) { + try { + const parsed = typeof value === "string" ? JSON.parse(value) : value; + return isRecord(parsed) ? parsed : null; + } catch { + return null; + } +} + +function toFiniteNumber(value) { + const parsed = Number(value); + return Number.isFinite(parsed) ? parsed : 0; +} + +function toStringOrNull(value) { + return typeof value === "string" && value.trim().length > 0 ? value.trim() : null; +} + +function buildFailurePayload(code, message) { + return { + type: "response.failed", + response: { + id: null, + status: "failed", + error: { + code, + message, + }, + }, + }; +} + +function getResponseErrorStatus(error) { + if (!isRecord(error)) return null; + const candidates = [ + error.status, + error.status_code, + error.statusCode, + error.code === "usage_limit_reached" ? 429 : null, + ]; + for (const candidate of candidates) { + const status = Number(candidate); + if (Number.isInteger(status) && status >= 400 && status <= 599) return status; + } + return null; +} + +function getTerminalResponseEvent(rawData) { + const message = parseJsonRecord(rawData); + if (!message) return null; + + const type = toStringOrNull(message.type) || ""; + const response = isRecord(message.response) ? message.response : {}; + const statusText = toStringOrNull(response.status) || ""; + + if (type === "response.completed" || type === "response.done" || statusText === "completed") { + return { + status: 200, + success: true, + terminalMessage: message, + responseBody: response, + }; + } + + if (type === "response.failed" || type === "response.error" || statusText === "failed") { + const error = isRecord(response.error) + ? response.error + : isRecord(message.error) + ? message.error + : null; + return { + status: getResponseErrorStatus(error) || 500, + success: false, + errorCode: toStringOrNull(error?.code) || "upstream_response_failed", + errorMessage: + toStringOrNull(error?.message) || + toStringOrNull(response.error) || + "Codex upstream response failed", + terminalMessage: message, + responseBody: response, + }; + } + + return null; +} + export function isResponsesWsPath(pathname) { return RESPONSES_WS_PUBLIC_PATHS.has(pathname); } @@ -314,6 +404,7 @@ class ResponsesWsSession { this.maxBufferBytes = normalizePositiveInteger(maxBufferBytes, DEFAULT_MAX_WS_BUFFER_BYTES); this.maxMessageBytes = normalizePositiveInteger(maxMessageBytes, DEFAULT_MAX_WS_MESSAGE_BYTES); this.sessionId = randomUUID(); + this.startedAt = Date.now(); this.closed = false; this.buffer = Buffer.alloc(0); this.fragmentOpcode = null; @@ -322,6 +413,9 @@ class ResponsesWsSession { this.processing = Promise.resolve(); this.upstream = null; this.upstreamReady = null; + this.firstResponseBody = null; + this.preparedContext = null; + this.historyLogged = false; this.lastSeenAt = Date.now(); this.pingTimer = setInterval(() => { @@ -376,17 +470,9 @@ class ResponsesWsSession { } sendFailure(code, message) { - this.sendJson({ - type: "response.failed", - response: { - id: null, - status: "failed", - error: { - code, - message, - }, - }, - }); + const payload = buildFailurePayload(code, message); + this.sendJson(payload); + return payload; } async onData(chunk) { @@ -490,6 +576,7 @@ class ResponsesWsSession { if (responseBody === null) { throw new Error("First Responses WebSocket message must be response.create"); } + this.firstResponseBody ||= responseBody; const prepared = await callInternal( this.fetchImpl, @@ -513,9 +600,21 @@ class ResponsesWsSession { const code = prepared.json?.error?.code || "codex_ws_prepare_failed"; const error = new Error(message); error.code = code; + error.status = prepared.status; throw error; } + this.preparedContext = { + upstreamUrl: toStringOrNull(prepared.json?.upstreamUrl), + connectionId: toStringOrNull(prepared.json?.connectionId), + account: toStringOrNull(prepared.json?.account), + provider: toStringOrNull(prepared.json?.provider) || "codex", + model: toStringOrNull(prepared.json?.model) || toStringOrNull(responseBody.model), + requestedModel: toStringOrNull(responseBody.model), + serviceTier: + toStringOrNull(responseBody.service_tier) || toStringOrNull(responseBody.serviceTier), + }; + const upstream = await this.wsFactory(prepared.json.upstreamUrl, { browser: prepared.json.browser || "chrome_142", os: prepared.json.os || "windows", @@ -526,17 +625,36 @@ class ResponsesWsSession { if (this.closed) return; const data = typeof event.data === "string" ? event.data : Buffer.from(event.data).toString("utf8"); + const terminalEvent = getTerminalResponseEvent(data); + if (terminalEvent) { + void this.persistHistory(terminalEvent); + } this.sendFrame(0x1, Buffer.from(data, "utf8")); }; upstream.onerror = (event) => { if (this.closed) return; - this.sendFailure( - "upstream_websocket_error", - event.message || "Codex upstream WebSocket error" - ); + const errorMessage = event.message || "Codex upstream WebSocket error"; + const failurePayload = this.sendFailure("upstream_websocket_error", errorMessage); + void this.persistHistory({ + status: 502, + success: false, + errorCode: "upstream_websocket_error", + errorMessage, + terminalMessage: failurePayload, + }); }; upstream.onclose = (event) => { if (this.closed) return; + void this.persistHistory({ + status: event.code === 1000 ? 499 : 502, + success: false, + errorCode: "upstream_websocket_closed", + errorMessage: event.reason || "Codex upstream WebSocket closed before completion", + terminalMessage: buildFailurePayload( + "upstream_websocket_closed", + event.reason || "Codex upstream WebSocket closed before completion" + ), + }); this.close(event.code || 1000, event.reason || "upstream_closed"); }; @@ -561,11 +679,56 @@ class ResponsesWsSession { } catch (error) { const code = error?.code || "upstream_websocket_connect_failed"; const messageText = error instanceof Error ? error.message : String(error); - this.sendFailure(code, messageText); + const failurePayload = this.sendFailure(code, messageText); + void this.persistHistory({ + status: Number.isInteger(error?.status) ? error.status : 502, + success: false, + errorCode: code, + errorMessage: messageText, + terminalMessage: failurePayload, + }); this.close(1011, "upstream_connect_failed"); } } + async persistHistory({ + status = 200, + success = true, + errorCode = null, + errorMessage = null, + terminalMessage = null, + responseBody = null, + } = {}) { + if (this.historyLogged || !this.firstResponseBody) return; + this.historyLogged = true; + + const finishedAt = Date.now(); + try { + await callInternal(this.fetchImpl, this.baseUrl, this.bridgeSecret, "log", { + sessionId: this.sessionId, + transport: "responses_websocket", + requestUrl: this.requestUrl, + headers: getAuthHeaders(this.requestUrl, this.requestHeaders), + path: new URL(this.requestUrl || "/v1/responses", "http://omniroute.local").pathname, + startedAt: new Date(this.startedAt).toISOString(), + completedAt: new Date(finishedAt).toISOString(), + durationMs: Math.max(0, finishedAt - this.startedAt), + status: toFiniteNumber(status), + success, + errorCode, + errorMessage, + clientRequest: this.firstResponseBody, + terminalMessage, + responseBody, + sourceFormat: "openai-responses", + targetFormat: "openai-responses", + ...this.preparedContext, + }); + } catch { + // History logging must never break an already-established WebSocket session. + } + } + close(code = 1000, reason = "normal_closure") { if (this.closed) return; this.closed = true; diff --git a/src/app/api/internal/codex-responses-ws/route.ts b/src/app/api/internal/codex-responses-ws/route.ts index d5b2cb715b..5bc72d44ca 100644 --- a/src/app/api/internal/codex-responses-ws/route.ts +++ b/src/app/api/internal/codex-responses-ws/route.ts @@ -9,6 +9,7 @@ import { getProviderCredentialsWithQuotaPreflight } from "@/sse/services/auth"; import { checkAndRefreshToken } from "@/sse/services/tokenRefresh"; import { resolveCodexWsModelInfo } from "./modelResolution"; import { isFeatureFlagEnabled } from "@/shared/utils/featureFlags"; +import { sanitizeErrorMessage } from "@omniroute/open-sse/utils/error.ts"; const CODEX_RESPONSES_WS_URL = "wss://chatgpt.com/backend-api/codex/responses"; const executor = new CodexExecutor(); @@ -28,6 +29,74 @@ function isRecord(value: unknown): value is JsonRecord { return typeof value === "object" && value !== null && !Array.isArray(value); } +function toStringOrNull(value: unknown): string | null { + return typeof value === "string" && value.trim().length > 0 ? value.trim() : null; +} + +function toFiniteNumber(value: unknown, fallback = 0): number { + const parsed = Number(value); + return Number.isFinite(parsed) ? parsed : fallback; +} + +function toHttpStatus(value: unknown, fallback: number): number { + const status = Number(value); + return Number.isInteger(status) && status >= 100 && status <= 599 ? status : fallback; +} + +function getResponseCreateBody(body: JsonRecord): JsonRecord { + if (isRecord(body.clientRequest)) return body.clientRequest; + if (isRecord(body.response)) return body.response; + return {}; +} + +function getTerminalMessage(body: JsonRecord): JsonRecord | null { + return isRecord(body.terminalMessage) ? body.terminalMessage : null; +} + +function getTerminalResponseBody(body: JsonRecord): JsonRecord | null { + if (isRecord(body.responseBody)) return body.responseBody; + const terminalMessage = getTerminalMessage(body); + if (isRecord(terminalMessage?.response)) return terminalMessage.response; + return terminalMessage; +} + +function getErrorRecord(body: JsonRecord, responseBody: JsonRecord | null): JsonRecord | null { + if (isRecord(body.error)) return body.error; + if (isRecord(responseBody?.error)) return responseBody.error; + const terminalMessage = getTerminalMessage(body); + if (isRecord(terminalMessage?.error)) return terminalMessage.error; + return null; +} + +function getTimestamp(value: unknown): string { + const raw = toStringOrNull(value); + if (!raw) return new Date().toISOString(); + const parsed = new Date(raw); + return Number.isNaN(parsed.getTime()) ? new Date().toISOString() : parsed.toISOString(); +} + +function getRequestPath(body: JsonRecord): string { + const explicitPath = toStringOrNull(body.path); + if (explicitPath) return explicitPath; + + try { + const requestUrl = toStringOrNull(body.requestUrl) || "/v1/responses"; + return new URL(requestUrl, "http://omniroute.local").pathname; + } catch { + return "/v1/responses"; + } +} + +function getServiceTier(requestBody: JsonRecord): string | null { + return toStringOrNull(requestBody.service_tier) || toStringOrNull(requestBody.serviceTier); +} + +async function getApiKeyMetadataFromBody(body: JsonRecord) { + const authRequest = getAuthRequest(body); + const apiKey = extractWsTokenFromRequest(authRequest); + return apiKey ? getApiKeyMetadata(apiKey).catch(() => null) : null; +} + function getBridgeSecret(): string { return process.env.OMNIROUTE_WS_BRIDGE_SECRET || ""; } @@ -184,12 +253,121 @@ async function prepare(body: JsonRecord) { browser: "chrome_142", os: "windows", connectionId: refreshedCredentials.connectionId, + provider, + account: refreshedCredentials.email || null, model, headers, response: transformed, }); } +async function persistResponsesWsCallHistory(body: JsonRecord) { + const [{ saveCallLog }, { saveRequestUsage }, { logProxyEvent }] = await Promise.all([ + import("@/lib/usage/callLogs"), + import("@/lib/usage/usageHistory"), + import("@/lib/proxyLogger"), + ]); + + const metadata = await getApiKeyMetadataFromBody(body); + const requestBody = getResponseCreateBody(body); + const terminalMessage = getTerminalMessage(body); + const responseBody = getTerminalResponseBody(body); + const usage = isRecord(responseBody?.usage) ? responseBody.usage : {}; + const errorRecord = getErrorRecord(body, responseBody); + const status = toHttpStatus( + body.status ?? errorRecord?.status_code ?? errorRecord?.status, + body.success === false ? 500 : 200 + ); + const success = typeof body.success === "boolean" ? body.success : status < 400; + const errorCode = + toStringOrNull(body.errorCode) || + toStringOrNull(errorRecord?.code) || + (success ? null : "responses_websocket_failed"); + const errorMessage = success + ? null + : sanitizeErrorMessage( + toStringOrNull(body.errorMessage) || + toStringOrNull(errorRecord?.message) || + "Responses WebSocket request failed" + ); + const timestamp = getTimestamp(body.startedAt); + const durationMs = Math.max(0, Math.round(toFiniteNumber(body.durationMs, 0))); + const provider = toStringOrNull(body.provider) || "codex"; + const model = + toStringOrNull(body.model) || + toStringOrNull(responseBody?.model) || + toStringOrNull(requestBody.model) || + "-"; + const requestedModel = toStringOrNull(body.requestedModel) || toStringOrNull(requestBody.model); + const connectionId = toStringOrNull(body.connectionId); + const apiKeyId = metadata?.id || null; + const apiKeyName = metadata?.name || null; + const noLog = metadata?.noLog === true; + const path = getRequestPath(body); + const sourceFormat = toStringOrNull(body.sourceFormat) || "openai-responses"; + const targetFormat = toStringOrNull(body.targetFormat) || "openai-responses"; + const targetUrl = toStringOrNull(body.upstreamUrl) || CODEX_RESPONSES_WS_URL; + const account = toStringOrNull(body.account); + + await saveCallLog({ + id: toStringOrNull(body.sessionId) || undefined, + timestamp, + method: "WEBSOCKET", + path, + status, + model, + requestedModel, + provider, + connectionId, + duration: durationMs, + tokens: usage, + requestType: "responses_websocket", + sourceFormat, + targetFormat, + apiKeyId, + apiKeyName, + noLog, + requestBody, + responseBody: responseBody ?? terminalMessage, + error: errorMessage ? { code: errorCode, message: errorMessage } : null, + pipelinePayloads: { + clientRequest: requestBody, + providerRequest: requestBody, + providerResponse: responseBody, + clientResponse: terminalMessage, + }, + }); + + await saveRequestUsage({ + timestamp, + provider, + model, + connectionId, + apiKeyId, + apiKeyName, + tokens: usage, + serviceTier: getServiceTier(requestBody), + status: String(status), + success, + latencyMs: durationMs, + timeToFirstTokenMs: durationMs, + errorCode, + }); + + logProxyEvent({ + status: success ? "success" : "error", + level: "direct", + provider, + targetUrl, + latencyMs: durationMs, + error: errorMessage, + connectionId, + account, + }); + + return NextResponse.json({ ok: true, logged: true }); +} + export async function POST(request: Request) { const expectedSecret = getBridgeSecret(); const receivedSecret = request.headers.get("x-omniroute-ws-bridge-secret") || ""; @@ -215,6 +393,17 @@ export async function POST(request: Request) { if (action === "prepare") { return prepare(body); } + if (action === "log") { + try { + return await persistResponsesWsCallHistory(body); + } catch (error) { + return jsonError( + 500, + "responses_ws_history_log_failed", + sanitizeErrorMessage(error instanceof Error ? error.message : String(error)) + ); + } + } return jsonError(400, "invalid_action", "Unsupported bridge action"); } diff --git a/tests/unit/responses-ws-proxy.test.mjs b/tests/unit/responses-ws-proxy.test.mjs index 382ced9b9c..e9de940650 100644 --- a/tests/unit/responses-ws-proxy.test.mjs +++ b/tests/unit/responses-ws-proxy.test.mjs @@ -75,6 +75,10 @@ test("responses ws proxy prepares and forwards OpenAI Responses websocket events ok: true, upstreamUrl: "wss://chatgpt.com/backend-api/codex/responses", headers: { Authorization: "Bearer upstream-token" }, + connectionId: "conn_1", + provider: "codex", + account: "codex@example.com", + model: "gpt-5.5", response: { ...body.response, model: "gpt-5.5", @@ -84,6 +88,12 @@ test("responses ws proxy prepares and forwards OpenAI Responses websocket events ); return; } + + if (body.action === "log") { + res.writeHead(200, { "content-type": "application/json" }); + res.end(JSON.stringify({ ok: true, logged: true })); + return; + } } res.writeHead(404, { "content-type": "application/json" }); @@ -97,7 +107,12 @@ test("responses ws proxy prepares and forwards OpenAI Responses websocket events fakeUpstream.onmessage?.({ data: JSON.stringify({ type: "response.completed", - response: { id: "resp_1", status: "completed" }, + response: { + id: "resp_1", + model: "gpt-5.5", + status: "completed", + usage: { input_tokens: 29, output_tokens: 42, total_tokens: 71 }, + }, }), }); }, 10); @@ -145,6 +160,7 @@ test("responses ws proxy prepares and forwards OpenAI Responses websocket events ); await waitFor(() => downstreamMessages.find((entry) => entry.type === "response.completed")); + const logRequest = await waitFor(() => internalRequests.find((entry) => entry.action === "log")); assert.equal(internalRequests[0].action, "authenticate"); assert.equal(internalRequests[1].action, "prepare"); @@ -153,6 +169,104 @@ test("responses ws proxy prepares and forwards OpenAI Responses websocket events assert.equal(upstreamSends[0].model, "gpt-5.5"); assert.equal(upstreamSends[0].reasoning.effort, "xhigh"); assert.equal("stream" in upstreamSends[0], false); + assert.equal(logRequest.transport, "responses_websocket"); + assert.equal(logRequest.status, 200); + assert.equal(logRequest.success, true); + assert.equal(logRequest.connectionId, "conn_1"); + assert.equal(logRequest.provider, "codex"); + assert.equal(logRequest.model, "gpt-5.5"); + assert.equal(logRequest.requestedModel, "gpt-5.5"); + assert.equal(logRequest.clientRequest.model, "gpt-5.5"); + assert.equal(logRequest.responseBody.usage.input_tokens, 29); + assert.equal(logRequest.responseBody.usage.output_tokens, 42); + assert.equal(logRequest.terminalMessage.type, "response.completed"); + + ws.close(); + await close(server); +}); + +test("responses ws proxy logs prepare failures to request history", async () => { + const internalRequests = []; + const downstreamMessages = []; + + const server = http.createServer(async (req, res) => { + const url = new URL(req.url || "/", `http://${req.headers.host}`); + if (url.pathname === "/api/internal/codex-responses-ws") { + const body = JSON.parse((await readRequestBody(req)) || "{}"); + internalRequests.push(body); + + if (body.action === "authenticate") { + res.writeHead(200, { "content-type": "application/json" }); + res.end(JSON.stringify({ ok: true, authenticated: true, authType: "api_key" })); + return; + } + + if (body.action === "prepare") { + res.writeHead(503, { "content-type": "application/json" }); + res.end( + JSON.stringify({ + error: { + code: "codex_credentials_unavailable", + message: "No available Codex OAuth connection for Responses WebSocket", + }, + }) + ); + return; + } + + if (body.action === "log") { + res.writeHead(200, { "content-type": "application/json" }); + res.end(JSON.stringify({ ok: true, logged: true })); + return; + } + } + + res.writeHead(404, { "content-type": "application/json" }); + res.end(JSON.stringify({ error: "not_found" })); + }); + + const port = await listen(server); + const proxy = createResponsesWsProxy({ + baseUrl: `http://127.0.0.1:${port}`, + bridgeSecret: "bridge-secret", + pingIntervalMs: 1000, + idleTimeoutMs: 10000, + wsFactory: async () => { + throw new Error("prepare failure should not connect upstream"); + }, + }); + + server.on("upgrade", async (req, socket, head) => { + const handled = await proxy.handleUpgrade(req, socket, head); + if (!handled && !socket.destroyed) { + socket.destroy(); + } + }); + + const ws = new WebSocket(`ws://127.0.0.1:${port}/api/v1/responses?api_key=local-token`); + ws.addEventListener("message", (event) => { + downstreamMessages.push(JSON.parse(String(event.data))); + }); + + await new Promise((resolve) => ws.addEventListener("open", resolve, { once: true })); + ws.send( + JSON.stringify({ + type: "response.create", + model: "gpt-5.5", + input: [{ role: "user", content: "hello" }], + }) + ); + + await waitFor(() => downstreamMessages.find((entry) => entry.type === "response.failed")); + const logRequest = await waitFor(() => internalRequests.find((entry) => entry.action === "log")); + + assert.equal(logRequest.transport, "responses_websocket"); + assert.equal(logRequest.status, 503); + assert.equal(logRequest.success, false); + assert.equal(logRequest.errorCode, "codex_credentials_unavailable"); + assert.match(logRequest.errorMessage, /No available Codex OAuth connection/); + assert.equal(logRequest.clientRequest.model, "gpt-5.5"); + assert.equal(logRequest.terminalMessage.type, "response.failed"); ws.close(); await close(server);