diff --git a/src/main/ai-vault/session-scanner-codex-discovery.ts b/src/main/ai-vault/session-scanner-codex-discovery.ts new file mode 100644 index 00000000000..8aadf78fb72 --- /dev/null +++ b/src/main/ai-vault/session-scanner-codex-discovery.ts @@ -0,0 +1,40 @@ +import type { AiVaultScanIssue } from '../../shared/ai-vault-types' +import { + CODEX_SESSION_ROLLOUT_EXTENSIONS, + isCodexSessionRolloutPath +} from './session-scanner-codex-paths' +import { discoverFiles } from './session-scanner-discovery' +import type { SessionFileDiscovery } from './session-scanner-types' + +export function codexSessionDiscoveries( + rootDirs: readonly string[], + limit: number, + issues: AiVaultScanIssue[] +): Promise[] { + return rootDirs.map((rootDir) => + discoverFiles({ + rootDir, + limit, + agent: 'codex', + issues, + extensions: [...CODEX_SESSION_ROLLOUT_EXTENSIONS], + // Why: `.zst` alone is too broad; only Codex `.jsonl.zst` rollouts qualify. + filePredicate: isCodexSessionRolloutPath, + // Why: sibling duplicates must not consume the recency limit, and the + // writable plain rollout remains authoritative even when zstd is newer. + candidatePathSelector: preferPlainCodexRolloutSiblings + }) + ) +} + +function preferPlainCodexRolloutSiblings(paths: readonly string[]): string[] { + const byPlainPath = new Map() + for (const path of paths) { + const plainPath = path.endsWith('.jsonl.zst') ? path.slice(0, -'.zst'.length) : path + const existing = byPlainPath.get(plainPath) + if (!existing || path.endsWith('.jsonl')) { + byPlainPath.set(plainPath, path) + } + } + return [...byPlainPath.values()] +} diff --git a/src/main/ai-vault/session-scanner-codex-parser.test.ts b/src/main/ai-vault/session-scanner-codex-parser.test.ts index f48d58f495a..6c68d0d74ac 100644 --- a/src/main/ai-vault/session-scanner-codex-parser.test.ts +++ b/src/main/ai-vault/session-scanner-codex-parser.test.ts @@ -2,7 +2,7 @@ import { mkdir, mkdtemp, rm, stat, writeFile } from 'node:fs/promises' import { tmpdir } from 'node:os' import { dirname, join } from 'node:path' import { afterEach, describe, expect, it } from 'vitest' -import { parseCodexSessionFile } from './session-scanner-codex-parser' +import { parseCodexSessionContent, parseCodexSessionFile } from './session-scanner-codex-parser' let tempRoots: string[] = [] @@ -16,6 +16,56 @@ function jsonLines(records: unknown[]): string { } describe('parseCodexSessionFile', () => { + it('parses paginated item_completed turn items for preview and title', async () => { + const session = await parseCodexSessionContent({ + file: { + path: '/tmp/rollout-paginated.jsonl', + mtimeMs: 1, + modifiedAt: '2026-06-18T10:00:00.000Z' + }, + content: jsonLines([ + { + timestamp: '2026-06-18T10:00:00.000Z', + type: 'session_meta', + payload: { + id: 'paginated-session', + cwd: '/repo/app', + history_mode: 'paginated' + } + }, + { + timestamp: '2026-06-18T10:00:01.000Z', + type: 'event_msg', + payload: { + type: 'item_completed', + item: { + type: 'user_message', + id: 'item-user-1', + content: [{ type: 'text', text: 'Paginated user prompt' }] + } + } + }, + { + timestamp: '2026-06-18T10:00:02.000Z', + type: 'event_msg', + payload: { + type: 'item_completed', + item: { + type: 'AgentMessage', + id: 'item-agent-1', + content: [{ type: 'text', text: 'Paginated assistant reply' }] + } + } + } + ]) + }) + + expect(session?.sessionId).toBe('paginated-session') + expect(session?.title).toBe('Paginated user prompt') + expect(session?.messageCount).toBe(2) + expect(session?.previewMessages?.map((message) => message.role)).toEqual(['user', 'assistant']) + }) + it('does not double-count usage when token count formats switch', async () => { const root = await mkdtemp(join(tmpdir(), 'orca-ai-vault-codex-token-switch-')) tempRoots.push(root) diff --git a/src/main/ai-vault/session-scanner-codex-parser.ts b/src/main/ai-vault/session-scanner-codex-parser.ts index 5127720b141..76ac51ea525 100644 --- a/src/main/ai-vault/session-scanner-codex-parser.ts +++ b/src/main/ai-vault/session-scanner-codex-parser.ts @@ -1,35 +1,22 @@ -import { createReadStream } from 'node:fs' -import { createInterface } from 'node:readline' import type { AiVaultSession } from '../../shared/ai-vault-types' import { readCodexSessionIndexTitle } from './session-scanner-codex-title-index' import type { ExecutionHostId } from '../../shared/execution-host' import { - addPreviewContent, cloneSessionAccumulator, createAccumulator, - finalizeSession, - sessionIdFromFileName, - updateTimeline + finalizeSession } from './session-scanner-accumulator' +import { codexRolloutBaseName } from './session-scanner-codex-paths' +import { + consumeCodexRecordLine, + type CodexSessionParseState +} from './session-scanner-codex-record-consume' +import { iterateCodexRolloutLines } from './session-scanner-codex-rollout-read' import type { - CodexUsageSnapshot, FileWithMtime, ResumableParseFinalizeOptions, - ResumableSessionParseState, - SessionAccumulator + ResumableSessionParseState } from './session-scanner-types' -import { - addCodexUsage, - asRecord, - extractContentText, - extractGitBranch, - extractModel, - extractString, - normalizeCodexUsage, - normalizeTitleText, - parseJsonObject, - subtractCodexUsage -} from './session-scanner-values' export async function parseCodexSessionFile( file: FileWithMtime, @@ -37,19 +24,19 @@ export async function parseCodexSessionFile( codexHome: string | null = null, executionHostId?: ExecutionHostId ): Promise { - const lines = createInterface({ - input: createReadStream(file.path, { encoding: 'utf-8' }), - crlfDelay: Infinity - }) - - return parseCodexSessionLines({ - file, - lines, - platform, - codexHome, - executionHostId, - titleReader: (sessionId) => readCodexSessionIndexTitle(file.path, codexHome, sessionId) - }) + const lines = iterateCodexRolloutLines(file.path) + try { + return await parseCodexSessionLines({ + file, + lines, + platform, + codexHome, + executionHostId, + titleReader: (sessionId) => readCodexSessionIndexTitle(file.path, codexHome, sessionId) + }) + } finally { + lines.close() + } } export async function parseCodexSessionContent(args: { @@ -72,22 +59,12 @@ export async function parseCodexSessionContent(args: { }) } -type CodexSessionParseState = { - accumulator: SessionAccumulator - previousTotals: CodexUsageSnapshot | null - rejectedWorkerSession: boolean - sawSessionMeta: boolean - // Which source set the current title; an index-file title outranks the raw - // first user prompt, so finalize must know whether 'meta' already won. - titleSource: 'meta' | 'user' | null -} - function createCodexParseState(file: FileWithMtime): CodexSessionParseState { return { accumulator: createAccumulator({ agent: 'codex', file, - sessionId: sessionIdFromFileName(file.path) + sessionId: sessionIdFromCodexRolloutPath(file.path) }), previousTotals: null, rejectedWorkerSession: false, @@ -96,6 +73,12 @@ function createCodexParseState(file: FileWithMtime): CodexSessionParseState { } } +function sessionIdFromCodexRolloutPath(filePath: string): string { + const baseName = codexRolloutBaseName(filePath) + const match = baseName.match(/[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$/i) + return match?.[0] ?? baseName +} + function cloneCodexParseState(state: CodexSessionParseState): CodexSessionParseState { return { // previousTotals snapshots are replaced, never mutated, so sharing is safe. @@ -104,124 +87,6 @@ function cloneCodexParseState(state: CodexSessionParseState): CodexSessionParseS } } -function consumeCodexRecordLine(state: CodexSessionParseState, line: string): void { - if (state.rejectedWorkerSession) { - return - } - const record = parseJsonObject(line) - if (!record) { - return - } - const { accumulator } = state - - updateTimeline(accumulator, extractString(record.timestamp)) - - const payload = asRecord(record.payload) - if (record.type === 'session_meta' && payload) { - if (isCodexWorkerSession(payload)) { - // Why: Codex writes internal worker/sub-agent transcripts into the same - // history tree; AI Vault should show user-started sessions only. - state.rejectedWorkerSession = true - return - } - state.sawSessionMeta = true - const sessionId = extractString(payload.id) - if (sessionId) { - accumulator.sessionId = sessionId - } - const metadataTitle = extractCodexSessionMetadataTitle(payload) - if (metadataTitle) { - accumulator.title = metadataTitle - state.titleSource = 'meta' - } - const cwd = extractString(payload.cwd) - if (cwd) { - accumulator.cwd = cwd - } - accumulator.branch = extractGitBranch(payload.git) ?? accumulator.branch - return - } - - if (record.type === 'turn_context' && payload) { - const cwd = extractString(payload.cwd) - if (cwd) { - accumulator.cwd = cwd - } - const model = extractModel(payload) - if (model) { - accumulator.model = model - } - return - } - - if (!payload) { - return - } - - if (record.type === 'response_item' && payload.type === 'message') { - accumulator.messageCount++ - if (payload.role === 'user' && !accumulator.title) { - accumulator.title = extractContentText(payload.content) - state.titleSource = accumulator.title ? 'user' : state.titleSource - } - addPreviewContent( - accumulator, - payload.role === 'assistant' ? 'assistant' : payload.role === 'user' ? 'user' : 'unknown', - payload.content, - record.timestamp - ) - return - } - - if (record.type !== 'event_msg') { - return - } - - if (payload.type === 'user_message') { - accumulator.messageCount++ - if (!accumulator.title) { - accumulator.title = extractContentText(payload.message) - state.titleSource = accumulator.title ? 'user' : state.titleSource - } - addPreviewContent(accumulator, 'user', payload.message, record.timestamp) - return - } - - if (payload.type === 'agent_message') { - accumulator.messageCount++ - addPreviewContent(accumulator, 'assistant', payload.message, record.timestamp) - return - } - - if (payload.type !== 'token_count') { - return - } - - const info = asRecord(payload.info) - if (!info) { - return - } - const totalUsage = normalizeCodexUsage(info.total_token_usage) - const lastUsage = normalizeCodexUsage(info.last_token_usage) - let delta: CodexUsageSnapshot | null = null - if (totalUsage) { - delta = subtractCodexUsage(totalUsage, state.previousTotals) - state.previousTotals = totalUsage - } else if (lastUsage) { - delta = lastUsage - state.previousTotals = state.previousTotals - ? addCodexUsage(state.previousTotals, lastUsage) - : lastUsage - } - if (delta) { - accumulator.totalTokens += delta.totalTokens - } - const model = extractModel(payload) - if (model) { - accumulator.model = model - } -} - async function finalizeCodexParseState( state: CodexSessionParseState, platform: NodeJS.Platform, @@ -303,25 +168,3 @@ async function parseCodexSessionLines(args: { executionHostPlatform: args.executionHostPlatform }) } - -function extractCodexThreadSource(payload: Record): string | null { - return extractString(payload.thread_source) ?? extractString(payload.threadSource) -} - -function isCodexWorkerSession(payload: Record): boolean { - const threadSource = extractCodexThreadSource(payload) - if (threadSource) { - return threadSource.toLowerCase() !== 'user' - } - - const source = asRecord(payload.source) - return Boolean(asRecord(source?.subagent)) -} - -function extractCodexSessionMetadataTitle(payload: Record): string | null { - return ( - normalizeTitleText(extractString(payload.title) ?? '') ?? - normalizeTitleText(extractString(payload.thread_name) ?? '') ?? - normalizeTitleText(extractString(payload.threadName) ?? '') - ) -} diff --git a/src/main/ai-vault/session-scanner-codex-paths.ts b/src/main/ai-vault/session-scanner-codex-paths.ts index 584833dc54c..1dbbddb2e6c 100644 --- a/src/main/ai-vault/session-scanner-codex-paths.ts +++ b/src/main/ai-vault/session-scanner-codex-paths.ts @@ -1,4 +1,7 @@ -import { dirname, resolve } from 'node:path' +import { basename, dirname, resolve } from 'node:path' + +/** Extensions accepted by Codex session discovery (plain + cold zstd). */ +export const CODEX_SESSION_ROLLOUT_EXTENSIONS = ['.jsonl', '.zst'] as const export function codexHomeForSessionsDir( sessionsDir: string, @@ -25,3 +28,29 @@ export function uniqueCodexSessionsDirs(paths: readonly string[]): string[] { } return unique } + +/** True for Codex rollout logs: `*.jsonl` or cold-compressed `*.jsonl.zst`. */ +export function isCodexSessionRolloutFileName(fileName: string): boolean { + return fileName.endsWith('.jsonl') || fileName.endsWith('.jsonl.zst') +} + +export function isCodexSessionRolloutPath(filePath: string): boolean { + return isCodexSessionRolloutFileName(basename(filePath)) +} + +/** True when the rollout was cold-compressed by Codex (`*.jsonl.zst`). */ +export function isCodexCompressedRolloutPath(filePath: string): boolean { + return filePath.endsWith('.jsonl.zst') +} + +/** Basename without rollout suffixes so UUID extraction also works for cold sessions. */ +export function codexRolloutBaseName(filePath: string): string { + const name = basename(filePath) + if (name.endsWith('.jsonl.zst')) { + return name.slice(0, -'.jsonl.zst'.length) + } + if (name.endsWith('.jsonl')) { + return name.slice(0, -'.jsonl'.length) + } + return name +} diff --git a/src/main/ai-vault/session-scanner-codex-record-consume.ts b/src/main/ai-vault/session-scanner-codex-record-consume.ts new file mode 100644 index 00000000000..bc1d53cb192 --- /dev/null +++ b/src/main/ai-vault/session-scanner-codex-record-consume.ts @@ -0,0 +1,204 @@ +import { addPreviewContent, updateTimeline } from './session-scanner-accumulator' +import type { CodexUsageSnapshot, SessionAccumulator } from './session-scanner-types' +import { + addCodexUsage, + asRecord, + extractContentText, + extractGitBranch, + extractModel, + extractString, + normalizeCodexUsage, + normalizeTitleText, + parseJsonObject, + subtractCodexUsage +} from './session-scanner-values' + +export type CodexSessionParseState = { + accumulator: SessionAccumulator + previousTotals: CodexUsageSnapshot | null + rejectedWorkerSession: boolean + sawSessionMeta: boolean + // An index/metadata title outranks the first user prompt at finalize time. + titleSource: 'meta' | 'user' | null +} + +export function consumeCodexRecordLine(state: CodexSessionParseState, line: string): void { + if (state.rejectedWorkerSession) { + return + } + const record = parseJsonObject(line) + if (!record) { + return + } + const { accumulator } = state + + updateTimeline(accumulator, extractString(record.timestamp)) + + const payload = asRecord(record.payload) + if (record.type === 'session_meta' && payload) { + if (isCodexWorkerSession(payload)) { + // Why: Codex writes worker transcripts into the same tree; AI Vault only + // lists user-started sessions. + state.rejectedWorkerSession = true + return + } + state.sawSessionMeta = true + const sessionId = extractString(payload.id) + if (sessionId) { + accumulator.sessionId = sessionId + } + const metadataTitle = extractCodexSessionMetadataTitle(payload) + if (metadataTitle) { + accumulator.title = metadataTitle + state.titleSource = 'meta' + } + const cwd = extractString(payload.cwd) + if (cwd) { + accumulator.cwd = cwd + } + accumulator.branch = extractGitBranch(payload.git) ?? accumulator.branch + return + } + + if (record.type === 'turn_context' && payload) { + const cwd = extractString(payload.cwd) + if (cwd) { + accumulator.cwd = cwd + } + const model = extractModel(payload) + if (model) { + accumulator.model = model + } + return + } + + if (!payload) { + return + } + + if (record.type === 'response_item' && payload.type === 'message') { + accumulator.messageCount++ + if (payload.role === 'user' && !accumulator.title) { + accumulator.title = extractContentText(payload.content) + state.titleSource = accumulator.title ? 'user' : state.titleSource + } + addPreviewContent( + accumulator, + payload.role === 'assistant' ? 'assistant' : payload.role === 'user' ? 'user' : 'unknown', + payload.content, + record.timestamp + ) + return + } + + if (record.type !== 'event_msg') { + return + } + + if (payload.type === 'user_message') { + accumulator.messageCount++ + if (!accumulator.title) { + accumulator.title = extractContentText(payload.message) + state.titleSource = accumulator.title ? 'user' : state.titleSource + } + addPreviewContent(accumulator, 'user', payload.message, record.timestamp) + return + } + + if (payload.type === 'agent_message') { + accumulator.messageCount++ + addPreviewContent(accumulator, 'assistant', payload.message, record.timestamp) + return + } + + // Why: paginated history persists TurnItems instead of the legacy event + // message variants. + if (payload.type === 'item_completed') { + consumeCodexCompletedTurnItem(state, payload, record.timestamp) + return + } + + if (payload.type !== 'token_count') { + return + } + + const info = asRecord(payload.info) + if (!info) { + return + } + const totalUsage = normalizeCodexUsage(info.total_token_usage) + const lastUsage = normalizeCodexUsage(info.last_token_usage) + let delta: CodexUsageSnapshot | null = null + if (totalUsage) { + delta = subtractCodexUsage(totalUsage, state.previousTotals) + state.previousTotals = totalUsage + } else if (lastUsage) { + delta = lastUsage + state.previousTotals = state.previousTotals + ? addCodexUsage(state.previousTotals, lastUsage) + : lastUsage + } + if (delta) { + accumulator.totalTokens += delta.totalTokens + } + const model = extractModel(payload) + if (model) { + accumulator.model = model + } +} + +function consumeCodexCompletedTurnItem( + state: CodexSessionParseState, + payload: Record, + timestamp: unknown +): void { + const item = asRecord(payload.item) + if (!item) { + return + } + const itemType = normalizeCodexTurnItemType(item.type) + const { accumulator } = state + if (itemType === 'user_message') { + const text = extractContentText(item.content) + accumulator.messageCount++ + if (!accumulator.title && text) { + accumulator.title = text + state.titleSource = 'user' + } + addPreviewContent(accumulator, 'user', item.content, timestamp) + } else if (itemType === 'agent_message') { + accumulator.messageCount++ + addPreviewContent(accumulator, 'assistant', item.content, timestamp) + } +} + +function normalizeCodexTurnItemType(value: unknown): string | null { + const raw = extractString(value) + if (!raw) { + return null + } + return raw + .replace(/([a-z0-9])([A-Z])/g, '$1_$2') + .replace(/([A-Z]+)([A-Z][a-z])/g, '$1_$2') + .toLowerCase() +} + +function extractCodexThreadSource(payload: Record): string | null { + return extractString(payload.thread_source) ?? extractString(payload.threadSource) +} + +function isCodexWorkerSession(payload: Record): boolean { + const threadSource = extractCodexThreadSource(payload) + if (threadSource) { + return threadSource.toLowerCase() !== 'user' + } + return Boolean(asRecord(asRecord(payload.source)?.subagent)) +} + +function extractCodexSessionMetadataTitle(payload: Record): string | null { + return ( + normalizeTitleText(extractString(payload.title) ?? '') ?? + normalizeTitleText(extractString(payload.thread_name) ?? '') ?? + normalizeTitleText(extractString(payload.threadName) ?? '') + ) +} diff --git a/src/main/ai-vault/session-scanner-codex-rollout-read.ts b/src/main/ai-vault/session-scanner-codex-rollout-read.ts new file mode 100644 index 00000000000..5c600647ace --- /dev/null +++ b/src/main/ai-vault/session-scanner-codex-rollout-read.ts @@ -0,0 +1,46 @@ +import { createReadStream } from 'node:fs' +import { createInterface } from 'node:readline' +import { compose, type Readable } from 'node:stream' +import { createZstdDecompress, type ZstdDecompress } from 'node:zlib' +import { isCodexCompressedRolloutPath } from './session-scanner-codex-paths' + +/** Opens a UTF-8 stream for either a plain or cold-compressed Codex rollout. */ +export function openCodexRolloutStream(filePath: string): Readable { + const raw = createReadStream(filePath) + if (!isCodexCompressedRolloutPath(filePath)) { + return raw + } + + // Why: Codex cold-compresses rollouts to `.jsonl.zst`; readers must accept + // both forms without loading the entire transcript into memory. + let decoder: ZstdDecompress + try { + decoder = createZstdDecompress() + } catch (error) { + raw.destroy() + throw new Error(`Zstd decompression is unavailable; cannot read ${filePath}`, { + cause: error + }) + } + // `compose` owns the complete source→decoder pipeline: source errors reach + // readers, and destroying the returned stream tears down both resources. + return compose(raw, decoder) +} + +/** Async lines plus an explicit close hook for early worker-session rejection. */ +export function iterateCodexRolloutLines( + filePath: string +): AsyncIterable & { close: () => void } { + const input = openCodexRolloutStream(filePath) + const lines = createInterface({ input, crlfDelay: Infinity }) + return { + [Symbol.asyncIterator]: () => lines[Symbol.asyncIterator](), + close: () => { + lines.close() + // Early worker-session rejection intentionally aborts the composed zstd + // pipeline; consume that close-only AbortError after parsing has stopped. + input.once('error', () => undefined) + input.destroy() + } + } +} diff --git a/src/main/ai-vault/session-scanner-codex-zstd.test.ts b/src/main/ai-vault/session-scanner-codex-zstd.test.ts new file mode 100644 index 00000000000..6a7a155e6ce --- /dev/null +++ b/src/main/ai-vault/session-scanner-codex-zstd.test.ts @@ -0,0 +1,200 @@ +import { mkdir, mkdtemp, rm, utimes, writeFile } from 'node:fs/promises' +import { tmpdir } from 'node:os' +import { dirname, join } from 'node:path' +import { zstdCompressSync } from 'node:zlib' +import { afterEach, describe, expect, it } from 'vitest' +import { scanAiVaultSessions } from './session-scanner' +import { parseCodexSessionFile } from './session-scanner-codex-parser' +import { openCodexRolloutStream } from './session-scanner-codex-rollout-read' +import { resetSessionParseCacheForTests } from './session-scanner-parse-cache' +import { isolatedScanRoots, jsonLines } from './session-scanner-test-fixtures' + +let tempRoots: string[] = [] + +afterEach(async () => { + resetSessionParseCacheForTests() + await Promise.all(tempRoots.map((root) => rm(root, { recursive: true, force: true }))) + tempRoots = [] +}) + +describe('Codex cold-compressed rollout scanning', () => { + it('discovers and fully reparses updated .jsonl.zst sessions', async () => { + const root = await mkdtemp(join(tmpdir(), 'orca-ai-vault-codex-zst-')) + tempRoots.push(root) + const roots = isolatedScanRoots(root) + const sessionId = '019f0000-1111-7222-8333-444444444444' + const sessionPath = join( + roots.codexSessionsDir, + '2026', + '06', + '18', + `rollout-2026-06-18T10-00-00-${sessionId}.jsonl.zst` + ) + await mkdir(dirname(sessionPath), { recursive: true }) + + const records = [ + { + timestamp: '2026-06-18T10:00:00.000Z', + type: 'session_meta', + payload: { id: sessionId, cwd: '/repo/app' } + }, + { + timestamp: '2026-06-18T10:00:01.000Z', + type: 'event_msg', + payload: { + type: 'item_completed', + item: { + type: 'user_message', + id: 'u1', + content: [{ type: 'text', text: 'Compressed rollout prompt' }] + } + } + } + ] + await writeFile(sessionPath, zstdCompressSync(Buffer.from(jsonLines(records), 'utf-8'))) + + const first = await scanAiVaultSessions({ ...roots, platform: 'darwin' }) + expect(first.issues).toEqual([]) + expect(first.sessions).toHaveLength(1) + expect(first.sessions[0]).toMatchObject({ + sessionId, + title: 'Compressed rollout prompt', + messageCount: 1 + }) + + records.push({ + timestamp: '2026-06-18T10:00:02.000Z', + type: 'event_msg', + payload: { + type: 'item_completed', + item: { + type: 'agent_message', + id: 'a1', + content: [{ type: 'text', text: 'Compressed reply' }] + } + } + }) + await writeFile(sessionPath, zstdCompressSync(Buffer.from(jsonLines(records), 'utf-8'))) + + const second = await scanAiVaultSessions({ ...roots, platform: 'darwin' }) + expect(second.issues).toEqual([]) + expect(second.sessions[0]?.messageCount).toBe(2) + }) + + it('prefers a plain rollout over its compressed sibling', async () => { + const root = await mkdtemp(join(tmpdir(), 'orca-ai-vault-codex-zst-sibling-')) + tempRoots.push(root) + const roots = isolatedScanRoots(root) + const sessionId = '019f0000-2222-7333-8444-555555555555' + const basePath = join( + roots.codexSessionsDir, + '2026', + '06', + '18', + `rollout-2026-06-18T10-00-00-${sessionId}.jsonl` + ) + await mkdir(dirname(basePath), { recursive: true }) + const records = [ + { + timestamp: '2026-06-18T10:00:00.000Z', + type: 'session_meta', + payload: { id: sessionId, cwd: '/repo/app' } + } + ] + const content = jsonLines(records) + await writeFile(`${basePath}.zst`, zstdCompressSync(Buffer.from(content, 'utf-8'))) + await writeFile(basePath, content) + + const result = await scanAiVaultSessions({ ...roots, platform: 'darwin' }) + + expect(result.issues).toEqual([]) + expect(result.sessions).toHaveLength(1) + expect(result.sessions[0]?.filePath).toBe(basePath) + }) + + it('selects plain siblings before applying the per-agent discovery limit', async () => { + const root = await mkdtemp(join(tmpdir(), 'orca-ai-vault-codex-zst-limit-')) + tempRoots.push(root) + const roots = isolatedScanRoots(root) + const sessionId = '019f0000-3333-7444-8555-666666666666' + const basePath = join( + roots.codexSessionsDir, + '2026', + '06', + '18', + `rollout-2026-06-18T10-00-00-${sessionId}.jsonl` + ) + await mkdir(dirname(basePath), { recursive: true }) + const content = jsonLines([ + { + timestamp: '2026-06-18T10:00:00.000Z', + type: 'session_meta', + payload: { id: sessionId, cwd: '/repo/app' } + } + ]) + await writeFile(basePath, content) + await writeFile(`${basePath}.zst`, zstdCompressSync(Buffer.from(content, 'utf-8'))) + await utimes( + basePath, + new Date('2026-06-18T10:00:00.000Z'), + new Date('2026-06-18T10:00:00.000Z') + ) + await utimes( + `${basePath}.zst`, + new Date('2026-06-18T10:00:01.000Z'), + new Date('2026-06-18T10:00:01.000Z') + ) + + const result = await scanAiVaultSessions({ + ...roots, + platform: 'darwin', + limitPerAgent: 1 + }) + + expect(result.issues).toEqual([]) + expect(result.sessions).toHaveLength(1) + expect(result.sessions[0]?.filePath).toBe(basePath) + }) + + it('propagates missing and corrupt compressed rollout errors', async () => { + const root = await mkdtemp(join(tmpdir(), 'orca-ai-vault-codex-zst-errors-')) + tempRoots.push(root) + const missingPath = join(root, 'missing.jsonl.zst') + await expect( + parseCodexSessionFile({ + path: missingPath, + mtimeMs: 0, + modifiedAt: new Date(0).toISOString() + }) + ).rejects.toThrow() + + const corruptPath = join(root, 'corrupt.jsonl.zst') + await writeFile(corruptPath, 'not-zstd-data', 'utf-8') + await expect( + parseCodexSessionFile({ + path: corruptPath, + mtimeMs: 0, + modifiedAt: new Date(0).toISOString() + }) + ).rejects.toThrow() + }) + + it('closes the complete compressed stream when a reader stops early', async () => { + const root = await mkdtemp(join(tmpdir(), 'orca-ai-vault-codex-zst-close-')) + tempRoots.push(root) + const sessionPath = join(root, 'rollout.jsonl.zst') + const content = + `${JSON.stringify({ type: 'session_meta', payload: { id: 'session' } })}\n`.repeat(10_000) + await writeFile(sessionPath, zstdCompressSync(Buffer.from(content, 'utf-8'))) + + const stream = openCodexRolloutStream(sessionPath) + const closed = new Promise((resolve) => stream.once('close', resolve)) + // `compose().destroy()` reports an expected AbortError while closing the + // owned pipeline; a consumer may intentionally stop before EOF. + stream.once('error', () => undefined) + stream.destroy() + await closed + + expect(stream.destroyed).toBe(true) + }) +}) diff --git a/src/main/ai-vault/session-scanner-discovery.ts b/src/main/ai-vault/session-scanner-discovery.ts index c176fb3c6e5..49f7b475117 100644 --- a/src/main/ai-vault/session-scanner-discovery.ts +++ b/src/main/ai-vault/session-scanner-discovery.ts @@ -12,12 +12,14 @@ export async function discoverFiles(args: { extensions: string[] filePredicate?: (path: string) => boolean directoryPredicate?: (name: string) => boolean + candidatePathSelector?: (paths: readonly string[]) => string[] }): Promise { - const paths = await walkSessionFiles(args.rootDir, args.agent, args.issues, { + const discoveredPaths = await walkSessionFiles(args.rootDir, args.agent, args.issues, { extensions: new Set(args.extensions), filePredicate: args.filePredicate, directoryPredicate: args.directoryPredicate }) + const paths = args.candidatePathSelector?.(discoveredPaths) ?? discoveredPaths const files: FileWithMtime[] = [] for (const path of paths) { try { diff --git a/src/main/ai-vault/session-scanner-parse-cache.ts b/src/main/ai-vault/session-scanner-parse-cache.ts index 70d0d311d6b..8b51bf5265d 100644 --- a/src/main/ai-vault/session-scanner-parse-cache.ts +++ b/src/main/ai-vault/session-scanner-parse-cache.ts @@ -50,7 +50,11 @@ function resumableStateFactoryFor( case 'claude': return () => createClaudeSessionResumeState(candidate.file) case 'codex': - return () => createCodexSessionResumeState(candidate.file, candidate.codexHome) + // Why: byte offsets only describe plain append-only JSONL. Cold zstd + // sessions must be decompressed from the beginning after each change. + return candidate.file.path.endsWith('.jsonl.zst') + ? null + : () => createCodexSessionResumeState(candidate.file, candidate.codexHome) case 'cursor': return () => createCursorSessionResumeState(candidate.file) case 'copilot': diff --git a/src/main/ai-vault/session-scanner-source-discovery.ts b/src/main/ai-vault/session-scanner-source-discovery.ts index f77404bfc5a..d6e2801d148 100644 --- a/src/main/ai-vault/session-scanner-source-discovery.ts +++ b/src/main/ai-vault/session-scanner-source-discovery.ts @@ -1,6 +1,7 @@ import { homedir } from 'node:os' import { basename, join } from 'node:path' import type { AiVaultScanIssue } from '../../shared/ai-vault-types' +import { codexSessionDiscoveries } from './session-scanner-codex-discovery' import { uniqueCodexSessionsDirs } from './session-scanner-codex-paths' import { SUBAGENT_DIR_NAME } from './session-scanner-subagent-transcripts' import { discoverFiles, discoverOpenClawFiles } from './session-scanner-discovery' @@ -74,7 +75,7 @@ export async function discoverAiVaultSessionSources(args: { // and the SQLite scanner (1.17.x); dedup by sessionId happens inside. ...opencodeDiscoveries(options, wslHomeDirs, limitPerAgent, issues), ...claudeDiscoveries(options, wslHomeDirs, limitPerAgent, issues), - ...codexDiscoveries(codexSessionsDirs, limitPerAgent, issues), + ...codexSessionDiscoveries(codexSessionsDirs, limitPerAgent, issues), ...standardDiscoveries(options, wslHomeDirs, limitPerAgent, issues), openClawDiscovery(options, wslHomeDirs, limitPerAgent, issues), ...droidDiscoveries(options, wslHomeDirs, limitPerAgent, issues), @@ -107,16 +108,6 @@ function claudeDiscoveries( ) } -function codexDiscoveries( - rootDirs: readonly string[], - limit: number, - issues: AiVaultScanIssue[] -): Promise[] { - return rootDirs.map((rootDir) => - discoverFiles({ rootDir, limit, agent: 'codex', issues, extensions: ['.jsonl'] }) - ) -} - function standardDiscoveries( options: AiVaultScanOptions, wslHomeDirs: readonly string[], diff --git a/src/main/codex/codex-session-bridge-link.ts b/src/main/codex/codex-session-bridge-link.ts new file mode 100644 index 00000000000..0a5ea083323 --- /dev/null +++ b/src/main/codex/codex-session-bridge-link.ts @@ -0,0 +1,301 @@ +import { + constants, + copyFileSync, + existsSync, + linkSync, + lstatSync, + readlinkSync, + renameSync, + rmSync, + symlinkSync +} from 'node:fs' +import { + clearCopiedCodexSessionMarker, + codexSessionSourceMatchesCopiedPrefix, + codexSessionStatsMatchMarker, + fingerprintCodexSessionFile, + readCopiedCodexSessionMarker, + writeCopiedCodexSessionMarker +} from './codex-session-copy-markers' +import { + hasPreservedCodexSession, + preservedCodexSessionPaths +} from './codex-session-preserved-copies' +import { installWithPreservedCodexSession } from './codex-session-preserved-install' + +type TargetIdentityCheck = (candidatePath: string) => boolean + +export function tryHardlinkSystemCodexSessionFile(sourcePath: string, targetPath: string): boolean { + try { + // Why: Codex resume ignores symlinks, while a hardlink keeps one physical log. + linkSync(sourcePath, targetPath) + return true + } catch { + return false + } +} + +export function tryCopySystemCodexSessionFile( + sourcePath: string, + targetPath: string, + relativePath: string +): boolean { + const replacementPath = `${targetPath}.orca-copy-${process.pid}-${Date.now()}` + try { + copyFileSync(sourcePath, replacementPath, constants.COPYFILE_EXCL) + // Why: the complete temp copy is published with an exclusive same-volume + // hardlink, so no process can observe partial bytes or be overwritten. + linkSync(replacementPath, targetPath) + } catch (error) { + console.warn('[codex-session-bridge] Failed to copy system Codex session:', sourcePath, error) + return false + } finally { + rmSync(replacementPath, { force: true }) + } + try { + writeCopiedCodexSessionMarker(relativePath, sourcePath, targetPath) + } catch (error) { + // The complete exclusive copy remains usable but deliberately unowned; a + // marker failure must not delete a file that another process may have opened. + console.warn('[codex-session-bridge] Failed to mark copied Codex session:', sourcePath, error) + } + return true +} + +/** Replaces an existing target with rollback, preferring a hardlink over a copy. */ +export function replaceSystemCodexSessionBridge( + sourcePath: string, + targetPath: string, + relativePath: string, + targetIdentityCheck: TargetIdentityCheck = () => true, + preserveOriginal = false +): boolean { + const replacementPath = `${targetPath}.orca-link-${process.pid}-${Date.now()}` + try { + const usesHardlink = tryHardlinkSystemCodexSessionFile(sourcePath, replacementPath) + if (!usesHardlink) { + copyFileSync(sourcePath, replacementPath) + } + return installPreparedCodexSessionBridge({ + sourcePath, + targetPath, + relativePath, + replacementPath, + usesHardlink, + targetIdentityCheck, + preserveOriginal + }) + } catch (error) { + console.warn( + '[codex-session-bridge] Failed to replace Codex session bridge:', + sourcePath, + error + ) + return false + } finally { + rmSync(replacementPath, { force: true }) + } +} + +/** Migrates an unchanged legacy copy only when a hardlink is now available. */ +export function migrateCopiedCodexSessionBridge( + sourcePath: string, + targetPath: string, + relativePath: string +): boolean { + if (hasPreservedCodexSession(relativePath)) { + warnPreservedCodexSessionRequiresReview(relativePath) + return false + } + const marker = readCopiedCodexSessionMarker(relativePath) + if (!marker || marker.sourcePath !== sourcePath) { + return false + } + let replacementPath: string | null = null + try { + replacementPath = `${targetPath}.orca-link-${process.pid}-${Date.now()}` + if (!tryHardlinkSystemCodexSessionFile(sourcePath, replacementPath)) { + return false + } + const targetIdentity = copiedTargetIdentity(marker, targetPath) + if ( + !targetIdentity || + !codexSessionSourceMatchesCopiedPrefix(sourcePath, marker, targetIdentity.fingerprintSha256) + ) { + return false + } + return installPreparedCodexSessionBridge({ + sourcePath, + targetPath, + relativePath, + replacementPath, + usesHardlink: true, + targetIdentityCheck: targetIdentity.check, + preserveOriginal: true + }) + } catch (error) { + console.warn( + '[codex-session-bridge] Failed to migrate copied Codex session:', + sourcePath, + error + ) + return false + } finally { + if (replacementPath) { + rmSync(replacementPath, { force: true }) + } + } +} + +/** Refreshes only an unchanged managed copy whose append-only source grew. */ +export function refreshCopiedCodexSessionBridge( + sourcePath: string, + targetPath: string, + relativePath: string +): boolean { + if (hasPreservedCodexSession(relativePath)) { + warnPreservedCodexSessionRequiresReview(relativePath) + return false + } + const marker = readCopiedCodexSessionMarker(relativePath) + if (!marker || marker.sourcePath !== sourcePath) { + return false + } + try { + const targetStat = lstatSync(targetPath) + const sourceStat = lstatSync(sourcePath) + if ( + targetStat.isSymbolicLink() || + targetStat.isDirectory() || + !codexSessionStatsMatchMarker(targetStat, marker, 'target') || + sourceStat.size <= marker.sourceSize + ) { + return false + } + const targetIdentity = copiedTargetIdentity(marker, targetPath) + if ( + !targetIdentity || + !codexSessionSourceMatchesCopiedPrefix(sourcePath, marker, targetIdentity.fingerprintSha256) + ) { + return false + } + return replaceSystemCodexSessionBridge( + sourcePath, + targetPath, + relativePath, + targetIdentity.check, + true + ) + } catch (error) { + console.warn( + '[codex-session-bridge] Failed to refresh copied Codex session:', + sourcePath, + error + ) + return false + } +} + +function copiedTargetIdentity( + marker: NonNullable>, + targetPath: string +): { check: TargetIdentityCheck; fingerprintSha256: string } | null { + const fingerprintSha256 = + marker.targetFingerprintSha256 ?? fingerprintCodexSessionFile(targetPath) + const check = (candidatePath: string): boolean => { + const stat = lstatSync(candidatePath) + return ( + !stat.isSymbolicLink() && + !stat.isDirectory() && + codexSessionStatsMatchMarker(stat, marker, 'target') && + fingerprintCodexSessionFile(candidatePath) === fingerprintSha256 + ) + } + return check(targetPath) ? { check, fingerprintSha256 } : null +} + +function installPreparedCodexSessionBridge(args: { + sourcePath: string + targetPath: string + relativePath: string + replacementPath: string + usesHardlink: boolean + targetIdentityCheck: TargetIdentityCheck + preserveOriginal: boolean +}): boolean { + if (args.preserveOriginal) { + return installWithPreservedCodexSession(args) + } + return installReplaceableCodexSessionBridge(args) +} + +function installReplaceableCodexSessionBridge(args: { + sourcePath: string + targetPath: string + relativePath: string + replacementPath: string + usesHardlink: boolean + targetIdentityCheck: TargetIdentityCheck +}): boolean { + const backupPath = `${args.targetPath}.orca-backup-${process.pid}-${Date.now()}` + if (!args.targetIdentityCheck(args.targetPath)) { + return false + } + + // Symlink metadata has no writable payload, but retain it until the prepared + // replacement has installed. Never remove an installed replacement to roll + // back a later marker failure; a writer may already have opened it. + renameSync(args.targetPath, backupPath) + if (!args.targetIdentityCheck(backupPath)) { + restoreSymlinkBackupExclusively(backupPath, args.targetPath) + return false + } + try { + linkSync(args.replacementPath, args.targetPath) + } catch (error) { + if (existsSync(args.targetPath)) { + console.warn( + '[codex-session-bridge] Target appeared during symlink replacement; original retained:', + backupPath, + error + ) + } else { + restoreSymlinkBackupExclusively(backupPath, args.targetPath) + } + return false + } + try { + if (args.usesHardlink) { + clearCopiedCodexSessionMarker(args.relativePath) + } else { + writeCopiedCodexSessionMarker(args.relativePath, args.sourcePath, args.targetPath) + } + } catch (error) { + console.warn( + '[codex-session-bridge] Installed replacement without marker update:', + args.targetPath, + error + ) + } + rmSync(backupPath, { force: true }) + return true +} + +function restoreSymlinkBackupExclusively(backupPath: string, targetPath: string): void { + try { + symlinkSync( + readlinkSync(backupPath), + targetPath, + process.platform === 'win32' ? 'file' : undefined + ) + } catch (error) { + console.warn('[codex-session-bridge] Symlink backup requires manual review:', backupPath, error) + } +} + +function warnPreservedCodexSessionRequiresReview(relativePath: string): void { + console.warn( + '[codex-session-bridge] Automatic refresh stopped; preserved copy requires review:', + preservedCodexSessionPaths(relativePath).dataPath + ) +} diff --git a/src/main/codex/codex-session-bridge.test.ts b/src/main/codex/codex-session-bridge.test.ts index ed3c838ea7c..dffb3064674 100644 --- a/src/main/codex/codex-session-bridge.test.ts +++ b/src/main/codex/codex-session-bridge.test.ts @@ -1,13 +1,16 @@ import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' import { + closeSync, existsSync, lstatSync, mkdtempSync, mkdirSync, + openSync, readFileSync, readlinkSync, rmSync, symlinkSync, + writeSync, writeFileSync } from 'node:fs' import type * as NodeFs from 'node:fs' @@ -23,6 +26,10 @@ const { fsMockState } = vi.hoisted(() => ({ fsMockState: { failLink: false, failSymlink: false, + failReplacementInstall: false, + afterReplacementCopy: null as null | (() => void), + afterFailedSourceLink: null as null | (() => void), + beforePreservedRestoreLink: null as null | (() => void), fakeSymlinks: new Map() } })) @@ -39,8 +46,39 @@ vi.mock('node:fs', async () => { const actual = await vi.importActual('node:fs') return { ...actual, + copyFileSync: (...args: Parameters) => { + const result = actual.copyFileSync(...args) + if (String(args[1]).includes('.orca-link-') && fsMockState.afterReplacementCopy !== null) { + const callback = fsMockState.afterReplacementCopy + fsMockState.afterReplacementCopy = null + callback() + } + return result + }, linkSync: (...args: Parameters) => { - if (fsMockState.failLink) { + const [sourcePath, targetPath] = args + if ( + String(sourcePath).includes('.orca-preserved.displaced-') && + fsMockState.beforePreservedRestoreLink !== null + ) { + const callback = fsMockState.beforePreservedRestoreLink + fsMockState.beforePreservedRestoreLink = null + callback() + } + if ( + fsMockState.failReplacementInstall && + String(sourcePath).includes('.orca-link-') && + !String(targetPath).includes('.orca-preserved') + ) { + fsMockState.failReplacementInstall = false + throw new Error('replacement install disabled for test') + } + if (fsMockState.failLink && !String(sourcePath).includes('codex-runtime-home')) { + if (fsMockState.afterFailedSourceLink !== null) { + const callback = fsMockState.afterFailedSourceLink + fsMockState.afterFailedSourceLink = null + callback() + } throw new Error('hardlink disabled for test') } return actual.linkSync(...args) @@ -120,6 +158,18 @@ function getRuntimeCodexHomePath(): string { return join(userDataDir, 'codex-runtime-home', 'home') } +function getPreservedSessionPath(relativePath: string): string { + return join( + getRuntimeCodexHomePath(), + '.orca-session-preserved', + `${relativePath}.orca-preserved` + ) +} + +function getPreservedSessionRecordPath(relativePath: string): string { + return join(getRuntimeCodexHomePath(), '.orca-session-preserved', `${relativePath}.json`) +} + function normalizeLinkTarget(linkTarget: string): string { return process.platform === 'win32' ? linkTarget.replace(/^\\\\\?\\/, '').toLowerCase() @@ -159,6 +209,10 @@ function writeLegacyCopyMarker(relativePath: string, sourcePath: string, targetP beforeEach(() => { fsMockState.failLink = false fsMockState.failSymlink = false + fsMockState.failReplacementInstall = false + fsMockState.afterReplacementCopy = null + fsMockState.afterFailedSourceLink = null + fsMockState.beforePreservedRestoreLink = null fsMockState.fakeSymlinks.clear() fakeHomeDir = mkdtempSync(join(tmpdir(), 'orca-codex-session-home-')) userDataDir = mkdtempSync(join(tmpdir(), 'orca-codex-session-user-data-')) @@ -260,7 +314,7 @@ describe('syncSystemCodexSessionsIntoManagedHome', () => { ).toBe(false) }) - it('falls back to symlinks when hardlinks are unavailable', () => { + it('falls back to a marked regular-file copy when hardlinks are unavailable', () => { fsMockState.failLink = true const systemSessionPath = join( getSystemCodexHomePath(), @@ -268,7 +322,7 @@ describe('syncSystemCodexSessionsIntoManagedHome', () => { '2026', '05', '26', - 'rollout-symlink-fallback.jsonl' + 'rollout-copy-fallback.jsonl' ) mkdirSync(dirname(systemSessionPath), { recursive: true }) writeFileSync(systemSessionPath, '{"id":"system"}\n', 'utf-8') @@ -281,12 +335,62 @@ describe('syncSystemCodexSessionsIntoManagedHome', () => { '2026', '05', '26', - 'rollout-symlink-fallback.jsonl' + 'rollout-copy-fallback.jsonl' ) - expect(lstatSync(runtimeSessionPath).isSymbolicLink()).toBe(true) - expect(normalizeLinkTarget(readlinkSync(runtimeSessionPath))).toBe( - normalizeLinkTarget(systemSessionPath) + expect(lstatSync(runtimeSessionPath).isSymbolicLink()).toBe(false) + expect(lstatSync(runtimeSessionPath).ino).not.toBe(lstatSync(systemSessionPath).ino) + expect(readFileSync(runtimeSessionPath, 'utf-8')).toBe('{"id":"system"}\n') + const markerPath = join( + getRuntimeCodexHomePath(), + '.orca-session-copies', + '2026', + '05', + '26', + 'rollout-copy-fallback.jsonl.json' ) + expect(existsSync(markerPath)).toBe(true) + expect(JSON.parse(readFileSync(markerPath, 'utf-8'))).toMatchObject({ + version: 2, + mtimePrecision: 'milliseconds', + targetFingerprintSha256: expect.stringMatching(/^[a-f0-9]{64}$/) + }) + }) + + it('never overwrites a target that appears during initial copy fallback', () => { + fsMockState.failLink = true + const relativeSessionPath = join('2026', '05', '26', 'rollout-copy-race-new.jsonl') + const systemSessionPath = join(getSystemCodexHomePath(), 'sessions', relativeSessionPath) + const runtimeSessionPath = join(getRuntimeCodexHomePath(), 'sessions', relativeSessionPath) + mkdirSync(dirname(systemSessionPath), { recursive: true }) + writeFileSync(systemSessionPath, '{"id":"source"}\n', 'utf-8') + fsMockState.afterFailedSourceLink = () => { + mkdirSync(dirname(runtimeSessionPath), { recursive: true }) + writeFileSync(runtimeSessionPath, '{"id":"concurrent"}\n', 'utf-8') + } + + syncSystemCodexSessionsIntoManagedHome() + + expect(readFileSync(runtimeSessionPath, 'utf-8')).toBe('{"id":"concurrent"}\n') + }) + + it('copies an existing symlink bridge when hardlink replacement is unavailable', () => { + const relativeSessionPath = join('sessions', '2026', '05', '26', 'rollout-symlink-copy.jsonl') + const systemSessionPath = join(getSystemCodexHomePath(), relativeSessionPath) + const runtimeSessionPath = join(getRuntimeCodexHomePath(), relativeSessionPath) + mkdirSync(dirname(systemSessionPath), { recursive: true }) + mkdirSync(dirname(runtimeSessionPath), { recursive: true }) + writeFileSync(systemSessionPath, '{"id":"system"}\n', 'utf-8') + symlinkSync( + systemSessionPath, + runtimeSessionPath, + process.platform === 'win32' ? 'file' : undefined + ) + fsMockState.failLink = true + + syncSystemCodexSessionsIntoManagedHome() + + expect(lstatSync(runtimeSessionPath).isSymbolicLink()).toBe(false) + expect(readFileSync(runtimeSessionPath, 'utf-8')).toBe('{"id":"system"}\n') }) it('does not overwrite runtime-owned session files', () => { @@ -322,27 +426,30 @@ describe('syncSystemCodexSessionsIntoManagedHome', () => { expectResourceLinked(runtimeSessionPath, systemSessionPath) }) - it('does not create independent session copies when file links are unavailable', () => { - fsMockState.failLink = true - fsMockState.failSymlink = true + it('bridges cold-compressed .jsonl.zst session files', () => { const systemSessionPath = join( getSystemCodexHomePath(), 'sessions', '2026', '05', '26', - 'rollout-unlinked.jsonl' + 'rollout-cold.jsonl.zst' ) mkdirSync(dirname(systemSessionPath), { recursive: true }) - writeFileSync(systemSessionPath, '{"id":"system"}\n', 'utf-8') + writeFileSync(systemSessionPath, 'fake-zstd-bytes', 'utf-8') syncSystemCodexSessionsIntoManagedHome() - expect( - existsSync( - join(getRuntimeCodexHomePath(), 'sessions', '2026', '05', '26', 'rollout-unlinked.jsonl') - ) - ).toBe(false) + const runtimeSessionPath = join( + getRuntimeCodexHomePath(), + 'sessions', + '2026', + '05', + '26', + 'rollout-cold.jsonl.zst' + ) + expect(readFileSync(runtimeSessionPath, 'utf-8')).toBe('fake-zstd-bytes') + expectResourceLinked(runtimeSessionPath, systemSessionPath) }) it('replaces unchanged legacy copied sessions with links', () => { @@ -359,9 +466,13 @@ describe('syncSystemCodexSessionsIntoManagedHome', () => { expect(readFileSync(runtimeSessionPath, 'utf-8')).toBe('{"id":"legacy"}\n') expectResourceLinked(runtimeSessionPath, systemSessionPath) + expect(readFileSync(getPreservedSessionPath(relativeSessionPath), 'utf-8')).toBe( + '{"id":"legacy"}\n' + ) + expect(existsSync(getPreservedSessionRecordPath(relativeSessionPath))).toBe(true) }) - it('preserves unchanged legacy copied sessions when relinking fails', () => { + it('preserves unchanged copied sessions when hardlink migration fails', () => { const relativeSessionPath = join('2026', '05', '26', 'rollout-legacy-unlinked.jsonl') const systemSessionPath = join(getSystemCodexHomePath(), 'sessions', relativeSessionPath) const runtimeSessionPath = join(getRuntimeCodexHomePath(), 'sessions', relativeSessionPath) @@ -371,7 +482,6 @@ describe('syncSystemCodexSessionsIntoManagedHome', () => { writeFileSync(runtimeSessionPath, '{"id":"legacy"}\n', 'utf-8') writeLegacyCopyMarker(relativeSessionPath, systemSessionPath, runtimeSessionPath) fsMockState.failLink = true - fsMockState.failSymlink = true syncSystemCodexSessionsIntoManagedHome() @@ -379,6 +489,148 @@ describe('syncSystemCodexSessionsIntoManagedHome', () => { expect(readFileSync(runtimeSessionPath, 'utf-8')).toBe('{"id":"legacy"}\n') }) + it('refreshes an unchanged copy when its source appends', () => { + fsMockState.failLink = true + const relativeSessionPath = join('2026', '05', '26', 'rollout-copy-append.jsonl') + const systemSessionPath = join(getSystemCodexHomePath(), 'sessions', relativeSessionPath) + const runtimeSessionPath = join(getRuntimeCodexHomePath(), 'sessions', relativeSessionPath) + mkdirSync(dirname(systemSessionPath), { recursive: true }) + writeFileSync(systemSessionPath, '{"id":"line-1"}\n', 'utf-8') + + syncSystemCodexSessionsIntoManagedHome() + writeFileSync(systemSessionPath, '{"id":"line-1"}\n{"id":"line-2"}\n', 'utf-8') + syncSystemCodexSessionsIntoManagedHome() + + expect(lstatSync(runtimeSessionPath).isSymbolicLink()).toBe(false) + expect(readFileSync(runtimeSessionPath, 'utf-8')).toBe('{"id":"line-1"}\n{"id":"line-2"}\n') + expect(readFileSync(getPreservedSessionPath(relativeSessionPath), 'utf-8')).toBe( + '{"id":"line-1"}\n' + ) + }) + + it('preserves late writes through an already-open target descriptor', () => { + fsMockState.failLink = true + const relativeSessionPath = join('2026', '05', '26', 'rollout-copy-late-write.jsonl') + const systemSessionPath = join(getSystemCodexHomePath(), 'sessions', relativeSessionPath) + const runtimeSessionPath = join(getRuntimeCodexHomePath(), 'sessions', relativeSessionPath) + mkdirSync(dirname(systemSessionPath), { recursive: true }) + writeFileSync(systemSessionPath, '{"id":"line-1"}\n', 'utf-8') + syncSystemCodexSessionsIntoManagedHome() + + const descriptor = openSync(runtimeSessionPath, 'a') + writeFileSync(systemSessionPath, '{"id":"line-1"}\n{"id":"line-2"}\n', 'utf-8') + syncSystemCodexSessionsIntoManagedHome() + writeSync(descriptor, '{"id":"late-target-write"}\n') + closeSync(descriptor) + + expect(readFileSync(runtimeSessionPath, 'utf-8')).toBe('{"id":"line-1"}\n{"id":"line-2"}\n') + expect(readFileSync(getPreservedSessionPath(relativeSessionPath), 'utf-8')).toBe( + '{"id":"line-1"}\n{"id":"late-target-write"}\n' + ) + }) + + it('refuses further automatic refresh after preserving one target inode', () => { + fsMockState.failLink = true + const relativeSessionPath = join('2026', '05', '26', 'rollout-copy-bounded.jsonl') + const systemSessionPath = join(getSystemCodexHomePath(), 'sessions', relativeSessionPath) + const runtimeSessionPath = join(getRuntimeCodexHomePath(), 'sessions', relativeSessionPath) + mkdirSync(dirname(systemSessionPath), { recursive: true }) + writeFileSync(systemSessionPath, 'one\n', 'utf-8') + syncSystemCodexSessionsIntoManagedHome() + writeFileSync(systemSessionPath, 'one\ntwo\n', 'utf-8') + syncSystemCodexSessionsIntoManagedHome() + writeFileSync(systemSessionPath, 'one\ntwo\nthree\n', 'utf-8') + const warn = vi.spyOn(console, 'warn').mockImplementation(() => undefined) + + syncSystemCodexSessionsIntoManagedHome() + + expect(readFileSync(runtimeSessionPath, 'utf-8')).toBe('one\ntwo\n') + expect(readFileSync(getPreservedSessionPath(relativeSessionPath), 'utf-8')).toBe('one\n') + expect(warn).toHaveBeenCalledWith( + expect.stringContaining('preserved copy requires review'), + expect.any(String) + ) + }) + + it('does not overwrite a managed copy that diverged after the bridge marker', () => { + fsMockState.failLink = true + const relativeSessionPath = join('2026', '05', '26', 'rollout-copy-diverged.jsonl') + const systemSessionPath = join(getSystemCodexHomePath(), 'sessions', relativeSessionPath) + const runtimeSessionPath = join(getRuntimeCodexHomePath(), 'sessions', relativeSessionPath) + mkdirSync(dirname(systemSessionPath), { recursive: true }) + writeFileSync(systemSessionPath, '{"id":"source-1"}\n', 'utf-8') + + syncSystemCodexSessionsIntoManagedHome() + // Same byte length as the original source and written in the same second: + // local markers must still use millisecond precision to detect divergence. + writeFileSync(runtimeSessionPath, '{"id":"target-1"}\n', 'utf-8') + writeFileSync(systemSessionPath, '{"id":"source-1"}\n{"id":"source-2"}\n', 'utf-8') + syncSystemCodexSessionsIntoManagedHome() + + expect(readFileSync(runtimeSessionPath, 'utf-8')).toBe('{"id":"target-1"}\n') + }) + + it('revalidates a managed copy after preparing its replacement', () => { + fsMockState.failLink = true + const relativeSessionPath = join('2026', '05', '26', 'rollout-copy-race.jsonl') + const systemSessionPath = join(getSystemCodexHomePath(), 'sessions', relativeSessionPath) + const runtimeSessionPath = join(getRuntimeCodexHomePath(), 'sessions', relativeSessionPath) + mkdirSync(dirname(systemSessionPath), { recursive: true }) + writeFileSync(systemSessionPath, '{"id":"source-1"}\n', 'utf-8') + syncSystemCodexSessionsIntoManagedHome() + + writeFileSync(systemSessionPath, '{"id":"source-1"}\n{"id":"source-2"}\n', 'utf-8') + fsMockState.afterReplacementCopy = () => { + writeFileSync(runtimeSessionPath, '{"id":"target-raced"}\n', 'utf-8') + } + syncSystemCodexSessionsIntoManagedHome() + + expect(readFileSync(runtimeSessionPath, 'utf-8')).toBe('{"id":"target-raced"}\n') + }) + + it('rolls back the original managed copy when replacement install fails', () => { + fsMockState.failLink = true + const relativeSessionPath = join('2026', '05', '26', 'rollout-copy-rollback.jsonl') + const systemSessionPath = join(getSystemCodexHomePath(), 'sessions', relativeSessionPath) + const runtimeSessionPath = join(getRuntimeCodexHomePath(), 'sessions', relativeSessionPath) + mkdirSync(dirname(systemSessionPath), { recursive: true }) + writeFileSync(systemSessionPath, '{"id":"source-1"}\n', 'utf-8') + syncSystemCodexSessionsIntoManagedHome() + + writeFileSync(systemSessionPath, '{"id":"source-1"}\n{"id":"source-2"}\n', 'utf-8') + fsMockState.failReplacementInstall = true + const warn = vi.spyOn(console, 'warn').mockImplementation(() => undefined) + syncSystemCodexSessionsIntoManagedHome() + + expect(readFileSync(runtimeSessionPath, 'utf-8')).toBe('{"id":"source-1"}\n') + expect(readFileSync(getPreservedSessionPath(relativeSessionPath), 'utf-8')).toBe( + '{"id":"source-1"}\n' + ) + expect(existsSync(getPreservedSessionRecordPath(relativeSessionPath))).toBe(true) + expect(warn).toHaveBeenCalled() + }) + + it('never overwrites a target that appears during exclusive preserved restore', () => { + fsMockState.failLink = true + const relativeSessionPath = join('2026', '05', '26', 'rollout-restore-race.jsonl') + const systemSessionPath = join(getSystemCodexHomePath(), 'sessions', relativeSessionPath) + const runtimeSessionPath = join(getRuntimeCodexHomePath(), 'sessions', relativeSessionPath) + mkdirSync(dirname(systemSessionPath), { recursive: true }) + writeFileSync(systemSessionPath, 'source-one\n', 'utf-8') + syncSystemCodexSessionsIntoManagedHome() + writeFileSync(systemSessionPath, 'source-one\nsource-two\n', 'utf-8') + fsMockState.failReplacementInstall = true + fsMockState.beforePreservedRestoreLink = () => { + writeFileSync(runtimeSessionPath, 'concurrent-target\n', 'utf-8') + } + vi.spyOn(console, 'warn').mockImplementation(() => undefined) + + syncSystemCodexSessionsIntoManagedHome() + + expect(readFileSync(runtimeSessionPath, 'utf-8')).toBe('concurrent-target\n') + expect(readFileSync(getPreservedSessionPath(relativeSessionPath), 'utf-8')).toBe('source-one\n') + }) + it('incrementally bridges session files without requiring the synchronous launch path', async () => { const systemSessionRoot = join(getSystemCodexHomePath(), 'sessions', '2026', '06', '18') mkdirSync(systemSessionRoot, { recursive: true }) diff --git a/src/main/codex/codex-session-bridge.ts b/src/main/codex/codex-session-bridge.ts index e546ce9d8f8..2462528ebcc 100644 --- a/src/main/codex/codex-session-bridge.ts +++ b/src/main/codex/codex-session-bridge.ts @@ -1,14 +1,4 @@ -import { - existsSync, - linkSync, - lstatSync, - mkdirSync, - readFileSync, - readlinkSync, - renameSync, - rmSync, - symlinkSync -} from 'node:fs' +import { existsSync, lstatSync, mkdirSync, readlinkSync } from 'node:fs' import { dirname, isAbsolute, join, relative, sep } from 'node:path' import { getOrcaManagedCodexHomePath, getSystemCodexHomePath } from './codex-home-paths' import { @@ -16,17 +6,25 @@ import { listCodexSessionJsonlFilesIncrementally } from './codex-session-file-listing' import type { CodexSessionBridgeIncrementalOptions } from './codex-session-file-listing' +import { + clearCopiedCodexSessionMarker, + codexSessionStatsMatchMarker, + readCopiedCodexSessionMarker +} from './codex-session-copy-markers' +import { + migrateCopiedCodexSessionBridge, + refreshCopiedCodexSessionBridge, + replaceSystemCodexSessionBridge, + tryCopySystemCodexSessionFile, + tryHardlinkSystemCodexSessionFile +} from './codex-session-bridge-link' +import { + hasPreservedCodexSession, + preservedCodexSessionPaths +} from './codex-session-preserved-copies' export type { CodexSessionBridgeIncrementalOptions } from './codex-session-file-listing' -type LegacyCopiedSessionMarker = { - sourcePath: string - sourceSize: number - sourceMtimeMs: number - targetSize: number - targetMtimeMs: number -} - export type LegacyCopiedCodexSessionBridgeScanPreference = { sourcePath: string preferManagedCopy: boolean @@ -130,79 +128,70 @@ function bridgeSystemCodexSessionFile( const relativePath = relative(systemSessionsRoot, systemSessionFilePath) const managedSessionFilePath = join(managedSessionsRoot, relativePath) if (existsSync(managedSessionFilePath)) { - if ( - replaceSymlinkSessionBridgeWithHardlink( - systemSessionFilePath, - managedSessionFilePath, - relativePath + if (hasPreservedCodexSession(relativePath)) { + if (pathsReferenceSameFile(systemSessionFilePath, managedSessionFilePath)) { + return false + } + console.warn( + '[codex-session-bridge] Automatic refresh stopped; preserved copy requires review:', + preservedCodexSessionPaths(relativePath).dataPath ) + return false + } + if (replaceSymlinkSessionBridge(systemSessionFilePath, managedSessionFilePath, relativePath)) { + return true + } + if ( + migrateCopiedCodexSessionBridge(systemSessionFilePath, managedSessionFilePath, relativePath) ) { return true } - migrateLegacyCopiedSessionBridge(systemSessionFilePath, managedSessionFilePath, relativePath) - return false + return refreshCopiedCodexSessionBridge( + systemSessionFilePath, + managedSessionFilePath, + relativePath + ) } mkdirSync(dirname(managedSessionFilePath), { recursive: true }) return linkSystemCodexSessionFile(systemSessionFilePath, managedSessionFilePath, relativePath) } +function pathsReferenceSameFile(leftPath: string, rightPath: string): boolean { + try { + const leftStat = lstatSync(leftPath) + const rightStat = lstatSync(rightPath) + return leftStat.ino !== 0 && leftStat.dev === rightStat.dev && leftStat.ino === rightStat.ino + } catch { + return false + } +} + /** - * Links a source session file and clears any stale copied-session marker. + * Hardlinks a source session or copy-syncs it when volumes differ. */ function linkSystemCodexSessionFile( sourcePath: string, targetPath: string, relativePath: string ): boolean { - const linked = tryLinkSystemCodexSessionFile(sourcePath, targetPath) - if (linked) { - clearLegacyCopiedSessionMarker(relativePath) - } - return linked -} - -/** - * Attempts to link a session file with hardlink first and symlink fallback. - */ -function tryLinkSystemCodexSessionFile(sourcePath: string, targetPath: string): boolean { if (tryHardlinkSystemCodexSessionFile(sourcePath, targetPath)) { + clearCopiedCodexSessionMarker(relativePath) return true } - try { - // Why fallback: hardlinks keep sessions visible to Codex resume, but can - // fail across volumes. A symlink is still better than a diverging copy. - symlinkSync(sourcePath, targetPath, process.platform === 'win32' ? 'file' : undefined) - return true - } catch (error) { - console.warn('[codex-session-bridge] Failed to link system Codex session:', sourcePath, error) - } - return false -} - -/** - * Attempts a hardlink so resume sees one physical JSONL session log. - */ -function tryHardlinkSystemCodexSessionFile(sourcePath: string, targetPath: string): boolean { - try { - // Why: Codex resume ignores symlinked JSONL sessions, while a hardlink - // preserves one physical log without copy divergence. - linkSync(sourcePath, targetPath) - return true - } catch { - return false - } + // Why: Codex resume ignores symlinks; a marked regular-file copy is the safe + // cross-volume fallback. + return tryCopySystemCodexSessionFile(sourcePath, targetPath, relativePath) } /** * Replaces an older symlink bridge with a hardlink when the target still points * at the expected source session. */ -function replaceSymlinkSessionBridgeWithHardlink( +function replaceSymlinkSessionBridge( sourcePath: string, targetPath: string, relativePath: string ): boolean { - let replacementPath: string | null = null try { const targetStat = lstatSync(targetPath) if (!targetStat.isSymbolicLink()) { @@ -216,74 +205,38 @@ function replaceSymlinkSessionBridgeWithHardlink( return false } - replacementPath = `${targetPath}.orca-link-${process.pid}-${Date.now()}` - if (!tryHardlinkSystemCodexSessionFile(sourcePath, replacementPath)) { - return false - } - rmSync(targetPath, { force: true }) - renameSync(replacementPath, targetPath) - clearLegacyCopiedSessionMarker(relativePath) - return true - } catch (error) { - console.warn( - '[codex-session-bridge] Failed to replace symlinked Codex session bridge:', + return replaceSystemCodexSessionBridge( sourcePath, - error + targetPath, + relativePath, + (candidatePath) => { + const candidateStat = lstatSync(candidatePath) + if (!candidateStat.isSymbolicLink()) { + return false + } + const candidateLinkTarget = readlinkSync(candidatePath) + return ( + (isAbsolute(candidateLinkTarget) + ? candidateLinkTarget + : join(dirname(candidatePath), candidateLinkTarget)) === sourcePath + ) + } ) - if (replacementPath) { - rmSync(replacementPath, { force: true }) - } - } - return false -} - -/** - * Migrates a legacy copied bridge to a linked bridge when the copied file still - * matches its marker. - */ -function migrateLegacyCopiedSessionBridge( - sourcePath: string, - targetPath: string, - relativePath: string -): void { - const marker = readLegacyCopiedSessionMarker(relativePath) - if (!marker || marker.sourcePath !== sourcePath) { - return - } - let replacementPath: string | null = null - try { - const targetStat = lstatSync(targetPath) - if (targetStat.isSymbolicLink()) { - clearLegacyCopiedSessionMarker(relativePath) - return - } - if (!fileStatsMatchMarker(targetStat, marker, 'target')) { - return - } - replacementPath = `${targetPath}.orca-link-${process.pid}-${Date.now()}` - if (!tryLinkSystemCodexSessionFile(sourcePath, replacementPath)) { - return - } - rmSync(targetPath, { force: true }) - renameSync(replacementPath, targetPath) - clearLegacyCopiedSessionMarker(relativePath) } catch (error) { console.warn( - '[codex-session-bridge] Failed to migrate copied system Codex session:', + '[codex-session-bridge] Failed to replace symlinked Codex session bridge:', sourcePath, error ) - if (replacementPath) { - rmSync(replacementPath, { force: true }) - } } + return false } /** * Resolves how scanners should treat a legacy copied session bridge. * * The result keeps resume scans coherent until the copied bridge is migrated to - * a hardlink or symlink. + * a hardlink. */ export function getLegacyCopiedCodexSessionBridgeScanPreference( sessionFilePath: string @@ -298,7 +251,7 @@ export function getLegacyCopiedCodexSessionBridgeScanPreference( ) { return null } - const marker = readLegacyCopiedSessionMarker(relativePath) + const marker = readCopiedCodexSessionMarker(relativePath) if (!marker) { return null } @@ -306,10 +259,14 @@ export function getLegacyCopiedCodexSessionBridgeScanPreference( let targetMatchesMarker = false let sourceMatchesMarker = false try { - targetMatchesMarker = fileStatsMatchMarker(lstatSync(sessionFilePath), marker, 'target') + targetMatchesMarker = codexSessionStatsMatchMarker(lstatSync(sessionFilePath), marker, 'target') } catch {} try { - sourceMatchesMarker = fileStatsMatchMarker(lstatSync(marker.sourcePath), marker, 'source') + sourceMatchesMarker = codexSessionStatsMatchMarker( + lstatSync(marker.sourcePath), + marker, + 'source' + ) } catch {} return { @@ -320,57 +277,3 @@ export function getLegacyCopiedCodexSessionBridgeScanPreference( sourceSkipBytes: !targetMatchesMarker && !sourceMatchesMarker ? marker.sourceSize : null } } - -/** - * Returns the marker path for a legacy copied session bridge. - */ -function getLegacySessionCopyMarkerPath(relativePath: string): string { - return join(getOrcaManagedCodexHomePath(), '.orca-session-copies', `${relativePath}.json`) -} - -/** - * Reads and validates the marker for a legacy copied session bridge. - */ -function readLegacyCopiedSessionMarker(relativePath: string): LegacyCopiedSessionMarker | null { - try { - const parsed: unknown = JSON.parse( - readFileSync(getLegacySessionCopyMarkerPath(relativePath), 'utf-8') - ) - if (!parsed || typeof parsed !== 'object' || Array.isArray(parsed)) { - return null - } - const marker = parsed as Record - if ( - typeof marker.sourcePath !== 'string' || - typeof marker.sourceSize !== 'number' || - typeof marker.sourceMtimeMs !== 'number' || - typeof marker.targetSize !== 'number' || - typeof marker.targetMtimeMs !== 'number' - ) { - return null - } - return marker as LegacyCopiedSessionMarker - } catch { - return null - } -} - -/** - * Checks whether source or target file stats still match a legacy bridge marker. - */ -function fileStatsMatchMarker( - stat: { size: number; mtimeMs: number }, - marker: LegacyCopiedSessionMarker, - kind: 'source' | 'target' -): boolean { - const expectedSize = kind === 'source' ? marker.sourceSize : marker.targetSize - const expectedMtimeMs = kind === 'source' ? marker.sourceMtimeMs : marker.targetMtimeMs - return stat.size === expectedSize && stat.mtimeMs === expectedMtimeMs -} - -/** - * Removes the marker after a copied session bridge has been migrated or retired. - */ -function clearLegacyCopiedSessionMarker(relativePath: string): void { - rmSync(getLegacySessionCopyMarkerPath(relativePath), { force: true }) -} diff --git a/src/main/codex/codex-session-copy-markers.ts b/src/main/codex/codex-session-copy-markers.ts new file mode 100644 index 00000000000..c4c6cc25798 --- /dev/null +++ b/src/main/codex/codex-session-copy-markers.ts @@ -0,0 +1,161 @@ +import { createHash } from 'node:crypto' +import { + closeSync, + lstatSync, + mkdirSync, + openSync, + readFileSync, + readSync, + rmSync, + writeFileSync +} from 'node:fs' +import { dirname, join } from 'node:path' +import { getOrcaManagedCodexHomePath } from './codex-home-paths' + +export type CopiedCodexSessionMarker = { + version?: 2 + mtimePrecision?: 'milliseconds' | 'seconds' + sourcePath: string + sourceSize: number + sourceMtimeMs: number + targetSize: number + targetMtimeMs: number + targetFingerprintSha256?: string +} + +export function copiedCodexSessionMarkerPath(relativePath: string): string { + return join(getOrcaManagedCodexHomePath(), '.orca-session-copies', `${relativePath}.json`) +} + +export function readCopiedCodexSessionMarker( + relativePath: string +): CopiedCodexSessionMarker | null { + try { + const parsed: unknown = JSON.parse( + readFileSync(copiedCodexSessionMarkerPath(relativePath), 'utf-8') + ) + if (!parsed || typeof parsed !== 'object' || Array.isArray(parsed)) { + return null + } + const marker = parsed as Record + if ( + typeof marker.sourcePath !== 'string' || + typeof marker.sourceSize !== 'number' || + typeof marker.sourceMtimeMs !== 'number' || + typeof marker.targetSize !== 'number' || + typeof marker.targetMtimeMs !== 'number' + ) { + return null + } + if ( + (marker.version !== undefined && marker.version !== 2) || + (marker.mtimePrecision !== undefined && + marker.mtimePrecision !== 'milliseconds' && + marker.mtimePrecision !== 'seconds') || + (marker.targetFingerprintSha256 !== undefined && + (typeof marker.targetFingerprintSha256 !== 'string' || + !/^[a-f0-9]{64}$/.test(marker.targetFingerprintSha256))) + ) { + return null + } + if ( + marker.version === 2 && + (marker.mtimePrecision === undefined || marker.targetFingerprintSha256 === undefined) + ) { + return null + } + return marker as CopiedCodexSessionMarker + } catch { + return null + } +} + +export function codexSessionStatsMatchMarker( + stat: { size: number; mtimeMs: number }, + marker: CopiedCodexSessionMarker, + kind: 'source' | 'target' +): boolean { + const expectedSize = kind === 'source' ? marker.sourceSize : marker.targetSize + const expectedMtimeMs = kind === 'source' ? marker.sourceMtimeMs : marker.targetMtimeMs + // Why: marker origin is explicit; inferring WSL precision from an exact-second + // local mtime can misclassify a same-second managed edit as unchanged. + const mtimeMatches = + marker.mtimePrecision === 'seconds' + ? Math.floor(stat.mtimeMs / 1000) === expectedMtimeMs / 1000 + : stat.mtimeMs === expectedMtimeMs + return stat.size === expectedSize && mtimeMatches +} + +export function codexSessionSourceMatchesCopiedPrefix( + sourcePath: string, + marker: CopiedCodexSessionMarker, + expectedFingerprintSha256 = marker.targetFingerprintSha256 +): boolean { + return ( + expectedFingerprintSha256 !== undefined && + lstatSync(sourcePath).size >= marker.targetSize && + fingerprintCodexSessionFile(sourcePath, marker.targetSize) === expectedFingerprintSha256 + ) +} + +export function fingerprintCodexSessionFile(filePath: string, maxBytes?: number): string { + const hash = createHash('sha256') + const buffer = Buffer.allocUnsafe(64 * 1024) + const descriptor = openSync(filePath, 'r') + let remaining = maxBytes ?? Number.POSITIVE_INFINITY + try { + while (remaining > 0) { + const bytesRead = readSync(descriptor, buffer, 0, Math.min(buffer.length, remaining), null) + if (bytesRead === 0) { + break + } + hash.update(buffer.subarray(0, bytesRead)) + remaining -= bytesRead + } + } finally { + closeSync(descriptor) + } + return hash.digest('hex') +} + +export function clearCopiedCodexSessionMarker(relativePath: string): void { + rmSync(copiedCodexSessionMarkerPath(relativePath), { force: true }) +} + +export function writeCopiedCodexSessionMarker( + relativePath: string, + sourcePath: string, + targetPath: string +): void { + const sourceStat = lstatSync(sourcePath) + const targetStat = lstatSync(targetPath) + const targetFingerprintSha256 = fingerprintCodexSessionFile(targetPath) + if ( + targetStat.size > sourceStat.size || + fingerprintCodexSessionFile(sourcePath, targetStat.size) !== targetFingerprintSha256 + ) { + throw new Error('Copied Codex session no longer matches its source prefix') + } + const markerPath = copiedCodexSessionMarkerPath(relativePath) + mkdirSync(dirname(markerPath), { recursive: true }) + writeFileSync( + markerPath, + `${JSON.stringify( + { + version: 2, + mtimePrecision: 'milliseconds', + sourcePath, + // Source may append while it is copied; the target size is the exact + // source prefix that this marker owns. + sourceSize: targetStat.size, + sourceMtimeMs: sourceStat.mtimeMs, + targetSize: targetStat.size, + targetMtimeMs: targetStat.mtimeMs, + targetFingerprintSha256 + } satisfies CopiedCodexSessionMarker, + null, + 2 + )}\n`, + 'utf-8' + ) +} diff --git a/src/main/codex/codex-session-file-listing.ts b/src/main/codex/codex-session-file-listing.ts index c6eeb9b1f77..c7073305a61 100644 --- a/src/main/codex/codex-session-file-listing.ts +++ b/src/main/codex/codex-session-file-listing.ts @@ -1,6 +1,7 @@ import { readdirSync } from 'node:fs' import { opendir } from 'node:fs/promises' import { join } from 'node:path' +import { isCodexSessionRolloutFileName } from '../ai-vault/session-scanner-codex-paths' export type CodexSessionBridgeIncrementalOptions = { /** Directory entries to process before yielding back to the event loop. */ @@ -27,7 +28,7 @@ export function listCodexSessionJsonlFiles(rootPath: string): string[] { appendSessionFilePaths(files, listCodexSessionJsonlFiles(childPath)) continue } - if (entry.isFile() && entry.name.endsWith('.jsonl')) { + if (entry.isFile() && isCodexSessionRolloutFileName(entry.name)) { files.push(childPath) } } @@ -74,7 +75,7 @@ export async function* listCodexSessionJsonlFilesIncrementally( const childPath = join(currentDirectory, entry.name) if (entry.isDirectory()) { pendingDirectories.push(childPath) - } else if (entry.isFile() && entry.name.endsWith('.jsonl')) { + } else if (entry.isFile() && isCodexSessionRolloutFileName(entry.name)) { yield childPath } entriesSinceYield += 1 diff --git a/src/main/codex/codex-session-preserved-copies.ts b/src/main/codex/codex-session-preserved-copies.ts new file mode 100644 index 00000000000..5cb7f87c30f --- /dev/null +++ b/src/main/codex/codex-session-preserved-copies.ts @@ -0,0 +1,50 @@ +import { existsSync, mkdirSync, writeFileSync } from 'node:fs' +import { dirname, join } from 'node:path' +import { getOrcaManagedCodexHomePath } from './codex-home-paths' +import { fingerprintCodexSessionFile } from './codex-session-copy-markers' + +export type PreservedCodexSessionPaths = { + dataPath: string + recordPath: string +} + +export function preservedCodexSessionPaths(relativePath: string): PreservedCodexSessionPaths { + const rootPath = join(getOrcaManagedCodexHomePath(), '.orca-session-preserved') + return { + // The suffix intentionally keeps this out of Codex rollout discovery. + dataPath: join(rootPath, `${relativePath}.orca-preserved`), + recordPath: join(rootPath, `${relativePath}.json`) + } +} + +export function hasPreservedCodexSession(relativePath: string): boolean { + const paths = preservedCodexSessionPaths(relativePath) + return existsSync(paths.dataPath) || existsSync(paths.recordPath) +} + +export function writePreservedCodexSessionRecord(args: { + relativePath: string + sourcePath: string + originalTargetPath: string + displacedTargetPath: string +}): void { + const paths = preservedCodexSessionPaths(args.relativePath) + mkdirSync(dirname(paths.recordPath), { recursive: true }) + writeFileSync( + paths.recordPath, + `${JSON.stringify( + { + version: 1, + sourcePath: args.sourcePath, + originalTargetPath: args.originalTargetPath, + preservedPath: paths.dataPath, + displacedTargetPath: args.displacedTargetPath, + preservedFingerprintSha256: fingerprintCodexSessionFile(paths.dataPath), + createdAt: new Date().toISOString() + }, + null, + 2 + )}\n`, + { encoding: 'utf-8', flag: 'wx' } + ) +} diff --git a/src/main/codex/codex-session-preserved-install.ts b/src/main/codex/codex-session-preserved-install.ts new file mode 100644 index 00000000000..233e201bccb --- /dev/null +++ b/src/main/codex/codex-session-preserved-install.ts @@ -0,0 +1,131 @@ +import { existsSync, linkSync, mkdirSync, renameSync } from 'node:fs' +import { dirname } from 'node:path' +import { + clearCopiedCodexSessionMarker, + writeCopiedCodexSessionMarker +} from './codex-session-copy-markers' +import { + hasPreservedCodexSession, + preservedCodexSessionPaths, + writePreservedCodexSessionRecord +} from './codex-session-preserved-copies' + +export type PreservedCodexSessionInstallArgs = { + sourcePath: string + targetPath: string + relativePath: string + replacementPath: string + usesHardlink: boolean + targetIdentityCheck: (candidatePath: string) => boolean +} + +/** Installs one bounded refresh while permanently retaining every displaced inode. */ +export function installWithPreservedCodexSession(args: PreservedCodexSessionInstallArgs): boolean { + const preservedPaths = preservedCodexSessionPaths(args.relativePath) + const displacedTargetPath = `${preservedPaths.dataPath}.displaced-${process.pid}-${Date.now()}` + if ( + hasPreservedCodexSession(args.relativePath) || + existsSync(displacedTargetPath) || + !args.targetIdentityCheck(args.targetPath) + ) { + return false + } + mkdirSync(dirname(preservedPaths.dataPath), { recursive: true }) + + // Hardlinking first keeps the old inode reachable even if an already-open + // writer appends after the canonical target name moves to the replacement. + linkSync(args.targetPath, preservedPaths.dataPath) + try { + writePreservedCodexSessionRecord({ + relativePath: args.relativePath, + sourcePath: args.sourcePath, + originalTargetPath: args.targetPath, + displacedTargetPath + }) + } catch (error) { + // The deterministic preserved path is itself the recovery pointer. Once + // created it is never removed automatically, even if its record failed. + console.warn( + '[codex-session-bridge] Preserved Codex session without sidecar record:', + preservedPaths.dataPath, + error + ) + return false + } + try { + if (!args.targetIdentityCheck(preservedPaths.dataPath)) { + return false + } + + // Rename, rather than unlink, whichever inode is at the canonical name at + // commit time. A racing replacement is therefore preserved too. + renameSync(args.targetPath, displacedTargetPath) + if (!installReplacementExclusively(args, displacedTargetPath)) { + return false + } + + updateInstalledBridgeMarker(args) + return true + } catch (error) { + console.warn( + '[codex-session-bridge] Preserved Codex session requires manual review:', + preservedPaths.dataPath, + error + ) + return false + } +} + +function installReplacementExclusively( + args: PreservedCodexSessionInstallArgs, + displacedTargetPath: string +): boolean { + try { + // replacementPath is in the target directory, so this is an exclusive, + // complete-file publication even when source and target volumes differ. + linkSync(args.replacementPath, args.targetPath) + return true + } catch (error) { + if (existsSync(args.targetPath)) { + console.warn( + '[codex-session-bridge] Target appeared during preserved install:', + args.targetPath, + error + ) + return false + } + try { + // Exclusive restore cannot overwrite a target that appears after the + // existence check; displaced and snapshot names remain recoverable. + linkSync(displacedTargetPath, args.targetPath) + } catch (restoreError) { + console.warn( + '[codex-session-bridge] Preserved replacement restore requires review:', + displacedTargetPath, + restoreError + ) + } + console.warn( + '[codex-session-bridge] Preserved replacement install failed:', + args.targetPath, + error + ) + return false + } +} + +function updateInstalledBridgeMarker(args: PreservedCodexSessionInstallArgs): void { + try { + if (args.usesHardlink) { + clearCopiedCodexSessionMarker(args.relativePath) + } else { + writeCopiedCodexSessionMarker(args.relativePath, args.sourcePath, args.targetPath) + } + } catch (error) { + console.warn( + '[codex-session-bridge] Installed preserved bridge without marker update:', + args.targetPath, + error + ) + } +} diff --git a/src/main/codex/wsl-codex-session-bridge.test.ts b/src/main/codex/wsl-codex-session-bridge.test.ts index c4218913a5b..3586e6defd9 100644 --- a/src/main/codex/wsl-codex-session-bridge.test.ts +++ b/src/main/codex/wsl-codex-session-bridge.test.ts @@ -65,11 +65,16 @@ describe('syncWslCodexSessionsIntoManagedHome', () => { expect(shellCommand).toContain( "managed_sessions_root='/home/alice/.local/share/orca/codex-runtime-home/home/sessions'" ) - expect(shellCommand).toContain(`find "\\$source_sessions_root" -type f -name '*.jsonl' -print0`) + expect(shellCommand).toContain( + `find "\\$source_sessions_root" -type f \\( -name '*.jsonl' -o -name '*.jsonl.zst' \\) -print0` + ) expect(shellCommand).toContain('ln -- "\\$source_file" "\\$target_file"') - expect(shellCommand).toContain('if [ -e "\\$target_file" ] || [ -L "\\$target_file" ]; then') + expect(shellCommand).toContain('cp -p -- "\\$source_file" "\\$replacement_file"') + expect(shellCommand).toContain('ln -- "\\$replacement_file" "\\$target_file"') + expect(shellCommand).toContain('.orca-session-copies') + expect(shellCommand).toContain('if [ -L "\\$target_file" ]; then') + expect(shellCommand).toContain('if [ -e "\\$target_file" ]; then') expect(shellCommand).not.toContain('ln -s') - expect(shellCommand).not.toContain('cp ') expect(shellCommand).not.toContain('sqlite') }) @@ -152,9 +157,46 @@ describe('buildWslCodexSessionBridgeShellCommand', () => { `source_sessions_root='/home/alice/.codex/sessions with '\\''quote'\\'''` ) expect(shellCommand).toContain(`-name '*.jsonl'`) + expect(shellCommand).toContain(`-name '*.jsonl.zst'`) + expect(shellCommand).toContain('cp -p --') + expect(shellCommand).toContain('.orca-session-copies') expect(shellCommand).not.toContain('.sqlite') }) + it('refreshes only marker-owned copies that still match the source prefix', () => { + const shellCommand = buildWslCodexSessionBridgeShellCommand({ + systemSessionsRoot: '/home/alice/.codex/sessions', + managedSessionsRoot: '/home/alice/.local/share/orca/codex-runtime-home/home/sessions' + }) + + expect(shellCommand).toContain( + 'copy_marker_matches "\\$marker_file" "\\$source_file" "\\$target_file"' + ) + expect(shellCommand).toContain( + '[ "\\$marker_content" = "\\${marker_prefix}\\${stored_source_mtime}\\${marker_suffix}" ]' + ) + expect(shellCommand).toContain('"sourcePath":"%s","sourceSize":%s') + expect(shellCommand).toContain('[ "\\$target_size" -lt "\\$source_size" ]') + expect(shellCommand).toContain( + 'cmp --silent --bytes="\\$target_size" -- "\\$target_file" "\\$source_file"' + ) + expect(shellCommand).toContain('cp -p -- "\\$source_file" "\\$replacement_file"') + expect(shellCommand).toContain('target_fingerprint_before=\\$(file_sha256 "\\$target_file")') + expect(shellCommand).toContain('target_fingerprint_after=\\$(file_sha256 "\\$target_file")') + expect(shellCommand).toContain('ln -- "\\$target_file" "\\$preserved_file"') + expect(shellCommand).toContain('mv -- "\\$target_file" "\\$displaced_file"') + expect(shellCommand).toContain('ln -- "\\$replacement_file" "\\$target_file"') + expect(shellCommand).toContain('ln -- "\\$displaced_file" "\\$target_file"') + expect(shellCommand).toContain('.orca-session-preserved') + expect(shellCommand).toContain('preserved copy requires review') + expect(shellCommand).toContain('value=\\${value//\\\\/\\\\\\\\}') + expect(shellCommand).toContain('source_path_json=\\$(json_escape "\\$source_file")') + expect(shellCommand).toContain('"version":2') + expect(shellCommand).toContain('"mtimePrecision":"seconds"') + expect(shellCommand).toContain('"targetFingerprintSha256":"%s"') + expect(shellCommand).toContain('"displacedTargetPath":"%s"') + }) + it('escapes Linux-side shell variable expansion for wsl.exe argv', () => { const shellCommand = buildWslCodexSessionBridgeShellCommand({ systemSessionsRoot: '/home/alice/.codex/sessions', diff --git a/src/main/codex/wsl-codex-session-bridge.ts b/src/main/codex/wsl-codex-session-bridge.ts index e83abfadd95..8371b6511c4 100644 --- a/src/main/codex/wsl-codex-session-bridge.ts +++ b/src/main/codex/wsl-codex-session-bridge.ts @@ -89,6 +89,125 @@ export function buildWslCodexSessionBridgeShellCommand( `managed_sessions_root=${quoteBashString(paths.managedSessionsRoot)}`, 'scanned_files=0', 'linked_files=0', + 'json_escape() {', + ' local value="$1"', + ' value=${value//\\\\/\\\\\\\\}', + ' value=${value//\\"/\\\\\\"}', + " value=${value//$'\\n'/\\\\n}", + " value=${value//$'\\r'/\\\\r}", + " value=${value//$'\\t'/\\\\t}", + ' printf \'%s\' "$value"', + '}', + 'file_sha256() {', + ' local output hash', + ' output=$(sha256sum -- "$1" 2>/dev/null) || return 1', + ' hash=${output%% *}', + ' [ "${#hash}" -eq 64 ] || return 1', + ' printf \'%s\' "$hash"', + '}', + // Why: sourceSize records the verified copied prefix, not the source's + // current full size, because an active rollout may still be growing. + 'write_copy_marker() {', + ' local marker_file="$1" source_file="$2" target_file="$3"', + ' mkdir -p -- "$(dirname -- "$marker_file")" 2>/dev/null || return 1', + ' local source_size target_size source_mtime_ms target_mtime_ms source_path_json target_fingerprint', + ' target_size=$(stat -c %s -- "$target_file" 2>/dev/null) || return 1', + ' source_size=$(stat -c %s -- "$source_file" 2>/dev/null) || return 1', + ' [ "$target_size" -le "$source_size" ] || return 1', + ' cmp --silent --bytes="$target_size" -- "$target_file" "$source_file" || return 1', + ' source_size=$target_size', + ' source_mtime_ms=$(($(stat -c %Y -- "$source_file" 2>/dev/null) * 1000)) || return 1', + ' target_mtime_ms=$(($(stat -c %Y -- "$target_file" 2>/dev/null) * 1000)) || return 1', + ' source_path_json=$(json_escape "$source_file") || return 1', + ' target_fingerprint=$(file_sha256 "$target_file") || return 1', + ` printf '{"version":2,"mtimePrecision":"seconds","sourcePath":"%s","sourceSize":%s,"sourceMtimeMs":%s,"targetSize":%s,"targetMtimeMs":%s,"targetFingerprintSha256":"%s"}\\n' "$source_path_json" "$source_size" "$source_mtime_ms" "$target_size" "$target_mtime_ms" "$target_fingerprint" >"$marker_file" 2>/dev/null`, + '}', + 'copy_marker_matches() {', + ' local marker_file="$1" source_file="$2" target_file="$3"', + ' local marker_content source_path_json target_size target_mtime_ms target_fingerprint marker_prefix marker_suffix stored_source_mtime', + ' [ -f "$marker_file" ] || return 1', + ' marker_content=$(<"$marker_file") || return 1', + ' source_path_json=$(json_escape "$source_file") || return 1', + ' target_size=$(stat -c %s -- "$target_file" 2>/dev/null) || return 1', + ' target_mtime_ms=$(($(stat -c %Y -- "$target_file" 2>/dev/null) * 1000)) || return 1', + ' target_fingerprint=$(file_sha256 "$target_file") || return 1', + ` marker_prefix=$(printf '{"version":2,"mtimePrecision":"seconds","sourcePath":"%s","sourceSize":%s,"sourceMtimeMs":' "$source_path_json" "$target_size")`, + ` marker_suffix=$(printf ',"targetSize":%s,"targetMtimeMs":%s,"targetFingerprintSha256":"%s"}' "$target_size" "$target_mtime_ms" "$target_fingerprint")`, + ' stored_source_mtime=${marker_content#"$marker_prefix"}', + ' stored_source_mtime=${stored_source_mtime%"$marker_suffix"}', + ' [ "$marker_content" = "${marker_prefix}${stored_source_mtime}${marker_suffix}" ] || return 1', + ' case "$stored_source_mtime" in ""|*[!0-9]*) return 1 ;; esac', + '}', + 'write_preserved_record() {', + ' local record_file="$1" source_file="$2" target_file="$3" preserved_file="$4" displaced_file="$5"', + ' local source_json target_json preserved_json displaced_json preserved_fingerprint', + ' source_json=$(json_escape "$source_file") || return 1', + ' target_json=$(json_escape "$target_file") || return 1', + ' preserved_json=$(json_escape "$preserved_file") || return 1', + ' displaced_json=$(json_escape "$displaced_file") || return 1', + ' preserved_fingerprint=$(file_sha256 "$preserved_file") || return 1', + ` ( set -o noclobber; printf '{"version":1,"sourcePath":"%s","originalTargetPath":"%s","preservedPath":"%s","displacedTargetPath":"%s","preservedFingerprintSha256":"%s"}\\n' "$source_json" "$target_json" "$preserved_json" "$displaced_json" "$preserved_fingerprint" >"$record_file" ) 2>/dev/null`, + '}', + // Why: preserve the pre-refresh inode so open file descriptors stay observable; + // once that evidence exists, halt automatic refresh until human review. + 'refresh_copy() {', + ' local marker_file="$1" source_file="$2" target_file="$3" relative_path="$4"', + ' local target_size source_size target_size_after target_fingerprint_before target_fingerprint_after', + ' local replacement_file preserved_file preserved_record displaced_file preserved_size preserved_fingerprint', + ' preserved_file="$managed_sessions_root/../.orca-session-preserved/${relative_path}.orca-preserved"', + ' preserved_record="$managed_sessions_root/../.orca-session-preserved/${relative_path}.json"', + ' displaced_file="${preserved_file}.displaced-$$"', + ' if [ -e "$preserved_file" ] || [ -e "$preserved_record" ] || [ -e "$displaced_file" ]; then', + ' printf \'%s\\n\' "[codex-session-bridge] Automatic refresh stopped; preserved copy requires review: $preserved_file" >&2', + ' return 1', + ' fi', + ' copy_marker_matches "$marker_file" "$source_file" "$target_file" || return 1', + ' target_size=$(stat -c %s -- "$target_file" 2>/dev/null) || return 1', + ' source_size=$(stat -c %s -- "$source_file" 2>/dev/null) || return 1', + ' [ "$target_size" -lt "$source_size" ] || return 1', + ' target_fingerprint_before=$(file_sha256 "$target_file") || return 1', + ' cmp --silent --bytes="$target_size" -- "$target_file" "$source_file" || return 1', + ' replacement_file="${target_file}.orca-copy-$$"', + ' if ! cp -p -- "$source_file" "$replacement_file"; then', + ' rm -f -- "$replacement_file"', + ' return 1', + ' fi', + ' target_size_after=$(stat -c %s -- "$target_file" 2>/dev/null) || { rm -f -- "$replacement_file"; return 1; }', + ' target_fingerprint_after=$(file_sha256 "$target_file") || { rm -f -- "$replacement_file"; return 1; }', + ' if [ "$target_size_after" != "$target_size" ] || [ "$target_fingerprint_after" != "$target_fingerprint_before" ] || ! cmp --silent --bytes="$target_size" -- "$target_file" "$source_file"; then', + ' rm -f -- "$replacement_file"', + ' return 1', + ' fi', + ' mkdir -p -- "$(dirname -- "$preserved_file")" || { rm -f -- "$replacement_file"; return 1; }', + ' if ! ln -- "$target_file" "$preserved_file"; then', + ' rm -f -- "$replacement_file"', + ' return 1', + ' fi', + ' if ! write_preserved_record "$preserved_record" "$source_file" "$target_file" "$preserved_file" "$displaced_file"; then', + ' rm -f -- "$replacement_file"', + ' printf \'%s\\n\' "[codex-session-bridge] Preserved copy has no sidecar record: $preserved_file" >&2', + ' return 1', + ' fi', + ' preserved_size=$(stat -c %s -- "$preserved_file" 2>/dev/null || printf 0)', + ' preserved_fingerprint=$(file_sha256 "$preserved_file" 2>/dev/null || printf invalid)', + ' if [ "$preserved_size" != "$target_size" ] || [ "$preserved_fingerprint" != "$target_fingerprint_before" ]; then', + ' rm -f -- "$replacement_file"', + ' return 1', + ' fi', + ' if ! mv -- "$target_file" "$displaced_file"; then', + ' rm -f -- "$replacement_file"', + ' return 1', + ' fi', + ' if ! ln -- "$replacement_file" "$target_file"; then', + ' rm -f -- "$replacement_file"', + ' ln -- "$displaced_file" "$target_file" 2>/dev/null || true', + ' return 1', + ' fi', + ' rm -f -- "$replacement_file"', + ' if ! write_copy_marker "$marker_file" "$source_file" "$target_file"; then', + ' printf \'%s\\n\' "[codex-session-bridge] Installed preserved bridge without marker update: $target_file" >&2', + ' fi', + '}', 'if [ ! -d "$source_sessions_root" ]; then', ` printf '{"scannedFiles":0,"linkedFiles":0}\\n'`, ' exit 0', @@ -97,17 +216,36 @@ export function buildWslCodexSessionBridgeShellCommand( ' scanned_files=$((scanned_files + 1))', ' relative_path=${source_file#"$source_sessions_root"/}', ' target_file="$managed_sessions_root/$relative_path"', - ' if [ -e "$target_file" ] || [ -L "$target_file" ]; then', + ' marker_file="$managed_sessions_root/../.orca-session-copies/${relative_path}.json"', + ' if [ -L "$target_file" ]; then', + ' continue', + ' fi', + ' if [ -e "$target_file" ]; then', + ' if refresh_copy "$marker_file" "$source_file" "$target_file" "$relative_path"; then', + ' linked_files=$((linked_files + 1))', + ' fi', ' continue', ' fi', ' target_dir=${target_file%/*}', ' mkdir -p -- "$target_dir" || continue', - // Why: Codex resume ignores symlinked JSONL, so WSL links must be - // Linux hardlinks created inside the distro filesystem. + // Why: Codex resume ignores symlinks; cross-filesystem hardlink failure + // falls back to a marked regular-file copy. ' if ln -- "$source_file" "$target_file"; then', + ' rm -f -- "$marker_file"', ' linked_files=$((linked_files + 1))', + ' else', + ' replacement_file="${target_file}.orca-copy-$$"', + ' if cp -p -- "$source_file" "$replacement_file" && ln -- "$replacement_file" "$target_file"; then', + ' rm -f -- "$replacement_file"', + ' if ! write_copy_marker "$marker_file" "$source_file" "$target_file"; then', + ' printf \'%s\\n\' "[codex-session-bridge] Installed unmarked exclusive copy: $target_file" >&2', + ' fi', + ' linked_files=$((linked_files + 1))', + ' else', + ' rm -f -- "$replacement_file"', + ' fi', ' fi', - `done < <(find "$source_sessions_root" -type f -name '*.jsonl' -print0 2>/dev/null)`, + `done < <(find "$source_sessions_root" -type f \\( -name '*.jsonl' -o -name '*.jsonl.zst' \\) -print0 2>/dev/null)`, `printf '{"scannedFiles":%s,"linkedFiles":%s}\\n' "$scanned_files" "$linked_files"` ].join('\n') return escapeWslShCommandForWindows(shellCommand) diff --git a/src/main/native-chat/session-file-resolver.test.ts b/src/main/native-chat/session-file-resolver.test.ts index 5dd3cbc8dab..26e7e147629 100644 --- a/src/main/native-chat/session-file-resolver.test.ts +++ b/src/main/native-chat/session-file-resolver.test.ts @@ -1,4 +1,4 @@ -import { mkdir, mkdtemp, rm, writeFile } from 'node:fs/promises' +import { mkdir, mkdtemp, realpath, rm, symlink, writeFile } from 'node:fs/promises' import { tmpdir } from 'node:os' import { dirname, join } from 'node:path' import { afterEach, describe, expect, it } from 'vitest' @@ -124,6 +124,26 @@ describe('resolveSessionFilePath', () => { expect(resolved).toBe(target) }) + it('matches cold-compressed Codex rollouts and prefers a plain sibling', async () => { + const root = await makeRoot('orca-native-chat-resolve-codex-zst-') + const codexSessionsDir = join(root, 'codex-sessions') + const dayDir = join(codexSessionsDir, '2026', '06', '04') + await mkdir(dayDir, { recursive: true }) + const compressed = join(dayDir, 'rollout-2026-06-04T10-00-00-dual-session.jsonl.zst') + await writeFile(compressed, 'zstd') + + await expect( + resolveSessionFilePath('codex', 'dual-session', { codexSessionsDirs: [codexSessionsDir] }) + ).resolves.toBe(compressed) + + const plain = join(dayDir, 'rollout-2026-06-04T10-00-00-dual-session.jsonl') + await writeFile(plain, '{}\n') + + await expect( + resolveSessionFilePath('codex', 'dual-session', { codexSessionsDirs: [codexSessionsDir] }) + ).resolves.toBe(plain) + }) + it('resolves a rollout from the orca-managed Codex home (ORCA_USER_DATA_PATH)', async () => { // Orca launches Codex with its own managed CODEX_HOME, so rollout files land // under /codex-runtime-home/home/sessions, NOT ~/.codex/sessions. @@ -196,9 +216,68 @@ describe('resolveSessionFilePath', () => { const resolved = await resolveSessionFilePath('claude', 'hook-session-id', { claudeProjectsDir, - transcriptPath: realFile + transcriptPath: realFile, + requireTranscriptPathInAgentRoots: true }) - expect(resolved).toBe(realFile) + expect(resolved).toBe(await realpath(realFile)) + }) + + it('rejects a client transcriptPath outside the agent transcript roots', async () => { + const root = await makeRoot('orca-native-chat-resolve-contained-') + const claudeProjectsDir = join(root, 'claude-projects') + const projectDir = join(claudeProjectsDir, '-Users-ada-repo') + await mkdir(projectDir, { recursive: true }) + const fallback = join(projectDir, 'hook-session-id.jsonl') + const outside = join(root, 'outside.jsonl') + await writeFile(fallback, '{}\n') + await writeFile(outside, '{"secret":true}\n') + + await expect( + resolveSessionFilePath('claude', 'hook-session-id', { + claudeProjectsDir, + transcriptPath: outside, + requireTranscriptPathInAgentRoots: true + }) + ).resolves.toBe(fallback) + }) + + it('rejects a client transcriptPath that escapes through a directory symlink', async () => { + const root = await makeRoot('orca-native-chat-resolve-symlink-') + const claudeProjectsDir = join(root, 'claude-projects') + const outsideDir = join(root, 'outside') + await mkdir(claudeProjectsDir, { recursive: true }) + await mkdir(outsideDir, { recursive: true }) + await writeFile(join(outsideDir, 'secret.jsonl'), '{"secret":true}\n') + const linkedDir = join(claudeProjectsDir, 'linked-project') + await symlink(outsideDir, linkedDir, process.platform === 'win32' ? 'junction' : 'dir') + + await expect( + resolveSessionFilePath('claude', 'missing', { + claudeProjectsDir, + transcriptPath: join(linkedDir, 'secret.jsonl'), + requireTranscriptPathInAgentRoots: true + }) + ).resolves.toBeNull() + }) + + it('accepts a hook-reported .jsonl.zst path only for Codex', async () => { + const root = await makeRoot('orca-native-chat-resolve-hook-zst-') + const compressed = join(root, 'rollout-hook-session.jsonl.zst') + await writeFile(compressed, 'zstd') + + await expect( + resolveSessionFilePath('codex', 'hook-session', { + codexSessionsDirs: [root], + transcriptPath: compressed, + requireTranscriptPathInAgentRoots: true + }) + ).resolves.toBe(await realpath(compressed)) + await expect( + resolveSessionFilePath('claude', 'hook-session', { + claudeProjectsDir: join(root, 'empty'), + transcriptPath: compressed + }) + ).resolves.toBeNull() }) it('falls back to the id glob when the hook transcriptPath does not exist', async () => { diff --git a/src/main/native-chat/session-file-resolver.ts b/src/main/native-chat/session-file-resolver.ts index adfa3856f0e..793a00c2780 100644 --- a/src/main/native-chat/session-file-resolver.ts +++ b/src/main/native-chat/session-file-resolver.ts @@ -1,7 +1,13 @@ import { existsSync } from 'node:fs' +import { realpath } from 'node:fs/promises' import { homedir } from 'node:os' -import { basename, extname, join } from 'node:path' +import { basename, extname, isAbsolute, join, relative, sep } from 'node:path' import type { AgentType } from '../../shared/native-chat-types' +import { + CODEX_SESSION_ROLLOUT_EXTENSIONS, + codexRolloutBaseName, + isCodexSessionRolloutPath +} from '../ai-vault/session-scanner-codex-paths' import { walkSessionFiles } from '../ai-vault/session-scanner-discovery' import { getOrcaManagedCodexHomePath } from '../codex/codex-home-paths' import { @@ -49,6 +55,51 @@ export type ResolveSessionFileOptions = { * directly — recent Claude Code names the transcript with a UUID that differs * from the hook session_id, so the id-based glob below would miss it. */ transcriptPath?: string + /** Require a client-provided transcript path to resolve inside this agent's + * transcript roots. Runtime RPC enables this for untrusted paired clients. */ + requireTranscriptPathInAgentRoots?: boolean +} + +function agentTranscriptRoots(agent: AgentType, options: ResolveSessionFileOptions): string[] { + if (agent === 'claude') { + return [options.claudeProjectsDir ?? claudeProjectsDir()] + } + if (agent === 'codex') { + return options.codexSessionsDirs ?? codexSessionsDirs() + } + if (agent === 'grok') { + return [options.grokSessionsDir ?? grokSessionsDir()] + } + return [] +} + +async function resolveContainedPath( + filePath: string, + roots: readonly string[] +): Promise { + let resolvedFile: string + try { + resolvedFile = await realpath(filePath) + } catch { + return null + } + + for (const root of roots) { + try { + const resolvedRoot = await realpath(root) + const relativePath = relative(resolvedRoot, resolvedFile) + if ( + relativePath === '' || + (!isAbsolute(relativePath) && relativePath !== '..' && !relativePath.startsWith(`..${sep}`)) + ) { + // Return the canonical path so a symlink cannot be swapped after validation. + return resolvedFile + } + } catch { + // Missing roots cannot contain an existing transcript; try the next root. + } + } + return null } /** @@ -71,8 +122,21 @@ export async function resolveSessionFilePath( // stale/remote path falls through to the id-based search rather than returning // a non-existent file. const hookPath = options.transcriptPath?.trim() - if (hookPath && extname(hookPath) === '.jsonl' && existsSync(hookPath)) { - return hookPath + const hookPathIsTranscript = + extname(hookPath ?? '') === '.jsonl' || + (agent === 'codex' && Boolean(hookPath && isCodexSessionRolloutPath(hookPath))) + if (hookPath && hookPathIsTranscript) { + if (options.requireTranscriptPathInAgentRoots) { + const containedPath = await resolveContainedPath( + hookPath, + agentTranscriptRoots(agent, options) + ) + if (containedPath) { + return containedPath + } + } else if (existsSync(hookPath)) { + return hookPath + } } const trimmedId = sessionId.trim() @@ -108,22 +172,25 @@ async function resolveCodexSessionFile( sessionId: string, sessionsDirs: string[] ): Promise { - // Codex rollout file names embed the session id (rollout--.jsonl), so - // match the id as a suffix of the file's base name rather than an exact name. + // Codex rollout file names embed the session id. Prefer plain `.jsonl` when + // both it and a cold `.jsonl.zst` sibling exist. // Search each candidate root (managed home first) and stop at the first match. for (const sessionsDir of sessionsDirs) { if (!existsSync(sessionsDir)) { continue } const files = await walkSessionFiles(sessionsDir, 'codex', [], { - extensions: new Set(['.jsonl']), + extensions: new Set(CODEX_SESSION_ROLLOUT_EXTENSIONS), filePredicate: (path) => { - const name = basename(path, extname(path)) + if (!isCodexSessionRolloutPath(path)) { + return false + } + const name = codexRolloutBaseName(path) return name === sessionId || name.endsWith(`-${sessionId}`) } }) - if (files[0]) { - return files[0] + if (files.length > 0) { + return files.find((path) => path.endsWith('.jsonl')) ?? files[0] ?? null } } return null diff --git a/src/main/native-chat/transcript-codex-turn-items.ts b/src/main/native-chat/transcript-codex-turn-items.ts new file mode 100644 index 00000000000..bdb25f5d123 --- /dev/null +++ b/src/main/native-chat/transcript-codex-turn-items.ts @@ -0,0 +1,138 @@ +import type { NativeChatMessage } from '../../shared/native-chat-types' +import { asRecord, extractString } from '../ai-vault/session-scanner-values' +import { claudeContentBlocks } from './transcript-record-blocks' + +/** Maps a paginated Codex TurnItem into Native Chat's transcript model. */ +export function codexTurnItem( + item: Record, + id: string, + timestamp: number | null +): NativeChatMessage | null { + const itemType = normalizeCodexTurnItemType(item.type) + if (itemType === 'user_message') { + const text = codexTurnItemText(item.content) + return text + ? { id, role: 'user', blocks: [{ type: 'text', text }], timestamp, source: 'transcript' } + : null + } + if (itemType === 'agent_message') { + const blocks = claudeContentBlocks(item.content) + const fallbackText = blocks.length === 0 ? codexTurnItemText(item.content) : null + if (blocks.length === 0 && !fallbackText) { + return null + } + return { + id, + role: 'assistant', + blocks: blocks.length > 0 ? blocks : [{ type: 'text', text: fallbackText! }], + timestamp, + source: 'transcript' + } + } + if (itemType === 'reasoning') { + const text = codexReasoningItemText(item) + return text + ? { + id, + role: 'reasoning', + blocks: [{ type: 'text', text }], + timestamp, + source: 'transcript' + } + : null + } + if (itemType === 'command_execution') { + return { + id, + role: 'assistant', + blocks: [ + { + type: 'tool-call', + name: 'command_execution', + input: { + command: item.command, + cwd: item.cwd, + status: item.status, + exit_code: item.exit_code + } + } + ], + timestamp, + source: 'transcript' + } + } + if (itemType === 'dynamic_tool_call' || itemType === 'mcp_tool_call') { + const name = + extractString(item.tool) ?? + extractString(item.name) ?? + (itemType === 'mcp_tool_call' ? 'mcp_tool' : 'tool') + return { + id, + role: 'assistant', + blocks: [{ type: 'tool-call', name, input: item.arguments ?? item.input ?? item }], + timestamp, + source: 'transcript' + } + } + return null +} + +function codexTurnItemText(content: unknown): string | null { + if (typeof content === 'string') { + return extractString(content) + } + if (!Array.isArray(content)) { + const record = asRecord(content) + return extractString(record?.text) ?? extractString(record?.message) + } + const parts: string[] = [] + for (const entry of content) { + if (typeof entry === 'string') { + if (entry.trim()) { + parts.push(entry) + } + continue + } + const record = asRecord(entry) + const text = extractString(record?.text) ?? extractString(record?.content) + if (text) { + parts.push(text) + } + } + return parts.length > 0 ? parts.join('') : null +} + +function codexReasoningItemText(item: Record): string | null { + if (Array.isArray(item.summary_text)) { + const parts = item.summary_text + .map((entry) => extractString(entry)) + .filter((entry): entry is string => Boolean(entry)) + if (parts.length > 0) { + return parts.join('\n') + } + } + if (Array.isArray(item.summary)) { + const parts: string[] = [] + for (const entry of item.summary) { + const text = extractString(asRecord(entry)?.text) ?? extractString(entry) + if (text) { + parts.push(text) + } + } + if (parts.length > 0) { + return parts.join('\n') + } + } + return extractString(item.text) +} + +function normalizeCodexTurnItemType(value: unknown): string | null { + const raw = extractString(value) + if (!raw) { + return null + } + return raw + .replace(/([a-z0-9])([A-Z])/g, '$1_$2') + .replace(/([A-Z]+)([A-Z][a-z])/g, '$1_$2') + .toLowerCase() +} diff --git a/src/main/native-chat/transcript-line-decoders-codex.ts b/src/main/native-chat/transcript-line-decoders-codex.ts index 64d04fa5ae3..803d17291a8 100644 --- a/src/main/native-chat/transcript-line-decoders-codex.ts +++ b/src/main/native-chat/transcript-line-decoders-codex.ts @@ -8,6 +8,7 @@ import { timestampMs } from '../ai-vault/session-scanner-values' import { claudeContentBlocks, toolResultOutput } from './transcript-record-blocks' +import { codexTurnItem } from './transcript-codex-turn-items' export function decodeCodexTranscriptLine( line: string, @@ -99,6 +100,12 @@ function codexEventMessage( ? { id, role: 'assistant', blocks: [{ type: 'text', text }], timestamp, source: 'transcript' } : null } + // Why: paginated history emits completed TurnItems instead of the legacy + // user_message/agent_message event variants. + if (payload.type === 'item_completed') { + const item = asRecord(payload.item) + return item ? codexTurnItem(item, extractString(item.id) ?? id, timestamp) : null + } return null } diff --git a/src/main/native-chat/transcript-read-cache.test.ts b/src/main/native-chat/transcript-read-cache.test.ts index 6f0fc39acb6..b3d6ec7e77e 100644 --- a/src/main/native-chat/transcript-read-cache.test.ts +++ b/src/main/native-chat/transcript-read-cache.test.ts @@ -1,6 +1,7 @@ import { mkdir, mkdtemp, rm, utimes, writeFile } from 'node:fs/promises' import { tmpdir } from 'node:os' import { join } from 'node:path' +import { zstdCompressSync } from 'node:zlib' import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' import type * as TranscriptReader from './transcript-reader' @@ -197,13 +198,36 @@ describe('readNativeChatTranscriptCached', () => { expect(readSpy).toHaveBeenCalledTimes(5) }) - it('keeps a single active transcript larger than the whole budget cached', async () => { - // A lone entry over budget must NOT be dropped, else every read re-parses. + it('does not retain a single transcript larger than the whole cache budget', async () => { setNativeChatTranscriptCacheMaxBytesForTests(1024) const big = await seedBigFile('huge', 8 * 1024) await readNativeChatTranscriptCached('claude', 'huge', big) await readNativeChatTranscriptCached('claude', 'huge', big) - expect(readSpy).toHaveBeenCalledTimes(1) + expect(readSpy).toHaveBeenCalledTimes(2) + }) + + it('does not use compressed file bytes as the cache weight for an unbounded parse', async () => { + setNativeChatTranscriptCacheMaxBytesForTests(1024) + const root = await mkdtemp(join(tmpdir(), 'orca-native-chat-cache-zst-')) + tempRoots.push(root) + const filePath = join(root, 'rollout-session.jsonl.zst') + const record = { + type: 'event_msg', + payload: { + type: 'item_completed', + item: { + type: 'user_message', + id: 'u-zst', + content: [{ type: 'text', text: 'x'.repeat(8 * 1024) }] + } + } + } + await writeFile(filePath, zstdCompressSync(Buffer.from(JSON.stringify(record), 'utf-8'))) + + await readNativeChatTranscriptCached('codex', 'zst', filePath) + await readNativeChatTranscriptCached('codex', 'zst', filePath) + + expect(readSpy).toHaveBeenCalledTimes(2) }) it('does not evict small entries under the default budget (no regression for typical use)', async () => { diff --git a/src/main/native-chat/transcript-read-cache.ts b/src/main/native-chat/transcript-read-cache.ts index 9f4df5bde87..1a0c67812e6 100644 --- a/src/main/native-chat/transcript-read-cache.ts +++ b/src/main/native-chat/transcript-read-cache.ts @@ -1,7 +1,9 @@ import { stat } from 'node:fs/promises' import type { AgentType } from '../../shared/native-chat-types' +import { isCodexCompressedRolloutPath } from '../ai-vault/session-scanner-codex-paths' import { resolveSessionFilePath } from './session-file-resolver' import { readNativeChatTranscript, type ReadTranscriptResult } from './transcript-reader' +import type { TranscriptDecodeLimits } from './transcript-stream-lines' // Why: both the desktop IPC handler and the runtime RPC handler read the same // host-filesystem transcript, so a single process-global cache keyed by the @@ -13,16 +15,15 @@ import { readNativeChatTranscript, type ReadTranscriptResult } from './transcrip // share one sessionId yet resolve to DIFFERENT files (the same session resumed // into a second worktree, which writes a new transcript file), and a // sessionId-only key let one worktree's cached parse be served to another when -// their file mtimes momentarily coincided (#7326). The cache stores ONE -// canonical, unwindowed parse; windowing and per-surface truncation stay in the -// callers so the same parse is reused across all `limit` values and every client kind. +// their file mtimes momentarily coincided (#7326). Desktop entries store a +// canonical unwindowed parse; paired clients use a separate fixed-size tail +// entry that is reused across requested windows. type CachedTranscript = { result: ReadTranscriptResult /** mtime of the resolved file when cached; a newer mtime invalidates it. */ mtimeMs: number - /** On-disk byte size of the resolved file — a cheap, monotonic proxy for this - * entry's parsed memory footprint, used to bound the cache by total bytes. */ + /** Conservative weight for this entry's parsed memory footprint. */ bytes: number } @@ -33,12 +34,10 @@ const cache = new Map() // the oldest entry (a simple LRU once re-inserts bump recency; see setCached). const MAX_CACHE_ENTRIES = 50 // Why: a heavy Claude/Codex coding session's JSONL is routinely tens of MB (tool -// results embed whole file contents, command output, and diffs), and each cached -// entry is the full unwindowed parse. The count cap alone let 50 such entries -// retain multiple GB in the one process that now serves desktop + every paired -// web/mobile client. Bound total cached file bytes too; we always keep the most- -// recent entry (see setCached) so an active transcript is never re-parsed on -// every read, which caps the regression to extra re-parses only past this budget. +// results embed whole file contents, command output, and diffs). The count cap +// alone let 50 such entries retain multiple GB in the process serving desktop +// and paired clients. Bound total cache weight too; entries over budget are read +// successfully but not retained. const MAX_CACHE_BYTES = 128 * 1024 * 1024 // Overridable only from tests so the byte-eviction path can be exercised without // writing hundreds of MB of fixtures; production always uses MAX_CACHE_BYTES. @@ -47,15 +46,18 @@ let maxCacheBytes = MAX_CACHE_BYTES function setCached(key: string, value: CachedTranscript): void { // Re-insert moves the key to the most-recent position for LRU eviction. cache.delete(key) + // Why: retaining one entry past the budget defeats the RSS guard entirely. + // Oversized reads still succeed; they are simply re-read instead of cached. + if (!Number.isFinite(value.bytes) || value.bytes > maxCacheBytes) { + return + } cache.set(key, value) let totalBytes = 0 for (const entry of cache.values()) { totalBytes += entry.bytes } - // Evict oldest until within BOTH caps, but never drop the most-recent entry - // (cache.size > 1): a single active transcript larger than the whole budget - // must stay cached or every read would re-parse the full file. - while (cache.size > 1 && (cache.size > MAX_CACHE_ENTRIES || totalBytes > maxCacheBytes)) { + // Evict oldest until within both caps. + while (cache.size > 0 && (cache.size > MAX_CACHE_ENTRIES || totalBytes > maxCacheBytes)) { const oldest = cache.keys().next().value if (oldest === undefined) { break @@ -65,8 +67,15 @@ function setCached(key: string, value: CachedTranscript): void { } } -function cacheKey(agent: AgentType, filePath: string): string { - return `${agent}:${filePath}` +function cacheKey( + agent: AgentType, + filePath: string, + limits: TranscriptDecodeLimits | undefined +): string { + const mode = limits + ? `${limits.maxDecodedBytes ?? ''}:${limits.maxLineBytes ?? ''}:${limits.maxMessages ?? ''}` + : 'full' + return `${agent}:${filePath}:${mode}` } async function fileStat(filePath: string): Promise<{ mtimeMs: number; bytes: number }> { @@ -79,23 +88,30 @@ async function fileStat(filePath: string): Promise<{ mtimeMs: number; bytes: num } /** - * Read the full transcript for an agent + session, returning the cached parse on - * an mtime hit and re-reading (and re-caching) when the file changed. Returns the - * canonical, unwindowed result; callers apply their own windowing/truncation. + * Read a transcript for an agent + session, returning the policy-specific cached + * parse on an mtime hit and re-reading when the file changed. Desktop callers + * omit limits; paired clients cache one fixed-size tail for subsequent windows. */ export async function readNativeChatTranscriptCached( agent: AgentType, sessionId: string, /** Hook-reported authoritative transcript path, preferred over the id glob. */ - transcriptPath?: string + transcriptPath?: string, + options: { + requireTranscriptPathInAgentRoots?: boolean + limits?: TranscriptDecodeLimits + } = {} ): Promise { - const filePath = await resolveSessionFilePath(agent, sessionId, { transcriptPath }) + const filePath = await resolveSessionFilePath(agent, sessionId, { + transcriptPath, + requireTranscriptPathInAgentRoots: options.requireTranscriptPathInAgentRoots + }) if (!filePath) { return { error: `No transcript found for ${agent} session ${sessionId}` } } - const key = cacheKey(agent, filePath) - const { mtimeMs, bytes } = await fileStat(filePath) + const key = cacheKey(agent, filePath, options.limits) + const { mtimeMs, bytes: onDiskBytes } = await fileStat(filePath) const cached = cache.get(key) if (cached && Number.isFinite(mtimeMs) && cached.mtimeMs === mtimeMs) { // Bump recency so a frequently-read session survives eviction. @@ -103,9 +119,18 @@ export async function readNativeChatTranscriptCached( return cached.result } - const result = await readNativeChatTranscript(agent, sessionId, { filePath }) + const result = await readNativeChatTranscript(agent, sessionId, { + filePath, + limits: options.limits + }) if (Number.isFinite(mtimeMs)) { - setCached(key, { result, mtimeMs, bytes }) + // A compressed file's disk size says nothing about retained decoded data. + // A bounded parse can safely use its decoded cap as an upper-bound weight; + // an unbounded compressed parse is deliberately not cached. + const cacheBytes = isCodexCompressedRolloutPath(filePath) + ? (options.limits?.maxDecodedBytes ?? Number.POSITIVE_INFINITY) + : onDiskBytes + setCached(key, { result, mtimeMs, bytes: Math.max(onDiskBytes, cacheBytes) }) } return result } diff --git a/src/main/native-chat/transcript-reader.test.ts b/src/main/native-chat/transcript-reader.test.ts index c6c2223c81f..9bb95297a9c 100644 --- a/src/main/native-chat/transcript-reader.test.ts +++ b/src/main/native-chat/transcript-reader.test.ts @@ -1,6 +1,7 @@ import { mkdtemp, rm, writeFile } from 'node:fs/promises' import { tmpdir } from 'node:os' import { join } from 'node:path' +import { zstdCompressSync } from 'node:zlib' import { afterEach, describe, expect, it } from 'vitest' import { readNativeChatTranscript } from './transcript-reader' @@ -173,6 +174,212 @@ describe('readNativeChatTranscript (claude)', () => { }) describe('readNativeChatTranscript (codex)', () => { + it('maps paginated item_completed turn items into chat messages', async () => { + const filePath = await writeFixture('orca-native-chat-codex-paginated-', [ + { + type: 'event_msg', + timestamp: '2026-06-01T10:00:01.000Z', + payload: { + type: 'item_completed', + item: { + type: 'UserMessage', + id: 'user-1', + content: [{ type: 'text', text: 'Paginated hello' }] + } + } + }, + { + type: 'event_msg', + timestamp: '2026-06-01T10:00:02.000Z', + payload: { + type: 'item_completed', + item: { + type: 'reasoning', + id: 'reason-1', + summary_text: ['Thinking about the answer'] + } + } + }, + { + type: 'event_msg', + timestamp: '2026-06-01T10:00:03.000Z', + payload: { + type: 'item_completed', + item: { + type: 'command_execution', + id: 'cmd-1', + command: ['bash', '-lc', 'ls'], + cwd: '/repo', + status: 'completed', + exit_code: 0 + } + } + }, + { + type: 'event_msg', + timestamp: '2026-06-01T10:00:04.000Z', + payload: { + type: 'item_completed', + item: { + type: 'agent_message', + id: 'agent-1', + content: [{ type: 'text', text: 'Done.' }] + } + } + } + ]) + + const result = await readNativeChatTranscript('codex', 'codex-paginated', { filePath }) + if (!('messages' in result)) { + throw new Error('expected messages') + } + + expect(result.messages.map((message) => message.role)).toEqual([ + 'user', + 'reasoning', + 'assistant', + 'assistant' + ]) + expect(result.messages[0]?.blocks[0]).toEqual({ type: 'text', text: 'Paginated hello' }) + expect(result.messages[1]?.blocks[0]).toEqual({ + type: 'text', + text: 'Thinking about the answer' + }) + expect(result.messages[2]?.blocks[0]).toEqual({ + type: 'tool-call', + name: 'command_execution', + input: { + command: ['bash', '-lc', 'ls'], + cwd: '/repo', + status: 'completed', + exit_code: 0 + } + }) + expect(result.messages[3]?.blocks[0]).toEqual({ type: 'text', text: 'Done.' }) + }) + + it('reads cold-compressed .jsonl.zst transcripts', async () => { + const root = await mkdtemp(join(tmpdir(), 'orca-native-chat-codex-zst-')) + tempRoots.push(root) + const filePath = join(root, 'rollout-session.jsonl.zst') + const plain = jsonLines([ + { + type: 'event_msg', + timestamp: '2026-06-01T10:00:01.000Z', + payload: { + type: 'item_completed', + item: { + type: 'user_message', + id: 'u1', + content: [{ type: 'text', text: 'From zst' }] + } + } + }, + { + type: 'event_msg', + timestamp: '2026-06-01T10:00:02.000Z', + payload: { + type: 'item_completed', + item: { + type: 'agent_message', + id: 'a1', + content: [{ type: 'text', text: 'Compressed reply' }] + } + } + } + ]) + await writeFile(filePath, zstdCompressSync(Buffer.from(plain, 'utf-8'))) + + const result = await readNativeChatTranscript('codex', 'session', { filePath }) + if (!('messages' in result)) { + throw new Error('expected messages') + } + expect(result.messages.map((message) => message.role)).toEqual(['user', 'assistant']) + expect(result.messages[0]?.blocks[0]).toEqual({ type: 'text', text: 'From zst' }) + }) + + it('stops compressed transcript decoding at the decompressed byte limit', async () => { + const root = await mkdtemp(join(tmpdir(), 'orca-native-chat-codex-zst-byte-limit-')) + tempRoots.push(root) + const filePath = join(root, 'rollout-session.jsonl.zst') + const plain = jsonLines( + Array.from({ length: 20 }, (_unused, index) => ({ + type: 'event_msg', + payload: { + type: 'item_completed', + item: { + type: 'user_message', + id: `u${index}`, + content: [{ type: 'text', text: 'x'.repeat(100) }] + } + } + })) + ) + await writeFile(filePath, zstdCompressSync(Buffer.from(plain, 'utf-8'))) + + const result = await readNativeChatTranscript('codex', 'session', { + filePath, + limits: { maxDecodedBytes: 512 } + }) + + expect(result).toEqual({ error: expect.stringContaining('decoded byte limit') }) + }) + + it('rejects a decompressed transcript line over the per-line byte limit', async () => { + const root = await mkdtemp(join(tmpdir(), 'orca-native-chat-codex-zst-line-limit-')) + tempRoots.push(root) + const filePath = join(root, 'rollout-session.jsonl.zst') + const plain = jsonLines([ + { + type: 'event_msg', + payload: { + type: 'item_completed', + item: { + type: 'user_message', + id: 'u1', + content: [{ type: 'text', text: 'x'.repeat(1024) }] + } + } + } + ]) + await writeFile(filePath, zstdCompressSync(Buffer.from(plain, 'utf-8'))) + + const result = await readNativeChatTranscript('codex', 'session', { + filePath, + limits: { maxDecodedBytes: 4096, maxLineBytes: 256 } + }) + + expect(result).toEqual({ error: expect.stringContaining('line byte limit') }) + }) + + it('keeps only the newest messages while decoding a bounded transcript window', async () => { + const filePath = await writeFixture( + 'orca-native-chat-codex-bounded-', + [1, 2, 3, 4].map((index) => ({ + type: 'event_msg', + timestamp: `2026-06-01T10:00:0${index}.000Z`, + payload: { + type: 'item_completed', + item: { + type: 'user_message', + id: `u${index}`, + content: [{ type: 'text', text: `message-${index}` }] + } + } + })) + ) + + const result = await readNativeChatTranscript('codex', 'session', { + filePath, + limits: { maxMessages: 2 } + }) + + if (!('messages' in result)) { + throw new Error(`expected messages, got ${result.error}`) + } + expect(result.messages.map((message) => message.id)).toEqual(['u3', 'u4']) + }) + it('maps tool calls and results to tool-call/tool-result blocks', async () => { const filePath = await writeFixture('orca-native-chat-codex-', [ { diff --git a/src/main/native-chat/transcript-reader.ts b/src/main/native-chat/transcript-reader.ts index 1dbbd7dd82a..75d2974919c 100644 --- a/src/main/native-chat/transcript-reader.ts +++ b/src/main/native-chat/transcript-reader.ts @@ -1,5 +1,7 @@ import { createReadStream } from 'node:fs' import type { AgentType, NativeChatMessage } from '../../shared/native-chat-types' +import { isCodexCompressedRolloutPath } from '../ai-vault/session-scanner-codex-paths' +import { openCodexRolloutStream } from '../ai-vault/session-scanner-codex-rollout-read' import { errorMessage } from '../ai-vault/session-scanner-values' import { resolveSessionFilePath, type ResolveSessionFileOptions } from './session-file-resolver' import { @@ -8,20 +10,24 @@ import { decodeGrokTranscriptLine } from './transcript-line-decoders' import { decodeTranscriptStream } from './transcript-stream-lines' +import type { TranscriptDecodeLimits } from './transcript-stream-lines' export type ReadTranscriptResult = { messages: NativeChatMessage[] } | { error: string } export type ReadTranscriptOptions = ResolveSessionFileOptions & { /** Resolve directly to this file, skipping path discovery (used by tests). */ filePath?: string + /** Optional streaming limits for remote/windowed readers. Omitted for the + * desktop full-history contract. */ + limits?: TranscriptDecodeLimits } /** - * Read the ENTIRE Claude/Codex JSONL transcript for an agent + session id into - * the NativeChatMessage model. Unlike the AI-Vault preview scan, this applies - * NO message cap. Unknown record types are skipped rather than throwing, so a - * single malformed/unrecognized line cannot fail the whole read. The per-line - * record-to-message mapping is shared with the live tailer. + * Read a Claude/Codex JSONL transcript for an agent + session id into the + * NativeChatMessage model. Desktop callers omit limits and retain full history; + * remote callers provide streaming limits. Unknown record types are skipped + * rather than failing the whole read. The per-line mapping is shared with the + * live tailer. */ export async function readNativeChatTranscript( agent: AgentType, @@ -34,13 +40,15 @@ export async function readNativeChatTranscript( } try { if (agent === 'claude') { - return { messages: await readTranscript(filePath, decodeClaudeTranscriptLine) } + return { + messages: await readTranscript(filePath, decodeClaudeTranscriptLine, options.limits) + } } if (agent === 'codex') { - return { messages: await readTranscript(filePath, decodeCodexTranscriptLine) } + return { messages: await readTranscript(filePath, decodeCodexTranscriptLine, options.limits) } } if (agent === 'grok') { - return { messages: await readTranscript(filePath, decodeGrokTranscriptLine) } + return { messages: await readTranscript(filePath, decodeGrokTranscriptLine, options.limits) } } return { error: `Unsupported agent for native chat transcript: ${agent}` } } catch (err) { @@ -50,9 +58,14 @@ export async function readNativeChatTranscript( async function readTranscript( filePath: string, - decode: (line: string, fallbackId: string) => NativeChatMessage | null + decode: (line: string, fallbackId: string) => NativeChatMessage | null, + limits?: TranscriptDecodeLimits ): Promise { - const stream = createReadStream(filePath, { encoding: 'utf-8' }) - const { messages } = await decodeTranscriptStream(stream, filePath, 0, decode, true) + // Why: Codex cold-compresses older rollouts; other agents remain plain JSONL. + // Keep plain reads as bytes so malformed UTF-8 cannot distort safety limits. + const stream = isCodexCompressedRolloutPath(filePath) + ? openCodexRolloutStream(filePath) + : createReadStream(filePath) + const { messages } = await decodeTranscriptStream(stream, filePath, 0, decode, true, limits) return messages } diff --git a/src/main/native-chat/transcript-stream-lines.test.ts b/src/main/native-chat/transcript-stream-lines.test.ts index 2bfa6f6f9d2..de6793a7f7d 100644 --- a/src/main/native-chat/transcript-stream-lines.test.ts +++ b/src/main/native-chat/transcript-stream-lines.test.ts @@ -11,6 +11,74 @@ const decode = (line: string, id: string) => ({ }) describe('decodeTranscriptStream', () => { + it.each([ + ['emoji', '🙂', 2], + ['CJK character', '汉', 1] + ])('preserves a split %s across buffer chunks', async (_name, character, splitOffset) => { + const prefix = '{"value":"' + const line = `${prefix}${character}"}` + const encoded = Buffer.from(`${line}\n`, 'utf8') + const splitIndex = Buffer.byteLength(prefix, 'utf8') + splitOffset + + const result = await decodeTranscriptStream( + Readable.from([encoded.subarray(0, splitIndex), encoded.subarray(splitIndex)]), + '/chat.jsonl', + 0, + decode, + true, + { maxDecodedBytes: encoded.length, maxLineBytes: Buffer.byteLength(line, 'utf8') } + ) + + expect(result.messages[0]?.blocks).toEqual([{ type: 'text', text: line }]) + expect(result.consumedBytes).toBe(encoded.length) + }) + + it('flushes an incomplete UTF-8 tail without advancing past its raw bytes', async () => { + const encoded = Buffer.concat([Buffer.from('partial:'), Buffer.from([0xf0, 0x9f])]) + + const result = await decodeTranscriptStream( + Readable.from([encoded]), + '/chat.jsonl', + 0, + decode, + true + ) + + expect(result.messages[0]?.blocks).toEqual([{ type: 'text', text: 'partial:\uFFFD' }]) + expect(result.consumedBytes).toBe(encoded.length) + }) + + it('uses raw byte offsets after malformed UTF-8 before a partial next line', async () => { + const complete = Buffer.from([0xff, 0x0a]) + const partial = Buffer.from('x') + + const result = await decodeTranscriptStream( + Readable.from([complete, partial]), + '/chat.jsonl', + 0, + decode, + false + ) + + expect(result.messages[0]?.blocks).toEqual([{ type: 'text', text: '\uFFFD' }]) + expect(result.consumedBytes).toBe(complete.length) + }) + + it('preserves a surrogate pair split across string chunks', async () => { + const line = '{"value":"🙂"}' + + const result = await decodeTranscriptStream( + Readable.from(['{"value":"\ud83d', '\ude42"}\n']), + '/chat.jsonl', + 0, + decode, + true + ) + + expect(result.messages[0]?.blocks).toEqual([{ type: 'text', text: line }]) + expect(result.consumedBytes).toBe(Buffer.byteLength(`${line}\n`, 'utf8')) + }) + it('uses identical absolute byte ids for full and incremental reads', async () => { const prefix = '{"first":"é"}\r\n' const appended = '{"second":true}\n' diff --git a/src/main/native-chat/transcript-stream-lines.ts b/src/main/native-chat/transcript-stream-lines.ts index a764941ad30..5ca47054445 100644 --- a/src/main/native-chat/transcript-stream-lines.ts +++ b/src/main/native-chat/transcript-stream-lines.ts @@ -1,47 +1,133 @@ import type { Readable } from 'node:stream' +import { StringDecoder } from 'node:string_decoder' import type { NativeChatMessage } from '../../shared/native-chat-types' import { transcriptFallbackId } from './transcript-fallback-id' type TranscriptDecoder = (line: string, fallbackId: string) => NativeChatMessage | null +export type TranscriptDecodeLimits = { + /** Maximum UTF-8 bytes accepted from the decoded stream. */ + maxDecodedBytes?: number + /** Maximum UTF-8 bytes accepted in one JSONL record. */ + maxLineBytes?: number + /** Keep only the newest decoded messages while continuing to scan the stream. */ + maxMessages?: number +} + +export class TranscriptDecodeLimitError extends Error { + constructor(kind: 'decoded byte' | 'line byte', limit: number) { + super(`Transcript ${kind} limit exceeded (${limit} bytes)`) + this.name = 'TranscriptDecodeLimitError' + } +} + export async function decodeTranscriptStream( stream: Readable, filePath: string, start: number, decode: TranscriptDecoder, - includeTrailingLine: boolean + includeTrailingLine: boolean, + limits: TranscriptDecodeLimits = {} ): Promise<{ messages: NativeChatMessage[]; consumedBytes: number }> { const messages: NativeChatMessage[] = [] let pending = '' + let pendingBytes = 0 + let decodedBytes = 0 let consumedBytes = 0 + let oldestMessageIndex = 0 + let pendingHighSurrogate = '' + const utf8Decoder = new StringDecoder('utf8') + + assertPositiveLimit(limits.maxDecodedBytes, 'decoded byte') + assertPositiveLimit(limits.maxLineBytes, 'line byte') + assertPositiveLimit(limits.maxMessages, 'message') for await (const chunk of stream) { - pending += typeof chunk === 'string' ? chunk : Buffer.from(chunk).toString('utf8') - let newlineIndex = pending.indexOf('\n') - while (newlineIndex !== -1) { - const segment = pending.slice(0, newlineIndex + 1) - decodeLine(segment.slice(0, -1), consumedBytes) - consumedBytes += Buffer.byteLength(segment, 'utf8') - pending = pending.slice(newlineIndex + 1) - newlineIndex = pending.indexOf('\n') + if (typeof chunk === 'string') { + let text = pendingHighSurrogate + chunk + pendingHighSurrogate = '' + const lastCodeUnit = text.charCodeAt(text.length - 1) + if (lastCodeUnit >= 0xd800 && lastCodeUnit <= 0xdbff) { + pendingHighSurrogate = text.slice(-1) + text = text.slice(0, -1) + } + consumeTranscriptBytes(Buffer.from(text, 'utf8')) + continue + } + if (pendingHighSurrogate) { + consumeTranscriptBytes(Buffer.from(pendingHighSurrogate, 'utf8')) + pendingHighSurrogate = '' } + consumeTranscriptBytes(Buffer.from(chunk)) + } + if (pendingHighSurrogate) { + consumeTranscriptBytes(Buffer.from(pendingHighSurrogate, 'utf8')) } + pending += utf8Decoder.end() if (includeTrailingLine && pending.length > 0) { - decodeLine(pending, consumedBytes) - consumedBytes += Buffer.byteLength(pending, 'utf8') + enforceLineLimit(pendingBytes) + decodeLine(pending, consumedBytes, pendingBytes) + consumedBytes += pendingBytes } - return { messages, consumedBytes } + const orderedMessages = + oldestMessageIndex === 0 + ? messages + : messages.slice(oldestMessageIndex).concat(messages.slice(0, oldestMessageIndex)) + return { messages: orderedMessages, consumedBytes } + + function consumeTranscriptBytes(bytes: Buffer): void { + decodedBytes += bytes.length + if (limits.maxDecodedBytes !== undefined && decodedBytes > limits.maxDecodedBytes) { + throw new TranscriptDecodeLimitError('decoded byte', limits.maxDecodedBytes) + } + let chunkOffset = 0 + let newlineIndex = bytes.indexOf(0x0a) + while (newlineIndex !== -1) { + const segment = bytes.subarray(chunkOffset, newlineIndex + 1) + pending += utf8Decoder.write(segment) + pendingBytes += segment.length + decodeLine(pending.slice(0, -1), consumedBytes, pendingBytes - 1) + consumedBytes += pendingBytes + pending = '' + pendingBytes = 0 + chunkOffset = newlineIndex + 1 + newlineIndex = bytes.indexOf(0x0a, chunkOffset) + } + const trailing = bytes.subarray(chunkOffset) + pending += utf8Decoder.write(trailing) + pendingBytes += trailing.length + enforceLineLimit(pendingBytes) + } - function decodeLine(rawLine: string, relativeOffset: number): void { - const line = rawLine.endsWith('\r') ? rawLine.slice(0, -1) : rawLine + function decodeLine(rawLine: string, relativeOffset: number, rawLineBytes: number): void { + const hasCarriageReturn = rawLine.endsWith('\r') + const line = hasCarriageReturn ? rawLine.slice(0, -1) : rawLine if (!line) { return } + enforceLineLimit(rawLineBytes - (hasCarriageReturn ? 1 : 0)) const message = decode(line, transcriptFallbackId(filePath, start + relativeOffset)) if (message) { - messages.push(message) + if (limits.maxMessages === undefined || messages.length < limits.maxMessages) { + messages.push(message) + } else { + messages[oldestMessageIndex] = message + oldestMessageIndex = (oldestMessageIndex + 1) % limits.maxMessages + } + } + } + + function enforceLineLimit(bytes: number): void { + if (limits.maxLineBytes !== undefined && bytes > limits.maxLineBytes) { + throw new TranscriptDecodeLimitError('line byte', limits.maxLineBytes) } } } + +function assertPositiveLimit(value: number | undefined, name: string): void { + if (value !== undefined && (!Number.isSafeInteger(value) || value <= 0)) { + throw new Error(`Transcript ${name} limit must be a positive safe integer`) + } +} diff --git a/src/main/native-chat/transcript-watch.test.ts b/src/main/native-chat/transcript-watch.test.ts index 55009521a99..ba89a3a5dbc 100644 --- a/src/main/native-chat/transcript-watch.test.ts +++ b/src/main/native-chat/transcript-watch.test.ts @@ -16,7 +16,7 @@ afterEach(async () => { tempRoots = [] }) -async function tempFile(initial: string): Promise { +async function tempFile(initial: string | Uint8Array): Promise { const root = await mkdtemp(join(tmpdir(), 'orca-native-chat-watch-')) tempRoots.push(root) const filePath = join(root, 'rollout.jsonl') @@ -122,6 +122,36 @@ describe('subscribeNativeChatTranscript', () => { expect(seen.some((m) => m.id === 'a-2')).toBe(true) }) + it('skips oversized initial history once and continues tailing later appends', async () => { + const filePath = await tempFile(claudeLine('u-large', 'user', 'x'.repeat(1024))) + const seen: NativeChatMessage[] = [] + let appendedDuringRead = false + + const sub = await subscribeNativeChatTranscript({ + agent: 'claude', + sessionId: 'ignored', + filePath, + onAppend: (messages) => seen.push(...messages), + debounceMs: 5, + limits: { maxDecodedBytes: 256, maxLineBytes: 2048, maxMessages: 40 }, + afterReadSnapshotForTests: async () => { + if (appendedDuringRead) { + return + } + appendedDuringRead = true + await appendFile(filePath, claudeLine('a-after-limit', 'assistant', 'still live')) + // Let the watcher turn this append into pendingReadRequested while the + // oversized seed read still owns the drain. + await new Promise((resolve) => setTimeout(resolve, 20)) + } + }) + + await waitFor(() => seen.some((message) => message.id === 'a-after-limit'), 500) + + sub.unsubscribe() + expect(seen.some((message) => message.id === 'u-large')).toBe(false) + }) + it('releases the watcher on unsubscribe (no leak)', async () => { const filePath = await tempFile(claudeLine('u-1', 'user', 'hi')) const before = getActiveNativeChatWatcherCount() @@ -196,6 +226,36 @@ describe('subscribeNativeChatTranscript', () => { expect(seen.filter((m) => m.id === 'a-partial')).toHaveLength(1) }) + it('keeps malformed UTF-8 from advancing into the next partial record', async () => { + const seed = claudeLine('u-seed', 'user', 'before malformed offset check') + const filePath = await tempFile( + Buffer.concat([Buffer.from([0xff, 0x0a]), Buffer.from(seed, 'utf8')]) + ) + const seen: NativeChatMessage[] = [] + + const sub = await subscribeNativeChatTranscript({ + agent: 'claude', + sessionId: 'ignored', + filePath, + onAppend: (messages) => seen.push(...messages), + debounceMs: 5 + }) + + await waitFor(() => seen.some((message) => message.id === 'u-seed')) + + const line = claudeLine('a-after-malformed', 'assistant', 'offset stayed aligned') + const splitAt = Math.floor(line.length / 2) + await appendFile(filePath, line.slice(0, splitAt)) + await new Promise((resolve) => setTimeout(resolve, 40)) + expect(seen.some((message) => message.id === 'a-after-malformed')).toBe(false) + + await appendFile(filePath, line.slice(splitAt)) + await waitFor(() => seen.some((message) => message.id === 'a-after-malformed')) + + sub.unsubscribe() + expect(seen.filter((message) => message.id === 'a-after-malformed')).toHaveLength(1) + }) + it('survives file replacement / rotation (offset reset on shrink)', async () => { const filePath = await tempFile( claudeLine('u-1', 'user', 'old') + claudeLine('a-1', 'assistant', 'old-reply') diff --git a/src/main/native-chat/transcript-watch.ts b/src/main/native-chat/transcript-watch.ts index 4bbd934f0eb..2784b27fadd 100644 --- a/src/main/native-chat/transcript-watch.ts +++ b/src/main/native-chat/transcript-watch.ts @@ -17,7 +17,8 @@ import { decodeCodexTranscriptLine, decodeGrokTranscriptLine } from './transcript-line-decoders' -import { decodeTranscriptStream } from './transcript-stream-lines' +import { decodeTranscriptStream, TranscriptDecodeLimitError } from './transcript-stream-lines' +import type { TranscriptDecodeLimits } from './transcript-stream-lines' export type SubscribeNativeChatTranscriptArgs = ResolveSessionFileOptions & { agent: AgentType @@ -29,6 +30,10 @@ export type SubscribeNativeChatTranscriptArgs = ResolveSessionFileOptions & { filePath?: string /** Coalesce window for rapid fs.watch events (ms). Defaults to 40ms. */ debounceMs?: number + /** Optional streaming limits for paired-client subscriptions. */ + limits?: TranscriptDecodeLimits + /** Test-only synchronization point after a read's EOF snapshot is fixed. */ + afterReadSnapshotForTests?: (end: number) => Promise } export type NativeChatTranscriptSubscription = { @@ -85,9 +90,10 @@ async function fileSize(filePath: string): Promise { async function readAppendedMessages( filePath: string, start: number, - decode: (line: string, fallbackId: string) => NativeChatMessage | null + end: number, + decode: (line: string, fallbackId: string) => NativeChatMessage | null, + limits?: TranscriptDecodeLimits ): Promise<{ messages: NativeChatMessage[]; consumedTo: number }> { - const end = await fileSize(filePath) if (end <= start) { // File shrank (rotation/replacement) or unchanged — caller resets offset. return { messages: [], consumedTo: end } @@ -95,8 +101,9 @@ async function readAppendedMessages( const handle = await open(filePath, 'r') try { + // Why: decoding here would turn malformed bytes into larger replacement + // sequences and advance the persisted file offset past unread records. const stream = handle.createReadStream({ - encoding: 'utf-8', start, end: end - 1, autoClose: false @@ -106,7 +113,8 @@ async function readAppendedMessages( filePath, start, decode, - false + false, + limits ) return { messages, consumedTo: start + consumedBytes } } finally { @@ -160,18 +168,34 @@ export async function subscribeNativeChatTranscript( try { do { pendingReadRequested = false + let attemptedEnd = offset try { const currentSize = await fileSize(filePath!) if (currentSize < offset) { // Rotation/replacement/truncation: re-read from the top. offset = 0 } - const { messages, consumedTo } = await readAppendedMessages(filePath!, offset, decode!) + attemptedEnd = currentSize + await args.afterReadSnapshotForTests?.(currentSize) + const { messages, consumedTo } = await readAppendedMessages( + filePath!, + offset, + currentSize, + decode!, + args.limits + ) offset = consumedTo if (!closed && messages.length > 0) { onAppend(messages) } - } catch { + } catch (error) { + if (error instanceof TranscriptDecodeLimitError) { + // Why: retrying an oversized seed from byte zero on every append + // creates a permanent decode loop. Skip that history once, then + // continue tailing records appended after this read snapshot. + offset = attemptedEnd + continue + } // Why: a transient read failure (EACCES/EIO/ENOENT during rotation) // must not leave the subscription permanently deaf. Stop this drain; // the finally resets `reading` so a later fs event re-arms the read. diff --git a/src/main/runtime/rpc/methods/native-chat.test.ts b/src/main/runtime/rpc/methods/native-chat.test.ts index 720ce127d2c..a59d7ebb30b 100644 --- a/src/main/runtime/rpc/methods/native-chat.test.ts +++ b/src/main/runtime/rpc/methods/native-chat.test.ts @@ -5,9 +5,19 @@ import type { RpcContext } from '../core' // Stub the shared cache so the handler returns a deterministic transcript with // one oversized tool-result block; the test then asserts clip behavior per client. const OVERSIZED = 'x'.repeat(5000) -const cachedResult = vi.hoisted(() => ({ value: { messages: [] as NativeChatMessage[] } })) +const { cachedResult, cachedReadSpy, subscribeSpy } = vi.hoisted(() => ({ + cachedResult: { value: { messages: [] as NativeChatMessage[] } }, + cachedReadSpy: vi.fn(), + subscribeSpy: vi.fn((_args: unknown) => Promise.resolve({ unsubscribe: vi.fn() })) +})) vi.mock('../../../native-chat/transcript-read-cache', () => ({ - readNativeChatTranscriptCached: () => Promise.resolve(cachedResult.value) + readNativeChatTranscriptCached: (...args: unknown[]) => { + cachedReadSpy(...args) + return Promise.resolve(cachedResult.value) + } +})) +vi.mock('../../../native-chat/transcript-watch', () => ({ + subscribeNativeChatTranscript: (args: unknown) => subscribeSpy(args) })) import { NATIVE_CHAT_METHODS } from './native-chat' @@ -30,6 +40,22 @@ function readSessionHandler(): (params: unknown, ctx: RpcContext) => Promise Promise } +function subscribeHandler(): ( + params: unknown, + ctx: RpcContext, + emit: (event: unknown) => void +) => Promise { + const method = NATIVE_CHAT_METHODS.find((m) => m.name === 'nativeChat.subscribe') + if (!method) { + throw new Error('subscribe method not registered') + } + return method.handler as ( + params: unknown, + ctx: RpcContext, + emit: (event: unknown) => void + ) => Promise +} + function ctxWith(clientKind: RpcContext['clientKind']): RpcContext { return { runtime: {} as RpcContext['runtime'], clientKind } } @@ -41,6 +67,55 @@ function firstOutput(result: unknown): string { } describe('nativeChat.readSession clientKind truncation gating', () => { + it('requests a root-contained, bounded parse before applying the client window', async () => { + cachedResult.value = { messages: [] } + cachedReadSpy.mockClear() + + await readSessionHandler()( + { + agent: 'codex', + sessionId: 's', + limit: 40, + transcriptPath: '/outside/rollout.jsonl.zst' + }, + ctxWith('runtime') + ) + + expect(cachedReadSpy).toHaveBeenCalledWith( + 'codex', + 's', + '/outside/rollout.jsonl.zst', + expect.objectContaining({ + requireTranscriptPathInAgentRoots: true, + limits: expect.objectContaining({ maxMessages: 2000 }) + }) + ) + }) + + it('applies the same path and stream limits to paired-client subscriptions', async () => { + subscribeSpy.mockClear() + const runtime = { + registerSubscriptionCleanup: vi.fn() + } as unknown as RpcContext['runtime'] + + await subscribeHandler()( + { + agent: 'codex', + sessionId: 's', + transcriptPath: '/outside/rollout.jsonl.zst' + }, + { runtime, connectionId: 'connection-1', clientKind: 'runtime' }, + vi.fn() + ) + + expect(subscribeSpy).toHaveBeenCalledWith( + expect.objectContaining({ + requireTranscriptPathInAgentRoots: true, + limits: expect.objectContaining({ maxMessages: 2000 }) + }) + ) + }) + it('clips oversized tool output for mobile clients', async () => { cachedResult.value = { messages: [makeMessage(OVERSIZED)] } const result = await readSessionHandler()( diff --git a/src/main/runtime/rpc/methods/native-chat.ts b/src/main/runtime/rpc/methods/native-chat.ts index 05558c8b76c..6681cfc9164 100644 --- a/src/main/runtime/rpc/methods/native-chat.ts +++ b/src/main/runtime/rpc/methods/native-chat.ts @@ -55,6 +55,10 @@ const NativeChatUnsubscribe = z.object({ // older history as the user scrolls back. const MOBILE_NATIVE_CHAT_DEFAULT_WINDOW = 40 const MOBILE_NATIVE_CHAT_MAX_WINDOW = 2000 +// Why: paired clients must never make the host retain or decode an unbounded +// transcript. The desktop IPC path remains uncapped for full-history access. +const REMOTE_TRANSCRIPT_MAX_DECODED_BYTES = 64 * 1024 * 1024 +const REMOTE_TRANSCRIPT_MAX_LINE_BYTES = 8 * 1024 * 1024 // Why: a single tool result (a big file read, a long diff) can be hundreds of KB. // The mobile view only previews block bodies, so truncate them on the wire to // keep the payload small; the marker tells the user content was clipped. @@ -116,7 +120,17 @@ export const NATIVE_CHAT_METHODS: readonly RpcAnyMethod[] = [ const result = await readNativeChatTranscriptCached( params.agent, params.sessionId, - params.transcriptPath + params.transcriptPath, + { + requireTranscriptPathInAgentRoots: true, + // Parse once into the largest supported tail so every smaller page + // reuses the same bounded cache entry. + limits: { + maxDecodedBytes: REMOTE_TRANSCRIPT_MAX_DECODED_BYTES, + maxLineBytes: REMOTE_TRANSCRIPT_MAX_LINE_BYTES, + maxMessages: MOBILE_NATIVE_CHAT_MAX_WINDOW + } + } ) // Window to the conversation tail (all clients); clip blocks for mobile only. return 'messages' in result @@ -157,6 +171,12 @@ export const NATIVE_CHAT_METHODS: readonly RpcAnyMethod[] = [ agent: params.agent, sessionId: params.sessionId, transcriptPath: params.transcriptPath, + requireTranscriptPathInAgentRoots: true, + limits: { + maxDecodedBytes: REMOTE_TRANSCRIPT_MAX_DECODED_BYTES, + maxLineBytes: REMOTE_TRANSCRIPT_MAX_LINE_BYTES, + maxMessages: MOBILE_NATIVE_CHAT_MAX_WINDOW + }, onAppend: (messages) => { if (closed) { return diff --git a/src/renderer/src/components/native-chat/native-chat-composer-state.test.ts b/src/renderer/src/components/native-chat/native-chat-composer-state.test.ts index 499d2b1610c..b9e377d4982 100644 --- a/src/renderer/src/components/native-chat/native-chat-composer-state.test.ts +++ b/src/renderer/src/components/native-chat/native-chat-composer-state.test.ts @@ -144,6 +144,27 @@ describe('filterSkillSuggestions', () => { ]) expect(filterSkillSuggestions(skills, 'h')).toEqual([]) }) + + it('dedupes same-name skills and prefers the repository source', () => { + const skills = [ + skill({ + id: 'home-review', + name: 'review', + sourceKind: 'home', + skillFilePath: '/Users/test/.agents/skills/review/SKILL.md' + }), + skill({ + id: 'repo-review', + name: 'Review', + sourceKind: 'repo', + skillFilePath: '/repo/.agents/skills/review/SKILL.md' + }) + ] + + const filtered = filterSkillSuggestions(skills, 'rev') + expect(filtered).toHaveLength(1) + expect(filtered[0]?.skillFilePath).toBe('/repo/.agents/skills/review/SKILL.md') + }) }) describe('history recall', () => { diff --git a/src/renderer/src/components/native-chat/native-chat-composer-state.ts b/src/renderer/src/components/native-chat/native-chat-composer-state.ts index 9871bba1b95..8c8d65d681c 100644 --- a/src/renderer/src/components/native-chat/native-chat-composer-state.ts +++ b/src/renderer/src/components/native-chat/native-chat-composer-state.ts @@ -75,16 +75,39 @@ export function filterSkillSuggestions( ): DiscoveredSkill[] { const normalized = query.toLowerCase() const installed = skills.filter((skill) => skill.installed) - if (normalized === '') { - return installed.slice(0, 12) + const matched = + normalized === '' + ? installed + : installed.filter((skill) => { + const name = skill.name.toLowerCase() + const dirName = skill.directoryPath.split(/[\\/]/).findLast(Boolean)?.toLowerCase() + return name.startsWith(normalized) || dirName?.startsWith(normalized) + }) + return dedupeSkillSuggestions(matched).slice(0, 12) +} + +const SKILL_SOURCE_PRIORITY: Record = { + repo: 0, + home: 1, + plugin: 2, + bundled: 3 +} + +function dedupeSkillSuggestions(skills: readonly DiscoveredSkill[]): DiscoveredSkill[] { + // Why: PTY fallback inserts only `$name`; prefer the repository definition + // when multiple roots expose an otherwise ambiguous name. + const byName = new Map() + for (const skill of skills) { + const key = skill.name.toLowerCase() + const existing = byName.get(key) + if ( + !existing || + SKILL_SOURCE_PRIORITY[skill.sourceKind] < SKILL_SOURCE_PRIORITY[existing.sourceKind] + ) { + byName.set(key, skill) + } } - return installed - .filter((skill) => { - const name = skill.name.toLowerCase() - const dirName = skill.directoryPath.split(/[\\/]/).findLast(Boolean)?.toLowerCase() - return name.startsWith(normalized) || dirName?.startsWith(normalized) - }) - .slice(0, 12) + return [...byName.values()] } export type HistoryState = { diff --git a/src/renderer/src/components/terminal-pane/pty-connection.test.ts b/src/renderer/src/components/terminal-pane/pty-connection.test.ts index 6c72e259c72..fb59cbd7c1f 100644 --- a/src/renderer/src/components/terminal-pane/pty-connection.test.ts +++ b/src/renderer/src/components/terminal-pane/pty-connection.test.ts @@ -895,6 +895,7 @@ describe('connectPanePty', () => { afterEach(() => { vi.useRealTimers() + vi.restoreAllMocks() if (originalRequestAnimationFrame) { globalThis.requestAnimationFrame = originalRequestAnimationFrame } else { @@ -16480,6 +16481,9 @@ describe('connectPanePty', () => { const transport = createMockTransport('pty-replaced-codex') transportFactoryQueue.push(transport) vi.useFakeTimers() + // Why: this assertion places the hook update between two active-cadence + // inspections, so pin the coordinator's poll jitter like its unit tests do. + vi.spyOn(Math, 'random').mockReturnValue(0.5) const getForegroundProcess = vi.mocked(window.api.pty.getForegroundProcess) getForegroundProcess.mockResolvedValue('codex') const paneKey = makePaneKey('tab-1', LEAF_1) diff --git a/src/shared/native-chat-slash-commands.test.ts b/src/shared/native-chat-slash-commands.test.ts index bd9f7984551..740275bdcef 100644 --- a/src/shared/native-chat-slash-commands.test.ts +++ b/src/shared/native-chat-slash-commands.test.ts @@ -15,6 +15,26 @@ describe('getAgentSlashCommands', () => { expect(names).toContain('diff') }) + it('includes current Codex TUI commands and aliases in presentation order', () => { + const names = getAgentSlashCommands('codex').map((command) => command.name) + for (const expected of [ + 'setup-default-sandbox', + 'sandbox-add-read-dir', + 'btw', + 'debug-config', + 'apps', + 'quit', + 'pet', + 'clean' + ]) { + expect(names).toContain(expected) + } + expect(names.indexOf('vim')).toBeLessThan(names.indexOf('setup-default-sandbox')) + expect(names.indexOf('side')).toBeLessThan(names.indexOf('btw')) + expect(names.indexOf('mcp')).toBeLessThan(names.indexOf('apps')) + expect(names.indexOf('quit')).toBeLessThan(names.indexOf('exit')) + }) + it('returns Claude commands for claude (no Codex-only /model)', () => { const names = getAgentSlashCommands('claude').map((c) => c.name) expect(names).toContain('clear') diff --git a/src/shared/native-chat-slash-commands.ts b/src/shared/native-chat-slash-commands.ts index 949567134b0..006917e68cf 100644 --- a/src/shared/native-chat-slash-commands.ts +++ b/src/shared/native-chat-slash-commands.ts @@ -30,12 +30,19 @@ const CLAUDE_COMMANDS: readonly SlashCommandSuggestion[] = [ { name: 'help', description: 'Show available commands' } ] +// Why: order mirrors Codex's TUI presentation order rather than alphabetical +// order, so Native Chat suggestions remain familiar. const CODEX_COMMANDS: readonly SlashCommandSuggestion[] = [ { name: 'model', description: 'Choose the model and reasoning effort' }, { name: 'ide', description: 'Include IDE context' }, { name: 'permissions', description: 'Choose what Codex is allowed to do' }, { name: 'keymap', description: 'Remap TUI shortcuts' }, { name: 'vim', description: 'Toggle Vim mode' }, + { name: 'setup-default-sandbox', description: 'Set up the elevated agent sandbox' }, + { + name: 'sandbox-add-read-dir', + description: 'Let the sandbox read another directory (Windows)' + }, { name: 'experimental', description: 'Toggle experimental features' }, { name: 'approve', description: 'Approve one auto-review retry' }, { name: 'memories', description: 'Configure memory use' }, @@ -56,23 +63,29 @@ const CODEX_COMMANDS: readonly SlashCommandSuggestion[] = [ { name: 'goal', description: 'Set or view the goal' }, { name: 'agent', description: 'Switch the active agent thread' }, { name: 'side', description: 'Start a side conversation' }, + { name: 'btw', description: 'Start a side conversation (alias of /side)' }, { name: 'copy', description: 'Copy the last response as markdown' }, { name: 'raw', description: 'Toggle raw scrollback mode' }, { name: 'diff', description: 'Show the working diff' }, { name: 'mention', description: 'Mention a file' }, { name: 'status', description: 'Show session configuration and usage' }, { name: 'usage', description: 'View account usage' }, + { name: 'debug-config', description: 'Show config layers and requirement sources' }, { name: 'title', description: 'Configure the terminal title' }, { name: 'statusline', description: 'Configure the status line' }, { name: 'theme', description: 'Choose a syntax highlighting theme' }, { name: 'pets', description: 'Choose or hide the terminal pet' }, + { name: 'pet', description: 'Alias for /pets' }, { name: 'mcp', description: 'List configured MCP tools' }, + { name: 'apps', description: 'Manage apps' }, { name: 'plugins', description: 'Browse plugins' }, { name: 'logout', description: 'Log out of Codex' }, + { name: 'quit', description: 'Exit Codex' }, { name: 'exit', description: 'Exit Codex' }, { name: 'feedback', description: 'Send logs to maintainers' }, { name: 'ps', description: 'List background terminals' }, { name: 'stop', description: 'Stop all background terminals' }, + { name: 'clean', description: 'Alias for /stop' }, { name: 'clear', description: 'Clear the terminal and start a new chat' }, { name: 'personality', description: 'Choose a communication style' }, { name: 'subagents', description: 'Switch the active agent thread' }