Add cloud client message submission and turn interruption (#59201)

## What changed

Expose `add_user_message` for plain-text input using the existing thread configuration, with explicit `StartIfIdle`, `Steer`, and `Queue` modes and typed admission outcomes. Expose `interrupt_turn` to interrupt a specific turn without removing queued input.

Distinguish local validation failures from unknown remote outcomes, require successful final gRPC status, and avoid automatic retries. Preserve client message identity and raw admission payloads; report duplicate admissions by original turn ID without implying current lifecycle status.

Document reconciliation requirements and extend the attach example to submit an optional message while draining events.

## Testing

Add wire-fixture tests for admission variants, request encoding, malformed responses, local validation, redacted diagnostics, uncertain RPC outcomes, final gRPC status requirements, and dropping pending submissions without retrying or interrupting.

GitOrigin-RevId: 1def92de3fc5ca8c1f89431b7ea29b9b80d64600

codex-rs/cloud-client/README.md

@@ -59,3 +59,29 @@ they must not create duplicate transcript rows. Disconnects, untyped recovery
 notifications and uncertain overlap require authoritative history reload. This
 crate exposes the read primitives; a coordinated resync/display state machine is
 still required before claiming lossless resume or live/history reconciliation.
+
+`add_user_message` submits plain text using the thread's existing configuration.
+Choose `StartIfIdle`, `Steer`, or `Queue` explicitly; admission distinguishes
+started, steered, duplicate and queued results. Persist a unique client message ID
+with each logical input and keep it unchanged when reconciling an uncertain
+submission. Success means admission, not completion. `interrupt_turn` addresses
+one thread/turn and does not stop the thread or delete queued input. Duplicate
+admission exposes only the original turn ID when supplied. Its wire turn status
+is synthesized by the backend and does not describe current lifecycle state;
+reload history or observe Attach instead of treating the receipt as an active turn.
+
+`MutationError::InvalidRequest` means local validation rejected the request before
+sending. `OutcomeUnknown` retains the source error: even a server error can follow
+admission, so status codes alone do not prove rejection. Malformed success replies
+and mismatched response IDs are also uncertain. No automatic retry is performed;
+dropping a pending call does not undo remote work. Reconcile with server state
+before deciding whether another attempt is safe. Full input/configuration overrides
+and automated outcome reconciliation are not implemented by this text-only slice.
+
+The attach example accepts two optional additional arguments to explicitly send
+one message after attaching: a persisted client message ID and the text. It uses
+`StartIfIdle`, drains events during admission, and prints only redacted event kinds:
+
+```sh
+cargo run -p codex-cloud-client --example attach -- "$CODEX_CLOUD_GRPC_ENDPOINT" "$CODEX_CLOUD_THREAD_ID" "$CODEX_CLOUD_MESSAGE_ID" 'Explain this project'
+```

codex-rs/cloud-client/examples/attach.rs

@@ -1,7 +1,10 @@
 //! Standalone consumer; prints event kinds without logging payloads.
+use codex_cloud_client::AddUserMessageRequest;
+use codex_cloud_client::AdmissionOutcome;
 use codex_cloud_client::Client;
 use codex_cloud_client::Credentials;
 use codex_cloud_client::ResumeRequest;
+use codex_cloud_client::SubmissionMode;
 use codex_http_client::HttpClientFactory;
 use codex_http_client::OutboundProxyPolicy;
 use futures::StreamExt;
@@ -15,6 +18,21 @@ async fn main() -> anyhow::Result<()> {
     let thread_id = args
         .next()
         .ok_or_else(|| anyhow::anyhow!("expected existing thread ID"))?;
+    // Supplying both extra arguments explicitly opts into sending one message.
+    let submission = match (args.next(), args.next()) {
+        (None, None) => None,
+        (Some(client_user_message_id), Some(text)) => Some(AddUserMessageRequest {
+            thread_id: thread_id.clone(),
+            client_user_message_id,
+            text,
+            mode: SubmissionMode::StartIfIdle,
+            shared_access: false,
+        }),
+        _ => anyhow::bail!("expected both client message ID and text, or neither"),
+    };
+    if args.next().is_some() {
+        anyhow::bail!("unexpected argument");
+    }
     let credentials = Credentials::new(
         &std::env::var("CODEX_CLOUD_ACCESS_TOKEN")?,
         &std::env::var("CODEX_CLOUD_ACCOUNT_ID")?,
@@ -32,6 +50,43 @@ async fn main() -> anyhow::Result<()> {
         })
         .await?;
     let mut events = client.attach(&thread_id).await?;
+    if let Some(request) = submission {
+        // Drain Attach while admission is pending; events can precede its response.
+        let admission = client.add_user_message(&request);
+        tokio::pin!(admission);
+        let (result, stream_error) = loop {
+            tokio::select! {
+                result = &mut admission => {
+                    break (result, None);
+                }
+                event = events.next() => match event {
+                    Some(Ok(event)) => println!("{event:?}"),
+                    Some(Err(error)) => break (admission.await, Some(anyhow::Error::new(error))),
+                    None => break (admission.await, Some(anyhow::anyhow!("attachment ended"))),
+                }
+            }
+        };
+        match result {
+            Ok(admission) => println!(
+                "admission: {}",
+                match admission.outcome {
+                    AdmissionOutcome::Started(_) => "started",
+                    AdmissionOutcome::Steered(_) => "steered",
+                    AdmissionOutcome::Duplicate { .. } => "duplicate",
+                    AdmissionOutcome::Queued { .. } => "queued",
+                }
+            ),
+            Err(error) => match stream_error {
+                Some(stream_error) => {
+                    anyhow::bail!("attachment failed: {stream_error}; admission failed: {error}");
+                }
+                None => return Err(error.into()),
+            },
+        }
+        if let Some(error) = stream_error {
+            return Err(error.context("attachment failed"));
+        }
+    }
     while let Some(event) = events.next().await {
         println!("{:?}", event?);
     }

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

@@ -10,6 +10,7 @@ mod item_wire;
 mod items;
 mod transcript;
 mod transcript_wire;
+mod turns;
 mod turns_wire;
 mod types;
 mod wire;
@@ -38,6 +39,12 @@ pub use transcript::TranscriptUpdate;
 pub use transcript::TurnFailure;
 pub use transcript::TurnStatus;
 pub use transcript::TurnSummary;
+pub use turns::AddUserMessageRequest;
+pub use turns::Admission;
+pub use turns::AdmissionOutcome;
+pub use turns::InterruptTurnRequest;
+pub use turns::MutationError;
+pub use turns::SubmissionMode;
 pub use types::Credentials;
 pub use types::Error;
 pub use types::Event;

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

@@ -298,3 +298,6 @@ mod items;
 
 #[path = "history_tests.rs"]
 mod history;
+
+#[path = "turns_tests.rs"]
+mod turns_tests;

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

@@ -0,0 +1,221 @@
+//! Explicit, non-retried mutations. A failed RPC may already have changed the thread.
+use crate::Client;
+use crate::Error;
+use crate::TurnSummary;
+use crate::transcript::DecodeFailure;
+use crate::transcript::required;
+use crate::turns_wire as wire;
+use bytes::Bytes;
+use prost::Message;
+use std::fmt;
+
+#[derive(Clone, Copy, Debug)]
+pub enum SubmissionMode {
+    Steer,
+    Queue,
+    StartIfIdle,
+}
+/// Plain text using the existing thread configuration. Persist the message ID with
+/// the input before calling; never reuse it for a different logical submission.
+#[derive(Clone)]
+pub struct AddUserMessageRequest {
+    pub thread_id: String,
+    pub client_user_message_id: String,
+    pub text: String,
+    pub mode: SubmissionMode,
+    pub shared_access: bool,
+}
+#[derive(Clone)]
+pub struct InterruptTurnRequest {
+    pub thread_id: String,
+    pub turn_id: String,
+    pub shared_access: bool,
+}
+#[derive(PartialEq)]
+pub enum AdmissionOutcome {
+    Started(TurnSummary),
+    Steered(TurnSummary),
+    /// The original turn ID, if assigned. Read history or observe Attach for its status.
+    Duplicate {
+        turn_id: Option<String>,
+    },
+    Queued {
+        submission_id: String,
+    },
+}
+pub struct Admission {
+    pub thread_id: String,
+    pub client_user_message_id: String,
+    pub outcome: AdmissionOutcome,
+    payload: Bytes,
+}
+impl Admission {
+    /// Complete original AddUserMessageResponse, including untyped metadata.
+    pub fn payload(&self) -> &Bytes {
+        &self.payload
+    }
+}
+/// Remote errors do not reliably identify whether admission happened. Inspect the
+/// source explicitly, reconcile with the server, and never blindly resubmit.
+#[derive(Debug, thiserror::Error)]
+pub enum MutationError {
+    #[error("invalid or oversized mutation request; nothing was sent")]
+    InvalidRequest,
+    #[error("mutation outcome unknown; do not automatically retry: {0}")]
+    OutcomeUnknown(#[source] Error),
+}
+macro_rules! redacted_debug {
+    ($($ty:ty),*) => {$(impl fmt::Debug for $ty {
+        fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { f.write_str(concat!(stringify!($ty), "([redacted])")) }
+    })*};
+}
+redacted_debug!(
+    AddUserMessageRequest,
+    InterruptTurnRequest,
+    Admission,
+    AdmissionOutcome
+);
+
+impl Client {
+    /// Sends once. Success is admission, not turn completion. Dropping this future
+    /// after polling can leave admission unknown; it does not cancel remote work.
+    pub async fn add_user_message(
+        &self,
+        request: &AddUserMessageRequest,
+    ) -> Result<Admission, MutationError> {
+        if request.thread_id.trim().is_empty()
+            || request.client_user_message_id.trim().is_empty()
+            || request.text.trim().is_empty()
+        {
+            return Err(MutationError::InvalidRequest);
+        }
+        let value = wire::AddRequest {
+            client_user_message_id: request.client_user_message_id.clone(),
+            input: Some(wire::Input {
+                values: vec![wire::UserInput {
+                    value: Some(wire::Value::Text(wire::Text {
+                        text: Some(request.text.clone()),
+                    })),
+                }],
+            }),
+            thread_id: request.thread_id.clone(),
+            access: request.shared_access.then_some(1),
+            mode: Some(match request.mode {
+                SubmissionMode::Steer => 1,
+                SubmissionMode::Queue => 2,
+                SubmissionMode::StartIfIdle => 3,
+            }),
+        };
+        let payload = self
+            .mutate(
+                value,
+                "/oaiproto.codex_app_server_backend.ThreadService/AddUserMessage",
+            )
+            .await?;
+        let outcome = decode_admission(&request.client_user_message_id, payload.clone())
+            .map_err(|_| MutationError::OutcomeUnknown(Error::InvalidResponse))?;
+        Ok(Admission {
+            thread_id: request.thread_id.clone(),
+            client_user_message_id: request.client_user_message_id.clone(),
+            outcome,
+            payload,
+        })
+    }
+    /// Interrupts this turn only, without stopping its thread or removing queued input.
+    /// Success acknowledges the RPC; observe turn lifecycle events for completion.
+    pub async fn interrupt_turn(
+        &self,
+        request: &InterruptTurnRequest,
+    ) -> Result<(), MutationError> {
+        if request.thread_id.trim().is_empty() || request.turn_id.trim().is_empty() {
+            return Err(MutationError::InvalidRequest);
+        }
+        let payload = self
+            .mutate(
+                wire::InterruptRequest {
+                    thread_id: request.thread_id.clone(),
+                    turn_id: request.turn_id.clone(),
+                    access: request.shared_access.then_some(1),
+                },
+                "/oaiproto.codex_app_server_backend.ThreadService/InterruptTurn",
+            )
+            .await?;
+        crate::wire::ResumeResponse::decode(payload)
+            .map_err(|_| MutationError::OutcomeUnknown(Error::InvalidResponse))?;
+        Ok(())
+    }
+    async fn mutate<T: Message + Send + 'static>(
+        &self,
+        value: T,
+        path: &'static str,
+    ) -> Result<Bytes, MutationError> {
+        if value.encoded_len() > crate::MAX_MESSAGE_BYTES {
+            return Err(MutationError::InvalidRequest);
+        }
+        let mut grpc = self.grpc.clone();
+        let mut request = tonic::Request::new(value);
+        *request.metadata_mut() =
+            tonic::metadata::MetadataMap::from_headers(self.credentials.headers.clone());
+        // Read the final trailers separately: initial metadata cannot acknowledge
+        // the mutation, and tonic permits HTTP 200 without a final gRPC status.
+        tokio::time::timeout(crate::REQUEST_TIMEOUT, async {
+            let response = grpc
+                .server_streaming(
+                    request,
+                    http::uri::PathAndQuery::from_static(path),
+                    crate::history_wire::RawResponseCodec::<T>(std::marker::PhantomData),
+                )
+                .await?;
+            // An initial OK makes tonic skip final-status validation. A unary
+            // success must carry its status after the response message instead.
+            if response.metadata().contains_key("grpc-status") {
+                return Err(Error::InvalidResponse);
+            }
+            let mut response = response.into_inner();
+            let payload = response.message().await?.ok_or(Error::InvalidResponse)?;
+            let trailers = response.trailers().await?.ok_or(Error::InvalidResponse)?;
+            if trailers
+                .get("grpc-status")
+                .is_none_or(|status| status != "0")
+            {
+                return Err(Error::InvalidResponse);
+            }
+            Ok(payload)
+        })
+        .await
+        .map_err(|_| MutationError::OutcomeUnknown(Error::Timeout))?
+        .map_err(MutationError::OutcomeUnknown)
+    }
+}
+fn decode_admission(message_id: &str, payload: Bytes) -> Result<AdmissionOutcome, DecodeFailure> {
+    let response = wire::AddResponse::decode(payload)?;
+    if required(response.client_user_message_id)? != message_id {
+        return Err(DecodeFailure::InvalidResponse);
+    }
+    if response.outcome == Some(3) {
+        let turn_id = response.turn.map(|turn| required(turn.id)).transpose()?;
+        if turn_id.as_ref().is_some_and(|id| id.trim().is_empty()) {
+            return Err(DecodeFailure::InvalidResponse);
+        }
+        return Ok(AdmissionOutcome::Duplicate { turn_id });
+    }
+    let turn = response.turn.map(TurnSummary::try_from).transpose()?;
+    if turn.as_ref().is_some_and(|turn| turn.id.trim().is_empty()) {
+        return Err(DecodeFailure::InvalidResponse);
+    }
+    Ok(match response.outcome {
+        Some(1) => AdmissionOutcome::Started(required(turn)?),
+        Some(2) => AdmissionOutcome::Steered(required(turn)?),
+        Some(4) => {
+            let queued = required(response.queued_submission)?;
+            let submission_id = required(queued.id)?;
+            if submission_id.trim().is_empty()
+                || required(queued.client_user_message_id)? != message_id
+            {
+                return Err(DecodeFailure::InvalidResponse);
+            }
+            AdmissionOutcome::Queued { submission_id }
+        }
+        _ => return Err(DecodeFailure::InvalidResponse),
+    })
+}

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

@@ -0,0 +1,336 @@
+//! Independent wire fixtures for admission, addressed interruption and uncertain outcomes.
+use super::*;
+use pretty_assertions::assert_eq;
+
+fn request() -> AddUserMessageRequest {
+    AddUserMessageRequest {
+        thread_id: "t".into(),
+        client_user_message_id: "m".into(),
+        text: "hello".into(),
+        mode: SubmissionMode::StartIfIdle,
+        shared_access: true,
+    }
+}
+fn admitted(outcome: u8) -> Vec<u8> {
+    [
+        message(/*tag*/ 1, b"\x22\x01u\x40\x04"),
+        vec![0x10, outcome, 0x1a, 1, b'm'],
+    ]
+    .concat()
+}
+#[tokio::test]
+async fn submission_preserves_admission_variants_and_wire_identity() {
+    for (mode, mode_number, outcome_number) in [
+        (SubmissionMode::StartIfIdle, 3, 1),
+        (SubmissionMode::Steer, 1, 2),
+        (SubmissionMode::Queue, 2, 3),
+        (SubmissionMode::Queue, 2, 4),
+    ] {
+        let mut payload = admitted(outcome_number);
+        if outcome_number == 4 {
+            payload = [
+                vec![0x10, 4, 0x1a, 1, b'm'],
+                message(/*tag*/ 5, b"\x0a\x01q\x1a\x01m"),
+            ]
+            .concat();
+        }
+        payload.extend([0xa0, 0x06, 7]); // Unknown metadata survives.
+        let mut fixture = fixture(response(vec![frame(&payload)], tonic::Status::ok(""))).await;
+        let result = fixture
+            .client
+            .add_user_message(&AddUserMessageRequest { mode, ..request() })
+            .await
+            .unwrap();
+        let turn = TurnSummary {
+            id: "u".into(),
+            status: TurnStatus::InProgress,
+            error: None,
+            started_at: None,
+            completed_at: None,
+            duration_ms: None,
+        };
+        let expected = match outcome_number {
+            1 => AdmissionOutcome::Started(turn),
+            2 => AdmissionOutcome::Steered(turn),
+            3 => AdmissionOutcome::Duplicate {
+                turn_id: Some("u".into()),
+            },
+            4 => AdmissionOutcome::Queued {
+                submission_id: "q".into(),
+            },
+            _ => unreachable!(),
+        };
+        assert_eq!(
+            (
+                &result.thread_id,
+                &result.client_user_message_id,
+                &result.outcome,
+                result.payload()
+            ),
+            (
+                &"t".to_owned(),
+                &"m".to_owned(),
+                &expected,
+                &Bytes::from(payload)
+            )
+        );
+        assert_eq!(
+            format!("{result:?} {:?}", result.outcome),
+            "Admission([redacted]) AdmissionOutcome([redacted])"
+        );
+        let expected_request = [
+            message(/*tag*/ 4, b"m"),
+            message(
+                /*tag*/ 11,
+                &message(
+                    /*tag*/ 1,
+                    &message(/*tag*/ 1, &message(/*tag*/ 1, b"hello")),
+                ),
+            ),
+            message(/*tag*/ 23, b"t"),
+            vec![0xd0, 1, 1, 0xf0, 1, mode_number],
+        ]
+        .concat();
+        assert_request(&mut fixture, "AddUserMessage", &expected_request).await;
+    }
+}
+#[tokio::test]
+async fn duplicate_can_omit_turn_but_malformed_admission_is_unknown() {
+    let payloads = vec![
+        vec![],
+        admitted(/*outcome*/ 0),
+        admitted(/*outcome*/ 99),
+        b"\x10\x01\x1a\x01m".to_vec(),
+        // Started with a turn ID but no turn status.
+        b"\x0a\x03\x22\x01u\x10\x01\x1a\x01m".to_vec(),
+        b"\x10\x04\x1a\x01m".to_vec(),
+        b"\x10\x03\x1a\x01x".to_vec(),
+        // A present duplicate turn must have a nonblank ID.
+        b"\x0a\x00\x10\x03\x1a\x01m".to_vec(),
+        b"\x0a\x02\x22\x00\x10\x03\x1a\x01m".to_vec(),
+        b"\x0a\x03\x22\x01 \x10\x03\x1a\x01m".to_vec(),
+        vec![0x80],
+        [
+            vec![0x10, 4, 0x1a, 1, b'm'],
+            message(/*tag*/ 5, b"\x0a\x01q\x1a\x01x"),
+        ]
+        .concat(),
+    ];
+    for payload in payloads {
+        let mut fixture = fixture(response(vec![frame(&payload)], tonic::Status::ok(""))).await;
+        assert!(matches!(
+            fixture.client.add_user_message(&request()).await,
+            Err(MutationError::OutcomeUnknown(Error::InvalidResponse))
+        ));
+        fixture.requests.recv().await.unwrap();
+        assert!(fixture.requests.try_recv().is_err());
+    }
+    // Duplicate identity is independent of absent or unknown lifecycle state.
+    for turn in [&b"\x22\x01u"[..], &b"\x22\x01u\x40\x63"[..]] {
+        let payload = [message(/*tag*/ 1, turn), b"\x10\x03\x1a\x01m".to_vec()].concat();
+        let fixture = fixture(response(vec![frame(&payload)], tonic::Status::ok(""))).await;
+        assert_eq!(
+            fixture
+                .client
+                .add_user_message(&request())
+                .await
+                .unwrap()
+                .outcome,
+            AdmissionOutcome::Duplicate {
+                turn_id: Some("u".into())
+            }
+        );
+    }
+    let fixture = fixture(response(
+        vec![frame(b"\x10\x03\x1a\x01m")],
+        tonic::Status::ok(""),
+    ))
+    .await;
+    assert_eq!(
+        fixture
+            .client
+            .add_user_message(&request())
+            .await
+            .unwrap()
+            .outcome,
+        AdmissionOutcome::Duplicate { turn_id: None }
+    );
+}
+#[tokio::test]
+async fn interrupt_addresses_original_turn_and_preserves_uncertain_errors() {
+    for status in [
+        tonic::Status::ok(""),
+        tonic::Status::permission_denied("private"),
+        tonic::Status::unavailable("private"),
+    ] {
+        let code = status.code() as i32;
+        let mut fixture = fixture(response(vec![frame(&[])], status)).await;
+        let result = fixture
+            .client
+            .interrupt_turn(&InterruptTurnRequest {
+                thread_id: "t".into(),
+                turn_id: "u".into(),
+                shared_access: true,
+            })
+            .await;
+        if code == 0 {
+            result.unwrap();
+        } else {
+            let error = result.unwrap_err();
+            assert!(!format!("{error:?} {error}").contains("private"));
+            let MutationError::OutcomeUnknown(Error::Rpc(status)) = error else {
+                panic!("expected unknown outcome")
+            };
+            assert_eq!((status.code, status.message), (code, "private".into()));
+        }
+        assert_request(&mut fixture, "InterruptTurn", b"\x0a\x01t\x12\x01u\x18\x01").await;
+        assert!(fixture.requests.try_recv().is_err());
+    }
+}
+#[tokio::test]
+async fn invalid_input_never_sends_and_diagnostics_redact() {
+    let mut fixture = fixture(response(vec![frame(&[])], tonic::Status::ok(""))).await;
+    for input in [
+        AddUserMessageRequest {
+            text: " ".into(),
+            ..request()
+        },
+        AddUserMessageRequest {
+            client_user_message_id: "".into(),
+            ..request()
+        },
+        AddUserMessageRequest {
+            thread_id: "".into(),
+            ..request()
+        },
+    ] {
+        assert_eq!(format!("{input:?}"), "AddUserMessageRequest([redacted])");
+        assert!(matches!(
+            fixture.client.add_user_message(&input).await,
+            Err(MutationError::InvalidRequest)
+        ));
+    }
+    let input = InterruptTurnRequest {
+        thread_id: "private".into(),
+        turn_id: "".into(),
+        shared_access: false,
+    };
+    assert_eq!(format!("{input:?}"), "InterruptTurnRequest([redacted])");
+    assert!(matches!(
+        fixture.client.interrupt_turn(&input).await,
+        Err(MutationError::InvalidRequest)
+    ));
+    assert!(fixture.requests.try_recv().is_err());
+}
+#[tokio::test]
+async fn dropping_pending_submission_does_not_send_a_retry_or_interrupt() {
+    let (sender, dropped) = tokio::sync::oneshot::channel::<()>();
+    let stream = async_stream::stream! {
+        let _sender = sender;
+        yield Ok::<_, Infallible>(Frame::data(frame(&admitted(/*outcome*/ 1))));
+        std::future::pending::<()>().await; // No final status yet: admission remains uncertain.
+    };
+    let mut fixture = fixture(
+        http::Response::builder()
+            .header("content-type", "application/grpc")
+            .body(Body::new(StreamBody::new(stream)))
+            .unwrap(),
+    )
+    .await;
+    {
+        let request = request();
+        let call = fixture.client.add_user_message(&request);
+        tokio::pin!(call);
+        tokio::select! {
+            result = &mut call => panic!("completed before trailers: {result:?}"),
+            observed = fixture.requests.recv() => { observed.unwrap(); }
+        }
+    }
+    assert!(
+        tokio::time::timeout(Duration::from_secs(5), dropped)
+            .await
+            .unwrap()
+            .is_err()
+    );
+    assert!(fixture.requests.try_recv().is_err());
+}
+#[tokio::test]
+async fn server_rejection_and_transport_status_do_not_prove_no_admission() {
+    for status in [
+        tonic::Status::invalid_argument("private"),
+        tonic::Status::deadline_exceeded("private"),
+        tonic::Status::unavailable("private"),
+    ] {
+        let code = status.code() as i32;
+        let mut fixture = fixture(response(vec![], status)).await;
+        let error = fixture
+            .client
+            .add_user_message(&request())
+            .await
+            .unwrap_err();
+        assert!(!format!("{error:?} {error}").contains("private"));
+        let MutationError::OutcomeUnknown(Error::Rpc(status)) = error else {
+            panic!("expected unknown outcome")
+        };
+        assert_eq!((status.code, status.message), (code, "private".into()));
+        fixture.requests.recv().await.unwrap();
+        assert!(fixture.requests.try_recv().is_err());
+    }
+}
+
+#[tokio::test]
+async fn mutations_require_explicit_successful_grpc_status() {
+    let mut ok_trailers = http::HeaderMap::new();
+    ok_trailers.insert("grpc-status", http::HeaderValue::from_static("0"));
+    for (initial_status, trailers, include_message) in [
+        (None, None, true),
+        (None, Some(http::HeaderMap::new()), true),
+        (Some("0"), None, true),
+        (Some("0"), None, false),
+        (Some("0"), Some(ok_trailers), true),
+    ] {
+        for method in ["AddUserMessage", "InterruptTurn"] {
+            let payload = if method == "AddUserMessage" {
+                admitted(/*outcome*/ 1)
+            } else {
+                vec![]
+            };
+            let frames = include_message
+                .then(|| Frame::data(frame(&payload)))
+                .into_iter()
+                .chain(trailers.clone().map(Frame::trailers));
+            let mut builder = http::Response::builder().header("content-type", "application/grpc");
+            if let Some(status) = initial_status {
+                builder = builder.header("grpc-status", status);
+            }
+            let fixture = fixture(
+                builder
+                    .body(Body::new(StreamBody::new(futures::stream::iter(
+                        frames.map(Ok::<_, Infallible>),
+                    ))))
+                    .unwrap(),
+            )
+            .await;
+            let result = if method == "AddUserMessage" {
+                fixture
+                    .client
+                    .add_user_message(&request())
+                    .await
+                    .map(|_| ())
+            } else {
+                fixture
+                    .client
+                    .interrupt_turn(&InterruptTurnRequest {
+                        thread_id: "t".into(),
+                        turn_id: "u".into(),
+                        shared_access: false,
+                    })
+                    .await
+            };
+            assert!(matches!(
+                result,
+                Err(MutationError::OutcomeUnknown(Error::InvalidResponse))
+            ));
+        }
+    }
+}

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

@@ -1,6 +1,19 @@
-//! Native user-input envelopes shared by item decoding and later turn submission.
+//! Native plain-text submission and interruption envelopes; omitted settings inherit thread config.
 use prost::Message;
 
+#[derive(Clone, Message)]
+pub(super) struct AddRequest {
+    #[prost(string, tag = "4")]
+    pub client_user_message_id: String,
+    #[prost(message, optional, tag = "11")]
+    pub input: Option<Input>,
+    #[prost(string, tag = "23")]
+    pub thread_id: String,
+    #[prost(int32, optional, tag = "26")]
+    pub access: Option<i32>,
+    #[prost(int32, optional, tag = "30")]
+    pub mode: Option<i32>,
+}
 #[derive(Clone, Message)]
 pub(super) struct Input {
     #[prost(message, repeated, tag = "1")]
@@ -33,3 +46,30 @@ pub(super) struct Text {
     #[prost(string, optional, tag = "1")]
     pub text: Option<String>,
 }
+#[derive(Clone, Message)]
+pub(super) struct AddResponse {
+    #[prost(message, optional, tag = "1")]
+    pub turn: Option<crate::transcript_wire::Turn>,
+    #[prost(int32, optional, tag = "2")]
+    pub outcome: Option<i32>,
+    #[prost(string, optional, tag = "3")]
+    pub client_user_message_id: Option<String>,
+    #[prost(message, optional, tag = "5")]
+    pub queued_submission: Option<QueuedSubmission>,
+}
+#[derive(Clone, Message)]
+pub(super) struct QueuedSubmission {
+    #[prost(string, optional, tag = "1")]
+    pub id: Option<String>,
+    #[prost(string, optional, tag = "3")]
+    pub client_user_message_id: Option<String>,
+}
+#[derive(Clone, Message)]
+pub(super) struct InterruptRequest {
+    #[prost(string, tag = "1")]
+    pub thread_id: String,
+    #[prost(string, tag = "2")]
+    pub turn_id: String,
+    #[prost(int32, optional, tag = "3")]
+    pub access: Option<i32>,
+}