From d8b4d7451c3399727c8051128dcebf8435e04a50 Mon Sep 17 00:00:00 2001 From: Andrei Hasna Date: Sun, 9 Aug 2026 21:05:27 +0300 Subject: [PATCH 01/13] test(tui): cover notification input starvation Agent: cossus --- codex-rs/tui/src/app/tests.rs | 56 +++++++++++++++++++++++++++++++++++ 1 file changed, 56 insertions(+) diff --git a/codex-rs/tui/src/app/tests.rs b/codex-rs/tui/src/app/tests.rs index 1bcc87d29..3ce8c924e 100644 --- a/codex-rs/tui/src/app/tests.rs +++ b/codex-rs/tui/src/app/tests.rs @@ -2869,6 +2869,62 @@ async fn refresh_pending_thread_approvals_only_lists_inactive_threads() { assert!(app.chat_widget.pending_thread_approvals().is_empty()); } +#[tokio::test] +async fn routine_notification_does_not_scan_every_inactive_thread_store() -> Result<()> { + const INACTIVE_THREAD_COUNT: usize = 64; + + let mut app = make_test_app().await; + let active_thread_id = ThreadId::new(); + app.primary_thread_id = Some(active_thread_id); + app.active_thread_id = Some(active_thread_id); + app.thread_event_channels + .insert(active_thread_id, ThreadEventChannel::new(/*capacity*/ 1)); + app.set_thread_active(active_thread_id, /*active*/ true) + .await; + + let mut blocked_store = None; + for index in 0..INACTIVE_THREAD_COUNT { + let thread_id = ThreadId::new(); + let channel = ThreadEventChannel::new(/*capacity*/ 1); + if index == 0 { + blocked_store = Some(Arc::clone(&channel.store)); + } + app.thread_event_channels.insert(thread_id, channel); + } + let blocked_store = blocked_store.expect("at least one inactive thread store"); + let blocked_guard = blocked_store.lock().await; + + let control = time::timeout( + std::time::Duration::from_millis(/*millis*/ 250), + app.refresh_pending_thread_approvals(), + ) + .await; + assert!( + control.is_err(), + "CONTROL: the held inactive store lock must block a full approvals scan" + ); + + let routed = time::timeout( + std::time::Duration::from_millis(/*millis*/ 250), + app.enqueue_thread_notification( + active_thread_id, + agent_message_delta_notification( + active_thread_id, + "turn-1", + "item-1", + "responsive typing", + ), + ), + ) + .await; + drop(blocked_guard); + + routed.expect( + "routine notification routing must not wait for unrelated inactive thread stores", + )?; + Ok(()) +} + #[tokio::test] async fn inactive_thread_approval_bubbles_into_active_view() -> Result<()> { let mut app = make_test_app().await; From 5679cc94bdf109681d0f2d260318df5c9ce611a3 Mon Sep 17 00:00:00 2001 From: Andrei Hasna Date: Sun, 9 Aug 2026 21:15:03 +0300 Subject: [PATCH 02/13] test(tui): cover closed-thread buffer retention Agent: cossus --- codex-rs/tui/src/app/tests.rs | 54 +++++++++++++++++++++++++++++++++++ 1 file changed, 54 insertions(+) diff --git a/codex-rs/tui/src/app/tests.rs b/codex-rs/tui/src/app/tests.rs index 3ce8c924e..dfc550c30 100644 --- a/codex-rs/tui/src/app/tests.rs +++ b/codex-rs/tui/src/app/tests.rs @@ -2925,6 +2925,60 @@ async fn routine_notification_does_not_scan_every_inactive_thread_store() -> Res Ok(()) } +#[tokio::test] +async fn closed_persisted_thread_releases_buffered_transcript_events() -> Result<()> { + const TRANSCRIPT_EVENT_COUNT: usize = 256; + + let mut app = make_test_app().await; + let thread_id = ThreadId::new(); + let channel = ThreadEventChannel::new_with_session( + THREAD_EVENT_CHANNEL_CAPACITY, + test_thread_session(thread_id, test_path_buf("/tmp/closed-agent")), + Vec::new(), + ); + let store = Arc::clone(&channel.store); + app.thread_event_channels.insert(thread_id, channel); + + { + let mut guard = store.lock().await; + for index in 0..TRANSCRIPT_EVENT_COUNT { + guard.push_notification(agent_message_delta_notification( + thread_id, + "turn-1", + "item-1", + &format!("delta-{index}"), + )); + } + guard.push_notification(schedule_updated_notification(thread_id)); + guard.push_notification(schedule_run_updated_notification( + thread_id, + "schedule-1", + "run-1", + )); + assert_eq!( + guard.buffer.len(), + TRANSCRIPT_EVENT_COUNT + 2, + "CONTROL: closed-thread compaction must start with retained transcript pressure" + ); + } + + app.enqueue_thread_notification(thread_id, thread_closed_notification(thread_id)) + .await?; + + let guard = store.lock().await; + assert_buffered_schedule_notifications(&guard.snapshot().events, thread_id); + assert_eq!( + guard.buffer.len(), + 2, + "persisted closed threads should keep replay-only state, not cloned transcript events" + ); + assert!( + guard.buffer.capacity() <= 4, + "closed-thread compaction must release the oversized transcript allocation" + ); + Ok(()) +} + #[tokio::test] async fn inactive_thread_approval_bubbles_into_active_view() -> Result<()> { let mut app = make_test_app().await; From 581766560e4b1d71c194bd87ffa9a2a5e58b5090 Mon Sep 17 00:00:00 2001 From: Andrei Hasna Date: Sun, 9 Aug 2026 21:25:30 +0300 Subject: [PATCH 03/13] test(tui): fix closed-buffer regression setup Agent: cossus --- codex-rs/tui/src/app/tests.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/codex-rs/tui/src/app/tests.rs b/codex-rs/tui/src/app/tests.rs index dfc550c30..4d43d6618 100644 --- a/codex-rs/tui/src/app/tests.rs +++ b/codex-rs/tui/src/app/tests.rs @@ -2949,7 +2949,7 @@ async fn closed_persisted_thread_releases_buffered_transcript_events() -> Result &format!("delta-{index}"), )); } - guard.push_notification(schedule_updated_notification(thread_id)); + guard.push_notification(schedule_updated_notification(thread_id, "schedule-1")); guard.push_notification(schedule_run_updated_notification( thread_id, "schedule-1", From cdd0d6dd0461c44452741c04de0bad619dbf59bb Mon Sep 17 00:00:00 2001 From: Andrei Hasna Date: Sun, 9 Aug 2026 21:32:16 +0300 Subject: [PATCH 04/13] test(app-server): cover thread subscription lineage Agent: cossus --- .../suite/v2/connection_handling_websocket.rs | 201 ++++++++++++++++++ 1 file changed, 201 insertions(+) diff --git a/codex-rs/app-server/tests/suite/v2/connection_handling_websocket.rs b/codex-rs/app-server/tests/suite/v2/connection_handling_websocket.rs index 27d3bdd08..8149b1427 100644 --- a/codex-rs/app-server/tests/suite/v2/connection_handling_websocket.rs +++ b/codex-rs/app-server/tests/suite/v2/connection_handling_websocket.rs @@ -2,6 +2,7 @@ use anyhow::Context; use anyhow::Result; use anyhow::bail; use app_test_support::DISABLE_PLUGIN_STARTUP_TASKS_ARG; +use app_test_support::create_mock_responses_server_repeating_assistant; use app_test_support::create_mock_responses_server_sequence_unchecked; use app_test_support::to_response; use base64::Engine; @@ -16,8 +17,15 @@ use codex_app_server_protocol::JSONRPCResponse; use codex_app_server_protocol::RequestId; use codex_app_server_protocol::ThreadLoadedListParams; use codex_app_server_protocol::ThreadLoadedListResponse; +use codex_app_server_protocol::ThreadResumeParams; +use codex_app_server_protocol::ThreadResumeResponse; +use codex_app_server_protocol::ThreadSource; use codex_app_server_protocol::ThreadStartParams; use codex_app_server_protocol::ThreadStartResponse; +use codex_app_server_protocol::ThreadStartedNotification; +use codex_app_server_protocol::TurnStartParams; +use codex_app_server_protocol::TurnStartedNotification; +use codex_app_server_protocol::UserInput as V2UserInput; use futures::SinkExt; use futures::StreamExt; use hmac::Hmac; @@ -104,6 +112,102 @@ async fn websocket_transport_routes_per_connection_handshake_and_responses() -> Ok(()) } +#[tokio::test] +async fn websocket_unsubscribed_connection_does_not_receive_top_level_thread_turn_events() +-> Result<()> { + let server = create_mock_responses_server_repeating_assistant("Done").await; + let codex_home = TempDir::new()?; + create_config_toml(codex_home.path(), &server.uri(), "never")?; + + let (mut process, bind_addr) = spawn_websocket_server(codex_home.path()).await?; + + let mut ws1 = connect_websocket(bind_addr).await?; + let mut ws2 = connect_websocket(bind_addr).await?; + initialize_websocket_client(&mut ws1, /*id*/ 1, "ws_thread_owner").await?; + initialize_websocket_client(&mut ws2, /*id*/ 2, "ws_unsubscribed_observer").await?; + + let thread = start_thread_with_notification(&mut ws1, /*id*/ 10, None).await?; + let discovered = read_thread_started_for_thread(&mut ws2, &thread.id).await?; + assert_eq!( + discovered.thread.id, thread.id, + "CONTROL: global thread discovery must remain visible to initialized clients" + ); + + // The thread-created receiver runs independently of the request processor. Give the current + // implementation a deterministic opportunity to attach any unintended listeners before + // producing listener-scoped traffic. + sleep(Duration::from_millis(100)).await; + + send_turn_start_request(&mut ws1, /*id*/ 11, &thread.id).await?; + let (turn_response, owner_turn_started) = + read_response_and_notification_for_method(&mut ws1, /*id*/ 11, "turn/started").await?; + assert_eq!(turn_response.id, RequestId::Integer(11)); + assert_turn_started_for_thread(owner_turn_started, &thread.id)?; + + assert_no_notification_for_method(&mut ws2, "turn/started", Duration::from_millis(500)).await?; + + process + .kill() + .await + .context("failed to stop websocket app-server process")?; + Ok(()) +} + +#[tokio::test] +async fn websocket_parent_subscribers_receive_child_thread_turn_events() -> Result<()> { + let server = create_mock_responses_server_repeating_assistant("Done").await; + let codex_home = TempDir::new()?; + create_config_toml(codex_home.path(), &server.uri(), "never")?; + + let (mut process, bind_addr) = spawn_websocket_server(codex_home.path()).await?; + + let mut ws1 = connect_websocket(bind_addr).await?; + let mut ws2 = connect_websocket(bind_addr).await?; + initialize_websocket_client(&mut ws1, /*id*/ 1, "ws_parent_owner").await?; + initialize_websocket_client(&mut ws2, /*id*/ 2, "ws_parent_subscriber").await?; + + let parent = start_thread_with_notification(&mut ws1, /*id*/ 20, None).await?; + read_thread_started_for_thread(&mut ws2, &parent.id).await?; + + send_request( + &mut ws2, + "thread/resume", + /*id*/ 21, + Some(serde_json::to_value(ThreadResumeParams { + thread_id: parent.id.clone(), + ..Default::default() + })?), + ) + .await?; + let resume_response = read_response_for_id(&mut ws2, /*id*/ 21).await?; + let resume: ThreadResumeResponse = to_response(resume_response)?; + assert_eq!(resume.thread.id, parent.id); + + let child = start_thread_with_notification( + &mut ws1, + /*id*/ 22, + Some((parent.id.clone(), ThreadSource::Subagent)), + ) + .await?; + read_thread_started_for_thread(&mut ws2, &child.id).await?; + sleep(Duration::from_millis(100)).await; + + send_turn_start_request(&mut ws1, /*id*/ 23, &child.id).await?; + let (turn_response, owner_turn_started) = + read_response_and_notification_for_method(&mut ws1, /*id*/ 23, "turn/started").await?; + assert_eq!(turn_response.id, RequestId::Integer(23)); + assert_turn_started_for_thread(owner_turn_started, &child.id)?; + + let subscriber_turn_started = read_notification_for_method(&mut ws2, "turn/started").await?; + assert_turn_started_for_thread(subscriber_turn_started, &child.id)?; + + process + .kill() + .await + .context("failed to stop websocket app-server process")?; + Ok(()) +} + #[tokio::test] async fn websocket_transport_serves_health_endpoints_on_same_listener() -> Result<()> { let server = create_mock_responses_server_sequence_unchecked(Vec::new()).await; @@ -627,6 +731,103 @@ async fn start_thread(stream: &mut WsClient, id: i64) -> Result { Ok(thread.id) } +async fn initialize_websocket_client( + stream: &mut WsClient, + id: i64, + client_name: &str, +) -> Result<()> { + send_initialize_request(stream, id, client_name).await?; + let response = read_response_for_id(stream, id).await?; + assert_eq!(response.id, RequestId::Integer(id)); + Ok(()) +} + +async fn start_thread_with_notification( + stream: &mut WsClient, + id: i64, + parent: Option<(String, ThreadSource)>, +) -> Result { + let (parent_thread_id, thread_source) = match parent { + Some((parent_thread_id, thread_source)) => (Some(parent_thread_id), Some(thread_source)), + None => (None, None), + }; + send_request( + stream, + "thread/start", + id, + Some(serde_json::to_value(ThreadStartParams { + model: Some("mock-model".to_string()), + parent_thread_id, + thread_source, + ..Default::default() + })?), + ) + .await?; + let (response, notification) = + read_response_and_notification_for_method(stream, id, "thread/started").await?; + let ThreadStartResponse { thread, .. } = to_response::(response)?; + let started: ThreadStartedNotification = + serde_json::from_value(notification.params.context("thread/started params")?)?; + assert_eq!(started.thread, thread); + Ok(thread) +} + +async fn send_turn_start_request(stream: &mut WsClient, id: i64, thread_id: &str) -> Result<()> { + send_request( + stream, + "turn/start", + id, + Some(serde_json::to_value(TurnStartParams { + thread_id: thread_id.to_string(), + client_user_message_id: None, + input: vec![V2UserInput::Text { + text: "Hello".to_string(), + text_elements: Vec::new(), + }], + ..Default::default() + })?), + ) + .await +} + +async fn read_thread_started_for_thread( + stream: &mut WsClient, + thread_id: &str, +) -> Result { + loop { + let notification = read_notification_for_method(stream, "thread/started").await?; + let started: ThreadStartedNotification = + serde_json::from_value(notification.params.context("thread/started params")?)?; + if started.thread.id == thread_id { + return Ok(started); + } + } +} + +fn assert_turn_started_for_thread( + notification: JSONRPCNotification, + thread_id: &str, +) -> Result<()> { + let started: TurnStartedNotification = + serde_json::from_value(notification.params.context("turn/started params")?)?; + assert_eq!(started.thread_id, thread_id); + Ok(()) +} + +async fn assert_no_notification_for_method( + stream: &mut WsClient, + method: &str, + wait_for: Duration, +) -> Result<()> { + match timeout(wait_for, read_notification_for_method(stream, method)).await { + Ok(Ok(notification)) => { + bail!("unexpected `{method}` notification: {notification:?}") + } + Ok(Err(err)) => Err(err), + Err(_) => Ok(()), + } +} + async fn assert_loaded_threads(stream: &mut WsClient, id: i64, expected: &[&str]) -> Result<()> { let response = request_loaded_threads(stream, id).await?; let mut actual = response.data; From 098ea8c91f4f0ee3040f268cd46ecc94a994ac8a Mon Sep 17 00:00:00 2001 From: Andrei Hasna Date: Sun, 9 Aug 2026 21:39:20 +0300 Subject: [PATCH 05/13] fix(tui): keep long-lived prompt input responsive Agent: cossus --- codex-rs/tui/src/app/thread_events.rs | 1 + codex-rs/tui/src/app/thread_routing.rs | 35 +++++++++++++++++++++----- 2 files changed, 30 insertions(+), 6 deletions(-) diff --git a/codex-rs/tui/src/app/thread_events.rs b/codex-rs/tui/src/app/thread_events.rs index 968bc15da..bc5509227 100644 --- a/codex-rs/tui/src/app/thread_events.rs +++ b/codex-rs/tui/src/app/thread_events.rs @@ -93,6 +93,7 @@ impl ThreadEventStore { pub(super) fn rebase_buffer_after_session_refresh(&mut self) { self.buffer.retain(Self::event_survives_session_refresh); + self.buffer.shrink_to_fit(); } pub(super) fn set_turns(&mut self, turns: Vec) { diff --git a/codex-rs/tui/src/app/thread_routing.rs b/codex-rs/tui/src/app/thread_routing.rs index 6d6316153..f6a7ca53a 100644 --- a/codex-rs/tui/src/app/thread_routing.rs +++ b/codex-rs/tui/src/app/thread_routing.rs @@ -1111,15 +1111,29 @@ impl App { (channel.sender.clone(), Arc::clone(&channel.store)) }; - let (should_send, pending_status) = { + let (should_send, pending_status, pending_approvals_changed) = { let mut guard = store.lock().await; if guard.session.is_none() && let Some(session) = inferred_session { guard.session = Some(session); } + let had_pending_approvals = guard.has_pending_thread_approvals(); + let should_rebase_closed_thread = + matches!(¬ification, ServerNotification::ThreadClosed(_)) + && guard + .session + .as_ref() + .is_some_and(|session| session.rollout_path.is_some()); guard.push_notification(notification.clone()); - (guard.active, guard.side_parent_pending_status()) + if should_rebase_closed_thread { + guard.rebase_buffer_after_session_refresh(); + } + ( + guard.active, + guard.side_parent_pending_status(), + had_pending_approvals != guard.has_pending_thread_approvals(), + ) }; let notification_status_change = SideParentStatusChange::for_notification(¬ification); @@ -1143,7 +1157,9 @@ impl App { } else if let Some(change) = notification_status_change { self.apply_side_parent_status_change(thread_id, change); } - self.refresh_pending_thread_approvals().await; + if pending_approvals_changed { + self.refresh_pending_thread_approvals().await; + } Ok(()) } @@ -1319,10 +1335,15 @@ impl App { (channel.sender.clone(), Arc::clone(&channel.store)) }; - let (should_send, pending_status) = { + let (should_send, pending_status, pending_approvals_changed) = { let mut guard = store.lock().await; + let had_pending_approvals = guard.has_pending_thread_approvals(); guard.push_request(request.clone()); - (guard.active, guard.side_parent_pending_status()) + ( + guard.active, + guard.side_parent_pending_status(), + had_pending_approvals != guard.has_pending_thread_approvals(), + ) }; let request_status = SideParentStatus::for_request(&request); @@ -1348,7 +1369,9 @@ impl App { if let Some(status) = pending_status.or(request_status) { self.set_side_parent_status(thread_id, Some(status)); } - self.refresh_pending_thread_approvals().await; + if pending_approvals_changed { + self.refresh_pending_thread_approvals().await; + } Ok(()) } From 24233e440857d141418f4838516fc341ca427c20 Mon Sep 17 00:00:00 2001 From: Andrei Hasna Date: Sun, 9 Aug 2026 21:51:28 +0300 Subject: [PATCH 06/13] test(runtime): cover lineage and daemon version isolation Agent: cossus --- .../suite/v2/connection_handling_websocket.rs | 192 +++++++++++------- codex-rs/tui/src/lib.rs | 117 ++++++++++- 2 files changed, 229 insertions(+), 80 deletions(-) diff --git a/codex-rs/app-server/tests/suite/v2/connection_handling_websocket.rs b/codex-rs/app-server/tests/suite/v2/connection_handling_websocket.rs index 8149b1427..526c360d3 100644 --- a/codex-rs/app-server/tests/suite/v2/connection_handling_websocket.rs +++ b/codex-rs/app-server/tests/suite/v2/connection_handling_websocket.rs @@ -19,13 +19,15 @@ use codex_app_server_protocol::ThreadLoadedListParams; use codex_app_server_protocol::ThreadLoadedListResponse; use codex_app_server_protocol::ThreadResumeParams; use codex_app_server_protocol::ThreadResumeResponse; -use codex_app_server_protocol::ThreadSource; use codex_app_server_protocol::ThreadStartParams; use codex_app_server_protocol::ThreadStartResponse; use codex_app_server_protocol::ThreadStartedNotification; +use codex_app_server_protocol::ThreadUnsubscribeParams; +use codex_app_server_protocol::ThreadUnsubscribeResponse; +use codex_app_server_protocol::ThreadUnsubscribeStatus; use codex_app_server_protocol::TurnStartParams; -use codex_app_server_protocol::TurnStartedNotification; use codex_app_server_protocol::UserInput as V2UserInput; +use core_test_support::responses; use futures::SinkExt; use futures::StreamExt; use hmac::Hmac; @@ -113,11 +115,11 @@ async fn websocket_transport_routes_per_connection_handshake_and_responses() -> } #[tokio::test] -async fn websocket_unsubscribed_connection_does_not_receive_top_level_thread_turn_events() --> Result<()> { - let server = create_mock_responses_server_repeating_assistant("Done").await; +async fn websocket_unrelated_connection_is_not_subscribed_to_spawned_child() -> Result<()> { + let server = + create_mock_responses_server_sequence_unchecked(spawn_agent_response_sequence()?).await; let codex_home = TempDir::new()?; - create_config_toml(codex_home.path(), &server.uri(), "never")?; + create_multi_agent_config_toml(codex_home.path(), &server.uri())?; let (mut process, bind_addr) = spawn_websocket_server(codex_home.path()).await?; @@ -126,25 +128,24 @@ async fn websocket_unsubscribed_connection_does_not_receive_top_level_thread_tur initialize_websocket_client(&mut ws1, /*id*/ 1, "ws_thread_owner").await?; initialize_websocket_client(&mut ws2, /*id*/ 2, "ws_unsubscribed_observer").await?; - let thread = start_thread_with_notification(&mut ws1, /*id*/ 10, None).await?; - let discovered = read_thread_started_for_thread(&mut ws2, &thread.id).await?; + let parent = start_thread_with_notification(&mut ws1, /*id*/ 10).await?; + let discovered = read_thread_started_for_thread(&mut ws2, &parent.id).await?; assert_eq!( - discovered.thread.id, thread.id, + discovered.thread.id, parent.id, "CONTROL: global thread discovery must remain visible to initialized clients" ); + assert_eq!( + unsubscribe_thread(&mut ws2, /*id*/ 11, &parent.id).await?, + ThreadUnsubscribeStatus::NotSubscribed, + "CONTROL: the observer must start unrelated to the parent" + ); - // The thread-created receiver runs independently of the request processor. Give the current - // implementation a deterministic opportunity to attach any unintended listeners before - // producing listener-scoped traffic. - sleep(Duration::from_millis(100)).await; - - send_turn_start_request(&mut ws1, /*id*/ 11, &thread.id).await?; - let (turn_response, owner_turn_started) = - read_response_and_notification_for_method(&mut ws1, /*id*/ 11, "turn/started").await?; - assert_eq!(turn_response.id, RequestId::Integer(11)); - assert_turn_started_for_thread(owner_turn_started, &thread.id)?; - - assert_no_notification_for_method(&mut ws2, "turn/started", Duration::from_millis(500)).await?; + let child = spawn_child_from_parent(&mut ws1, &mut ws2, /*id*/ 12, &parent.id).await?; + assert_eq!( + unsubscribe_thread(&mut ws2, /*id*/ 13, &child.id).await?, + ThreadUnsubscribeStatus::NotSubscribed, + "unrelated initialized connections must not inherit spawned-child subscriptions" + ); process .kill() @@ -154,10 +155,11 @@ async fn websocket_unsubscribed_connection_does_not_receive_top_level_thread_tur } #[tokio::test] -async fn websocket_parent_subscribers_receive_child_thread_turn_events() -> Result<()> { - let server = create_mock_responses_server_repeating_assistant("Done").await; +async fn websocket_parent_subscriber_inherits_spawned_child_subscription() -> Result<()> { + let server = + create_mock_responses_server_sequence_unchecked(spawn_agent_response_sequence()?).await; let codex_home = TempDir::new()?; - create_config_toml(codex_home.path(), &server.uri(), "never")?; + create_multi_agent_config_toml(codex_home.path(), &server.uri())?; let (mut process, bind_addr) = spawn_websocket_server(codex_home.path()).await?; @@ -166,7 +168,7 @@ async fn websocket_parent_subscribers_receive_child_thread_turn_events() -> Resu initialize_websocket_client(&mut ws1, /*id*/ 1, "ws_parent_owner").await?; initialize_websocket_client(&mut ws2, /*id*/ 2, "ws_parent_subscriber").await?; - let parent = start_thread_with_notification(&mut ws1, /*id*/ 20, None).await?; + let parent = start_thread_with_notification(&mut ws1, /*id*/ 20).await?; read_thread_started_for_thread(&mut ws2, &parent.id).await?; send_request( @@ -183,23 +185,12 @@ async fn websocket_parent_subscribers_receive_child_thread_turn_events() -> Resu let resume: ThreadResumeResponse = to_response(resume_response)?; assert_eq!(resume.thread.id, parent.id); - let child = start_thread_with_notification( - &mut ws1, - /*id*/ 22, - Some((parent.id.clone(), ThreadSource::Subagent)), - ) - .await?; - read_thread_started_for_thread(&mut ws2, &child.id).await?; - sleep(Duration::from_millis(100)).await; - - send_turn_start_request(&mut ws1, /*id*/ 23, &child.id).await?; - let (turn_response, owner_turn_started) = - read_response_and_notification_for_method(&mut ws1, /*id*/ 23, "turn/started").await?; - assert_eq!(turn_response.id, RequestId::Integer(23)); - assert_turn_started_for_thread(owner_turn_started, &child.id)?; - - let subscriber_turn_started = read_notification_for_method(&mut ws2, "turn/started").await?; - assert_turn_started_for_thread(subscriber_turn_started, &child.id)?; + let child = spawn_child_from_parent(&mut ws1, &mut ws2, /*id*/ 22, &parent.id).await?; + assert_eq!( + unsubscribe_thread(&mut ws2, /*id*/ 23, &child.id).await?, + ThreadUnsubscribeStatus::Unsubscribed, + "connections subscribed to the parent must inherit the spawned child" + ); process .kill() @@ -745,20 +736,13 @@ async fn initialize_websocket_client( async fn start_thread_with_notification( stream: &mut WsClient, id: i64, - parent: Option<(String, ThreadSource)>, ) -> Result { - let (parent_thread_id, thread_source) = match parent { - Some((parent_thread_id, thread_source)) => (Some(parent_thread_id), Some(thread_source)), - None => (None, None), - }; send_request( stream, "thread/start", id, Some(serde_json::to_value(ThreadStartParams { model: Some("mock-model".to_string()), - parent_thread_id, - thread_source, ..Default::default() })?), ) @@ -772,6 +756,54 @@ async fn start_thread_with_notification( Ok(thread) } +async fn spawn_child_from_parent( + owner: &mut WsClient, + observer: &mut WsClient, + id: i64, + parent_thread_id: &str, +) -> Result { + send_turn_start_request(owner, id, parent_thread_id).await?; + let response = read_response_for_id(owner, id).await?; + assert_eq!(response.id, RequestId::Integer(id)); + + let child = read_spawned_child_for_parent(observer, parent_thread_id).await?; + sleep(Duration::from_millis(100)).await; + Ok(child) +} + +async fn read_spawned_child_for_parent( + stream: &mut WsClient, + parent_thread_id: &str, +) -> Result { + loop { + let notification = read_notification_for_method(stream, "thread/started").await?; + let started: ThreadStartedNotification = + serde_json::from_value(notification.params.context("thread/started params")?)?; + if started.thread.parent_thread_id.as_deref() == Some(parent_thread_id) { + return Ok(started.thread); + } + } +} + +async fn unsubscribe_thread( + stream: &mut WsClient, + id: i64, + thread_id: &str, +) -> Result { + send_request( + stream, + "thread/unsubscribe", + id, + Some(serde_json::to_value(ThreadUnsubscribeParams { + thread_id: thread_id.to_string(), + })?), + ) + .await?; + let response = read_response_for_id(stream, id).await?; + let response: ThreadUnsubscribeResponse = to_response(response)?; + Ok(response.status) +} + async fn send_turn_start_request(stream: &mut WsClient, id: i64, thread_id: &str) -> Result<()> { send_request( stream, @@ -804,30 +836,6 @@ async fn read_thread_started_for_thread( } } -fn assert_turn_started_for_thread( - notification: JSONRPCNotification, - thread_id: &str, -) -> Result<()> { - let started: TurnStartedNotification = - serde_json::from_value(notification.params.context("turn/started params")?)?; - assert_eq!(started.thread_id, thread_id); - Ok(()) -} - -async fn assert_no_notification_for_method( - stream: &mut WsClient, - method: &str, - wait_for: Duration, -) -> Result<()> { - match timeout(wait_for, read_notification_for_method(stream, method)).await { - Ok(Ok(notification)) => { - bail!("unexpected `{method}` notification: {notification:?}") - } - Ok(Err(err)) => Err(err), - Err(_) => Ok(()), - } -} - async fn assert_loaded_threads(stream: &mut WsClient, id: i64, expected: &[&str]) -> Result<()> { let response = request_loaded_threads(stream, id).await?; let mut actual = response.data; @@ -1056,6 +1064,44 @@ stream_max_retries = 0 ) } +fn create_multi_agent_config_toml(codex_home: &Path, server_uri: &str) -> std::io::Result<()> { + create_config_toml(codex_home, server_uri, "never")?; + let config_toml = codex_home.join("config.toml"); + let mut config = std::fs::read_to_string(&config_toml)?; + config.push_str( + r#" + +[features.multi_agent_v2] +enabled = true +non_code_mode_only = false +"#, + ); + std::fs::write(config_toml, config) +} + +fn spawn_agent_response_sequence() -> Result> { + let spawn_arguments = serde_json::to_string(&json!({ + "fork_turns": "none", + "message": "Wait for follow-up.", + "task_name": "listener_child", + }))?; + let mut response_sequence = vec![responses::sse(vec![ + responses::ev_response_created("resp-spawn"), + responses::ev_function_call("call-spawn", "spawn_agent", &spawn_arguments), + responses::ev_completed("resp-spawn"), + ])]; + for index in 0..8 { + let response_id = format!("resp-done-{index}"); + let message_id = format!("msg-done-{index}"); + response_sequence.push(responses::sse(vec![ + responses::ev_response_created(&response_id), + responses::ev_assistant_message(&message_id, "Done"), + responses::ev_completed(&response_id), + ])); + } + Ok(response_sequence) +} + fn connectable_bind_addr(bind_addr: SocketAddr) -> SocketAddr { match bind_addr { SocketAddr::V4(addr) if addr.ip().is_unspecified() => { diff --git a/codex-rs/tui/src/lib.rs b/codex-rs/tui/src/lib.rs index 8c80f7ea3..0302d5503 100644 --- a/codex-rs/tui/src/lib.rs +++ b/codex-rs/tui/src/lib.rs @@ -302,6 +302,14 @@ const TUI_LOG_FILE_NAME: &str = "codex-tui.log"; const AUTO_CONNECT_DAEMON_CONNECT_TIMEOUT: std::time::Duration = std::time::Duration::from_millis(50); +#[cfg(unix)] +#[derive(Debug, PartialEq, Eq)] +enum DefaultDaemonSocketProbe { + Reuse(AbsolutePathBuf), + StartAllowed, + UseEmbedded, +} + #[allow(clippy::too_many_arguments)] async fn start_embedded_app_server( arg0_paths: Arg0DispatchPaths, @@ -479,9 +487,31 @@ async fn connect_remote_app_server( #[cfg(unix)] async fn maybe_probe_default_daemon_socket(codex_home: &Path) -> Option { - let socket_path = codex_app_server_client::app_server_control_socket_path(codex_home).ok()?; + match probe_default_daemon_socket_with(codex_home, |_| async { + Ok(env!("CARGO_PKG_VERSION").to_string()) + }) + .await + { + DefaultDaemonSocketProbe::Reuse(socket_path) => Some(socket_path), + DefaultDaemonSocketProbe::StartAllowed | DefaultDaemonSocketProbe::UseEmbedded => None, + } +} + +#[cfg(unix)] +async fn probe_default_daemon_socket_with( + codex_home: &Path, + probe_version: F, +) -> DefaultDaemonSocketProbe +where + F: FnOnce(PathBuf) -> Fut, + Fut: std::future::Future>, +{ + let Ok(socket_path) = codex_app_server_client::app_server_control_socket_path(codex_home) + else { + return DefaultDaemonSocketProbe::StartAllowed; + }; if !socket_path.as_path().try_exists().unwrap_or(false) { - return None; + return DefaultDaemonSocketProbe::StartAllowed; } match tokio::time::timeout( @@ -490,10 +520,13 @@ async fn maybe_probe_default_daemon_socket(codex_home: &Path) -> Option Some(socket_path), + Ok(Ok(_stream)) => { + let _ = probe_version(socket_path.as_path().to_path_buf()).await; + DefaultDaemonSocketProbe::Reuse(socket_path) + } Ok(Err(err)) => { tracing::debug!(%err, socket_path = %socket_path.display(), "skipping default app-server daemon socket"); - None + DefaultDaemonSocketProbe::StartAllowed } Err(_) => { tracing::debug!( @@ -501,7 +534,7 @@ async fn maybe_probe_default_daemon_socket(codex_home: &Path) -> Option Fut, Fut: std::future::Future>, { - if let Some(socket_path) = maybe_probe_default_daemon_socket(codex_home).await { - return Some(socket_path); + maybe_start_default_daemon_socket_with_probes( + codex_home, + codex_bin, + |_| async { Ok(env!("CARGO_PKG_VERSION").to_string()) }, + start_daemon, + ) + .await +} + +#[cfg(unix)] +async fn maybe_start_default_daemon_socket_with_probes( + codex_home: &Path, + codex_bin: Option<&Path>, + probe_version: P, + start_daemon: F, +) -> Option +where + P: FnOnce(PathBuf) -> PFut, + PFut: std::future::Future>, + F: FnOnce(PathBuf) -> Fut, + Fut: std::future::Future>, +{ + match probe_default_daemon_socket_with(codex_home, probe_version).await { + DefaultDaemonSocketProbe::Reuse(socket_path) => return Some(socket_path), + DefaultDaemonSocketProbe::UseEmbedded => return None, + DefaultDaemonSocketProbe::StartAllowed => {} } let Some(codex_bin) = codex_bin else { @@ -2582,6 +2639,52 @@ mod tests { Ok(()) } + #[cfg(unix)] + #[tokio::test] + async fn default_daemon_auto_start_rejects_mismatched_version_without_starting() + -> color_eyre::Result<()> { + let codex_home = TempDir::new()?; + let socket_path = + codex_app_server_client::app_server_control_socket_path(codex_home.path())?; + std::fs::create_dir_all(socket_path.as_path().parent().expect("socket parent"))?; + let _listener = tokio::net::UnixListener::bind(socket_path.as_path())?; + let probe_count = Arc::new(AtomicUsize::new(0)); + let probe_count_for_closure = Arc::clone(&probe_count); + let start_count = Arc::new(AtomicUsize::new(0)); + let start_count_for_closure = Arc::clone(&start_count); + let unexpected_socket_path = codex_home.path().join("unexpected.sock"); + + let resolved_socket_path = maybe_start_default_daemon_socket_with_probes( + codex_home.path(), + Some(std::path::Path::new("/bin/codewith")), + move |_| async move { + probe_count_for_closure.fetch_add(1, Ordering::SeqCst); + Ok("0.1.82".to_string()) + }, + move |_| async move { + start_count_for_closure.fetch_add(1, Ordering::SeqCst); + Ok(unexpected_socket_path) + }, + ) + .await; + + assert_eq!( + probe_count.load(Ordering::SeqCst), + 1, + "CONTROL: a reachable socket must be version-probed" + ); + assert_eq!( + start_count.load(Ordering::SeqCst), + 0, + "CONTROL: an existing reachable daemon must never be started or replaced" + ); + assert_eq!( + resolved_socket_path, None, + "a reachable daemon with a mismatched version must fall back to embedded" + ); + Ok(()) + } + #[cfg(unix)] #[tokio::test] async fn default_daemon_auto_start_starts_when_socket_is_missing() -> color_eyre::Result<()> { From e289993fd068958236451f1b7b4b829e077b6f05 Mon Sep 17 00:00:00 2001 From: Andrei Hasna Date: Sun, 9 Aug 2026 21:58:35 +0300 Subject: [PATCH 07/13] test(app-server): discover spawned children via loaded list Agent: cossus --- .../suite/v2/connection_handling_websocket.rs | 43 ++++++++++++------- 1 file changed, 27 insertions(+), 16 deletions(-) diff --git a/codex-rs/app-server/tests/suite/v2/connection_handling_websocket.rs b/codex-rs/app-server/tests/suite/v2/connection_handling_websocket.rs index 526c360d3..e0924ceb9 100644 --- a/codex-rs/app-server/tests/suite/v2/connection_handling_websocket.rs +++ b/codex-rs/app-server/tests/suite/v2/connection_handling_websocket.rs @@ -140,9 +140,9 @@ async fn websocket_unrelated_connection_is_not_subscribed_to_spawned_child() -> "CONTROL: the observer must start unrelated to the parent" ); - let child = spawn_child_from_parent(&mut ws1, &mut ws2, /*id*/ 12, &parent.id).await?; + let child_id = spawn_child_from_parent(&mut ws1, &mut ws2, /*id*/ 12, &parent.id).await?; assert_eq!( - unsubscribe_thread(&mut ws2, /*id*/ 13, &child.id).await?, + unsubscribe_thread(&mut ws2, /*id*/ 13, &child_id).await?, ThreadUnsubscribeStatus::NotSubscribed, "unrelated initialized connections must not inherit spawned-child subscriptions" ); @@ -185,9 +185,9 @@ async fn websocket_parent_subscriber_inherits_spawned_child_subscription() -> Re let resume: ThreadResumeResponse = to_response(resume_response)?; assert_eq!(resume.thread.id, parent.id); - let child = spawn_child_from_parent(&mut ws1, &mut ws2, /*id*/ 22, &parent.id).await?; + let child_id = spawn_child_from_parent(&mut ws1, &mut ws2, /*id*/ 22, &parent.id).await?; assert_eq!( - unsubscribe_thread(&mut ws2, /*id*/ 23, &child.id).await?, + unsubscribe_thread(&mut ws2, /*id*/ 23, &child_id).await?, ThreadUnsubscribeStatus::Unsubscribed, "connections subscribed to the parent must inherit the spawned child" ); @@ -761,28 +761,39 @@ async fn spawn_child_from_parent( observer: &mut WsClient, id: i64, parent_thread_id: &str, -) -> Result { +) -> Result { send_turn_start_request(owner, id, parent_thread_id).await?; let response = read_response_for_id(owner, id).await?; assert_eq!(response.id, RequestId::Integer(id)); - let child = read_spawned_child_for_parent(observer, parent_thread_id).await?; + let child_id = + wait_for_new_loaded_child(observer, /*first_id*/ id + 100, parent_thread_id).await?; sleep(Duration::from_millis(100)).await; - Ok(child) + Ok(child_id) } -async fn read_spawned_child_for_parent( +async fn wait_for_new_loaded_child( stream: &mut WsClient, + first_id: i64, parent_thread_id: &str, -) -> Result { - loop { - let notification = read_notification_for_method(stream, "thread/started").await?; - let started: ThreadStartedNotification = - serde_json::from_value(notification.params.context("thread/started params")?)?; - if started.thread.parent_thread_id.as_deref() == Some(parent_thread_id) { - return Ok(started.thread); +) -> Result { + let mut next_id = first_id; + timeout(DEFAULT_READ_TIMEOUT, async { + loop { + let response = request_loaded_threads(stream, next_id).await?; + next_id += 1; + if let Some(child_id) = response + .data + .into_iter() + .find(|thread_id| thread_id != parent_thread_id) + { + return Ok::(child_id); + } + sleep(Duration::from_millis(50)).await; } - } + }) + .await + .context("timed out waiting for spawned child in thread/loaded/list")? } async fn unsubscribe_thread( From 2907ab47d366aca780d3cb48f92ed81a8841cffa Mon Sep 17 00:00:00 2001 From: Andrei Hasna Date: Sun, 9 Aug 2026 22:08:30 +0300 Subject: [PATCH 08/13] fix(runtime): isolate spawned lineage and daemon versions Agent: cossus --- CHANGELOG.md | 12 +- codex-rs/app-server/src/in_process.rs | 7 +- codex-rs/app-server/src/lib.rs | 11 +- codex-rs/app-server/src/message_processor.rs | 8 +- .../request_processors/thread_processor.rs | 72 +++++++-- codex-rs/tui/src/lib.rs | 142 ++++++++++++++++-- 6 files changed, 204 insertions(+), 48 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index ea822d1e0..adb2f5c5b 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -54,13 +54,23 @@ npm: Compare: This release prevents capability-shaped values stored in coordination metadata -from reaching the model through compact mission-control overview fields. +from reaching the model through compact mission-control overview fields. It +also keeps long-lived prompt input responsive, scopes spawned-child listeners +to their parent lineage, and avoids reusing incompatible app-server daemons. ### Fixed - Mission control: redact capability-shaped values from session titles and compact schedule or interaction previews before returning model-visible overview data. (#525) +- App server: inherit spawned-child listeners only from connections subscribed + to the parent thread, while preserving global manual-thread discovery. +- TUI: update pending-approval state incrementally and compact buffered + transcript events when a persisted closed thread can be reloaded. +- TUI: reuse the default app-server daemon only when its reported version + exactly matches the TUI; fall back to the embedded server after a mismatch, + failed handshake, or two-second version-probe timeout without restarting the + existing daemon. ## [0.1.86] - 2026-08-09 diff --git a/codex-rs/app-server/src/in_process.rs b/codex-rs/app-server/src/in_process.rs index f4e5475ac..604a108bf 100644 --- a/codex-rs/app-server/src/in_process.rs +++ b/codex-rs/app-server/src/in_process.rs @@ -492,13 +492,8 @@ async fn start_uninitialized(args: InProcessStartArgs) -> IoResult { match created { Ok(thread_id) => { - let connection_ids = if session.initialized() { - vec![IN_PROCESS_CONNECTION_ID] - } else { - Vec::::new() - }; processor - .try_attach_thread_listener(thread_id, connection_ids) + .try_attach_thread_lineage_listeners(thread_id) .await; } Err(tokio::sync::broadcast::error::RecvError::Lagged(_)) => { diff --git a/codex-rs/app-server/src/lib.rs b/codex-rs/app-server/src/lib.rs index 2a8858e57..fc3bb1a7e 100644 --- a/codex-rs/app-server/src/lib.rs +++ b/codex-rs/app-server/src/lib.rs @@ -1121,17 +1121,8 @@ pub async fn run_main_with_transport_options( created = thread_created_rx.recv(), if listen_for_threads => { match created { Ok(thread_id) => { - let mut initialized_connection_ids = Vec::new(); - for (connection_id, connection_state) in &connections { - if connection_state.session.initialized() { - initialized_connection_ids.push(*connection_id); - } - } processor - .try_attach_thread_listener( - thread_id, - initialized_connection_ids, - ) + .try_attach_thread_lineage_listeners(thread_id) .await; } Err(tokio::sync::broadcast::error::RecvError::Lagged(_)) => { diff --git a/codex-rs/app-server/src/message_processor.rs b/codex-rs/app-server/src/message_processor.rs index 9069686c6..ce61f3c5f 100644 --- a/codex-rs/app-server/src/message_processor.rs +++ b/codex-rs/app-server/src/message_processor.rs @@ -904,13 +904,9 @@ impl MessageProcessor { .await; } - pub(crate) async fn try_attach_thread_listener( - &self, - thread_id: ThreadId, - connection_ids: Vec, - ) { + pub(crate) async fn try_attach_thread_lineage_listeners(&self, thread_id: ThreadId) { self.thread_processor - .try_attach_thread_listener(thread_id, connection_ids) + .try_attach_thread_lineage_listeners(thread_id) .await; } diff --git a/codex-rs/app-server/src/request_processors/thread_processor.rs b/codex-rs/app-server/src/request_processors/thread_processor.rs index 0cd02bffe..8d16b60b6 100644 --- a/codex-rs/app-server/src/request_processors/thread_processor.rs +++ b/codex-rs/app-server/src/request_processors/thread_processor.rs @@ -2933,12 +2933,9 @@ impl ThreadRequestProcessor { self.thread_watch_manager.subscribe_running_turn_count() } - /// Best-effort: ensure initialized connections are subscribed to this thread. - pub(crate) async fn try_attach_thread_listener( - &self, - thread_id: ThreadId, - connection_ids: Vec, - ) { + /// Best-effort: inherit listeners from the newly created thread's parent. + pub(crate) async fn try_attach_thread_lineage_listeners(&self, thread_id: ThreadId) { + let mut connection_ids = Vec::new(); let mut raw_events_enabled = false; if let Ok(thread) = self.thread_manager.get_thread(thread_id).await { let config_snapshot = thread.config_snapshot().await; @@ -2949,7 +2946,14 @@ impl ThreadRequestProcessor { thread.rollout_path(), ); self.thread_watch_manager.upsert_thread(loaded_thread).await; - if let Some(parent_thread_id) = config_snapshot.parent_thread_id { + if let Some(parent_thread_id) = resolve_thread_parent_id( + config_snapshot.parent_thread_id, + &config_snapshot.session_source, + ) { + connection_ids = self + .thread_state_manager + .subscribed_connection_ids(parent_thread_id) + .await; raw_events_enabled = self .thread_state_manager .thread_state(parent_thread_id) @@ -4948,6 +4952,13 @@ fn permission_profile_trusts_project( } } +fn resolve_thread_parent_id( + explicit_parent_thread_id: Option, + session_source: &CoreSessionSource, +) -> Option { + explicit_parent_thread_id.or_else(|| session_source.parent_thread_id()) +} + fn build_thread_from_snapshot( thread_id: ThreadId, session_id: String, @@ -4955,9 +4966,10 @@ fn build_thread_from_snapshot( path: Option, ) -> Thread { let now = time::OffsetDateTime::now_utc().unix_timestamp(); - let parent_thread_id = config_snapshot - .parent_thread_id - .or_else(|| config_snapshot.session_source.parent_thread_id()); + let parent_thread_id = resolve_thread_parent_id( + config_snapshot.parent_thread_id, + &config_snapshot.session_source, + ); Thread { id: thread_id.to_string(), session_id, @@ -5001,6 +5013,46 @@ fn build_thread_from_loaded_snapshot( ) } +#[cfg(test)] +mod lineage_parent_tests { + use super::*; + + #[test] + fn explicit_parent_takes_precedence_over_session_source() { + let explicit_parent_thread_id = ThreadId::new(); + let source_parent_thread_id = ThreadId::new(); + let session_source = CoreSessionSource::SubAgent(CoreSubAgentSource::ThreadSpawn { + parent_thread_id: source_parent_thread_id, + depth: 1, + agent_path: None, + agent_nickname: None, + agent_role: None, + }); + + assert_eq!( + resolve_thread_parent_id(Some(explicit_parent_thread_id), &session_source), + Some(explicit_parent_thread_id) + ); + } + + #[test] + fn spawned_thread_uses_session_source_parent() { + let parent_thread_id = ThreadId::new(); + let session_source = CoreSessionSource::SubAgent(CoreSubAgentSource::ThreadSpawn { + parent_thread_id, + depth: 1, + agent_path: None, + agent_nickname: None, + agent_role: None, + }); + + assert_eq!( + resolve_thread_parent_id(None, &session_source), + Some(parent_thread_id) + ); + } +} + #[cfg(test)] #[path = "thread_processor_tests.rs"] mod thread_processor_tests; diff --git a/codex-rs/tui/src/lib.rs b/codex-rs/tui/src/lib.rs index 0302d5503..97d35ff6e 100644 --- a/codex-rs/tui/src/lib.rs +++ b/codex-rs/tui/src/lib.rs @@ -301,6 +301,8 @@ const TUI_LOG_FILE_NAME: &str = "codex-tui.log"; #[cfg(unix)] const AUTO_CONNECT_DAEMON_CONNECT_TIMEOUT: std::time::Duration = std::time::Duration::from_millis(50); +#[cfg(unix)] +const AUTO_CONNECT_DAEMON_VERSION_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(2); #[cfg(unix)] #[derive(Debug, PartialEq, Eq)] @@ -487,8 +489,8 @@ async fn connect_remote_app_server( #[cfg(unix)] async fn maybe_probe_default_daemon_socket(codex_home: &Path) -> Option { - match probe_default_daemon_socket_with(codex_home, |_| async { - Ok(env!("CARGO_PKG_VERSION").to_string()) + match probe_default_daemon_socket_with(codex_home, |socket_path| async move { + codex_app_server_daemon::probe_app_server_version(&socket_path).await }) .await { @@ -520,9 +522,43 @@ where ) .await { - Ok(Ok(_stream)) => { - let _ = probe_version(socket_path.as_path().to_path_buf()).await; - DefaultDaemonSocketProbe::Reuse(socket_path) + Ok(Ok(stream)) => { + drop(stream); + match tokio::time::timeout( + AUTO_CONNECT_DAEMON_VERSION_TIMEOUT, + probe_version(socket_path.as_path().to_path_buf()), + ) + .await + { + Ok(Ok(version)) if version == env!("CARGO_PKG_VERSION") => { + DefaultDaemonSocketProbe::Reuse(socket_path) + } + Ok(Ok(version)) => { + tracing::debug!( + socket_path = %socket_path.display(), + daemon_version = %version, + tui_version = env!("CARGO_PKG_VERSION"), + "default app-server daemon version does not match; using embedded app-server" + ); + DefaultDaemonSocketProbe::UseEmbedded + } + Ok(Err(err)) => { + tracing::debug!( + %err, + socket_path = %socket_path.display(), + "failed to read default app-server daemon version; using embedded app-server" + ); + DefaultDaemonSocketProbe::UseEmbedded + } + Err(_) => { + tracing::debug!( + socket_path = %socket_path.display(), + timeout_ms = AUTO_CONNECT_DAEMON_VERSION_TIMEOUT.as_millis(), + "timed out reading default app-server daemon version; using embedded app-server" + ); + DefaultDaemonSocketProbe::UseEmbedded + } + } } Ok(Err(err)) => { tracing::debug!(%err, socket_path = %socket_path.display(), "skipping default app-server daemon socket"); @@ -549,16 +585,23 @@ async fn maybe_start_default_daemon_socket( codex_home: &Path, codex_bin: Option<&Path>, ) -> Option { - maybe_start_default_daemon_socket_with(codex_home, codex_bin, |codex_bin| async move { - codex_app_server_daemon::ensure_local_daemon_started( - codex_app_server_daemon::LocalDaemonStartOptions { codex_bin }, - ) - .await - }) + maybe_start_default_daemon_socket_with_probes( + codex_home, + codex_bin, + |socket_path| async move { + codex_app_server_daemon::probe_app_server_version(&socket_path).await + }, + |codex_bin| async move { + codex_app_server_daemon::ensure_local_daemon_started( + codex_app_server_daemon::LocalDaemonStartOptions { codex_bin }, + ) + .await + }, + ) .await } -#[cfg(unix)] +#[cfg(all(unix, test))] async fn maybe_start_default_daemon_socket_with( codex_home: &Path, codex_bin: Option<&Path>, @@ -2573,7 +2616,7 @@ mod tests { #[cfg(unix)] #[tokio::test] - async fn default_daemon_auto_connect_probes_socket_only() -> color_eyre::Result<()> { + async fn default_daemon_auto_connect_reuses_matching_socket() -> color_eyre::Result<()> { let codex_home = TempDir::new()?; let socket_path = codex_app_server_client::app_server_control_socket_path(codex_home.path())?; @@ -2581,8 +2624,11 @@ mod tests { let _listener = tokio::net::UnixListener::bind(socket_path.as_path())?; assert_eq!( - maybe_probe_default_daemon_socket(codex_home.path()).await, - Some(socket_path) + probe_default_daemon_socket_with(codex_home.path(), |_| async { + Ok(env!("CARGO_PKG_VERSION").to_string()) + }) + .await, + DefaultDaemonSocketProbe::Reuse(socket_path) ); Ok(()) } @@ -2685,6 +2731,72 @@ mod tests { Ok(()) } + #[cfg(unix)] + #[tokio::test] + async fn default_daemon_auto_start_uses_embedded_when_version_probe_fails() + -> color_eyre::Result<()> { + let codex_home = TempDir::new()?; + let socket_path = + codex_app_server_client::app_server_control_socket_path(codex_home.path())?; + std::fs::create_dir_all(socket_path.as_path().parent().expect("socket parent"))?; + let _listener = tokio::net::UnixListener::bind(socket_path.as_path())?; + let start_count = Arc::new(AtomicUsize::new(0)); + let start_count_for_closure = Arc::clone(&start_count); + let unexpected_socket_path = codex_home.path().join("unexpected.sock"); + + let resolved_socket_path = maybe_start_default_daemon_socket_with_probes( + codex_home.path(), + Some(std::path::Path::new("/bin/codewith")), + |_| async { anyhow::bail!("test handshake failure") }, + move |_| async move { + start_count_for_closure.fetch_add(1, Ordering::SeqCst); + Ok(unexpected_socket_path) + }, + ) + .await; + + assert_eq!(resolved_socket_path, None); + assert_eq!( + start_count.load(Ordering::SeqCst), + 0, + "a reachable daemon with a failed handshake must never be started or replaced" + ); + Ok(()) + } + + #[cfg(unix)] + #[tokio::test(start_paused = true)] + async fn default_daemon_auto_start_uses_embedded_when_version_probe_times_out() + -> color_eyre::Result<()> { + let codex_home = TempDir::new()?; + let socket_path = + codex_app_server_client::app_server_control_socket_path(codex_home.path())?; + std::fs::create_dir_all(socket_path.as_path().parent().expect("socket parent"))?; + let _listener = tokio::net::UnixListener::bind(socket_path.as_path())?; + let start_count = Arc::new(AtomicUsize::new(0)); + let start_count_for_closure = Arc::clone(&start_count); + let unexpected_socket_path = codex_home.path().join("unexpected.sock"); + + let resolved_socket_path = maybe_start_default_daemon_socket_with_probes( + codex_home.path(), + Some(std::path::Path::new("/bin/codewith")), + |_| std::future::pending::>(), + move |_| async move { + start_count_for_closure.fetch_add(1, Ordering::SeqCst); + Ok(unexpected_socket_path) + }, + ) + .await; + + assert_eq!(resolved_socket_path, None); + assert_eq!( + start_count.load(Ordering::SeqCst), + 0, + "a reachable unresponsive daemon must never be started or replaced" + ); + Ok(()) + } + #[cfg(unix)] #[tokio::test] async fn default_daemon_auto_start_starts_when_socket_is_missing() -> color_eyre::Result<()> { From 0cbac1b1d84c5e81d322bc3b017c676ccbec1436 Mon Sep 17 00:00:00 2001 From: Andrei Hasna Date: Sun, 9 Aug 2026 22:13:02 +0300 Subject: [PATCH 09/13] style(tui): annotate daemon timeout argument Agent: cossus --- codex-rs/tui/src/lib.rs | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/codex-rs/tui/src/lib.rs b/codex-rs/tui/src/lib.rs index 97d35ff6e..77b9d0c78 100644 --- a/codex-rs/tui/src/lib.rs +++ b/codex-rs/tui/src/lib.rs @@ -302,7 +302,8 @@ const TUI_LOG_FILE_NAME: &str = "codex-tui.log"; const AUTO_CONNECT_DAEMON_CONNECT_TIMEOUT: std::time::Duration = std::time::Duration::from_millis(50); #[cfg(unix)] -const AUTO_CONNECT_DAEMON_VERSION_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(2); +const AUTO_CONNECT_DAEMON_VERSION_TIMEOUT: std::time::Duration = + std::time::Duration::from_secs(/*secs*/ 2); #[cfg(unix)] #[derive(Debug, PartialEq, Eq)] From 7d76e5ac5c0359638158d77701c27f483f00c63e Mon Sep 17 00:00:00 2001 From: Andrei Hasna Date: Sun, 9 Aug 2026 22:16:37 +0300 Subject: [PATCH 10/13] test(app-server): subscribe parent without resume race Agent: cossus --- .../suite/v2/connection_handling_websocket.rs | 29 +++++++------------ 1 file changed, 11 insertions(+), 18 deletions(-) diff --git a/codex-rs/app-server/tests/suite/v2/connection_handling_websocket.rs b/codex-rs/app-server/tests/suite/v2/connection_handling_websocket.rs index e0924ceb9..ac9c50540 100644 --- a/codex-rs/app-server/tests/suite/v2/connection_handling_websocket.rs +++ b/codex-rs/app-server/tests/suite/v2/connection_handling_websocket.rs @@ -17,8 +17,6 @@ use codex_app_server_protocol::JSONRPCResponse; use codex_app_server_protocol::RequestId; use codex_app_server_protocol::ThreadLoadedListParams; use codex_app_server_protocol::ThreadLoadedListResponse; -use codex_app_server_protocol::ThreadResumeParams; -use codex_app_server_protocol::ThreadResumeResponse; use codex_app_server_protocol::ThreadStartParams; use codex_app_server_protocol::ThreadStartResponse; use codex_app_server_protocol::ThreadStartedNotification; @@ -168,22 +166,17 @@ async fn websocket_parent_subscriber_inherits_spawned_child_subscription() -> Re initialize_websocket_client(&mut ws1, /*id*/ 1, "ws_parent_owner").await?; initialize_websocket_client(&mut ws2, /*id*/ 2, "ws_parent_subscriber").await?; - let parent = start_thread_with_notification(&mut ws1, /*id*/ 20).await?; - read_thread_started_for_thread(&mut ws2, &parent.id).await?; - - send_request( - &mut ws2, - "thread/resume", - /*id*/ 21, - Some(serde_json::to_value(ThreadResumeParams { - thread_id: parent.id.clone(), - ..Default::default() - })?), - ) - .await?; - let resume_response = read_response_for_id(&mut ws2, /*id*/ 21).await?; - let resume: ThreadResumeResponse = to_response(resume_response)?; - assert_eq!(resume.thread.id, parent.id); + let parent = start_thread_with_notification(&mut ws2, /*id*/ 20).await?; + let discovered = read_thread_started_for_thread(&mut ws1, &parent.id).await?; + assert_eq!( + discovered.thread.id, parent.id, + "CONTROL: the turn-driving connection must discover the parent without subscribing" + ); + assert_eq!( + unsubscribe_thread(&mut ws1, /*id*/ 21, &parent.id).await?, + ThreadUnsubscribeStatus::NotSubscribed, + "CONTROL: global discovery must not subscribe the turn-driving connection" + ); let child_id = spawn_child_from_parent(&mut ws1, &mut ws2, /*id*/ 22, &parent.id).await?; assert_eq!( From e86bc98d44d84caec529e093fe6acb9e3fdcabaa Mon Sep 17 00:00:00 2001 From: Andrei Hasna Date: Sun, 9 Aug 2026 22:33:11 +0300 Subject: [PATCH 11/13] test(tui): avoid awaiting with held store lock Agent: cossus --- codex-rs/tui/src/app/tests.rs | 19 +++++++++++++++++-- 1 file changed, 17 insertions(+), 2 deletions(-) diff --git a/codex-rs/tui/src/app/tests.rs b/codex-rs/tui/src/app/tests.rs index 4d43d6618..39a812119 100644 --- a/codex-rs/tui/src/app/tests.rs +++ b/codex-rs/tui/src/app/tests.rs @@ -2892,7 +2892,17 @@ async fn routine_notification_does_not_scan_every_inactive_thread_store() -> Res app.thread_event_channels.insert(thread_id, channel); } let blocked_store = blocked_store.expect("at least one inactive thread store"); - let blocked_guard = blocked_store.lock().await; + let (lock_acquired_tx, lock_acquired_rx) = tokio::sync::oneshot::channel(); + let (release_lock_tx, release_lock_rx) = std::sync::mpsc::channel(); + let blocked_store_for_task = Arc::clone(&blocked_store); + let blocked_task = tokio::task::spawn_blocking(move || { + let _guard = blocked_store_for_task.blocking_lock(); + let _ = lock_acquired_tx.send(()); + let _ = release_lock_rx.recv(); + }); + lock_acquired_rx + .await + .expect("blocking lock-holder task should acquire the inactive store"); let control = time::timeout( std::time::Duration::from_millis(/*millis*/ 250), @@ -2917,7 +2927,12 @@ async fn routine_notification_does_not_scan_every_inactive_thread_store() -> Res ), ) .await; - drop(blocked_guard); + release_lock_tx + .send(()) + .expect("blocking lock-holder task should remain available"); + blocked_task + .await + .expect("blocking lock-holder task should exit cleanly"); routed.expect( "routine notification routing must not wait for unrelated inactive thread stores", From 9d64a9f5f35edb4c60f7fbf4e33f499e06e2ca31 Mon Sep 17 00:00:00 2001 From: Andrei Hasna Date: Sun, 9 Aug 2026 22:41:01 +0300 Subject: [PATCH 12/13] test(app-server): remove stale websocket helper import Agent: cossus --- .../app-server/tests/suite/v2/connection_handling_websocket.rs | 1 - 1 file changed, 1 deletion(-) diff --git a/codex-rs/app-server/tests/suite/v2/connection_handling_websocket.rs b/codex-rs/app-server/tests/suite/v2/connection_handling_websocket.rs index ac9c50540..736f6689e 100644 --- a/codex-rs/app-server/tests/suite/v2/connection_handling_websocket.rs +++ b/codex-rs/app-server/tests/suite/v2/connection_handling_websocket.rs @@ -2,7 +2,6 @@ use anyhow::Context; use anyhow::Result; use anyhow::bail; use app_test_support::DISABLE_PLUGIN_STARTUP_TASKS_ARG; -use app_test_support::create_mock_responses_server_repeating_assistant; use app_test_support::create_mock_responses_server_sequence_unchecked; use app_test_support::to_response; use base64::Engine; From 88964ed9ee29fb309c088c9fe91c0b0331b349d1 Mon Sep 17 00:00:00 2001 From: Andrei Hasna Date: Sun, 9 Aug 2026 23:04:58 +0300 Subject: [PATCH 13/13] fix(app-server): annotate explicit parent argument Agent: cossus --- codex-rs/app-server/src/request_processors/thread_processor.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/codex-rs/app-server/src/request_processors/thread_processor.rs b/codex-rs/app-server/src/request_processors/thread_processor.rs index 8d16b60b6..67d31ecb4 100644 --- a/codex-rs/app-server/src/request_processors/thread_processor.rs +++ b/codex-rs/app-server/src/request_processors/thread_processor.rs @@ -5047,7 +5047,7 @@ mod lineage_parent_tests { }); assert_eq!( - resolve_thread_parent_id(None, &session_source), + resolve_thread_parent_id(/*explicit_parent_thread_id*/ None, &session_source), Some(parent_thread_id) ); }