"""Explicit, secret-safe live IOP boundary for benchmark callers.

Only the six ``IOP_BENCH_<CALLER>_{BASE_URL,SECRET_ENV}`` values select this
boundary.  The referenced secret is retained in a private runtime object and
is never included in connectivity evidence, command output, or run metadata.
"""

from __future__ import annotations

import hashlib
import json
import os
import re
import shutil
import subprocess
from collections.abc import Callable, Mapping
from dataclasses import dataclass, replace
from pathlib import Path
from typing import Any
from urllib.parse import urlsplit, urlunsplit
from urllib.error import HTTPError
from urllib.request import Request, urlopen

from scripts.agent_benchmark.agy_iop import (
    AGY_CALLER,
    AgyEventParser,
    AgyRuntimeInputs,
    AgyRuntimeObservation,
    build_agy_invocation,
    inspect_agy_iop_capability,
    preflight_agy_iop,
    run_agy_invocation,
    _runtime_identity as _agy_runtime_identity,
)
from scripts.agent_benchmark.attempts import Attempt, ExecutionAdapter, PreflightObservation
from scripts.agent_benchmark.claude_iop import ClaudeIopAdapter, ClaudeIopRuntime, claude_capability
from scripts.agent_benchmark.codex_iop import (
    BASE_URL_ENV_KEY,
    SECRET_ENV_KEY,
    build_codex_invocation,
    codex_capability,
    run_codex_invocation,
    runtime_from_environment,
)
from scripts.agent_benchmark.connectivity import (
    ISSUE_CODE_ORDER,
    ISSUE_RESUME_CODES,
    CallerCapability,
    ConnectivityIssue,
    EffectiveBinding,
    RequestedEffectiveBinding,
    make_result,
)
from scripts.agent_benchmark.lifecycle import InvocationResult, run_invocation, spec_digest
from scripts.agent_benchmark.manifest import CALLER_ENUM, MatrixCell, TOKEN_RE, Timeout
from scripts.agent_benchmark.workspace import PreparedWorkspace


_ENV_NAME = re.compile(r"^[A-Za-z_][A-Za-z0-9_]{0,63}$")
_CALLERS = ("claude", "agy", "codex")
_TIMEOUT_SECONDS = 10


class LiveIopError(Exception):
    """Private boundary failure converted into one closed public issue."""

    def __init__(self, issue_code: str) -> None:
        if issue_code not in ISSUE_RESUME_CODES:
            raise ValueError("invalid live IOP issue")
        self.issue_code = issue_code
        super().__init__(issue_code)


@dataclass(frozen=True)
class _Runtime:
    caller: str
    base_url: str
    secret: str
    endpoint_identity: str
    config: "_ConfigObservation"


@dataclass(frozen=True)
class _RouteObservation:
    route_kind: str
    route_id: str
    model: str


@dataclass(frozen=True)
class _ConfigObservation:
    routes: tuple[_RouteObservation, ...]
    identity: str


@dataclass(frozen=True)
class _Observation:
    catalog_models: tuple[str, ...]
    config_identity: str
    caller_ready: bool
    agy_version: str = ""
    agy_help: str = ""


@dataclass(frozen=True)
class _RuntimeResolution:
    runtime: _Runtime | None
    issue_code: str | None


@dataclass(frozen=True)
class _InvokerSeams:
    """Test seams below registry construction; production uses real callers."""

    claude: Callable[..., InvocationResult]
    agy: Callable[..., InvocationResult]
    codex: Callable[..., Any]


def _identity(label: str, value: str) -> str:
    return "sha256:" + hashlib.sha256(
        b"iop-benchmark-live-v1\0" + label.encode("ascii") + b"\0" + value.encode("utf-8")
    ).hexdigest()


def _base_url(value: str) -> str:
    if not isinstance(value, str) or not value:
        raise LiveIopError("endpoint_incompatible")
    parsed = urlsplit(value)
    if parsed.scheme not in ("http", "https") or not parsed.netloc or parsed.query or parsed.fragment:
        raise LiveIopError("endpoint_incompatible")
    return value.rstrip("/")


def _models_url(base_url: str) -> str:
    parsed = urlsplit(base_url)
    path = parsed.path.rstrip("/")
    if path.endswith("/v1"):
        path += "/models"
    else:
        path += "/v1/models"
    return urlunsplit((parsed.scheme, parsed.netloc, path, "", ""))


def _command(argv: tuple[str, ...]) -> str:
    try:
        completed = subprocess.run(
            argv, check=False, capture_output=True, text=True, timeout=_TIMEOUT_SECONDS,
            env={"PATH": os.environ.get("PATH", "/usr/bin:/bin"), "LANG": "C.UTF-8", "LC_ALL": "C.UTF-8"},
        )
    except (OSError, subprocess.SubprocessError) as exc:
        raise LiveIopError("stream_incompatible") from exc
    if completed.returncode != 0:
        raise LiveIopError("stream_incompatible")
    return completed.stdout + completed.stderr


def _caller_binary(name: str) -> str:
    path = shutil.which(name)
    if not path or not os.path.isfile(path) or not os.access(path, os.X_OK):
        raise LiveIopError("stream_incompatible")
    return os.path.realpath(path)


def _catalog(runtime: _Runtime) -> tuple[tuple[str, ...], str]:
    request = Request(_models_url(runtime.base_url), headers={"Authorization": f"Bearer {runtime.secret}"})
    try:
        with urlopen(request, timeout=_TIMEOUT_SECONDS) as response:
            if response.status in (401, 403):
                raise LiveIopError("auth_incompatible")
            if response.status != 200:
                raise LiveIopError("endpoint_incompatible")
            payload = json.loads(response.read().decode("utf-8"))
    except HTTPError as exc:
        if exc.code in (401, 403):
            raise LiveIopError("auth_incompatible") from exc
        raise LiveIopError("endpoint_incompatible") from exc
    except LiveIopError:
        raise
    except OSError as exc:
        # Connection refusal, DNS failure, and timeout all mean that this
        # boundary cannot reach a compatible endpoint.  They are distinct
        # from a reachable endpoint with an invalid response schema.
        raise LiveIopError("endpoint_incompatible") from exc
    except ValueError as exc:
        raise LiveIopError("protocol_incompatible") from exc
    records = payload.get("data") if isinstance(payload, dict) else None
    if not isinstance(records, list):
        raise LiveIopError("protocol_incompatible")
    model_ids: list[str] = []
    for item in records:
        model_id = item.get("id") if isinstance(item, dict) else None
        if not isinstance(model_id, str) or not model_id.strip():
            raise LiveIopError("protocol_incompatible")
        model_ids.append(model_id)
    models = tuple(sorted(model_ids))
    if not models or len(set(models)) != len(models):
        raise LiveIopError("protocol_incompatible")
    return models, _identity("catalog", "\n".join(models))


def _observe(runtime: _Runtime) -> _Observation:
    models, config_identity = _catalog(runtime)
    if runtime.caller == "claude":
        _command(("claude", "--version"))
        _command(("claude", "--help"))
        return _Observation(models, config_identity, True)
    if runtime.caller == AGY_CALLER:
        version = _command(("agy", "--version"))
        help_output = _command(("agy", "--help"))
        return _Observation(models, config_identity, True, version, help_output)
    if runtime.caller == "codex":
        _command(("codex", "--version"))
        _command(("codex", "exec", "--help"))
        return _Observation(models, config_identity, True)
    raise LiveIopError("protocol_incompatible")


def _config_from_environment(environment: Mapping[str, str]) -> _ConfigObservation:
    reference = environment.get("IOP_BENCH_CONFIG_OBSERVATION_ENV")
    if not isinstance(reference, str) or not _ENV_NAME.fullmatch(reference):
        raise LiveIopError("route_missing")
    raw = environment.get(reference)
    if not isinstance(raw, str) or not raw:
        raise LiveIopError("route_missing")
    try:
        value = json.loads(raw)
    except (TypeError, json.JSONDecodeError) as exc:
        raise LiveIopError("protocol_incompatible") from exc
    if not isinstance(value, dict) or set(value) != {"schema_version", "routes"} or value.get("schema_version") != "1":
        raise LiveIopError("protocol_incompatible")
    raw_routes = value.get("routes")
    if not isinstance(raw_routes, list) or not raw_routes:
        raise LiveIopError("route_missing")
    routes: list[_RouteObservation] = []
    for item in raw_routes:
        if not isinstance(item, dict) or set(item) != {"route_kind", "route_id", "model"}:
            raise LiveIopError("protocol_incompatible")
        route_kind, route_id, model = (item.get(name) for name in ("route_kind", "route_id", "model"))
        if route_kind != "direct" or not all(isinstance(field, str) and TOKEN_RE.fullmatch(field) for field in (route_id, model)):
            raise LiveIopError("protocol_incompatible")
        routes.append(_RouteObservation(route_kind, route_id, model))
    if len({(item.route_kind, item.route_id) for item in routes}) != len(routes):
        raise LiveIopError("protocol_incompatible")
    routes.sort(key=lambda item: (item.route_kind, item.route_id, item.model))
    canonical = json.dumps({"schema_version": "1", "routes": [item.__dict__ for item in routes]}, sort_keys=True, separators=(",", ":"), ensure_ascii=True)
    return _ConfigObservation(tuple(routes), _identity("config", canonical))


def _runtime_from_environment(caller: str, environment: Mapping[str, str]) -> _RuntimeResolution:
    prefix = f"IOP_BENCH_{caller.upper()}_"
    base_key = prefix + "BASE_URL"
    secret_ref_key = prefix + "SECRET_ENV"
    base_url = environment.get(base_key)
    secret_ref = environment.get(secret_ref_key)
    try:
        normalized = _base_url(base_url)
    except LiveIopError as exc:
        return _RuntimeResolution(None, exc.issue_code)
    if not isinstance(secret_ref, str) or not _ENV_NAME.fullmatch(secret_ref):
        return _RuntimeResolution(None, "credential_missing")
    secret = environment.get(secret_ref)
    if not isinstance(secret, str) or not secret:
        return _RuntimeResolution(None, "credential_missing")
    try:
        config = _config_from_environment(environment)
    except LiveIopError as exc:
        return _RuntimeResolution(None, exc.issue_code)
    return _RuntimeResolution(_Runtime(caller, normalized, secret, _identity("endpoint", normalized), config), None)


def _requested(cell: MatrixCell) -> RequestedEffectiveBinding:
    return RequestedEffectiveBinding(
        cell.id, cell.caller, cell.iop.route_kind, cell.iop.route_id,
        cell.iop.request_model, cell.iop.requested_effort,
    )


def _issues(*codes: str) -> tuple[ConnectivityIssue, ...]:
    selected = set(codes)
    return tuple(ConnectivityIssue(code, ISSUE_RESUME_CODES[code]) for code in ISSUE_CODE_ORDER if code in selected)


def _binding_from_config(cell: MatrixCell, capability: CallerCapability, config: _ConfigObservation) -> tuple[RequestedEffectiveBinding, tuple[ConnectivityIssue, ...]]:
    """Admit a direct cell from config ownership and caller capability only."""
    requested = _requested(cell)
    if cell.iop.route_kind not in capability.route_kinds:
        return requested, _issues("protocol_incompatible")
    if cell.iop.requested_effort not in capability.efforts:
        return requested, _issues("effort_unsupported")
    route = next((item for item in config.routes if (item.route_kind, item.route_id) == (cell.iop.route_kind, cell.iop.route_id)), None)
    if route is None:
        return requested, _issues("route_missing")
    if route.model != cell.iop.request_model:
        return requested, _issues("model_missing")
    return RequestedEffectiveBinding(
        cell.id, cell.caller, cell.iop.route_kind, cell.iop.route_id,
        cell.iop.request_model, cell.iop.requested_effort,
        route.route_kind, route.route_id, route.model, cell.iop.requested_effort,
        (EffectiveBinding("request", route.model, cell.iop.requested_effort),),
    ), ()


def _default_claude_invoker(adapter: ClaudeIopAdapter, spec: Any, **kwargs: Any) -> InvocationResult:
    return run_invocation(spec, **kwargs)


def _default_agy_invoker(spec: Any, parser: AgyEventParser, preflight: Any, on_started: Callable[..., None]) -> InvocationResult:
    return run_agy_invocation(spec, parser, preflight, on_started)


def _default_codex_invoker(invocation: Any, on_started: Callable[..., None]) -> Any:
    return run_codex_invocation(invocation, on_started)


_DEFAULT_INVOKERS = _InvokerSeams(_default_claude_invoker, _default_agy_invoker, _default_codex_invoker)


def _bind_live_spec(
    cell: MatrixCell,
    prepared: PreparedWorkspace,
    attempt: Attempt,
    control_dir: str,
    spec: Any,
) -> Any:
    """Immutably bind one caller spec to the controller-owned attempt paths."""
    if (
        not isinstance(prepared, PreparedWorkspace)
        or not isinstance(attempt, Attempt)
        or cell.id != attempt.identity.cell_id
        or prepared.identity != attempt.identity
        or not isinstance(control_dir, str)
        or not control_dir
    ):
        raise LiveIopError("stream_incompatible")
    try:
        attempt_root = Path(attempt.root).resolve(strict=True)
        prepared_root = Path(prepared.attempt_root).resolve(strict=True)
        workspace_root = Path(prepared.workspace_dir).resolve(strict=True)
        session_root = Path(prepared.session_dir).resolve(strict=True)
        evidence_root = Path(spec.evidence_dir).resolve(strict=True)
        control_path = Path(control_dir)
        alias = control_path.parent
        if (
            not control_path.is_absolute()
            or control_path.name != "control"
            or control_path.exists()
            or control_path.is_symlink()
            or not alias.is_symlink()
            or alias.resolve(strict=True) != attempt_root
            or control_path.resolve(strict=False) != attempt_root / "control"
        ):
            raise LiveIopError("stream_incompatible")
    except (OSError, RuntimeError, TypeError, ValueError) as exc:
        raise LiveIopError("stream_incompatible") from exc
    if (
        prepared_root != attempt_root
        or evidence_root != attempt_root
        or workspace_root != attempt_root / "workspace"
        or session_root != attempt_root / "session"
    ):
        raise LiveIopError("stream_incompatible")
    return replace(spec, control_dir=control_dir)


class _LiveAdapter:
    """One caller's live observation and invocation boundary."""

    def __init__(
        self, caller: str, capability: CallerCapability, runtime: _RuntimeResolution,
        observer: Callable[[_Runtime], _Observation] = _observe,
        binary_resolver: Callable[[str], str] = _caller_binary,
        invokers: _InvokerSeams = _DEFAULT_INVOKERS,
    ) -> None:
        self.caller = caller
        self.capability = capability
        self._runtime_resolution = runtime
        self._observer = observer
        self._binary_resolver = binary_resolver
        self._invokers = invokers
        self._agy_preflight: Any = None
        self._admitted_bindings: dict[str, RequestedEffectiveBinding] = {}

    def preflight(self, cell: MatrixCell) -> PreflightObservation:
        if cell.caller != self.caller:
            raise LiveIopError("protocol_incompatible")
        resolution = self._runtime_resolution
        runtime = resolution.runtime
        if runtime is None:
            result = make_result(cell, self.capability, _requested(cell), _issues(resolution.issue_code or "credential_missing"))
            return PreflightObservation(result, _identity("missing", self.caller), _identity("missing-config", self.caller))
        try:
            observed = self._observer(runtime)
        except LiveIopError as exc:
            result = make_result(cell, self.capability, _requested(cell), _issues(exc.issue_code))
            return PreflightObservation(result, runtime.endpoint_identity, _identity("unobserved-config", self.caller))
        if not observed.caller_ready:
            result = make_result(cell, self.capability, _requested(cell), _issues("stream_incompatible"))
            return PreflightObservation(result, runtime.endpoint_identity, runtime.config.identity)
        if self.caller == AGY_CALLER:
            capability = inspect_agy_iop_capability(observed.agy_version, observed.agy_help)
            agy_observation = AgyRuntimeObservation(
                cell.id, cell.iop.route_kind, cell.iop.route_id,
                _agy_runtime_identity("endpoint", runtime.base_url),
                _agy_runtime_identity("credential", runtime.secret), runtime.config.identity,
            )
            try:
                self._agy_preflight = preflight_agy_iop(
                    cell, capability,
                    AgyRuntimeInputs(self._binary_resolver("agy"), runtime.base_url, runtime.secret),
                    agy_observation,
                )
            except Exception:
                result = make_result(cell, self.capability, _requested(cell), _issues("stream_incompatible"))
                return PreflightObservation(result, runtime.endpoint_identity, runtime.config.identity)
            if self._agy_preflight.issues:
                result = make_result(cell, self.capability, _requested(cell), self._agy_preflight.issues)
                return PreflightObservation(result, runtime.endpoint_identity, runtime.config.identity)
        binding, issues = _binding_from_config(cell, self.capability, runtime.config)
        if not issues and binding.effective_model not in observed.catalog_models:
            binding, issues = _requested(cell), _issues("model_missing")
        result = make_result(cell, self.capability, binding, issues)
        if result.status == "ready":
            self._admitted_bindings[cell.id] = result.binding
        return PreflightObservation(result, runtime.endpoint_identity, runtime.config.identity)

    def invoke(
        self,
        cell: MatrixCell,
        prepared: PreparedWorkspace,
        attempt: Attempt,
        control_dir: str,
        task_payload: bytes,
        timeout: Timeout,
        on_started: Callable[..., None],
    ) -> InvocationResult:
        if cell.caller != self.caller:
            raise LiveIopError("protocol_incompatible")
        runtime = self._runtime_resolution.runtime
        if runtime is None:
            raise LiveIopError(self._runtime_resolution.issue_code or "protocol_incompatible")
        admitted = self._admitted_bindings.get(cell.id)
        if admitted is None:
            raise LiveIopError("stream_incompatible")
        if self.caller == "claude":
            adapter = ClaudeIopAdapter(cell, prepared, ClaudeIopRuntime("claude", runtime.base_url, runtime.secret))
            spec = _bind_live_spec(
                cell,
                prepared,
                attempt,
                control_dir,
                adapter.invocation(task_payload.decode("utf-8"), attempt.root, timeout),
            )
            return self._invokers.claude(adapter, spec, parse_event=adapter.parser(), redact=adapter.redactor(task_payload.decode("utf-8")), on_started=lambda locator: on_started(locator, spec_digest(spec)))
        if self.caller == AGY_CALLER:
            if self._agy_preflight is None:
                raise LiveIopError("stream_incompatible")
            spec = _bind_live_spec(
                cell,
                prepared,
                attempt,
                control_dir,
                build_agy_invocation(
                    cell, prepared, task_payload, timeout, self._agy_preflight
                ),
            )
            parser = AgyEventParser(cell)
            result = self._invokers.agy(spec, parser, self._agy_preflight, lambda locator: on_started(locator, spec_digest(spec)))
            observed = parser.observed_result(self._agy_preflight.capability, result)
            if observed.status != "ready" or observed.binding != admitted:
                raise LiveIopError("stream_incompatible")
            return result
        if self.caller == "codex":
            invocation = build_codex_invocation(cell, prepared, runtime_from_environment({BASE_URL_ENV_KEY: runtime.base_url, SECRET_ENV_KEY: runtime.secret, "PATH": os.environ.get("PATH", "/usr/bin:/bin")}), task_payload, timeout)
            invocation = replace(
                invocation,
                spec=_bind_live_spec(
                    cell,
                    prepared,
                    attempt,
                    control_dir,
                    invocation.spec,
                ),
            )
            result = self._invokers.codex(invocation, lambda locator: on_started(locator, spec_digest(invocation.spec)))
            expected = (admitted.effective_route_kind, admitted.effective_route_id, admitted.effective_model, admitted.effective_effort)
            if result.effective_binding != expected:
                raise LiveIopError("stream_incompatible")
            return result.lifecycle
        raise LiveIopError("protocol_incompatible")


def build_live_adapter_registry(
    environment: Mapping[str, str], *, observer: Callable[[_Runtime], _Observation] = _observe,
    binary_resolver: Callable[[str], str] = _caller_binary,
    invokers: _InvokerSeams = _DEFAULT_INVOKERS,
) -> dict[str, ExecutionAdapter]:
    """Build the fixed caller registry without reading ambient caller settings."""
    if not isinstance(environment, Mapping):
        raise LiveIopError("protocol_incompatible")
    registry: dict[str, ExecutionAdapter] = {
        "claude": _LiveAdapter("claude", claude_capability(), _runtime_from_environment("claude", environment), observer, binary_resolver, invokers),
        "agy": _LiveAdapter(AGY_CALLER, CallerCapability(AGY_CALLER, ("direct", "execution_preset"), ("high", "low", "medium")), _runtime_from_environment("agy", environment), observer, binary_resolver, invokers),
        "codex": _LiveAdapter("codex", codex_capability(), _runtime_from_environment("codex", environment), observer, binary_resolver, invokers),
    }
    if tuple(registry) != CALLER_ENUM:
        raise LiveIopError("protocol_incompatible")
    return registry