diff --git a/changelog.d/fixes/7388-codex-ws-history-per-turn.md b/changelog.d/fixes/7388-codex-ws-history-per-turn.md new file mode 100644 index 0000000000..2e6e23a280 --- /dev/null +++ b/changelog.d/fixes/7388-codex-ws-history-per-turn.md @@ -0,0 +1 @@ +- fix(cli): log Codex Responses WebSocket history/usage per logical turn instead of once per connection (#7388) diff --git a/scripts/dev/responses-ws-proxy.mjs b/scripts/dev/responses-ws-proxy.mjs index 411cb78828..1d66c57cb1 100644 --- a/scripts/dev/responses-ws-proxy.mjs +++ b/scripts/dev/responses-ws-proxy.mjs @@ -31,6 +31,9 @@ const WS_QUERY_TOKEN_KEYS = ["api_key", "token", "access_token"]; const textDecoder = new TextDecoder(); const DEFAULT_MAX_WS_BUFFER_BYTES = 16 * 1024 * 1024; const DEFAULT_MAX_WS_MESSAGE_BYTES = 16 * 1024 * 1024; +// #7388: sentinel turn key for session-ending terminal events that don't carry +// a `response.id` (prepare failure, upstream error/close, connect failure). +const SESSION_TERMINAL_TURN_KEY = "__session_terminal__"; class WebSocketInputTooLargeError extends Error { constructor(message, reason = "message_too_large") { @@ -414,8 +417,16 @@ class ResponsesWsSession { this.upstream = null; this.upstreamReady = null; this.firstResponseBody = null; + this.currentRequestBody = null; this.preparedContext = null; - this.historyLogged = false; + // #7388: logging must be scoped per logical turn (one `response.create` + // through its terminal event), not once for the lifetime of the WS + // connection — a single boolean here silently dropped every turn after + // the first on a reused connection. Terminal events carry a + // `response.id` we can key on; session-ending failure paths (prepare + // failure, upstream error/close, connect failure) don't, so they fall + // back to a session-scoped sentinel key that still logs exactly once. + this.loggedTurnIds = new Set(); this.lastSeenAt = Date.now(); this.pingTimer = setInterval(() => { @@ -577,6 +588,7 @@ class ResponsesWsSession { throw new Error("First Responses WebSocket message must be response.create"); } this.firstResponseBody ||= responseBody; + this.currentRequestBody = responseBody; const prepared = await callInternal( this.fetchImpl, @@ -681,6 +693,12 @@ class ResponsesWsSession { upstream.send(jsonStringifySafe(firstMessage)); return; } + // #7388: a reused WS connection forwards subsequent response.create + // turns straight through (ensureUpstream() only runs once); track each + // turn's own request body so persistHistory() attaches the right + // clientRequest instead of always the first turn's. + const nextTurnBody = getResponseCreatePayload(message); + if (nextTurnBody !== null) this.currentRequestBody = nextTurnBody; this.upstream.send(jsonStringifySafe(message)); } catch (error) { const code = error?.code || "upstream_websocket_connect_failed"; @@ -705,8 +723,17 @@ class ResponsesWsSession { terminalMessage = null, responseBody = null, } = {}) { - if (this.historyLogged || !this.firstResponseBody) return; - this.historyLogged = true; + if (!this.firstResponseBody) return; + // #7388: key the "already logged" guard per logical turn instead of once + // per WS connection. Terminal events from a real response carry + // `response.id` — use it so each turn on a reused connection logs + // independently, while the same id firing twice (retries) still logs + // exactly once. Session-ending failure paths (prepare failure, upstream + // error/close, connect failure) don't carry a response id — they end the + // session, so they share one sentinel key and still log exactly once. + const turnId = toStringOrNull(terminalMessage?.response?.id) || SESSION_TERMINAL_TURN_KEY; + if (this.loggedTurnIds.has(turnId)) return; + this.loggedTurnIds.add(turnId); const finishedAt = Date.now(); try { @@ -723,7 +750,7 @@ class ResponsesWsSession { success, errorCode, errorMessage, - clientRequest: this.firstResponseBody, + clientRequest: this.currentRequestBody || this.firstResponseBody, terminalMessage, responseBody, sourceFormat: "openai-responses", diff --git a/tests/unit/responses-ws-proxy-multi-turn-history.test.ts b/tests/unit/responses-ws-proxy-multi-turn-history.test.ts new file mode 100644 index 0000000000..fc9248a55e --- /dev/null +++ b/tests/unit/responses-ws-proxy-multi-turn-history.test.ts @@ -0,0 +1,256 @@ +// Regression test for issue #7388: Responses WebSocket history/usage logging +// was scoped to `ResponsesWsSession.historyLogged` — a single boolean per +// WebSocket CONNECTION — instead of per logical `response.create` turn. When +// a Codex client reuses one WebSocket connection for two sequential turns, +// only the first terminal event (`response.completed`) was persisted to +// `call_logs`; the second turn's usage/history was silently dropped. +// +// This test opens ONE WebSocket connection, sends two `response.create` +// messages sequentially, and has the fake upstream emit two distinct +// `response.completed` events (different `response.id`, different usage). +// EXPECTED (post-fix): two "log" internal requests, one per turn, each +// carrying its own terminal response id and its own request body. +import test from "node:test"; +import assert from "node:assert/strict"; +import http from "node:http"; + +const { createResponsesWsProxy } = await import("../../scripts/dev/responses-ws-proxy.mjs"); + +function listen(server: http.Server): Promise { + return new Promise((resolve) => { + server.listen(0, "127.0.0.1", () => { + const address = server.address(); + resolve((address as { port: number }).port); + }); + }); +} + +function close(server: http.Server): Promise { + return new Promise((resolve) => { + server.close(() => resolve()); + }); +} + +function readRequestBody(req: http.IncomingMessage): Promise { + return new Promise((resolve, reject) => { + const chunks: Buffer[] = []; + req.on("data", (chunk) => chunks.push(chunk)); + req.on("end", () => resolve(Buffer.concat(chunks).toString("utf8"))); + req.on("error", reject); + }); +} + +function waitFor( + predicate: () => T | undefined | null | false, + { timeoutMs = 3000, intervalMs = 10 }: { timeoutMs?: number; intervalMs?: number } = {} +): Promise { + const startedAt = Date.now(); + return new Promise((resolve, reject) => { + const timer = setInterval(() => { + try { + const value = predicate(); + if (value) { + clearInterval(timer); + resolve(value); + return; + } + if (Date.now() - startedAt >= timeoutMs) { + clearInterval(timer); + reject(new Error("Timed out waiting for condition")); + } + } catch (error) { + clearInterval(timer); + reject(error); + } + }, intervalMs); + }); +} + +test("#7388: a reused Responses WebSocket connection logs both of two logical turns", async () => { + const internalRequests: Array> = []; + const downstreamMessages: Array> = []; + const upstreamSends: Array> = []; + + 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(200, { "content-type": "application/json" }); + res.end( + JSON.stringify({ + 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.4-mini", + response: { ...body.response, model: "gpt-5.4-mini", stream: undefined }, + }) + ); + 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" })); + }); + + // Fake upstream: reply to whichever turn was just sent with a distinct + // response.completed event (distinct response.id + usage), matching the + // issue's minimal reproduction of two sequential turns on one socket. + let turn = 0; + const fakeUpstream = { + send(data: string) { + const parsed = JSON.parse(data); + upstreamSends.push(parsed); + if (parsed.type !== "response.create") return; + turn += 1; + const currentTurn = turn; + setTimeout(() => { + fakeUpstream.onmessage?.({ + data: JSON.stringify({ + type: "response.completed", + response: { + id: `resp_${currentTurn}`, + model: "gpt-5.4-mini", + status: "completed", + usage: { + input_tokens: 10 * currentTurn, + output_tokens: 20 * currentTurn, + total_tokens: 30 * currentTurn, + }, + }, + }), + }); + }, 10); + }, + close() {}, + onmessage: null as ((event: { data: string }) => void) | null, + onerror: null, + onclose: null, + }; + + const port = await listen(server); + const proxy = createResponsesWsProxy({ + baseUrl: `http://127.0.0.1:${port}`, + bridgeSecret: "bridge-secret", + pingIntervalMs: 1000, + idleTimeoutMs: 10000, + wsFactory: async () => fakeUpstream, + }); + + 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))); + }); + + try { + await new Promise((resolve) => ws.addEventListener("open", resolve, { once: true })); + + // Turn 1 on this single, reused WebSocket connection. + ws.send( + JSON.stringify({ + type: "response.create", + model: "gpt-5.4-mini", + input: [{ role: "user", content: "Reply with exactly: pong1" }], + stream: true, + }) + ); + + await waitFor( + () => downstreamMessages.filter((entry) => entry.type === "response.completed").length === 1 + ); + + // Turn 2 on the SAME WebSocket connection (client reuse), per the issue's + // repro: "Codex clients may reuse one WebSocket connection for multiple + // logical turns." + ws.send( + JSON.stringify({ + type: "response.create", + model: "gpt-5.4-mini", + input: [{ role: "user", content: "Reply with exactly: pong2" }], + stream: true, + }) + ); + + await waitFor( + () => downstreamMessages.filter((entry) => entry.type === "response.completed").length === 2 + ); + + // Both logical turns completed downstream — confirms the repro precondition + // from the issue ("The WebSocket received two terminal events"). + assert.equal( + upstreamSends.filter((entry) => entry.type === "response.create").length, + 2 + ); + assert.equal( + downstreamMessages.filter((entry) => entry.type === "response.completed").length, + 2 + ); + + // Give any async persistHistory() calls a moment to land, then assert on + // the internal "log" calls actually issued to the bridge. + await new Promise((resolve) => setTimeout(resolve, 150)); + const logRequests = internalRequests.filter((entry) => entry.action === "log"); + + // One call-log row per logical turn (2) — the second turn must not be + // dropped by a session-level "already logged" guard (#7388). + assert.equal( + logRequests.length, + 2, + `expected 2 call-log entries (one per logical turn), got ${logRequests.length} — ` + + "second turn's history/usage was dropped by the session-level historyLogged guard (#7388)" + ); + + const respIds = logRequests + .map((entry) => (entry.terminalMessage as { response?: { id?: string } } | null)?.response?.id) + .sort(); + assert.deepEqual( + respIds, + ["resp_1", "resp_2"], + "each logged call should carry its own terminal response.id, not just the first turn's" + ); + + // Each logged call's clientRequest must reflect the request body of the + // turn actually being finalized, not always turn 1's body (#7388). + const contentByTurn = logRequests + .map((entry) => { + const clientRequest = entry.clientRequest as { + input?: Array<{ content?: string }>; + } | null; + return clientRequest?.input?.[0]?.content; + }) + .sort(); + assert.deepEqual( + contentByTurn, + ["Reply with exactly: pong1", "Reply with exactly: pong2"], + "each logged call's clientRequest should carry its own turn's request body" + ); + } finally { + ws.close(); + await close(server); + } +});