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 }