cortex_llm/

claude_http.rs

//! HTTP adapter that posts to the Anthropic Messages API.
//!
//! [`ClaudeHttpAdapter`] implements [`LlmAdapter`] by forwarding requests to
//! `https://api.anthropic.com/v1/messages`. Because `ureq` is synchronous,
//! the blocking I/O is wrapped with `tokio::task::spawn_blocking` so the
//! adapter can satisfy the async trait contract without blocking the async
//! executor.
//!
//! ## Streaming
//!
//! [`ClaudeHttpAdapter::stream_boxed`] overrides the trait default and posts
//! to `/v1/messages` with `"stream": true`. Anthropic returns a Server-Sent
//! Events stream. The blocking reader filters `content_block_delta` events to
//! collect token deltas and emits a terminal [`StreamChunk`] on `message_stop`.
//!
//! ## Runtime ceiling
//!
//! Per the forthcoming ADR 0048, this adapter carries a `RemoteUnsigned`
//! runtime ceiling — lower than `OllamaHttpAdapter`'s `LocalUnsigned`
//! ceiling, because responses arrive from a remote endpoint without
//! supply-chain or cryptographic non-repudiation. The `RemoteUnsigned`
//! variant is being added in a separate task; until it lands, callers MUST
//! treat responses from this adapter as bounded at `LocalUnsigned` or weaker.
//!
//! ## Data classification
//!
//! Before dispatching a prompt to a remote endpoint, callers should invoke
//! [`check_prompt_sensitivity`] to verify that no high-sensitivity memory
//! content is included. The real classification logic (ADR 0030) will be
//! wired in a follow-on task; the current stub always permits.
//!
//! ## Construction
//!
//! Construction reads `CORTEX_CLAUDE_API_KEY` from the environment and fails
//! closed if the variable is absent or empty. This prevents accidental use in
//! environments where the key has not been provisioned.

use std::time::Duration;

use async_trait::async_trait;
use serde::{Deserialize, Serialize};

use crate::adapter::{
    blake3_hex, BoxStream, LlmAdapter, LlmError, LlmRequest, LlmResponse, LlmRole, StreamChunk,
};
use crate::sensitivity::{check_remote_prompt_sensitivity, MaxSensitivity};

/// Stable invariant: `CORTEX_CLAUDE_API_KEY` env var absent or empty (ADR 0048 §4).
pub const CLAUDE_ADAPTER_API_KEY_MISSING_INVARIANT: &str = "cortex.run.claude.api_key_missing";
/// Stable invariant: model ID not in compile-time allowlist (ADR 0048 §5).
pub const CLAUDE_ADAPTER_MODEL_NOT_ALLOWED_INVARIANT: &str = "cortex.run.claude.model_not_allowed";
/// Stable invariant: endpoint rejected — not `api.anthropic.com` (ADR 0048 §2).
pub const CLAUDE_ADAPTER_ENDPOINT_REJECTED_INVARIANT: &str = "cortex.run.claude.endpoint_rejected";

/// HTTP adapter that routes to the Anthropic Messages API.
///
/// Construction reads `CORTEX_CLAUDE_API_KEY` from the environment and
/// validates the model string. The adapter is `Send + Sync` and may be placed
/// behind an `Arc<dyn LlmAdapter>`.
///
/// The `max_sensitivity` field enforces the ADR 0048 §3 data-classification
/// gate: prompts containing inline high-sensitivity markers are rejected before
/// any bytes leave the machine when the gate is set below `High`.
#[derive(Debug, Clone)]
pub struct ClaudeHttpAdapter {
    /// API key loaded from [`ClaudeHttpAdapter::ANTHROPIC_API_KEY_ENV`] at
    /// construction time.
    api_key: String,
    /// Anthropic model identifier, e.g. `claude-3-5-sonnet-20241022`.
    model: String,
    /// Base URL for the API; defaults to [`ClaudeHttpAdapter::ANTHROPIC_API_BASE`].
    /// Overridable via [`ClaudeHttpAdapter::new_with_base_url`] for testing.
    base_url: String,
    /// Maximum data-classification level permitted in remote prompts (ADR 0048 §3).
    /// Defaults to [`MaxSensitivity::Medium`] when constructed via [`Self::new`].
    max_sensitivity: MaxSensitivity,
}

impl ClaudeHttpAdapter {
    /// The base URL for all Anthropic API requests.
    pub const ANTHROPIC_API_BASE: &'static str = "https://api.anthropic.com";

    /// Environment variable that must contain the Anthropic API key.
    ///
    /// Construction fails with [`LlmError::InvalidRequest`] if this variable
    /// is absent or empty.
    pub const ANTHROPIC_API_KEY_ENV: &'static str = "CORTEX_CLAUDE_API_KEY";

    /// `anthropic-version` header value required by the Messages API.
    pub const ANTHROPIC_VERSION_HEADER: &'static str = "2023-06-01";

    /// Construct a `ClaudeHttpAdapter` for `model` with `max_sensitivity`.
    ///
    /// `max_sensitivity` controls the data-classification gate (ADR 0048 §3).
    /// Pass `None` to use the default of [`MaxSensitivity::Medium`], which
    /// blocks high-sensitivity memories from being sent to the remote endpoint.
    ///
    /// Returns [`LlmError::InvalidRequest`] when:
    /// - `CORTEX_CLAUDE_API_KEY` is absent or empty.
    /// - `model` is empty.
    /// - `model` contains the string `"latest"` (forbidden to preserve
    ///   audit-trail identity; see ADR 0044 §3 and ADR 0048).
    pub fn new(model: String, max_sensitivity: Option<MaxSensitivity>) -> Result<Self, LlmError> {
        let api_key = std::env::var(Self::ANTHROPIC_API_KEY_ENV)
            .ok()
            .filter(|v| !v.is_empty())
            .ok_or_else(|| {
                LlmError::InvalidRequest(format!(
                    "env var {} is absent or empty; refusing to construct ClaudeHttpAdapter",
                    Self::ANTHROPIC_API_KEY_ENV
                ))
            })?;

        if model.is_empty() {
            return Err(LlmError::InvalidRequest(
                "model must not be empty".to_string(),
            ));
        }
        if model.contains("latest") {
            return Err(LlmError::InvalidRequest(format!(
                "model '{model}' contains 'latest' alias; pin to a specific version for audit-trail identity"
            )));
        }

        Ok(Self {
            api_key,
            model,
            base_url: Self::ANTHROPIC_API_BASE.to_string(),
            max_sensitivity: max_sensitivity.unwrap_or(MaxSensitivity::Medium),
        })
    }

    /// Construct a `ClaudeHttpAdapter` with an explicit `base_url`.
    ///
    /// This constructor is intended for testing only — it allows tests to point
    /// the adapter at a mock `TcpListener` instead of `api.anthropic.com`. The
    /// API key validation and model validation rules are identical to
    /// [`Self::new`]. `max_sensitivity` follows the same defaulting rule:
    /// `None` resolves to [`MaxSensitivity::Medium`].
    #[doc(hidden)]
    pub fn new_with_base_url(
        model: String,
        base_url: String,
        max_sensitivity: Option<MaxSensitivity>,
    ) -> Result<Self, LlmError> {
        let api_key = std::env::var(Self::ANTHROPIC_API_KEY_ENV)
            .ok()
            .filter(|v| !v.is_empty())
            .ok_or_else(|| {
                LlmError::InvalidRequest(format!(
                    "env var {} is absent or empty; refusing to construct ClaudeHttpAdapter",
                    Self::ANTHROPIC_API_KEY_ENV
                ))
            })?;

        if model.is_empty() {
            return Err(LlmError::InvalidRequest(
                "model must not be empty".to_string(),
            ));
        }
        if model.contains("latest") {
            return Err(LlmError::InvalidRequest(format!(
                "model '{model}' contains 'latest' alias; pin to a specific version for audit-trail identity"
            )));
        }

        Ok(Self {
            api_key,
            model,
            base_url,
            max_sensitivity: max_sensitivity.unwrap_or(MaxSensitivity::Medium),
        })
    }
}

// ---------------------------------------------------------------------------
// Wire types
// ---------------------------------------------------------------------------

/// Outgoing body for `POST /v1/messages`.
#[derive(Debug, Serialize)]
struct MessagesRequest<'a> {
    model: &'a str,
    max_tokens: u32,
    messages: Vec<AnthropicMessage<'a>>,
    stream: bool,
}

// ---------------------------------------------------------------------------
// Streaming SSE wire types
// ---------------------------------------------------------------------------

/// Top-level envelope for a `content_block_delta` SSE event.
#[derive(Debug, Deserialize)]
struct SseEvent {
    #[serde(rename = "type")]
    event_type: String,
    #[serde(default)]
    delta: Option<SseDelta>,
}

/// The `delta` field inside a `content_block_delta` event.
#[derive(Debug, Deserialize)]
struct SseDelta {
    #[serde(rename = "type")]
    delta_type: String,
    #[serde(default)]
    text: String,
}

/// One message in the Anthropic chat format.
#[derive(Debug, Serialize)]
struct AnthropicMessage<'a> {
    role: &'a str,
    content: &'a str,
}

/// Top-level Anthropic Messages API response envelope.
#[derive(Debug, Deserialize)]
struct MessagesResponse {
    #[serde(default)]
    content: Vec<ContentBlock>,
    #[serde(default)]
    model: String,
    #[serde(default)]
    usage: Option<AnthropicUsage>,
}

/// One content block in the response `content` array.
#[derive(Debug, Deserialize)]
struct ContentBlock {
    #[serde(rename = "type")]
    block_type: String,
    #[serde(default)]
    text: String,
}

/// Token-usage field from the Anthropic response.
#[derive(Debug, Deserialize)]
struct AnthropicUsage {
    input_tokens: u32,
    output_tokens: u32,
}

// ---------------------------------------------------------------------------
// LlmAdapter implementation
// ---------------------------------------------------------------------------

#[async_trait]
impl LlmAdapter for ClaudeHttpAdapter {
    fn adapter_id(&self) -> &'static str {
        "claude"
    }

    async fn complete(&self, req: LlmRequest) -> Result<LlmResponse, LlmError> {
        // ADR 0048 §3: data-classification gate before any remote dispatch.
        // Assemble the full prompt text (system + all message contents) and run
        // the sensitivity gate so that high-sensitivity markers are caught
        // before any bytes leave the machine.
        let prompt_text: String = std::iter::once(req.system.as_str())
            .chain(req.messages.iter().map(|m| m.content.as_str()))
            .collect::<Vec<_>>()
            .join("\n");
        check_remote_prompt_sensitivity(&prompt_text, self.max_sensitivity)?;

        let api_key = self.api_key.clone();
        let model = self.model.clone();
        let base_url = self.base_url.clone();
        let timeout_ms = req.timeout_ms;

        let result = tokio::task::spawn_blocking(move || {
            call_claude(&api_key, &model, &base_url, &req, timeout_ms)
        })
        .await
        .map_err(|e| LlmError::Transport(format!("spawn_blocking join error: {e}")))?;

        result
    }

    /// Override with true Anthropic SSE streaming via `POST /v1/messages` with
    /// `"stream": true`.
    ///
    /// Uses `ureq` (synchronous) inside `spawn_blocking`. The blocking reader
    /// collects all SSE lines into a `Vec` before yielding them; the
    /// `async_stream::stream!` block then emits items one by one.
    ///
    /// TODO: replace `ureq` with an async HTTP client to achieve true
    /// line-by-line streaming without buffering the entire response.
    fn stream_boxed(&self, req: LlmRequest) -> BoxStream<'_> {
        stream_claude_sse(
            self.api_key.clone(),
            self.model.clone(),
            self.base_url.clone(),
            req,
        )
    }
}

/// Synchronous Anthropic HTTP call, executed inside `spawn_blocking`.
fn call_claude(
    api_key: &str,
    model: &str,
    base_url: &str,
    req: &LlmRequest,
    timeout_ms: u64,
) -> Result<LlmResponse, LlmError> {
    let url = format!("{base_url}/v1/messages");

    // Build message list from request messages; Anthropic only accepts
    // `user` and `assistant` roles in the `messages` array.
    let messages: Vec<AnthropicMessage<'_>> = req
        .messages
        .iter()
        .map(|m| AnthropicMessage {
            role: m.role.as_anthropic_str(),
            content: &m.content,
        })
        .collect();

    let body = MessagesRequest {
        model,
        max_tokens: req.max_tokens,
        messages,
        stream: false,
    };

    let body_value = serde_json::to_value(&body)
        .map_err(|e| LlmError::Transport(format!("request serialization failed: {e}")))?;

    let timeout = Duration::from_millis(timeout_ms);
    let agent = ureq::AgentBuilder::new().timeout(timeout).build();

    let raw_response = agent
        .post(&url)
        .set("x-api-key", api_key)
        .set(
            "anthropic-version",
            ClaudeHttpAdapter::ANTHROPIC_VERSION_HEADER,
        )
        .set("content-type", "application/json")
        .send_json(body_value)
        .map_err(|err| map_ureq_error(err, timeout_ms))?;

    let status = raw_response.status();
    if status != 200 {
        return Err(LlmError::Upstream(format!("HTTP {status}")));
    }

    let response_text = raw_response
        .into_string()
        .map_err(|e| LlmError::Transport(format!("reading response body: {e}")))?;

    let parsed: MessagesResponse = serde_json::from_str(&response_text)
        .map_err(|e| LlmError::Parse(format!("anthropic response parse: {e}")))?;

    // Extract the first text block from content[].
    let text = parsed
        .content
        .into_iter()
        .find(|block| block.block_type == "text")
        .map(|block| block.text)
        .ok_or_else(|| {
            LlmError::Parse("anthropic response contained no text content block".to_string())
        })?;

    let raw_hash = blake3_hex(response_text.as_bytes());
    let usage = parsed.usage.map(|u| crate::adapter::TokenUsage {
        prompt_tokens: u.input_tokens,
        completion_tokens: u.output_tokens,
    });

    // Use the model echoed by the provider when present; fall back to the
    // adapter's configured model so the field is never empty.
    let response_model = if parsed.model.is_empty() {
        model.to_string()
    } else {
        parsed.model
    };

    Ok(LlmResponse {
        text,
        parsed_json: None,
        model: response_model,
        usage,
        raw_hash,
    })
}

// ---------------------------------------------------------------------------
// Streaming implementation
// ---------------------------------------------------------------------------

/// Build a `BoxStream` that drives Anthropic SSE streaming.
///
/// Extracted as a free function so the `async_stream::stream!` macro is not
/// nested inside an `impl` block, which can confuse lifetime inference.
fn stream_claude_sse(
    api_key: String,
    model: String,
    base_url: String,
    req: LlmRequest,
) -> BoxStream<'static> {
    Box::pin(async_stream::stream! {
        let timeout_ms = req.timeout_ms;
        let result = tokio::task::spawn_blocking(move || {
            call_claude_streaming(&api_key, &model, &base_url, &req, timeout_ms)
        })
        .await;

        match result {
            Ok(chunks) => {
                for chunk in chunks {
                    yield chunk;
                }
            }
            Err(e) => yield Err(LlmError::Transport(format!("spawn_blocking join error: {e}"))),
        }
    })
}

/// Synchronous Anthropic SSE streaming call, executed inside `spawn_blocking`.
///
/// Posts to `/v1/messages` with `stream: true`, then reads the response body
/// line by line. SSE protocol:
/// - Empty lines are separators — skip them.
/// - Lines beginning with `event:` are event-type hints — skip them (we parse
///   the type from the `data:` JSON instead).
/// - Lines beginning with `data:` carry the JSON payload.
///
/// For `content_block_delta` events with `delta.type == "text_delta"` we emit
/// a [`StreamChunk`] carrying the token text. On `message_stop` we emit a
/// terminal chunk with `finish_reason = Some("stop")` and return.
fn call_claude_streaming(
    api_key: &str,
    model: &str,
    base_url: &str,
    req: &LlmRequest,
    timeout_ms: u64,
) -> Vec<Result<StreamChunk, LlmError>> {
    let url = format!("{base_url}/v1/messages");

    let messages: Vec<AnthropicMessage<'_>> = req
        .messages
        .iter()
        .map(|m| AnthropicMessage {
            role: m.role.as_anthropic_str(),
            content: &m.content,
        })
        .collect();

    let body = MessagesRequest {
        model,
        max_tokens: req.max_tokens,
        messages,
        stream: true,
    };

    let body_value = match serde_json::to_value(&body) {
        Ok(v) => v,
        Err(e) => {
            return vec![Err(LlmError::Transport(format!(
                "request serialization failed: {e}"
            )))]
        }
    };

    let timeout = Duration::from_millis(timeout_ms);
    let agent = ureq::AgentBuilder::new().timeout(timeout).build();

    let raw_response = match agent
        .post(&url)
        .set("x-api-key", api_key)
        .set(
            "anthropic-version",
            ClaudeHttpAdapter::ANTHROPIC_VERSION_HEADER,
        )
        .set("content-type", "application/json")
        .send_json(body_value)
    {
        Ok(r) => r,
        Err(err) => return vec![Err(map_ureq_error(err, timeout_ms))],
    };

    let status = raw_response.status();
    if status != 200 {
        return vec![Err(LlmError::Upstream(format!("HTTP {status}")))];
    }

    let body_text = match raw_response.into_string() {
        Ok(s) => s,
        Err(e) => {
            return vec![Err(LlmError::Transport(format!(
                "reading streaming response body: {e}"
            )))]
        }
    };

    let mut chunks = Vec::new();

    for line in body_text.lines() {
        // Skip empty lines (SSE event separators) and event-type hint lines.
        if line.is_empty() || line.starts_with("event:") {
            continue;
        }

        let data = match line.strip_prefix("data:") {
            Some(rest) => rest.trim(),
            None => continue,
        };

        let event: SseEvent = match serde_json::from_str(data) {
            Ok(v) => v,
            Err(e) => {
                chunks.push(Err(LlmError::Parse(format!(
                    "claude SSE data parse: {e}: {data}"
                ))));
                continue;
            }
        };

        match event.event_type.as_str() {
            "content_block_delta" => {
                if let Some(delta) = event.delta {
                    if delta.delta_type == "text_delta" {
                        chunks.push(Ok(StreamChunk {
                            delta: delta.text,
                            finish_reason: None,
                        }));
                    }
                }
            }
            "message_stop" => {
                chunks.push(Ok(StreamChunk {
                    delta: String::new(),
                    finish_reason: Some("stop".into()),
                }));
                // Terminal event — no further lines need processing.
                return chunks;
            }
            _ => {
                // Informational events (message_start, content_block_start,
                // message_delta, ping, etc.) are intentionally ignored.
            }
        }
    }

    chunks
}

/// Map a `ureq` error to an [`LlmError`] variant.
fn map_ureq_error(err: ureq::Error, timeout_ms: u64) -> LlmError {
    match err {
        ureq::Error::Transport(t) => {
            let msg = t.to_string();
            if is_timeout_message(&msg) {
                LlmError::Timeout { timeout_ms }
            } else {
                LlmError::Transport(msg)
            }
        }
        ureq::Error::Status(code, _) => LlmError::Upstream(format!("HTTP {code}")),
    }
}

/// Heuristic: does the transport error message look like a timeout?
fn is_timeout_message(msg: &str) -> bool {
    let lower = msg.to_ascii_lowercase();
    lower.contains("timed out") || lower.contains("deadline exceeded") || lower.contains("timeout")
}

// ---------------------------------------------------------------------------
// Role serialization helper
// ---------------------------------------------------------------------------

impl LlmRole {
    /// Return the lowercase string representation used by Anthropic's API.
    ///
    /// Anthropic accepts `user` and `assistant`; `tool` is mapped to `user`
    /// as a conservative fallback (tool-result multi-turn is out of scope
    /// for this adapter version).
    fn as_anthropic_str(self) -> &'static str {
        match self {
            LlmRole::User | LlmRole::Tool => "user",
            LlmRole::Assistant => "assistant",
        }
    }
}