"""Tracker-side validation of the capability report a Node presents at registration.

A Node proves locally that it can execute one exact combination — model artifact,
shard range, recipe, backend/device — and ships that proof with its registration
(ADR-0023, NCA-001/002/003). The tracker does not re-run the forward; it decides
whether the presented proof *covers what the node is advertising*, records the
verdict as a small sanitized enum, and routes only to nodes whose verdict is
`admitted`.

Two properties this module deliberately keeps:

* **No model knowledge.** Model ids, recipe ids, backend ids and device names are
  opaque labels. They are compared, never interpreted; no vendor string is a
  code path here.
* **Evidence, not assertion.** A report is treated as a claim about identity, and
  the tracker only ever *narrows* what a node may serve with it. Nothing in a
  report can widen a node's eligibility or its routing weight — throughput
  routing stays measurement-driven (ADR-0013/0021).

Older nodes that predate the capability protocol present no report at all. They
are handled by an explicit policy (`POLICY_COMPAT` vs `POLICY_ENFORCE`), never by
silently treating "no proof" as "proven" — see `docs/adr/0023-…` for the rollout.
"""

from __future__ import annotations

import os
import re
import time
from dataclasses import dataclass, replace
from typing import Any, Callable, Mapping

from .recipe import (
    CertificationLedger,
    FingerprintMismatch,
    RecipeIdentityError,
    parse_identity,
)

# The capability report layout this tracker reads (meshnet_node.capability).
SUPPORTED_SCHEMA_VERSION = 1

# The oldest recipe catalogue whose recipe semantics this tracker still trusts.
# A node carrying an older catalogue may be running a recipe whose id has since
# been redefined, so its proof cannot be matched to a name reliably.
MIN_CATALOGUE_VERSION = "2026.07.1"

# How old a proof may be *at the moment it is presented*. Freshness after that is
# carried by liveness: a registration is re-asserted on tracker restart and the
# node is purged once heartbeats stop.
DEFAULT_MAX_REPORT_AGE_SECONDS = 900.0

# A proof timestamped further ahead than this is not fresh, it is wrong.
MAX_CLOCK_SKEW_SECONDS = 60.0

STATUS_PASSED = "passed"

# --- Admission verdicts. `admitted` is the only routable one under `enforce`. ---
STATE_ADMITTED = "admitted"
STATE_ABSENT = "absent"
STATE_INVALID = "invalid"
STATE_FAILED = "failed"
STATE_STALE = "stale"
STATE_MODEL_MISMATCH = "model-mismatch"
STATE_SHARD_MISMATCH = "shard-mismatch"
STATE_RECIPE_MISMATCH = "recipe-mismatch"
STATE_CATALOGUE_INCOMPATIBLE = "catalogue-incompatible"
# The node presented a DGR-003 identity block whose declared fingerprint is not
# the digest of the axes it declared. Identity is derived, never asserted.
STATE_FINGERPRINT_MISMATCH = "fingerprint-mismatch"
# Registered-but-dark: a known recipe no real distributed forward has certified.
# Visible to an operator, never routable for user traffic.
STATE_UNCERTIFIED = "uncertified"

ALL_STATES = (
    STATE_ADMITTED,
    STATE_ABSENT,
    STATE_INVALID,
    STATE_FAILED,
    STATE_STALE,
    STATE_MODEL_MISMATCH,
    STATE_SHARD_MISMATCH,
    STATE_RECIPE_MISMATCH,
    STATE_CATALOGUE_INCOMPATIBLE,
    STATE_FINGERPRINT_MISMATCH,
    STATE_UNCERTIFIED,
)

# --- Compatibility policy for nodes that predate the capability protocol. ---
# `compat`  — a node presenting *no* proof still routes (legacy behaviour), but a
#             node presenting a *bad* proof never does. Presenting a broken or
#             mismatched proof is a stronger signal than presenting none.
# `enforce` — only `admitted` routes. Absent proof is not routable.
POLICY_COMPAT = "compat"
POLICY_ENFORCE = "enforce"
ALL_POLICIES = (POLICY_COMPAT, POLICY_ENFORCE)
DEFAULT_POLICY = POLICY_COMPAT

POLICY_ENV_VAR = "MESHNET_TRACKER_CAPABILITY_POLICY"

# Operator-facing detail strings are short and never carry a raw exception.
_MAX_DETAIL_CHARS = 240
_MAX_DIAGNOSTICS = 3

_CREDENTIAL_PATTERNS = (
    re.compile(r"\b[A-Za-z0-9_]{2,6}_[A-Za-z0-9]{16,}\b"),  # hf_…, ghp_…, sk_live_…
    re.compile(r"\bsk-[A-Za-z0-9_-]{16,}\b"),
    re.compile(r"(?i)\bbearer\s+\S+"),
    re.compile(r"(?i)\b(?:token|api[_-]?key|password|secret)\s*[=:]\s*\S+"),
)
_REDACTED = "[redacted]"


def normalize_policy(value: Any) -> str:
    """Return a known policy name, falling back to the default for anything else."""
    if isinstance(value, str) and value.strip().lower() in ALL_POLICIES:
        return value.strip().lower()
    return DEFAULT_POLICY


def policy_from_env(environ: Mapping[str, str] | None = None) -> str:
    env = os.environ if environ is None else environ
    return normalize_policy(env.get(POLICY_ENV_VAR))


def sanitize_detail(text: Any) -> str:
    """Collapse, redact and clip a string bound for an operator view."""
    cleaned = " ".join(str(text).split())
    for pattern in _CREDENTIAL_PATTERNS:
        cleaned = pattern.sub(_REDACTED, cleaned)
    if len(cleaned) > _MAX_DETAIL_CHARS:
        cleaned = cleaned[: _MAX_DETAIL_CHARS - 1].rstrip() + "…"
    return cleaned


def catalogue_is_compatible(version: Any) -> bool:
    """True when `version` is at least `MIN_CATALOGUE_VERSION`.

    Versions are dotted integer sequences (`2026.07.1`). Anything that does not
    parse is incompatible — an unparseable catalogue version cannot be shown to
    be new enough.
    """
    parsed = _parse_version(version)
    if parsed is None:
        return False
    return parsed >= _parse_version(MIN_CATALOGUE_VERSION)  # type: ignore[operator]


def _parse_version(value: Any) -> tuple[int, ...] | None:
    if not isinstance(value, str) or not value.strip():
        return None
    parts = value.strip().split(".")
    try:
        return tuple(int(part) for part in parts)
    except ValueError:
        return None


@dataclass(frozen=True)
class CapabilityState:
    """The tracker's sanitized verdict on one node's presented proof.

    This is what the network map exposes and what route selection consults. It
    holds identity labels and a verdict — never a raw exception, a file path, or
    a credential.
    """

    state: str
    detail: str = ""
    model_id: str | None = None
    shard_start: int | None = None
    shard_end: int | None = None
    recipe_id: str | None = None
    recipe_version: str | None = None
    catalogue_version: str | None = None
    backend_id: str | None = None
    device: str | None = None
    quantization: str | None = None
    validated_at: float | None = None
    recorded_at: float = 0.0
    schema_version: int | None = None
    diagnostics: tuple[str, ...] = ()
    # The DGR-003 compatibility fingerprint, *re-derived* by this tracker from
    # the axes the node declared — never copied from what the node claimed.
    # Absent for a node that predates DGR-003.
    model_artifact_digest: str | None = None
    runtime_recipe_digest: str | None = None
    shard_binding_digest: str | None = None
    # The tracker ledger's verdict on that fingerprint at evaluation time
    # ("dark"/"certified"), so the network map answers "why is this exact node
    # not routing" without a second query. None when no identity was presented.
    certification: str | None = None

    @property
    def proven(self) -> bool:
        """The presented proof covers exactly what the node advertised."""
        return self.state == STATE_ADMITTED

    @property
    def fingerprint(self) -> tuple[str, str] | None:
        """What route formation compares. `None` when the node declares no identity."""
        if self.model_artifact_digest is None or self.runtime_recipe_digest is None:
            return None
        return (self.model_artifact_digest, self.runtime_recipe_digest)

    def routable_under(self, policy: str) -> bool:
        if self.proven:
            return True
        return self.state == STATE_ABSENT and normalize_policy(policy) == POLICY_COMPAT

    def with_state(self, state: str, detail: str) -> CapabilityState:
        """Re-verdict a recorded proof against what the node advertises *now*."""
        return replace(self, state=state, detail=sanitize_detail(detail))

    def to_dict(self) -> dict:
        return {
            "state": self.state,
            "detail": self.detail,
            "model_id": self.model_id,
            "shard_start": self.shard_start,
            "shard_end": self.shard_end,
            "recipe_id": self.recipe_id,
            "recipe_version": self.recipe_version,
            "catalogue_version": self.catalogue_version,
            "backend_id": self.backend_id,
            "device": self.device,
            "quantization": self.quantization,
            "validated_at": self.validated_at,
            "recorded_at": self.recorded_at,
            "schema_version": self.schema_version,
            "diagnostics": list(self.diagnostics),
            "model_artifact_digest": self.model_artifact_digest,
            "runtime_recipe_digest": self.runtime_recipe_digest,
            "shard_binding_digest": self.shard_binding_digest,
            "certification": self.certification,
        }


def absent_state(detail: str = "", *, now: float | None = None) -> CapabilityState:
    """The verdict for a node that presented no proof at all (legacy node)."""
    return CapabilityState(
        state=STATE_ABSENT,
        detail=sanitize_detail(
            detail
            or "node registered without a capability report; it predates the "
            "capability protocol or ran with admission disabled"
        ),
        recorded_at=time.time() if now is None else now,
    )


def evaluate_report(
    report: Any,
    *,
    model_matches: Callable[[str], bool],
    advertised_model: str | None,
    shard_start: int | None,
    shard_end: int | None,
    declared_recipe_id: str | None = None,
    declared_recipe_version: str | None = None,
    now: float | None = None,
    max_age_seconds: float = DEFAULT_MAX_REPORT_AGE_SECONDS,
    ledger: CertificationLedger | None = None,
) -> CapabilityState:
    """Judge the proof a node presented against what that node is advertising.

    `model_matches` is the tracker's own alias-aware comparison against the
    node's registered model / hf_repo, so an opaque model id never has to be
    parsed here.

    Returns a verdict for *every* input, including malformed ones: a bad proof is
    recorded and shown to the operator rather than dropped, so "why is my node not
    routing" has an answer in the network map.
    """
    now = time.time() if now is None else now

    if report is None:
        return absent_state(now=now)

    if not isinstance(report, Mapping):
        return CapabilityState(
            state=STATE_INVALID,
            detail=sanitize_detail(
                f"capability_report must be a JSON object, got "
                f"{type(report).__name__}"
            ),
            recorded_at=now,
        )

    try:
        parsed = _parse_report(report)
    except _ReportError as exc:
        return CapabilityState(
            state=STATE_INVALID,
            detail=sanitize_detail(str(exc)),
            recorded_at=now,
            schema_version=_maybe_int(report.get("schema_version")),
        )

    status = parsed.pop("_status")
    base = CapabilityState(state=STATE_ADMITTED, recorded_at=now, **parsed)

    if base.schema_version != SUPPORTED_SCHEMA_VERSION:
        return base.with_state(
            STATE_INVALID,
            f"capability report declares schema version {base.schema_version}; "
            f"this tracker reads version {SUPPORTED_SCHEMA_VERSION}",
        )

    if not catalogue_is_compatible(base.catalogue_version):
        return base.with_state(
            STATE_CATALOGUE_INCOMPATIBLE,
            f"recipe catalogue {base.catalogue_version!r} is older than the "
            f"minimum this tracker trusts ({MIN_CATALOGUE_VERSION}); upgrade the node",
        )

    if not model_matches(base.model_id or ""):
        return base.with_state(
            STATE_MODEL_MISMATCH,
            f"proof is for model {base.model_id!r}, but the node registered "
            f"{advertised_model!r}",
        )

    if shard_start is not None and shard_end is not None:
        if (base.shard_start, base.shard_end) != (shard_start, shard_end):
            return base.with_state(
                STATE_SHARD_MISMATCH,
                f"proof is for layers {base.shard_start}–{base.shard_end}, but the "
                f"node registered layers {shard_start}–{shard_end}",
            )

    if declared_recipe_id is not None and base.recipe_id != declared_recipe_id:
        return base.with_state(
            STATE_RECIPE_MISMATCH,
            f"proof is for recipe {base.recipe_id!r}, but the node declared it "
            f"serves with {declared_recipe_id!r}",
        )
    if (
        declared_recipe_version is not None
        and base.recipe_version != declared_recipe_version
    ):
        return base.with_state(
            STATE_RECIPE_MISMATCH,
            f"proof is for recipe {base.recipe_id!r} v{base.recipe_version}, but "
            f"the node declared v{declared_recipe_version}",
        )

    identity = None
    if report.get("identity") is not None:
        try:
            identity = parse_identity(report["identity"])
        except FingerprintMismatch as exc:
            return base.with_state(STATE_FINGERPRINT_MISMATCH, str(exc))
        except RecipeIdentityError as exc:
            return base.with_state(
                STATE_INVALID, f"capability identity block is unusable: {exc}"
            )

        # The report's `shard` range is inclusive/inclusive (the CLI and backend
        # convention); the identity's is inclusive/exclusive (the protocol's, and
        # ADR-0012's). A node whose two halves disagree has not proven the range
        # it claims, whichever one is right.
        if (identity.shard_start, identity.shard_end) != (
            base.shard_start,
            (base.shard_end or 0) + 1,
        ):
            return base.with_state(
                STATE_SHARD_MISMATCH,
                f"identity covers layers {identity.shard_start}–{identity.shard_end} "
                f"(end-exclusive), but the proof is for layers {base.shard_start}–"
                f"{base.shard_end} (end-inclusive)",
            )

        if not model_matches(identity.artifact_id):
            return base.with_state(
                STATE_MODEL_MISMATCH,
                f"identity is for artifact {identity.artifact_id!r}, but the node "
                f"registered {advertised_model!r}",
            )
        model_claim = report["model"]
        if model_claim.get("revision") != identity.revision:
            return base.with_state(
                STATE_MODEL_MISMATCH,
                "identity revision does not match the capability proof",
            )
        config_fingerprint = model_claim.get("config_fingerprint")
        if isinstance(config_fingerprint, str) and config_fingerprint.startswith(
            "sha256:"
        ):
            config_fingerprint = config_fingerprint.removeprefix("sha256:")
        if config_fingerprint != identity.architecture_digest:
            return base.with_state(
                STATE_MODEL_MISMATCH,
                "identity architecture/config digest does not match the capability proof",
            )
        if (
            identity.recipe_id != base.recipe_id
            or identity.recipe_version != base.recipe_version
            or identity.catalogue_version != base.catalogue_version
        ):
            return base.with_state(
                STATE_RECIPE_MISMATCH,
                "identity recipe labels do not match the capability proof",
            )
        if identity.axes["backend_id"] != base.backend_id:
            return base.with_state(
                STATE_RECIPE_MISMATCH,
                "identity backend does not match the capability proof",
            )
        if (
            base.quantization is not None
            and identity.axes["weight_quantization"] != base.quantization
        ):
            return base.with_state(
                STATE_RECIPE_MISMATCH,
                "identity weight quantization does not match the capability proof",
            )

        base = replace(
            base,
            model_artifact_digest=identity.model_artifact_digest,
            runtime_recipe_digest=identity.runtime_recipe_digest,
            shard_binding_digest=identity.shard_binding_digest,
        )

    if status != STATUS_PASSED:
        return base.with_state(
            STATE_FAILED,
            f"capability validation {status} on the node"
            + (f" — {' '.join(base.diagnostics)}" if base.diagnostics else ""),
        )

    age = now - (base.validated_at or 0.0)
    if age > max_age_seconds:
        return base.with_state(
            STATE_STALE,
            f"proof is {age / 60:.0f} min old (limit {max_age_seconds / 60:.0f} min); "
            "the node must re-validate before it can be routed",
        )
    if age < -MAX_CLOCK_SKEW_SECONDS:
        return base.with_state(
            STATE_STALE,
            f"proof is timestamped {-age:.0f}s in the future; check the node's clock",
        )

    # The digest only establishes that this report is self-consistent. It does
    # not authenticate a node or prove a distributed forward. Exact recipes are
    # therefore registered-but-dark until tracker-owned certification records
    # that forward.
    #
    # `ledger is None` means the caller owns no certification authority. It is
    # not "certify anything" and it is not "make one up": a disposable ledger
    # would register the recipe into state that is discarded on return, which
    # reads like certification is wired when nothing is recording it. The only
    # safe reading is that nothing here has certified this recipe, so it stays
    # dark. `TrackerServer` owns the real ledger and passes it in.
    if identity is not None:
        if ledger is None:
            return base.with_state(
                STATE_UNCERTIFIED,
                "no certification ledger; an exact recipe is dark until a "
                "tracker-owned distributed forward certifies it",
            )
        recipe_status = ledger.register(identity)
        base = replace(base, certification=recipe_status.status)
        if not recipe_status.may_serve:
            return base.with_state(STATE_UNCERTIFIED, recipe_status.detail)

    return base.with_state(
        STATE_ADMITTED,
        f"{base.model_id} layers {base.shard_start}–{base.shard_end} proven on "
        f"{base.device} with recipe {base.recipe_id} (v{base.recipe_version})",
    )


class _ReportError(ValueError):
    """Malformed report input. Messages name the field, never echo a payload."""


def _parse_report(doc: Mapping[str, Any]) -> dict:
    model = _object(doc.get("model"), "model")
    shard = _object(doc.get("shard"), "shard")
    recipe = _object(doc.get("recipe"), "recipe")
    backend = _object(doc.get("backend"), "backend")

    validated_at = doc.get("validated_at")
    if isinstance(validated_at, bool) or not isinstance(validated_at, (int, float)):
        raise _ReportError("'validated_at' must be a Unix timestamp")

    schema_version = doc.get("schema_version")
    if isinstance(schema_version, bool) or not isinstance(schema_version, int):
        raise _ReportError("'schema_version' must be an integer")

    return {
        "model_id": _text(model.get("model_id"), "model.model_id"),
        "shard_start": _index(shard.get("start"), "shard.start"),
        "shard_end": _index(shard.get("end"), "shard.end"),
        "recipe_id": _text(recipe.get("recipe_id"), "recipe.recipe_id"),
        "recipe_version": _text(recipe.get("recipe_version"), "recipe.recipe_version"),
        "catalogue_version": _text(
            recipe.get("catalogue_version"), "recipe.catalogue_version"
        ),
        "backend_id": _text(backend.get("backend_id"), "backend.backend_id"),
        "device": _text(backend.get("device"), "backend.device"),
        "quantization": _optional_text(
            backend.get("quantization"), "backend.quantization"
        ),
        "validated_at": float(validated_at),
        "schema_version": schema_version,
        "diagnostics": _diagnostics(doc.get("diagnostics")),
        "_status": _text(doc.get("status"), "status"),
    }


def _object(value: Any, field_name: str) -> Mapping[str, Any]:
    if not isinstance(value, Mapping):
        raise _ReportError(f"{field_name!r} must be a JSON object")
    return value


def _text(value: Any, field_name: str) -> str:
    if not isinstance(value, str) or not value.strip():
        raise _ReportError(f"{field_name!r} must be a non-empty string")
    return value


def _optional_text(value: Any, field_name: str) -> str | None:
    if value is None:
        return None
    return _text(value, field_name)


def _index(value: Any, field_name: str) -> int:
    if isinstance(value, bool) or not isinstance(value, int) or value < 0:
        raise _ReportError(f"{field_name!r} must be a non-negative integer")
    return value


def _maybe_int(value: Any) -> int | None:
    if isinstance(value, bool) or not isinstance(value, int):
        return None
    return value


def _diagnostics(value: Any) -> tuple[str, ...]:
    if not isinstance(value, list):
        return ()
    out = [
        sanitize_detail(item)
        for item in value[:_MAX_DIAGNOSTICS]
        if isinstance(item, str) and item.strip()
    ]
    return tuple(out)