openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-09-14 11:57:03 +00:00
Code Issues Packages Projects Releases Wiki Activity

R2

Browse Source
This commit is contained in:
jif-oai
2025-11-11 14:58:19 +00:00
parent f6494aa85c
commit 0c609d441b
10 changed files with 224 additions and 43 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
codex-rs/api-client/src/aggregate.rs
View File

@@ -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)))) => {

1
codex-rs/api-client/src/chat.rs
View File

@@ -1,4 +1,3 @@
use async_trait::async_trait;
use codex_otel::otel_event_manager::OtelEventManager;
use codex_protocol::protocol::SessionSource;

9
codex-rs/api-client/src/common.rs
View File

@@ -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(

4
codex-rs/api-client/src/decode/responses.rs
View File

@@ -34,9 +34,7 @@ pub struct ErrorBody {
pub resets_at: Option<i64>,
}
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 {

2
codex-rs/api-client/src/lib.rs
View File

@@ -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;

6
codex-rs/api-client/src/responses.rs
View File

@@ -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))
}
}

73
codex-rs/api-client/src/routed_client.rs
View File

@@ -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<WireResponseStream> {
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<WireEvent> {
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<codex_protocol::protocol::RateLimitWindow>| -> Option<crate::stream::WireRateLimitWindow> {
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),
})
}
})
}

40
codex-rs/api-client/src/stream.rs
View File

@@ -81,3 +81,43 @@ impl<T> Stream for EventStream<T> {
}
pub type ResponseStream = EventStream<Result<ResponseEvent>>;
#[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<f64>,
pub window_minutes: Option<i64>,
pub resets_at: Option<i64>,
}
#[derive(Debug, Clone)]
pub struct WireRateLimitSnapshot {
pub primary: Option<WireRateLimitWindow>,
pub secondary: Option<WireRateLimitWindow>,
}
#[derive(Debug)]
pub enum WireEvent {
Created,
OutputItemDone(serde_json::Value),
OutputItemAdded(serde_json::Value),
Completed {
response_id: String,
token_usage: Option<WireTokenUsage>,
},
OutputTextDelta(String),
ReasoningSummaryDelta(String),
ReasoningContentDelta(String),
ReasoningSummaryPartAdded,
RateLimits(WireRateLimitSnapshot),
}
pub type WireResponseStream = EventStream<Result<WireEvent>>;

72
codex-rs/core/src/client.rs
View File

@@ -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<RoutedApiClient> {
@@ -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::<Result<ResponseEvent>>(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<WireRateLimitWindow>| -> Option<RateLimitWindow> {
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::<codex_protocol::models::ResponseItem>(v)
.unwrap_or(codex_protocol::models::ResponseItem::Other);
ResponseEvent::OutputItemAdded(item)
}
codex_api_client::WireEvent::OutputItemDone(v) => {
let item = serde_json::from_value::<codex_protocol::models::ResponseItem>(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),

52
codex-rs/core/src/wire_payload.rs
View File

@@ -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::<Value>::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<Value> = 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 { .. } => {}
}
}