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 }, + ]) + }) +})