"""Naming contract for strategy-router (auto-router) pseudo-models.

A deployment whose ``litellm_params.model`` starts with ``auto_router/`` does not
name a provider model; the string is the discriminator that selects which
pre-routing strategy owns the deployment. This module is the single source of
truth for classifying that string (``Router._is_*_router_deployment`` delegates
here) and for checking that a client-supplied write leaves the deployment
coherent, so management endpoints can reject corruption with a 400 instead of
the router silently dropping the deployment at load time under
``ignore_invalid_deployments``.
"""

from collections.abc import Mapping, Sequence
from dataclasses import dataclass
from types import MappingProxyType
from typing import Final, Literal, TypeAlias

from litellm.router_strategy.complexity_router.config import (
    COMPLEXITY_ROUTER_CONFIG_KEYS,
    LLM_CLASSIFIER_TYPES,
)

AUTO_ROUTER_MODEL_PREFIX: Final = "auto_router/"

StrategyRouterKind = Literal["semantic", "complexity", "adaptive", "quality"]

StrategyRouterDependencyRole: TypeAlias = Literal["tier", "default", "classifier", "embedding"]


@dataclass(frozen=True, slots=True)
class StrategyRouterDependency:
    """A model name a strategy router must be able to reach to do its job."""

    model_name: str
    role: StrategyRouterDependencyRole


STRATEGY_ROUTER_PARAM_FIELDS: Final[frozenset[str]] = frozenset(
    {
        "auto_router_config",
        "auto_router_config_path",
        "auto_router_default_model",
        "auto_router_embedding_model",
        "auto_router_max_input_chars",
        "complexity_router_config",
        "complexity_router_default_model",
        "adaptive_router_config",
        "quality_router_config",
        "quality_router_default_model",
    }
)

_REQUIRED_FIELD_GROUPS: Final[Mapping[StrategyRouterKind, tuple[tuple[str, ...], ...]]] = {
    "semantic": (
        ("auto_router_config", "auto_router_config_path"),
        ("auto_router_default_model",),
        ("auto_router_embedding_model",),
    ),
    "complexity": (("complexity_router_config", "complexity_router_default_model"),),
    "adaptive": (("adaptive_router_config",),),
    "quality": (("quality_router_config", "quality_router_default_model"),),
}


def classify_strategy_router_model(model: str) -> StrategyRouterKind | None:
    """Classify a ``litellm_params.model`` string the way the Router does.

    Returns None for regular provider models. Mirrors Router registration
    exactly: reserved names are matched by prefix, everything else under
    ``auto_router/`` is a semantic router.
    """
    if not model.startswith(AUTO_ROUTER_MODEL_PREFIX):
        return None
    remainder: Final = model[len(AUTO_ROUTER_MODEL_PREFIX) :]
    if remainder.startswith("complexity_router"):
        return "complexity"
    if remainder.startswith("adaptive_router"):
        return "adaptive"
    if remainder.startswith("quality_router"):
        return "quality"
    return "semantic"


def _named(value: object, role: StrategyRouterDependencyRole) -> tuple[StrategyRouterDependency, ...]:
    """One dependency from a scalar field, or none when it is absent or not a name."""
    return (StrategyRouterDependency(value, role),) if isinstance(value, str) and value else ()


def _pool(value: object, role: StrategyRouterDependencyRole) -> tuple[StrategyRouterDependency, ...]:
    """Dependencies from a field holding either a single name or a pool of them."""
    if isinstance(value, str):
        return _named(value, role)
    if isinstance(value, Sequence):
        return tuple(dep for entry in value for dep in _named(entry, role))
    return ()


_NO_CONFIG: Final[Mapping[str, object]] = MappingProxyType({})


def _mapping(value: object) -> Mapping[str, object]:
    return value if isinstance(value, Mapping) else _NO_CONFIG


def strategy_router_dependencies(
    litellm_params: Mapping[str, object],
) -> tuple[StrategyRouterDependency, ...]:
    """The model names a strategy-router deployment must reach, in no particular order.

    A field is a dependency only under the condition the runtime itself reads it: the
    classifier model needs `classifier_type: llm`, and the complexity embedding model needs
    `semantic_keyword_matching`. Listing one the router never calls reds a working deployment.

    The two default-model spellings are not symmetric. A quality router falls back to its
    config's `default_model`, so both are read. A complexity router ignores that field and
    derives its default from the tiers instead (`fallback_tier`, then MEDIUM, then SIMPLE),
    overwriting the config value at init, so only the `litellm_params` spelling is a
    dependency here; the derived one is already covered as a tier.

    Returns empty for a regular deployment, and for any name this module cannot reach from
    the deployment dict alone: a semantic router's routes live in an `auto_router_config`
    JSON string or an `auto_router_config_path` file, so only its default and embedding
    models are enumerable here. Every field is read defensively, since a caller may hold a
    config the router itself would refuse, and a health check must not raise on one.
    """
    kind: Final = classify_strategy_router_model(str(litellm_params.get("model", "")))
    if kind is None:
        return ()
    if kind == "semantic":
        return _named(litellm_params.get("auto_router_default_model"), "default") + _named(
            litellm_params.get("auto_router_embedding_model"), "embedding"
        )
    if kind == "adaptive":
        return _pool(_mapping(litellm_params.get("adaptive_router_config")).get("available_models"), "tier")
    if kind == "quality":
        quality: Final = _mapping(litellm_params.get("quality_router_config"))
        return tuple(
            dict.fromkeys(
                _pool(quality.get("available_models"), "tier")
                + _named(
                    litellm_params.get("quality_router_default_model") or quality.get("default_model"),
                    "default",
                )
            )
        )
    complexity: Final = _mapping(litellm_params.get("complexity_router_config"))
    classifier: Final = _mapping(complexity.get("classifier_llm_config"))
    return tuple(
        dict.fromkeys(
            tuple(dep for tier in _mapping(complexity.get("tiers")).values() for dep in _pool(tier, "tier"))
            + _named(litellm_params.get("complexity_router_default_model"), "default")
            + (
                _named(classifier.get("model"), "classifier")
                if complexity.get("classifier_type") in LLM_CLASSIFIER_TYPES
                else ()
            )
            + (
                _named(complexity.get("embedding_model"), "embedding")
                if complexity.get("semantic_keyword_matching")
                else ()
            )
        )
    )


def validate_complexity_router_config_write(complexity_router_config: Mapping[str, object] | None) -> str | None:
    """Reject a complexity config the router would refuse to build a deployment from.

    Parsed with the router's own ``ComplexityRouterConfig`` rather than a copy of
    its rules, so the boundary rejects exactly what the load would. Plugins are
    resolved from dotted paths only on the config.yaml path, so a written config
    reaches this function in the same shape the load hands to the same model.
    Judged on the config alone: a patch may write one without naming a model, and
    the stored model is encrypted at rest, so it cannot be classified here.
    """
    from pydantic import ValidationError

    from litellm.router_strategy.complexity_router.config import ComplexityRouterConfig

    if complexity_router_config is None:
        return None
    try:
        _ = ComplexityRouterConfig.model_validate(complexity_router_config)
    except ValidationError as exc:
        first: Final = exc.errors()[0]
        location: Final = ".".join(str(part) for part in first.get("loc", ())) or "complexity_router_config"
        return (
            f"complexity_router_config is invalid at {location}: {first.get('msg', 'invalid value')}. "
            "The router would drop this deployment at load time, so the write is rejected instead."
        )
    return None


_COMPLEXITY_ROUTER_FIELDS: Final[frozenset[str]] = frozenset(
    field for group in _REQUIRED_FIELD_GROUPS["complexity"] for field in group
)


def carries_complexity_router_settings(model: str | None, present_fields: frozenset[str]) -> bool:
    """Whether this deployment configures a complexity router, so is judged on its key set.

    Scoped rather than applied to every deployment because the setting names are only
    unambiguous in this context: ``embedding_model``, for one, is a legitimate flat param
    on an s3_vectors vector store. ``present_fields`` carries the same merged view
    ``validate_strategy_router_model_write`` is judged on, so a router named only by its
    default model is in scope, and a field added to the table above is covered here for free.
    """
    return classify_strategy_router_model(model or "") == "complexity" or bool(
        present_fields & _COMPLEXITY_ROUTER_FIELDS
    )


def validate_complexity_router_config_placement(litellm_params: Mapping[str, object] | None) -> str | None:
    """Reject a complexity-router setting written beside ``complexity_router_config``.

    The router reads its settings only from ``litellm_params.complexity_router_config``, so a
    key one level too high configures nothing. It does not stay inert: the alias-marker
    forwarding carries every unrecognized ``litellm_params`` key onto the outbound request,
    where the provider rejects it as an unknown body field, and the deployment then fails
    every call with an error naming an internal config key. Caller scopes; this judges.
    """
    if litellm_params is None:
        return None
    misplaced: Final = tuple(sorted(frozenset(litellm_params) & COMPLEXITY_ROUTER_CONFIG_KEYS))
    if not misplaced:
        return None
    return (
        f"litellm_params sets complexity_router_config settings directly: {', '.join(misplaced)}. "
        "The router reads these only from complexity_router_config, so there they configure nothing "
        "and are forwarded to the provider as unknown request params, which rejects the call. "
        "Move them under complexity_router_config."
    )


def validate_strategy_router_model_write(model: str, present_fields: frozenset[str]) -> str | None:
    """Check that writing ``model`` leaves a deployment the router can load.

    ``present_fields`` is the set of strategy-router param fields that are
    non-None on the deployment after the write (stored fields merged with the
    incoming ones). Returns a human-readable violation, or None when coherent.
    A config's contents are ``validate_complexity_router_config_write``'s to
    judge, since a write may carry one without naming a model at all.
    """
    kind: Final = classify_strategy_router_model(model)
    if kind is None:
        offending: Final = sorted(present_fields & STRATEGY_ROUTER_PARAM_FIELDS)
        if offending:
            return (
                f"litellm_params.model='{model}' does not start with '{AUTO_ROUTER_MODEL_PREFIX}' but the "
                f"deployment carries auto-router settings ({', '.join(offending)}), so the router could not "
                f"load it. Keep the '{AUTO_ROUTER_MODEL_PREFIX}' prefix; to change the name clients call, "
                "edit the public model_name instead."
            )
        return None
    remainder: Final = model[len(AUTO_ROUTER_MODEL_PREFIX) :]
    if remainder.startswith(AUTO_ROUTER_MODEL_PREFIX):
        return (
            f"litellm_params.model='{model}' repeats the '{AUTO_ROUTER_MODEL_PREFIX}' prefix, so the router "
            f"could not load it. Use '{remainder}'; to change the name clients call, edit the public "
            "model_name instead."
        )
    if not remainder:
        return (
            f"litellm_params.model='{model}' is missing the router name after the '{AUTO_ROUTER_MODEL_PREFIX}' prefix."
        )
    missing: Final = tuple(
        " or ".join(group) for group in _REQUIRED_FIELD_GROUPS[kind] if not any(f in present_fields for f in group)
    )
    if missing:
        return (
            f"litellm_params.model='{model}' selects the {kind} router, which requires "
            f"{'; '.join(missing)} in litellm_params."
        )
    return None