Log Responses WebSocket calls in history (#3616)

Integrated into release/v3.8.22 — Codex Responses WebSocket call history logging.

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;

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");
 }

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);