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/ + ); + }); +});