diff --git a/codex-rs/Cargo.lock b/codex-rs/Cargo.lock index 9e048223da..c2ce088a6f 100644 --- a/codex-rs/Cargo.lock +++ b/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", diff --git a/codex-rs/app-server-client/Cargo.toml b/codex-rs/app-server-client/Cargo.toml index 272173b83a..45a8ac0eb2 100644 --- a/codex-rs/app-server-client/Cargo.toml +++ b/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 } diff --git a/codex-rs/app-server-client/src/cloud_backend/connection_tests.rs b/codex-rs/app-server-client/src/cloud_backend/connection_tests.rs new file mode 100644 index 0000000000..d67380b25c --- /dev/null +++ b/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(()) +} diff --git a/codex-rs/app-server-client/src/cloud_backend/events.rs b/codex-rs/app-server-client/src/cloud_backend/events.rs new file mode 100644 index 0000000000..621ecdb0b7 --- /dev/null +++ b/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 { + 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> { + 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), + })) +} diff --git a/codex-rs/app-server-client/src/cloud_backend/mod.rs b/codex-rs/app-server-client/src/cloud_backend/mod.rs new file mode 100644 index 0000000000..1db1a3b70c --- /dev/null +++ b/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>; + +struct Attachment { + thread_id: String, + stream: ReplyStream, +} + +#[derive(Default)] +struct Replies { + active: Option, + // 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) { + 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( + &mut self, + future: impl Future, + output: &mpsc::UnboundedSender, + ) -> 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: &mpsc::UnboundedSender, + ) -> 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, + ) -> 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::() + .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, +) -> Result { + if let protocol::ClientRequest::ThreadRead { params, .. } = &request { + return requests.read_thread(params, replies.thread_status(¶ms.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(¶ms.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(¶ms.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, + 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, + 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; diff --git a/codex-rs/app-server-client/src/cloud_backend/requests.rs b/codex-rs/app-server-client/src/cloud_backend/requests.rs new file mode 100644 index 0000000000..c8014a17e4 --- /dev/null +++ b/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>, +} + +pub(super) enum Response { + Value(Value), + Attach(Box), +} + +impl Requests { + pub(super) fn read_thread( + &self, + params: &protocol::ThreadReadParams, + status: Option<&cloud::ThreadStatus>, + ) -> Result { + 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 { + 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(¶ms.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::>(); + json!({"data": ids, "nextCursor": null}) + } + other => return Err(UnsupportedOperation(other.method_name().to_owned()).into()), + }; + Ok(Response::Value(value)) + } +} diff --git a/codex-rs/app-server-client/src/lib.rs b/codex-rs/app-server-client/src/lib.rs index c4b9011f60..008e165e11 100644 --- a/codex-rs/app-server-client/src/lib.rs +++ b/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; diff --git a/codex-rs/cloud-client/src/activation.rs b/codex-rs/cloud-client/src/activation.rs index 327e6c6c57..1734c8586f 100644 --- a/codex-rs/cloud-client/src/activation.rs +++ b/codex-rs/cloud-client/src/activation.rs @@ -482,7 +482,7 @@ fn parse_subagent_source(value: wire::SubAgentSource) -> Result Result { +pub(super) fn parse_status(value: wire::ThreadStatus) -> Result { match required(value.value)? { wire::thread_status::Value::NotLoaded(_) => Ok(ThreadStatus::NotLoaded), wire::thread_status::Value::Idle(_) => Ok(ThreadStatus::Idle), diff --git a/codex-rs/cloud-client/src/transcript.rs b/codex-rs/cloud-client/src/transcript.rs index a355b0b894..5b146d55c9 100644 --- a/codex-rs/cloud-client/src/transcript.rs +++ b/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 for DecodeFailure { pub(super) fn required(value: Option) -> Result { 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 { + 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::>(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, Error> { match Self::decode_present_fields(thread_id, payload) { diff --git a/codex-rs/cloud-client/src/transcript_wire.rs b/codex-rs/cloud-client/src/transcript_wire.rs index 670a6b134b..9e68b6c43a 100644 --- a/codex-rs/cloud-client/src/transcript_wire.rs +++ b/codex-rs/cloud-client/src/transcript_wire.rs @@ -67,6 +67,13 @@ pub(super) struct Envelope { pub params: Option, } #[derive(Clone, Message)] +pub(super) struct StatusUpdate { + #[prost(message, optional, tag = "1")] + pub status: Option, + #[prost(string, optional, tag = "2")] + pub thread_id: Option, +} +#[derive(Clone, Message)] pub(super) struct PatchUpdate { #[prost(message, optional, tag = "1")] pub changes: Option,