feat(sync): archive Codex's compressed archived rollouts

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<string>, 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>): 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

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

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 = [
     '<INSTRUCTIONS-TO-EPISODIC-MEMORY>DO NOT INDEX THIS CHAT</INSTRUCTIONS-TO-EPISODIC-MEMORY>',
     '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,56 +245,68 @@ 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);
             continue;
         }
         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-<hex>.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-<hex>.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) {