localharness 0.25.0

//! Async HTTPS client for the Anthropic Messages API.
//!
//! Single public type: [`AnthropicClient`]. Two methods:
//! [`stream_messages`][AnthropicClient::stream_messages] (SSE turn loop)
//! and [`messages`][AnthropicClient::messages] (non-streaming one-shot,
//! used by `start_subagent` / compaction summary).
//!
//! The SSE decoder ([`MessagesSseStream`]) reuses the wire-agnostic
//! frame-buffering skeleton from the Gemini backend: CRLF+LF-tolerant
//! `take_frame`, partial-chunk buffering. The only Anthropic-specific
//! piece is payload decoding — Anthropic frames are
//! `event: <name>\ndata: <json>\n\n`; we ignore the `event:` line and
//! decode the `data:` JSON as a [`StreamEvent`] (the JSON carries the
//! event name in its `"type"` field, so the `event:` line is redundant).
//!
//! The API key is held as a `Box<str>` so it never appears in `Debug`
//! output and is dropped reliably when the client is dropped.

use std::fmt;
use std::pin::Pin;
use std::sync::Arc;
use std::task::{Context, Poll};
#[cfg(not(target_arch = "wasm32"))]
use std::time::Duration;

use bytes::{Bytes, BytesMut};
use futures_core::Stream;
use futures_util::stream::StreamExt;
use reqwest::{Client, Url};
use tracing::trace;

use crate::backends::anthropic::wire::{
    MessagesRequest, MessagesResponse, StreamEvent, ANTHROPIC_VERSION,
};
use crate::error::{Error, Result};

const DEFAULT_BASE_URL: &str = "https://api.anthropic.com";
#[cfg(not(target_arch = "wasm32"))]
const DEFAULT_TIMEOUT: Duration = Duration::from_secs(300);

/// HTTPS client for `api.anthropic.com` (or a credit-proxy base URL).
pub struct AnthropicClient {
    http: Client,
    api_key: Box<str>,
    base_url: Url,
}

impl fmt::Debug for AnthropicClient {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_struct("AnthropicClient")
            .field("base_url", &self.base_url.as_str())
            .field("api_key", &"<redacted>")
            .finish()
    }
}

impl AnthropicClient {
    /// Build a client for the given API key, talking directly to
    /// `api.anthropic.com`.
    pub fn new(api_key: impl Into<String>) -> Result<Self> {
        let builder =
            Client::builder().user_agent(concat!("localharness/", env!("CARGO_PKG_VERSION")));
        // reqwest's wasm builder doesn't have .timeout() — browser fetch
        // controls timeouts, not the client config.
        #[cfg(not(target_arch = "wasm32"))]
        let builder = builder.timeout(DEFAULT_TIMEOUT);
        let http = builder
            .build()
            .map_err(|e| Error::other(format!("reqwest client build: {e}")))?;
        Ok(Self {
            http,
            api_key: api_key.into().into_boxed_str(),
            base_url: Url::parse(DEFAULT_BASE_URL).expect("default base url is valid"),
        })
    }

    /// Override the base URL (e.g. the future localharness credit proxy,
    /// or a test server). In credits mode the `api_key` carries the proxy
    /// auth token rather than a raw Anthropic key.
    pub fn with_base_url(mut self, url: Url) -> Self {
        self.base_url = url;
        self
    }

    fn messages_url(&self) -> Result<Url> {
        self.base_url
            .join("v1/messages")
            .map_err(|e| Error::other(format!("invalid messages url: {e}")))
    }

    /// Non-streaming `POST /v1/messages`. Used for one-shot completions
    /// (subagent, compaction summary).
    pub async fn messages(&self, req: &MessagesRequest) -> Result<MessagesResponse> {
        let url = self.messages_url()?;
        // Force non-stream on the one-shot path regardless of caller flag.
        let mut body = req.clone();
        body.stream = false;
        let response = self
            .http
            .post(url)
            .header("x-api-key", self.api_key.as_ref())
            .header("anthropic-version", ANTHROPIC_VERSION)
            .header("content-type", "application/json")
            .json(&body)
            .send()
            .await
            .map_err(|e| Error::other(format!("anthropic POST: {e}")))?;

        if !response.status().is_success() {
            let status = response.status();
            let body = response
                .text()
                .await
                .unwrap_or_else(|_| "<no body>".to_string());
            return Err(Error::other(format!("anthropic HTTP {status}: {body}")));
        }

        response
            .json::<MessagesResponse>()
            .await
            .map_err(|e| Error::other(format!("anthropic JSON: {e}")))
    }

    /// Streaming `POST /v1/messages` (`stream: true`). Returns a
    /// [`MessagesSseStream`] yielding one [`StreamEvent`] per SSE frame.
    pub async fn stream_messages(&self, req: &MessagesRequest) -> Result<MessagesSseStream> {
        let url = self.messages_url()?;
        let mut body = req.clone();
        body.stream = true;
        let response = self
            .http
            .post(url)
            .header("x-api-key", self.api_key.as_ref())
            .header("anthropic-version", ANTHROPIC_VERSION)
            .header("content-type", "application/json")
            .header("accept", "text/event-stream")
            .json(&body)
            .send()
            .await
            .map_err(|e| Error::other(format!("anthropic POST: {e}")))?;

        let debug_sse = std::env::var("LH_DEBUG_SSE").is_ok();
        if debug_sse {
            eprintln!(
                "[anthropic resp] status={} content-type={:?}",
                response.status(),
                response.headers().get("content-type"),
            );
        }
        if !response.status().is_success() {
            let status = response.status();
            let body = response
                .text()
                .await
                .unwrap_or_else(|_| "<no body>".to_string());
            if debug_sse {
                eprintln!("[anthropic ERROR] HTTP {status}: {body}");
            }
            return Err(Error::other(format!("anthropic HTTP {status}: {body}")));
        }

        let byte_stream = response
            .bytes_stream()
            .map(|res| res.map_err(|e| Error::other(format!("anthropic chunk read: {e}"))));
        Ok(MessagesSseStream::new(Box::pin(byte_stream)))
    }
}

// =============================================================================
// SSE stream
// =============================================================================

// On native, the SSE byte stream must be `Send` so it can move into a
// `tokio::spawn`'d turn. On wasm32, browser fetch streams aren't Send —
// fine, everything single-threads through `spawn_local`.
#[cfg(not(target_arch = "wasm32"))]
type ByteStream = Pin<Box<dyn Stream<Item = Result<Bytes>> + Send + 'static>>;
#[cfg(target_arch = "wasm32")]
type ByteStream = Pin<Box<dyn Stream<Item = Result<Bytes>> + 'static>>;

/// Decodes an Anthropic Messages SSE byte stream into [`StreamEvent`]s.
///
/// Frame format: `event: <name>\ndata: <json>\n\n`. A frame ends at the
/// first blank line — `\n\n` (LF) or `\r\n\r\n` (CRLF; browser fetch
/// surfaces CRLF, the wasm gotcha). We concatenate every `data:` line in
/// a frame and decode the result as a [`StreamEvent`], ignoring the
/// `event:` line (the JSON's own `"type"` field carries the same name).
pub struct MessagesSseStream {
    upstream: ByteStream,
    buffer: BytesMut,
    done: bool,
}

impl MessagesSseStream {
    /// Wrap a raw byte stream. Public so unit tests can feed canned bytes.
    pub fn new(upstream: ByteStream) -> Self {
        Self {
            upstream,
            buffer: BytesMut::with_capacity(8 * 1024),
            done: false,
        }
    }

    /// Pull a complete frame's `data:` payload from `self.buffer` if one
    /// is fully buffered. Returns `None` when no frame boundary is present
    /// yet (partial chunk). Boundary = first blank line, CRLF or LF.
    fn take_frame(&mut self) -> Option<Vec<u8>> {
        let bytes = &self.buffer[..];
        let mut i = 0;
        while i < bytes.len() {
            // Prefer the longer CRLF boundary so we consume it whole.
            if i + 3 < bytes.len()
                && bytes[i] == b'\r'
                && bytes[i + 1] == b'\n'
                && bytes[i + 2] == b'\r'
                && bytes[i + 3] == b'\n'
            {
                let frame = self.buffer.split_to(i + 4);
                return Some(extract_data_payload(&frame));
            }
            if i + 1 < bytes.len() && bytes[i] == b'\n' && bytes[i + 1] == b'\n' {
                let frame = self.buffer.split_to(i + 2);
                return Some(extract_data_payload(&frame));
            }
            i += 1;
        }
        None
    }

    /// Consume the ENTIRE remaining buffer as one final frame, regardless of
    /// whether it ends in a blank line. Called ONLY at upstream EOF: the
    /// WHATWG event-stream rule is that a stream's last event need not be
    /// terminated by a trailing blank line, so a complete final
    /// `data: {...}` with no `\n\n` would otherwise be silently dropped — and
    /// for Anthropic that frame can be the `message_delta` (carrying
    /// `stop_reason` + cumulative `output_tokens`) or the `message_stop`.
    /// Returns `None` once the buffer is empty.
    fn take_remaining(&mut self) -> Option<Vec<u8>> {
        if self.buffer.is_empty() {
            return None;
        }
        let frame = self.buffer.split_to(self.buffer.len());
        Some(extract_data_payload(&frame))
    }
}

impl Stream for MessagesSseStream {
    type Item = Result<StreamEvent>;

    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
        loop {
            if self.done {
                // Flush any remaining buffered frame even after upstream EOF.
                // First drain blank-line-terminated frames; once those are
                // gone, treat whatever is left as the final UNTERMINATED frame
                // (SSE permits the last event to omit the trailing blank line
                // — see `take_remaining`). Without this, a final
                // `message_delta` / `message_stop` with no trailing `\n\n`
                // would be silently dropped.
                let payload = match self.take_frame() {
                    Some(p) => p,
                    None => match self.take_remaining() {
                        Some(p) => p,
                        None => return Poll::Ready(None),
                    },
                };
                if payload.is_empty() {
                    continue;
                }
                return Poll::Ready(Some(decode_event(&payload)));
            }

            if let Some(payload) = self.take_frame() {
                if payload.is_empty() {
                    continue;
                }
                return Poll::Ready(Some(decode_event(&payload)));
            }

            match self.upstream.as_mut().poll_next(cx) {
                Poll::Pending => return Poll::Pending,
                Poll::Ready(Some(Ok(bytes))) => {
                    trace!(len = bytes.len(), "anthropic sse bytes");
                    self.buffer.extend_from_slice(&bytes);
                }
                Poll::Ready(Some(Err(e))) => return Poll::Ready(Some(Err(e))),
                Poll::Ready(None) => self.done = true,
            }
        }
    }
}

fn extract_data_payload(frame: &[u8]) -> Vec<u8> {
    // Concatenate every `data:` line in this frame, trimming the prefix
    // and any single leading space. `event:`/`id:`/`retry:` lines ignored.
    let mut out = Vec::with_capacity(frame.len());
    let text = std::str::from_utf8(frame).unwrap_or("");
    for line in text.split('\n') {
        let line = line.trim_end_matches('\r');
        if let Some(rest) = line.strip_prefix("data:") {
            let rest = rest.strip_prefix(' ').unwrap_or(rest);
            if !out.is_empty() {
                out.push(b'\n');
            }
            out.extend_from_slice(rest.as_bytes());
        }
    }
    out
}

fn decode_event(payload: &[u8]) -> Result<StreamEvent> {
    serde_json::from_slice::<StreamEvent>(payload).map_err(|e| {
        Error::other(format!(
            "anthropic sse decode: {e}; payload: {}",
            String::from_utf8_lossy(payload)
        ))
    })
}

/// Re-export an `Arc<AnthropicClient>` for ergonomic cloning into spawned
/// tasks.
pub type SharedClient = Arc<AnthropicClient>;

#[cfg(test)]
mod tests {
    use super::*;
    use crate::backends::anthropic::wire::{Block, BlockDelta, StopReason};
    use futures_util::stream;

    fn bytes_from(parts: &[&[u8]]) -> ByteStream {
        let owned: Vec<Bytes> = parts.iter().map(|b| Bytes::copy_from_slice(b)).collect();
        Box::pin(stream::iter(owned.into_iter().map(Ok)))
    }

    /// The canonical full streaming sequence the task spec asks for:
    /// message_start → content_block_start(text) → 2 text_deltas →
    /// content_block_stop → content_block_start(tool_use) → 2
    /// input_json_delta fragments → content_block_stop → message_delta
    /// (stop_reason: tool_use) → message_stop. As a sequence of complete
    /// SSE frames (one `event:`/`data:`/blank-line group each).
    fn canonical_frames() -> Vec<Vec<u8>> {
        let raw = [
            "event: message_start\ndata: {\"type\":\"message_start\",\"message\":{\"id\":\"msg_1\",\"model\":\"claude-haiku-4-5-20251001\",\"role\":\"assistant\",\"usage\":{\"input_tokens\":12,\"output_tokens\":1}}}\n\n",
            "event: content_block_start\ndata: {\"type\":\"content_block_start\",\"index\":0,\"content_block\":{\"type\":\"text\",\"text\":\"\"}}\n\n",
            "event: content_block_delta\ndata: {\"type\":\"content_block_delta\",\"index\":0,\"delta\":{\"type\":\"text_delta\",\"text\":\"Read\"}}\n\n",
            "event: content_block_delta\ndata: {\"type\":\"content_block_delta\",\"index\":0,\"delta\":{\"type\":\"text_delta\",\"text\":\"ing.\"}}\n\n",
            "event: content_block_stop\ndata: {\"type\":\"content_block_stop\",\"index\":0}\n\n",
            "event: content_block_start\ndata: {\"type\":\"content_block_start\",\"index\":1,\"content_block\":{\"type\":\"tool_use\",\"id\":\"toolu_x\",\"name\":\"view_file\"}}\n\n",
            "event: content_block_delta\ndata: {\"type\":\"content_block_delta\",\"index\":1,\"delta\":{\"type\":\"input_json_delta\",\"partial_json\":\"{\\\"path\\\":\"}}\n\n",
            "event: content_block_delta\ndata: {\"type\":\"content_block_delta\",\"index\":1,\"delta\":{\"type\":\"input_json_delta\",\"partial_json\":\"\\\"main.rs\\\"}\"}}\n\n",
            "event: content_block_stop\ndata: {\"type\":\"content_block_stop\",\"index\":1}\n\n",
            "event: message_delta\ndata: {\"type\":\"message_delta\",\"delta\":{\"stop_reason\":\"tool_use\"},\"usage\":{\"output_tokens\":33}}\n\n",
            "event: message_stop\ndata: {\"type\":\"message_stop\"}\n\n",
        ];
        raw.iter().map(|s| s.as_bytes().to_vec()).collect()
    }

    /// Drive the stream to completion and assert the decoded events
    /// reconstruct the right text, a tool_use with concatenated args, and
    /// the terminal stop reason + usage. Shared by the three framing
    /// variants below.
    async fn assert_canonical(mut s: MessagesSseStream) {
        let mut text = String::new();
        let mut tool_id = String::new();
        let mut tool_name = String::new();
        let mut tool_args = String::new();
        let mut stop: Option<StopReason> = None;
        let mut out_tokens: Option<i32> = None;
        let mut in_tokens: Option<i32> = None;
        let mut saw_stop_event = false;

        while let Some(ev) = s.next().await {
            match ev.unwrap() {
                StreamEvent::MessageStart { message } => {
                    in_tokens = message.usage.and_then(|u| u.input_tokens);
                }
                StreamEvent::ContentBlockStart {
                    content_block: Block::ToolUse { id, name, .. },
                    ..
                } => {
                    tool_id = id;
                    tool_name = name;
                }
                StreamEvent::ContentBlockDelta { delta, .. } => match delta {
                    BlockDelta::TextDelta { text: t } => text.push_str(&t),
                    BlockDelta::InputJsonDelta { partial_json } => tool_args.push_str(&partial_json),
                    _ => {}
                },
                StreamEvent::MessageDelta { delta, usage } => {
                    stop = delta.stop_reason;
                    out_tokens = usage.and_then(|u| u.output_tokens);
                }
                StreamEvent::MessageStop => saw_stop_event = true,
                _ => {}
            }
        }

        assert_eq!(text, "Reading.");
        assert_eq!(tool_id, "toolu_x");
        assert_eq!(tool_name, "view_file");
        // The two fragments concatenate to valid JSON.
        let parsed: serde_json::Value = serde_json::from_str(&tool_args).unwrap();
        assert_eq!(parsed["path"], "main.rs");
        assert_eq!(stop, Some(StopReason::ToolUse));
        assert_eq!(out_tokens, Some(33));
        assert_eq!(in_tokens, Some(12));
        assert!(saw_stop_event);
    }

    #[tokio::test]
    async fn decodes_canonical_sequence_one_chunk() {
        // Every frame concatenated into a single byte chunk.
        let blob: Vec<u8> = canonical_frames().concat();
        let s = MessagesSseStream::new(bytes_from(&[&blob]));
        assert_canonical(s).await;
    }

    #[tokio::test]
    async fn decodes_canonical_sequence_split_mid_frame() {
        // Concatenate all frames, then re-chop into chunks that fall in
        // the MIDDLE of frames (every 17 bytes), exercising partial-frame
        // buffering across chunk boundaries.
        let blob: Vec<u8> = canonical_frames().concat();
        let chunks: Vec<&[u8]> = blob.chunks(17).collect();
        let s = MessagesSseStream::new(bytes_from(&chunks));
        assert_canonical(s).await;
    }

    #[tokio::test]
    async fn decodes_canonical_sequence_crlf() {
        // Browser fetch surfaces SSE with CRLF frame separators (the wasm
        // gotcha). Rewrite every "\n" to "\r\n" and re-chop mid-frame.
        let blob: Vec<u8> = canonical_frames().concat();
        let crlf: Vec<u8> = String::from_utf8(blob)
            .unwrap()
            .replace('\n', "\r\n")
            .into_bytes();
        let chunks: Vec<&[u8]> = crlf.chunks(13).collect();
        let s = MessagesSseStream::new(bytes_from(&chunks));
        assert_canonical(s).await;
    }

    #[tokio::test]
    async fn ignores_ping_and_unknown_events() {
        let frames = [
            "event: ping\ndata: {\"type\":\"ping\"}\n\n".as_bytes(),
            "event: message_stop\ndata: {\"type\":\"message_stop\"}\n\n".as_bytes(),
        ];
        let mut s = MessagesSseStream::new(bytes_from(&frames));
        let first = s.next().await.unwrap().unwrap();
        assert_eq!(first, StreamEvent::Ping);
        let second = s.next().await.unwrap().unwrap();
        assert_eq!(second, StreamEvent::MessageStop);
        assert!(s.next().await.is_none());
    }

    /// Drain a stream into the list of decoded event variant tags (cheap,
    /// order-preserving), surfacing decode errors as the literal "ERR".
    async fn collect_kinds(mut s: MessagesSseStream) -> Vec<String> {
        let mut out = Vec::new();
        while let Some(ev) = s.next().await {
            match ev {
                Ok(StreamEvent::MessageStart { .. }) => out.push("message_start".into()),
                Ok(StreamEvent::ContentBlockStart { .. }) => out.push("content_block_start".into()),
                Ok(StreamEvent::ContentBlockDelta { .. }) => out.push("content_block_delta".into()),
                Ok(StreamEvent::ContentBlockStop { .. }) => out.push("content_block_stop".into()),
                Ok(StreamEvent::MessageDelta { .. }) => out.push("message_delta".into()),
                Ok(StreamEvent::MessageStop) => out.push("message_stop".into()),
                Ok(StreamEvent::Ping) => out.push("ping".into()),
                Ok(StreamEvent::Error { .. }) => out.push("error".into()),
                Ok(StreamEvent::Unknown) => out.push("unknown".into()),
                Err(_) => out.push("ERR".into()),
            }
        }
        out
    }

    /// REGRESSION: SSE permits the stream's LAST event to omit the trailing
    /// blank line (WHATWG event-stream rule). The Gemini decoder handles this
    /// with `take_remaining`; the Anthropic decoder originally only re-ran
    /// `take_frame` at EOF, which requires a blank-line boundary — so a final
    /// `data: {...}` with no trailing `\n\n` was silently dropped. For
    /// Anthropic that frame can be the `message_delta` carrying `stop_reason`
    /// + cumulative `output_tokens`, OR the `message_stop`.
    #[tokio::test]
    async fn flushes_final_frame_without_trailing_blank_line() {
        let frames = [
            "event: content_block_stop\ndata: {\"type\":\"content_block_stop\",\"index\":0}\n\n".as_bytes(),
            // Final frame: complete event, NO trailing blank line.
            "event: message_delta\ndata: {\"type\":\"message_delta\",\"delta\":{\"stop_reason\":\"end_turn\"},\"usage\":{\"output_tokens\":7}}".as_bytes(),
        ];
        let s = MessagesSseStream::new(bytes_from(&frames));
        let kinds = collect_kinds(s).await;
        assert_eq!(
            kinds,
            vec!["content_block_stop", "message_delta"],
            "the final unterminated frame must still be decoded, not dropped at EOF"
        );
    }

    /// A single complete frame with NO terminator at all — the whole stream is
    /// one EOF flush.
    #[tokio::test]
    async fn flushes_single_unterminated_frame() {
        let frames =
            ["event: message_stop\ndata: {\"type\":\"message_stop\"}".as_bytes()];
        let s = MessagesSseStream::new(bytes_from(&frames));
        assert_eq!(collect_kinds(s).await, vec!["message_stop"]);
    }

    /// Final frame terminated by a single `\n` (not a blank line) at EOF — the
    /// SSE field is complete but the event-terminating blank line is missing.
    #[tokio::test]
    async fn flushes_final_frame_with_single_newline() {
        let frames =
            ["event: message_stop\ndata: {\"type\":\"message_stop\"}\n".as_bytes()];
        let s = MessagesSseStream::new(bytes_from(&frames));
        assert_eq!(collect_kinds(s).await, vec!["message_stop"]);
    }

    /// A `:`-comment / keepalive line and a bare blank-line heartbeat must not
    /// produce a phantom event (empty `data:` payload → skipped, not decoded
    /// as an error). Anthropic sends periodic `event: ping` but a raw `:`
    /// comment is also legal SSE.
    #[tokio::test]
    async fn skips_comment_and_blank_heartbeat_frames() {
        let frames = [
            ": this is an SSE comment / keepalive\n\n".as_bytes(),
            "\n".as_bytes(), // bare blank-line heartbeat (no data:)
            "event: message_stop\ndata: {\"type\":\"message_stop\"}\n\n".as_bytes(),
        ];
        let s = MessagesSseStream::new(bytes_from(&frames));
        assert_eq!(
            collect_kinds(s).await,
            vec!["message_stop"],
            "comment + blank-line frames must be skipped, not surfaced as events/errors"
        );
    }

    /// A multi-line `data:` field (two `data:` lines in one frame) is
    /// newline-joined before JSON decode — the SSE spec way to carry a payload
    /// containing a newline. (Anthropic doesn't currently split JSON this way,
    /// but the decoder must follow the spec.)
    #[tokio::test]
    async fn joins_multiline_data_field() {
        // {"type":"message_stop"} split across two data: lines.
        let frames =
            ["event: message_stop\ndata: {\"type\":\ndata: \"message_stop\"}\n\n".as_bytes()];
        let s = MessagesSseStream::new(bytes_from(&frames));
        assert_eq!(collect_kinds(s).await, vec!["message_stop"]);
    }

    /// A multibyte UTF-8 character ("é" = 0xC3 0xA9) split across two network
    /// chunks must not corrupt the decoded text. The frame stays buffered
    /// until the blank-line terminator, by which point the char is whole.
    #[tokio::test]
    async fn multibyte_char_split_across_chunks() {
        let full = "event: content_block_delta\ndata: {\"type\":\"content_block_delta\",\"index\":0,\"delta\":{\"type\":\"text_delta\",\"text\":\"é\"}}\n\n";
        let raw = full.as_bytes();
        let split_at = raw.iter().position(|&b| b == 0xC3).unwrap();
        let (head, tail) = raw.split_at(split_at + 1); // 0xC3 lead byte in head
        let mut s = MessagesSseStream::new(bytes_from(&[head, tail]));
        let ev = s.next().await.unwrap().unwrap();
        match ev {
            StreamEvent::ContentBlockDelta {
                delta: BlockDelta::TextDelta { text },
                ..
            } => assert_eq!(text, "é"),
            other => panic!("expected a text_delta carrying é, got {other:?}"),
        }
        assert!(s.next().await.is_none());
    }

    /// A genuinely malformed JSON `data:` payload surfaces as an `Err` item
    /// (not a panic, not a silent drop).
    #[tokio::test]
    async fn malformed_json_yields_error_not_panic() {
        let frames = ["event: message_delta\ndata: {not valid json}\n\n".as_bytes()];
        let mut s = MessagesSseStream::new(bytes_from(&frames));
        let item = s.next().await.unwrap();
        assert!(item.is_err(), "malformed JSON must be an Err, got {item:?}");
    }

    /// Accumulate a multi-block stream EXACTLY as `loop.rs::run_turn` does —
    /// text deltas across TWO separate text blocks (indices 0 and 2) all fold
    /// into one text string; `input_json_delta` fragments route to the right
    /// tool block by `index`; and a `tool_use` block with ZERO input deltas
    /// (no args) yields an empty fragment (the loop resolves that to `{}`).
    /// This locks the per-index routing contract the loop relies on.
    #[tokio::test]
    async fn accumulates_multiple_blocks_with_empty_and_filled_tool_args() {
        let frames: Vec<&[u8]> = vec![
            "event: message_start\ndata: {\"type\":\"message_start\",\"message\":{\"id\":\"m\",\"usage\":{\"input_tokens\":5,\"output_tokens\":1}}}\n\n".as_bytes(),
            // Block 0: text.
            "event: content_block_start\ndata: {\"type\":\"content_block_start\",\"index\":0,\"content_block\":{\"type\":\"text\",\"text\":\"\"}}\n\n".as_bytes(),
            "event: content_block_delta\ndata: {\"type\":\"content_block_delta\",\"index\":0,\"delta\":{\"type\":\"text_delta\",\"text\":\"foo \"}}\n\n".as_bytes(),
            "event: content_block_stop\ndata: {\"type\":\"content_block_stop\",\"index\":0}\n\n".as_bytes(),
            // Block 1: tool_use with NO input deltas (empty args).
            "event: content_block_start\ndata: {\"type\":\"content_block_start\",\"index\":1,\"content_block\":{\"type\":\"tool_use\",\"id\":\"toolu_empty\",\"name\":\"list_subdomains\"}}\n\n".as_bytes(),
            "event: content_block_stop\ndata: {\"type\":\"content_block_stop\",\"index\":1}\n\n".as_bytes(),
            // Block 2: more text — must concat with block 0's text.
            "event: content_block_start\ndata: {\"type\":\"content_block_start\",\"index\":2,\"content_block\":{\"type\":\"text\",\"text\":\"\"}}\n\n".as_bytes(),
            "event: content_block_delta\ndata: {\"type\":\"content_block_delta\",\"index\":2,\"delta\":{\"type\":\"text_delta\",\"text\":\"bar\"}}\n\n".as_bytes(),
            "event: content_block_stop\ndata: {\"type\":\"content_block_stop\",\"index\":2}\n\n".as_bytes(),
            // Block 3: tool_use WITH args split across two fragments.
            "event: content_block_start\ndata: {\"type\":\"content_block_start\",\"index\":3,\"content_block\":{\"type\":\"tool_use\",\"id\":\"toolu_full\",\"name\":\"view_file\"}}\n\n".as_bytes(),
            "event: content_block_delta\ndata: {\"type\":\"content_block_delta\",\"index\":3,\"delta\":{\"type\":\"input_json_delta\",\"partial_json\":\"{\\\"path\\\":\"}}\n\n".as_bytes(),
            "event: content_block_delta\ndata: {\"type\":\"content_block_delta\",\"index\":3,\"delta\":{\"type\":\"input_json_delta\",\"partial_json\":\"\\\"a.rs\\\"}\"}}\n\n".as_bytes(),
            "event: content_block_stop\ndata: {\"type\":\"content_block_stop\",\"index\":3}\n\n".as_bytes(),
            "event: message_delta\ndata: {\"type\":\"message_delta\",\"delta\":{\"stop_reason\":\"tool_use\"},\"usage\":{\"output_tokens\":20}}\n\n".as_bytes(),
            "event: message_stop\ndata: {\"type\":\"message_stop\"}\n\n".as_bytes(),
        ];
        let mut s = MessagesSseStream::new(bytes_from(&frames));

        let mut text = String::new();
        // index → (id, name, concatenated args)
        let mut tools: std::collections::BTreeMap<u32, (String, String, String)> =
            std::collections::BTreeMap::new();
        while let Some(ev) = s.next().await {
            match ev.unwrap() {
                StreamEvent::ContentBlockStart {
                    index,
                    content_block: Block::ToolUse { id, name, .. },
                } => {
                    tools.insert(index, (id, name, String::new()));
                }
                StreamEvent::ContentBlockDelta { index, delta } => match delta {
                    BlockDelta::TextDelta { text: t } => text.push_str(&t),
                    BlockDelta::InputJsonDelta { partial_json } => {
                        if let Some(e) = tools.get_mut(&index) {
                            e.2.push_str(&partial_json);
                        }
                    }
                    _ => {}
                },
                _ => {}
            }
        }

        assert_eq!(text, "foo bar", "text from blocks 0 and 2 concatenates");
        // Empty-args tool: no fragments → empty string (loop → {}).
        let empty = &tools[&1];
        assert_eq!(empty.0, "toolu_empty");
        assert_eq!(empty.1, "list_subdomains");
        assert_eq!(empty.2, "", "no input deltas → empty args fragment");
        // Filled tool: fragments concatenate into valid JSON, no bleed from
        // the empty block.
        let full = &tools[&3];
        assert_eq!(full.0, "toolu_full");
        let parsed: serde_json::Value = serde_json::from_str(&full.2).unwrap();
        assert_eq!(parsed["path"], "a.rs");
    }
}