diff --git a/codex-rs/app-server-transport/src/transport/remote_control/enroll.rs b/codex-rs/app-server-transport/src/transport/remote_control/enroll.rs index 9ec6e4d004..cb05230633 100644 --- a/codex-rs/app-server-transport/src/transport/remote_control/enroll.rs +++ b/codex-rs/app-server-transport/src/transport/remote_control/enroll.rs @@ -4,6 +4,7 @@ use super::protocol::RemoteControlTarget; use axum::http::HeaderMap; use codex_api::SharedAuthProvider; use codex_login::default_client::build_reqwest_client; +use codex_login::default_client::get_codex_advertised_version; use codex_state::RemoteControlEnrollmentRecord; use codex_state::StateRuntime; use std::io; @@ -201,7 +202,7 @@ pub(super) async fn enroll_remote_control_server( name: server_name.to_string(), os: std::env::consts::OS, arch: std::env::consts::ARCH, - app_server_version: env!("CARGO_PKG_VERSION"), + app_server_version: get_codex_advertised_version(), installation_id: installation_id.to_string(), }; let client = build_reqwest_client(); diff --git a/codex-rs/app-server-transport/src/transport/remote_control/tests.rs b/codex-rs/app-server-transport/src/transport/remote_control/tests.rs index fb1512fedb..780164f1be 100644 --- a/codex-rs/app-server-transport/src/transport/remote_control/tests.rs +++ b/codex-rs/app-server-transport/src/transport/remote_control/tests.rs @@ -25,6 +25,7 @@ use codex_app_server_protocol::ServerNotification; use codex_config::types::AuthCredentialsStoreMode; use codex_core::test_support::auth_manager_from_auth; use codex_core::test_support::auth_manager_from_auth_with_home; +use codex_login::default_client::get_codex_advertised_version; use codex_login::AuthDotJson; use codex_login::AuthManager; use codex_login::CodexAuth; @@ -1002,7 +1003,7 @@ async fn remote_control_http_mode_enrolls_before_connecting() { "name": expected_server_name, "os": std::env::consts::OS, "arch": std::env::consts::ARCH, - "app_server_version": env!("CARGO_PKG_VERSION"), + "app_server_version": get_codex_advertised_version(), "installation_id": TEST_INSTALLATION_ID, }) ); diff --git a/codex-rs/app-server/src/in_process.rs b/codex-rs/app-server/src/in_process.rs index c75c2d5ad1..2de1c34d98 100644 --- a/codex-rs/app-server/src/in_process.rs +++ b/codex-rs/app-server/src/in_process.rs @@ -449,6 +449,7 @@ async fn start_uninitialized(args: InProcessStartArgs) -> IoResult, connection_id: ConnectionId, + origin: ConnectionOrigin, request: JSONRPCRequest, transport: &AppServerTransport, session: Arc, @@ -548,6 +550,7 @@ impl MessageProcessor { self.handle_client_request( request_id.clone(), codex_request, + origin, Arc::clone(&session), /*outbound_initialized*/ None, request_context.clone(), @@ -571,6 +574,7 @@ impl MessageProcessor { pub(crate) async fn process_client_request( self: &Arc, connection_id: ConnectionId, + origin: ConnectionOrigin, request: ClientRequest, session: Arc, outbound_initialized: &AtomicBool, @@ -599,6 +603,7 @@ impl MessageProcessor { .handle_client_request( request_id.clone(), request, + origin, Arc::clone(&session), Some(outbound_initialized), request_context.clone(), @@ -654,6 +659,7 @@ impl MessageProcessor { pub(crate) async fn connection_initialized( &self, connection_id: ConnectionId, + origin: ConnectionOrigin, request_attestation: bool, ) { self.thread_processor @@ -662,6 +668,7 @@ impl MessageProcessor { ConnectionCapabilities { request_attestation, }, + origin, ) .await; } @@ -737,6 +744,7 @@ impl MessageProcessor { self: &Arc, connection_request_id: ConnectionRequestId, codex_request: ClientRequest, + origin: ConnectionOrigin, session: Arc, // `Some(...)` means the caller wants initialize to immediately mark the // connection outbound-ready. Websocket JSON-RPC calls pass `None` so @@ -763,6 +771,7 @@ impl MessageProcessor { ConnectionCapabilities { request_attestation: session.request_attestation(), }, + origin, ) .await; } diff --git a/codex-rs/app-server/src/message_processor_tracing_tests.rs b/codex-rs/app-server/src/message_processor_tracing_tests.rs index a9625d3086..d78bdd7a33 100644 --- a/codex-rs/app-server/src/message_processor_tracing_tests.rs +++ b/codex-rs/app-server/src/message_processor_tracing_tests.rs @@ -179,6 +179,7 @@ impl TracingHarness { self.processor .process_request( TEST_CONNECTION_ID, + crate::transport::ConnectionOrigin::Stdio, request, &AppServerTransport::Stdio, Arc::clone(&self.session), diff --git a/codex-rs/app-server/src/outgoing_message.rs b/codex-rs/app-server/src/outgoing_message.rs index 97ef37f742..94d3fda79f 100644 --- a/codex-rs/app-server/src/outgoing_message.rs +++ b/codex-rs/app-server/src/outgoing_message.rs @@ -164,6 +164,12 @@ impl ThreadScopedOutgoingMessageSender { if self.connection_ids.is_empty() { return; } + if matches!(notification, ServerNotification::TurnStarted(_)) { + tracing::info!( + connection_count = self.connection_ids.len(), + "sending thread-scoped turn/started notification" + ); + } self.outgoing .send_server_notification_to_connections(self.connection_ids.as_slice(), notification) .await; @@ -1319,4 +1325,50 @@ mod tests { .is_empty() ); } + + #[tokio::test] + async fn thread_scoped_notifications_fan_out_to_all_subscribed_connections() { + let (tx, mut rx) = mpsc::channel::(8); + let outgoing = Arc::new(OutgoingMessageSender::new( + tx, + codex_analytics::AnalyticsEventsClient::disabled(), + )); + let thread_id = ThreadId::new(); + let notification = ServerNotification::GuardianWarning(GuardianWarningNotification { + thread_id: thread_id.to_string(), + message: "warn".to_string(), + }); + let thread_outgoing = ThreadScopedOutgoingMessageSender::new( + outgoing, + vec![ConnectionId(2), ConnectionId(1)], + thread_id, + ); + + thread_outgoing + .send_server_notification(notification.clone()) + .await; + + let mut deliveries = Vec::new(); + for _ in 0..2 { + let envelope = rx.recv().await.expect("notification should be delivered"); + let OutgoingEnvelope::ToConnection { + connection_id, + message: OutgoingMessage::AppServerNotification(sent_notification), + .. + } = envelope + else { + panic!("expected thread-scoped notification delivery"); + }; + match sent_notification { + ServerNotification::GuardianWarning(sent) => { + assert_eq!(sent.thread_id, thread_id.to_string()); + assert_eq!(sent.message, "warn"); + } + other => panic!("unexpected notification: {other:?}"), + } + deliveries.push(connection_id); + } + deliveries.sort_by_key(|connection_id| connection_id.0); + assert_eq!(deliveries, vec![ConnectionId(1), ConnectionId(2)]); + } } 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 dd2dd6a57e..998b036b93 100644 --- a/codex-rs/app-server/src/request_processors/thread_processor.rs +++ b/codex-rs/app-server/src/request_processors/thread_processor.rs @@ -1,5 +1,6 @@ use super::*; use crate::error_code::method_not_found; +use crate::transport::ConnectionOrigin; use codex_protocol::models::BUILT_IN_PERMISSION_PROFILE_DANGER_FULL_ACCESS; use codex_protocol::models::BUILT_IN_PERMISSION_PROFILE_WORKSPACE; @@ -1159,6 +1160,34 @@ impl ThreadRequestProcessor { request_id.connection_id, "thread", ); + let local_stdio_connection_ids = listener_task_context + .thread_state_manager + .local_stdio_connection_ids_for_remote_control_thread_start(request_id.connection_id) + .await; + if !local_stdio_connection_ids.is_empty() { + tracing::info!( + local_stdio_connection_count = local_stdio_connection_ids.len(), + "remote-control thread/start attaching local stdio companions" + ); + } + for connection_id in local_stdio_connection_ids { + log_listener_attach_result( + super::thread_lifecycle::ensure_conversation_listener( + listener_task_context.clone(), + thread_id, + connection_id, + /*raw_events_enabled*/ false, + ) + .instrument(tracing::info_span!( + "app_server.thread_start.attach_stdio_listener", + otel.name = "app_server.thread_start.attach_stdio_listener", + )) + .await, + thread_id, + connection_id, + "thread", + ); + } listener_task_context .thread_watch_manager @@ -2243,9 +2272,10 @@ impl ThreadRequestProcessor { &self, connection_id: ConnectionId, capabilities: ConnectionCapabilities, + origin: ConnectionOrigin, ) { self.thread_state_manager - .connection_initialized(connection_id, capabilities) + .connection_initialized(connection_id, capabilities, origin) .await; } diff --git a/codex-rs/app-server/src/request_processors/thread_processor_tests.rs b/codex-rs/app-server/src/request_processors/thread_processor_tests.rs index aa33fc623f..d1bb66d1d8 100644 --- a/codex-rs/app-server/src/request_processors/thread_processor_tests.rs +++ b/codex-rs/app-server/src/request_processors/thread_processor_tests.rs @@ -48,6 +48,7 @@ mod thread_processor_behavior_tests { use super::super::*; use crate::outgoing_message::OutgoingEnvelope; use crate::outgoing_message::OutgoingMessage; + use crate::transport::ConnectionOrigin; use anyhow::Result; use chrono::DateTime; use chrono::Utc; @@ -1139,7 +1140,11 @@ mod thread_processor_behavior_tests { let (cancel_tx, cancel_rx) = oneshot::channel(); manager - .connection_initialized(connection, ConnectionCapabilities::default()) + .connection_initialized( + connection, + ConnectionCapabilities::default(), + ConnectionOrigin::Stdio, + ) .await; manager .try_ensure_connection_subscribed( @@ -1184,10 +1189,18 @@ mod thread_processor_behavior_tests { let (cancel_tx, mut cancel_rx) = oneshot::channel(); manager - .connection_initialized(connection_a, ConnectionCapabilities::default()) + .connection_initialized( + connection_a, + ConnectionCapabilities::default(), + ConnectionOrigin::Stdio, + ) .await; manager - .connection_initialized(connection_b, ConnectionCapabilities::default()) + .connection_initialized( + connection_b, + ConnectionCapabilities::default(), + ConnectionOrigin::Stdio, + ) .await; manager .try_ensure_connection_subscribed( @@ -1233,10 +1246,18 @@ mod thread_processor_behavior_tests { let connection_b = ConnectionId(2); manager - .connection_initialized(connection_a, ConnectionCapabilities::default()) + .connection_initialized( + connection_a, + ConnectionCapabilities::default(), + ConnectionOrigin::Stdio, + ) .await; manager - .connection_initialized(connection_b, ConnectionCapabilities::default()) + .connection_initialized( + connection_b, + ConnectionCapabilities::default(), + ConnectionOrigin::Stdio, + ) .await; manager .try_ensure_connection_subscribed( @@ -1283,7 +1304,11 @@ mod thread_processor_behavior_tests { let connection = ConnectionId(1); manager - .connection_initialized(connection, ConnectionCapabilities::default()) + .connection_initialized( + connection, + ConnectionCapabilities::default(), + ConnectionOrigin::Stdio, + ) .await; let threads_to_unload = manager.remove_connection(connection).await; assert_eq!(threads_to_unload, Vec::::new()); @@ -1317,6 +1342,7 @@ mod thread_processor_behavior_tests { ConnectionCapabilities { request_attestation: true, }, + ConnectionOrigin::Stdio, ) .await; manager @@ -1325,6 +1351,7 @@ mod thread_processor_behavior_tests { ConnectionCapabilities { request_attestation: true, }, + ConnectionOrigin::Stdio, ) .await; manager @@ -1333,10 +1360,15 @@ mod thread_processor_behavior_tests { ConnectionCapabilities { request_attestation: true, }, + ConnectionOrigin::Stdio, ) .await; manager - .connection_initialized(unsupported_connection, ConnectionCapabilities::default()) + .connection_initialized( + unsupported_connection, + ConnectionCapabilities::default(), + ConnectionOrigin::Stdio, + ) .await; assert!( @@ -1374,4 +1406,101 @@ mod thread_processor_behavior_tests { ); Ok(()) } + + #[tokio::test] + async fn remote_control_thread_start_also_subscribes_local_stdio_connections() -> Result<()> { + let manager = ThreadStateManager::new(); + let thread_id = ThreadId::from_string("4f1ddc9b-e4f5-4c59-b0d7-1ea3db8c9e4a")?; + let remote_control_connection = ConnectionId(1); + let stdio_connection = ConnectionId(2); + let websocket_connection = ConnectionId(3); + + manager + .connection_initialized( + remote_control_connection, + ConnectionCapabilities::default(), + ConnectionOrigin::RemoteControl, + ) + .await; + manager + .connection_initialized( + stdio_connection, + ConnectionCapabilities::default(), + ConnectionOrigin::Stdio, + ) + .await; + manager + .connection_initialized( + websocket_connection, + ConnectionCapabilities::default(), + ConnectionOrigin::WebSocket, + ) + .await; + + manager + .try_ensure_connection_subscribed( + thread_id, + remote_control_connection, + /*experimental_raw_events*/ false, + ) + .await + .expect("remote-control connection should be live"); + for connection_id in manager + .local_stdio_connection_ids_for_remote_control_thread_start(remote_control_connection) + .await + { + manager + .try_ensure_connection_subscribed( + thread_id, + connection_id, + /*experimental_raw_events*/ false, + ) + .await + .expect("stdio companion should be live"); + } + + let mut subscribed_connection_ids = manager.subscribed_connection_ids(thread_id).await; + subscribed_connection_ids.sort_by_key(|connection_id| connection_id.0); + assert_eq!( + subscribed_connection_ids, + vec![remote_control_connection, stdio_connection] + ); + Ok(()) + } + + #[tokio::test] + async fn non_remote_control_thread_start_does_not_add_stdio_companions() -> Result<()> { + let manager = ThreadStateManager::new(); + let stdio_connection = ConnectionId(1); + let websocket_connection = ConnectionId(2); + + manager + .connection_initialized( + stdio_connection, + ConnectionCapabilities::default(), + ConnectionOrigin::Stdio, + ) + .await; + manager + .connection_initialized( + websocket_connection, + ConnectionCapabilities::default(), + ConnectionOrigin::WebSocket, + ) + .await; + + assert_eq!( + manager + .local_stdio_connection_ids_for_remote_control_thread_start(stdio_connection) + .await, + Vec::::new() + ); + assert_eq!( + manager + .local_stdio_connection_ids_for_remote_control_thread_start(websocket_connection) + .await, + Vec::::new() + ); + Ok(()) + } } diff --git a/codex-rs/app-server/src/thread_state.rs b/codex-rs/app-server/src/thread_state.rs index 32dfcc325d..9a1807cdd8 100644 --- a/codex-rs/app-server/src/thread_state.rs +++ b/codex-rs/app-server/src/thread_state.rs @@ -1,5 +1,6 @@ use crate::outgoing_message::ConnectionId; use crate::outgoing_message::ConnectionRequestId; +use crate::transport::ConnectionOrigin; use codex_app_server_protocol::RequestId; use codex_app_server_protocol::ThreadGoal; use codex_app_server_protocol::ThreadHistoryBuilder; @@ -205,7 +206,7 @@ impl ThreadEntry { #[derive(Default)] struct ThreadStateManagerInner { - live_connections: HashMap, + live_connections: HashMap, threads: HashMap, thread_ids_by_connection: HashMap>, } @@ -215,6 +216,12 @@ pub(crate) struct ConnectionCapabilities { pub(crate) request_attestation: bool, } +#[derive(Clone, Copy)] +struct LiveConnection { + capabilities: ConnectionCapabilities, + origin: ConnectionOrigin, +} + #[derive(Clone, Default)] pub(crate) struct ThreadStateManager { state: Arc>, @@ -229,12 +236,15 @@ impl ThreadStateManager { &self, connection_id: ConnectionId, capabilities: ConnectionCapabilities, + origin: ConnectionOrigin, ) { - self.state - .lock() - .await - .live_connections - .insert(connection_id, capabilities); + self.state.lock().await.live_connections.insert( + connection_id, + LiveConnection { + capabilities, + origin, + }, + ); } pub(crate) async fn first_attestation_capable_connection_for_thread( @@ -251,12 +261,36 @@ impl ThreadStateManager { state .live_connections .get(connection_id)? + .capabilities .request_attestation .then_some(*connection_id) }) .min_by_key(|connection_id| connection_id.0) } + pub(crate) async fn local_stdio_connection_ids_for_remote_control_thread_start( + &self, + request_connection_id: ConnectionId, + ) -> Vec { + let state = self.state.lock().await; + let Some(request_connection) = state.live_connections.get(&request_connection_id) else { + return Vec::new(); + }; + if request_connection.origin != ConnectionOrigin::RemoteControl { + return Vec::new(); + } + + let mut connection_ids = state + .live_connections + .iter() + .filter_map(|(connection_id, live_connection)| { + (live_connection.origin == ConnectionOrigin::Stdio).then_some(*connection_id) + }) + .collect::>(); + connection_ids.sort_by_key(|connection_id| connection_id.0); + connection_ids + } + pub(crate) async fn subscribed_connection_ids(&self, thread_id: ThreadId) -> Vec { let state = self.state.lock().await; state diff --git a/codex-rs/app-server/src/transport.rs b/codex-rs/app-server/src/transport.rs index f7222bd28f..ff082e991d 100644 --- a/codex-rs/app-server/src/transport.rs +++ b/codex-rs/app-server/src/transport.rs @@ -30,6 +30,7 @@ pub(crate) use codex_app_server_transport::start_stdio_connection; pub(crate) use codex_app_server_transport::start_websocket_acceptor; pub(crate) struct ConnectionState { + pub(crate) origin: ConnectionOrigin, pub(crate) outbound_initialized: Arc, pub(crate) outbound_experimental_api_enabled: Arc, pub(crate) outbound_opted_out_notification_methods: Arc>>, @@ -38,12 +39,13 @@ pub(crate) struct ConnectionState { impl ConnectionState { pub(crate) fn new( - _origin: ConnectionOrigin, + origin: ConnectionOrigin, outbound_initialized: Arc, outbound_experimental_api_enabled: Arc, outbound_opted_out_notification_methods: Arc>>, ) -> Self { Self { + origin, outbound_initialized, outbound_experimental_api_enabled, outbound_opted_out_notification_methods, diff --git a/codex-rs/login/src/auth/default_client.rs b/codex-rs/login/src/auth/default_client.rs index ee51edf9b5..c4bd5a4f75 100644 --- a/codex-rs/login/src/auth/default_client.rs +++ b/codex-rs/login/src/auth/default_client.rs @@ -36,6 +36,9 @@ pub static USER_AGENT_SUFFIX: LazyLock>> = LazyLock::new(|| pub const DEFAULT_ORIGINATOR: &str = "codex_cli_rs"; pub const CODEX_INTERNAL_ORIGINATOR_OVERRIDE_ENV_VAR: &str = "CODEX_INTERNAL_ORIGINATOR_OVERRIDE"; pub const RESIDENCY_HEADER_NAME: &str = "x-openai-internal-codex-residency"; +const SOURCE_BUILD_VERSION: &str = "0.0.0"; +// Keep this aligned with the remote client minimum version that source builds should impersonate. +const SOURCE_BUILD_REMOTE_COMPAT_VERSION: &str = "0.129.0-alpha.6"; pub use codex_config::ResidencyRequirement; @@ -130,8 +133,15 @@ pub fn is_first_party_chat_originator(originator_value: &str) -> bool { originator_value == "codex_atlas" || originator_value == "codex_chatgpt_desktop" } +pub fn get_codex_advertised_version() -> &'static str { + match env!("CARGO_PKG_VERSION") { + SOURCE_BUILD_VERSION => SOURCE_BUILD_REMOTE_COMPAT_VERSION, + version => version, + } +} + pub fn get_codex_user_agent() -> String { - let build_version = env!("CARGO_PKG_VERSION"); + let build_version = get_codex_advertised_version(); let os_info = os_info::get(); let originator = originator(); let prefix = format!( diff --git a/codex-rs/login/src/auth/default_client_tests.rs b/codex-rs/login/src/auth/default_client_tests.rs index be8e6bc392..8658193029 100644 --- a/codex-rs/login/src/auth/default_client_tests.rs +++ b/codex-rs/login/src/auth/default_client_tests.rs @@ -7,10 +7,20 @@ use pretty_assertions::assert_eq; fn test_get_codex_user_agent() { let user_agent = get_codex_user_agent(); let originator = originator().value; - let prefix = format!("{originator}/"); + let prefix = format!("{originator}/{}", get_codex_advertised_version()); assert!(user_agent.starts_with(&prefix)); } +#[test] +fn source_builds_use_remote_compatible_advertised_version() { + let expected = match env!("CARGO_PKG_VERSION") { + "0.0.0" => "0.129.0-alpha.6", + version => version, + }; + + assert_eq!(get_codex_advertised_version(), expected); +} + #[test] fn is_first_party_originator_matches_known_values() { assert_eq!(is_first_party_originator(DEFAULT_ORIGINATOR), true);