diff --git a/codex-rs/api-client/src/api.rs b/codex-rs/api-client/src/api.rs deleted file mode 100644 index 048c357e3c..0000000000 --- a/codex-rs/api-client/src/api.rs +++ /dev/null @@ -1,20 +0,0 @@ -use async_trait::async_trait; - -use crate::error::Result; -use crate::stream::ResponseStream; -use codex_protocol::protocol::SessionSource; -use serde_json::Value; - -#[async_trait] -pub trait PayloadClient: Sized { - type Config; - - fn new(config: Self::Config) -> Result; - - /// Start a streaming request for a pre-built wire JSON payload. - async fn stream_payload( - &self, - payload_json: &Value, - session_source: Option<&SessionSource>, - ) -> Result; -} diff --git a/codex-rs/api-client/src/chat.rs b/codex-rs/api-client/src/chat.rs index 5f727baf1e..0efe20a42a 100644 --- a/codex-rs/api-client/src/chat.rs +++ b/codex-rs/api-client/src/chat.rs @@ -1,16 +1,9 @@ -use async_trait::async_trait; -use codex_otel::otel_event_manager::OtelEventManager; -use codex_protocol::protocol::SessionSource; -use futures::TryStreamExt; -use tokio::sync::mpsc; - -use crate::api::PayloadClient; use crate::error::Error; use crate::error::Result; -use crate::stream::ResponseEvent; -use crate::stream::ResponseStream; use crate::stream::WireResponseStream; +use codex_otel::otel_event_manager::OtelEventManager; use codex_provider_config::ModelProviderInfo; +use futures::TryStreamExt; #[derive(Clone)] /// Configuration for the Chat Completions client (OpenAI-compatible `/v1/chat/completions`). @@ -25,7 +18,6 @@ pub struct ChatCompletionsApiClientConfig { pub provider: ModelProviderInfo, pub model: String, pub otel_event_manager: OtelEventManager, - pub session_source: SessionSource, pub extra_headers: Vec<(String, String)>, } @@ -34,23 +26,16 @@ pub struct ChatCompletionsApiClient { config: ChatCompletionsApiClientConfig, } -// prompt-based API removed; use PayloadClient::stream_payload instead - -// prompt-based API removed - -#[async_trait] -impl PayloadClient for ChatCompletionsApiClient { - type Config = ChatCompletionsApiClientConfig; - - fn new(config: Self::Config) -> Result { +impl ChatCompletionsApiClient { + pub fn new(config: ChatCompletionsApiClientConfig) -> Result { Ok(Self { config }) } - async fn stream_payload( + pub async fn stream_payload_wire( &self, payload_json: &serde_json::Value, - session_source: Option<&codex_protocol::protocol::SessionSource>, - ) -> Result { + _session_source: Option<&codex_protocol::protocol::SessionSource>, + ) -> Result { if self.config.provider.wire_api != codex_provider_config::WireApi::Chat { return Err(crate::error::Error::UnsupportedOperation( "ChatCompletionsApiClient requires a Chat provider".to_string(), @@ -68,7 +53,6 @@ impl PayloadClient for ChatCompletionsApiClient { &self.config.http_client, &self.config.provider, &auth, - session_source, &extra_headers, ) .await?; @@ -83,7 +67,8 @@ impl PayloadClient for ChatCompletionsApiClient { .log_request(0, || req_builder.send()) .await?; - let (tx_event, rx_event) = mpsc::channel::>(1600); + let (tx_event, rx_event) = + tokio::sync::mpsc::channel::>(1600); let stream = res .bytes_stream() .map_err(|err| Error::ResponseStreamFailed { @@ -92,36 +77,14 @@ impl PayloadClient for ChatCompletionsApiClient { }); let idle_timeout = self.config.provider.stream_idle_timeout(); let otel = self.config.otel_event_manager.clone(); - tokio::spawn(crate::client::sse::process_sse( + tokio::spawn(crate::client::sse::process_sse_wire( stream, tx_event, idle_timeout, otel, - crate::decode::chat::ChatSseDecoder::new(), + crate::decode_wire::chat::WireChatSseDecoder::new(), )); Ok(crate::stream::EventStream::from_receiver(rx_event)) } } - -impl ChatCompletionsApiClient { - pub async fn stream_payload_wire( - &self, - payload_json: &serde_json::Value, - session_source: Option<&codex_protocol::protocol::SessionSource>, - ) -> Result { - use futures::StreamExt; - let legacy = self.stream_payload(payload_json, session_source).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| crate::wire::map_response_event_to_wire(ev)); - if tx.send(converted).await.is_err() { - break; - } - } - }); - Ok(crate::stream::EventStream::from_receiver(rx)) - } -} diff --git a/codex-rs/api-client/src/client/fixtures.rs b/codex-rs/api-client/src/client/fixtures.rs index d638ab3fbd..a0e4c28488 100644 --- a/codex-rs/api-client/src/client/fixtures.rs +++ b/codex-rs/api-client/src/client/fixtures.rs @@ -8,16 +8,14 @@ use tokio_util::io::ReaderStream; use crate::error::Error; use crate::error::Result; -use crate::stream::ResponseEvent; -use crate::stream::ResponseStream; use codex_provider_config::ModelProviderInfo; -pub async fn stream_from_fixture( +pub async fn stream_from_fixture_wire( path: impl AsRef, provider: ModelProviderInfo, otel_event_manager: OtelEventManager, -) -> Result { - let (tx_event, rx_event) = mpsc::channel::>(1600); +) -> 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(|err| Error::Other(format!("failed to open fixture {display_path}: {err}")))?; @@ -34,12 +32,12 @@ pub async fn stream_from_fixture( let rdr = std::io::Cursor::new(content); let stream = ReaderStream::new(rdr).map_err(|err| Error::Other(err.to_string())); - tokio::spawn(crate::client::sse::process_sse( + tokio::spawn(crate::client::sse::process_sse_wire( stream, tx_event, provider.stream_idle_timeout(), otel_event_manager, - crate::decode::responses::ResponsesSseDecoder, + crate::decode_wire::responses::WireResponsesSseDecoder, )); - Ok(ResponseStream { rx_event }) + Ok(crate::stream::EventStream::from_receiver(rx_event)) } diff --git a/codex-rs/api-client/src/client/http.rs b/codex-rs/api-client/src/client/http.rs index b9c9281c47..a6e44230c9 100644 --- a/codex-rs/api-client/src/client/http.rs +++ b/codex-rs/api-client/src/client/http.rs @@ -1,10 +1,7 @@ use std::sync::Arc; -use codex_protocol::protocol::SessionSource; - use crate::auth::AuthContext; use crate::auth::AuthProvider; -use crate::common::apply_subagent_header; use crate::error::Result; use codex_provider_config::ModelProviderInfo; @@ -13,7 +10,6 @@ pub async fn build_request( http_client: &reqwest::Client, provider: &ModelProviderInfo, auth: &Option, - session_source: Option<&SessionSource>, extra_headers: &[(&str, String)], ) -> Result { let mut builder = provider @@ -36,7 +32,6 @@ pub async fn build_request( } => instructions.clone(), }, })?; - builder = apply_subagent_header(builder, session_source); for (name, value) in extra_headers { builder = builder.header(*name, value); } diff --git a/codex-rs/api-client/src/client/mod.rs b/codex-rs/api-client/src/client/mod.rs index e7ebea0915..67be6c7b4b 100644 --- a/codex-rs/api-client/src/client/mod.rs +++ b/codex-rs/api-client/src/client/mod.rs @@ -3,21 +3,22 @@ use codex_otel::otel_event_manager::OtelEventManager; use tokio::sync::mpsc; use crate::error::Result; -use crate::stream::ResponseEvent; +use crate::stream::WireEvent; pub mod fixtures; pub mod http; pub mod rate_limits; pub mod sse; -/// Decodes framed SSE JSON into ResponseEvent(s). -/// Implementations may keep state across frames (e.g., Chat function-call state). +// Legacy ResponseEvent-based decoder removed + +/// Decodes framed SSE JSON into WireEvent(s). #[async_trait] -pub trait ResponseDecoder { +pub trait WireResponseDecoder { async fn on_frame( &mut self, json: &str, - tx: &mpsc::Sender>, + tx: &mpsc::Sender>, otel: &OtelEventManager, ) -> Result<()>; } diff --git a/codex-rs/api-client/src/client/sse.rs b/codex-rs/api-client/src/client/sse.rs index 6d44cea363..70b31681ee 100644 --- a/codex-rs/api-client/src/client/sse.rs +++ b/codex-rs/api-client/src/client/sse.rs @@ -7,22 +7,24 @@ use futures::StreamExt; use tokio::sync::mpsc; use tokio::time::timeout; -use crate::client::ResponseDecoder; use crate::error::Error; use crate::error::Result; -use crate::stream::ResponseEvent; +// Legacy ResponseEvent-based SSE framer removed +use crate::stream::WireEvent; -/// Generic SSE framer: turns a Byte stream into framed JSON and delegates to a ResponseDecoder. +// Legacy ResponseEvent-based SSE framer removed + +/// Generic SSE framer for wire events: Byte stream -> framed JSON -> WireResponseDecoder. #[allow(clippy::too_many_arguments)] -pub async fn process_sse( +pub async fn process_sse_wire( stream: S, - tx_event: mpsc::Sender>, + tx_event: mpsc::Sender>, max_idle_duration: Duration, otel_event_manager: OtelEventManager, mut decoder: D, ) where S: Stream> + Send + 'static + Unpin, - D: ResponseDecoder + Send, + D: crate::client::WireResponseDecoder + Send, { let mut stream = stream; let mut data_buffer = String::new(); @@ -60,20 +62,18 @@ pub async fn process_sse( if let Some(tail) = line.strip_prefix("data:") { data_buffer.push_str(tail.trim_start()); } else if !line.is_empty() && !data_buffer.is_empty() { - // Continuation of a long data: line split across chunks; append raw. data_buffer.push_str(line); } if line.is_empty() && !data_buffer.is_empty() { - // One full JSON frame ready – delegate to decoder - if let Err(err) = decoder - .on_frame(&data_buffer, &tx_event, &otel_event_manager) + let json = std::mem::take(&mut data_buffer); + if let Err(e) = decoder + .on_frame(&json, &tx_event, &otel_event_manager) .await { - let _ = tx_event.send(Err(err)).await; + let _ = tx_event.send(Err(e)).await; return; } - data_buffer.clear(); } } } diff --git a/codex-rs/api-client/src/common.rs b/codex-rs/api-client/src/common.rs deleted file mode 100644 index 560da36c8e..0000000000 --- a/codex-rs/api-client/src/common.rs +++ /dev/null @@ -1,22 +0,0 @@ -use codex_protocol::protocol::SessionSource; -use codex_protocol::protocol::SubAgentSource; - -/// 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( - mut builder: reqwest::RequestBuilder, - session_source: Option<&SessionSource>, -) -> reqwest::RequestBuilder { - if let Some(SessionSource::SubAgent(sub)) = 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()) - }; - builder = builder.header("x-openai-subagent", subagent); - } - builder -} diff --git a/codex-rs/api-client/src/decode/responses.rs b/codex-rs/api-client/src/decode/responses.rs deleted file mode 100644 index 4d4c15e760..0000000000 --- a/codex-rs/api-client/src/decode/responses.rs +++ /dev/null @@ -1,490 +0,0 @@ -use async_trait::async_trait; -use codex_otel::otel_event_manager::OtelEventManager; -use codex_protocol::models::ResponseItem; -use codex_protocol::protocol::TokenUsage; -use serde::Deserialize; -use serde::Serialize; -use serde_json::Value; -use std::time::Duration; -use tokio::sync::mpsc; -use tracing::debug; -use tracing::trace; - -use crate::error::Error; -use crate::error::Result; -use crate::stream::ResponseEvent; - -#[derive(Debug, Deserialize)] -pub struct StreamResponseCompleted { - pub id: String, - pub usage: Option, -} - -#[derive(Debug, Deserialize)] -pub struct ErrorResponse { - pub error: ErrorBody, -} - -#[derive(Debug, Deserialize, Serialize)] -pub struct ErrorBody { - pub r#type: Option, - pub code: Option, - pub message: Option, - pub plan_type: Option, - pub resets_at: Option, -} - -// legacy helper removed; decoupled error handling in core - -#[derive(Debug, Deserialize)] -pub struct StreamEvent { - pub r#type: String, - pub response: Option, - pub item: Option, - pub error: Option, - #[serde(default)] - pub delta: Option, -} - -#[derive(Debug, Deserialize)] -pub struct TokenUsagePartial { - #[serde(default)] - pub input_tokens: i64, - #[serde(default)] - pub cached_input_tokens: i64, - #[serde(default)] - pub input_tokens_details: Option, - #[serde(default)] - pub output_tokens: i64, - #[serde(default)] - pub output_tokens_details: Option, - #[serde(default)] - pub reasoning_output_tokens: i64, - #[serde(default)] - pub total_tokens: i64, -} - -impl From for TokenUsage { - fn from(value: TokenUsagePartial) -> Self { - let cached_input_tokens = if value.cached_input_tokens > 0 { - Some(value.cached_input_tokens) - } else { - value - .input_tokens_details - .and_then(|d| d.cached_tokens) - .filter(|v| *v > 0) - }; - let reasoning_output_tokens = if value.reasoning_output_tokens > 0 { - Some(value.reasoning_output_tokens) - } else { - value - .output_tokens_details - .and_then(|d| d.reasoning_tokens) - .filter(|v| *v > 0) - }; - Self { - input_tokens: value.input_tokens, - cached_input_tokens: cached_input_tokens.unwrap_or(0), - output_tokens: value.output_tokens, - reasoning_output_tokens: reasoning_output_tokens.unwrap_or(0), - total_tokens: value.total_tokens, - } - } -} - -#[derive(Debug, Deserialize)] -pub struct TokenUsageInputDetails { - #[serde(default)] - pub cached_tokens: Option, -} - -#[derive(Debug, Deserialize)] -pub struct TokenUsageOutputDetails { - #[serde(default)] - pub reasoning_tokens: Option, -} - -pub async fn handle_sse_payload( - payload: sse::Payload, - tx_event: &mpsc::Sender>, - otel_event_manager: &OtelEventManager, -) -> Result<()> { - if let Some(responses) = payload.responses { - for ev in responses { - let event = match ev { - sse::Response::Completed(complete) => { - if let Some(usage) = &complete.usage { - otel_event_manager.sse_event_completed( - usage.input_tokens, - usage.output_tokens, - Some(usage.cached_input_tokens), - Some(usage.reasoning_output_tokens), - usage.total_tokens, - ); - } else { - otel_event_manager - .see_event_completed_failed(&"missing token usage".to_string()); - } - ResponseEvent::Completed { - response_id: complete.id, - token_usage: complete.usage, - } - } - sse::Response::Error(err) => { - let retry_after = err - .retry_after - .map(|secs| Duration::from_secs(if secs < 0 { 0 } else { secs as u64 })); - return Err(Error::Stream( - err.message.unwrap_or_else(|| "fatal error".to_string()), - retry_after, - )); - } - }; - tx_event.send(Ok(event)).await.ok(); - } - } - - if let Some(message_delta) = payload.response_message_delta { - let ev = ResponseEvent::OutputTextDelta(message_delta.text.clone()); - tx_event.send(Ok(ev)).await.ok(); - } - - if let Some(_response_content) = payload.response_content { - // Not used currently - } - - if let Some(ev) = payload.response_event { - debug!("Unhandled response_event: {ev:?}"); - } - - if let Some(item) = payload.response_output_item { - match item.r#type { - sse::OutputItem::Created => { - tx_event.send(Ok(ResponseEvent::Created)).await.ok(); - otel_event_manager.sse_event_kind("response.output_item.done"); - } - } - } - - if let Some(done) = payload.response_output_text_delta { - tx_event - .send(Ok(ResponseEvent::OutputTextDelta(done.text))) - .await - .ok(); - } - - if let Some(completed) = payload.response_output_item_done { - let response_item = - serde_json::from_value::(completed.item).map_err(Error::Json)?; - tx_event - .send(Ok(ResponseEvent::OutputItemDone(response_item))) - .await - .ok(); - otel_event_manager.sse_event_kind("response.output_item.done"); - } - - if let Some(reasoning_content_delta) = payload.response_output_reasoning_delta { - tx_event - .send(Ok(ResponseEvent::ReasoningContentDelta( - reasoning_content_delta.text, - ))) - .await - .ok(); - } - - if let Some(reasoning_summary_delta) = payload.response_output_reasoning_summary_delta { - tx_event - .send(Ok(ResponseEvent::ReasoningSummaryDelta( - reasoning_summary_delta.text, - ))) - .await - .ok(); - } - - if let Some(ev) = payload.response_error - && ev.code.as_deref() == Some("max_response_tokens") - { - let _ = tx_event - .send(Err(Error::Stream( - "context window exceeded".to_string(), - None, - ))) - .await; - } - - Ok(()) -} - -#[derive(Debug, Deserialize)] -pub struct TextDelta { - pub delta: String, -} - -pub async fn handle_stream_event( - event: StreamEvent, - tx_event: mpsc::Sender>, - otel_event_manager: &OtelEventManager, -) { - trace!("response event: {}", event.r#type); - match event.r#type.as_str() { - "response.created" => { - let _ = tx_event.send(Ok(ResponseEvent::Created)).await; - } - "response.output_text.delta" => { - if let Some(item_val) = event.item { - let resp = serde_json::from_value::(item_val); - if let Ok(delta) = resp { - let event = ResponseEvent::OutputTextDelta(delta.delta); - let _ = tx_event.send(Ok(event)).await; - } - } else if let Some(delta) = event.delta { - let _ = tx_event - .send(Ok(ResponseEvent::OutputTextDelta(delta))) - .await; - } - } - "response.reasoning_text.delta" => { - if let Some(delta) = event.delta { - let event = ResponseEvent::ReasoningContentDelta(delta); - let _ = tx_event.send(Ok(event)).await; - } - } - "response.reasoning_summary_text.delta" => { - if let Some(delta) = event.delta { - let event = ResponseEvent::ReasoningSummaryDelta(delta); - let _ = tx_event.send(Ok(event)).await; - } - } - "response.output_item.done" => { - if let Some(item_val) = event.item - && let Ok(item) = serde_json::from_value::(item_val) - { - let event = ResponseEvent::OutputItemDone(item); - if tx_event.send(Ok(event)).await.is_err() {} - } - } - "response.failed" => { - if let Some(resp_val) = event.response { - otel_event_manager.sse_event_failed( - Some(&"response.failed".to_string()), - Duration::from_millis(0), - &resp_val, - ); - - if let Some(err) = resp_val - .get("error") - .cloned() - .and_then(|v| serde_json::from_value::(v).ok()) - { - let msg = if err.code.as_deref() == Some("context_length_exceeded") { - "context window exceeded".to_string() - } else if err.code.as_deref() == Some("insufficient_quota") { - "quota exceeded".to_string() - } else { - err.message.unwrap_or_else(|| "fatal error".to_string()) - }; - let _ = tx_event.send(Err(Error::Stream(msg, None))).await; - } - } - } - "response.error" => { - if let Some(err_val) = event.error { - let err_resp = serde_json::from_value::(err_val); - if let Ok(err) = err_resp { - let retry_after = try_parse_retry_after(&err); - let _ = tx_event - .send(Err(Error::Stream( - err.error - .message - .unwrap_or_else(|| "unknown error".to_string()), - retry_after, - ))) - .await; - } - } - } - "response.completed" => { - if let Some(resp_val) = event.response - && let Ok(resp) = serde_json::from_value::(resp_val) - { - let usage = resp.usage.map(TokenUsage::from); - let ev = ResponseEvent::Completed { - response_id: resp.id, - token_usage: usage.clone(), - }; - let _ = tx_event.send(Ok(ev)).await; - if let Some(usage) = &usage { - otel_event_manager.sse_event_completed( - usage.input_tokens, - usage.output_tokens, - Some(usage.cached_input_tokens), - Some(usage.reasoning_output_tokens), - usage.total_tokens, - ); - } else { - otel_event_manager - .see_event_completed_failed(&"missing token usage".to_string()); - } - } - } - "response.output_item.added" => { - if let Some(item_val) = event.item - && let Ok(item) = serde_json::from_value::(item_val) - { - let event = ResponseEvent::OutputItemAdded(item); - if tx_event.send(Ok(event)).await.is_err() {} - } - } - "response.reasoning_summary_part.added" => { - let event = ResponseEvent::ReasoningSummaryPartAdded; - let _ = tx_event.send(Ok(event)).await; - } - _ => {} - } -} - -#[derive(Debug, Deserialize)] -pub struct ResponseErrorBody { - pub code: Option, -} - -fn try_parse_retry_after(err: &ErrorResponse) -> Option { - if err.error.r#type.as_deref() == Some("rate_limit_exceeded") { - let retry_after = serde_json::to_value(&err.error) - .ok() - .and_then(|v| v.get("retry_after").cloned()) - .and_then(|v| serde_json::from_value::(v).ok()) - .and_then(|v| v.code) - .and_then(parse_retry_after); - return retry_after; - } - None -} - -fn parse_retry_after(s: String) -> Option { - let minutes_pattern = regex_lite::Regex::new(r"^(\d+)m$").ok()?; - if let Some(cap) = minutes_pattern.captures(&s) - && let Some(m) = cap.get(1).and_then(|m| m.as_str().parse::().ok()) - { - return Some(Duration::from_secs(m * 60)); - } - s.parse::().ok().map(Duration::from_secs) -} - -pub mod sse { - use serde::Deserialize; - use serde_json::Value; - - #[derive(Debug, Deserialize)] - pub struct Payload { - pub responses: Option>, - pub response_content: Option, - pub response_error: Option, - pub response_event: Option, - pub response_message_delta: Option, - pub response_output_item: Option, - pub response_output_text_delta: Option, - pub response_output_item_done: Option, - pub response_output_reasoning_delta: Option, - pub response_output_reasoning_summary_delta: Option, - } - - #[derive(Debug, Deserialize)] - pub enum Response { - #[serde(rename = "response.completed")] - Completed(ResponseCompleted), - #[serde(rename = "response.error")] - Error(ResponseError), - } - - #[derive(Debug, Deserialize)] - pub struct ResponseCompleted { - pub id: String, - pub usage: Option, - } - - #[derive(Debug, Deserialize)] - pub struct ResponseError { - pub code: Option, - pub message: Option, - pub retry_after: Option, - } - - #[derive(Debug, Deserialize)] - pub struct ResponseMessageDelta { - pub text: String, - } - - #[derive(Debug, Deserialize)] - pub enum OutputItem { - #[serde(rename = "response.output_item.created")] - Created, - } - - #[derive(Debug, Deserialize)] - pub struct ResponseOutputItem { - pub r#type: OutputItem, - } - - #[derive(Debug, Deserialize)] - pub struct ResponseOutputTextDelta { - pub text: String, - } - - #[derive(Debug, Deserialize)] - pub struct ResponseOutputItemDone { - pub item: Value, - } - - #[derive(Debug, Deserialize)] - pub struct ResponseOutputReasoningDelta { - pub text: String, - } - - #[derive(Debug, Deserialize)] - pub struct ResponseOutputReasoningSummaryDelta { - pub text: String, - } -} - -pub struct ResponsesSseDecoder; - -impl Default for ResponsesSseDecoder { - fn default() -> Self { - Self - } -} - -#[async_trait] -impl crate::client::ResponseDecoder for ResponsesSseDecoder { - async fn on_frame( - &mut self, - json: &str, - tx: &mpsc::Sender>, - otel_event_manager: &OtelEventManager, - ) -> Result<()> { - if let Ok(event) = serde_json::from_str::(json) { - otel_event_manager.sse_event_kind(&event.r#type); - handle_stream_event(event, tx.clone(), otel_event_manager).await; - return Ok(()); - } - - otel_event_manager.sse_event_failed( - None, - Duration::from_millis(0), - &format!("Cannot parse SSE JSON: {json}"), - ); - - match serde_json::from_str::(json) { - Ok(payload) => handle_sse_payload(payload, tx, otel_event_manager).await, - Err(err) => { - otel_event_manager.sse_event_failed( - None, - Duration::from_millis(0), - &format!("Cannot parse SSE JSON: {err}"), - ); - Err(Error::Other(format!("Cannot parse SSE JSON: {err}"))) - } - } - } -} diff --git a/codex-rs/api-client/src/decode/chat.rs b/codex-rs/api-client/src/decode_wire/chat.rs similarity index 54% rename from codex-rs/api-client/src/decode/chat.rs rename to codex-rs/api-client/src/decode_wire/chat.rs index 43e83c1da9..eb9cc6471f 100644 --- a/codex-rs/api-client/src/decode/chat.rs +++ b/codex-rs/api-client/src/decode_wire/chat.rs @@ -1,48 +1,43 @@ use async_trait::async_trait; use codex_otel::otel_event_manager::OtelEventManager; -use codex_protocol::models::ContentItem; -use codex_protocol::models::ReasoningItemContent; -use codex_protocol::models::ResponseItem; +use serde_json::Value; use tokio::sync::mpsc; use tracing::debug; +use crate::client::WireResponseDecoder; use crate::error::Result; -use crate::stream::ResponseEvent; - -pub struct ChatSseDecoder { - fn_call_state: FunctionCallState, - assistant_item: Option, - reasoning_item: Option, -} +use crate::stream::WireEvent; #[derive(Default)] struct FunctionCallState { + active: bool, + call_id: Option, name: Option, arguments: String, - call_id: Option, - active: bool, } -impl ChatSseDecoder { +pub struct WireChatSseDecoder { + fn_call_state: FunctionCallState, +} + +impl WireChatSseDecoder { pub fn new() -> Self { Self { fn_call_state: FunctionCallState::default(), - assistant_item: None, - reasoning_item: None, } } } #[async_trait] -impl crate::client::ResponseDecoder for ChatSseDecoder { +impl WireResponseDecoder for WireChatSseDecoder { async fn on_frame( &mut self, json: &str, - tx: &mpsc::Sender>, + tx: &mpsc::Sender>, _otel: &OtelEventManager, ) -> Result<()> { - // Chat sends a terminal "[DONE]" frame; we ignore it here. Caller should handle end-of-stream. - let Ok(parsed_chunk) = serde_json::from_str::(json) else { + // Chat sends a terminal "[DONE]" frame; ignore it. + let Ok(parsed_chunk) = serde_json::from_str::(json) else { debug!("failed to parse Chat SSE JSON: {}", json); return Ok(()); }; @@ -58,10 +53,8 @@ impl crate::client::ResponseDecoder for ChatSseDecoder { if let Some(content) = delta.get("content").and_then(|c| c.as_array()) { for piece in content { if let Some(text) = piece.get("text").and_then(|t| t.as_str()) { - append_assistant_text(tx, &mut self.assistant_item, text.to_string()) - .await; let _ = tx - .send(Ok(ResponseEvent::OutputTextDelta(text.to_string()))) + .send(Ok(WireEvent::OutputTextDelta(text.to_string()))) .await; } } @@ -87,7 +80,8 @@ impl crate::client::ResponseDecoder for ChatSseDecoder { if let Some(reasoning) = delta.get("reasoning_content").and_then(|c| c.as_array()) { for entry in reasoning { if let Some(text) = entry.get("text").and_then(|t| t.as_str()) { - append_reasoning_text(tx, &mut self.reasoning_item, text.to_string()) + let _ = tx + .send(Ok(WireEvent::ReasoningContentDelta(text.to_string()))) .await; } } @@ -103,65 +97,17 @@ impl crate::client::ResponseDecoder for ChatSseDecoder { let arguments = self.fn_call_state.arguments.clone(); self.fn_call_state = FunctionCallState::default(); - let item = ResponseItem::FunctionCall { - id: Some(call_id.clone()), - call_id, - name: function_name, - arguments, - }; - let _ = tx.send(Ok(ResponseEvent::OutputItemDone(item))).await; + let item = serde_json::json!({ + "type": "function_call", + "id": call_id, + "call_id": call_id, + "name": function_name, + "arguments": arguments, + }); + let _ = tx.send(Ok(WireEvent::OutputItemDone(item))).await; } } Ok(()) } } - -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 }); - } -} - -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 }); - } -} diff --git a/codex-rs/api-client/src/decode_wire/mod.rs b/codex-rs/api-client/src/decode_wire/mod.rs new file mode 100644 index 0000000000..74e360ea71 --- /dev/null +++ b/codex-rs/api-client/src/decode_wire/mod.rs @@ -0,0 +1,2 @@ +pub mod chat; +pub mod responses; diff --git a/codex-rs/api-client/src/decode_wire/responses.rs b/codex-rs/api-client/src/decode_wire/responses.rs new file mode 100644 index 0000000000..98af5a86f1 --- /dev/null +++ b/codex-rs/api-client/src/decode_wire/responses.rs @@ -0,0 +1,166 @@ +use async_trait::async_trait; +use codex_otel::otel_event_manager::OtelEventManager; +use serde::Deserialize; +use serde_json::Value; +use tokio::sync::mpsc; +use tracing::debug; + +use crate::client::WireResponseDecoder; +use crate::error::Error; +use crate::error::Result; +use crate::stream::WireEvent; +use crate::stream::WireTokenUsage; + +#[derive(Debug, Deserialize)] +struct StreamEvent { + #[serde(rename = "type")] + event_type: String, + #[serde(default)] + response: Option, + #[serde(default)] + item: Option, + #[serde(default)] + error: Option, + #[serde(default)] + delta: Option, +} + +pub struct WireResponsesSseDecoder; + +#[async_trait] +impl WireResponseDecoder for WireResponsesSseDecoder { + async fn on_frame( + &mut self, + json: &str, + tx: &mpsc::Sender>, + otel: &OtelEventManager, + ) -> Result<()> { + let Ok(event) = serde_json::from_str::(json) else { + debug!("failed to parse Responses SSE JSON: {}", json); + return Ok(()); + }; + + match event.event_type.as_str() { + "response.created" => { + let _ = tx.send(Ok(WireEvent::Created)).await; + } + "response.output_text.delta" => { + if let Some(delta) = event.delta.or_else(|| { + event.item.and_then(|v| { + v.get("delta") + .and_then(|d| d.as_str().map(|s| s.to_string())) + }) + }) { + let _ = tx.send(Ok(WireEvent::OutputTextDelta(delta))).await; + } + } + "response.reasoning_text.delta" => { + if let Some(delta) = event.delta { + let _ = tx.send(Ok(WireEvent::ReasoningContentDelta(delta))).await; + } + } + "response.reasoning_summary_text.delta" => { + if let Some(delta) = event.delta { + let _ = tx.send(Ok(WireEvent::ReasoningSummaryDelta(delta))).await; + } + } + "response.output_item.done" => { + if let Some(item_val) = event.item { + let _ = tx.send(Ok(WireEvent::OutputItemDone(item_val))).await; + } + } + "response.output_item.added" => { + if let Some(item_val) = event.item { + let _ = tx.send(Ok(WireEvent::OutputItemAdded(item_val))).await; + } + } + "response.reasoning_summary_part.added" => { + let _ = tx.send(Ok(WireEvent::ReasoningSummaryPartAdded)).await; + } + "response.completed" => { + if let Some(resp) = event.response { + let response_id = resp + .get("id") + .and_then(|v| v.as_str()) + .unwrap_or_default() + .to_string(); + let usage = parse_wire_usage(&resp); + if let Some(u) = &usage { + otel.sse_event_completed( + u.input_tokens, + u.output_tokens, + Some(u.cached_input_tokens), + Some(u.reasoning_output_tokens), + u.total_tokens, + ); + } else { + otel.see_event_completed_failed(&"missing token usage".to_string()); + } + let _ = tx + .send(Ok(WireEvent::Completed { + response_id, + token_usage: usage, + })) + .await; + } + } + "response.error" | "response.failed" => { + let message = event + .error + .as_ref() + .and_then(|v| v.get("message")) + .and_then(|v| v.as_str()) + .map(|s| s.to_string()) + .unwrap_or_else(|| "unknown error".to_string()); + let _ = tx.send(Err(Error::Stream(message, None))).await; + } + _ => {} + } + + Ok(()) + } +} + +fn parse_wire_usage(resp: &Value) -> Option { + let usage = resp.get("usage").cloned()?; + let input_tokens = usage + .get("input_tokens") + .and_then(|v| v.as_i64()) + .unwrap_or(0); + let cached_input_tokens = usage + .get("cached_input_tokens") + .and_then(|v| v.as_i64()) + .or_else(|| { + usage + .get("input_tokens_details") + .and_then(|d| d.get("cached_tokens")) + .and_then(|v| v.as_i64()) + }) + .unwrap_or(0); + let output_tokens = usage + .get("output_tokens") + .and_then(|v| v.as_i64()) + .unwrap_or(0); + let reasoning_output_tokens = usage + .get("reasoning_output_tokens") + .and_then(|v| v.as_i64()) + .or_else(|| { + usage + .get("output_tokens_details") + .and_then(|d| d.get("reasoning_tokens")) + .and_then(|v| v.as_i64()) + }) + .unwrap_or(0); + let total_tokens = usage + .get("total_tokens") + .and_then(|v| v.as_i64()) + .unwrap_or(0); + + Some(WireTokenUsage { + input_tokens, + cached_input_tokens, + output_tokens, + reasoning_output_tokens, + total_tokens, + }) +} diff --git a/codex-rs/api-client/src/lib.rs b/codex-rs/api-client/src/lib.rs index 63550d6b7a..ba38736490 100644 --- a/codex-rs/api-client/src/lib.rs +++ b/codex-rs/api-client/src/lib.rs @@ -1,21 +1,18 @@ -pub mod api; pub mod auth; pub mod chat; mod client; -mod common; -mod decode; +// Legacy payload decoding has been removed; wire decoding lives in decode_wire +mod decode_wire; pub mod error; // payload building lives in codex-core now pub mod responses; pub mod routed_client; pub mod stream; -mod wire; pub use crate::auth::AuthContext; pub use crate::auth::AuthProvider; pub use crate::chat::ChatCompletionsApiClient; pub use crate::chat::ChatCompletionsApiClientConfig; -pub use crate::client::fixtures::stream_from_fixture; pub use crate::error::Error; pub use crate::error::Result; pub use crate::responses::ResponsesApiClient; diff --git a/codex-rs/api-client/src/responses.rs b/codex-rs/api-client/src/responses.rs index 4e07a2d99d..84b2354268 100644 --- a/codex-rs/api-client/src/responses.rs +++ b/codex-rs/api-client/src/responses.rs @@ -1,6 +1,5 @@ use std::sync::Arc; -use async_trait::async_trait; use codex_app_server_protocol::AuthMode; use codex_otel::otel_event_manager::OtelEventManager; use codex_protocol::ConversationId; @@ -10,12 +9,9 @@ use tokio::sync::mpsc; use tracing::debug; use tracing::trace; -use crate::api::PayloadClient; use crate::auth::AuthProvider; use crate::error::Error; use crate::error::Result; -use crate::stream::ResponseEvent; -use crate::stream::ResponseStream; use crate::stream::WireResponseStream; use codex_provider_config::ModelProviderInfo; @@ -43,19 +39,18 @@ pub struct ResponsesApiClient { config: ResponsesApiClientConfig, } -#[async_trait] -impl PayloadClient for ResponsesApiClient { - type Config = ResponsesApiClientConfig; - - fn new(config: Self::Config) -> Result { +impl ResponsesApiClient { + pub fn new(config: ResponsesApiClientConfig) -> Result { Ok(Self { config }) } +} - async fn stream_payload( +impl ResponsesApiClient { + pub async fn stream_payload_wire( &self, payload_json: &Value, - session_source: Option<&codex_protocol::protocol::SessionSource>, - ) -> Result { + _session_source: Option<&codex_protocol::protocol::SessionSource>, + ) -> Result { if self.config.provider.wire_api != codex_provider_config::WireApi::Responses { return Err(Error::UnsupportedOperation( "ResponsesApiClient requires a Responses provider".to_string(), @@ -98,7 +93,6 @@ impl PayloadClient for ResponsesApiClient { &self.config.http_client, &self.config.provider, &auth, - session_source, &extra_headers, ) .await?; @@ -124,10 +118,27 @@ impl PayloadClient for ResponsesApiClient { request_id: None, })?; - let (tx_event, rx_event) = mpsc::channel::>(1600); + let (tx_event, rx_event) = mpsc::channel::>(1600); if let Some(snapshot) = crate::client::rate_limits::parse_rate_limit_snapshot(res.headers()) && tx_event - .send(Ok(ResponseEvent::RateLimits(snapshot))) + .send(Ok(crate::stream::WireEvent::RateLimits( + crate::stream::WireRateLimitSnapshot { + primary: snapshot + .primary + .map(|w| crate::stream::WireRateLimitWindow { + used_percent: Some(w.used_percent), + window_minutes: w.window_minutes, + resets_at: w.resets_at, + }), + secondary: snapshot + .secondary + .map(|w| crate::stream::WireRateLimitWindow { + used_percent: Some(w.used_percent), + window_minutes: w.window_minutes, + resets_at: w.resets_at, + }), + }, + ))) .await .is_err() { @@ -142,36 +153,14 @@ impl PayloadClient for ResponsesApiClient { }); let idle_timeout = self.config.provider.stream_idle_timeout(); let otel = self.config.otel_event_manager.clone(); - tokio::spawn(crate::client::sse::process_sse( + tokio::spawn(crate::client::sse::process_sse_wire( stream, tx_event, idle_timeout, otel, - crate::decode::responses::ResponsesSseDecoder, + crate::decode_wire::responses::WireResponsesSseDecoder, )); Ok(crate::stream::EventStream::from_receiver(rx_event)) } } - -impl ResponsesApiClient { - pub async fn stream_payload_wire( - &self, - payload_json: &Value, - session_source: Option<&codex_protocol::protocol::SessionSource>, - ) -> Result { - use futures::StreamExt; - let legacy = self.stream_payload(payload_json, session_source).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| crate::wire::map_response_event_to_wire(ev)); - if tx.send(converted).await.is_err() { - break; - } - } - }); - Ok(crate::stream::EventStream::from_receiver(rx)) - } -} diff --git a/codex-rs/api-client/src/routed_client.rs b/codex-rs/api-client/src/routed_client.rs index e772961d35..c96cec7d08 100644 --- a/codex-rs/api-client/src/routed_client.rs +++ b/codex-rs/api-client/src/routed_client.rs @@ -3,20 +3,15 @@ use std::sync::Arc; use codex_otel::otel_event_manager::OtelEventManager; use codex_protocol::ConversationId; -use codex_protocol::protocol::SessionSource; use crate::ChatCompletionsApiClient; use crate::ChatCompletionsApiClientConfig; -use crate::ResponseStream; 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 codex_provider_config::ModelProviderInfo; /// Dispatches to the appropriate API client implementation based on the provider wire API. @@ -28,8 +23,8 @@ pub struct RoutedApiClientConfig { pub conversation_id: ConversationId, pub auth_provider: Option>, pub otel_event_manager: OtelEventManager, - pub session_source: SessionSource, pub responses_fixture_path: Option, + pub extra_headers: Vec<(String, String)>, } #[derive(Clone)] @@ -42,18 +37,12 @@ impl RoutedApiClient { Self { config } } - pub async fn stream_payload(&self, payload_json: &serde_json::Value) -> Result { + pub async fn stream_payload_wire( + &self, + payload_json: &serde_json::Value, + ) -> Result { match self.config.provider.wire_api { WireApi::Responses => { - if let Some(path) = &self.config.responses_fixture_path { - return stream_from_fixture( - path, - self.config.provider.clone(), - self.config.otel_event_manager.clone(), - ) - .await; - } - let cfg = ResponsesApiClientConfig { http_client: self.config.http_client.clone(), provider: self.config.provider.clone(), @@ -61,12 +50,18 @@ impl RoutedApiClient { conversation_id: self.config.conversation_id, auth_provider: self.config.auth_provider.clone(), otel_event_manager: self.config.otel_event_manager.clone(), - extra_headers: vec![], + extra_headers: self.config.extra_headers.clone(), }; - let client = ::new(cfg)?; - client - .stream_payload(payload_json, Some(&self.config.session_source)) - .await + if let Some(path) = &self.config.responses_fixture_path { + return crate::client::fixtures::stream_from_fixture_wire( + path, + self.config.provider.clone(), + self.config.otel_event_manager.clone(), + ) + .await; + } + let client = ResponsesApiClient::new(cfg)?; + client.stream_payload_wire(payload_json, None).await } WireApi::Chat => { let cfg = ChatCompletionsApiClientConfig { @@ -74,54 +69,11 @@ impl RoutedApiClient { provider: self.config.provider.clone(), model: self.config.model.clone(), otel_event_manager: self.config.otel_event_manager.clone(), - session_source: self.config.session_source.clone(), - extra_headers: vec![], + extra_headers: self.config.extra_headers.clone(), }; - let client = ::new(cfg)?; - client - .stream_payload(payload_json, Some(&self.config.session_source)) - .await + let client = ChatCompletionsApiClient::new(cfg)?; + client.stream_payload_wire(payload_json, None).await } } } - - 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] -impl PayloadClient for RoutedApiClient { - type Config = RoutedApiClientConfig; - - fn new(config: Self::Config) -> Result { - Ok(Self::new(config)) - } - - async fn stream_payload( - &self, - payload_json: &serde_json::Value, - _session_source: Option<&codex_protocol::protocol::SessionSource>, - ) -> Result { - self.stream_payload(payload_json).await - } -} - -fn map_response_event_to_wire(ev: crate::stream::ResponseEvent) -> Result { - crate::wire::map_response_event_to_wire(ev) } diff --git a/codex-rs/api-client/src/wire.rs b/codex-rs/api-client/src/wire.rs deleted file mode 100644 index c526043af6..0000000000 --- a/codex-rs/api-client/src/wire.rs +++ /dev/null @@ -1,55 +0,0 @@ -use crate::error::Result; -use crate::stream::WireEvent; -use crate::stream::WireRateLimitSnapshot; -use crate::stream::WireRateLimitWindow; - -pub 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| crate::stream::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| WireRateLimitWindow { - used_percent: Some(w.used_percent), - window_minutes: w.window_minutes, - resets_at: w.resets_at, - }) - }; - WireEvent::RateLimits(WireRateLimitSnapshot { - primary: to_win(s.primary), - secondary: to_win(s.secondary), - }) - } - }) -} diff --git a/codex-rs/core/src/client.rs b/codex-rs/core/src/client.rs index 6c277bc7c8..4418a3be8e 100644 --- a/codex-rs/core/src/client.rs +++ b/codex-rs/core/src/client.rs @@ -177,6 +177,18 @@ impl ModelClient { let responses_fixture_path: Option = CODEX_RS_SSE_FIXTURE.as_ref().map(PathBuf::from); let http_client = create_client().clone_inner(); + // Compose extra headers (conversation/session + subagent) + let mut extra_headers: Vec<(String, String)> = vec![ + ( + "conversation_id".to_string(), + self.conversation_id.to_string(), + ), + ("session_id".to_string(), self.conversation_id.to_string()), + ]; + if let Some((name, value)) = build_subagent_header(&self.session_source) { + extra_headers.push((name, value)); + } + let config = RoutedApiClientConfig { http_client, provider: self.provider.clone(), @@ -184,8 +196,8 @@ impl ModelClient { conversation_id: self.conversation_id, auth_provider, otel_event_manager: self.otel_event_manager.clone(), - session_source: self.session_source.clone(), responses_fixture_path, + extra_headers, }; Ok(RoutedApiClient::new(config)) @@ -224,6 +236,22 @@ impl ModelClient { } } +fn build_subagent_header(session_source: &SessionSource) -> Option<(String, String)> { + use codex_protocol::protocol::SubAgentSource; + if let SessionSource::SubAgent(sub) = session_source { + let value = match sub { + SubAgentSource::Other(label) => label.clone(), + _ => serde_json::to_value(sub) + .ok() + .and_then(|v| v.as_str().map(std::string::ToString::to_string)) + .unwrap_or_else(|| "other".to_string()), + }; + Some(("x-openai-subagent".to_string(), value)) + } else { + None + } +} + struct AuthManagerProvider { manager: Arc, }