Add a native cloud backend to the app-server client (#59493)

## What changed

Expose `cloud_backend::connect` to adapt an authenticated `codex-cloud-client` client to `AppServerClient`. Support account reads, resuming a conversation by ID with saved settings, reading its metadata, listing the loaded thread, and unsubscribing.

Forward turn lifecycle events and agent message deltas, and track thread status from attachment notifications. Keep the previous attachment when a replacement fails, report stream disconnections, and reject turns without an active attachment. Unsupported requests and approval replies return explicit errors.

## Testing

Add tests covering failed attachment replacement, turn admission after disconnect, and thread status tracking across notifications and reattachment.

GitOrigin-RevId: 5a131e814d588cadeb9fb94513b99e92b47e3da4

codex-rs/Cargo.lock

@@ -2205,6 +2205,7 @@ dependencies = [
  "codex-app-server-protocol",
  "codex-app-server-transport",
  "codex-arg0",
+ "codex-cloud-client",
  "codex-config",
  "codex-core",
  "codex-exec-server",
@@ -2213,11 +2214,13 @@ dependencies = [
  "codex-uds",
  "codex-utils-absolute-path",
  "codex-utils-rustls-provider",
+ "color-eyre",
  "futures",
  "pretty_assertions",
  "serde",
  "serde_json",
  "tempfile",
+ "thiserror 2.0.18",
  "tokio",
  "tokio-tungstenite",
  "toml 0.9.11+spec-1.1.0",

codex-rs/app-server-client/Cargo.toml

@@ -17,6 +17,7 @@ codex-app-server = { workspace = true }
 codex-app-server-protocol = { workspace = true }
 codex-app-server-transport = { workspace = true }
 codex-arg0 = { workspace = true }
+codex-cloud-client = { path = "../cloud-client" }
 codex-config = { workspace = true }
 codex-core = { workspace = true }
 codex-exec-server = { workspace = true }
@@ -25,9 +26,11 @@ codex-protocol = { workspace = true }
 codex-uds = { workspace = true }
 codex-utils-absolute-path = { workspace = true }
 codex-utils-rustls-provider = { workspace = true }
+color-eyre = { workspace = true }
 futures = { workspace = true }
 serde = { workspace = true }
 serde_json = { workspace = true }
+thiserror = { workspace = true }
 tokio = { workspace = true, features = ["sync", "time", "rt"] }
 tokio-tungstenite = { workspace = true }
 toml = { workspace = true }

codex-rs/app-server-client/src/cloud_backend/connection_tests.rs

@@ -0,0 +1,92 @@
+#![cfg(test)]
+
+//! Ensures only a successfully attached thread can receive new turns.
+
+use super::*;
+use futures::stream;
+use pretty_assertions::assert_eq;
+
+#[tokio::test]
+async fn failed_replacement_keeps_old_thread_until_new_attachment_succeeds() -> Result<()> {
+    let (output, mut notifications) = mpsc::unbounded_channel();
+    let mut replies = Replies::default();
+    replies
+        .replace("old", async { Ok(stream::pending().boxed()) }, &output)
+        .await?;
+
+    for failure in [
+        Err(cloud::Error::InvalidResponse),
+        Ok(stream::empty().boxed()),
+        Ok(stream::iter([Err(cloud::Error::InvalidResponse)]).boxed()),
+    ] {
+        assert!(
+            replies
+                .replace("new", async { failure }, &output)
+                .await
+                .is_err()
+        );
+        replies.admit_turn("old", &output)?;
+        assert!(replies.admit_turn("new", &output).is_err());
+    }
+
+    let (close, closed) = tokio::sync::oneshot::channel::<()>();
+    let stream = stream::once(async move {
+        let _ = closed.await;
+        Err(cloud::Error::InvalidResponse)
+    })
+    .boxed();
+    replies
+        .replace("new", async { Ok(stream) }, &output)
+        .await?;
+    replies.detach("old");
+    assert!(replies.admit_turn("old", &output).is_err());
+    replies.admit_turn("new", &output)?;
+
+    let _ = close.send(());
+    assert!(replies.admit_turn("new", &output).is_err());
+    let AppServerEvent::ServerNotification(notification) = notifications.try_recv()? else {
+        panic!("expected a disconnect notification");
+    };
+    assert_eq!(
+        serde_json::to_value(notification)?,
+        serde_json::to_value(cloud_error(
+            "new",
+            "Cloud stream ended. Reopen the conversation to reconnect."
+        ))?
+    );
+    Ok(())
+}
+
+#[tokio::test]
+async fn attachment_status_follows_its_server_notifications() -> Result<()> {
+    let (output, _notifications) = mpsc::unbounded_channel();
+    let mut replies = Replies::default();
+    let notifications: [&[u8]; 3] = [
+        b"\x1a\x0d\x12\x0b\x0a\x06\x22\x04\x0a\x02\x08\x01\x12\x01t", // t is waiting
+        b"\x1a\x09\x12\x07\x0a\x02\x12\x00\x12\x01x",                 // x is idle
+        b"\x1a\x09\x12\x07\x0a\x02\x12\x00\x12\x01t",                 // t is idle
+    ];
+    let stream =
+        stream::iter(notifications.map(|bytes| Ok(cloud::Event::Notification(bytes.into()))));
+    replies
+        .replace("t", async { Ok(stream.boxed()) }, &output)
+        .await?;
+
+    let waiting = cloud::ThreadStatus::Active {
+        active_flags: vec![cloud::ThreadActiveFlag::WaitingOnApproval],
+    };
+    assert_eq!(replies.thread_status("t"), Some(&waiting));
+    replies.forward_next(&output).await;
+    assert_eq!(replies.thread_status("t"), Some(&waiting));
+
+    replies.forward_next(&output).await;
+    assert_eq!(replies.thread_status("t"), Some(&cloud::ThreadStatus::Idle));
+    replies.forward_next(&output).await;
+    assert_eq!(replies.thread_status("t"), Some(&cloud::ThreadStatus::Idle));
+
+    replies
+        .replace("t", async { Ok(stream::pending().boxed()) }, &output)
+        .await?;
+    assert_eq!(replies.thread_status("t"), None);
+    Ok(())
+}

codex-rs/app-server-client/src/cloud_backend/events.rs

@@ -0,0 +1,67 @@
+//! Converts supported cloud transcript events without interpreting executor paths locally.
+
+use codex_app_server_protocol as protocol;
+use codex_cloud_client as cloud;
+use color_eyre::eyre::Result;
+use color_eyre::eyre::bail;
+
+pub(super) fn turn(summary: cloud::TurnSummary) -> Result<protocol::Turn> {
+    Ok(protocol::Turn {
+        id: summary.id,
+        root_turn_id: None,
+        items: Vec::new(),
+        items_view: protocol::TurnItemsView::NotLoaded,
+        status: match summary.status {
+            cloud::TurnStatus::Completed => protocol::TurnStatus::Completed,
+            cloud::TurnStatus::Interrupted => protocol::TurnStatus::Interrupted,
+            cloud::TurnStatus::Failed => protocol::TurnStatus::Failed,
+            cloud::TurnStatus::InProgress => protocol::TurnStatus::InProgress,
+            cloud::TurnStatus::Unknown(_) => {
+                bail!("The cloud returned an unsupported turn status.");
+            }
+        },
+        error: summary.error.map(|error| protocol::TurnError {
+            message: error.message,
+            additional_details: error.additional_details,
+            codex_error_info: None,
+            misalignment: None,
+        }),
+        started_at: summary.started_at,
+        completed_at: summary.completed_at,
+        duration_ms: summary.duration_ms,
+    })
+}
+
+pub(super) fn transcript(
+    event: cloud::TranscriptEvent,
+) -> Result<Option<protocol::ServerNotification>> {
+    use protocol::ServerNotification as Notification;
+    let thread_id = event.thread_id;
+    Ok(Some(match event.update {
+        cloud::TranscriptUpdate::TurnStarted(summary) => {
+            Notification::TurnStarted(protocol::TurnStartedNotification {
+                thread_id,
+                turn: turn(summary)?,
+            })
+        }
+        cloud::TranscriptUpdate::TurnCompleted(summary) => {
+            Notification::TurnCompleted(protocol::TurnCompletedNotification {
+                thread_id,
+                turn: turn(summary)?,
+            })
+        }
+        cloud::TranscriptUpdate::AgentMessageDelta(delta) => {
+            Notification::AgentMessageDelta(protocol::AgentMessageDeltaNotification {
+                thread_id,
+                turn_id: delta.turn_id,
+                item_id: delta.item_id,
+                delta: delta.delta,
+            })
+        }
+        cloud::TranscriptUpdate::ItemStarted(_)
+        | cloud::TranscriptUpdate::ItemCompleted(_)
+        | cloud::TranscriptUpdate::CommandOutputDelta(_)
+        | cloud::TranscriptUpdate::FileChangeOutputDelta(_)
+        | cloud::TranscriptUpdate::FileChangePatchUpdated(_) => return Ok(None),
+    }))
+}

codex-rs/app-server-client/src/cloud_backend/mod.rs

@@ -0,0 +1,278 @@
+//! Adapts native Cloud requests and the currently open conversation’s reply stream.
+//! A replacement becomes active only after its native attachment opens successfully.
+
+mod events;
+mod requests;
+
+use crate::AppServerClient;
+use crate::AppServerEvent;
+use crate::CloudClient;
+use crate::CloudOperation;
+use codex_app_server_protocol as protocol;
+use codex_cloud_client as cloud;
+use color_eyre::eyre::Result;
+use color_eyre::eyre::bail;
+use color_eyre::eyre::eyre;
+use futures::FutureExt;
+use futures::StreamExt;
+use futures::stream::BoxStream;
+use serde_json::Value;
+use tokio::sync::mpsc;
+
+type ReplyStream = BoxStream<'static, Result<cloud::Event, cloud::Error>>;
+
+struct Attachment {
+    thread_id: String,
+    stream: ReplyStream,
+}
+
+#[derive(Default)]
+struct Replies {
+    active: Option<Attachment>,
+    // Keep the last observed status through EOF until this conversation is unloaded.
+    latest_status: Option<(String, cloud::ThreadStatus)>,
+}
+
+impl Replies {
+    async fn forward_next(&mut self, output: &mpsc::UnboundedSender<AppServerEvent>) {
+        let Some(active) = &mut self.active else {
+            return std::future::pending().await;
+        };
+        match active.stream.next().await {
+            Some(Ok(event)) => forward(&active.thread_id, event, output, &mut self.latest_status),
+            Some(Err(_)) | None => {
+                send_notification(
+                    output,
+                    cloud_error(
+                        &active.thread_id,
+                        "Cloud stream ended. Reopen the conversation to reconnect.",
+                    ),
+                );
+                self.active = None;
+            }
+        }
+    }
+
+    async fn while_listening<T>(
+        &mut self,
+        future: impl Future<Output = T>,
+        output: &mpsc::UnboundedSender<AppServerEvent>,
+    ) -> T {
+        tokio::pin!(future);
+        loop {
+            tokio::select! {
+                biased;
+                result = &mut future => return result,
+                () = self.forward_next(output) => {}
+            }
+        }
+    }
+
+    async fn replace(
+        &mut self,
+        id: &str,
+        open: impl Future<Output = Result<ReplyStream, cloud::Error>>,
+        output: &mpsc::UnboundedSender<AppServerEvent>,
+    ) -> Result<()> {
+        let mut stream = self.while_listening(open, output).await?;
+        // A server can close Attach immediately after returning successful headers.
+        // A healthy idle stream needs no event to prove that it is ready.
+        let first_event = match tokio::task::unconstrained(stream.next()).now_or_never() {
+            Some(None | Some(Err(_))) => bail!(
+                "Cloud conversation activated but its stream did not connect. Reopen it before sending."
+            ),
+            Some(Some(Ok(event))) => Some(event),
+            None => None,
+        };
+        self.latest_status = None;
+        if let Some(event) = first_event {
+            forward(id, event, output, &mut self.latest_status);
+        }
+        self.active = Some(Attachment {
+            thread_id: id.to_owned(),
+            stream,
+        });
+        Ok(())
+    }
+
+    fn thread_status(&self, id: &str) -> Option<&cloud::ThreadStatus> {
+        self.latest_status
+            .as_ref()
+            .filter(|(thread_id, _)| thread_id == id)
+            .map(|(_, status)| status)
+    }
+
+    fn detach(&mut self, id: &str) {
+        if self.thread_status(id).is_some() {
+            self.latest_status = None;
+        }
+        if self
+            .active
+            .as_ref()
+            .is_some_and(|active| active.thread_id == id)
+        {
+            self.active = None;
+        }
+    }
+
+    fn admit_turn(
+        &mut self,
+        id: &str,
+        output: &mpsc::UnboundedSender<AppServerEvent>,
+    ) -> Result<()> {
+        // Notice a ready disconnect before sending. A continuous stream of events
+        // must still allow commands to make progress.
+        let _ = tokio::task::unconstrained(self.forward_next(output)).now_or_never();
+        if self
+            .active
+            .as_ref()
+            .is_none_or(|active| active.thread_id != id)
+        {
+            bail!("Cloud stream is not connected. Reopen the conversation before sending.");
+        }
+        Ok(())
+    }
+}
+
+/// Opens an app-server client backed by the authenticated native Cloud client.
+pub fn connect(client: cloud::Client, account: protocol::GetAccountResponse) -> AppServerClient {
+    let (app_client, mut commands, output) = CloudClient::channel(/*capacity*/ 32);
+    let mut requests = requests::Requests {
+        client,
+        account,
+        loaded: None,
+    };
+    tokio::spawn(async move {
+        let mut replies = Replies::default();
+        loop {
+            let command = tokio::select! {
+                _ = output.closed() => break,
+                () = replies.forward_next(&output) => continue,
+                command = commands.recv() => match command { Some(command) => command, None => break },
+            };
+            let result = match command.operation {
+                CloudOperation::Shutdown => {
+                    let _ = command.response.send(Ok(Ok(Value::Null)));
+                    break;
+                }
+                CloudOperation::Request(request) => tokio::select! {
+                    _ = output.closed() => break,
+                    result = handle_request(*request, &mut requests, &mut replies, &output) => result,
+                },
+                CloudOperation::Notify(_) => {
+                    Err(eyre!("Cloud client notifications are not supported yet."))
+                }
+                CloudOperation::Resolve { .. } | CloudOperation::Reject { .. } => Err(eyre!(
+                    "Answer this cloud approval in another client; TUI approval replies are not supported yet."
+                )),
+            };
+            let result = result.map_err(|error| protocol::JSONRPCErrorError {
+                code: if error
+                    .downcast_ref::<requests::UnsupportedOperation>()
+                    .is_some()
+                {
+                    -32601
+                } else {
+                    -32000
+                },
+                message: error.to_string(),
+                data: None,
+            });
+            let _ = command.response.send(Ok(result));
+        }
+    });
+    AppServerClient::Cloud(app_client)
+}
+
+async fn handle_request(
+    request: protocol::ClientRequest,
+    requests: &mut requests::Requests,
+    replies: &mut Replies,
+    output: &mpsc::UnboundedSender<AppServerEvent>,
+) -> Result<Value> {
+    if let protocol::ClientRequest::ThreadRead { params, .. } = &request {
+        return requests.read_thread(params, replies.thread_status(&params.thread_id));
+    }
+    if let protocol::ClientRequest::ThreadUnsubscribe { params, .. } = &request {
+        let status = if requests
+            .loaded
+            .as_ref()
+            .is_some_and(|loaded| loaded.thread.id == params.thread_id)
+        {
+            replies.detach(&params.thread_id);
+            requests.loaded = None;
+            protocol::ThreadUnsubscribeStatus::Unsubscribed
+        } else {
+            protocol::ThreadUnsubscribeStatus::NotSubscribed
+        };
+        return Ok(serde_json::json!({ "status": status }));
+    }
+    if let protocol::ClientRequest::TurnStart { params, .. } = &request {
+        replies.admit_turn(&params.thread_id, output)?;
+    }
+    match replies
+        .while_listening(requests.handle(request), output)
+        .await?
+    {
+        requests::Response::Value(value) => Ok(value),
+        requests::Response::Attach(settings) => {
+            let id = &settings.thread.id;
+            let result = serde_json::to_value(&settings)?;
+            replies
+                .replace(id, requests.client.attach(id), output)
+                .await?;
+            requests.loaded = Some(settings);
+            Ok(result)
+        }
+    }
+}
+
+fn forward(
+    id: &str,
+    event: cloud::Event,
+    output: &mpsc::UnboundedSender<AppServerEvent>,
+    latest_status: &mut Option<(String, cloud::ThreadStatus)>,
+) {
+    if let Some(status) = event.thread_status_for(id) {
+        *latest_status = Some((id.to_owned(), status));
+        return;
+    }
+    let notification = match event {
+        cloud::Event::Transcript(event) => match events::transcript(*event) {
+            Ok(Some(notification)) => notification,
+            Ok(None) => return,
+            Err(_) => cloud_error(id, "The cloud sent an unsupported transcript event."),
+        },
+        cloud::Event::ServerRequest(_) => protocol::ServerNotification::Warning(protocol::WarningNotification {
+            thread_id: Some(id.to_owned()),
+            message: "This cloud turn requested an unsupported interaction. Open it in another client to continue.".into(),
+        }),
+        cloud::Event::Notification(_) => return,
+    };
+    send_notification(output, notification);
+}
+
+fn send_notification(
+    output: &mpsc::UnboundedSender<AppServerEvent>,
+    notification: protocol::ServerNotification,
+) {
+    let _ = output.send(AppServerEvent::ServerNotification(Box::new(notification)));
+}
+
+fn cloud_error(id: &str, message: &str) -> protocol::ServerNotification {
+    protocol::ServerNotification::Error(protocol::ErrorNotification {
+        thread_id: id.to_owned(),
+        turn_id: String::new(),
+        will_retry: false,
+        error: protocol::TurnError {
+            message: message.to_owned(),
+            additional_details: None,
+            codex_error_info: None,
+            misalignment: None,
+        },
+    })
+}
+
+#[cfg(test)]
+#[path = "connection_tests.rs"]
+mod tests;

codex-rs/app-server-client/src/cloud_backend/requests.rs

@@ -0,0 +1,80 @@
+//! Maps conversation requests onto native Cloud operations.
+
+use codex_app_server_protocol as protocol;
+use codex_cloud_client as cloud;
+use color_eyre::eyre::Result;
+use color_eyre::eyre::bail;
+use color_eyre::eyre::eyre;
+use serde_json::Value;
+use serde_json::json;
+
+#[derive(Debug, thiserror::Error)]
+#[error("{0} is not supported by the cloud backend yet.")]
+pub(super) struct UnsupportedOperation(pub(super) String);
+
+pub(super) struct Requests {
+    pub(super) client: cloud::Client,
+    pub(super) account: protocol::GetAccountResponse,
+    pub(super) loaded: Option<Box<protocol::ThreadResumeResponse>>,
+}
+
+pub(super) enum Response {
+    Value(Value),
+    Attach(Box<protocol::ThreadResumeResponse>),
+}
+
+impl Requests {
+    pub(super) fn read_thread(
+        &self,
+        params: &protocol::ThreadReadParams,
+        status: Option<&cloud::ThreadStatus>,
+    ) -> Result<Value> {
+        if params.include_turns {
+            bail!("Use paginated history for cloud conversations.");
+        }
+        let loaded = self
+            .loaded
+            .as_deref()
+            .filter(|loaded| loaded.thread.id == params.thread_id)
+            .ok_or_else(|| eyre!("Open this cloud conversation before reading its transcript."))?;
+        let mut thread = serde_json::to_value(&loaded.thread)?;
+        if let Some(status) = status {
+            thread["status"] = serde_json::to_value(status)?;
+        }
+        Ok(json!({"thread": thread, "readState": null}))
+    }
+
+    pub(super) async fn handle(&self, request: protocol::ClientRequest) -> Result<Response> {
+        use protocol::ClientRequest as Request;
+
+        let value = match request {
+            Request::GetAccount { .. } => serde_json::to_value(&self.account)?,
+            Request::ThreadResume { params, .. } => {
+                let supported = protocol::ThreadResumeParams {
+                    thread_id: params.thread_id.clone(),
+                    exclude_turns: true,
+                    ..Default::default()
+                };
+                if params != supported {
+                    bail!(
+                        "Cloud conversations resume by ID with saved settings. Set excludeTurns to true and load history separately."
+                    );
+                }
+                let settings = self.client.activate(&params.thread_id).await?;
+                let resolved = serde_json::from_value(serde_json::to_value(settings)?)
+                    .map_err(|_| eyre!("The cloud returned unsupported conversation settings."))?;
+                return Ok(Response::Attach(resolved));
+            }
+            Request::ThreadLoadedList { .. } => {
+                let ids = self
+                    .loaded
+                    .iter()
+                    .map(|loaded| &loaded.thread.id)
+                    .collect::<Vec<_>>();
+                json!({"data": ids, "nextCursor": null})
+            }
+            other => return Err(UnsupportedOperation(other.method_name().to_owned()).into()),
+        };
+        Ok(Response::Value(value))
+    }
+}

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

@@ -21,6 +21,7 @@
 //! unread notifications cannot prevent request responses from being delivered.
 
 mod cloud;
+pub mod cloud_backend;
 mod path;
 mod remote;
 mod remote_websocket;

codex-rs/cloud-client/src/activation.rs

@@ -482,7 +482,7 @@ fn parse_subagent_source(value: wire::SubAgentSource) -> Result<SubAgentSource,
     }
 }
 
-fn parse_status(value: wire::ThreadStatus) -> Result<ThreadStatus, Error> {
+pub(super) fn parse_status(value: wire::ThreadStatus) -> Result<ThreadStatus, Error> {
     match required(value.value)? {
         wire::thread_status::Value::NotLoaded(_) => Ok(ThreadStatus::NotLoaded),
         wire::thread_status::Value::Idle(_) => Ok(ThreadStatus::Idle),

codex-rs/cloud-client/src/transcript.rs

@@ -1,5 +1,7 @@
 //! Typed live transcript updates for the attached thread.
 use crate::Error;
+use crate::Event;
+use crate::ThreadStatus;
 use crate::transcript_wire as wire;
 use std::fmt;
 
@@ -97,6 +99,7 @@ redacted_debug!(
 );
 
 // Field numbers in the ASB Notification.value oneof.
+const THREAD_STATUS_CHANGED: u32 = 3;
 const TURN_STARTED: u32 = 21;
 const TURN_COMPLETED: u32 = 23;
 pub(super) const ITEM_STARTED: u32 = 27;
@@ -120,6 +123,25 @@ impl From<prost::DecodeError> for DecodeFailure {
 pub(super) fn required<T>(value: Option<T>) -> Result<T, DecodeFailure> {
     value.ok_or(DecodeFailure::MissingField)
 }
+impl Event {
+    /// Returns a supported status update for this thread from an Attach notification.
+    pub fn thread_status_for(&self, thread_id: &str) -> Option<ThreadStatus> {
+        let Self::Notification(bytes) = self else {
+            return None;
+        };
+        let (tag, values) = wire::notification(bytes).ok().flatten()?;
+        if tag != THREAD_STATUS_CHANGED {
+            return None;
+        }
+        let envelope = wire::merge_messages::<wire::Envelope<wire::StatusUpdate>>(values).ok()?;
+        let params = envelope.params?;
+        if params.thread_id.as_deref() != Some(thread_id) {
+            return None;
+        }
+        crate::activation::parse_status(params.status?).ok()
+    }
+}
+
 impl TranscriptEvent {
     pub(crate) fn decode(thread_id: &str, payload: &[u8]) -> Result<Option<Self>, Error> {
         match Self::decode_present_fields(thread_id, payload) {

codex-rs/cloud-client/src/transcript_wire.rs

@@ -67,6 +67,13 @@ pub(super) struct Envelope<T: Message + Default + std::fmt::Debug> {
     pub params: Option<T>,
 }
 #[derive(Clone, Message)]
+pub(super) struct StatusUpdate {
+    #[prost(message, optional, tag = "1")]
+    pub status: Option<crate::activation_wire::ThreadStatus>,
+    #[prost(string, optional, tag = "2")]
+    pub thread_id: Option<String>,
+}
+#[derive(Clone, Message)]
 pub(super) struct PatchUpdate {
     #[prost(message, optional, tag = "1")]
     pub changes: Option<Changes>,