diff --git a/codex-rs/app-server-protocol/schema/precomputed/app-server-exports-stable.json.zst b/codex-rs/app-server-protocol/schema/precomputed/app-server-exports-stable.json.zst index c4ceee2265..1ce21e841b 100644 Binary files a/codex-rs/app-server-protocol/schema/precomputed/app-server-exports-stable.json.zst and b/codex-rs/app-server-protocol/schema/precomputed/app-server-exports-stable.json.zst differ diff --git a/codex-rs/app-server/tests/suite/v2/guardian_v2_history_tests.rs b/codex-rs/app-server/tests/suite/v2/guardian_v2_history_tests.rs index 6ce65f7e1d..7f8ea10e22 100644 --- a/codex-rs/app-server/tests/suite/v2/guardian_v2_history_tests.rs +++ b/codex-rs/app-server/tests/suite/v2/guardian_v2_history_tests.rs @@ -67,17 +67,23 @@ enum ContextPath { ThreadOwned, } -#[test_case(ContextPath::ThreadOwned, "matching", 0, AnswerSize::Normal; "compatible checkpoint")] -#[test_case(ContextPath::ThreadOwned, "different", 0, AnswerSize::Normal; "incompatible checkpoint")] -#[test_case(ContextPath::ThreadOwned, "matching", 140, AnswerSize::Normal; "source call evicted")] -#[test_case(ContextPath::ThreadOwned, "matching", 0, AnswerSize::Oversized; "incomplete answers reject fresh low score")] -#[test_case(ContextPath::Legacy, "matching", 0, AnswerSize::Normal; "legacy answers remain runtime only")] -#[test_case(ContextPath::Legacy, "matching", 140, AnswerSize::Normal; "legacy source call evicted")] -#[test_case(ContextPath::Legacy, "matching", 0, AnswerSize::Oversized; "legacy answer truncation")] +#[test_case(ContextPath::ThreadOwned, Some("matching"), Some("matching"), 0, AnswerSize::Normal; "compatible checkpoint")] +#[test_case(ContextPath::ThreadOwned, Some("matching"), Some("different"), 0, AnswerSize::Normal; "incompatible checkpoint")] +#[test_case(ContextPath::ThreadOwned, Some("matching"), None, 0, AnswerSize::Normal; "unknown Luna compatibility")] +#[test_case(ContextPath::ThreadOwned, Some("matching"), Some(""), 0, AnswerSize::Normal; "empty Luna compatibility")] +#[test_case(ContextPath::ThreadOwned, None, Some("matching"), 0, AnswerSize::Normal; "unknown producer remains unknown after model switch")] +#[test_case(ContextPath::ThreadOwned, Some("matching"), Some("matching"), 140, AnswerSize::Normal; "source call evicted")] +#[test_case(ContextPath::ThreadOwned, Some("matching"), Some("matching"), 0, AnswerSize::Oversized; "incomplete answers reject fresh low score")] +#[test_case(ContextPath::Legacy, Some("matching"), Some("matching"), 0, AnswerSize::Normal; "legacy answers remain runtime only")] +#[test_case(ContextPath::Legacy, Some("matching"), Some("different"), 0, AnswerSize::Normal; "legacy incompatible checkpoint still samples")] +#[test_case(ContextPath::Legacy, Some("matching"), None, 0, AnswerSize::Normal; "legacy unknown Luna compatibility still samples")] +#[test_case(ContextPath::Legacy, Some("matching"), Some("matching"), 140, AnswerSize::Normal; "legacy source call evicted")] +#[test_case(ContextPath::Legacy, Some("matching"), Some("matching"), 0, AnswerSize::Oversized; "legacy answer truncation")] #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn guardians_retain_evidence_after_compaction_and_discard_it_after_rollback( context_path: ContextPath, - luna_hash: &str, + parent_hash: Option<&str>, + luna_hash: Option<&str>, tool_traffic: usize, answer_size: AnswerSize, ) -> Result<()> { @@ -86,6 +92,8 @@ async fn guardians_retain_evidence_after_compaction_and_discard_it_after_rollbac const RESTRICTION: &str = "Only publish to a private repository."; const EVIDENCE: &str = "repository is private"; const SUMMARY: &str = "Repository inspection was summarized."; + let compatible = parent_hash == Some("matching") && luna_hash == parent_hash; + let requires_sync = matches!(context_path, ContextPath::ThreadOwned) && !compatible; let answer = match answer_size { AnswerSize::Normal => USER_INPUT_RESTRICTION.to_owned(), AnswerSize::Oversized => "Never publish this repository publicly. ".repeat(200), @@ -104,7 +112,11 @@ async fn guardians_retain_evidence_after_compaction_and_discard_it_after_rollbac (ContextPath::ThreadOwned, AnswerSize::Oversized) ); let classifier = Arc::new(MockResponsesState { - luna_score: if rejects_incomplete_score { 0.0 } else { 1.0 }, + luna_score: if rejects_incomplete_score || requires_sync { + 0.0 + } else { + 1.0 + }, ..Default::default() }); let router = Router::new() @@ -223,14 +235,18 @@ async fn guardians_retain_evidence_after_compaction_and_discard_it_after_rollbac }; mock_config.write(codex_home.path())?; let config = load_default_config_for_test(&codex_home).await; - let models = [(MODEL, "matching"), ("gpt-5.6-luna", luna_hash)] - .into_iter() - .map(|(model, hash)| { - let mut info = codex_core::test_support::construct_model_info_offline(model, &config); - info.comp_hash = Some(hash.to_owned()); - info - }) - .collect(); + let models = [ + (MODEL, parent_hash), + ("gpt-5.6-luna", luna_hash), + ("resumed-parent", Some("matching")), + ] + .into_iter() + .map(|(model, hash)| { + let mut info = codex_core::test_support::construct_model_info_offline(model, &config); + info.comp_hash = hash.map(str::to_owned); + info + }) + .collect(); write_models_cache_with_models(codex_home.path(), models)?; write_chatgpt_auth( codex_home.path(), @@ -263,7 +279,7 @@ async fn guardians_retain_evidence_after_compaction_and_discard_it_after_rollbac .into_iter() .enumerate() { - if (1..=2).contains(&index) { + if index == 2 || (index == 1 && !requires_sync) { app_server.shutdown_gracefully().await?; app_server = TestAppServer::builder() .with_codex_home(codex_home.path()) @@ -273,6 +289,7 @@ async fn guardians_retain_evidence_after_compaction_and_discard_it_after_rollbac let id = app_server .send_thread_resume_request(ThreadResumeParams { thread_id: thread_id.clone(), + model: parent_hash.is_none().then(|| "resumed-parent".to_owned()), ..Default::default() }) .await?; @@ -304,15 +321,26 @@ async fn guardians_retain_evidence_after_compaction_and_discard_it_after_rollbac let id = app_server .send_turn_start_request(TurnStartParams { thread_id: thread_id.clone(), - input: vec![UserInput::Text { - text: prompt.to_owned(), - text_elements: Vec::new(), - }], + input: if index == 1 && requires_sync { + Vec::new() + } else { + vec![UserInput::Text { + text: prompt.to_owned(), + text_elements: Vec::new(), + }] + }, ..Default::default() }) .await?; let _: TurnStartResponse = timeout(TIMEOUT, app_server.read_response(id)).await??; if index == 0 { + // Wait for the pre-answer sample before replying; scoring runs asynchronously. + let before_answer = wait_for_luna_request(&classifier, /*index*/ 0).await?; + assert!( + !before_answer + .to_string() + .contains(">>> TRUSTED USER ANSWERS START") + ); submit_user_input_response( &mut app_server, json!({ @@ -320,19 +348,61 @@ async fn guardians_retain_evidence_after_compaction_and_discard_it_after_rollbac }), ) .await?; - // request_user_input itself is scored before the host receives the answer. - let before_answer = wait_for_luna_request(&classifier, /*index*/ 0).await?; - assert!( - !before_answer - .to_string() - .contains(">>> TRUSTED USER ANSWERS START") - ); classifier.allow_luna.notify_one(); } let completed: TurnCompletedNotification = timeout(TIMEOUT, app_server.read_notification("turn/completed")).await??; assert_eq!(completed.turn.status, TurnStatus::Completed); - let request = wait_for_luna_request(&classifier, index + 1).await?; + if requires_sync && (1..=3).contains(&index) { + let reviews = review_requests.lock().expect("request log lock"); + assert_eq!( + reviews.len(), + index + 1, + "incompatible checkpoints require sync review" + ); + let review = &reviews[index]; + assert!( + review["input"] + .as_array() + .expect("request input array") + .contains(&checkpoint) + ); + let text = review.to_string(); + if index >= 2 { + assert!( + text.contains(prompt), + "current user input missing from sync review: {text}" + ); + } + if index == 3 { + assert!( + !text.contains("Inspect after resume."), + "rolled-back user input remains in sync review: {text}" + ); + } + assert!(text.contains(USER_INPUT_RESTRICTION)); + assert!(text.contains("TRUSTED USER ANSWERS START")); + assert!(text.contains("TRANSCRIPT START")); + assert!(!text.contains("TRANSCRIPT DELTA START")); + assert_eq!( + classifier + .luna_requests + .lock() + .expect("classifier request lock") + .len(), + 2, + "no checkpoint-less async request may run after compaction" + ); + continue; + } + let sample_index = if !requires_sync { + index + 1 + } else if index == 0 { + 1 + } else { + 2 + }; + let request = wait_for_luna_request(&classifier, sample_index).await?; let content = request["input"] .as_array() .expect("request input array") @@ -365,7 +435,7 @@ async fn guardians_retain_evidence_after_compaction_and_discard_it_after_rollbac assert_eq!(sync_input.contains(&checkpoint), (1..=3).contains(&index)); assert_eq!( async_input.contains(&checkpoint), - (1..=3).contains(&index) && luna_hash == "matching" + (1..=3).contains(&index) && compatible ); let sync_text = sync_input .iter() @@ -485,7 +555,7 @@ async fn guardians_retain_evidence_after_compaction_and_discard_it_after_rollbac } } classifier.allow_luna.notify_one(); - if rejects_incomplete_score { + if rejects_incomplete_score || (index == 0 && requires_sync) { // Wait for host publication, not merely a mock response. The next action must // see a fresh low score with unchanged authorization, rather than a stale score. let path = thread.path.as_ref().expect("legacy rollout path"); @@ -507,6 +577,8 @@ async fn guardians_retain_evidence_after_compaction_and_discard_it_after_rollbac } }) .await?; + } + if rejects_incomplete_score { let completed = app_server .start_turn_and_wait_for_completion(TurnStartParams { thread_id: thread_id.clone(), @@ -551,6 +623,27 @@ async fn guardians_retain_evidence_after_compaction_and_discard_it_after_rollbac usize::from(matches!(context_path, ContextPath::ThreadOwned)), "only the enabled path may persist a retained-answer event", ); + for line in &items { + if let RolloutItem::Compacted(checkpoint) = &line.item { + for envelope in checkpoint.replacement_history.iter().flatten() { + if matches!( + envelope.item, + codex_protocol::models::ResponseItem::Compaction { .. } + ) { + assert_eq!( + envelope + .metadata + .as_ref() + .and_then(|metadata| metadata.compaction_model_hash.as_deref()), + matches!(context_path, ContextPath::ThreadOwned) + .then_some(parent_hash) + .flatten(), + "only the enabled path records checkpoint producer provenance", + ); + } + } + } + } if matches!(context_path, ContextPath::Legacy) { for line in &items { if let RolloutItem::Compacted(checkpoint) = &line.item { diff --git a/codex-rs/core/src/codex_thread.rs b/codex-rs/core/src/codex_thread.rs index 888d80fe2b..a2ec9fb99f 100644 --- a/codex-rs/core/src/codex_thread.rs +++ b/codex-rs/core/src/codex_thread.rs @@ -11,6 +11,7 @@ use crate::session::step_settings::StepSettingsUpdate; use codex_diagnostics::Gauge; use codex_diagnostics::GaugeGuard; use codex_exec_server::SelectedCapabilityRootsStatus; +use codex_extension_api::ConversationHistorySnapshot; use codex_extension_api::ThreadIdleCause; use codex_features::Feature; use codex_history::RolloutItem; @@ -804,9 +805,14 @@ impl CodexThread { self.session.multi_agent_version() } + /// Shares an immutable view of the live parent model context and retained host facts. + pub async fn conversation_history_snapshot(&self) -> Arc { + self.session.conversation_history_snapshot().await + } + /// Returns the current user-authorization revision for Guardian. pub async fn guardian_authorization_version(&self) -> GuardianAuthorizationVersion { - let history = self.session.conversation_history_snapshot().await; + let history = self.conversation_history_snapshot().await; self.thread_extension_data() .get_or_init(GuardianReviewEvidence::default) .authorization_version(history.as_ref()) diff --git a/codex-rs/core/src/compact.rs b/codex-rs/core/src/compact.rs index 49306c357c..5e473b0547 100644 --- a/codex-rs/core/src/compact.rs +++ b/codex-rs/core/src/compact.rs @@ -86,6 +86,7 @@ pub(crate) struct CompactedHistoryMetadata { pub(crate) window_number: u64, pub(crate) window_ids: AutoCompactWindowIds, pub(crate) compaction_response_id: Option, + pub(crate) compaction_model_hash: Option, } pub(crate) async fn build_compaction_initial_context( @@ -385,6 +386,7 @@ async fn run_compact_task_inner_impl( window_number, window_ids, compaction_response_id: Some(compaction_response_id), + compaction_model_hash: turn_context.model_info().comp_hash.clone(), }, ) .await; diff --git a/codex-rs/core/src/compact_remote.rs b/codex-rs/core/src/compact_remote.rs index 442c4a0ce0..d1eb2124eb 100644 --- a/codex-rs/core/src/compact_remote.rs +++ b/codex-rs/core/src/compact_remote.rs @@ -301,6 +301,7 @@ async fn run_remote_compact_task_inner_impl( window_number: new_window_number, window_ids: new_window_ids, compaction_response_id: None, + compaction_model_hash: compaction_turn_context.model_info().comp_hash.clone(), }, ) .await; diff --git a/codex-rs/core/src/compact_remote_v2.rs b/codex-rs/core/src/compact_remote_v2.rs index 68e2eaeda6..f02e2db678 100644 --- a/codex-rs/core/src/compact_remote_v2.rs +++ b/codex-rs/core/src/compact_remote_v2.rs @@ -351,6 +351,7 @@ async fn run_remote_compact_task_inner_impl( window_number: new_window_number, window_ids: new_window_ids, compaction_response_id: Some(compaction_response_id), + compaction_model_hash: compaction_turn_context.model_info().comp_hash.clone(), }, ) .await; diff --git a/codex-rs/core/src/context/guardian_review_evidence.rs b/codex-rs/core/src/context/guardian_review_evidence.rs index d6824982eb..d36d785a14 100644 --- a/codex-rs/core/src/context/guardian_review_evidence.rs +++ b/codex-rs/core/src/context/guardian_review_evidence.rs @@ -64,6 +64,11 @@ 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(crate) fn from_features(features: &Features) -> Self { Self { mode: if features.enabled(Feature::GuardianThreadContext) { diff --git a/codex-rs/core/src/context_manager/history.rs b/codex-rs/core/src/context_manager/history.rs index 404088301f..b02d2188d8 100644 --- a/codex-rs/core/src/context_manager/history.rs +++ b/codex-rs/core/src/context_manager/history.rs @@ -95,6 +95,20 @@ pub(crate) enum HistoryReplacement { } impl ConversationHistorySnapshot for SharedConversationHistory { + fn latest_compaction_model_hash(&self) -> Option<&str> { + self.items + .iter() + .rev() + .find(|envelope| { + matches!( + envelope.item, + ResponseItem::Compaction { .. } | ResponseItem::ContextCompaction { .. } + ) + }) + .and_then(|envelope| envelope.metadata.as_ref()) + .and_then(|metadata| metadata.compaction_model_hash.as_deref()) + } + fn retained_context(&self) -> Option<&RetainedContext> { Some(&self.retained_context) } diff --git a/codex-rs/core/src/context_manager/history_tests.rs b/codex-rs/core/src/context_manager/history_tests.rs index 8fbca13c55..c6e926ee0a 100644 --- a/codex-rs/core/src/context_manager/history_tests.rs +++ b/codex-rs/core/src/context_manager/history_tests.rs @@ -153,6 +153,48 @@ fn conversation_history_snapshot_shares_response_items_until_history_changes() { ); } +#[test] +fn conversation_history_snapshot_binds_compaction_hash_to_the_latest_item() { + let checkpoint = ResponseItemEnvelope { + item: serde_json::from_value(serde_json::json!({ + "type": "compaction", "id": "known", "encrypted_content": "opaque checkpoint" + })) + .expect("checkpoint fixture"), + metadata: Some(CodexHarnessMetadata { + compaction_model_hash: Some("producer-hash".to_owned()), + ..Default::default() + }), + }; + let mut history = ContextManager::new(); + history.replace_annotated(vec![checkpoint.clone()]); + let snapshot = history.conversation_history_snapshot(); + let mut unknown = checkpoint.clone(); + unknown.metadata = None; + unknown.item = serde_json::from_value(serde_json::json!({ + "type": "compaction", "id": "unknown", "encrypted_content": "newer opaque checkpoint" + })) + .expect("unknown checkpoint fixture"); + history.replace_annotated(vec![checkpoint.clone(), unknown]); + assert_eq!( + history + .conversation_history_snapshot() + .latest_compaction_model_hash(), + None + ); + assert_eq!( + snapshot.latest_compaction_model_hash(), + Some("producer-hash") + ); + // Checkpoint replay/rollback restores the item's own provenance, not a new model's metadata. + history.replace_annotated(vec![checkpoint]); + assert_eq!( + history + .conversation_history_snapshot() + .latest_compaction_model_hash(), + Some("producer-hash") + ); +} + #[test] fn conversation_history_snapshot_excludes_contextual_user_messages() { let contextual_message = crate::context::ContextualUserFragment::into(UserInstructions { diff --git a/codex-rs/core/src/session/mod.rs b/codex-rs/core/src/session/mod.rs index 9153bfad4a..d08feebaef 100644 --- a/codex-rs/core/src/session/mod.rs +++ b/codex-rs/core/src/session/mod.rs @@ -3759,6 +3759,19 @@ impl Session { for envelope in &mut items { Self::assign_missing_response_item_id(&mut envelope.item); } + if self.enabled(Feature::GuardianThreadContext) + && let Some(checkpoint) = items.iter_mut().rev().find(|envelope| { + matches!( + envelope.item, + ResponseItem::Compaction { .. } | ResponseItem::ContextCompaction { .. } + ) + }) + { + checkpoint + .metadata + .get_or_insert_default() + .compaction_model_hash = metadata.compaction_model_hash; + } let mut compacted_item = CompactedItem { message: metadata.message, replacement_history: Some(items.clone()), @@ -4227,6 +4240,7 @@ impl Session { window_number, window_ids, compaction_response_id: None, + compaction_model_hash: None, }, ) .await; diff --git a/codex-rs/core/src/session/tests.rs b/codex-rs/core/src/session/tests.rs index 1d30b0c219..d9c1f72214 100644 --- a/codex-rs/core/src/session/tests.rs +++ b/codex-rs/core/src/session/tests.rs @@ -5952,6 +5952,7 @@ async fn compaction_checkpoint_waits_for_accepted_settings_persistence() { window_number, window_ids, compaction_response_id: None, + compaction_model_hash: None, }, ), )); diff --git a/codex-rs/ext/extension-api/src/capabilities/conversation_history.rs b/codex-rs/ext/extension-api/src/capabilities/conversation_history.rs index 7e8ff75031..8942a384c5 100644 --- a/codex-rs/ext/extension-api/src/capabilities/conversation_history.rs +++ b/codex-rs/ext/extension-api/src/capabilities/conversation_history.rs @@ -22,6 +22,12 @@ pub trait ConversationHistorySnapshot: Send + Sync { None } + /// Producer compatibility recorded on the latest opaque checkpoint. Missing provenance + /// must not be inferred from the currently selected model, including after resume. + fn latest_compaction_model_hash(&self) -> Option<&str> { + None + } + /// Original review evidence retained across parent compaction, in conversation order. /// Hosts without separate retention provide their current history. fn review_items(&self) -> Box + Send + '_> { 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 bc125a3bd5..30159d0ee5 100644 --- a/codex-rs/ext/guardian-v2/src/async_scorer/extension.rs +++ b/codex-rs/ext/guardian-v2/src/async_scorer/extension.rs @@ -15,6 +15,7 @@ use codex_core::config::Config; 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; @@ -107,6 +108,7 @@ fn should_classify_tool( pub enum StrictReviewReason { ElevatedRisk, StaleScore, + IncompatibleCompaction, } enum ClassificationOutcome { @@ -273,6 +275,9 @@ impl ApprovalReviewContributor for GuardianV2Extension { Box::pin(async move { thread_store.get::()?; let guardian_config = thread_store.get::()?; + let guardian_evidence = thread_store.get_or_init(GuardianReviewEvidence::default); + let thread_context_enabled = guardian_evidence.uses_thread_owned_context(); + let mut initial_cua_call = false; if guardian_config.review_scope == GuardianV2ReviewScope::ComputerUseOnly { let Ok(action) = serde_json::from_str::(prompt) else { record_fast_decision(extension_metrics.as_deref(), "deferred", "out_of_scope"); @@ -290,9 +295,10 @@ impl ApprovalReviewContributor for GuardianV2Extension { record_fast_decision(extension_metrics.as_deref(), "deferred", "out_of_scope"); return None; } - // The first REPL execution never waits for synchronous Guardian review. - // The async classifier still runs, and later calls use the normal policy. - if action.get("tool_name").and_then(serde_json::Value::as_str) == Some("js") + // Legacy mode keeps the first REPL execution's review bypass. + // Thread-owned context first checks whether Luna can use the checkpoint. + initial_cua_call = action.get("tool_name").and_then(serde_json::Value::as_str) + == Some("js") && action .get("connector_id") .and_then(serde_json::Value::as_str) @@ -301,8 +307,8 @@ impl ApprovalReviewContributor for GuardianV2Extension { .get::()? .js_executions .load(Ordering::Acquire) - == 1 - { + == 1; + if initial_cua_call && !thread_context_enabled { record_fast_decision( extension_metrics.as_deref(), "approved", @@ -339,7 +345,31 @@ impl ApprovalReviewContributor for GuardianV2Extension { record_fast_decision(extension_metrics.as_deref(), "deferred", "scoring_failure"); return None; }; - let current_authorization = ScoreAuthorization::current(&thread).await; + let root_authorization = thread + .guardian_root_snapshot() + .await + .map(|snapshot| snapshot.authorization_version); + let history = thread.conversation_history_snapshot().await; + if thread_context_enabled { + let sampler = thread_store.get::()?; + if requires_sync_for_compaction(&guardian_config, history.as_ref(), &sampler) { + thread_store.insert(StrictReviewReason::IncompatibleCompaction); + record_fast_decision( + extension_metrics.as_deref(), + "deferred", + "incompatible_compaction", + ); + return None; + } + } + if initial_cua_call { + record_fast_decision(extension_metrics.as_deref(), "approved", "initial_cua_call"); + return Some(ReviewDecision::Approved); + } + let current_authorization = ScoreAuthorization { + local: guardian_evidence.authorization_version(history.as_ref()), + root: root_authorization, + }; if !current_authorization.local.retained_context_complete || current_authorization .root @@ -571,6 +601,38 @@ impl GuardianV2Extension { .enable_image_capture(); } input.thread_store.insert(guardian_config.clone()); + 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", + ); + return; + } + let parent_compaction_hash = if thread_context_enabled { + input + .conversation_history + .latest_compaction_model_hash() + .map(str::to_owned) + } else { + 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(), @@ -590,22 +652,23 @@ impl GuardianV2Extension { } 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 call_id = input.call_id.to_owned(); let mcp_tool = input.mcp_tool.cloned(); let action = GuardianAction { tool_name: input.tool_name.clone(), payload: input.payload.clone(), }; - let parent_compaction_hash = parent_model - .as_ref() - .and_then(|model| model.comp_hash.clone()); let review_model_override = parent_model .as_ref() .and_then(|model| model.auto_review_model_override.clone()); // Snapshot before spawning so a delayed sample cannot see later reviews. - let guardian_evidence = input - .thread_store - .get_or_init(GuardianReviewEvidence::default); let sync_reviews = guardian_evidence.snapshot(); let codex_core::context::GuardianUserInputSnapshot { fragments: trusted_user_inputs, @@ -895,6 +958,22 @@ enum ParentCompactionError { Oversized, } +// Sampling and fast approval must apply the same checkpoint eligibility policy. +fn requires_sync_for_compaction( + config: &GuardianV2Config, + history: &dyn ConversationHistorySnapshot, + sampler: &LunaSampler, +) -> bool { + config.reuse_parent_compaction + && history.items().any(|item| { + matches!( + item, + ResponseItem::Compaction { .. } | ResponseItem::ContextCompaction { .. } + ) + }) + && !sampler.supports_parent_compaction(history.latest_compaction_model_hash()) +} + // An unusable latest compaction must never fall back to an older one. Missing // encrypted content can be omitted; content that cannot be bounded rejects the sample. fn encrypted_parent_compaction<'a>( 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 9a7011f879..64d25a2418 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 @@ -75,6 +75,7 @@ use super::should_classify_tool; use crate::async_scorer::config::CLASSIFICATION_OUTPUT_INSTRUCTIONS; use crate::async_scorer::config::DEFAULT_MODEL_CONTEXT_ITEM_TOKENS; use crate::async_scorer::config::DEFAULT_PARENT_COMPACTION_TOKENS; +use crate::async_scorer::config::GuardianV2Config; use crate::async_scorer::config::GuardianV2ReviewScope; use crate::async_scorer::metrics::CLASSIFICATION_DURATION_METRIC; use crate::async_scorer::metrics::CLASSIFICATION_METRIC; @@ -367,9 +368,13 @@ struct TestConversationHistory(Vec); struct TestRetainedHistory { current: TestConversationHistory, retained: Vec, + compaction_model_hash: Option, } impl ConversationHistorySnapshot for TestRetainedHistory { + fn latest_compaction_model_hash(&self) -> Option<&str> { + self.compaction_model_hash.as_deref() + } fn history_version(&self) -> u64 { self.current.history_version() } @@ -1097,10 +1102,16 @@ struct GuardianFailureFixture { impl GuardianFailureFixture { async fn new() -> Result { - let (_, test, registry) = sample_conversation_history( + Self::with_config("").await + } + + async fn with_config(guardian_config: &str) -> Result { + let (_, test, registry) = sample_configured_conversation_history( Vec::new(), r#"{"path":"README.md"}"#, Some(TEST_GUARDIAN_POLICY), + guardian_config, + /*model_defaults*/ None, ) .await?; let thread_store = test.codex.thread_extension_data(); @@ -2430,6 +2441,110 @@ async fn cached_score_survives_compaction_and_internal_context_but_not_user_inpu Ok(()) } +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn incompatible_compaction_blocks_cached_score_and_initial_cua_allowance() -> Result<()> { + assert_compaction_approval_policy(/*thread_context_enabled*/ true).await +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn legacy_incompatible_compaction_preserves_cached_score_and_initial_cua_allowance() +-> Result<()> { + assert_compaction_approval_policy(/*thread_context_enabled*/ false).await +} + +async fn assert_compaction_approval_policy(thread_context_enabled: bool) -> Result<()> { + skip_if_no_network!(Ok(())); + + let fixture = GuardianFailureFixture::with_config(&format!( + "[features]\nguardian_thread_context = {thread_context_enabled}\n" + )) + .await?; + let thread_store = fixture.test.codex.thread_extension_data(); + thread_store.insert(SecurityRiskScore { + scores: BTreeMap::from([("action_risk".to_owned(), 0.0)]), + call_id: None, + action: None, + sampled_at: None, + }); + assert_eq!( + fixture + .registry + .fast_approval_decision( + &fixture.session_store, + thread_store, + "review action", + /*extension_metrics*/ None + ) + .await, + Some(ReviewDecision::Approved) + ); + let authorization = fixture.test.codex.guardian_authorization_version().await; + fixture + .test + .codex + .inject_response_items(vec![ResponseItem::Compaction { + id: Some(ResponseItemId::from_server( + "incompatible-checkpoint".to_owned(), + )), + encrypted_content: "opaque parent summary".to_owned(), + internal_chat_message_metadata_passthrough: None, + }]) + .await?; + let mut model = (*thread_store + .get::() + .expect("parent model metadata")) + .clone(); + model.comp_hash = Some("incompatible-parent".to_owned()); + model.node_repl_auto_review_required = true; + thread_store.insert(model); + assert_eq!( + fixture.test.codex.guardian_authorization_version().await, + authorization + ); + // No new sample runs: only the enabled path rejects cached and initial-call approvals. + for (scope, prompt) in [ + ( + GuardianV2ReviewScope::Standard { + sandboxed_exec_commands: true, + }, + "review action", + ), + ( + GuardianV2ReviewScope::ComputerUseOnly, + r#"{"tool":"mcp_tool_call","server":"node_repl","connector_id":"node_repl","tool_name":"js"}"#, + ), + ] { + let mut config = (*thread_store + .get::() + .expect("Guardian configuration")) + .clone(); + config.review_scope = scope; + thread_store.insert(config); + thread_store + .get::() + .expect("score progress") + .js_executions + .store(/*val*/ 1, Ordering::Release); + assert_eq!( + fixture + .registry + .fast_approval_decision( + &fixture.session_store, + thread_store, + prompt, + /*extension_metrics*/ None + ) + .await, + (!thread_context_enabled).then_some(ReviewDecision::Approved) + ); + assert_eq!( + thread_store.remove::().as_deref(), + thread_context_enabled.then_some(&StrictReviewReason::IncompatibleCompaction) + ); + } + Ok(()) +} + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn contributor_counts_failed_thread_lookups_toward_score_lag() -> Result<()> { skip_if_no_network!(Ok(())); @@ -2864,11 +2979,26 @@ async fn contributor_sends_compacted_conversation_history_to_luna() -> Result<() #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn contributor_reuses_the_latest_compatible_parent_compaction() -> Result<()> { + assert_parent_compaction_reuse(/*thread_context_enabled*/ true).await +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn legacy_contributor_reuses_the_latest_compatible_parent_compaction() -> Result<()> { + assert_parent_compaction_reuse(/*thread_context_enabled*/ false).await +} + +async fn assert_parent_compaction_reuse(thread_context_enabled: bool) -> Result<()> { skip_if_no_network!(Ok(())); let thread_server = responses::start_mock_server().await; let test = test_codex() - .with_config(|config| config.approvals_reviewer = ApprovalsReviewer::AutoReview) + .with_config(move |config| { + config.approvals_reviewer = ApprovalsReviewer::AutoReview; + config + .features + .set_enabled(Feature::GuardianThreadContext, thread_context_enabled) + .expect("test context mode"); + }) .with_pre_build_hook(|home| { std::fs::write( home.join("config.toml"), @@ -2966,6 +3096,7 @@ async fn contributor_reuses_the_latest_compatible_parent_compaction() -> Result< let conversation_history = TestRetainedHistory { retained, current: conversation_history, + compaction_model_hash: parent_model.comp_hash.clone(), }; thread_store.insert(parent_model); @@ -3026,6 +3157,8 @@ async fn contributor_reuses_the_latest_compatible_parent_compaction() -> Result< }) .await?; assert_eq!(previous_score.scores.get("action_risk"), Some(&0.0)); + // Raw injection did not attach producer provenance to the live checkpoint. + // The sample's mock snapshot cannot make that live checkpoint safe for approval. assert_eq!( registry .fast_approval_decision( @@ -3035,7 +3168,7 @@ async fn contributor_reuses_the_latest_compatible_parent_compaction() -> Result< /*extension_metrics*/ None, ) .await, - Some(ReviewDecision::Approved) + (!thread_context_enabled).then_some(ReviewDecision::Approved), ); let oversized_compaction = ResponseItem::Compaction { @@ -3054,10 +3187,13 @@ async fn contributor_reuses_the_latest_compatible_parent_compaction() -> Result< tool_name: &tool_name, mcp_tool: None, payload: &tool_payload, - conversation_history: Arc::new(TestConversationHistory(vec![ - latest_compaction, - oversized_compaction, - ])), + conversation_history: Arc::new(TestRetainedHistory { + current: TestConversationHistory(vec![latest_compaction, oversized_compaction]), + retained: Vec::new(), + compaction_model_hash: thread_store + .get::() + .and_then(|model| model.comp_hash.clone()), + }), source: ToolCallSource::Direct, }) .await; @@ -3123,7 +3259,7 @@ async fn contributor_can_disable_parent_compaction_reuse() -> Result<()> { internal_chat_message_metadata_passthrough: None, }, ]; - let configuration = "[features.guardianv2]\nenabled = true\nreuse_parent_compaction = false\nmax_parent_compaction_tokens = 256\n"; + let configuration = "[features]\nguardian_thread_context = true\n\n[features.guardianv2]\nenabled = true\nreuse_parent_compaction = false\nmax_parent_compaction_tokens = 256\n"; let (request, test, _registry) = sample_configured_conversation_history( conversation_history, r#"{"path":"README.md"}"#, diff --git a/codex-rs/ext/guardian-v2/src/async_scorer/sampler.rs b/codex-rs/ext/guardian-v2/src/async_scorer/sampler.rs index f7f5aa411b..6a822cfd71 100644 --- a/codex-rs/ext/guardian-v2/src/async_scorer/sampler.rs +++ b/codex-rs/ext/guardian-v2/src/async_scorer/sampler.rs @@ -104,7 +104,7 @@ pub struct LunaSamplingRequest { pub images: Vec, /// Opaque parent compaction to reuse only for compatible model configurations. pub parent_compaction: Option, - /// Current parent model's encrypted-compaction compatibility hash. + /// Host-selected compatibility hash for the supplied parent checkpoint. pub parent_compaction_hash: Option, /// Reasoning budget explicitly selected for this request. pub reasoning_effort: ReasoningEffort, @@ -135,6 +135,9 @@ pub enum LunaSamplerError { /// A newer classification replaced this request when the pool was full. #[error("Luna request was superseded by a newer classification")] Superseded, + /// The supplied parent checkpoint cannot be consumed by this Luna configuration. + #[error("parent compaction is incompatible with Luna")] + IncompatibleCompaction, } struct PooledConnection { @@ -203,6 +206,15 @@ pub struct LunaSampler { } impl LunaSampler { + /// A checkpoint is reusable only when both models declare the same nonempty hash. + pub(super) fn supports_parent_compaction(&self, parent_hash: Option<&str>) -> bool { + parent_hash + .zip(self.config.luna_compaction_hash.as_deref()) + .is_some_and(|(parent_hash, luna_hash)| { + !parent_hash.is_empty() && parent_hash == luna_hash + }) + } + pub(super) fn new(config: LunaSamplerConfig) -> Self { Self { config, @@ -412,6 +424,7 @@ impl LunaSampler { | LunaSamplerError::MissingOutput | LunaSamplerError::OutputTooLarge | LunaSamplerError::Superseded + | LunaSamplerError::IncompatibleCompaction | LunaSamplerError::Api( ApiError::Transport(TransportError::Build(_)) | ApiError::ContextWindowExceeded @@ -432,6 +445,11 @@ impl LunaSampler { /// Sends one tool-less classification request on an exclusively leased WebSocket. pub async fn sample(&self, request: LunaSamplingRequest) -> Result { + if request.parent_compaction.is_some() + && !self.supports_parent_compaction(request.parent_compaction_hash.as_deref()) + { + return Err(LunaSamplerError::IncompatibleCompaction); + } // A classification is its own inference turn; retries keep that identity. let turn_id = Uuid::now_v7().to_string(); let parent_turn_id = request.parent_turn_id; @@ -452,15 +470,7 @@ impl LunaSampler { internal_chat_message_metadata_passthrough: None, }, ]; - if request - .parent_compaction_hash - .as_deref() - .zip(self.config.luna_compaction_hash.as_deref()) - .is_some_and(|(parent_hash, luna_hash)| { - !parent_hash.is_empty() && parent_hash == luna_hash - }) - && let Some(parent_compaction) = request.parent_compaction - { + if let Some(parent_compaction) = request.parent_compaction { input.push(parent_compaction); } if !request.trusted_review_evidence.is_empty() { diff --git a/codex-rs/ext/guardian-v2/src/async_scorer/sampler_tests.rs b/codex-rs/ext/guardian-v2/src/async_scorer/sampler_tests.rs index f797aad3c0..4750b3d7f8 100644 --- a/codex-rs/ext/guardian-v2/src/async_scorer/sampler_tests.rs +++ b/codex-rs/ext/guardian-v2/src/async_scorer/sampler_tests.rs @@ -39,6 +39,7 @@ use super::CLASSIFICATION_TOKEN_USAGE_METRIC; use super::INITIAL_WEBSOCKET_CONNECTIONS; use super::LunaSampler; use super::LunaSamplerConfig; +use super::LunaSamplerError; use super::LunaSamplingRequest; use super::MAX_SAMPLING_RETRIES; use super::MAX_WEBSOCKET_CONNECTIONS; @@ -570,7 +571,20 @@ async fn sampler_reuses_parent_compaction_only_for_matching_model_hashes() -> Re request.parent_compaction_hash = parent_hash.map(str::to_owned); request.trusted_review_evidence = vec!["trusted review".to_owned()]; - assert_eq!(sampler.sample(request).await?, "low"); + let result = sampler.sample(request).await; + if !should_reuse { + assert!(matches!( + result, + Err(LunaSamplerError::IncompatibleCompaction) + )); + assert!(server.connections().iter().all(Vec::is_empty)); + assert_eq!( + sampler.sample(sample_request("uncompacted-turn")).await?, + "low" + ); + continue; + } + assert_eq!(result?, "low"); let request = server .wait_for_request( @@ -582,31 +596,20 @@ async fn sampler_reuses_parent_compaction_only_for_matching_model_hashes() -> Re let input = request["input"].as_array().expect("input items"); assert_eq!(input[0]["type"], "additional_tools"); assert_eq!(input[1]["role"], "developer"); - if should_reuse { - assert_eq!(input.len(), 5); - assert_eq!(input[2], serde_json::to_value(&parent_compaction)?); - assert_eq!(input[3]["role"], "developer"); - assert_eq!(input[3]["content"][1]["text"], "trusted review"); - assert_eq!(input[4]["role"], "user"); + assert_eq!(input.len(), 5); + assert_eq!(input[2], serde_json::to_value(&parent_compaction)?); + assert_eq!(input[3]["role"], "developer"); + assert_eq!(input[3]["content"][1]["text"], "trusted review"); + assert_eq!(input[4]["role"], "user"); - let mut switched_request = sample_request("turn-2"); - switched_request.parent_compaction = Some(parent_compaction); - switched_request.parent_compaction_hash = Some("incompatible".to_owned()); - assert_eq!(sampler.sample(switched_request).await?, "low"); - let switched_request = server - .wait_for_request( - /*connection_index*/ INITIAL_WEBSOCKET_CONNECTIONS - 1, - /*request_index*/ 1, - ) - .await - .body_json(); - assert_eq!(switched_request["input"][2]["role"], "user"); - } else { - assert_eq!(input.len(), 4); - assert_eq!(input[2]["role"], "developer"); - assert_eq!(input[2]["content"][1]["text"], "trusted review"); - assert_eq!(input[3]["role"], "user"); - } + let mut switched_request = sample_request("turn-2"); + switched_request.parent_compaction = Some(parent_compaction); + switched_request.parent_compaction_hash = Some("incompatible".to_owned()); + assert!(matches!( + sampler.sample(switched_request).await, + Err(LunaSamplerError::IncompatibleCompaction) + )); + assert_eq!(server.connections().iter().map(Vec::len).sum::(), 1); } Ok(()) diff --git a/codex-rs/history/src/lib.rs b/codex-rs/history/src/lib.rs index 74e5b01c0a..734ec8e673 100644 --- a/codex-rs/history/src/lib.rs +++ b/codex-rs/history/src/lib.rs @@ -58,6 +58,10 @@ pub struct CodexHarnessMetadata { /// Whether a response configuration update was created by the Codex harness itself. #[serde(default, skip_serializing_if = "std::ops::Not::not")] pub harness_authored_configuration: bool, + + /// Producer compatibility for an opaque compaction item, never the currently selected model. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub compaction_model_hash: Option, } impl ResponseItemEnvelope {