github/codex
4
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-10-19 16:17:49 +00:00
Code Issues Packages Projects Releases Wiki Activity

Restrict TUI MCP startup notifications to owned threads (#50781)

Browse Source 
## Why

MCP startup notifications from unrelated threads could create TUI event channels, allowing subsequent approval requests from those threads to appear in the current session.

## What changed

Check the parent of an untracked subagent before accepting its MCP startup notifications. Read thread metadata with bounded retries for threads that are not yet found or loaded, and ignore notifications unless the parent belongs to the TUI.

Share thread ownership checks across notification and approval routing, including side threads.

## Testing

Extend approval-routing coverage to verify that owned subagent approvals are preserved both before thread-start notifications and after MCP startup notifications, while unrelated threads do not create event channels or display approval prompts.

GitOrigin-RevId: 6f78f237f7441aadf10b491ed191357df75c3a6a
This commit is contained in:
Eric Traut
2026-10-04 03:51:52 +00:00
committed by copyberry
parent dde5f8c5ae
commit f365d5754b
5 changed files with 159 additions and 30 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/tui/src/app.rs
+1
View File
@@ -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;
codex-rs/tui/src/app/app_server_events.rs
+7 -23
View File
@@ -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;
}
@@ -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
codex-rs/tui/src/app/app_server_thread_ownership.rs
+83
View File
@@ -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))
}
}
codex-rs/tui/src/app/tests.rs
+6
View File
@@ -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,
codex-rs/tui/src/app/tests/startup.rs
+62 -7
View File
@@ -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(
@@ -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;