diff --git a/codex-rs/ext/guardian-v2/src/async_scorer/approval.rs b/codex-rs/ext/guardian-v2/src/async_scorer/approval.rs new file mode 100644 index 0000000000..c5d9ed717c --- /dev/null +++ b/codex-rs/ext/guardian-v2/src/async_scorer/approval.rs @@ -0,0 +1,228 @@ +//! Owns approval routing and the choice between cached evidence and a fresh assessment. +//! Registration does not depend on the async scorer starting successfully. + +use super::authorization::ScoreAuthorization; +use super::config::GuardianV2Config; +use super::coverage::GuardianPolicy; +use super::extension::GuardianV2ScoreProgress; +use super::extension::requires_sync_for_compaction; +use super::metrics::TOOL_CALL_LAG_METRIC; +use super::metrics::record_fast_decision; +use super::sampler::LunaSampler; +use codex_core::CodexThread; +use codex_core::ThreadManager; +use codex_core::context::GuardianReviewEvidence; +use codex_extension_api::ApprovalDecision; +use codex_extension_api::ApprovalDecisionInput; +use codex_extension_api::ApprovalReviewContributor; +use codex_extension_api::ExtensionFuture; +use codex_protocol::approvals::GuardianReviewReason; +use codex_protocol::config_types::ApprovalsReviewer; +use codex_protocol::openai_models::GuardianReviewMode; +use codex_protocol::openai_models::GuardianScope; +use codex_protocol::protocol::AskForApproval; +use codex_protocol::security_risk::SecurityRiskScore; +use std::sync::Weak; +use std::sync::atomic::Ordering; + +pub(super) struct GuardianApprovalReviewer { + pub(super) thread_manager: Weak, +} + +impl ApprovalReviewContributor for GuardianApprovalReviewer { + fn decide<'a>( + &'a self, + input: &'a ApprovalDecisionInput<'_>, + ) -> ExtensionFuture<'a, Option> { + Box::pin(async move { + // If the extension is unavailable, core keeps its existing 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"); + return None; + }; + Some(self.decide_request(&thread, input).await) + }) + } +} + +impl GuardianApprovalReviewer { + #[tracing::instrument(skip_all, fields(approval_id = input.approval_id))] + async fn decide_request( + &self, + thread: &CodexThread, + input: &ApprovalDecisionInput<'_>, + ) -> ApprovalDecision { + if input.full_access { + return ApprovalDecision::Allow; + } + if !input.require_guardian + && (input.approvals_reviewer == ApprovalsReviewer::User + || !matches!( + input.approval_policy, + AskForApproval::OnRequest | AskForApproval::Granular(_) + )) + { + return ApprovalDecision::AskUser; + } + let config = thread.config().await; + let model = input + .thread_store + .get::(); + let guardian_config = input + .thread_store + .get::() + .map(|config| (*config).clone()) + .map_or_else(|| GuardianV2Config::resolve(&config), Ok) + .ok(); + let mut policy = guardian_config.as_ref().map_or_else( + || GuardianPolicy::from_legacy(/*scope*/ None).for_model(model.as_deref()), + |config| config.policy_for_model(model.as_deref()), + ); + if model.as_ref().is_some_and(|model| { + config + .config_layer_stack + .requirements() + .auto_review_required_for_model(&model.slug) + }) { + policy.enforce_required_model(); + } + let mode = policy.mode(input.category); + if mode != GuardianReviewMode::Adaptive { + record_fast_decision(input.metrics.as_deref(), "deferred", "out_of_scope"); + } + if mode == GuardianReviewMode::Disabled && !input.require_guardian { + return ApprovalDecision::AskUser; + } + 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, + }, + None => { + record_fast_decision(input.metrics.as_deref(), "deferred", "scoring_failure"); + GuardianReviewReason::ScoringFailure + } + } + } else if input.require_fresh_review { + GuardianReviewReason::FreshRequired + } else { + GuardianReviewReason::Policy + }; + tracing::debug!( + decision_source = "synchronous_assessment", + ?reason, + "reviewing approval" + ); + ApprovalDecision::Reviewed(input.synchronous_reviewer.review(reason).await) + } +} + +async fn cached_evidence( + thread: &CodexThread, + input: &ApprovalDecisionInput<'_>, + config: &GuardianV2Config, + policy: &GuardianPolicy, +) -> 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); + }; + if store + .get_or_init(GuardianReviewEvidence::default) + .uses_thread_owned_context() + { + let sampler = store + .get::() + .ok_or(GuardianReviewReason::MissingScore)?; + let history = thread.conversation_history_snapshot().await; + if requires_sync_for_compaction(config, history.as_ref(), &sampler) { + record_fast_decision(metrics, "deferred", "incompatible_compaction"); + return Err(GuardianReviewReason::IncompatibleCompaction); + } + } + let action = input.action; + if input.category == GuardianScope::ComputerUse + && policy.initial_cua_call + && action.get("tool_name").and_then(serde_json::Value::as_str) == Some("js") + && action + .get("connector_id") + .and_then(serde_json::Value::as_str) + == Some("node_repl") + && progress.js_executions.load(Ordering::Acquire) == 1 + { + record_fast_decision(metrics, "approved", "initial_cua_call"); + return Ok(()); + } + let current = ScoreAuthorization::current(thread).await; + if !current.local.retained_context_complete + || current + .root + .is_some_and(|root| !root.retained_context_complete) + { + record_fast_decision(metrics, "deferred", "incomplete_authorization"); + return Err(GuardianReviewReason::Policy); + } + let scored_authorization = progress + .authorization + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + let latest_scored = progress.latest_scored_tool_call.load(Ordering::Acquire); + let lag = progress + .latest_tool_call + .load(Ordering::Acquire) + .saturating_sub(latest_scored); + if let Some(metrics) = metrics { + metrics.histogram( + TOOL_CALL_LAG_METRIC, + i64::try_from(lag).unwrap_or(i64::MAX), + &[], + ); + } + // Reuse the latest thread score within the lag limit, even across categories + // and while the current action's async score is still in flight. + let score = store + .get::() + .and_then(|score| score.scores.get("action_risk").copied()); + let (reason, label) = match score { + _ if lag > config.max_tool_call_lag => { + if let Some(metrics) = metrics { + metrics.counter( + super::metrics::REVIEW_FALLBACK_METRIC, + /*inc*/ 1, + &[("fallback_reason", "score_lag")], + ); + } + (GuardianReviewReason::StaleScore, "stale_score") + } + _ if progress.latest_failed_tool_call.load(Ordering::Acquire) > latest_scored => { + (GuardianReviewReason::ScoringFailure, "scoring_failure") + } + None => (GuardianReviewReason::MissingScore, "missing_score"), + Some(score) if score < config.review_threshold => { + if scored_authorization.as_ref() != Some(¤t) { + ( + GuardianReviewReason::AuthorizationChanged, + "authorization_changed", + ) + } else { + record_fast_decision(metrics, "approved", "low_risk"); + return Ok(()); + } + } + Some(score) if score >= config.review_threshold => { + (GuardianReviewReason::ElevatedRisk, "elevated_risk") + } + Some(_) => (GuardianReviewReason::InvalidScore, "invalid_score"), + }; + tracing::debug!( + approval_id = input.approval_id, + fallback_reason = label, + "requesting synchronous review" + ); + record_fast_decision(metrics, "deferred", label); + Err(reason) +} diff --git a/codex-rs/ext/guardian-v2/src/async_scorer/extension.rs b/codex-rs/ext/guardian-v2/src/async_scorer/extension.rs index c6a41bec6d..f9fb09d0ef 100644 --- a/codex-rs/ext/guardian-v2/src/async_scorer/extension.rs +++ b/codex-rs/ext/guardian-v2/src/async_scorer/extension.rs @@ -85,14 +85,14 @@ enum ClassificationOutcome { } #[derive(Default)] -struct GuardianV2ScoreProgress { - latest_tool_call: AtomicUsize, +pub(super) struct GuardianV2ScoreProgress { + pub(super) latest_tool_call: AtomicUsize, // Setup and reset calls must not consume the first JS execution allowance. - js_executions: AtomicUsize, - latest_scored_tool_call: AtomicUsize, - latest_failed_tool_call: AtomicUsize, + pub(super) js_executions: AtomicUsize, + pub(super) latest_scored_tool_call: AtomicUsize, + pub(super) latest_failed_tool_call: AtomicUsize, // Serialize successful score publication with its authorization metadata. - authorization: Mutex>, + pub(super) authorization: Mutex>, metrics: Option>, } @@ -959,7 +959,7 @@ enum ParentCompactionError { } // Sampling and fast approval must apply the same checkpoint eligibility policy. -fn requires_sync_for_compaction( +pub(super) fn requires_sync_for_compaction( config: &GuardianV2Config, history: &dyn ConversationHistorySnapshot, sampler: &LunaSampler, @@ -1026,9 +1026,12 @@ pub fn install( let extension = Arc::new(GuardianV2Extension { auth_manager, event_sink: registry.event_sink(), - thread_manager, + thread_manager: thread_manager.clone(), }); registry.thread_lifecycle_contributor(extension.clone()); + registry.approval_review_contributor(Arc::new(super::approval::GuardianApprovalReviewer { + thread_manager, + })); registry.approval_review_contributor(extension.clone()); registry.skill_invocation_contributor(extension.clone()); registry.tool_lifecycle_contributor(extension); diff --git a/codex-rs/ext/guardian-v2/src/async_scorer/mod.rs b/codex-rs/ext/guardian-v2/src/async_scorer/mod.rs index 4b57eeda00..7b6b82039e 100644 --- a/codex-rs/ext/guardian-v2/src/async_scorer/mod.rs +++ b/codex-rs/ext/guardian-v2/src/async_scorer/mod.rs @@ -1,4 +1,5 @@ mod action; +mod approval; mod authorization; mod config; mod coverage;