openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-09-14 11:57:03 +00:00
Code Issues Packages Projects Releases Wiki Activity

client

Browse Source
This commit is contained in:
Ahmed Ibrahim
2025-12-02 14:45:44 -08:00
parent db70faab42
commit 47ef2cd9ca
4 changed files with 168 additions and 24 deletions
Download Patch File Download Diff File Expand all files Collapse all files

26
codex-rs/core/src/codex.rs
View File

@@ -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<ComponentHealth> {
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<Config>,
@@ -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(

2
codex-rs/core/src/state/service.rs
View File

@@ -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<AuthManager>,
pub(crate) otel_event_manager: OtelEventManager,
pub(crate) tool_approvals: Mutex<ApprovalStore>,
pub(crate) codex_backend_status: Mutex<Option<ComponentHealth>>,
}

55
codex-rs/core/src/status.rs
View File

@@ -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<String> {
if self.warning_sent {
pub(crate) async fn maybe_warning_message(&mut self, session: &Session) -> Option<String> {
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<ComponentHealth> {
.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

109
codex-rs/core/tests/suite/idle_warning.rs
View File

@@ -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<AtomicUsize>,
}
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)
}
}