From b707714ae4200db0a0385da24b3981d99139fa62 Mon Sep 17 00:00:00 2001 From: Benjamin Carlsson Date: Thu, 1 Oct 2026 22:04:41 +0000 Subject: [PATCH] Add a native gRPC client for cloud thread resume and attach (#50113) ## What changed Add `codex-cloud-client`, a reusable Rust client for `ThreadService.Resume` and live `ThreadService.Attach` over HTTP/2. Callers supply the native gRPC origin, bearer token, account ID, and `HttpClientFactory`. - Reuse shared proxy, custom CA, and network policy handling for HTTPS. Reject redirects and allow plaintext HTTP only on direct loopback with unrestricted, unmanaged policy. - Preserve notification and server-request protobuf payloads as opaque bytes and expose native gRPC status details while redacting credentials, payloads, and error details from diagnostics. - Bound resume and attach setup to 150 seconds without retrying requests or imposing an idle deadline on live streams. Attach ends on its first error and provides no history or automatic reconnect. - Select Rustls in `HttpClientBuilder::http2_prior_knowledge()` so HTTPS advertises `h2` through ALPN. Include a standalone attach example and documentation of admission semantics and stream behavior. ## Testing Add HTTP/2 fixture tests for protobuf framing, authentication metadata, status trailers, event preservation, malformed responses, stream cancellation, redirect rejection, and diagnostic redaction. Add checks for unsafe origins, network policy restrictions, and TLS `h2` advertisement. GitOrigin-RevId: 55db70438bb5e59272c6995c67a46ced5fa28ff4 --- codex-rs/Cargo.lock | 24 ++ codex-rs/Cargo.toml | 1 + codex-rs/cloud-client/BUILD.bazel | 6 + codex-rs/cloud-client/Cargo.toml | 35 +++ codex-rs/cloud-client/README.md | 40 +++ codex-rs/cloud-client/examples/attach.rs | 39 +++ codex-rs/cloud-client/src/client_tests.rs | 348 +++++++++++++++++++++ codex-rs/cloud-client/src/lib.rs | 177 +++++++++++ codex-rs/cloud-client/src/types.rs | 104 ++++++ codex-rs/cloud-client/src/wire.rs | 25 ++ codex-rs/http-client/src/client_builder.rs | 2 + 11 files changed, 801 insertions(+) create mode 100644 codex-rs/cloud-client/BUILD.bazel create mode 100644 codex-rs/cloud-client/Cargo.toml create mode 100644 codex-rs/cloud-client/README.md create mode 100644 codex-rs/cloud-client/examples/attach.rs create mode 100644 codex-rs/cloud-client/src/client_tests.rs create mode 100644 codex-rs/cloud-client/src/lib.rs create mode 100644 codex-rs/cloud-client/src/types.rs create mode 100644 codex-rs/cloud-client/src/wire.rs diff --git a/codex-rs/Cargo.lock b/codex-rs/Cargo.lock index ff2059d2f5..04841ac6a2 100644 --- a/codex-rs/Cargo.lock +++ b/codex-rs/Cargo.lock @@ -2615,6 +2615,30 @@ dependencies = [ "url", ] +[[package]] +name = "codex-cloud-client" +version = "0.0.0" +dependencies = [ + "anyhow", + "async-stream", + "bytes", + "codex-http-client", + "futures", + "http 1.4.0", + "http-body-util", + "hyper", + "hyper-util", + "pretty_assertions", + "prost", + "thiserror 2.0.18", + "tokio", + "tokio-rustls", + "tonic", + "tonic-prost", + "tower", + "url", +] + [[package]] name = "codex-cloud-config" version = "0.0.0" diff --git a/codex-rs/Cargo.toml b/codex-rs/Cargo.toml index 66186f8968..ca575245dc 100644 --- a/codex-rs/Cargo.toml +++ b/codex-rs/Cargo.toml @@ -7,6 +7,7 @@ members = [ "agent-identity", "agent-roles", "backend-client", + "cloud-client", "bwrap", "build-info", "ansi-escape", diff --git a/codex-rs/cloud-client/BUILD.bazel b/codex-rs/cloud-client/BUILD.bazel new file mode 100644 index 0000000000..3686a06b5c --- /dev/null +++ b/codex-rs/cloud-client/BUILD.bazel @@ -0,0 +1,6 @@ +load("//:defs.bzl", "codex_rust_crate") + +codex_rust_crate( + name = "cloud-client", + crate_name = "codex_cloud_client", +) diff --git a/codex-rs/cloud-client/Cargo.toml b/codex-rs/cloud-client/Cargo.toml new file mode 100644 index 0000000000..58ba2d07e8 --- /dev/null +++ b/codex-rs/cloud-client/Cargo.toml @@ -0,0 +1,35 @@ +[package] +name = "codex-cloud-client" +version.workspace = true +edition.workspace = true +license.workspace = true +publish = false + +[lib] +doctest = false + +[lints] +workspace = true + +[dependencies] +async-stream = { workspace = true } +bytes = { workspace = true } +codex-http-client = { workspace = true } +futures = { workspace = true } +http = { workspace = true } +http-body-util = "0.1.3" +prost = "0.14.3" +tonic = { workspace = true } +tonic-prost = { workspace = true } +tower = { version = "0.5.3", features = ["util"] } +thiserror = { workspace = true } +tokio = { workspace = true, features = ["time"] } +url = { workspace = true } + +[dev-dependencies] +anyhow = { workspace = true } +pretty_assertions = { workspace = true } +tokio = { workspace = true, features = ["macros", "rt-multi-thread"] } +hyper = { version = "1.8.1", features = ["server", "http2"] } +hyper-util = { version = "0.1.20", features = ["tokio"] } +tokio-rustls = { workspace = true } diff --git a/codex-rs/cloud-client/README.md b/codex-rs/cloud-client/README.md new file mode 100644 index 0000000000..7d357f8fcf --- /dev/null +++ b/codex-rs/cloud-client/README.md @@ -0,0 +1,40 @@ +# Codex cloud client + +Reusable Rust client for native ThreadService gRPC over HTTP/2, with no TUI, +conventional app-server or private monorepo dependency. This slice supports +`Resume` and live `Attach`. It does not use the HTTP `/grpc` JSON gateway. + +The caller supplies a trusted native gRPC origin (scheme/host/port, no path), +ChatGPT bearer token, selected account ID and configured `HttpClientFactory`. +HTTPS is required except for direct loopback HTTP with an unrestricted, unmanaged +factory. This exception bypasses proxies to keep plaintext credentials local. +For HTTPS, the shared HTTP client supplies proxy, +custom CA and network policy handling; redirects are rejected. Tonic supplies +gRPC framing, protobuf encoding, message bounds (64 MiB) and status/trailer handling. +Bind factory policy before loading credentials; rebuild on login/account changes. + +`ResumeRequest.wait = false` acknowledges admission, not readiness. Requests are +never retried by this crate: a timeout or disconnect can leave work admitted. +`Attach` has no history snapshot or replay guarantee. Dropping it detaches without +stopping the thread or answering approvals. Streams end after their first error. +Resume and Attach setup have a 150-second local deadline; live streams have no idle +deadline. Setup does not send a gRPC deadline that would expire a healthy stream. + +Event variants distinguish notifications and server requests; payloads remain the +complete nested protobuf message bytes, preserving unknown fields. These are not +ProtoJSON or app-server JSON-RPC. Native status code, message and binary details +are available explicitly; credentials, event and error diagnostics are redacted. +Full payload typing, history, approval replies and reconnect are subsequent slices. +The minimal envelopes in `wire.rs` and `ResumeRequest` mirror ThreadService field +numbers; no monorepo checkout or protobuf compiler is required to build this crate. + +Provide `CODEX_CLOUD_ACCESS_TOKEN` and `CODEX_CLOUD_ACCOUNT_ID` through your environment, then run +this standalone consumer from `codex-rs`: + +```sh +cargo run -p codex-cloud-client --example attach -- "$CODEX_CLOUD_GRPC_ENDPOINT" "$CODEX_CLOUD_THREAD_ID" +``` + +Use an existing authorized thread and native gRPC origin. No production URL is +assumed. Public native ingress is a rollout assumption, not proved by fixture +tests. This change does not publish a package to a registry. diff --git a/codex-rs/cloud-client/examples/attach.rs b/codex-rs/cloud-client/examples/attach.rs new file mode 100644 index 0000000000..6f777b3f23 --- /dev/null +++ b/codex-rs/cloud-client/examples/attach.rs @@ -0,0 +1,39 @@ +//! Standalone consumer; prints event kinds without logging payloads. +use codex_cloud_client::Client; +use codex_cloud_client::Credentials; +use codex_cloud_client::ResumeRequest; +use codex_http_client::HttpClientFactory; +use codex_http_client::OutboundProxyPolicy; +use futures::StreamExt; + +#[tokio::main] +async fn main() -> anyhow::Result<()> { + let mut args = std::env::args().skip(1); + let endpoint = args + .next() + .ok_or_else(|| anyhow::anyhow!("expected native gRPC origin"))?; + let thread_id = args + .next() + .ok_or_else(|| anyhow::anyhow!("expected existing thread ID"))?; + let credentials = Credentials::new( + &std::env::var("CODEX_CLOUD_ACCESS_TOKEN")?, + &std::env::var("CODEX_CLOUD_ACCOUNT_ID")?, + )?; + let client = Client::new( + &HttpClientFactory::new(OutboundProxyPolicy::RespectSystemProxy), + &endpoint, + credentials, + )?; + client + .resume(&ResumeRequest { + thread_id: thread_id.clone(), + wait: false, + shared_access: false, + }) + .await?; + let mut events = client.attach(&thread_id).await?; + while let Some(event) = events.next().await { + println!("{:?}", event?); + } + Ok(()) +} diff --git a/codex-rs/cloud-client/src/client_tests.rs b/codex-rs/cloud-client/src/client_tests.rs new file mode 100644 index 0000000000..f571f4ce38 --- /dev/null +++ b/codex-rs/cloud-client/src/client_tests.rs @@ -0,0 +1,348 @@ +//! Exercises native HTTP/2 framing, metadata, trailers and cancellation with synthetic data. +use super::*; +use bytes::Bytes; +use codex_http_client::OutboundProxyPolicy; +use http_body_util::StreamBody; +use hyper::body::Frame; +use hyper::service::service_fn; +use hyper_util::rt::TokioExecutor; +use hyper_util::rt::TokioIo; +use pretty_assertions::assert_eq; +use std::convert::Infallible; +use tokio::sync::mpsc; + +struct Fixture { + client: Client, + requests: mpsc::UnboundedReceiver<(http::request::Parts, Bytes)>, + task: tokio::task::JoinHandle<()>, +} +impl Drop for Fixture { + fn drop(&mut self) { + self.task.abort(); + } +} +async fn fixture(response: http::Response) -> Fixture { + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let endpoint = format!("http://{}", listener.local_addr().unwrap()); + let (sender, requests) = mpsc::unbounded_channel(); + let response = std::sync::Arc::new(std::sync::Mutex::new(Some(response))); + let task = tokio::spawn(async move { + let (socket, _) = listener.accept().await.unwrap(); + let service = service_fn(move |request: http::Request| { + let sender = sender.clone(); + let response = response.lock().unwrap().take().unwrap(); + async move { + let (parts, body) = request.into_parts(); + sender + .send((parts, body.collect().await.unwrap().to_bytes())) + .unwrap(); + Ok::<_, Infallible>(response) + } + }); + let _ = hyper::server::conn::http2::Builder::new(TokioExecutor::new()) + .serve_connection(TokioIo::new(socket), service) + .await; + }); + let client = Client::new( + &HttpClientFactory::new(OutboundProxyPolicy::ReqwestDefault), + &endpoint, + Credentials::new("synthetic-token", "synthetic-account").unwrap(), + ) + .unwrap(); + Fixture { + client, + requests, + task, + } +} +fn frame(payload: &[u8]) -> Bytes { + let mut bytes = vec![0]; + bytes.extend_from_slice(&(payload.len() as u32).to_be_bytes()); + bytes.extend_from_slice(payload); + bytes.into() +} +fn response(messages: Vec, status: tonic::Status) -> http::Response { + let frames = messages + .into_iter() + .map(Frame::data) + .chain([Frame::trailers( + status.into_http::().into_parts().0.headers, + )]); + http::Response::builder() + .header("content-type", "application/grpc") + .body(Body::new(StreamBody::new(futures::stream::iter( + frames.map(Ok::<_, Infallible>), + )))) + .unwrap() +} +async fn assert_request(fixture: &mut Fixture, method: &str, expected: &[u8]) { + let (parts, body) = tokio::time::timeout(Duration::from_secs(5), fixture.requests.recv()) + .await + .unwrap() + .unwrap(); + assert_eq!(parts.method, http::Method::POST); + assert_eq!(parts.version, http::Version::HTTP_2); + assert_eq!( + parts.uri.path(), + format!("/oaiproto.codex_app_server_backend.ThreadService/{method}") + ); + assert_eq!(parts.headers["authorization"], "Bearer synthetic-token"); + assert_eq!(parts.headers["chatgpt-account-id"], "synthetic-account"); + assert_eq!(parts.headers["content-type"], "application/grpc"); + assert!(!parts.headers.contains_key("grpc-timeout")); + assert_eq!(body, frame(expected)); +} +#[tokio::test] +async fn resume_sends_protobuf_and_preserves_admission_flags() { + for wait in [false, true] { + let mut fixture = fixture(response(vec![frame(&[])], tonic::Status::ok(""))).await; + fixture + .client + .resume(&ResumeRequest { + thread_id: "t".into(), + wait, + shared_access: true, + }) + .await + .unwrap(); + // Independent protobuf wire fixtures, not the client's encoder. + let expected: &[u8] = if wait { + &[10, 1, b't', 16, 1, 24, 1] + } else { + &[10, 1, b't', 24, 1] + }; + assert_request(&mut fixture, "Resume", expected).await; + } +} +#[tokio::test] +async fn attach_preserves_opaque_events_and_native_error_details() { + let secret = Bytes::from_static(b"private-error-details"); + let status = tonic::Status::with_details( + tonic::Code::PermissionDenied, + "private-message", + secret.clone(), + ); + let data = [frame(&[10, 3, 10, 1, b'n']), frame(&[18, 3, 10, 1, b'r'])].concat(); + let mut fixture = fixture(response( + data.iter().map(|b| Bytes::from(vec![*b])).collect(), + status, + )) + .await; + let mut events = fixture.client.attach("t").await.unwrap(); + assert_request(&mut fixture, "Attach", &[10, 1, b't']).await; + let notification = events.next().await.unwrap().unwrap(); + assert_eq!(format!("{notification:?}"), "Notification([redacted])"); + let Event::Notification(bytes) = notification else { + panic!("notification expected") + }; + assert_eq!(bytes.as_ref(), &[10, 1, b'n']); + let Event::ServerRequest(bytes) = events.next().await.unwrap().unwrap() else { + panic!("server request expected") + }; + assert_eq!(bytes.as_ref(), &[10, 1, b'r']); + let error = events.next().await.unwrap().unwrap_err(); + assert!(!format!("{error:?} {error}").contains("private")); + let Error::Rpc(status) = error else { + panic!("native status expected") + }; + assert_eq!( + (status.code, status.message, status.details), + (7, "private-message".into(), secret) + ); + assert!(events.next().await.is_none()); +} +#[tokio::test] +async fn successful_stream_ends_and_bad_frames_terminate() { + let fixture = fixture(response(vec![], tonic::Status::ok(""))).await; + assert!( + fixture + .client + .attach("t") + .await + .unwrap() + .next() + .await + .is_none() + ); + for message in [ + frame(&[]), + frame(&[10, 99]), + Bytes::from_static(&[0, 4, 0, 0, 1]), + ] { + let fixture = self::fixture(response(vec![message], tonic::Status::ok(""))).await; + let mut stream = fixture.client.attach("t").await.unwrap(); + assert!(stream.next().await.unwrap().is_err()); + assert!(stream.next().await.is_none()); + } +} +#[tokio::test] +async fn unary_checks_status_trailers_and_preserves_details() { + let fixture = fixture(response( + vec![frame(&[])], + tonic::Status::with_details( + tonic::Code::Unavailable, + "secret", + Bytes::from_static(b"details"), + ), + )) + .await; + let error = fixture + .client + .resume(&ResumeRequest { + thread_id: "t".into(), + wait: false, + shared_access: false, + }) + .await + .unwrap_err(); + let Error::Rpc(status) = error else { + panic!("RPC status expected") + }; + assert_eq!( + (status.code, status.details), + (14, Bytes::from_static(b"details")) + ); +} +#[tokio::test] +async fn dropping_attachment_cancels_idle_response() { + let (sender, dropped) = tokio::sync::oneshot::channel::<()>(); + let stream = async_stream::stream! { + let _sender = sender; + yield Ok::<_, Infallible>(Frame::data(frame(&[10, 0]))); + std::future::pending::<()>().await; + }; + let fixture = fixture( + http::Response::builder() + .header("content-type", "application/grpc") + .body(Body::new(StreamBody::new(stream))) + .unwrap(), + ) + .await; + let mut events = fixture.client.attach("t").await.unwrap(); + events.next().await.unwrap().unwrap(); + drop(events); + assert!( + tokio::time::timeout(Duration::from_secs(5), dropped) + .await + .unwrap() + .is_err() + ); +} +#[tokio::test] +async fn redirects_and_http_failures_are_not_followed() { + for status in [307, 403] { + let fixture = fixture( + http::Response::builder() + .status(status) + .header("location", "http://127.0.0.1:1/private") + .body(Body::empty()) + .unwrap(), + ) + .await; + let error = match fixture.client.attach("t").await { + Ok(mut stream) => stream.next().await.unwrap().unwrap_err(), + Err(error) => error, + }; + assert!(matches!(error, Error::Rpc(_))); + assert!(!format!("{error:?} {error}").contains("private")); + } +} +#[test] +fn rejects_unsafe_origins_and_redacts_credentials() { + for endpoint in [ + "http://example.com", + "https://user:secret@example.com", + "https://example.com?token=secret", + "https://example.com#secret", + "https://example.com/grpc", + ] { + assert!(matches!( + Client::new( + &HttpClientFactory::new(OutboundProxyPolicy::ReqwestDefault), + endpoint, + Credentials::new("token", "account").unwrap(), + ), + Err(Error::InvalidEndpoint) + )); + } + assert!(Credentials::new("secret\nvalue", "account").is_err()); + assert!( + !format!( + "{:?}", + Credentials::new("synthetic-token", "synthetic-account").unwrap() + ) + .contains("synthetic") + ); +} + +#[test] +fn loopback_http_rejects_managed_and_endpoint_restricted_policies() { + for policy in [ + codex_http_client::NetworkPolicyController::default().policy(), + codex_http_client::NetworkPolicy::unmanaged().restrict_to_endpoints(Default::default()), + ] { + let factory = HttpClientFactory::new(OutboundProxyPolicy::RespectSystemProxy) + .with_network_policy(policy); + assert!(matches!( + Client::new( + &factory, + "http://127.0.0.1:1234", + Credentials::new("token", "account").unwrap(), + ), + Err(Error::InvalidEndpoint) + )); + } +} + +#[tokio::test] +async fn https_advertises_h2_in_tls_client_hello() { + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let client = HttpClientBuilder::new() + .http2_prior_knowledge() + .build_direct() + .unwrap(); + let request = client.get(format!("https://{}", listener.local_addr().unwrap())); + let handshake = async { + let (socket, _) = listener.accept().await.unwrap(); + let hello = tokio_rustls::LazyConfigAcceptor::new(Default::default(), socket) + .await + .unwrap(); + let protocols: Vec<_> = hello + .client_hello() + .alpn() + .into_iter() + .flatten() + .map(<[u8]>::to_vec) + .collect(); + assert_eq!(protocols, vec![b"h2".to_vec()]); + // Inspect negotiation without trusting a test CA or completing the TLS handshake. + }; + tokio::time::timeout(Duration::from_secs(5), async { + let (_, result) = tokio::join!(handshake, request.send()); + assert!(result.is_err()); + }) + .await + .unwrap(); +} + +#[tokio::test] +async fn https_enforces_unmanaged_endpoint_restrictions_before_connecting() { + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let factory = HttpClientFactory::new(OutboundProxyPolicy::ReqwestDefault).with_network_policy( + codex_http_client::NetworkPolicy::unmanaged().restrict_to_endpoints(Default::default()), + ); + let client = Client::new( + &factory, + &format!("https://{}", listener.local_addr().unwrap()), + Credentials::new("token", "account").unwrap(), + ) + .unwrap(); + tokio::time::timeout(Duration::from_secs(5), async { + tokio::select! { + result = client.attach("t") => assert!(result.is_err()), + _ = listener.accept() => panic!("denied destination was contacted"), + } + }) + .await + .unwrap(); +} diff --git a/codex-rs/cloud-client/src/lib.rs b/codex-rs/cloud-client/src/lib.rs new file mode 100644 index 0000000000..ff1fdd5540 --- /dev/null +++ b/codex-rs/cloud-client/src/lib.rs @@ -0,0 +1,177 @@ +//! Native gRPC client for the Codex cloud ThreadService, using shared HTTP transport policy. +//! +//! Credentials and account selection belong to the caller. Requests are never retried: +//! a timeout or disconnect can leave Resume admitted. Attach delivers live events only; +//! dropping its stream detaches without interrupting the thread or answering approvals. + +mod types; +mod wire; + +pub use types::Credentials; +pub use types::Error; +pub use types::Event; +pub use types::ResumeRequest; +pub use types::RpcStatus; + +use codex_http_client::ClientRouteClass; +use codex_http_client::HttpClientBuilder; +use codex_http_client::HttpClientFactory; +use codex_http_client::RouteAwareClientPool; +use futures::StreamExt; +use futures::stream::BoxStream; +use http_body_util::BodyExt; +use std::io; +use std::time::Duration; +use tonic::body::Body; +use tonic::client::Grpc; +use tonic_prost::ProstCodec; +use tower::service_fn; +use tower::util::BoxCloneSyncService; +use url::Url; + +const REQUEST_TIMEOUT: Duration = Duration::from_secs(150); +const MAX_MESSAGE_BYTES: usize = 64 * 1024 * 1024; +type Transport = BoxCloneSyncService, http::Response, io::Error>; + +/// One fixed origin and account. Construct a new client when credentials change. +/// Supply a trusted native gRPC origin, without a path (not the HTTP `/grpc` gateway). +pub struct Client { + grpc: Grpc, + credentials: Credentials, +} + +impl Client { + pub fn new( + factory: &HttpClientFactory, + endpoint: &str, + credentials: Credentials, + ) -> Result { + let endpoint = Url::parse(endpoint).map_err(|_| Error::InvalidEndpoint)?; + let loopback = match endpoint.host() { + Some(url::Host::Ipv4(ip)) => ip.is_loopback(), + Some(url::Host::Ipv6(ip)) => ip.is_loopback(), + Some(url::Host::Domain(name)) => name == "localhost", + None => false, + }; + if (endpoint.scheme() != "https" && !(endpoint.scheme() == "http" && loopback)) + || !endpoint.username().is_empty() + || endpoint.password().is_some() + || endpoint.query().is_some() + || endpoint.fragment().is_some() + || endpoint.path() != "/" + { + return Err(Error::InvalidEndpoint); + } + let local_http = endpoint.scheme() == "http"; + if local_http + && (factory.network_policy().is_managed() + || factory + .network_policy() + .acquire_for_unsupported_sdk() + .is_err()) + { + return Err(Error::InvalidEndpoint); + } + let builder = HttpClientBuilder::new() + .without_redirects() + .without_request_logging() + .http2_prior_knowledge() + .connect_timeout(Duration::from_secs(30)); + let http = if local_http { + // Plaintext credentials must never leave loopback through an outbound proxy. + builder.build_direct().map_err(|_| Error::ClientBuild)? + } else { + RouteAwareClientPool::with_builder(factory.clone(), ClientRouteClass::Api, builder) + .into_client() + }; + let origin = endpoint + .as_str() + .parse() + .map_err(|_| Error::InvalidEndpoint)?; + let transport = service_fn(move |request: http::Request| { + let http = http.clone(); + async move { + let (parts, body) = request.into_parts(); + let response = http + .request(parts.method, parts.uri.to_string()) + .version(parts.version) + .headers(parts.headers) + .body_stream(body.into_data_stream()) + .send() + .await + .map_err(|_| io::Error::other("gRPC transport failed"))?; + Ok(response.into_http_response().map(Body::new)) + } + }); + Ok(Self { + grpc: Grpc::with_origin(BoxCloneSyncService::new(transport), origin) + .max_decoding_message_size(MAX_MESSAGE_BYTES) + .max_encoding_message_size(MAX_MESSAGE_BYTES), + credentials, + }) + } + + /// With `wait: false`, success means admission only. Failures may leave work admitted. + pub async fn resume(&self, request: &ResumeRequest) -> Result<(), Error> { + let mut grpc = self.grpc.clone(); + let mut request = tonic::Request::new(request.clone()); + *request.metadata_mut() = + tonic::metadata::MetadataMap::from_headers(self.credentials.headers.clone()); + // Bound the whole unary RPC, including its response body and trailers. + tokio::time::timeout( + REQUEST_TIMEOUT, + grpc.unary( + request, + http::uri::PathAndQuery::from_static( + "/oaiproto.codex_app_server_backend.ThreadService/Resume", + ), + ProstCodec::::default(), + ), + ) + .await + .map_err(|_| Error::Timeout)? + .map_err(Error::from)?; + Ok(()) + } + + /// Live events only, without history or automatic reconnect. The stream terminates + /// on its first error. Setup is bounded; there is no idle or total stream deadline. + pub async fn attach( + &self, + thread_id: &str, + ) -> Result>, Error> { + let mut grpc = self.grpc.clone(); + let mut request = tonic::Request::new(wire::AttachRequest { + thread_id: thread_id.to_owned(), + }); + *request.metadata_mut() = + tonic::metadata::MetadataMap::from_headers(self.credentials.headers.clone()); + let response = tokio::time::timeout( + REQUEST_TIMEOUT, + grpc.server_streaming( + request, + http::uri::PathAndQuery::from_static( + "/oaiproto.codex_app_server_backend.ThreadService/Attach", + ), + ProstCodec::::default(), + ), + ) + .await + .map_err(|_| Error::Timeout)? + .map_err(Error::from)?; + let mut stream = response.into_inner(); + Ok(async_stream::try_stream! { + while let Some(message) = stream.message().await.map_err(Error::from)? { + yield match message.event.ok_or(Error::InvalidResponse)? { + wire::Payload::Notification(bytes) => Event::Notification(bytes), + wire::Payload::ServerRequest(bytes) => Event::ServerRequest(bytes), + }; + } + } + .boxed()) + } +} + +#[cfg(test)] +#[path = "client_tests.rs"] +mod tests; diff --git a/codex-rs/cloud-client/src/types.rs b/codex-rs/cloud-client/src/types.rs new file mode 100644 index 0000000000..666afc07de --- /dev/null +++ b/codex-rs/cloud-client/src/types.rs @@ -0,0 +1,104 @@ +//! Caller-facing native gRPC types. Diagnostics never include credentials or server payloads. + +use bytes::Bytes; +use http::HeaderMap; +use http::HeaderValue; +use http::header::AUTHORIZATION; +use std::fmt; + +/// Caller-owned ChatGPT bearer token and selected account, validated as HTTP headers. +#[derive(Clone, Debug)] +pub struct Credentials { + pub(crate) headers: HeaderMap, +} + +impl Credentials { + pub fn new(bearer_token: &str, account_id: &str) -> Result { + if bearer_token.is_empty() || account_id.is_empty() { + return Err(Error::InvalidCredentials); + } + let mut headers = HeaderMap::new(); + for (name, value) in [ + (AUTHORIZATION, format!("Bearer {bearer_token}")), + ( + http::HeaderName::from_static("chatgpt-account-id"), + account_id.to_owned(), + ), + ] { + let mut value = HeaderValue::from_str(&value).map_err(|_| Error::InvalidCredentials)?; + value.set_sensitive(true); + headers.insert(name, value); + } + Ok(Self { headers }) + } +} + +/// Resume an existing thread; `wait` distinguishes admission from readiness. +#[derive(Clone, prost::Message)] +#[prost(skip_debug)] +pub struct ResumeRequest { + #[prost(string, tag = "1")] + pub thread_id: String, + #[prost(bool, tag = "2")] + pub wait: bool, + #[prost(bool, tag = "3")] + pub shared_access: bool, +} + +/// An Attach event containing the complete nested protobuf message bytes. +/// Payloads are native Notification/ServerRequest messages, not JSON-RPC or ProtoJSON. +/// Keeping them opaque preserves unknown fields until full payload typing is added. +pub enum Event { + Notification(Bytes), + ServerRequest(Bytes), +} + +impl fmt::Debug for Event { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.write_str(match self { + Self::Notification(_) => "Notification([redacted])", + Self::ServerRequest(_) => "ServerRequest([redacted])", + }) + } +} + +/// Native gRPC status, including binary error details. Payloads are redacted in diagnostics. +pub struct RpcStatus { + pub code: i32, + pub message: String, + pub details: Bytes, +} + +impl fmt::Debug for RpcStatus { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.debug_struct("RpcStatus") + .field("code", &self.code) + .finish_non_exhaustive() + } +} + +impl From for Error { + fn from(status: tonic::Status) -> Self { + Self::Rpc(RpcStatus { + code: status.code() as i32, + message: status.message().to_owned(), + details: Bytes::copy_from_slice(status.details()), + }) + } +} + +#[derive(Debug, thiserror::Error)] +pub enum Error { + #[error("invalid gRPC origin (HTTPS required except on unrestricted unmanaged loopback)")] + InvalidEndpoint, + #[error("invalid bearer token or account header")] + InvalidCredentials, + #[error("could not construct the HTTP client")] + ClientBuild, + #[error("gRPC request timed out; request outcome may be unknown")] + Timeout, + #[error("gRPC failed; request outcome may be unknown: {0:?}")] + Rpc(RpcStatus), + #[error("invalid gRPC event envelope")] + InvalidResponse, +} diff --git a/codex-rs/cloud-client/src/wire.rs b/codex-rs/cloud-client/src/wire.rs new file mode 100644 index 0000000000..644ead33ef --- /dev/null +++ b/codex-rs/cloud-client/src/wire.rs @@ -0,0 +1,25 @@ +//! Minimal ThreadService protobuf envelopes; field numbers are the wire contract. +//! Length-delimited nested events remain intact, including unknown protobuf fields. + +#[derive(Clone, prost::Message)] +pub(crate) struct ResumeResponse {} + +#[derive(Clone, prost::Message)] +pub(crate) struct AttachRequest { + #[prost(string, tag = "1")] + pub thread_id: String, +} + +#[derive(Clone, prost::Message)] +pub(crate) struct AttachResponse { + #[prost(oneof = "Payload", tags = "1, 2")] + pub event: Option, +} + +#[derive(Clone, prost::Oneof)] +pub(crate) enum Payload { + #[prost(bytes = "bytes", tag = "1")] + Notification(bytes::Bytes), + #[prost(bytes = "bytes", tag = "2")] + ServerRequest(bytes::Bytes), +} diff --git a/codex-rs/http-client/src/client_builder.rs b/codex-rs/http-client/src/client_builder.rs index a3ad63d007..d3b8054767 100644 --- a/codex-rs/http-client/src/client_builder.rs +++ b/codex-rs/http-client/src/client_builder.rs @@ -101,6 +101,8 @@ impl HttpClientBuilder { /// Uses HTTP/2 for SDKs such as gRPC that require framed bidirectional bodies. pub fn http2_prior_knowledge(mut self) -> Self { + // Native TLS may lack ALPN support; HTTPS HTTP/2 requires advertising h2. + self.tls_backend = TlsBackend::Rustls; self.http2_prior_knowledge = true; self }