From 0f4b957bfd62c414fb140f6c13405019eabd0a81 Mon Sep 17 00:00:00 2001 From: Nishant Joshi Date: Tue, 13 Oct 2026 18:04:59 -0700 Subject: [PATCH] fix(codex): read rollouts Codex has compressed to .jsonl.zst Codex compresses a rollout it hasn't written to for 7 days, replacing rollout--.jsonl with rollout--.jsonl.zst. The store matched .jsonl only, so those sessions stopped listing. Discovery now lists a compressed rollout, decoding only as far as its header line, load decodes it, and delete accepts its path. When both forms are on disk the plain one lists, as Codex itself reads it. Saving still writes plain .jsonl. Decoding uses the zstd crate behind a new default feature, codex_zstd, so the wasm build goes without the C library. Fixes #84 Co-Authored-By: Claude Opus 5.5 --- Cargo.lock | 1 + Cargo.toml | 11 +++- docs/formats/codex.md | 18 +++++- docs/usage.md | 5 +- src/harness/codex.rs | 94 ++++++++++++++++++++++++---- tests/integration/codex.rs | 58 +++++++++++++++++ tests/integration/store_delete.rs | 31 +++++++++ tests/regression/codex_compressed.rs | 47 ++++++++++++++ tests/regression/main.rs | 2 + 9 files changed, 249 insertions(+), 18 deletions(-) create mode 100644 tests/regression/codex_compressed.rs 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};