From bd89f60348adda35e19685550174977749f15a8e Mon Sep 17 00:00:00 2001 From: jif-oai Date: Tue, 10 Feb 2026 14:24:07 +0000 Subject: [PATCH] Plan outbound router ordering fixes --- codex-rs/app-server/src/lib.rs | 18 +++++++++++++++++- codex-rs/app-server/src/transport.rs | 8 ++++++-- 2 files changed, 23 insertions(+), 3 deletions(-) diff --git a/codex-rs/app-server/src/lib.rs b/codex-rs/app-server/src/lib.rs index 3d78dcdadf..fed9314f25 100644 --- a/codex-rs/app-server/src/lib.rs +++ b/codex-rs/app-server/src/lib.rs @@ -366,6 +366,7 @@ pub async fn run_main_with_transport( } } + let transport_event_tx_for_outbound = transport_event_tx.clone(); let outbound_handle = tokio::spawn(async move { let mut outbound_connections = HashMap::::new(); loop { @@ -374,7 +375,22 @@ pub async fn run_main_with_transport( let Some(envelope) = envelope else { break; }; - route_outgoing_envelope(&mut outbound_connections, envelope).await; + let disconnected_connections = + route_outgoing_envelope(&mut outbound_connections, envelope).await; + let mut should_exit = false; + for connection_id in disconnected_connections { + if transport_event_tx_for_outbound + .send(TransportEvent::ConnectionClosed { connection_id }) + .await + .is_err() + { + should_exit = true; + break; + } + } + if should_exit { + break; + } } event = outbound_control_rx.recv() => { let Some(event) = event else { diff --git a/codex-rs/app-server/src/transport.rs b/codex-rs/app-server/src/transport.rs index e4155cf463..ae25720474 100644 --- a/codex-rs/app-server/src/transport.rs +++ b/codex-rs/app-server/src/transport.rs @@ -422,7 +422,8 @@ fn serialize_outgoing_message(outgoing_message: OutgoingMessage) -> Option, envelope: OutgoingEnvelope, -) { +) -> Vec { + let mut disconnected = Vec::new(); match envelope { OutgoingEnvelope::ToConnection { connection_id, @@ -433,10 +434,11 @@ pub(crate) async fn route_outgoing_envelope( "dropping message for disconnected connection: {:?}", connection_id ); - return; + return disconnected; }; if connection_state.writer.send(message).await.is_err() { connections.remove(&connection_id); + disconnected.push(connection_id); } } OutgoingEnvelope::Broadcast { message } => { @@ -457,10 +459,12 @@ pub(crate) async fn route_outgoing_envelope( }; if connection_state.writer.send(message.clone()).await.is_err() { connections.remove(&connection_id); + disconnected.push(connection_id); } } } } + disconnected } pub(crate) fn has_initialized_connections(