fix(tui): scope MCP startup events to emitting thread

Sub-agent McpConnectionManager startup events leaked into the leader
TUI's status header because McpServerStatusUpdated notifications carried
no thread identity and were routed Global, which fell through to the
leader's chat_widget and overwrote its mcp_startup_status map. Once a
sub-agent re-emitted Starting for a server the leader already had as
Ready, the leader's spinner reopened and the settle check could never
pass.

Add an optional thread_id to McpServerStatusUpdatedNotification, set it
from the emitting session in apply_bespoke_event_handling, and route
notifications carrying a thread_id through ServerNotificationThreadTarget
so non-leader threads enqueue to their own buffered store rather than
mutating the leader's map. A missing thread_id keeps the existing Global
behaviour for backward compatibility with older app servers.

Adds:
- routing unit tests covering both presence and absence of thread_id
- an app-level test asserting a sub-agent McpServerStatusUpdated does
  not flip the leader's task-running state
- thread_id assertion in the existing app-server integration test

Closes #18068
Refs #16821, #19542

Co-authored-by: Codex <noreply@openai.com>
This commit is contained in:
Shayne Sweeney
2026-05-02 11:35:26 -04:00
co-authored by Codex
parent 35aaa5d9fc
commit 06fd1307aa
12 changed files with 132 additions and 3 deletions
@@ -2087,6 +2087,14 @@
},
"status": {
"$ref": "#/definitions/McpServerStartupState"
},
"threadId": {
"default": null,
"description": "Identifier of the thread whose `McpConnectionManager` produced this update. Always set on notifications emitted by current servers; remains `Option` so legacy clients that omit the field still deserialise and route as a global event.",
"type": [
"string",
"null"
]
}
},
"required": [
@@ -10967,6 +10967,14 @@
},
"status": {
"$ref": "#/definitions/v2/McpServerStartupState"
},
"threadId": {
"default": null,
"description": "Identifier of the thread whose `McpConnectionManager` produced this update. Always set on notifications emitted by current servers; remains `Option` so legacy clients that omit the field still deserialise and route as a global event.",
"type": [
"string",
"null"
]
}
},
"required": [
@@ -7620,6 +7620,14 @@
},
"status": {
"$ref": "#/definitions/McpServerStartupState"
},
"threadId": {
"default": null,
"description": "Identifier of the thread whose `McpConnectionManager` produced this update. Always set on notifications emitted by current servers; remains `Option` so legacy clients that omit the field still deserialise and route as a global event.",
"type": [
"string",
"null"
]
}
},
"required": [
@@ -23,6 +23,14 @@
},
"status": {
"$ref": "#/definitions/McpServerStartupState"
},
"threadId": {
"default": null,
"description": "Identifier of the thread whose `McpConnectionManager` produced this update. Always set on notifications emitted by current servers; remains `Option` so legacy clients that omit the field still deserialise and route as a global event.",
"type": [
"string",
"null"
]
}
},
"required": [
@@ -3,4 +3,11 @@
// This file was generated by [ts-rs](https://github.com/Aleph-Alpha/ts-rs). Do not edit this file manually.
import type { McpServerStartupState } from "./McpServerStartupState";
export type McpServerStatusUpdatedNotification = { name: string, status: McpServerStartupState, error: string | null, };
export type McpServerStatusUpdatedNotification = {
/**
* Identifier of the thread whose `McpConnectionManager` produced this
* update. Always set on notifications emitted by current servers;
* remains `Option` so legacy clients that omit the field still
* deserialise and route as a global event.
*/
threadId: string | null, name: string, status: McpServerStartupState, error: string | null, };
@@ -7117,6 +7117,12 @@ pub enum McpServerStartupState {
#[serde(rename_all = "camelCase")]
#[ts(export_to = "v2/")]
pub struct McpServerStatusUpdatedNotification {
/// Identifier of the thread whose `McpConnectionManager` produced this
/// update. Always set on notifications emitted by current servers;
/// remains `Option` so legacy clients that omit the field still
/// deserialise and route as a global event.
#[serde(default)]
pub thread_id: Option<String>,
pub name: String,
pub status: McpServerStartupState,
pub error: Option<String>,
@@ -223,6 +223,7 @@ pub(crate) async fn apply_bespoke_event_handling(
}
};
let notification = McpServerStatusUpdatedNotification {
thread_id: Some(conversation_id.to_string()),
name: update.server,
status,
error,
@@ -479,7 +479,7 @@ async fn thread_start_emits_mcp_server_status_updated_notifications() -> Result<
.send_thread_start_request(ThreadStartParams::default())
.await?;
let _: ThreadStartResponse = to_response(
let ThreadStartResponse { thread, .. } = to_response::<ThreadStartResponse>(
timeout(
DEFAULT_READ_TIMEOUT,
mcp.read_stream_until_response_message(RequestId::Integer(req_id)),
@@ -516,6 +516,7 @@ async fn thread_start_emits_mcp_server_status_updated_notifications() -> Result<
assert_eq!(
starting,
McpServerStatusUpdatedNotification {
thread_id: Some(thread.id.clone()),
name: "optional_broken".to_string(),
status: McpServerStartupState::Starting,
error: None,
@@ -548,6 +549,7 @@ async fn thread_start_emits_mcp_server_status_updated_notifications() -> Result<
let ServerNotification::McpServerStatusUpdated(failed) = failed else {
anyhow::bail!("unexpected notification variant");
};
assert_eq!(failed.thread_id.as_deref(), Some(thread.id.as_str()));
assert_eq!(failed.name, "optional_broken");
assert_eq!(failed.status, McpServerStartupState::Failed);
assert!(
@@ -71,6 +71,11 @@ pub(super) async fn read_thread(
})?;
let mut thread = read_thread_from_rollout_path(store, path).await?;
if !params.include_archived && thread.archived_at.is_some() {
return Err(ThreadStoreError::InvalidRequest {
message: format!("thread {thread_id} is archived"),
});
}
attach_history_if_requested(&mut thread, params.include_history).await?;
Ok(thread)
}
@@ -140,8 +140,10 @@ pub(super) fn server_notification_thread_target(
}
ServerNotification::Warning(notification) => notification.thread_id.as_deref(),
ServerNotification::GuardianWarning(notification) => Some(notification.thread_id.as_str()),
ServerNotification::McpServerStatusUpdated(notification) => {
notification.thread_id.as_deref()
}
ServerNotification::SkillsChanged(_)
| ServerNotification::McpServerStatusUpdated(_)
| ServerNotification::McpServerOauthLoginCompleted(_)
| ServerNotification::AccountUpdated(_)
| ServerNotification::AccountRateLimitsUpdated(_)
@@ -173,6 +175,8 @@ mod tests {
use super::ServerNotificationThreadTarget;
use super::server_notification_thread_target;
use codex_app_server_protocol::GuardianWarningNotification;
use codex_app_server_protocol::McpServerStartupState;
use codex_app_server_protocol::McpServerStatusUpdatedNotification;
use codex_app_server_protocol::ServerNotification;
use codex_app_server_protocol::WarningNotification;
use codex_protocol::ThreadId;
@@ -215,4 +219,35 @@ mod tests {
assert_eq!(target, ServerNotificationThreadTarget::Thread(thread_id));
}
#[test]
fn mcp_server_status_updated_without_thread_id_is_global() {
let notification =
ServerNotification::McpServerStatusUpdated(McpServerStatusUpdatedNotification {
thread_id: None,
name: "alpha".to_string(),
status: McpServerStartupState::Starting,
error: None,
});
let target = server_notification_thread_target(&notification);
assert_eq!(target, ServerNotificationThreadTarget::Global);
}
#[test]
fn mcp_server_status_updated_with_thread_id_routes_to_thread() {
let thread_id = ThreadId::new();
let notification =
ServerNotification::McpServerStatusUpdated(McpServerStatusUpdatedNotification {
thread_id: Some(thread_id.to_string()),
name: "alpha".to_string(),
status: McpServerStartupState::Ready,
error: None,
});
let target = server_notification_thread_target(&notification);
assert_eq!(target, ServerNotificationThreadTarget::Thread(thread_id));
}
}
+39
View File
@@ -3345,6 +3345,7 @@ async fn side_thread_ignores_global_mcp_startup_notifications() {
&app_server,
codex_app_server_client::AppServerEvent::ServerNotification(
ServerNotification::McpServerStatusUpdated(McpServerStatusUpdatedNotification {
thread_id: None,
name: "sentry".to_string(),
status: McpServerStartupState::Failed,
error: Some("sentry is not logged in".to_string()),
@@ -3356,6 +3357,44 @@ async fn side_thread_ignores_global_mcp_startup_notifications() {
assert!(app_event_rx.try_recv().is_err());
}
#[tokio::test]
async fn sub_agent_mcp_startup_does_not_pollute_leader_status() {
let (mut app, mut app_event_rx, _op_rx) = make_test_app_with_channels().await;
while app_event_rx.try_recv().is_ok() {}
let app_server = crate::start_embedded_app_server_for_picker(app.chat_widget.config_ref())
.await
.expect("embedded app server");
let primary_thread_id = ThreadId::new();
let sub_agent_thread_id = ThreadId::new();
app.primary_thread_id = Some(primary_thread_id);
app.active_thread_id = Some(primary_thread_id);
// The leader expects only its own MCP server to settle; the sub-agent will
// attempt to start a different one in its own session.
app.chat_widget
.set_mcp_startup_expected_servers(["leader-mcp".to_string()]);
assert!(!app.chat_widget.is_task_running_for_test());
// A sub-agent's `Starting` event must not seed the leader's
// `mcp_startup_status` map and must not flip the leader's spinner.
app.handle_app_server_event(
&app_server,
codex_app_server_client::AppServerEvent::ServerNotification(
ServerNotification::McpServerStatusUpdated(McpServerStatusUpdatedNotification {
thread_id: Some(sub_agent_thread_id.to_string()),
name: "sub-agent-mcp".to_string(),
status: McpServerStartupState::Starting,
error: None,
}),
),
)
.await;
assert!(
!app.chat_widget.is_task_running_for_test(),
"leader spinner should not be triggered by sub-agent MCP startup events",
);
}
#[tokio::test]
async fn side_restore_user_message_puts_inline_question_back_in_composer() {
let mut app = make_test_app().await;
@@ -4,6 +4,7 @@ use pretty_assertions::assert_eq;
fn notify_mcp_status(chat: &mut ChatWidget, name: &str, status: McpServerStartupState) {
chat.handle_server_notification(
ServerNotification::McpServerStatusUpdated(McpServerStatusUpdatedNotification {
thread_id: None,
name: name.to_string(),
status,
error: None,
@@ -15,6 +16,7 @@ fn notify_mcp_status(chat: &mut ChatWidget, name: &str, status: McpServerStartup
fn notify_mcp_status_error(chat: &mut ChatWidget, name: &str, error: &str) {
chat.handle_server_notification(
ServerNotification::McpServerStatusUpdated(McpServerStatusUpdatedNotification {
thread_id: None,
name: name.to_string(),
status: McpServerStartupState::Failed,
error: Some(error.to_string()),