# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/where-must-an-instance-of-a-capability-registry-refuse-to-serve-a-match-when-semantic-similarity-and-contract-version-compatibility-disagree.md
# Remnant Warrant:    https://corpusfetch.com/c/where-must-an-instance-of-a-capability-registry-refuse-to-serve-a-match-when-semantic-similarity-and-contract-version-compatibility-disagree.warrant.json
# Markdown SHA-256:   b04faba679ef0ce8c2110e25c4ce14f808b76a09ab3da4d9e5a1844575a50cb4
# Code AST SHA-256:   d766b0a8a088dfe3d220dd6f94a1a6f845642f34a8b6855a64acc3e5c2389afb
# Verified Classes:   CapabilityResolutionError, NoCompatibleContractError, LowConfidenceMatchError, ContractVersion, Candidate
# Defensive Guards:   3 fail-closed invariants (zero external deps)

from __future__ import annotations

from dataclasses import dataclass
from typing import Iterable, Mapping, Sequence


class CapabilityResolutionError(RuntimeError):
    """Base class for capability-registry resolution failures."""


class NoCompatibleContractError(CapabilityResolutionError):
    """No registered tool satisfies the caller's version constraint.

    Raised instead of falling back to the nearest semantic match, because a
    cross-major nearest match is exactly the misroute this registry forbids.
    """


class LowConfidenceMatchError(CapabilityResolutionError):
    """Compatible candidates exist, but none clears the semantic floor tau."""


@dataclass(frozen=True)
class ContractVersion:
    major: int
    minor: int
    patch: int

    def __str__(self) -> str:
        return f"{self.major}.{self.minor}.{self.patch}"


@dataclass(frozen=True)
class Candidate:
    capability_family: str
    contract: ContractVersion
    description_variant: str
    semantic_score: float


def _is_compatible(
    candidate: ContractVersion,
    constraint: tuple[int, int, int],
) -> bool:
    """True iff candidate lies within the caller's [min, max) major/minor range."""
    lo_major, lo_minor, hi_major = constraint
    if candidate.major < lo_major or candidate.major > hi_major:
        return False
    if candidate.major == lo_major and candidate.minor < lo_minor:
        return False
    return True


def resolve_capability(
    candidates: Sequence[Candidate],
    constraint: tuple[int, int, int],
    *,
    tau: float = 0.62,
    shadow_delta: float = 0.05,
) -> Candidate:
    """Select a compatible candidate, ranking by semantics only within the
    compatibility class. Never crosses a compatibility boundary.
    """
    compatible: list[Candidate] = [
        c for c in candidates if _is_compatible(c.contract, constraint)
    ]
    if not compatible:
        raise NoCompatibleContractError(
            f"no registered contract satisfies constraint {constraint}; "
            f"refusing to fall back to {len(candidates)} incompatible candidates"
        )

    # Deduplicate A/B description variants: one slot per (family, major).
    seen: set[tuple[str, int]] = set()
    deduped: list[Candidate] = []
    for c in sorted(compatible, key=lambda c: c.semantic_score, reverse=True):
        key = (c.capability_family, c.contract.major)
        if key in seen:
            continue
        seen.add(key)
        deduped.append(c)

    best = deduped[0]
    if best.semantic_score < tau:
        raise LowConfidenceMatchError(
            f"best compatible score {best.semantic_score:.3f} < tau {tau}; "
            f"capability {best.capability_family} @ {best.contract}"
        )

    incompatible_best = max(
        (c for c in candidates if not _is_compatible(c.contract, constraint)),
        key=lambda c: c.semantic_score,
        default=None,
    )
    if (
        incompatible_best is not None
        and incompatible_best.semantic_score - best.semantic_score > shadow_delta
    ):
        # Advisory only: emit telemetry, do not change the selection.
        _emit_shadowed_match(best, incompatible_best)

    return best


def _emit_shadowed_match(compatible: Candidate, shadowing: Candidate) -> None:
    # Placeholder for structured telemetry; intentionally non-fatal.
    _ = (compatible, shadowing)


def resolve_or_refuse(
    candidates: Iterable[Mapping[str, object]],
    constraint: tuple[int, int, int],
) -> Candidate:
    """Adapter that normalizes raw registry rows and delegates to the resolver."""
    normalized = [
        Candidate(
            capability_family=str(row["family"]),
            contract=ContractVersion(*map(int, str(row["contract"]).split("."))),
            description_variant=str(row.get("variant", "default")),
            semantic_score=float(row["score"]),
        )
        for row in candidates
    ]
    if not normalized:
        raise NoCompatibleContractError("registry returned zero candidates")
    return resolve_capability(normalized, constraint)