From e8e7103cb96e41b1b14f0aff3ca18ea50555fdcb Mon Sep 17 00:00:00 2001 From: jif Date: Wed, 9 Sep 2026 17:16:54 +0000 Subject: [PATCH] Extract Guardian review policy into a dedicated crate (#44227) ## What changed Move assessment parsing and schema, model selection, review outcomes, retry handling, and rejection circuit breakers into `codex-guardian-reviewer`. Update core to use the extracted APIs while retaining session execution and decision enforcement in the host. ## Testing Move existing assessment, circuit breaker, and retry tests into the new crate, including coverage for transient errors, cancellation, and deadlines. GitOrigin-RevId: 96ec9989a0066acb012bf3bec0b9f7d8bf11a4ef --- codex-rs/Cargo.lock | 16 + codex-rs/Cargo.toml | 2 + codex-rs/core/Cargo.toml | 1 + codex-rs/core/src/guardian/mod.rs | 101 +---- codex-rs/core/src/guardian/prompt.rs | 4 +- codex-rs/core/src/guardian/review.rs | 384 ++---------------- codex-rs/core/src/guardian/review_session.rs | 13 +- codex-rs/core/src/guardian/tests.rs | 105 ----- codex-rs/ext/guardian-reviewer/BUILD.bazel | 6 + codex-rs/ext/guardian-reviewer/Cargo.toml | 26 ++ .../guardian-reviewer/src}/assessment.rs | 2 +- .../src}/assessment_tests.rs | 0 .../guardian-reviewer/src/circuit_breaker.rs | 90 ++++ .../src/circuit_breaker_tests.rs | 107 +++++ codex-rs/ext/guardian-reviewer/src/lib.rs | 27 ++ codex-rs/ext/guardian-reviewer/src/model.rs | 76 ++++ codex-rs/ext/guardian-reviewer/src/outcome.rs | 111 +++++ codex-rs/ext/guardian-reviewer/src/retry.rs | 94 +++++ .../ext/guardian-reviewer/src/retry_tests.rs | 135 ++++++ 19 files changed, 730 insertions(+), 570 deletions(-) create mode 100644 codex-rs/ext/guardian-reviewer/BUILD.bazel create mode 100644 codex-rs/ext/guardian-reviewer/Cargo.toml rename codex-rs/{core/src/guardian => ext/guardian-reviewer/src}/assessment.rs (98%) rename codex-rs/{core/src/guardian => ext/guardian-reviewer/src}/assessment_tests.rs (100%) create mode 100644 codex-rs/ext/guardian-reviewer/src/circuit_breaker.rs create mode 100644 codex-rs/ext/guardian-reviewer/src/circuit_breaker_tests.rs create mode 100644 codex-rs/ext/guardian-reviewer/src/lib.rs create mode 100644 codex-rs/ext/guardian-reviewer/src/model.rs create mode 100644 codex-rs/ext/guardian-reviewer/src/outcome.rs create mode 100644 codex-rs/ext/guardian-reviewer/src/retry.rs create mode 100644 codex-rs/ext/guardian-reviewer/src/retry_tests.rs diff --git a/codex-rs/Cargo.lock b/codex-rs/Cargo.lock index 94deb8b6d9..ae499d41a7 100644 --- a/codex-rs/Cargo.lock +++ b/codex-rs/Cargo.lock @@ -2988,6 +2988,7 @@ dependencies = [ "codex-file-system", "codex-git-utils", "codex-guardian-context", + "codex-guardian-reviewer", "codex-history", "codex-home", "codex-hooks", @@ -3550,6 +3551,21 @@ dependencies = [ "serde_json", ] +[[package]] +name = "codex-guardian-reviewer" +version = "0.0.0" +dependencies = [ + "anyhow", + "codex-analytics", + "codex-protocol", + "pretty_assertions", + "rand 0.9.3", + "serde", + "serde_json", + "tokio", + "tokio-util", +] + [[package]] name = "codex-guardian-v2" version = "0.0.0" diff --git a/codex-rs/Cargo.toml b/codex-rs/Cargo.toml index 77cc3c3fc5..e07725207c 100644 --- a/codex-rs/Cargo.toml +++ b/codex-rs/Cargo.toml @@ -62,6 +62,7 @@ members = [ "ext/extension-api", "ext/goal", "ext/git-attribution", + "ext/guardian-reviewer", "ext/guardian-v2", "ext/history-notes", "ext/image-generation", @@ -213,6 +214,7 @@ codex-extension-api = { path = "ext/extension-api" } codex-extension-items = { path = "ext/items" } codex-goal-extension = { path = "ext/goal" } codex-git-attribution = { path = "ext/git-attribution" } +codex-guardian-reviewer = { path = "ext/guardian-reviewer" } codex-guardian-v2 = { path = "ext/guardian-v2" } codex-history-notes-extension = { path = "ext/history-notes" } codex-image-generation-extension = { path = "ext/image-generation" } diff --git a/codex-rs/core/Cargo.toml b/codex-rs/core/Cargo.toml index e7947e1715..24c6640cc1 100644 --- a/codex-rs/core/Cargo.toml +++ b/codex-rs/core/Cargo.toml @@ -48,6 +48,7 @@ codex-shell-command = { workspace = true } codex-execpolicy = { workspace = true } codex-git-utils = { workspace = true } codex-guardian-context = { workspace = true } +codex-guardian-reviewer = { workspace = true } codex-history = { workspace = true } codex-hooks = { workspace = true } codex-http-client = { workspace = true } diff --git a/codex-rs/core/src/guardian/mod.rs b/codex-rs/core/src/guardian/mod.rs index f04041bf0b..99d0713ae3 100644 --- a/codex-rs/core/src/guardian/mod.rs +++ b/codex-rs/core/src/guardian/mod.rs @@ -3,7 +3,6 @@ //! review requirements. Each approval retains its issuing context and cancellation. mod approval_request; -mod assessment; mod coverage; mod decision; mod feedback; @@ -17,7 +16,6 @@ mod reviewer_config; mod runtime; use std::sync::Arc; -use std::time::Duration; use codex_protocol::config_types::ApprovalsReviewer; use codex_protocol::config_types::ReasoningSummary; @@ -54,13 +52,9 @@ pub use review_session::GuardianReviewSessionManager; pub(crate) use review_session::prompt_cache_key_override_for_review_session; pub(crate) use runtime::ReviewAction; -pub(crate) const GUARDIAN_REVIEW_TIMEOUT: Duration = Duration::from_secs(90); +pub(crate) use codex_guardian_reviewer::REVIEW_TIMEOUT as GUARDIAN_REVIEW_TIMEOUT; pub(crate) const GUARDIAN_REVIEWER_NAME: &str = "guardian"; -pub(crate) const MAX_CONSECUTIVE_CYBER_GUARDIAN_DENIALS_PER_TURN: u32 = 1; -pub(crate) const MAX_CONSECUTIVE_GUARDIAN_DENIALS_PER_TURN: u32 = 3; -pub(crate) const MAX_RECENT_CYBER_AUTO_REVIEW_DENIALS_PER_TURN: u32 = 1; -pub(crate) const MAX_RECENT_AUTO_REVIEW_DENIALS_PER_TURN: u32 = 10; -pub(crate) const AUTO_REVIEW_DENIAL_WINDOW_SIZE: usize = 50; +pub(crate) use codex_guardian_reviewer::AUTO_REVIEW_DENIAL_WINDOW_SIZE; pub(crate) const AUTO_REVIEW_DENIED_ACTION_APPROVAL_DEVELOPER_PREFIX: &str = codex_guardian_context::MANUAL_APPROVAL_DEVELOPER_PREFIX; const GUARDIAN_MAX_TOOL_ENTRY_TOKENS: usize = codex_guardian_context::ContextProfile::synchronous() @@ -166,89 +160,14 @@ impl From<&Arc> for GuardianReviewContext { } } -pub use assessment::GuardianAssessment; -pub use assessment::guardian_output_schema; -pub use assessment::parse_guardian_assessment; +pub use codex_guardian_reviewer::GuardianAssessment; +pub use codex_guardian_reviewer::guardian_output_schema; +pub use codex_guardian_reviewer::parse_guardian_assessment; pub use reviewer_config::build_guardian_review_session_config; -#[derive(Debug, Default)] -pub(crate) struct GuardianRejectionCircuitBreaker { - turns: std::collections::HashMap, -} - -#[derive(Debug, Default)] -struct GuardianRejectionCircuitBreakerTurn { - consecutive_denials: u32, - recent_denials: std::collections::VecDeque, - interrupt_triggered: bool, -} - -#[derive(Clone, Copy, Debug, PartialEq, Eq)] -pub(crate) enum GuardianRejectionCircuitBreakerPolicy { - Standard, - CyberModel, -} - -#[derive(Clone, Copy, Debug, PartialEq, Eq)] -pub(crate) enum GuardianRejectionCircuitBreakerAction { - Continue, - InterruptTurn { - consecutive_denials: u32, - recent_denials: u32, - }, -} - -impl GuardianRejectionCircuitBreaker { - pub(crate) fn clear_turn(&mut self, turn_id: &str) { - self.turns.remove(turn_id); - } - - pub(crate) fn record_denial( - &mut self, - turn_id: &str, - policy: GuardianRejectionCircuitBreakerPolicy, - ) -> GuardianRejectionCircuitBreakerAction { - let turn = self.turns.entry(turn_id.to_string()).or_default(); - turn.consecutive_denials = turn.consecutive_denials.saturating_add(1); - Self::record_recent_review(turn, /*denied*/ true); - let recent_denials = turn.recent_denials.iter().filter(|denied| **denied).count() as u32; - let (max_consecutive_denials, max_recent_denials) = match policy { - GuardianRejectionCircuitBreakerPolicy::Standard => ( - MAX_CONSECUTIVE_GUARDIAN_DENIALS_PER_TURN, - MAX_RECENT_AUTO_REVIEW_DENIALS_PER_TURN, - ), - GuardianRejectionCircuitBreakerPolicy::CyberModel => ( - MAX_CONSECUTIVE_CYBER_GUARDIAN_DENIALS_PER_TURN, - MAX_RECENT_CYBER_AUTO_REVIEW_DENIALS_PER_TURN, - ), - }; - if !turn.interrupt_triggered - && (turn.consecutive_denials >= max_consecutive_denials - || recent_denials >= max_recent_denials) - { - turn.interrupt_triggered = true; - GuardianRejectionCircuitBreakerAction::InterruptTurn { - consecutive_denials: turn.consecutive_denials, - recent_denials, - } - } else { - GuardianRejectionCircuitBreakerAction::Continue - } - } - - pub(crate) fn record_non_denial(&mut self, turn_id: &str) { - let turn = self.turns.entry(turn_id.to_string()).or_default(); - turn.consecutive_denials = 0; - Self::record_recent_review(turn, /*denied*/ false); - } - - fn record_recent_review(turn: &mut GuardianRejectionCircuitBreakerTurn, denied: bool) { - turn.recent_denials.push_back(denied); - if turn.recent_denials.len() > AUTO_REVIEW_DENIAL_WINDOW_SIZE { - turn.recent_denials.pop_front(); - } - } -} +pub(crate) use codex_guardian_reviewer::GuardianRejectionCircuitBreaker; +pub(crate) use codex_guardian_reviewer::GuardianRejectionCircuitBreakerAction; +pub(crate) use codex_guardian_reviewer::GuardianRejectionCircuitBreakerPolicy; pub(crate) use approval_request::format_guardian_action_pretty; #[cfg(test)] @@ -256,6 +175,8 @@ use approval_request::guardian_assessment_action; #[cfg(test)] use approval_request::guardian_request_turn_id; #[cfg(test)] +use codex_guardian_reviewer::GuardianReviewOutcome; +#[cfg(test)] use prompt::GuardianPromptMode; #[cfg(test)] use prompt::GuardianTranscriptCursor; @@ -266,8 +187,6 @@ use prompt::build_guardian_prompt_items_with_parent_turn; #[cfg(test)] use prompt::render_guardian_transcript_entries; #[cfg(test)] -use review::GuardianReviewOutcome; -#[cfg(test)] use review::run_guardian_review_session_with_retry as run_guardian_review_session_for_test; #[cfg(test)] use review_session::build_guardian_review_session_config as build_guardian_review_session_config_for_test; diff --git a/codex-rs/core/src/guardian/prompt.rs b/codex-rs/core/src/guardian/prompt.rs index 52ab173d3e..f22a3bcafd 100644 --- a/codex-rs/core/src/guardian/prompt.rs +++ b/codex-rs/core/src/guardian/prompt.rs @@ -362,8 +362,8 @@ pub(crate) fn guardian_truncate_text(content: &str, token_cap: usize) -> (String ) } -use super::assessment::guardian_output_contract_prompt; -pub use super::assessment::parse_guardian_assessment; +use codex_guardian_reviewer::guardian_output_contract_prompt; +pub use codex_guardian_reviewer::parse_guardian_assessment; pub(crate) const BUNDLED_GUARDIAN_POLICY: &str = include_str!("../../assets/guardian/policy.md"); pub(crate) const BUNDLED_GUARDIAN_POLICY_TEMPLATE: &str = diff --git a/codex-rs/core/src/guardian/review.rs b/codex-rs/core/src/guardian/review.rs index b3cb30a6ad..1520d30c42 100644 --- a/codex-rs/core/src/guardian/review.rs +++ b/codex-rs/core/src/guardian/review.rs @@ -9,11 +9,13 @@ use codex_analytics::GuardianReviewedAction; use codex_core_plugins::PluginCommandAttribution; use codex_extension_api::ThreadIdleCause; use codex_features::Feature; +use codex_guardian_reviewer::GuardianReviewError; +use codex_guardian_reviewer::GuardianReviewOutcome; +use codex_guardian_reviewer::GuardianReviewSessionLimits; use codex_protocol::config_types::ApprovalsReviewer; use codex_protocol::openai_models::MODEL_SPECIALTY_CYBER; use codex_protocol::openai_models::ModelInfo; use codex_protocol::protocol::AskForApproval; -use codex_protocol::protocol::CodexErrorInfo; use codex_protocol::protocol::EventMsg; use codex_protocol::protocol::GuardianAssessmentDecisionSource; use codex_protocol::protocol::GuardianAssessmentEvent; @@ -29,7 +31,6 @@ use codex_protocol::protocol::WarningEvent; use std::sync::Arc; use std::time::Duration; use tokio::time::Instant; -use tokio::time::sleep_until; use tokio_util::sync::CancellationToken; use crate::context::GuardianNodeReplPolicy; @@ -37,7 +38,6 @@ use crate::context::GuardianReviewEvidence; use crate::session::session::Session; use crate::session::turn_context::TurnContext; use crate::turn_timing::now_unix_timestamp_ms; -use crate::util::backoff; use super::AUTO_REVIEW_DENIAL_WINDOW_SIZE; use super::ApprovalRequestReasons; @@ -54,12 +54,10 @@ use super::approval_request::guardian_assessment_action; use super::approval_request::guardian_request_target_item_id; use super::approval_request::guardian_request_turn_id; use super::approval_request::guardian_reviewed_action; -use super::assessment::guardian_output_schema; use super::metrics::emit_guardian_review_metrics; -use super::prompt::parse_guardian_assessment; -use super::review_session::GuardianReviewSessionOutcome; use super::review_session::GuardianReviewSessionParams; use super::review_session::build_guardian_review_session_config; +use codex_guardian_reviewer::guardian_output_schema; const GUARDIAN_REJECTION_INSTRUCTIONS: &str = concat!( "The agent must not attempt to achieve the same outcome via workaround, ", @@ -75,7 +73,6 @@ const GUARDIAN_TIMEOUT_INSTRUCTIONS: &str = concat!( "You may retry once, or ask the user for guidance or explicit approval.", ); -const GUARDIAN_REVIEW_MAX_ATTEMPTS: i64 = 3; const GUARDIAN_PLUGIN_ATTRIBUTION_TIMEOUT: Duration = Duration::from_secs(5); async fn plugin_attribution_for_guardian_request( @@ -138,66 +135,6 @@ pub(crate) fn guardian_timeout_message(model_info: &ModelInfo) -> String { .to_string() } -#[derive(Debug)] -pub(super) enum GuardianReviewOutcome { - Completed(GuardianAssessment), - Error(GuardianReviewError), -} - -#[derive(Debug)] -pub(super) enum GuardianReviewError { - PromptBuild { - message: String, - }, - Session { - message: String, - error_info: Option, - }, - Parse { - message: String, - }, - Timeout, - Cancelled, -} - -impl GuardianReviewError { - fn prompt_build(err: anyhow::Error) -> Self { - Self::PromptBuild { - message: err.to_string(), - } - } - - fn session(err: anyhow::Error) -> Self { - Self::Session { - message: err.to_string(), - error_info: None, - } - } - - fn session_with_error_info(err: anyhow::Error, error_info: CodexErrorInfo) -> Self { - Self::Session { - message: err.to_string(), - error_info: Some(error_info), - } - } - - fn parse(err: anyhow::Error) -> Self { - Self::Parse { - message: err.to_string(), - } - } - - fn failure_reason(&self) -> GuardianReviewFailureReason { - match self { - Self::PromptBuild { .. } => GuardianReviewFailureReason::PromptBuildError, - Self::Session { .. } => GuardianReviewFailureReason::SessionError, - Self::Parse { .. } => GuardianReviewFailureReason::ParseError, - Self::Timeout => GuardianReviewFailureReason::Timeout, - Self::Cancelled => GuardianReviewFailureReason::Cancelled, - } - } -} - fn guardian_risk_level_str(level: GuardianRiskLevel) -> &'static str { match level { GuardianRiskLevel::Low => "low", @@ -511,7 +448,7 @@ pub(super) async fn run_synchronous_review( schema, external_cancel, GuardianReviewSessionLimits { - max_attempts: GUARDIAN_REVIEW_MAX_ATTEMPTS, + max_attempts: codex_guardian_reviewer::MAX_REVIEW_ATTEMPTS, deadline, }, )) @@ -824,49 +761,19 @@ pub(super) async fn guardian_review_session_config( ) .await; let default_review_model_id = turn.provider.approval_review_preferred_model(); - let preferred_reasoning_effort = |supports_low: bool, fallback| { - if supports_low { - Some(codex_protocol::openai_models::ReasoningEffort::Low) - } else { - fallback - } - }; - let model_override = turn.model_info().auto_review_model_override.as_deref(); - let review_model_id = model_override.unwrap_or(default_review_model_id); - let review_model = available_models - .iter() - .find(|preset| preset.model == review_model_id); - let guardian_catalog_contains_auto_review = available_models - .iter() - .any(|preset| preset.model == default_review_model_id); - let guardian_review_model_overridden = model_override.is_some(); - let guardian_review_model_override = model_override.map(str::to_string); - let (guardian_model, guardian_reasoning_effort) = if let Some(preset) = review_model { - let reasoning_effort = preferred_reasoning_effort( - preset - .supported_reasoning_efforts - .iter() - .any(|effort| effort.effort == codex_protocol::openai_models::ReasoningEffort::Low), - Some(preset.default_reasoning_effort.clone()), - ); - (review_model_id.to_string(), reasoning_effort) - } else { - let reasoning_effort = preferred_reasoning_effort( - turn.model_info() - .supported_reasoning_levels - .iter() - .any(|preset| preset.effort == codex_protocol::openai_models::ReasoningEffort::Low), - turn.reasoning_effort() - .or(turn.model_info().default_reasoning_level.as_ref()) - .cloned(), - ); - ( - model_override - .unwrap_or(turn.model_info().slug.as_str()) - .to_string(), - reasoning_effort, - ) - }; + let codex_guardian_reviewer::ReviewModel { + model: guardian_model, + reasoning_effort: guardian_reasoning_effort, + default_review_model_id, + catalog_contains_auto_review: guardian_catalog_contains_auto_review, + model_overridden: guardian_review_model_overridden, + model_override: guardian_review_model_override, + } = codex_guardian_reviewer::select_review_model( + turn.model_info(), + turn.reasoning_effort(), + default_review_model_id, + &available_models, + ); let guardian_model_info = session .services @@ -905,7 +812,7 @@ pub(super) async fn guardian_review_session_config( ), model: guardian_model, reasoning_effort: guardian_reasoning_effort, - default_review_model_id: default_review_model_id.to_string(), + default_review_model_id, catalog_contains_auto_review: guardian_catalog_contains_auto_review, model_overridden: guardian_review_model_overridden, model_override: guardian_review_model_override, @@ -973,55 +880,7 @@ async fn run_guardian_review_session_before_deadline( ) .await; - match session_outcome { - GuardianReviewSessionOutcome::Completed(Ok(last_agent_message)) => match last_agent_message - { - Some(last_agent_message) => { - match parse_guardian_assessment(Some(&last_agent_message)) { - Ok(assessment) => ( - GuardianReviewOutcome::Completed(assessment), - session_analytics_result, - ), - Err(err) => ( - GuardianReviewOutcome::Error(GuardianReviewError::parse(err)), - session_analytics_result, - ), - } - } - None => ( - GuardianReviewOutcome::Error(GuardianReviewError::session(anyhow::anyhow!( - "guardian review completed without an assessment payload" - ))), - session_analytics_result, - ), - }, - GuardianReviewSessionOutcome::Completed(Err(err)) => ( - GuardianReviewOutcome::Error(GuardianReviewError::session(err)), - session_analytics_result, - ), - GuardianReviewSessionOutcome::PromptBuildFailed(err) => ( - GuardianReviewOutcome::Error(GuardianReviewError::prompt_build(err)), - session_analytics_result, - ), - GuardianReviewSessionOutcome::SessionFailed { error, error_info } => { - let error = match error_info { - Some(error_info) => GuardianReviewError::session_with_error_info(error, error_info), - None => GuardianReviewError::session(error), - }; - ( - GuardianReviewOutcome::Error(error), - session_analytics_result, - ) - } - GuardianReviewSessionOutcome::TimedOut => ( - GuardianReviewOutcome::Error(GuardianReviewError::Timeout), - session_analytics_result, - ), - GuardianReviewSessionOutcome::Aborted => ( - GuardianReviewOutcome::Error(GuardianReviewError::Cancelled), - session_analytics_result, - ), - } + (session_outcome.into(), session_analytics_result) } #[cfg(test)] @@ -1049,11 +908,6 @@ pub(super) async fn run_guardian_review_session_with_retry( .await } -struct GuardianReviewSessionLimits { - max_attempts: i64, - deadline: Instant, -} - async fn run_guardian_review_session_with_retry_before_deadline( session: Arc, context: impl Into, @@ -1063,15 +917,9 @@ async fn run_guardian_review_session_with_retry_before_deadline( external_cancel: Option, limits: GuardianReviewSessionLimits, ) -> (GuardianReviewOutcome, GuardianReviewAnalyticsResult) { - let GuardianReviewSessionLimits { - max_attempts, - deadline, - } = limits; let context = context.into(); - assert!(max_attempts > 0, "guardian review must run at least once"); - let mut attempt_count = 1; - loop { - let (outcome, mut analytics_result) = run_guardian_review_session_before_deadline( + codex_guardian_reviewer::run_with_retry(limits, external_cancel.as_ref(), |deadline| { + run_guardian_review_session_before_deadline( Arc::clone(&session), context.clone(), request.clone(), @@ -1080,190 +928,6 @@ async fn run_guardian_review_session_with_retry_before_deadline( external_cancel.clone(), deadline, ) - .await; - analytics_result.attempt_count = attempt_count; - if attempt_count >= max_attempts || !should_retry_guardian_review(&outcome) { - return (outcome, analytics_result); - } - if let Some(error) = - wait_before_guardian_retry(attempt_count, deadline, external_cancel.as_ref()).await - { - return (GuardianReviewOutcome::Error(error), analytics_result); - } - attempt_count += 1; - } -} - -async fn wait_before_guardian_retry( - attempt_count: i64, - deadline: Instant, - external_cancel: Option<&CancellationToken>, -) -> Option { - let retry_delay = backoff(attempt_count as u64); - let retry_at = (Instant::now() + retry_delay).min(deadline); - tokio::select! { - _ = sleep_until(retry_at) => { - (Instant::now() >= deadline).then_some(GuardianReviewError::Timeout) - } - _ = async { - if let Some(cancel_token) = external_cancel { - cancel_token.cancelled().await; - } else { - std::future::pending::<()>().await; - } - } => Some(GuardianReviewError::Cancelled), - } -} - -fn should_retry_guardian_review(outcome: &GuardianReviewOutcome) -> bool { - matches!( - outcome, - GuardianReviewOutcome::Error( - GuardianReviewError::Session { - error_info: Some( - CodexErrorInfo::ServerOverloaded - | CodexErrorInfo::HttpConnectionFailed { .. } - | CodexErrorInfo::ResponseStreamConnectionFailed { .. } - | CodexErrorInfo::InternalServerError - | CodexErrorInfo::ResponseStreamDisconnected { .. } - ), - .. - } | GuardianReviewError::Parse { .. } - ) - ) -} - -#[cfg(test)] -mod review_tests { - use super::*; - use std::time::Duration; - - #[test] - fn guardian_review_error_reason_distinguishes_error_kinds() { - let parse_error = GuardianReviewError::parse(anyhow::anyhow!("bad guardian JSON")); - let prompt_error = GuardianReviewError::prompt_build(anyhow::anyhow!("bad prompt/config")); - let session_error = - GuardianReviewError::session(anyhow::anyhow!("guardian runtime failed")); - let structured_session_error = GuardianReviewError::session_with_error_info( - anyhow::anyhow!("temporary guardian failure"), - CodexErrorInfo::ServerOverloaded, - ); - - assert!(matches!( - parse_error.failure_reason(), - GuardianReviewFailureReason::ParseError - )); - assert!(matches!( - prompt_error.failure_reason(), - GuardianReviewFailureReason::PromptBuildError - )); - assert!(matches!( - session_error.failure_reason(), - GuardianReviewFailureReason::SessionError - )); - assert!(matches!( - structured_session_error.failure_reason(), - GuardianReviewFailureReason::SessionError - )); - } - - #[test] - fn guardian_review_retry_only_retries_transient_session_and_parse_errors() { - let assessment = GuardianAssessment { - risk_level: GuardianRiskLevel::High, - user_authorization: GuardianUserAuthorization::Unknown, - outcome: GuardianAssessmentOutcome::Deny, - rationale: "deny".to_string(), - }; - let transient_error_info = [ - CodexErrorInfo::ServerOverloaded, - CodexErrorInfo::HttpConnectionFailed { - http_status_code: Some(502), - }, - CodexErrorInfo::ResponseStreamConnectionFailed { - http_status_code: Some(503), - }, - CodexErrorInfo::InternalServerError, - CodexErrorInfo::ResponseStreamDisconnected { - http_status_code: None, - }, - ]; - let mut outcomes = transient_error_info - .into_iter() - .map(|error_info| { - ( - GuardianReviewOutcome::Error(GuardianReviewError::session_with_error_info( - anyhow::anyhow!("transient session"), - error_info, - )), - true, - ) - }) - .collect::>(); - outcomes.extend([ - (GuardianReviewOutcome::Completed(assessment), false), - ( - GuardianReviewOutcome::Error(GuardianReviewError::prompt_build(anyhow::anyhow!( - "prompt" - ))), - false, - ), - ( - GuardianReviewOutcome::Error(GuardianReviewError::session(anyhow::anyhow!( - "session" - ))), - false, - ), - ( - GuardianReviewOutcome::Error(GuardianReviewError::session_with_error_info( - anyhow::anyhow!("bad request"), - CodexErrorInfo::BadRequest, - )), - false, - ), - ( - GuardianReviewOutcome::Error(GuardianReviewError::parse(anyhow::anyhow!("parse"))), - true, - ), - ( - GuardianReviewOutcome::Error(GuardianReviewError::Timeout), - false, - ), - ( - GuardianReviewOutcome::Error(GuardianReviewError::Cancelled), - false, - ), - ]); - - for (outcome, expected) in outcomes { - assert_eq!(should_retry_guardian_review(&outcome), expected); - } - } - - #[tokio::test] - async fn guardian_review_retry_wait_honors_cancellation() { - let cancel_token = CancellationToken::new(); - cancel_token.cancel(); - - let error = wait_before_guardian_retry( - /*attempt_count*/ 1, - Instant::now() + Duration::from_secs(/*secs*/ 1), - Some(&cancel_token), - ) - .await; - - assert!(matches!(error, Some(GuardianReviewError::Cancelled))); - } - - #[tokio::test] - async fn guardian_review_retry_wait_honors_deadline() { - let error = wait_before_guardian_retry( - /*attempt_count*/ 1, - Instant::now(), - /*external_cancel*/ None, - ) - .await; - - assert!(matches!(error, Some(GuardianReviewError::Timeout))); - } + }) + .await } diff --git a/codex-rs/core/src/guardian/review_session.rs b/codex-rs/core/src/guardian/review_session.rs index 753be4d3b7..d792e2870b 100644 --- a/codex-rs/core/src/guardian/review_session.rs +++ b/codex-rs/core/src/guardian/review_session.rs @@ -36,6 +36,7 @@ use codex_protocol::models::ResponseItem; use codex_protocol::openai_models::InputModality; use codex_protocol::openai_models::ReasoningEffort as ReasoningEffortConfig; use codex_protocol::protocol::AskForApproval; +#[cfg(test)] use codex_protocol::protocol::CodexErrorInfo; use codex_protocol::protocol::EnvironmentConfigState; use codex_protocol::protocol::ErrorEvent; @@ -100,17 +101,7 @@ use super::reviewer_config::read_only_guardian_permission_profile; const GUARDIAN_INTERRUPT_DRAIN_TIMEOUT: Duration = Duration::from_secs(5); const GUARDIAN_MAX_IMAGE_ITEM_TOKENS: i64 = 10_000; -#[derive(Debug)] -pub(crate) enum GuardianReviewSessionOutcome { - Completed(anyhow::Result>), - PromptBuildFailed(anyhow::Error), - SessionFailed { - error: anyhow::Error, - error_info: Option, - }, - TimedOut, - Aborted, -} +pub(crate) use codex_guardian_reviewer::GuardianReviewSessionOutcome; pub(crate) struct GuardianReviewSessionParams { pub(crate) parent_session: Arc, diff --git a/codex-rs/core/src/guardian/tests.rs b/codex-rs/core/src/guardian/tests.rs index 543f86a898..7494bc814b 100644 --- a/codex-rs/core/src/guardian/tests.rs +++ b/codex-rs/core/src/guardian/tests.rs @@ -146,111 +146,6 @@ impl codex_extension_api::ContextContributor for GuardianMemoryContextProbe { } } -#[test] -fn guardian_rejection_circuit_breaker_interrupts_after_three_consecutive_denials() { - let mut circuit_breaker = GuardianRejectionCircuitBreaker::default(); - assert_eq!( - circuit_breaker.record_denial("turn-1", GuardianRejectionCircuitBreakerPolicy::Standard), - GuardianRejectionCircuitBreakerAction::Continue - ); - assert_eq!( - circuit_breaker.record_denial("turn-1", GuardianRejectionCircuitBreakerPolicy::Standard), - GuardianRejectionCircuitBreakerAction::Continue - ); - assert_eq!( - circuit_breaker.record_denial("turn-1", GuardianRejectionCircuitBreakerPolicy::Standard), - GuardianRejectionCircuitBreakerAction::InterruptTurn { - consecutive_denials: 3, - recent_denials: 3, - } - ); - assert_eq!( - circuit_breaker.record_denial("turn-1", GuardianRejectionCircuitBreakerPolicy::Standard), - GuardianRejectionCircuitBreakerAction::Continue - ); -} - -#[test] -fn guardian_rejection_circuit_breaker_interrupts_cyber_models_after_one_denial() { - let mut circuit_breaker = GuardianRejectionCircuitBreaker::default(); - assert_eq!( - circuit_breaker.record_denial("turn-1", GuardianRejectionCircuitBreakerPolicy::CyberModel), - GuardianRejectionCircuitBreakerAction::InterruptTurn { - consecutive_denials: 1, - recent_denials: 1, - } - ); - assert_eq!( - circuit_breaker.record_denial("turn-1", GuardianRejectionCircuitBreakerPolicy::CyberModel), - GuardianRejectionCircuitBreakerAction::Continue - ); -} - -#[test] -fn guardian_rejection_circuit_breaker_resets_consecutive_denials_on_non_denial() { - let mut circuit_breaker = GuardianRejectionCircuitBreaker::default(); - assert_eq!( - circuit_breaker.record_denial("turn-1", GuardianRejectionCircuitBreakerPolicy::Standard), - GuardianRejectionCircuitBreakerAction::Continue - ); - circuit_breaker.record_non_denial("turn-1"); - assert_eq!( - circuit_breaker.record_denial("turn-1", GuardianRejectionCircuitBreakerPolicy::Standard), - GuardianRejectionCircuitBreakerAction::Continue - ); - assert_eq!( - circuit_breaker.record_denial("turn-1", GuardianRejectionCircuitBreakerPolicy::Standard), - GuardianRejectionCircuitBreakerAction::Continue - ); - assert_eq!( - circuit_breaker.record_denial("turn-1", GuardianRejectionCircuitBreakerPolicy::Standard), - GuardianRejectionCircuitBreakerAction::InterruptTurn { - consecutive_denials: 3, - recent_denials: 4, - } - ); -} - -#[test] -fn auto_review_rejection_circuit_breaker_interrupts_after_ten_recent_denials() { - let mut circuit_breaker = GuardianRejectionCircuitBreaker::default(); - for _ in 0..9 { - assert_eq!( - circuit_breaker - .record_denial("turn-1", GuardianRejectionCircuitBreakerPolicy::Standard), - GuardianRejectionCircuitBreakerAction::Continue - ); - circuit_breaker.record_non_denial("turn-1"); - } - assert_eq!( - circuit_breaker.record_denial("turn-1", GuardianRejectionCircuitBreakerPolicy::Standard), - GuardianRejectionCircuitBreakerAction::InterruptTurn { - consecutive_denials: 1, - recent_denials: 10, - } - ); -} - -#[test] -fn auto_review_rejection_circuit_breaker_forgets_denials_outside_recent_review_window() { - let mut circuit_breaker = GuardianRejectionCircuitBreaker::default(); - for _ in 0..9 { - assert_eq!( - circuit_breaker - .record_denial("turn-1", GuardianRejectionCircuitBreakerPolicy::Standard), - GuardianRejectionCircuitBreakerAction::Continue - ); - circuit_breaker.record_non_denial("turn-1"); - } - for _ in 0..(AUTO_REVIEW_DENIAL_WINDOW_SIZE - 18) { - circuit_breaker.record_non_denial("turn-1"); - } - assert_eq!( - circuit_breaker.record_denial("turn-1", GuardianRejectionCircuitBreakerPolicy::Standard), - GuardianRejectionCircuitBreakerAction::Continue - ); -} - async fn guardian_test_session_and_turn( server: &wiremock::MockServer, ) -> (Arc, Arc) { diff --git a/codex-rs/ext/guardian-reviewer/BUILD.bazel b/codex-rs/ext/guardian-reviewer/BUILD.bazel new file mode 100644 index 0000000000..2c7a70b2c4 --- /dev/null +++ b/codex-rs/ext/guardian-reviewer/BUILD.bazel @@ -0,0 +1,6 @@ +load("//:defs.bzl", "codex_rust_crate") + +codex_rust_crate( + name = "guardian-reviewer", + crate_name = "codex_guardian_reviewer", +) diff --git a/codex-rs/ext/guardian-reviewer/Cargo.toml b/codex-rs/ext/guardian-reviewer/Cargo.toml new file mode 100644 index 0000000000..381e341f31 --- /dev/null +++ b/codex-rs/ext/guardian-reviewer/Cargo.toml @@ -0,0 +1,26 @@ +[package] +edition.workspace = true +license.workspace = true +name = "codex-guardian-reviewer" +version.workspace = true + +[lib] +doctest = false +name = "codex_guardian_reviewer" +path = "src/lib.rs" + +[lints] +workspace = true + +[dependencies] +anyhow = { workspace = true } +codex-analytics = { workspace = true } +codex-protocol = { workspace = true } +rand = { workspace = true } +serde = { workspace = true } +serde_json = { workspace = true } +tokio = { workspace = true } +tokio-util = { workspace = true } + +[dev-dependencies] +pretty_assertions = { workspace = true } diff --git a/codex-rs/core/src/guardian/assessment.rs b/codex-rs/ext/guardian-reviewer/src/assessment.rs similarity index 98% rename from codex-rs/core/src/guardian/assessment.rs rename to codex-rs/ext/guardian-reviewer/src/assessment.rs index 5ce3cf57a0..43c4024ec6 100644 --- a/codex-rs/core/src/guardian/assessment.rs +++ b/codex-rs/ext/guardian-reviewer/src/assessment.rs @@ -104,7 +104,7 @@ pub fn guardian_output_schema() -> Value { /// Prompt fragment that describes the exact JSON contract paired with /// `guardian_output_schema()`. -pub(super) fn guardian_output_contract_prompt() -> &'static str { +pub fn guardian_output_contract_prompt() -> &'static str { r#"You may use read-only tool checks to gather any additional context you need before deciding. When you are ready to answer, your final message must be strict JSON. For low-risk actions, give the final answer directly: {"outcome":"allow"}. diff --git a/codex-rs/core/src/guardian/assessment_tests.rs b/codex-rs/ext/guardian-reviewer/src/assessment_tests.rs similarity index 100% rename from codex-rs/core/src/guardian/assessment_tests.rs rename to codex-rs/ext/guardian-reviewer/src/assessment_tests.rs diff --git a/codex-rs/ext/guardian-reviewer/src/circuit_breaker.rs b/codex-rs/ext/guardian-reviewer/src/circuit_breaker.rs new file mode 100644 index 0000000000..7f1e5ee4eb --- /dev/null +++ b/codex-rs/ext/guardian-reviewer/src/circuit_breaker.rs @@ -0,0 +1,90 @@ +//! Counts policy denials per turn; the host carries out any requested interruption. + +const MAX_CONSECUTIVE_CYBER_GUARDIAN_DENIALS_PER_TURN: u32 = 1; +const MAX_CONSECUTIVE_GUARDIAN_DENIALS_PER_TURN: u32 = 3; +const MAX_RECENT_CYBER_AUTO_REVIEW_DENIALS_PER_TURN: u32 = 1; +const MAX_RECENT_AUTO_REVIEW_DENIALS_PER_TURN: u32 = 10; +pub const AUTO_REVIEW_DENIAL_WINDOW_SIZE: usize = 50; + +#[derive(Debug, Default)] +pub struct GuardianRejectionCircuitBreaker { + turns: std::collections::HashMap, +} + +#[derive(Debug, Default)] +struct GuardianRejectionCircuitBreakerTurn { + consecutive_denials: u32, + recent_denials: std::collections::VecDeque, + interrupt_triggered: bool, +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum GuardianRejectionCircuitBreakerPolicy { + Standard, + CyberModel, +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum GuardianRejectionCircuitBreakerAction { + Continue, + InterruptTurn { + consecutive_denials: u32, + recent_denials: u32, + }, +} + +impl GuardianRejectionCircuitBreaker { + pub fn clear_turn(&mut self, turn_id: &str) { + self.turns.remove(turn_id); + } + + pub fn record_denial( + &mut self, + turn_id: &str, + policy: GuardianRejectionCircuitBreakerPolicy, + ) -> GuardianRejectionCircuitBreakerAction { + let turn = self.turns.entry(turn_id.to_string()).or_default(); + turn.consecutive_denials = turn.consecutive_denials.saturating_add(1); + Self::record_recent_review(turn, /*denied*/ true); + let recent_denials = turn.recent_denials.iter().filter(|denied| **denied).count() as u32; + let (max_consecutive_denials, max_recent_denials) = match policy { + GuardianRejectionCircuitBreakerPolicy::Standard => ( + MAX_CONSECUTIVE_GUARDIAN_DENIALS_PER_TURN, + MAX_RECENT_AUTO_REVIEW_DENIALS_PER_TURN, + ), + GuardianRejectionCircuitBreakerPolicy::CyberModel => ( + MAX_CONSECUTIVE_CYBER_GUARDIAN_DENIALS_PER_TURN, + MAX_RECENT_CYBER_AUTO_REVIEW_DENIALS_PER_TURN, + ), + }; + if !turn.interrupt_triggered + && (turn.consecutive_denials >= max_consecutive_denials + || recent_denials >= max_recent_denials) + { + turn.interrupt_triggered = true; + GuardianRejectionCircuitBreakerAction::InterruptTurn { + consecutive_denials: turn.consecutive_denials, + recent_denials, + } + } else { + GuardianRejectionCircuitBreakerAction::Continue + } + } + + pub fn record_non_denial(&mut self, turn_id: &str) { + let turn = self.turns.entry(turn_id.to_string()).or_default(); + turn.consecutive_denials = 0; + Self::record_recent_review(turn, /*denied*/ false); + } + + fn record_recent_review(turn: &mut GuardianRejectionCircuitBreakerTurn, denied: bool) { + turn.recent_denials.push_back(denied); + if turn.recent_denials.len() > AUTO_REVIEW_DENIAL_WINDOW_SIZE { + turn.recent_denials.pop_front(); + } + } +} + +#[cfg(test)] +#[path = "circuit_breaker_tests.rs"] +mod tests; diff --git a/codex-rs/ext/guardian-reviewer/src/circuit_breaker_tests.rs b/codex-rs/ext/guardian-reviewer/src/circuit_breaker_tests.rs new file mode 100644 index 0000000000..8250c5b150 --- /dev/null +++ b/codex-rs/ext/guardian-reviewer/src/circuit_breaker_tests.rs @@ -0,0 +1,107 @@ +use super::*; +use pretty_assertions::assert_eq; + +#[test] +fn guardian_rejection_circuit_breaker_interrupts_after_three_consecutive_denials() { + let mut circuit_breaker = GuardianRejectionCircuitBreaker::default(); + assert_eq!( + circuit_breaker.record_denial("turn-1", GuardianRejectionCircuitBreakerPolicy::Standard), + GuardianRejectionCircuitBreakerAction::Continue + ); + assert_eq!( + circuit_breaker.record_denial("turn-1", GuardianRejectionCircuitBreakerPolicy::Standard), + GuardianRejectionCircuitBreakerAction::Continue + ); + assert_eq!( + circuit_breaker.record_denial("turn-1", GuardianRejectionCircuitBreakerPolicy::Standard), + GuardianRejectionCircuitBreakerAction::InterruptTurn { + consecutive_denials: 3, + recent_denials: 3, + } + ); + assert_eq!( + circuit_breaker.record_denial("turn-1", GuardianRejectionCircuitBreakerPolicy::Standard), + GuardianRejectionCircuitBreakerAction::Continue + ); +} + +#[test] +fn guardian_rejection_circuit_breaker_interrupts_cyber_models_after_one_denial() { + let mut circuit_breaker = GuardianRejectionCircuitBreaker::default(); + assert_eq!( + circuit_breaker.record_denial("turn-1", GuardianRejectionCircuitBreakerPolicy::CyberModel), + GuardianRejectionCircuitBreakerAction::InterruptTurn { + consecutive_denials: 1, + recent_denials: 1, + } + ); + assert_eq!( + circuit_breaker.record_denial("turn-1", GuardianRejectionCircuitBreakerPolicy::CyberModel), + GuardianRejectionCircuitBreakerAction::Continue + ); +} + +#[test] +fn guardian_rejection_circuit_breaker_resets_consecutive_denials_on_non_denial() { + let mut circuit_breaker = GuardianRejectionCircuitBreaker::default(); + assert_eq!( + circuit_breaker.record_denial("turn-1", GuardianRejectionCircuitBreakerPolicy::Standard), + GuardianRejectionCircuitBreakerAction::Continue + ); + circuit_breaker.record_non_denial("turn-1"); + assert_eq!( + circuit_breaker.record_denial("turn-1", GuardianRejectionCircuitBreakerPolicy::Standard), + GuardianRejectionCircuitBreakerAction::Continue + ); + assert_eq!( + circuit_breaker.record_denial("turn-1", GuardianRejectionCircuitBreakerPolicy::Standard), + GuardianRejectionCircuitBreakerAction::Continue + ); + assert_eq!( + circuit_breaker.record_denial("turn-1", GuardianRejectionCircuitBreakerPolicy::Standard), + GuardianRejectionCircuitBreakerAction::InterruptTurn { + consecutive_denials: 3, + recent_denials: 4, + } + ); +} + +#[test] +fn auto_review_rejection_circuit_breaker_interrupts_after_ten_recent_denials() { + let mut circuit_breaker = GuardianRejectionCircuitBreaker::default(); + for _ in 0..9 { + assert_eq!( + circuit_breaker + .record_denial("turn-1", GuardianRejectionCircuitBreakerPolicy::Standard), + GuardianRejectionCircuitBreakerAction::Continue + ); + circuit_breaker.record_non_denial("turn-1"); + } + assert_eq!( + circuit_breaker.record_denial("turn-1", GuardianRejectionCircuitBreakerPolicy::Standard), + GuardianRejectionCircuitBreakerAction::InterruptTurn { + consecutive_denials: 1, + recent_denials: 10, + } + ); +} + +#[test] +fn auto_review_rejection_circuit_breaker_forgets_denials_outside_recent_review_window() { + let mut circuit_breaker = GuardianRejectionCircuitBreaker::default(); + for _ in 0..9 { + assert_eq!( + circuit_breaker + .record_denial("turn-1", GuardianRejectionCircuitBreakerPolicy::Standard), + GuardianRejectionCircuitBreakerAction::Continue + ); + circuit_breaker.record_non_denial("turn-1"); + } + for _ in 0..(AUTO_REVIEW_DENIAL_WINDOW_SIZE - 18) { + circuit_breaker.record_non_denial("turn-1"); + } + assert_eq!( + circuit_breaker.record_denial("turn-1", GuardianRejectionCircuitBreakerPolicy::Standard), + GuardianRejectionCircuitBreakerAction::Continue + ); +} diff --git a/codex-rs/ext/guardian-reviewer/src/lib.rs b/codex-rs/ext/guardian-reviewer/src/lib.rs new file mode 100644 index 0000000000..4bb0ae8bcb --- /dev/null +++ b/codex-rs/ext/guardian-reviewer/src/lib.rs @@ -0,0 +1,27 @@ +//! Owns synchronous Guardian review policy independently of the host session runtime. +//! The host supplies review attempts and enforces the resulting decision on the bound action. + +mod assessment; +mod circuit_breaker; +mod model; +mod outcome; +mod retry; + +pub use assessment::GuardianAssessment; +pub use assessment::guardian_output_contract_prompt; +pub use assessment::guardian_output_schema; +pub use assessment::parse_guardian_assessment; +pub use circuit_breaker::AUTO_REVIEW_DENIAL_WINDOW_SIZE; +pub use circuit_breaker::GuardianRejectionCircuitBreaker; +pub use circuit_breaker::GuardianRejectionCircuitBreakerAction; +pub use circuit_breaker::GuardianRejectionCircuitBreakerPolicy; +pub use model::ReviewModel; +pub use model::select_review_model; +pub use outcome::GuardianReviewError; +pub use outcome::GuardianReviewOutcome; +pub use outcome::GuardianReviewSessionOutcome; +pub use retry::GuardianReviewSessionLimits; +pub use retry::run_with_retry; + +pub const MAX_REVIEW_ATTEMPTS: i64 = 3; +pub const REVIEW_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(90); diff --git a/codex-rs/ext/guardian-reviewer/src/model.rs b/codex-rs/ext/guardian-reviewer/src/model.rs new file mode 100644 index 0000000000..df9204ade7 --- /dev/null +++ b/codex-rs/ext/guardian-reviewer/src/model.rs @@ -0,0 +1,76 @@ +//! Selects the synchronous review model from the parent's metadata and provider catalog. + +use codex_protocol::openai_models::ModelInfo; +use codex_protocol::openai_models::ModelPreset; +use codex_protocol::openai_models::ReasoningEffort; + +/// Effective model choice, including the attribution carried by existing review events. +pub struct ReviewModel { + pub model: String, + pub reasoning_effort: Option, + pub default_review_model_id: String, + pub catalog_contains_auto_review: bool, + pub model_overridden: bool, + pub model_override: Option, +} + +/// Prefers the review model and low effort when advertised; preserves the parent fallback. +pub fn select_review_model( + parent_model: &ModelInfo, + parent_reasoning_effort: Option<&ReasoningEffort>, + default_review_model_id: &str, + available_models: &[ModelPreset], +) -> ReviewModel { + let preferred_reasoning_effort = |supports_low: bool, fallback| { + if supports_low { + Some(codex_protocol::openai_models::ReasoningEffort::Low) + } else { + fallback + } + }; + let model_override = parent_model.auto_review_model_override.as_deref(); + let review_model_id = model_override.unwrap_or(default_review_model_id); + let review_model = available_models + .iter() + .find(|preset| preset.model == review_model_id); + let guardian_catalog_contains_auto_review = available_models + .iter() + .any(|preset| preset.model == default_review_model_id); + let guardian_review_model_overridden = model_override.is_some(); + let guardian_review_model_override = model_override.map(str::to_string); + let (guardian_model, guardian_reasoning_effort) = if let Some(preset) = review_model { + let reasoning_effort = preferred_reasoning_effort( + preset + .supported_reasoning_efforts + .iter() + .any(|effort| effort.effort == codex_protocol::openai_models::ReasoningEffort::Low), + Some(preset.default_reasoning_effort.clone()), + ); + (review_model_id.to_string(), reasoning_effort) + } else { + let reasoning_effort = preferred_reasoning_effort( + parent_model + .supported_reasoning_levels + .iter() + .any(|preset| preset.effort == codex_protocol::openai_models::ReasoningEffort::Low), + parent_reasoning_effort + .or(parent_model.default_reasoning_level.as_ref()) + .cloned(), + ); + ( + model_override + .unwrap_or(parent_model.slug.as_str()) + .to_string(), + reasoning_effort, + ) + }; + + ReviewModel { + model: guardian_model, + reasoning_effort: guardian_reasoning_effort, + default_review_model_id: default_review_model_id.to_string(), + catalog_contains_auto_review: guardian_catalog_contains_auto_review, + model_overridden: guardian_review_model_overridden, + model_override: guardian_review_model_override, + } +} diff --git a/codex-rs/ext/guardian-reviewer/src/outcome.rs b/codex-rs/ext/guardian-reviewer/src/outcome.rs new file mode 100644 index 0000000000..e376897cbc --- /dev/null +++ b/codex-rs/ext/guardian-reviewer/src/outcome.rs @@ -0,0 +1,111 @@ +//! Distinguishes completed assessments from failures without assigning risk to errors. + +use crate::GuardianAssessment; +use codex_analytics::GuardianReviewFailureReason; +use codex_protocol::protocol::CodexErrorInfo; + +#[derive(Debug)] +pub enum GuardianReviewOutcome { + Completed(GuardianAssessment), + Error(GuardianReviewError), +} + +#[derive(Debug)] +pub enum GuardianReviewError { + PromptBuild { + message: String, + }, + Session { + message: String, + error_info: Option, + }, + Parse { + message: String, + }, + Timeout, + Cancelled, +} + +impl GuardianReviewError { + pub fn prompt_build(err: anyhow::Error) -> Self { + Self::PromptBuild { + message: err.to_string(), + } + } + + pub fn session(err: anyhow::Error) -> Self { + Self::Session { + message: err.to_string(), + error_info: None, + } + } + + pub fn session_with_error_info(err: anyhow::Error, error_info: CodexErrorInfo) -> Self { + Self::Session { + message: err.to_string(), + error_info: Some(error_info), + } + } + + pub fn parse(err: anyhow::Error) -> Self { + Self::Parse { + message: err.to_string(), + } + } + + pub fn failure_reason(&self) -> GuardianReviewFailureReason { + match self { + Self::PromptBuild { .. } => GuardianReviewFailureReason::PromptBuildError, + Self::Session { .. } => GuardianReviewFailureReason::SessionError, + Self::Parse { .. } => GuardianReviewFailureReason::ParseError, + Self::Timeout => GuardianReviewFailureReason::Timeout, + Self::Cancelled => GuardianReviewFailureReason::Cancelled, + } + } +} + +#[derive(Debug)] +pub enum GuardianReviewSessionOutcome { + Completed(anyhow::Result>), + PromptBuildFailed(anyhow::Error), + SessionFailed { + error: anyhow::Error, + error_info: Option, + }, + TimedOut, + Aborted, +} + +impl From for GuardianReviewOutcome { + fn from(outcome: GuardianReviewSessionOutcome) -> Self { + match outcome { + GuardianReviewSessionOutcome::Completed(Ok(Some(message))) => { + match crate::parse_guardian_assessment(Some(&message)) { + Ok(assessment) => Self::Completed(assessment), + Err(error) => Self::Error(GuardianReviewError::parse(error)), + } + } + GuardianReviewSessionOutcome::Completed(Ok(None)) => { + Self::Error(GuardianReviewError::session(anyhow::anyhow!( + "guardian review completed without an assessment payload" + ))) + } + GuardianReviewSessionOutcome::Completed(Err(error)) => { + Self::Error(GuardianReviewError::session(error)) + } + GuardianReviewSessionOutcome::PromptBuildFailed(error) => { + Self::Error(GuardianReviewError::prompt_build(error)) + } + GuardianReviewSessionOutcome::SessionFailed { error, error_info } => { + Self::Error(match error_info { + Some(error_info) => { + GuardianReviewError::session_with_error_info(error, error_info) + } + None => GuardianReviewError::session(error), + }) + } + GuardianReviewSessionOutcome::TimedOut => Self::Error(GuardianReviewError::Timeout), + GuardianReviewSessionOutcome::Aborted => Self::Error(GuardianReviewError::Cancelled), + } + } +} diff --git a/codex-rs/ext/guardian-reviewer/src/retry.rs b/codex-rs/ext/guardian-reviewer/src/retry.rs new file mode 100644 index 0000000000..90166bfbbe --- /dev/null +++ b/codex-rs/ext/guardian-reviewer/src/retry.rs @@ -0,0 +1,94 @@ +//! Applies one review's attempt budget, deadline and cancellation to host-owned sessions. +//! Resubmissions are independent; this layer keeps no state between approval requests. + +use crate::GuardianReviewError; +use crate::GuardianReviewOutcome; +use codex_analytics::GuardianReviewAnalyticsResult; +use codex_protocol::protocol::CodexErrorInfo; +use rand::Rng; +use std::future::Future; +use std::time::Duration; +use tokio::time::Instant; +use tokio::time::sleep_until; +use tokio_util::sync::CancellationToken; + +pub struct GuardianReviewSessionLimits { + pub max_attempts: i64, + pub deadline: Instant, +} + +/// Retries only recoverable review failures within one shared deadline. +pub async fn run_with_retry( + limits: GuardianReviewSessionLimits, + external_cancel: Option<&CancellationToken>, + mut run_attempt: F, +) -> (GuardianReviewOutcome, GuardianReviewAnalyticsResult) +where + F: FnMut(Instant) -> Attempt, + Attempt: Future, +{ + let GuardianReviewSessionLimits { + max_attempts, + deadline, + } = limits; + assert!(max_attempts > 0, "guardian review must run at least once"); + let mut attempt_count = 1; + loop { + let (outcome, mut analytics_result) = run_attempt(deadline).await; + analytics_result.attempt_count = attempt_count; + if attempt_count >= max_attempts || !should_retry_guardian_review(&outcome) { + return (outcome, analytics_result); + } + if let Some(error) = + wait_before_guardian_retry(attempt_count, deadline, external_cancel).await + { + return (GuardianReviewOutcome::Error(error), analytics_result); + } + attempt_count += 1; + } +} + +async fn wait_before_guardian_retry( + attempt_count: i64, + deadline: Instant, + external_cancel: Option<&CancellationToken>, +) -> Option { + let exponential_delay = 200.0 * 2.0_f64.powi(attempt_count.saturating_sub(1) as i32); + let jitter = rand::rng().random_range(0.9..1.1); + let retry_delay = Duration::from_millis((exponential_delay * jitter) as u64); + let retry_at = (Instant::now() + retry_delay).min(deadline); + tokio::select! { + _ = sleep_until(retry_at) => { + (Instant::now() >= deadline).then_some(GuardianReviewError::Timeout) + } + _ = async { + if let Some(cancel_token) = external_cancel { + cancel_token.cancelled().await; + } else { + std::future::pending::<()>().await; + } + } => Some(GuardianReviewError::Cancelled), + } +} + +fn should_retry_guardian_review(outcome: &GuardianReviewOutcome) -> bool { + matches!( + outcome, + GuardianReviewOutcome::Error( + GuardianReviewError::Session { + error_info: Some( + CodexErrorInfo::ServerOverloaded + | CodexErrorInfo::HttpConnectionFailed { .. } + | CodexErrorInfo::ResponseStreamConnectionFailed { .. } + | CodexErrorInfo::InternalServerError + | CodexErrorInfo::ResponseStreamDisconnected { .. } + ), + .. + } | GuardianReviewError::Parse { .. } + ) + ) +} + +#[cfg(test)] +#[path = "retry_tests.rs"] +mod tests; diff --git a/codex-rs/ext/guardian-reviewer/src/retry_tests.rs b/codex-rs/ext/guardian-reviewer/src/retry_tests.rs new file mode 100644 index 0000000000..715598c850 --- /dev/null +++ b/codex-rs/ext/guardian-reviewer/src/retry_tests.rs @@ -0,0 +1,135 @@ +#[cfg(test)] +use super::*; +use crate::GuardianAssessment; +use codex_analytics::GuardianReviewFailureReason; +use codex_protocol::protocol::GuardianAssessmentOutcome; +use codex_protocol::protocol::GuardianRiskLevel; +use codex_protocol::protocol::GuardianUserAuthorization; +use pretty_assertions::assert_eq; +use std::time::Duration; + +#[test] +fn guardian_review_error_reason_distinguishes_error_kinds() { + let parse_error = GuardianReviewError::parse(anyhow::anyhow!("bad guardian JSON")); + let prompt_error = GuardianReviewError::prompt_build(anyhow::anyhow!("bad prompt/config")); + let session_error = GuardianReviewError::session(anyhow::anyhow!("guardian runtime failed")); + let structured_session_error = GuardianReviewError::session_with_error_info( + anyhow::anyhow!("temporary guardian failure"), + CodexErrorInfo::ServerOverloaded, + ); + + assert!(matches!( + parse_error.failure_reason(), + GuardianReviewFailureReason::ParseError + )); + assert!(matches!( + prompt_error.failure_reason(), + GuardianReviewFailureReason::PromptBuildError + )); + assert!(matches!( + session_error.failure_reason(), + GuardianReviewFailureReason::SessionError + )); + assert!(matches!( + structured_session_error.failure_reason(), + GuardianReviewFailureReason::SessionError + )); +} + +#[test] +fn guardian_review_retry_only_retries_transient_session_and_parse_errors() { + let assessment = GuardianAssessment { + risk_level: GuardianRiskLevel::High, + user_authorization: GuardianUserAuthorization::Unknown, + outcome: GuardianAssessmentOutcome::Deny, + rationale: "deny".to_string(), + }; + let transient_error_info = [ + CodexErrorInfo::ServerOverloaded, + CodexErrorInfo::HttpConnectionFailed { + http_status_code: Some(502), + }, + CodexErrorInfo::ResponseStreamConnectionFailed { + http_status_code: Some(503), + }, + CodexErrorInfo::InternalServerError, + CodexErrorInfo::ResponseStreamDisconnected { + http_status_code: None, + }, + ]; + let mut outcomes = transient_error_info + .into_iter() + .map(|error_info| { + ( + GuardianReviewOutcome::Error(GuardianReviewError::session_with_error_info( + anyhow::anyhow!("transient session"), + error_info, + )), + true, + ) + }) + .collect::>(); + outcomes.extend([ + (GuardianReviewOutcome::Completed(assessment), false), + ( + GuardianReviewOutcome::Error(GuardianReviewError::prompt_build(anyhow::anyhow!( + "prompt" + ))), + false, + ), + ( + GuardianReviewOutcome::Error(GuardianReviewError::session(anyhow::anyhow!("session"))), + false, + ), + ( + GuardianReviewOutcome::Error(GuardianReviewError::session_with_error_info( + anyhow::anyhow!("bad request"), + CodexErrorInfo::BadRequest, + )), + false, + ), + ( + GuardianReviewOutcome::Error(GuardianReviewError::parse(anyhow::anyhow!("parse"))), + true, + ), + ( + GuardianReviewOutcome::Error(GuardianReviewError::Timeout), + false, + ), + ( + GuardianReviewOutcome::Error(GuardianReviewError::Cancelled), + false, + ), + ]); + + for (outcome, expected) in outcomes { + assert_eq!(should_retry_guardian_review(&outcome), expected); + } +} + +#[tokio::test] +async fn guardian_review_retry_wait_honors_cancellation() { + let cancel_token = CancellationToken::new(); + cancel_token.cancel(); + + let error = wait_before_guardian_retry( + /*attempt_count*/ 1, + Instant::now() + Duration::from_secs(/*secs*/ 1), + Some(&cancel_token), + ) + .await; + + assert!(matches!(error, Some(GuardianReviewError::Cancelled))); +} + +#[tokio::test] +async fn guardian_review_retry_wait_honors_deadline() { + let error = wait_before_guardian_retry( + /*attempt_count*/ 1, + Instant::now(), + /*external_cancel*/ None, + ) + .await; + + assert!(matches!(error, Some(GuardianReviewError::Timeout))); +}