mirror of
https://github.com/openai/codex.git
synced 2026-09-06 15:29:32 +00:00
Route Guardian approvals independently of async scoring (#42819)
## Why Approval routing must remain available even when the async scorer does not start successfully. ## What changed - Register a dedicated Guardian approval reviewer before the existing async scorer contributor. - Reuse eligible cached low-risk evidence, including the initial computer-use allowance, without waiting for a fresh score. - Route requests to the user when automatic review does not apply, and fall back to a synchronous assessment with the applicable review reason when cached evidence cannot be used. GitOrigin-RevId: 956427040456bed4c497da952f0bed7f22e93fa6
This commit is contained in:
228
codex-rs/ext/guardian-v2/src/async_scorer/approval.rs
Normal file
228
codex-rs/ext/guardian-v2/src/async_scorer/approval.rs
Normal file
@@ -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<ThreadManager>,
|
||||
}
|
||||
|
||||
impl ApprovalReviewContributor for GuardianApprovalReviewer {
|
||||
fn decide<'a>(
|
||||
&'a self,
|
||||
input: &'a ApprovalDecisionInput<'_>,
|
||||
) -> ExtensionFuture<'a, Option<ApprovalDecision>> {
|
||||
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::<codex_protocol::openai_models::ModelInfo>();
|
||||
let guardian_config = input
|
||||
.thread_store
|
||||
.get::<GuardianV2Config>()
|
||||
.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::<GuardianV2ScoreProgress>() 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::<LunaSampler>()
|
||||
.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::<SecurityRiskScore>()
|
||||
.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)
|
||||
}
|
||||
@@ -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<Option<ScoreAuthorization>>,
|
||||
pub(super) authorization: Mutex<Option<ScoreAuthorization>>,
|
||||
metrics: Option<Arc<dyn ExtensionMetrics>>,
|
||||
}
|
||||
|
||||
@@ -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);
|
||||
|
||||
@@ -1,4 +1,5 @@
|
||||
mod action;
|
||||
mod approval;
|
||||
mod authorization;
|
||||
mod config;
|
||||
mod coverage;
|
||||
|
||||
Reference in New Issue
Block a user