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(()) +}