diff --git a/codex-rs/Cargo.lock b/codex-rs/Cargo.lock index 9a72d028f3..59d3eb730b 100644 --- a/codex-rs/Cargo.lock +++ b/codex-rs/Cargo.lock @@ -2232,6 +2232,7 @@ dependencies = [ "toml 0.9.11+spec-1.1.0", "tracing", "url", + "wiremock", ] [[package]] diff --git a/codex-rs/app-server-client/Cargo.toml b/codex-rs/app-server-client/Cargo.toml index 45a8ac0eb2..6d9201276c 100644 --- a/codex-rs/app-server-client/Cargo.toml +++ b/codex-rs/app-server-client/Cargo.toml @@ -42,3 +42,4 @@ pretty_assertions = { workspace = true } serde_json = { workspace = true } tempfile = { workspace = true } tokio = { workspace = true, features = ["macros", "rt-multi-thread"] } +wiremock = { workspace = true } diff --git a/codex-rs/app-server-client/src/cloud.rs b/codex-rs/app-server-client/src/cloud.rs index bc5260aa4e..03553d14db 100644 --- a/codex-rs/app-server-client/src/cloud.rs +++ b/codex-rs/app-server-client/src/cloud.rs @@ -3,11 +3,14 @@ use crate::AppServerEvent; use crate::RequestResult; +use crate::TypedRequestError; 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 codex_app_server_protocol::Turn; +use codex_app_server_protocol::TurnStartParams; use std::io; use tokio::sync::mpsc; use tokio::sync::oneshot; @@ -16,6 +19,7 @@ use tokio::sync::oneshot; /// Unsupported operations must return an explicit error; they must not claim success. pub enum CloudOperation { Request(Box), + SubmitUserMessage(Box), Notify(ClientNotification), Resolve { request_id: RequestId, @@ -28,6 +32,15 @@ pub enum CloudOperation { Shutdown, } +/// One server-decided admission, not completion. A duplicate carries identity +/// only; read history or observe the reply stream before inferring its status. +#[derive(serde::Serialize, serde::Deserialize)] +pub enum CloudUserMessageReceipt { + Started(Turn), + Steered(Turn), + Duplicate { turn_id: Option }, +} + /// 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 { @@ -111,6 +124,18 @@ impl CloudClient { } impl CloudRequestHandle { + /// Submit once with a stable message ID and saved thread settings. The server + /// starts or steers; an error can mean admission is unknown. Never auto-resend. + pub async fn submit_user_message( + &self, + params: TurnStartParams, + ) -> Result { + let response = self + .send(CloudOperation::SubmitUserMessage(Box::new(params))) + .await; + crate::decode_typed_response("cloud/submitUserMessage", response) + } + async fn send(&self, operation: CloudOperation) -> io::Result { let (response, result) = oneshot::channel(); self.commands 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 f1d9be3e80..7b7c86ed79 100644 --- a/codex-rs/app-server-client/src/cloud_backend/mod.rs +++ b/codex-rs/app-server-client/src/cloud_backend/mod.rs @@ -21,6 +21,11 @@ use tokio::sync::mpsc; type ReplyStream = BoxStream<'static, Result>; +/// Rejected locally before an admission request can be sent. +#[derive(Debug, thiserror::Error)] +#[error("{0}")] +struct PreflightRejection(&'static str); + struct Attachment { thread_id: String, stream: ReplyStream, @@ -119,7 +124,7 @@ impl Replies { &mut self, id: &str, output: &mpsc::UnboundedSender, - ) -> Result<()> { + ) -> Result<(), PreflightRejection> { // 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(); @@ -128,7 +133,9 @@ impl Replies { .as_ref() .is_none_or(|active| active.thread_id != id) { - bail!("Cloud stream is not connected. Reopen the conversation before sending."); + return Err(PreflightRejection( + "Cloud stream is not connected. Reopen the conversation before sending.", + )); } Ok(()) } @@ -136,7 +143,7 @@ impl Replies { /// 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 (app_client, commands, output) = CloudClient::channel(/*capacity*/ 32); let mut requests = requests::Requests { client, account, @@ -144,49 +151,80 @@ pub fn connect(client: cloud::Client, account: protocol::GetAccountResponse) -> warned_partial_history: 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)); - } + run_worker(commands, &mut requests, &mut Replies::default(), &output).await; }); AppServerClient::Cloud(app_client) } +async fn run_worker( + mut commands: mpsc::Receiver, + requests: &mut requests::Requests, + replies: &mut Replies, + output: &mpsc::UnboundedSender, +) { + loop { + let command = tokio::select! { + _ = output.closed() => return, + () = replies.forward_next(output) => continue, + command = commands.recv() => match command { Some(command) => command, None => return }, + }; + let user_message_submission = + matches!(&command.operation, CloudOperation::SubmitUserMessage(_)); + let request = match command.operation { + CloudOperation::Shutdown => { + let _ = command.response.send(Ok(Ok(Value::Null))); + return; + } + CloudOperation::Request(request) => Ok((*request, cloud::SubmissionMode::StartIfIdle)), + CloudOperation::SubmitUserMessage(params) => Ok(( + protocol::ClientRequest::TurnStart { + // Native admission is keyed by the caller's stable message ID. + request_id: protocol::RequestId::Integer(0), + params: *params, + }, + cloud::SubmissionMode::Steer, + )), + 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 = match request { + Ok((request, mode)) => tokio::select! { + _ = output.closed() => return, + result = handle_request(request, mode, requests, replies, output) => result, + }, + Err(error) => Err(error), + }; + let result = result.map_err(|error| protocol::JSONRPCErrorError { + code: if error + .downcast_ref::() + .is_some() + { + -32601 + } else if user_message_submission + && (error.is::() + || matches!( + error.downcast_ref::(), + Some(cloud::MutationError::InvalidRequest) + )) + { + -32602 + } else { + -32000 + }, + message: error.to_string(), + data: None, + }); + let _ = command.response.send(Ok(result)); + } +} + async fn handle_request( request: protocol::ClientRequest, + mode: cloud::SubmissionMode, requests: &mut requests::Requests, replies: &mut Replies, output: &mpsc::UnboundedSender, @@ -212,7 +250,7 @@ async fn handle_request( replies.admit_turn(¶ms.thread_id, output)?; } match replies - .while_listening(requests.handle(request, output), output) + .while_listening(requests.handle(request, mode, output), output) .await? { requests::Response::Value(value) => Ok(value), @@ -278,3 +316,7 @@ fn cloud_error(id: &str, message: &str) -> protocol::ServerNotification { #[cfg(test)] #[path = "connection_tests.rs"] mod tests; + +#[cfg(test)] +#[path = "submission_tests.rs"] +mod submission_tests; 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 index 16fdd5f709..d372a40304 100644 --- a/codex-rs/app-server-client/src/cloud_backend/request_tests.rs +++ b/codex-rs/app-server-client/src/cloud_backend/request_tests.rs @@ -108,3 +108,43 @@ fn duplicate_and_queued_admissions_never_invent_a_running_turn() { ); } } + +#[test] +fn admission_receipts_preserve_server_turns_and_duplicate_identity() -> Result<()> { + let turn = |id: &str, status| cloud::TurnSummary { + id: id.into(), + status, + error: None, + started_at: None, + completed_at: None, + duration_ms: None, + }; + for (outcome, expected) in [ + ( + cloud::AdmissionOutcome::Started(turn("u", cloud::TurnStatus::InProgress)), + ("started", "u", protocol::TurnStatus::InProgress), + ), + ( + cloud::AdmissionOutcome::Steered(turn("v", cloud::TurnStatus::Completed)), + ("steered", "v", protocol::TurnStatus::Completed), + ), + ] { + let (kind, turn) = match admitted_message(outcome)? { + CloudUserMessageReceipt::Started(turn) => ("started", turn), + CloudUserMessageReceipt::Steered(turn) => ("steered", turn), + CloudUserMessageReceipt::Duplicate { .. } => panic!("expected a turn"), + }; + assert_eq!((kind, turn.id.as_str(), turn.status), expected); + } + for turn_id in [None, Some("original".into())] { + let CloudUserMessageReceipt::Duplicate { turn_id: actual } = + admitted_message(cloud::AdmissionOutcome::Duplicate { + turn_id: turn_id.clone(), + })? + else { + panic!("a duplicate must not become a running turn"); + }; + assert_eq!(actual, turn_id); + } + Ok(()) +} 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 d5d82e87d9..6d5814be45 100644 --- a/codex-rs/app-server-client/src/cloud_backend/requests.rs +++ b/codex-rs/app-server-client/src/cloud_backend/requests.rs @@ -1,7 +1,9 @@ //! Maps conversation requests onto native Cloud operations. +use super::PreflightRejection; use super::events; use super::send_notification; +use crate::CloudUserMessageReceipt; use codex_app_server_protocol as protocol; use codex_cloud_client as cloud; use color_eyre::eyre::Result; @@ -51,6 +53,7 @@ impl Requests { pub(super) async fn handle( &mut self, request: protocol::ClientRequest, + mode: cloud::SubmissionMode, output: &mpsc::UnboundedSender, ) -> Result { use protocol::ClientRequest as Request; @@ -188,12 +191,15 @@ impl Requests { })? } 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 })? + let mut message = cloud_message(params)?; + message.mode = mode; + let admission = self.client.add_user_message(&message).await?; + if matches!(mode, cloud::SubmissionMode::Steer) { + serde_json::to_value(admitted_message(admission.outcome)?)? + } else { + let turn = admitted_turn(admission.outcome)?; + serde_json::to_value(protocol::TurnStartResponse { turn })? + } } Request::TurnInterrupt { params, .. } => { self.client @@ -275,16 +281,22 @@ fn history_order(direction: protocol::SortDirection) -> cloud::HistoryOrder { } } -fn cloud_message(mut params: protocol::TurnStartParams) -> Result { +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."))?; + .ok_or(PreflightRejection( + "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."); + return Err(PreflightRejection( + "Changing cloud settings or sending structured input is not supported yet.", + )); } let text = input .into_iter() @@ -293,9 +305,11 @@ fn cloud_message(mut params: protocol::TurnStartParams) -> Result Ok(text), - _ => bail!("Cloud conversations currently accept plain text only."), + _ => Err(PreflightRejection( + "Cloud conversations currently accept plain text only.", + )), }) - .collect::>>()? + .collect::, _>>()? .join("\n"); Ok(cloud::AddUserMessageRequest { thread_id, @@ -307,16 +321,29 @@ fn cloud_message(mut params: protocol::TurnStartParams) -> Result Result { + Ok(match outcome { + cloud::AdmissionOutcome::Started(turn) => { + CloudUserMessageReceipt::Started(events::turn(turn)?) + } + cloud::AdmissionOutcome::Steered(turn) => { + CloudUserMessageReceipt::Steered(events::turn(turn)?) + } + cloud::AdmissionOutcome::Duplicate { turn_id } => { + CloudUserMessageReceipt::Duplicate { turn_id } + } + cloud::AdmissionOutcome::Queued { .. } => bail!( + "Cloud already accepted this message without returning a turn. Reopen the conversation to see its status before sending again." + ), + }) +} + 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." - ) - } + match admitted_message(outcome)? { + CloudUserMessageReceipt::Started(turn) | CloudUserMessageReceipt::Steered(turn) => Ok(turn), + CloudUserMessageReceipt::Duplicate { .. } => bail!( + "Cloud already accepted this message without returning a turn. Reopen the conversation to see its status before sending again." + ), } } diff --git a/codex-rs/app-server-client/src/cloud_backend/submission_tests.rs b/codex-rs/app-server-client/src/cloud_backend/submission_tests.rs new file mode 100644 index 0000000000..0130c44896 --- /dev/null +++ b/codex-rs/app-server-client/src/cloud_backend/submission_tests.rs @@ -0,0 +1,148 @@ +#![cfg(test)] +//! Exercises native admission and local validation through the command worker. + +use super::*; +use pretty_assertions::assert_eq; +use wiremock::Mock; +use wiremock::MockServer; +use wiremock::ResponseTemplate; +use wiremock::matchers::header; +use wiremock::matchers::method; +use wiremock::matchers::path; + +fn submission(message_id: &str) -> protocol::TurnStartParams { + protocol::TurnStartParams { + thread_id: "t".into(), + client_user_message_id: Some(message_id.into()), + input: vec![protocol::UserInput::Text { + text: "hello".into(), + text_elements: Vec::new(), + }], + ..Default::default() + } +} + +#[tokio::test] +async fn native_submission_uses_one_admission_and_keeps_preflight_guards() -> Result<()> { + let server = MockServer::start().await; + let home = tempfile::tempdir()?; + let config = crate::legacy_core::config::ConfigBuilder::default() + .codex_home(home.path().to_owned()) + .fallback_cwd(Some(home.path().to_owned())) + .without_project_context() + .build() + .await?; + let native = cloud::Client::new( + &config.http_client_factory(), + &server.uri(), + cloud::Credentials::new("synthetic-bearer", "synthetic-account")?, + )?; + let account = protocol::GetAccountResponse { + account: None, + requires_openai_auth: true, + workspace_routing: None, + }; + let (client, commands, output) = crate::CloudClient::channel(/*capacity*/ 4); + let client = crate::AppServerClient::Cloud(client); + let handle = client.request_handle(); + let crate::AppServerRequestHandle::Cloud(cloud_handle) = &handle else { + panic!("expected a Cloud handle"); + }; + let (reply_sender, stream) = futures::channel::mpsc::unbounded(); + let mut replies = Replies { + active: Some(Attachment { + thread_id: "t".into(), + stream: stream.boxed(), + }), + ..Replies::default() + }; + let worker = tokio::spawn(async move { + let mut requests = crate::cloud_backend::requests::Requests { + client: native, + account, + loaded: None, + warned_partial_history: None, + }; + crate::cloud_backend::run_worker(commands, &mut requests, &mut replies, &output).await + }); + let rpc = "/oaiproto.codex_app_server_backend.ThreadService/AddUserMessage"; + Mock::given(method("POST")) + .and(path(rpc)) + .and(header("authorization", "Bearer synthetic-bearer")) + .respond_with(ResponseTemplate::new(503)) + .expect(2) + .mount(&server) + .await; + let Err(crate::TypedRequestError::Server { source, .. }) = + cloud_handle.submit_user_message(submission("x")).await + else { + panic!("a lost receipt must remain uncertain"); + }; + assert_eq!(source.code, -32000); + assert!(source.message.contains("mutation outcome unknown")); + // Existing typed turn/start callers keep StartIfIdle and the same native error. + let Err(crate::TypedRequestError::Server { source, .. }) = handle + .request_typed::(protocol::ClientRequest::TurnStart { + request_id: protocol::RequestId::Integer(1), + params: submission("m"), + }) + .await + else { + panic!("ordinary turn/start must retain its server error"); + }; + assert_eq!(source.code, -32000); + assert!(source.message.contains("mutation outcome unknown")); + + for params in [ + protocol::TurnStartParams { + model: Some("other".into()), + ..submission("invalid") + }, + protocol::TurnStartParams { + client_user_message_id: None, + ..submission("missing") + }, + protocol::TurnStartParams { + input: Vec::new(), + ..submission("empty") + }, + protocol::TurnStartParams { + thread_id: "other".into(), + ..submission("foreign") + }, + ] { + let Err(crate::TypedRequestError::Server { source, .. }) = + cloud_handle.submit_user_message(params).await + else { + panic!("invalid input must fail before admission"); + }; + assert_eq!(source.code, -32602); + } + drop(reply_sender); + let Err(crate::TypedRequestError::Server { source, .. }) = + cloud_handle.submit_user_message(submission("closed")).await + else { + panic!("a disconnected stream must reject input"); + }; + assert_eq!(source.code, -32602); + let ordinary = handle + .request(protocol::ClientRequest::TurnStart { + request_id: protocol::RequestId::Integer(2), + params: submission("closed"), + }) + .await? + .expect_err("ordinary turn/start keeps its existing disconnected error"); + assert_eq!(ordinary.code, -32000); + + client.shutdown().await?; + worker.await?; + let calls = server.received_requests().await.unwrap(); + assert_eq!( + calls + .iter() + .map(|request| (request.body[7], *request.body.last().unwrap())) + .collect::>(), + vec![(b'x', 1), (b'm', 3)] + ); + Ok(()) +} diff --git a/codex-rs/app-server-client/src/lib.rs b/codex-rs/app-server-client/src/lib.rs index 008e165e11..dae43d2a02 100644 --- a/codex-rs/app-server-client/src/lib.rs +++ b/codex-rs/app-server-client/src/lib.rs @@ -29,6 +29,7 @@ pub use cloud::CloudClient; pub use cloud::CloudCommand; pub use cloud::CloudOperation; pub use cloud::CloudRequestHandle; +pub use cloud::CloudUserMessageReceipt; mod server_request; pub use server_request::AppServerReplyHandle;