diff --git a/codex-rs/app-server/src/codex_message_processor.rs b/codex-rs/app-server/src/codex_message_processor.rs index 4e3bc70cb3..94ee478b9d 100644 --- a/codex-rs/app-server/src/codex_message_processor.rs +++ b/codex-rs/app-server/src/codex_message_processor.rs @@ -402,6 +402,7 @@ enum AppListLoadResult { enum ThreadShutdownResult { Complete, + ShutdownFailed(String), SubmitFailed, TimedOut, } @@ -2171,6 +2172,9 @@ impl CodexMessageProcessor { for thread_id in report.submit_failed { warn!("failed to submit Shutdown to thread {thread_id}"); } + for (thread_id, error) in report.shutdown_failed { + warn!("thread {thread_id} shutdown completed with error: {error}"); + } for thread_id in report.timed_out { warn!("timed out waiting for thread {thread_id} to shut down"); } @@ -5433,6 +5437,9 @@ impl CodexMessageProcessor { async fn wait_for_thread_shutdown(thread: &Arc) -> ThreadShutdownResult { match tokio::time::timeout(Duration::from_secs(10), thread.shutdown_and_wait()).await { Ok(Ok(())) => ThreadShutdownResult::Complete, + Ok(Err(CodexErr::ShutdownFailed(message))) => { + ThreadShutdownResult::ShutdownFailed(message) + } Ok(Err(_)) => ThreadShutdownResult::SubmitFailed, Err(_) => ThreadShutdownResult::TimedOut, } @@ -5515,40 +5522,41 @@ impl CodexMessageProcessor { let thread_manager = self.thread_manager.clone(); let thread_watch_manager = self.thread_watch_manager.clone(); tokio::spawn(async move { - match Self::wait_for_thread_shutdown(&thread).await { - ThreadShutdownResult::Complete => { - if thread_manager.remove_thread(&thread_id).await.is_none() { - info!( - "thread {thread_id} was already removed before unsubscribe finalized" - ); - thread_watch_manager - .remove_thread(&thread_id.to_string()) - .await; - pending_thread_unloads.lock().await.remove(&thread_id); - return; - } - thread_watch_manager - .remove_thread(&thread_id.to_string()) - .await; - let notification = ThreadClosedNotification { - thread_id: thread_id.to_string(), - }; - outgoing - .send_server_notification(ServerNotification::ThreadClosed( - notification, - )) - .await; - pending_thread_unloads.lock().await.remove(&thread_id); - } + let shutdown_error = match Self::wait_for_thread_shutdown(&thread).await { + ThreadShutdownResult::Complete => None, + ThreadShutdownResult::ShutdownFailed(error) => Some(error), ThreadShutdownResult::SubmitFailed => { pending_thread_unloads.lock().await.remove(&thread_id); warn!("failed to submit Shutdown to thread {thread_id}"); + return; } ThreadShutdownResult::TimedOut => { pending_thread_unloads.lock().await.remove(&thread_id); warn!("thread {thread_id} shutdown timed out; leaving thread loaded"); + return; } + }; + if let Some(error) = shutdown_error { + error!("thread {thread_id} shutdown completed with error: {error}"); } + if thread_manager.remove_thread(&thread_id).await.is_none() { + info!("thread {thread_id} was already removed before unsubscribe finalized"); + thread_watch_manager + .remove_thread(&thread_id.to_string()) + .await; + pending_thread_unloads.lock().await.remove(&thread_id); + return; + } + thread_watch_manager + .remove_thread(&thread_id.to_string()) + .await; + let notification = ThreadClosedNotification { + thread_id: thread_id.to_string(), + }; + outgoing + .send_server_notification(ServerNotification::ThreadClosed(notification)) + .await; + pending_thread_unloads.lock().await.remove(&thread_id); }); } @@ -5635,6 +5643,9 @@ impl CodexMessageProcessor { info!("thread {thread_id} was active; shutting down"); match Self::wait_for_thread_shutdown(&conversation).await { ThreadShutdownResult::Complete => {} + ThreadShutdownResult::ShutdownFailed(error) => { + error!("thread {thread_id} shutdown completed with error: {error}"); + } ThreadShutdownResult::SubmitFailed => { error!( "failed to submit Shutdown to thread {thread_id}; proceeding with archive" diff --git a/codex-rs/core/src/codex.rs b/codex-rs/core/src/codex.rs index f53a5456c4..a24b2872f5 100644 --- a/codex-rs/core/src/codex.rs +++ b/codex-rs/core/src/codex.rs @@ -748,7 +748,7 @@ impl Codex { } session_loop_termination.await; match self.session.shutdown_failure.borrow().clone() { - Some(message) => Err(CodexErr::Fatal(message)), + Some(message) => Err(CodexErr::ShutdownFailed(message)), _ => Ok(()), } } diff --git a/codex-rs/core/src/codex_tests.rs b/codex-rs/core/src/codex_tests.rs index e6b3068683..b73830bf10 100644 --- a/codex-rs/core/src/codex_tests.rs +++ b/codex-rs/core/src/codex_tests.rs @@ -3526,7 +3526,7 @@ async fn shutdown_and_wait_returns_shutdown_error_status() { .expect_err("shutdown error status should propagate"); match err { - CodexErr::Fatal(message) => assert_eq!(message, error_message), + CodexErr::ShutdownFailed(message) => assert_eq!(message, error_message), other => panic!("unexpected error: {other:?}"), } } diff --git a/codex-rs/core/src/thread_manager.rs b/codex-rs/core/src/thread_manager.rs index 2dd4ad5532..2ef2952614 100644 --- a/codex-rs/core/src/thread_manager.rs +++ b/codex-rs/core/src/thread_manager.rs @@ -179,12 +179,14 @@ impl From for ForkSnapshot { #[derive(Debug, Default, PartialEq, Eq)] pub struct ThreadShutdownReport { pub completed: Vec, + pub shutdown_failed: Vec<(ThreadId, String)>, pub submit_failed: Vec, pub timed_out: Vec, } enum ShutdownOutcome { Complete, + ShutdownFailed(String), SubmitFailed, TimedOut, } @@ -605,6 +607,9 @@ impl ThreadManager { let outcome = match tokio::time::timeout(timeout, thread.shutdown_and_wait()).await { Ok(Ok(())) => ShutdownOutcome::Complete, + Ok(Err(CodexErr::ShutdownFailed(message))) => { + ShutdownOutcome::ShutdownFailed(message) + } Ok(Err(_)) => ShutdownOutcome::SubmitFailed, Err(_) => ShutdownOutcome::TimedOut, }; @@ -616,6 +621,9 @@ impl ThreadManager { while let Some((thread_id, outcome)) = shutdowns.next().await { match outcome { ShutdownOutcome::Complete => report.completed.push(thread_id), + ShutdownOutcome::ShutdownFailed(message) => { + report.shutdown_failed.push((thread_id, message)); + } ShutdownOutcome::SubmitFailed => report.submit_failed.push(thread_id), ShutdownOutcome::TimedOut => report.timed_out.push(thread_id), } @@ -625,10 +633,16 @@ impl ThreadManager { for thread_id in &report.completed { tracked_threads.remove(thread_id); } + for (thread_id, _) in &report.shutdown_failed { + tracked_threads.remove(thread_id); + } report .completed .sort_by_key(std::string::ToString::to_string); + report + .shutdown_failed + .sort_by_key(|(thread_id, _)| thread_id.to_string()); report .submit_failed .sort_by_key(std::string::ToString::to_string); diff --git a/codex-rs/protocol/src/error.rs b/codex-rs/protocol/src/error.rs index 3ffdeb48e0..92a2143081 100644 --- a/codex-rs/protocol/src/error.rs +++ b/codex-rs/protocol/src/error.rs @@ -137,6 +137,8 @@ pub enum CodexErr { UnsupportedOperation(String), #[error("{0}")] RefreshTokenFailed(RefreshTokenFailedError), + #[error("shutdown failed: {0}")] + ShutdownFailed(String), #[error("Fatal error: {0}")] Fatal(String), // ----------------------------------------------------------------- @@ -171,6 +173,7 @@ impl CodexErr { | CodexErr::Interrupted | CodexErr::EnvVar(_) | CodexErr::Fatal(_) + | CodexErr::ShutdownFailed(_) | CodexErr::UsageNotIncluded | CodexErr::QuotaExceeded | CodexErr::InvalidImageRequest()