codex: distinguish shutdown durability failures (#16599)

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<CodexThread>) -> 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"

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(()),
         }
     }

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:?}"),
     }
 }

codex-rs/core/src/thread_manager.rs

@@ -179,12 +179,14 @@ impl From<usize> for ForkSnapshot {
 #[derive(Debug, Default, PartialEq, Eq)]
 pub struct ThreadShutdownReport {
     pub completed: Vec<ThreadId>,
+    pub shutdown_failed: Vec<(ThreadId, String)>,
     pub submit_failed: Vec<ThreadId>,
     pub timed_out: Vec<ThreadId>,
 }
 
 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);

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()