release/v1/scripts/run_reproduction.py

# /// script
# requires-python = "==3.10.*"
# dependencies = [
#   "gym==0.26.2",
#   "gymnasium==1.0.0",
#   "huggingface-hub==0.26.2",
#   "huggingface-sb3==3.0",
#   "numpy==1.26.4",
#   "pandas==2.2.3",
#   "plotly==5.24.1",
#   "psutil==6.1.0",
#   "pygame==2.6.1",
#   "pyyaml==6.0.2",
#   "rl-zoo3==2.4.0",
#   "sb3-contrib==2.4.0",
#   "scipy==1.14.0",
#   "stable-baselines3==2.4.0",
#   "tensorboard==2.18.0",
#   "tensordict==0.5.0",
#   "torch==2.4.0",
#   "torchrl==0.5.0",
# ]
# ///
"""Fail-closed runner for the ICML 2026 Chebyshev Policies reproduction.

The script is deliberately self-contained so the exact same file can be sent
to a Hugging Face Job. It imports the authors' code only after fetching and
verifying immutable Git commits and baseline checkpoint hashes.
"""

from __future__ import annotations

import argparse
import contextlib
import csv
import hashlib
import importlib.metadata
import io
import json
import math
import multiprocessing
import os
import platform
import random
import resource
import shutil
import subprocess
import sys
import time
import traceback
from collections import Counter
from concurrent.futures import ProcessPoolExecutor, as_completed
from dataclasses import dataclass
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Iterable, Iterator, Mapping, Sequence

import numpy as np
import psutil


SCHEMA_VERSION = "1.0.0"
EXPERIMENT_ID = "icml2026_chebyshev_v1"
SPEC_VERSION = "1.0.0"
PAPER_VERSION = "2605.22305v4"
OPENREVIEW_ID = "aNWIVNjocB"
ANALYTIC_C1 = 4.3346
ANALYTIC_C2 = 4.8358
MOUNTAINCAR_START_LOW = -0.6
MOUNTAINCAR_START_HIGH = -0.4
UPSTREAM_NAMES = ("paper", "companion")
DEPENDENCIES_TO_RECORD = (
    "gym",
    "gymnasium",
    "huggingface-hub",
    "huggingface-sb3",
    "numpy",
    "psutil",
    "pyyaml",
    "rl-zoo3",
    "sb3-contrib",
    "scipy",
    "stable-baselines3",
    "tensorboard",
    "tensordict",
    "torch",
    "torchrl",
)


def utc_now() -> str:
    return datetime.now(timezone.utc).isoformat()


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 jsonable(value: Any) -> Any:
    if isinstance(value, Path):
        return str(value)
    if isinstance(value, np.ndarray):
        return value.tolist()
    if isinstance(value, np.generic):
        return value.item()
    if isinstance(value, Mapping):
        return {str(key): jsonable(item) for key, item in value.items()}
    if isinstance(value, (list, tuple)):
        return [jsonable(item) for item in value]
    if isinstance(value, float) and not math.isfinite(value):
        raise ValueError(f"Non-finite float cannot enter evidence: {value!r}")
    return value


def write_json_once(path: Path, payload: Mapping[str, Any]) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    if path.exists():
        raise FileExistsError(f"Refusing to overwrite evidence file: {path}")
    path.write_text(
        json.dumps(jsonable(payload), indent=2, sort_keys=True) + "\n",
        encoding="utf-8",
    )


def package_versions() -> dict[str, str]:
    versions: dict[str, str] = {}
    for name in DEPENDENCIES_TO_RECORD:
        versions[name] = importlib.metadata.version(name)
    return versions


def hardware_snapshot() -> dict[str, Any]:
    import torch

    return {
        "platform": platform.platform(),
        "python": sys.version,
        "machine": platform.machine(),
        "logical_cpu_count": psutil.cpu_count(logical=True),
        "physical_cpu_count": psutil.cpu_count(logical=False),
        "system_ram_bytes": psutil.virtual_memory().total,
        "cuda_available": torch.cuda.is_available(),
        "cuda_device_count": torch.cuda.device_count(),
        "cuda_devices": [
            torch.cuda.get_device_name(index)
            for index in range(torch.cuda.device_count())
        ],
        "environment": {
            key: os.environ.get(key)
            for key in (
                "JOB_ID",
                "HF_JOB_ID",
                "ACCELERATOR",
                "CUDA_VISIBLE_DEVICES",
                "OMP_NUM_THREADS",
                "MKL_NUM_THREADS",
            )
            if os.environ.get(key) is not None
        },
    }


def peak_rss_mb() -> float:
    value = resource.getrusage(resource.RUSAGE_SELF).ru_maxrss
    if platform.system() == "Darwin":
        return value / (1024 * 1024)
    return value / 1024


def seed_everything(seed: int) -> None:
    import torch

    random.seed(seed)
    np.random.seed(seed)
    torch.manual_seed(seed)
    torch.set_num_threads(1)
    if hasattr(torch, "set_num_interop_threads"):
        try:
            torch.set_num_interop_threads(1)
        except RuntimeError:
            # PyTorch permits setting inter-op threads only before parallel work.
            # A previous identical setting is safe and is verified in the manifest.
            if torch.get_num_interop_threads() != 1:
                raise


@contextlib.contextmanager
def working_directory(path: Path) -> Iterator[None]:
    previous = Path.cwd()
    os.chdir(path)
    try:
        yield
    finally:
        os.chdir(previous)


@dataclass(frozen=True)
class UpstreamPaths:
    paper: Path
    companion: Path


class EvidenceWriter:
    """Append-only structured evidence for one unique collection batch."""

    def __init__(
        self,
        output_root: Path,
        batch_id: str,
        config_path: Path,
        spec_path: Path,
        lock_path: Path,
    ) -> None:
        if not batch_id or any(char not in "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789-_" for char in batch_id):
            raise ValueError("batch_id must contain only letters, digits, '-' and '_'")
        self.batch_id = batch_id
        self.batch_dir = output_root / batch_id
        self.batch_dir.mkdir(parents=True, exist_ok=False)
        self.records_path = self.batch_dir / "records.jsonl"
        self.artifacts_dir = self.batch_dir / "artifacts"
        self.artifacts_dir.mkdir()
        self.inputs_dir = self.batch_dir / "inputs"
        self.inputs_dir.mkdir()
        self.started_at = utc_now()
        self.started_monotonic = time.perf_counter()
        self.counts: Counter[str] = Counter()
        self._manifest_written = False
        for source in (config_path, spec_path, lock_path):
            destination = self.inputs_dir / source.name
            shutil.copy2(source, destination)
        self.input_hashes = {
            path.name: sha256_file(path)
            for path in sorted(self.inputs_dir.iterdir())
            if path.is_file()
        }

    def record(self, record_type: str, payload: Mapping[str, Any]) -> None:
        row = {
            "schema_version": SCHEMA_VERSION,
            "experiment_id": EXPERIMENT_ID,
            "spec_version": SPEC_VERSION,
            "paper_version": PAPER_VERSION,
            "openreview_id": OPENREVIEW_ID,
            "batch_id": self.batch_id,
            "record_type": record_type,
            "recorded_at": utc_now(),
            **payload,
        }
        with self.records_path.open("a", encoding="utf-8") as handle:
            handle.write(json.dumps(jsonable(row), sort_keys=True) + "\n")
        self.counts[record_type] += 1

    def artifact(self, relative_path: str, payload: Mapping[str, Any]) -> Path:
        path = self.artifacts_dir / relative_path
        write_json_once(path, payload)
        self.record(
            "artifact",
            {
                "status": "success",
                "artifact_path": str(path.relative_to(self.batch_dir)),
                "artifact_sha256": sha256_file(path),
                "artifact_bytes": path.stat().st_size,
            },
        )
        return path

    def finalize(
        self,
        status: str,
        config: Mapping[str, Any],
        upstream_commits: Mapping[str, str],
        error: str | None = None,
    ) -> None:
        if self._manifest_written:
            raise RuntimeError("Manifest was already finalized")
        manifest = {
            "schema_version": SCHEMA_VERSION,
            "experiment_id": EXPERIMENT_ID,
            "spec_version": SPEC_VERSION,
            "paper_version": PAPER_VERSION,
            "openreview_id": OPENREVIEW_ID,
            "batch_id": self.batch_id,
            "status": status,
            "error": error,
            "command": [sys.executable, *sys.argv],
            "runner_path": str(Path(__file__).resolve()),
            "runner_sha256": sha256_file(Path(__file__).resolve()),
            "working_directory": str(Path.cwd()),
            "started_at": self.started_at,
            "finished_at": utc_now(),
            "wall_time_seconds": time.perf_counter() - self.started_monotonic,
            "peak_rss_mb": peak_rss_mb(),
            "record_counts": dict(sorted(self.counts.items())),
            "records_sha256": sha256_file(self.records_path) if self.records_path.exists() else None,
            "input_hashes": self.input_hashes,
            "upstream_commits": dict(upstream_commits),
            "config": config,
            "packages": package_versions(),
            "hardware": hardware_snapshot(),
        }
        write_json_once(self.batch_dir / "manifest.json", manifest)
        self._manifest_written = True


def run_command(command: Sequence[str], cwd: Path | None = None) -> str:
    print("+", " ".join(command), flush=True)
    completed = subprocess.run(
        list(command),
        cwd=cwd,
        check=True,
        text=True,
        stdout=subprocess.PIPE,
        stderr=subprocess.STDOUT,
    )
    if completed.stdout:
        print(completed.stdout, end="", flush=True)
    return completed.stdout


def ensure_repository(target: Path, url: str, commit: str) -> str:
    if target.exists():
        if not (target / ".git").is_dir():
            raise RuntimeError(f"Existing upstream path is not a Git repository: {target}")
    else:
        target.parent.mkdir(parents=True, exist_ok=True)
        target.mkdir()
        run_command(["git", "init", str(target)])
        run_command(["git", "remote", "add", "origin", url], cwd=target)
        run_command(["git", "fetch", "--depth", "1", "origin", commit], cwd=target)
        run_command(["git", "checkout", "--detach", "FETCH_HEAD"], cwd=target)
    actual = run_command(["git", "rev-parse", "HEAD"], cwd=target).strip()
    if actual != commit:
        raise RuntimeError(f"Upstream commit mismatch for {target}: {actual} != {commit}")
    dirty = run_command(["git", "status", "--porcelain"], cwd=target).strip()
    if dirty:
        raise RuntimeError(f"Upstream checkout is dirty: {target}\n{dirty}")
    return actual


def prepare_upstream(lock: Mapping[str, Any], upstream_root: Path) -> tuple[UpstreamPaths, dict[str, str]]:
    repositories = lock.get("repositories")
    if not isinstance(repositories, dict) or set(repositories) != set(UPSTREAM_NAMES):
        raise ValueError(f"Lock must contain exactly repositories {UPSTREAM_NAMES}")
    commits: dict[str, str] = {}
    paths: dict[str, Path] = {}
    for name in UPSTREAM_NAMES:
        entry = repositories[name]
        target = upstream_root / name
        commits[name] = ensure_repository(target, entry["url"], entry["commit"])
        paths[name] = target
    upstream = UpstreamPaths(paper=paths["paper"], companion=paths["companion"])
    for asset_name, asset in lock.get("assets", {}).items():
        repository_path = paths[asset["repository"]]
        asset_path = repository_path / asset["path"]
        if not asset_path.is_file():
            raise FileNotFoundError(f"Missing locked asset {asset_name}: {asset_path}")
        actual_hash = sha256_file(asset_path)
        if actual_hash != asset["sha256"]:
            raise RuntimeError(
                f"Asset hash mismatch for {asset_name}: {actual_hash} != {asset['sha256']}"
            )
    return upstream, commits


def activate_upstream(upstream: UpstreamPaths) -> None:
    companion_src = str(upstream.companion / "src")
    paper_src = str(upstream.paper)
    for path in (companion_src, paper_src):
        if path not in sys.path:
            sys.path.insert(0, path)


def load_json_object(path: Path) -> dict[str, Any]:
    value = json.loads(path.read_text(encoding="utf-8"))
    if not isinstance(value, dict):
        raise ValueError(f"Expected a JSON object: {path}")
    return value


def validate_config(config: Mapping[str, Any]) -> None:
    if config.get("schema_version") != SCHEMA_VERSION:
        raise ValueError(f"Unsupported config schema: {config.get('schema_version')!r}")
    if config.get("experiment_id") != EXPERIMENT_ID:
        raise ValueError(f"Unexpected experiment id: {config.get('experiment_id')!r}")
    max_workers = config.get("max_workers")
    if not isinstance(max_workers, int) or max_workers < 1 or max_workers > 8:
        raise ValueError("max_workers must be an integer in [1, 8]")
    claims = config.get("claims")
    if not isinstance(claims, dict) or not claims:
        raise ValueError("Config must enable at least one claim")
    unknown = set(claims) - {"claim1", "claim2", "claim3", "claim4", "claim5"}
    if unknown:
        raise ValueError(f"Unknown claim configs: {sorted(unknown)}")


def finite_float(value: Any, label: str) -> float:
    result = float(np.asarray(value).reshape(-1)[0])
    if not math.isfinite(result):
        raise RuntimeError(f"Non-finite {label}: {result!r}")
    return result


def mountaincar_starts(count: int) -> np.ndarray:
    if not isinstance(count, int) or count < 2:
        raise ValueError("Mountain Car evaluation requires at least two starts")
    return np.linspace(MOUNTAINCAR_START_LOW, MOUNTAINCAR_START_HIGH, count)


def run_analytic_episode(start_position: float) -> dict[str, Any]:
    import gymnasium as gym

    env = gym.make("MountainCarContinuous-v0")
    observation, _ = env.reset(options={"low": start_position, "high": start_position})
    reward_sum = 0.0
    phase = 1
    trajectory: list[dict[str, float | int]] = []
    for step in range(1000):
        position = float(observation[0])
        velocity = float(observation[1])
        if phase == 1 and abs(position - (-1.2)) <= 1e-3:
            phase = 2
        coefficient = ANALYTIC_C1 if phase == 1 else ANALYTIC_C2
        action = coefficient * abs(velocity)
        if abs(position - (-math.pi / 6)) < 1e-2:
            action = max(0.1, action)
        if velocity <= -0.0:
            action = -action
        trajectory.append(
            {
                "step": step,
                "position": position,
                "velocity": velocity,
                "action": float(action),
                "phase": phase,
            }
        )
        observation, reward, terminated, truncated, _ = env.step([action])
        reward_sum += float(reward)
        if terminated or truncated:
            env.close()
            return {
                "return": reward_sum,
                "episode_length": step + 1,
                "terminal_velocity": float(observation[1]),
                "terminated": bool(terminated),
                "truncated": bool(truncated),
                "trajectory": trajectory,
            }
    env.close()
    raise RuntimeError("Analytic episode exceeded the Gymnasium horizon")


def run_claim1(config: Mapping[str, Any], writer: EvidenceWriter) -> dict[str, Any]:
    count = int(config["perturbation_count"])
    if count < 1:
        raise ValueError("Claim 1 perturbation_count must be positive")
    rng = np.random.default_rng(2401)
    velocity = np.linspace(0.01, 0.07, 256)
    delta_xi = 1.0 / velocity.size
    required_action_integral = 0.5
    coefficient = required_action_integral / (float(np.sum(velocity)) * delta_xi)
    optimum = coefficient * velocity
    optimum_loss = float(np.sum((optimum**2) / velocity) * delta_xi)
    gaps: list[float] = []
    constraint_residuals: list[float] = []
    for _ in range(count):
        perturbation = rng.normal(0.0, 0.02, size=velocity.size)
        perturbation -= np.mean(perturbation)
        candidate = optimum + perturbation
        residual = float(np.sum(candidate - optimum) * delta_xi)
        loss = float(np.sum((candidate**2) / velocity) * delta_xi)
        gaps.append(loss - optimum_loss)
        constraint_residuals.append(residual)
    kkt_ratio = optimum / velocity
    result = {
        "status": "success",
        "claim_id": "claim1",
        "task_id": "cauchy_schwarz_discrete_certificate",
        "claim_scope": "continuous_unconstrained_theorem_mechanism",
        "experimental_unit_count": count,
        "velocity_grid_count": velocity.size,
        "derived_coefficient": coefficient,
        "optimum_loss": optimum_loss,
        "minimum_perturbed_loss_gap": min(gaps),
        "median_perturbed_loss_gap": float(np.median(gaps)),
        "maximum_constraint_residual": max(abs(value) for value in constraint_residuals),
        "maximum_kkt_ratio_deviation": float(np.max(np.abs(kkt_ratio - coefficient))),
        "historical_priority_tested": False,
        "discrete_global_optimality_tested": False,
    }
    if result["minimum_perturbed_loss_gap"] < -1e-10:
        raise RuntimeError("Claim 1 certificate found a lower feasible perturbation")
    writer.record("proof_certificate", result)
    return result


def evaluate_rl_zoo_mountaincar_episode(
    model: Any,
    env: Any,
    start_position: float,
    keep_trajectory: bool,
) -> dict[str, Any]:
    from stable_baselines3.common.vec_env import DummyVecEnv, VecNormalize

    if isinstance(env, VecNormalize):
        normalized_rewards = bool(env.norm_reward)
        normalized_observations = bool(env.norm_obs)
    elif isinstance(env, DummyVecEnv):
        # PPO and SAC released assets have no VecNormalize statistics. This is
        # an explicit official-asset type, not an attribute-based fallback.
        normalized_rewards = False
        normalized_observations = False
    else:
        raise RuntimeError(
            f"Unsupported official baseline VecEnv type: {type(env).__module__}.{type(env).__name__}"
        )
    env.set_options({"low": start_position, "high": start_position})
    env.seed(seed=0)
    observation = env.reset()
    if normalized_rewards:
        raise RuntimeError("Baseline environment unexpectedly normalizes rewards")
    reward_sum = 0.0
    trajectory: list[dict[str, float | int]] = []
    for step in range(1000):
        action, _ = model.predict(observation, deterministic=True)
        next_observation, reward, done, infos = env.step(action)
        reward_sum += finite_float(reward[0], "baseline reward")
        if keep_trajectory:
            physical = (
                env.get_original_obs()[0]
                if normalized_observations
                else next_observation[0]
            )
            trajectory.append(
                {
                    "step": step,
                    "position": float(physical[0]),
                    "velocity": float(physical[1]),
                    "action": finite_float(action, "baseline action"),
                }
            )
        observation = next_observation
        if bool(done[0]):
            terminal = infos[0].get("terminal_observation")
            if terminal is None:
                raise RuntimeError("Missing terminal observation from baseline environment")
            if normalized_observations:
                terminal = terminal * np.sqrt(env.obs_rms.var + env.epsilon) + env.obs_rms.mean
            return {
                "return": reward_sum,
                "episode_length": step + 1,
                "terminal_velocity": float(terminal[1]),
                "trajectory": trajectory,
            }
    raise RuntimeError("Baseline Mountain Car episode exceeded the horizon")


def parameter_count(model: Any) -> int:
    count = sum(parameter.numel() for _, parameter in model.policy.named_parameters())
    if count <= 0:
        raise RuntimeError("Executable model exposes no trainable parameters")
    return int(count)


def run_claim2(
    config: Mapping[str, Any],
    writer: EvidenceWriter,
    upstream: UpstreamPaths,
    scratch_root: Path,
) -> dict[str, Any]:
    activate_upstream(upstream)
    from utils import parallel

    starts = mountaincar_starts(int(config["start_count"]))
    representative_index = int(np.argmin(np.abs(starts - (-0.55))))
    method_summaries: dict[str, dict[str, float]] = {}
    analytic_returns: list[float] = []
    for index, start in enumerate(starts):
        result = run_analytic_episode(float(start))
        analytic_returns.append(result["return"])
        writer.record(
            "episode",
            {
                "status": "success",
                "claim_id": "claim2",
                "task_id": f"mountaincar_analytic_start_{index:03d}",
                "attempt_id": 1,
                "method": "analytic_two_phase",
                "environment": "MountainCarContinuous-v0",
                "start_position": float(start),
                "return": result["return"],
                "episode_length": result["episode_length"],
                "terminal_velocity": result["terminal_velocity"],
                "panel_id": "mountaincar_primary_v1",
            },
        )
        if index == representative_index:
            writer.artifact(
                "trajectories/analytic_representative.json",
                {
                    "method": "analytic_two_phase",
                    "start_position": float(start),
                    "trajectory": result["trajectory"],
                },
            )
    method_summaries["analytic_two_phase"] = summarize_values(analytic_returns)

    runtime_dir = scratch_root / writer.batch_id / "claim2"
    runtime_dir.mkdir(parents=True)
    with working_directory(runtime_dir):
        for algorithm in config["baselines"]:
            if algorithm not in {"ars", "ppo", "sac"}:
                raise ValueError(f"Unsupported Claim 2 baseline: {algorithm}")
            model, env = parallel.get_rl_zoo3_model_and_generate_env(
                algorithm,
                str(upstream.paper / "rl-trained-agents"),
                "MountainCarContinuous-v0",
            )
            model_count = parameter_count(model)
            writer.record(
                "parameter_audit",
                {
                    "status": "success",
                    "claim_id": "claim3",
                    "task_id": f"bundled_{algorithm}_executable_parameter_count",
                    "method": algorithm,
                    "count_source": "torch_named_parameters",
                    "parameter_count": model_count,
                },
            )
            returns: list[float] = []
            for index, start in enumerate(starts):
                result = evaluate_rl_zoo_mountaincar_episode(
                    model,
                    env,
                    float(start),
                    keep_trajectory=index == representative_index,
                )
                returns.append(result["return"])
                writer.record(
                    "episode",
                    {
                        "status": "success",
                        "claim_id": "claim2",
                        "task_id": f"mountaincar_{algorithm}_start_{index:03d}",
                        "attempt_id": 1,
                        "method": algorithm,
                        "environment": "MountainCarContinuous-v0",
                        "start_position": float(start),
                        "return": result["return"],
                        "episode_length": result["episode_length"],
                        "terminal_velocity": result["terminal_velocity"],
                        "panel_id": "mountaincar_primary_v1",
                    },
                )
                if index == representative_index:
                    writer.artifact(
                        f"trajectories/{algorithm}_representative.json",
                        {
                            "method": algorithm,
                            "start_position": float(start),
                            "trajectory": result["trajectory"],
                        },
                    )
            method_summaries[algorithm] = summarize_values(returns)
            env.close()
    expected_episode_count = len(starts) * (1 + len(config["baselines"]))
    actual_episode_count = sum(
        summary["count"] for summary in method_summaries.values()
    )
    if actual_episode_count != expected_episode_count:
        raise RuntimeError(
            f"Claim 2 coverage mismatch: {actual_episode_count} != {expected_episode_count}"
        )
    return method_summaries


def summarize_values(values: Sequence[float]) -> dict[str, float]:
    array = np.asarray(values, dtype=np.float64)
    if array.size == 0 or not np.all(np.isfinite(array)):
        raise RuntimeError("Cannot summarize empty or non-finite values")
    return {
        "count": int(array.size),
        "mean": float(np.mean(array)),
        "std": float(np.std(array)),
        "min": float(np.min(array)),
        "max": float(np.max(array)),
    }


def tensor_to_list(value: Any) -> Any:
    """Convert tensors/arrays produced by upstream code into strict JSON values."""
    try:
        import torch

        if torch.is_tensor(value):
            return value.detach().cpu().numpy().tolist()
    except ImportError:
        pass
    if isinstance(value, np.ndarray):
        return value.tolist()
    if isinstance(value, np.generic):
        return value.item()
    if isinstance(value, Mapping):
        return {str(key): tensor_to_list(item) for key, item in value.items()}
    if isinstance(value, (list, tuple)):
        return [tensor_to_list(item) for item in value]
    return value


def extract_polynomial_coefficients(model: Any, algorithm: str) -> list[Any]:
    parameters = model.policy.parameters()
    if algorithm == "ars":
        coefficients = [tensor_to_list(parameters)]
    elif algorithm == "ppo":
        coefficients = [tensor_to_list(parameter) for parameter in parameters]
    else:
        raise ValueError(f"Unsupported polynomial algorithm: {algorithm}")
    flat_count = sum(np.asarray(item).size for item in coefficients)
    if flat_count <= 0:
        raise RuntimeError("Polynomial policy produced no coefficients")
    return coefficients


def run_polynomial_episode(
    model: Any,
    env: Any,
    options: Mapping[str, float],
) -> float:
    """Evaluate one exact deterministic start using the authors' helper."""
    from utils import exp_run

    mean_reward, std_reward = exp_run.run_sb3_model(
        model,
        env,
        episodes=1,
        options=dict(options),
    )
    reward = finite_float(mean_reward, "polynomial episode reward")
    if abs(finite_float(std_reward, "polynomial episode std")) > 1e-12:
        raise RuntimeError("One deterministic episode unexpectedly has nonzero std")
    return reward


def sb3_training_worker(job: Mapping[str, Any]) -> dict[str, Any]:
    """Train and fully evaluate one polynomial SB3 policy in an isolated process."""
    captured = io.StringIO()
    started = time.perf_counter()
    try:
        with contextlib.redirect_stdout(captured), contextlib.redirect_stderr(captured):
            upstream = UpstreamPaths(
                paper=Path(job["paper_path"]),
                companion=Path(job["companion_path"]),
            )
            activate_upstream(upstream)
            seed = int(job["seed"])
            seed_everything(seed)
            from stable_baselines3.common.callbacks import EvalCallback
            from utils import exp_run

            algorithm = str(job["algorithm"])
            environment = str(job["environment"])
            tensorboard_dir = Path(job["work_dir"]) / f"{job['claim_id']}_{algorithm}_{seed}"
            tensorboard_dir.mkdir(parents=True, exist_ok=False)
            model, eval_env = exp_run.get_sb3_polynomial_model_and_eval_env(
                basis="chebyshev",
                degree=int(job["degree"]),
                algo=algorithm,
                learning_rate=float(job["learning_rate"]),
                clip_range=float(job.get("clip_range", 0.4)),
                clip_range_vf=job.get("clip_range_vf"),
                batch_size=int(job.get("batch_size", 64)),
                n_steps=int(job.get("n_steps", 1024)),
                n_epochs=int(job.get("n_epochs", 1)),
                delta_std=float(job.get("delta_std", 0.05)),
                n_delta=int(job.get("n_delta", 8)),
                n_top=job.get("n_top"),
                env_name=environment,
                tensorboard_log_dir=str(tensorboard_dir),
                seed=seed,
                verbose=0,
                normalize_actions=bool(job.get("normalize_actions", True)),
            )
            evaluation_log_dir = tensorboard_dir / "periodic_eval"
            evaluation_log_dir.mkdir()
            callback = EvalCallback(
                eval_env,
                eval_freq=int(job.get("evaluate_every_n_steps", 10_000)),
                deterministic=True,
                render=False,
                n_eval_episodes=1,
                log_path=str(evaluation_log_dir),
                verbose=0,
            )
            model.learn(
                total_timesteps=int(job["timesteps"]),
                tb_log_name=f"seed_{seed}",
                callback=callback,
            )
            actual_timesteps = int(model.num_timesteps)
            coefficients = extract_polynomial_coefficients(model, algorithm)
            episodes: list[dict[str, Any]] = []
            if environment == "MountainCarContinuous-v0":
                starts = mountaincar_starts(int(job["evaluation_start_count"]))
                for index, start in enumerate(starts):
                    episodes.append(
                        {
                            "evaluation_index": index,
                            "start_position": float(start),
                            "return": run_polynomial_episode(
                                model,
                                eval_env,
                                {"low": float(start), "high": float(start)},
                            ),
                        }
                    )
            elif environment == "Pendulum-v1":
                from gymnasium.envs.registration import registry

                deterministic_name = "DeterministicPendulum-v1"
                if deterministic_name not in registry:
                    exp_run.register_deterministic_pendulum_env(deterministic_name)
                # Evaluation requires a separately wrapped deterministic environment.
                _, deterministic_eval_env = exp_run.get_sb3_polynomial_model_and_eval_env(
                    basis="chebyshev",
                    degree=int(job["degree"]),
                    algo=algorithm,
                    coeffs=coefficients[0] if algorithm == "ars" else coefficients,
                    env_name=deterministic_name,
                    normalize_actions=bool(job.get("normalize_actions", True)),
                    verbose=0,
                )
                grid_size = int(job["grid_points_per_dimension"])
                angles = np.linspace(-math.pi, math.pi, grid_size)
                angular_velocities = np.linspace(-1.0, 1.0, grid_size)
                index = 0
                # Match np.meshgrid(...).ravel(): angle varies fastest.
                for angular_velocity in angular_velocities:
                    for angle in angles:
                        episodes.append(
                            {
                                "evaluation_index": index,
                                "start_angle": float(angle),
                                "start_angular_velocity": float(angular_velocity),
                                "return": run_polynomial_episode(
                                    model,
                                    deterministic_eval_env,
                                    {
                                        "x_init": float(angle),
                                        "y_init": float(angular_velocity),
                                    },
                                ),
                            }
                        )
                        index += 1
                deterministic_eval_env.close()
            else:
                raise ValueError(f"Unsupported training environment: {environment}")
            evaluation_returns = [float(row["return"]) for row in episodes]
            training_curve = []
            for timesteps, results in zip(
                callback.evaluations_timesteps,
                callback.evaluations_results,
                strict=True,
            ):
                training_curve.append(
                    {
                        "timesteps": int(timesteps),
                        "mean_return": float(np.mean(results)),
                    }
                )
            eval_env.close()
            training_env = model.get_env()
            if training_env is not None:
                training_env.close()
        return {
            "status": "success",
            "claim_id": str(job["claim_id"]),
            "method": str(job["method"]),
            "seed": int(job["seed"]),
            "configured_timesteps": int(job["timesteps"]),
            "actual_timesteps": actual_timesteps,
            "configuration": {
                key: tensor_to_list(value)
                for key, value in job.items()
                if key not in {"paper_path", "companion_path", "work_dir"}
            },
            "coefficients": coefficients,
            "coefficient_count": int(
                sum(np.asarray(item, dtype=object).size for item in coefficients)
            ),
            "episodes": episodes,
            "evaluation_summary": summarize_values(evaluation_returns),
            "training_curve": training_curve,
            "wall_time_seconds": time.perf_counter() - started,
            "peak_rss_mb": peak_rss_mb(),
            "stdout": captured.getvalue(),
        }
    except Exception:
        return {
            "status": "error",
            "claim_id": str(job.get("claim_id")),
            "method": str(job.get("method")),
            "seed": int(job.get("seed", -1)),
            "wall_time_seconds": time.perf_counter() - started,
            "peak_rss_mb": peak_rss_mb(),
            "error": traceback.format_exc(),
            "stdout": captured.getvalue(),
        }


def evaluate_reinforce_episode(
    mrp: Any,
    *,
    options: Mapping[str, float] | None = None,
    seed: int | None = None,
) -> dict[str, Any]:
    if seed is not None:
        observation, _ = mrp.env.reset(seed=seed, options=dict(options or {}))
    else:
        observation, _ = mrp.reset(options=dict(options or {}))
    reward_sum = 0.0
    for step in range(1000):
        observation, reward, terminated, truncated, _, action, _ = mrp.step(
            observation,
            sigma=1e-12,
        )
        reward_sum += finite_float(reward, "REINFORCE reward")
        finite_float(action, "REINFORCE action")
        if terminated or truncated:
            return {"return": reward_sum, "episode_length": step + 1}
    raise RuntimeError("REINFORCE evaluation exceeded the environment horizon")


def reinforce_training_worker(job: Mapping[str, Any]) -> dict[str, Any]:
    """Fresh fixed-seed AdamW replication of one REINFORCE candidate."""
    captured = io.StringIO()
    started = time.perf_counter()
    try:
        with contextlib.redirect_stdout(captured), contextlib.redirect_stderr(captured):
            upstream = UpstreamPaths(
                paper=Path(job["paper_path"]),
                companion=Path(job["companion_path"]),
            )
            activate_upstream(upstream)
            seed = int(job["seed"])
            seed_everything(seed)
            import gymnasium as gym
            from algorithms import polynomial_agents

            env = gym.make("MountainCarContinuous-v0")
            # Seed Gymnasium's private generator before upstream training performs
            # its first unseeded reset. Subsequent resets advance deterministically.
            env.reset(seed=seed)
            mrp = polynomial_agents.TrainableContinuousMRPWrapper(
                env,
                basis="chebyshev",
                degree=3,
                normalize_observations=True,
                initial_sigma=0.25,
            )
            rewards: list[Any] = []
            steps: list[Any] = []
            losses: list[Any] = []
            coefficient_history: list[Any] = []
            mrp.train(
                alpha_mu=0.0003,
                alpha_sigma=0.00003,
                epochs=int(job["reinforce_episodes"]),
                discount=0.9,
                method="reinforce_autodiff",
                learning_history=rewards,
                steps_history=steps,
                loss_history=losses,
                coeffs_history=coefficient_history,
                mu_optimizer="adamw",
                sigma_optimizer="adamw",
                verbose=False,
            )
            mu_coefficients = tensor_to_list(mrp.agent.mu_approximator.coeffs)
            sigma_coefficients = tensor_to_list(mrp.agent.sigma_approximator.coeffs)
            selection_episodes: list[dict[str, Any]] = []
            for index in range(int(job["selection_episodes"])):
                evaluation_seed = 9_100_000 + index
                result = evaluate_reinforce_episode(mrp, seed=evaluation_seed)
                selection_episodes.append(
                    {
                        "evaluation_index": index,
                        "evaluation_seed": evaluation_seed,
                        **result,
                    }
                )
            env.close()
            training_rewards = [finite_float(value, "REINFORCE training reward") for value in rewards]
            training_steps = [int(finite_float(value, "REINFORCE training steps")) for value in steps]
            if len(training_rewards) != int(job["reinforce_episodes"]):
                raise RuntimeError(
                    "REINFORCE training history length does not match configured episodes"
                )
            selection_returns = [row["return"] for row in selection_episodes]
        return {
            "status": "success",
            "claim_id": "claim4",
            "method": "ch3_reinforce_adamw_seeded",
            "seed": seed,
            "configured_episodes": int(job["reinforce_episodes"]),
            "actual_training_episodes": len(training_rewards),
            "training_rewards": training_rewards,
            "training_steps": training_steps,
            "training_losses": tensor_to_list(losses),
            "mu_coefficients": mu_coefficients,
            "sigma_coefficients": sigma_coefficients,
            "coefficient_count": int(np.asarray(mu_coefficients).size),
            "selection_episodes": selection_episodes,
            "selection_summary": summarize_values(selection_returns),
            "configuration": {
                "degree": 3,
                "alpha_mu": 0.0003,
                "alpha_sigma": 0.00003,
                "discount": 0.9,
                "initial_sigma": 0.25,
                "mu_optimizer": "adamw",
                "sigma_optimizer": "adamw",
                "evaluation_sigma": 1e-12,
                "selection_seed_schedule": "9100000 + episode_index",
            },
            "wall_time_seconds": time.perf_counter() - started,
            "peak_rss_mb": peak_rss_mb(),
            "stdout": captured.getvalue(),
        }
    except Exception:
        return {
            "status": "error",
            "claim_id": "claim4",
            "method": "ch3_reinforce_adamw_seeded",
            "seed": int(job.get("seed", -1)),
            "wall_time_seconds": time.perf_counter() - started,
            "peak_rss_mb": peak_rss_mb(),
            "error": traceback.format_exc(),
            "stdout": captured.getvalue(),
        }


def run_worker_jobs(
    worker: Any,
    jobs: Sequence[Mapping[str, Any]],
    max_workers: int,
) -> list[dict[str, Any]]:
    if not jobs:
        raise ValueError("At least one worker job is required")
    results: list[dict[str, Any]] = []
    context = multiprocessing.get_context("spawn")
    with ProcessPoolExecutor(max_workers=max_workers, mp_context=context) as pool:
        futures = {pool.submit(worker, dict(job)): int(job["seed"]) for job in jobs}
        for future in as_completed(futures):
            seed = futures[future]
            try:
                result = future.result()
            except BaseException:
                result = {
                    "status": "error",
                    "claim_id": str(jobs[0].get("claim_id")),
                    "method": str(jobs[0].get("method")),
                    "seed": seed,
                    "error": traceback.format_exc(),
                    "stdout": "",
                }
            results.append(result)
            print(
                f"worker method={result.get('method')} seed={seed} "
                f"status={result.get('status')}",
                flush=True,
            )
    return sorted(results, key=lambda result: int(result["seed"]))


def persist_worker_result(
    writer: EvidenceWriter,
    result: Mapping[str, Any],
    panel_id: str,
) -> None:
    method = str(result["method"])
    seed = int(result["seed"])
    safe_method = method.replace("/", "_")
    writer.artifact(
        f"logs/{safe_method}_seed_{seed}.json",
        {
            "status": result["status"],
            "method": method,
            "seed": seed,
            "stdout": result.get("stdout", ""),
            "error": result.get("error"),
        },
    )
    training_payload = {
        key: value
        for key, value in result.items()
        if key
        not in {
            "episodes",
            "selection_episodes",
            "training_rewards",
            "training_steps",
            "training_losses",
            "coefficients",
            "mu_coefficients",
            "sigma_coefficients",
            "training_curve",
            "stdout",
        }
    }
    training_payload["panel_id"] = panel_id
    writer.record("training_run", training_payload)
    if result["status"] != "success":
        return
    coefficient_payload: dict[str, Any]
    if "coefficients" in result:
        coefficient_payload = {"coefficients": result["coefficients"]}
    else:
        coefficient_payload = {
            "mu_coefficients": result["mu_coefficients"],
            "sigma_coefficients": result["sigma_coefficients"],
        }
    writer.artifact(
        f"checkpoints/{safe_method}_seed_{seed}.json",
        {
            "status": "success",
            "method": method,
            "seed": seed,
            **coefficient_payload,
        },
    )
    for index, metric in enumerate(result.get("training_curve", [])):
        writer.record(
            "training_metric",
            {
                "status": "success",
                "claim_id": result["claim_id"],
                "task_id": f"{safe_method}_seed_{seed}_periodic_{index:04d}",
                "method": method,
                "seed": seed,
                "panel_id": panel_id,
                **metric,
            },
        )
    training_rewards = result.get("training_rewards", [])
    training_steps = result.get("training_steps", [])
    for index, reward in enumerate(training_rewards):
        writer.record(
            "training_metric",
            {
                "status": "success",
                "claim_id": result["claim_id"],
                "task_id": f"{safe_method}_seed_{seed}_episode_{index:04d}",
                "method": method,
                "seed": seed,
                "panel_id": panel_id,
                "training_episode": index,
                "return": reward,
                "episode_length": training_steps[index],
            },
        )


def require_all_workers_success(results: Sequence[Mapping[str, Any]], label: str) -> None:
    failed = [result for result in results if result.get("status") != "success"]
    if failed:
        compact = [
            {
                "seed": result.get("seed"),
                "error": str(result.get("error", ""))[-2000:],
            }
            for result in failed
        ]
        raise RuntimeError(f"{label} had failed workers: {json.dumps(compact)}")


def record_sb3_episodes(
    writer: EvidenceWriter,
    result: Mapping[str, Any],
    *,
    claim_id: str,
    panel_id: str,
    environment: str,
) -> None:
    method = str(result["method"])
    seed = int(result["seed"])
    for row in result["episodes"]:
        index = int(row["evaluation_index"])
        writer.record(
            "episode",
            {
                "status": "success",
                "claim_id": claim_id,
                "task_id": f"{method}_seed_{seed}_evaluation_{index:05d}",
                "attempt_id": 1,
                "method": method,
                "seed": seed,
                "environment": environment,
                "panel_id": panel_id,
                **row,
            },
        )


def select_best_result(
    results: Sequence[Mapping[str, Any]],
    summary_key: str,
) -> Mapping[str, Any]:
    if not results:
        raise ValueError("Cannot select from no results")
    return max(
        results,
        key=lambda result: (float(result[summary_key]["mean"]), -int(result["seed"])),
    )


def run_claim3(
    config: Mapping[str, Any],
    writer: EvidenceWriter,
    upstream: UpstreamPaths,
    max_workers: int,
    scratch_root: Path,
) -> dict[str, Any]:
    seeds = [int(seed) for seed in config["seeds"]]
    if len(seeds) != len(set(seeds)) or not seeds:
        raise ValueError("Claim 3 seeds must be unique and non-empty")
    jobs = [
        {
            "paper_path": str(upstream.paper),
            "companion_path": str(upstream.companion),
            "work_dir": str(scratch_root / writer.batch_id / "claim3"),
            "claim_id": "claim3",
            "method": "ch3_ars",
            "algorithm": "ars",
            "environment": "MountainCarContinuous-v0",
            "seed": seed,
            "degree": 3,
            "learning_rate": 0.018,
            "delta_std": 0.1,
            "n_delta": 4,
            "n_top": 1,
            "timesteps": int(config["timesteps"]),
            "evaluate_every_n_steps": 10_000,
            "evaluation_start_count": int(config["evaluation_start_count"]),
            "normalize_actions": True,
        }
        for seed in seeds
    ]
    (scratch_root / writer.batch_id / "claim3").mkdir(parents=True)
    results = run_worker_jobs(sb3_training_worker, jobs, min(max_workers, len(jobs)))
    for result in results:
        persist_worker_result(writer, result, "mountaincar_ch3_ars_seed_grid_v1")
    require_all_workers_success(results, "Claim 3 CH-3-ARS")
    for result in results:
        record_sb3_episodes(
            writer,
            result,
            claim_id="claim3",
            panel_id="mountaincar_ch3_ars_seed_grid_v1",
            environment="MountainCarContinuous-v0",
        )
    expected_episodes = len(seeds) * int(config["evaluation_start_count"])
    actual_episodes = sum(len(result["episodes"]) for result in results)
    if actual_episodes != expected_episodes:
        raise RuntimeError(
            f"Claim 3 coverage mismatch: {actual_episodes} != {expected_episodes}"
        )
    selected = select_best_result(results, "evaluation_summary")
    writer.record(
        "selection",
        {
            "status": "success",
            "claim_id": "claim3",
            "task_id": "select_best_ch3_ars_on_fixed_start_grid",
            "method": "ch3_ars",
            "selection_rule": "maximum mean return across the fixed Mountain Car start grid",
            "candidate_count": len(results),
            "selected_seed": int(selected["seed"]),
            "selected_summary": selected["evaluation_summary"],
            "selection_and_reporting_grid_reused": True,
        },
    )
    paper_mlp_count = 4_355
    correct_dense_count = 64 * (2 + 1) + 64 * (64 + 1) + 1 * (64 + 1)
    coefficient_count = 16
    parameter_audit = {
        "status": "success",
        "claim_id": "claim3",
        "task_id": "claim3_parameter_arithmetic_audit",
        "method": "parameter_formula_audit",
        "chebyshev_parameter_count": coefficient_count,
        "paper_stated_mlp_parameter_count": paper_mlp_count,
        "paper_stated_ratio": 277.0,
        "ratio_using_paper_count": paper_mlp_count / coefficient_count,
        "correct_dense_2x64x64x1_parameter_count": correct_dense_count,
        "ratio_using_correct_dense_count": correct_dense_count / coefficient_count,
        "released_ars_comparator_parameter_count": 65,
        "ratio_using_released_comparator": 65 / coefficient_count,
    }
    writer.record("parameter_audit", parameter_audit)
    return {
        "selected_seed": int(selected["seed"]),
        "selected_summary": selected["evaluation_summary"],
        "all_seed_means": {
            str(result["seed"]): result["evaluation_summary"]["mean"]
            for result in results
        },
        "parameter_audit": parameter_audit,
    }


def evaluate_selected_reinforce_on_fixed_grid(
    result: Mapping[str, Any],
    start_count: int,
    upstream: UpstreamPaths,
) -> list[dict[str, Any]]:
    activate_upstream(upstream)
    import gymnasium as gym
    import torch
    from algorithms import polynomial_agents

    previous_grad_state = torch.is_grad_enabled()
    try:
        env = gym.make("MountainCarContinuous-v0")
        mrp = polynomial_agents.TrainableContinuousMRPWrapper(
            env,
            basis="chebyshev",
            degree=3,
            normalize_observations=True,
            initial_sigma=0.25,
            mu_coeffs=result["mu_coefficients"],
            sigma_coeffs=result["sigma_coefficients"],
        )
        rows: list[dict[str, Any]] = []
        for index, start in enumerate(mountaincar_starts(start_count)):
            episode = evaluate_reinforce_episode(
                mrp,
                options={"low": float(start), "high": float(start)},
            )
            rows.append(
                {
                    "evaluation_index": index,
                    "start_position": float(start),
                    **episode,
                }
            )
        env.close()
    finally:
        torch.set_grad_enabled(previous_grad_state)
    return rows


def run_claim4(
    config: Mapping[str, Any],
    writer: EvidenceWriter,
    upstream: UpstreamPaths,
    max_workers: int,
    scratch_root: Path,
) -> dict[str, Any]:
    ppo_seeds = [int(seed) for seed in config["ppo_seeds"]]
    reinforce_seeds = [int(seed) for seed in config["reinforce_seeds"]]
    if not ppo_seeds or len(ppo_seeds) != len(set(ppo_seeds)):
        raise ValueError("Claim 4 PPO seeds must be unique and non-empty")
    if not reinforce_seeds or len(reinforce_seeds) != len(set(reinforce_seeds)):
        raise ValueError("Claim 4 REINFORCE seeds must be unique and non-empty")
    worker_root = scratch_root / writer.batch_id / "claim4"
    worker_root.mkdir(parents=True)
    ppo_jobs = [
        {
            "paper_path": str(upstream.paper),
            "companion_path": str(upstream.companion),
            "work_dir": str(worker_root),
            "claim_id": "claim4",
            "method": "ch3_ppo",
            "algorithm": "ppo",
            "environment": "MountainCarContinuous-v0",
            "seed": seed,
            "degree": 3,
            "learning_rate": 0.001,
            "n_steps": 2048,
            "batch_size": 1048,
            "n_epochs": 2,
            "clip_range": 0.4,
            "clip_range_vf": 0.4,
            "timesteps": int(config["ppo_timesteps"]),
            "evaluate_every_n_steps": 10_000,
            "evaluation_start_count": int(config["evaluation_start_count"]),
            "normalize_actions": True,
        }
        for seed in ppo_seeds
    ]
    ppo_results = run_worker_jobs(
        sb3_training_worker,
        ppo_jobs,
        min(max_workers, len(ppo_jobs)),
    )
    for result in ppo_results:
        persist_worker_result(writer, result, "mountaincar_ch3_ppo_seed_grid_v1")
    require_all_workers_success(ppo_results, "Claim 4 CH-3-PPO")
    for result in ppo_results:
        record_sb3_episodes(
            writer,
            result,
            claim_id="claim4",
            panel_id="mountaincar_ch3_ppo_seed_grid_v1",
            environment="MountainCarContinuous-v0",
        )
    selected_ppo = select_best_result(ppo_results, "evaluation_summary")
    writer.record(
        "selection",
        {
            "status": "success",
            "claim_id": "claim4",
            "task_id": "select_best_ch3_ppo_on_fixed_start_grid",
            "method": "ch3_ppo",
            "selection_rule": "maximum mean return across the fixed Mountain Car start grid",
            "candidate_count": len(ppo_results),
            "selected_seed": int(selected_ppo["seed"]),
            "selected_summary": selected_ppo["evaluation_summary"],
            "selection_and_reporting_grid_reused": True,
        },
    )

    reinforce_jobs = [
        {
            "paper_path": str(upstream.paper),
            "companion_path": str(upstream.companion),
            "claim_id": "claim4",
            "method": "ch3_reinforce_adamw_seeded",
            "seed": seed,
            "reinforce_episodes": int(config["reinforce_episodes"]),
            "selection_episodes": int(config["reinforce_selection_episodes"]),
        }
        for seed in reinforce_seeds
    ]
    reinforce_results = run_worker_jobs(
        reinforce_training_worker,
        reinforce_jobs,
        min(max_workers, len(reinforce_jobs)),
    )
    for result in reinforce_results:
        persist_worker_result(writer, result, "mountaincar_ch3_reinforce_seed_grid_v1")
    require_all_workers_success(reinforce_results, "Claim 4 CH-3-REINFORCE")
    for result in reinforce_results:
        seed = int(result["seed"])
        for row in result["selection_episodes"]:
            index = int(row["evaluation_index"])
            writer.record(
                "episode",
                {
                    "status": "success",
                    "claim_id": "claim4",
                    "task_id": f"ch3_reinforce_seed_{seed}_selection_{index:04d}",
                    "attempt_id": 1,
                    "method": "ch3_reinforce_adamw_seeded",
                    "seed": seed,
                    "environment": "MountainCarContinuous-v0",
                    "panel_id": "mountaincar_ch3_reinforce_selection_v1",
                    **row,
                },
            )
    selected_reinforce = select_best_result(reinforce_results, "selection_summary")
    fixed_grid_rows = evaluate_selected_reinforce_on_fixed_grid(
        selected_reinforce,
        int(config["evaluation_start_count"]),
        upstream,
    )
    for row in fixed_grid_rows:
        index = int(row["evaluation_index"])
        writer.record(
            "episode",
            {
                "status": "success",
                "claim_id": "claim4",
                "task_id": f"ch3_reinforce_selected_fixed_grid_{index:04d}",
                "attempt_id": 1,
                "method": "ch3_reinforce_adamw_seeded_selected",
                "seed": int(selected_reinforce["seed"]),
                "environment": "MountainCarContinuous-v0",
                "panel_id": "mountaincar_ch3_reinforce_primary_v1",
                **row,
            },
        )
    fixed_grid_summary = summarize_values([row["return"] for row in fixed_grid_rows])
    writer.record(
        "selection",
        {
            "status": "success",
            "claim_id": "claim4",
            "task_id": "select_best_seeded_adamw_reinforce_candidate",
            "method": "ch3_reinforce_adamw_seeded",
            "selection_rule": "maximum mean return over 50 common-random-number episodes",
            "candidate_count": len(reinforce_results),
            "selected_seed": int(selected_reinforce["seed"]),
            "selection_summary": selected_reinforce["selection_summary"],
            "fixed_grid_summary": fixed_grid_summary,
            "exact_original_replay": False,
            "determinization_extension": (
                "seeded Python, NumPy, PyTorch, and Gymnasium; common evaluation seeds"
            ),
            "reason": (
                "The paper released neither random seeds nor the selected coefficients; "
                "upstream asynchronous completion also made the stored policy index unstable."
            ),
        },
    )
    expected_ppo_episodes = len(ppo_seeds) * int(config["evaluation_start_count"])
    if sum(len(result["episodes"]) for result in ppo_results) != expected_ppo_episodes:
        raise RuntimeError("Claim 4 PPO coverage mismatch")
    expected_selection = len(reinforce_seeds) * int(config["reinforce_selection_episodes"])
    if sum(len(result["selection_episodes"]) for result in reinforce_results) != expected_selection:
        raise RuntimeError("Claim 4 REINFORCE selection coverage mismatch")
    return {
        "ppo_selected_seed": int(selected_ppo["seed"]),
        "ppo_selected_summary": selected_ppo["evaluation_summary"],
        "reinforce_selected_seed": int(selected_reinforce["seed"]),
        "reinforce_selection_summary": selected_reinforce["selection_summary"],
        "reinforce_fixed_grid_summary": fixed_grid_summary,
        "reinforce_reproduction_type": "fresh_fixed_seed_independent_replication",
    }


def evaluate_pendulum_baseline(
    algorithm: str,
    grid_size: int,
    upstream: UpstreamPaths,
    runtime_dir: Path,
) -> tuple[dict[str, list[dict[str, Any]]], int]:
    activate_upstream(upstream)
    from envs import custom_gymnasium
    from utils import parallel

    runtime_dir.mkdir(parents=True)
    with working_directory(runtime_dir):
        model, env = parallel.get_rl_zoo3_model_and_generate_env(
            algorithm,
            str(upstream.paper / "rl-trained-agents"),
            "Pendulum-v1",
        )
        model_count = parameter_count(model)
        if bool(env.norm_reward):
            raise RuntimeError("Pendulum baseline unexpectedly normalizes rewards")
        # Preserve every released observation/action wrapper while replacing only
        # the innermost stochastic Pendulum environment, exactly as the notebook.
        innermost = env.envs[0].env.env.env
        if not hasattr(innermost, "env"):
            raise RuntimeError("Unexpected Pendulum baseline wrapper structure")
        innermost.env = custom_gymnasium.DeterministicPendulumEnv()
        author_exact_rows: list[dict[str, Any]] = []
        corrected_rows: list[dict[str, Any]] = []

        def finish_episode(observation: np.ndarray) -> float:
            reward_sum = 0.0
            for _ in range(200):
                action, _ = model.predict(observation, deterministic=True)
                observation, reward, done, _ = env.step(action)
                reward_sum += finite_float(reward[0], "Pendulum baseline reward")
                if bool(done[0]):
                    return reward_sum
            raise RuntimeError("Pendulum baseline episode exceeded 200 steps")

        index = 0
        for angular_velocity in np.linspace(-1.0, 1.0, grid_size):
            for angle in np.linspace(-math.pi, math.pi, grid_size):
                options = {
                    "x_init": float(angle),
                    "y_init": float(angular_velocity),
                }
                # Author-exact protocol: the first observation bypasses the
                # released VecNormalize wrapper because the notebook resets the
                # leaf directly. Preserve it for headline-number comparison.
                env.reset()
                observation = np.asarray([innermost.env.reset(options=options)[0]])
                author_exact_rows.append(
                    {
                        "evaluation_index": index,
                        "start_angle": float(angle),
                        "start_angular_velocity": float(angular_velocity),
                        "return": finish_episode(observation),
                    }
                )
                # Corrected sensitivity analysis: reset through the full VecEnv
                # chain so the first observation is normalized like later ones.
                env.set_options(options)
                corrected_observation = env.reset()
                corrected_rows.append(
                    {
                        "evaluation_index": index,
                        "start_angle": float(angle),
                        "start_angular_velocity": float(angular_velocity),
                        "return": finish_episode(corrected_observation),
                    }
                )
                index += 1
        env.close()
    return {
        "author_exact": author_exact_rows,
        "corrected_first_observation": corrected_rows,
    }, model_count


def run_claim5(
    config: Mapping[str, Any],
    writer: EvidenceWriter,
    upstream: UpstreamPaths,
    max_workers: int,
    scratch_root: Path,
) -> dict[str, Any]:
    seeds = [int(seed) for seed in config["seeds"]]
    if not seeds or len(seeds) != len(set(seeds)):
        raise ValueError("Claim 5 seeds must be unique and non-empty")
    baselines = list(config["baselines"])
    if baselines != ["ars"]:
        raise ValueError("Claim 5 currently requires exactly the released ARS baseline")
    worker_root = scratch_root / writer.batch_id / "claim5"
    worker_root.mkdir(parents=True)
    jobs = [
        {
            "paper_path": str(upstream.paper),
            "companion_path": str(upstream.companion),
            "work_dir": str(worker_root),
            "claim_id": "claim5",
            "method": "ch6_ars_pendulum",
            "algorithm": "ars",
            "environment": "Pendulum-v1",
            "seed": seed,
            "degree": 6,
            "learning_rate": 0.018,
            "delta_std": 0.3,
            "n_delta": 8,
            "n_top": None,
            "timesteps": int(config["timesteps"]),
            "evaluate_every_n_steps": 10_000,
            "grid_points_per_dimension": int(config["grid_points_per_dimension"]),
            "normalize_actions": True,
        }
        for seed in seeds
    ]
    results = run_worker_jobs(sb3_training_worker, jobs, min(max_workers, len(jobs)))
    for result in results:
        persist_worker_result(writer, result, "pendulum_ch6_ars_seed_grid_v1")
    require_all_workers_success(results, "Claim 5 Pendulum CH-6-ARS")
    for result in results:
        record_sb3_episodes(
            writer,
            result,
            claim_id="claim5",
            panel_id="pendulum_ch6_ars_seed_grid_v1",
            environment="DeterministicPendulum-v1",
        )
    selected = select_best_result(results, "evaluation_summary")
    writer.record(
        "selection",
        {
            "status": "success",
            "claim_id": "claim5",
            "task_id": "select_best_ch6_ars_pendulum_grid",
            "method": "ch6_ars_pendulum",
            "selection_rule": "maximum mean return across the 2D deterministic evaluation grid",
            "candidate_count": len(results),
            "selected_seed": int(selected["seed"]),
            "selected_summary": selected["evaluation_summary"],
            "selection_and_reporting_grid_reused": True,
        },
    )
    grid_size = int(config["grid_points_per_dimension"])
    baseline_protocols, baseline_parameter_count = evaluate_pendulum_baseline(
        "ars",
        grid_size,
        upstream,
        scratch_root / writer.batch_id / "claim5_baseline",
    )
    baseline_rows = baseline_protocols["author_exact"]
    corrected_baseline_rows = baseline_protocols["corrected_first_observation"]
    for row in baseline_rows:
        index = int(row["evaluation_index"])
        writer.record(
            "episode",
            {
                "status": "success",
                "claim_id": "claim5",
                "task_id": f"pendulum_ars_baseline_grid_{index:05d}",
                "attempt_id": 1,
                "method": "ars_baseline_pendulum_released",
                "seed": 0,
                "environment": "DeterministicPendulum-v1",
                "panel_id": "pendulum_primary_comparison_v1",
                "evaluation_protocol": "author_exact_first_observation_bypasses_wrappers",
                **row,
            },
        )
    for row in corrected_baseline_rows:
        index = int(row["evaluation_index"])
        writer.record(
            "episode",
            {
                "status": "success",
                "claim_id": "claim5",
                "task_id": f"pendulum_ars_baseline_corrected_grid_{index:05d}",
                "attempt_id": 1,
                "method": "ars_baseline_pendulum_released_corrected_reset",
                "seed": 0,
                "environment": "DeterministicPendulum-v1",
                "panel_id": "pendulum_baseline_reset_sensitivity_v1",
                "evaluation_protocol": "corrected_first_observation_through_vecnormalize",
                **row,
            },
        )
    # Duplicate the selected learned-policy rows into the comparison panel so
    # downstream paired heatmaps never infer the selected seed themselves.
    for row in selected["episodes"]:
        index = int(row["evaluation_index"])
        writer.record(
            "episode",
            {
                "status": "success",
                "claim_id": "claim5",
                "task_id": f"pendulum_ch6_ars_selected_grid_{index:05d}",
                "attempt_id": 1,
                "method": "ch6_ars_pendulum_selected",
                "seed": int(selected["seed"]),
                "environment": "DeterministicPendulum-v1",
                "panel_id": "pendulum_primary_comparison_v1",
                "evaluation_protocol": "author_exact_polynomial_helper",
                **row,
            },
        )
    writer.record(
        "parameter_audit",
        {
            "status": "success",
            "claim_id": "claim5",
            "task_id": "pendulum_released_ars_parameter_count",
            "method": "ars_baseline_pendulum_released",
            "count_source": "torch_named_parameters",
            "parameter_count": baseline_parameter_count,
        },
    )
    writer.record(
        "scope_limitation",
        {
            "status": "not_executed",
            "claim_id": "claim5",
            "task_id": "aero2_physical_hardware",
            "method": "physical_quanser_aero2",
            "reason": (
                "Requires a physical Quanser Aero 2 and proprietary HIL runtime; "
                "the artifact also omits the selected Chebyshev checkpoint, real-hardware "
                "evaluation driver, and raw hardware traces. Simulation is not treated as "
                "equivalent evidence for sim-to-real transfer."
            ),
        },
    )
    expected_grid = grid_size * grid_size
    if any(len(result["episodes"]) != expected_grid for result in results):
        raise RuntimeError("Claim 5 learned-policy grid coverage mismatch")
    if len(baseline_rows) != expected_grid:
        raise RuntimeError("Claim 5 baseline grid coverage mismatch")
    if len(corrected_baseline_rows) != expected_grid:
        raise RuntimeError("Claim 5 corrected baseline grid coverage mismatch")
    return {
        "pendulum_selected_seed": int(selected["seed"]),
        "pendulum_selected_summary": selected["evaluation_summary"],
        "pendulum_baseline_summary": summarize_values(
            [row["return"] for row in baseline_rows]
        ),
        "pendulum_baseline_corrected_reset_summary": summarize_values(
            [row["return"] for row in corrected_baseline_rows]
        ),
        "aero2_status": "inconclusive_physical_hardware_unavailable",
    }


def parse_args() -> argparse.Namespace:
    script_root = Path(__file__).resolve().parents[1]
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--config", type=Path, required=True)
    parser.add_argument("--batch-id", required=True)
    parser.add_argument(
        "--output-root",
        type=Path,
        default=script_root / "runs" / "raw",
    )
    parser.add_argument(
        "--lock-path",
        type=Path,
        default=script_root / "upstream.lock.json",
    )
    parser.add_argument(
        "--spec-path",
        type=Path,
        default=script_root / "experiments" / "icml2026_chebyshev" / "SPEC.md",
    )
    parser.add_argument(
        "--upstream-root",
        type=Path,
        default=script_root / ".cache" / "upstream",
    )
    parser.add_argument(
        "--scratch-root",
        type=Path,
        default=Path("/tmp/chebyshev-reproduction-scratch"),
        help="Local ephemeral training logs; must not be inside output-root.",
    )
    return parser.parse_args()


def main() -> int:
    args = parse_args()
    config_path = args.config.resolve()
    lock_path = args.lock_path.resolve()
    spec_path = args.spec_path.resolve()
    output_root = args.output_root.resolve()
    upstream_root = args.upstream_root.resolve()
    scratch_root = args.scratch_root.resolve()
    for path in (config_path, lock_path, spec_path):
        if not path.is_file():
            raise FileNotFoundError(path)
    config = load_json_object(config_path)
    validate_config(config)
    lock = load_json_object(lock_path)
    prospective_batch_dir = output_root / args.batch_id
    if (
        scratch_root == prospective_batch_dir
        or prospective_batch_dir in scratch_root.parents
        or scratch_root in prospective_batch_dir.parents
    ):
        raise ValueError("scratch-root and the append-only evidence batch must be disjoint")
    writer = EvidenceWriter(
        output_root=output_root,
        batch_id=args.batch_id,
        config_path=config_path,
        spec_path=spec_path,
        lock_path=lock_path,
    )
    commits: dict[str, str] = {}
    summaries: dict[str, Any] = {}
    try:
        upstream, commits = prepare_upstream(lock, upstream_root)
        activate_upstream(upstream)
        writer.record(
            "environment",
            {
                "status": "success",
                "task_id": "runtime_environment_snapshot",
                "packages": package_versions(),
                "hardware": hardware_snapshot(),
                "upstream_commits": commits,
                "scratch_root": str(scratch_root),
            },
        )
        for claim_id in ("claim1", "claim2", "claim3", "claim4", "claim5"):
            claim_config = config["claims"].get(claim_id)
            if not claim_config or not bool(claim_config.get("enabled")):
                continue
            print(f"starting {claim_id}", flush=True)
            if claim_id == "claim1":
                summaries[claim_id] = run_claim1(claim_config, writer)
            elif claim_id == "claim2":
                summaries[claim_id] = run_claim2(
                    claim_config,
                    writer,
                    upstream,
                    scratch_root,
                )
            elif claim_id == "claim3":
                summaries[claim_id] = run_claim3(
                    claim_config,
                    writer,
                    upstream,
                    int(config["max_workers"]),
                    scratch_root,
                )
            elif claim_id == "claim4":
                summaries[claim_id] = run_claim4(
                    claim_config,
                    writer,
                    upstream,
                    int(config["max_workers"]),
                    scratch_root,
                )
            elif claim_id == "claim5":
                summaries[claim_id] = run_claim5(
                    claim_config,
                    writer,
                    upstream,
                    int(config["max_workers"]),
                    scratch_root,
                )
            writer.record(
                "claim_summary",
                {
                    "status": "success",
                    "claim_id": claim_id,
                    "summary": summaries[claim_id],
                },
            )
        if not summaries:
            raise RuntimeError("No enabled claims were executed")
        writer.finalize("success", config, commits)
        print(
            "REPRO_RESULT="
            + json.dumps(
                {
                    "status": "success",
                    "batch_id": args.batch_id,
                    "batch_dir": str(writer.batch_dir),
                    "summaries": summaries,
                },
                sort_keys=True,
            ),
            flush=True,
        )
        return 0
    except BaseException:
        error = traceback.format_exc()
        try:
            writer.record(
                "failure",
                {
                    "status": "error",
                    "task_id": "batch_failure",
                    "error": error,
                },
            )
            writer.finalize("error", config, commits, error=error)
        except BaseException:
            print("Failed while finalizing failure evidence:", file=sys.stderr)
            traceback.print_exc()
        print(error, file=sys.stderr, flush=True)
        return 1


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