apps/api/app/core/claude.py

"""Claude model router + a deterministic offline engine.

Model routing by task weight (spec §3): Haiku 4.5 plans and routes, Sonnet 4.6
does the worker and writer reasoning. Cost discipline is stated as architecture,
not as a comment.

Two engines implement the same `LLMEngine` protocol:

``AnthropicEngine``
    Real Claude calls. Used whenever ``ANTHROPIC_API_KEY`` is present.

``DeterministicEngine``
    A rule-based stand-in that drives the **identical graph, the identical MCP
    tools and the identical deterministic verifier**. It exists so the quality
    gates (pytest, the 20-run eval, CI) run at ``$0`` and without network access
    to Anthropic. It never invents market data: every number still comes from the
    MCP tools, which still call yfinance and the real RSS feeds. What it replaces
    is only the *decision-making* an LLM would otherwise contribute.
"""

from __future__ import annotations

import json
import re
from collections.abc import Sequence
from dataclasses import dataclass, field
from enum import StrEnum
from typing import Any, Protocol, cast

from app.core.budget import Usage
from app.core.settings import Settings, get_settings


class AgentRole(StrEnum):
    """Which routed model handles a call."""

    SUPERVISOR = "supervisor"
    DATA = "data_agent"
    NEWS = "news_agent"
    WRITER = "writer"


class PromptHint(StrEnum):
    """Opaque tag identifying *which* decision is being asked for.

    Ignored entirely by :class:`AnthropicEngine`; used by
    :class:`DeterministicEngine` to dispatch to the matching rule.
    """

    SUPERVISOR_PLAN = "supervisor.plan"
    DATA_ANALYSE = "data_agent.analyse"
    NEWS_ASSESS = "news_agent.assess"
    WRITER_COMPOSE = "writer.compose"


class LLMRefusalError(RuntimeError):
    """The model declined the request (``stop_reason == "refusal"``)."""


@dataclass(frozen=True, slots=True)
class ToolSpec:
    """A tool advertised to the model."""

    name: str
    description: str
    input_schema: dict[str, Any]

    def to_anthropic(self) -> dict[str, Any]:
        return {
            "name": self.name,
            "description": self.description,
            "input_schema": self.input_schema,
        }


@dataclass(frozen=True, slots=True)
class ToolCall:
    """A tool invocation requested by the model."""

    id: str
    name: str
    arguments: dict[str, Any]


@dataclass(frozen=True, slots=True)
class LLMRequest:
    """One model call."""

    role: AgentRole
    hint: PromptHint
    system: str
    messages: list[dict[str, Any]]
    tools: Sequence[ToolSpec] = ()
    forced_tool: str | None = None
    max_tokens: int | None = None
    #: Structured data also rendered into `messages`. The Anthropic engine never
    #: reads it; the deterministic engine uses it instead of parsing prose.
    context: dict[str, Any] = field(default_factory=dict)


@dataclass(frozen=True, slots=True)
class LLMResult:
    """One model response."""

    model: str
    text: str
    tool_calls: list[ToolCall]
    usage: Usage
    stop_reason: str
    engine: str

    def first_tool(self, name: str) -> dict[str, Any] | None:
        for call in self.tool_calls:
            if call.name == name:
                return call.arguments
        return None


class LLMEngine(Protocol):
    """Everything the graph needs from a language model."""

    name: str

    def model_for(self, role: AgentRole) -> str:
        """The model id this engine would use for `role`."""
        ...

    async def complete(self, request: LLMRequest) -> LLMResult:
        """Execute one model call."""
        ...


# --------------------------------------------------------------------------- #
# Anthropic engine                                                            #
# --------------------------------------------------------------------------- #


class AnthropicEngine:
    """Real Claude calls through the official SDK."""

    name = "anthropic"

    def __init__(self, settings: Settings | None = None) -> None:
        import anthropic  # imported lazily so the package is optional offline

        self._settings = settings or get_settings()
        if not self._settings.anthropic_api_key:
            raise ValueError("ANTHROPIC_API_KEY is required for the anthropic engine")
        self._client = anthropic.AsyncAnthropic(
            api_key=self._settings.anthropic_api_key,
            timeout=self._settings.llm_timeout_seconds,
            max_retries=self._settings.llm_max_retries,
        )

    def model_for(self, role: AgentRole) -> str:
        cfg = self._settings
        if role is AgentRole.SUPERVISOR:
            return cfg.model_supervisor
        if role is AgentRole.WRITER:
            return cfg.model_writer
        return cfg.model_worker

    async def complete(self, request: LLMRequest) -> LLMResult:
        model = self.model_for(request.role)
        kwargs: dict[str, Any] = {
            "model": model,
            "max_tokens": request.max_tokens or self._settings.llm_max_tokens,
            "system": request.system,
            "messages": request.messages,
        }
        if request.tools:
            kwargs["tools"] = [t.to_anthropic() for t in request.tools]
            if request.forced_tool:
                kwargs["tool_choice"] = {"type": "tool", "name": request.forced_tool}

        # Haiku 4.5 rejects `effort`; Sonnet 4.6 supports it. Adaptive thinking is
        # enabled for the reasoning workers but not for the writer, whose call
        # uses a forced tool choice and needs no exploration.
        if request.role is not AgentRole.SUPERVISOR:
            kwargs["output_config"] = {"effort": self._settings.worker_effort}
            if request.role in (AgentRole.DATA, AgentRole.NEWS):
                kwargs["thinking"] = {"type": "adaptive"}

        response = await self._client.messages.create(**kwargs)

        stop_reason = str(response.stop_reason or "end_turn")
        if stop_reason == "refusal":
            raise LLMRefusalError(f"{model} declined the request")

        text_parts: list[str] = []
        tool_calls: list[ToolCall] = []
        for block in response.content:
            block_type = getattr(block, "type", "")
            if block_type == "text":
                text_parts.append(str(getattr(block, "text", "")))
            elif block_type == "tool_use":
                tool_calls.append(
                    ToolCall(
                        id=str(getattr(block, "id", "")),
                        name=str(getattr(block, "name", "")),
                        arguments=cast(dict[str, Any], getattr(block, "input", {}) or {}),
                    )
                )

        raw_usage = response.usage
        usage = Usage(
            input_tokens=int(getattr(raw_usage, "input_tokens", 0) or 0),
            output_tokens=int(getattr(raw_usage, "output_tokens", 0) or 0),
            cache_read_tokens=int(getattr(raw_usage, "cache_read_input_tokens", 0) or 0),
            cache_write_tokens=int(getattr(raw_usage, "cache_creation_input_tokens", 0) or 0),
        )
        return LLMResult(
            model=model,
            text="\n".join(text_parts).strip(),
            tool_calls=tool_calls,
            usage=usage,
            stop_reason=stop_reason,
            engine=self.name,
        )


# --------------------------------------------------------------------------- #
# Deterministic engine                                                         #
# --------------------------------------------------------------------------- #

#: Transparent sentiment lexicon. Applied to *real* headline text pulled from the
#: live RSS feeds — this is a baseline scorer, not fabricated sentiment.
_POSITIVE_TERMS: frozenset[str] = frozenset(
    {
        "beat",
        "beats",
        "surge",
        "surges",
        "record",
        "upgrade",
        "upgraded",
        "rally",
        "gain",
        "gains",
        "growth",
        "profit",
        "strong",
        "soar",
        "soars",
        "jump",
        "jumps",
        "outperform",
        "raises",
        "raise",
        "expands",
        "wins",
        "approval",
        "breakthrough",
        "bullish",
        "rebound",
        "optimism",
        "milestone",
    }
)
_NEGATIVE_TERMS: frozenset[str] = frozenset(
    {
        "miss",
        "misses",
        "plunge",
        "plunges",
        "slump",
        "downgrade",
        "downgraded",
        "loss",
        "losses",
        "weak",
        "fall",
        "falls",
        "drop",
        "drops",
        "lawsuit",
        "probe",
        "investigation",
        "recall",
        "layoff",
        "layoffs",
        "cuts",
        "cut",
        "warning",
        "warns",
        "bearish",
        "decline",
        "declines",
        "halt",
        "fine",
        "penalty",
        "delay",
        "delays",
        "risk",
        "slowdown",
        "selloff",
    }
)
#: Headline patterns that constitute a flagged risk event.
_RISK_PATTERNS: tuple[tuple[str, str], ...] = (
    (r"\b(lawsuit|sued|litigation|class action)\b", "legal"),
    (r"\b(probe|investigation|subpoena|antitrust|regulator\w*)\b", "regulatory"),
    (r"\b(recall|defect|safety)\b", "product"),
    (r"\b(layoff\w*|job cuts|restructur\w*)\b", "restructuring"),
    (r"\b(guidance cut|lowers guidance|profit warning|downgrade[ds]?)\b", "guidance"),
    (r"\b(breach|hack|ransomware|outage)\b", "security"),
    (r"\b(short seller|fraud|accounting)\b", "integrity"),
)


def _tokenise(text: str) -> list[str]:
    return re.findall(r"[a-z']+", text.lower())


def score_headlines(headlines: Sequence[str]) -> tuple[float, str]:
    """Lexicon sentiment over real headlines → (score in [-1, 1], reasoning).

    The reasoning string is deliberately **numeral-free**: it is interpolated into
    brief narrative, where the schema forbids bare digits (every figure must be a
    verified claim reference). Counts are therefore described qualitatively and
    the score itself is carried as a claim.
    """
    if not headlines:
        return 0.0, "No headlines were retrieved, so sentiment defaults to neutral."
    positive = 0
    negative = 0
    hits: list[str] = []
    for headline in headlines:
        for token in _tokenise(headline):
            if token in _POSITIVE_TERMS:
                positive += 1
                hits.append(token)
            elif token in _NEGATIVE_TERMS:
                negative += 1
                hits.append(token)
    total = positive + negative
    if total == 0:
        return 0.0, "The retrieved headlines carried no directional language."
    score = round((positive - negative) / total, 3)
    if positive > negative:
        lead = "Positive language outweighed negative"
    elif negative > positive:
        lead = "Negative language outweighed positive"
    else:
        lead = "Positive and negative language were evenly balanced"
    sample = ", ".join(dict.fromkeys(hits[:6]))
    return score, f"{lead} across the retrieved headlines ({sample})."


def extract_risk_events(ticker: str, headlines: Sequence[str]) -> list[dict[str, str]]:
    """Flag risk headlines by pattern match over real headline text."""
    events: list[dict[str, str]] = []
    for headline in headlines:
        lowered = headline.lower()
        for pattern, category in _RISK_PATTERNS:
            if re.search(pattern, lowered):
                events.append(
                    {"ticker": ticker, "category": category, "headline": headline.strip()}
                )
                break
    return events


class DeterministicEngine:
    """Rule-based stand-in for Claude used by CI, the eval harness and demos.

    Produces the same structured tool calls the real agents would, so the graph,
    the MCP tool layer and the verifier are all exercised identically.
    """

    name = "deterministic"

    def __init__(self, settings: Settings | None = None) -> None:
        self._settings = settings or get_settings()
        self._counter = 0

    def model_for(self, role: AgentRole) -> str:
        cfg = self._settings
        if role is AgentRole.SUPERVISOR:
            return f"{cfg.model_supervisor}#deterministic"
        if role is AgentRole.WRITER:
            return f"{cfg.model_writer}#deterministic"
        return f"{cfg.model_worker}#deterministic"

    def _next_id(self, prefix: str) -> str:
        self._counter += 1
        return f"{prefix}_{self._counter:04d}"

    async def complete(self, request: LLMRequest) -> LLMResult:
        handlers = {
            PromptHint.SUPERVISOR_PLAN: self._plan,
            PromptHint.DATA_ANALYSE: self._analyse,
            PromptHint.NEWS_ASSESS: self._assess,
            PromptHint.WRITER_COMPOSE: self._compose,
        }
        text, calls = handlers[request.hint](request)
        # Deterministic engine costs nothing; usage is reported as zero so the
        # budget guard and cost readout never over-state real spend.
        return LLMResult(
            model=self.model_for(request.role),
            text=text,
            tool_calls=calls,
            usage=Usage(),
            stop_reason="end_turn",
            engine=self.name,
        )

    # ------------------------------------------------------------- handlers ---
    def _plan(self, request: LLMRequest) -> tuple[str, list[ToolCall]]:
        pending: list[str] = list(request.context.get("pending_tickers", []))
        plan = {
            "dispatch": ["data_agent", "news_agent"] if pending else [],
            "tickers": pending,
            "reason": (
                f"{len(pending)} ticker(s) still missing data — dispatching both "
                "workers in parallel."
                if pending
                else "All tickers complete — routing to the writer."
            ),
        }
        return json.dumps(plan), []

    def _analyse(self, request: LLMRequest) -> tuple[str, list[ToolCall]]:
        tickers: list[str] = list(request.context.get("tickers", []))
        days = int(request.context.get("days", self._settings.price_history_days))
        calls = [
            ToolCall(
                id=self._next_id("data"),
                name="plan_market_data",
                arguments={"tickers": tickers, "days": days},
            )
        ]
        return (
            f"Fetching {days}d price history, fundamentals and computed metrics "
            f"for {', '.join(tickers) or 'no tickers'} via the MCP tool server."
        ), calls

    def _assess(self, request: LLMRequest) -> tuple[str, list[ToolCall]]:
        per_ticker: dict[str, list[str]] = request.context.get("headlines", {})
        assessments: list[dict[str, Any]] = []
        for ticker, headlines in per_ticker.items():
            score, reasoning = score_headlines(headlines)
            assessments.append(
                {
                    "ticker": ticker,
                    "score": score,
                    "reasoning": reasoning,
                    "risk_events": extract_risk_events(ticker, headlines),
                }
            )
        calls = [
            ToolCall(
                id=self._next_id("news"),
                name="emit_sentiment",
                arguments={"assessments": assessments},
            )
        ]
        return f"Scored {len(assessments)} ticker(s) from live RSS headlines.", calls

    def _compose(self, request: LLMRequest) -> tuple[str, list[ToolCall]]:
        from app.graph.compose import compose_brief_payload

        payload = compose_brief_payload(request.context)
        calls = [
            ToolCall(
                id=self._next_id("brief"),
                name="emit_brief",
                arguments=payload,
            )
        ]
        return "Assembled the brief from verified state — no arithmetic performed.", calls


# --------------------------------------------------------------------------- #
# Factory                                                                      #
# --------------------------------------------------------------------------- #


def build_engine(settings: Settings | None = None) -> LLMEngine:
    """Return the engine this process should use."""
    cfg = settings or get_settings()
    if cfg.resolved_engine == "anthropic":
        return AnthropicEngine(cfg)
    return DeterministicEngine(cfg)