#!/usr/bin/env python3
"""pkh end-to-end package harness.

Runs real packages from catalog.yml through the full pkh workflow:

    pkh pull  (from the archive, by default)
    pkh chlog --rebuild   (no-change rebuild entry; EDITOR is stubbed out)
    pkh build             (.dsc + _source.changes next to the tree)
    pkh deb               (binary packages in an ephemeral chroot)

Every stage is logged per package, timed, and checked for its artifacts;
results land in <work>/runs/<id>/results.json and results.md. Pulls run in
parallel (network-bound), the build pipeline runs sequentially (one heavy
build at a time, and pkh's shared chroot/apt/ccache caches stay warm).

Usage:
    run.py --list
    run.py --dry-run
    run.py                          # smoke + standard tiers
    run.py --tier long              # the multi-hour builds
    run.py --only hello,dracut      # exact names
    run.py --resume runs/<id>/results.json   # continue an interrupted run

See README.md in this directory for the full documentation.
"""

from __future__ import annotations

import argparse
import concurrent.futures
import datetime as dt
import json
import os
import platform
import shlex
import shutil
import signal
import subprocess
import sys
import threading
import time
from dataclasses import dataclass, field
from pathlib import Path

TIERS = ("smoke", "standard", "long")
DEFAULT_TIERS = ("smoke", "standard")
TIER_TIMEOUT_MIN = {"smoke": 30, "standard": 120, "long": 360}
STAGES = ("pull", "chlog", "build", "deb")
PIPELINE_STAGES = ("chlog", "build", "deb")
TERMINAL_OUTCOMES = ("pass", "xpass", "xfail")
REPO_ROOT = Path(__file__).resolve().parents[2]
SCRIPT_DIR = Path(__file__).resolve().parent


class CatalogError(Exception):
    """Invalid catalog content or unusable options."""


@dataclass
class Entry:
    """One catalog entry: a source package and how to build it."""

    name: str
    category: str = "misc"
    tier: str = "standard"
    dist: str = "debian"
    series: str = ""
    arch: str = ""
    cross: bool = False
    pocket: str = ""
    ppa: list[str] = field(default_factory=list)
    timeout: float = 0.0  # minutes; 0 = the tier default
    xfail: str = ""
    notes: str = ""
    tags: list[str] = field(default_factory=list)

    def timeout_min(self) -> float:
        return self.timeout if self.timeout else TIER_TIMEOUT_MIN.get(self.tier, 120)


# ---------------------------------------------------------------------------
# Catalog loading: PyYAML when available, a schema-specific fallback parser
# otherwise (the subset it accepts is documented in catalog.yml).
# ---------------------------------------------------------------------------


def _scalar(text: str):
    text = text.strip()
    if text.startswith("[") and text.endswith("]"):
        inner = text[1:-1].strip()
        return [v.strip().strip("'\"") for v in inner.split(",") if v.strip()]
    if len(text) >= 2 and text[0] == text[-1] and text[0] in "'\"":
        return text[1:-1]
    if text in ("true", "false"):
        return text == "true"
    if text.isdigit():
        return int(text)
    return text


def _parse_mini_yaml(text: str) -> dict:
    """Parse the flat 'defaults' mapping + flat 'packages' list subset."""
    doc: dict = {"defaults": {}, "packages": []}
    mode = None  # None | "defaults" | "packages"
    item: dict | None = None
    for lineno, raw in enumerate(text.splitlines(), 1):
        if not raw.strip() or raw.lstrip().startswith("#"):
            continue
        indent = len(raw) - len(raw.lstrip(" "))
        stripped = raw.strip()
        if indent == 0:
            if stripped in ("defaults:", "packages:"):
                mode = stripped[:-1]
                item = None
                continue
            raise CatalogError(f"catalog:{lineno}: unexpected top-level '{stripped}'")
        if mode is None:
            raise CatalogError(f"catalog:{lineno}: content outside a top-level key")
        if mode == "defaults":
            key, sep, value = stripped.partition(":")
            if not sep:
                raise CatalogError(f"catalog:{lineno}: expected 'key: value'")
            doc["defaults"][key.strip()] = _scalar(value)
        else:
            if stripped.startswith("- "):
                item = {}
                doc["packages"].append(item)
                stripped = stripped[2:].strip()
                if not stripped:
                    continue
            if item is None:
                raise CatalogError(f"catalog:{lineno}: list items must start with '- '")
            key, sep, value = stripped.partition(":")
            if not sep:
                raise CatalogError(f"catalog:{lineno}: expected 'key: value'")
            item[key.strip()] = _scalar(value)
    return doc


def load_catalog(path: Path) -> list[Entry]:
    text = path.read_text(encoding="utf-8")
    use_pyyaml = not os.environ.get("PKH_E2E_NO_PYYAML")
    try:
        if use_pyyaml:
            import yaml  # type: ignore import-not-found

            doc = yaml.safe_load(text)
        else:
            doc = _parse_mini_yaml(text)
    except CatalogError:
        raise
    except Exception as exc:
        if use_pyyaml:
            raise CatalogError(f"cannot parse {path}: {exc}") from exc
        raise

    defaults = doc.get("defaults") or {}
    raw_entries = doc.get("packages") or []
    if not raw_entries:
        raise CatalogError(f"{path}: no packages listed")

    entries: list[Entry] = []
    seen: set[str] = set()
    for i, raw in enumerate(raw_entries):
        name = str(raw.get("name", "")).strip()
        if not name:
            raise CatalogError(f"{path}: package #{i + 1} has no name")
        if name in seen:
            raise CatalogError(f"{path}: duplicate package '{name}'")
        seen.add(name)

        entry = Entry(
            name=name,
            category=str(raw.get("category", "misc")),
            tier=str(raw.get("tier", defaults.get("tier", "standard"))),
            dist=str(raw.get("dist", defaults.get("dist", "debian"))),
            series=str(raw.get("series", "")),
            arch=str(raw.get("arch", "")),
            cross=bool(raw.get("cross", False)),
            pocket=str(raw.get("pocket", "")),
            ppa=[str(p) for p in (raw.get("ppa") or [])],
            timeout=float(raw.get("timeout", 0)),
            xfail=str(raw.get("xfail", "")),
            notes=str(raw.get("notes", "")),
            tags=[str(t) for t in (raw.get("tags") or [])],
        )
        if entry.tier not in TIERS:
            raise CatalogError(f"{path}: {name}: unknown tier '{entry.tier}'")
        if entry.dist not in ("debian", "ubuntu"):
            raise CatalogError(f"{path}: {name}: unknown dist '{entry.dist}'")
        if entry.cross and not entry.arch:
            raise CatalogError(f"{path}: {name}: cross: true needs an arch")
        entries.append(entry)
    return entries


# ---------------------------------------------------------------------------
# Selection and rendering
# ---------------------------------------------------------------------------


def select_entries(
    entries: list[Entry], args: argparse.Namespace
) -> tuple[list[Entry], list[str]]:
    """Apply the CLI filters; returns the selection and the filtered-out names."""
    tiers = (
        list(TIERS) if args.tier == "all" else [t.strip() for t in args.tier.split(",")]
    )
    for tier in tiers:
        if tier not in TIERS:
            raise CatalogError(f"unknown tier '{tier}' (expected: {', '.join(TIERS)})")

    selected = [e for e in entries if e.tier in tiers]
    dropped = [e.name for e in entries if e.tier not in tiers]

    if args.category:
        wanted = {c.strip() for c in args.category.split(",") if c.strip()}
        dropped += [e.name for e in selected if e.category not in wanted]
        selected = [e for e in selected if e.category in wanted]
    if args.tag:
        wanted = {t.strip() for t in args.tag.split(",") if t.strip()}
        dropped += [e.name for e in selected if not wanted.intersection(e.tags)]
        selected = [e for e in selected if wanted.intersection(e.tags)]
    if args.only:
        names = {n.strip() for n in args.only.split(",") if n.strip()}
        unknown = sorted(names - {e.name for e in entries})
        if unknown:
            raise CatalogError(f"--only: not in the catalog: {', '.join(unknown)}")
        dropped += [e.name for e in selected if e.name not in names]
        selected = [e for e in selected if e.name in names]
    if args.skip:
        names = {n.strip() for n in args.skip.split(",") if n.strip()}
        dropped += [e.name for e in selected if e.name in names]
        selected = [e for e in selected if e.name not in names]
    return selected, dropped


def fmt_duration(seconds: float | None) -> str:
    if seconds is None:
        return "-"
    seconds = int(seconds)
    if seconds < 60:
        return f"{seconds}s"
    if seconds < 3600:
        return f"{seconds // 60}m{seconds % 60:02d}s"
    return f"{seconds // 3600}h{(seconds % 3600) // 60:02d}m"


def render_list(entries: list[Entry]) -> str:
    lines = [
        f"{'PACKAGE':<20} {'TIER':<9} {'DIST':<7} {'CATEGORY':<14} "
        f"{'TAGS':<30} NOTES"
    ]
    for e in entries:
        extra = []
        if e.cross:
            extra.append(f"cross {e.arch}")
        if e.series:
            extra.append(f"series {e.series}")
        note = " ".join(extra + [e.xfail or e.notes])
        lines.append(
            f"{e.name:<20} {e.tier:<9} {e.dist:<7} {e.category:<14} "
            f"{','.join(e.tags):<30} {note}"
        )
    return "\n".join(lines) + f"\n\n{len(entries)} package(s)\n"


# ---------------------------------------------------------------------------
# Command construction and child-process plumbing
# ---------------------------------------------------------------------------


def stage_command(args: argparse.Namespace, entry: Entry, stage: str, pkh: Path) -> list[str]:
    cmd = [str(pkh), stage]
    if stage == "pull":
        cmd += [entry.name, "-d", args.dist or entry.dist]
        if args.series or entry.series:
            cmd += ["-s", args.series or entry.series]
        if args.pocket or entry.pocket:
            cmd += ["-p", args.pocket or entry.pocket]
        if not args.git:
            cmd += ["--archive"]
    elif stage == "chlog":
        cmd += ["--rebuild"]
        if args.series or entry.series:
            cmd += ["-s", args.series or entry.series]
    elif stage == "build":
        cmd += ["--verbose"]
    elif stage == "deb":
        cmd += ["--verbose"]
        if args.series or entry.series:
            cmd += ["-s", args.series or entry.series]
        arch = args.arch or entry.arch
        if arch:
            cmd += ["-a", arch]
        if entry.cross or args.cross:
            cmd += ["--cross"]
        if args.pocket or entry.pocket:
            cmd += ["-p", args.pocket or entry.pocket]
        for ppa in entry.ppa + args.ppa:
            cmd += ["--ppa", ppa]
        if args.jobs:
            cmd += ["-j", str(args.jobs)]
        if args.no_ccache:
            cmd += ["--no-ccache"]
    return cmd


def child_env() -> dict:
    """Environment for pkh children: no editor, a deterministic identity."""
    env = dict(os.environ)
    env["EDITOR"] = "true"  # chlog launches it after generating; nothing to edit
    env.setdefault("DEBEMAIL", "pkh-e2e@invalid")
    env.setdefault("DEBFULLNAME", "pkh e2e")
    return env


def run_stage(
    cmd: list[str], cwd: Path, log_path: Path, timeout_min: float
) -> tuple[str, float]:
    """Run one stage, logging to log_path; returns (status, seconds).

    Status is 'ok', 'fail' or 'timeout'. The whole process group is killed
    on timeout: builds fan out into make/compiler children, and killing only
    the direct child would leave them running. On Ctrl+C the group gets a
    graceful SIGTERM first, mirroring pkh's own interrupt cleanup.
    """
    with log_path.open("ab") as log:
        log.write(f"\n$ {shlex.join(cmd)}\n(cwd: {cwd})\n".encode())
        log.flush()
        start = time.monotonic()
        proc = subprocess.Popen(
            cmd,
            cwd=cwd,
            env=child_env(),
            stdin=subprocess.DEVNULL,
            stdout=log,
            stderr=subprocess.STDOUT,
            start_new_session=True,
        )
        try:
            proc.wait(timeout=timeout_min * 60)
        except subprocess.TimeoutExpired:
            _kill_group(proc, signal.SIGKILL)
            return "timeout", time.monotonic() - start
        except KeyboardInterrupt:
            # A graceful stop first: SIGTERM lets pkh run its own interrupt
            # cleanup; a child ignoring it is killed after a grace period.
            _kill_group(proc, signal.SIGTERM, wait=10)
            _kill_group(proc, signal.SIGKILL)
            raise
        return ("ok" if proc.returncode == 0 else "fail"), time.monotonic() - start


def _kill_group(
    proc: subprocess.Popen, sig: signal.Signals, wait: float | None = None
) -> None:
    try:
        os.killpg(os.getpgid(proc.pid), sig)
    except ProcessLookupError:
        pass
    try:
        proc.wait(timeout=wait)
    except subprocess.TimeoutExpired:
        pass


# ---------------------------------------------------------------------------
# The harness proper
# ---------------------------------------------------------------------------


class Harness:
    def __init__(self, args: argparse.Namespace, entries: list[Entry], pkh: Path):
        self.args = args
        self.entries = entries
        self.pkh = pkh
        self.stages = [s.strip() for s in args.stages.split(",") if s.strip()]
        self.records: dict[str, dict] = {}
        self.pull_results: dict[str, tuple[str, float]] = {}
        self.lock = threading.Lock()
        self.run_dir = self._open_run_dir()
        self.results_path = self.run_dir / "results.json"
        (self.run_dir / "logs").mkdir(parents=True, exist_ok=True)

    # -- layout -------------------------------------------------------------

    def _open_run_dir(self) -> Path:
        if self.args.resume:
            try:
                data = json.loads(Path(self.args.resume).read_text())
                run_dir = Path(data["run_dir"])
            except (OSError, KeyError, json.JSONDecodeError) as exc:
                raise CatalogError(f"--resume: cannot read {self.args.resume}: {exc}")
            if not run_dir.is_dir():
                raise CatalogError(f"--resume: missing run directory {run_dir}")
            for record in data.get("entries", []):
                self.records[record["name"]] = record
            return run_dir
        stamp = dt.datetime.now().strftime("%Y%m%d-%H%M%S")
        return SCRIPT_DIR / "runs" / f"{stamp}-{os.getpid()}"

    def log_path(self, name: str) -> Path:
        return self.run_dir / "logs" / f"{name}.log"

    def workroot(self, entry: Entry) -> Path:
        # pkh pull (run from the run root) puts the tree at
        # ./<pkg>/<pkg>/: this directory is both the tree's parent and
        # where pkh build / pkh deb drop their artifacts.
        return self.run_dir / entry.name

    def tree(self, entry: Entry) -> Path:
        return self.workroot(entry) / entry.name

    # -- stages ---------------------------------------------------------------

    def do_pull(self, entry: Entry) -> tuple[str, float]:
        # Pull from the run root: pkh pull (cwd-less) extracts to
        # ./<pkg>/<pkg>/ under its cwd, so every package still gets its own
        # subtree and parallel pulls never collide. A failed pull leaves a
        # partial tree that pkh pull would refuse to overwrite and that a
        # later --resume must not trust: wipe it before retrying and after
        # the last attempt.
        for attempt in range(self.args.pull_retries + 1):
            if attempt:
                shutil.rmtree(self.workroot(entry), ignore_errors=True)
            status, seconds = run_stage(
                stage_command(self.args, entry, "pull", self.pkh),
                self.run_dir,
                self.log_path(entry.name),
                entry.timeout_min(),
            )
            if status == "ok":
                return status, seconds
            shutil.rmtree(self.workroot(entry), ignore_errors=True)
        return status, seconds

    def do_chlog(self, entry: Entry) -> tuple[str, float]:
        return run_stage(
            stage_command(self.args, entry, "chlog", self.pkh),
            self.tree(entry),
            self.log_path(entry.name),
            entry.timeout_min(),
        )

    def do_build(self, entry: Entry) -> tuple[tuple[str, float], list[str]]:
        record = run_stage(
            stage_command(self.args, entry, "build", self.pkh),
            self.tree(entry),
            self.log_path(entry.name),
            entry.timeout_min(),
        )
        artifacts = self.build_artifacts(entry)
        return self._check_artifacts(entry, record, artifacts, "dsc/changes"), artifacts

    def do_deb(self, entry: Entry) -> tuple[tuple[str, float], list[str]]:
        record = run_stage(
            stage_command(self.args, entry, "deb", self.pkh),
            self.tree(entry),
            self.log_path(entry.name),
            entry.timeout_min(),
        )
        debs = self.deb_artifacts(entry)
        return self._check_artifacts(entry, record, debs, ".deb"), debs

    def build_artifacts(self, entry: Entry) -> list[str]:
        return sorted(
            p.name
            for pattern in ("*.dsc", "*_source.changes")
            for p in self.workroot(entry).glob(pattern)
        )

    def deb_artifacts(self, entry: Entry) -> list[str]:
        return sorted(p.name for p in self.workroot(entry).glob("*.deb"))

    def _check_artifacts(
        self, entry: Entry, record: tuple[str, float], artifacts: list[str], what: str
    ) -> tuple[str, float]:
        """A green stage with no artifacts is a failure: log it and flip."""
        if record[0] == "ok" and not artifacts:
            with self.log_path(entry.name).open("ab") as log:
                log.write(
                    f"\npkh-e2e: stage reported success but produced no {what}\n".encode()
                )
            return ("fail", record[1])
        return record

    def changelog_version(self, entry: Entry) -> str:
        try:
            out = subprocess.run(
                ["dpkg-parsechangelog", "-S", "Version"],
                cwd=self.tree(entry),
                capture_output=True,
                text=True,
                check=True,
            )
            return out.stdout.strip()
        except Exception:
            return ""

    # -- per-entry pipeline ------------------------------------------------------

    def run_entry(self, entry: Entry) -> dict:
        """Full pipeline for one entry; never raises."""
        # A resumed run keeps what the previous record already achieved
        previous = self.records.get(entry.name)
        record = {
            "name": entry.name,
            "category": entry.category,
            "tier": entry.tier,
            "dist": self.args.dist or entry.dist,
            "series": self.args.series or entry.series,
            "arch": self.args.arch or entry.arch or platform.machine(),
            "cross": entry.cross or self.args.cross,
            "tags": entry.tags,
            "xfail": entry.xfail,
            "notes": entry.notes,
            "outcome": "fail",
            "version": "",
            "stages": {},
            "artifacts": [],
            "total_seconds": 0.0,
            "log": str(self.log_path(entry.name).relative_to(self.run_dir)),
        }
        started = time.monotonic()
        try:
            self._run_entry(record, entry, previous)
        finally:
            record["total_seconds"] = time.monotonic() - started
            if entry.xfail:
                record["outcome"] = "xfail" if record["outcome"] == "fail" else "xpass"
            with self.lock:
                self.records[entry.name] = record
            self.flush()
            color = {"pass": 32, "xfail": 33, "xpass": 33}.get(record["outcome"], 31)
            print(
                f"[{entry.name}] \033[{color}m{record['outcome']}\033[0m "
                f"({fmt_duration(record['total_seconds'])})"
                + (f" — {record['error']}" if record.get("error") else "")
            )
        return record

    def _run_entry(self, record: dict, entry: Entry, previous: dict | None) -> None:
        # Stages the resumed record already completed are reused rather than
        # re-run: a resumed chlog must not bump the version a second time.
        # Reuse only while the build selection is unchanged.
        reusable = previous is not None and all(
            previous.get(key) == record.get(key)
            for key in ("dist", "series", "arch", "cross")
        )
        prev_stages = previous.get("stages", {}) if previous else {}

        if "pull" in self.stages:
            if (self.tree(entry) / "debian").is_dir():
                # Resumed run: the tree survived, keep it (pkh pull refuses
                # existing directories anyway).
                record["stages"]["pull"] = {"status": "ok", "seconds": 0.0,
                                            "cached": True}
                print(f"[{entry.name}]   pull ok (tree already present)")
            else:
                status, seconds = self.pull_results.get(entry.name, ("fail", 0.0))
                record["stages"]["pull"] = {"status": status, "seconds": seconds}
                if status != "ok":
                    record["error"] = f"pull {status}"
                    return
                # Already reported by the pull phase; no duplicate line.

        for stage in PIPELINE_STAGES:
            if stage not in self.stages:
                continue
            prior = prev_stages.get(stage) if reusable else None
            if prior and prior.get("status") == "ok":
                record["stages"][stage] = {**prior, "reused": True}
                if stage == "chlog":
                    record["version"] = previous.get("version", "")
                elif stage == "build":
                    record["artifacts"] += self.build_artifacts(entry)
                elif stage == "deb":
                    record["artifacts"] += self.deb_artifacts(entry)
                print(f"[{entry.name}]   {stage} ok (reused)")
                continue
            if not (self.tree(entry) / "debian").is_dir():
                record["error"] = (
                    f"no source tree for {stage}; include the pull stage or "
                    "resume the original run directory"
                )
                return

            if stage == "chlog":
                status, seconds = self.do_chlog(entry)
                record["stages"]["chlog"] = {"status": status, "seconds": seconds}
                if status != "ok":
                    record["error"] = f"chlog {status}"
                    return
                record["version"] = self.changelog_version(entry)
                print(
                    f"[{entry.name}]   chlog ok ({fmt_duration(seconds)})"
                    + (f" -> {record['version']}" if record["version"] else "")
                )
            elif stage == "build":
                (status, seconds), artifacts = self.do_build(entry)
                record["stages"]["build"] = {"status": status, "seconds": seconds}
                record["artifacts"] += artifacts
                if status != "ok":
                    record["error"] = f"build {status}"
                    return
                print(f"[{entry.name}]   build ok ({fmt_duration(seconds)})")
            elif stage == "deb":
                (status, seconds), debs = self.do_deb(entry)
                record["stages"]["deb"] = {"status": status, "seconds": seconds}
                record["artifacts"] += debs
                if status != "ok":
                    record["error"] = f"deb {status}"
                    return
                print(
                    f"[{entry.name}]   deb ok ({fmt_duration(seconds)}), "
                    f"{len(debs)} binary package(s)"
                )

        record["outcome"] = "pass"

    # -- run orchestration ---------------------------------------------------------

    def preflight(self) -> None:
        version = subprocess.run(
            [str(self.pkh), "--version"], capture_output=True, text=True
        )
        print(f"[pkh-e2e] pkh: {self.pkh} ({version.stdout.strip() or '?'})")
        if "deb" in self.stages:
            missing = [
                tool
                for tool in ("mmdebstrap", "unshare", "sudo")
                if shutil.which(tool) is None
            ]
            if missing:
                print(f"[pkh-e2e] WARNING: deb needs: {', '.join(missing)}")
            if subprocess.run(["sudo", "-n", "true"], capture_output=True).returncode:
                print(
                    "[pkh-e2e] WARNING: passwordless sudo unavailable; "
                    "ephemeral chroot builds will fail"
                )
        free_g = shutil.disk_usage(self.run_dir).free / 2**30
        if free_g < 20:
            print(f"[pkh-e2e] WARNING: only {free_g:.0f}G free on {self.run_dir}")
        print(
            f"[pkh-e2e] {len(self.entries)} package(s), stages: "
            f"{'+'.join(self.stages)} -> {self.run_dir}"
        )

    def resume_skip(self, entry: Entry) -> bool:
        previous = self.records.get(entry.name)
        if not previous or previous.get("outcome") not in TERMINAL_OUTCOMES:
            return False
        return all(s in previous.get("stages", {}) for s in self.stages)

    def pull_phase(self) -> None:
        if "pull" not in self.stages:
            return
        todo = [
            e
            for e in self.entries
            if not self.resume_skip(e) and not (self.tree(e) / "debian").is_dir()
        ]
        if not todo:
            return
        print(
            f"[pkh-e2e] pulling {len(todo)} package(s) "
            f"({self.args.pull_jobs} at a time)"
        )
        with concurrent.futures.ThreadPoolExecutor(
            max_workers=self.args.pull_jobs
        ) as pool:
            futures = {pool.submit(self.do_pull, e): e for e in todo}
            for future in concurrent.futures.as_completed(futures):
                entry = futures[future]
                try:
                    status, seconds = future.result()
                except Exception as exc:  # keep the run going on worker errors
                    status, seconds = "fail", 0.0
                    print(f"[{entry.name}]   pull crashed: {exc}")
                self.pull_results[entry.name] = (status, seconds)
                print(
                    f"[{entry.name}]   pull {'ok' if status == 'ok' else status} "
                    f"({fmt_duration(seconds)})"
                )

    def run(self) -> int:
        self.preflight()
        self.flush()

        self.pull_phase()

        failures = 0
        for entry in self.entries:
            if self.resume_skip(entry):
                print(f"[{entry.name}] already {self.records[entry.name]['outcome']},"
                      " skipping")
                continue
            if self.args.fail_fast and failures:
                print(f"[{entry.name}] skipped (fail-fast)")
                continue
            record = self.run_entry(entry)
            if record["outcome"] in ("fail", "timeout"):
                failures += 1

        self.write_report()
        self.print_summary()
        return self.exit_code(failures)

    def print_summary(self) -> None:
        rows = sorted(self.records.values(), key=lambda r: r["name"])
        print(f"\n=== summary ({len(rows)} recorded) ===")
        print(
            f"{'PACKAGE':<20} {'TIER':<9} {'RESULT':<8} "
            f"{'PULL':>7} {'CHLOG':>7} {'BUILD':>8} {'DEB':>8} {'TOTAL':>8}  VERSION"
        )
        for r in rows:
            def sec(stage):
                data = r["stages"].get(stage)
                return fmt_duration(data["seconds"]) if data else "-"

            print(
                f"{r['name']:<20} {r['tier']:<9} {r['outcome']:<8} "
                f"{sec('pull'):>7} {sec('chlog'):>7} {sec('build'):>8} "
                f"{sec('deb'):>8} {fmt_duration(r['total_seconds']):>8}  "
                f"{r['version']}"
            )
        tally: dict[str, int] = {}
        for r in rows:
            tally[r["outcome"]] = tally.get(r["outcome"], 0) + 1
        print("\n" + ", ".join(f"{v} {k}" for k, v in sorted(tally.items())))

    def exit_code(self, failures: int) -> int:
        if failures:
            return 1
        if self.args.strict and any(
            r["outcome"] == "xpass" for r in self.records.values()
        ):
            return 1
        return 0

    # -- persistence ---------------------------------------------------------------

    def flush(self) -> None:
        """Persist results progressively: a killed run keeps its records."""
        if not hasattr(self, "_started"):
            self._started = dt.datetime.now().astimezone().isoformat()
        meta = {
            "run_dir": str(self.run_dir),
            "pkh": str(self.pkh),
            "filters": {
                "tier": self.args.tier,
                "category": self.args.category,
                "tag": self.args.tag,
                "only": self.args.only,
                "skip": self.args.skip,
                "stages": self.stages,
            },
            "host": {
                "kernel": platform.release(),
                "arch": platform.machine(),
                "python": platform.python_version(),
            },
            "started": self._started,
            "updated": dt.datetime.now().astimezone().isoformat(),
            "entries": sorted(self.records.values(), key=lambda r: r["name"]),
        }
        tmp = self.results_path.with_suffix(".json.tmp")
        tmp.write_text(json.dumps(meta, indent=2) + "\n")
        tmp.replace(self.results_path)

    def write_report(self) -> None:
        rows = sorted(self.records.values(), key=lambda r: (r["outcome"], r["name"]))
        lines = [
            f"# pkh e2e run — {self.run_dir.name}",
            "",
            f"- pkh: `{self.pkh}`",
            f"- started: {getattr(self, '_started', '?')}",
            f"- stages: {'+'.join(self.stages)}",
            "",
            "| package | tier | result | pull | chlog | build | deb | total |"
            " version |",
            "|---|---|---|---|---|---|---|---|---|",
        ]
        for r in rows:
            def sec(stage):
                data = r["stages"].get(stage)
                return fmt_duration(data["seconds"]) if data else "-"

            lines.append(
                f"| {r['name']} | {r['tier']} | {r['outcome']} | {sec('pull')} | "
                f"{sec('chlog')} | {sec('build')} | {sec('deb')} | "
                f"{fmt_duration(r['total_seconds'])} | {r['version']} |"
            )
        (self.run_dir / "results.md").write_text("\n".join(lines) + "\n")


# ---------------------------------------------------------------------------
# Entry point
# ---------------------------------------------------------------------------


def parse_args(argv: list[str]) -> argparse.Namespace:
    parser = argparse.ArgumentParser(
        description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter
    )
    parser.add_argument("--catalog", type=Path, default=SCRIPT_DIR / "catalog.yml",
                        help="catalog file (default: catalog.yml)")
    parser.add_argument("--tier", default=",".join(DEFAULT_TIERS),
                        help="tiers to run, comma-separated: smoke, standard, long, "
                             "or 'all' (default: smoke,standard)")
    parser.add_argument("--category", default="",
                        help="comma-separated categories to keep")
    parser.add_argument("--tag", default="",
                        help="comma-separated tags; keep packages carrying any")
    parser.add_argument("--only", default="",
                        help="comma-separated exact package names to run")
    parser.add_argument("--skip", default="",
                        help="comma-separated exact package names to skip")
    parser.add_argument("--stages", default=",".join(STAGES),
                        help=f"stages to run (default: {','.join(STAGES)})")
    parser.add_argument("--pkh", default="",
                        help="pkh binary to exercise (default: the repo build, "
                             "then PATH)")
    parser.add_argument("--build-pkh", action="store_true",
                        help="run 'cargo build --release' before resolving pkh")
    parser.add_argument("--git", action="store_true",
                        help="pull through the package VCS instead of the archive")
    parser.add_argument("--dist", default="",
                        help="override the catalog default dist (per-entry dist wins)")
    parser.add_argument("--series", default="",
                        help="override the series (per-entry series wins)")
    parser.add_argument("--arch", default="",
                        help="force the 'pkh deb -a' architecture on every entry")
    parser.add_argument("--cross", action="store_true",
                        help="pass --cross to every deb build")
    parser.add_argument("--pocket", default="",
                        help="distribution pocket passed to pull and deb")
    parser.add_argument("--ppa", action="append", default=[],
                        help="PPA passed to 'pkh deb --ppa' (repeatable)")
    parser.add_argument("--jobs", type=int, default=0,
                        help="pass -j N to 'pkh deb' (default: pkh chooses)")
    parser.add_argument("--no-ccache", action="store_true",
                        help="pass --no-ccache to 'pkh deb'")
    parser.add_argument("--pull-jobs", type=int, default=4,
                        help="parallel pulls (default: 4)")
    parser.add_argument("--pull-retries", type=int, default=1,
                        help="retries per failed pull (default: 1)")
    parser.add_argument("--fail-fast", action="store_true",
                        help="stop the pipeline at the first failure")
    parser.add_argument("--strict", action="store_true",
                        help="also fail the run on xpass")
    parser.add_argument("--resume", default="",
                        help="results.json of a previous run: reuse its run "
                             "directory and skip finished packages")
    parser.add_argument("--list", action="store_true",
                        help="print the selected packages and exit")
    parser.add_argument("--dry-run", action="store_true",
                        help="print the plan and the exact commands, run nothing")
    return parser.parse_args(argv)


def main(argv: list[str]) -> int:
    args = parse_args(argv)
    try:
        entries = load_catalog(args.catalog)
        selected, dropped = select_entries(entries, args)

        stages = [s.strip() for s in args.stages.split(",") if s.strip()]
        for stage in stages:
            if stage not in STAGES:
                raise CatalogError(
                    f"unknown stage '{stage}' (expected: {', '.join(STAGES)})"
                )
        if args.pull_jobs < 1:
            raise CatalogError("--pull-jobs must be at least 1")

        if args.list:
            print(render_list(selected))
            if dropped:
                print(f"(filtered out: {len(dropped)})")
            return 0
        if not selected:
            print("nothing selected", file=sys.stderr)
            return 2

        pkh = resolve_pkh(args)
        if args.dry_run:
            print(f"pkh: {pkh}")
            print(f"selected {len(selected)} package(s), stages: {args.stages}")
            for entry in selected:
                where = f" {entry.series}" if entry.series else ""
                print(f"\n[{entry.name}] {entry.tier}/{entry.category} "
                      f"({entry.dist}{where})")
                for stage in stages:
                    cmd = stage_command(args, entry, stage, pkh)
                    print(f"  {shlex.join(cmd)}"
                          f"  # timeout {entry.timeout_min():.0f}m")
            return 0

        harness = Harness(args, selected, pkh)
        return harness.run()
    except CatalogError as exc:
        print(f"error: {exc}", file=sys.stderr)
        return 2
    except KeyboardInterrupt:
        print("\ninterrupted", file=sys.stderr)
        return 130


def repo_source_epoch() -> float | None:
    """The newest evidence of source changes: HEAD's commit time, or the
    newest mtime under src/ for uncommitted edits. None outside a git
    checkout with no src/ tree."""
    stamps = []
    try:
        out = subprocess.run(
            ["git", "log", "-1", "--format=%ct"],
            cwd=REPO_ROOT, capture_output=True, text=True, check=True,
        )
        stamps.append(int(out.stdout.strip()))
    except Exception:
        pass
    src = REPO_ROOT / "src"
    if src.is_dir():
        stamps.extend(path.stat().st_mtime for path in src.rglob("*.rs"))
    return max(stamps) if stamps else None


def resolve_pkh(args: argparse.Namespace) -> Path:
    if args.pkh:
        path = Path(args.pkh).resolve()
        if not path.is_file():
            raise CatalogError(f"--pkh: no such file: {path}")
        return path

    # A repo binary older than the sources measures whatever the last
    # `cargo build --release` happened to include — yesterday's code.
    # Rebuild unless the caller pinned a binary or asked for it explicitly.
    if not args.build_pkh:
        head = repo_source_epoch()
        fresh = head is not None and any(
            candidate.is_file() and candidate.stat().st_mtime >= head
            for candidate in (REPO_ROOT / "target/release/pkh",
                              REPO_ROOT / "target/debug/pkh")
        )
        if fresh:
            return next(candidate
                        for candidate in (REPO_ROOT / "target/release/pkh",
                                          REPO_ROOT / "target/debug/pkh")
                        if candidate.is_file()
                        and candidate.stat().st_mtime >= head)
        print("[pkh-e2e] the repo pkh binary is older than the sources; "
              "running 'cargo build --release'")

    subprocess.run(["cargo", "build", "--release"], cwd=REPO_ROOT, check=True)
    for candidate in (REPO_ROOT / "target/release/pkh", REPO_ROOT / "target/debug/pkh"):
        if candidate.is_file():
            return candidate
    found = shutil.which("pkh")
    if not found:
        raise CatalogError(
            "no pkh binary: pass --pkh, use --build-pkh, or install pkh on PATH"
        )
    return Path(found)


if __name__ == "__main__":
    sys.exit(main(sys.argv[1:]))