io-harness 0.22.0

//! Anthropic provider over an own HTTP + SSE client.
//!
//! Anthropic's `/v1/messages` wire format differs from the OpenAI-style one:
//! `system` is top-level, tools carry an `input_schema`, and the stream is a
//! sequence of typed events (`content_block_start`, `content_block_delta` with
//! `text_delta` / `input_json_delta`, `message_delta` carrying output-token
//! usage). Tool-call arguments arrive as `partial_json` fragments accumulated by
//! block index here — no vendor SDK.

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

use serde_json::json;

use super::{read_sse, CompletionRequest, CompletionResponse, Provider, ToolCall, Usage};
use crate::error::{Error, Result};

/// The request deadline this provider uses unless [`Anthropic::with_timeout`]
/// replaces it.
pub use crate::net::REQUEST_TIMEOUT;

const ENDPOINT: &str = "https://api.anthropic.com/v1/messages";
const API_VERSION: &str = "2023-06-01";
/// Anthropic versions a server tool by a dated `type` string, so the constant is
/// one line to change when the vendor supersedes it — and the body test names it,
/// so a stale one fails here rather than on the wire.
const WEB_SEARCH_TYPE: &str = "web_search_20250305";
const WEB_FETCH_TYPE: &str = "web_fetch_20250910";
/// Web fetch is beta-gated; web search is not. The header is sent only when fetch
/// is asked for, so a search-only request is byte-identical to what it would have
/// been without the feature.
const WEB_FETCH_BETA: &str = "web-fetch-2025-09-10";
// ponytail: Anthropic requires max_tokens; fixed cap. Thread through from the
// contract if agent outputs get truncated.
const MAX_TOKENS: u64 = 8192;

/// An Anthropic-backed [`Provider`].
///
/// ```no_run
/// use io_harness::{run_with, Anthropic, ApproveAll, Policy, Store, TaskContract, Verification};
///
/// # async fn demo() -> io_harness::Result<()> {
/// // `ANTHROPIC_API_KEY` and `ANTHROPIC_MODEL`; the key is read here and never
/// // logged. `Anthropic::new` takes both explicitly when they come from your own
/// // configuration rather than the environment.
/// let provider = Anthropic::from_env()?;
///
/// let contract = TaskContract::workspace(
///     "summarise the repo's README into NOTES.md",
///     "/path/to/repo",
///     Verification::WorkspaceFileContains { file: "NOTES.md".into(), needle: "#".into() },
/// );
/// let policy = Policy::default().layer("app").allow_read("*").allow_write("NOTES.md");
/// let result = run_with(&contract, &provider, &Store::memory()?, &policy, &ApproveAll).await?;
/// println!("{:?}", result.outcome);
/// # Ok(())
/// # }
/// ```
///
/// The harness contributes `api.anthropic.com` as the `provider` policy layer, so
/// a deny-by-default network policy still reaches this model and the trace records
/// why it was allowed.
pub struct Anthropic {
    client: reqwest::Client,
    api_key: String,
    model: String,
    endpoint: String,
}

impl Anthropic {
    /// Build from an explicit key and model slug (e.g. `claude-sonnet-4`).
    pub fn new(api_key: impl Into<String>, model: impl Into<String>) -> Self {
        Self {
            client: crate::net::http_client(),
            api_key: api_key.into(),
            model: model.into(),
            endpoint: ENDPOINT.to_string(),
        }
    }

    /// Set the deadline for one request, replacing the [`REQUEST_TIMEOUT`] default.
    ///
    /// For the case [`REQUEST_TIMEOUT`] names and could not serve until now: a
    /// model slower than ten minutes per completion, or a caller who would rather
    /// abandon a hung socket sooner than the default does. Rebuilds the client, so
    /// call it before handing the provider to a run.
    pub fn with_timeout(mut self, timeout: Duration) -> Self {
        self.client = crate::net::http_client_with_timeout(timeout);
        self
    }

    /// The same provider pointed at `endpoint` with `timeout` as its deadline, so
    /// the failure tests can drive the real HTTP and SSE path against a local
    /// socket. Test-only: the endpoint is not configurable in the public API.
    #[cfg(test)]
    pub(crate) fn at(endpoint: impl Into<String>, timeout: std::time::Duration) -> Self {
        Self {
            client: crate::net::http_client_with_timeout(timeout),
            api_key: "test-key".into(),
            model: "test-model".into(),
            endpoint: endpoint.into(),
        }
    }

    /// Build from the environment: `ANTHROPIC_API_KEY` (required) and
    /// `ANTHROPIC_MODEL` (required — no default guessed). The key is read here
    /// and never logged.
    pub fn from_env() -> Result<Self> {
        let api_key = std::env::var("ANTHROPIC_API_KEY")
            .map_err(|_| Error::Config("ANTHROPIC_API_KEY is not set".into()))?;
        let model = std::env::var("ANTHROPIC_MODEL")
            .map_err(|_| Error::Config("ANTHROPIC_MODEL is not set".into()))?;
        Ok(Self::new(api_key, model))
    }

    fn body(&self, request: &CompletionRequest) -> serde_json::Value {
        let mut tools: Vec<serde_json::Value> = request
            .tools
            .iter()
            .map(|t| {
                json!({
                    "name": t.name,
                    "description": t.description,
                    "input_schema": t.parameters,
                })
            })
            .collect();
        // 0.22.0 — provider-executed search and fetch are declared as tools in the
        // same array, distinguished by a dated `type` rather than an
        // `input_schema`: Anthropic runs them itself and the model never sends
        // this crate a call to dispatch.
        tools.extend(Self::web_tools(request.web.as_ref()));

        json!({
            // 0.21.0 — a per-request model override, for a named agent definition
            // spawned into a tree that shares this one provider. `None` is the
            // model this provider was constructed with.
            "model": request.model.as_deref().unwrap_or(&self.model),
            "max_tokens": MAX_TOKENS,
            "stream": true,
            "system": request.system,
            "messages": [
                { "role": "user", "content": Self::user_content(request) },
            ],
            "tools": tools,
        })
    }

    /// The server-tool entries a [`WebAccess`](crate::WebAccess) declaration adds
    /// to the `tools` array, in Anthropic's shape.
    ///
    /// Empty for `None` and for a declaration with nothing switched on, which is
    /// what keeps a non-searching request's body byte-identical to 0.21.0's.
    fn web_tools(web: Option<&crate::web::WebAccess>) -> Vec<serde_json::Value> {
        let Some(web) = web.filter(|w| w.enabled()) else {
            return Vec::new();
        };
        let (allowed, blocked) = web.vendor_filter();
        let entry = |type_: &str, name: &str| {
            let mut tool = json!({ "type": type_, "name": name });
            let map = tool.as_object_mut().expect("a json object");
            if let Some(uses) = web.max_uses {
                map.insert("max_uses".into(), json!(uses));
            }
            if !allowed.is_empty() {
                map.insert("allowed_domains".into(), json!(allowed));
            }
            if !blocked.is_empty() {
                map.insert("blocked_domains".into(), json!(blocked));
            }
            tool
        };
        let mut tools = Vec::new();
        if web.search {
            tools.push(entry(WEB_SEARCH_TYPE, "web_search"));
        }
        if web.fetch {
            tools.push(entry(WEB_FETCH_TYPE, "web_fetch"));
        }
        tools
    }

    /// The user turn's `content`: a bare string when there is no image, and
    /// Anthropic's content-block array when there is.
    ///
    /// Text-only requests keep exactly the body 0.14.0 sent, so upgrading
    /// changes nothing on the wire for a caller who sends no image.
    #[cfg(feature = "media")]
    fn user_content(request: &CompletionRequest) -> serde_json::Value {
        if request.media.is_empty() {
            return json!(request.user);
        }
        // Images before text: what Anthropic's own guidance recommends for
        // prompts that ask a question about an image.
        let mut parts: Vec<serde_json::Value> = request
            .media
            .iter()
            .map(|m| {
                json!({
                    "type": "image",
                    "source": {
                        "type": "base64",
                        "media_type": m.media_type,
                        "data": m.base64,
                    },
                })
            })
            .collect();
        parts.push(json!({ "type": "text", "text": request.user }));
        json!(parts)
    }

    #[cfg(not(feature = "media"))]
    fn user_content(request: &CompletionRequest) -> serde_json::Value {
        json!(request.user)
    }
}

impl Provider for Anthropic {
    fn name(&self) -> &str {
        "anthropic"
    }

    fn endpoint(&self) -> Option<&str> {
        Some(&self.endpoint)
    }

    #[cfg(feature = "media")]
    fn accepts_images(&self) -> bool {
        true
    }

    async fn complete(&self, request: CompletionRequest) -> Result<CompletionResponse> {
        self.stream(request, &|_| {}).await
    }

    async fn complete_streaming(
        &self,
        request: CompletionRequest,
        on_token: &(dyn Fn(&str) + Send + Sync),
    ) -> Result<CompletionResponse> {
        self.stream(request, on_token).await
    }
}

impl Anthropic {
    /// One completion, with each text delta handed to `on_token` on its way into
    /// the accumulator.
    ///
    /// Both trait methods are this function; `complete` passes a sink that does
    /// nothing. The stream was always consumed delta by delta — 0.20.0 only stops
    /// throwing each one away before anything else can see it.
    async fn stream(
        &self,
        request: CompletionRequest,
        on_token: &(dyn Fn(&str) + Send + Sync),
    ) -> Result<CompletionResponse> {
        #[cfg(feature = "media")]
        super::ensure_media_accepted(self.name(), self.accepts_images(), &request)?;
        // Time to first token is measured from here — before the socket is
        // opened — because that is the wait a caller actually experiences. It
        // therefore includes connection setup, which `CONTRACT.md` states rather
        // than quietly excluding to produce a flattering number.
        let sent = Instant::now();
        let mut post = self
            .client
            .post(&self.endpoint)
            .header("x-api-key", &self.api_key)
            .header("anthropic-version", API_VERSION);
        // Only when fetch is asked for: an unnecessary beta header opts a request
        // into a preview it does not use, and a search-only request should send
        // exactly what 0.21.0 sent plus its tool entry.
        if request.web.as_ref().is_some_and(|w| w.fetch) {
            post = post.header("anthropic-beta", WEB_FETCH_BETA);
        }
        let resp = post.json(&self.body(&request)).send().await?;
        let resp = super::ensure_success(resp).await?;

        let mut acc = Accumulator::since(sent);
        read_sse(resp, |data| {
            if let Ok(value) = serde_json::from_str::<serde_json::Value>(data) {
                if value.get("type").and_then(|t| t.as_str()) == Some("message_stop") {
                    return true;
                }
                // Before `ingest`, so the delta a caller renders is the same string
                // the accumulated text ends up carrying rather than a re-derivation
                // of it.
                if let Some(delta) = text_delta(&value) {
                    on_token(delta);
                }
                acc.ingest(&value);
            }
            false
        })
        .await?;
        // A stream where nothing at all parsed is a failure, not a quiet model.
        super::ensure_parsed(acc.finish())
    }
}

/// The assistant-text delta an event carries, if it carries one.
///
/// Text only. A `input_json_delta` fragment of a tool call is not renderable and
/// is not safe to act on — the accumulator owns reassembling those.
fn text_delta(value: &serde_json::Value) -> Option<&str> {
    if value.get("type").and_then(|t| t.as_str()) != Some("content_block_delta") {
        return None;
    }
    let delta = value.get("delta")?;
    if delta.get("type").and_then(|t| t.as_str()) != Some("text_delta") {
        return None;
    }
    delta.get("text")?.as_str()
}

/// Accumulates Anthropic's typed stream events into one response.
#[derive(Default)]
struct Accumulator {
    text: String,
    /// block index -> (tool name, input-json fragments joined)
    tool_calls: BTreeMap<u64, (String, String)>,
    input_tokens: u64,
    output_tokens: u64,
    /// 0.18.0 — the cache breakdown of `input_tokens`, the model that answered,
    /// why it stopped, and the provider-executed tool requests it made. All
    /// carried on events the accumulator already reads and, until now, dropped.
    cache_write_tokens: u64,
    cache_read_tokens: u64,
    server_tool_requests: u64,
    model: Option<String>,
    finish_reason: Option<String>,
    /// 0.22.0 — what the provider cited, and what its own server tools did. The
    /// tool name is remembered per `tool_use_id` from the `server_tool_use` block
    /// that asked for it, so the result block a few events later is attributed to
    /// `web_search` or `web_fetch` rather than guessed at.
    citations: Vec<crate::web::Citation>,
    server_tools: Vec<crate::web::ServerToolCall>,
    server_tool_names: BTreeMap<String, String>,
    /// When the request was sent, and the elapsed time at the first
    /// content-bearing event. `None` in a unit test that feeds events directly:
    /// nothing measured the wait, so the response reports no TTFT rather than
    /// zero.
    sent: Option<Instant>,
    ttft_ms: Option<u64>,
}

impl Accumulator {
    /// An accumulator that measures time to first token from `sent`.
    fn since(sent: Instant) -> Self {
        Self {
            sent: Some(sent),
            ..Default::default()
        }
    }

    /// The first content-bearing event stops the TTFT clock. Later events do
    /// not: `Option::get_or_insert_with` is what makes it first-token rather
    /// than last-token.
    fn mark_first_token(&mut self) {
        if let Some(sent) = self.sent {
            self.ttft_ms
                .get_or_insert(sent.elapsed().as_millis() as u64);
        }
    }

    fn ingest(&mut self, value: &serde_json::Value) {
        let index = || value.get("index").and_then(|i| i.as_u64()).unwrap_or(0);
        match value.get("type").and_then(|t| t.as_str()) {
            Some("message_start") => {
                if let Some(n) = value
                    .pointer("/message/usage/input_tokens")
                    .and_then(|v| v.as_u64())
                {
                    self.input_tokens = n;
                }
                if let Some(m) = value
                    .pointer("/message/model")
                    .and_then(|v| v.as_str())
                    .filter(|m| !m.is_empty())
                {
                    self.model = Some(m.to_string());
                }
                self.ingest_usage(value.pointer("/message/usage"));
            }
            Some("content_block_start") => {
                self.mark_first_token();
                if let Some(cb) = value.get("content_block") {
                    match cb.get("type").and_then(|t| t.as_str()) {
                        Some("tool_use") => {
                            let name = cb
                                .get("name")
                                .and_then(|n| n.as_str())
                                .unwrap_or_default()
                                .to_string();
                            self.tool_calls.entry(index()).or_default().0 = name;
                        }
                        // The model asking Anthropic to run a search: not a call
                        // this crate dispatches, so it never joins `tool_calls`.
                        // Its id is kept so the result block can be named.
                        Some("server_tool_use") => self.ingest_server_tool_use(cb),
                        // The result, which Anthropic sends whole rather than as
                        // deltas — including, inside an HTTP 200, the error object
                        // that means the search failed.
                        Some(t) if t.ends_with("_tool_result") => {
                            self.ingest_server_tool_result(t, cb);
                        }
                        // A text block can arrive with its citations already
                        // attached when the stream is replayed or recorded.
                        _ => self.ingest_citations(cb.get("citations")),
                    }
                }
            }
            Some("content_block_delta") => {
                self.mark_first_token();
                let delta = value.get("delta");
                match delta.and_then(|d| d.get("type")).and_then(|t| t.as_str()) {
                    Some("text_delta") => {
                        if let Some(t) = delta.and_then(|d| d.get("text")).and_then(|t| t.as_str())
                        {
                            self.text.push_str(t);
                        }
                    }
                    Some("input_json_delta") => {
                        if let Some(p) = delta
                            .and_then(|d| d.get("partial_json"))
                            .and_then(|p| p.as_str())
                        {
                            self.tool_calls.entry(index()).or_default().1.push_str(p);
                        }
                    }
                    // 0.22.0 — a source arriving mid-sentence, one delta per
                    // citation, on the text block it supports.
                    Some("citations_delta") => {
                        self.push_citation(delta.and_then(|d| d.get("citation")));
                    }
                    _ => {}
                }
            }
            Some("message_delta") => {
                if let Some(n) = value
                    .pointer("/usage/output_tokens")
                    .and_then(|v| v.as_u64())
                {
                    self.output_tokens = n;
                }
                if let Some(r) = value
                    .pointer("/delta/stop_reason")
                    .and_then(|v| v.as_str())
                    .filter(|r| !r.is_empty())
                {
                    self.finish_reason = Some(r.to_string());
                }
                self.ingest_usage(value.pointer("/usage"));
            }
            _ => {}
        }
    }

    /// A `server_tool_use` block: remember which tool the id belongs to, so the
    /// result block that follows is named rather than assumed to be a search.
    fn ingest_server_tool_use(&mut self, block: &serde_json::Value) {
        let (Some(id), Some(name)) = (
            block.get("id").and_then(|v| v.as_str()),
            block.get("name").and_then(|v| v.as_str()),
        ) else {
            return;
        };
        self.server_tool_names
            .insert(id.to_string(), name.to_string());
    }

    /// A `web_search_tool_result` / `web_fetch_tool_result` block.
    ///
    /// The failure this release exists to catch lives here: the vendor reports a
    /// broken search as an error *object* inside a 200, and a parser that only
    /// looks for results reads it as a search that found nothing. The `_error`
    /// content type is what tells the two apart, and its `error_code` is recorded
    /// in the vendor's own words.
    fn ingest_server_tool_result(&mut self, block_type: &str, block: &serde_json::Value) {
        let tool = block
            .get("tool_use_id")
            .and_then(|v| v.as_str())
            .and_then(|id| self.server_tool_names.get(id))
            .cloned()
            // The result block names its own kind, so an unmatched id still
            // records the right tool rather than a guess.
            .unwrap_or_else(|| block_type.trim_end_matches("_tool_result").to_string());
        let content = block.get("content");
        let error = content
            .and_then(|c| c.get("type"))
            .and_then(|t| t.as_str())
            .filter(|t| t.ends_with("_error"))
            .and(
                content
                    .and_then(|c| c.get("error_code"))
                    .and_then(|c| c.as_str()),
            );
        self.server_tools.push(match error {
            Some(code) => crate::web::ServerToolCall::failed("anthropic", tool, code),
            None => crate::web::ServerToolCall::ok("anthropic", tool),
        });
        // A fetch result carries the page it read; a search result carries the
        // pages it found. Both are sources the answer may draw on, so both are
        // recorded as citations rather than only the ones the model quotes.
        if let Some(results) = content.and_then(|c| c.as_array()) {
            for result in results {
                self.push_citation(Some(result));
            }
        }
    }

    /// Every citation on a block that carries a `citations` array.
    fn ingest_citations(&mut self, citations: Option<&serde_json::Value>) {
        let Some(list) = citations.and_then(|c| c.as_array()) else {
            return;
        };
        for citation in list {
            self.push_citation(Some(citation));
        }
    }

    /// One citation, from whichever shape carried it. A block with no `url` is
    /// not a source and is skipped rather than recorded as an empty one.
    fn push_citation(&mut self, citation: Option<&serde_json::Value>) {
        let Some(url) = citation
            .and_then(|c| c.get("url"))
            .and_then(|u| u.as_str())
            .filter(|u| !u.is_empty())
        else {
            return;
        };
        let text = |key: &str| {
            citation
                .and_then(|c| c.get(key))
                .and_then(|v| v.as_str())
                .filter(|s| !s.is_empty())
                .map(str::to_string)
        };
        let found = crate::web::Citation {
            url: url.to_string(),
            title: text("title"),
            cited_text: text("cited_text"),
        };
        // A page cited twice in one answer is one source. Anthropic repeats the
        // url on every sentence it supports, and a trace with the same row forty
        // times is a trace nobody reads. The second mention still enriches the
        // first: a result block gives the url and title, and the citation delta a
        // few events later adds the quoted passage.
        if let Some(seen) = self.citations.iter_mut().find(|c| c.url == found.url) {
            seen.title = seen.title.take().or(found.title);
            seen.cited_text = seen.cited_text.take().or(found.cited_text);
            return;
        }
        self.citations.push(found);
    }

    /// The counters Anthropic reports inside a `usage` object, wherever that
    /// object arrives. Cache tokens land on `message_start`, the server-tool
    /// count can land on either event, and a field that is absent leaves the
    /// running value alone rather than resetting it to zero.
    fn ingest_usage(&mut self, usage: Option<&serde_json::Value>) {
        let Some(usage) = usage else { return };
        let get = |k: &str| usage.get(k).and_then(|v| v.as_u64());
        if let Some(n) = get("cache_creation_input_tokens") {
            self.cache_write_tokens = n;
        }
        if let Some(n) = get("cache_read_input_tokens") {
            self.cache_read_tokens = n;
        }
        // `server_tool_use` is an object of per-tool counters; their sum is the
        // number of billed requests, and summing rather than naming one keeps a
        // tool Anthropic adds later from being silently uncounted.
        if let Some(counts) = usage.get("server_tool_use").and_then(|v| v.as_object()) {
            let sum = counts.values().filter_map(|v| v.as_u64()).sum();
            if sum > 0 {
                self.server_tool_requests = sum;
            }
        }
    }

    fn finish(self) -> CompletionResponse {
        let tool_calls = self
            .tool_calls
            .into_values()
            .filter(|(name, _)| !name.is_empty())
            .map(|(name, args)| ToolCall {
                name,
                // An empty-arg tool call streams no partial_json; treat as {}.
                arguments: serde_json::from_str(if args.is_empty() { "{}" } else { &args })
                    .unwrap_or(serde_json::Value::Null),
            })
            .collect();

        // Anthropic's `input_tokens` EXCLUDES the cached ones — it reports the
        // three counts side by side and every one of them is billed — where the
        // OpenAI wire's `prompt_tokens` includes them. `Usage::prompt_tokens` is
        // defined as the whole prompt, so the vendors are reconciled here, at the
        // wire boundary, rather than leaving every reader of the trace to know
        // which vendor it came from. Before 0.18.0 the cached tokens were dropped
        // entirely, so a cache-heavy run under-reported its prompt.
        let prompt = self.input_tokens + self.cache_read_tokens + self.cache_write_tokens;
        // Anthropic reports no total, so this one is summed rather than taken as
        // reported.
        let total = prompt + self.output_tokens;
        CompletionResponse {
            text: if self.text.is_empty() {
                None
            } else {
                Some(self.text)
            },
            tool_calls,
            usage: (total > 0).then_some(Usage {
                prompt_tokens: prompt,
                completion_tokens: self.output_tokens,
                total_tokens: total,
                cache_read_tokens: self.cache_read_tokens,
                cache_write_tokens: self.cache_write_tokens,
                // Anthropic bills extended thinking inside `output_tokens` and
                // reports no separate figure, so this stays zero rather than
                // being guessed at from the text.
                reasoning_tokens: 0,
                server_tool_requests: self.server_tool_requests,
            }),
            model: self.model,
            finish_reason: self.finish_reason,
            ttft_ms: self.ttft_ms,
            citations: self.citations,
            server_tools: self.server_tools,
        }
    }
}

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

    #[test]
    fn body_maps_tools_to_input_schema_and_system_top_level() {
        let a = Anthropic::new("k", "claude-x");
        #[allow(clippy::needless_update)] // `media` is cfg'd out in the default build
        let req = CompletionRequest {
            system: "sys".into(),
            user: "hi".into(),
            tools: vec![ToolSpec {
                name: "write_file".into(),
                description: "w".into(),
                parameters: json!({"type":"object"}),
            }],
            ..Default::default()
        };
        let b = a.body(&req);
        assert_eq!(b["system"], "sys");
        assert_eq!(b["messages"][0]["content"], "hi");
        assert_eq!(b["tools"][0]["name"], "write_file");
        assert_eq!(b["tools"][0]["input_schema"], json!({"type":"object"}));
        assert!(b["max_tokens"].is_u64());
    }

    #[test]
    fn accumulates_tool_use_from_input_json_deltas_and_usage() {
        let mut acc = Accumulator::default();
        acc.ingest(&json!({"type":"message_start","message":{"usage":{"input_tokens":11}}}));
        acc.ingest(&json!({"type":"content_block_start","index":0,
            "content_block":{"type":"tool_use","id":"t1","name":"write_file","input":{}}}));
        acc.ingest(&json!({"type":"content_block_delta","index":0,
            "delta":{"type":"input_json_delta","partial_json":"{\"path\":\"a"}}));
        acc.ingest(&json!({"type":"content_block_delta","index":0,
            "delta":{"type":"input_json_delta","partial_json":".rs\",\"content\":\"x\"}"}}));
        acc.ingest(&json!({"type":"message_delta","usage":{"output_tokens":7}}));

        let out = acc.finish();
        assert_eq!(out.tool_calls.len(), 1);
        assert_eq!(out.tool_calls[0].name, "write_file");
        assert_eq!(out.tool_calls[0].arguments["path"], "a.rs");
        assert_eq!(out.tool_calls[0].arguments["content"], "x");
        let u = out.usage.unwrap();
        assert_eq!(u.prompt_tokens, 11);
        assert_eq!(u.completion_tokens, 7);
        assert_eq!(u.total_tokens, 18);
    }

    /// F3, F6, F7 — the counters Anthropic already sends and 0.17.0 dropped.
    #[test]
    fn cache_tokens_the_model_reports_reach_usage_with_the_model_and_stop_reason() {
        let mut acc = Accumulator::default();
        acc.ingest(&json!({"type":"message_start","message":{
            "model":"claude-sonnet-4-5",
            "usage":{"input_tokens":11,"cache_creation_input_tokens":300,
                     "cache_read_input_tokens":1_200}}}));
        acc.ingest(
            &json!({"type":"message_delta","delta":{"stop_reason":"max_tokens"},
            "usage":{"output_tokens":7,"server_tool_use":{"web_search_requests":2}}}),
        );

        let out = acc.finish();
        let u = out.usage.unwrap();
        assert_eq!(u.cache_write_tokens, 300);
        assert_eq!(u.cache_read_tokens, 1_200);
        assert_eq!(u.server_tool_requests, 2);
        // Anthropic reports `input_tokens` EXCLUDING the cached ones, so the
        // prompt is the three added together — all three are billed, and a
        // reader of the trace should not have to know which vendor wrote it.
        assert_eq!(u.prompt_tokens, 11 + 300 + 1_200);
        assert_eq!(u.total_tokens, 11 + 300 + 1_200 + 7);
        // Extended thinking is billed inside `output_tokens` and reported
        // nowhere separately, so this stays zero rather than being guessed at.
        assert_eq!(u.reasoning_tokens, 0);
        assert_eq!(out.model.as_deref(), Some("claude-sonnet-4-5"));
        assert_eq!(out.finish_reason.as_deref(), Some("max_tokens"));
    }

    /// The negative control for the test above: a stream reporting none of it
    /// yields zeros and `None`s, not an error and not a fabricated figure. This
    /// is also every pre-cache request, so it is the common case.
    #[test]
    fn a_stream_that_reports_no_cache_or_stop_reason_yields_zeros_and_nones() {
        let mut acc = Accumulator::default();
        acc.ingest(&json!({"type":"message_start","message":{"usage":{"input_tokens":11}}}));
        acc.ingest(&json!({"type":"message_delta","usage":{"output_tokens":7}}));

        let out = acc.finish();
        let u = out.usage.unwrap();
        assert_eq!((u.cache_read_tokens, u.cache_write_tokens), (0, 0));
        assert_eq!(u.server_tool_requests, 0);
        assert_eq!(u.prompt_tokens, 11);
        assert_eq!(u.total_tokens, 18);
        assert_eq!(out.model, None);
        assert_eq!(out.finish_reason, None);
        // Nothing measured the stream, so TTFT is unknown rather than instant.
        assert_eq!(out.ttft_ms, None);
    }

    /// F5, at the point of measurement: the clock stops on the FIRST
    /// content-bearing event and not on a later one.
    #[test]
    fn the_ttft_clock_stops_at_the_first_content_event() {
        let mut acc = Accumulator::since(Instant::now() - Duration::from_millis(40));
        acc.ingest(&json!({"type":"content_block_delta","index":0,
            "delta":{"type":"text_delta","text":"a"}}));
        let first = acc.ttft_ms.expect("measured");
        std::thread::sleep(Duration::from_millis(15));
        acc.ingest(&json!({"type":"content_block_delta","index":0,
            "delta":{"type":"text_delta","text":"b"}}));
        assert_eq!(acc.ttft_ms, Some(first), "a later chunk moved the clock");
        assert!(
            first >= 40,
            "the clock runs from the request, got {first}ms"
        );
    }

    #[test]
    fn accumulates_plain_text() {
        let mut acc = Accumulator::default();
        acc.ingest(&json!({"type":"content_block_delta","index":0,
            "delta":{"type":"text_delta","text":"hello "}}));
        acc.ingest(&json!({"type":"content_block_delta","index":0,
            "delta":{"type":"text_delta","text":"world"}}));
        let out = acc.finish();
        assert_eq!(out.text.as_deref(), Some("hello world"));
        assert!(out.tool_calls.is_empty());
    }
}

/// 0.22.0 — the provider-executed web tools: what is declared on the wire, what
/// header a beta-gated tool adds, and what comes back.
#[cfg(test)]
mod web_wire {
    use std::io::{Read, Write};
    use std::net::TcpListener;
    use std::sync::mpsc;

    use super::*;
    use crate::web::WebAccess;

    #[allow(clippy::needless_update)] // `media` is cfg'd out in the default build
    fn req(web: Option<WebAccess>) -> CompletionRequest {
        CompletionRequest {
            system: "sys".into(),
            user: "what shipped this week".into(),
            web,
            ..Default::default()
        }
    }

    /// F1 — one declaration, Anthropic's shape.
    #[test]
    fn a_declaration_becomes_dated_server_tool_entries() {
        let web = WebAccess::search()
            .with_fetch()
            .max_uses(5)
            .allow("docs.rs")
            .allow("crates.io");
        let b = Anthropic::new("k", "claude-x").body(&req(Some(web)));
        let tools = b["tools"].as_array().expect("a tools array");
        assert_eq!(tools.len(), 2, "search and fetch, got {tools:?}");
        assert_eq!(tools[0]["type"], WEB_SEARCH_TYPE);
        assert_eq!(tools[0]["name"], "web_search");
        assert_eq!(tools[0]["max_uses"], 5);
        assert_eq!(tools[0]["allowed_domains"], json!(["docs.rs", "crates.io"]));
        assert_eq!(tools[1]["type"], WEB_FETCH_TYPE);
        assert_eq!(tools[1]["name"], "web_fetch");
        // A vendor rejects both lists at once, so only the narrower one is sent.
        assert!(tools[0].get("blocked_domains").is_none());
    }

    /// The block-list alone reaches the vendor when there is no allow-list.
    #[test]
    fn a_block_list_alone_is_sent_as_one() {
        let b = Anthropic::new("k", "claude-x")
            .body(&req(Some(WebAccess::search().block("evil.test"))));
        assert_eq!(b["tools"][0]["blocked_domains"], json!(["evil.test"]));
        assert!(b["tools"][0].get("allowed_domains").is_none());
        assert!(b["tools"][0].get("max_uses").is_none(), "no cap declared");
    }

    /// NF3, the negative control: a request that declares nothing sends exactly
    /// the body 0.21.0 sent.
    #[test]
    fn no_declaration_sends_the_0_21_0_body() {
        let b = Anthropic::new("k", "claude-x").body(&req(None));
        assert_eq!(b["tools"], json!([]));
        // And a declaration with both switches off is the same as none: a filter
        // around a capability nobody asked for is not a capability.
        let off =
            Anthropic::new("k", "claude-x").body(&req(Some(WebAccess::default().allow("docs.rs"))));
        assert_eq!(off["tools"], json!([]));
    }

    /// F2 — the fetch beta header, and its absence.
    #[test]
    fn the_fetch_beta_header_is_sent_only_when_fetch_is_asked_for() {
        for (web, expected) in [
            (WebAccess::search().with_fetch(), true),
            (WebAccess::search(), false),
        ] {
            let head = head_of_request(web);
            assert_eq!(
                head.contains(&format!("anthropic-beta: {WEB_FETCH_BETA}")),
                expected,
                "wrong beta header for fetch={}, head was:\n{head}",
                expected
            );
        }
    }

    /// Send one real request at a local socket and hand back the request head, so
    /// the header assertion runs through the same code a live call does.
    fn head_of_request(web: WebAccess) -> String {
        let listener = TcpListener::bind("127.0.0.1:0").unwrap();
        let url = format!("http://{}/v1/messages", listener.local_addr().unwrap());
        let (tx, rx) = mpsc::channel();
        std::thread::spawn(move || {
            let Ok(mut stream) = listener.incoming().next().expect("one connection") else {
                return;
            };
            let mut seen = Vec::new();
            let mut byte = [0u8; 1];
            while stream.read(&mut byte).unwrap_or(0) == 1 {
                seen.push(byte[0]);
                if seen.ends_with(b"\r\n\r\n") {
                    break;
                }
            }
            let _ = tx.send(String::from_utf8_lossy(&seen).to_ascii_lowercase());
            let body = "data: {\"type\":\"message_stop\"}\n\n";
            let _ = stream.write_all(
                format!(
                    "HTTP/1.1 200 OK\r\nContent-Length: {}\r\n\r\n{body}",
                    body.len()
                )
                .as_bytes(),
            );
            let _ = stream.flush();
        });
        let provider = Anthropic::at(url, Duration::from_secs(5));
        let runtime = tokio::runtime::Builder::new_current_thread()
            .enable_all()
            .build()
            .unwrap();
        // The response is deliberately empty, so the call itself fails; the head
        // is what this test is about and it has already been captured.
        let _ = runtime.block_on(provider.complete(req(Some(web))));
        rx.recv().expect("the request head")
    }

    /// F3 — citations reach the response, deduplicated by url.
    #[test]
    fn citations_arrive_from_deltas_and_from_result_blocks() {
        let mut acc = Accumulator::default();
        acc.ingest(&json!({"type":"content_block_start","index":0,
            "content_block":{"type":"server_tool_use","id":"srvtoolu_1","name":"web_search"}}));
        acc.ingest(&json!({"type":"content_block_start","index":1,
            "content_block":{"type":"web_search_tool_result","tool_use_id":"srvtoolu_1",
                "content":[{"type":"web_search_result","url":"https://docs.rs/io-harness",
                            "title":"io-harness"}]}}));
        acc.ingest(&json!({"type":"content_block_delta","index":2,
            "delta":{"type":"text_delta","text":"0.22.0 adds web search"}}));
        acc.ingest(&json!({"type":"content_block_delta","index":2,
            "delta":{"type":"citations_delta","citation":{"type":"web_search_result_location",
                "url":"https://docs.rs/io-harness","title":"io-harness",
                "cited_text":"provider-executed web search"}}}));
        acc.ingest(
            &json!({"type":"message_delta","delta":{"stop_reason":"end_turn"},
            "usage":{"output_tokens":9,"server_tool_use":{"web_search_requests":1}}}),
        );

        let out = acc.finish();
        assert_eq!(
            out.citations.len(),
            1,
            "one page, cited twice: {:?}",
            out.citations
        );
        assert_eq!(out.citations[0].url, "https://docs.rs/io-harness");
        assert_eq!(out.citations[0].title.as_deref(), Some("io-harness"));
        assert_eq!(
            out.citations[0].cited_text.as_deref(),
            Some("provider-executed web search"),
            "the quoted passage arrives on the delta, not on the result block"
        );
        assert_eq!(
            out.server_tools,
            vec![crate::web::ServerToolCall::ok("anthropic", "web_search")]
        );
        assert_eq!(out.usage.unwrap().server_tool_requests, 1);
    }

    /// F4 — a search that failed inside an HTTP 200, and the negative control of
    /// one that succeeded and found nothing.
    #[test]
    fn a_search_that_failed_inside_a_200_is_recorded_as_a_failure() {
        let mut acc = Accumulator::default();
        acc.ingest(&json!({"type":"content_block_start","index":0,
            "content_block":{"type":"server_tool_use","id":"srvtoolu_1","name":"web_search"}}));
        acc.ingest(&json!({"type":"content_block_start","index":1,
            "content_block":{"type":"web_search_tool_result","tool_use_id":"srvtoolu_1",
                "content":{"type":"web_search_tool_result_error",
                           "error_code":"max_uses_exceeded"}}}));
        acc.ingest(&json!({"type":"message_delta","usage":{"output_tokens":3}}));

        let out = acc.finish();
        assert_eq!(
            out.server_tools,
            vec![crate::web::ServerToolCall::failed(
                "anthropic",
                "web_search",
                "max_uses_exceeded"
            )]
        );
        assert!(out.citations.is_empty(), "a failed search cites nothing");

        // The control: a search that worked and returned nothing is a SUCCESSFUL
        // call with no citations. Reading that as a failure — or reading the
        // failure above as this — is the defect the release exists to prevent.
        let mut acc = Accumulator::default();
        acc.ingest(&json!({"type":"content_block_start","index":0,
            "content_block":{"type":"server_tool_use","id":"srvtoolu_2","name":"web_search"}}));
        acc.ingest(&json!({"type":"content_block_start","index":1,
            "content_block":{"type":"web_search_tool_result","tool_use_id":"srvtoolu_2",
                "content":[]}}));
        acc.ingest(&json!({"type":"message_delta","usage":{"output_tokens":3}}));
        let out = acc.finish();
        assert_eq!(out.server_tools.len(), 1);
        assert!(out.server_tools[0].succeeded());
        assert!(out.citations.is_empty());
    }

    /// A fetch result is attributed to `web_fetch`, from the id its request block
    /// carried rather than from the block type alone.
    #[test]
    fn a_fetch_result_is_named_by_the_request_that_asked_for_it() {
        let mut acc = Accumulator::default();
        acc.ingest(&json!({"type":"content_block_start","index":0,
            "content_block":{"type":"server_tool_use","id":"srvtoolu_9","name":"web_fetch"}}));
        acc.ingest(&json!({"type":"content_block_start","index":1,
            "content_block":{"type":"web_fetch_tool_result","tool_use_id":"srvtoolu_9",
                "content":[{"type":"web_fetch_result","url":"https://example.test/page"}]}}));
        acc.ingest(&json!({"type":"message_delta","usage":{"output_tokens":2}}));

        let out = acc.finish();
        assert_eq!(out.server_tools[0].tool, "web_fetch");
        assert_eq!(out.citations[0].url, "https://example.test/page");
        // No title and no quote reported is `None`, not an empty string.
        assert_eq!(out.citations[0].title, None);
    }

    /// The negative control for the whole module: a 0.21.0-shaped stream carries
    /// no citations and no server-tool rows.
    #[test]
    fn a_stream_with_no_web_activity_reports_none() {
        let mut acc = Accumulator::default();
        acc.ingest(&json!({"type":"content_block_delta","index":0,
            "delta":{"type":"text_delta","text":"hello"}}));
        acc.ingest(&json!({"type":"message_delta","usage":{"output_tokens":1}}));
        let out = acc.finish();
        assert!(out.citations.is_empty());
        assert!(out.server_tools.is_empty());
        assert_eq!(out.usage.unwrap().server_tool_requests, 0);
    }
}

/// The image content-block shape, against Anthropic's documented format.
#[cfg(all(test, feature = "media"))]
mod media_wire {
    use super::*;
    use crate::provider::Media;

    #[allow(clippy::needless_update)] // `media` is cfg'd out in the default build
    fn req_with_image() -> CompletionRequest {
        CompletionRequest {
            system: "sys".into(),
            user: "what is this".into(),
            media: vec![Media::image("image/png", &[1, 2, 3]).unwrap()],
            ..Default::default()
        }
    }

    #[test]
    fn an_image_becomes_a_base64_source_block_before_the_text() {
        let b = Anthropic::new("k", "claude-x").body(&req_with_image());
        let content = &b["messages"][0]["content"];
        assert!(content.is_array(), "content must be blocks, got {content}");
        assert_eq!(content[0]["type"], "image");
        assert_eq!(content[0]["source"]["type"], "base64");
        assert_eq!(content[0]["source"]["media_type"], "image/png");
        assert_eq!(content[0]["source"]["data"], "AQID");
        // Text after the image, which is what Anthropic's guidance recommends
        // when the question is about the picture.
        assert_eq!(content[1]["type"], "text");
        assert_eq!(content[1]["text"], "what is this");
    }

    #[test]
    fn a_request_without_an_image_still_sends_a_bare_string() {
        // The negative control, and the compatibility guarantee: a text-only
        // body is byte-identical to the one 0.14.0 sent, so upgrading alone
        // changes nothing on the wire and invalidates no recording.
        #[allow(clippy::needless_update)] // `media` is cfg'd out in the default build
        let b = Anthropic::new("k", "claude-x").body(&CompletionRequest {
            system: "sys".into(),
            user: "no picture".into(),
            ..Default::default()
        });
        assert_eq!(b["messages"][0]["content"], "no picture");
    }
}