car_inference/

stream.rs

//! Streaming inference — SSE parsing for real-time token output.
//!
//! Supports streaming from OpenAI-compatible, Anthropic, and Google APIs.
//! Each provider uses Server-Sent Events (SSE) with different JSON schemas.

use crate::tasks::generate::ToolCall;
use crate::TokenUsage;
use std::collections::HashMap;

/// Events emitted during a streaming inference response.
///
/// `Serialize`/`Deserialize` are derived so the whole event stream can cross
/// the on-device inference-worker IPC boundary (car-releases#74) losslessly —
/// including the `StopReason` variant the daemon's hand-rolled WS-runner JSON
/// mapping omits. The representation is internal (same crate version on both
/// ends of the pipe), so the default externally-tagged enum shape is fine.
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
pub enum StreamEvent {
    /// Partial text token from the model.
    TextDelta(String),
    /// A tool call is starting (name known, arguments pending).
    ToolCallStart {
        name: String,
        index: usize,
        id: Option<String>,
    },
    /// Partial tool call arguments (JSON fragment).
    ToolCallDelta {
        index: usize,
        arguments_delta: String,
    },
    /// A provider snapshot of the tool-call arguments. Completed snapshots are
    /// authoritative; an early `output_item.added` snapshot is only a fallback
    /// when no deltas or completed item arrive.
    ToolCallArguments {
        index: usize,
        arguments: String,
        authoritative: bool,
    },
    /// Provider-reported cumulative token usage observed mid-stream.
    ///
    /// Anthropic emits this twice: once in `message_start` with the
    /// finalized `input_tokens` (plus a stub `output_tokens: 1`), and
    /// again in `message_delta` at end of stream with the real
    /// `output_tokens`. Consumers should prefer per-field monotonicity
    /// (see [`StreamAccumulator`]) rather than overwriting blindly.
    Usage {
        /// Uncached prompt tokens. For providers whose streamed usage reports
        /// the cached subset inside the prompt total (OpenAI), the parser
        /// subtracts it here so this stays the uncached portion — matching the
        /// non-streaming convention.
        input_tokens: u64,
        output_tokens: u64,
        /// Prompt-cache read (hit) tokens reported mid-stream (Anthropic
        /// `message_start.usage.cache_read_input_tokens`, OpenAI
        /// `prompt_tokens_details.cached_tokens`). `0` when uncached / absent.
        cache_read_input_tokens: u64,
        /// Prompt-cache write tokens reported mid-stream (Anthropic
        /// `cache_creation_input_tokens`). OpenAI has no write bucket. `0` when
        /// absent.
        cache_creation_input_tokens: u64,
    },
    /// Provider-reported termination reason, surfaced mid-stream as the raw
    /// provider string (OpenAI `finish_reason`, Anthropic `delta.stop_reason`).
    /// A value of `"length"`/`"max_tokens"` signals the output was cut off at
    /// the token cap. Captured by [`StreamAccumulator`] and returned from
    /// [`StreamAccumulator::finish_with_usage`].
    StopReason(String),
    /// Opaque provider output item that must be replayed verbatim on a later
    /// turn. The OpenAI Responses API emits completed reasoning items through
    /// `response.output_item.done`; CAR does not interpret or rewrite their
    /// `id`, `status`, `summary`, or `encrypted_content`.
    ProviderOutputItem(serde_json::Value),
    /// Provider or transport failure observed after the HTTP stream began.
    /// This is terminal and must never be interpreted as a normal stop/success.
    /// Messages are sanitized at the provider adapter before entering the event.
    Error(String),
    /// Sent immediately before an [`Self::Error`] whose cause the producer
    /// knows was not the model's — an account rejection, a spend limit, a
    /// content ruling. Not terminal and not forwarded to clients; it lets the
    /// outcome tracker attribute the error by type instead of re-reading its
    /// text. Consumers that do not track outcomes ignore it.
    Attribution(crate::outcome::UnattributedReason),
    /// The first event of a stream served by the on-device model CAR appended
    /// behind a remote-only chain, naming the model the request was routed to
    /// and why it did not serve — the streaming counterpart of
    /// [`crate::InferenceResult::local_last_resort`] plus its first
    /// [`crate::FallbackFrom`] hop (Parslee-ai/car#1439). Absent from every
    /// other stream. Not text: an accumulator ignores it, and each consumer
    /// carries it onto its own terminal frame with
    /// [`StreamSubstitution::annotate`].
    Substituted(StreamSubstitution),
    /// Stream is complete. Contains the final aggregated result.
    Done {
        text: String,
        tool_calls: Vec<ToolCall>,
    },
}

/// What [`StreamEvent::Substituted`] reports.
#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
pub struct StreamSubstitution {
    /// The installed on-device model that is serving the stream.
    pub model_id: String,
    /// The model the request was routed to — the first candidate the chain
    /// moved past. `None` only if no hop was recorded.
    pub substituted_from: Option<String>,
    /// Why [`Self::substituted_from`] did not serve.
    pub substitution_reason: Option<crate::FallbackReason>,
}

impl StreamSubstitution {
    /// Put the substitution on a consumer's terminal frame as
    /// `local_last_resort: true` plus `substituted_from` /
    /// `substitution_reason` — the field names the non-streaming surfaces
    /// already use, so one vocabulary covers every path.
    pub fn annotate(&self, frame: &mut serde_json::Value) {
        frame["local_last_resort"] = serde_json::json!(true);
        if let Some(from) = &self.substituted_from {
            frame["substituted_from"] = serde_json::json!(from);
        }
        if let Some(reason) = self.substitution_reason {
            frame["substitution_reason"] = serde_json::json!(reason);
        }
    }
}

/// Parse an OpenAI **Responses API** SSE event into `StreamEvent`s. Unlike
/// chat/completions (a single anonymous `data:` delta stream), the Responses
/// API tags each event with a typed `event:` line — `response.output_text.delta`
/// (text), `response.output_item.added` (a function_call starting),
/// `response.function_call_arguments.delta` (streamed tool-arg JSON), completed
/// argument snapshots, and a terminal `response.completed` /
/// `response.incomplete` carrying usage + status.
pub fn parse_openai_responses_sse_line(event_type: &str, data: &str) -> Vec<StreamEvent> {
    let json: serde_json::Value = match serde_json::from_str(data) {
        Ok(v) => v,
        Err(_) => return Vec::new(),
    };
    let mut events = Vec::new();
    match event_type {
        "response.output_text.delta" => {
            if let Some(d) = json.get("delta").and_then(|d| d.as_str()) {
                if !d.is_empty() {
                    events.push(StreamEvent::TextDelta(d.to_string()));
                }
            }
        }
        "response.output_item.added" | "response.output_item.done" => {
            if let Some(item) = json.get("item") {
                if item.get("type").and_then(|t| t.as_str()) == Some("function_call") {
                    let name = item
                        .get("name")
                        .and_then(|n| n.as_str())
                        .unwrap_or("")
                        .to_string();
                    let id = item
                        .get("call_id")
                        .or_else(|| item.get("id"))
                        .and_then(|i| i.as_str())
                        .map(|s| s.to_string());
                    let index = json
                        .get("output_index")
                        .and_then(|v| v.as_u64())
                        .unwrap_or(0) as usize;
                    // The call starts once, on `added`; `done` only completes
                    // its arguments. Re-emitting a start on `done` would put a
                    // second `tool_start` on every streaming client's wire.
                    if !name.is_empty() && event_type == "response.output_item.added" {
                        events.push(StreamEvent::ToolCallStart { name, index, id });
                    }
                    if let Some(arguments) = item
                        .get("arguments")
                        .and_then(|value| value.as_str())
                        .filter(|arguments| !arguments.is_empty())
                    {
                        events.push(StreamEvent::ToolCallArguments {
                            index,
                            arguments: arguments.to_string(),
                            authoritative: event_type == "response.output_item.done",
                        });
                    }
                } else if event_type == "response.output_item.done"
                    && item.get("type").and_then(|value| value.as_str()) == Some("reasoning")
                {
                    events.push(StreamEvent::ProviderOutputItem(item.clone()));
                }
            }
        }
        "response.function_call_arguments.delta" => {
            if let Some(d) = json.get("delta").and_then(|d| d.as_str()) {
                let index = json
                    .get("output_index")
                    .and_then(|v| v.as_u64())
                    .unwrap_or(0) as usize;
                events.push(StreamEvent::ToolCallDelta {
                    index,
                    arguments_delta: d.to_string(),
                });
            }
        }
        "response.function_call_arguments.done" => {
            if let Some(arguments) = json.get("arguments").and_then(|value| value.as_str()) {
                let index = json
                    .get("output_index")
                    .and_then(|value| value.as_u64())
                    .unwrap_or(0) as usize;
                events.push(StreamEvent::ToolCallArguments {
                    index,
                    arguments: arguments.to_string(),
                    authoritative: true,
                });
            }
        }
        "response.completed" | "response.incomplete" => {
            if let Some(resp) = json.get("response") {
                if let Some(u) = resp.get("usage") {
                    // OpenAI Responses: input_tokens is the TOTAL (cached
                    // included); subtract the cached subset so input_tokens
                    // stays the uncached portion (non-streaming convention).
                    let input_total = u.get("input_tokens").and_then(|v| v.as_u64()).unwrap_or(0);
                    let cached = u
                        .get("input_tokens_details")
                        .and_then(|d| d.get("cached_tokens"))
                        .and_then(|v| v.as_u64())
                        .unwrap_or(0)
                        .min(input_total);
                    events.push(StreamEvent::Usage {
                        input_tokens: input_total - cached,
                        output_tokens: u.get("output_tokens").and_then(|v| v.as_u64()).unwrap_or(0),
                        cache_read_input_tokens: cached,
                        cache_creation_input_tokens: 0,
                    });
                }
                if event_type == "response.incomplete" {
                    let reason = resp
                        .pointer("/incomplete_details/reason")
                        .and_then(|r| r.as_str())
                        .unwrap_or("incomplete");
                    events.push(StreamEvent::StopReason(reason.to_string()));
                    events.push(StreamEvent::Error(format!(
                        "managed inference incomplete: {reason}"
                    )));
                } else {
                    // A typed response.completed event is the only positive
                    // terminal proof accepted by the managed Responses path.
                    // `[DONE]`, transport EOF, deltas, and usage are not
                    // completion evidence.
                    events.push(StreamEvent::Done {
                        text: String::new(),
                        tool_calls: Vec::new(),
                    });
                }
            }
        }
        "error" | "response.failed" => {
            let pick = |field: &str| {
                json.pointer(&format!("/error/{field}"))
                    .or_else(|| json.pointer(&format!("/response/error/{field}")))
                    .and_then(|value| value.as_str())
                    .map(str::trim)
                    .filter(|value| !value.is_empty())
                    .map(str::to_string)
            };
            let message = pick("message");
            // `type` and `code` are CLASSIFICATION, not payload — and they are
            // the difference between "the gateway refused this content" and
            // "inference crashed". Dropping them left every failure looking
            // identical: a benchmark could not record a policy refusal as
            // distinct from a crash, a retry loop burned its budget retrying
            // into a decision that would never change, and an operator could
            // not tell a misconfiguration from a content ruling
            // (Parslee-ai/car#796).
            //
            // A bare "managed inference failed" is what a caller sees when the
            // gateway sends an error object with no message at all — so
            // reporting the fields we DO have matters most in exactly the case
            // that reads as least informative.
            let kind = pick("type");
            let code = pick("code");
            let mut detail = message.unwrap_or_else(|| "managed inference failed".to_string());
            let tags: Vec<String> = [("type", kind), ("code", code)]
                .into_iter()
                .filter_map(|(label, value)| value.map(|v| format!("{label}={v}")))
                .collect();
            if !tags.is_empty() {
                detail.push_str(&format!(" ({})", tags.join(", ")));
            }
            // Provider payloads, request IDs and stack details still stay out of
            // the cross-surface event: only the message and these two
            // classification fields are forwarded.
            events.push(StreamEvent::Error(detail));
        }
        _ => {}
    }
    events
}

/// Recover the `type` / `code` classification tags this module appends to a
/// gateway error message.
///
/// Deliberately adjacent to the code that writes them: #804 forwarded the tags
/// so a *consumer* could tell a policy refusal from a crash, but CAR itself is
/// also such a consumer — it decides whether to retry, and whether to blame the
/// model's health record. Reading them back out of the formatted string keeps
/// the tags on one wire type instead of threading a structured error through
/// every stream path; the cost is that writer and reader must agree, which is
/// why they live together and are tested against each other
/// (Parslee-ai/car#796).
pub fn error_tags(detail: &str) -> (Option<&str>, Option<&str>) {
    let Some(open) = detail.rfind(" (") else {
        return (None, None);
    };
    let Some(close) = detail[open..].rfind(')') else {
        return (None, None);
    };
    let mut kind = None;
    let mut code = None;
    for part in detail[open + 2..open + close].split(", ") {
        if let Some(v) = part.strip_prefix("type=") {
            kind = Some(v);
        } else if let Some(v) = part.strip_prefix("code=") {
            code = Some(v);
        }
    }
    (kind, code)
}

/// Classify a gateway error as a **content refusal**, from the `type`/`code`
/// tags #804 forwards.
///
/// Lives beside [`error_tags`], which reads what this module's SSE writer
/// appends, because those three are one contract: writer, tag reader, and the
/// verdict derived from the tags. It is `pub` because the daemon needs the same
/// verdict — a refusal that arrives mid-stream reaches `car-server-core` as
/// flattened `StreamEvent::Error` text rather than a typed `InferenceError`, and
/// re-deriving "was this a refusal?" there with a fresh substring rule would be
/// a second, drifting definition of the same thing (Parslee-ai/car#796).
///
/// Matched on the classification fields, never the prose. Deliberately NARROW:
/// over-classifying is the more dangerous direction, because a real inference
/// failure mislabelled as a refusal is excluded from the model's health record
/// and silently stops being retried — a crash that looks like a policy decision
/// is harder to find than a policy decision that looks like a crash.
///
/// The matched set is what a filter calls itself across the providers CAR has
/// seen: `content_policy_violation` (OpenAI-family, and the value in the remote
/// module's own fixtures), plus the `content_filter` / `moderation` / `safety`
/// families. #796's own gateway values are NOT yet known — the issue asks for a
/// re-run to discover them — so this list is expected to grow, and the test
/// pins the over-classification boundary rather than the exact membership.
pub fn content_refusal_tags(detail: &str) -> Option<(Option<String>, Option<String>)> {
    let (kind, code) = error_tags(detail);
    let refused = |v: &str| {
        let v = v.to_ascii_lowercase();
        v.contains("content_policy")
            || v.contains("content_filter")
            || v.contains("moderation")
            || v.contains("safety")
    };
    (code.is_some_and(refused) || kind.is_some_and(refused))
        .then(|| (kind.map(str::to_string), code.map(str::to_string)))
}

/// Output tokens from a Gemini `usageMetadata`: the response's
/// `candidatesTokenCount` plus the thinking model's `thoughtsTokenCount`.
///
/// Gemini reports thinking separately from the candidates — its
/// `totalTokenCount` is prompt + candidates + thoughts — and bills thoughts as
/// output, so reading `candidatesTokenCount` alone dropped every reasoning
/// token from a thinking model's output count (car#1765). Shared by the
/// streaming and non-streaming Google parsers so the two cannot disagree.
pub(crate) fn google_output_tokens(usage: &serde_json::Value) -> u64 {
    let count = |field: &str| usage.get(field).and_then(|v| v.as_u64()).unwrap_or(0);
    count("candidatesTokenCount") + count("thoughtsTokenCount")
}

/// Parse one Gemini `:streamGenerateContent?alt=sse` chunk (the JSON after
/// `data:`) into `StreamEvent`s. Each chunk is a `GenerateContentResponse`:
/// `candidates[0].content.parts[]` carry `text` deltas and/or complete
/// `functionCall`s (Gemini sends the whole call in one chunk, not incrementally
/// — so emit a `ToolCallStart` plus a single `ToolCallDelta` with the full
/// args), `usageMetadata` carries token counts, and `finishReason` (terminal
/// chunk) maps to `StopReason`.
pub fn parse_google_sse_line(data: &str) -> Vec<StreamEvent> {
    let json: serde_json::Value = match serde_json::from_str(data) {
        Ok(v) => v,
        Err(_) => return Vec::new(),
    };
    let mut events = Vec::new();
    if let Some(parts) = json
        .pointer("/candidates/0/content/parts")
        .and_then(|p| p.as_array())
    {
        for (i, part) in parts.iter().enumerate() {
            if let Some(t) = part.get("text").and_then(|t| t.as_str()) {
                if !t.is_empty() {
                    events.push(StreamEvent::TextDelta(t.to_string()));
                }
            }
            if let Some(fc) = part.get("functionCall") {
                let name = fc
                    .get("name")
                    .and_then(|n| n.as_str())
                    .unwrap_or("")
                    .to_string();
                if !name.is_empty() {
                    let args = fc
                        .get("args")
                        .map(|a| a.to_string())
                        .unwrap_or_else(|| "{}".to_string());
                    events.push(StreamEvent::ToolCallStart {
                        name,
                        index: i,
                        id: None,
                    });
                    events.push(StreamEvent::ToolCallDelta {
                        index: i,
                        arguments_delta: args,
                    });
                }
            }
        }
    }
    if let Some(u) = json.get("usageMetadata") {
        events.push(StreamEvent::Usage {
            input_tokens: u
                .get("promptTokenCount")
                .and_then(|v| v.as_u64())
                .unwrap_or(0),
            output_tokens: google_output_tokens(u),
            // Google prompt caching is not parsed by CAR.
            cache_read_input_tokens: 0,
            cache_creation_input_tokens: 0,
        });
    }
    if let Some(fr) = json
        .pointer("/candidates/0/finishReason")
        .and_then(|r| r.as_str())
    {
        events.push(StreamEvent::StopReason(fr.to_string()));
    }
    events
}

/// Parse a single SSE data line from an OpenAI-compatible streaming response.
/// Returns all events found in the line (supports multiple tool calls per chunk).
pub fn parse_openai_sse_line(line: &str) -> Vec<StreamEvent> {
    let data = match line.strip_prefix("data: ") {
        Some(d) => d,
        None => return Vec::new(),
    };
    if data == "[DONE]" {
        return Vec::new();
    }

    let json: serde_json::Value = match serde_json::from_str(data) {
        Ok(v) => v,
        Err(_) => return Vec::new(),
    };

    let mut events = Vec::new();

    // choices[0].finish_reason — set on the terminal content chunk
    // (e.g. "stop", "length", "tool_calls"). Surface it so truncation
    // ("length") is observable on the streaming path.
    if let Some(reason) = json
        .get("choices")
        .and_then(|c| c.as_array())
        .and_then(|c| c.first())
        .and_then(|c| c.get("finish_reason"))
        .and_then(|r| r.as_str())
    {
        if !reason.is_empty() {
            events.push(StreamEvent::StopReason(reason.to_string()));
        }
    }

    // choices[].delta — present on every text/tool chunk, absent on the
    // final usage-only chunk when `stream_options.include_usage=true`.
    if let Some(delta) = json
        .get("choices")
        .and_then(|c| c.as_array())
        .and_then(|c| c.first())
        .and_then(|c| c.get("delta"))
    {
        if let Some(content) = delta.get("content").and_then(|c| c.as_str()) {
            if !content.is_empty() {
                events.push(StreamEvent::TextDelta(content.to_string()));
            }
        }

        // Tool calls — collect ALL tool call events from this chunk
        if let Some(tool_calls) = delta.get("tool_calls").and_then(|t| t.as_array()) {
            for tc in tool_calls {
                let index = tc.get("index").and_then(|i| i.as_u64()).unwrap_or(0) as usize;
                if let Some(function) = tc.get("function") {
                    if let Some(name) = function.get("name").and_then(|n| n.as_str()) {
                        let id = tc.get("id").and_then(|i| i.as_str()).map(|s| s.to_string());
                        events.push(StreamEvent::ToolCallStart {
                            name: name.to_string(),
                            index,
                            id,
                        });
                    }
                    if let Some(args) = function.get("arguments").and_then(|a| a.as_str()) {
                        if !args.is_empty() {
                            events.push(StreamEvent::ToolCallDelta {
                                index,
                                arguments_delta: args.to_string(),
                            });
                        }
                    }
                }
            }
        }
    }

    // OpenAI sends real usage only when the request sets
    // `stream_options.include_usage=true`; it arrives in a final chunk
    // with `"choices": []` and a top-level `"usage"` object.
    if let Some(usage) = json.get("usage") {
        let input = usage
            .get("prompt_tokens")
            .and_then(|n| n.as_u64())
            .unwrap_or(0);
        let output = usage
            .get("completion_tokens")
            .and_then(|n| n.as_u64())
            .unwrap_or(0);
        // prompt_tokens is the TOTAL (cached included); subtract the cached
        // subset so input_tokens stays uncached (non-streaming convention).
        let cached = usage
            .get("prompt_tokens_details")
            .and_then(|d| d.get("cached_tokens"))
            .and_then(|n| n.as_u64())
            .unwrap_or(0)
            .min(input);
        if input != 0 || output != 0 {
            events.push(StreamEvent::Usage {
                input_tokens: input - cached,
                output_tokens: output,
                cache_read_input_tokens: cached,
                cache_creation_input_tokens: 0,
            });
        }
    }

    events
}

/// Parse a single SSE data line from an Anthropic streaming response.
pub fn parse_anthropic_sse_line(event_type: &str, data: &str) -> Vec<StreamEvent> {
    match event_type {
        "content_block_delta" => {
            let json: serde_json::Value = match serde_json::from_str(data) {
                Ok(v) => v,
                Err(_) => return Vec::new(),
            };
            let delta = match json.get("delta") {
                Some(d) => d,
                None => return Vec::new(),
            };
            let delta_type = match delta.get("type").and_then(|t| t.as_str()) {
                Some(t) => t,
                None => return Vec::new(),
            };

            match delta_type {
                "text_delta" => match delta.get("text").and_then(|t| t.as_str()) {
                    Some(text) => vec![StreamEvent::TextDelta(text.to_string())],
                    None => Vec::new(),
                },
                "input_json_delta" => match delta.get("partial_json").and_then(|p| p.as_str()) {
                    Some(partial) => {
                        let index =
                            json.get("index").and_then(|i| i.as_u64()).unwrap_or(0) as usize;
                        vec![StreamEvent::ToolCallDelta {
                            index,
                            arguments_delta: partial.to_string(),
                        }]
                    }
                    None => Vec::new(),
                },
                _ => Vec::new(),
            }
        }
        "content_block_start" => {
            let json: serde_json::Value = match serde_json::from_str(data) {
                Ok(v) => v,
                Err(_) => return Vec::new(),
            };
            let block = match json.get("content_block") {
                Some(b) => b,
                None => return Vec::new(),
            };
            if block.get("type").and_then(|t| t.as_str()) == Some("tool_use") {
                if let Some(name) = block.get("name").and_then(|n| n.as_str()) {
                    let index = json.get("index").and_then(|i| i.as_u64()).unwrap_or(0) as usize;
                    let id = block
                        .get("id")
                        .and_then(|i| i.as_str())
                        .map(|s| s.to_string());
                    return vec![StreamEvent::ToolCallStart {
                        name: name.to_string(),
                        index,
                        id,
                    }];
                }
            }
            Vec::new()
        }
        // Beginning of the response — Anthropic reports the finalized
        // `input_tokens` here along with a stub `output_tokens: 1`.
        // Shape: `{"message":{"usage":{"input_tokens":123,"output_tokens":1}}}`
        "message_start" => {
            let json: serde_json::Value = match serde_json::from_str(data) {
                Ok(v) => v,
                Err(_) => return Vec::new(),
            };
            let Some(usage) = json.pointer("/message/usage") else {
                return Vec::new();
            };
            let input = usage
                .get("input_tokens")
                .and_then(|n| n.as_u64())
                .unwrap_or(0);
            let output = usage
                .get("output_tokens")
                .and_then(|n| n.as_u64())
                .unwrap_or(0);
            // Anthropic reports the cache split in message_start's usage too —
            // input_tokens is the uncached prefix; the cached portion is here.
            let cache_read = usage
                .get("cache_read_input_tokens")
                .and_then(|n| n.as_u64())
                .unwrap_or(0);
            let cache_creation = usage
                .get("cache_creation_input_tokens")
                .and_then(|n| n.as_u64())
                .unwrap_or(0);
            if input == 0 && output == 0 && cache_read == 0 && cache_creation == 0 {
                return Vec::new();
            }
            vec![StreamEvent::Usage {
                input_tokens: input,
                output_tokens: output,
                cache_read_input_tokens: cache_read,
                cache_creation_input_tokens: cache_creation,
            }]
        }
        // End of the response — Anthropic reports the final
        // `output_tokens` here (input is already known from
        // `message_start`). Shape: `{"usage":{"output_tokens":456}}`.
        "message_delta" => {
            let json: serde_json::Value = match serde_json::from_str(data) {
                Ok(v) => v,
                Err(_) => return Vec::new(),
            };
            let mut events = Vec::new();
            // Anthropic carries the termination reason in `delta.stop_reason`
            // on this terminal event (`end_turn`, `tool_use`, `max_tokens`, …).
            if let Some(reason) = json.pointer("/delta/stop_reason").and_then(|r| r.as_str()) {
                if !reason.is_empty() {
                    events.push(StreamEvent::StopReason(reason.to_string()));
                }
            }
            if let Some(usage) = json.get("usage") {
                let input = usage
                    .get("input_tokens")
                    .and_then(|n| n.as_u64())
                    .unwrap_or(0);
                let output = usage
                    .get("output_tokens")
                    .and_then(|n| n.as_u64())
                    .unwrap_or(0);
                if input != 0 || output != 0 {
                    events.push(StreamEvent::Usage {
                        input_tokens: input,
                        output_tokens: output,
                        // Cache tokens arrive in message_start, not here; the
                        // accumulator's per-field max preserves them.
                        cache_read_input_tokens: 0,
                        cache_creation_input_tokens: 0,
                    });
                }
            }
            events
        }
        _ => Vec::new(),
    }
}

/// Accumulator for building the final result from stream events.
#[derive(Default)]
pub struct StreamAccumulator {
    pub text: String,
    tool_names: HashMap<usize, String>,
    tool_args: HashMap<usize, String>,
    tool_args_fallback: HashMap<usize, String>,
    tool_args_authoritative: HashMap<usize, String>,
    tool_ids: HashMap<usize, String>,
    /// Highest `input_tokens` value seen in a `Usage` event. Anthropic
    /// only sends this on `message_start`; other providers may send it
    /// multiple times and we keep the largest as the authoritative
    /// count.
    input_tokens: u64,
    /// Highest `output_tokens` value seen in a `Usage` event. For
    /// Anthropic this grows from the `message_start` stub (`1`) to the
    /// final count in `message_delta`, so we track monotonically.
    output_tokens: u64,
    /// Highest prompt-cache read tokens seen in a `Usage` event.
    cache_read_input_tokens: u64,
    /// Highest prompt-cache write tokens seen in a `Usage` event.
    cache_creation_input_tokens: u64,
    /// Whether any `Usage` event was observed. `false` means the
    /// provider never reported usage (e.g. OpenAI without
    /// `stream_options.include_usage=true`) and [`finish_with_usage`]
    /// should return `None`.
    saw_usage: bool,
    /// Provider-reported termination reason, captured from the last
    /// [`StreamEvent::StopReason`] seen. `None` until one arrives.
    stop_reason: Option<String>,
    /// Opaque provider output items observed in stream order.
    provider_output_items: Vec<serde_json::Value>,
}

impl StreamAccumulator {
    pub fn push(&mut self, event: &StreamEvent) {
        match event {
            StreamEvent::TextDelta(t) => self.text.push_str(t),
            StreamEvent::ToolCallStart { name, index, id } => {
                self.tool_names.insert(*index, name.clone());
                self.tool_args.entry(*index).or_default();
                if let Some(id) = id {
                    self.tool_ids.insert(*index, id.clone());
                }
            }
            StreamEvent::ToolCallDelta {
                index,
                arguments_delta,
            } => {
                self.tool_args
                    .entry(*index)
                    .or_default()
                    .push_str(arguments_delta);
            }
            StreamEvent::ToolCallArguments {
                index,
                arguments,
                authoritative,
            } => {
                let destination = if *authoritative {
                    &mut self.tool_args_authoritative
                } else {
                    &mut self.tool_args_fallback
                };
                destination.insert(*index, arguments.clone());
            }
            StreamEvent::Usage {
                input_tokens,
                output_tokens,
                cache_read_input_tokens,
                cache_creation_input_tokens,
            } => {
                self.saw_usage = true;
                // Per-field max: Anthropic's `message_start` carries
                // real input + cache tokens + stub output=1; `message_delta`
                // carries only final output. Neither event should be allowed
                // to clobber the other's authoritative value. Cache tokens
                // arrive once (message_start / final OpenAI chunk), so max
                // preserves them across the otherwise-zero deltas.
                if *input_tokens > self.input_tokens {
                    self.input_tokens = *input_tokens;
                }
                if *output_tokens > self.output_tokens {
                    self.output_tokens = *output_tokens;
                }
                if *cache_read_input_tokens > self.cache_read_input_tokens {
                    self.cache_read_input_tokens = *cache_read_input_tokens;
                }
                if *cache_creation_input_tokens > self.cache_creation_input_tokens {
                    self.cache_creation_input_tokens = *cache_creation_input_tokens;
                }
            }
            StreamEvent::StopReason(reason) => {
                self.stop_reason = Some(reason.clone());
            }
            StreamEvent::ProviderOutputItem(item) => {
                self.provider_output_items.push(item.clone());
            }
            StreamEvent::Error(_) | StreamEvent::Attribution(_) | StreamEvent::Substituted(_) => {}
            StreamEvent::Done { .. } => {}
        }
    }

    pub fn finish(self) -> (String, Vec<ToolCall>) {
        let (text, tool_calls, _, _) = self.finish_with_usage();
        (text, tool_calls)
    }

    /// Like [`finish`](crate::stream::StreamAccumulator::finish) but also returns the accumulated [`TokenUsage`]
    /// when the provider reported any (and the provider-reported
    /// `stop_reason` when one arrived). Returns `None` for usage if no
    /// `Usage` event was observed — callers can fall back to their
    /// own estimator. The 4th element is the raw provider stop_reason
    /// (`None` if the provider didn't report one) — see
    /// [`crate::InferenceResult::was_truncated`].
    pub fn finish_with_usage(self) -> (String, Vec<ToolCall>, Option<TokenUsage>, Option<String>) {
        let (text, tool_calls, usage, stop_reason, _) = self.finish_with_provider_output_items();
        (text, tool_calls, usage, stop_reason)
    }

    /// Like [`finish_with_usage`](crate::stream::StreamAccumulator::finish_with_usage) but also returns opaque provider output
    /// items in the order the Responses stream emitted them.
    pub fn finish_with_provider_output_items(
        self,
    ) -> (
        String,
        Vec<ToolCall>,
        Option<TokenUsage>,
        Option<String>,
        Vec<serde_json::Value>,
    ) {
        let mut tool_calls = Vec::new();
        let mut indices: Vec<usize> = self.tool_names.keys().copied().collect();
        indices.sort();

        for idx in indices {
            let id = self.tool_ids.get(&idx).cloned();
            let name = self.tool_names.get(&idx).cloned().unwrap_or_default();
            let args_str = self
                .tool_args_authoritative
                .get(&idx)
                .or_else(|| {
                    self.tool_args
                        .get(&idx)
                        .filter(|arguments| !arguments.is_empty())
                })
                .or_else(|| self.tool_args_fallback.get(&idx))
                .cloned()
                .unwrap_or_default();
            let arguments: HashMap<String, serde_json::Value> = if args_str.trim().is_empty() {
                HashMap::new()
            } else {
                match serde_json::from_str(&args_str) {
                    Ok(arguments) => arguments,
                    Err(error) => {
                        tracing::warn!(
                            tool = %name,
                            output_index = idx,
                            %error,
                            "tool-call arguments were not valid JSON; returning an empty argument map"
                        );
                        HashMap::new()
                    }
                }
            };
            tool_calls.push(ToolCall {
                id,
                name,
                arguments,
            });
        }

        let usage = if self.saw_usage {
            Some(TokenUsage {
                prompt_tokens: self.input_tokens,
                completion_tokens: self.output_tokens,
                total_tokens: self.input_tokens + self.output_tokens,
                // Context-window sizing comes from model metadata, not
                // per-response usage — leave it zero and let the
                // caller populate it if needed.
                context_window: 0,
                // Cache buckets decoded from the streamed usage events, so
                // streamed calls price cache the same as non-streaming.
                cache_read_input_tokens: self.cache_read_input_tokens,
                cache_creation_input_tokens: self.cache_creation_input_tokens,
            })
        } else {
            None
        };

        // In-process (MLX/candle) streams have no structured tool-call channel:
        // the local model emits `<tool_call>{…}</tool_call>` tags inline in the
        // text. Recover them into structured tool_calls and strip the tags from
        // the visible text. Remote streams already produced structured
        // tool_calls (and carry no tags), so this is a no-op for them.
        let (text, tag_calls) = crate::tasks::generate::parse_tool_calls(&self.text);
        let (text, tool_calls) = if tool_calls.is_empty() && !tag_calls.is_empty() {
            (text, tag_calls)
        } else {
            // Keep structured calls; still drop any stray tags from the text.
            (text, tool_calls)
        };

        (
            text,
            tool_calls,
            usage,
            self.stop_reason,
            self.provider_output_items,
        )
    }
}

/// Parse SSE lines from a raw byte stream. Handles both OpenAI and Anthropic formats.
/// Returns (event_type, data) pairs. OpenAI doesn't send event types (always "message").
pub fn parse_sse_lines(chunk: &str) -> Vec<(String, String)> {
    let mut events = Vec::new();
    let mut current_event = String::new();
    let mut current_data = String::new();

    for line in chunk.lines() {
        if let Some(rest) = line.strip_prefix("event: ") {
            current_event = rest.to_string();
        } else if let Some(rest) = line.strip_prefix("data: ") {
            current_data = rest.to_string();
        } else if line.is_empty() && !current_data.is_empty() {
            events.push((
                if current_event.is_empty() {
                    "message".to_string()
                } else {
                    current_event.clone()
                },
                current_data.clone(),
            ));
            current_event.clear();
            current_data.clear();
        }
    }

    // Handle case where stream doesn't end with empty line
    if !current_data.is_empty() {
        events.push((
            if current_event.is_empty() {
                "message".to_string()
            } else {
                current_event
            },
            current_data,
        ));
    }

    events
}

#[cfg(test)]
mod tests {
    use super::*;

    /// A runner's reported usage must survive accumulation (#795).
    ///
    /// The delegated-inference path called `finish()`, which drops the usage
    /// `finish_with_usage()` returns — so a runner that DID report counts had
    /// them collected and then discarded, and every delegated call surfaced
    /// `usage: null`. A consumer summing `total_tokens` read a silent zero,
    /// which is worse than an error because it looks like a valid answer.
    #[test]
    fn accumulated_usage_and_stop_reason_survive_finish() {
        let mut acc = StreamAccumulator::default();
        acc.push(&StreamEvent::TextDelta("hello".into()));
        acc.push(&StreamEvent::Usage {
            input_tokens: 28,
            output_tokens: 5,
            cache_read_input_tokens: 0,
            cache_creation_input_tokens: 0,
        });
        acc.push(&StreamEvent::StopReason("length".into()));

        let (text, _tools, usage, stop) = acc.finish_with_usage();
        assert_eq!(text, "hello");
        let usage = usage.expect("a reported Usage event must not be dropped");
        assert_eq!(usage.prompt_tokens, 28);
        assert_eq!(usage.completion_tokens, 5);
        assert_eq!(usage.total_tokens, 33);
        assert_eq!(
            stop.as_deref(),
            Some("length"),
            "the provider stop_reason feeds was_truncated and was being dropped too"
        );
    }

    /// No usage event means `None`, not a fabricated zero. CAR cannot know a
    /// foreign runner's tokenization, and inventing 0 is the bug being fixed.
    #[test]
    fn absent_usage_stays_none_rather_than_zero() {
        let mut acc = StreamAccumulator::default();
        acc.push(&StreamEvent::TextDelta("hi".into()));
        let (_text, _tools, usage, stop) = acc.finish_with_usage();
        assert!(
            usage.is_none(),
            "no Usage event must yield None so callers can fall back to an estimator"
        );
        assert!(stop.is_none());
    }

    /// A gateway error must carry its CLASSIFICATION, not just its prose.
    ///
    /// `type` / `code` are what let a consumer tell "refused by policy" from
    /// "inference crashed" — a benchmark scoring a refusal, a retry loop
    /// declining to retry a decision that will not change, an operator telling
    /// a misconfiguration from a content ruling. CAR forwarded only `message`,
    /// so all three were indistinguishable (Parslee-ai/car#796).
    #[test]
    fn managed_error_events_carry_type_and_code() {
        let events = parse_openai_responses_sse_line(
            "error",
            r#"{"error":{"message":"content refused","type":"invalid_request_error","code":"content_policy_violation"}}"#,
        );
        let StreamEvent::Error(msg) = events.first().expect("an error event") else {
            panic!("expected StreamEvent::Error, got {:?}", events.first());
        };
        assert!(msg.contains("content refused"), "message dropped: {msg}");
        assert!(
            msg.contains("type=invalid_request_error"),
            "type dropped: {msg}"
        );
        assert!(
            msg.contains("code=content_policy_violation"),
            "code dropped: {msg}"
        );
    }

    /// The worst case is an error object with NO message: the caller saw a bare
    /// "managed inference failed" and nothing else. Whatever classification the
    /// gateway did send must still come through.
    #[test]
    fn a_messageless_managed_error_still_reports_its_code() {
        let events = parse_openai_responses_sse_line(
            "response.failed",
            r#"{"response":{"error":{"code":"content_filter"}}}"#,
        );
        let StreamEvent::Error(msg) = events.first().expect("an error event") else {
            panic!("expected StreamEvent::Error");
        };
        assert!(msg.contains("managed inference failed"), "{msg}");
        assert!(
            msg.contains("code=content_filter"),
            "classification lost: {msg}"
        );
    }

    /// No classification fields → unchanged text, so nothing downstream that
    /// matches on the old string breaks.
    #[test]
    fn a_bare_managed_error_is_unchanged() {
        let events = parse_openai_responses_sse_line("error", r#"{"error":{}}"#);
        let StreamEvent::Error(msg) = events.first().expect("an error event") else {
            panic!("expected StreamEvent::Error");
        };
        assert_eq!(msg, "managed inference failed");
    }

    #[test]
    fn parse_openai_text_delta() {
        let line = r#"data: {"choices":[{"delta":{"content":"Hello"}}]}"#;
        let events = parse_openai_sse_line(line);
        assert_eq!(events.len(), 1);
        match &events[0] {
            StreamEvent::TextDelta(t) => assert_eq!(t, "Hello"),
            other => panic!("expected TextDelta, got {:?}", other),
        }
    }

    #[test]
    fn parse_openai_tool_call_start() {
        let line = r#"data: {"choices":[{"delta":{"tool_calls":[{"index":0,"function":{"name":"edit_file"}}]}}]}"#;
        let events = parse_openai_sse_line(line);
        assert_eq!(events.len(), 1);
        match &events[0] {
            StreamEvent::ToolCallStart { name, index, .. } => {
                assert_eq!(name, "edit_file");
                assert_eq!(*index, 0);
            }
            other => panic!("expected ToolCallStart, got {:?}", other),
        }
    }

    #[test]
    fn parse_openai_tool_call_delta() {
        let line = r#"data: {"choices":[{"delta":{"tool_calls":[{"index":0,"function":{"arguments":"{\"path\":"}}]}}]}"#;
        let events = parse_openai_sse_line(line);
        assert_eq!(events.len(), 1);
        match &events[0] {
            StreamEvent::ToolCallDelta {
                index,
                arguments_delta,
            } => {
                assert_eq!(*index, 0);
                assert!(arguments_delta.contains("path"));
            }
            other => panic!("expected ToolCallDelta, got {:?}", other),
        }
    }

    #[test]
    fn parse_openai_multiple_tool_calls_in_chunk() {
        // When OpenAI sends multiple tool call deltas in a single SSE chunk
        let line = r#"data: {"choices":[{"delta":{"tool_calls":[{"index":0,"function":{"name":"read_file"}},{"index":1,"function":{"name":"search"}}]}}]}"#;
        let events = parse_openai_sse_line(line);
        assert_eq!(events.len(), 2);
        match &events[0] {
            StreamEvent::ToolCallStart { name, index, .. } => {
                assert_eq!(name, "read_file");
                assert_eq!(*index, 0);
            }
            other => panic!("expected ToolCallStart, got {:?}", other),
        }
        match &events[1] {
            StreamEvent::ToolCallStart { name, index, .. } => {
                assert_eq!(name, "search");
                assert_eq!(*index, 1);
            }
            other => panic!("expected ToolCallStart, got {:?}", other),
        }
    }

    #[test]
    fn parse_openai_done() {
        assert!(parse_openai_sse_line("data: [DONE]").is_empty());
    }

    #[test]
    fn parse_anthropic_text_delta() {
        let data = r#"{"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":"world"}}"#;
        let events = parse_anthropic_sse_line("content_block_delta", data);
        assert_eq!(events.len(), 1);
        match &events[0] {
            StreamEvent::TextDelta(t) => assert_eq!(t, "world"),
            other => panic!("expected TextDelta, got {:?}", other),
        }
    }

    #[test]
    fn parse_anthropic_tool_start() {
        let data = r#"{"type":"content_block_start","index":1,"content_block":{"type":"tool_use","id":"t1","name":"search","input":{}}}"#;
        let events = parse_anthropic_sse_line("content_block_start", data);
        assert_eq!(events.len(), 1);
        match &events[0] {
            StreamEvent::ToolCallStart { name, index, .. } => {
                assert_eq!(name, "search");
                assert_eq!(*index, 1);
            }
            other => panic!("expected ToolCallStart, got {:?}", other),
        }
    }

    #[test]
    fn accumulator_builds_result() {
        let mut acc = StreamAccumulator::default();
        acc.push(&StreamEvent::TextDelta("Hello ".into()));
        acc.push(&StreamEvent::TextDelta("world".into()));
        acc.push(&StreamEvent::ToolCallStart {
            name: "search".into(),
            index: 0,
            id: None,
        });
        acc.push(&StreamEvent::ToolCallDelta {
            index: 0,
            arguments_delta: r#"{"q":"test"}"#.into(),
        });

        let (text, tools) = acc.finish();
        assert_eq!(text, "Hello world");
        assert_eq!(tools.len(), 1);
        assert_eq!(tools[0].name, "search");
        assert!(tools[0].arguments.contains_key("q"));
    }

    #[test]
    fn parse_sse_lines_openai_format() {
        let chunk = "data: {\"choices\":[{\"delta\":{\"content\":\"Hi\"}}]}\n\ndata: [DONE]\n\n";
        let events = parse_sse_lines(chunk);
        assert_eq!(events.len(), 2);
        assert_eq!(events[0].0, "message");
        assert_eq!(events[1].1, "[DONE]");
    }

    #[test]
    fn parse_sse_lines_anthropic_format() {
        let chunk = "event: content_block_delta\ndata: {\"delta\":{\"type\":\"text_delta\",\"text\":\"Hi\"}}\n\n";
        let events = parse_sse_lines(chunk);
        assert_eq!(events.len(), 1);
        assert_eq!(events[0].0, "content_block_delta");
    }

    #[test]
    fn parse_anthropic_message_start_emits_usage() {
        let data = r#"{"type":"message_start","message":{"id":"msg_1","role":"assistant","usage":{"input_tokens":245,"output_tokens":1}}}"#;
        let events = parse_anthropic_sse_line("message_start", data);
        assert_eq!(events.len(), 1);
        match &events[0] {
            StreamEvent::Usage {
                input_tokens,
                output_tokens,
                ..
            } => {
                assert_eq!(*input_tokens, 245);
                assert_eq!(*output_tokens, 1);
            }
            other => panic!("expected Usage, got {:?}", other),
        }
    }

    #[test]
    fn parse_anthropic_message_delta_emits_stop_reason_and_usage() {
        let data = r#"{"type":"message_delta","delta":{"stop_reason":"end_turn"},"usage":{"output_tokens":87}}"#;
        let events = parse_anthropic_sse_line("message_delta", data);
        // message_delta now carries both the termination reason and the
        // final output-token count, emitted as StopReason then Usage.
        assert_eq!(events.len(), 2);
        match &events[0] {
            StreamEvent::StopReason(reason) => assert_eq!(reason, "end_turn"),
            other => panic!("expected StopReason, got {:?}", other),
        }
        match &events[1] {
            StreamEvent::Usage {
                input_tokens,
                output_tokens,
                ..
            } => {
                assert_eq!(*input_tokens, 0);
                assert_eq!(*output_tokens, 87);
            }
            other => panic!("expected Usage, got {:?}", other),
        }
    }

    #[test]
    fn parse_anthropic_message_delta_max_tokens_stop_reason() {
        // A truncation: stop_reason "max_tokens" must surface so
        // InferenceResult::was_truncated() can fire on the streaming path.
        let data = r#"{"type":"message_delta","delta":{"stop_reason":"max_tokens"},"usage":{"output_tokens":4096}}"#;
        let events = parse_anthropic_sse_line("message_delta", data);
        assert!(matches!(
            &events[0],
            StreamEvent::StopReason(r) if r == "max_tokens"
        ));
    }

    #[test]
    fn parse_openai_finish_reason_length_surfaces() {
        // OpenAI's terminal content chunk carries finish_reason="length"
        // on truncation; it must surface as a StopReason event.
        let line = r#"data: {"choices":[{"delta":{"content":""},"finish_reason":"length"}]}"#;
        let events = parse_openai_sse_line(line);
        assert!(events
            .iter()
            .any(|e| matches!(e, StreamEvent::StopReason(r) if r == "length")));
    }

    #[test]
    fn accumulator_captures_stop_reason() {
        let mut acc = StreamAccumulator::default();
        acc.push(&StreamEvent::TextDelta("partial".into()));
        acc.push(&StreamEvent::StopReason("max_tokens".into()));
        let (_, _, _, stop) = acc.finish_with_usage();
        assert_eq!(stop.as_deref(), Some("max_tokens"));
    }

    #[test]
    fn parse_anthropic_message_start_without_usage_is_empty() {
        // Some forward-compat payloads may omit usage; don't crash.
        let data = r#"{"type":"message_start","message":{"id":"msg_1"}}"#;
        assert!(parse_anthropic_sse_line("message_start", data).is_empty());
    }

    #[test]
    fn accumulator_tracks_usage_across_anthropic_stream() {
        // Simulate the exact shape of a real Anthropic stream:
        // message_start → content_block_start → content_block_delta × 3 → message_delta.
        let mut acc = StreamAccumulator::default();
        for event in parse_anthropic_sse_line(
            "message_start",
            r#"{"message":{"usage":{"input_tokens":245,"output_tokens":1}}}"#,
        ) {
            acc.push(&event);
        }
        for event in parse_anthropic_sse_line(
            "content_block_start",
            r#"{"index":0,"content_block":{"type":"text","text":""}}"#,
        ) {
            acc.push(&event);
        }
        for (chunk, _) in [
            (r#"{"delta":{"type":"text_delta","text":"Hello"}}"#, ()),
            (r#"{"delta":{"type":"text_delta","text":", "}}"#, ()),
            (r#"{"delta":{"type":"text_delta","text":"world"}}"#, ()),
        ] {
            for event in parse_anthropic_sse_line("content_block_delta", chunk) {
                acc.push(&event);
            }
        }
        for event in parse_anthropic_sse_line("message_delta", r#"{"usage":{"output_tokens":87}}"#)
        {
            acc.push(&event);
        }

        let (text, tools, usage, _stop) = acc.finish_with_usage();
        assert_eq!(text, "Hello, world");
        assert!(tools.is_empty());
        let usage = usage.expect("provider reported usage; must surface");
        assert_eq!(usage.prompt_tokens, 245);
        // message_delta output (87) must win over message_start stub (1).
        assert_eq!(usage.completion_tokens, 87);
        assert_eq!(usage.total_tokens, 332);
    }

    #[test]
    fn parse_openai_final_chunk_emits_usage() {
        // OpenAI's final usage chunk when `stream_options.include_usage`
        // is set: `choices` is empty and `usage` carries the real counts.
        let line = r#"data: {"id":"chatcmpl-1","object":"chat.completion.chunk","choices":[],"usage":{"prompt_tokens":245,"completion_tokens":87,"total_tokens":332}}"#;
        let events = parse_openai_sse_line(line);
        assert_eq!(events.len(), 1);
        match &events[0] {
            StreamEvent::Usage {
                input_tokens,
                output_tokens,
                ..
            } => {
                assert_eq!(*input_tokens, 245);
                assert_eq!(*output_tokens, 87);
            }
            other => panic!("expected Usage, got {:?}", other),
        }
    }

    #[test]
    fn accumulator_tracks_usage_across_openai_stream() {
        // Simulate a full OpenAI stream with `stream_options.include_usage`:
        // text delta chunks followed by a choiceless usage-only chunk.
        let mut acc = StreamAccumulator::default();
        for line in [
            r#"data: {"choices":[{"delta":{"content":"Hello"}}]}"#,
            r#"data: {"choices":[{"delta":{"content":", "}}]}"#,
            r#"data: {"choices":[{"delta":{"content":"world"}}]}"#,
            r#"data: {"id":"chatcmpl-1","choices":[],"usage":{"prompt_tokens":245,"completion_tokens":87}}"#,
        ] {
            for event in parse_openai_sse_line(line) {
                acc.push(&event);
            }
        }

        let (text, tools, usage, _stop) = acc.finish_with_usage();
        assert_eq!(text, "Hello, world");
        assert!(tools.is_empty());
        let usage = usage.expect("provider reported usage; must surface");
        assert_eq!(usage.prompt_tokens, 245);
        assert_eq!(usage.completion_tokens, 87);
        assert_eq!(usage.total_tokens, 332);
    }

    #[test]
    fn accumulator_returns_no_usage_when_provider_silent() {
        // OpenAI without `stream_options.include_usage` — no Usage
        // events. `finish_with_usage` returns None so callers can fall
        // back to their own estimator.
        let mut acc = StreamAccumulator::default();
        acc.push(&StreamEvent::TextDelta("hi".into()));
        let (_, _, usage, _stop) = acc.finish_with_usage();
        assert!(usage.is_none());
    }

    #[test]
    fn anthropic_stream_decodes_cache_tokens_from_message_start() {
        // message_start carries the cache split + uncached input; message_delta
        // carries the final output. The accumulator must preserve cache tokens
        // (which arrive once) across the otherwise-zero delta.
        let mut acc = StreamAccumulator::default();
        let start = r#"{"message":{"usage":{"input_tokens":50,"output_tokens":1,"cache_read_input_tokens":4000,"cache_creation_input_tokens":600}}}"#;
        for e in parse_anthropic_sse_line("message_start", start) {
            acc.push(&e);
        }
        let delta = r#"{"delta":{"stop_reason":"end_turn"},"usage":{"output_tokens":87}}"#;
        for e in parse_anthropic_sse_line("message_delta", delta) {
            acc.push(&e);
        }
        let (_t, _c, usage, stop) = acc.finish_with_usage();
        let u = usage.expect("usage surfaced");
        assert_eq!(u.prompt_tokens, 50, "uncached prefix");
        assert_eq!(u.completion_tokens, 87, "final output from message_delta");
        assert_eq!(u.cache_read_input_tokens, 4000);
        assert_eq!(u.cache_creation_input_tokens, 600);
        assert_eq!(stop.as_deref(), Some("end_turn"));
    }

    #[test]
    fn openai_stream_normalizes_cached_tokens_out_of_prompt() {
        // OpenAI's streamed prompt_tokens (1000) INCLUDES the cached subset
        // (800); the parser must subtract so prompt_tokens is the uncached 200
        // and 800 lands in cache_read — else cost double-charges the cached part.
        let mut acc = StreamAccumulator::default();
        let chunk = r#"data: {"choices":[],"usage":{"prompt_tokens":1000,"completion_tokens":40,"prompt_tokens_details":{"cached_tokens":800}}}"#;
        for e in parse_openai_sse_line(chunk) {
            acc.push(&e);
        }
        let (_t, _c, usage, _s) = acc.finish_with_usage();
        let u = usage.expect("usage surfaced");
        assert_eq!(u.prompt_tokens, 200, "uncached = 1000 - 800");
        assert_eq!(u.cache_read_input_tokens, 800);
        assert_eq!(
            u.cache_creation_input_tokens, 0,
            "OpenAI has no write bucket"
        );
    }

    #[test]
    fn responses_failure_is_a_terminal_safe_error_event() {
        let events = parse_openai_responses_sse_line(
            "response.failed",
            r#"{"response":{"error":{"message":"managed model unavailable","stack":"secret"}}}"#,
        );
        assert!(matches!(
            events.as_slice(),
            [StreamEvent::Error(message)] if message == "managed model unavailable"
        ));
        assert!(!format!("{events:?}").contains("secret"));
    }

    #[test]
    fn responses_reasoning_item_done_is_retained_verbatim() {
        let data = r#"{"output_index":0,"item":{"type":"reasoning","id":"rs_1","status":"completed","summary":[{"type":"summary_text","text":"safe summary"}],"encrypted_content":"opaque-ciphertext"}}"#;
        let events = parse_openai_responses_sse_line("response.output_item.done", data);
        let expected = serde_json::json!({
            "type": "reasoning",
            "id": "rs_1",
            "status": "completed",
            "summary": [{"type": "summary_text", "text": "safe summary"}],
            "encrypted_content": "opaque-ciphertext",
        });
        assert!(matches!(
            events.as_slice(),
            [StreamEvent::ProviderOutputItem(item)] if item == &expected
        ));
        let mut accumulator = StreamAccumulator::default();
        accumulator.push(&events[0]);
        let (_, _, _, _, items) = accumulator.finish_with_provider_output_items();
        assert_eq!(items, vec![expected]);
    }

    #[test]
    fn responses_incomplete_is_terminal_failure_not_success() {
        let events = parse_openai_responses_sse_line(
            "response.incomplete",
            r#"{"response":{"status":"incomplete","incomplete_details":{"reason":"max_output_tokens"},"usage":{"input_tokens":17,"output_tokens":9}}}"#,
        );
        assert!(
            events
                .iter()
                .any(|event| matches!(event, StreamEvent::Error(_))),
            "response.incomplete must emit a terminal error"
        );
    }
}