From 47768aec2cc43fae14dd7b2e4899a23db26d7005 Mon Sep 17 00:00:00 2001 From: Fabian Ponce Date: Tue, 23 Jun 2026 14:44:44 +0000 Subject: [PATCH] fix(app-server): clear subscriber snapshots on teardown --- .../src/request_processors/thread_processor_tests.rs | 12 ++++++++++++ codex-rs/app-server/src/thread_state.rs | 12 ++++++++---- 2 files changed, 20 insertions(+), 4 deletions(-) diff --git a/codex-rs/app-server/src/request_processors/thread_processor_tests.rs b/codex-rs/app-server/src/request_processors/thread_processor_tests.rs index 347b8d8d83..88dfdb5862 100644 --- a/codex-rs/app-server/src/request_processors/thread_processor_tests.rs +++ b/codex-rs/app-server/src/request_processors/thread_processor_tests.rs @@ -1255,6 +1255,16 @@ mod thread_processor_behavior_tests { ) .await .expect("connection should be live"); + let connection_ids = manager + .subscribe_to_connection_ids(thread_id) + .await + .expect("thread should have a connection-ids watcher"); + let has_connections = manager + .subscribe_to_has_connections(thread_id) + .await + .expect("thread should have a has-connections watcher"); + assert_eq!(connection_ids.borrow().as_ref(), &[connection]); + assert!(*has_connections.borrow()); { let state = manager.thread_state(thread_id).await; let mut state = state.lock().await; @@ -1273,6 +1283,8 @@ mod thread_processor_behavior_tests { manager.remove_thread_state(thread_id).await; assert_eq!(cancel_rx.await, Ok(())); + assert!(connection_ids.borrow().is_empty()); + assert!(!*has_connections.borrow()); let state = manager.thread_state(thread_id).await; let subscribed_connection_ids = manager.subscribed_connection_ids(thread_id).await; diff --git a/codex-rs/app-server/src/thread_state.rs b/codex-rs/app-server/src/thread_state.rs index e1e2643ed7..6dc227183f 100644 --- a/codex-rs/app-server/src/thread_state.rs +++ b/codex-rs/app-server/src/thread_state.rs @@ -429,10 +429,14 @@ impl ThreadStateManager { pub(crate) async fn remove_thread_state(&self, thread_id: ThreadId) { let thread_state = { let mut state = self.state.lock().await; - let thread_state = state - .threads - .remove(&thread_id) - .map(|thread_entry| thread_entry.state); + let thread_state = state.threads.remove(&thread_id).map(|mut thread_entry| { + // Watch receivers retain their last value after the sender is dropped. Publish + // a terminal empty snapshot so an in-flight listener cannot dispatch to stale + // subscribers while teardown cancellation is being observed. + thread_entry.connection_ids.clear(); + thread_entry.publish_connections(); + thread_entry.state + }); state.thread_ids_by_connection.retain(|_, thread_ids| { thread_ids.remove(&thread_id); !thread_ids.is_empty()