codex: surface shutdown durability failures (#16599)

codex-rs/core/src/codex.rs

@@ -747,7 +747,10 @@ impl Codex {
             Err(err) => return Err(err),
         }
         session_loop_termination.await;
-        Ok(())
+        match self.session.shutdown_failure.borrow().clone() {
+            Some(message) => Err(CodexErr::Fatal(message)),
+            _ => Ok(()),
+        }
     }
 
     pub async fn next_event(&self) -> CodexResult<Event> {
@@ -821,6 +824,7 @@ pub(crate) struct Session {
     pub(crate) conversation_id: ThreadId,
     tx_event: Sender<Event>,
     agent_status: watch::Sender<AgentStatus>,
+    shutdown_failure: watch::Sender<Option<String>>,
     out_of_band_elicitation_paused: watch::Sender<bool>,
     state: Mutex<SessionState>,
     /// Serializes rebuild/apply cycles for the running proxy; each cycle
@@ -2039,12 +2043,14 @@ impl Session {
         ));
         let (out_of_band_elicitation_paused, _out_of_band_elicitation_paused_rx) =
             watch::channel(false);
+        let (shutdown_failure, _shutdown_failure_rx) = watch::channel(None);
 
         let (mailbox, mailbox_rx) = Mailbox::new();
         let sess = Arc::new(Session {
             conversation_id,
             tx_event: tx_event.clone(),
             agent_status,
+            shutdown_failure,
             out_of_band_elicitation_paused,
             state: Mutex::new(state),
             managed_network_proxy_refresh_lock: Mutex::new(()),
@@ -5637,10 +5643,12 @@ mod handlers {
             && let Err(e) = rec.shutdown().await
         {
             warn!("failed to shutdown rollout recorder: {e}");
+            let message = "Failed to shutdown rollout recorder".to_string();
+            sess.shutdown_failure.send_replace(Some(message.clone()));
             let event = Event {
                 id: sub_id.clone(),
                 msg: EventMsg::Error(ErrorEvent {
-                    message: "Failed to shutdown rollout recorder".to_string(),
+                    message,
                     codex_error_info: Some(CodexErrorInfo::Other),
                 }),
             };
@@ -5651,10 +5659,12 @@ mod handlers {
             && let Err(e) = state_db.checkpoint_wal().await
         {
             warn!("failed to checkpoint state db WAL during shutdown: {e}");
+            let message = "Failed to checkpoint state database WAL".to_string();
+            sess.shutdown_failure.send_replace(Some(message.clone()));
             let event = Event {
                 id: sub_id.clone(),
                 msg: EventMsg::Error(ErrorEvent {
-                    message: "Failed to checkpoint state database WAL".to_string(),
+                    message,
                     codex_error_info: Some(CodexErrorInfo::Other),
                 }),
             };

codex-rs/core/src/codex_tests.rs

@@ -2949,6 +2949,7 @@ pub(crate) async fn make_session_and_context() -> (Session, TurnContext) {
         conversation_id,
         tx_event,
         agent_status: agent_status_tx,
+        shutdown_failure: watch::channel(None).0,
         out_of_band_elicitation_paused: watch::channel(false).0,
         state: Mutex::new(state),
         managed_network_proxy_refresh_lock: Mutex::new(()),
@@ -3494,6 +3495,42 @@ async fn shutdown_and_wait_waits_when_shutdown_is_already_in_progress() {
         .expect("shutdown waiter");
 }
 
+#[tokio::test]
+async fn shutdown_and_wait_returns_shutdown_error_status() {
+    let (session, _turn_context) = make_session_and_context().await;
+    let (tx_sub, rx_sub) = async_channel::bounded(4);
+    let (_tx_event, rx_event) = async_channel::unbounded();
+    let (_agent_status_tx, agent_status) = watch::channel(AgentStatus::PendingInit);
+    let error_message = "shutdown durability failed".to_string();
+    let status_message = error_message.clone();
+    let session = Arc::new(session);
+    let session_for_loop = Arc::clone(&session);
+    let session_loop_handle = tokio::spawn(async move {
+        let shutdown: Submission = rx_sub.recv().await.expect("shutdown submission");
+        assert_eq!(shutdown.op, Op::Shutdown);
+        session_for_loop
+            .shutdown_failure
+            .send_replace(Some(status_message));
+    });
+    let codex = Arc::new(Codex {
+        tx_sub,
+        rx_event,
+        agent_status,
+        session,
+        session_loop_termination: session_loop_termination_from_handle(session_loop_handle),
+    });
+
+    let err = codex
+        .shutdown_and_wait()
+        .await
+        .expect_err("shutdown error status should propagate");
+
+    match err {
+        CodexErr::Fatal(message) => assert_eq!(message, error_message),
+        other => panic!("unexpected error: {other:?}"),
+    }
+}
+
 #[tokio::test]
 async fn shutdown_and_wait_shuts_down_cached_guardian_subagent() {
     let (parent_session, parent_turn_context) = make_session_and_context().await;
@@ -3793,6 +3830,7 @@ pub(crate) async fn make_session_and_context_with_dynamic_tools_and_rx(
         conversation_id,
         tx_event,
         agent_status: agent_status_tx,
+        shutdown_failure: watch::channel(None).0,
         out_of_band_elicitation_paused: watch::channel(false).0,
         state: Mutex::new(state),
         managed_network_proxy_refresh_lock: Mutex::new(()),