orchestrate.py

#!/usr/bin/env python3
"""Read-only inventory and non-privileged UMSH test orchestration.

The orchestrator never invokes sudo or changes kernel/module state.  Commands
which require such setup must remain explicit unsupported/external_setup rows
in the manifest and be prepared outside this program.
"""

from __future__ import annotations

import argparse
import csv
import dataclasses
import datetime as dt
import hashlib
import io
import itertools
import json
import math
import os
import posixpath
import re
import shlex
import subprocess
import sys
import time
from collections import Counter
from pathlib import Path
from typing import Any, Iterable, Mapping, Sequence


SCHEMA_VERSION = 2
HARNESS_VERSION = "2"
SAFE_ID = re.compile(r"^[A-Za-z0-9][A-Za-z0-9_.-]*$")
CASE_STATES = frozenset(("ready", "unsupported", "external_setup"))
SUITES = frozenset(("correctness", "performance"))
PERFORMANCE_DIRECTIONS = frozenset(
    (
        "bind",
        "cpu_read",
        "cpu_write",
        "gpu_read",
        "gpu_write",
        "gpu_contention_read",
        "io_to_gpu",
    )
)
CLASSIFICATION_SOURCES = frozenset(("csv", "sidecar", "csv+sidecar", "not_applicable"))
FORBIDDEN_EXECUTABLES = frozenset(("sudo", "su", "insmod", "rmmod", "modprobe"))
DEFAULT_MANIFEST = Path(__file__).with_name("platforms.json")
REPO_ROOT = Path(__file__).resolve().parents[2]


DIRTY_HASH_PROGRAM = r"""
import hashlib
import os
import stat
import subprocess

h = hashlib.sha256()
h.update(b"tracked-diff\0")
h.update(subprocess.check_output(["git", "diff", "--binary", "HEAD"]))
untracked = subprocess.check_output(
    ["git", "ls-files", "--others", "--exclude-standard", "-z"]
).split(b"\0")
for raw_path in sorted(path for path in untracked if path):
    path = os.fsdecode(raw_path)
    h.update(b"untracked\0")
    h.update(len(raw_path).to_bytes(8, "big"))
    h.update(raw_path)
    metadata = os.lstat(path)
    h.update(stat.S_IFMT(metadata.st_mode).to_bytes(8, "big"))
    if stat.S_ISLNK(metadata.st_mode):
        h.update(os.fsencode(os.readlink(path)))
    elif stat.S_ISREG(metadata.st_mode):
        with open(path, "rb") as stream:
            while True:
                chunk = stream.read(1024 * 1024)
                if not chunk:
                    break
                h.update(chunk)
print(h.hexdigest())
""".strip()


HUGEPAGE_PROGRAM = r"""
from pathlib import Path

roots = [Path("/sys/kernel/mm/hugepages"), Path("/sys/kernel/mm/transparent_hugepage")]
for root in roots:
    if not root.exists():
        print(f"MISSING {root}")
        continue
    for path in sorted(item for item in root.rglob("*") if item.is_file()):
        try:
            value = path.read_text(encoding="utf-8", errors="replace").strip()
        except OSError as error:
            value = f"ERROR {error}"
        print(f"{path}={value}")
""".strip()


MODULE_PARAMETER_PROGRAM = r"""
from pathlib import Path

for module in ("i915", "xe", "nvidia", "nvidia_uvm", "uc_mem", "wbinvd_mod", "arm64_attr_mmap"):
    root = Path("/sys/module") / module
    if not root.exists():
        continue
    print(f"MODULE {module}")
    for relative in ("version", "srcversion", "taint"):
        path = root / relative
        if path.is_file():
            try:
                print(f"{path}={path.read_text(errors='replace').strip()}")
            except OSError as error:
                print(f"{path}=ERROR {error}")
    parameters = root / "parameters"
    if parameters.is_dir():
        for path in sorted(item for item in parameters.iterdir() if item.is_file()):
            try:
                value = path.read_text(errors="replace")[:4096].strip()
            except OSError as error:
                value = f"ERROR {error}"
            print(f"{path}={value}")
""".strip()


MODULE_FILE_HASH_PROGRAM = r"""
import hashlib
import subprocess
from pathlib import Path

for module in ("i915", "xe", "nvidia", "nvidia_uvm", "uc_mem", "wbinvd_mod", "arm64_attr_mmap"):
    result = subprocess.run(["modinfo", "-n", module], capture_output=True, text=True)
    if result.returncode != 0:
        continue
    path = Path(result.stdout.strip())
    if not path.is_file():
        print(f"{module}\t{path}\tMISSING")
        continue
    digest = hashlib.sha256()
    with path.open("rb") as stream:
        while True:
            chunk = stream.read(1024 * 1024)
            if not chunk:
                break
            digest.update(chunk)
    print(f"{module}\t{path}\t{digest.hexdigest()}")
""".strip()


@dataclasses.dataclass
class CommandResult:
    returncode: int
    stdout: bytes
    stderr: bytes
    duration_seconds: float
    timed_out: bool = False


class ManifestError(ValueError):
    pass


class Transport:
    def __init__(self, platform: Mapping[str, Any]):
        self.kind = str(platform["transport"])
        self.host = platform.get("host")
        self.ssh_options = [str(value) for value in platform.get("ssh_options", [])]

    def describe(self) -> str:
        return "local" if self.kind == "local" else f"ssh:{self.host}"

    def run(
        self,
        command: Sequence[str],
        *,
        cwd: str | None = None,
        timeout: float | None = None,
    ) -> CommandResult:
        logical = [str(item) for item in command]
        if self.kind == "local":
            actual = logical
            local_cwd = cwd
        else:
            remote = shlex.join(logical)
            if cwd:
                remote = f"cd -- {shlex.quote(cwd)} && exec {remote}"
            else:
                remote = f"exec {remote}"
            actual = ["ssh", *self.ssh_options, str(self.host), remote]
            local_cwd = None

        started = time.monotonic()
        try:
            completed = subprocess.run(
                actual,
                cwd=local_cwd,
                stdout=subprocess.PIPE,
                stderr=subprocess.PIPE,
                timeout=timeout,
                check=False,
            )
            return CommandResult(
                returncode=completed.returncode,
                stdout=completed.stdout,
                stderr=completed.stderr,
                duration_seconds=time.monotonic() - started,
            )
        except subprocess.TimeoutExpired as error:
            return CommandResult(
                returncode=124,
                stdout=error.stdout or b"",
                stderr=(error.stderr or b"") + b"\nORCHESTRATOR_TIMEOUT\n",
                duration_seconds=time.monotonic() - started,
                timed_out=True,
            )
        except (FileNotFoundError, NotADirectoryError) as error:
            return CommandResult(
                returncode=127,
                stdout=b"",
                stderr=f"{type(error).__name__}: {error}\n".encode(),
                duration_seconds=time.monotonic() - started,
            )

    def read_file(self, path: str, *, timeout: float = 60.0) -> CommandResult:
        if self.kind == "local":
            started = time.monotonic()
            try:
                payload = Path(path).read_bytes()
                return CommandResult(0, payload, b"", time.monotonic() - started)
            except OSError as error:
                return CommandResult(
                    1,
                    b"",
                    f"{type(error).__name__}: {error}\n".encode(),
                    time.monotonic() - started,
                )
        return self.run(["cat", "--", path], timeout=timeout)

    def archive_directory(self, path: str, *, timeout: float = 120.0) -> CommandResult:
        parent = posixpath.dirname(path.rstrip("/")) or "."
        name = posixpath.basename(path.rstrip("/"))
        return self.run(
            ["tar", "-C", parent, "-cf", "-", "--", name], timeout=timeout
        )


def utc_now() -> str:
    return dt.datetime.now(dt.timezone.utc).isoformat().replace("+00:00", "Z")


def default_run_id() -> str:
    stamp = dt.datetime.now(dt.timezone.utc).strftime("%Y%m%dT%H%M%SZ")
    nonce = hashlib.sha256(os.urandom(16)).hexdigest()[:8]
    return f"{stamp}-{nonce}"


def decode(payload: bytes) -> str:
    return payload.decode("utf-8", errors="replace")


def sha256_bytes(payload: bytes) -> str:
    return hashlib.sha256(payload).hexdigest()


def sha256_file(path: Path) -> str:
    digest = hashlib.sha256()
    with path.open("rb") as stream:
        while True:
            chunk = stream.read(1024 * 1024)
            if not chunk:
                break
            digest.update(chunk)
    return digest.hexdigest()


def write_bytes(path: Path, payload: bytes) -> dict[str, Any]:
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_bytes(payload)
    return {"path": str(path), "bytes": len(payload), "sha256": sha256_bytes(payload)}


def write_json(path: Path, value: Any) -> None:
    payload = (json.dumps(value, indent=2, sort_keys=True) + "\n").encode()
    write_bytes(path, payload)


def safe_id(value: str, description: str) -> None:
    if not SAFE_ID.fullmatch(value):
        raise ManifestError(f"invalid {description} {value!r}")


def load_manifest(path: Path) -> tuple[dict[str, Any], bytes]:
    try:
        payload = path.read_bytes()
        manifest = json.loads(payload)
    except (OSError, json.JSONDecodeError) as error:
        raise ManifestError(f"cannot load manifest {path}: {error}") from error
    if not isinstance(manifest, dict):
        raise ManifestError("manifest root must be an object")
    validate_manifest(manifest)
    return manifest, payload


def resolve_platform_cases(
    manifest: Mapping[str, Any], platform_id: str
) -> list[dict[str, Any]]:
    platform = manifest["platforms"][platform_id]
    catalog = manifest["case_catalog"]
    resolved: list[dict[str, Any]] = []
    for entry in platform["cases"]:
        if isinstance(entry, str):
            case = dict(catalog[entry])
        else:
            case = {}
            reference = entry.get("ref")
            if reference:
                case.update(catalog[reference])
            case.update({key: value for key, value in entry.items() if key != "ref"})
        resolved.append(case)
    return resolved


def validate_command(command: Sequence[Any], context: str) -> None:
    if not command or not all(isinstance(item, str) and item for item in command):
        raise ManifestError(f"{context}: ready command must be a non-empty string list")
    for token in command:
        executable = posixpath.basename(token)
        if executable in FORBIDDEN_EXECUTABLES:
            raise ManifestError(f"{context}: forbidden privileged executable {token!r}")
        if "/sys/module/" in token or "/proc/sys/" in token:
            raise ManifestError(f"{context}: mutable kernel path is forbidden in test command")


def validate_performance_metadata(
    case: Mapping[str, Any], context: str, csv_columns: set[str]
) -> None:
    metadata = case.get("performance_metadata")
    if not isinstance(metadata, dict):
        raise ManifestError(f"{context}: ready performance case requires performance_metadata")
    direction = metadata.get("direction")
    if direction not in PERFORMANCE_DIRECTIONS:
        raise ManifestError(f"{context}: invalid performance direction {direction!r}")
    result_scope = metadata.get("result_scope")
    if not isinstance(result_scope, str) or not result_scope:
        raise ManifestError(f"{context}: performance result_scope is required")
    if result_scope != case.get("result_scope"):
        raise ManifestError(
            f"{context}: performance result_scope must match the case result_scope"
        )
    timing_boundary = metadata.get("timing_boundary")
    if not isinstance(timing_boundary, str) or not timing_boundary:
        raise ManifestError(f"{context}: performance timing_boundary is required")

    units = metadata.get("units")
    if not isinstance(units, dict) or not units:
        raise ManifestError(f"{context}: performance units must be a non-empty object")
    for metric, unit in units.items():
        if not isinstance(metric, str) or metric not in csv_columns:
            raise ManifestError(
                f"{context}: unit metric {metric!r} is not a required CSV column"
            )
        if not isinstance(unit, str) or not unit:
            raise ManifestError(f"{context}: unit for {metric!r} must be a string")

    for field in ("binding_classification", "policy_classification"):
        classification = metadata.get(field)
        if not isinstance(classification, dict):
            raise ManifestError(f"{context}: {field} must be an object")
        source = classification.get("source")
        if source not in CLASSIFICATION_SOURCES:
            raise ManifestError(f"{context}: invalid {field} source {source!r}")
        value = classification.get("value")
        if not isinstance(value, str) or not value:
            raise ManifestError(f"{context}: {field} value is required")
        columns = classification.get("columns")
        if not isinstance(columns, list) or not all(
            isinstance(column, str) and column for column in columns
        ):
            raise ManifestError(f"{context}: {field} columns must be a string list")
        if len(columns) != len(set(columns)):
            raise ManifestError(f"{context}: {field} columns must be unique")
        if source in ("csv", "csv+sidecar") and not columns:
            raise ManifestError(f"{context}: {field} CSV source requires columns")
        if source in ("sidecar", "not_applicable") and columns:
            raise ManifestError(
                f"{context}: {field} source {source!r} cannot declare CSV columns"
            )
        missing = sorted(set(columns) - csv_columns)
        if missing:
            raise ManifestError(
                f"{context}: {field} columns are not required by CSV artifacts: "
                f"{', '.join(missing)}"
            )


def validate_csv_contract(
    artifact: Mapping[str, Any],
    context: str,
    required_columns: Sequence[str],
    *,
    performance: bool,
) -> None:
    required = set(required_columns)
    expected_rows = artifact.get("expected_rows")
    if expected_rows is not None:
        if not isinstance(expected_rows, dict) or not expected_rows:
            raise ManifestError(f"{context}: CSV expected_rows must be a non-empty object")
        unknown = sorted(set(expected_rows) - {"exact", "minimum", "maximum"})
        if unknown:
            raise ManifestError(
                f"{context}: CSV expected_rows has unknown keys: {', '.join(unknown)}"
            )
        for key, value in expected_rows.items():
            if isinstance(value, bool) or not isinstance(value, int) or value < 0:
                raise ManifestError(
                    f"{context}: CSV expected_rows {key!r} must be a non-negative integer"
                )
        if "exact" in expected_rows and len(expected_rows) != 1:
            raise ManifestError(
                f"{context}: CSV expected_rows exact cannot be combined with a range"
            )
        if expected_rows.get("minimum", 0) > expected_rows.get("maximum", sys.maxsize):
            raise ManifestError(f"{context}: CSV expected_rows minimum exceeds maximum")

    unique_coordinates = artifact.get("unique_coordinates")
    if unique_coordinates is not None:
        if not isinstance(unique_coordinates, list) or not unique_coordinates or not all(
            isinstance(column, str) and column for column in unique_coordinates
        ):
            raise ManifestError(
                f"{context}: CSV unique_coordinates must be a non-empty string list"
            )
        if len(unique_coordinates) != len(set(unique_coordinates)):
            raise ManifestError(f"{context}: CSV unique_coordinates must be unique")
        missing = sorted(set(unique_coordinates) - required)
        if missing:
            raise ManifestError(
                f"{context}: CSV unique_coordinates are not required columns: "
                f"{', '.join(missing)}"
            )

    numeric_columns = artifact.get("numeric_columns")
    if numeric_columns is not None:
        if not isinstance(numeric_columns, dict) or not numeric_columns:
            raise ManifestError(f"{context}: CSV numeric_columns must be a non-empty object")
        for column, constraints in numeric_columns.items():
            if column not in required:
                raise ManifestError(f"{context}: numeric column {column!r} is not required")
            if not isinstance(constraints, dict) or not constraints:
                raise ManifestError(
                    f"{context}: numeric column {column!r} needs constraints"
                )
            unknown = sorted(set(constraints) - {"finite", "positive"})
            if unknown:
                raise ManifestError(
                    f"{context}: numeric column {column!r} has unknown constraints: "
                    f"{', '.join(unknown)}"
                )
            if not all(isinstance(value, bool) for value in constraints.values()):
                raise ManifestError(
                    f"{context}: numeric constraints for {column!r} must be booleans"
                )
            if not any(constraints.values()):
                raise ManifestError(
                    f"{context}: numeric column {column!r} has no enabled constraint"
                )

    expected_matrix = artifact.get("expected_matrix")
    if expected_matrix is not None:
        if not isinstance(expected_matrix, dict) or not expected_matrix:
            raise ManifestError(
                f"{context}: CSV expected_matrix must be a non-empty object"
            )
        for column, values in expected_matrix.items():
            if column not in required:
                raise ManifestError(
                    f"{context}: expected_matrix column {column!r} is not required"
                )
            if not isinstance(values, list) or not values or not all(
                isinstance(value, str) for value in values
            ):
                raise ManifestError(
                    f"{context}: expected_matrix values for {column!r} must be "
                    "a non-empty string list"
                )
            if len(values) != len(set(values)):
                raise ManifestError(
                    f"{context}: expected_matrix values for {column!r} must be unique"
                )
        if unique_coordinates != list(expected_matrix):
            raise ManifestError(
                f"{context}: unique_coordinates must exactly match expected_matrix keys"
            )
        matrix_rows = math.prod(len(values) for values in expected_matrix.values())
        if expected_rows is not None and expected_rows.get("exact") != matrix_rows:
            raise ManifestError(
                f"{context}: expected_rows.exact must equal expected_matrix size "
                f"{matrix_rows}"
            )

    row_rules = artifact.get("row_rules")
    if row_rules is not None:
        if not isinstance(row_rules, list) or not row_rules:
            raise ManifestError(f"{context}: CSV row_rules must be a non-empty list")
        for index, rule in enumerate(row_rules):
            if not isinstance(rule, dict) or set(rule) != {"when", "require"}:
                raise ManifestError(
                    f"{context}: row_rules[{index}] needs only when and require"
                )
            for field in ("when", "require"):
                values = rule[field]
                if not isinstance(values, dict) or not values:
                    raise ManifestError(
                        f"{context}: row_rules[{index}].{field} must be an object"
                    )
                for column, accepted in values.items():
                    if column not in required:
                        raise ManifestError(
                            f"{context}: row_rules[{index}] column {column!r} "
                            "is not required"
                        )
                    if isinstance(accepted, str):
                        accepted = [accepted]
                    if not isinstance(accepted, list) or not accepted or not all(
                        isinstance(value, str) for value in accepted
                    ):
                        raise ManifestError(
                            f"{context}: row_rules[{index}] values for {column!r} "
                            "must be a string or non-empty string list"
                        )

    invalid_values = artifact.get("evidence_invalid_values")
    if invalid_values is not None:
        if not isinstance(invalid_values, dict) or not invalid_values:
            raise ManifestError(
                f"{context}: CSV evidence_invalid_values must be a non-empty object"
            )
        for column, values in invalid_values.items():
            if column not in required:
                raise ManifestError(
                    f"{context}: evidence-invalid column {column!r} is not required"
                )
            if not isinstance(values, list) or not values or not all(
                isinstance(value, str) for value in values
            ):
                raise ManifestError(
                    f"{context}: evidence-invalid values for {column!r} must be "
                    "a non-empty string list"
                )
            if len(values) != len(set(values)):
                raise ManifestError(
                    f"{context}: evidence-invalid values for {column!r} must be unique"
                )

    if performance:
        if expected_rows is None and unique_coordinates is None:
            raise ManifestError(
                f"{context}: required performance CSV needs expected_rows or "
                "unique_coordinates"
            )
        if numeric_columns is None:
            raise ManifestError(f"{context}: required performance CSV needs numeric_columns")
        if not isinstance(invalid_values, dict) or "Failure" not in invalid_values.get(
            "status", []
        ):
            raise ManifestError(
                f"{context}: required performance CSV must mark status=Failure "
                "as evidence-invalid"
            )


def validate_case_artifacts(
    case: Mapping[str, Any], context: str
) -> tuple[set[str], bool]:
    outputs = case.get("artifacts", [])
    if not isinstance(outputs, list):
        raise ManifestError(f"{context}: artifacts must be a list")
    artifact_ids: set[str] = set()
    performance_csv_columns: set[str] = set()
    required_performance_csv = False
    for artifact in outputs:
        if not isinstance(artifact, dict):
            raise ManifestError(f"{context}: artifact must be an object")
        artifact_id = artifact.get("id")
        if not isinstance(artifact_id, str):
            raise ManifestError(f"{context}: artifact id is required")
        safe_id(artifact_id, "artifact id")
        if artifact_id in artifact_ids:
            raise ManifestError(f"{context}: duplicate artifact id {artifact_id!r}")
        artifact_ids.add(artifact_id)
        if artifact.get("kind", "file") not in ("file", "directory"):
            raise ManifestError(f"{context}: invalid artifact kind")
        if not isinstance(artifact.get("path"), str):
            raise ManifestError(f"{context}: artifact path is required")
        if artifact.get("format") not in (None, "csv", "json"):
            raise ManifestError(f"{context}: unsupported artifact format")
        if artifact.get("format") == "json":
            required_values = artifact.get("required_values", {})
            if not isinstance(required_values, dict) or not all(
                isinstance(path, str) and path for path in required_values
            ):
                raise ManifestError(
                    f"{context}: JSON required_values must be an object with "
                    "non-empty path keys"
                )
            required_nonempty = artifact.get("required_nonempty", [])
            if not isinstance(required_nonempty, list) or not all(
                isinstance(path, str) and path for path in required_nonempty
            ):
                raise ManifestError(
                    f"{context}: JSON required_nonempty must be a string list"
                )
            if len(required_nonempty) != len(set(required_nonempty)):
                raise ManifestError(
                    f"{context}: JSON required_nonempty paths must be unique"
                )
            required_types = artifact.get("required_types", {})
            valid_types = {
                "array", "boolean", "integer", "null", "number", "object", "string"
            }
            if not isinstance(required_types, dict) or not all(
                isinstance(path, str) and path and isinstance(value, str) and
                value in valid_types
                for path, value in required_types.items()
            ):
                raise ManifestError(
                    f"{context}: JSON required_types must map paths to valid type names"
                )
            continue
        if artifact.get("format") != "csv":
            continue

        required_columns = artifact.get("required_columns", [])
        if not isinstance(required_columns, list) or not all(
            isinstance(column, str) and column for column in required_columns
        ):
            raise ManifestError(f"{context}: CSV required_columns must be a string list")
        if len(required_columns) != len(set(required_columns)):
            raise ManifestError(f"{context}: CSV required_columns must be unique")
        allowed_values = artifact.get("allowed_values", {})
        if not isinstance(allowed_values, dict):
            raise ManifestError(f"{context}: CSV allowed_values must be an object")
        for column, values in allowed_values.items():
            if column not in required_columns:
                raise ManifestError(
                    f"{context}: allowed_values column {column!r} is not required"
                )
            if not isinstance(values, list) or not values or not all(
                isinstance(value, str) for value in values
            ):
                raise ManifestError(
                    f"{context}: allowed_values for {column!r} must be a "
                    "non-empty string list"
                )
            if len(values) != len(set(values)):
                raise ManifestError(
                    f"{context}: allowed_values for {column!r} must be unique"
                )
        performance_contract = (
            case.get("suite") == "performance" and artifact.get("required", False)
        )
        if performance_contract:
            if not required_columns:
                raise ManifestError(
                    f"{context}: required performance CSV needs required_columns"
                )
            required_performance_csv = True
            performance_csv_columns.update(required_columns)
        validate_csv_contract(
            artifact,
            context,
            required_columns,
            performance=performance_contract,
        )
    return performance_csv_columns, required_performance_csv


def validate_case(case: Mapping[str, Any], context: str) -> None:
    case_id = case.get("id")
    if not isinstance(case_id, str):
        raise ManifestError(f"{context}: case id is required")
    safe_id(case_id, "case id")
    if case.get("suite") not in SUITES:
        raise ManifestError(f"{context}: suite must be correctness or performance")
    state = case.get("state")
    if state not in CASE_STATES:
        raise ManifestError(f"{context}: invalid state {state!r}")
    profiles = case.get("profiles")
    if not isinstance(profiles, list) or not profiles or not all(
        isinstance(item, str) and item for item in profiles
    ):
        raise ManifestError(f"{context}: profiles must be a non-empty string list")
    command = case.get("command")
    if state == "ready":
        validate_command(command, context)
    else:
        if command != []:
            raise ManifestError(
                f"{context}: {state} case must have an explicitly empty command"
            )
        if not case.get("reason"):
            raise ManifestError(f"{context}: {state} case requires a reason")

    has_declared_evidence = "artifacts" in case or "performance_metadata" in case
    if state == "ready" or has_declared_evidence:
        performance_csv_columns, required_performance_csv = validate_case_artifacts(
            case, context
        )
        if case.get("suite") == "performance":
            if not required_performance_csv:
                raise ManifestError(
                    f"{context}: performance case with evidence metadata requires a "
                    "required CSV artifact"
                )
            validate_performance_metadata(case, context, performance_csv_columns)


def validate_manifest(manifest: Mapping[str, Any]) -> None:
    if manifest.get("schema_version") != SCHEMA_VERSION:
        raise ManifestError(
            f"schema_version must be {SCHEMA_VERSION}, got {manifest.get('schema_version')!r}"
        )
    catalog = manifest.get("case_catalog")
    platforms = manifest.get("platforms")
    if not isinstance(catalog, dict) or not isinstance(platforms, dict):
        raise ManifestError("case_catalog and platforms must be objects")
    for name, case in catalog.items():
        safe_id(name, "catalog key")
        if not isinstance(case, dict):
            raise ManifestError(f"catalog case {name!r} must be an object")
        validate_case(case, f"catalog case {name}")
    if not platforms:
        raise ManifestError("at least one platform is required")
    for platform_id, platform in platforms.items():
        safe_id(platform_id, "platform id")
        if not isinstance(platform, dict):
            raise ManifestError(f"platform {platform_id!r} must be an object")
        transport = platform.get("transport")
        if transport not in ("local", "ssh"):
            raise ManifestError(f"platform {platform_id}: invalid transport")
        if transport == "ssh" and not platform.get("host"):
            raise ManifestError(f"platform {platform_id}: ssh host is required")
        if not isinstance(platform.get("repo_root"), str):
            raise ManifestError(f"platform {platform_id}: repo_root is required")
        if not isinstance(platform.get("build_root"), str):
            raise ManifestError(f"platform {platform_id}: build_root is required")
        entries = platform.get("cases")
        if not isinstance(entries, list) or not entries:
            raise ManifestError(f"platform {platform_id}: cases must be non-empty")
        for entry in entries:
            if isinstance(entry, str):
                if entry not in catalog:
                    raise ManifestError(
                        f"platform {platform_id}: unknown catalog case {entry!r}"
                    )
            elif isinstance(entry, dict):
                reference = entry.get("ref")
                if reference and reference not in catalog:
                    raise ManifestError(
                        f"platform {platform_id}: unknown catalog case {reference!r}"
                    )
            else:
                raise ManifestError(f"platform {platform_id}: invalid case entry")
        case_ids: set[str] = set()
        for case in resolve_platform_cases(manifest, platform_id):
            validate_case(case, f"platform {platform_id} case")
            if case["id"] in case_ids:
                raise ManifestError(
                    f"platform {platform_id}: duplicate case id {case['id']!r}"
                )
            case_ids.add(case["id"])
            if case["state"] == "ready":
                context = platform_context(
                    platform_id,
                    platform,
                    run_id="VALIDATE",
                    case_id=case["id"],
                )
                format_values(case["command"], context)
                for artifact in case.get("artifacts", []):
                    try:
                        str(artifact["path"]).format_map(context)
                    except KeyError as error:
                        raise ManifestError(
                            f"platform {platform_id} case {case['id']}: "
                            f"unknown artifact placeholder {error.args[0]!r}"
                        ) from error


def platform_context(
    platform_id: str,
    platform: Mapping[str, Any],
    *,
    run_id: str,
    case_id: str,
) -> dict[str, str]:
    repo = str(platform["repo_root"])
    build_root = str(platform["build_root"])
    build = build_root if posixpath.isabs(build_root) else posixpath.join(repo, build_root)
    remote_root = str(platform.get("remote_work_root", "/tmp/umsh-cross-platform"))
    work = posixpath.join(remote_root, run_id, platform_id, case_id)
    return {
        "platform": platform_id,
        "repo": repo,
        "build": build,
        "work": work,
        "run_id": run_id,
        "python": str(platform.get("python", "python3")),
    }


def format_values(values: Sequence[str], context: Mapping[str, str]) -> list[str]:
    try:
        return [value.format_map(context) for value in values]
    except KeyError as error:
        raise ManifestError(f"unknown command placeholder {error.args[0]!r}") from error


def select_platforms(
    manifest: Mapping[str, Any], requested: Sequence[str] | None
) -> list[str]:
    available = list(manifest["platforms"])
    if not requested:
        return available
    unknown = sorted(set(requested) - set(available))
    if unknown:
        raise ManifestError(f"unknown platform(s): {', '.join(unknown)}")
    requested_set = set(requested)
    return [name for name in available if name in requested_set]


def select_cases(
    manifest: Mapping[str, Any],
    platform_id: str,
    *,
    suites: Sequence[str] | None,
    profile: str,
) -> list[dict[str, Any]]:
    suite_set = set(suites or SUITES)
    cases = resolve_platform_cases(manifest, platform_id)
    return [
        case
        for case in cases
        if case["suite"] in suite_set
        and (profile == "all" or profile in case["profiles"])
    ]


def plan_rows(
    manifest: Mapping[str, Any],
    platform_ids: Sequence[str],
    *,
    suites: Sequence[str] | None,
    profile: str,
) -> list[dict[str, Any]]:
    rows: list[dict[str, Any]] = []
    for platform_id in platform_ids:
        platform = manifest["platforms"][platform_id]
        for case in select_cases(
            manifest, platform_id, suites=suites, profile=profile
        ):
            context = platform_context(
                platform_id, platform, run_id="DRY_RUN", case_id=case["id"]
            )
            command = (
                format_values(case["command"], context)
                if case["state"] == "ready"
                else []
            )
            rows.append(
                {
                    "platform": platform_id,
                    "backend": platform["backend"],
                    "suite": case["suite"],
                    "case": case["id"],
                    "state": case["state"],
                    "result_scope": case.get("result_scope", "UNSPECIFIED"),
                    "paper_claim": case.get("paper_claim", "none"),
                    "performance_metadata": case.get("performance_metadata"),
                    "reason": case.get("reason", ""),
                    "command": command,
                }
            )
    return rows


def print_plan(rows: Sequence[Mapping[str, Any]], *, as_json: bool) -> None:
    if as_json:
        print(json.dumps(list(rows), indent=2, sort_keys=True))
        return
    for row in rows:
        prefix = (
            f"{row['platform']:<10} {row['suite']:<11} "
            f"{row['state']:<14} {row['case']}"
        )
        print(prefix)
        if row["command"]:
            print(f"  {shlex.join(row['command'])}")
        else:
            print(f"  reason={row['reason']}")
    counts = Counter(str(row["state"]) for row in rows)
    print(
        "plan_summary "
        + " ".join(f"{state}={counts[state]}" for state in sorted(counts))
    )


def probe_definitions() -> list[tuple[str, list[str], bool]]:
    return [
        ("hostname", ["hostname", "--fqdn"], True),
        ("uname", ["uname", "-a"], True),
        ("os_release", ["cat", "/etc/os-release"], False),
        ("kernel_cmdline", ["cat", "/proc/cmdline"], False),
        ("lscpu", ["lscpu"], True),
        ("lspci", ["lspci", "-nnk"], False),
        ("dri_devices", ["ls", "-l", "/dev/dri"], False),
        ("opencl", ["clinfo", "--list"], False),
        ("nvidia_smi", ["nvidia-smi"], False),
        ("cc", ["cc", "--version"], False),
        ("cxx", ["c++", "--version"], False),
        ("clangxx", ["clang++", "--version"], False),
        ("nvcc", ["nvcc", "--version"], False),
        ("cmake", ["cmake", "--version"], False),
        ("ninja", ["ninja", "--version"], False),
        ("loaded_modules", ["cat", "/proc/modules"], True),
        ("module_parameters", ["python3", "-c", MODULE_PARAMETER_PROGRAM], False),
        ("module_file_hashes", ["python3", "-c", MODULE_FILE_HASH_PROGRAM], False),
        ("meminfo", ["cat", "/proc/meminfo"], True),
        ("base_page_size", ["getconf", "PAGESIZE"], False),
        ("hugepages", ["python3", "-c", HUGEPAGE_PROGRAM], False),
    ]


def archive_probe(
    transport: Transport,
    destination: Path,
    name: str,
    command: Sequence[str],
    *,
    cwd: str | None = None,
    timeout: float = 60.0,
    required: bool = False,
) -> tuple[dict[str, Any], CommandResult]:
    result = transport.run(command, cwd=cwd, timeout=timeout)
    stdout_path = destination / "raw" / f"{name}.stdout.txt"
    stderr_path = destination / "raw" / f"{name}.stderr.txt"
    stdout_info = write_bytes(stdout_path, result.stdout)
    stderr_info = write_bytes(stderr_path, result.stderr)
    record = {
        "command": list(command),
        "cwd": cwd,
        "required": required,
        "exit_code": result.returncode,
        "timed_out": result.timed_out,
        "duration_seconds": round(result.duration_seconds, 6),
        "stdout": stdout_info,
        "stderr": stderr_info,
    }
    return record, result


def parse_loaded_modules(payload: bytes) -> list[str]:
    names: list[str] = []
    for line in decode(payload).splitlines():
        fields = line.split()
        if fields:
            names.append(fields[0])
    return sorted(set(names))


def collect_repo_snapshot(
    transport: Transport,
    repo: str,
    *,
    destination: Path | None = None,
) -> dict[str, Any]:
    commands = {
        "inside_worktree": ["git", "rev-parse", "--is-inside-work-tree"],
        "commit": ["git", "rev-parse", "HEAD"],
        "branch": ["git", "rev-parse", "--abbrev-ref", "HEAD"],
        "status": ["git", "status", "--porcelain=v1", "--untracked-files=all"],
        "dirty_hash": ["python3", "-c", DIRTY_HASH_PROGRAM],
    }
    results: dict[str, CommandResult] = {}
    records: dict[str, Any] = {}
    for name, command in commands.items():
        if destination is None:
            result = transport.run(command, cwd=repo, timeout=120.0)
            results[name] = result
        else:
            record, result = archive_probe(
                transport,
                destination,
                f"repo_{name}",
                command,
                cwd=repo,
                timeout=120.0,
                required=name in ("inside_worktree", "commit", "status"),
            )
            records[name] = record
            results[name] = result
    available = (
        results["inside_worktree"].returncode == 0
        and decode(results["inside_worktree"].stdout).strip() == "true"
    )
    status_text = decode(results["status"].stdout) if available else ""
    dirty = bool(status_text.strip()) if available else None
    snapshot = {
        "root": repo,
        "available": available,
        "commit": decode(results["commit"].stdout).strip() if available else None,
        "branch": decode(results["branch"].stdout).strip() if available else None,
        "dirty": dirty,
        "dirty_sha256": (
            decode(results["dirty_hash"].stdout).strip()
            if available and dirty and results["dirty_hash"].returncode == 0
            else None
        ),
        "status_sha256": sha256_bytes(status_text.encode()) if available else None,
    }
    if records:
        snapshot["probes"] = records
    return snapshot


def collect_inventory(
    manifest: Mapping[str, Any],
    platform_id: str,
    destination: Path,
) -> dict[str, Any]:
    platform = manifest["platforms"][platform_id]
    transport = Transport(platform)
    destination.mkdir(parents=True, exist_ok=False)
    probes: dict[str, Any] = {}
    raw_results: dict[str, CommandResult] = {}
    for name, command, required in probe_definitions():
        record, result = archive_probe(
            transport,
            destination,
            name,
            command,
            required=required,
        )
        probes[name] = record
        raw_results[name] = result
    repo = collect_repo_snapshot(
        transport, str(platform["repo_root"]), destination=destination
    )
    required_failures = [
        name
        for name, record in probes.items()
        if record["required"] and record["exit_code"] != 0
    ]
    loaded_modules = parse_loaded_modules(raw_results["loaded_modules"].stdout)
    inventory = {
        "schema_version": SCHEMA_VERSION,
        "captured_at": utc_now(),
        "platform": platform_id,
        "backend": platform["backend"],
        "expected_hardware": platform.get("expected_hardware", {}),
        "transport": transport.describe(),
        "repo": repo,
        "loaded_modules": loaded_modules,
        "required_probe_failures": required_failures,
        "status": "complete" if not required_failures else "partial",
        "probes": probes,
    }
    write_json(destination / "inventory.json", inventory)
    return inventory


def csv_summary(
    payload: bytes,
    required_columns: Sequence[str],
    allowed_values: Mapping[str, Sequence[str]] | None = None,
    contract: Mapping[str, Any] | None = None,
) -> dict[str, Any]:
    try:
        stream = io.StringIO(payload.decode("utf-8-sig"))
        reader = csv.DictReader(stream)
        if reader.fieldnames is None:
            raise ValueError("CSV has no header")
        rows = list(reader)
    except (UnicodeDecodeError, csv.Error, ValueError) as error:
        return {"valid": False, "error": str(error)}
    fieldnames = list(reader.fieldnames)
    missing = sorted(set(required_columns) - set(fieldnames))
    duplicate_columns = sorted(
        column for column, count in Counter(fieldnames).items() if count > 1
    )
    malformed_rows: list[dict[str, Any]] = []
    for row_number, row in enumerate(rows, start=2):
        missing_values = [field for field in fieldnames if row.get(field) is None]
        extra_values = row.get(None)
        if missing_values or extra_values:
            malformed_rows.append(
                {
                    "row": row_number,
                    "missing_values": missing_values,
                    "extra_value_count": len(extra_values or []),
                }
            )
    unexpected_values: dict[str, list[str]] = {}
    for field, expected in (allowed_values or {}).items():
        if field not in fieldnames:
            continue
        unexpected = sorted(
            set((row.get(field) or "") for row in rows) - set(expected)
        )
        if unexpected:
            unexpected_values[field] = unexpected
    contract = contract or {}
    row_count_error: dict[str, int] | None = None
    expected_rows = contract.get("expected_rows", {})
    if "exact" in expected_rows and len(rows) != expected_rows["exact"]:
        row_count_error = {"actual": len(rows), "exact": expected_rows["exact"]}
    elif len(rows) < expected_rows.get("minimum", 0):
        row_count_error = {
            "actual": len(rows),
            "minimum": expected_rows["minimum"],
        }
    elif len(rows) > expected_rows.get("maximum", sys.maxsize):
        row_count_error = {
            "actual": len(rows),
            "maximum": expected_rows["maximum"],
        }

    numeric_errors: dict[str, list[dict[str, Any]]] = {}
    for field, constraints in contract.get("numeric_columns", {}).items():
        failures: list[dict[str, Any]] = []
        for row_number, row in enumerate(rows, start=2):
            raw_value = row.get(field) or ""
            try:
                value = float(raw_value)
            except (TypeError, ValueError):
                failures.append(
                    {"row": row_number, "value": raw_value, "reason": "not_numeric"}
                )
                continue
            if constraints.get("finite") and not math.isfinite(value):
                failures.append(
                    {"row": row_number, "value": raw_value, "reason": "not_finite"}
                )
            elif constraints.get("positive") and not value > 0.0:
                failures.append(
                    {"row": row_number, "value": raw_value, "reason": "not_positive"}
                )
        if failures:
            numeric_errors[field] = failures[:100]

    duplicate_coordinates: list[dict[str, Any]] = []
    coordinate_columns = contract.get("unique_coordinates", [])
    if coordinate_columns:
        first_rows: dict[tuple[str, ...], int] = {}
        for row_number, row in enumerate(rows, start=2):
            coordinate = tuple((row.get(field) or "") for field in coordinate_columns)
            if coordinate in first_rows:
                duplicate_coordinates.append(
                    {
                        "coordinate": dict(zip(coordinate_columns, coordinate)),
                        "first_row": first_rows[coordinate],
                        "duplicate_row": row_number,
                    }
                )
            else:
                first_rows[coordinate] = row_number

    missing_coordinates: list[dict[str, str]] = []
    unexpected_coordinates: list[dict[str, str]] = []
    expected_matrix = contract.get("expected_matrix", {})
    if expected_matrix:
        matrix_columns = list(expected_matrix)
        expected_coordinates = set(itertools.product(
            *(expected_matrix[column] for column in matrix_columns)
        ))
        actual_coordinates = {
            tuple((row.get(column) or "") for column in matrix_columns)
            for row in rows
        }
        missing_coordinates = [
            dict(zip(matrix_columns, coordinate))
            for coordinate in sorted(expected_coordinates - actual_coordinates)
        ]
        unexpected_coordinates = [
            dict(zip(matrix_columns, coordinate))
            for coordinate in sorted(actual_coordinates - expected_coordinates)
        ]

    row_rule_violations: list[dict[str, Any]] = []
    for rule_index, rule in enumerate(contract.get("row_rules", [])):
        when = {
            column: [accepted] if isinstance(accepted, str) else accepted
            for column, accepted in rule["when"].items()
        }
        required = {
            column: [accepted] if isinstance(accepted, str) else accepted
            for column, accepted in rule["require"].items()
        }
        for row_number, row in enumerate(rows, start=2):
            if not all((row.get(column) or "") in accepted
                       for column, accepted in when.items()):
                continue
            failures = {
                column: row.get(column) or ""
                for column, accepted in required.items()
                if (row.get(column) or "") not in accepted
            }
            if failures:
                row_rule_violations.append({
                    "rule": rule_index,
                    "row": row_number,
                    "values": failures,
                })

    evidence_invalid_values: dict[str, dict[str, int]] = {}
    for field, rejected in contract.get("evidence_invalid_values", {}).items():
        counts = Counter((row.get(field) or "") for row in rows)
        found = {value: counts[value] for value in rejected if counts[value]}
        if found:
            evidence_invalid_values[field] = found

    structural_valid = (
        not missing
        and not duplicate_columns
        and not malformed_rows
        and not unexpected_values
        and bool(rows)
    )
    contract_valid = (
        row_count_error is None
        and not numeric_errors
        and not duplicate_coordinates
        and not missing_coordinates
        and not unexpected_coordinates
        and not row_rule_violations
    )
    evidence_valid = structural_valid and contract_valid and not evidence_invalid_values
    summary: dict[str, Any] = {
        "valid": evidence_valid,
        "structural_valid": structural_valid,
        "contract_valid": contract_valid,
        "evidence_valid": evidence_valid,
        "columns": fieldnames,
        "rows": len(rows),
        "missing_required_columns": missing,
        "duplicate_columns": duplicate_columns,
        "malformed_rows": malformed_rows[:100],
        "malformed_row_count": len(malformed_rows),
        "unexpected_values": unexpected_values,
        "row_count_error": row_count_error,
        "numeric_errors": numeric_errors,
        "duplicate_coordinates": duplicate_coordinates[:100],
        "duplicate_coordinate_count": len(duplicate_coordinates),
        "missing_coordinates": missing_coordinates[:100],
        "missing_coordinate_count": len(missing_coordinates),
        "unexpected_coordinates": unexpected_coordinates[:100],
        "unexpected_coordinate_count": len(unexpected_coordinates),
        "row_rule_violations": row_rule_violations[:100],
        "row_rule_violation_count": len(row_rule_violations),
        "evidence_invalid_values": evidence_invalid_values,
    }
    if not rows:
        summary["error"] = "CSV contains no data rows"
    summary_fields = {
        "status",
        "result",
        "supported",
        "executed",
        "result_scope",
        *(allowed_values or {}).keys(),
    }
    for field in sorted(summary_fields):
        if field in fieldnames:
            summary[f"{field}_counts"] = dict(
                sorted(Counter((row.get(field) or "") for row in rows).items())
            )
    evidence_errors: list[str] = []
    if row_count_error is not None:
        evidence_errors.append("unexpected_row_count")
    if numeric_errors:
        evidence_errors.append("invalid_numeric_value")
    if duplicate_coordinates:
        evidence_errors.append("duplicate_coordinate")
    if missing_coordinates:
        evidence_errors.append("missing_coordinate")
    if unexpected_coordinates:
        evidence_errors.append("unexpected_coordinate")
    if row_rule_violations:
        evidence_errors.append("row_rule_violation")
    if evidence_invalid_values:
        evidence_errors.append("declared_evidence_invalid_value")
    summary["evidence_errors"] = evidence_errors
    return summary


def _json_path(payload: Any, path: str) -> tuple[bool, Any]:
    current = payload
    for component in path.split("."):
        if not isinstance(current, dict) or component not in current:
            return False, None
        current = current[component]
    return True, current


def json_summary(
    payload: bytes,
    required_values: Mapping[str, Any] | None = None,
    required_nonempty: Sequence[str] | None = None,
    required_types: Mapping[str, str] | None = None,
) -> dict[str, Any]:
    try:
        document = json.loads(payload)
    except (UnicodeDecodeError, json.JSONDecodeError) as error:
        return {
            "valid": False,
            "structural_valid": False,
            "evidence_valid": False,
            "error": str(error),
        }
    if not isinstance(document, dict):
        return {
            "valid": False,
            "structural_valid": False,
            "evidence_valid": False,
            "error": "JSON root must be an object",
        }

    missing_paths: list[str] = []
    mismatched_values: dict[str, dict[str, Any]] = {}
    for path, expected in (required_values or {}).items():
        exists, actual = _json_path(document, path)
        if not exists:
            missing_paths.append(path)
        elif type(actual) is not type(expected) or actual != expected:
            mismatched_values[path] = {"expected": expected, "actual": actual}

    empty_paths: list[str] = []
    for path in required_nonempty or []:
        exists, actual = _json_path(document, path)
        if not exists:
            missing_paths.append(path)
        elif actual is None or actual == "" or actual == [] or actual == {}:
            empty_paths.append(path)

    type_errors: dict[str, dict[str, str]] = {}
    type_checks = {
        "array": lambda value: isinstance(value, list),
        "boolean": lambda value: isinstance(value, bool),
        "integer": lambda value: isinstance(value, int) and not isinstance(value, bool),
        "null": lambda value: value is None,
        "number": lambda value: isinstance(value, (int, float)) and not isinstance(value, bool),
        "object": lambda value: isinstance(value, dict),
        "string": lambda value: isinstance(value, str),
    }
    for path, expected_type in (required_types or {}).items():
        exists, actual = _json_path(document, path)
        if not exists:
            missing_paths.append(path)
        elif not type_checks[expected_type](actual):
            type_errors[path] = {
                "expected": expected_type,
                "actual": type(actual).__name__,
            }

    missing_paths = sorted(set(missing_paths))
    evidence_valid = (
        not missing_paths and not mismatched_values and not empty_paths and
        not type_errors
    )
    return {
        "valid": evidence_valid,
        "structural_valid": True,
        "evidence_valid": evidence_valid,
        "missing_paths": missing_paths,
        "mismatched_values": mismatched_values,
        "empty_paths": empty_paths,
        "type_errors": type_errors,
    }


def check_case_preconditions(
    case: Mapping[str, Any], inventory: Mapping[str, Any]
) -> list[str]:
    failures: list[str] = []
    requirements = case.get("requires", {})
    loaded = set(inventory.get("loaded_modules", []))
    for module in requirements.get("loaded_modules", []):
        if module not in loaded:
            failures.append(f"module_not_loaded:{module}")
    return failures


def collect_case_artifacts(
    transport: Transport,
    case: Mapping[str, Any],
    context: Mapping[str, str],
    destination: Path,
) -> tuple[list[dict[str, Any]], list[str]]:
    records: list[dict[str, Any]] = []
    errors: list[str] = []
    artifact_dir = destination / "artifacts"
    for specification in case.get("artifacts", []):
        artifact_id = str(specification["id"])
        safe_id(artifact_id, "artifact id")
        source = str(specification["path"]).format_map(context)
        kind = specification.get("kind", "file")
        required = bool(specification.get("required", False))
        if kind == "file":
            result = transport.read_file(source)
            suffix = Path(source).suffix
            local_path = artifact_dir / f"{artifact_id}{suffix}"
        elif kind == "directory":
            result = transport.archive_directory(source)
            local_path = artifact_dir / f"{artifact_id}.tar"
        else:
            raise ManifestError(f"case {case['id']}: invalid artifact kind {kind!r}")
        record: dict[str, Any] = {
            "id": artifact_id,
            "kind": kind,
            "source": source,
            "required": required,
            "fetch_exit_code": result.returncode,
            "fetch_stderr": decode(result.stderr),
        }
        artifact_format = specification.get("format")
        if artifact_format in ("csv", "json"):
            record["format"] = artifact_format
        if artifact_format == "csv":
            record["required_columns"] = specification.get("required_columns", [])
            record["allowed_values"] = specification.get("allowed_values", {})
            for field in (
                "expected_rows",
                "unique_coordinates",
                "numeric_columns",
                "evidence_invalid_values",
                "expected_matrix",
                "row_rules",
            ):
                if field in specification:
                    record[field] = specification[field]
        if result.returncode == 0:
            record.update(write_bytes(local_path, result.stdout))
            if artifact_format == "csv":
                summary = csv_summary(
                    result.stdout,
                    specification.get("required_columns", []),
                    specification.get("allowed_values", {}),
                    specification,
                )
                record["csv_summary"] = summary
                if required and not summary["valid"]:
                    error_kind = (
                        "evidence_invalid_csv"
                        if summary.get("structural_valid")
                        else "invalid_csv"
                    )
                    errors.append(f"{error_kind}:{artifact_id}")
            elif artifact_format == "json":
                summary = json_summary(
                    result.stdout,
                    specification.get("required_values", {}),
                    specification.get("required_nonempty", []),
                    specification.get("required_types", {}),
                )
                record["required_values"] = specification.get(
                    "required_values", {}
                )
                record["required_nonempty"] = specification.get(
                    "required_nonempty", []
                )
                record["required_types"] = specification.get(
                    "required_types", {}
                )
                record["json_summary"] = summary
                if required and not summary["valid"]:
                    error_kind = (
                        "evidence_invalid_json"
                        if summary.get("structural_valid")
                        else "invalid_json"
                    )
                    errors.append(f"{error_kind}:{artifact_id}")
        elif required:
            errors.append(f"missing_artifact:{artifact_id}")
        records.append(record)
    return records, errors


def archive_performance_metadata(
    platform_id: str,
    platform: Mapping[str, Any],
    case: Mapping[str, Any],
    destination: Path,
    case_record: dict[str, Any],
) -> None:
    metadata = case.get("performance_metadata")
    if case.get("suite") != "performance" or not isinstance(metadata, dict):
        return
    sidecar = {
        "schema_version": SCHEMA_VERSION,
        "platform": platform_id,
        "backend": platform["backend"],
        "case": case["id"],
        "paper_claim": case.get("paper_claim", "none"),
        "coordinates": metadata,
    }
    path = destination / "performance-metadata.json"
    write_json(path, sidecar)
    case_record["performance_metadata"] = metadata
    case_record["performance_metadata_sidecar"] = path.name


def execute_case(
    manifest: Mapping[str, Any],
    platform_id: str,
    case: Mapping[str, Any],
    inventory: Mapping[str, Any],
    destination: Path,
    *,
    run_id: str,
    timeout_override: float | None,
) -> dict[str, Any]:
    platform = manifest["platforms"][platform_id]
    transport = Transport(platform)
    destination.mkdir(parents=True, exist_ok=False)
    context = platform_context(
        platform_id, platform, run_id=run_id, case_id=case["id"]
    )
    base = {
        "platform": platform_id,
        "backend": platform["backend"],
        "suite": case["suite"],
        "case": case["id"],
        "declared_state": case["state"],
        "result_scope": case.get("result_scope", "UNSPECIFIED"),
        "paper_claim": case.get("paper_claim", "none"),
        "started_at": utc_now(),
        "command": [],
        "work_directory": context["work"],
        "artifacts": [],
        "reason": "",
    }
    archive_performance_metadata(platform_id, platform, case, destination, base)
    if case["state"] != "ready":
        base["status"] = (
            "SKIP_EXTERNAL_SETUP"
            if case["state"] == "external_setup"
            else "SKIP_UNSUPPORTED"
        )
        base["reason"] = case["reason"]
        base["completed_at"] = utc_now()
        write_json(destination / "case.json", base)
        return base
    if not inventory["repo"]["available"]:
        base["status"] = "ERROR_PRECONDITION"
        base["reason"] = "repository_not_available"
        base["completed_at"] = utc_now()
        write_json(destination / "case.json", base)
        return base
    preconditions = check_case_preconditions(case, inventory)
    if preconditions:
        base["status"] = "SKIP_PRECONDITION"
        base["reason"] = ";".join(preconditions)
        base["completed_at"] = utc_now()
        write_json(destination / "case.json", base)
        return base

    mkdir_result = transport.run(
        ["mkdir", "-p", "--", context["work"]], timeout=30.0
    )
    if mkdir_result.returncode != 0:
        base["status"] = "ERROR_PRECONDITION"
        base["reason"] = f"cannot_create_work_directory:{decode(mkdir_result.stderr).strip()}"
        base["completed_at"] = utc_now()
        write_json(destination / "case.json", base)
        return base

    command = format_values(case["command"], context)
    validate_command(command, f"case {case['id']}")
    timeout = timeout_override or float(case.get("timeout_seconds", 3600.0))
    result = transport.run(command, cwd=context["repo"], timeout=timeout)
    base["command"] = command
    base["exit_code"] = result.returncode
    base["timed_out"] = result.timed_out
    base["duration_seconds"] = round(result.duration_seconds, 6)
    base["stdout"] = write_bytes(destination / "stdout.txt", result.stdout)
    base["stderr"] = write_bytes(destination / "stderr.txt", result.stderr)
    artifacts, artifact_errors = collect_case_artifacts(
        transport, case, context, destination
    )
    base["artifacts"] = artifacts
    if result.timed_out:
        base["status"] = "ERROR_TIMEOUT"
        base["reason"] = f"case_timeout_seconds:{timeout}"
    elif artifact_errors:
        base["status"] = "ERROR_ARTIFACT"
        reason_parts = list(artifact_errors)
        if result.returncode != 0:
            reason_parts.append(f"command_exit_code:{result.returncode}")
        base["reason"] = ";".join(reason_parts)
    elif result.returncode != 0:
        base["status"] = "ERROR_COMMAND"
        base["reason"] = f"exit_code:{result.returncode}"
    else:
        base["status"] = "RECORDED"
        base["reason"] = "command_and_artifact_contract_completed"
    base["completed_at"] = utc_now()
    write_json(destination / "case.json", base)
    return base


def source_requirement_errors(
    inventory: Mapping[str, Any],
    *,
    expected_commit: str | None,
    require_clean: bool,
) -> list[str]:
    repo = inventory["repo"]
    errors: list[str] = []
    if not repo["available"]:
        return ["repository_not_available"]
    commit = str(repo["commit"])
    if expected_commit and not commit.startswith(expected_commit):
        errors.append(f"commit_mismatch:expected={expected_commit}:actual={commit}")
    if require_clean and repo["dirty"]:
        errors.append(f"dirty_worktree:{repo['dirty_sha256']}")
    return errors


def write_case_summary(path: Path, cases: Sequence[Mapping[str, Any]]) -> None:
    fields = (
        "platform",
        "backend",
        "suite",
        "case",
        "declared_state",
        "status",
        "result_scope",
        "paper_claim",
        "exit_code",
        "duration_seconds",
        "reason",
    )
    path.parent.mkdir(parents=True, exist_ok=True)
    with path.open("w", newline="", encoding="utf-8") as stream:
        writer = csv.DictWriter(stream, fieldnames=fields)
        writer.writeheader()
        for case in cases:
            writer.writerow({field: case.get(field, "") for field in fields})


def write_hash_index(root: Path) -> None:
    entries: list[str] = []
    for path in sorted(item for item in root.rglob("*") if item.is_file()):
        if path.name == "SHA256SUMS":
            continue
        entries.append(f"{sha256_file(path)}  {path.relative_to(root)}")
    write_bytes(root / "SHA256SUMS", ("\n".join(entries) + "\n").encode())


def create_campaign_root(archive_root: Path, run_id: str) -> Path:
    safe_id(run_id, "run id")
    destination = archive_root / run_id
    destination.mkdir(parents=True, exist_ok=False)
    return destination


def execute_run(
    manifest: Mapping[str, Any],
    manifest_payload: bytes,
    platform_ids: Sequence[str],
    *,
    suites: Sequence[str] | None,
    profile: str,
    archive_root: Path,
    run_id: str,
    timeout_override: float | None,
    expected_commit: str | None,
    require_clean: bool,
    notes: Mapping[str, str],
) -> tuple[Path, int]:
    root = create_campaign_root(archive_root, run_id)
    write_bytes(root / "manifest.snapshot.json", manifest_payload)
    campaign: dict[str, Any] = {
        "schema_version": SCHEMA_VERSION,
        "harness_version": HARNESS_VERSION,
        "kind": "test_run",
        "run_id": run_id,
        "started_at": utc_now(),
        "profile": profile,
        "suites": list(suites or sorted(SUITES)),
        "platforms": list(platform_ids),
        "manifest_sha256": sha256_bytes(manifest_payload),
        "source_requirements": {
            "expected_commit": expected_commit,
            "require_clean": require_clean,
        },
        "notes": dict(notes),
        "platform_records": {},
    }
    all_cases: list[dict[str, Any]] = []
    infrastructure_errors = 0
    for platform_id in platform_ids:
        platform_root = root / platform_id
        inventory = collect_inventory(
            manifest, platform_id, platform_root / "inventory"
        )
        if inventory["status"] != "complete":
            infrastructure_errors += 1
        requirement_errors = source_requirement_errors(
            inventory,
            expected_commit=expected_commit,
            require_clean=require_clean,
        )
        selected = select_cases(
            manifest, platform_id, suites=suites, profile=profile
        )
        case_records: list[dict[str, Any]] = []
        for ordinal, case in enumerate(selected, start=1):
            destination = platform_root / "cases" / f"{ordinal:03d}-{case['id']}"
            if requirement_errors and case["state"] == "ready":
                destination.mkdir(parents=True, exist_ok=False)
                record = {
                    "platform": platform_id,
                    "backend": manifest["platforms"][platform_id]["backend"],
                    "suite": case["suite"],
                    "case": case["id"],
                    "declared_state": case["state"],
                    "status": "ERROR_PRECONDITION",
                    "result_scope": case.get("result_scope", "UNSPECIFIED"),
                    "paper_claim": case.get("paper_claim", "none"),
                    "reason": ";".join(requirement_errors),
                    "command": [],
                    "artifacts": [],
                    "started_at": utc_now(),
                    "completed_at": utc_now(),
                }
                archive_performance_metadata(
                    platform_id,
                    manifest["platforms"][platform_id],
                    case,
                    destination,
                    record,
                )
                write_json(destination / "case.json", record)
            else:
                record = execute_case(
                    manifest,
                    platform_id,
                    case,
                    inventory,
                    destination,
                    run_id=run_id,
                    timeout_override=timeout_override,
                )
            case_records.append(record)
            all_cases.append(record)
            if str(record["status"]).startswith("ERROR_"):
                infrastructure_errors += 1
        post_source = collect_repo_snapshot(
            Transport(manifest["platforms"][platform_id]),
            str(manifest["platforms"][platform_id]["repo_root"]),
        )
        source_changed = any(
            inventory["repo"].get(field) != post_source.get(field)
            for field in ("commit", "dirty", "dirty_sha256", "status_sha256")
        )
        if source_changed:
            infrastructure_errors += 1
        campaign["platform_records"][platform_id] = {
            "inventory": str(
                (platform_root / "inventory" / "inventory.json").relative_to(root)
            ),
            "source_before": inventory["repo"],
            "source_after": post_source,
            "source_changed_during_run": source_changed,
            "case_status_counts": dict(
                sorted(Counter(record["status"] for record in case_records).items())
            ),
        }
    write_case_summary(root / "cases.csv", all_cases)
    campaign["completed_at"] = utc_now()
    campaign["case_status_counts"] = dict(
        sorted(Counter(record["status"] for record in all_cases).items())
    )
    campaign["infrastructure_errors"] = infrastructure_errors
    campaign["status"] = "complete" if infrastructure_errors == 0 else "error"
    write_json(root / "campaign.json", campaign)
    write_hash_index(root)
    return root, 0 if infrastructure_errors == 0 else 1


def execute_inventory(
    manifest: Mapping[str, Any],
    manifest_payload: bytes,
    platform_ids: Sequence[str],
    *,
    archive_root: Path,
    run_id: str,
    notes: Mapping[str, str],
) -> tuple[Path, int]:
    root = create_campaign_root(archive_root, run_id)
    write_bytes(root / "manifest.snapshot.json", manifest_payload)
    records: dict[str, Any] = {}
    partial = 0
    for platform_id in platform_ids:
        inventory = collect_inventory(
            manifest, platform_id, root / platform_id / "inventory"
        )
        records[platform_id] = {
            "status": inventory["status"],
            "repo": inventory["repo"],
        }
        if inventory["status"] != "complete":
            partial += 1
    campaign = {
        "schema_version": SCHEMA_VERSION,
        "harness_version": HARNESS_VERSION,
        "kind": "inventory",
        "run_id": run_id,
        "captured_at": utc_now(),
        "platforms": list(platform_ids),
        "manifest_sha256": sha256_bytes(manifest_payload),
        "notes": dict(notes),
        "platform_records": records,
        "status": "complete" if partial == 0 else "partial",
    }
    write_json(root / "campaign.json", campaign)
    write_hash_index(root)
    return root, 0 if partial == 0 else 1


def parse_notes(values: Sequence[str]) -> dict[str, str]:
    notes: dict[str, str] = {}
    for value in values:
        key, separator, content = value.partition("=")
        if not separator or not key:
            raise ManifestError(f"note must use KEY=VALUE syntax: {value!r}")
        safe_id(key, "note key")
        notes[key] = content
    return notes


def archive_root_from(manifest: Mapping[str, Any], value: str | None) -> Path:
    raw = value or str(
        manifest.get("default_archive_root", "build/experiment-results/cross-platform")
    )
    path = Path(raw).expanduser()
    return path if path.is_absolute() else REPO_ROOT / path


def add_selection_arguments(parser: argparse.ArgumentParser) -> None:
    parser.add_argument(
        "--platform",
        action="append",
        default=[],
        help="platform id; repeat to select multiple (default: all)",
    )


def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument(
        "--manifest", type=Path, default=DEFAULT_MANIFEST, help="platform manifest"
    )
    subparsers = parser.add_subparsers(dest="action", required=True)

    subparsers.add_parser("validate", help="validate the manifest and exit")

    plan = subparsers.add_parser("plan", help="print commands without connecting")
    add_selection_arguments(plan)
    plan.add_argument("--suite", action="append", choices=sorted(SUITES), default=[])
    plan.add_argument(
        "--profile", choices=("smoke", "full", "all"), default="smoke"
    )
    plan.add_argument("--json", action="store_true", help="emit JSON")

    inventory = subparsers.add_parser(
        "inventory", help="plan or execute read-only platform inventory"
    )
    add_selection_arguments(inventory)
    inventory.add_argument("--execute", action="store_true")
    inventory.add_argument("--archive-root")
    inventory.add_argument("--run-id")
    inventory.add_argument("--note", action="append", default=[], metavar="KEY=VALUE")

    run = subparsers.add_parser(
        "run", help="dry-run by default; --execute creates an archived campaign"
    )
    add_selection_arguments(run)
    run.add_argument("--suite", action="append", choices=sorted(SUITES), default=[])
    run.add_argument(
        "--profile", choices=("smoke", "full", "all"), default="smoke"
    )
    run.add_argument("--execute", action="store_true")
    run.add_argument("--archive-root")
    run.add_argument("--run-id")
    run.add_argument("--timeout-seconds", type=float)
    run.add_argument("--expected-commit")
    run.add_argument("--require-clean", action="store_true")
    run.add_argument("--note", action="append", default=[], metavar="KEY=VALUE")
    run.add_argument("--json", action="store_true", help="JSON dry-run output")
    return parser


def main(argv: Sequence[str] | None = None) -> int:
    parser = build_parser()
    args = parser.parse_args(argv)
    try:
        manifest, payload = load_manifest(args.manifest.expanduser())
        if args.action == "validate":
            print(
                f"manifest_valid schema={SCHEMA_VERSION} "
                f"platforms={len(manifest['platforms'])} "
                f"catalog_cases={len(manifest['case_catalog'])}"
            )
            return 0
        platform_ids = select_platforms(manifest, args.platform)
        if args.action == "plan":
            rows = plan_rows(
                manifest,
                platform_ids,
                suites=args.suite,
                profile=args.profile,
            )
            if not rows:
                raise ManifestError("selection matched no cases")
            print_plan(rows, as_json=args.json)
            return 0
        notes = parse_notes(args.note)
        run_id = args.run_id or default_run_id()
        archive_root = archive_root_from(manifest, args.archive_root)
        if args.action == "inventory":
            if not args.execute:
                print("inventory_dry_run no connections made")
                for platform_id in platform_ids:
                    platform = manifest["platforms"][platform_id]
                    print(
                        f"{platform_id:<10} transport={Transport(platform).describe()} "
                        f"repo={platform['repo_root']} probes={len(probe_definitions()) + 5}"
                    )
                return 0
            root, status = execute_inventory(
                manifest,
                payload,
                platform_ids,
                archive_root=archive_root,
                run_id=run_id,
                notes=notes,
            )
            print(f"inventory_archived path={root}")
            return status
        rows = plan_rows(
            manifest,
            platform_ids,
            suites=args.suite,
            profile=args.profile,
        )
        if not rows:
            raise ManifestError("selection matched no cases")
        if not args.execute:
            print_plan(rows, as_json=args.json)
            print("run_dry_run no connections or files created")
            return 0
        if args.timeout_seconds is not None and args.timeout_seconds <= 0:
            raise ManifestError("--timeout-seconds must be positive")
        root, status = execute_run(
            manifest,
            payload,
            platform_ids,
            suites=args.suite,
            profile=args.profile,
            archive_root=archive_root,
            run_id=run_id,
            timeout_override=args.timeout_seconds,
            expected_commit=args.expected_commit,
            require_clean=args.require_clean,
            notes=notes,
        )
        print(f"campaign_archived path={root} exit_status={status}")
        return status
    except (ManifestError, FileExistsError, OSError) as error:
        parser.error(str(error))
    return 2


if __name__ == "__main__":
    sys.exit(main())