feat(codex): ingest compressed and shared-history rollouts

crates/adapter-support/Cargo.toml

@@ -4,6 +4,7 @@ version.workspace = true
 edition.workspace = true
 
 [dependencies]
+rosetta-capture = { path = "../capture" }
 rosetta-media = { path = "../media" }
 base64 = { workspace = true }
 blake3 = { workspace = true }

crates/adapter-support/src/adapter.rs

@@ -14,6 +14,16 @@ use rosetta_core::{EvidenceOrder, IngestOp, SourceIdentity};
 
 use crate::GroundTruthDoc;
 
+pub use rosetta_capture::input::{InputCandidate, InputEncoding, InputSource};
+
+/// Physical representations of one logical transcript. Paths are relative
+/// to the runtime root; the stable key survives compression and relocation.
+#[derive(Debug, Clone)]
+pub struct TranscriptLocation {
+    pub logical_path: String,
+    pub source: InputSource,
+}
+
 /// A stream parser the generic pipeline knows how to frame. Layout inventory
 /// is intentionally broader than this list: an adapter must opt a stream into
 /// normalization instead of every matching file being treated as a transcript.
@@ -256,6 +266,40 @@ impl NormalizeCtx {
 /// One implementation per agent. Object-safe: the pipeline holds a registry
 /// of `&'static dyn AgentAdapter`.
 pub trait AgentAdapter: Send + Sync {
+    /// Native position of a range's final record; capture remains format-blind.
+    fn transcript_record_index(&self, _record: &[u8]) -> Option<u64> {
+        None
+    }
+
+    fn transcript_header(
+        &self,
+        _path: &str,
+        _first: &[u8],
+    ) -> Result<Option<crate::transcripts::TranscriptHeader>, String> {
+        Ok(None)
+    }
+
+    /// A promise that snapshot records emit no canonical operations. Such
+    /// shared metadata sources can precede independently publishable trees.
+    fn snapshot_is_metadata_only(&self, stream: &str) -> bool {
+        self.snapshot_plan(stream)
+            .is_some_and(|plan| plan.record_queries.is_empty())
+    }
+
+    fn transcript_selection(
+        &self,
+        _stream: &str,
+        _row: &serde_json::Value,
+    ) -> Option<crate::transcripts::TranscriptSelection> {
+        None
+    }
+
+    /// Opt in to representation-independent, decoded-byte capture. Native
+    /// naming rules stay in the adapter; capture only opens declared paths.
+    fn transcript_location(&self, _path: &str) -> Option<TranscriptLocation> {
+        None
+    }
+
     /// Agent kind, e.g. "claude-code".
     fn kind(&self) -> &'static str;
 

crates/adapter-support/src/lib.rs

@@ -159,3 +159,5 @@ impl GroundTruthDoc {
 pub fn glob_match(pattern: &str, path: &str) -> bool {
     Glob::new(pattern).matches(path)
 }
+
+pub mod transcripts;

crates/adapter-support/src/transcripts.rs

@@ -0,0 +1,23 @@
+//! Pure native metadata for resolving shared transcript prefixes.
+
+#[derive(Debug, Clone)]
+pub struct TranscriptReference {
+    pub source_id: String,
+    pub end_offset: u64,
+    pub end_ordinal: u64,
+}
+
+#[derive(Debug, Clone)]
+pub struct TranscriptHeader {
+    pub source_id: String,
+    pub session_id: String,
+    pub base: Option<TranscriptReference>,
+    pub start_ordinal: u64,
+    pub paginated: bool,
+}
+
+#[derive(Debug, Clone)]
+pub struct TranscriptSelection {
+    /// Native path, absolute or relative to the runtime root.
+    pub path: String,
+}

crates/adapters/codex/src/adapter.rs

@@ -18,6 +18,40 @@ use crate::time::rfc3339_ms;
 pub struct CodexAdapter;
 
 impl AgentAdapter for CodexAdapter {
+    fn transcript_record_index(&self, record: &[u8]) -> Option<u64> {
+        serde_json::from_slice::<Value>(record)
+            .ok()?
+            .get("ordinal")?
+            .as_u64()
+    }
+
+    fn transcript_header(
+        &self,
+        path: &str,
+        first: &[u8],
+    ) -> Result<Option<rosetta_adapter_support::transcripts::TranscriptHeader>, String> {
+        crate::storage::header(path, first)
+    }
+    fn transcript_selection(
+        &self,
+        stream: &str,
+        row: &Value,
+    ) -> Option<rosetta_adapter_support::transcripts::TranscriptSelection> {
+        if stream != spec::STATE_DB {
+            return None;
+        }
+        Some(rosetta_adapter_support::transcripts::TranscriptSelection {
+            path: crate::storage::selection_path(row.get("rollout_path")?.as_str()?),
+        })
+    }
+
+    fn transcript_location(
+        &self,
+        path: &str,
+    ) -> Option<rosetta_adapter_support::adapter::TranscriptLocation> {
+        crate::storage::location(path)
+    }
+
     fn kind(&self) -> &'static str {
         "codex"
     }
@@ -28,21 +62,29 @@ impl AgentAdapter for CodexAdapter {
 
     fn stream_driver(&self, stream_id: &str) -> Option<StreamDriver> {
         match stream_id {
-            spec::ROLLOUT => Some(StreamDriver::JsonLines),
+            spec::ROLLOUT
+            | "rollout-compressed"
+            | "archived-rollout"
+            | "archived-rollout-compressed" => Some(StreamDriver::JsonLines),
             spec::STATE_DB => Some(StreamDriver::SqliteSnapshot),
             _ => None,
         }
     }
 
     fn census_session_streams(&self) -> Option<&'static [&'static str]> {
-        Some(&[spec::ROLLOUT])
+        Some(&[
+            spec::ROLLOUT,
+            "rollout-compressed",
+            "archived-rollout",
+            "archived-rollout-compressed",
+        ])
     }
 
     /// A rollout writes its own session, except a guardian review's usage,
     /// which lands on the parent its header names: metadata links the
     /// guardian to that parent, so both are one tree. The state database is
-    /// shared by every tree but declares no record queries, so its members
-    /// emit no ops (`snapshot_plan`).
+    /// shared by every tree; its native selector records emit no canonical
+    /// ops, preserving closure (`snapshot_plan`, `ingest_line`).
     fn cold_admission(&self) -> rosetta_adapter_support::ColdAdmission {
         rosetta_adapter_support::ColdAdmission::RootTree
     }
@@ -61,7 +103,7 @@ impl AgentAdapter for CodexAdapter {
         // renditions made after import (2026-11-01): legacy bundle refs stay
         // valid, so re-streaming every source is not worth it; a rebuild
         // converts them (docs/ARCHITECTURE.md).
-        21
+        22
     }
 
     fn html_artifact_hints(
@@ -84,14 +126,23 @@ impl AgentAdapter for CodexAdapter {
         }
     }
 
+    fn snapshot_is_metadata_only(&self, stream: &str) -> bool {
+        stream == spec::STATE_DB
+    }
+
     fn snapshot_plan(&self, stream_id: &str) -> Option<SnapshotPlan> {
         (stream_id == spec::STATE_DB).then(|| SnapshotPlan {
             // `SELECT *` is intentional compatibility machinery: the base
             // `threads` columns predate optional `name` and millisecond stamp
             // columns. `snapshot_session` inspects whichever optional columns
             // this version exposes without emitting schema-dependent SQL.
             sessions_sql: "SELECT * FROM threads ORDER BY created_at, id".to_owned(),
-            record_queries: Vec::new(),
+            // Preserve the native selection with the archive. It projects no
+            // rows, but keeps the authoritative pointer available offline.
+            record_queries: vec![rosetta_adapter_support::adapter::SnapshotQuery {
+                name: "thread".into(),
+                sql: "SELECT * FROM threads WHERE id = ?1".into(),
+            }],
         })
     }
 
@@ -184,7 +235,7 @@ impl AgentAdapter for CodexAdapter {
     /// opening record, and the file's mtime already gives the pipeline the
     /// session's `updated_at`.
     fn skim(&self, stream_id: &str, stat: &FileStat, head: &[u8], _tail: &[u8]) -> SkimResult {
-        if stream_id != spec::ROLLOUT {
+        if !spec::is_rollout(stream_id) {
             return SkimResult::default();
         }
 
@@ -234,6 +285,9 @@ impl AgentAdapter for CodexAdapter {
     }
 
     fn ingest_line(&self, ctx: &mut NormalizeCtx, stream_id: &str, line: &[u8]) -> LineOutcome {
+        if stream_id == spec::STATE_DB {
+            return LineOutcome::parsed(Vec::new());
+        }
         // Owning the state for the call keeps it out of the borrow of `ctx`,
         // which every row needs mutably for its ordinal. One restore point, so
         // no projection path can lose it.

crates/adapters/codex/src/lib.rs

@@ -22,6 +22,7 @@ pub mod normalize;
 pub mod raw;
 mod session_policy;
 pub mod spec;
+mod storage;
 pub mod time;
 pub mod tool;
 mod usage;

crates/adapters/codex/src/normalize.rs

@@ -89,6 +89,12 @@ pub struct CodexState {
     /// configures the file.
     #[serde(default)]
     header_seen: bool,
+    #[serde(default)]
+    pub(crate) shared_history: bool,
+    #[serde(default)]
+    attached_thread: bool,
+    #[serde(default)]
+    pub(crate) origin_thread: Option<String>,
     /// Current execution cwd used only to resolve relative tool arguments.
     /// The canonical session cwd is selected independently during meta ingest.
     pub(crate) tool_cwd: Option<String>,
@@ -191,6 +197,13 @@ pub fn ingest_line(
         return LineOutcome::parsed(Vec::new());
     }
 
+    if state.shared_history
+        && state.attached_thread
+        && state.origin_thread != state.thread_id
+        && !matches!(record.payload, RolloutPayload::SessionMeta(_))
+    {
+        return LineOutcome::parsed(Vec::new());
+    }
     let mut ops = Vec::new();
     // Permissive parsing means novel fields land in the flatten maps instead of
     // failing the line — that is the whole point, but it must still be visible.
@@ -207,6 +220,14 @@ pub fn ingest_line(
     }
 
     let status = project(ctx, state, stream_id, &record, ts, &mut ops);
+    if state.shared_history && state.origin_thread != state.thread_id {
+        ops.retain(|op| {
+            !matches!(
+                op,
+                IngestOp::HtmlArtifact(_) | IngestOp::HtmlArtifactContent(_)
+            )
+        });
+    }
     LineOutcome { ops, status }
 }
 
@@ -311,8 +332,17 @@ fn session_meta(
     ts: Option<Ms>,
 ) {
     if state.header_seen {
+        if state.shared_history {
+            state.origin_thread = Some(meta.id.clone());
+        }
         return;
     }
+    state.attached_thread = meta.attached_thread.unwrap_or(false);
+    state.shared_history = meta
+        .history_base
+        .as_ref()
+        .is_some_and(|base| !base.is_null());
+    state.origin_thread = Some(meta.id.clone());
     state.header_seen = true;
     ctx.agent_version = Some(meta.cli_version.clone());
     state.thread_id = Some(meta.id.clone());

crates/adapters/codex/src/raw/rollout.rs

@@ -225,6 +225,10 @@ pub struct SessionMeta {
     #[serde(skip_serializing_if = "Option::is_none")]
     pub history_base: Option<Value>,
     #[serde(skip_serializing_if = "Option::is_none")]
+    pub attached_thread: Option<bool>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub forked_from_ordinal_exclusive: Option<u64>,
+    #[serde(skip_serializing_if = "Option::is_none")]
     pub dynamic_tools: Option<Value>,
     #[serde(skip_serializing_if = "Option::is_none")]
     pub subagent_history_start_ordinal: Option<u64>,

crates/adapters/codex/src/spec.rs

@@ -81,6 +81,13 @@ struct FacetView {
     record_types: BTreeMap<String, IgnoredAny>,
 }
 
+pub fn is_rollout(id: &str) -> bool {
+    matches!(
+        id,
+        ROLLOUT | "rollout-compressed" | "archived-rollout" | "archived-rollout-compressed"
+    )
+}
+
 #[cfg(test)]
 mod tests {
     use super::{ROLLOUT, STATE_DB, declared_event_subtypes, declared_response_items, doc};