diff --git a/codex-rs/app-server/src/outgoing_message.rs b/codex-rs/app-server/src/outgoing_message.rs index 647481cb3c..92e5a88391 100644 --- a/codex-rs/app-server/src/outgoing_message.rs +++ b/codex-rs/app-server/src/outgoing_message.rs @@ -797,12 +797,13 @@ impl OutgoingMessageSender { } return; } - for connection_id in connection_ids { + let messages = std::iter::repeat_n(outgoing_message, connection_ids.len()); + for (connection_id, message) in connection_ids.iter().zip(messages) { if let Err(err) = self .sender .send(OutgoingEnvelope::ToConnection { connection_id: *connection_id, - message: outgoing_message.clone(), + message, write_complete_tx: None, }) .await @@ -1365,41 +1366,66 @@ mod tests { } } + /// Empty target lists broadcast; explicit targets preserve order, payload, and one timestamp. #[tokio::test] async fn send_server_notification_to_connections_reuses_timestamp() { - let (tx, mut rx) = mpsc::channel::(2); - let outgoing = - OutgoingMessageSender::new(tx, codex_analytics::AnalyticsEventsClient::disabled()); + for connection_ids in [ + vec![], + vec![ConnectionId(1)], + vec![ConnectionId(1), ConnectionId(2)], + ] { + let (tx, mut rx) = mpsc::channel::(2); + let outgoing = + OutgoingMessageSender::new(tx, codex_analytics::AnalyticsEventsClient::disabled()); + let notification = ServerNotification::ConfigWarning(ConfigWarningNotification { + summary: "test".to_string(), + details: None, + path: None, + range: None, + }); + let expected = serde_json::to_value(¬ification).expect("serialize notification"); - outgoing - .send_server_notification_to_connections( - &[ConnectionId(1), ConnectionId(2)], - ServerNotification::ConfigWarning(ConfigWarningNotification { - summary: "test".to_string(), - details: None, - path: None, - range: None, - }), - ) - .await; + outgoing + .send_server_notification_to_connections(&connection_ids, notification) + .await; - let timestamps = [ - rx.recv() - .await - .expect("first connection should receive notification"), - rx.recv() - .await - .expect("second connection should receive notification"), - ] - .map(|envelope| match envelope { - OutgoingEnvelope::ToConnection { - message: OutgoingMessage::AppServerNotification(envelope), - .. - } => envelope.emitted_at_ms, - _ => panic!("expected targeted server notification"), - }); - - assert_eq!(timestamps[0], timestamps[1]); + let mut timestamps = Vec::new(); + for index in 0..connection_ids.len().max(1) { + let message = match rx.try_recv().expect("notification should be queued") { + OutgoingEnvelope::Broadcast { message } => { + assert!(connection_ids.is_empty()); + message + } + OutgoingEnvelope::ToConnection { + connection_id, + message, + write_complete_tx, + } => { + assert_eq!(connection_id, connection_ids[index]); + assert!(write_complete_tx.is_none()); + message + } + }; + let OutgoingMessage::AppServerNotification(envelope) = message else { + panic!("expected server notification"); + }; + assert_eq!( + serde_json::to_value(&envelope.notification).unwrap(), + expected + ); + timestamps.push( + envelope + .emitted_at_ms + .expect("notification has a timestamp"), + ); + } + assert!( + timestamps + .iter() + .all(|timestamp| *timestamp == timestamps[0]) + ); + assert!(rx.try_recv().is_err()); + } } #[tokio::test] diff --git a/codex-rs/app-server/src/transport.rs b/codex-rs/app-server/src/transport.rs index 98aeeeac84..4aef699835 100644 --- a/codex-rs/app-server/src/transport.rs +++ b/codex-rs/app-server/src/transport.rs @@ -230,11 +230,12 @@ pub(crate) async fn route_outgoing_envelope( }) .collect(); - for connection_id in target_connections { + let messages = std::iter::repeat_n(message, target_connections.len()); + for (connection_id, message) in target_connections.into_iter().zip(messages) { let _ = send_message_to_connection( connections, connection_id, - message.clone(), + message, /*write_complete_tx*/ None, ) .await; diff --git a/codex-rs/app-server/src/transport_tests.rs b/codex-rs/app-server/src/transport_tests.rs index 5b98f9bbc8..fe591ccbce 100644 --- a/codex-rs/app-server/src/transport_tests.rs +++ b/codex-rs/app-server/src/transport_tests.rs @@ -415,6 +415,56 @@ async fn command_execution_request_approval_keeps_additional_permissions_with_ca ); } +/// Broadcasts preserve their payload for eligible recipients and skip every other connection. +/// The empty eligible set must not accidentally send to an uninitialized or opted-out client. +#[tokio::test] +async fn broadcast_only_reaches_eligible_connections() { + for eligible_count in 0..=2 { + let notification = thread_realtime_started_notification(); + let method = notification.to_string(); + let message = app_server_notification(notification); + let expected = serde_json::to_value(&message).expect("serialize notification"); + let mut connections = HashMap::new(); + let mut receivers = Vec::new(); + let configs = [ + (false, true, HashSet::new()), + (true, false, HashSet::new()), + (true, true, HashSet::from([method])), + ] + .into_iter() + .chain((0..eligible_count).map(|_| (true, true, HashSet::new()))); + for (index, (initialized, experimental_api, opted_out_methods)) in configs.enumerate() { + let (writer_tx, writer_rx) = mpsc::channel(1); + connections.insert( + ConnectionId(index as u64), + OutboundConnectionState::new( + writer_tx, + Arc::new(AtomicBool::new(initialized)), + Arc::new(AtomicBool::new(experimental_api)), + Arc::new(RwLock::new(opted_out_methods)), + /*disconnect_sender*/ None, + ), + ); + receivers.push(writer_rx); + } + + route_outgoing_envelope(&mut connections, OutgoingEnvelope::Broadcast { message }).await; + + for (index, mut receiver) in receivers.into_iter().enumerate() { + if index < 3 { + assert!(receiver.try_recv().is_err()); + } else { + let queued = receiver + .try_recv() + .expect("eligible client receives notification"); + assert_eq!(serde_json::to_value(&queued.message).unwrap(), expected); + assert!(queued.write_complete_tx.is_none()); + assert!(receiver.try_recv().is_err()); + } + } + } +} + #[tokio::test] async fn broadcast_does_not_block_on_slow_connection() { let fast_connection_id = ConnectionId(1);