rpki/scripts/rpctl/rpctl

#!/usr/bin/env python3
from __future__ import annotations

import argparse
import csv
import gzip
import hashlib
import json
import os
import platform
import re
import shutil
import signal
import subprocess
import sys
import time
from pathlib import Path
from typing import Any

RIR_TAL = {
    "afrinic": "afrinic.tal",
    "apnic": "apnic-rfc7730-https.tal",
    "arin": "arin.tal",
    "lacnic": "lacnic.tal",
    "ripe": "ripe-ncc.tal",
}

TAL_URLS = {
    "afrinic": "https://rpki.afrinic.net/tal/afrinic.tal",
    "apnic": "https://tal.apnic.net/apnic.tal",
    "arin": "https://www.arin.net/resources/manage/rpki/arin.tal",
    "lacnic": "https://www.lacnic.net/innovaportal/file/4983/1/lacnic.tal",
    "ripe": "https://tal.rpki.ripe.net/ripe-ncc.tal",
}

PROFILES = {
    "ours-no-cache": "ours-rp",
    "ours-pp-object-cache": "ours-rp",
    "routinator-latest-release": "routinator",
    "rpki-client-latest-release": "rpki-client",
    "fort-latest-release": "fort",
    "rpki-prover-incremental": "rpki-prover",
    "rpki-prover-full": "rpki-prover",
}

ALIASES = {
    "ours": "ours-pp-object-cache",
    "ours-rp": "ours-pp-object-cache",
    "ours-cached": "ours-pp-object-cache",
    "ours-baseline": "ours-no-cache",
    "ours-no-cache": "ours-no-cache",
    "ours-pp-object-cache": "ours-pp-object-cache",
    "routinator": "routinator-latest-release",
    "routinator-latest-release": "routinator-latest-release",
    "rpki-client": "rpki-client-latest-release",
    "rpki-client-latest-release": "rpki-client-latest-release",
    "fort": "fort-latest-release",
    "fort-latest-release": "fort-latest-release",
    "rpki-prover": "rpki-prover-incremental",
    "rpki-prover-incremental": "rpki-prover-incremental",
    "rpki-prover-full": "rpki-prover-full",
    "rpki-prover-no-incremental": "rpki-prover-full",
}

RPKI_PROVER_IMAGE_TAG = "rpki-prover:6857d4bf"
RPKI_PROVER_VERSION = "rpki-prover-0.10.1"

DEFAULT_CONFIG = {
    "root": "/root/rpki_4rp_ctl",
    "bundle": "",
}


def utc_now() -> str:
    return time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())


def utc_id() -> str:
    return time.strftime("%Y%m%dT%H%M%SZ", time.gmtime())


def script_root() -> Path:
    return Path(__file__).resolve().parent


def ctl_root() -> Path:
    return script_root()


def state_dir() -> Path:
    return ctl_root() / "state"


def logs_dir() -> Path:
    return ctl_root() / "logs"


def runs_dir() -> Path:
    return ctl_root() / "runs"


def config_path() -> Path:
    return ctl_root() / "config.json"


def current_path() -> Path:
    return state_dir() / "current.json"


def pid_path() -> Path:
    return state_dir() / "runner.pid"


def pgid_path() -> Path:
    return state_dir() / "runner.pgid"


def ensure_dirs() -> None:
    for path in [state_dir(), logs_dir(), runs_dir()]:
        path.mkdir(parents=True, exist_ok=True)


def read_json(path: Path, default: Any) -> Any:
    if not path.exists():
        return default
    try:
        return json.loads(path.read_text(encoding="utf-8"))
    except Exception:
        return default


def normalize_experiment_metadata(payload: Any) -> dict[str, Any]:
    if not isinstance(payload, dict):
        return {}
    normalized = dict(payload)
    if "expId" not in normalized and normalized.get("runId"):
        normalized["expId"] = normalized["runId"]
    if "experimentRoot" not in normalized and normalized.get("runRoot"):
        normalized["experimentRoot"] = normalized["runRoot"]
    normalized.pop("runId", None)
    normalized.pop("runRoot", None)
    return normalized


def read_current_state() -> dict[str, Any]:
    return normalize_experiment_metadata(read_json(current_path(), {}))


def write_json(path: Path, payload: Any) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    tmp = path.with_suffix(path.suffix + ".tmp")
    tmp.write_text(json.dumps(payload, ensure_ascii=False, indent=2, sort_keys=True) + "\n", encoding="utf-8")
    tmp.replace(path)


def load_config() -> dict[str, Any]:
    cfg = dict(DEFAULT_CONFIG)
    cfg.update(read_json(config_path(), {}))
    return cfg


def save_config(cfg: dict[str, Any]) -> None:
    write_json(config_path(), cfg)


def pid_alive(pid: int) -> bool:
    if pid <= 0:
        return False
    try:
        os.kill(pid, 0)
    except ProcessLookupError:
        return False
    except PermissionError:
        return True
    return True


def read_pid(path: Path) -> int:
    try:
        return int(path.read_text(encoding="utf-8").strip())
    except Exception:
        return 0


def normalize_profile(value: str) -> str:
    key = value.strip()
    if key not in ALIASES:
        raise SystemExit(f"unknown rp/profile: {value}; use one of: {', '.join(sorted(ALIASES))}")
    return ALIASES[key]


def parse_rirs(value: str) -> list[str]:
    if value in ("all5", "all"):
        return ["afrinic", "apnic", "arin", "lacnic", "ripe"]
    rirs = [part.strip().lower() for part in value.split(",") if part.strip()]
    bad = [rir for rir in rirs if rir not in RIR_TAL]
    if bad:
        raise SystemExit(f"unknown RIR(s): {', '.join(bad)}")
    if not rirs:
        raise SystemExit("empty --rirs")
    return rirs


def parse_interval(value: str) -> int:
    text = value.strip().lower()
    if text in ("0", "0s", "now"):
        return 0
    match = re.fullmatch(r"(\d+)([smh]?)", text)
    if not match:
        raise SystemExit(f"invalid interval: {value}")
    n = int(match.group(1))
    unit = match.group(2) or "s"
    return n * {"s": 1, "m": 60, "h": 3600}[unit]


def command_string(argv: list[str]) -> str:
    import shlex
    return " ".join(shlex.quote(x) for x in argv)


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 check_bundle(root: Path) -> None:
    required = ["bin/rpki", "bin/routinator", "bin/rpki-client", "bin/fort", "fixtures/tal", "scripts/normalize_and_collect.py", "scripts/write_group_report.py", "versions.json"]
    missing = [item for item in required if not (root / item).exists()]
    if missing:
        raise SystemExit(f"runtime bundle incomplete at {root}: missing {missing}")


def install_from_bundle(bundle: Path) -> None:
    check_bundle(bundle)
    for name in ["bin", "lib", "fixtures", "scripts"]:
        src = bundle / name
        dst = ctl_root() / name
        if dst.exists():
            if dst.is_dir():
                shutil.rmtree(dst)
            else:
                dst.unlink()
        shutil.copytree(src, dst)
    shutil.copy2(bundle / "versions.json", ctl_root() / "versions.json")
    for path in [ctl_root() / "bin", ctl_root() / "scripts"]:
        for item in path.iterdir():
            if item.is_file():
                item.chmod(item.stat().st_mode | 0o111)


def validate_rpki_prover_host() -> None:
    arch = platform.machine().lower()
    if arch not in {"x86_64", "amd64"}:
        raise SystemExit(f"rpki-prover image requires amd64 host, found: {arch}")
    if shutil.which("docker") is None:
        raise SystemExit("docker is required for rpki-prover profiles")
    result = subprocess.run(
        ["docker", "info"],
        stdout=subprocess.DEVNULL,
        stderr=subprocess.PIPE,
        text=True,
        check=False,
    )
    if result.returncode != 0:
        raise SystemExit(f"docker is not available: {result.stderr.strip()}")


def inspect_rpki_prover_image() -> dict[str, str]:
    inspect = subprocess.run(
        ["docker", "image", "inspect", RPKI_PROVER_IMAGE_TAG],
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
        text=True,
        check=False,
    )
    if inspect.returncode != 0:
        raise SystemExit(f"loaded image is missing tag {RPKI_PROVER_IMAGE_TAG}: {inspect.stderr.strip()}")
    data = json.loads(inspect.stdout)[0]
    architecture = str(data.get("Architecture") or "")
    if architecture != "amd64":
        raise SystemExit(f"rpki-prover image architecture mismatch: expected amd64, found {architecture or '<unknown>'}")
    version = subprocess.run(
        ["docker", "run", "--rm", "--entrypoint", "/opt/app/rpki-prover", RPKI_PROVER_IMAGE_TAG, "--version"],
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
        text=True,
        check=False,
    )
    version_text = version.stdout.strip()
    if version.returncode != 0 or version_text != RPKI_PROVER_VERSION:
        raise SystemExit(
            f"rpki-prover version check failed: expected {RPKI_PROVER_VERSION}, "
            f"exit={version.returncode} output={version_text or version.stderr.strip()}"
        )
    return {
        "tag": RPKI_PROVER_IMAGE_TAG,
        "imageId": str(data.get("Id") or ""),
        "architecture": architecture,
        "version": version_text,
    }


def install_rpki_prover_image(archive: Path) -> dict[str, str]:
    if not archive.is_file():
        raise SystemExit(f"rpki-prover image archive does not exist: {archive}")
    validate_rpki_prover_host()
    image_dir = ctl_root() / "images"
    image_dir.mkdir(parents=True, exist_ok=True)
    destination = image_dir / archive.name
    if archive != destination:
        shutil.copy2(archive, destination)
    load = subprocess.run(
        ["docker", "load", "--input", str(destination)],
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
        text=True,
        check=False,
    )
    if load.returncode != 0:
        raise SystemExit(f"failed to load rpki-prover image archive: {load.stderr.strip()}")
    metadata = inspect_rpki_prover_image()
    metadata.update({
        "archive": str(destination),
        "archiveSha256": sha256_file(destination),
        "loadedAtUtc": utc_now(),
    })
    return metadata


def require_rpki_prover_image() -> dict[str, Any]:
    metadata = load_config().get("rpkiProverImage")
    if not isinstance(metadata, dict) or metadata.get("tag") != RPKI_PROVER_IMAGE_TAG:
        raise SystemExit(
            "rpki-prover image is not initialized; run ./rpctl init --bundle <bundle> "
            "--rpki-prover-image <docker.tar.gz>"
        )
    validate_rpki_prover_host()
    inspected = inspect_rpki_prover_image()
    if metadata.get("imageId") and metadata.get("imageId") != inspected.get("imageId"):
        raise SystemExit(
            f"rpki-prover image id changed: configured={metadata.get('imageId')} "
            f"loaded={inspected.get('imageId')}"
        )
    return metadata


def cmd_init(args: argparse.Namespace) -> None:
    ensure_dirs()
    bundle = Path(args.bundle).resolve()
    install_from_bundle(bundle)
    cfg = load_config()
    cfg["bundle"] = str(bundle)
    cfg["initializedAtUtc"] = utc_now()
    if args.rpki_prover_image:
        cfg["rpkiProverImage"] = install_rpki_prover_image(Path(args.rpki_prover_image).resolve())
    save_config(cfg)
    print(f"initialized root={ctl_root()}")
    print(f"bundle={bundle}")
    if cfg.get("rpkiProverImage"):
        image = cfg["rpkiProverImage"]
        print(f"rpki_prover_image={image.get('tag')}")
        print(f"rpki_prover_image_id={image.get('imageId')}")
        print(f"rpki_prover_archive_sha256={image.get('archiveSha256')}")


def timed_run(run_dir: Path, argv: list[str], env: dict[str, str] | None = None) -> int:
    run_dir.mkdir(parents=True, exist_ok=True)
    (run_dir / "start-utc.txt").write_text(utc_now() + "\n", encoding="utf-8")
    stdout = (run_dir / "stdout.log").open("wb")
    stderr = (run_dir / "stderr.log").open("wb")
    time_path = run_dir / "process-time.txt"
    full = ["/usr/bin/time", "-v", "-o", str(time_path), "--", *argv]
    try:
        proc = subprocess.run(full, stdout=stdout, stderr=stderr, env=env, check=False)
        code = proc.returncode
    finally:
        stdout.close()
        stderr.close()
    (run_dir / "end-utc.txt").write_text(utc_now() + "\n", encoding="utf-8")
    (run_dir / "exit-code.txt").write_text(str(code) + "\n", encoding="utf-8")
    return code


def clean_state_for_snapshot(profile_root: Path) -> None:
    state = profile_root / "state"
    if state.exists():
        shutil.rmtree(state)
    state.mkdir(parents=True, exist_ok=True)


def prepare_state(profile_root: Path, snapshot: bool) -> Path:
    state = profile_root / "state"
    if snapshot:
        clean_state_for_snapshot(profile_root)
    else:
        state.mkdir(parents=True, exist_ok=True)
    return state


def build_ours_args(profile: str, run_root: Path, run_dir: Path, state: Path, rirs: list[str]) -> list[str]:
    argv = [
        str(ctl_root() / "bin" / "rpki"),
        "--db", str(state / "work-db"),
        "--repo-bytes-db", str(state / "repo-bytes.db"),
        "--rsync-mirror-root", str(state / "rsync-mirror"),
        "--rsync-scope", "module-root",
        "--report-json", str(run_dir / "report.json"),
        "--report-json-compact",
        "--ccr-out", str(run_dir / "result.ccr"),
        "--cir-enable",
        "--cir-out", str(run_dir / "input.cir"),
        "--vrps-csv-out", str(run_dir / "vrps.csv"),
        "--vaps-csv-out", str(run_dir / "vaps.csv"),
        "--compare-view-trust-anchor", "all5" if len(rirs) > 1 else rirs[0],
        "--parallel-phase2-ready-batch-size", "256",
        "--parallel-phase2-ready-batch-wall-time-budget-ms", "100",
        "--parallel-phase2-result-drain-batch-size", "2048",
        "--parallel-phase2-finalize-batch-size", "256",
        "--parallel-phase2-finalize-batch-wall-time-budget-ms", "100",
    ]
    for rir in rirs:
        argv += ["--tal-url", TAL_URLS[rir]]
    for rir in rirs:
        argv += ["--cir-tal-uri", TAL_URLS[rir]]
    if profile == "ours-pp-object-cache":
        argv += [
            "--enable-publication-point-validation-cache",
            "--enable-roa-validation-cache",
            "--enable-child-certificate-validation-cache",
        ]
    return argv


def build_routinator_args(run_dir: Path, state: Path, rirs: list[str], snapshot: bool) -> list[str]:
    extra_tals = state / "extra-tals"
    repo = state / "repository"
    extra_tals.mkdir(parents=True, exist_ok=True)
    repo.mkdir(parents=True, exist_ok=True)
    for rir in rirs:
        shutil.copy2(ctl_root() / "fixtures" / "tal" / RIR_TAL[rir], extra_tals / RIR_TAL[rir])
    argv = [
        str(ctl_root() / "bin" / "routinator"),
        "-r", str(repo),
        "--no-rir-tals",
        "--extra-tals-dir", str(extra_tals),
        "--enable-aspa",
    ]
    if snapshot:
        argv.append("--fresh")
    argv += ["vrps", "-f", "jsonext", "-o", str(run_dir / "routinator.json")]
    return argv


def ensure_rpki_client_user() -> None:
    subprocess.run(["id", "-u", "_rpki-client"], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, check=False)
    if subprocess.run(["id", "-u", "_rpki-client"], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, check=False).returncode != 0:
        subprocess.run(["useradd", "-r", "-M", "-s", "/usr/sbin/nologin", "_rpki-client"], check=False)


def build_rpki_client_args(run_dir: Path, state: Path, rirs: list[str]) -> tuple[list[str], dict[str, str]]:
    ensure_rpki_client_user()
    cache = state / "cache"
    cache.mkdir(parents=True, exist_ok=True)
    run_dir.mkdir(parents=True, exist_ok=True)
    skiplist = state / "skiplist"
    skiplist.touch(exist_ok=True)
    for path in [state, cache, run_dir]:
        path.chmod(0o755)
    subprocess.run(["chown", "-R", "_rpki-client:_rpki-client", str(state), str(run_dir)], check=False)
    subprocess.run(["chmod", "-R", "u+rwX,go+rX", str(state), str(run_dir)], check=False)
    tal_args: list[str] = []
    for rir in rirs:
        tal_args += ["-t", str(ctl_root() / "fixtures" / "tal" / RIR_TAL[rir])]
    argv = [
        str(ctl_root() / "bin" / "rpki-client"),
        "-p", "4",
        "-j",
        "-c",
        "-S", str(skiplist),
        "-d", str(cache),
        *tal_args,
        str(run_dir),
    ]
    env = os.environ.copy()
    env["LD_LIBRARY_PATH"] = str(ctl_root() / "lib") + ":" + env.get("LD_LIBRARY_PATH", "")
    return argv, env


def build_fort_args(run_dir: Path, state: Path, rirs: list[str]) -> tuple[list[str], dict[str, str]]:
    repo = state / "repository"
    tal_dir = state / "tal"
    repo.mkdir(parents=True, exist_ok=True)
    tal_dir.mkdir(parents=True, exist_ok=True)
    for old_tal in tal_dir.glob("*.tal"):
        old_tal.unlink()
    for rir in rirs:
        shutil.copy2(ctl_root() / "fixtures" / "tal" / RIR_TAL[rir], tal_dir / RIR_TAL[rir])
    argv = [
        str(ctl_root() / "bin" / "fort"),
        "--mode", "standalone",
        "--tal", str(tal_dir),
        "--local-repository", str(repo),
        "--output.roa", str(run_dir / "vrps.csv"),
        "--output.format", "csv",
        "--log.output", "console",
        "--log.level", "warning",
        "--thread-pool.validation.max", "5",
        "--http.connect-timeout", "30",
        "--http.transfer-timeout", "180",
        "--rsync.strategy", "root-except-ta",
    ]
    env = os.environ.copy()
    env["MALLOC_ARENA_MAX"] = "2"
    env["LD_LIBRARY_PATH"] = str(ctl_root() / "lib") + ":" + env.get("LD_LIBRARY_PATH", "")
    return argv, env


def prepare_rpki_prover_state(state: Path, rirs: list[str]) -> None:
    tal_dir = state / "tals"
    tal_dir.mkdir(parents=True, exist_ok=True)
    for old_tal in tal_dir.glob("*.tal"):
        old_tal.unlink()
    for rir in rirs:
        shutil.copy2(ctl_root() / "fixtures" / "tal" / RIR_TAL[rir], tal_dir / RIR_TAL[rir])


def rpki_prover_arguments(profile: str) -> list[str]:
    argv = [
        "--once",
        "--vrp-output", "/outputs/vrps.csv",
        "--no-rir-tals",
        "--rpki-root-directory", "/rpki-data",
        "--cpu-count", "4",
        "--fetcher-count", "8",
        "--log-level", "info",
    ]
    if profile == "rpki-prover-full":
        argv.append("--no-incremental-validation")
    return argv


def build_rpki_prover_command(profile: str, run_dir: Path, state: Path, container_name: str) -> list[str]:
    return [
        "docker", "run", "--name", container_name,
        "-v", f"{state.resolve()}:/rpki-data",
        "-v", f"{run_dir.resolve()}:/outputs",
        RPKI_PROVER_IMAGE_TAG,
        *rpki_prover_arguments(profile),
    ]


def container_rss_kb(container_name: str) -> tuple[int, int]:
    top = subprocess.run(
        ["docker", "top", container_name, "-eo", "pid"],
        stdout=subprocess.PIPE,
        stderr=subprocess.DEVNULL,
        text=True,
        check=False,
    )
    total = 0
    maximum = 0
    for line in top.stdout.splitlines():
        pid = line.strip()
        if not pid.isdigit():
            continue
        try:
            status = Path(f"/proc/{pid}/status").read_text(encoding="utf-8", errors="replace")
        except OSError:
            continue
        match = re.search(r"^VmRSS:\s+(\d+)\s+kB$", status, re.MULTILINE)
        if not match:
            continue
        rss = int(match.group(1))
        total += rss
        maximum = max(maximum, rss)
    return total, maximum


def timed_rpki_prover_run(
    profile: str,
    run_dir: Path,
    state: Path,
    container_name: str,
) -> tuple[int, int, int, int, dict[str, Any]]:
    run_dir.mkdir(parents=True, exist_ok=True)
    start_utc = utc_now()
    start_ns = time.monotonic_ns()
    (run_dir / "start-utc.txt").write_text(start_utc + "\n", encoding="utf-8")
    create_argv = [
        "docker", "create", "--name", container_name,
        "-v", f"{state.resolve()}:/rpki-data",
        "-v", f"{run_dir.resolve()}:/outputs",
        RPKI_PROVER_IMAGE_TAG,
        *rpki_prover_arguments(profile),
    ]
    create = subprocess.run(create_argv, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, check=False)
    if create.returncode != 0:
        (run_dir / "stderr.log").write_text(create.stderr, encoding="utf-8")
        (run_dir / "stdout.log").write_text(create.stdout, encoding="utf-8")
        (run_dir / "end-utc.txt").write_text(utc_now() + "\n", encoding="utf-8")
        (run_dir / "exit-code.txt").write_text(str(create.returncode) + "\n", encoding="utf-8")
        return create.returncode, 0, 0, (time.monotonic_ns() - start_ns) // 1_000_000, {}

    previous_sigterm = signal.getsignal(signal.SIGTERM)

    def terminate_container(signum: int, _frame: Any) -> None:
        subprocess.run(
            ["docker", "rm", "--force", container_name],
            stdout=subprocess.DEVNULL,
            stderr=subprocess.DEVNULL,
            check=False,
        )
        raise SystemExit(128 + signum)

    signal.signal(signal.SIGTERM, terminate_container)

    max_total_rss_kb = 0
    max_process_rss_kb = 0
    inspect_data: dict[str, Any] = {}
    samples_path = run_dir / "rss-samples.tsv"
    with (run_dir / "stdout.log").open("wb") as stdout, (run_dir / "stderr.log").open("wb") as stderr, samples_path.open("w", encoding="utf-8") as samples:
        samples.write("epoch_ms\ttotal_rss_kb\tmax_process_rss_kb\n")
        start = subprocess.Popen(["docker", "start", "--attach", container_name], stdout=stdout, stderr=stderr)
        while start.poll() is None:
            total_rss_kb, process_rss_kb = container_rss_kb(container_name)
            max_total_rss_kb = max(max_total_rss_kb, total_rss_kb)
            max_process_rss_kb = max(max_process_rss_kb, process_rss_kb)
            samples.write(f"{int(time.time() * 1000)}\t{total_rss_kb}\t{process_rss_kb}\n")
            samples.flush()
            time.sleep(0.25)
        start.wait()

    try:
        inspect = subprocess.run(
            ["docker", "inspect", container_name],
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
            text=True,
            check=False,
        )
        if inspect.returncode == 0:
            parsed = json.loads(inspect.stdout)
            if parsed:
                inspect_data = parsed[0]
                write_json(run_dir / "container-inspect.json", inspect_data)
        state_data = inspect_data.get("State", {}) if inspect_data else {}
        exit_code = int(state_data.get("ExitCode", start.returncode if start.returncode is not None else 1))
    finally:
        subprocess.run(["docker", "rm", "--force", container_name], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, check=False)
        signal.signal(signal.SIGTERM, previous_sigterm)
    wall_ms = (time.monotonic_ns() - start_ns) // 1_000_000
    (run_dir / "end-utc.txt").write_text(utc_now() + "\n", encoding="utf-8")
    (run_dir / "exit-code.txt").write_text(str(exit_code) + "\n", encoding="utf-8")
    return exit_code, max_total_rss_kb, max_process_rss_kb, wall_ms, inspect_data


def collect_rpki_prover_vrps(path: Path) -> tuple[int, set[str]]:
    unique: set[str] = set()
    raw = 0
    if not path.is_file():
        return raw, unique
    with path.open("r", encoding="utf-8", errors="replace", newline="") as handle:
        for row in csv.reader(handle):
            if not row or row[0].strip().lower() == "asn":
                continue
            if len(row) < 3:
                continue
            raw += 1
            unique.add(",".join(value.strip() for value in row[:3]))
    return raw, unique


def write_rpki_prover_run_metadata(
    run_root: Path,
    profile: str,
    run_seq: int,
    mode: str,
    wall_ms: int,
    max_rss_kb: int,
    max_process_rss_kb: int,
    exit_code: int,
) -> dict[str, Any]:
    run_dir = run_root / "profiles" / profile / "runs" / f"run_{run_seq:04d}"
    raw_vrps, unique_vrps = collect_rpki_prover_vrps(run_dir / "vrps.csv")
    with gzip.open(run_dir / "normalized-vrps.txt.gz", "wt", encoding="utf-8") as handle:
        for row in sorted(unique_vrps):
            handle.write(row + "\n")
    image = load_config().get("rpkiProverImage", {})
    meta = {
        "profile": profile,
        "rp": "rpki-prover",
        "runSeq": run_seq,
        "syncMode": mode,
        "startUtc": read_text_strip(run_dir / "start-utc.txt"),
        "endUtc": read_text_strip(run_dir / "end-utc.txt"),
        "exitCode": exit_code,
        "timing": {
            "wallMs": wall_ms,
            "maxRssKb": max_rss_kb,
            "maxProcessRssKb": max_process_rss_kb,
        },
        "counts": {
            "vrps": raw_vrps,
            "vrpsRaw": raw_vrps,
            "vrpsUnique": len(unique_vrps),
            "vaps": "n/a",
        },
        "vapsAvailable": False,
        "validationAlgorithm": "full-every-iteration" if profile == "rpki-prover-full" else "incremental",
        "image": image,
        "remoteRunDir": str(run_dir),
    }
    write_json(run_dir / "run-meta.json", meta)
    return meta


def write_rpki_prover_group_report(run_root: Path, profile: str) -> None:
    rows: list[dict[str, Any]] = []
    run_dirs = sorted((run_root / "profiles" / profile / "runs").glob("run_*"))
    for run_dir in run_dirs:
        meta = read_json(run_dir / "run-meta.json", {})
        if not meta:
            continue
        rows.append({
            "profile": profile,
            "rp": "rpki-prover",
            "run_seq": meta.get("runSeq", ""),
            "sync_mode": meta.get("syncMode", ""),
            "start_utc": meta.get("startUtc", ""),
            "end_utc": meta.get("endUtc", ""),
            "exit_code": meta.get("exitCode", ""),
            "wall_ms": meta.get("timing", {}).get("wallMs", ""),
            "max_rss_kb": meta.get("timing", {}).get("maxRssKb", ""),
            "max_process_rss_kb": meta.get("timing", {}).get("maxProcessRssKb", ""),
            "vrps": meta.get("counts", {}).get("vrpsRaw", ""),
            "vrps_unique": meta.get("counts", {}).get("vrpsUnique", ""),
            "vaps": "n/a",
            "vaps_available": False,
            "validation_algorithm": meta.get("validationAlgorithm", ""),
            "remote_run_dir": str(run_dir),
        })
    reports = run_root / "reports"
    reports.mkdir(parents=True, exist_ok=True)
    if rows:
        with (reports / "per_run_metrics.csv").open("w", encoding="utf-8", newline="") as handle:
            writer = csv.DictWriter(handle, fieldnames=list(rows[0]))
            writer.writeheader()
            writer.writerows(rows)
    write_json(reports / "per_run_metrics.json", {"generatedAtUtc": utc_now(), "rows": rows})


def normalize_one(run_root: Path, profile: str, run_seq: int) -> None:
    script = ctl_root() / "scripts" / "normalize_and_collect.py"
    if not script.exists():
        return
    subprocess.run(["python3", str(script), "--root", str(run_root), "--profile", profile, "--run-seq", str(run_seq)], check=False)


def write_group_report(run_root: Path, interval_secs: int, runs: int) -> None:
    script = ctl_root() / "scripts" / "write_group_report.py"
    if not script.exists():
        return
    delta_count = max(0, runs - 1)
    interval_minutes = max(0, interval_secs // 60)
    subprocess.run([
        "python3", str(script),
        "--root", str(run_root),
        "--interval-minutes", str(interval_minutes),
        "--delta-count", str(delta_count),
        "--mode", "manual-rpctl",
    ], check=False)


def parse_latest_meta(run_root: Path, profile: str, run_seq: int) -> dict[str, Any]:
    meta = run_root / "profiles" / profile / "runs" / f"run_{run_seq:04d}" / "run-meta.json"
    return read_json(meta, {})


def update_current(**fields: Any) -> None:
    cur = read_current_state()
    cur.update(fields)
    cur["updatedAtUtc"] = utc_now()
    write_json(current_path(), cur)


def is_zero_exit(value: Any) -> bool:
    return str(value) in {"0", "0.0"}


def completed_run_count(profile: str, run_root: Path) -> int:
    profile_runs = run_root / "profiles" / profile / "runs"
    if not profile_runs.exists():
        return 0
    count = 0
    for expected_seq, run_dir in enumerate(sorted(profile_runs.glob("run_*")), start=1):
        try:
            seq = int(run_dir.name.replace("run_", ""))
        except ValueError:
            raise RuntimeError(f"invalid run directory name: {run_dir}")
        if seq != expected_seq:
            raise RuntimeError(f"non-contiguous run directories under {profile_runs}: expected run_{expected_seq:04d}, got {run_dir.name}")
        exit_code = read_text_strip(run_dir / "exit-code.txt")
        if not is_zero_exit(exit_code):
            raise RuntimeError(f"run is not successful: {run_dir} exit_code={exit_code or '<missing>'}")
        count = seq
    return count


def parse_entry_rirs(entry: dict[str, Any]) -> list[str]:
    value = entry.get("rirs") or ""
    if isinstance(value, list):
        return parse_rirs(",".join(str(item) for item in value))
    return parse_rirs(str(value))


def write_experiment_versions(run_root: Path) -> None:
    versions = read_json(ctl_root() / "versions.json", {})
    image = load_config().get("rpkiProverImage")
    if image:
        versions["rpki-prover"] = image
        profiles = versions.setdefault("profiles", [])
        for profile in ["rpki-prover-incremental", "rpki-prover-full"]:
            if profile not in profiles:
                profiles.append(profile)
    write_json(run_root / "versions.json", versions)


def run_profile(profile: str, exp_id: str, runs: int, interval_secs: int, rirs: list[str], append: bool = False) -> int:
    ensure_dirs()
    rp = PROFILES[profile]
    run_root = runs_dir() / profile / exp_id
    profile_root = run_root / "profiles" / profile
    (run_root / "logs").mkdir(parents=True, exist_ok=True)
    (run_root / "reports").mkdir(parents=True, exist_ok=True)
    write_experiment_versions(run_root)
    now = utc_now()
    if append:
        run_info = normalize_experiment_metadata(read_json(run_root / "run-info.json", {}))
        if not run_info:
            raise RuntimeError(f"cannot continue experiment without run-info.json: {run_root}")
        if run_info.get("state") != "success" or not is_zero_exit(run_info.get("exitCode")):
            raise RuntimeError(f"can only continue a successful experiment: exp_id={exp_id} state={run_info.get('state')} exitCode={run_info.get('exitCode')}")
        existing_runs = completed_run_count(profile, run_root)
        if existing_runs < 1:
            raise RuntimeError(f"cannot continue experiment without successful child runs: {run_root}")
        first_run_seq = existing_runs + 1
        last_run_seq = existing_runs + runs
        run_info.update({
            "runsRequested": last_run_seq,
            "intervalSecs": interval_secs,
            "state": "running",
            "exitCode": "",
            "lastContinueAtUtc": now,
            "lastContinueFromRunSeq": first_run_seq,
            "lastContinueRunsRequested": runs,
        })
    else:
        first_run_seq = 1
        last_run_seq = runs
        run_info = {
            "expId": exp_id,
            "profile": profile,
            "rp": rp,
            "rirs": rirs,
            "runsRequested": runs,
            "intervalSecs": interval_secs,
            "cacheProfile": cache_profile_name(profile),
            "startedAtUtc": now,
            "experimentRoot": str(run_root),
            "stateRoot": str(profile_root / "state"),
        }
        if rp == "rpki-prover":
            run_info["validationAlgorithm"] = cache_profile_name(profile)
            run_info["rpkiProverImage"] = load_config().get("rpkiProverImage", {})
    write_json(run_root / "run-info.json", run_info)
    status = {
        "state": "running",
        "rp": rp,
        "profile": profile,
        "pid": os.getpid(),
        "pgid": os.getpgrp(),
        "expId": exp_id,
        "experimentRoot": str(run_root),
        "intervalSecs": interval_secs,
        "runsRequested": last_run_seq,
        "continueMode": append,
        "firstRunSeq": first_run_seq,
        "lastRunSeqRequested": last_run_seq,
        "rirs": rirs,
        "startedAtUtc": now,
        "currentRun": None,
    }
    write_json(current_path(), status)
    exit_code = 0
    last_start = 0.0
    completed_seq = first_run_seq - 1
    for run_seq in range(first_run_seq, last_run_seq + 1):
        if run_seq > first_run_seq and interval_secs > 0:
            target = last_start + interval_secs
            sleep_seconds = max(0.0, target - time.time())
            update_current(state="waiting", nextRunSeq=run_seq, nextRunAtEpoch=target, sleepSeconds=round(sleep_seconds, 3))
            if sleep_seconds > 0:
                time.sleep(sleep_seconds)
        last_start = time.time()
        snapshot = run_seq == 1
        mode = "snapshot" if snapshot else ("warm" if rp == "rpki-prover" else "delta")
        run_dir = profile_root / "runs" / f"run_{run_seq:04d}"
        if run_dir.exists() and any(run_dir.iterdir()):
            raise RuntimeError(f"refusing to overwrite existing run directory: {run_dir}")
        state = prepare_state(profile_root, snapshot)
        update_current(state="running", currentRun=run_seq, currentMode=mode, currentRunDir=str(run_dir), lastRunStartedAtUtc=utc_now())
        env = None
        if rp == "ours-rp":
            argv = build_ours_args(profile, run_root, run_dir, state, rirs)
        elif rp == "routinator":
            argv = build_routinator_args(run_dir, state, rirs, snapshot)
        elif rp == "rpki-client":
            argv, env = build_rpki_client_args(run_dir, state, rirs)
        elif rp == "fort":
            argv, env = build_fort_args(run_dir, state, rirs)
        elif rp == "rpki-prover":
            prepare_rpki_prover_state(state, rirs)
            container_suffix = hashlib.sha1(f"{run_root}:{run_seq}".encode("utf-8")).hexdigest()[:12]
            container_name = f"rpctl-prover-{container_suffix}"
            argv = build_rpki_prover_command(profile, run_dir, state, container_name)
        else:
            raise RuntimeError(rp)
        run_dir.mkdir(parents=True, exist_ok=True)
        (run_dir / "command.txt").write_text(command_string(argv) + "\n", encoding="utf-8")
        if rp == "rpki-prover":
            code, max_rss_kb, max_process_rss_kb, wall_ms, _ = timed_rpki_prover_run(
                profile, run_dir, state, container_name
            )
            meta = write_rpki_prover_run_metadata(
                run_root,
                profile,
                run_seq,
                mode,
                wall_ms,
                max_rss_kb,
                max_process_rss_kb,
                code,
            )
            write_rpki_prover_group_report(run_root, profile)
        else:
            code = timed_run(run_dir, argv, env=env)
            normalize_one(run_root, profile, run_seq)
            meta = parse_latest_meta(run_root, profile, run_seq)
        update_current(
            state="running",
            lastRunSeq=run_seq,
            lastRunExitCode=code,
            lastRunWallMs=meta.get("timing", {}).get("wallMs"),
            lastRunMaxRssKb=meta.get("timing", {}).get("maxRssKb"),
            lastRunVrps=meta.get("counts", {}).get("vrps"),
            lastRunVaps=meta.get("counts", {}).get("vaps"),
            lastRunDir=str(run_dir),
        )
        completed_seq = run_seq
        if code != 0:
            exit_code = code
            break
    if rp == "rpki-prover":
        write_rpki_prover_group_report(run_root, profile)
    else:
        write_group_report(run_root, interval_secs, completed_seq)
    final_state = "success" if exit_code == 0 else "failed"
    update_current(state=final_state, finishedAtUtc=utc_now(), exitCode=exit_code)
    run_info = normalize_experiment_metadata(read_json(run_root / "run-info.json", {}))
    run_info.update({
        "finishedAtUtc": utc_now(),
        "state": final_state,
        "exitCode": exit_code,
        "completedRuns": completed_seq,
    })
    write_json(run_root / "run-info.json", run_info)
    return exit_code


def cmd_runner(args: argparse.Namespace) -> None:
    code = run_profile(args.profile, args.exp_id, args.runs, args.interval_secs, args.rirs.split(","), append=args.append)
    sys.exit(code)


def current_live_status() -> dict[str, Any]:
    raw_current = read_json(current_path(), {})
    cur = normalize_experiment_metadata(raw_current)
    migrated_legacy_fields = cur != raw_current
    pid = read_pid(pid_path()) or int(cur.get("pid") or 0)
    if not cur and not pid:
        return {}
    alive = pid_alive(pid)
    if cur.get("state") in {"starting", "running", "waiting"} and not alive:
        cur = dict(cur)
        cur["state"] = "exited"
        cur["pidAlive"] = False
        cur["observedExitAtUtc"] = utc_now()
        write_json(current_path(), cur)
    else:
        cur["pidAlive"] = alive
        if migrated_legacy_fields:
            write_json(current_path(), cur)
    if pid:
        cur["pid"] = pid
    return cur


def cmd_status(args: argparse.Namespace) -> None:
    ensure_dirs()
    cur = current_live_status()
    if args.json:
        print(json.dumps(cur, ensure_ascii=False, indent=2, sort_keys=True))
        return
    if not cur:
        print("state=idle")
        return
    keys = ["state", "profile", "rp", "pid", "pidAlive", "expId", "experimentRoot", "currentRun", "currentMode", "lastRunSeq", "lastRunExitCode", "lastRunWallMs", "lastRunMaxRssKb", "lastRunVrps", "lastRunVaps", "lastRunDir"]
    for key in keys:
        if key in cur:
            print(f"{key}={cur[key]}")


def cmd_list(args: argparse.Namespace) -> None:
    print("profile\trp\taliases")
    reverse: dict[str, list[str]] = {p: [] for p in PROFILES}
    for alias, profile in ALIASES.items():
        reverse.setdefault(profile, []).append(alias)
    for profile, rp in PROFILES.items():
        print(f"{profile}\t{rp}\t{','.join(sorted(reverse.get(profile, [])))}")


def cmd_paths(args: argparse.Namespace) -> None:
    cfg = load_config()
    print(f"root={ctl_root()}")
    print(f"config={config_path()}")
    print(f"state={state_dir()}")
    print(f"runs={runs_dir()}")
    print(f"logs={logs_dir()}")
    print(f"bundle={cfg.get('bundle','')}")
    image = cfg.get("rpkiProverImage", {})
    print(f"rpki_prover_image={image.get('tag', '')}")
    print(f"rpki_prover_image_id={image.get('imageId', '')}")
    print(f"rpki_prover_image_archive={image.get('archive', '')}")
    cur = current_live_status()
    if cur.get("experimentRoot"):
        print(f"current_experiment_root={cur['experimentRoot']}")
    if cur.get("lastRunDir"):
        print(f"last_run_dir={cur['lastRunDir']}")


def refuse_if_running() -> None:
    cur = current_live_status()
    if cur.get("state") in {"starting", "running", "waiting"} and cur.get("pidAlive"):
        raise SystemExit(f"another rpctl experiment is active: profile={cur.get('profile')} pid={cur.get('pid')} experimentRoot={cur.get('experimentRoot')}")


def cmd_start(args: argparse.Namespace) -> None:
    ensure_dirs()
    refuse_if_running()
    profile = normalize_profile(args.rp)
    rirs = parse_rirs(args.rirs)
    interval_secs = parse_interval(args.interval)
    if args.runs < 1:
        raise SystemExit("--runs must be >= 1")
    if not (ctl_root() / "versions.json").exists():
        raise SystemExit("rpctl is not initialized; run ./rpctl init --bundle <bundle-root> first")
    if PROFILES[profile] == "rpki-prover":
        require_rpki_prover_image()
    exp_id = args.exp_id or f"{utc_id()}_{profile}_{args.runs}runs_{interval_secs}s"
    runner_args = [
        sys.executable,
        str(Path(__file__).resolve()),
        "_runner",
        "--profile", profile,
        "--exp-id", exp_id,
        "--runs", str(args.runs),
        "--interval-secs", str(interval_secs),
        "--rirs", ",".join(rirs),
    ]
    nohup = logs_dir() / f"runner-{exp_id}.log"
    out = nohup.open("ab")
    proc = subprocess.Popen(runner_args, stdout=out, stderr=subprocess.STDOUT, start_new_session=True, cwd=str(ctl_root()))
    pid_path().write_text(str(proc.pid) + "\n", encoding="utf-8")
    try:
        pgid = os.getpgid(proc.pid)
    except Exception:
        pgid = proc.pid
    pgid_path().write_text(str(pgid) + "\n", encoding="utf-8")
    write_json(current_path(), {
        "state": "starting",
        "profile": profile,
        "rp": PROFILES[profile],
        "pid": proc.pid,
        "pgid": pgid,
        "expId": exp_id,
        "experimentRoot": str(runs_dir() / profile / exp_id),
        "intervalSecs": interval_secs,
        "runsRequested": args.runs,
        "rirs": rirs,
        "runnerLog": str(nohup),
        "startedAtUtc": utc_now(),
    })
    print(f"started profile={profile} pid={proc.pid} pgid={pgid} exp_id={exp_id}")
    print(f"log={nohup}")


def cmd_continue(args: argparse.Namespace) -> None:
    ensure_dirs()
    refuse_if_running()
    if args.runs < 1:
        raise SystemExit("--runs must be >= 1")
    if not (ctl_root() / "versions.json").exists():
        raise SystemExit("rpctl is not initialized; run ./rpctl init --bundle <bundle-root> first")
    entry = find_experiment_entry(args.rp, args.exp_id)
    if entry["state"] != "success":
        raise SystemExit(f"can only continue a successful experiment: exp_id={args.exp_id} state={entry['state']}")
    profile = entry["profile"]
    if PROFILES[profile] == "rpki-prover":
        require_rpki_prover_image()
    rirs = parse_entry_rirs(entry)
    interval_secs = parse_interval(args.interval)
    runner_args = [
        sys.executable,
        str(Path(__file__).resolve()),
        "_runner",
        "--profile", profile,
        "--exp-id", args.exp_id,
        "--runs", str(args.runs),
        "--interval-secs", str(interval_secs),
        "--rirs", ",".join(rirs),
        "--append",
    ]
    nohup = logs_dir() / f"runner-{args.exp_id}-continue-{utc_id()}.log"
    out = nohup.open("ab")
    proc = subprocess.Popen(runner_args, stdout=out, stderr=subprocess.STDOUT, start_new_session=True, cwd=str(ctl_root()))
    pid_path().write_text(str(proc.pid) + "\n", encoding="utf-8")
    try:
        pgid = os.getpgid(proc.pid)
    except Exception:
        pgid = proc.pid
    pgid_path().write_text(str(pgid) + "\n", encoding="utf-8")
    previous_runs = int(entry.get("runCount") or 0)
    write_json(current_path(), {
        "state": "starting",
        "profile": profile,
        "rp": PROFILES[profile],
        "pid": proc.pid,
        "pgid": pgid,
        "expId": args.exp_id,
        "experimentRoot": entry["experimentRoot"],
        "intervalSecs": interval_secs,
        "runsRequested": previous_runs + args.runs,
        "continueMode": True,
        "continueRunsRequested": args.runs,
        "previousRunCount": previous_runs,
        "rirs": rirs,
        "runnerLog": str(nohup),
        "startedAtUtc": utc_now(),
    })
    print(f"continued profile={profile} pid={proc.pid} pgid={pgid} exp_id={args.exp_id} add_runs={args.runs}")
    print(f"log={nohup}")


def cmd_stop(args: argparse.Namespace) -> None:
    ensure_dirs()
    cur = current_live_status()
    pid = int(cur.get("pid") or read_pid(pid_path()) or 0)
    pgid = int(cur.get("pgid") or read_pid(pgid_path()) or pid or 0)
    if not pid:
        print("no runner pid")
        return
    if not pid_alive(pid):
        update_current(state="exited", pidAlive=False)
        print(f"pid {pid} is not alive")
        return
    sig = signal.SIGKILL if args.force else signal.SIGTERM
    try:
        os.killpg(pgid, sig)
        print(f"sent {sig.name} to pgid={pgid}")
    except ProcessLookupError:
        try:
            os.kill(pid, sig)
            print(f"sent {sig.name} to pid={pid}")
        except ProcessLookupError:
            print(f"pid {pid} disappeared")
    deadline = time.time() + (2 if args.force else 10)
    while time.time() < deadline:
        if not pid_alive(pid):
            break
        time.sleep(0.2)
    alive = pid_alive(pid)
    update_current(state="stopped" if not alive else "stopping", pidAlive=alive, stoppedAtUtc=utc_now())
    print(f"pidAlive={alive}")


def cmd_tail(args: argparse.Namespace) -> None:
    cur = current_live_status()
    candidates = []
    runner_log = cur.get("runnerLog")
    if runner_log:
        candidates.append(Path(runner_log))
    experiment_root = cur.get("experimentRoot")
    if experiment_root:
        candidates.append(Path(experiment_root) / "logs" / "driver-progress.log")
    candidates.extend(sorted(logs_dir().glob("runner-*.log"), key=lambda p: p.stat().st_mtime if p.exists() else 0, reverse=True))
    path = next((p for p in candidates if p.is_file()), None)
    if path is None:
        print("no log found")
        return
    lines = path.read_text(encoding="utf-8", errors="replace").splitlines()
    for line in lines[-args.lines:]:
        print(line)


def load_csv_rows(path: Path) -> list[dict[str, str]]:
    if not path.exists():
        return []
    try:
        with path.open("r", encoding="utf-8", errors="replace", newline="") as handle:
            return list(csv.DictReader(handle))
    except Exception:
        return []


def first_non_empty(*values: Any) -> str:
    for value in values:
        if value is not None and str(value) != "":
            return str(value)
    return ""


def display_value(value: Any) -> str:
    text = first_non_empty(value)
    return text if text else "-"


def read_text_strip(path: Path) -> str:
    try:
        return path.read_text(encoding="utf-8", errors="replace").strip()
    except Exception:
        return ""


def parse_command_summary(command: str) -> str:
    if not command:
        return ""
    import shlex
    try:
        tokens = shlex.split(command)
    except ValueError:
        tokens = command.split()
    token_set = set(tokens)
    keys = []
    for flag in [
        "--enable-publication-point-validation-cache",
        "--enable-roa-validation-cache",
        "--enable-child-certificate-validation-cache",
        "--enable-aspa",
        "-p",
        "-j",
        "-c",
    ]:
        if flag in token_set:
            keys.append(flag)
    for idx, token in enumerate(tokens):
        if token == "--rsync-scope" and idx + 1 < len(tokens):
            keys.append(f"--rsync-scope={tokens[idx + 1]}")
        elif token.startswith("--rsync-scope="):
            keys.append(token)
    if RPKI_PROVER_IMAGE_TAG in token_set:
        algorithm = "full-every-iteration" if "--no-incremental-validation" in token_set else "incremental"
        keys.append(f"algorithm={algorithm}")
    return " ".join(keys)


def parse_process_time(path: Path) -> dict[str, int]:
    if not path.exists():
        return {}
    result: dict[str, int] = {}
    for line in path.read_text(encoding="utf-8", errors="replace").splitlines():
        if "Elapsed (wall clock) time" in line:
            raw = line.rsplit("):", 1)[-1].strip() if "):" in line else line.rsplit(":", 1)[-1].strip()
            try:
                result["wallMs"] = elapsed_to_ms(raw)
            except Exception:
                pass
        elif "Maximum resident set size" in line:
            try:
                result["maxRssKb"] = int(line.rsplit(":", 1)[1].strip() or 0)
            except Exception:
                pass
    return result


def elapsed_to_ms(raw: str) -> int:
    raw = raw.strip()
    days = 0
    if "-" in raw:
        day_text, raw = raw.split("-", 1)
        days = int(day_text)
    parts = raw.split(":")
    if len(parts) == 3:
        hours, minutes, seconds = parts
    elif len(parts) == 2:
        hours, minutes, seconds = "0", parts[0], parts[1]
    else:
        hours, minutes, seconds = "0", "0", parts[0]
    return int(round((days * 86400 + int(hours) * 3600 + int(minutes) * 60 + float(seconds)) * 1000))


def human_ms(value: Any) -> str:
    text = first_non_empty(value)
    if not text:
        return "-"
    try:
        ms = int(float(text))
    except ValueError:
        return text
    if ms >= 60_000:
        return f"{ms / 60_000:.1f}m"
    if ms >= 1000:
        return f"{ms / 1000:.1f}s"
    return f"{ms}ms"


def compact_utc(value: Any) -> str:
    text = first_non_empty(value)
    if not text:
        return "-"
    match = re.fullmatch(r"(\d{4})-(\d{2})-(\d{2})T(\d{2}:\d{2}:\d{2})Z", text)
    if match:
        return f"{match.group(2)}-{match.group(3)} {match.group(4)}"
    return text


def compact_rirs(value: Any) -> str:
    text = first_non_empty(value)
    if not text:
        return "-"
    rirs = [part.strip() for part in text.split(",") if part.strip()]
    if rirs == ["afrinic", "apnic", "arin", "lacnic", "ripe"]:
        return "all5"
    return ",".join(rirs) if rirs else text


def cache_profile_name(profile: str) -> str:
    if profile == "ours-pp-object-cache":
        return "pp+roa+crt"
    if profile == "ours-no-cache":
        return "off"
    if profile == "rpki-prover-incremental":
        return "incremental"
    if profile == "rpki-prover-full":
        return "full-every-iteration"
    return "n/a"


def format_run_params(entry: dict[str, Any]) -> str:
    parts = []
    runs_requested = first_non_empty(entry.get("runsRequested"))
    interval = first_non_empty(entry.get("intervalSecs"))
    rirs = compact_rirs(entry.get("rirs"))
    cache = first_non_empty(entry.get("cacheProfile"))
    if runs_requested or interval:
        interval_text = f"{interval}s" if interval else "?"
        parts.append(f"{runs_requested or '?'}@{interval_text}")
    if rirs != "-":
        parts.append(rirs)
    if cache:
        parts.append(cache)
    return " ".join(parts)


def profile_label(profile: str) -> str:
    labels = {
        "ours-pp-object-cache": "ours-cache",
        "ours-no-cache": "ours-off",
        "routinator-latest-release": "routinator",
        "rpki-client-latest-release": "rpki-client",
        "fort-latest-release": "fort",
        "rpki-prover-incremental": "prover-inc",
        "rpki-prover-full": "prover-full",
    }
    return labels.get(profile, profile)


def count_gzip_lines(path: Path) -> str:
    if not path.exists():
        return ""
    try:
        with gzip.open(path, "rt", encoding="utf-8", errors="replace") as handle:
            return str(sum(1 for line in handle if line.strip()))
    except Exception:
        return ""


def count_csv_data_rows(path: Path) -> str:
    if not path.exists():
        return ""
    try:
        with path.open("r", encoding="utf-8", errors="replace", newline="") as handle:
            return str(sum(1 for _ in csv.DictReader(handle)))
    except Exception:
        return ""


def count_vrp_json(path: Path) -> str:
    if not path.exists():
        return ""
    try:
        data = json.loads(path.read_text(encoding="utf-8", errors="replace"))
    except Exception:
        return ""
    if not isinstance(data, dict):
        return ""
    for key in ("roas", "routeOrigins", "valid_roas"):
        items = data.get(key)
        if isinstance(items, list):
            return str(len(items))
    return ""


def count_vap_json(path: Path) -> str:
    if not path.exists():
        return ""
    try:
        data = json.loads(path.read_text(encoding="utf-8", errors="replace"))
    except Exception:
        return ""
    if not isinstance(data, dict):
        return ""
    for key in ("aspas", "aspaAssertions", "vaps"):
        items = data.get(key)
        if isinstance(items, list):
            return str(len(items))
    return ""


def infer_output_counts(profile: str, run_dir: Path) -> tuple[str, str]:
    normalized_vrps = count_gzip_lines(run_dir / "normalized-vrps.txt.gz")
    normalized_vaps = count_gzip_lines(run_dir / "normalized-vaps.txt.gz")
    if normalized_vrps or normalized_vaps:
        return normalized_vrps, normalized_vaps
    rp = PROFILES.get(profile, profile)
    if rp == "ours-rp":
        return count_csv_data_rows(run_dir / "vrps.csv"), count_csv_data_rows(run_dir / "vaps.csv")
    if rp == "routinator":
        return count_vrp_json(run_dir / "routinator.json"), count_vap_json(run_dir / "routinator.json")
    if rp == "rpki-client":
        vrps = count_csv_data_rows(run_dir / "csv") or count_vrp_json(run_dir / "json")
        vaps = count_vap_json(run_dir / "json")
        return vrps, vaps
    if rp == "fort":
        return count_csv_data_rows(run_dir / "vrps.csv"), "0" if (run_dir / "vrps.csv").exists() else ""
    if rp == "rpki-prover":
        raw, _ = collect_rpki_prover_vrps(run_dir / "vrps.csv")
        return str(raw) if (run_dir / "vrps.csv").exists() else "", "n/a" if (run_dir / "vrps.csv").exists() else ""
    return "", ""


def discover_run_entries() -> list[dict[str, Any]]:
    ensure_dirs()
    current = current_live_status()
    entries: list[dict[str, Any]] = []
    for profile_dir in sorted([p for p in runs_dir().iterdir() if p.is_dir()]):
        profile = profile_dir.name
        for run_root in sorted([p for p in profile_dir.iterdir() if p.is_dir()]):
            exp_id = run_root.name
            run_info = normalize_experiment_metadata(read_json(run_root / "run-info.json", {}))
            reports_csv = run_root / "reports" / "per_run_metrics.csv"
            rows = load_csv_rows(reports_csv)
            profile_run_root = run_root / "profiles" / profile / "runs"
            run_dirs = sorted(profile_run_root.glob("run_*")) if profile_run_root.exists() else []
            started = first_non_empty(run_info.get("startedAtUtc"))
            ended = first_non_empty(run_info.get("finishedAtUtc"))
            exit_codes: list[str] = []
            walls: list[int] = []
            max_rss_values: list[int] = []
            vrps = ""
            vaps = ""
            run_count = 0
            failed_runs: list[str] = []
            if rows:
                run_count = len(rows)
                started = first_non_empty(rows[0].get("start_utc"), rows[0].get("startUtc"))
                ended = first_non_empty(rows[-1].get("end_utc"), rows[-1].get("endUtc"))
                for row in rows:
                    seq = first_non_empty(row.get("run_seq"), row.get("runSeq"))
                    exit_code = first_non_empty(row.get("exit_code"), row.get("exitCode"))
                    if exit_code:
                        exit_codes.append(exit_code)
                        if exit_code not in {"0", "0.0"}:
                            failed_runs.append(seq or "?")
                    try:
                        walls.append(int(float(first_non_empty(row.get("wall_ms"), row.get("wallMs")) or 0)))
                    except ValueError:
                        pass
                    try:
                        max_rss_values.append(int(float(first_non_empty(row.get("max_rss_kb"), row.get("maxRssKb")) or 0)))
                    except ValueError:
                        pass
                vrps = first_non_empty(rows[-1].get("vrps"), rows[-1].get("VRPs"))
                vaps = first_non_empty(rows[-1].get("vaps"), rows[-1].get("VAPs"))
                if run_dirs and (not vrps or not vaps):
                    inferred_vrps, inferred_vaps = infer_output_counts(profile, run_dirs[-1])
                    vrps = first_non_empty(vrps, inferred_vrps)
                    vaps = first_non_empty(vaps, inferred_vaps)
            else:
                run_count = len(run_dirs)
                for run_dir in run_dirs:
                    seq = run_dir.name.replace("run_", "")
                    started = started or read_text_strip(run_dir / "start-utc.txt")
                    end = read_text_strip(run_dir / "end-utc.txt")
                    ended = end or ended
                    exit_code = read_text_strip(run_dir / "exit-code.txt")
                    if exit_code:
                        exit_codes.append(exit_code)
                        if exit_code != "0":
                            failed_runs.append(seq)
                    meta = read_json(run_dir / "run-meta.json", {})
                    if meta:
                        try:
                            walls.append(int(meta.get("timing", {}).get("wallMs") or 0))
                        except Exception:
                            pass
                        try:
                            max_rss_values.append(int(meta.get("timing", {}).get("maxRssKb") or 0))
                        except Exception:
                            pass
                        vrps = str(meta.get("counts", {}).get("vrps", vrps))
                        vaps = str(meta.get("counts", {}).get("vaps", vaps))
                    else:
                        timing = parse_process_time(run_dir / "process-time.txt")
                        if timing.get("wallMs"):
                            walls.append(timing["wallMs"])
                        if timing.get("maxRssKb"):
                            max_rss_values.append(timing["maxRssKb"])
                    inferred_vrps, inferred_vaps = infer_output_counts(profile, run_dir)
                    vrps = first_non_empty(inferred_vrps, vrps)
                    vaps = first_non_empty(inferred_vaps, vaps)
            command = ""
            first_cmd = run_root / "profiles" / profile / "runs" / "run_0001" / "command.txt"
            command = read_text_strip(first_cmd)
            state = "unknown"
            if current.get("experimentRoot") == str(run_root) and current.get("state") in {"starting", "running", "waiting", "stopping", "stopped", "success", "failed", "exited"}:
                state = str(current.get("state"))
            elif failed_runs:
                state = "failed"
            elif exit_codes and all(code in {"0", "0.0"} for code in exit_codes):
                state = "success"
            elif run_dirs and not ended:
                state = "incomplete"
            elif run_root.exists():
                state = "partial"
            interval = first_non_empty(run_info.get("intervalSecs"))
            runs_requested = first_non_empty(run_info.get("runsRequested"))
            rirs = ",".join(run_info.get("rirs", [])) if isinstance(run_info.get("rirs"), list) else first_non_empty(run_info.get("rirs"))
            cache_profile = first_non_empty(run_info.get("cacheProfile"), cache_profile_name(profile))
            if current.get("experimentRoot") == str(run_root):
                interval = first_non_empty(current.get("intervalSecs"), interval)
                runs_requested = first_non_empty(current.get("runsRequested"), runs_requested)
                current_rirs = current.get("rirs")
                if isinstance(current_rirs, list):
                    rirs = ",".join(current_rirs)
            entries.append({
                "profile": profile,
                "rp": PROFILES.get(profile, profile),
                "expId": exp_id,
                "state": state,
                "startedAt": started,
                "endedAt": ended,
                "runCount": run_count,
                "runsRequested": runs_requested,
                "intervalSecs": interval,
                "rirs": rirs,
                "cacheProfile": cache_profile,
                "exitCodes": ",".join(exit_codes),
                "failedRuns": ",".join(failed_runs),
                "wallMsTotal": sum(walls) if walls else "",
                "wallMsMax": max(walls) if walls else "",
                "maxRssKb": max(max_rss_values) if max_rss_values else "",
                "vrps": vrps,
                "vaps": vaps,
                "experimentRoot": str(run_root),
                "commandSummary": parse_command_summary(command),
                "command": command,
            })
    entries.sort(key=lambda item: (item.get("startedAt") or "", item.get("experimentRoot") or ""), reverse=True)
    return entries


def select_show_entries(args: argparse.Namespace) -> list[dict[str, Any]]:
    entries = discover_run_entries()
    if args.rp:
        profile = normalize_profile(args.rp)
        entries = [entry for entry in entries if entry["profile"] == profile]
    if args.failed:
        entries = [entry for entry in entries if entry["state"] in {"failed", "incomplete", "partial", "exited"} or entry.get("failedRuns")]
    if args.limit and args.limit > 0:
        entries = entries[: args.limit]
    return entries


def print_table(rows: list[list[Any]], headers: list[str]) -> None:
    data = [[display_value(cell) for cell in row] for row in rows]
    widths = [len(header) for header in headers]
    for row in data:
        for idx, cell in enumerate(row):
            widths[idx] = max(widths[idx], min(len(cell), 80))
    def trim(cell: str, width: int) -> str:
        return cell if len(cell) <= width else cell[: max(0, width - 1)] + "…"
    print("  ".join(header.ljust(widths[idx]) for idx, header in enumerate(headers)))
    print("  ".join("-" * widths[idx] for idx in range(len(headers))))
    for row in data:
        print("  ".join(trim(cell, widths[idx]).ljust(widths[idx]) for idx, cell in enumerate(row)))


def cmd_show(args: argparse.Namespace) -> None:
    entries = select_show_entries(args)
    if args.json:
        print(json.dumps(entries, ensure_ascii=False, indent=2, sort_keys=True))
        return
    headers = ["profile", "exp_id", "state", "runs", "params", "start", "end", "failed", "wall", "vrps", "vaps"]
    rows = [[
        profile_label(e["profile"]),
        e["expId"],
        e["state"],
        e["runCount"],
        format_run_params(e),
        compact_utc(e["startedAt"]),
        compact_utc(e["endedAt"]),
        e["failedRuns"],
        human_ms(e["wallMsTotal"]),
        e["vrps"],
        e["vaps"],
    ] for e in entries]
    if args.paths:
        headers.append("path")
        for row, entry in zip(rows, entries):
            row.append(entry["experimentRoot"])
    if not rows:
        print("no experiments found")
        return
    print_table(rows, headers)


def find_experiment_entry(profile_or_alias: str | None, exp_id: str) -> dict[str, Any]:
    profile = normalize_profile(profile_or_alias) if profile_or_alias else None
    entries = discover_run_entries()
    matches = [entry for entry in entries if entry["expId"] == exp_id and (profile is None or entry["profile"] == profile)]
    if not matches:
        raise SystemExit(f"experiment not found: exp_id={exp_id}" + (f" profile={profile}" if profile else ""))
    if len(matches) > 1:
        profiles = ", ".join(entry["profile"] for entry in matches)
        raise SystemExit(f"exp_id is ambiguous; specify --rp. matches: {profiles}")
    return matches[0]


def is_running_entry(entry: dict[str, Any]) -> bool:
    run_root = str(Path(str(entry.get("experimentRoot"))).resolve())
    cur = current_live_status()
    if cur.get("experimentRoot"):
        try:
            current_root = Path(str(cur.get("experimentRoot"))).resolve()
            target_root = Path(run_root)
            if (
                current_root == target_root
                and cur.get("state") in {"starting", "running", "waiting", "stopping"}
                and bool(cur.get("pidAlive"))
            ):
                return True
        except Exception:
            pass
    try:
        proc = subprocess.run(
            ["ps", "-eo", "pid=,args="],
            text=True,
            stdout=subprocess.PIPE,
            stderr=subprocess.DEVNULL,
            check=False,
        )
    except Exception:
        return False
    for line in proc.stdout.splitlines():
        if run_root in line and " rpctl delete " not in line:
            return True
    return False


def safe_run_root(path: Path) -> Path:
    resolved = path.resolve()
    base = runs_dir().resolve()
    if resolved == base or base not in resolved.parents:
        raise SystemExit(f"refusing to delete path outside runs dir: {resolved}")
    if len(resolved.relative_to(base).parts) < 2:
        raise SystemExit(f"refusing to delete non-experiment root path: {resolved}")
    return resolved


def matching_runner_logs(exp_id: str) -> list[Path]:
    prefix = f"runner-{exp_id}"
    return sorted(path for path in logs_dir().glob("runner-*.log") if path.name.startswith(prefix))


def clear_current_if_points_to(run_root: Path) -> None:
    cur = current_live_status()
    if not cur.get("experimentRoot"):
        return
    try:
        if Path(str(cur.get("experimentRoot"))).resolve() != run_root.resolve():
            return
    except Exception:
        return
    for path in [current_path(), pid_path(), pgid_path()]:
        try:
            path.unlink()
        except FileNotFoundError:
            pass


def cmd_delete(args: argparse.Namespace) -> None:
    entry = find_experiment_entry(args.rp, args.exp_id)
    if is_running_entry(entry):
        raise SystemExit(f"refusing to delete active experiment: exp_id={args.exp_id}. stop it first.")
    run_root = safe_run_root(Path(entry["experimentRoot"]))
    logs = matching_runner_logs(args.exp_id)
    if args.dry_run:
        print(f"experiment_root={run_root}")
        for log in logs:
            print(f"log={log}")
        return
    if not args.yes:
        raise SystemExit(f"refusing to delete without --yes. target={run_root}")
    if run_root.exists():
        shutil.rmtree(run_root)
    deleted_logs = 0
    for log in logs:
        try:
            log.unlink()
            deleted_logs += 1
        except FileNotFoundError:
            pass
    clear_current_if_points_to(run_root)
    print(f"deleted exp_id={args.exp_id}")
    print(f"experiment_root={run_root}")
    print(f"runner_logs_deleted={deleted_logs}")


def run_rows_for_entry(entry: dict[str, Any]) -> list[dict[str, str]]:
    return load_csv_rows(Path(entry["experimentRoot"]) / "reports" / "per_run_metrics.csv")


def fallback_run_details(entry: dict[str, Any]) -> list[dict[str, Any]]:
    profile = entry["profile"]
    root = Path(entry["experimentRoot"]) / "profiles" / profile / "runs"
    details = []
    for run_dir in sorted(root.glob("run_*")):
        meta = read_json(run_dir / "run-meta.json", {})
        timing = meta.get("timing", {}) if meta else parse_process_time(run_dir / "process-time.txt")
        inferred_vrps, inferred_vaps = infer_output_counts(profile, run_dir)
        details.append({
            "run_seq": run_dir.name.replace("run_", ""),
            "sync_mode": meta.get("syncMode") or ("snapshot" if run_dir.name.endswith("0001") else "delta"),
            "start_utc": read_text_strip(run_dir / "start-utc.txt"),
            "end_utc": read_text_strip(run_dir / "end-utc.txt"),
            "exit_code": read_text_strip(run_dir / "exit-code.txt"),
            "wall_ms": timing.get("wallMs", ""),
            "max_rss_kb": timing.get("maxRssKb", ""),
            "vrps": first_non_empty(meta.get("counts", {}).get("vrps") if meta else "", inferred_vrps),
            "vrps_unique": first_non_empty(meta.get("counts", {}).get("vrpsUnique") if meta else ""),
            "vaps": first_non_empty(meta.get("counts", {}).get("vaps") if meta else "", inferred_vaps),
            "remote_run_dir": str(run_dir),
        })
    return details


def cmd_view(args: argparse.Namespace) -> None:
    entry = find_experiment_entry(args.rp, args.exp_id)
    rows = run_rows_for_entry(entry) or fallback_run_details(entry)
    payload = {"summary": entry, "runs": rows}
    if args.json:
        print(json.dumps(payload, ensure_ascii=False, indent=2, sort_keys=True))
        return
    print(f"exp_id={display_value(entry['expId'])}")
    print(f"profile={display_value(entry['profile'])}")
    print(f"rp={display_value(entry['rp'])}")
    print(f"state={display_value(entry['state'])}")
    print(f"start={display_value(entry['startedAt'])}")
    print(f"end={display_value(entry['endedAt'])}")
    print(f"run_count={display_value(entry['runCount'])}")
    print(f"runs_requested={display_value(entry['runsRequested'])}")
    print(f"interval_secs={display_value(entry['intervalSecs'])}")
    print(f"rirs={display_value(entry['rirs'])}")
    print(f"cache_profile={display_value(entry['cacheProfile'])}")
    print(f"failed_runs={display_value(entry['failedRuns'])}")
    print(f"wall_total_ms={display_value(entry['wallMsTotal'])}")
    print(f"wall_max_ms={display_value(entry['wallMsMax'])}")
    print(f"max_rss_kb={display_value(entry['maxRssKb'])}")
    print(f"vrps={display_value(entry['vrps'])}")
    print(f"vaps={display_value(entry['vaps'])}")
    print(f"experiment_root={display_value(entry['experimentRoot'])}")
    print(f"command_summary={display_value(entry['commandSummary'])}")
    if args.command and entry.get("command"):
        print("command=" + entry["command"])
    if rows:
        print("\nper-run:")
        headers = ["run", "mode", "exit", "start", "end", "wall_ms", "rss_kb", "vrps", "vrps_unique", "vaps", "path"]
        table_rows = []
        for row in rows:
            table_rows.append([
                first_non_empty(row.get("run_seq"), row.get("runSeq")),
                first_non_empty(row.get("sync_mode"), row.get("syncMode")),
                first_non_empty(row.get("exit_code"), row.get("exitCode")),
                first_non_empty(row.get("start_utc"), row.get("startUtc")),
                first_non_empty(row.get("end_utc"), row.get("endUtc")),
                first_non_empty(row.get("wall_ms"), row.get("wallMs")),
                first_non_empty(row.get("max_rss_kb"), row.get("maxRssKb")),
                first_non_empty(row.get("vrps"), row.get("VRPs")),
                first_non_empty(row.get("vrps_unique"), row.get("vrpsUnique")),
                first_non_empty(row.get("vaps"), row.get("VAPs")),
                first_non_empty(row.get("remote_run_dir"), row.get("remoteRunDir")),
            ])
        print_table(table_rows, headers)


def build_parser() -> argparse.ArgumentParser:
    formatter = argparse.RawDescriptionHelpFormatter
    parser = argparse.ArgumentParser(
        description="Control one local five-RP performance run on this server.",
        formatter_class=formatter,
        epilog="""
RP/profile naming:
  ours / ours-rp / ours-cached
      Alias of ours-pp-object-cache. Starts ours RP with PP cache + ROA cache
      + child certificate cache enabled.

  ours-no-cache / ours-baseline
      Starts ours RP without the three cache acceleration flags. This is the
      baseline/no-cache profile.

  routinator
      Alias of routinator-latest-release. Uses the archived Routinator release
      binary and enables ASPA output when supported.

  rpki-client
      Alias of rpki-client-latest-release. Uses archived official rpki-client
      release binary with the same parameters as the #100 experiment.

  fort
      Alias of fort-latest-release. Uses archived FORT validator release binary.

  rpki-prover / rpki-prover-incremental
      Uses the archived rpki-prover Docker image with its default incremental
      validation algorithm and persistent state.

  rpki-prover-full / rpki-prover-no-incremental
      Uses the same image with --no-incremental-validation. Its state is kept
      separate from the incremental profile.

Cache difference for ours RP:
  --rp ours adds:
    --enable-publication-point-validation-cache
    --enable-roa-validation-cache
    --enable-child-certificate-validation-cache

  --rp ours-no-cache does not add those flags.

Typical usage:
  ./rpctl init --bundle /data/rpki_4rp_ctl_runtime_bundle \\
    --rpki-prover-image /data/rpki-prover-6857d4bf-linux-amd64.docker.tar.gz
  ./rpctl list
  ./rpctl start --rp ours --interval 10m --runs 6 --rirs all5
  ./rpctl start --rp ours-no-cache --interval 10m --runs 6 --rirs all5
  ./rpctl start --rp rpki-prover --interval 0 --runs 2 --rirs all5
  ./rpctl status
  ./rpctl show
  ./rpctl show --failed
  ./rpctl view --exp-id 20260709T081122Z_ours-pp-object-cache_2runs_600s
  ./rpctl delete --exp-id old_experiment_id --dry-run
  ./rpctl tail -n 120
  ./rpctl stop

Data model:
  - Experiment data stays under ./runs/<profile>/<exp-id>/.
  - Each experiment starts with a snapshot; later child runs reuse the profile state.
  - rpki-prover labels reused-state runs as warm because its native storage decides snapshot/delta behavior.
  - By default only one RP/profile can run at a time.
""",
    )
    sub = parser.add_subparsers(dest="cmd", required=True, metavar="COMMAND")

    p = sub.add_parser(
        "init",
        help="initialize runtime files from a #100 runtime bundle",
        description="Initialize rpctl by copying the archived runtime bundle into this local control directory.",
        formatter_class=formatter,
        epilog="""
The bundle must contain:
  bin/rpki
  bin/routinator
  bin/rpki-client
  bin/fort
  lib/
  fixtures/tal/
  scripts/normalize_and_collect.py
  scripts/write_group_report.py
  versions.json

Optional rpki-prover image archive:
  Pass --rpki-prover-image to copy and load the saved linux/amd64 Docker image.
  Init verifies image architecture, tag and rpki-prover version without pulling online.

Recommended source on 47.77.237.41:
  /root/rpki_4rp_ctl_runtime_bundle

Example:
  ./rpctl init --bundle /data/rpki_4rp_ctl_runtime_bundle \\
    --rpki-prover-image /data/rpki-prover-6857d4bf-linux-amd64.docker.tar.gz
""",
    )
    p.add_argument("--bundle", required=True, help="Path to the archived #100 runtime bundle root.")
    p.add_argument(
        "--rpki-prover-image",
        default="",
        help="Optional saved rpki-prover linux/amd64 Docker image archive. Existing four-RP profiles do not require it.",
    )
    p.set_defaults(func=cmd_init)

    p = sub.add_parser(
        "list",
        help="list supported RPs/profiles and aliases",
        description="Show supported profile names, underlying RP name, and accepted aliases.",
        formatter_class=formatter,
        epilog="""
Use this command when unsure which value to pass to --rp.

Important ours RP profiles:
  ours-pp-object-cache  cache-enabled profile
  ours-no-cache         no-cache baseline profile

rpki-prover profiles:
  rpki-prover-incremental  default incremental validation
  rpki-prover-full         --no-incremental-validation baseline
""",
    )
    p.set_defaults(func=cmd_list)

    p = sub.add_parser(
        "start",
        help="start one RP/profile in background",
        description="Start a single RP/profile run loop in the background on this server.",
        formatter_class=formatter,
        epilog="""
Run semantics:
  --runs 1 means one snapshot run.
  --runs 6 means one snapshot run plus five delta runs.
  --interval controls target start-to-start spacing between adjacent runs.
  The first run always resets that profile's state directory.
  Delta runs reuse the same profile state directory.

RP/profile examples:
  ./rpctl start --rp ours --interval 10m --runs 6 --rirs all5
  ./rpctl start --rp ours-no-cache --interval 10m --runs 6 --rirs all5
  ./rpctl start --rp routinator --interval 15m --runs 2 --rirs apnic
  ./rpctl start --rp rpki-client --interval 0 --runs 1 --rirs apnic,arin
  ./rpctl start --rp fort --interval 0 --runs 1 --rirs all5
  ./rpctl start --rp rpki-prover --interval 0 --runs 2 --rirs all5
  ./rpctl start --rp rpki-prover-full --interval 0 --runs 1 --rirs all5

Cache mapping:
  --rp ours         enables PP cache + ROA cache + child certificate cache.
  --rp ours-no-cache disables those cache flags.

Safety:
  rpctl refuses to start a second RP while another rpctl runner is alive.
""",
    )
    p.add_argument(
        "--rp",
        required=True,
        help="RP/profile or alias. Common values: ours, ours-no-cache, routinator, rpki-client, fort, rpki-prover, rpki-prover-full.",
    )
    p.add_argument(
        "--interval",
        default="10m",
        help="Target start-to-start interval between runs. Supports seconds/minutes/hours, e.g. 0, 30s, 10m, 1h. Default: 10m.",
    )
    p.add_argument(
        "--runs",
        type=int,
        default=6,
        help="Total run count. 1=snapshot only; 6=snapshot+5 deltas. Default: 6.",
    )
    p.add_argument(
        "--rirs",
        default="all5",
        help="RIR set: all5/all, or comma list from afrinic,apnic,arin,lacnic,ripe. Default: all5.",
    )
    p.add_argument(
        "--exp-id",
        default="",
        help="Optional experiment id used in ./runs/<profile>/<exp-id>. Default: generated UTC id.",
    )
    p.set_defaults(func=cmd_start)

    p = sub.add_parser(
        "continue",
        help="append child runs to a successful historical experiment",
        description="Continue an existing successful experiment by appending child runs without resetting RP state.",
        formatter_class=formatter,
        epilog="""
Run semantics:
  --exp-id must refer to a historical experiment whose existing child runs all exited 0.
  The command appends run_000N directories under the same experiment root.
  rpctl does not clean profile state and does not force snapshot-only flags.
  Each RP decides delta/snapshot behavior from its persisted state and native implementation.
  The original RIR set is reused; continue does not allow changing scope mid-run.

Examples:
  ./rpctl continue --exp-id 20260709T081122Z_ours-pp-object-cache_2runs_600s --runs 3 --interval 10m
  ./rpctl continue --rp routinator --exp-id m18_237_routinator_all5_snapshot_delta_20260710T002726Z --runs 1 --interval 0
  ./rpctl continue --rp rpki-prover --exp-id prover_all5 --runs 1 --interval 0
""",
    )
    p.add_argument("--exp-id", required=True, help="Existing successful experiment id to append to.")
    p.add_argument("--rp", default="", help="Optional RP/profile filter if the experiment id exists under multiple profiles.")
    p.add_argument(
        "--interval",
        default="10m",
        help="Target start-to-start interval between newly appended runs. The first appended run starts immediately. Default: 10m.",
    )
    p.add_argument("--runs", type=int, default=1, help="Number of additional runs to append. Default: 1.")
    p.set_defaults(func=cmd_continue)

    p = sub.add_parser(
        "stop",
        help="stop active experiment",
        description="Stop the currently active rpctl experiment runner process group.",
        formatter_class=formatter,
        epilog="""
Default stop sends SIGTERM to the runner process group and waits briefly.
Use --force to send SIGKILL when the RP does not exit.

Stopping does not delete experiment data.

Examples:
  ./rpctl stop
  ./rpctl stop --force
""",
    )
    p.add_argument("--force", action="store_true", help="Send SIGKILL instead of SIGTERM.")
    p.set_defaults(func=cmd_stop)

    p = sub.add_parser(
        "status",
        help="show current status",
        description="Show current or last rpctl experiment status from local state files and PID liveness.",
        formatter_class=formatter,
        epilog="""
Human output includes:
  state, profile, rp, pid, pidAlive, expId, experimentRoot,
  current child run/mode, last child-run wall/RSS/counts if available.

States:
  idle      no state file and no runner pid
  starting  runner just launched
  running   currently executing an RP run
  waiting   waiting for next interval
  stopped   stopped by rpctl stop
  failed    runner finished with non-zero exit code
  success   all requested child runs completed successfully
  exited    state said running, but PID is gone

Examples:
  ./rpctl status
  ./rpctl status --json
""",
    )
    p.add_argument("--json", action="store_true", help="Print machine-readable JSON.")
    p.set_defaults(func=cmd_status)

    p = sub.add_parser(
        "paths",
        help="show important paths",
        description="Print local control, state, log, bundle and experiment data paths.",
        formatter_class=formatter,
        epilog="""
Useful paths:
  root       rpctl installation root
  config     init metadata
  state      current.json, runner.pid, runner.pgid
  runs       all generated experiment data
  logs       runner nohup logs
  bundle     original bundle path recorded by init

Example:
  ./rpctl paths
""",
    )
    p.set_defaults(func=cmd_paths)

    p = sub.add_parser(
        "tail",
        help="show runner log tail",
        description="Print the tail of the latest/current rpctl runner log.",
        formatter_class=formatter,
        epilog="""
This shows rpctl runner errors and Python stack traces if the wrapper fails.
Per-RP stdout/stderr are stored inside the current child-run directory.
Use ./rpctl paths to find the latest experiment directory.

Examples:
  ./rpctl tail
  ./rpctl tail -n 200
""",
    )
    p.add_argument("-n", "--lines", type=int, default=80, help="Number of log lines to print. Default: 80.")
    p.set_defaults(func=cmd_tail)

    p = sub.add_parser(
        "show",
        help="list historical experiment ids and summaries",
        description="List historical rpctl experiment roots with status, start/end time, parameters and product counts.",
        formatter_class=formatter,
        epilog="""
Show scans ./runs/<profile>/<exp-id>/ and does not require the experiment to be active.
It can identify failed/partial runs from exit codes, missing end time, or current status.

Examples:
  ./rpctl show
  ./rpctl show --failed
  ./rpctl show --rp ours --limit 20
  ./rpctl show --paths
  ./rpctl show --json
""",
    )
    p.add_argument("--rp", default="", help="Optional RP/profile filter, e.g. ours, ours-no-cache, routinator, rpki-client, fort, rpki-prover.")
    p.add_argument("--failed", action="store_true", help="Only show failed, partial, incomplete or exited experiments.")
    p.add_argument("--limit", type=int, default=50, help="Maximum rows to show. 0 means no limit. Default: 50.")
    p.add_argument("--paths", action="store_true", help="Also show full experiment root paths. Hidden by default to keep the table readable.")
    p.add_argument("--json", action="store_true", help="Print machine-readable JSON.")
    p.set_defaults(func=cmd_show)

    p = sub.add_parser(
        "view",
        help="show one historical experiment summary",
        description="Show summary and child-run details for a specific historical experiment id.",
        formatter_class=formatter,
        epilog="""
Use show first to find the experiment id. If the same experiment id exists under multiple
profiles, add --rp to disambiguate.

Examples:
  ./rpctl view --exp-id 20260709T081122Z_ours-pp-object-cache_2runs_600s
  ./rpctl view --rp ours --exp-id 20260709T081122Z_ours-pp-object-cache_2runs_600s
  ./rpctl view --exp-id m3_ours_apnic_smoke --command
  ./rpctl view --exp-id m3_ours_apnic_smoke --json
""",
    )
    p.add_argument("--exp-id", required=True, help="Experiment id under ./runs/<profile>/<exp-id>.")
    p.add_argument("--rp", default="", help="Optional RP/profile filter to disambiguate duplicated experiment ids.")
    p.add_argument("--command", action="store_true", help="Also print the first run command line.")
    p.add_argument("--json", action="store_true", help="Print machine-readable JSON.")
    p.set_defaults(func=cmd_view)

    p = sub.add_parser(
        "delete",
        help="delete one non-running experiment id and its local data",
        description="Delete a historical experiment root and matching runner logs. Active/running experiments are refused.",
        formatter_class=formatter,
        epilog="""
Safety rules:
  --exp-id must point to a non-running experiment.
  If the same experiment id exists under multiple profiles, add --rp.
  Use --dry-run first to print the exact experiment root and matching runner logs.
  Actual deletion requires --yes.

Examples:
  ./rpctl delete --exp-id 20260710T025439Z_ours-pp-object-cache_4runs_180s --dry-run
  ./rpctl delete --exp-id 20260710T025439Z_ours-pp-object-cache_4runs_180s --yes
  ./rpctl delete --rp ours --exp-id duplicated_experiment_id --yes
""",
    )
    p.add_argument("--exp-id", required=True, help="Historical non-running experiment id to delete.")
    p.add_argument("--rp", default="", help="Optional RP/profile filter to disambiguate duplicated experiment ids.")
    p.add_argument("--dry-run", action="store_true", help="Only print the paths that would be deleted.")
    p.add_argument("--yes", action="store_true", help="Confirm deletion.")
    p.set_defaults(func=cmd_delete)

    p = sub.add_parser("_runner", help=argparse.SUPPRESS)
    p.add_argument("--profile", required=True)
    p.add_argument("--exp-id", required=True)
    p.add_argument("--runs", type=int, required=True)
    p.add_argument("--interval-secs", type=int, required=True)
    p.add_argument("--rirs", required=True)
    p.add_argument("--append", action="store_true")
    p.set_defaults(func=cmd_runner)
    return parser

def main() -> None:
    args = build_parser().parse_args()
    args.func(args)


if __name__ == "__main__":
    main()