diff --git a/src/main/ai-vault/scanners/codex/codex-session-root-dedup.ts b/src/main/ai-vault/scanners/codex/codex-session-root-dedup.ts index 4cfe85d9b78..1003eacbe0b 100644 --- a/src/main/ai-vault/scanners/codex/codex-session-root-dedup.ts +++ b/src/main/ai-vault/scanners/codex/codex-session-root-dedup.ts @@ -8,13 +8,24 @@ import { sessionSortTime } from '../../session-scanner-accumulator' // Bridging and backfill expose one rollout through multiple roots; list only its canonical alias (#7521). // Rollout names narrow candidates; inode or parsed session ID must still prove identity. -const CODEX_ROLLOUT_FILE_NAME_PATTERN = /^rollout-.+\.jsonl$/ +const CODEX_ROLLOUT_FILE_NAME_PATTERN = /^rollout-.+\.jsonl(?:\.zst)?$/ // Hosts can scan foreign-platform paths, so handle both separators independently of node:path. function lastPathSegment(filePath: string): string { return filePath.split(/[\\/]/).at(-1) ?? '' } +// Codex archives idle rollouts as .jsonl.zst and restores the .jsonl on resume, so both names alias one rollout. +function codexRolloutAliasName(fileName: string): string { + return fileName.endsWith('.zst') ? fileName.slice(0, -'.zst'.length) : fileName +} + +/** Returns null when both or neither alias is archived; otherwise whether `filePath` is the live one. */ +function codexLiveAliasBeats(filePath: string, bestPath: string): boolean | null { + const archived = filePath.endsWith('.zst') + return archived === bestPath.endsWith('.zst') ? null : !archived +} + export function codexRolloutHardlinkIdentity(file: { dev?: number ino?: number @@ -133,7 +144,7 @@ export async function dedupeCodexRolloutCopyAliases( if (!CODEX_ROLLOUT_FILE_NAME_PATTERN.test(fileName)) { continue } - const key = `${sessionFileExecutionNamespace(filePath)}\0${fileName}` + const key = `${sessionFileExecutionNamespace(filePath)}\0${codexRolloutAliasName(fileName)}` const group = groups.get(key) if (group) { group.push(candidate) @@ -173,7 +184,12 @@ export async function dedupeCodexRolloutCopyAliases( const filePath = accessors.getFilePath(candidate) const rank = codexSessionRootRank(accessors.getCodexHome(candidate)) const best = bestById.get(id) - if (!best || rank < best.rank || (rank === best.rank && filePath < best.filePath)) { + if ( + !best || + rank < best.rank || + (rank === best.rank && + (codexLiveAliasBeats(filePath, best.filePath) ?? filePath < best.filePath)) + ) { bestById.set(id, { candidate, rank, filePath }) } } @@ -195,7 +211,7 @@ export function codexSessionAliasKey(session: AiVaultSession): string | null { if (!CODEX_ROLLOUT_FILE_NAME_PATTERN.test(fileName)) { return null } - return `${session.executionHostId}\0${sessionFileExecutionNamespace(session.filePath)}\0${session.sessionId}\0${fileName}` + return `${session.executionHostId}\0${sessionFileExecutionNamespace(session.filePath)}\0${session.sessionId}\0${codexRolloutAliasName(fileName)}` } export function codexSessionAliasBeats(candidate: AiVaultSession, best: AiVaultSession): boolean { @@ -204,6 +220,10 @@ export function codexSessionAliasBeats(candidate: AiVaultSession, best: AiVaultS if (candidateRank !== bestRank) { return candidateRank < bestRank } + const live = codexLiveAliasBeats(candidate.filePath, best.filePath) + if (live !== null) { + return live + } const candidateTime = sessionSortTime(candidate) const bestTime = sessionSortTime(best) if (candidateTime !== bestTime) { diff --git a/src/main/ai-vault/scanners/codex/session-scanner-codex-fast-path.test.ts b/src/main/ai-vault/scanners/codex/session-scanner-codex-fast-path.test.ts index b2e44506531..ba065c3b55f 100644 --- a/src/main/ai-vault/scanners/codex/session-scanner-codex-fast-path.test.ts +++ b/src/main/ai-vault/scanners/codex/session-scanner-codex-fast-path.test.ts @@ -1,6 +1,7 @@ -import { mkdir, mkdtemp, rm, stat, writeFile } from 'node:fs/promises' +import { mkdir, mkdtemp, readFile, rm, stat, writeFile } from 'node:fs/promises' import { tmpdir } from 'node:os' import { dirname, join } from 'node:path' +import { zstdCompressSync } from 'node:zlib' import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' import { parseCodexSessionFile } from './session-scanner-codex-parser' import { @@ -145,4 +146,45 @@ describe('Codex resumable parser fast path', () => { expect(session).toBeNull() expect(stats.bytesRead).toBeLessThan((file.sizeBytes ?? 0) / 2) }) + + it('indexes a Codex rollout archived as .jsonl.zst (#40312)', async () => { + const lines = [ + record( + 'session_meta', + { id: 'archived-session', cwd: '/repo/archived' }, + '2026-08-01T10:00:00.000Z' + ), + record( + 'response_item', + { type: 'message', role: 'user', content: 'archived question' }, + '2026-08-01T10:00:05.000Z' + ), + record( + 'response_item', + { type: 'message', role: 'assistant', content: 'archived answer' }, + '2026-08-01T10:00:10.000Z' + ) + ] + const plain = await writeTranscript(lines, 'rollout-2026-08-01T10-00-00-archived.jsonl') + const file: FileWithMtime = { + ...plain, + path: `${plain.path}.zst`, + sizeBytes: undefined + } + await writeFile(file.path, zstdCompressSync(await readFile(plain.path))) + const fileStat = await stat(file.path) + file.mtimeMs = fileStat.mtimeMs + file.sizeBytes = fileStat.size + + const session = await parseAgentSessionFileCached( + { agent: 'codex', file, codexHome: null }, + process.platform + ) + + expect(session).toMatchObject({ + sessionId: 'archived-session', + messageCount: 2, + updatedAt: '2026-08-01T10:00:10.000Z' + }) + }) }) diff --git a/src/main/ai-vault/scanners/codex/session-scanner-codex-parser.ts b/src/main/ai-vault/scanners/codex/session-scanner-codex-parser.ts index 246b642f591..43778625ff7 100644 --- a/src/main/ai-vault/scanners/codex/session-scanner-codex-parser.ts +++ b/src/main/ai-vault/scanners/codex/session-scanner-codex-parser.ts @@ -1,5 +1,9 @@ import { waitForAiVaultScan } from '../../ai-vault-scan-cancellation' import { openTranscriptReadStream } from '../../../native-chat/wsl-transcript-fs-access' +import { + isZstdTranscriptPath, + zstdTranscriptStream +} from '../../../native-chat/transcript-zstd-decompression' import { createInterface } from 'node:readline' import type { AiVaultSession, @@ -53,19 +57,25 @@ export async function parseCodexSessionFile( executionHostId?: ExecutionHostId, messages?: TranscriptMessageSink ): Promise { - const lines = createInterface({ - input: openTranscriptReadStream(file.path, { encoding: 'utf-8' }, 'scan'), - crlfDelay: Infinity - }) - return parseCodexSessionLines({ - file, - lines, - platform, - codexHome, - executionHostId, - messages, - titleReader: (sessionId) => readCodexSessionIndexTitle(file.path, codexHome, sessionId) - }) + // readline decodes UTF-8 across chunk boundaries, so both streams stay binary. + const raw = openTranscriptReadStream(file.path, {}, 'scan') + const input = isZstdTranscriptPath(file.path) ? zstdTranscriptStream(raw) : raw + const lines = createInterface({ input, crlfDelay: Infinity }) + try { + return await parseCodexSessionLines({ + file, + lines, + platform, + codexHome, + executionHostId, + messages, + titleReader: (sessionId) => readCodexSessionIndexTitle(file.path, codexHome, sessionId) + }) + } finally { + // readline.close() leaves its input open; an abandoned zstd decoder would keep inflating. + lines.close() + input.destroy() + } } type CodexSessionParseState = { accumulator: SessionAccumulator diff --git a/src/main/ai-vault/session-file-discovery.test.ts b/src/main/ai-vault/session-file-discovery.test.ts new file mode 100644 index 00000000000..c2f842f78c0 --- /dev/null +++ b/src/main/ai-vault/session-file-discovery.test.ts @@ -0,0 +1,27 @@ +import { describe, expect, it } from 'vitest' +import { AI_VAULT_AGENT_SOURCES } from './session-scanner-agent-sources' +import { isDiscoverableSessionFile } from './session-file-discovery' + +const codex = AI_VAULT_AGENT_SOURCES.codex +if (!codex) { + throw new Error('codex agent source missing') +} +const rootDir = '/home/u/.codex/sessions' + +describe('Codex session file discovery', () => { + it('admits rollout-named compressed archives and plain rollouts', () => { + expect( + isDiscoverableSessionFile(codex, rootDir, `${rootDir}/2026/08/20/rollout-2026-a.jsonl.zst`) + ).toBe(true) + expect( + isDiscoverableSessionFile(codex, rootDir, `${rootDir}/2026/08/20/rollout-2026-a.jsonl`) + ).toBe(true) + }) + + it('rejects compressed files that are not rollout archives', () => { + expect(isDiscoverableSessionFile(codex, rootDir, `${rootDir}/2026/08/20/notes.zst`)).toBe(false) + expect(isDiscoverableSessionFile(codex, rootDir, `${rootDir}/2026/08/20/backup.tar.zst`)).toBe( + false + ) + }) +}) diff --git a/src/main/ai-vault/session-scanner-agent-sources.ts b/src/main/ai-vault/session-scanner-agent-sources.ts index 3395c430251..4db2ab7c288 100644 --- a/src/main/ai-vault/session-scanner-agent-sources.ts +++ b/src/main/ai-vault/session-scanner-agent-sources.ts @@ -124,7 +124,11 @@ export const AI_VAULT_AGENT_SOURCES: AiVaultAgentSourceTable = { ...wslHomeSessionDirs('codex', wslHomeDirs), ...(options.additionalCodexSessionsDirs ?? []) ]), - extensions: ['.jsonl'] + // Codex archives rollouts idle 7+ days as .jsonl.zst (#40312); only rollout-named archives. + extensions: ['.jsonl', '.zst'], + filePredicate: (filePath) => + !filePath.toLowerCase().endsWith('.zst') || + /^rollout-.+\.jsonl\.zst$/.test(basename(filePath)) }, gemini: { rootDirs: (options, wslHomeDirs) => diff --git a/src/main/ai-vault/session-scanner-jsonl-reader.ts b/src/main/ai-vault/session-scanner-jsonl-reader.ts index 6af54b2fce4..cedcbe72037 100644 --- a/src/main/ai-vault/session-scanner-jsonl-reader.ts +++ b/src/main/ai-vault/session-scanner-jsonl-reader.ts @@ -1,6 +1,10 @@ import { throwIfAiVaultScanCancelled, waitForAiVaultScan } from './ai-vault-scan-cancellation' import type { Hash } from 'node:crypto' import { openTranscriptReadStream } from '../native-chat/wsl-transcript-fs-access' +import { + isZstdTranscriptPath, + zstdTranscriptStream +} from '../native-chat/transcript-zstd-decompression' import { MAX_SESSION_TRANSCRIPT_RECORD_BYTES, type SkippedTranscriptRecord @@ -60,7 +64,18 @@ export async function consumeCompleteJsonlLines(args: { let discarding = false const skippedRecords: SkippedTranscriptRecord[] = [] - const stream = openTranscriptReadStream(args.path, { start: args.start }, 'scan', args.signal) + // Compressed transcripts are immutable archives; every read starts from zero and decodes. + const compressed = isZstdTranscriptPath(args.path) + if (compressed && args.start !== 0) { + throw new Error(`compressed transcript cannot resume at byte ${args.start}: ${args.path}`) + } + const rawStream = openTranscriptReadStream( + args.path, + { start: compressed ? 0 : args.start }, + 'scan', + args.signal + ) + const stream = compressed ? zstdTranscriptStream(rawStream) : rawStream let chunkStart = args.start try { for await (const chunk of stream as AsyncIterable) { diff --git a/src/main/ai-vault/session-scanner-scope-walk.test.ts b/src/main/ai-vault/session-scanner-scope-walk.test.ts index 2b695cdf609..09e0403e631 100644 --- a/src/main/ai-vault/session-scanner-scope-walk.test.ts +++ b/src/main/ai-vault/session-scanner-scope-walk.test.ts @@ -1,6 +1,7 @@ -import { link, mkdtemp, mkdir, rm, stat, utimes, writeFile } from 'node:fs/promises' +import { link, mkdtemp, mkdir, readFile, rm, stat, utimes, 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, vi } from 'vitest' import { truncateAiVaultListResult } from '../../shared/ai-vault/scanning/ai-vault-session-depth' import type { AiVaultListResult } from '../../shared/ai-vault/contracts/ai-vault-types' @@ -139,6 +140,55 @@ async function writeRollout(sessionsDir: string, id: string, cwd: string, day: s } describe('scanAiVaultSessions — scope before the recency cap', () => { + it.each([undefined, ['/repo']])( + 'lists a rollout once, as its plain file, beside its .zst archive (scope %j)', + async (scopePaths) => { + const root = await mkdtemp(join(tmpdir(), 'orca-ai-vault-zst-twin-')) + tempRoots.push(root) + const roots = isolatedScanRoots(root) + const plain = await writeRollout(roots.codexSessionsDir, 'twin', '/repo', '11') + await writeFile(`${plain}.zst`, zstdCompressSync(await readFile(plain))) + const archived = await writeRollout(roots.codexSessionsDir, 'archived-only', '/repo', '13') + await writeFile(`${archived}.zst`, zstdCompressSync(await readFile(archived))) + await rm(archived) + + const result = await scanAiVaultSessions({ ...roots, platform: 'darwin', scopePaths }) + + expect( + result.sessions + .filter((session) => session.agent === 'codex') + .map((session) => [session.sessionId, session.filePath]) + ).toEqual( + expect.arrayContaining([ + ['twin', plain], + ['archived-only', `${archived}.zst`] + ]) + ) + expect(result.sessions.filter((session) => session.agent === 'codex')).toHaveLength(2) + } + ) + + it('lists the live rollout when another account home holds only its archive', async () => { + const root = await mkdtemp(join(tmpdir(), 'orca-ai-vault-zst-homes-')) + tempRoots.push(root) + const homeA = join(root, 'codex-accounts', 'a', 'home', 'sessions') + const homeB = join(root, 'codex-accounts', 'b', 'home', 'sessions') + const archived = await writeRollout(homeA, 'moved', '/repo', '11') + await writeFile(`${archived}.zst`, zstdCompressSync(await readFile(archived))) + await rm(archived) + const live = await writeRollout(homeB, 'moved', '/repo', '11') + + const result = await scanAiVaultSessions({ + ...isolatedScanRoots(root), + additionalCodexSessionsDirs: [homeA, homeB], + platform: 'darwin' + }) + + expect( + result.sessions.filter((session) => session.agent === 'codex').map((s) => s.filePath) + ).toEqual([live]) + }) + it('lists an older in-scope Codex session once across hardlinked homes', async () => { const root = await mkdtemp(join(tmpdir(), 'orca-ai-vault-scope-walk-')) tempRoots.push(root) diff --git a/src/main/ai-vault/session-scope-cwd.ts b/src/main/ai-vault/session-scope-cwd.ts index 92c4fbc33b3..612362d380e 100644 --- a/src/main/ai-vault/session-scope-cwd.ts +++ b/src/main/ai-vault/session-scope-cwd.ts @@ -1,5 +1,14 @@ import type { AiVaultAgent } from '../../shared/ai-vault/contracts/ai-vault-types' -import { readTranscriptSlice, wslGatedReadFile } from '../native-chat/wsl-transcript-fs-access' +import { + openTranscriptReadStream, + readTranscriptSlice, + wslGatedReadFile +} from '../native-chat/wsl-transcript-fs-access' +import { + isZstdTranscriptPath, + readZstdTranscriptSlice, + zstdTranscriptStream +} from '../native-chat/transcript-zstd-decompression' import { CLAUDE_CWD_BUCKET_LAYOUT, CODEBUDDY_CWD_BUCKET_LAYOUT, @@ -182,7 +191,12 @@ export async function readTranscriptHeadCwd( : [first]) { let head: Buffer try { - head = await readTranscriptSlice(file.path, 0, length, 'scan', signal) + head = isZstdTranscriptPath(file.path) + ? await readZstdTranscriptSlice( + zstdTranscriptStream(openTranscriptReadStream(file.path, {}, 'scan', signal)), + length + ) + : await readTranscriptSlice(file.path, 0, length, 'scan', signal) } catch (error) { if (signal?.aborted) { throw error diff --git a/src/main/ai-vault/transcripts/session-transcript-consumers.test.ts b/src/main/ai-vault/transcripts/session-transcript-consumers.test.ts index 693fdfc7f5c..c9b040071cf 100644 --- a/src/main/ai-vault/transcripts/session-transcript-consumers.test.ts +++ b/src/main/ai-vault/transcripts/session-transcript-consumers.test.ts @@ -1,6 +1,7 @@ import { appendFile, mkdir, mkdtemp, rm, stat, truncate, utimes, writeFile } from 'node:fs/promises' import { tmpdir } from 'node:os' import { join } from 'node:path' +import { zstdCompressSync } from 'node:zlib' import { afterEach, expect, it, vi } from 'vitest' import { isAiVaultScanCancelledError } from '../../../shared/ai-vault/contracts/ai-vault-types' import { scanAiVaultSessions } from '../session-scanner' @@ -307,3 +308,39 @@ it('reports the transcript size, not the cache key, as a whole-file read offset' const read = consumer.reads.find((entry) => entry.start.candidate.agent === 'cline') expect(read?.outcome?.byteOffset).toBe((await stat(metadataPath)).size) }) + +it('covers an archived Codex rollout through its compressed size, never resuming into it', async () => { + const root = await mkdtemp(join(tmpdir(), 'orca-transcript-consumers-zst-')) + tempRoots.push(root) + const path = join(root, 'rollout-2026-05-01T10-00-00-archived.jsonl.zst') + const records = [ + { type: 'session_meta', payload: { id: 'archived', cwd: '/repo' } }, + { + timestamp: '2026-05-01T10:00:01.000Z', + type: 'response_item', + payload: { type: 'message', role: 'user', content: [{ type: 'text', text: 'old ask' }] } + } + ] + await writeFile(path, zstdCompressSync(`${jsonLines(records)}\n`)) + const stats = await stat(path) + const candidate: SessionFileCandidate = { + agent: 'codex', + file: { + path, + mtimeMs: stats.mtimeMs, + modifiedAt: stats.mtime.toISOString(), + sizeBytes: stats.size + }, + codexHome: null + } + const consumer = recordingConsumer() + + await parseAgentSessionFileCached(candidate, 'darwin', undefined, 'any') + await parseAgentSessionFileCached(candidate, 'darwin', undefined, 'any') + + expect(consumer.reads.map((read) => [read.start.mode, read.outcome?.byteOffset])).toEqual([ + ['replace', stats.size], + ['replace', stats.size] + ]) + expect(textsFor(consumer.reads, 'codex')).toEqual(['user:old ask', 'user:old ask']) +}) diff --git a/src/main/ai-vault/transcripts/session-transcript-reader.ts b/src/main/ai-vault/transcripts/session-transcript-reader.ts index c61449a5104..6f3b8e77936 100644 --- a/src/main/ai-vault/transcripts/session-transcript-reader.ts +++ b/src/main/ai-vault/transcripts/session-transcript-reader.ts @@ -4,6 +4,7 @@ import { readTranscriptSlice } from '../../native-chat/wsl-transcript-fs-access' import type { AiVaultSession } from '../../../shared/ai-vault/contracts/ai-vault-types' import { parseAgentSessionFile } from '../session-scanner-agent-parser' import { consumeCompleteJsonlLines } from '../session-scanner-jsonl-reader' +import { isZstdTranscriptPath } from '../../native-chat/transcript-zstd-decompression' import type { ResumableSessionParseState, SessionFileCandidate } from '../session-scanner-types' import { invalidateSessionParseCacheEntry, @@ -55,7 +56,10 @@ export async function readResumableTranscript( ): Promise { const { file } = args.candidate const resume = args.resume + // Codex .jsonl.zst archives never grow; decoded offsets cannot address their compressed bytes. + const archived = isZstdTranscriptPath(file.path) const canResume = + !archived && resume !== null && typeof file.sizeBytes === 'number' && file.sizeBytes >= resume.byteOffset && @@ -83,7 +87,7 @@ export async function readResumableTranscript( // Hash only for consumers: they compare prefixes after a restart drops this resume point. const prefixHash = canResume ? (resume.prefixHash?.copy() ?? null) - : hasTranscriptConsumers() + : !archived && hasTranscriptConsumers() ? createHash('sha256') : null const prefixOffsets = @@ -98,6 +102,8 @@ export async function readResumableTranscript( prefixOffsets }) try { + // Archives are covered through their compressed size, which search cursors compare to stat size. + const coveredThrough = archived ? (file.sizeBytes ?? 0) : null const readResult = await consumeCompleteJsonlLines({ path: file.path, start: startOffset, @@ -135,7 +141,7 @@ export async function readResumableTranscript( throwIfAiVaultScanCancelled(args.signal) channel.finishRead({ session: consumerSession, - byteOffset: readResult.consumedThrough, + byteOffset: coveredThrough ?? readResult.consumedThrough, incomplete: false, prefixHash: readResult.prefixHash?.copy().digest('hex') }) @@ -143,7 +149,7 @@ export async function readResumableTranscript( session, resume: { state, - byteOffset: readResult.consumedThrough, + byteOffset: coveredThrough ?? readResult.consumedThrough, mtimeMs: file.mtimeMs, sizeBytes: file.sizeBytes, channel, diff --git a/src/main/codex/history/rollouts/codex-rollout-session-meta.test.ts b/src/main/codex/history/rollouts/codex-rollout-session-meta.test.ts index 799c5123275..e24e4c72115 100644 --- a/src/main/codex/history/rollouts/codex-rollout-session-meta.test.ts +++ b/src/main/codex/history/rollouts/codex-rollout-session-meta.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, vi } from 'vitest' import type * as WslFsAccessModule from '../../../native-chat/wsl-transcript-fs-access' import { readCodexRolloutSessionMetaId } from './codex-rollout-session-meta' @@ -44,6 +45,22 @@ describe('readCodexRolloutSessionMetaId', () => { await expect(readCodexRolloutSessionMetaId(malformed)).resolves.toBeNull() }) + it('reads the session id from a compressed rollout', async () => { + const root = await mkdtemp(join(tmpdir(), 'orca-codex-session-meta-')) + tempRoots.push(root) + const rollout = join(root, 'rollout.jsonl.zst') + await writeFile( + rollout, + zstdCompressSync( + Buffer.from( + `${JSON.stringify({ type: 'session_meta', payload: { id: 'archived-id' } })}\nignored\n` + ) + ) + ) + + await expect(readCodexRolloutSessionMetaId(rollout)).resolves.toBe('archived-id') + }) + it('surfaces an aborted read instead of reporting an unprovable rollout', async () => { const root = await mkdtemp(join(tmpdir(), 'orca-codex-session-meta-')) tempRoots.push(root) diff --git a/src/main/codex/history/rollouts/codex-rollout-session-meta.ts b/src/main/codex/history/rollouts/codex-rollout-session-meta.ts index 0cffe593b2a..74317f0f5de 100644 --- a/src/main/codex/history/rollouts/codex-rollout-session-meta.ts +++ b/src/main/codex/history/rollouts/codex-rollout-session-meta.ts @@ -1,4 +1,12 @@ -import { readTranscriptSlice } from '../../../native-chat/wsl-transcript-fs-access' +import { + openTranscriptReadStream, + readTranscriptSlice +} from '../../../native-chat/wsl-transcript-fs-access' +import { + isZstdTranscriptPath, + readZstdTranscriptSlice, + zstdTranscriptStream +} from '../../../native-chat/transcript-zstd-decompression' import type { WslTranscriptFsTaskPriority } from '../../../native-chat/wsl-transcript-fs-gate' const ROLLOUT_READ_LIMIT = 64 * 1024 @@ -12,7 +20,12 @@ export async function readCodexRolloutSessionMetaId( let head: Buffer try { // The transcript gate bounds stalled WSL reads; missing or pruned rollouts cannot abort a scan (#15453). - head = await readTranscriptSlice(filePath, 0, ROLLOUT_READ_LIMIT, priority, signal) + head = isZstdTranscriptPath(filePath) + ? await readZstdTranscriptSlice( + zstdTranscriptStream(openTranscriptReadStream(filePath, {}, priority, signal)), + ROLLOUT_READ_LIMIT + ) + : await readTranscriptSlice(filePath, 0, ROLLOUT_READ_LIMIT, priority, signal) } catch (error) { // Preserve cancellation instead of treating it as an unreadable rollout. if (signal?.aborted) { diff --git a/src/main/native-chat/transcript-zstd-decompression.test.ts b/src/main/native-chat/transcript-zstd-decompression.test.ts new file mode 100644 index 00000000000..8242eb6112e --- /dev/null +++ b/src/main/native-chat/transcript-zstd-decompression.test.ts @@ -0,0 +1,24 @@ +import { describe, expect, it } from 'vitest' +import { Readable } from 'node:stream' +import { randomBytes } from 'node:crypto' +import { zstdCompressSync } from 'node:zlib' +import { readZstdTranscriptSlice, zstdTranscriptStream } from './transcript-zstd-decompression' + +describe('zstdTranscriptStream', () => { + it('closes the raw stream when the decoded stream stops before EOF', async () => { + const compressed = zstdCompressSync(randomBytes(1024 * 1024)) + async function* rawChunks(): AsyncGenerator { + for (let offset = 0; offset < compressed.length; offset += 16 * 1024) { + yield compressed.subarray(offset, offset + 16 * 1024) + } + } + const raw = Readable.from(rawChunks()) + const decoded = zstdTranscriptStream(raw) + + const head = await readZstdTranscriptSlice(decoded, 64 * 1024) + + expect(head.length).toBe(64 * 1024) + await new Promise((resolve) => setImmediate(resolve)) + expect(raw.destroyed).toBe(true) + }) +}) diff --git a/src/main/native-chat/transcript-zstd-decompression.ts b/src/main/native-chat/transcript-zstd-decompression.ts new file mode 100644 index 00000000000..4e34d42f9de --- /dev/null +++ b/src/main/native-chat/transcript-zstd-decompression.ts @@ -0,0 +1,29 @@ +import { pipeline, type Readable } from 'node:stream' +import { createZstdDecompress } from 'node:zlib' + +export function isZstdTranscriptPath(path: string): boolean { + return path.endsWith('.zst') +} + +/** Pipes raw compressed bytes through the zstd decoder; errors on the raw stream surface on the output. */ +export function zstdTranscriptStream(raw: Readable): Readable { + const decoder = createZstdDecompress() + // pipeline links lifetimes so an early close of the decoded stream releases the raw stream's fd. + pipeline(raw, decoder, () => undefined) + return decoder +} + +/** Collects at most `length` decompressed bytes; compressed transcripts are small enough to keep in memory. */ +export async function readZstdTranscriptSlice(stream: Readable, length: number): Promise { + const parts: Buffer[] = [] + let total = 0 + for await (const chunk of stream) { + const buffer = Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk) + parts.push(buffer) + total += buffer.length + if (total >= length) { + break + } + } + return Buffer.concat(parts).subarray(0, Math.min(total, length)) +}