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),
    /// The provider's own dollar figure for the stream (OpenRouter's
    /// `usage.cost` on its final usage chunk). The accumulator carries the
    /// latest one into [`TokenUsage::reported_cost_usd`]. Not forwarded to
    /// clients as an event; a stream without it has no provider figure.
    ProviderCost(f64),
    /// 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, and Anthropic's exact `refusal`. #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();
        // Anthropic's `stop_reason: "refusal"`, matched exactly.
        v == "refusal"
            || 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
}

/// Fold a complete OpenAI chat-completions SSE body into the non-streaming
/// response it stands for, so the ordinary `parse_response` reads it.
///
/// This exists for vllm-mlx serving gpt-oss. Its non-streaming path cleans the
/// harmony channel tokens out of the output *before* its reasoning parser runs
/// (vllm-mlx 0.6.0, `engine/simple.py`), so the parser never sees the channels:
/// a finished answer comes back with its reasoning dropped, and one cut off at
/// `max_tokens` inside the analysis channel comes back with that reasoning *as
/// the answer*. The streaming path splits them correctly into
/// `delta.reasoning_content` and `delta.content`, so CAR asks for a stream and
/// folds it here. Unlike [`parse_openai_sse_line`] this keeps the reasoning,
/// which has no [`StreamEvent`] of its own.
pub(crate) fn fold_openai_chat_sse(body: &str) -> Result<String, crate::InferenceError> {
    // A server that ignored `stream` answered with the ordinary body.
    if body.trim_start().starts_with('{') {
        return Ok(body.to_string());
    }
    let mut chunks = 0usize;
    let mut done = false;
    let mut content = String::new();
    let mut reasoning = String::new();
    // index -> (id, name, arguments), in the order the indices first appear.
    let mut tools: Vec<(u64, String, String, String)> = Vec::new();
    let mut finish_reason = serde_json::Value::Null;
    let mut usage = serde_json::Value::Null;
    for line in body.lines() {
        let Some(data) = line.strip_prefix("data:").map(str::trim) else {
            continue;
        };
        if data == "[DONE]" {
            done = true;
            continue;
        }
        if data.is_empty() {
            continue;
        }
        let chunk: serde_json::Value = serde_json::from_str(data).map_err(|e| {
            crate::InferenceError::InferenceFailed(format!("parse stream chunk: {e}"))
        })?;
        chunks += 1;
        if let Some(error) = chunk.get("error") {
            let message = error
                .get("message")
                .and_then(|m| m.as_str())
                .unwrap_or("the server reported an error mid-stream")
                .to_string();
            // After the 200, a refusal the non-streaming path answered with a
            // status (a busy engine's 503, an out-of-memory 500) arrives here
            // with that status as its `code`; keep it retryable.
            let status = error.get("code").and_then(|c| c.as_u64()).map(|c| c as u16);
            return Err(match status {
                Some(code) if code == 429 || code >= 500 => crate::InferenceError::Transient {
                    status: Some(code),
                    message,
                },
                _ => crate::InferenceError::InferenceFailed(message),
            });
        }
        if let Some(u) = chunk.get("usage").filter(|u| !u.is_null()) {
            usage = u.clone();
        }
        let Some(choice) = chunk.pointer("/choices/0") else {
            continue;
        };
        if let Some(fr) = choice.get("finish_reason").filter(|f| !f.is_null()) {
            finish_reason = fr.clone();
        }
        let Some(delta) = choice.get("delta") else {
            continue;
        };
        if let Some(t) = delta.get("content").and_then(|c| c.as_str()) {
            content.push_str(t);
        }
        if let Some(t) = delta
            .get("reasoning_content")
            .or_else(|| delta.get("reasoning"))
            .and_then(|c| c.as_str())
        {
            reasoning.push_str(t);
        }
        for call in delta
            .get("tool_calls")
            .and_then(|t| t.as_array())
            .into_iter()
            .flatten()
        {
            let index = call.get("index").and_then(|i| i.as_u64()).unwrap_or(0);
            let slot = match tools.iter().position(|t| t.0 == index) {
                Some(i) => &mut tools[i],
                None => {
                    tools.push((index, String::new(), String::new(), String::new()));
                    tools.last_mut().expect("just pushed")
                }
            };
            if let Some(id) = call.get("id").and_then(|v| v.as_str()) {
                slot.1 = id.to_string();
            }
            if let Some(name) = call.pointer("/function/name").and_then(|v| v.as_str()) {
                slot.2.push_str(name);
            }
            if let Some(args) = call.pointer("/function/arguments").and_then(|v| v.as_str()) {
                slot.3.push_str(args);
            }
        }
    }
    // A stream that stopped with neither a finish reason nor `[DONE]` was cut
    // (vllm-mlx's inactivity guard ends one that way), not completed.
    if chunks == 0 || (!done && finish_reason.is_null()) {
        return Err(crate::InferenceError::InferenceFailed(
            "the stream ended before the server finished the response".to_string(),
        ));
    }
    let mut message = serde_json::json!({ "role": "assistant", "content": content });
    if !reasoning.is_empty() {
        message["reasoning_content"] = serde_json::Value::String(reasoning);
    }
    if !tools.is_empty() {
        message["tool_calls"] = tools
            .into_iter()
            .map(|(_, id, name, arguments)| {
                serde_json::json!({
                    "id": id,
                    "type": "function",
                    "function": { "name": name, "arguments": arguments },
                })
            })
            .collect();
    }
    let mut response = serde_json::json!({
        "choices": [{ "index": 0, "message": message, "finish_reason": finish_reason }],
    });
    // Absent, not null: a null `usage` would read as a measured zero.
    if !usage.is_null() {
        response["usage"] = usage;
    }
    Ok(response.to_string())
}

/// 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(),
    }
}

/// Cut a stream at the caller's stop sequences (`GenerateParams::stop`).
///
/// The streaming counterpart of
/// [`crate::tasks::generate::apply_stop_backstop`]: text is relayed with just
/// enough held back that a stop sequence split across deltas is still caught,
/// nothing after the earliest stop sequence is forwarded, the terminal
/// `Done { text }` is cut the same way, and once the stream has been cut its
/// stop reason reads [`crate::tasks::generate::STOP_SEQUENCE_STOP_REASON`].
/// Every non-text event (usage, tool calls, errors) passes through, so a
/// stream cut early still reports what the provider billed. The provider's
/// own `StopReason` is held until `Done` (or the end of the stream), so a
/// stream whose stop sequence first shows up in the `Done` text still reports
/// exactly one reason. With no non-empty stop sequence the receiver is
/// returned untouched.
///
/// **Cost.** Cutting does not cancel the upstream request: where the provider
/// took no native stop (the OpenAI Responses API, the Codex subscription, a
/// fifth-and-later OpenAI sequence) the model keeps generating, and billing,
/// up to `max_tokens` while CAR discards the text. The source is drained
/// rather than dropped so its usage still reaches the spend ledger.
pub fn enforce_stop_sequences(
    mut source: tokio::sync::mpsc::Receiver<StreamEvent>,
    stops: Vec<String>,
) -> tokio::sync::mpsc::Receiver<StreamEvent> {
    let stops: Vec<String> = stops.into_iter().filter(|stop| !stop.is_empty()).collect();
    if stops.is_empty() {
        return source;
    }
    let (tx, rx) = tokio::sync::mpsc::channel(64);
    tokio::spawn(async move {
        let mut filter = StopSequenceFilter::new(stops);
        while let Some(event) = source.recv().await {
            for out in filter.push(event) {
                if tx.send(out).await.is_err() {
                    return;
                }
            }
        }
        for out in filter.finish() {
            if tx.send(out).await.is_err() {
                return;
            }
        }
    });
    rx
}

/// The state behind [`enforce_stop_sequences`], separate from the channel
/// plumbing so its event-by-event behavior is testable directly.
struct StopSequenceFilter {
    stops: Vec<String>,
    /// Text received but not yet forwarded: a suffix that could still be the
    /// start of a stop sequence.
    held: String,
    /// A stop sequence has been seen; no further text is forwarded.
    stopped: bool,
    /// The stream's one `StopReason` has been forwarded.
    reported_stop: bool,
    /// The provider's own `StopReason`, held until `Done` decides whether the
    /// stream was cut.
    provider_reason: Option<String>,
}

impl StopSequenceFilter {
    fn new(stops: Vec<String>) -> Self {
        Self {
            stops,
            held: String::new(),
            stopped: false,
            reported_stop: false,
            provider_reason: None,
        }
    }

    /// Byte length of the longest suffix of `held` that is a proper prefix of
    /// some stop sequence, at a char boundary.
    fn ambiguous_tail(&self) -> usize {
        self.held
            .char_indices()
            .map(|(index, _)| index)
            .find(|&index| {
                let tail = &self.held[index..];
                self.stops
                    .iter()
                    .any(|stop| stop.len() > tail.len() && stop.starts_with(tail))
            })
            .map_or(0, |index| self.held.len() - index)
    }

    fn take_held(&mut self) -> Vec<StreamEvent> {
        if self.held.is_empty() {
            Vec::new()
        } else {
            vec![StreamEvent::TextDelta(std::mem::take(&mut self.held))]
        }
    }

    fn stop_reason(&mut self) -> StreamEvent {
        self.reported_stop = true;
        StreamEvent::StopReason(crate::tasks::generate::STOP_SEQUENCE_STOP_REASON.to_string())
    }

    fn push(&mut self, event: StreamEvent) -> Vec<StreamEvent> {
        match event {
            StreamEvent::TextDelta(delta) => {
                if self.stopped {
                    return Vec::new();
                }
                self.held.push_str(&delta);
                let earliest = self
                    .stops
                    .iter()
                    .filter_map(|stop| self.held.find(stop.as_str()))
                    .min();
                if let Some(cut) = earliest {
                    self.held.truncate(cut);
                    self.stopped = true;
                    return self.take_held();
                }
                let emit = self.held.len() - self.ambiguous_tail();
                if emit == 0 {
                    return Vec::new();
                }
                let rest = self.held.split_off(emit);
                vec![StreamEvent::TextDelta(std::mem::replace(
                    &mut self.held,
                    rest,
                ))]
            }
            StreamEvent::StopReason(reason) => {
                // Held: `Done` may still reveal a stop sequence, and then the
                // provider's reason no longer describes the text.
                self.provider_reason = Some(reason);
                Vec::new()
            }
            StreamEvent::Done { text, tool_calls } => {
                let mut out = self.take_held();
                let cut = crate::tasks::generate::truncate_at_stop(&text, &self.stops);
                if cut.len() < text.len() {
                    self.stopped = true;
                }
                out.extend(self.reason_event());
                if !tool_calls.is_empty() {
                    tracing::warn!(
                        tool_calls = tool_calls.len(),
                        "stop sequences apply to streamed text only; tool-call arguments \
                         in this stream were not checked against them"
                    );
                }
                out.push(StreamEvent::Done {
                    text: cut,
                    tool_calls,
                });
                out
            }
            StreamEvent::Error(message) => {
                // Terminal: whatever text is held is all the text there is.
                let mut out = self.take_held();
                out.extend(self.reason_event());
                out.push(StreamEvent::Error(message));
                out
            }
            // Usage, tool-call and bookkeeping events can interleave with
            // text, so they pass straight through and the held tail stays
            // held: the next delta may still complete a stop sequence.
            other @ (StreamEvent::ToolCallStart { .. }
            | StreamEvent::ToolCallDelta { .. }
            | StreamEvent::ToolCallArguments { .. }
            | StreamEvent::Usage { .. }
            | StreamEvent::ProviderCost(_)
            | StreamEvent::ProviderOutputItem(_)
            | StreamEvent::Attribution(_)
            | StreamEvent::Substituted(_)) => vec![other],
        }
    }

    /// The one stop reason this stream reports, once: `stop_sequence` when
    /// it was cut, else the provider's held reason.
    fn reason_event(&mut self) -> Option<StreamEvent> {
        if self.reported_stop {
            return None;
        }
        if self.stopped {
            self.provider_reason = None;
            return Some(self.stop_reason());
        }
        self.provider_reason.take().map(|reason| {
            self.reported_stop = true;
            StreamEvent::StopReason(reason)
        })
    }

    /// The source closed without a `Done`: forward what is still held.
    fn finish(&mut self) -> Vec<StreamEvent> {
        let mut out = self.take_held();
        out.extend(self.reason_event());
        out
    }
}

/// 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>,
    /// The latest [`StreamEvent::ProviderCost`], the provider's own figure.
    reported_cost_usd: Option<f64>,
}

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::ProviderCost(cost) => {
                self.reported_cost_usd = Some(*cost);
            }
            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,
                reported_cost_usd: self.reported_cost_usd,
                // Streamed usage events carry no reasoning split yet.
                reasoning_tokens: None,
            })
        } 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 {

    #[test]
    fn a_folded_stream_must_be_a_complete_stream() {
        // A server that ignored `stream` sent the ordinary body: passed through.
        let json = r#"{"choices":[{"message":{"content":"hi"},"finish_reason":"stop"}]}"#;
        assert_eq!(fold_openai_chat_sse(json).unwrap(), json);
        // Cut with no finish reason and no [DONE]: not a completed response.
        let cut = "data: {\"choices\":[{\"index\":0,\"delta\":{\"content\":\"par\"}}]}\n\n";
        assert!(fold_openai_chat_sse(cut).is_err());
        assert!(fold_openai_chat_sse("").is_err());
        // No usage chunk: `usage` is absent, not a measured zero.
        let no_usage = "data: {\"choices\":[{\"index\":0,\"delta\":{\"content\":\"ok\"},\"finish_reason\":\"stop\"}]}\n\ndata: [DONE]\n\n";
        let folded: serde_json::Value =
            serde_json::from_str(&fold_openai_chat_sse(no_usage).unwrap()).unwrap();
        assert!(folded.get("usage").is_none(), "{folded}");
        assert_eq!(
            folded
                .pointer("/choices/0/message/content")
                .and_then(|c| c.as_str()),
            Some("ok")
        );
    }

    /// A busy engine refuses after the 200 has gone out, as an error chunk
    /// carrying the status the non-streaming path would have answered with.
    #[test]
    fn a_mid_stream_server_error_keeps_its_status() {
        let busy = "data: {\"error\":{\"message\":\"Internal server error\",\"code\":500}}\n\n";
        assert!(matches!(
            fold_openai_chat_sse(busy),
            Err(crate::InferenceError::Transient {
                status: Some(500),
                ..
            })
        ));
        let bad = "data: {\"error\":{\"message\":\"bad request\",\"code\":400}}\n\n";
        assert!(matches!(
            fold_openai_chat_sse(bad),
            Err(crate::InferenceError::InferenceFailed(_))
        ));
    }

    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"
        );
    }

    fn run_stop_filter(stops: &[&str], events: Vec<StreamEvent>) -> Vec<StreamEvent> {
        let mut filter = StopSequenceFilter::new(stops.iter().map(|s| s.to_string()).collect());
        let mut out = Vec::new();
        for event in events {
            out.extend(filter.push(event));
        }
        out.extend(filter.finish());
        out
    }

    fn streamed_text(events: &[StreamEvent]) -> String {
        events
            .iter()
            .filter_map(|event| match event {
                StreamEvent::TextDelta(delta) => Some(delta.as_str()),
                _ => None,
            })
            .collect()
    }

    #[test]
    fn stop_filter_catches_a_sequence_split_across_three_deltas() {
        let out = run_stop_filter(
            &["STOPHERE"],
            vec![
                StreamEvent::TextDelta("Paris S".into()),
                StreamEvent::TextDelta("TOPH".into()),
                StreamEvent::Usage {
                    input_tokens: 3,
                    output_tokens: 2,
                    cache_read_input_tokens: 0,
                    cache_creation_input_tokens: 0,
                },
                StreamEvent::TextDelta("ERE indeed".into()),
                StreamEvent::TextDelta(" and more".into()),
                StreamEvent::StopReason("length".into()),
                StreamEvent::Done {
                    text: "Paris STOPHERE indeed and more".into(),
                    tool_calls: Vec::new(),
                },
            ],
        );
        assert_eq!(streamed_text(&out), "Paris ");
        let reasons: Vec<&str> = out
            .iter()
            .filter_map(|event| match event {
                StreamEvent::StopReason(reason) => Some(reason.as_str()),
                _ => None,
            })
            .collect();
        assert_eq!(reasons, ["stop_sequence"]);
        assert!(matches!(
            out.last(),
            Some(StreamEvent::Done { text, .. }) if text == "Paris "
        ));
        // Usage still reaches the consumer after the cut.
        assert!(out
            .iter()
            .any(|event| matches!(event, StreamEvent::Usage { .. })));
    }

    /// A held tail that never becomes a stop sequence is released, and text
    /// with no stop sequence passes through whole with the provider's reason.
    #[test]
    fn stop_filter_releases_a_false_alarm_tail_unchanged() {
        let out = run_stop_filter(
            &["STOPHERE"],
            vec![
                StreamEvent::TextDelta("almost STO".into()),
                StreamEvent::TextDelta("P but not quite".into()),
                StreamEvent::StopReason("stop".into()),
                StreamEvent::Done {
                    text: "almost STOP but not quite".into(),
                    tool_calls: Vec::new(),
                },
            ],
        );
        assert_eq!(streamed_text(&out), "almost STOP but not quite");
        assert!(out
            .iter()
            .any(|event| matches!(event, StreamEvent::StopReason(reason) if reason == "stop")));
        assert!(matches!(
            out.last(),
            Some(StreamEvent::Done { text, .. }) if text == "almost STOP but not quite"
        ));
    }

    /// The provider's reason arrives before a `Done` whose text is the only
    /// place the stop sequence appears (no deltas): exactly one reason goes
    /// out, and it is `stop_sequence`.
    #[test]
    fn stop_filter_reports_one_reason_when_only_done_carries_the_stop() {
        let out = run_stop_filter(
            &["STOPHERE"],
            vec![
                StreamEvent::StopReason("length".into()),
                StreamEvent::Done {
                    text: "Paris STOPHERE indeed".into(),
                    tool_calls: Vec::new(),
                },
            ],
        );
        let reasons: Vec<&str> = out
            .iter()
            .filter_map(|event| match event {
                StreamEvent::StopReason(reason) => Some(reason.as_str()),
                _ => None,
            })
            .collect();
        assert_eq!(reasons, ["stop_sequence"]);
        assert!(matches!(
            out.last(),
            Some(StreamEvent::Done { text, .. }) if text == "Paris "
        ));
    }

    #[test]
    fn stop_filter_flushes_a_held_tail_when_the_source_closes() {
        let out = run_stop_filter(
            &["STOPHERE"],
            vec![StreamEvent::TextDelta("ends ST".into())],
        );
        assert_eq!(streamed_text(&out), "ends ST");
    }

    #[test]
    fn stop_filter_holds_back_only_at_char_boundaries() {
        let out = run_stop_filter(
            &["\u{e9}\u{e9}"],
            vec![
                StreamEvent::TextDelta("caf\u{e9}".into()),
                StreamEvent::TextDelta("\u{e9} au lait".into()),
            ],
        );
        assert_eq!(streamed_text(&out), "caf");
    }

    #[test]
    fn apply_stop_backstop_cuts_text_and_reports_stop_sequence() {
        use crate::tasks::generate::apply_stop_backstop;
        let stops = vec!["STOPHERE".to_string()];
        assert_eq!(
            apply_stop_backstop(
                "Paris STOPHERE indeed".into(),
                Some("length".into()),
                &stops,
                0
            ),
            ("Paris ".to_string(), Some("stop_sequence".to_string()))
        );
        // Nothing to cut: the provider's reason stands.
        assert_eq!(
            apply_stop_backstop("Paris".into(), Some("stop".into()), &stops, 0),
            ("Paris".to_string(), Some("stop".to_string()))
        );
        // No stop list: untouched, even when the text contains the string.
        assert_eq!(
            apply_stop_backstop("a STOPHERE b".into(), None, &[], 0),
            ("a STOPHERE b".to_string(), None)
        );
    }
}