diff --git a/codex-rs/core/src/codex.rs b/codex-rs/core/src/codex.rs index 1d08a6c9e6..ab5366aad5 100644 --- a/codex-rs/core/src/codex.rs +++ b/codex-rs/core/src/codex.rs @@ -109,6 +109,7 @@ use crate::shell; use crate::state::ActiveTurn; use crate::state::SessionServices; use crate::state::SessionState; +use crate::status::ComponentHealth; use crate::status::IdleWarning; use crate::tasks::GhostSnapshotTask; use crate::tasks::ReviewTask; @@ -455,6 +456,16 @@ impl Session { } } + pub(crate) async fn replace_codex_backend_status( + &self, + status: ComponentHealth, + ) -> Option { + let mut guard = self.services.codex_backend_status.lock().await; + let previous = *guard; + *guard = Some(status); + previous + } + async fn new( session_configuration: SessionConfiguration, config: Arc, @@ -571,6 +582,7 @@ impl Session { auth_manager: Arc::clone(&auth_manager), otel_event_manager, tool_approvals: Mutex::new(ApprovalStore::default()), + codex_backend_status: Mutex::new(None), }; let sess = Arc::new(Session { @@ -2217,13 +2229,17 @@ async fn try_run_turn( biased; result = stream.next().or_cancel(&cancellation_token) => result, _ = sleep_until(idle_warning.deadline()) => { - if let Some(message) = idle_warning.maybe_warning_message().await { + if let Some(message) = idle_warning + .maybe_warning_message(sess.as_ref()) + .await + { sess.send_event( &turn_context, EventMsg::Warning(WarningEvent { message }), ) .await; } + idle_warning.mark_event(); continue; } }; @@ -2445,13 +2461,17 @@ where biased; result = &mut stream_future => return result?, _ = sleep_until(idle_warning.deadline()) => { - if let Some(message) = idle_warning.maybe_warning_message().await { + if let Some(message) = idle_warning + .maybe_warning_message(sess.as_ref()) + .await + { sess.send_event( turn_context, EventMsg::Warning(WarningEvent { message }), ) .await; } + idle_warning.mark_event(); } } } @@ -2750,6 +2770,7 @@ mod tests { auth_manager: Arc::clone(&auth_manager), otel_event_manager: otel_event_manager.clone(), tool_approvals: Mutex::new(ApprovalStore::default()), + codex_backend_status: Mutex::new(None), }; let turn_context = Session::make_turn_context( @@ -2828,6 +2849,7 @@ mod tests { auth_manager: Arc::clone(&auth_manager), otel_event_manager: otel_event_manager.clone(), tool_approvals: Mutex::new(ApprovalStore::default()), + codex_backend_status: Mutex::new(None), }; let turn_context = Arc::new(Session::make_turn_context( diff --git a/codex-rs/core/src/state/service.rs b/codex-rs/core/src/state/service.rs index 287fb73d25..9c3d271f9d 100644 --- a/codex-rs/core/src/state/service.rs +++ b/codex-rs/core/src/state/service.rs @@ -3,6 +3,7 @@ use std::sync::Arc; use crate::AuthManager; use crate::RolloutRecorder; use crate::mcp_connection_manager::McpConnectionManager; +use crate::status::ComponentHealth; use crate::tools::sandboxing::ApprovalStore; use crate::unified_exec::UnifiedExecSessionManager; use crate::user_notification::UserNotifier; @@ -22,4 +23,5 @@ pub(crate) struct SessionServices { pub(crate) auth_manager: Arc, pub(crate) otel_event_manager: OtelEventManager, pub(crate) tool_approvals: Mutex, + pub(crate) codex_backend_status: Mutex>, } diff --git a/codex-rs/core/src/status.rs b/codex-rs/core/src/status.rs index aa307890de..98de8fd79c 100644 --- a/codex-rs/core/src/status.rs +++ b/codex-rs/core/src/status.rs @@ -1,6 +1,7 @@ use std::sync::OnceLock; use std::time::Duration; +use crate::codex::Session; use anyhow::Context; use anyhow::Result; use anyhow::anyhow; @@ -8,6 +9,9 @@ use anyhow::bail; use codex_client::HttpTransport; use codex_client::Request; use codex_client::ReqwestTransport; +use codex_client::RetryOn; +use codex_client::RetryPolicy; +use codex_client::run_with_retry; use http::header::CONTENT_TYPE; use reqwest::Method; use serde::Deserialize; @@ -53,7 +57,6 @@ impl ComponentHealth { pub(crate) struct IdleWarning { last_event: Instant, idle_timeout: Duration, - warning_sent: bool, } impl IdleWarning { @@ -61,7 +64,6 @@ impl IdleWarning { Self { last_event: Instant::now(), idle_timeout, - warning_sent: false, } } @@ -73,22 +75,20 @@ impl IdleWarning { self.last_event = Instant::now(); } - pub(crate) async fn maybe_warning_message(&mut self) -> Option { - if self.warning_sent { + pub(crate) async fn maybe_warning_message(&mut self, session: &Session) -> Option { + let Ok(status) = fetch_codex_health().await else { + return None; + }; + + let previous = session.replace_codex_backend_status(status).await; + if status.is_operational() || previous == Some(status) { return None; } - if let Ok(status) = fetch_codex_health().await - && !status.is_operational() - { - self.warning_sent = true; - self.mark_event(); - return Some(format!( - "Codex is experiencing a {status}. If a response stalls, try again later. You can follow incident updates at status.openai.com." - )); - } - - None + self.mark_event(); + Some(format!( + "Codex is experiencing a {status}. If a response stalls, try again later. You can follow incident updates at status.openai.com." + )) } } @@ -107,10 +107,27 @@ async fn fetch_codex_health() -> Result { .build() .context("building HTTP client")?; - let response = ReqwestTransport::new(client) - .execute(Request::new(Method::GET, status_widget_url.clone())) - .await - .context("requesting status widget")?; + let transport = ReqwestTransport::new(client); + let policy = RetryPolicy { + max_attempts: 2, + base_delay: Duration::from_millis(200), + retry_on: RetryOn { + retry_429: true, + retry_5xx: true, + retry_transport: true, + }, + }; + + let response = run_with_retry( + policy, + || Request::new(Method::GET, status_widget_url.clone()), + |req, _attempt| { + let transport = transport.clone(); + async move { transport.execute(req).await } + }, + ) + .await + .context("requesting status widget")?; let content_type = response .headers diff --git a/codex-rs/core/tests/suite/idle_warning.rs b/codex-rs/core/tests/suite/idle_warning.rs index 65c46f1aa6..d4c88fbe48 100644 --- a/codex-rs/core/tests/suite/idle_warning.rs +++ b/codex-rs/core/tests/suite/idle_warning.rs @@ -1,5 +1,8 @@ #![allow(clippy::unwrap_used, clippy::expect_used)] +use std::sync::Arc; +use std::sync::atomic::AtomicUsize; +use std::sync::atomic::Ordering; use std::time::Duration; use codex_core::protocol::EventMsg; @@ -18,6 +21,8 @@ use core_test_support::test_codex::test_codex; use core_test_support::wait_for_event; use pretty_assertions::assert_eq; use wiremock::Mock; +use wiremock::Request; +use wiremock::Respond; use wiremock::ResponseTemplate; use wiremock::matchers::method; use wiremock::matchers::path; @@ -29,7 +34,7 @@ async fn emits_warning_when_stream_is_idle_and_status_is_degraded() { Mock::given(method("GET")) .and(path(status_path)) - .respond_with(status_payload()) + .respond_with(status_payload("major_outage")) .mount(&status_server) .await; @@ -82,7 +87,76 @@ async fn emits_warning_when_stream_is_idle_and_status_is_degraded() { wait_for_event(&codex, |event| matches!(event, EventMsg::TaskComplete(_))).await; } -fn status_payload() -> ResponseTemplate { +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn warns_once_per_status_change_only_when_unhealthy() { + let status_server = start_mock_server().await; + let status_path = "/proxy/status.openai.com"; + + let responder = SequenceResponder::new(vec!["major_outage", "partial_outage"]); + + Mock::given(method("GET")) + .and(path(status_path)) + .respond_with(responder) + .mount(&status_server) + .await; + + set_test_status_widget_url(format!("{}{}", status_server.uri(), status_path)); + set_test_idle_timeout(Duration::from_millis(100)); + + let responses_server = start_mock_server().await; + let stalled_response = sse(vec![ + ev_response_created("resp-1"), + ev_assistant_message("msg-1", "finally"), + ev_completed("resp-1"), + ]); + + let _responses_mock = mount_sse_once_with_delay( + &responses_server, + stalled_response, + Duration::from_millis(2000), + ) + .await; + + let test_codex = test_codex().build(&responses_server).await.unwrap(); + let codex = test_codex.codex; + + codex + .submit(Op::UserInput { + items: vec![UserInput::Text { text: "hi".into() }], + }) + .await + .unwrap(); + + let mut warnings = Vec::new(); + loop { + let event = codex.next_event().await.expect("event"); + match event.msg { + EventMsg::Warning(WarningEvent { message }) => warnings.push(message), + EventMsg::TaskComplete(_) => break, + _ => {} + } + } + + let expected_messages = vec![ + "Codex is experiencing a major outage. If a response stalls, try again later. You can follow incident updates at status.openai.com.".to_string(), + "Codex is experiencing a partial outage. If a response stalls, try again later. You can follow incident updates at status.openai.com.".to_string(), + ]; + + assert!( + !warnings.is_empty(), + "expected at least one warning for non-operational status" + ); + assert!( + warnings.len() <= expected_messages.len(), + "unexpected extra warnings: {warnings:?}" + ); + assert_eq!(warnings[0], expected_messages[0], "first warning mismatch"); + if warnings.len() > 1 { + assert_eq!(warnings[1], expected_messages[1], "second warning mismatch"); + } +} + +fn status_payload(status: &str) -> ResponseTemplate { ResponseTemplate::new(200) .insert_header("content-type", "application/json") .set_body_json(serde_json::json!({ @@ -91,8 +165,37 @@ fn status_payload() -> ResponseTemplate { {"id": "cmp-1", "name": "Codex", "status_page_id": "page-1"} ], "affected_components": [ - {"component_id": "cmp-1", "status": "major_outage"} + {"component_id": "cmp-1", "status": status} ] } })) } + +#[derive(Clone)] +struct SequenceResponder { + statuses: Vec<&'static str>, + calls: Arc, +} + +impl SequenceResponder { + fn new(statuses: Vec<&'static str>) -> Self { + Self { + statuses, + calls: Arc::new(AtomicUsize::new(0)), + } + } +} + +impl Respond for SequenceResponder { + fn respond(&self, _request: &Request) -> ResponseTemplate { + let call = self.calls.fetch_add(1, Ordering::SeqCst); + let idx = usize::try_from(call).unwrap_or(0); + let status = self + .statuses + .get(idx) + .copied() + .or_else(|| self.statuses.last().copied()) + .unwrap_or("operational"); + status_payload(status) + } +}