From ba7d8e166b39865ae6c1a9713d45016bc4d3b152 Mon Sep 17 00:00:00 2001 From: Christo Wilken Date: Sun, 13 Sep 2026 16:11:11 +0200 Subject: [PATCH 1/2] feat(oauth): log the usage-limit state the Responses socket already reports The OpenAI Responses WebSocket sends a codex.rate_limits event on the connection clodex already holds; native Codex parses it (codex-rs/codex-api/src/endpoint/responses_websocket.rs, parse_rate_limit_event, rust-v0.154.0). clodex dropped it: handleSocketMessage returns before parsing when no request is in flight, and during a response the frame was never recorded. With --ws-diagnostics this records a ws_rate_limits event per frame: rate_limits, additional_rate_limits, code_review_rate_limits, credits and promo, uncoerced, each with its serialized size and dropped above 8,000 characters, plus bounded field names and plan type. A frame inside a response is emitted through the request's own diagnostic sink, so it carries that request's ids. Idle frames use a connection sink with an explicit empty correlation: socket callbacks run in the async context of the request that created the socket, so the ambient context would stamp an older request's ids on a reused connection. The sink is wired at both createConnection call sites, including the transport-retry replacement. Observation only. No request, head decision or upstream write changes. Documented in .claude/docs/oauth-continuation.md; README and --help mention the events. Session: a5ae056f-79a5-4beb-831d-c16d7b2d37dc --- .claude/docs/oauth-continuation.md | 34 ++++ README.md | 2 +- src/cli.ts | 5 +- src/oauth/responses-websocket.ts | 126 +++++++++++- tests/responses-websocket.test.ts | 313 +++++++++++++++++++++++++++++ 5 files changed, 476 insertions(+), 4 deletions(-) diff --git a/.claude/docs/oauth-continuation.md b/.claude/docs/oauth-continuation.md index ed014bd1..6009feb0 100644 --- a/.claude/docs/oauth-continuation.md +++ b/.claude/docs/oauth-continuation.md @@ -589,6 +589,40 @@ Trade-off: a request whose `tools` omit a tool that appears in its own history g that tool, which is the pre-fix behaviour. Subagent histories never contain the parent's calls, so the residual is narrow. +### Account-meter diagnostics + +The Responses socket carries the account meter. Native Codex parses a `codex.rate_limits` event off +this same connection (`codex-rs/codex-api/src/endpoint/responses_websocket.rs`, +`parse_rate_limit_event`, rust-v0.154.0). With WS diagnostics enabled, clodex records each one as a +`ws_rate_limits` event: + +- `phase` is `during_response` when the frame arrived while a request was in flight and `idle` when + it did not. Keep the two apart when attributing a debit: an idle frame's change belongs to no + particular response. +- Correlation follows the phase. A `during_response` frame carries the in-flight request's + `requestId` and `claudeSessionId`; an `idle` frame carries neither. Socket callbacks run in the + async context of the request that created the socket, so reading the ambient diagnostic context + there would stamp an older request's ids on a reused head's frames. The connection sink passes an + explicit empty correlation for that reason. +- `rateLimits`, `additionalRateLimits`, `codeReviewRateLimits`, `credits` and `promo` are the + frame's `rate_limits`, `additional_rate_limits`, `code_review_rate_limits`, `credits` and `promo` + values passed through uncoerced, so a fractional percent survives and a field the server omitted + stays absent instead of reading as zero. Each has a `…Bytes` sibling with its serialized size in UTF-8 + bytes, and the value itself is dropped when that exceeds 8,000 bytes. `additionalRateLimits` holds the + separately metered allowances, keyed by allowance name, so it answers whether one of those moved. +- `fieldCount` is the number of top-level keys; `fieldsPresent` lists their names (at most 24, each + through `boundedDiagnosticIdentifier`); `planType` passes through the same helper. +- Unlike the rest of this log, which records upstream strings only as bounded identifiers or hashes, + these values are the server's own objects, recorded verbatim. They include account state such as + credit balance and promotions. + +Before this, `handleSocketMessage` returned before parsing whenever no request was in flight, so +those frames were never read. The idle path needs a connection-scoped sink because +`RequestContext.emitDiagnostic` only exists mid-request; it is wired at every `createConnection` +caller, including the transport-retry replacement. + +Observation only: nothing here changes a request, a head decision or what is sent upstream. + ### Mismatch diagnostics On a history mismatch the head-decision log includes `expected_hash`/`actual_hash` (SHA-256 of each diff --git a/README.md b/README.md index 0100c871..df123065 100644 --- a/README.md +++ b/README.md @@ -146,7 +146,7 @@ Common options (both modes): | `--save-mode` | With `--endpoint`/`--proxy`: save that mode as the `server` default | | `--port <1-65535>` | Listen port (default 17645) | | `--no-discovery` | Don't advertise this server in `~/.clodex/server-runtime.json` (`CLODEX_NO_DISCOVERY=1` also works). Use it for a standalone endpoint the `clodex-claude` wrapper should ignore. | -| `--ws-diagnostics` | Log sanitized request envelopes and WebSocket head decisions | +| `--ws-diagnostics` | Log sanitized request envelopes and WebSocket head decisions, plus the usage-limit reports the OpenAI socket sends, verbatim (they include account state such as credits) | | `--help`, `--version` | Help / version | Endpoint mode only (an error if combined with `--proxy`): diff --git a/src/cli.ts b/src/cli.ts index 429f6a2a..7f6b338e 100644 --- a/src/cli.ts +++ b/src/cli.ts @@ -514,8 +514,9 @@ ${pc.bold('Common options (both modes):')} ~/.clodex/server-runtime.json, so the clodex-claude wrapper never bridges to it (CLODEX_NO_DISCOVERY=1 works too) - --ws-diagnostics Log sanitized request envelopes and WebSocket - head decisions + --ws-diagnostics Log sanitized request envelopes, WebSocket + head decisions, and the server's usage-limit + reports verbatim (includes account credits) --help, --version Help / version ${pc.bold('Endpoint mode only')} ${pc.dim('(error if combined with --proxy)')}: diff --git a/src/oauth/responses-websocket.ts b/src/oauth/responses-websocket.ts index 405704f7..2e75ab71 100644 --- a/src/oauth/responses-websocket.ts +++ b/src/oauth/responses-websocket.ts @@ -224,6 +224,14 @@ interface ConnectionEntry { canonicalToolDefaultsId?: string; options: Required>; debug: (message: string) => void; + /** + * Connection-scoped diagnostic sink. `RequestContext.emitDiagnostic` only exists + * while a request is in flight, and `codex.rate_limits` frames can arrive between + * or after responses, so they belong to the CONNECTION. Without this they + * are observed by nobody: the message handler returns before parsing when there is + * no active context. + */ + connectionDiagnostic?: (event: { event: string } & Record) => void; } // A Claude session partition can have multiple valid conversation heads at @@ -1799,7 +1807,11 @@ function transportReplaySafe(ctx: RequestContext): boolean { function handleSocketMessage(entry: ConnectionEntry, data: RawData): void { const ctx = entry.current; - if (!ctx || ctx.closed) return; + if (!ctx || ctx.closed) { + // A usage-limit frame between or after responses is only read when diagnostics are on. + if (entry.connectionDiagnostic) observeIdleFrame(entry, data); + return; + } const text = Array.isArray(data) ? Buffer.concat(data).toString('utf8') : data.toString('utf8'); ctx.frameCount += 1; if (ctx.transportRetryPending) { @@ -1820,6 +1832,10 @@ function handleSocketMessage(entry: ConnectionEntry, data: RawData): void { } const type = eventType(event); + // Emitted through the request's own sink so the frame carries THIS request's ids. + // The connection sink would not: socket callbacks run in the async context of + // whichever request created the socket, which on a reused head is an older one. + if (isQuotaEvent(type)) observeQuotaEvent(entry, event, 'during_response', ctx.emitDiagnostic); trackReasoningProtocol(entry, ctx, event, type); captureOutput(ctx, event); if (type === 'response.completed') { @@ -2134,6 +2150,101 @@ function numericRetryAfterHeader(value: string | string[] | undefined): number | : undefined; } +/** The upstream event that carries account-meter state rather than response data. */ +function isQuotaEvent(type: string | undefined): boolean { + return type === 'codex.rate_limits'; +} + +/** Largest serialized ledger (UTF-8 bytes) recorded verbatim; bigger ones keep only their size. */ +const QUOTA_LEDGER_MAX_BYTES = 8000; +/** Bound on how many top-level field names one event may list. */ +const QUOTA_FIELDS_MAX_COUNT = 24; + +function boundedLedger(value: unknown): { value?: unknown; bytes?: number } { + let serialized: string | undefined; + try { + serialized = JSON.stringify(value); + } catch { + return {}; + } + if (serialized === undefined) return {}; + const bytes = Buffer.byteLength(serialized); + return bytes <= QUOTA_LEDGER_MAX_BYTES ? { value, bytes } : { bytes }; +} + +/** + * Record what an upstream frame says about the ACCOUNT's allowance, verbatim. + * + * Native Codex parses `codex.rate_limits` off this same socket + * (`codex-rs/codex-api/src/endpoint/responses_websocket.rs` → `parse_rate_limit_event` + * at `rust-v0.154.0`), so the protocol carries the signal even though clodex has + * never looked at it. Nothing here changes inference: it observes and returns. + * + * Two rules the measurement depends on: + * - values are passed through uncoerced — no `?? 0`, no Number() coercion — so a + * fractional percent survives and a missing field stays distinguishable from a + * measured zero (`fieldsPresent` says which keys actually existed); + * - `phase` records whether the frame arrived inside a response or between them. + * An idle frame belongs to the connection; attributing its debit to the last + * response would invent a number. + * + * `emit` decides the correlation: the in-flight request's sink during a response, + * the uncorrelated connection sink while idle. + */ +function observeQuotaEvent( + entry: ConnectionEntry, + event: unknown, + phase: 'during_response' | 'idle', + emit: ConnectionEntry['connectionDiagnostic'], +): void { + if (!emit) return; + const record = event as Record; + // The sibling ledgers ride the same frame: `additional_rate_limits` holds the + // separately metered allowances, `code_review_rate_limits` the code-review one, + // and `credits` and `promo` the account's credit and promotion state. Capturing + // only `rate_limits` would leave "did a separate allowance move" unanswerable. + const rate = boundedLedger(record.rate_limits); + const additional = boundedLedger(record.additional_rate_limits); + const codeReview = boundedLedger(record.code_review_rate_limits); + const credits = boundedLedger(record.credits); + const promo = boundedLedger(record.promo); + emit({ + event: 'ws_rate_limits', + connectionId: entry.debugId, + generation: entry.generation, + phase, + upstreamEventType: 'codex.rate_limits', + fieldCount: Object.keys(record).length, + fieldsPresent: Object.keys(record) + .slice(0, QUOTA_FIELDS_MAX_COUNT) + .map(boundedDiagnosticIdentifier) + .filter((name): name is string => name !== undefined), + rateLimits: rate.value, + rateLimitsBytes: rate.bytes, + additionalRateLimits: additional.value, + additionalRateLimitsBytes: additional.bytes, + codeReviewRateLimits: codeReview.value, + codeReviewRateLimitsBytes: codeReview.bytes, + credits: credits.value, + creditsBytes: credits.bytes, + promo: promo.value, + promoBytes: promo.bytes, + planType: boundedDiagnosticIdentifier(record.plan_type), + }); +} + +/** Parse a frame that arrived with no request in flight, purely to observe quota. */ +function observeIdleFrame(entry: ConnectionEntry, data: RawData): void { + let event: unknown; + try { + event = JSON.parse(Array.isArray(data) ? Buffer.concat(data).toString('utf8') : data.toString('utf8')); + } catch { + return; + } + if (!isQuotaEvent(eventType(event))) return; + observeQuotaEvent(entry, event, 'idle', entry.connectionDiagnostic); +} + function createConnection( WebSocket: WebSocketConstructor, wsUrl: string, @@ -2144,6 +2255,7 @@ function createConnection( debug: ConnectionEntry['debug'], /** Optional HTTP(S)_PROXY CONNECT-tunnel agent (see src/outbound-proxy.ts). */ agent?: import('node:http').Agent, + connectionDiagnostic?: ConnectionEntry['connectionDiagnostic'], ): ConnectionEntry { const now = options.now(); const socket = new WebSocket(wsUrl, agent ? { headers, agent } : { headers }); @@ -2160,6 +2272,7 @@ function createConnection( inFlight: false, options, debug, + connectionDiagnostic, }; if (persistent && key) registerEntry(entry); debug( @@ -2748,6 +2861,15 @@ export function createResponsesWebSocketFetch( evictions, }, diagnosticCorrelation); + // Connection-scoped sink, deliberately uncorrelated. Its caller is a socket + // callback, and those run in the async context of the request that CREATED the + // socket, so the default (`diagnosticContext.getStore()`) would stamp an idle frame + // with that first request's ids. The explicit empty + // correlation keeps them unattributed; in-response frames use `ctx.emitDiagnostic`. + const connectionDiagnostic: ConnectionEntry['connectionDiagnostic'] = options.onDiagnostic + ? event => emitDiagnostic(options, event, {}) + : undefined; + let activeContext: RequestContext | undefined; const stream = new ReadableStream({ start(controller) { @@ -2783,6 +2905,7 @@ export function createResponsesWebSocketFetch( resolvedOptions, debug, proxyAgent, + connectionDiagnostic, ), }; activeContext = ctx; @@ -2796,6 +2919,7 @@ export function createResponsesWebSocketFetch( resolvedOptions, debug, proxyAgent, + connectionDiagnostic, ); dispatchContext(entry, ctx); diff --git a/tests/responses-websocket.test.ts b/tests/responses-websocket.test.ts index 5027b7a5..8f413da0 100644 --- a/tests/responses-websocket.test.ts +++ b/tests/responses-websocket.test.ts @@ -6747,3 +6747,316 @@ describe('new-connection pacing', () => { } }); }); + +describe('usage-limit diagnostics', () => { + // A missing event has to mean the server sent nothing, not that nobody looked. + + function quotaFrame(overrides: Record = {}): string { + return JSON.stringify({ + type: 'codex.rate_limits', + rate_limits: { + // Fractional on purpose, to show the value is not coerced. + primary: { used_percent: 12.5, window_minutes: 10_080, reset_at: 1_789_805_434 }, + // A present zero must stay distinguishable from an absent field. + secondary: { used_percent: 0 }, + }, + plan_type: 'pro', + ...overrides, + }); + } + + it('captures a rate-limit frame that arrives DURING a response', async () => { + const diagnostics: ResponsesWebSocketDiagnosticEvent[] = []; + const wsFetch = createResponsesWebSocketFetch(WS_URL, () => {}, { + onDiagnostic: event => diagnostics.push(event), + }); + const res = await wsFetch('https://x', { method: 'POST', headers: {}, body: '{}' }); + const socket = lastSocket(); + socket.emit('open'); + socket.emit('message', Buffer.from(quotaFrame())); + socket.emit('message', Buffer.from(JSON.stringify({ type: 'response.completed' }))); + await readAll(res); + + const observed = diagnostics.filter(d => d.event === 'ws_rate_limits'); + expect(observed).toHaveLength(1); + expect(observed[0]!.phase).toBe('during_response'); + expect(observed[0]!.planType).toBe('pro'); + const limits = observed[0]!.rateLimits as { + primary: { used_percent: number; window_minutes: number }; + secondary: { used_percent: number }; + }; + // Fractional precision survives, and is not coerced to an integer. + expect(limits.primary.used_percent).toBe(12.5); + expect(limits.primary.window_minutes).toBe(10_080); + // A measured zero is recorded AS zero, not dropped. + expect(limits.secondary.used_percent).toBe(0); + }); + + it('captures a rate-limit frame that arrives with NO request in flight', async () => { + // The case the transport used to discard outright: handleSocketMessage returned + // before parsing whenever `entry.current` was absent, so an account-meter frame + // between or after responses was seen by nobody. + const diagnostics: ResponsesWebSocketDiagnosticEvent[] = []; + const wsFetch = createResponsesWebSocketFetch(WS_URL, () => {}, { + onDiagnostic: event => diagnostics.push(event), + }); + const res = await wsFetch('https://x', { method: 'POST', headers: {}, body: '{}' }); + const socket = lastSocket(); + socket.emit('open'); + socket.emit('message', Buffer.from(JSON.stringify({ type: 'response.completed' }))); + await readAll(res); + diagnostics.length = 0; + + socket.emit('message', Buffer.from(quotaFrame())); + + const observed = diagnostics.filter(d => d.event === 'ws_rate_limits'); + expect(observed).toHaveLength(1); + expect(observed[0]!.phase).toBe('idle'); + }); + + it('omits a field the server did not send rather than reporting it as zero', async () => { + const diagnostics: ResponsesWebSocketDiagnosticEvent[] = []; + const wsFetch = createResponsesWebSocketFetch(WS_URL, () => {}, { + onDiagnostic: event => diagnostics.push(event), + }); + const res = await wsFetch('https://x', { method: 'POST', headers: {}, body: '{}' }); + const socket = lastSocket(); + socket.emit('open'); + socket.emit('message', Buffer.from(JSON.stringify({ + type: 'codex.rate_limits', + rate_limits: { primary: { used_percent: 3 } }, + }))); + socket.emit('message', Buffer.from(JSON.stringify({ type: 'response.completed' }))); + await readAll(res); + + const observed = diagnostics.find(d => d.event === 'ws_rate_limits')!; + const limits = observed.rateLimits as { primary: Record }; + expect(limits.primary.used_percent).toBe(3); + expect('window_minutes' in limits.primary).toBe(false); + expect(observed.planType).toBeUndefined(); + }); + + it('captures the sibling allowance ledgers that ride the same frame', async () => { + const diagnostics: ResponsesWebSocketDiagnosticEvent[] = []; + const wsFetch = createResponsesWebSocketFetch(WS_URL, () => {}, { + onDiagnostic: event => diagnostics.push(event), + }); + const res = await wsFetch('https://x', { method: 'POST', headers: {}, body: '{}' }); + const socket = lastSocket(); + socket.emit('open'); + socket.emit('message', Buffer.from(JSON.stringify({ + type: 'codex.rate_limits', + rate_limits: { primary: { used_percent: 2 } }, + // Keyed by allowance name, as live frames send it. + additional_rate_limits: { + 'gpt-reserve': { + allowed: true, + limit_reached: false, + primary: { used_percent: 2, window_minutes: 10_080, reset_at: 1_789_382_732 }, + secondary: null, + }, + }, + code_review_rate_limits: { allowed: true, primary: { used_percent: 4 } }, + credits: null, + promo: { active: false }, + }))); + socket.emit('message', Buffer.from(JSON.stringify({ type: 'response.completed' }))); + await readAll(res); + + const observed = diagnostics.find(d => d.event === 'ws_rate_limits')!; + const additional = observed.additionalRateLimits as Record; + expect(additional['gpt-reserve']!.primary.used_percent).toBe(2); + // A null ledger is recorded as null, not dropped as if absent. + expect(observed.credits).toBeNull(); + expect('credits' in observed).toBe(true); + expect((observed.codeReviewRateLimits as { primary: { used_percent: number } }).primary.used_percent).toBe(4); + expect(observed.promo).toEqual({ active: false }); + }); + + it('stays silent on an idle frame that carries no meter state', async () => { + // Silence has to mean silence, or a null result is unreadable. + const diagnostics: ResponsesWebSocketDiagnosticEvent[] = []; + const wsFetch = createResponsesWebSocketFetch(WS_URL, () => {}, { + onDiagnostic: event => diagnostics.push(event), + }); + const res = await wsFetch('https://x', { method: 'POST', headers: {}, body: '{}' }); + const socket = lastSocket(); + socket.emit('open'); + socket.emit('message', Buffer.from(JSON.stringify({ type: 'response.completed' }))); + await readAll(res); + diagnostics.length = 0; + + socket.emit('message', Buffer.from(JSON.stringify({ type: 'response.output_text.delta', delta: 'x' }))); + socket.emit('message', Buffer.from('not json at all')); + socket.emit('message', Buffer.from(JSON.stringify({ type: 'codex.response.metadata', rate_limits: {} }))); + expect(diagnostics.filter(d => d.event === 'ws_rate_limits')).toHaveLength(0); + + // The same idle path does report a real meter frame, so the silence above came + // from the filter and not from an observer that was never listening. + socket.emit('message', Buffer.from(quotaFrame())); + expect(diagnostics.filter(d => d.event === 'ws_rate_limits')).toHaveLength(1); + }); + + it('attributes a meter frame to the request in flight, not to the request that opened the socket', async () => { + // Socket callbacks run in the async context of the request that CREATED the + // socket. On a reused head that is an older request, so a frame correlated from + // the ambient store carries the wrong requestId and session. The emits below run + // inside the first request's context to reproduce that. + const diagnostics: ResponsesWebSocketDiagnosticEvent[] = []; + const wsFetch = createResponsesWebSocketFetch(WS_URL, () => {}, { + onDiagnostic: event => diagnostics.push(event), + }); + const contextA = { requestId: 'req-A', claudeSessionId: 'session-A' }; + const inA = (fn: () => void) => withResponsesWebSocketDiagnosticContext(contextA, fn); + const firstUser = { role: 'user', content: [{ type: 'input_text', text: 'first' }] }; + + const socketsBefore = socketCount(); + const first = await withResponsesWebSocketDiagnosticContext( + contextA, + () => wsFetch('https://x', { + method: 'POST', headers: {}, body: JSON.stringify(sessionPayload([firstUser])), + }), + ); + const socket = lastSocket(); + inA(() => { + socket.emit('open'); + emitTextResponse(socket, 'resp_A', 'first answer'); + }); + await readAll(first); + + const secondInput = [ + firstUser, + { role: 'assistant', content: [{ type: 'output_text', text: 'first answer' }] }, + { role: 'user', content: [{ type: 'input_text', text: 'second' }] }, + ]; + const second = await withResponsesWebSocketDiagnosticContext( + { requestId: 'req-B', claudeSessionId: 'session-B' }, + () => wsFetch('https://x', { + method: 'POST', headers: {}, body: JSON.stringify(sessionPayload(secondInput)), + }), + ); + expect(socketCount()).toBe(socketsBefore + 1); + inA(() => { + socket.emit('message', Buffer.from(quotaFrame())); + emitTextResponse(socket, 'resp_B', 'second answer'); + }); + await readAll(second); + inA(() => socket.emit('message', Buffer.from(quotaFrame()))); + + const [during, idle] = diagnostics.filter(d => d.event === 'ws_rate_limits'); + expect(during).toMatchObject({ phase: 'during_response', requestId: 'req-B', claudeSessionId: 'session-B' }); + expect(idle!.phase).toBe('idle'); + expect(idle!.requestId).toBeUndefined(); + expect(idle!.claudeSessionId).toBeUndefined(); + }); + + it('observes idle meter frames on a replacement socket after a transport retry', async () => { + const diagnostics: ResponsesWebSocketDiagnosticEvent[] = []; + const wsFetch = createResponsesWebSocketFetch(WS_URL, () => {}, { + onDiagnostic: event => diagnostics.push(event), + }); + const res = await wsFetch('https://x', { + method: 'POST', headers: {}, body: JSON.stringify(sessionPayload([])), + }); + const socketsBefore = socketCount(); + lastSocket().emit('error', Object.assign(new Error('reset'), { code: 'ECONNRESET' })); + expect(socketCount()).toBe(socketsBefore + 1); + const replacement = lastSocket(); + replacement.emit('open'); + emitTextResponse(replacement, 'resp_replacement', 'recovered'); + await readAll(res); + diagnostics.length = 0; + + replacement.emit('message', Buffer.from(quotaFrame())); + + const observed = diagnostics.filter(d => d.event === 'ws_rate_limits'); + expect(observed).toHaveLength(1); + expect(observed[0]!.phase).toBe('idle'); + }); + + it('keeps the size of a ledger too large to record verbatim', async () => { + const diagnostics: ResponsesWebSocketDiagnosticEvent[] = []; + const wsFetch = createResponsesWebSocketFetch(WS_URL, () => {}, { + onDiagnostic: event => diagnostics.push(event), + }); + const res = await wsFetch('https://x', { method: 'POST', headers: {}, body: '{}' }); + const socket = lastSocket(); + socket.emit('open'); + socket.emit('message', Buffer.from(quotaFrame({ + additional_rate_limits: [{ limit_name: 'x'.repeat(9000) }], + credits: { balance: '0' }, + }))); + socket.emit('message', Buffer.from(JSON.stringify({ type: 'response.completed' }))); + await readAll(res); + + const observed = diagnostics.find(d => d.event === 'ws_rate_limits')!; + expect(observed.additionalRateLimits).toBeUndefined(); + expect(observed.additionalRateLimitsBytes).toBeGreaterThan(8000); + expect(observed.credits).toEqual({ balance: '0' }); + expect(observed.creditsBytes).toBe(JSON.stringify({ balance: '0' }).length); + }); + + it('measures ledger size in UTF-8 bytes', async () => { + const diagnostics: ResponsesWebSocketDiagnosticEvent[] = []; + const wsFetch = createResponsesWebSocketFetch(WS_URL, () => {}, { + onDiagnostic: event => diagnostics.push(event), + }); + const res = await wsFetch('https://x', { method: 'POST', headers: {}, body: '{}' }); + const socket = lastSocket(); + socket.emit('open'); + const promo = { label: 'Früh bucher €' }; + socket.emit('message', Buffer.from(quotaFrame({ promo }))); + socket.emit('message', Buffer.from(JSON.stringify({ type: 'response.completed' }))); + await readAll(res); + + const observed = diagnostics.find(d => d.event === 'ws_rate_limits')!; + expect(observed.promoBytes).toBe(Buffer.byteLength(JSON.stringify(promo))); + expect(observed.promoBytes).toBeGreaterThan(JSON.stringify(promo).length); + // Under the limit in characters but over it in bytes: dropped, size kept. + diagnostics.length = 0; + const res2 = await wsFetch('https://x', { method: 'POST', headers: {}, body: '{}' }); + const socket2 = lastSocket(); + socket2.emit('open'); + const wide = { label: '€'.repeat(3000) }; + socket2.emit('message', Buffer.from(quotaFrame({ promo: wide }))); + socket2.emit('message', Buffer.from(JSON.stringify({ type: 'response.completed' }))); + await readAll(res2); + const dropped = diagnostics.find(d => d.event === 'ws_rate_limits')!; + expect(dropped.promo).toBeUndefined(); + expect(dropped.promoBytes).toBe(Buffer.byteLength(JSON.stringify(wide))); + }); + + it('leaves a response untouched when diagnostics are off', async () => { + // Without a diagnostic sink the meter frame must be skipped, not dereferenced. + const wsFetch = createResponsesWebSocketFetch(WS_URL, () => {}); + const res = await wsFetch('https://x', { method: 'POST', headers: {}, body: '{}' }); + const socket = lastSocket(); + socket.emit('open'); + socket.emit('message', Buffer.from(quotaFrame())); + emitTextResponse(socket, 'resp_no_diagnostics', 'still answered'); + expect(await readAll(res)).toContain('still answered'); + }); + + it('bounds and sanitizes the field names and identifiers it records', async () => { + const diagnostics: ResponsesWebSocketDiagnosticEvent[] = []; + const wsFetch = createResponsesWebSocketFetch(WS_URL, () => {}, { + onDiagnostic: event => diagnostics.push(event), + }); + const res = await wsFetch('https://x', { method: 'POST', headers: {}, body: '{}' }); + const socket = lastSocket(); + socket.emit('open'); + const extra: Record = { 'bad\nkey': 1 }; + for (let i = 0; i < 40; i += 1) extra[`field_${i}`] = i; + socket.emit('message', Buffer.from(quotaFrame({ plan_type: 'pro\nforged', ...extra }))); + socket.emit('message', Buffer.from(JSON.stringify({ type: 'response.completed' }))); + await readAll(res); + + const observed = diagnostics.find(d => d.event === 'ws_rate_limits')!; + const fields = observed.fieldsPresent as string[]; + expect(fields.length).toBeLessThanOrEqual(24); + expect(fields).toContain('rate_limits'); + expect(fields).not.toContain('bad\nkey'); + expect(observed.fieldCount).toBe(44); + expect(observed.planType).toBeUndefined(); + }); +}); From 86890ec725a8b3c624dacdbd42981861e40ae2d0 Mon Sep 17 00:00:00 2001 From: integ Date: Wed, 16 Sep 2026 10:53:49 -0500 Subject: [PATCH 2/2] fix(oauth): tell users at startup that usage-limit reports are logged verbatim Apply the non-blocking review notes on #237 before landing it: - the `--ws-diagnostics` startup line in both server modes now says the log also holds OpenAI usage-limit reports verbatim, including credits and promotions, matching README and `--help` - the sibling-ledger test pins every `...Bytes` size to its own ledger and the event's `connectionId`/`generation` to the head decision that created the socket - `idle` is documented as "cannot be attributed to a request", not "no response caused it" --- .claude/docs/oauth-continuation.md | 5 +++-- src/http-proxy/index.ts | 1 + src/oauth/responses-websocket.ts | 5 +++-- src/server/index.ts | 1 + tests/responses-websocket.test.ts | 30 ++++++++++++++++++++++++++++-- 5 files changed, 36 insertions(+), 6 deletions(-) diff --git a/.claude/docs/oauth-continuation.md b/.claude/docs/oauth-continuation.md index 66af48f9..4f3d2be4 100644 --- a/.claude/docs/oauth-continuation.md +++ b/.claude/docs/oauth-continuation.md @@ -637,8 +637,9 @@ this same connection (`codex-rs/codex-api/src/endpoint/responses_websocket.rs`, `ws_rate_limits` event: - `phase` is `during_response` when the frame arrived while a request was in flight and `idle` when - it did not. Keep the two apart when attributing a debit: an idle frame's change belongs to no - particular response. + it did not. `idle` means clodex cannot safely attribute the frame to a request, not that no + response caused it: `response.completed` clears the in-flight request on a persistent head, so a + meter frame sent right after completion by that very response is still labeled `idle`. - Correlation follows the phase. A `during_response` frame carries the in-flight request's `requestId` and `claudeSessionId`; an `idle` frame carries neither. Socket callbacks run in the async context of the request that created the socket, so reading the ambient diagnostic context diff --git a/src/http-proxy/index.ts b/src/http-proxy/index.ts index c03e5fe8..02c43e07 100644 --- a/src/http-proxy/index.ts +++ b/src/http-proxy/index.ts @@ -243,6 +243,7 @@ export async function runHttpProxyServerCommand( if (handle.webSocketDiagnosticsLogPath) { console.log(` WebSocket diagnostics: ${handle.webSocketDiagnosticsLogPath}`); console.log(pc.yellow(' Diagnostic mode records request headers and metadata; credential headers are redacted.')); + console.log(pc.yellow(' It also records OpenAI usage-limit reports verbatim, including account credits and promotions.')); } console.log(''); printHttpProxyModels(loaded.routes, loaded.aliases); diff --git a/src/oauth/responses-websocket.ts b/src/oauth/responses-websocket.ts index 5640e8cb..fa139429 100644 --- a/src/oauth/responses-websocket.ts +++ b/src/oauth/responses-websocket.ts @@ -2300,8 +2300,9 @@ function boundedLedger(value: unknown): { value?: unknown; bytes?: number } { * fractional percent survives and a missing field stays distinguishable from a * measured zero (`fieldsPresent` says which keys actually existed); * - `phase` records whether the frame arrived inside a response or between them. - * An idle frame belongs to the connection; attributing its debit to the last - * response would invent a number. + * `idle` only says no request was in flight when the frame arrived; the last + * response may well have caused it (`response.completed` clears `current` + * before a trailing meter frame lands), but clodex cannot safely attribute it. * * `emit` decides the correlation: the in-flight request's sink during a response, * the uncorrelated connection sink while idle. diff --git a/src/server/index.ts b/src/server/index.ts index d98eee42..e3dce26f 100644 --- a/src/server/index.ts +++ b/src/server/index.ts @@ -512,6 +512,7 @@ export async function runServerCommand(options: ServerCommandOptions = {}): Prom if (webSocketDiagnosticsLogPath) { console.log(` WebSocket diagnostics: ${webSocketDiagnosticsLogPath}`); console.log(pc.yellow(' Diagnostic mode records request headers and metadata; credential headers are redacted.')); + console.log(pc.yellow(' It also records OpenAI usage-limit reports verbatim, including account credits and promotions.')); } if (mode === 'network') { for (const { name, address } of getLocalIps()) { diff --git a/tests/responses-websocket.test.ts b/tests/responses-websocket.test.ts index 51fbce34..a8fdc0eb 100644 --- a/tests/responses-websocket.test.ts +++ b/tests/responses-websocket.test.ts @@ -6863,7 +6863,7 @@ describe('usage-limit diagnostics', () => { const res = await wsFetch('https://x', { method: 'POST', headers: {}, body: '{}' }); const socket = lastSocket(); socket.emit('open'); - socket.emit('message', Buffer.from(JSON.stringify({ + const frame = { type: 'codex.rate_limits', rate_limits: { primary: { used_percent: 2 } }, // Keyed by allowance name, as live frames send it. @@ -6878,7 +6878,8 @@ describe('usage-limit diagnostics', () => { code_review_rate_limits: { allowed: true, primary: { used_percent: 4 } }, credits: null, promo: { active: false }, - }))); + }; + socket.emit('message', Buffer.from(JSON.stringify(frame))); socket.emit('message', Buffer.from(JSON.stringify({ type: 'response.completed' }))); await readAll(res); @@ -6890,6 +6891,31 @@ describe('usage-limit diagnostics', () => { expect('credits' in observed).toBe(true); expect((observed.codeReviewRateLimits as { primary: { used_percent: number } }).primary.used_percent).toBe(4); expect(observed.promo).toEqual({ active: false }); + + // Every ledger's size is the size of THAT ledger, and the event names the + // socket the head decision for this request created. A swapped byte count + // or a stale connection id would otherwise pass the field-by-field checks. + const size = (value: unknown) => Buffer.byteLength(JSON.stringify(value)); + const decision = diagnostics.find(d => d.event === 'ws_head_decision')!; + expect(observed).toMatchObject({ + connectionId: decision.createdConnectionId, + generation: decision.createdGeneration, + phase: 'during_response', + upstreamEventType: 'codex.rate_limits', + fieldCount: 6, + fieldsPresent: ['type', 'rate_limits', 'additional_rate_limits', 'code_review_rate_limits', 'credits', 'promo'], + rateLimits: frame.rate_limits, + rateLimitsBytes: size(frame.rate_limits), + additionalRateLimits: frame.additional_rate_limits, + additionalRateLimitsBytes: size(frame.additional_rate_limits), + codeReviewRateLimits: frame.code_review_rate_limits, + codeReviewRateLimitsBytes: size(frame.code_review_rate_limits), + credits: null, + creditsBytes: size(null), + promo: frame.promo, + promoBytes: size(frame.promo), + }); + expect(observed.planType).toBeUndefined(); }); it('stays silent on an idle frame that carries no meter state', async () => {