"""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 stat
import subprocess
import tempfile
from collections.abc import Callable, Mapping
from contextlib import ExitStack, contextmanager
from dataclasses import dataclass, replace
from pathlib import Path
from typing import Any, Iterator
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,
    STAGE_ENUM,
    MatrixCell,
    TOKEN_RE,
    Timeout,
)
from scripts.agent_benchmark.scoring import (
    BlindWorkspace,
    ScoringAdapter,
    ScoringEvidenceFinalization,
    ScoringInvocationResult,
)
from scripts.agent_benchmark.workspace import (
    AttemptIdentity,
    PreparedWorkspace,
    TestbedProvenance,
)


_ENV_NAME = re.compile(r"^[A-Za-z_][A-Za-z0-9_]{0,63}$")
_CALLERS = ("claude", "agy", "codex")
_TIMEOUT_SECONDS = 10
_SECRET_SCAN_MAX_ENTRIES = 100_000
_SECRET_SCAN_MAX_DEPTH = 64
_SECRET_SCAN_MAX_FILE_BYTES = 32 * 1024 * 1024


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
    bindings: tuple[EffectiveBinding, ...]


@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", "bindings"
        }:
            raise LiveIopError("protocol_incompatible")
        route_kind, route_id, model = (item.get(name) for name in ("route_kind", "route_id", "model"))
        if route_kind not in ("direct", "execution_preset") or not all(isinstance(field, str) and TOKEN_RE.fullmatch(field) for field in (route_id, model)):
            raise LiveIopError("protocol_incompatible")
        raw_bindings = item.get("bindings")
        if not isinstance(raw_bindings, list) or not raw_bindings:
            raise LiveIopError("protocol_incompatible")
        bindings: list[EffectiveBinding] = []
        stages: set[str] = set()
        for raw_binding in raw_bindings:
            if not isinstance(raw_binding, dict) or set(raw_binding) != {
                "stage", "model", "effort"
            }:
                raise LiveIopError("protocol_incompatible")
            stage = raw_binding.get("stage")
            bound_model = raw_binding.get("model")
            effort = raw_binding.get("effort")
            if (
                stage not in STAGE_ENUM
                or stage in stages
                or not isinstance(bound_model, str)
                or not TOKEN_RE.fullmatch(bound_model)
                or (
                    effort is not None
                    and (
                        not isinstance(effort, str)
                        or not TOKEN_RE.fullmatch(effort)
                    )
                )
            ):
                raise LiveIopError("protocol_incompatible")
            stages.add(stage)
            bindings.append(EffectiveBinding(stage, bound_model, effort))
        routes.append(
            _RouteObservation(route_kind, route_id, model, tuple(bindings))
        )
    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_routes = [
        {
            "route_kind": item.route_kind,
            "route_id": item.route_id,
            "model": item.model,
            "bindings": [
                {
                    "stage": binding.stage,
                    "model": binding.model,
                    "effort": binding.effort,
                }
                for binding in item.bindings
            ],
        }
        for item in routes
    ]
    canonical = json.dumps(
        {"schema_version": "1", "routes": canonical_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 manifest route 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")
    expected = tuple(
        EffectiveBinding(binding.stage, binding.model, binding.effort)
        for binding in cell.iop.expected_bindings
    )
    if route.bindings != expected:
        return requested, _issues("protocol_incompatible")
    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,
        route.bindings,
    ), ()


def _bound_observations(
    result: InvocationResult, admitted: RequestedEffectiveBinding
) -> InvocationResult:
    """Return the lifecycle result only when every observation stays bound.

    Typed observations ride on the lifecycle result, so this boundary keeps the
    result intact and refuses any observation whose model label names something
    other than the admitted binding.
    """
    if not isinstance(result, InvocationResult):
        raise LiveIopError("stream_incompatible")
    admitted_models = {
        value for value in (admitted.effective_model, admitted.requested_model) if value
    }
    admitted_models.update(
        binding.model for binding in admitted.effective_bindings if binding.model
    )
    admitted_by_stage = {
        binding.stage: binding.model
        for binding in admitted.effective_bindings
        if binding.stage and binding.model
    }
    for metric in result.metrics:
        if (
            metric.stage
            and metric.model
            and admitted_by_stage.get(metric.stage) != metric.model
        ):
            raise LiveIopError("stream_incompatible")
        if metric.model and metric.model not in admitted_models:
            raise LiveIopError("stream_incompatible")
    return result


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)


@contextmanager
def _temporary_owned_permissions(
    path: Path,
    required: int,
    *,
    expected: os.stat_result | None = None,
) -> Iterator[os.stat_result]:
    """Grant minimum owner access and restore a retained inode exactly."""
    try:
        info = os.lstat(path)
    except OSError as exc:
        raise LiveIopError("stream_incompatible") from exc
    current_uid = getattr(os, "geteuid", lambda: info.st_uid)()
    if (
        stat.S_ISLNK(info.st_mode)
        or info.st_uid != current_uid
        or (
            expected is not None
            and (info.st_dev, info.st_ino) != (expected.st_dev, expected.st_ino)
        )
    ):
        raise LiveIopError("stream_incompatible")
    original_mode = stat.S_IMODE(info.st_mode)
    temporary_mode = original_mode | required
    changed = temporary_mode != original_mode
    try:
        if changed:
            os.chmod(path, temporary_mode, follow_symlinks=False)
        current = os.lstat(path)
        if (
            stat.S_ISLNK(current.st_mode)
            or current.st_uid != current_uid
            or (current.st_dev, current.st_ino) != (info.st_dev, info.st_ino)
            or stat.S_IMODE(current.st_mode) & required != required
        ):
            raise LiveIopError("stream_incompatible")
        yield current
    except OSError as exc:
        raise LiveIopError("stream_incompatible") from exc
    finally:
        if changed:
            try:
                current = os.lstat(path)
                if (
                    stat.S_ISLNK(current.st_mode)
                    or current.st_uid != current_uid
                    or (current.st_dev, current.st_ino) != (info.st_dev, info.st_ino)
                ):
                    raise LiveIopError("stream_incompatible")
                os.chmod(path, original_mode, follow_symlinks=False)
                restored = os.lstat(path)
                if (
                    (restored.st_dev, restored.st_ino) != (info.st_dev, info.st_ino)
                    or stat.S_IMODE(restored.st_mode) != original_mode
                ):
                    raise LiveIopError("stream_incompatible")
            except OSError as exc:
                raise LiveIopError("stream_incompatible") from exc


@contextmanager
def _temporary_directory_chain(
    root: Path, parent: Path, *, writable_parent: bool = False
) -> Iterator[None]:
    try:
        relative = parent.relative_to(root)
    except ValueError as exc:
        raise LiveIopError("stream_incompatible") from exc
    chain = [root]
    current = root
    for part in relative.parts:
        current = current / part
        chain.append(current)
    with ExitStack() as stack:
        for index, directory in enumerate(chain):
            required = stat.S_IXUSR
            if writable_parent and index == len(chain) - 1:
                required |= stat.S_IWUSR
            stack.enter_context(_temporary_owned_permissions(directory, required))
        yield


def _owned_plain_directory(path: Path) -> Path:
    try:
        info = os.lstat(path)
    except OSError as exc:
        raise LiveIopError("stream_incompatible") from exc
    current_uid = getattr(os, "geteuid", lambda: info.st_uid)()
    if (
        not stat.S_ISDIR(info.st_mode)
        or stat.S_ISLNK(info.st_mode)
        or info.st_uid != current_uid
    ):
        raise LiveIopError("stream_incompatible")
    with _temporary_owned_permissions(path, stat.S_IXUSR, expected=info):
        try:
            resolved = path.resolve(strict=True)
        except OSError as exc:
            raise LiveIopError("stream_incompatible") from exc
    return resolved


def _blind_controlled_roots(
    blind: BlindWorkspace,
) -> tuple[Path, tuple[Path, Path, Path]]:
    blind_path = Path(blind.root)
    blind_root = _owned_plain_directory(blind_path)
    controlled = (
        _owned_plain_directory(Path(blind.input_dir)),
        _owned_plain_directory(Path(blind.session_dir)),
        _owned_plain_directory(Path(blind.output_dir)),
    )
    expected = tuple(blind_root / name for name in ("input", "session", "output"))
    if controlled != expected:
        raise LiveIopError("stream_incompatible")
    return blind_root, controlled


def _walk_no_follow(root: Path) -> tuple[tuple[Path, os.stat_result], ...]:
    entries: list[tuple[Path, os.stat_result]] = []

    def visit(
        directory: Path, depth: int, expected: os.stat_result | None = None
    ) -> None:
        if depth > _SECRET_SCAN_MAX_DEPTH:
            raise LiveIopError("stream_incompatible")
        with _temporary_owned_permissions(
            directory, stat.S_IRUSR | stat.S_IXUSR, expected=expected
        ):
            try:
                with os.scandir(directory) as iterator:
                    children = sorted(iterator, key=lambda item: item.name)
            except OSError as exc:
                raise LiveIopError("stream_incompatible") from exc
            for entry in children:
                try:
                    info = entry.stat(follow_symlinks=False)
                except OSError as exc:
                    raise LiveIopError("stream_incompatible") from exc
                path = directory / entry.name
                entries.append((path, info))
                if len(entries) > _SECRET_SCAN_MAX_ENTRIES:
                    raise LiveIopError("stream_incompatible")
                if stat.S_ISDIR(info.st_mode) and not stat.S_ISLNK(info.st_mode):
                    visit(path, depth + 1, info)

    visit(root, 0)
    return tuple(entries)


def _read_bounded_regular(
    root: Path, path: Path, expected: os.stat_result
) -> bytes:
    if expected.st_size > _SECRET_SCAN_MAX_FILE_BYTES:
        raise LiveIopError("stream_incompatible")
    flags = os.O_RDONLY | os.O_CLOEXEC | os.O_NONBLOCK
    if hasattr(os, "O_NOFOLLOW"):
        flags |= os.O_NOFOLLOW
    with _temporary_directory_chain(root, path.parent):
        with _temporary_owned_permissions(path, stat.S_IRUSR, expected=expected):
            try:
                descriptor = os.open(path, flags)
            except OSError as exc:
                raise LiveIopError("stream_incompatible") from exc
            try:
                current = os.fstat(descriptor)
                if (
                    not stat.S_ISREG(current.st_mode)
                    or current.st_uid
                    != getattr(os, "geteuid", lambda: current.st_uid)()
                    or current.st_dev != expected.st_dev
                    or current.st_ino != expected.st_ino
                    or current.st_size != expected.st_size
                    or current.st_size > _SECRET_SCAN_MAX_FILE_BYTES
                ):
                    raise LiveIopError("stream_incompatible")
                data = bytearray()
                while len(data) < current.st_size:
                    chunk = os.read(descriptor, current.st_size - len(data))
                    if not chunk:
                        raise LiveIopError("stream_incompatible")
                    data.extend(chunk)
                if os.read(descriptor, 1):
                    raise LiveIopError("stream_incompatible")
                final = os.fstat(descriptor)
                if (
                    final.st_size != current.st_size
                    or final.st_mtime_ns != current.st_mtime_ns
                ):
                    raise LiveIopError("stream_incompatible")
                return bytes(data)
            except OSError as exc:
                raise LiveIopError("stream_incompatible") from exc
            finally:
                os.close(descriptor)


def _contains_sensitive(data: bytes, sensitive: tuple[bytes, ...]) -> bool:
    return any(value in data for value in sensitive)


def _relative_bytes(root: Path, path: Path) -> bytes:
    try:
        return os.fsencode(path.relative_to(root).as_posix())
    except ValueError as exc:
        raise LiveIopError("stream_incompatible") from exc


def _unlink_owned_entry(
    root: Path,
    path: Path,
    expected: os.stat_result,
    *,
    directory: bool,
) -> None:
    with _temporary_directory_chain(root, path.parent, writable_parent=True):
        try:
            current = os.lstat(path)
            current_uid = getattr(os, "geteuid", lambda: current.st_uid)()
            if (
                current.st_uid != current_uid
                or (current.st_dev, current.st_ino)
                != (expected.st_dev, expected.st_ino)
                or (directory and not stat.S_ISDIR(current.st_mode))
                or (not directory and stat.S_ISDIR(current.st_mode))
            ):
                raise LiveIopError("stream_incompatible")
            if directory:
                path.rmdir()
            else:
                path.unlink()
        except OSError as exc:
            raise LiveIopError("stream_incompatible") from exc


@dataclass
class _SanitizationOutcome:
    secret: bool = False
    input_invalid: bool = False
    output_invalid: bool = False

    def record(self, root_kind: str, *, secret: bool, invalid: bool) -> None:
        self.secret = self.secret or secret
        if invalid and root_kind == "input":
            self.input_invalid = True
        elif invalid:
            self.output_invalid = True

    def finalization(self) -> ScoringEvidenceFinalization:
        if self.secret:
            return ScoringEvidenceFinalization(False, "runtime_secret_leak")
        if self.input_invalid:
            return ScoringEvidenceFinalization(False, "input_mutated")
        if self.output_invalid:
            return ScoringEvidenceFinalization(False, "evaluator_output_leak")
        return ScoringEvidenceFinalization(True)


def _remove_sensitive_blind_paths(
    roots: tuple[Path, Path, Path], sensitive: tuple[bytes, ...]
) -> _SanitizationOutcome:
    outcome = _SanitizationOutcome()
    for root_kind, root in zip(("input", "session", "output"), roots):
        entries = sorted(
            _walk_no_follow(root),
            key=lambda item: (len(item[0].parts), item[0].as_posix()),
            reverse=True,
        )
        for path, info in entries:
            current_uid = getattr(os, "geteuid", lambda: info.st_uid)()
            if info.st_uid != current_uid:
                raise LiveIopError("stream_incompatible")
            path_leak = _contains_sensitive(_relative_bytes(root, path), sensitive)
            mode = info.st_mode
            if stat.S_ISDIR(mode) and not stat.S_ISLNK(mode):
                if path_leak:
                    _unlink_owned_entry(root, path, info, directory=True)
                    outcome.record(root_kind, secret=True, invalid=False)
                continue
            if stat.S_ISLNK(mode):
                with _temporary_directory_chain(root, path.parent):
                    try:
                        link_value = os.fsencode(os.readlink(path))
                        current = os.lstat(path)
                    except OSError as exc:
                        raise LiveIopError("stream_incompatible") from exc
                if (current.st_dev, current.st_ino) != (info.st_dev, info.st_ino):
                    raise LiveIopError("stream_incompatible")
                link_leak = path_leak or _contains_sensitive(link_value, sensitive)
                _unlink_owned_entry(root, path, info, directory=False)
                outcome.record(root_kind, secret=link_leak, invalid=not link_leak)
                continue
            if stat.S_ISREG(mode):
                content_leak = False
                if not path_leak:
                    content_leak = _contains_sensitive(
                        _read_bounded_regular(root, path, info), sensitive
                    )
                if path_leak or content_leak:
                    _unlink_owned_entry(root, path, info, directory=False)
                    outcome.record(root_kind, secret=True, invalid=False)
                continue
            if stat.S_ISSOCK(mode):
                if path_leak:
                    _unlink_owned_entry(root, path, info, directory=False)
                    outcome.record(root_kind, secret=True, invalid=False)
                continue
            if path_leak:
                _unlink_owned_entry(root, path, info, directory=False)
                outcome.record(root_kind, secret=True, invalid=False)
                continue
            raise LiveIopError("stream_incompatible")
    return outcome


def _freeze_sanitized_input(root: Path) -> None:
    entries = _walk_no_follow(root)
    directories = [root]
    current_uid = getattr(os, "geteuid", lambda: os.lstat(root).st_uid)()
    for path, info in entries:
        if info.st_uid != current_uid:
            raise LiveIopError("stream_incompatible")
        try:
            if stat.S_ISDIR(info.st_mode) and not stat.S_ISLNK(info.st_mode):
                directories.append(path)
            elif stat.S_ISREG(info.st_mode) and not stat.S_ISLNK(info.st_mode):
                os.chmod(path, 0o400, follow_symlinks=False)
            else:
                raise LiveIopError("stream_incompatible")
        except OSError as exc:
            raise LiveIopError("stream_incompatible") from exc
    for directory in sorted(
        directories, key=lambda item: len(item.parts), reverse=True
    ):
        try:
            os.chmod(directory, 0o500, follow_symlinks=False)
        except OSError as exc:
            raise LiveIopError("stream_incompatible") from exc


def _verify_sensitive_absent(root: Path, sensitive: tuple[bytes, ...]) -> None:
    root = _owned_plain_directory(root)
    for path, info in _walk_no_follow(root):
        if _contains_sensitive(_relative_bytes(root, path), sensitive):
            raise LiveIopError("stream_incompatible")
        if stat.S_ISREG(info.st_mode):
            if _contains_sensitive(
                _read_bounded_regular(root, path, info), sensitive
            ):
                raise LiveIopError("stream_incompatible")
        elif stat.S_ISLNK(info.st_mode):
            with _temporary_directory_chain(root, path.parent):
                try:
                    link_value = os.fsencode(os.readlink(path))
                except OSError as exc:
                    raise LiveIopError("stream_incompatible") from exc
            if _contains_sensitive(link_value, sensitive):
                raise LiveIopError("stream_incompatible")


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 _bound_observations(
                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))),
                admitted,
            )
        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 _bound_observations(result, admitted)
        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 _bound_observations(result.lifecycle, admitted)
        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


class _LiveScoringAdapter:
    """Codex-only scoring adapter sharing the live config and secret boundary."""

    def __init__(
        self,
        live: _LiveAdapter,
        *,
        invoker: Callable[[Any, Callable[..., None]], Any] = _default_codex_invoker,
    ) -> None:
        self.capability = codex_capability()
        self._live = live
        self._invoker = invoker
        self._control_aliases: dict[str, Path] = {}

    def preflight(self, cell: MatrixCell) -> PreflightObservation:
        if cell.id != "evaluator" or cell.caller != "codex":
            raise LiveIopError("protocol_incompatible")
        return self._live.preflight(cell)

    def invoke(
        self,
        cell: MatrixCell,
        blind: BlindWorkspace,
        task_payload: bytes,
        timeout: Timeout,
        on_started: Callable[..., None],
    ) -> ScoringInvocationResult:
        if cell.id != "evaluator" or cell.caller != "codex":
            raise LiveIopError("protocol_incompatible")
        runtime = self._live._runtime_resolution.runtime
        admitted = self._live._admitted_bindings.get(cell.id)
        if runtime is None or admitted is None:
            raise LiveIopError("stream_incompatible")
        blind_root = Path(blind.root).resolve(strict=True)
        input_root = Path(blind.input_dir).resolve(strict=True)
        session_root = Path(blind.session_dir).resolve(strict=True)
        output_root = Path(blind.output_dir).resolve(strict=True)
        if (
            input_root != blind_root / "input"
            or session_root != blind_root / "session"
            or output_root != blind_root / "output"
            or any(path.is_symlink() for path in (blind_root, input_root, session_root, output_root))
        ):
            raise LiveIopError("stream_incompatible")
        prepared = PreparedWorkspace(
            identity=AttemptIdentity("run-blind", "blind", 1, 1),
            attempt_root=str(output_root),
            workspace_dir=str(blind_root),
            session_dir=str(session_root),
            session_id=blind.session_identity,
            session_is_fresh=True,
            workspace_checksum=blind.input_digest,
            setup_cache_policy="isolated",
            testbed_provenance=TestbedProvenance(
                path="opaque", branch="opaque", head="opaque",
                status_digest="sha256:" + "0" * 64, clean=True,
            ),
            prepared_at="opaque",
        )
        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,
        )
        alias = Path(tempfile.gettempdir()).resolve() / (
            "iop-bench-score-"
            + hashlib.sha256(os.fsencode(str(output_root))).hexdigest()[:20]
        )
        try:
            os.symlink(str(output_root), alias, target_is_directory=True)
        except FileExistsError:
            try:
                if not alias.is_symlink() or alias.resolve(strict=True) != output_root:
                    raise LiveIopError("stream_incompatible")
            except OSError as exc:
                raise LiveIopError("stream_incompatible") from exc
        except OSError as exc:
            raise LiveIopError("stream_incompatible") from exc
        self._control_aliases[blind.blind_id] = alias
        invocation = replace(
            invocation,
            spec=replace(
                invocation.spec,
                evidence_dir=str(output_root),
                control_dir=str(alias / "codex-control"),
            ),
        )
        result = self._invoker(
            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:
            return ScoringInvocationResult(
                False, "binding_mismatch", result.effective_binding
            )
        lifecycle = _bound_observations(result.lifecycle, admitted)
        return ScoringInvocationResult(
            lifecycle.success,
            lifecycle.terminal_reason,
            result.effective_binding,
        )

    def finalize_evidence(
        self, blind: BlindWorkspace
    ) -> ScoringEvidenceFinalization:
        """Remove exact evaluator runtime values before controller publication."""
        runtime = self._live._runtime_resolution.runtime
        if runtime is None:
            raise LiveIopError("stream_incompatible")
        blind_root, controlled = _blind_controlled_roots(blind)
        sensitive = tuple(
            dict.fromkeys(
                value.encode("utf-8")
                for value in (runtime.secret, runtime.base_url)
                if value
            )
        )
        alias = self._control_aliases.pop(blind.blind_id, None)
        if alias is not None and (alias.exists() or alias.is_symlink()):
            try:
                if not alias.is_symlink() or alias.resolve(strict=True) != controlled[2]:
                    raise LiveIopError("stream_incompatible")
                alias.unlink()
            except OSError as exc:
                raise LiveIopError("stream_incompatible") from exc
        outcome = _remove_sensitive_blind_paths(controlled, sensitive)
        _freeze_sanitized_input(controlled[0])
        _verify_sensitive_absent(blind_root.parent.parent, sensitive)
        return outcome.finalization()


def build_live_scoring_adapter(
    environment: Mapping[str, str],
    *,
    observer: Callable[[_Runtime], _Observation] = _observe,
    binary_resolver: Callable[[str], str] = _caller_binary,
    invoker: Callable[[Any, Callable[..., None]], Any] = _default_codex_invoker,
) -> ScoringAdapter:
    """Build the manifest-bound Codex evaluator without caller fallback."""
    if not isinstance(environment, Mapping):
        raise LiveIopError("protocol_incompatible")
    live = _LiveAdapter(
        "codex",
        codex_capability(),
        _runtime_from_environment("codex", environment),
        observer,
        binary_resolver,
        _DEFAULT_INVOKERS,
    )
    return _LiveScoringAdapter(live, invoker=invoker)