From 7b5094f28854094192e496d620f444a1d831d7e3 Mon Sep 17 00:00:00 2001 From: Ahmed Ibrahim <219906144+aibrahim-oai@users.noreply.github.com> Date: Tue, 17 Mar 2026 18:22:04 +0000 Subject: [PATCH] Preserve in-process request ordering for app-server clients Co-authored-by: Codex --- codex-rs/app-server-client/src/lib.rs | 23 ++++--- codex-rs/app-server/src/in_process.rs | 90 +++++++++++++++++++++++---- 2 files changed, 94 insertions(+), 19 deletions(-) diff --git a/codex-rs/app-server-client/src/lib.rs b/codex-rs/app-server-client/src/lib.rs index 1452eb590a..1eed8f315f 100644 --- a/codex-rs/app-server-client/src/lib.rs +++ b/codex-rs/app-server-client/src/lib.rs @@ -352,14 +352,21 @@ impl InProcessAppServerClient { command = command_rx.recv() => { match command { Some(ClientCommand::Request { request, response_tx }) => { - let request_sender = request_sender.clone(); - // Request waits happen on a detached task so - // this loop can keep draining runtime events - // while the request is blocked on client input. - tokio::spawn(async move { - let result = request_sender.request(*request).await; - let _ = response_tx.send(result); - }); + // Enqueue synchronously so the embedded runtime + // observes request order matching the caller's + // command order, then detach only the response + // wait so this loop can keep draining events. + match request_sender.start_request(*request) { + Ok(pending_request) => { + tokio::spawn(async move { + let result = pending_request.recv().await; + let _ = response_tx.send(result); + }); + } + Err(err) => { + let _ = response_tx.send(Err(err)); + } + } } Some(ClientCommand::Notify { notification, diff --git a/codex-rs/app-server/src/in_process.rs b/codex-rs/app-server/src/in_process.rs index 4288d15393..b28aac9aa6 100644 --- a/codex-rs/app-server/src/in_process.rs +++ b/codex-rs/app-server/src/in_process.rs @@ -92,7 +92,24 @@ const SHUTDOWN_TIMEOUT: Duration = Duration::from_secs(5); /// Default bounded channel capacity for in-process runtime queues. pub const DEFAULT_IN_PROCESS_CHANNEL_CAPACITY: usize = CHANNEL_CAPACITY; -type PendingClientRequestResponse = std::result::Result; +/// JSON-RPC application response returned by in-process client requests. +pub type InProcessRequestResponse = std::result::Result; + +/// Handle for a request that has been enqueued but whose response has not yet been awaited. +pub struct PendingInProcessRequest { + response_rx: oneshot::Receiver, +} + +impl PendingInProcessRequest { + pub async fn recv(self) -> IoResult { + self.response_rx.await.map_err(|err| { + IoError::new( + ErrorKind::BrokenPipe, + format!("in-process request response channel closed: {err}"), + ) + }) + } +} fn server_notification_requires_delivery(notification: &ServerNotification) -> bool { matches!(notification, ServerNotification::TurnCompleted(_)) @@ -171,7 +188,7 @@ pub enum InProcessServerEvent { enum InProcessClientMessage { Request { request: Box, - response_tx: oneshot::Sender, + response_tx: oneshot::Sender, }, Notification { notification: ClientNotification, @@ -200,18 +217,17 @@ pub struct InProcessClientSender { } impl InProcessClientSender { - pub async fn request(&self, request: ClientRequest) -> IoResult { + pub fn start_request(&self, request: ClientRequest) -> IoResult { let (response_tx, response_rx) = oneshot::channel(); self.try_send_client_message(InProcessClientMessage::Request { request: Box::new(request), response_tx, })?; - response_rx.await.map_err(|err| { - IoError::new( - ErrorKind::BrokenPipe, - format!("in-process request response channel closed: {err}"), - ) - }) + Ok(PendingInProcessRequest { response_rx }) + } + + pub async fn request(&self, request: ClientRequest) -> IoResult { + self.start_request(request)?.recv().await } pub fn notify(&self, notification: ClientNotification) -> IoResult<()> { @@ -263,6 +279,11 @@ pub struct InProcessClientHandle { } impl InProcessClientHandle { + /// Enqueues a typed client request and returns a handle for awaiting the response later. + pub fn start_request(&self, request: ClientRequest) -> IoResult { + self.client.start_request(request) + } + /// Sends a typed client request into the in-process runtime. /// /// The returned value is a transport-level `IoResult` containing either a @@ -270,7 +291,7 @@ impl InProcessClientHandle { /// request IDs unique among concurrent requests; reusing an in-flight ID /// produces an `INVALID_REQUEST` response and can make request routing /// ambiguous in the caller. - pub async fn request(&self, request: ClientRequest) -> IoResult { + pub async fn request(&self, request: ClientRequest) -> IoResult { self.client.request(request).await } @@ -490,7 +511,7 @@ fn start_uninitialized(args: InProcessStartArgs) -> InProcessClientHandle { processor.shutdown_threads().await; }); let mut pending_request_responses = - HashMap::>::new(); + HashMap::>::new(); let mut shutdown_ack = None; loop { @@ -732,6 +753,8 @@ mod tests { use super::*; use codex_app_server_protocol::ClientInfo; use codex_app_server_protocol::ConfigRequirementsReadResponse; + use codex_app_server_protocol::FuzzyFileSearchSessionStartParams; + use codex_app_server_protocol::FuzzyFileSearchSessionUpdateParams; use codex_app_server_protocol::SessionSource as ApiSessionSource; use codex_app_server_protocol::ThreadStartParams; use codex_app_server_protocol::ThreadStartResponse; @@ -740,6 +763,7 @@ mod tests { use codex_app_server_protocol::TurnStatus; use codex_core::config::ConfigBuilder; use pretty_assertions::assert_eq; + use tempfile::TempDir; async fn build_test_config() -> Config { match ConfigBuilder::default().build().await { @@ -857,6 +881,50 @@ mod tests { .expect("in-process runtime should shutdown cleanly"); } + #[tokio::test] + async fn start_request_preserves_enqueue_order_for_dependent_requests() { + let client = start_test_client(SessionSource::Cli).await; + let root = TempDir::new().expect("temp dir should be created"); + std::fs::write(root.path().join("alpha.txt"), "contents") + .expect("fixture file should be written"); + let root_path = root.path().to_string_lossy().to_string(); + + let start_request = client + .start_request(ClientRequest::FuzzyFileSearchSessionStart { + request_id: RequestId::Integer(10), + params: FuzzyFileSearchSessionStartParams { + session_id: "session-1".to_string(), + roots: vec![root_path], + }, + }) + .expect("session start should enqueue"); + let update_request = client + .start_request(ClientRequest::FuzzyFileSearchSessionUpdate { + request_id: RequestId::Integer(11), + params: FuzzyFileSearchSessionUpdateParams { + session_id: "session-1".to_string(), + query: "alpha".to_string(), + }, + }) + .expect("session update should enqueue"); + + start_request + .recv() + .await + .expect("session start transport should succeed") + .expect("session start should succeed"); + update_request + .recv() + .await + .expect("session update transport should succeed") + .expect("session update should succeed"); + + client + .shutdown() + .await + .expect("in-process runtime should shutdown cleanly"); + } + #[test] fn guaranteed_delivery_helpers_cover_terminal_notifications() { assert!(server_notification_requires_delivery(