From 0c609d441b2a50480ea880f8c3aa2aed1d74d00e Mon Sep 17 00:00:00 2001 From: jif-oai Date: Tue, 11 Nov 2025 14:58:19 +0000 Subject: [PATCH] R2 --- codex-rs/api-client/src/aggregate.rs | 8 +-- codex-rs/api-client/src/chat.rs | 1 - codex-rs/api-client/src/common.rs | 9 --- codex-rs/api-client/src/decode/responses.rs | 4 +- codex-rs/api-client/src/lib.rs | 2 + codex-rs/api-client/src/responses.rs | 6 -- codex-rs/api-client/src/routed_client.rs | 73 +++++++++++++++++++++ codex-rs/api-client/src/stream.rs | 40 +++++++++++ codex-rs/core/src/client.rs | 72 ++++++++++++++++++-- codex-rs/core/src/wire_payload.rs | 52 +++++++++++---- 10 files changed, 224 insertions(+), 43 deletions(-) diff --git a/codex-rs/api-client/src/aggregate.rs b/codex-rs/api-client/src/aggregate.rs index 080abfa414..44c02e1f88 100644 --- a/codex-rs/api-client/src/aggregate.rs +++ b/codex-rs/api-client/src/aggregate.rs @@ -106,9 +106,7 @@ where } } } else { - return Poll::Ready(Some(Ok(ResponseEvent::OutputItemDone( - item, - )))); + return Poll::Ready(Some(Ok(ResponseEvent::OutputItemDone(item)))); } } Poll::Ready(Some(Ok(ResponseEvent::OutputItemAdded(item)))) => { @@ -116,9 +114,7 @@ where &item, ResponseItem::Message { role, .. } if role == "assistant" ) { - return Poll::Ready(Some(Ok(ResponseEvent::OutputItemAdded( - item, - )))); + return Poll::Ready(Some(Ok(ResponseEvent::OutputItemAdded(item)))); } } Poll::Ready(Some(Ok(ResponseEvent::ReasoningContentDelta(delta)))) => { diff --git a/codex-rs/api-client/src/chat.rs b/codex-rs/api-client/src/chat.rs index 20de2b7953..4edf267eb6 100644 --- a/codex-rs/api-client/src/chat.rs +++ b/codex-rs/api-client/src/chat.rs @@ -1,4 +1,3 @@ - use async_trait::async_trait; use codex_otel::otel_event_manager::OtelEventManager; use codex_protocol::protocol::SessionSource; diff --git a/codex-rs/api-client/src/common.rs b/codex-rs/api-client/src/common.rs index 0a08f42326..560da36c8e 100644 --- a/codex-rs/api-client/src/common.rs +++ b/codex-rs/api-client/src/common.rs @@ -1,15 +1,6 @@ -use std::time::Duration; - use codex_protocol::protocol::SessionSource; use codex_protocol::protocol::SubAgentSource; -/// Exponential backoff with a 100ms base and a cap on the exponent to avoid -/// unbounded growth. The attempt number is clamped to [0, 6]. -pub(crate) fn backoff(attempt: i64) -> Duration { - let capped = attempt.clamp(0, 6) as u32; - Duration::from_millis(100 * 2_i64.pow(capped) as u64) -} - /// Apply the `x-openai-subagent` header when the session source indicates a /// subagent. Returns the original builder unchanged when not applicable. pub(crate) fn apply_subagent_header( diff --git a/codex-rs/api-client/src/decode/responses.rs b/codex-rs/api-client/src/decode/responses.rs index 0e8c07e53b..4d4c15e760 100644 --- a/codex-rs/api-client/src/decode/responses.rs +++ b/codex-rs/api-client/src/decode/responses.rs @@ -34,9 +34,7 @@ pub struct ErrorBody { pub resets_at: Option, } -pub fn is_quota_exceeded_error(error: &ErrorBody) -> bool { - error.code.as_deref() == Some("quota_exceeded") -} +// legacy helper removed; decoupled error handling in core #[derive(Debug, Deserialize)] pub struct StreamEvent { diff --git a/codex-rs/api-client/src/lib.rs b/codex-rs/api-client/src/lib.rs index d3a78611bd..ab3240d624 100644 --- a/codex-rs/api-client/src/lib.rs +++ b/codex-rs/api-client/src/lib.rs @@ -30,6 +30,8 @@ pub use crate::stream::ResponseStream; pub use crate::stream::TextControls; pub use crate::stream::TextFormat; pub use crate::stream::TextFormatType; +pub use crate::stream::WireEvent; +pub use crate::stream::WireResponseStream; pub use codex_provider_config::BUILT_IN_OSS_MODEL_PROVIDER_ID; pub use codex_provider_config::ModelProviderInfo; pub use codex_provider_config::WireApi; diff --git a/codex-rs/api-client/src/responses.rs b/codex-rs/api-client/src/responses.rs index f62ad26baa..c26f9f9959 100644 --- a/codex-rs/api-client/src/responses.rs +++ b/codex-rs/api-client/src/responses.rs @@ -1,12 +1,10 @@ use std::sync::Arc; -use std::time::Duration; use async_trait::async_trait; use codex_app_server_protocol::AuthMode; use codex_otel::otel_event_manager::OtelEventManager; use codex_protocol::ConversationId; use futures::TryStreamExt; -use reqwest::StatusCode; use serde_json::Value; use tokio::sync::mpsc; use tracing::debug; @@ -14,8 +12,6 @@ use tracing::trace; use crate::api::PayloadClient; use crate::auth::AuthProvider; -use crate::common::backoff; -use crate::decode::responses::ErrorResponse; use crate::error::Error; use crate::error::Result; use crate::stream::ResponseEvent; @@ -45,7 +41,6 @@ pub struct ResponsesApiClient { config: ResponsesApiClientConfig, } - #[async_trait] impl PayloadClient for ResponsesApiClient { type Config = ResponsesApiClientConfig; @@ -145,4 +140,3 @@ impl PayloadClient for ResponsesApiClient { Ok(crate::stream::EventStream::from_receiver(rx_event)) } } - diff --git a/codex-rs/api-client/src/routed_client.rs b/codex-rs/api-client/src/routed_client.rs index 5c0e8dc0c0..0ca4def2b6 100644 --- a/codex-rs/api-client/src/routed_client.rs +++ b/codex-rs/api-client/src/routed_client.rs @@ -12,9 +12,12 @@ use crate::ResponsesApiClient; use crate::ResponsesApiClientConfig; use crate::Result; use crate::WireApi; +use crate::WireEvent; +use crate::WireResponseStream; use crate::api::PayloadClient; use crate::auth::AuthProvider; use crate::client::fixtures::stream_from_fixture; +use crate::stream::WireTokenUsage; use codex_provider_config::ModelProviderInfo; /// Dispatches to the appropriate API client implementation based on the provider wire API. @@ -80,6 +83,25 @@ impl RoutedApiClient { } } } + + pub async fn stream_payload_wire( + &self, + payload_json: &serde_json::Value, + ) -> Result { + use futures::StreamExt; + let legacy = self.stream_payload(payload_json).await?; + let (tx, rx) = tokio::sync::mpsc::channel(1600); + tokio::spawn(async move { + futures::pin_mut!(legacy); + while let Some(item) = legacy.next().await { + let converted = item.and_then(|ev| map_response_event_to_wire(ev)); + if tx.send(converted).await.is_err() { + break; + } + } + }); + Ok(crate::stream::EventStream::from_receiver(rx)) + } } #[async_trait::async_trait] @@ -98,3 +120,54 @@ impl PayloadClient for RoutedApiClient { self.stream_payload(payload_json).await } } + +fn map_response_event_to_wire(ev: crate::stream::ResponseEvent) -> Result { + Ok(match ev { + crate::stream::ResponseEvent::Created => WireEvent::Created, + crate::stream::ResponseEvent::OutputItemDone(item) => { + WireEvent::OutputItemDone(serde_json::to_value(item).unwrap_or(serde_json::Value::Null)) + } + crate::stream::ResponseEvent::OutputItemAdded(item) => WireEvent::OutputItemAdded( + serde_json::to_value(item).unwrap_or(serde_json::Value::Null), + ), + crate::stream::ResponseEvent::Completed { + response_id, + token_usage, + } => { + let mapped = token_usage.map(|u| WireTokenUsage { + input_tokens: u.input_tokens, + cached_input_tokens: u.cached_input_tokens, + output_tokens: u.output_tokens, + reasoning_output_tokens: u.reasoning_output_tokens, + total_tokens: u.total_tokens, + }); + WireEvent::Completed { + response_id, + token_usage: mapped, + } + } + crate::stream::ResponseEvent::OutputTextDelta(s) => WireEvent::OutputTextDelta(s), + crate::stream::ResponseEvent::ReasoningSummaryDelta(s) => { + WireEvent::ReasoningSummaryDelta(s) + } + crate::stream::ResponseEvent::ReasoningContentDelta(s) => { + WireEvent::ReasoningContentDelta(s) + } + crate::stream::ResponseEvent::ReasoningSummaryPartAdded => { + WireEvent::ReasoningSummaryPartAdded + } + crate::stream::ResponseEvent::RateLimits(s) => { + let to_win = |w: Option| -> Option { + w.map(|w| crate::stream::WireRateLimitWindow { + used_percent: Some(w.used_percent), + window_minutes: w.window_minutes, + resets_at: w.resets_at, + }) + }; + WireEvent::RateLimits(crate::stream::WireRateLimitSnapshot { + primary: to_win(s.primary), + secondary: to_win(s.secondary), + }) + } + }) +} diff --git a/codex-rs/api-client/src/stream.rs b/codex-rs/api-client/src/stream.rs index ac76b282fa..100bf45edf 100644 --- a/codex-rs/api-client/src/stream.rs +++ b/codex-rs/api-client/src/stream.rs @@ -81,3 +81,43 @@ impl Stream for EventStream { } pub type ResponseStream = EventStream>; + +#[derive(Debug, Clone)] +pub struct WireTokenUsage { + pub input_tokens: i64, + pub cached_input_tokens: i64, + pub output_tokens: i64, + pub reasoning_output_tokens: i64, + pub total_tokens: i64, +} + +#[derive(Debug, Clone)] +pub struct WireRateLimitWindow { + pub used_percent: Option, + pub window_minutes: Option, + pub resets_at: Option, +} + +#[derive(Debug, Clone)] +pub struct WireRateLimitSnapshot { + pub primary: Option, + pub secondary: Option, +} + +#[derive(Debug)] +pub enum WireEvent { + Created, + OutputItemDone(serde_json::Value), + OutputItemAdded(serde_json::Value), + Completed { + response_id: String, + token_usage: Option, + }, + OutputTextDelta(String), + ReasoningSummaryDelta(String), + ReasoningContentDelta(String), + ReasoningSummaryPartAdded, + RateLimits(WireRateLimitSnapshot), +} + +pub type WireResponseStream = EventStream>; diff --git a/codex-rs/core/src/client.rs b/codex-rs/core/src/client.rs index 6029be2235..6c277bc7c8 100644 --- a/codex-rs/core/src/client.rs +++ b/codex-rs/core/src/client.rs @@ -9,6 +9,7 @@ use codex_api_client::Result as ApiClientResult; use codex_api_client::RoutedApiClient; use codex_api_client::RoutedApiClientConfig; use codex_api_client::WireApi; +use codex_api_client::stream::WireRateLimitWindow; use codex_otel::otel_event_manager::OtelEventManager; use codex_protocol::ConversationId; use codex_protocol::config_types::ReasoningEffort as ReasoningEffortConfig; @@ -42,8 +43,6 @@ use crate::model_family::ModelFamily; use crate::openai_model_info::get_model_info; use crate::token_data::KnownPlan; use crate::token_data::PlanType; -use crate::tools::spec::create_tools_json_for_chat_completions_api; -use crate::tools::spec::create_tools_json_for_responses_api; #[derive(Clone)] pub struct ModelClient { @@ -153,7 +152,9 @@ impl ModelClient { text_controls, instructions, ), - WireApi::Chat => crate::wire_payload::build_chat_payload(prompt, &self.config.model, instructions), + WireApi::Chat => { + crate::wire_payload::build_chat_payload(prompt, &self.config.model, instructions) + } }; let client = self @@ -163,10 +164,10 @@ impl ModelClient { .map_err(map_api_error)?; let api_stream = client - .stream_payload(&payload_json) + .stream_payload_wire(&payload_json) .await .map_err(map_api_error)?; - Ok(wrap_stream(api_stream)) + Ok(wrap_wire_stream(api_stream)) } async fn build_api_client(&self) -> ApiClientResult { @@ -263,13 +264,13 @@ impl AuthProvider for AuthManagerProvider { } } -fn wrap_stream(stream: codex_api_client::ResponseStream) -> ResponseStream { +fn wrap_wire_stream(stream: codex_api_client::WireResponseStream) -> ResponseStream { let (tx, rx) = mpsc::channel::>(1600); tokio::spawn(async move { let mut stream = stream; while let Some(item) = stream.next().await { - let mapped = item.map_err(map_api_error); + let mapped = item.map(|ev| map_wire_event(ev)).map_err(map_api_error); if tx.send(mapped).await.is_err() { break; } @@ -279,6 +280,63 @@ fn wrap_stream(stream: codex_api_client::ResponseStream) -> ResponseStream { codex_api_client::EventStream::from_receiver(rx) } +fn map_wire_event(ev: codex_api_client::WireEvent) -> ResponseEvent { + match ev { + codex_api_client::WireEvent::Created => ResponseEvent::Created, + codex_api_client::WireEvent::OutputTextDelta(s) => ResponseEvent::OutputTextDelta(s), + codex_api_client::WireEvent::ReasoningSummaryDelta(s) => { + ResponseEvent::ReasoningSummaryDelta(s) + } + codex_api_client::WireEvent::ReasoningContentDelta(s) => { + ResponseEvent::ReasoningContentDelta(s) + } + codex_api_client::WireEvent::ReasoningSummaryPartAdded => { + ResponseEvent::ReasoningSummaryPartAdded + } + codex_api_client::WireEvent::RateLimits(w) => { + use codex_protocol::protocol::RateLimitSnapshot; + use codex_protocol::protocol::RateLimitWindow; + let to_win = |ow: Option| -> Option { + ow.map(|w| RateLimitWindow { + used_percent: w.used_percent.unwrap_or(0.0), + window_minutes: w.window_minutes, + resets_at: w.resets_at, + }) + }; + ResponseEvent::RateLimits(RateLimitSnapshot { + primary: to_win(w.primary), + secondary: to_win(w.secondary), + }) + } + codex_api_client::WireEvent::Completed { + response_id, + token_usage, + } => { + let mapped = token_usage.map(|u| codex_protocol::protocol::TokenUsage { + input_tokens: u.input_tokens, + cached_input_tokens: u.cached_input_tokens, + output_tokens: u.output_tokens, + reasoning_output_tokens: u.reasoning_output_tokens, + total_tokens: u.total_tokens, + }); + ResponseEvent::Completed { + response_id, + token_usage: mapped, + } + } + codex_api_client::WireEvent::OutputItemAdded(v) => { + let item = serde_json::from_value::(v) + .unwrap_or(codex_protocol::models::ResponseItem::Other); + ResponseEvent::OutputItemAdded(item) + } + codex_api_client::WireEvent::OutputItemDone(v) => { + let item = serde_json::from_value::(v) + .unwrap_or(codex_protocol::models::ResponseItem::Other); + ResponseEvent::OutputItemDone(item) + } + } +} + fn map_api_error(err: codex_api_client::Error) -> CodexErr { match err { codex_api_client::Error::UnsupportedOperation(msg) => CodexErr::UnsupportedOperation(msg), diff --git a/codex-rs/core/src/wire_payload.rs b/codex-rs/core/src/wire_payload.rs index 9a9228f29f..565a0afd15 100644 --- a/codex-rs/core/src/wire_payload.rs +++ b/codex-rs/core/src/wire_payload.rs @@ -108,11 +108,11 @@ fn attach_item_ids_array(json_array: &mut [Value], prompt_input: &[ResponseItem] } pub fn build_chat_payload(prompt: &Prompt, model: &str, instructions: String) -> Value { + use crate::tools::spec::create_tools_json_for_chat_completions_api; use codex_protocol::models::ContentItem; use codex_protocol::models::FunctionCallOutputContentItem; use codex_protocol::models::ReasoningItemContent; use std::collections::HashMap; - use crate::tools::spec::create_tools_json_for_chat_completions_api; let mut messages = Vec::::new(); messages.push(json!({ "role": "system", "content": instructions })); @@ -153,7 +153,11 @@ pub fn build_chat_payload(prompt: &Prompt, model: &str, instructions: String) -> continue; } - if let ResponseItem::Reasoning { content: Some(items), .. } = item { + if let ResponseItem::Reasoning { + content: Some(items), + .. + } = item + { let mut text = String::new(); for entry in items { match entry { @@ -210,7 +214,8 @@ pub fn build_chat_payload(prompt: &Prompt, model: &str, instructions: String) -> for c in content { match c { - ContentItem::InputText { text: t } | ContentItem::OutputText { text: t } => { + ContentItem::InputText { text: t } + | ContentItem::OutputText { text: t } => { text.push_str(t); items.push(json!({"type":"text","text": t})); } @@ -246,7 +251,12 @@ pub fn build_chat_payload(prompt: &Prompt, model: &str, instructions: String) -> } messages.push(message); } - ResponseItem::FunctionCall { name, arguments, call_id, .. } => { + ResponseItem::FunctionCall { + name, + arguments, + call_id, + .. + } => { messages.push(json!({ "role": "assistant", "tool_calls": [{ @@ -261,17 +271,28 @@ pub fn build_chat_payload(prompt: &Prompt, model: &str, instructions: String) -> 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}}), + 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 })); + messages.push( + json!({ "role": "tool", "tool_call_id": call_id, "content": content_value }), + ); } - ResponseItem::LocalShellCall { id, call_id, action, .. } => { + ResponseItem::LocalShellCall { + id, + call_id, + action, + .. + } => { let tool_id = call_id .clone() .filter(|value| !value.is_empty()) @@ -289,7 +310,12 @@ pub fn build_chat_payload(prompt: &Prompt, model: &str, instructions: String) -> }], })); } - ResponseItem::CustomToolCall { call_id, name, input, .. } => { + ResponseItem::CustomToolCall { + call_id, + name, + input, + .. + } => { messages.push(json!({ "role": "assistant", "tool_calls": [{ @@ -300,9 +326,13 @@ pub fn build_chat_payload(prompt: &Prompt, model: &str, instructions: String) -> })); } ResponseItem::CustomToolCallOutput { call_id, output } => { - messages.push(json!({ "role": "tool", "tool_call_id": call_id, "content": output })); + messages + .push(json!({ "role": "tool", "tool_call_id": call_id, "content": output })); } - ResponseItem::WebSearchCall { .. } | ResponseItem::Reasoning { .. } | ResponseItem::Other | ResponseItem::GhostSnapshot { .. } => {} + ResponseItem::WebSearchCall { .. } + | ResponseItem::Reasoning { .. } + | ResponseItem::Other + | ResponseItem::GhostSnapshot { .. } => {} } }