From 098a47fd3627b1fc1b0741eed4d3098df984bd6a Mon Sep 17 00:00:00 2001 From: Katia Bazzi Date: Wed, 10 Jun 2026 21:04:24 -0700 Subject: [PATCH] Handle Guardian infrastructure failures separately --- codex-rs/core/src/guardian/review.rs | 116 ++++++++++++++++--- codex-rs/core/src/guardian/review_session.rs | 54 +++++++++ codex-rs/core/src/guardian/tests.rs | 82 +++++++++++++ codex-rs/protocol/src/error.rs | 3 + codex-rs/protocol/src/error_tests.rs | 21 ++++ 5 files changed, 259 insertions(+), 17 deletions(-) diff --git a/codex-rs/core/src/guardian/review.rs b/codex-rs/core/src/guardian/review.rs index 02b4b04138..02c7d2eac2 100644 --- a/codex-rs/core/src/guardian/review.rs +++ b/codex-rs/core/src/guardian/review.rs @@ -480,16 +480,14 @@ async fn run_guardian_review( record_guardian_non_denial(&session, &assessment_turn_id).await; return ReviewDecision::Abort; } - GuardianReviewError::PromptBuild { .. } - | GuardianReviewError::Session { .. } - | GuardianReviewError::Parse { .. } => { + GuardianReviewError::PromptBuild { .. } | GuardianReviewError::Parse { .. } => { let message = match &error { GuardianReviewError::PromptBuild { message } - | GuardianReviewError::Session { message, .. } | GuardianReviewError::Parse { message } => message, GuardianReviewError::Timeout | GuardianReviewError::Cancelled => { "guardian review failed" } + GuardianReviewError::Session { .. } => unreachable!(), }; let rationale = format!("Automatic approval review failed: {message}"); track_guardian_review( @@ -515,6 +513,78 @@ async fn run_guardian_review( false, ) } + GuardianReviewError::Session { + message, + error_info: Some(error_info), + } if is_transient_guardian_session_error_info(&error_info) => { + let rationale = format!("Automatic approval review failed: {message}"); + track_guardian_review( + session.as_ref(), + &review_tracking, + approval_request_source, + &reviewed_action, + GuardianReviewAnalyticsResult { + decision: GuardianReviewDecision::Denied, + terminal_status: GuardianReviewTerminalStatus::TimedOut, + failure_reason: Some(GuardianReviewFailureReason::SessionError), + ..analytics_result + }, + completed_at_ms.try_into().unwrap_or_default(), + ); + session + .send_event( + turn.as_ref(), + EventMsg::GuardianWarning(WarningEvent { + message: rationale.clone(), + }), + ) + .await; + session + .send_event( + turn.as_ref(), + EventMsg::GuardianAssessment(GuardianAssessmentEvent { + id: review_id, + target_item_id, + turn_id: assessment_turn_id.clone(), + started_at_ms, + completed_at_ms: Some(completed_at_ms), + status: GuardianAssessmentStatus::TimedOut, + risk_level: None, + user_authorization: None, + rationale: Some(rationale), + decision_source: Some(GuardianAssessmentDecisionSource::Agent), + action: terminal_action, + }), + ) + .await; + record_guardian_non_denial(&session, &assessment_turn_id).await; + return ReviewDecision::TimedOut; + } + GuardianReviewError::Session { message, .. } => { + let rationale = format!("Automatic approval review failed: {message}"); + track_guardian_review( + session.as_ref(), + &review_tracking, + approval_request_source, + &reviewed_action, + GuardianReviewAnalyticsResult { + decision: GuardianReviewDecision::Denied, + terminal_status: GuardianReviewTerminalStatus::FailedClosed, + failure_reason: Some(GuardianReviewFailureReason::SessionError), + ..analytics_result + }, + completed_at_ms.try_into().unwrap_or_default(), + ); + ( + GuardianAssessment { + risk_level: GuardianRiskLevel::High, + user_authorization: GuardianUserAuthorization::Unknown, + outcome: GuardianAssessmentOutcome::Deny, + rationale, + }, + false, + ) + } }, }; @@ -889,20 +959,32 @@ async fn wait_before_guardian_retry( } fn should_retry_guardian_review(outcome: &GuardianReviewOutcome) -> bool { + match outcome { + GuardianReviewOutcome::Error(GuardianReviewError::Session { + error_info: Some(error_info), + .. + }) => is_transient_guardian_session_error_info(error_info), + GuardianReviewOutcome::Error(GuardianReviewError::Parse { .. }) => true, + GuardianReviewOutcome::Completed(_) + | GuardianReviewOutcome::Error( + GuardianReviewError::PromptBuild { .. } + | GuardianReviewError::Session { + error_info: None, .. + } + | GuardianReviewError::Timeout + | GuardianReviewError::Cancelled, + ) => false, + } +} + +fn is_transient_guardian_session_error_info(error_info: &CodexErrorInfo) -> bool { matches!( - outcome, - GuardianReviewOutcome::Error( - GuardianReviewError::Session { - error_info: Some( - CodexErrorInfo::ServerOverloaded - | CodexErrorInfo::HttpConnectionFailed { .. } - | CodexErrorInfo::ResponseStreamConnectionFailed { .. } - | CodexErrorInfo::InternalServerError - | CodexErrorInfo::ResponseStreamDisconnected { .. } - ), - .. - } | GuardianReviewError::Parse { .. } - ) + error_info, + CodexErrorInfo::ServerOverloaded + | CodexErrorInfo::HttpConnectionFailed { .. } + | CodexErrorInfo::ResponseStreamConnectionFailed { .. } + | CodexErrorInfo::InternalServerError + | CodexErrorInfo::ResponseStreamDisconnected { .. } ) } diff --git a/codex-rs/core/src/guardian/review_session.rs b/codex-rs/core/src/guardian/review_session.rs index 54282c1c0a..a247b7db9f 100644 --- a/codex-rs/core/src/guardian/review_session.rs +++ b/codex-rs/core/src/guardian/review_session.rs @@ -1058,6 +1058,7 @@ async fn interrupt_and_drain_turn(codex: &Codex, expected_turn_id: &str) -> anyh #[cfg(test)] mod tests { use super::*; + use codex_protocol::error::CodexErr; use codex_protocol::protocol::AgentStatus; use codex_protocol::protocol::ErrorEvent; use codex_protocol::protocol::Submission; @@ -1627,6 +1628,59 @@ mod tests { assert!(capture_token_usage); } + #[tokio::test] + async fn wait_for_guardian_review_preserves_remote_compact_stream_disconnect() { + let (review_session, tx_event, _rx_sub) = test_review_session().await; + let stream_error = CodexErr::Stream( + "WebSocket protocol error: Connection reset without closing handshake".to_string(), + None, + ); + tx_event + .send(Event { + id: "current-turn".to_string(), + msg: EventMsg::Error( + stream_error + .to_error_event(Some("Error running remote compact task".to_string())), + ), + }) + .await + .expect("queue guardian error"); + tx_event + .send(turn_complete_event( + "current-turn", + /*last_agent_message*/ None, + Some(42), + )) + .await + .expect("queue current turn completion"); + + let mut analytics_result = GuardianReviewAnalyticsResult::without_session(); + let (outcome, keep_review_session, capture_token_usage) = wait_for_guardian_review( + &review_session, + "current-turn", + tokio::time::Instant::now() + Duration::from_secs(1), + /*external_cancel*/ None, + &mut analytics_result, + ) + .await; + + let GuardianReviewSessionOutcome::SessionFailed { error, error_info } = outcome else { + panic!("expected structured session failure"); + }; + assert_eq!( + error.to_string(), + "Error running remote compact task: stream disconnected before completion: WebSocket protocol error: Connection reset without closing handshake" + ); + assert_eq!( + error_info, + Some(CodexErrorInfo::ResponseStreamDisconnected { + http_status_code: None, + }) + ); + assert!(keep_review_session); + assert!(capture_token_usage); + } + #[tokio::test] async fn wait_for_guardian_review_ignores_prior_turn_aborts() { let (review_session, tx_event, _rx_sub) = test_review_session().await; diff --git a/codex-rs/core/src/guardian/tests.rs b/codex-rs/core/src/guardian/tests.rs index 8a1382c7bf..55a62381be 100644 --- a/codex-rs/core/src/guardian/tests.rs +++ b/codex-rs/core/src/guardian/tests.rs @@ -2192,6 +2192,88 @@ async fn guardian_review_retries_transient_session_failure_then_approves() -> an Ok(()) } +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn guardian_review_exhausted_model_capacity_failures_do_not_deny() -> anyhow::Result<()> { + skip_if_no_network!(Ok(())); + + let server = start_mock_server().await; + let selected_model_capacity = "Selected model is at capacity. Please try a different model."; + let request_log = mount_sse_sequence( + &server, + vec![ + sse_failed( + "resp-selected-model-capacity-1", + "server_is_overloaded", + selected_model_capacity, + ), + sse_failed( + "resp-selected-model-capacity-2", + "server_is_overloaded", + selected_model_capacity, + ), + sse_failed( + "resp-selected-model-capacity-3", + "server_is_overloaded", + selected_model_capacity, + ), + ], + ) + .await; + let (session, turn, rx) = guardian_test_session_turn_and_rx(&server).await; + seed_guardian_parent_history(&session, &turn).await; + + let decision = review_approval_request( + &session, + &turn, + "review-selected-model-capacity".to_string(), + guardian_shell_request("shell-selected-model-capacity"), + /*retry_reason*/ None, + ) + .await; + + assert_eq!(decision, ReviewDecision::TimedOut); + assert_eq!(request_log.requests().len(), 3); + let mut statuses = Vec::new(); + let mut warnings = Vec::new(); + while let Ok(event) = rx.try_recv() { + match event.msg { + EventMsg::GuardianAssessment(event) => statuses.push(event.status), + EventMsg::GuardianWarning(event) => warnings.push(event.message), + _ => {} + } + } + assert_eq!( + statuses, + vec![ + GuardianAssessmentStatus::InProgress, + GuardianAssessmentStatus::TimedOut, + ] + ); + assert!( + warnings + .iter() + .any(|message| message.contains(selected_model_capacity)), + "warning should preserve the capacity failure" + ); + assert!( + warnings + .iter() + .all(|message| !message.contains("denied (risk:")), + "exhausted infrastructure failures should not surface as Guardian denials" + ); + assert!( + session + .services + .guardian_rejections + .lock() + .await + .get("review-selected-model-capacity") + .is_none(), + "selected model capacity failures must not be stored as a policy denial" + ); + Ok(()) +} + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn guardian_review_does_not_retry_missing_assessment_payload() -> anyhow::Result<()> { skip_if_no_network!(Ok(())); diff --git a/codex-rs/protocol/src/error.rs b/codex-rs/protocol/src/error.rs index d7e953af0b..2dbe2a532f 100644 --- a/codex-rs/protocol/src/error.rs +++ b/codex-rs/protocol/src/error.rs @@ -228,6 +228,9 @@ impl CodexErr { CodexErr::RetryLimit(_) => CodexErrorInfo::ResponseTooManyFailedAttempts { http_status_code: self.http_status_code_value(), }, + CodexErr::Stream(..) => CodexErrorInfo::ResponseStreamDisconnected { + http_status_code: self.http_status_code_value(), + }, CodexErr::ConnectionFailed(_) => CodexErrorInfo::HttpConnectionFailed { http_status_code: self.http_status_code_value(), }, diff --git a/codex-rs/protocol/src/error_tests.rs b/codex-rs/protocol/src/error_tests.rs index bd7a5f0058..f234d6336f 100644 --- a/codex-rs/protocol/src/error_tests.rs +++ b/codex-rs/protocol/src/error_tests.rs @@ -190,6 +190,27 @@ fn to_error_event_handles_response_stream_failed() { ); } +#[test] +fn to_error_event_handles_response_stream_disconnected() { + let err = CodexErr::Stream( + "WebSocket protocol error: Connection reset without closing handshake".to_string(), + None, + ); + + let event = err.to_error_event(Some("Error running remote compact task".to_string())); + + assert_eq!( + event.message, + "Error running remote compact task: stream disconnected before completion: WebSocket protocol error: Connection reset without closing handshake" + ); + assert_eq!( + event.codex_error_info, + Some(CodexErrorInfo::ResponseStreamDisconnected { + http_status_code: None, + }) + ); +} + #[test] fn sandbox_denied_reports_exit_code_when_no_output_available() { let output = ExecToolCallOutput {