From 0bba25eff38b0220a499f6331df9668a06375ec9 Mon Sep 17 00:00:00 2001 From: jif-oai Date: Thu, 30 Oct 2025 15:34:21 +0000 Subject: [PATCH] V3 --- codex-rs/Cargo.lock | 13 + codex-rs/api-client/Cargo.toml | 23 +- codex-rs/api-client/src/api.rs | 16 + codex-rs/api-client/src/auth.rs | 15 + codex-rs/api-client/src/chat.rs | 855 ++++++++++ codex-rs/api-client/src/error.rs | 42 + codex-rs/api-client/src/lib.rs | 26 + .../src/model_provider.rs} | 191 +-- codex-rs/api-client/src/prompt.rs | 33 +- codex-rs/api-client/src/responses.rs | 741 ++++++++ codex-rs/api-client/src/stream.rs | 41 +- codex-rs/core/src/chat_completions.rs | 966 ----------- codex-rs/core/src/client.rs | 1515 ++--------------- codex-rs/core/src/client_common.rs | 19 +- codex-rs/core/src/codex.rs | 23 +- codex-rs/core/src/codex/compact.rs | 16 +- codex-rs/core/src/config/mod.rs | 4 +- codex-rs/core/src/default_client.rs | 8 + codex-rs/core/src/lib.rs | 12 +- codex-rs/core/src/sandboxing/assessment.rs | 20 +- codex-rs/core/src/tools/spec.rs | 28 - 21 files changed, 2076 insertions(+), 2531 deletions(-) create mode 100644 codex-rs/api-client/src/api.rs create mode 100644 codex-rs/api-client/src/auth.rs create mode 100644 codex-rs/api-client/src/chat.rs create mode 100644 codex-rs/api-client/src/error.rs rename codex-rs/{core/src/model_provider_info.rs => api-client/src/model_provider.rs} (64%) create mode 100644 codex-rs/api-client/src/responses.rs delete mode 100644 codex-rs/core/src/chat_completions.rs diff --git a/codex-rs/Cargo.lock b/codex-rs/Cargo.lock index 85cec820d9..baa98b401f 100644 --- a/codex-rs/Cargo.lock +++ b/codex-rs/Cargo.lock @@ -833,11 +833,24 @@ dependencies = [ name = "codex-api-client" version = "0.0.0" dependencies = [ + "async-trait", + "bytes", + "codex-app-server-protocol", + "codex-otel", "codex-protocol", + "eventsource-stream", "futures", + "maplit", + "once_cell", + "regex-lite", + "reqwest", "serde", "serde_json", + "thiserror 2.0.16", "tokio", + "tokio-util", + "toml", + "tracing", ] [[package]] diff --git a/codex-rs/api-client/Cargo.toml b/codex-rs/api-client/Cargo.toml index 67808e80b9..4ebe4d7991 100644 --- a/codex-rs/api-client/Cargo.toml +++ b/codex-rs/api-client/Cargo.toml @@ -4,11 +4,26 @@ version.workspace = true edition.workspace = true [dependencies] -futures = "0.3.31" -serde = { version = "1.0", features = ["derive"] } -serde_json = "1.0" -tokio = { version = "1.41", features = ["sync"] } +async-trait = { workspace = true } +bytes = { workspace = true } +codex-app-server-protocol = { workspace = true } +codex-otel = { workspace = true } codex-protocol = { path = "../protocol" } +eventsource-stream = { workspace = true } +futures = { workspace = true, default-features = false, features = ["std"] } +once_cell = "1.20.2" +regex-lite = { workspace = true } +reqwest = { workspace = true, features = ["json", "stream"] } +serde = { workspace = true, features = ["derive"] } +serde_json = { workspace = true } +thiserror = { workspace = true } +tokio = { workspace = true, features = ["sync", "time", "rt", "rt-multi-thread", "macros", "io-util"] } +tokio-util = { workspace = true } +tracing = { workspace = true } + +[dev-dependencies] +maplit = "1.0.2" +toml = { workspace = true } [lints] workspace = true diff --git a/codex-rs/api-client/src/api.rs b/codex-rs/api-client/src/api.rs new file mode 100644 index 0000000000..8ebb356a1a --- /dev/null +++ b/codex-rs/api-client/src/api.rs @@ -0,0 +1,16 @@ +use async_trait::async_trait; + +use crate::error::Error; +use crate::prompt::Prompt; +use crate::stream::ResponseStream; + +#[async_trait] +pub trait ApiClient: Send + Sync { + type Config: Send + Sync; + + async fn new(config: Self::Config) -> Result + where + Self: Sized; + + async fn stream(&self, prompt: Prompt) -> Result; +} diff --git a/codex-rs/api-client/src/auth.rs b/codex-rs/api-client/src/auth.rs new file mode 100644 index 0000000000..0c6fb94490 --- /dev/null +++ b/codex-rs/api-client/src/auth.rs @@ -0,0 +1,15 @@ +use async_trait::async_trait; +use codex_app_server_protocol::AuthMode; + +#[derive(Debug, Clone)] +pub struct AuthContext { + pub mode: AuthMode, + pub bearer_token: Option, + pub account_id: Option, +} + +#[async_trait] +pub trait AuthProvider: Send + Sync { + async fn auth_context(&self) -> Option; + async fn refresh_token(&self) -> Result, String>; +} diff --git a/codex-rs/api-client/src/chat.rs b/codex-rs/api-client/src/chat.rs new file mode 100644 index 0000000000..d990320f92 --- /dev/null +++ b/codex-rs/api-client/src/chat.rs @@ -0,0 +1,855 @@ +use std::collections::VecDeque; +use std::pin::Pin; +use std::task::Context; +use std::task::Poll; +use std::time::Duration; + +use async_trait::async_trait; +use bytes::Bytes; +use codex_otel::otel_event_manager::OtelEventManager; +use codex_protocol::models::ContentItem; +use codex_protocol::models::FunctionCallOutputContentItem; +use codex_protocol::models::ReasoningItemContent; +use codex_protocol::models::ResponseItem; +use codex_protocol::protocol::SessionSource; +use codex_protocol::protocol::SubAgentSource; +use eventsource_stream::Eventsource; +use futures::Stream; +use futures::StreamExt; +use futures::TryStreamExt; +use serde_json::json; +use tokio::sync::mpsc; +use tokio::time::timeout; +use tracing::debug; +use tracing::trace; + +use crate::api::ApiClient; +use crate::error::Error; +use crate::model_provider::ModelProviderInfo; +use crate::prompt::Prompt; +use crate::stream::ResponseEvent; +use crate::stream::ResponseStream; + +pub type Result = std::result::Result; + +#[derive(Clone, Copy, Debug)] +pub enum ChatAggregationMode { + AggregatedOnly, + Streaming, +} + +#[derive(Clone)] +pub struct ChatCompletionsApiClientConfig { + pub http_client: reqwest::Client, + pub provider: ModelProviderInfo, + pub model: String, + pub otel_event_manager: OtelEventManager, + pub session_source: SessionSource, + pub aggregation_mode: ChatAggregationMode, +} + +#[derive(Clone)] +pub struct ChatCompletionsApiClient { + config: ChatCompletionsApiClientConfig, +} + +#[async_trait] +impl ApiClient for ChatCompletionsApiClient { + type Config = ChatCompletionsApiClientConfig; + + async fn new(config: Self::Config) -> Result { + Ok(Self { config }) + } + + async fn stream(&self, prompt: Prompt) -> Result { + Self::validate_prompt(&prompt)?; + + let payload = self.build_payload(&prompt)?; + let (tx_event, rx_event) = mpsc::channel::>(1600); + + let mut attempt = 0u64; + let max_retries = self.config.provider.request_max_retries(); + + loop { + attempt += 1; + + let mut req_builder = self + .config + .provider + .create_request_builder(&self.config.http_client, &None) + .await?; + + if let SessionSource::SubAgent(sub) = &self.config.session_source { + let subagent = if let SubAgentSource::Other(label) = sub { + label.clone() + } else { + serde_json::to_value(sub) + .ok() + .and_then(|v| v.as_str().map(std::string::ToString::to_string)) + .unwrap_or_else(|| "other".to_string()) + }; + req_builder = req_builder.header("x-openai-subagent", subagent); + } + + let res = self + .config + .otel_event_manager + .log_request(attempt, || { + 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(); + let mode = self.config.aggregation_mode; + + tokio::spawn(process_chat_sse( + stream, + tx_event.clone(), + idle_timeout, + otel, + mode, + )); + + 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(|s| Duration::from_secs(s)); + 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(), + )); + } + Ok(()) + } + + fn build_payload(&self, prompt: &Prompt) -> Result { + let mut messages = Vec::::new(); + messages.push(json!({ "role": "system", "content": prompt.instructions })); + + let mut reasoning_by_anchor_index: std::collections::HashMap = + std::collections::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(|v| v.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(|v| v.push_str(&text)) + .or_insert(text.clone()); + } + ResponseItem::Message { role, .. } if role == "assistant" => { + reasoning_by_anchor_index + .entry(idx + 1) + .and_modify(|v| v.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) { + message + .as_object_mut() + .expect("message") + .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 { + call_id, action, .. + } => { + messages.push(json!({ + "role": "assistant", + "tool_calls": [{ + "id": call_id.clone().unwrap_or_default(), + "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": json!(input).to_string(), + }, + }], + })); + } + 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.config.model, + "messages": messages, + "stream": true, + "tools": tools_json, + }); + + trace!("chat completions payload: {}", payload); + Ok(payload) + } +} + +async fn append_assistant_text( + tx_event: &mpsc::Sender>, + assistant_item: &mut Option, + text: String, +) { + if assistant_item.is_none() { + let item = ResponseItem::Message { + id: None, + role: "assistant".to_string(), + content: vec![], + }; + *assistant_item = Some(item.clone()); + let _ = tx_event + .send(Ok(ResponseEvent::OutputItemAdded(item))) + .await; + } + + if let Some(ResponseItem::Message { content, .. }) = assistant_item { + content.push(ContentItem::OutputText { text: text.clone() }); + let _ = tx_event + .send(Ok(ResponseEvent::OutputTextDelta(text.clone()))) + .await; + } +} + +async fn append_reasoning_text( + tx_event: &mpsc::Sender>, + reasoning_item: &mut Option, + text: String, +) { + if reasoning_item.is_none() { + let item = ResponseItem::Reasoning { + id: String::new(), + summary: Vec::new(), + content: Some(vec![]), + encrypted_content: None, + }; + *reasoning_item = Some(item.clone()); + let _ = tx_event + .send(Ok(ResponseEvent::OutputItemAdded(item))) + .await; + } + + if let Some(ResponseItem::Reasoning { + content: Some(content), + .. + }) = reasoning_item + { + content.push(ReasoningItemContent::ReasoningText { text: text.clone() }); + + let _ = tx_event + .send(Ok(ResponseEvent::ReasoningContentDelta(text.clone()))) + .await; + } +} + +async fn process_chat_sse( + stream: S, + tx_event: mpsc::Sender>, + idle_timeout: Duration, + otel_event_manager: OtelEventManager, + _aggregation_mode: ChatAggregationMode, +) where + S: Stream> + Unpin, +{ + let mut stream = stream.eventsource(); + + #[derive(Default)] + struct FunctionCallState { + name: Option, + arguments: String, + call_id: Option, + } + + let mut function_call_state = FunctionCallState::default(); + let mut assistant_item: Option = None; + let mut reasoning_item: Option = None; + + loop { + let response = timeout(idle_timeout, stream.next()).await; + otel_event_manager.log_sse_event(&response, idle_timeout); + + let sse = match response { + Ok(Some(Ok(sse))) => sse, + Ok(Some(Err(e))) => { + debug!("SSE Error: {e:#}"); + let event = Error::Stream(e.to_string(), None); + let _ = tx_event.send(Err(event)).await; + return; + } + Ok(None) => { + if let Some(item) = assistant_item.take() { + let _ = tx_event.send(Ok(ResponseEvent::OutputItemDone(item))).await; + } + if let Some(item) = reasoning_item.take() { + let _ = tx_event.send(Ok(ResponseEvent::OutputItemDone(item))).await; + } + let _ = tx_event + .send(Ok(ResponseEvent::Completed { + response_id: String::new(), + token_usage: None, + })) + .await; + return; + } + Err(_) => { + let _ = tx_event + .send(Err(Error::Stream( + "idle timeout waiting for SSE".into(), + None, + ))) + .await; + return; + } + }; + + trace!("chat_completions received SSE chunk: {}", sse.data); + + if sse.data.trim() == "[DONE]" { + if let Some(item) = assistant_item.take() { + let _ = tx_event.send(Ok(ResponseEvent::OutputItemDone(item))).await; + } + if let Some(item) = reasoning_item.take() { + let _ = tx_event.send(Ok(ResponseEvent::OutputItemDone(item))).await; + } + let _ = tx_event + .send(Ok(ResponseEvent::Completed { + response_id: String::new(), + token_usage: None, + })) + .await; + return; + } + + let chunk: serde_json::Value = match serde_json::from_str(&sse.data) { + Ok(v) => v, + Err(_) => continue, + }; + + let choice_opt = chunk.get("choices").and_then(|c| c.get(0)); + + if let Some(choice) = choice_opt { + if let Some(content) = choice + .get("delta") + .and_then(|d| d.get("content")) + .and_then(|c| c.as_str()) + && !content.is_empty() + { + append_assistant_text(&tx_event, &mut assistant_item, content.to_string()).await; + } + + if let Some(reasoning_val) = choice.get("delta").and_then(|d| d.get("reasoning")) { + let mut maybe_text = reasoning_val + .as_str() + .map(str::to_string) + .filter(|s| !s.is_empty()); + + if maybe_text.is_none() && reasoning_val.is_object() { + if let Some(s) = reasoning_val + .get("text") + .and_then(|t| t.as_str()) + .filter(|s| !s.is_empty()) + { + maybe_text = Some(s.to_string()); + } else if let Some(s) = reasoning_val + .get("content") + .and_then(|t| t.as_str()) + .filter(|s| !s.is_empty()) + { + maybe_text = Some(s.to_string()); + } + } + + if let Some(reasoning) = maybe_text { + append_reasoning_text(&tx_event, &mut reasoning_item, reasoning).await; + } + } + + if let Some(message_reasoning) = choice.get("message").and_then(|m| m.get("reasoning")) + { + if let Some(s) = message_reasoning.as_str() { + if !s.is_empty() { + append_reasoning_text(&tx_event, &mut reasoning_item, s.to_string()).await; + } + } else if let Some(obj) = message_reasoning.as_object() + && let Some(s) = obj + .get("text") + .and_then(|v| v.as_str()) + .or_else(|| obj.get("content").and_then(|v| v.as_str())) + && !s.is_empty() + { + append_reasoning_text(&tx_event, &mut reasoning_item, s.to_string()).await; + } + } + + if let Some(tool_calls) = choice + .get("delta") + .and_then(|d| d.get("tool_calls")) + .and_then(|v| v.as_array()) + { + for call in tool_calls { + if let Some(index) = call.get("index").and_then(|i| i.as_u64()) + && index == 0 + && let Some(function) = call.get("function") + { + if let Some(name) = function.get("name").and_then(|n| n.as_str()) { + function_call_state.name = Some(name.to_string()); + } + if let Some(arguments) = function.get("arguments").and_then(|a| a.as_str()) + { + function_call_state.arguments.push_str(arguments); + } + if let Some(id) = call.get("id").and_then(|i| i.as_str()) { + function_call_state.call_id = Some(id.to_string()); + } + + if let Some(finish) = choice.get("finish_reason").and_then(|f| f.as_str()) + && finish == "tool_calls" + { + if let Some(name) = function_call_state.name.take() { + let call_id = + function_call_state.call_id.take().unwrap_or_default(); + let arguments = std::mem::take(&mut function_call_state.arguments); + let item = ResponseItem::FunctionCall { + id: None, + name, + arguments, + call_id, + }; + let _ = + tx_event.send(Ok(ResponseEvent::OutputItemDone(item))).await; + } + } + } + } + } + } + } +} + +pub trait AggregateStreamExt: Stream> + Sized { + fn aggregate(self) -> AggregatedChatStream + where + Self: Unpin, + { + AggregatedChatStream::new(self, AggregateMode::AggregatedOnly) + } + + fn streaming_mode(self) -> AggregatedChatStream + where + Self: Unpin, + { + AggregatedChatStream::new(self, AggregateMode::Streaming) + } +} + +impl AggregateStreamExt for S where S: Stream> + Sized + Unpin {} + +enum AggregateMode { + AggregatedOnly, + Streaming, +} + +pub struct AggregatedChatStream { + inner: S, + cumulative: String, + cumulative_reasoning: String, + pending: VecDeque, + mode: AggregateMode, +} + +impl AggregatedChatStream +where + S: Stream> + Unpin, +{ + fn new(inner: S, mode: AggregateMode) -> Self { + Self { + inner, + cumulative: String::new(), + cumulative_reasoning: String::new(), + pending: VecDeque::new(), + mode, + } + } +} + +impl Stream for AggregatedChatStream +where + S: Stream> + Unpin, +{ + type Item = Result; + + fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + if let Some(ev) = self.pending.pop_front() { + return Poll::Ready(Some(Ok(ev))); + } + + loop { + match Pin::new(&mut self.inner).poll_next(cx) { + Poll::Pending => return Poll::Pending, + Poll::Ready(None) => return Poll::Ready(None), + Poll::Ready(Some(Err(e))) => return Poll::Ready(Some(Err(e))), + Poll::Ready(Some(Ok(ResponseEvent::OutputItemDone(item)))) => { + let is_assistant_message = matches!( + &item, + ResponseItem::Message { role, .. } if role == "assistant" + ); + + if is_assistant_message { + match self.mode { + AggregateMode::AggregatedOnly => { + if self.cumulative.is_empty() + && let ResponseItem::Message { content, .. } = &item + && let Some(text) = content.iter().find_map(|c| match c { + ContentItem::OutputText { text } => Some(text), + _ => None, + }) + { + self.cumulative.push_str(text); + } + continue; + } + AggregateMode::Streaming => { + if self.cumulative.is_empty() { + return Poll::Ready(Some(Ok(ResponseEvent::OutputItemDone( + item, + )))); + } else { + continue; + } + } + } + } + + return Poll::Ready(Some(Ok(ResponseEvent::OutputItemDone(item)))); + } + Poll::Ready(Some(Ok(ResponseEvent::RateLimits(snapshot)))) => { + return Poll::Ready(Some(Ok(ResponseEvent::RateLimits(snapshot)))); + } + Poll::Ready(Some(Ok(ResponseEvent::Completed { + response_id, + token_usage, + }))) => { + let mut emitted_any = false; + + if !self.cumulative_reasoning.is_empty() + && matches!(self.mode, AggregateMode::AggregatedOnly) + { + let aggregated_reasoning = ResponseItem::Reasoning { + id: String::new(), + summary: Vec::new(), + content: Some(vec![ReasoningItemContent::ReasoningText { + text: std::mem::take(&mut self.cumulative_reasoning), + }]), + encrypted_content: None, + }; + self.pending + .push_back(ResponseEvent::OutputItemDone(aggregated_reasoning)); + emitted_any = true; + } + + if !self.cumulative.is_empty() { + let aggregated_message = ResponseItem::Message { + id: None, + role: "assistant".to_string(), + content: vec![ContentItem::OutputText { + text: std::mem::take(&mut self.cumulative), + }], + }; + self.pending + .push_back(ResponseEvent::OutputItemDone(aggregated_message)); + emitted_any = true; + } + + if emitted_any { + self.pending.push_back(ResponseEvent::Completed { + response_id: response_id.clone(), + token_usage: token_usage.clone(), + }); + if let Some(ev) = self.pending.pop_front() { + return Poll::Ready(Some(Ok(ev))); + } + } + + return Poll::Ready(Some(Ok(ResponseEvent::Completed { + response_id, + token_usage, + }))); + } + Poll::Ready(Some(Ok(ResponseEvent::Created))) => continue, + Poll::Ready(Some(Ok(ResponseEvent::OutputTextDelta(delta)))) => { + self.cumulative.push_str(&delta); + if matches!(self.mode, AggregateMode::Streaming) { + return Poll::Ready(Some(Ok(ResponseEvent::OutputTextDelta(delta)))); + } + } + Poll::Ready(Some(Ok(ResponseEvent::ReasoningContentDelta(delta)))) => { + self.cumulative_reasoning.push_str(&delta); + if matches!(self.mode, AggregateMode::Streaming) { + return Poll::Ready(Some(Ok(ResponseEvent::ReasoningContentDelta(delta)))); + } + } + Poll::Ready(Some(Ok(ResponseEvent::ReasoningSummaryDelta(_)))) => continue, + Poll::Ready(Some(Ok(ResponseEvent::ReasoningSummaryPartAdded))) => continue, + Poll::Ready(Some(Ok(ResponseEvent::OutputItemAdded(item)))) => { + return Poll::Ready(Some(Ok(ResponseEvent::OutputItemAdded(item)))); + } + } + } + } +} + +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; + } + + tool.as_object().map(|map| { + let mut function = map.clone(); + function.remove("type"); + json!({ + "type": "function", + "function": function, + }) + }) + }) + .collect::>(); + Ok(tools_json) +} + +fn backoff(attempt: u64) -> Duration { + let capped = attempt.min(6); + Duration::from_millis(100 * 2u64.pow(capped as u32)) +} diff --git a/codex-rs/api-client/src/error.rs b/codex-rs/api-client/src/error.rs new file mode 100644 index 0000000000..c02dcab71a --- /dev/null +++ b/codex-rs/api-client/src/error.rs @@ -0,0 +1,42 @@ +use std::time::Duration; + +use thiserror::Error; + +pub type Result = std::result::Result; + +#[derive(Debug, Error)] +pub enum Error { + #[error("{0}")] + UnsupportedOperation(String), + #[error(transparent)] + Http(#[from] reqwest::Error), + #[error("{source}")] + ResponseStreamFailed { + #[source] + source: reqwest::Error, + request_id: Option, + }, + #[error("{0}")] + Stream(String, Option), + #[error("unexpected status {status}: {body}")] + UnexpectedStatus { + status: reqwest::StatusCode, + body: String, + }, + #[error("retry limit reached (status {status}, request id: {request_id:?})")] + RetryLimit { + status: reqwest::StatusCode, + request_id: Option, + }, + #[error("missing environment variable {var}")] + MissingEnvVar { + var: String, + instructions: Option, + }, + #[error("{0}")] + Auth(String), + #[error(transparent)] + Json(#[from] serde_json::Error), + #[error("{0}")] + Other(String), +} diff --git a/codex-rs/api-client/src/lib.rs b/codex-rs/api-client/src/lib.rs index d755580a35..1b135bee00 100644 --- a/codex-rs/api-client/src/lib.rs +++ b/codex-rs/api-client/src/lib.rs @@ -1,8 +1,34 @@ +pub mod api; +pub mod auth; +pub mod chat; +pub mod error; +pub mod model_provider; pub mod prompt; +pub mod responses; pub mod stream; +pub use crate::api::ApiClient; +pub use crate::auth::AuthContext; +pub use crate::auth::AuthProvider; +pub use crate::chat::AggregateStreamExt; +pub use crate::chat::ChatAggregationMode; +pub use crate::chat::ChatCompletionsApiClient; +pub use crate::chat::ChatCompletionsApiClientConfig; +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::responses::ResponsesApiClient; +pub use crate::responses::ResponsesApiClientConfig; +pub use crate::responses::stream_from_fixture; pub use crate::stream::Reasoning; +pub use crate::stream::ResponseEvent; +pub use crate::stream::ResponseStream; pub use crate::stream::TextControls; pub use crate::stream::TextFormat; pub use crate::stream::TextFormatType; diff --git a/codex-rs/core/src/model_provider_info.rs b/codex-rs/api-client/src/model_provider.rs similarity index 64% rename from codex-rs/core/src/model_provider_info.rs rename to codex-rs/api-client/src/model_provider.rs index 8dc252aa7c..c2c2e82093 100644 --- a/codex-rs/core/src/model_provider_info.rs +++ b/codex-rs/api-client/src/model_provider.rs @@ -1,136 +1,79 @@ -//! Registry of model providers supported by Codex. -//! -//! 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. - -use crate::CodexAuth; -use crate::default_client::CodexHttpClient; -use crate::default_client::CodexRequestBuilder; -use codex_app_server_protocol::AuthMode; -use serde::Deserialize; -use serde::Serialize; use std::collections::HashMap; use std::env::VarError; use std::time::Duration; -use crate::error::EnvVarError; +use codex_app_server_protocol::AuthMode; +use serde::Deserialize; +use serde::Serialize; + +use crate::auth::AuthContext; +use crate::error::Error; +use crate::error::Result; + const DEFAULT_STREAM_IDLE_TIMEOUT_MS: u64 = 300_000; const DEFAULT_STREAM_MAX_RETRIES: u64 = 5; const DEFAULT_REQUEST_MAX_RETRIES: u64 = 4; -/// Hard cap for user-configured `stream_max_retries`. const MAX_STREAM_MAX_RETRIES: u64 = 100; -/// Hard cap for user-configured `request_max_retries`. const MAX_REQUEST_MAX_RETRIES: u64 = 100; +const DEFAULT_OLLAMA_PORT: u32 = 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. #[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)] #[serde(rename_all = "lowercase")] pub enum WireApi { - /// The Responses API exposed by OpenAI at `/v1/responses`. Responses, - - /// Regular Chat Completions compatible with `/v1/chat/completions`. #[default] Chat, } -/// Serializable representation of a provider definition. #[derive(Debug, Clone, Deserialize, Serialize, PartialEq)] pub struct ModelProviderInfo { - /// Friendly display name. pub name: String, - /// Base URL for the provider's OpenAI-compatible API. 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. 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. 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. 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. pub env_http_headers: Option>, - - /// Maximum number of times to retry a failed HTTP request to this provider. 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. pub stream_idle_timeout_ms: Option, - - /// Does this provider require an OpenAI API Key or ChatGPT login token? If true, - /// user is presented with login screen on first run, and login preference and token/key - /// are stored in auth.json. If false (which is the default), login screen is skipped, - /// and API key (if needed) comes from the "env_key" environment variable. #[serde(default)] pub requires_openai_auth: bool, } impl ModelProviderInfo { - /// Construct a `POST` RequestBuilder for the given URL using the provided - /// [`CodexHttpClient`] applying: - /// • provider-specific headers (static + env 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/empty, returns an [`Err`] identical to the - /// one produced by [`ModelProviderInfo::api_key`]. - pub async fn create_request_builder<'a>( - &'a self, - client: &'a CodexHttpClient, - auth: &Option, - ) -> crate::error::Result { + 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(CodexAuth::from_api_key(secret_key)) + Some(AuthContext { + mode: AuthMode::ApiKey, + bearer_token: Some(secret_key.clone()), + account_id: None, + }) } else { - match self.api_key() { - Ok(Some(key)) => Some(CodexAuth::from_api_key(&key)), - Ok(None) => auth.clone(), - Err(err) => { - if auth.is_some() { - auth.clone() - } else { - return Err(err); - } - } + 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); - + let url = self.get_full_url(effective_auth.as_ref()); let mut builder = client.post(url); - if let Some(auth) = effective_auth.as_ref() { - builder = builder.bearer_auth(auth.get_token().await?); + if let Some(context) = effective_auth.as_ref() { + if let Some(token) = context.bearer_token.as_ref() { + builder = builder.bearer_auth(token); + } } Ok(self.apply_http_headers(builder)) @@ -149,10 +92,10 @@ impl ModelProviderInfo { }) } - pub(crate) fn get_full_url(&self, auth: &Option) -> String { + pub fn get_full_url(&self, auth: Option<&AuthContext>) -> String { let default_base_url = if matches!( auth, - Some(CodexAuth { + Some(AuthContext { mode: AuthMode::ChatGPT, .. }) @@ -165,7 +108,7 @@ impl ModelProviderInfo { let base_url = self .base_url .clone() - .unwrap_or(default_base_url.to_string()); + .unwrap_or_else(|| default_base_url.to_string()); match self.wire_api { WireApi::Responses => format!("{base_url}/responses{query_string}"), @@ -173,7 +116,7 @@ impl ModelProviderInfo { } } - pub(crate) fn is_azure_responses_endpoint(&self) -> bool { + pub fn is_azure_responses_endpoint(&self) -> bool { if self.wire_api != WireApi::Responses { return false; } @@ -188,10 +131,7 @@ impl ModelProviderInfo { .unwrap_or(false) } - /// Apply provider-specific HTTP headers (both static and environment-based) - /// onto an existing [`CodexRequestBuilder`] and return the updated - /// builder. - fn apply_http_headers(&self, mut builder: CodexRequestBuilder) -> CodexRequestBuilder { + 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); @@ -210,10 +150,7 @@ impl ModelProviderInfo { builder } - /// If `env_key` is Some, returns the API key for this provider if present - /// (and non-empty) in the environment. If `env_key` is required but - /// cannot be found, returns an error. - pub fn api_key(&self) -> crate::error::Result> { + pub fn api_key(&self) -> Result> { match &self.env_key { Some(env_key) => { let env_value = std::env::var(env_key); @@ -225,32 +162,27 @@ impl ModelProviderInfo { Ok(Some(v)) } }) - .map_err(|_| { - crate::error::CodexErr::EnvVar(EnvVarError { - var: env_key.clone(), - instructions: self.env_key_instructions.clone(), - }) + .map_err(|_| Error::MissingEnvVar { + var: env_key.clone(), + instructions: self.env_key_instructions.clone(), }) } None => Ok(None), } } - /// Effective maximum number of request retries for this provider. pub fn request_max_retries(&self) -> u64 { self.request_max_retries .unwrap_or(DEFAULT_REQUEST_MAX_RETRIES) .min(MAX_REQUEST_MAX_RETRIES) } - /// Effective maximum number of stream reconnection attempts for this provider. pub fn stream_max_retries(&self) -> u64 { self.stream_max_retries .unwrap_or(DEFAULT_STREAM_MAX_RETRIES) .min(MAX_STREAM_MAX_RETRIES) } - /// Effective idle timeout for streaming responses. pub fn stream_idle_timeout(&self) -> Duration { self.stream_idle_timeout_ms .map(Duration::from_millis) @@ -258,28 +190,16 @@ impl ModelProviderInfo { } } -const DEFAULT_OLLAMA_PORT: u32 = 11434; - pub const BUILT_IN_OSS_MODEL_PROVIDER_ID: &str = "oss"; -/// Built-in default provider list. pub fn built_in_model_providers() -> HashMap { use ModelProviderInfo as P; - // We do not want to be in the business of adjucating which third-party - // providers are bundled with Codex CLI, so we only include the OpenAI and - // open source ("oss") providers by default. Users are encouraged to add to - // `model_providers` in config.toml to add their own providers. [ ( "openai", P { name: "OpenAI".into(), - // Allow users to override the default OpenAI endpoint by - // exporting `OPENAI_BASE_URL`. This is useful when pointing - // Codex at a proxy, mock server, or Azure-style deployment - // without requiring a full TOML override for the built-in - // OpenAI provider. base_url: std::env::var("OPENAI_BASE_URL") .ok() .filter(|v| !v.trim().is_empty()), @@ -304,7 +224,6 @@ pub fn built_in_model_providers() -> HashMap { .into_iter() .collect(), ), - // Use global defaults for retry/timeout unless overridden in config.toml. request_max_retries: None, stream_max_retries: None, stream_idle_timeout_ms: None, @@ -319,8 +238,6 @@ pub fn built_in_model_providers() -> HashMap { } pub fn create_oss_provider() -> ModelProviderInfo { - // These CODEX_OSS_ environment variables are experimental: we may - // switch to reading values from config.toml instead. let codex_oss_base_url = match std::env::var("CODEX_OSS_BASE_URL") .ok() .filter(|v| !v.trim().is_empty()) @@ -366,23 +283,23 @@ fn matches_azure_responses_base_url(base_url: &str) -> bool { "azure-api.", "azurefd.", ]; - AZURE_MARKERS.iter().any(|marker| base.contains(marker)) + AZURE_MARKERS.iter().any(|needle| base.contains(needle)) } #[cfg(test)] mod tests { use super::*; - use pretty_assertions::assert_eq; + use maplit::hashmap; #[test] - fn test_deserialize_ollama_model_provider_toml() { + fn deserializes_defaults_without_optional_fields() { let azure_provider_toml = r#" -name = "Ollama" -base_url = "http://localhost:11434/v1" +name = "Azure" +base_url = "https://xxxxx.openai.azure.com/openai" "#; let expected_provider = ModelProviderInfo { - name: "Ollama".into(), - base_url: Some("http://localhost:11434/v1".into()), + name: "Azure".into(), + base_url: Some("https://xxxxx.openai.azure.com/openai".into()), env_key: None, env_key_instructions: None, experimental_bearer_token: None, @@ -415,7 +332,7 @@ query_params = { api-version = "2025-04-01-preview" } env_key_instructions: None, experimental_bearer_token: None, wire_api: WireApi::Chat, - query_params: Some(maplit::hashmap! { + query_params: Some(hashmap! { "api-version".to_string() => "2025-04-01-preview".to_string(), }), http_headers: None, @@ -447,10 +364,10 @@ env_http_headers = { "X-Example-Env-Header" = "EXAMPLE_ENV_VAR" } experimental_bearer_token: None, wire_api: WireApi::Chat, query_params: None, - http_headers: Some(maplit::hashmap! { + http_headers: Some(hashmap! { "X-Example-Header".to_string() => "example-value".to_string(), }), - env_http_headers: Some(maplit::hashmap! { + env_http_headers: Some(hashmap! { "X-Example-Env-Header".to_string() => "EXAMPLE_ENV_VAR".to_string(), }), request_max_retries: None, @@ -516,16 +433,12 @@ env_http_headers = { "X-Example-Env-Header" = "EXAMPLE_ENV_VAR" } }; assert!(named_provider.is_azure_responses_endpoint()); - let negative_cases = [ - "https://api.openai.com/v1", - "https://example.com/openai", - "https://myproxy.azurewebsites.net/openai", - ]; + let negative_cases = ["https://api.openai.com/v1", "https://example.com"]; for base_url in negative_cases { let provider = provider_for(base_url); assert!( !provider.is_azure_responses_endpoint(), - "expected {base_url} not to be detected as Azure" + "expected {base_url} to be non-Azure" ); } } diff --git a/codex-rs/api-client/src/prompt.rs b/codex-rs/api-client/src/prompt.rs index 4b15a9aa63..5dce2980e7 100644 --- a/codex-rs/api-client/src/prompt.rs +++ b/codex-rs/api-client/src/prompt.rs @@ -1,31 +1,52 @@ use codex_protocol::models::ResponseItem; -use serde::Serialize; +use codex_protocol::protocol::SessionSource; use serde_json::Value; -#[derive(Debug, Clone, Default, Serialize)] +use crate::Reasoning; +use crate::TextControls; + +#[derive(Debug, Clone, Default)] pub struct Prompt { - #[serde(skip_serializing)] + pub instructions: String, pub input: Vec, - #[serde(skip_serializing)] pub tools: Vec, - #[serde(skip_serializing)] pub parallel_tool_calls: bool, - #[serde(skip_serializing)] pub output_schema: Option, + pub reasoning: Option, + pub text_controls: Option, + pub store_response: bool, + pub prompt_cache_key: Option, + pub previous_response_id: 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, + store_response: bool, + prompt_cache_key: Option, + previous_response_id: Option, + session_source: Option, ) -> Self { Self { + instructions, input, tools, parallel_tool_calls, output_schema, + reasoning, + text_controls, + store_response, + prompt_cache_key, + previous_response_id, + session_source, } } } diff --git a/codex-rs/api-client/src/responses.rs b/codex-rs/api-client/src/responses.rs new file mode 100644 index 0000000000..7e10adab0f --- /dev/null +++ b/codex-rs/api-client/src/responses.rs @@ -0,0 +1,741 @@ +use std::io::BufRead; +use std::path::Path; +use std::sync::Arc; +use std::time::Duration; + +use async_trait::async_trait; +use bytes::Bytes; +use codex_app_server_protocol::AuthMode; +use codex_otel::otel_event_manager::OtelEventManager; +use codex_protocol::ConversationId; +use codex_protocol::models::ResponseItem; +use codex_protocol::protocol::RateLimitSnapshot; +use codex_protocol::protocol::RateLimitWindow; +use codex_protocol::protocol::SessionSource; +use codex_protocol::protocol::SubAgentSource; +use codex_protocol::protocol::TokenUsage; +use eventsource_stream::Eventsource; +use futures::Stream; +use futures::StreamExt; +use futures::TryStreamExt; +use once_cell::sync::OnceCell; +use regex_lite::Regex; +use reqwest::StatusCode; +use reqwest::header::HeaderMap; +use serde::Deserialize; +use serde::Serialize; +use serde_json::Value; +use serde_json::json; +use tokio::sync::mpsc; +use tokio::time::timeout; +use tokio_util::io::ReaderStream; +use tracing::debug; +use tracing::trace; + +use crate::api::ApiClient; +use crate::auth::AuthProvider; +use crate::error::Error; +use crate::model_provider::ModelProviderInfo; +use crate::prompt::Prompt; +use crate::stream::ResponseEvent; +use crate::stream::ResponseStream; + +type Result = std::result::Result; + +#[derive(Clone)] +pub struct ResponsesApiClientConfig { + pub http_client: reqwest::Client, + pub provider: ModelProviderInfo, + pub model: String, + pub conversation_id: ConversationId, + pub auth_provider: Option>, + pub otel_event_manager: OtelEventManager, +} + +#[derive(Clone)] +pub struct ResponsesApiClient { + config: ResponsesApiClientConfig, +} + +#[async_trait] +impl ApiClient for ResponsesApiClient { + type Config = ResponsesApiClientConfig; + + async 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 { + return Err(Error::UnsupportedOperation( + "ResponsesApiClient requires a Responses provider".to_string(), + )); + } + + let mut payload_json = self.build_payload(&prompt)?; + + if self.config.provider.is_azure_responses_endpoint() { + if let Some(input_value) = payload_json.get_mut("input") + && let Some(array) = input_value.as_array_mut() + { + attach_item_ids_array(array, &prompt.input); + } + } + + 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 { + fn build_payload(&self, prompt: &Prompt) -> Result { + let azure_workaround = self.config.provider.is_azure_responses_endpoint(); + + let mut payload = json!({ + "model": self.config.model, + "instructions": prompt.instructions, + "input": prompt.input, + "tools": prompt.tools, + "tool_choice": "auto", + "parallel_tool_calls": prompt.parallel_tool_calls, + "store": azure_workaround || prompt.store_response, + "stream": true, + "prompt_cache_key": prompt + .prompt_cache_key + .clone() + .unwrap_or_else(|| self.config.conversation_id.to_string()), + }); + + if let Some(reasoning) = prompt.reasoning.as_ref() { + payload + .as_object_mut() + .expect("payload object") + .insert("reasoning".to_string(), serde_json::to_value(reasoning)?); + } + + if let Some(text) = prompt.text_controls.as_ref() { + payload + .as_object_mut() + .expect("payload object") + .insert("text".to_string(), serde_json::to_value(text)?); + } + + if let Some(prev) = prompt.previous_response_id.as_ref() { + payload.as_object_mut().expect("payload object").insert( + "previous_response_id".to_string(), + Value::String(prev.clone()), + ); + } + + let include = if prompt.reasoning.is_some() { + vec!["reasoning.encrypted_content".to_string()] + } else { + Vec::new() + }; + payload.as_object_mut().expect("payload object").insert( + "include".to_string(), + Value::Array(include.into_iter().map(Value::String).collect()), + ); + + Ok(payload) + } + + async fn attempt_stream_responses( + &self, + attempt: u64, + prompt: &Prompt, + payload_json: &Value, + ) -> std::result::Result { + let auth = match &self.config.auth_provider { + Some(provider) => provider.auth_context().await, + None => None, + }; + + trace!( + "POST to {}: {:?}", + self.config.provider.get_full_url(auth.as_ref()), + serde_json::to_string(payload_json) + .unwrap_or_else(|_| "".to_string()) + ); + + let mut req_builder = self + .config + .provider + .create_request_builder(&self.config.http_client, &auth) + .await + .map_err(StreamAttemptError::Fatal)?; + + if let Some(SessionSource::SubAgent(sub)) = prompt.session_source.as_ref() { + let subagent = match sub { + SubAgentSource::Other(label) => label.clone(), + other => serde_json::to_value(other) + .ok() + .and_then(|v| v.as_str().map(ToString::to_string)) + .unwrap_or_else(|| "other".to_string()), + }; + req_builder = req_builder.header("x-openai-subagent", subagent); + } + + req_builder = req_builder + .header("conversation_id", self.config.conversation_id.to_string()) + .header("session_id", self.config.conversation_id.to_string()) + .header(reqwest::header::ACCEPT, "text/event-stream") + .json(payload_json); + + if let Some(ctx) = auth.as_ref() + && ctx.mode == AuthMode::ChatGPT + && let Some(account_id) = ctx.account_id.as_ref() + { + req_builder = req_builder.header("chatgpt-account-id", account_id); + } + + let res = self + .config + .otel_event_manager + .log_request(attempt, || req_builder.send()) + .await; + + 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(|s| s.to_string()); + } + + match res { + Ok(resp) if resp.status().is_success() => { + let (tx_event, rx_event) = mpsc::channel::>(1600); + + if let Some(snapshot) = parse_rate_limit_snapshot(resp.headers()) + && tx_event + .send(Ok(ResponseEvent::RateLimits(snapshot))) + .await + .is_err() + { + debug!("receiver dropped rate limit snapshot event"); + } + + let request_id_for_stream = request_id.clone(); + let stream = resp + .bytes_stream() + .map_err(move |err| Error::ResponseStreamFailed { + source: err, + request_id: request_id_for_stream.clone(), + }); + tokio::spawn(process_sse( + stream, + tx_event, + self.config.provider.stream_idle_timeout(), + self.config.otel_event_manager.clone(), + )); + + Ok(ResponseStream { rx_event }) + } + Ok(res) => { + let status = res.status(); + + let retry_after_secs = res + .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(|s| Duration::from_millis(s * 1_000)); + + if status == StatusCode::UNAUTHORIZED { + if let Some(provider) = self.config.auth_provider.as_ref() + && let Some(ctx) = auth.as_ref() + && ctx.mode == AuthMode::ChatGPT + { + provider + .refresh_token() + .await + .map_err(|err| StreamAttemptError::Fatal(Error::Auth(err)))?; + } + } + + if !(status == StatusCode::TOO_MANY_REQUESTS + || status == StatusCode::UNAUTHORIZED + || status.is_server_error()) + { + // Surface error body. + let body = res + .text() + .await + .unwrap_or_else(|_| "".to_string()); + return Err(StreamAttemptError::Fatal(Error::UnexpectedStatus { + status, + body, + })); + } + + Err(StreamAttemptError::RetryableHttpError { + status, + retry_after, + request_id, + }) + } + Err(err) => Err(StreamAttemptError::RetryableTransportError(Error::Http( + err, + ))), + } + } +} + +enum StreamAttemptError { + RetryableHttpError { + status: StatusCode, + retry_after: Option, + request_id: Option, + }, + RetryableTransportError(Error), + Fatal(Error), +} + +impl StreamAttemptError { + fn delay(&self, attempt: u64) -> Duration { + let backoff_attempt = attempt + 1; + match self { + StreamAttemptError::RetryableHttpError { retry_after, .. } => { + retry_after.unwrap_or_else(|| backoff(backoff_attempt)) + } + StreamAttemptError::RetryableTransportError { .. } => backoff(backoff_attempt), + StreamAttemptError::Fatal(_) => Duration::from_secs(0), + } + } + + fn into_error(self) -> Error { + match self { + StreamAttemptError::RetryableHttpError { + status, request_id, .. + } => Error::RetryLimit { status, request_id }, + StreamAttemptError::RetryableTransportError(error) => error, + StreamAttemptError::Fatal(error) => error, + } + } +} + +#[derive(Debug, Deserialize, Serialize)] +struct SseEvent { + #[serde(rename = "type")] + kind: String, + response: Option, + item: Option, + delta: Option, +} + +#[derive(Debug, Deserialize)] +struct ResponseCompleted { + id: String, + usage: Option, +} + +#[derive(Debug, Deserialize)] +struct ResponseCompletedUsage { + input_tokens: i64, + input_tokens_details: Option, + output_tokens: i64, + output_tokens_details: Option, + total_tokens: i64, +} + +impl From for TokenUsage { + fn from(val: ResponseCompletedUsage) -> Self { + TokenUsage { + input_tokens: val.input_tokens, + cached_input_tokens: val + .input_tokens_details + .map(|d| d.cached_tokens) + .unwrap_or(0), + output_tokens: val.output_tokens, + reasoning_output_tokens: val + .output_tokens_details + .map(|d| d.reasoning_tokens) + .unwrap_or(0), + total_tokens: val.total_tokens, + } + } +} + +#[derive(Debug, Deserialize)] +struct ResponseCompletedInputTokensDetails { + cached_tokens: i64, +} + +#[derive(Debug, Deserialize)] +struct ResponseCompletedOutputTokensDetails { + reasoning_tokens: i64, +} + +fn attach_item_ids_array(items: &mut [Value], original_items: &[ResponseItem]) { + for (value, item) in items.iter_mut().zip(original_items.iter()) { + if let ResponseItem::Reasoning { id, .. } + | ResponseItem::Message { id: Some(id), .. } + | ResponseItem::WebSearchCall { id: Some(id), .. } + | ResponseItem::FunctionCall { id: Some(id), .. } + | ResponseItem::LocalShellCall { id: Some(id), .. } + | ResponseItem::CustomToolCall { id: Some(id), .. } + | ResponseItem::CustomToolCallOutput { call_id: id, .. } + | ResponseItem::FunctionCallOutput { call_id: id, .. } = item + { + if id.is_empty() { + continue; + } + + if let Some(obj) = value.as_object_mut() { + obj.insert("id".to_string(), Value::String(id.clone())); + } + } + } +} + +fn parse_rate_limit_snapshot(headers: &HeaderMap) -> Option { + let primary = parse_rate_limit_window( + headers, + "x-codex-primary-used-percent", + "x-codex-primary-window-minutes", + "x-codex-primary-reset-at", + ); + + let secondary = parse_rate_limit_window( + headers, + "x-codex-secondary-used-percent", + "x-codex-secondary-window-minutes", + "x-codex-secondary-reset-at", + ); + + Some(RateLimitSnapshot { primary, secondary }) +} + +fn parse_rate_limit_window( + headers: &HeaderMap, + used_percent_header: &str, + window_minutes_header: &str, + resets_at_header: &str, +) -> Option { + let used_percent: Option = parse_header_f64(headers, used_percent_header); + + used_percent.and_then(|used_percent| { + let window_minutes = parse_header_i64(headers, window_minutes_header); + let resets_at = parse_header_i64(headers, resets_at_header); + + let has_data = used_percent != 0.0 + || window_minutes.is_some_and(|minutes| minutes != 0) + || resets_at.is_some(); + + has_data.then_some(RateLimitWindow { + used_percent, + window_minutes, + resets_at, + }) + }) +} + +fn parse_header_f64(headers: &HeaderMap, name: &str) -> Option { + parse_header_str(headers, name)? + .parse::() + .ok() + .filter(|v| v.is_finite()) +} + +fn parse_header_i64(headers: &HeaderMap, name: &str) -> Option { + parse_header_str(headers, name)?.parse::().ok() +} + +fn parse_header_str<'a>(headers: &'a HeaderMap, name: &str) -> Option<&'a str> { + headers.get(name)?.to_str().ok() +} + +async fn process_sse( + stream: S, + tx_event: mpsc::Sender>, + idle_timeout: Duration, + otel_event_manager: OtelEventManager, +) where + S: Stream> + Unpin + Send + 'static, +{ + let mut stream = stream.eventsource(); + + let mut response_completed: Option = None; + let mut response_error: Option = None; + + loop { + let start = std::time::Instant::now(); + let response = timeout(idle_timeout, stream.next()).await; + let duration = start.elapsed(); + otel_event_manager.log_sse_event(&response, duration); + + let sse = match response { + Ok(Some(Ok(sse))) => sse, + Ok(Some(Err(e))) => { + debug!("SSE Error: {e:#}"); + let event = Error::Stream(e.to_string(), None); + let _ = tx_event.send(Err(event)).await; + return; + } + Ok(None) => { + match response_completed { + Some(ResponseCompleted { + id: response_id, + usage, + }) => { + if let Some(token_usage) = &usage { + otel_event_manager.sse_event_completed( + token_usage.input_tokens, + token_usage.output_tokens, + token_usage + .input_tokens_details + .as_ref() + .map(|d| d.cached_tokens), + token_usage + .output_tokens_details + .as_ref() + .map(|d| d.reasoning_tokens), + token_usage.total_tokens, + ); + } + let event = ResponseEvent::Completed { + response_id, + token_usage: usage.map(Into::into), + }; + let _ = tx_event.send(Ok(event)).await; + } + None => { + let error = response_error.unwrap_or(Error::Stream( + "stream closed before response.completed".into(), + None, + )); + otel_event_manager.see_event_completed_failed(&error); + + let _ = tx_event.send(Err(error)).await; + } + } + return; + } + Err(_) => { + let _ = tx_event + .send(Err(Error::Stream( + "idle timeout waiting for SSE".into(), + None, + ))) + .await; + return; + } + }; + + let raw = sse.data.clone(); + trace!("SSE event: {}", raw); + + let event: SseEvent = match serde_json::from_str(&sse.data) { + Ok(event) => event, + Err(e) => { + debug!("Failed to parse SSE event: {e}, data: {}", &sse.data); + continue; + } + }; + + match event.kind.as_str() { + "response.output_item.done" => { + let Some(item_val) = event.item else { continue }; + let Ok(item) = serde_json::from_value::(item_val) else { + debug!("failed to parse ResponseItem from output_item.done"); + continue; + }; + + let event = ResponseEvent::OutputItemDone(item); + if tx_event.send(Ok(event)).await.is_err() { + return; + } + } + "response.output_text.delta" => { + if let Some(delta) = event.delta { + let event = ResponseEvent::OutputTextDelta(delta); + if tx_event.send(Ok(event)).await.is_err() { + return; + } + } + } + "response.reasoning_summary_text.delta" => { + if let Some(delta) = event.delta { + let event = ResponseEvent::ReasoningSummaryDelta(delta); + if tx_event.send(Ok(event)).await.is_err() { + return; + } + } + } + "response.reasoning_text.delta" => { + if let Some(delta) = event.delta { + let event = ResponseEvent::ReasoningContentDelta(delta); + if tx_event.send(Ok(event)).await.is_err() { + return; + } + } + } + "response.created" => { + if event.response.is_some() { + let _ = tx_event.send(Ok(ResponseEvent::Created)).await; + } + } + "response.failed" => { + if let Some(resp_val) = event.response { + response_error = Some(Error::Stream( + "response.failed event received".to_string(), + None, + )); + + if let Some(error) = resp_val.get("error") { + match serde_json::from_value::(error.clone()) { + Ok(error) => { + if is_context_window_error(&error) { + response_error = Some(Error::UnsupportedOperation( + "context window exceeded".to_string(), + )); + } else { + let delay = try_parse_retry_after(&error); + let message = error.message.clone().unwrap_or_default(); + response_error = Some(Error::Stream(message, delay)); + } + } + Err(e) => { + let error = format!("failed to parse ErrorResponse: {e}"); + debug!(error); + response_error = Some(Error::Stream(error, None)) + } + } + } + } + } + "response.completed" => { + if let Some(resp_val) = event.response { + match serde_json::from_value::(resp_val) { + Ok(r) => { + response_completed = Some(r); + } + Err(e) => { + let error = format!("failed to parse ResponseCompleted: {e}"); + debug!(error); + response_error = Some(Error::Stream(error, None)); + continue; + } + }; + }; + } + "response.output_item.added" => { + let Some(item_val) = event.item else { continue }; + let Ok(item) = serde_json::from_value::(item_val) else { + debug!("failed to parse ResponseItem from output_item.done"); + continue; + }; + + let event = ResponseEvent::OutputItemAdded(item); + if tx_event.send(Ok(event)).await.is_err() { + return; + } + } + "response.reasoning_summary_part.added" => { + let event = ResponseEvent::ReasoningSummaryPartAdded; + if tx_event.send(Ok(event)).await.is_err() { + return; + } + } + _ => {} + } + } +} + +#[derive(Debug, Deserialize)] +struct ErrorResponse { + code: Option, + message: Option, +} + +fn backoff(attempt: u64) -> Duration { + let exponent = attempt.min(6) as u32; + let base = 2u64.pow(exponent); + Duration::from_millis(base * 100) +} + +fn rate_limit_regex() -> &'static Regex { + static RE: OnceCell = OnceCell::new(); + + RE.get_or_init(|| Regex::new(r"Please try again in (\d+(?:\.\d+)?)(s|ms)").unwrap()) +} + +fn try_parse_retry_after(err: &ErrorResponse) -> Option { + if err.code.as_deref() != Some("rate_limit_exceeded") { + return None; + } + + let re = rate_limit_regex(); + if let Some(message) = &err.message + && let Some(captures) = re.captures(message) + { + let seconds = captures.get(1); + let unit = captures.get(2); + + if let (Some(value), Some(unit)) = (seconds, unit) { + let value = value.as_str().parse::().ok()?; + let unit = unit.as_str(); + + if unit == "s" { + return Some(Duration::from_secs_f64(value)); + } else if unit == "ms" { + return Some(Duration::from_millis(value as u64)); + } + } + } + None +} + +fn is_context_window_error(error: &ErrorResponse) -> bool { + error.code.as_deref() == Some("context_length_exceeded") +} + +/// used in tests to stream from a text SSE file +pub async fn stream_from_fixture( + path: impl AsRef, + provider: ModelProviderInfo, + otel_event_manager: OtelEventManager, +) -> Result { + let (tx_event, rx_event) = mpsc::channel::>(1600); + let display_path = path.as_ref().display().to_string(); + let file = std::fs::File::open(path.as_ref()) + .map_err(|e| Error::Other(format!("failed to open fixture {}: {}", display_path, e)))?; + let lines = std::io::BufReader::new(file).lines(); + + let mut content = String::new(); + for line in lines { + let line = line + .map_err(|e| Error::Other(format!("failed to read fixture {}: {}", display_path, e)))?; + content.push_str(&line); + content.push_str("\n\n"); + } + + let rdr = std::io::Cursor::new(content); + let stream = ReaderStream::new(rdr).map_err(|e| Error::Other(e.to_string())); + tokio::spawn(process_sse( + stream, + tx_event, + provider.stream_idle_timeout(), + otel_event_manager, + )); + Ok(ResponseStream { rx_event }) +} diff --git a/codex-rs/api-client/src/stream.rs b/codex-rs/api-client/src/stream.rs index dcb5fc1dd2..6054aa62f5 100644 --- a/codex-rs/api-client/src/stream.rs +++ b/codex-rs/api-client/src/stream.rs @@ -1,9 +1,20 @@ +use std::pin::Pin; +use std::task::Context; +use std::task::Poll; + use codex_protocol::config_types::ReasoningEffort as ReasoningEffortConfig; use codex_protocol::config_types::ReasoningSummary as ReasoningSummaryConfig; +use codex_protocol::models::ResponseItem; +use codex_protocol::protocol::RateLimitSnapshot; +use codex_protocol::protocol::TokenUsage; +use futures::Stream; use serde::Serialize; use serde_json::Value; +use tokio::sync::mpsc; -#[derive(Debug, Serialize)] +use crate::error::Result; + +#[derive(Debug, Serialize, Clone)] pub struct Reasoning { #[serde(skip_serializing_if = "Option::is_none")] pub effort: Option, @@ -33,3 +44,31 @@ pub struct TextControls { #[serde(skip_serializing_if = "Option::is_none")] pub format: Option, } + +#[derive(Debug)] +pub enum ResponseEvent { + Created, + OutputItemDone(ResponseItem), + OutputItemAdded(ResponseItem), + Completed { + response_id: String, + token_usage: Option, + }, + OutputTextDelta(String), + ReasoningSummaryDelta(String), + ReasoningContentDelta(String), + ReasoningSummaryPartAdded, + RateLimits(RateLimitSnapshot), +} + +pub struct ResponseStream { + pub(crate) rx_event: mpsc::Receiver>, +} + +impl Stream for ResponseStream { + type Item = Result; + + fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + self.rx_event.poll_recv(cx) + } +} diff --git a/codex-rs/core/src/chat_completions.rs b/codex-rs/core/src/chat_completions.rs deleted file mode 100644 index 6590900218..0000000000 --- a/codex-rs/core/src/chat_completions.rs +++ /dev/null @@ -1,966 +0,0 @@ -use std::time::Duration; - -use crate::ModelProviderInfo; -use crate::client::StreamPayload; -use crate::client_common::ResponseEvent; -use crate::client_common::ResponseStream; -use crate::default_client::CodexHttpClient; -use crate::error::CodexErr; -use crate::error::ConnectionFailedError; -use crate::error::ResponseStreamFailed; -use crate::error::Result; -use crate::error::RetryLimitReachedError; -use crate::error::UnexpectedResponseError; -use crate::model_family::ModelFamily; -use crate::tools::spec::create_tools_json_for_chat_completions_api; -use crate::util::backoff; -use bytes::Bytes; -use codex_otel::otel_event_manager::OtelEventManager; -use codex_protocol::models::ContentItem; -use codex_protocol::models::FunctionCallOutputContentItem; -use codex_protocol::models::ReasoningItemContent; -use codex_protocol::models::ResponseItem; -use codex_protocol::protocol::SessionSource; -use codex_protocol::protocol::SubAgentSource; -use eventsource_stream::Eventsource; -use futures::Stream; -use futures::StreamExt; -use futures::TryStreamExt; -use reqwest::StatusCode; -use serde_json::json; -use std::pin::Pin; -use std::task::Context; -use std::task::Poll; -use tokio::sync::mpsc; -use tokio::time::timeout; -use tracing::debug; -use tracing::trace; - -/// Implementation for the classic Chat Completions API. -pub(crate) async fn stream_chat_completions( - payload: &StreamPayload, - model_family: &ModelFamily, - client: &CodexHttpClient, - provider: &ModelProviderInfo, - otel_event_manager: &OtelEventManager, - session_source: &SessionSource, -) -> Result { - if payload.prompt.output_schema.is_some() { - return Err(CodexErr::UnsupportedOperation( - "output_schema is not supported for Chat Completions API".to_string(), - )); - } - - // Build messages array - let mut messages = Vec::::new(); - - messages.push(json!({"role": "system", "content": payload.instructions})); - - let input = payload.prompt.input.clone(); - - // Pre-scan: map Reasoning blocks to the adjacent assistant anchor after the last user. - // - If the last emitted message is a user message, drop all reasoning. - // - Otherwise, for each Reasoning item after the last user message, attach it - // to the immediate previous assistant message (stop turns) or the immediate - // next assistant anchor (tool-call turns: function/local shell call, or assistant message). - let mut reasoning_by_anchor_index: std::collections::HashMap = - std::collections::HashMap::new(); - - // Determine the last role that would be emitted to Chat Completions. - let mut last_emitted_role: Option<&str> = None; - for item in &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 { .. } => {} - } - } - - // Find the last user message index in the input. - let mut last_user_index: Option = None; - for (idx, item) in input.iter().enumerate() { - if let ResponseItem::Message { role, .. } = item - && role == "user" - { - last_user_index = Some(idx); - } - } - - // Attach reasoning only if the conversation does not end with a user message. - if !matches!(last_emitted_role, Some("user")) { - for (idx, item) in input.iter().enumerate() { - // Only consider reasoning that appears after the last user message. - 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; - } - - // Prefer immediate previous assistant message (stop turns) - let mut attached = false; - if idx > 0 - && let ResponseItem::Message { role, .. } = &input[idx - 1] - && role == "assistant" - { - reasoning_by_anchor_index - .entry(idx - 1) - .and_modify(|v| v.push_str(&text)) - .or_insert(text.clone()); - attached = true; - } - - // Otherwise, attach to immediate next assistant anchor (tool-calls or assistant message) - if !attached && idx + 1 < input.len() { - match &input[idx + 1] { - ResponseItem::FunctionCall { .. } | ResponseItem::LocalShellCall { .. } => { - reasoning_by_anchor_index - .entry(idx + 1) - .and_modify(|v| v.push_str(&text)) - .or_insert(text.clone()); - } - ResponseItem::Message { role, .. } if role == "assistant" => { - reasoning_by_anchor_index - .entry(idx + 1) - .and_modify(|v| v.push_str(&text)) - .or_insert(text.clone()); - } - _ => {} - } - } - } - } - } - - // Track last assistant text we emitted to avoid duplicate assistant messages - // in the outbound Chat Completions payload (can happen if a final - // aggregated assistant message was recorded alongside an earlier partial). - let mut last_assistant_text: Option = None; - - for (idx, item) in input.iter().enumerate() { - match item { - ResponseItem::Message { role, content, .. } => { - // Build content either as a plain string (typical for assistant text) - // or as an array of content items when images are present (user/tool multimodal). - 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}})); - } - } - } - - // Skip exact-duplicate assistant messages. - if role == "assistant" { - if let Some(prev) = &last_assistant_text - && prev == &text - { - continue; - } - last_assistant_text = Some(text.clone()); - } - - // For assistant messages, always send a plain string for compatibility. - // For user messages, if an image is present, send an array of content items. - let content_value = if role == "assistant" { - json!(text) - } else if saw_image { - json!(items) - } else { - json!(text) - }; - - let mut msg = json!({"role": role, "content": content_value}); - if role == "assistant" - && let Some(reasoning) = reasoning_by_anchor_index.get(&idx) - && let Some(obj) = msg.as_object_mut() - { - obj.insert("reasoning".to_string(), json!(reasoning)); - } - messages.push(msg); - } - ResponseItem::FunctionCall { - name, - arguments, - call_id, - .. - } => { - let mut msg = json!({ - "role": "assistant", - "content": null, - "tool_calls": [{ - "id": call_id, - "type": "function", - "function": { - "name": name, - "arguments": arguments, - } - }] - }); - if let Some(reasoning) = reasoning_by_anchor_index.get(&idx) - && let Some(obj) = msg.as_object_mut() - { - obj.insert("reasoning".to_string(), json!(reasoning)); - } - messages.push(msg); - } - ResponseItem::LocalShellCall { - id, - call_id: _, - status, - action, - } => { - // Confirm with API team. - let mut msg = json!({ - "role": "assistant", - "content": null, - "tool_calls": [{ - "id": id.clone().unwrap_or_else(|| "".to_string()), - "type": "local_shell_call", - "status": status, - "action": action, - }] - }); - if let Some(reasoning) = reasoning_by_anchor_index.get(&idx) - && let Some(obj) = msg.as_object_mut() - { - obj.insert("reasoning".to_string(), json!(reasoning)); - } - messages.push(msg); - } - ResponseItem::FunctionCallOutput { call_id, output } => { - // Prefer structured content items when available (e.g., images) - // otherwise fall back to the legacy plain-string content. - let content_value = if let Some(items) = &output.content_items { - let mapped: Vec = items - .iter() - .map(|it| match it { - 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::CustomToolCall { - id, - call_id: _, - name, - input, - status: _, - } => { - messages.push(json!({ - "role": "assistant", - "content": null, - "tool_calls": [{ - "id": id, - "type": "custom", - "custom": { - "name": name, - "input": input, - } - }] - })); - } - ResponseItem::CustomToolCallOutput { call_id, output } => { - messages.push(json!({ - "role": "tool", - "tool_call_id": call_id, - "content": output, - })); - } - ResponseItem::GhostSnapshot { .. } => { - // Ghost snapshots annotate history but are not sent to the model. - continue; - } - ResponseItem::Reasoning { .. } - | ResponseItem::WebSearchCall { .. } - | ResponseItem::Other => { - // Omit these items from the conversation history. - continue; - } - } - } - - let tools_json = create_tools_json_for_chat_completions_api(&payload.prompt.tools)?; - let payload = json!({ - "model": model_family.slug, - "messages": messages, - "stream": true, - "tools": tools_json, - }); - - debug!( - "POST to {}: {}", - provider.get_full_url(&None), - serde_json::to_string_pretty(&payload).unwrap_or_default() - ); - - let mut attempt = 0; - let max_retries = provider.request_max_retries(); - loop { - attempt += 1; - - let mut req_builder = provider.create_request_builder(client, &None).await?; - - // Include subagent header only for subagent sessions. - if let SessionSource::SubAgent(sub) = session_source.clone() { - let subagent = if let SubAgentSource::Other(label) = sub { - label - } else { - serde_json::to_value(&sub) - .ok() - .and_then(|v| v.as_str().map(std::string::ToString::to_string)) - .unwrap_or_else(|| "other".to_string()) - }; - req_builder = req_builder.header("x-openai-subagent", subagent); - } - - let res = otel_event_manager - .log_request(attempt, || { - req_builder - .header(reqwest::header::ACCEPT, "text/event-stream") - .json(&payload) - .send() - }) - .await; - - match res { - Ok(resp) if resp.status().is_success() => { - let (tx_event, rx_event) = mpsc::channel::>(1600); - let stream = resp.bytes_stream().map_err(|e| { - CodexErr::ResponseStreamFailed(ResponseStreamFailed { - source: e, - request_id: None, - }) - }); - tokio::spawn(process_chat_sse( - stream, - tx_event, - provider.stream_idle_timeout(), - otel_event_manager.clone(), - )); - return Ok(ResponseStream { rx_event }); - } - Ok(res) => { - let status = res.status(); - if !(status == StatusCode::TOO_MANY_REQUESTS || status.is_server_error()) { - let body = (res.text().await).unwrap_or_default(); - return Err(CodexErr::UnexpectedStatus(UnexpectedResponseError { - status, - body, - request_id: None, - })); - } - - if attempt > max_retries { - return Err(CodexErr::RetryLimit(RetryLimitReachedError { - status, - request_id: None, - })); - } - - let retry_after_secs = res - .headers() - .get(reqwest::header::RETRY_AFTER) - .and_then(|v| v.to_str().ok()) - .and_then(|s| s.parse::().ok()); - - let delay = retry_after_secs - .map(|s| Duration::from_millis(s * 1_000)) - .unwrap_or_else(|| backoff(attempt)); - tokio::time::sleep(delay).await; - } - Err(e) => { - if attempt > max_retries { - return Err(CodexErr::ConnectionFailed(ConnectionFailedError { - source: e, - })); - } - let delay = backoff(attempt); - tokio::time::sleep(delay).await; - } - } - } -} - -async fn append_assistant_text( - tx_event: &mpsc::Sender>, - assistant_item: &mut Option, - text: String, -) { - if assistant_item.is_none() { - let item = ResponseItem::Message { - id: None, - role: "assistant".to_string(), - content: vec![], - }; - *assistant_item = Some(item.clone()); - let _ = tx_event - .send(Ok(ResponseEvent::OutputItemAdded(item))) - .await; - } - - if let Some(ResponseItem::Message { content, .. }) = assistant_item { - content.push(ContentItem::OutputText { text: text.clone() }); - let _ = tx_event - .send(Ok(ResponseEvent::OutputTextDelta(text.clone()))) - .await; - } -} - -async fn append_reasoning_text( - tx_event: &mpsc::Sender>, - reasoning_item: &mut Option, - text: String, -) { - if reasoning_item.is_none() { - let item = ResponseItem::Reasoning { - id: String::new(), - summary: Vec::new(), - content: Some(vec![]), - encrypted_content: None, - }; - *reasoning_item = Some(item.clone()); - let _ = tx_event - .send(Ok(ResponseEvent::OutputItemAdded(item))) - .await; - } - - if let Some(ResponseItem::Reasoning { - content: Some(content), - .. - }) = reasoning_item - { - content.push(ReasoningItemContent::ReasoningText { text: text.clone() }); - - let _ = tx_event - .send(Ok(ResponseEvent::ReasoningContentDelta(text.clone()))) - .await; - } -} -/// Lightweight SSE processor for the Chat Completions streaming format. The -/// output is mapped onto Codex's internal [`ResponseEvent`] so that the rest -/// of the pipeline can stay agnostic of the underlying wire format. -async fn process_chat_sse( - stream: S, - tx_event: mpsc::Sender>, - idle_timeout: Duration, - otel_event_manager: OtelEventManager, -) where - S: Stream> + Unpin, -{ - let mut stream = stream.eventsource(); - - // State to accumulate a function call across streaming chunks. - // OpenAI may split the `arguments` string over multiple `delta` events - // until the chunk whose `finish_reason` is `tool_calls` is emitted. We - // keep collecting the pieces here and forward a single - // `ResponseItem::FunctionCall` once the call is complete. - #[derive(Default)] - struct FunctionCallState { - name: Option, - arguments: String, - call_id: Option, - active: bool, - } - - let mut fn_call_state = FunctionCallState::default(); - let mut assistant_item: Option = None; - let mut reasoning_item: Option = None; - - loop { - let start = std::time::Instant::now(); - let response = timeout(idle_timeout, stream.next()).await; - let duration = start.elapsed(); - otel_event_manager.log_sse_event(&response, duration); - - let sse = match response { - Ok(Some(Ok(ev))) => ev, - Ok(Some(Err(e))) => { - let _ = tx_event - .send(Err(CodexErr::Stream(e.to_string(), None))) - .await; - return; - } - Ok(None) => { - // Stream closed gracefully – emit Completed with dummy id. - let _ = tx_event - .send(Ok(ResponseEvent::Completed { - response_id: String::new(), - token_usage: None, - })) - .await; - return; - } - Err(_) => { - let _ = tx_event - .send(Err(CodexErr::Stream( - "idle timeout waiting for SSE".into(), - None, - ))) - .await; - return; - } - }; - - // OpenAI Chat streaming sends a literal string "[DONE]" when finished. - if sse.data.trim() == "[DONE]" { - // Emit any finalized items before closing so downstream consumers receive - // terminal events for both assistant content and raw reasoning. - if let Some(item) = assistant_item { - let _ = tx_event.send(Ok(ResponseEvent::OutputItemDone(item))).await; - } - - if let Some(item) = reasoning_item { - let _ = tx_event.send(Ok(ResponseEvent::OutputItemDone(item))).await; - } - - let _ = tx_event - .send(Ok(ResponseEvent::Completed { - response_id: String::new(), - token_usage: None, - })) - .await; - return; - } - - // Parse JSON chunk - let chunk: serde_json::Value = match serde_json::from_str(&sse.data) { - Ok(v) => v, - Err(_) => continue, - }; - trace!("chat_completions received SSE chunk: {chunk:?}"); - - let choice_opt = chunk.get("choices").and_then(|c| c.get(0)); - - if let Some(choice) = choice_opt { - // Handle assistant content tokens as streaming deltas. - if let Some(content) = choice - .get("delta") - .and_then(|d| d.get("content")) - .and_then(|c| c.as_str()) - && !content.is_empty() - { - append_assistant_text(&tx_event, &mut assistant_item, content.to_string()).await; - } - - // Forward any reasoning/thinking deltas if present. - // Some providers stream `reasoning` as a plain string while others - // nest the text under an object (e.g. `{ "reasoning": { "text": "…" } }`). - if let Some(reasoning_val) = choice.get("delta").and_then(|d| d.get("reasoning")) { - let mut maybe_text = reasoning_val - .as_str() - .map(str::to_string) - .filter(|s| !s.is_empty()); - - if maybe_text.is_none() && reasoning_val.is_object() { - if let Some(s) = reasoning_val - .get("text") - .and_then(|t| t.as_str()) - .filter(|s| !s.is_empty()) - { - maybe_text = Some(s.to_string()); - } else if let Some(s) = reasoning_val - .get("content") - .and_then(|t| t.as_str()) - .filter(|s| !s.is_empty()) - { - maybe_text = Some(s.to_string()); - } - } - - if let Some(reasoning) = maybe_text { - // Accumulate so we can emit a terminal Reasoning item at the end. - append_reasoning_text(&tx_event, &mut reasoning_item, reasoning).await; - } - } - - // Some providers only include reasoning on the final message object. - if let Some(message_reasoning) = choice.get("message").and_then(|m| m.get("reasoning")) - { - // Accept either a plain string or an object with { text | content } - if let Some(s) = message_reasoning.as_str() { - if !s.is_empty() { - append_reasoning_text(&tx_event, &mut reasoning_item, s.to_string()).await; - } - } else if let Some(obj) = message_reasoning.as_object() - && let Some(s) = obj - .get("text") - .and_then(|v| v.as_str()) - .or_else(|| obj.get("content").and_then(|v| v.as_str())) - && !s.is_empty() - { - append_reasoning_text(&tx_event, &mut reasoning_item, s.to_string()).await; - } - } - - // Handle streaming function / tool calls. - if let Some(tool_calls) = choice - .get("delta") - .and_then(|d| d.get("tool_calls")) - .and_then(|tc| tc.as_array()) - && let Some(tool_call) = tool_calls.first() - { - // Mark that we have an active function call in progress. - fn_call_state.active = true; - - // Extract call_id if present. - if let Some(id) = tool_call.get("id").and_then(|v| v.as_str()) { - fn_call_state.call_id.get_or_insert_with(|| id.to_string()); - } - - // Extract function details if present. - if let Some(function) = tool_call.get("function") { - if let Some(name) = function.get("name").and_then(|n| n.as_str()) { - fn_call_state.name.get_or_insert_with(|| name.to_string()); - } - - if let Some(args_fragment) = function.get("arguments").and_then(|a| a.as_str()) - { - fn_call_state.arguments.push_str(args_fragment); - } - } - } - - // Emit end-of-turn when finish_reason signals completion. - if let Some(finish_reason) = choice.get("finish_reason").and_then(|v| v.as_str()) { - match finish_reason { - "tool_calls" if fn_call_state.active => { - // First, flush the terminal raw reasoning so UIs can finalize - // the reasoning stream before any exec/tool events begin. - if let Some(item) = reasoning_item.take() { - let _ = tx_event.send(Ok(ResponseEvent::OutputItemDone(item))).await; - } - - // Then emit the FunctionCall response item. - let item = ResponseItem::FunctionCall { - id: None, - name: fn_call_state.name.clone().unwrap_or_else(|| "".to_string()), - arguments: fn_call_state.arguments.clone(), - call_id: fn_call_state.call_id.clone().unwrap_or_else(String::new), - }; - - let _ = tx_event.send(Ok(ResponseEvent::OutputItemDone(item))).await; - } - "stop" => { - // Regular turn without tool-call. Emit the final assistant message - // as a single OutputItemDone so non-delta consumers see the result. - if let Some(item) = assistant_item.take() { - let _ = tx_event.send(Ok(ResponseEvent::OutputItemDone(item))).await; - } - // Also emit a terminal Reasoning item so UIs can finalize raw reasoning. - if let Some(item) = reasoning_item.take() { - let _ = tx_event.send(Ok(ResponseEvent::OutputItemDone(item))).await; - } - } - _ => {} - } - - // Emit Completed regardless of reason so the agent can advance. - let _ = tx_event - .send(Ok(ResponseEvent::Completed { - response_id: String::new(), - token_usage: None, - })) - .await; - - // Prepare for potential next turn (should not happen in same stream). - // fn_call_state = FunctionCallState::default(); - - return; // End processing for this SSE stream. - } - } - } -} - -/// Optional client-side aggregation helper -/// -/// Stream adapter that merges the incremental `OutputItemDone` chunks coming from -/// [`process_chat_sse`] into a *running* assistant message, **suppressing the -/// per-token deltas**. The stream stays silent while the model is thinking -/// and only emits two events per turn: -/// -/// 1. `ResponseEvent::OutputItemDone` with the *complete* assistant message -/// (fully concatenated). -/// 2. The original `ResponseEvent::Completed` right after it. -/// -/// This mirrors the behaviour the TypeScript CLI exposes to its higher layers. -/// -/// The adapter is intentionally *lossless*: callers who do **not** opt in via -/// [`AggregateStreamExt::aggregate()`] keep receiving the original unmodified -/// events. -#[derive(Copy, Clone, Eq, PartialEq)] -enum AggregateMode { - AggregatedOnly, - Streaming, -} -pub(crate) struct AggregatedChatStream { - inner: S, - cumulative: String, - cumulative_reasoning: String, - pending: std::collections::VecDeque, - mode: AggregateMode, -} - -impl Stream for AggregatedChatStream -where - S: Stream> + Unpin, -{ - type Item = Result; - - fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { - let this = self.get_mut(); - - // First, flush any buffered events from the previous call. - if let Some(ev) = this.pending.pop_front() { - return Poll::Ready(Some(Ok(ev))); - } - - loop { - match Pin::new(&mut this.inner).poll_next(cx) { - Poll::Pending => return Poll::Pending, - Poll::Ready(None) => return Poll::Ready(None), - Poll::Ready(Some(Err(e))) => return Poll::Ready(Some(Err(e))), - Poll::Ready(Some(Ok(ResponseEvent::OutputItemDone(item)))) => { - // If this is an incremental assistant message chunk, accumulate but - // do NOT emit yet. Forward any other item (e.g. FunctionCall) right - // away so downstream consumers see it. - - let is_assistant_message = matches!( - &item, - codex_protocol::models::ResponseItem::Message { role, .. } if role == "assistant" - ); - - if is_assistant_message { - match this.mode { - AggregateMode::AggregatedOnly => { - // Only use the final assistant message if we have not - // seen any deltas; otherwise, deltas already built the - // cumulative text and this would duplicate it. - if this.cumulative.is_empty() - && let codex_protocol::models::ResponseItem::Message { - content, - .. - } = &item - && let Some(text) = content.iter().find_map(|c| match c { - codex_protocol::models::ContentItem::OutputText { - text, - } => Some(text), - _ => None, - }) - { - this.cumulative.push_str(text); - } - // Swallow assistant message here; emit on Completed. - continue; - } - AggregateMode::Streaming => { - // In streaming mode, if we have not seen any deltas, forward - // the final assistant message directly. If deltas were seen, - // suppress the final message to avoid duplication. - if this.cumulative.is_empty() { - return Poll::Ready(Some(Ok(ResponseEvent::OutputItemDone( - item, - )))); - } else { - continue; - } - } - } - } - - // Not an assistant message – forward immediately. - return Poll::Ready(Some(Ok(ResponseEvent::OutputItemDone(item)))); - } - Poll::Ready(Some(Ok(ResponseEvent::RateLimits(snapshot)))) => { - return Poll::Ready(Some(Ok(ResponseEvent::RateLimits(snapshot)))); - } - Poll::Ready(Some(Ok(ResponseEvent::Completed { - response_id, - token_usage, - }))) => { - // Build any aggregated items in the correct order: Reasoning first, then Message. - let mut emitted_any = false; - - if !this.cumulative_reasoning.is_empty() - && matches!(this.mode, AggregateMode::AggregatedOnly) - { - let aggregated_reasoning = - codex_protocol::models::ResponseItem::Reasoning { - id: String::new(), - summary: Vec::new(), - content: Some(vec![ - codex_protocol::models::ReasoningItemContent::ReasoningText { - text: std::mem::take(&mut this.cumulative_reasoning), - }, - ]), - encrypted_content: None, - }; - this.pending - .push_back(ResponseEvent::OutputItemDone(aggregated_reasoning)); - emitted_any = true; - } - - // Always emit the final aggregated assistant message when any - // content deltas have been observed. In AggregatedOnly mode this - // is the sole assistant output; in Streaming mode this finalizes - // the streamed deltas into a terminal OutputItemDone so callers - // can persist/render the message once per turn. - if !this.cumulative.is_empty() { - let aggregated_message = codex_protocol::models::ResponseItem::Message { - id: None, - role: "assistant".to_string(), - content: vec![codex_protocol::models::ContentItem::OutputText { - text: std::mem::take(&mut this.cumulative), - }], - }; - this.pending - .push_back(ResponseEvent::OutputItemDone(aggregated_message)); - emitted_any = true; - } - - // Always emit Completed last when anything was aggregated. - if emitted_any { - this.pending.push_back(ResponseEvent::Completed { - response_id: response_id.clone(), - token_usage: token_usage.clone(), - }); - // Return the first pending event now. - if let Some(ev) = this.pending.pop_front() { - return Poll::Ready(Some(Ok(ev))); - } - } - - // Nothing aggregated – forward Completed directly. - return Poll::Ready(Some(Ok(ResponseEvent::Completed { - response_id, - token_usage, - }))); - } - Poll::Ready(Some(Ok(ResponseEvent::Created))) => { - // These events are exclusive to the Responses API and - // will never appear in a Chat Completions stream. - continue; - } - Poll::Ready(Some(Ok(ResponseEvent::OutputTextDelta(delta)))) => { - // Always accumulate deltas so we can emit a final OutputItemDone at Completed. - this.cumulative.push_str(&delta); - if matches!(this.mode, AggregateMode::Streaming) { - // In streaming mode, also forward the delta immediately. - return Poll::Ready(Some(Ok(ResponseEvent::OutputTextDelta(delta)))); - } else { - continue; - } - } - Poll::Ready(Some(Ok(ResponseEvent::ReasoningContentDelta(delta)))) => { - // Always accumulate reasoning deltas so we can emit a final Reasoning item at Completed. - this.cumulative_reasoning.push_str(&delta); - if matches!(this.mode, AggregateMode::Streaming) { - // In streaming mode, also forward the delta immediately. - return Poll::Ready(Some(Ok(ResponseEvent::ReasoningContentDelta(delta)))); - } else { - continue; - } - } - Poll::Ready(Some(Ok(ResponseEvent::ReasoningSummaryDelta(_)))) => { - continue; - } - Poll::Ready(Some(Ok(ResponseEvent::ReasoningSummaryPartAdded))) => { - continue; - } - Poll::Ready(Some(Ok(ResponseEvent::OutputItemAdded(item)))) => { - return Poll::Ready(Some(Ok(ResponseEvent::OutputItemAdded(item)))); - } - } - } - } -} - -/// Extension trait that activates aggregation on any stream of [`ResponseEvent`]. -pub(crate) trait AggregateStreamExt: Stream> + Sized { - /// Returns a new stream that emits **only** the final assistant message - /// per turn instead of every incremental delta. The produced - /// `ResponseEvent` sequence for a typical text turn looks like: - /// - /// ```ignore - /// OutputItemDone() - /// Completed - /// ``` - /// - /// No other `OutputItemDone` events will be seen by the caller. - /// - /// Usage: - /// - /// ```ignore - /// let agg_stream = client.stream(&prompt).await?.aggregate(); - /// while let Some(event) = agg_stream.next().await { - /// // event now contains cumulative text - /// } - /// ``` - fn aggregate(self) -> AggregatedChatStream { - AggregatedChatStream::new(self, AggregateMode::AggregatedOnly) - } -} - -impl AggregateStreamExt for T where T: Stream> + Sized {} - -impl AggregatedChatStream { - fn new(inner: S, mode: AggregateMode) -> Self { - AggregatedChatStream { - inner, - cumulative: String::new(), - cumulative_reasoning: String::new(), - pending: std::collections::VecDeque::new(), - mode, - } - } - - pub(crate) fn streaming_mode(inner: S) -> Self { - Self::new(inner, AggregateMode::Streaming) - } -} diff --git a/codex-rs/core/src/client.rs b/codex-rs/core/src/client.rs index 7508391f5f..1675b9eee8 100644 --- a/codex-rs/core/src/client.rs +++ b/codex-rs/core/src/client.rs @@ -1,38 +1,29 @@ -use std::io::BufRead; -use std::path::Path; use std::sync::Arc; -use std::sync::OnceLock; -use std::time::Duration; -use bytes::Bytes; -use chrono::DateTime; -use chrono::Utc; -use codex_app_server_protocol::AuthMode; +use async_trait::async_trait; +use codex_api_client::AggregateStreamExt; +use codex_api_client::ApiClient; +use codex_api_client::AuthContext; +use codex_api_client::AuthProvider; +use codex_api_client::ChatAggregationMode; +use codex_api_client::ChatCompletionsApiClient; +use codex_api_client::ChatCompletionsApiClientConfig; +use codex_api_client::ResponsesApiClient; +use codex_api_client::ResponsesApiClientConfig; +use codex_api_client::Result as ApiClientResult; +use codex_api_client::stream_from_fixture; use codex_otel::otel_event_manager::OtelEventManager; use codex_protocol::ConversationId; use codex_protocol::config_types::ReasoningEffort as ReasoningEffortConfig; use codex_protocol::config_types::ReasoningSummary as ReasoningSummaryConfig; -use codex_protocol::models::ResponseItem; use codex_protocol::protocol::SessionSource; -use eventsource_stream::Eventsource; -use futures::prelude::*; -use regex_lite::Regex; -use reqwest::StatusCode; -use reqwest::header::HeaderMap; -use serde::Deserialize; -use serde::Serialize; -use serde_json::Value; +use futures::Stream; use tokio::sync::mpsc; -use tokio::time::timeout; -use tokio_util::io::ReaderStream; -use tracing::debug; -use tracing::trace; use tracing::warn; use crate::AuthManager; -use crate::auth::CodexAuth; -use crate::chat_completions::AggregateStreamExt; -use crate::chat_completions::stream_chat_completions; +use crate::ModelProviderInfo; +use crate::WireApi; use crate::client_common::Prompt; use crate::client_common::ResponseEvent; use crate::client_common::ResponseStream; @@ -43,38 +34,15 @@ use crate::default_client::CodexHttpClient; use crate::default_client::create_client; use crate::error::CodexErr; use crate::error::ConnectionFailedError; +use crate::error::EnvVarError; use crate::error::ResponseStreamFailed; use crate::error::Result; use crate::error::RetryLimitReachedError; use crate::error::UnexpectedResponseError; -use crate::error::UsageLimitReachedError; use crate::features::Feature; use crate::flags::CODEX_RS_SSE_FIXTURE; use crate::model_family::ModelFamily; -use crate::model_provider_info::ModelProviderInfo; -use crate::model_provider_info::WireApi; use crate::openai_model_info::get_model_info; -use crate::protocol::RateLimitSnapshot; -use crate::protocol::RateLimitWindow; -use crate::protocol::TokenUsage; -use crate::token_data::PlanType; -use crate::util::backoff; - -#[derive(Debug, Deserialize)] -struct ErrorResponse { - error: Error, -} - -#[derive(Debug, Deserialize)] -struct Error { - r#type: Option, - code: Option, - message: Option, - - // Optional fields available on "usage_limit_reached" and "usage_not_included" errors - plan_type: Option, - resets_at: Option, -} #[derive(Debug, Clone)] pub struct ModelClient { @@ -92,9 +60,6 @@ pub struct ModelClient { #[derive(Debug, Clone)] pub struct StreamPayload { pub prompt: Prompt, - pub instructions: String, - pub store_response: bool, - pub previous_response_id: Option, } #[allow(clippy::too_many_arguments)] @@ -152,360 +117,116 @@ impl ModelClient { } pub async fn stream(&self, payload: &StreamPayload) -> Result { + let mut prompt = payload.prompt.clone(); + self.populate_prompt(&mut prompt); + match self.provider.wire_api { - WireApi::Responses => self.stream_responses(payload).await, - WireApi::Chat => { - // Create the raw streaming connection first. - let response_stream = stream_chat_completions( - payload, - &self.config.model_family, - &self.client, - &self.provider, - &self.otel_event_manager, - &self.session_source, - ) - .await?; - - // Wrap it with the aggregation adapter so callers see *only* - // the final assistant message per turn (matching the - // behaviour of the Responses API). - let mut aggregated = if self.config.show_raw_agent_reasoning { - crate::chat_completions::AggregatedChatStream::streaming_mode(response_stream) - } else { - response_stream.aggregate() - }; - - // Bridge the aggregated stream back into a standard - // `ResponseStream` by forwarding events through a channel. - let (tx, rx) = mpsc::channel::>(16); - - tokio::spawn(async move { - use futures::StreamExt; - while let Some(ev) = aggregated.next().await { - // Exit early if receiver hung up. - if tx.send(ev).await.is_err() { - break; - } - } - }); - - Ok(ResponseStream { rx_event: rx }) - } + WireApi::Responses => self.stream_via_responses(prompt).await, + WireApi::Chat => self.stream_via_chat(prompt).await, } } + fn populate_prompt(&self, prompt: &mut Prompt) { + if prompt.prompt_cache_key.is_none() { + prompt.prompt_cache_key = Some(self.conversation_id.to_string()); + } + + prompt.session_source = Some(self.session_source.clone()); + + prompt.reasoning = create_reasoning_param_for_request( + &self.config.model_family, + self.effort, + self.summary, + ); + + let verbosity = if self.config.model_family.support_verbosity { + self.config.model_verbosity + } else { + if self.config.model_verbosity.is_some() { + warn!( + "model_verbosity is set but ignored as the model does not support verbosity: {}", + self.config.model_family.family + ); + } + None + }; + + prompt.text_controls = create_text_param_for_request(verbosity, &prompt.output_schema); + } + + async fn stream_via_responses(&self, prompt: Prompt) -> Result { + if let Some(path) = &*CODEX_RS_SSE_FIXTURE { + warn!(path, "Streaming from fixture"); + let stream = + stream_from_fixture(path, self.provider.clone(), self.otel_event_manager.clone()) + .await + .map_err(map_api_error)?; + return Ok(wrap_stream(stream)); + } + + let auth_provider = self.auth_manager.as_ref().map(|manager| { + Arc::new(AuthManagerProvider::new(Arc::clone(manager))) as Arc + }); + + let config = ResponsesApiClientConfig { + http_client: self.client.clone_inner(), + provider: self.provider.clone(), + model: self.config.model.clone(), + conversation_id: self.conversation_id, + auth_provider, + otel_event_manager: self.otel_event_manager.clone(), + }; + + let client = ResponsesApiClient::new(config) + .await + .map_err(map_api_error)?; + + let stream = client.stream(prompt).await.map_err(map_api_error)?; + + Ok(wrap_stream(stream)) + } + + async fn stream_via_chat(&self, prompt: Prompt) -> Result { + let config = ChatCompletionsApiClientConfig { + http_client: self.client.clone_inner(), + provider: self.provider.clone(), + model: self.config.model.clone(), + otel_event_manager: self.otel_event_manager.clone(), + session_source: self.session_source.clone(), + aggregation_mode: if self.config.show_raw_agent_reasoning { + ChatAggregationMode::Streaming + } else { + ChatAggregationMode::AggregatedOnly + }, + }; + + let client = ChatCompletionsApiClient::new(config) + .await + .map_err(map_api_error)?; + + let stream = client.stream(prompt).await.map_err(map_api_error)?; + + let stream = if self.config.show_raw_agent_reasoning { + stream.streaming_mode() + } else { + stream.aggregate() + }; + + Ok(wrap_stream(stream)) + } + pub async fn stream_for_test(&self, mut prompt: Prompt) -> Result { crate::conversation_history::format_prompt_items(&mut prompt.input, false); let instructions = crate::client_common::compute_full_instructions(None, &self.config.model_family, false) .into_owned(); - let payload = StreamPayload { - prompt, - instructions, - store_response: false, - previous_response_id: None, - }; + prompt.instructions = instructions; + prompt.store_response = false; + prompt.previous_response_id = None; + let payload = StreamPayload { prompt }; self.stream(&payload).await } - /// Implementation for the OpenAI *Responses* experimental API. - async fn stream_responses(&self, payload: &StreamPayload) -> Result { - if let Some(path) = &*CODEX_RS_SSE_FIXTURE { - // short circuit for tests - warn!(path, "Streaming from fixture"); - return stream_from_fixture( - path, - self.provider.clone(), - self.otel_event_manager.clone(), - ) - .await; - } - - let auth_manager = self.auth_manager.clone(); - - let prompt = &payload.prompt; - let tools_json = prompt.tools.clone(); - let reasoning = create_reasoning_param_for_request( - &self.config.model_family, - self.effort, - self.summary, - ); - - let include: Vec = if reasoning.is_some() { - vec!["reasoning.encrypted_content".to_string()] - } else { - vec![] - }; - - let input_with_instructions = prompt.input.clone(); - - let verbosity = if self.config.model_family.support_verbosity { - self.config.model_verbosity - } else { - warn!( - "model_verbosity is set but ignored as the model does not support verbosity: {}", - self.config.model_family.family - ); - None - }; - - // Only include `text.verbosity` for GPT-5 family models - let text = create_text_param_for_request(verbosity, &prompt.output_schema); - - // In general, we want to explicitly send `store: false` when using the Responses API, - // but in practice, the Azure Responses API rejects `store: false`: - // - // - If store = false and id is sent an error is thrown that ID is not found - // - If store = false and id is not sent an error is thrown that ID is required - // - // For Azure, we send `store: true` and preserve reasoning item IDs. - let azure_workaround = self.provider.is_azure_responses_endpoint(); - - let mut payload_json = serde_json::json!({ - "model": self.config.model, - "instructions": payload.instructions, - "input": input_with_instructions, - "tools": tools_json, - "tool_choice": "auto", - "parallel_tool_calls": prompt.parallel_tool_calls, - "store": azure_workaround || payload.store_response, - "stream": true, - "include": include, - "prompt_cache_key": self.conversation_id.to_string(), - }); - - if let Some(reasoning) = reasoning { - payload_json - .as_object_mut() - .expect("payload object") - .insert("reasoning".to_string(), serde_json::to_value(reasoning)?); - } - - if let Some(text) = text { - payload_json - .as_object_mut() - .expect("payload object") - .insert("text".to_string(), serde_json::to_value(text)?); - } - - if let Some(previous) = payload.previous_response_id.as_ref() { - payload_json - .as_object_mut() - .expect("payload object") - .insert( - "previous_response_id".to_string(), - serde_json::Value::String(previous.clone()), - ); - } - - if azure_workaround { - if let Some(input_value) = payload_json.get_mut("input") - && let Some(array) = input_value.as_array_mut() - { - attach_item_ids_array(array, &prompt.input); - } - } - - let max_attempts = self.provider.request_max_retries(); - for attempt in 0..=max_attempts { - match self - .attempt_stream_responses(attempt, &payload_json, &auth_manager) - .await - { - Ok(stream) => { - return Ok(stream); - } - Err(StreamAttemptError::Fatal(e)) => { - return Err(e); - } - Err(retryable_attempt_error) => { - if attempt == max_attempts { - return Err(retryable_attempt_error.into_error()); - } - - tokio::time::sleep(retryable_attempt_error.delay(attempt)).await; - } - } - } - - unreachable!("stream_responses_attempt should always return"); - } - - /// Single attempt to start a streaming Responses API call. - async fn attempt_stream_responses( - &self, - attempt: u64, - payload_json: &Value, - auth_manager: &Option>, - ) -> std::result::Result { - // Always fetch the latest auth in case a prior attempt refreshed the token. - let auth = auth_manager.as_ref().and_then(|m| m.auth()); - - trace!( - "POST to {}: {:?}", - self.provider.get_full_url(&auth), - serde_json::to_string(payload_json) - .unwrap_or("".to_string()) - ); - - let mut req_builder = self - .provider - .create_request_builder(&self.client, &auth) - .await - .map_err(StreamAttemptError::Fatal)?; - - // Include subagent header only for subagent sessions. - if let SessionSource::SubAgent(sub) = &self.session_source { - let subagent = if let crate::protocol::SubAgentSource::Other(label) = sub { - label.clone() - } else { - serde_json::to_value(sub) - .ok() - .and_then(|v| v.as_str().map(std::string::ToString::to_string)) - .unwrap_or_else(|| "other".to_string()) - }; - req_builder = req_builder.header("x-openai-subagent", subagent); - } - - req_builder = req_builder - // Send session_id for compatibility. - .header("conversation_id", self.conversation_id.to_string()) - .header("session_id", self.conversation_id.to_string()) - .header(reqwest::header::ACCEPT, "text/event-stream") - .json(payload_json); - - if let Some(auth) = auth.as_ref() - && auth.mode == AuthMode::ChatGPT - && let Some(account_id) = auth.get_account_id() - { - req_builder = req_builder.header("chatgpt-account-id", account_id); - } - - let res = self - .otel_event_manager - .log_request(attempt, || req_builder.send()) - .await; - - let mut request_id = None; - if let Ok(resp) = &res { - request_id = resp - .headers() - .get("cf-ray") - .map(|v| v.to_str().unwrap_or_default().to_string()); - } - - match res { - Ok(resp) if resp.status().is_success() => { - let (tx_event, rx_event) = mpsc::channel::>(1600); - - if let Some(snapshot) = parse_rate_limit_snapshot(resp.headers()) - && tx_event - .send(Ok(ResponseEvent::RateLimits(snapshot))) - .await - .is_err() - { - debug!("receiver dropped rate limit snapshot event"); - } - - // spawn task to process SSE - let stream = resp.bytes_stream().map_err(move |e| { - CodexErr::ResponseStreamFailed(ResponseStreamFailed { - source: e, - request_id: request_id.clone(), - }) - }); - tokio::spawn(process_sse( - stream, - tx_event, - self.provider.stream_idle_timeout(), - self.otel_event_manager.clone(), - )); - - Ok(ResponseStream { rx_event }) - } - Ok(res) => { - let status = res.status(); - - // Pull out Retry‑After header if present. - let retry_after_secs = res - .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(|s| Duration::from_millis(s * 1_000)); - - if status == StatusCode::UNAUTHORIZED - && let Some(manager) = auth_manager.as_ref() - && let Some(auth) = auth.as_ref() - && auth.mode == AuthMode::ChatGPT - { - manager.refresh_token().await.map_err(|err| { - StreamAttemptError::Fatal(CodexErr::Fatal(format!( - "Failed to refresh ChatGPT credentials: {err}" - ))) - })?; - } - - // The OpenAI Responses endpoint returns structured JSON bodies even for 4xx/5xx - // errors. When we bubble early with only the HTTP status the caller sees an opaque - // "unexpected status 400 Bad Request" which makes debugging nearly impossible. - // Instead, read (and include) the response text so higher layers and users see the - // exact error message (e.g. "Unknown parameter: 'input[0].metadata'"). The body is - // small and this branch only runs on error paths so the extra allocation is - // negligible. - if !(status == StatusCode::TOO_MANY_REQUESTS - || status == StatusCode::UNAUTHORIZED - || status.is_server_error()) - { - // Surface the error body to callers. Use `unwrap_or_default` per Clippy. - let body = res.text().await.unwrap_or_default(); - return Err(StreamAttemptError::Fatal(CodexErr::UnexpectedStatus( - UnexpectedResponseError { - status, - body, - request_id: None, - }, - ))); - } - - if status == StatusCode::TOO_MANY_REQUESTS { - let rate_limit_snapshot = parse_rate_limit_snapshot(res.headers()); - let body = res.json::().await.ok(); - if let Some(ErrorResponse { error }) = body { - if error.r#type.as_deref() == Some("usage_limit_reached") { - // Prefer the plan_type provided in the error message if present - // because it's more up to date than the one encoded in the auth - // token. - let plan_type = error - .plan_type - .or_else(|| auth.as_ref().and_then(CodexAuth::get_plan_type)); - let resets_at = error - .resets_at - .and_then(|seconds| DateTime::::from_timestamp(seconds, 0)); - let codex_err = CodexErr::UsageLimitReached(UsageLimitReachedError { - plan_type, - resets_at, - rate_limits: rate_limit_snapshot, - }); - return Err(StreamAttemptError::Fatal(codex_err)); - } else if error.r#type.as_deref() == Some("usage_not_included") { - return Err(StreamAttemptError::Fatal(CodexErr::UsageNotIncluded)); - } - } - } - - Err(StreamAttemptError::RetryableHttpError { - status, - retry_after, - request_id, - }) - } - Err(e) => Err(StreamAttemptError::RetryableTransportError( - CodexErr::ConnectionFailed(ConnectionFailedError { source: e }), - )), - } - } - pub fn get_provider(&self) -> ModelProviderInfo { self.provider.clone() } @@ -543,958 +264,96 @@ impl ModelClient { } } -enum StreamAttemptError { - RetryableHttpError { - status: StatusCode, - retry_after: Option, - request_id: Option, - }, - RetryableTransportError(CodexErr), - Fatal(CodexErr), +struct AuthManagerProvider { + manager: Arc, } -impl StreamAttemptError { - /// attempt is 0-based. - fn delay(&self, attempt: u64) -> Duration { - // backoff() uses 1-based attempts. - let backoff_attempt = attempt + 1; - match self { - Self::RetryableHttpError { retry_after, .. } => { - retry_after.unwrap_or_else(|| backoff(backoff_attempt)) - } - Self::RetryableTransportError { .. } => backoff(backoff_attempt), - Self::Fatal(_) => { - // Should not be called on Fatal errors. - Duration::from_secs(0) - } - } - } - - fn into_error(self) -> CodexErr { - match self { - Self::RetryableHttpError { - status, request_id, .. - } => { - if status == StatusCode::INTERNAL_SERVER_ERROR { - CodexErr::InternalServerError - } else { - CodexErr::RetryLimit(RetryLimitReachedError { status, request_id }) - } - } - Self::RetryableTransportError(error) => error, - Self::Fatal(error) => error, - } +impl AuthManagerProvider { + fn new(manager: Arc) -> Self { + Self { manager } } } -#[derive(Debug, Deserialize, Serialize)] -struct SseEvent { - #[serde(rename = "type")] - kind: String, - response: Option, - item: Option, - delta: Option, -} - -#[derive(Debug, Deserialize)] -struct ResponseCompleted { - id: String, - usage: Option, -} - -#[derive(Debug, Deserialize)] -struct ResponseCompletedUsage { - input_tokens: i64, - input_tokens_details: Option, - output_tokens: i64, - output_tokens_details: Option, - total_tokens: i64, -} - -impl From for TokenUsage { - fn from(val: ResponseCompletedUsage) -> Self { - TokenUsage { - input_tokens: val.input_tokens, - cached_input_tokens: val - .input_tokens_details - .map(|d| d.cached_tokens) - .unwrap_or(0), - output_tokens: val.output_tokens, - reasoning_output_tokens: val - .output_tokens_details - .map(|d| d.reasoning_tokens) - .unwrap_or(0), - total_tokens: val.total_tokens, - } - } -} - -#[derive(Debug, Deserialize)] -struct ResponseCompletedInputTokensDetails { - cached_tokens: i64, -} - -#[derive(Debug, Deserialize)] -struct ResponseCompletedOutputTokensDetails { - reasoning_tokens: i64, -} - -fn attach_item_ids_array(items: &mut [Value], original_items: &[ResponseItem]) { - for (value, item) in items.iter_mut().zip(original_items.iter()) { - if let ResponseItem::Reasoning { id, .. } - | ResponseItem::Message { id: Some(id), .. } - | ResponseItem::WebSearchCall { id: Some(id), .. } - | ResponseItem::FunctionCall { id: Some(id), .. } - | ResponseItem::LocalShellCall { id: Some(id), .. } - | ResponseItem::CustomToolCall { id: Some(id), .. } = item - { - if id.is_empty() { - continue; +#[async_trait] +impl AuthProvider for AuthManagerProvider { + async fn auth_context(&self) -> Option { + let auth = self.manager.auth()?; + let mode = auth.mode; + let account_id = auth.get_account_id(); + let bearer_token = match auth.get_token().await { + Ok(token) if !token.is_empty() => Some(token), + Ok(_) => None, + Err(err) => { + warn!("failed to resolve auth token: {err}"); + None } + }; - if let Some(obj) = value.as_object_mut() { - obj.insert("id".to_string(), Value::String(id.clone())); - } - } - } -} - -fn parse_rate_limit_snapshot(headers: &HeaderMap) -> Option { - let primary = parse_rate_limit_window( - headers, - "x-codex-primary-used-percent", - "x-codex-primary-window-minutes", - "x-codex-primary-reset-at", - ); - - let secondary = parse_rate_limit_window( - headers, - "x-codex-secondary-used-percent", - "x-codex-secondary-window-minutes", - "x-codex-secondary-reset-at", - ); - - Some(RateLimitSnapshot { primary, secondary }) -} - -fn parse_rate_limit_window( - headers: &HeaderMap, - used_percent_header: &str, - window_minutes_header: &str, - resets_at_header: &str, -) -> Option { - let used_percent: Option = parse_header_f64(headers, used_percent_header); - - used_percent.and_then(|used_percent| { - let window_minutes = parse_header_i64(headers, window_minutes_header); - let resets_at = parse_header_i64(headers, resets_at_header); - - let has_data = used_percent != 0.0 - || window_minutes.is_some_and(|minutes| minutes != 0) - || resets_at.is_some(); - - has_data.then_some(RateLimitWindow { - used_percent, - window_minutes, - resets_at, + Some(AuthContext { + mode, + bearer_token, + account_id, }) - }) + } + + async fn refresh_token(&self) -> std::result::Result, String> { + self.manager + .refresh_token() + .await + .map_err(|err| err.to_string()) + } } -fn parse_header_f64(headers: &HeaderMap, name: &str) -> Option { - parse_header_str(headers, name)? - .parse::() - .ok() - .filter(|v| v.is_finite()) -} - -fn parse_header_i64(headers: &HeaderMap, name: &str) -> Option { - parse_header_str(headers, name)?.parse::().ok() -} - -fn parse_header_str<'a>(headers: &'a HeaderMap, name: &str) -> Option<&'a str> { - headers.get(name)?.to_str().ok() -} - -async fn process_sse( - stream: S, - tx_event: mpsc::Sender>, - idle_timeout: Duration, - otel_event_manager: OtelEventManager, -) where - S: Stream> + Unpin, +fn wrap_stream(stream: S) -> ResponseStream +where + S: Stream> + Send + Unpin + 'static, { - let mut stream = stream.eventsource(); + let (tx, rx) = mpsc::channel::>(1600); - // If the stream stays completely silent for an extended period treat it as disconnected. - // The response id returned from the "complete" message. - let mut response_completed: Option = None; - let mut response_error: Option = None; + tokio::spawn(async move { + use futures::StreamExt; - loop { - let start = std::time::Instant::now(); - let response = timeout(idle_timeout, stream.next()).await; - let duration = start.elapsed(); - otel_event_manager.log_sse_event(&response, duration); - - let sse = match response { - Ok(Some(Ok(sse))) => sse, - Ok(Some(Err(e))) => { - debug!("SSE Error: {e:#}"); - let event = CodexErr::Stream(e.to_string(), None); - let _ = tx_event.send(Err(event)).await; - return; - } - Ok(None) => { - match response_completed { - Some(ResponseCompleted { - id: response_id, - usage, - }) => { - if let Some(token_usage) = &usage { - otel_event_manager.sse_event_completed( - token_usage.input_tokens, - token_usage.output_tokens, - token_usage - .input_tokens_details - .as_ref() - .map(|d| d.cached_tokens), - token_usage - .output_tokens_details - .as_ref() - .map(|d| d.reasoning_tokens), - token_usage.total_tokens, - ); - } - let event = ResponseEvent::Completed { - response_id, - token_usage: usage.map(Into::into), - }; - let _ = tx_event.send(Ok(event)).await; - } - None => { - let error = response_error.unwrap_or(CodexErr::Stream( - "stream closed before response.completed".into(), - None, - )); - otel_event_manager.see_event_completed_failed(&error); - - let _ = tx_event.send(Err(error)).await; - } - } - return; - } - Err(_) => { - let _ = tx_event - .send(Err(CodexErr::Stream( - "idle timeout waiting for SSE".into(), - None, - ))) - .await; - return; - } - }; - - let raw = sse.data.clone(); - trace!("SSE event: {}", raw); - - let event: SseEvent = match serde_json::from_str(&sse.data) { - Ok(event) => event, - Err(e) => { - debug!("Failed to parse SSE event: {e}, data: {}", &sse.data); - continue; - } - }; - - match event.kind.as_str() { - // Individual output item finalised. Forward immediately so the - // rest of the agent can stream assistant text/functions *live* - // instead of waiting for the final `response.completed` envelope. - // - // IMPORTANT: We used to ignore these events and forward the - // duplicated `output` array embedded in the `response.completed` - // payload. That produced two concrete issues: - // 1. No real‑time streaming – the user only saw output after the - // entire turn had finished, which broke the "typing" UX and - // made long‑running turns look stalled. - // 2. Duplicate `function_call_output` items – both the - // individual *and* the completed array were forwarded, which - // confused the backend and triggered 400 - // "previous_response_not_found" errors because the duplicated - // IDs did not match the incremental turn chain. - // - // The fix is to forward the incremental events *as they come* and - // drop the duplicated list inside `response.completed`. - "response.output_item.done" => { - let Some(item_val) = event.item else { continue }; - let Ok(item) = serde_json::from_value::(item_val) else { - debug!("failed to parse ResponseItem from output_item.done"); - continue; - }; - - let event = ResponseEvent::OutputItemDone(item); - if tx_event.send(Ok(event)).await.is_err() { - return; - } - } - "response.output_text.delta" => { - if let Some(delta) = event.delta { - let event = ResponseEvent::OutputTextDelta(delta); - if tx_event.send(Ok(event)).await.is_err() { - return; - } - } - } - "response.reasoning_summary_text.delta" => { - if let Some(delta) = event.delta { - let event = ResponseEvent::ReasoningSummaryDelta(delta); - if tx_event.send(Ok(event)).await.is_err() { - return; - } - } - } - "response.reasoning_text.delta" => { - if let Some(delta) = event.delta { - let event = ResponseEvent::ReasoningContentDelta(delta); - if tx_event.send(Ok(event)).await.is_err() { - return; - } - } - } - "response.created" => { - if event.response.is_some() { - let _ = tx_event.send(Ok(ResponseEvent::Created {})).await; - } - } - "response.failed" => { - if let Some(resp_val) = event.response { - response_error = Some(CodexErr::Stream( - "response.failed event received".to_string(), - None, - )); - - let error = resp_val.get("error"); - - if let Some(error) = error { - match serde_json::from_value::(error.clone()) { - Ok(error) => { - if is_context_window_error(&error) { - response_error = Some(CodexErr::ContextWindowExceeded); - } else { - let delay = try_parse_retry_after(&error); - let message = error.message.clone().unwrap_or_default(); - response_error = Some(CodexErr::Stream(message, delay)); - } - } - Err(e) => { - let error = format!("failed to parse ErrorResponse: {e}"); - debug!(error); - response_error = Some(CodexErr::Stream(error, None)) - } - } - } - } - } - // Final response completed – includes array of output items & id - "response.completed" => { - if let Some(resp_val) = event.response { - match serde_json::from_value::(resp_val) { - Ok(r) => { - response_completed = Some(r); - } - Err(e) => { - let error = format!("failed to parse ResponseCompleted: {e}"); - debug!(error); - response_error = Some(CodexErr::Stream(error, None)); - continue; - } - }; - }; - } - "response.content_part.done" - | "response.function_call_arguments.delta" - | "response.custom_tool_call_input.delta" - | "response.custom_tool_call_input.done" // also emitted as response.output_item.done - | "response.in_progress" - | "response.output_text.done" => {} - "response.output_item.added" => { - let Some(item_val) = event.item else { continue }; - let Ok(item) = serde_json::from_value::(item_val) else { - debug!("failed to parse ResponseItem from output_item.done"); - continue; - }; - - let event = ResponseEvent::OutputItemAdded(item); - if tx_event.send(Ok(event)).await.is_err() { - return; - } - } - "response.reasoning_summary_part.added" => { - // Boundary between reasoning summary sections (e.g., titles). - let event = ResponseEvent::ReasoningSummaryPartAdded; - if tx_event.send(Ok(event)).await.is_err() { - return; - } - } - "response.reasoning_summary_text.done" => {} - _ => {} - } - } -} - -/// used in tests to stream from a text SSE file -async fn stream_from_fixture( - path: impl AsRef, - provider: ModelProviderInfo, - otel_event_manager: OtelEventManager, -) -> Result { - let (tx_event, rx_event) = mpsc::channel::>(1600); - let f = std::fs::File::open(path.as_ref())?; - let lines = std::io::BufReader::new(f).lines(); - - // insert \n\n after each line for proper SSE parsing - let mut content = String::new(); - for line in lines { - content.push_str(&line?); - content.push_str("\n\n"); - } - - let rdr = std::io::Cursor::new(content); - let stream = ReaderStream::new(rdr).map_err(CodexErr::Io); - tokio::spawn(process_sse( - stream, - tx_event, - provider.stream_idle_timeout(), - otel_event_manager, - )); - Ok(ResponseStream { rx_event }) -} - -fn rate_limit_regex() -> &'static Regex { - static RE: OnceLock = OnceLock::new(); - - #[expect(clippy::unwrap_used)] - RE.get_or_init(|| Regex::new(r"Please try again in (\d+(?:\.\d+)?)(s|ms)").unwrap()) -} - -fn try_parse_retry_after(err: &Error) -> Option { - if err.code != Some("rate_limit_exceeded".to_string()) { - return None; - } - - // parse the Please try again in 1.898s format using regex - let re = rate_limit_regex(); - if let Some(message) = &err.message - && let Some(captures) = re.captures(message) - { - let seconds = captures.get(1); - let unit = captures.get(2); - - if let (Some(value), Some(unit)) = (seconds, unit) { - let value = value.as_str().parse::().ok()?; - let unit = unit.as_str(); - - if unit == "s" { - return Some(Duration::from_secs_f64(value)); - } else if unit == "ms" { - return Some(Duration::from_millis(value as u64)); - } - } - } - None -} - -fn is_context_window_error(error: &Error) -> bool { - error.code.as_deref() == Some("context_length_exceeded") -} - -#[cfg(test)] -mod tests { - use super::*; - use assert_matches::assert_matches; - use serde_json::json; - use tokio::sync::mpsc; - use tokio_test::io::Builder as IoBuilder; - use tokio_util::io::ReaderStream; - - // ──────────────────────────── - // Helpers - // ──────────────────────────── - - /// Runs the SSE parser on pre-chunked byte slices and returns every event - /// (including any final `Err` from a stream-closure check). - async fn collect_events( - chunks: &[&[u8]], - provider: ModelProviderInfo, - otel_event_manager: OtelEventManager, - ) -> Vec> { - let mut builder = IoBuilder::new(); - for chunk in chunks { - builder.read(chunk); - } - - let reader = builder.build(); - let stream = ReaderStream::new(reader).map_err(CodexErr::Io); - let (tx, mut rx) = mpsc::channel::>(16); - tokio::spawn(process_sse( - stream, - tx, - provider.stream_idle_timeout(), - otel_event_manager, - )); - - let mut events = Vec::new(); - while let Some(ev) = rx.recv().await { - events.push(ev); - } - events - } - - /// Builds an in-memory SSE stream from JSON fixtures and returns only the - /// successfully parsed events (panics on internal channel errors). - async fn run_sse( - events: Vec, - provider: ModelProviderInfo, - otel_event_manager: OtelEventManager, - ) -> Vec { - let mut body = String::new(); - for e in events { - let kind = e - .get("type") - .and_then(|v| v.as_str()) - .expect("fixture event missing type"); - if e.as_object().map(|o| o.len() == 1).unwrap_or(false) { - body.push_str(&format!("event: {kind}\n\n")); - } else { - body.push_str(&format!("event: {kind}\ndata: {e}\n\n")); - } - } - - let (tx, mut rx) = mpsc::channel::>(8); - let stream = ReaderStream::new(std::io::Cursor::new(body)).map_err(CodexErr::Io); - tokio::spawn(process_sse( - stream, - tx, - provider.stream_idle_timeout(), - otel_event_manager, - )); - - let mut out = Vec::new(); - while let Some(ev) = rx.recv().await { - out.push(ev.expect("channel closed")); - } - out - } - - fn otel_event_manager() -> OtelEventManager { - OtelEventManager::new( - ConversationId::new(), - "test", - "test", - None, - Some("test@test.com".to_string()), - Some(AuthMode::ChatGPT), - false, - "test".to_string(), - ) - } - - // ──────────────────────────── - // Tests from `implement-test-for-responses-api-sse-parser` - // ──────────────────────────── - - #[tokio::test] - async fn parses_items_and_completed() { - let item1 = json!({ - "type": "response.output_item.done", - "item": { - "type": "message", - "role": "assistant", - "content": [{"type": "output_text", "text": "Hello"}] - } - }) - .to_string(); - - let item2 = json!({ - "type": "response.output_item.done", - "item": { - "type": "message", - "role": "assistant", - "content": [{"type": "output_text", "text": "World"}] - } - }) - .to_string(); - - let completed = json!({ - "type": "response.completed", - "response": { "id": "resp1" } - }) - .to_string(); - - let sse1 = format!("event: response.output_item.done\ndata: {item1}\n\n"); - let sse2 = format!("event: response.output_item.done\ndata: {item2}\n\n"); - let sse3 = format!("event: response.completed\ndata: {completed}\n\n"); - - let provider = ModelProviderInfo { - name: "test".to_string(), - base_url: Some("https://test.com".to_string()), - env_key: Some("TEST_API_KEY".to_string()), - env_key_instructions: None, - experimental_bearer_token: None, - wire_api: WireApi::Responses, - query_params: None, - http_headers: None, - env_http_headers: None, - request_max_retries: Some(0), - stream_max_retries: Some(0), - stream_idle_timeout_ms: Some(1000), - requires_openai_auth: false, - }; - - let otel_event_manager = otel_event_manager(); - - let events = collect_events( - &[sse1.as_bytes(), sse2.as_bytes(), sse3.as_bytes()], - provider, - otel_event_manager, - ) - .await; - - assert_eq!(events.len(), 3); - - matches!( - &events[0], - Ok(ResponseEvent::OutputItemDone(ResponseItem::Message { role, .. })) - if role == "assistant" - ); - - matches!( - &events[1], - Ok(ResponseEvent::OutputItemDone(ResponseItem::Message { role, .. })) - if role == "assistant" - ); - - match &events[2] { - Ok(ResponseEvent::Completed { - response_id, - token_usage, - }) => { - assert_eq!(response_id, "resp1"); - assert!(token_usage.is_none()); - } - other => panic!("unexpected third event: {other:?}"), - } - } - - #[tokio::test] - async fn error_when_missing_completed() { - let item1 = json!({ - "type": "response.output_item.done", - "item": { - "type": "message", - "role": "assistant", - "content": [{"type": "output_text", "text": "Hello"}] - } - }) - .to_string(); - - let sse1 = format!("event: response.output_item.done\ndata: {item1}\n\n"); - let provider = ModelProviderInfo { - name: "test".to_string(), - base_url: Some("https://test.com".to_string()), - env_key: Some("TEST_API_KEY".to_string()), - env_key_instructions: None, - experimental_bearer_token: None, - wire_api: WireApi::Responses, - query_params: None, - http_headers: None, - env_http_headers: None, - request_max_retries: Some(0), - stream_max_retries: Some(0), - stream_idle_timeout_ms: Some(1000), - requires_openai_auth: false, - }; - - let otel_event_manager = otel_event_manager(); - - let events = collect_events(&[sse1.as_bytes()], provider, otel_event_manager).await; - - assert_eq!(events.len(), 2); - - matches!(events[0], Ok(ResponseEvent::OutputItemDone(_))); - - match &events[1] { - Err(CodexErr::Stream(msg, _)) => { - assert_eq!(msg, "stream closed before response.completed") - } - other => panic!("unexpected second event: {other:?}"), - } - } - - #[tokio::test] - async fn error_when_error_event() { - let raw_error = r#"{"type":"response.failed","sequence_number":3,"response":{"id":"resp_689bcf18d7f08194bf3440ba62fe05d803fee0cdac429894","object":"response","created_at":1755041560,"status":"failed","background":false,"error":{"code":"rate_limit_exceeded","message":"Rate limit reached for gpt-5 in organization org-AAA on tokens per min (TPM): Limit 30000, Used 22999, Requested 12528. Please try again in 11.054s. Visit https://platform.openai.com/account/rate-limits to learn more."}, "usage":null,"user":null,"metadata":{}}}"#; - - let sse1 = format!("event: response.failed\ndata: {raw_error}\n\n"); - let provider = ModelProviderInfo { - name: "test".to_string(), - base_url: Some("https://test.com".to_string()), - env_key: Some("TEST_API_KEY".to_string()), - env_key_instructions: None, - experimental_bearer_token: None, - wire_api: WireApi::Responses, - query_params: None, - http_headers: None, - env_http_headers: None, - request_max_retries: Some(0), - stream_max_retries: Some(0), - stream_idle_timeout_ms: Some(1000), - requires_openai_auth: false, - }; - - let otel_event_manager = otel_event_manager(); - - let events = collect_events(&[sse1.as_bytes()], provider, otel_event_manager).await; - - assert_eq!(events.len(), 1); - - match &events[0] { - Err(CodexErr::Stream(msg, delay)) => { - assert_eq!( - msg, - "Rate limit reached for gpt-5 in organization org-AAA on tokens per min (TPM): Limit 30000, Used 22999, Requested 12528. Please try again in 11.054s. Visit https://platform.openai.com/account/rate-limits to learn more." - ); - assert_eq!(*delay, Some(Duration::from_secs_f64(11.054))); - } - other => panic!("unexpected second event: {other:?}"), - } - } - - #[tokio::test] - async fn context_window_error_is_fatal() { - let raw_error = r#"{"type":"response.failed","sequence_number":3,"response":{"id":"resp_5c66275b97b9baef1ed95550adb3b7ec13b17aafd1d2f11b","object":"response","created_at":1759510079,"status":"failed","background":false,"error":{"code":"context_length_exceeded","message":"Your input exceeds the context window of this model. Please adjust your input and try again."},"usage":null,"user":null,"metadata":{}}}"#; - - let sse1 = format!("event: response.failed\ndata: {raw_error}\n\n"); - let provider = ModelProviderInfo { - name: "test".to_string(), - base_url: Some("https://test.com".to_string()), - env_key: Some("TEST_API_KEY".to_string()), - env_key_instructions: None, - experimental_bearer_token: None, - wire_api: WireApi::Responses, - query_params: None, - http_headers: None, - env_http_headers: None, - request_max_retries: Some(0), - stream_max_retries: Some(0), - stream_idle_timeout_ms: Some(1000), - requires_openai_auth: false, - }; - - let otel_event_manager = otel_event_manager(); - - let events = collect_events(&[sse1.as_bytes()], provider, otel_event_manager).await; - - assert_eq!(events.len(), 1); - - match &events[0] { - Err(err @ CodexErr::ContextWindowExceeded) => { - assert_eq!(err.to_string(), CodexErr::ContextWindowExceeded.to_string()); - } - other => panic!("unexpected context window event: {other:?}"), - } - } - - #[tokio::test] - async fn context_window_error_with_newline_is_fatal() { - let raw_error = r#"{"type":"response.failed","sequence_number":4,"response":{"id":"resp_fatal_newline","object":"response","created_at":1759510080,"status":"failed","background":false,"error":{"code":"context_length_exceeded","message":"Your input exceeds the context window of this model. Please adjust your input and try\nagain."},"usage":null,"user":null,"metadata":{}}}"#; - - let sse1 = format!("event: response.failed\ndata: {raw_error}\n\n"); - let provider = ModelProviderInfo { - name: "test".to_string(), - base_url: Some("https://test.com".to_string()), - env_key: Some("TEST_API_KEY".to_string()), - env_key_instructions: None, - experimental_bearer_token: None, - wire_api: WireApi::Responses, - query_params: None, - http_headers: None, - env_http_headers: None, - request_max_retries: Some(0), - stream_max_retries: Some(0), - stream_idle_timeout_ms: Some(1000), - requires_openai_auth: false, - }; - - let otel_event_manager = otel_event_manager(); - - let events = collect_events(&[sse1.as_bytes()], provider, otel_event_manager).await; - - assert_eq!(events.len(), 1); - - match &events[0] { - Err(err @ CodexErr::ContextWindowExceeded) => { - assert_eq!(err.to_string(), CodexErr::ContextWindowExceeded.to_string()); - } - other => panic!("unexpected context window event: {other:?}"), - } - } - - // ──────────────────────────── - // Table-driven test from `main` - // ──────────────────────────── - - /// Verifies that the adapter produces the right `ResponseEvent` for a - /// variety of incoming `type` values. - #[tokio::test] - async fn table_driven_event_kinds() { - struct TestCase { - name: &'static str, - event: serde_json::Value, - expect_first: fn(&ResponseEvent) -> bool, - expected_len: usize, - } - - fn is_created(ev: &ResponseEvent) -> bool { - matches!(ev, ResponseEvent::Created) - } - fn is_output(ev: &ResponseEvent) -> bool { - matches!(ev, ResponseEvent::OutputItemDone(_)) - } - fn is_completed(ev: &ResponseEvent) -> bool { - matches!(ev, ResponseEvent::Completed { .. }) - } - - let completed = json!({ - "type": "response.completed", - "response": { - "id": "c", - "usage": { - "input_tokens": 0, - "input_tokens_details": null, - "output_tokens": 0, - "output_tokens_details": null, - "total_tokens": 0 - }, - "output": [] - } - }); - - let cases = vec![ - TestCase { - name: "created", - event: json!({"type": "response.created", "response": {}}), - expect_first: is_created, - expected_len: 2, - }, - TestCase { - name: "output_item.done", - event: json!({ - "type": "response.output_item.done", - "item": { - "type": "message", - "role": "assistant", - "content": [ - {"type": "output_text", "text": "hi"} - ] - } - }), - expect_first: is_output, - expected_len: 2, - }, - TestCase { - name: "unknown", - event: json!({"type": "response.new_tool_event"}), - expect_first: is_completed, - expected_len: 1, - }, - ]; - - for case in cases { - let mut evs = vec![case.event]; - evs.push(completed.clone()); - - let provider = ModelProviderInfo { - name: "test".to_string(), - base_url: Some("https://test.com".to_string()), - env_key: Some("TEST_API_KEY".to_string()), - env_key_instructions: None, - experimental_bearer_token: None, - wire_api: WireApi::Responses, - query_params: None, - http_headers: None, - env_http_headers: None, - request_max_retries: Some(0), - stream_max_retries: Some(0), - stream_idle_timeout_ms: Some(1000), - requires_openai_auth: false, + let mut stream = stream; + while let Some(item) = stream.next().await { + let mapped = match item { + Ok(event) => Ok(event), + Err(err) => Err(map_api_error(err)), }; - let otel_event_manager = otel_event_manager(); - - let out = run_sse(evs, provider, otel_event_manager).await; - assert_eq!(out.len(), case.expected_len, "case {}", case.name); - assert!( - (case.expect_first)(&out[0]), - "first event mismatch in case {}", - case.name - ); + if tx.send(mapped).await.is_err() { + break; + } } - } + }); - #[test] - fn test_try_parse_retry_after() { - let err = Error { - r#type: None, - message: Some("Rate limit reached for gpt-5 in organization org- on tokens per min (TPM): Limit 1, Used 1, Requested 19304. Please try again in 28ms. Visit https://platform.openai.com/account/rate-limits to learn more.".to_string()), - code: Some("rate_limit_exceeded".to_string()), - plan_type: None, - resets_at: None - }; + ResponseStream { rx_event: rx } +} - let delay = try_parse_retry_after(&err); - assert_eq!(delay, Some(Duration::from_millis(28))); - } - - #[test] - fn test_try_parse_retry_after_no_delay() { - let err = Error { - r#type: None, - message: Some("Rate limit reached for gpt-5 in organization on tokens per min (TPM): Limit 30000, Used 6899, Requested 24050. Please try again in 1.898s. Visit https://platform.openai.com/account/rate-limits to learn more.".to_string()), - code: Some("rate_limit_exceeded".to_string()), - plan_type: None, - resets_at: None - }; - let delay = try_parse_retry_after(&err); - assert_eq!(delay, Some(Duration::from_secs_f64(1.898))); - } - - #[test] - fn error_response_deserializes_schema_known_plan_type_and_serializes_back() { - use crate::token_data::KnownPlan; - use crate::token_data::PlanType; - - let json = - r#"{"error":{"type":"usage_limit_reached","plan_type":"pro","resets_at":1704067200}}"#; - let resp: ErrorResponse = serde_json::from_str(json).expect("should deserialize schema"); - - assert_matches!(resp.error.plan_type, Some(PlanType::Known(KnownPlan::Pro))); - - let plan_json = serde_json::to_string(&resp.error.plan_type).expect("serialize plan_type"); - assert_eq!(plan_json, "\"pro\""); - } - - #[test] - fn error_response_deserializes_schema_unknown_plan_type_and_serializes_back() { - use crate::token_data::PlanType; - - let json = - r#"{"error":{"type":"usage_limit_reached","plan_type":"vip","resets_at":1704067260}}"#; - let resp: ErrorResponse = serde_json::from_str(json).expect("should deserialize schema"); - - assert_matches!(resp.error.plan_type, Some(PlanType::Unknown(ref s)) if s == "vip"); - - let plan_json = serde_json::to_string(&resp.error.plan_type).expect("serialize plan_type"); - assert_eq!(plan_json, "\"vip\""); +fn map_api_error(err: codex_api_client::Error) -> CodexErr { + match err { + codex_api_client::Error::UnsupportedOperation(msg) => CodexErr::UnsupportedOperation(msg), + codex_api_client::Error::Http(source) => { + CodexErr::ConnectionFailed(ConnectionFailedError { source }) + } + codex_api_client::Error::ResponseStreamFailed { source, request_id } => { + CodexErr::ResponseStreamFailed(ResponseStreamFailed { source, request_id }) + } + codex_api_client::Error::Stream(message, delay) => CodexErr::Stream(message, delay), + codex_api_client::Error::UnexpectedStatus { status, body } => { + CodexErr::UnexpectedStatus(UnexpectedResponseError { + status, + body, + request_id: None, + }) + } + codex_api_client::Error::RetryLimit { status, request_id } => { + CodexErr::RetryLimit(RetryLimitReachedError { status, request_id }) + } + codex_api_client::Error::MissingEnvVar { var, instructions } => { + CodexErr::EnvVar(EnvVarError { var, instructions }) + } + codex_api_client::Error::Auth(message) => CodexErr::Fatal(message), + codex_api_client::Error::Json(err) => CodexErr::Json(err), + codex_api_client::Error::Other(message) => CodexErr::Fatal(message), } } diff --git a/codex-rs/core/src/client_common.rs b/codex-rs/core/src/client_common.rs index 861a9c90b7..bbf771bbb0 100644 --- a/codex-rs/core/src/client_common.rs +++ b/codex-rs/core/src/client_common.rs @@ -17,9 +17,6 @@ use codex_apply_patch::APPLY_PATCH_TOOL_INSTRUCTIONS; use codex_protocol::config_types::ReasoningEffort as ReasoningEffortConfig; use codex_protocol::config_types::ReasoningSummary as ReasoningSummaryConfig; use codex_protocol::config_types::Verbosity as VerbosityConfig; -use codex_protocol::models::ResponseItem; -use codex_protocol::protocol::RateLimitSnapshot; -use codex_protocol::protocol::TokenUsage; use serde_json::Value; use crate::model_family::ModelFamily; @@ -85,21 +82,7 @@ pub fn create_text_param_for_request( }) } -#[derive(Debug)] -pub enum ResponseEvent { - Created, - OutputItemDone(ResponseItem), - OutputItemAdded(ResponseItem), - Completed { - response_id: String, - token_usage: Option, - }, - OutputTextDelta(String), - ReasoningSummaryDelta(String), - ReasoningContentDelta(String), - ReasoningSummaryPartAdded, - RateLimits(RateLimitSnapshot), -} +pub use codex_api_client::ResponseEvent; pub struct ResponseStream { pub(crate) rx_event: mpsc::Receiver>, diff --git a/codex-rs/core/src/codex.rs b/codex-rs/core/src/codex.rs index d46b041eaf..0075bbcf7e 100644 --- a/codex-rs/core/src/codex.rs +++ b/codex-rs/core/src/codex.rs @@ -1949,19 +1949,16 @@ async fn run_turn( .get_model_family() .supports_parallel_tool_calls; let parallel_tool_calls = model_supports_parallel; - let prompt = Prompt::new( - request_items, - tools_json, - parallel_tool_calls, - turn_context.final_output_json_schema.clone(), - ); + let mut prompt = Prompt::default(); + prompt.instructions = instructions.clone(); + prompt.input = request_items; + prompt.tools = tools_json; + prompt.parallel_tool_calls = parallel_tool_calls; + prompt.output_schema = turn_context.final_output_json_schema.clone(); + prompt.store_response = store_response; + prompt.previous_response_id = previous_response_id.clone(); - let payload = StreamPayload { - prompt, - instructions, - store_response, - previous_response_id, - }; + let payload = StreamPayload { prompt }; let mut retries = 0; loop { @@ -2058,7 +2055,7 @@ async fn try_run_turn( full_prompt_items: Vec, cancellation_token: CancellationToken, ) -> CodexResult { - let chaining_intent = payload.store_response; + let chaining_intent = payload.prompt.store_response; let rollout_item = RolloutItem::TurnContext(TurnContextItem { cwd: turn_context.cwd.clone(), diff --git a/codex-rs/core/src/codex/compact.rs b/codex-rs/core/src/codex/compact.rs index c18a158d41..62fe1d599c 100644 --- a/codex-rs/core/src/codex/compact.rs +++ b/codex-rs/core/src/codex/compact.rs @@ -88,19 +88,21 @@ async fn run_compact_task_inner( let mut turn_input = history.get_history_for_prompt(); let turn_input_len = turn_input.len(); crate::conversation_history::format_prompt_items(&mut turn_input, false); - let prompt = Prompt::new(turn_input, Vec::new(), false, None); + let mut prompt = Prompt::default(); + prompt.input = turn_input; + prompt.tools = Vec::new(); + prompt.parallel_tool_calls = false; + prompt.output_schema = None; + prompt.store_response = false; let instructions = crate::client_common::compute_full_instructions( turn_context.base_instructions.as_deref(), &turn_context.client.get_model_family(), false, ) .into_owned(); - let payload = StreamPayload { - prompt, - instructions, - store_response: false, - previous_response_id: None, - }; + prompt.instructions = instructions.clone(); + prompt.previous_response_id = None; + let payload = StreamPayload { prompt }; let attempt_result = drain_to_completed(&sess, turn_context.as_ref(), payload).await; match attempt_result { diff --git a/codex-rs/core/src/config/mod.rs b/codex-rs/core/src/config/mod.rs index 3759a3adc0..688acde83f 100644 --- a/codex-rs/core/src/config/mod.rs +++ b/codex-rs/core/src/config/mod.rs @@ -1,4 +1,6 @@ +use crate::ModelProviderInfo; use crate::auth::AuthCredentialsStoreMode; +use crate::built_in_model_providers; use crate::config::types::DEFAULT_OTEL_ENVIRONMENT; use crate::config::types::History; use crate::config::types::McpServerConfig; @@ -25,8 +27,6 @@ use crate::git_info::resolve_root_git_project_for_trust; use crate::model_family::ModelFamily; use crate::model_family::derive_default_model_family; use crate::model_family::find_family_for_model; -use crate::model_provider_info::ModelProviderInfo; -use crate::model_provider_info::built_in_model_providers; use crate::openai_model_info::get_model_info; use crate::project_doc::DEFAULT_PROJECT_DOC_FILENAME; use crate::project_doc::LOCAL_PROJECT_DOC_FILENAME; diff --git a/codex-rs/core/src/default_client.rs b/codex-rs/core/src/default_client.rs index 8e4635460c..b29f06e29c 100644 --- a/codex-rs/core/src/default_client.rs +++ b/codex-rs/core/src/default_client.rs @@ -41,6 +41,14 @@ impl CodexHttpClient { Self { inner } } + pub fn inner(&self) -> &reqwest::Client { + &self.inner + } + + pub fn clone_inner(&self) -> reqwest::Client { + self.inner.clone() + } + pub fn get(&self, url: U) -> CodexRequestBuilder where U: IntoUrl, diff --git a/codex-rs/core/src/lib.rs b/codex-rs/core/src/lib.rs index b9bd97ca23..d303ccb7f7 100644 --- a/codex-rs/core/src/lib.rs +++ b/codex-rs/core/src/lib.rs @@ -8,7 +8,6 @@ mod apply_patch; pub mod auth; pub mod bash; -mod chat_completions; mod client; mod client_common; pub mod codex; @@ -32,7 +31,6 @@ pub mod mcp; mod mcp_connection_manager; mod mcp_tool_call; mod message_history; -mod model_provider_info; pub mod parse_command; mod response_processing; pub mod sandboxing; @@ -40,11 +38,11 @@ pub mod token_data; mod truncate; mod unified_exec; mod user_instructions; -pub use model_provider_info::BUILT_IN_OSS_MODEL_PROVIDER_ID; -pub use model_provider_info::ModelProviderInfo; -pub use model_provider_info::WireApi; -pub use model_provider_info::built_in_model_providers; -pub use model_provider_info::create_oss_provider_with_base_url; +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 conversation_manager; mod event_mapping; pub mod review_format; diff --git a/codex-rs/core/src/sandboxing/assessment.rs b/codex-rs/core/src/sandboxing/assessment.rs index 28013e2723..1f0a1119c5 100644 --- a/codex-rs/core/src/sandboxing/assessment.rs +++ b/codex-rs/core/src/sandboxing/assessment.rs @@ -128,12 +128,12 @@ pub(crate) async fn assess_command( }]; crate::conversation_history::format_prompt_items(&mut prompt_items, false); - let prompt = Prompt::new( - prompt_items, - Vec::new(), - false, - Some(sandbox_assessment_schema()), - ); + let mut prompt = Prompt::default(); + prompt.input = prompt_items; + prompt.tools = Vec::new(); + prompt.parallel_tool_calls = false; + prompt.output_schema = Some(sandbox_assessment_schema()); + prompt.store_response = false; let instructions = crate::client_common::compute_full_instructions( Some(system_prompt.as_str()), &config.model_family, @@ -141,12 +141,8 @@ pub(crate) async fn assess_command( ) .into_owned(); - let payload = StreamPayload { - prompt, - instructions, - store_response: false, - previous_response_id: None, - }; + prompt.instructions = instructions.clone(); + let payload = StreamPayload { prompt }; let child_otel = parent_otel.with_model(config.model.as_str(), config.model_family.slug.as_str()); diff --git a/codex-rs/core/src/tools/spec.rs b/codex-rs/core/src/tools/spec.rs index c47937b4d6..24e887c6f1 100644 --- a/codex-rs/core/src/tools/spec.rs +++ b/codex-rs/core/src/tools/spec.rs @@ -710,9 +710,6 @@ pub(crate) struct ApplyPatchToolArgs { pub(crate) input: String, } -/// Returns JSON values that are compatible with Function Calling in the -/// Responses API: -/// https://platform.openai.com/docs/guides/function-calling?api-mode=responses pub fn create_tools_json_for_responses_api( tools: &[ToolSpec], ) -> crate::error::Result> { @@ -736,31 +733,6 @@ pub fn tools_metadata_for_prompt( }); Ok((tools_json, has_freeform_apply_patch)) } -/// Returns JSON values that are compatible with Function Calling in the -/// Chat Completions API: -/// https://platform.openai.com/docs/guides/function-calling?api-mode=chat -pub(crate) fn create_tools_json_for_chat_completions_api( - tools: &[serde_json::Value], -) -> crate::error::Result> { - let tools_json = tools - .iter() - .filter_map(|tool| { - if tool.get("type") != Some(&serde_json::Value::String("function".to_string())) { - return None; - } - - tool.as_object().map(|map| { - let mut function = map.clone(); - function.remove("type"); - json!({ - "type": "function", - "function": function, - }) - }) - }) - .collect::>(); - Ok(tools_json) -} pub(crate) fn mcp_tool_to_openai_tool( fully_qualified_name: String,