diff --git a/codex-rs/app-server-client/src/cloud_backend/mod.rs b/codex-rs/app-server-client/src/cloud_backend/mod.rs index d2c9ca81cc..f1d9be3e80 100644 --- a/codex-rs/app-server-client/src/cloud_backend/mod.rs +++ b/codex-rs/app-server-client/src/cloud_backend/mod.rs @@ -141,6 +141,7 @@ pub fn connect(client: cloud::Client, account: protocol::GetAccountResponse) -> client, account, loaded: None, + warned_partial_history: None, }; tokio::spawn(async move { let mut replies = Replies::default(); @@ -211,7 +212,7 @@ async fn handle_request( replies.admit_turn(¶ms.thread_id, output)?; } match replies - .while_listening(requests.handle(request), output) + .while_listening(requests.handle(request, output), output) .await? { requests::Response::Value(value) => Ok(value), @@ -222,6 +223,7 @@ async fn handle_request( .replace(id, requests.client.attach(id), output) .await?; requests.loaded = Some(settings); + requests.warned_partial_history = None; Ok(result) } } diff --git a/codex-rs/app-server-client/src/cloud_backend/request_tests.rs b/codex-rs/app-server-client/src/cloud_backend/request_tests.rs new file mode 100644 index 0000000000..16fdd5f709 --- /dev/null +++ b/codex-rs/app-server-client/src/cloud_backend/request_tests.rs @@ -0,0 +1,110 @@ +#![cfg(test)] + +use super::*; +use pretty_assertions::assert_eq; + +#[test] +fn ordinary_multi_backend_resume_queries_cloud_interactive_threads_on_every_page() { + // The mixed picker has no shared cwd/provider filter. It sends conventional + // interactive source names, even when the connection belongs to Cloud. + for (cursor, state_db_only) in [(None, true), (Some("page-2"), false)] { + let params = serde_json::from_value(json!({ + "currentAccountOnly": true, + "cursor": cursor, + "limit": 25, + "sortKey": "updated_at", + "modelProviders": [], + "sourceKinds": ["cli", "vscode"], + "archived": false, + "useStateDbOnly": state_db_only, + })) + .unwrap(); + let query = thread_list_query(params).unwrap(); + assert_eq!(query.cursor.as_deref(), cursor); + assert_eq!(query.limit, Some(25)); + assert!(matches!(query.sort_key, cloud::ThreadSortKey::UpdatedAt)); + assert!(matches!( + query.sort_direction, + cloud::ThreadSortDirection::Desc + )); + assert!(query.originators.is_empty()); + } +} + +#[test] +fn unsupported_filters_do_not_return_unrelated_cloud_rows() { + for filter in [ + json!({"archived": true}), + json!({"sourceKinds": ["exec"]}), + json!({"attachedToThreadId": "other"}), + json!({"modelProviders": ["other"]}), + ] { + let params = serde_json::from_value(filter).unwrap(); + assert!(thread_list_query(params).is_err()); + } +} + +fn message() -> protocol::TurnStartParams { + let text = |text: &str| protocol::UserInput::Text { + text: text.to_owned(), + text_elements: Vec::new(), + }; + protocol::TurnStartParams { + thread_id: "thread-a".into(), + client_user_message_id: Some("submission-a".into()), + turn_trigger: Some("safety_recourse_follow_up".into()), + input: vec![text("one"), text("two")], + ..Default::default() + } +} + +#[test] +fn user_input_uses_stable_id_and_the_existing_cloud_configuration() { + let request = cloud_message(message()).unwrap(); + assert_eq!(request.thread_id, "thread-a"); + assert_eq!(request.client_user_message_id, "submission-a"); + assert_eq!(request.text, "one\ntwo"); + assert_eq!( + request.turn_trigger.as_deref(), + Some("safety_recourse_follow_up") + ); + assert!(matches!(request.mode, cloud::SubmissionMode::StartIfIdle)); + assert!(!request.shared_access); + + for tier in ["default", "priority"] { + assert!( + cloud_message(protocol::TurnStartParams { + service_tier_for_turn: Some(tier.into()), + ..message() + }) + .is_err() + ); + } + assert!( + cloud_message(protocol::TurnStartParams { + client_user_message_id: None, + ..message() + }) + .is_err() + ); +} + +#[test] +fn duplicate_and_queued_admissions_never_invent_a_running_turn() { + for outcome in [ + cloud::AdmissionOutcome::Duplicate { turn_id: None }, + cloud::AdmissionOutcome::Duplicate { + turn_id: Some("completed-or-running".into()), + }, + cloud::AdmissionOutcome::Queued { + submission_id: "queued".into(), + }, + ] { + assert!( + admitted_turn(outcome) + .unwrap_err() + .to_string() + .contains("already accepted") + ); + } +} diff --git a/codex-rs/app-server-client/src/cloud_backend/requests.rs b/codex-rs/app-server-client/src/cloud_backend/requests.rs index c8014a17e4..61fb99a408 100644 --- a/codex-rs/app-server-client/src/cloud_backend/requests.rs +++ b/codex-rs/app-server-client/src/cloud_backend/requests.rs @@ -1,5 +1,7 @@ //! Maps conversation requests onto native Cloud operations. +use super::events; +use super::send_notification; use codex_app_server_protocol as protocol; use codex_cloud_client as cloud; use color_eyre::eyre::Result; @@ -7,6 +9,7 @@ use color_eyre::eyre::bail; use color_eyre::eyre::eyre; use serde_json::Value; use serde_json::json; +use tokio::sync::mpsc; #[derive(Debug, thiserror::Error)] #[error("{0} is not supported by the cloud backend yet.")] @@ -16,6 +19,7 @@ pub(super) struct Requests { pub(super) client: cloud::Client, pub(super) account: protocol::GetAccountResponse, pub(super) loaded: Option>, + pub(super) warned_partial_history: Option, } pub(super) enum Response { @@ -44,11 +48,37 @@ impl Requests { Ok(json!({"thread": thread, "readState": null})) } - pub(super) async fn handle(&self, request: protocol::ClientRequest) -> Result { + pub(super) async fn handle( + &mut self, + request: protocol::ClientRequest, + output: &mpsc::UnboundedSender, + ) -> Result { use protocol::ClientRequest as Request; let value = match request { Request::GetAccount { .. } => serde_json::to_value(&self.account)?, + Request::ThreadList { params, .. } => { + let query = thread_list_query(params)?; + let mut page = self.client.list_threads(&query).await?; + for thread in &mut page.data { + use_paginated_history(thread); + } + serde_json::to_value(page)? + } + Request::ThreadStart { mut params, .. } => { + let model = params.model.take(); + let provider = params.model_provider.take(); + if params.ephemeral == Some(false) { + params.ephemeral = None; + } + if params != protocol::ThreadStartParams::default() { + bail!( + "Cloud conversations start with server defaults. Custom settings are not supported yet." + ); + } + let settings = self.client.prewarm_default_thread(model, provider).await?; + return attach_response(settings); + } Request::ThreadResume { params, .. } => { let supported = protocol::ThreadResumeParams { thread_id: params.thread_id.clone(), @@ -61,9 +91,7 @@ impl Requests { ); } 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)); + return attach_response(settings); } Request::ThreadLoadedList { .. } => { let ids = self @@ -73,8 +101,224 @@ impl Requests { .collect::>(); json!({"data": ids, "nextCursor": null}) } + Request::ThreadTurnsList { params, .. } => { + if params.items_view != Some(protocol::TurnItemsView::NotLoaded) { + bail!("Read cloud turn items separately with thread/items/list."); + } + let page = self + .client + .list_turns(&cloud::ListTurnsRequest { + thread_id: params.thread_id, + page: cloud::HistoryQuery { + cursor: params.cursor, + limit: params.limit, + order: history_order( + params + .sort_direction + .unwrap_or(protocol::SortDirection::Desc), + ), + ..Default::default() + }, + }) + .await?; + serde_json::to_value(protocol::ThreadTurnsListResponse { + data: page + .data + .into_iter() + .map(|entry| events::turn(entry.summary)) + .collect::>()?, + next_cursor: page.next_cursor, + backwards_cursor: page.backwards_cursor, + })? + } + Request::ThreadItemsList { params, .. } => { + let cursor = match params.cursor { + Some(protocol::ThreadItemsListCursor::Opaque(cursor)) => Some(cursor), + None => None, + Some(protocol::ThreadItemsListCursor::Anchor(_)) => { + bail!("Cloud item anchors are not supported yet.") + } + }; + let page = self + .client + .list_items(&cloud::ListItemsRequest { + thread_id: params.thread_id.clone(), + turn_id: params.turn_id, + page: cloud::HistoryQuery { + cursor, + limit: params.limit, + order: history_order( + params + .sort_direction + .unwrap_or(protocol::SortDirection::Asc), + ), + ..Default::default() + }, + }) + .await?; + let size = page.data.len(); + let data = page + .data + .into_iter() + .filter_map(|entry| { + Some(protocol::ThreadItemEntry { + turn_id: entry.turn_id, + item: events::item_summary(entry.summary?)?, + started_at_ms: entry.started_at_ms, + completed_at_ms: entry.completed_at_ms, + }) + }) + .collect::>(); + if data.len() != size + && self.warned_partial_history.as_ref() != Some(¶ms.thread_id) + { + self.warned_partial_history = Some(params.thread_id.clone()); + let warning = protocol::WarningNotification { + thread_id: Some(params.thread_id), + message: "Some Cloud history items cannot be displayed in this preview." + .to_owned(), + }; + send_notification(output, protocol::ServerNotification::Warning(warning)); + } + serde_json::to_value(protocol::ThreadItemsListResponse { + data, + next_cursor: page.next_cursor, + backwards_cursor: page.backwards_cursor, + })? + } + Request::TurnStart { params, .. } => { + let admission = self + .client + .add_user_message(&cloud_message(params)?) + .await?; + let turn = admitted_turn(admission.outcome)?; + serde_json::to_value(protocol::TurnStartResponse { turn })? + } + Request::TurnInterrupt { params, .. } => { + self.client + .interrupt_turn(&cloud::InterruptTurnRequest { + thread_id: params.thread_id, + turn_id: params.turn_id, + shared_access: false, + }) + .await?; + serde_json::to_value(protocol::TurnInterruptResponse {})? + } other => return Err(UnsupportedOperation(other.method_name().to_owned()).into()), }; Ok(Response::Value(value)) } } + +fn attach_response(mut settings: cloud::ThreadActivation) -> Result { + use_paginated_history(&mut settings.thread); + let resolved = serde_json::from_value(serde_json::to_value(settings)?) + .map_err(|_| eyre!("The cloud returned unsupported conversation settings."))?; + Ok(Response::Attach(resolved)) +} + +// Cloud serves history in pages even when an older thread used the legacy rollout format. +fn use_paginated_history(thread: &mut cloud::ThreadSnapshot) { + thread.history_mode = cloud::ThreadHistoryMode::Paginated; +} + +fn thread_list_query(params: protocol::ThreadListParams) -> Result { + use protocol::ThreadSourceKind::Cli; + use protocol::ThreadSourceKind::VsCode; + // The resume picker's ordinary interactive list maps to Cloud's own interactive default. + // Its explicit CLI/VS Code defaults are conventional source names. + let sources = params.source_kinds.as_deref().unwrap_or_default(); + if !matches!(sources, [] | [Cli, VsCode] | [VsCode, Cli]) + || params.archived == Some(true) + || params.cwd.is_some() + || params.search_term.is_some() + || params.section_id.is_some() + || params.project_id.is_some() + || params.parent_thread_id.is_some() + || params.ancestor_thread_id.is_some() + || params.attached_to_thread_id.is_some() + || params + .model_providers + .as_ref() + .is_some_and(|values| !values.is_empty()) + || params + .excluded_thread_ids + .as_ref() + .is_some_and(|values| !values.is_empty()) + { + bail!("These cloud conversation filters are not supported yet."); + } + Ok(cloud::ThreadListQuery { + cursor: params.cursor, + limit: params.limit, + originators: params.originators.unwrap_or_default(), + sort_key: match params.sort_key { + Some(protocol::ThreadSortKey::CreatedAt) | None => cloud::ThreadSortKey::CreatedAt, + Some(protocol::ThreadSortKey::UpdatedAt) => cloud::ThreadSortKey::UpdatedAt, + Some(protocol::ThreadSortKey::RecencyAt) => cloud::ThreadSortKey::RecencyAt, + Some(protocol::ThreadSortKey::SectionPosition) => { + bail!("Cloud section ordering is not supported yet.") + } + }, + sort_direction: match params.sort_direction { + Some(protocol::SortDirection::Asc) => cloud::ThreadSortDirection::Asc, + Some(protocol::SortDirection::Desc) | None => cloud::ThreadSortDirection::Desc, + }, + }) +} + +fn history_order(direction: protocol::SortDirection) -> cloud::HistoryOrder { + match direction { + protocol::SortDirection::Asc => cloud::HistoryOrder::Ascending, + protocol::SortDirection::Desc => cloud::HistoryOrder::Descending, + } +} + +fn cloud_message(mut params: protocol::TurnStartParams) -> Result { + let thread_id = std::mem::take(&mut params.thread_id); + let client_user_message_id = params + .client_user_message_id + .take() + .ok_or_else(|| eyre!("A cloud message needs a stable submission ID."))?; + let input = std::mem::take(&mut params.input); + let turn_trigger = params.turn_trigger.take(); + if params != protocol::TurnStartParams::default() { + bail!("Changing cloud settings or sending structured input is not supported yet."); + } + let text = input + .into_iter() + .map(|item| match item { + protocol::UserInput::Text { + text, + text_elements, + } if text_elements.is_empty() => Ok(text), + _ => bail!("Cloud conversations currently accept plain text only."), + }) + .collect::>>()? + .join("\n"); + Ok(cloud::AddUserMessageRequest { + thread_id, + client_user_message_id, + text, + turn_trigger, + mode: cloud::SubmissionMode::StartIfIdle, + shared_access: false, + }) +} + +fn admitted_turn(outcome: cloud::AdmissionOutcome) -> Result { + match outcome { + cloud::AdmissionOutcome::Started(turn) | cloud::AdmissionOutcome::Steered(turn) => { + events::turn(turn) + } + cloud::AdmissionOutcome::Duplicate { .. } | cloud::AdmissionOutcome::Queued { .. } => { + bail!( + "Cloud already accepted this message without returning a turn. Reopen the conversation to see its status before sending again." + ) + } + } +} + +#[cfg(test)] +#[path = "request_tests.rs"] +mod tests; diff --git a/codex-rs/cloud-client/examples/attach.rs b/codex-rs/cloud-client/examples/attach.rs index 9e8c550cd5..94f786cda5 100644 --- a/codex-rs/cloud-client/examples/attach.rs +++ b/codex-rs/cloud-client/examples/attach.rs @@ -25,6 +25,7 @@ async fn main() -> anyhow::Result<()> { thread_id: thread_id.clone(), client_user_message_id, text, + turn_trigger: None, mode: SubmissionMode::StartIfIdle, shared_access: false, }), diff --git a/codex-rs/cloud-client/src/turns.rs b/codex-rs/cloud-client/src/turns.rs index 0d8213ddc3..28f8b86359 100644 --- a/codex-rs/cloud-client/src/turns.rs +++ b/codex-rs/cloud-client/src/turns.rs @@ -22,6 +22,8 @@ pub struct AddUserMessageRequest { pub thread_id: String, pub client_user_message_id: String, pub text: String, + /// Optional classification supplied by the caller for this turn. + pub turn_trigger: Option, pub mode: SubmissionMode, pub shared_access: bool, } @@ -99,6 +101,7 @@ impl Client { }], }), thread_id: request.thread_id.clone(), + turn_trigger: request.turn_trigger.clone(), access: request.shared_access.then_some(1), mode: Some(match request.mode { SubmissionMode::Steer => 1, diff --git a/codex-rs/cloud-client/src/turns_tests.rs b/codex-rs/cloud-client/src/turns_tests.rs index 7c6ebd212d..adb7914e9b 100644 --- a/codex-rs/cloud-client/src/turns_tests.rs +++ b/codex-rs/cloud-client/src/turns_tests.rs @@ -7,6 +7,7 @@ fn request() -> AddUserMessageRequest { thread_id: "t".into(), client_user_message_id: "m".into(), text: "hello".into(), + turn_trigger: None, mode: SubmissionMode::StartIfIdle, shared_access: true, } @@ -36,9 +37,14 @@ async fn submission_preserves_admission_variants_and_wire_identity() { } payload.extend([0xa0, 0x06, 7]); // Unknown metadata survives. let mut fixture = fixture(response(vec![frame(&payload)], tonic::Status::ok(""))).await; + let turn_trigger = (mode_number == 3).then(|| "safety_recourse_follow_up".to_owned()); let result = fixture .client - .add_user_message(&AddUserMessageRequest { mode, ..request() }) + .add_user_message(&AddUserMessageRequest { + turn_trigger, + mode, + ..request() + }) .await .unwrap(); let turn = TurnSummary { @@ -88,6 +94,11 @@ async fn submission_preserves_admission_variants_and_wire_identity() { ), ), message(/*tag*/ 23, b"t"), + if mode_number == 3 { + message(/*tag*/ 25, b"safety_recourse_follow_up") + } else { + Vec::new() + }, vec![0xd0, 1, 1, 0xf0, 1, mode_number], ] .concat(); diff --git a/codex-rs/cloud-client/src/turns_wire.rs b/codex-rs/cloud-client/src/turns_wire.rs index 7333e24412..fc5a57bbde 100644 --- a/codex-rs/cloud-client/src/turns_wire.rs +++ b/codex-rs/cloud-client/src/turns_wire.rs @@ -9,6 +9,8 @@ pub(super) struct AddRequest { pub input: Option, #[prost(string, tag = "23")] pub thread_id: String, + #[prost(string, optional, tag = "25")] + pub turn_trigger: Option, #[prost(int32, optional, tag = "26")] pub access: Option, #[prost(int32, optional, tag = "30")]