diff --git a/API.md b/API.md index 3dc1e6c..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. 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 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 7fb7255..bf1c763 100644 --- a/SECURITY.md +++ b/SECURITY.md @@ -42,6 +42,13 @@ 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 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. 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 new file mode 100644 index 0000000..1e783cc --- /dev/null +++ b/docs/plans/2026-09-27-latest-request-body.md @@ -0,0 +1,58 @@ +# 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. + +## 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: 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. +- **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. + +## 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). + +## 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. + +## 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.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.test.ts b/src/proxy/latestRequestBody.test.ts new file mode 100644 index 0000000..4c9117f --- /dev/null +++ b/src/proxy/latestRequestBody.test.ts @@ -0,0 +1,84 @@ +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) +}) + +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 new file mode 100644 index 0000000..b1de04f --- /dev/null +++ b/src/proxy/latestRequestBody.ts @@ -0,0 +1,189 @@ +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 +// 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 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 holds the +// newest main-turn body, even when the log also has it (at most one body +// appears twice in a bundle). +// +// 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). 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. +// +// "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 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. 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. + +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. 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 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): 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 + : { generation: this.generation, requestId, endpoint, body } + this.kick() + } + + /** Resolves once every recorded body has been written, or dropped. */ + flush(): Promise { + 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 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}.${entry.generation}.tmp` + try { + await writeEncoded(temp, entry) + // 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 + } + 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() + } + } + + private async removeTempFiles(): Promise { + // 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 */ } + } +} + +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 new file mode 100644 index 0000000..61ef2f4 --- /dev/null +++ b/src/proxy/responsesProxy.latestBody.test.ts @@ -0,0 +1,187 @@ +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) +}) + +// #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' }) +}) + +// #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) +}) + +// #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 b6355c2..f87daaf 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. @@ -214,6 +215,15 @@ 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). 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 session_id: string | null @@ -247,6 +257,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 @@ -268,6 +280,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, @@ -392,6 +405,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 +419,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 +435,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 +545,7 @@ export class ResponsesProxy extends EventEmitter { await this.stopServer() } finally { await this.mirror?.close() + await this.latestBody?.flush() } } @@ -698,6 +718,24 @@ export class ResponsesProxy extends EventEmitter { ...(bodyB64 !== undefined ? { body_b64: bodyB64 } : {}), ...(requestShape !== null ? { request_shape: requestShape } : {}), }) + // 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). + // Subagent calls are skipped too (#70 review b): Codex tags every + // non-main Responses call with `x-openai-subagent` (codex-api + // 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. + 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) + } const abort = new AbortController() const headersTimer = setTimeout(() => {