diff --git a/Cargo.lock b/Cargo.lock index fe2c912..ab1498b 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2846,6 +2846,7 @@ dependencies = [ "wasm-bindgen", "wreq", "wreq-util", + "zstd", ] [[package]] diff --git a/Cargo.toml b/Cargo.toml index 1ec6c6a..5d9f64d 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -60,6 +60,8 @@ wreq = { version = "6.0.0-rc.31", default-features = false, features = ["brotli" wreq-util = { version = "3.0.0-rc.14", features = ["emulation-compression"], optional = true } uuid = { version = "1.23.4", features = ["v4", "v5"] } wasm-bindgen = { version = "0.2.126", optional = true } +# Decoding the rollouts Codex compresses; the library Codex writes them with. +zstd = { version = "0.13.3", default-features = false, optional = true } [target.'cfg(target_os = "macos")'.dependencies] aes = { version = "0.8.4", optional = true } @@ -72,6 +74,8 @@ sha2 = { version = "0.10.9", optional = true } criterion = "0.8.2" proptest = "1.11.0" tempfile = "3.27.0" +# Tests write compressed rollouts the way Codex does. +zstd = { version = "0.13.3", default-features = false } # cdylib for the WASM/C-ABI output; rlib so the cli crate and tests still # link it. Library name follows the package: `use txcript::…`, WASM artifact @@ -100,7 +104,7 @@ required-features = ["search"] [features] # Explicit cloud Cowork reads reuse the Claude Desktop transport. cowork_remote = ["claude_chat"] -default = ["opencode", "hermes", "claude_chat", "cowork_remote", "chatgpt", "search"] +default = ["opencode", "hermes", "claude_chat", "cowork_remote", "chatgpt", "search", "codex_zstd"] # The OpenCode harness store reads OpenCode's SQLite database. The OpenCode # codec itself is always available; only the on-disk store needs rusqlite. opencode = ["dep:rusqlite"] @@ -131,6 +135,11 @@ chatgpt = [ "dep:wreq", "dep:wreq-util", ] +# Codex compresses a rollout it hasn't written to for a week, to +# `.jsonl.zst`. The Codex store reads plain rollouts without this feature and +# leaves the compressed ones out of discovery; with it, both list and load. +# zstd is a C library, so the wasm build goes without. +codex_zstd = ["dep:zstd"] # Fuzzy/substring search over transcripts (`txcript::search`). nucleo-matcher # is pure Rust (helix's matcher) and compiles to wasm32; combine with `wasm` # to expose the bindings. diff --git a/docs/formats/codex.md b/docs/formats/codex.md index c312d7b..3efe583 100644 --- a/docs/formats/codex.md +++ b/docs/formats/codex.md @@ -32,6 +32,16 @@ are not followed, guarding against cycles; symlinked files still list). A file o session if it contains a `session_meta` line carrying an `id`; discovery parses just those lines and skips message payloads entirely. On load, a missing id falls back to the filename's uuid. +Codex compresses a rollout it hasn't written to for 7 days. It replaces the file in place with +`rollout--.jsonl.zst`, one zstd frame at level 3, and expands it back to `.jsonl` the +next time the session is resumed. Upstream gates this behind its +`local_thread_store_compression` feature. With txcript's `codex_zstd` feature, on by default, +discovery lists a compressed rollout and `load` decodes it. Discovery reads only as far as the +header line, so listing a compressed session costs about what a plain one does. Both forms can be +on disk for a moment while Codex compresses or expands a session. Codex reads the plain one then, +and discovery lists only that one. Without the feature a compressed rollout isn't listed. zstd is +a C library, so the wasm build leaves it out. + Codex's own `/archive` (TUI) and `codex archive`/`codex unarchive` (CLI) move a rollout out of this dated tree into a flat sibling directory, `archived_sessions` (no `YYYY/MM/DD` sharding). `CodexStore::default_root` sets `archived_sessions_dir` to that sibling, and discovery walks it @@ -40,8 +50,8 @@ won't show it until it's unarchived back into `sessions/`. A `CodexStore` built custom `sessions_dir` has no archived directory unless one is set with `with_archived_sessions_dir`. -Deletion accepts rollout files in either configured directory, including when the active -directory is missing. It resolves symlinks before checking containment and refuses paths +Deletion accepts rollout files, plain or compressed, in either configured directory, including +when the active directory is missing. It resolves symlinks before checking containment and refuses paths outside both directories. Saving always writes into `sessions_dir`. For Rust callers, the new public field changes struct-literal construction. Replace @@ -148,7 +158,9 @@ Writing lineage back into a rollout is not implemented yet — `from_common` nev (accessed 2026-08-10) - Payload item types (`ResponseItem`, `EventMsg`, `SessionMeta`, `TurnContext`): (accessed 2026-08-10) +- Upstream compression (the `.jsonl.zst` name, the 7-day age, plain read before compressed): + (accessed 2026-10-13) - Authoritative mapping in this repo: `src/harness/codex.rs` (codec, store, tool normalization), with shape fixtures and aggregation assertions in `tests/integration/codex.rs`. -Last verified: 2026-09-26, against src/harness/codex.rs and real local sessions. +Last verified: 2026-10-13, against src/harness/codex.rs and real local sessions. diff --git a/docs/usage.md b/docs/usage.md index 85b9dc6..e9ad6c8 100644 --- a/docs/usage.md +++ b/docs/usage.md @@ -118,11 +118,12 @@ txcript completion fish > ~/.config/fish/completions/txcript.fish [dependencies] txcript = "0.14" # Codecs only: drops the SQLite-backed stores, the live Claude Chat and -# ChatGPT sources, and search. Every codec stays available. +# ChatGPT sources, search, and reading Codex's compressed rollouts. Every +# codec stays available. # txcript = { version = "0.14", default-features = false } ``` -Default features: `opencode` (the SQLite stores: OpenCode, both Cursors, Antigravity), `hermes`, `claude_chat`, `cowork_remote`, `chatgpt`, and `search`. +Default features: `opencode` (the SQLite stores: OpenCode, both Cursors, Antigravity), `hermes`, `claude_chat`, `cowork_remote`, `chatgpt`, `search`, and `codex_zstd` (the rollouts Codex compresses to `.jsonl.zst`). Three layers, smallest to largest: diff --git a/src/harness/codex.rs b/src/harness/codex.rs index 85d0b4c..092f675 100644 --- a/src/harness/codex.rs +++ b/src/harness/codex.rs @@ -12,10 +12,14 @@ //! `from_common` emits both `response_item` and `event_msg` logs so Codex can //! resume and replay the session. It also emits `turn_context`, `token_count`, //! and `task_complete` records for model and usage metadata. +//! +//! Codex compresses a rollout it hasn't written to for a week, in place, to +//! `rollout--.jsonl.zst`, and expands it again when the session is +//! resumed. The store reads both; it always writes the plain form. use std::collections::{HashMap, HashSet}; use std::fs; -use std::io::{BufRead, BufReader}; +use std::io::{self, BufRead, BufReader, Read}; use std::path::{Path, PathBuf}; use chrono::{DateTime, Datelike, SecondsFormat, Utc}; @@ -933,10 +937,10 @@ impl Store for CodexStore { // mean pulling every byte of every rollout on the machine // through a JSON probe to learn nothing more. let has_id = |l: &Line| l.payload.get("id").and_then(Value::as_str).is_some(); - let file = fs::File::open(path).ok()?; + let rollout = open_rollout(path).ok()?; let mut first: Option = None; let mut found_id = false; - for line in BufReader::new(file).lines().map_while(std::io::Result::ok) { + for line in rollout.lines().map_while(io::Result::ok) { if line.trim().is_empty() || !is_session_meta(&line) { continue; } @@ -955,7 +959,7 @@ impl Store for CodexStore { found_id.then(|| { let mut meta = meta_from_lines(std::slice::from_ref(&first)); if meta.id.is_empty() { - meta.id = jsonl::file_id(path); + meta.id = file_id(path); } Discovered { meta, @@ -966,9 +970,9 @@ impl Store for CodexStore { } fn load(&self, reference: &PathBuf) -> Result> { - let mut transcript = Codex::from_text(&fs::read_to_string(reference)?)?; + let mut transcript = Codex::from_text(&read_rollout(reference)?)?; if transcript.meta.id.is_empty() { - transcript.meta.id = jsonl::file_id(reference); + transcript.meta.id = file_id(reference); } Ok(transcript) } @@ -993,10 +997,11 @@ impl Store for CodexStore { } /// Removes a Codex rollout log. Guarded on shape and containment: - /// the reference must be a `.jsonl` file resolving within `sessions_dir` - /// or the configured `archived_sessions_dir`. Neither root itself is deletable. + /// the reference must be a `.jsonl` or `.jsonl.zst` file resolving within + /// `sessions_dir` or the configured `archived_sessions_dir`. Neither root + /// itself is deletable. fn delete(&self, reference: &PathBuf) -> Result<()> { - if reference.extension().is_none_or(|ext| ext != "jsonl") { + if reference.extension().is_none_or(|ext| ext != "jsonl") && !is_compressed(reference) { return Err(Error::Malformed { harness: Codex::NAME, detail: format!("not a codex session file: {}", reference.display()), @@ -1155,9 +1160,7 @@ fn collect_rollouts(dir: &Path, out: &mut Vec) { collect_rollouts(&path, out); } else if let Some(name) = path.file_name().and_then(|n| n.to_str()) && name.starts_with("rollout-") - && path - .extension() - .is_some_and(|ext| ext.eq_ignore_ascii_case("jsonl")) + && (has_extension(&path, "jsonl") || lists_compressed(&path)) { out.push(path); } @@ -1165,6 +1168,73 @@ fn collect_rollouts(dir: &Path, out: &mut Vec) { } } +fn has_extension(path: &Path, extension: &str) -> bool { + path.extension() + .is_some_and(|ext| ext.eq_ignore_ascii_case(extension)) +} + +/// Whether `path` names a rollout Codex has compressed: `….jsonl.zst`, per +/// upstream `codex-rs/rollout/src/compression.rs`. +fn is_compressed(path: &Path) -> bool { + has_extension(path, "zst") + && path + .file_stem() + .is_some_and(|stem| has_extension(Path::new(stem), "jsonl")) +} + +/// Whether discovery lists a compressed rollout. Codex reads the plain file +/// when both forms are on disk, which they are for a moment while it +/// compresses or expands one, so only that one lists and the session doesn't +/// appear twice. Without the feature nothing could read it, so it isn't a +/// session. +fn lists_compressed(path: &Path) -> bool { + cfg!(feature = "codex_zstd") && is_compressed(path) && !path.with_extension("").exists() +} + +/// A rollout's lines, decoded as they're read when Codex has compressed it. +fn open_rollout(path: &Path) -> io::Result> { + let file = fs::File::open(path)?; + if is_compressed(path) { + decoded(file) + } else { + Ok(Box::new(BufReader::new(file))) + } +} + +#[cfg(feature = "codex_zstd")] +fn decoded(file: fs::File) -> io::Result> { + Ok(Box::new(BufReader::new(zstd::stream::read::Decoder::new( + file, + )?))) +} + +#[cfg(not(feature = "codex_zstd"))] +fn decoded(_: fs::File) -> io::Result> { + Err(io::Error::new( + io::ErrorKind::Unsupported, + "a compressed rollout needs txcript's `codex_zstd` feature", + )) +} + +/// A whole rollout as text. +fn read_rollout(path: &Path) -> io::Result { + if !is_compressed(path) { + return fs::read_to_string(path); + } + let mut text = String::new(); + open_rollout(path)?.read_to_string(&mut text)?; + Ok(text) +} + +/// The id a rollout's file name carries, in either form. +fn file_id(path: &Path) -> String { + if is_compressed(path) { + jsonl::file_id(&path.with_extension("")) + } else { + jsonl::file_id(path) + } +} + // ── tool normalization (codex native → canonical) ────────────────────── fn normalize_function_tool(name: &str, input: Value) -> (String, Value) { diff --git a/tests/integration/codex.rs b/tests/integration/codex.rs index df7b44c..012d91b 100644 --- a/tests/integration/codex.rs +++ b/tests/integration/codex.rs @@ -241,6 +241,64 @@ fn discover_without_archived_dir_only_finds_active_sessions() { assert_eq!(found[0].meta.id, "sess-1"); } +/// Codex reads the plain rollout when a compressed copy sits beside it, as +/// one does for a moment while Codex compresses or expands a session. +/// Discovery lists that one, so the session appears once. +#[cfg(feature = "codex_zstd")] +#[test] +fn discover_lists_a_rollout_once_when_both_forms_exist() { + let dir = tempfile::tempdir().unwrap(); + let rollout = format!( + "{}\n", + r#"{"timestamp":"2026-01-02T03:04:05.000Z","type":"session_meta","payload":{"id":"sess-1"}}"#, + ); + let plain = dir.path().join("rollout-2026-01-02T03-04-05-sess-1.jsonl"); + std::fs::write(&plain, &rollout).unwrap(); + std::fs::write( + dir.path() + .join("rollout-2026-01-02T03-04-05-sess-1.jsonl.zst"), + zstd::stream::encode_all(rollout.as_bytes(), 3).unwrap(), + ) + .unwrap(); + + let found = codex::CodexStore::new(dir.path()).discover().unwrap(); + assert_eq!(found.len(), 1); + assert_eq!(found[0].reference, plain); +} + +/// A rollout with no id inside takes its file name's, and a compressed +/// rollout's name is the plain name with `.zst` after it: both forms of one +/// session must come out with the same id. +#[cfg(feature = "codex_zstd")] +#[test] +fn load_names_a_compressed_rollout_as_its_plain_form() { + let dir = tempfile::tempdir().unwrap(); + let rollout = format!( + "{}\n", + r#"{"timestamp":"2026-01-02T03:04:06.000Z","type":"response_item","payload":{"type":"message","role":"user","content":[{"type":"input_text","text":"hi"}]}}"#, + ); + let plain = dir.path().join("rollout-2026-01-02T03-04-05-sess-1.jsonl"); + let compressed = dir + .path() + .join("rollout-2026-01-02T03-04-05-sess-2.jsonl.zst"); + std::fs::write(&plain, &rollout).unwrap(); + std::fs::write( + &compressed, + zstd::stream::encode_all(rollout.as_bytes(), 3).unwrap(), + ) + .unwrap(); + + let store = codex::CodexStore::new(dir.path()); + assert_eq!( + store.load(&plain).unwrap().meta.id, + "rollout-2026-01-02T03-04-05-sess-1" + ); + assert_eq!( + store.load(&compressed).unwrap().meta.id, + "rollout-2026-01-02T03-04-05-sess-2" + ); +} + /// Shaped at codex's granularity: each assistant block is its own message /// (codex stores one `response_item` per line), every assistant turn carries a /// model, only the final text turn carries usage, and `stop_reason` is None diff --git a/tests/integration/store_delete.rs b/tests/integration/store_delete.rs index 2c03d38..7dd918d 100644 --- a/tests/integration/store_delete.rs +++ b/tests/integration/store_delete.rs @@ -309,6 +309,37 @@ fn codex_delete_archived_session_with_or_without_active_directory() { } } +/// A session Codex has compressed is still the user's to delete, and no +/// other `.zst` file is a session. +#[cfg(feature = "codex_zstd")] +#[test] +fn codex_delete_removes_a_compressed_rollout() { + let dir = tempfile::tempdir().unwrap(); + let store = codex::CodexStore::new(dir.path()); + let native = codex::Codex::from_common(&small_common("cold-session")).unwrap(); + let saved = store.save(&native).unwrap(); + + // What Codex leaves behind: `.jsonl.zst`, the plain file gone. + let mut name = saved.reference.clone().into_os_string(); + name.push(".zst"); + let reference = std::path::PathBuf::from(name); + let plain = std::fs::read(&saved.reference).unwrap(); + std::fs::write(&reference, zstd::stream::encode_all(&plain[..], 3).unwrap()).unwrap(); + std::fs::remove_file(&saved.reference).unwrap(); + + let found = store.discover().unwrap(); + assert_eq!(found.len(), 1); + assert_eq!(found[0].reference, reference); + store.delete(&reference).unwrap(); + assert!(!reference.exists()); + assert!(store.discover().unwrap().is_empty()); + + let other = dir.path().join("notes.zst"); + std::fs::write(&other, b"not a session").unwrap(); + assert!(store.delete(&other).is_err()); + assert!(other.exists()); +} + #[test] fn codex_active_delete_works_when_archive_directory_is_missing() { let dir = tempfile::tempdir().unwrap(); diff --git a/tests/regression/codex_compressed.rs b/tests/regression/codex_compressed.rs new file mode 100644 index 0000000..0d362e5 --- /dev/null +++ b/tests/regression/codex_compressed.rs @@ -0,0 +1,47 @@ +#![allow(clippy::expect_used, clippy::panic, clippy::unwrap_used)] + +//! Regression for the rollouts Codex compresses. + +use txcript::Store; +use txcript::harness::codex; + +/// Issue #84. Codex began compressing a rollout it hadn't written to for a +/// week, in place, to `rollout-….jsonl.zst` (seen with Codex 0.162.1). +/// Discovery matched `.jsonl` alone, so on one machine 781 of 814 sessions +/// stopped listing in one pass, with every file still on disk. A compressed +/// rollout must list and load exactly as its plain form does. +#[test] +fn compressed_rollouts_list_and_load_like_plain_ones() { + let rollout = [ + r#"{"timestamp":"2026-10-01T00:00:00.000Z","type":"session_meta","payload":{"id":"sess-cold","timestamp":"2026-10-01T00:00:00.000Z","cwd":"/repo","cli_version":"0.162.1"}}"#, + r#"{"timestamp":"2026-10-01T00:00:01.000Z","type":"response_item","payload":{"type":"message","role":"user","content":[{"type":"input_text","text":"hi"}]}}"#, + r#"{"timestamp":"2026-10-01T00:00:02.000Z","type":"response_item","payload":{"type":"message","role":"assistant","content":[{"type":"output_text","text":"hello"}]}}"#, + ] + .join("\n") + + "\n"; + // Upstream writes one zstd frame at level 3. + let compressed = zstd::stream::encode_all(rollout.as_bytes(), 3).unwrap(); + + let store_holding = |name: &str, bytes: &[u8]| { + let dir = tempfile::tempdir().unwrap(); + let day = dir.path().join("2026").join("10").join("01"); + std::fs::create_dir_all(&day).unwrap(); + std::fs::write(day.join(name), bytes).unwrap(); + dir + }; + let name = "rollout-2026-10-01T00-00-00-sess-cold.jsonl"; + let plain_dir = store_holding(name, rollout.as_bytes()); + let compressed_dir = store_holding(&format!("{name}.zst"), &compressed); + let plain = codex::CodexStore::new(plain_dir.path()); + let store = codex::CodexStore::new(compressed_dir.path()); + + let expected = plain.discover().unwrap(); + let found = store.discover().unwrap(); + assert_eq!(found.len(), 1); + assert_eq!(found[0].meta, expected[0].meta); + assert_eq!(found[0].meta.id, "sess-cold"); + + let loaded = store.load(&found[0].reference).unwrap(); + assert_eq!(loaded.body.len(), 3); + assert_eq!(loaded, plain.load(&expected[0].reference).unwrap()); +} diff --git a/tests/regression/main.rs b/tests/regression/main.rs index fd47007..dc8cae5 100644 --- a/tests/regression/main.rs +++ b/tests/regression/main.rs @@ -7,6 +7,8 @@ //! belong in `tests/integration/`; see `tests/README.md`. mod claude_tool_result; +#[cfg(feature = "codex_zstd")] +mod codex_compressed; mod codex_pairing; use chrono::{DateTime, Utc};