diff --git a/dist/paths.d.ts b/dist/paths.d.ts index 6cee3813..470188e2 100644 --- a/dist/paths.d.ts +++ b/dist/paths.d.ts @@ -46,8 +46,8 @@ export type ConversationSourceHarness = 'claude' | 'codex' | 'cursor' | 'opencod /** * Get all directories where supported harnesses store conversation files. * Checks Claude Code legacy (projects/) and current (transcripts/) locations, - * Codex sessions, Cursor agent transcripts (live and legacy exports), and - * generated opencode transcripts. + * Codex live and archived sessions, Cursor agent transcripts (live and legacy + * exports), and generated opencode transcripts. * Returns only directories that exist. */ export declare function getConversationSourceDirs(only?: ConversationSourceHarness[]): string[]; @@ -59,8 +59,20 @@ export declare function getConversationSourceDirs(only?: ConversationSourceHarne * the set, at any depth. Top-level project skipping at the caller is the * usual case; this parameter handles nested directories like `subagents/` * inside session UUIDs (#80). + * + * `includeCompressed` also returns Codex's zstd-compressed `.jsonl.zst` + * rollouts, for callers that can read them. + */ +export declare function findJsonlFiles(dir: string, excludedDirNames?: ReadonlySet, options?: { + includeCompressed?: boolean; +}): string[]; +/** + * Codex moves a finished rollout from `sessions/YYYY/MM/DD/` to a flat + * `archived_sessions/` directory, usually compressing it. Map such a flat + * rollout back to the dated archive path it had before it moved, so a moved + * session keeps one archive identity. Returns null for any other file name. */ -export declare function findJsonlFiles(dir: string, excludedDirNames?: ReadonlySet): string[]; +export declare function archivedRolloutRelativePath(fileName: string): string | null; /** * statSync that follows symlinks but returns null instead of throwing when * the entry cannot be stat'ed — a dangling symlink (e.g. left behind by a diff --git a/dist/paths.js b/dist/paths.js index e80076b1..9b2301f3 100644 --- a/dist/paths.js +++ b/dist/paths.js @@ -80,8 +80,8 @@ export function getOpencodeTranscriptDir() { /** * Get all directories where supported harnesses store conversation files. * Checks Claude Code legacy (projects/) and current (transcripts/) locations, - * Codex sessions, Cursor agent transcripts (live and legacy exports), and - * generated opencode transcripts. + * Codex live and archived sessions, Cursor agent transcripts (live and legacy + * exports), and generated opencode transcripts. * Returns only directories that exist. */ export function getConversationSourceDirs(only) { @@ -96,6 +96,7 @@ export function getConversationSourceDirs(only) { { harness: 'claude', dir: path.join(claudeDir, 'projects') }, { harness: 'claude', dir: path.join(claudeDir, 'transcripts') }, { harness: 'codex', dir: path.join(codexDir, 'sessions') }, + { harness: 'codex', dir: path.join(codexDir, 'archived_sessions') }, { harness: 'cursor', dir: path.join(cursorDir, 'projects') }, { harness: 'cursor', dir: getCursorLegacyExportDir() }, { harness: 'opencode', dir: getOpencodeTranscriptDir() }, @@ -114,20 +115,23 @@ export function getConversationSourceDirs(only) { * the set, at any depth. Top-level project skipping at the caller is the * usual case; this parameter handles nested directories like `subagents/` * inside session UUIDs (#80). + * + * `includeCompressed` also returns Codex's zstd-compressed `.jsonl.zst` + * rollouts, for callers that can read them. */ -export function findJsonlFiles(dir, excludedDirNames) { +export function findJsonlFiles(dir, excludedDirNames, options = {}) { const results = []; try { const entries = fs.readdirSync(dir, { withFileTypes: true }); for (const entry of entries) { - if (entry.isFile() && entry.name.endsWith('.jsonl')) { + if (entry.isFile() && (entry.name.endsWith('.jsonl') || options.includeCompressed && entry.name.endsWith('.jsonl.zst'))) { results.push(entry.name); } else if (entry.isDirectory()) { if (excludedDirNames?.has(entry.name)) continue; const subDir = path.join(dir, entry.name); - for (const f of findJsonlFiles(subDir, excludedDirNames)) { + for (const f of findJsonlFiles(subDir, excludedDirNames, options)) { results.push(path.join(entry.name, f)); } } @@ -138,6 +142,18 @@ export function findJsonlFiles(dir, excludedDirNames) { } return results; } +/** + * Codex moves a finished rollout from `sessions/YYYY/MM/DD/` to a flat + * `archived_sessions/` directory, usually compressing it. Map such a flat + * rollout back to the dated archive path it had before it moved, so a moved + * session keeps one archive identity. Returns null for any other file name. + */ +export function archivedRolloutRelativePath(fileName) { + const match = /^rollout-(\d{4})-(\d{2})-(\d{2})T.*\.jsonl(?:\.zst)?$/.exec(fileName); + if (!match) + return null; + return path.join(match[1], match[2], match[3], fileName.replace(/\.zst$/, '')); +} /** * statSync that follows symlinks but returns null instead of throwing when * the entry cannot be stat'ed — a dangling symlink (e.g. left behind by a diff --git a/dist/sync.js b/dist/sync.js index 0629761d..b92442d3 100644 --- a/dist/sync.js +++ b/dist/sync.js @@ -1,10 +1,18 @@ import fs from 'fs'; import path from 'path'; +import readline from 'readline'; +import zlib from 'zlib'; +import { pipeline } from 'stream/promises'; import { StringDecoder } from 'string_decoder'; import { SUMMARIZER_CONTEXT_MARKER } from './constants.js'; -import { getExcludedProjects, findJsonlFiles, statIfExists } from './paths.js'; +import { archivedRolloutRelativePath, getExcludedProjects, findJsonlFiles, statIfExists } from './paths.js'; import { formatErrorSentinel, shouldQueueForSummary } from './summary-sentinel.js'; import { getMaxMessageBytes, isOversizeExchange } from './message-size.js'; +/** Codex compresses archived rollouts with zstd, which node:zlib reads from Node 22.15. */ +const zstdAvailable = typeof zlib.createZstdDecompress === 'function'; +function isCompressedRollout(file) { + return file.endsWith('.jsonl.zst'); +} const EXCLUSION_MARKERS = [ 'DO NOT INDEX THIS CHAT', 'Only use NO_INSIGHTS_FOUND', @@ -59,11 +67,19 @@ export function shouldSkipConversation(filePath) { * every run — one new stub per summary generated. A transcript that has no * messages *yet* (a session that just started) is skipped this run and picked * up on a later sync once it has content, since its mtime keeps advancing. + * Reading streams until the first message line, so a large or compressed + * transcript is never loaded whole. */ -function hasConversationContent(filePath) { +async function hasConversationContent(filePath) { + const input = fs.createReadStream(filePath); + const decompressed = isCompressedRollout(filePath) ? zlib.createZstdDecompress() : undefined; + if (decompressed) { + input.on('error', error => decompressed.destroy(error)); + input.pipe(decompressed); + } + const lines = readline.createInterface({ input: decompressed ?? input, crlfDelay: Infinity }); try { - const content = fs.readFileSync(filePath, 'utf-8'); - for (const line of content.split('\n')) { + for await (const line of lines) { if (!line.trim()) continue; try { @@ -103,6 +119,11 @@ function hasConversationContent(filePath) { // If we can't read the file, let the normal pipeline handle it return true; } + finally { + lines.close(); + decompressed?.destroy(); + input.destroy(); + } } /** * Derive sync options from the process environment. @@ -122,32 +143,78 @@ function hasConversationContent(filePath) { export function buildSyncOptionsFromEnv(env) { return { skipSummaries: env.EPISODIC_MEMORY_SKIP_SUMMARIES === '1' }; } -function copyIfNewer(src, dest) { +/** True when `file` begins with every byte of `prefix`. */ +function startsWithFile(file, prefix) { + const prefixSize = fs.statSync(prefix).size; + if (fs.statSync(file).size < prefixSize) + return false; + const fileFd = fs.openSync(file, 'r'); + const prefixFd = fs.openSync(prefix, 'r'); + try { + const a = Buffer.allocUnsafe(MARKER_SCAN_CHUNK_BYTES); + const b = Buffer.allocUnsafe(MARKER_SCAN_CHUNK_BYTES); + for (let position = 0; position < prefixSize;) { + const want = Math.min(a.length, prefixSize - position); + const read = fs.readSync(prefixFd, b, 0, want, position); + if (read <= 0 || fs.readSync(fileFd, a, 0, read, position) !== read) + return false; + if (!a.subarray(0, read).equals(b.subarray(0, read))) + return false; + position += read; + } + return true; + } + finally { + fs.closeSync(fileFd); + fs.closeSync(prefixFd); + } +} +async function copyIfNewer(src, dest) { // Ensure destination directory exists const destDir = path.dirname(dest); if (!fs.existsSync(destDir)) { fs.mkdirSync(destDir, { recursive: true }); } // Check if destination exists and is up-to-date - if (fs.existsSync(dest)) { + const existing = fs.existsSync(dest); + if (existing) { const srcStat = fs.statSync(src); const destStat = fs.statSync(dest); if (destStat.mtimeMs >= srcStat.mtimeMs) { - return false; // Dest is current, skip + return 'current'; // Dest is current, skip } } // Atomic copy: temp file + rename const tempDest = dest + '.tmp.' + process.pid; - fs.copyFileSync(src, tempDest); - fs.renameSync(tempDest, dest); // Atomic on same filesystem + let outcome = 'copied'; + try { + if (isCompressedRollout(src)) { + await pipeline(fs.createReadStream(src), zlib.createZstdDecompress(), fs.createWriteStream(tempDest)); + // A rollout archived earlier as plain JSONL may have been rewritten by Codex + // before it was compressed. Index rows address the existing archive's lines, + // so only content that extends that archive replaces it. + if (existing && !startsWithFile(tempDest, dest)) + outcome = 'kept'; + } + else { + fs.copyFileSync(src, tempDest); + } + if (outcome === 'copied') + fs.renameSync(tempDest, dest); // Atomic on same filesystem + } + finally { + if (fs.existsSync(tempDest)) + fs.unlinkSync(tempDest); + } // Preserve source mtime: harnesses without per-message timestamps (Cursor // agent transcripts) fall back to file mtime. Round up to the next whole // millisecond — utimes can't always represent the source's sub-millisecond // precision, and a dest mtime even fractionally older would defeat the - // skip-if-current check above on every subsequent sync. + // skip-if-current check above on every subsequent sync. A kept archive is + // stamped too, so an unchanged rewritten source is not decompressed again. const srcStat = fs.statSync(src); fs.utimesSync(dest, srcStat.atimeMs / 1000, Math.ceil(srcStat.mtimeMs) / 1000); - return true; + return outcome; } export function extractSessionIdFromPath(filePath) { // Extract session ID from Claude filename or Codex rollout filename. @@ -178,6 +245,7 @@ export async function syncConversations(sourceDir, destDir, options = {}) { const projects = fs.readdirSync(sourceDir); const excludedProjects = getExcludedProjects(); const excludedDirSet = new Set(excludedProjects); + const sources = []; for (const project of projects) { if (excludedProjects.includes(project)) { console.log("\nSkipping excluded project: " + project); @@ -185,49 +253,60 @@ export async function syncConversations(sourceDir, destDir, options = {}) { } const projectPath = path.join(sourceDir, project); const stat = statIfExists(projectPath); + if (stat?.isFile()) { + // A finished Codex rollout moved out of its dated directory. + const relative = zstdAvailable || !isCompressedRollout(project) ? archivedRolloutRelativePath(project) : null; + if (relative) + sources.push({ srcFile: projectPath, destFile: path.join(destDir, relative) }); + continue; + } if (!stat?.isDirectory()) continue; - const files = findJsonlFiles(projectPath, excludedDirSet); - for (const file of files) { - const srcFile = path.join(projectPath, file); - const destFile = path.join(destDir, project, file); - try { - // Skip message-less transcripts (summarizer-spawned stubs, sessions - // that haven't produced content yet) before they enter the archive. - if (!hasConversationContent(srcFile)) { - result.skipped++; - continue; - } - const wasCopied = copyIfNewer(srcFile, destFile); - if (wasCopied) { - result.copied++; - filesToIndex.push(destFile); - } - else { - result.skipped++; - } - // Check if this file needs a summary (whether newly copied or existing). - // shouldQueueForSummary skips files that already have a real summary or - // an empty zero-exchange sentinel, and retries stale error sentinels (#96). - if (!options.skipSummaries) { - const summaryPath = destFile.replace('.jsonl', '-summary.txt'); - if (shouldQueueForSummary(summaryPath) && !shouldSkipConversation(destFile)) { - // sessionId enables Claude session-resume summarization; when the - // filename has no UUID to extract (e.g. subagent transcripts named - // agent-.jsonl), queue anyway — summarizeConversation falls - // back to summarizing from the transcript text. - const sessionId = extractSessionIdFromPath(destFile) ?? undefined; - filesToSummarize.push({ path: destFile, sessionId }); - } + for (const file of findJsonlFiles(projectPath, excludedDirSet, { includeCompressed: zstdAvailable })) { + // A compressed rollout is archived as the plain transcript it holds. + sources.push({ srcFile: path.join(projectPath, file), destFile: path.join(destDir, project, file.replace(/\.zst$/, '')) }); + } + } + for (const { srcFile, destFile } of sources) { + try { + // Skip message-less transcripts (summarizer-spawned stubs, sessions + // that haven't produced content yet) before they enter the archive. + if (!(await hasConversationContent(srcFile))) { + result.skipped++; + continue; + } + const outcome = await copyIfNewer(srcFile, destFile); + if (outcome === 'copied') { + result.copied++; + filesToIndex.push(destFile); + } + else { + result.skipped++; + if (outcome === 'kept') { + console.log(` Kept ${destFile}: its source was rewritten and no longer extends the indexed archive`); } } - catch (error) { - result.errors.push({ - file: srcFile, - error: error instanceof Error ? error.message : String(error) - }); + // Check if this file needs a summary (whether newly copied or existing). + // shouldQueueForSummary skips files that already have a real summary or + // an empty zero-exchange sentinel, and retries stale error sentinels (#96). + if (!options.skipSummaries) { + const summaryPath = destFile.replace('.jsonl', '-summary.txt'); + if (shouldQueueForSummary(summaryPath) && !shouldSkipConversation(destFile)) { + // sessionId enables Claude session-resume summarization; when the + // filename has no UUID to extract (e.g. subagent transcripts named + // agent-.jsonl), queue anyway — summarizeConversation falls + // back to summarizing from the transcript text. + const sessionId = extractSessionIdFromPath(destFile) ?? undefined; + filesToSummarize.push({ path: destFile, sessionId }); + } } } + catch (error) { + result.errors.push({ + file: srcFile, + error: error instanceof Error ? error.message : String(error) + }); + } } // Index copied files (unless skipIndex is set) if (!options.skipIndex && filesToIndex.length > 0) { diff --git a/src/paths.ts b/src/paths.ts index 4a08f2fe..331d7478 100644 --- a/src/paths.ts +++ b/src/paths.ts @@ -93,8 +93,8 @@ export type ConversationSourceHarness = 'claude' | 'codex' | 'cursor' | 'opencod /** * Get all directories where supported harnesses store conversation files. * Checks Claude Code legacy (projects/) and current (transcripts/) locations, - * Codex sessions, Cursor agent transcripts (live and legacy exports), and - * generated opencode transcripts. + * Codex live and archived sessions, Cursor agent transcripts (live and legacy + * exports), and generated opencode transcripts. * Returns only directories that exist. */ export function getConversationSourceDirs(only?: ConversationSourceHarness[]): string[] { @@ -109,6 +109,7 @@ export function getConversationSourceDirs(only?: ConversationSourceHarness[]): s { harness: 'claude', dir: path.join(claudeDir, 'projects') }, { harness: 'claude', dir: path.join(claudeDir, 'transcripts') }, { harness: 'codex', dir: path.join(codexDir, 'sessions') }, + { harness: 'codex', dir: path.join(codexDir, 'archived_sessions') }, { harness: 'cursor', dir: path.join(cursorDir, 'projects') }, { harness: 'cursor', dir: getCursorLegacyExportDir() }, { harness: 'opencode', dir: getOpencodeTranscriptDir() }, @@ -128,18 +129,22 @@ export function getConversationSourceDirs(only?: ConversationSourceHarness[]): s * the set, at any depth. Top-level project skipping at the caller is the * usual case; this parameter handles nested directories like `subagents/` * inside session UUIDs (#80). + * + * `includeCompressed` also returns Codex's zstd-compressed `.jsonl.zst` + * rollouts, for callers that can read them. */ -export function findJsonlFiles(dir: string, excludedDirNames?: ReadonlySet): string[] { +export function findJsonlFiles(dir: string, excludedDirNames?: ReadonlySet, + options: { includeCompressed?: boolean } = {}): string[] { const results: string[] = []; try { const entries = fs.readdirSync(dir, { withFileTypes: true }); for (const entry of entries) { - if (entry.isFile() && entry.name.endsWith('.jsonl')) { + if (entry.isFile() && (entry.name.endsWith('.jsonl') || options.includeCompressed && entry.name.endsWith('.jsonl.zst'))) { results.push(entry.name); } else if (entry.isDirectory()) { if (excludedDirNames?.has(entry.name)) continue; const subDir = path.join(dir, entry.name); - for (const f of findJsonlFiles(subDir, excludedDirNames)) { + for (const f of findJsonlFiles(subDir, excludedDirNames, options)) { results.push(path.join(entry.name, f)); } } @@ -150,6 +155,18 @@ export function findJsonlFiles(dir: string, excludedDirNames?: ReadonlySetDO NOT INDEX THIS CHAT', 'Only use NO_INSIGHTS_FOUND', @@ -56,11 +66,19 @@ export function shouldSkipConversation(filePath: string): boolean { * every run — one new stub per summary generated. A transcript that has no * messages *yet* (a session that just started) is skipped this run and picked * up on a later sync once it has content, since its mtime keeps advancing. + * Reading streams until the first message line, so a large or compressed + * transcript is never loaded whole. */ -function hasConversationContent(filePath: string): boolean { +async function hasConversationContent(filePath: string): Promise { + const input = fs.createReadStream(filePath); + const decompressed = isCompressedRollout(filePath) ? zlib.createZstdDecompress() : undefined; + if (decompressed) { + input.on('error', error => decompressed.destroy(error)); + input.pipe(decompressed); + } + const lines = readline.createInterface({ input: decompressed ?? input, crlfDelay: Infinity }); try { - const content = fs.readFileSync(filePath, 'utf-8'); - for (const line of content.split('\n')) { + for await (const line of lines) { if (!line.trim()) continue; try { const parsed = JSON.parse(line); @@ -96,6 +114,10 @@ function hasConversationContent(filePath: string): boolean { } catch { // If we can't read the file, let the normal pipeline handle it return true; + } finally { + lines.close(); + decompressed?.destroy(); + input.destroy(); } } @@ -132,7 +154,32 @@ export function buildSyncOptionsFromEnv(env: NodeJS.ProcessEnv): SyncOptions { return { skipSummaries: env.EPISODIC_MEMORY_SKIP_SUMMARIES === '1' }; } -function copyIfNewer(src: string, dest: string): boolean { +/** True when `file` begins with every byte of `prefix`. */ +function startsWithFile(file: string, prefix: string): boolean { + const prefixSize = fs.statSync(prefix).size; + if (fs.statSync(file).size < prefixSize) return false; + const fileFd = fs.openSync(file, 'r'); + const prefixFd = fs.openSync(prefix, 'r'); + try { + const a = Buffer.allocUnsafe(MARKER_SCAN_CHUNK_BYTES); + const b = Buffer.allocUnsafe(MARKER_SCAN_CHUNK_BYTES); + for (let position = 0; position < prefixSize;) { + const want = Math.min(a.length, prefixSize - position); + const read = fs.readSync(prefixFd, b, 0, want, position); + if (read <= 0 || fs.readSync(fileFd, a, 0, read, position) !== read) return false; + if (!a.subarray(0, read).equals(b.subarray(0, read))) return false; + position += read; + } + return true; + } finally { + fs.closeSync(fileFd); + fs.closeSync(prefixFd); + } +} + +type CopyOutcome = 'copied' | 'current' | 'kept'; + +async function copyIfNewer(src: string, dest: string): Promise { // Ensure destination directory exists const destDir = path.dirname(dest); if (!fs.existsSync(destDir)) { @@ -140,27 +187,42 @@ function copyIfNewer(src: string, dest: string): boolean { } // Check if destination exists and is up-to-date - if (fs.existsSync(dest)) { + const existing = fs.existsSync(dest); + if (existing) { const srcStat = fs.statSync(src); const destStat = fs.statSync(dest); if (destStat.mtimeMs >= srcStat.mtimeMs) { - return false; // Dest is current, skip + return 'current'; // Dest is current, skip } } // Atomic copy: temp file + rename const tempDest = dest + '.tmp.' + process.pid; - fs.copyFileSync(src, tempDest); - fs.renameSync(tempDest, dest); // Atomic on same filesystem + let outcome: CopyOutcome = 'copied'; + try { + if (isCompressedRollout(src)) { + await pipeline(fs.createReadStream(src), zlib.createZstdDecompress(), fs.createWriteStream(tempDest)); + // A rollout archived earlier as plain JSONL may have been rewritten by Codex + // before it was compressed. Index rows address the existing archive's lines, + // so only content that extends that archive replaces it. + if (existing && !startsWithFile(tempDest, dest)) outcome = 'kept'; + } else { + fs.copyFileSync(src, tempDest); + } + if (outcome === 'copied') fs.renameSync(tempDest, dest); // Atomic on same filesystem + } finally { + if (fs.existsSync(tempDest)) fs.unlinkSync(tempDest); + } // Preserve source mtime: harnesses without per-message timestamps (Cursor // agent transcripts) fall back to file mtime. Round up to the next whole // millisecond — utimes can't always represent the source's sub-millisecond // precision, and a dest mtime even fractionally older would defeat the - // skip-if-current check above on every subsequent sync. + // skip-if-current check above on every subsequent sync. A kept archive is + // stamped too, so an unchanged rewritten source is not decompressed again. const srcStat = fs.statSync(src); fs.utimesSync(dest, srcStat.atimeMs / 1000, Math.ceil(srcStat.mtimeMs) / 1000); - return true; + return outcome; } export function extractSessionIdFromPath(filePath: string): string | null { @@ -201,6 +263,7 @@ export async function syncConversations( const excludedProjects = getExcludedProjects(); const excludedDirSet = new Set(excludedProjects); + const sources: Array<{ srcFile: string; destFile: string }> = []; for (const project of projects) { if (excludedProjects.includes(project)) { console.log("\nSkipping excluded project: " + project); @@ -210,50 +273,59 @@ export async function syncConversations( const projectPath = path.join(sourceDir, project); const stat = statIfExists(projectPath); + if (stat?.isFile()) { + // A finished Codex rollout moved out of its dated directory. + const relative = zstdAvailable || !isCompressedRollout(project) ? archivedRolloutRelativePath(project) : null; + if (relative) sources.push({ srcFile: projectPath, destFile: path.join(destDir, relative) }); + continue; + } if (!stat?.isDirectory()) continue; - const files = findJsonlFiles(projectPath, excludedDirSet); - - for (const file of files) { - const srcFile = path.join(projectPath, file); - const destFile = path.join(destDir, project, file); + for (const file of findJsonlFiles(projectPath, excludedDirSet, { includeCompressed: zstdAvailable })) { + // A compressed rollout is archived as the plain transcript it holds. + sources.push({ srcFile: path.join(projectPath, file), destFile: path.join(destDir, project, file.replace(/\.zst$/, '')) }); + } + } - try { - // Skip message-less transcripts (summarizer-spawned stubs, sessions - // that haven't produced content yet) before they enter the archive. - if (!hasConversationContent(srcFile)) { - result.skipped++; - continue; - } + for (const { srcFile, destFile } of sources) { + try { + // Skip message-less transcripts (summarizer-spawned stubs, sessions + // that haven't produced content yet) before they enter the archive. + if (!(await hasConversationContent(srcFile))) { + result.skipped++; + continue; + } - const wasCopied = copyIfNewer(srcFile, destFile); - if (wasCopied) { - result.copied++; - filesToIndex.push(destFile); - } else { - result.skipped++; + const outcome = await copyIfNewer(srcFile, destFile); + if (outcome === 'copied') { + result.copied++; + filesToIndex.push(destFile); + } else { + result.skipped++; + if (outcome === 'kept') { + console.log(` Kept ${destFile}: its source was rewritten and no longer extends the indexed archive`); } + } - // Check if this file needs a summary (whether newly copied or existing). - // shouldQueueForSummary skips files that already have a real summary or - // an empty zero-exchange sentinel, and retries stale error sentinels (#96). - if (!options.skipSummaries) { - const summaryPath = destFile.replace('.jsonl', '-summary.txt'); - if (shouldQueueForSummary(summaryPath) && !shouldSkipConversation(destFile)) { - // sessionId enables Claude session-resume summarization; when the - // filename has no UUID to extract (e.g. subagent transcripts named - // agent-.jsonl), queue anyway — summarizeConversation falls - // back to summarizing from the transcript text. - const sessionId = extractSessionIdFromPath(destFile) ?? undefined; - filesToSummarize.push({ path: destFile, sessionId }); - } + // Check if this file needs a summary (whether newly copied or existing). + // shouldQueueForSummary skips files that already have a real summary or + // an empty zero-exchange sentinel, and retries stale error sentinels (#96). + if (!options.skipSummaries) { + const summaryPath = destFile.replace('.jsonl', '-summary.txt'); + if (shouldQueueForSummary(summaryPath) && !shouldSkipConversation(destFile)) { + // sessionId enables Claude session-resume summarization; when the + // filename has no UUID to extract (e.g. subagent transcripts named + // agent-.jsonl), queue anyway — summarizeConversation falls + // back to summarizing from the transcript text. + const sessionId = extractSessionIdFromPath(destFile) ?? undefined; + filesToSummarize.push({ path: destFile, sessionId }); } - } catch (error) { - result.errors.push({ - file: srcFile, - error: error instanceof Error ? error.message : String(error) - }); } + } catch (error) { + result.errors.push({ + file: srcFile, + error: error instanceof Error ? error.message : String(error) + }); } } diff --git a/test/sync-codex-archives.test.ts b/test/sync-codex-archives.test.ts new file mode 100644 index 00000000..efe9eed9 --- /dev/null +++ b/test/sync-codex-archives.test.ts @@ -0,0 +1,120 @@ +import { afterEach, beforeEach, describe, expect, it } from 'vitest'; +import fs from 'node:fs'; +import os from 'node:os'; +import path from 'node:path'; +import zlib from 'node:zlib'; +import { syncConversations } from '../src/sync.js'; +import { archivedRolloutRelativePath } from '../src/paths.js'; + +const zstdAvailable = typeof (zlib as any).zstdCompressSync === 'function'; +const SESSION = '019f0000-0000-7000-8000-000000000001'; +const NAME = `rollout-2026-10-15T10-00-01-${SESSION}.jsonl`; + +let root: string; +let archive: string; + +function codexLines(turns: number, from = 1): string { + const lines: string[] = []; + if (from === 1) lines.push(JSON.stringify({ type: 'session_meta', payload: { id: SESSION, cwd: '/synthetic' } })); + for (let turn = from; turn < from + turns; turn++) { + lines.push(JSON.stringify({ type: 'response_item', payload: { type: 'message', role: 'user', + content: [{ type: 'input_text', text: `question ${turn}` }] } })); + lines.push(JSON.stringify({ type: 'response_item', payload: { type: 'message', role: 'assistant', + content: [{ type: 'output_text', text: `answer ${turn}` }] } })); + } + return lines.join('\n') + '\n'; +} + +function write(file: string, content: string, compressed: boolean): string { + fs.mkdirSync(path.dirname(file), { recursive: true }); + fs.writeFileSync(file, compressed ? (zlib as any).zstdCompressSync(Buffer.from(content)) : content); + return file; +} + +// A later source time, as when Codex archives or rewrites a rollout after it was synced. +function later(file: string): void { + const future = new Date(Date.now() + 5000); + fs.utimesSync(file, future, future); +} + +beforeEach(() => { + root = fs.mkdtempSync(path.join(os.tmpdir(), 'sync-codex-archives-')); + archive = path.join(root, 'archive'); +}); + +afterEach(() => { + fs.rmSync(root, { recursive: true, force: true }); +}); + +describe('Codex archived rollouts', () => { + it('maps a flat archived rollout to its dated archive path', () => { + expect(archivedRolloutRelativePath(`${NAME}.zst`)).toBe(path.join('2026', '10', '15', NAME)); + expect(archivedRolloutRelativePath(NAME)).toBe(path.join('2026', '10', '15', NAME)); + expect(archivedRolloutRelativePath('notes.jsonl')).toBeNull(); + }); + + it('archives an uncompressed rollout from archived_sessions at its dated path', async () => { + const sourceDir = path.join(root, 'archived_sessions'); + write(path.join(sourceDir, NAME), codexLines(1), false); + const result = await syncConversations(sourceDir, archive, { skipIndex: true, skipSummaries: true }); + expect(result.copied).toBe(1); + expect(fs.readFileSync(path.join(archive, '2026', '10', '15', NAME), 'utf8')).toBe(codexLines(1)); + }); +}); + +describe.skipIf(!zstdAvailable)('compressed Codex rollouts', () => { + it('archives a compressed archived rollout as the plain transcript it holds', async () => { + const sourceDir = path.join(root, 'archived_sessions'); + write(path.join(sourceDir, `${NAME}.zst`), codexLines(2), true); + const result = await syncConversations(sourceDir, archive, { skipIndex: true, skipSummaries: true }); + expect(result.copied).toBe(1); + expect(fs.readFileSync(path.join(archive, '2026', '10', '15', NAME), 'utf8')).toBe(codexLines(2)); + + const again = await syncConversations(sourceDir, archive, { skipIndex: true, skipSummaries: true }); + expect(again.copied).toBe(0); + }); + + it('archives a compressed rollout inside the sessions tree without its .zst suffix', async () => { + const sourceDir = path.join(root, 'sessions'); + write(path.join(sourceDir, '2026', '10', '15', `${NAME}.zst`), codexLines(1), true); + const result = await syncConversations(sourceDir, archive, { skipIndex: true, skipSummaries: true }); + expect(result.copied).toBe(1); + expect(fs.existsSync(path.join(archive, '2026', '10', '15', NAME))).toBe(true); + }); + + it('extends an archived plain rollout from its compressed continuation', async () => { + const live = path.join(root, 'sessions'); + const plain = write(path.join(live, '2026', '10', '15', NAME), codexLines(1), false); + await syncConversations(live, archive, { skipIndex: true, skipSummaries: true }); + + // Codex moves the finished rollout out of sessions/ and compresses it. + fs.rmSync(plain); + const archived = path.join(root, 'archived_sessions'); + later(write(path.join(archived, `${NAME}.zst`), codexLines(1) + codexLines(1, 2), true)); + const result = await syncConversations(archived, archive, { skipIndex: true, skipSummaries: true }); + expect(result.copied).toBe(1); + expect(fs.readFileSync(path.join(archive, '2026', '10', '15', NAME), 'utf8')).toBe(codexLines(1) + codexLines(1, 2)); + }); + + it('keeps an archive whose compressed source was rewritten, and does not decompress it again', async () => { + const live = path.join(root, 'sessions'); + const plain = write(path.join(live, '2026', '10', '15', NAME), codexLines(2), false); + await syncConversations(live, archive, { skipIndex: true, skipSummaries: true }); + const target = path.join(archive, '2026', '10', '15', NAME); + const previous = fs.readFileSync(target, 'utf8'); + + // A format migration rewrites every record before the rollout is compressed. + fs.rmSync(plain); + const rewritten = codexLines(2).trimEnd().split('\n') + .map((line, ordinal) => JSON.stringify({ ordinal, ...JSON.parse(line) })).join('\n') + '\n'; + const archived = path.join(root, 'archived_sessions'); + const source = write(path.join(archived, `${NAME}.zst`), rewritten, true); + later(source); + const result = await syncConversations(archived, archive, { skipIndex: true, skipSummaries: true }); + expect(result.copied).toBe(0); + expect(fs.readFileSync(target, 'utf8')).toBe(previous); + // The kept archive carries the source time, so an unchanged rewrite is current next time. + expect(fs.statSync(target).mtimeMs).toBeGreaterThanOrEqual(fs.statSync(source).mtimeMs); + expect(fs.readdirSync(path.dirname(target)).filter(name => name.includes('.tmp.'))).toEqual([]); + }); +});