diff --git a/packages/host/agent-adapter/src/__tests__/codex-history-files.test.ts b/packages/host/agent-adapter/src/__tests__/codex-history-files.test.ts new file mode 100644 index 000000000..73cac2a3b --- /dev/null +++ b/packages/host/agent-adapter/src/__tests__/codex-history-files.test.ts @@ -0,0 +1,153 @@ +import { createReadStream } from 'node:fs'; +import { mkdir, mkdtemp, rm, writeFile } from 'node:fs/promises'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; + +// Delegating spy: lets tests assert WHICH rollout files a lookup opened — the observable +// difference between the filename fast path and the whole-corpus fallback. +vi.mock('node:fs', async (importOriginal) => { + const actual = await importOriginal(); + return { ...actual, createReadStream: vi.fn(actual.createReadStream) }; +}); + +import { asHistoryId } from '../history-util'; +import { + findCodexTranscript, + readCodexIndex, + readCodexTranscriptSummaries, + readJsonlFile, +} from '../native/codex/history'; + +const THREAD_ID = '019f1111-2222-7333-8444-555566667777'; + +function rolloutLines(id: string): string[] { + return [ + JSON.stringify({ + timestamp: '2026-08-01T10:00:00.000Z', + type: 'session_meta', + payload: { id, cwd: '/repo', model: 'sol-1', cli_version: '0.144.1' }, + }), + JSON.stringify({ + timestamp: '2026-08-01T10:00:01.000Z', + type: 'event_msg', + payload: { type: 'user_message', message: 'real prompt' }, + }), + // Machine-injected row: marker-bearing, never echoed — must not count or title. + JSON.stringify({ + type: 'response_item', + payload: { + type: 'message', + role: 'user', + content: [{ type: 'input_text', text: '...' }], + }, + }), + JSON.stringify({ + timestamp: '2026-08-01T10:00:01.000Z', + type: 'response_item', + payload: { + type: 'message', + role: 'user', + content: [{ type: 'input_text', text: 'real prompt' }], + }, + }), + JSON.stringify({ + timestamp: '2026-08-01T10:00:05.000Z', + type: 'response_item', + payload: { + type: 'message', + role: 'assistant', + content: [{ type: 'output_text', text: 'an answer' }], + }, + }), + ]; +} + +/** Exercises the streaming rollout reads against a throwaway `CODEX_HOME` — the summary pass and + * the filename fast path both changed for the 2026-08 daemon OOM fix and must keep the whole-file + * pass's semantics. */ +describe('codex rollout file reads', () => { + let home: string; + + beforeEach(async () => { + home = await mkdtemp(join(tmpdir(), 'codex-history-')); + }); + + afterEach(async () => { + await rm(home, { recursive: true, force: true }); + }); + + async function writeRollout(relative: string, lines: string[]): Promise { + const path = join(home, relative); + await mkdir(join(path, '..'), { recursive: true }); + await writeFile(path, `${lines.join('\n')}\n`); + return path; + } + + it('summarizes a rollout in one streaming pass with the whole-file semantics', async () => { + await writeRollout( + `sessions/2026/08/01/rollout-2026-08-01T10-00-00-${THREAD_ID}.jsonl`, + rolloutLines(THREAD_ID), + ); + const summaries = await readCodexTranscriptSummaries(await readCodexIndex(home), home); + + expect(summaries).toHaveLength(1); + expect(summaries[0]).toMatchObject({ + id: THREAD_ID, + cwd: '/repo', + model: 'sol-1', + title: 'real prompt', + messageCount: 2, + createdAt: Date.parse('2026-08-01T10:00:00.000Z'), + updatedAt: Date.parse('2026-08-01T10:00:05.000Z'), + }); + }); + + it('skips corrupt lines and ignores empty files', async () => { + const path = await writeRollout( + `sessions/2026/08/01/rollout-2026-08-01T10-00-00-${THREAD_ID}.jsonl`, + [...rolloutLines(THREAD_ID), '{"truncated": '], + ); + await writeRollout('sessions/2026/08/01/rollout-empty.jsonl', ['']); + + expect(await readJsonlFile(path)).toHaveLength(5); + const summaries = await readCodexTranscriptSummaries(await readCodexIndex(home), home); + expect(summaries.map((summary) => summary.id)).toEqual([THREAD_ID]); + }); + + it('finds a transcript through the filename fast path without opening the rest', async () => { + // A decoy whose name carries a different id must not satisfy the lookup — and the fast path + // must never even open it (a fallback full scan would, which is the OOM this guards against). + const decoyPath = await writeRollout( + 'sessions/2026/08/01/rollout-2026-08-01T09-00-00-019f0000-aaaa-7bbb-8ccc-dddd00000000.jsonl', + rolloutLines('019f0000-aaaa-7bbb-8ccc-dddd00000000'), + ); + await writeRollout( + `archived_sessions/rollout-2026-08-01T10-00-00-${THREAD_ID}.jsonl`, + rolloutLines(THREAD_ID), + ); + + vi.mocked(createReadStream).mockClear(); + const found = await findCodexTranscript(asHistoryId(THREAD_ID), home); + expect(found).toMatchObject({ id: THREAD_ID, title: 'real prompt' }); + const opened = vi.mocked(createReadStream).mock.calls.map((call) => String(call[0])); + expect(opened).not.toContain(decoyPath); + }); + + it('falls back to the full scan when the filename does not carry the id', async () => { + await writeRollout('sessions/renamed-rollout.jsonl', rolloutLines(THREAD_ID)); + + const found = await findCodexTranscript(asHistoryId(THREAD_ID), home); + expect(found).toMatchObject({ id: THREAD_ID, cwd: '/repo' }); + }); + + it('returns undefined for an unknown id', async () => { + await writeRollout( + `sessions/rollout-2026-08-01T10-00-00-${THREAD_ID}.jsonl`, + rolloutLines(THREAD_ID), + ); + expect( + await findCodexTranscript(asHistoryId('019f9999-0000-7000-8000-000000000000'), home), + ).toBeUndefined(); + }); +}); diff --git a/packages/host/agent-adapter/src/__tests__/codex-history.test.ts b/packages/host/agent-adapter/src/__tests__/codex-history.test.ts index 61a25fa93..740a023a8 100644 --- a/packages/host/agent-adapter/src/__tests__/codex-history.test.ts +++ b/packages/host/agent-adapter/src/__tests__/codex-history.test.ts @@ -346,6 +346,22 @@ describe('mapCodexHistoryEvents', () => { } }); + it('does not pair distinct marker prompts containing lone surrogates', () => { + const events = mapCodexHistoryEvents(HID, [ + { + type: 'event_msg', + payload: { type: 'user_message', message: '# AGENTS.md instructions\u{D800}' }, + }, + responseItem({ + type: 'message', + role: 'user', + content: [{ type: 'input_text', text: '# AGENTS.md instructions\u{D801}' }], + }), + ]); + + expect(events).toHaveLength(0); + }); + it('still drops a glued row when an unmarked twin of one part was echoed as a real prompt', () => { // The echoed prompt text coincides with the glued row's env part; the AGENTS.md part was // never echoed, so the injected row must stay filtered while the real prompt replays. diff --git a/packages/host/agent-adapter/src/native/codex/adapter.ts b/packages/host/agent-adapter/src/native/codex/adapter.ts index b6453c428..d8b0ef9c3 100644 --- a/packages/host/agent-adapter/src/native/codex/adapter.ts +++ b/packages/host/agent-adapter/src/native/codex/adapter.ts @@ -1469,6 +1469,8 @@ export class CodexAdapter extends BaseAgentAdapter { this.pendingCompactionId = null; this.emit({ type: 'compaction', compactionId: pending, status: 'completed' }); } + // A turn ended by cancel/exit never reaches handleTurnCompleted, which owns the normal clear. + this.streamedTextLen.clear(); super.teardown(); } diff --git a/packages/host/agent-adapter/src/native/codex/history.ts b/packages/host/agent-adapter/src/native/codex/history.ts index 90a3cd1a2..0f5b51f03 100644 --- a/packages/host/agent-adapter/src/native/codex/history.ts +++ b/packages/host/agent-adapter/src/native/codex/history.ts @@ -1,8 +1,11 @@ +import { createHash } from 'node:crypto'; import type { Stats } from 'node:fs'; -import { readdir, readFile, stat } from 'node:fs/promises'; +import { createReadStream } from 'node:fs'; +import { readdir, stat } from 'node:fs/promises'; import { homedir } from 'node:os'; import { basename, join } from 'node:path'; import { env } from 'node:process'; +import { createInterface } from 'node:readline'; import type { AgentHistoryEvent, AgentHistoryId, @@ -17,6 +20,7 @@ import { textBlock, } from '@linkcode/schema'; import { appendArrayInPlace } from 'foxts/append-array-in-place'; +import { createFixedArray } from 'foxts/create-fixed-array'; import { not } from 'foxts/guard'; import { encodeHistoryBranchCursor } from '../../history-branch'; import { @@ -113,37 +117,73 @@ function isCodexImageMarker(parts: unknown[], index: number): boolean { * so the row is rescued when every marker-bearing part is echoed as an `event_msg`/`user_message` * (real prompts always are, both TUI- and app-server-written; injected rows never are). Only the * marked parts count — an unmarked part that happens to equal a real prompt must not drag the - * injected parts of a glued row back in. Rollouts without event_msg rows degrade to marker-only. */ + * injected parts of a glued row back in. Rollouts without event_msg rows degrade to marker-only. + * Echo comparison is by {@link promptTextFingerprint}, so scans never retain prompt bodies. */ export function isSyntheticCodexUserPayload( payload: JsonRecord, - realPromptTexts?: ReadonlySet, + realPromptFingerprints?: ReadonlySet, ): boolean { + return isSyntheticCodexUserDigest(digestCodexUserPayload(payload), realPromptFingerprints); +} + +/** Equality-only stand-in for a prompt text, so holding one per row stays a few dozen bytes even + * when the text is a pasted file or an injected AGENTS.md blob. */ +function promptTextFingerprint(text: string): string { + return createHash('sha256').update(text, 'utf16le').digest('base64url'); +} + +/** The parts of a user row the synthetic judgment needs, small enough to hold for every user row + * of a rollout while the surrounding rows stream by (the judgment needs the complete echoed-prompt + * set, which is only known at end of file). */ +interface CodexUserRowDigest { + markedFingerprints: string[]; + hasImage: boolean; + echoedFingerprint: string; +} + +function digestCodexUserPayload(payload: JsonRecord): CodexUserRowDigest { const content = payload.content; const parts = Array.isArray(content) ? content : [payload]; const texts = parts.map((part) => textFromUnknown(part)); - const marked = texts.filter((text) => isSyntheticCodexUserText(text)); - if (marked.length === 0) return false; - if (!realPromptTexts) return true; - if (marked.every((text) => realPromptTexts.has(text))) return false; const hasImage = parts.some( (part) => isRecord(part) && stringField(part, 'type') === 'input_image', ); - if (!hasImage) return true; - const echoedText = texts.filter((_text, index) => !isCodexImageMarker(parts, index)).join(''); - return !realPromptTexts.has(echoedText); + return { + markedFingerprints: texts.flatMap((text) => + isSyntheticCodexUserText(text) ? [promptTextFingerprint(text)] : [], + ), + hasImage, + echoedFingerprint: hasImage + ? promptTextFingerprint( + texts.filter((_text, index) => !isCodexImageMarker(parts, index)).join(''), + ) + : '', + }; +} + +function isSyntheticCodexUserDigest( + digest: CodexUserRowDigest, + realPromptFingerprints?: ReadonlySet, +): boolean { + if (digest.markedFingerprints.length === 0) return false; + if (!realPromptFingerprints) return true; + if (digest.markedFingerprints.every((print) => realPromptFingerprints.has(print))) return false; + if (!digest.hasImage) return true; + return !realPromptFingerprints.has(digest.echoedFingerprint); } -/** The texts codex echoed as `event_msg`/`user_message` rows — the real prompts of the rollout. */ -export function collectCodexPromptTexts(rows: JsonRecord[]): Set { - const texts = new Set(); +/** Fingerprints of the texts codex echoed as `event_msg`/`user_message` rows — the real prompts + * of the rollout. */ +export function collectCodexPromptFingerprints(rows: JsonRecord[]): Set { + const prints = new Set(); for (const row of rows) { if (stringField(row, 'type') !== 'event_msg') continue; const payload = recordField(row, 'payload'); if (!payload || stringField(payload, 'type') !== 'user_message') continue; const message = stringField(payload, 'message'); - if (message) texts.add(message); + if (message) prints.add(promptTextFingerprint(message)); } - return texts; + return prints; } /** Convert Codex's persisted response content without trusting an arbitrary URL or local path. @@ -251,15 +291,27 @@ export async function readCodexIndex(home = codexHome()): Promise, home = codexHome(), ): Promise { - const roots = [join(home, 'sessions'), join(home, 'archived_sessions')]; - const fileSets = await Promise.all(roots.map((root) => collectJsonlFiles(root))); - const files = fileSets.flat(); - const summaries = await Promise.all(files.map((file) => readCodexTranscriptSummary(file, index))); - return summaries.filter(not(undefined)); + const files = await collectCodexRolloutFiles(home); + const summaries: CodexTranscriptSummary[] = []; + const workers = createFixedArray(Math.min(SUMMARY_READ_CONCURRENCY, files.length)).map( + async () => { + for (let file = files.pop(); file !== undefined; file = files.pop()) { + // eslint-disable-next-line no-await-in-loop -- the loop is one bounded-concurrency worker. + const summary = await readCodexTranscriptSummary(file, index); + if (summary) summaries.push(summary); + } + }, + ); + await Promise.all(workers); + return summaries; } export async function findCodexTranscript( @@ -267,11 +319,31 @@ export async function findCodexTranscript( home = codexHome(), ): Promise { const index = await readCodexIndex(home); + const id: string = historyId; + // Rollout filenames end with the thread id (`rollout--.jsonl`), so a suffix match reads + // one file instead of the whole corpus; session_meta stays the identity check. + const files = await collectCodexRolloutFiles(home); + const candidates = files.filter((path) => { + const name = basename(path, '.jsonl'); + return path === id || name === id || name.endsWith(`-${id}`); + }); + const fastSummaries = await Promise.all( + candidates.map((candidate) => readCodexTranscriptSummary(candidate, index)), + ); + const fastHit = fastSummaries + .filter(not(undefined)) + .find((summary) => summary.id === id || summary.path === id); + if (fastHit) return fastHit; const summaries = await readCodexTranscriptSummaries(index, home); - const id = historyId; return summaries.find((summary) => summary.id === id || summary.path === id); } +async function collectCodexRolloutFiles(home: string): Promise { + const roots = [join(home, 'sessions'), join(home, 'archived_sessions')]; + const fileSets = await Promise.all(roots.map((root) => collectJsonlFiles(root))); + return fileSets.flat(); +} + async function collectJsonlFiles(root: string, depth = 8): Promise { if (depth < 0) return []; let entries: DirectoryEntry[]; @@ -292,39 +364,60 @@ async function collectJsonlFiles(root: string, depth = 8): Promise { } export async function readJsonlFile(path: string): Promise { - let raw: string; - try { - raw = await readFile(path, 'utf8'); - } catch { - return []; - } const rows: JsonRecord[] = []; - for (const line of raw.split('\n')) { - if (line.trim().length === 0) continue; - try { - const parsed: unknown = JSON.parse(line); - if (isRecord(parsed)) rows.push(parsed); - } catch { - // Ignore corrupt partial lines; Codex may be writing the active transcript. + await forEachJsonlRow(path, (row) => rows.push(row)); + return rows; +} + +/** Stream one rollout's rows without ever holding the whole file — a single transcript can be + * over 100MB, and whole-file reads were half of the 2026-08 daemon OOM. Returns the row count. */ +async function forEachJsonlRow(path: string, onRow: (row: JsonRecord) => void): Promise { + const lines = createInterface({ + input: createReadStream(path, { encoding: 'utf8' }), + crlfDelay: Number.POSITIVE_INFINITY, + }); + let count = 0; + try { + for await (const line of lines) { + if (line.trim().length === 0) continue; + try { + const parsed: unknown = JSON.parse(line); + if (isRecord(parsed)) { + count += 1; + onRow(parsed); + } + } catch { + // Ignore corrupt partial lines; Codex may be writing the active transcript. + } } + } catch { + // An unreadable or vanished file reads as empty, matching the old readFile fallback. + } finally { + lines.close(); } - return rows; + return count; +} + +/** A user row's summary contribution, deferred to end of stream: whether it counts (and previews) + * depends on the complete echoed-prompt set. Holds previews and digests, never the row itself. */ +interface PendingUserRow { + digest: CodexUserRowDigest; + empty: boolean; + preview?: string; } async function readCodexTranscriptSummary( path: string, index: Map, ): Promise { - const [rows, fileStat] = await Promise.all([readJsonlFile(path), statOrUndefined(path)]); - if (rows.length === 0) return undefined; - const promptTexts = collectCodexPromptTexts(rows); + const promptFingerprints = new Set(); + const userRows: PendingUserRow[] = []; let id: string | undefined; let cwd: string | undefined; let model: string | undefined; let createdAt: number | undefined; let updatedAt: number | undefined; - let firstUserText: string | undefined; let firstAssistantText: string | undefined; let messageCount = 0; let cliVersion: string | undefined; @@ -333,7 +426,7 @@ async function readCodexTranscriptSummary( let modelProvider: string | undefined; let gitBranch: string | undefined; - for (const row of rows) { + const rowCount = await forEachJsonlRow(path, (row) => { const rowType = stringField(row, 'type'); const rowTs = timestampMs(row.timestamp); if (rowTs !== undefined) { @@ -342,9 +435,15 @@ async function readCodexTranscriptSummary( } const payload = recordField(row, 'payload'); - if (!payload) continue; + if (!payload) return; switch (rowType) { + case 'event_msg': { + if (stringField(payload, 'type') !== 'user_message') break; + const message = stringField(payload, 'message'); + if (message) promptFingerprints.add(promptTextFingerprint(message)); + break; + } case 'session_meta': { id = stringField(payload, 'id') ?? id; cwd = stringField(payload, 'cwd') ?? cwd; @@ -367,26 +466,36 @@ async function readCodexTranscriptSummary( } case 'response_item': { const role = stringField(payload, 'role'); - if (role !== 'user' && role !== 'assistant') continue; - let text: string; if (role === 'user') { - if (isSyntheticCodexUserPayload(payload, promptTexts)) continue; const content = codexUserContent(payload.content); - if (content.length === 0) continue; - text = content.flatMap((block) => (block.type === 'text' ? [block.text] : [])).join('\n'); - } else { - text = textFromUnknown(payload); - if (text.trim().length === 0) continue; + const text = content + .flatMap((block) => (block.type === 'text' ? [block.text] : [])) + .join('\n'); + userRows.push({ + digest: digestCodexUserPayload(payload), + empty: content.length === 0, + ...(text.trim().length > 0 && { preview: previewText(text) }), + }); + } else if (role === 'assistant') { + const text = textFromUnknown(payload); + if (text.trim().length === 0) break; + messageCount += 1; + firstAssistantText ??= previewText(text); } - messageCount += 1; - if (role === 'user' && text.trim().length > 0) firstUserText ??= previewText(text); - else if (role === 'assistant') firstAssistantText ??= previewText(text); - break; } default: break; } + }); + if (rowCount === 0) return undefined; + const fileStat = await statOrUndefined(path); + + let firstUserText: string | undefined; + for (const userRow of userRows) { + if (isSyntheticCodexUserDigest(userRow.digest, promptFingerprints) || userRow.empty) continue; + messageCount += 1; + if (userRow.preview !== undefined) firstUserText ??= userRow.preview; } id ??= idFromFilename(path); @@ -518,7 +627,7 @@ export function mapCodexHistoryEvents( const events: AgentHistoryEvent[] = []; const announced = new Map(); const persistedMcpIdentities = collectCodexMcpIdentities(rows); - const promptTexts = collectCodexPromptTexts(rows); + const promptFingerprints = collectCodexPromptFingerprints(rows); const { callIds: respondedCallIds, outputCallIds: responseOutputCallIds } = collectRespondedToolCallIds(rows); const mcpEndStates = collectMcpEndStates(rows); @@ -615,7 +724,7 @@ export function mapCodexHistoryEvents( const role = stringField(payload, 'role'); if (role !== 'user' && role !== 'assistant') return; - if (role === 'user' && isSyntheticCodexUserPayload(payload, promptTexts)) return; + if (role === 'user' && isSyntheticCodexUserPayload(payload, promptFingerprints)) return; const itemId = stringField(payload, 'id') ?? stringField(row, 'id') ?? `${role}-${index.toString(36)}`; const event = diff --git a/packages/host/engine/src/__tests__/history-service.test.ts b/packages/host/engine/src/__tests__/history-service.test.ts index 39c606608..83e7ba4be 100644 --- a/packages/host/engine/src/__tests__/history-service.test.ts +++ b/packages/host/engine/src/__tests__/history-service.test.ts @@ -54,6 +54,20 @@ describe('HistoryService', () => { expect(state.lastReadOptions?.mcpServerNames).toBeUndefined(); }); + it('evicts expired cache entries instead of keeping dead transcripts', async () => { + const state: FakeHistoryState = { listCalls: 0, readCalls: 0, resumeCalls: 0 }; + let now = 0; + const service = new HistoryService(fakeHistoryFactory(state), { ttlMs: 1000, now: () => now }); + + await Effect.runPromise(service.list('codex', { cwd: '/repo' })); + await Effect.runPromise(service.read('codex', { historyId })); + expect(service.cacheSizes()).toEqual({ list: 1, events: 1 }); + + now = 1000; + await Effect.runPromise(service.list('codex', { cwd: '/other' })); + expect(service.cacheSizes()).toEqual({ list: 1, events: 0 }); + }); + it('removes injected resource context from provider history', async () => { const state = { listCalls: 0, diff --git a/packages/host/engine/src/session/history-service.ts b/packages/host/engine/src/session/history-service.ts index 4fc0b80d9..5cae16e76 100644 --- a/packages/host/engine/src/session/history-service.ts +++ b/packages/host/engine/src/session/history-service.ts @@ -72,8 +72,9 @@ export class HistoryService { opts: HistoryListOptions = {}, ): Effect.Effect { const key = listCacheKey(kind, opts); - const cached = this.listCache.get(key); const now = this.now(); + this.sweepExpired(now); + const cached = this.listCache.get(key); if (cached && !opts.forceRefresh && cached.expiresAt > now) { return Effect.succeed(cloneListResult(cached.result)); } @@ -115,8 +116,9 @@ export class HistoryService { const limit = boundedLimit(opts.limit, 1000, 1000); const key = eventCacheKey(kind, opts.historyId); const cwd = opts.cwd ?? this.historyCwdById.get(key); - const cached = this.eventCache.get(key); const now = this.now(); + this.sweepExpired(now); + const cached = this.eventCache.get(key); if ( cached && @@ -266,6 +268,21 @@ export class HistoryService { this.historyCwdById.clear(); } + /** Cache occupancy, for eviction tests and diagnostics. */ + cacheSizes(): { list: number; events: number } { + return { list: this.listCache.size, events: this.eventCache.size }; + } + + /** Expired entries are dead weight — a full event cache entry holds a whole transcript. */ + private sweepExpired(now: number): void { + for (const [key, entry] of this.listCache) { + if (entry.expiresAt <= now) this.listCache.delete(key); + } + for (const [key, entry] of this.eventCache) { + if (entry.expiresAt <= now) this.eventCache.delete(key); + } + } + private invalidateEventCacheFromList(kind: AgentKind, sessions: AgentHistorySession[]): void { for (const session of sessions) { const key = eventCacheKey(kind, session.historyId);