From 7f01a84effccef40d4726c3ca12e6c839ec98d7a Mon Sep 17 00:00:00 2001 From: jif Date: Tue, 15 Sep 2026 12:44:37 +0000 Subject: [PATCH] Move Guardian approval routing into the reviewer extension (#45693) ## What changed Introduce `ReviewRequest` in `codex-guardian-reviewer` to own contributor routing, cached approvals, synchronous fallback, and review cancellation. Keep action validation and session-specific preparation in the host. Track the complete approval operation through parent shutdown, driving cancellation through reporting and reviewer cleanup before releasing it. Preserve fresh-review requirements and validate cached approvals before recording their outcome. ## Testing Extend coverage for cancellation before routing and during cached approval, parent shutdown cleanup, and required fresh review overriding a cached allow result with matching assessment events and a denial warning. GitOrigin-RevId: 8dcf26cc3660e91ef94e28e0bee7327ebe34ae6c --- codex-rs/core/src/guardian/decision.rs | 141 ++----------- codex-rs/core/src/guardian/review.rs | 28 +-- codex-rs/core/src/guardian/review_request.rs | 31 ++- codex-rs/core/src/guardian/runtime.rs | 1 - codex-rs/core/src/guardian/test_host.rs | 5 +- codex-rs/core/src/guardian/tests.rs | 51 ++++- .../core/src/session/tests/guardian_tests.rs | 38 +++- codex-rs/core/tests/suite/guardian_review.rs | 69 ++++++- codex-rs/ext/guardian-reviewer/src/lib.rs | 5 +- codex-rs/ext/guardian-reviewer/src/review.rs | 38 ++-- codex-rs/ext/guardian-reviewer/src/routing.rs | 185 ++++++++++++++++++ .../guardian-v2/src/async_scorer/approval.rs | 2 +- 12 files changed, 387 insertions(+), 207 deletions(-) create mode 100644 codex-rs/ext/guardian-reviewer/src/routing.rs diff --git a/codex-rs/core/src/guardian/decision.rs b/codex-rs/core/src/guardian/decision.rs index 0f9332d2fc..e12fa15dd1 100644 --- a/codex-rs/core/src/guardian/decision.rs +++ b/codex-rs/core/src/guardian/decision.rs @@ -2,20 +2,14 @@ //! The synchronous service captures one action; no outcome is stored by tool-call ID. use codex_async_utils::THREAD_STACK_SIZE_BYTES; -use codex_extension_api::ApprovalDecision; -use codex_extension_api::ApprovalDecisionInput; -use codex_extension_api::SynchronousApprovalReviewer; -use codex_protocol::approvals::GuardianReviewReason; use codex_protocol::protocol::ReviewDecision; use std::sync::Arc; use tokio::sync::oneshot; -use tokio_util::sync::CancellationToken; use super::ApprovalRequestReasons; use super::GuardianApprovalRequest; use super::GuardianReviewContext; use super::GuardianReviewOptions; -use super::review::record_guardian_non_denial; use super::runtime::ReviewAction; use super::runtime::ReviewRuntime; use crate::session::session::Session; @@ -56,17 +50,6 @@ pub(crate) async fn decide_approval( reasons: ApprovalRequestReasons, mut options: GuardianReviewOptions, ) -> Option { - let runtime = session - .services - .thread_extension_data - .get::(); - let _task = runtime.as_ref().map(|runtime| runtime.tasks.token()); - if runtime - .as_ref() - .is_some_and(|runtime| runtime.cancellation.is_cancelled()) - { - return Some(ReviewDecision::Abort); - } let context = context.into(); let request = request.into(); let (_, history_reset) = session.history_reset().await; @@ -75,14 +58,6 @@ pub(crate) async fn decide_approval( None => history_reset.child_token(), }; let _cancel_on_drop = cancellation.clone().drop_guard(); - let review_cancellation = match runtime.as_ref() { - Some(runtime) => { - crate::exec::cancel_when_either(runtime.cancellation.clone(), cancellation.clone()) - } - None => cancellation.clone(), - }; - let _review_cancel_on_drop = review_cancellation.clone().drop_guard(); - options.external_cancel = Some(review_cancellation.clone()); let turn = context.turn(); let live_config = session.get_config().await; let requirements = live_config.config_layer_stack.requirements(); @@ -94,62 +69,21 @@ pub(crate) async fn decide_approval( .approvals_reviewer .can_set(&codex_protocol::config_types::ApprovalsReviewer::User) .is_err(); - let require_fresh_review = options.require_synchronous_review - || model_requires_review - && !turn - .config - .features - .enabled(codex_features::Feature::GuardianV2) - || options - .external_cancel - .as_ref() - .is_some_and(CancellationToken::is_cancelled) - || reasons.retry.is_some() - || matches!(&request.request, Ok(GuardianApprovalRequest::ExecCommand { sandbox_permissions, .. }) - if sandbox_permissions.requires_escalated_permissions()); let full_access = context.environments().has_full_access( context.approval_policy, &turn.config.permissions.effective_permission_profile(), ); - if full_access { - return Some(if let Err(decision) = request.validate(&context) { - decision - } else if options - .external_cancel - .as_ref() - .is_some_and(CancellationToken::is_cancelled) - { - ReviewDecision::Abort - } else { - ReviewDecision::Approved - }); - } - let action = match &request.action { - Ok(action) => action, - Err(_) => { - return Some(ReviewDecision::denied( - "automatic approval review could not prepare the action", - )); - } - }; - let runtime = codex_guardian_reviewer::SynchronousReview { + let require_synchronous_review = options.require_synchronous_review; + let retried = reasons.retry.is_some(); + let decision = codex_guardian_reviewer::ReviewRequest { host: ReviewRuntime { session: Arc::clone(&session), history_reset: history_reset.clone(), context: context.clone(), - review_id: review_id.clone(), request: request.clone(), reasons, options, }, - thread_store: &session.services.thread_extension_data, - // Keep the existing turn-level reporting policy during this ownership move. - model: turn.model_info(), - require_guardian, - telemetry: &session.services.session_telemetry, - analytics: &session.services.analytics_events_client, - }; - let input = ApprovalDecisionInput { approval_id: &review_id, tool_call_id: request .request @@ -165,71 +99,32 @@ pub(crate) async fn decide_approval( GuardianApprovalRequest::Execve { .. } => None, _ => super::approval_request::guardian_request_target_item_id(request), }), - action, + action: request.action.as_ref().ok(), thread_id: session.thread_id, thread_store: &session.services.thread_extension_data, category: request.category, approval_policy: context.approval_policy, approvals_reviewer: context.approvals_reviewer, require_guardian, - require_fresh_review, + require_synchronous_review, + model_requires_review, + async_enabled: turn.config.features.enabled(codex_features::Feature::GuardianV2), + retried, + escalated_exec: matches!(&request.request, Ok(GuardianApprovalRequest::ExecCommand { sandbox_permissions, .. }) if sandbox_permissions.requires_escalated_permissions()), full_access, + cancellation: cancellation.clone(), + // Keep the existing turn-level reporting policy during this ownership move. + model: turn.model_info(), + telemetry: &session.services.session_telemetry, + analytics: &session.services.analytics_events_client, metrics: Some(crate::session::extension_metrics::from_session_telemetry( turn.session_telemetry.clone(), )), - synchronous_reviewer: &runtime, - }; - let decision = match session.services.extensions.decide_approval(&input).await { - Some(ApprovalDecision::Reviewed(decision)) => Some(decision), - Some(ApprovalDecision::Allow) if !require_fresh_review => { - let request = match request.validate(&context) { - Ok(request) => request, - Err(decision) => return Some(decision), - }; - let turn_id = super::approval_request::guardian_request_turn_id(request, &turn.sub_id); - if session - .services - .thread_extension_data - .get::() - .is_some() - { - session - .services - .analytics_events_client - .track_guardian_v2_event(codex_analytics::GuardianV2Event { - thread_id: session.thread_id.to_string(), - turn_id: turn_id.to_owned(), - item_id: super::approval_request::guardian_request_target_item_id(request) - .map(str::to_owned), - model: Some(turn.model_info().slug.clone()), - occurred_at_ms: codex_analytics::now_unix_millis(), - kind: codex_analytics::GuardianV2EventKind::FastDecision { - decision: "approved", - }, - }); - } - record_guardian_non_denial(&session).await; - Some(ReviewDecision::Approved) - } - Some(ApprovalDecision::Allow) => runtime.review(GuardianReviewReason::FreshRequired).await, - Some(ApprovalDecision::AskUser) if !require_guardian => None, - None if !require_guardian - && !super::review::routes_approval_policy_to_guardian( - context.approval_policy, - context.approvals_reviewer, - ) => - { - None - } - None | Some(ApprovalDecision::AskUser) => { - runtime.review(GuardianReviewReason::Policy).await - } - }; + } + .decide(&session.services.extensions) + .await; // Enforce cancellation after extension callbacks too, including cached decisions. - if history_reset.is_cancelled() - || cancellation.is_cancelled() - || review_cancellation.is_cancelled() - { + if history_reset.is_cancelled() || cancellation.is_cancelled() { Some(ReviewDecision::Abort) } else { decision diff --git a/codex-rs/core/src/guardian/review.rs b/codex-rs/core/src/guardian/review.rs index 2f73c631ec..7235cce9e3 100644 --- a/codex-rs/core/src/guardian/review.rs +++ b/codex-rs/core/src/guardian/review.rs @@ -1,5 +1,5 @@ //! Supplies host review preparation and context-dependent configuration. -//! Guardian's extension owns execution, reporting and denial accounting. +//! Guardian's extension owns routing, execution, reporting and denial accounting. #[path = "review_request.rs"] mod request; @@ -14,8 +14,6 @@ use codex_guardian_reviewer::GuardianReviewOutcome; #[cfg(test)] use codex_guardian_reviewer::GuardianReviewSessionLimits; use codex_guardian_reviewer::ReviewModel; -use codex_protocol::config_types::ApprovalsReviewer; -use codex_protocol::protocol::AskForApproval; use codex_protocol::protocol::EventMsg; use codex_protocol::protocol::InternalSessionSource; use codex_protocol::protocol::ReviewDecision; @@ -98,16 +96,7 @@ pub(crate) fn new_guardian_review_id() -> String { uuid::Uuid::new_v4().to_string() } -/// Whether an exact approval policy and reviewer should route through Guardian. -pub(crate) fn routes_approval_policy_to_guardian( - approval_policy: AskForApproval, - approvals_reviewer: ApprovalsReviewer, -) -> bool { - matches!( - approval_policy, - AskForApproval::OnRequest | AskForApproval::Granular(_) - ) && approvals_reviewer == ApprovalsReviewer::AutoReview -} +pub(crate) use codex_guardian_reviewer::routes_approval_policy_to_guardian; pub(crate) fn is_basic_session_source(session_source: &SessionSource) -> bool { match session_source { @@ -117,19 +106,6 @@ pub(crate) fn is_basic_session_source(session_source: &SessionSource) -> bool { } } -pub(super) async fn record_guardian_non_denial(session: &Arc) { - let turn_id = { - let active = session.active_turn.lock().await; - let Some(task) = active.as_ref().and_then(|active| active.task.as_ref()) else { - return; - }; - task.turn_context.sub_id.clone() - }; - codex_guardian_reviewer::ReviewDenials::for_thread(&session.services.thread_extension_data) - .record_non_denial(&turn_id) - .await; -} - #[derive(Clone)] pub(crate) struct GuardianReviewOptions { /// Requires Guardian rather than a manual approval; cached evidence may still satisfy it. diff --git a/codex-rs/core/src/guardian/review_request.rs b/codex-rs/core/src/guardian/review_request.rs index 37aa7df4fb..3ac5138301 100644 --- a/codex-rs/core/src/guardian/review_request.rs +++ b/codex-rs/core/src/guardian/review_request.rs @@ -21,10 +21,6 @@ pub(in crate::guardian) struct PreparedApproval { impl ReviewHost for super::super::runtime::ReviewRuntime { type Prepared = PreparedApproval; - fn cancellation(&self) -> Option<&CancellationToken> { - self.options.external_cancel.as_ref() - } - async fn servicing_turn( &self, ) -> Option<(String, Arc)> { @@ -35,14 +31,15 @@ impl ReviewHost for super::super::runtime::ReviewRuntime { async fn prepare( &self, + review_id: &str, review_reason: GuardianReviewReason, deadline: Instant, + cancellation: &CancellationToken, ) -> Result<(PreparedApproval, codex_guardian_reviewer::ReviewReport), ReviewDecision> { let super::super::runtime::ReviewRuntime { session, history_reset: _, context, - review_id, request, reasons: _, options, @@ -56,7 +53,7 @@ impl ReviewHost for super::super::runtime::ReviewRuntime { let GuardianReviewOptions { plugin_attribution_override, approval_request_source, - external_cancel, + external_cancel: _, require_synchronous_review: _, require_guardian: _, } = options; @@ -65,9 +62,6 @@ impl ReviewHost for super::super::runtime::ReviewRuntime { let plugin_attribution = match plugin_attribution_override { Some(attribution) => Some(attribution), None if matches!(&request, GuardianApprovalRequest::ExecCommand { .. }) => { - let cancellation = external_cancel - .clone() - .unwrap_or_else(CancellationToken::new); let attribution_deadline = std::cmp::min( deadline, Instant::now() + GUARDIAN_PLUGIN_ATTRIBUTION_TIMEOUT, @@ -101,7 +95,7 @@ impl ReviewHost for super::super::runtime::ReviewRuntime { codex_guardian_reviewer::ReviewReport::new(codex_guardian_reviewer::ReviewMetadata { thread_id: session.thread_id.to_string(), turn_id: assessment_turn_id, - review_id, + review_id: review_id.to_owned(), target_item_id, plugin_id, script_path, @@ -153,6 +147,7 @@ impl ReviewHost for super::super::runtime::ReviewRuntime { &self, prepared: &PreparedApproval, deadline: Instant, + cancellation: &CancellationToken, ) -> (GuardianReviewOutcome, GuardianReviewAnalyticsResult) { let (mut outcome, analytics) = run_guardian_review_session_before_deadline( Arc::clone(&self.session), @@ -160,7 +155,7 @@ impl ReviewHost for super::super::runtime::ReviewRuntime { prepared.request.clone(), self.reasons.clone(), guardian_output_schema(), - self.options.external_cancel.clone(), + Some(cancellation.clone()), deadline, ) .await; @@ -182,11 +177,7 @@ impl ReviewHost for super::super::runtime::ReviewRuntime { .await .user_message_revision())) || self.history_reset.is_cancelled() - || self - .options - .external_cancel - .as_ref() - .is_some_and(CancellationToken::is_cancelled)) + || cancellation.is_cancelled()) { // A completed approval cannot outlive the owning-session or root evidence // it evaluated, including when either changed before prompt construction. @@ -196,6 +187,14 @@ impl ReviewHost for super::super::runtime::ReviewRuntime { (outcome, analytics) } + fn validate_action(&self) -> Result<(&str, Option<&str>), ReviewDecision> { + let request = self.request.validate(&self.context)?; + Ok(( + guardian_request_turn_id(request, &self.context.turn().sub_id), + guardian_request_target_item_id(request), + )) + } + async fn emit(&self, event: EventMsg) { self.session.send_event(self.context.turn(), event).await; } diff --git a/codex-rs/core/src/guardian/runtime.rs b/codex-rs/core/src/guardian/runtime.rs index 75d20b96d3..272126d79f 100644 --- a/codex-rs/core/src/guardian/runtime.rs +++ b/codex-rs/core/src/guardian/runtime.rs @@ -77,7 +77,6 @@ pub(super) struct ReviewRuntime { pub(super) session: Arc, pub(super) history_reset: CancellationToken, pub(super) context: GuardianReviewContext, - pub(super) review_id: String, pub(super) request: ReviewAction, pub(super) reasons: ApprovalRequestReasons, pub(super) options: GuardianReviewOptions, diff --git a/codex-rs/core/src/guardian/test_host.rs b/codex-rs/core/src/guardian/test_host.rs index 9529fb499f..c9bef66ea9 100644 --- a/codex-rs/core/src/guardian/test_host.rs +++ b/codex-rs/core/src/guardian/test_host.rs @@ -35,7 +35,10 @@ pub(crate) fn install(session: &Session, config: &Config) { /*attestation_provider*/ None, /*external_time_provider*/ None, )); - let runtime = Arc::new(codex_guardian_reviewer::ReviewerTasks::default()); + let runtime = session + .services + .thread_extension_data + .get_or_init(codex_guardian_reviewer::ReviewerTasks::default); session .services .thread_extension_data diff --git a/codex-rs/core/src/guardian/tests.rs b/codex-rs/core/src/guardian/tests.rs index 10a3157338..68888d9973 100644 --- a/codex-rs/core/src/guardian/tests.rs +++ b/codex-rs/core/src/guardian/tests.rs @@ -1466,11 +1466,45 @@ fn guardian_request_target_item_id_omits_network_access_trigger_call_id() { assert_eq!(guardian_request_target_item_id(&network_access), None); } +#[derive(Clone, Copy)] +enum ApprovalCancellation { + BeforeRouting, + BeforeCachedResult, +} + +#[test_case::test_case(ApprovalCancellation::BeforeRouting; "before_routing")] +#[test_case::test_case(ApprovalCancellation::BeforeCachedResult; "before_cached_result")] #[tokio::test] -async fn cancelled_guardian_review_emits_terminal_abort_without_warning() { - let (session, turn, rx) = crate::session::tests::make_session_and_context_with_rx().await; +async fn cancelled_guardian_review_emits_terminal_abort_without_warning( + moment: ApprovalCancellation, +) { + let (mut session, turn, rx) = crate::session::tests::make_session_and_context_with_rx().await; let cancel_token = CancellationToken::new(); - cancel_token.cancel(); + match moment { + ApprovalCancellation::BeforeRouting => cancel_token.cancel(), + ApprovalCancellation::BeforeCachedResult => { + struct CancelWhileApproving(CancellationToken); + impl codex_extension_api::ApprovalReviewContributor for CancelWhileApproving { + fn decide<'a>( + &'a self, + _input: &'a codex_extension_api::ApprovalDecisionInput<'_>, + ) -> codex_extension_api::ExtensionFuture< + 'a, + Option, + > { + self.0.cancel(); + Box::pin(async { Some(codex_extension_api::ApprovalDecision::Allow) }) + } + } + let mut extensions = codex_extension_api::ExtensionRegistryBuilder::::new(); + extensions + .approval_review_contributor(Arc::new(CancelWhileApproving(cancel_token.clone()))); + Arc::get_mut(&mut session) + .expect("unique test session") + .services + .extensions = Arc::new(extensions.build()); + } + } let decision = super::decide_approval( Arc::clone(&session), @@ -1511,10 +1545,13 @@ async fn cancelled_guardian_review_emits_terminal_abort_without_warning() { assert_eq!( guardian_statuses, - vec![ - GuardianAssessmentStatus::InProgress, - GuardianAssessmentStatus::Aborted, - ] + match moment { + ApprovalCancellation::BeforeRouting => vec![ + GuardianAssessmentStatus::InProgress, + GuardianAssessmentStatus::Aborted + ], + ApprovalCancellation::BeforeCachedResult => vec![], + } ); assert!(warnings.is_empty()); } diff --git a/codex-rs/core/src/session/tests/guardian_tests.rs b/codex-rs/core/src/session/tests/guardian_tests.rs index b1baa6609a..eef4b2cf47 100644 --- a/codex-rs/core/src/session/tests/guardian_tests.rs +++ b/codex-rs/core/src/session/tests/guardian_tests.rs @@ -389,8 +389,18 @@ async fn request_permissions_uses_issuing_step_policy_and_reviewer() { ); } +#[derive(Clone, Copy)] +enum ReviewCancellationSource { + Action, + ParentShutdown, +} + +#[test_case::test_case(ReviewCancellationSource::Action; "action")] +#[test_case::test_case(ReviewCancellationSource::ParentShutdown; "parent_shutdown")] #[tokio::test] -async fn request_permissions_guardian_review_stops_when_cancelled() { +async fn request_permissions_guardian_review_stops_when_cancelled( + source: ReviewCancellationSource, +) { let server = start_mock_server().await; let _guardian_request_log = mount_response_once( &server, @@ -479,13 +489,35 @@ async fn request_permissions_guardian_review_stops_when_cancelled() { .await .expect("guardian review should start before cancellation"); - cancellation_token.cancel(); + let reviewer_tasks = session + .services + .thread_extension_data + .get::() + .expect("reviewer tasks installed"); + match source { + ReviewCancellationSource::Action => cancellation_token.cancel(), + ReviewCancellationSource::ParentShutdown => reviewer_tasks.cancellation.cancel(), + } let response = timeout(Duration::from_secs(5), request_handle) .await .expect("request_permissions should stop when cancelled") .expect("request_permissions task should not panic"); - assert_eq!(response, None); + let expected_response = match source { + ReviewCancellationSource::Action => None, + ReviewCancellationSource::ParentShutdown => Some(RequestPermissionsResponse { + permissions: RequestPermissionProfile::default(), + scope: PermissionGrantScope::Turn, + strict_auto_review: false, + }), + }; + assert_eq!(response, expected_response); + if matches!(source, ReviewCancellationSource::ParentShutdown) { + reviewer_tasks.tasks.close(); + timeout(Duration::from_secs(5), reviewer_tasks.tasks.wait()) + .await + .expect("parent shutdown must finish reviewer cleanup"); + } assert_eq!( session .granted_turn_permissions(codex_exec_server::LOCAL_ENVIRONMENT_ID) diff --git a/codex-rs/core/tests/suite/guardian_review.rs b/codex-rs/core/tests/suite/guardian_review.rs index 166df45251..a88ff05af7 100644 --- a/codex-rs/core/tests/suite/guardian_review.rs +++ b/codex-rs/core/tests/suite/guardian_review.rs @@ -2289,8 +2289,32 @@ async fn guardian_timeout_rejects_tool_call_with_acting_model_instructions( Ok(()) } +#[derive(Clone, Copy)] +enum ApprovalPath { + SynchronousFallback, + CachedContributor, +} + +struct AttemptCachedApproval(Arc); + +impl codex_extension_api::ApprovalReviewContributor for AttemptCachedApproval { + fn decide<'a>( + &'a self, + _input: &'a codex_extension_api::ApprovalDecisionInput<'_>, + ) -> codex_extension_api::ExtensionFuture<'a, Option> + { + self.0 + .fetch_add(/*val*/ 1, std::sync::atomic::Ordering::SeqCst); + Box::pin(async { Some(codex_extension_api::ApprovalDecision::Allow) }) + } +} + +#[test_case(ApprovalPath::SynchronousFallback; "synchronous_fallback")] +#[test_case(ApprovalPath::CachedContributor; "cached_result_requires_fresh_review")] #[tokio::test(flavor = "multi_thread", worker_threads = 2)] -async fn cyber_model_guardian_denial_interrupts_turn_immediately() -> Result<()> { +async fn cyber_model_guardian_denial_interrupts_turn_immediately( + approval_path: ApprovalPath, +) -> Result<()> { skip_if_no_network!(Ok(())); skip_if_sandbox!(Ok(())); skip_if_wine_exec!( @@ -2318,6 +2342,14 @@ async fn cyber_model_guardian_denial_interrupts_turn_immediately() -> Result<()> .set_legacy_sandbox_policy(sandbox_policy_for_config) .expect("set sandbox policy"); }); + let cached_calls = Arc::new(std::sync::atomic::AtomicUsize::new(/*v*/ 0)); + if matches!(approval_path, ApprovalPath::CachedContributor) { + let mut extensions = ExtensionRegistryBuilder::default(); + extensions.approval_review_contributor(Arc::new(AttemptCachedApproval(Arc::clone( + &cached_calls, + )))); + builder = builder.with_extensions(Arc::new(extensions.build())); + } let test = builder.build_with_auto_env(&server).await?; let output_file = test.cwd.path().join("cyber-guardian-denied.txt"); @@ -2373,14 +2405,41 @@ async fn cyber_model_guardian_denial_interrupts_turn_immediately() -> Result<()> ) .await?; + let mut assessments = Vec::new(); let warning = wait_for_event(&test.codex, |event| { - matches!( - event, + match event { + EventMsg::GuardianAssessment(event) => assessments.push(event.clone()), EventMsg::GuardianWarning(warning) - if warning.message.contains("too many approval requests") - ) + if warning.message.contains("too many approval requests") => + { + return true; + } + EventMsg::TurnComplete(_) | EventMsg::TurnAborted(_) => { + panic!("turn ended without Guardian's denial warning") + } + _ => {} + } + false }) .await; + assert_eq!( + assessments + .iter() + .map(|event| event.status) + .collect::>(), + vec![ + codex_protocol::protocol::GuardianAssessmentStatus::InProgress, + codex_protocol::protocol::GuardianAssessmentStatus::Denied + ] + ); + assert_eq!(assessments[0].id, assessments[1].id); + assert_eq!( + cached_calls.load(std::sync::atomic::Ordering::SeqCst), + match approval_path { + ApprovalPath::SynchronousFallback => 0, + ApprovalPath::CachedContributor => 1, + } + ); let EventMsg::GuardianWarning(warning) = warning else { unreachable!("wait_for_event returned a non-warning event") }; diff --git a/codex-rs/ext/guardian-reviewer/src/lib.rs b/codex-rs/ext/guardian-reviewer/src/lib.rs index ad2aafbb02..57fe1cc228 100644 --- a/codex-rs/ext/guardian-reviewer/src/lib.rs +++ b/codex-rs/ext/guardian-reviewer/src/lib.rs @@ -18,6 +18,7 @@ mod pool; mod reporting; mod retry; mod review; +mod routing; mod settings; pub use assessment::GuardianAssessment; @@ -47,7 +48,6 @@ pub use pool::ReviewerTasks; pub use pool::SessionDisposition; pub use review::ReviewHost; -pub use review::SynchronousReview; pub use completion::ReviewCompletion; pub use completion::complete_review; @@ -66,3 +66,6 @@ pub use feedback::ReviewFeedbackSettings; pub use reporting::ReviewDenials; pub use reporting::ReviewMetadata; pub use reporting::ReviewReport; + +pub use routing::ReviewRequest; +pub use routing::routes_approval_policy_to_guardian; diff --git a/codex-rs/ext/guardian-reviewer/src/review.rs b/codex-rs/ext/guardian-reviewer/src/review.rs index 2e5bbf72f8..285f06e833 100644 --- a/codex-rs/ext/guardian-reviewer/src/review.rs +++ b/codex-rs/ext/guardian-reviewer/src/review.rs @@ -6,12 +6,10 @@ use crate::GuardianReviewOutcome; use crate::GuardianReviewSessionLimits; use crate::ReviewDenials; use crate::ReviewReport; -use codex_analytics::AnalyticsEventsClient; +use crate::ReviewRequest; use codex_analytics::GuardianReviewAnalyticsResult; -use codex_extension_api::ExtensionData; use codex_extension_api::ExtensionFuture; use codex_extension_api::SynchronousApprovalReviewer; -use codex_otel::SessionTelemetry; use codex_protocol::approvals::GuardianReviewReason; use codex_protocol::openai_models::ModelInfo; use codex_protocol::protocol::EventMsg; @@ -29,19 +27,23 @@ use tokio_util::sync::CancellationToken; /// they do not select Guardian outcomes, retry policy or reporting effects. pub trait ReviewHost: Send + Sync { type Prepared: Send + Sync; - fn cancellation(&self) -> Option<&CancellationToken>; /// Captures the turn currently servicing reviews, which may differ from a yielded cell's origin. fn servicing_turn(&self) -> impl Future)>> + Send; + /// Returns the owning turn and optional target item after validating the action. + fn validate_action(&self) -> Result<(&str, Option<&str>), ReviewDecision>; fn prepare( &self, + approval_id: &str, reason: GuardianReviewReason, deadline: Instant, + cancellation: &CancellationToken, ) -> impl Future> + Send; /// Rejects stale approvals before returning the attempt's outcome. fn attempt( &self, prepared: &Self::Prepared, deadline: Instant, + cancellation: &CancellationToken, ) -> impl Future + Send; fn emit(&self, event: EventMsg) -> impl Future + Send; fn record_evidence( @@ -52,32 +54,22 @@ pub trait ReviewHost: Send + Sync { fn interrupt(&self, turn_id: &str, warning: EventMsg) -> impl Future + Send; } -/// One review bound by the host before Guardian's approval policy chooses to run it. -pub struct SynchronousReview<'a, H> { - pub host: H, - pub thread_store: &'a ExtensionData, - pub model: &'a ModelInfo, - pub require_guardian: bool, - pub telemetry: &'a SessionTelemetry, - pub analytics: &'a AnalyticsEventsClient, -} - -impl SynchronousApprovalReviewer for SynchronousReview<'_, H> { +impl SynchronousApprovalReviewer for ReviewRequest<'_, H> { fn review(&self, reason: GuardianReviewReason) -> ExtensionFuture<'_, Option> { Box::pin(async move { let deadline = Instant::now() + crate::REVIEW_TIMEOUT; - let (context, report) = match self.host.prepare(reason, deadline).await { + let (context, report) = match self + .host + .prepare(self.approval_id, reason, deadline, &self.cancellation) + .await + { Ok(prepared) => prepared, Err(decision) => return Some(decision), }; self.host .emit(EventMsg::GuardianAssessment(report.started_event())) .await; - let (outcome, analytics) = if self - .host - .cancellation() - .is_some_and(CancellationToken::is_cancelled) - { + let (outcome, analytics) = if self.cancellation.is_cancelled() { ( GuardianReviewOutcome::Error(GuardianReviewError::Cancelled), GuardianReviewAnalyticsResult::without_session(), @@ -88,8 +80,8 @@ impl SynchronousApprovalReviewer for SynchronousReview<'_, H> { max_attempts: crate::MAX_REVIEW_ATTEMPTS, deadline, }, - self.host.cancellation(), - |deadline| self.host.attempt(&context, deadline), + Some(&self.cancellation), + |deadline| self.host.attempt(&context, deadline, &self.cancellation), )) .await }; diff --git a/codex-rs/ext/guardian-reviewer/src/routing.rs b/codex-rs/ext/guardian-reviewer/src/routing.rs new file mode 100644 index 0000000000..9dfbfd2b06 --- /dev/null +++ b/codex-rs/ext/guardian-reviewer/src/routing.rs @@ -0,0 +1,185 @@ +//! Resolves approval contributors and the synchronous fallback in one place. +//! Cached approvals still validate the bound action before recording their outcome. + +use crate::ReviewDenials; +use crate::ReviewHost; +use codex_analytics::AnalyticsEventsClient; +use codex_extension_api::ApprovalDecision; +use codex_extension_api::ApprovalDecisionInput; +use codex_extension_api::ExtensionData; +use codex_extension_api::ExtensionMetrics; +use codex_extension_api::ExtensionRegistry; +use codex_extension_api::GuardianV2Enabled; +use codex_extension_api::SynchronousApprovalReviewer; +use codex_otel::SessionTelemetry; +use codex_protocol::ThreadId; +use codex_protocol::approvals::GuardianReviewReason; +use codex_protocol::config_types::ApprovalsReviewer; +use codex_protocol::openai_models::GuardianScope; +use codex_protocol::openai_models::ModelInfo; +use codex_protocol::protocol::AskForApproval; +use codex_protocol::protocol::ReviewDecision; +use std::sync::Arc; +use tokio_util::sync::CancellationToken; + +/// One captured action and its host-enforced constraints. Guardian derives the +/// contributor input and owns cancellation for the complete review operation. +pub struct ReviewRequest<'a, H> { + pub host: H, + pub approval_id: &'a str, + pub tool_call_id: Option<&'a str>, + /// Absent when the host could not render the action for review. + pub action: Option<&'a serde_json::Value>, + pub thread_id: ThreadId, + pub thread_store: &'a ExtensionData, + pub category: GuardianScope, + pub approval_policy: AskForApproval, + pub approvals_reviewer: ApprovalsReviewer, + pub require_guardian: bool, + pub require_synchronous_review: bool, + pub model_requires_review: bool, + pub async_enabled: bool, + pub retried: bool, + pub escalated_exec: bool, + pub full_access: bool, + pub cancellation: CancellationToken, + pub model: &'a ModelInfo, + pub telemetry: &'a SessionTelemetry, + pub analytics: &'a AnalyticsEventsClient, + pub metrics: Option>, +} + +pub fn routes_approval_policy_to_guardian( + policy: AskForApproval, + reviewer: ApprovalsReviewer, +) -> bool { + matches!( + policy, + AskForApproval::OnRequest | AskForApproval::Granular(_) + ) && reviewer == ApprovalsReviewer::AutoReview +} + +impl ReviewRequest<'_, H> { + pub async fn decide( + mut self, + registry: &ExtensionRegistry, + ) -> Option { + let runtime = self.thread_store.get::(); + let _task = runtime.as_ref().map(|runtime| runtime.tasks.token()); + if runtime + .as_ref() + .is_some_and(|runtime| runtime.cancellation.is_cancelled()) + { + return Some(ReviewDecision::Abort); + } + let cancellation = self.cancellation.child_token(); + let _cancel_on_drop = cancellation.clone().drop_guard(); + self.cancellation = cancellation.clone(); + let decision = async { + if self.full_access { + return Some(match self.host.validate_action() { + Ok(_) => ReviewDecision::Approved, + Err(decision) => decision, + }); + } + let Some(action) = self.action else { + return Some(ReviewDecision::denied( + "automatic approval review could not prepare the action", + )); + }; + let input = ApprovalDecisionInput { + approval_id: self.approval_id, + tool_call_id: self.tool_call_id, + action, + thread_id: self.thread_id, + thread_store: self.thread_store, + category: self.category, + approval_policy: self.approval_policy, + approvals_reviewer: self.approvals_reviewer, + require_guardian: self.require_guardian, + require_fresh_review: self.require_synchronous_review + || self.model_requires_review && !self.async_enabled + || self.cancellation.is_cancelled() + || self.retried + || self.escalated_exec, + full_access: self.full_access, + metrics: self.metrics.clone(), + synchronous_reviewer: &self, + }; + match registry.decide_approval(&input).await { + Some(ApprovalDecision::Reviewed(decision)) => Some(decision), + Some(ApprovalDecision::Allow) if !input.require_fresh_review => { + Some(self.cached_approval().await) + } + Some(ApprovalDecision::Allow) => { + self.review(GuardianReviewReason::FreshRequired).await + } + Some(ApprovalDecision::AskUser) if !self.require_guardian => None, + None if !self.require_guardian + && !routes_approval_policy_to_guardian( + self.approval_policy, + self.approvals_reviewer, + ) => + { + None + } + None | Some(ApprovalDecision::AskUser) => { + self.review(GuardianReviewReason::Policy).await + } + } + }; + tokio::pin!(decision); + let result = if let Some(runtime) = runtime.as_ref() { + tokio::select! { + biased; + _ = runtime.cancellation.cancelled() => { + cancellation.cancel(); + // Drive cancellation through reporting and agent cleanup before releasing + // the tracked operation. Parent stop joins this work before closing history. + decision.await + } + result = &mut decision => result, + } + } else { + decision.await + }; + if cancellation.is_cancelled() + || runtime + .as_ref() + .is_some_and(|runtime| runtime.cancellation.is_cancelled()) + { + Some(ReviewDecision::Abort) + } else { + result + } + } + + async fn cached_approval(&self) -> ReviewDecision { + let (turn_id, item_id) = match self.host.validate_action() { + Ok(target) => target, + Err(decision) => return decision, + }; + if self.cancellation.is_cancelled() { + return ReviewDecision::Abort; + } + if self.thread_store.get::().is_some() { + self.analytics + .track_guardian_v2_event(codex_analytics::GuardianV2Event { + thread_id: self.thread_id.to_string(), + turn_id: turn_id.to_owned(), + item_id: item_id.map(str::to_owned), + model: Some(self.model.slug.clone()), + occurred_at_ms: codex_analytics::now_unix_millis(), + kind: codex_analytics::GuardianV2EventKind::FastDecision { + decision: "approved", + }, + }); + } + if let Some((turn_id, _)) = self.host.servicing_turn().await { + ReviewDenials::for_thread(self.thread_store) + .record_non_denial(&turn_id) + .await; + } + ReviewDecision::Approved + } +} 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 e61fda2bff..f67f8aac99 100644 --- a/codex-rs/ext/guardian-v2/src/async_scorer/approval.rs +++ b/codex-rs/ext/guardian-v2/src/async_scorer/approval.rs @@ -37,7 +37,7 @@ impl ApprovalReviewContributor for GuardianApprovalReviewer { input: &'a ApprovalDecisionInput<'_>, ) -> ExtensionFuture<'a, Option> { Box::pin(async move { - // If the extension is unavailable, core keeps its existing synchronous fallback. + // If the scorer is unavailable, the reviewer extension runs its synchronous fallback. let manager = self.thread_manager.upgrade()?; let Ok(thread) = manager.get_thread(input.thread_id).await else { record_fast_decision(input.metrics.as_deref(), "deferred", "scoring_failure");