kite/backend/ai/anthropic_provider.py

"""Anthropic AI provider — singleton client, output_config structured output, smart truncation.

D-03: Uses output_config={"format": {"type": "json_schema", "schema": ...}} with constrained
      decoding available in anthropic SDK >=0.95.0 (GA, no beta headers needed).
D-07: self._client = AsyncAnthropic(...) created once in __init__ and reused — never recreated
      per API call to preserve the httpx connection pool.
D-12/D-13: Global char constant removed; uses self._context_chars with 60/40 smart truncation.

Security: api_key is accepted from the caller (loaded from system_settings by ai_config.py
and decrypted before being passed here). The key is never stored beyond this instance's
lifetime. T-07-06 mitigated: this class never reads the api_key from env vars directly.
"""
from __future__ import annotations

import anthropic
from ai.base import AIProvider, ClassificationResult
from ai.utils import parse_classification, parse_suggestions

# ── Output schemas for constrained decoding (D-03 / RESEARCH.md) ────────────────────────────
# additionalProperties=False required by Anthropic output_config grammar.
# "reasoning" is intentionally absent from "required" so legacy prompts that don't include it
# still produce valid JSON (Anthropic will emit it because it is declared in properties, but
# we do not enforce it in the schema to avoid refusal on minimal responses).

_CLASSIFICATION_SCHEMA: dict = {
    "type": "object",
    "properties": {
        "assigned_topics": {"type": "array", "items": {"type": "string"}},
        "new_topic_suggestions": {"type": "array", "items": {"type": "string"}},
        "reasoning": {"type": "string"},
    },
    "required": ["assigned_topics", "new_topic_suggestions"],
    "additionalProperties": False,
}

_SUGGESTIONS_SCHEMA: dict = {
    "type": "object",
    "properties": {
        "suggested_topics": {"type": "array", "items": {"type": "string"}},
    },
    "required": ["suggested_topics"],
    "additionalProperties": False,
}


class AnthropicProvider(AIProvider):
    """Anthropic Claude provider with singleton client and output_config structured output.

    Constructor signature matches the uniform factory contract in ai/__init__.py:
        api_key, model, context_chars, base_url (accepted but unused — Anthropic SDK
        manages the endpoint; widened so get_provider() can call all providers uniformly).
    """

    def __init__(
        self,
        api_key: str,
        model: str = "claude-sonnet-4-6",
        context_chars: int = 180_000,
        base_url: str | None = None,  # accepted for uniform factory signature; unused
    ):
        self._api_key = api_key
        self._model = model
        self._context_chars = context_chars
        # Singleton: created once in __init__, reused for all calls on this instance.
        # Do NOT recreate per API call — AsyncAnthropic wraps an httpx.AsyncClient
        # that maintains a connection pool; recreating per call destroys pool reuse
        # and forces a new TLS handshake per request (D-07 / RESEARCH.md).
        self._client = anthropic.AsyncAnthropic(api_key=self._api_key)

    def _truncate(self, text: str) -> str:
        """D-13 smart truncation: first 60% + last 40% of context window.

        Captures both document introduction and conclusion, which carry the
        most topic signal for long documents.
        """
        if len(text) <= self._context_chars:
            return text
        head_len = int(self._context_chars * 0.6)
        tail_len = self._context_chars - head_len
        return text[:head_len] + "\n[...truncated...]\n" + text[-tail_len:]

    async def classify(
        self,
        document_text: str,
        existing_topics: list[str],
        system_prompt: str,
    ) -> ClassificationResult:
        topics_str = ", ".join(existing_topics) if existing_topics else "(none yet)"
        user_msg = (
            f"Existing topics: [{topics_str}]\n\n"
            f"Document text:\n{self._truncate(document_text)}"
        )
        response = await self._client.messages.create(
            model=self._model,
            max_tokens=1024,
            system=system_prompt,
            messages=[{"role": "user", "content": user_msg}],
            output_config={"format": {"type": "json_schema", "schema": _CLASSIFICATION_SCHEMA}},
        )
        # Graceful degradation (T-07-08): when stop_reason is "refusal" or "max_tokens"
        # the constrained decoding did not complete — fall back to parse_classification("")
        # which returns an empty ClassificationResult rather than raising an exception.
        stop_reason = getattr(response, "stop_reason", "end_turn")
        if response.content and stop_reason == "end_turn":
            raw = response.content[0].text
        else:
            raw = ""
        return parse_classification(raw)

    async def suggest_topics(
        self,
        document_text: str,
        system_prompt: str,
    ) -> list[str]:
        user_msg = (
            "Suggest 3-5 topic names for this document. "
            "Return ONLY valid JSON: {\"suggested_topics\": [\"topic1\", \"topic2\"]}\n\n"
            f"Document text:\n{self._truncate(document_text)}"
        )
        response = await self._client.messages.create(
            model=self._model,
            max_tokens=256,
            system=system_prompt,
            messages=[{"role": "user", "content": user_msg}],
            output_config={"format": {"type": "json_schema", "schema": _SUGGESTIONS_SCHEMA}},
        )
        stop_reason = getattr(response, "stop_reason", "end_turn")
        if response.content and stop_reason == "end_turn":
            raw = response.content[0].text
        else:
            raw = ""
        return parse_suggestions(raw)

    async def health_check(self) -> bool:
        """Verify API key validity and connectivity by sending a minimal message.

        Does NOT pass output_config — the response shape does not matter here;
        this only confirms the api_key and network path are working.
        """
        try:
            await self._client.messages.create(
                model=self._model,
                max_tokens=8,
                messages=[{"role": "user", "content": "ping"}],
            )
            return True
        except Exception:
            return False