From 8260619cb6dbb23fc43fd9b8567e98fb103aebd4 Mon Sep 17 00:00:00 2001 From: felixxia-oai Date: Mon, 7 Sep 2026 12:23:43 +0000 Subject: [PATCH] Centralize Guardian context mode and checkpoint policy (#43458) ## What changed Resolve `GuardianContextMode` once at session construction and share it across history retention, replay, evidence capture, compaction, and review consumers. Extract synchronous reviewer context policy and consolidate Guardian V2 parent checkpoint selection so sampling and fast approval use the same eligibility checks. Preserve legacy checkpoint omission and strict compatibility requirements for thread-owned context. ## Testing Adapt existing review and history tests to the explicit context mode. Move checkpoint tests alongside the extracted selector, retaining coverage for latest-item identity, unusable checkpoints, and complete-item size limits. GitOrigin-RevId: c73fb6fea556921acfffe758a5f709d7494019dd --- codex-rs/core/src/agent/control/spawn.rs | 20 +- .../src/agent/control/user_authorization.rs | 5 +- codex-rs/core/src/compact.rs | 4 +- .../core/src/context/guardian_context_mode.rs | 23 +++ .../src/context/guardian_review_evidence.rs | 22 +-- codex-rs/core/src/context/mod.rs | 2 + codex-rs/core/src/context_manager/history.rs | 33 ++-- .../core/src/context_manager/history_tests.rs | 3 +- codex-rs/core/src/guardian/review.rs | 3 +- codex-rs/core/src/guardian/review_session.rs | 142 ++++---------- .../src/guardian/review_session_context.rs | 105 +++++++++++ .../core/src/guardian/review_session_tests.rs | 19 +- codex-rs/core/src/session/mod.rs | 3 +- codex-rs/core/src/session/retained_context.rs | 4 +- .../src/session/rollout_reconstruction.rs | 17 +- codex-rs/core/src/session/session.rs | 14 +- codex-rs/core/src/session/tests.rs | 9 +- codex-rs/core/src/state/session.rs | 5 +- .../src/tools/handlers/request_user_input.rs | 3 +- .../handlers/request_user_input_tests.rs | 4 +- .../guardian-v2/src/async_scorer/approval.rs | 21 ++- .../guardian-v2/src/async_scorer/extension.rs | 177 +++++------------- .../src/async_scorer/extension_tests.rs | 162 ---------------- .../ext/guardian-v2/src/async_scorer/mod.rs | 1 + .../src/async_scorer/parent_compaction.rs | 112 +++++++++++ .../async_scorer/parent_compaction_tests.rs | 168 +++++++++++++++++ 26 files changed, 599 insertions(+), 482 deletions(-) create mode 100644 codex-rs/core/src/context/guardian_context_mode.rs create mode 100644 codex-rs/core/src/guardian/review_session_context.rs create mode 100644 codex-rs/ext/guardian-v2/src/async_scorer/parent_compaction.rs create mode 100644 codex-rs/ext/guardian-v2/src/async_scorer/parent_compaction_tests.rs diff --git a/codex-rs/core/src/agent/control/spawn.rs b/codex-rs/core/src/agent/control/spawn.rs index 34f281df9e..6dccadb96d 100644 --- a/codex-rs/core/src/agent/control/spawn.rs +++ b/codex-rs/core/src/agent/control/spawn.rs @@ -6,6 +6,7 @@ use crate::config::PermissionProfileSnapshot; use crate::context::ContextualUserFragment; use crate::context::CurrentTimeReminder; use crate::context::DeveloperInstructions; +use crate::context::GuardianContextMode; use crate::context::ManagedDeveloperInstructions; use crate::context::MultiAgentModeInstructions; use crate::context::MultiAgentRoleInstructions; @@ -15,7 +16,6 @@ use crate::tools::handlers::multi_agents_common::build_agent_resume_config; use codex_context_fragments::set_annotated_content; use codex_context_fragments::to_annotated_content; use codex_extension_api::ExtensionDataInit; -use codex_features::Feature; use codex_history::ResponseItemEnvelope; use codex_protocol::intersect_effective_permission_profiles; use codex_protocol::protocol::EnvironmentConfigState; @@ -109,7 +109,7 @@ fn keep_forked_rollout_item(item: &RolloutItem, preserve_reference_context_item: fn retain_forked_developer_message( item: &mut ResponseItem, usage_hint_texts: &[String], - features: &crate::config::ManagedFeatures, + context_mode: GuardianContextMode, ) -> bool { if !matches!(item, ResponseItem::Message { role, .. } if role == "developer") { return true; @@ -119,7 +119,7 @@ fn retain_forked_developer_message( return false; }; content.retain(|content_item| { - if features.enabled(Feature::GuardianThreadContext) + if context_mode == GuardianContextMode::ThreadOwned && content_item.kind().0 == "guardian.approved_action" { return false; @@ -129,7 +129,7 @@ fn retain_forked_developer_message( }; !(MultiAgentRoleInstructions::matches_text(text) - || (features.enabled(Feature::GuardianThreadContext) + || (context_mode == GuardianContextMode::ThreadOwned && text.starts_with( crate::guardian::AUTO_REVIEW_DENIED_ACTION_APPROVAL_DEVELOPER_PREFIX, )) @@ -929,12 +929,13 @@ impl AgentControl { break; } } + let context_mode = GuardianContextMode::from_features(&config.features); let mut replaced_parent_developer_instructions = false; // Scrub inherited hints and replace only the parent's developer-instruction fragment. // Compaction stores response items separately, so sanitize both top-level messages and // compacted replacement histories with the same policy. let retain_forked_item = |envelope: &mut ResponseItemEnvelope, replaced: &mut bool| { - if config.features.enabled(Feature::GuardianThreadContext) + if context_mode == GuardianContextMode::ThreadOwned && multi_agent_version == MultiAgentVersion::V2 && matches!(&envelope.item, ResponseItem::Message { role, .. } if role == "user") { @@ -952,7 +953,7 @@ impl AgentControl { if !retain_forked_developer_message( response_item, &multi_agent_v2_usage_hint_texts_to_filter, - &config.features, + context_mode, ) { return false; } @@ -1031,10 +1032,9 @@ impl AgentControl { compacted.guardian_history = None; // Only V2 fetches root authorization live. Its local scope starts known-empty; // V1 must remain incomplete when inherited authorization has been stripped. - compacted.retained_context = - (config.features.enabled(Feature::GuardianThreadContext) - && multi_agent_version == MultiAgentVersion::V2) - .then(codex_history::RetainedContext::default); + compacted.retained_context = (context_mode == GuardianContextMode::ThreadOwned + && multi_agent_version == MultiAgentVersion::V2) + .then(codex_history::RetainedContext::default); if let Some(replacement_history) = compacted.replacement_history.as_mut() { // Matches before this checkpoint cannot survive its replacement history. replaced_parent_developer_instructions = false; diff --git a/codex-rs/core/src/agent/control/user_authorization.rs b/codex-rs/core/src/agent/control/user_authorization.rs index a84f5ccd27..263ab7419a 100644 --- a/codex-rs/core/src/agent/control/user_authorization.rs +++ b/codex-rs/core/src/agent/control/user_authorization.rs @@ -8,6 +8,7 @@ use super::AgentControl; use crate::codex_thread::GuardianRootMessage; use crate::codex_thread::GuardianRootSnapshot; use crate::compact::is_summary_message; +use crate::context::GuardianContextMode; use crate::context::GuardianReviewEvidence; use crate::context::is_contextual_user_fragment; use crate::event_mapping::parse_turn_item; @@ -48,7 +49,9 @@ impl AgentControl { .thread_extension_data .get_or_init(GuardianReviewEvidence::default); let mut latest_user_turn_id = None; - let (messages, authorization_version) = if root_evidence.uses_thread_owned_context() { + let (messages, authorization_version) = if root_evidence.context_mode() + == GuardianContextMode::ThreadOwned + { let mut missing_root_instructions = false; let mut messages = history .retained_context() diff --git a/codex-rs/core/src/compact.rs b/codex-rs/core/src/compact.rs index 796ae1bb22..4aacc97d10 100644 --- a/codex-rs/core/src/compact.rs +++ b/codex-rs/core/src/compact.rs @@ -1,3 +1,4 @@ +use crate::context::GuardianContextMode; use std::sync::Arc; use std::time::Instant; @@ -32,7 +33,6 @@ use codex_analytics::CompactionTrigger; use codex_analytics::now_unix_seconds; use codex_context_fragments::AnnotatedContent; use codex_context_fragments::set_annotated_content; -use codex_features::Feature; use codex_history::CodexHarnessMetadata; use codex_history::ResponseItemEnvelope; use codex_protocol::ResponseItemId; @@ -357,7 +357,7 @@ async fn run_compact_task_inner_impl( let summary_suffix = get_last_assistant_message_from_turn(history_snapshot.raw_items()).unwrap_or_default(); let summary_text = format!("{SUMMARY_PREFIX}\n{summary_suffix}"); - let identity = if sess.enabled(Feature::GuardianThreadContext) { + let identity = if sess.guardian_context_mode == GuardianContextMode::ThreadOwned { CompactedMessageIdentity::Preserve } else { CompactedMessageIdentity::Regenerate diff --git a/codex-rs/core/src/context/guardian_context_mode.rs b/codex-rs/core/src/context/guardian_context_mode.rs new file mode 100644 index 0000000000..0b9cf655ae --- /dev/null +++ b/codex-rs/core/src/context/guardian_context_mode.rs @@ -0,0 +1,23 @@ +//! Immutable session mode shared by history, evidence projection, and reviewer policy. +//! Resolve the rollout flag at session construction, not independently at each consumer. + +use codex_features::Feature; +use codex_features::Features; + +/// Selects legacy compatibility or thread-owned evidence for a session's lifetime. +#[derive(Debug, Default, Clone, Copy, PartialEq, Eq)] +pub enum GuardianContextMode { + #[default] + Legacy, + ThreadOwned, +} + +impl GuardianContextMode { + pub(crate) fn from_features(features: &Features) -> Self { + if features.enabled(Feature::GuardianThreadContext) { + Self::ThreadOwned + } else { + Self::Legacy + } + } +} diff --git a/codex-rs/core/src/context/guardian_review_evidence.rs b/codex-rs/core/src/context/guardian_review_evidence.rs index d36d785a14..4d898845b8 100644 --- a/codex-rs/core/src/context/guardian_review_evidence.rs +++ b/codex-rs/core/src/context/guardian_review_evidence.rs @@ -10,8 +10,6 @@ use std::sync::Mutex; use std::sync::PoisonError; use codex_extension_api::ConversationHistorySnapshot; -use codex_features::Feature; -use codex_features::Features; use codex_protocol::models::ContentItemKind; use codex_protocol::models::ResponseItem; use codex_protocol::protocol::GuardianAssessmentEvent; @@ -20,6 +18,7 @@ use codex_protocol::request_user_input::RequestUserInputResponse; use serde_json::json; use super::ContextualUserFragment; +use super::GuardianContextMode; use crate::codex_thread::GuardianAuthorizationVersion; use crate::codex_thread::GuardianRootMessage; use crate::guardian::guardian_truncate_text; @@ -30,13 +29,6 @@ const MAX_TRUSTED_SKILL_PATHS_BYTES: usize = 2_048; const MAX_GUARDIAN_USER_INPUT_ANSWERS: usize = 8; const MAX_GUARDIAN_USER_INPUT_TOKENS: usize = 900; -#[derive(Debug, Default)] -enum GuardianContextMode { - #[default] - Legacy, - ThreadOwned, -} - /// Selected answer fragments and the authorization state they describe. pub struct GuardianUserInputSnapshot { pub fragments: Vec, @@ -65,17 +57,13 @@ struct GuardianReviewEvidenceState { impl GuardianReviewEvidence { /// Reports the fixed thread mode used for both capture and reviewer policy. - pub fn uses_thread_owned_context(&self) -> bool { - matches!(self.mode, GuardianContextMode::ThreadOwned) + pub fn context_mode(&self) -> GuardianContextMode { + self.mode } - pub(crate) fn from_features(features: &Features) -> Self { + pub(crate) fn new(mode: GuardianContextMode) -> Self { Self { - mode: if features.enabled(Feature::GuardianThreadContext) { - GuardianContextMode::ThreadOwned - } else { - GuardianContextMode::Legacy - }, + mode, state: Mutex::default(), } } diff --git a/codex-rs/core/src/context/mod.rs b/codex-rs/core/src/context/mod.rs index d1ac427700..23b5a97e9f 100644 --- a/codex-rs/core/src/context/mod.rs +++ b/codex-rs/core/src/context/mod.rs @@ -11,6 +11,7 @@ mod developer_instructions; mod environment_context; mod environments_instructions; mod guardian_approved_action; +mod guardian_context_mode; mod guardian_followup_review_reminder; mod guardian_node_repl_policy; mod guardian_policy; @@ -65,6 +66,7 @@ pub(crate) use current_time_reminder::CurrentTimeReminder; pub(crate) use developer_instructions::DeveloperInstructions; pub(crate) use environments_instructions::EnvironmentsInstructions; pub(crate) use guardian_approved_action::GuardianApprovedAction; +pub use guardian_context_mode::GuardianContextMode; pub(crate) use guardian_followup_review_reminder::GuardianFollowupReviewReminder; pub(crate) use guardian_node_repl_policy::GuardianNodeReplPolicy; pub(crate) use guardian_policy::GuardianPolicy; diff --git a/codex-rs/core/src/context_manager/history.rs b/codex-rs/core/src/context_manager/history.rs index 4796f0729e..a6e2eae621 100644 --- a/codex-rs/core/src/context_manager/history.rs +++ b/codex-rs/core/src/context_manager/history.rs @@ -57,6 +57,8 @@ use std::ops::Deref; use std::sync::Arc; use std::sync::LazyLock; +use crate::context::GuardianContextMode; + /// Transcript of thread history #[derive(Debug, Clone, Default)] pub(crate) struct ContextManager { @@ -67,8 +69,8 @@ pub(crate) struct ContextManager { review_history: Option, /// Host facts independent of the model window; snapshots share immutable state. retained_context: Arc, - /// Live and replay instruction capture are enabled together by the session feature flag. - retain_user_messages: bool, + /// Capture, replay, and snapshot selection share the immutable session mode. + guardian_context_mode: GuardianContextMode, /// Bumped whenever history is rewritten, such as compaction or rollback. history_version: u64, /// Monotonic user-input/reset revision, independent of compaction's history generation. @@ -93,7 +95,7 @@ struct SharedConversationHistory { items: Arc>, review_history: Option, retained_context: Arc, - expose_retained_context: bool, + guardian_context_mode: GuardianContextMode, history_version: u64, user_message_revision: u64, } @@ -119,7 +121,7 @@ impl ConversationHistorySnapshot for SharedConversationHistory { } fn retained_context(&self) -> Option<&RetainedContext> { - self.expose_retained_context + (self.guardian_context_mode == GuardianContextMode::ThreadOwned) .then_some(&self.retained_context) } @@ -166,7 +168,7 @@ impl ContextManager { items: Arc::new(Vec::new()), review_history: None, retained_context: Arc::default(), - retain_user_messages: false, + guardian_context_mode: GuardianContextMode::Legacy, history_version: 0, user_message_revision: 0, token_info: TokenUsageInfo::new_or_append( @@ -182,7 +184,7 @@ impl ContextManager { items: Arc::clone(&self.items), review_history: self.review_history.clone(), retained_context: Arc::clone(&self.retained_context), - expose_retained_context: self.retain_user_messages, + guardian_context_mode: self.guardian_context_mode, history_version: self.history_version, user_message_revision: self.user_message_revision, }) @@ -192,8 +194,11 @@ impl ContextManager { &self.retained_context } - pub(crate) fn enable_user_message_retention(&mut self) { - self.retain_user_messages = true; + pub(crate) fn with_guardian_context_mode(guardian_context_mode: GuardianContextMode) -> Self { + Self { + guardian_context_mode, + ..Self::new() + } } pub(crate) fn reserve_input_order(&mut self) -> u64 { @@ -213,7 +218,7 @@ impl ContextManager { } pub(crate) fn guardian_history_checkpoint(&self) -> Option { - if self.retain_user_messages { + if self.guardian_context_mode == GuardianContextMode::ThreadOwned { return None; } self.review_history @@ -227,7 +232,7 @@ impl ContextManager { checkpoint: Option<&GuardianHistoryCheckpoint>, ) { self.restore_retained_context(retained_context); - if self.retain_user_messages { + if self.guardian_context_mode == GuardianContextMode::ThreadOwned { // Older retained checkpoints cleared oversized instructions. Recover their // bounded root excerpts before discarding the legacy source transcript. let items = &self.items; @@ -365,7 +370,7 @@ impl ContextManager { } Arc::make_mut(&mut self.items).push(processed); if crate::context::is_user_authorization_message(item) { - if !self.retain_user_messages { + if self.guardian_context_mode == GuardianContextMode::Legacy { Arc::make_mut(&mut self.retained_context).mark_user_messages_incomplete(); } else if !metadata.is_some_and(|metadata| metadata.inherited_user_message) && let ResponseItem::Message { @@ -528,7 +533,9 @@ impl ContextManager { /// Compaction changes the model's history without changing the user's authorization. pub(crate) fn replace_compacted(&mut self, items: Vec) { - if !self.retain_user_messages && self.review_history.is_none() { + if self.guardian_context_mode == GuardianContextMode::Legacy + && self.review_history.is_none() + { let mut retained = TranscriptHistory::new(self.history_version.saturating_add(1)); for item in self.raw_items().filter(|item| { !matches!(item, ResponseItem::Message { role, content, .. } @@ -627,7 +634,7 @@ impl ContextManager { .iter() .filter_map(|item| item.turn_id()) .collect::>(); - if self.retain_user_messages { + if self.guardian_context_mode == GuardianContextMode::ThreadOwned { Arc::make_mut(&mut retained_context).rollback( &removed_turns, first_removed_message_id, diff --git a/codex-rs/core/src/context_manager/history_tests.rs b/codex-rs/core/src/context_manager/history_tests.rs index 98ad19ef8e..f634b0049e 100644 --- a/codex-rs/core/src/context_manager/history_tests.rs +++ b/codex-rs/core/src/context_manager/history_tests.rs @@ -1269,8 +1269,7 @@ fn drop_last_n_user_turns_preserves_prefix() { // A steered message shares its source turn, but rollback must keep the earlier // instruction and answer as complete evidence, including after the next compaction. - let mut history = ContextManager::default(); - history.enable_user_message_retention(); + let mut history = ContextManager::with_guardian_context_mode(GuardianContextMode::ThreadOwned); let mut expected = None; for (id, text) in [ ("restriction", "Never publish publicly."), diff --git a/codex-rs/core/src/guardian/review.rs b/codex-rs/core/src/guardian/review.rs index 8902d1796b..8b14e449c4 100644 --- a/codex-rs/core/src/guardian/review.rs +++ b/codex-rs/core/src/guardian/review.rs @@ -1,3 +1,4 @@ +use crate::context::GuardianContextMode; use codex_analytics::GuardianApprovalRequestSource; use codex_analytics::GuardianReviewAnalyticsResult; use codex_analytics::GuardianReviewDecision; @@ -627,7 +628,7 @@ pub(super) async fn run_synchronous_review( }, )) .await; - if session.enabled(Feature::GuardianThreadContext) + if session.guardian_context_mode == GuardianContextMode::ThreadOwned && matches!(&outcome, GuardianReviewOutcome::Completed(assessment) if assessment.outcome == GuardianAssessmentOutcome::Allow) && (root_authorization_version != session diff --git a/codex-rs/core/src/guardian/review_session.rs b/codex-rs/core/src/guardian/review_session.rs index 09785fd807..9f7ff85192 100644 --- a/codex-rs/core/src/guardian/review_session.rs +++ b/codex-rs/core/src/guardian/review_session.rs @@ -1,3 +1,7 @@ +#[path = "review_session_context.rs"] +mod context_policy; +use context_policy::ReviewContextPolicy; + use std::collections::HashMap; use std::collections::HashSet; use std::future::Future; @@ -52,6 +56,7 @@ use crate::config::ManagedFeatures; use crate::config::NetworkProxySpec; use crate::config::Permissions; use crate::context::ContextualUserFragment; +use crate::context::GuardianContextMode; use crate::context::GuardianFollowupReviewReminder; use crate::context::GuardianNodeReplPolicy; use crate::context_manager::ContextManager; @@ -228,19 +233,17 @@ impl GuardianReviewSessionReuseKey { spawn_config: &Config, user_instructions: Option, parent_history_version: u64, + context_mode: GuardianContextMode, ) -> Self { Self { root_authorization_version: None, - parent_history_version: if spawn_config - .features - .enabled(Feature::GuardianReuseParentCompaction) - || spawn_config - .features - .enabled(Feature::GuardianThreadContext) - { - parent_history_version - } else { - 0 + parent_history_version: match ReviewContextPolicy::for_context( + context_mode, + &spawn_config.features, + ) { + ReviewContextPolicy::Legacy => 0, + ReviewContextPolicy::LegacyWithCheckpointReuse + | ReviewContextPolicy::ThreadOwned => parent_history_version, }, node_repl_auto_review_required: false, node_repl_policy: String::new(), @@ -287,64 +290,6 @@ impl GuardianReviewSessionReuseKey { } } -fn encrypted_parent_compaction( - history: &ContextManager, - features: &ManagedFeatures, - reviewer_compaction_hash: Option<&str>, -) -> anyhow::Result> { - let strict = features.enabled(Feature::GuardianThreadContext); - if !strict && !features.enabled(Feature::GuardianReuseParentCompaction) { - return Ok(None); - } - let Some(envelope) = history.annotated_items().iter().rev().find(|envelope| { - matches!( - envelope.item, - ResponseItem::Compaction { .. } | ResponseItem::ContextCompaction { .. } - ) - }) else { - return Ok(None); - }; - - let item = &envelope.item; - let valid = match item { - ResponseItem::Compaction { - id: Some(_), - encrypted_content, - .. - } if !encrypted_content.is_empty() => true, - ResponseItem::ContextCompaction { - id: Some(_), - encrypted_content: Some(encrypted_content), - .. - } if !encrypted_content.is_empty() => true, - _ => false, - }; - if !valid && !strict { - return Ok(None); - } - anyhow::ensure!( - valid, - "parent compaction checkpoint is unusable for Guardian review" - ); - if strict { - // A resumed parent may now use a different model. Compare the actual - // checkpoint producer with the selected reviewer, not the live parent model. - let producer_hash = envelope - .metadata - .as_ref() - .and_then(|metadata| metadata.compaction_model_hash.as_deref()); - anyhow::ensure!( - producer_hash - .zip(reviewer_compaction_hash) - .is_some_and(|(producer, reviewer)| { - !producer.is_empty() && producer == reviewer - }), - "parent compaction checkpoint is incompatible with the Guardian review model or its compatibility is unknown" - ); - } - Ok(Some(item.clone())) -} - pub(crate) fn prompt_cache_key_override_for_review_session( session_source: &SessionSource, parent_thread_id: Option, @@ -473,20 +418,15 @@ impl GuardianReviewSessionManager { guardian_review_session_config(&parent_session, &parent_turn).await?; let spawn_config = session_config.spawn_config; let parent_history = parent_session.clone_history().await; - let root_authorization_version = - if parent_session.enabled(Feature::GuardianThreadContext) { - parent_session - .services - .agent_control - .root_user_authorization(parent_session.thread_id) - .await - .map(|snapshot| snapshot.authorization_version) - } else { - None - }; - let parent_compaction = encrypted_parent_compaction( - &parent_history, + let context_policy = ReviewContextPolicy::for_context( + parent_session.guardian_context_mode, &spawn_config.features, + ); + let root_authorization_version = context_policy + .root_authorization_version(&parent_session) + .await; + let parent_compaction = context_policy.parent_compaction( + &parent_history, session_config.compaction_model_hash.as_deref(), )?; let parent_context = GuardianReviewContext::from(parent_turn); @@ -494,6 +434,7 @@ impl GuardianReviewSessionManager { &spawn_config, parent_session.user_instructions().await, parent_history.history_version(), + parent_session.guardian_context_mode, ) .with_environments(parent_context.environments()) .with_node_repl_policy_eligibility( @@ -575,25 +516,16 @@ impl GuardianReviewSessionManager { ) -> (GuardianReviewSessionOutcome, GuardianReviewAnalyticsResult) { let deadline = params.deadline; let parent_history = ¶ms.parent_history; - let root_authorization_version = if params - .parent_session - .enabled(Feature::GuardianThreadContext) - { - params - .parent_session - .services - .agent_control - .root_user_authorization(params.parent_session.thread_id) - .await - .map(|snapshot| snapshot.authorization_version) - } else { - None - }; - let parent_compaction = match encrypted_parent_compaction( - parent_history, + let context_policy = ReviewContextPolicy::for_context( + params.parent_session.guardian_context_mode, ¶ms.spawn_config.features, - params.compaction_model_hash.as_deref(), - ) { + ); + let root_authorization_version = context_policy + .root_authorization_version(¶ms.parent_session) + .await; + let parent_compaction = match context_policy + .parent_compaction(parent_history, params.compaction_model_hash.as_deref()) + { Ok(checkpoint) => checkpoint, Err(error) => { return ( @@ -606,6 +538,7 @@ impl GuardianReviewSessionManager { ¶ms.spawn_config, params.parent_session.user_instructions().await, parent_history.history_version(), + params.parent_session.guardian_context_mode, ) .with_environments(params.parent_context.environments()) .with_node_repl_policy_eligibility( @@ -626,9 +559,7 @@ impl GuardianReviewSessionManager { .await { Ok(mut state) => { - if !params - .parent_session - .enabled(Feature::GuardianThreadContext) + if context_policy != ReviewContextPolicy::ThreadOwned && parent_compaction.is_none() && let Some(trunk) = state.trunk.as_ref() { @@ -752,6 +683,7 @@ impl GuardianReviewSessionManager { session.get_config().await.as_ref(), session.user_instructions().await, session.clone_history().await.history_version(), + session.guardian_context_mode, ); self.state.lock().await.trunk = Some(Arc::new(GuardianReviewSession { reuse_key, @@ -775,6 +707,7 @@ impl GuardianReviewSessionManager { session.get_config().await.as_ref(), session.user_instructions().await, session.clone_history().await.history_version(), + session.guardian_context_mode, ); self.state .lock() @@ -1129,9 +1062,8 @@ async fn run_review_on_session( .sync_session_approved_hosts_to(&review_session.session.services.network_approval) .await; - let history = if params - .parent_session - .enabled(Feature::GuardianThreadContext) + let history = if params.parent_session.guardian_context_mode + == GuardianContextMode::ThreadOwned { params.parent_history.conversation_history_snapshot() } else { diff --git a/codex-rs/core/src/guardian/review_session_context.rs b/codex-rs/core/src/guardian/review_session_context.rs new file mode 100644 index 0000000000..f1fa01536d --- /dev/null +++ b/codex-rs/core/src/guardian/review_session_context.rs @@ -0,0 +1,105 @@ +//! Owns sync reviewer checkpoint and invalidation policy for both context modes. +//! Legacy may keep its existing transcript; thread-owned mode requires current parent context. + +use codex_features::Feature; +use codex_protocol::models::ResponseItem; + +use crate::codex_thread::GuardianAuthorizationVersion; +use crate::config::ManagedFeatures; +use crate::context::GuardianContextMode; +use crate::context_manager::ContextManager; +use crate::session::session::Session; + +#[derive(Clone, Copy, PartialEq, Eq)] +pub(super) enum ReviewContextPolicy { + Legacy, + LegacyWithCheckpointReuse, + ThreadOwned, +} + +impl ReviewContextPolicy { + pub(super) fn for_context(mode: GuardianContextMode, features: &ManagedFeatures) -> Self { + match mode { + GuardianContextMode::ThreadOwned => Self::ThreadOwned, + GuardianContextMode::Legacy + if features.enabled(Feature::GuardianReuseParentCompaction) => + { + Self::LegacyWithCheckpointReuse + } + GuardianContextMode::Legacy => Self::Legacy, + } + } + + pub(super) async fn root_authorization_version( + self, + session: &Session, + ) -> Option { + if self != Self::ThreadOwned { + return None; + } + session + .services + .agent_control + .root_user_authorization(session.thread_id) + .await + .map(|snapshot| snapshot.authorization_version) + } + + pub(super) fn parent_compaction( + self, + history: &ContextManager, + reviewer_compaction_hash: Option<&str>, + ) -> anyhow::Result> { + let strict = self == Self::ThreadOwned; + if self == Self::Legacy { + return Ok(None); + } + let Some(envelope) = history.annotated_items().iter().rev().find(|envelope| { + matches!( + envelope.item, + ResponseItem::Compaction { .. } | ResponseItem::ContextCompaction { .. } + ) + }) else { + return Ok(None); + }; + + let item = &envelope.item; + let valid = match item { + ResponseItem::Compaction { + id: Some(_), + encrypted_content, + .. + } if !encrypted_content.is_empty() => true, + ResponseItem::ContextCompaction { + id: Some(_), + encrypted_content: Some(encrypted_content), + .. + } if !encrypted_content.is_empty() => true, + _ => false, + }; + if !valid && !strict { + return Ok(None); + } + anyhow::ensure!( + valid, + "parent compaction checkpoint is unusable for Guardian review" + ); + if strict { + // A resumed parent may now use a different model. Compare the actual + // checkpoint producer with the selected reviewer, not the live parent model. + let producer_hash = envelope + .metadata + .as_ref() + .and_then(|metadata| metadata.compaction_model_hash.as_deref()); + anyhow::ensure!( + producer_hash + .zip(reviewer_compaction_hash) + .is_some_and(|(producer, reviewer)| { + !producer.is_empty() && producer == reviewer + }), + "parent compaction checkpoint is incompatible with the Guardian review model or its compatibility is unknown" + ); + } + Ok(Some(item.clone())) + } +} diff --git a/codex-rs/core/src/guardian/review_session_tests.rs b/codex-rs/core/src/guardian/review_session_tests.rs index 62a6e96de3..7f8558e2d5 100644 --- a/codex-rs/core/src/guardian/review_session_tests.rs +++ b/codex-rs/core/src/guardian/review_session_tests.rs @@ -1,4 +1,5 @@ use super::*; +use crate::context_manager::ContextManager; use codex_history::CodexHarnessMetadata; use codex_history::ResponseItemEnvelope; use codex_protocol::openai_models::AutoReviewMessages; @@ -54,6 +55,7 @@ async fn run_review_preserves_evidence_during_parent_compaction() { ¶ms.spawn_config, parent.user_instructions().await, params.parent_history.history_version(), + parent.guardian_context_mode, ) .with_environments(params.parent_context.environments()) .with_node_repl_policy_eligibility( @@ -146,6 +148,7 @@ async fn test_review_session() -> ( session.get_config().await.as_ref(), session.user_instructions().await, session.clone_history().await.history_version(), + GuardianContextMode::Legacy, ); ( @@ -298,6 +301,7 @@ async fn guardian_review_session_config_change_invalidates_cached_session() { &cached_spawn_config, /*user_instructions*/ None, /*parent_history_version*/ 0, + GuardianContextMode::Legacy, ); let mut changed_parent_config = parent_config; @@ -315,6 +319,7 @@ async fn guardian_review_session_config_change_invalidates_cached_session() { &next_spawn_config, /*user_instructions*/ None, /*parent_history_version*/ 0, + GuardianContextMode::Legacy, ); assert_eq!( @@ -328,6 +333,7 @@ async fn guardian_review_session_config_change_invalidates_cached_session() { &cached_spawn_config, /*user_instructions*/ None, /*parent_history_version*/ 0, + GuardianContextMode::Legacy, ) ); @@ -337,6 +343,7 @@ async fn guardian_review_session_config_change_invalidates_cached_session() { &cached_spawn_config, /*user_instructions*/ None, /*parent_history_version*/ 1, + GuardianContextMode::Legacy, ) ); assert_ne!( @@ -372,11 +379,13 @@ async fn guardian_review_session_config_change_invalidates_cached_session() { &compaction_enabled_config, /*user_instructions*/ None, /*parent_history_version*/ 0, + GuardianContextMode::Legacy, ), GuardianReviewSessionReuseKey::from_spawn_config( &compaction_enabled_config, /*user_instructions*/ None, /*parent_history_version*/ 1, + GuardianContextMode::Legacy, ) ); } @@ -393,6 +402,8 @@ async fn encrypted_parent_compaction_requires_original_item_id(thread_context_en features .set_enabled(Feature::GuardianThreadContext, thread_context_enabled) .expect("context mode"); + let policy = + ReviewContextPolicy::for_context(GuardianContextMode::from_features(&features), &features); let item = ResponseItem::Compaction { id: Some(codex_protocol::ResponseItemId::from_server( "cmp_guardian_parent_summary".to_string(), @@ -410,7 +421,8 @@ async fn encrypted_parent_compaction_requires_original_item_id(thread_context_en }), }]); assert_eq!( - encrypted_parent_compaction(&history, &features, Some("compatible")) + policy + .parent_compaction(&history, Some("compatible")) .expect("valid checkpoint"), Some(item) ); @@ -425,7 +437,7 @@ async fn encrypted_parent_compaction_requires_original_item_id(thread_context_en .into(), ); history.replace_annotated(items); - let result = encrypted_parent_compaction(&history, &features, Some("compatible")); + let result = policy.parent_compaction(&history, Some("compatible")); if thread_context_enabled { assert!(result.is_err()); } else { @@ -484,6 +496,7 @@ async fn guardian_review_session_compact_scope_change_invalidates_cached_session &cached_spawn_config, /*user_instructions*/ None, /*parent_history_version*/ 0, + GuardianContextMode::Legacy, ); let mut changed_parent_config = parent_config; @@ -501,6 +514,7 @@ async fn guardian_review_session_compact_scope_change_invalidates_cached_session &next_spawn_config, /*user_instructions*/ None, /*parent_history_version*/ 0, + GuardianContextMode::Legacy, ); assert_ne!(cached_reuse_key, next_reuse_key); @@ -897,6 +911,7 @@ async fn run_review_removes_trunk_when_event_stream_is_broken() { .clone_history() .await .history_version(), + GuardianContextMode::Legacy, ) .with_environments(params.parent_context.environments()) .with_node_repl_policy(¶ms.node_repl_policy); diff --git a/codex-rs/core/src/session/mod.rs b/codex-rs/core/src/session/mod.rs index 849f07c935..8ffc144d02 100644 --- a/codex-rs/core/src/session/mod.rs +++ b/codex-rs/core/src/session/mod.rs @@ -1,3 +1,4 @@ +use crate::context::GuardianContextMode; use std::borrow::Cow; use std::collections::HashMap; use std::collections::HashSet; @@ -3777,7 +3778,7 @@ impl Session { for envelope in &mut items { Self::assign_missing_response_item_id(&mut envelope.item); } - if self.enabled(Feature::GuardianThreadContext) + if self.guardian_context_mode == GuardianContextMode::ThreadOwned && let Some(checkpoint) = items.iter_mut().rev().find(|envelope| { matches!( envelope.item, diff --git a/codex-rs/core/src/session/retained_context.rs b/codex-rs/core/src/session/retained_context.rs index 71506e68f4..8d1eb3e615 100644 --- a/codex-rs/core/src/session/retained_context.rs +++ b/codex-rs/core/src/session/retained_context.rs @@ -1,6 +1,6 @@ //! Orders retained inputs and records host-verified facts at the context checkpoint boundary. -use codex_features::Feature; +use crate::context::GuardianContextMode; use codex_history::RetainedContextEvent; use codex_history::RolloutItem; @@ -10,7 +10,7 @@ use super::thread_settings; impl Session { /// Legacy mode neither reserves a sequence nor takes the session-state lock. pub(crate) async fn reserve_user_input_order(&self) -> Option { - if !self.enabled(Feature::GuardianThreadContext) { + if self.guardian_context_mode == GuardianContextMode::Legacy { return None; } Some(self.state.lock().await.history.reserve_input_order()) diff --git a/codex-rs/core/src/session/rollout_reconstruction.rs b/codex-rs/core/src/session/rollout_reconstruction.rs index b9ce27b7c3..4ee1497ba1 100644 --- a/codex-rs/core/src/session/rollout_reconstruction.rs +++ b/codex-rs/core/src/session/rollout_reconstruction.rs @@ -1,4 +1,5 @@ use super::*; +use crate::context::GuardianContextMode; use crate::context::world_state::WorldStateSnapshot; use crate::context_manager::is_user_turn_boundary; use codex_history::ResponseItemEnvelope; @@ -338,10 +339,7 @@ impl Session { ) .unwrap_or(u64::MAX); - let mut history = ContextManager::new(); - if self.enabled(Feature::GuardianThreadContext) { - history.enable_user_message_retention(); - } + let mut history = ContextManager::with_guardian_context_mode(self.guardian_context_mode); let mut saw_legacy_compaction_without_replacement_history = false; if let Some(checkpoint) = base_compaction && let Some(items) = &checkpoint.compacted.replacement_history @@ -389,11 +387,12 @@ impl Session { // prompt shape. // TODO(ccunningham): if we drop support for None replacement_history compaction items, // we can get rid of this second loop entirely and just build `history` directly in the first loop. - let identity = if self.enabled(Feature::GuardianThreadContext) { - compact::CompactedMessageIdentity::Preserve - } else { - compact::CompactedMessageIdentity::Regenerate - }; + let identity = + if self.guardian_context_mode == GuardianContextMode::ThreadOwned { + compact::CompactedMessageIdentity::Preserve + } else { + compact::CompactedMessageIdentity::Regenerate + }; let user_messages = compact::collect_annotated_user_messages( history.annotated_items(), identity, diff --git a/codex-rs/core/src/session/session.rs b/codex-rs/core/src/session/session.rs index 29bda89d2d..b283528024 100644 --- a/codex-rs/core/src/session/session.rs +++ b/codex-rs/core/src/session/session.rs @@ -7,6 +7,7 @@ use super::step_settings::StepSettingsUpdate; use super::*; use crate::agents_md_manager::AgentsMdManager; use crate::config::ConstraintError; +use crate::context::GuardianContextMode; use crate::environment_selection::ThreadEnvironments; use crate::environment_selection::TurnEnvironmentSnapshot; use crate::hook_mcp_executor::CoreHookMcpExecutor; @@ -54,6 +55,7 @@ pub(crate) struct Session { /// The set of enabled features should be invariant for the lifetime of the /// session. pub(super) features: ManagedFeatures, + pub(crate) guardian_context_mode: GuardianContextMode, pub(crate) windows_sandbox_proxy_settings_mode: codex_sandboxing::WindowsSandboxProxySettingsMode, pub(super) multi_agent_version: OnceLock, @@ -840,9 +842,10 @@ impl Session { thread_id.to_string(), thread_extension_init, ); - // Select the answer path once, before extensions or tool handlers observe the thread. - thread_extension_data.insert(crate::context::GuardianReviewEvidence::from_features( - &config.features, + // Resolve once for live history, replay, and all reviewer consumers. + let guardian_context_mode = GuardianContextMode::from_features(&config.features); + thread_extension_data.insert(crate::context::GuardianReviewEvidence::new( + guardian_context_mode, )); // Kick off independent async setup tasks in parallel to reduce startup latency. // @@ -1264,10 +1267,8 @@ impl Session { let mut state = SessionState::new_with_auto_compact_window_ids( session_configuration.clone(), initial_auto_compact_window_ids, + guardian_context_mode, ); - if config.features.enabled(Feature::GuardianThreadContext) { - state.history.enable_user_message_retention(); - } state.base_instructions_provenance = base_instructions_provenance.clone(); let managed_network_requirements_configured = config .config_layer_stack @@ -1505,6 +1506,7 @@ impl Session { thread_settings_persistence: Semaphore::new(/*permits*/ 1), managed_network_proxy_refresh_lock: Semaphore::new(/*permits*/ 1), features: config.features.clone(), + guardian_context_mode, windows_sandbox_proxy_settings_mode, multi_agent_version, mcp_refresh: McpRefresh::new(), diff --git a/codex-rs/core/src/session/tests.rs b/codex-rs/core/src/session/tests.rs index 79dc4f069f..9e4f5992c3 100644 --- a/codex-rs/core/src/session/tests.rs +++ b/codex-rs/core/src/session/tests.rs @@ -12,6 +12,7 @@ use crate::config::ConfigOverrides; use crate::config::test_config; use crate::context::ContextualUserFragment; use crate::context::DeveloperInstructions; +use crate::context::GuardianContextMode; use crate::context::TurnAborted; use crate::environment_selection::EnvironmentConfigOrigin; use crate::environment_selection::ThreadEnvironments; @@ -6620,6 +6621,7 @@ pub(crate) async fn make_session_and_context() -> (Session, TurnContext) { thread_settings_persistence: Semaphore::new(/*permits*/ 1), managed_network_proxy_refresh_lock: Semaphore::new(/*permits*/ 1), features: config.features.clone(), + guardian_context_mode: GuardianContextMode::from_features(&config.features), windows_sandbox_proxy_settings_mode: codex_sandboxing::WindowsSandboxProxySettingsMode::Reconcile, multi_agent_version: OnceLock::from(config.multi_agent_version_from_features()), @@ -8784,9 +8786,9 @@ where ); let mut state = SessionState::new(session_configuration.clone()); - if config.features.enabled(Feature::GuardianThreadContext) { - state.history.enable_user_message_retention(); - } + state.history = ContextManager::with_guardian_context_mode(GuardianContextMode::from_features( + &config.features, + )); let (environment_manager, resolved_turn_environments) = resolved_environments_for_configuration(&session_configuration, &default_environments) .await; @@ -8925,6 +8927,7 @@ where thread_settings_persistence: Semaphore::new(/*permits*/ 1), managed_network_proxy_refresh_lock: Semaphore::new(/*permits*/ 1), features: config.features.clone(), + guardian_context_mode: GuardianContextMode::from_features(&config.features), windows_sandbox_proxy_settings_mode: codex_sandboxing::WindowsSandboxProxySettingsMode::Reconcile, multi_agent_version: OnceLock::from(config.multi_agent_version_from_features()), diff --git a/codex-rs/core/src/state/session.rs b/codex-rs/core/src/state/session.rs index e9e42cf00a..d4d04d37aa 100644 --- a/codex-rs/core/src/state/session.rs +++ b/codex-rs/core/src/state/session.rs @@ -12,6 +12,7 @@ use super::AdditionalContextStore; use super::auto_compact_window::AutoCompactWindow; use super::auto_compact_window::AutoCompactWindowIds; use super::auto_compact_window::AutoCompactWindowSnapshot; +use crate::context::GuardianContextMode; use crate::context_manager::ContextManager; use crate::context_manager::HistoryReplacement; use crate::session::PreviousTurnSettings; @@ -64,14 +65,16 @@ impl SessionState { Self::new_with_auto_compact_window_ids( session_configuration, AutoCompactWindowIds::new_initial(), + GuardianContextMode::Legacy, ) } pub(crate) fn new_with_auto_compact_window_ids( session_configuration: SessionConfiguration, auto_compact_window_ids: AutoCompactWindowIds, + guardian_context_mode: GuardianContextMode, ) -> Self { - let history = ContextManager::new(); + let history = ContextManager::with_guardian_context_mode(guardian_context_mode); Self { session_configuration, base_instructions_provenance: None, diff --git a/codex-rs/core/src/tools/handlers/request_user_input.rs b/codex-rs/core/src/tools/handlers/request_user_input.rs index a00781ed24..95b83eca64 100644 --- a/codex-rs/core/src/tools/handlers/request_user_input.rs +++ b/codex-rs/core/src/tools/handlers/request_user_input.rs @@ -1,3 +1,4 @@ +use crate::context::GuardianContextMode; use crate::context::GuardianReviewEvidence; use crate::function_tool::FunctionCallError; use crate::tools::context::FunctionToolOutput; @@ -111,7 +112,7 @@ impl RequestUserInputHandler { // Capture and consumption use the same fixed thread feature setting. Legacy // threads must not construct retained answers, persist them, or advance their revision. if turn.config.features.enabled(Feature::GuardianApproval) - && session.enabled(Feature::GuardianThreadContext) + && session.guardian_context_mode == GuardianContextMode::ThreadOwned { let user_input = questions .iter() diff --git a/codex-rs/core/src/tools/handlers/request_user_input_tests.rs b/codex-rs/core/src/tools/handlers/request_user_input_tests.rs index ddf261983f..96bc27c43c 100644 --- a/codex-rs/core/src/tools/handlers/request_user_input_tests.rs +++ b/codex-rs/core/src/tools/handlers/request_user_input_tests.rs @@ -117,9 +117,7 @@ async fn request_user_input_sets_non_blocking_outside_plan_mode( session .services .thread_extension_data - .insert(GuardianReviewEvidence::from_features( - session.features().get(), - )); + .insert(GuardianReviewEvidence::new(session.guardian_context_mode)); let original_history = session.conversation_history_snapshot().await; *session.active_turn.lock().await = Some(ActiveTurn::default()); diff --git a/codex-rs/ext/guardian-v2/src/async_scorer/approval.rs b/codex-rs/ext/guardian-v2/src/async_scorer/approval.rs index d783468db3..fb96cbcac4 100644 --- a/codex-rs/ext/guardian-v2/src/async_scorer/approval.rs +++ b/codex-rs/ext/guardian-v2/src/async_scorer/approval.rs @@ -5,13 +5,13 @@ use super::authorization::ScoreAuthorization; use super::config::GuardianV2Config; use super::coverage::GuardianPolicy; use super::extension::GuardianV2ScoreProgress; -use super::extension::encrypted_parent_compaction; -use super::extension::requires_sync_for_compaction; use super::metrics::TOOL_CALL_LAG_METRIC; use super::metrics::record_fast_decision; +use super::parent_compaction::select_parent_compaction; use super::sampler::LunaSampler; use codex_core::CodexThread; use codex_core::ThreadManager; +use codex_core::context::GuardianContextMode; use codex_core::context::GuardianReviewEvidence; use codex_extension_api::ApprovalDecision; use codex_extension_api::ApprovalDecisionInput; @@ -132,17 +132,22 @@ async fn cached_evidence( record_fast_decision(metrics, "deferred", "missing_score"); return Err(GuardianReviewReason::MissingScore); }; - if store + let context_mode = store .get_or_init(GuardianReviewEvidence::default) - .uses_thread_owned_context() - { + .context_mode(); + if context_mode == GuardianContextMode::ThreadOwned { let sampler = store .get::() .ok_or(GuardianReviewReason::MissingScore)?; let history = thread.conversation_history_snapshot().await; - if requires_sync_for_compaction(config, history.as_ref(), &sampler) - || encrypted_parent_compaction(history.items(), config.max_parent_compaction_tokens) - .is_err() + if select_parent_compaction( + context_mode, + config, + history.as_ref(), + &sampler, + /*legacy_model_hash*/ None, + ) + .is_err() { record_fast_decision(metrics, "deferred", "incompatible_compaction"); return Err(GuardianReviewReason::IncompatibleCompaction); diff --git a/codex-rs/ext/guardian-v2/src/async_scorer/extension.rs b/codex-rs/ext/guardian-v2/src/async_scorer/extension.rs index 34bb6e5b6a..428aff3097 100644 --- a/codex-rs/ext/guardian-v2/src/async_scorer/extension.rs +++ b/codex-rs/ext/guardian-v2/src/async_scorer/extension.rs @@ -12,10 +12,10 @@ use codex_analytics::GuardianV2Event; use codex_analytics::GuardianV2EventKind; use codex_core::ThreadManager; use codex_core::config::Config; +use codex_core::context::GuardianContextMode; use codex_core::context::GuardianReviewEvidence; use codex_core::context::NodeReplReviewEvidence; use codex_extension_api::ApprovalReviewContributor; -use codex_extension_api::ConversationHistorySnapshot; use codex_extension_api::ExtensionData; use codex_extension_api::ExtensionEventSink; use codex_extension_api::ExtensionFuture; @@ -23,7 +23,6 @@ use codex_extension_api::ExtensionMetrics; use codex_extension_api::ExtensionRegistryBuilder; use codex_extension_api::ExtensionWarning; use codex_extension_api::GuardianV2Enabled; -use codex_extension_api::ResponseItem; use codex_extension_api::SkillInvocationContributor; use codex_extension_api::SkillInvocationInput; use codex_extension_api::ThreadLifecycleContributor; @@ -45,7 +44,6 @@ use codex_protocol::openai_models::GuardianReviewMode; use codex_protocol::openai_models::GuardianScope; use codex_protocol::openai_models::ModelInfo; use codex_protocol::protocol::ReviewDecision; -use codex_protocol::protocol::TruncationPolicy; use codex_protocol::protocol::has_full_access; use codex_protocol::security_risk::SecurityRiskScore; @@ -61,6 +59,8 @@ use super::metrics::record_classification; use super::metrics::record_classification_risk; use super::metrics::record_fast_decision; use super::metrics::sampler_failure_reason; +use super::parent_compaction::ParentCompactionError; +use super::parent_compaction::select_parent_compaction; use super::review_evidence::render_review_evidence; use super::sampler::LunaSampler; use super::sampler::LunaSamplerConfig; @@ -263,7 +263,7 @@ impl ApprovalReviewContributor for GuardianV2Extension { return None; } let guardian_evidence = thread_store.get_or_init(GuardianReviewEvidence::default); - let thread_context_enabled = guardian_evidence.uses_thread_owned_context(); + let context_mode = guardian_evidence.context_mode(); let mut initial_cua_call = false; if scope == Some(GuardianScope::ComputerUse) && policy.initial_cua_call { // Thread-owned context checks checkpoint compatibility before the first CUA allowance. @@ -278,7 +278,7 @@ impl ApprovalReviewContributor for GuardianV2Extension { .js_executions .load(Ordering::Acquire) == 1; - if initial_cua_call && !thread_context_enabled { + if initial_cua_call && context_mode == GuardianContextMode::Legacy { record_fast_decision( extension_metrics.as_deref(), "approved", @@ -323,14 +323,16 @@ impl ApprovalReviewContributor for GuardianV2Extension { return None; }; let history = thread.conversation_history_snapshot().await; - if thread_context_enabled { + if context_mode == GuardianContextMode::ThreadOwned { let sampler = thread_store.get::()?; - if requires_sync_for_compaction(&guardian_config, history.as_ref(), &sampler) - || encrypted_parent_compaction( - history.items(), - guardian_config.max_parent_compaction_tokens, - ) - .is_err() + if select_parent_compaction( + context_mode, + &guardian_config, + history.as_ref(), + &sampler, + /*legacy_model_hash*/ None, + ) + .is_err() { thread_store.insert(StrictReviewReason::IncompatibleCompaction); record_fast_decision( @@ -608,64 +610,38 @@ impl GuardianV2Extension { let guardian_evidence = input .thread_store .get_or_init(GuardianReviewEvidence::default); - let thread_context_enabled = guardian_evidence.uses_thread_owned_context(); - if thread_context_enabled - && requires_sync_for_compaction( - &guardian_config, - input.conversation_history.as_ref(), - &sampler, - ) - { - score_progress - .latest_failed_tool_call - .fetch_max(tool_call_index, Ordering::Release); - Self::record_fail_closed_score(input.thread_store, sampled_at); - record_classification( - metrics.as_deref(), - classification_started_at.elapsed(), - "skipped", - /*failure_reason*/ None, - ); - return; - } - let parent_compaction_hash = if thread_context_enabled { - input - .conversation_history - .latest_compaction_model_hash() - .map(str::to_owned) - } else { + let context_mode = guardian_evidence.context_mode(); + let selected_compaction = match select_parent_compaction( + context_mode, + &guardian_config, + input.conversation_history.as_ref(), + &sampler, parent_model .as_ref() - .and_then(|model| model.comp_hash.clone()) - }; - let parent_compaction = if guardian_config.reuse_parent_compaction { - match encrypted_parent_compaction( - input.conversation_history.items(), - guardian_config.max_parent_compaction_tokens, - ) { - Ok(compaction) => compaction, - Err(ParentCompactionError::Unusable) if !thread_context_enabled => None, - Err(_) => { - Self::record_fail_closed_score(input.thread_store, sampled_at); - record_classification( - metrics.as_deref(), - classification_started_at.elapsed(), - "failure", - Some("parent_compaction_error"), - ); - return; - } + .and_then(|model| model.comp_hash.as_deref()), + ) { + Ok(compaction) => compaction, + Err(error) => { + let (outcome, failure_reason) = if error == ParentCompactionError::RequiresSync { + score_progress + .latest_failed_tool_call + .fetch_max(tool_call_index, Ordering::Release); + ("skipped", None) + } else { + ("failure", Some("parent_compaction_error")) + }; + Self::record_fail_closed_score(input.thread_store, sampled_at); + record_classification( + metrics.as_deref(), + classification_started_at.elapsed(), + outcome, + failure_reason, + ); + return; } - } else { - None }; - // Legacy requests may omit an incompatible checkpoint because their raw - // review transcript is still retained. The sampler rejects supplied - // incompatible checkpoints, so preserve that legacy omission here. - let parent_compaction = parent_compaction.filter(|_| { - thread_context_enabled - || sampler.supports_parent_compaction(parent_compaction_hash.as_deref()) - }); + let parent_compaction = selected_compaction.item; + let parent_compaction_hash = selected_compaction.model_hash; let call_id = input.call_id.to_owned(); let mcp_tool = input.mcp_tool.cloned(); let action = GuardianAction { @@ -697,7 +673,7 @@ impl GuardianV2Extension { .images(input.conversation_history.review_items(), node_repl_images); // Capture root evidence before background metadata resolution or model I/O. // Later root changes invalidate this sample through its captured authorization version. - let root_snapshot = if thread_context_enabled { + let root_snapshot = if context_mode == GuardianContextMode::ThreadOwned { thread.guardian_root_snapshot().await } else { None @@ -712,7 +688,7 @@ impl GuardianV2Extension { } None => None, }; - let root_snapshot = if thread_context_enabled { + let root_snapshot = if context_mode == GuardianContextMode::ThreadOwned { root_snapshot } else { thread.guardian_root_snapshot().await @@ -986,71 +962,6 @@ impl GuardianV2Extension { } } -#[derive(Debug, Eq, PartialEq)] -pub(super) enum ParentCompactionError { - Serialization, - Oversized, - Unusable, -} - -// Sampling and fast approval must apply the same checkpoint eligibility policy. -pub(super) fn requires_sync_for_compaction( - config: &GuardianV2Config, - history: &dyn ConversationHistorySnapshot, - sampler: &LunaSampler, -) -> bool { - history.items().any(|item| { - matches!( - item, - ResponseItem::Compaction { .. } | ResponseItem::ContextCompaction { .. } - ) - }) && (!config.reuse_parent_compaction - || !sampler.supports_parent_compaction(history.latest_compaction_model_hash())) -} - -// An unusable latest compaction must never fall back to an older one. Missing -// encrypted content is rejected here; only legacy callers may omit that checkpoint. -pub(super) fn encrypted_parent_compaction<'a>( - items: impl Iterator, - max_parent_compaction_tokens: usize, -) -> Result, ParentCompactionError> { - let max_compaction_bytes = TruncationPolicy::Tokens(max_parent_compaction_tokens).byte_budget(); - let Some(item) = items - .filter(|item| { - matches!( - item, - ResponseItem::Compaction { .. } | ResponseItem::ContextCompaction { .. } - ) - }) - .last() - else { - return Ok(None); - }; - - let encrypted_content = match item { - ResponseItem::Compaction { - id: Some(_), - encrypted_content, - .. - } - | ResponseItem::ContextCompaction { - id: Some(_), - encrypted_content: Some(encrypted_content), - .. - } => encrypted_content, - _ => return Err(ParentCompactionError::Unusable), - }; - if encrypted_content.is_empty() { - return Err(ParentCompactionError::Unusable); - } - let serialized = serde_json::to_vec(item).map_err(|_| ParentCompactionError::Serialization)?; - if serialized.len() > max_compaction_bytes { - return Err(ParentCompactionError::Oversized); - } - - Ok(Some(item.clone())) -} - /// Installs feature-gated Guardian V2 tool classification for each thread. pub fn install( registry: &mut ExtensionRegistryBuilder, diff --git a/codex-rs/ext/guardian-v2/src/async_scorer/extension_tests.rs b/codex-rs/ext/guardian-v2/src/async_scorer/extension_tests.rs index acbbc6161a..a15355c656 100644 --- a/codex-rs/ext/guardian-v2/src/async_scorer/extension_tests.rs +++ b/codex-rs/ext/guardian-v2/src/async_scorer/extension_tests.rs @@ -69,9 +69,7 @@ use serde_json::json; use super::GuardianV2Extension; use super::GuardianV2ScoreProgress; -use super::ParentCompactionError; use super::StrictReviewReason; -use super::encrypted_parent_compaction; use crate::async_scorer::config::CLASSIFICATION_OUTPUT_INSTRUCTIONS; use crate::async_scorer::config::DEFAULT_MODEL_CONTEXT_ITEM_TOKENS; @@ -791,166 +789,6 @@ async fn computer_use_only_scores_cannot_approve_other_actions() -> Result<()> { Ok(()) } -#[test] -fn encrypted_parent_compaction_preserves_the_latest_valid_item() { - let older = ResponseItem::Compaction { - id: Some(ResponseItemId::from_server("cmp_older".to_owned())), - encrypted_content: "older encrypted summary".to_owned(), - internal_chat_message_metadata_passthrough: None, - }; - let latest = ResponseItem::ContextCompaction { - id: Some(ResponseItemId::from_server("cmp_latest".to_owned())), - encrypted_content: Some("latest encrypted summary".to_owned()), - internal_chat_message_metadata_passthrough: None, - }; - - assert_eq!( - encrypted_parent_compaction( - [&older, &latest].into_iter(), - DEFAULT_PARENT_COMPACTION_TOKENS, - ), - Ok(Some(latest.clone())) - ); - assert_eq!( - encrypted_parent_compaction( - [&latest, &older].into_iter(), - DEFAULT_PARENT_COMPACTION_TOKENS, - ), - Ok(Some(older)) - ); -} - -#[test] -fn encrypted_parent_compaction_rejects_invalid_latest_item() { - let older = ResponseItem::Compaction { - id: Some(ResponseItemId::from_server("cmp_older".to_owned())), - encrypted_content: "older encrypted summary".to_owned(), - internal_chat_message_metadata_passthrough: None, - }; - let invalid = [ - ResponseItem::Compaction { - id: None, - encrypted_content: "encrypted summary without an ID".to_owned(), - internal_chat_message_metadata_passthrough: None, - }, - ResponseItem::Compaction { - id: Some(ResponseItemId::from_server("cmp_empty".to_owned())), - encrypted_content: String::new(), - internal_chat_message_metadata_passthrough: None, - }, - ResponseItem::ContextCompaction { - id: None, - encrypted_content: Some("encrypted context without an ID".to_owned()), - internal_chat_message_metadata_passthrough: None, - }, - ResponseItem::ContextCompaction { - id: Some(ResponseItemId::from_server("cmp_missing".to_owned())), - encrypted_content: None, - internal_chat_message_metadata_passthrough: None, - }, - ResponseItem::ContextCompaction { - id: Some(ResponseItemId::from_server("cmp_empty".to_owned())), - encrypted_content: Some(String::new()), - internal_chat_message_metadata_passthrough: None, - }, - ]; - - for latest in &invalid { - assert_eq!( - encrypted_parent_compaction( - [&older, latest].into_iter(), - DEFAULT_PARENT_COMPACTION_TOKENS, - ), - Err(ParentCompactionError::Unusable), - "an unusable latest summary must not resurrect older context" - ); - } -} - -#[test] -fn encrypted_parent_compaction_rejects_oversized_latest_item() -> Result<()> { - let max_compaction_bytes = - TruncationPolicy::Tokens(DEFAULT_PARENT_COMPACTION_TOKENS).byte_budget(); - let mut bounded = [ - ResponseItem::Compaction { - id: Some(ResponseItemId::from_server("cmp_bounded".to_owned())), - encrypted_content: String::new(), - internal_chat_message_metadata_passthrough: None, - }, - ResponseItem::ContextCompaction { - id: Some(ResponseItemId::from_server("ctx_bounded".to_owned())), - encrypted_content: Some(String::new()), - internal_chat_message_metadata_passthrough: None, - }, - ]; - - for item in &mut bounded { - let envelope_bytes = serde_json::to_vec(&*item)?.len(); - let encrypted_content = match item { - ResponseItem::Compaction { - encrypted_content, .. - } - | ResponseItem::ContextCompaction { - encrypted_content: Some(encrypted_content), - .. - } => encrypted_content, - _ => unreachable!("test fixtures are encrypted compaction items"), - }; - *encrypted_content = "a".repeat(max_compaction_bytes - envelope_bytes); - assert_eq!(serde_json::to_vec(&*item)?.len(), max_compaction_bytes); - assert_eq!( - encrypted_parent_compaction(std::iter::once(&*item), DEFAULT_PARENT_COMPACTION_TOKENS,), - Ok(Some(item.clone())) - ); - - let mut oversized = item.clone(); - match &mut oversized { - ResponseItem::Compaction { - encrypted_content, .. - } - | ResponseItem::ContextCompaction { - encrypted_content: Some(encrypted_content), - .. - } => encrypted_content.push('a'), - _ => unreachable!("test fixtures are encrypted compaction items"), - } - assert_eq!( - serde_json::to_vec(&oversized)?.len(), - max_compaction_bytes + 1 - ); - assert_eq!( - encrypted_parent_compaction( - [&*item, &oversized].into_iter(), - DEFAULT_PARENT_COMPACTION_TOKENS, - ), - Err(ParentCompactionError::Oversized), - "an oversized latest summary must not resurrect older context" - ); - } - - let oversized_metadata = ResponseItem::ContextCompaction { - id: Some(ResponseItemId::from_server( - "ctx_oversized_metadata".to_owned(), - )), - encrypted_content: Some("bounded encrypted summary".to_owned()), - internal_chat_message_metadata_passthrough: Some(InternalChatMessageMetadataPassthrough { - turn_id: Some("a".repeat(max_compaction_bytes)), - ..Default::default() - }), - }; - assert!(serde_json::to_vec(&oversized_metadata)?.len() > max_compaction_bytes); - assert_eq!( - encrypted_parent_compaction( - [&bounded[0], &oversized_metadata].into_iter(), - DEFAULT_PARENT_COMPACTION_TOKENS, - ), - Err(ParentCompactionError::Oversized), - "oversized passthrough metadata must not bypass the complete-item limit" - ); - - Ok(()) -} - async fn sample_conversation_history( conversation_history: Vec, arguments: &str, diff --git a/codex-rs/ext/guardian-v2/src/async_scorer/mod.rs b/codex-rs/ext/guardian-v2/src/async_scorer/mod.rs index 7b6b82039e..cb4b12d55b 100644 --- a/codex-rs/ext/guardian-v2/src/async_scorer/mod.rs +++ b/codex-rs/ext/guardian-v2/src/async_scorer/mod.rs @@ -5,6 +5,7 @@ mod config; mod coverage; mod extension; mod metrics; +mod parent_compaction; mod review_evidence; mod sampler; mod transcript; diff --git a/codex-rs/ext/guardian-v2/src/async_scorer/parent_compaction.rs b/codex-rs/ext/guardian-v2/src/async_scorer/parent_compaction.rs new file mode 100644 index 0000000000..311da0be5c --- /dev/null +++ b/codex-rs/ext/guardian-v2/src/async_scorer/parent_compaction.rs @@ -0,0 +1,112 @@ +//! Selects the Luna checkpoint once under the session's context mode. +//! Sampling and fast approval share eligibility; legacy omission stays distinct from rejection. + +use codex_core::context::GuardianContextMode; +use codex_extension_api::ConversationHistorySnapshot; +use codex_extension_api::ResponseItem; +use codex_protocol::protocol::TruncationPolicy; + +use super::config::GuardianV2Config; +use super::sampler::LunaSampler; + +#[derive(Debug, Eq, PartialEq)] +pub(super) enum ParentCompactionError { + RequiresSync, + Serialization, + Oversized, + Unusable, +} + +pub(super) struct ParentCompaction { + pub(super) item: Option, + pub(super) model_hash: Option, +} + +pub(super) fn select_parent_compaction( + mode: GuardianContextMode, + config: &GuardianV2Config, + history: &dyn ConversationHistorySnapshot, + sampler: &LunaSampler, + legacy_model_hash: Option<&str>, +) -> Result { + let model_hash = match mode { + GuardianContextMode::Legacy => legacy_model_hash, + GuardianContextMode::ThreadOwned => history.latest_compaction_model_hash(), + }; + if mode == GuardianContextMode::ThreadOwned + && history.items().any(|item| { + matches!( + item, + ResponseItem::Compaction { .. } | ResponseItem::ContextCompaction { .. } + ) + }) + && (!config.reuse_parent_compaction || !sampler.supports_parent_compaction(model_hash)) + { + return Err(ParentCompactionError::RequiresSync); + } + let item = if config.reuse_parent_compaction { + match encrypted_parent_compaction(history.items(), config.max_parent_compaction_tokens) { + Ok(item) => item, + Err(ParentCompactionError::Unusable) if mode == GuardianContextMode::Legacy => None, + Err(error) => return Err(error), + } + } else { + None + }; + // Legacy has raw history, so it can omit an incompatible checkpoint. Validate its size + // before omission, preserving the previous failure behavior for oversized items. + let item = item.filter(|_| { + mode == GuardianContextMode::ThreadOwned || sampler.supports_parent_compaction(model_hash) + }); + Ok(ParentCompaction { + item, + model_hash: model_hash.map(str::to_owned), + }) +} + +// An unusable latest compaction must never fall back to an older one. Missing +// encrypted content is rejected here; only legacy callers may omit that checkpoint. +fn encrypted_parent_compaction<'a>( + items: impl Iterator, + max_parent_compaction_tokens: usize, +) -> Result, ParentCompactionError> { + let max_compaction_bytes = TruncationPolicy::Tokens(max_parent_compaction_tokens).byte_budget(); + let Some(item) = items + .filter(|item| { + matches!( + item, + ResponseItem::Compaction { .. } | ResponseItem::ContextCompaction { .. } + ) + }) + .last() + else { + return Ok(None); + }; + + let encrypted_content = match item { + ResponseItem::Compaction { + id: Some(_), + encrypted_content, + .. + } + | ResponseItem::ContextCompaction { + id: Some(_), + encrypted_content: Some(encrypted_content), + .. + } => encrypted_content, + _ => return Err(ParentCompactionError::Unusable), + }; + if encrypted_content.is_empty() { + return Err(ParentCompactionError::Unusable); + } + let serialized = serde_json::to_vec(item).map_err(|_| ParentCompactionError::Serialization)?; + if serialized.len() > max_compaction_bytes { + return Err(ParentCompactionError::Oversized); + } + + Ok(Some(item.clone())) +} + +#[cfg(test)] +#[path = "parent_compaction_tests.rs"] +mod tests; diff --git a/codex-rs/ext/guardian-v2/src/async_scorer/parent_compaction_tests.rs b/codex-rs/ext/guardian-v2/src/async_scorer/parent_compaction_tests.rs new file mode 100644 index 0000000000..c921061fa9 --- /dev/null +++ b/codex-rs/ext/guardian-v2/src/async_scorer/parent_compaction_tests.rs @@ -0,0 +1,168 @@ +//! Checkpoint identity and complete-item bounds remain unchanged after policy extraction. + +use super::super::config::DEFAULT_PARENT_COMPACTION_TOKENS; +use super::*; +use anyhow::Result; +use codex_protocol::ResponseItemId; +use codex_protocol::models::InternalChatMessageMetadataPassthrough; +use pretty_assertions::assert_eq; + +#[test] +fn encrypted_parent_compaction_preserves_the_latest_valid_item() { + let older = ResponseItem::Compaction { + id: Some(ResponseItemId::from_server("cmp_older".to_owned())), + encrypted_content: "older encrypted summary".to_owned(), + internal_chat_message_metadata_passthrough: None, + }; + let latest = ResponseItem::ContextCompaction { + id: Some(ResponseItemId::from_server("cmp_latest".to_owned())), + encrypted_content: Some("latest encrypted summary".to_owned()), + internal_chat_message_metadata_passthrough: None, + }; + + assert_eq!( + encrypted_parent_compaction( + [&older, &latest].into_iter(), + DEFAULT_PARENT_COMPACTION_TOKENS, + ), + Ok(Some(latest.clone())) + ); + assert_eq!( + encrypted_parent_compaction( + [&latest, &older].into_iter(), + DEFAULT_PARENT_COMPACTION_TOKENS, + ), + Ok(Some(older)) + ); +} + +#[test] +fn encrypted_parent_compaction_rejects_invalid_latest_item() { + let older = ResponseItem::Compaction { + id: Some(ResponseItemId::from_server("cmp_older".to_owned())), + encrypted_content: "older encrypted summary".to_owned(), + internal_chat_message_metadata_passthrough: None, + }; + let invalid = [ + ResponseItem::Compaction { + id: None, + encrypted_content: "encrypted summary without an ID".to_owned(), + internal_chat_message_metadata_passthrough: None, + }, + ResponseItem::Compaction { + id: Some(ResponseItemId::from_server("cmp_empty".to_owned())), + encrypted_content: String::new(), + internal_chat_message_metadata_passthrough: None, + }, + ResponseItem::ContextCompaction { + id: None, + encrypted_content: Some("encrypted context without an ID".to_owned()), + internal_chat_message_metadata_passthrough: None, + }, + ResponseItem::ContextCompaction { + id: Some(ResponseItemId::from_server("cmp_missing".to_owned())), + encrypted_content: None, + internal_chat_message_metadata_passthrough: None, + }, + ResponseItem::ContextCompaction { + id: Some(ResponseItemId::from_server("cmp_empty".to_owned())), + encrypted_content: Some(String::new()), + internal_chat_message_metadata_passthrough: None, + }, + ]; + + for latest in &invalid { + assert_eq!( + encrypted_parent_compaction( + [&older, latest].into_iter(), + DEFAULT_PARENT_COMPACTION_TOKENS, + ), + Err(ParentCompactionError::Unusable), + "an unusable latest summary must not resurrect older context" + ); + } +} + +#[test] +fn encrypted_parent_compaction_rejects_oversized_latest_item() -> Result<()> { + let max_compaction_bytes = + TruncationPolicy::Tokens(DEFAULT_PARENT_COMPACTION_TOKENS).byte_budget(); + let mut bounded = [ + ResponseItem::Compaction { + id: Some(ResponseItemId::from_server("cmp_bounded".to_owned())), + encrypted_content: String::new(), + internal_chat_message_metadata_passthrough: None, + }, + ResponseItem::ContextCompaction { + id: Some(ResponseItemId::from_server("ctx_bounded".to_owned())), + encrypted_content: Some(String::new()), + internal_chat_message_metadata_passthrough: None, + }, + ]; + + for item in &mut bounded { + let envelope_bytes = serde_json::to_vec(&*item)?.len(); + let encrypted_content = match item { + ResponseItem::Compaction { + encrypted_content, .. + } + | ResponseItem::ContextCompaction { + encrypted_content: Some(encrypted_content), + .. + } => encrypted_content, + _ => unreachable!("test fixtures are encrypted compaction items"), + }; + *encrypted_content = "a".repeat(max_compaction_bytes - envelope_bytes); + assert_eq!(serde_json::to_vec(&*item)?.len(), max_compaction_bytes); + assert_eq!( + encrypted_parent_compaction(std::iter::once(&*item), DEFAULT_PARENT_COMPACTION_TOKENS,), + Ok(Some(item.clone())) + ); + + let mut oversized = item.clone(); + match &mut oversized { + ResponseItem::Compaction { + encrypted_content, .. + } + | ResponseItem::ContextCompaction { + encrypted_content: Some(encrypted_content), + .. + } => encrypted_content.push('a'), + _ => unreachable!("test fixtures are encrypted compaction items"), + } + assert_eq!( + serde_json::to_vec(&oversized)?.len(), + max_compaction_bytes + 1 + ); + assert_eq!( + encrypted_parent_compaction( + [&*item, &oversized].into_iter(), + DEFAULT_PARENT_COMPACTION_TOKENS, + ), + Err(ParentCompactionError::Oversized), + "an oversized latest summary must not resurrect older context" + ); + } + + let oversized_metadata = ResponseItem::ContextCompaction { + id: Some(ResponseItemId::from_server( + "ctx_oversized_metadata".to_owned(), + )), + encrypted_content: Some("bounded encrypted summary".to_owned()), + internal_chat_message_metadata_passthrough: Some(InternalChatMessageMetadataPassthrough { + turn_id: Some("a".repeat(max_compaction_bytes)), + ..Default::default() + }), + }; + assert!(serde_json::to_vec(&oversized_metadata)?.len() > max_compaction_bytes); + assert_eq!( + encrypted_parent_compaction( + [&bounded[0], &oversized_metadata].into_iter(), + DEFAULT_PARENT_COMPACTION_TOKENS, + ), + Err(ParentCompactionError::Oversized), + "oversized passthrough metadata must not bypass the complete-item limit" + ); + + Ok(()) +}