From 4e6d5c0a96bdc8a8ef3b571bcf2c173141f5e7ee Mon Sep 17 00:00:00 2001 From: jif Date: Thu, 10 Sep 2026 14:03:41 +0000 Subject: [PATCH] Move Guardian reporting and denial accounting into the extension (#44544) ## What changed - Move assessment event construction, metrics, and analytics tracking into `codex-guardian-reviewer` through `ReviewReport`. - Store denial accounting in thread extension data through `ReviewDenials`, with core retaining turn interruption and lifecycle cleanup. - Move failed-review record selection and bounded serialization into the extension, with core supplying captured review context. ## Testing Add coverage that denial accounting clears on turn completion and interruption. Move the oversized-record test into the extension and exercise the new API. GitOrigin-RevId: 6e5b3d9d4128b356f99b1d0e74a68111e40261ab --- codex-rs/Cargo.lock | 7 + codex-rs/core/src/guardian/feedback.rs | 102 ++--------- codex-rs/core/src/guardian/mod.rs | 6 - codex-rs/core/src/guardian/prompt.rs | 1 - codex-rs/core/src/guardian/review.rs | 66 ++----- codex-rs/core/src/guardian/review_request.rs | 151 ++++------------ codex-rs/core/src/session/session.rs | 1 - codex-rs/core/src/session/tests.rs | 2 - .../core/src/session/tests/guardian_tests.rs | 45 +++++ codex-rs/core/src/state/service.rs | 2 - codex-rs/core/src/tasks/mod.rs | 30 ++-- codex-rs/ext/guardian-reviewer/Cargo.toml | 7 + .../ext/guardian-reviewer/src/feedback.rs | 141 +++++++++++++++ .../guardian-reviewer/src}/feedback_tests.rs | 17 +- codex-rs/ext/guardian-reviewer/src/lib.rs | 10 ++ .../guardian-reviewer/src}/metrics.rs | 2 + .../ext/guardian-reviewer/src/reporting.rs | 161 ++++++++++++++++++ 17 files changed, 455 insertions(+), 296 deletions(-) create mode 100644 codex-rs/ext/guardian-reviewer/src/feedback.rs rename codex-rs/{core/src/guardian => ext/guardian-reviewer/src}/feedback_tests.rs (79%) rename codex-rs/{core/src/guardian => ext/guardian-reviewer/src}/metrics.rs (99%) create mode 100644 codex-rs/ext/guardian-reviewer/src/reporting.rs diff --git a/codex-rs/Cargo.lock b/codex-rs/Cargo.lock index 037d27b64d..5394382df2 100644 --- a/codex-rs/Cargo.lock +++ b/codex-rs/Cargo.lock @@ -3563,13 +3563,20 @@ dependencies = [ "codex-analytics", "codex-extension-api", "codex-features", + "codex-feedback", + "codex-mcp", + "codex-otel", "codex-protocol", + "opentelemetry", + "opentelemetry_sdk", "pretty_assertions", "rand 0.9.3", "serde", "serde_json", + "test-case", "tokio", "tokio-util", + "tracing", ] [[package]] diff --git a/codex-rs/core/src/guardian/feedback.rs b/codex-rs/core/src/guardian/feedback.rs index fa58870461..64c964611f 100644 --- a/codex-rs/core/src/guardian/feedback.rs +++ b/codex-rs/core/src/guardian/feedback.rs @@ -1,63 +1,25 @@ -//! Captures bounded failed-review evidence for opt-in feedback uploads. -//! Records outlive replaced and short-lived reviewers without writing extra history files. +//! Captures failed-review context for the extension's bounded feedback recorder. -use super::GuardianAssessmentOutcome; use super::approval_request::format_guardian_action_pretty; use super::approval_request::guardian_request_target_item_id; use super::approval_request::guardian_request_turn_id; -use super::prompt::parse_guardian_assessment; use super::review_session::GuardianReviewSessionOutcome; use super::review_session::GuardianReviewSessionParams; use crate::session::session::Session; -use codex_feedback::record_guardian_review_failure; -use codex_protocol::ThreadId; -use codex_protocol::models::ResponseItem; -use serde::Serialize; -use std::io; -use std::io::Write; - -const MAX_RECORD_BYTES: usize = 4 * 1024 * 1024; - -#[derive(Serialize)] -struct ReviewFeedbackRecord<'a> { - reviewed_thread_id: ThreadId, - reviewed_turn_id: &'a str, - target_item_id: Option<&'a str>, - reviewer_thread_id: ThreadId, - model: &'a str, - status: &'a str, - decision: Option<&'a str>, - action: &'a str, - action_truncated: bool, - instructions: Option<&'a str>, - history: Vec<&'a ResponseItem>, - context_omitted: bool, -} pub(super) async fn record_failed_review( reviewer: &Session, params: &GuardianReviewSessionParams, outcome: &GuardianReviewSessionOutcome, ) { - if !params.spawn_config.feedback_enabled || params.spawn_config.ephemeral { + let Some(feedback) = codex_guardian_reviewer::FailedReviewFeedback::for_outcome( + outcome, + codex_guardian_reviewer::ReviewFeedbackSettings { + enabled: params.spawn_config.feedback_enabled, + ephemeral: params.spawn_config.ephemeral, + }, + ) else { return; - } - let (status, decision) = match outcome { - GuardianReviewSessionOutcome::Completed(Ok(Some(decision))) => { - match parse_guardian_assessment(Some(decision)) { - Ok(assessment) if assessment.outcome == GuardianAssessmentOutcome::Allow => { - return; - } - Ok(_) => ("denied", Some(decision.as_str())), - Err(_) => ("invalid_decision", Some(decision.as_str())), - } - } - GuardianReviewSessionOutcome::Completed(Ok(None)) => ("missing_decision", None), - GuardianReviewSessionOutcome::Completed(Err(_)) - | GuardianReviewSessionOutcome::PromptBuildFailed(_) - | GuardianReviewSessionOutcome::SessionFailed { .. } => ("failed", None), - GuardianReviewSessionOutcome::TimedOut => ("timed_out", None), - GuardianReviewSessionOutcome::Aborted => ("aborted", None), }; let Ok(action) = format_guardian_action_pretty(¶ms.request) else { tracing::warn!("Could not serialize Guardian feedback action"); @@ -65,7 +27,7 @@ pub(super) async fn record_failed_review( }; let instructions = reviewer.get_prompt_base_instructions().await; let history = reviewer.clone_history().await; - ReviewFeedbackRecord { + feedback.store(codex_guardian_reviewer::ReviewFeedbackContext { reviewed_thread_id: params.parent_session.thread_id(), reviewed_turn_id: guardian_request_turn_id( ¶ms.request, @@ -74,53 +36,9 @@ pub(super) async fn record_failed_review( target_item_id: guardian_request_target_item_id(¶ms.request), reviewer_thread_id: reviewer.thread_id(), model: ¶ms.model, - status, - decision, action: &action.text, action_truncated: action.truncated, instructions: Some(&instructions.text), history: history.raw_items().collect(), - context_omitted: false, - } - .store(); + }); } - -impl ReviewFeedbackRecord<'_> { - fn store(mut self) { - let mut buffer = BoundedBuffer(Vec::new()); - if serde_json::to_writer(&mut buffer, &self).is_err() { - // Preserve the action and decision even when optional reviewer history is too large. - self.instructions = None; - self.history.clear(); - self.context_omitted = true; - buffer = BoundedBuffer(Vec::new()); - if serde_json::to_writer(&mut buffer, &self).is_err() { - tracing::warn!("Guardian feedback action and decision exceed the record limit"); - return; - } - } - record_guardian_review_failure(self.reviewed_thread_id, buffer.0); - } -} - -struct BoundedBuffer(Vec); - -impl Write for BoundedBuffer { - fn write(&mut self, bytes: &[u8]) -> io::Result { - if bytes.len() > MAX_RECORD_BYTES.saturating_sub(self.0.len()) { - return Err(io::Error::other( - "Guardian feedback record exceeds its size limit", - )); - } - self.0.extend_from_slice(bytes); - Ok(bytes.len()) - } - - fn flush(&mut self) -> io::Result<()> { - Ok(()) - } -} - -#[cfg(test)] -#[path = "feedback_tests.rs"] -mod tests; diff --git a/codex-rs/core/src/guardian/mod.rs b/codex-rs/core/src/guardian/mod.rs index b384b302a1..862d2eb995 100644 --- a/codex-rs/core/src/guardian/mod.rs +++ b/codex-rs/core/src/guardian/mod.rs @@ -7,7 +7,6 @@ mod coverage; mod decision; mod feedback; mod input_budget; -mod metrics; mod prompt; pub(crate) use input_budget::PendingReviewContext; pub(crate) use input_budget::check_pending as check_pending_guardian_input; @@ -62,7 +61,6 @@ pub(crate) use runtime::ReviewAction; pub(crate) use codex_guardian_reviewer::REVIEW_TIMEOUT as GUARDIAN_REVIEW_TIMEOUT; pub(crate) const GUARDIAN_REVIEWER_NAME: &str = "guardian"; -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() @@ -171,10 +169,6 @@ impl From<&Arc> for GuardianReviewContext { #[cfg(test)] use codex_guardian_reviewer::guardian_output_schema; -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)] use approval_request::guardian_assessment_action; diff --git a/codex-rs/core/src/guardian/prompt.rs b/codex-rs/core/src/guardian/prompt.rs index 4ff50b7940..953cb94c61 100644 --- a/codex-rs/core/src/guardian/prompt.rs +++ b/codex-rs/core/src/guardian/prompt.rs @@ -361,7 +361,6 @@ pub(crate) fn guardian_truncate_text(content: &str, token_cap: usize) -> (String } use codex_guardian_reviewer::guardian_output_contract_prompt; -pub(super) 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 840f031508..f88c987219 100644 --- a/codex-rs/core/src/guardian/review.rs +++ b/codex-rs/core/src/guardian/review.rs @@ -1,4 +1,4 @@ -//! Supplies host review preparation, telemetry and runtime configuration. +//! Supplies host review preparation, interruption and runtime configuration. //! Guardian's extension owns the synchronous review loop and pool. #[path = "review_request.rs"] @@ -7,11 +7,6 @@ mod request; use crate::context::GuardianContextMode; use codex_analytics::GuardianApprovalRequestSource; use codex_analytics::GuardianReviewAnalyticsResult; -use codex_analytics::GuardianReviewDecision; -use codex_analytics::GuardianReviewFailureReason; -use codex_analytics::GuardianReviewTerminalStatus; -use codex_analytics::GuardianReviewTrackContext; -use codex_analytics::GuardianReviewedAction; use codex_core_plugins::PluginCommandAttribution; use codex_extension_api::ThreadIdleCause; use codex_features::Feature; @@ -22,9 +17,6 @@ use codex_guardian_reviewer::GuardianReviewSessionLimits; use codex_protocol::config_types::ApprovalsReviewer; use codex_protocol::protocol::AskForApproval; use codex_protocol::protocol::EventMsg; -use codex_protocol::protocol::GuardianAssessmentDecisionSource; -use codex_protocol::protocol::GuardianAssessmentEvent; -use codex_protocol::protocol::GuardianAssessmentStatus; use codex_protocol::protocol::InternalSessionSource; use codex_protocol::protocol::ReviewDecision; use codex_protocol::protocol::SessionSource; @@ -42,21 +34,18 @@ use crate::session::session::Session; use crate::session::turn_context::TurnContext; use crate::turn_timing::now_unix_timestamp_ms; -use super::AUTO_REVIEW_DENIAL_WINDOW_SIZE; use super::ApprovalRequestReasons; +#[cfg(test)] use super::GUARDIAN_REVIEW_TIMEOUT; use super::GUARDIAN_REVIEWER_NAME; use super::GuardianApprovalRequest; use super::GuardianAssessmentOutcome; -use super::GuardianRejectionCircuitBreakerAction; -use super::GuardianRejectionCircuitBreakerPolicy; use super::GuardianReviewContext; use super::approval_request::format_guardian_action_pretty; 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::metrics::emit_guardian_review_metrics; use super::review_session::GuardianReviewSessionParams; use super::review_session::build_guardian_review_session_config; use codex_guardian_reviewer::guardian_output_schema; @@ -147,48 +136,17 @@ pub(crate) fn is_basic_session_source(session_source: &SessionSource) -> bool { } } -fn track_guardian_review( - session: &Session, - tracking: &GuardianReviewTrackContext, - approval_request_source: GuardianApprovalRequestSource, - reviewed_action: &GuardianReviewedAction, - result: GuardianReviewAnalyticsResult, - completed_at_ms: u64, -) { - emit_guardian_review_metrics( - &session.services.session_telemetry, - &result, - approval_request_source, - reviewed_action, - completed_at_ms.saturating_sub(tracking.started_at_ms), - ); - session - .services - .analytics_events_client - .track_guardian_review(tracking, result, completed_at_ms); -} - pub(super) async fn record_guardian_non_denial(session: &Arc, turn_id: &str) { - session - .services - .guardian_rejection_circuit_breaker - .lock() - .await - .record_non_denial(turn_id); + codex_guardian_reviewer::ReviewDenials::for_thread(&session.services.thread_extension_data) + .record_non_denial(turn_id) + .await; } async fn record_guardian_denial(session: &Arc, turn: &Arc, turn_id: &str) { - let policy = GuardianRejectionCircuitBreakerPolicy::from(turn.model_info().as_ref()); - let action = session - .services - .guardian_rejection_circuit_breaker - .lock() - .await - .record_denial(turn_id, policy); - let GuardianRejectionCircuitBreakerAction::InterruptTurn { - consecutive_denials, - recent_denials, - } = action + let Some(message) = + codex_guardian_reviewer::ReviewDenials::for_thread(&session.services.thread_extension_data) + .record_denial(turn_id, turn.model_info()) + .await else { return; }; @@ -200,11 +158,7 @@ async fn record_guardian_denial(session: &Arc, turn: &Arc, session .send_event( turn.as_ref(), - EventMsg::GuardianWarning(WarningEvent { - message: format!( - "Automatic approval review rejected too many approval requests for this turn ({consecutive_denials} consecutive, {recent_denials} in the last {AUTO_REVIEW_DENIAL_WINDOW_SIZE} reviews); interrupting the turn." - ), - }), + EventMsg::GuardianWarning(WarningEvent { message }), ) .await; diff --git a/codex-rs/core/src/guardian/review_request.rs b/codex-rs/core/src/guardian/review_request.rs index 478dbc9e08..c546c5aebc 100644 --- a/codex-rs/core/src/guardian/review_request.rs +++ b/codex-rs/core/src/guardian/review_request.rs @@ -4,21 +4,12 @@ use super::*; use crate::codex_thread::GuardianAuthorizationVersion; use codex_guardian_reviewer::ReviewHost; -use codex_protocol::approvals::GuardianAssessmentAction; use codex_protocol::approvals::GuardianReviewReason; pub(in crate::guardian) struct PreparedApproval { request: GuardianApprovalRequest, turn: Arc, - review_reason: GuardianReviewReason, - assessment_turn_id: String, - target_item_id: Option, - plugin_id: Option, - script_path: Option, - action_summary: GuardianAssessmentAction, - reviewed_action: GuardianReviewedAction, - review_tracking: GuardianReviewTrackContext, - started_at_ms: i64, + report: codex_guardian_reviewer::ReviewReport, root_authorization_version: Option, user_message_revision: u64, review_evidence: Option<( @@ -98,80 +89,46 @@ impl ReviewHost for super::super::runtime::ReviewRuntime { .as_ref() .map(PluginCommandAttribution::serialized_fields) .unzip(); - let action_summary = guardian_assessment_action(&request); - let reviewed_action = guardian_reviewed_action(&request); - let review_tracking = GuardianReviewTrackContext::new( - session.thread_id.to_string(), - assessment_turn_id.clone(), - review_id.clone(), - target_item_id.clone(), - approval_request_source, - reviewed_action.clone(), - GUARDIAN_REVIEW_TIMEOUT.as_millis() as u64, - ); - let started_at_ms = review_tracking.started_at_ms.try_into().unwrap_or_default(); + let report = + codex_guardian_reviewer::ReviewReport::new(codex_guardian_reviewer::ReviewMetadata { + thread_id: session.thread_id.to_string(), + turn_id: assessment_turn_id, + review_id, + target_item_id, + plugin_id, + script_path, + approval_request_source, + reviewed_action: guardian_reviewed_action(&request), + action: guardian_assessment_action(&request), + review_reason, + }); session .send_event( turn.as_ref(), - EventMsg::GuardianAssessment(GuardianAssessmentEvent { - review_reason: Some(review_reason), - id: review_id.clone(), - target_item_id: target_item_id.clone(), - plugin_id: plugin_id.clone(), - script_path: script_path.clone(), - turn_id: assessment_turn_id.clone(), - started_at_ms, - completed_at_ms: None, - status: GuardianAssessmentStatus::InProgress, - risk_level: None, - user_authorization: None, - rationale: None, - decision_source: None, - action: action_summary.clone(), - }), + EventMsg::GuardianAssessment(report.started_event()), ) .await; - if external_cancel .as_ref() .is_some_and(CancellationToken::is_cancelled) { let completed_at_ms = now_unix_timestamp_ms(); - track_guardian_review( - session.as_ref(), - &review_tracking, - approval_request_source, - &reviewed_action, - GuardianReviewAnalyticsResult { - decision: GuardianReviewDecision::Aborted, - terminal_status: GuardianReviewTerminalStatus::Aborted, - failure_reason: Some(GuardianReviewFailureReason::Cancelled), - ..GuardianReviewAnalyticsResult::without_session() - }, + let completed = report.complete( + GuardianReviewOutcome::Error(GuardianReviewError::Cancelled), + turn.model_info(), + GuardianReviewAnalyticsResult::without_session(), + completed_at_ms, + ); + report.track( + &session.services.session_telemetry, + &session.services.analytics_events_client, + completed.analytics, completed_at_ms.try_into().unwrap_or_default(), ); session - .send_event( - turn.as_ref(), - EventMsg::GuardianAssessment(GuardianAssessmentEvent { - review_reason: Some(review_reason), - id: review_id, - target_item_id, - plugin_id: plugin_id.clone(), - script_path: script_path.clone(), - turn_id: assessment_turn_id.clone(), - started_at_ms, - completed_at_ms: Some(completed_at_ms), - status: GuardianAssessmentStatus::Aborted, - risk_level: None, - user_authorization: None, - rationale: None, - decision_source: Some(GuardianAssessmentDecisionSource::Agent), - action: action_summary, - }), - ) + .send_event(turn.as_ref(), EventMsg::GuardianAssessment(completed.event)) .await; - record_guardian_non_denial(&session, &assessment_turn_id).await; + record_guardian_non_denial(&session, report.turn_id()).await; return Err(ReviewDecision::Abort); } @@ -205,15 +162,7 @@ impl ReviewHost for super::super::runtime::ReviewRuntime { Ok(PreparedApproval { request, turn, - review_reason, - assessment_turn_id, - target_item_id, - plugin_id, - script_path, - action_summary, - reviewed_action, - review_tracking, - started_at_ms, + report, root_authorization_version, user_message_revision, review_evidence, @@ -246,23 +195,12 @@ impl ReviewHost for super::super::runtime::ReviewRuntime { let PreparedApproval { request: _, turn, - review_reason, - assessment_turn_id, - target_item_id, - plugin_id, - script_path, - action_summary, - reviewed_action, - review_tracking, - started_at_ms, + report, root_authorization_version, user_message_revision, review_evidence, } = prepared; let session = Arc::clone(&self.session); - let review_id = self.review_id.clone(); - let approval_request_source = self.options.approval_request_source; - let terminal_action = action_summary.clone(); if session.guardian_context_mode == GuardianContextMode::ThreadOwned && matches!(&outcome, GuardianReviewOutcome::Completed(assessment) if assessment.outcome == GuardianAssessmentOutcome::Allow) && (root_authorization_version @@ -284,32 +222,15 @@ impl ReviewHost for super::super::runtime::ReviewRuntime { } let completed_at_ms = now_unix_timestamp_ms(); - let completed = codex_guardian_reviewer::complete_review( + let completed = report.complete( outcome, turn.model_info(), - GuardianAssessmentEvent { - review_reason: Some(review_reason), - id: review_id, - target_item_id, - plugin_id, - script_path, - turn_id: assessment_turn_id.clone(), - started_at_ms, - completed_at_ms: Some(completed_at_ms), - status: GuardianAssessmentStatus::InProgress, - risk_level: None, - user_authorization: None, - rationale: None, - decision_source: Some(GuardianAssessmentDecisionSource::Agent), - action: terminal_action, - }, analytics_result, + completed_at_ms, ); - track_guardian_review( - session.as_ref(), - &review_tracking, - approval_request_source, - &reviewed_action, + report.track( + &session.services.session_telemetry, + &session.services.analytics_events_client, completed.analytics, completed_at_ms.try_into().unwrap_or_default(), ); @@ -336,9 +257,9 @@ impl ReviewHost for super::super::runtime::ReviewRuntime { .send_event(turn.as_ref(), EventMsg::GuardianAssessment(completed.event)) .await; if completed.assessment_outcome == Some(GuardianAssessmentOutcome::Deny) { - record_guardian_denial(&session, &turn, &assessment_turn_id).await; + record_guardian_denial(&session, &turn, report.turn_id()).await; } else { - record_guardian_non_denial(&session, &assessment_turn_id).await; + record_guardian_non_denial(&session, report.turn_id()).await; } completed.decision } diff --git a/codex-rs/core/src/session/session.rs b/codex-rs/core/src/session/session.rs index ffe66620c4..1f2a4b0517 100644 --- a/codex-rs/core/src/session/session.rs +++ b/codex-rs/core/src/session/session.rs @@ -1514,7 +1514,6 @@ impl Session { models_manager: Arc::clone(&models_manager), git_root_discovery, tool_approvals: Mutex::new(ApprovalStore::default()), - guardian_rejection_circuit_breaker: Mutex::new(Default::default()), runtime_handle: tokio::runtime::Handle::current(), skills_service, agents_md_manager, diff --git a/codex-rs/core/src/session/tests.rs b/codex-rs/core/src/session/tests.rs index 2089299b59..aa828752a3 100644 --- a/codex-rs/core/src/session/tests.rs +++ b/codex-rs/core/src/session/tests.rs @@ -6635,7 +6635,6 @@ pub(crate) async fn make_session_and_context() -> (Session, TurnContext) { models_manager: Arc::clone(&models_manager), git_root_discovery: Arc::default(), tool_approvals: Mutex::new(ApprovalStore::default()), - guardian_rejection_circuit_breaker: Mutex::new(Default::default()), runtime_handle: tokio::runtime::Handle::current(), skills_service, agents_md_manager: Arc::new(AgentsMdManager::new(/*user_instructions*/ None)), @@ -8809,7 +8808,6 @@ where models_manager: Arc::clone(&models_manager), git_root_discovery: Arc::default(), tool_approvals: Mutex::new(ApprovalStore::default()), - guardian_rejection_circuit_breaker: Mutex::new(Default::default()), runtime_handle: tokio::runtime::Handle::current(), skills_service, agents_md_manager: Arc::new(AgentsMdManager::new(/*user_instructions*/ None)), diff --git a/codex-rs/core/src/session/tests/guardian_tests.rs b/codex-rs/core/src/session/tests/guardian_tests.rs index d51009d986..5a4c79a98a 100644 --- a/codex-rs/core/src/session/tests/guardian_tests.rs +++ b/codex-rs/core/src/session/tests/guardian_tests.rs @@ -1324,3 +1324,48 @@ async fn guardian_subagent_does_not_inherit_parent_exec_policy_rules() { ); drop(io); } + +#[test_case(TerminalEventKind::TurnComplete; "completion")] +#[test_case(TerminalEventKind::TurnAborted; "interruption")] +#[tokio::test] +async fn terminal_turn_clears_extension_owned_denials(terminal: TerminalEventKind) { + let (session, turn, events) = make_session_and_context_with_rx().await; + let finish = Arc::new(tokio::sync::Notify::new()); + session + .spawn_task( + Arc::clone(&turn), + Vec::new(), + HeldStepTask { + kind: TaskKind::Regular, + finish: Arc::clone(&finish), + }, + ) + .await; + let denials = + codex_guardian_reviewer::ReviewDenials::for_thread(&session.services.thread_extension_data); + for _ in 0..2 { + assert_eq!( + denials.record_denial(&turn.sub_id, turn.model_info()).await, + None + ); + } + match terminal { + TerminalEventKind::TurnComplete => finish.notify_one(), + TerminalEventKind::TurnAborted => { + session.abort_all_tasks(TurnAbortReason::Interrupted).await + } + } + recv_terminal_event(&events, terminal).await; + // Delivery precedes accounting cleanup. Wait for the runtime to finish the turn. + timeout(Duration::from_secs(/*secs*/ 5), async { + while session.active_turn.lock().await.is_some() { + tokio::task::yield_now().await; + } + }) + .await + .expect("turn becomes idle"); + assert_eq!( + denials.record_denial(&turn.sub_id, turn.model_info()).await, + None + ); +} diff --git a/codex-rs/core/src/state/service.rs b/codex-rs/core/src/state/service.rs index 6524bbfbb3..6248b2c977 100644 --- a/codex-rs/core/src/state/service.rs +++ b/codex-rs/core/src/state/service.rs @@ -11,7 +11,6 @@ use crate::current_time::TimeProvider; use crate::elicitation::ElicitationService; use crate::environment_selection::ThreadEnvironments; use crate::exec_policy::ExecPolicyManager; -use crate::guardian::GuardianRejectionCircuitBreaker; use crate::mcp::McpManager; use crate::mcp_tool_exposure::McpHandlerCache; use crate::tools::ExecutedToolCalls; @@ -68,7 +67,6 @@ pub(crate) struct SessionServices { pub(crate) git_root_discovery: Arc, pub(crate) session_telemetry: SessionTelemetry, pub(crate) tool_approvals: Mutex, - pub(crate) guardian_rejection_circuit_breaker: Mutex, pub(crate) runtime_handle: Handle, pub(crate) skills_service: Arc, pub(crate) agents_md_manager: Arc, diff --git a/codex-rs/core/src/tasks/mod.rs b/codex-rs/core/src/tasks/mod.rs index 2a6de503aa..8ac1030269 100644 --- a/codex-rs/core/src/tasks/mod.rs +++ b/codex-rs/core/src/tasks/mod.rs @@ -306,11 +306,11 @@ impl Session { let cancellation_token = CancellationToken::new(); let done = Arc::new(Notify::new()); - self.services - .guardian_rejection_circuit_breaker - .lock() - .await - .clear_turn(&turn_context.sub_id); + codex_guardian_reviewer::ReviewDenials::clear_turn( + &self.services.thread_extension_data, + &turn_context.sub_id, + ) + .await; let (pending_items, _) = self.input_queue.drain_mailbox_input_items().await; let turn_state = { @@ -835,11 +835,11 @@ impl Session { }) }; self.send_event(turn_context.as_ref(), event).await; - self.services - .guardian_rejection_circuit_breaker - .lock() - .await - .clear_turn(&turn_context.sub_id); + codex_guardian_reviewer::ReviewDenials::clear_turn( + &self.services.thread_extension_data, + &turn_context.sub_id, + ) + .await; let cleared_active_turn = { let mut active = self.active_turn.lock().await; @@ -991,11 +991,11 @@ impl Session { duration_ms, }); self.send_event(task.turn_context.as_ref(), event).await; - self.services - .guardian_rejection_circuit_breaker - .lock() - .await - .clear_turn(&task.turn_context.sub_id); + codex_guardian_reviewer::ReviewDenials::clear_turn( + &self.services.thread_extension_data, + &task.turn_context.sub_id, + ) + .await; // Regular items were flushed before this terminal event was appended; buffering // thread writers may not flush it without another explicit barrier. if let Err(err) = self.flush_rollout().await { diff --git a/codex-rs/ext/guardian-reviewer/Cargo.toml b/codex-rs/ext/guardian-reviewer/Cargo.toml index 9aac21a4c5..8e24a8e218 100644 --- a/codex-rs/ext/guardian-reviewer/Cargo.toml +++ b/codex-rs/ext/guardian-reviewer/Cargo.toml @@ -17,12 +17,19 @@ anyhow = { workspace = true } codex-analytics = { workspace = true } codex-extension-api = { workspace = true } codex-features = { workspace = true } +codex-feedback = { workspace = true } +codex-mcp = { workspace = true } +codex-otel = { workspace = true } codex-protocol = { workspace = true } rand = { workspace = true } serde = { workspace = true } serde_json = { workspace = true } tokio = { workspace = true } tokio-util = { workspace = true } +tracing = { workspace = true } [dev-dependencies] +opentelemetry = { workspace = true } +opentelemetry_sdk = { workspace = true, features = ["experimental_metrics_custom_reader", "metrics"] } +test-case = "3.3.1" pretty_assertions = { workspace = true } diff --git a/codex-rs/ext/guardian-reviewer/src/feedback.rs b/codex-rs/ext/guardian-reviewer/src/feedback.rs new file mode 100644 index 0000000000..7a9c3a20a3 --- /dev/null +++ b/codex-rs/ext/guardian-reviewer/src/feedback.rs @@ -0,0 +1,141 @@ +//! Selects failed reviews for opt-in feedback and bounds each serialized record. +//! The host supplies captured context; records outlive short-lived reviewer sessions. + +use crate::GuardianReviewSessionOutcome; +use crate::parse_guardian_assessment; +use codex_feedback::record_guardian_review_failure; +use codex_protocol::ThreadId; +use codex_protocol::models::ResponseItem; +use codex_protocol::protocol::GuardianAssessmentOutcome; +use serde::Serialize; +use std::io; +use std::io::Write; + +const MAX_RECORD_BYTES: usize = 4 * 1024 * 1024; + +#[derive(Serialize)] +struct ReviewFeedbackRecord<'a> { + reviewed_thread_id: ThreadId, + reviewed_turn_id: &'a str, + target_item_id: Option<&'a str>, + reviewer_thread_id: ThreadId, + model: &'a str, + status: &'a str, + decision: Option<&'a str>, + action: &'a str, + action_truncated: bool, + instructions: Option<&'a str>, + history: Vec<&'a ResponseItem>, + context_omitted: bool, +} + +/// Context captured by the host only after feedback policy selects a failed review. +pub struct ReviewFeedbackContext<'a> { + pub reviewed_thread_id: ThreadId, + pub reviewed_turn_id: &'a str, + pub target_item_id: Option<&'a str>, + pub reviewer_thread_id: ThreadId, + pub model: &'a str, + pub action: &'a str, + pub action_truncated: bool, + pub instructions: Option<&'a str>, + pub history: Vec<&'a ResponseItem>, +} + +/// Host-supplied opt-in and persistence facts for the reviewed session. +pub struct ReviewFeedbackSettings { + pub enabled: bool, + pub ephemeral: bool, +} + +pub struct FailedReviewFeedback<'a> { + status: &'static str, + decision: Option<&'a str>, +} + +impl<'a> FailedReviewFeedback<'a> { + pub fn for_outcome( + outcome: &'a GuardianReviewSessionOutcome, + settings: ReviewFeedbackSettings, + ) -> Option { + if !settings.enabled || settings.ephemeral { + return None; + } + let (status, decision) = match outcome { + GuardianReviewSessionOutcome::Completed(Ok(Some(decision))) => { + match parse_guardian_assessment(Some(decision)) { + Ok(assessment) if assessment.outcome == GuardianAssessmentOutcome::Allow => { + return None; + } + Ok(_) => ("denied", Some(decision.as_str())), + Err(_) => ("invalid_decision", Some(decision.as_str())), + } + } + GuardianReviewSessionOutcome::Completed(Ok(None)) => ("missing_decision", None), + GuardianReviewSessionOutcome::Completed(Err(_)) + | GuardianReviewSessionOutcome::PromptBuildFailed(_) + | GuardianReviewSessionOutcome::SessionFailed { .. } => ("failed", None), + GuardianReviewSessionOutcome::TimedOut => ("timed_out", None), + GuardianReviewSessionOutcome::Aborted => ("aborted", None), + }; + Some(Self { status, decision }) + } + + pub fn store(self, context: ReviewFeedbackContext<'_>) { + ReviewFeedbackRecord { + reviewed_thread_id: context.reviewed_thread_id, + reviewed_turn_id: context.reviewed_turn_id, + target_item_id: context.target_item_id, + reviewer_thread_id: context.reviewer_thread_id, + model: context.model, + action: context.action, + action_truncated: context.action_truncated, + instructions: context.instructions, + history: context.history, + status: self.status, + decision: self.decision, + context_omitted: false, + } + .store(); + } +} + +impl ReviewFeedbackRecord<'_> { + fn store(mut self) { + let mut buffer = BoundedBuffer(Vec::new()); + if serde_json::to_writer(&mut buffer, &self).is_err() { + // Preserve the action and decision even when optional reviewer history is too large. + self.instructions = None; + self.history.clear(); + self.context_omitted = true; + buffer = BoundedBuffer(Vec::new()); + if serde_json::to_writer(&mut buffer, &self).is_err() { + tracing::warn!("Guardian feedback action and decision exceed the record limit"); + return; + } + } + record_guardian_review_failure(self.reviewed_thread_id, buffer.0); + } +} + +struct BoundedBuffer(Vec); + +impl Write for BoundedBuffer { + fn write(&mut self, bytes: &[u8]) -> io::Result { + if bytes.len() > MAX_RECORD_BYTES.saturating_sub(self.0.len()) { + return Err(io::Error::other( + "Guardian feedback record exceeds its size limit", + )); + } + self.0.extend_from_slice(bytes); + Ok(bytes.len()) + } + + fn flush(&mut self) -> io::Result<()> { + Ok(()) + } +} + +#[cfg(test)] +#[path = "feedback_tests.rs"] +mod tests; diff --git a/codex-rs/core/src/guardian/feedback_tests.rs b/codex-rs/ext/guardian-reviewer/src/feedback_tests.rs similarity index 79% rename from codex-rs/core/src/guardian/feedback_tests.rs rename to codex-rs/ext/guardian-reviewer/src/feedback_tests.rs index 3b256e3ad8..c6f46bef43 100644 --- a/codex-rs/core/src/guardian/feedback_tests.rs +++ b/codex-rs/ext/guardian-reviewer/src/feedback_tests.rs @@ -8,21 +8,26 @@ fn oversized_context_keeps_the_action_and_decision() -> anyhow::Result<()> { let reviewer_thread_id = ThreadId::new(); let decision = r#"{"outcome":"deny","rationale":"Missing approval."}"#; let action = r#"{"command":"git push"}"#; - ReviewFeedbackRecord { + let outcome = GuardianReviewSessionOutcome::Completed(Ok(Some(decision.to_owned()))); + let feedback = FailedReviewFeedback::for_outcome( + &outcome, + ReviewFeedbackSettings { + enabled: true, + ephemeral: false, + }, + ) + .expect("denials retain feedback"); + feedback.store(ReviewFeedbackContext { reviewed_thread_id: thread_id, reviewed_turn_id: "parent-turn", target_item_id: Some("push-call"), reviewer_thread_id, model: "review-model", - status: "denied", - decision: Some(decision), action, action_truncated: false, instructions: Some(&"x".repeat(MAX_RECORD_BYTES)), history: Vec::new(), - context_omitted: false, - } - .store(); + }); let contents = codex_feedback::guardian_review_failures(&[thread_id]) .attachment .expect("failed-review record") diff --git a/codex-rs/ext/guardian-reviewer/src/lib.rs b/codex-rs/ext/guardian-reviewer/src/lib.rs index e12467ee5d..3c5d7f31da 100644 --- a/codex-rs/ext/guardian-reviewer/src/lib.rs +++ b/codex-rs/ext/guardian-reviewer/src/lib.rs @@ -6,9 +6,12 @@ mod circuit_breaker; mod completion; mod deadline; mod execution; +mod feedback; +mod metrics; mod model; mod outcome; mod pool; +mod reporting; mod retry; mod review; mod settings; @@ -53,3 +56,10 @@ pub use execution::wait_for_guardian_review; pub use settings::ReviewerConfigOverrides; pub use settings::ReviewerTurn; pub use settings::reviewer_config_overrides; + +pub use feedback::FailedReviewFeedback; +pub use feedback::ReviewFeedbackContext; +pub use feedback::ReviewFeedbackSettings; +pub use reporting::ReviewDenials; +pub use reporting::ReviewMetadata; +pub use reporting::ReviewReport; diff --git a/codex-rs/core/src/guardian/metrics.rs b/codex-rs/ext/guardian-reviewer/src/metrics.rs similarity index 99% rename from codex-rs/core/src/guardian/metrics.rs rename to codex-rs/ext/guardian-reviewer/src/metrics.rs index c60c7d961b..9ec547fdc7 100644 --- a/codex-rs/core/src/guardian/metrics.rs +++ b/codex-rs/ext/guardian-reviewer/src/metrics.rs @@ -1,3 +1,5 @@ +//! Emits Guardian metrics with bounded tags and the existing telemetry schema. + use std::time::Duration; use codex_analytics::GuardianApprovalRequestSource; diff --git a/codex-rs/ext/guardian-reviewer/src/reporting.rs b/codex-rs/ext/guardian-reviewer/src/reporting.rs new file mode 100644 index 0000000000..c2e2815e87 --- /dev/null +++ b/codex-rs/ext/guardian-reviewer/src/reporting.rs @@ -0,0 +1,161 @@ +//! Owns review reporting and the per-thread denial window. The host publishes +//! events and enforces interruption only on the still-active owning turn. + +use std::sync::Arc; + +use codex_analytics::AnalyticsEventsClient; +use codex_analytics::GuardianApprovalRequestSource; +use codex_analytics::GuardianReviewAnalyticsResult; +use codex_analytics::GuardianReviewTrackContext; +use codex_analytics::GuardianReviewedAction; +use codex_extension_api::ExtensionData; +use codex_otel::SessionTelemetry; +use codex_protocol::approvals::GuardianAssessmentAction; +use codex_protocol::approvals::GuardianReviewReason; +use codex_protocol::openai_models::ModelInfo; +use codex_protocol::protocol::GuardianAssessmentDecisionSource; +use codex_protocol::protocol::GuardianAssessmentEvent; +use codex_protocol::protocol::GuardianAssessmentStatus; +use tokio::sync::Mutex; + +use crate::GuardianRejectionCircuitBreaker; +use crate::GuardianRejectionCircuitBreakerAction; +use crate::GuardianRejectionCircuitBreakerPolicy; +use crate::GuardianReviewOutcome; +use crate::ReviewCompletion; + +/// Captured action attribution used for both the started and terminal report. +pub struct ReviewMetadata { + pub thread_id: String, + pub turn_id: String, + pub review_id: String, + pub target_item_id: Option, + pub plugin_id: Option, + pub script_path: Option, + pub approval_request_source: GuardianApprovalRequestSource, + pub reviewed_action: GuardianReviewedAction, + pub action: GuardianAssessmentAction, + pub review_reason: GuardianReviewReason, +} + +pub struct ReviewReport { + started: GuardianAssessmentEvent, + tracking: GuardianReviewTrackContext, + approval_request_source: GuardianApprovalRequestSource, + reviewed_action: GuardianReviewedAction, +} + +impl ReviewReport { + pub fn new(metadata: ReviewMetadata) -> Self { + let tracking = GuardianReviewTrackContext::new( + metadata.thread_id, + metadata.turn_id.clone(), + metadata.review_id.clone(), + metadata.target_item_id.clone(), + metadata.approval_request_source, + metadata.reviewed_action.clone(), + crate::REVIEW_TIMEOUT.as_millis() as u64, + ); + let started = GuardianAssessmentEvent { + review_reason: Some(metadata.review_reason), + id: metadata.review_id, + target_item_id: metadata.target_item_id, + plugin_id: metadata.plugin_id, + script_path: metadata.script_path, + turn_id: metadata.turn_id, + started_at_ms: tracking.started_at_ms.try_into().unwrap_or_default(), + completed_at_ms: None, + status: GuardianAssessmentStatus::InProgress, + risk_level: None, + user_authorization: None, + rationale: None, + decision_source: None, + action: metadata.action, + }; + Self { + started, + tracking, + approval_request_source: metadata.approval_request_source, + reviewed_action: metadata.reviewed_action, + } + } + + pub fn turn_id(&self) -> &str { + &self.started.turn_id + } + + pub fn started_event(&self) -> GuardianAssessmentEvent { + self.started.clone() + } + + pub fn complete( + &self, + outcome: GuardianReviewOutcome, + model: &ModelInfo, + analytics: GuardianReviewAnalyticsResult, + completed_at_ms: i64, + ) -> ReviewCompletion { + let mut event = self.started.clone(); + event.completed_at_ms = Some(completed_at_ms); + event.decision_source = Some(GuardianAssessmentDecisionSource::Agent); + crate::complete_review(outcome, model, event, analytics) + } + + pub fn track( + &self, + telemetry: &SessionTelemetry, + analytics: &AnalyticsEventsClient, + result: GuardianReviewAnalyticsResult, + completed_at_ms: u64, + ) { + crate::metrics::emit_guardian_review_metrics( + telemetry, + &result, + self.approval_request_source, + &self.reviewed_action, + completed_at_ms.saturating_sub(self.tracking.started_at_ms), + ); + analytics.track_guardian_review(&self.tracking, result, completed_at_ms); + } +} + +/// Thread-scoped accounting shared by synchronous and cached approvals. +#[derive(Default)] +pub struct ReviewDenials(Mutex); + +impl ReviewDenials { + pub fn for_thread(store: &ExtensionData) -> Arc { + store.get_or_init(Self::default) + } + + pub async fn clear_turn(store: &ExtensionData, turn_id: &str) { + if let Some(state) = store.get::() { + state.0.lock().await.clear_turn(turn_id); + } + } + + pub async fn record_non_denial(&self, turn_id: &str) { + self.0.lock().await.record_non_denial(turn_id); + } + + /// Returns a warning exactly once when this turn reaches its denial limit. + pub async fn record_denial(&self, turn_id: &str, model: &ModelInfo) -> Option { + let action = self + .0 + .lock() + .await + .record_denial(turn_id, GuardianRejectionCircuitBreakerPolicy::from(model)); + match action { + GuardianRejectionCircuitBreakerAction::Continue => None, + GuardianRejectionCircuitBreakerAction::InterruptTurn { + consecutive_denials, + recent_denials, + } => { + let window = crate::AUTO_REVIEW_DENIAL_WINDOW_SIZE; + Some(format!( + "Automatic approval review rejected too many approval requests for this turn ({consecutive_denials} consecutive, {recent_denials} in the last {window} reviews); interrupting the turn." + )) + } + } + } +}