"""Blind, append-only S13 scoring over immutable execution attempts.

The original cell identity remains under ``cells/`` and in a run-owned mapping
that is never copied into the evaluator tree.  A scorer receives only one
opaque ``blind/<id>`` directory, anonymous input bytes, the fixed D12 rubric,
and a fresh session/output pair.
"""

from __future__ import annotations

import hashlib
import json
import os
import re
import secrets
import stat
import time
from dataclasses import dataclass
from pathlib import Path, PurePosixPath
from typing import Any, Callable, Mapping, Protocol

from scripts.agent_benchmark.attempts import (
    Attempt,
    AttemptStateError,
    PreflightObservation,
    RunIdentity,
    RunStore,
    TERMINAL_STATES,
)
from scripts.agent_benchmark.connectivity import (
    CallerCapability,
    canonical_evidence_bytes,
    validate_result,
)
from scripts.agent_benchmark.manifest import Manifest, MatrixCell, Timeout
from scripts.agent_benchmark.lifecycle import (
    LifecycleRecoveryError,
    REASON_CONTROLLER_LOST,
    REASON_RECOVERED_STOP,
    SupervisorLocator,
    TERMINAL_REASONS,
    recover_invocation,
)
from scripts.agent_benchmark.rubric import (
    RUBRIC_CATEGORIES,
    RubricError,
    Worksheet,
    canonical_worksheet_bytes,
    load_worksheet,
)
from scripts.agent_benchmark.web_validation import (
    GENERATED_FILES,
    WEB_GATES,
    WEB_VALIDATION_FILENAME,
    WebValidationError,
    load_web_validation,
)


SCORING_VERSION = 1
SCORE_RE = re.compile(r"^score-([0-9]{6})$")
BLIND_ID_RE = re.compile(r"^blind-[0-9a-f]{32}$")
DIGEST_RE = re.compile(r"^sha256:[0-9a-f]{64}$")
IMAGE_SUFFIXES = frozenset((".png", ".jpg", ".jpeg", ".gif", ".webp", ".svg"))
MAX_INPUT_FILE_BYTES = 32 * 1024 * 1024
MAX_RECORD_BYTES = 256 * 1024
UNSCORED_FILENAME = "unscored.json"
ALLOCATION_FILENAME = "allocation.json"
INPUT_FILENAME = "input.json"
RESULT_FILENAME = "result.json"
RUNNER_FILENAME = "runner.json"
SCORING_STATUSES = ("scored", "unscored", "scoring_failed", "blocked")
_POST_CLEANUP_TIMEOUT_SECONDS = 2.0
_POST_CLEANUP_QUIET_SECONDS = 0.2
_POST_CLEANUP_POLL_SECONDS = 0.01


class ScoringError(Exception):
    """Scoring state or evaluator evidence cannot be trusted."""


@dataclass(frozen=True)
class BlindWorkspace:
    blind_id: str
    root: str
    input_dir: str
    session_dir: str
    output_dir: str
    input_digest: str
    session_identity: str


@dataclass(frozen=True)
class ScoringInvocationResult:
    success: bool
    terminal_reason: str
    effective_binding: tuple[str, str, str, str] | None


@dataclass(frozen=True)
class ScoringEvidenceFinalization:
    """Closed post-invocation projection from the secret-owning adapter."""

    safe: bool
    reason: str = ""


@dataclass(frozen=True)
class ProducerIdentity:
    """Producer-only identity values, separated from evaluator evidence."""

    exact_tokens: tuple[str, ...]
    path_tokens: tuple[str, ...]
    producer_tokens: tuple[str, ...]
    evaluator_shared_tokens: tuple[str, ...]


@dataclass(frozen=True)
class ScoringSummary:
    run_id: str
    scored: int
    unscored: int
    scoring_failed: int
    blocked: int


class ScoringAdapter(Protocol):
    capability: CallerCapability

    def preflight(self, cell: MatrixCell) -> PreflightObservation:
        """Return one manifest-bound, secret-free evaluator observation."""

    def invoke(
        self,
        cell: MatrixCell,
        blind: BlindWorkspace,
        task_payload: bytes,
        timeout: Timeout,
        on_started: Callable[[SupervisorLocator, str], None],
    ) -> ScoringInvocationResult:
        """Run exactly one fresh evaluator session for this score id."""

    def finalize_evidence(
        self, blind: BlindWorkspace
    ) -> ScoringEvidenceFinalization:
        """Scrub secret-owned output after cleanup and return a closed status."""


def _digest(data: bytes) -> str:
    return "sha256:" + hashlib.sha256(data).hexdigest()


def _json_bytes(value: Any) -> bytes:
    return (
        json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=True)
        .encode("ascii")
        + b"\n"
    )


def _fsync_dir(path: Path) -> None:
    fd = os.open(path, os.O_RDONLY | os.O_DIRECTORY)
    try:
        os.fsync(fd)
    finally:
        os.close(fd)


def _mkdir_new(path: Path) -> None:
    try:
        path.mkdir(mode=0o700)
        _fsync_dir(path.parent)
    except FileExistsError as exc:
        raise ScoringError("scoring allocation collision") from exc
    except OSError as exc:
        raise ScoringError("scoring directory is unavailable") from exc


def _ensure_directory(path: Path) -> None:
    try:
        mode = os.lstat(path).st_mode
    except OSError as exc:
        raise ScoringError("scoring directory is unavailable") from exc
    if not stat.S_ISDIR(mode) or path.is_symlink():
        raise ScoringError("scoring directory is invalid")


def _write_new(path: Path, data: bytes) -> None:
    flags = os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_CLOEXEC
    if hasattr(os, "O_NOFOLLOW"):
        flags |= os.O_NOFOLLOW
    try:
        fd = os.open(path, flags, 0o600)
    except OSError as exc:
        raise ScoringError("scoring evidence collision") from exc
    try:
        os.write(fd, data)
        os.fsync(fd)
    except OSError as exc:
        raise ScoringError("scoring evidence write failed") from exc
    finally:
        os.close(fd)
    _fsync_dir(path.parent)


def _read_regular(path: Path, label: str, *, maximum: int = MAX_RECORD_BYTES) -> bytes:
    flags = os.O_RDONLY | os.O_CLOEXEC | os.O_NONBLOCK
    if hasattr(os, "O_NOFOLLOW"):
        flags |= os.O_NOFOLLOW
    try:
        fd = os.open(path, flags)
    except OSError as exc:
        raise ScoringError(f"{label} is unavailable") from exc
    try:
        info = os.fstat(fd)
        if not stat.S_ISREG(info.st_mode) or info.st_size > maximum:
            raise ScoringError(f"{label} is invalid")
        data = bytearray()
        while len(data) < info.st_size:
            chunk = os.read(fd, info.st_size - len(data))
            if not chunk:
                raise ScoringError(f"{label} changed while reading")
            data.extend(chunk)
        if os.read(fd, 1):
            raise ScoringError(f"{label} changed while reading")
        return bytes(data)
    except OSError as exc:
        raise ScoringError(f"{label} is unavailable") from exc
    finally:
        os.close(fd)


def _load_canonical(path: Path, label: str) -> dict[str, Any]:
    raw = _read_regular(path, label)
    try:
        value = json.loads(raw.decode("ascii"))
    except (UnicodeDecodeError, json.JSONDecodeError) as exc:
        raise ScoringError(f"{label} is invalid") from exc
    if not isinstance(value, dict) or _json_bytes(value) != raw:
        raise ScoringError(f"{label} is not canonical")
    return value


def _relative(value: str) -> str:
    if not isinstance(value, str) or not value or "\\" in value or ":" in value:
        raise ScoringError("scoring path is invalid")
    path = PurePosixPath(value)
    if path.is_absolute() or str(path) != value or any(
        part in ("", ".", "..") for part in path.parts
    ):
        raise ScoringError("scoring path is invalid")
    return value


def _safe_source(root: Path, relative: str) -> bytes:
    """Read one bounded regular file without following any component."""
    parts = PurePosixPath(_relative(relative)).parts
    directory_flags = os.O_RDONLY | os.O_DIRECTORY | os.O_CLOEXEC
    file_flags = os.O_RDONLY | os.O_CLOEXEC | os.O_NONBLOCK
    if hasattr(os, "O_NOFOLLOW"):
        directory_flags |= os.O_NOFOLLOW
        file_flags |= os.O_NOFOLLOW
    descriptors: list[int] = []
    try:
        current = os.open(root, directory_flags)
        descriptors.append(current)
        for component in parts[:-1]:
            current = os.open(component, directory_flags, dir_fd=current)
            descriptors.append(current)
        fd = os.open(parts[-1], file_flags, dir_fd=current)
        descriptors.append(fd)
        info = os.fstat(fd)
        if not stat.S_ISREG(info.st_mode) or info.st_size > MAX_INPUT_FILE_BYTES:
            raise OSError("not a bounded regular file")
        data = bytearray()
        while len(data) < info.st_size:
            chunk = os.read(fd, info.st_size - len(data))
            if not chunk:
                raise OSError("short read")
            data.extend(chunk)
        if os.read(fd, 1):
            raise OSError("file grew while reading")
        return bytes(data)
    except OSError as exc:
        raise ScoringError("blind input source is invalid") from exc
    finally:
        for fd in reversed(descriptors):
            os.close(fd)


def _write_relative(root: Path, relative: str, data: bytes) -> None:
    target = root / _relative(relative)
    try:
        target.parent.mkdir(mode=0o700, parents=True, exist_ok=True)
    except OSError as exc:
        raise ScoringError("blind input path is unavailable") from exc
    current = root
    for part in target.relative_to(root).parts[:-1]:
        current = current / part
        _ensure_directory(current)
    _write_new(target, data)


def _identity_values(manifest: Manifest, attempt: Attempt) -> ProducerIdentity:
    cell = next(
        (item for item in manifest.matrix if item.id == attempt.identity.cell_id),
        None,
    )
    if cell is None:
        raise ScoringError("execution attempt cell is unavailable")
    evaluator = manifest.evaluator.iop
    shared = {
        evaluator.route_kind,
        evaluator.route_id,
        evaluator.request_model,
        evaluator.requested_effort,
    }
    shared.update(item.model for item in evaluator.expected_bindings)
    shared.update(
        item.effort for item in evaluator.expected_bindings if item.effort
    )
    producer = {
        cell.iop.route_kind,
        cell.iop.route_id,
        cell.iop.request_model,
        cell.iop.requested_effort,
    }
    producer.update(item.model for item in cell.iop.expected_bindings)
    producer.update(
        item.effort for item in cell.iop.expected_bindings if item.effort
    )
    return ProducerIdentity(
        exact_tokens=tuple(
            sorted(value for value in (attempt.identity.cell_id, cell.caller) if value)
        ),
        path_tokens=(str(Path(attempt.root).resolve()),),
        producer_tokens=tuple(sorted(value for value in producer if value)),
        evaluator_shared_tokens=tuple(sorted(value for value in shared if value)),
    )


def _ascii_identity_bytes(value: str) -> bytes:
    try:
        return value.encode("ascii")
    except UnicodeEncodeError:
        return b""


def _route_token_present_bytes(data: bytes, value: str) -> bool:
    candidate = _ascii_identity_bytes(value)
    if not candidate:
        return False
    pattern = rb"(?<![a-z0-9_.+-])" + re.escape(candidate) + rb"(?![a-z0-9_.+-])"
    return re.search(pattern, data, flags=re.IGNORECASE | re.ASCII) is not None


def _exact_identity_present_bytes(data: bytes, value: str) -> bool:
    """Match an ASCII caller/cell identity on the original evidence bytes."""
    candidate = _ascii_identity_bytes(value)
    if not candidate:
        return False
    pattern = rb"(?<![a-z0-9])" + re.escape(candidate) + rb"(?![a-z0-9])"
    return re.search(pattern, data, flags=re.IGNORECASE | re.ASCII) is not None


def _contains_identity(data: bytes, identity: ProducerIdentity) -> bool:
    """Detect producer identity without rejecting legitimate evaluator binding."""
    if any(
        _exact_identity_present_bytes(data, value)
        for value in identity.exact_tokens
    ):
        return True
    shared = {value.casefold() for value in identity.evaluator_shared_tokens}
    if any(
        value.casefold() not in shared
        and _route_token_present_bytes(data, value)
        for value in identity.producer_tokens
    ):
        return True

    # Paths can contain non-ASCII values, so retain decoded comparison only for
    # that typed field.  ASCII identities above are matched before lossy decode
    # can join byte runs across invalid image/screenshot bytes.
    text = data.decode("utf-8", errors="ignore")
    lowered = text.casefold()
    for value in identity.path_tokens:
        if value.casefold() in lowered:
            return True
    return False


def _path_bytes(value: str) -> bytes:
    # Frame paths explicitly instead of through the process filesystem codec so
    # ordinary Unicode keeps canonical UTF-8 bytes under any locale while
    # surrogateescaped raw POSIX filename bytes are restored exactly.
    try:
        return value.encode("utf-8", errors="surrogateescape")
    except UnicodeEncodeError as exc:
        raise ScoringError("scoring path is invalid") from exc


def _input_digest(files: list[tuple[str, bytes]]) -> str:
    framed = bytearray(b"IOP-BENCH-BLIND-INPUT-V1\0")
    for relative, data in sorted(files):
        path_bytes = _path_bytes(relative)
        framed += len(path_bytes).to_bytes(8, "big") + path_bytes
        framed += len(data).to_bytes(8, "big") + data
    return _digest(bytes(framed))


def _evaluator_cell(manifest: Manifest) -> MatrixCell:
    return MatrixCell("evaluator", manifest.evaluator.caller, manifest.evaluator.iop)


def _evaluator_payload(manifest: Manifest) -> dict[str, Any]:
    evaluator = manifest.evaluator
    return {
        "caller": evaluator.caller,
        "route_kind": evaluator.iop.route_kind,
        "route_id": evaluator.iop.route_id,
        "request_model": evaluator.iop.request_model,
        "requested_effort": evaluator.iop.requested_effort,
        "expected_bindings": [
            {
                "stage": item.stage,
                "model": item.model,
                "effort": item.effort,
            }
            for item in evaluator.iop.expected_bindings
        ],
    }


def _score_root(attempt: Attempt, *, create: bool) -> Path:
    root = Path(attempt.root) / "scoring"
    if root.exists() or root.is_symlink():
        _ensure_directory(root)
    elif create:
        _mkdir_new(root)
    return root


def _score_dirs(root: Path) -> tuple[Path, ...]:
    if not root.exists() and not root.is_symlink():
        return ()
    _ensure_directory(root)
    found: list[Path] = []
    for expected, child in enumerate(sorted(root.iterdir()), start=1):
        if child.name == UNSCORED_FILENAME:
            continue
        match = SCORE_RE.fullmatch(child.name)
        if match is None or int(match.group(1)) != expected:
            raise ScoringError("scoring attempt sequence is invalid")
        _ensure_directory(child)
        found.append(child)
    return tuple(found)


def _record_digest(path: Path, label: str) -> str:
    return _digest(_read_regular(path, label, maximum=MAX_INPUT_FILE_BYTES))


def _load_json(path: Path, label: str) -> dict[str, Any]:
    try:
        value = json.loads(
            _read_regular(path, label, maximum=MAX_INPUT_FILE_BYTES).decode("utf-8")
        )
    except (UnicodeDecodeError, json.JSONDecodeError) as exc:
        raise ScoringError(f"{label} is invalid") from exc
    if not isinstance(value, dict):
        raise ScoringError(f"{label} is invalid")
    return value


def _contained(path: Path, root: Path) -> bool:
    try:
        path.resolve(strict=False).relative_to(root.resolve(strict=True))
    except (OSError, RuntimeError, ValueError):
        return False
    return True


def _locator_payload(locator: SupervisorLocator) -> dict[str, Any]:
    return {
        "supervisor_pid": locator.supervisor_pid,
        "start_identity": locator.start_identity,
        "socket_path": locator.socket_path,
        "challenge": locator.challenge,
        "control_dir": locator.control_dir,
        "created_at": locator.created_at,
    }


def _locator_public(locator: SupervisorLocator) -> dict[str, Any]:
    return {
        "supervisor_pid": locator.supervisor_pid,
        "start_identity": locator.start_identity,
        "socket_path": locator.socket_path,
        "control_dir": locator.control_dir,
        "challenge_digest": hashlib.sha256(
            locator.challenge.encode("utf-8")
        ).hexdigest(),
        "created_at": locator.created_at,
    }


def _validate_locator(
    locator: SupervisorLocator,
    blind_root: Path,
    *,
    control_target: Path | None = None,
) -> Path:
    if (
        not isinstance(locator, SupervisorLocator)
        or isinstance(locator.supervisor_pid, bool)
        or locator.supervisor_pid < 1
        or any(
            not isinstance(value, str) or not value
            for value in (
                locator.start_identity,
                locator.socket_path,
                locator.challenge,
                locator.control_dir,
                locator.created_at,
            )
        )
    ):
        raise ScoringError("evaluator locator is invalid")
    control = Path(locator.control_dir)
    socket = Path(locator.socket_path)
    if control_target is None:
        try:
            resolved_control = control.resolve(strict=True)
        except OSError as exc:
            raise ScoringError("evaluator control directory is unavailable") from exc
    else:
        try:
            resolved_control = control_target.resolve(strict=True)
        except OSError as exc:
            raise ScoringError("evaluator control directory is unavailable") from exc
    if (
        not control.is_absolute()
        or not socket.is_absolute()
        or socket.parent != control
        or not _contained(resolved_control, blind_root)
        or not _contained(resolved_control / socket.name, resolved_control)
    ):
        raise ScoringError("evaluator locator escapes blind workspace")
    if control.exists() or control.is_symlink():
        try:
            if control.resolve(strict=True) != resolved_control:
                raise ScoringError("evaluator control alias is invalid")
        except OSError as exc:
            raise ScoringError("evaluator control alias is invalid") from exc
    registered = _load_json(
        resolved_control / "locator.json", "registered evaluator locator"
    )
    if registered != _locator_payload(locator):
        raise ScoringError("registered evaluator locator is invalid")
    return resolved_control


def _publish_runner(
    score_root: Path,
    blind_root: Path,
    blind: BlindWorkspace,
    run: RunIdentity,
    attempt: Attempt,
    locator: SupervisorLocator,
    invocation_digest: str,
) -> None:
    if not isinstance(invocation_digest, str) or not DIGEST_RE.fullmatch(
        invocation_digest
    ):
        raise ScoringError("evaluator invocation digest is invalid")
    control_target = _validate_locator(locator, blind_root)
    control = Path(locator.control_dir)
    alias = control.parent if control != control_target else None
    if alias is not None:
        try:
            if not alias.is_symlink() or alias.resolve(strict=True) != control_target.parent:
                raise ScoringError("evaluator control alias is invalid")
        except OSError as exc:
            raise ScoringError("evaluator control alias is invalid") from exc
    record = {
        "record": "scoring-runner",
        "scoring_version": SCORING_VERSION,
        "run_id": run.run_id,
        "cell_id": attempt.identity.cell_id,
        "repetition": attempt.identity.repetition,
        "attempt": attempt.identity.attempt,
        "score_id": score_root.name,
        "blind_id": blind.blind_id,
        "session_identity": blind.session_identity,
        "spec_digest": invocation_digest,
        "control_target": str(control_target),
        "control_alias": "" if alias is None else str(alias),
        "locator": _locator_payload(locator),
    }
    _write_new(score_root / RUNNER_FILENAME, _json_bytes(record))


def _validate_runner(
    score_root: Path,
    blind_root: Path,
    allocation: Mapping[str, Any],
    run: RunIdentity,
    attempt: Attempt,
) -> tuple[dict[str, Any], SupervisorLocator, str] | None:
    path = score_root / RUNNER_FILENAME
    if not path.exists() and not path.is_symlink():
        return None
    value = _load_canonical(path, "evaluator runner")
    fields = {
        "record", "scoring_version", "run_id", "cell_id", "repetition",
        "attempt", "score_id", "blind_id", "session_identity", "spec_digest",
        "control_target", "control_alias", "locator",
    }
    raw_locator = value.get("locator")
    locator_fields = {
        "supervisor_pid", "start_identity", "socket_path", "challenge",
        "control_dir", "created_at",
    }
    if not isinstance(raw_locator, dict) or set(raw_locator) != locator_fields:
        raise ScoringError("evaluator runner is invalid")
    try:
        locator = SupervisorLocator(**raw_locator)
    except TypeError as exc:
        raise ScoringError("evaluator runner is invalid") from exc
    if (
        set(value) != fields
        or value["record"] != "scoring-runner"
        or value["scoring_version"] != SCORING_VERSION
        or value["run_id"] != run.run_id
        or value["cell_id"] != attempt.identity.cell_id
        or value["repetition"] != attempt.identity.repetition
        or value["attempt"] != attempt.identity.attempt
        or value["score_id"] != score_root.name
        or value["blind_id"] != allocation["blind_id"]
        or value["session_identity"] != allocation["session_identity"]
        or not isinstance(value["spec_digest"], str)
        or not DIGEST_RE.fullmatch(value["spec_digest"])
        or not isinstance(value["control_target"], str)
        or not value["control_target"]
        or not isinstance(value["control_alias"], str)
    ):
        raise ScoringError("evaluator runner is invalid")
    control_target = Path(value["control_target"])
    resolved_control = _validate_locator(
        locator, blind_root, control_target=control_target
    )
    if resolved_control != control_target.resolve(strict=True):
        raise ScoringError("evaluator runner control target is invalid")
    alias = value["control_alias"]
    if alias:
        alias_path = Path(alias)
        if Path(locator.control_dir).parent != alias_path:
            raise ScoringError("evaluator runner control alias is invalid")
        if alias_path.exists() or alias_path.is_symlink():
            try:
                if (
                    not alias_path.is_symlink()
                    or alias_path.resolve(strict=True) != resolved_control.parent
                ):
                    raise ScoringError("evaluator runner control alias is invalid")
            except OSError as exc:
                raise ScoringError("evaluator runner control alias is invalid") from exc
    elif Path(locator.control_dir) != resolved_control:
        raise ScoringError("evaluator runner control alias is invalid")
    return value, locator, _digest(_read_regular(path, "evaluator runner"))


def _validate_cleanup_receipt(
    locator: SupervisorLocator,
    *,
    expected_reason: str | None = None,
    control_target: Path | None = None,
) -> tuple[dict[str, Any], str]:
    control = control_target or Path(locator.control_dir)
    path = control / "cleanup-receipt.json"
    receipt = _load_json(path, "evaluator cleanup receipt")
    required = {
        "receipt_version", "supervisor_pid", "challenge_digest", "reason",
        "exit_code", "signal", "caller_launched", "cleanup_complete",
        "process_group_alive", "completed_at",
    }
    if (
        set(receipt) != required
        or receipt["receipt_version"] != 1
        or receipt["supervisor_pid"] != locator.supervisor_pid
        or receipt["challenge_digest"]
        != hashlib.sha256(locator.challenge.encode("utf-8")).hexdigest()
        or receipt["reason"] not in TERMINAL_REASONS
        or (expected_reason is not None and receipt["reason"] != expected_reason)
        or not isinstance(receipt["caller_launched"], bool)
        or receipt["cleanup_complete"] is not True
        or receipt["process_group_alive"] is not False
        or not isinstance(receipt["completed_at"], str)
    ):
        raise ScoringError("evaluator cleanup receipt is invalid")
    for name in ("exit_code", "signal"):
        if receipt[name] is not None and (
            isinstance(receipt[name], bool) or not isinstance(receipt[name], int)
        ):
            raise ScoringError("evaluator cleanup receipt is invalid")
    return receipt, _digest(_read_regular(path, "evaluator cleanup receipt"))


def _validate_lifecycle_binding(
    blind_root: Path,
    locator: SupervisorLocator,
    invocation_digest: str,
    *,
    control_target: Path | None = None,
) -> str | None:
    path = blind_root / "output" / "lifecycle-result.json"
    if not path.exists() and not path.is_symlink():
        return None
    value = _load_json(path, "evaluator lifecycle")
    if (
        value.get("record") != "result"
        or value.get("spec_digest") != invocation_digest
        or value.get("locator") != _locator_public(locator)
        or value.get("terminal_reason") not in TERMINAL_REASONS
        or value.get("cleanup_complete") is not True
        or value.get("process_group_alive") is not False
        or value.get("success")
        is not (value.get("terminal_reason") == "success")
    ):
        raise ScoringError("evaluator lifecycle binding is invalid")
    _validate_cleanup_receipt(
        locator,
        expected_reason=str(value["terminal_reason"]),
        control_target=control_target,
    )
    journal = blind_root / "output" / "lifecycle-journal.jsonl"
    try:
        lines = _read_regular(
            journal, "evaluator lifecycle journal", maximum=MAX_INPUT_FILE_BYTES
        ).decode("utf-8").splitlines()
        header = json.loads(lines[0])
        terminal = json.loads(lines[-1])
    except (IndexError, UnicodeDecodeError, json.JSONDecodeError) as exc:
        raise ScoringError("evaluator lifecycle journal is invalid") from exc
    if (
        not isinstance(header, dict)
        or header.get("record") != "header"
        or header.get("spec_digest") != invocation_digest
        or not isinstance(terminal, dict)
        or terminal.get("record") != "terminal"
        or terminal.get("terminal_reason") != value["terminal_reason"]
        or terminal.get("cleanup_complete") is not True
        or terminal.get("process_group_alive") is not False
    ):
        raise ScoringError("evaluator lifecycle journal is invalid")
    return _digest(_read_regular(path, "evaluator lifecycle"))


def _wait_post_cleanup_quiet(
    root: Path,
    *,
    lifecycle_validator: Callable[[], str | None] | None = None,
) -> str | None:
    """Wait for required publication and then one stable quiet interval."""
    deadline = time.monotonic() + _POST_CLEANUP_TIMEOUT_SECONDS
    quiet_since = time.monotonic()
    previous: tuple[tuple[str, int, int], ...] | None = None
    while True:
        snapshot: list[tuple[str, int, int]] = []
        for path in sorted(root.rglob("*")):
            try:
                info = os.lstat(path)
            except FileNotFoundError:
                # Atomic lifecycle publication uses short-lived staging files;
                # disappearance is itself a change and the next poll observes
                # the stable post-cleanup tree.
                continue
            except OSError as exc:
                raise ScoringError("evaluator post-cleanup state is unavailable") from exc
            if stat.S_ISREG(info.st_mode):
                snapshot.append(
                    (path.relative_to(root).as_posix(), info.st_size, info.st_mtime_ns)
                )
        current = tuple(snapshot)
        now = time.monotonic()
        if current != previous:
            previous = current
            quiet_since = now

        lifecycle_digest: str | None = None
        if lifecycle_validator is not None:
            journal = root / "output" / "lifecycle-journal.jsonl"
            result = root / "output" / "lifecycle-result.json"
            journal_published = journal.exists() or journal.is_symlink()
            result_published = result.exists() or result.is_symlink()
            if journal_published and result_published:
                lifecycle_digest = lifecycle_validator()
                if lifecycle_digest is None:
                    raise ScoringError(
                        "evaluator lifecycle publication is incomplete"
                    )
        if (
            now - quiet_since >= _POST_CLEANUP_QUIET_SECONDS
            and (lifecycle_validator is None or lifecycle_digest is not None)
        ):
            if lifecycle_validator is None:
                return None
            final_digest = lifecycle_validator()
            if final_digest != lifecycle_digest:
                raise ScoringError("evaluator lifecycle publication changed")
            return final_digest
        if now >= deadline:
            if lifecycle_validator is not None:
                raise ScoringError("evaluator lifecycle publication is incomplete")
            raise ScoringError("evaluator post-cleanup state did not quiesce")
        time.sleep(_POST_CLEANUP_POLL_SECONDS)


def _remove_cleaned_socket(
    locator: SupervisorLocator, control_target: Path
) -> None:
    socket = control_target / Path(locator.socket_path).name
    try:
        mode = os.lstat(socket).st_mode
    except FileNotFoundError:
        return
    except OSError as exc:
        raise ScoringError("cleaned evaluator control socket is unavailable") from exc
    if not stat.S_ISSOCK(mode):
        raise ScoringError("cleaned evaluator control socket is invalid")
    try:
        socket.unlink()
        _fsync_dir(control_target)
    except OSError as exc:
        raise ScoringError("cleaned evaluator control socket cleanup failed") from exc


def _recover_runner(
    blind_root: Path,
    locator: SupervisorLocator,
    invocation_digest: str,
    *,
    control_target: Path,
) -> tuple[str | None, str]:
    lifecycle = _validate_lifecycle_binding(
        blind_root,
        locator,
        invocation_digest,
        control_target=control_target,
    )
    if lifecycle is not None:
        _, receipt_digest = _validate_cleanup_receipt(
            locator, control_target=control_target
        )
        stable_lifecycle = _wait_post_cleanup_quiet(
            blind_root,
            lifecycle_validator=lambda: _validate_lifecycle_binding(
                blind_root,
                locator,
                invocation_digest,
                control_target=control_target,
            ),
        )
        if stable_lifecycle != lifecycle:
            raise ScoringError("evaluator lifecycle publication changed")
        _remove_cleaned_socket(locator, control_target)
        return stable_lifecycle, receipt_digest
    receipt_path = control_target / "cleanup-receipt.json"
    if receipt_path.exists() or receipt_path.is_symlink():
        _, receipt_digest = _validate_cleanup_receipt(
            locator, control_target=control_target
        )
        lifecycle = _wait_post_cleanup_quiet(
            blind_root,
            lifecycle_validator=lambda: _validate_lifecycle_binding(
                blind_root,
                locator,
                invocation_digest,
                control_target=control_target,
            ),
        )
        if lifecycle is None:
            raise ScoringError("evaluator lifecycle publication is incomplete")
        _remove_cleaned_socket(locator, control_target)
        return lifecycle, receipt_digest
    try:
        outcome = recover_invocation(locator, stop=True)
    except LifecycleRecoveryError as exc:
        try:
            receipt, receipt_digest = _validate_cleanup_receipt(
                locator, control_target=control_target
            )
        except ScoringError:
            raise ScoringError("evaluator recovery is unverified") from exc
        if receipt["reason"] not in {
            REASON_CONTROLLER_LOST,
            REASON_RECOVERED_STOP,
        }:
            raise ScoringError("evaluator recovery is unverified") from exc
    else:
        if not outcome.cleanup_complete or outcome.process_group_alive:
            raise ScoringError("evaluator cleanup is unverified")
        _, receipt_digest = _validate_cleanup_receipt(
            locator,
            expected_reason=outcome.reason,
            control_target=control_target,
        )
    _wait_post_cleanup_quiet(blind_root)
    _remove_cleaned_socket(locator, control_target)
    lifecycle = _validate_lifecycle_binding(
        blind_root,
        locator,
        invocation_digest,
        control_target=control_target,
    )
    return lifecycle, receipt_digest


def _release_runner_alias(runner: Mapping[str, Any]) -> None:
    raw = runner.get("control_alias")
    if not isinstance(raw, str) or not raw:
        return
    alias = Path(raw)
    target = Path(str(runner.get("control_target", ""))).parent
    try:
        mode = os.lstat(alias).st_mode
    except FileNotFoundError:
        return
    except OSError as exc:
        raise ScoringError("evaluator control alias is unavailable") from exc
    if not stat.S_ISLNK(mode):
        raise ScoringError("evaluator control alias is invalid")
    try:
        if alias.resolve(strict=True) != target.resolve(strict=True):
            raise ScoringError("evaluator control alias is invalid")
        alias.unlink()
        _fsync_dir(alias.parent)
    except OSError as exc:
        raise ScoringError("evaluator control alias cleanup failed") from exc


def _eligibility(manifest: Manifest, attempt: Attempt) -> tuple[bool, tuple[str, ...]]:
    if attempt.state not in TERMINAL_STATES:
        return False, ("lifecycle_running",)
    if attempt.state != "success":
        return False, (f"lifecycle_{attempt.state}",)
    try:
        web = load_web_validation(attempt.root, manifest=manifest)
    except WebValidationError as exc:
        raise ScoringError("web eligibility evidence is invalid") from exc
    reasons: list[str] = []
    if web.status != "passed":
        reasons.append(f"web_{web.status}")
        if web.record["reason"]:
            reasons.append(f"web_reason_{web.record['reason']}")
    gates = web.record["gates"]
    if [item["id"] for item in gates] != list(WEB_GATES):
        raise ScoringError("web eligibility gates are invalid")
    reasons.extend(
        f"gate_{item['id']}" for item in gates if not item["passed"]
    )
    return not reasons, tuple(reasons)


def _publish_unscored(
    run: RunIdentity, manifest: Manifest, attempt: Attempt, reasons: tuple[str, ...]
) -> None:
    root = _score_root(attempt, create=True)
    if _score_dirs(root):
        raise ScoringError("eligible scoring and unscored evidence conflict")
    path = root / UNSCORED_FILENAME
    record = {
        "record": "unscored",
        "scoring_version": SCORING_VERSION,
        "status": "unscored",
        "reasons": list(reasons),
        "manifest_digest": run.manifest_digest,
        "attempt_digest": _record_digest(
            Path(attempt.root) / "attempt.json", "execution attempt"
        ),
        "web_validation_digest": _record_digest(
            Path(attempt.root) / WEB_VALIDATION_FILENAME, "web validation"
        ),
    }
    raw = _json_bytes(record)
    if path.exists() or path.is_symlink():
        if _read_regular(path, "unscored evidence") != raw:
            raise ScoringError("unscored evidence is immutable")
        return
    _write_new(path, raw)


def _validate_unscored(
    run: RunIdentity, manifest: Manifest, attempt: Attempt
) -> bool:
    path = _score_root(attempt, create=False) / UNSCORED_FILENAME
    if not path.exists() and not path.is_symlink():
        return False
    eligible, reasons = _eligibility(manifest, attempt)
    if eligible:
        raise ScoringError("unscored evidence conflicts with eligibility")
    expected = {
        "record": "unscored",
        "scoring_version": SCORING_VERSION,
        "status": "unscored",
        "reasons": list(reasons),
        "manifest_digest": run.manifest_digest,
        "attempt_digest": _record_digest(
            Path(attempt.root) / "attempt.json", "execution attempt"
        ),
        "web_validation_digest": _record_digest(
            Path(attempt.root) / WEB_VALIDATION_FILENAME, "web validation"
        ),
    }
    if _load_canonical(path, "unscored evidence") != expected:
        raise ScoringError("unscored evidence is invalid")
    if any(key in expected for key in ("score", "total", "worksheet")):
        raise ScoringError("unscored evidence contains a score")
    return True


def _append_preflight(
    run: RunIdentity,
    cell: MatrixCell,
    observation: PreflightObservation,
) -> tuple[int, str, str]:
    if not isinstance(observation, PreflightObservation):
        raise ScoringError("evaluator preflight is invalid")
    try:
        validate_result(cell, observation.result)
        evidence = canonical_evidence_bytes(
            cell,
            observation.result,
            observation.endpoint_identity,
            observation.config_identity,
        )
    except Exception as exc:
        raise ScoringError("evaluator preflight is invalid") from exc
    root = Path(run.root) / "scoring-preflight"
    if root.exists() or root.is_symlink():
        _ensure_directory(root)
    else:
        _mkdir_new(root)
    children = sorted(root.iterdir())
    for index, child in enumerate(children, start=1):
        if child.name != f"preflight-{index:06d}.json":
            raise ScoringError("evaluator preflight sequence is invalid")
        _read_regular(child, "evaluator preflight")
    sequence = len(children) + 1
    path = root / f"preflight-{sequence:06d}.json"
    _write_new(path, evidence)
    return sequence, _digest(evidence), observation.result.status


def _blind_id(
    manifest_digest: str, ordinal: int, score_number: int, nonce: bytes
) -> str:
    material = (
        b"IOP-BENCH-BLIND-ID-V1\0"
        + manifest_digest.encode("ascii")
        + ordinal.to_bytes(8, "big")
        + score_number.to_bytes(8, "big")
        + nonce
    )
    return "blind-" + hashlib.sha256(material).hexdigest()[:32]


def _allocate_score(
    run: RunIdentity,
    manifest: Manifest,
    attempt: Attempt,
    ordinal: int,
    score_number: int,
    preflight_sequence: int,
    preflight_digest: str,
) -> tuple[Path, str, Path, str]:
    root = _score_root(attempt, create=True)
    score_id = f"score-{score_number:06d}"
    score_root = root / score_id
    _mkdir_new(score_root)
    nonce = secrets.token_bytes(32)
    blind_id = _blind_id(manifest.digest, ordinal, score_number, nonce)
    if not BLIND_ID_RE.fullmatch(blind_id):
        raise ScoringError("blind id allocation failed")
    blind_root = Path(run.root) / "blind" / blind_id
    blind_parent = blind_root.parent
    if blind_parent.exists() or blind_parent.is_symlink():
        _ensure_directory(blind_parent)
    else:
        _mkdir_new(blind_parent)
    _mkdir_new(blind_root)
    for name in ("input", "session", "output"):
        _mkdir_new(blind_root / name)

    relative_blind = f"blind/{blind_id}"
    session_identity = _digest(
        b"IOP-BENCH-SCORING-SESSION-V1\0" + nonce
    )
    allocation = {
        "record": "scoring-allocation",
        "scoring_version": SCORING_VERSION,
        "score_id": score_id,
        "rubric_version": manifest.rubric_version,
        "blind_id": blind_id,
        "blind_path": relative_blind,
        "session_identity": session_identity,
        "manifest_digest": manifest.digest,
        "evaluator": _evaluator_payload(manifest),
        "preflight_sequence": preflight_sequence,
        "preflight_digest": preflight_digest,
    }
    _write_new(score_root / ALLOCATION_FILENAME, _json_bytes(allocation))

    mappings = Path(run.root) / "blind-mappings"
    if mappings.exists() or mappings.is_symlink():
        _ensure_directory(mappings)
    else:
        _mkdir_new(mappings)
    mapping = {
        "record": "blind-mapping",
        "scoring_version": SCORING_VERSION,
        "blind_id": blind_id,
        "blind_path": relative_blind,
        "score_id": score_id,
        "attempt_ordinal": ordinal,
        "attempt": {
            "run_id": attempt.identity.run_id,
            "cell_id": attempt.identity.cell_id,
            "repetition": attempt.identity.repetition,
            "attempt": attempt.identity.attempt,
        },
        "nonce_digest": _digest(nonce),
    }
    _write_new(mappings / f"{blind_id}.json", _json_bytes(mapping))
    return score_root, blind_id, blind_root, session_identity


def _materialize_blind(
    manifest: Manifest,
    attempt: Attempt,
    blind_id: str,
    blind_root: Path,
    session_identity: str,
) -> BlindWorkspace:
    web = load_web_validation(attempt.root, manifest=manifest)
    if web.status != "passed" or not all(item["passed"] for item in web.record["gates"]):
        raise ScoringError("execution attempt is not eligible")
    identities = _identity_values(manifest, attempt)
    if _contains_identity(str(blind_root).encode("utf-8"), identities):
        raise ScoringError("blind path leaks execution identity")

    generated = {item["path"]: item for item in web.record["workspace"]["generated"]}
    files: list[tuple[str, bytes]] = []
    workspace_root = Path(attempt.root) / "workspace"
    for name in GENERATED_FILES:
        fact = generated.get(name)
        if not isinstance(fact, dict) or fact.get("state") != "regular":
            raise ScoringError("blind generated input is unavailable")
        data = _safe_source(workspace_root, name)
        if _digest(data) != fact["digest"] or len(data) != fact["size"]:
            raise ScoringError("blind generated input digest is invalid")
        files.append((f"input/{name}", data))

    image_assets = [
        item.workspace_path
        for item in manifest.fixture.assets
        if Path(item.workspace_path).suffix.lower() in IMAGE_SUFFIXES
    ]
    if len(image_assets) != 2:
        raise ScoringError("blind input requires exactly two local images")
    input_facts = {item["path"]: item for item in web.record["workspace"]["inputs"]}
    for relative in sorted(image_assets):
        fact = input_facts.get(relative)
        data = _safe_source(workspace_root, relative)
        if (
            not isinstance(fact, dict)
            or fact.get("state") != "regular"
            or _digest(data) != fact.get("digest")
            or len(data) != fact.get("size")
        ):
            raise ScoringError("blind image input digest is invalid")
        files.append((f"input/{relative}", data))

    for screenshot in web.record["screenshots"]:
        source = _relative(screenshot["file"])
        data = _safe_source(Path(attempt.root), source)
        if _digest(data) != screenshot["digest"] or len(data) != screenshot["size"]:
            raise ScoringError("blind screenshot digest is invalid")
        files.append((f"input/screenshots/{Path(source).name}", data))

    for relative, data in files:
        if _contains_identity(relative.encode("utf-8"), identities) or _contains_identity(data, identities):
            raise ScoringError("blind input leaks execution identity")
        _write_relative(blind_root, relative, data)
    digest = _input_digest(files)
    return BlindWorkspace(
        blind_id=blind_id,
        root=str(blind_root),
        input_dir=str(blind_root / "input"),
        session_dir=str(blind_root / "session"),
        output_dir=str(blind_root / "output"),
        input_digest=digest,
        session_identity=session_identity,
    )


def _prompt(blind: BlindWorkspace) -> bytes:
    categories = ", ".join(
        f"{ident} ({maximum})" for ident, maximum in RUBRIC_CATEGORIES
    )
    value = (
        "Evaluate the anonymous landing page using only the files under input/. "
        "Inspect index.html, styles.css, script.js, the two local images, and all "
        "screenshots. Do not use network access or infer the producing system. "
        f"Use exactly these categories: {categories}. "
        "Write exactly one JSON object to output/worksheet.json with fields "
        "rubric_version, categories, and total. Each category must have exactly "
        "id, max_score, score, and non-empty evidence. Use rubric_version "
        "landing-quality-v1, integer scores within each maximum, and total equal "
        "to their sum. Do not add automatic gate points or any extra field."
    )
    return value.encode("utf-8")


def _publish_input_record(score_root: Path, blind: BlindWorkspace) -> None:
    record = {
        "record": "scoring-input",
        "scoring_version": SCORING_VERSION,
        "blind_id": blind.blind_id,
        "input_digest": blind.input_digest,
        "session_identity": blind.session_identity,
    }
    _write_new(score_root / INPUT_FILENAME, _json_bytes(record))


def _lifecycle_digest(blind_root: Path) -> str | None:
    path = blind_root / "output" / "lifecycle-result.json"
    if not path.exists() and not path.is_symlink():
        return None
    return _record_digest(path, "evaluator lifecycle")


def _publish_failure(
    score_root: Path,
    blind_id: str,
    reason: str,
    *,
    lifecycle_digest: str | None = None,
    runner_digest: str | None = None,
    cleanup_receipt_digest: str | None = None,
    post_tree_digest: str,
) -> None:
    record = {
        "record": "scoring-result",
        "scoring_version": SCORING_VERSION,
        "status": "scoring_failed",
        "blind_id": blind_id,
        "reason": reason,
        "lifecycle_digest": lifecycle_digest,
        "runner_digest": runner_digest,
        "cleanup_receipt_digest": cleanup_receipt_digest,
        "post_tree_digest": post_tree_digest,
    }
    _write_new(score_root / RESULT_FILENAME, _json_bytes(record))


def _publish_success(
    score_root: Path,
    blind: BlindWorkspace,
    worksheet: Worksheet,
    lifecycle_digest: str,
    runner_digest: str,
    cleanup_receipt_digest: str,
    post_tree_digest: str,
) -> None:
    canonical = canonical_worksheet_bytes(worksheet)
    record = {
        "record": "scoring-result",
        "scoring_version": SCORING_VERSION,
        "status": "scored",
        "blind_id": blind.blind_id,
        "reason": "",
        "lifecycle_digest": lifecycle_digest,
        "runner_digest": runner_digest,
        "cleanup_receipt_digest": cleanup_receipt_digest,
        "post_tree_digest": post_tree_digest,
        "input_digest": blind.input_digest,
        "worksheet_digest": _digest(canonical),
        "worksheet": worksheet.as_dict(),
    }
    _write_new(score_root / RESULT_FILENAME, _json_bytes(record))


def _validate_allocation(
    path: Path,
    run: RunIdentity,
    manifest: Manifest,
    attempt: Attempt,
    score_id: str,
) -> dict[str, Any]:
    value = _load_canonical(path, "scoring allocation")
    if not isinstance(value.get("evaluator"), dict):
        raise ScoringError("scoring allocation is invalid")
    expected_fields = {
        "record", "scoring_version", "score_id", "rubric_version", "blind_id",
        "blind_path", "session_identity", "manifest_digest", "evaluator",
        "preflight_sequence", "preflight_digest",
    }
    if (
        set(value) != expected_fields
        or value["record"] != "scoring-allocation"
        or value["scoring_version"] != SCORING_VERSION
        or value["score_id"] != score_id
        or value["rubric_version"] != manifest.rubric_version
        or not isinstance(value["blind_id"], str)
        or not BLIND_ID_RE.fullmatch(value["blind_id"])
        or value["blind_path"] != f"blind/{value['blind_id']}"
        or not isinstance(value["session_identity"], str)
        or not DIGEST_RE.fullmatch(value["session_identity"])
        or value["manifest_digest"] != run.manifest_digest
        or value["evaluator"] != _evaluator_payload(manifest)
        or isinstance(value["preflight_sequence"], bool)
        or not isinstance(value["preflight_sequence"], int)
        or value["preflight_sequence"] < 1
        or not isinstance(value["preflight_digest"], str)
        or not DIGEST_RE.fullmatch(value["preflight_digest"])
    ):
        raise ScoringError("scoring allocation is invalid")
    blind = Path(run.root) / value["blind_path"]
    _ensure_directory(blind)
    for name in ("input", "session", "output"):
        _ensure_directory(blind / name)
    preflight_path = (
        Path(run.root)
        / "scoring-preflight"
        / f"preflight-{value['preflight_sequence']:06d}.json"
    )
    if _record_digest(preflight_path, "evaluator preflight") != value["preflight_digest"]:
        raise ScoringError("scoring preflight binding is invalid")
    mapping = _load_canonical(
        Path(run.root) / "blind-mappings" / f"{value['blind_id']}.json",
        "blind mapping",
    )
    if (
        set(mapping) != {
            "record", "scoring_version", "blind_id", "blind_path", "score_id",
            "attempt_ordinal", "attempt", "nonce_digest",
        }
        or mapping["record"] != "blind-mapping"
        or mapping["scoring_version"] != SCORING_VERSION
        or mapping["blind_id"] != value["blind_id"]
        or mapping["blind_path"] != value["blind_path"]
        or mapping["score_id"] != score_id
        or isinstance(mapping["attempt_ordinal"], bool)
        or not isinstance(mapping["attempt_ordinal"], int)
        or mapping["attempt_ordinal"] < 1
        or mapping["attempt"] != {
            "run_id": attempt.identity.run_id,
            "cell_id": attempt.identity.cell_id,
            "repetition": attempt.identity.repetition,
            "attempt": attempt.identity.attempt,
        }
        or not isinstance(mapping["nonce_digest"], str)
        or not DIGEST_RE.fullmatch(mapping["nonce_digest"])
    ):
        raise ScoringError("blind mapping is invalid")
    return value


def _blind_tree_digest(root: Path) -> str:
    _ensure_directory(root)
    files: list[tuple[str, bytes]] = []

    def restore_mode(
        path: Path, expected: os.stat_result, original_mode: int
    ) -> None:
        try:
            current = os.lstat(path)
            if (
                stat.S_ISLNK(current.st_mode)
                or (current.st_dev, current.st_ino)
                != (expected.st_dev, expected.st_ino)
            ):
                raise ScoringError("blind input path changed")
            os.chmod(path, original_mode, follow_symlinks=False)
            restored = os.lstat(path)
            if (
                (restored.st_dev, restored.st_ino)
                != (expected.st_dev, expected.st_ino)
                or stat.S_IMODE(restored.st_mode) != original_mode
            ):
                raise ScoringError("blind input mode restoration failed")
        except OSError as exc:
            raise ScoringError("blind input is unavailable") from exc

    def visit(
        directory: Path,
        expected: os.stat_result | None = None,
        depth: int = 0,
    ) -> None:
        if depth > 64:
            raise ScoringError("blind input tree is too deep")
        try:
            info = os.lstat(directory)
        except OSError as exc:
            raise ScoringError("blind input is unavailable") 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
            or (
                expected is not None
                and (info.st_dev, info.st_ino)
                != (expected.st_dev, expected.st_ino)
            )
        ):
            raise ScoringError("blind input path is invalid")
        original_mode = stat.S_IMODE(info.st_mode)
        temporary_mode = original_mode | stat.S_IRUSR | stat.S_IXUSR
        changed = temporary_mode != original_mode
        try:
            if changed:
                os.chmod(directory, temporary_mode, follow_symlinks=False)
            current = os.lstat(directory)
            if (
                (current.st_dev, current.st_ino) != (info.st_dev, info.st_ino)
                or stat.S_ISLNK(current.st_mode)
            ):
                raise ScoringError("blind input path changed")
            with os.scandir(directory) as iterator:
                children = sorted(iterator, key=lambda item: item.name)
            for entry in children:
                try:
                    child_info = entry.stat(follow_symlinks=False)
                except OSError as exc:
                    raise ScoringError("blind input is unavailable") from exc
                path = directory / entry.name
                if stat.S_ISDIR(child_info.st_mode) and not stat.S_ISLNK(
                    child_info.st_mode
                ):
                    visit(path, child_info, depth + 1)
                    continue
                if (
                    not stat.S_ISREG(child_info.st_mode)
                    or stat.S_ISLNK(child_info.st_mode)
                    or child_info.st_uid != current_uid
                ):
                    raise ScoringError("blind input path is invalid")
                file_mode = stat.S_IMODE(child_info.st_mode)
                readable_mode = file_mode | stat.S_IRUSR
                file_changed = readable_mode != file_mode
                try:
                    if file_changed:
                        os.chmod(path, readable_mode, follow_symlinks=False)
                    current_file = os.lstat(path)
                    if (
                        (current_file.st_dev, current_file.st_ino)
                        != (child_info.st_dev, child_info.st_ino)
                        or stat.S_ISLNK(current_file.st_mode)
                    ):
                        raise ScoringError("blind input path changed")
                    relative = path.relative_to(root.parent).as_posix()
                    files.append(
                        (
                            relative,
                            _read_regular(
                                path,
                                "blind input",
                                maximum=MAX_INPUT_FILE_BYTES,
                            ),
                        )
                    )
                    if len(files) > 100_000:
                        raise ScoringError("blind input tree has too many files")
                finally:
                    if file_changed:
                        restore_mode(path, child_info, file_mode)
        except OSError as exc:
            raise ScoringError("blind input is unavailable") from exc
        finally:
            if changed:
                restore_mode(directory, info, original_mode)

    visit(root)
    for relative, _data in files:
        if not relative:
            # Defensive only: every collected entry must be below ``root``.
            raise ScoringError("blind input path is invalid")
    return _input_digest(files)


def _freeze_input_tree(root: Path) -> None:
    _ensure_directory(root)
    directories: list[Path] = [root]
    for path in sorted(root.rglob("*")):
        try:
            mode = os.lstat(path).st_mode
        except OSError as exc:
            raise ScoringError("blind input is unavailable") from exc
        if stat.S_ISDIR(mode):
            if path.is_symlink():
                raise ScoringError("blind input path is invalid")
            directories.append(path)
        elif stat.S_ISREG(mode) and not path.is_symlink():
            try:
                os.chmod(path, 0o400, follow_symlinks=False)
            except OSError as exc:
                raise ScoringError("blind input could not be frozen") from exc
        else:
            raise ScoringError("blind input path is invalid")
    for directory in reversed(directories):
        try:
            os.chmod(directory, 0o500, follow_symlinks=False)
        except OSError as exc:
            raise ScoringError("blind input could not be frozen") from exc


def _validate_input_record(
    score_root: Path,
    allocation: Mapping[str, Any],
    run: RunIdentity,
    *,
    verify_tree: bool = True,
) -> str | None:
    path = score_root / INPUT_FILENAME
    if not path.exists() and not path.is_symlink():
        return None
    value = _load_canonical(path, "scoring input")
    if (
        set(value) != {
            "record", "scoring_version", "blind_id", "input_digest",
            "session_identity",
        }
        or value["record"] != "scoring-input"
        or value["scoring_version"] != SCORING_VERSION
        or value["blind_id"] != allocation["blind_id"]
        or value["session_identity"] != allocation["session_identity"]
        or not isinstance(value["input_digest"], str)
        or not DIGEST_RE.fullmatch(value["input_digest"])
    ):
        raise ScoringError("scoring input is invalid")
    if verify_tree:
        input_root = Path(run.root) / allocation["blind_path"] / "input"
        actual = _blind_tree_digest(input_root)
        if actual != value["input_digest"]:
            raise ScoringError("blind input changed after allocation")
        for candidate in (input_root, *sorted(input_root.rglob("*"))):
            try:
                mode = os.lstat(candidate).st_mode
            except OSError as exc:
                raise ScoringError("blind input is unavailable") from exc
            if mode & 0o222:
                raise ScoringError("blind input changed after allocation")
    return str(value["input_digest"])


def _result_status(
    score_root: Path,
    run: RunIdentity,
    manifest: Manifest,
    attempt: Attempt,
) -> str | None:
    allocation = _validate_allocation(
        score_root / ALLOCATION_FILENAME,
        run,
        manifest,
        attempt,
        score_root.name,
    )
    input_digest = _validate_input_record(
        score_root, allocation, run, verify_tree=False
    )
    blind_root = Path(run.root) / allocation["blind_path"]
    runner = _validate_runner(
        score_root, blind_root, allocation, run, attempt
    )
    path = score_root / RESULT_FILENAME
    if not path.exists() and not path.is_symlink():
        return None
    value = _load_canonical(path, "scoring result")
    common = {
        "record", "scoring_version", "status", "blind_id", "reason",
        "lifecycle_digest", "runner_digest", "cleanup_receipt_digest",
        "post_tree_digest",
    }
    status = value.get("status")
    if (
        status not in {"scored", "scoring_failed"}
        or value.get("record") != "scoring-result"
        or value.get("scoring_version") != SCORING_VERSION
        or value.get("blind_id") != allocation["blind_id"]
        or not isinstance(value.get("reason"), str)
        or (
            value.get("lifecycle_digest") is not None
            and (
                not isinstance(value["lifecycle_digest"], str)
                or not DIGEST_RE.fullmatch(value["lifecycle_digest"])
            )
        )
        or (
            value.get("runner_digest") is not None
            and (
                not isinstance(value["runner_digest"], str)
                or not DIGEST_RE.fullmatch(value["runner_digest"])
            )
        )
        or (
            value.get("cleanup_receipt_digest") is not None
            and (
                not isinstance(value["cleanup_receipt_digest"], str)
                or not DIGEST_RE.fullmatch(value["cleanup_receipt_digest"])
            )
        )
        or not isinstance(value.get("post_tree_digest"), str)
        or not DIGEST_RE.fullmatch(value["post_tree_digest"])
    ):
        raise ScoringError("scoring result is invalid")
    actual_post_tree = _blind_tree_digest(blind_root)
    if actual_post_tree != value["post_tree_digest"]:
        raise ScoringError("scoring post-tree changed")
    actual_runner = None if runner is None else runner[2]
    if value["runner_digest"] != actual_runner:
        raise ScoringError("scoring runner binding is invalid")
    actual_receipt: str | None = None
    actual_lifecycle: str | None = None
    if runner is not None:
        runner_record, locator, _ = runner
        control_target = Path(runner_record["control_target"])
        receipt_path = control_target / "cleanup-receipt.json"
        if receipt_path.exists() or receipt_path.is_symlink():
            _, actual_receipt = _validate_cleanup_receipt(
                locator, control_target=control_target
            )
        lifecycle_path = blind_root / "output" / "lifecycle-result.json"
        if lifecycle_path.exists() or lifecycle_path.is_symlink():
            actual_lifecycle = _validate_lifecycle_binding(
                blind_root,
                locator,
                str(runner_record["spec_digest"]),
                control_target=control_target,
            )
    if value["cleanup_receipt_digest"] != actual_receipt:
        raise ScoringError("scoring cleanup binding is invalid")
    if value["lifecycle_digest"] != actual_lifecycle:
        raise ScoringError("scoring lifecycle binding is invalid")
    if status == "scoring_failed":
        if set(value) != common or not value["reason"]:
            raise ScoringError("scoring failure is invalid")
        return status
    if set(value) != common | {
        "input_digest", "worksheet_digest", "worksheet",
    } or value["reason"]:
        raise ScoringError("scored result is invalid")
    try:
        worksheet = load_worksheet(
            Path(run.root) / allocation["blind_path"] / "output" / "worksheet.json"
        )
    except RubricError as exc:
        raise ScoringError("scored worksheet is invalid") from exc
    canonical = canonical_worksheet_bytes(worksheet)
    if (
        value["worksheet"] != worksheet.as_dict()
        or value["worksheet_digest"] != _digest(canonical)
        or input_digest is None
        or value["input_digest"] != input_digest
        or value["lifecycle_digest"] is None
        or value["runner_digest"] is None
        or value["cleanup_receipt_digest"] is None
    ):
        raise ScoringError("scored worksheet binding is invalid")
    _validate_input_record(score_root, allocation, run, verify_tree=True)
    return status


def _blind_from_allocation(
    run: RunIdentity,
    allocation: Mapping[str, Any],
    input_digest: str | None,
) -> BlindWorkspace:
    root = Path(run.root) / str(allocation["blind_path"])
    return BlindWorkspace(
        blind_id=str(allocation["blind_id"]),
        root=str(root),
        input_dir=str(root / "input"),
        session_dir=str(root / "session"),
        output_dir=str(root / "output"),
        input_digest=input_digest or _blind_tree_digest(root / "input"),
        session_identity=str(allocation["session_identity"]),
    )


def _finalize_adapter_evidence(
    adapter: ScoringAdapter, blind: BlindWorkspace
) -> ScoringEvidenceFinalization:
    try:
        finalized = adapter.finalize_evidence(blind)
    except Exception as exc:
        raise ScoringError("evaluator evidence finalization failed") from exc
    if (
        not isinstance(finalized, ScoringEvidenceFinalization)
        or not isinstance(finalized.safe, bool)
        or not isinstance(finalized.reason, str)
        or (finalized.safe and finalized.reason)
        or (
            not finalized.safe
            and finalized.reason
            not in {
                "runtime_secret_leak",
                "input_mutated",
                "evaluator_output_leak",
            }
        )
    ):
        raise ScoringError("evaluator evidence finalization is invalid")
    return finalized


def _evidence_digests(
    score_root: Path,
    blind_root: Path,
    allocation: Mapping[str, Any],
    run: RunIdentity,
    attempt: Attempt,
) -> tuple[str | None, str | None, str | None, str]:
    runner = _validate_runner(
        score_root, blind_root, allocation, run, attempt
    )
    if runner is None:
        runner_digest = lifecycle_digest = receipt_digest = None
    else:
        runner_record, locator, runner_digest = runner
        control_target = Path(runner_record["control_target"])
        lifecycle_path = blind_root / "output" / "lifecycle-result.json"
        lifecycle_digest = (
            _validate_lifecycle_binding(
                blind_root,
                locator,
                str(runner_record["spec_digest"]),
                control_target=control_target,
            )
            if lifecycle_path.exists() or lifecycle_path.is_symlink()
            else None
        )
        receipt_path = control_target / "cleanup-receipt.json"
        receipt_digest = (
            _validate_cleanup_receipt(
                locator, control_target=control_target
            )[1]
            if receipt_path.exists() or receipt_path.is_symlink()
            else None
        )
    return (
        lifecycle_digest,
        runner_digest,
        receipt_digest,
        _blind_tree_digest(blind_root),
    )


def _publish_current_failure(
    score_root: Path,
    blind: BlindWorkspace,
    allocation: Mapping[str, Any],
    run: RunIdentity,
    attempt: Attempt,
    reason: str,
) -> None:
    lifecycle, runner, receipt, post_tree = _evidence_digests(
        score_root, Path(blind.root), allocation, run, attempt
    )
    _publish_failure(
        score_root,
        blind.blind_id,
        reason,
        lifecycle_digest=lifecycle,
        runner_digest=runner,
        cleanup_receipt_digest=receipt,
        post_tree_digest=post_tree,
    )


def _complete_interrupted(
    adapter: ScoringAdapter,
    score_root: Path,
    run: RunIdentity,
    manifest: Manifest,
    attempt: Attempt,
) -> str:
    allocation = _validate_allocation(
        score_root / ALLOCATION_FILENAME,
        run,
        manifest,
        attempt,
        score_root.name,
    )
    status = _result_status(score_root, run, manifest, attempt)
    if status is not None:
        return status
    blind_root = Path(run.root) / allocation["blind_path"]
    input_digest = _validate_input_record(
        score_root, allocation, run, verify_tree=False
    )
    blind = _blind_from_allocation(run, allocation, input_digest)
    runner = _validate_runner(
        score_root, blind_root, allocation, run, attempt
    )
    if runner is not None:
        runner_record, locator, _ = runner
        _recover_runner(
            blind_root,
            locator,
            str(runner_record["spec_digest"]),
            control_target=Path(runner_record["control_target"]),
        )
        _release_runner_alias(runner_record)
    finalized = _finalize_adapter_evidence(adapter, blind)
    reason = "interrupted" if finalized.safe else finalized.reason
    _publish_current_failure(
        score_root, blind, allocation, run, attempt, reason
    )
    return "scoring_failed"


def _scan_visible_tree(root: Path, identities: ProducerIdentity) -> None:
    _ensure_directory(root)
    for path in sorted(root.rglob("*")):
        try:
            mode = os.lstat(path).st_mode
        except OSError as exc:
            raise ScoringError("evaluator-visible state is unavailable") from exc
        if stat.S_ISDIR(mode):
            if path.is_symlink():
                raise ScoringError("evaluator-visible path is invalid")
            continue
        if not stat.S_ISREG(mode) or path.is_symlink():
            raise ScoringError("evaluator-visible path is invalid")
        relative = _path_bytes(path.relative_to(root).as_posix())
        data = _read_regular(
            path, "evaluator-visible evidence", maximum=MAX_INPUT_FILE_BYTES
        )
        if _contains_identity(relative, identities) or _contains_identity(
            data, identities
        ):
            raise ScoringError("evaluator-visible evidence leaks execution identity")


def _score_one(
    adapter: ScoringAdapter,
    run: RunIdentity,
    manifest: Manifest,
    attempt: Attempt,
    ordinal: int,
    score_number: int,
    preflight_sequence: int,
    preflight_digest: str,
) -> str:
    score_root, blind_id, blind_root, session_identity = _allocate_score(
        run,
        manifest,
        attempt,
        ordinal,
        score_number,
        preflight_sequence,
        preflight_digest,
    )
    allocation = _validate_allocation(
        score_root / ALLOCATION_FILENAME,
        run,
        manifest,
        attempt,
        score_root.name,
    )
    try:
        blind = _materialize_blind(
            manifest, attempt, blind_id, blind_root, session_identity
        )
        _publish_input_record(score_root, blind)
        _freeze_input_tree(Path(blind.input_dir))
    except Exception:
        _publish_failure(
            score_root,
            blind_id,
            "blind_preparation_failed",
            post_tree_digest=_blind_tree_digest(blind_root),
        )
        return "scoring_failed"

    prompt = _prompt(blind)
    identities = _identity_values(manifest, attempt)
    if _contains_identity(prompt, identities):
        _publish_current_failure(
            score_root, blind, allocation, run, attempt, "blind_prompt_leak"
        )
        return "scoring_failed"
    cell = _evaluator_cell(manifest)
    invocation: ScoringInvocationResult | None = None
    invocation_failed = False

    def on_started(locator: SupervisorLocator, invocation_digest: str) -> None:
        _publish_runner(
            score_root,
            blind_root,
            blind,
            run,
            attempt,
            locator,
            invocation_digest,
        )

    try:
        invocation = adapter.invoke(
            cell, blind, prompt, manifest.timeout, on_started
        )
    except Exception:
        invocation_failed = True

    runner = _validate_runner(
        score_root, blind_root, allocation, run, attempt
    )
    if runner is not None:
        runner_record, locator, _ = runner
        _recover_runner(
            blind_root,
            locator,
            str(runner_record["spec_digest"]),
            control_target=Path(runner_record["control_target"]),
        )
        _release_runner_alias(runner_record)
    finalized = _finalize_adapter_evidence(adapter, blind)
    if not finalized.safe:
        _publish_current_failure(
            score_root, blind, allocation, run, attempt, finalized.reason
        )
        return "scoring_failed"
    try:
        _validate_input_record(score_root, allocation, run, verify_tree=True)
    except ScoringError:
        _publish_current_failure(
            score_root, blind, allocation, run, attempt, "input_mutated"
        )
        return "scoring_failed"
    if invocation_failed:
        _publish_current_failure(
            score_root, blind, allocation, run, attempt, "evaluator_failed"
        )
        return "scoring_failed"
    if runner is None:
        _publish_current_failure(
            score_root, blind, allocation, run, attempt, "evaluator_owner_missing"
        )
        return "scoring_failed"
    if not isinstance(invocation, ScoringInvocationResult):
        _publish_current_failure(
            score_root, blind, allocation, run, attempt, "evaluator_protocol_failed"
        )
        return "scoring_failed"
    expected_binding = (
        manifest.evaluator.iop.route_kind,
        manifest.evaluator.iop.route_id,
        manifest.evaluator.iop.request_model,
        manifest.evaluator.iop.requested_effort,
    )
    lifecycle, runner_digest, receipt_digest, _ = _evidence_digests(
        score_root, blind_root, allocation, run, attempt
    )
    if (
        not invocation.success
        or invocation.terminal_reason != "success"
        or invocation.effective_binding != expected_binding
        or lifecycle is None
        or runner_digest is None
        or receipt_digest is None
    ):
        _publish_current_failure(
            score_root, blind, allocation, run, attempt, "evaluator_failed"
        )
        return "scoring_failed"
    try:
        _scan_visible_tree(blind_root, identities)
    except ScoringError:
        _publish_current_failure(
            score_root, blind, allocation, run, attempt, "evaluator_output_leak"
        )
        return "scoring_failed"
    try:
        worksheet = load_worksheet(blind_root / "output" / "worksheet.json")
        post_tree = _blind_tree_digest(blind_root)
        _publish_success(
            score_root,
            blind,
            worksheet,
            lifecycle,
            runner_digest,
            receipt_digest,
            post_tree,
        )
    except (RubricError, ScoringError):
        _publish_current_failure(
            score_root, blind, allocation, run, attempt, "invalid_worksheet"
        )
        return "scoring_failed"
    return "scored"


def score_run(
    store: RunStore,
    run: RunIdentity,
    manifest: Manifest,
    *,
    adapter: ScoringAdapter,
    retry_scoring_failed: bool = False,
) -> ScoringSummary:
    """Classify every execution attempt and append only explicitly allowed work."""
    if not isinstance(store, RunStore) or not isinstance(manifest, Manifest):
        raise ScoringError("scoring inputs are invalid")
    if not callable(getattr(adapter, "preflight", None)) or not callable(
        getattr(adapter, "invoke", None)
    ) or not callable(getattr(adapter, "finalize_evidence", None)):
        raise ScoringError("scoring adapter is unavailable")
    bound = store.open(manifest, run.run_id)
    if bound != run:
        raise ScoringError("run identity is invalid")

    counts = {status: 0 for status in SCORING_STATUSES}
    with store.writer(bound):
        retained = store.execution_attempts(bound, manifest)
        retained_by_slot = {
            (item.identity.cell_id, item.identity.repetition) for item in retained
        }
        counts["blocked"] += len(store.slots(manifest)) - len(retained_by_slot)

        pending: list[tuple[int, Attempt, int]] = []
        for ordinal, attempt in enumerate(retained, start=1):
            eligible, reasons = _eligibility(manifest, attempt)
            if not eligible:
                if reasons == ("lifecycle_running",):
                    counts["blocked"] += 1
                    continue
                _publish_unscored(bound, manifest, attempt, reasons)
                _validate_unscored(bound, manifest, attempt)
                counts["unscored"] += 1
                continue
            if _validate_unscored(bound, manifest, attempt):
                raise ScoringError("eligible attempt is marked unscored")
            score_root = _score_root(attempt, create=False)
            score_dirs = _score_dirs(score_root)
            statuses = [
                _complete_interrupted(adapter, item, bound, manifest, attempt)
                for item in score_dirs
            ]
            if "scored" in statuses:
                if statuses[-1] != "scored" or statuses.count("scored") != 1:
                    raise ScoringError("successful scoring is not terminal")
                counts["scored"] += 1
                continue
            if statuses and not retry_scoring_failed:
                counts["scoring_failed"] += 1
                continue
            pending.append((ordinal, attempt, len(score_dirs) + 1))

        if not pending:
            return ScoringSummary(bound.run_id, **counts)

        cell = _evaluator_cell(manifest)
        observation = adapter.preflight(cell)
        sequence, preflight_digest, status = _append_preflight(
            bound, cell, observation
        )
        if status != "ready":
            counts["blocked"] += len(pending)
            return ScoringSummary(bound.run_id, **counts)

        for ordinal, attempt, score_number in pending:
            outcome = _score_one(
                adapter,
                bound,
                manifest,
                attempt,
                ordinal,
                score_number,
                sequence,
                preflight_digest,
            )
            counts[outcome] += 1
    return ScoringSummary(bound.run_id, **counts)