feat: add auto refresh on thread listeners (#9105)

This PR is in the scope of multi-agent work. 

An agent (=thread) can now spawn other agents. Those other agents are
not attached to any clients. We need a way to make sure that the clients
are aware of the new threads to look at (for approval for example). This
PR adds a channel to the `ThreadManager` that pushes the ID of those
newly created agents such that the client (here the app-server) can also
subscribe to those ones.

codex-rs/core/src/agent/control.rs

@@ -1,10 +1,8 @@
-use crate::CodexThread;
 use crate::agent::AgentStatus;
 use crate::error::CodexErr;
 use crate::error::Result as CodexResult;
 use crate::thread_manager::ThreadManagerState;
 use codex_protocol::ThreadId;
-use codex_protocol::protocol::EventMsg;
 use codex_protocol::protocol::Op;
 use codex_protocol::user_input::UserInput;
 use std::sync::Arc;
@@ -29,21 +27,18 @@ impl AgentControl {
     }
 
     /// Spawn a new agent thread and submit the initial prompt.
-    ///
-    /// If `headless` is true, a background drain task is spawned to prevent unbounded event growth
-    /// of the channel queue when there is no client actively reading the thread events.
     pub(crate) async fn spawn_agent(
         &self,
         config: crate::config::Config,
         prompt: String,
-        headless: bool,
     ) -> CodexResult<ThreadId> {
         let state = self.upgrade()?;
         let new_thread = state.spawn_new_thread(config, self.clone()).await?;
 
-        if headless {
-            spawn_headless_drain(Arc::clone(&new_thread.thread));
-        }
+        // Notify a new thread has been created. This notification will be processed by clients
+        // to subscribe or drain this newly created thread.
+        // TODO(jif) add helper for drain
+        state.notify_thread_created(new_thread.thread_id);
 
         self.send_prompt(new_thread.thread_id, prompt).await?;
 
@@ -110,38 +105,18 @@ impl AgentControl {
     }
 }
 
-/// When an agent is spawned "headless" (no UI/view attached), there may be no consumer polling
-/// `CodexThread::next_event()`. The underlying event channel is unbounded, so the producer can
-/// accumulate events indefinitely. This drain task prevents that memory growth by polling and
-/// discarding events until shutdown.
-fn spawn_headless_drain(thread: Arc<CodexThread>) {
-    tokio::spawn(async move {
-        loop {
-            match thread.next_event().await {
-                Ok(event) => {
-                    if matches!(event.msg, EventMsg::ShutdownComplete) {
-                        break;
-                    }
-                }
-                Err(err) => {
-                    tracing::warn!("failed to receive event from agent: {err:?}");
-                    break;
-                }
-            }
-        }
-    });
-}
-
 #[cfg(test)]
 mod tests {
     use super::*;
     use crate::CodexAuth;
+    use crate::CodexThread;
     use crate::ThreadManager;
     use crate::agent::agent_status_from_event;
     use crate::config::Config;
     use crate::config::ConfigBuilder;
     use assert_matches::assert_matches;
     use codex_protocol::protocol::ErrorEvent;
+    use codex_protocol::protocol::EventMsg;
     use codex_protocol::protocol::TurnAbortReason;
     use codex_protocol::protocol::TurnAbortedEvent;
     use codex_protocol::protocol::TurnCompleteEvent;
@@ -262,7 +237,7 @@ mod tests {
         let control = AgentControl::default();
         let (_home, config) = test_config().await;
         let err = control
-            .spawn_agent(config, "hello".to_string(), false)
+            .spawn_agent(config, "hello".to_string())
             .await
             .expect_err("spawn_agent should fail without a manager");
         assert_eq!(
@@ -363,7 +338,7 @@ mod tests {
         let harness = AgentControlHarness::new().await;
         let thread_id = harness
             .control
-            .spawn_agent(harness.config.clone(), "spawned".to_string(), false)
+            .spawn_agent(harness.config.clone(), "spawned".to_string())
             .await
             .expect("spawn_agent should succeed");
         let _thread = harness

codex-rs/core/src/thread_manager.rs

@@ -31,8 +31,11 @@ use std::sync::Arc;
 #[cfg(any(test, feature = "test-support"))]
 use tempfile::TempDir;
 use tokio::sync::RwLock;
+use tokio::sync::broadcast;
 use tracing::warn;
 
+const THREAD_CREATED_CHANNEL_CAPACITY: usize = 1024;
+
 /// Represents a newly created Codex thread (formerly called a conversation), including the first event
 /// (which is [`EventMsg::SessionConfigured`]).
 pub struct NewThread {
@@ -54,6 +57,7 @@ pub struct ThreadManager {
 /// function to require an `Arc<&Self>`.
 pub(crate) struct ThreadManagerState {
     threads: Arc<RwLock<HashMap<ThreadId, Arc<CodexThread>>>>,
+    thread_created_tx: broadcast::Sender<ThreadId>,
     auth_manager: Arc<AuthManager>,
     models_manager: Arc<ModelsManager>,
     skills_manager: Arc<SkillsManager>,
@@ -70,9 +74,11 @@ impl ThreadManager {
         auth_manager: Arc<AuthManager>,
         session_source: SessionSource,
     ) -> Self {
+        let (thread_created_tx, _) = broadcast::channel(THREAD_CREATED_CHANNEL_CAPACITY);
         Self {
             state: Arc::new(ThreadManagerState {
                 threads: Arc::new(RwLock::new(HashMap::new())),
+                thread_created_tx,
                 models_manager: Arc::new(ModelsManager::new(
                     codex_home.clone(),
                     auth_manager.clone(),
@@ -108,9 +114,11 @@ impl ThreadManager {
         codex_home: PathBuf,
     ) -> Self {
         let auth_manager = AuthManager::from_auth_for_testing(auth);
+        let (thread_created_tx, _) = broadcast::channel(THREAD_CREATED_CHANNEL_CAPACITY);
         Self {
             state: Arc::new(ThreadManagerState {
                 threads: Arc::new(RwLock::new(HashMap::new())),
+                thread_created_tx,
                 models_manager: Arc::new(ModelsManager::with_provider(
                     codex_home.clone(),
                     auth_manager.clone(),
@@ -174,6 +182,10 @@ impl ThreadManager {
         }
     }
 
+    pub fn subscribe_thread_created(&self) -> broadcast::Receiver<ThreadId> {
+        self.state.thread_created_tx.subscribe()
+    }
+
     pub async fn get_thread(&self, thread_id: ThreadId) -> CodexResult<Arc<CodexThread>> {
         self.state.get_thread(thread_id).await
     }
@@ -348,6 +360,10 @@ impl ThreadManagerState {
             session_configured,
         })
     }
+
+    pub(crate) fn notify_thread_created(&self, thread_id: ThreadId) {
+        let _ = self.thread_created_tx.send(thread_id);
+    }
 }
 
 /// Return a prefix of `items` obtained by cutting strictly before the nth user message

codex-rs/core/src/tools/handlers/collab.rs

@@ -94,7 +94,7 @@ mod spawn {
         let result = session
             .services
             .agent_control
-            .spawn_agent(config, args.message, true)
+            .spawn_agent(config, args.message)
             .await
             .map_err(collab_spawn_error)?;