From 5b85aea9797ce75152d3dfa3651b58f1bbd7d218 Mon Sep 17 00:00:00 2001 From: felixxia-oai Date: Mon, 7 Sep 2026 11:37:16 +0000 Subject: [PATCH] Keep Guardian review evidence consistent and reject stale approvals (#43442) ## Why Concurrent parent compaction can remove evidence between Guardian checkpoint selection and prompt construction. New user instructions or answers received during a review can also invalidate an approval based on earlier authorization. ## What changed - With `GuardianThreadContext` enabled, use the same parent history snapshot for checkpoint selection and prompt evidence. - Cancel completed allow decisions if the owning session's user-message revision or root authorization version changed during the review, even without cacheable review evidence. ## Testing Add regression coverage for evidence preservation during concurrent compaction and aborted approvals after new user instructions or verified answers. Extend reviewer context rollover coverage to exercise `GuardianThreadContext`. GitOrigin-RevId: 1c2c82197a27c0d3d45b6fe08d200a835d80cc30 --- codex-rs/core/src/guardian/prompt.rs | 7 +- codex-rs/core/src/guardian/review.rs | 18 +- codex-rs/core/src/guardian/review_session.rs | 15 +- .../core/src/guardian/review_session_tests.rs | 125 +++++++++++ codex-rs/core/src/guardian/tests.rs | 5 + .../tests/suite/guardian_authorization.rs | 201 ++++++++++++++++++ codex-rs/core/tests/suite/guardian_review.rs | 11 +- 7 files changed, 369 insertions(+), 13 deletions(-) diff --git a/codex-rs/core/src/guardian/prompt.rs b/codex-rs/core/src/guardian/prompt.rs index 1e408783a9..47b3678160 100644 --- a/codex-rs/core/src/guardian/prompt.rs +++ b/codex-rs/core/src/guardian/prompt.rs @@ -87,6 +87,7 @@ pub(crate) async fn build_guardian_prompt_items( ) -> anyhow::Result { build_guardian_prompt_items_with_parent_turn( session, + session.conversation_history_snapshot().await.as_ref(), /*parent_context*/ None, ApprovalRequestReasons { approval: None, @@ -101,6 +102,7 @@ pub(crate) async fn build_guardian_prompt_items( pub(crate) async fn build_guardian_prompt_items_with_parent_turn( session: &Session, + history: &dyn ConversationHistorySnapshot, parent_context: Option<&GuardianReviewContext>, reasons: ApprovalRequestReasons, request: GuardianApprovalRequest, @@ -116,7 +118,6 @@ pub(crate) async fn build_guardian_prompt_items_with_parent_turn( } else { GUARDIAN_MAX_TOOL_ENTRY_TOKENS }; - let history = session.conversation_history_snapshot().await; let root_authorization = session .services .agent_control @@ -127,13 +128,13 @@ pub(crate) async fn build_guardian_prompt_items_with_parent_turn( .services .thread_extension_data .get_or_init(GuardianReviewEvidence::default) - .user_input_snapshot(history.as_ref()) + .user_input_snapshot(history) .fragments; let ComposedContext { authorization, transcript: transcript_entries, } = collect_guardian_context( - &GuardianReviewHistory(history.as_ref()), + &GuardianReviewHistory(history), node_repl_result_token_limit, root_authorization.as_deref().unwrap_or_default(), &trusted_user_inputs, diff --git a/codex-rs/core/src/guardian/review.rs b/codex-rs/core/src/guardian/review.rs index 3347abe33d..116045bc30 100644 --- a/codex-rs/core/src/guardian/review.rs +++ b/codex-rs/core/src/guardian/review.rs @@ -587,14 +587,14 @@ pub(super) async fn run_synchronous_review( .root_user_authorization(session.thread_id) .await .map(|snapshot| snapshot.authorization_version); + // Keep the authorization revision even when no cacheable review evidence exists. + let history = session.conversation_history_snapshot().await; + let user_message_revision = history.user_message_revision(); let review_evidence = if let Some(evidence) = session .services .thread_extension_data .get::() { - // Root rewrites and new user messages during this review make its evidence - // stale even if it later completes against a newer prompt snapshot. - let history = session.conversation_history_snapshot().await; let authorization_version = evidence.authorization_version(history.as_ref()); format_guardian_action_pretty(&request).ok().map(|action| { ( @@ -607,6 +607,7 @@ pub(super) async fn run_synchronous_review( } else { None }; + drop(history); let (mut outcome, analytics_result) = Box::pin(run_guardian_review_session_with_retry_before_deadline( session.clone(), @@ -623,15 +624,21 @@ pub(super) async fn run_synchronous_review( .await; if session.enabled(Feature::GuardianThreadContext) && matches!(&outcome, GuardianReviewOutcome::Completed(assessment) if assessment.outcome == GuardianAssessmentOutcome::Allow) - && root_authorization_version + && (root_authorization_version != session .services .agent_control .root_user_authorization(session.thread_id) .await .map(|snapshot| snapshot.authorization_version) + || user_message_revision + != session + .conversation_history_snapshot() + .await + .user_message_revision()) { - // A completed approval cannot outlive the root evidence it evaluated. + // A completed approval cannot outlive the owning-session or root evidence + // it evaluated, including when either changed before prompt construction. outcome = GuardianReviewOutcome::Error(GuardianReviewError::Cancelled); } @@ -1100,6 +1107,7 @@ async fn run_guardian_review_session_before_deadline( .run_review(GuardianReviewSessionParams { parent_session: Arc::clone(&session), parent_context: context.clone(), + parent_history: session.clone_history().await, spawn_config: session_config.spawn_config, node_repl_policy: session_config.node_repl_policy, request, diff --git a/codex-rs/core/src/guardian/review_session.rs b/codex-rs/core/src/guardian/review_session.rs index 3e99448917..e145d86ef7 100644 --- a/codex-rs/core/src/guardian/review_session.rs +++ b/codex-rs/core/src/guardian/review_session.rs @@ -108,6 +108,8 @@ pub(crate) enum GuardianReviewSessionOutcome { pub(crate) struct GuardianReviewSessionParams { pub(crate) parent_session: Arc, pub(crate) parent_context: GuardianReviewContext, + // Checkpoint selection and thread-owned prompt evidence must use the same history. + pub(crate) parent_history: ContextManager, pub(crate) spawn_config: Config, pub(crate) node_repl_policy: GuardianNodeReplPolicy, pub(crate) request: GuardianApprovalRequest, @@ -572,7 +574,7 @@ impl GuardianReviewSessionManager { params: GuardianReviewSessionParams, ) -> (GuardianReviewSessionOutcome, GuardianReviewAnalyticsResult) { let deadline = params.deadline; - let parent_history = params.parent_session.clone_history().await; + let parent_history = ¶ms.parent_history; let root_authorization_version = if params .parent_session .enabled(Feature::GuardianThreadContext) @@ -588,7 +590,7 @@ impl GuardianReviewSessionManager { None }; let parent_compaction = match encrypted_parent_compaction( - &parent_history, + parent_history, ¶ms.spawn_config.features, params.compaction_model_hash.as_deref(), ) { @@ -1127,8 +1129,17 @@ 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) + { + params.parent_history.conversation_history_snapshot() + } else { + params.parent_session.conversation_history_snapshot().await + }; let mut prompt_items = build_guardian_prompt_items_with_parent_turn( params.parent_session.as_ref(), + history.as_ref(), Some(¶ms.parent_context), params.reasons.clone(), params.request.clone(), diff --git a/codex-rs/core/src/guardian/review_session_tests.rs b/codex-rs/core/src/guardian/review_session_tests.rs index 372bb20f0c..62a6e96de3 100644 --- a/codex-rs/core/src/guardian/review_session_tests.rs +++ b/codex-rs/core/src/guardian/review_session_tests.rs @@ -9,6 +9,130 @@ use codex_protocol::protocol::TurnAbortReason; use codex_protocol::protocol::TurnAbortedEvent; use codex_protocol::protocol::TurnCompleteEvent; +#[tokio::test] +#[allow( + clippy::await_holding_invalid_type, + reason = "hold session selection while the parent compacts to reproduce the race" +)] +async fn run_review_preserves_evidence_during_parent_compaction() { + const EVIDENCE: &str = "The inspected repository is public."; + let (parent, turn, _events) = + crate::session::tests::make_session_and_context_with_auth_and_config_and_rx( + codex_login::CodexAuth::from_api_key("Test API Key"), + Vec::new(), + |config| { + config + .features + .enable(Feature::GuardianThreadContext) + .unwrap(); + config.features.disable(Feature::TokenBudget).unwrap(); + }, + ) + .await; + let mut params = test_review_params().await; + params.spawn_config = build_guardian_review_session_config( + turn.config.as_ref(), + /*live_network_config*/ None, + ¶ms.model, + params.reasoning_effort.clone(), + /*model_messages*/ None, + ) + .unwrap(); + params.parent_session = Arc::clone(&parent); + params.parent_context = GuardianReviewContext::from(Arc::clone(&turn)); + params.compaction_model_hash = Some("matching".to_owned()); + let evidence: ResponseItem = serde_json::from_value(serde_json::json!({ + "type": "function_call_output", "call_id": "prior-inspection", "output": EVIDENCE + })) + .unwrap(); + parent.record_conversation_items(&turn, &[evidence]).await; + params.parent_history = parent.clone_history().await; + + // An idle, prewarmed reviewer is a normal manager state. + let (mut reviewer, tx_event, rx_sub) = test_review_session().await; + reviewer.reuse_key = GuardianReviewSessionReuseKey::from_spawn_config( + ¶ms.spawn_config, + parent.user_instructions().await, + params.parent_history.history_version(), + ) + .with_environments(params.parent_context.environments()) + .with_node_repl_policy_eligibility( + params + .parent_context + .turn() + .model_info() + .node_repl_auto_review_required, + ) + .with_node_repl_policy(¶ms.node_repl_policy); + let manager = GuardianReviewSessionManager { + state: Arc::new(Mutex::new(GuardianReviewSessionState { + trunk: Some(Arc::new(reviewer)), + ephemeral_reviews: Vec::new(), + })), + ..Default::default() + }; + let gate = manager.state.lock().await; + let review = manager.run_review(params); + tokio::pin!(review); + // Earlier awaits use unlocked in-memory state; this polls through checkpoint + // selection and deterministically stops at the held manager mutex. + assert!(futures::poll!(&mut review).is_pending()); + let checkpoint: ResponseItem = serde_json::from_value(serde_json::json!({ + "type": "compaction", "id": "cmp_new", "encrypted_content": "new-checkpoint" + })) + .unwrap(); + let (window_number, window_ids) = parent.advance_auto_compact_window().await; + parent + .replace_compacted_history( + vec![checkpoint.into()], + /*reference_context_item*/ None, + /*world_state_baseline*/ None, + crate::compact::CompactedHistoryMetadata { + message: String::new(), + window_number, + window_ids, + compaction_response_id: None, + compaction_model_hash: Some("matching".to_owned()), + }, + ) + .await; + drop(gate); + + let ((outcome, _), submitted_text) = tokio::join!(review, async { + let submission = rx_sub.recv().await.unwrap(); + let id = submission.id; + let Op::TurnInput { request, reply, .. } = submission.op else { + panic!("expected reviewer prompt"); + }; + let codex_protocol::turn_input::TurnInput::UserInput { content, .. } = request.input else { + panic!("expected user input"); + }; + let text = serde_json::to_string(&content).unwrap(); + reply + .send(Ok(TurnInputSubmission::Started { + turn_id: id.clone(), + })) + .unwrap(); + tx_event + .send(turn_complete_event( + &id, + Some("review finished"), + /*time_to_first_token_ms*/ None, + )) + .await + .unwrap(); + text + }); + assert!(matches!( + outcome, + GuardianReviewSessionOutcome::Completed(Ok(_)) + )); + assert!( + submitted_text.contains(EVIDENCE), + "review must retain evidence from before the concurrent compaction: {submitted_text}" + ); +} + async fn test_review_session() -> ( GuardianReviewSession, async_channel::Sender, @@ -99,6 +223,7 @@ async fn test_review_params() -> GuardianReviewSessionParams { .expect("guardian config"); GuardianReviewSessionParams { + parent_history: session.clone_history().await, parent_session: Arc::new(session), parent_context: GuardianReviewContext::from(Arc::new(turn)), spawn_config, diff --git a/codex-rs/core/src/guardian/tests.rs b/codex-rs/core/src/guardian/tests.rs index 71190f951d..d4810c480a 100644 --- a/codex-rs/core/src/guardian/tests.rs +++ b/codex-rs/core/src/guardian/tests.rs @@ -548,6 +548,7 @@ async fn build_guardian_prompt_prefers_retry_reason_over_approval_reason() -> an let prompt = build_guardian_prompt_items_with_parent_turn( session.as_ref(), + session.conversation_history_snapshot().await.as_ref(), Some(&context), ApprovalRequestReasons { approval: Some("A policy rule requires approval.".to_string()), @@ -589,6 +590,7 @@ async fn build_guardian_prompt_truncates_oversized_approval_reason() -> anyhow:: let prompt = build_guardian_prompt_items_with_parent_turn( session.as_ref(), + session.conversation_history_snapshot().await.as_ref(), Some(&context), ApprovalRequestReasons { approval: Some(approval_reason), @@ -682,6 +684,7 @@ async fn build_guardian_prompt_includes_parent_turn_denied_reads() -> anyhow::Re let prompt = build_guardian_prompt_items_with_parent_turn( session.as_ref(), + session.conversation_history_snapshot().await.as_ref(), Some(&context), ApprovalRequestReasons { approval: None, @@ -1319,6 +1322,7 @@ async fn build_guardian_prompt_items_keeps_required_node_repl_reviews_generic() let prompt = build_guardian_prompt_items_with_parent_turn( session.as_ref(), + session.conversation_history_snapshot().await.as_ref(), Some(&context), ApprovalRequestReasons { approval: None, @@ -1357,6 +1361,7 @@ async fn build_guardian_prompt_items_keeps_other_requests_generic() -> anyhow::R ] { let prompt = build_guardian_prompt_items_with_parent_turn( session.as_ref(), + session.conversation_history_snapshot().await.as_ref(), Some(&context), ApprovalRequestReasons::default(), request, diff --git a/codex-rs/core/tests/suite/guardian_authorization.rs b/codex-rs/core/tests/suite/guardian_authorization.rs index 602dcd2201..e793c5cb21 100644 --- a/codex-rs/core/tests/suite/guardian_authorization.rs +++ b/codex-rs/core/tests/suite/guardian_authorization.rs @@ -1,19 +1,220 @@ //! Ensures Guardian authorization survives compaction and internal context, but not user changes. +use anyhow::Context; use anyhow::Result; use codex_core::TurnInputRequest; +use codex_core::config::Constrained; use codex_core::context::ContextualUserFragment; use codex_core::context::InternalContextSource; use codex_core::context::InternalModelContextFragment; use codex_features::Feature; +use codex_protocol::config_types::ApprovalsReviewer; +use codex_protocol::protocol::AskForApproval; use codex_protocol::protocol::EventMsg; +use codex_protocol::protocol::GuardianAssessmentStatus; use codex_protocol::protocol::Op; +use codex_protocol::protocol::SandboxPolicy; +use codex_protocol::protocol::ThreadSettingsOverrides; +use codex_protocol::request_user_input::RequestUserInputAnswer; +use codex_protocol::request_user_input::RequestUserInputResponse; use codex_protocol::user_input::UserInput; use core_test_support::responses; use core_test_support::skip_if_no_network; +use core_test_support::skip_if_sandbox; +use core_test_support::skip_if_wine_exec; +use core_test_support::streaming_sse::StreamingSseChunk; +use core_test_support::streaming_sse::start_streaming_sse_server; +use core_test_support::test_codex::local_selections; use core_test_support::test_codex::test_codex; use core_test_support::wait_for_event; +use core_test_support::wait_for_event_match; use pretty_assertions::assert_eq; +use serde_json::Value; +use serde_json::json; +use std::collections::HashMap; +use std::time::Duration; +use test_case::test_case; +use tokio::sync::oneshot; + +#[derive(Clone, Copy)] +enum PendingReviewChange { + UserInstruction, + VerifiedAnswer, +} + +#[test_case(PendingReviewChange::UserInstruction; "new user instruction")] +#[test_case(PendingReviewChange::VerifiedAnswer; "verified answer")] +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn guardian_revalidates_owning_session_before_allow( + change: PendingReviewChange, +) -> Result<()> { + skip_if_no_network!(Ok(())); + skip_if_sandbox!(Ok(())); + skip_if_wine_exec!( + Ok(()), + "Guardian approval actions require host-native paths" + ); + + let (parent_completion_tx, parent_completion_rx) = oneshot::channel(); + let (review_completion_tx, review_completion_rx) = oneshot::channel(); + let parent_followup = if matches!(change, PendingReviewChange::VerifiedAnswer) { + responses::sse(vec![ + responses::ev_function_call( + "confirm-command", + "request_user_input", + &json!({"questions": [{ + "id": "execute", "header": "Execute", "question": "May I run the command?", + "options": [ + {"label": "Yes", "description": "Run the command."}, + {"label": "No", "description": "Do not run the command."} + ] + }]}) + .to_string(), + ), + responses::ev_completed("parent-followup"), + ]) + } else { + responses::sse(vec![responses::ev_completed("parent-followup")]) + }; + let (streaming_server, _) = start_streaming_sse_server(vec![ + vec![ + StreamingSseChunk { + gate: None, + body: responses::sse(vec![ + responses::ev_response_created("parent-start"), + responses::ev_custom_tool_call( + "reviewed-tool", + "exec", + "yield_control(); await tools.exec_command({cmd: 'printf authorized', login: false, sandbox_permissions: 'require_escalated', justification: 'Exercise Guardian authorization freshness.'});", + ), + ]), + }, + StreamingSseChunk { + gate: Some(parent_completion_rx), + body: responses::sse(vec![responses::ev_completed("parent-start")]), + }, + ], + vec![StreamingSseChunk { + gate: Some(review_completion_rx), + body: responses::sse(vec![ + responses::ev_response_created("review"), + responses::ev_assistant_message( + "review-result", + &json!({ + "risk_level": "low", "user_authorization": "high", "outcome": "allow", + "rationale": "The user authorized this command in the reviewed snapshot.", + }).to_string(), + ), + responses::ev_completed("review"), + ]), + }], + vec![StreamingSseChunk { gate: None, body: parent_followup }], + vec![StreamingSseChunk { + gate: None, + body: responses::sse(vec![responses::ev_completed("user-change")]), + }], + ]).await; + let base_url = format!("{}/v1", streaming_server.uri()); + let server = responses::start_mock_server().await; + let test = test_codex() + .with_model("test-gpt-5.1-codex") + .with_config(move |config| { + config.model_provider.base_url = Some(base_url); + config.permissions.approval_policy = Constrained::allow_any(AskForApproval::OnRequest); + config.approvals_reviewer = ApprovalsReviewer::AutoReview; + config + .set_legacy_sandbox_policy(SandboxPolicy::new_workspace_write_policy()) + .expect("set sandbox policy"); + for feature in [ + Feature::GuardianThreadContext, + Feature::CodeMode, + Feature::CodeModeInterrupt, + Feature::DefaultModeRequestUserInput, + ] { + config + .features + .enable(feature) + .expect("enable test feature"); + } + }) + .with_code_mode_host_program(codex_utils_cargo_bin::cargo_bin("codex-code-mode-host")?) + .build_with_auto_env(&server) + .await?; + test.codex + .start_or_steer_turn( + TurnInputRequest::user_input(vec![UserInput::Text { + text: "Run the command in a background cell.".into(), + text_elements: Vec::new(), + }]) + .with_thread_settings(ThreadSettingsOverrides { + environments: Some(local_selections(test.config.cwd.clone())), + ..Default::default() + }), + ) + .await?; + + // The parent stream cannot finish until Guardian has captured its prompt, so the + // second request is deterministically the pending review, not a parent follow-up. + tokio::time::timeout( + Duration::from_secs(/*secs*/ 10), + streaming_server.wait_for_request_count(/*count*/ 2), + ) + .await + .context("Guardian review did not start")?; + let review: Value = serde_json::from_slice(&streaming_server.requests().await[1])?; + assert_eq!( + review.pointer("/client_metadata/x-openai-subagent"), + Some(&json!("guardian")) + ); + let before = test.codex.guardian_authorization_version().await; + parent_completion_tx + .send(()) + .expect("release parent completion"); + if matches!(change, PendingReviewChange::VerifiedAnswer) { + let request = wait_for_event_match(&test.codex, |event| match event { + EventMsg::RequestUserInput(request) => Some(request.clone()), + _ => None, + }) + .await; + test.codex + .submit(Op::UserInputAnswer { + id: request.turn_id, + response: RequestUserInputResponse { + answers: HashMap::from([( + "execute".to_owned(), + RequestUserInputAnswer { + answers: vec!["No. Do not run the command.".to_owned()], + }, + )]), + }, + }) + .await?; + } + wait_for_event(&test.codex, |event| { + matches!(event, EventMsg::TurnComplete(_)) + }) + .await; + if matches!(change, PendingReviewChange::UserInstruction) { + test.submit_text_turn("Do not run the command.").await?; + } + assert_ne!(before, test.codex.guardian_authorization_version().await); + review_completion_tx + .send(()) + .expect("release Guardian response"); + let status = wait_for_event_match(&test.codex, |event| match event { + EventMsg::GuardianAssessment(assessment) + if assessment.status != GuardianAssessmentStatus::InProgress => + { + Some(assessment.status) + } + _ => None, + }) + .await; + assert_eq!(status, GuardianAssessmentStatus::Aborted); + test.codex.shutdown_and_wait().await?; + streaming_server.shutdown().await; + Ok(()) +} #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn guardian_authorization_revision_survives_compaction_not_user_input_or_rollback() diff --git a/codex-rs/core/tests/suite/guardian_review.rs b/codex-rs/core/tests/suite/guardian_review.rs index 538def3138..ea0a27b8b4 100644 --- a/codex-rs/core/tests/suite/guardian_review.rs +++ b/codex-rs/core/tests/suite/guardian_review.rs @@ -287,10 +287,11 @@ async fn guardian_session_inherits_parent_http_fallback( } #[tokio::test(flavor = "multi_thread", worker_threads = 2)] -#[test_case(60_000; "reviewer_window_is_already_exhausted")] -#[test_case(49_990; "followup_reminder_exhausts_reviewer_window")] +#[test_case(60_000, false; "legacy_reviewer_window_is_already_exhausted")] +#[test_case(49_990, true; "retained_followup_reminder_exhausts_reviewer_window")] async fn guardian_review_resends_full_transcript_after_reviewer_context_rollover( first_review_total_tokens: i64, + thread_owned: bool, ) -> Result<()> { skip_if_no_network!(Ok(())); skip_if_wine_exec!( @@ -303,7 +304,11 @@ async fn guardian_review_resends_full_transcript_after_reviewer_context_rollover .with_model_info_override("gpt-5.5", |model| { model.auto_review_model_override = Some(model.slug.clone()); }) - .with_config(|config| { + .with_config(move |config| { + config + .features + .set_enabled(Feature::GuardianThreadContext, thread_owned) + .expect("configure Guardian context mode"); config.model_context_window = Some(100_000); config.model_auto_compact_token_limit = Some(50_000); config.permissions.approval_policy = Constrained::allow_any(AskForApproval::OnRequest);