diff --git a/codex-rs/core/src/codex.rs b/codex-rs/core/src/codex.rs index b43e4df81d..62083bcb10 100644 --- a/codex-rs/core/src/codex.rs +++ b/codex-rs/core/src/codex.rs @@ -2182,14 +2182,28 @@ async fn try_run_turn( }); sess.persist_rollout_items(&[rollout_item]).await; - let mut stream = turn_context - .client - .clone() - .stream(prompt) - .or_cancel(&cancellation_token) - .await??; - let mut idle_warning = IdleWarning::default(); + let client = turn_context.client.clone(); + let mut stream_future = Box::pin(client.stream(prompt).or_cancel(&cancellation_token)); + + let mut stream = loop { + tokio::select! { + biased; + result = &mut stream_future => break result??, + _ = sleep_until(idle_warning.deadline()) => { + if let Some(message) = idle_warning.maybe_warning_message().await { + sess.send_event( + &turn_context, + EventMsg::Warning(WarningEvent { message }), + ) + .await; + } + continue; + } + } + }; + + idle_warning.mark_event(); let tool_runtime = ToolCallRuntime::new( Arc::clone(&router), diff --git a/codex-rs/core/src/status.rs b/codex-rs/core/src/status.rs index ec73d2e136..e4df839189 100644 --- a/codex-rs/core/src/status.rs +++ b/codex-rs/core/src/status.rs @@ -1,3 +1,4 @@ +use std::sync::OnceLock; use std::time::Duration; use anyhow::Context; @@ -18,6 +19,9 @@ const STATUS_WIDGET_URL: &str = "https://status.openai.com/proxy/status.openai.c const CODEX_COMPONENT_NAME: &str = "Codex"; const DEFAULT_IDLE_TIMEOUT: Duration = Duration::from_secs(30); +static TEST_STATUS_WIDGET_URL: OnceLock = OnceLock::new(); +static TEST_IDLE_TIMEOUT: OnceLock = OnceLock::new(); + #[derive(Debug, Clone, Display, Copy, PartialEq, Eq, Serialize, Deserialize)] #[serde(rename_all = "snake_case")] pub(crate) enum ComponentHealth { @@ -83,11 +87,13 @@ impl IdleWarning { impl Default for IdleWarning { fn default() -> Self { - Self::new(DEFAULT_IDLE_TIMEOUT) + Self::new(idle_timeout()) } } async fn fetch_codex_health() -> Result { + let status_widget_url = status_widget_url(); + let client = reqwest::Client::builder() .connect_timeout(Duration::from_secs(5)) .timeout(Duration::from_secs(10)) @@ -95,7 +101,7 @@ async fn fetch_codex_health() -> Result { .context("building HTTP client")?; let response = ReqwestTransport::new(client) - .execute(Request::new(Method::GET, STATUS_WIDGET_URL.to_string())) + .execute(Request::new(Method::GET, status_widget_url.clone())) .await .context("requesting status widget")?; @@ -113,7 +119,7 @@ async fn fetch_codex_health() -> Result { .collect::(); bail!( - "Expected JSON from {STATUS_WIDGET_URL}: Content-Type={content_type}. Body starts with: {snippet:?}" + "Expected JSON from {status_widget_url}: Content-Type={content_type}. Body starts with: {snippet:?}" ); } @@ -172,6 +178,32 @@ fn derive_component_health( Ok(status) } +fn idle_timeout() -> Duration { + TEST_IDLE_TIMEOUT + .get() + .copied() + .unwrap_or(DEFAULT_IDLE_TIMEOUT) +} + +fn status_widget_url() -> String { + TEST_STATUS_WIDGET_URL + .get() + .cloned() + .unwrap_or_else(|| STATUS_WIDGET_URL.to_string()) +} + +#[doc(hidden)] +#[cfg_attr(not(test), allow(dead_code))] +pub fn set_test_status_widget_url(url: impl Into) { + let _ = TEST_STATUS_WIDGET_URL.set(url.into()); +} + +#[doc(hidden)] +#[cfg_attr(not(test), allow(dead_code))] +pub fn set_test_idle_timeout(duration: Duration) { + let _ = TEST_IDLE_TIMEOUT.set(duration); +} + #[cfg(test)] mod tests { use super::*; diff --git a/codex-rs/core/tests/common/responses.rs b/codex-rs/core/tests/common/responses.rs index 7cc59f26ef..edd17553dc 100644 --- a/codex-rs/core/tests/common/responses.rs +++ b/codex-rs/core/tests/common/responses.rs @@ -1,5 +1,6 @@ use std::sync::Arc; use std::sync::Mutex; +use std::time::Duration; use anyhow::Result; use base64::Engine; @@ -546,6 +547,19 @@ pub async fn mount_sse_once(server: &MockServer, body: String) -> ResponseMock { response_mock } +pub async fn mount_sse_once_with_delay( + server: &MockServer, + body: String, + delay: Duration, +) -> ResponseMock { + let (mock, response_mock) = base_mock(); + mock.respond_with(sse_response(body).set_delay(delay)) + .up_to_n_times(1) + .mount(server) + .await; + response_mock +} + pub async fn mount_compact_json_once_match( server: &MockServer, matcher: M, diff --git a/codex-rs/core/tests/suite/idle_warning.rs b/codex-rs/core/tests/suite/idle_warning.rs new file mode 100644 index 0000000000..d35c90e95c --- /dev/null +++ b/codex-rs/core/tests/suite/idle_warning.rs @@ -0,0 +1,100 @@ +#![allow(clippy::unwrap_used, clippy::expect_used)] + +use std::time::Duration; + +use codex_core::protocol::EventMsg; +use codex_core::protocol::Op; +use codex_core::protocol::WarningEvent; +use codex_core::status::set_test_idle_timeout; +use codex_core::status::set_test_status_widget_url; +use codex_protocol::user_input::UserInput; +use core_test_support::responses::ev_assistant_message; +use core_test_support::responses::ev_completed; +use core_test_support::responses::ev_response_created; +use core_test_support::responses::mount_sse_once_with_delay; +use core_test_support::responses::sse; +use core_test_support::responses::start_mock_server; +use core_test_support::test_codex::test_codex; +use core_test_support::wait_for_event; +use wiremock::Mock; +use wiremock::ResponseTemplate; +use wiremock::matchers::method; +use wiremock::matchers::path; + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn emits_warning_when_stream_is_idle_and_status_is_degraded() { + let status_server = start_mock_server().await; + let status_path = "/proxy/status.openai.com"; + + Mock::given(method("GET")) + .and(path(status_path)) + .respond_with(status_payload()) + .mount(&status_server) + .await; + + set_test_status_widget_url(format!("{}{}", status_server.uri(), status_path)); + set_test_idle_timeout(Duration::from_millis(30)); + + 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(40), + ) + .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(); + + tokio::time::sleep(Duration::from_millis(35)).await; + + let status_requests = status_server + .received_requests() + .await + .expect("status server running"); + assert!( + !status_requests.is_empty(), + "status widget was not queried before idle warning" + ); + + let warning = wait_for_event(&codex, |event| matches!(event, EventMsg::Warning(_))).await; + let EventMsg::Warning(WarningEvent { message }) = warning else { + panic!("expected warning event"); + }; + assert!( + message.contains("OpenAI status: MajorOutage"), + "unexpected warning message: {message}" + ); + + tokio::time::sleep(Duration::from_millis(25)).await; + + wait_for_event(&codex, |event| matches!(event, EventMsg::TaskComplete(_))).await; +} + +fn status_payload() -> ResponseTemplate { + ResponseTemplate::new(200) + .insert_header("content-type", "application/json") + .set_body_json(serde_json::json!({ + "summary": { + "components": [ + {"id": "cmp-1", "name": "Codex", "status_page_id": "page-1"} + ], + "affected_components": [ + {"component_id": "cmp-1", "status": "major_outage"} + ] + } + })) +} diff --git a/codex-rs/core/tests/suite/mod.rs b/codex-rs/core/tests/suite/mod.rs index b877663614..8978ea41bd 100644 --- a/codex-rs/core/tests/suite/mod.rs +++ b/codex-rs/core/tests/suite/mod.rs @@ -31,6 +31,7 @@ mod exec; mod exec_policy; mod fork_conversation; mod grep_files; +mod idle_warning; mod items; mod json_result; mod list_dir;