Skip to content

Restrict TUI MCP startup notifications to owned threads #50781

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
copyberry merged 1 commit into from
Oct 4, 2026
+159 −30
Merged

Restrict TUI MCP startup notifications to owned threads #50781

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions codex-rs/tui/src/app.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -212,6 +212,7 @@ mod activity_groups;
mod app_server_event_targets;
mod app_server_events;
pub(crate) mod app_server_requests;
mod app_server_thread_ownership;
mod backend_banner_fallback;
mod background_requests;
mod composer_hints;
Expand Down
30 changes: 7 additions & 23 deletions codex-rs/tui/src/app/app_server_events.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -460,25 +460,12 @@ impl App {
{
return;
}
if self.primary_thread_id.is_some()
&& self.primary_thread_id != Some(thread_id)
&& !self.thread_event_channels.contains_key(&thread_id)
&& self.agent_navigation.get(&thread_id).is_none()
&& !self.side_threads.contains_key(&thread_id)
&& !matches!(&notification, ServerNotification::McpServerStatusUpdated(_))
&& !matches!(
&notification,
ServerNotification::ThreadStarted(started)
if matches!(
&started.thread.source,
SessionSource::SubAgent(SubAgentSource::ThreadSpawn {
parent_thread_id,
..
}) if self.primary_thread_id == Some(*parent_thread_id)
|| self.thread_event_channels.contains_key(parent_thread_id)
|| self.agent_navigation.get(parent_thread_id).is_some()
)
)
let untracked_thread =
self.primary_thread_id.is_some() && !self.owns_thread_for_routing(thread_id);
if untracked_thread
&& !self
.owns_untracked_notification(app_server_client, thread_id, &notification)
.await
{
return;
}
Expand Down Expand Up @@ -754,10 +741,7 @@ impl App {
else {
return;
};
if self.primary_thread_id != Some(parent_thread_id)
&& !self.thread_event_channels.contains_key(&parent_thread_id)
&& self.agent_navigation.get(&parent_thread_id).is_none()
{
if !self.owns_thread_for_routing(parent_thread_id) {
if self
.agents_overview
.dispatched_requests
Expand Down
83 changes: 83 additions & 0 deletions codex-rs/tui/src/app/app_server_thread_ownership.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,83 @@
//! Ownership checks for routing app-server events to a TUI root.

use super::App;
use crate::app_server_session::AppServerSession;
use codex_app_server_client::TypedRequestError;
use codex_app_server_protocol::ClientRequest;
use codex_app_server_protocol::RequestId;
use codex_app_server_protocol::ServerNotification;
use codex_app_server_protocol::SessionSource;
use codex_app_server_protocol::ThreadReadParams;
use codex_app_server_protocol::ThreadReadResponse;
use codex_protocol::ThreadId;
use codex_protocol::protocol::SubAgentSource;
use std::time::Duration;

async fn read_mcp_subagent_parent_thread_id(
app_server_client: &AppServerSession,
thread_id: ThreadId,
) -> Option<ThreadId> {
let response = tokio::time::timeout(Duration::from_secs(/*secs*/ 1), async {
let mut retry_delay = Duration::from_millis(50);
loop {
match app_server_client
.request_handle()
.request_typed::<ThreadReadResponse>(ClientRequest::ThreadRead {
request_id: RequestId::String(uuid::Uuid::new_v4().to_string()),
params: ThreadReadParams {
thread_id: thread_id.to_string(),
include_turns: false,
},
})
.await
{
Ok(response) => return Some(response),
Err(TypedRequestError::Server { source, .. })
if source.message.starts_with("thread not found:")
|| source.message.starts_with("thread not loaded:") =>
{
tokio::time::sleep(retry_delay).await;
retry_delay = retry_delay.saturating_mul(2).min(Duration::from_secs(1));
}
Err(_) => return None,
}
}
})
.await
.ok()??;
subagent_parent(&response.thread.source)
}

fn subagent_parent(source: &SessionSource) -> Option<ThreadId> {
match source {
SessionSource::SubAgent(SubAgentSource::ThreadSpawn {
parent_thread_id, ..
}) => Some(*parent_thread_id),
_ => None,
}
}

impl App {
pub(super) fn owns_thread_for_routing(&self, thread_id: ThreadId) -> bool {
self.primary_thread_id == Some(thread_id)
|| self.thread_event_channels.contains_key(&thread_id)
|| self.side_threads.contains_key(&thread_id)
|| self.agent_navigation.get(&thread_id).is_some()
}

pub(super) async fn owns_untracked_notification(
&self,
app_server_client: &AppServerSession,
thread_id: ThreadId,
notification: &ServerNotification,
) -> bool {
let parent_thread_id = match notification {
ServerNotification::ThreadStarted(started) => subagent_parent(&started.thread.source),
ServerNotification::McpServerStatusUpdated(_) => {
read_mcp_subagent_parent_thread_id(app_server_client, thread_id).await
}
_ => None,
};
parent_thread_id.is_some_and(|thread_id| self.owns_thread_for_routing(thread_id))
}
}
6 changes: 6 additions & 0 deletions codex-rs/tui/src/app/tests.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -5436,6 +5436,12 @@ async fn primary_thread_ignores_child_mcp_startup_notifications() {
let child_thread_id = ThreadId::new();
app.primary_thread_id = Some(parent_thread_id);
app.active_thread_id = Some(parent_thread_id);
app.upsert_agent_picker_thread(
child_thread_id,
/*agent_nickname*/ None,
/*agent_role*/ None,
/*is_closed*/ false,
);

app.handle_app_server_event(
&app_server,
Expand Down
69 changes: 62 additions & 7 deletions codex-rs/tui/src/app/tests/startup.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1394,14 +1394,35 @@ async fn startup_thread_started_does_not_replay_resolved_approval() -> Result<()
}

#[tokio::test]
async fn owned_subagent_approval_before_thread_started_is_preserved() -> Result<()> {
async fn subagent_approval_respects_root_ownership() -> Result<()> {
check_subagent_approval_routing(ApprovalRouting::OwnedApprovalFirst).await?;
check_subagent_approval_routing(ApprovalRouting::OwnedAfterMcp).await?;
check_subagent_approval_routing(ApprovalRouting::ForeignAfterMcp).await
}

enum ApprovalRouting {
OwnedApprovalFirst,
OwnedAfterMcp,
ForeignAfterMcp,
}

async fn check_subagent_approval_routing(routing: ApprovalRouting) -> Result<()> {
let owned_parent = !matches!(&routing, ApprovalRouting::ForeignAfterMcp);
let (mut app, _app_event_rx, _op_rx) = make_test_app_with_channels().await;
let codex_home = tempdir()?;
app.config.codex_home = codex_home.path().to_path_buf().abs();
app.config.sqlite = codex_state::SqliteConfig::new_for_testing(codex_home.path().abs());
let mut app_server = Box::pin(crate::start_embedded_app_server_for_picker(&app.config)).await?;
let parent = app_server.start_thread(&app.config).await?;
let parent_thread_id = parent.session.thread_id;
let parent_thread_id = if owned_parent {
parent.session.thread_id
} else {
app_server
.start_thread(&app.config)
.await?
.session
.thread_id
};
app.enqueue_primary_thread_session(parent.session, parent.turns)
.await?;
let child_thread_id = ThreadId::from_string(
Expand Down Expand Up @@ -1439,20 +1460,54 @@ async fn owned_subagent_approval_before_thread_started_is_preserved() -> Result<
/*approval_id*/ None,
);

if matches!(&routing, ApprovalRouting::ForeignAfterMcp) {
send_failed_mcp_startup(&mut app, &app_server, parent_thread_id).await;
}
if !matches!(&routing, ApprovalRouting::OwnedApprovalFirst) {
send_failed_mcp_startup(&mut app, &app_server, child_thread_id).await;
let has_channel = app.thread_event_channels.contains_key(&child_thread_id);
assert_eq!(has_channel, owned_parent);
}

app.handle_app_server_event(
&app_server,
codex_app_server_client::AppServerEvent::ServerRequest(Box::new(request.clone())),
)
.await;

assert!(
app.pending_app_server_requests
.contains_server_request(&request)
);
assert!(app.thread_event_channels.contains_key(&child_thread_id));
let requests = &app.pending_app_server_requests;
assert_eq!(requests.contains_server_request(&request), owned_parent);
assert_eq!(app.chat_widget.has_active_view(), owned_parent);
if !owned_parent {
let popup = render_bottom_popup(&app.chat_widget, /*width*/ 80);
let approval = popup
.lines()
.find(|line| line.contains("Would you like to run"));
insta::assert_snapshot!(approval.unwrap_or_default(), @"");
}
Ok(())
}

async fn send_failed_mcp_startup(
app: &mut App,
app_server: &AppServerSession,
thread_id: ThreadId,
) {
app.handle_app_server_event(
app_server,
codex_app_server_client::AppServerEvent::ServerNotification(Box::new(
ServerNotification::McpServerStatusUpdated(McpServerStatusUpdatedNotification {
thread_id: Some(thread_id.to_string()),
name: "fixture".to_string(),
status: McpServerStartupState::Failed,
error: Some("fixture failed".to_string()),
failure_reason: None,
}),
)),
)
.await;
}

#[tokio::test]
async fn startup_thread_start_failure_returns_error() {
let (mut app, _app_event_rx, _op_rx) = make_test_app_with_channels().await;
Expand Down
Loading