release/v1/scripts/build_tables.py

#!/usr/bin/env python3
"""Build deterministic canonical tables from append-only evidence batches.

This transformation is intentionally strict.  It accepts only completed raw
batches, verifies every hash and declared record count, validates each record
according to its type, and refuses unknown record types.  Error and exclusion
records remain visible in dedicated canonical tables; they are never filtered
away while constructing summaries.
"""

from __future__ import annotations

import argparse
import csv
import hashlib
import json
import math
import os
import shutil
import statistics
import sys
import tempfile
from collections import Counter, defaultdict
from dataclasses import dataclass
from datetime import datetime
from pathlib import Path
from typing import Any, Iterable, Mapping, Sequence


BUILDER_VERSION = "1.1.0"
SUPPORTED_SCHEMA_VERSION = "1.0.0"
ALLOWED_RECORD_TYPES = {
    "episode",
    "training_run",
    "training_metric",
    "proof_certificate",
    "parameter_audit",
    "artifact",
    "selection",
    "environment",
    "claim_summary",
    "scope_limitation",
    "failure",
    "error",
    "exclusion",
}
FAILURE_STATUSES = {"error", "failed", "failure"}
EXCLUSION_STATUSES = {"excluded", "not_executed"}
ALLOWED_STATUSES = {"success", *EXCLUSION_STATUSES, *FAILURE_STATUSES}

PROVENANCE_COLUMNS = (
    "source_batch",
    "source_line_number",
    "record_sha256",
)
COMMON_COLUMNS = (
    "schema_version",
    "experiment_id",
    "spec_version",
    "paper_version",
    "openreview_id",
    "batch_id",
    "record_type",
    "recorded_at",
    "status",
    "claim_id",
    "panel_id",
    "task_id",
    "attempt_id",
    "method",
    "environment",
    "seed",
)
TABLE_CORE_COLUMNS: dict[str, tuple[str, ...]] = {
    "episodes.csv": (
        *PROVENANCE_COLUMNS,
        *COMMON_COLUMNS,
        "start_position",
        "initial_angle",
        "initial_angular_velocity",
        "return",
        "episode_length",
        "terminal_velocity",
    ),
    "training_runs.csv": (
        *PROVENANCE_COLUMNS,
        *COMMON_COLUMNS,
        "timesteps",
        "episodes",
        "evaluation_mean_return",
        "selection_mean_return",
        "mean_return",
        "wall_time_seconds",
        "peak_rss_mb",
        "selected",
    ),
    "training_metrics.csv": (
        *PROVENANCE_COLUMNS,
        *COMMON_COLUMNS,
        "timesteps",
        "mean_return",
        "training_episode",
        "return",
        "episode_length",
    ),
    "proof_certificates.csv": (
        *PROVENANCE_COLUMNS,
        *COMMON_COLUMNS,
        "claim_scope",
        "historical_priority_tested",
        "discrete_global_optimality_tested",
    ),
    "parameter_audit.csv": (
        *PROVENANCE_COLUMNS,
        *COMMON_COLUMNS,
        "count_source",
        "parameter_count",
        "performance_method",
        "reduction_factor",
    ),
    "artifacts.csv": (
        *PROVENANCE_COLUMNS,
        *COMMON_COLUMNS,
        "artifact_path",
        "artifact_sha256",
        "artifact_bytes",
    ),
    "selections.csv": (
        *PROVENANCE_COLUMNS,
        *COMMON_COLUMNS,
        "selection_rule",
        "candidate_count",
        "selected_seed",
        "selected_summary",
    ),
    "environments.csv": (
        *PROVENANCE_COLUMNS,
        *COMMON_COLUMNS,
        "packages",
        "hardware",
        "upstream_commits",
    ),
    "claim_summaries.csv": (
        *PROVENANCE_COLUMNS,
        *COMMON_COLUMNS,
        "summary",
    ),
    "errors.csv": (
        *PROVENANCE_COLUMNS,
        *COMMON_COLUMNS,
        "error",
        "message",
        "traceback",
    ),
    "exclusions.csv": (
        *PROVENANCE_COLUMNS,
        *COMMON_COLUMNS,
        "exclusion_reason",
        "reason",
    ),
    "summaries.csv": (
        "summary_type",
        "experiment_id",
        "spec_version",
        "paper_version",
        "openreview_id",
        "batch_id",
        "claim_id",
        "panel_id",
        "environment",
        "method",
        "metric_name",
        "record_count",
        "success_count",
        "error_count",
        "exclusion_count",
        "value_count",
        "mean",
        "std_population",
        "min",
        "max",
    ),
}


class ValidationError(RuntimeError):
    """Raised when raw evidence does not satisfy the canonical contract."""


@dataclass(frozen=True)
class BatchData:
    directory: Path
    manifest: dict[str, Any]
    records: tuple[dict[str, Any], ...]
    records_sha256: str
    manifest_sha256: str
    coverage_checks: tuple[dict[str, Any], ...]


def sha256_file(path: Path) -> str:
    digest = hashlib.sha256()
    with path.open("rb") as handle:
        for chunk in iter(lambda: handle.read(1024 * 1024), b""):
            digest.update(chunk)
    return digest.hexdigest()


def canonical_json(value: Any) -> str:
    return json.dumps(
        value,
        ensure_ascii=False,
        allow_nan=False,
        separators=(",", ":"),
        sort_keys=True,
    )


def _reject_json_constant(value: str) -> None:
    raise ValidationError(f"Non-standard JSON constant is forbidden: {value}")


def _unique_object(pairs: Sequence[tuple[str, Any]]) -> dict[str, Any]:
    result: dict[str, Any] = {}
    for key, value in pairs:
        if key in result:
            raise ValidationError(f"Duplicate JSON object key: {key!r}")
        result[key] = value
    return result


def parse_json(text: str, source: str) -> Any:
    try:
        return json.loads(
            text,
            object_pairs_hook=_unique_object,
            parse_constant=_reject_json_constant,
        )
    except (json.JSONDecodeError, ValidationError) as exc:
        raise ValidationError(f"Invalid JSON in {source}: {exc}") from exc


def require_object(value: Any, source: str) -> dict[str, Any]:
    if not isinstance(value, dict):
        raise ValidationError(f"Expected a JSON object in {source}")
    return value


def require_string(record: Mapping[str, Any], key: str, source: str) -> str:
    value = record.get(key)
    if not isinstance(value, str) or not value.strip():
        raise ValidationError(f"{source}: {key!r} must be a non-empty string")
    return value


def require_integer(
    record: Mapping[str, Any], key: str, source: str, minimum: int | None = None
) -> int:
    value = record.get(key)
    if isinstance(value, bool) or not isinstance(value, int):
        raise ValidationError(f"{source}: {key!r} must be an integer")
    if minimum is not None and value < minimum:
        raise ValidationError(f"{source}: {key!r} must be >= {minimum}")
    return value


def require_number(record: Mapping[str, Any], key: str, source: str) -> float:
    value = record.get(key)
    if isinstance(value, bool) or not isinstance(value, (int, float)):
        raise ValidationError(f"{source}: {key!r} must be numeric")
    number = float(value)
    if not math.isfinite(number):
        raise ValidationError(f"{source}: {key!r} must be finite")
    return number


def validate_optional_number(record: Mapping[str, Any], key: str, source: str) -> None:
    if key in record and record[key] is not None:
        require_number(record, key, source)


def validate_optional_integer(record: Mapping[str, Any], key: str, source: str) -> None:
    if key in record and record[key] is not None:
        require_integer(record, key, source)


def validate_timestamp(value: Any, source: str) -> None:
    if not isinstance(value, str) or not value:
        raise ValidationError(f"{source}: recorded_at must be a non-empty ISO timestamp")
    normalized = value[:-1] + "+00:00" if value.endswith("Z") else value
    try:
        parsed = datetime.fromisoformat(normalized)
    except ValueError as exc:
        raise ValidationError(f"{source}: invalid recorded_at timestamp {value!r}") from exc
    if parsed.tzinfo is None or parsed.utcoffset() is None:
        raise ValidationError(f"{source}: recorded_at must include a UTC offset")


def validate_json_tree(value: Any, source: str) -> None:
    if value is None or isinstance(value, (str, bool, int)):
        return
    if isinstance(value, float):
        if not math.isfinite(value):
            raise ValidationError(f"{source}: non-finite numeric value")
        return
    if isinstance(value, list):
        for index, item in enumerate(value):
            validate_json_tree(item, f"{source}[{index}]")
        return
    if isinstance(value, dict):
        for key, item in value.items():
            if not isinstance(key, str):
                raise ValidationError(f"{source}: JSON object key is not a string")
            validate_json_tree(item, f"{source}.{key}")
        return
    raise ValidationError(f"{source}: unsupported JSON value type {type(value).__name__}")


def validate_failure_reason(record: Mapping[str, Any], source: str) -> None:
    if not any(
        isinstance(record.get(key), str) and bool(record[key].strip())
        for key in ("error", "message", "reason", "exclusion_reason")
    ):
        raise ValidationError(f"{source}: failed/excluded record has no reason")


def validate_return_summary(value: Any, label: str) -> None:
    if not isinstance(value, dict):
        raise ValidationError(f"{label} must be an object")
    for key in ("count", "mean", "std", "min", "max"):
        if key not in value:
            raise ValidationError(f"{label} is missing {key!r}")
    require_integer(value, "count", label, minimum=1)
    for key in ("mean", "std", "min", "max"):
        require_number(value, key, label)
    if float(value["std"]) < 0:
        raise ValidationError(f"{label}.std must be non-negative")
    if float(value["min"]) > float(value["mean"]) or float(value["mean"]) > float(value["max"]):
        raise ValidationError(f"{label} must satisfy min <= mean <= max")


def validate_record(record: dict[str, Any], source: str) -> None:
    validate_json_tree(record, source)
    for key in (
        "schema_version",
        "experiment_id",
        "spec_version",
        "paper_version",
        "openreview_id",
        "batch_id",
        "record_type",
    ):
        require_string(record, key, source)
    validate_timestamp(record.get("recorded_at"), source)
    record_type = record["record_type"]
    if record_type not in ALLOWED_RECORD_TYPES:
        raise ValidationError(
            f"{source}: unknown record_type {record_type!r}; refusing to drop it"
        )
    status = require_string(record, "status", source)
    if status not in ALLOWED_STATUSES:
        raise ValidationError(f"{source}: unsupported status {status!r}")
    if status != "success":
        validate_failure_reason(record, source)

    if record_type == "episode":
        for key in ("claim_id", "task_id", "method", "environment", "panel_id"):
            require_string(record, key, source)
        require_integer(record, "attempt_id", source, minimum=1)
        validate_optional_integer(record, "seed", source)
        for key in (
            "start_position",
            "initial_angle",
            "initial_angular_velocity",
            "terminal_velocity",
        ):
            validate_optional_number(record, key, source)
        if status == "success":
            require_number(record, "return", source)
            if "episode_length" in record and record["episode_length"] is not None:
                require_integer(record, "episode_length", source, minimum=1)
    elif record_type == "training_run":
        for key in ("claim_id", "method", "panel_id"):
            require_string(record, key, source)
        require_integer(record, "seed", source)
        for key in (
            "configured_timesteps",
            "actual_timesteps",
            "configured_episodes",
            "actual_training_episodes",
        ):
            if key in record and record[key] is not None:
                require_integer(record, key, source, minimum=0)
        for key in ("wall_time_seconds", "peak_rss_mb"):
            validate_optional_number(record, key, source)
        summaries = [
            key
            for key in ("evaluation_summary", "selection_summary")
            if key in record and record[key] is not None
        ]
        if status == "success" and not summaries:
            raise ValidationError(
                f"{source}: successful training_run has no declared return metric"
            )
        for key in summaries:
            validate_return_summary(record[key], f"{source}.{key}")
    elif record_type == "training_metric":
        for key in ("claim_id", "task_id", "method", "panel_id"):
            require_string(record, key, source)
        require_integer(record, "seed", source)
        has_periodic = "timesteps" in record or "mean_return" in record
        has_episode = "training_episode" in record or "return" in record
        if has_periodic == has_episode:
            raise ValidationError(
                f"{source}: training_metric must be exactly one periodic or episode metric"
            )
        if has_periodic:
            require_integer(record, "timesteps", source, minimum=0)
            require_number(record, "mean_return", source)
        else:
            require_integer(record, "training_episode", source, minimum=0)
            require_number(record, "return", source)
            require_integer(record, "episode_length", source, minimum=1)
    elif record_type == "proof_certificate":
        require_string(record, "claim_id", source)
        require_string(record, "task_id", source)
    elif record_type == "parameter_audit":
        for key in ("claim_id", "task_id", "method"):
            require_string(record, key, source)
        if status == "success":
            if "parameter_count" in record:
                require_string(record, "count_source", source)
                require_integer(record, "parameter_count", source, minimum=1)
            else:
                count_keys = (
                    "chebyshev_parameter_count",
                    "paper_stated_mlp_parameter_count",
                    "correct_dense_2x64x64x1_parameter_count",
                    "released_ars_comparator_parameter_count",
                )
                for key in count_keys:
                    require_integer(record, key, source, minimum=1)
                ratio_contracts = (
                    ("ratio_using_paper_count", "paper_stated_mlp_parameter_count"),
                    ("ratio_using_correct_dense_count", "correct_dense_2x64x64x1_parameter_count"),
                    ("ratio_using_released_comparator", "released_ars_comparator_parameter_count"),
                )
                denominator = float(record["chebyshev_parameter_count"])
                for ratio_key, numerator_key in ratio_contracts:
                    ratio = require_number(record, ratio_key, source)
                    expected = float(record[numerator_key]) / denominator
                    if not math.isclose(ratio, expected, rel_tol=1e-12, abs_tol=1e-12):
                        raise ValidationError(
                            f"{source}: {ratio_key}={ratio} does not equal {numerator_key}/"
                            "chebyshev_parameter_count"
                        )
                require_number(record, "paper_stated_ratio", source)
        validate_optional_number(record, "reduction_factor", source)
    elif record_type == "artifact":
        if status == "success":
            require_string(record, "artifact_path", source)
            digest = require_string(record, "artifact_sha256", source)
            if len(digest) != 64 or any(char not in "0123456789abcdef" for char in digest):
                raise ValidationError(f"{source}: artifact_sha256 is not lowercase SHA-256")
            require_integer(record, "artifact_bytes", source, minimum=0)
    elif record_type == "error":
        if status not in FAILURE_STATUSES:
            raise ValidationError(f"{source}: error record must have a failure status")
    elif record_type == "selection":
        for key in ("claim_id", "task_id", "method", "selection_rule"):
            require_string(record, key, source)
        require_integer(record, "candidate_count", source, minimum=1)
        require_integer(record, "selected_seed", source)
    elif record_type == "environment":
        require_string(record, "task_id", source)
        for key in ("packages", "hardware", "upstream_commits"):
            if not isinstance(record.get(key), dict) or not record[key]:
                raise ValidationError(f"{source}: environment.{key} must be a non-empty object")
    elif record_type == "claim_summary":
        require_string(record, "claim_id", source)
        if not isinstance(record.get("summary"), dict) or not record["summary"]:
            raise ValidationError(f"{source}: claim_summary.summary must be non-empty")
    elif record_type == "scope_limitation":
        for key in ("claim_id", "task_id", "method", "reason"):
            require_string(record, key, source)
        if status != "not_executed":
            raise ValidationError(
                f"{source}: scope_limitation must have status='not_executed'"
            )
    elif record_type == "failure":
        require_string(record, "task_id", source)
        require_string(record, "error", source)
        if status not in FAILURE_STATUSES:
            raise ValidationError(f"{source}: failure record must have a failure status")
    elif record_type == "exclusion" and status not in EXCLUSION_STATUSES:
        raise ValidationError(
            f"{source}: exclusion record must have an exclusion status"
        )


def _safe_relative_path(value: str, source: str) -> Path:
    path = Path(value)
    if path.is_absolute() or not path.parts or any(part in {"", ".", ".."} for part in path.parts):
        raise ValidationError(f"{source}: unsafe relative path {value!r}")
    return path


def validate_artifacts(
    batch_dir: Path, records: Sequence[Mapping[str, Any]], source: str
) -> None:
    artifact_records = [
        record
        for record in records
        if record["record_type"] == "artifact" and record["status"] == "success"
    ]
    declared: set[str] = set()
    batch_root = batch_dir.resolve()
    for record in artifact_records:
        relative = _safe_relative_path(str(record["artifact_path"]), source)
        canonical_relative = relative.as_posix()
        if canonical_relative in declared:
            raise ValidationError(f"{source}: duplicate artifact path {canonical_relative!r}")
        declared.add(canonical_relative)
        candidate = batch_dir / relative
        if candidate.is_symlink() or not candidate.is_file():
            raise ValidationError(f"{source}: missing or symlinked artifact {relative}")
        resolved = candidate.resolve()
        if batch_root not in resolved.parents:
            raise ValidationError(f"{source}: artifact escapes its batch: {relative}")
        if candidate.stat().st_size != record["artifact_bytes"]:
            raise ValidationError(f"{source}: artifact byte count mismatch for {relative}")
        if sha256_file(candidate) != record["artifact_sha256"]:
            raise ValidationError(f"{source}: artifact hash mismatch for {relative}")

    artifact_root = batch_dir / "artifacts"
    actual: set[str] = set()
    if artifact_root.exists():
        if artifact_root.is_symlink() or not artifact_root.is_dir():
            raise ValidationError(f"{source}: artifacts path is not a real directory")
        for path in sorted(artifact_root.rglob("*")):
            if path.is_symlink():
                raise ValidationError(f"{source}: symlink found under artifacts: {path}")
            if path.is_file():
                actual.add(path.relative_to(batch_dir).as_posix())
    if actual != declared:
        missing_records = sorted(actual - declared)
        missing_files = sorted(declared - actual)
        raise ValidationError(
            f"{source}: artifact inventory mismatch; unrecorded={missing_records}, "
            f"missing={missing_files}"
        )


def validate_input_snapshots(batch_dir: Path, manifest: Mapping[str, Any], source: str) -> None:
    hashes = manifest.get("input_hashes")
    if not isinstance(hashes, dict) or not hashes:
        raise ValidationError(f"{source}: manifest input_hashes must be a non-empty object")
    inputs_dir = batch_dir / "inputs"
    if inputs_dir.is_symlink() or not inputs_dir.is_dir():
        raise ValidationError(f"{source}: missing real inputs directory")
    actual_names = {
        path.name
        for path in inputs_dir.iterdir()
        if path.is_file() and not path.is_symlink()
    }
    if actual_names != set(hashes):
        raise ValidationError(
            f"{source}: input inventory mismatch; actual={sorted(actual_names)}, "
            f"declared={sorted(hashes)}"
        )
    for name, expected_hash in hashes.items():
        if Path(name).name != name:
            raise ValidationError(f"{source}: unsafe input snapshot name {name!r}")
        if not isinstance(expected_hash, str) or len(expected_hash) != 64:
            raise ValidationError(f"{source}: invalid input SHA-256 for {name}")
        path = inputs_dir / name
        if sha256_file(path) != expected_hash:
            raise ValidationError(f"{source}: input snapshot hash mismatch for {name}")


def _successful(records: Sequence[Mapping[str, Any]], record_type: str, claim_id: str) -> list[Mapping[str, Any]]:
    return [
        row
        for row in records
        if row["record_type"] == record_type
        and row["status"] == "success"
        and row.get("claim_id") == claim_id
    ]


def validate_protocol_coverage(
    manifest: Mapping[str, Any], records: Sequence[Mapping[str, Any]], source: str
) -> tuple[dict[str, Any], ...]:
    """Validate predeclared unit counts encoded in the immutable config."""

    config = manifest.get("config")
    if not isinstance(config, dict):
        raise ValidationError(f"{source}: manifest config must be an object")
    claims = config.get("claims")
    if not isinstance(claims, dict) or not claims:
        raise ValidationError(f"{source}: manifest config.claims must be non-empty")
    checks: list[dict[str, Any]] = []

    def exact(label: str, actual: int, expected: int) -> None:
        check = {"check": label, "actual": actual, "expected": expected, "status": "pass"}
        checks.append(check)
        if actual != expected:
            raise ValidationError(f"{source}: coverage {label} is {actual}, expected {expected}")

    def exact_counter(
        label: str,
        actual: Mapping[tuple[str, int | None], int],
        expected: Mapping[tuple[str, int | None], int],
    ) -> None:
        def printable(value: Mapping[tuple[str, int | None], int]) -> dict[str, int]:
            return {
                f"{method}|seed={seed if seed is not None else 'none'}": count
                for (method, seed), count in sorted(
                    value.items(), key=lambda item: (item[0][0], -1 if item[0][1] is None else item[0][1])
                )
            }

        actual_dict = dict(actual)
        expected_dict = dict(expected)
        check = {
            "check": label,
            "actual": printable(actual_dict),
            "expected": printable(expected_dict),
            "status": "pass",
        }
        checks.append(check)
        if actual_dict != expected_dict:
            raise ValidationError(
                f"{source}: coverage {label} mismatch; "
                f"actual={printable(actual_dict)}, expected={printable(expected_dict)}"
            )

    def method_seed_counter(rows: Sequence[Mapping[str, Any]]) -> Counter[tuple[str, int | None]]:
        return Counter(
            (
                str(row.get("method")),
                row.get("seed") if isinstance(row.get("seed"), int) else None,
            )
            for row in rows
        )

    def validate_mountaincar_start_grids(
        label: str,
        rows: Sequence[Mapping[str, Any]],
        expected_groups: Mapping[tuple[str, int | None], int],
    ) -> None:
        groups: dict[tuple[str, int | None], list[float]] = defaultdict(list)
        for row in rows:
            key = (
                str(row.get("method")),
                row.get("seed") if isinstance(row.get("seed"), int) else None,
            )
            value = row.get("start_position")
            if isinstance(value, bool) or not isinstance(value, (int, float)):
                raise ValidationError(f"{source}: {label} has a non-numeric start position")
            groups[key].append(float(value))
        if set(groups) != set(expected_groups):
            raise ValidationError(
                f"{source}: {label} grid groups differ from expected method/seed groups"
            )
        reference: tuple[float, ...] | None = None
        for key, expected_count in expected_groups.items():
            values = tuple(sorted(groups[key]))
            if len(values) != expected_count or len(set(values)) != expected_count:
                raise ValidationError(
                    f"{source}: {label} {key} does not contain {expected_count} unique starts"
                )
            if not math.isclose(values[0], -0.6, abs_tol=1e-12) or not math.isclose(
                values[-1], -0.4, abs_tol=1e-12
            ):
                raise ValidationError(
                    f"{source}: {label} {key} does not span the preregistered [-0.6, -0.4] range"
                )
            if reference is None:
                reference = values
            elif values != reference:
                raise ValidationError(f"{source}: {label} start grids are not identical")
        checks.append(
            {
                "check": label,
                "actual": f"{len(expected_groups)} identical complete grids",
                "expected": f"{len(expected_groups)} identical complete grids",
                "status": "pass",
            }
        )

    def validate_pendulum_grids(
        label: str,
        rows: Sequence[Mapping[str, Any]],
        expected_groups: Mapping[tuple[str, int | None], int],
    ) -> None:
        groups: dict[tuple[str, int | None], list[tuple[float, float]]] = defaultdict(list)
        for row in rows:
            key = (
                str(row.get("method")),
                row.get("seed") if isinstance(row.get("seed"), int) else None,
            )
            angle = row.get("start_angle")
            velocity = row.get("start_angular_velocity")
            if (
                isinstance(angle, bool)
                or not isinstance(angle, (int, float))
                or isinstance(velocity, bool)
                or not isinstance(velocity, (int, float))
            ):
                raise ValidationError(f"{source}: {label} has a non-numeric grid coordinate")
            groups[key].append((float(angle), float(velocity)))
        if set(groups) != set(expected_groups):
            raise ValidationError(
                f"{source}: {label} grid groups differ from expected method/seed groups"
            )
        reference: tuple[tuple[float, float], ...] | None = None
        for key, expected_count in expected_groups.items():
            values = tuple(sorted(groups[key]))
            if len(values) != expected_count or len(set(values)) != expected_count:
                raise ValidationError(
                    f"{source}: {label} {key} does not contain {expected_count} unique points"
                )
            if reference is None:
                reference = values
            elif values != reference:
                raise ValidationError(f"{source}: {label} evaluation grids are not identical")
        checks.append(
            {
                "check": label,
                "actual": f"{len(expected_groups)} identical complete grids",
                "expected": f"{len(expected_groups)} identical complete grids",
                "status": "pass",
            }
        )

    claim1 = claims.get("claim1")
    if isinstance(claim1, dict) and claim1.get("enabled") is True:
        exact("claim1.proof_certificates", len(_successful(records, "proof_certificate", "claim1")), 1)

    claim2 = claims.get("claim2")
    if isinstance(claim2, dict) and claim2.get("enabled") is True:
        starts = claim2.get("start_count")
        baselines = claim2.get("baselines")
        if not isinstance(starts, int) or starts < 2 or not isinstance(baselines, list):
            raise ValidationError(f"{source}: invalid Claim 2 coverage config")
        expected = starts * (1 + len(baselines))
        episode_rows = _successful(records, "episode", "claim2")
        exact("claim2.episodes", len(episode_rows), expected)
        expected_groups = {
            (method, None): starts
            for method in ("analytic_two_phase", *(str(value) for value in baselines))
        }
        exact_counter(
            "claim2.method_episode_counts",
            method_seed_counter(episode_rows),
            expected_groups,
        )
        validate_mountaincar_start_grids(
            "claim2.fixed_start_grids", episode_rows, expected_groups
        )
        parameter_methods = {
            str(row.get("method"))
            for row in records
            if row["record_type"] == "parameter_audit" and row["status"] == "success"
        }
        missing = sorted(set(str(value) for value in baselines) - parameter_methods)
        if missing:
            raise ValidationError(f"{source}: missing executable parameter audits for {missing}")
        checks.append(
            {
                "check": "claim2.baseline_parameter_methods",
                "actual": sorted(parameter_methods & set(str(value) for value in baselines)),
                "expected": sorted(str(value) for value in baselines),
                "status": "pass",
            }
        )

    claim3 = claims.get("claim3")
    if isinstance(claim3, dict) and claim3.get("enabled") is True:
        seeds = claim3.get("seeds")
        starts = claim3.get("evaluation_start_count")
        if not isinstance(seeds, list) or not seeds or not isinstance(starts, int) or starts < 2:
            raise ValidationError(f"{source}: invalid Claim 3 coverage config")
        exact("claim3.training_runs", len(_successful(records, "training_run", "claim3")), len(seeds))
        training_rows = _successful(records, "training_run", "claim3")
        expected_training_groups = {("ch3_ars", int(seed)): 1 for seed in seeds}
        exact_counter(
            "claim3.method_seed_training_runs",
            method_seed_counter(training_rows),
            expected_training_groups,
        )
        episode_rows = _successful(records, "episode", "claim3")
        exact("claim3.episodes", len(episode_rows), len(seeds) * starts)
        expected_episode_groups = {
            ("ch3_ars", int(seed)): starts for seed in seeds
        }
        exact_counter(
            "claim3.method_seed_episode_counts",
            method_seed_counter(episode_rows),
            expected_episode_groups,
        )
        validate_mountaincar_start_grids(
            "claim3.fixed_start_grids", episode_rows, expected_episode_groups
        )
        selection_rows = _successful(records, "selection", "claim3")
        if (
            len(selection_rows) != 1
            or selection_rows[0].get("method") != "ch3_ars"
            or selection_rows[0].get("candidate_count") != len(seeds)
            or selection_rows[0].get("selected_seed") not in seeds
        ):
            raise ValidationError(
                f"{source}: Claim 3 must contain one complete CH-3-ARS selection"
            )
        chebyshev_parameter_rows = [
            row
            for row in _successful(records, "parameter_audit", "claim3")
            if "ch" in str(row.get("method", "")).lower()
            or "chebyshev" in str(row.get("method", "")).lower()
            or "chebyshev_parameter_count" in row
        ]
        if not chebyshev_parameter_rows:
            raise ValidationError(f"{source}: Claim 3 has no Chebyshev parameter audit")
        checks.append(
            {
                "check": "claim3.chebyshev_parameter_audit",
                "actual": len(chebyshev_parameter_rows),
                "expected": ">=1",
                "status": "pass",
            }
        )

    claim4 = claims.get("claim4")
    if isinstance(claim4, dict) and claim4.get("enabled") is True:
        ppo_seeds = claim4.get("ppo_seeds")
        reinforce_seeds = claim4.get("reinforce_seeds")
        start_count = claim4.get("evaluation_start_count")
        selection_count = claim4.get("reinforce_selection_episodes")
        if (
            not isinstance(ppo_seeds, list)
            or not ppo_seeds
            or not isinstance(reinforce_seeds, list)
            or not reinforce_seeds
            or not isinstance(start_count, int)
            or start_count < 2
            or not isinstance(selection_count, int)
            or selection_count < 1
        ):
            raise ValidationError(f"{source}: invalid Claim 4 coverage config")
        exact(
            "claim4.training_runs",
            len(_successful(records, "training_run", "claim4")),
            len(ppo_seeds) + len(reinforce_seeds),
        )
        training_rows = _successful(records, "training_run", "claim4")
        expected_training_groups = {
            **{("ch3_ppo", int(seed)): 1 for seed in ppo_seeds},
            **{
                ("ch3_reinforce_adamw_seeded", int(seed)): 1
                for seed in reinforce_seeds
            },
        }
        exact_counter(
            "claim4.method_seed_training_runs",
            method_seed_counter(training_rows),
            expected_training_groups,
        )
        expected_episodes = (
            len(ppo_seeds) * start_count
            + len(reinforce_seeds) * selection_count
            + start_count
        )
        episode_rows = _successful(records, "episode", "claim4")
        exact("claim4.episodes", len(episode_rows), expected_episodes)
        selection_rows = _successful(records, "selection", "claim4")
        selection_by_method = {
            str(row.get("method")): row for row in selection_rows
        }
        if len(selection_rows) != 2 or set(selection_by_method) != {
            "ch3_ppo",
            "ch3_reinforce_adamw_seeded",
        }:
            raise ValidationError(
                f"{source}: Claim 4 must contain exactly one PPO and one REINFORCE selection"
            )
        for method, candidates in (
            ("ch3_ppo", len(ppo_seeds)),
            ("ch3_reinforce_adamw_seeded", len(reinforce_seeds)),
        ):
            if selection_by_method[method].get("candidate_count") != candidates:
                raise ValidationError(
                    f"{source}: Claim 4 selection {method} has the wrong candidate count"
                )
        selected_reinforce_seed = selection_by_method[
            "ch3_reinforce_adamw_seeded"
        ].get("selected_seed")
        if not isinstance(selected_reinforce_seed, int):
            raise ValidationError(f"{source}: Claim 4 REINFORCE selected seed is invalid")
        expected_episode_groups = {
            **{("ch3_ppo", int(seed)): start_count for seed in ppo_seeds},
            **{
                ("ch3_reinforce_adamw_seeded", int(seed)): selection_count
                for seed in reinforce_seeds
            },
            ("ch3_reinforce_adamw_seeded_selected", selected_reinforce_seed): start_count,
        }
        exact_counter(
            "claim4.method_seed_episode_counts",
            method_seed_counter(episode_rows),
            expected_episode_groups,
        )
        fixed_grid_rows = [
            row
            for row in episode_rows
            if row.get("method")
            in {"ch3_ppo", "ch3_reinforce_adamw_seeded_selected"}
        ]
        fixed_grid_groups = {
            key: count
            for key, count in expected_episode_groups.items()
            if key[0] in {"ch3_ppo", "ch3_reinforce_adamw_seeded_selected"}
        }
        validate_mountaincar_start_grids(
            "claim4.fixed_start_grids", fixed_grid_rows, fixed_grid_groups
        )

    claim5 = claims.get("claim5")
    if isinstance(claim5, dict) and claim5.get("enabled") is True:
        seeds = claim5.get("seeds")
        points = claim5.get("grid_points_per_dimension")
        baselines = claim5.get("baselines")
        if (
            not isinstance(seeds, list)
            or not seeds
            or not isinstance(points, int)
            or points < 2
            or not isinstance(baselines, list)
        ):
            raise ValidationError(f"{source}: invalid Claim 5 coverage config")
        training_rows = _successful(records, "training_run", "claim5")
        exact("claim5.training_runs", len(training_rows), len(seeds))
        expected_training_groups = {
            ("ch6_ars_pendulum", int(seed)): 1 for seed in seeds
        }
        exact_counter(
            "claim5.method_seed_training_runs",
            method_seed_counter(training_rows),
            expected_training_groups,
        )
        # The runner keeps every learned-policy grid, emits a separately named
        # copy of the selected grid, evaluates the author-exact released
        # baseline, and adds one corrected-reset sensitivity grid disclosed in
        # DRIFT.md D-003.
        expected_episodes = (len(seeds) + len(baselines) + 2) * points * points
        episode_rows = _successful(records, "episode", "claim5")
        exact("claim5.episodes", len(episode_rows), expected_episodes)
        if [str(value) for value in baselines] != ["ars"]:
            raise ValidationError(
                f"{source}: Claim 5 canonical contract expects exactly the released ARS baseline"
            )
        selection_rows = _successful(records, "selection", "claim5")
        if (
            len(selection_rows) != 1
            or selection_rows[0].get("method") != "ch6_ars_pendulum"
            or selection_rows[0].get("candidate_count") != len(seeds)
            or not isinstance(selection_rows[0].get("selected_seed"), int)
        ):
            raise ValidationError(
                f"{source}: Claim 5 must contain one complete CH-6-ARS selection"
            )
        selected_seed = int(selection_rows[0]["selected_seed"])
        expected_episode_groups = {
            **{
                ("ch6_ars_pendulum", int(seed)): points * points
                for seed in seeds
            },
            ("ch6_ars_pendulum_selected", selected_seed): points * points,
            ("ars_baseline_pendulum_released", 0): points * points,
            (
                "ars_baseline_pendulum_released_corrected_reset",
                0,
            ): points * points,
        }
        exact_counter(
            "claim5.method_seed_episode_counts",
            method_seed_counter(episode_rows),
            expected_episode_groups,
        )
        validate_pendulum_grids(
            "claim5.fixed_evaluation_grids", episode_rows, expected_episode_groups
        )

    return tuple(checks)


def canonicalize_record(record: Mapping[str, Any]) -> dict[str, Any]:
    """Add declared canonical fields without removing any raw field."""

    output = dict(record)
    if record["record_type"] == "episode":
        if "start_angle" in record:
            output["initial_angle"] = record["start_angle"]
        if "start_angular_velocity" in record:
            output["initial_angular_velocity"] = record["start_angular_velocity"]
    elif record["record_type"] == "training_run":
        configuration = record.get("configuration")
        if isinstance(configuration, dict) and isinstance(configuration.get("environment"), str):
            output["environment"] = configuration["environment"]
        if "actual_timesteps" in record:
            output["timesteps"] = record["actual_timesteps"]
            output["timesteps_source"] = "actual_timesteps"
        elif "configured_timesteps" in record:
            output["timesteps"] = record["configured_timesteps"]
            output["timesteps_source"] = "configured_timesteps"
        if "actual_training_episodes" in record:
            output["episodes"] = record["actual_training_episodes"]
            output["episodes_source"] = "actual_training_episodes"
        elif "configured_episodes" in record:
            output["episodes"] = record["configured_episodes"]
            output["episodes_source"] = "configured_episodes"
        for source_key, prefix in (
            ("evaluation_summary", "evaluation"),
            ("selection_summary", "selection"),
        ):
            summary = record.get(source_key)
            if isinstance(summary, dict):
                for metric in ("count", "mean", "std", "min", "max"):
                    output[f"{prefix}_{metric}_return" if metric != "count" else f"{prefix}_count"] = summary[metric]
        if "task_id" not in output:
            output["task_id"] = f"{record['method']}_seed_{record['seed']}"
            output["task_id_source"] = "canonical_method_seed"
        if "attempt_id" not in output:
            output["attempt_id"] = 1
            output["attempt_id_source"] = "canonical_single_raw_attempt"
    return output


def load_batch(batch_dir: Path) -> BatchData:
    batch_dir = batch_dir.resolve()
    source = str(batch_dir)
    manifest_path = batch_dir / "manifest.json"
    records_path = batch_dir / "records.jsonl"
    if not manifest_path.is_file() or manifest_path.is_symlink():
        raise ValidationError(f"{source}: missing real manifest.json")
    if not records_path.is_file() or records_path.is_symlink():
        raise ValidationError(f"{source}: missing real records.jsonl")

    manifest = require_object(
        parse_json(manifest_path.read_text(encoding="utf-8"), str(manifest_path)),
        str(manifest_path),
    )
    for key in (
        "schema_version",
        "experiment_id",
        "spec_version",
        "paper_version",
        "openreview_id",
        "batch_id",
        "status",
        "records_sha256",
    ):
        require_string(manifest, key, str(manifest_path))
    if manifest["schema_version"] != SUPPORTED_SCHEMA_VERSION:
        raise ValidationError(
            f"{source}: unsupported schema_version {manifest['schema_version']!r}"
        )
    if manifest["status"] != "success":
        raise ValidationError(
            f"{source}: raw manifest status is {manifest['status']!r}, not 'success'"
        )
    if manifest.get("error") not in (None, ""):
        raise ValidationError(f"{source}: successful manifest contains an error")
    if batch_dir.name != manifest["batch_id"]:
        raise ValidationError(
            f"{source}: directory name and manifest batch_id do not match"
        )
    actual_records_hash = sha256_file(records_path)
    if actual_records_hash != manifest["records_sha256"]:
        raise ValidationError(f"{source}: records.jsonl SHA-256 mismatch")

    records: list[dict[str, Any]] = []
    seen_hashes: set[str] = set()
    seen_task_ids: set[tuple[str, str, int]] = set()
    seen_training_units: set[tuple[str, int, str]] = set()
    with records_path.open("r", encoding="utf-8", newline="") as handle:
        for line_number, line in enumerate(handle, start=1):
            if not line.strip():
                raise ValidationError(f"{records_path}:{line_number}: blank JSONL line")
            record = require_object(
                parse_json(line, f"{records_path}:{line_number}"),
                f"{records_path}:{line_number}",
            )
            record_source = f"{records_path}:{line_number}"
            validate_record(record, record_source)
            for key in (
                "schema_version",
                "experiment_id",
                "spec_version",
                "paper_version",
                "openreview_id",
                "batch_id",
            ):
                if record[key] != manifest[key]:
                    raise ValidationError(
                        f"{record_source}: {key} differs from its manifest"
                    )
            digest = hashlib.sha256(canonical_json(record).encode("utf-8")).hexdigest()
            if digest in seen_hashes:
                raise ValidationError(f"{record_source}: duplicate canonical record")
            seen_hashes.add(digest)
            if "task_id" in record:
                attempt_id = record.get("attempt_id", 1)
                if isinstance(attempt_id, bool) or not isinstance(attempt_id, int):
                    raise ValidationError(f"{record_source}: invalid attempt_id")
                identity = (record["record_type"], str(record["task_id"]), attempt_id)
                if identity in seen_task_ids:
                    raise ValidationError(f"{record_source}: duplicate task/attempt identity {identity}")
                seen_task_ids.add(identity)
            if record["record_type"] == "training_run":
                training_identity = (
                    str(record["method"]),
                    int(record["seed"]),
                    str(record["panel_id"]),
                )
                if training_identity in seen_training_units:
                    raise ValidationError(
                        f"{record_source}: duplicate training method/seed/panel {training_identity}"
                    )
                seen_training_units.add(training_identity)
            enriched = canonicalize_record(record)
            enriched["source_batch"] = manifest["batch_id"]
            enriched["source_line_number"] = line_number
            enriched["record_sha256"] = digest
            records.append(enriched)

    if not records:
        raise ValidationError(f"{source}: records.jsonl is empty")
    declared_counts = manifest.get("record_counts")
    if not isinstance(declared_counts, dict):
        raise ValidationError(f"{source}: manifest record_counts must be an object")
    if any(isinstance(value, bool) or not isinstance(value, int) or value < 0 for value in declared_counts.values()):
        raise ValidationError(f"{source}: manifest record_counts contains invalid counts")
    actual_counts = Counter(str(record["record_type"]) for record in records)
    if dict(sorted(actual_counts.items())) != dict(sorted(declared_counts.items())):
        raise ValidationError(
            f"{source}: record_counts mismatch; actual={dict(sorted(actual_counts.items()))}, "
            f"declared={dict(sorted(declared_counts.items()))}"
        )

    validate_input_snapshots(batch_dir, manifest, source)
    validate_artifacts(batch_dir, records, source)
    coverage = validate_protocol_coverage(manifest, records, source)
    return BatchData(
        directory=batch_dir,
        manifest=manifest,
        records=tuple(records),
        records_sha256=actual_records_hash,
        manifest_sha256=sha256_file(manifest_path),
        coverage_checks=coverage,
    )


def discover_batches(raw_roots: Sequence[Path]) -> list[Path]:
    discovered: dict[Path, None] = {}
    for raw_root in raw_roots:
        root = raw_root.resolve()
        if not root.exists():
            raise ValidationError(f"Raw root does not exist: {root}")
        if (root / "manifest.json").is_file() or (root / "records.jsonl").is_file():
            candidates = [root]
        elif root.is_dir():
            candidates = sorted(
                path.parent
                for path in root.rglob("manifest.json")
                if (path.parent / "records.jsonl").is_file()
            )
        else:
            raise ValidationError(f"Raw root is not a directory: {root}")
        if not candidates:
            raise ValidationError(f"No evidence batches found under {root}")
        for candidate in candidates:
            discovered[candidate.resolve()] = None
    return sorted(discovered)


def validate_cross_batch_identity(batches: Sequence[BatchData]) -> None:
    if not batches:
        raise ValidationError("At least one batch is required")
    identity_keys = (
        "schema_version",
        "experiment_id",
        "spec_version",
        "paper_version",
        "openreview_id",
    )
    expected = {key: batches[0].manifest[key] for key in identity_keys}
    seen_ids: set[str] = set()
    for batch in batches:
        actual = {key: batch.manifest[key] for key in identity_keys}
        if actual != expected:
            raise ValidationError(
                f"Cross-batch identity mismatch in {batch.directory}: {actual} != {expected}"
            )
        batch_id = str(batch.manifest["batch_id"])
        if batch_id in seen_ids:
            raise ValidationError(f"Duplicate batch_id across inputs: {batch_id}")
        seen_ids.add(batch_id)


def table_for_record_type(record_type: str) -> str:
    mapping = {
        "episode": "episodes.csv",
        "training_run": "training_runs.csv",
        "training_metric": "training_metrics.csv",
        "proof_certificate": "proof_certificates.csv",
        "parameter_audit": "parameter_audit.csv",
        "artifact": "artifacts.csv",
        "selection": "selections.csv",
        "environment": "environments.csv",
        "claim_summary": "claim_summaries.csv",
        "scope_limitation": "exclusions.csv",
        "failure": "errors.csv",
        "error": "errors.csv",
        "exclusion": "exclusions.csv",
    }
    return mapping[record_type]


def sort_component(value: Any) -> tuple[int, Any]:
    if value is None or value == "":
        return (0, "")
    if isinstance(value, bool):
        return (1, int(value))
    if isinstance(value, (int, float)):
        return (2, float(value))
    return (3, canonical_json(value) if isinstance(value, (dict, list)) else str(value))


SORT_FIELDS: dict[str, tuple[str, ...]] = {
    "episodes.csv": (
        "experiment_id",
        "batch_id",
        "claim_id",
        "panel_id",
        "environment",
        "method",
        "seed",
        "start_position",
        "initial_angle",
        "initial_angular_velocity",
        "task_id",
        "attempt_id",
        "source_line_number",
    ),
    "training_runs.csv": (
        "experiment_id",
        "batch_id",
        "claim_id",
        "panel_id",
        "environment",
        "method",
        "seed",
        "attempt_id",
        "task_id",
        "source_line_number",
    ),
    "training_metrics.csv": (
        "experiment_id",
        "batch_id",
        "claim_id",
        "panel_id",
        "method",
        "seed",
        "timesteps",
        "training_episode",
        "task_id",
        "source_line_number",
    ),
    "proof_certificates.csv": (
        "experiment_id",
        "batch_id",
        "claim_id",
        "task_id",
        "source_line_number",
    ),
    "parameter_audit.csv": (
        "experiment_id",
        "batch_id",
        "claim_id",
        "method",
        "count_source",
        "task_id",
        "source_line_number",
    ),
    "artifacts.csv": ("experiment_id", "batch_id", "artifact_path", "source_line_number"),
    "selections.csv": (
        "experiment_id",
        "batch_id",
        "claim_id",
        "method",
        "task_id",
        "source_line_number",
    ),
    "environments.csv": ("experiment_id", "batch_id", "task_id", "source_line_number"),
    "claim_summaries.csv": (
        "experiment_id",
        "batch_id",
        "claim_id",
        "source_line_number",
    ),
    "errors.csv": ("experiment_id", "batch_id", "claim_id", "task_id", "source_line_number"),
    "exclusions.csv": ("experiment_id", "batch_id", "claim_id", "task_id", "source_line_number"),
}


def deterministic_sort(rows: Iterable[dict[str, Any]], table_name: str) -> list[dict[str, Any]]:
    fields = SORT_FIELDS[table_name]
    return sorted(rows, key=lambda row: tuple(sort_component(row.get(field)) for field in fields))


def build_record_tables(batches: Sequence[BatchData]) -> dict[str, list[dict[str, Any]]]:
    tables = {name: [] for name in TABLE_CORE_COLUMNS if name != "summaries.csv"}
    assigned_hashes: set[str] = set()
    all_hashes: set[str] = set()
    for batch in batches:
        for record in batch.records:
            digest = str(record["record_sha256"])
            all_hashes.add(digest)
            target = table_for_record_type(str(record["record_type"]))
            tables[target].append(dict(record))
            assigned_hashes.add(digest)
            if record["status"] in FAILURE_STATUSES and target != "errors.csv":
                tables["errors.csv"].append(dict(record))
            if record["status"] in EXCLUSION_STATUSES and target != "exclusions.csv":
                tables["exclusions.csv"].append(dict(record))
    if all_hashes != assigned_hashes:
        raise AssertionError("Internal error: a validated raw record was not assigned")
    for name in tables:
        tables[name] = deterministic_sort(tables[name], name)
    return tables


SUMMARY_GROUP_FIELDS = (
    "experiment_id",
    "spec_version",
    "paper_version",
    "openreview_id",
    "batch_id",
    "claim_id",
    "panel_id",
    "environment",
    "method",
)


def summarize_group(
    summary_type: str,
    group_key: tuple[Any, ...],
    rows: Sequence[Mapping[str, Any]],
    metric_name: str,
) -> dict[str, Any]:
    successful = [row for row in rows if row.get("status") == "success"]
    values: list[float] = []
    for row in successful:
        value = row.get(metric_name)
        if value is None:
            continue
        if isinstance(value, bool) or not isinstance(value, (int, float)) or not math.isfinite(float(value)):
            raise ValidationError(
                f"Cannot summarize non-numeric {metric_name!r} in record {row.get('record_sha256')}"
            )
        values.append(float(value))
    result = dict(zip(SUMMARY_GROUP_FIELDS, group_key))
    result.update(
        {
            "summary_type": summary_type,
            "metric_name": metric_name,
            "record_count": len(rows),
            "success_count": len(successful),
            "error_count": sum(row.get("status") in FAILURE_STATUSES for row in rows),
            "exclusion_count": sum(row.get("status") in EXCLUSION_STATUSES for row in rows),
            "value_count": len(values),
            "mean": statistics.fmean(values) if values else None,
            "std_population": statistics.pstdev(values) if values else None,
            "min": min(values) if values else None,
            "max": max(values) if values else None,
        }
    )
    return result


def build_summaries(tables: Mapping[str, Sequence[Mapping[str, Any]]]) -> list[dict[str, Any]]:
    output: list[dict[str, Any]] = []
    episode_groups: dict[tuple[Any, ...], list[Mapping[str, Any]]] = defaultdict(list)
    for row in tables["episodes.csv"]:
        episode_groups[tuple(row.get(field) for field in SUMMARY_GROUP_FIELDS)].append(row)
    for key, rows in sorted(episode_groups.items(), key=lambda item: tuple(sort_component(v) for v in item[0])):
        output.append(summarize_group("episodes", key, rows, "return"))

    training_groups: dict[tuple[Any, ...], list[Mapping[str, Any]]] = defaultdict(list)
    for row in tables["training_runs.csv"]:
        training_groups[tuple(row.get(field) for field in SUMMARY_GROUP_FIELDS)].append(row)
    metric_fields = (
        "evaluation_mean_return",
        "selection_mean_return",
        "mean_return",
        "return",
    )
    for key, rows in sorted(training_groups.items(), key=lambda item: tuple(sort_component(v) for v in item[0])):
        present = [metric for metric in metric_fields if any(row.get(metric) is not None for row in rows)]
        if not present:
            output.append(summarize_group("training_runs", key, rows, ""))
        else:
            output.extend(summarize_group("training_runs", key, rows, metric) for metric in present)
    return output


def csv_value(value: Any) -> str | int:
    if value is None:
        return ""
    if isinstance(value, bool):
        return "true" if value else "false"
    if isinstance(value, int):
        return value
    if isinstance(value, float):
        if not math.isfinite(value):
            raise ValidationError("Cannot write non-finite float to canonical CSV")
        return format(value, ".17g")
    if isinstance(value, (dict, list)):
        return canonical_json(value)
    return str(value)


def ordered_columns(table_name: str, rows: Sequence[Mapping[str, Any]]) -> list[str]:
    preferred = list(dict.fromkeys(TABLE_CORE_COLUMNS[table_name]))
    extras = sorted({key for row in rows for key in row} - set(preferred))
    return [*preferred, *extras]


def write_csv(path: Path, table_name: str, rows: Sequence[Mapping[str, Any]]) -> list[str]:
    columns = ordered_columns(table_name, rows)
    with path.open("w", encoding="utf-8", newline="") as handle:
        writer = csv.DictWriter(
            handle,
            fieldnames=columns,
            extrasaction="raise",
            lineterminator="\n",
        )
        writer.writeheader()
        for row in rows:
            writer.writerow({column: csv_value(row.get(column)) for column in columns})
    return columns


def publish_directory(staging: Path, destination: Path, replace: bool) -> None:
    if destination.is_symlink():
        raise ValidationError(f"Refusing to publish through a symlink: {destination}")
    destination = destination.resolve()
    if destination.exists() and not destination.is_dir():
        raise ValidationError(f"Output path exists and is not a directory: {destination}")
    if destination.exists() and not replace:
        raise ValidationError(
            f"Output directory already exists: {destination}; use --replace for derived outputs"
        )
    destination.parent.mkdir(parents=True, exist_ok=True)
    if not destination.exists():
        os.replace(staging, destination)
        return
    backup = destination.parent / f".{destination.name}.old-{os.getpid()}"
    if backup.exists():
        raise ValidationError(f"Refusing to overwrite stale backup directory: {backup}")
    os.replace(destination, backup)
    try:
        os.replace(staging, destination)
    except BaseException:
        os.replace(backup, destination)
        raise
    shutil.rmtree(backup)


def build_tables(raw_roots: Sequence[Path], output_root: Path, replace: bool) -> dict[str, Any]:
    batch_dirs = discover_batches(raw_roots)
    output_resolved = output_root.resolve()
    for batch_dir in batch_dirs:
        if output_resolved == batch_dir or batch_dir in output_resolved.parents:
            raise ValidationError("Canonical output must not overwrite or be nested in a raw batch")
    batches = [load_batch(path) for path in batch_dirs]
    validate_cross_batch_identity(batches)
    tables = build_record_tables(batches)
    tables["summaries.csv"] = build_summaries(tables)

    output_root.parent.mkdir(parents=True, exist_ok=True)
    staging = Path(
        tempfile.mkdtemp(prefix=f".{output_root.name}.tmp-", dir=output_root.parent.resolve())
    )
    try:
        table_manifest: dict[str, Any] = {}
        for table_name in TABLE_CORE_COLUMNS:
            rows = tables[table_name]
            columns = write_csv(staging / table_name, table_name, rows)
            table_manifest[table_name] = {
                "path": table_name,
                "rows": len(rows),
                "columns": columns,
                "sha256": sha256_file(staging / table_name),
            }
        identity = batches[0].manifest
        manifest = {
            "schema_version": SUPPORTED_SCHEMA_VERSION,
            "builder": "build_tables.py",
            "builder_version": BUILDER_VERSION,
            "builder_sha256": sha256_file(Path(__file__).resolve()),
            "command": build_command(raw_roots, output_root, replace),
            "status": "success",
            "experiment_id": identity["experiment_id"],
            "spec_version": identity["spec_version"],
            "paper_version": identity["paper_version"],
            "openreview_id": identity["openreview_id"],
            "source_batches": [
                {
                    "batch_id": batch.manifest["batch_id"],
                    "directory_name": batch.directory.name,
                    "manifest_sha256": batch.manifest_sha256,
                    "records_sha256": batch.records_sha256,
                    "record_counts": batch.manifest["record_counts"],
                    "coverage_checks": list(batch.coverage_checks),
                }
                for batch in batches
            ],
            "record_preservation": {
                "unknown_record_policy": "fail",
                "failure_records": "errors.csv (and their typed table when applicable)",
                "exclusion_records": "exclusions.csv (and their typed table when applicable)",
            },
            "summary_contract": {
                "standard_deviation": "population",
                "successful_values_only": True,
                "status_counts_include_all_records": True,
                "group_fields": list(SUMMARY_GROUP_FIELDS),
            },
            "tables": table_manifest,
        }
        (staging / "MANIFEST.json").write_text(
            json.dumps(manifest, ensure_ascii=False, allow_nan=False, indent=2, sort_keys=True)
            + "\n",
            encoding="utf-8",
        )
        publish_directory(staging, output_root, replace)
        return manifest
    except BaseException:
        if staging.exists():
            shutil.rmtree(staging)
        raise


def build_command(
    raw_roots: Sequence[Path], output_root: Path, replace: bool
) -> list[str]:
    """Return the canonical, interpreter-independent command for this build."""
    command = [str(Path(__file__).resolve())]
    for raw_root in raw_roots:
        command.extend(("--raw-root", str(raw_root.resolve())))
    command.extend(("--output-root", str(output_root.resolve())))
    if replace:
        command.append("--replace")
    return command


def parse_args(argv: Sequence[str] | None = None) -> argparse.Namespace:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument(
        "--raw-root",
        type=Path,
        action="append",
        required=True,
        help="Raw batch directory or a parent containing batches; repeatable.",
    )
    parser.add_argument("--output-root", type=Path, required=True)
    parser.add_argument(
        "--replace",
        action="store_true",
        help="Atomically replace an existing derived output directory.",
    )
    return parser.parse_args(argv)


def main(argv: Sequence[str] | None = None) -> int:
    args = parse_args(argv)
    try:
        manifest = build_tables(args.raw_root, args.output_root, args.replace)
    except (OSError, ValidationError, ValueError) as exc:
        print(f"build_tables: ERROR: {exc}", file=sys.stderr)
        return 2
    total_rows = sum(table["rows"] for table in manifest["tables"].values())
    print(
        f"Built {len(manifest['tables'])} canonical tables from "
        f"{len(manifest['source_batches'])} batch(es), {total_rows} table rows."
    )
    print(args.output_root.resolve() / "MANIFEST.json")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())