V1

codex-rs/core/src/client/chat_completions.rs

@@ -1,15 +1,14 @@
 use std::time::Duration;
 
 use crate::ModelProviderInfo;
+use crate::client::ResponseEvent;
+use crate::client::ResponseStream;
+use crate::client::http::CodexHttpClient;
 use crate::client_common::Prompt;
-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;
@@ -55,7 +54,7 @@ pub(crate) async fn stream_chat_completions(
     let mut messages = Vec::<serde_json::Value>::new();
 
     let full_instructions = prompt.get_full_instructions(model_family);
-    messages.push(json!({"role": "system", "content": full_instructions}));
+    messages.push(json!({ "role": "system", "content": full_instructions }));
 
     let input = prompt.get_formatted_input();
 
@@ -128,99 +127,73 @@ pub(crate) async fn stream_chat_completions(
                 {
                     reasoning_by_anchor_index
                         .entry(idx - 1)
-                        .and_modify(|v| v.push_str(&text))
+                        .and_modify(|existing| existing.push_str(text.as_str()))
                         .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());
+                // Otherwise, attach to the first future assistant anchor.
+                if !attached {
+                    for anchor_idx in idx + 1..input.len() {
+                        match &input[anchor_idx] {
+                            ResponseItem::Message { role, .. } if role == "assistant" => {
+                                reasoning_by_anchor_index
+                                    .entry(anchor_idx)
+                                    .and_modify(|existing| existing.push_str(text.as_str()))
+                                    .or_insert(text.clone());
+                                attached = true;
+                                break;
+                            }
+                            ResponseItem::FunctionCall { .. }
+                            | ResponseItem::LocalShellCall { .. }
+                            | ResponseItem::FunctionCallOutput { .. } => {
+                                continue;
+                            }
+                            _ => break,
                         }
-                        ResponseItem::Message { role, .. } if role == "assistant" => {
-                            reasoning_by_anchor_index
-                                .entry(idx + 1)
-                                .and_modify(|v| v.push_str(&text))
-                                .or_insert(text.clone());
-                        }
-                        _ => {}
                     }
                 }
+
+                // Either attached or dropped, move on.
             }
         }
     }
 
-    // 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() {
+    for (index, 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}}));
+                let mut content_text = String::new();
+                for item in content {
+                    match item {
+                        ContentItem::InputText { text } | ContentItem::OutputText { text } => {
+                            content_text.push_str(text)
                         }
+                        ContentItem::InputImage { .. } => {}
                     }
                 }
-
-                // Skip exact-duplicate assistant messages.
-                if role == "assistant" {
-                    if let Some(prev) = &last_assistant_text
-                        && prev == &text
-                    {
-                        continue;
-                    }
-                    last_assistant_text = Some(text.clone());
+                if content_text.trim().is_empty() {
+                    continue;
                 }
 
-                // 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));
+                // Append reasoning when mapped to this anchor.
+                if let Some(reasoning) = reasoning_by_anchor_index.remove(&index) {
+                    content_text.push_str(reasoning.as_str());
                 }
-                messages.push(msg);
+
+                messages.push(json!({
+                    "role": role,
+                    "content": content_text,
+                }));
             }
+
             ResponseItem::FunctionCall {
                 name,
                 arguments,
                 call_id,
                 ..
             } => {
-                let mut msg = json!({
+                messages.push(json!({
                     "role": "assistant",
-                    "content": null,
                     "tool_calls": [{
                         "id": call_id,
                         "type": "function",
@@ -228,39 +201,10 @@ pub(crate) async fn stream_chat_completions(
                             "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.
@@ -286,27 +230,43 @@ pub(crate) async fn stream_chat_completions(
                     "tool_call_id": call_id,
                     "content": content_value,
                 }));
+
+                if let Some(reasoning) = reasoning_by_anchor_index.remove(&index) {
+                    messages.push(json!({
+                        "role": "assistant",
+                        "content": reasoning,
+                    }));
+                }
             }
-            ResponseItem::CustomToolCall {
-                id,
-                call_id: _,
-                name,
-                input,
-                status: _,
-            } => {
+
+            ResponseItem::LocalShellCall { call_id, .. } => {
                 messages.push(json!({
                     "role": "assistant",
-                    "content": null,
                     "tool_calls": [{
-                        "id": id,
-                        "type": "custom",
-                        "custom": {
-                            "name": name,
-                            "input": input,
+                        "id": call_id,
+                        "type": "function",
+                        "function": {
+                            "name": "shell",
+                            "arguments": "{}", // arguments are defined via `input` only.
                         }
-                    }]
+                    }],
                 }));
             }
+
+            ResponseItem::CustomToolCall { call_id, name, .. } => {
+                messages.push(json!({
+                    "role": "assistant",
+                    "tool_calls": [{
+                        "id": call_id,
+                        "type": "function",
+                        "function": {
+                            "name": name,
+                            "arguments": "{}", // arguments are defined via `input` only.
+                        }
+                    }],
+                }));
+            }
+
             ResponseItem::CustomToolCallOutput { call_id, output } => {
                 messages.push(json!({
                     "role": "tool",
@@ -314,118 +274,159 @@ pub(crate) async fn stream_chat_completions(
                     "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;
+
+            ResponseItem::Reasoning { .. } => {
+                // Reasoning is mapped onto adjacent assistant anchors above.
             }
+
+            ResponseItem::WebSearchCall { .. } => {}
+
+            ResponseItem::Other { .. } => {}
+
+            ResponseItem::GhostSnapshot { .. } => {}
+        }
+    }
+
+    // Attach any reasoning still not mapped (e.g., if the last input items are Reasoning).
+    if messages.len() == 1 {
+        if let Some(text) = reasoning_by_anchor_index.remove(&0) {
+            messages.push(json!({
+                "role": "assistant",
+                "content": text,
+            }));
         }
     }
 
     let tools_json = create_tools_json_for_chat_completions_api(&prompt.tools)?;
-    let payload = json!({
+
+    let mut body = json!({
         "model": model_family.slug,
         "messages": messages,
         "stream": true,
-        "tools": tools_json,
+        "stream_options": {
+            "include_usage": true,
+        },
     });
 
-    debug!(
+    if !tools_json.is_empty() {
+        body["tools"] = json!(tools_json);
+        body["tool_choice"] = json!("auto");
+    }
+
+    if let SessionSource::SubAgent(sub) = session_source {
+        let subagent = if let SubAgentSource::Other(label) = sub {
+            label.clone()
+        } else {
+            serde_json::to_value(sub)
+                .ok()
+                .and_then(|v| v.as_str().map(std::string::ToString::to_string))
+                .unwrap_or_else(|| "other".to_string())
+        };
+        body["metadata"] = json!({
+            "x-openai-subagent": subagent,
+        });
+    }
+
+    let max_attempts = provider.request_max_retries();
+    let mut last_error = None;
+    for attempt in 0..=max_attempts {
+        match stream_single_chat_completion(
+            attempt,
+            client,
+            provider,
+            otel_event_manager,
+            body.clone(),
+        )
+        .await
+        {
+            Ok(stream) => return Ok(stream),
+            Err(e) => {
+                last_error = Some(e);
+                if attempt != max_attempts {
+                    tokio::time::sleep(backoff(attempt)).await;
+                }
+            }
+        }
+    }
+
+    Err(last_error.unwrap_or(CodexErr::InternalServerError))
+}
+
+async fn stream_single_chat_completion(
+    attempt: u64,
+    client: &CodexHttpClient,
+    provider: &ModelProviderInfo,
+    otel_event_manager: &OtelEventManager,
+    body: serde_json::Value,
+) -> Result<ResponseStream> {
+    trace!(
         "POST to {}: {}",
         provider.get_full_url(&None),
-        payload.to_string()
+        body.to_string()
     );
 
-    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?;
+    req_builder = req_builder
+        .header(reqwest::header::ACCEPT, "text/event-stream")
+        .json(&body);
 
-        let mut req_builder = provider.create_request_builder(client, &None).await?;
+    let res = otel_event_manager
+        .log_request(attempt, || req_builder.send())
+        .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 mut request_id = None;
+    if let Ok(resp) = &res {
+        request_id = resp
+            .headers()
+            .get("cf-ray")
+            .map(|v| v.to_str().unwrap_or_default().to_string());
+    }
+
+    match res {
+        Ok(resp) if resp.status().is_success() => {
+            let (tx_event, rx_event) = mpsc::channel::<Result<ResponseEvent>>(1600);
+
+            // spawn task to process SSE
+            let stream = resp.bytes_stream().map_err(move |e| {
+                CodexErr::ResponseStreamFailed(ResponseStreamFailed {
+                    source: e,
+                    request_id: request_id.clone(),
+                })
+            });
+            tokio::spawn(process_chat_sse(
+                stream,
+                tx_event,
+                provider.stream_idle_timeout(),
+                otel_event_manager.clone(),
+            ));
+
+            Ok(ResponseStream { rx_event })
         }
+        Ok(res) => {
+            let status = res.status();
 
-        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 });
+            if !(status == StatusCode::TOO_MANY_REQUESTS
+                || status == StatusCode::UNAUTHORIZED
+                || status.is_server_error())
+            {
+                // Surface the error body to callers. Use `unwrap_or_default` per Clippy.
+                let body = res.text().await.unwrap_or_default();
+                return Err(CodexErr::UnexpectedStatus(UnexpectedResponseError {
+                    status,
+                    body,
+                    request_id: None,
+                }));
             }
-            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;
-            }
+            Err(CodexErr::UnexpectedStatus(UnexpectedResponseError {
+                status,
+                body: String::new(),
+                request_id,
+            }))
         }
+        Err(e) => Err(CodexErr::ConnectionFailed(ConnectionFailedError {
+            source: e,
+        })),
     }
 }
 
@@ -484,9 +485,7 @@ async fn append_reasoning_text(
             .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>>,
@@ -719,33 +718,28 @@ async fn process_chat_sse<S>(
     }
 }
 
-/// 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> {
+/// Adapter that aggregates Chat Completions SSE output into the final assistant
+/// message plus optional reasoning, mirroring the Responses API contract.
+pub(crate) struct AggregatedChatStream<S>
+where
+    S: Stream<Item = Result<ResponseEvent>> + Unpin,
+{
     inner: S,
-    cumulative: String,
-    cumulative_reasoning: String,
     pending: std::collections::VecDeque<ResponseEvent>,
-    mode: AggregateMode,
+    finished: bool,
+}
+
+impl<S> AggregatedChatStream<S>
+where
+    S: Stream<Item = Result<ResponseEvent>> + Unpin,
+{
+    pub fn streaming_mode(inner: S) -> Self {
+        Self {
+            inner,
+            pending: std::collections::VecDeque::new(),
+            finished: false,
+        }
+    }
 }
 
 impl<S> Stream for AggregatedChatStream<S>
@@ -757,165 +751,44 @@ where
     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)));
+        if let Some(event) = this.pending.pop_front() {
+            return Poll::Ready(Some(Ok(event)));
+        }
+
+        if this.finished {
+            return Poll::Ready(None);
         }
 
         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))));
+                    this.pending.push_back(ResponseEvent::OutputItemDone(item));
+                    continue;
                 }
                 Poll::Ready(Some(Ok(ResponseEvent::RateLimits(snapshot)))) => {
-                    return Poll::Ready(Some(Ok(ResponseEvent::RateLimits(snapshot))));
+                    this.pending.push_back(ResponseEvent::RateLimits(snapshot));
+                    continue;
                 }
                 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 {
+                    this.pending.push_back(ResponseEvent::Completed {
                         response_id,
                         token_usage,
-                    })));
+                    });
+                    this.finished = true;
+                    return Poll::Ready(this.pending.pop_front().map(Ok));
                 }
-                Poll::Ready(Some(Ok(ResponseEvent::Created))) => {
-                    // These events are exclusive to the Responses API and
-                    // will never appear in a Chat Completions stream.
+                Poll::Ready(Some(Ok(other))) => {
+                    this.pending.push_back(other);
                     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))));
+                Poll::Ready(Some(Err(e))) => return Poll::Ready(Some(Err(e))),
+                Poll::Ready(None) => {
+                    this.finished = true;
+                    return Poll::Ready(None);
                 }
             }
         }
@@ -923,45 +796,12 @@ where
 }
 
 /// 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
-    /// }
-    /// ```
+pub(crate) trait AggregateStreamExt:
+    Stream<Item = Result<ResponseEvent>> + Sized + Unpin
+{
     fn aggregate(self) -> AggregatedChatStream<Self> {
-        AggregatedChatStream::new(self, AggregateMode::AggregatedOnly)
+        AggregatedChatStream::streaming_mode(self)
     }
 }
 
-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)
-    }
-}
+impl<T> AggregateStreamExt for T where T: Stream<Item = Result<ResponseEvent>> + Sized + Unpin {}

codex-rs/core/src/client/http.rs

@@ -0,0 +1,377 @@
+use crate::spawn::CODEX_SANDBOX_ENV_VAR;
+use http::Error as HttpError;
+use reqwest::IntoUrl;
+use reqwest::Method;
+use reqwest::Response;
+use reqwest::header::HeaderName;
+use reqwest::header::HeaderValue;
+use serde::Serialize;
+use std::collections::HashMap;
+use std::fmt::Display;
+use std::sync::LazyLock;
+use std::sync::Mutex;
+use std::sync::OnceLock;
+
+/// Set this to add a suffix to the User-Agent string.
+///
+/// It is not ideal that we're using a global singleton for this.
+/// This is primarily designed to differentiate MCP clients from each other.
+/// Because there can only be one MCP server per process, it should be safe for this to be a global static.
+/// However, future users of this should use this with caution as a result.
+/// In addition, we want to be confident that this value is used for ALL clients and doing that requires a
+/// lot of wiring and it's easy to miss code paths by doing so.
+/// See https://github.com/openai/codex/pull/3388/files for an example of what that would look like.
+/// Finally, we want to make sure this is set for ALL mcp clients without needing to know a special env var
+/// or having to set data that they already specified in the mcp initialize request somewhere else.
+///
+/// A space is automatically added between the suffix and the rest of the User-Agent string.
+/// The full user agent string is returned from the mcp initialize response.
+/// Parenthesis will be added by Codex. This should only specify what goes inside of the parenthesis.
+pub static USER_AGENT_SUFFIX: LazyLock<Mutex<Option<String>>> = LazyLock::new(|| Mutex::new(None));
+pub const DEFAULT_ORIGINATOR: &str = "codex_cli_rs";
+pub const CODEX_INTERNAL_ORIGINATOR_OVERRIDE_ENV_VAR: &str = "CODEX_INTERNAL_ORIGINATOR_OVERRIDE";
+
+#[derive(Clone, Debug)]
+pub struct CodexHttpClient {
+    inner: reqwest::Client,
+}
+
+impl CodexHttpClient {
+    fn new(inner: reqwest::Client) -> Self {
+        Self { inner }
+    }
+
+    pub fn get<U>(&self, url: U) -> CodexRequestBuilder
+    where
+        U: IntoUrl,
+    {
+        self.request(Method::GET, url)
+    }
+
+    pub fn post<U>(&self, url: U) -> CodexRequestBuilder
+    where
+        U: IntoUrl,
+    {
+        self.request(Method::POST, url)
+    }
+
+    pub fn request<U>(&self, method: Method, url: U) -> CodexRequestBuilder
+    where
+        U: IntoUrl,
+    {
+        let url_str = url.as_str().to_string();
+        CodexRequestBuilder::new(self.inner.request(method.clone(), url), method, url_str)
+    }
+}
+
+#[must_use = "requests are not sent unless `send` is awaited"]
+#[derive(Debug)]
+pub struct CodexRequestBuilder {
+    builder: reqwest::RequestBuilder,
+    method: Method,
+    url: String,
+}
+
+impl CodexRequestBuilder {
+    fn new(builder: reqwest::RequestBuilder, method: Method, url: String) -> Self {
+        Self {
+            builder,
+            method,
+            url,
+        }
+    }
+
+    fn map(self, f: impl FnOnce(reqwest::RequestBuilder) -> reqwest::RequestBuilder) -> Self {
+        Self {
+            builder: f(self.builder),
+            method: self.method,
+            url: self.url,
+        }
+    }
+
+    pub fn header<K, V>(self, key: K, value: V) -> Self
+    where
+        HeaderName: TryFrom<K>,
+        <HeaderName as TryFrom<K>>::Error: Into<HttpError>,
+        HeaderValue: TryFrom<V>,
+        <HeaderValue as TryFrom<V>>::Error: Into<HttpError>,
+    {
+        self.map(|builder| builder.header(key, value))
+    }
+
+    pub fn bearer_auth<T>(self, token: T) -> Self
+    where
+        T: Display,
+    {
+        self.map(|builder| builder.bearer_auth(token))
+    }
+
+    pub fn json<T>(self, value: &T) -> Self
+    where
+        T: ?Sized + Serialize,
+    {
+        self.map(|builder| builder.json(value))
+    }
+
+    pub async fn send(self) -> Result<Response, reqwest::Error> {
+        match self.builder.send().await {
+            Ok(response) => {
+                let request_ids = Self::extract_request_ids(&response);
+                tracing::debug!(
+                    method = %self.method,
+                    url = %self.url,
+                    status = %response.status(),
+                    request_ids = ?request_ids,
+                    version = ?response.version(),
+                    "Request completed"
+                );
+
+                Ok(response)
+            }
+            Err(error) => {
+                let status = error.status();
+                tracing::debug!(
+                    method = %self.method,
+                    url = %self.url,
+                    status = status.map(|s| s.as_u16()),
+                    error = %error,
+                    "Request failed"
+                );
+                Err(error)
+            }
+        }
+    }
+
+    fn extract_request_ids(response: &Response) -> HashMap<String, String> {
+        ["cf-ray", "x-request-id", "x-oai-request-id"]
+            .iter()
+            .filter_map(|&name| {
+                let header_name = HeaderName::from_static(name);
+                let value = response.headers().get(header_name)?;
+                let value = value.to_str().ok()?.to_owned();
+                Some((name.to_owned(), value))
+            })
+            .collect()
+    }
+}
+
+#[derive(Debug, Clone)]
+pub struct Originator {
+    pub value: String,
+    pub header_value: HeaderValue,
+}
+
+static ORIGINATOR: OnceLock<Originator> = OnceLock::new();
+
+#[derive(Debug)]
+pub enum SetOriginatorError {
+    InvalidHeaderValue,
+    AlreadyInitialized,
+}
+
+fn get_originator_value(provided: Option<String>) -> Originator {
+    let value = std::env::var(CODEX_INTERNAL_ORIGINATOR_OVERRIDE_ENV_VAR)
+        .ok()
+        .or(provided)
+        .unwrap_or(DEFAULT_ORIGINATOR.to_string());
+
+    match HeaderValue::from_str(&value) {
+        Ok(header_value) => Originator {
+            value,
+            header_value,
+        },
+        Err(e) => {
+            tracing::error!("Unable to turn originator override {value} into header value: {e}");
+            Originator {
+                value: DEFAULT_ORIGINATOR.to_string(),
+                header_value: HeaderValue::from_static(DEFAULT_ORIGINATOR),
+            }
+        }
+    }
+}
+
+pub fn set_default_originator(value: String) -> Result<(), SetOriginatorError> {
+    let originator = get_originator_value(Some(value));
+    ORIGINATOR
+        .set(originator)
+        .map_err(|_| SetOriginatorError::AlreadyInitialized)
+}
+
+pub fn originator() -> &'static Originator {
+    ORIGINATOR.get_or_init(|| get_originator_value(None))
+}
+
+pub fn get_codex_user_agent() -> String {
+    let build_version = env!("CARGO_PKG_VERSION");
+    let os_info = os_info::get();
+    let prefix = format!(
+        "{}/{build_version} ({} {}; {}) {}",
+        originator().value.as_str(),
+        os_info.os_type(),
+        os_info.version(),
+        os_info.architecture().unwrap_or("unknown"),
+        crate::terminal::user_agent()
+    );
+    let suffix = USER_AGENT_SUFFIX
+        .lock()
+        .ok()
+        .and_then(|guard| guard.clone());
+    let suffix = suffix
+        .as_deref()
+        .map(str::trim)
+        .filter(|value| !value.is_empty())
+        .map_or_else(String::new, |value| format!(" ({value})"));
+
+    let candidate = format!("{prefix}{suffix}");
+    sanitize_user_agent(candidate, &prefix)
+}
+
+/// Sanitize the user agent string.
+///
+/// Invalid characters are replaced with an underscore.
+///
+/// If the user agent fails to parse, it falls back to fallback and then to ORIGINATOR.
+fn sanitize_user_agent(candidate: String, fallback: &str) -> String {
+    if HeaderValue::from_str(candidate.as_str()).is_ok() {
+        return candidate;
+    }
+
+    let sanitized: String = candidate
+        .chars()
+        .map(|ch| if matches!(ch, ' '..='~') { ch } else { '_' })
+        .collect();
+    if !sanitized.is_empty() && HeaderValue::from_str(sanitized.as_str()).is_ok() {
+        tracing::warn!(
+            "Sanitized Codex user agent because provided suffix contained invalid header characters"
+        );
+        sanitized
+    } else if HeaderValue::from_str(fallback).is_ok() {
+        tracing::warn!(
+            "Falling back to base Codex user agent because provided suffix could not be sanitized"
+        );
+        fallback.to_string()
+    } else {
+        tracing::warn!(
+            "Falling back to default Codex originator because base user agent string is invalid"
+        );
+        originator().value.clone()
+    }
+}
+
+/// Create an HTTP client with default `originator` and `User-Agent` headers set.
+pub fn create_client() -> CodexHttpClient {
+    use reqwest::header::HeaderMap;
+
+    let mut headers = HeaderMap::new();
+    headers.insert("originator", originator().header_value.clone());
+    let ua = get_codex_user_agent();
+
+    let mut builder = reqwest::Client::builder()
+        // Set UA via dedicated helper to avoid header validation pitfalls
+        .user_agent(ua)
+        .default_headers(headers);
+    if is_sandboxed() {
+        builder = builder.no_proxy();
+    }
+
+    let inner = builder.build().unwrap_or_else(|_| reqwest::Client::new());
+    CodexHttpClient::new(inner)
+}
+
+fn is_sandboxed() -> bool {
+    std::env::var(CODEX_SANDBOX_ENV_VAR).as_deref() == Ok("seatbelt")
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use core_test_support::skip_if_no_network;
+
+    #[test]
+    fn test_get_codex_user_agent() {
+        let user_agent = get_codex_user_agent();
+        assert!(user_agent.starts_with("codex_cli_rs/"));
+    }
+
+    #[tokio::test]
+    async fn test_create_client_sets_default_headers() {
+        skip_if_no_network!();
+
+        use wiremock::Mock;
+        use wiremock::MockServer;
+        use wiremock::ResponseTemplate;
+        use wiremock::matchers::method;
+        use wiremock::matchers::path;
+
+        let client = create_client();
+
+        // Spin up a local mock server and capture a request.
+        let server = MockServer::start().await;
+        Mock::given(method("GET"))
+            .and(path("/"))
+            .respond_with(ResponseTemplate::new(200))
+            .mount(&server)
+            .await;
+
+        let resp = client
+            .get(server.uri())
+            .send()
+            .await
+            .expect("failed to send request");
+        assert!(resp.status().is_success());
+
+        let requests = server
+            .received_requests()
+            .await
+            .expect("failed to fetch received requests");
+        assert!(!requests.is_empty());
+        let headers = &requests[0].headers;
+
+        // originator header is set to the provided value
+        let originator_header = headers
+            .get("originator")
+            .expect("originator header missing");
+        assert_eq!(originator_header.to_str().unwrap(), "codex_cli_rs");
+
+        // User-Agent matches the computed Codex UA for that originator
+        let expected_ua = get_codex_user_agent();
+        let ua_header = headers
+            .get("user-agent")
+            .expect("user-agent header missing");
+        assert_eq!(ua_header.to_str().unwrap(), expected_ua);
+    }
+
+    #[test]
+    fn test_invalid_suffix_is_sanitized() {
+        let prefix = "codex_cli_rs/0.0.0";
+        let suffix = "bad\rsuffix";
+
+        assert_eq!(
+            sanitize_user_agent(format!("{prefix} ({suffix})"), prefix),
+            "codex_cli_rs/0.0.0 (bad_suffix)"
+        );
+    }
+
+    #[test]
+    fn test_invalid_suffix_is_sanitized2() {
+        let prefix = "codex_cli_rs/0.0.0";
+        let suffix = "bad\0suffix";
+
+        assert_eq!(
+            sanitize_user_agent(format!("{prefix} ({suffix})"), prefix),
+            "codex_cli_rs/0.0.0 (bad_suffix)"
+        );
+    }
+
+    #[test]
+    #[cfg(target_os = "macos")]
+    fn test_macos() {
+        use regex_lite::Regex;
+        let user_agent = get_codex_user_agent();
+        let re = Regex::new(
+            r"^codex_cli_rs/\d+\.\d+\.\d+ \(Mac OS \d+\.\d+\.\d+; (x86_64|arm64)\) (\S+)$",
+        )
+        .unwrap();
+        assert!(re.is_match(&user_agent));
+    }
+}

codex-rs/core/src/client/mod.rs

@@ -0,0 +1,18 @@
+mod chat_completions;
+pub mod http;
+mod responses;
+pub mod types;
+
+pub(crate) use chat_completions::AggregateStreamExt;
+pub(crate) use chat_completions::AggregatedChatStream;
+pub(crate) use chat_completions::stream_chat_completions;
+pub use responses::ModelClient;
+pub(crate) use types::FreeformTool;
+pub(crate) use types::FreeformToolFormat;
+pub(crate) use types::Reasoning;
+pub use types::ResponseEvent;
+pub use types::ResponseStream;
+pub(crate) use types::ResponsesApiRequest;
+pub(crate) use types::ResponsesApiTool;
+pub(crate) use types::ToolSpec;
+pub(crate) use types::create_text_param_for_request;

codex-rs/core/src/client/types.rs

@@ -0,0 +1,307 @@
+use crate::error::Result;
+use crate::protocol::RateLimitSnapshot;
+use crate::protocol::TokenUsage;
+use crate::tools::spec::JsonSchema;
+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 futures::Stream;
+use serde::Serialize;
+use serde_json::Value;
+use std::pin::Pin;
+use std::task::Context;
+use std::task::Poll;
+use tokio::sync::mpsc;
+
+#[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),
+}
+
+#[derive(Debug, Serialize)]
+pub(crate) struct Reasoning {
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub(crate) effort: Option<ReasoningEffortConfig>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub(crate) summary: Option<ReasoningSummaryConfig>,
+}
+
+#[derive(Debug, Serialize, Default, Clone)]
+#[serde(rename_all = "snake_case")]
+pub(crate) enum TextFormatType {
+    #[default]
+    JsonSchema,
+}
+
+#[derive(Debug, Serialize, Default, Clone)]
+pub(crate) struct TextFormat {
+    pub(crate) r#type: TextFormatType,
+    pub(crate) strict: bool,
+    pub(crate) schema: Value,
+    pub(crate) name: String,
+}
+
+/// Controls under the `text` field in the Responses API for GPT-5.
+#[derive(Debug, Serialize, Default, Clone)]
+pub(crate) struct TextControls {
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub(crate) verbosity: Option<OpenAiVerbosity>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub(crate) format: Option<TextFormat>,
+}
+
+#[derive(Debug, Serialize, Default, Clone)]
+#[serde(rename_all = "lowercase")]
+pub(crate) enum OpenAiVerbosity {
+    Low,
+    #[default]
+    Medium,
+    High,
+}
+
+impl From<VerbosityConfig> for OpenAiVerbosity {
+    fn from(v: VerbosityConfig) -> Self {
+        match v {
+            VerbosityConfig::Low => OpenAiVerbosity::Low,
+            VerbosityConfig::Medium => OpenAiVerbosity::Medium,
+            VerbosityConfig::High => OpenAiVerbosity::High,
+        }
+    }
+}
+
+/// Request object that is serialized as JSON and POST'ed when using the
+/// Responses API.
+#[derive(Debug, Serialize)]
+pub(crate) struct ResponsesApiRequest<'a> {
+    pub(crate) model: &'a str,
+    pub(crate) instructions: &'a str,
+    // TODO(mbolin): ResponseItem::Other should not be serialized. Currently,
+    // we code defensively to avoid this case, but perhaps we should use a
+    // separate enum for serialization.
+    pub(crate) input: &'a Vec<ResponseItem>,
+    pub(crate) tools: &'a [serde_json::Value],
+    pub(crate) tool_choice: &'static str,
+    pub(crate) parallel_tool_calls: bool,
+    pub(crate) reasoning: Option<Reasoning>,
+    pub(crate) store: bool,
+    pub(crate) stream: bool,
+    pub(crate) include: Vec<String>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub(crate) prompt_cache_key: Option<String>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub(crate) text: Option<TextControls>,
+}
+
+pub(crate) mod tools {
+    use super::JsonSchema;
+    use serde::Deserialize;
+    use serde::Serialize;
+
+    /// When serialized as JSON, this produces a valid "Tool" in the OpenAI
+    /// Responses API.
+    #[derive(Debug, Clone, Serialize, PartialEq)]
+    #[serde(tag = "type")]
+    pub(crate) enum ToolSpec {
+        #[serde(rename = "function")]
+        Function(ResponsesApiTool),
+        #[serde(rename = "local_shell")]
+        LocalShell {},
+        // TODO: Understand why we get an error on web_search although the API docs say it's supported.
+        // https://platform.openai.com/docs/guides/tools-web-search?api-mode=responses#:~:text=%7B%20type%3A%20%22web_search%22%20%7D%2C
+        #[serde(rename = "web_search")]
+        WebSearch {},
+        #[serde(rename = "custom")]
+        Freeform(FreeformTool),
+    }
+
+    impl ToolSpec {
+        pub(crate) fn name(&self) -> &str {
+            match self {
+                ToolSpec::Function(tool) => tool.name.as_str(),
+                ToolSpec::LocalShell {} => "local_shell",
+                ToolSpec::WebSearch {} => "web_search",
+                ToolSpec::Freeform(tool) => tool.name.as_str(),
+            }
+        }
+    }
+
+    #[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
+    pub struct FreeformTool {
+        pub(crate) name: String,
+        pub(crate) description: String,
+        pub(crate) format: FreeformToolFormat,
+    }
+
+    #[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
+    pub struct FreeformToolFormat {
+        pub(crate) r#type: String,
+        pub(crate) syntax: String,
+        pub(crate) definition: String,
+    }
+
+    #[derive(Debug, Clone, Serialize, PartialEq)]
+    pub struct ResponsesApiTool {
+        pub(crate) name: String,
+        pub(crate) description: String,
+        /// TODO: Validation. When strict is set to true, the JSON schema,
+        /// `required` and `additional_properties` must be present. All fields in
+        /// `properties` must be present in `required`.
+        pub(crate) strict: bool,
+        pub(crate) parameters: JsonSchema,
+    }
+}
+
+pub(crate) use tools::FreeformTool;
+pub(crate) use tools::FreeformToolFormat;
+pub(crate) use tools::ResponsesApiTool;
+pub(crate) use tools::ToolSpec;
+
+pub(crate) fn create_text_param_for_request(
+    verbosity: Option<VerbosityConfig>,
+    output_schema: &Option<Value>,
+) -> Option<TextControls> {
+    if verbosity.is_none() && output_schema.is_none() {
+        return None;
+    }
+
+    Some(TextControls {
+        verbosity: verbosity.map(std::convert::Into::into),
+        format: output_schema.as_ref().map(|schema| TextFormat {
+            r#type: TextFormatType::JsonSchema,
+            strict: true,
+            schema: schema.clone(),
+            name: "codex_output_schema".to_string(),
+        }),
+    })
+}
+
+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)
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use codex_protocol::models::ResponseItem;
+    use pretty_assertions::assert_eq;
+
+    #[test]
+    fn serializes_text_verbosity_when_set() {
+        let input: Vec<ResponseItem> = vec![];
+        let tools: Vec<serde_json::Value> = vec![];
+        let req = ResponsesApiRequest {
+            model: "gpt-5",
+            instructions: "i",
+            input: &input,
+            tools: &tools,
+            tool_choice: "auto",
+            parallel_tool_calls: true,
+            reasoning: None,
+            store: false,
+            stream: true,
+            include: vec![],
+            prompt_cache_key: None,
+            text: Some(TextControls {
+                verbosity: Some(OpenAiVerbosity::Low),
+                format: None,
+            }),
+        };
+
+        let v = serde_json::to_value(&req).expect("json");
+        assert_eq!(
+            v.get("text")
+                .and_then(|t| t.get("verbosity"))
+                .and_then(|s| s.as_str()),
+            Some("low")
+        );
+    }
+
+    #[test]
+    fn serializes_text_schema_with_strict_format() {
+        let input: Vec<ResponseItem> = vec![];
+        let tools: Vec<serde_json::Value> = vec![];
+        let schema = serde_json::json!({
+            "type": "object",
+            "properties": {
+                "answer": {"type": "string"}
+            },
+            "required": ["answer"],
+        });
+        let text_controls =
+            create_text_param_for_request(None, &Some(schema.clone())).expect("text controls");
+
+        let req = ResponsesApiRequest {
+            model: "gpt-5",
+            instructions: "i",
+            input: &input,
+            tools: &tools,
+            tool_choice: "auto",
+            parallel_tool_calls: true,
+            reasoning: None,
+            store: false,
+            stream: true,
+            include: vec![],
+            prompt_cache_key: None,
+            text: Some(text_controls),
+        };
+
+        let v = serde_json::to_value(&req).expect("json");
+        let text = v.get("text").expect("text field");
+        assert!(text.get("verbosity").is_none());
+        let format = text.get("format").expect("format field");
+
+        assert_eq!(
+            format.get("name"),
+            Some(&serde_json::Value::String("codex_output_schema".into()))
+        );
+        assert_eq!(
+            format.get("type"),
+            Some(&serde_json::Value::String("json_schema".into()))
+        );
+        assert_eq!(format.get("strict"), Some(&serde_json::Value::Bool(true)));
+        assert_eq!(format.get("schema"), Some(&schema));
+    }
+
+    #[test]
+    fn omits_text_when_not_set() {
+        let input: Vec<ResponseItem> = vec![];
+        let tools: Vec<serde_json::Value> = vec![];
+        let req = ResponsesApiRequest {
+            model: "gpt-5",
+            instructions: "i",
+            input: &input,
+            tools: &tools,
+            tool_choice: "auto",
+            parallel_tool_calls: true,
+            reasoning: None,
+            store: false,
+            stream: true,
+            include: vec![],
+            prompt_cache_key: None,
+            text: None,
+        };
+
+        let v = serde_json::to_value(&req).expect("json");
+        assert!(v.get("text").is_none());
+    }
+}

codex-rs/core/src/client_common.rs

@@ -1,24 +1,12 @@
-use crate::client_common::tools::ToolSpec;
-use crate::error::Result;
+use crate::client::ToolSpec;
 use crate::model_family::ModelFamily;
-use crate::protocol::RateLimitSnapshot;
-use crate::protocol::TokenUsage;
 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 futures::Stream;
 use serde::Deserialize;
-use serde::Serialize;
 use serde_json::Value;
 use std::borrow::Cow;
 use std::collections::HashSet;
 use std::ops::Deref;
-use std::pin::Pin;
-use std::task::Context;
-use std::task::Poll;
-use tokio::sync::mpsc;
 
 /// Review thread system prompt. Edit `core/src/review_prompt.md` to customize.
 pub const REVIEW_PROMPT: &str = include_str!("../review_prompt.md");
@@ -193,186 +181,6 @@ fn strip_total_output_header(output: &str) -> Option<&str> {
     Some(remainder)
 }
 
-#[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),
-}
-
-#[derive(Debug, Serialize)]
-pub(crate) struct Reasoning {
-    #[serde(skip_serializing_if = "Option::is_none")]
-    pub(crate) effort: Option<ReasoningEffortConfig>,
-    #[serde(skip_serializing_if = "Option::is_none")]
-    pub(crate) summary: Option<ReasoningSummaryConfig>,
-}
-
-#[derive(Debug, Serialize, Default, Clone)]
-#[serde(rename_all = "snake_case")]
-pub(crate) enum TextFormatType {
-    #[default]
-    JsonSchema,
-}
-
-#[derive(Debug, Serialize, Default, Clone)]
-pub(crate) struct TextFormat {
-    pub(crate) r#type: TextFormatType,
-    pub(crate) strict: bool,
-    pub(crate) schema: Value,
-    pub(crate) name: String,
-}
-
-/// Controls under the `text` field in the Responses API for GPT-5.
-#[derive(Debug, Serialize, Default, Clone)]
-pub(crate) struct TextControls {
-    #[serde(skip_serializing_if = "Option::is_none")]
-    pub(crate) verbosity: Option<OpenAiVerbosity>,
-    #[serde(skip_serializing_if = "Option::is_none")]
-    pub(crate) format: Option<TextFormat>,
-}
-
-#[derive(Debug, Serialize, Default, Clone)]
-#[serde(rename_all = "lowercase")]
-pub(crate) enum OpenAiVerbosity {
-    Low,
-    #[default]
-    Medium,
-    High,
-}
-
-impl From<VerbosityConfig> for OpenAiVerbosity {
-    fn from(v: VerbosityConfig) -> Self {
-        match v {
-            VerbosityConfig::Low => OpenAiVerbosity::Low,
-            VerbosityConfig::Medium => OpenAiVerbosity::Medium,
-            VerbosityConfig::High => OpenAiVerbosity::High,
-        }
-    }
-}
-
-/// Request object that is serialized as JSON and POST'ed when using the
-/// Responses API.
-#[derive(Debug, Serialize)]
-pub(crate) struct ResponsesApiRequest<'a> {
-    pub(crate) model: &'a str,
-    pub(crate) instructions: &'a str,
-    // TODO(mbolin): ResponseItem::Other should not be serialized. Currently,
-    // we code defensively to avoid this case, but perhaps we should use a
-    // separate enum for serialization.
-    pub(crate) input: &'a Vec<ResponseItem>,
-    pub(crate) tools: &'a [serde_json::Value],
-    pub(crate) tool_choice: &'static str,
-    pub(crate) parallel_tool_calls: bool,
-    pub(crate) reasoning: Option<Reasoning>,
-    pub(crate) store: bool,
-    pub(crate) stream: bool,
-    pub(crate) include: Vec<String>,
-    #[serde(skip_serializing_if = "Option::is_none")]
-    pub(crate) prompt_cache_key: Option<String>,
-    #[serde(skip_serializing_if = "Option::is_none")]
-    pub(crate) text: Option<TextControls>,
-}
-
-pub(crate) mod tools {
-    use crate::tools::spec::JsonSchema;
-    use serde::Deserialize;
-    use serde::Serialize;
-
-    /// When serialized as JSON, this produces a valid "Tool" in the OpenAI
-    /// Responses API.
-    #[derive(Debug, Clone, Serialize, PartialEq)]
-    #[serde(tag = "type")]
-    pub(crate) enum ToolSpec {
-        #[serde(rename = "function")]
-        Function(ResponsesApiTool),
-        #[serde(rename = "local_shell")]
-        LocalShell {},
-        // TODO: Understand why we get an error on web_search although the API docs say it's supported.
-        // https://platform.openai.com/docs/guides/tools-web-search?api-mode=responses#:~:text=%7B%20type%3A%20%22web_search%22%20%7D%2C
-        #[serde(rename = "web_search")]
-        WebSearch {},
-        #[serde(rename = "custom")]
-        Freeform(FreeformTool),
-    }
-
-    impl ToolSpec {
-        pub(crate) fn name(&self) -> &str {
-            match self {
-                ToolSpec::Function(tool) => tool.name.as_str(),
-                ToolSpec::LocalShell {} => "local_shell",
-                ToolSpec::WebSearch {} => "web_search",
-                ToolSpec::Freeform(tool) => tool.name.as_str(),
-            }
-        }
-    }
-
-    #[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
-    pub struct FreeformTool {
-        pub(crate) name: String,
-        pub(crate) description: String,
-        pub(crate) format: FreeformToolFormat,
-    }
-
-    #[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
-    pub struct FreeformToolFormat {
-        pub(crate) r#type: String,
-        pub(crate) syntax: String,
-        pub(crate) definition: String,
-    }
-
-    #[derive(Debug, Clone, Serialize, PartialEq)]
-    pub struct ResponsesApiTool {
-        pub(crate) name: String,
-        pub(crate) description: String,
-        /// TODO: Validation. When strict is set to true, the JSON schema,
-        /// `required` and `additional_properties` must be present. All fields in
-        /// `properties` must be present in `required`.
-        pub(crate) strict: bool,
-        pub(crate) parameters: JsonSchema,
-    }
-}
-
-pub(crate) fn create_text_param_for_request(
-    verbosity: Option<VerbosityConfig>,
-    output_schema: &Option<Value>,
-) -> Option<TextControls> {
-    if verbosity.is_none() && output_schema.is_none() {
-        return None;
-    }
-
-    Some(TextControls {
-        verbosity: verbosity.map(std::convert::Into::into),
-        format: output_schema.as_ref().map(|schema| TextFormat {
-            r#type: TextFormatType::JsonSchema,
-            strict: true,
-            schema: schema.clone(),
-            name: "codex_output_schema".to_string(),
-        }),
-    })
-}
-
-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)
-    }
-}
-
 #[cfg(test)]
 mod tests {
     use crate::model_family::find_family_for_model;
@@ -443,104 +251,4 @@ mod tests {
             assert_eq!(full, expected);
         }
     }
-
-    #[test]
-    fn serializes_text_verbosity_when_set() {
-        let input: Vec<ResponseItem> = vec![];
-        let tools: Vec<serde_json::Value> = vec![];
-        let req = ResponsesApiRequest {
-            model: "gpt-5",
-            instructions: "i",
-            input: &input,
-            tools: &tools,
-            tool_choice: "auto",
-            parallel_tool_calls: true,
-            reasoning: None,
-            store: false,
-            stream: true,
-            include: vec![],
-            prompt_cache_key: None,
-            text: Some(TextControls {
-                verbosity: Some(OpenAiVerbosity::Low),
-                format: None,
-            }),
-        };
-
-        let v = serde_json::to_value(&req).expect("json");
-        assert_eq!(
-            v.get("text")
-                .and_then(|t| t.get("verbosity"))
-                .and_then(|s| s.as_str()),
-            Some("low")
-        );
-    }
-
-    #[test]
-    fn serializes_text_schema_with_strict_format() {
-        let input: Vec<ResponseItem> = vec![];
-        let tools: Vec<serde_json::Value> = vec![];
-        let schema = serde_json::json!({
-            "type": "object",
-            "properties": {
-                "answer": {"type": "string"}
-            },
-            "required": ["answer"],
-        });
-        let text_controls =
-            create_text_param_for_request(None, &Some(schema.clone())).expect("text controls");
-
-        let req = ResponsesApiRequest {
-            model: "gpt-5",
-            instructions: "i",
-            input: &input,
-            tools: &tools,
-            tool_choice: "auto",
-            parallel_tool_calls: true,
-            reasoning: None,
-            store: false,
-            stream: true,
-            include: vec![],
-            prompt_cache_key: None,
-            text: Some(text_controls),
-        };
-
-        let v = serde_json::to_value(&req).expect("json");
-        let text = v.get("text").expect("text field");
-        assert!(text.get("verbosity").is_none());
-        let format = text.get("format").expect("format field");
-
-        assert_eq!(
-            format.get("name"),
-            Some(&serde_json::Value::String("codex_output_schema".into()))
-        );
-        assert_eq!(
-            format.get("type"),
-            Some(&serde_json::Value::String("json_schema".into()))
-        );
-        assert_eq!(format.get("strict"), Some(&serde_json::Value::Bool(true)));
-        assert_eq!(format.get("schema"), Some(&schema));
-    }
-
-    #[test]
-    fn omits_text_when_not_set() {
-        let input: Vec<ResponseItem> = vec![];
-        let tools: Vec<serde_json::Value> = vec![];
-        let req = ResponsesApiRequest {
-            model: "gpt-5",
-            instructions: "i",
-            input: &input,
-            tools: &tools,
-            tool_choice: "auto",
-            parallel_tool_calls: true,
-            reasoning: None,
-            store: false,
-            stream: true,
-            include: vec![],
-            prompt_cache_key: None,
-            text: None,
-        };
-
-        let v = serde_json::to_value(&req).expect("json");
-        assert!(v.get("text").is_none());
-    }
 }

codex-rs/core/src/codex.rs

@@ -5,6 +5,7 @@ use std::sync::Arc;
 use std::sync::atomic::AtomicU64;
 
 use crate::AuthManager;
+use crate::ResponseEvent;
 use crate::client_common::REVIEW_PROMPT;
 use crate::compact;
 use crate::features::Feature;
@@ -55,7 +56,6 @@ use tracing::warn;
 use crate::ModelProviderInfo;
 use crate::client::ModelClient;
 use crate::client_common::Prompt;
-use crate::client_common::ResponseEvent;
 use crate::config::Config;
 use crate::config::types::McpServerTransportConfig;
 use crate::config::types::ShellEnvironmentPolicy;

codex-rs/core/src/compact.rs

@@ -1,7 +1,7 @@
 use std::sync::Arc;
 
 use crate::Prompt;
-use crate::client_common::ResponseEvent;
+use crate::ResponseEvent;
 use crate::codex::Session;
 use crate::codex::TurnContext;
 use crate::codex::get_last_assistant_message_from_turn;

codex-rs/core/src/default_client.rs

@@ -1,375 +1 @@
-use crate::spawn::CODEX_SANDBOX_ENV_VAR;
-use http::Error as HttpError;
-use reqwest::IntoUrl;
-use reqwest::Method;
-use reqwest::Response;
-use reqwest::header::HeaderName;
-use reqwest::header::HeaderValue;
-use serde::Serialize;
-use std::collections::HashMap;
-use std::fmt::Display;
-use std::sync::LazyLock;
-use std::sync::Mutex;
-use std::sync::OnceLock;
-
-/// Set this to add a suffix to the User-Agent string.
-///
-/// It is not ideal that we're using a global singleton for this.
-/// This is primarily designed to differentiate MCP clients from each other.
-/// Because there can only be one MCP server per process, it should be safe for this to be a global static.
-/// However, future users of this should use this with caution as a result.
-/// In addition, we want to be confident that this value is used for ALL clients and doing that requires a
-/// lot of wiring and it's easy to miss code paths by doing so.
-/// See https://github.com/openai/codex/pull/3388/files for an example of what that would look like.
-/// Finally, we want to make sure this is set for ALL mcp clients without needing to know a special env var
-/// or having to set data that they already specified in the mcp initialize request somewhere else.
-///
-/// A space is automatically added between the suffix and the rest of the User-Agent string.
-/// The full user agent string is returned from the mcp initialize response.
-/// Parenthesis will be added by Codex. This should only specify what goes inside of the parenthesis.
-pub static USER_AGENT_SUFFIX: LazyLock<Mutex<Option<String>>> = LazyLock::new(|| Mutex::new(None));
-pub const DEFAULT_ORIGINATOR: &str = "codex_cli_rs";
-pub const CODEX_INTERNAL_ORIGINATOR_OVERRIDE_ENV_VAR: &str = "CODEX_INTERNAL_ORIGINATOR_OVERRIDE";
-
-#[derive(Clone, Debug)]
-pub struct CodexHttpClient {
-    inner: reqwest::Client,
-}
-
-impl CodexHttpClient {
-    fn new(inner: reqwest::Client) -> Self {
-        Self { inner }
-    }
-
-    pub fn get<U>(&self, url: U) -> CodexRequestBuilder
-    where
-        U: IntoUrl,
-    {
-        self.request(Method::GET, url)
-    }
-
-    pub fn post<U>(&self, url: U) -> CodexRequestBuilder
-    where
-        U: IntoUrl,
-    {
-        self.request(Method::POST, url)
-    }
-
-    pub fn request<U>(&self, method: Method, url: U) -> CodexRequestBuilder
-    where
-        U: IntoUrl,
-    {
-        let url_str = url.as_str().to_string();
-        CodexRequestBuilder::new(self.inner.request(method.clone(), url), method, url_str)
-    }
-}
-
-#[must_use = "requests are not sent unless `send` is awaited"]
-#[derive(Debug)]
-pub struct CodexRequestBuilder {
-    builder: reqwest::RequestBuilder,
-    method: Method,
-    url: String,
-}
-
-impl CodexRequestBuilder {
-    fn new(builder: reqwest::RequestBuilder, method: Method, url: String) -> Self {
-        Self {
-            builder,
-            method,
-            url,
-        }
-    }
-
-    fn map(self, f: impl FnOnce(reqwest::RequestBuilder) -> reqwest::RequestBuilder) -> Self {
-        Self {
-            builder: f(self.builder),
-            method: self.method,
-            url: self.url,
-        }
-    }
-
-    pub fn header<K, V>(self, key: K, value: V) -> Self
-    where
-        HeaderName: TryFrom<K>,
-        <HeaderName as TryFrom<K>>::Error: Into<HttpError>,
-        HeaderValue: TryFrom<V>,
-        <HeaderValue as TryFrom<V>>::Error: Into<HttpError>,
-    {
-        self.map(|builder| builder.header(key, value))
-    }
-
-    pub fn bearer_auth<T>(self, token: T) -> Self
-    where
-        T: Display,
-    {
-        self.map(|builder| builder.bearer_auth(token))
-    }
-
-    pub fn json<T>(self, value: &T) -> Self
-    where
-        T: ?Sized + Serialize,
-    {
-        self.map(|builder| builder.json(value))
-    }
-
-    pub async fn send(self) -> Result<Response, reqwest::Error> {
-        match self.builder.send().await {
-            Ok(response) => {
-                let request_ids = Self::extract_request_ids(&response);
-                tracing::debug!(
-                    method = %self.method,
-                    url = %self.url,
-                    status = %response.status(),
-                    request_ids = ?request_ids,
-                    version = ?response.version(),
-                    "Request completed"
-                );
-
-                Ok(response)
-            }
-            Err(error) => {
-                let status = error.status();
-                tracing::debug!(
-                    method = %self.method,
-                    url = %self.url,
-                    status = status.map(|s| s.as_u16()),
-                    error = %error,
-                    "Request failed"
-                );
-                Err(error)
-            }
-        }
-    }
-
-    fn extract_request_ids(response: &Response) -> HashMap<String, String> {
-        ["cf-ray", "x-request-id", "x-oai-request-id"]
-            .iter()
-            .filter_map(|&name| {
-                let header_name = HeaderName::from_static(name);
-                let value = response.headers().get(header_name)?;
-                let value = value.to_str().ok()?.to_owned();
-                Some((name.to_owned(), value))
-            })
-            .collect()
-    }
-}
-#[derive(Debug, Clone)]
-pub struct Originator {
-    pub value: String,
-    pub header_value: HeaderValue,
-}
-static ORIGINATOR: OnceLock<Originator> = OnceLock::new();
-
-#[derive(Debug)]
-pub enum SetOriginatorError {
-    InvalidHeaderValue,
-    AlreadyInitialized,
-}
-
-fn get_originator_value(provided: Option<String>) -> Originator {
-    let value = std::env::var(CODEX_INTERNAL_ORIGINATOR_OVERRIDE_ENV_VAR)
-        .ok()
-        .or(provided)
-        .unwrap_or(DEFAULT_ORIGINATOR.to_string());
-
-    match HeaderValue::from_str(&value) {
-        Ok(header_value) => Originator {
-            value,
-            header_value,
-        },
-        Err(e) => {
-            tracing::error!("Unable to turn originator override {value} into header value: {e}");
-            Originator {
-                value: DEFAULT_ORIGINATOR.to_string(),
-                header_value: HeaderValue::from_static(DEFAULT_ORIGINATOR),
-            }
-        }
-    }
-}
-
-pub fn set_default_originator(value: String) -> Result<(), SetOriginatorError> {
-    let originator = get_originator_value(Some(value));
-    ORIGINATOR
-        .set(originator)
-        .map_err(|_| SetOriginatorError::AlreadyInitialized)
-}
-
-pub fn originator() -> &'static Originator {
-    ORIGINATOR.get_or_init(|| get_originator_value(None))
-}
-
-pub fn get_codex_user_agent() -> String {
-    let build_version = env!("CARGO_PKG_VERSION");
-    let os_info = os_info::get();
-    let prefix = format!(
-        "{}/{build_version} ({} {}; {}) {}",
-        originator().value.as_str(),
-        os_info.os_type(),
-        os_info.version(),
-        os_info.architecture().unwrap_or("unknown"),
-        crate::terminal::user_agent()
-    );
-    let suffix = USER_AGENT_SUFFIX
-        .lock()
-        .ok()
-        .and_then(|guard| guard.clone());
-    let suffix = suffix
-        .as_deref()
-        .map(str::trim)
-        .filter(|value| !value.is_empty())
-        .map_or_else(String::new, |value| format!(" ({value})"));
-
-    let candidate = format!("{prefix}{suffix}");
-    sanitize_user_agent(candidate, &prefix)
-}
-
-/// Sanitize the user agent string.
-///
-/// Invalid characters are replaced with an underscore.
-///
-/// If the user agent fails to parse, it falls back to fallback and then to ORIGINATOR.
-fn sanitize_user_agent(candidate: String, fallback: &str) -> String {
-    if HeaderValue::from_str(candidate.as_str()).is_ok() {
-        return candidate;
-    }
-
-    let sanitized: String = candidate
-        .chars()
-        .map(|ch| if matches!(ch, ' '..='~') { ch } else { '_' })
-        .collect();
-    if !sanitized.is_empty() && HeaderValue::from_str(sanitized.as_str()).is_ok() {
-        tracing::warn!(
-            "Sanitized Codex user agent because provided suffix contained invalid header characters"
-        );
-        sanitized
-    } else if HeaderValue::from_str(fallback).is_ok() {
-        tracing::warn!(
-            "Falling back to base Codex user agent because provided suffix could not be sanitized"
-        );
-        fallback.to_string()
-    } else {
-        tracing::warn!(
-            "Falling back to default Codex originator because base user agent string is invalid"
-        );
-        originator().value.clone()
-    }
-}
-
-/// Create an HTTP client with default `originator` and `User-Agent` headers set.
-pub fn create_client() -> CodexHttpClient {
-    use reqwest::header::HeaderMap;
-
-    let mut headers = HeaderMap::new();
-    headers.insert("originator", originator().header_value.clone());
-    let ua = get_codex_user_agent();
-
-    let mut builder = reqwest::Client::builder()
-        // Set UA via dedicated helper to avoid header validation pitfalls
-        .user_agent(ua)
-        .default_headers(headers);
-    if is_sandboxed() {
-        builder = builder.no_proxy();
-    }
-
-    let inner = builder.build().unwrap_or_else(|_| reqwest::Client::new());
-    CodexHttpClient::new(inner)
-}
-
-fn is_sandboxed() -> bool {
-    std::env::var(CODEX_SANDBOX_ENV_VAR).as_deref() == Ok("seatbelt")
-}
-
-#[cfg(test)]
-mod tests {
-    use super::*;
-    use core_test_support::skip_if_no_network;
-
-    #[test]
-    fn test_get_codex_user_agent() {
-        let user_agent = get_codex_user_agent();
-        assert!(user_agent.starts_with("codex_cli_rs/"));
-    }
-
-    #[tokio::test]
-    async fn test_create_client_sets_default_headers() {
-        skip_if_no_network!();
-
-        use wiremock::Mock;
-        use wiremock::MockServer;
-        use wiremock::ResponseTemplate;
-        use wiremock::matchers::method;
-        use wiremock::matchers::path;
-
-        let client = create_client();
-
-        // Spin up a local mock server and capture a request.
-        let server = MockServer::start().await;
-        Mock::given(method("GET"))
-            .and(path("/"))
-            .respond_with(ResponseTemplate::new(200))
-            .mount(&server)
-            .await;
-
-        let resp = client
-            .get(server.uri())
-            .send()
-            .await
-            .expect("failed to send request");
-        assert!(resp.status().is_success());
-
-        let requests = server
-            .received_requests()
-            .await
-            .expect("failed to fetch received requests");
-        assert!(!requests.is_empty());
-        let headers = &requests[0].headers;
-
-        // originator header is set to the provided value
-        let originator_header = headers
-            .get("originator")
-            .expect("originator header missing");
-        assert_eq!(originator_header.to_str().unwrap(), "codex_cli_rs");
-
-        // User-Agent matches the computed Codex UA for that originator
-        let expected_ua = get_codex_user_agent();
-        let ua_header = headers
-            .get("user-agent")
-            .expect("user-agent header missing");
-        assert_eq!(ua_header.to_str().unwrap(), expected_ua);
-    }
-
-    #[test]
-    fn test_invalid_suffix_is_sanitized() {
-        let prefix = "codex_cli_rs/0.0.0";
-        let suffix = "bad\rsuffix";
-
-        assert_eq!(
-            sanitize_user_agent(format!("{prefix} ({suffix})"), prefix),
-            "codex_cli_rs/0.0.0 (bad_suffix)"
-        );
-    }
-
-    #[test]
-    fn test_invalid_suffix_is_sanitized2() {
-        let prefix = "codex_cli_rs/0.0.0";
-        let suffix = "bad\0suffix";
-
-        assert_eq!(
-            sanitize_user_agent(format!("{prefix} ({suffix})"), prefix),
-            "codex_cli_rs/0.0.0 (bad_suffix)"
-        );
-    }
-
-    #[test]
-    #[cfg(target_os = "macos")]
-    fn test_macos() {
-        use regex_lite::Regex;
-        let user_agent = get_codex_user_agent();
-        let re = Regex::new(
-            r"^codex_cli_rs/\d+\.\d+\.\d+ \(Mac OS \d+\.\d+\.\d+; (x86_64|arm64)\) (\S+)$",
-        )
-        .unwrap();
-        assert!(re.is_match(&user_agent));
-    }
-}
+pub use crate::client::http::*;

codex-rs/core/src/lib.rs

@@ -8,8 +8,7 @@
 mod apply_patch;
 pub mod auth;
 pub mod bash;
-mod chat_completions;
-mod client;
+pub mod client;
 mod client_common;
 pub mod codex;
 mod codex_conversation;
@@ -96,10 +95,10 @@ pub use codex_protocol::protocol;
 pub use codex_protocol::config_types as protocol_config_types;
 
 pub use client::ModelClient;
+pub use client::ResponseEvent;
+pub use client::ResponseStream;
 pub use client_common::Prompt;
 pub use client_common::REVIEW_PROMPT;
-pub use client_common::ResponseEvent;
-pub use client_common::ResponseStream;
 pub use codex_protocol::models::ContentItem;
 pub use codex_protocol::models::LocalShellAction;
 pub use codex_protocol::models::LocalShellExecAction;

codex-rs/core/src/sandboxing/assessment.rs

@@ -6,9 +6,9 @@ use std::time::Instant;
 
 use crate::AuthManager;
 use crate::ModelProviderInfo;
+use crate::ResponseEvent;
 use crate::client::ModelClient;
 use crate::client_common::Prompt;
-use crate::client_common::ResponseEvent;
 use crate::config::Config;
 use crate::protocol::SandboxPolicy;
 use askama::Template;

codex-rs/core/src/tools/handlers/apply_patch.rs

@@ -3,10 +3,10 @@ use std::collections::BTreeMap;
 use crate::apply_patch;
 use crate::apply_patch::InternalApplyPatchInvocation;
 use crate::apply_patch::convert_apply_patch_to_protocol;
-use crate::client_common::tools::FreeformTool;
-use crate::client_common::tools::FreeformToolFormat;
-use crate::client_common::tools::ResponsesApiTool;
-use crate::client_common::tools::ToolSpec;
+use crate::client::FreeformTool;
+use crate::client::FreeformToolFormat;
+use crate::client::ResponsesApiTool;
+use crate::client::ToolSpec;
 use crate::function_tool::FunctionCallError;
 use crate::tools::context::ToolInvocation;
 use crate::tools::context::ToolOutput;

codex-rs/core/src/tools/handlers/plan.rs

@@ -1,5 +1,5 @@
-use crate::client_common::tools::ResponsesApiTool;
-use crate::client_common::tools::ToolSpec;
+use crate::client::ResponsesApiTool;
+use crate::client::ToolSpec;
 use crate::codex::Session;
 use crate::codex::TurnContext;
 use crate::function_tool::FunctionCallError;

codex-rs/core/src/tools/registry.rs

@@ -2,7 +2,7 @@ use std::collections::HashMap;
 use std::sync::Arc;
 use std::time::Duration;
 
-use crate::client_common::tools::ToolSpec;
+use crate::client::ToolSpec;
 use crate::function_tool::FunctionCallError;
 use crate::tools::context::ToolInvocation;
 use crate::tools::context::ToolOutput;

codex-rs/core/src/tools/router.rs

@@ -1,7 +1,7 @@
 use std::collections::HashMap;
 use std::sync::Arc;
 
-use crate::client_common::tools::ToolSpec;
+use crate::client::ToolSpec;
 use crate::codex::Session;
 use crate::codex::TurnContext;
 use crate::function_tool::FunctionCallError;

codex-rs/core/src/tools/spec.rs

@@ -1,5 +1,5 @@
-use crate::client_common::tools::ResponsesApiTool;
-use crate::client_common::tools::ToolSpec;
+use crate::client::ResponsesApiTool;
+use crate::client::ToolSpec;
 use crate::features::Feature;
 use crate::features::Features;
 use crate::model_family::ModelFamily;
@@ -1074,7 +1074,7 @@ pub(crate) fn build_specs(
 
 #[cfg(test)]
 mod tests {
-    use crate::client_common::tools::FreeformTool;
+    use crate::client::FreeformTool;
     use crate::model_family::find_family_for_model;
     use crate::tools::registry::ConfiguredToolSpec;
     use mcp_types::ToolInputSchema;