Show warning on Codex incidents

codex-rs/core/Cargo.toml

@@ -1,8 +1,8 @@
 [package]
-name = "codex-core"
-version.workspace = true
 edition.workspace = true
 license.workspace = true
+name = "codex-core"
+version.workspace = true
 
 [lib]
 doctest = false
@@ -18,12 +18,13 @@ askama = { workspace = true }
 async-channel = { workspace = true }
 async-trait = { workspace = true }
 base64 = { workspace = true }
-chrono = { workspace = true, features = ["serde"] }
 chardetng = { workspace = true }
+chrono = { workspace = true, features = ["serde"] }
+codex-api = { workspace = true }
 codex-app-server-protocol = { workspace = true }
 codex-apply-patch = { workspace = true }
 codex-async-utils = { workspace = true }
-codex-api = { workspace = true }
+codex-client = { workspace = true }
 codex-execpolicy = { workspace = true }
 codex-file-search = { workspace = true }
 codex-git = { workspace = true }
@@ -37,17 +38,19 @@ codex-utils-string = { workspace = true }
 codex-windows-sandbox = { package = "codex-windows-sandbox", path = "../windows-sandbox-rs" }
 dirs = { workspace = true }
 dunce = { workspace = true }
-env-flags = { workspace = true }
 encoding_rs = { workspace = true }
+env-flags = { workspace = true }
 eventsource-stream = { workspace = true }
 futures = { workspace = true }
 http = { workspace = true }
 indexmap = { workspace = true }
 keyring = { workspace = true, features = ["crypto-rust"] }
 libc = { workspace = true }
 mcp-types = { workspace = true }
+once_cell = { workspace = true }
 os_info = { workspace = true }
 rand = { workspace = true }
+regex = { workspace = true }
 regex-lite = { workspace = true }
 reqwest = { workspace = true, features = ["json", "stream"] }
 serde = { workspace = true, features = ["derive"] }
@@ -57,9 +60,6 @@ sha2 = { workspace = true }
 shlex = { workspace = true }
 similar = { workspace = true }
 strum_macros = { workspace = true }
-url = { workspace = true }
-once_cell = { workspace = true }
-regex = { workspace = true }
 tempfile = { workspace = true }
 test-case = "3.3.1"
 test-log = { workspace = true }
@@ -83,6 +83,7 @@ toml_edit = { workspace = true }
 tracing = { workspace = true, features = ["log"] }
 tree-sitter = { workspace = true }
 tree-sitter-bash = { workspace = true }
+url = { workspace = true }
 uuid = { workspace = true, features = ["serde", "v4", "v5"] }
 which = { workspace = true }
 wildmatch = { workspace = true }
@@ -92,9 +93,9 @@ deterministic_process_ids = []
 
 
 [target.'cfg(target_os = "linux")'.dependencies]
+keyring = { workspace = true, features = ["linux-native-async-persistent"] }
 landlock = { workspace = true }
 seccompiler = { workspace = true }
-keyring = { workspace = true, features = ["linux-native-async-persistent"] }
 
 [target.'cfg(target_os = "macos")'.dependencies]
 core-foundation = "0.9"

codex-rs/core/src/codex.rs

@@ -1,10 +1,12 @@
 use std::collections::HashMap;
 use std::fmt::Debug;
 use std::path::PathBuf;
+use std::pin::Pin;
 use std::sync::Arc;
 use std::sync::atomic::AtomicU64;
 
 use crate::AuthManager;
+use crate::ResponseStream;
 use crate::SandboxState;
 use crate::client_common::REVIEW_PROMPT;
 use crate::compact;
@@ -52,6 +54,7 @@ use serde_json::Value;
 use tokio::sync::Mutex;
 use tokio::sync::RwLock;
 use tokio::sync::oneshot;
+use tokio::time::sleep_until;
 use tokio_util::sync::CancellationToken;
 use tracing::debug;
 use tracing::error;
@@ -106,6 +109,7 @@ use crate::shell;
 use crate::state::ActiveTurn;
 use crate::state::SessionServices;
 use crate::state::SessionState;
+use crate::status::IdleWarning;
 use crate::tasks::GhostSnapshotTask;
 use crate::tasks::ReviewTask;
 use crate::tasks::SessionTask;
@@ -2180,12 +2184,19 @@ 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 = await_stream_with_idle_warning(
+        stream_future.as_mut(),
+        &mut idle_warning,
+        &sess,
+        &turn_context,
+    )
+    .await?;
+
+    idle_warning.mark_event();
 
     let tool_runtime = ToolCallRuntime::new(
         Arc::clone(&router),
@@ -2202,7 +2213,24 @@ async fn try_run_turn(
         // Poll the next item from the model stream. We must inspect *both* Ok and Err
         // cases so that transient stream failures (e.g., dropped SSE connection before
         // `response.completed`) bubble up and trigger the caller's retry logic.
-        let event = match stream.next().or_cancel(&cancellation_token).await {
+        let event = tokio::select! {
+            biased;
+            result = stream.next().or_cancel(&cancellation_token) => 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 event = match event {
             Ok(event) => event,
             Err(codex_async_utils::CancelErr::Cancelled) => {
                 let processed_items = output.try_collect().await?;
@@ -2401,6 +2429,34 @@ async fn try_run_turn(
     }
 }
 
+async fn await_stream_with_idle_warning<F>(
+    mut stream_future: Pin<&mut F>,
+    idle_warning: &mut IdleWarning,
+    sess: &Arc<Session>,
+    turn_context: &Arc<TurnContext>,
+) -> CodexResult<ResponseStream>
+where
+    F: std::future::Future<
+            Output = Result<CodexResult<ResponseStream>, codex_async_utils::CancelErr>,
+        > + Send,
+{
+    loop {
+        tokio::select! {
+            biased;
+            result = &mut stream_future => return 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;
+                }
+            }
+        }
+    }
+}
+
 async fn handle_non_tool_response_item(item: &ResponseItem) -> Option<TurnItem> {
     debug!(?item, "Output item");
 

codex-rs/core/src/lib.rs

@@ -42,6 +42,7 @@ pub mod parse_command;
 pub mod powershell;
 mod response_processing;
 pub mod sandboxing;
+pub mod status;
 mod text_encoding;
 pub mod token_data;
 mod truncate;