aither_openai/

client.rs

use crate::PROVIDER_NAME;
use crate::{
    DEEPSEEK_BASE_URL, DEFAULT_AUDIO_FORMAT, DEFAULT_AUDIO_MODEL, DEFAULT_AUDIO_VOICE,
    DEFAULT_BASE_URL, DEFAULT_EMBEDDING_DIM, DEFAULT_EMBEDDING_MODEL, DEFAULT_IMAGE_MODEL,
    DEFAULT_MODEL, DEFAULT_MODERATION_MODEL, DEFAULT_TRANSCRIPTION_MODEL, OPENROUTER_BASE_URL,
    error::OpenAIError,
    request::{
        ChatCompletionRequest, ChatMessagePayload, ParameterSnapshot, ResponsesInputItem,
        ResponsesRequest, ResponsesTool, ToolPayload, convert_responses_tools, convert_tools,
        responses_tool_choice, to_chat_messages, to_responses_input,
    },
    response::{
        ChatCompletionChunk, ChatCompletionUsage, DeltaToolCall, ResponsesOutputItem,
        ResponsesStreamEvent, ResponsesUsage, should_skip_event,
    },
};
use aither_core::{
    LanguageModel,
    llm::{
        Event, LLMRequest, ReasoningState, ToolCall, Usage,
        model::{Ability, Profile as ModelProfile, ToolChoice},
        oneshot,
    },
};
use futures_core::Stream;
use futures_lite::StreamExt;
use std::{collections::HashMap, future::Future, sync::Arc, time::Duration};
use zenwave::{Client, client, header};

/// Configuration for request retry behavior.
#[derive(Debug, Clone)]
pub struct RetryConfig {
    /// Maximum number of retry attempts (0 = no retries).
    pub max_retries: u32,
    /// Initial delay before first retry.
    pub initial_delay: Duration,
    /// Maximum delay between retries.
    pub max_delay: Duration,
    /// Multiplier for exponential backoff.
    pub backoff_multiplier: f64,
}

impl Default for RetryConfig {
    fn default() -> Self {
        Self {
            max_retries: 3,
            initial_delay: Duration::from_millis(500),
            max_delay: Duration::from_secs(30),
            backoff_multiplier: 2.0,
        }
    }
}

impl RetryConfig {
    /// Create a config with no retries.
    #[must_use]
    pub fn none() -> Self {
        Self {
            max_retries: 0,
            ..Default::default()
        }
    }

    /// Calculate delay for a given attempt number (0-indexed).
    #[allow(
        clippy::cast_possible_truncation,
        clippy::cast_possible_wrap,
        clippy::cast_precision_loss,
        clippy::cast_sign_loss
    )]
    fn delay_for_attempt(&self, attempt: u32) -> Duration {
        let delay_ms =
            self.initial_delay.as_millis() as f64 * self.backoff_multiplier.powi(attempt as i32);
        let delay = Duration::from_millis(delay_ms as u64);
        delay.min(self.max_delay)
    }
}

fn is_retryable_json_error(error: &serde_json::Error) -> bool {
    matches!(error.classify(), serde_json::error::Category::Eof)
}

/// Check if an error is retryable.
fn is_retryable_error(err: &OpenAIError) -> bool {
    match err {
        OpenAIError::Http(_)
        | OpenAIError::Body(_)
        | OpenAIError::Stream(_)
        | OpenAIError::RateLimit { .. }
        | OpenAIError::ServerError { .. }
        | OpenAIError::Timeout => true,
        // API and decode errors are generally not retryable.
        OpenAIError::Api(_) | OpenAIError::Decode(_) => false,
        // EOF while parsing JSON is typically a truncated/empty upstream body.
        OpenAIError::Json(error) => is_retryable_json_error(error),
    }
}

/// Get retry delay for an error, respecting Retry-After header for rate limits.
fn get_retry_delay(err: &OpenAIError, attempt: u32, config: &RetryConfig) -> Duration {
    if let OpenAIError::RateLimit {
        retry_after: Some(delay),
        ..
    } = err
    {
        // Respect Retry-After header, but cap at max_delay
        return (*delay).min(config.max_delay);
    }
    config.delay_for_attempt(attempt)
}

/// Sleep for the given duration (runtime-agnostic).
async fn sleep(duration: Duration) {
    #[cfg(not(target_arch = "wasm32"))]
    {
        async_io::Timer::after(duration).await;
    }
    #[cfg(target_arch = "wasm32")]
    {
        let _ = duration;
    }
}

fn map_zenwave_error(error: zenwave::Error) -> OpenAIError {
    if matches!(error, zenwave::Error::Timeout) {
        OpenAIError::Timeout
    } else {
        OpenAIError::Http(error)
    }
}

async fn request_with_timeout<T>(
    timeout: Duration,
    fut: impl Future<Output = Result<T, zenwave::Error>>,
) -> Result<T, OpenAIError> {
    #[cfg(not(target_arch = "wasm32"))]
    {
        futures_lite::future::or(
            async move { fut.await.map_err(map_zenwave_error) },
            async move {
                async_io::Timer::after(timeout).await;
                Err(OpenAIError::Timeout)
            },
        )
        .await
    }
    #[cfg(target_arch = "wasm32")]
    {
        // The browser timer is not `Send`, and `LanguageModel` requires the
        // futures built on top of this one to be. Run the timer on the JS event
        // loop where it belongs and wait on a channel, which is `Send`.
        let millis = timeout.as_millis().min(u128::from(u32::MAX)) as u32;
        let (expired_tx, expired_rx) = async_channel::bounded::<()>(1);
        wasm_bindgen_futures::spawn_local(async move {
            gloo_timers::future::TimeoutFuture::new(millis).await;
            let _ = expired_tx.send(()).await;
        });

        futures_lite::future::or(
            async move { fut.await.map_err(map_zenwave_error) },
            async move {
                // A closed channel means the timer task was dropped, which can
                // only happen once the whole request is being torn down.
                let _ = expired_rx.recv().await;
                Err(OpenAIError::Timeout)
            },
        )
        .await
    }
}

/// Result of attempting to establish an SSE stream.
type SseStreamResult = Result<zenwave::sse::SseStream, OpenAIError>;

/// Attempt to make an SSE request with retry logic.
///
/// Returns a streaming SSE connection on success, or the last error on failure.
async fn sse_request_with_retry<F, Fut>(cfg: &Config, make_request: F) -> SseStreamResult
where
    F: Fn() -> Fut,
    Fut: std::future::Future<Output = SseStreamResult>,
{
    let retry_config = &cfg.retry;
    let mut attempt = 0;

    loop {
        match make_request().await {
            Ok(stream) => return Ok(stream),
            Err(err) => {
                // Check if we should retry
                if attempt < retry_config.max_retries && is_retryable_error(&err) {
                    let delay = get_retry_delay(&err, attempt, retry_config);
                    tracing::warn!(
                        attempt = attempt + 1,
                        max_retries = retry_config.max_retries,
                        delay_ms = delay.as_millis(),
                        error = %err,
                        "Request failed, retrying"
                    );
                    sleep(delay).await;
                    attempt += 1;
                } else {
                    return Err(err);
                }
            }
        }
    }
}

/// `OpenAI` model backed by the Responses API by default, with legacy
/// `chat.completions` support for compatibility.
#[derive(Clone, Debug)]
pub struct OpenAI {
    inner: Arc<Config>,
}

/// Selects which `OpenAI` API surface to use.
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum ApiKind {
    /// The recommended Responses API.
    Responses,
    /// Legacy Chat Completions API (deprecated by `OpenAI`).
    ChatCompletions,
}

impl OpenAI {
    /// Create a new client using the provided API key and default model.
    pub fn new(api_key: impl Into<String>) -> Self {
        Self::builder(api_key).build()
    }

    /// Create a client configured for [`Deepseek`](https://api-docs.deepseek.com)'s OpenAI-compatible endpoint.
    pub fn deepseek(api_key: impl Into<String>) -> Self {
        Self::builder(api_key).base_url(DEEPSEEK_BASE_URL).build()
    }

    /// Create a client configured for [`OpenRouter`](https://openrouter.ai)'s OpenAI-compatible endpoint.
    pub fn openrouter(api_key: impl Into<String>) -> Self {
        Self::builder(api_key).base_url(OPENROUTER_BASE_URL).build()
    }

    /// Start building an [`OpenAI`] client with custom configuration.
    #[must_use]
    pub fn builder(api_key: impl Into<String>) -> Builder {
        Builder::new(api_key)
    }

    /// Override the default chat model in-place.
    #[must_use]
    pub fn with_model(mut self, model: impl Into<String>) -> Self {
        Arc::make_mut(&mut self.inner).chat_model = sanitize_model(model);
        self
    }

    /// Override the REST base URL (useful for OpenAI-compatible endpoints).
    #[must_use]
    pub fn with_base_url(mut self, base_url: impl Into<String>) -> Self {
        Arc::make_mut(&mut self.inner).base_url = base_url.into();
        self
    }

    /// Select which `OpenAI` API to call.
    #[must_use]
    pub fn with_api(mut self, api: ApiKind) -> Self {
        Arc::make_mut(&mut self.inner).api_kind = api;
        self
    }

    /// Use the recommended Responses API.
    #[must_use]
    pub fn with_responses_api(self) -> Self {
        self.with_api(ApiKind::Responses)
    }

    /// Use the legacy Chat Completions API (deprecated by `OpenAI`).
    #[must_use]
    pub fn with_chat_completions_api(self) -> Self {
        self.with_api(ApiKind::ChatCompletions)
    }

    /// Send deprecated `max_tokens` alongside `max_completion_tokens` for compatibility.
    ///
    /// `OpenAI` deprecates `max_tokens` and it is incompatible with reasoning models.
    #[must_use]
    pub fn with_legacy_max_tokens(mut self, enabled: bool) -> Self {
        Arc::make_mut(&mut self.inner).legacy_max_tokens = enabled;
        self
    }

    /// Override the embeddings model identifier.
    #[must_use]
    pub fn with_embedding_model(mut self, model: impl Into<String>) -> Self {
        let model = sanitize_model(model);
        let cfg = Arc::make_mut(&mut self.inner);
        cfg.embedding_model = model;
        if let Some(dim) = infer_embedding_dim(&cfg.embedding_model) {
            cfg.embedding_dimensions = dim;
        }
        self
    }

    /// Override the embedding dimension (defaults depend on model).
    #[must_use]
    pub fn with_embedding_dimensions(mut self, dimensions: usize) -> Self {
        Arc::make_mut(&mut self.inner).embedding_dimensions = dimensions;
        self
    }

    /// Override the image generation model identifier.
    #[must_use]
    pub fn with_image_model(mut self, model: impl Into<String>) -> Self {
        Arc::make_mut(&mut self.inner).image_model = sanitize_model(model);
        self
    }

    /// Override the default text-to-speech model identifier.
    #[must_use]
    pub fn with_audio_model(mut self, model: impl Into<String>) -> Self {
        Arc::make_mut(&mut self.inner).audio_model = sanitize_model(model);
        self
    }

    /// Set the default text-to-speech voice (e.g., `alloy`).
    #[must_use]
    pub fn with_audio_voice(mut self, voice: impl Into<String>) -> Self {
        Arc::make_mut(&mut self.inner).audio_voice = voice.into();
        self
    }

    /// Set the preferred audio format (`mp3`, `wav`, `flac`).
    #[must_use]
    pub fn with_audio_format(mut self, format: impl Into<String>) -> Self {
        Arc::make_mut(&mut self.inner).audio_format = format.into();
        self
    }

    /// Override the transcription model identifier.
    #[must_use]
    pub fn with_transcription_model(mut self, model: impl Into<String>) -> Self {
        Arc::make_mut(&mut self.inner).transcription_model = sanitize_model(model);
        self
    }

    /// Override the moderation model identifier.
    #[must_use]
    pub fn with_moderation_model(mut self, model: impl Into<String>) -> Self {
        Arc::make_mut(&mut self.inner).moderation_model = sanitize_model(model);
        self
    }

    pub(crate) fn config(&self) -> Arc<Config> {
        self.inner.clone()
    }
}

impl LanguageModel for OpenAI {
    type Error = OpenAIError;

    #[allow(clippy::too_many_lines)]
    fn respond(
        &self,
        request: LLMRequest,
    ) -> impl Stream<Item = Result<Event, Self::Error>> + Send {
        let cfg = self.inner.clone();
        let (messages, parameters, tool_defs) = request.into_parts();
        let tool_defs = filter_tool_definitions(tool_defs, &parameters.tool_choice);
        let mut snapshot = ParameterSnapshot::from(&parameters);
        snapshot.legacy_max_tokens = cfg.legacy_max_tokens;
        let has_attachments = messages.iter().any(|msg| !msg.attachments().is_empty());
        let has_audio = messages.iter().any(|message| {
            message
                .attachments()
                .iter()
                .any(|attachment| attachment.media_type().as_ref().starts_with("audio/"))
        });
        let effective_api_kind = if cfg.api_kind == ApiKind::Responses && has_audio {
            ApiKind::ChatCompletions
        } else {
            cfg.api_kind
        };

        async_stream::stream! {
            if parameters.cache.claude.is_some() || parameters.cache.gemini.is_some() {
                yield Err(OpenAIError::Api(
                    "OpenAI provider only accepts cache.openai settings".to_string(),
                ));
                return;
            }

            if effective_api_kind == ApiKind::ChatCompletions
                && messages.iter().flat_map(aither_core::llm::Message::attachments).any(|attachment| {
                    let media_type = attachment.media_type().as_ref();
                    !media_type.starts_with("image/") && !media_type.starts_with("audio/")
                })
            {
                yield Err(OpenAIError::Api(
                    "Chat Completions accepts only image and audio attachments".to_string(),
                ));
                return;
            }

            if effective_api_kind == ApiKind::Responses
                && messages.iter().flat_map(aither_core::llm::Message::attachments).any(|attachment| {
                    attachment.media_type().as_ref().starts_with("video/")
                })
            {
                yield Err(OpenAIError::Api(
                    "OpenAI Responses API does not support video attachments".to_string(),
                ));
                return;
            }

            if effective_api_kind == ApiKind::ChatCompletions
                && (!snapshot.openai_tools.is_empty() || snapshot.websearch || snapshot.code_execution)
            {
                yield Err(OpenAIError::Api(
                    "OpenAI native hosted tools require the Responses API".to_string(),
                ));
                return;
            }

            let messages = if has_attachments && effective_api_kind == ApiKind::Responses {
                #[cfg(not(target_arch = "wasm32"))]
                {
                    match crate::attachments::resolve_messages(&cfg, messages).await {
                        Ok(resolved) => resolved,
                        Err(err) => {
                            yield Err(err);
                            return;
                        }
                    }
                }
                #[cfg(target_arch = "wasm32")]
                {
                    yield Err(OpenAIError::Api(
                        "file:// attachments are not supported on wasm32 targets".to_string(),
                    ));
                    return;
                }
            } else {
                messages
            };

            match effective_api_kind {
                ApiKind::ChatCompletions => {
                    let payload_messages = match to_chat_messages(&messages).await {
                        Ok(payload) => payload,
                        Err(error) => {
                            yield Err(error);
                            return;
                        }
                    };
                    let openai_tools = if tool_defs.is_empty() {
                        None
                    } else {
                        Some(convert_tools(tool_defs))
                    };

                    let mut events = chat_completions_stream(
                        cfg,
                        payload_messages,
                        snapshot,
                        openai_tools,
                    );

                    while let Some(event) = events.next().await {
                        yield event;
                    }
                }
                ApiKind::Responses => {
                    let responses_input = match to_responses_input(&messages) {
                        Ok(input) => input,
                        Err(err) => {
                            yield Err(err);
                            return;
                        }
                    };

                    let mut events = responses_stream(
                        cfg,
                        responses_input,
                        snapshot,
                        tool_defs,
                    );

                    while let Some(event) = events.next().await {
                        yield event;
                    }
                }
            }
        }
    }

    fn complete(&self, prefix: &str) -> impl Stream<Item = Result<Event, Self::Error>> + Send {
        self.respond(oneshot(
            "Continue the user provided text without additional commentary.",
            prefix,
        ))
    }

    fn profile(&self) -> impl Future<Output = Result<ModelProfile, Self::Error>> + Send {
        let cfg = self.inner.clone();
        async move {
            // Try to fetch context window from API, fallback to models database
            let context_length = match fetch_model_context_length(&cfg).await {
                Ok(len) => len,
                Err(e) => {
                    tracing::debug!("API did not return context_length: {e}");
                    aither_models::lookup(&cfg.chat_model)
                        .and_then(aither_models::ModelEntry::max_input_tokens)
                        .ok_or_else(|| {
                            OpenAIError::Api(format!(
                                "cannot determine context length for model '{model}': \
                                 {models_url} request failed ({e}) and aither-models has no entry",
                                model = cfg.chat_model,
                                models_url = cfg.request_url("/models"),
                            ))
                        })?
                }
            };

            let mut profile = ModelProfile::new(
                cfg.chat_model.clone(),
                "OpenAI",
                cfg.chat_model.clone(),
                "OpenAI GPT family model",
                context_length,
            )
            .with_ability(Ability::ToolUse);
            for ability in &cfg.native_abilities {
                if !profile.abilities.contains(ability) {
                    profile.abilities.push(*ability);
                }
            }
            Ok(profile)
        }
    }
}

/// Fetch context window size from the models API.
async fn fetch_model_context_length(cfg: &Config) -> Result<u32, OpenAIError> {
    use crate::response::ModelsListResponse;

    let url = cfg.request_url("/models");
    let mut backend = client();
    let mut builder = backend
        .get(&url)
        .map_err(OpenAIError::Http)?
        .header(
            header::AUTHORIZATION.as_str(),
            format!("Bearer {}", cfg.api_key),
        )
        .map_err(OpenAIError::Http)?;
    for (name, value) in &cfg.extra_headers {
        builder = builder
            .header(name.clone(), value.clone())
            .map_err(OpenAIError::Http)?;
    }
    let response: ModelsListResponse =
        request_with_timeout(cfg.request_timeout, builder.json()).await?;

    // Find matching model
    let mut model_found = false;
    for model in response.data {
        if model.id == cfg.chat_model {
            model_found = true;
            if let Some(ctx) = model.context_length.or(model.max_tokens) {
                tracing::debug!(
                    "Fetched context_length={} for model '{}'",
                    ctx,
                    cfg.chat_model
                );
                return Ok(ctx);
            }
        }
    }

    // More specific error message
    if model_found {
        Err(OpenAIError::Api(format!(
            "Model '{}' found but missing context_length field (proxy may not support it)",
            cfg.chat_model
        )))
    } else {
        Err(OpenAIError::Api(format!(
            "Model '{}' not found in /models response",
            cfg.chat_model
        )))
    }
}

fn chat_completions_stream(
    cfg: Arc<Config>,
    payload_messages: Vec<ChatMessagePayload>,
    snapshot: ParameterSnapshot,
    openai_tools: Option<Vec<ToolPayload>>,
) -> impl Stream<Item = Result<Event, OpenAIError>> + Send + Unpin {
    Box::pin(chat_completions_stream_inner(
        cfg,
        payload_messages,
        snapshot,
        openai_tools,
    ))
}

/// Accumulated tool call state for streaming.
#[derive(Debug, Default)]
struct ToolCallAccumulator {
    id: Option<String>,
    name: Option<String>,
    arguments: String,
    /// Bytes of `arguments` already emitted as `ToolCallDelta` fragments.
    emitted: usize,
    /// Whether the first `ToolCallDelta` announcing id and name went out.
    announced: bool,
}

/// Merge one `delta.tool_calls` entry into the accumulator for its index.
///
/// Compatible servers are expected to omit `id`/`name` on continuation
/// deltas; some instead send them again as empty strings. Empty values are
/// not authoritative — they must not overwrite the id or name captured
/// from the first delta.
fn accumulate_tool_call_delta<'a>(
    tool_calls: &'a mut HashMap<usize, ToolCallAccumulator>,
    call: &DeltaToolCall,
) -> &'a mut ToolCallAccumulator {
    let index = call.index.unwrap_or(0);
    let acc = tool_calls.entry(index).or_default();
    if let Some(id) = &call.id
        && !id.is_empty()
    {
        acc.id = Some(id.clone());
    }
    if let Some(function) = &call.function {
        if let Some(name) = &function.name
            && !name.is_empty()
        {
            acc.name = Some(name.clone());
        }
        if let Some(args) = &function.arguments {
            acc.arguments.push_str(args);
        }
    }
    acc
}

/// Make a chat completions SSE request (single attempt).
async fn chat_completions_request(
    cfg: &Config,
    request: &ChatCompletionRequest,
) -> SseStreamResult {
    let endpoint = cfg.request_url("/chat/completions");
    let mut backend = client();

    let build_result = backend
        .post(endpoint)
        .and_then(|b| b.header(header::AUTHORIZATION.as_str(), cfg.request_auth()))
        .and_then(|b| b.header(header::USER_AGENT.as_str(), "aither-openai/0.1"))
        .and_then(|b| b.header(header::ACCEPT.as_str(), "text/event-stream"));

    let mut builder = build_result.map_err(OpenAIError::Http)?;

    if let Some(org) = &cfg.organization {
        builder = builder
            .header("OpenAI-Organization", org.clone())
            .map_err(OpenAIError::Http)?;
    }
    for (name, value) in &cfg.extra_headers {
        builder = builder
            .header(name.clone(), value.clone())
            .map_err(OpenAIError::Http)?;
    }

    request_with_timeout(
        cfg.request_timeout,
        builder.json_body(request).map_err(OpenAIError::Http)?.sse(),
    )
    .await
}

#[allow(clippy::too_many_lines)]
fn chat_completions_stream_inner(
    cfg: Arc<Config>,
    payload_messages: Vec<ChatMessagePayload>,
    snapshot: ParameterSnapshot,
    openai_tools: Option<Vec<ToolPayload>>,
) -> impl Stream<Item = Result<Event, OpenAIError>> + Send {
    let include_reasoning = snapshot.include_reasoning;

    async_stream::stream! {
        let request = ChatCompletionRequest::new(
            cfg.chat_model.clone(),
            payload_messages,
            &snapshot,
            openai_tools,
            true,
        );

        tracing::debug!(request = %serde_json::to_string_pretty(&request).unwrap_or_default(), "Sending chat completion request");

        // Make request with retry
        let sse_stream = match sse_request_with_retry(&cfg, || chat_completions_request(&cfg, &request)).await {
            Ok(stream) => stream,
            Err(e) => {
                yield Err(e);
                return;
            }
        };
        futures_lite::pin!(sse_stream);

        // Accumulate tool calls by index - streaming sends id/name first, then arguments incrementally
        let mut tool_calls: std::collections::HashMap<usize, ToolCallAccumulator> =
            std::collections::HashMap::new();
        let mut usage: Option<Usage> = None;
        while let Some(event) = sse_stream.next().await {
            match event {
                Ok(e) => {
                    if should_skip_event(&e) {
                        continue;
                    }
                    let data = e.text_data();
                    if data == "[DONE]" {
                        continue;
                    }
                    tracing::debug!(sse_event = %data, "Received SSE event");

                    // Check for API error response
                    if let Ok(error_obj) = serde_json::from_str::<serde_json::Value>(data)
                        && let Some(error) = error_obj.get("error") {
                            let msg = error.get("message")
                                .and_then(|m| m.as_str())
                                .unwrap_or("Unknown API error");
                            yield Err(OpenAIError::Api(msg.to_string()));
                            return;
                        }

                    match serde_json::from_str::<ChatCompletionChunk>(data) {
                        Ok(chunk) => {
                            if let Some(chunk_usage) = &chunk.usage {
                                usage = Some(usage_from_chat_completion(chunk_usage));
                            }
                            // Emit text events
                            for choice in &chunk.choices {
                                // Check for malformed function call
                                if let Some(ref reason) = choice.finish_reason
                                    && reason.contains("malformed_function_call") {
                                        yield Err(OpenAIError::Api("malformed function call".to_string()));
                                        return;
                                    }

                                if let Some(content) = &choice.delta.content
                                    && !content.is_empty() {
                                        yield Ok(Event::Text(content.clone()));
                                    }
                                // Emit reasoning if enabled
                                if include_reasoning
                                    && let Some(reasoning) = &choice.delta.reasoning_content
                                        && !reasoning.is_empty() {
                                            yield Ok(Event::Reasoning(reasoning.clone()));
                                        }
                                // Accumulate tool calls and emit deltas
                                if let Some(calls) = &choice.delta.tool_calls {
                                    for call in calls {
                                        let acc = accumulate_tool_call_delta(&mut tool_calls, call);
                                        // Emit incremental fragments for UI
                                        // progress; the first emission may
                                        // precede any argument bytes, so the
                                        // tool name surfaces as soon as both
                                        // id and name are known.
                                        if let (Some(id), Some(name)) = (&acc.id, &acc.name) {
                                            let fragment =
                                                acc.arguments[acc.emitted..].to_string();
                                            if !fragment.is_empty() || !acc.announced {
                                                acc.emitted = acc.arguments.len();
                                                acc.announced = true;
                                                yield Ok(Event::ToolCallDelta {
                                                    id: id.clone(),
                                                    name: name.clone(),
                                                    arguments_fragment: fragment,
                                                });
                                            }
                                        }
                                    }
                                }
                            }
                        }
                        Err(e) => {
                            yield Err(OpenAIError::from(e));
                            return;
                        }
                    }
                }
                Err(e) => {
                    yield Err(OpenAIError::from(e));
                    return;
                }
            }
        }

        // Emit accumulated tool calls at the end
        let mut sorted_calls: Vec<_> = tool_calls.into_iter().collect();
        sorted_calls.sort_by_key(|(index, _)| *index);
        for (_, acc) in sorted_calls {
            if let (Some(id), Some(name)) = (acc.id, acc.name) {
                let arguments = if acc.arguments.is_empty() {
                    serde_json::Value::Object(serde_json::Map::default())
                } else {
                    serde_json::from_str(&acc.arguments).unwrap_or_else(|_| {
                        serde_json::Value::Object(serde_json::Map::default())
                    })
                };
                yield Ok(Event::ToolCall(ToolCall {
                    id,
                    name,
                    arguments,
                    reasoning_state: None,
                }));
            }
        }
        if let Some(final_usage) = usage {
            yield Ok(Event::Usage(final_usage));
        }
    }
}

fn responses_stream(
    cfg: Arc<Config>,
    input: Vec<ResponsesInputItem>,
    snapshot: ParameterSnapshot,
    tool_defs: Vec<aither_core::llm::tool::ToolDefinition>,
) -> impl Stream<Item = Result<Event, OpenAIError>> + Send + Unpin {
    Box::pin(responses_stream_inner(cfg, input, snapshot, tool_defs))
}

/// Accumulated function call state for Responses API streaming.
#[derive(Debug, Default)]
struct FunctionCallAccumulator {
    item_id: Option<String>,
    call_id: Option<String>,
    name: Option<String>,
    arguments: String,
    emitted: bool,
}

fn normalize_response_call_id(call_id: Option<String>, fallback: &str) -> String {
    match call_id {
        Some(call_id) if !call_id.trim().is_empty() => call_id,
        _ => fallback.to_string(),
    }
}

fn non_empty_response_id(value: Option<String>) -> Option<String> {
    value.filter(|value| !value.trim().is_empty())
}

fn fallback_response_item_id(output_index: usize) -> String {
    format!("response_output_{output_index}")
}

fn parse_tool_call_arguments(arguments: &str) -> serde_json::Value {
    if arguments.is_empty() {
        serde_json::Value::Object(serde_json::Map::default())
    } else {
        serde_json::from_str(arguments)
            .unwrap_or_else(|_| serde_json::Value::Object(serde_json::Map::default()))
    }
}

fn usage_from_chat_completion(raw: &ChatCompletionUsage) -> Usage {
    Usage {
        prompt_tokens: raw.prompt_tokens,
        completion_tokens: raw.completion_tokens,
        total_tokens: raw.total_tokens,
        reasoning_tokens: raw
            .completion_tokens_details
            .as_ref()
            .and_then(|details| details.reasoning_tokens),
        cache_read_tokens: raw
            .prompt_tokens_details
            .as_ref()
            .and_then(|details| details.cached_tokens),
        cache_write_tokens: None,
        cost_usd: None,
        stop_reason: None,
    }
}

fn usage_from_responses(raw: &ResponsesUsage) -> Usage {
    Usage {
        prompt_tokens: raw.input_tokens,
        completion_tokens: raw.output_tokens,
        total_tokens: raw.total_tokens,
        reasoning_tokens: raw
            .output_token_details
            .as_ref()
            .and_then(|details| details.reasoning_tokens),
        cache_read_tokens: raw
            .input_token_details
            .as_ref()
            .and_then(|details| details.cached_tokens),
        cache_write_tokens: None,
        cost_usd: None,
        stop_reason: None,
    }
}

fn mark_and_emit_done_function_call(
    function_calls: &mut HashMap<usize, FunctionCallAccumulator>,
    output_index: usize,
    item_id: Option<String>,
    response_id: String,
    call_id: Option<String>,
    name: String,
    arguments: &str,
) -> ToolCall {
    let fallback_id = non_empty_response_id(item_id.clone())
        .or_else(|| non_empty_response_id(Some(response_id.clone())))
        .unwrap_or_else(|| fallback_response_item_id(output_index));
    let resolved_call_id = normalize_response_call_id(call_id, &fallback_id);

    let acc = function_calls.entry(output_index).or_default();
    acc.item_id =
        non_empty_response_id(item_id).or_else(|| non_empty_response_id(Some(response_id)));
    acc.call_id = Some(resolved_call_id.clone());
    acc.name = Some(name.clone());
    acc.arguments.clear();
    acc.arguments.push_str(arguments);
    acc.emitted = true;

    ToolCall {
        id: resolved_call_id,
        name,
        arguments: parse_tool_call_arguments(arguments),
        reasoning_state: None,
    }
}

fn drain_pending_function_calls(
    function_calls: HashMap<usize, FunctionCallAccumulator>,
) -> Vec<ToolCall> {
    function_calls
        .into_iter()
        .filter(|(_, acc)| !acc.emitted)
        .filter_map(|(output_index, acc)| {
            let name = acc.name?;
            let fallback_id = acc
                .item_id
                .clone()
                .unwrap_or_else(|| fallback_response_item_id(output_index));
            let call_id = normalize_response_call_id(acc.call_id, &fallback_id);
            Some(ToolCall {
                id: call_id,
                name,
                arguments: parse_tool_call_arguments(&acc.arguments),
                reasoning_state: None,
            })
        })
        .collect()
}

#[allow(clippy::items_after_statements)]
fn validate_serialized_responses_request(request: &ResponsesRequest) -> Result<(), OpenAIError> {
    let request_value = serde_json::to_value(request).map_err(OpenAIError::Json)?;
    let Some(items) = request_value
        .get("input")
        .and_then(serde_json::Value::as_array)
    else {
        return Ok(());
    };

    fn find_blank_call_id(value: &serde_json::Value, path: &str) -> Option<(String, String)> {
        match value {
            serde_json::Value::Object(map) => {
                for (key, child) in map {
                    let child_path = format!("{path}.{key}");
                    if key == "call_id"
                        && let Some(call_id) = child.as_str()
                        && call_id.trim().is_empty()
                    {
                        return Some((child_path, call_id.to_string()));
                    }
                    if let Some(found) = find_blank_call_id(child, &child_path) {
                        return Some(found);
                    }
                }
                None
            }
            serde_json::Value::Array(items) => {
                for (index, child) in items.iter().enumerate() {
                    let child_path = format!("{path}[{index}]");
                    if let Some(found) = find_blank_call_id(child, &child_path) {
                        return Some(found);
                    }
                }
                None
            }
            serde_json::Value::Null
            | serde_json::Value::Bool(_)
            | serde_json::Value::Number(_)
            | serde_json::Value::String(_) => None,
        }
    }

    for (index, item) in items.iter().enumerate() {
        if let Some((path, call_id)) = find_blank_call_id(item, &format!("input[{index}]")) {
            let snippet = serde_json::to_string(item).map_err(OpenAIError::Json)?;
            return Err(OpenAIError::Api(format!(
                "responses request serialized blank call_id at {path}: {call_id:?}; item={snippet}"
            )));
        }
    }
    Ok(())
}

/// Make a responses API SSE request (single attempt).
async fn responses_request(cfg: &Config, request: &ResponsesRequest) -> SseStreamResult {
    validate_serialized_responses_request(request)?;
    let serialized_request = serde_json::to_string(request).map_err(OpenAIError::Json)?;
    let endpoint = cfg.request_url("/responses");
    let mut backend = client();

    let build_result = backend
        .post(endpoint)
        .and_then(|b| b.header(header::AUTHORIZATION.as_str(), cfg.request_auth()))
        .and_then(|b| b.header(header::USER_AGENT.as_str(), "aither-openai/0.1"))
        .and_then(|b| b.header(header::ACCEPT.as_str(), "text/event-stream"));

    let mut builder = build_result.map_err(OpenAIError::Http)?;

    if let Some(org) = &cfg.organization {
        builder = builder
            .header("OpenAI-Organization", org.clone())
            .map_err(OpenAIError::Http)?;
    }
    for (name, value) in &cfg.extra_headers {
        builder = builder
            .header(name.clone(), value.clone())
            .map_err(OpenAIError::Http)?;
    }

    match request_with_timeout(
        cfg.request_timeout,
        builder.json_body(request).map_err(OpenAIError::Http)?.sse(),
    )
    .await
    {
        Ok(stream) => Ok(stream),
        Err(error) => {
            if error.to_string().contains("call_id") {
                return Err(OpenAIError::Api(format!(
                    "{error}\nSerialized responses request: {serialized_request}"
                )));
            }
            Err(error)
        }
    }
}

#[allow(clippy::too_many_lines, clippy::collapsible_match)]
fn responses_stream_inner(
    cfg: Arc<Config>,
    input: Vec<ResponsesInputItem>,
    snapshot: ParameterSnapshot,
    tool_defs: Vec<aither_core::llm::tool::ToolDefinition>,
) -> impl Stream<Item = Result<Event, OpenAIError>> + Send {
    let include_reasoning = snapshot.include_reasoning;

    async_stream::stream! {
        let mut response_tools = convert_responses_tools(tool_defs);
        let allow_builtins = matches!(snapshot.tool_choice, ToolChoice::Auto | ToolChoice::Required);
        if allow_builtins {
            let native_tools = snapshot.openai_tools.clone();
            if snapshot.websearch || native_tools.web_search.is_some() {
                let tool = native_tools.web_search.unwrap_or_default();
                response_tools.push(ResponsesTool::from(tool));
            }
            for tool in native_tools.file_search {
                response_tools.push(ResponsesTool::from(tool));
            }
            if snapshot.code_execution || native_tools.code_interpreter.is_some() {
                let tool = native_tools.code_interpreter.unwrap_or_default();
                response_tools.push(ResponsesTool::from(tool));
            }
            if let Some(tool) = native_tools.image_generation {
                response_tools.push(ResponsesTool::from(tool));
            }
            for tool in native_tools.mcp {
                response_tools.push(ResponsesTool::from(tool));
            }
            if let Some(tool) = native_tools.computer_use {
                response_tools.push(ResponsesTool::from(tool));
            }
        }
        let response_tools = if response_tools.is_empty() {
            None
        } else {
            Some(response_tools)
        };
        let has_tools = response_tools.is_some();
        let tool_choice = responses_tool_choice(&snapshot, has_tools);

        let mut request = ResponsesRequest::new(
            cfg.chat_model.clone(),
            input,
            &snapshot,
            response_tools,
            tool_choice,
            true, // stream: true
        );
        request.store = cfg.store;
        let serialized_request = serde_json::to_string(&request).unwrap_or_default();

        tracing::debug!(request = %serde_json::to_string_pretty(&request).unwrap_or_default(), "Sending responses request");

        // Make request with retry
        let sse_stream = match sse_request_with_retry(&cfg, || responses_request(&cfg, &request)).await {
            Ok(stream) => stream,
            Err(e) => {
                yield Err(e);
                return;
            }
        };
        futures_lite::pin!(sse_stream);

        // Accumulate function calls by output_index because provider item identifiers can be
        // absent or empty on some streamed events.
        let mut function_calls: HashMap<usize, FunctionCallAccumulator> = HashMap::new();
        let mut usage: Option<Usage> = None;
        let mut usage_emitted = false;

        while let Some(event) = sse_stream.next().await {
            match event {
                Ok(e) => {
                    if should_skip_event(&e) {
                        continue;
                    }
                    let data = e.text_data();
                    if data == "[DONE]" {
                        continue;
                    }
                    tracing::trace!(sse_event = %data, "Received Responses API SSE event");

                    // Check for API error response
                    if let Ok(error_obj) = serde_json::from_str::<serde_json::Value>(data)
                        && let Some(error) = error_obj.get("error") {
                            let msg = error.get("message")
                                .and_then(|m| m.as_str())
                                .unwrap_or("Unknown API error");
                            if msg.contains("call_id") {
                                yield Err(OpenAIError::Api(format!(
                                    "{msg}\nSerialized responses request: {serialized_request}"
                                )));
                            } else {
                                yield Err(OpenAIError::Api(msg.to_string()));
                            }
                            return;
                        }

                    match serde_json::from_str::<ResponsesStreamEvent>(data) {
                        Ok(stream_event) => {
                            match stream_event {
                                ResponsesStreamEvent::ResponseCreated { response } => {
                                    if let Some(meta) = response.and_then(|resp| resp.usage) {
                                        usage = Some(usage_from_responses(&meta));
                                    }
                                }
                                ResponsesStreamEvent::OutputTextDelta { delta, .. } => {
                                    if !delta.is_empty() {
                                        yield Ok(Event::Text(delta));
                                    }
                                }
                                ResponsesStreamEvent::ReasoningTextDelta { delta, .. } |
                                ResponsesStreamEvent::ReasoningSummaryTextDelta { delta, .. } => {
                                    if include_reasoning && !delta.is_empty() {
                                        yield Ok(Event::Reasoning(delta));
                                    }
                                }
                                ResponsesStreamEvent::OutputItemAdded { item, item_id, output_index } => {
                                    if let Some(event) = builtin_call_event(&item) {
                                        yield Ok(event);
                                    }
                                    // When a function_call item is added, capture id and name
                                    if let ResponsesOutputItem::FunctionCall { id, call_id, name, .. } = item {
                                        let fallback_id = non_empty_response_id(item_id.clone())
                                            .or_else(|| non_empty_response_id(Some(id.clone())))
                                            .unwrap_or_else(|| fallback_response_item_id(output_index));
                                        let acc = function_calls.entry(output_index).or_default();
                                        acc.item_id = non_empty_response_id(item_id)
                                            .or_else(|| non_empty_response_id(Some(id)));
                                        acc.call_id = Some(normalize_response_call_id(call_id, &fallback_id));
                                        acc.name = Some(name.clone());
                                        // Emit initial delta with tool name
                                        if let Some(ref call_id) = acc.call_id {
                                            yield Ok(Event::ToolCallDelta {
                                                id: call_id.clone(),
                                                name,
                                                arguments_fragment: String::new(),
                                            });
                                        }
                                    }
                                }
                                ResponsesStreamEvent::FunctionCallArgumentsDelta { delta, output_index, .. } => {
                                    let acc = function_calls.entry(output_index).or_default();
                                    acc.arguments.push_str(&delta);
                                    // Emit the wire delta verbatim for UI progress
                                    if let (Some(call_id), Some(name)) = (&acc.call_id, &acc.name) {
                                        yield Ok(Event::ToolCallDelta {
                                            id: call_id.clone(),
                                            name: name.clone(),
                                            arguments_fragment: delta,
                                        });
                                    }
                                }
                                ResponsesStreamEvent::FunctionCallArgumentsDone { arguments, output_index, .. } => {
                                    let acc = function_calls.entry(output_index).or_default();
                                    acc.arguments = arguments;
                                }
                                ResponsesStreamEvent::OutputItemDone { item, item_id, output_index } => {
                                    if let Some(event) = builtin_result_event(&item) {
                                        yield Ok(event);
                                    }
                                    // Emit function call when item is done
                                    if let ResponsesOutputItem::FunctionCall { id, call_id, name, arguments } = item {
                                        let tool_call = mark_and_emit_done_function_call(
                                            &mut function_calls,
                                            output_index,
                                            item_id,
                                            id,
                                            call_id,
                                            name,
                                            &arguments,
                                        );
                                        yield Ok(Event::ToolCall(tool_call));
                                    } else if let ResponsesOutputItem::Reasoning {
                                        id, encrypted_content, ..
                                    } = item
                                    {
                                        // Only encrypted reasoning is replayable;
                                        // a summary alone cannot be verified.
                                        if let Some(encrypted) = encrypted_content {
                                            let payload = serde_json::json!({
                                                "type": "reasoning",
                                                "id": id,
                                                "encrypted_content": encrypted,
                                            });
                                            yield Ok(Event::ReasoningState(ReasoningState::new(
                                                PROVIDER_NAME,
                                                payload.to_string(),
                                            )));
                                        }
                                    }
                                }
                                ResponsesStreamEvent::ResponseCompleted { response } => {
                                    if let Some(meta) = response.usage {
                                        let final_usage = usage_from_responses(&meta);
                                        usage = Some(final_usage.clone());
                                        yield Ok(Event::Usage(final_usage));
                                        usage_emitted = true;
                                    }
                                }
                                ResponsesStreamEvent::ResponseFailed { error } => {
                                    let msg = error
                                        .and_then(|e| e.message)
                                        .unwrap_or_else(|| "Response failed".to_string());
                                    if msg.contains("call_id") {
                                        yield Err(OpenAIError::Api(format!(
                                            "{msg}\nSerialized responses request: {serialized_request}"
                                        )));
                                    } else {
                                        yield Err(OpenAIError::Api(msg));
                                    }
                                    return;
                                }
                                ResponsesStreamEvent::Error { message, .. } => {
                                    let msg = message.unwrap_or_else(|| "Unknown error".to_string());
                                    if msg.contains("call_id") {
                                        yield Err(OpenAIError::Api(format!(
                                            "{msg}\nSerialized responses request: {serialized_request}"
                                        )));
                                    } else {
                                        yield Err(OpenAIError::Api(msg));
                                    }
                                    return;
                                }
                                // Ignore other events
                                _ => {}
                            }
                        }
                        Err(e) => {
                            tracing::warn!(error = %e, data = %data, "Failed to parse Responses API event");
                        }
                    }
                }
                Err(e) => {
                    yield Err(OpenAIError::from(e));
                    return;
                }
            }
        }

        // Emit any remaining accumulated function calls (fallback if OutputItemDone wasn't received)
        for tool_call in drain_pending_function_calls(function_calls) {
            yield Ok(Event::ToolCall(tool_call));
        }
        if !usage_emitted && let Some(final_usage) = usage {
            yield Ok(Event::Usage(final_usage));
        }
    }
}

/// Start event for a provider-executed (hosted) tool item, when the output
/// item is one we model — the provider runs it, no consumer execution.
fn builtin_call_event(item: &ResponsesOutputItem) -> Option<Event> {
    let (id, tool, detail) = match item {
        ResponsesOutputItem::WebSearchCall { id, action, .. } => (
            id.as_str(),
            "web_search",
            action
                .as_ref()
                .and_then(|action| {
                    action
                        .get("query")
                        .or_else(|| action.get("url"))
                        .and_then(serde_json::Value::as_str)
                })
                .unwrap_or_default()
                .to_string(),
        ),
        ResponsesOutputItem::FileSearchCall { id, queries, .. } => {
            (id.as_str(), "file_search", queries.join(", "))
        }
        ResponsesOutputItem::CodeInterpreterCall { id, code, .. } => (
            id.as_str(),
            "code_interpreter",
            code.clone().unwrap_or_default(),
        ),
        _ => return None,
    };
    Some(Event::builtin_call(id, tool, detail))
}

/// Completion event for a provider-executed (hosted) tool item.
fn builtin_result_event(item: &ResponsesOutputItem) -> Option<Event> {
    let (id, tool, result) = match item {
        ResponsesOutputItem::WebSearchCall { id, status, action } => {
            let detail = action.as_ref().and_then(|action| {
                action
                    .get("query")
                    .or_else(|| action.get("url"))
                    .and_then(serde_json::Value::as_str)
            });
            let result = detail.map_or_else(
                || status.clone().unwrap_or_else(|| "completed".to_string()),
                |detail| format!("\"{detail}\""),
            );
            (id.as_str(), "web_search", result)
        }
        ResponsesOutputItem::FileSearchCall {
            id,
            queries,
            results,
            status,
        } => {
            let count = results.as_ref().map_or_else(
                || status.clone().unwrap_or_else(|| "completed".to_string()),
                |results| format!("{} results", results.len()),
            );
            (
                id.as_str(),
                "file_search",
                format!("{count} for {}", queries.join(", ")),
            )
        }
        ResponsesOutputItem::CodeInterpreterCall { id, status, code } => {
            let result = code
                .clone()
                .unwrap_or_else(|| status.clone().unwrap_or_else(|| "completed".to_string()));
            (id.as_str(), "code_interpreter", result)
        }
        _ => return None,
    };
    Some(Event::builtin_result(id, tool, result))
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::response::{
        ChatCompletionTokensDetails, ChatCompletionUsage, ChatPromptTokensDetails,
        DeltaToolFunction, ResponsesInputTokenDetails, ResponsesOutputTokenDetails, ResponsesUsage,
    };
    use std::time::Duration;

    fn chat_delta(id: Option<&str>, name: Option<&str>, arguments: &str) -> DeltaToolCall {
        DeltaToolCall {
            index: Some(0),
            id: id.map(str::to_string),
            function: Some(DeltaToolFunction {
                name: name.map(str::to_string),
                arguments: Some(arguments.to_string()),
            }),
        }
    }

    #[test]
    fn chat_tool_call_accumulates_name_and_arguments() {
        let mut tool_calls = HashMap::new();
        accumulate_tool_call_delta(
            &mut tool_calls,
            &chat_delta(Some("call_1"), Some("search"), ""),
        );
        accumulate_tool_call_delta(&mut tool_calls, &chat_delta(None, None, "{\"q\":"));
        accumulate_tool_call_delta(&mut tool_calls, &chat_delta(None, None, "\"rust\"}"));

        let acc = &tool_calls[&0];
        assert_eq!(acc.id.as_deref(), Some("call_1"));
        assert_eq!(acc.name.as_deref(), Some("search"));
        assert_eq!(acc.arguments, "{\"q\":\"rust\"}");
    }

    #[test]
    fn chat_tool_call_empty_delta_fields_do_not_overwrite() {
        let mut tool_calls = HashMap::new();
        accumulate_tool_call_delta(
            &mut tool_calls,
            &chat_delta(Some("call_1"), Some("search"), ""),
        );
        // Some compatible servers repeat `id`/`name` as empty strings on
        // every continuation delta instead of omitting them.
        accumulate_tool_call_delta(&mut tool_calls, &chat_delta(Some(""), Some(""), "{\"q\":"));
        accumulate_tool_call_delta(
            &mut tool_calls,
            &chat_delta(Some(""), Some(""), "\"rust\"}"),
        );

        let acc = &tool_calls[&0];
        assert_eq!(acc.id.as_deref(), Some("call_1"));
        assert_eq!(acc.name.as_deref(), Some("search"));
        assert_eq!(acc.arguments, "{\"q\":\"rust\"}");
    }

    #[test]
    fn done_function_call_is_not_emitted_twice() {
        let mut function_calls = HashMap::new();
        let emitted = mark_and_emit_done_function_call(
            &mut function_calls,
            1,
            Some("item_1".to_string()),
            "item_1".to_string(),
            Some("call_1".to_string()),
            "search".to_string(),
            r#"{"q":"rust"}"#,
        );
        assert_eq!(emitted.id, "call_1");
        assert_eq!(emitted.name, "search");
        assert_eq!(drain_pending_function_calls(function_calls).len(), 0);
    }

    #[test]
    fn pending_function_call_is_emitted_once_via_fallback() {
        let mut function_calls = HashMap::new();
        function_calls.insert(
            2,
            FunctionCallAccumulator {
                item_id: Some("item_2".to_string()),
                call_id: Some("call_2".to_string()),
                name: Some("lookup".to_string()),
                arguments: r#"{"id":42}"#.to_string(),
                emitted: false,
            },
        );
        let pending = drain_pending_function_calls(function_calls);
        assert_eq!(pending.len(), 1);
        assert_eq!(pending[0].id, "call_2");
        assert_eq!(pending[0].name, "lookup");
    }

    #[test]
    fn empty_done_call_id_falls_back_to_item_id() {
        let mut function_calls = HashMap::new();
        let emitted = mark_and_emit_done_function_call(
            &mut function_calls,
            3,
            Some("item_3".to_string()),
            "item_3".to_string(),
            Some(String::new()),
            "search".to_string(),
            r#"{"q":"terminal first"}"#,
        );
        assert_eq!(emitted.id, "item_3");
    }

    #[test]
    fn empty_pending_call_id_falls_back_to_item_id() {
        let mut function_calls = HashMap::new();
        function_calls.insert(
            4,
            FunctionCallAccumulator {
                item_id: Some("item_4".to_string()),
                call_id: Some("   ".to_string()),
                name: Some("lookup".to_string()),
                arguments: r#"{"id":7}"#.to_string(),
                emitted: false,
            },
        );
        let pending = drain_pending_function_calls(function_calls);
        assert_eq!(pending.len(), 1);
        assert_eq!(pending[0].id, "item_4");
        assert_eq!(pending[0].name, "lookup");
    }

    #[test]
    fn empty_stream_identifiers_fall_back_to_output_index() {
        let mut function_calls = HashMap::new();
        function_calls.insert(
            12,
            FunctionCallAccumulator {
                item_id: None,
                call_id: None,
                name: None,
                arguments: r#"{"q":"may"}"#.to_string(),
                emitted: false,
            },
        );
        let emitted = mark_and_emit_done_function_call(
            &mut function_calls,
            12,
            None,
            String::new(),
            Some(String::new()),
            "search".to_string(),
            r#"{"q":"may"}"#,
        );
        assert_eq!(emitted.id, "response_output_12");
    }

    #[test]
    fn serialized_responses_request_rejects_empty_call_id() {
        let snapshot = ParameterSnapshot::from(&aither_core::llm::model::Parameters::default());
        let request = ResponsesRequest::new(
            "gpt-5.4".to_string(),
            vec![ResponsesInputItem::FunctionCall {
                kind: "function_call",
                call_id: String::new(),
                name: "search".to_string(),
                arguments: "{}".to_string(),
            }],
            &snapshot,
            None,
            responses_tool_choice(&snapshot, false),
            false,
        );

        let error = validate_serialized_responses_request(&request)
            .expect_err("empty serialized call_id must fail");
        assert!(
            error.to_string().contains("input[0]"),
            "unexpected error: {error}"
        );
    }

    #[test]
    fn serialized_responses_request_rejects_whitespace_call_id() {
        let snapshot = ParameterSnapshot::from(&aither_core::llm::model::Parameters::default());
        let request = ResponsesRequest::new(
            "gpt-5.4".to_string(),
            vec![ResponsesInputItem::FunctionCallOutput {
                kind: "function_call_output",
                call_id: "   ".to_string(),
                output: "{}".to_string(),
            }],
            &snapshot,
            None,
            responses_tool_choice(&snapshot, false),
            false,
        );

        let error = validate_serialized_responses_request(&request)
            .expect_err("whitespace serialized call_id must fail");
        assert!(
            error.to_string().contains("blank call_id"),
            "unexpected error: {error}"
        );
    }

    /// Serves a single fixed HTTP response on a loopback address.
    fn serve_once(body: &'static str) -> String {
        use std::io::{Read, Write};

        let listener = std::net::TcpListener::bind("127.0.0.1:0").expect("bind stub server");
        let addr = listener.local_addr().expect("stub server address");
        std::thread::spawn(move || {
            let Ok((mut stream, _)) = listener.accept() else {
                return;
            };
            let mut request = Vec::new();
            let mut chunk = [0u8; 4096];
            // Read until the request headers are complete; the stub has no body to consume.
            while !request.windows(4).any(|w| w == b"\r\n\r\n") {
                let Ok(n) = stream.read(&mut chunk) else {
                    return;
                };
                if n == 0 {
                    return;
                }
                request.extend_from_slice(&chunk[..n]);
            }
            let response = format!(
                "HTTP/1.1 200 OK\r\ncontent-type: application/json\r\ncontent-length: {}\r\nconnection: close\r\n\r\n{body}",
                body.len(),
            );
            let _ = stream.write_all(response.as_bytes());
            let _ = stream.flush();
        });
        format!("http://{addr}")
    }

    #[tokio::test]
    async fn profile_errors_when_context_length_is_unknown() {
        // aither#181: a /models listing without context_length, for a model
        // the aither-models registry does not know, must surface an error
        // instead of silently assuming a context window.
        let base_url = serve_once(
            r#"{"object":"list","data":[{"id":"unregistered-stub-model","object":"model","created":1,"owned_by":"stub"}]}"#,
        );
        let model = OpenAI::builder("test-key")
            .base_url(base_url)
            .model("unregistered-stub-model")
            .build();

        let err = model
            .profile()
            .await
            .expect_err("profile must fail when the context window is unknown");
        let message = err.to_string();
        assert!(
            message.contains("unregistered-stub-model"),
            "error must name the model: {message}"
        );
        assert!(
            message.contains("/models"),
            "error must name the endpoint that was asked: {message}"
        );
    }

    #[tokio::test]
    async fn configured_timeout_is_enforced_for_sse_requests() {
        let timeout = request_with_timeout(
            Duration::from_millis(20),
            futures_lite::future::pending::<Result<(), zenwave::Error>>(),
        )
        .await;
        assert!(matches!(timeout, Err(OpenAIError::Timeout)));
    }

    #[test]
    fn chat_usage_mapping_keeps_reasoning_and_cache_tokens() {
        let usage = ChatCompletionUsage {
            prompt_tokens: Some(11),
            completion_tokens: Some(7),
            total_tokens: Some(18),
            prompt_tokens_details: Some(ChatPromptTokensDetails {
                cached_tokens: Some(3),
            }),
            completion_tokens_details: Some(ChatCompletionTokensDetails {
                reasoning_tokens: Some(2),
            }),
        };
        let mapped = usage_from_chat_completion(&usage);
        assert_eq!(mapped.prompt_tokens, Some(11));
        assert_eq!(mapped.completion_tokens, Some(7));
        assert_eq!(mapped.total_tokens, Some(18));
        assert_eq!(mapped.reasoning_tokens, Some(2));
        assert_eq!(mapped.cache_read_tokens, Some(3));
    }

    #[test]
    fn responses_usage_mapping_keeps_reasoning_and_cache_tokens() {
        let usage = ResponsesUsage {
            input_tokens: Some(19),
            output_tokens: Some(5),
            total_tokens: Some(24),
            input_token_details: Some(ResponsesInputTokenDetails {
                cached_tokens: Some(4),
            }),
            output_token_details: Some(ResponsesOutputTokenDetails {
                reasoning_tokens: Some(1),
            }),
        };
        let mapped = usage_from_responses(&usage);
        assert_eq!(mapped.prompt_tokens, Some(19));
        assert_eq!(mapped.completion_tokens, Some(5));
        assert_eq!(mapped.total_tokens, Some(24));
        assert_eq!(mapped.reasoning_tokens, Some(1));
        assert_eq!(mapped.cache_read_tokens, Some(4));
    }
}

fn filter_tool_definitions(
    defs: Vec<aither_core::llm::tool::ToolDefinition>,
    choice: &ToolChoice,
) -> Vec<aither_core::llm::tool::ToolDefinition> {
    match choice {
        ToolChoice::None => Vec::new(),
        ToolChoice::Exact(name) => defs
            .into_iter()
            .filter(|tool| tool.name() == name)
            .collect(),
        ToolChoice::Auto | ToolChoice::Required => defs,
    }
}

/// Builder for [`OpenAI`] clients.
#[derive(Debug)]
pub struct Builder {
    api_key: String,
    base_url: String,
    api_kind: ApiKind,
    chat_model: String,
    embedding_model: String,
    embedding_dimensions: usize,
    image_model: String,
    audio_model: String,
    audio_voice: String,
    audio_format: String,
    transcription_model: String,
    moderation_model: String,
    legacy_max_tokens: bool,
    organization: Option<String>,
    native_abilities: Vec<Ability>,
    retry: RetryConfig,
    request_timeout: Duration,
    extra_headers: Vec<(String, String)>,
    store: Option<bool>,
}

/// Default request timeout (5 minutes - generous for long completions).
const DEFAULT_REQUEST_TIMEOUT: Duration = Duration::from_secs(300);

impl Builder {
    fn new(api_key: impl Into<String>) -> Self {
        Self {
            api_key: api_key.into(),
            base_url: DEFAULT_BASE_URL.to_string(),
            api_kind: ApiKind::Responses,
            chat_model: DEFAULT_MODEL.to_string(),
            embedding_model: DEFAULT_EMBEDDING_MODEL.to_string(),
            embedding_dimensions: DEFAULT_EMBEDDING_DIM,
            image_model: DEFAULT_IMAGE_MODEL.to_string(),
            audio_model: DEFAULT_AUDIO_MODEL.to_string(),
            audio_voice: DEFAULT_AUDIO_VOICE.to_string(),
            audio_format: DEFAULT_AUDIO_FORMAT.to_string(),
            transcription_model: DEFAULT_TRANSCRIPTION_MODEL.to_string(),
            moderation_model: DEFAULT_MODERATION_MODEL.to_string(),
            legacy_max_tokens: false,
            organization: None,
            native_abilities: Vec::new(),
            retry: RetryConfig::default(),
            request_timeout: DEFAULT_REQUEST_TIMEOUT,
            extra_headers: Vec::new(),
            store: None,
        }
    }

    /// Set a custom API base URL.
    #[must_use]
    pub fn base_url(mut self, url: impl Into<String>) -> Self {
        self.base_url = url.into();
        self
    }

    /// Use the recommended Responses API.
    #[must_use]
    pub const fn use_responses_api(mut self) -> Self {
        self.api_kind = ApiKind::Responses;
        self
    }

    /// Use the legacy Chat Completions API (deprecated by `OpenAI`).
    #[must_use]
    pub const fn use_chat_completions_api(mut self) -> Self {
        self.api_kind = ApiKind::ChatCompletions;
        self
    }

    /// Send deprecated `max_tokens` alongside `max_completion_tokens` for compatibility.
    ///
    /// `OpenAI` deprecates `max_tokens` and it is incompatible with reasoning models.
    #[must_use]
    pub const fn legacy_max_tokens(mut self, enabled: bool) -> Self {
        self.legacy_max_tokens = enabled;
        self
    }

    /// Select a model identifier (e.g., `gpt-4o-mini`, `o1-mini`).
    #[must_use]
    pub fn model(mut self, model: impl Into<String>) -> Self {
        self.chat_model = sanitize_model(model);
        self
    }

    /// Select the embeddings model identifier.
    #[must_use]
    pub fn embedding_model(mut self, model: impl Into<String>) -> Self {
        let model = sanitize_model(model);
        if let Some(dim) = infer_embedding_dim(&model) {
            self.embedding_dimensions = dim;
        }
        self.embedding_model = model;
        self
    }

    /// Override the embedding vector dimension.
    #[must_use]
    pub const fn embedding_dimensions(mut self, dimensions: usize) -> Self {
        self.embedding_dimensions = dimensions;
        self
    }

    /// Select the image model identifier.
    #[must_use]
    pub fn image_model(mut self, model: impl Into<String>) -> Self {
        self.image_model = sanitize_model(model);
        self
    }

    /// Select the default audio generation model.
    #[must_use]
    pub fn audio_model(mut self, model: impl Into<String>) -> Self {
        self.audio_model = sanitize_model(model);
        self
    }

    /// Override the default TTS voice.
    #[must_use]
    pub fn audio_voice(mut self, voice: impl Into<String>) -> Self {
        self.audio_voice = voice.into();
        self
    }

    /// Override the synthesized audio format.
    #[must_use]
    pub fn audio_format(mut self, format: impl Into<String>) -> Self {
        self.audio_format = format.into();
        self
    }

    /// Select the transcription model identifier.
    #[must_use]
    pub fn transcription_model(mut self, model: impl Into<String>) -> Self {
        self.transcription_model = sanitize_model(model);
        self
    }

    /// Select the moderation model identifier.
    #[must_use]
    pub fn moderation_model(mut self, model: impl Into<String>) -> Self {
        self.moderation_model = sanitize_model(model);
        self
    }

    /// Attach an `OpenAI` organization header.
    #[must_use]
    pub fn organization(mut self, organization: impl Into<String>) -> Self {
        self.organization = Some(organization.into());
        self
    }

    /// Attach an extra header sent with every request.
    ///
    /// Used by endpoints that authenticate or route on custom headers (for
    /// example the `ChatGPT` Codex backend's `ChatGPT-Account-Id`). Headers set
    /// here are sent alongside the built-in `Authorization` header.
    #[must_use]
    pub fn extra_header(mut self, name: impl Into<String>, value: impl Into<String>) -> Self {
        self.extra_headers.push((name.into(), value.into()));
        self
    }

    /// Set the Responses API `store` field.
    ///
    /// `store: false` disables server-side response persistence. Some
    /// backends (the `ChatGPT` Codex endpoint) require it to be present and
    /// false; the default `None` omits the field entirely.
    #[must_use]
    pub const fn store(mut self, store: bool) -> Self {
        self.store = Some(store);
        self
    }

    /// Declare extra native capabilities (e.g., web search, PDF understanding) supported by the upstream model.
    #[must_use]
    pub fn native_capabilities(mut self, abilities: impl IntoIterator<Item = Ability>) -> Self {
        for ability in abilities {
            if !self.native_abilities.contains(&ability) {
                self.native_abilities.push(ability);
            }
        }
        self
    }

    /// Mark this model as having built-in web search support.
    #[must_use]
    pub fn enable_native_web_search(self) -> Self {
        self.native_capabilities([Ability::WebSearch])
    }

    /// Mark this model as having native PDF/document understanding.
    #[must_use]
    pub fn enable_native_pdf(self) -> Self {
        self.native_capabilities([Ability::Pdf])
    }

    /// Configure retry behavior for failed requests.
    ///
    /// By default, requests are retried up to 3 times with exponential backoff.
    /// Retries happen on network errors and certain HTTP status codes (429, 500, 502, 503, 504).
    #[must_use]
    pub const fn retry(mut self, config: RetryConfig) -> Self {
        self.retry = config;
        self
    }

    /// Set maximum number of retry attempts.
    #[must_use]
    pub const fn max_retries(mut self, max_retries: u32) -> Self {
        self.retry.max_retries = max_retries;
        self
    }

    /// Disable retries entirely.
    #[must_use]
    pub fn no_retry(mut self) -> Self {
        self.retry = RetryConfig::none();
        self
    }

    /// Set the request timeout.
    ///
    /// Default is 5 minutes, which is generous for long completions.
    /// The timeout applies to the entire request, including connection and response streaming.
    #[must_use]
    pub const fn timeout(mut self, timeout: Duration) -> Self {
        self.request_timeout = timeout;
        self
    }

    /// Consume the builder and create an [`OpenAI`] client.
    #[must_use]
    pub fn build(self) -> OpenAI {
        OpenAI {
            inner: Arc::new(Config {
                api_key: self.api_key,
                base_url: self.base_url,
                api_kind: self.api_kind,
                chat_model: self.chat_model,
                embedding_model: self.embedding_model,
                embedding_dimensions: self.embedding_dimensions,
                image_model: self.image_model,
                audio_model: self.audio_model,
                audio_voice: self.audio_voice,
                audio_format: self.audio_format,
                transcription_model: self.transcription_model,
                moderation_model: self.moderation_model,
                legacy_max_tokens: self.legacy_max_tokens,
                organization: self.organization,
                native_abilities: self.native_abilities,
                retry: self.retry,
                request_timeout: self.request_timeout,
                extra_headers: self.extra_headers,
                store: self.store,
            }),
        }
    }
}

#[derive(Debug, Clone)]
pub struct Config {
    pub(crate) api_key: String,
    pub(crate) base_url: String,
    pub(crate) api_kind: ApiKind,
    pub(crate) chat_model: String,
    pub(crate) embedding_model: String,
    pub(crate) embedding_dimensions: usize,
    pub(crate) image_model: String,
    pub(crate) audio_model: String,
    pub(crate) audio_voice: String,
    pub(crate) audio_format: String,
    pub(crate) transcription_model: String,
    pub(crate) moderation_model: String,
    pub(crate) legacy_max_tokens: bool,
    pub(crate) organization: Option<String>,
    pub(crate) native_abilities: Vec<Ability>,
    pub(crate) retry: RetryConfig,
    pub(crate) request_timeout: Duration,
    pub(crate) extra_headers: Vec<(String, String)>,
    pub(crate) store: Option<bool>,
}

impl Config {
    pub(crate) fn request_url(&self, path: &str) -> String {
        format!(
            "{}/{}",
            self.base_url.trim_end_matches('/'),
            path.trim_start_matches('/')
        )
    }

    pub(crate) fn request_auth(&self) -> String {
        format!("Bearer {}", self.api_key)
    }
}

fn sanitize_model(model: impl Into<String>) -> String {
    model.into().trim().to_string()
}

fn infer_embedding_dim(model: &str) -> Option<usize> {
    match model {
        "text-embedding-3-large" => Some(3072),
        "text-embedding-3-small" | "text-embedding-ada-002" => Some(1536),
        _ => None,
    }
}