From b4e464cb067cda437a41d7d3c39cd8ec75a005aa Mon Sep 17 00:00:00 2001 From: Julius Olsson Date: Sun, 27 Sep 2026 02:52:45 -0700 Subject: [PATCH 1/5] feat(proxy): keep the newest Responses request body beside the events file (agent-code#1336) Co-Authored-By: Claude Opus 5.5 --- API.md | 2 +- SECURITY.md | 5 + docs/plans/2026-09-27-latest-request-body.md | 28 ++++++ src/proxy/latestRequestBody.ts | 98 +++++++++++++++++++ src/proxy/responsesProxy.latestBody.test.ts | 99 ++++++++++++++++++++ src/proxy/responsesProxy.ts | 13 ++- 6 files changed, 242 insertions(+), 3 deletions(-) create mode 100644 docs/plans/2026-09-27-latest-request-body.md create mode 100644 src/proxy/latestRequestBody.ts create mode 100644 src/proxy/responsesProxy.latestBody.test.ts diff --git a/API.md b/API.md index 3dc1e6c..0ec7675 100644 --- a/API.md +++ b/API.md @@ -1601,7 +1601,7 @@ static ResponsesProxy.create(options?: { | --- | --- | --- | --- | | `authMode` | `CodexAuthMode` (`'apikey' \| 'chatgpt'`) | auto-detected | Which auth path Codex uses. Auto-detection reads `~/.codex/auth.json`; absent → `'apikey'`. | | `upstreamBaseUrl` | `string` | derived from `authMode` | The real upstream. Defaults: `apikey` → `https://api.openai.com/v1`, `chatgpt` → `https://chatgpt.com/backend-api/codex`. | -| `eventsFile` | `string` | unset | If set, every emitted `event` is also written as a JSON line to this file (forensic mirror; `Buffer` payloads are inlined as `{ _buffer_b64 }`. Dumps written before agent-code#372's fix hold `{"type":"Buffer","data":[…]}` byte arrays instead, so a reader of older files must accept both). Written asynchronously through one ordered stream, never on the emitting call. When a line would push the total unwritten bytes (across the live and any rotated-out file) past 16 MiB, it is dropped and counted. A `{kind:'mirror-dropped', droppedEvents, droppedBytes}` line precedes the next written line, or is written at `stop()` if no line follows. Markers obey the same bounds: with a queue cap smaller than one marker (only ever a test setting), the marker is not written and the drops stay counted in the mirror's stats. Call `flushMirror()` to wait for the queue; `stop()` flushes and closes it. | +| `eventsFile` | `string` | unset | If set, every emitted `event` is also written as a JSON line to this file (forensic mirror; `Buffer` payloads are inlined as `{ _buffer_b64 }`. Dumps written before agent-code#372's fix hold `{"type":"Buffer","data":[…]}` byte arrays instead, so a reader of older files must accept both). Written asynchronously through one ordered stream, never on the emitting call. When a line would push the total unwritten bytes (across the live and any rotated-out file) past 16 MiB, it is dropped and counted. A `{kind:'mirror-dropped', droppedEvents, droppedBytes}` line precedes the next written line, or is written at `stop()` if no line follows. Markers obey the same bounds: with a queue cap smaller than one marker (only ever a test setting), the marker is not written and the drops stay counted in the mirror's stats. Also keeps the newest `responses*` request body in `latest-request-body.json` beside the file: one line, `{kind:'request-body-latest', requestId, endpoint, body_b64}`, the name and kind of the Claude addon's sidecar. It is replaced by each newer body, never by a bodiless request. It is removed when the newest body is over 16 MiB or a write fails, so it is never an older prompt. Writes are asynchronous and ordered. Call `flushMirror()` to wait for the queue and the sidecar; `stop()` flushes and closes them. | | `eventsFileMaxBytes` | `number` | 64 MiB | When the next line would pass this size, the file is renamed to `.1.jsonl` (replacing any earlier one) and a fresh file starts with `{kind:'mirror-rotated', rotatedBytes, rotations, droppedEvents, droppedBytes}`. Every file stays within the cap, markers included, so a run holds at most two files of this size. A line that cannot fit even in a fresh file is dropped and counted. A restart that appends to a file already at the cap rotates on its first event. | `create()` starts listening before resolving. The resolved instance diff --git a/SECURITY.md b/SECURITY.md index 7fb7255..ebb7616 100644 --- a/SECURITY.md +++ b/SECURITY.md @@ -42,6 +42,11 @@ While the proxy is in use: `Authorization` header** so a leaked bundle cannot expose bearer tokens — but it **does** record request bodies (your prompts, base64, capped at 2 MiB) and response bytes. Treat that file as sensitive. +- With `eventsFile` set, the proxy also keeps the newest Responses request + body (up to 16 MiB) in `latest-request-body.json` next to it, so a debug + bundle has the prompt even after the events tail has moved past it + (agent-code#1336). It is the same prompt text as the mirror. Treat it as + sensitive too. The proxy is **opt-in**. If you don't construct it, no interception happens — Codex talks to the upstream directly and the package observes diff --git a/docs/plans/2026-09-27-latest-request-body.md b/docs/plans/2026-09-27-latest-request-body.md new file mode 100644 index 0000000..57caea9 --- /dev/null +++ b/docs/plans/2026-09-27-latest-request-body.md @@ -0,0 +1,28 @@ +# Keep the newest Responses request body beside the events file (agent-code#1336) + +## Evidence +- **What the bundle carries.** Agent Code's debug bundle includes the last 5 MiB of `proxy-events.jsonl` (plus `.1`). +- **Why the prompt falls out.** On Codex the response chunks are the bulk of that file. So after a long stream, or a run of `/v1/models` refreshes, the `request` event with its `body_b64` is no longer in the tail, and the bundle has no prompt. +- **Measured.** Review C of agent-code#1332 found the last inline body more than 5 MiB before EOF in 40 of 65 Codex files: 10 of 65 after the base64 fix (#53). +- **The Claude precedent.** Claude solved the same loss with `latest-request-body.json` (claude-code-headless#62, agent-code#1273). Agent Code's `readLatestRequestBody` already appends `/latest-request-body.json` for any provider. Codex never writes one. + +## Change +- **New module, `src/proxy/latestRequestBody.ts`:** `LatestRequestBodySidecar`. For each `responses*` request with a body, it replaces the sidecar with `{kind:'request-body-latest', requestId, endpoint, body_b64}`. + - **Ordered writes.** One ordered chain of async writes (temp file plus rename): no sync I/O on the main process, and an older body can never rename over a newer one. + - **Removal instead of staleness.** It removes the sidecar when the newest body is over 16 MiB (the Claude cap) or a write fails, so it never shows an older prompt as current. + - **Bodiless requests are skipped.** `/models` GETs never touch it. +- **Wiring in `ResponsesProxy`.** It is created with the mirror (same `eventsFile` opt-in) and recorded right after the `request` event. `flushMirror()` and `stop()` also wait for it. +- **Docs.** `API.md` and `SECURITY.md` document the file, including that it holds prompt text. + +## Invariant, and how it differs from Claude +Claude's sidecar holds the newest body that is NOT in its log, because that log omits bodies past a budget. Codex never omits bodies up to 2 MiB; the loss is in the bundle's tail, which the proxy cannot see. So this sidecar always holds the newest Responses body, even when the log also has it. The cost is at most one body appearing twice in a bundle. + +## Tests (`responsesProxy.latestBody.test.ts`, real requests through the proxy to a local upstream) +1. After two POSTs, the sidecar holds the second body, with its `requestId`, as one line. +2. A `/models` GET after a prompt leaves the prompt in place. +3. A body over 16 MiB removes an existing sidecar. + +All three are red with the wiring removed. + +## Then +The app bump carries this to Agent Code. Its reader's WHY comment, which describes only the Claude invariant, gets the Codex one. diff --git a/src/proxy/latestRequestBody.ts b/src/proxy/latestRequestBody.ts new file mode 100644 index 0000000..ff418a6 --- /dev/null +++ b/src/proxy/latestRequestBody.ts @@ -0,0 +1,98 @@ +import { readdir, rename, rm, writeFile } from 'fs/promises' +import { basename, dirname, join } from 'path' + +// The newest Responses request body, kept in a sidecar next to the proxy +// events file (agent-code#1336). +// +// WHY a sidecar at all: Agent Code's debug bundle carries the last 5 MiB of +// `proxy-events.jsonl` (plus the rotated `.1`). On Codex the response chunks +// are the bulk of that file, so a turn that streams more than 5 MiB, or a run +// of `/v1/models` refreshes, pushes the request event (and its `body_b64`) +// out of the tail. The bundle then holds recent chunks and no prompt, which +// is the one thing a bug report needs. Measured on the live corpus (review C +// of agent-code#1332): the last inline body lay more than 5 MiB before EOF in +// 40 of 65 Codex files (10 of 65 once chunks are base64). Rotation cannot +// help, because the problem is the tail, not the file size. +// +// WHY this name and line shape: it is the Claude addon's sidecar +// (claude-code-headless `mitmAddon.py`, agent-code#1273), and Agent Code's +// `readLatestRequestBody` already appends `/latest-request-body.json` +// to the bundle's proxy section for ANY provider. One JSON line with its own +// `kind`, so nothing that replays the events file mistakes it for a request. +// +// The INVARIANT differs from Claude's, on purpose. Claude's sidecar holds the +// newest body that is NOT in the log (its log omits bodies past a budget). +// Codex never omits bodies up to the 2 MiB inline cap, and the loss here is +// the bundle's tail, which the proxy cannot see. So this sidecar always holds +// the newest Responses request body, even when the log also has it. The cost +// is at most one body (bounded below) appearing twice in a bundle. +// +// WHY only `responses*` endpoints: they carry the prompt (the main turn, +// `responses/compact`). `/v1/models` is a GET with no body. A turn's later +// requests (tool results) resend the whole input, so "newest" still holds +// every prompt so far. A title side request (0.157 `thread_title.rs`) is a +// Responses request too, so the sidecar can briefly hold one; it carries the +// same user prompt, and the next main-turn request replaces it. +// +// WHY one ordered chain of async writes: the proxy runs on Agent Code's main +// process, where the old synchronous mirror appends were the problem +// (agent-code#372), so no sync I/O. But independent async writes can finish +// out of order and rename an OLDER body over a newer one. Chaining them keeps +// "latest" true. A failed write removes the sidecar, so a reader never sees a +// stale prompt as the current one. Best-effort throughout: forensics must +// never disturb the live proxy. + +export const LATEST_REQUEST_BODY_FILE_NAME = 'latest-request-body.json' + +// Same cap as the Claude addon's sidecar (`_LATEST_BODY_CAP`). Larger than the +// 2 MiB inline cap on purpose: a body too big to inline is exactly the one the +// events file cannot show at all. +export const LATEST_REQUEST_BODY_CAP = 16 * 1024 * 1024 + +export class LatestRequestBodySidecar { + readonly path: string + private chain: Promise = Promise.resolve() + private writes = 0 + + constructor(eventsFile: string) { + this.path = join(dirname(eventsFile), LATEST_REQUEST_BODY_FILE_NAME) + } + + record(requestId: string, endpoint: string, body: Buffer | undefined): void { + if (!endpoint.startsWith('responses') || !body || body.length === 0) return + const write = this.writes++ + const line = body.length > LATEST_REQUEST_BODY_CAP + ? null + : JSON.stringify({ kind: 'request-body-latest', requestId, endpoint, body_b64: body.toString('base64') }) + '\n' + this.chain = this.chain.then(() => (line === null ? this.remove() : this.replace(line, write))) + } + + /** Resolves once every recorded body has been written or removed. */ + flush(): Promise { + return this.chain + } + + private async replace(line: string, write: number): Promise { + // A per-write temp name, so a crash mid-write never leaves a half body + // under the real name. + const temp = `${this.path}.${process.pid}.${write}.tmp` + try { + await writeFile(temp, line, 'utf-8') + await rename(temp, this.path) + } catch { + await rm(temp, { force: true }).catch(() => {}) + await this.remove() + } + } + + private async remove(): Promise { + await rm(this.path, { force: true }).catch(() => {}) + // Temp files a crash between write and rename left behind. + try { + const prefix = `${basename(this.path)}.` + for (const name of await readdir(dirname(this.path))) { + if (name.startsWith(prefix) && name.endsWith('.tmp')) await rm(join(dirname(this.path), name), { force: true }).catch(() => {}) + } + } catch { /* directory gone: nothing to clean */ } + } +} diff --git a/src/proxy/responsesProxy.latestBody.test.ts b/src/proxy/responsesProxy.latestBody.test.ts new file mode 100644 index 0000000..55f5834 --- /dev/null +++ b/src/proxy/responsesProxy.latestBody.test.ts @@ -0,0 +1,99 @@ +import { existsSync, mkdtempSync, readFileSync, rmSync } from 'node:fs' +import { createServer, request as httpRequest, type Server } from 'node:http' +import { tmpdir } from 'node:os' +import { join } from 'node:path' + +import { afterEach, expect, it } from 'vitest' + +import { ResponsesProxy } from './responsesProxy.js' + +// agent-code#1336: Agent Code's debug bundle tails the newest 5 MiB of +// proxy-events.jsonl. On Codex the response chunks are the bulk, so after a +// long stream the request event and its body_b64 fall out of the tail and the +// bundle has no prompt (40 of 65 recorded files, review C of #1332). The +// Claude addon solved the same loss with `latest-request-body.json`, which +// Agent Code already appends for any provider. These drive REAL requests +// through the proxy to a local upstream and read what lands on disk. + +// Literal on purpose, not imported: Agent Code's reader looks for this exact +// name, and the test must fail on a proxy that never writes it. +const LATEST_REQUEST_BODY_FILE_NAME = 'latest-request-body.json' +// The Claude addon's sidecar cap, mirrored. +const LATEST_REQUEST_BODY_CAP = 16 * 1024 * 1024 + +const cleanup: Array<() => Promise | void> = [] +afterEach(async () => { for (const step of cleanup.splice(0).reverse()) await step() }) + +async function startProxy(): Promise<{ proxy: ResponsesProxy; eventsFile: string; sidecar: string }> { + const upstream: Server = createServer((req, res) => { + req.resume() + req.on('end', () => { + res.writeHead(200, { 'content-type': 'text/event-stream' }) + res.end('event: response.completed\ndata: {"type":"response.completed"}\n\n') + }) + }) + await new Promise(resolve => upstream.listen(0, '127.0.0.1', resolve)) + cleanup.push(() => new Promise(resolve => { upstream.closeAllConnections(); upstream.close(() => resolve()) })) + const address = upstream.address() + if (!address || typeof address === 'string') throw new Error('upstream did not bind') + const dir = mkdtempSync(join(tmpdir(), 'cxh-latest-body-')) + cleanup.push(() => rmSync(dir, { recursive: true, force: true })) + const eventsFile = join(dir, 'proxy-events.jsonl') + const proxy = await ResponsesProxy.create({ + upstreamBaseUrl: `http://127.0.0.1:${address.port}/v1`, + authMode: 'apikey', + eventsFile, + }) + cleanup.push(() => proxy.stop()) + return { proxy, eventsFile, sidecar: join(dir, LATEST_REQUEST_BODY_FILE_NAME) } +} + +function send(proxy: ResponsesProxy, path: string, method: string, body?: Buffer): Promise { + return new Promise((resolve, reject) => { + const client = httpRequest(`${proxy.info.proxyBaseUrl}${path}`, { method, headers: { 'content-type': 'application/json' } }, res => { + res.resume() + res.on('end', () => resolve()) + }) + client.on('error', reject) + client.end(body) + }) +} + +const sidecarLine = (path: string) => JSON.parse(readFileSync(path, 'utf8').trim()) as Record + +it('keeps the newest Responses request body next to the events file', async () => { + const { proxy, sidecar } = await startProxy() + const first = Buffer.from(JSON.stringify({ model: 'gpt-5', input: [{ role: 'user', content: 'first prompt' }] })) + const second = Buffer.from(JSON.stringify({ model: 'gpt-5', input: [{ role: 'user', content: 'second prompt' }] })) + + await send(proxy, '/responses', 'POST', first) + await send(proxy, '/responses', 'POST', second) + await proxy.flushMirror() + + const line = sidecarLine(sidecar) + expect(line).toMatchObject({ kind: 'request-body-latest', requestId: 'req-2', endpoint: 'responses' }) + expect(Buffer.from(line.body_b64 as string, 'base64').equals(second)).toBe(true) + // One line: the reader appends it verbatim to the events tail. + expect(readFileSync(sidecar, 'utf8').split('\n').filter(Boolean)).toHaveLength(1) +}) + +it('is not replaced by a bodiless /models refresh', async () => { + const { proxy, sidecar } = await startProxy() + const prompt = Buffer.from(JSON.stringify({ input: [{ role: 'user', content: 'the prompt' }] })) + await send(proxy, '/responses', 'POST', prompt) + await send(proxy, '/models', 'GET') + await proxy.flushMirror() + + expect(Buffer.from(sidecarLine(sidecar).body_b64 as string, 'base64').equals(prompt)).toBe(true) +}) + +it('removes the sidecar rather than keep an older prompt when the newest body is over the cap', async () => { + const { proxy, sidecar } = await startProxy() + await send(proxy, '/responses', 'POST', Buffer.from('{"input":"older prompt"}')) + await proxy.flushMirror() + expect(existsSync(sidecar)).toBe(true) + + await send(proxy, '/responses', 'POST', Buffer.alloc(LATEST_REQUEST_BODY_CAP + 1, 0x20)) + await proxy.flushMirror() + expect(existsSync(sidecar)).toBe(false) +}) diff --git a/src/proxy/responsesProxy.ts b/src/proxy/responsesProxy.ts index b6355c2..010c634 100644 --- a/src/proxy/responsesProxy.ts +++ b/src/proxy/responsesProxy.ts @@ -7,6 +7,7 @@ import { homedir } from 'os' import { join } from 'path' import { EventsMirror } from './eventsMirror.js' +import { LatestRequestBodySidecar } from './latestRequestBody.js' import { decompressZstdBounded } from './zstd.js' // Local HTTP proxy for Codex's Responses API. @@ -392,6 +393,9 @@ export class ResponsesProxy extends EventEmitter { // bounded async stream; see eventsMirror.ts for why it is no longer a // synchronous append per event (agent-code#372). private readonly mirror: EventsMirror | null + // The newest Responses request body, next to the events file, so a debug + // bundle has the prompt even when the events tail does not (latestRequestBody.ts). + private readonly latestBody: LatestRequestBodySidecar | null constructor( info: CodexResponsesProxyInfo, @@ -403,6 +407,8 @@ export class ResponsesProxy extends EventEmitter { this.mirror = eventsFile ? new EventsMirror(eventsFile, { maxFileBytes: mirrorOptions.eventsFileMaxBytes }) : null + // Same opt-in as the mirror: no events file, no sidecar (agent-code#1336). + this.latestBody = eventsFile ? new LatestRequestBodySidecar(eventsFile) : null } // Override emit to mirror events into the on-disk JSONL. Done at @@ -417,9 +423,10 @@ export class ResponsesProxy extends EventEmitter { return super.emit(event, ...args) } - /** Resolves once every mirrored event so far has reached the OS. */ + /** Resolves once every mirrored event, and the latest-request-body sidecar, + * has reached the OS. */ flushMirror(): Promise { - return this.mirror?.flush() ?? Promise.resolve() + return Promise.all([this.mirror?.flush(), this.latestBody?.flush()]).then(() => undefined) } static async create(options: Options = {}): Promise { @@ -526,6 +533,7 @@ export class ResponsesProxy extends EventEmitter { await this.stopServer() } finally { await this.mirror?.close() + await this.latestBody?.flush() } } @@ -698,6 +706,7 @@ export class ResponsesProxy extends EventEmitter { ...(bodyB64 !== undefined ? { body_b64: bodyB64 } : {}), ...(requestShape !== null ? { request_shape: requestShape } : {}), }) + this.latestBody?.record(requestId, endpoint, body) const abort = new AbortController() const headersTimer = setTimeout(() => { From a54cfe760ee1feb4ea17aa43aa9f0922cd3d934d Mon Sep 17 00:00:00 2001 From: Julius Olsson Date: Sun, 27 Sep 2026 03:08:57 -0700 Subject: [PATCH 2/5] fix(proxy): latest body is never stale, skips title turns, encodes in slices (review a) Co-Authored-By: Claude Opus 5.5 --- API.md | 2 +- SECURITY.md | 10 +- docs/plans/2026-09-27-latest-request-body.md | 13 ++ src/proxy/latestRequestBody.test.ts | 72 +++++++++ src/proxy/latestRequestBody.ts | 146 ++++++++++++++----- src/proxy/responsesProxy.latestBody.test.ts | 27 ++++ src/proxy/responsesProxy.ts | 17 ++- 7 files changed, 244 insertions(+), 43 deletions(-) create mode 100644 src/proxy/latestRequestBody.test.ts diff --git a/API.md b/API.md index 0ec7675..abdd347 100644 --- a/API.md +++ b/API.md @@ -1601,7 +1601,7 @@ static ResponsesProxy.create(options?: { | --- | --- | --- | --- | | `authMode` | `CodexAuthMode` (`'apikey' \| 'chatgpt'`) | auto-detected | Which auth path Codex uses. Auto-detection reads `~/.codex/auth.json`; absent → `'apikey'`. | | `upstreamBaseUrl` | `string` | derived from `authMode` | The real upstream. Defaults: `apikey` → `https://api.openai.com/v1`, `chatgpt` → `https://chatgpt.com/backend-api/codex`. | -| `eventsFile` | `string` | unset | If set, every emitted `event` is also written as a JSON line to this file (forensic mirror; `Buffer` payloads are inlined as `{ _buffer_b64 }`. Dumps written before agent-code#372's fix hold `{"type":"Buffer","data":[…]}` byte arrays instead, so a reader of older files must accept both). Written asynchronously through one ordered stream, never on the emitting call. When a line would push the total unwritten bytes (across the live and any rotated-out file) past 16 MiB, it is dropped and counted. A `{kind:'mirror-dropped', droppedEvents, droppedBytes}` line precedes the next written line, or is written at `stop()` if no line follows. Markers obey the same bounds: with a queue cap smaller than one marker (only ever a test setting), the marker is not written and the drops stay counted in the mirror's stats. Also keeps the newest `responses*` request body in `latest-request-body.json` beside the file: one line, `{kind:'request-body-latest', requestId, endpoint, body_b64}`, the name and kind of the Claude addon's sidecar. It is replaced by each newer body, never by a bodiless request. It is removed when the newest body is over 16 MiB or a write fails, so it is never an older prompt. Writes are asynchronous and ordered. Call `flushMirror()` to wait for the queue and the sidecar; `stop()` flushes and closes them. | +| `eventsFile` | `string` | unset | If set, every emitted `event` is also written as a JSON line to this file (forensic mirror; `Buffer` payloads are inlined as `{ _buffer_b64 }`. Dumps written before agent-code#372's fix hold `{"type":"Buffer","data":[…]}` byte arrays instead, so a reader of older files must accept both). Written asynchronously through one ordered stream, never on the emitting call. When a line would push the total unwritten bytes (across the live and any rotated-out file) past 16 MiB, it is dropped and counted. A `{kind:'mirror-dropped', droppedEvents, droppedBytes}` line precedes the next written line, or is written at `stop()` if no line follows. Markers obey the same bounds: with a queue cap smaller than one marker (only ever a test setting), the marker is not written and the drops stay counted in the mirror's stats. Also keeps the newest main-turn `responses*` request body in `latest-request-body.json` beside the file: one line, `{kind:'request-body-latest', requestId, endpoint, body_b64}` (raw on-wire bytes, zstd on current Codex), the name and kind of the Claude addon's sidecar. Requests with an output schema (`request_shape.has_output_schema`, e.g. title generation) and bodiless requests never replace it. A newer body removes the old file at once. A write superseded by a newer body never lands, and a body over 16 MiB leaves no file, so the file is never an older prompt. Writes are asynchronous, encoded in 768 KiB slices, and keep at most one pending body. Call `flushMirror()` to wait for the queue and the sidecar; `stop()` flushes and closes them. | | `eventsFileMaxBytes` | `number` | 64 MiB | When the next line would pass this size, the file is renamed to `.1.jsonl` (replacing any earlier one) and a fresh file starts with `{kind:'mirror-rotated', rotatedBytes, rotations, droppedEvents, droppedBytes}`. Every file stays within the cap, markers included, so a run holds at most two files of this size. A line that cannot fit even in a fresh file is dropped and counted. A restart that appends to a file already at the cap rotates on its first event. | `create()` starts listening before resolving. The resolved instance diff --git a/SECURITY.md b/SECURITY.md index ebb7616..e9f2d94 100644 --- a/SECURITY.md +++ b/SECURITY.md @@ -42,11 +42,11 @@ While the proxy is in use: `Authorization` header** so a leaked bundle cannot expose bearer tokens — but it **does** record request bodies (your prompts, base64, capped at 2 MiB) and response bytes. Treat that file as sensitive. -- With `eventsFile` set, the proxy also keeps the newest Responses request - body (up to 16 MiB) in `latest-request-body.json` next to it, so a debug - bundle has the prompt even after the events tail has moved past it - (agent-code#1336). It is the same prompt text as the mirror. Treat it as - sensitive too. +- With `eventsFile` set, the proxy also keeps the newest main-turn Responses + request body (up to 16 MiB) in `latest-request-body.json` next to it, so a + debug bundle has the prompt even after the events tail has moved past it + (agent-code#1336). A body between 2 MiB and 16 MiB is persisted **only** + there: the mirror omits it. Treat that file as sensitive too. The proxy is **opt-in**. If you don't construct it, no interception happens — Codex talks to the upstream directly and the package observes diff --git a/docs/plans/2026-09-27-latest-request-body.md b/docs/plans/2026-09-27-latest-request-body.md index 57caea9..db548b8 100644 --- a/docs/plans/2026-09-27-latest-request-body.md +++ b/docs/plans/2026-09-27-latest-request-body.md @@ -26,3 +26,16 @@ All three are red with the wiring removed. ## Then The app bump carries this to Agent Code. Its reader's WHY comment, which describes only the Claude invariant, gets the Codex one. + +## Review a (round 1) +- **Stale "latest"**, after a crash between write and rename, a failed removal, or a read while a newer write was queued. Fixed with three rules (see the module header): + - `record()` unlinks the real-name file synchronously; + - a superseded write never renames; + - only the newest pending body is kept. + If the directory refuses the unlink, nothing can make the file current; that stays unreported, like every mirror failure. +- **Title generation replaced the main prompt.** Codex 0.157 title turns carry an output schema (`text.format`, `codex_output_schema`). `request_shape` gains `has_output_schema`, and such requests are skipped. A body whose shape cannot be read is still kept. +- **Main-process cost.** Base64 is encoded in 768 KiB slices, one per turn after a drain, instead of 21 MiB in one call. At most one body is pending. +- **SECURITY.md accuracy.** Bodies between 2 MiB and 16 MiB are persisted only in the sidecar. +- **Fresh sessions never reach the file (P1).** This is an app-side selection bug, not a package bug: the bundle asks for `resume-` while a fresh run lives under `shell-`. It is fixed in the app PR that bumps this package, where it is reachable and testable. +- **Tests:** compaction is recorded, the title turn is skipped, a newer record removes the file at once, a superseded write cannot resurrect an older body, and a multi-slice body round-trips. + - Mutations killed: the endpoint narrowing, the schema check, the generation check and the unlink. diff --git a/src/proxy/latestRequestBody.test.ts b/src/proxy/latestRequestBody.test.ts new file mode 100644 index 0000000..62b8265 --- /dev/null +++ b/src/proxy/latestRequestBody.test.ts @@ -0,0 +1,72 @@ +import { existsSync, mkdtempSync, readFileSync, readdirSync, rmSync } from 'node:fs' +import { tmpdir } from 'node:os' +import { join } from 'node:path' + +import { afterEach, expect, it } from 'vitest' + +import { LatestRequestBodySidecar } from './latestRequestBody.js' + +// #70 review a: "never an older prompt" must hold at every instant a bundle +// can read the file, not only once writes settle, and a large body must not +// be encoded in one main-process turn. + +const dirs: string[] = [] +afterEach(() => { for (const dir of dirs.splice(0)) rmSync(dir, { recursive: true, force: true }) }) +function sidecarIn(): LatestRequestBodySidecar { + const dir = mkdtempSync(join(tmpdir(), 'cxh-latest-body-unit-')) + dirs.push(dir) + return new LatestRequestBodySidecar(join(dir, 'proxy-events.jsonl')) +} +const bodyOf = (path: string) => Buffer.from((JSON.parse(readFileSync(path, 'utf8')) as { body_b64: string }).body_b64, 'base64') + +it('removes the older body the moment a newer one is recorded', async () => { + const sidecar = sidecarIn() + sidecar.record('req-1', 'responses', Buffer.from('older prompt')) + await sidecar.flush() + expect(existsSync(sidecar.path)).toBe(true) + + sidecar.record('req-2', 'responses', Buffer.from('newer prompt')) + // Synchronously, before any write: a reader (or a crash) now sees no body, + // never the older one. + expect(existsSync(sidecar.path)).toBe(false) + await sidecar.flush() + expect(bodyOf(sidecar.path).toString()).toBe('newer prompt') +}) + +it('never lets a superseded in-flight write rename over a newer body', async () => { + const sidecar = sidecarIn() + // A large body takes several turns to encode, so the next record lands + // while it is still in flight. + const large = Buffer.alloc(4 * 1024 * 1024, 0x61) + sidecar.record('req-1', 'responses', large) + sidecar.record('req-2', 'responses', Buffer.from('newest prompt')) + await sidecar.flush() + + expect(bodyOf(sidecar.path).toString()).toBe('newest prompt') + // The superseded write left no temp file behind. + expect(readdirSync(join(sidecar.path, '..')).filter(name => name.endsWith('.tmp'))).toEqual([]) +}) + +it('writes a multi-slice body that round-trips byte for byte', async () => { + const sidecar = sidecarIn() + // Not a multiple of the 768 KiB slice, and arbitrary bytes (a zstd frame is + // binary), so slice boundaries and padding are both exercised. + const body = Buffer.from(Array.from({ length: 2 * 1024 * 1024 + 7 }, (_, i) => (i * 131 + 7) % 256)) + sidecar.record('req-1', 'responses', body) + await sidecar.flush() + + expect(bodyOf(sidecar.path).equals(body)).toBe(true) + expect(readFileSync(sidecar.path, 'utf8').split('\n').filter(Boolean)).toHaveLength(1) +}) + +it('does not resurrect an in-flight older body after a newer one was too big to keep', async () => { + // The newer body is over the cap, so nothing replaces the file; only the + // superseded write's own generation check stops it renaming the older body + // back in as "latest". + const sidecar = sidecarIn() + sidecar.record('req-1', 'responses', Buffer.alloc(4 * 1024 * 1024, 0x61)) + sidecar.record('req-2', 'responses', Buffer.alloc(16 * 1024 * 1024 + 1, 0x62)) + await sidecar.flush() + + expect(existsSync(sidecar.path)).toBe(false) +}) diff --git a/src/proxy/latestRequestBody.ts b/src/proxy/latestRequestBody.ts index ff418a6..a9f4a65 100644 --- a/src/proxy/latestRequestBody.ts +++ b/src/proxy/latestRequestBody.ts @@ -1,8 +1,9 @@ -import { readdir, rename, rm, writeFile } from 'fs/promises' +import { createWriteStream, unlinkSync } from 'fs' +import { readdir, rename, rm } from 'fs/promises' import { basename, dirname, join } from 'path' -// The newest Responses request body, kept in a sidecar next to the proxy -// events file (agent-code#1336). +// The newest main-turn Responses request body, kept in a sidecar next to the +// proxy events file (agent-code#1336). // // WHY a sidecar at all: Agent Code's debug bundle carries the last 5 MiB of // `proxy-events.jsonl` (plus the rotated `.1`). On Codex the response chunks @@ -19,74 +20,118 @@ import { basename, dirname, join } from 'path' // `readLatestRequestBody` already appends `/latest-request-body.json` // to the bundle's proxy section for ANY provider. One JSON line with its own // `kind`, so nothing that replays the events file mistakes it for a request. +// The body is the raw on-wire bytes, exactly as the `request` event's +// `body_b64` (zstd-compressed on current Codex; readers detect the frame). // // The INVARIANT differs from Claude's, on purpose. Claude's sidecar holds the // newest body that is NOT in the log (its log omits bodies past a budget). // Codex never omits bodies up to the 2 MiB inline cap, and the loss here is -// the bundle's tail, which the proxy cannot see. So this sidecar always holds -// the newest Responses request body, even when the log also has it. The cost -// is at most one body (bounded below) appearing twice in a bundle. +// the bundle's tail, which the proxy cannot see. So this sidecar holds the +// newest main-turn body, even when the log also has it (at most one body +// appears twice in a bundle). // -// WHY only `responses*` endpoints: they carry the prompt (the main turn, -// `responses/compact`). `/v1/models` is a GET with no body. A turn's later -// requests (tool results) resend the whole input, so "newest" still holds -// every prompt so far. A title side request (0.157 `thread_title.rs`) is a -// Responses request too, so the sidecar can briefly hold one; it carries the -// same user prompt, and the next main-turn request replaces it. +// WHICH requests (the caller decides, see ResponsesProxy): `responses*` +// endpoints with a body, excluding temporary structured turns. Codex 0.157's +// title generation (tui `thread_title.rs`) runs an ephemeral thread whose +// request carries an output schema (`text.format`, name +// `codex_output_schema`); ordinary TUI turns never do. Letting it in replaced +// the main prompt with a 960-byte title prompt (#70 review a). A body whose +// shape cannot be read (not JSON, or a zstd frame over the decode bound) is +// still recorded: an unclassified prompt is better evidence than none. // -// WHY one ordered chain of async writes: the proxy runs on Agent Code's main -// process, where the old synchronous mirror appends were the problem -// (agent-code#372), so no sync I/O. But independent async writes can finish -// out of order and rename an OLDER body over a newer one. Chaining them keeps -// "latest" true. A failed write removes the sidecar, so a reader never sees a -// stale prompt as the current one. Best-effort throughout: forensics must -// never disturb the live proxy. +// "NEVER AN OLDER PROMPT" (#70 review a): three rules make it hold at every +// instant a reader can look, not only after writes settle. +// 1. A newer body removes the real-name file SYNCHRONOUSLY when it is +// recorded. `unlinkSync` is one metadata call, not a data write. From then +// on a reader sees either nothing or the newer body, including after a +// crash anywhere in the write. +// 2. A write that has been superseded by a newer record never renames: it +// checks its generation right before `rename` and discards its temp file. +// 3. Only the newest pending body is kept. Older pending bodies are dropped +// unwritten, so memory holds at most one body in flight and one pending. +// If the directory refuses the unlink (permissions), nothing here can make the +// file current; that failure is not reported, as for every mirror failure. +// +// MAIN-PROCESS COST (#70 review a): the proxy runs on Agent Code's main +// process. A 16 MiB body is 21.3 MiB of base64, measured at up to 376 ms when +// encoded in one call. It is encoded in slices of ENCODE_SLICE_BYTES, each in +// its own turn after the stream drains, so no single turn does more than a +// slice. export const LATEST_REQUEST_BODY_FILE_NAME = 'latest-request-body.json' // Same cap as the Claude addon's sidecar (`_LATEST_BODY_CAP`). Larger than the // 2 MiB inline cap on purpose: a body too big to inline is exactly the one the -// events file cannot show at all. +// events file cannot show at all. Such bodies are persisted ONLY here +// (SECURITY.md says so). export const LATEST_REQUEST_BODY_CAP = 16 * 1024 * 1024 +// A multiple of 3, so every slice but the last encodes to base64 without +// padding and the slices concatenate into one valid base64 string. +const ENCODE_SLICE_BYTES = 3 * 256 * 1024 + +type Pending = { generation: number; requestId: string; endpoint: string; body: Buffer } + export class LatestRequestBodySidecar { readonly path: string - private chain: Promise = Promise.resolve() - private writes = 0 + private generation = 0 + private pending: Pending | null = null + private inFlight: Promise | null = null + private idleWaiters: Array<() => void> = [] constructor(eventsFile: string) { this.path = join(dirname(eventsFile), LATEST_REQUEST_BODY_FILE_NAME) } - record(requestId: string, endpoint: string, body: Buffer | undefined): void { - if (!endpoint.startsWith('responses') || !body || body.length === 0) return - const write = this.writes++ - const line = body.length > LATEST_REQUEST_BODY_CAP + record(requestId: string, endpoint: string, body: Buffer): void { + this.generation += 1 + // Rule 1: from now on the old body is not "latest", on disk too. + try { unlinkSync(this.path) } catch { /* absent, or the directory refuses */ } + this.pending = body.length > LATEST_REQUEST_BODY_CAP ? null - : JSON.stringify({ kind: 'request-body-latest', requestId, endpoint, body_b64: body.toString('base64') }) + '\n' - this.chain = this.chain.then(() => (line === null ? this.remove() : this.replace(line, write))) + : { generation: this.generation, requestId, endpoint, body } + this.kick() } - /** Resolves once every recorded body has been written or removed. */ + /** Resolves once every recorded body has been written, or dropped. */ flush(): Promise { - return this.chain + if (!this.inFlight && !this.pending) return Promise.resolve() + return new Promise(resolve => this.idleWaiters.push(resolve)) + } + + private kick(): void { + if (this.inFlight) return + const next = this.pending + this.pending = null + if (!next) { + for (const resolve of this.idleWaiters.splice(0)) resolve() + return + } + this.inFlight = this.write(next).finally(() => { + this.inFlight = null + this.kick() + }) } - private async replace(line: string, write: number): Promise { + private async write(entry: Pending): Promise { // A per-write temp name, so a crash mid-write never leaves a half body // under the real name. - const temp = `${this.path}.${process.pid}.${write}.tmp` + const temp = `${this.path}.${process.pid}.${entry.generation}.tmp` try { - await writeFile(temp, line, 'utf-8') + await writeEncoded(temp, entry) + // Rule 2. + if (entry.generation !== this.generation) { + await rm(temp, { force: true }) + return + } await rename(temp, this.path) } catch { await rm(temp, { force: true }).catch(() => {}) - await this.remove() + await this.removeTempFiles() } } - private async remove(): Promise { - await rm(this.path, { force: true }).catch(() => {}) + private async removeTempFiles(): Promise { // Temp files a crash between write and rename left behind. try { const prefix = `${basename(this.path)}.` @@ -96,3 +141,32 @@ export class LatestRequestBodySidecar { } catch { /* directory gone: nothing to clean */ } } } + +async function writeEncoded(path: string, entry: Pending): Promise { + const stream = createWriteStream(path, { encoding: 'utf-8' }) + const done = new Promise((resolve, reject) => { + stream.once('error', reject) + stream.once('finish', resolve) + }) + // Observed here so an error while a slice is still being encoded is not an + // unhandled rejection; it is re-raised by the race below and the final await. + done.catch(() => {}) + const put = async (text: string): Promise => { + // Raced with `done`: a stream that errors never emits 'drain'. + if (!stream.write(text)) await Promise.race([new Promise(resolve => stream.once('drain', resolve)), done]) + // Yield so the next slice's encoding runs in its own turn. + await new Promise(resolve => setImmediate(resolve)) + } + try { + const head = JSON.stringify({ kind: 'request-body-latest', requestId: entry.requestId, endpoint: entry.endpoint }) + await put(`${head.slice(0, -1)},"body_b64":"`) + for (let offset = 0; offset < entry.body.length; offset += ENCODE_SLICE_BYTES) { + await put(entry.body.subarray(offset, offset + ENCODE_SLICE_BYTES).toString('base64')) + } + stream.end('"}\n') + } catch (error) { + stream.destroy() + throw error + } + await done +} diff --git a/src/proxy/responsesProxy.latestBody.test.ts b/src/proxy/responsesProxy.latestBody.test.ts index 55f5834..89a87e3 100644 --- a/src/proxy/responsesProxy.latestBody.test.ts +++ b/src/proxy/responsesProxy.latestBody.test.ts @@ -97,3 +97,30 @@ it('removes the sidecar rather than keep an older prompt when the newest body is await proxy.flushMirror() expect(existsSync(sidecar)).toBe(false) }) + +// #70 review a. Codex 0.157 title generation runs an ephemeral thread whose +// request carries an output schema (tui `thread_title.rs` through +// codex-api `create_text_param_for_request`: text.format, name +// `codex_output_schema`). It must not replace the main prompt. +it('keeps the main prompt when a title-generation request follows it', async () => { + const { proxy, sidecar } = await startProxy() + const prompt = Buffer.from(JSON.stringify({ input: [{ role: 'user', content: 'the main prompt' }], tools: [{ type: 'function' }] })) + const title = Buffer.from(JSON.stringify({ + input: [{ role: 'user', content: 'Generate a concise, single-line task title' }], + text: { format: { type: 'json_schema', strict: true, name: 'codex_output_schema', schema: {} } }, + })) + await send(proxy, '/responses', 'POST', prompt) + await send(proxy, '/responses', 'POST', title) + await proxy.flushMirror() + + expect(Buffer.from(sidecarLine(sidecar).body_b64 as string, 'base64').equals(prompt)).toBe(true) +}) + +it('records a compaction request, which also carries the conversation', async () => { + const { proxy, sidecar } = await startProxy() + const compact = Buffer.from(JSON.stringify({ input: [{ role: 'user', content: 'compact me' }] })) + await send(proxy, '/responses/compact', 'POST', compact) + await proxy.flushMirror() + + expect(sidecarLine(sidecar)).toMatchObject({ endpoint: 'responses/compact' }) +}) diff --git a/src/proxy/responsesProxy.ts b/src/proxy/responsesProxy.ts index 010c634..31b10b8 100644 --- a/src/proxy/responsesProxy.ts +++ b/src/proxy/responsesProxy.ts @@ -215,6 +215,11 @@ export type CodexRequestShape = { input_items_count: number | null tools_count: number | null has_reasoning: boolean + /** True when the request asks for structured output (`text.format`, an + * output schema). Codex sets it only on temporary structured turns, such + * as 0.157 title generation (tui `thread_title.rs`); ordinary turns never + * carry one. The latest-request-body sidecar skips these (#70 review a). */ + has_output_schema: boolean client_metadata: { thread_id: string | null session_id: string | null @@ -248,6 +253,8 @@ function extractRequestShape(body: Buffer): CodexRequestShape | null { const input = Array.isArray(obj.input) ? obj.input.length : null const tools = Array.isArray(obj.tools) ? obj.tools.length : null const hasReasoning = obj.reasoning != null && typeof obj.reasoning === 'object' + const text = obj.text && typeof obj.text === 'object' ? obj.text as Record : null + const hasOutputSchema = text?.format != null && typeof text.format === 'object' const metadataObject = obj.client_metadata const metadata = metadataObject && typeof metadataObject === 'object' ? metadataObject as Record @@ -269,6 +276,7 @@ function extractRequestShape(body: Buffer): CodexRequestShape | null { input_items_count: input, tools_count: tools, has_reasoning: hasReasoning, + has_output_schema: hasOutputSchema, client_metadata: clientMetadata, provider_session_id: threadId !== null && threadId === sessionId ? threadId : null, @@ -706,7 +714,14 @@ export class ResponsesProxy extends EventEmitter { ...(bodyB64 !== undefined ? { body_b64: bodyB64 } : {}), ...(requestShape !== null ? { request_shape: requestShape } : {}), }) - this.latestBody?.record(requestId, endpoint, body) + // The newest MAIN-turn prompt, for debug bundles whose events tail no + // longer holds this request (agent-code#1336). Temporary structured turns + // (title generation) are skipped: see latestRequestBody.ts for why, and + // for why a body whose shape could not be read is still kept (a prompt we + // cannot classify is better evidence than none). + if (body && body.length > 0 && endpoint.startsWith('responses') && requestShape?.has_output_schema !== true) { + this.latestBody?.record(requestId, endpoint, body) + } const abort = new AbortController() const headersTimer = setTimeout(() => { From aff509f2102b394f4260f92aa4f860dcc74adbe5 Mon Sep 17 00:00:00 2001 From: Julius Olsson Date: Sun, 27 Sep 2026 03:12:33 -0700 Subject: [PATCH 3/5] fix(proxy): enforce the sidecar generation at the commit boundary (steering q96) Co-Authored-By: Claude Opus 5.5 --- docs/plans/2026-09-27-latest-request-body.md | 7 +- .../latestRequestBody.commitFence.test.ts | 91 +++++++++++++++++++ src/proxy/latestRequestBody.ts | 23 +++-- 3 files changed, 114 insertions(+), 7 deletions(-) create mode 100644 src/proxy/latestRequestBody.commitFence.test.ts diff --git a/docs/plans/2026-09-27-latest-request-body.md b/docs/plans/2026-09-27-latest-request-body.md index db548b8..3a07163 100644 --- a/docs/plans/2026-09-27-latest-request-body.md +++ b/docs/plans/2026-09-27-latest-request-body.md @@ -30,7 +30,7 @@ The app bump carries this to Agent Code. Its reader's WHY comment, which describ ## Review a (round 1) - **Stale "latest"**, after a crash between write and rename, a failed removal, or a read while a newer write was queued. Fixed with three rules (see the module header): - `record()` unlinks the real-name file synchronously; - - a superseded write never renames; + - a superseded write never renames: the generation check and `renameSync` run in one synchronous turn (steering q96: an awaited async rename after the check let `record(B)` unlink the file and A's late rename restore it); - only the newest pending body is kept. If the directory refuses the unlink, nothing can make the file current; that stays unreported, like every mirror failure. - **Title generation replaced the main prompt.** Codex 0.157 title turns carry an output schema (`text.format`, `codex_output_schema`). `request_shape` gains `has_output_schema`, and such requests are skipped. A body whose shape cannot be read is still kept. @@ -39,3 +39,8 @@ The app bump carries this to Agent Code. Its reader's WHY comment, which describ - **Fresh sessions never reach the file (P1).** This is an app-side selection bug, not a package bug: the bundle asks for `resume-` while a fresh run lives under `shell-`. It is fixed in the app PR that bumps this package, where it is reachable and testable. - **Tests:** compaction is recorded, the title turn is skipped, a newer record removes the file at once, a superseded write cannot resurrect an older body, and a multi-slice body round-trips. - Mutations killed: the endpoint narrowing, the schema check, the generation check and the unlink. + +## Steering q96 +- **A late async rename restored an older body.** The first round checked the generation, then awaited an async `rename`. A `record(B)` in that gap unlinked the public file, and A's rename then restored A while B was still being written. +- **Fix: commit in one turn.** The commit is now `renameSync` in the same turn as the check. It is one metadata call; body bytes are still written asynchronously in slices. +- **Test: `latestRequestBody.commitFence.test.ts`.** It holds every async `fs/promises` rename at a gate and spies `renameSync`. It records B while A's commit is pending, releases A, and reads the public file before B commits: the file is absent or B, never A. It was red on `a54cfe7` (A restored). diff --git a/src/proxy/latestRequestBody.commitFence.test.ts b/src/proxy/latestRequestBody.commitFence.test.ts new file mode 100644 index 0000000..d914bd7 --- /dev/null +++ b/src/proxy/latestRequestBody.commitFence.test.ts @@ -0,0 +1,91 @@ +import { existsSync, mkdtempSync, readFileSync, rmSync } from 'node:fs' +import { tmpdir } from 'node:os' +import { join } from 'node:path' + +import { afterEach, expect, it, vi } from 'vitest' + +// steering q96 / #70: the "never an older prompt" invariant must hold at the +// COMMIT boundary. The first fix checked the generation and THEN awaited an +// async rename. `record(B)` could run while that rename was in flight: it +// unlinked the public file, then A's rename landed and restored A as +// "latest" while B was still being written. A bundle read or a crash in that +// window labelled an older prompt current. +// +// This holds every async `rename` from fs/promises at a gate and spies the +// synchronous one, so the test can record B exactly while A's commit is +// pending, whichever API the implementation commits with. +const gate = vi.hoisted(() => ({ + held: [] as Array<() => void>, + commits: 0, +})) + +vi.mock('fs/promises', async importOriginal => { + const actual = await importOriginal() + return { + ...actual, + rename: async (from: string, to: string) => { + gate.commits += 1 + await new Promise(resolve => gate.held.push(resolve)) + return actual.rename(from, to) + }, + } +}) + +vi.mock('fs', async importOriginal => { + const actual = await importOriginal() + return { + ...actual, + renameSync: (from: string, to: string) => { + gate.commits += 1 + return actual.renameSync(from, to) + }, + } +}) + +const { LatestRequestBodySidecar } = await import('./latestRequestBody.js') + +const dirs: string[] = [] +afterEach(() => { + for (const release of gate.held.splice(0)) release() + gate.commits = 0 + for (const dir of dirs.splice(0)) rmSync(dir, { recursive: true, force: true }) +}) + +async function until(predicate: () => boolean): Promise { + const deadline = Date.now() + 5_000 + while (!predicate()) { + if (Date.now() > deadline) throw new Error('timed out') + await new Promise(resolve => setTimeout(resolve, 1)) + } +} + +it('never restores an older body when a newer one is recorded during its commit', async () => { + const dir = mkdtempSync(join(tmpdir(), 'cxh-latest-body-fence-')) + dirs.push(dir) + const sidecar = new LatestRequestBodySidecar(join(dir, 'proxy-events.jsonl')) + + sidecar.record('req-a', 'responses', Buffer.from('OLDER prompt A')) + // A has written its temp file and reached its commit. + await until(() => gate.commits === 1) + + // B arrives now. It is large, so its own write spans many turns and cannot + // commit within the window below. + sidecar.record('req-b', 'responses', Buffer.alloc(4 * 1024 * 1024, 0x62)) + for (const release of gate.held.splice(0)) release() + // Let A's commit, if it was still pending, land. + await new Promise(resolve => setTimeout(resolve, 20)) + + // Before B commits, the public file is absent or B: never A. + if (existsSync(sidecar.path)) { + expect(readFileSync(sidecar.path, 'utf8')).not.toContain(Buffer.from('OLDER prompt A').toString('base64')) + } + + // Let every remaining write finish, releasing any rename that is held. + const releaser = setInterval(() => { for (const release of gate.held.splice(0)) release() }, 1) + try { + await sidecar.flush() + } finally { + clearInterval(releaser) + } + expect(readFileSync(sidecar.path, 'utf8')).toContain('"requestId":"req-b"') +}) diff --git a/src/proxy/latestRequestBody.ts b/src/proxy/latestRequestBody.ts index a9f4a65..f8066c7 100644 --- a/src/proxy/latestRequestBody.ts +++ b/src/proxy/latestRequestBody.ts @@ -1,5 +1,5 @@ -import { createWriteStream, unlinkSync } from 'fs' -import { readdir, rename, rm } from 'fs/promises' +import { createWriteStream, renameSync, unlinkSync } from 'fs' +import { readdir, rm } from 'fs/promises' import { basename, dirname, join } from 'path' // The newest main-turn Responses request body, kept in a sidecar next to the @@ -46,14 +46,18 @@ import { basename, dirname, join } from 'path' // on a reader sees either nothing or the newer body, including after a // crash anywhere in the write. // 2. A write that has been superseded by a newer record never renames: it -// checks its generation right before `rename` and discards its temp file. +// checks its generation and publishes with `renameSync` in the same +// synchronous turn, so no record() can slip between check and commit +// (steering q96), and a superseded write discards its temp file. // 3. Only the newest pending body is kept. Older pending bodies are dropped // unwritten, so memory holds at most one body in flight and one pending. // If the directory refuses the unlink (permissions), nothing here can make the // file current; that failure is not reported, as for every mirror failure. // // MAIN-PROCESS COST (#70 review a): the proxy runs on Agent Code's main -// process. A 16 MiB body is 21.3 MiB of base64, measured at up to 376 ms when +// process. The only synchronous calls are two metadata operations (the +// unlink in record, the rename at commit); every data byte is written +// asynchronously. A 16 MiB body is 21.3 MiB of base64, measured at up to 376 ms when // encoded in one call. It is encoded in slices of ENCODE_SLICE_BYTES, each in // its own turn after the stream drains, so no single turn does more than a // slice. @@ -119,12 +123,19 @@ export class LatestRequestBodySidecar { const temp = `${this.path}.${process.pid}.${entry.generation}.tmp` try { await writeEncoded(temp, entry) - // Rule 2. + // Rule 2, enforced AT the commit (steering q96). The generation check + // and the publish run in ONE synchronous turn, so no `record()` can run + // between them. The first version checked, then awaited an async + // `rename`: a `record(B)` in that gap unlinked the public file, and A's + // rename then landed and restored A as "latest" while B was still being + // written. `renameSync` is one metadata call (like the `unlinkSync` in + // record), not a data write; the body itself is still written + // asynchronously in slices above, so a large body never blocks a turn. if (entry.generation !== this.generation) { await rm(temp, { force: true }) return } - await rename(temp, this.path) + renameSync(temp, this.path) } catch { await rm(temp, { force: true }).catch(() => {}) await this.removeTempFiles() From d56589b259917cfdc417623581ddb63f7e3a65ea Mon Sep 17 00:00:00 2001 From: Julius Olsson Date: Sun, 27 Sep 2026 03:38:34 -0700 Subject: [PATCH 4/5] fix(proxy): subagent requests never replace the latest main-turn body (review b) Co-Authored-By: Claude Opus 5.5 --- API.md | 2 +- SECURITY.md | 4 +- docs/plans/2026-09-27-latest-request-body.md | 6 +++ src/proxy/latestRequestBody.ts | 4 +- src/proxy/responsesProxy.latestBody.test.ts | 46 ++++++++++++++++++++ src/proxy/responsesProxy.ts | 11 ++++- 6 files changed, 69 insertions(+), 4 deletions(-) diff --git a/API.md b/API.md index abdd347..d00fec6 100644 --- a/API.md +++ b/API.md @@ -1601,7 +1601,7 @@ static ResponsesProxy.create(options?: { | --- | --- | --- | --- | | `authMode` | `CodexAuthMode` (`'apikey' \| 'chatgpt'`) | auto-detected | Which auth path Codex uses. Auto-detection reads `~/.codex/auth.json`; absent → `'apikey'`. | | `upstreamBaseUrl` | `string` | derived from `authMode` | The real upstream. Defaults: `apikey` → `https://api.openai.com/v1`, `chatgpt` → `https://chatgpt.com/backend-api/codex`. | -| `eventsFile` | `string` | unset | If set, every emitted `event` is also written as a JSON line to this file (forensic mirror; `Buffer` payloads are inlined as `{ _buffer_b64 }`. Dumps written before agent-code#372's fix hold `{"type":"Buffer","data":[…]}` byte arrays instead, so a reader of older files must accept both). Written asynchronously through one ordered stream, never on the emitting call. When a line would push the total unwritten bytes (across the live and any rotated-out file) past 16 MiB, it is dropped and counted. A `{kind:'mirror-dropped', droppedEvents, droppedBytes}` line precedes the next written line, or is written at `stop()` if no line follows. Markers obey the same bounds: with a queue cap smaller than one marker (only ever a test setting), the marker is not written and the drops stay counted in the mirror's stats. Also keeps the newest main-turn `responses*` request body in `latest-request-body.json` beside the file: one line, `{kind:'request-body-latest', requestId, endpoint, body_b64}` (raw on-wire bytes, zstd on current Codex), the name and kind of the Claude addon's sidecar. Requests with an output schema (`request_shape.has_output_schema`, e.g. title generation) and bodiless requests never replace it. A newer body removes the old file at once. A write superseded by a newer body never lands, and a body over 16 MiB leaves no file, so the file is never an older prompt. Writes are asynchronous, encoded in 768 KiB slices, and keep at most one pending body. Call `flushMirror()` to wait for the queue and the sidecar; `stop()` flushes and closes them. | +| `eventsFile` | `string` | unset | If set, every emitted `event` is also written as a JSON line to this file (forensic mirror; `Buffer` payloads are inlined as `{ _buffer_b64 }`. Dumps written before agent-code#372's fix hold `{"type":"Buffer","data":[…]}` byte arrays instead, so a reader of older files must accept both). Written asynchronously through one ordered stream, never on the emitting call. When a line would push the total unwritten bytes (across the live and any rotated-out file) past 16 MiB, it is dropped and counted. A `{kind:'mirror-dropped', droppedEvents, droppedBytes}` line precedes the next written line, or is written at `stop()` if no line follows. Markers obey the same bounds: with a queue cap smaller than one marker (only ever a test setting), the marker is not written and the drops stay counted in the mirror's stats. Also keeps the newest main-turn `responses*` request body in `latest-request-body.json` beside the file: one line, `{kind:'request-body-latest', requestId, endpoint, body_b64}` (raw on-wire bytes, zstd on current Codex), the name and kind of the Claude addon's sidecar. Requests with an output schema (`request_shape.has_output_schema`, e.g. title generation), subagent requests (`x-openai-subagent` other than `compact`), and bodiless requests never replace it. A newer body removes the old file at once. A write superseded by a newer body never lands, and a body over 16 MiB leaves no file, so the file is never an older prompt. Writes are asynchronous, encoded in 768 KiB slices, and keep at most one pending body. Call `flushMirror()` to wait for the queue and the sidecar; `stop()` flushes and closes them. | | `eventsFileMaxBytes` | `number` | 64 MiB | When the next line would pass this size, the file is renamed to `.1.jsonl` (replacing any earlier one) and a fresh file starts with `{kind:'mirror-rotated', rotatedBytes, rotations, droppedEvents, droppedBytes}`. Every file stays within the cap, markers included, so a run holds at most two files of this size. A line that cannot fit even in a fresh file is dropped and counted. A restart that appends to a file already at the cap rotates on its first event. | `create()` starts listening before resolving. The resolved instance diff --git a/SECURITY.md b/SECURITY.md index e9f2d94..bf1c763 100644 --- a/SECURITY.md +++ b/SECURITY.md @@ -46,7 +46,9 @@ While the proxy is in use: request body (up to 16 MiB) in `latest-request-body.json` next to it, so a debug bundle has the prompt even after the events tail has moved past it (agent-code#1336). A body between 2 MiB and 16 MiB is persisted **only** - there: the mirror omits it. Treat that file as sensitive too. + there: the mirror omits it. A crash between writing and renaming can also + leave a `latest-request-body.json...tmp` copy beside it. Treat all + of these as sensitive. The proxy is **opt-in**. If you don't construct it, no interception happens — Codex talks to the upstream directly and the package observes diff --git a/docs/plans/2026-09-27-latest-request-body.md b/docs/plans/2026-09-27-latest-request-body.md index 3a07163..1ad6bcc 100644 --- a/docs/plans/2026-09-27-latest-request-body.md +++ b/docs/plans/2026-09-27-latest-request-body.md @@ -44,3 +44,9 @@ The app bump carries this to Agent Code. Its reader's WHY comment, which describ - **A late async rename restored an older body.** The first round checked the generation, then awaited an async `rename`. A `record(B)` in that gap unlinked the public file, and A's rename then restored A while B was still being written. - **Fix: commit in one turn.** The commit is now `renameSync` in the same turn as the check. It is one metadata call; body bytes are still written asynchronously in slices. - **Test: `latestRequestBody.commitFence.test.ts`.** It holds every async `fs/promises` rename at a gate and spies `renameSync`. It records B while A's commit is pending, releases A, and reads the public file before B commits: the file is absent or B, never A. It was red on `a54cfe7` (A restored). + +## Review b (round 1) +- **Subagent calls replaced the main prompt.** Codex tags every non-main Responses call with `x-openai-subagent` (codex-api `requests/headers.rs`: `review`, `compact`, `thread_spawn`, `memory_consolidation`, or a label). They are now skipped, except `compact`, which carries the main conversation. There are tests for `thread_spawn` and `review`, and for a kept `compact`. +- **16 MiB cap untested.** A body of exactly 16 MiB is now kept, so lowering the cap fails a test. +- **Crash-left temp copies.** They are now listed in `SECURITY.md`. +- **P1 (the app side), no package change.** Covered by agent-code#1399, plus a bump that follows #1366. This PR is "For", not "Fixes", #1336. diff --git a/src/proxy/latestRequestBody.ts b/src/proxy/latestRequestBody.ts index f8066c7..374c52e 100644 --- a/src/proxy/latestRequestBody.ts +++ b/src/proxy/latestRequestBody.ts @@ -35,7 +35,9 @@ import { basename, dirname, join } from 'path' // title generation (tui `thread_title.rs`) runs an ephemeral thread whose // request carries an output schema (`text.format`, name // `codex_output_schema`); ordinary TUI turns never do. Letting it in replaced -// the main prompt with a 960-byte title prompt (#70 review a). A body whose +// the main prompt with a 960-byte title prompt (#70 review a). Subagent calls +// (`x-openai-subagent`, except `compact`) are skipped for the same reason +// (#70 review b). A body whose // shape cannot be read (not JSON, or a zstd frame over the decode bound) is // still recorded: an unclassified prompt is better evidence than none. // diff --git a/src/proxy/responsesProxy.latestBody.test.ts b/src/proxy/responsesProxy.latestBody.test.ts index 89a87e3..a224772 100644 --- a/src/proxy/responsesProxy.latestBody.test.ts +++ b/src/proxy/responsesProxy.latestBody.test.ts @@ -124,3 +124,49 @@ it('records a compaction request, which also carries the conversation', async () expect(sidecarLine(sidecar)).toMatchObject({ endpoint: 'responses/compact' }) }) + +// #70 review b: Codex tags every non-main Responses call with +// x-openai-subagent (codex-api requests/headers.rs). A spawned worker's task +// must not replace the top-level prompt; a compaction must still be kept. +function sendWithSubagent(proxy: ResponsesProxy, path: string, body: Buffer, subagent: string): Promise { + return new Promise((resolve, reject) => { + const client = httpRequest(`${proxy.info.proxyBaseUrl}${path}`, { + method: 'POST', + headers: { 'content-type': 'application/json', 'x-openai-subagent': subagent }, + }, res => { + res.resume() + res.on('end', () => resolve()) + }) + client.on('error', reject) + client.end(body) + }) +} + +it('keeps the main prompt when a spawned subagent sends its own request', async () => { + const { proxy, sidecar } = await startProxy() + const prompt = Buffer.from(JSON.stringify({ input: [{ role: 'user', content: 'the main prompt' }] })) + await send(proxy, '/responses', 'POST', prompt) + await sendWithSubagent(proxy, '/responses', Buffer.from('{"input":"worker task"}'), 'thread_spawn') + await sendWithSubagent(proxy, '/responses', Buffer.from('{"input":"review task"}'), 'review') + await proxy.flushMirror() + + expect(Buffer.from(sidecarLine(sidecar).body_b64 as string, 'base64').equals(prompt)).toBe(true) +}) + +it('still records a compaction tagged as the compact subagent', async () => { + const { proxy, sidecar } = await startProxy() + await send(proxy, '/responses', 'POST', Buffer.from('{"input":"the main prompt"}')) + await sendWithSubagent(proxy, '/responses', Buffer.from('{"input":"compact me"}'), 'compact') + await proxy.flushMirror() + + expect(Buffer.from(sidecarLine(sidecar).body_b64 as string, 'base64').toString()).toBe('{"input":"compact me"}') +}) + +it('keeps a body of exactly the cap', async () => { + const { proxy, sidecar } = await startProxy() + const body = Buffer.alloc(LATEST_REQUEST_BODY_CAP, 0x20) + await send(proxy, '/responses', 'POST', body) + await proxy.flushMirror() + + expect(Buffer.from(sidecarLine(sidecar).body_b64 as string, 'base64').length).toBe(LATEST_REQUEST_BODY_CAP) +}) diff --git a/src/proxy/responsesProxy.ts b/src/proxy/responsesProxy.ts index 31b10b8..107e0aa 100644 --- a/src/proxy/responsesProxy.ts +++ b/src/proxy/responsesProxy.ts @@ -719,7 +719,16 @@ export class ResponsesProxy extends EventEmitter { // (title generation) are skipped: see latestRequestBody.ts for why, and // for why a body whose shape could not be read is still kept (a prompt we // cannot classify is better evidence than none). - if (body && body.length > 0 && endpoint.startsWith('responses') && requestShape?.has_output_schema !== true) { + // Subagent calls are skipped too (#70 review b): Codex tags every + // non-main Responses call with `x-openai-subagent` (codex-api + // requests/headers.rs: review, compact, thread_spawn, + // memory_consolidation, or a custom label). A spawned worker's task has + // no output schema, so without this it replaced the top-level prompt in + // multi-agent sessions. `compact` is kept: a compaction request carries + // the main conversation, and it is what a bundle after compaction needs. + const subagent = req.headers['x-openai-subagent'] + const auxiliary = subagent !== undefined && subagent !== 'compact' + if (body && body.length > 0 && endpoint.startsWith('responses') && requestShape?.has_output_schema !== true && !auxiliary) { this.latestBody?.record(requestId, endpoint, body) } From 880c6d8d2fff1f57edf068e6e71a3dd75d20bed9 Mon Sep 17 00:00:00 2001 From: Julius Olsson Date: Sun, 27 Sep 2026 06:09:30 -0700 Subject: [PATCH 5/5] test(proxy): pin the endpoint filter; sweep crash-left temps on success; honest comments (review c) Co-Authored-By: Claude Opus 5.5 --- docs/plans/2026-09-27-latest-request-body.md | 6 ++++++ src/proxy/latestRequestBody.test.ts | 12 ++++++++++++ src/proxy/latestRequestBody.ts | 4 ++++ src/proxy/responsesProxy.latestBody.test.ts | 15 +++++++++++++++ src/proxy/responsesProxy.ts | 15 ++++++++++----- 5 files changed, 47 insertions(+), 5 deletions(-) diff --git a/docs/plans/2026-09-27-latest-request-body.md b/docs/plans/2026-09-27-latest-request-body.md index 1ad6bcc..1e783cc 100644 --- a/docs/plans/2026-09-27-latest-request-body.md +++ b/docs/plans/2026-09-27-latest-request-body.md @@ -50,3 +50,9 @@ The app bump carries this to Agent Code. Its reader's WHY comment, which describ - **16 MiB cap untested.** A body of exactly 16 MiB is now kept, so lowering the cap fails a test. - **Crash-left temp copies.** They are now listed in `SECURITY.md`. - **P1 (the app side), no package change.** Covered by agent-code#1399, plus a bump that follows #1366. This PR is "For", not "Fixes", #1336. + +## Review c +- **The endpoint filter was unpinned, and my "mutations killed" claim for it was false.** The `/models` test is bodiless, so the filter was never consulted. A new test sends bodied `/memories/trace_summarize` and `/alpha/search` POSTs after a main prompt, and the prompt must stay. Replacing the filter with `true` now fails it. +- **`has_output_schema` is a heuristic, not a title detector.** Output schemas are a per-turn option upstream. Agent Code sends none on main turns, and a structured main turn would leave the sidecar one turn behind. The comment now says so. +- **Crash-left temp files were cleaned only after a failure.** They are now also swept after a successful commit, with a test. +- **Subagent label list in the comment.** Corrected to upstream's (`collab_spawn`, `guardian`, …). The rule itself is label-agnostic. diff --git a/src/proxy/latestRequestBody.test.ts b/src/proxy/latestRequestBody.test.ts index 62b8265..4c9117f 100644 --- a/src/proxy/latestRequestBody.test.ts +++ b/src/proxy/latestRequestBody.test.ts @@ -70,3 +70,15 @@ it('does not resurrect an in-flight older body after a newer one was too big to expect(existsSync(sidecar.path)).toBe(false) }) + +it('sweeps a crash-left temp file on the next successful commit', async () => { + // #70 review c: a crash between writing and renaming leaves a temp file + // holding a full prompt. It used to be cleaned only after a failed write. + const sidecar = sidecarIn() + const { writeFileSync } = await import('node:fs') + writeFileSync(`${sidecar.path}.99999.7.tmp`, 'crash-left prompt body') + sidecar.record('req-1', 'responses', Buffer.from('prompt')) + await sidecar.flush() + expect(readdirSync(join(sidecar.path, '..')).filter(name => name.endsWith('.tmp'))).toEqual([]) + expect(bodyOf(sidecar.path).toString()).toBe('prompt') +}) diff --git a/src/proxy/latestRequestBody.ts b/src/proxy/latestRequestBody.ts index 374c52e..b1de04f 100644 --- a/src/proxy/latestRequestBody.ts +++ b/src/proxy/latestRequestBody.ts @@ -138,6 +138,10 @@ export class LatestRequestBodySidecar { return } renameSync(temp, this.path) + // Temp files a crash left between write and rename hold full prompt + // bodies; sweep them on a successful commit too, not only after a + // failure (#70 review c). Not awaited in the commit's turn. + await this.removeTempFiles() } catch { await rm(temp, { force: true }).catch(() => {}) await this.removeTempFiles() diff --git a/src/proxy/responsesProxy.latestBody.test.ts b/src/proxy/responsesProxy.latestBody.test.ts index a224772..61ef2f4 100644 --- a/src/proxy/responsesProxy.latestBody.test.ts +++ b/src/proxy/responsesProxy.latestBody.test.ts @@ -170,3 +170,18 @@ it('keeps a body of exactly the cap', async () => { expect(Buffer.from(sidecarLine(sidecar).body_b64 as string, 'base64').length).toBe(LATEST_REQUEST_BODY_CAP) }) + +// #70 review c: the endpoint filter is the only guard against auxiliary +// POSTs with a body (memory summarization, the web-search sideband) replacing +// the prompt. Nothing pinned it: `/models` is bodiless, so it is skipped +// before the endpoint is ever consulted. +it('keeps the main prompt when a bodied non-Responses request follows it', async () => { + const { proxy, sidecar } = await startProxy() + const prompt = Buffer.from(JSON.stringify({ input: [{ role: 'user', content: 'the main prompt' }] })) + await send(proxy, '/responses', 'POST', prompt) + await send(proxy, '/memories/trace_summarize', 'POST', Buffer.from('{"traces":["memory summary input"]}')) + await send(proxy, '/alpha/search', 'POST', Buffer.from('{"query":"a web search"}')) + await proxy.flushMirror() + + expect(Buffer.from(sidecarLine(sidecar).body_b64 as string, 'base64').equals(prompt)).toBe(true) +}) diff --git a/src/proxy/responsesProxy.ts b/src/proxy/responsesProxy.ts index 107e0aa..f87daaf 100644 --- a/src/proxy/responsesProxy.ts +++ b/src/proxy/responsesProxy.ts @@ -216,9 +216,13 @@ export type CodexRequestShape = { tools_count: number | null has_reasoning: boolean /** True when the request asks for structured output (`text.format`, an - * output schema). Codex sets it only on temporary structured turns, such - * as 0.157 title generation (tui `thread_title.rs`); ordinary turns never - * carry one. The latest-request-body sidecar skips these (#70 review a). */ + * output schema). In the TUI that Agent Code runs, Codex sets it on + * temporary structured turns such as 0.157 title generation + * (`thread_title.rs`). It is a per-turn option upstream, though (exec, + * app-server `turn_start`), so a main turn COULD carry one; Agent Code + * sends none today. The latest-request-body sidecar skips these (#70 + * reviews a, c): a structured main turn would leave the sidecar one turn + * behind, which is the accepted cost of keeping title prompts out. */ has_output_schema: boolean client_metadata: { thread_id: string | null @@ -721,8 +725,9 @@ export class ResponsesProxy extends EventEmitter { // cannot classify is better evidence than none). // Subagent calls are skipped too (#70 review b): Codex tags every // non-main Responses call with `x-openai-subagent` (codex-api - // requests/headers.rs: review, compact, thread_spawn, - // memory_consolidation, or a custom label). A spawned worker's task has + // requests/headers.rs and responses_metadata.rs: review, compact, + // collab_spawn, guardian, memory_consolidation, or a custom label; the + // rule below is label-agnostic, so the exact list does not matter). A spawned worker's task has // no output schema, so without this it replaced the top-level prompt in // multi-agent sessions. `compact` is kept: a compaction request carries // the main conversation, and it is what a bundle after compaction needs.