# Haiku-tier cacheable system-prompt seed.
# Composed deterministically from src/symfonic/core/providers.py so
# live capture is fully reproducible. Target: ~3k tokens (just over
# the Haiku 2048-token minimum cacheable prefix per Anthropic docs).
#
# This seed MUST be byte-identical across every turn of fixtures
# cached_haiku_repeat and cached_haiku_with_rewrite — that is the
# precondition for Anthropic prompt-cache HITs on turns 2+.

You are an ops bot embedded in the symfonic-core framework. The
reference source you must answer from is the providers module of
the framework, reproduced below verbatim. Be concise; answer in one
or two sentences unless asked to elaborate.

## Reference: src/symfonic/core/providers.py

"""ModelProvider protocol + concrete implementations.

Per ADR-PFX-003 and TDD SS2.4:
- ModelProvider: protocol for LLM instantiation
- AnthropicProvider: default, installed via symfonic-core[anthropic]
- OpenAIProvider: installed via symfonic-core[openai]
- DeepSeekProvider: installed via symfonic-core[deepseek]
- GoogleProvider: installed via symfonic-core[google]
- OllamaProvider: installed via symfonic-core[ollama]
"""

from __future__ import annotations

import logging
import os
from typing import TYPE_CHECKING, Any, ClassVar, Protocol, runtime_checkable

# BaseChatModel is used only as a return-type annotation and we have
# `from __future__ import annotations`, so all hints are stringified.
# Importing it at runtime pulls langchain_core.language_models which
# transitively eager-loads PIL.Image (via chat_template_utils). Keeping
# this under TYPE_CHECKING preserves the lazy-import contract of
# symfonic.agent.attachments.
if TYPE_CHECKING:
    from langchain_core.language_models import BaseChatModel

from .config import ModelConfig

logger = logging.getLogger("symfonic.core.providers")


def _api_key_fingerprint(api_key: str | None) -> str:
    """Return last 8 chars of an API key, or '<none>' if unset."""
    if not api_key:
        return "<none>"
    if len(api_key) < 12:
        return "***"
    return f"…{api_key[-8:]}"


@runtime_checkable
class ModelProvider(Protocol):
    """Protocol for LLM model construction."""

    def get_chat_model(self, config: ModelConfig) -> BaseChatModel: ...

    def supports_thinking(self) -> bool: ...

    def supports_streaming(self) -> bool: ...

    def supports_forced_tool_choice(self, config: ModelConfig) -> bool:
        """v7.8.5: provider-level introspection for the v7.8.3
        ``procedural_force_first_action_tool`` lever.

        Returns ``True`` when the provider can honour
        ``llm.bind_tools(tools, tool_choice=<name>)`` under the given
        ``ModelConfig``.  Returns ``False`` when the provider would
        reject forced tool_choice (Anthropic's API returns ``400`` when
        ``thinking`` is enabled alongside ``tool_choice={type:tool,...}``;
        see Anthropic's published extended-thinking constraint).

        Default implementation returns ``True`` -- providers that have
        a no-force constraint override.  The engine's
        ``_maybe_resolve_forced_tool_choice`` consults this BEFORE
        stamping ``forced_tool_choice`` into the LangGraph state, so
        adopters who configure ``thinking`` see a clean WARN log and
        a forced no-op rather than a runtime 400 from the API
        boundary.

        Implementations MUST be pure functions of the passed
        ``ModelConfig`` -- no I/O, no provider state mutation -- so
        the engine can call this on every turn without cost.

        v7.8.3 architectural-line preservation: the framework
        constrains the choice set when it can; abstains cleanly when
        it cannot.  Auto-disabling ``config.thinking`` would cross the
        line (reasoning is model-owned).
        """
        return True

    # v7.19.2 NOTE: ``refusal_reason(config) -> str | None`` is an
    # OPTIONAL diagnostic companion the engine consults via
    # :func:`getattr` (see ``engine.py::_maybe_resolve_forced_tool_choice``).
    # It is NOT in the Protocol body deliberately so existing third-
    # party providers continue to satisfy the ``runtime_checkable``
    # Protocol without modification.  Providers that override
    # ``supports_forced_tool_choice`` are ENCOURAGED to also implement
    # ``refusal_reason`` so the per-turn refusal WARN names the
    # structural cause; providers that omit it fall through to the
    # engine's generic v7.8.5-contract fallback text.


class AnthropicProvider:
    """Anthropic-specific model construction. Requires symfonic-core[anthropic].

    Reads ``ANTHROPIC_API_KEY`` from the environment on first
    ``get_chat_model()`` call; logs an INFO line with the last 8 chars
    of the key (never the full value) so operators can tell whether
    the running process picked up the expected key.
    """

    #: v7.13.4: self-declare the wire dialect so
    #: :func:`_detect_provider_family` doesn't need to introspect via
    #: ``isinstance()`` (kept as a fallback for forks that subclass).
    #: Adopters who implement :class:`ModelProvider` Protocol directly
    #: (e.g. for per-request key injection) declare the same ClassVar
    #: on their own class with the appropriate value.
    _symfonic_provider_family: ClassVar[str] = "anthropic"

    _logged_init: bool = False

    def get_chat_model(self, config: ModelConfig) -> BaseChatModel:
        try:
            from langchain_anthropic import ChatAnthropic
        except ImportError as e:
            raise ImportError(
                "langchain-anthropic is required for AnthropicProvider. "
                "Install it via: pip install symfonic-core[anthropic]"
            ) from e

        # Log once per provider instance: which key is in use.
        if not self._logged_init:
            logger.info(
                "AnthropicProvider initialized: api_key=%s model=%s",
                _api_key_fingerprint(os.environ.get("ANTHROPIC_API_KEY")),
                config.model_name,
            )
            self._logged_init = True

        kwargs: dict[str, Any] = {
            "model": config.model_name,
            "temperature": config.temperature,
            "max_tokens": config.max_tokens,
            "max_retries": config.max_retries,
        }
        if config.extra_headers:
            kwargs["extra_headers"] = config.extra_headers
        if config.thinking:
            # v7.13.2: pass ``thinking`` as a first-class kwarg.
            # ``langchain-anthropic>=0.3.20`` exposes ``thinking`` as a
            # native ChatAnthropic field; threading it via
            # ``model_kwargs`` triggers ``UserWarning: Parameters
            # {'thinking'} should be specified explicitly``.  Jarvio
            # flagged this informationally on 2026-05-31; pin bump in
            # pyproject.toml guarantees the field is available.
            kwargs["thinking"] = config.thinking
        return ChatAnthropic(**kwargs)

    def supports_thinking(self) -> bool:
        return True

    def supports_streaming(self) -> bool:
        return True

    def supports_forced_tool_choice(self, config: ModelConfig) -> bool:
        """v7.8.5: Anthropic API rejects ``tool_choice={type:tool,
        name:X}`` and ``{type:any}`` when ``thinking`` is enabled.

        Per the published extended-thinking constraint at
        ``platform.claude.com/docs/en/docs/build-with-claude/
        extended-thinking``: when thinking is enabled, only
        ``{type:auto}`` and ``{type:none}`` are accepted; forced
        tool_choice returns ``400``.

        Returns ``False`` when ``config.thinking`` is set, so the
        engine's force lever cleanly abstains instead of crashing the
        API call at runtime.  Returns ``True`` otherwise (the
        unconstrained Anthropic path).
        """
        return not bool(config.thinking)

    def refusal_reason(self, config: ModelConfig) -> str | None:
        """v7.19.2: human-readable reason for the extended-thinking
        refusal, surfaced in the engine's per-turn refusal WARN."""
        if config.thinking:
            return (
                "Anthropic API rejects tool_choice={type:tool,name:X} "
                "when extended thinking is enabled (returns HTTP 400 "
                "per the published constraint)."
            )
        return None

    @staticmethod
    def extract_cache_ttl(input_token_details: Any) -> str | None:
        """v7.20.0 T-7.20.0.6: mine the Anthropic 1h-cache TTL from a
        LangChain ``usage_metadata.input_token_details`` dict.

        Anthropic's API surfaces the prompt-cache write rate via the
        ``cache_creation: {ephemeral_5m_input_tokens, ephemeral_1h_input_tokens}``
        sub-object on the response usage block.  langchain-anthropic
        (>= 0.3.20) flattens that into ``input_token_details`` with the
        same key names — see ``langchain_anthropic.chat_models:2600-2610``.

        When ``ephemeral_1h_input_tokens > 0`` (the 1h cache was used on
        this turn), this returns ``"1h"`` so the cost layer picks the
        ``cache_write_1h`` rate from the pricing row (T-7.20.0.5).
        Otherwise — pure cache_read, only 5m cache, or no cache_creation
        at all — returns ``None`` so the cost layer falls through to
        ``cache_write`` (the 5-min rate, which IS the documented default
        in the registry).

        Provider boundary placement (architect verdict 2026-06-05 §5.c):
        wire-shape translation lives on the provider that owns the wire
        contract.  The framework-level usage extractors in
        ``callbacks/emit.py`` and ``nodes/react.py`` consult this helper
        so they stay provider-agnostic — they pass through the
        ``input_token_details`` dict without coupling themselves to
        Anthropic-specific key names.

        Args:
            input_token_details: The ``input_token_details`` sub-dict of
                a LangChain ``usage_metadata`` payload.  May be ``None``
                (no cache info), an empty dict, or any non-dict value
                (defensive — never raises).

        Returns:
            ``"1h"`` when the 1h cache was used this turn.  ``None``
            otherwise (5m cache only, no cache, or non-dict input).
        """
        if not isinstance(input_token_details, dict):
            return None
        # ``or 0`` guards against the field being present but ``None``
        # (older langchain-anthropic surfaced a None when 1h cache was
        # absent rather than omitting the key).
        ephemeral_1h = input_token_details.get("ephemeral_1h_input_tokens") or 0
        if ephemeral_1h > 0:
            return "1h"
        return None


class OpenAIProvider:
    """OpenAI model construction. Requires langchain-openai.

    Install via: pip install symfonic-core[openai]
    """

    #: v7.13.4: see :class:`AnthropicProvider` for the rationale.
    _symfonic_provider_family: ClassVar[str] = "openai"

    def __init__(self, base_url: str | None = None, api_key: str | None = None):
        self._base_url = base_url
        self._api_key = api_key

    def get_chat_model(self, config: ModelConfig) -> BaseChatModel:
        try:
            from langchain_openai import ChatOpenAI
        except ImportError as e:
            raise ImportError(
                "langchain-openai is required for OpenAIProvider. "
                "Install it via: pip install symfonic-core[openai]"
            ) from e

        kwargs: dict[str, Any] = {
            "model": config.model_name,
            "temperature": config.temperature,
            "max_tokens": config.max_tokens,
            "max_retries": config.max_retries,
            # Emit a usage chunk at the end of every SSE stream so
            # input_tokens / output_tokens are non-zero on streamed calls.
            # LangChain defaults stream_usage=False.
            "stream_usage": True,
        }
        if self._base_url:
            kwargs["base_url"] = self._base_url
        if self._api_key:
            kwargs["api_key"] = self._api_key
        return ChatOpenAI(**kwargs)

    def supports_thinking(self) -> bool:
        return False  # Unless o1/o3 models

    def supports_streaming(self) -> bool:
        return True

    def supports_forced_tool_choice(self, config: ModelConfig) -> bool:
        # v7.8.5: OpenAI honours ``tool_choice`` unconditionally.
        return True


class DeepSeekProvider:
    """DeepSeek via OpenAI-compatible API. Requires langchain-openai.

    Install via: pip install symfonic-core[deepseek]
    """

    #: v7.13.4: see :class:`AnthropicProvider` for the rationale.
    _symfonic_provider_family: ClassVar[str] = "openai"

    def __init__(self, api_key: str | None = None):
        self._api_key = api_key

    def get_chat_model(self, config: ModelConfig) -> BaseChatModel:
        try:
            from langchain_openai import ChatOpenAI
        except ImportError as e:
            raise ImportError(
                "langchain-openai is required for DeepSeekProvider. "
                "Install it via: pip install symfonic-core[deepseek]"
            ) from e

        kwargs: dict[str, Any] = {
            "model": config.model_name or "deepseek-chat",
            "temperature": config.temperature,
            "max_tokens": config.max_tokens,
            "base_url": "https://api.deepseek.com/v1",
            # Emit a usage chunk at the end of every SSE stream so
            # input_tokens / output_tokens are non-zero on streamed calls.
            # LangChain defaults stream_usage=False.
            "stream_usage": True,
        }
        if self._api_key:
            kwargs["api_key"] = self._api_key
        return ChatOpenAI(**kwargs)

    def supports_thinking(self) -> bool:
        return True  # DeepSeek-R1 has reasoning

    def supports_streaming(self) -> bool:
        return True

    def supports_forced_tool_choice(self, config: ModelConfig) -> bool:
        # v7.8.5: DeepSeek goes through the OpenAI-compatible API
        # which honours ``tool_choice`` unconditionally.
        return True


class GoogleProvider:
    """Google Gemini models. Requires langchain-google-genai.

    Install via: pip install symfonic-core[google]
    """

    #: v7.13.4: see :class:`AnthropicProvider` for the rationale.
    _symfonic_provider_family: ClassVar[str] = "google"

    def get_chat_model(self, config: ModelConfig) -> BaseChatModel:
        try:
            from langchain_google_genai import ChatGoogleGenerativeAI
        except ImportError as e:
            raise ImportError(
                "langchain-google-genai is required for GoogleProvider. "
                "Install it via: pip install symfonic-core[google]"
            ) from e

        return ChatGoogleGenerativeAI(
            # v7.19.8 (2026-06-08 registry refresh): default flipped from
            # ``gemini-2.0-flash`` (shut down 2026-06-01 at the provider)
            # to ``gemini-2.5-flash``.  Adopters who relied on the old
            # default were already broken at the wire by the provider
            # retirement; new builds get a live default.
            model=config.model_name or "gemini-2.5-flash",
            temperature=config.temperature,
            max_tokens=config.max_tokens,
        )

    def supports_thinking(self) -> bool:
        return True  # Gemini 2.5 has thinking

    def supports_streaming(self) -> bool:
        return True

    def supports_forced_tool_choice(self, config: ModelConfig) -> bool:
        # v7.8.5: Google Gemini honours ``tool_choice`` (function
        # calling mode) unconditionally.  No known thinking
        # constraint analogous to Anthropic's.
        return True


class OllamaProvider:
    """Local Ollama models. Requires langchain-ollama.

    Install via: pip install symfonic-core[ollama]
    """

    #: v7.13.4: see :class:`AnthropicProvider` for the rationale.
    _symfonic_provider_family: ClassVar[str] = "openai"

    def __init__(self, base_url: str = "http://localhost:11434"):
        self._base_url = base_url

    def get_chat_model(self, config: ModelConfig) -> BaseChatModel:
        try:
            from langchain_ollama import ChatOllama
        except ImportError as e:
            raise ImportError(
                "langchain-ollama is required for OllamaProvider. "
                "Install it via: pip install symfonic-core[ollama]"
            ) from e

        return ChatOllama(
            model=config.model_name or "llama3.2",
            base_url=self._base_url,
            temperature=config.temperature,
        )

    def supports_thinking(self) -> bool:
        return False

    def supports_streaming(self) -> bool:
        return True

    def supports_forced_tool_choice(self, config: ModelConfig) -> bool:
        # v7.8.5: Ollama tool-call support depends on the underlying
        # model.  Default ``True`` -- adopters running tool-bearing
        # local models (e.g. llama3.1) honour ``tool_choice``.
        return True


class MultiProviderRouter:
    """v7.9.1: prefix-based ``ModelProvider`` dispatcher.

    Closes the v7.9.0 adoption gap: ``FrameworkConfig.role_models``
    lets adopters route different roles to different
    ``ModelConfig`` instances, but ``SymfonicAgent`` accepts a
    SINGLE ``model_provider`` -- so a ``ModelConfig(model_name=
    "gpt-4.1-nano")`` route handed to ``AnthropicProvider``
    builds ``ChatAnthropic(model="gpt-4.1-nano")`` and crashes at
    the API boundary with ``404 model_not_found``.

    This router satisfies the ``ModelProvider`` protocol and
    dispatches each call to a backing provider based on the
    ``ModelConfig.model_name`` prefix.  Adopters compose it once
    at agent construction; ``SymfonicAgent`` never sees the
    multi-provider topology.

    Example::

        from symfonic.core.providers import (
            AnthropicProvider, MultiProviderRouter, OpenAIProvider,
        )

        provider = MultiProviderRouter(
            default=AnthropicProvider(),
            routes={
                "gpt-": OpenAIProvider(),
                "o1-": OpenAIProvider(),
                "o3-": OpenAIProvider(),
            },
        )
        agent = SymfonicAgent(
            model_provider=provider,
            config=FrameworkConfig(
                agent=AgentConfig(
                    model=ModelConfig(model_name="claude-opus-4-6"),
                ),
                role_models={
                    "action": ModelConfig(model_name="gpt-4.1-nano"),
                },
            ),
        )

    Routes are matched in insertion order via ``startswith`` (the
    first matching prefix wins).  Unmatched ``model_name`` strings
    fall back to ``default``.  Empty / ``None`` ``model_name``
    falls back to ``default`` (the construction-time provider's
    default model).

    Architectural-line preserved (v7.8.3): the router is wiring,
    not policy.  It exposes the ``ModelProvider`` surface and
    delegates per call; it does not modify ``ModelConfig``
    instances, does not log model selection (adopter telemetry is
    the right place), and does not maintain per-call state.
    """

    def __init__(
        self,
        *,
        default: ModelProvider,
        routes: dict[str, ModelProvider] | None = None,
    ) -> None:
        """Construct the router.

        Args:
            default: The provider used for any ``ModelConfig`` whose
                ``model_name`` does not match a registered prefix
                (or has an empty / ``None`` ``model_name``).
            routes: Mapping from prefix string to ``ModelProvider``.
                Matched in insertion order via ``startswith``.  An
                empty / ``None`` mapping makes the router a
                pass-through to ``default``.
        """
        self._default: ModelProvider = default
        self._routes: dict[str, ModelProvider] = dict(routes or {})
        # v7.11.0 (closed-enumeration audit, Part C): warn adopters
        # when an earlier prefix would shadow a longer one that can
        # never fire.  Example: ``{"gpt-": OpenAIProvider(),
        # "gpt-4.1-": SpecificProvider()}`` -- ``"gpt-"`` matches
        # every gpt-4.1-* model first, so SpecificProvider() is dead
        # code.  Insertion-order semantics are documented but the
        # footgun is silent -- this WARN surfaces it once at
        # construction so adopters can re-order.  Architectural
        # principle: closed-set enumerations are framework
        # boundaries; cross-boundary extension (custom route maps)
        # needs an adopter seam AND honest feedback when the seam is
        # misconfigured.
        self._warn_shadowed_routes()

    def _warn_shadowed_routes(self) -> None:
        """Detect and WARN when a route's prefix is a strict prefix
        of a later-registered route's prefix.  The later route can
        never fire because the earlier one matches first."""
        keys = [k for k in self._routes if k]
        shadowed: list[tuple[str, str]] = []
        for i, earlier in enumerate(keys):
            for later in keys[i + 1:]:
                if later.startswith(earlier) and later != earlier:
                    shadowed.append((earlier, later))
        if shadowed:
            import logging
            _log = logging.getLogger(__name__)
            for earlier, later in shadowed:
                _log.warning(
                    "MultiProviderRouter: route %r is shadowed by "
                    "the earlier-registered route %r and will never "
                    "fire.  Re-register routes longest-prefix-first "
                    "to resolve.",
                    later, earlier,
                )

    def _pick(self, config: ModelConfig) -> ModelProvider:
        """Resolve the backing provider for the given ``config``.

        Pure function of ``config.model_name``; no side effects.
        """
        name = (config.model_name or "")
        for prefix, provider in self._routes.items():
            if prefix and name.startswith(prefix):
                return provider
        return self._default

    def get_chat_model(self, config: ModelConfig) -> BaseChatModel:
        return self._pick(config).get_chat_model(config)

    def supports_thinking(self) -> bool:
        """Delegate to ``default`` -- this method is called without a
        ``config`` per the v7.x protocol, so the safest answer is
        the default provider's capability."""
        return self._default.supports_thinking()

    def supports_streaming(self) -> bool:
        """Delegate to ``default`` for the same reason as
        ``supports_thinking``."""
        return self._default.supports_streaming()

    def supports_forced_tool_choice(self, config: ModelConfig) -> bool:
        """Delegate to the route-matched provider so per-role forcing
        eligibility tracks the right model.  v7.8.5 refuse-to-force
        contract: the engine queries this BEFORE binding tools, so
        the answer must reflect the model that will actually receive
        the call -- not the default."""
        return self._pick(config).supports_forced_tool_choice(config)

    def refusal_reason(self, config: ModelConfig) -> str | None:
        """v7.19.2: delegate refusal-reason diagnostics to the route-
        matched provider so the per-turn WARN names the right cause."""
        picked = self._pick(config)
        getter = getattr(picked, "refusal_reason", None)
        if callable(getter):
            try:
                return getter(config)
            except Exception:
                return None
        return None
