Add a channel-backed cloud app-server client (#56037)

## What changed

- Add `CloudClient` and integrate it with `AppServerClient` and `AppServerRequestHandle` for requests, notifications, server-request replies, events, and shutdown.
- Keep retained request and reply handles bound to their cloud backend, with event delivery independent of request completion and a timeout for shutdown.
- Return no platform or Codex home metadata for cloud clients, and reject TUI filesystem operations for this backend.

## Testing

Add tests for backend ownership of retained handles, independent event delivery, errors after the command receiver closes, and shutdown timeouts when the worker does not respond.

GitOrigin-RevId: 4cf308dc0ea6fa6e7b14dd4542d93e8030b0162d

codex-rs/app-server-client/src/cloud.rs

@@ -0,0 +1,163 @@
+//! Channel-backed cloud client. Requests and retained replies stay on one owner;
+//! events are drained independently so a foreground request cannot block its response.
+
+use crate::AppServerEvent;
+use crate::RequestResult;
+use codex_app_server_protocol::ClientNotification;
+use codex_app_server_protocol::ClientRequest;
+use codex_app_server_protocol::JSONRPCErrorError;
+use codex_app_server_protocol::RequestId;
+use codex_app_server_protocol::Result as JsonRpcResult;
+use std::io;
+use tokio::sync::mpsc;
+use tokio::sync::oneshot;
+
+/// Operations the cloud worker must handle without forwarding to another backend.
+/// Unsupported operations must return an explicit error; they must not claim success.
+pub enum CloudOperation {
+    Request(Box<ClientRequest>),
+    Notify(ClientNotification),
+    Resolve {
+        request_id: RequestId,
+        result: JsonRpcResult,
+    },
+    Reject {
+        request_id: RequestId,
+        error: JSONRPCErrorError,
+    },
+    Shutdown,
+}
+
+/// An operation and its single response. Dropping a caller does not undo admission;
+/// cloud workers must not retry mutations when the receiver has gone away.
+pub struct CloudCommand {
+    pub operation: CloudOperation,
+    pub response: oneshot::Sender<io::Result<RequestResult>>,
+}
+
+#[derive(Clone)]
+pub struct CloudRequestHandle {
+    commands: mpsc::Sender<CloudCommand>,
+}
+
+pub struct CloudClient {
+    handle: CloudRequestHandle,
+    events: mpsc::UnboundedReceiver<AppServerEvent>,
+}
+
+impl CloudClient {
+    /// Creates a cloud transport. The owner must drain commands and stop when
+    /// its receiver closes. Event delivery is independent of request completion,
+    /// matching conventional transports; cloud workers must bound remote payloads.
+    pub fn channel(
+        capacity: usize,
+    ) -> (
+        Self,
+        mpsc::Receiver<CloudCommand>,
+        mpsc::UnboundedSender<AppServerEvent>,
+    ) {
+        let (commands, receiver) = mpsc::channel(capacity.max(1));
+        let (events, event_receiver) = mpsc::unbounded_channel();
+        (
+            Self {
+                handle: CloudRequestHandle { commands },
+                events: event_receiver,
+            },
+            receiver,
+            events,
+        )
+    }
+
+    pub(crate) fn request_handle(&self) -> CloudRequestHandle {
+        self.handle.clone()
+    }
+
+    pub(crate) async fn next_event(&mut self) -> Option<AppServerEvent> {
+        self.events.recv().await
+    }
+
+    pub(crate) async fn request(&self, request: ClientRequest) -> io::Result<RequestResult> {
+        self.handle.request(request).await
+    }
+
+    pub(crate) async fn notify(&self, notification: ClientNotification) -> io::Result<()> {
+        self.handle.unit(CloudOperation::Notify(notification)).await
+    }
+
+    pub(crate) async fn resolve_server_request(
+        &self,
+        request_id: RequestId,
+        result: JsonRpcResult,
+    ) -> io::Result<()> {
+        self.handle.resolve_server_request(request_id, result).await
+    }
+
+    pub(crate) async fn reject_server_request(
+        &self,
+        request_id: RequestId,
+        error: JSONRPCErrorError,
+    ) -> io::Result<()> {
+        self.handle.reject_server_request(request_id, error).await
+    }
+
+    pub(crate) async fn shutdown(self) -> io::Result<()> {
+        tokio::time::timeout(
+            crate::SHUTDOWN_TIMEOUT,
+            self.handle.unit(CloudOperation::Shutdown),
+        )
+        .await
+        .map_err(|_| io::Error::new(io::ErrorKind::TimedOut, "cloud backend shutdown timed out"))?
+    }
+}
+
+impl CloudRequestHandle {
+    async fn send(&self, operation: CloudOperation) -> io::Result<RequestResult> {
+        let (response, result) = oneshot::channel();
+        self.commands
+            .send(CloudCommand {
+                operation,
+                response,
+            })
+            .await
+            .map_err(|_| io::Error::new(io::ErrorKind::BrokenPipe, "cloud backend is closed"))?;
+        result.await.map_err(|_| {
+            io::Error::new(
+                io::ErrorKind::BrokenPipe,
+                "cloud backend dropped its response",
+            )
+        })?
+    }
+
+    async fn unit(&self, operation: CloudOperation) -> io::Result<()> {
+        self.send(operation)
+            .await?
+            .map(|_| ())
+            .map_err(|error| io::Error::other(error.message))
+    }
+
+    pub(crate) async fn request(&self, request: ClientRequest) -> io::Result<RequestResult> {
+        self.send(CloudOperation::Request(Box::new(request))).await
+    }
+
+    pub(crate) async fn resolve_server_request(
+        &self,
+        request_id: RequestId,
+        result: JsonRpcResult,
+    ) -> io::Result<()> {
+        self.unit(CloudOperation::Resolve { request_id, result })
+            .await
+    }
+
+    pub(crate) async fn reject_server_request(
+        &self,
+        request_id: RequestId,
+        error: JSONRPCErrorError,
+    ) -> io::Result<()> {
+        self.unit(CloudOperation::Reject { request_id, error })
+            .await
+    }
+}
+
+#[cfg(test)]
+#[path = "cloud_tests.rs"]
+mod tests;

codex-rs/app-server-client/src/cloud_tests.rs

@@ -0,0 +1,104 @@
+use super::*;
+use crate::AppServerClient;
+use crate::AppServerEvent;
+use codex_app_server_protocol::ThreadReadParams;
+use pretty_assertions::assert_eq;
+
+#[tokio::test]
+async fn retained_request_and_reply_stay_with_their_cloud_backend() {
+    let (first, mut first_commands, first_events) = CloudClient::channel(/*capacity*/ 1);
+    let (second, mut second_commands, _second_events) = CloudClient::channel(/*capacity*/ 1);
+    let mut first = AppServerClient::Cloud(first);
+    let second = AppServerClient::Cloud(second);
+    let handle = first.request_handle();
+    let reply = handle.server_request_reply_handle(RequestId::Integer(7));
+    let worker = tokio::spawn(async move {
+        let command = first_commands.recv().await.unwrap();
+        let CloudOperation::Request(request) = command.operation else {
+            panic!("expected thread read");
+        };
+        let ClientRequest::ThreadRead { params, .. } = *request else {
+            panic!("expected thread read");
+        };
+        assert_eq!(params.thread_id, "thread-1");
+        first_events
+            .send(AppServerEvent::Disconnected {
+                message: "stream ended".into(),
+            })
+            .unwrap();
+        command
+            .response
+            .send(Ok(Ok(serde_json::json!({"owner": "first"}))))
+            .unwrap();
+        let command = first_commands.recv().await.unwrap();
+        let CloudOperation::Resolve { request_id, result } = command.operation else {
+            panic!("expected retained reply");
+        };
+        assert_eq!(
+            (request_id, result),
+            (
+                RequestId::Integer(7),
+                serde_json::json!({"decision": "accept"})
+            )
+        );
+        command
+            .response
+            .send(Ok(Ok(serde_json::Value::Null)))
+            .unwrap();
+    });
+    assert_eq!(
+        handle
+            .request(ClientRequest::ThreadRead {
+                request_id: RequestId::Integer(1),
+                params: ThreadReadParams {
+                    thread_id: "thread-1".into(),
+                    include_turns: false
+                },
+            })
+            .await
+            .unwrap()
+            .unwrap(),
+        serde_json::json!({"owner": "first"})
+    );
+    reply
+        .resolve(serde_json::json!({"decision": "accept"}))
+        .await
+        .unwrap();
+    assert!(second_commands.try_recv().is_err());
+    assert!(matches!(
+        first.next_event().await,
+        Some(AppServerEvent::Disconnected { .. })
+    ));
+    worker.await.unwrap();
+    assert!(
+        handle
+            .request(ClientRequest::ThreadRead {
+                request_id: RequestId::Integer(2),
+                params: ThreadReadParams {
+                    thread_id: "thread-1".into(),
+                    include_turns: false
+                },
+            })
+            .await
+            .is_err()
+    );
+    drop(second);
+}
+
+#[tokio::test(start_paused = true)]
+async fn cloud_shutdown_times_out_when_owner_does_not_respond() {
+    let (client, mut commands, events) = CloudClient::channel(/*capacity*/ 1);
+    let error = tokio::time::timeout(
+        crate::SHUTDOWN_TIMEOUT + std::time::Duration::from_secs(1),
+        AppServerClient::Cloud(client).shutdown(),
+    )
+    .await
+    .expect("shutdown should be bounded")
+    .expect_err("an unresponsive cloud worker should time out");
+    assert_eq!(error.kind(), io::ErrorKind::TimedOut);
+    let command = commands.try_recv().expect("shutdown was admitted");
+    assert!(matches!(command.operation, CloudOperation::Shutdown));
+    assert!(command.response.is_closed());
+    assert!(events.is_closed());
+    assert!(commands.recv().await.is_none());
+}

codex-rs/app-server-client/src/lib.rs

@@ -20,8 +20,13 @@
 //! runtime remain bounded; the local consumer event queue is unbounded so
 //! unread notifications cannot prevent request responses from being delivered.
 
+mod cloud;
 mod path;
 mod remote;
+pub use cloud::CloudClient;
+pub use cloud::CloudCommand;
+pub use cloud::CloudOperation;
+pub use cloud::CloudRequestHandle;
 mod server_request;
 pub use server_request::AppServerReplyHandle;
 
@@ -345,11 +350,13 @@ pub struct InProcessAppServerRequestHandle {
 
 #[derive(Clone)]
 pub enum AppServerRequestHandle {
+    Cloud(CloudRequestHandle),
     InProcess(InProcessAppServerRequestHandle),
     Remote(crate::remote::RemoteAppServerRequestHandle),
 }
 
 pub enum AppServerClient {
+    Cloud(CloudClient),
     InProcess(InProcessAppServerClient),
     Remote(RemoteAppServerClient),
 }
@@ -708,6 +715,7 @@ impl AppServerRequestHandle {
         match self {
             Self::InProcess(handle) => handle.resolve_server_request(request_id, result).await,
             Self::Remote(handle) => handle.resolve_server_request(request_id, result).await,
+            Self::Cloud(handle) => handle.resolve_server_request(request_id, result).await,
         }
     }
 
@@ -719,6 +727,7 @@ impl AppServerRequestHandle {
         match self {
             Self::InProcess(handle) => handle.reject_server_request(request_id, error).await,
             Self::Remote(handle) => handle.reject_server_request(request_id, error).await,
+            Self::Cloud(handle) => handle.reject_server_request(request_id, error).await,
         }
     }
 
@@ -726,6 +735,7 @@ impl AppServerRequestHandle {
         match self {
             Self::InProcess(handle) => handle.request(request).await,
             Self::Remote(handle) => handle.request(request).await,
+            Self::Cloud(handle) => handle.request(request).await,
         }
     }
 
@@ -746,6 +756,7 @@ impl AppServerClient {
         match self {
             Self::InProcess(_) => Some(std::env::consts::FAMILY),
             Self::Remote(client) => client.platform_family(),
+            Self::Cloud(_) => None,
         }
     }
 
@@ -754,6 +765,7 @@ impl AppServerClient {
         match self {
             Self::InProcess(_) => Some(std::env::consts::OS),
             Self::Remote(client) => client.platform_os(),
+            Self::Cloud(_) => None,
         }
     }
 
@@ -763,6 +775,7 @@ impl AppServerClient {
                 local_codex_home.display().to_string(),
             )),
             Self::Remote(client) => client.codex_home().map(AppServerPath::from_app_server),
+            Self::Cloud(_) => None,
         }
     }
 
@@ -770,6 +783,7 @@ impl AppServerClient {
         match self {
             Self::InProcess(client) => client.request(request).await,
             Self::Remote(client) => client.request(request).await,
+            Self::Cloud(client) => client.request(request).await,
         }
     }
 
@@ -786,6 +800,7 @@ impl AppServerClient {
         match self {
             Self::InProcess(client) => client.notify(notification).await,
             Self::Remote(client) => client.notify(notification).await,
+            Self::Cloud(client) => client.notify(notification).await,
         }
     }
 
@@ -797,6 +812,7 @@ impl AppServerClient {
         match self {
             Self::InProcess(client) => client.resolve_server_request(request_id, result).await,
             Self::Remote(client) => client.resolve_server_request(request_id, result).await,
+            Self::Cloud(client) => client.resolve_server_request(request_id, result).await,
         }
     }
 
@@ -808,6 +824,7 @@ impl AppServerClient {
         match self {
             Self::InProcess(client) => client.reject_server_request(request_id, error).await,
             Self::Remote(client) => client.reject_server_request(request_id, error).await,
+            Self::Cloud(client) => client.reject_server_request(request_id, error).await,
         }
     }
 
@@ -815,6 +832,7 @@ impl AppServerClient {
         match self {
             Self::InProcess(client) => client.next_event().await.map(Into::into),
             Self::Remote(client) => client.next_event().await,
+            Self::Cloud(client) => client.next_event().await,
         }
     }
 
@@ -822,6 +840,7 @@ impl AppServerClient {
         match self {
             Self::InProcess(client) => client.shutdown().await,
             Self::Remote(client) => client.shutdown().await,
+            Self::Cloud(client) => client.shutdown().await,
         }
     }
 
@@ -829,6 +848,7 @@ impl AppServerClient {
         match self {
             Self::InProcess(client) => AppServerRequestHandle::InProcess(client.request_handle()),
             Self::Remote(client) => AppServerRequestHandle::Remote(client.request_handle()),
+            Self::Cloud(client) => AppServerRequestHandle::Cloud(client.request_handle()),
         }
     }
 }

codex-rs/tui/src/app_server_session/fs.rs

@@ -119,6 +119,9 @@ impl AppServerFileSystem {
     ) -> Result<T> {
         let request_id = RequestId::String(format!("tui-fs-{}", Uuid::new_v4()));
         match &self.request_handle {
+            AppServerRequestHandle::Cloud(_) => Err(color_eyre::eyre::eyre!(
+                "{method} is unavailable for this backend"
+            )),
             AppServerRequestHandle::Remote(handle) => {
                 let response = handle
                     .request_json_rpc(JSONRPCRequest {