From fad8b3d5bcb32dcefb2660a9d46e4d16233697c7 Mon Sep 17 00:00:00 2001 From: felixxia-oai Date: Tue, 10 Nov 2026 21:18:36 +0000 Subject: [PATCH] Initialize Guardian evidence cursors from thread history state (#60670) ## What changed Expose history initialization and the effective history mode to thread-ready contributors. Add `EvidenceCursor::at_thread_ready` with paginated-history validation: - Start new or cleared threads at position zero without requiring an existing rollout. - Start resumed, forked, or supplied-history threads at the durable tail to avoid backfilling existing evidence. - Skip the initial `SessionMeta` header when capturing evidence from position zero. ## Testing Extend evidence tests to cover initialization before a rollout exists, capture of the first original record, rejection of resume without a rollout, and exclusion of existing records on resume. GitOrigin-RevId: 32aa8e2afb71eb4e882b5f1fb107ac8616b33c8e --- codex-rs/core/src/codex_thread.rs | 5 +- codex-rs/core/src/session/mod.rs | 12 +++++ .../src/contributors/thread_lifecycle.rs | 6 +++ codex-rs/ext/extension-api/src/lib.rs | 1 + codex-rs/ext/guardian-v3/src/evidence.rs | 37 +++++++++++++- codex-rs/ext/guardian-v3/tests/evidence.rs | 48 +++++++++++++++---- codex-rs/ext/mcp/src/stream_manager_tests.rs | 2 + 7 files changed, 101 insertions(+), 10 deletions(-) diff --git a/codex-rs/core/src/codex_thread.rs b/codex-rs/core/src/codex_thread.rs index e97fe781ab..c639095147 100644 --- a/codex-rs/core/src/codex_thread.rs +++ b/codex-rs/core/src/codex_thread.rs @@ -433,12 +433,15 @@ impl CodexThread { if contributors.is_empty() { return; } - let config = self.config().await; + let (config, history_initialization, history_mode) = + self.session.thread_ready_config().await; for contributor in contributors { contributor .on_thread_ready(codex_extension_api::ThreadReadyInput { config: config.as_ref(), session_source: &self.session_source, + history_initialization, + history_mode, session_store: &self.session.services.session_extension_data, thread_store: &self.session.services.thread_extension_data, }) diff --git a/codex-rs/core/src/session/mod.rs b/codex-rs/core/src/session/mod.rs index 72c69d087a..b477663de7 100644 --- a/codex-rs/core/src/session/mod.rs +++ b/codex-rs/core/src/session/mod.rs @@ -2473,6 +2473,18 @@ impl Session { ) } + pub(crate) async fn thread_ready_config( + &self, + ) -> (Arc, HistoryInitialization, ThreadHistoryMode) { + let state = self.state.lock().await; + let config = &state.session_configuration; + ( + Arc::clone(&config.original_config_do_not_use), + self.history_initialization, + config.history_mode, + ) + } + pub(crate) async fn get_config(&self) -> std::sync::Arc { let state = self.state.lock().await; state diff --git a/codex-rs/ext/extension-api/src/contributors/thread_lifecycle.rs b/codex-rs/ext/extension-api/src/contributors/thread_lifecycle.rs index b614d536f6..05e3492f67 100644 --- a/codex-rs/ext/extension-api/src/contributors/thread_lifecycle.rs +++ b/codex-rs/ext/extension-api/src/contributors/thread_lifecycle.rs @@ -2,8 +2,10 @@ use std::sync::Arc; use crate::ExtensionData; use crate::ExtensionMetrics; +use codex_history::HistoryInitialization; use codex_mcp::McpResourceClient; use codex_protocol::protocol::SessionSource; +use codex_protocol::protocol::ThreadHistoryMode; use codex_protocol::protocol::TurnEnvironmentSelection; /// Trusted, host-resolved billing attribution for a thread. @@ -40,6 +42,10 @@ pub struct ThreadReadyInput<'a, C> { pub config: &'a C, /// Source that created the session for this thread. pub session_source: &'a SessionSource, + /// How the host initialized the history for this runtime. + pub history_initialization: HistoryInitialization, + /// The effective history mode, including the persisted mode on resume. + pub history_mode: ThreadHistoryMode, /// Store scoped to the host session runtime. pub session_store: &'a ExtensionData, /// Store scoped to this thread runtime. diff --git a/codex-rs/ext/extension-api/src/lib.rs b/codex-rs/ext/extension-api/src/lib.rs index d78b0a8472..7295cd03b8 100644 --- a/codex-rs/ext/extension-api/src/lib.rs +++ b/codex-rs/ext/extension-api/src/lib.rs @@ -26,6 +26,7 @@ pub use capabilities::ResponseItemInjector; pub use capabilities::record_catalog_entries; pub use catalog_refresh::CatalogRefreshBackoff; pub use codex_context_fragments::ContextualUserFragment; +pub use codex_history::HistoryInitialization; pub use codex_mcp::McpProtocolMode; pub use codex_mcp::ToolInfo as McpToolInfo; pub use codex_protocol::models::ContentItem; diff --git a/codex-rs/ext/guardian-v3/src/evidence.rs b/codex-rs/ext/guardian-v3/src/evidence.rs index 39a0389583..5e42cf97d7 100644 --- a/codex-rs/ext/guardian-v3/src/evidence.rs +++ b/codex-rs/ext/guardian-v3/src/evidence.rs @@ -3,6 +3,7 @@ use anyhow::ensure; use codex_history::CodexHarnessMetadata; +use codex_history::HistoryInitialization; use codex_history::ResponseItemEnvelope; use codex_history::RetainedContextEvent; use codex_history::RolloutItem; @@ -11,6 +12,7 @@ use codex_protocol::models::ContentItemProvenance; use codex_protocol::models::ResponseItem; use codex_protocol::protocol::EventMsg; use codex_protocol::protocol::HistoryPosition; +use codex_protocol::protocol::ThreadHistoryMode; use codex_thread_store::OriginalHistoryRequest; use codex_thread_store::ThreadStore; use codex_utils_string::to_json_string_bounded; @@ -81,7 +83,7 @@ impl EvidenceSnapshot { /// 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. +/// history before each capture and coordinate lifetime with thread deletion. #[derive(Clone, Debug, PartialEq, Eq)] pub struct EvidenceCursor { thread_id: ThreadId, @@ -89,6 +91,33 @@ pub struct EvidenceCursor { } impl EvidenceCursor { + /// Fresh roots have no prior evidence. Resumed roots start at their durable tail. + pub async fn at_thread_ready( + store: &dyn ThreadStore, + thread_id: ThreadId, + history_mode: ThreadHistoryMode, + initialization: HistoryInitialization, + ) -> anyhow::Result { + ensure!( + history_mode == ThreadHistoryMode::Paginated, + "Guardian evidence requires paginated root history" + ); + match initialization { + HistoryInitialization::New | HistoryInitialization::Cleared => Ok(Self { + thread_id, + position: HistoryPosition { + thread_id, + end_ordinal_exclusive: 0, + end_byte_offset: 0, + }, + }), + HistoryInitialization::ColdResume + | HistoryInitialization::SuppliedHistory + | HistoryInitialization::WarmFork + | HistoryInitialization::ColdFork => Self::from_now(store, thread_id).await, + } + } + /// 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 @@ -112,6 +141,12 @@ impl EvidenceCursor { let end = history.end; let mut records = Vec::new(); for record in history.items { + // The first capture of a previously empty rollout includes its validated header. + if record.end.end_ordinal_exclusive == 1 + && matches!(record.item, RolloutItem::SessionMeta(_)) + { + continue; + } if let Some(original) = review_original(record.item)? { records.push(OriginalEvidence { id: EvidenceId { diff --git a/codex-rs/ext/guardian-v3/tests/evidence.rs b/codex-rs/ext/guardian-v3/tests/evidence.rs index 09e9bbeba5..d612659429 100644 --- a/codex-rs/ext/guardian-v3/tests/evidence.rs +++ b/codex-rs/ext/guardian-v3/tests/evidence.rs @@ -8,6 +8,7 @@ use codex_guardian_v3::EvidenceCursor; use codex_guardian_v3::OriginalEvidence; use codex_history::AcceptedFileAudio; use codex_history::CodexHarnessMetadata; +use codex_history::HistoryInitialization; use codex_history::ResponseItemEnvelope; use codex_history::RetainedContextEvent; use codex_history::RetainedUserMessage; @@ -175,20 +176,57 @@ Approved action: {"action":{"command":"write /workspace/report.txt"},"outcome":" RolloutItem::ResponseItem(approval), delivered, ]; + let fresh = EvidenceCursor::at_thread_ready( + &store, + owner, + ThreadHistoryMode::Paginated, + HistoryInitialization::New, + ) + .await?; + assert!(!store.live_rollout_path(owner).await?.exists()); + // A resume must establish its tail even if the rollout is missing. + assert!( + EvidenceCursor::at_thread_ready( + &store, + owner, + ThreadHistoryMode::Paginated, + HistoryInitialization::ColdResume, + ) + .await + .is_err() + ); 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"}]}))?; + let old = RolloutItem::ResponseItem(old); + let first_original = serde_json::to_value(&old)?; store .append_items(AppendThreadItemsParams { thread_id: owner, - items: vec![RolloutItem::ResponseItem(old)], + items: vec![old], }) .await?; store .persist_thread(owner, PersistContext::Standard) .await?; store.flush_thread(owner).await?; - let mut cursor = EvidenceCursor::from_now(&store, owner).await?; + let board_originals = |records: &[OriginalEvidence]| -> serde_json::Result> { + records + .iter() + .map(|record| Ok(serde_json::to_value(record)?["original"].take())) + .collect() + }; + assert_eq!( + board_originals(fresh.capture(&store).await?.records())?, + [first_original] + ); + let mut cursor = EvidenceCursor::at_thread_ready( + &store, + owner, + ThreadHistoryMode::Paginated, + HistoryInitialization::ColdResume, + ) + .await?; assert!(cursor.capture(&store).await?.records().is_empty()); for item in &mut items { @@ -224,12 +262,6 @@ Approved action: {"action":{"command":"write /workspace/report.txt"},"outcome":" store.flush_thread(owner).await?; let captured = cursor.capture(&store).await?; // Check the serialized board post, independently of the rollout serializer. - let board_originals = |records: &[OriginalEvidence]| -> serde_json::Result> { - records - .iter() - .map(|record| Ok(serde_json::to_value(record)?["original"].take())) - .collect() - }; assert_eq!(board_originals(captured.records())?, expected); let mut later = user.clone(); diff --git a/codex-rs/ext/mcp/src/stream_manager_tests.rs b/codex-rs/ext/mcp/src/stream_manager_tests.rs index 68e2c3e538..4357fd52b5 100644 --- a/codex-rs/ext/mcp/src/stream_manager_tests.rs +++ b/codex-rs/ext/mcp/src/stream_manager_tests.rs @@ -62,6 +62,8 @@ async fn connect_loaded_task( .on_thread_ready(ThreadReadyInput { config: &(), session_source: &SessionSource::Exec, + history_initialization: codex_extension_api::HistoryInitialization::New, + history_mode: codex_protocol::protocol::ThreadHistoryMode::Legacy, session_store: &session_store, thread_store: &thread_store, })