"""Closed, secret-safe connectivity preflight contracts for benchmark cells.

This module deliberately has no caller, provider, or network dependency.  Caller
adapters provide typed observations, while this boundary proves that those
observations exactly match one immutable manifest cell before they can be stored.
A blocked caller may report that it observed no effective binding at all, but it
can never report a partial or manifest-derived synthetic one.
"""

from __future__ import annotations

import json
import os
import re
import stat
from dataclasses import dataclass
from pathlib import Path
from typing import Any, Protocol

from scripts.agent_benchmark.manifest import (
    CALLER_ENUM,
    ROUTE_KIND_ENUM,
    STAGE_ENUM,
    STAGE_RANK,
    ExpectedBinding,
    MatrixCell,
    TOKEN_RE,
)


SCHEMA_VERSION = "1"
IDENTITY_RE = re.compile(r"^sha256:[0-9a-f]{64}$")
MAX_EVIDENCE_BYTES = 65536

# Canonical issue order: registration blockers first, then implementation gaps.
ISSUE_CODE_ORDER = (
    "credential_missing",
    "model_missing",
    "route_missing",
    "effort_unsupported",
    "endpoint_incompatible",
    "auth_incompatible",
    "protocol_incompatible",
    "stream_incompatible",
)
REGISTRATION_ISSUE_CODES = frozenset(ISSUE_CODE_ORDER[:4])
IMPLEMENTATION_ISSUE_CODES = frozenset(ISSUE_CODE_ORDER[4:])
ISSUE_CODES = REGISTRATION_ISSUE_CODES | IMPLEMENTATION_ISSUE_CODES
ISSUE_RANK: dict[str, int] = {code: rank for rank, code in enumerate(ISSUE_CODE_ORDER)}
# The only resume vocabulary; callers can never attach their own text.
ISSUE_RESUME_CODES: dict[str, str] = {
    "credential_missing": "register_credential",
    "model_missing": "register_model",
    "route_missing": "register_route",
    "effort_unsupported": "register_effort_support",
    "endpoint_incompatible": "implement_endpoint_adapter",
    "auth_incompatible": "implement_auth_adapter",
    "protocol_incompatible": "implement_protocol_adapter",
    "stream_incompatible": "implement_stream_adapter",
}
RESUME_CODES = frozenset(ISSUE_RESUME_CODES.values())
RESULT_STATUSES = ("ready", "registration_required", "implementation_gap")
EFFECTIVE_FIELDS = (
    "effective_route_kind",
    "effective_route_id",
    "effective_model",
    "effective_effort",
)


class ConnectivityError(Exception):
    """Base class for closed connectivity contract failures."""


class ConnectivityValidationError(ConnectivityError):
    """Raised when a caller observation or evidence value is not admissible."""


class ConnectivityEvidenceError(ConnectivityError):
    """Raised when durable evidence cannot be safely written or read."""


@dataclass(frozen=True)
class CallerCapability:
    """Static public capability claimed by one benchmark caller adapter."""

    caller: str
    route_kinds: tuple[str, ...]
    efforts: tuple[str, ...]


@dataclass(frozen=True)
class EffectiveBinding:
    """One observed stage binding, including its exact effective effort."""

    stage: str
    model: str
    effort: str | None = None


@dataclass(frozen=True)
class RequestedEffectiveBinding:
    """Requested identity plus one all-or-none observed effective binding.

    The requested fields are always required.  The effective group is either
    fully absent, which only a blocked result may report, or complete and exact.
    """

    cell_id: str
    caller: str
    requested_route_kind: str
    requested_route_id: str
    requested_model: str
    requested_effort: str
    effective_route_kind: str | None = None
    effective_route_id: str | None = None
    effective_model: str | None = None
    effective_effort: str | None = None
    effective_bindings: tuple[EffectiveBinding, ...] = ()


@dataclass(frozen=True)
class ConnectivityIssue:
    """Closed blocker code paired with its one permitted resume code."""

    code: str
    resume_code: str


@dataclass(frozen=True)
class ConnectivityResult:
    """Validated immutable result for exactly one caller/cell preflight."""

    capability: CallerCapability
    binding: RequestedEffectiveBinding
    issues: tuple[ConnectivityIssue, ...]
    status: str


class CallerPreflight(Protocol):
    """Adapter boundary; implementations must not return raw caller output."""

    capability: CallerCapability

    def preflight(
        self, cell: MatrixCell
    ) -> tuple[RequestedEffectiveBinding, tuple[ConnectivityIssue, ...]]:
        """Return typed, secret-free observation for ``cell``."""


def _fail(message: str) -> None:
    # Messages contain only fixed contract field names, never input values.
    raise ConnectivityValidationError(message)


def _require_identifier(value: Any, field_name: str) -> str:
    if not isinstance(value, str) or not TOKEN_RE.fullmatch(value):
        _fail(f"invalid {field_name}")
    return value


def _require_identity(value: Any, field_name: str) -> str:
    if not isinstance(value, str) or not IDENTITY_RE.fullmatch(value):
        raise ConnectivityEvidenceError(f"invalid {field_name}")
    return value


def _validate_closed_tuple(
    values: Any,
    field_name: str,
    *,
    allowed: tuple[str, ...] | None = None,
) -> None:
    """Check shape, item type, membership and uniqueness before canonical order.

    Ordering is checked last so unknown or unhashable entries can never reach
    ``set`` or an enum index lookup and escape as a built-in exception.
    """
    if not isinstance(values, tuple) or not values:
        _fail(f"invalid {field_name}")
    for item in values:
        if not isinstance(item, str):
            _fail(f"invalid {field_name} item")
        if allowed is None:
            _require_identifier(item, f"{field_name} item")
        elif item not in allowed:
            _fail(f"invalid {field_name} item")
    if len(set(values)) != len(values):
        _fail(f"duplicate {field_name} item")
    if tuple(sorted(values, key=allowed.index if allowed is not None else None)) != values:
        _fail(f"non-canonical {field_name}")


def _validate_capability(capability: CallerCapability) -> None:
    if not isinstance(capability, CallerCapability):
        _fail("invalid caller capability")
    if capability.caller not in CALLER_ENUM:
        _fail("invalid capability caller")
    _validate_closed_tuple(
        capability.route_kinds, "capability route_kinds", allowed=ROUTE_KIND_ENUM
    )
    _validate_closed_tuple(capability.efforts, "capability efforts")


def _validate_binding_shape(binding: RequestedEffectiveBinding) -> bool:
    """Validate requested fields, then the all-or-none effective group.

    Returns whether the caller reported an effective observation at all.
    """
    if not isinstance(binding, RequestedEffectiveBinding):
        _fail("invalid requested/effective binding")
    _require_identifier(binding.cell_id, "binding cell_id")
    if binding.caller not in CALLER_ENUM:
        _fail("invalid binding caller")
    if binding.requested_route_kind not in ROUTE_KIND_ENUM:
        _fail("invalid requested route_kind")
    for name in ("requested_route_id", "requested_model", "requested_effort"):
        _require_identifier(getattr(binding, name), f"binding {name}")

    if not isinstance(binding.effective_bindings, tuple):
        _fail("invalid effective_bindings")
    observed = tuple(getattr(binding, name) is not None for name in EFFECTIVE_FIELDS)
    if any(observed) != all(observed):
        _fail("partial effective observation")
    if not any(observed):
        if binding.effective_bindings:
            _fail("partial effective observation")
        return False

    if binding.effective_route_kind not in ROUTE_KIND_ENUM:
        _fail("invalid effective route_kind")
    for name in ("effective_route_id", "effective_model", "effective_effort"):
        _require_identifier(getattr(binding, name), f"binding {name}")
    if not binding.effective_bindings:
        _fail("partial effective observation")
    stages: list[str] = []
    for item in binding.effective_bindings:
        if not isinstance(item, EffectiveBinding):
            _fail("invalid effective binding item")
        if item.stage not in STAGE_ENUM:
            _fail("invalid effective binding stage")
        _require_identifier(item.model, "effective binding model")
        if item.effort is not None:
            _require_identifier(item.effort, "effective binding effort")
        stages.append(item.stage)
    if len(set(stages)) != len(stages):
        _fail("duplicate effective binding stage")
    if stages != sorted(stages, key=STAGE_RANK.__getitem__):
        _fail("non-canonical effective binding order")
    return True


def _expected_as_effective(expected: ExpectedBinding) -> EffectiveBinding:
    return EffectiveBinding(expected.stage, expected.model, expected.effort)


def _cell_binding(cell: MatrixCell) -> tuple[str, str, str, str]:
    return (
        cell.iop.route_kind,
        cell.iop.route_id,
        cell.iop.request_model,
        cell.iop.requested_effort,
    )


def validate_requested_binding(
    cell: MatrixCell,
    capability: CallerCapability,
    binding: RequestedEffectiveBinding,
) -> None:
    """Fail closed unless caller identity and requested route/model/effort are exact."""
    if not isinstance(cell, MatrixCell):
        _fail("invalid matrix cell")
    _validate_capability(capability)
    _validate_binding_shape(binding)

    if capability.caller != cell.caller or binding.caller != cell.caller:
        _fail("caller identity mismatch")
    if binding.cell_id != cell.id:
        _fail("cell identity mismatch")
    if cell.iop.route_kind not in capability.route_kinds:
        _fail("unsupported route_kind")
    if cell.iop.requested_effort not in capability.efforts:
        _fail("unsupported effort")

    requested = (
        binding.requested_route_kind,
        binding.requested_route_id,
        binding.requested_model,
        binding.requested_effort,
    )
    if requested != _cell_binding(cell):
        _fail("requested binding mismatch")


def validate_effective_binding(
    cell: MatrixCell,
    binding: RequestedEffectiveBinding,
    *,
    required: bool,
) -> None:
    """Require the observation only for ``ready``; any present group must be exact."""
    if not isinstance(cell, MatrixCell):
        _fail("invalid matrix cell")
    if not _validate_binding_shape(binding):
        if required:
            _fail("missing effective observation")
        return

    effective = (
        binding.effective_route_kind,
        binding.effective_route_id,
        binding.effective_model,
        binding.effective_effort,
    )
    if effective != _cell_binding(cell):
        _fail("effective binding substitution")
    expected_stages = tuple(_expected_as_effective(item) for item in cell.iop.expected_bindings)
    if binding.effective_bindings != expected_stages:
        _fail("effective stage binding mismatch")


def validate_binding(
    cell: MatrixCell,
    capability: CallerCapability,
    binding: RequestedEffectiveBinding,
    *,
    require_effective: bool = True,
) -> None:
    """Validate the requested phase and then the effective phase of one binding."""
    validate_requested_binding(cell, capability, binding)
    validate_effective_binding(cell, binding, required=require_effective)


def _validate_issue(issue: ConnectivityIssue) -> None:
    if not isinstance(issue, ConnectivityIssue):
        _fail("invalid issue")
    if not isinstance(issue.code, str) or issue.code not in ISSUE_CODES:
        _fail("invalid issue code")
    if not isinstance(issue.resume_code, str) or issue.resume_code != ISSUE_RESUME_CODES[issue.code]:
        _fail("invalid issue resume_code")


def classify_issues(issues: tuple[ConnectivityIssue, ...]) -> str:
    """Return the only permitted status, with implementation gaps taking precedence."""
    if not isinstance(issues, tuple):
        _fail("issues must be a tuple")
    codes: list[str] = []
    for issue in issues:
        _validate_issue(issue)
        codes.append(issue.code)
    if len(set(codes)) != len(codes):
        _fail("duplicate issue code")
    if codes != sorted(codes, key=ISSUE_RANK.__getitem__):
        _fail("non-canonical issue order")
    if any(code in IMPLEMENTATION_ISSUE_CODES for code in codes):
        return "implementation_gap"
    if any(code in REGISTRATION_ISSUE_CODES for code in codes):
        return "registration_required"
    return "ready"


def make_result(
    cell: MatrixCell,
    capability: CallerCapability,
    binding: RequestedEffectiveBinding,
    issues: tuple[ConnectivityIssue, ...] = (),
) -> ConnectivityResult:
    """Construct a result only after phase-exact no-substitution validation."""
    status = classify_issues(issues)
    validate_requested_binding(cell, capability, binding)
    validate_effective_binding(cell, binding, required=status == "ready")
    return ConnectivityResult(capability, binding, issues, status)


def validate_result(cell: MatrixCell, result: ConnectivityResult) -> None:
    """Revalidate a received result before it is consumed or persisted."""
    if not isinstance(result, ConnectivityResult):
        _fail("invalid connectivity result")
    status = classify_issues(result.issues)
    if result.status not in RESULT_STATUSES or result.status != status:
        _fail("result status mismatch")
    validate_requested_binding(cell, result.capability, result.binding)
    validate_effective_binding(cell, result.binding, required=status == "ready")


def _binding_dict(binding: RequestedEffectiveBinding) -> dict[str, Any]:
    return {
        "cell_id": binding.cell_id,
        "caller": binding.caller,
        "requested_route_kind": binding.requested_route_kind,
        "requested_route_id": binding.requested_route_id,
        "requested_model": binding.requested_model,
        "requested_effort": binding.requested_effort,
        "effective_route_kind": binding.effective_route_kind,
        "effective_route_id": binding.effective_route_id,
        "effective_model": binding.effective_model,
        "effective_effort": binding.effective_effort,
        "effective_bindings": [
            {"stage": item.stage, "model": item.model, "effort": item.effort}
            for item in binding.effective_bindings
        ],
    }


def canonical_evidence_bytes(
    cell: MatrixCell,
    result: ConnectivityResult,
    endpoint_identity: str,
    config_identity: str,
) -> bytes:
    """Return deterministic schema-closed evidence containing no raw endpoint data."""
    validate_result(cell, result)
    _require_identity(endpoint_identity, "endpoint_identity")
    _require_identity(config_identity, "config_identity")
    payload = {
        "schema_version": SCHEMA_VERSION,
        "cell": {"id": cell.id, "caller": cell.caller},
        "status": result.status,
        "binding": _binding_dict(result.binding),
        "issues": [
            {"code": issue.code, "resume_code": issue.resume_code}
            for issue in result.issues
        ],
        "endpoint_identity": endpoint_identity,
        "config_identity": config_identity,
    }
    return json.dumps(payload, sort_keys=True, separators=(",", ":"), ensure_ascii=True).encode("ascii") + b"\n"


def _safe_relative_path(relative_path: str | Path) -> Path:
    if not isinstance(relative_path, (str, Path)):
        raise ConnectivityEvidenceError("invalid evidence relative path")
    path = Path(relative_path)
    if (
        path.is_absolute()
        or not path.parts
        or any(part in ("", ".", "..") for part in path.parts)
        or path.suffix != ".json"
    ):
        raise ConnectivityEvidenceError("invalid evidence relative path")
    return path


def _require_nofollow_support() -> None:
    if (
        not hasattr(os, "O_NOFOLLOW")
        or not hasattr(os, "O_DIRECTORY")
        or os.open not in os.supports_dir_fd
        or os.mkdir not in os.supports_dir_fd
    ):
        raise ConnectivityEvidenceError("evidence no-follow traversal unsupported")


def _open_directory(name: str, dir_fd: int | None, message: str) -> int:
    flags = os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW
    try:
        return os.open(name, flags, dir_fd=dir_fd)
    except OSError as exc:
        raise ConnectivityEvidenceError(message) from exc


def _open_root(root: str | Path) -> int:
    """Open the evidence root by descending one no-follow component at a time."""
    if not isinstance(root, (str, Path)):
        raise ConnectivityEvidenceError("invalid evidence root")
    path = Path(root)
    parts = list(path.parts)
    if path.is_absolute():
        descriptor = _open_directory(parts[0], None, "invalid evidence root")
        parts = parts[1:]
    else:
        descriptor = _open_directory(".", None, "invalid evidence root")
    try:
        for part in parts:
            if part in ("", ".", ".."):
                raise ConnectivityEvidenceError("invalid evidence root")
            child = _open_directory(part, descriptor, "invalid evidence root")
            os.close(descriptor)
            descriptor = child
    except BaseException:
        os.close(descriptor)
        raise
    return descriptor


def _open_evidence_parent(
    root: str | Path, relative_path: str | Path, *, create: bool
) -> tuple[int, str]:
    """Return a descriptor for the verified parent directory and the final name."""
    relative = _safe_relative_path(relative_path)
    _require_nofollow_support()
    descriptor = _open_root(root)
    try:
        for part in relative.parts[:-1]:
            if create:
                try:
                    os.mkdir(part, 0o700, dir_fd=descriptor)
                except FileExistsError:
                    pass
                except OSError as exc:
                    raise ConnectivityEvidenceError("unsafe evidence parent") from exc
            child = _open_directory(part, descriptor, "unsafe evidence parent")
            os.close(descriptor)
            descriptor = child
    except BaseException:
        os.close(descriptor)
        raise
    return descriptor, relative.parts[-1]


def write_evidence(
    root: str | Path,
    relative_path: str | Path,
    cell: MatrixCell,
    result: ConnectivityResult,
    endpoint_identity: str,
    config_identity: str,
) -> None:
    """Atomically create evidence once; any existing or symlinked target is rejected."""
    payload = canonical_evidence_bytes(cell, result, endpoint_identity, config_identity)
    parent, name = _open_evidence_parent(root, relative_path, create=True)
    try:
        flags = os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW
        try:
            descriptor = os.open(name, flags, 0o600, dir_fd=parent)
        except FileExistsError as exc:
            raise ConnectivityEvidenceError("evidence target already exists") from exc
        except OSError as exc:
            raise ConnectivityEvidenceError("evidence write rejected") from exc
        try:
            with os.fdopen(descriptor, "wb") as stream:
                stream.write(payload)
                stream.flush()
                os.fsync(stream.fileno())
        except OSError as exc:
            raise ConnectivityEvidenceError("evidence write rejected") from exc
    finally:
        os.close(parent)


def _read_bounded(descriptor: int) -> bytes:
    """Read at most one byte past the cap so oversized input fails closed."""
    chunks: list[bytes] = []
    remaining = MAX_EVIDENCE_BYTES + 1
    try:
        while remaining > 0:
            chunk = os.read(descriptor, remaining)
            if not chunk:
                break
            chunks.append(chunk)
            remaining -= len(chunk)
    except OSError as exc:
        raise ConnectivityEvidenceError("invalid evidence bytes") from exc
    raw = b"".join(chunks)
    if len(raw) > MAX_EVIDENCE_BYTES:
        raise ConnectivityEvidenceError("evidence target too large")
    return raw


def _read_evidence_bytes(root: str | Path, relative_path: str | Path) -> bytes:
    parent, name = _open_evidence_parent(root, relative_path, create=False)
    try:
        try:
            descriptor = os.open(
                name, os.O_RDONLY | os.O_NOFOLLOW | os.O_NONBLOCK, dir_fd=parent
            )
        except OSError as exc:
            raise ConnectivityEvidenceError("unsafe evidence target") from exc
        try:
            info = os.fstat(descriptor)
            if not stat.S_ISREG(info.st_mode):
                raise ConnectivityEvidenceError("unsafe evidence target")
            if info.st_size > MAX_EVIDENCE_BYTES:
                raise ConnectivityEvidenceError("evidence target too large")
            return _read_bounded(descriptor)
        finally:
            os.close(descriptor)
    finally:
        os.close(parent)


def _no_duplicate_object(pairs: list[tuple[str, Any]]) -> dict[str, Any]:
    result: dict[str, Any] = {}
    for key, value in pairs:
        if key in result:
            raise ConnectivityEvidenceError("duplicate evidence field")
        result[key] = value
    return result


def _read_binding(data: Any) -> RequestedEffectiveBinding:
    if not isinstance(data, dict) or set(data) != {
        "cell_id", "caller", "requested_route_kind", "requested_route_id",
        "requested_model", "requested_effort", "effective_route_kind",
        "effective_route_id", "effective_model", "effective_effort", "effective_bindings",
    }:
        raise ConnectivityEvidenceError("invalid evidence binding")
    stages = data["effective_bindings"]
    if not isinstance(stages, list):
        raise ConnectivityEvidenceError("invalid evidence binding")
    bindings: list[EffectiveBinding] = []
    for item in stages:
        if not isinstance(item, dict) or set(item) != {"stage", "model", "effort"}:
            raise ConnectivityEvidenceError("invalid evidence binding")
        bindings.append(EffectiveBinding(item["stage"], item["model"], item["effort"]))
    try:
        binding = RequestedEffectiveBinding(
            data["cell_id"], data["caller"], data["requested_route_kind"],
            data["requested_route_id"], data["requested_model"], data["requested_effort"],
            data["effective_route_kind"], data["effective_route_id"],
            data["effective_model"], data["effective_effort"], tuple(bindings),
        )
        _validate_binding_shape(binding)
    except ConnectivityValidationError as exc:
        raise ConnectivityEvidenceError("invalid evidence binding") from exc
    return binding


def _validate_evidence_payload(payload: dict[str, Any], cell: MatrixCell) -> None:
    if not isinstance(cell, MatrixCell):
        raise ConnectivityEvidenceError("invalid matrix cell")
    if not isinstance(payload.get("cell"), dict) or set(payload["cell"]) != {"id", "caller"}:
        raise ConnectivityEvidenceError("invalid evidence cell")
    binding = _read_binding(payload["binding"])
    if payload["cell"] != {"id": binding.cell_id, "caller": binding.caller}:
        raise ConnectivityEvidenceError("evidence cell mismatch")
    if binding.cell_id != cell.id or binding.caller != cell.caller:
        raise ConnectivityEvidenceError("evidence cell mismatch")
    requested = (
        binding.requested_route_kind,
        binding.requested_route_id,
        binding.requested_model,
        binding.requested_effort,
    )
    if requested != _cell_binding(cell):
        raise ConnectivityEvidenceError("requested binding mismatch")
    try:
        validate_effective_binding(cell, binding, required=payload["status"] == "ready")
    except ConnectivityValidationError as exc:
        raise ConnectivityEvidenceError("invalid evidence binding") from exc
    issues_raw = payload["issues"]
    if not isinstance(issues_raw, list):
        raise ConnectivityEvidenceError("invalid evidence issues")
    try:
        issues = tuple(
            ConnectivityIssue(item["code"], item["resume_code"])
            for item in issues_raw
            if isinstance(item, dict) and set(item) == {"code", "resume_code"}
        )
        if len(issues) != len(issues_raw) or classify_issues(issues) != payload["status"]:
            raise ConnectivityValidationError("result status mismatch")
    except (ConnectivityValidationError, KeyError, TypeError) as exc:
        raise ConnectivityEvidenceError("invalid evidence issues") from exc


def read_evidence(
    root: str | Path, relative_path: str | Path, cell: MatrixCell
) -> dict[str, Any]:
    """Read only canonical, schema-closed evidence; corruption fails closed."""
    raw = _read_evidence_bytes(root, relative_path)
    try:
        parsed = json.loads(raw.decode("ascii"), object_pairs_hook=_no_duplicate_object)
    except (UnicodeDecodeError, json.JSONDecodeError) as exc:
        raise ConnectivityEvidenceError("invalid evidence bytes") from exc
    if not isinstance(parsed, dict) or set(parsed) != {
        "schema_version", "cell", "status", "binding", "issues",
        "endpoint_identity", "config_identity",
    }:
        raise ConnectivityEvidenceError("invalid evidence schema")
    if parsed.get("schema_version") != SCHEMA_VERSION or parsed.get("status") not in RESULT_STATUSES:
        raise ConnectivityEvidenceError("invalid evidence schema")
    _require_identity(parsed.get("endpoint_identity"), "endpoint_identity")
    _require_identity(parsed.get("config_identity"), "config_identity")
    _validate_evidence_payload(parsed, cell)
    canonical = json.dumps(parsed, sort_keys=True, separators=(",", ":"), ensure_ascii=True).encode("ascii") + b"\n"
    if raw != canonical:
        raise ConnectivityEvidenceError("non-canonical evidence")
    return parsed