From 002d876556d89967e1c37072de03e1f740502361 Mon Sep 17 00:00:00 2001 From: Aamer Akhter Date: Sat, 24 Oct 2026 23:00:49 -0400 Subject: [PATCH] fix(history): read forked threads, large stores and live rollouts A forked or spawned thread's rollout carries its own session_meta followed by its parent's and a copy of the parent's history. A later session_meta for a declared ancestor now opens an inherited prefix that is dropped: by ordinal below subagent_history_start_ordinal in paginated history, and up to the thread_settings_applied event Codex appends after the copy in legacy history, checked against the parent's own messages when the copy holds more than one candidate. An unrelated id still refuses the file. Store attribution now has its own budget (--metadata-bytes) separate from the transcript budget (--scan-bytes), reads each head in 32 KiB steps, and --max-entries is a flag, so id lookups work on stores of thousands of rollouts. An unattributable store file or unreadable directory is counted instead of aborting the lookup, and a rollout that a running Codex appends to is read as a snapshot up to its last complete line. User records carry no phase, date bounds are UTC calendar dates, the injected-context rule is the shared one, a symlinked sessions root is followed as the Past Sessions scanner follows it, and --json with --format text is refused. --- src/cli-history.ts | 70 +++- src/local-history.ts | 638 ++++++++++++++++++++++---------- test/fixtures/codex-rollout.ts | 125 +++++++ test/local-history-live.test.ts | 129 +++++++ test/local-history.test.ts | 472 ++++++++++++++++++++++- 5 files changed, 1234 insertions(+), 200 deletions(-) create mode 100644 test/fixtures/codex-rollout.ts create mode 100644 test/local-history-live.test.ts diff --git a/src/cli-history.ts b/src/cli-history.ts index 614a416e..11be6088 100644 --- a/src/cli-history.ts +++ b/src/cli-history.ts @@ -1,6 +1,33 @@ -/** Local offline transcript commands. Deliberately separate from the agent API guard. */ +/** + * @fileoverview `codeman history read`: the command-line front of the offline + * Codex history reader (`local-history.ts`). + * + * Deliberately separate from `codeman agent` (`cli-agent.ts`): that command is + * a client of a running Codeman and refuses outside a managed session, while + * this one reads rollout files from disk and needs neither a server nor + * `CODEMAN_MUX`. It never writes, never inspects a live pane and never sends + * input. + * + * Every bound the reader enforces has a flag here with the reader's own + * maximum, so a large store is a matter of raising a number rather than an + * impossible lookup. Output is JSON unless `--format text` is given; `--json` + * is the JSON spelling and refuses to combine with `--format text`. + * + * @module cli-history + */ import type { Command } from 'commander'; -import { formatLocalHistory, HISTORY_DEFAULT_SCAN_BYTES, readLocalHistory } from './local-history.js'; +import { + formatLocalHistory, + HISTORY_DEFAULT_MAX_ENTRIES, + HISTORY_DEFAULT_METADATA_BYTES, + HISTORY_DEFAULT_OUTPUT_BYTES, + HISTORY_DEFAULT_SCAN_BYTES, + HISTORY_MAX_ENTRIES, + HISTORY_MAX_METADATA_BYTES, + HISTORY_MAX_OUTPUT_BYTES, + HISTORY_MAX_SCAN_BYTES, + readLocalHistory, +} from './local-history.js'; export function registerHistoryCommands(program: Command): Command { const history = program @@ -12,22 +39,41 @@ export function registerHistoryCommands(program: Command): Command { .option('--file ', 'Explicit local JSONL transcript (regular file only)') .option('--session-id ', 'Native Codex conversation UUID or unique prefix (at least 8 characters)') .option('--root-session-id ', 'Native root UUID, including transitive child conversations') - .option('--sessions-root ', 'Local Codex sessions directory (default: CODEX_HOME/sessions)') + .option( + '--sessions-root ', + 'Local Codex sessions directory (default: $CODEX_HOME/sessions, else ~/.codex/sessions)' + ) .option('--topic ', 'Case-insensitive literal match, preserving whole matching turns') - .option('--after ', 'Inclusive YYYY-MM-DD lower bound; preserves whole matching turns') - .option('--before ', 'Inclusive YYYY-MM-DD upper bound; preserves whole matching turns') - .option('--max-bytes ', 'Maximum complete output bytes (1–1048576)', '8192') + .option('--after ', 'Inclusive YYYY-MM-DD lower bound, as a UTC date; preserves whole matching turns') + .option('--before ', 'Inclusive YYYY-MM-DD upper bound, as a UTC date; preserves whole matching turns') + .option( + '--max-bytes ', + `Maximum complete output bytes (1-${HISTORY_MAX_OUTPUT_BYTES})`, + String(HISTORY_DEFAULT_OUTPUT_BYTES) + ) .option( '--scan-bytes ', - 'Total metadata and selected transcript read budget (1–67108864)', + `Read budget for the selected transcripts (1-${HISTORY_MAX_SCAN_BYTES})`, String(HISTORY_DEFAULT_SCAN_BYTES) ) + .option( + '--metadata-bytes ', + `Read budget for attributing every rollout in the store (1-${HISTORY_MAX_METADATA_BYTES})`, + String(HISTORY_DEFAULT_METADATA_BYTES) + ) + .option( + '--max-entries ', + `Directory entries visited in the store (1-${HISTORY_MAX_ENTRIES})`, + String(HISTORY_DEFAULT_MAX_ENTRIES) + ) .option('--format ', 'json or text', 'json') - .option('--json', 'JSON output (also the default)') - .action(async (options: Record) => { + .option('--json', 'JSON output (the default; conflicts with --format text)') + .action(async (options: Record, command: Command) => { try { - const format = options.json ? 'json' : options.format; - if (format !== 'json' && format !== 'text') throw new Error('--format must be json or text'); + if (options.format !== 'json' && options.format !== 'text') throw new Error('--format must be json or text'); + if (options.json && options.format === 'text' && command.getOptionValueSource('format') === 'cli') + throw new Error('--json conflicts with --format text; pass one of them'); + const format = options.format; const result = await readLocalHistory({ file: options.file as string | undefined, sessionId: options.sessionId as string | undefined, @@ -37,6 +83,8 @@ export function registerHistoryCommands(program: Command): Command { after: options.after as string | undefined, before: options.before as string | undefined, scanBytes: Number(options.scanBytes), + metadataBytes: Number(options.metadataBytes), + maxEntries: Number(options.maxEntries), }); program.configureOutput().writeOut?.(`${formatLocalHistory(result, format, Number(options.maxBytes))}\n`); } catch (error) { diff --git a/src/local-history.ts b/src/local-history.ts index ac7fb04f..e6162c0d 100644 --- a/src/local-history.ts +++ b/src/local-history.ts @@ -1,26 +1,117 @@ -/** Offline, local-operator Codex history. No HTTP routes or running session required. */ +/** + * @fileoverview Offline reader for saved Codex conversations, behind + * `codeman history read` (`cli-history.ts`). It reads rollout JSONL files from + * disk directly: no HTTP route, no running server and no managed session. + * + * ## Selecting conversations + * + * Exactly one selector: `file` (one rollout, no store search), `sessionId` (a + * native thread id or unique prefix of at least 8 characters) or + * `rootSessionId` (a full id plus every transitive descendant). Identity always + * comes from each file's `session_meta`, never from its filename. + * + * ## Budgets + * + * Three separate bounds, each raisable from the CLI up to a fixed maximum, and + * each failing closed with an error rather than returning a partial answer: + * + * - `metadataBytes`: the head reads that attribute every rollout in the store. + * A `session_meta` line embeds Codex's base instructions (about 20 KiB on + * every rollout measured), so this scales with the store, not the query, + * and is kept apart from the transcript budget: a lookup on a store of + * thousands of rollouts must not spend the budget meant for the transcripts + * it returns. Each file's head read is also capped at 128 KiB. + * - `scanBytes`: the selected transcripts, read whole. + * - `maxEntries`: directory entries visited. + * + * ## Forks and sub-agents + * + * A thread Codex forks or spawns as a sub-agent can carry a copy of its + * parent's history in its own file (codex-rs `ForkPersistence::Copied`, written + * by `record_initial_history` in `core/src/session/mod.rs`; the copy keeps the + * parent's `session_meta`, see `keep_forked_rollout_item` in + * `core/src/agent/control/spawn.rs`). Such a file starts with its own + * `session_meta`, then the parent's, then the copied records, then one + * `thread_settings_applied` event, then the child's own records. The copy + * already lives in the parent's own file, so it is dropped here: + * + * - Paginated history (`history_mode: "paginated"`) numbers every record with + * an `ordinal` and declares `subagent_history_start_ordinal`; records below + * it are inherited, exactly as Codex's own projection treats them + * (`thread-store/src/local/thread_history_materialization.rs`). A paginated + * thread with a `history_base` references its prefix in another file and has + * nothing inherited to drop. + * - Legacy history has no ordinals. The prefix runs from the inherited + * `session_meta` to the `thread_settings_applied` event Codex appends after + * the copy. A legacy copy keeps the parent's own `thread_settings_applied` + * events too, so when more than one candidate follows, the boundary is the + * last one whose preceding messages are still a prefix of the parent's own + * messages, read from the parent's file. Without the parent's file that + * cannot be proven: the cut then falls at the LAST candidate, so parent text + * is never attributed to the child, and the result says so + * (`unverifiedPrefixes`, `complete: false`). + * + * A later `session_meta` must be the file's own id or a declared ancestor + * (`parent_thread_id`, `forked_from_id`, or the shared root `session_id`, and + * transitively the ancestors those declare); any other id still refuses the + * whole file rather than attribute one conversation's answer to another. + * + * ## Live rollouts + * + * A running Codex appends to its rollout. Reads take a snapshot: the file's + * identity (device and inode) must match the attributed one and its size may + * only have grown; everything up to the last complete line is read, and an + * unterminated tail is reported as a partial record. + * + * ## Format + * + * The line shapes and the injected-context rule are shared with the Past + * Sessions scanner through `codex-rollout-format.ts`. + * + * @module local-history + */ import { constants } from 'node:fs'; import { isUtf8 } from 'node:buffer'; -import { lstat, open, opendir, type FileHandle } from 'node:fs/promises'; -import { homedir } from 'node:os'; +import { lstat, open, opendir, stat, type FileHandle } from 'node:fs/promises'; import { join, resolve } from 'node:path'; -export const HISTORY_DEFAULT_SCAN_BYTES = 16 * 1024 * 1024; -export const HISTORY_MAX_SCAN_BYTES = 64 * 1024 * 1024; -export const HISTORY_MAX_OUTPUT_BYTES = 1024 * 1024; +import { codexMessageFromLine, codexSessionsRoot, isCodexInjectedContext } from './codex-rollout-format.js'; + +const KIB = 1024; +const MIB = 1024 * KIB; +const GIB = 1024 * MIB; + +export const HISTORY_DEFAULT_SCAN_BYTES = 16 * MIB; +export const HISTORY_MAX_SCAN_BYTES = 256 * MIB; +export const HISTORY_DEFAULT_METADATA_BYTES = 1 * GIB; +export const HISTORY_MAX_METADATA_BYTES = 16 * GIB; +export const HISTORY_DEFAULT_MAX_ENTRIES = 65536; +export const HISTORY_MAX_ENTRIES = 1048576; +export const HISTORY_DEFAULT_OUTPUT_BYTES = 8192; +export const HISTORY_MAX_OUTPUT_BYTES = 1 * MIB; +/** Per-file head window for `session_meta`. The line itself runs to about 20 KiB. */ +const METADATA_HEAD_BYTES = 128 * KIB; +/** Read the head in steps, so a typical rollout costs one step past its session_meta line. */ +const METADATA_CHUNK_BYTES = 32 * KIB; +/** Codex writes `sessions/YYYY/MM/DD`; this leaves five levels of headroom. */ +const MAX_DIRECTORY_DEPTH = 8; + const ID = /^[0-9a-f]{8}(?:[0-9a-f-]{0,28})$/i; const FULL_ID = /^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$/i; -const METADATA_BYTES = 128 * 1024; type Json = Record; + export interface LocalHistoryOptions { file?: string; sessionId?: string; rootSessionId?: string; sessionsRoot?: string; topic?: string; + /** Inclusive UTC date, `YYYY-MM-DD`. */ after?: string; + /** Inclusive UTC date, `YYYY-MM-DD`. */ before?: string; scanBytes?: number; + metadataBytes?: number; maxEntries?: number; } export interface HistoryRecord { @@ -28,16 +119,23 @@ export interface HistoryRecord { sessionId: string; timestamp: string; role: 'user' | 'assistant'; - phase: string; + /** Codex's own phase for assistant messages (`commentary`, `final_answer`); null when none was recorded. */ + phase: string | null; message: string; } interface Source { path: string; sessionId: string; parentId?: string; + forkedFromId?: string; rootId?: string; + /** `subagent_history_start_ordinal`: records below it are inherited. */ + historyStart?: number; timestamp: string; - fingerprint?: string; + /** `dev:ino` of the attributed file. */ + identity: string; + /** Size when attributed; a later read may only see it grow. */ + size: number; } export interface LocalHistoryResult { source: 'codex-jsonl'; @@ -45,25 +143,49 @@ export interface LocalHistoryResult { sessions: Array<{ sessionId: string; rootSessionId: string | null; parentSessionId?: string; path: string }>; records: HistoryRecord[]; scan: { + /** Transcript bytes read (`--scan-bytes`). */ bytesRead: number; + /** Head bytes read to attribute rollouts (`--metadata-bytes`). */ + metadataBytesRead: number; entries: number; malformedRecords: number; partialRecords: number; skippedSymlinks: number; unattributedFiles: number; + unreadableDirectories: number; + /** Inherited fork or sub-agent prefixes dropped (informational). */ + inheritedPrefixes: number; + /** Inherited prefixes that never reached their end marker. */ + incompletePrefixes: number; + /** Legacy prefixes whose end could not be checked against the parent's own file. */ + unverifiedPrefixes: number; }; } interface Budget { - remaining: number; + transcriptRemaining: number; + metadataRemaining: number; entriesRemaining: number; scan: LocalHistoryResult['scan']; } + +/** A budget ran out: always fatal, never counted. */ +class HistoryBudgetError extends Error {} +/** One store file could not be attributed: fatal for `--file`, counted for a store scan. */ +class UnattributableFileError extends Error {} + function object(value: unknown): Json { return value && typeof value === 'object' && !Array.isArray(value) ? (value as Json) : {}; } function string(value: unknown): string | undefined { return typeof value === 'string' ? value : undefined; } +function fullId(value: unknown): string | undefined { + const id = string(value); + return id && FULL_ID.test(id) ? id.toLowerCase() : undefined; +} +function fileIdentity(value: { dev: number; ino: number }): string { + return `${value.dev}:${value.ino}`; +} function validDate(value: string | undefined): boolean { return ( value === undefined || @@ -72,6 +194,14 @@ function validDate(value: string | undefined): boolean { new Date(value).toISOString().slice(0, 10) === value) ); } +/** The UTC calendar date of a native timestamp, or undefined when it does not parse. */ +function utcDate(timestamp: string): string | undefined { + const ms = Date.parse(timestamp); + return Number.isFinite(ms) ? new Date(ms).toISOString().slice(0, 10) : undefined; +} +function checkBudget(value: number, max: number, flag: string): void { + if (!Number.isSafeInteger(value) || value < 1 || value > max) throw new Error(`Invalid ${flag} (1-${max})`); +} function validate(options: LocalHistoryOptions): void { if ([options.file, options.sessionId, options.rootSessionId].filter((v) => v !== undefined).length !== 1) throw new Error('Select exactly one of --file, --session-id, or --root-session-id'); @@ -84,13 +214,11 @@ function validate(options: LocalHistoryOptions): void { !validDate(options.before) || (options.after && options.before && options.after > options.before) ) - throw new Error('Invalid date or date range (use YYYY-MM-DD)'); + throw new Error('Invalid date or date range (use YYYY-MM-DD, UTC)'); if (options.topic !== undefined && !options.topic.trim()) throw new Error('Invalid empty topic'); - for (const [value, max] of [ - [options.scanBytes ?? HISTORY_DEFAULT_SCAN_BYTES, HISTORY_MAX_SCAN_BYTES], - [options.maxEntries ?? 8192, 65536], - ]) - if (!Number.isSafeInteger(value) || value < 1 || value > max) throw new Error('Invalid scan budget'); + checkBudget(options.scanBytes ?? HISTORY_DEFAULT_SCAN_BYTES, HISTORY_MAX_SCAN_BYTES, '--scan-bytes'); + checkBudget(options.metadataBytes ?? HISTORY_DEFAULT_METADATA_BYTES, HISTORY_MAX_METADATA_BYTES, '--metadata-bytes'); + checkBudget(options.maxEntries ?? HISTORY_DEFAULT_MAX_ENTRIES, HISTORY_MAX_ENTRIES, '--max-entries'); } async function regularFile(path: string): Promise { // O_NOFOLLOW prevents a final-component link swap between lstat and open. fstat @@ -103,88 +231,148 @@ async function regularFile(path: string): Promise { } return handle; } -async function readChunk(handle: FileHandle, position: number, size: number, budget: Budget): Promise { - if (size > budget.remaining) - throw new Error('History scan byte budget exhausted; narrow the store or raise --scan-bytes'); +async function readAt(handle: FileHandle, position: number, size: number): Promise { const buffer = Buffer.alloc(size); - const { bytesRead } = await handle.read(buffer, 0, size, position); - budget.remaining -= bytesRead; - budget.scan.bytesRead += bytesRead; - return buffer.subarray(0, bytesRead); + let filled = 0; + while (filled < size) { + const { bytesRead } = await handle.read(buffer, filled, size - filled, position + filled); + if (!bytesRead) break; + filled += bytesRead; + } + return buffer.subarray(0, filled); } -function metadata(value: unknown, path: string): Source | undefined { +async function readMetadataChunk(handle: FileHandle, position: number, size: number, budget: Budget) { + if (size > budget.metadataRemaining) + throw new HistoryBudgetError('History metadata byte budget exhausted; raise --metadata-bytes'); + const chunk = await readAt(handle, position, size); + budget.metadataRemaining -= chunk.length; + budget.scan.metadataBytesRead += chunk.length; + return chunk; +} +async function readTranscript(handle: FileHandle, size: number, budget: Budget): Promise { + if (size > budget.transcriptRemaining) + throw new HistoryBudgetError('History scan byte budget exhausted; narrow the selection or raise --scan-bytes'); + const buffer = await readAt(handle, 0, size); + budget.transcriptRemaining -= buffer.length; + budget.scan.bytesRead += buffer.length; + return buffer; +} + +type Identity = Omit; +function metadata(value: unknown, path: string): Identity | undefined { const entry = object(value); if (entry.type !== 'session_meta') return undefined; const p = object(entry.payload); - const sessionId = string(p.id) ?? string(p.session_id); - if (!sessionId || !FULL_ID.test(sessionId)) return undefined; - const source = object(p.source); - const spawn = object(object(source.subagent).thread_spawn); - const parentId = string(p.parent_thread_id) ?? string(p.parent_session_id) ?? string(spawn.parent_thread_id); - const declaredRoot = string(p.root_session_id) ?? string(p.session_id); + const sessionId = fullId(p.id) ?? fullId(p.session_id); + if (!sessionId) return undefined; + const spawn = object(object(object(p.source).subagent).thread_spawn); + const forkedFromId = fullId(p.forked_from_id); + // A fork with no spawn parent (a `/fork` in the TUI) still descends from its source thread. + const parentId = + fullId(p.parent_thread_id) ?? fullId(p.parent_session_id) ?? fullId(spawn.parent_thread_id) ?? forkedFromId; + // Codex: "session_id is equal to the root thread's ID". + const declaredRoot = fullId(p.root_session_id) ?? fullId(p.session_id); + const start = p.subagent_history_start_ordinal; return { path, - sessionId: sessionId.toLowerCase(), - parentId: parentId && FULL_ID.test(parentId) ? parentId.toLowerCase() : undefined, - rootId: - declaredRoot && declaredRoot !== sessionId && FULL_ID.test(declaredRoot) ? declaredRoot.toLowerCase() : undefined, + sessionId, + parentId, + forkedFromId, + rootId: declaredRoot && declaredRoot !== sessionId ? declaredRoot : undefined, + historyStart: Number.isSafeInteger(start) && (start as number) >= 0 ? (start as number) : undefined, timestamp: string(p.timestamp) ?? string(entry.timestamp) ?? '', }; } +function sameIdentity(a: Identity, b: Identity): boolean { + return ( + a.sessionId === b.sessionId && + a.parentId === b.parentId && + a.forkedFromId === b.forkedFromId && + a.rootId === b.rootId && + a.historyStart === b.historyStart + ); +} + +/** + * The first `session_meta` within the file's head. Problems with the file + * itself throw {@link UnattributableFileError}; a budget throws + * {@link HistoryBudgetError}; a short file with no metadata yet returns undefined. + */ async function readMetadata(path: string, budget: Budget): Promise { - const handle = await regularFile(path); + let handle: FileHandle; + try { + handle = await regularFile(path); + } catch (error) { + throw new UnattributableFileError(error instanceof Error ? error.message : String(error)); + } try { const before = await handle.stat(); - const parts: Buffer[] = []; - let offset = 0; + const limit = Math.min(before.size, METADATA_HEAD_BYTES); + let buffer = Buffer.alloc(0); let parsed = 0; - while (offset < before.size && offset < METADATA_BYTES) { - const chunk = await readChunk( + while (buffer.length < limit) { + const chunk = await readMetadataChunk( handle, - offset, - Math.min(4096, before.size - offset, METADATA_BYTES - offset), + buffer.length, + Math.min(METADATA_CHUNK_BYTES, limit - buffer.length), budget ); if (!chunk.length) break; - parts.push(chunk); - offset += chunk.length; - const buffer = Buffer.concat(parts); + buffer = Buffer.concat([buffer, chunk]); let end = buffer.indexOf(10, parsed); while (end >= 0) { const line = buffer.subarray(parsed, end); - if (!isUtf8(line)) throw new Error('History metadata contains invalid UTF8'); - try { - const found = metadata(JSON.parse(line.toString('utf8')), path); - if (found) { - const after = await handle.stat(); - const fingerprint = (value: typeof before) => - `${value.dev}:${value.ino}:${value.size}:${value.mtimeMs}:${value.ctimeMs}`; - if (fingerprint(before) !== fingerprint(after)) - throw new Error('History source changed during metadata scan'); - return { ...found, fingerprint: fingerprint(before) }; - } - } catch (error) { - if (!(error instanceof SyntaxError)) throw error; - } parsed = end + 1; end = buffer.indexOf(10, parsed); + if (!isUtf8(line)) throw new UnattributableFileError('History metadata contains invalid UTF8'); + let value: unknown; + try { + value = JSON.parse(line.toString('utf8')); + } catch { + continue; + } + const found = metadata(value, path); + if (!found) continue; + // session_meta is written once; a growing file is a live rollout, not a change. + const after = await handle.stat(); + if (fileIdentity(after) !== fileIdentity(before) || after.size < before.size) + throw new UnattributableFileError('History source changed during metadata scan'); + return { ...found, identity: fileIdentity(before), size: before.size }; } } - if (offset < before.size) - throw new Error('History metadata scan budget exhausted; cannot prove family completeness'); + if (buffer.length < limit) throw new UnattributableFileError('History source changed during metadata scan'); + if (before.size > METADATA_HEAD_BYTES) + throw new UnattributableFileError('No session metadata within the first 128 KiB'); return undefined; } finally { await handle.close(); } } + async function findSources(root: string, budget: Budget): Promise { - if (!(await lstat(root)).isDirectory()) throw new Error('Sessions root must be a real directory, not a symlink'); + // The root may itself be a symlink, as the Past Sessions scanner allows + // (`codex-transcript.ts` reads it with readdir, which follows it); links + // BELOW the root are never followed by either reader. + let rootStat; + try { + rootStat = await stat(root); + } catch { + throw new Error(`Sessions root not found: ${root}`); + } + if (!rootStat.isDirectory()) throw new Error('Sessions root must be a directory'); const sources: Source[] = []; const visit = async (dir: string, depth: number): Promise => { - if (depth > 8) throw new Error('History directory scan depth budget exhausted'); - const entries = await opendir(dir); + if (depth > MAX_DIRECTORY_DEPTH) throw new HistoryBudgetError('History directory scan depth budget exhausted'); + let entries; + try { + entries = await opendir(dir); + } catch { + budget.scan.unreadableDirectories++; + return; + } for await (const entry of entries) { - if (--budget.entriesRemaining < 0) throw new Error('History directory scan entry budget exhausted'); + if (--budget.entriesRemaining < 0) + throw new HistoryBudgetError('History directory scan entry budget exhausted; raise --max-entries'); budget.scan.entries++; const path = join(dir, entry.name); if (entry.isSymbolicLink()) { @@ -193,10 +381,19 @@ async function findSources(root: string, budget: Budget): Promise { } if (entry.isDirectory()) { // Recheck immediately before descending; never intentionally follow links. - if (!(await lstat(path)).isDirectory()) throw new Error('History directory changed during scan'); + const now = await lstat(path).catch(() => undefined); + if (!now?.isDirectory()) { + budget.scan.unreadableDirectories++; + continue; + } await visit(path, depth + 1); } else if (entry.isFile() && entry.name.endsWith('.jsonl')) { - const source = await readMetadata(path, budget); + let source: Source | undefined; + try { + source = await readMetadata(path, budget); + } catch (error) { + if (!(error instanceof UnattributableFileError)) throw error; + } if (source) sources.push(source); else budget.scan.unattributedFiles++; } @@ -205,134 +402,167 @@ async function findSources(root: string, budget: Budget): Promise { await visit(root, 0); return sources; } -function contentText(value: unknown): string { - if (typeof value === 'string') return value; - if (!Array.isArray(value)) return ''; - return value - .map((block) => { - const item = object(block); - return typeof item.type === 'string' && ['text', 'input_text', 'output_text'].includes(item.type.toLowerCase()) - ? (string(item.text) ?? '') - : ''; - }) - .filter(Boolean) - .join('\n'); -} + interface Message { representation: 'event' | 'modern' | 'response'; role: 'user' | 'assistant'; message: string; - phase: string; + phase: string | null; timestamp: string; } function message(value: unknown): Message | undefined { const entry = object(value); const p = object(entry.payload); if (entry.isSidechain || p.isSidechain) return undefined; - let role: Message['role'] | undefined; - let representation: Message['representation'] = 'event'; - let text = ''; - let phase = string(p.phase) ?? 'progress'; - if (entry.type === 'event_msg') { - if (p.type === 'user_message' || p.type === 'agent_message') { - role = p.type === 'user_message' ? 'user' : 'assistant'; - text = contentText(p.message); - } else if (p.type === 'item_completed') { - representation = 'modern'; - const item = object(p.item); - if (item.type === 'UserMessage' || item.type === 'AgentMessage') { - role = item.type === 'UserMessage' ? 'user' : 'assistant'; - text = contentText(item.content); - phase = string(item.phase) ?? phase; - } - } - } else if ( - entry.type === 'response_item' && - (p.type === 'message' || p.type === undefined) && - (p.role === 'user' || p.role === 'assistant') - ) { - role = p.role; - representation = 'response'; - text = contentText(p.content); - // Codex context is not a human turn. Do not omit arbitrary markdown prompts. - if ( - role === 'user' && - /^(?:# AGENTS\.md instructions|||)/.test(text) - ) - return undefined; - } - if (!role || !text) return undefined; + const found = codexMessageFromLine(value); + if (!found) return undefined; + // Codex context is not a human turn. Real prompts also arrive as events, which are never filtered. + if (found.injectionProne && isCodexInjectedContext(found.text)) return undefined; return { - role, - message: text, - phase, - representation, + role: found.role, + message: found.text, + // Codex records a phase on assistant messages only; a user turn has none. + phase: found.role === 'assistant' ? (found.phase ?? null) : null, + representation: found.shape === 'item' ? 'modern' : found.shape, timestamp: string(entry.timestamp) ?? string(p.timestamp) ?? '', }; } -async function readMessages(source: Source, budget: Budget): Promise { - const handle = await regularFile(source.path); - try { - const before = await handle.stat(); - if (`${before.dev}:${before.ino}:${before.size}:${before.mtimeMs}:${before.ctimeMs}` !== source.fingerprint) - throw new Error('History source changed after attribution; retry explicitly'); - const buffer = await readChunk(handle, 0, before.size, budget); - const after = await handle.stat(); +/** Native events and response_item commonly duplicate one accepted message. */ +function collapse(items: Message[]): Message[] { + const records: Message[] = []; + let representations = new Set(); + for (const raw of items) { + const item = { ...raw }; + const previous = records.at(-1); if ( - buffer.length !== before.size || - before.size !== after.size || - before.mtimeMs !== after.mtimeMs || - before.ctimeMs !== after.ctimeMs - ) - throw new Error('History source changed while reading; retry explicitly'); - if (!isUtf8(buffer)) throw new Error('History source contains invalid UTF8'); - const lines = buffer.toString('utf8').split('\n'); - const partial = lines.pop(); - if (partial) budget.scan.partialRecords++; - const records: Message[] = []; - let representations = new Set(); - for (const line of lines) { - if (!line.trim()) continue; - let parsed: unknown; - try { - parsed = JSON.parse(line); - } catch { - budget.scan.malformedRecords++; - continue; - } - if (object(parsed).type === 'session_meta') { - const declared = metadata(parsed, source.path); - if ( - !declared || - declared.sessionId !== source.sessionId || - declared.parentId !== source.parentId || - declared.rootId !== source.rootId - ) - throw new Error('History source contains conflicting native metadata; attribution refused'); - continue; - } - const item = message(parsed); - if (!item) continue; - const previous = records.at(-1); - // Native events and response_item commonly duplicate one accepted message. - if ( - previous?.role === item.role && - previous.message === item.message && - !representations.has(item.representation) - ) { - representations.add(item.representation); - if (!previous.timestamp) previous.timestamp = item.timestamp; - if (item.phase === 'final_answer') previous.phase = item.phase; - continue; - } - records.push(item); - representations = new Set([item.representation]); + previous?.role === item.role && + previous.message === item.message && + !representations.has(item.representation) + ) { + representations.add(item.representation); + if (!previous.timestamp) previous.timestamp = item.timestamp; + if (item.phase === 'final_answer' || previous.phase === null) previous.phase = item.phase ?? previous.phase; + continue; } - return records; + records.push(item); + representations = new Set([item.representation]); + } + return records; +} +function isPrefixOf(prefix: Message[], whole: Message[]): boolean { + return ( + prefix.length <= whole.length && + prefix.every((item, i) => item.role === whole[i].role && item.message === whole[i].message) + ); +} + +interface ReadContext { + /** The ancestor's own messages, uncut, or undefined when its file is not in the store. */ + ancestorMessages(id: string): Promise; +} +type Piece = Message | { boundary: true }; +interface LegacyPrefix { + ancestorId: string; + /** Messages and `thread_settings_applied` boundaries after the inherited session_meta. */ + pieces: Piece[]; +} + +async function readMessages(source: Source, budget: Budget, context: ReadContext, raw = false): Promise { + const handle = await regularFile(source.path); + let buffer: Buffer; + try { + const now = await handle.stat(); + // A live rollout only grows. A different file or a shorter one is a rewrite. + if (fileIdentity(now) !== source.identity || now.size < source.size) + throw new Error('History source changed after attribution; retry explicitly'); + buffer = await readTranscript(handle, now.size, budget); + if (buffer.length !== now.size) throw new Error('History source shrank while reading; retry explicitly'); } finally { await handle.close(); } + // Snapshot up to the last complete line: a running Codex may be mid-append. + const end = buffer.lastIndexOf(10) + 1; + const complete = buffer.subarray(0, end); + if (buffer.subarray(end).toString('latin1').trim()) budget.scan.partialRecords++; + if (!isUtf8(complete)) throw new Error('History source contains invalid UTF8'); + + const ancestors = new Set([source.parentId, source.forkedFromId, source.rootId].filter((id): id is string => !!id)); + const own: Message[] = []; + let legacy: LegacyPrefix | undefined; + let lastOrdinal = -1; + for (const line of complete.toString('utf8').split('\n')) { + if (!line.trim()) continue; + let parsed: unknown; + try { + parsed = JSON.parse(line); + } catch { + budget.scan.malformedRecords++; + continue; + } + const entry = object(parsed); + const ordinal = typeof entry.ordinal === 'number' ? entry.ordinal : undefined; + if (ordinal !== undefined) lastOrdinal = Math.max(lastOrdinal, ordinal); + if (source.historyStart !== undefined && ordinal === undefined) { + // Codex's projection skips a paginated record without an ordinal; so does this, visibly. + budget.scan.malformedRecords++; + continue; + } + const inherited = source.historyStart !== undefined && ordinal! < source.historyStart; + if (entry.type === 'session_meta') { + const declared = metadata(parsed, source.path); + if (declared && declared.sessionId === source.sessionId && sameIdentity(declared, source)) continue; + if (!declared || declared.sessionId === source.sessionId || !ancestors.has(declared.sessionId)) + throw new Error('History source contains conflicting native metadata; attribution refused'); + // An ancestor that was itself a fork carries its own ancestors' metadata in the copy. + for (const id of [declared.parentId, declared.forkedFromId, declared.rootId]) if (id) ancestors.add(id); + if (raw || inherited || legacy) continue; + budget.scan.inheritedPrefixes++; + legacy = { ancestorId: declared.sessionId, pieces: [] }; + continue; + } + if (inherited && !raw) continue; + if (legacy && entry.type === 'event_msg' && object(entry.payload).type === 'thread_settings_applied') { + legacy.pieces.push({ boundary: true }); + continue; + } + const item = message(parsed); + if (!item) continue; + if (legacy) legacy.pieces.push(item); + else own.push(item); + } + if (!raw && source.historyStart !== undefined && source.historyStart > 0) { + budget.scan.inheritedPrefixes++; + // Initialization died while copying the prefix: Codex refuses to resume such a file. + if (lastOrdinal < source.historyStart - 1) budget.scan.incompletePrefixes++; + } + if (legacy) own.push(...(await afterLegacyPrefix(legacy, budget, context))); + return collapse(own); } + +/** The child's own messages after a legacy inherited prefix. */ +async function afterLegacyPrefix(prefix: LegacyPrefix, budget: Budget, context: ReadContext): Promise { + const boundaries = prefix.pieces.flatMap((piece, i) => ('boundary' in piece ? [i] : [])); + if (!boundaries.length) { + budget.scan.incompletePrefixes++; + return []; + } + const messagesBefore = (index: number) => + collapse(prefix.pieces.slice(0, index).filter((piece): piece is Message => !('boundary' in piece))); + let cut = boundaries[0]; + if (boundaries.length > 1) { + // The last candidate whose preceding messages the parent's own file also holds. + const parent = await context.ancestorMessages(prefix.ancestorId); + const proven = parent ? boundaries.filter((candidate) => isPrefixOf(messagesBefore(candidate), parent)) : []; + if (proven.length) cut = proven.at(-1)!; + else { + // Unprovable: drop up to the last candidate, so parent text is never attributed to the child. + budget.scan.unverifiedPrefixes++; + cut = boundaries.at(-1)!; + } + } + return prefix.pieces.slice(cut + 1).filter((piece): piece is Message => !('boundary' in piece)); +} + function selectedTurns(messages: Message[], options: LocalHistoryOptions): Message[] { const turns: Message[][] = []; for (const item of messages) { @@ -345,9 +575,11 @@ function selectedTurns(messages: Message[], options: LocalHistoryOptions): Messa const dateMatches = (!options.after && !options.before) || turn.some((item) => { - const date = item.timestamp.slice(0, 10); + const date = utcDate(item.timestamp); return ( - validDate(date) && (!options.after || date >= options.after) && (!options.before || date <= options.before) + date !== undefined && + (!options.after || date >= options.after) && + (!options.before || date <= options.before) ); }); return dateMatches && (!needle || turn.some((item) => item.message.toLocaleLowerCase().includes(needle))); @@ -375,30 +607,46 @@ export async function readLocalHistory(options: LocalHistoryOptions): Promise source.sessionId.toLowerCase().startsWith(wanted)); - if (!matches.length) throw new Error('Native session missing from history store'); + const matches = all.filter((source) => source.sessionId.startsWith(wanted)); + if (!matches.length) + throw new Error( + scan.unattributedFiles + ? `Native session missing from history store (${scan.unattributedFiles} file(s) could not be attributed)` + : 'Native session missing from history store' + ); if (matches.length !== 1) throw new Error('Native session id is ambiguous'); const target = matches[0]; attributionSources = all; @@ -428,10 +676,23 @@ export async function readLocalHistory(options: LocalHistoryOptions): Promise a.timestamp.localeCompare(b.timestamp) || a.sessionId.localeCompare(b.sessionId)); const roots = new Map(sources.map((source) => [source.sessionId, nativeRoot(source, attributionSources)])); + const ancestorCache = new Map>(); + const context: ReadContext = { + // A raw read never opens a legacy prefix, so this never recurses. + ancestorMessages(id) { + const cached = ancestorCache.get(id); + if (cached) return cached; + const candidates = attributionSources.filter((source) => source.sessionId === id); + if (candidates.length !== 1) return Promise.resolve(undefined); + const read = readMessages(candidates[0], budget, context, true); + ancestorCache.set(id, read); + return read; + }, + }; const records: HistoryRecord[] = []; for (const source of sources) { records.push( - ...selectedTurns(await readMessages(source, budget), options).map((item) => ({ + ...selectedTurns(await readMessages(source, budget, context), options).map((item) => ({ rootSessionId: roots.get(source.sessionId) ?? null, sessionId: source.sessionId, role: item.role, @@ -449,6 +710,9 @@ export async function readLocalHistory(options: LocalHistoryOptions): Promise root !== null), sessions: sources.map((source) => ({ sessionId: source.sessionId, @@ -464,18 +728,18 @@ export async function readLocalHistory(options: LocalHistoryOptions): Promise HISTORY_MAX_OUTPUT_BYTES) - throw new Error('Invalid output byte budget'); + throw new Error(`Invalid --max-bytes (1-${HISTORY_MAX_OUTPUT_BYTES})`); const body = format === 'json' ? JSON.stringify(result) : [ - `source=${result.source} complete=${result.complete} scannedBytes=${result.scan.bytesRead}`, + `source=${result.source} complete=${result.complete} scannedBytes=${result.scan.bytesRead} metadataBytes=${result.scan.metadataBytesRead}`, ...result.records.map( (row) => - `[root=${row.rootSessionId} session=${row.sessionId} ${row.timestamp} ${row.role} ${row.phase}]\n${row.message}` + `[root=${row.rootSessionId} session=${row.sessionId} ${row.timestamp} ${row.role}${row.phase ? ` ${row.phase}` : ''}]\n${row.message}` ), ].join('\n\n'); if (Buffer.byteLength(body) + 1 > maxBytes) diff --git a/test/fixtures/codex-rollout.ts b/test/fixtures/codex-rollout.ts new file mode 100644 index 00000000..9f90f5c5 --- /dev/null +++ b/test/fixtures/codex-rollout.ts @@ -0,0 +1,125 @@ +/** + * @fileoverview Codex rollout lines shaped like real ones, for the offline + * history tests. The structure (keys, their order, the event types around a + * fork boundary) was taken from real rollouts written by codex 0.120 to 0.163 + * and from codex-rs `SessionMeta` (`protocol/src/protocol.rs`); no + * conversation content was copied. + * + * A forked or spawned thread's file, as `record_initial_history` writes it for + * `ForkPersistence::Copied`: + * + * own session_meta, parent session_meta, copied parent records, + * thread_settings_applied, the child's own records + * + * In paginated history every line also carries an `ordinal`, and the child's + * own `session_meta` declares `subagent_history_start_ordinal`. + */ +import { writeFile } from 'node:fs/promises'; +import { join } from 'node:path'; + +/** Codex's base instructions make every real session_meta line about 20 KiB. */ +export const BASE_INSTRUCTIONS_BYTES = 19 * 1024; + +export interface MetaOptions { + id: string; + /** Spawn parent: sets parent_thread_id, forked_from_id and source.subagent.thread_spawn. */ + parent?: string; + /** Shared root (`session_id`); defaults to `parent ?? id`, as Codex writes it. */ + root?: string; + mode?: 'legacy' | 'paginated'; + historyStart?: number; + timestamp?: string; + baseInstructionsBytes?: number; +} + +export function realMeta(options: MetaOptions): Record { + const { id, parent, mode = 'legacy', timestamp = '2026-10-24T01:00:00.000Z' } = options; + const payload: Record = { + session_id: options.root ?? parent ?? id, + id, + ...(parent ? { forked_from_id: parent, parent_thread_id: parent } : {}), + timestamp, + cwd: '/work/repo', + originator: 'codex_exec', + cli_version: '0.163.0', + source: parent + ? { + subagent: { + thread_spawn: { + parent_thread_id: parent, + depth: 1, + agent_path: null, + agent_nickname: 'Kepler', + agent_role: null, + }, + }, + } + : 'exec', + thread_source: parent ? 'subagent' : 'user', + ...(parent ? { agent_nickname: 'Kepler' } : {}), + model_provider: 'openai', + base_instructions: { text: 'b'.repeat(options.baseInstructionsBytes ?? BASE_INSTRUCTIONS_BYTES) }, + history_mode: mode, + ...(options.historyStart !== undefined ? { subagent_history_start_ordinal: options.historyStart } : {}), + ...(parent ? { multi_agent_version: 'v1' } : {}), + context_window: { window_id: `${id.slice(0, 8)}-window` }, + }; + return { timestamp, type: 'session_meta', payload }; +} + +export const taskStarted = () => ({ + timestamp: '2026-10-24T01:00:00.000Z', + type: 'event_msg', + payload: { type: 'task_started' }, +}); +export const settingsApplied = () => ({ + timestamp: '2026-10-24T01:00:00.000Z', + type: 'event_msg', + payload: { type: 'thread_settings_applied', thread_settings: { model: 'gpt-5.5', cwd: '/work/repo' } }, +}); +export const developer = () => ({ + timestamp: '2026-10-24T01:00:00.000Z', + type: 'response_item', + payload: { type: 'message', role: 'developer', content: [{ type: 'input_text', text: 'policy' }] }, +}); +export const injected = (tag = 'environment_context') => ({ + timestamp: '2026-10-24T01:00:00.000Z', + type: 'response_item', + payload: { type: 'message', role: 'user', content: [{ type: 'input_text', text: `<${tag}>\n ctx\n` }] }, +}); +/** A real user turn: the response_item Codex sent plus its user_message event twin. */ +export const userTurn = (text: string, timestamp = '2026-10-24T01:00:00.000Z') => [ + { + timestamp, + type: 'response_item', + payload: { type: 'message', role: 'user', content: [{ type: 'input_text', text }] }, + }, + { timestamp, type: 'event_msg', payload: { type: 'user_message', message: text, images: [] } }, +]; +/** An assistant message: the event and the response_item, both carrying Codex's phase. */ +export const assistantTurn = (text: string, phase: 'commentary' | 'final_answer' = 'final_answer') => [ + { + timestamp: '2026-10-24T01:00:01.000Z', + type: 'event_msg', + payload: { type: 'agent_message', message: text, phase }, + }, + { + timestamp: '2026-10-24T01:00:01.000Z', + type: 'response_item', + payload: { type: 'message', role: 'assistant', content: [{ type: 'output_text', text }], phase }, + }, +]; + +/** Number every line, as paginated history does. */ +export function withOrdinals(lines: Array>): Array> { + return lines.map((line, ordinal) => { + const { timestamp, type, payload, ...rest } = line; + return { timestamp, ordinal, type, payload, ...rest }; + }); +} + +export async function writeRollout(dir: string, name: string, lines: unknown[]): Promise { + const path = join(dir, name); + await writeFile(path, lines.map((line) => JSON.stringify(line)).join('\n') + '\n'); + return path; +} diff --git a/test/local-history-live.test.ts b/test/local-history-live.test.ts new file mode 100644 index 00000000..aa0348a6 --- /dev/null +++ b/test/local-history-live.test.ts @@ -0,0 +1,129 @@ +/** + * @fileoverview The offline history reader against rollouts that change while + * it reads them: a running Codex appending to its own rollout (or a child's), + * an unrelated rollout rewritten mid-scan, and a selected rollout replaced + * outright. `node:fs/promises` is wrapped so a test can act at an exact point + * of the read (after the open, after a read call) without timing races. + */ +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; +import { appendFile, mkdtemp, rename, rm, truncate, writeFile } from 'node:fs/promises'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; + +type Hook = (path: string, openCount: number, handle: import('node:fs/promises').FileHandle) => Promise | void; +const hooks: { onOpen?: Hook } = {}; +const opens = new Map(); + +vi.mock('node:fs/promises', async (importOriginal) => { + const actual = await importOriginal(); + return { + ...actual, + open: async (...args: Parameters) => { + const handle = await actual.open(...args); + const path = String(args[0]); + const count = (opens.get(path) ?? 0) + 1; + opens.set(path, count); + await hooks.onOpen?.(path, count, handle); + return handle; + }, + }; +}); + +const history = await import('../src/local-history.js'); +const { assistantTurn, realMeta, userTurn, writeRollout } = await import('./fixtures/codex-rollout.js'); + +const ROOT = 'aaaaaaaa-1111-4111-8111-111111111111'; +const CHILD = 'bbbbbbbb-1111-4111-8111-111111111111'; +let dir: string; +const line = (value: unknown) => `${JSON.stringify(value)}\n`; + +beforeEach(async () => { + dir = await mkdtemp(join(tmpdir(), 'local-history-live-')); + opens.clear(); + hooks.onOpen = undefined; +}); +afterEach(async () => { + hooks.onOpen = undefined; + await rm(dir, { recursive: true, force: true }); +}); + +describe('rollouts that change while they are read', () => { + it('reads a snapshot of a rollout a running Codex appends to after attribution', async () => { + const path = await writeRollout(dir, 'rollout-root.jsonl', [realMeta({ id: ROOT }), ...userTurn('question')]); + hooks.onOpen = async (opened, count) => { + // The second open is the transcript read; Codex writes a whole record and starts the next. + if (opened === path && count === 2) { + await appendFile(path, assistantTurn('answer').map(line).join('')); + await appendFile(path, '{"timestamp":"2026-10-24T01:00:02.000Z","type":"event_'); + } + }; + const result = await history.readLocalHistory({ sessionId: ROOT, sessionsRoot: dir }); + expect(result.records.map((r) => r.message)).toEqual(['question', 'answer']); + // The record cut mid-write is reported, not parsed and not fatal. + expect(result.scan.partialRecords).toBe(1); + expect(result.scan.malformedRecords).toBe(0); + expect(result.complete).toBe(false); + }); + + it('reads a family whose child is still running', async () => { + await writeRollout(dir, 'rollout-root.jsonl', [realMeta({ id: ROOT }), ...userTurn('root question')]); + const child = await writeRollout(dir, 'rollout-child.jsonl', [ + realMeta({ id: CHILD, parent: ROOT, mode: 'legacy' }), + ...userTurn('child task'), + ]); + hooks.onOpen = async (opened, count) => { + // Grows during the metadata scan and again before its transcript read. + if (opened === child) + await appendFile(child, assistantTurn(`child progress ${count}`, 'commentary').map(line).join('')); + }; + const result = await history.readLocalHistory({ rootSessionId: ROOT, sessionsRoot: dir }); + expect(result.records.map((r) => r.message)).toEqual([ + 'root question', + 'child task', + 'child progress 1', + 'child progress 2', + ]); + expect(result.complete).toBe(true); + }); + + it('counts an unrelated rollout that shrinks during its head read as unattributed', async () => { + await writeRollout(dir, 'rollout-root.jsonl', [realMeta({ id: ROOT }), ...userTurn('question')]); + const other = await writeRollout(dir, 'rollout-other.jsonl', [realMeta({ id: CHILD }), ...userTurn('other')]); + hooks.onOpen = async (opened, count, handle) => { + if (opened !== other || count !== 1) return; + const read = handle.read.bind(handle) as (...args: unknown[]) => Promise; + (handle as unknown as { read: (...args: unknown[]) => Promise }).read = async (...args) => { + const result = await read(...args); + await truncate(other, 10); + return result; + }; + }; + const result = await history.readLocalHistory({ sessionId: ROOT, sessionsRoot: dir }); + expect(result.records.map((r) => r.message)).toEqual(['question']); + expect(result.scan.unattributedFiles).toBe(1); + expect(result.complete).toBe(false); + }); + + it('still refuses a selected rollout replaced by another file after attribution', async () => { + const path = await writeRollout(dir, 'rollout-root.jsonl', [realMeta({ id: ROOT }), ...userTurn('question')]); + hooks.onOpen = async (opened, count) => { + if (opened !== path || count !== 1) return; + const replacement = join(dir, 'replacement.tmp'); + await writeFile(replacement, [realMeta({ id: ROOT }), ...userTurn('rewritten')].map(line).join('')); + await rename(replacement, path); + }; + await expect(history.readLocalHistory({ sessionId: ROOT, sessionsRoot: dir })).rejects.toThrow( + /changed after attribution/ + ); + }); + + it('still refuses a selected rollout truncated after attribution', async () => { + const path = await writeRollout(dir, 'rollout-root.jsonl', [realMeta({ id: ROOT }), ...userTurn('question')]); + hooks.onOpen = async (opened, count) => { + if (opened === path && count === 2) await truncate(path, 100); + }; + await expect(history.readLocalHistory({ sessionId: ROOT, sessionsRoot: dir })).rejects.toThrow( + /changed after attribution/ + ); + }); +}); diff --git a/test/local-history.test.ts b/test/local-history.test.ts index 7ffa4c4b..0a7cf6dc 100644 --- a/test/local-history.test.ts +++ b/test/local-history.test.ts @@ -1,10 +1,21 @@ import { afterEach, beforeEach, describe, expect, it } from 'vitest'; -import { mkdtemp, mkdir, rm, symlink, writeFile } from 'node:fs/promises'; +import { chmod, mkdtemp, mkdir, rm, symlink, writeFile } from 'node:fs/promises'; import { tmpdir } from 'node:os'; import { join } from 'node:path'; import * as history from '../src/local-history.js'; import { Command } from 'commander'; import { registerHistoryCommands } from '../src/cli-history.js'; +import { + assistantTurn, + developer, + injected, + realMeta, + settingsApplied, + taskStarted, + userTurn, + withOrdinals, + writeRollout, +} from './fixtures/codex-rollout.js'; const ROOT = 'aaaaaaaa-1111-4111-8111-111111111111'; const CHILD = 'bbbbbbbb-1111-4111-8111-111111111111'; @@ -155,9 +166,19 @@ describe('offline native history', () => { expect(result.complete).toBe(false); expect(result.scan.unattributedFiles).toBe(1); }); - it('refuses conflicting session metadata instead of attributing another conversation answer', async () => { + it('refuses a later session_meta whose id is not a declared ancestor', async () => { + // ROOT declares no parent, fork source or shared root, so CHILD can only be another conversation. const file = await fixture(ROOT, [row('user', 'root question'), meta(CHILD), row('assistant', 'child answer')]); await expect(history.readLocalHistory({ file })).rejects.toThrow(/conflicting.*metadata/); + // A fork child is refused too when the inherited metadata names a thread it does not descend from. + const forked = await writeRollout(dir, 'unrelated-fork.jsonl', [ + realMeta({ id: CHILD, parent: ROOT, mode: 'legacy' }), + realMeta({ id: GRANDCHILD, mode: 'legacy' }), + row('assistant', 'not ours'), + settingsApplied(), + row('assistant', 'child answer'), + ]); + await expect(history.readLocalHistory({ file: forked })).rejects.toThrow(/conflicting.*metadata/); }); it('accepts repeated metadata only when native identity and genealogy agree', async () => { const file = await fixture(ROOT, [row('user', 'question'), meta(ROOT), row('assistant', 'answer')]); @@ -244,3 +265,450 @@ describe('offline native history', () => { expect(program.commands.find((c) => c.name() === 'history')?.commands.map((c) => c.name())).toEqual(['read']); }); }); + +const messages = (result: history.LocalHistoryResult) => + result.records.map((r) => `${r.sessionId.slice(0, 1)}:${r.message}`); + +/** A parent R, and a legacy child C whose file copies R's history before its own, as Codex writes it. */ +async function legacyFamily(root: string, parentExtra: unknown[] = [], copiedExtra: unknown[] = []) { + await mkdir(root, { recursive: true }); + const parentHistory = [ + taskStarted(), + developer(), + injected(), + ...userTurn('parent question'), + ...assistantTurn('parent commentary', 'commentary'), + ...assistantTurn('parent answer'), + ]; + await writeRollout(root, `rollout-${ROOT}.jsonl`, [ + realMeta({ id: ROOT }), + ...parentHistory, + ...parentExtra, + ...userTurn('parent later question'), + ...assistantTurn('parent later answer'), + ]); + return writeRollout(root, `rollout-${CHILD}.jsonl`, [ + realMeta({ id: CHILD, parent: ROOT }), + realMeta({ id: ROOT }), + ...parentHistory, + ...copiedExtra, + settingsApplied(), + taskStarted(), + ...userTurn('child task'), + ...assistantTurn('child commentary', 'commentary'), + ...assistantTurn('child answer'), + ]); +} + +describe('forked and spawned threads', () => { + it('drops the parent history a legacy fork copies, so the child reads as itself', async () => { + const childFile = await legacyFamily(dir); + const child = await history.readLocalHistory({ sessionId: CHILD, sessionsRoot: dir }); + expect(messages(child)).toEqual(['b:child task', 'b:child commentary', 'b:child answer']); + expect(child.complete).toBe(true); + expect(child.scan.inheritedPrefixes).toBe(1); + expect(child.sessions[0]).toMatchObject({ sessionId: CHILD, parentSessionId: ROOT, rootSessionId: ROOT }); + // The whole family: every message once, the copy not repeated under the child. + const family = await history.readLocalHistory({ rootSessionId: ROOT, sessionsRoot: dir }); + expect(messages(family)).toEqual([ + 'a:parent question', + 'a:parent commentary', + 'a:parent answer', + 'a:parent later question', + 'a:parent later answer', + 'b:child task', + 'b:child commentary', + 'b:child answer', + ]); + expect(family.complete).toBe(true); + // An explicit file needs no store: the copy ends at the one thread_settings_applied. + const file = await history.readLocalHistory({ file: childFile }); + expect(messages(file)).toEqual(['b:child task', 'b:child commentary', 'b:child answer']); + expect(file.complete).toBe(true); + }); + + it('finds the end of a legacy copy that carries the parent own thread_settings_applied events', async () => { + // The parent compacted (or checkpointed its first turn) before the fork, so the copy holds that event too. + const parentSettings = [ + settingsApplied(), + ...userTurn('parent second question'), + ...assistantTurn('parent second answer'), + ]; + const childFile = await legacyFamily(dir, parentSettings, parentSettings); + const child = await history.readLocalHistory({ sessionId: CHILD, sessionsRoot: dir }); + expect(messages(child)).toEqual(['b:child task', 'b:child commentary', 'b:child answer']); + expect(child.complete).toBe(true); + // Without the parent's file the boundary cannot be proven: nothing of the parent is attributed to the child. + const alone = await history.readLocalHistory({ file: childFile }); + expect(messages(alone)).toEqual(['b:child task', 'b:child commentary', 'b:child answer']); + expect(alone.scan.unverifiedPrefixes).toBe(1); + expect(alone.complete).toBe(false); + }); + + it('keeps the child own later thread_settings_applied events on its side of the cut', async () => { + await legacyFamily(dir); + const path = join(dir, `rollout-${CHILD}.jsonl`); + const { appendFile } = await import('node:fs/promises'); + await appendFile( + path, + [settingsApplied(), ...userTurn('child second task'), ...assistantTurn('child second answer')] + .map((line) => JSON.stringify(line)) + .join('\n') + '\n' + ); + const child = await history.readLocalHistory({ sessionId: CHILD, sessionsRoot: dir }); + expect(messages(child)).toEqual([ + 'b:child task', + 'b:child commentary', + 'b:child answer', + 'b:child second task', + 'b:child second answer', + ]); + expect(child.complete).toBe(true); + }); + + it('accepts the copied metadata of an ancestor that was itself a fork', async () => { + const file = await writeRollout(dir, 'grandchild.jsonl', [ + realMeta({ id: GRANDCHILD, parent: CHILD, root: ROOT }), + realMeta({ id: CHILD, parent: ROOT }), + realMeta({ id: ROOT }), + ...userTurn('root question'), + settingsApplied(), + ...userTurn('child question'), + settingsApplied(), + ...userTurn('grandchild task'), + ...assistantTurn('grandchild answer'), + ]); + const result = await history.readLocalHistory({ file }); + expect(messages(result)).toEqual(['c:grandchild task', 'c:grandchild answer']); + expect(result.records[0].rootSessionId).toBe(ROOT); + }); + + it('marks a legacy copy that never reached its end marker incomplete and keeps none of it', async () => { + const file = await writeRollout(dir, 'cut.jsonl', [ + realMeta({ id: CHILD, parent: ROOT }), + realMeta({ id: ROOT }), + ...userTurn('parent question'), + ]); + await expect(history.readLocalHistory({ file })).rejects.toThrow(/evidence/); + await writeRollout(dir, 'cut.jsonl', [ + realMeta({ id: CHILD, parent: ROOT }), + ...userTurn('child before copy'), + realMeta({ id: ROOT }), + ...userTurn('parent question'), + ]); + const result = await history.readLocalHistory({ file }); + expect(messages(result)).toEqual(['b:child before copy']); + expect(result.scan.incompletePrefixes).toBe(1); + expect(result.complete).toBe(false); + }); + + it('drops the inherited records of a paginated sub-agent by ordinal, as Codex projects them', async () => { + const inherited = [ + realMeta({ id: CHILD, parent: ROOT, mode: 'paginated', historyStart: 7 }), + realMeta({ id: ROOT, mode: 'paginated' }), + taskStarted(), + injected(), + ...userTurn('parent question'), + ...assistantTurn('parent answer').slice(1), + ]; + expect(inherited).toHaveLength(7); + const lines = withOrdinals([ + ...inherited, + settingsApplied(), + taskStarted(), + ...assistantTurn('child commentary', 'commentary'), + ...assistantTurn('child answer'), + ]); + await writeRollout( + dir, + `rollout-${ROOT}.jsonl`, + withOrdinals([realMeta({ id: ROOT, mode: 'paginated' }), ...userTurn('parent question')]) + ); + await writeRollout(dir, `rollout-${CHILD}.jsonl`, lines); + const child = await history.readLocalHistory({ sessionId: CHILD, sessionsRoot: dir }); + expect(messages(child)).toEqual(['b:child commentary', 'b:child answer']); + expect(child.records.map((r) => r.phase)).toEqual(['commentary', 'final_answer']); + expect(child.complete).toBe(true); + const family = await history.readLocalHistory({ rootSessionId: ROOT, sessionsRoot: dir }); + expect(messages(family)).toEqual(['a:parent question', 'b:child commentary', 'b:child answer']); + }); + + it('reports a paginated copy that ends before its declared start ordinal', async () => { + const file = await writeRollout( + dir, + 'paginated-cut.jsonl', + withOrdinals([ + realMeta({ id: CHILD, parent: ROOT, mode: 'paginated', historyStart: 9 }), + realMeta({ id: ROOT, mode: 'paginated' }), + ...userTurn('parent question'), + ...assistantTurn('parent answer'), + ]) + ); + await expect(history.readLocalHistory({ file })).rejects.toThrow(/evidence/); + // In a family read the parent still answers, and the cut child makes the result incomplete. + await writeRollout( + dir, + `rollout-${ROOT}.jsonl`, + withOrdinals([realMeta({ id: ROOT, mode: 'paginated' }), ...userTurn('parent question')]) + ); + const family = await history.readLocalHistory({ rootSessionId: ROOT, sessionsRoot: dir }); + expect(messages(family)).toEqual(['a:parent question']); + expect(family.scan.incompletePrefixes).toBe(1); + expect(family.complete).toBe(false); + await rm(join(dir, `rollout-${ROOT}.jsonl`)); + // A paginated record without an ordinal is skipped, visibly, as Codex's projection skips it. + const lines = withOrdinals([ + realMeta({ id: CHILD, parent: ROOT, mode: 'paginated', historyStart: 2 }), + realMeta({ id: ROOT, mode: 'paginated' }), + settingsApplied(), + ...assistantTurn('child answer'), + ]); + await writeRollout(dir, 'paginated-cut.jsonl', [...lines, assistantTurn('stray')[0]]); + const result = await history.readLocalHistory({ file }); + expect(messages(result)).toEqual(['b:child answer']); + expect(result.scan.malformedRecords).toBe(1); + expect(result.complete).toBe(false); + }); +}); + +describe('large and untidy stores', () => { + it('finds one id among thousands of rollouts with real-sized session_meta at the default budgets', async () => { + const count = 3500; + const base = realMeta({ id: ROOT }); + const pad = JSON.stringify(base).length; + expect(pad).toBeGreaterThan(19 * 1024); + const writes: Array> = []; + for (let i = 0; i < count; i++) { + const day = join(dir, '2026', String(1 + (i % 12)).padStart(2, '0'), String(1 + (i % 28)).padStart(2, '0')); + const id = `${i.toString(16).padStart(8, '0')}-2222-4222-8222-222222222222`; + writes.push( + mkdir(day, { recursive: true }).then(() => + writeFile( + join(day, `rollout-${id}.jsonl`), + [realMeta({ id }), ...userTurn(`question ${i}`), ...assistantTurn(`answer ${i}`)] + .map((line) => JSON.stringify(line)) + .join('\n') + '\n' + ) + ) + ); + } + await Promise.all(writes); + const target = `${(count - 1).toString(16).padStart(8, '0')}-2222-4222-8222-222222222222`; + const result = await history.readLocalHistory({ sessionId: target, sessionsRoot: dir }); + expect(messages(result)).toEqual([`0:question ${count - 1}`, `0:answer ${count - 1}`]); + expect(result.complete).toBe(true); + // Metadata is charged to its own budget; the transcript budget only paid for the one rollout. + expect(result.scan.metadataBytesRead).toBeGreaterThan(history.HISTORY_MAX_SCAN_BYTES / 4); + expect(result.scan.bytesRead).toBeLessThan(64 * 1024); + // The store-wide caps fail closed and are raisable. + await expect( + history.readLocalHistory({ sessionId: target, sessionsRoot: dir, metadataBytes: 1024 * 1024 }) + ).rejects.toThrow(/metadata.*--metadata-bytes/); + await expect(history.readLocalHistory({ sessionId: target, sessionsRoot: dir, maxEntries: 100 })).rejects.toThrow( + /entry budget.*--max-entries/ + ); + }, 60_000); + + it('reads only the head of each rollout to attribute it, not its whole window', async () => { + await writeRollout(dir, 'rollout-big.jsonl', [ + realMeta({ id: ROOT }), + ...Array.from({ length: 400 }, (_, i) => assistantTurn(`${i} ${'x'.repeat(2048)}`)).flat(), + ]); + const result = await history.readLocalHistory({ sessionId: ROOT, sessionsRoot: dir, scanBytes: 8 * 1024 * 1024 }); + expect(result.records).toHaveLength(400); + // One 32 KiB step holds the ~20 KiB session_meta line; the 128 KiB window is only the ceiling. + expect(result.scan.metadataBytesRead).toBe(32 * 1024); + }); + + it.each([ + ['invalid UTF8 in its head', Buffer.from([0x7b, 0xff, 0xfe, 0x7d, 0x0a])], + [ + 'no session_meta within 128 KiB', + Buffer.from(`${JSON.stringify({ type: 'event_msg', payload: {} })}\n`.repeat(6000)), + ], + ])('counts one unrelated file with %s as unattributed instead of failing the lookup', async (_label, bytes) => { + await fixture(ROOT, [row('user', 'question'), row('assistant', 'answer')]); + await writeFile(join(dir, 'rollout-broken.jsonl'), bytes); + const result = await history.readLocalHistory({ sessionId: ROOT, sessionsRoot: dir }); + expect(result.records.map((r) => r.message)).toEqual(['question', 'answer']); + expect(result.scan.unattributedFiles).toBe(1); + expect(result.complete).toBe(false); + // The same file named explicitly is still an error. + await expect(history.readLocalHistory({ file: join(dir, 'rollout-broken.jsonl') })).rejects.toThrow(/UTF8|128 KiB/); + }); + + it('counts an unreadable directory instead of failing the lookup', async () => { + await fixture(ROOT, [row('assistant', 'answer')]); + const locked = join(dir, 'locked'); + await mkdir(locked); + await chmod(locked, 0); + try { + const result = await history.readLocalHistory({ sessionId: ROOT, sessionsRoot: dir }); + expect(result.records[0].message).toBe('answer'); + expect(result.scan.unreadableDirectories).toBe(process.getuid?.() === 0 ? 0 : 1); + } finally { + await chmod(locked, 0o700); + } + }); + + it('follows a sessions root that is itself a symlink, and still skips links below it', async () => { + const real = join(dir, 'real-sessions'); + await mkdir(real); + await writeRollout(real, 'rollout-root.jsonl', [ + realMeta({ id: ROOT }), + ...userTurn('question'), + ...assistantTurn('answer'), + ]); + const linkedRoot = join(dir, 'sessions-link'); + await symlink(real, linkedRoot); + const result = await history.readLocalHistory({ sessionId: ROOT, sessionsRoot: linkedRoot }); + expect(result.records.map((r) => r.message)).toEqual(['question', 'answer']); + expect(result.complete).toBe(true); + await symlink(join(real, 'rollout-root.jsonl'), join(real, 'rollout-link.jsonl')); + const withLink = await history.readLocalHistory({ sessionId: ROOT, sessionsRoot: linkedRoot }); + expect(withLink.scan.skippedSymlinks).toBe(1); + await expect( + history.readLocalHistory({ sessionId: ROOT, sessionsRoot: join(real, 'rollout-root.jsonl') }) + ).rejects.toThrow(/must be a directory/); + }); +}); + +describe('records, filters and output', () => { + it('gives user records no phase and keeps the phase Codex recorded on assistant messages', async () => { + const file = await writeRollout(dir, 'phases.jsonl', [ + realMeta({ id: ROOT }), + ...userTurn('question'), + ...assistantTurn('working', 'commentary'), + ...assistantTurn('done'), + row('assistant', 'legacy answer without a phase'), + ]); + const result = await history.readLocalHistory({ file }); + expect(result.records.map((r) => [r.role, r.phase])).toEqual([ + ['user', null], + ['assistant', 'commentary'], + ['assistant', 'final_answer'], + ['assistant', null], + ]); + }); + + it('drops every injected-context wrapper but never a typed prompt that starts with #', async () => { + const file = await writeRollout(dir, 'injected.jsonl', [ + realMeta({ id: ROOT }), + injected('environment_context'), + injected('recommended_plugins'), + injected('subagent_notification'), + { + type: 'response_item', + payload: { + type: 'message', + role: 'user', + content: [{ type: 'input_text', text: '# AGENTS.md instructions for /x' }], + }, + }, + ...userTurn('# a heading the user typed'), + ...assistantTurn('answer'), + ]); + const result = await history.readLocalHistory({ file }); + expect(result.records.map((r) => r.message)).toEqual(['# a heading the user typed', 'answer']); + }); + + it('compares date bounds as UTC calendar dates', async () => { + const file = await writeRollout(dir, 'dates.jsonl', [ + realMeta({ id: ROOT }), + // 23:30 at UTC-2 is 01:30 UTC the next day. + ...userTurn('late evening west of UTC', '2026-10-24T23:30:00.000-02:00'), + ...userTurn('plain UTC', '2026-10-24T12:00:00.000Z'), + ]); + const after = await history.readLocalHistory({ file, after: '2026-10-25' }); + expect(after.records.map((r) => r.message)).toEqual(['late evening west of UTC']); + const on = await history.readLocalHistory({ file, before: '2026-10-24' }); + expect(on.records.map((r) => r.message)).toEqual(['plain UTC']); + }); + + it('reads the default store at $CODEX_HOME/sessions, else ~/.codex/sessions', async () => { + const saved = { CODEX_HOME: process.env.CODEX_HOME, HOME: process.env.HOME }; + try { + process.env.CODEX_HOME = join(dir, 'codex-home'); + await mkdir(join(dir, 'codex-home', 'sessions', '2026', '10', '24'), { recursive: true }); + await writeRollout(join(dir, 'codex-home', 'sessions', '2026', '10', '24'), 'rollout-a.jsonl', [ + realMeta({ id: ROOT }), + ...userTurn('from CODEX_HOME'), + ]); + expect((await history.readLocalHistory({ sessionId: ROOT })).records[0].message).toBe('from CODEX_HOME'); + delete process.env.CODEX_HOME; + process.env.HOME = join(dir, 'home'); + await mkdir(join(dir, 'home', '.codex', 'sessions'), { recursive: true }); + await writeRollout(join(dir, 'home', '.codex', 'sessions'), 'rollout-b.jsonl', [ + realMeta({ id: ROOT }), + ...userTurn('from ~/.codex'), + ]); + const result = await history.readLocalHistory({ sessionId: ROOT }); + expect(result.records[0].message).toBe('from ~/.codex'); + expect(result.sessions[0].path).toBe(join(dir, 'home', '.codex', 'sessions', 'rollout-b.jsonl')); + } finally { + for (const [key, value] of Object.entries(saved)) { + if (value === undefined) delete process.env[key]; + else process.env[key] = value; + } + } + }); +}); + +describe('history CLI flags', () => { + const run = async (args: string[]) => { + let output = ''; + const program = new Command().exitOverride().configureOutput({ + writeOut: (text) => { + output += text; + }, + writeErr: () => {}, + }); + registerHistoryCommands(program); + await program.parseAsync(['history', 'read', ...args], { from: 'user' }); + return output; + }; + + it('prints text with --format text, a header line and one block per record', async () => { + const file = await writeRollout(dir, 'text.jsonl', [ + realMeta({ id: ROOT }), + ...userTurn('question'), + ...assistantTurn('answer'), + ]); + const output = await run(['--file', file, '--format', 'text']); + const blocks = output.trimEnd().split('\n\n'); + expect(blocks[0]).toMatch(/^source=codex-jsonl complete=true scannedBytes=\d+ metadataBytes=\d+$/); + expect(blocks[1]).toBe(`[root=${ROOT} session=${ROOT} 2026-10-24T01:00:00.000Z user]\nquestion`); + expect(blocks[2]).toBe(`[root=${ROOT} session=${ROOT} 2026-10-24T01:00:01.000Z assistant final_answer]\nanswer`); + expect(() => JSON.parse(output)).toThrow(); + }); + + it('refuses --json together with --format text, and accepts either alone', async () => { + const file = await writeRollout(dir, 'flags.jsonl', [realMeta({ id: ROOT }), ...assistantTurn('answer')]); + await expect(run(['--file', file, '--json', '--format', 'text'])).rejects.toThrow( + /--json conflicts with --format text/ + ); + expect(JSON.parse(await run(['--file', file, '--json'])).records[0].message).toBe('answer'); + expect(JSON.parse(await run(['--file', file, '--json', '--format', 'json'])).records[0].message).toBe('answer'); + expect(await run(['--file', file, '--format', 'text'])).toMatch(/^source=codex-jsonl/); + }); + + it('passes the store caps through and validates them', async () => { + await fixture(ROOT, [row('assistant', 'answer')]); + await fixture(CHILD, [row('assistant', 'child')], ROOT); + await expect(run(['--session-id', ROOT, '--sessions-root', dir, '--max-entries', '1'])).rejects.toThrow( + /--max-entries/ + ); + expect(JSON.parse(await run(['--session-id', ROOT, '--sessions-root', dir, '--max-entries', '2'])).complete).toBe( + true + ); + await expect(run(['--session-id', ROOT, '--sessions-root', dir, '--metadata-bytes', '16'])).rejects.toThrow( + /--metadata-bytes/ + ); + await expect(run(['--session-id', ROOT, '--sessions-root', dir, '--metadata-bytes', 'lots'])).rejects.toThrow( + /Invalid --metadata-bytes/ + ); + await expect(run(['--session-id', ROOT, '--sessions-root', dir, '--max-entries', '0'])).rejects.toThrow( + /Invalid --max-entries/ + ); + }); +});