Add a manager for MCP event streams (#41906)

## What changed - Add `McpEventStreamManager` to own event stream workers by thread and subscription. - Wait for activation before completing a start, tag updates with a stream attempt ID, and keep active streams alive independently of their originating runtime. - Support cancellation and shutdown without depending on output queue capacity, and close streams when MCP access changes or an event server is removed. - Reconnect reloaded threads to surviving streams so server removal still cancels them. ## Testing Add coverage for activation, replacement attempts, task reloads, cancellation before activation, full output queues, access changes, and shutdown. GitOrigin-RevId: b7156bbe62f9375b2138eb62868f052124e2a884
2026-10-18 17:18:52 +00:00 · 2026-08-31 20:31:29 +00:00
parent 13bc770eaf
commit ae0ecd7484
5 changed files with 661 additions and 3 deletions
@@ -13,6 +13,7 @@ doctest = false
 workspace = true

 [dependencies]
+anyhow = { workspace = true }
 codex-core = { workspace = true }
 codex-core-plugins = { workspace = true }
 codex-config = { workspace = true }
@@ -32,7 +33,6 @@ tracing = { workspace = true }
 tokio = { workspace = true, features = ["sync"] }

 [dev-dependencies]
-anyhow = { workspace = true }
 axum = { workspace = true }
 codex-exec-server-test-support = { workspace = true }
 codex-login = { workspace = true }
@@ -70,6 +70,7 @@ pub(super) struct Fixture {
    _home: tempfile::TempDir,
    _server: tokio_util::task::AbortOnDropHandle<std::io::Result<()>>,
    pub runtime: Arc<McpRuntime>,
+    pub runtime_input: McpRuntimeInput,
    pub auth: Arc<AuthManager>,
    pub opened: mpsc::UnboundedReceiver<TestStream>,
 }
@@ -153,7 +154,7 @@ impl Fixture {
        ));
        config.mcp_server_catalog = catalog.build();
        let config = Arc::new(config);
-        let runtime_input = McpRuntimeInput {
+        let runtime_input = || McpRuntimeInput {
            startup_policy: McpStartupPolicy::Eager,
            config: Arc::clone(&config),
            plugins_available: false,
@@ -178,7 +179,8 @@ impl Fixture {
            elicitation_reviewer: None,
            elicitation_lifecycle: None,
        };
-        let runtime = Arc::new(McpRuntime::new(runtime_input).await);
+        let runtime = Arc::new(McpRuntime::new(runtime_input()).await);
+        let runtime_input = runtime_input();
        assert!(
            runtime
                .latest_wait_for_server_ready(CODEX_APPS_MCP_SERVER_NAME, WAIT)
@@ -188,6 +190,7 @@ impl Fixture {
            _home: home,
            _server: tokio_util::task::AbortOnDropHandle::new(server),
            runtime,
+            runtime_input,
            auth,
            opened,
        })
@@ -11,6 +11,14 @@ use codex_mcp::hosted_plugin_runtime_mcp_server_config;
 #[path = "event_stream_tests.rs"]
 mod event_stream_tests;
 mod executor_plugin;
+mod stream_manager;
+
+pub use stream_manager::McpEventStreamManager;
+pub use stream_manager::McpEventStreamUpdate;
+
+#[cfg(test)]
+#[path = "stream_manager_tests.rs"]
+mod stream_manager_tests;

 struct HostedPluginRuntimeExtension;

@@ -0,0 +1,290 @@
+//! Owns task-scoped MCP event streams without retaining their thread or runtime.
+//! Each subscription has one worker, and cancellation never waits for output capacity.
+
+use std::collections::HashMap;
+use std::sync::Arc;
+use std::sync::Mutex;
+use std::sync::PoisonError;
+use std::sync::atomic::AtomicBool;
+use std::sync::atomic::AtomicU64;
+use std::sync::atomic::Ordering;
+use std::time::Duration;
+
+use anyhow::Result;
+use anyhow::anyhow;
+use codex_extension_api::ExtensionFuture;
+use codex_extension_api::ThreadLifecycleContributor;
+use codex_extension_api::ThreadReadyInput;
+use codex_extension_api::ThreadStartInput;
+use codex_mcp::McpEventNotification;
+use codex_mcp::McpEventStreamOpener;
+use codex_mcp::McpResourceClient;
+use codex_protocol::ThreadId;
+use serde_json::Map;
+use serde_json::Value;
+use tokio::sync::mpsc;
+use tokio::sync::oneshot;
+use tokio::sync::watch;
+use tokio::task::JoinHandle;
+use tokio::time::Instant;
+use tokio::time::timeout_at;
+
+const EVENT_UPDATE_CAPACITY: usize = 64;
+const ACTIVATION_TIMEOUT: Duration = Duration::from_secs(/*secs*/ 90);
+
+/// A notification or stream ending, identified by task, subscription, and stream attempt.
+pub enum McpEventStreamUpdate {
+    Notification {
+        thread_id: ThreadId,
+        subscription_id: String,
+        stream_attempt_id: u64,
+        notification: McpEventNotification,
+    },
+    Ended {
+        thread_id: ThreadId,
+        subscription_id: String,
+        stream_attempt_id: u64,
+        result: Result<()>,
+    },
+}
+
+struct ManagedEventStream {
+    stream_attempt_id: u64,
+    active: Arc<AtomicBool>,
+    worker: JoinHandle<()>,
+    cancel_event_streams_on_server_removal: watch::Sender<()>,
+}
+
+/// Tracks event streams by thread and subscription ID.
+///
+/// Notifications and stream endings go into a queue. Cancelling a
+/// stream does not require adding an `Ended` update, so it still works when the queue is full.
+pub struct McpEventStreamManager {
+    streams: Mutex<HashMap<(ThreadId, String), ManagedEventStream>>,
+    next_stream_attempt_id: AtomicU64,
+    updates: mpsc::Sender<McpEventStreamUpdate>,
+    shutdown: AtomicBool,
+}
+
+impl McpEventStreamManager {
+    pub fn new() -> (Self, mpsc::Receiver<McpEventStreamUpdate>) {
+        let (updates, receiver) = mpsc::channel(EVENT_UPDATE_CAPACITY);
+        (
+            Self {
+                streams: Mutex::default(),
+                next_stream_attempt_id: AtomicU64::new(/*v*/ 1),
+                updates,
+                shutdown: AtomicBool::new(/*v*/ false),
+            },
+            receiver,
+        )
+    }
+
+    /// Opens a stream and returns its attempt ID after the server confirms activation.
+    pub async fn start(
+        &self,
+        thread_id: ThreadId,
+        subscription_id: String,
+        opener: McpEventStreamOpener,
+        event_name: String,
+        arguments: Value,
+        request_meta: Option<Map<String, Value>>,
+    ) -> Result<u64> {
+        let key = (thread_id, subscription_id.clone());
+        let (ready_tx, ready_rx) = oneshot::channel();
+        {
+            let mut streams = self.streams.lock().unwrap_or_else(PoisonError::into_inner);
+            if self.shutdown.load(Ordering::Acquire) {
+                return Err(anyhow!("MCP event stream manager is shut down"));
+            }
+            streams.retain(|_, stream| !stream.worker.is_finished());
+            if let Some(stream) = streams.get(&key) {
+                if stream.active.load(Ordering::Acquire) {
+                    return Ok(stream.stream_attempt_id);
+                }
+                return Err(anyhow!(
+                    "MCP event subscription is already starting or finishing"
+                ));
+            }
+
+            let stream_attempt_id = self
+                .next_stream_attempt_id
+                .fetch_add(/*val*/ 1, Ordering::Relaxed);
+            let active = Arc::new(AtomicBool::new(/*v*/ false));
+            let worker_active = Arc::clone(&active);
+            let cancel_event_streams_on_server_removal = opener.event_stream_cancellation_sender();
+            let updates = self.updates.clone();
+            let deadline = Instant::now() + ACTIVATION_TIMEOUT;
+            // Keep the stream running if the caller stops waiting for activation.
+            let worker = tokio::spawn(async move {
+                let mut ready = Some(ready_tx);
+                let forward = async {
+                    let mut stream = timeout_at(
+                        deadline,
+                        opener.open(&event_name, &arguments, request_meta.as_ref()),
+                    )
+                    .await
+                    .map_err(|_| anyhow!("MCP event stream activation timed out"))??;
+
+                    loop {
+                        let notification = if ready.is_some() {
+                            timeout_at(deadline, stream.recv())
+                                .await
+                                .map_err(|_| anyhow!("MCP event stream activation timed out"))??
+                        } else {
+                            stream.recv().await?
+                        };
+                        let Some(notification) = notification else {
+                            return Ok(());
+                        };
+
+                        let activates = notification.method == "notifications/events/active";
+                        let terminates = notification.method == "notifications/events/terminated";
+                        if activates {
+                            worker_active.store(/*val*/ true, Ordering::Release);
+                            if let Some(ready) = ready.take() {
+                                let _ = ready.send(Ok(stream_attempt_id));
+                            }
+                        }
+
+                        let send = updates.send(McpEventStreamUpdate::Notification {
+                            thread_id,
+                            subscription_id: subscription_id.clone(),
+                            stream_attempt_id,
+                            notification,
+                        });
+                        let sent = if ready.is_some() {
+                            timeout_at(deadline, send)
+                                .await
+                                .map_err(|_| anyhow!("MCP event stream activation timed out"))?
+                        } else {
+                            send.await
+                        };
+                        sent.map_err(|_| anyhow!("MCP event stream receiver was dropped"))?;
+
+                        if terminates {
+                            return Ok(());
+                        }
+                    }
+                };
+                let result = tokio::select! {
+                    biased;
+                    () = opener.wait_for_access_change() => {
+                        Err(anyhow!("MCP event stream access changed"))
+                    }
+                    () = updates.closed() => {
+                        Err(anyhow!("MCP event stream receiver was dropped"))
+                    }
+                    result = forward => result,
+                };
+
+                worker_active.store(/*val*/ false, Ordering::Release);
+                drop(opener);
+                if let Some(ready) = ready {
+                    let error = match &result {
+                        Ok(()) => anyhow!("MCP event stream ended before becoming active"),
+                        Err(error) => anyhow!("{error:#}"),
+                    };
+                    let _ = ready.send(Err(error));
+                }
+
+                let _ = updates
+                    .send(McpEventStreamUpdate::Ended {
+                        thread_id,
+                        subscription_id,
+                        stream_attempt_id,
+                        result,
+                    })
+                    .await;
+            });
+            streams.insert(
+                key,
+                ManagedEventStream {
+                    stream_attempt_id,
+                    active,
+                    worker,
+                    cancel_event_streams_on_server_removal,
+                },
+            );
+        }
+
+        ready_rx
+            .await
+            .map_err(|_| anyhow!("MCP event stream ended before becoming active"))?
+    }
+
+    /// Await cancellation before starting the same subscription again.
+    pub async fn cancel(&self, thread_id: ThreadId, subscription_id: &str) {
+        let stream = self
+            .streams
+            .lock()
+            .unwrap_or_else(PoisonError::into_inner)
+            .remove(&(thread_id, subscription_id.to_owned()));
+        if let Some(stream) = stream {
+            stream.worker.abort();
+            let _ = stream.worker.await;
+        }
+    }
+
+    /// Permanently rejects new starts and waits for all current workers to release their streams.
+    pub async fn shutdown(&self) {
+        self.shutdown.store(/*val*/ true, Ordering::Release);
+        let streams = self
+            .streams
+            .lock()
+            .unwrap_or_else(PoisonError::into_inner)
+            .drain()
+            .map(|(_, stream)| stream)
+            .collect::<Vec<_>>();
+        for stream in &streams {
+            stream.worker.abort();
+        }
+        for stream in streams {
+            let _ = stream.worker.await;
+        }
+    }
+}
+
+impl<C: Sync> ThreadLifecycleContributor<C> for McpEventStreamManager {
+    fn on_thread_start<'a>(&'a self, input: ThreadStartInput<'a, C>) -> ExtensionFuture<'a, ()> {
+        Box::pin(async move {
+            if let Some(client) = input.mcp_resource_client {
+                input.thread_store.insert(client.as_ref().clone());
+            }
+        })
+    }
+
+    fn on_thread_ready<'a>(&'a self, input: ThreadReadyInput<'a, C>) -> ExtensionFuture<'a, ()> {
+        Box::pin(async move {
+            let Some(client) = input.thread_store.get::<McpResourceClient>() else {
+                return;
+            };
+            let Ok(thread_id) = ThreadId::from_string(input.thread_store.level_id()) else {
+                return;
+            };
+            let cancellation = self
+                .streams
+                .lock()
+                .unwrap_or_else(PoisonError::into_inner)
+                .iter()
+                .find(|((owner, _), stream)| {
+                    let belongs_to_task = *owner == thread_id;
+                    let worker_is_running = !stream.worker.is_finished();
+                    belongs_to_task && worker_is_running
+                })
+                .map(|(_, stream)| stream.cancel_event_streams_on_server_removal.clone());
+            if let Some(cancellation) = cancellation {
+                client.forward_event_server_removals_to(cancellation);
+            }
+        })
+    }
+}
+
+impl Drop for McpEventStreamManager {
+    fn drop(&mut self) {
+        let mut streams = self.streams.lock().unwrap_or_else(PoisonError::into_inner);
+        for (_, stream) in streams.drain() {
+            stream.worker.abort();
+        }
+    }
+}
@@ -0,0 +1,357 @@
+//! Exercises stream activation, task ownership, and cancellation.
+#![allow(clippy::unwrap_used)]
+
+use anyhow::Result;
+use codex_extension_api::ExtensionData;
+use codex_extension_api::ThreadLifecycleContributor;
+use codex_extension_api::ThreadReadyInput;
+use codex_extension_api::ThreadStartInput;
+use codex_login::CodexAuth;
+use codex_mcp::McpEventStreamOpener;
+use codex_mcp::McpResourceClient;
+use codex_protocol::ThreadId;
+use codex_protocol::protocol::SessionSource;
+use pretty_assertions::assert_eq;
+use serde_json::json;
+use std::sync::Arc;
+use tokio::time::timeout;
+
+use crate::McpEventStreamManager;
+use crate::McpEventStreamUpdate;
+use crate::event_stream_tests::Fixture;
+use crate::event_stream_tests::StaticAuth;
+use crate::event_stream_tests::WAIT;
+
+async fn start_subscription(
+    manager: &McpEventStreamManager,
+    opener: &McpEventStreamOpener,
+    thread_id: ThreadId,
+) -> Result<u64> {
+    manager
+        .start(
+            thread_id,
+            "sub".into(),
+            opener.clone(),
+            "test.event".into(),
+            json!({}),
+            /*request_meta*/ None,
+        )
+        .await
+}
+
+async fn connect_loaded_task(
+    manager: &McpEventStreamManager,
+    thread_id: ThreadId,
+    client: Arc<McpResourceClient>,
+) {
+    let session_store = ExtensionData::new("test");
+    let thread_store = ExtensionData::new(thread_id.to_string());
+    manager
+        .on_thread_start(ThreadStartInput {
+            config: &(),
+            session_source: &SessionSource::Exec,
+            persistent_thread_state_available: false,
+            environments: &[],
+            mcp_resource_client: Some(client),
+            extension_metrics: None,
+            session_store: &session_store,
+            thread_store: &thread_store,
+        })
+        .await;
+    manager
+        .on_thread_ready(ThreadReadyInput {
+            config: &(),
+            session_source: &SessionSource::Exec,
+            session_store: &session_store,
+            thread_store: &thread_store,
+        })
+        .await;
+}
+
+#[tokio::test]
+async fn reloaded_tasks_can_cancel_surviving_streams() -> Result<()> {
+    for removed_before_ready in [false, true] {
+        let mut fixture = Fixture::new().await?;
+        let opener = McpResourceClient::new(Arc::clone(&fixture.runtime)).event_stream_opener()?;
+        let (manager, mut updates) = McpEventStreamManager::new();
+        let thread_id = ThreadId::new();
+        let (result, stream) =
+            tokio::join!(start_subscription(&manager, &opener, thread_id), async {
+                let stream = timeout(WAIT, fixture.opened.recv()).await.unwrap().unwrap();
+                stream.notify("notifications/events/active");
+                stream
+            });
+        result?;
+        timeout(WAIT, updates.recv()).await?.unwrap();
+        fixture.runtime.shutdown().await;
+        drop(fixture.runtime);
+
+        let mut replacement = Fixture::new().await?;
+        let client = Arc::new(McpResourceClient::new(Arc::clone(&replacement.runtime)));
+        replacement.runtime_input.mcp_servers.clear();
+        if removed_before_ready {
+            replacement.runtime.replace(replacement.runtime_input).await;
+            connect_loaded_task(&manager, thread_id, client).await;
+        } else {
+            connect_loaded_task(&manager, thread_id, client).await;
+            let expected = stream.notify("notifications/events/event");
+            let McpEventStreamUpdate::Notification { notification, .. } =
+                timeout(WAIT, updates.recv()).await?.unwrap()
+            else {
+                panic!("event expected after task reload");
+            };
+            assert_eq!(notification, expected);
+            replacement.runtime.replace(replacement.runtime_input).await;
+        }
+
+        timeout(WAIT, stream.notifications.closed()).await?;
+        let McpEventStreamUpdate::Ended { result, .. } =
+            timeout(WAIT, updates.recv()).await?.unwrap()
+        else {
+            panic!("stream should end after the event server is removed");
+        };
+        assert!(result.is_err());
+        manager.shutdown().await;
+    }
+    Ok(())
+}
+
+#[tokio::test]
+async fn streams_outlive_tasks_and_activate_with_full_output() -> Result<()> {
+    let mut fixture = Fixture::new().await?;
+    let opener = McpResourceClient::new(Arc::clone(&fixture.runtime)).event_stream_opener()?;
+    let (manager, mut updates) = McpEventStreamManager::new();
+    let first_thread = ThreadId::new();
+    let second_thread = ThreadId::new();
+    let mut streams = Vec::new();
+    for thread_id in [first_thread, second_thread] {
+        let start = manager.start(
+            thread_id,
+            "sub".into(),
+            opener.clone(),
+            "test.event".into(),
+            json!({"resource_id": "resource-123"}),
+            Some(serde_json::Map::from_iter([(
+                "saved_id".into(),
+                json!("sub"),
+            )])),
+        );
+        tokio::pin!(start);
+        let stream = tokio::select! {
+            result = &mut start => panic!("start returned before activation: {result:?}"),
+            stream = timeout(WAIT, fixture.opened.recv()) => stream?.unwrap(),
+        };
+        let mut params = stream.request["params"].clone();
+        params["_meta"]
+            .as_object_mut()
+            .unwrap()
+            .remove("progressToken");
+        assert_eq!(
+            params,
+            json!({"name": "test.event", "arguments": {"resource_id": "resource-123"}, "_meta": {"saved_id": "sub"}})
+        );
+        stream.notify("notifications/events/active");
+        timeout(WAIT, start).await??;
+        timeout(WAIT, updates.recv()).await?.unwrap();
+        streams.push(stream);
+    }
+    start_subscription(&manager, &opener, first_thread).await?;
+    assert!(fixture.opened.try_recv().is_err());
+
+    fixture.runtime.shutdown().await;
+    let expected = streams[1].notify("notifications/events/event");
+    let McpEventStreamUpdate::Notification {
+        thread_id,
+        subscription_id,
+        notification,
+        ..
+    } = timeout(WAIT, updates.recv()).await?.unwrap()
+    else {
+        panic!("event expected");
+    };
+    assert_eq!(
+        (thread_id, subscription_id, notification),
+        (second_thread, "sub".into(), expected)
+    );
+    for _ in 0..=updates.max_capacity() {
+        streams[0].notify("notifications/events/event");
+    }
+    timeout(WAIT, async {
+        while updates.capacity() != 0 {
+            tokio::task::yield_now().await;
+        }
+    })
+    .await?;
+
+    let third_thread = ThreadId::new();
+    let (started, stream) = tokio::join!(
+        timeout(WAIT, start_subscription(&manager, &opener, third_thread)),
+        async {
+            let stream = timeout(WAIT, fixture.opened.recv()).await.unwrap().unwrap();
+            stream.notify("notifications/events/active");
+            stream
+        }
+    );
+    started??;
+    assert_eq!(updates.capacity(), 0);
+    manager.cancel(third_thread, "sub").await;
+    timeout(WAIT, stream.notifications.closed()).await?;
+
+    manager.cancel(first_thread, "sub").await;
+    timeout(WAIT, streams[0].notifications.closed()).await?;
+    manager.shutdown().await;
+    timeout(WAIT, streams[1].notifications.closed()).await?;
+
+    Ok(())
+}
+
+#[tokio::test]
+async fn queued_updates_keep_their_stream_attempt_after_replacement() -> Result<()> {
+    let mut fixture = Fixture::new().await?;
+    let opener = McpResourceClient::new(Arc::clone(&fixture.runtime)).event_stream_opener()?;
+    let (manager, mut updates) = McpEventStreamManager::new();
+    let thread_id = ThreadId::new();
+    let (first_attempt, stream) =
+        tokio::join!(start_subscription(&manager, &opener, thread_id), async {
+            let stream = timeout(WAIT, fixture.opened.recv()).await.unwrap().unwrap();
+            stream.notify("notifications/events/active");
+            stream
+        });
+    let first_attempt = first_attempt?;
+    assert_eq!(
+        start_subscription(&manager, &opener, thread_id).await?,
+        first_attempt
+    );
+
+    stream.notify("notifications/events/terminated");
+    // Leave activation, termination, and the ending queued while opening a replacement.
+    timeout(WAIT, async {
+        while updates.len() < 3 {
+            tokio::task::yield_now().await;
+        }
+    })
+    .await?;
+    manager.cancel(thread_id, "sub").await;
+
+    let (second_attempt, replacement) =
+        tokio::join!(start_subscription(&manager, &opener, thread_id), async {
+            let stream = timeout(WAIT, fixture.opened.recv()).await.unwrap().unwrap();
+            stream.notify("notifications/events/active");
+            stream
+        });
+    let second_attempt = second_attempt?;
+    assert_ne!(first_attempt, second_attempt);
+
+    let mut queued_attempts = Vec::new();
+    for _ in 0..4 {
+        let stream_attempt_id = match timeout(WAIT, updates.recv()).await?.unwrap() {
+            McpEventStreamUpdate::Notification {
+                stream_attempt_id, ..
+            } => stream_attempt_id,
+            McpEventStreamUpdate::Ended {
+                stream_attempt_id,
+                result,
+                ..
+            } => {
+                result?;
+                assert_eq!(stream_attempt_id, first_attempt);
+                stream_attempt_id
+            }
+        };
+        queued_attempts.push(stream_attempt_id);
+    }
+    assert_eq!(
+        queued_attempts,
+        vec![first_attempt, first_attempt, first_attempt, second_attempt]
+    );
+    manager.shutdown().await;
+    timeout(WAIT, replacement.notifications.closed()).await?;
+    Ok(())
+}
+
+#[tokio::test]
+async fn cancelling_before_activation_releases_the_subscription_id() -> Result<()> {
+    let mut fixture = Fixture::new().await?;
+    let opener = McpResourceClient::new(Arc::clone(&fixture.runtime)).event_stream_opener()?;
+    let (manager, _updates) = McpEventStreamManager::new();
+    let thread_id = ThreadId::new();
+    let start = start_subscription(&manager, &opener, thread_id);
+    tokio::pin!(start);
+    let stream = tokio::select! {
+        result = &mut start => panic!("start returned before activation: {result:?}"),
+        stream = timeout(WAIT, fixture.opened.recv()) => stream?.unwrap(),
+    };
+    manager.cancel(thread_id, "sub").await;
+    assert!(timeout(WAIT, start).await?.is_err());
+    timeout(WAIT, stream.notifications.closed()).await?;
+    let (result, replacement) =
+        tokio::join!(start_subscription(&manager, &opener, thread_id), async {
+            let stream = timeout(WAIT, fixture.opened.recv()).await.unwrap().unwrap();
+            stream.notify("notifications/events/active");
+            stream
+        });
+    result?;
+    manager.shutdown().await;
+    timeout(WAIT, replacement.notifications.closed()).await?;
+    Ok(())
+}
+
+#[tokio::test]
+async fn access_changes_cancel_streams_with_full_output() -> Result<()> {
+    for remove_server in [false, true] {
+        let mut fixture = Fixture::new().await?;
+        let opener = McpResourceClient::new(Arc::clone(&fixture.runtime)).event_stream_opener()?;
+        let (manager, mut updates) = McpEventStreamManager::new();
+        let thread_id = ThreadId::new();
+        let (result, stream) =
+            tokio::join!(start_subscription(&manager, &opener, thread_id), async {
+                let stream = timeout(WAIT, fixture.opened.recv()).await.unwrap().unwrap();
+                stream.notify("notifications/events/active");
+                stream
+            });
+        result?;
+        timeout(WAIT, updates.recv()).await?.unwrap();
+        for _ in 0..=updates.max_capacity() {
+            stream.notify("notifications/events/event");
+        }
+        timeout(WAIT, async {
+            while updates.capacity() != 0 {
+                tokio::task::yield_now().await;
+            }
+        })
+        .await?;
+        if remove_server {
+            fixture.runtime_input.mcp_servers.clear();
+            fixture.runtime.replace(fixture.runtime_input).await;
+        } else {
+            fixture
+                .auth
+                .set_external_auth(Arc::new(StaticAuth(
+                    CodexAuth::from_external_chatgpt_tokens(
+                        "header.e30.other",
+                        "other-account",
+                        /*chatgpt_plan_type*/ None,
+                    )?,
+                )))
+                .await?;
+        }
+        // Access changes close the stream even while its output is blocked.
+        timeout(WAIT, stream.notifications.closed()).await?;
+        let result = timeout(WAIT, async {
+            loop {
+                if let McpEventStreamUpdate::Ended { result, .. } = updates.recv().await.unwrap() {
+                    break result;
+                }
+            }
+        })
+        .await?;
+        assert!(result.is_err());
+        assert!(
+            opener
+                .open("test.event", &json!({}), /*request_meta*/ None)
+                .await
+                .is_err()
+        );
+    }
+    Ok(())
+}