From 781d89b91e2487c3653208b9c13287e1a09f6f04 Mon Sep 17 00:00:00 2001 From: Benjamin Carlsson Date: Tue, 3 Nov 2026 17:54:00 +0000 Subject: [PATCH] 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 | 26 ++ codex-rs/cloud-client/examples/attach.rs | 55 +++ codex-rs/cloud-client/src/lib.rs | 7 + codex-rs/cloud-client/src/transcript_tests.rs | 3 + codex-rs/cloud-client/src/turns.rs | 221 ++++++++++++ codex-rs/cloud-client/src/turns_tests.rs | 336 ++++++++++++++++++ codex-rs/cloud-client/src/turns_wire.rs | 42 ++- 7 files changed, 689 insertions(+), 1 deletion(-) create mode 100644 codex-rs/cloud-client/src/turns.rs create mode 100644 codex-rs/cloud-client/src/turns_tests.rs diff --git a/codex-rs/cloud-client/README.md b/codex-rs/cloud-client/README.md index a7c2389f4a..3953afac5e 100644 --- a/codex-rs/cloud-client/README.md +++ b/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' +``` diff --git a/codex-rs/cloud-client/examples/attach.rs b/codex-rs/cloud-client/examples/attach.rs index 6f777b3f23..9e8c550cd5 100644 --- a/codex-rs/cloud-client/examples/attach.rs +++ b/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?); } diff --git a/codex-rs/cloud-client/src/lib.rs b/codex-rs/cloud-client/src/lib.rs index 51609f8736..bdff9a5004 100644 --- a/codex-rs/cloud-client/src/lib.rs +++ b/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; diff --git a/codex-rs/cloud-client/src/transcript_tests.rs b/codex-rs/cloud-client/src/transcript_tests.rs index a2c324057a..977ea64317 100644 --- a/codex-rs/cloud-client/src/transcript_tests.rs +++ b/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; diff --git a/codex-rs/cloud-client/src/turns.rs b/codex-rs/cloud-client/src/turns.rs new file mode 100644 index 0000000000..fe14a943be --- /dev/null +++ b/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, + }, + 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 { + 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( + &self, + value: T, + path: &'static str, + ) -> Result { + 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::(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 { + 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), + }) +} diff --git a/codex-rs/cloud-client/src/turns_tests.rs b/codex-rs/cloud-client/src/turns_tests.rs new file mode 100644 index 0000000000..09b8b69481 --- /dev/null +++ b/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 { + [ + 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)) + )); + } + } +} diff --git a/codex-rs/cloud-client/src/turns_wire.rs b/codex-rs/cloud-client/src/turns_wire.rs index 91cf93e1d5..7333e24412 100644 --- a/codex-rs/cloud-client/src/turns_wire.rs +++ b/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, + #[prost(string, tag = "23")] + pub thread_id: String, + #[prost(int32, optional, tag = "26")] + pub access: Option, + #[prost(int32, optional, tag = "30")] + pub mode: Option, +} #[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, } +#[derive(Clone, Message)] +pub(super) struct AddResponse { + #[prost(message, optional, tag = "1")] + pub turn: Option, + #[prost(int32, optional, tag = "2")] + pub outcome: Option, + #[prost(string, optional, tag = "3")] + pub client_user_message_id: Option, + #[prost(message, optional, tag = "5")] + pub queued_submission: Option, +} +#[derive(Clone, Message)] +pub(super) struct QueuedSubmission { + #[prost(string, optional, tag = "1")] + pub id: Option, + #[prost(string, optional, tag = "3")] + pub client_user_message_id: Option, +} +#[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, +}