rsclaw-provider 0.1.0

//! Anthropic Messages API provider.
//!
//! Implements streaming via `anthropic-version: 2023-06-01` SSE.

use anyhow::{Context, Result};
use futures::{StreamExt, TryStreamExt, future::BoxFuture};
use reqwest::Client;
use serde_json::{Value, json};

use super::{
    ContentPart, LlmProvider, LlmRequest, LlmStream, Message, MessageContent, Role, StreamEvent,
    TokenUsage,
};

pub const ANTHROPIC_API_BASE: &str = "https://api.anthropic.com";
const ANTHROPIC_VERSION: &str = "2023-06-01";
const DEFAULT_MAX_TOKENS: u32 = 8192;

pub struct AnthropicProvider {
    client: Client,
    api_key: String,
    base_url: String,
    user_agent: Option<String>,
}

impl AnthropicProvider {
    pub fn new(api_key: impl Into<String>) -> Self {
        Self::with_base_url(api_key, ANTHROPIC_API_BASE)
    }

    pub fn with_base_url(api_key: impl Into<String>, base_url: impl Into<String>) -> Self {
        Self {
            client: super::http_client(),
            api_key: api_key.into(),
            base_url: base_url.into(),
            user_agent: None,
        }
    }

    /// Create a provider with custom User-Agent.
    pub fn with_user_agent(
        api_key: impl Into<String>,
        base_url: impl Into<String>,
        user_agent: Option<String>,
    ) -> Self {
        Self {
            client: super::http_client_with_ua(user_agent.as_deref()),
            api_key: api_key.into(),
            base_url: base_url.into(),
            user_agent,
        }
    }
}

impl LlmProvider for AnthropicProvider {
    fn name(&self) -> &str {
        "anthropic"
    }

    fn stream(&self, req: LlmRequest) -> BoxFuture<'_, Result<LlmStream>> {
        Box::pin(async move {
            super::warn_unsupported_kv_cache_mode_2(self.name(), &req);
            let body = build_request_body(&req)?;
            let url = format!("{}/v1/messages", self.base_url.trim_end_matches('/'));
            // Capture model + URL up-front so the failure path can
            // surface them in the error message — a bare "404 Not
            // Found" is otherwise impossible to triage.
            let model_for_log = req.model.clone();

            let resp = self
                .client
                .post(&url)
                .header("x-api-key", &self.api_key)
                .header("anthropic-version", ANTHROPIC_VERSION)
                .header("content-type", "application/json")
                .header(
                    "user-agent",
                    self.user_agent
                        .as_deref()
                        .unwrap_or(super::DEFAULT_USER_AGENT),
                )
                .json(&body)
                .timeout(std::time::Duration::from_secs(120))
                .send()
                .await
                .with_context(|| format!("Anthropic request failed (url={url})"))?;

            let status = resp.status();
            if !status.is_success() {
                let resp_body = resp.text().await.unwrap_or_default();
                let req_body_str = serde_json::to_string(&body).unwrap_or_default();
                let req_body_preview = if req_body_str.len() > 4000 {
                    format!(
                        "{}...[truncated, total {} bytes]",
                        rsclaw_util::truncate_str(&req_body_str, 4000),
                        req_body_str.len()
                    )
                } else {
                    req_body_str
                };
                tracing::warn!(
                    url = %url,
                    model = %model_for_log,
                    status = %status,
                    request_body = %req_body_preview,
                    response_body = %resp_body,
                    "Anthropic provider non-2xx response"
                );
                anyhow::bail!(
                    "Anthropic API error {status} at {url} (model={model_for_log}): {resp_body}"
                );
            }

            let byte_stream = resp.bytes_stream();
            let line_buffer = std::sync::Arc::new(tokio::sync::Mutex::new(String::new()));
            let event_stream = byte_stream
                .map_err(|e| anyhow::anyhow!("stream read error: {e}"))
                .then(move |chunk| {
                    let line_buffer = line_buffer.clone();
                    async move { parse_sse_chunk_buffered(chunk, &line_buffer).await }
                })
                .flat_map(|events| futures::stream::iter(events));

            let stream: LlmStream = Box::pin(event_stream);
            Ok(stream)
        })
    }
}

// ---------------------------------------------------------------------------
// Request body builder
// ---------------------------------------------------------------------------

fn build_request_body(req: &LlmRequest) -> Result<Value> {
    // Split system messages from conversation messages.
    let (system, messages) = split_system_messages(&req.messages, req.system.as_deref());

    let mut body = json!({
        "model":      req.model,
        "max_tokens": req.max_tokens.unwrap_or(DEFAULT_MAX_TOKENS),
        "stream":     true,
        "messages":   messages,
    });

    if let Some(sys) = system {
        body["system"] = json!(sys);
    }

    // Inject prompt caching markers (system_and_3 strategy).
    inject_cache_control(&mut body);

    if let Some(t) = req.temperature {
        body["temperature"] = super::json_f32(t);
    }

    if !req.tools.is_empty() {
        // Anthropic enforces the same `^[a-zA-Z0-9_-]{1,64}$` pattern
        // as OpenAI on tool names, so plugin tools like `wechat.send_text`
        // need the same wire encoding. Restore happens in the agent
        // runtime after accumulating streamed name fragments.
        let tools: Vec<Value> = req
            .tools
            .iter()
            .map(|t| {
                json!({
                    "name":         crate::openai::sanitize_tool_name(&t.name),
                    "description":  t.description,
                    "input_schema": t.parameters,
                })
            })
            .collect();
        body["tools"] = json!(tools);
        // Set tool_choice explicitly. The real Anthropic API defaults this to
        // {"type":"auto"} so it's a no-op for Claude — but anthropic-COMPATIBLE
        // third parties (notably Kimi's api.kimi.com/coding endpoint) ignore the
        // `tools` list entirely unless tool_choice is present, so the model never
        // function-calls and instead tells the user to run commands themselves.
        // The OpenAI-completions path encodes the same Kimi quirk (see
        // `openai::build_request_body`); the anthropic path needs it too. Being
        // explicit is safe for genuine Claude and fixes every compatible endpoint
        // with this behavior.
        body["tool_choice"] = json!({ "type": "auto" });
    }

    // Extended thinking: if budget > 0, enable thinking with the specified budget.
    if let Some(budget) = req.thinking_budget
        && budget > 0
    {
        body["thinking"] = json!({
            "type": "enabled",
            "budget_tokens": budget,
        });
    }

    Ok(body)
}

fn split_system_messages<'a>(
    messages: &'a [Message],
    extra_system: Option<&'a str>,
) -> (Option<String>, Vec<Value>) {
    let mut system_parts: Vec<String> =
        extra_system.map(|s| vec![s.to_owned()]).unwrap_or_default();

    let mut conv: Vec<Value> = Vec::new();

    for msg in messages {
        match msg.role {
            Role::System => {
                if let MessageContent::Text(t) = &msg.content {
                    system_parts.push(t.clone());
                }
            }
            Role::User | Role::Assistant | Role::Tool => {
                conv.push(serialize_message(msg));
            }
        }
    }

    let system = if system_parts.is_empty() {
        None
    } else {
        Some(system_parts.join("\n\n"))
    };

    (system, conv)
}

// TODO: Tool role maps to "user" but loses the tool_use_id, which Anthropic
// requires for tool_result blocks. This may cause issues with multi-turn
// tool-use conversations.
fn serialize_message(msg: &Message) -> Value {
    let role = match msg.role {
        Role::User | Role::Tool => "user",
        Role::Assistant => "assistant",
        Role::System => "user", // fallback, shouldn't happen
    };

    // Anthropic's API rejects messages with empty `content` ("the
    // message at position N with role 'user' must not be empty").
    // Empty content here is almost always a bug somewhere upstream
    // (compaction, tool-result truncation, an aborted turn that left
    // a husk in history). Rather than 400-out the entire turn, we
    // substitute a single-char placeholder so the conversation can
    // make forward progress; the upstream bug should still be fixed,
    // but a placeholder beats a hard failure on every retry.
    const EMPTY_PLACEHOLDER: &str = "(empty turn)";
    let content = match &msg.content {
        MessageContent::Text(t) => {
            if t.trim().is_empty() {
                tracing::warn!(role, "empty text-content message; substituting placeholder");
                json!(EMPTY_PLACEHOLDER)
            } else {
                json!(t)
            }
        }
        MessageContent::Parts(parts) => {
            let serialized: Vec<Value> = parts.iter().map(serialize_part).collect();
            // Reject entirely-empty parts arrays, and arrays where
            // every Text/Reasoning part is whitespace.
            let has_meaningful_content = !serialized.is_empty()
                && serialized.iter().any(|p| {
                    let t = p.get("type").and_then(|v| v.as_str()).unwrap_or("");
                    if t == "text" {
                        p.get("text")
                            .and_then(|v| v.as_str())
                            .map(|s| !s.trim().is_empty())
                            .unwrap_or(false)
                    } else {
                        // image / tool_use / tool_result are non-empty
                        // by construction — they always carry data.
                        true
                    }
                });
            if has_meaningful_content {
                json!(serialized)
            } else {
                tracing::warn!(
                    role,
                    parts_len = serialized.len(),
                    "all-empty parts array; substituting placeholder"
                );
                json!([{ "type": "text", "text": EMPTY_PLACEHOLDER }])
            }
        }
    };

    json!({ "role": role, "content": content })
}

fn serialize_part(part: &ContentPart) -> Value {
    match part {
        ContentPart::Text { text } => json!({ "type": "text", "text": text }),
        ContentPart::Image { url } => json!({
            "type": "image",
            "source": { "type": "url", "url": url }
        }),
        ContentPart::ToolUse { id, name, input } => json!({
            "type": "tool_use",
            "id":    id,
            // Echo the wire-encoded name so Anthropic accepts the
            // history block. The runtime stores names in their
            // original (restored) form, so re-sanitize on serialize.
            "name":  crate::openai::sanitize_tool_name(name),
            "input": input,
        }),
        ContentPart::ToolResult {
            tool_use_id,
            content,
            is_error,
        } => json!({
            "type":        "tool_result",
            "tool_use_id": tool_use_id,
            "content":     content,
            "is_error":    is_error.unwrap_or(false),
        }),
        ContentPart::Reasoning { text } => json!({
            "type": "text",
            "text": text,
        }),
    }
}

// ---------------------------------------------------------------------------
// Prompt caching — "system_and_3" strategy
// ---------------------------------------------------------------------------

/// Inject `cache_control: {"type": "ephemeral"}` markers into the request body.
///
/// Strategy (matches hermes-agent "system_and_3"):
/// - 1 breakpoint on the system prompt (stable prefix, high cache-hit rate)
/// - Up to 3 rolling breakpoints on the most recent non-system messages
///
/// The system prompt is converted from a plain string to a content-block array
/// so that the `cache_control` field can be attached to the last block.
fn inject_cache_control(body: &mut Value) {
    let cache_marker = json!({"type": "ephemeral"});

    // -- System prompt breakpoint --
    if let Some(system_val) = body.get_mut("system") {
        match system_val {
            // Plain string -> convert to content-block array with cache_control.
            Value::String(text) => {
                let block = json!([{
                    "type": "text",
                    "text": text.clone(),
                    "cache_control": cache_marker.clone(),
                }]);
                *system_val = block;
            }
            // Already an array of content blocks -> tag the last one.
            Value::Array(blocks) => {
                if let Some(last) = blocks.last_mut() {
                    last["cache_control"] = cache_marker.clone();
                }
            }
            _ => {}
        }
    }

    // -- Message breakpoints: 2+1 anchor strategy --
    //
    // Anchor 1 (stable): first user message — the task baseline.  Once set it
    // never changes even as the conversation grows or messages are trimmed from
    // the middle, so the prefix before this anchor is always a cache hit.
    //
    // Dynamic (volatile): last message only — the freshest addition.
    //
    // This replaces the old "last 3" approach which caused cache churn whenever
    // middle messages were pruned: the three-message window would shift and
    // invalidate the previously-cached prefix.
    if let Some(Value::Array(messages)) = body.get_mut("messages") {
        let len = messages.len();
        if len == 0 {
            return;
        }

        // Anchor: first user message.
        if messages[0].get("role").and_then(|r| r.as_str()) == Some("user") {
            tag_last_content_block(&mut messages[0], &cache_marker);
        }

        // Dynamic: latest message (only if it is different from the first).
        if len > 1 {
            let last_idx = len - 1;
            tag_last_content_block(&mut messages[last_idx], &cache_marker);
        }
    }
}

/// Add `cache_control` to the last content block of a message.
///
/// If the message content is a plain string, convert it to a content-block
/// array so the marker can be attached.
fn tag_last_content_block(msg: &mut Value, marker: &Value) {
    let content = match msg.get_mut("content") {
        Some(c) => c,
        None => return,
    };

    match content {
        // Plain string -> convert to [{type: "text", text: "...", cache_control: ...}]
        Value::String(text) => {
            let block = json!([{
                "type": "text",
                "text": text.clone(),
                "cache_control": marker.clone(),
            }]);
            *content = block;
        }
        // Array of content blocks -> tag the last one.
        Value::Array(blocks) => {
            if let Some(last) = blocks.last_mut() {
                last["cache_control"] = marker.clone();
            }
        }
        _ => {}
    }
}

// ---------------------------------------------------------------------------
// SSE parser
// ---------------------------------------------------------------------------

/// Buffered SSE parser — handles TCP chunk boundaries that split lines.
// TODO: SSE buffered parsing is duplicated across openai.rs, anthropic.rs,
// gemini.rs — extract shared utility
async fn parse_sse_chunk_buffered(
    chunk: Result<bytes::Bytes>,
    line_buffer: &tokio::sync::Mutex<String>,
) -> Vec<Result<StreamEvent>> {
    let bytes = match chunk {
        Ok(b) => b,
        Err(e) => return vec![Err(e)],
    };

    let text = match std::str::from_utf8(&bytes) {
        Ok(t) => std::borrow::Cow::Borrowed(t),
        Err(e) => {
            tracing::warn!(
                "anthropic: UTF-8 decode error at byte {}, replacing: {}",
                e.valid_up_to(),
                e
            );
            std::borrow::Cow::Owned(String::from_utf8_lossy(&bytes).into_owned())
        }
    };

    let mut buffer = line_buffer.lock().await;
    buffer.push_str(&text);

    let last_newline_pos = match buffer.rfind('\n') {
        Some(pos) => pos,
        None => return vec![],
    };

    let complete_portion = buffer[..last_newline_pos].to_owned();
    let incomplete_portion = buffer[last_newline_pos + 1..].to_owned();
    buffer.clear();
    buffer.push_str(&incomplete_portion);

    let mut events = Vec::new();
    for line in complete_portion.lines() {
        if let Some(data) = line
            .strip_prefix("data: ")
            .or_else(|| line.strip_prefix("data:"))
        {
            if data == "[DONE]" {
                continue;
            }
            if let Some(event) = parse_event(data) {
                events.push(Ok(event));
            }
        }
    }

    events
}

fn parse_event(data: &str) -> Option<StreamEvent> {
    let v: Value = serde_json::from_str(data).ok()?;
    let event_type = v["type"].as_str()?;

    match event_type {
        "content_block_delta" => {
            let delta_type = v["delta"]["type"].as_str()?;
            match delta_type {
                "text_delta" => {
                    let text = v["delta"]["text"].as_str()?.to_owned();
                    Some(StreamEvent::TextDelta(text))
                }
                "thinking_delta" => {
                    let text = v["delta"]["thinking"].as_str().unwrap_or("").to_owned();
                    if text.is_empty() {
                        None
                    } else {
                        Some(StreamEvent::ReasoningDelta(text))
                    }
                }
                "input_json_delta" => {
                    // Tool input streaming — emit as ToolCall so the agent loop
                    // accumulates partial JSON fragments (same pattern as OpenAI).
                    let partial = v["delta"]["partial_json"].as_str().unwrap_or("");
                    if partial.is_empty() {
                        None
                    } else {
                        Some(StreamEvent::ToolCall {
                            id: String::new(),
                            name: String::new(),
                            input: Value::String(partial.to_owned()),
                        })
                    }
                }
                _ => None,
            }
        }
        "content_block_start" => {
            let block = &v["content_block"];
            match block["type"].as_str() {
                Some("tool_use") => {
                    // Tool call start — emit immediately so the agent loop knows.
                    Some(StreamEvent::ToolCall {
                        id: block["id"].as_str().unwrap_or("").to_owned(),
                        name: block["name"].as_str().unwrap_or("").to_owned(),
                        input: serde_json::Value::Object(Default::default()),
                    })
                }
                Some("thinking") => {
                    // Thinking block start — no action needed.
                    None
                }
                _ => None,
            }
        }
        "message_delta" => {
            let usage = v["usage"].as_object().map(|u| TokenUsage {
                input: u.get("input_tokens").and_then(Value::as_u64).unwrap_or(0),
                output: u.get("output_tokens").and_then(Value::as_u64).unwrap_or(0),
                cache_creation: u
                    .get("cache_creation_input_tokens")
                    .and_then(Value::as_u64)
                    .unwrap_or(0),
                cache_read: u
                    .get("cache_read_input_tokens")
                    .and_then(Value::as_u64)
                    .unwrap_or(0),
                ..Default::default()
            });
            if v["delta"]["stop_reason"].is_string() {
                Some(StreamEvent::Done { usage })
            } else {
                None
            }
        }
        "error" => {
            let msg = v["error"]["message"]
                .as_str()
                .unwrap_or("unknown error")
                .to_owned();
            Some(StreamEvent::Error(msg))
        }
        _ => None,
    }
}

// ---------------------------------------------------------------------------
// Tests
// ---------------------------------------------------------------------------

#[cfg(test)]
mod tests {
    use super::{
        super::{LlmRequest, Message, MessageContent, Role},
        *,
    };

    fn make_request() -> LlmRequest {
        LlmRequest {
            fallback_models: Vec::new(),
            model: "claude-3-5-sonnet-20241022".to_owned(),
            ..Default::default()
        }
    }

    #[test]
    fn request_serializes_messages() {
        let req = LlmRequest {
            fallback_models: Vec::new(),
            messages: vec![
                Message {
                    role: Role::User,
                    content: MessageContent::Text("hi".to_owned()),
                    rsclaw_hidden: None,
                },
                Message {
                    role: Role::Assistant,
                    content: MessageContent::Text("hello".to_owned()),
                    rsclaw_hidden: None,
                },
            ],
            ..make_request()
        };
        let body = build_request_body(&req).expect("build request body");
        let msgs = body["messages"].as_array().expect("messages is array");
        assert_eq!(msgs.len(), 2);
        assert_eq!(msgs[0]["role"].as_str().expect("role is str"), "user");
        assert_eq!(msgs[1]["role"].as_str().expect("role is str"), "assistant");
    }

    #[test]
    fn system_field_present() {
        let req = LlmRequest {
            fallback_models: Vec::new(),
            system: Some("hello".to_owned()),
            ..make_request()
        };
        let body = build_request_body(&req).expect("build request body");
        // After cache_control injection, system is an array of content blocks.
        let blocks = body["system"]
            .as_array()
            .expect("system should be content-block array");
        assert_eq!(blocks.len(), 1);
        assert_eq!(blocks[0]["text"].as_str().expect("text is str"), "hello");
        assert_eq!(
            blocks[0]["cache_control"]["type"]
                .as_str()
                .expect("cache_control type is str"),
            "ephemeral"
        );
    }

    /// Regression: when tools are present, the body must carry an explicit
    /// `tool_choice: {"type":"auto"}`. Genuine Claude defaults to auto, but
    /// anthropic-compatible endpoints (Kimi's api.kimi.com/coding) ignore the
    /// tools list without it, so the model never function-calls.
    #[test]
    fn tool_choice_auto_set_when_tools_present() {
        let req = LlmRequest {
            fallback_models: Vec::new(),
            tools: vec![crate::ToolDef {
                name: "shell".to_owned(),
                description: "run a shell command".to_owned(),
                parameters: serde_json::json!({"type": "object"}),
            }],
            ..make_request()
        };
        let body = build_request_body(&req).expect("build request body");
        assert_eq!(
            body["tool_choice"]["type"].as_str(),
            Some("auto"),
            "tool_choice must be set to auto when tools are present"
        );
        assert!(body["tools"].as_array().is_some_and(|t| t.len() == 1));
    }

    /// No tools → no `tool_choice` key (avoid sending it with an empty tools
    /// list, which some endpoints reject).
    #[test]
    fn tool_choice_absent_when_no_tools() {
        let body = build_request_body(&make_request()).expect("build request body");
        assert!(
            body.get("tool_choice").is_none(),
            "tool_choice must be omitted when there are no tools"
        );
    }

    #[test]
    fn cache_control_system_and_anchors() {
        // 2+1 anchor strategy: system + first user message + last message.
        let req = LlmRequest {
            fallback_models: Vec::new(),
            system: Some("system prompt".to_owned()),
            messages: vec![
                Message {
                    role: Role::User,
                    content: MessageContent::Text("m1".to_owned()),
                    rsclaw_hidden: None,
                },
                Message {
                    role: Role::Assistant,
                    content: MessageContent::Text("m2".to_owned()),
                    rsclaw_hidden: None,
                },
                Message {
                    role: Role::User,
                    content: MessageContent::Text("m3".to_owned()),
                    rsclaw_hidden: None,
                },
                Message {
                    role: Role::Assistant,
                    content: MessageContent::Text("m4".to_owned()),
                    rsclaw_hidden: None,
                },
                Message {
                    role: Role::User,
                    content: MessageContent::Text("m5".to_owned()),
                    rsclaw_hidden: None,
                },
            ],
            ..make_request()
        };
        let body = build_request_body(&req).expect("build request body");

        // System should have cache_control.
        let sys_blocks = body["system"].as_array().expect("system content blocks");
        assert_eq!(
            sys_blocks[0]["cache_control"]["type"]
                .as_str()
                .expect("cache_control type"),
            "ephemeral"
        );

        let msgs = body["messages"].as_array().expect("messages is array");
        assert_eq!(msgs.len(), 5);

        // m1 (first user): cache_control — stable task anchor.
        let m1_content = &msgs[0]["content"];
        assert!(
            m1_content.is_array(),
            "m1 should be converted to content-block array"
        );
        assert_eq!(
            m1_content[0]["cache_control"]["type"]
                .as_str()
                .expect("m1 cache_control type"),
            "ephemeral"
        );

        // m2, m3, m4: no cache_control (middle messages are not anchored).
        for i in 1..4 {
            let content = &msgs[i]["content"];
            if content.is_array() {
                assert!(
                    content[0].get("cache_control").is_none(),
                    "message {i} should not have cache_control"
                );
            }
            // plain string content also means no cache_control — acceptable
        }

        // m5 (last): cache_control — dynamic breakpoint.
        let m5_content = &msgs[4]["content"];
        assert!(m5_content.is_array(), "m5 should be content-block array");
        assert_eq!(
            m5_content[0]["cache_control"]["type"]
                .as_str()
                .expect("m5 cache_control type"),
            "ephemeral"
        );
    }

    #[test]
    fn cache_control_fewer_than_3_messages() {
        let req = LlmRequest {
            fallback_models: Vec::new(),
            messages: vec![Message {
                role: Role::User,
                content: MessageContent::Text("only one".to_owned()),
                rsclaw_hidden: None,
            }],
            ..make_request()
        };
        let body = build_request_body(&req).expect("build request body");
        let msgs = body["messages"].as_array().expect("messages is array");
        // Single message should still get cache_control.
        let content = &msgs[0]["content"];
        assert!(content.is_array());
        assert_eq!(
            content[0]["cache_control"]["type"]
                .as_str()
                .expect("cache_control type"),
            "ephemeral"
        );
    }

    #[test]
    fn temperature_serializes() {
        let req = LlmRequest {
            fallback_models: Vec::new(),
            temperature: Some(0.7),
            ..make_request()
        };
        let body = build_request_body(&req).expect("build request body");
        let t = body["temperature"].as_f64().expect("temperature is f64");
        assert!((t - 0.7).abs() < 1e-4);
    }
}