//! Shared LLM API transport: provider dispatch + request send + streaming
//! (SSE, NDJSON) and non-streaming response consumption. Provider-specific
//! request-body construction lives in `crate::llm::providers`; this file is
//! the wire-format layer below that.

use std::time::{Duration, Instant};

use crate::agent_events::{AgentEvent, ToolCallStatus};
use crate::value::{VmError, VmValue};

use super::openai_normalize::{
    append_paragraph, debug_log_message_shapes, extract_openai_delta_field_str,
};
use super::options::{DeltaSender, LlmRequestPayload};
use super::partial_tool_args::{project_partial, DeltaCoalescer, PartialToolArgs};
use super::response::{
    billed_noncommittal_completion_error, extract_cache_read_tokens, extract_cache_write_tokens,
    is_billed_noncommittal_completion, parse_llm_response, CompletionContractSignals,
};
use super::result::LlmResult;
use super::telemetry::{elapsed_ms, source as telemetry_source, ProviderTelemetry};
use super::thinking::ThinkingStreamSplitter;

fn parse_ollama_tool_arguments(arguments: &serde_json::Value) -> serde_json::Value {
    match arguments {
        serde_json::Value::Object(_) | serde_json::Value::Array(_) | serde_json::Value::Null => {
            arguments.clone()
        }
        serde_json::Value::String(text) => serde_json::from_str(text).unwrap_or_else(|err| {
            serde_json::json!({
                "__parse_error": format!(
                    "Could not parse tool arguments as JSON: {}. Raw input: {}",
                    err,
                    &text[..text.len().min(200)]
                )
            })
        }),
        other => other.clone(),
    }
}

fn append_ollama_tool_calls(
    message: &serde_json::Value,
    tool_calls: &mut Vec<serde_json::Value>,
    blocks: &mut Vec<serde_json::Value>,
) {
    let Some(calls) = message.get("tool_calls").and_then(|value| value.as_array()) else {
        return;
    };

    for (idx, call) in calls.iter().enumerate() {
        let function = call.get("function").unwrap_or(call);
        let name = function
            .get("name")
            .and_then(|value| value.as_str())
            .unwrap_or("")
            .to_string();
        if name.is_empty() {
            continue;
        }
        let arguments = parse_ollama_tool_arguments(
            function
                .get("arguments")
                .unwrap_or(&serde_json::Value::Object(Default::default())),
        );
        let id = call
            .get("id")
            .and_then(|value| value.as_str())
            .map(str::to_string)
            .or_else(|| {
                function
                    .get("index")
                    .and_then(|value| value.as_i64())
                    .map(|index| format!("ollama_tool_{index}"))
            })
            .unwrap_or_else(|| format!("ollama_tool_{}", tool_calls.len() + idx));
        tool_calls.push(serde_json::json!({
            "id": id,
            "name": name,
            "arguments": arguments,
        }));
        blocks.push(serde_json::json!({
            "type": "tool_call",
            "id": id,
            "name": function.get("name").cloned().unwrap_or(serde_json::json!("")),
            "arguments": arguments,
            "visibility": "internal",
        }));
    }
}

fn should_request_stream_usage(is_anthropic_style: bool, is_ollama: bool, endpoint: &str) -> bool {
    if is_anthropic_style {
        return false;
    }
    // OpenAI-compatible streams expose aggregate usage in a final chunk
    // when requested. Ollama's native `/api/chat` shape does not use
    // this field, but its `/v1/chat/completions` compatibility endpoint
    // does.
    !is_ollama || endpoint.contains("/v1/")
}

fn classify_transport_http_error(
    provider: &str,
    status: reqwest::StatusCode,
    retry_after: Option<&str>,
    body: &str,
    is_anthropic_style: bool,
    is_ollama: bool,
) -> String {
    if is_anthropic_style {
        return crate::llm::providers::AnthropicProvider::classify_http_error(
            status,
            retry_after,
            body,
        )
        .message;
    }
    if is_ollama {
        return crate::llm::providers::OllamaProvider::classify_http_error(
            status,
            retry_after,
            body,
        )
        .message;
    }
    crate::llm::providers::OpenAiCompatibleProvider::classify_http_error(
        provider,
        status,
        retry_after,
        body,
    )
    .message
}

/// Dispatch an LLM API call to the appropriate provider. This is the main
/// entry point that routes to provider-specific implementations via the
/// provider plugin architecture.
///
/// The dispatch order is:
/// 1. Route providers declared with `protocol = "acp"` to the ACP adapter
/// 2. Check the thread-local provider registry (populated by `register_default_providers`)
/// 3. Fall back to config-based resolution (for dynamically-configured providers)
/// 4. Use the legacy inline dispatch as a final fallback
pub(super) async fn vm_call_llm_api(
    opts: &LlmRequestPayload,
    delta_tx: Option<DeltaSender>,
) -> Result<LlmResult, VmError> {
    let provider = &opts.provider;

    if crate::llm::providers::AcpProvider::is_configured_acp(provider) {
        return crate::llm::providers::AcpProvider::new(provider.clone())
            .chat_impl(opts, delta_tx)
            .await;
    }

    if crate::llm::provider::is_provider_registered(provider) {
        return dispatch_to_registered_provider(opts, delta_tx).await;
    }

    // Fallback for unregistered providers: dispatch by API style.
    let is_ollama = crate::llm::provider::provider_uses_ollama_messages(provider, &opts.model);
    let is_anthropic =
        crate::llm::provider::provider_uses_anthropic_messages(provider, &opts.model);

    if is_ollama {
        return crate::llm::providers::OllamaProvider
            .chat_impl(opts, delta_tx)
            .await;
    }

    let body = if is_anthropic {
        crate::llm::providers::AnthropicProvider::build_request_body(opts)
    } else {
        crate::llm::providers::OpenAiCompatibleProvider::build_request_body(opts, false)
    };

    vm_call_llm_api_with_body(opts, delta_tx, body, is_anthropic, is_ollama).await
}

/// Dispatch to a registered provider by name.
///
/// Provider selection uses trait methods (`is_mock()`, `is_local()`,
/// `is_anthropic_style()`) instead of string comparisons so that each
/// provider owns its own dispatch semantics.
async fn dispatch_to_registered_provider(
    opts: &LlmRequestPayload,
    delta_tx: Option<DeltaSender>,
) -> Result<LlmResult, VmError> {
    use crate::llm::provider::LlmProvider;

    // Providers are zero-cost unit structs constructed inline to avoid
    // RefCell-across-await conflicts on a shared registry.
    let provider = &opts.provider;

    let mock = crate::llm::providers::MockProvider;
    if mock.is_mock() && provider == mock.name() {
        return mock.chat_impl(opts, delta_tx).await;
    }

    if crate::llm::fake::FakeLlmProvider::should_intercept(provider) {
        return crate::llm::fake::FakeLlmProvider
            .chat_impl(opts, delta_tx)
            .await;
    }

    let ollama = crate::llm::providers::OllamaProvider;
    if (provider == ollama.name()
        || crate::llm::provider::provider_uses_ollama_messages(provider, &opts.model))
        && ollama.is_local()
    {
        return ollama.chat_impl(opts, delta_tx).await;
    }

    let gemini = crate::llm::providers::GeminiProvider;
    if provider == gemini.name() {
        return gemini.chat_impl(opts, delta_tx).await;
    }

    if provider == "bedrock" {
        return crate::llm::providers::BedrockProvider
            .chat_impl(opts, delta_tx)
            .await;
    }

    if provider == "azure_openai" {
        return crate::llm::providers::AzureOpenAiProvider
            .chat_impl(opts, delta_tx)
            .await;
    }

    if provider == "vertex" {
        return crate::llm::providers::VertexProvider
            .chat_impl(opts, delta_tx)
            .await;
    }

    if crate::llm::provider::provider_uses_anthropic_messages(provider, &opts.model) {
        let anthropic = crate::llm::providers::AnthropicProvider;
        return anthropic.chat_impl(opts, delta_tx).await;
    }

    crate::llm::providers::OpenAiCompatibleProvider::new(provider.clone())
        .chat_impl(opts, delta_tx)
        .await
}

/// Execute an LLM API call with a pre-built request body. This is the shared
/// transport layer used by all provider implementations. It handles:
/// - Provider-specific overrides merging
/// - Stream vs non-stream transport selection
/// - HTTP error classification
/// - SSE and NDJSON response parsing
///
/// Provider implementations call this after building their provider-specific
/// request body via `build_request_body()`.
pub(crate) async fn vm_call_llm_api_with_body(
    opts: &LlmRequestPayload,
    delta_tx: Option<DeltaSender>,
    body: serde_json::Value,
    is_anthropic_style: bool,
    is_ollama: bool,
) -> Result<LlmResult, VmError> {
    let started = Instant::now();
    let mut result =
        vm_call_llm_api_with_body_inner(opts, delta_tx, body, is_anthropic_style, is_ollama)
            .await?;
    // Reserved-token tool-call delimiter remap (single boundary).
    //
    // For models that reserve `<tool_call>`/`</tool_call>` as special tokens
    // (`reserved_tool_call_token` in capabilities.toml) the prompt is sent with
    // the delimiters swapped for a non-special wire form (`[[CALL]]`; see
    // `build_request_body` + `tool_delimiter`). The completion comes back in
    // that same wire form and MUST be mapped back to canonical before the
    // transcript and tagged tool-call parser ever see it.
    //
    // This lives in the shared transport funnel — not in `chat_impl` — so it
    // fires identically for every route into this function: the registered
    // OpenAI-compat path (`chat_impl`), the *unregistered* OpenAI-compat
    // fallback in `vm_call_llm_api` (e.g. a `llamacpp` provider configured via
    // providers.toml but never `provider_register`-ed), and both the streaming
    // (SSE/NDJSON) and non-streaming transports. Previously the remap lived
    // only in `chat_impl`, so an unregistered `llamacpp` qwen3.6 route returned
    // raw `[[CALL]]` text, the parser found zero `<tool_call>` blocks, and the
    // agent dispatched no tools (convergence-fatal). The streamed live deltas
    // are canonicalized separately by `canonicalizing_delta_tx`; this remaps the
    // assembled `result.text` that the parser/transcript consume.
    if crate::llm::capabilities::lookup(&opts.provider, &opts.model).reserved_tool_call_token {
        let wire_open = result.text.matches("[[CALL]]").count();
        result.text = crate::llm::tool_delimiter::wire_to_canonical(&result.text);
        let canon_open = result.text.matches("<tool_call>").count();
        tracing::debug!(
            target: "harn::llm::tool_delimiter",
            provider = %opts.provider,
            model = %opts.model,
            wire_open_markers = wire_open,
            canonical_open_blocks = canon_open,
            "reserved-token wire->canonical remap applied to assembled completion",
        );
    }
    // Preserve a per-call wall clock regardless of provider. Server-side
    // timings (when available) cover only the model's view; client_wall_ms
    // captures network + streaming overhead the server cannot see, so eval
    // dashboards can decompose total latency end-to-end.
    if result.telemetry.client_wall_ms.is_none() {
        result.telemetry.client_wall_ms = Some(elapsed_ms(started));
    }
    if result.telemetry.source.is_empty() {
        result.telemetry.source = telemetry_source::UNKNOWN.to_string();
    }
    Ok(result)
}

async fn vm_call_llm_api_with_body_inner(
    opts: &LlmRequestPayload,
    delta_tx: Option<DeltaSender>,
    mut body: serde_json::Value,
    is_anthropic_style: bool,
    is_ollama: bool,
) -> Result<LlmResult, VmError> {
    let provider = &opts.provider;
    let model = &opts.model;
    let wants_streaming = delta_tx.is_some() && opts.stream;
    // Whether this request offered any tools to the model. Used by the
    // billed-no-op contract guard so a deliberately terse text answer to a
    // tool-less prompt is never misclassified as a missing tool call.
    let tools_offered = body
        .get("tools")
        .and_then(|tools| tools.as_array())
        .map(|tools| !tools.is_empty())
        .unwrap_or(false);

    let resolved = crate::llm::helpers::ResolvedProvider::resolve(provider);
    let use_stream_transport = if is_ollama && !opts.stream {
        crate::events::log_warn(
            "llm",
            "stream=false is not supported by Ollama, using streaming",
        );
        true
    } else {
        wants_streaming || is_ollama
    };

    if !is_ollama {
        if let Some(ref overrides) = opts.provider_overrides {
            if let Some(obj) = overrides.as_object() {
                for (k, v) in obj {
                    body[k] = v.clone();
                }
            }
        }
    }
    if provider == "openrouter"
        && (body.get("response_format").is_some() || body.get("top_k").is_some())
    {
        crate::llm::providers::openai_compat::ensure_openrouter_require_parameters(&mut body);
    }

    if let Some(messages) = body.get("messages").and_then(|value| value.as_array()) {
        debug_log_message_shapes(
            &format!("outbound provider={provider} model={model}"),
            messages,
        );
    }

    if use_stream_transport {
        body["stream"] = serde_json::json!(true);
        // OpenAI-style: request usage in the final streaming chunk.
        if should_request_stream_usage(is_anthropic_style, is_ollama, &resolved.endpoint) {
            body["stream_options"] = serde_json::json!({"include_usage": true});
        }
    }

    let client = if use_stream_transport {
        crate::llm::shared_streaming_client().clone()
    } else {
        crate::llm::shared_blocking_client().clone()
    };

    let req = client
        .post(resolved.url())
        .header("Content-Type", "application/json")
        .timeout(std::time::Duration::from_secs(opts.resolve_timeout()))
        .json(&body);
    let mut req = resolved.apply_headers(req, &opts.api_key);
    if is_anthropic_style && !opts.anthropic_beta_features.is_empty() {
        req = req.header("anthropic-beta", opts.anthropic_beta_features.join(","));
    }

    if use_stream_transport {
        let tx = if let Some(tx) = delta_tx {
            tx
        } else {
            let (tx, _rx) = tokio::sync::mpsc::unbounded_channel::<String>();
            tx
        };
        let max_attempts = if is_ollama { 2 } else { 1 };
        let unload_grace = if is_ollama {
            crate::llm::api::ollama_unload_grace_duration_from_env()
        } else {
            Duration::ZERO
        };
        let mut ollama_warmup_gate = false;
        for attempt in 0..max_attempts {
            let req = client
                .post(resolved.url())
                .header("Content-Type", "application/json")
                .timeout(std::time::Duration::from_secs(opts.resolve_timeout()))
                .json(&body);
            let mut req = resolved.apply_headers(req, &opts.api_key);
            if is_anthropic_style && !opts.anthropic_beta_features.is_empty() {
                req = req.header("anthropic-beta", opts.anthropic_beta_features.join(","));
            }
            let response = send_stream_request_with_ollama_warmup(
                req,
                provider,
                model,
                is_ollama,
                unload_grace,
                &mut ollama_warmup_gate,
            )
            .await?;
            if !response.status().is_success() {
                let status = response.status();
                let retry_after = response
                    .headers()
                    .get("retry-after")
                    .and_then(|v| v.to_str().ok())
                    .map(|s| s.to_string());
                let body = response.text().await.unwrap_or_default();
                let msg = classify_transport_http_error(
                    provider,
                    status,
                    retry_after.as_deref(),
                    &body,
                    is_anthropic_style,
                    is_ollama,
                );
                return Err(VmError::Thrown(VmValue::String(std::sync::Arc::from(msg))));
            }
            let is_sse = response
                .headers()
                .get("content-type")
                .and_then(|v| v.to_str().ok())
                .is_some_and(|ct| ct.contains("text/event-stream"));
            // Build a fresh schema watch per attempt so an Ollama-retry
            // restart doesn't see chunks from the previous run.
            let schema_watch = super::schema_stream::StreamSchemaWatch::from_payload(opts);
            if is_sse {
                return vm_call_llm_api_sse_from_response(
                    response,
                    provider,
                    model,
                    is_anthropic_style,
                    tx,
                    opts.session_id.as_deref(),
                    schema_watch,
                    tools_offered,
                )
                .await;
            }
            match vm_call_llm_api_ndjson_from_response(
                response,
                provider,
                model,
                tx.clone(),
                unload_grace,
                &mut ollama_warmup_gate,
                schema_watch,
            )
            .await
            {
                // A `done_reason == "length"` truncation now returns Ok with an
                // empty body and `stop_reason: Some("length")`, so it bypasses
                // the retry guard below entirely — a deterministic token-cap cut
                // would just re-truncate on every retry. Only the genuine
                // empty-content parser bug (done_reason stop/absent) is retried.
                Ok(result) => return Ok(result),
                Err(err)
                    if is_ollama
                        && attempt + 1 < max_attempts
                        && is_ollama_empty_content_parser_bug(&err) =>
                {
                    crate::events::log_warn(
                        "llm",
                        &format!(
                            "ollama model {model} returned empty content with eval_count; retrying once"
                        ),
                    );
                    continue;
                }
                Err(err) => return Err(err),
            }
        }
        unreachable!("streaming LLM attempt loop exhausted without returning");
    }

    let response = req.send().await.map_err(|e| {
        VmError::Thrown(VmValue::String(std::sync::Arc::from(format!(
            "{provider} API error: {e}"
        ))))
    })?;

    // Check HTTP status BEFORE parsing the body as LLM response, or error
    // responses (e.g. vLLM "prompt too long" 400) silently become malformed
    // parse results and the agent loop retries against the same bad context.
    if !response.status().is_success() {
        let status = response.status();
        let retry_after = response
            .headers()
            .get("retry-after")
            .and_then(|v| v.to_str().ok())
            .map(|s| s.to_string());
        let body = response.text().await.unwrap_or_default();
        let msg = classify_transport_http_error(
            provider,
            status,
            retry_after.as_deref(),
            &body,
            is_anthropic_style,
            is_ollama,
        );
        return Err(VmError::Thrown(VmValue::String(std::sync::Arc::from(msg))));
    }

    let json: serde_json::Value = response.json().await.map_err(|e| {
        VmError::Thrown(VmValue::String(std::sync::Arc::from(format!(
            "{provider} response parse error: {e}"
        ))))
    })?;

    let is_anthropic_style =
        crate::llm::provider::provider_uses_anthropic_messages(provider, model);
    parse_llm_response(&json, provider, model, is_anthropic_style, tools_offered)
}

/// Consume an SSE streaming response from an already-sent request.
/// Parses `data: {...}` lines from the response body, then defers to
/// [`consume_sse_lines`] for the parsing and event-emission logic so
/// tests can drive the same code path against an in-memory `AsyncBufRead`.
#[allow(clippy::too_many_arguments)]
async fn vm_call_llm_api_sse_from_response(
    response: reqwest::Response,
    provider: &str,
    model: &str,
    is_anthropic_style: bool,
    delta_tx: DeltaSender,
    session_id: Option<&str>,
    schema_watch: Option<super::schema_stream::StreamSchemaWatch>,
    tools_offered: bool,
) -> Result<LlmResult, VmError> {
    use tokio_stream::StreamExt;

    let stream = response.bytes_stream();
    let reader = tokio::io::BufReader::new(tokio_util::io::StreamReader::new(
        stream.map(|r| r.map_err(std::io::Error::other)),
    ));
    consume_sse_lines(
        reader,
        provider,
        model,
        is_anthropic_style,
        delta_tx,
        session_id,
        schema_watch,
        tools_offered,
    )
    .await
}

/// Try to publish the live `(tool_call_id, tool_name, accumulated_bytes)`
/// triple as a `ToolCallUpdate(Pending, raw_input | raw_input_partial)`
/// event. Coalescing + partial-parse logic lives here so both the
/// Anthropic and OpenAI branches of the SSE loop share one emit site.
fn try_emit_partial_tool_args(
    session_id: Option<&str>,
    tool_call_id: &str,
    tool_name: &str,
    accumulated: &str,
    coalescer: &mut DeltaCoalescer,
    now: Instant,
) {
    let Some(session_id) = session_id else {
        return;
    };
    if !coalescer.should_emit(now) {
        return;
    }
    let PartialToolArgs { value, raw_partial } = project_partial(accumulated);
    if value.is_none() && raw_partial.is_none() {
        return;
    }
    let event = AgentEvent::ToolCallUpdate {
        session_id: session_id.to_string(),
        tool_call_id: tool_call_id.to_string(),
        tool_name: tool_name.to_string(),
        status: ToolCallStatus::Pending,
        raw_output: None,
        error: None,
        duration_ms: None,
        execution_duration_ms: None,
        error_category: None,
        executor: None,
        raw_input: value,
        raw_input_partial: raw_partial,
        audit: crate::orchestration::current_mutation_session(),

        parsing: None,
    };
    crate::llm::agent_runtime::emit_agent_event_sync(&event);
}

async fn send_stream_request_with_ollama_warmup(
    req: reqwest::RequestBuilder,
    provider: &str,
    model: &str,
    is_ollama: bool,
    unload_grace: Duration,
    warmup_gate: &mut bool,
) -> Result<reqwest::Response, VmError> {
    let send = req.send();
    if !is_ollama || *warmup_gate || unload_grace.is_zero() {
        return send
            .await
            .map_err(|error| stream_send_error(provider, error));
    }

    tokio::pin!(send);
    tokio::select! {
        response = &mut send => response.map_err(|error| stream_send_error(provider, error)),
        _ = tokio::time::sleep(unload_grace) => {
            *warmup_gate = true;
            emit_ollama_warmup_progress(model);
            send.await.map_err(|error| stream_send_error(provider, error))
        }
    }
}

fn stream_send_error(provider: &str, error: reqwest::Error) -> VmError {
    let kind = if error.is_timeout() {
        "timeout"
    } else if error.is_connect() {
        "connect"
    } else if error.is_request() {
        "request_build"
    } else if error.is_body() {
        "body"
    } else {
        "unknown"
    };
    // "unknown" uses Debug repr to surface the inner cause.
    let detail = if kind == "unknown" {
        format!("{provider} stream error ({kind}): {error:?}")
    } else {
        format!("{provider} stream error ({kind}): {error}")
    };
    VmError::Thrown(VmValue::String(std::sync::Arc::from(detail)))
}

/// Map an Anthropic `tool_use` block id (or, as a fallback, an
/// iteration-relative index) to the canonical `tool-{id}` shape that
/// `tool_dispatch.rs` later emits from. Keeping the two sites in sync
/// lets clients correlate streaming `Pending` updates with the eventual
/// `InProgress`/`Completed` lifecycle.
fn streaming_tool_call_id(provider_id: &str, fallback_index: usize) -> String {
    if provider_id.is_empty() {
        format!("tool-stream-{fallback_index}")
    } else {
        format!("tool-{provider_id}")
    }
}

/// Pure SSE-line consumer extracted so #693's streaming-partial-args
/// behavior can be tested against canned byte streams without standing
/// up a full `reqwest::Response`. The Anthropic / OpenAI branches and
/// the trailing accumulator drain that finalize the call live here.
#[allow(clippy::too_many_arguments)]
pub(super) async fn consume_sse_lines<R: tokio::io::AsyncBufRead + Unpin>(
    reader: R,
    provider: &str,
    model: &str,
    is_anthropic_style: bool,
    delta_tx: DeltaSender,
    session_id: Option<&str>,
    mut schema_watch: Option<super::schema_stream::StreamSchemaWatch>,
    tools_offered: bool,
) -> Result<LlmResult, VmError> {
    use tokio::io::AsyncBufReadExt;
    let mut lines = reader.lines();

    let mut text = String::new();
    let mut input_tokens: i64 = 0;
    let mut output_tokens: i64 = 0;
    let mut tool_calls: Vec<serde_json::Value> = Vec::new();
    let mut blocks: Vec<serde_json::Value> = Vec::new();
    let mut telemetry = ProviderTelemetry::default();
    let mut anth_request_id: Option<String> = None;
    // Set once the provider echoes the fast-mode knob (`speed` /
    // `service_tier`) on a streamed event; drives premium-tier billing.
    let mut served_fast = false;

    struct ToolBlock {
        id: String,
        name: String,
        input_json: String,
        /// Stable id used for `AgentEvent::ToolCall*` streaming
        /// emissions (#693). Must match the shape `tool_dispatch.rs`
        /// constructs later so clients can correlate the streaming
        /// `Pending` updates with the eventual `InProgress`/`Completed`
        /// lifecycle.
        tool_call_id: String,
        /// Coalescing gate so a tool that arrives in 30 small deltas
        /// emits ~6 `ToolCallUpdate` events instead of 30.
        coalescer: DeltaCoalescer,
    }
    let mut current_tool: Option<ToolBlock> = None;
    // Mirror structure for server-side tool-search queries: Anthropic
    // streams the query JSON the same way as a regular tool_use, but we
    // route it to a `tool_search_query` transcript event instead of the
    // dispatchable `tool_calls` vector.
    struct ServerToolBlock {
        id: String,
        name: String,
        input_json: String,
    }
    let mut current_server_tool: Option<ServerToolBlock> = None;
    let mut thinking_text = String::new();
    let mut in_thinking_block = false;
    let mut stop_reason: Option<String> = None;
    let mut cache_read_tokens: i64 = 0;
    let mut cache_write_tokens: i64 = 0;
    // Counter for fallback streaming-tool-call ids when a provider sent
    // an empty id on the first tool_use block. Kept stable across the
    // stream so the coalesced updates reuse the same id the dispatcher
    // would compute.
    let mut anth_tool_block_index: usize = 0;

    /// Per-tool-call OpenAI streaming state. Tracks the accumulated
    /// arguments string, the tool name (filled when the first delta
    /// carries `function.name`), the synthetic `tool_call_id` we use
    /// for `AgentEvent::ToolCall` emission, whether the initial
    /// `ToolCall(Pending)` event has fired yet, and a coalescer so
    /// argument-delta storms don't fan out per-byte.
    struct OaiToolStream {
        id: String,
        name: String,
        args: String,
        tool_call_id: String,
        announced: bool,
        coalescer: DeltaCoalescer,
    }
    let mut oai_tool_map: std::collections::HashMap<u64, OaiToolStream> =
        std::collections::HashMap::new();
    // Qwen3/3.5 via vLLM emit inline `<think>...</think>`. Strip these
    // out of the visible delta stream so the tool-call parser / progress
    // UI only see the real response.
    let mut oai_thinking_splitter = ThinkingStreamSplitter::new();

    loop {
        let line = match lines.next_line().await {
            Ok(Some(line)) => line,
            // Clean EOF without `[DONE]` stays a normal stop: Anthropic ends
            // streams with `message_stop` (no `[DONE]` sentinel) and some
            // OpenAI-compatible servers close the connection cleanly.
            Ok(None) => break,
            Err(error) => {
                // A mid-body read failure (the whole-request reqwest timeout
                // from `resolve_timeout`, a connection reset, ...) used to be
                // swallowed here, silently truncating the stream and returning
                // a zero-token "success" that the agent loop accepted as an
                // empty assistant turn (observed live: an OpenRouter call hung
                // 133s and came back with output_tokens=0). Surface it in the
                // same transient stream-error class other transports use so
                // the existing retry machinery picks it up.
                return Err(VmError::Thrown(VmValue::String(std::sync::Arc::from(
                    format!("{provider} stream error (mid-stream read): {error}"),
                ))));
            }
        };
        let data = if let Some(d) = line.strip_prefix("data: ") {
            d
        } else {
            continue;
        };
        if data == "[DONE]" {
            break;
        }
        let json: serde_json::Value = match serde_json::from_str(data) {
            Ok(v) => v,
            Err(_) => continue,
        };

        if is_anthropic_style {
            match json["type"].as_str() {
                Some("message_start") => {
                    if let Some(n) = json["message"]["usage"]["input_tokens"].as_i64() {
                        input_tokens = n;
                    }
                    served_fast |= crate::llm::fast_mode::served_fast(model, &json["message"]);
                    let usage = &json["message"]["usage"];
                    let cr = extract_cache_read_tokens(usage);
                    if cr > 0 {
                        cache_read_tokens = cr;
                    }
                    let cw = extract_cache_write_tokens(usage);
                    if cw > 0 {
                        cache_write_tokens = cw;
                    }
                    if let Some(rid) = json["message"]["id"].as_str() {
                        if !rid.is_empty() {
                            anth_request_id = Some(rid.to_string());
                        }
                    }
                }
                Some("content_block_start") => {
                    let block = &json["content_block"];
                    match block["type"].as_str() {
                        Some("tool_use") => {
                            let id = block["id"].as_str().unwrap_or("").to_string();
                            let name = block["name"].as_str().unwrap_or("").to_string();
                            anth_tool_block_index += 1;
                            let tool_call_id = streaming_tool_call_id(&id, anth_tool_block_index);
                            // Streaming announcement: emit before any
                            // arg deltas so ACP clients can render
                            // "calling search_web…" with zero latency.
                            if let Some(sid) = session_id {
                                let tool_kind =
                                    crate::orchestration::current_tool_annotations(&name)
                                        .map(|a| a.kind);
                                crate::llm::agent_runtime::emit_agent_event_sync(
                                    &AgentEvent::ToolCall {
                                        session_id: sid.to_string(),
                                        tool_call_id: tool_call_id.clone(),
                                        tool_name: name.clone(),
                                        kind: tool_kind,
                                        status: ToolCallStatus::Pending,
                                        raw_input: serde_json::Value::Object(Default::default()),
                                        audit: crate::orchestration::current_mutation_session(),

                                        parsing: None,
                                    },
                                );
                            }
                            current_tool = Some(ToolBlock {
                                id,
                                name,
                                input_json: String::new(),
                                tool_call_id,
                                coalescer: DeltaCoalescer::new(),
                            });
                        }
                        Some("server_tool_use") => {
                            current_server_tool = Some(ServerToolBlock {
                                id: block["id"].as_str().unwrap_or("").to_string(),
                                name: block["name"].as_str().unwrap_or("").to_string(),
                                input_json: String::new(),
                            });
                        }
                        Some("tool_search_tool_result") => {
                            // Non-streaming content: Anthropic embeds the
                            // references directly in the block_start
                            // payload. Record immediately — no deltas
                            // follow for this block type.
                            let refs: Vec<serde_json::Value> = block["content"]["tool_references"]
                                .as_array()
                                .cloned()
                                .unwrap_or_default();
                            blocks.push(serde_json::json!({
                                "type": "tool_search_result",
                                "tool_use_id": block["tool_use_id"].clone(),
                                "tool_references": refs,
                                "visibility": "internal",
                            }));
                        }
                        Some("thinking") => {
                            in_thinking_block = true;
                        }
                        _ => {}
                    }
                }
                Some("content_block_delta") => {
                    let delta = &json["delta"];
                    match delta["type"].as_str() {
                        Some("text_delta") => {
                            if let Some(t) = delta["text"].as_str() {
                                text.push_str(t);
                                let _ = delta_tx.send(t.to_string());
                                blocks.push(serde_json::json!({"type": "output_text", "text": t, "visibility": "public"}));
                                if let Some(watch) = schema_watch.as_mut() {
                                    if let Some(abort) = watch.observe(t) {
                                        return Err(abort.into_vm_error());
                                    }
                                }
                            }
                        }
                        Some("thinking_delta") => {
                            if let Some(t) = delta["thinking"].as_str() {
                                thinking_text.push_str(t);
                                blocks.push(serde_json::json!({"type": "reasoning", "text": t, "visibility": "private"}));
                            }
                        }
                        Some("input_json_delta") => {
                            if let Some(ref mut tool) = current_tool {
                                if let Some(j) = delta["partial_json"].as_str() {
                                    tool.input_json.push_str(j);
                                }
                                try_emit_partial_tool_args(
                                    session_id,
                                    &tool.tool_call_id,
                                    &tool.name,
                                    &tool.input_json,
                                    &mut tool.coalescer,
                                    Instant::now(),
                                );
                            } else if let Some(ref mut server_tool) = current_server_tool {
                                if let Some(j) = delta["partial_json"].as_str() {
                                    server_tool.input_json.push_str(j);
                                }
                            }
                        }
                        _ => {}
                    }
                }
                Some("content_block_stop") => {
                    if let Some(tool) = current_tool.take() {
                        let args = serde_json::from_str::<serde_json::Value>(&tool.input_json)
                            .unwrap_or(serde_json::Value::Object(Default::default()));
                        tool_calls.push(serde_json::json!({
                            "id": tool.id, "name": tool.name, "arguments": args,
                        }));
                        blocks.push(serde_json::json!({"type": "tool_call", "id": tool.id, "name": tool.name, "arguments": args, "visibility": "internal"}));
                    } else if let Some(server_tool) = current_server_tool.take() {
                        // Emit a `tool_search_query` transcript event —
                        // not dispatchable, just observability.
                        let query =
                            serde_json::from_str::<serde_json::Value>(&server_tool.input_json)
                                .unwrap_or(serde_json::Value::Object(Default::default()));
                        blocks.push(serde_json::json!({
                            "type": "tool_search_query",
                            "id": server_tool.id,
                            "name": server_tool.name,
                            "query": query,
                            "visibility": "internal",
                        }));
                    }
                    in_thinking_block = false;
                }
                Some("message_delta") => {
                    if let Some(n) = json["usage"]["output_tokens"].as_i64() {
                        output_tokens = n;
                    }
                    let usage = &json["usage"];
                    let cr = extract_cache_read_tokens(usage);
                    if cr > 0 {
                        cache_read_tokens = cr;
                    }
                    let cw = extract_cache_write_tokens(usage);
                    if cw > 0 {
                        cache_write_tokens = cw;
                    }
                    if let Some(sr) = json["delta"]["stop_reason"].as_str() {
                        stop_reason = Some(sr.to_string());
                    }
                }
                _ => {}
            }
        } else {
            let choice = &json["choices"][0];
            let delta = &choice["delta"];

            if let Some(content) = delta["content"].as_str() {
                let visible = oai_thinking_splitter.push(content);
                if !visible.is_empty() {
                    text.push_str(&visible);
                    let _ = delta_tx.send(visible.clone());
                    blocks.push(serde_json::json!({"type": "output_text", "text": visible, "visibility": "public"}));
                    if let Some(watch) = schema_watch.as_mut() {
                        if let Some(abort) = watch.observe(&visible) {
                            return Err(abort.into_vm_error());
                        }
                    }
                }
            }
            // Streaming deltas for `reasoning` (Ollama OpenAI-compat,
            // OpenRouter passthrough), `reasoning_content` (DashScope,
            // Together), and `reasoning_details` (MiniMax) arrive as
            // token-sized fragments — `"Here"`,
            // `"'s"`, `" a"`, `" thinking"`. Concatenate them verbatim;
            // `extract_openai_message_field_as_text` + `append_paragraph`
            // would `.trim()` each fragment (losing inter-token spaces)
            // and inject a newline between every chunk, producing the
            // one-token-per-line reasoning text we used to surface as
            // `"The\ntask\nis\nto\nextend"`. The non-streaming response
            // path still uses `append_paragraph` because there each
            // field arrives as a single complete block.
            let reasoning_delta = extract_openai_delta_field_str(
                delta,
                &["reasoning", "reasoning_content", "reasoning_details"],
            );
            if !reasoning_delta.is_empty() {
                thinking_text.push_str(reasoning_delta);
                blocks.push(serde_json::json!({"type": "reasoning", "text": reasoning_delta, "visibility": "private"}));
            }

            // Only capture finish_reason once; OpenRouter can send
            // duplicates (qwen-code#2402) that truncate in-progress tool
            // calls.
            if stop_reason.is_none() {
                if let Some(fr) = choice["finish_reason"].as_str() {
                    stop_reason = Some(fr.to_string());
                }
            }

            if let Some(tcs) = delta["tool_calls"].as_array() {
                for tc in tcs {
                    // OpenAI Responses-API server-side tool_search
                    // (harn#71) streams as `tool_search_call` /
                    // `tool_search_output` entries in the tool_calls
                    // array. Record them as transcript events, never
                    // as dispatchable calls.
                    let tc_type = tc["type"].as_str().unwrap_or("");
                    if tc_type == "tool_search_call" {
                        let id = tc["id"].as_str().unwrap_or("").to_string();
                        let query = tc.get("query").cloned().unwrap_or_else(|| {
                            tc.get("input").cloned().unwrap_or(serde_json::Value::Null)
                        });
                        blocks.push(serde_json::json!({
                            "type": "tool_search_query",
                            "id": id,
                            "name": "tool_search",
                            "query": query,
                            "visibility": "internal",
                        }));
                        continue;
                    }
                    if tc_type == "tool_search_output" {
                        let tool_use_id = tc["call_id"]
                            .as_str()
                            .or_else(|| tc["id"].as_str())
                            .unwrap_or("")
                            .to_string();
                        let references = tc["tool_references"]
                            .as_array()
                            .cloned()
                            .unwrap_or_default();
                        blocks.push(serde_json::json!({
                            "type": "tool_search_result",
                            "tool_use_id": tool_use_id,
                            "tool_references": references,
                            "visibility": "internal",
                        }));
                        continue;
                    }
                    let idx = tc["index"].as_u64().unwrap_or(0);
                    let stream_index = idx as usize + 1;
                    let entry = oai_tool_map.entry(idx).or_insert_with(|| {
                        let id = tc["id"].as_str().unwrap_or("").to_string();
                        let name = tc["function"]["name"].as_str().unwrap_or("").to_string();
                        let tool_call_id = streaming_tool_call_id(&id, stream_index);
                        OaiToolStream {
                            id,
                            name,
                            args: String::new(),
                            tool_call_id,
                            announced: false,
                            coalescer: DeltaCoalescer::new(),
                        }
                    });
                    // OpenAI sometimes splits the metadata across deltas:
                    // the first one carries `name`, later ones carry only
                    // `arguments`. Patch missing fields if a later delta
                    // fills them in.
                    if entry.id.is_empty() {
                        if let Some(id) = tc["id"].as_str() {
                            if !id.is_empty() {
                                entry.id = id.to_string();
                                entry.tool_call_id =
                                    streaming_tool_call_id(&entry.id, stream_index);
                            }
                        }
                    }
                    if entry.name.is_empty() {
                        if let Some(name) = tc["function"]["name"].as_str() {
                            if !name.is_empty() {
                                entry.name = name.to_string();
                            }
                        }
                    }
                    // Announce the tool call as soon as we have a name —
                    // before any arg deltas, so clients render "calling
                    // X…" immediately. If only arguments arrive first
                    // (rare; some self-hosted vLLM builds), we hold off
                    // and announce on the first real partial-args emit
                    // below.
                    if !entry.announced && !entry.name.is_empty() {
                        if let Some(sid) = session_id {
                            let tool_kind =
                                crate::orchestration::current_tool_annotations(&entry.name)
                                    .map(|a| a.kind);
                            crate::llm::agent_runtime::emit_agent_event_sync(
                                &AgentEvent::ToolCall {
                                    session_id: sid.to_string(),
                                    tool_call_id: entry.tool_call_id.clone(),
                                    tool_name: entry.name.clone(),
                                    kind: tool_kind,
                                    status: ToolCallStatus::Pending,
                                    raw_input: serde_json::Value::Object(Default::default()),
                                    audit: crate::orchestration::current_mutation_session(),

                                    parsing: None,
                                },
                            );
                            entry.announced = true;
                        }
                    }
                    if let Some(args) = tc["function"]["arguments"].as_str() {
                        entry.args.push_str(args);
                    }
                    if entry.announced {
                        try_emit_partial_tool_args(
                            session_id,
                            &entry.tool_call_id,
                            &entry.name,
                            &entry.args,
                            &mut entry.coalescer,
                            Instant::now(),
                        );
                    }
                }
            }

            served_fast |= crate::llm::fast_mode::served_fast(model, &json);
            if let Some(usage) = json.get("usage") {
                if let Some(n) = usage["prompt_tokens"].as_i64() {
                    input_tokens = n;
                }
                if let Some(n) = usage["completion_tokens"].as_i64() {
                    output_tokens = n;
                }
                let cr = extract_cache_read_tokens(usage);
                if cr > 0 {
                    cache_read_tokens = cr;
                }
                let cw = extract_cache_write_tokens(usage);
                if cw > 0 {
                    cache_write_tokens = cw;
                }
                let request_id = json
                    .get("id")
                    .and_then(serde_json::Value::as_str)
                    .filter(|value| !value.is_empty());
                telemetry = ProviderTelemetry::from_openai_usage(usage, request_id);
            }
        }
    }

    for (_, stream) in oai_tool_map {
        let args = serde_json::from_str::<serde_json::Value>(&stream.args)
            .unwrap_or(serde_json::Value::Object(Default::default()));
        tool_calls.push(serde_json::json!({
            "id": stream.id, "name": stream.name, "arguments": args,
        }));
        blocks.push(serde_json::json!({
            "type": "tool_call",
            "id": stream.id,
            "name": stream.name,
            "arguments": args,
            "visibility": "internal",
        }));
    }

    let final_visible = oai_thinking_splitter.flush();
    if !final_visible.is_empty() {
        text.push_str(&final_visible);
        let _ = delta_tx.send(final_visible.clone());
        blocks.push(serde_json::json!({"type": "output_text", "text": final_visible, "visibility": "public"}));
        if let Some(watch) = schema_watch.as_mut() {
            if let Some(abort) = watch.observe(&final_visible) {
                return Err(abort.into_vm_error());
            }
        }
    }
    if !oai_thinking_splitter.thinking.is_empty() {
        append_paragraph(&mut thinking_text, &oai_thinking_splitter.thinking);
    }

    let _ = in_thinking_block; // suppress unused warning

    // When the stream is cut off mid-thought (finish_reason == "length") with
    // no committed visible content, the reasoning trace is a partial, garbage
    // not-an-answer. Promoting it into `.text` surfaces that garbage as the
    // final answer. Leave `text` empty and expose the partial trace only via
    // `thinking`, mirroring `openai_normalize::normalize_openai_message_text`.
    let truncated = stop_reason.as_deref() == Some("length");
    // When the turn also carries a tool call, the reasoning is intermediate
    // chain-of-thought (the tool call is the real action), not a final answer.
    // gpt-oss / harmony models stream their analysis channel into the reasoning
    // delta and emit a tool call with no committed content; promoting that
    // reasoning into `.text` leaks private chain-of-thought into the user-facing
    // assistant message AND the transcript the eval grader mines. Keep `.text`
    // empty and surface the reasoning only via `thinking`, mirroring
    // `openai_normalize::normalize_openai_message_text`.
    if !truncated && tool_calls.is_empty() && text.is_empty() && !thinking_text.is_empty() {
        text = thinking_text.clone();
        blocks
            .push(serde_json::json!({"type": "output_text", "text": text, "visibility": "public"}));
    }
    let has_tool_search_block = blocks.iter().any(|b| {
        matches!(
            b.get("type").and_then(|v| v.as_str()),
            Some("tool_search_query") | Some("tool_search_result")
        )
    });
    if text.is_empty()
        && thinking_text.is_empty()
        && output_tokens > 0
        && tool_calls.is_empty()
        && !has_tool_search_block
    {
        return Err(VmError::Thrown(VmValue::String(std::sync::Arc::from(format!(
            "openai-compatible model {model} reported completion_tokens={output_tokens} but delivered no content, reasoning, or tool calls"
        )))));
    }
    // Deterministic upstream contract-violation backstop (streaming path).
    // Mirrors the non-streaming detector in `response::parse_llm_response`: a
    // short, clean, tool-offered turn that billed output but dispatched no tool
    // call is a billed no-op (the action went only to the reasoning channel).
    // The bare `text.is_empty()` guard above misses it because the upstream
    // emitted a tiny content preamble.
    if is_billed_noncommittal_completion(&CompletionContractSignals {
        stop_reason: stop_reason.as_deref(),
        output_tokens,
        tools_offered,
        tool_call_count: tool_calls.len(),
        has_tool_search_block,
        text: &text,
    }) {
        return Err(billed_noncommittal_completion_error(
            provider,
            model,
            output_tokens,
        ));
    }

    // Use the caller-supplied provider id rather than collapsing every
    // non-anthropic stream to "openai". The provider name shows up in the
    // observability transcript (`agent_observe::dump_llm_response`) and is
    // load-bearing for downstream classifiers (e.g. honors_chat_template_kwargs
    // routing in capability lookup) — collapsing it to "openai" hides which
    // OpenAI-compatible server (vLLM, llama.cpp, OpenRouter, llamacpp) the
    // call actually went to. Anthropic's classic SSE shape still implies
    // provider="anthropic" because the wire protocol is anthropic-specific
    // even when the configured provider name disagrees (proxies / mocks).
    let result_provider = if is_anthropic_style {
        "anthropic".to_string()
    } else {
        provider.to_string()
    };
    if telemetry.is_empty() && is_anthropic_style && (input_tokens > 0 || output_tokens > 0) {
        let usage = serde_json::json!({
            "input_tokens": input_tokens,
            "output_tokens": output_tokens,
        });
        telemetry = ProviderTelemetry::from_anthropic_usage(&usage, anth_request_id.as_deref());
    }
    Ok(LlmResult {
        text,
        tool_calls,
        input_tokens,
        output_tokens,
        cache_read_tokens,
        cache_write_tokens,
        cache_supported: true,
        model: model.to_string(),
        provider: result_provider,
        thinking: if thinking_text.is_empty() {
            None
        } else {
            Some(thinking_text)
        },
        thinking_summary: None,
        stop_reason,
        served_fast,
        blocks,
        logprobs: Vec::new(),
        telemetry,
    })
}

/// Consume an NDJSON streaming response, forwarding text deltas to `delta_tx`
/// while accumulating the full result.
///
/// Supports Ollama format (one JSON object per line):
/// `{"message":{"role":"assistant","content":"Hi"},"done":false}`
/// Final line has `"done":true` with token counts.
///
/// Also supports OpenAI-compatible NDJSON where each line is `data: {...}`.
async fn vm_call_llm_api_ndjson_from_response(
    response: reqwest::Response,
    provider: &str,
    model: &str,
    delta_tx: DeltaSender,
    unload_grace: Duration,
    warmup_gate: &mut bool,
    schema_watch: Option<super::schema_stream::StreamSchemaWatch>,
) -> Result<LlmResult, VmError> {
    use tokio_stream::StreamExt;

    let stream = response.bytes_stream();
    let reader = tokio::io::BufReader::new(tokio_util::io::StreamReader::new(
        stream.map(|r| r.map_err(std::io::Error::other)),
    ));
    consume_ollama_ndjson_lines(
        reader,
        provider,
        model,
        delta_tx,
        unload_grace,
        warmup_gate,
        schema_watch,
    )
    .await
}

async fn consume_ollama_ndjson_lines<R>(
    reader: R,
    provider: &str,
    model: &str,
    delta_tx: DeltaSender,
    unload_grace: Duration,
    warmup_gate: &mut bool,
    mut schema_watch: Option<super::schema_stream::StreamSchemaWatch>,
) -> Result<LlmResult, VmError>
where
    R: tokio::io::AsyncBufRead + Unpin,
{
    use tokio::io::AsyncBufReadExt;

    let mut lines = reader.lines();

    let mut text = String::new();
    let mut input_tokens: i64 = 0;
    let mut output_tokens: i64 = 0;
    let mut result_model = model.to_string();

    let mut thinking_text = String::new();
    let mut tool_calls = Vec::new();
    let mut blocks = Vec::new();
    let mut saw_done = false;
    let mut saw_chunk = false;
    let mut done_reason: Option<String> = None;
    let mut telemetry = ProviderTelemetry::default();

    loop {
        let line = next_ollama_ndjson_line(&mut lines, model, unload_grace, warmup_gate)
            .await
            .map_err(|error| {
                VmError::Thrown(VmValue::String(std::sync::Arc::from(format!(
                    "ollama stream error: unexpected EOF or read failure before done=true: {error}"
                ))))
            })?;
        let Some(line) = line else {
            break;
        };
        *warmup_gate = true;
        if line.is_empty() {
            continue;
        }
        let data = line.strip_prefix("data: ").unwrap_or(&line);
        if data == "[DONE]" {
            break;
        }
        let json: serde_json::Value = serde_json::from_str(data).map_err(|error| {
            VmError::Thrown(VmValue::String(std::sync::Arc::from(format!(
                "ollama stream parse error: partial or invalid NDJSON frame before done=true: {error}; line={}",
                &data[..data.len().min(200)]
            ))))
        })?;
        saw_chunk = true;

        // Ollama streams content and thinking as separate channels for
        // reasoning-capable models (gemma3/4, qwen3, etc.); we always set
        // `think: true` so thinking tokens aren't dropped.
        let content = json["message"]["content"].as_str().unwrap_or("");
        let thinking = json["message"]["thinking"].as_str().unwrap_or("");
        if !content.is_empty() {
            text.push_str(content);
            let _ = delta_tx.send(content.to_string());
            blocks.push(
                serde_json::json!({"type": "output_text", "text": content, "visibility": "public"}),
            );
            if let Some(watch) = schema_watch.as_mut() {
                if let Some(abort) = watch.observe(content) {
                    return Err(abort.into_vm_error());
                }
            }
        } else if !thinking.is_empty() {
            thinking_text.push_str(thinking);
            let _ = delta_tx.send(thinking.to_string());
            blocks.push(
                serde_json::json!({"type": "reasoning", "text": thinking, "visibility": "private"}),
            );
        }
        append_ollama_tool_calls(&json["message"], &mut tool_calls, &mut blocks);

        if let Some(m) = json["model"].as_str() {
            result_model = m.to_string();
        }

        if json["done"].as_bool() == Some(true) {
            if let Some(n) = json["prompt_eval_count"].as_i64() {
                input_tokens = n;
            }
            if let Some(n) = json["eval_count"].as_i64() {
                output_tokens = n;
            }
            // Capture Ollama's `done_reason` so length-truncation is visible on
            // the most-used local chat path. Without this the agent never learns
            // it was cut off at the token cap.
            done_reason = json["done_reason"].as_str().map(str::to_string);
            telemetry = ProviderTelemetry::from_ollama_done(&json, telemetry_source::OLLAMA_CHAT);
            saw_done = true;
            break;
        }
    }

    if !saw_done {
        let suffix = if saw_chunk {
            " after partial content"
        } else {
            " before any chunks"
        };
        return Err(VmError::Thrown(VmValue::String(std::sync::Arc::from(
            format!("ollama stream error: unexpected EOF before done=true{suffix}"),
        ))));
    }

    let thinking = if thinking_text.is_empty() {
        None
    } else {
        Some(thinking_text.clone())
    };
    if text.is_empty() && !thinking_text.is_empty() {
        text = thinking_text;
    }

    // Guard against upstream parser bugs reporting generated tokens with
    // no visible content. Observed with `gemma4:26b` + ollama's
    // server-side `PARSER gemma4` on tool-heavy prompts: eval_count is
    // nonzero but every delta and the done chunk are empty strings.
    // Silently returning empty text would make the agent loop burn
    // iterations on a no-op.
    //
    // BUT: `done_reason == "length"` is a deterministic token-cap
    // truncation, not a parser bug. Re-running it just re-truncates, so we
    // must NOT raise the retryable parser-bug error here. Surface the empty
    // result carrying `stop_reason: Some("length")` so the caller sees a
    // non-retryable truncation signal instead. The parser-bug error stays
    // reserved for `done_reason` == stop/absent.
    if text.is_empty()
        && tool_calls.is_empty()
        && output_tokens > 0
        && done_reason.as_deref() != Some("length")
    {
        return Err(VmError::Thrown(VmValue::String(std::sync::Arc::from(format!(
            "ollama model {model} reported eval_count={output_tokens} but delivered no content or thinking [ollama_empty_content_parser_bug]"
        )))));
    }

    Ok(LlmResult {
        text,
        tool_calls,
        input_tokens,
        output_tokens,
        // Native Ollama `/api/chat` NDJSON done frames carry no cache field;
        // 0 is "unknown", not a real miss. The `/v1` shim on these hosts also
        // omits `prompt_tokens_details`, so routing to `/v1` is not a fix.
        // `cache_supported: false` keeps cost/telemetry from scoring a local
        // model as a 100% cache miss.
        cache_read_tokens: 0,
        cache_write_tokens: 0,
        cache_supported: false,
        model: result_model,
        // NDJSON is currently only consumed for Ollama's `/api/chat`, but
        // pass the caller-supplied provider through anyway so the result
        // matches `opts.provider` exactly. Future engines that adopt
        // NDJSON streaming (some llama.cpp builds, mlx-vlm) will get the
        // right label without additional plumbing.
        provider: provider.to_string(),
        thinking,
        thinking_summary: None,
        stop_reason: done_reason,
        // NDJSON (Ollama) has no accelerated-serving tier.
        served_fast: false,
        blocks,
        logprobs: Vec::new(),
        telemetry,
    })
}

async fn next_ollama_ndjson_line<R>(
    lines: &mut tokio::io::Lines<R>,
    model: &str,
    unload_grace: Duration,
    warmup_progress_sent: &mut bool,
) -> std::io::Result<Option<String>>
where
    R: tokio::io::AsyncBufRead + Unpin,
{
    if *warmup_progress_sent || unload_grace.is_zero() {
        return lines.next_line().await;
    }

    tokio::select! {
        line = lines.next_line() => line,
        _ = tokio::time::sleep(unload_grace) => {
            *warmup_progress_sent = true;
            emit_ollama_warmup_progress(model);
            lines.next_line().await
        }
    }
}

fn emit_ollama_warmup_progress(model: &str) {
    let message = format!("warming up Ollama model {model}");
    if let Some(bridge) = crate::llm::current_host_bridge() {
        bridge.send_progress(
            "llm",
            &message,
            None,
            None,
            Some(serde_json::json!({
                "provider": "ollama",
                "model": model,
                "reason": "model_unload",
            })),
        );
    }
    crate::events::log_info("llm", &message);
}

fn is_ollama_empty_content_parser_bug(err: &VmError) -> bool {
    match err {
        VmError::Thrown(VmValue::String(message)) => {
            message.contains("[ollama_empty_content_parser_bug]")
        }
        VmError::Runtime(message) => message.contains("[ollama_empty_content_parser_bug]"),
        _ => false,
    }
}

#[cfg(test)]
mod tests {
    use super::{
        append_ollama_tool_calls, consume_ollama_ndjson_lines, parse_ollama_tool_arguments,
        should_request_stream_usage, telemetry_source,
    };
    use std::time::Duration;

    #[test]
    fn stream_usage_requested_for_openai_compatible_endpoints() {
        assert!(should_request_stream_usage(
            false,
            false,
            "/chat/completions"
        ));
        assert!(should_request_stream_usage(
            false,
            true,
            "/v1/chat/completions"
        ));
        assert!(!should_request_stream_usage(false, true, "/api/chat"));
        assert!(!should_request_stream_usage(true, false, "/messages"));
    }

    #[test]
    fn ollama_tool_arguments_accept_object_shape() {
        let arguments = serde_json::json!({"path": "README.md"});
        assert_eq!(parse_ollama_tool_arguments(&arguments), arguments);
    }

    #[test]
    fn ollama_tool_arguments_parse_json_strings() {
        let parsed = parse_ollama_tool_arguments(&serde_json::json!("{\"path\":\"README.md\"}"));
        assert_eq!(parsed["path"], "README.md");
    }

    #[test]
    fn ollama_stream_chunks_surface_tool_calls() {
        let message = serde_json::json!({
            "role": "assistant",
            "content": "",
            "tool_calls": [{
                "function": {
                    "name": "read_file",
                    "arguments": {
                        "path": "README.md"
                    }
                }
            }]
        });
        let mut tool_calls = Vec::new();
        let mut blocks = Vec::new();
        append_ollama_tool_calls(&message, &mut tool_calls, &mut blocks);

        assert_eq!(tool_calls.len(), 1);
        assert_eq!(tool_calls[0]["name"], "read_file");
        assert_eq!(tool_calls[0]["arguments"]["path"], "README.md");
        assert_eq!(blocks.len(), 1);
        assert_eq!(blocks[0]["type"], "tool_call");
    }

    #[tokio::test]
    async fn ollama_ndjson_empty_content_eval_count_is_marked_for_retry() {
        let body = b"{\"message\":{\"role\":\"assistant\",\"content\":\"\"},\"done\":true,\"prompt_eval_count\":10,\"eval_count\":4}\n";
        let (tx, _rx) = tokio::sync::mpsc::unbounded_channel();
        let mut warmup_gate = false;
        let err = consume_ollama_ndjson_lines(
            &body[..],
            "ollama",
            "stub-model",
            tx,
            Duration::ZERO,
            &mut warmup_gate,
            None,
        )
        .await
        .expect_err("empty Ollama parser-bug response should fail");
        assert!(
            err.to_string()
                .contains("[ollama_empty_content_parser_bug]"),
            "err was: {err}"
        );
    }

    #[tokio::test]
    async fn ollama_ndjson_requires_done_frame() {
        let body =
            b"{\"message\":{\"role\":\"assistant\",\"content\":\"partial\"},\"done\":false}\n";
        let (tx, _rx) = tokio::sync::mpsc::unbounded_channel();
        let mut warmup_gate = false;
        let err = consume_ollama_ndjson_lines(
            &body[..],
            "ollama",
            "stub-model",
            tx,
            Duration::ZERO,
            &mut warmup_gate,
            None,
        )
        .await
        .expect_err("truncated Ollama stream should fail");
        assert!(err.to_string().contains("unexpected EOF before done=true"));
    }

    #[tokio::test]
    async fn ollama_ndjson_captures_server_telemetry_from_done_frame() {
        let body = b"{\"message\":{\"role\":\"assistant\",\"content\":\"ok\"},\"done\":false}\n\
            {\"message\":{\"role\":\"assistant\",\"content\":\"\"},\"done\":true,\
            \"model\":\"devstral-small-2:24b\",\
            \"total_duration\":4500000000,\"load_duration\":300000000,\
            \"prompt_eval_count\":42,\"prompt_eval_duration\":1100000000,\
            \"eval_count\":7,\"eval_duration\":3000000000}\n";
        let (tx, _rx) = tokio::sync::mpsc::unbounded_channel();
        let mut warmup_gate = false;
        let result = consume_ollama_ndjson_lines(
            &body[..],
            "ollama",
            "devstral-small-2:24b",
            tx,
            Duration::ZERO,
            &mut warmup_gate,
            None,
        )
        .await
        .expect("ollama stream parses");
        assert_eq!(result.telemetry.source, telemetry_source::OLLAMA_CHAT);
        assert_eq!(result.telemetry.server_total_ms, Some(4500));
        assert_eq!(result.telemetry.server_load_ms, Some(300));
        assert_eq!(result.telemetry.server_prompt_eval_ms, Some(1100));
        assert_eq!(result.telemetry.server_generation_ms, Some(3000));
        assert_eq!(result.telemetry.server_prompt_tokens, Some(42));
        assert_eq!(result.telemetry.server_output_tokens, Some(7));
        assert_eq!(
            result.telemetry.runtime_loaded_model.as_deref(),
            Some("devstral-small-2:24b")
        );
    }

    #[tokio::test]
    async fn ollama_ndjson_done_reason_stop_is_captured_as_stop_reason() {
        // Fix 1: a normal `done_reason:"stop"` frame surfaces stop_reason.
        let body = b"{\"message\":{\"role\":\"assistant\",\"content\":\"hi\"},\"done\":true,\
            \"done_reason\":\"stop\",\"prompt_eval_count\":5,\"eval_count\":2}\n";
        let (tx, _rx) = tokio::sync::mpsc::unbounded_channel();
        let mut warmup_gate = false;
        let result = consume_ollama_ndjson_lines(
            &body[..],
            "ollama",
            "stub-model",
            tx,
            Duration::ZERO,
            &mut warmup_gate,
            None,
        )
        .await
        .expect("ollama stream parses");
        assert_eq!(result.stop_reason.as_deref(), Some("stop"));
        // Fix 3: native Ollama reports no cache field — it is unsupported,
        // not a real 0% cache hit.
        assert!(!result.cache_supported);
    }

    #[tokio::test]
    async fn ollama_ndjson_done_reason_length_surfaces_truncation_not_parser_bug() {
        // Fix 1 + Fix 2: a length-capped frame with empty content must NOT
        // raise the retryable parser-bug error; instead it returns Ok carrying
        // stop_reason: Some("length") as a non-retryable truncation signal.
        let body = b"{\"message\":{\"role\":\"assistant\",\"content\":\"\"},\"done\":true,\
            \"done_reason\":\"length\",\"prompt_eval_count\":10,\"eval_count\":8}\n";
        let (tx, _rx) = tokio::sync::mpsc::unbounded_channel();
        let mut warmup_gate = false;
        let result = consume_ollama_ndjson_lines(
            &body[..],
            "ollama",
            "stub-model",
            tx,
            Duration::ZERO,
            &mut warmup_gate,
            None,
        )
        .await
        .expect("length truncation should return Ok, not the parser-bug error");
        assert_eq!(result.stop_reason.as_deref(), Some("length"));
        assert!(result.text.is_empty());
        assert!(result.tool_calls.is_empty());
    }

    #[tokio::test]
    async fn ollama_ndjson_empty_content_without_done_reason_is_still_parser_bug() {
        // Fix 2 guardrail: empty content with no `done_reason` (or stop) keeps
        // the retryable parser-bug behavior — only "length" is special-cased.
        let body = b"{\"message\":{\"role\":\"assistant\",\"content\":\"\"},\"done\":true,\
            \"done_reason\":\"stop\",\"prompt_eval_count\":10,\"eval_count\":4}\n";
        let (tx, _rx) = tokio::sync::mpsc::unbounded_channel();
        let mut warmup_gate = false;
        let err = consume_ollama_ndjson_lines(
            &body[..],
            "ollama",
            "stub-model",
            tx,
            Duration::ZERO,
            &mut warmup_gate,
            None,
        )
        .await
        .expect_err("empty content with done_reason=stop is still a parser bug");
        assert!(
            err.to_string()
                .contains("[ollama_empty_content_parser_bug]"),
            "err was: {err}"
        );
    }
}

#[cfg(test)]
mod streaming_tool_call_tests {
    //! Streaming-tool-call partial-arg event tests (#693). Drive
    //! [`consume_sse_lines`] against canned byte streams that mimic the
    //! Anthropic / OpenAI wire format and assert the
    //! `AgentEvent::ToolCall` / `AgentEvent::ToolCallUpdate` sequence
    //! lands as expected: an initial `Pending` announcement, one or
    //! more coalesced partial-arg updates, and `raw_input_partial`
    //! when the partial bytes can't be parsed as JSON yet.
    //!
    //! Events are captured via the global session-sink registry (which
    //! is what real ACP/A2A consumers use) rather than the per-loop
    //! thread-local — drive a fresh session id per test so they don't
    //! cross-talk under `cargo test`'s thread pool.

    use super::*;
    use crate::agent_events::{
        clear_session_sinks, register_sink, AgentEvent, AgentEventSink, ToolCallStatus,
    };
    use std::sync::{Arc, Mutex};

    struct CapturingSink {
        events: Arc<Mutex<Vec<AgentEvent>>>,
    }

    impl AgentEventSink for CapturingSink {
        fn handle_event(&self, event: &AgentEvent) {
            self.events
                .lock()
                .expect("capture mutex")
                .push(event.clone());
        }
    }

    fn install_capturing_sink(session_id: &str) -> Arc<Mutex<Vec<AgentEvent>>> {
        let events: Arc<Mutex<Vec<AgentEvent>>> = Arc::new(Mutex::new(Vec::new()));
        register_sink(
            session_id,
            Arc::new(CapturingSink {
                events: events.clone(),
            }),
        );
        events
    }

    /// Build a fresh per-test session id so concurrent test threads
    /// can't poison each other's captured-event vector via the global
    /// registry.
    fn fresh_session_id(label: &str) -> String {
        format!("{label}-{}", uuid::Uuid::now_v7())
    }

    /// Drive `consume_sse_lines` against a canned SSE byte buffer and
    /// return the captured agent events plus the parsed `LlmResult`.
    /// Helper so each test can stay focused on the assertion.
    async fn drive(
        bytes: &[u8],
        session_id: &str,
        is_anthropic: bool,
    ) -> (LlmResult, Vec<AgentEvent>) {
        let events = install_capturing_sink(session_id);
        let (delta_tx, _delta_rx) = tokio::sync::mpsc::unbounded_channel::<String>();
        let reader = tokio::io::BufReader::new(bytes);
        let result = consume_sse_lines(
            reader,
            if is_anthropic { "anthropic" } else { "openai" },
            "test-model",
            is_anthropic,
            delta_tx,
            Some(session_id),
            None,
            false,
        )
        .await
        .expect("sse parse should succeed");
        let captured = events.lock().expect("capture mutex").clone();
        (result, captured)
    }

    #[tokio::test(flavor = "current_thread")]
    async fn anthropic_stream_announces_tool_call_then_streams_partials() {
        // Force COALESCE_WINDOW pauses by emitting deltas across the
        // 50ms boundary so the coalescer flushes more than once.
        let body = concat!(
            "data: {\"type\":\"message_start\",\"message\":{\"usage\":{\"input_tokens\":3}}}\n",
            "data: {\"type\":\"content_block_start\",\"index\":0,\"content_block\":{\"type\":\"tool_use\",\"id\":\"toolu_a1\",\"name\":\"search_web\"}}\n",
            "data: {\"type\":\"content_block_delta\",\"index\":0,\"delta\":{\"type\":\"input_json_delta\",\"partial_json\":\"{\\\"q\\\":\\\"ant\"}}\n",
            "data: {\"type\":\"content_block_delta\",\"index\":0,\"delta\":{\"type\":\"input_json_delta\",\"partial_json\":\"hropic\"}}\n",
            "data: {\"type\":\"content_block_delta\",\"index\":0,\"delta\":{\"type\":\"input_json_delta\",\"partial_json\":\"\\\"}\"}}\n",
            "data: {\"type\":\"content_block_stop\",\"index\":0}\n",
            "data: {\"type\":\"message_delta\",\"usage\":{\"output_tokens\":5},\"delta\":{\"stop_reason\":\"tool_use\"}}\n",
            "data: [DONE]\n",
        );
        let session_id = fresh_session_id("anth-stream");
        let (result, events) = drive(body.as_bytes(), &session_id, true).await;

        // ── Initial ToolCall(Pending) announcement ──────────────────
        let announcements: Vec<&AgentEvent> = events
            .iter()
            .filter(|e| matches!(e, AgentEvent::ToolCall { .. }))
            .collect();
        assert_eq!(
            announcements.len(),
            1,
            "expected exactly one initial ToolCall(Pending), got {events:#?}"
        );
        match announcements[0] {
            AgentEvent::ToolCall {
                tool_name,
                status,
                tool_call_id,
                raw_input,
                ..
            } => {
                assert_eq!(tool_name, "search_web");
                assert_eq!(*status, ToolCallStatus::Pending);
                assert_eq!(tool_call_id, "tool-toolu_a1");
                assert_eq!(*raw_input, serde_json::json!({}));
            }
            _ => unreachable!(),
        }

        // ── Partial-arg ToolCallUpdate(Pending) updates fired before
        //    content_block_stop ─────────────────────────────────────
        let partial_updates: Vec<&AgentEvent> = events
            .iter()
            .filter(|e| {
                matches!(
                    e,
                    AgentEvent::ToolCallUpdate {
                        status: ToolCallStatus::Pending,
                        ..
                    }
                )
            })
            .collect();
        assert!(
            !partial_updates.is_empty(),
            "expected at least one Pending tool_call_update from streaming deltas, got {events:#?}"
        );
        // Some update must carry either a parsed `raw_input` or a
        // `raw_input_partial` so clients can render the args live.
        let has_payload = partial_updates.iter().any(|e| match e {
            AgentEvent::ToolCallUpdate {
                raw_input,
                raw_input_partial,
                ..
            } => raw_input.is_some() || raw_input_partial.is_some(),
            _ => false,
        });
        assert!(
            has_payload,
            "expected at least one Pending update to carry raw_input or raw_input_partial; got {partial_updates:#?}"
        );

        // ── Final tool call result was still parsed correctly ───────
        assert_eq!(result.tool_calls.len(), 1);
        assert_eq!(result.tool_calls[0]["name"], "search_web");
        assert_eq!(result.tool_calls[0]["arguments"]["q"], "anthropic");

        clear_session_sinks(&session_id);
    }

    #[tokio::test(flavor = "current_thread")]
    async fn anthropic_stream_emits_raw_input_partial_when_args_unparseable() {
        // The model emits an unterminated string before the close —
        // the recovery path can't synthesize a value, so the transport
        // must publish `raw_input_partial` carrying the raw bytes.
        let body = concat!(
            "data: {\"type\":\"content_block_start\",\"index\":0,\"content_block\":{\"type\":\"tool_use\",\"id\":\"toolu_b1\",\"name\":\"edit\"}}\n",
            "data: {\"type\":\"content_block_delta\",\"index\":0,\"delta\":{\"type\":\"input_json_delta\",\"partial_json\":\"{\\\"path\\\":\\\"foo.swift\\\",\\\"replace\\\":\\\"hello\"}}\n",
            "data: {\"type\":\"content_block_delta\",\"index\":0,\"delta\":{\"type\":\"input_json_delta\",\"partial_json\":\" world\\\"}\"}}\n",
            "data: {\"type\":\"content_block_stop\",\"index\":0}\n",
            "data: {\"type\":\"message_delta\",\"usage\":{\"output_tokens\":3},\"delta\":{\"stop_reason\":\"tool_use\"}}\n",
            "data: [DONE]\n",
        );
        let session_id = fresh_session_id("anth-raw-partial");
        let (_, events) = drive(body.as_bytes(), &session_id, true).await;

        // The first Pending update fires immediately after the first
        // delta. At that point the buffer contains an unterminated
        // string, so `raw_input_partial` should be set.
        let first_partial = events.iter().find_map(|e| match e {
            AgentEvent::ToolCallUpdate {
                status: ToolCallStatus::Pending,
                raw_input,
                raw_input_partial,
                ..
            } => Some((raw_input.clone(), raw_input_partial.clone())),
            _ => None,
        });
        let (first_value, first_raw) =
            first_partial.expect("expected at least one Pending tool_call_update during streaming");
        assert!(
            first_value.is_none() && first_raw.is_some(),
            "first partial must surface raw_input_partial when JSON isn't yet parseable; got value={first_value:?} raw={first_raw:?}"
        );
        assert!(
            first_raw.unwrap().contains("hello"),
            "raw_input_partial should carry the concatenated bytes verbatim"
        );

        clear_session_sinks(&session_id);
    }

    #[tokio::test(flavor = "current_thread")]
    async fn openai_stream_truncated_reasoning_does_not_leak_into_text() {
        // A reasoning model cut off mid-thought streams only `reasoning`
        // deltas with no committed `content`, then finish_reason="length".
        // The partial trace must NOT be promoted into `.text` (that garbage
        // would surface as the answer); it stays under `.thinking`.
        let body = concat!(
            "data: {\"choices\":[{\"index\":0,\"delta\":{\"reasoning\":\"Let me think step\"}}]}\n",
            "data: {\"choices\":[{\"index\":0,\"delta\":{\"reasoning\":\" by step about\"}}]}\n",
            "data: {\"choices\":[{\"index\":0,\"finish_reason\":\"length\",\"delta\":{}}],\"usage\":{\"prompt_tokens\":4,\"completion_tokens\":6}}\n",
            "data: [DONE]\n",
        );
        let session_id = fresh_session_id("oai-trunc");
        let (result, _events) = drive(body.as_bytes(), &session_id, false).await;

        assert_eq!(result.stop_reason.as_deref(), Some("length"));
        assert_eq!(
            result.text, "",
            "truncated reasoning leaked into visible text: {:?}",
            result.text
        );
        assert_eq!(
            result.thinking.as_deref(),
            Some("Let me think step by step about"),
            "partial reasoning trace must survive under thinking"
        );

        clear_session_sinks(&session_id);
    }

    #[tokio::test(flavor = "current_thread")]
    async fn openai_stream_reasoning_promotes_to_text_on_clean_stop() {
        // Control: the same reasoning-only stream on a clean finish
        // (finish_reason != "length") still promotes the trace into `.text`,
        // preserving behaviour for models that answer in the reasoning channel.
        let body = concat!(
            "data: {\"choices\":[{\"index\":0,\"delta\":{\"reasoning\":\"the answer\"}}]}\n",
            "data: {\"choices\":[{\"index\":0,\"delta\":{\"reasoning\":\" is 42\"}}]}\n",
            "data: {\"choices\":[{\"index\":0,\"finish_reason\":\"stop\",\"delta\":{}}],\"usage\":{\"prompt_tokens\":4,\"completion_tokens\":3}}\n",
            "data: [DONE]\n",
        );
        let session_id = fresh_session_id("oai-clean");
        let (result, _events) = drive(body.as_bytes(), &session_id, false).await;

        assert_eq!(result.stop_reason.as_deref(), Some("stop"));
        assert_eq!(result.text, "the answer is 42");
        assert_eq!(result.thinking.as_deref(), Some("the answer is 42"));

        clear_session_sinks(&session_id);
    }

    #[tokio::test(flavor = "current_thread")]
    async fn openai_stream_reasoning_does_not_leak_into_text_when_tool_call_present() {
        // gpt-oss / harmony streaming: the analysis channel arrives as
        // `reasoning` deltas, the model emits a tool call, and no `content`
        // delta is committed. The reasoning is intermediate chain-of-thought,
        // not a final answer — the tool call is the action. It must NOT be
        // promoted into `.text` (that would leak private CoT into the
        // user-facing message and into the transcript the eval grader mines);
        // it stays under `.thinking`.
        let body = concat!(
            "data: {\"choices\":[{\"index\":0,\"delta\":{\"reasoning\":\"We need to inspect\"}}]}\n",
            "data: {\"choices\":[{\"index\":0,\"delta\":{\"reasoning\":\" parser.rs first.\"}}]}\n",
            "data: {\"choices\":[{\"index\":0,\"delta\":{\"tool_calls\":[{\"index\":0,\"id\":\"call_a\",\"function\":{\"name\":\"look\",\"arguments\":\"{\\\"path\\\":\\\"parser.rs\\\"}\"}}]}}]}\n",
            "data: {\"choices\":[{\"index\":0,\"finish_reason\":\"tool_calls\",\"delta\":{}}],\"usage\":{\"prompt_tokens\":4,\"completion_tokens\":6}}\n",
            "data: [DONE]\n",
        );
        let session_id = fresh_session_id("oai-reason-toolcall");
        let (result, _events) = drive(body.as_bytes(), &session_id, false).await;

        assert_eq!(result.stop_reason.as_deref(), Some("tool_calls"));
        assert_eq!(
            result.text, "",
            "reasoning leaked into visible text on a tool-call turn: {:?}",
            result.text
        );
        assert_eq!(
            result.thinking.as_deref(),
            Some("We need to inspect parser.rs first."),
            "reasoning trace must survive under thinking"
        );
        assert_eq!(result.tool_calls.len(), 1);
        assert_eq!(result.tool_calls[0]["name"], "look");

        clear_session_sinks(&session_id);
    }

    #[tokio::test(flavor = "current_thread")]
    async fn openai_stream_tool_call_cut_by_length_keeps_finish_reason() {
        // burin-code#2121 evidence shape: the output-token cap cuts a native
        // tool call mid-arguments. OpenRouter delivers the truncated args
        // deltas and a final chunk that carries BOTH a `tool_calls` delta and
        // `finish_reason:"length"`. The accumulated args fail to parse (-> the
        // empty-object fallback the agent loop sees as an empty-args call),
        // and `stop_reason` MUST still surface so the empty-args feedback can
        // name the truncation cause.
        let body = concat!(
            "data: {\"choices\":[{\"index\":0,\"delta\":{\"tool_calls\":[{\"index\":0,\"id\":\"chatcmpl-tool-1\",\"function\":{\"name\":\"edit\",\"arguments\":\"{\\\"pa\"}}]}}]}\n",
            "data: {\"choices\":[{\"index\":0,\"finish_reason\":\"length\",\"delta\":{\"tool_calls\":[{\"index\":0,\"function\":{\"arguments\":\"th\\\":\\\"src/ma\"}}]}}],\"usage\":{\"prompt_tokens\":4,\"completion_tokens\":9}}\n",
            "data: [DONE]\n",
        );
        let session_id = fresh_session_id("oai-toolcall-length");
        let (result, _events) = drive(body.as_bytes(), &session_id, false).await;

        assert_eq!(result.stop_reason.as_deref(), Some("length"));
        assert_eq!(result.tool_calls.len(), 1);
        assert_eq!(result.tool_calls[0]["name"], "edit");
        assert_eq!(
            result.tool_calls[0]["arguments"],
            serde_json::json!({}),
            "truncated unparseable args fall back to the empty object"
        );

        clear_session_sinks(&session_id);
    }

    #[tokio::test(flavor = "current_thread")]
    async fn openai_stream_announces_and_streams_partials() {
        // OpenAI Chat-Completions-style: tool name on the first delta,
        // arguments string concatenated across subsequent deltas.
        let body = concat!(
            "data: {\"choices\":[{\"index\":0,\"delta\":{\"tool_calls\":[{\"index\":0,\"id\":\"call_a\",\"function\":{\"name\":\"read_file\",\"arguments\":\"{\\\"pa\"}}]}}]}\n",
            "data: {\"choices\":[{\"index\":0,\"delta\":{\"tool_calls\":[{\"index\":0,\"function\":{\"arguments\":\"th\\\":\\\"REA\"}}]}}]}\n",
            "data: {\"choices\":[{\"index\":0,\"delta\":{\"tool_calls\":[{\"index\":0,\"function\":{\"arguments\":\"DME.md\\\"}\"}}]}}]}\n",
            "data: {\"choices\":[{\"index\":0,\"finish_reason\":\"tool_calls\",\"delta\":{}}],\"usage\":{\"prompt_tokens\":4,\"completion_tokens\":3}}\n",
            "data: [DONE]\n",
        );
        let session_id = fresh_session_id("oai-stream");
        let (result, events) = drive(body.as_bytes(), &session_id, false).await;

        // Initial ToolCall(Pending) announcement on the first delta
        // that carries `function.name`.
        let announcements: Vec<&AgentEvent> = events
            .iter()
            .filter(|e| matches!(e, AgentEvent::ToolCall { .. }))
            .collect();
        assert_eq!(
            announcements.len(),
            1,
            "expected one initial ToolCall(Pending); got {events:#?}"
        );
        match announcements[0] {
            AgentEvent::ToolCall {
                tool_name,
                tool_call_id,
                status,
                ..
            } => {
                assert_eq!(tool_name, "read_file");
                assert_eq!(*status, ToolCallStatus::Pending);
                assert_eq!(tool_call_id, "tool-call_a");
            }
            _ => unreachable!(),
        }

        // At least one partial-arg ToolCallUpdate fired during streaming.
        let partial_updates = events
            .iter()
            .filter(|e| {
                matches!(
                    e,
                    AgentEvent::ToolCallUpdate {
                        status: ToolCallStatus::Pending,
                        ..
                    }
                )
            })
            .count();
        assert!(
            partial_updates >= 1,
            "expected at least one Pending tool_call_update during streaming; got {events:#?}"
        );

        // Final tool call result still surfaces the canonical args.
        assert_eq!(result.tool_calls.len(), 1);
        assert_eq!(result.tool_calls[0]["name"], "read_file");
        assert_eq!(result.tool_calls[0]["arguments"]["path"], "README.md");

        clear_session_sinks(&session_id);
    }

    #[tokio::test(flavor = "current_thread")]
    async fn no_session_id_means_no_streaming_events() {
        // Without an opt-in session id the transport must remain silent
        // — the dispatch-time lifecycle still owns the canonical events
        // for raw `llm_call(...)` invocations from script context.
        let body = concat!(
            "data: {\"type\":\"content_block_start\",\"index\":0,\"content_block\":{\"type\":\"tool_use\",\"id\":\"toolu_x\",\"name\":\"fake\"}}\n",
            "data: {\"type\":\"content_block_delta\",\"index\":0,\"delta\":{\"type\":\"input_json_delta\",\"partial_json\":\"{\\\"k\\\":1}\"}}\n",
            "data: {\"type\":\"content_block_stop\",\"index\":0}\n",
            "data: [DONE]\n",
        );
        let session_id = fresh_session_id("anth-silent");
        let events = install_capturing_sink(&session_id);
        let (delta_tx, _delta_rx) = tokio::sync::mpsc::unbounded_channel::<String>();
        let reader = tokio::io::BufReader::new(body.as_bytes());
        let _result = consume_sse_lines(
            reader,
            "anthropic",
            "test-model",
            true,
            delta_tx,
            None,
            None,
            false,
        )
        .await
        .expect("parse");
        let captured = events.lock().expect("capture mutex").clone();
        assert!(
            captured.is_empty(),
            "transport must emit no events when session_id is None; got {captured:#?}"
        );
        clear_session_sinks(&session_id);
    }

    #[tokio::test(flavor = "current_thread")]
    async fn coalescing_caps_event_count_under_burst() {
        // Build a burst of 20 input_json_deltas emitted in tight
        // succession (no real timer pauses inside the test). With
        // 50ms coalescing, only the very first delta should fire an
        // immediate ToolCallUpdate; the others should fold into it.
        let mut body = String::new();
        body.push_str(
            "data: {\"type\":\"content_block_start\",\"index\":0,\"content_block\":{\"type\":\"tool_use\",\"id\":\"toolu_burst\",\"name\":\"big\"}}\n",
        );
        for i in 0..20 {
            body.push_str(&format!(
                "data: {{\"type\":\"content_block_delta\",\"index\":0,\"delta\":{{\"type\":\"input_json_delta\",\"partial_json\":\"chunk{i} \"}}}}\n",
            ));
        }
        body.push_str("data: {\"type\":\"content_block_stop\",\"index\":0}\n");
        body.push_str("data: [DONE]\n");
        let session_id = fresh_session_id("anth-coalesce");
        let (_, events) = drive(body.as_bytes(), &session_id, true).await;

        // 20 deltas in well under 50ms must not produce 20 events.
        // We allow up to 3 (first delta + boundary races) so the test
        // tolerates clock jitter on busy CI hardware while still
        // guarding against the per-byte regression.
        let pending_updates = events
            .iter()
            .filter(|e| {
                matches!(
                    e,
                    AgentEvent::ToolCallUpdate {
                        status: ToolCallStatus::Pending,
                        ..
                    }
                )
            })
            .count();
        assert!(
            pending_updates < 20,
            "coalescing must cap the burst — got {pending_updates} pending updates from 20 deltas"
        );
        clear_session_sinks(&session_id);
    }
}

#[cfg(test)]
mod sse_read_error_tests {
    //! Mid-stream read failures must surface as transient errors, not as a
    //! silently truncated zero-token "success". The whole-request reqwest
    //! timeout (`resolve_timeout`, now fed by the model catalog's
    //! `stream_timeout`) materializes as exactly such a body-read error, so
    //! these tests drive `consume_sse_lines` with an erroring byte stream —
    //! the same shape `reqwest::Response::bytes_stream()` produces when the
    //! deadline fires mid-body.

    use super::*;

    /// Reader whose underlying stream yields some SSE bytes and then an
    /// io error, mimicking a reqwest total-timeout firing mid-stream.
    fn erroring_reader(
        head: &'static [u8],
        error: std::io::Error,
    ) -> tokio::io::BufReader<impl tokio::io::AsyncBufRead + Unpin> {
        let chunks: Vec<Result<&'static [u8], std::io::Error>> = vec![Ok(head), Err(error)];
        tokio::io::BufReader::new(tokio_util::io::StreamReader::new(tokio_stream::iter(
            chunks,
        )))
    }

    #[tokio::test]
    async fn mid_stream_timeout_surfaces_as_retryable_error_not_empty_success() {
        let reader = erroring_reader(
            b"data: {\"choices\":[{\"delta\":{\"content\":\"par\"}}]}\n",
            std::io::Error::new(std::io::ErrorKind::TimedOut, "operation timed out"),
        );
        let (delta_tx, _delta_rx) = tokio::sync::mpsc::unbounded_channel::<String>();
        let err = consume_sse_lines(
            reader,
            "openrouter",
            "test-model",
            false,
            delta_tx,
            None,
            None,
            false,
        )
        .await
        .expect_err("mid-stream read failure must not return a truncated success");
        let message = err.to_string();
        assert!(
            message.contains("openrouter stream error (mid-stream read)"),
            "message was: {message}"
        );
        assert!(
            crate::llm::agent_observe::is_retryable_llm_error(&err),
            "mid-stream timeout must classify as transient/retryable; message was: {message}"
        );
    }

    #[tokio::test]
    async fn mid_stream_connection_reset_is_also_retryable() {
        let reader = erroring_reader(
            b"data: {\"choices\":[{\"delta\":{\"content\":\"par\"}}]}\n",
            std::io::Error::new(
                std::io::ErrorKind::ConnectionReset,
                "connection reset by peer",
            ),
        );
        let (delta_tx, _delta_rx) = tokio::sync::mpsc::unbounded_channel::<String>();
        let err = consume_sse_lines(
            reader,
            "llamacpp",
            "test-model",
            false,
            delta_tx,
            None,
            None,
            false,
        )
        .await
        .expect_err("mid-stream reset must surface as an error");
        assert!(
            crate::llm::agent_observe::is_retryable_llm_error(&err),
            "mid-stream reset must classify as transient/retryable; got: {err}"
        );
    }

    #[tokio::test]
    async fn clean_eof_without_done_sentinel_still_returns_ok() {
        // Anthropic ends streams with `message_stop` (no `[DONE]`), and some
        // OpenAI-compatible servers close cleanly without the sentinel; a
        // clean EOF must keep returning the accumulated result.
        let body: &[u8] = concat!(
            "data: {\"choices\":[{\"delta\":{\"content\":\"hello\"}}]}\n",
            "data: {\"choices\":[{\"index\":0,\"finish_reason\":\"stop\",\"delta\":{}}],\"usage\":{\"prompt_tokens\":2,\"completion_tokens\":1}}\n",
        )
        .as_bytes();
        let (delta_tx, _delta_rx) = tokio::sync::mpsc::unbounded_channel::<String>();
        let reader = tokio::io::BufReader::new(body);
        let result = consume_sse_lines(
            reader,
            "openai",
            "test-model",
            false,
            delta_tx,
            None,
            None,
            false,
        )
        .await
        .expect("clean EOF without [DONE] parses");
        assert_eq!(result.text, "hello");
        assert_eq!(result.stop_reason.as_deref(), Some("stop"));
    }
}

#[cfg(test)]
mod schema_stream_abort_tests {
    //! End-to-end coverage for the streaming `output_schema` abort
    //! (`schema_stream_abort` — harn#1775). Drives [`consume_sse_lines`]
    //! against a canned OpenAI-shaped SSE body whose content delta
    //! immediately violates the schema, asserts:
    //!
    //! - the consumer returns a categorized `SchemaStreamAborted` error,
    //! - a `SchemaStreamAborted` transcript event is emitted, and
    //! - the `harn_llm_schema_stream_aborted_total` counter increments.
    //!
    //! Driven via `consume_sse_lines` rather than a full HTTP stack so
    //! the test stays deterministic and offline.
    use super::*;
    use crate::llm::api::StreamSchemaWatch;
    use crate::llm::trace::{peek_agent_trace, reset_agent_trace_state, AgentTraceEvent};
    use crate::value::ErrorCategory;
    use crate::{install_active_metrics_registry, MetricsRegistry};
    use std::sync::Arc;

    fn schema_with_int_age() -> serde_json::Value {
        serde_json::json!({
            "type": "object",
            "required": ["age"],
            "properties": {"age": {"type": "integer"}}
        })
    }

    fn build_payload(schema: serde_json::Value) -> crate::llm::api::LlmRequestPayload {
        let mut opts = crate::llm::api::options::base_opts("openai");
        opts.model = "gpt-test".to_string();
        opts.output_schema = Some(schema);
        opts.schema_stream_abort = true;
        opts.session_id = None;
        crate::llm::api::LlmRequestPayload::from(&opts)
    }

    #[tokio::test(flavor = "current_thread")]
    async fn openai_stream_aborts_on_impossible_property_type() {
        reset_agent_trace_state();
        let metrics = Arc::new(MetricsRegistry::default());
        install_active_metrics_registry(metrics.clone());

        let payload = build_payload(schema_with_int_age());
        let watch = StreamSchemaWatch::from_payload(&payload).expect("schema is canonicalizable");

        // First content delta opens `"age":"`, which is incompatible with
        // `age: int`. The mid-stream watch should fire on the trailing
        // string quote; the second delta is never read because we abort.
        let body = concat!(
            "data: {\"choices\":[{\"delta\":{\"content\":\"{\\\"age\\\":\"}}]}\n",
            "data: {\"choices\":[{\"delta\":{\"content\":\"\\\"twenty\"}}]}\n",
            "data: [DONE]\n",
        );
        let (delta_tx, _delta_rx) = tokio::sync::mpsc::unbounded_channel::<String>();
        let reader = tokio::io::BufReader::new(body.as_bytes());
        let err = consume_sse_lines(
            reader,
            "openai",
            "gpt-test",
            false,
            delta_tx,
            None,
            Some(watch),
            false,
        )
        .await
        .expect_err("schema abort must surface as error");

        match &err {
            VmError::CategorizedError { category, message } => {
                assert_eq!(*category, ErrorCategory::SchemaStreamAborted);
                assert!(
                    message.contains("$.age"),
                    "abort message should include JSON path; got: {message}"
                );
            }
            other => panic!("expected CategorizedError; got {other:?}"),
        }

        let events = peek_agent_trace();
        let aborts: Vec<_> = events
            .iter()
            .filter_map(|e| match e {
                AgentTraceEvent::SchemaStreamAborted {
                    provider,
                    model,
                    reason,
                    path,
                    chunks_consumed,
                } => Some((
                    provider.clone(),
                    model.clone(),
                    reason.clone(),
                    path.clone(),
                    *chunks_consumed,
                )),
                _ => None,
            })
            .collect();
        assert_eq!(
            aborts.len(),
            1,
            "expected exactly one SchemaStreamAborted event; got {events:#?}"
        );
        let (provider, model, _reason, path, chunks) = &aborts[0];
        assert_eq!(provider, "openai");
        assert_eq!(model, "gpt-test");
        assert_eq!(path, "$.age");
        assert!(*chunks >= 1);

        // Telemetry counter incremented through the installed registry.
        let rendered = metrics.render_prometheus();
        assert!(
            rendered.contains("harn_llm_schema_stream_aborted_total"),
            "metric family missing from prometheus render"
        );
        assert!(
            rendered.contains(
                "harn_llm_schema_stream_aborted_total{model=\"gpt-test\",provider=\"openai\"} 1"
            ),
            "expected labelled counter increment; got:\n{rendered}"
        );

        reset_agent_trace_state();
        crate::clear_active_metrics_registry();
    }

    #[tokio::test(flavor = "current_thread")]
    async fn opt_out_keeps_stream_alive_through_invalid_content() {
        reset_agent_trace_state();

        let mut payload = build_payload(schema_with_int_age());
        payload.schema_stream_abort = false;
        // With the watch disabled, even a clearly-invalid stream must
        // run to completion; only the post-hoc validator catches it.
        assert!(StreamSchemaWatch::from_payload(&payload).is_none());

        reset_agent_trace_state();
    }
}
