diff --git a/codex-rs/app-server/tests/suite/v2/guardian_v2.rs b/codex-rs/app-server/tests/suite/v2/guardian_v2.rs index bfd3d8e6ac..2d3b581daf 100644 --- a/codex-rs/app-server/tests/suite/v2/guardian_v2.rs +++ b/codex-rs/app-server/tests/suite/v2/guardian_v2.rs @@ -184,6 +184,8 @@ struct MockResponsesState { allow_luna: Notify, allow_guardian_review: Notify, gate_each_guardian_review: bool, + gate_second_tool_on_classification: bool, + allow_second_tool: Notify, classification_completed: Notify, truncation_recorded: Notify, context_metric_bounds: Mutex>>, @@ -520,6 +522,9 @@ async fn parent_response( .contains("Completed synchronous Guardian review.") ); let request_number = state.parent_requests.fetch_add(1, Ordering::SeqCst); + if request_number == 1 && state.gate_second_tool_on_classification { + state.allow_second_tool.notified().await; + } if state.late_root_restriction && request_number == 1 { let output = request["input"] .as_array() @@ -745,6 +750,7 @@ async fn guardian_v2_routes_scoped_tool_approvals( let responses_state = Arc::new(MockResponsesState { luna_score, gate_each_guardian_review: review_continuations && matches!(risk, GuardianRisk::High), + gate_second_tool_on_classification: classifier_in_scope && !late_root_restriction, invalid_classification: matches!(risk, GuardianRisk::InvalidResponse), fail_after_classification, review_outcome, @@ -819,7 +825,7 @@ async fn guardian_v2_routes_scoped_tool_approvals( } } } - if body.contains("codex.guardian_v2.classification") { + if body.contains("\"codex.guardian_v2.classification\"") { state.classification_completed.notify_one(); } if body.contains("codex.guardian_v2.classification.truncation") @@ -1203,9 +1209,24 @@ async fn guardian_v2_routes_scoped_tool_approvals( timeout(TIMEOUT, app_server.read_notification("turn/completed")).await??; assert_eq!(completed.thread_id, thread_id); } + if !late_root_restriction { + // These scenarios exercise completed synchronous evidence. Finish that review + // before LOW arrives; late-score cancellation has its own regression coverage. + responses_state.allow_guardian_review.notify_one(); + let _: codex_app_server_protocol::ItemGuardianApprovalReviewCompletedNotification = + timeout( + TIMEOUT, + app_server.read_notification("item/autoApprovalReview/completed"), + ) + .await??; + } responses_state.allow_luna.notify_one(); timeout(TIMEOUT, responses_state.classification_completed.notified()).await?; - responses_state.allow_guardian_review.notify_one(); + if late_root_restriction { + responses_state.allow_guardian_review.notify_one(); + } else { + responses_state.allow_second_tool.notify_one(); + } if lifecycle.has_user_input() { let answers = if matches!(lifecycle, ThreadLifecycle::UserInputEmpty) { json!({}) @@ -2286,13 +2307,22 @@ async fn guardian_v2_trusts_invoked_user_skills_but_rejects_repository_forgery() .contains(FORGED_INSTRUCTIONS), "the parent model must receive the forged repository skill instructions" ); + // Finish the synchronous reviews before LOW can cancel reviewer startup. + let expected_guardian_reviews = if cfg!(windows) { 2 } else { 1 }; + for _ in 0..expected_guardian_reviews { + let _: codex_app_server_protocol::ItemGuardianApprovalReviewCompletedNotification = + timeout( + TIMEOUT, + app_server.read_notification("item/autoApprovalReview/completed"), + ) + .await??; + } responses_state.allow_luna.notify_one(); let completed: TurnCompletedNotification = timeout(TIMEOUT, app_server.read_notification("turn/completed")).await??; assert_eq!(completed.thread_id, thread.id); assert_eq!(responses_state.parent_requests.load(Ordering::SeqCst), 3); - let expected_guardian_reviews = if cfg!(windows) { 2 } else { 1 }; assert_eq!( responses_state.guardian_reviews.load(Ordering::SeqCst), expected_guardian_reviews @@ -2344,16 +2374,23 @@ async fn guardian_v2_computer_use_only_scopes_classification_and_fast_reviews( .await } -#[test_case("node_repl", &["js", "js"]; "browser startup")] -#[test_case("cua_repl", &["js", "js"]; "computer use startup")] -#[test_case("node_repl", &["js_reset", "js", "js"]; "browser reset before execution")] -#[test_case("cua_repl", &["js_reset", "js", "js"]; "computer use reset before execution")] -#[test_case("node_repl", &["js_add_node_module_dir", "js", "js"]; "browser setup before execution")] -#[test_case("cua_repl", &["js_add_node_module_dir", "js", "js"]; "computer use setup before execution")] +enum ApprovalCompletion { + Sync, + AsyncLow, +} + +#[test_case("cua_repl", &["js", "js"], ApprovalCompletion::AsyncLow; "late low cancels review")] +#[test_case("node_repl", &["js", "js"], ApprovalCompletion::Sync; "browser startup")] +#[test_case("cua_repl", &["js", "js"], ApprovalCompletion::Sync; "computer use startup")] +#[test_case("node_repl", &["js_reset", "js", "js"], ApprovalCompletion::Sync; "browser reset before execution")] +#[test_case("cua_repl", &["js_reset", "js", "js"], ApprovalCompletion::Sync; "computer use reset before execution")] +#[test_case("node_repl", &["js_add_node_module_dir", "js", "js"], ApprovalCompletion::Sync; "browser setup before execution")] +#[test_case("cua_repl", &["js_add_node_module_dir", "js", "js"], ApprovalCompletion::Sync; "computer use setup before execution")] #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn first_cua_review_does_not_wait_for_initial_score( server_name: &'static str, tool_sequence: &'static [&'static str], + completion: ApprovalCompletion, ) -> Result<()> { skip_if_no_network!(Ok(())); let state = Arc::new(MockResponsesState { @@ -2430,7 +2467,24 @@ async fn first_cua_review_does_not_wait_for_initial_score( state.parent_requests.load(Ordering::SeqCst), tool_sequence.len() ); - state.allow_guardian_review.notify_one(); + if !matches!(completion, ApprovalCompletion::Sync) { + state.allow_luna.notify_one(); + let review: codex_app_server_protocol::ItemGuardianApprovalReviewCompletedNotification = + timeout( + TIMEOUT, + app_server.read_notification("item/autoApprovalReview/completed"), + ) + .await??; + assert_eq!( + (review.review.status, review.review.risk_level), + ( + codex_app_server_protocol::GuardianApprovalReviewStatus::Approved, + None + ) + ); + } else { + state.allow_guardian_review.notify_one(); + } let completed: TurnCompletedNotification = timeout(TIMEOUT, app_server.read_notification("turn/completed")).await??; assert_eq!(completed.thread_id, thread.id); diff --git a/codex-rs/core/src/guardian/review_request.rs b/codex-rs/core/src/guardian/review_request.rs index 746b4caba4..2f4783355e 100644 --- a/codex-rs/core/src/guardian/review_request.rs +++ b/codex-rs/core/src/guardian/review_request.rs @@ -240,11 +240,24 @@ impl ReviewHost for super::super::runtime::ReviewRuntime { let review_evidence = match &outcome { GuardianReviewOutcome::Completed(_) => review_evidence, - GuardianReviewOutcome::Error(_) => None, + GuardianReviewOutcome::CachedApproval | GuardianReviewOutcome::Error(_) => None, }; (outcome, analytics, review_evidence) } + async fn cached_approval_is_current(&self, prepared: &PreparedApproval) -> bool { + let root = self + .session + .services + .agent_control + .get_guardian_package(self.session.thread_id) + .await; + !self.history_reset.is_cancelled() + && self.validate_action().is_ok() + && prepared.root_history + == root.map(|snapshot| (snapshot.root_thread_id, snapshot.history_reset_version)) + } + fn validate_action(&self) -> Result<(&str, Option<&str>), ReviewDecision> { let request = self.request.validate(&self.context)?; Ok(( diff --git a/codex-rs/ext/extension-api/src/contributors/approval_review.rs b/codex-rs/ext/extension-api/src/contributors/approval_review.rs index 2f813c0849..cb37dac7d6 100644 --- a/codex-rs/ext/extension-api/src/contributors/approval_review.rs +++ b/codex-rs/ext/extension-api/src/contributors/approval_review.rs @@ -18,14 +18,16 @@ pub enum ApprovalDecision { AskUser, } -/// Runs an extension-owned synchronous review already bound to an action by the host. -/// Implementations must not reuse an async score. `None` requests the host user -/// flow when automatic review exhausts its input budget and host policy permits it. +/// Runs an extension-owned review already bound to an action by the host. +/// `None` requests the host user flow when automatic review exhausts its input budget +/// and host policy permits it. An optional async approval resolves only when current +/// cached evidence authorizes the action; it races the complete synchronous review. pub trait SynchronousApprovalReviewer: Send + Sync { - fn review( - &self, + fn review<'a>( + &'a self, reason: codex_protocol::approvals::GuardianReviewReason, - ) -> crate::ExtensionFuture<'_, Option>; + async_approval: Option>, + ) -> crate::ExtensionFuture<'a, Option>; } /// Inputs to Guardian's policy choice. Conversation and scores stay thread-owned. diff --git a/codex-rs/ext/extension-api/tests/registry.rs b/codex-rs/ext/extension-api/tests/registry.rs index daf919269a..e9a86b9cb3 100644 --- a/codex-rs/ext/extension-api/tests/registry.rs +++ b/codex-rs/ext/extension-api/tests/registry.rs @@ -165,10 +165,11 @@ impl ApprovalReviewContributor for AllContributors { } impl codex_extension_api::SynchronousApprovalReviewer for AllContributors { - fn review( - &self, + fn review<'a>( + &'a self, _reason: codex_protocol::approvals::GuardianReviewReason, - ) -> ExtensionFuture<'_, Option> { + _async_approval: Option>, + ) -> ExtensionFuture<'a, Option> { Box::pin(std::future::ready(Some( codex_protocol::protocol::ReviewDecision::Approved, ))) diff --git a/codex-rs/ext/guardian-reviewer/src/completion.rs b/codex-rs/ext/guardian-reviewer/src/completion.rs index 169e8b2e4b..72befe0bd1 100644 --- a/codex-rs/ext/guardian-reviewer/src/completion.rs +++ b/codex-rs/ext/guardian-reviewer/src/completion.rs @@ -44,9 +44,24 @@ pub fn complete_review( ) -> ReviewCompletion { let completed_assessment = match &outcome { GuardianReviewOutcome::Completed(assessment) => Some(assessment.outcome), - GuardianReviewOutcome::Error(_) => None, + GuardianReviewOutcome::CachedApproval | GuardianReviewOutcome::Error(_) => None, }; let assessment = match outcome { + GuardianReviewOutcome::CachedApproval => { + event.status = GuardianAssessmentStatus::Approved; + event.decision_source = None; + event.rationale = Some("Approved using a current low-risk async score.".to_owned()); + analytics.decision = GuardianReviewDecision::Approved; + analytics.terminal_status = GuardianReviewTerminalStatus::Approved; + analytics.failure_reason = None; + return ReviewCompletion { + decision: Some(ReviewDecision::Approved), + event, + warning: None, + analytics, + assessment_outcome: None, + }; + } GuardianReviewOutcome::Completed(assessment) => { let approved = matches!(assessment.outcome, GuardianAssessmentOutcome::Allow); analytics.decision = if approved { diff --git a/codex-rs/ext/guardian-reviewer/src/outcome.rs b/codex-rs/ext/guardian-reviewer/src/outcome.rs index aa97f16be4..882a38f40b 100644 --- a/codex-rs/ext/guardian-reviewer/src/outcome.rs +++ b/codex-rs/ext/guardian-reviewer/src/outcome.rs @@ -7,6 +7,8 @@ use tokio::time::Instant; #[derive(Debug)] pub enum GuardianReviewOutcome { + /// Approved by current async evidence after the redundant attempt was drained. + CachedApproval, Completed(GuardianAssessment), Error(GuardianReviewError), } diff --git a/codex-rs/ext/guardian-reviewer/src/retry.rs b/codex-rs/ext/guardian-reviewer/src/retry.rs index 3adcb4d83a..2dbb61a45b 100644 --- a/codex-rs/ext/guardian-reviewer/src/retry.rs +++ b/codex-rs/ext/guardian-reviewer/src/retry.rs @@ -125,7 +125,8 @@ fn should_retry_guardian_review(outcome: &GuardianReviewOutcome) -> bool { | CodexErrorInfo::ThreadRollbackFailed | CodexErrorInfo::Other => false, }, - GuardianReviewOutcome::Completed(_) + GuardianReviewOutcome::CachedApproval + | GuardianReviewOutcome::Completed(_) | GuardianReviewOutcome::Error( GuardianReviewError::InputBudgetExceeded | GuardianReviewError::PromptBuild { .. } diff --git a/codex-rs/ext/guardian-reviewer/src/review.rs b/codex-rs/ext/guardian-reviewer/src/review.rs index 7938f7f201..e8bcd4bb2a 100644 --- a/codex-rs/ext/guardian-reviewer/src/review.rs +++ b/codex-rs/ext/guardian-reviewer/src/review.rs @@ -57,6 +57,11 @@ pub trait ReviewHost: Send + Sync { Option, ), > + Send; + /// Revalidates the original action and history lifetime before using cached approval. + fn cached_approval_is_current( + &self, + prepared: &Self::Prepared, + ) -> impl Future + Send; fn emit(&self, event: EventMsg) -> impl Future + Send; fn record_evidence( &self, @@ -73,7 +78,11 @@ pub trait ReviewHost: Send + Sync { } impl SynchronousApprovalReviewer for ReviewRequest<'_, H> { - fn review(&self, reason: GuardianReviewReason) -> ExtensionFuture<'_, Option> { + fn review<'a>( + &'a self, + reason: GuardianReviewReason, + async_approval: Option>, + ) -> ExtensionFuture<'a, Option> { Box::pin(async move { let deadline = Instant::now() + crate::REVIEW_TIMEOUT; let (prepared, report) = match self @@ -87,24 +96,40 @@ impl SynchronousApprovalReviewer for ReviewRequest<'_, H> { self.host .emit(EventMsg::GuardianAssessment(report.started_event())) .await; - let (outcome, analytics, evidence) = if self.cancellation.is_cancelled() { - ( - GuardianReviewOutcome::Error(GuardianReviewError::Cancelled), - GuardianReviewAnalyticsResult::without_session(), - None, - ) + let cancellation = self.cancellation.child_token(); + let mut review = Box::pin(crate::run_with_retry( + GuardianReviewSessionLimits { + max_attempts: crate::MAX_REVIEW_ATTEMPTS, + deadline, + }, + Some(&cancellation), + |deadline| self.host.attempt(&prepared, deadline, &cancellation), + )); + let (outcome, analytics, evidence) = if let Some(async_approval) = async_approval { + tokio::select! { + biased; + result = &mut review => result, + () = async_approval => { + // The async approval won. Finish cancellation through the existing + // reviewer cleanup; the retry loop is never restarted. + cancellation.cancel(); + let (_, analytics, _) = review.await; + let outcome = if !self.host.cached_approval_is_current(&prepared).await + || self.cancellation.is_cancelled() { + GuardianReviewOutcome::Error(GuardianReviewError::Cancelled) + } else if Instant::now() >= deadline { + GuardianReviewOutcome::Error(GuardianReviewError::Timeout) + } else { + GuardianReviewOutcome::CachedApproval + }; + (outcome, analytics, None) + } + } } else { - Box::pin(crate::run_with_retry( - GuardianReviewSessionLimits { - max_attempts: crate::MAX_REVIEW_ATTEMPTS, - deadline, - }, - Some(&self.cancellation), - |deadline| self.host.attempt(&prepared, deadline, &self.cancellation), - )) - .await + review.await }; let completed_at_ms = codex_analytics::now_unix_millis(); + let cached_approval = matches!(outcome, GuardianReviewOutcome::CachedApproval); let completed = report.complete( outcome, self.model, @@ -112,7 +137,7 @@ impl SynchronousApprovalReviewer for ReviewRequest<'_, H> { analytics, completed_at_ms.try_into().unwrap_or_default(), ); - if self.log_assessments { + if self.log_assessments && !cached_approval { self.telemetry .guardian_assessment(&completed.event, completed.assessment_outcome); } @@ -137,6 +162,9 @@ impl SynchronousApprovalReviewer for ReviewRequest<'_, H> { self.host .emit(EventMsg::GuardianAssessment(completed.event)) .await; + if cached_approval { + return Some(self.cached_approval().await); + } if let Some((turn_id, model)) = self.host.servicing_turn().await { let denials = ReviewDenials::for_thread(self.thread_store); if completed.assessment_outcome == Some(GuardianAssessmentOutcome::Deny) { diff --git a/codex-rs/ext/guardian-reviewer/src/routing.rs b/codex-rs/ext/guardian-reviewer/src/routing.rs index aed5e95409..c33aa1ae7e 100644 --- a/codex-rs/ext/guardian-reviewer/src/routing.rs +++ b/codex-rs/ext/guardian-reviewer/src/routing.rs @@ -121,7 +121,11 @@ impl ReviewRequest<'_, H> { Some(self.cached_approval().await) } Some(ApprovalDecision::Allow) => { - self.review(GuardianReviewReason::FreshRequired).await + self.review( + GuardianReviewReason::FreshRequired, + /*async_approval*/ None, + ) + .await } Some(ApprovalDecision::AskUser) if !self.require_guardian => None, None if !self.require_guardian @@ -133,7 +137,8 @@ impl ReviewRequest<'_, H> { None } None | Some(ApprovalDecision::AskUser) => { - self.review(GuardianReviewReason::Policy).await + self.review(GuardianReviewReason::Policy, /*async_approval*/ None) + .await } } }; @@ -163,7 +168,7 @@ impl ReviewRequest<'_, H> { } } - async fn cached_approval(&self) -> ReviewDecision { + pub(super) async fn cached_approval(&self) -> ReviewDecision { let (turn_id, item_id) = match self.host.validate_action() { Ok(target) => target, Err(decision) => return decision, 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 1f41cd6cb9..c979adb7d2 100644 --- a/codex-rs/ext/guardian-v2/src/async_scorer/approval.rs +++ b/codex-rs/ext/guardian-v2/src/async_scorer/approval.rs @@ -15,6 +15,7 @@ use codex_extension_api::ApprovalDecision; use codex_extension_api::ApprovalDecisionInput; use codex_extension_api::ApprovalReviewContributor; use codex_extension_api::ExtensionFuture; +use codex_extension_api::ExtensionMetrics; use codex_protocol::approvals::GuardianReviewReason; use codex_protocol::config_types::ApprovalsReviewer; use codex_protocol::mcp::is_node_repl_backed_connector; @@ -100,10 +101,14 @@ impl GuardianApprovalReviewer { } let reason = if mode == GuardianReviewMode::Adaptive && !input.require_fresh_review { match guardian_config.as_ref() { - Some(config) => match cached_evidence(thread, input, config, &policy).await { - Ok(()) => return ApprovalDecision::Allow, - Err(reason) => reason, - }, + Some(config) => { + match cached_evidence(thread, input, config, &policy, input.metrics.as_deref()) + .await + { + Ok(()) => return ApprovalDecision::Allow, + Err(reason) => reason, + } + } None => { record_fast_decision(input.metrics.as_deref(), "deferred", "scoring_failure"); GuardianReviewReason::ScoringFailure @@ -119,7 +124,33 @@ impl GuardianApprovalReviewer { ?reason, "reviewing approval" ); - match input.synchronous_reviewer.review(reason).await { + let async_approval = async { + if mode == GuardianReviewMode::Adaptive + && !input.require_fresh_review + && let Some(config) = guardian_config.as_ref() + && let Some(progress) = input.thread_store.get::() + { + // Subscribe before checking so a concurrently published score is not lost. + let mut updates = progress.updates.subscribe(); + loop { + if cached_evidence(thread, input, config, &policy, /*metrics*/ None) + .await + .is_ok() + { + return; + } + if updates.changed().await.is_err() { + break; + } + } + } + std::future::pending::<()>().await; + }; + match input + .synchronous_reviewer + .review(reason, Some(Box::pin(async_approval))) + .await + { Some(decision) => ApprovalDecision::Reviewed(decision), None => ApprovalDecision::AskUser, } @@ -131,9 +162,9 @@ async fn cached_evidence( input: &ApprovalDecisionInput<'_>, config: &GuardianV2Config, policy: &GuardianModelPolicy, + metrics: Option<&dyn ExtensionMetrics>, ) -> Result<(), GuardianReviewReason> { let store = input.thread_store; - let metrics = input.metrics.as_deref(); let Some(progress) = store.get::() else { record_fast_decision(metrics, "deferred", "missing_score"); return Err(GuardianReviewReason::MissingScore); 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 772e7189e2..768719fe3e 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 @@ -3355,10 +3355,11 @@ async fn assert_parent_compaction_reuse(parent_context_for_review: bool) -> Resu struct CacheMiss; impl codex_extension_api::SynchronousApprovalReviewer for CacheMiss { - fn review( - &self, + fn review<'a>( + &'a self, _reason: codex_protocol::approvals::GuardianReviewReason, - ) -> codex_extension_api::ExtensionFuture<'_, Option> { + _async_approval: Option>, + ) -> codex_extension_api::ExtensionFuture<'a, Option> { Box::pin(async { Some(ReviewDecision::denied("cache miss")) }) } } diff --git a/codex-rs/ext/guardian-v2/src/async_scorer/score.rs b/codex-rs/ext/guardian-v2/src/async_scorer/score.rs index ade8ece519..99acdea8b7 100644 --- a/codex-rs/ext/guardian-v2/src/async_scorer/score.rs +++ b/codex-rs/ext/guardian-v2/src/async_scorer/score.rs @@ -18,6 +18,7 @@ use super::wrapper_lag::WrapperLag; #[derive(Default)] pub(super) struct GuardianV2ScoreProgress { state: Mutex, + pub(super) updates: tokio::sync::watch::Sender<()>, pub(super) metrics: Option>, } @@ -116,6 +117,7 @@ impl GuardianV2ScoreProgress { state.score = Some(score); state.authorization = Some(authorization); state.latest_scored_tool_call = state.latest_scored_tool_call.max(index); + self.updates.send_replace(()); } accepted }