From f6494aa85c3cdd673146f2a69878944cc7fabe4e Mon Sep 17 00:00:00 2001 From: jif-oai Date: Tue, 11 Nov 2025 14:45:06 +0000 Subject: [PATCH] R1 --- codex-rs/Cargo.lock | 13 + codex-rs/Cargo.toml | 2 + codex-rs/api-client/Cargo.toml | 1 + codex-rs/api-client/src/aggregate.rs | 30 +- codex-rs/api-client/src/api.rs | 19 +- codex-rs/api-client/src/chat.rs | 149 +++------ codex-rs/api-client/src/client/fixtures.rs | 2 +- codex-rs/api-client/src/client/http.rs | 23 +- codex-rs/api-client/src/client/mod.rs | 6 - codex-rs/api-client/src/lib.rs | 19 +- codex-rs/api-client/src/payload/chat.rs | 306 ----------------- codex-rs/api-client/src/payload/mod.rs | 2 - codex-rs/api-client/src/payload/responses.rs | 125 ------- codex-rs/api-client/src/prompt.rs | 125 ------- codex-rs/api-client/src/responses.rs | 240 +++----------- codex-rs/api-client/src/routed_client.rs | 89 ++--- codex-rs/core/Cargo.toml | 1 + codex-rs/core/src/client.rs | 55 ++- codex-rs/core/src/lib.rs | 11 +- codex-rs/core/src/wire_payload.rs | 312 ++++++++++++++++++ codex-rs/provider-config/Cargo.toml | 18 + .../src/lib.rs} | 257 +++++++-------- 22 files changed, 703 insertions(+), 1102 deletions(-) delete mode 100644 codex-rs/api-client/src/payload/chat.rs delete mode 100644 codex-rs/api-client/src/payload/mod.rs delete mode 100644 codex-rs/api-client/src/payload/responses.rs delete mode 100644 codex-rs/api-client/src/prompt.rs create mode 100644 codex-rs/core/src/wire_payload.rs create mode 100644 codex-rs/provider-config/Cargo.toml rename codex-rs/{api-client/src/model_provider.rs => provider-config/src/lib.rs} (67%) diff --git a/codex-rs/Cargo.lock b/codex-rs/Cargo.lock index 2667bef650..8b40912993 100644 --- a/codex-rs/Cargo.lock +++ b/codex-rs/Cargo.lock @@ -839,6 +839,7 @@ dependencies = [ "codex-app-server-protocol", "codex-otel", "codex-protocol", + "codex-provider-config", "futures", "maplit", "regex-lite", @@ -1096,6 +1097,7 @@ dependencies = [ "codex-keyring-store", "codex-otel", "codex-protocol", + "codex-provider-config", "codex-rmcp-client", "codex-utils-pty", "codex-utils-readiness", @@ -1390,6 +1392,17 @@ dependencies = [ "uuid", ] +[[package]] +name = "codex-provider-config" +version = "0.0.0" +dependencies = [ + "codex-app-server-protocol", + "reqwest", + "serde", + "serde_json", + "thiserror 2.0.17", +] + [[package]] name = "codex-responses-api-proxy" version = "0.0.0" diff --git a/codex-rs/Cargo.toml b/codex-rs/Cargo.toml index 821eeeec9a..cfeb406ee7 100644 --- a/codex-rs/Cargo.toml +++ b/codex-rs/Cargo.toml @@ -26,6 +26,7 @@ members = [ "ollama", "process-hardening", "protocol", + "provider-config", "rmcp-client", "responses-api-proxy", "stdio-to-uds", @@ -76,6 +77,7 @@ codex-ollama = { path = "ollama" } codex-otel = { path = "otel" } codex-process-hardening = { path = "process-hardening" } codex-protocol = { path = "protocol" } +codex-provider-config = { path = "provider-config" } codex-responses-api-proxy = { path = "responses-api-proxy" } codex-rmcp-client = { path = "rmcp-client" } codex-stdio-to-uds = { path = "stdio-to-uds" } diff --git a/codex-rs/api-client/Cargo.toml b/codex-rs/api-client/Cargo.toml index 5518876656..a36d539207 100644 --- a/codex-rs/api-client/Cargo.toml +++ b/codex-rs/api-client/Cargo.toml @@ -9,6 +9,7 @@ bytes = { workspace = true } codex-app-server-protocol = { workspace = true } codex-otel = { workspace = true } codex-protocol = { path = "../protocol" } +codex-provider-config = { path = "../provider-config" } futures = { workspace = true, default-features = false, features = ["std"] } maplit = { workspace = true } regex-lite = { workspace = true } diff --git a/codex-rs/api-client/src/aggregate.rs b/codex-rs/api-client/src/aggregate.rs index e22fd47d92..080abfa414 100644 --- a/codex-rs/api-client/src/aggregate.rs +++ b/codex-rs/api-client/src/aggregate.rs @@ -69,12 +69,12 @@ where loop { match Pin::new(&mut self.inner).poll_next(cx) { - std::task::Poll::Pending => return Poll::Pending, - std::task::Poll::Ready(None) => return std::task::Poll::Ready(None), - std::task::Poll::Ready(Some(Err(err))) => { - return std::task::Poll::Ready(Some(Err(err))); + Poll::Pending => return Poll::Pending, + Poll::Ready(None) => return Poll::Ready(None), + Poll::Ready(Some(Err(err))) => { + return Poll::Ready(Some(Err(err))); } - std::task::Poll::Ready(Some(Ok(ResponseEvent::OutputItemDone(item)))) => { + Poll::Ready(Some(Ok(ResponseEvent::OutputItemDone(item)))) => { let is_assistant_message = matches!( &item, ResponseItem::Message { role, .. } if role == "assistant" @@ -106,22 +106,22 @@ where } } } else { - return std::task::Poll::Ready(Some(Ok(ResponseEvent::OutputItemDone( + return Poll::Ready(Some(Ok(ResponseEvent::OutputItemDone( item, )))); } } - std::task::Poll::Ready(Some(Ok(ResponseEvent::OutputItemAdded(item)))) => { + Poll::Ready(Some(Ok(ResponseEvent::OutputItemAdded(item)))) => { if !matches!( &item, ResponseItem::Message { role, .. } if role == "assistant" ) { - return std::task::Poll::Ready(Some(Ok(ResponseEvent::OutputItemAdded( + return Poll::Ready(Some(Ok(ResponseEvent::OutputItemAdded( item, )))); } } - std::task::Poll::Ready(Some(Ok(ResponseEvent::ReasoningContentDelta(delta)))) => { + Poll::Ready(Some(Ok(ResponseEvent::ReasoningContentDelta(delta)))) => { self.cumulative_reasoning.push_str(&delta); if matches!(self.mode, AggregateMode::Streaming) { let ev = @@ -129,13 +129,13 @@ where self.pending.push_back(ev); } } - std::task::Poll::Ready(Some(Ok(ResponseEvent::ReasoningSummaryDelta(delta)))) => { + Poll::Ready(Some(Ok(ResponseEvent::ReasoningSummaryDelta(delta)))) => { if matches!(self.mode, AggregateMode::Streaming) { let ev = ResponseEvent::ReasoningSummaryDelta(delta); self.pending.push_back(ev); } } - std::task::Poll::Ready(Some(Ok(ResponseEvent::Completed { + Poll::Ready(Some(Ok(ResponseEvent::Completed { response_id, token_usage, }))) => { @@ -155,16 +155,16 @@ where self.pending.push_back(assistant_event); self.pending.push_back(completion_event); } else { - return std::task::Poll::Ready(Some(Ok(assistant_event))); + return Poll::Ready(Some(Ok(assistant_event))); } } - std::task::Poll::Ready(Some(Ok(ev))) => { - return std::task::Poll::Ready(Some(Ok(ev))); + Poll::Ready(Some(Ok(ev))) => { + return Poll::Ready(Some(Ok(ev))); } } if let Some(ev) = self.pending.pop_front() { - return std::task::Poll::Ready(Some(Ok(ev))); + return Poll::Ready(Some(Ok(ev))); } } } diff --git a/codex-rs/api-client/src/api.rs b/codex-rs/api-client/src/api.rs index c8349bdf2a..048c357e3c 100644 --- a/codex-rs/api-client/src/api.rs +++ b/codex-rs/api-client/src/api.rs @@ -1,21 +1,20 @@ use async_trait::async_trait; use crate::error::Result; -use crate::prompt::Prompt; use crate::stream::ResponseStream; +use codex_protocol::protocol::SessionSource; +use serde_json::Value; #[async_trait] -pub trait ApiClient: Sized { +pub trait PayloadClient: Sized { type Config; - /// Construct a new client instance from the provided configuration. - /// - /// This is synchronous to avoid forcing callers to `await` when no async - /// work is needed during construction. If an implementation needs async - /// initialization, prefer doing it inside `stream` or provide an explicit - /// async initializer on the concrete type. fn new(config: Self::Config) -> Result; - /// Start a streaming request for the given prompt. - async fn stream(&self, prompt: &Prompt) -> Result; + /// Start a streaming request for a pre-built wire JSON payload. + async fn stream_payload( + &self, + payload_json: &Value, + session_source: Option<&SessionSource>, + ) -> Result; } diff --git a/codex-rs/api-client/src/chat.rs b/codex-rs/api-client/src/chat.rs index 36bf7adfd1..20de2b7953 100644 --- a/codex-rs/api-client/src/chat.rs +++ b/codex-rs/api-client/src/chat.rs @@ -1,4 +1,3 @@ -use std::time::Duration; use async_trait::async_trait; use codex_otel::otel_event_manager::OtelEventManager; @@ -6,15 +5,12 @@ use codex_protocol::protocol::SessionSource; use futures::TryStreamExt; use tokio::sync::mpsc; -use crate::api::ApiClient; -use crate::client::PayloadBuilder; -use crate::common::backoff; +use crate::api::PayloadClient; use crate::error::Error; use crate::error::Result; -use crate::model_provider::ModelProviderInfo; -use crate::prompt::Prompt; use crate::stream::ResponseEvent; use crate::stream::ResponseStream; +use codex_provider_config::ModelProviderInfo; #[derive(Clone)] /// Configuration for the Chat Completions client (OpenAI-compatible `/v1/chat/completions`). @@ -37,103 +33,66 @@ pub struct ChatCompletionsApiClient { config: ChatCompletionsApiClientConfig, } +// prompt-based API removed; use PayloadClient::stream_payload instead + +// prompt-based API removed + #[async_trait] -impl ApiClient for ChatCompletionsApiClient { +impl PayloadClient for ChatCompletionsApiClient { type Config = ChatCompletionsApiClientConfig; fn new(config: Self::Config) -> Result { Ok(Self { config }) } - async fn stream(&self, prompt: &Prompt) -> Result { - Self::validate_prompt(prompt)?; - - let payload = crate::payload::chat::ChatPayloadBuilder::new(self.config.model.clone()) - .build(prompt)?; - let (tx_event, rx_event) = mpsc::channel::>(1600); - - let mut attempt: i64 = 0; - let max_retries = self.config.provider.request_max_retries(); - - loop { - attempt += 1; - - let req_builder = crate::client::http::build_request( - &self.config.http_client, - &self.config.provider, - &None, - Some(&self.config.session_source), - &[], - ) - .await?; - - let res = self - .config - .otel_event_manager - .log_request(attempt as u64, || { - req_builder - .header(reqwest::header::ACCEPT, "text/event-stream") - .json(&payload) - .send() - }) - .await; - - match res { - Ok(resp) if resp.status().is_success() => { - let stream = resp - .bytes_stream() - .map_err(|err| Error::ResponseStreamFailed { - source: err, - request_id: None, - }); - let idle_timeout = self.config.provider.stream_idle_timeout(); - let otel = self.config.otel_event_manager.clone(); - tokio::spawn(crate::client::sse::process_sse( - stream, - tx_event.clone(), - idle_timeout, - otel, - crate::decode::chat::ChatSseDecoder::new(), - )); - - return Ok(ResponseStream { rx_event }); - } - Ok(resp) => { - if attempt >= max_retries { - let status = resp.status(); - let body = resp - .text() - .await - .unwrap_or_else(|_| "".to_string()); - return Err(Error::UnexpectedStatus { status, body }); - } - - let retry_after = resp - .headers() - .get(reqwest::header::RETRY_AFTER) - .and_then(|v| v.to_str().ok()) - .and_then(|s| s.parse::().ok()) - .map(|secs| Duration::from_secs(if secs < 0 { 0 } else { secs as u64 })); - tokio::time::sleep(retry_after.unwrap_or_else(|| backoff(attempt))).await; - } - Err(error) => { - if attempt >= max_retries { - return Err(Error::Http(error)); - } - tokio::time::sleep(backoff(attempt)).await; - } - } - } - } -} - -impl ChatCompletionsApiClient { - fn validate_prompt(prompt: &Prompt) -> Result<()> { - if prompt.output_schema.is_some() { - return Err(Error::UnsupportedOperation( - "output_schema is not supported for Chat Completions API".to_string(), + async fn stream_payload( + &self, + payload_json: &serde_json::Value, + session_source: Option<&codex_protocol::protocol::SessionSource>, + ) -> Result { + if self.config.provider.wire_api != codex_provider_config::WireApi::Chat { + return Err(crate::error::Error::UnsupportedOperation( + "ChatCompletionsApiClient requires a Chat provider".to_string(), )); } - Ok(()) + + let auth = crate::client::http::resolve_auth(&None).await; + let mut req_builder = crate::client::http::build_request( + &self.config.http_client, + &self.config.provider, + &auth, + session_source, + &[], + ) + .await?; + + req_builder = req_builder + .header(reqwest::header::ACCEPT, "text/event-stream") + .json(payload_json); + + let res = self + .config + .otel_event_manager + .log_request(0, || req_builder.send()) + .await?; + + let (tx_event, rx_event) = mpsc::channel::>(1600); + let stream = res + .bytes_stream() + .map_err(|err| Error::ResponseStreamFailed { + source: err, + request_id: None, + }); + let idle_timeout = self.config.provider.stream_idle_timeout(); + let otel = self.config.otel_event_manager.clone(); + tokio::spawn(crate::client::sse::process_sse( + stream, + tx_event, + idle_timeout, + otel, + crate::decode::chat::ChatSseDecoder::new(), + )); + + Ok(crate::stream::EventStream::from_receiver(rx_event)) } } diff --git a/codex-rs/api-client/src/client/fixtures.rs b/codex-rs/api-client/src/client/fixtures.rs index 08ab9e0ac4..d638ab3fbd 100644 --- a/codex-rs/api-client/src/client/fixtures.rs +++ b/codex-rs/api-client/src/client/fixtures.rs @@ -8,9 +8,9 @@ use tokio_util::io::ReaderStream; use crate::error::Error; use crate::error::Result; -use crate::model_provider::ModelProviderInfo; use crate::stream::ResponseEvent; use crate::stream::ResponseStream; +use codex_provider_config::ModelProviderInfo; pub async fn stream_from_fixture( path: impl AsRef, diff --git a/codex-rs/api-client/src/client/http.rs b/codex-rs/api-client/src/client/http.rs index e79fc4df1b..b9c9281c47 100644 --- a/codex-rs/api-client/src/client/http.rs +++ b/codex-rs/api-client/src/client/http.rs @@ -6,7 +6,7 @@ use crate::auth::AuthContext; use crate::auth::AuthProvider; use crate::common::apply_subagent_header; use crate::error::Result; -use crate::model_provider::ModelProviderInfo; +use codex_provider_config::ModelProviderInfo; /// Build a request builder with provider/auth/session headers applied. pub async fn build_request( @@ -16,7 +16,26 @@ pub async fn build_request( session_source: Option<&SessionSource>, extra_headers: &[(&str, String)], ) -> Result { - let mut builder = provider.create_request_builder(http_client, auth).await?; + let mut builder = provider + .create_request_builder( + http_client, + &auth.as_ref().map(|a| codex_provider_config::AuthContext { + mode: a.mode, + bearer_token: a.bearer_token.clone(), + account_id: a.account_id.clone(), + }), + ) + .await + .map_err(|e| crate::error::Error::MissingEnvVar { + var: match e { + codex_provider_config::Error::MissingEnvVar { ref var, .. } => var.clone(), + }, + instructions: match e { + codex_provider_config::Error::MissingEnvVar { + ref instructions, .. + } => instructions.clone(), + }, + })?; builder = apply_subagent_header(builder, session_source); for (name, value) in extra_headers { builder = builder.header(*name, value); diff --git a/codex-rs/api-client/src/client/mod.rs b/codex-rs/api-client/src/client/mod.rs index 80ac1273dd..e7ebea0915 100644 --- a/codex-rs/api-client/src/client/mod.rs +++ b/codex-rs/api-client/src/client/mod.rs @@ -3,7 +3,6 @@ use codex_otel::otel_event_manager::OtelEventManager; use tokio::sync::mpsc; use crate::error::Result; -use crate::prompt::Prompt; use crate::stream::ResponseEvent; pub mod fixtures; @@ -11,11 +10,6 @@ pub mod http; pub mod rate_limits; pub mod sse; -/// Builds provider-specific JSON payloads from a Prompt. -pub trait PayloadBuilder { - fn build(&self, prompt: &Prompt) -> Result; -} - /// Decodes framed SSE JSON into ResponseEvent(s). /// Implementations may keep state across frames (e.g., Chat function-call state). #[async_trait] diff --git a/codex-rs/api-client/src/lib.rs b/codex-rs/api-client/src/lib.rs index 99374b15f2..d3a78611bd 100644 --- a/codex-rs/api-client/src/lib.rs +++ b/codex-rs/api-client/src/lib.rs @@ -6,15 +6,12 @@ mod client; mod common; mod decode; pub mod error; -pub mod model_provider; -mod payload; -pub mod prompt; +// payload building lives in codex-core now pub mod responses; pub mod routed_client; pub mod stream; pub use crate::aggregate::AggregateStreamExt; -pub use crate::api::ApiClient; pub use crate::auth::AuthContext; pub use crate::auth::AuthProvider; pub use crate::chat::ChatCompletionsApiClient; @@ -22,14 +19,6 @@ pub use crate::chat::ChatCompletionsApiClientConfig; pub use crate::client::fixtures::stream_from_fixture; pub use crate::error::Error; pub use crate::error::Result; -pub use crate::model_provider::BUILT_IN_OSS_MODEL_PROVIDER_ID; -pub use crate::model_provider::ModelProviderInfo; -pub use crate::model_provider::WireApi; -pub use crate::model_provider::built_in_model_providers; -pub use crate::model_provider::create_oss_provider; -pub use crate::model_provider::create_oss_provider_with_base_url; -pub use crate::prompt::Prompt; -pub use crate::prompt::PromptBuilder; pub use crate::responses::ResponsesApiClient; pub use crate::responses::ResponsesApiClientConfig; pub use crate::routed_client::RoutedApiClient; @@ -41,3 +30,9 @@ pub use crate::stream::ResponseStream; pub use crate::stream::TextControls; pub use crate::stream::TextFormat; pub use crate::stream::TextFormatType; +pub use codex_provider_config::BUILT_IN_OSS_MODEL_PROVIDER_ID; +pub use codex_provider_config::ModelProviderInfo; +pub use codex_provider_config::WireApi; +pub use codex_provider_config::built_in_model_providers; +pub use codex_provider_config::create_oss_provider; +pub use codex_provider_config::create_oss_provider_with_base_url; diff --git a/codex-rs/api-client/src/payload/chat.rs b/codex-rs/api-client/src/payload/chat.rs deleted file mode 100644 index 6f143847f0..0000000000 --- a/codex-rs/api-client/src/payload/chat.rs +++ /dev/null @@ -1,306 +0,0 @@ -use serde_json::Value; -use serde_json::json; -use std::collections::HashMap; - -use crate::client::PayloadBuilder; -use crate::error::Result; -use crate::prompt::Prompt; - -use codex_protocol::models::ContentItem; -use codex_protocol::models::FunctionCallOutputContentItem; -use codex_protocol::models::ReasoningItemContent; -use codex_protocol::models::ResponseItem; - -pub struct ChatPayloadBuilder { - model: String, -} - -impl ChatPayloadBuilder { - pub fn new(model: String) -> Self { - Self { model } - } -} - -impl PayloadBuilder for ChatPayloadBuilder { - fn build(&self, prompt: &Prompt) -> Result { - let mut messages = Vec::::new(); - messages.push(json!({ "role": "system", "content": prompt.instructions })); - - let mut reasoning_by_anchor_index: HashMap = HashMap::new(); - - let mut last_emitted_role: Option<&str> = None; - for item in &prompt.input { - match item { - ResponseItem::Message { role, .. } => last_emitted_role = Some(role.as_str()), - ResponseItem::FunctionCall { .. } | ResponseItem::LocalShellCall { .. } => { - last_emitted_role = Some("assistant"); - } - ResponseItem::FunctionCallOutput { .. } => last_emitted_role = Some("tool"), - ResponseItem::Reasoning { .. } - | ResponseItem::Other - | ResponseItem::CustomToolCall { .. } - | ResponseItem::CustomToolCallOutput { .. } - | ResponseItem::WebSearchCall { .. } - | ResponseItem::GhostSnapshot { .. } => {} - } - } - - let mut last_user_index: Option = None; - for (idx, item) in prompt.input.iter().enumerate() { - if let ResponseItem::Message { role, .. } = item - && role == "user" - { - last_user_index = Some(idx); - } - } - - if !matches!(last_emitted_role, Some("user")) { - for (idx, item) in prompt.input.iter().enumerate() { - if let Some(u_idx) = last_user_index - && idx <= u_idx - { - continue; - } - - if let ResponseItem::Reasoning { - content: Some(items), - .. - } = item - { - let mut text = String::new(); - for entry in items { - match entry { - ReasoningItemContent::ReasoningText { text: segment } - | ReasoningItemContent::Text { text: segment } => { - text.push_str(segment); - } - } - } - if text.trim().is_empty() { - continue; - } - - let mut attached = false; - if idx > 0 - && let ResponseItem::Message { role, .. } = &prompt.input[idx - 1] - && role == "assistant" - { - reasoning_by_anchor_index - .entry(idx - 1) - .and_modify(|val| val.push_str(&text)) - .or_insert(text.clone()); - attached = true; - } - - if !attached && idx + 1 < prompt.input.len() { - match &prompt.input[idx + 1] { - ResponseItem::FunctionCall { .. } - | ResponseItem::LocalShellCall { .. } => { - reasoning_by_anchor_index - .entry(idx + 1) - .and_modify(|val| val.push_str(&text)) - .or_insert(text.clone()); - } - ResponseItem::Message { role, .. } if role == "assistant" => { - reasoning_by_anchor_index - .entry(idx + 1) - .and_modify(|val| val.push_str(&text)) - .or_insert(text.clone()); - } - _ => {} - } - } - } - } - } - - let mut last_assistant_text: Option = None; - - for (idx, item) in prompt.input.iter().enumerate() { - match item { - ResponseItem::Message { role, content, .. } => { - let mut text = String::new(); - let mut items: Vec = Vec::new(); - let mut saw_image = false; - - for c in content { - match c { - ContentItem::InputText { text: t } - | ContentItem::OutputText { text: t } => { - text.push_str(t); - items.push(json!({"type":"text","text": t})); - } - ContentItem::InputImage { image_url } => { - saw_image = true; - items.push( - json!({"type":"image_url","image_url": {"url": image_url}}), - ); - } - } - } - - if role == "assistant" { - if let Some(prev) = &last_assistant_text - && prev == &text - { - continue; - } - last_assistant_text = Some(text.clone()); - } - - let content_value = if role == "assistant" { - json!(text) - } else if saw_image { - json!(items) - } else { - json!(text) - }; - - let mut message = json!({ - "role": role, - "content": content_value, - }); - - if let Some(reasoning) = reasoning_by_anchor_index.get(&idx) - && let Some(obj) = message.as_object_mut() - { - obj.insert("reasoning".to_string(), json!({"text": reasoning})); - } - - messages.push(message); - } - ResponseItem::FunctionCall { - name, - arguments, - call_id, - .. - } => { - messages.push(json!({ - "role": "assistant", - "tool_calls": [{ - "id": call_id, - "type": "function", - "function": { - "name": name, - "arguments": arguments, - }, - }], - })); - } - ResponseItem::FunctionCallOutput { call_id, output } => { - let content_value = if let Some(items) = &output.content_items { - let mapped: Vec = items - .iter() - .map(|item| match item { - FunctionCallOutputContentItem::InputText { text } => { - json!({"type":"text","text": text}) - } - FunctionCallOutputContentItem::InputImage { image_url } => { - json!({"type":"image_url","image_url": {"url": image_url}}) - } - }) - .collect(); - json!(mapped) - } else { - json!(output.content) - }; - messages.push(json!({ - "role": "tool", - "tool_call_id": call_id, - "content": content_value, - })); - } - ResponseItem::LocalShellCall { - id, - call_id, - action, - .. - } => { - let tool_id = call_id - .clone() - .filter(|value| !value.is_empty()) - .or_else(|| id.clone()) - .unwrap_or_default(); - messages.push(json!({ - "role": "assistant", - "tool_calls": [{ - "id": tool_id, - "type": "function", - "function": { - "name": "shell", - "arguments": serde_json::to_string(action).unwrap_or_default(), - }, - }], - })); - } - ResponseItem::CustomToolCall { - call_id, - name, - input, - .. - } => { - messages.push(json!({ - "role": "assistant", - "tool_calls": [{ - "id": call_id.clone(), - "type": "function", - "function": { - "name": name, - "arguments": input, - }, - }], - })); - } - ResponseItem::CustomToolCallOutput { call_id, output } => { - messages.push(json!({ - "role": "tool", - "tool_call_id": call_id, - "content": output, - })); - } - ResponseItem::WebSearchCall { .. } - | ResponseItem::Reasoning { .. } - | ResponseItem::Other - | ResponseItem::GhostSnapshot { .. } => {} - } - } - - let tools_json = create_tools_json_for_chat_completions_api(&prompt.tools)?; - let payload = json!({ - "model": self.model, - "messages": messages, - "stream": true, - "tools": tools_json, - }); - Ok(payload) - } -} - -fn create_tools_json_for_chat_completions_api( - tools: &[serde_json::Value], -) -> Result> { - let tools_json = tools - .iter() - .filter_map(|tool| { - if tool.get("type") != Some(&serde_json::Value::String("function".to_string())) { - return None; - } - - let function_value = if let Some(function) = tool.get("function") { - function.clone() - } else if let Some(map) = tool.as_object() { - let mut function = map.clone(); - function.remove("type"); - Value::Object(function) - } else { - return None; - }; - - Some(json!({ - "type": "function", - "function": function_value, - })) - }) - .collect::>(); - Ok(tools_json) -} diff --git a/codex-rs/api-client/src/payload/mod.rs b/codex-rs/api-client/src/payload/mod.rs deleted file mode 100644 index 74e360ea71..0000000000 --- a/codex-rs/api-client/src/payload/mod.rs +++ /dev/null @@ -1,2 +0,0 @@ -pub mod chat; -pub mod responses; diff --git a/codex-rs/api-client/src/payload/responses.rs b/codex-rs/api-client/src/payload/responses.rs deleted file mode 100644 index 7ace3aad72..0000000000 --- a/codex-rs/api-client/src/payload/responses.rs +++ /dev/null @@ -1,125 +0,0 @@ -use serde_json::Value; -use serde_json::json; - -use crate::client::PayloadBuilder; -use crate::error::Result; -use crate::prompt::Prompt; - -use codex_protocol::ConversationId; -use codex_protocol::models::ResponseItem; - -pub struct ResponsesPayloadBuilder { - model: String, - conversation_id: ConversationId, - azure_workaround: bool, -} - -impl ResponsesPayloadBuilder { - pub fn new(model: String, conversation_id: ConversationId, azure_workaround: bool) -> Self { - Self { - model, - conversation_id, - azure_workaround, - } - } -} - -impl PayloadBuilder for ResponsesPayloadBuilder { - fn build(&self, prompt: &Prompt) -> Result { - let azure = self.azure_workaround; - let mut payload = json!({ - "model": self.model, - "instructions": prompt.instructions, - "input": prompt.input, - "tools": prompt.tools, - "tool_choice": "auto", - "parallel_tool_calls": prompt.parallel_tool_calls, - "store": azure, - "stream": true, - "prompt_cache_key": prompt - .prompt_cache_key - .clone() - .unwrap_or_else(|| self.conversation_id.to_string()), - }); - - if let Some(reasoning) = prompt.reasoning.as_ref() - && let Some(obj) = payload.as_object_mut() - { - obj.insert("reasoning".to_string(), serde_json::to_value(reasoning)?); - } - - if let Some(text) = prompt.text_controls.as_ref() - && let Some(obj) = payload.as_object_mut() - { - obj.insert("text".to_string(), serde_json::to_value(text)?); - } - - let include = if prompt.reasoning.is_some() { - vec!["reasoning.encrypted_content".to_string()] - } else { - Vec::new() - }; - if let Some(obj) = payload.as_object_mut() { - obj.insert( - "include".to_string(), - Value::Array(include.into_iter().map(Value::String).collect()), - ); - } - - // Azure Responses requires ids attached to input items - if azure - && let Some(input_value) = payload.get_mut("input") - && let Some(array) = input_value.as_array_mut() - { - attach_item_ids_array(array, &prompt.input); - } - - Ok(payload) - } -} - -fn attach_item_ids_array(json_array: &mut [Value], prompt_input: &[ResponseItem]) { - for (json_item, item) in json_array.iter_mut().zip(prompt_input.iter()) { - let Some(obj) = json_item.as_object_mut() else { - continue; - }; - - let mut set_id_if_absent = |id: &str| match obj.get("id") { - Some(Value::String(s)) if !s.is_empty() => {} - Some(Value::Null) | None => { - obj.insert("id".to_string(), Value::String(id.to_string())); - } - _ => {} - }; - - match item { - ResponseItem::Reasoning { id, .. } => set_id_if_absent(id), - ResponseItem::Message { id, .. } => { - if let Some(id) = id.as_ref() { - set_id_if_absent(id); - } - } - ResponseItem::WebSearchCall { id, .. } => { - if let Some(id) = id.as_ref() { - set_id_if_absent(id); - } - } - ResponseItem::FunctionCall { id, .. } => { - if let Some(id) = id.as_ref() { - set_id_if_absent(id); - } - } - ResponseItem::LocalShellCall { id, .. } => { - if let Some(id) = id.as_ref() { - set_id_if_absent(id); - } - } - ResponseItem::CustomToolCall { id, .. } => { - if let Some(id) = id.as_ref() { - set_id_if_absent(id); - } - } - _ => {} - } - } -} diff --git a/codex-rs/api-client/src/prompt.rs b/codex-rs/api-client/src/prompt.rs deleted file mode 100644 index b708d8abef..0000000000 --- a/codex-rs/api-client/src/prompt.rs +++ /dev/null @@ -1,125 +0,0 @@ -use codex_protocol::models::ResponseItem; -use codex_protocol::protocol::SessionSource; -use serde_json::Value; - -use crate::Reasoning; -use crate::TextControls; - -#[derive(Debug, Clone, Default)] -pub struct Prompt { - pub instructions: String, - pub input: Vec, - pub tools: Vec, - pub parallel_tool_calls: bool, - pub output_schema: Option, - pub reasoning: Option, - pub text_controls: Option, - pub prompt_cache_key: Option, - pub session_source: Option, -} - -impl Prompt { - #[allow(clippy::too_many_arguments)] - pub fn new( - instructions: String, - input: Vec, - tools: Vec, - parallel_tool_calls: bool, - output_schema: Option, - reasoning: Option, - text_controls: Option, - prompt_cache_key: Option, - session_source: Option, - ) -> Self { - Self { - instructions, - input, - tools, - parallel_tool_calls, - output_schema, - reasoning, - text_controls, - prompt_cache_key, - session_source, - } - } - - /// Start building a `Prompt` with a fluent API. - pub fn builder() -> PromptBuilder { - PromptBuilder::default() - } -} - -#[derive(Debug, Clone, Default)] -pub struct PromptBuilder { - instructions: String, - input: Vec, - tools: Vec, - parallel_tool_calls: bool, - output_schema: Option, - reasoning: Option, - text_controls: Option, - prompt_cache_key: Option, - session_source: Option, -} - -impl PromptBuilder { - pub fn instructions(mut self, instructions: impl Into) -> Self { - self.instructions = instructions.into(); - self - } - - pub fn input(mut self, input: Vec) -> Self { - self.input = input; - self - } - - pub fn tools(mut self, tools: Vec) -> Self { - self.tools = tools; - self - } - - pub fn parallel_tool_calls(mut self, enabled: bool) -> Self { - self.parallel_tool_calls = enabled; - self - } - - pub fn output_schema(mut self, schema: Option) -> Self { - self.output_schema = schema; - self - } - - pub fn reasoning(mut self, reasoning: Option) -> Self { - self.reasoning = reasoning; - self - } - - pub fn text_controls(mut self, text_controls: Option) -> Self { - self.text_controls = text_controls; - self - } - - pub fn prompt_cache_key(mut self, key: Option) -> Self { - self.prompt_cache_key = key; - self - } - - pub fn session_source(mut self, session: Option) -> Self { - self.session_source = session; - self - } - - pub fn build(self) -> Prompt { - Prompt { - instructions: self.instructions, - input: self.input, - tools: self.tools, - parallel_tool_calls: self.parallel_tool_calls, - output_schema: self.output_schema, - reasoning: self.reasoning, - text_controls: self.text_controls, - prompt_cache_key: self.prompt_cache_key, - session_source: self.session_source, - } - } -} diff --git a/codex-rs/api-client/src/responses.rs b/codex-rs/api-client/src/responses.rs index c957526b0c..f62ad26baa 100644 --- a/codex-rs/api-client/src/responses.rs +++ b/codex-rs/api-client/src/responses.rs @@ -12,17 +12,15 @@ use tokio::sync::mpsc; use tracing::debug; use tracing::trace; -use crate::api::ApiClient; +use crate::api::PayloadClient; use crate::auth::AuthProvider; -use crate::client::PayloadBuilder; use crate::common::backoff; use crate::decode::responses::ErrorResponse; use crate::error::Error; use crate::error::Result; -use crate::model_provider::ModelProviderInfo; -use crate::prompt::Prompt; use crate::stream::ResponseEvent; use crate::stream::ResponseStream; +use codex_provider_config::ModelProviderInfo; #[derive(Clone)] /// Configuration for the OpenAI Responses API client (`/v1/responses`). @@ -47,62 +45,39 @@ pub struct ResponsesApiClient { config: ResponsesApiClientConfig, } + #[async_trait] -impl ApiClient for ResponsesApiClient { +impl PayloadClient for ResponsesApiClient { type Config = ResponsesApiClientConfig; fn new(config: Self::Config) -> Result { Ok(Self { config }) } - async fn stream(&self, prompt: &Prompt) -> Result { - if self.config.provider.wire_api != crate::model_provider::WireApi::Responses { + async fn stream_payload( + &self, + payload_json: &Value, + session_source: Option<&codex_protocol::protocol::SessionSource>, + ) -> Result { + if self.config.provider.wire_api != codex_provider_config::WireApi::Responses { return Err(Error::UnsupportedOperation( "ResponsesApiClient requires a Responses provider".to_string(), )); } - let payload_json = crate::payload::responses::ResponsesPayloadBuilder::new( - self.config.model.clone(), - self.config.conversation_id, - self.config.provider.is_azure_responses_endpoint(), - ) - .build(prompt)?; - - let max_attempts = self.config.provider.request_max_retries(); - for attempt in 0..=max_attempts { - match self - .attempt_stream_responses(attempt, prompt, &payload_json) - .await - { - Ok(stream) => return Ok(stream), - Err(StreamAttemptError::Fatal(err)) => return Err(err), - Err(retryable) => { - if attempt == max_attempts { - return Err(retryable.into_error()); - } - - tokio::time::sleep(retryable.delay(attempt)).await; - } - } - } - - unreachable!("attempt_stream_responses should always return"); - } -} - -impl ResponsesApiClient { - async fn attempt_stream_responses( - &self, - attempt: i64, - prompt: &Prompt, - payload_json: &Value, - ) -> std::result::Result { let auth = crate::client::http::resolve_auth(&self.config.auth_provider).await; trace!( "POST to {}: {:?}", - self.config.provider.get_full_url(auth.as_ref()), + self.config.provider.get_full_url( + auth.as_ref() + .map(|a| codex_provider_config::AuthContext { + mode: a.mode, + bearer_token: a.bearer_token.clone(), + account_id: a.account_id.clone(), + }) + .as_ref() + ), serde_json::to_string(payload_json) .unwrap_or_else(|_| "".to_string()) ); @@ -115,11 +90,10 @@ impl ResponsesApiClient { &self.config.http_client, &self.config.provider, &auth, - prompt.session_source.as_ref(), + session_source, &extra_headers, ) - .await - .map_err(StreamAttemptError::Fatal)?; + .await?; req_builder = req_builder .header(reqwest::header::ACCEPT, "text/event-stream") @@ -135,152 +109,40 @@ impl ResponsesApiClient { let res = self .config .otel_event_manager - .log_request(attempt as u64, || req_builder.send()) - .await; + .log_request(0, || req_builder.send()) + .await + .map_err(|source| Error::ResponseStreamFailed { + source, + request_id: None, + })?; - let mut request_id = None; - if let Ok(resp) = &res { - request_id = resp - .headers() - .get("cf-ray") - .and_then(|v| v.to_str().ok()) - .map(std::string::ToString::to_string); + let (tx_event, rx_event) = mpsc::channel::>(1600); + if let Some(snapshot) = crate::client::rate_limits::parse_rate_limit_snapshot(res.headers()) + && tx_event + .send(Ok(ResponseEvent::RateLimits(snapshot))) + .await + .is_err() + { + debug!("receiver dropped rate limit snapshot event"); } - match res { - Ok(resp) if resp.status().is_success() => { - let (tx_event, rx_event) = mpsc::channel::>(1600); + let stream = res + .bytes_stream() + .map_err(|err| Error::ResponseStreamFailed { + source: err, + request_id: None, + }); + let idle_timeout = self.config.provider.stream_idle_timeout(); + let otel = self.config.otel_event_manager.clone(); + tokio::spawn(crate::client::sse::process_sse( + stream, + tx_event, + idle_timeout, + otel, + crate::decode::responses::ResponsesSseDecoder, + )); - if let Some(snapshot) = - crate::client::rate_limits::parse_rate_limit_snapshot(resp.headers()) - && tx_event - .send(Ok(ResponseEvent::RateLimits(snapshot))) - .await - .is_err() - { - debug!("receiver dropped rate limit snapshot event"); - } - - let stream = resp - .bytes_stream() - .map_err(move |err| Error::ResponseStreamFailed { - source: err, - request_id: request_id.clone(), - }); - let idle_timeout = self.config.provider.stream_idle_timeout(); - let otel = self.config.otel_event_manager.clone(); - - tokio::spawn(crate::client::sse::process_sse( - stream, - tx_event, - idle_timeout, - otel, - crate::decode::responses::ResponsesSseDecoder, - )); - - Ok(ResponseStream { rx_event }) - } - Ok(resp) => Err(handle_error_response(resp, request_id, &self.config).await), - Err(err) => Err(StreamAttemptError::RetryableTransportError(Error::Http( - err, - ))), - } + Ok(crate::stream::EventStream::from_receiver(rx_event)) } } -// payload building is provided by crate::payload::responses - -async fn handle_error_response( - resp: reqwest::Response, - request_id: Option, - _config: &ResponsesApiClientConfig, -) -> StreamAttemptError { - let status = resp.status(); - let retry_after_secs = resp - .headers() - .get(reqwest::header::RETRY_AFTER) - .and_then(|v| v.to_str().ok()) - .and_then(|s| s.parse::().ok()); - let retry_after = retry_after_secs.map(|secs| { - let clamped = if secs < 0 { 0 } else { secs as u64 }; - Duration::from_secs(clamped) - }); - - if !(status == StatusCode::TOO_MANY_REQUESTS - || status == StatusCode::UNAUTHORIZED - || status.is_server_error()) - { - let body = resp.text().await.unwrap_or_default(); - return StreamAttemptError::Fatal(Error::UnexpectedStatus { status, body }); - } - - if status == StatusCode::TOO_MANY_REQUESTS { - let rate_limits = crate::client::rate_limits::parse_rate_limit_snapshot(resp.headers()); - let body = resp.json::().await.ok(); - if let Some(ErrorResponse { error }) = body { - if error.r#type.as_deref() == Some("usage_limit_reached") { - return StreamAttemptError::Fatal(Error::UsageLimitReached { - plan_type: error.plan_type, - resets_at: error.resets_at, - rate_limits, - }); - } else if error.r#type.as_deref() == Some("usage_not_included") { - return StreamAttemptError::Fatal(Error::Stream( - "usage not included".to_string(), - None, - )); - } else if crate::decode::responses::is_quota_exceeded_error(&error) { - return StreamAttemptError::Fatal(Error::Stream( - "quota exceeded".to_string(), - None, - )); - } - } - } - - StreamAttemptError::RetryableHttpError { - status, - retry_after, - request_id, - } -} - -enum StreamAttemptError { - RetryableHttpError { - status: StatusCode, - retry_after: Option, - request_id: Option, - }, - RetryableTransportError(Error), - Fatal(Error), -} - -impl StreamAttemptError { - fn delay(&self, attempt: i64) -> Duration { - match self { - StreamAttemptError::RetryableHttpError { - retry_after: Some(retry_after), - .. - } => *retry_after, - StreamAttemptError::RetryableHttpError { - retry_after: None, .. - } - | StreamAttemptError::RetryableTransportError(..) => backoff(attempt), - StreamAttemptError::Fatal(..) => Duration::from_millis(0), - } - } - - fn into_error(self) -> Error { - match self { - StreamAttemptError::RetryableHttpError { - status, request_id, .. - } => Error::RetryLimit { - status: Some(status), - request_id, - }, - StreamAttemptError::RetryableTransportError(err) | StreamAttemptError::Fatal(err) => { - err - } - } - } -} diff --git a/codex-rs/api-client/src/routed_client.rs b/codex-rs/api-client/src/routed_client.rs index 5c90094a2f..5c0e8dc0c0 100644 --- a/codex-rs/api-client/src/routed_client.rs +++ b/codex-rs/api-client/src/routed_client.rs @@ -5,18 +5,17 @@ use codex_otel::otel_event_manager::OtelEventManager; use codex_protocol::ConversationId; use codex_protocol::protocol::SessionSource; -use crate::ApiClient; use crate::ChatCompletionsApiClient; use crate::ChatCompletionsApiClientConfig; -use crate::Prompt; use crate::ResponseStream; use crate::ResponsesApiClient; use crate::ResponsesApiClientConfig; use crate::Result; use crate::WireApi; +use crate::api::PayloadClient; use crate::auth::AuthProvider; use crate::client::fixtures::stream_from_fixture; -use crate::model_provider::ModelProviderInfo; +use codex_provider_config::ModelProviderInfo; /// Dispatches to the appropriate API client implementation based on the provider wire API. #[derive(Clone)] @@ -41,57 +40,61 @@ impl RoutedApiClient { Self { config } } - pub async fn stream(&self, prompt: &Prompt) -> Result { + pub async fn stream_payload(&self, payload_json: &serde_json::Value) -> Result { match self.config.provider.wire_api { - WireApi::Responses => self.stream_responses(prompt).await, - WireApi::Chat => self.stream_chat(prompt).await, + WireApi::Responses => { + if let Some(path) = &self.config.responses_fixture_path { + return stream_from_fixture( + path, + self.config.provider.clone(), + self.config.otel_event_manager.clone(), + ) + .await; + } + + let cfg = ResponsesApiClientConfig { + http_client: self.config.http_client.clone(), + provider: self.config.provider.clone(), + model: self.config.model.clone(), + conversation_id: self.config.conversation_id, + auth_provider: self.config.auth_provider.clone(), + otel_event_manager: self.config.otel_event_manager.clone(), + }; + let client = ::new(cfg)?; + client + .stream_payload(payload_json, Some(&self.config.session_source)) + .await + } + WireApi::Chat => { + let cfg = ChatCompletionsApiClientConfig { + http_client: self.config.http_client.clone(), + provider: self.config.provider.clone(), + model: self.config.model.clone(), + otel_event_manager: self.config.otel_event_manager.clone(), + session_source: self.config.session_source.clone(), + }; + let client = ::new(cfg)?; + client + .stream_payload(payload_json, Some(&self.config.session_source)) + .await + } } } - - async fn stream_responses(&self, prompt: &Prompt) -> Result { - if let Some(path) = &self.config.responses_fixture_path { - return stream_from_fixture( - path, - self.config.provider.clone(), - self.config.otel_event_manager.clone(), - ) - .await; - } - - let cfg = ResponsesApiClientConfig { - http_client: self.config.http_client.clone(), - provider: self.config.provider.clone(), - model: self.config.model.clone(), - conversation_id: self.config.conversation_id, - auth_provider: self.config.auth_provider.clone(), - otel_event_manager: self.config.otel_event_manager.clone(), - }; - let client = ResponsesApiClient::new(cfg)?; - client.stream(prompt).await - } - - async fn stream_chat(&self, prompt: &Prompt) -> Result { - let cfg = ChatCompletionsApiClientConfig { - http_client: self.config.http_client.clone(), - provider: self.config.provider.clone(), - model: self.config.model.clone(), - otel_event_manager: self.config.otel_event_manager.clone(), - session_source: self.config.session_source.clone(), - }; - let client = ChatCompletionsApiClient::new(cfg)?; - client.stream(prompt).await - } } #[async_trait::async_trait] -impl ApiClient for RoutedApiClient { +impl PayloadClient for RoutedApiClient { type Config = RoutedApiClientConfig; fn new(config: Self::Config) -> Result { Ok(Self::new(config)) } - async fn stream(&self, prompt: &Prompt) -> Result { - RoutedApiClient::stream(self, prompt).await + async fn stream_payload( + &self, + payload_json: &serde_json::Value, + _session_source: Option<&codex_protocol::protocol::SessionSource>, + ) -> Result { + self.stream_payload(payload_json).await } } diff --git a/codex-rs/core/Cargo.toml b/codex-rs/core/Cargo.toml index 7ab2e80d5e..3de9ddb0b2 100644 --- a/codex-rs/core/Cargo.toml +++ b/codex-rs/core/Cargo.toml @@ -23,6 +23,7 @@ codex-app-server-protocol = { workspace = true } codex-apply-patch = { workspace = true } codex-async-utils = { workspace = true } codex-api-client = { workspace = true } +codex-provider-config = { workspace = true } codex-file-search = { workspace = true } codex-git = { workspace = true } codex-keyring-store = { workspace = true } diff --git a/codex-rs/core/src/client.rs b/codex-rs/core/src/client.rs index 9472e2f856..6029be2235 100644 --- a/codex-rs/core/src/client.rs +++ b/codex-rs/core/src/client.rs @@ -119,28 +119,9 @@ impl ModelClient { } pub async fn stream(&self, prompt: &Prompt) -> Result { - let api_prompt = self.build_api_prompt(prompt)?; - - let client = self - .api_client - .get_or_try_init(|| async { self.build_api_client().await }) - .await - .map_err(map_api_error)?; - - let api_stream = client.stream(&api_prompt).await.map_err(map_api_error)?; - Ok(wrap_stream(api_stream)) - } - - fn build_api_prompt(&self, prompt: &Prompt) -> Result { let instructions = prompt .get_full_instructions(&self.config.model_family) .into_owned(); - let input = prompt.get_formatted_input(); - - let tools = match self.provider.wire_api { - WireApi::Responses => create_tools_json_for_responses_api(&prompt.tools)?, - WireApi::Chat => create_tools_json_for_chat_completions_api(&prompt.tools)?, - }; let reasoning = create_reasoning_param_for_request( &self.config.model_family, @@ -160,20 +141,32 @@ impl ModelClient { } else { None }; - let text_controls = create_text_param_for_request(verbosity, &prompt.output_schema); - Ok(codex_api_client::Prompt { - instructions, - input, - tools, - parallel_tool_calls: prompt.parallel_tool_calls, - output_schema: prompt.output_schema.clone(), - reasoning, - text_controls, - prompt_cache_key: Some(self.conversation_id.to_string()), - session_source: Some(self.session_source.clone()), - }) + let payload_json = match self.provider.wire_api { + WireApi::Responses => crate::wire_payload::build_responses_payload( + prompt, + &self.config.model, + self.conversation_id, + self.provider.is_azure_responses_endpoint(), + reasoning, + text_controls, + instructions, + ), + WireApi::Chat => crate::wire_payload::build_chat_payload(prompt, &self.config.model, instructions), + }; + + let client = self + .api_client + .get_or_try_init(|| async { self.build_api_client().await }) + .await + .map_err(map_api_error)?; + + let api_stream = client + .stream_payload(&payload_json) + .await + .map_err(map_api_error)?; + Ok(wrap_stream(api_stream)) } async fn build_api_client(&self) -> ApiClientResult { diff --git a/codex-rs/core/src/lib.rs b/codex-rs/core/src/lib.rs index 020ba52c89..88b340d462 100644 --- a/codex-rs/core/src/lib.rs +++ b/codex-rs/core/src/lib.rs @@ -38,11 +38,12 @@ pub mod token_data; mod truncate; mod unified_exec; mod user_instructions; -pub use codex_api_client::BUILT_IN_OSS_MODEL_PROVIDER_ID; -pub use codex_api_client::ModelProviderInfo; -pub use codex_api_client::WireApi; -pub use codex_api_client::built_in_model_providers; -pub use codex_api_client::create_oss_provider_with_base_url; +mod wire_payload; +pub use codex_provider_config::BUILT_IN_OSS_MODEL_PROVIDER_ID; +pub use codex_provider_config::ModelProviderInfo; +pub use codex_provider_config::WireApi; +pub use codex_provider_config::built_in_model_providers; +pub use codex_provider_config::create_oss_provider_with_base_url; mod conversation_manager; mod event_mapping; pub mod review_format; diff --git a/codex-rs/core/src/wire_payload.rs b/codex-rs/core/src/wire_payload.rs new file mode 100644 index 0000000000..9a9228f29f --- /dev/null +++ b/codex-rs/core/src/wire_payload.rs @@ -0,0 +1,312 @@ +use codex_protocol::ConversationId; +use codex_protocol::models::ResponseItem; +use serde_json::Value; +use serde_json::json; + +use crate::client_common::Prompt; +use crate::tools::spec::create_tools_json_for_responses_api; + +pub fn build_responses_payload( + prompt: &Prompt, + model: &str, + conversation_id: ConversationId, + azure_workaround: bool, + reasoning: Option, + text_controls: Option, + instructions: String, +) -> Value { + let tools = + create_tools_json_for_responses_api(&prompt.tools).unwrap_or_else(|_| Vec::::new()); + + let mut payload = json!({ + "model": model, + "instructions": instructions, + "input": prompt.get_formatted_input(), + "tools": tools, + "tool_choice": "auto", + "parallel_tool_calls": prompt.parallel_tool_calls, + "store": azure_workaround, + "stream": true, + "prompt_cache_key": conversation_id.to_string(), + }); + + if let Some(reasoning) = reasoning + && let Some(obj) = payload.as_object_mut() + { + obj.insert( + "reasoning".to_string(), + serde_json::to_value(reasoning).unwrap_or(Value::Null), + ); + } + + if let Some(text) = text_controls + && let Some(obj) = payload.as_object_mut() + { + obj.insert( + "text".to_string(), + serde_json::to_value(text).unwrap_or(Value::Null), + ); + } + + let include = if prompt + .get_formatted_input() + .iter() + .any(|it| matches!(it, ResponseItem::Reasoning { .. })) + { + vec!["reasoning.encrypted_content".to_string()] + } else { + Vec::new() + }; + if let Some(obj) = payload.as_object_mut() { + obj.insert( + "include".to_string(), + Value::Array(include.into_iter().map(Value::String).collect()), + ); + } + + // Azure Responses requires ids attached to input items + if azure_workaround + && let Some(input_value) = payload.get_mut("input") + && let Some(array) = input_value.as_array_mut() + { + attach_item_ids_array(array, &prompt.get_formatted_input()); + } + + payload +} + +fn attach_item_ids_array(json_array: &mut [Value], prompt_input: &[ResponseItem]) { + for (json_item, item) in json_array.iter_mut().zip(prompt_input.iter()) { + let Some(obj) = json_item.as_object_mut() else { + continue; + }; + let mut set_id_if_absent = |id: &str| match obj.get("id") { + Some(Value::String(s)) if !s.is_empty() => {} + Some(Value::Null) | None => { + obj.insert("id".to_string(), Value::String(id.to_string())); + } + _ => {} + }; + match item { + ResponseItem::Reasoning { id, .. } => set_id_if_absent(id), + ResponseItem::Message { id, .. } => { + if let Some(id) = id.as_ref() { + set_id_if_absent(id); + } + } + ResponseItem::WebSearchCall { id, .. } + | ResponseItem::FunctionCall { id, .. } + | ResponseItem::LocalShellCall { id, .. } + | ResponseItem::CustomToolCall { id, .. } => { + if let Some(id) = id.as_ref() { + set_id_if_absent(id); + } + } + _ => {} + } + } +} + +pub fn build_chat_payload(prompt: &Prompt, model: &str, instructions: String) -> Value { + use codex_protocol::models::ContentItem; + use codex_protocol::models::FunctionCallOutputContentItem; + use codex_protocol::models::ReasoningItemContent; + use std::collections::HashMap; + use crate::tools::spec::create_tools_json_for_chat_completions_api; + + let mut messages = Vec::::new(); + messages.push(json!({ "role": "system", "content": instructions })); + + let mut reasoning_by_anchor_index: HashMap = HashMap::new(); + + let mut last_emitted_role: Option<&str> = None; + for item in &prompt.input { + match item { + ResponseItem::Message { role, .. } => last_emitted_role = Some(role.as_str()), + ResponseItem::FunctionCall { .. } | ResponseItem::LocalShellCall { .. } => { + last_emitted_role = Some("assistant"); + } + ResponseItem::FunctionCallOutput { .. } => last_emitted_role = Some("tool"), + ResponseItem::Reasoning { .. } + | ResponseItem::Other + | ResponseItem::CustomToolCall { .. } + | ResponseItem::CustomToolCallOutput { .. } + | ResponseItem::WebSearchCall { .. } + | ResponseItem::GhostSnapshot { .. } => {} + } + } + + let mut last_user_index: Option = None; + for (idx, item) in prompt.input.iter().enumerate() { + if let ResponseItem::Message { role, .. } = item + && role == "user" + { + last_user_index = Some(idx); + } + } + + if !matches!(last_emitted_role, Some("user")) { + for (idx, item) in prompt.input.iter().enumerate() { + if let Some(u_idx) = last_user_index + && idx <= u_idx + { + continue; + } + + if let ResponseItem::Reasoning { content: Some(items), .. } = item { + let mut text = String::new(); + for entry in items { + match entry { + ReasoningItemContent::ReasoningText { text: segment } + | ReasoningItemContent::Text { text: segment } => { + text.push_str(segment); + } + } + } + if text.trim().is_empty() { + continue; + } + + let mut attached = false; + if idx > 0 + && let ResponseItem::Message { role, .. } = &prompt.input[idx - 1] + && role == "assistant" + { + reasoning_by_anchor_index + .entry(idx - 1) + .and_modify(|val| val.push_str(&text)) + .or_insert(text.clone()); + attached = true; + } + + if !attached && idx + 1 < prompt.input.len() { + match &prompt.input[idx + 1] { + ResponseItem::FunctionCall { .. } | ResponseItem::LocalShellCall { .. } => { + reasoning_by_anchor_index + .entry(idx + 1) + .and_modify(|val| val.push_str(&text)) + .or_insert(text.clone()); + } + ResponseItem::Message { role, .. } if role == "assistant" => { + reasoning_by_anchor_index + .entry(idx + 1) + .and_modify(|val| val.push_str(&text)) + .or_insert(text.clone()); + } + _ => {} + } + } + } + } + } + + let mut last_assistant_text: Option = None; + for (idx, item) in prompt.input.iter().enumerate() { + match item { + ResponseItem::Message { role, content, .. } => { + let mut text = String::new(); + let mut items: Vec = Vec::new(); + let mut saw_image = false; + + for c in content { + match c { + ContentItem::InputText { text: t } | ContentItem::OutputText { text: t } => { + text.push_str(t); + items.push(json!({"type":"text","text": t})); + } + ContentItem::InputImage { image_url } => { + saw_image = true; + items.push(json!({"type":"image_url","image_url": {"url": image_url}})); + } + } + } + + if role == "assistant" { + if let Some(prev) = &last_assistant_text + && prev == &text + { + continue; + } + last_assistant_text = Some(text.clone()); + } + + let content_value = if role == "assistant" { + json!(text) + } else if saw_image { + json!(items) + } else { + json!(text) + }; + + let mut message = json!({ "role": role, "content": content_value }); + if let Some(reasoning) = reasoning_by_anchor_index.get(&idx) + && let Some(obj) = message.as_object_mut() + { + obj.insert("reasoning".to_string(), json!({"text": reasoning})); + } + messages.push(message); + } + ResponseItem::FunctionCall { name, arguments, call_id, .. } => { + messages.push(json!({ + "role": "assistant", + "tool_calls": [{ + "id": call_id, + "type": "function", + "function": { "name": name, "arguments": arguments }, + }], + })); + } + ResponseItem::FunctionCallOutput { call_id, output } => { + let content_value = if let Some(items) = &output.content_items { + let mapped: Vec = items + .iter() + .map(|item| match item { + FunctionCallOutputContentItem::InputText { text } => json!({"type":"text","text": text}), + FunctionCallOutputContentItem::InputImage { image_url } => json!({"type":"image_url","image_url": {"url": image_url}}), + }) + .collect(); + json!(mapped) + } else { + json!(output.content) + }; + messages.push(json!({ "role": "tool", "tool_call_id": call_id, "content": content_value })); + } + ResponseItem::LocalShellCall { id, call_id, action, .. } => { + let tool_id = call_id + .clone() + .filter(|value| !value.is_empty()) + .or_else(|| id.clone()) + .unwrap_or_default(); + messages.push(json!({ + "role": "assistant", + "tool_calls": [{ + "id": tool_id, + "type": "function", + "function": { + "name": "shell", + "arguments": serde_json::to_string(action).unwrap_or_default(), + }, + }], + })); + } + ResponseItem::CustomToolCall { call_id, name, input, .. } => { + messages.push(json!({ + "role": "assistant", + "tool_calls": [{ + "id": call_id.clone(), + "type": "function", + "function": { "name": name, "arguments": input }, + }], + })); + } + ResponseItem::CustomToolCallOutput { call_id, output } => { + messages.push(json!({ "role": "tool", "tool_call_id": call_id, "content": output })); + } + ResponseItem::WebSearchCall { .. } | ResponseItem::Reasoning { .. } | ResponseItem::Other | ResponseItem::GhostSnapshot { .. } => {} + } + } + + let tools_json = create_tools_json_for_chat_completions_api(&prompt.tools) + .unwrap_or_else(|_| Vec::::new()); + json!({ "model": model, "messages": messages, "stream": true, "tools": tools_json }) +} diff --git a/codex-rs/provider-config/Cargo.toml b/codex-rs/provider-config/Cargo.toml new file mode 100644 index 0000000000..99aa9a0371 --- /dev/null +++ b/codex-rs/provider-config/Cargo.toml @@ -0,0 +1,18 @@ +[package] +name = "codex-provider-config" +version.workspace = true +edition.workspace = true + +[lib] +name = "codex_provider_config" +path = "src/lib.rs" + +[lints] +workspace = true + +[dependencies] +codex-app-server-protocol = { workspace = true } +serde = { workspace = true, features = ["derive"] } +serde_json = { workspace = true } +thiserror = { workspace = true } +reqwest = { workspace = true } diff --git a/codex-rs/api-client/src/model_provider.rs b/codex-rs/provider-config/src/lib.rs similarity index 67% rename from codex-rs/api-client/src/model_provider.rs rename to codex-rs/provider-config/src/lib.rs index 1b67a91ea5..beed624829 100644 --- a/codex-rs/api-client/src/model_provider.rs +++ b/codex-rs/provider-config/src/lib.rs @@ -1,9 +1,8 @@ -//! Registry of model providers supported by Codex. +//! Provider configuration shared across Codex layers. //! -//! Providers can be defined in two places: -//! 1. Built-in defaults compiled into the binary so Codex works out-of-the-box. -//! 2. User-defined entries inside `~/.codex/config.toml` under the `model_providers` -//! key. These override or extend the defaults at runtime. +//! This crate defines the provider-agnostic configuration and wire API +//! selection that higher layers (core/app/client) can use. It intentionally +//! avoids Codex-domain concepts like prompts, token counting, or event types. use std::collections::HashMap; use std::env::VarError; @@ -13,9 +12,16 @@ use codex_app_server_protocol::AuthMode; use serde::Deserialize; use serde::Serialize; -use crate::auth::AuthContext; -use crate::error::Error; -use crate::error::Result; +#[derive(Debug, thiserror::Error)] +pub enum Error { + #[error("missing environment variable {var}")] + MissingEnvVar { + var: String, + instructions: Option, + }, +} + +pub type Result = std::result::Result; const DEFAULT_STREAM_IDLE_TIMEOUT_MS: i64 = 300_000; const DEFAULT_STREAM_MAX_RETRIES: i64 = 5; @@ -26,12 +32,7 @@ const MAX_STREAM_MAX_RETRIES: i64 = 100; const MAX_REQUEST_MAX_RETRIES: i64 = 100; const DEFAULT_OLLAMA_PORT: i32 = 11434; -/// Wire protocol that the provider speaks. Most third-party services only -/// implement the classic OpenAI Chat Completions JSON schema, whereas OpenAI -/// itself (and a handful of others) additionally expose the more modern -/// Responses API. The two protocols use different request/response shapes -/// and cannot be auto-detected at runtime, therefore each provider entry -/// must declare which one it expects. +/// Wire protocol that the provider speaks. #[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)] #[serde(rename_all = "lowercase")] pub enum WireApi { @@ -51,97 +52,32 @@ pub struct ModelProviderInfo { pub base_url: Option, /// Environment variable that stores the user's API key for this provider. pub env_key: Option, - /// Optional instructions to help the user get a valid value for the - /// variable and set it. + /// Optional instructions to help the user set the environment variable. pub env_key_instructions: Option, - /// Value to use with `Authorization: Bearer ` header. Use of this - /// config is discouraged in favor of `env_key` for security reasons, but - /// this may be necessary when using this programmatically. + /// Value to use with `Authorization: Bearer ` header. Prefer `env_key` when possible. pub experimental_bearer_token: Option, /// Which wire protocol this provider expects. #[serde(default)] pub wire_api: WireApi, /// Optional query parameters to append to the base URL. pub query_params: Option>, - /// Additional HTTP headers to include in requests to this provider where - /// the (key, value) pairs are the header name and value. + /// Additional static HTTP headers to include in requests. pub http_headers: Option>, - /// Optional HTTP headers to include in requests to this provider where the - /// (key, value) pairs are the header name and environment variable whose - /// value should be used. If the environment variable is not set, or the - /// value is empty, the header will not be included in the request. + /// Optional HTTP headers whose values come from environment variables. pub env_http_headers: Option>, - /// Maximum number of times to retry a failed HTTP request to this provider. + /// Maximum number of times to retry a failed HTTP request. pub request_max_retries: Option, /// Number of times to retry reconnecting a dropped streaming response before failing. pub stream_max_retries: Option, - /// Idle timeout (in milliseconds) to wait for activity on a streaming response before treating - /// the connection as lost. + /// Idle timeout (in milliseconds) to wait for activity on a streaming response. pub stream_idle_timeout_ms: Option, - /// Does this provider require an OpenAI API Key or ChatGPT login token? If true, - /// the user is presented with a login screen on first run, and login preference and token/key - /// are stored in auth.json. If false (which is the default), the login screen is skipped, - /// and the API key (if needed) comes from the `env_key` environment variable. + /// If true, user is prompted for OpenAI login; otherwise uses `env_key`. #[serde(default)] pub requires_openai_auth: bool, } impl ModelProviderInfo { - /// Construct a `POST` request builder for the given URL using the provided - /// [`reqwest::Client`] applying: - /// - provider-specific headers (static and environment based) - /// - Bearer auth header when an API key is available - /// - Auth token for OAuth - /// - /// If the provider declares an `env_key` but the variable is missing or empty, this returns an - /// error identical to the one produced by [`ModelProviderInfo::api_key`]. - pub async fn create_request_builder( - &self, - client: &reqwest::Client, - auth: &Option, - ) -> Result { - let effective_auth = if let Some(secret_key) = &self.experimental_bearer_token { - Some(AuthContext { - mode: AuthMode::ApiKey, - bearer_token: Some(secret_key.clone()), - account_id: None, - }) - } else { - match self.api_key()? { - Some(key) => Some(AuthContext { - mode: AuthMode::ApiKey, - bearer_token: Some(key), - account_id: None, - }), - None => auth.clone(), - } - }; - - let url = self.get_full_url(effective_auth.as_ref()); - let mut builder = client.post(url); - - if let Some(context) = effective_auth.as_ref() - && let Some(token) = context.bearer_token.as_ref() - { - builder = builder.bearer_auth(token); - } - - Ok(self.apply_http_headers(builder)) - } - - fn get_query_string(&self) -> String { - self.query_params - .as_ref() - .map_or_else(String::new, |params| { - let full_params = params - .iter() - .map(|(k, v)| format!("{k}={v}")) - .collect::>() - .join("&"); - format!("?{full_params}") - }) - } - + /// Construct a `POST` request URL for the configured wire API. pub fn get_full_url(&self, auth: Option<&AuthContext>) -> String { let default_base_url = if matches!( auth, @@ -166,30 +102,42 @@ impl ModelProviderInfo { } } + fn get_query_string(&self) -> String { + self.query_params + .as_ref() + .map_or_else(String::new, |params| { + let full_params = params + .iter() + .map(|(k, v)| format!("{k}={v}")) + .collect::>() + .join("&"); + format!("?{full_params}") + }) + } + pub fn is_azure_responses_endpoint(&self) -> bool { if self.wire_api != WireApi::Responses { return false; } - if self.name.eq_ignore_ascii_case("azure") { return true; } - self.base_url .as_ref() .map(|base| matches_azure_responses_base_url(base)) .unwrap_or(false) } - /// Apply provider-specific HTTP headers (both static and environment-based) onto an existing - /// [`reqwest::RequestBuilder`] and return the updated builder. - fn apply_http_headers(&self, mut builder: reqwest::RequestBuilder) -> reqwest::RequestBuilder { + /// Apply static and env-derived headers to the provided builder. + pub fn apply_http_headers( + &self, + mut builder: reqwest::RequestBuilder, + ) -> reqwest::RequestBuilder { if let Some(extra) = &self.http_headers { for (k, v) in extra { builder = builder.header(k, v); } } - if let Some(env_headers) = &self.env_http_headers { for (header, env_var) in env_headers { if let Ok(val) = std::env::var(env_var) @@ -199,7 +147,6 @@ impl ModelProviderInfo { } } } - builder } @@ -259,18 +206,41 @@ pub const BUILT_IN_OSS_MODEL_PROVIDER_ID: &str = "openai/compatible"; pub const OPENAI_MODEL_PROVIDER_ID: &str = "openai"; pub const ANTHROPIC_MODEL_PROVIDER_ID: &str = "anthropic"; -/// Returns the baked-in list of providers. These can be overridden by a `[model_providers]` -/// entry inside `~/.codex/config.toml`. -pub fn built_in_model_providers() -> HashMap { - let mut providers = HashMap::new(); +/// Convenience helper to construct a default `openai/compatible` provider pointing at localhost. +pub fn create_oss_provider_with_base_url(url: &str) -> ModelProviderInfo { + ModelProviderInfo { + name: "openai/compatible".to_string(), + base_url: Some(url.to_string()), + env_key: None, + env_key_instructions: None, + experimental_bearer_token: None, + wire_api: WireApi::Chat, + query_params: None, + http_headers: None, + env_http_headers: None, + request_max_retries: None, + stream_max_retries: None, + stream_idle_timeout_ms: None, + requires_openai_auth: false, + } +} - providers.insert( +pub fn create_oss_provider() -> ModelProviderInfo { + create_oss_provider_with_base_url(&format!("http://localhost:{DEFAULT_OLLAMA_PORT}")) +} + +pub fn built_in_model_providers() -> HashMap { + let mut map = HashMap::new(); + + map.insert( OPENAI_MODEL_PROVIDER_ID.to_string(), ModelProviderInfo { name: "OpenAI".to_string(), base_url: None, env_key: Some("OPENAI_API_KEY".to_string()), - env_key_instructions: Some("Log in to OpenAI and create a new API key at https://platform.openai.com/api-keys. Then paste it here.".to_string()), + env_key_instructions: Some( + "Log in to OpenAI and create a new API key at https://platform.openai.com/api-keys. Then paste it here.".to_string(), + ), experimental_bearer_token: None, wire_api: WireApi::Responses, query_params: None, @@ -283,21 +253,17 @@ pub fn built_in_model_providers() -> HashMap { }, ); - providers.insert( + map.insert( ANTHROPIC_MODEL_PROVIDER_ID.to_string(), ModelProviderInfo { name: "Anthropic".to_string(), - base_url: Some("https://api.anthropic.com/v1/messages".to_string()), + base_url: Some("https://api.anthropic.com/v1".to_string()), env_key: Some("ANTHROPIC_API_KEY".to_string()), - env_key_instructions: Some("Create a new API key at https://console.anthropic.com/settings/keys and paste it here.".to_string()), + env_key_instructions: None, experimental_bearer_token: None, wire_api: WireApi::Chat, query_params: None, - http_headers: Some( - maplit::hashmap! { - "anthropic-version".to_string() => "2023-06-01".to_string(), - } - ), + http_headers: None, env_http_headers: None, request_max_retries: None, stream_max_retries: None, @@ -306,38 +272,59 @@ pub fn built_in_model_providers() -> HashMap { }, ); - providers.insert( + map.insert( BUILT_IN_OSS_MODEL_PROVIDER_ID.to_string(), create_oss_provider_with_base_url("http://localhost:11434"), ); - providers + map } -pub fn create_oss_provider_with_base_url(url: &str) -> ModelProviderInfo { - let http_headers = maplit::hashmap! { - "x-oss-provider".to_string() => "ollama".to_string(), - }; - ModelProviderInfo { - name: "Self-hosted OpenAI-compatible (OSS)".to_string(), - base_url: Some(url.to_string()), - env_key: Some("CODEX_OSS_PROVIDER_API_KEY".to_string()), - env_key_instructions: Some( - "Set CODEX_OSS_PROVIDER_API_KEY to authenticate with this provider.".to_string(), - ), - experimental_bearer_token: None, - wire_api: WireApi::Chat, - query_params: None, - http_headers: Some(http_headers), - env_http_headers: None, - request_max_retries: None, - stream_max_retries: None, - stream_idle_timeout_ms: None, - requires_openai_auth: false, +/// Minimal auth context used only for computing URLs and headers. +#[derive(Debug, Clone)] +pub struct AuthContext { + pub mode: AuthMode, + pub bearer_token: Option, + pub account_id: Option, +} + +impl ModelProviderInfo { + /// Convenience to create a request builder with provider and auth headers. + pub async fn create_request_builder( + &self, + client: &reqwest::Client, + auth: &Option, + ) -> Result { + let effective_auth = if let Some(secret_key) = &self.experimental_bearer_token { + Some(AuthContext { + mode: AuthMode::ApiKey, + bearer_token: Some(secret_key.clone()), + account_id: None, + }) + } else { + match self.api_key()? { + Some(key) => Some(AuthContext { + mode: AuthMode::ApiKey, + bearer_token: Some(key), + account_id: None, + }), + None => auth.clone(), + } + }; + + let mut builder = client.post(self.get_full_url(effective_auth.as_ref())); + builder = self.apply_http_headers(builder); + + if let Some(context) = effective_auth.as_ref() { + if let Some(token) = context.bearer_token.as_ref() { + builder = builder.bearer_auth(token); + } + if let Some(account) = context.account_id.as_ref() { + builder = builder.header("OpenAI-Beta", "codex-2"); + builder = builder.header("OpenAI-Organization", account); + } + } + + Ok(builder) } } - -/// Convenience helper to construct a default `openai/compatible` provider pointing at localhost. -pub fn create_oss_provider() -> ModelProviderInfo { - create_oss_provider_with_base_url(&format!("http://localhost:{DEFAULT_OLLAMA_PORT}")) -}