From f64a52b6aa40038dc09dff0f920733caa8302b5e Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Mon, 24 Aug 2026 16:09:55 +0200 Subject: [PATCH] Isolate Codex auth and provider transport --- AGENTS.md | 17 +- Cargo.lock | 248 +++++++++ Cargo.toml | 3 + README.md | 30 +- src/lib.rs | 8 + src/provider_backend.rs | 1168 +++++++++++++++++++++++++++++++++++++++ 6 files changed, 1459 insertions(+), 15 deletions(-) create mode 100644 src/provider_backend.rs diff --git a/AGENTS.md b/AGENTS.md index 6b397e7..2b6608b 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -7,11 +7,18 @@ owner, tool executor, decision store, or consumer runtime. Keep one Cargo package and one production daemon entrypoint. Add another target only when an independent lifecycle or privilege invariant requires it. -The daemon may own credential loading and refresh, upstream client identity, -authenticated caller admission, bounded concurrency/replay, provider transport, -typed events, and transport receipts. Consumers own native requests, provider -lowering, prompts, schemas, tool execution, retry between passes, interpretation, -decisions, and state admission. +The public daemon owns upstream client identity, authenticated caller admission, +bounded concurrency/replay, provider transport, typed events, and transport +receipts. A private pinned official `codex app-server` child is the sole +credential-store writer and refresh authority. The daemon may read its +credential only after their private auth RPC completes; it may never write or +repair the store. Consumers own native requests, provider lowering, prompts, +schemas, tool execution, retry between passes, interpretation, decisions, and +state admission. + +Do not link upstream Codex crates into this package. Idunn freezes the official +Codex binary as an exact package input. The child receives auth RPC only, never +consumer identity, prompts, tools, provider requests, or model output. Use typed Rust contracts, CultCache state, CultNet transport, and CultMesh/Eve projection. JSON is permitted only at the upstream provider boundary or for diff --git a/Cargo.lock b/Cargo.lock index 783aafd..90f8929 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -43,6 +43,12 @@ version = "1.5.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f2032f911046de80f0a198e0901378627c33f59ea0ac00e363d481118bd70a53" +[[package]] +name = "base64" +version = "0.23.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ac07cdecf99051d9a5238b80f35af32cdeba5b336e55d957b318b50137e18da5" + [[package]] name = "block-buffer" version = "0.10.4" @@ -52,6 +58,22 @@ dependencies = [ "generic-array", ] +[[package]] +name = "bytes" +version = "1.12.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "fc652a48c352aef3ea3aed32080501cf3ef6ed5da78602a020c991775b0aff04" + +[[package]] +name = "cc" +version = "1.4.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0ad534f4357a5264cce5019c989cf66a4f0dc4e0d1b1d15f8aacec0ff7360273" +dependencies = [ + "find-msvc-tools", + "shlex", +] + [[package]] name = "cfg-if" version = "1.0.4" @@ -73,12 +95,15 @@ name = "codex-connector" version = "0.1.0" dependencies = [ "aes-gcm", + "base64", "rand_core", "rmp-serde", "serde", "serde_json", "sha2", "thiserror", + "ureq", + "zeroize", ] [[package]] @@ -120,6 +145,12 @@ dependencies = [ "crypto-common", ] +[[package]] +name = "find-msvc-tools" +version = "0.1.11" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d45db016d36b838f563236e9193d0ee6ce38f3f68b6c94e914b4929c96bbb890" + [[package]] name = "generic-array" version = "0.14.7" @@ -151,6 +182,22 @@ dependencies = [ "polyval", ] +[[package]] +name = "http" +version = "1.5.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "918d3568bebf352712bc2ef3d46a8bcf1a75b373be6539de198e9105cbbf9ce0" +dependencies = [ + "bytes", + "itoa", +] + +[[package]] +name = "httparse" +version = "1.10.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6dbf3de79e51f3d586ab4cb9d5c3e2c14aa28ed23d180cf89b4df0454a69cc87" + [[package]] name = "inout" version = "0.1.4" @@ -172,6 +219,12 @@ version = "0.2.189" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "3eaf3ede3fee6db1a4c2ee091bf8a8b4dccdc6d17f656fb07896ee72867612f2" +[[package]] +name = "log" +version = "0.4.34" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f9f8bd3e56ce4dfc153cf470fffbfa98c7620958b312ca5c3a4b8d5181fd13c6" + [[package]] name = "memchr" version = "2.8.3" @@ -187,12 +240,24 @@ dependencies = [ "autocfg", ] +[[package]] +name = "once_cell" +version = "1.21.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9f7c3e4beb33f85d45ae3e3a1792185706c8e16d043238c593331cc7cd313b50" + [[package]] name = "opaque-debug" version = "0.3.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "c08d65885ee38876c4f86fa503fb49d7b507c2b62552df7c70b2fce627e06381" +[[package]] +name = "percent-encoding" +version = "2.3.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9b4f627cb1b25917193a259e49bdad08f671f8d9708acfd5fe0a8c1455d87220" + [[package]] name = "polyval" version = "0.6.2" @@ -232,6 +297,20 @@ dependencies = [ "getrandom", ] +[[package]] +name = "ring" +version = "0.17.14" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a4689e6c2294d81e88dc6261c768b63bc4fcdb852be6d1352498b114f61383b7" +dependencies = [ + "cc", + "cfg-if", + "getrandom", + "libc", + "untrusted", + "windows-sys", +] + [[package]] name = "rmp" version = "0.8.15" @@ -251,6 +330,41 @@ dependencies = [ "serde", ] +[[package]] +name = "rustls" +version = "0.23.43" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0283386ce02abc0151e1761d08802dfe86c173b0b494af5cbc086574e453da06" +dependencies = [ + "log", + "once_cell", + "ring", + "rustls-pki-types", + "rustls-webpki", + "subtle", + "zeroize", +] + +[[package]] +name = "rustls-pki-types" +version = "1.15.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2f4925028c7eb5d1fcdaf196971378ed9d2c1c4efc7dc5d011256f76c99c0a96" +dependencies = [ + "zeroize", +] + +[[package]] +name = "rustls-webpki" +version = "0.103.15" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f3c3cf1d8b1e7d4927e2d154c3fcb02979afb9939629c62cd9048d4f07b60ac2" +dependencies = [ + "ring", + "rustls-pki-types", + "untrusted", +] + [[package]] name = "serde" version = "1.0.229" @@ -305,6 +419,12 @@ dependencies = [ "digest", ] +[[package]] +name = "shlex" +version = "2.0.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f8fadd59c855ef2080decdef8ff161eb6661b86933c9d82e5ba29dc602a55aba" + [[package]] name = "subtle" version = "2.6.1" @@ -364,6 +484,46 @@ dependencies = [ "subtle", ] +[[package]] +name = "untrusted" +version = "0.9.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8ecb6da28b8a351d773b68d5825ac39017e680750f980f3a1a85cd8dd28a47c1" + +[[package]] +name = "ureq" +version = "3.4.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "972d7902c8735f2695410b8aed7df6ed12a47394aa1c8d7af49f0497b731a94d" +dependencies = [ + "base64", + "log", + "percent-encoding", + "rustls", + "rustls-pki-types", + "ureq-proto", + "utf8-zero", + "webpki-roots", +] + +[[package]] +name = "ureq-proto" +version = "0.6.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "da5f78b09e6941e1a0f2e30e695e4b120377b54d5e0aec11b594bb57b3971613" +dependencies = [ + "base64", + "http", + "httparse", + "log", +] + +[[package]] +name = "utf8-zero" +version = "0.8.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b8c0a043c9540bae7c578c88f91dda8bd82e59ae27c21baca69c8b191aaf5a6e" + [[package]] name = "version_check" version = "0.9.5" @@ -376,6 +536,94 @@ version = "0.11.1+wasi-snapshot-preview1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ccf3ec651a847eb01de73ccad15eb7d99f80485de043efb2f370cd654f4ea44b" +[[package]] +name = "webpki-roots" +version = "1.0.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7dcd9d09a39985f5344844e66b0c530a33843579125f23e21e9f0f220850f22a" +dependencies = [ + "rustls-pki-types", +] + +[[package]] +name = "windows-sys" +version = "0.52.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "282be5f36a8ce781fad8c8ae18fa3f9beff57ec1b52cb3de0789201425d9a33d" +dependencies = [ + "windows-targets", +] + +[[package]] +name = "windows-targets" +version = "0.52.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9b724f72796e036ab90c1021d4780d4d3d648aca59e491e6b98e725b84e99973" +dependencies = [ + "windows_aarch64_gnullvm", + "windows_aarch64_msvc", + "windows_i686_gnu", + "windows_i686_gnullvm", + "windows_i686_msvc", + "windows_x86_64_gnu", + "windows_x86_64_gnullvm", + "windows_x86_64_msvc", +] + +[[package]] +name = "windows_aarch64_gnullvm" +version = "0.52.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "32a4622180e7a0ec044bb555404c800bc9fd9ec262ec147edd5989ccd0c02cd3" + +[[package]] +name = "windows_aarch64_msvc" +version = "0.52.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "09ec2a7bb152e2252b53fa7803150007879548bc709c039df7627cabbd05d469" + +[[package]] +name = "windows_i686_gnu" +version = "0.52.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8e9b5ad5ab802e97eb8e295ac6720e509ee4c243f69d781394014ebfe8bbfa0b" + +[[package]] +name = "windows_i686_gnullvm" +version = "0.52.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0eee52d38c090b3caa76c563b86c3a4bd71ef1a819287c19d586d7334ae8ed66" + +[[package]] +name = "windows_i686_msvc" +version = "0.52.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "240948bc05c5e7c6dabba28bf89d89ffce3e303022809e73deaefe4f6ec56c66" + +[[package]] +name = "windows_x86_64_gnu" +version = "0.52.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "147a5c80aabfbf0c7d901cb5895d1de30ef2907eb21fbbab29ca94c5b08b1a78" + +[[package]] +name = "windows_x86_64_gnullvm" +version = "0.52.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "24d5b23dc417412679681396f2b49f3de8c1473deb516bd34410872eff51ed0d" + +[[package]] +name = "windows_x86_64_msvc" +version = "0.52.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "589f6da84c646204747d1270a2a5661ea66ed1cced2631d546fdfb155959f9ec" + +[[package]] +name = "zeroize" +version = "1.9.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e13c156562582aa81c60cb29407084cdb54c4164760106ab78e6c5b0858cf64e" + [[package]] name = "zmij" version = "1.0.23" diff --git a/Cargo.toml b/Cargo.toml index c807822..f681ec3 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -13,9 +13,12 @@ path = "src/lib.rs" [dependencies] aes-gcm = "0.10" +base64 = "0.23" rand_core = { version = "0.6", features = ["getrandom"] } rmp-serde = "1" serde = { version = "1", features = ["derive"] } serde_json = "1" sha2 = "0.10" thiserror = "2" +ureq = { version = "3.4", default-features = false, features = ["rustls"] } +zeroize = "1" diff --git a/README.md b/README.md index 6d5d216..cb384a5 100644 --- a/README.md +++ b/README.md @@ -5,17 +5,27 @@ subscription inference. Epiphany and Ghostlight remain separate minds. They derive their own exact provider requests, execute their own tools, interpret their own results, and admit their own state. -The connector owns one writable credential/refresh store, authenticated caller -admission, bounded transport physiology, exact provider-request receipts, and -redacted CultMesh health. It owns no prompt policy or decision authority. +The service owns authenticated caller admission, bounded transport physiology, +exact provider-request receipts, and redacted CultMesh health. A private pinned +official `codex app-server` child is the sole writer and refresh authority for +one credential store. The public connector reads the credential only after the +child's private auth RPC completes, then performs the exact Responses request. +It owns no prompt policy or decision authority. -The current source establishes the v2 typed contract and its transport-neutral -service core before adding the daemon shell. The core admits distinct caller -keys and policies, returns an exact cached response for byte-identical retries, -refuses conflicting or concurrent reuse of one caller/request identity, and -leaves provider execution outside its lock. The existing Yggdrasil connector -already proves the process seam, but its single-caller, fixed-model, stateless, -no-tools wire law cannot become the shared contract by inertia. +The current source establishes the v2 typed contract, its transport-neutral +service core, and the private Codex-auth/provider backend before adding the +public listener shell. The core admits distinct caller keys and policies, +returns an exact cached response for byte-identical retries, refuses conflicting +or concurrent reuse of one caller/request identity, and leaves provider +execution outside its lock. The backend verifies the exact child binary and +reported Codex home, limits child RPC to authentication, reads API-key or +ChatGPT credentials without publishing their identity, retries one ChatGPT 401 +only after the official writer advances the credential store, and lowers raw +SSE into typed text, tool-call, usage, and failure receipts. + +No upstream Codex crate is linked into the package. The official binary is a +pinned deployment input, keeping Codex's application build graph outside both +consumers and this small daemon. ## Contract invariant diff --git a/src/lib.rs b/src/lib.rs index d6e8662..478c719 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -7,6 +7,14 @@ use serde::{Deserialize, Serialize}; use sha2::{Digest, Sha256}; use thiserror::Error; +mod provider_backend; + +pub use provider_backend::CodexAppServerConfig; +pub use provider_backend::CodexAuthMode; +pub use provider_backend::CodexAuthReadiness; +pub use provider_backend::CodexProviderBackend; +pub use provider_backend::CodexProviderBackendError; + pub const PROVIDER_REQUEST_SCHEMA_ID: &str = "gamecult.codex.provider_request.v2"; pub const INVOCATION_SCHEMA_ID: &str = "gamecult.codex.transport_invocation.v2"; pub const RESULT_SCHEMA_ID: &str = "gamecult.codex.transport_result.v2"; diff --git a/src/provider_backend.rs b/src/provider_backend.rs new file mode 100644 index 0000000..83e1ae0 --- /dev/null +++ b/src/provider_backend.rs @@ -0,0 +1,1168 @@ +use std::collections::HashMap; +use std::fs::File; +use std::io::{self, BufRead, BufReader, Read, Write}; +use std::path::{Path, PathBuf}; +use std::process::{Child, ChildStdin, ChildStdout, Command, Stdio}; +use std::sync::Mutex; +use std::time::Duration; + +use base64::Engine; +use base64::engine::general_purpose::URL_SAFE_NO_PAD; +use serde::Deserialize; +use serde_json::{Value, json}; +use sha2::{Digest, Sha256}; +use thiserror::Error; +use zeroize::Zeroizing; + +use crate::CodexTransportEvent; +use crate::CodexTransportEventPayload; +use crate::CodexTransportInvocation; +use crate::CodexTransportOutcome; +use crate::CodexTransportReceipt; +use crate::CodexTransportResult; +use crate::RECEIPT_SCHEMA_ID; +use crate::canonical_responses_body_bytes; + +const MAX_APP_SERVER_LINE_BYTES: usize = 1024 * 1024; +const MAX_INTERLEAVED_MESSAGES: usize = 64; +const MAX_SSE_FRAME_BYTES: usize = 2 * 1024 * 1024; +const MAX_STREAM_BYTES: usize = 64 * 1024 * 1024; +const OPENAI_RESPONSES_URL: &str = "https://api.openai.com/v1/responses"; +const CHATGPT_RESPONSES_URL: &str = "https://chatgpt.com/backend-api/codex/responses"; +const CONNECTOR_ORIGINATOR: &str = "gamecult_codex_connector"; +const TRANSPORT_ID: &str = "codex_connector_responses_http_v2"; + +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct CodexAppServerConfig { + pub executable: PathBuf, + pub executable_sha256: [u8; 32], + pub codex_home: PathBuf, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum CodexAuthMode { + ApiKey, + Chatgpt, +} + +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct CodexAuthReadiness { + pub auth_mode: CodexAuthMode, + pub credential_loaded: bool, + pub fedramp_routing: bool, + pub app_server_user_agent: String, +} + +pub struct CodexProviderBackend { + authority: Mutex, + http: ureq::Agent, +} + +impl CodexProviderBackend { + pub fn start(config: CodexAppServerConfig) -> Result { + let http: ureq::Agent = ureq::Agent::config_builder() + .timeout_connect(Some(Duration::from_secs(30))) + .timeout_recv_response(Some(Duration::from_secs(30))) + .timeout_recv_body(Some(Duration::from_secs(300))) + .max_redirects(0) + .build() + .into(); + Ok(Self { + authority: Mutex::new(CodexAppServerAuthority::spawn(config)?), + http, + }) + } + + pub fn readiness(&self) -> Result { + let mut authority = self + .authority + .lock() + .map_err(|_| CodexProviderBackendError::CredentialAuthorityPoisoned)?; + let credential = authority.read_credential(false)?; + Ok(credential.redacted(&authority.app_server_user_agent)) + } + + pub fn execute( + &self, + invocation: &CodexTransportInvocation, + ) -> Result { + let body = canonical_responses_body_bytes(&invocation.request) + .map_err(|_| CodexProviderBackendError::ProviderRequest)?; + let (first, user_agent) = { + let mut authority = self + .authority + .lock() + .map_err(|_| CodexProviderBackendError::CredentialAuthorityPoisoned)?; + let credential = authority.read_credential(false)?; + (credential, authority.app_server_user_agent.clone()) + }; + let first_digest = first.auth_file_sha256; + let response = match send_provider_request( + &self.http, + provider_url(first.mode), + invocation, + &body, + &first, + &user_agent, + ) { + Err(ureq::Error::StatusCode(401)) if first.mode == CodexAuthMode::Chatgpt => { + let (refreshed, user_agent) = { + let mut authority = self + .authority + .lock() + .map_err(|_| CodexProviderBackendError::CredentialAuthorityPoisoned)?; + let current = authority.read_credential(false)?; + let credential = if current.auth_file_sha256 != first_digest { + current + } else { + authority.read_credential(true)? + }; + (credential, authority.app_server_user_agent.clone()) + }; + if refreshed.auth_file_sha256 == first_digest { + return Ok(failed_result( + invocation, + "credential_refresh", + "Codex credential refresh did not advance the credential store", + )); + } + send_provider_request( + &self.http, + provider_url(refreshed.mode), + invocation, + &body, + &refreshed, + &user_agent, + ) + } + result => result, + }; + let response = match response { + Ok(response) => response, + Err(ureq::Error::StatusCode(status)) => { + return Ok(failed_result( + invocation, + "provider_http_status", + &format!("provider returned HTTP {status}"), + )); + } + Err(_) => { + return Ok(failed_result( + invocation, + "provider_transport", + "provider transport failed", + )); + } + }; + let parsed = match parse_responses_sse(response.into_body().into_reader()) { + Ok(parsed) => parsed, + Err(message) => return Ok(failed_result(invocation, "provider_stream", message)), + }; + let receipt = CodexTransportReceipt { + schema_id: RECEIPT_SCHEMA_ID.to_string(), + request_id: invocation.request_id().to_string(), + caller_runtime_id: invocation.caller_runtime_id.clone(), + native_request_sha256: invocation.native_request_sha256, + provider_request_sha256: invocation.provider_request_sha256, + model: invocation.request.model.clone(), + transport: TRANSPORT_ID.to_string(), + outcome: parsed.outcome, + }; + Ok(CodexTransportResult::transported( + invocation, + parsed.events, + receipt, + )) + } +} + +fn provider_url(mode: CodexAuthMode) -> &'static str { + match mode { + CodexAuthMode::ApiKey => OPENAI_RESPONSES_URL, + CodexAuthMode::Chatgpt => CHATGPT_RESPONSES_URL, + } +} + +fn send_provider_request( + http: &ureq::Agent, + url: &str, + invocation: &CodexTransportInvocation, + body: &[u8], + credential: &CodexCredentialSnapshot, + user_agent: &str, +) -> Result, ureq::Error> { + let authorization = Zeroizing::new(format!("Bearer {}", credential.bearer.as_str())); + let mut request = http + .post(url) + .header("authorization", authorization.as_str()) + .header("accept", "text/event-stream") + .header("content-type", "application/json") + .header("originator", CONNECTOR_ORIGINATOR) + .header("user-agent", user_agent) + .header("session_id", &invocation.request.conversation_id) + .header("x-client-request-id", invocation.request_id()) + .header("version", env!("CARGO_PKG_VERSION")); + if let Some(account_id) = &credential.account_id { + request = request.header("ChatGPT-Account-ID", account_id); + } + if credential.is_fedramp { + request = request.header("X-OpenAI-Fedramp", "true"); + } + request.send(body) +} + +struct ParsedProviderStream { + events: Vec, + outcome: CodexTransportOutcome, +} + +#[derive(Default)] +struct PendingToolCall { + call_id: Option, + name: Option, + arguments: String, +} + +struct StreamState { + events: Vec, + pending_tools: HashMap, + outcome: Option, +} + +impl StreamState { + fn new() -> Self { + Self { + events: Vec::new(), + pending_tools: HashMap::new(), + outcome: None, + } + } + + fn push(&mut self, payload: CodexTransportEventPayload) { + self.events.push(CodexTransportEvent { + sequence: self.events.len() as u64, + payload, + }); + } + + fn accept_frame(&mut self, data: &[u8]) -> Result<(), &'static str> { + if data == b"[DONE]" { + return Ok(()); + } + let event: Value = + serde_json::from_slice(data).map_err(|_| "provider emitted malformed SSE JSON")?; + let kind = event + .get("type") + .and_then(Value::as_str) + .ok_or("provider SSE event omitted its type")?; + match kind { + "response.output_text.delta" => { + let text = event + .get("delta") + .and_then(Value::as_str) + .ok_or("provider text delta omitted text")?; + if !text.is_empty() { + self.push(CodexTransportEventPayload::TextDelta { + text: text.to_string(), + }); + } + } + "response.output_item.added" => { + if let Some(item) = event.get("item") { + self.seed_tool(item)?; + } + } + "response.function_call_arguments.delta" => { + let identity = event + .get("item_id") + .or_else(|| event.get("call_id")) + .and_then(Value::as_str) + .ok_or("provider tool delta omitted its identity")?; + let delta = event + .get("delta") + .and_then(Value::as_str) + .ok_or("provider tool delta omitted arguments")?; + let pending = self.pending_tools.entry(identity.to_string()).or_default(); + if pending.call_id.is_none() { + pending.call_id = event + .get("call_id") + .and_then(Value::as_str) + .map(str::to_owned); + } + pending.arguments.push_str(delta); + } + "response.output_item.done" => { + if let Some(item) = event.get("item") { + self.finish_tool(item)?; + } + } + "response.completed" => self.complete(event.get("response"))?, + "response.failed" => { + self.outcome = Some(CodexTransportOutcome::Failed { + failure_kind: "response_failed".to_string(), + message: "provider declared response.failed".to_string(), + }); + } + "response.incomplete" => { + self.outcome = Some(CodexTransportOutcome::Failed { + failure_kind: "response_incomplete".to_string(), + message: "provider declared response.incomplete".to_string(), + }); + } + _ => {} + } + Ok(()) + } + + fn seed_tool(&mut self, item: &Value) -> Result<(), &'static str> { + if item.get("type").and_then(Value::as_str) != Some("function_call") { + return Ok(()); + } + let identity = item_identity(item).ok_or("provider tool item omitted its identity")?; + self.pending_tools.insert( + identity, + PendingToolCall { + call_id: item_string(item, "call_id"), + name: item_string(item, "name"), + arguments: item_string(item, "arguments").unwrap_or_default(), + }, + ); + Ok(()) + } + + fn finish_tool(&mut self, item: &Value) -> Result<(), &'static str> { + if item.get("type").and_then(Value::as_str) != Some("function_call") { + return Ok(()); + } + let identity = item_identity(item).ok_or("provider tool item omitted its identity")?; + let pending = self.pending_tools.remove(&identity).unwrap_or_default(); + let call_id = item_string(item, "call_id") + .or(pending.call_id) + .ok_or("provider tool item omitted its call ID")?; + let name = item_string(item, "name") + .or(pending.name) + .ok_or("provider tool item omitted its name")?; + let arguments = item_string(item, "arguments").unwrap_or(pending.arguments); + if arguments.is_empty() { + return Err("provider tool item omitted its arguments"); + } + self.push(CodexTransportEventPayload::ToolCall { + call_id, + name, + arguments, + }); + Ok(()) + } + + fn complete(&mut self, response: Option<&Value>) -> Result<(), &'static str> { + if self.outcome.is_some() { + return Err("provider emitted more than one terminal event"); + } + if !self.pending_tools.is_empty() { + return Err("provider completed with an unfinished tool call"); + } + let response = response.ok_or("provider completion omitted its response")?; + let response_id = response + .get("id") + .and_then(Value::as_str) + .filter(|value| !value.is_empty()) + .ok_or("provider completion omitted its response ID")?; + let usage = response.get("usage"); + self.outcome = Some(CodexTransportOutcome::Completed { + provider_response_id: Some(response_id.to_string()), + input_tokens: usage + .and_then(|value| value.get("input_tokens")) + .and_then(Value::as_u64), + output_tokens: usage + .and_then(|value| value.get("output_tokens")) + .and_then(Value::as_u64), + reasoning_output_tokens: usage + .and_then(|value| value.get("output_tokens_details")) + .and_then(|value| value.get("reasoning_tokens")) + .and_then(Value::as_u64), + cached_input_tokens: usage + .and_then(|value| value.get("input_tokens_details")) + .and_then(|value| value.get("cached_tokens")) + .and_then(Value::as_u64), + }); + Ok(()) + } +} + +fn item_identity(item: &Value) -> Option { + item_string(item, "id").or_else(|| item_string(item, "call_id")) +} + +fn item_string(item: &Value, field: &str) -> Option { + item.get(field).and_then(Value::as_str).map(str::to_owned) +} + +fn parse_responses_sse(reader: impl Read) -> Result { + let mut reader = BufReader::new(reader); + let mut state = StreamState::new(); + let mut data = Vec::new(); + let mut total_bytes = 0_usize; + loop { + let mut line = Vec::new(); + let read = reader + .by_ref() + .take((MAX_SSE_FRAME_BYTES + 1) as u64) + .read_until(b'\n', &mut line) + .map_err(|_| "provider SSE read failed")?; + if read == 0 { + if !data.is_empty() { + state.accept_frame(&data)?; + } + break; + } + total_bytes = total_bytes + .checked_add(read) + .filter(|total| *total <= MAX_STREAM_BYTES) + .ok_or("provider stream exceeded its byte bound")?; + if line.len() > MAX_SSE_FRAME_BYTES { + return Err("provider SSE frame exceeded its byte bound"); + } + let line = line.strip_suffix(b"\n").unwrap_or(&line); + let line = line.strip_suffix(b"\r").unwrap_or(line); + if line.is_empty() { + if !data.is_empty() { + state.accept_frame(&data)?; + data.clear(); + if state.outcome.is_some() { + break; + } + } + } else if let Some(fragment) = line.strip_prefix(b"data:") { + let fragment = fragment.strip_prefix(b" ").unwrap_or(fragment); + if !data.is_empty() { + data.push(b'\n'); + } + data.extend_from_slice(fragment); + if data.len() > MAX_SSE_FRAME_BYTES { + return Err("provider SSE frame exceeded its byte bound"); + } + } + } + let outcome = state + .outcome + .ok_or("provider stream closed without a terminal event")?; + Ok(ParsedProviderStream { + events: state.events, + outcome, + }) +} + +fn failed_result( + invocation: &CodexTransportInvocation, + failure_kind: &str, + message: &str, +) -> CodexTransportResult { + CodexTransportResult::transported( + invocation, + Vec::new(), + CodexTransportReceipt { + schema_id: RECEIPT_SCHEMA_ID.to_string(), + request_id: invocation.request_id().to_string(), + caller_runtime_id: invocation.caller_runtime_id.clone(), + native_request_sha256: invocation.native_request_sha256, + provider_request_sha256: invocation.provider_request_sha256, + model: invocation.request.model.clone(), + transport: TRANSPORT_ID.to_string(), + outcome: CodexTransportOutcome::Failed { + failure_kind: failure_kind.to_string(), + message: message.to_string(), + }, + }, + ) +} + +struct CodexAppServerAuthority { + child: Child, + rpc: AppServerRpc, ChildStdin>, + codex_home: PathBuf, + app_server_user_agent: String, +} + +impl CodexAppServerAuthority { + fn spawn(config: CodexAppServerConfig) -> Result { + if config.executable_sha256 == [0; 32] { + return Err(CodexProviderBackendError::InvalidConfiguration( + "executable_sha256", + )); + } + let executable = canonical_file(&config.executable)?; + let observed_sha256 = sha256_file(&executable)?; + if observed_sha256 != config.executable_sha256 { + return Err(CodexProviderBackendError::ExecutableDigest); + } + let expected_home = canonical_directory(&config.codex_home)?; + + let mut command = Command::new(executable); + command + .arg("app-server") + .arg("--listen") + .arg("stdio://") + .current_dir(&expected_home) + .env("CODEX_HOME", &expected_home) + .env("CODEX_APP_SERVER_DISABLE_MANAGED_CONFIG", "1") + .env("RUST_LOG", "warn") + .stdin(Stdio::piped()) + .stdout(Stdio::piped()) + .stderr(Stdio::inherit()); + let mut child = command.spawn().map_err(CodexProviderBackendError::Spawn)?; + let stdin = match child.stdin.take() { + Some(stdin) => stdin, + None => { + terminate_child(&mut child); + return Err(CodexProviderBackendError::MissingChildPipe("stdin")); + } + }; + let stdout = match child.stdout.take() { + Some(stdout) => stdout, + None => { + terminate_child(&mut child); + return Err(CodexProviderBackendError::MissingChildPipe("stdout")); + } + }; + let mut rpc = AppServerRpc::new(BufReader::new(stdout), stdin); + let initialized = match rpc.initialize() { + Ok(initialized) => initialized, + Err(error) => { + terminate_child(&mut child); + return Err(error); + } + }; + let reported_home = match canonical_directory(Path::new(&initialized.codex_home)) { + Ok(home) => home, + Err(error) => { + terminate_child(&mut child); + return Err(error); + } + }; + if reported_home != expected_home { + terminate_child(&mut child); + return Err(CodexProviderBackendError::CodexHomeIdentity); + } + + Ok(Self { + child, + rpc, + codex_home: expected_home, + app_server_user_agent: initialized.user_agent, + }) + } + + fn read_credential( + &mut self, + refresh_token: bool, + ) -> Result { + let account = self.rpc.read_account(refresh_token)?; + let auth_bytes = std::fs::read(self.codex_home.join("auth.json")) + .map_err(CodexProviderBackendError::AuthFile)?; + parse_auth_file(&auth_bytes, account) + } +} + +impl Drop for CodexAppServerAuthority { + fn drop(&mut self) { + terminate_child(&mut self.child); + } +} + +struct AppServerRpc { + reader: R, + writer: W, + next_request_id: i64, +} + +impl AppServerRpc { + fn new(reader: R, writer: W) -> Self { + Self { + reader, + writer, + next_request_id: 0, + } + } + + fn initialize(&mut self) -> Result { + let result = self.request( + "initialize", + json!({ + "clientInfo": { + "name": "gamecult_codex_connector", + "title": "GameCult CodexConnector", + "version": env!("CARGO_PKG_VERSION") + } + }), + )?; + self.notify("initialized")?; + serde_json::from_value(result).map_err(|_| CodexProviderBackendError::AppServerProtocol) + } + + fn read_account( + &mut self, + refresh_token: bool, + ) -> Result { + let result = self.request("account/read", json!({ "refreshToken": refresh_token }))?; + let response: AppServerAccountResponse = serde_json::from_value(result) + .map_err(|_| CodexProviderBackendError::AppServerProtocol)?; + if !response.requires_openai_auth { + return Err(CodexProviderBackendError::AuthNotRequired); + } + response + .account + .ok_or(CodexProviderBackendError::CredentialUnavailable) + } + + fn request(&mut self, method: &str, params: Value) -> Result { + let request_id = self.next_request_id; + self.next_request_id = self + .next_request_id + .checked_add(1) + .ok_or(CodexProviderBackendError::AppServerProtocol)?; + write_json_line( + &mut self.writer, + &json!({ "id": request_id, "method": method, "params": params }), + )?; + + for _ in 0..MAX_INTERLEAVED_MESSAGES { + let message = read_json_line(&mut self.reader)?; + match message.get("id").and_then(Value::as_i64) { + Some(id) if id == request_id => { + if let Some(result) = message.get("result") { + return Ok(result.clone()); + } + if message.get("error").is_some() { + return Err(CodexProviderBackendError::AppServerRejected); + } + return Err(CodexProviderBackendError::AppServerProtocol); + } + Some(_) => return Err(CodexProviderBackendError::AppServerProtocol), + None if message.get("method").and_then(Value::as_str).is_some() => continue, + None => return Err(CodexProviderBackendError::AppServerProtocol), + } + } + Err(CodexProviderBackendError::AppServerProtocol) + } + + fn notify(&mut self, method: &str) -> Result<(), CodexProviderBackendError> { + write_json_line(&mut self.writer, &json!({ "method": method })) + } +} + +#[derive(Deserialize)] +#[serde(rename_all = "camelCase")] +struct AppServerInitialization { + user_agent: String, + codex_home: String, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Deserialize)] +#[serde(tag = "type")] +enum AppServerAccount { + #[serde(rename = "apiKey")] + ApiKey, + #[serde(rename = "chatgpt")] + Chatgpt, +} + +#[derive(Deserialize)] +#[serde(rename_all = "camelCase")] +struct AppServerAccountResponse { + account: Option, + requires_openai_auth: bool, +} + +struct CodexCredentialSnapshot { + mode: CodexAuthMode, + bearer: Zeroizing, + account_id: Option, + is_fedramp: bool, + auth_file_sha256: [u8; 32], +} + +impl CodexCredentialSnapshot { + fn redacted(&self, app_server_user_agent: &str) -> CodexAuthReadiness { + CodexAuthReadiness { + auth_mode: self.mode, + credential_loaded: !self.bearer.is_empty(), + fedramp_routing: self.is_fedramp, + app_server_user_agent: app_server_user_agent.to_string(), + } + } +} + +#[derive(Deserialize)] +struct AuthFile { + auth_mode: String, + #[serde(rename = "OPENAI_API_KEY")] + openai_api_key: Option, + tokens: Option, +} + +#[derive(Deserialize)] +struct AuthTokens { + id_token: String, + access_token: String, + account_id: Option, +} + +fn parse_auth_file( + bytes: &[u8], + account: AppServerAccount, +) -> Result { + let auth_file_sha256 = Sha256::digest(bytes).into(); + let auth: AuthFile = + serde_json::from_slice(bytes).map_err(|_| CodexProviderBackendError::AuthFileShape)?; + match (auth.auth_mode.as_str(), account) { + ("apikey", AppServerAccount::ApiKey) => { + let bearer = required_secret(auth.openai_api_key)?; + Ok(CodexCredentialSnapshot { + mode: CodexAuthMode::ApiKey, + bearer, + account_id: None, + is_fedramp: false, + auth_file_sha256, + }) + } + ("chatgpt", AppServerAccount::Chatgpt) => { + let tokens = auth + .tokens + .ok_or(CodexProviderBackendError::CredentialUnavailable)?; + let bearer = required_secret(Some(tokens.access_token))?; + let claims = parse_id_token_claims(&tokens.id_token)?; + let account_id = tokens + .account_id + .or(claims.account_id) + .filter(|value| !value.trim().is_empty() && value.trim() == value) + .ok_or(CodexProviderBackendError::AccountIdentity)?; + Ok(CodexCredentialSnapshot { + mode: CodexAuthMode::Chatgpt, + bearer, + account_id: Some(account_id), + is_fedramp: claims.is_fedramp, + auth_file_sha256, + }) + } + ("chatgptAuthTokens" | "agentIdentity", _) => { + Err(CodexProviderBackendError::UnsupportedAuthMode) + } + _ => Err(CodexProviderBackendError::AuthModeSubstitution), + } +} + +fn required_secret(value: Option) -> Result, CodexProviderBackendError> { + let value = value.ok_or(CodexProviderBackendError::CredentialUnavailable)?; + if value.trim().is_empty() || value.trim() != value { + return Err(CodexProviderBackendError::CredentialUnavailable); + } + Ok(Zeroizing::new(value)) +} + +struct IdTokenClaims { + account_id: Option, + is_fedramp: bool, +} + +fn parse_id_token_claims(token: &str) -> Result { + let mut parts = token.split('.'); + let (Some(header), Some(payload), Some(signature), None) = + (parts.next(), parts.next(), parts.next(), parts.next()) + else { + return Err(CodexProviderBackendError::IdTokenShape); + }; + if header.is_empty() || payload.is_empty() || signature.is_empty() { + return Err(CodexProviderBackendError::IdTokenShape); + } + let decoded = URL_SAFE_NO_PAD + .decode(payload) + .map_err(|_| CodexProviderBackendError::IdTokenShape)?; + let claims: Value = + serde_json::from_slice(&decoded).map_err(|_| CodexProviderBackendError::IdTokenShape)?; + let auth = claims + .get("https://api.openai.com/auth") + .and_then(Value::as_object); + Ok(IdTokenClaims { + account_id: auth + .and_then(|value| value.get("chatgpt_account_id")) + .and_then(Value::as_str) + .map(str::to_owned), + is_fedramp: auth + .and_then(|value| value.get("chatgpt_account_is_fedramp")) + .and_then(Value::as_bool) + .unwrap_or(false), + }) +} + +fn write_json_line( + writer: &mut impl Write, + value: &Value, +) -> Result<(), CodexProviderBackendError> { + serde_json::to_writer(&mut *writer, value) + .map_err(|_| CodexProviderBackendError::AppServerProtocol)?; + writer + .write_all(b"\n") + .and_then(|()| writer.flush()) + .map_err(CodexProviderBackendError::AppServerIo) +} + +fn read_json_line(reader: &mut impl BufRead) -> Result { + let mut line = Vec::new(); + let read = reader + .take((MAX_APP_SERVER_LINE_BYTES + 1) as u64) + .read_until(b'\n', &mut line) + .map_err(CodexProviderBackendError::AppServerIo)?; + if read == 0 || line.len() > MAX_APP_SERVER_LINE_BYTES { + return Err(CodexProviderBackendError::AppServerProtocol); + } + serde_json::from_slice(&line).map_err(|_| CodexProviderBackendError::AppServerProtocol) +} + +fn canonical_directory(path: &Path) -> Result { + let canonical = path + .canonicalize() + .map_err(CodexProviderBackendError::CodexHome)?; + if !canonical.is_dir() { + return Err(CodexProviderBackendError::InvalidConfiguration( + "codex_home", + )); + } + Ok(canonical) +} + +fn canonical_file(path: &Path) -> Result { + let canonical = path + .canonicalize() + .map_err(CodexProviderBackendError::Executable)?; + if !canonical.is_file() { + return Err(CodexProviderBackendError::InvalidConfiguration( + "executable", + )); + } + Ok(canonical) +} + +fn terminate_child(child: &mut Child) { + let _ = child.kill(); + let _ = child.wait(); +} + +fn sha256_file(path: &Path) -> Result<[u8; 32], CodexProviderBackendError> { + let mut file = File::open(path).map_err(CodexProviderBackendError::Executable)?; + let mut digest = Sha256::new(); + let mut buffer = [0_u8; 64 * 1024]; + loop { + let read = file + .read(&mut buffer) + .map_err(CodexProviderBackendError::Executable)?; + if read == 0 { + break; + } + digest.update(&buffer[..read]); + } + Ok(digest.finalize().into()) +} + +#[derive(Debug, Error)] +pub enum CodexProviderBackendError { + #[error("invalid connector configuration field {0}")] + InvalidConfiguration(&'static str), + #[error("failed to read pinned Codex executable")] + Executable(#[source] io::Error), + #[error("pinned Codex executable digest mismatch")] + ExecutableDigest, + #[error("failed to start pinned Codex app-server")] + Spawn(#[source] io::Error), + #[error("pinned Codex app-server did not expose {0}")] + MissingChildPipe(&'static str), + #[error("Codex home is unavailable")] + CodexHome(#[source] io::Error), + #[error("pinned Codex app-server reported a different Codex home")] + CodexHomeIdentity, + #[error("Codex app-server protocol failed")] + AppServerProtocol, + #[error("Codex app-server IO failed")] + AppServerIo(#[source] io::Error), + #[error("Codex app-server refused the authentication request")] + AppServerRejected, + #[error("Codex app-server reports that OpenAI authentication is disabled")] + AuthNotRequired, + #[error("Codex credential file is unavailable")] + AuthFile(#[source] io::Error), + #[error("Codex credential file has an unexpected shape")] + AuthFileShape, + #[error("Codex credential is unavailable")] + CredentialUnavailable, + #[error("Codex credential mode disagrees with app-server")] + AuthModeSubstitution, + #[error("Codex credential mode is not admitted")] + UnsupportedAuthMode, + #[error("ChatGPT account identity is unavailable")] + AccountIdentity, + #[error("ChatGPT ID token has an unexpected shape")] + IdTokenShape, + #[error("provider request could not be rendered")] + ProviderRequest, + #[error("Codex credential authority lock was poisoned")] + CredentialAuthorityPoisoned, +} + +#[cfg(test)] +mod tests { + use super::*; + use std::net::TcpListener; + use std::thread; + + fn jwt(payload: Value) -> String { + let payload = URL_SAFE_NO_PAD.encode(serde_json::to_vec(&payload).unwrap()); + format!("e30.{payload}.signature") + } + + #[test] + fn app_server_rpc_uses_only_initialize_and_account_read() { + let input = concat!( + "{\"id\":0,\"result\":{\"userAgent\":\"codex/1.2.3\",\"codexHome\":\"C:/codex\"}}\n", + "{\"method\":\"account/updated\",\"params\":{}}\n", + "{\"id\":1,\"result\":{\"account\":{\"type\":\"apiKey\"},\"requiresOpenaiAuth\":true}}\n" + ); + let reader = BufReader::new(io::Cursor::new(input.as_bytes())); + let mut written = Vec::new(); + let mut rpc = AppServerRpc::new(reader, &mut written); + + let initialized = rpc.initialize().unwrap(); + assert_eq!(initialized.user_agent, "codex/1.2.3"); + assert_eq!(rpc.read_account(false).unwrap(), AppServerAccount::ApiKey); + drop(rpc); + + let messages = String::from_utf8(written) + .unwrap() + .lines() + .map(|line| serde_json::from_str::(line).unwrap()) + .collect::>(); + assert_eq!(messages.len(), 3); + assert_eq!(messages[0]["method"], "initialize"); + assert_eq!(messages[1], json!({"method": "initialized"})); + assert_eq!( + messages[2], + json!({"id": 1, "method": "account/read", "params": {"refreshToken": false}}) + ); + assert!(messages.iter().all(|message| { + let text = message.to_string(); + !text.contains("prompt") && !text.contains("tool") && !text.contains("model") + })); + } + + #[test] + fn api_key_readiness_exposes_no_credential_identity() { + let bytes = br#"{ + "auth_mode":"apikey", + "OPENAI_API_KEY":"secret-api-key", + "tokens":null + }"#; + let credential = parse_auth_file(bytes, AppServerAccount::ApiKey).unwrap(); + let readiness = credential.redacted("codex/1.2.3"); + assert_eq!(readiness.auth_mode, CodexAuthMode::ApiKey); + assert!(readiness.credential_loaded); + assert!(!readiness.fedramp_routing); + assert_eq!(readiness.app_server_user_agent, "codex/1.2.3"); + assert!(!format!("{readiness:?}").contains("secret-api-key")); + } + + #[test] + fn chatgpt_readiness_derives_account_and_fedramp_without_exposing_tokens() { + let id_token = jwt(json!({ + "https://api.openai.com/auth": { + "chatgpt_account_id": "workspace-7", + "chatgpt_account_is_fedramp": true + } + })); + let bytes = serde_json::to_vec(&json!({ + "auth_mode": "chatgpt", + "OPENAI_API_KEY": null, + "tokens": { + "id_token": id_token, + "access_token": "secret-access-token", + "refresh_token": "secret-refresh-token", + "account_id": null + } + })) + .unwrap(); + let credential = parse_auth_file(&bytes, AppServerAccount::Chatgpt).unwrap(); + assert!(credential.is_fedramp); + let readiness = credential.redacted("codex/1.2.3"); + assert_eq!(readiness.auth_mode, CodexAuthMode::Chatgpt); + assert!(readiness.credential_loaded); + assert!(readiness.fedramp_routing); + let debug = format!("{readiness:?}"); + assert!(!debug.contains("secret-access-token")); + assert!(!debug.contains("secret-refresh-token")); + assert!(!debug.contains("workspace-7")); + } + + #[test] + fn app_server_and_auth_file_mode_must_match() { + let bytes = br#"{ + "auth_mode":"apikey", + "OPENAI_API_KEY":"secret-api-key", + "tokens":null + }"#; + assert!(matches!( + parse_auth_file(bytes, AppServerAccount::Chatgpt), + Err(CodexProviderBackendError::AuthModeSubstitution) + )); + } + + #[test] + fn responses_stream_preserves_text_tool_and_usage_consequences() { + let stream = concat!( + "data: {\"type\":\"response.output_text.delta\",\"delta\":\"answer\"}\n\n", + "data: {\"type\":\"response.output_item.added\",\"item\":{\"id\":\"fc-1\",\"type\":\"function_call\",\"call_id\":\"call-1\",\"name\":\"read_source\",\"arguments\":\"\"}}\n\n", + "data: {\"type\":\"response.function_call_arguments.delta\",\"item_id\":\"fc-1\",\"call_id\":\"call-1\",\"delta\":\"{\\\"path\\\":\"}\n\n", + "data: {\"type\":\"response.function_call_arguments.delta\",\"item_id\":\"fc-1\",\"call_id\":\"call-1\",\"delta\":\"\\\"README.md\\\"}\"}\n\n", + "data: {\"type\":\"response.output_item.done\",\"item\":{\"id\":\"fc-1\",\"type\":\"function_call\",\"call_id\":\"call-1\",\"name\":\"read_source\"}}\n\n", + "data: {\"type\":\"response.completed\",\"response\":{\"id\":\"resp-1\",\"usage\":{\"input_tokens\":11,\"input_tokens_details\":{\"cached_tokens\":3},\"output_tokens\":7,\"output_tokens_details\":{\"reasoning_tokens\":2}}}}\n\n" + ); + let parsed = parse_responses_sse(stream.as_bytes()).unwrap(); + assert_eq!( + parsed.events, + vec![ + CodexTransportEvent { + sequence: 0, + payload: CodexTransportEventPayload::TextDelta { + text: "answer".to_string() + } + }, + CodexTransportEvent { + sequence: 1, + payload: CodexTransportEventPayload::ToolCall { + call_id: "call-1".to_string(), + name: "read_source".to_string(), + arguments: "{\"path\":\"README.md\"}".to_string() + } + } + ] + ); + assert_eq!( + parsed.outcome, + CodexTransportOutcome::Completed { + provider_response_id: Some("resp-1".to_string()), + input_tokens: Some(11), + output_tokens: Some(7), + reasoning_output_tokens: Some(2), + cached_input_tokens: Some(3), + } + ); + } + + #[test] + fn responses_stream_refuses_terminality_with_an_unfinished_tool() { + let stream = concat!( + "data: {\"type\":\"response.output_item.added\",\"item\":{\"id\":\"fc-1\",\"type\":\"function_call\",\"call_id\":\"call-1\",\"name\":\"read_source\",\"arguments\":\"\"}}\n\n", + "data: {\"type\":\"response.completed\",\"response\":{\"id\":\"resp-1\"}}\n\n" + ); + assert_eq!( + parse_responses_sse(stream.as_bytes()).err(), + Some("provider completed with an unfinished tool call") + ); + } + + #[test] + fn provider_http_transmits_exact_body_and_required_identity_headers() { + let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + let address = listener.local_addr().unwrap(); + let server = thread::spawn(move || { + let (mut socket, _) = listener.accept().unwrap(); + socket + .set_read_timeout(Some(Duration::from_secs(2))) + .unwrap(); + let mut received = Vec::new(); + let mut buffer = [0_u8; 4096]; + loop { + let read = socket.read(&mut buffer).unwrap(); + assert!(read > 0); + received.extend_from_slice(&buffer[..read]); + let Some(headers_end) = received.windows(4).position(|part| part == b"\r\n\r\n") + else { + continue; + }; + let headers = String::from_utf8_lossy(&received[..headers_end]); + let content_length = headers + .lines() + .find_map(|line| { + let (name, value) = line.split_once(':')?; + name.eq_ignore_ascii_case("content-length") + .then(|| value.trim().parse::().unwrap()) + }) + .unwrap(); + if received.len() >= headers_end + 4 + content_length { + break; + } + } + let response_body = concat!( + "data: {\"type\":\"response.completed\",", + "\"response\":{\"id\":\"resp-local\"}}\n\n" + ); + write!( + socket, + "HTTP/1.1 200 OK\r\nContent-Type: text/event-stream\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{}", + response_body.len(), + response_body + ) + .unwrap(); + received + }); + + let mut request = crate::CodexProviderRequest::new( + "request-http", + "conversation-http", + "gpt-test", + "Return a result.", + ); + request.input.push(crate::CodexInputItem::UserText { + text: "projected state".to_string(), + }); + let invocation = + CodexTransportInvocation::new("caller-http", 2_000, [9; 32], request).unwrap(); + let expected_body = canonical_responses_body_bytes(&invocation.request).unwrap(); + let credential = CodexCredentialSnapshot { + mode: CodexAuthMode::ApiKey, + bearer: Zeroizing::new("secret-http-token".to_string()), + account_id: None, + is_fedramp: false, + auth_file_sha256: [7; 32], + }; + let agent: ureq::Agent = ureq::Agent::config_builder() + .max_redirects(0) + .build() + .into(); + let response = send_provider_request( + &agent, + &format!("http://{address}/v1/responses"), + &invocation, + &expected_body, + &credential, + "codex/test", + ) + .unwrap(); + let parsed = parse_responses_sse(response.into_body().into_reader()).unwrap(); + assert!(matches!( + parsed.outcome, + CodexTransportOutcome::Completed { + provider_response_id: Some(ref id), + .. + } if id == "resp-local" + )); + + let received = server.join().unwrap(); + let headers_end = received + .windows(4) + .position(|part| part == b"\r\n\r\n") + .unwrap(); + let headers = String::from_utf8_lossy(&received[..headers_end]).to_ascii_lowercase(); + assert!(headers.contains("authorization: bearer secret-http-token")); + assert!(headers.contains("originator: gamecult_codex_connector")); + assert!(headers.contains("session_id: conversation-http")); + assert!(headers.contains("x-client-request-id: request-http")); + assert_eq!(&received[headers_end + 4..], expected_body); + } +}