V3

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]]

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

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<Self, Error>
+    where
+        Self: Sized;
+
+    async fn stream(&self, prompt: Prompt) -> Result<ResponseStream, Error>;
+}

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<String>,
+    pub account_id: Option<String>,
+}
+
+#[async_trait]
+pub trait AuthProvider: Send + Sync {
+    async fn auth_context(&self) -> Option<AuthContext>;
+    async fn refresh_token(&self) -> Result<Option<String>, String>;
+}

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<T> = std::result::Result<T, Error>;
+
+#[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<Self> {
+        Ok(Self { config })
+    }
+
+    async fn stream(&self, prompt: Prompt) -> Result<ResponseStream> {
+        Self::validate_prompt(&prompt)?;
+
+        let payload = self.build_payload(&prompt)?;
+        let (tx_event, rx_event) = mpsc::channel::<Result<ResponseEvent>>(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(|_| "<failed to read response>".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::<u64>().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<serde_json::Value> {
+        let mut messages = Vec::<serde_json::Value>::new();
+        messages.push(json!({ "role": "system", "content": prompt.instructions }));
+
+        let mut reasoning_by_anchor_index: std::collections::HashMap<usize, String> =
+            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<usize> = 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<String> = None;
+
+        for (idx, item) in prompt.input.iter().enumerate() {
+            match item {
+                ResponseItem::Message { role, content, .. } => {
+                    let mut text = String::new();
+                    let mut items: Vec<serde_json::Value> = 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<serde_json::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}})
+                                }
+                            })
+                            .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<Result<ResponseEvent>>,
+    assistant_item: &mut Option<ResponseItem>,
+    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<Result<ResponseEvent>>,
+    reasoning_item: &mut Option<ResponseItem>,
+    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<S>(
+    stream: S,
+    tx_event: mpsc::Sender<Result<ResponseEvent>>,
+    idle_timeout: Duration,
+    otel_event_manager: OtelEventManager,
+    _aggregation_mode: ChatAggregationMode,
+) where
+    S: Stream<Item = Result<Bytes>> + Unpin,
+{
+    let mut stream = stream.eventsource();
+
+    #[derive(Default)]
+    struct FunctionCallState {
+        name: Option<String>,
+        arguments: String,
+        call_id: Option<String>,
+    }
+
+    let mut function_call_state = FunctionCallState::default();
+    let mut assistant_item: Option<ResponseItem> = None;
+    let mut reasoning_item: Option<ResponseItem> = 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<Item = Result<ResponseEvent>> + Sized {
+    fn aggregate(self) -> AggregatedChatStream<Self>
+    where
+        Self: Unpin,
+    {
+        AggregatedChatStream::new(self, AggregateMode::AggregatedOnly)
+    }
+
+    fn streaming_mode(self) -> AggregatedChatStream<Self>
+    where
+        Self: Unpin,
+    {
+        AggregatedChatStream::new(self, AggregateMode::Streaming)
+    }
+}
+
+impl<S> AggregateStreamExt for S where S: Stream<Item = Result<ResponseEvent>> + Sized + Unpin {}
+
+enum AggregateMode {
+    AggregatedOnly,
+    Streaming,
+}
+
+pub struct AggregatedChatStream<S> {
+    inner: S,
+    cumulative: String,
+    cumulative_reasoning: String,
+    pending: VecDeque<ResponseEvent>,
+    mode: AggregateMode,
+}
+
+impl<S> AggregatedChatStream<S>
+where
+    S: Stream<Item = Result<ResponseEvent>> + Unpin,
+{
+    fn new(inner: S, mode: AggregateMode) -> Self {
+        Self {
+            inner,
+            cumulative: String::new(),
+            cumulative_reasoning: String::new(),
+            pending: VecDeque::new(),
+            mode,
+        }
+    }
+}
+
+impl<S> Stream for AggregatedChatStream<S>
+where
+    S: Stream<Item = Result<ResponseEvent>> + Unpin,
+{
+    type Item = Result<ResponseEvent>;
+
+    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
+        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<Vec<serde_json::Value>> {
+    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::<Vec<serde_json::Value>>();
+    Ok(tools_json)
+}
+
+fn backoff(attempt: u64) -> Duration {
+    let capped = attempt.min(6);
+    Duration::from_millis(100 * 2u64.pow(capped as u32))
+}

codex-rs/api-client/src/error.rs

@@ -0,0 +1,42 @@
+use std::time::Duration;
+
+use thiserror::Error;
+
+pub type Result<T> = std::result::Result<T, Error>;
+
+#[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<String>,
+    },
+    #[error("{0}")]
+    Stream(String, Option<Duration>),
+    #[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<String>,
+    },
+    #[error("missing environment variable {var}")]
+    MissingEnvVar {
+        var: String,
+        instructions: Option<String>,
+    },
+    #[error("{0}")]
+    Auth(String),
+    #[error(transparent)]
+    Json(#[from] serde_json::Error),
+    #[error("{0}")]
+    Other(String),
+}

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;

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<String>,
-    /// Environment variable that stores the user's API key for this provider.
     pub env_key: Option<String>,
-
-    /// Optional instructions to help the user get a valid value for the
-    /// variable and set it.
     pub env_key_instructions: Option<String>,
-
-    /// Value to use with `Authorization: Bearer <token>` 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<String>,
-
-    /// 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<HashMap<String, String>>,
-
-    /// 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<HashMap<String, String>>,
-
-    /// 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<HashMap<String, String>>,
-
-    /// Maximum number of times to retry a failed HTTP request to this provider.
     pub request_max_retries: Option<u64>,
-
-    /// Number of times to retry reconnecting a dropped streaming response before failing.
     pub stream_max_retries: Option<u64>,
-
-    /// Idle timeout (in milliseconds) to wait for activity on a streaming response before treating
-    /// the connection as lost.
     pub stream_idle_timeout_ms: Option<u64>,
-
-    /// 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<CodexAuth>,
-    ) -> crate::error::Result<CodexRequestBuilder> {
+    pub async fn create_request_builder(
+        &self,
+        client: &reqwest::Client,
+        auth: &Option<AuthContext>,
+    ) -> Result<reqwest::RequestBuilder> {
         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<CodexAuth>) -> 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<Option<String>> {
+    pub fn api_key(&self) -> Result<Option<String>> {
         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<String, ModelProviderInfo> {
     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<String, ModelProviderInfo> {
                     .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<String, ModelProviderInfo> {
 }
 
 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"
             );
         }
     }

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<ResponseItem>,
-    #[serde(skip_serializing)]
     pub tools: Vec<Value>,
-    #[serde(skip_serializing)]
     pub parallel_tool_calls: bool,
-    #[serde(skip_serializing)]
     pub output_schema: Option<Value>,
+    pub reasoning: Option<Reasoning>,
+    pub text_controls: Option<TextControls>,
+    pub store_response: bool,
+    pub prompt_cache_key: Option<String>,
+    pub previous_response_id: Option<String>,
+    pub session_source: Option<SessionSource>,
 }
 
 impl Prompt {
+    #[allow(clippy::too_many_arguments)]
     pub fn new(
+        instructions: String,
         input: Vec<ResponseItem>,
         tools: Vec<Value>,
         parallel_tool_calls: bool,
         output_schema: Option<Value>,
+        reasoning: Option<Reasoning>,
+        text_controls: Option<TextControls>,
+        store_response: bool,
+        prompt_cache_key: Option<String>,
+        previous_response_id: Option<String>,
+        session_source: Option<SessionSource>,
     ) -> Self {
         Self {
+            instructions,
             input,
             tools,
             parallel_tool_calls,
             output_schema,
+            reasoning,
+            text_controls,
+            store_response,
+            prompt_cache_key,
+            previous_response_id,
+            session_source,
         }
     }
 }

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<T> = std::result::Result<T, Error>;
+
+#[derive(Clone)]
+pub struct ResponsesApiClientConfig {
+    pub http_client: reqwest::Client,
+    pub provider: ModelProviderInfo,
+    pub model: String,
+    pub conversation_id: ConversationId,
+    pub auth_provider: Option<Arc<dyn AuthProvider>>,
+    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<Self> {
+        Ok(Self { config })
+    }
+
+    async fn stream(&self, prompt: Prompt) -> Result<ResponseStream> {
+        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<Value> {
+        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<ResponseStream, StreamAttemptError> {
+        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(|_| "<unable to serialize payload>".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::<Result<ResponseEvent>>(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::<u64>().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(|_| "<failed to read response>".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<Duration>,
+        request_id: Option<String>,
+    },
+    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<Value>,
+    item: Option<Value>,
+    delta: Option<String>,
+}
+
+#[derive(Debug, Deserialize)]
+struct ResponseCompleted {
+    id: String,
+    usage: Option<ResponseCompletedUsage>,
+}
+
+#[derive(Debug, Deserialize)]
+struct ResponseCompletedUsage {
+    input_tokens: i64,
+    input_tokens_details: Option<ResponseCompletedInputTokensDetails>,
+    output_tokens: i64,
+    output_tokens_details: Option<ResponseCompletedOutputTokensDetails>,
+    total_tokens: i64,
+}
+
+impl From<ResponseCompletedUsage> 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<RateLimitSnapshot> {
+    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<RateLimitWindow> {
+    let used_percent: Option<f64> = 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<f64> {
+    parse_header_str(headers, name)?
+        .parse::<f64>()
+        .ok()
+        .filter(|v| v.is_finite())
+}
+
+fn parse_header_i64(headers: &HeaderMap, name: &str) -> Option<i64> {
+    parse_header_str(headers, name)?.parse::<i64>().ok()
+}
+
+fn parse_header_str<'a>(headers: &'a HeaderMap, name: &str) -> Option<&'a str> {
+    headers.get(name)?.to_str().ok()
+}
+
+async fn process_sse<S>(
+    stream: S,
+    tx_event: mpsc::Sender<Result<ResponseEvent>>,
+    idle_timeout: Duration,
+    otel_event_manager: OtelEventManager,
+) where
+    S: Stream<Item = Result<Bytes>> + Unpin + Send + 'static,
+{
+    let mut stream = stream.eventsource();
+
+    let mut response_completed: Option<ResponseCompleted> = None;
+    let mut response_error: Option<Error> = 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::<ResponseItem>(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::<ErrorResponse>(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::<ResponseCompleted>(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::<ResponseItem>(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<String>,
+    message: Option<String>,
+}
+
+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<Regex> = 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<Duration> {
+    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::<f64>().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<Path>,
+    provider: ModelProviderInfo,
+    otel_event_manager: OtelEventManager,
+) -> Result<ResponseStream> {
+    let (tx_event, rx_event) = mpsc::channel::<Result<ResponseEvent>>(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 })
+}

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<ReasoningEffortConfig>,
@@ -33,3 +44,31 @@ pub struct TextControls {
     #[serde(skip_serializing_if = "Option::is_none")]
     pub format: Option<TextFormat>,
 }
+
+#[derive(Debug)]
+pub enum ResponseEvent {
+    Created,
+    OutputItemDone(ResponseItem),
+    OutputItemAdded(ResponseItem),
+    Completed {
+        response_id: String,
+        token_usage: Option<TokenUsage>,
+    },
+    OutputTextDelta(String),
+    ReasoningSummaryDelta(String),
+    ReasoningContentDelta(String),
+    ReasoningSummaryPartAdded,
+    RateLimits(RateLimitSnapshot),
+}
+
+pub struct ResponseStream {
+    pub(crate) rx_event: mpsc::Receiver<Result<ResponseEvent>>,
+}
+
+impl Stream for ResponseStream {
+    type Item = Result<ResponseEvent>;
+
+    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
+        self.rx_event.poll_recv(cx)
+    }
+}

codex-rs/core/src/chat_completions.rs

@@ -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<ResponseStream> {
-    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::<serde_json::Value>::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<usize, String> =
-        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<usize> = 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<String> = 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<serde_json::Value> = 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<serde_json::Value> = 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::<Result<ResponseEvent>>(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::<u64>().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<Result<ResponseEvent>>,
-    assistant_item: &mut Option<ResponseItem>,
-    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<Result<ResponseEvent>>,
-    reasoning_item: &mut Option<ResponseItem>,
-    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<S>(
-    stream: S,
-    tx_event: mpsc::Sender<Result<ResponseEvent>>,
-    idle_timeout: Duration,
-    otel_event_manager: OtelEventManager,
-) where
-    S: Stream<Item = Result<Bytes>> + 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<String>,
-        arguments: String,
-        call_id: Option<String>,
-        active: bool,
-    }
-
-    let mut fn_call_state = FunctionCallState::default();
-    let mut assistant_item: Option<ResponseItem> = None;
-    let mut reasoning_item: Option<ResponseItem> = 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<S> {
-    inner: S,
-    cumulative: String,
-    cumulative_reasoning: String,
-    pending: std::collections::VecDeque<ResponseEvent>,
-    mode: AggregateMode,
-}
-
-impl<S> Stream for AggregatedChatStream<S>
-where
-    S: Stream<Item = Result<ResponseEvent>> + Unpin,
-{
-    type Item = Result<ResponseEvent>;
-
-    fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
-        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<Item = Result<ResponseEvent>> + 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(<full message>)
-    ///     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<Self> {
-        AggregatedChatStream::new(self, AggregateMode::AggregatedOnly)
-    }
-}
-
-impl<T> AggregateStreamExt for T where T: Stream<Item = Result<ResponseEvent>> + Sized {}
-
-impl<S> AggregatedChatStream<S> {
-    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)
-    }
-}

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<TokenUsage>,
-    },
-    OutputTextDelta(String),
-    ReasoningSummaryDelta(String),
-    ReasoningContentDelta(String),
-    ReasoningSummaryPartAdded,
-    RateLimits(RateLimitSnapshot),
-}
+pub use codex_api_client::ResponseEvent;
 
 pub struct ResponseStream {
     pub(crate) rx_event: mpsc::Receiver<Result<ResponseEvent>>,

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<ResponseItem>,
     cancellation_token: CancellationToken,
 ) -> CodexResult<TurnRunResult> {
-    let chaining_intent = payload.store_response;
+    let chaining_intent = payload.prompt.store_response;
 
     let rollout_item = RolloutItem::TurnContext(TurnContextItem {
         cwd: turn_context.cwd.clone(),

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 {

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;

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<U>(&self, url: U) -> CodexRequestBuilder
     where
         U: IntoUrl,

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;

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());

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<Vec<serde_json::Value>> {
@@ -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<Vec<serde_json::Value>> {
-    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::<Vec<serde_json::Value>>();
-    Ok(tools_json)
-}
 
 pub(crate) fn mcp_tool_to_openai_tool(
     fully_qualified_name: String,