Capture Guardian evidence incrementally with explicit acknowledgements (#59670)

## Why

Evidence capture previously reloaded the full history without tracking consumer progress. Live consumers need to read new originals from activation onward and acknowledge them after downstream writes succeed.

## What changed

- Add an in-memory `EvidenceCursor` that starts at the current complete tail without backfilling, captures a fixed unread range, and supports partial acknowledgements. Reads leave progress unchanged.
- Add `ThreadStore::read_original_history` to read original records and their source positions through a fixed cutoff. Local storage supports append-only paginated root histories, including compressed rollouts, and defers unfinished records.
- Reject unsupported histories and invalid ranges, and retain source ordinals as evidence IDs.

## Testing

Add coverage for partial records, compressed suffixes, fixed cutoffs, mismatched thread positions, and truncation. Extend evidence tests to cover activation without backfill, partial acknowledgements, later appends, stale captures, and unchanged cursor progress after capture errors.

GitOrigin-RevId: c7a13749fedc4f44fadddc9dd9875043bccebd95

codex-rs/ext/guardian-v3/src/evidence.rs

@@ -1,5 +1,5 @@
-//! Captures original conversation, tool activity and host-verified evidence from history.
-//! The prototype supports append-only root histories; it has no durable intake state.
+//! Captures original evidence after an in-memory source cursor, without backfilling.
+//! A capture has a fixed end; consumers acknowledge records only after their writes succeed.
 
 use anyhow::ensure;
 use codex_history::ResponseItemEnvelope;
@@ -9,7 +9,8 @@ use codex_protocol::ThreadId;
 use codex_protocol::models::ContentItemProvenance;
 use codex_protocol::models::ResponseItem;
 use codex_protocol::protocol::EventMsg;
-use codex_thread_store::LoadThreadHistoryParams;
+use codex_protocol::protocol::HistoryPosition;
+use codex_thread_store::OriginalHistoryRequest;
 use codex_thread_store::ThreadStore;
 use serde::Serialize;
 use std::sync::Arc;
@@ -17,7 +18,7 @@ use std::sync::Arc;
 /// Position in the containing snapshot's root history, independent of channel routing.
 #[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize)]
 pub struct EvidenceId {
-    pub ordinal: usize,
+    pub ordinal: u64,
 }
 
 /// The original rollout record retains roles, call IDs and host provenance unchanged.
@@ -26,6 +27,8 @@ pub struct EvidenceId {
 pub struct OriginalEvidence {
     pub id: EvidenceId,
     pub original: RolloutItem,
+    #[serde(skip)]
+    end: HistoryPosition,
 }
 
 /// An immutable capture. Later history appends cannot enter its reads or routing.
@@ -33,27 +36,55 @@ pub struct OriginalEvidence {
 #[derive(Clone)]
 pub struct EvidenceSnapshot {
     thread_id: ThreadId,
+    start: HistoryPosition,
+    end: HistoryPosition,
     records: Arc<[OriginalEvidence]>,
 }
 
 impl EvidenceSnapshot {
-    /// Load through the existing history API after the host flushes pending originals.
-    /// This deliberately reloads history; incremental IO and restart recovery are deferred.
-    /// Forked, reverted and ambiguously attributed user histories are unsupported.
-    pub async fn capture(store: &dyn ThreadStore, thread_id: ThreadId) -> anyhow::Result<Self> {
-        let history = store
-            .load_history(LoadThreadHistoryParams {
-                thread_id,
-                include_archived: false,
-            })
+    pub fn thread_id(&self) -> ThreadId {
+        self.thread_id
+    }
+
+    pub fn records(&self) -> &[OriginalEvidence] {
+        &self.records
+    }
+}
+
+/// Live progress for one board. Never serialized or reconstructed after restart.
+/// Create it at board activation, before accepting new input. Hosts flush pending
+/// history before activation/capture and coordinate lifetime with thread deletion.
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub struct EvidenceCursor {
+    thread_id: ThreadId,
+    position: HistoryPosition,
+}
+
+impl EvidenceCursor {
+    /// Start at the current complete tail. Existing records are not backfilled.
+    pub async fn from_now(store: &dyn ThreadStore, thread_id: ThreadId) -> anyhow::Result<Self> {
+        let tail = store
+            .read_original_history(thread_id, OriginalHistoryRequest::Tail)
             .await?;
-        ensure!(
-            history.thread_id == thread_id,
-            "evidence history owner mismatch"
-        );
+        Ok(Self {
+            thread_id,
+            position: tail.end,
+        })
+    }
+
+    pub fn thread_id(&self) -> ThreadId {
+        self.thread_id
+    }
+
+    /// Read the unread suffix through a fixed cutoff without advancing progress.
+    pub async fn capture(&self, store: &dyn ThreadStore) -> anyhow::Result<EvidenceSnapshot> {
+        let history = store
+            .read_original_history(self.thread_id, OriginalHistoryRequest::After(self.position))
+            .await?;
+        let end = history.end;
         let mut records = Vec::new();
-        for (ordinal, item) in history.items.into_iter().enumerate() {
-            let include = match &item {
+        for record in history.items {
+            let include = match &record.item {
                 RolloutItem::ResponseItem(envelope) => selected(envelope)?,
                 RolloutItem::RetainedContext(RetainedContextEvent::VerifiedAnswer {
                     answer,
@@ -69,15 +100,9 @@ impl EvidenceSnapshot {
                     ensure!(message.complete, "incomplete delivered assistant message");
                     true
                 }
-                RolloutItem::SessionMeta(meta) => {
-                    ensure!(
-                        meta.meta.forked_from_id.is_none(),
-                        "forked evidence is unsupported"
-                    );
-                    continue;
-                }
-                RolloutItem::EventMsg(EventMsg::ThreadRolledBack(_)) => {
-                    anyhow::bail!("reverted evidence is unsupported");
+                RolloutItem::SessionMeta(_)
+                | RolloutItem::EventMsg(EventMsg::ThreadRolledBack(_)) => {
+                    anyhow::bail!("evidence history was replaced");
                 }
                 RolloutItem::RetainedContext(RetainedContextEvent::AcceptedFileAudio {
                     ..
@@ -91,27 +116,65 @@ impl EvidenceSnapshot {
                 | RolloutItem::WorldState(_)
                 | RolloutItem::SecurityRiskScore(_)
                 | RolloutItem::EventMsg(_)
-                | RolloutItem::RealtimeItem(_) => continue,
+                | RolloutItem::RealtimeItem(_) => false,
             };
             if include {
                 records.push(OriginalEvidence {
-                    id: EvidenceId { ordinal },
-                    original: item,
+                    id: EvidenceId {
+                        ordinal: record.end.end_ordinal_exclusive.checked_sub(1).ok_or_else(
+                            || anyhow::anyhow!("evidence record has no source ordinal"),
+                        )?,
+                    },
+                    original: record.item,
+                    end: record.end,
                 });
             }
         }
-        Ok(Self {
-            thread_id,
+        Ok(EvidenceSnapshot {
+            thread_id: self.thread_id,
+            start: self.position,
+            end,
             records: records.into(),
         })
     }
 
-    pub fn thread_id(&self) -> ThreadId {
-        self.thread_id
+    /// Records from this capture that have not yet been acknowledged.
+    pub fn pending<'a>(
+        &self,
+        snapshot: &'a EvidenceSnapshot,
+    ) -> anyhow::Result<&'a [OriginalEvidence]> {
+        ensure!(
+            snapshot.thread_id == self.thread_id
+                && snapshot.start.thread_id == self.position.thread_id
+                && snapshot.end.thread_id == self.position.thread_id
+                && snapshot.start.end_byte_offset <= self.position.end_byte_offset
+                && self.position.end_byte_offset <= snapshot.end.end_byte_offset
+                && snapshot.start.end_ordinal_exclusive <= self.position.end_ordinal_exclusive
+                && self.position.end_ordinal_exclusive <= snapshot.end.end_ordinal_exclusive,
+            "evidence capture does not contain the current cursor"
+        );
+        let start = snapshot
+            .records
+            .partition_point(|record| record.end.end_byte_offset <= self.position.end_byte_offset);
+        Ok(&snapshot.records[start..])
     }
 
-    pub fn records(&self) -> &[OriginalEvidence] {
-        &self.records
+    /// Acknowledge a successfully consumed prefix of the pending evidence.
+    /// Completing a capture also skips trailing records excluded by evidence selection.
+    pub fn acknowledge(&mut self, snapshot: &EvidenceSnapshot, count: usize) -> anyhow::Result<()> {
+        let pending = self.pending(snapshot)?;
+        ensure!(
+            count <= pending.len(),
+            "acknowledgement exceeds pending evidence"
+        );
+        self.position = if count == pending.len() {
+            snapshot.end
+        } else if count > 0 {
+            pending[count - 1].end
+        } else {
+            self.position
+        };
+        Ok(())
     }
 }
 

codex-rs/ext/guardian-v3/src/lib.rs

@@ -10,6 +10,7 @@ use codex_protocol::SessionId;
 use codex_state::SqliteConfig;
 
 mod evidence;
+pub use evidence::EvidenceCursor;
 pub use evidence::EvidenceId;
 pub use evidence::EvidenceSnapshot;
 pub use evidence::OriginalEvidence;

codex-rs/ext/guardian-v3/tests/evidence.rs

@@ -4,7 +4,7 @@ use codex_context_fragments::AdditionalContextUserFragment;
 use codex_context_fragments::AnnotatedContent;
 use codex_context_fragments::ContextualUserFragment;
 use codex_context_fragments::message_from_parts;
-use codex_guardian_v3::EvidenceSnapshot;
+use codex_guardian_v3::EvidenceCursor;
 use codex_history::AcceptedFileAudio;
 use codex_history::CodexHarnessMetadata;
 use codex_history::ResponseItemEnvelope;
@@ -62,7 +62,7 @@ async fn captures_originals_without_generated_context_or_later_appends() -> anyh
             dynamic_tools: Vec::new(),
             selected_capability_roots: Vec::new(),
             multi_agent_version: None,
-            history_mode: ThreadHistoryMode::Legacy,
+            history_mode: ThreadHistoryMode::Paginated,
             history_base: None,
             subagent_history_start_ordinal: None,
             initial_window_id: "window-1".into(),
@@ -147,6 +147,22 @@ Approved action: {"action":{"command":"write /workspace/report.txt"},"outcome":"
         RolloutItem::ResponseItem(approval),
         delivered,
     ];
+    let mut old = user.clone();
+    old.item = serde_json::from_value(json!({"type":"message","id":"old","role":"user",
+        "content":[{"type":"input_text","text":"Before board activation"}]}))?;
+    store
+        .append_items(AppendThreadItemsParams {
+            thread_id: owner,
+            items: vec![RolloutItem::ResponseItem(old)],
+        })
+        .await?;
+    store
+        .persist_thread(owner, PersistContext::Standard)
+        .await?;
+    store.flush_thread(owner).await?;
+    let mut cursor = EvidenceCursor::from_now(&store, owner).await?;
+    assert!(cursor.capture(&store).await?.records().is_empty());
+
     let mut items = expected.clone();
     items.insert(1, RolloutItem::ResponseItem(generated));
     items.extend([
@@ -171,7 +187,7 @@ Approved action: {"action":{"command":"write /workspace/report.txt"},"outcome":"
         .persist_thread(owner, PersistContext::Standard)
         .await?;
     store.flush_thread(owner).await?;
-    let captured = EvidenceSnapshot::capture(&store, owner).await?;
+    let captured = cursor.capture(&store).await?;
     let originals = captured
         .records()
         .iter()
@@ -179,7 +195,7 @@ Approved action: {"action":{"command":"write /workspace/report.txt"},"outcome":"
         .collect::<Vec<_>>();
     assert_eq!(
         serde_json::to_value(originals)?,
-        serde_json::to_value(expected)?
+        serde_json::to_value(&expected)?
     );
 
     let mut later = user.clone();
@@ -196,17 +212,36 @@ Approved action: {"action":{"command":"write /workspace/report.txt"},"outcome":"
         })
         .await?;
     store.flush_thread(owner).await?;
-    let latest = EvidenceSnapshot::capture(&store, owner).await?;
-    let new_record = latest.records().last().expect("later original");
+    // Reads do not acknowledge delivery, and a later append cannot enter the old capture.
+    assert_eq!(cursor.pending(&captured)?.len(), expected.len());
+    cursor.acknowledge(&captured, /*count*/ 2)?;
     assert_eq!(
-        serde_json::to_value(&new_record.original)?,
-        serde_json::to_value(RolloutItem::ResponseItem(later.clone()))?
+        serde_json::to_value(
+            cursor
+                .pending(&captured)?
+                .iter()
+                .map(|r| &r.original)
+                .collect::<Vec<_>>()
+        )?,
+        serde_json::to_value(&expected[2..])?
     );
-    assert_eq!(latest.records().len(), captured.records().len() + 1);
+    cursor.acknowledge(&captured, expected.len() - 2)?;
+    assert!(cursor.pending(&captured)?.is_empty());
+    let latest = cursor.capture(&store).await?;
     assert_eq!(
-        serde_json::to_value(&latest.records()[..captured.records().len()])?,
-        serde_json::to_value(captured.records())?
+        serde_json::to_value(
+            latest
+                .records()
+                .iter()
+                .map(|r| &r.original)
+                .collect::<Vec<_>>()
+        )?,
+        serde_json::to_value(vec![RolloutItem::ResponseItem(later.clone())])?
     );
+    assert_eq!(captured.records().len(), expected.len());
+    cursor.acknowledge(&latest, /*count*/ 1)?;
+    assert!(cursor.capture(&store).await?.records().is_empty());
+    assert!(cursor.acknowledge(&captured, /*count*/ 0).is_err());
 
     let mut ambiguous = later;
     ambiguous.metadata = None;
@@ -217,7 +252,9 @@ Approved action: {"action":{"command":"write /workspace/report.txt"},"outcome":"
         })
         .await?;
     store.flush_thread(owner).await?;
-    assert!(EvidenceSnapshot::capture(&store, owner).await.is_err());
+    let before = cursor.clone();
+    assert!(cursor.capture(&store).await.is_err());
+    assert_eq!(cursor, before);
     store.shutdown_thread(owner).await?;
     Ok(())
 }

codex-rs/thread-store/src/lib.rs

@@ -9,6 +9,7 @@ mod in_memory;
 mod live_thread;
 mod local;
 mod local_history_account;
+mod original_history;
 mod projects;
 mod queue_store;
 mod store;
@@ -44,6 +45,9 @@ pub use local::RolloutMigrationProgress;
 pub use local::RolloutMigrationReport;
 pub use local::RolloutMigrationStatus;
 pub use local_history_account::LocalHistoryAccount;
+pub use original_history::OriginalHistoryItem;
+pub use original_history::OriginalHistoryRead;
+pub use original_history::OriginalHistoryRequest;
 pub use projects::CreateProjectParams;
 pub use projects::CreatedProject;
 pub use projects::DeletedProject;

codex-rs/thread-store/src/local/mod.rs

@@ -7,6 +7,7 @@ mod list_threads;
 mod live_writer;
 mod model_context;
 mod move_thread_to_section;
+mod original_history;
 mod paginated_fork;
 mod pending_thread_metadata;
 mod projects;
@@ -616,6 +617,14 @@ impl ThreadStore for LocalThreadStore {
         Box::pin(async move { model_context::load_latest_model_context(self, params).await })
     }
 
+    fn read_original_history(
+        &self,
+        thread_id: ThreadId,
+        request: crate::OriginalHistoryRequest,
+    ) -> ThreadStoreFuture<'_, crate::OriginalHistoryRead> {
+        Box::pin(original_history::read(self, thread_id, request))
+    }
+
     fn is_model_context_current(
         &self,
         params: LoadThreadHistoryParams,

codex-rs/thread-store/src/local/original_history.rs

@@ -0,0 +1,150 @@
+//! Captures a fixed tail and reads its original-record suffix through one rollout reader.
+//! Unfinished lines are excluded; suffix reads require contiguous, complete records.
+
+use std::io;
+use std::io::BufRead;
+use std::io::BufReader;
+use std::io::Read;
+use std::io::Seek;
+use std::io::SeekFrom;
+
+use codex_protocol::ThreadId;
+use codex_protocol::protocol::HistoryPosition;
+use codex_protocol::protocol::ThreadHistoryMode;
+use codex_rollout::ReverseJsonlScanner;
+use codex_rollout::RolloutItem;
+use codex_rollout::ScanOutcome;
+use codex_rollout::open_rollout_seekable_reader;
+use codex_rollout::parse_rollout_line_bytes;
+
+use super::LocalThreadStore;
+use super::thread_rollout_resolver;
+use crate::OriginalHistoryItem;
+use crate::OriginalHistoryRead;
+use crate::OriginalHistoryRequest;
+use crate::ThreadStoreError;
+use crate::ThreadStoreResult;
+
+#[cfg(test)]
+#[path = "original_history_tests.rs"]
+mod tests;
+
+pub(super) async fn read(
+    store: &LocalThreadStore,
+    thread_id: ThreadId,
+    request: OriginalHistoryRequest,
+) -> ThreadStoreResult<OriginalHistoryRead> {
+    let source = thread_rollout_resolver::resolve_current(store, thread_id)
+        .await?
+        .ok_or(ThreadStoreError::ThreadNotFound { thread_id })?;
+    if source.rollout_id != thread_id {
+        return Err(ThreadStoreError::Unsupported {
+            operation: "original_history_after_revert",
+        });
+    }
+    let (stop, keep_running) = tokio::sync::oneshot::channel::<()>();
+    let result = tokio::task::spawn_blocking(move || {
+        let mut file = open_rollout_seekable_reader(&source.path, || stop.is_closed())?;
+        let mut header = Vec::new();
+        BufReader::new(&mut file).read_until(b'\n', &mut header)?;
+        let line = parse_rollout_line_bytes(&header).map_err(io::Error::other)?;
+        let RolloutItem::SessionMeta(meta) = line.item else {
+            return Err(io::Error::other("original history has no session header"));
+        };
+        if meta.meta.id != thread_id
+            || meta.meta.history_mode != ThreadHistoryMode::Paginated
+            || meta.meta.history_base.is_some()
+            || meta.meta.forked_from_id.is_some()
+            || meta.meta.parent_thread_id.is_some()
+            || meta.meta.subagent_history_start_ordinal.is_some()
+        {
+            return Err(io::Error::other(
+                "original history requires a paginated root",
+            ));
+        }
+        let end_byte_offset = complete_end(&mut file)?;
+        let end_ordinal_exclusive = {
+            let mut scanner = ReverseJsonlScanner::new_at(&mut file, end_byte_offset)?;
+            let Some(ScanOutcome::Parsed(last)) = scanner.scan_next_rollout_line()? else {
+                return Err(io::Error::other("original history has an invalid tail"));
+            };
+            last.ordinal
+                .and_then(|ordinal| ordinal.checked_add(1))
+                .ok_or_else(|| io::Error::other("original history tail has no next ordinal"))?
+        };
+        let end = HistoryPosition {
+            thread_id,
+            end_ordinal_exclusive,
+            end_byte_offset,
+        };
+        match request {
+            OriginalHistoryRequest::Tail => Ok(OriginalHistoryRead {
+                items: Vec::new(),
+                end,
+            }),
+            OriginalHistoryRequest::After(start) => {
+                if start.thread_id != thread_id
+                    || start.end_byte_offset > end.end_byte_offset
+                    || start.end_ordinal_exclusive > end.end_ordinal_exclusive
+                {
+                    return Err(io::Error::other("original history range no longer applies"));
+                }
+                file.seek(SeekFrom::Start(start.end_byte_offset))?;
+                let mut reader =
+                    BufReader::new(file.take(end.end_byte_offset - start.end_byte_offset));
+                let mut position = start;
+                let mut bytes = Vec::new();
+                let mut items = Vec::new();
+                while reader.read_until(b'\n', &mut bytes)? != 0 {
+                    if !bytes.ends_with(b"\n") {
+                        return Err(io::Error::other("original history record is incomplete"));
+                    }
+                    let line = parse_rollout_line_bytes(&bytes).map_err(io::Error::other)?;
+                    if line.ordinal != Some(position.end_ordinal_exclusive) {
+                        return Err(io::Error::other("original history ordinal changed"));
+                    }
+                    position.end_byte_offset += bytes.len() as u64;
+                    position.end_ordinal_exclusive = position
+                        .end_ordinal_exclusive
+                        .checked_add(1)
+                        .ok_or_else(|| io::Error::other("original history ordinal overflow"))?;
+                    items.push(OriginalHistoryItem {
+                        item: line.item,
+                        end: position,
+                    });
+                    bytes.clear();
+                }
+                if position != end {
+                    return Err(io::Error::other("original history ended before its cutoff"));
+                }
+                Ok(OriginalHistoryRead { items, end })
+            }
+        }
+    })
+    .await;
+    drop(keep_running);
+    result
+        .map_err(|error| ThreadStoreError::Internal {
+            message: format!("original history reader failed: {error}"),
+        })?
+        .map_err(|error| ThreadStoreError::Internal {
+            message: format!("cannot read original history: {error}"),
+        })
+}
+
+/// Find the last complete record boundary without scanning the earlier history.
+fn complete_end(file: &mut (impl Read + Seek)) -> io::Result<u64> {
+    let mut end = file.seek(SeekFrom::End(0))?;
+    let mut bytes = [0; 8192];
+    while end > 0 {
+        let count = end.min(bytes.len() as u64) as usize;
+        let start = end - count as u64;
+        file.seek(SeekFrom::Start(start))?;
+        file.read_exact(&mut bytes[..count])?;
+        if let Some(index) = bytes[..count].iter().rposition(|byte| *byte == b'\n') {
+            return Ok(start + index as u64 + 1);
+        }
+        end = start;
+    }
+    Err(io::Error::other("original history has no complete record"))
+}

codex-rs/thread-store/src/local/original_history_tests.rs

@@ -0,0 +1,194 @@
+#![cfg(test)]
+//! Raw history captures retain complete originals and defer unfinished suffixes.
+
+use std::fs::OpenOptions;
+use std::io::Write;
+
+use codex_protocol::ThreadId;
+use codex_protocol::protocol::HistoryPosition;
+use codex_protocol::protocol::ThreadHistoryMode;
+use pretty_assertions::assert_eq;
+use serde_json::json;
+
+use super::super::test_support::test_config;
+use super::super::test_support::write_session_file_with_history_mode;
+use crate::LocalThreadStore;
+use crate::OriginalHistoryRequest;
+use crate::ThreadStore;
+
+#[tokio::test]
+async fn reads_complete_fixed_suffix_and_rejects_truncated_source()
+-> Result<(), Box<dyn std::error::Error>> {
+    let dir = tempfile::tempdir()?;
+    let id = uuid::Uuid::new_v4();
+    let owner = ThreadId::from_string(&id.to_string())?;
+    let path = write_session_file_with_history_mode(
+        dir.path(),
+        "2025-01-03T10-00-00",
+        id,
+        ThreadHistoryMode::Paginated,
+    )?;
+    let store = LocalThreadStore::new(test_config(dir.path()), /*state_db*/ None);
+    let start = store
+        .read_original_history(owner, OriginalHistoryRequest::Tail)
+        .await?
+        .end;
+    let original = json!({"timestamp":"2025-01-03T10:00:01Z","ordinal":1,"type":"response_item",
+        "payload":{"type":"message","role":"assistant","content":[{"type":"output_text","text":"Original é🙂"}]}});
+    let line = format!("{original}\n");
+    let mut file = OpenOptions::new().append(true).open(&path)?;
+    // The tail ignores a partially appended record rather than losing it.
+    file.write_all(&line.as_bytes()[..12])?;
+    assert_eq!(
+        store
+            .read_original_history(owner, OriginalHistoryRequest::Tail)
+            .await?
+            .end,
+        start
+    );
+    let partial = store
+        .read_original_history(owner, OriginalHistoryRequest::After(start))
+        .await?;
+    assert!(partial.items.is_empty());
+    assert_eq!(partial.end, start);
+    file.write_all(&line.as_bytes()[12..])?;
+    let read = store
+        .read_original_history(owner, OriginalHistoryRequest::After(start))
+        .await?;
+    let end = HistoryPosition {
+        end_byte_offset: start.end_byte_offset + line.len() as u64,
+        end_ordinal_exclusive: start.end_ordinal_exclusive + 1,
+        ..start
+    };
+    let mut later = original.clone();
+    later["ordinal"] = json!(2);
+    later["payload"]["content"][0]["text"] = json!("After the cutoff");
+    writeln!(file, "{later}")?;
+    assert_eq!(read.end, end);
+    assert_eq!(read.items.len(), 1);
+    assert_eq!(read.items[0].end, end);
+    assert_eq!(
+        serde_json::to_value(&read.items[0].item)?,
+        json!({"type":"response_item","payload":original["payload"]})
+    );
+    let suffix = store
+        .read_original_history(owner, OriginalHistoryRequest::After(end))
+        .await?;
+    assert_eq!(suffix.items.len(), 1);
+    assert_eq!(
+        serde_json::to_value(&suffix.items[0].item)?,
+        json!({"type":"response_item","payload":later["payload"]})
+    );
+    let other = ThreadId::new();
+    let wrong = HistoryPosition {
+        thread_id: other,
+        ..start
+    };
+    assert!(
+        store
+            .read_original_history(owner, OriginalHistoryRequest::After(wrong))
+            .await
+            .is_err()
+    );
+    // Windows append-only handles cannot truncate the file.
+    drop(file);
+    OpenOptions::new()
+        .write(true)
+        .open(&path)?
+        .set_len(start.end_byte_offset)?;
+    assert!(
+        store
+            .read_original_history(owner, OriginalHistoryRequest::After(end))
+            .await
+            .is_err()
+    );
+    Ok(())
+}
+
+#[tokio::test]
+async fn reads_compressed_suffix_and_defers_unfinished_record()
+-> Result<(), Box<dyn std::error::Error>> {
+    let dir = tempfile::tempdir()?;
+    let id = uuid::Uuid::new_v4();
+    let owner = ThreadId::from_string(&id.to_string())?;
+    let path = write_session_file_with_history_mode(
+        dir.path(),
+        "2025-01-03T10-00-00",
+        id,
+        ThreadHistoryMode::Paginated,
+    )?;
+    let store = LocalThreadStore::new(test_config(dir.path()), /*state_db*/ None);
+    let start = store
+        .read_original_history(owner, OriginalHistoryRequest::Tail)
+        .await?
+        .end;
+    let original = json!({"timestamp":"2025-01-03T10:00:01Z","ordinal":1,"type":"response_item",
+        "payload":{"type":"message","role":"assistant","content":[{"type":"output_text","text":"Original é🙂"}]}});
+    let mut later = original.clone();
+    later["ordinal"] = json!(2);
+    later["payload"]["content"][0]["text"] = json!("Completed in the next frame");
+    let original_line = format!("{original}\n");
+    let later_line = format!("{later}\n");
+    let mut file = OpenOptions::new().append(true).open(&path)?;
+    file.write_all(original_line.as_bytes())?;
+    file.write_all(&later_line.as_bytes()[..12])?;
+    drop(file);
+
+    let compressed_path = path.with_extension("jsonl.zst");
+    let contents = std::fs::read(&path)?;
+    std::fs::write(
+        &compressed_path,
+        zstd::stream::encode_all(contents.as_slice(), /*level*/ 3)?,
+    )?;
+    std::fs::remove_file(&path)?;
+    let read = store
+        .read_original_history(owner, OriginalHistoryRequest::After(start))
+        .await?;
+    let end = HistoryPosition {
+        end_byte_offset: start.end_byte_offset + original_line.len() as u64,
+        end_ordinal_exclusive: start.end_ordinal_exclusive + 1,
+        ..start
+    };
+    assert_eq!(read.end, end);
+    assert_eq!(read.items.len(), 1);
+    assert_eq!(
+        (
+            serde_json::to_value(&read.items[0].item)?,
+            read.items[0].end
+        ),
+        (
+            json!({"type":"response_item","payload":original["payload"]}),
+            end
+        )
+    );
+
+    // A complete zstd frame may still end with an unfinished logical JSONL record.
+    let mut compressed = OpenOptions::new().append(true).open(&compressed_path)?;
+    compressed.write_all(&zstd::stream::encode_all(
+        &later_line.as_bytes()[12..],
+        /*level*/ 3,
+    )?)?;
+    drop(compressed);
+    let suffix = store
+        .read_original_history(owner, OriginalHistoryRequest::After(end))
+        .await?;
+    let tail = HistoryPosition {
+        end_byte_offset: end.end_byte_offset + later_line.len() as u64,
+        end_ordinal_exclusive: end.end_ordinal_exclusive + 1,
+        ..end
+    };
+    assert_eq!(suffix.end, tail);
+    assert_eq!(suffix.items.len(), 1);
+    assert_eq!(
+        (
+            serde_json::to_value(&suffix.items[0].item)?,
+            suffix.items[0].end
+        ),
+        (
+            json!({"type":"response_item","payload":later["payload"]}),
+            tail
+        )
+    );
+    assert!(!path.exists());
+    Ok(())
+}

codex-rs/thread-store/src/original_history.rs

@@ -0,0 +1,28 @@
+//! Fixed ranges of original records for live, forward-only history consumers.
+//! Positions identify persisted source records; they do not acknowledge downstream writes.
+
+use codex_protocol::protocol::HistoryPosition;
+use codex_rollout::RolloutItem;
+
+/// Capture the current complete tail, optionally reading records after a saved position.
+/// This API deliberately provides no implicit backfill or restart recovery.
+pub enum OriginalHistoryRequest {
+    /// Return only the current complete end position.
+    Tail,
+    /// Freeze the current complete tail and read originals after this position.
+    /// Appends beyond that cutoff are excluded from the returned records.
+    After(HistoryPosition),
+}
+
+/// One original record and the exclusive position immediately after it.
+pub struct OriginalHistoryItem {
+    pub item: RolloutItem,
+    pub end: HistoryPosition,
+}
+
+/// Original records in source order and their fixed upper boundary.
+/// A tail request returns no items. Reading never advances consumer progress.
+pub struct OriginalHistoryRead {
+    pub items: Vec<OriginalHistoryItem>,
+    pub end: HistoryPosition,
+}

codex-rs/thread-store/src/store.rs

@@ -254,6 +254,23 @@ pub trait ThreadStore: Any + Send + Sync {
         params: LoadThreadHistoryParams,
     ) -> ThreadStoreFuture<'_, StoredThreadHistory>;
 
+    /// Captures a complete tail, optionally reading originals after a saved position.
+    /// Local storage uses one reader for the cutoff and records; compressed rollouts
+    /// are decoded once per call. It supports append-only paginated roots;
+    /// other histories and backends must fail explicitly rather than omit originals.
+    /// Callers flush pending writes before capturing a boundary and own progress.
+    fn read_original_history(
+        &self,
+        _thread_id: ThreadId,
+        _request: crate::OriginalHistoryRequest,
+    ) -> ThreadStoreFuture<'_, crate::OriginalHistoryRead> {
+        Box::pin(async {
+            Err(ThreadStoreError::Unsupported {
+                operation: "read_original_history",
+            })
+        })
+    }
+
     /// Loads the persisted rollout items needed to reconstruct the latest model-visible context.
     ///
     /// Implementations that cannot perform a targeted read may return the full persisted history.