app-server: Only unload threads which were unused for some time (#17398)

Currently app-server may unload actively running threads once the last connection disconnects, which is not expected. Instead track when was the last active turn & when there were any subscribers the last time, also add 30 minute idleness/no subscribers timer to reduce the churn.
2026-07-01 00:31:56 +08:00 · 2026-04-13 12:25:26 -07:00
parent d905376628
commit a5507b59c4
6 changed files with 495 additions and 160 deletions
@@ -143,7 +143,7 @@ Example with notification opt-out:
 - `thread/memoryMode/set` — experimental; set a thread’s persisted memory eligibility to `"enabled"` or `"disabled"` for either a loaded thread or a stored rollout; returns `{}` on success.
 - `thread/status/changed` — notification emitted when a loaded thread’s status changes (`threadId` + new `status`).
 - `thread/archive` — move a thread’s rollout file into the archived directory; returns `{}` on success and emits `thread/archived`.
- `thread/unsubscribe` — unsubscribe this connection from thread turn/item events. If this was the last subscriber, the server shuts down and unloads the thread, then emits `thread/closed`.
+- `thread/unsubscribe` — unsubscribe this connection from thread turn/item events. If this was the last subscriber, the server keeps the thread loaded and unloads it only after it has had no subscribers and no thread activity for 30 minutes, then emits `thread/closed`.
 - `thread/name/set` — set or update a thread’s user-facing name for either a loaded thread or a persisted rollout; returns `{}` on success and emits `thread/name/updated` to initialized, opted-in clients. Thread names are not required to be unique; name lookups resolve to the most recently updated thread.
 - `thread/unarchive` — move an archived rollout file back into the sessions directory; returns the restored `thread` on success and emits `thread/unarchived`.
 - `thread/compact/start` — trigger conversation history compaction for a thread; returns `{}` immediately while progress streams through standard turn/item notifications.
@@ -338,11 +338,16 @@ When `nextCursor` is `null`, you’ve reached the final page.
 - `notSubscribed` when the connection was not subscribed to that thread.
 - `notLoaded` when the thread is not loaded.

-If this was the last subscriber, the server unloads the thread and emits `thread/closed` and a `thread/status/changed` transition to `notLoaded`.
+If this was the last subscriber, the server does not unload the thread immediately. It unloads the thread after the thread has had no subscribers and no thread activity for 30 minutes, then emits `thread/closed` and a `thread/status/changed` transition to `notLoaded`.

 ```json
 { "method": "thread/unsubscribe", "id": 22, "params": { "threadId": "thr_123" } }
 { "id": 22, "result": { "status": "unsubscribed" } }
+```
+
+Later, after the idle unload timeout:
+
+```json
 { "method": "thread/status/changed", "params": {
    "threadId": "thr_123",
    "status": { "type": "notLoaded" }
@@ -329,6 +329,7 @@ use std::sync::RwLock;
 use std::sync::atomic::AtomicBool;
 use std::sync::atomic::Ordering;
 use std::time::Duration;
+use std::time::Instant;
 use std::time::SystemTime;
 use tokio::sync::Mutex;
 use tokio::sync::broadcast;
@@ -371,6 +372,7 @@ struct ThreadListFilters {
 const LOGIN_CHATGPT_TIMEOUT: Duration = Duration::from_secs(10 * 60);
 const LOGIN_ISSUER_OVERRIDE_ENV_VAR: &str = "CODEX_APP_SERVER_LOGIN_ISSUER";
 const APP_LIST_LOAD_TIMEOUT: Duration = Duration::from_secs(90);
+const THREAD_UNLOADING_DELAY: Duration = Duration::from_secs(30 * 60);

 enum ActiveLogin {
    Browser {
@@ -460,6 +462,7 @@ struct ListenerTaskContext {
    thread_manager: Arc<ThreadManager>,
    thread_state_manager: ThreadStateManager,
    outgoing: Arc<OutgoingMessageSender>,
+    pending_thread_unloads: Arc<Mutex<HashSet<ThreadId>>>,
    analytics_events_client: AnalyticsEventsClient,
    general_analytics_enabled: bool,
    thread_watch_manager: ThreadWatchManager,
@@ -480,6 +483,110 @@ enum RefreshTokenRequestOutcome {
    FailedPermanently,
 }

+struct UnloadingState {
+    delay: Duration,
+    has_subscribers_rx: watch::Receiver<bool>,
+    has_subscribers: (bool, Instant),
+    thread_status_rx: watch::Receiver<ThreadStatus>,
+    is_active: (bool, Instant),
+}
+
+impl UnloadingState {
+    async fn new(
+        listener_task_context: &ListenerTaskContext,
+        thread_id: ThreadId,
+        delay: Duration,
+    ) -> Option<Self> {
+        let has_subscribers_rx = listener_task_context
+            .thread_state_manager
+            .subscribe_to_has_connections(thread_id)
+            .await?;
+        let thread_status_rx = listener_task_context
+            .thread_watch_manager
+            .subscribe(thread_id)
+            .await?;
+        let has_subscribers = (*has_subscribers_rx.borrow(), Instant::now());
+        let is_active = (
+            matches!(*thread_status_rx.borrow(), ThreadStatus::Active { .. }),
+            Instant::now(),
+        );
+        Some(Self {
+            delay,
+            has_subscribers_rx,
+            thread_status_rx,
+            has_subscribers,
+            is_active,
+        })
+    }
+
+    fn unloading_target(&self) -> Option<Instant> {
+        match (self.has_subscribers, self.is_active) {
+            ((false, has_no_subscribers_since), (false, is_inactive_since)) => {
+                Some(std::cmp::max(has_no_subscribers_since, is_inactive_since) + self.delay)
+            }
+            _ => None,
+        }
+    }
+
+    fn sync_receiver_values(&mut self) {
+        let has_subscribers = *self.has_subscribers_rx.borrow();
+        if self.has_subscribers.0 != has_subscribers {
+            self.has_subscribers = (has_subscribers, Instant::now());
+        }
+
+        let is_active = matches!(*self.thread_status_rx.borrow(), ThreadStatus::Active { .. });
+        if self.is_active.0 != is_active {
+            self.is_active = (is_active, Instant::now());
+        }
+    }
+
+    fn should_unload_now(&mut self) -> bool {
+        self.sync_receiver_values();
+        self.unloading_target()
+            .is_some_and(|target| target <= Instant::now())
+    }
+
+    fn note_thread_activity_observed(&mut self) {
+        if !self.is_active.0 {
+            self.is_active = (false, Instant::now());
+        }
+    }
+
+    async fn wait_for_unloading_trigger(&mut self) -> bool {
+        loop {
+            self.sync_receiver_values();
+            let unloading_target = self.unloading_target();
+            if let Some(target) = unloading_target
+                && target <= Instant::now()
+            {
+                return true;
+            }
+            let unloading_sleep = async {
+                if let Some(target) = unloading_target {
+                    tokio::time::sleep_until(target.into()).await;
+                } else {
+                    futures::future::pending::<()>().await;
+                }
+            };
+            tokio::select! {
+                _ = unloading_sleep => return true,
+                changed = self.has_subscribers_rx.changed() => {
+                    if changed.is_err() {
+                        return false;
+                    }
+                    self.sync_receiver_values();
+                },
+                changed = self.thread_status_rx.changed() => {
+                    if changed.is_err() {
+                        return false;
+                    }
+                    self.sync_receiver_values();
+                },
+            }
+        }
+    }
+}
+
 pub(crate) struct CodexMessageProcessorArgs {
    pub(crate) auth_manager: Arc<AuthManager>,
    pub(crate) thread_manager: Arc<ThreadManager>,
@@ -2149,6 +2256,7 @@ impl CodexMessageProcessor {
            thread_manager: Arc::clone(&self.thread_manager),
            thread_state_manager: self.thread_state_manager.clone(),
            outgoing: Arc::clone(&self.outgoing),
+            pending_thread_unloads: Arc::clone(&self.pending_thread_unloads),
            analytics_events_client: self.analytics_events_client.clone(),
            general_analytics_enabled: self.config.features.enabled(Feature::GeneralAnalytics),
            thread_watch_manager: self.thread_watch_manager.clone(),
@@ -3884,17 +3992,17 @@ impl CodexMessageProcessor {
        self.command_exec_manager
            .connection_closed(connection_id)
            .await;
-        let thread_ids_with_no_subscribers = self
+        let thread_ids = self
            .thread_state_manager
            .remove_connection(connection_id)
            .await;
-        for thread_id in thread_ids_with_no_subscribers {
-            let Ok(thread) = self.thread_manager.get_thread(thread_id).await else {
+
+        for thread_id in thread_ids {
+            if self.thread_manager.get_thread(thread_id).await.is_err() {
+                // Reconcile stale app-server bookkeeping when the thread has already been
+                // removed from the core manager.
                self.finalize_thread_teardown(thread_id).await;
-                continue;
-            };
-            self.unload_thread_without_subscribers(thread_id, thread)
-                .await;
+            }
        }
    }

@@ -4260,13 +4368,18 @@ impl CodexMessageProcessor {
                .thread_state_manager
                .thread_state(existing_thread_id)
                .await;
-            self.ensure_listener_task_running(
-                existing_thread_id,
-                existing_thread.clone(),
-                thread_state.clone(),
-                ApiVersion::V2,
-            )
-            .await;
+            if let Err(error) = self
+                .ensure_listener_task_running(
+                    existing_thread_id,
+                    existing_thread.clone(),
+                    thread_state.clone(),
+                    ApiVersion::V2,
+                )
+                .await
+            {
+                self.outgoing.send_error(request_id, error).await;
+                return true;
+            }

            let config_snapshot = existing_thread.config_snapshot().await;
            let mismatch_details = collect_resume_override_mismatches(params, &config_snapshot);
@@ -5653,31 +5766,23 @@ impl CodexMessageProcessor {
    }

    async fn unload_thread_without_subscribers(
-        &self,
+        thread_manager: Arc<ThreadManager>,
+        outgoing: Arc<OutgoingMessageSender>,
+        pending_thread_unloads: Arc<Mutex<HashSet<ThreadId>>>,
+        thread_state_manager: ThreadStateManager,
+        thread_watch_manager: ThreadWatchManager,
        thread_id: ThreadId,
        thread: Arc<CodexThread>,
    ) {
-        // This connection was the last subscriber. Only now do we unload the thread.
-        info!("thread {thread_id} has no subscribers; shutting down");
-        let should_start_unload_task = self.pending_thread_unloads.lock().await.insert(thread_id);
+        info!("thread {thread_id} has no subscribers and is idle; shutting down");

        // Any pending app-server -> client requests for this thread can no longer be
        // answered; cancel their callbacks before shutdown/unload.
-        self.outgoing
+        outgoing
            .cancel_requests_for_thread(thread_id, /*error*/ None)
            .await;
-        self.thread_state_manager
-            .remove_thread_state(thread_id)
-            .await;
+        thread_state_manager.remove_thread_state(thread_id).await;

-        if !should_start_unload_task {
-            return;
-        }
-
-        let outgoing = self.outgoing.clone();
-        let pending_thread_unloads = self.pending_thread_unloads.clone();
-        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 => {
@@ -5726,7 +5831,7 @@ impl CodexMessageProcessor {
            }
        };

-        let Ok(thread) = self.thread_manager.get_thread(thread_id).await else {
+        if self.thread_manager.get_thread(thread_id).await.is_err() {
            // Reconcile stale app-server bookkeeping when the thread has already been
            // removed from the core manager. This keeps loaded-status/subscription state
            // consistent with the source of truth before reporting NotLoaded.
@@ -5746,30 +5851,14 @@ impl CodexMessageProcessor {
            .thread_state_manager
            .unsubscribe_connection_from_thread(thread_id, request_id.connection_id)
            .await;
-        if !was_subscribed {
-            self.outgoing
-                .send_response(
-                    request_id,
-                    ThreadUnsubscribeResponse {
-                        status: ThreadUnsubscribeStatus::NotSubscribed,
-                    },
-                )
-                .await;
-            return;
-        }
-
-        if !self.thread_state_manager.has_subscribers(thread_id).await {
-            self.unload_thread_without_subscribers(thread_id, thread)
-                .await;
-        }

+        let status = if was_subscribed {
+            ThreadUnsubscribeStatus::Unsubscribed
+        } else {
+            ThreadUnsubscribeStatus::NotSubscribed
+        };
        self.outgoing
-            .send_response(
-                request_id,
-                ThreadUnsubscribeResponse {
-                    status: ThreadUnsubscribeStatus::Unsubscribed,
-                },
-            )
+            .send_response(request_id, ThreadUnsubscribeResponse { status })
            .await;
    }

@@ -7514,6 +7603,7 @@ impl CodexMessageProcessor {
                thread_manager: Arc::clone(&self.thread_manager),
                thread_state_manager: self.thread_state_manager.clone(),
                outgoing: Arc::clone(&self.outgoing),
+                pending_thread_unloads: Arc::clone(&self.pending_thread_unloads),
                analytics_events_client: self.analytics_events_client.clone(),
                general_analytics_enabled: self.config.features.enabled(Feature::GeneralAnalytics),
                thread_watch_manager: self.thread_watch_manager.clone(),
@@ -7549,21 +7639,45 @@ impl CodexMessageProcessor {
                });
            }
        };
-        let Some(thread_state) = listener_task_context
-            .thread_state_manager
-            .try_ensure_connection_subscribed(conversation_id, connection_id, raw_events_enabled)
-            .await
-        else {
-            return Ok(EnsureConversationListenerResult::ConnectionClosed);
+        let thread_state = {
+            let pending_thread_unloads = listener_task_context.pending_thread_unloads.lock().await;
+            if pending_thread_unloads.contains(&conversation_id) {
+                return Err(JSONRPCErrorError {
+                    code: INVALID_REQUEST_ERROR_CODE,
+                    message: format!(
+                        "thread {conversation_id} is closing; retry after the thread is closed"
+                    ),
+                    data: None,
+                });
+            }
+            let Some(thread_state) = listener_task_context
+                .thread_state_manager
+                .try_ensure_connection_subscribed(
+                    conversation_id,
+                    connection_id,
+                    raw_events_enabled,
+                )
+                .await
+            else {
+                return Ok(EnsureConversationListenerResult::ConnectionClosed);
+            };
+            thread_state
        };
-        Self::ensure_listener_task_running_task(
-            listener_task_context,
+        if let Err(error) = Self::ensure_listener_task_running_task(
+            listener_task_context.clone(),
            conversation_id,
            conversation,
            thread_state,
            api_version,
        )
-        .await;
+        .await
+        {
+            let _ = listener_task_context
+                .thread_state_manager
+                .unsubscribe_connection_from_thread(conversation_id, connection_id)
+                .await;
+            return Err(error);
+        }
        Ok(EnsureConversationListenerResult::Attached)
    }

@@ -7597,12 +7711,13 @@ impl CodexMessageProcessor {
        conversation: Arc<CodexThread>,
        thread_state: Arc<Mutex<ThreadState>>,
        api_version: ApiVersion,
-    ) {
+    ) -> Result<(), JSONRPCErrorError> {
        Self::ensure_listener_task_running_task(
            ListenerTaskContext {
                thread_manager: Arc::clone(&self.thread_manager),
                thread_state_manager: self.thread_state_manager.clone(),
                outgoing: Arc::clone(&self.outgoing),
+                pending_thread_unloads: Arc::clone(&self.pending_thread_unloads),
                analytics_events_client: self.analytics_events_client.clone(),
                general_analytics_enabled: self.config.features.enabled(Feature::GeneralAnalytics),
                thread_watch_manager: self.thread_watch_manager.clone(),
@@ -7614,7 +7729,7 @@ impl CodexMessageProcessor {
            thread_state,
            api_version,
        )
-        .await;
+        .await
    }

    async fn ensure_listener_task_running_task(
@@ -7623,12 +7738,27 @@ impl CodexMessageProcessor {
        conversation: Arc<CodexThread>,
        thread_state: Arc<Mutex<ThreadState>>,
        api_version: ApiVersion,
-    ) {
+    ) -> Result<(), JSONRPCErrorError> {
        let (cancel_tx, mut cancel_rx) = oneshot::channel();
+        let Some(mut unloading_state) = UnloadingState::new(
+            &listener_task_context,
+            conversation_id,
+            THREAD_UNLOADING_DELAY,
+        )
+        .await
+        else {
+            return Err(JSONRPCErrorError {
+                code: INVALID_REQUEST_ERROR_CODE,
+                message: format!(
+                    "thread {conversation_id} is closing; retry after the thread is closed"
+                ),
+                data: None,
+            });
+        };
        let (mut listener_command_rx, listener_generation) = {
            let mut thread_state = thread_state.lock().await;
            if thread_state.listener_matches(&conversation) {
-                return;
+                return Ok(());
            }
            thread_state.set_listener(cancel_tx, &conversation)
        };
@@ -7636,6 +7766,7 @@ impl CodexMessageProcessor {
            outgoing,
            thread_manager,
            thread_state_manager,
+            pending_thread_unloads,
            analytics_events_client: _,
            general_analytics_enabled: _,
            thread_watch_manager,
@@ -7646,10 +7777,28 @@ impl CodexMessageProcessor {
        tokio::spawn(async move {
            loop {
                tokio::select! {
+                    biased;
                    _ = &mut cancel_rx => {
                        // Listener was superseded or the thread is being torn down.
                        break;
                    }
+                    listener_command = listener_command_rx.recv() => {
+                        let Some(listener_command) = listener_command else {
+                            break;
+                        };
+                        handle_thread_listener_command(
+                            conversation_id,
+                            &conversation,
+                            codex_home.as_path(),
+                            &thread_state_manager,
+                            &thread_state,
+                            &thread_watch_manager,
+                            &outgoing_for_task,
+                            &pending_thread_unloads,
+                            listener_command,
+                        )
+                        .await;
+                    }
                    event = conversation.next_event() => {
                        let event = match event {
                            Ok(event) => event,
@@ -7704,21 +7853,38 @@ impl CodexMessageProcessor {
                        )
                        .await;
                    }
-                    listener_command = listener_command_rx.recv() => {
-                        let Some(listener_command) = listener_command else {
+                    unloading_watchers_open = unloading_state.wait_for_unloading_trigger() => {
+                        if !unloading_watchers_open {
                            break;
-                        };
-                        handle_thread_listener_command(
+                        }
+                        if !unloading_state.should_unload_now() {
+                            continue;
+                        }
+                        if matches!(conversation.agent_status().await, AgentStatus::Running) {
+                            unloading_state.note_thread_activity_observed();
+                            continue;
+                        }
+                        {
+                            let mut pending_thread_unloads = pending_thread_unloads.lock().await;
+                            if pending_thread_unloads.contains(&conversation_id) {
+                                continue;
+                            }
+                            if !unloading_state.should_unload_now() {
+                                continue;
+                            }
+                            pending_thread_unloads.insert(conversation_id);
+                        }
+                        Self::unload_thread_without_subscribers(
+                            thread_manager.clone(),
+                            outgoing_for_task.clone(),
+                            pending_thread_unloads.clone(),
+                            thread_state_manager.clone(),
+                            thread_watch_manager.clone(),
                            conversation_id,
-                            &conversation,
-                            codex_home.as_path(),
-                            &thread_state_manager,
-                            &thread_state,
-                            &thread_watch_manager,
-                            &outgoing_for_task,
-                            listener_command,
+                            conversation.clone(),
                        )
                        .await;
+                        break;
                    }
                }
            }
@@ -7728,6 +7894,7 @@ impl CodexMessageProcessor {
                thread_state.clear_listener();
            }
        });
+        Ok(())
    }
    async fn git_diff_to_origin(&self, request_id: ConnectionRequestId, cwd: PathBuf) {
        let diff = git_diff_to_remote(&cwd).await;
@@ -8218,6 +8385,7 @@ async fn handle_thread_listener_command(
    thread_state: &Arc<Mutex<ThreadState>>,
    thread_watch_manager: &ThreadWatchManager,
    outgoing: &Arc<OutgoingMessageSender>,
+    pending_thread_unloads: &Arc<Mutex<HashSet<ThreadId>>>,
    listener_command: ThreadListenerCommand,
 ) {
    match listener_command {
@@ -8230,6 +8398,7 @@ async fn handle_thread_listener_command(
                thread_state,
                thread_watch_manager,
                outgoing,
+                pending_thread_unloads,
                *resume_request,
            )
            .await;
@@ -8259,6 +8428,7 @@ async fn handle_pending_thread_resume_request(
    thread_state: &Arc<Mutex<ThreadState>>,
    thread_watch_manager: &ThreadWatchManager,
    outgoing: &Arc<OutgoingMessageSender>,
+    pending_thread_unloads: &Arc<Mutex<HashSet<ThreadId>>>,
    pending: crate::thread_state::PendingThreadResumeRequest,
 ) {
    let active_turn = {
@@ -8312,6 +8482,37 @@ async fn handle_pending_thread_resume_request(
        has_live_in_progress_turn,
    );

+    {
+        let pending_thread_unloads = pending_thread_unloads.lock().await;
+        if pending_thread_unloads.contains(&conversation_id) {
+            drop(pending_thread_unloads);
+            outgoing
+                .send_error(
+                    request_id,
+                    JSONRPCErrorError {
+                        code: INVALID_REQUEST_ERROR_CODE,
+                        message: format!(
+                            "thread {conversation_id} is closing; retry thread/resume after the thread is closed"
+                        ),
+                        data: None,
+                    },
+                )
+                .await;
+            return;
+        }
+        if !thread_state_manager
+            .try_add_connection_to_thread(conversation_id, connection_id)
+            .await
+        {
+            tracing::debug!(
+                thread_id = %conversation_id,
+                connection_id = ?connection_id,
+                "skipping running thread resume for closed connection"
+            );
+            return;
+        }
+    }
+
    let ThreadConfigSnapshot {
        model,
        model_provider_id,
@@ -8340,9 +8541,6 @@ async fn handle_pending_thread_resume_request(
    outgoing
        .replay_requests_to_connection_for_thread(connection_id, conversation_id)
        .await;
-    let _attached = thread_state_manager
-        .try_add_connection_to_thread(conversation_id, connection_id)
-        .await;
 }

 enum ThreadTurnSource<'a> {
@@ -10137,6 +10335,53 @@ mod tests {
        Ok(())
    }

+    #[tokio::test]
+    async fn adding_connection_to_thread_updates_has_connections_watcher() -> Result<()> {
+        let manager = ThreadStateManager::new();
+        let thread_id = ThreadId::from_string("ad7f0408-99b8-4f6e-a46f-bd0eec433370")?;
+        let connection_a = ConnectionId(1);
+        let connection_b = ConnectionId(2);
+
+        manager.connection_initialized(connection_a).await;
+        manager.connection_initialized(connection_b).await;
+        manager
+            .try_ensure_connection_subscribed(
+                thread_id,
+                connection_a,
+                /*experimental_raw_events*/ false,
+            )
+            .await
+            .expect("connection_a should be live");
+        let mut has_connections = manager
+            .subscribe_to_has_connections(thread_id)
+            .await
+            .expect("thread should have a has-connections watcher");
+        assert!(*has_connections.borrow());
+
+        assert!(
+            manager
+                .unsubscribe_connection_from_thread(thread_id, connection_a)
+                .await
+        );
+        tokio::time::timeout(Duration::from_secs(1), has_connections.changed())
+            .await
+            .expect("timed out waiting for no-subscriber update")
+            .expect("has-connections watcher should remain open");
+        assert!(!*has_connections.borrow());
+
+        assert!(
+            manager
+                .try_add_connection_to_thread(thread_id, connection_b)
+                .await
+        );
+        tokio::time::timeout(Duration::from_secs(1), has_connections.changed())
+            .await
+            .expect("timed out waiting for subscriber update")
+            .expect("has-connections watcher should remain open");
+        assert!(*has_connections.borrow());
+        Ok(())
+    }
+
    #[tokio::test]
    async fn closed_connection_cannot_be_reintroduced_by_auto_subscribe() -> Result<()> {
        let manager = ThreadStateManager::new();
@@ -16,6 +16,7 @@ use std::sync::Weak;
 use tokio::sync::Mutex;
 use tokio::sync::mpsc;
 use tokio::sync::oneshot;
+use tokio::sync::watch;
 use tracing::error;

 type PendingInterruptQueue = Vec<(
@@ -159,6 +160,7 @@ pub(crate) async fn resolve_server_request_on_thread_listener(
 struct ThreadEntry {
    state: Arc<Mutex<ThreadState>>,
    connection_ids: HashSet<ConnectionId>,
+    has_connections_watcher: watch::Sender<bool>,
 }

 impl Default for ThreadEntry {
@@ -166,10 +168,21 @@ impl Default for ThreadEntry {
        Self {
            state: Arc::new(Mutex::new(ThreadState::default())),
            connection_ids: HashSet::new(),
+            has_connections_watcher: watch::channel(false).0,
        }
    }
 }

+impl ThreadEntry {
+    fn update_has_connections(&self) {
+        let _ = self.has_connections_watcher.send_if_modified(|current| {
+            let prev = *current;
+            *current = !self.connection_ids.is_empty();
+            prev != *current
+        });
+    }
+}
+
 #[derive(Default)]
 struct ThreadStateManagerInner {
    live_connections: HashSet<ConnectionId>,
@@ -286,12 +299,14 @@ impl ThreadStateManager {
            }
            if let Some(thread_entry) = state.threads.get_mut(&thread_id) {
                thread_entry.connection_ids.remove(&connection_id);
+                thread_entry.update_has_connections();
            }
        };

        true
    }

+    #[cfg(test)]
    pub(crate) async fn has_subscribers(&self, thread_id: ThreadId) -> bool {
        self.state
            .lock()
@@ -319,6 +334,7 @@ impl ThreadStateManager {
                .insert(thread_id);
            let thread_entry = state.threads.entry(thread_id).or_default();
            thread_entry.connection_ids.insert(connection_id);
+            thread_entry.update_has_connections();
            thread_entry.state.clone()
        };
        {
@@ -344,12 +360,9 @@ impl ThreadStateManager {
            .entry(connection_id)
            .or_default()
            .insert(thread_id);
-        state
-            .threads
-            .entry(thread_id)
-            .or_default()
-            .connection_ids
-            .insert(connection_id);
+        let thread_entry = state.threads.entry(thread_id).or_default();
+        thread_entry.connection_ids.insert(connection_id);
+        thread_entry.update_has_connections();
        true
    }

@@ -364,6 +377,7 @@ impl ThreadStateManager {
            for thread_id in &thread_ids {
                if let Some(thread_entry) = state.threads.get_mut(thread_id) {
                    thread_entry.connection_ids.remove(&connection_id);
+                    thread_entry.update_has_connections();
                }
            }
            thread_ids
@@ -377,4 +391,15 @@ impl ThreadStateManager {
                .collect::<Vec<_>>()
        }
    }
+
+    pub(crate) async fn subscribe_to_has_connections(
+        &self,
+        thread_id: ThreadId,
+    ) -> Option<watch::Receiver<bool>> {
+        let state = self.state.lock().await;
+        state
+            .threads
+            .get(&thread_id)
+            .map(|thread_entry| thread_entry.has_connections_watcher.subscribe())
+    }
 }
@@ -8,6 +8,7 @@ use codex_app_server_protocol::Thread;
 use codex_app_server_protocol::ThreadActiveFlag;
 use codex_app_server_protocol::ThreadStatus;
 use codex_app_server_protocol::ThreadStatusChangedNotification;
+use codex_protocol::ThreadId;
 use std::collections::HashMap;
 #[cfg(test)]
 use std::path::PathBuf;
@@ -244,6 +245,13 @@ impl ThreadWatchManager {
        }
    }

+    pub(crate) async fn subscribe(
+        &self,
+        thread_id: ThreadId,
+    ) -> Option<watch::Receiver<ThreadStatus>> {
+        Some(self.state.lock().await.subscribe(thread_id.to_string()))
+    }
+
    async fn note_active_guard_released(
        &self,
        thread_id: String,
@@ -295,6 +303,7 @@ pub(crate) fn resolve_thread_status(
 #[derive(Default)]
 struct ThreadWatchState {
    runtime_by_thread_id: HashMap<String, RuntimeFacts>,
+    status_watcher_by_thread_id: HashMap<String, watch::Sender<ThreadStatus>>,
 }

 impl ThreadWatchState {
@@ -309,6 +318,7 @@ impl ThreadWatchState {
            .entry(thread_id.clone())
            .or_default();
        runtime.is_loaded = true;
+        self.update_status_watcher_for_thread(&thread_id);
        if emit_notification {
            self.status_changed_notification(thread_id, previous_status)
        } else {
@@ -319,6 +329,7 @@ impl ThreadWatchState {
    fn remove_thread(&mut self, thread_id: &str) -> Option<ThreadStatusChangedNotification> {
        let previous_status = self.status_for(thread_id);
        self.runtime_by_thread_id.remove(thread_id);
+        self.update_status_watcher(thread_id, &ThreadStatus::NotLoaded);
        if previous_status.is_some() && previous_status != Some(ThreadStatus::NotLoaded) {
            Some(ThreadStatusChangedNotification {
                thread_id: thread_id.to_string(),
@@ -344,6 +355,7 @@ impl ThreadWatchState {
            .or_default();
        runtime.is_loaded = true;
        mutate(runtime);
+        self.update_status_watcher_for_thread(thread_id);
        self.status_changed_notification(thread_id.to_string(), previous_status)
    }

@@ -358,6 +370,40 @@ impl ThreadWatchState {
            .unwrap_or(ThreadStatus::NotLoaded)
    }

+    fn subscribe(&mut self, thread_id: String) -> watch::Receiver<ThreadStatus> {
+        let status = self.loaded_status_for_thread(&thread_id);
+        let sender = self
+            .status_watcher_by_thread_id
+            .entry(thread_id)
+            .or_insert_with(|| watch::channel(status.clone()).0);
+        sender.subscribe()
+    }
+
+    fn update_status_watcher_for_thread(&mut self, thread_id: &str) {
+        let status = self.loaded_status_for_thread(thread_id);
+        self.update_status_watcher(thread_id, &status);
+    }
+
+    fn update_status_watcher(&mut self, thread_id: &str, status: &ThreadStatus) {
+        let remove_watcher = if let Some(sender) = self.status_watcher_by_thread_id.get(thread_id) {
+            let status = status.clone();
+            let _ = sender.send_if_modified(|current| {
+                if *current == status {
+                    false
+                } else {
+                    *current = status;
+                    true
+                }
+            });
+            sender.receiver_count() == 0
+        } else {
+            false
+        };
+        if remove_watcher {
+            self.status_watcher_by_thread_id.remove(thread_id);
+        }
+    }
+
    fn status_changed_notification(
        &self,
        thread_id: String,
@@ -752,6 +798,55 @@ mod tests {
        );
    }

+    #[tokio::test]
+    async fn status_watchers_receive_only_their_thread_updates() {
+        let manager = ThreadWatchManager::new();
+        manager
+            .upsert_thread(test_thread(
+                INTERACTIVE_THREAD_ID,
+                codex_app_server_protocol::SessionSource::Cli,
+            ))
+            .await;
+        manager
+            .upsert_thread(test_thread(
+                NON_INTERACTIVE_THREAD_ID,
+                codex_app_server_protocol::SessionSource::AppServer,
+            ))
+            .await;
+        let interactive_thread_id = ThreadId::from_string(INTERACTIVE_THREAD_ID)
+            .expect("interactive thread id should parse");
+        let non_interactive_thread_id = ThreadId::from_string(NON_INTERACTIVE_THREAD_ID)
+            .expect("non-interactive thread id should parse");
+        let mut interactive_rx = manager
+            .subscribe(interactive_thread_id)
+            .await
+            .expect("interactive status watcher should subscribe");
+        let mut non_interactive_rx = manager
+            .subscribe(non_interactive_thread_id)
+            .await
+            .expect("non-interactive status watcher should subscribe");
+
+        manager.note_turn_started(INTERACTIVE_THREAD_ID).await;
+
+        timeout(Duration::from_secs(1), interactive_rx.changed())
+            .await
+            .expect("timed out waiting for interactive status update")
+            .expect("interactive status watcher should remain open");
+        assert_eq!(
+            *interactive_rx.borrow(),
+            ThreadStatus::Active {
+                active_flags: vec![],
+            },
+        );
+        assert!(
+            timeout(Duration::from_millis(100), non_interactive_rx.changed())
+                .await
+                .is_err(),
+            "unrelated thread watcher should not receive an update"
+        );
+        assert_eq!(*non_interactive_rx.borrow(), ThreadStatus::Idle);
+    }
+
    async fn wait_for_status(
        manager: &ThreadWatchManager,
        thread_id: &str,
@@ -338,7 +338,8 @@ async fn websocket_transport_allows_unauthenticated_non_loopback_startup_by_defa
 }

 #[tokio::test]
-async fn websocket_disconnect_unloads_last_subscribed_thread() -> Result<()> {
+async fn websocket_disconnect_keeps_last_subscribed_thread_loaded_until_idle_timeout() -> Result<()>
+{
    let server = create_mock_responses_server_sequence_unchecked(Vec::new()).await;
    let codex_home = TempDir::new()?;
    create_config_toml(codex_home.path(), &server.uri(), "never")?;
@@ -359,7 +360,7 @@ async fn websocket_disconnect_unloads_last_subscribed_thread() -> Result<()> {
    send_initialize_request(&mut ws2, /*id*/ 4, "ws_reconnect_client").await?;
    read_response_for_id(&mut ws2, /*id*/ 4).await?;

-    wait_for_loaded_threads(&mut ws2, /*first_id*/ 5, &[]).await?;
+    wait_for_loaded_threads(&mut ws2, /*first_id*/ 5, &[thread_id.as_str()]).await?;

    process
        .kill()
@@ -7,10 +7,8 @@ use app_test_support::create_mock_responses_server_sequence_unchecked;
 use app_test_support::create_shell_command_sse_response;
 use app_test_support::to_response;
 use codex_app_server_protocol::ItemStartedNotification;
-use codex_app_server_protocol::JSONRPCNotification;
 use codex_app_server_protocol::JSONRPCResponse;
 use codex_app_server_protocol::RequestId;
-use codex_app_server_protocol::ServerNotification;
 use codex_app_server_protocol::ThreadItem;
 use codex_app_server_protocol::ThreadLoadedListParams;
 use codex_app_server_protocol::ThreadLoadedListResponse;
@@ -21,7 +19,6 @@ use codex_app_server_protocol::ThreadResumeResponse;
 use codex_app_server_protocol::ThreadStartParams;
 use codex_app_server_protocol::ThreadStartResponse;
 use codex_app_server_protocol::ThreadStatus;
-use codex_app_server_protocol::ThreadStatusChangedNotification;
 use codex_app_server_protocol::ThreadUnsubscribeParams;
 use codex_app_server_protocol::ThreadUnsubscribeResponse;
 use codex_app_server_protocol::ThreadUnsubscribeStatus;
@@ -81,7 +78,7 @@ async fn wait_for_responses_request_count_to_stabilize(
 }

 #[tokio::test]
-async fn thread_unsubscribe_unloads_thread_and_emits_thread_closed_notification() -> Result<()> {
+async fn thread_unsubscribe_keeps_thread_loaded_until_idle_timeout() -> Result<()> {
    let server = create_mock_responses_server_repeating_assistant("Done").await;
    let codex_home = TempDir::new()?;
    create_config_toml(codex_home.path(), &server.uri())?;
@@ -104,20 +101,14 @@ async fn thread_unsubscribe_unloads_thread_and_emits_thread_closed_notification(
    let unsubscribe = to_response::<ThreadUnsubscribeResponse>(unsubscribe_resp)?;
    assert_eq!(unsubscribe.status, ThreadUnsubscribeStatus::Unsubscribed);

-    let closed_notif: JSONRPCNotification = timeout(
-        DEFAULT_READ_TIMEOUT,
-        mcp.read_stream_until_notification_message("thread/closed"),
-    )
-    .await??;
-    let parsed: ServerNotification = closed_notif.try_into()?;
-    let ServerNotification::ThreadClosed(payload) = parsed else {
-        anyhow::bail!("expected thread/closed notification");
-    };
-    assert_eq!(payload.thread_id, thread_id);
-
-    let status_changed = wait_for_thread_status_not_loaded(&mut mcp, &payload.thread_id).await?;
-    assert_eq!(status_changed.thread_id, payload.thread_id);
-    assert_eq!(status_changed.status, ThreadStatus::NotLoaded);
+    assert!(
+        timeout(
+            std::time::Duration::from_millis(250),
+            mcp.read_stream_until_notification_message("thread/closed"),
+        )
+        .await
+        .is_err()
+    );

    let list_id = mcp
        .send_thread_loaded_list_request(ThreadLoadedListParams::default())
@@ -129,22 +120,22 @@ async fn thread_unsubscribe_unloads_thread_and_emits_thread_closed_notification(
    .await??;
    let ThreadLoadedListResponse { data, next_cursor } =
        to_response::<ThreadLoadedListResponse>(list_resp)?;
-    assert_eq!(data, Vec::<String>::new());
+    assert_eq!(data, vec![thread_id]);
    assert_eq!(next_cursor, None);

    Ok(())
 }

 #[tokio::test]
-async fn thread_unsubscribe_during_turn_interrupts_turn_and_emits_thread_closed() -> Result<()> {
+async fn thread_unsubscribe_during_turn_keeps_turn_running() -> Result<()> {
    #[cfg(target_os = "windows")]
    let shell_command = vec![
        "powershell".to_string(),
        "-Command".to_string(),
-        "Start-Sleep -Seconds 10".to_string(),
+        "Start-Sleep -Seconds 1".to_string(),
    ];
    #[cfg(not(target_os = "windows"))]
-    let shell_command = vec!["sleep".to_string(), "10".to_string()];
+    let shell_command = vec!["sleep".to_string(), "1".to_string()];

    let tmp = TempDir::new()?;
    let codex_home = tmp.path().join("codex_home");
@@ -206,20 +197,18 @@ async fn thread_unsubscribe_during_turn_interrupts_turn_and_emits_thread_closed(
    let unsubscribe = to_response::<ThreadUnsubscribeResponse>(unsubscribe_resp)?;
    assert_eq!(unsubscribe.status, ThreadUnsubscribeStatus::Unsubscribed);

-    let closed_notif: JSONRPCNotification = timeout(
-        DEFAULT_READ_TIMEOUT,
-        mcp.read_stream_until_notification_message("thread/closed"),
-    )
-    .await??;
-    let parsed: ServerNotification = closed_notif.try_into()?;
-    let ServerNotification::ThreadClosed(payload) = parsed else {
-        anyhow::bail!("expected thread/closed notification");
-    };
-    assert_eq!(payload.thread_id, thread_id);
+    assert!(
+        timeout(
+            std::time::Duration::from_millis(250),
+            mcp.read_stream_until_notification_message("thread/closed"),
+        )
+        .await
+        .is_err()
+    );

    wait_for_responses_request_count_to_stabilize(
        &server,
-        /*expected_count*/ 1,
+        /*expected_count*/ 2,
        std::time::Duration::from_millis(200),
    )
    .await?;
@@ -228,7 +217,7 @@ async fn thread_unsubscribe_during_turn_interrupts_turn_and_emits_thread_closed(
 }

 #[tokio::test]
-async fn thread_unsubscribe_clears_cached_status_before_resume() -> Result<()> {
+async fn thread_unsubscribe_preserves_cached_status_before_idle_unload() -> Result<()> {
    let server = responses::start_mock_server().await;
    let _response_mock = responses::mount_sse_once(
        &server,
@@ -291,11 +280,14 @@ async fn thread_unsubscribe_clears_cached_status_before_resume() -> Result<()> {
    .await??;
    let unsubscribe = to_response::<ThreadUnsubscribeResponse>(unsubscribe_resp)?;
    assert_eq!(unsubscribe.status, ThreadUnsubscribeStatus::Unsubscribed);
-    timeout(
-        DEFAULT_READ_TIMEOUT,
-        mcp.read_stream_until_notification_message("thread/closed"),
-    )
-    .await??;
+    assert!(
+        timeout(
+            std::time::Duration::from_millis(250),
+            mcp.read_stream_until_notification_message("thread/closed"),
+        )
+        .await
+        .is_err()
+    );

    let resume_id = mcp
        .send_thread_resume_request(ThreadResumeParams {
@@ -309,13 +301,13 @@ async fn thread_unsubscribe_clears_cached_status_before_resume() -> Result<()> {
    )
    .await??;
    let resume: ThreadResumeResponse = to_response::<ThreadResumeResponse>(resume_resp)?;
-    assert_eq!(resume.thread.status, ThreadStatus::Idle);
+    assert_eq!(resume.thread.status, ThreadStatus::SystemError);

    Ok(())
 }

 #[tokio::test]
-async fn thread_unsubscribe_reports_not_loaded_after_thread_is_unloaded() -> Result<()> {
+async fn thread_unsubscribe_reports_not_subscribed_before_idle_unload() -> Result<()> {
    let server = create_mock_responses_server_repeating_assistant("Done").await;
    let codex_home = TempDir::new()?;
    create_config_toml(codex_home.path(), &server.uri())?;
@@ -341,12 +333,6 @@ async fn thread_unsubscribe_reports_not_loaded_after_thread_is_unloaded() -> Res
        ThreadUnsubscribeStatus::Unsubscribed
    );

-    timeout(
-        DEFAULT_READ_TIMEOUT,
-        mcp.read_stream_until_notification_message("thread/closed"),
-    )
-    .await??;
-
    let second_unsubscribe_id = mcp
        .send_thread_unsubscribe_request(ThreadUnsubscribeParams { thread_id })
        .await?;
@@ -358,7 +344,7 @@ async fn thread_unsubscribe_reports_not_loaded_after_thread_is_unloaded() -> Res
    let second_unsubscribe = to_response::<ThreadUnsubscribeResponse>(second_unsubscribe_resp)?;
    assert_eq!(
        second_unsubscribe.status,
-        ThreadUnsubscribeStatus::NotLoaded
+        ThreadUnsubscribeStatus::NotSubscribed
    );

    Ok(())
@@ -377,28 +363,6 @@ async fn wait_for_command_execution_item_started(mcp: &mut McpProcess) -> Result
    }
 }

-async fn wait_for_thread_status_not_loaded(
-    mcp: &mut McpProcess,
-    thread_id: &str,
-) -> Result<ThreadStatusChangedNotification> {
-    loop {
-        let status_changed_notif: JSONRPCNotification = timeout(
-            DEFAULT_READ_TIMEOUT,
-            mcp.read_stream_until_notification_message("thread/status/changed"),
-        )
-        .await??;
-        let status_changed_params = status_changed_notif
-            .params
-            .context("thread/status/changed params must be present")?;
-        let status_changed: ThreadStatusChangedNotification =
-            serde_json::from_value(status_changed_params)?;
-        if status_changed.thread_id == thread_id && status_changed.status == ThreadStatus::NotLoaded
-        {
-            return Ok(status_changed);
-        }
-    }
-}
-
 fn create_config_toml(codex_home: &std::path::Path, server_uri: &str) -> std::io::Result<()> {
    let config_toml = codex_home.join("config.toml");
    std::fs::write(