From 6d0573fc74d77049fe7538a0aff91bc7e0bd140f Mon Sep 17 00:00:00 2001 From: Drew Stone Date: Fri, 14 Aug 2026 23:22:55 -0600 Subject: [PATCH] feat(intake): stream session transcripts past the 512MB string ceiling parseCodeAgentJsonl takes the whole transcript as one string. V8 cannot build a string longer than 536,870,888 characters, so a session above that size throws ERR_STRING_TOO_LONG and no part of it can be ingested. The largest real Codex rollout in the local corpus is 696,493,387 bytes and is invisible to every trace analysis today. The longest sessions are the long-horizon runs a research program most needs to read. Add streamCodeAgentJsonlFile, which reads a transcript one line at a time and retains only the unterminated tail, and parseCodeAgentJsonlFile, which returns the existing { entries, malformedLines } shape from a file path. The string entrypoint keeps its signature and behavior. Both paths call one readCodeAgentJsonlLine helper, so blank-line skipping, malformed-line counting, and entry parsing cannot drift apart. The stream breaks lines on \n only, which is what split('\n') does. node:readline also breaks on a bare \r, which reports a different malformed count for the same file, so it is not used. Measured on the 696,493,387-byte rollout: the string path throws ERR_STRING_TOO_LONG; the stream reads 90,170 entries and 1 malformed line under a hard 128MB heap cap. On three real 158-189MB sessions, including one with 2 malformed lines, both paths produce identical per-entry SHA-256 digests and identical counts. --- src/contract/index.ts | 3 + src/contract/intake/code-agent-session.ts | 78 ++++++++++++++-- src/contract/intake/index.ts | 3 + tests/contract-code-agent-intake.test.ts | 105 +++++++++++++++++++++- 4 files changed, 179 insertions(+), 10 deletions(-) diff --git a/src/contract/index.ts b/src/contract/index.ts index 8056d031..084a57f1 100644 --- a/src/contract/index.ts +++ b/src/contract/index.ts @@ -297,6 +297,7 @@ export { type AgentTraceRange, type AgentTraceRecord, type AuthoringProvenance, + type CodeAgentJsonlLine, type CodeAgentSessionAction, type CodeAgentSessionActionKind, type CodeAgentSessionActionStatus, @@ -330,6 +331,8 @@ export { type PartitionByAuthoringModelResult, parseAgentTrace, parseCodeAgentJsonl, + parseCodeAgentJsonlFile, partitionRunsByAuthoringModel, type RunRecordRejection, + streamCodeAgentJsonlFile, } from './intake' diff --git a/src/contract/intake/code-agent-session.ts b/src/contract/intake/code-agent-session.ts index 93ad7671..1f85930a 100644 --- a/src/contract/intake/code-agent-session.ts +++ b/src/contract/intake/code-agent-session.ts @@ -1,4 +1,5 @@ import { createHash } from 'node:crypto' +import { createReadStream } from 'node:fs' import { estimateCost, isModelPriced } from '../../metrics' import type { RunCostProvenance, @@ -121,17 +122,82 @@ export interface CodeAgentSessionIntakeOptions { execution?: CodeAgentSessionExecutionReceipt } +/** One transcript line after the intake rule ran on it. A blank line produces + * nothing, so every value here is either a parsed entry or a counted defect. */ +export type CodeAgentJsonlLine = + | { kind: 'entry'; lineNumber: number; entry: unknown } + | { kind: 'malformed'; lineNumber: number } + +/** The single per-line rule. Both the string path and the file path call this, + * so malformed-line handling and entry validation cannot drift apart. */ +function readCodeAgentJsonlLine(line: string, lineNumber: number): CodeAgentJsonlLine | undefined { + const trimmed = line.trim() + if (!trimmed) return undefined + try { + return { kind: 'entry', lineNumber, entry: JSON.parse(trimmed) } + } catch { + return { kind: 'malformed', lineNumber } + } +} + export function parseCodeAgentJsonl(jsonl: string): ParsedCodeAgentJsonl { const entries: unknown[] = [] let malformedLines = 0 + let lineNumber = 0 for (const line of jsonl.split('\n')) { - const trimmed = line.trim() - if (!trimmed) continue - try { - entries.push(JSON.parse(trimmed)) - } catch { - malformedLines += 1 + lineNumber += 1 + const read = readCodeAgentJsonlLine(line, lineNumber) + if (!read) continue + if (read.kind === 'malformed') malformedLines += 1 + else entries.push(read.entry) + } + return { entries, malformedLines } +} + +/** Reads a transcript one line at a time and never holds the file as a single + * string. `parseCodeAgentJsonl` needs the whole file in one string, so a + * session above V8's ~512MB string ceiling throws `ERR_STRING_TOO_LONG` and + * cannot be ingested at all; the largest real Codex rollout on record is 695MB. + * + * Lines break on `\n` only, which is what the string path's `split('\n')` does. + * `node:readline` also breaks on a bare `\r`, so it is deliberately not used + * here: a lone carriage return inside a line must stay inside that line for the + * two paths to report the same malformed count. */ +export async function* streamCodeAgentJsonlFile(path: string): AsyncGenerator { + const stream = createReadStream(path, { encoding: 'utf8' }) + let pending = '' + let lineNumber = 0 + try { + for await (const chunk of stream) { + pending += chunk + let start = 0 + for (let at = pending.indexOf('\n'); at !== -1; at = pending.indexOf('\n', start)) { + lineNumber += 1 + const read = readCodeAgentJsonlLine(pending.slice(start, at), lineNumber) + if (read) yield read + start = at + 1 + } + // Only the unterminated tail is retained, so live memory is bounded by + // the longest single line rather than by the file size. + pending = pending.slice(start) } + } finally { + stream.destroy() + } + const last = readCodeAgentJsonlLine(pending, lineNumber + 1) + if (last) yield last +} + +/** Streaming counterpart to `parseCodeAgentJsonl` for a transcript on disk. + * It returns the same shape, so a caller that holds every entry keeps working + * above the string ceiling. The entry array still grows with the transcript; + * consume `streamCodeAgentJsonlFile` directly when memory must stay flat. */ +export async function parseCodeAgentJsonlFile(path: string): Promise { + const entries: unknown[] = [] + let malformedLines = 0 + for await (const read of streamCodeAgentJsonlFile(path)) { + if (read.kind === 'malformed') malformedLines += 1 + else entries.push(read.entry) } return { entries, malformedLines } } diff --git a/src/contract/intake/index.ts b/src/contract/intake/index.ts index 3294d91a..8cdcbb48 100644 --- a/src/contract/intake/index.ts +++ b/src/contract/intake/index.ts @@ -32,6 +32,7 @@ export { partitionRunsByAuthoringModel, } from './agent-trace' export { + type CodeAgentJsonlLine, type CodeAgentSessionAction, type CodeAgentSessionActionKind, type CodeAgentSessionActionStatus, @@ -53,6 +54,8 @@ export { observeCodeAgentSession, type ParsedCodeAgentJsonl, parseCodeAgentJsonl, + parseCodeAgentJsonlFile, + streamCodeAgentJsonlFile, } from './code-agent-session' export { type FeedbackTableMeta, diff --git a/tests/contract-code-agent-intake.test.ts b/tests/contract-code-agent-intake.test.ts index 31f2d9f2..7eaa6ea1 100644 --- a/tests/contract-code-agent-intake.test.ts +++ b/tests/contract-code-agent-intake.test.ts @@ -1,13 +1,19 @@ -import { readFileSync } from 'node:fs' -import { describe, expect, it } from 'vitest' +import { mkdtempSync, readFileSync, rmSync, writeFileSync } from 'node:fs' +import { tmpdir } from 'node:os' +import { join } from 'node:path' +import { fileURLToPath } from 'node:url' +import { afterAll, describe, expect, it } from 'vitest' import { analyzeRuns, + type CodeAgentJsonlLine, fromClaudeCodeSession, fromCodexSession, fromKimiCodeSession, fromOpenCodeSession, fromPiSession, parseCodeAgentJsonl, + parseCodeAgentJsonlFile, + streamCodeAgentJsonlFile, } from '../src/contract' import { validateRunRecord } from '../src/run-record' @@ -331,7 +337,11 @@ describe('code-agent session intake', () => { { timestamp: '2026-08-14T00:00:03.000Z', type: 'response_item', - payload: { type: 'message', role: 'assistant', content: [{ type: 'text', text: 'done' }] }, + payload: { + type: 'message', + role: 'assistant', + content: [{ type: 'text', text: 'done' }], + }, }, // The execution record. Reasoning and CommandExecution repeat the // transcript above; the rest appear nowhere else in the file. @@ -352,7 +362,8 @@ describe('code-agent session intake', () => { changes: { '/repo/src/lib/billing.ts': { type: 'update', - unified_diff: '@@ -448,2 +448,3 @@\n interface BillingServiceDeps {\n+ enabled?: boolean;\n', + unified_diff: + '@@ -448,2 +448,3 @@\n interface BillingServiceDeps {\n+ enabled?: boolean;\n', }, }, stdout: '', @@ -1187,3 +1198,89 @@ describe('code-agent session intake', () => { }, ) }) + +describe('streaming code-agent JSONL intake', () => { + const dir = mkdtempSync(join(tmpdir(), 'code-agent-jsonl-')) + const write = (name: string, body: string) => { + const path = join(dir, name) + writeFileSync(path, body) + return path + } + + afterAll(() => rmSync(dir, { recursive: true, force: true })) + + it('matches the string path on a real committed session fixture', async () => { + const url = new URL('./fixtures/codex-exec-0.144.1.jsonl', import.meta.url) + const fromString = parseCodeAgentJsonl(readFileSync(url, 'utf8')) + const fromFile = await parseCodeAgentJsonlFile(fileURLToPath(url)) + + expect(fromFile.entries).toHaveLength(15) + expect(fromFile).toEqual(fromString) + }) + + it('counts malformed lines exactly as the string path does', async () => { + const body = '{"type":"ok"}\nnot-json\n\n \n{"type":"also-ok"}\n{"unterminated":\n' + const path = write('malformed.jsonl', body) + + const fromFile = await parseCodeAgentJsonlFile(path) + + expect(fromFile.malformedLines).toBe(2) + expect(fromFile).toEqual(parseCodeAgentJsonl(body)) + }) + + it.each([ + ['no trailing newline', '{"a":1}\n{"b":2}'], + ['CRLF line breaks', '{"a":1}\r\n{"b":2}\r\n'], + ['blank and whitespace-only lines', '\n{"a":1}\n \t \n{"b":2}\n\n'], + ['empty file', ''], + ['only malformed content', 'nope\nalso nope\n'], + ])('agrees with the string path on %s', async (name, body) => { + const path = write(`${name.replace(/\s+/g, '-')}.jsonl`, body) + + expect(await parseCodeAgentJsonlFile(path)).toEqual(parseCodeAgentJsonl(body)) + }) + + it('keeps a bare carriage return inside its line, as split by newline does', async () => { + // `node:readline` breaks on a lone `\r`, which would report two malformed + // lines here instead of one and silently diverge from the string path. + const body = '{"a":1}\nbroken\rline\n{"b":2}\n' + const path = write('bare-cr.jsonl', body) + + const fromFile = await parseCodeAgentJsonlFile(path) + + expect(fromFile.malformedLines).toBe(1) + expect(fromFile).toEqual(parseCodeAgentJsonl(body)) + }) + + it('reassembles entries that span read-stream chunk boundaries', async () => { + // Each line is far larger than the 64KB default chunk, so a splitter that + // does not retain the unterminated tail loses or corrupts every entry. + const long = Array.from({ length: 8 }, (_, i) => + JSON.stringify({ index: i, filler: 'x'.repeat(200_000) }), + ).join('\n') + const path = write('chunked.jsonl', `${long}\n`) + + const fromFile = await parseCodeAgentJsonlFile(path) + + expect(fromFile.entries).toHaveLength(8) + expect(fromFile.entries.map((e) => (e as { index: number }).index)).toEqual([ + 0, 1, 2, 3, 4, 5, 6, 7, + ]) + expect(fromFile).toEqual(parseCodeAgentJsonl(long)) + }) + + it('reports line numbers and stops reading when the consumer stops', async () => { + const path = write('numbered.jsonl', '\n{"a":1}\nnot-json\n{"b":2}\n') + + const seen: CodeAgentJsonlLine[] = [] + for await (const line of streamCodeAgentJsonlFile(path)) { + seen.push(line) + if (seen.length === 2) break + } + + expect(seen).toEqual([ + { kind: 'entry', lineNumber: 2, entry: { a: 1 } }, + { kind: 'malformed', lineNumber: 3 }, + ]) + }) +})