Add typed Cloud user-message submission receipts (#59746)

## What changed

Add `CloudRequestHandle::submit_user_message` to submit a message with a stable ID using `SubmissionMode::Steer`. Return a `CloudUserMessageReceipt` that distinguishes started, steered, and duplicate admissions while preserving server-provided turn status and duplicate identity.

Report preflight rejections and invalid requests as `-32602`, while preserving uncertain admission outcomes as errors without automatic retries. Keep existing `turn/start` callers on `StartIfIdle` with their existing response and error behavior.

## Testing

Add receipt mapping tests and a mock-server worker test covering submission modes, single admission attempts, uncertain outcomes, input validation, and disconnected reply streams.

GitOrigin-RevId: eba120cf47cf0297b072ea373341ead663fb9bc9

codex-rs/Cargo.lock

@@ -2232,6 +2232,7 @@ dependencies = [
  "toml 0.9.11+spec-1.1.0",
  "tracing",
  "url",
+ "wiremock",
 ]
 
 [[package]]

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 }

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<ClientRequest>),
+    SubmitUserMessage(Box<TurnStartParams>),
     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<String> },
+}
+
 /// 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<CloudUserMessageReceipt, TypedRequestError> {
+        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<RequestResult> {
         let (response, result) = oneshot::channel();
         self.commands

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

@@ -21,6 +21,11 @@ use tokio::sync::mpsc;
 
 type ReplyStream = BoxStream<'static, Result<cloud::Event, cloud::Error>>;
 
+/// 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<AppServerEvent>,
-    ) -> 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::<requests::UnsupportedOperation>()
-                    .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<crate::CloudCommand>,
+    requests: &mut requests::Requests,
+    replies: &mut Replies,
+    output: &mpsc::UnboundedSender<AppServerEvent>,
+) {
+    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::<requests::UnsupportedOperation>()
+                .is_some()
+            {
+                -32601
+            } else if user_message_submission
+                && (error.is::<PreflightRejection>()
+                    || matches!(
+                        error.downcast_ref::<cloud::MutationError>(),
+                        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<AppServerEvent>,
@@ -212,7 +250,7 @@ async fn handle_request(
         replies.admit_turn(&params.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;

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(())
+}

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<crate::AppServerEvent>,
     ) -> Result<Response> {
         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<cloud::AddUserMessageRequest> {
+fn cloud_message(
+    mut params: protocol::TurnStartParams,
+) -> Result<cloud::AddUserMessageRequest, PreflightRejection> {
     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<cloud::AddUser
                 text,
                 text_elements,
             } if text_elements.is_empty() => Ok(text),
-            _ => bail!("Cloud conversations currently accept plain text only."),
+            _ => Err(PreflightRejection(
+                "Cloud conversations currently accept plain text only.",
+            )),
         })
-        .collect::<Result<Vec<_>>>()?
+        .collect::<Result<Vec<_>, _>>()?
         .join("\n");
     Ok(cloud::AddUserMessageRequest {
         thread_id,
@@ -307,16 +321,29 @@ fn cloud_message(mut params: protocol::TurnStartParams) -> Result<cloud::AddUser
     })
 }
 
+fn admitted_message(outcome: cloud::AdmissionOutcome) -> Result<CloudUserMessageReceipt> {
+    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<protocol::Turn> {
-    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."
+        ),
     }
 }
 

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::TurnStartResponse>(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<_>>(),
+        vec![(b'x', 1), (b'm', 3)]
+    );
+    Ok(())
+}

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;