Let late low-risk scores complete pending Guardian reviews (#51396)

## Why

A synchronous Guardian review could keep an action waiting even after an asynchronous score supplied valid low-risk evidence for approval.

## What changed

- In adaptive mode, race the synchronous review against score updates that satisfy the existing cached-approval checks, unless a fresh review is required.
- When asynchronous approval wins, cancel and drain the pending review, then revalidate the action and history lifetime and enforce cancellation and timeout limits before approving.
- Emit an approved review completion without a synchronous assessment or risk level, and use the cached-approval path for the final decision.

## Testing

Add an app-server regression case that holds the synchronous review pending, releases a low-risk score, and verifies approval completes without a risk level. Sequence existing scenarios to finish synchronous reviews before releasing classification results when they test synchronous evidence.

GitOrigin-RevId: 04f5ad6b960b819b565fccb1b85c8039ccbd0479

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<BTreeMap<(String, String), Option<f64>>>,
@@ -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);

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((

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<codex_protocol::protocol::ReviewDecision>>;
+        async_approval: Option<crate::ExtensionFuture<'a, ()>>,
+    ) -> crate::ExtensionFuture<'a, Option<codex_protocol::protocol::ReviewDecision>>;
 }
 
 /// Inputs to Guardian's policy choice. Conversation and scores stay thread-owned.

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<codex_protocol::protocol::ReviewDecision>> {
+        _async_approval: Option<codex_extension_api::ExtensionFuture<'a, ()>>,
+    ) -> ExtensionFuture<'a, Option<codex_protocol::protocol::ReviewDecision>> {
         Box::pin(std::future::ready(Some(
             codex_protocol::protocol::ReviewDecision::Approved,
         )))

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 {

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),
 }

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 { .. }

codex-rs/ext/guardian-reviewer/src/review.rs

@@ -57,6 +57,11 @@ pub trait ReviewHost: Send + Sync {
             Option<Self::Evidence>,
         ),
     > + Send;
+    /// Revalidates the original action and history lifetime before using cached approval.
+    fn cached_approval_is_current(
+        &self,
+        prepared: &Self::Prepared,
+    ) -> impl Future<Output = bool> + Send;
     fn emit(&self, event: EventMsg) -> impl Future<Output = ()> + Send;
     fn record_evidence(
         &self,
@@ -73,7 +78,11 @@ pub trait ReviewHost: Send + Sync {
 }
 
 impl<H: ReviewHost> SynchronousApprovalReviewer for ReviewRequest<'_, H> {
-    fn review(&self, reason: GuardianReviewReason) -> ExtensionFuture<'_, Option<ReviewDecision>> {
+    fn review<'a>(
+        &'a self,
+        reason: GuardianReviewReason,
+        async_approval: Option<ExtensionFuture<'a, ()>>,
+    ) -> ExtensionFuture<'a, Option<ReviewDecision>> {
         Box::pin(async move {
             let deadline = Instant::now() + crate::REVIEW_TIMEOUT;
             let (prepared, report) = match self
@@ -87,24 +96,40 @@ impl<H: ReviewHost> 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<H: ReviewHost> 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<H: ReviewHost> 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) {

codex-rs/ext/guardian-reviewer/src/routing.rs

@@ -121,7 +121,11 @@ impl<H: ReviewHost> 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<H: ReviewHost> 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<H: ReviewHost> 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,

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::<GuardianV2ScoreProgress>()
+            {
+                // 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::<GuardianV2ScoreProgress>() else {
         record_fast_decision(metrics, "deferred", "missing_score");
         return Err(GuardianReviewReason::MissingScore);

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<ReviewDecision>> {
+        _async_approval: Option<codex_extension_api::ExtensionFuture<'a, ()>>,
+    ) -> codex_extension_api::ExtensionFuture<'a, Option<ReviewDecision>> {
         Box::pin(async { Some(ReviewDecision::denied("cache miss")) })
     }
 }

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<ScoreState>,
+    pub(super) updates: tokio::sync::watch::Sender<()>,
     pub(super) metrics: Option<Arc<dyn ExtensionMetrics>>,
 }
 
@@ -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
     }