codecraft/core/src/provider/anthropic.rs

//! Anthropic Messages API (streaming). Also works with Anthropic-compatible
//! third-party endpoints; first-party-only features are gated on the host.

use super::{ChatRequest, Endpoint, Provider, StreamEvent, http_error};
use crate::config::ModelConfig;
use crate::types::{Message, Part, Role, Usage};
use anyhow::{Result, anyhow};
use async_trait::async_trait;
use eventsource_stream::Eventsource;
use futures::StreamExt;
use serde_json::{Value, json};
use std::collections::BTreeMap;
use tokio::sync::mpsc;
use tokio_util::sync::CancellationToken;

const API_VERSION: &str = "2026-10-21";

pub struct Anthropic {
    client: reqwest::Client,
    ep: Endpoint,
}

impl Anthropic {
    pub fn new(client: reqwest::Client, ep: Endpoint) -> Self {
        Self { client, ep }
    }

    fn first_party(&self) -> bool {
        self.ep.base_url.contains("api.anthropic.com")
    }

    fn request(&self, method: reqwest::Method, path: &str) -> reqwest::RequestBuilder {
        let mut rb = self.client.request(method, format!("{}{}", self.ep.base_url, path));
        if !self.ep.headers.keys().any(|k| k.eq_ignore_ascii_case("anthropic-version")) {
            rb = rb.header("anthropic-version", API_VERSION);
        }
        self.ep.apply(rb)
    }
}

/// Convert our neutral history into Anthropic message params.
pub fn to_wire_messages(messages: &[Message]) -> Vec<Value> {
    let mut out: Vec<Value> = Vec::new();
    for m in messages {
        let mut content = Vec::new();
        for p in &m.parts {
            match p {
                Part::Text { text } if !text.is_empty() => {
                    content.push(json!({"type": "text", "text": text}))
                }
                Part::Summary { text } => content.push(json!({
                    "type": "text",
                    "text": format!("<conversation-summary>\n{text}\n</conversation-summary>")
                })),
                Part::Image { image } => content.push(json!({
                    "type": "image",
                    "source": {"type": "base64", "media_type": image.media_type, "data": image.data}
                })),
                // Thinking blocks are replayed unchanged (assistant turns only).
                Part::Reasoning { text, signature } if m.role == Role::Assistant => {
                    match signature.as_deref() {
                        Some(sig) if sig.starts_with("redacted:") => content.push(json!({
                            "type": "redacted_thinking",
                            "data": &sig["redacted:".len()..]
                        })),
                        Some(sig) => content.push(json!({
                            "type": "thinking", "thinking": text, "signature": sig
                        })),
                        // Reasoning from another provider has no signature – drop it.
                        None => {}
                    }
                }
                Part::ToolCall { id, name, input, .. } => content.push(json!({
                    "type": "tool_use", "id": id, "name": name, "input": input
                })),
                Part::ToolResult { call_id, content: text, images, is_error, .. } => {
                    let mut blocks = vec![json!({"type": "text", "text": if text.is_empty() { "(no output)" } else { text }})];
                    for img in images {
                        blocks.push(json!({
                            "type": "image",
                            "source": {"type": "base64", "media_type": img.media_type, "data": img.data}
                        }));
                    }
                    content.push(json!({
                        "type": "tool_result", "tool_use_id": call_id,
                        "content": blocks, "is_error": is_error
                    }))
                }
                _ => {}
            }
        }
        if content.is_empty() {
            continue;
        }
        let role = match m.role {
            Role::User => "user",
            Role::Assistant => "assistant",
        };
        // Merge consecutive messages of the same role (the API requires alternation).
        if let Some(last) = out.last_mut() {
            if last["role"] == role {
                last["content"].as_array_mut().unwrap().extend(content);
                continue;
            }
        }
        out.push(json!({"role": role, "content": content}));
    }
    // tool_result blocks must come first in a user message.
    for msg in out.iter_mut().filter(|m| m["role"] == "user") {
        let arr = msg["content"].as_array_mut().unwrap();
        arr.sort_by_key(|b| if b["type"] == "tool_result" { 0 } else { 1 });
    }
    out
}

#[derive(Default)]
struct BlockState {
    kind: String,
    id: String,
    name: String,
    json: String,
    signature: String,
}

#[async_trait]
impl Provider for Anthropic {
    async fn stream(
        &self,
        req: ChatRequest,
        tx: mpsc::Sender<StreamEvent>,
        cancel: CancellationToken,
    ) -> Result<()> {
        let first_party = self.first_party();
        let tools: Vec<Value> = req
            .tools
            .iter()
            .map(|t| {
                let mut v = json!({
                    "name": t.name, "description": t.description, "input_schema": t.input_schema
                });
                if first_party {
                    v["eager_input_streaming"] = json!(true);
                }
                v
            })
            .collect();
        let mut body = json!({
            "model": req.model.id,
            "max_tokens": req.max_tokens,
            "stream": true,
            "system": [{"type": "text", "text": req.system, "cache_control": {"type": "ephemeral"}}],
            "messages": to_wire_messages(&req.messages),
        });
        if !tools.is_empty() {
            body["tools"] = Value::Array(tools);
        }
        if first_party {
            // Automatic breakpoint on the last cacheable block (conversation prefix).
            body["cache_control"] = json!({"type": "ephemeral"});
        }
        if req.reasoning && req.model.reasoning {
            body["thinking"] = json!({"type": "adaptive", "display": "summarized"});
        }
        if let Some(effort) = &req.model.effort {
            body["output_config"] = json!({"effort": effort});
        }

        for (k, v) in &req.model.options {
            body[k] = v.clone();
        }
        let rb = self.request(reqwest::Method::POST, "/messages").json(&body);
        let resp = self.ep.with_session(rb, req.session_id.as_deref()).send().await?;
        if !resp.status().is_success() {
            return Err(http_error(resp).await);
        }

        let mut events = resp.bytes_stream().eventsource();
        let mut blocks: BTreeMap<u64, BlockState> = BTreeMap::new();
        let mut usage = Usage::default();
        let mut stop_reason = String::from("end_turn");

        loop {
            let ev = tokio::select! {
                _ = cancel.cancelled() => return Ok(()),
                ev = events.next() => ev,
            };
            let Some(ev) = ev else { break };
            let ev = ev.map_err(|e| anyhow!("stream error: {e}"))?;
            if ev.data.is_empty() {
                continue;
            }
            let data: Value = serde_json::from_str(&ev.data)?;
            match data["type"].as_str().unwrap_or_default() {
                "message_start" => {
                    let u = &data["message"]["usage"];
                    usage.input_tokens = u["input_tokens"].as_u64().unwrap_or(0);
                    usage.cache_read_tokens = u["cache_read_input_tokens"].as_u64().unwrap_or(0);
                    usage.cache_write_tokens = u["cache_creation_input_tokens"].as_u64().unwrap_or(0);
                }
                "content_block_start" => {
                    let idx = data["index"].as_u64().unwrap_or(0);
                    let cb = &data["content_block"];
                    let mut st = BlockState {
                        kind: cb["type"].as_str().unwrap_or_default().to_string(),
                        ..Default::default()
                    };
                    match st.kind.as_str() {
                        "tool_use" => {
                            st.id = cb["id"].as_str().unwrap_or_default().into();
                            st.name = cb["name"].as_str().unwrap_or_default().into();
                        }
                        "redacted_thinking" => {
                            st.signature = format!("redacted:{}", cb["data"].as_str().unwrap_or_default());
                        }
                        "text" => {
                            if let Some(t) = cb["text"].as_str().filter(|t| !t.is_empty()) {
                                tx.send(StreamEvent::TextDelta(t.into())).await.ok();
                            }
                        }
                        _ => {}
                    }
                    blocks.insert(idx, st);
                }
                "content_block_delta" => {
                    let idx = data["index"].as_u64().unwrap_or(0);
                    let d = &data["delta"];
                    let st = blocks.entry(idx).or_default();
                    match d["type"].as_str().unwrap_or_default() {
                        "text_delta" => {
                            tx.send(StreamEvent::TextDelta(d["text"].as_str().unwrap_or_default().into())).await.ok();
                        }
                        "thinking_delta" => {
                            tx.send(StreamEvent::ReasoningDelta(d["thinking"].as_str().unwrap_or_default().into())).await.ok();
                        }
                        "signature_delta" => st.signature.push_str(d["signature"].as_str().unwrap_or_default()),
                        "input_json_delta" => st.json.push_str(d["partial_json"].as_str().unwrap_or_default()),
                        _ => {}
                    }
                }
                "content_block_stop" => {
                    let idx = data["index"].as_u64().unwrap_or(0);
                    if let Some(st) = blocks.remove(&idx) {
                        match st.kind.as_str() {
                            "tool_use" => {
                                let input = if st.json.trim().is_empty() {
                                    json!({})
                                } else {
                                    // Invalid JSON (e.g. truncated) is passed on as a marker; the
                                    // agent loop answers it with an error tool_result.
                                    serde_json::from_str(&st.json)
                                        .unwrap_or_else(|_| json!({"__invalid_json": st.json}))
                                };
                                tx.send(StreamEvent::ToolCall { id: st.id, name: st.name, input, extra: None }).await.ok();
                            }
                            "thinking" | "redacted_thinking" => {
                                tx.send(StreamEvent::ReasoningSignature(st.signature)).await.ok();
                            }
                            _ => {}
                        }
                    }
                }
                "message_delta" => {
                    if let Some(r) = data["delta"]["stop_reason"].as_str() {
                        stop_reason = r.to_string();
                    }
                    if let Some(o) = data["usage"]["output_tokens"].as_u64() {
                        usage.output_tokens = o;
                    }
                    usage.read_reported_cost(&data["usage"]);
                }
                "error" => {
                    let msg = data["error"]["message"].as_str().unwrap_or("unknown error");
                    return Err(anyhow!("provider error: {msg}"));
                }
                _ => {}
            }
        }
        tx.send(StreamEvent::Usage(usage)).await.ok();
        tx.send(StreamEvent::Done { stop_reason }).await.ok();
        Ok(())
    }

    async fn list_models(&self) -> Result<Vec<ModelConfig>> {
        let resp = self.request(reqwest::Method::GET, "/models?limit=100").send().await?;
        if !resp.status().is_success() {
            return Err(http_error(resp).await);
        }
        let v: Value = resp.json().await?;
        Ok(v["data"]
            .as_array()
            .map(|a| {
                a.iter()
                    .map(|m| ModelConfig {
                        id: m["id"].as_str().unwrap_or_default().into(),
                        name: m["display_name"].as_str().map(String::from),
                        context: m["max_input_tokens"].as_u64(),
                        max_output: m["max_tokens"].as_u64(),
                        reasoning: true,
                        vision: true,
                        ..Default::default()
                    })
                    .collect()
            })
            .unwrap_or_default())
    }
}