feat: support the chat completions API in the Rust CLI

codex-rs/core/src/chat_completions.rs

@@ -0,0 +1,298 @@
+use std::time::Duration;
+
+use bytes::Bytes;
+use eventsource_stream::Eventsource;
+// (Already imported at top of file) -- no second import needed.
+use futures::StreamExt;
+use futures::TryStreamExt;
+use reqwest::StatusCode;
+use serde_json::json;
+use tokio::sync::mpsc;
+use tokio::time::timeout;
+use tracing::debug;
+use tracing::trace;
+
+use crate::ModelProviderInfo;
+use crate::client_common::Prompt;
+use crate::client_common::ResponseEvent;
+use crate::client_common::ResponseStream;
+use crate::error::CodexErr;
+use crate::error::Result;
+use crate::flags::OPENAI_REQUEST_MAX_RETRIES;
+use crate::flags::OPENAI_STREAM_IDLE_TIMEOUT_MS;
+use crate::models::ContentItem;
+use crate::models::ResponseItem;
+use crate::util::backoff;
+
+// -------------------------------------------------------------------------------------------------
+// Optional client-side aggregation helper
+// -------------------------------------------------------------------------------------------------
+
+use futures::Stream;
+use std::pin::Pin;
+use std::task::Context;
+use std::task::Poll;
+
+/// Stream adapter that merges the incremental `OutputItemDone` chunks coming from
+/// [`process_chat_sse`] into a *running* assistant message. Each time we receive
+/// a new delta we emit a fresh [`ResponseEvent::OutputItemDone`] that contains
+/// the **entire** content so far. This mirrors the on-the-fly aggregation the
+/// TypeScript CLI performs.
+///
+/// The adapter is intentionally *lossless*: callers who do **not** opt in via
+/// [`AggregateStreamExt::aggregate()`] keep receiving the original unmodified
+/// events.
+pub struct AggregatedChatStream<S> {
+    inner: S,
+    cumulative: String,
+}
+
+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();
+
+        match Pin::new(&mut this.inner).poll_next(cx) {
+            Poll::Ready(Some(Ok(ResponseEvent::OutputItemDone(item)))) => {
+                // We only aggregate assistant text messages. All other items pass through
+                // unchanged.
+                match &item {
+                    crate::models::ResponseItem::Message { role, content }
+                        if role == "assistant" =>
+                    {
+                        // Find the first OutputText part – Chat/Responses guarantee a single part
+                        // per chunk.
+                        if let Some(text) = content.iter().find_map(|c| match c {
+                            crate::models::ContentItem::OutputText { text } => Some(text),
+                            _ => None,
+                        }) {
+                            this.cumulative.push_str(text);
+
+                            // Build a new aggregated ResponseItem with the concatenated string.
+                            let agg_item = crate::models::ResponseItem::Message {
+                                role: "assistant".to_string(),
+                                content: vec![crate::models::ContentItem::OutputText {
+                                    text: this.cumulative.clone(),
+                                }],
+                            };
+
+                            return Poll::Ready(Some(Ok(ResponseEvent::OutputItemDone(agg_item))));
+                        }
+
+                        // If the message does not contain an OutputText (should never happen) just
+                        // forward unchanged.
+                        Poll::Ready(Some(Ok(ResponseEvent::OutputItemDone(item))))
+                    }
+                    _ => {
+                        // Non-assistant or non-message: forward unchanged.
+                        Poll::Ready(Some(Ok(ResponseEvent::OutputItemDone(item))))
+                    }
+                }
+            }
+            Poll::Ready(Some(event)) => {
+                // On any other event we forward as-is. Reset cumulative buffer when the
+                // stream signals completion so subsequent turns start fresh.
+                if matches!(&event, Ok(ResponseEvent::Completed { .. })) {
+                    this.cumulative.clear();
+                }
+                Poll::Ready(Some(event))
+            }
+            other => other,
+        }
+    }
+}
+
+/// Extension trait that activates aggregation on any stream of [`ResponseEvent`].
+pub trait AggregateStreamExt: Stream<Item = Result<ResponseEvent>> + Sized {
+    /// Returns a new stream that yields assistant messages whose `text` field
+    /// is the concatenation of **all** deltas received so far.
+    ///
+    /// 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 {
+            inner: self,
+            cumulative: String::new(),
+        }
+    }
+}
+
+impl<T> AggregateStreamExt for T where T: Stream<Item = Result<ResponseEvent>> + Sized {}
+
+/// Implementation for the classic Chat Completions API. This is intentionally
+/// minimal: we only stream back plain assistant text.
+pub(crate) async fn stream_chat_completions(
+    prompt: &Prompt,
+    model: &str,
+    client: &reqwest::Client,
+    provider: &ModelProviderInfo,
+) -> Result<ResponseStream> {
+    // Build messages array
+    let mut messages = Vec::<serde_json::Value>::new();
+
+    if let Some(instr) = &prompt.instructions {
+        messages.push(json!({"role": "system", "content": instr}));
+    }
+
+    for item in &prompt.input {
+        if let ResponseItem::Message { role, content } = item {
+            let mut text = String::new();
+            for c in content {
+                match c {
+                    ContentItem::InputText { text: t } | ContentItem::OutputText { text: t } => {
+                        text.push_str(t);
+                    }
+                    _ => {}
+                }
+            }
+            messages.push(json!({"role": role, "content": text}));
+        }
+    }
+
+    let payload = json!({
+        "model": model,
+        "messages": messages,
+        "stream": true
+    });
+
+    let base_url = provider.base_url.trim_end_matches('/');
+    let url = format!("{}/chat/completions", base_url);
+
+    debug!(url, "POST (chat)");
+    trace!("request payload: {}", payload);
+
+    let api_key = provider.api_key()?;
+    let mut attempt = 0;
+    loop {
+        attempt += 1;
+
+        let mut req_builder = client.post(&url);
+        if let Some(api_key) = &api_key {
+            req_builder = req_builder.bearer_auth(api_key.clone());
+        }
+        let res = 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>>(16);
+                let stream = resp.bytes_stream().map_err(CodexErr::Reqwest);
+                tokio::spawn(process_chat_sse(stream, tx_event));
+                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(status, body));
+                }
+
+                if attempt > *OPENAI_REQUEST_MAX_RETRIES {
+                    return Err(CodexErr::RetryLimit(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 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 > *OPENAI_REQUEST_MAX_RETRIES {
+                    return Err(e.into());
+                }
+                let delay = backoff(attempt);
+                tokio::time::sleep(delay).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>>)
+where
+    S: Stream<Item = Result<Bytes>> + Unpin,
+{
+    let mut stream = stream.eventsource();
+
+    let idle_timeout = *OPENAI_STREAM_IDLE_TIMEOUT_MS;
+
+    loop {
+        let sse = match timeout(idle_timeout, stream.next()).await {
+            Ok(Some(Ok(ev))) => ev,
+            Ok(Some(Err(e))) => {
+                let _ = tx_event.send(Err(CodexErr::Stream(e.to_string()))).await;
+                return;
+            }
+            Ok(None) => {
+                // Stream closed gracefully – emit Completed with dummy id.
+                let _ = tx_event
+                    .send(Ok(ResponseEvent::Completed {
+                        response_id: String::new(),
+                    }))
+                    .await;
+                return;
+            }
+            Err(_) => {
+                let _ = tx_event
+                    .send(Err(CodexErr::Stream("idle timeout waiting for SSE".into())))
+                    .await;
+                return;
+            }
+        };
+
+        // OpenAI Chat streaming sends a literal string "[DONE]" when finished.
+        if sse.data.trim() == "[DONE]" {
+            let _ = tx_event
+                .send(Ok(ResponseEvent::Completed {
+                    response_id: String::new(),
+                }))
+                .await;
+            return;
+        }
+
+        // Parse JSON chunk
+        let chunk: serde_json::Value = match serde_json::from_str(&sse.data) {
+            Ok(v) => v,
+            Err(_) => continue,
+        };
+
+        let content_opt = chunk
+            .get("choices")
+            .and_then(|c| c.get(0))
+            .and_then(|c| c.get("delta"))
+            .and_then(|d| d.get("content"))
+            .and_then(|c| c.as_str());
+
+        if let Some(content) = content_opt {
+            let item = ResponseItem::Message {
+                role: "assistant".to_string(),
+                content: vec![ContentItem::OutputText {
+                    text: content.to_string(),
+                }],
+            };
+
+            let _ = tx_event.send(Ok(ResponseEvent::OutputItemDone(item))).await;
+        }
+    }
+}

codex-rs/core/src/client.rs

@@ -1,11 +1,7 @@
 use std::collections::BTreeMap;
-use std::collections::HashMap;
 use std::io::BufRead;
 use std::path::Path;
-use std::pin::Pin;
 use std::sync::LazyLock;
-use std::task::Context;
-use std::task::Poll;
 use std::time::Duration;
 
 use bytes::Bytes;
@@ -23,66 +19,22 @@ use tracing::debug;
 use tracing::trace;
 use tracing::warn;
 
+use crate::chat_completions::stream_chat_completions;
+use crate::client_common::Payload;
+use crate::client_common::Prompt;
+use crate::client_common::Reasoning;
+use crate::client_common::ResponseEvent;
+use crate::client_common::ResponseStream;
 use crate::error::CodexErr;
 use crate::error::Result;
 use crate::flags::CODEX_RS_SSE_FIXTURE;
 use crate::flags::OPENAI_REQUEST_MAX_RETRIES;
 use crate::flags::OPENAI_STREAM_IDLE_TIMEOUT_MS;
 use crate::model_provider_info::ModelProviderInfo;
+use crate::model_provider_info::WireApi;
 use crate::models::ResponseItem;
 use crate::util::backoff;
 
-/// API request payload for a single model turn.
-#[derive(Default, Debug, Clone)]
-pub struct Prompt {
-    /// Conversation context input items.
-    pub input: Vec<ResponseItem>,
-    /// Optional previous response ID (when storage is enabled).
-    pub prev_id: Option<String>,
-    /// Optional initial instructions (only sent on first turn).
-    pub instructions: Option<String>,
-    /// Whether to store response on server side (disable_response_storage = !store).
-    pub store: bool,
-
-    /// Additional tools sourced from external MCP servers. Note each key is
-    /// the "fully qualified" tool name (i.e., prefixed with the server name),
-    /// which should be reported to the model in place of Tool::name.
-    pub extra_tools: HashMap<String, mcp_types::Tool>,
-}
-
-#[derive(Debug)]
-pub enum ResponseEvent {
-    OutputItemDone(ResponseItem),
-    Completed { response_id: String },
-}
-
-#[derive(Debug, Serialize)]
-struct Payload<'a> {
-    model: &'a str,
-    #[serde(skip_serializing_if = "Option::is_none")]
-    instructions: Option<&'a String>,
-    // 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.
-    input: &'a Vec<ResponseItem>,
-    tools: &'a [serde_json::Value],
-    tool_choice: &'static str,
-    parallel_tool_calls: bool,
-    reasoning: Option<Reasoning>,
-    #[serde(skip_serializing_if = "Option::is_none")]
-    previous_response_id: Option<String>,
-    /// true when using the Responses API.
-    store: bool,
-    stream: bool,
-}
-
-#[derive(Debug, Serialize)]
-struct Reasoning {
-    effort: &'static str,
-    #[serde(skip_serializing_if = "Option::is_none")]
-    generate_summary: Option<bool>,
-}
-
 /// When serialized as JSON, this produces a valid "Tool" in the OpenAI
 /// Responses API.
 #[derive(Debug, Serialize)]
@@ -152,7 +104,20 @@ impl ModelClient {
         }
     }
 
-    pub async fn stream(&mut self, prompt: &Prompt) -> Result<ResponseStream> {
+    /// Dispatches to either the Responses or Chat implementation depending on
+    /// the provider config.  Public callers always invoke `stream()` – the
+    /// specialised helpers are private to avoid accidental misuse.
+    pub async fn stream(&self, prompt: &Prompt) -> Result<ResponseStream> {
+        match self.provider.wire_api {
+            WireApi::Responses => self.stream_responses(prompt).await,
+            WireApi::Chat => {
+                stream_chat_completions(prompt, &self.model, &self.client, &self.provider).await
+            }
+        }
+    }
+
+    /// Implementation for the OpenAI *Responses* experimental API.
+    async fn stream_responses(&self, prompt: &Prompt) -> Result<ResponseStream> {
         if let Some(path) = &*CODEX_RS_SSE_FIXTURE {
             // short circuit for tests
             warn!(path, "Streaming from fixture");
@@ -202,8 +167,8 @@ impl ModelClient {
 
             let api_key = self
                 .provider
-                .api_key()
-                .ok_or_else(|| crate::error::CodexErr::EnvVar("API_KEY"))?;
+                .api_key()?
+                .expect("Repsones API requires an API key");
             let res = self
                 .client
                 .post(&url)
@@ -396,18 +361,6 @@ where
     }
 }
 
-pub struct ResponseStream {
-    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)
-    }
-}
-
 /// used in tests to stream from a text SSE file
 async fn stream_from_fixture(path: impl AsRef<Path>) -> Result<ResponseStream> {
     let (tx_event, rx_event) = mpsc::channel::<Result<ResponseEvent>>(16);

codex-rs/core/src/client_common.rs

@@ -0,0 +1,72 @@
+use crate::error::Result;
+use crate::models::ResponseItem;
+use futures::Stream;
+use serde::Serialize;
+use std::collections::HashMap;
+use std::pin::Pin;
+use std::task::Context;
+use std::task::Poll;
+use tokio::sync::mpsc;
+
+/// API request payload for a single model turn.
+#[derive(Default, Debug, Clone)]
+pub struct Prompt {
+    /// Conversation context input items.
+    pub input: Vec<ResponseItem>,
+    /// Optional previous response ID (when storage is enabled).
+    pub prev_id: Option<String>,
+    /// Optional initial instructions (only sent on first turn).
+    pub instructions: Option<String>,
+    /// Whether to store response on server side (disable_response_storage = !store).
+    pub store: bool,
+
+    /// Additional tools sourced from external MCP servers. Note each key is
+    /// the "fully qualified" tool name (i.e., prefixed with the server name),
+    /// which should be reported to the model in place of Tool::name.
+    pub extra_tools: HashMap<String, mcp_types::Tool>,
+}
+
+#[derive(Debug)]
+pub enum ResponseEvent {
+    OutputItemDone(ResponseItem),
+    Completed { response_id: String },
+}
+
+#[derive(Debug, Serialize)]
+pub(crate) struct Reasoning {
+    pub(crate) effort: &'static str,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub(crate) generate_summary: Option<bool>,
+}
+
+#[derive(Debug, Serialize)]
+pub(crate) struct Payload<'a> {
+    pub(crate) model: &'a str,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub(crate) instructions: Option<&'a String>,
+    // 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>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub(crate) previous_response_id: Option<String>,
+    /// true when using the Responses API.
+    pub(crate) store: bool,
+    pub(crate) stream: bool,
+}
+
+pub(crate) 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/codex.rs

@@ -28,9 +28,10 @@ use tracing::info;
 use tracing::trace;
 use tracing::warn;
 
+use crate::AggregateStreamExt;
 use crate::client::ModelClient;
-use crate::client::Prompt;
-use crate::client::ResponseEvent;
+use crate::client_common::Prompt;
+use crate::client_common::ResponseEvent;
 use crate::config::Config;
 use crate::error::CodexErr;
 use crate::error::Result as CodexResult;
@@ -791,6 +792,7 @@ async fn run_turn(
         match try_run_turn(sess, &sub_id, &prompt).await {
             Ok(output) => return Ok(output),
             Err(CodexErr::Interrupted) => return Err(CodexErr::Interrupted),
+            Err(CodexErr::EnvVar(var)) => return Err(CodexErr::EnvVar(var)),
             Err(e) => {
                 if retries < *OPENAI_STREAM_MAX_RETRIES {
                     retries += 1;
@@ -835,7 +837,7 @@ async fn try_run_turn(
     sub_id: &str,
     prompt: &Prompt,
 ) -> CodexResult<Vec<ProcessedResponseItem>> {
-    let mut stream = sess.client.clone().stream(prompt).await?;
+    let mut stream = sess.client.clone().stream(prompt).await?.aggregate();
 
     // Buffer all the incoming messages from the stream first, then execute them.
     // If we execute a function call in the middle of handling the stream, it can time out.

codex-rs/core/src/error.rs

@@ -55,7 +55,7 @@ pub enum CodexErr {
 
     /// Returned by run_command_stream when the user pressed Ctrl‑C (SIGINT). Session uses this to
     /// surface a polite FunctionCallOutput back to the model instead of crashing the CLI.
-    #[error("interrupted (Ctrl‑C)")]
+    #[error("interrupted (Ctrl-C)")]
     Interrupted,
 
     /// Unexpected HTTP status code.
@@ -97,8 +97,28 @@ pub enum CodexErr {
     #[error(transparent)]
     TokioJoin(#[from] JoinError),
 
-    #[error("missing environment variable {0}")]
-    EnvVar(&'static str),
+    #[error("{0}")]
+    EnvVar(EnvVarError),
+}
+
+#[derive(Debug)]
+pub struct EnvVarError {
+    /// Name of the environment variable that is missing.
+    pub var: String,
+
+    /// Optional instructions to help the user get a valid value for the
+    /// variable and set it.
+    pub instructions: Option<String>,
+}
+
+impl std::fmt::Display for EnvVarError {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        write!(f, "Missing environment variable: `{}`.", self.var)?;
+        if let Some(instructions) = &self.instructions {
+            write!(f, "\n{instructions}")?;
+        }
+        Ok(())
+    }
 }
 
 impl CodexErr {

codex-rs/core/src/lib.rs

@@ -5,7 +5,13 @@
 // the TUI or the tracing stack).
 #![deny(clippy::print_stdout, clippy::print_stderr)]
 
+mod chat_completions;
+
+// Re-export the optional aggregation helper so callers can opt-in with
+// `stream.aggregate()` without needing to know the private module layout.
+pub use chat_completions::AggregateStreamExt;
 mod client;
+mod client_common;
 pub mod codex;
 pub use codex::Codex;
 pub mod codex_wrapper;
@@ -21,6 +27,7 @@ pub mod mcp_server_config;
 mod mcp_tool_call;
 mod model_provider_info;
 pub use model_provider_info::ModelProviderInfo;
+pub use model_provider_info::WireApi;
 mod models;
 pub mod protocol;
 mod rollout;

codex-rs/core/src/model_provider_info.rs

@@ -9,6 +9,24 @@ use serde::Deserialize;
 use serde::Serialize;
 use std::collections::HashMap;
 
+use crate::error::EnvVarError;
+
+/// 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 experimental “Responses” API exposed by OpenAI at `/v1/responses`.
+    #[default]
+    Responses,
+    /// Regular Chat Completions compatible with `/v1/chat/completions`.
+    Chat,
+}
+
 /// Serializable representation of a provider definition.
 #[derive(Debug, Clone, Deserialize, Serialize)]
 pub struct ModelProviderInfo {
@@ -17,13 +35,28 @@ pub struct ModelProviderInfo {
     /// Base URL for the provider's OpenAI-compatible API.
     pub base_url: String,
     /// Environment variable that stores the user's API key for this provider.
-    pub env_key: String,
+    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>,
+
+    /// Which wire protocol this provider expects.
+    pub wire_api: WireApi,
 }
 
 impl ModelProviderInfo {
     /// Returns the API key for this provider if present in the environment.
-    pub fn api_key(&self) -> Option<String> {
-        std::env::var(&self.env_key).ok()
+    pub fn api_key(&self) -> crate::error::Result<Option<String>> {
+        match &self.env_key {
+            Some(env_key) => std::env::var(env_key).map(Some).map_err(|_| {
+                crate::error::CodexErr::EnvVar(EnvVarError {
+                    var: env_key.clone(),
+                    instructions: self.env_key_instructions.clone(),
+                })
+            }),
+            None => Ok(None),
+        }
     }
 }
 
@@ -37,7 +70,9 @@ pub fn built_in_model_providers() -> HashMap<String, ModelProviderInfo> {
             P {
                 name: "OpenAI".into(),
                 base_url: "https://api.openai.com/v1".into(),
-                env_key: "OPENAI_API_KEY".into(),
+                env_key: Some("OPENAI_API_KEY".into()),
+                env_key_instructions: Some("Create an API key (https://platform.openai.com) and export as an environment variable".into()),
+                wire_api: WireApi::Responses,
             },
         ),
         (
@@ -45,7 +80,9 @@ pub fn built_in_model_providers() -> HashMap<String, ModelProviderInfo> {
             P {
                 name: "OpenRouter".into(),
                 base_url: "https://openrouter.ai/api/v1".into(),
-                env_key: "OPENROUTER_API_KEY".into(),
+                env_key: Some("OPENROUTER_API_KEY".into()),
+                env_key_instructions: None,
+                wire_api: WireApi::Chat,
             },
         ),
         (
@@ -53,7 +90,9 @@ pub fn built_in_model_providers() -> HashMap<String, ModelProviderInfo> {
             P {
                 name: "Gemini".into(),
                 base_url: "https://generativelanguage.googleapis.com/v1beta/openai".into(),
-                env_key: "GEMINI_API_KEY".into(),
+                env_key: Some("GEMINI_API_KEY".into()),
+                env_key_instructions: None,
+                wire_api: WireApi::Chat,
             },
         ),
         (
@@ -61,7 +100,9 @@ pub fn built_in_model_providers() -> HashMap<String, ModelProviderInfo> {
             P {
                 name: "Ollama".into(),
                 base_url: "http://localhost:11434/v1".into(),
-                env_key: "OLLAMA_API_KEY".into(),
+                env_key: None,
+                env_key_instructions: None,
+                wire_api: WireApi::Chat,
             },
         ),
         (
@@ -69,7 +110,9 @@ pub fn built_in_model_providers() -> HashMap<String, ModelProviderInfo> {
             P {
                 name: "Mistral".into(),
                 base_url: "https://api.mistral.ai/v1".into(),
-                env_key: "MISTRAL_API_KEY".into(),
+                env_key: Some("MISTRAL_API_KEY".into()),
+                env_key_instructions: None,
+                wire_api: WireApi::Chat,
             },
         ),
         (
@@ -77,7 +120,9 @@ pub fn built_in_model_providers() -> HashMap<String, ModelProviderInfo> {
             P {
                 name: "DeepSeek".into(),
                 base_url: "https://api.deepseek.com".into(),
-                env_key: "DEEPSEEK_API_KEY".into(),
+                env_key: Some("DEEPSEEK_API_KEY".into()),
+                env_key_instructions: None,
+                wire_api: WireApi::Chat,
             },
         ),
         (
@@ -85,7 +130,9 @@ pub fn built_in_model_providers() -> HashMap<String, ModelProviderInfo> {
             P {
                 name: "xAI".into(),
                 base_url: "https://api.x.ai/v1".into(),
-                env_key: "XAI_API_KEY".into(),
+                env_key: Some("XAI_API_KEY".into()),
+                env_key_instructions: None,
+                wire_api: WireApi::Chat,
             },
         ),
         (
@@ -93,7 +140,9 @@ pub fn built_in_model_providers() -> HashMap<String, ModelProviderInfo> {
             P {
                 name: "Groq".into(),
                 base_url: "https://api.groq.com/openai/v1".into(),
-                env_key: "GROQ_API_KEY".into(),
+                env_key: Some("GROQ_API_KEY".into()),
+                env_key_instructions: None,
+                wire_api: WireApi::Chat,
             },
         ),
     ]

codex-rs/core/src/models.rs

@@ -116,7 +116,6 @@ pub struct ShellToolCallParams {
     pub timeout_ms: Option<u64>,
 }
 
-#[expect(dead_code)]
 #[derive(Deserialize, Debug, Clone)]
 pub struct FunctionCallOutputPayload {
     pub content: String,

codex-rs/core/src/protocol.rs

@@ -25,6 +25,7 @@ pub struct Submission {
 /// Submission operation
 #[derive(Debug, Clone, Deserialize, Serialize)]
 #[serde(tag = "type", rename_all = "snake_case")]
+#[allow(clippy::large_enum_variant)]
 #[non_exhaustive]
 pub enum Op {
     /// Configure the model session.

codex-rs/core/tests/previous_response_id.rs

@@ -90,7 +90,9 @@ async fn keeps_previous_response_id_between_tasks() {
         // Environment variable that should exist in the test environment.
         // ModelClient will return an error if the environment variable for the
         // provider is not set.
-        env_key: "PATH".into(),
+        env_key: Some("PATH".into()),
+        env_key_instructions: None,
+        wire_api: codex_core::WireApi::Responses,
     };
 
     // Init session

codex-rs/core/tests/stream_no_completed.rs

@@ -80,7 +80,9 @@ async fn retries_on_early_close() {
         // Environment variable that should exist in the test environment.
         // ModelClient will return an error if the environment variable for the
         // provider is not set.
-        env_key: "PATH".into(),
+        env_key: Some("PATH".into()),
+        env_key_instructions: None,
+        wire_api: codex_core::WireApi::Responses,
     };
 
     let ctrl_c = std::sync::Arc::new(tokio::sync::Notify::new());

codex-rs/exec/src/lib.rs

@@ -16,8 +16,6 @@ use codex_core::protocol::Op;
 use codex_core::protocol::SandboxPolicy;
 use codex_core::util::is_inside_git_repo;
 use event_processor::EventProcessor;
-use owo_colors::OwoColorize;
-use owo_colors::Style;
 use tracing::debug;
 use tracing::error;
 use tracing::info;
@@ -45,8 +43,6 @@ pub async fn run_main(cli: Cli) -> anyhow::Result<()> {
         ),
     };
 
-    assert_api_key(stderr_with_ansi);
-
     let sandbox_policy = if full_auto {
         Some(SandboxPolicy::new_full_auto_policy())
     } else {
@@ -161,38 +157,3 @@ pub async fn run_main(cli: Cli) -> anyhow::Result<()> {
 
     Ok(())
 }
-
-/// If a valid API key is not present in the environment, print an error to
-/// stderr and exits with 1; otherwise, does nothing.
-fn assert_api_key(stderr_with_ansi: bool) {
-    if !has_api_key() {
-        let (msg_style, var_style, url_style) = if stderr_with_ansi {
-            (
-                Style::new().red(),
-                Style::new().bold(),
-                Style::new().bold().underline(),
-            )
-        } else {
-            (Style::new(), Style::new(), Style::new())
-        };
-
-        eprintln!(
-            "\n{msg}\n\nSet the environment variable {var} and re-run this command.\nYou can create a key here: {url}\n",
-            msg = "Missing OpenAI API key.".style(msg_style),
-            var = "OPENAI_API_KEY".style(var_style),
-            url = "https://platform.openai.com/account/api-keys".style(url_style),
-        );
-        std::process::exit(1);
-    }
-}
-
-/// Returns `true` if a recognized API key is present in the environment.
-///
-/// At present we only support `OPENAI_API_KEY`, mirroring the behavior of the
-/// Node-based `codex-cli`. Additional providers can be added here when the
-/// Rust implementation gains first-class support for them.
-fn has_api_key() -> bool {
-    std::env::var("OPENAI_API_KEY")
-        .map(|s| !s.trim().is_empty())
-        .unwrap_or(false)
-}

codex-rs/tui/src/app_event.rs

@@ -1,6 +1,7 @@
 use codex_core::protocol::Event;
 use crossterm::event::KeyEvent;
 
+#[allow(clippy::large_enum_variant)]
 pub(crate) enum AppEvent {
     CodexEvent(Event),
 

codex-rs/tui/src/chatwidget.rs

@@ -240,8 +240,7 @@ impl ChatWidget<'_> {
                 self.request_redraw()?;
             }
             EventMsg::Error { message } => {
-                self.conversation_history
-                    .add_background_event(format!("Error: {message}"));
+                self.conversation_history.add_error(message);
                 self.bottom_pane.set_task_running(false)?;
             }
             EventMsg::ExecApprovalRequest {

codex-rs/tui/src/conversation_history_widget.rs

@@ -174,6 +174,10 @@ impl ConversationHistoryWidget {
         self.add_to_history(HistoryCell::new_background_event(message));
     }
 
+    pub fn add_error(&mut self, message: String) {
+        self.add_to_history(HistoryCell::new_error_event(message));
+    }
+
     /// Add a pending patch entry (before user approval).
     pub fn add_patch_event(
         &mut self,

codex-rs/tui/src/history_cell.rs

@@ -69,6 +69,9 @@ pub(crate) enum HistoryCell {
     /// Background event
     BackgroundEvent { lines: Vec<Line<'static>> },
 
+    /// Error event from the backend.
+    ErrorEvent { lines: Vec<Line<'static>> },
+
     /// Info describing the newly‑initialized session.
     SessionInfo { lines: Vec<Line<'static>> },
 
@@ -245,6 +248,12 @@ impl HistoryCell {
         HistoryCell::BackgroundEvent { lines }
     }
 
+    pub(crate) fn new_error_event(message: String) -> Self {
+        let lines: Vec<Line<'static>> =
+            vec![vec!["ERROR".red().bold(), message.into()].into(), "".into()];
+        HistoryCell::ErrorEvent { lines }
+    }
+
     pub(crate) fn new_session_info(config: &Config, model: String) -> Self {
         let mut lines: Vec<Line<'static>> = Vec::new();
 
@@ -332,6 +341,7 @@ impl HistoryCell {
             HistoryCell::UserPrompt { lines, .. }
             | HistoryCell::AgentMessage { lines, .. }
             | HistoryCell::BackgroundEvent { lines, .. }
+            | HistoryCell::ErrorEvent { lines, .. }
             | HistoryCell::SessionInfo { lines, .. }
             | HistoryCell::ActiveExecCommand { lines, .. }
             | HistoryCell::CompletedExecCommand { lines, .. }

codex-rs/tui/src/lib.rs

@@ -34,8 +34,6 @@ mod user_approval_widget;
 pub use cli::Cli;
 
 pub fn run_main(cli: Cli) -> std::io::Result<()> {
-    assert_env_var_set();
-
     let (sandbox_policy, approval_policy) = if cli.full_auto {
         (
             Some(SandboxPolicy::new_full_auto_policy()),
@@ -173,20 +171,6 @@ fn run_ratatui_app(
     app_result
 }
 
-#[expect(
-    clippy::print_stderr,
-    reason = "TUI should not have been displayed yet, so we can write to stderr."
-)]
-fn assert_env_var_set() {
-    if std::env::var("OPENAI_API_KEY").is_err() {
-        eprintln!("Welcome to codex! It looks like you're missing: `OPENAI_API_KEY`");
-        eprintln!(
-            "Create an API key (https://platform.openai.com) and export as an environment variable"
-        );
-        std::process::exit(1);
-    }
-}
-
 #[expect(
     clippy::print_stderr,
     reason = "TUI should no longer be displayed, so we can write to stderr."