diff --git a/codex-rs/ext/guardian-v3/src/evidence.rs b/codex-rs/ext/guardian-v3/src/evidence.rs index 5f0f7ee6a6..14c47a8eeb 100644 --- a/codex-rs/ext/guardian-v3/src/evidence.rs +++ b/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 { - 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 { + 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 { + 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(()) } } diff --git a/codex-rs/ext/guardian-v3/src/lib.rs b/codex-rs/ext/guardian-v3/src/lib.rs index b94eec8fd6..edf5dc110a 100644 --- a/codex-rs/ext/guardian-v3/src/lib.rs +++ b/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; diff --git a/codex-rs/ext/guardian-v3/tests/evidence.rs b/codex-rs/ext/guardian-v3/tests/evidence.rs index 166b10b03d..13e2cacd40 100644 --- a/codex-rs/ext/guardian-v3/tests/evidence.rs +++ b/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::>(); 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::>() + )?, + 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::>() + )?, + 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(()) } diff --git a/codex-rs/thread-store/src/lib.rs b/codex-rs/thread-store/src/lib.rs index 596b060780..ede0d1022f 100644 --- a/codex-rs/thread-store/src/lib.rs +++ b/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; diff --git a/codex-rs/thread-store/src/local/mod.rs b/codex-rs/thread-store/src/local/mod.rs index dab03b904b..831153df86 100644 --- a/codex-rs/thread-store/src/local/mod.rs +++ b/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, diff --git a/codex-rs/thread-store/src/local/original_history.rs b/codex-rs/thread-store/src/local/original_history.rs new file mode 100644 index 0000000000..9cb5907201 --- /dev/null +++ b/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 { + 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 { + 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")) +} diff --git a/codex-rs/thread-store/src/local/original_history_tests.rs b/codex-rs/thread-store/src/local/original_history_tests.rs new file mode 100644 index 0000000000..1b2e2372be --- /dev/null +++ b/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> { + 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> { + 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(()) +} diff --git a/codex-rs/thread-store/src/original_history.rs b/codex-rs/thread-store/src/original_history.rs new file mode 100644 index 0000000000..486311d30b --- /dev/null +++ b/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, + pub end: HistoryPosition, +} diff --git a/codex-rs/thread-store/src/store.rs b/codex-rs/thread-store/src/store.rs index 8b2cbe3e21..b5e43dae12 100644 --- a/codex-rs/thread-store/src/store.rs +++ b/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.