#!/usr/bin/python3
"""afl-health — health, trend, and action analysis for AFL++/cargo-afl/ziggy campaigns.

(c) 2026 Marc "vanHauser" Heuse

A single-file, dependency-free (stdlib-only) CLI that diagnoses AFL++ / cargo-afl /
ziggy fuzzing campaigns — locally or over SSH — and emits a verdict, a within/cross-run
trend, and ranked machine-actionable action directives. It is read-only: it never
writes to or otherwise touches the campaign it inspects.

Run it directly (no install needed):

    python3 afl-health.py /path/to/out
    ./afl-health.py user@host:/path/to/out --json
    ./afl-health.py --summary host:/runs/output/*/afl   # glob expands remotely

It needs only Python >= 3.11 locally; the remote side of an ssh run needs nothing
beyond POSIX sh + coreutils.

The pipeline is one-directional, mirrored by the sections below:

    collect    Target ──ssh|local──▶ RawCampaign      (raw fuzzer_stats / plot_data text + counts)
    snapshot   RawCampaign ─────────▶ CampaignSnapshot (normalized facts)
    analyze    CampaignSnapshot ─────▶ Report          (+ optional prior) verdict/findings/actions
    report     Report ──────────────▶ text | JSON
    store      Report ──────────────▶ SQLite row       (for next run's cross-run delta)
    cli        orchestrates all of the above

License: AGPL-3.0 (GNU Affero General Public License, version 3).
"""

from __future__ import annotations

import argparse
import contextlib
import glob
import json
import os
import re
import shlex
import sqlite3
import subprocess
import sys
import time
from collections import Counter
from concurrent.futures import ThreadPoolExecutor
from dataclasses import asdict, dataclass, field, replace
from pathlib import Path
from statistics import median
from typing import Any

__version__ = "0.1"
__license__ = "AGPL-3.0"
SCHEMA_VERSION = 2


# ============================================================================
# model — raw collector output, the normalized snapshot, and the analysis report
# ============================================================================


# --- raw collector output (one per campaign) ------------------------------------
@dataclass
class RawInstance:
    name: str
    fuzzer_stats: str
    plot_data: str | None
    newest_cov_mtime: float | None
    pid_alive: bool
    crash_count: int
    hang_count: int
    core_bytes: int
    newest_queue_mtime: float | None = None
    fastresume_present: bool = False
    main_node_file: bool = False
    no_fastresume_env: bool = False
    starting: bool = False  # fuzzer_stats is missing or older than fuzzer_setup, and afl-fuzz is running
    setup_mtime: float | None = None
    setup_cmdline: str = ""  # the '# command line:' line of fuzzer_setup, still shell-quoted


@dataclass
class RawCampaign:
    campaign_id: str
    host: str
    output_dir: str
    now: int
    layout: str  # "flat" | "ziggy-afl-subdir"
    honggfuzz_present: bool
    collection_error: str | None
    instances: list[RawInstance]


# --- normalized snapshot --------------------------------------------------------
@dataclass
class Instance:
    name: str
    role: str  # main | secondary | solo
    variant: str  # generic | cmplog | asan | compcov | cfisan
    toolchain: str
    alive: bool
    pid_alive: bool
    target_mode: str
    execs_per_sec: float
    execs_ps_last_min: float
    stability: float
    bitmap_cvg: float
    edges_found: int
    total_edges: int
    pending_favs: int
    pending_total: int
    cycles_done: int
    cycles_wo_finds: int
    time_wo_finds: int
    corpus_count: int
    corpus_variable: int
    saved_crashes: int
    saved_hangs: int
    crash_files: int
    hang_files: int
    slowest_exec_ms: int
    peak_rss_mb: int
    exec_timeout: int
    run_time_s: int
    last_update_age_s: int
    last_find_age_s: int | None
    start_time: int = 0
    afl_version: str = ""
    death: str = ""
    starting: bool = False  # running but in its dry run/calibration: no stats for this run yet
    absent_fields: list[str] = field(default_factory=list)
    status: str = ""  # filled by analyze
    issues: list[str] = field(default_factory=list)


@dataclass
class Trend:
    source: str  # plot_data | none
    window: str
    edges_found_now: int
    edges_found_peak: int
    edges_slope_recent_per_min: float
    execs_per_sec_now: float
    execs_per_sec_peak: float
    execs_per_sec_recent: float
    execs_per_sec_baseline: float
    last_cov_entry_age_s: int | None
    last_find_age_s: int | None
    expected_eps: float | None = None
    prior: dict[str, Any] | None = None  # previous stored snapshot (store)
    cross_run: dict[str, Any] | None = None  # deltas vs prior (analyze)


@dataclass
class Crashes:
    total: int
    hangs_total: int
    core_bytes: int
    by_instance: dict[str, int]


@dataclass
class CampaignSnapshot:
    campaign_id: str
    host: str
    output_dir: str
    generated_at: int
    toolchain: str
    layout: str
    honggfuzz_present: bool
    run_time_s: int
    instance_count: int
    alive_count: int
    roles: dict[str, int]
    variants: dict[str, int]
    trend: Trend
    crashes: Crashes
    instances: list[Instance]
    collection_error: str | None = None
    no_pids: bool = False
    starting_count: int = 0

    def edges_found_max(self) -> int:
        """Representative campaign edge count: the max across (synced) instances."""
        return max((i.edges_found for i in self.instances), default=0)

    def bitmap_cvg_max(self) -> float:
        """Top map coverage (%) across instances."""
        return max((i.bitmap_cvg for i in self.instances), default=0.0)

    def pending_favs_total(self) -> int:
        """Favored queue entries still awaiting fuzzing, summed across *running* instances.

        A stopped instance's backlog is frozen process state, not campaign work in flight:
        nothing will ever fuzz it, so counting it both overstates the todo list and hides
        a plateau (a drained live queue looks busy).
        """
        return sum(i.pending_favs for i in self.instances if i.alive)

    def to_dict(self) -> dict[str, Any]:
        return {"schema_version": SCHEMA_VERSION, **asdict(self)}


# --- analysis -------------------------------------------------------------------
@dataclass
class Finding:
    severity: str  # high | medium | low
    scope: str  # campaign | instance
    instance: str | None
    signal: str
    likely_cause: str


@dataclass
class ActionDirective:
    action: str
    severity: str
    scope: str
    target: dict[str, Any]
    rationale: str
    evidence: dict[str, Any]
    suggested_command: str | None = None


@dataclass
class Report:
    snapshot: CampaignSnapshot
    overall: str  # healthy|degraded|stalled|misconfigured|dead|starting|no_instances|collection_error
    verdict_explainer: str
    findings: list[Finding] = field(default_factory=list)
    actions: list[ActionDirective] = field(default_factory=list)
    notes: list[str] = field(default_factory=list)
    headline: dict[str, str] = field(default_factory=dict)  # one-word status per dimension
    degraded_reasons: list[str] = field(default_factory=list)

    def to_dict(self) -> dict[str, Any]:
        d = self.snapshot.to_dict()
        d.update(
            {
                "overall": self.overall,
                "verdict_explainer": self.verdict_explainer,
                "degraded_reasons": list(self.degraded_reasons),
                "headline": dict(self.headline),
                "findings": [asdict(f) for f in self.findings],
                "actions": [asdict(a) for a in self.actions],
                "notes": list(self.notes),
            },
        )
        return d


# ============================================================================
# parse — version-robust parsing and classification of AFL++ campaign artifacts.
#
# Pure functions only — no filesystem or network. They turn raw fuzzer_stats /
# plot_data text and command lines into normalized facts. build_snapshot()
# assembles these into a CampaignSnapshot.
# ============================================================================


def parse_fuzzer_stats(text: str) -> dict[str, str]:
    """Parse ``key : value`` lines into a dict (order-independent).

    Splits on the first colon so values may themselves contain colons. Trailing
    ``%`` is left on (callers strip it where numeric). Missing keys are simply
    absent — callers must treat absence as "unknown", never as ``0``.
    """
    kv: dict[str, str] = {}
    for line in text.splitlines():
        if ":" not in line:
            continue
        key, _, value = line.partition(":")
        kv[key.strip()] = value.strip()
    return kv


def unquote_cmdline(line: str) -> str:
    """The ``# command line:`` line of fuzzer_setup, shell-unquoted back to plain argv.

    fuzzer_setup single-quotes every argument, while fuzzer_stats' ``command_line`` is
    plain; the classifiers want the plain form. A line that does not parse is returned
    unchanged rather than dropped.
    """
    line = line.strip()
    try:
        return " ".join(shlex.split(line))
    except ValueError:
        return line


def cmplog_enabled(cmdline: str) -> bool:
    """True if CMPLOG is enabled on this afl-fuzz command line.

    ``-c <path>`` or ``-c0`` (the fuzz target doubles as the cmplog binary) enable
    it; ``-c-`` / ``-c -`` explicitly disable it; absent means off. Never inferred
    from an instance's name.
    """
    prev = ""
    for tok in cmdline.split():
        if tok == "-c-":
            return False
        if tok == "-c0":
            return True
        if tok.startswith("-c") and len(tok) > 2:  # attached value, e.g. -c/path or -c-
            return tok[2:] != "-"
        if tok == "-c":
            prev = "-c"
            continue
        if prev == "-c":
            return tok != "-"
        prev = ""
    return False


def classify_role(cmdline: str, instance_count: int) -> str:
    """Topology role: ``main`` (-M), ``secondary`` (-S), or ``solo`` (lone, no -M/-S).

    Both spellings of the flag count — ``-M main`` and the attached ``-Mmain`` — and only
    afl-fuzz's own arguments are looked at: everything from ``--`` on belongs to the target.
    """
    toks = cmdline.split()
    if "--" in toks:
        toks = toks[: toks.index("--")]
    if any(t.startswith("-M") for t in toks):
        return "main"
    if instance_count == 1 and not any(t.startswith("-S") for t in toks):
        return "solo"
    return "secondary"


def classify_variant(cmdline: str, banner: str) -> str:
    """Instance type, orthogonal to role.

    A build-instrumentation banner (asan / laf==compcov / cfisan) wins; otherwise
    ``cmplog`` if CMPLOG is enabled; otherwise ``generic``.
    """
    low = banner.lower()
    if "asan" in low:
        return "asan"
    if "laf" in low or "compcov" in low:
        return "compcov"
    if "cfisan" in low:
        return "cfisan"
    if cmplog_enabled(cmdline):
        return "cmplog"
    return "generic"


def classify_toolchain(cmdline: str, *, is_afl_subdir: bool) -> str:
    """``ziggy`` / ``cargo-afl`` / ``aflpp`` from layout + command line."""
    if is_afl_subdir or "target/afl/" in cmdline:
        return "ziggy"
    if "target/debug/" in cmdline or "target/release/" in cmdline or "cargo afl" in cmdline:
        return "cargo-afl"
    return "aflpp"


def _to_float(value: str) -> float:
    try:
        return float(value)
    except (TypeError, ValueError):
        return 0.0


EPS_MIN_SAMPLES = 10  # plot_data samples below which a self-relative rate judgment is not meaningful


def parse_plot_data(text: str) -> dict[str, float | int]:
    """Header-indexed trend from ``plot_data``.

    Tolerant of column drift: reads ``edges_found`` (falling back to ``map_size``),
    ``execs_per_sec``, ``relative_time`` by *name*. Returns peak vs now for edges
    and execs/sec plus the edge slope (edges/min) over the last 25% of the run.
    ``rows == 0`` means no usable data (trend source ``none``).

    The two execs/sec *judgment* values are medians, not extremes: ``recent`` over the
    last 25% of the run and ``baseline`` over everything collected. plot_data's
    execs_per_sec is a smoothed instantaneous rate, so single rows swing wide with the
    input being fuzzed; and a campaign's rate legitimately decays as its seeds get
    deeper, which makes the all-time peak an unreachable reference. Median vs median
    compares the campaign against its own recent norm instead. ``baseline`` is 0.0
    below ``EPS_MIN_SAMPLES`` usable rows — too little history to judge against.
    """
    cols: dict[str, int] = {}
    rows: list[list[str]] = []
    for line in text.splitlines():
        s = line.strip()
        if not s:
            continue
        if s.startswith("#"):
            names = [c.strip() for c in s.lstrip("#").split(",")]
            cols = {n: i for i, n in enumerate(names)}
            continue
        parts = [p.strip() for p in s.split(",")]
        if len(parts) >= 2:
            rows.append(parts)

    out: dict[str, float | int] = {
        "rows": len(rows),
        "edges_found_now": 0,
        "edges_found_peak": 0,
        "execs_per_sec_now": 0.0,
        "execs_per_sec_peak": 0.0,
        "execs_per_sec_recent": 0.0,
        "execs_per_sec_baseline": 0.0,
        "edges_slope_recent_per_min": 0.0,
    }
    if not rows or not cols:
        return out

    ef_idx = cols.get("edges_found", cols.get("map_size"))
    eps_idx = cols.get("execs_per_sec")
    rt_idx = cols.get("relative_time")

    def col(row: list[str], idx: int | None) -> float:
        return _to_float(row[idx]) if idx is not None and idx < len(row) else 0.0

    efs = [col(r, ef_idx) for r in rows]
    epss = [col(r, eps_idx) for r in rows]
    ts = [col(r, rt_idx) for r in rows] if rt_idx is not None else [float(i) for i in range(len(rows))]

    out["edges_found_now"] = int(efs[-1])
    out["edges_found_peak"] = int(max(efs))
    out["execs_per_sec_now"] = epss[-1]
    out["execs_per_sec_peak"] = max(epss)

    tspan = ts[-1] - ts[0]
    wstart = ts[-1] - 0.25 * tspan
    wi = len(ts) - 1
    for i in range(len(ts) - 1, -1, -1):
        if ts[i] >= wstart:
            wi = i
    dt = ts[-1] - ts[wi]
    de = efs[-1] - efs[wi]
    out["edges_slope_recent_per_min"] = (de / (dt / 60.0)) if dt > 0 else 0.0

    live = [e for e in epss if e > 0]
    recent = [e for e in epss[wi:] if e > 0]
    out["execs_per_sec_recent"] = median(recent) if recent else 0.0
    out["execs_per_sec_baseline"] = median(live) if len(live) >= EPS_MIN_SAMPLES else 0.0
    return out


# ============================================================================
# snapshot — assemble a normalized CampaignSnapshot from raw collector artifacts.
# ============================================================================

STALE_SECONDS = 180  # since last_update within which an instance counts as freshly reporting
NO_PIDS_STALE_SECONDS = 128  # --no-pids: since last_update within which an instance counts as alive

# Fields newer AFL++ builds may omit; tracked as "absent" so callers never read 0.
_WATCHED_OPTIONAL = ("slowest_exec_ms", "peak_rss_mb")

FASTRESUME_MIN_VERSION = (4, 22)
_VERSION_RE = re.compile(r"(\d+)\.(\d+)")


def supports_fastresume(afl_version: str) -> bool:
    """True if this afl-fuzz build writes ``fastresume.bin`` on a clean exit.

    afl-fuzz writes it only from ``stop_fuzzing()`` — the graceful shutdown path — and
    unlinks it again the moment it resumes from it. So on a *non-running* instance the
    file is a positive marker of a clean shutdown. It exists since AFL++ 4.22a; older
    builds leave no marker at all, which makes absence meaningless.
    """
    m = _VERSION_RE.search(afl_version or "")
    if not m:
        return False
    return (int(m.group(1)), int(m.group(2))) >= FASTRESUME_MIN_VERSION


def classify_death(inst: Instance, raw: RawInstance, *, generation_start: int | None, last_update: int) -> str:
    """Why a non-running instance is not running.

    ``leftover``  started *and* stopped before the currently-running instances began —
                  a previous campaign generation that simply was not restarted.
    ``stopped``   left a clean-exit marker: a user stop, or a ``-V``/``-E`` limit reached.
    ``killed``    died mid-campaign with no such marker: OOM-killed, SIGKILL, a crash of
                  afl-fuzz itself, a FATAL, or the machine going down.
    ``unknown``   no marker is available (AFL++ < 4.22a, or AFL_NO_FASTRESUME suppressed
                  it), so a clean exit cannot be ruled out either way.
    ``unreadable``
                  fuzzer_stats itself could not be read, so nothing about this instance,
                  liveness included, can be determined.

    A sixth class, ``startup``, is not decided here: it is assigned in build_snapshot to a
    dir that never got a fuzzer_stats at all and has no live afl-fuzz behind it.

    ``generation_start`` is the start_time of the earliest *freshly reporting* instance:
    liveness alone cannot define it, since an instance that was stopped moments ago still
    belongs to the generation its siblings are running in.

    Only ``killed`` and ``unknown`` reflect badly on the campaign; ``unreadable`` reflects
    on the collection, not on the campaign.
    """
    if not raw.fuzzer_stats:
        return "unreadable"
    if generation_start is not None and 0 < inst.start_time < generation_start and last_update < generation_start:
        return "leftover"
    if raw.fastresume_present:
        return "stopped"
    if raw.no_fastresume_env or not supports_fastresume(inst.afl_version):
        return "killed" if inst.role == "main" and raw.main_node_file else "unknown"
    return "killed"


def _fnum(kv: dict[str, str], key: str) -> tuple[float, bool]:
    """(value, absent). Strips a trailing ``%``; non-numeric/missing -> (0.0, True)."""
    raw = kv.get(key)
    if raw is None:
        return 0.0, True
    try:
        return float(raw.rstrip("%")), False
    except ValueError:
        return 0.0, True


def _inum(kv: dict[str, str], key: str) -> int:
    return int(_fnum(kv, key)[0])


def build_snapshot(raw: RawCampaign, *, no_pids: bool = False) -> CampaignSnapshot:
    n = len(raw.instances)
    is_afl_subdir = raw.layout == "ziggy-afl-subdir"

    instances: list[Instance] = []
    roles: dict[str, int] = {}
    variants: dict[str, int] = {}
    by_instance: dict[str, int] = {}
    crash_total = hang_total = core_bytes = 0
    run_time_max = alive_count = 0
    tc_ziggy = tc_cargo = False

    for ri in raw.instances:
        kv = parse_fuzzer_stats(ri.fuzzer_stats)
        cl = kv.get("command_line", "") or unquote_cmdline(ri.setup_cmdline)
        banner = kv.get("afl_banner", "")
        role = classify_role(cl, n)
        variant = classify_variant(cl, banner)
        tc = classify_toolchain(cl, is_afl_subdir=is_afl_subdir)
        tc_ziggy = tc_ziggy or tc == "ziggy"
        tc_cargo = tc_cargo or tc == "cargo-afl"

        # A starting instance has published no numbers for this run: whatever fuzzer_stats
        # holds is the previous run's, so metrics come from an empty source and only the
        # identity fields (command line, banner, version) are taken from the file.
        mv = {} if ri.starting else kv

        run_time = _inum(mv, "run_time")
        last_update = _inum(mv, "last_update")
        lu_age = max(0, raw.now - last_update)
        last_find = _inum(mv, "last_find")
        lf_age = max(0, raw.now - last_find) if last_find > 0 else None
        alive = lu_age < NO_PIDS_STALE_SECONDS if no_pids else ri.pid_alive
        if ri.starting:
            # The setup file's age is the only clock there is, and only a live process can
            # tell "starting up" from "aborted before it ever wrote stats".
            run_time = lu_age = max(0, int(raw.now - ri.setup_mtime)) if ri.setup_mtime else 0
            lf_age = None
            alive = ri.pid_alive
        else:
            run_time_max = max(run_time_max, run_time)
        if alive:
            alive_count += 1

        edges_found = _inum(mv, "edges_found")

        crash_total += ri.crash_count
        hang_total += ri.hang_count
        core_bytes += ri.core_bytes
        if ri.crash_count:
            by_instance[ri.name] = ri.crash_count

        roles[role] = roles.get(role, 0) + 1
        variants[variant] = variants.get(variant, 0) + 1
        absent = [f for f in _WATCHED_OPTIONAL if f not in kv] if kv else []

        instances.append(
            Instance(
                name=ri.name,
                role=role,
                variant=variant,
                toolchain=tc,
                alive=alive,
                pid_alive=ri.pid_alive,
                target_mode=kv.get("target_mode", ""),
                execs_per_sec=_fnum(mv, "execs_per_sec")[0],
                execs_ps_last_min=_fnum(mv, "execs_ps_last_min")[0],
                stability=_fnum(mv, "stability")[0],
                bitmap_cvg=_fnum(mv, "bitmap_cvg")[0],
                edges_found=edges_found,
                total_edges=_inum(mv, "total_edges"),
                pending_favs=_inum(mv, "pending_favs"),
                pending_total=_inum(mv, "pending_total"),
                cycles_done=_inum(mv, "cycles_done"),
                cycles_wo_finds=_inum(mv, "cycles_wo_finds"),
                time_wo_finds=_inum(mv, "time_wo_finds"),
                corpus_count=_inum(mv, "corpus_count"),
                corpus_variable=_inum(mv, "corpus_variable"),
                saved_crashes=_inum(mv, "saved_crashes"),
                saved_hangs=_inum(mv, "saved_hangs"),
                crash_files=ri.crash_count,
                hang_files=ri.hang_count,
                slowest_exec_ms=_inum(mv, "slowest_exec_ms"),
                peak_rss_mb=_inum(mv, "peak_rss_mb"),
                exec_timeout=_inum(mv, "exec_timeout"),
                run_time_s=run_time,
                last_update_age_s=lu_age,
                last_find_age_s=lf_age,
                start_time=_inum(mv, "start_time"),
                afl_version=kv.get("afl_version", ""),
                starting=ri.starting,
                absent_fields=absent,
            ),
        )

    # A starting instance has no start_time yet, so its fuzzer_setup stands in for one.
    fresh_starts = [i.start_time for i in instances if i.last_update_age_s < STALE_SECONDS and i.start_time > 0]
    fresh_starts += [
        int(ri.setup_mtime) for i, ri in zip(instances, raw.instances) if i.starting and i.alive and ri.setup_mtime
    ]
    generation_start = min(fresh_starts, default=None)
    for i, ri in zip(instances, raw.instances):
        if i.alive:
            continue
        if ri.starting:
            setup_ts = int(ri.setup_mtime or 0)
            older = generation_start is not None and 0 < setup_ts < generation_start
            i.death = "leftover" if older else "startup"
        else:
            i.death = classify_death(
                i,
                ri,
                generation_start=generation_start,
                last_update=raw.now - i.last_update_age_s,
            )

    toolchain = "ziggy" if tc_ziggy else "cargo-afl" if tc_cargo else "aflpp"

    # Campaign-level trend comes from the running main's queue (it syncs everything), else
    # from the newest/leading running instance — never from one that is no longer running,
    # and never from one that is still starting up (its plot_data holds only the header).
    pairs = [p for p in zip(instances, raw.instances) if not p[0].starting]
    running = [p for p in pairs if p[0].alive] or pairs
    main = next((p for p in running if p[0].role == "main"), None)

    def newest(attr: str) -> float | None:
        return max((getattr(ri, attr) for _, ri in running if getattr(ri, attr) is not None), default=None)

    cov_mtime = (main[1].newest_cov_mtime if main else None) or newest("newest_cov_mtime")
    queue_mtime = (main[1].newest_queue_mtime if main else None) or newest("newest_queue_mtime")

    def age(ts: float | None) -> int | None:
        return max(0, int(raw.now - ts)) if ts is not None else None

    last_cov_age = age(cov_mtime)
    stats_find = max((raw.now - i.last_find_age_s for i, _ in running if i.last_find_age_s is not None), default=None)
    last_find_age = age(max([t for t in (stats_find, queue_mtime) if t is not None], default=None))

    primary_plot = main[1].plot_data if main else None
    if primary_plot is None:
        best = max(running, key=lambda p: p[0].edges_found, default=None)
        primary_plot = best[1].plot_data if best else None

    pt = parse_plot_data(primary_plot or "")
    trend = Trend(
        source="plot_data" if pt["rows"] else "none",
        window="last 25% of run",
        edges_found_now=int(pt["edges_found_now"]),
        edges_found_peak=int(pt["edges_found_peak"]),
        edges_slope_recent_per_min=float(pt["edges_slope_recent_per_min"]),
        execs_per_sec_now=float(pt["execs_per_sec_now"]),
        execs_per_sec_peak=float(pt["execs_per_sec_peak"]),
        execs_per_sec_recent=float(pt["execs_per_sec_recent"]),
        execs_per_sec_baseline=float(pt["execs_per_sec_baseline"]),
        last_cov_entry_age_s=last_cov_age,
        last_find_age_s=last_find_age,
    )

    return CampaignSnapshot(
        campaign_id=raw.campaign_id,
        host=raw.host,
        output_dir=raw.output_dir,
        generated_at=raw.now,
        toolchain=toolchain,
        layout=raw.layout,
        honggfuzz_present=raw.honggfuzz_present,
        run_time_s=run_time_max,
        instance_count=n,
        alive_count=alive_count,
        roles=roles,
        variants=variants,
        trend=trend,
        crashes=Crashes(crash_total, hang_total, core_bytes, by_instance),
        instances=instances,
        collection_error=raw.collection_error,
        no_pids=no_pids,
        starting_count=sum(1 for i in instances if i.starting),
    )


# ============================================================================
# analyze — snapshot (+ optional history) -> verdict, findings, directives.
#
# All rate/size judgments are self-relative (vs the campaign's own plot_data norm,
# prior snapshots, or an optional expected eps). The only absolute judgments are
# objective/degenerate (execs==0, edges==0, dead) and AFL-normalized (stability
# bands, bitmap saturation). Read-only: directives are emitted, never executed.
# ============================================================================

STABILITY_BAD = 85.0
STABILITY_WARN = 90.0
FLAT_SLOPE = 0.5  # edges/min at/under which growth is too slow to rule out a plateau
SATURATION_CVG = 99.5
CORE_BYTES_WARN = 2 * 1024**3  # 2 GB of .core dumps
THROUGHPUT_RUNTIME = 6 * 3600  # only call a campaign "throughput-bound" after this long
# A campaign's exec rate falls off naturally as its seeds reach deeper, so only a
# sustained collapse below a quarter of its own recent norm (median vs median, see
# parse_plot_data) counts as a slowdown - not the slow drift away from an early peak.
EPS_DROP_FRACTION = 0.25

DEGRADED_REASON_TEXT = {
    "dead-instances": "instance(s) died unexpectedly",
    "low-stability": "low stability",
    "exec-slowdown": "dropped exec rate",
}

DEATH_UNEXPECTED = ("killed", "unknown", "startup", "")
DEATH_ORDER = ("killed", "unknown", "startup", "stopped", "leftover")

DEATH_CAUSE = {
    "unreadable": "fuzzer_stats could not be read: permissions (campaign owned by another user) "
    "or the file went away mid-collection",
    "killed": "OOM-killed, SIGKILL, a crash of afl-fuzz itself, a FATAL, or the machine went down",
    "unknown": "no clean-exit marker available (AFL++ < 4.22a or AFL_NO_FASTRESUME) — cause undeterminable",
    "stopped": "shut down cleanly: stopped by the user, or a -V/-E limit was reached",
    "leftover": "ran before the current instances started: a previous campaign generation, not restarted",
    "startup": "afl-fuzz gave up before finishing its initial dry run: a FATAL on a seed, an OOM-killed "
    "or timed-out dry run, or the launcher stopped it again right away",
    "": "crashed or OOM-killed, or the campaign was stopped",
}

DEATH_ISSUE = {
    "unreadable": "fuzzer_stats unreadable; state undeterminable",
    "killed": "died unexpectedly (no clean-exit marker)",
    "unknown": "not running; clean exit could not be confirmed",
    "stopped": "stopped cleanly (user stop or -V/-E limit)",
    "leftover": "leftover from an earlier campaign generation",
    "startup": "aborted during startup (never wrote fuzzer_stats)",
}


def dead_breakdown(snap: CampaignSnapshot) -> str:
    """``2 killed`` — instances that died on us. Empty unless some actually did.

    Instances that stopped cleanly or are leftovers of an earlier generation are
    deliberately not listed: they are expected states, reported in the notes instead.
    """
    by_class = Counter(i.death for i in snap.instances if not i.alive and i.death in DEATH_UNEXPECTED and i.death)
    return ", ".join(f"{by_class[c]} {c}" for c in DEATH_ORDER if by_class[c])


def _target(snap: CampaignSnapshot, instance: str | None = None) -> dict[str, Any]:
    t = {"host": snap.host, "output_dir": snap.output_dir}
    if instance:
        t["instance"] = instance
    return t


def analyze(
    snap: CampaignSnapshot,
    *,
    expected_eps: float | None = None,
    prior: dict[str, Any] | None = None,
) -> Report:
    if snap.collection_error:
        return Report(
            snap, "collection_error", snap.collection_error, notes=[f"collection failed: {snap.collection_error}"]
        )

    if snap.instance_count == 0:
        explainer = (
            f"No fuzzer_stats or fuzzer_setup found under {snap.output_dir} (searched 3 levels deep) — not "
            "an AFL++/cargo-afl/ziggy output directory, a campaign that never got as far as creating its "
            "instance dirs, or files that this user may not read. Nothing was measured, so nothing is "
            "claimed about health."
        )
        if snap.honggfuzz_present:
            explainer += " A ziggy honggfuzz instance is present but is not interpreted (out of scope)."
        return Report(snap, "no_instances", explainer, notes=[explainer])

    starting = [i for i in snap.instances if i.starting]
    live_starting = [i for i in starting if i.alive]
    if len(starting) == snap.instance_count and live_starting:
        # Every instance is still in its initial dry run: there is no fuzzer_stats to read,
        # so every metric would be a fake zero. Report the state, judge nothing.
        age = max((i.run_time_s for i in live_starting), default=0)
        explainer = (
            f"{len(live_starting)} of {snap.instance_count} instance"
            f"{'' if snap.instance_count == 1 else 's'} still starting up, no stats for this run yet — "
            f"afl-fuzz writes its first stats only once the initial dry run and calibration are "
            f"through, which on a large corpus or a slow target takes a while (launched {_hm(age)} ago). "
            f"Nothing was measured, so nothing is claimed about health."
        )
        notes = [explainer]
        if len(live_starting) != snap.instance_count:
            notes.append(
                f"{snap.instance_count - len(live_starting)} of them have a startup directory but no "
                f"live afl-fuzz process: {DEATH_CAUSE['startup']}."
            )
        return Report(snap, "starting", explainer, notes=notes)

    findings: list[Finding] = []
    actions: list[ActionDirective] = []
    notes: list[str] = []
    snap.trend.expected_eps = expected_eps
    snap.trend.prior = prior

    n = snap.instance_count
    reporting = [i for i in snap.instances if not i.starting]
    alive = [i for i in snap.instances if i.alive]
    dead = [i for i in snap.instances if not i.alive]
    died = [i for i in dead if i.death in DEATH_UNEXPECTED]
    idle = [i for i in dead if i.death not in DEATH_UNEXPECTED and i.death != "unreadable"]
    t = snap.trend
    pending_favs_total = snap.pending_favs_total()
    cycles_max = max((i.cycles_done for i in snap.instances), default=0)

    # --- cross-run delta (from a prior stored snapshot) -------------------------
    if prior:
        prev_edges = prior.get("edges_found")
        t.cross_run = {
            "since_ts": prior.get("generated_at"),
            "edges_delta": (snap.edges_found_max() - prev_edges) if isinstance(prev_edges, (int, float)) else None,
        }

    # --- crashes ----------------------------------------------------------------
    if snap.crashes.total > 0:
        findings.append(
            Finding(
                "high",
                "campaign",
                None,
                f"{snap.crashes.total} crash testcases saved",
                "the target crashed on saved inputs — real findings to triage",
            )
        )
        actions.append(
            ActionDirective(
                "triage_crashes",
                "high",
                "campaign",
                _target(snap),
                f"{snap.crashes.total} crashes ({snap.crashes.hangs_total} hangs) await triage",
                {"crashes": snap.crashes.total, "by_instance": snap.crashes.by_instance},
                suggested_command=f"casr-afl -i {snap.output_dir} -o {snap.output_dir}-triaged",
            )
        )

    if snap.crashes.core_bytes > CORE_BYTES_WARN:
        gb = snap.crashes.core_bytes / 1024**3
        findings.append(
            Finding(
                "low",
                "campaign",
                None,
                f"{gb:.0f} GB of .core dumps next to crashes",
                "the crash handler is writing core dumps; disk will fill",
            )
        )
        actions.append(
            ActionDirective(
                "prune_cores",
                "medium",
                "campaign",
                _target(snap),
                f"{gb:.0f} GB of .core files and growing",
                {"core_bytes": snap.crashes.core_bytes},
                suggested_command="disable core dumps (ulimit -c 0) or prune */crashes/*.core",
            )
        )

    # --- objective / AFL-normalized: misconfigured ------------------------------
    all_zero_edges = bool(reporting) and snap.run_time_s > 0 and all(i.edges_found == 0 for i in reporting)
    all_zero_eps = bool(reporting) and snap.run_time_s > 0 and all(i.execs_per_sec == 0 for i in reporting)
    saturated = [i for i in snap.instances if i.bitmap_cvg >= SATURATION_CVG]
    if all_zero_edges:
        findings.append(
            Finding(
                "high",
                "campaign",
                None,
                f"edges_found == 0 after run_time = {snap.run_time_s}s",
                "instrumentation didn't take, or afl-fuzz is on the wrong binary",
            )
        )
    if all_zero_eps:
        findings.append(
            Finding("high", "campaign", None, "execs_per_sec == 0 everywhere", "campaign not actually executing")
        )
    if saturated:
        findings.append(
            Finding(
                "high",
                "campaign",
                None,
                f"bitmap_cvg ~100% on {len(saturated)} instance(s)",
                "map saturation: hash collisions degrade the feedback signal",
            )
        )
    misconfigured = bool(all_zero_edges or all_zero_eps or saturated)

    # --- structural: no main on a multi-instance run ----------------------------
    if n > 1 and snap.roles.get("main", 0) == 0:
        findings.append(
            Finding(
                "medium",
                "campaign",
                None,
                "no -M main instance",
                "no sync hub / final-sync; cross-pollination is weaker",
            )
        )

    # --- per-instance: stability, liveness, eps drop ----------------------------
    low_stability = False
    eps_drop = False
    for i in snap.instances:
        issues: list[str] = []
        status = "healthy"
        if not i.alive:
            status = i.death or "dead"
            issues.append(DEATH_ISSUE.get(i.death, "not running"))
            if i.death == "unreadable":
                findings.append(
                    Finding(
                        "medium",
                        "instance",
                        i.name,
                        "fuzzer_stats could not be read",
                        DEATH_CAUSE["unreadable"],
                    )
                )
                actions.append(
                    ActionDirective(
                        "fix_stats_access",
                        "medium",
                        "instance",
                        _target(snap, i.name),
                        "fuzzer_stats is unreadable, so this instance cannot be assessed at all",
                        {"death": i.death},
                        suggested_command="check access to "
                        f"{snap.output_dir}{'/afl' if snap.layout == 'ziggy-afl-subdir' else ''}"
                        f"/{i.name}/fuzzer_stats",
                    )
                )
            elif i.death in DEATH_UNEXPECTED:
                findings.append(
                    Finding(
                        "medium" if alive else "high",
                        "instance",
                        i.name,
                        f"instance {DEATH_ISSUE.get(i.death, 'not running')}",
                        DEATH_CAUSE[i.death],
                    )
                )
                actions.append(
                    ActionDirective(
                        "relaunch_instance",
                        "medium",
                        "instance",
                        _target(snap, i.name),
                        "instance is not running while others are" if alive else "instance not running",
                        {"last_update_age_s": i.last_update_age_s, "death": i.death},
                        suggested_command="re-run its launch line with AFL_AUTORESUME=1",
                    )
                )
            else:
                findings.append(
                    Finding(
                        "low",
                        "instance",
                        i.name,
                        f"instance {DEATH_ISSUE[i.death]}",
                        DEATH_CAUSE[i.death],
                    )
                )
        elif i.starting:
            status = "starting"
            issues.append(f"starting up for {_hm(i.run_time_s)}, no stats for this run yet")
        if i.stability and i.stability < STABILITY_BAD:
            low_stability = True
            if i.alive:
                status = "unstable"
            issues.append(f"stability {i.stability:.0f}% (<{STABILITY_BAD:.0f}%)")
            findings.append(
                Finding(
                    "high",
                    "instance",
                    i.name,
                    f"stability {i.stability:.0f}% (<{STABILITY_BAD:.0f}%)",
                    "nondeterministic edges: uninit memory / hashmap order / "
                    "threads / persistent-loop state (Rust: HashMap RandomState)",
                )
            )
            actions.append(
                ActionDirective(
                    "investigate_instability",
                    "high",
                    "instance",
                    _target(snap, i.name),
                    f"stability {i.stability:.0f}% is below the 85% floor",
                    {"stability": i.stability, "variant": i.variant},
                    suggested_command="diagnose unstable edges; consider moving off ASAN",
                )
            )
        elif i.stability and i.stability < STABILITY_WARN:
            issues.append(f"stability {i.stability:.0f}% (<{STABILITY_WARN:.0f}%)")
            findings.append(
                Finding(
                    "low",
                    "instance",
                    i.name,
                    f"stability {i.stability:.0f}% (<{STABILITY_WARN:.0f}%)",
                    "mild nondeterminism",
                )
            )
        i.status = status
        i.issues = issues

    eps_reference = expected_eps or t.execs_per_sec_baseline
    if eps_reference > 0 and 0 < t.execs_per_sec_recent < EPS_DROP_FRACTION * eps_reference:
        eps_drop = True
        against = "the expected rate" if expected_eps else "its own recent norm"
        findings.append(
            Finding(
                "medium",
                "campaign",
                None,
                f"execs/sec {t.execs_per_sec_recent:.0f} over the last 25% of the run is far "
                f"below {against} ({eps_reference:.0f})",
                "slow inputs accumulating, sync overhead, or a slow path",
            )
        )
        actions.append(
            ActionDirective(
                "tune",
                "medium",
                "campaign",
                _target(snap),
                f"recent exec rate dropped to under a quarter of {against}",
                {
                    "eps_recent": t.execs_per_sec_recent,
                    "eps_reference": eps_reference,
                    "eps_peak": t.execs_per_sec_peak,
                },
                suggested_command="verify persistent mode; afl-persistent-config + afl-system-config; raise AFL_TESTCACHE_SIZE",
            )
        )

    # --- throughput-bound vs plateau (need a trend, not a snapshot) -------------
    cov_old = t.last_cov_entry_age_s is not None and t.last_cov_entry_age_s >= max(1800, snap.run_time_s // 4)
    flat = t.source == "plot_data" and t.edges_slope_recent_per_min <= FLAT_SLOPE
    # favored queue essentially exhausted (only instances that report a queue can drain one)
    drained = pending_favs_total <= max(1, sum(1 for i in reporting if i.alive))
    throughput_bound = cycles_max == 0 and snap.run_time_s >= THROUGHPUT_RUNTIME and pending_favs_total > 0
    plateau = flat and cov_old and drained and not throughput_bound

    if throughput_bound:
        notes.append(
            "throughput-bound, not stalled: 0 completed cycles after a long run with a "
            "large pending-favorites backlog — coverage is gated by exec speed, not corpus."
        )
        findings.append(
            Finding(
                "medium",
                "campaign",
                None,
                f"cycles_done = 0 after {snap.run_time_s // 3600}h; {pending_favs_total} favorites pending",
                "heavy per-exec target caps coverage growth (throughput-bound)",
            )
        )
        actions.append(
            ActionDirective(
                "tune",
                "medium",
                "campaign",
                _target(snap),
                "growth is limited by exec speed, not by corpus",
                {"cycles_done": cycles_max, "pending_favs": pending_favs_total, "eps_now": t.execs_per_sec_now},
                suggested_command="profile the harness/target per-exec cost; faster execs > more seeds",
            )
        )

    if plateau:
        findings.append(
            Finding(
                "high",
                "campaign",
                None,
                f"edge slope ~{t.edges_slope_recent_per_min:.1f}/min, last +cov "
                f"{t.last_cov_entry_age_s}s ago, favorites drained",
                "genuine coverage plateau (bitmap not saturated → input/harness limited)",
            )
        )
        actions.append(
            ActionDirective(
                "inject_seeds",
                "high",
                "campaign",
                _target(snap),
                "coverage plateaued with the favored queue exhausted",
                {"slope_per_min": t.edges_slope_recent_per_min, "last_cov_age_s": t.last_cov_entry_age_s},
                suggested_command="perform a seed corpus and dictionary refresh: build/inject richer seeds + a dictionary",
            )
        )
        actions.append(
            ActionDirective(
                "run_coverage_analysis",
                "medium",
                "campaign",
                _target(snap),
                "find which reachable code is still uncovered",
                {},
                suggested_command="perform a coverage analysis with cov-analysis",
            )
        )

    if (plateau or throughput_bound) and snap.variants.get("cmplog", 0) == 0:
        actions.append(
            ActionDirective(
                "add_cmplog",
                "medium",
                "campaign",
                _target(snap),
                "no cmplog instance present; comparison-gated edges may be blocking progress",
                {"variants": snap.variants},
                suggested_command="add a -c <cmplog_bin> or -c0 instance (AFL_CMPLOG_ONLY_NEW=1 on restart)",
            )
        )

    if live_starting:
        notes.append(
            f"{len(live_starting)} of {snap.instance_count} instance(s) are still starting up "
            f"(oldest {_hm(max(i.run_time_s for i in live_starting))}, no stats for this run yet) — "
            "they contribute nothing to the measurements above."
        )

    if idle:
        by_class = Counter(i.death for i in idle)
        notes.append(
            f"{len(idle)} of {snap.instance_count} instance(s) are not running but did not die: "
            + "; ".join(f"{n}x {cls} ({DEATH_CAUSE[cls]})" for cls, n in by_class.items())
            + " — not counted against the verdict."
        )

    if snap.honggfuzz_present:
        notes.append(
            "ziggy honggfuzz instance present but not interpreted (out of scope); "
            "perform a crash triage (casr-afl) on the shared crashes/ dir for honggfuzz findings."
        )
    absent = sorted({f for i in snap.instances for f in i.absent_fields})
    if absent:
        notes.append(
            f"AFL build omits {', '.join(absent)} from fuzzer_stats — reported as 0 but "
            f"unknown; do not interpret as real zeros."
        )

    # --- verdict ----------------------------------------------------------------
    degraded_reasons: list[str] = []
    if died:
        degraded_reasons.append("dead-instances")
    if low_stability:
        degraded_reasons.append("low-stability")
    if eps_drop and not throughput_bound:
        degraded_reasons.append("exec-slowdown")

    if misconfigured:
        overall = "misconfigured"
    elif n > 0 and snap.alive_count == 0:
        overall = "dead"
    elif plateau:
        overall = "stalled"
    elif degraded_reasons:
        overall = "degraded"
    else:
        overall = "healthy"
    if overall not in ("degraded", "stalled"):
        degraded_reasons = []

    headline = _headline(
        snap,
        low_stability=low_stability,
        eps_drop=eps_drop,
        throughput_bound=throughput_bound,
        plateau=plateau,
        all_zero_edges=all_zero_edges,
        all_zero_eps=all_zero_eps,
        saturated=bool(saturated),
        cov_recent=t.last_cov_entry_age_s is not None and not cov_old,
    )
    explainer = _explain(overall, snap, throughput_bound=throughput_bound, degraded_reasons=degraded_reasons)
    return Report(
        snap,
        overall,
        explainer,
        findings=findings,
        actions=actions,
        notes=notes,
        headline=headline,
        degraded_reasons=degraded_reasons,
    )


def _headline(
    snap: CampaignSnapshot,
    *,
    low_stability: bool,
    eps_drop: bool,
    throughput_bound: bool,
    plateau: bool,
    all_zero_edges: bool,
    all_zero_eps: bool,
    saturated: bool,
    cov_recent: bool,
) -> dict[str, str]:
    """One-word status per dimension for the header block.

    ``coverage`` states the recent *direction*, not a rate: plot_data's edges_found only
    accumulates, so any positive slope means new edges landed inside the window. ``flat``
    is therefore reserved for windows with no new edge at all and no fresh +cov entry.
    """
    stab_vals = [i.stability for i in snap.instances if i.stability > 0]
    stab_min = min(stab_vals) if stab_vals else None
    if low_stability or (stab_min is not None and stab_min < STABILITY_BAD):
        stability = "low"
    elif stab_min is not None and stab_min < STABILITY_WARN:
        stability = "marginal"
    else:
        stability = "OK"

    if all_zero_eps:
        speed = "none"
    elif throughput_bound:
        speed = "slow"
    elif eps_drop:
        speed = "dropping"
    else:
        speed = "OK"

    if all_zero_edges:
        coverage = "none"
    elif saturated:
        coverage = "saturated"
    elif plateau:
        coverage = "plateaued"
    elif throughput_bound or snap.trend.edges_slope_recent_per_min > 0 or cov_recent:
        coverage = "gaining coverage"
    else:
        coverage = "flat"

    return {"stability": stability, "speed": speed, "coverage": coverage}


def degraded_tag(reasons: list[str]) -> str:
    """The single-word degradation reason tags, joined for one-line output."""
    return "+".join(reasons)


def _explain(
    overall: str,
    snap: CampaignSnapshot,
    *,
    throughput_bound: bool,
    degraded_reasons: list[str] | None = None,
) -> str:
    t = snap.trend
    breakdown = dead_breakdown(snap)
    alive = f"{snap.alive_count}/{snap.instance_count} instances alive" + (f" ({breakdown})" if breakdown else "")
    if overall == "misconfigured":
        return f"Setup failure: {alive}, but coverage/feedback is not working (see findings)."
    if overall == "dead":
        clean = all(i.death in ("stopped", "leftover") for i in snap.instances if not i.alive)
        detail = f" ({breakdown})" if breakdown else ""
        return (
            f"No instance is running, {snap.instance_count} total{detail} — "
            f"campaign {'was stopped cleanly' if clean else 'died or was stopped'}."
        )
    if overall == "degraded":
        why = ", ".join(DEGRADED_REASON_TEXT.get(r, r) for r in degraded_reasons or []) or (
            "dead instance, low stability, or dropped exec rate"
        )
        return f"{alive}, but a subset is unwell ({why})."
    if overall == "stalled":
        why = ", ".join(DEGRADED_REASON_TEXT.get(r, r) for r in degraded_reasons or [])
        return (
            f"{alive} and running, but coverage plateaued: edge slope "
            f"~{t.edges_slope_recent_per_min:.1f}/min, last +cov {t.last_cov_entry_age_s}s ago, "
            f"favored queue drained." + (f" A subset is also unwell ({why})." if why else "")
        )
    tail = " Throughput-bound: still gaining coverage, just slowly (exec-speed limited)." if throughput_bound else ""
    return f"{alive}; coverage progressing and stability OK.{tail}"


# ============================================================================
# report — machine JSON (single or fleet envelope) and a human text report.
# ============================================================================

# Process exit codes for cron/CI.
EXIT = {
    "healthy": 0,
    "degraded": 1,
    "stalled": 1,
    "misconfigured": 2,
    "dead": 2,
    "starting": 0,
    "no_instances": 3,
    "collection_error": 3,
}

# Verdicts where no instance could be measured: every metric would be a fake zero, so
# these report the reason instead of a header block, a table and a trend.
NOTHING_MEASURED = {
    "starting": "STARTING",
    "no_instances": "NO INSTANCES",
    "collection_error": "COLLECTION ERROR",
}


def exit_code(reports: list[Report]) -> int:
    return max((EXIT.get(r.overall, 0) for r in reports), default=0)


def fleet_verdict(reports: list[Report]) -> str:
    """The worst verdict across the fleet (by exit-code rank).

    Ties are broken towards the more informative state — ``stalled`` over ``degraded``,
    ``starting`` over ``healthy`` — so the roll-up does not depend on argument order.
    """
    if not reports:
        return "healthy"
    return max(reports, key=lambda r: (EXIT.get(r.overall, 0), r.overall in ("stalled", "starting"))).overall


def render_json(reports: list[Report], *, now: int, version: str) -> str:
    if len(reports) == 1:
        return json.dumps(reports[0].to_dict(), indent=2)
    envelope = {
        "tool": "afl-health",
        "version": version,
        "generated_at": now,
        "campaigns": [r.to_dict() for r in reports],
        "fleet": {
            "verdict": fleet_verdict(reports),
            "counts": dict(Counter(r.overall for r in reports)),
        },
    }
    return json.dumps(envelope, indent=2)


def _dead_suffix(snap: CampaignSnapshot) -> str:
    breakdown = dead_breakdown(snap)
    return f" [{breakdown}]" if breakdown else ""


def _hm(seconds: int) -> str:
    h, m = divmod(max(0, seconds) // 60, 60)
    return f"{h}h{m:02d}m"


def _toolchain_label(toolchain: str) -> str:
    return "AFL++" if toolchain == "aflpp" else toolchain


_ACTION_NUM = re.compile(r"\d+(?:\.\d+)?%?")
_SEV_RANK = {"high": 0, "medium": 1, "low": 2}


def _group_actions(actions: list[ActionDirective]) -> list[tuple[str, str, str, list[ActionDirective]]]:
    """Collapse directives that differ only in embedded numbers into one group.

    Keyed by (severity, action, rationale-with-numbers-blanked), insertion order preserved.
    The same finding across many instances becomes one group whose members are the affected
    instances, so the finding is stated once instead of once per instance.
    """
    groups: dict[tuple[str, str, str], list[ActionDirective]] = {}
    order: list[tuple[str, str, str]] = []
    for a in actions:
        key = (a.severity, a.action, _ACTION_NUM.sub("\x00", a.rationale))
        if key not in groups:
            groups[key] = []
            order.append(key)
        groups[key].append(a)
    return [(k[0], k[1], k[2], groups[k]) for k in order]


def _action_group_view(template: str, members: list[ActionDirective]) -> tuple[str, list[str]]:
    """Split a group into a shared title and per-member labels.

    Numeric tokens shared by every member are restored into the title; tokens that vary
    are elided from the title and shown per member (as ``instance (values)``), so the
    common finding reads once and each instance keeps its own distinguishing values.
    """
    nums = [_ACTION_NUM.findall(m.rationale) for m in members]
    slots = template.count("\x00")
    varying = [len({n[j] for n in nums}) > 1 for j in range(slots)]
    segs = template.split("\x00")
    title = segs[0]
    for j in range(slots):
        title += ("" if varying[j] else nums[0][j]) + segs[j + 1]
    title = re.sub(r"\s{2,}", " ", title).strip()
    labels: list[str] = []
    for m, ns in zip(members, nums):
        detail = " ".join(ns[j] for j in range(slots) if varying[j])
        name = m.target.get("instance") or "campaign"
        labels.append(f"{name} ({detail})" if detail else name)
    return title, labels


def _wrap_labels(labels: list[str], indent: str, width: int = 92) -> list[str]:
    """Comma-join labels into indented lines, wrapping at ``width`` without splitting a label."""
    out: list[str] = []
    cur = ""
    for idx, label in enumerate(labels):
        piece = label + ("," if idx < len(labels) - 1 else "")
        if cur and len(indent) + len(cur) + 1 + len(piece) > width:
            out.append(indent + cur)
            cur = piece
        else:
            cur = f"{cur} {piece}" if cur else piece
    if cur:
        out.append(indent + cur)
    return out


def render_text(report: Report, summary: bool = False) -> str:
    s = report.snapshot
    if report.overall in NOTHING_MEASURED:
        head = f"=== {s.campaign_id} ===\n{NOTHING_MEASURED[report.overall]}: {report.verdict_explainer}\n"
        return head + "".join(f"note: {x}\n" for x in report.notes if x != report.verdict_explainer)

    hl = report.headline
    tc = _toolchain_label(s.toolchain)

    stab_vals = [i.stability for i in s.instances if i.stability > 0]
    stability = f"{min(stab_vals):.1f}%" if stab_vals else hl.get("stability", "?")
    eps_now = sum((i.execs_ps_last_min or i.execs_per_sec) for i in s.instances if i.alive)
    speed = f"{hl.get('speed', '?')} ({eps_now:.0f} exec/s)"
    coverage = hl.get("coverage", "?")
    ages = [
        f"last find: {_hm(s.trend.last_find_age_s)}" if s.trend.last_find_age_s is not None else "",
        f"last +cov: {_hm(s.trend.last_cov_entry_age_s)}" if s.trend.last_cov_entry_age_s is not None else "",
    ]
    ages = [a for a in ages if a]
    if ages:
        coverage += f" ({', '.join(ages)})"

    verdict = report.overall.upper()
    if report.degraded_reasons:
        verdict += f" ({degraded_tag(report.degraded_reasons)})"

    lines: list[str] = [f"=== {s.campaign_id} ==="]

    if summary:
        lines.append(
            f"Verdict: {verdict}  Type: {tc}  "
            f"Running: {s.alive_count}/{s.instance_count}{_dead_suffix(s)}  Runtime: {_hm(s.run_time_s)}  "
            f"Stability: {stability}  Speed: {speed}  "
            f"Coverage: {s.bitmap_cvg_max():.2f}% {coverage}  "
            f"Favs pending: {s.pending_favs_total()}"
        )
    else:
        lines += [
            f"{'Type:':<10} {tc}",
            f"{'Running:':<10} {s.alive_count}/{s.instance_count}{_dead_suffix(s)}",
            f"{'Runtime:':<10} {_hm(s.run_time_s)}",
            f"{'Stability:':<10} {stability}",
            f"{'Speed:':<10} {speed}",
            f"{'Coverage:':<10} {coverage}",
            "",
            f"{'Verdict:':<10} {verdict}",
            "",
        ]

        header = (
            f"{'instance name':<18}{'role/variant':<18}{'alive':<6}{'exec/s':>8}{'stab':>7}"
            f"{'cvg':>9}{'favs':>7}{'crash':>6}  status"
        )
        lines.append(header)
        lines.append("-" * len(header))
        for i in sorted(s.instances, key=lambda x: -x.crash_files):
            lines.append(
                f"{i.name:<18}{i.role + '/' + i.variant:<18}{('yes' if i.alive else 'NO'):<6}"
                f"{i.execs_per_sec:>8.0f}{i.stability:>6.1f}%{i.bitmap_cvg:>8.2f}%{i.pending_favs:>7}"
                f"{i.crash_files:>6}  {i.status}",
            )

        t = s.trend
        lines.append("")
        delta = ""
        if t.cross_run and t.cross_run.get("edges_delta") is not None:
            delta = f", Δ{t.cross_run['edges_delta']:+d} edges since last run"
        cov = "n/a" if t.last_cov_entry_age_s is None else _hm(t.last_cov_entry_age_s)
        norm = f"{t.execs_per_sec_baseline:.1f}" if t.execs_per_sec_baseline > 0 else "n/a"
        total_edges = max((i.total_edges for i in s.instances), default=0)
        lines.append(
            f"Trend [source: {t.source}]: edges {t.edges_found_now}/{total_edges} (peak {t.edges_found_peak}, "
            f"{t.edges_slope_recent_per_min:+.1f}/min), +cov age {cov}, "
            f"exec/s {t.execs_per_sec_recent:.1f} (norm {norm}, peak {t.execs_per_sec_peak:.1f}){delta}",
        )
        lines.append(
            f"Crashes: {s.crashes.total} ({s.crashes.hangs_total} hangs)"
            + (f", cores {s.crashes.core_bytes / 1024**3:.0f}GB" if s.crashes.core_bytes else ""),
        )

    if report.actions:
        lines.append("")
        lines.append("Action required:")
        for severity, action, template, members in sorted(
            _group_actions(report.actions), key=lambda g: _SEV_RANK.get(g[0], 3)
        ):
            cmd = next((m.suggested_command for m in members if m.suggested_command), None)
            if len(members) == 1:
                where = members[0].target.get("instance") or "campaign"
                lines.append(f"  [{severity}] {action} ({where}) — {members[0].rationale}")
            else:
                title, labels = _action_group_view(template, members)
                head = f"  [{severity}] {action} ({len(members)} instances)"
                lines.append(f"{head} — {title}" if title else head)
                lines += _wrap_labels(labels, indent="          ")
            if cmd:
                lines.append(f"          $ {cmd}")
    for note in report.notes:
        lines.append(f"note: {note}")
    return "\n".join(lines) + "\n"


# ============================================================================
# store — persisted snapshot history (SQLite) for cross-run trend.
#
# One row per run, keyed by campaign_id (host:abspath). prior() returns the most
# recent stored row's key metrics so the analyzer can compute cross-run deltas
# (call prior() before save() so it reflects the *previous* run). sqlite3 is
# stdlib, so this works everywhere with no extra dependency.
# ============================================================================

_SCHEMA = """
CREATE TABLE IF NOT EXISTS snapshots (
    campaign_id    TEXT NOT NULL,
    generated_at   INTEGER NOT NULL,
    run_time_s     INTEGER,
    edges_found    INTEGER,
    eps_now        REAL,
    eps_peak       REAL,
    corpus_count   INTEGER,
    pending_favs   INTEGER,
    crashes        INTEGER,
    hangs          INTEGER,
    last_cov_age_s INTEGER,
    verdict        TEXT,
    raw_json       TEXT
);
CREATE INDEX IF NOT EXISTS idx_snapshots_cid_ts ON snapshots(campaign_id, generated_at);
"""

_METRIC_COLS = (
    "generated_at",
    "run_time_s",
    "edges_found",
    "eps_now",
    "eps_peak",
    "corpus_count",
    "pending_favs",
    "crashes",
    "hangs",
    "last_cov_age_s",
    "verdict",
)


class TrendStore:
    def __init__(self, path: str | Path) -> None:
        self.path = Path(path)
        self.path.parent.mkdir(parents=True, exist_ok=True)
        self._conn = sqlite3.connect(str(self.path))
        self._conn.row_factory = sqlite3.Row
        self._conn.executescript(_SCHEMA)
        self._conn.commit()

    def save(self, report: Report) -> None:
        s = report.snapshot
        t = s.trend
        if report.overall in NOTHING_MEASURED:
            # No instance could be measured: store the verdict only, with NULL metrics, so the
            # next run's cross-run delta is suppressed instead of computed against fake zeros.
            self._conn.execute(
                "INSERT INTO snapshots (campaign_id, generated_at, verdict, raw_json) VALUES (?,?,?,?)",
                (s.campaign_id, s.generated_at, report.overall, json.dumps(report.to_dict())),
            )
            self._conn.commit()
            return
        self._conn.execute(
            "INSERT INTO snapshots (campaign_id, generated_at, run_time_s, edges_found, "
            "eps_now, eps_peak, corpus_count, pending_favs, crashes, hangs, last_cov_age_s, "
            "verdict, raw_json) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?)",
            (
                s.campaign_id,
                s.generated_at,
                s.run_time_s,
                s.edges_found_max(),
                t.execs_per_sec_now,
                t.execs_per_sec_peak,
                sum(i.corpus_count for i in s.instances),
                sum(i.pending_favs for i in s.instances),
                s.crashes.total,
                s.crashes.hangs_total,
                t.last_cov_entry_age_s,
                report.overall,
                json.dumps(report.to_dict()),
            ),
        )
        self._conn.commit()

    def prior(self, campaign_id: str) -> dict[str, Any] | None:
        """Most recent stored metrics for this campaign, or None if never seen."""
        row = self._conn.execute(
            f"SELECT {', '.join(_METRIC_COLS)} FROM snapshots WHERE campaign_id = ? "  # noqa: S608
            "ORDER BY generated_at DESC, rowid DESC LIMIT 1",
            (campaign_id,),
        ).fetchone()
        return dict(row) if row is not None else None

    def close(self) -> None:
        self._conn.close()


# ============================================================================
# collect — gather raw campaign artifacts, locally or over SSH, into a RawCampaign.
#
# Both paths run the *same* POSIX-sh collector so behavior is identical. The remote
# needs nothing beyond sh + coreutils; we never transfer corpora or core dumps —
# only fuzzer_stats, a plot_data tail, the newest +cov mtime, and crash/hang/core
# counts. The collector emits a length-prefixed, binary-safe dump that parse_dump
# turns into a RawCampaign. Every blob length is measured from the same snapshot
# that is written out, because afl-fuzz replaces fuzzer_stats by rename(): reading
# the file twice can see two different inodes and would desync the stream.
#
# Instances are discovered by fuzzer_stats *or* fuzzer_setup: afl-fuzz writes the
# setup file at startup but the first fuzzer_stats only once the initial dry run and
# calibration are done, which on a large corpus takes many minutes. A dir whose stats
# are missing — or older than the setup file, i.e. a restarted instance whose stats are
# still the previous run's — is therefore an instance that is starting up (STARTING 1).
# It has no usable fuzzer_pid, so it is matched to its live afl-fuzz through /proc; a
# restarted dir with no such process is not starting at all and keeps the normal path.
# ============================================================================

# POSIX-sh collector. $1 = campaign output dir. Honours $AFL_HEALTH_NOW for tests.
COLLECTOR_SH = r"""
DIR=$1
nopids=$2
abs=$(cd "$DIR" 2>/dev/null && pwd) || { printf 'ERROR no such dir: %s\n' "$DIR"; exit 9; }
now=${AFL_HEALTH_NOW:-$(date +%s)}
found=$(find "$abs" -maxdepth 3 \( -name queue -o -name crashes -o -name hangs \) -prune -o -type f \( -name fuzzer_stats -o -name fuzzer_setup \) -print 2>/dev/null)
dirs=$(printf '%s\n' "$found" | sed -n 's|/[^/]*$||p' | sort -u)
hf=0
if [ -n "$(find "$abs" \( -name queue -o -name crashes -o -name hangs \) -prune -o -type d -name honggfuzz -print 2>/dev/null | head -n1)" ]; then hf=1; fi
layout=flat
if printf '%s\n' "$found" | grep -q '/afl/[^/]*/fuzzer_s'; then
  layout=ziggy-afl-subdir
fi
OWN_AWK='
    NR==1 { n=split($0,a,"/"); if (a[n] ~ /afl-fuzz/) bin=1; next }
    stop { next }
    wo { if (fo=="") fo=$0; wo=0; next }
    wn { if (fn=="") fn=$0; wn=0; next }
    $0=="--" { stop=1; next }
    $0=="-o" { wo=1; next }
    $0=="-M" || $0=="-S" { wn=1; next }
    substr($0,1,2)=="-o" { if (fo=="") fo=substr($0,3); next }
    substr($0,1,2)=="-M" || substr($0,1,2)=="-S" { if (fn=="") fn=substr($0,3); next }
    END {
      if (!bin) exit 1
      if (fn=="") fn="default"
      printf "%s|%s\n", fn, fo
    }'
pid_ident() {
  tr '\0' '\n' < "/proc/$1/cmdline" 2>/dev/null | awk "$OWN_AWK"
}
phys() {
  (cd "$1" 2>/dev/null && pwd -P)
}
base_of() {
  b=${1%/}; printf '%s' "${b##*/}"
}
own_dir() {
  case $1 in
    "") return 0 ;;
    /*) phys "$1" ;;
    *) c=$(readlink "/proc/$2/cwd" 2>/dev/null); [ -n "$c" ] && phys "$c/$1" ;;
  esac
}
pid_owns() {
  ident=$(pid_ident "$1") || return 1
  [ "${ident%%|*}" = "$2" ] || return 1
  oraw=${ident#*|}
  od=$(own_dir "$oraw" "$1")
  if [ -n "$od" ]; then
    [ "$od" = "$3" ] || return 1
  else
    ob=$(base_of "$oraw")
    [ -z "$ob" ] || [ "$ob" = "$4" ] || return 1
  fi
  return 0
}
owns_dir() {
  printf '%s' "$owners" | (
    while IFS='|' read -r onm ophys obase; do
      [ "$onm" = "$1" ] || continue
      if [ -n "$ophys" ]; then
        if [ "$ophys" = "$2" ]; then exit 0; fi
      elif [ -z "$obase" ] || [ "$obase" = "$3" ]; then
        exit 0
      fi
    done
    exit 1
  )
}
starting_dirs=$(printf '%s\n' "$dirs" | while IFS= read -r d; do
  [ -n "$d" ] || continue
  if [ ! -f "$d/fuzzer_stats" ]; then
    printf '%s\n' "$d"
  elif [ -f "$d/fuzzer_setup" ] && [ "$d/fuzzer_stats" -ot "$d/fuzzer_setup" ]; then
    printf '%s\n' "$d"
  fi
done)
owners=
if [ "$nopids" != 1 ] && [ -n "$starting_dirs" ] && [ -d /proc ]; then
  cands=$(grep -las afl-fuzz /proc/[0-9]*/cmdline 2>/dev/null)
  if [ $? -gt 1 ]; then cands=$(echo /proc/[0-9]*/cmdline); fi
  for c in $cands; do
    p=${c#/proc/}; p=${p%/cmdline}
    ident=$(pid_ident "$p") || continue
    oraw=${ident#*|}
    owners="$owners${ident%%|*}|$(own_dir "$oraw" "$p")|$(base_of "$oraw")
"
  done
fi
printf 'CAMPAIGN now=%s honggfuzz=%s layout=%s\n' "$now" "$hf" "$layout"
printf 'DIR %s\n' "$abs"
printf '%s\n' "$dirs" | while IFS= read -r d; do
  [ -n "$d" ] || continue
  f=$d/fuzzer_stats
  name=$(basename "$d")
  parent=$(dirname "$d")
  pphys=$(phys "$parent")
  pbase=$(base_of "$parent")
  starting=0
  restarted=0
  if [ ! -f "$f" ]; then
    starting=1
  elif [ -f "$d/fuzzer_setup" ] && [ "$f" -ot "$d/fuzzer_setup" ]; then
    restarted=1
  fi
  pa=0
  if [ "$nopids" != 1 ]; then
    if [ "$starting" = 1 ] || [ "$restarted" = 1 ]; then
      if owns_dir "$name" "$pphys" "$pbase"; then
        pa=1
        starting=1
      fi
    fi
    if [ "$starting" = 0 ]; then
      pid=$(awk -F: '/^fuzzer_pid/{gsub(/[ \t]/,"",$2); print $2; exit}' "$f")
      if [ -n "$pid" ]; then
        if [ -d "/proc/$pid" ]; then
          pa=1
          if [ -r "/proc/$pid/cmdline" ]; then
            pid_owns "$pid" "$name" "$pphys" "$pbase" || pa=0
          fi
        elif [ ! -d /proc ] && kill -0 "$pid" 2>/dev/null; then
          pa=1
        fi
      fi
    fi
  fi
  qm=$(find "$d/queue" -maxdepth 1 -type f -name 'id:*' -printf '%T@ %f\n' 2>/dev/null | awk '{ if ($1+0>q) q=$1+0; if (index($2,"+cov") && $1+0>c) c=$1+0 } END { printf "%.0f %.0f", q+0, c+0 }')
  qmtime=${qm%% *}; cov=${qm##* }
  cr=$(find "$d/crashes" -maxdepth 1 -type f -name 'id:*' ! -name '*.core' ! -name '*.txt' ! -name '*.metadata' ! -name 'README.txt' 2>/dev/null | wc -l | tr -d ' ')
  hg=$(find "$d/hangs" -maxdepth 1 -type f -name 'id:*' ! -name '*.core' ! -name '*.txt' ! -name '*.metadata' ! -name 'README.txt' 2>/dev/null | wc -l | tr -d ' ')
  cb=$(find "$d/crashes" -maxdepth 1 -type f -name '*.core' -printf '%s\n' 2>/dev/null | awk '{s+=$1} END{printf "%d", s+0}')
  fr=0; if [ -f "$d/fastresume.bin" ]; then fr=1; fi
  mn=0; if [ -f "$d/is_main_node" ]; then mn=1; fi
  nf=0; if grep -q '^AFL_NO_FASTRESUME=.' "$d/fuzzer_setup" 2>/dev/null; then nf=1; fi
  sm=$(find "$d" -maxdepth 1 -type f -name fuzzer_setup -printf '%T@\n' 2>/dev/null | awk '{ printf "%.0f", $1+0; exit }')
  printf 'INSTANCE %s\n' "$name"
  printf 'PIDALIVE %s\n' "$pa"
  printf 'COVMTIME %s\n' "$cov"
  printf 'QUEUEMTIME %s\n' "$qmtime"
  printf 'CRASHES %s\n' "$cr"
  printf 'HANGS %s\n' "$hg"
  printf 'COREBYTES %s\n' "$cb"
  printf 'FASTRESUME %s\n' "$fr"
  printf 'MAINNODEFILE %s\n' "$mn"
  printf 'NOFASTRESUME %s\n' "$nf"
  printf 'STARTING %s\n' "$starting"
  printf 'SETUPMTIME %s\n' "${sm:-0}"
  if sdata=$(cat "$f" 2>/dev/null); then
    sn=$(printf '%s\n' "$sdata" | wc -c | tr -d ' ')
    printf 'BLOB stats %s\n' "$sn"
    printf '%s\n' "$sdata"
  else
    printf 'BLOB stats 0\n'
  fi
  if [ -f "$d/plot_data" ]; then
    pt=$( { head -n1 "$d/plot_data"; tail -n 200 "$d/plot_data"; } )
    pn=$(printf '%s' "$pt" | wc -c | tr -d ' ')
    printf 'BLOB plot %s\n' "$pn"
    printf '%s' "$pt"
  else
    printf 'BLOB plot 0\n'
  fi
  sc=
  if [ ! -f "$f" ]; then
    sc=$(awk '/^# command line:/{getline; print; exit}' "$d/fuzzer_setup" 2>/dev/null)
  fi
  if [ -n "$sc" ]; then
    cn=$(printf '%s\n' "$sc" | wc -c | tr -d ' ')
    printf 'BLOB setup %s\n' "$cn"
    printf '%s\n' "$sc"
  else
    printf 'BLOB setup 0\n'
  fi
  printf 'ENDINSTANCE\n'
done
printf 'END\n'
"""


class DumpFormatError(Exception):
    """The collector output did not follow the length-prefixed dump protocol."""


@dataclass
class Target:
    output_dir: str
    host: str = "local"  # "local" or an ssh destination ([user@]host)
    port: int | None = None
    ssh_opts: list[str] = field(default_factory=list)
    ssh_exe: str = "ssh"
    connect_timeout: int = 12

    @property
    def is_local(self) -> bool:
        return self.host == "local"


def _err(target: Target, msg: str, now: int | None) -> RawCampaign:
    return RawCampaign(
        campaign_id=f"{target.host}:{target.output_dir}",
        host=target.host,
        output_dir=target.output_dir,
        now=now if now is not None else int(time.time()),
        layout="flat",
        honggfuzz_present=False,
        collection_error=msg,
        instances=[],
    )


def _ssh_base_opts(target: Target) -> list[str]:
    """Common ssh ``-o`` options for both collection and glob expansion."""
    opts = ["-o", "BatchMode=yes", "-o", f"ConnectTimeout={target.connect_timeout}"]
    if target.port:
        opts += ["-p", str(target.port)]
    return opts + target.ssh_opts


def _run(target: Target, no_pids: bool = False) -> subprocess.CompletedProcess[bytes]:
    script = COLLECTOR_SH.encode()
    np = "1" if no_pids else "0"
    if target.is_local:
        argv = ["sh", "-s", target.output_dir, np]
    else:
        argv = [target.ssh_exe, *_ssh_base_opts(target), target.host, f"sh -s {shlex.quote(target.output_dir)} {np}"]
    return subprocess.run(
        argv,
        input=script,
        capture_output=True,
        timeout=target.connect_timeout + 60,
        check=False,
    )


def collect(target: Target, *, now_override: int | None = None, no_pids: bool = False) -> RawCampaign:
    try:
        proc = _run(target, no_pids)
    except (subprocess.TimeoutExpired, OSError) as e:
        return _err(target, f"{type(e).__name__}: {e}", now_override)
    out = proc.stdout
    if not out.startswith((b"CAMPAIGN", b"ERROR")):
        stderr = proc.stderr.decode("utf-8", "replace").strip()
        return _err(target, stderr or f"collector exited {proc.returncode}", now_override)
    try:
        raw = parse_dump(out, host=target.host, output_dir=target.output_dir)
    except DumpFormatError as e:
        return _err(target, f"malformed collector output: {e}", now_override)
    if now_override is not None:
        raw.now = now_override
    return raw


def _opt_mtime(token: str) -> float | None:
    """An mtime field from the collector; absent/zero (no such file) becomes None."""
    try:
        value = float(token.strip())
    except ValueError:
        return None
    return value if value > 0 else None


def _opt_count(token: str) -> int:
    """A count field from the collector; absent or non-numeric becomes 0."""
    try:
        return int(token.strip())
    except ValueError:
        return 0


def parse_dump(data: bytes, *, host: str, output_dir: str = "") -> RawCampaign:
    pos = 0

    def read_line() -> str:
        nonlocal pos
        nl = data.find(b"\n", pos)
        if nl < 0:
            s = data[pos:].decode("utf-8", "replace")
            pos = len(data)
            return s
        s = data[pos:nl].decode("utf-8", "replace")
        pos = nl + 1
        return s

    def read_bytes(n: int) -> bytes:
        nonlocal pos
        b = data[pos : pos + n]
        pos += n
        return b

    def field_after_space() -> str:
        parts = read_line().split(" ", 1)
        return parts[1] if len(parts) > 1 else ""

    def read_blob(kind: str) -> str:
        header = read_line()
        parts = header.split()
        if len(parts) != 3 or parts[0] != "BLOB" or parts[1] != kind or not parts[2].isdigit():
            raise DumpFormatError(f"expected 'BLOB {kind} <len>', got {header[:60]!r}")
        n = int(parts[2])
        blob = read_bytes(n)
        if len(blob) != n:
            raise DumpFormatError(f"{kind} blob truncated: want {n} bytes, got {len(blob)}")
        return blob.decode("utf-8", "replace")

    first = read_line()
    if first.startswith("ERROR"):
        return RawCampaign(
            f"{host}:{output_dir or '?'}",
            host,
            output_dir,
            0,
            "flat",
            honggfuzz_present=False,
            collection_error=first,
            instances=[],
        )

    meta = dict(tok.split("=", 1) for tok in first.split()[1:] if "=" in tok)
    now = int(meta.get("now") or 0)
    honggfuzz = meta.get("honggfuzz") == "1"
    layout = meta.get("layout", "flat")
    dirline = read_line()
    absdir = dirline[4:] if dirline.startswith("DIR ") else ""

    instances: list[RawInstance] = []
    while True:
        line = read_line()
        if line in ("END", ""):
            break
        if not line.startswith("INSTANCE "):
            continue
        name = line[len("INSTANCE ") :]
        pid_alive = field_after_space() == "1"
        cov_mtime = _opt_mtime(field_after_space())
        queue_mtime = _opt_mtime(field_after_space())
        crash_count = _opt_count(field_after_space())
        hang_count = _opt_count(field_after_space())
        core_bytes = _opt_count(field_after_space())
        fastresume = field_after_space() == "1"
        main_node_file = field_after_space() == "1"
        no_fastresume_env = field_after_space() == "1"
        starting = field_after_space() == "1"
        setup_mtime = _opt_mtime(field_after_space())
        stats = read_blob("stats")
        plot = read_blob("plot") or None
        setup_cmdline = read_blob("setup")
        end = read_line()
        if end != "ENDINSTANCE":
            raise DumpFormatError(f"expected 'ENDINSTANCE' after {name}, got {end[:60]!r}")
        instances.append(
            RawInstance(
                name,
                stats,
                plot,
                cov_mtime,
                pid_alive,
                crash_count,
                hang_count,
                core_bytes,
                queue_mtime,
                fastresume,
                main_node_file,
                no_fastresume_env,
                starting,
                setup_mtime,
                setup_cmdline,
            )
        )

    return RawCampaign(
        f"{host}:{absdir}",
        host,
        absdir,
        now,
        layout,
        honggfuzz_present=honggfuzz,
        collection_error=None,
        instances=instances,
    )


# ============================================================================
# config — target resolution: CLI specs + optional JSON config into Target objects.
# ============================================================================


def parse_target(spec: str, *, ssh_opts: list[str], connect_timeout: int) -> Target:
    """Parse one target spec.

    ``[user@]host:/path`` is an ssh target; anything else is a local path. The ssh
    form is recognized when the part before the first ``:`` contains no ``/``.
    """
    if ":" in spec:
        head, _, path = spec.partition(":")
        if head and "/" not in head and head != ".":
            return Target(output_dir=path, host=head, ssh_opts=list(ssh_opts), connect_timeout=connect_timeout)
    return Target(output_dir=spec, ssh_opts=list(ssh_opts), connect_timeout=connect_timeout)


_GLOB_META = re.compile(r"[*?\[]")


def _has_glob(path: str) -> bool:
    return bool(_GLOB_META.search(path))


def _shell_glob_quote(path: str) -> str:
    """Quote ``path`` for a remote shell while leaving glob metacharacters bare.

    shlex.quote would single-quote ``*?[]`` and kill globbing; instead we quote
    only the literal runs so the *remote* shell expands the pattern, yet paths with
    spaces / shell metacharacters stay injection-safe.
    """
    parts = re.split(r"([*?\[\]])", path)
    return "".join(tok if tok in "*?[]" else shlex.quote(tok) for tok in parts if tok != "")


def _expand_remote_glob(target: Target) -> list[str]:
    """Expand a glob ``output_dir`` on the remote host, returning matched directories."""
    pattern = _shell_glob_quote(target.output_dir)
    remote = f'for d in {pattern}; do [ -d "$d" ] && printf "%s\\n" "$d"; done'
    argv = [target.ssh_exe, *_ssh_base_opts(target), target.host, remote]
    try:
        proc = subprocess.run(argv, capture_output=True, timeout=target.connect_timeout + 30, check=False)
    except (subprocess.TimeoutExpired, OSError):
        return []
    return [ln for ln in proc.stdout.decode("utf-8", "replace").splitlines() if ln.strip()]


def expand_globs(targets: list[Target]) -> list[Target]:
    """Expand any glob ``output_dir`` into one Target per matched directory.

    Local globs that the shell left literal (e.g. quoted) are expanded with the
    ``glob`` module; remote globs are expanded over ssh. A pattern that matches
    nothing is kept verbatim so it surfaces a clean "no such dir" collection error.
    """
    out: list[Target] = []
    for t in targets:
        if not _has_glob(t.output_dir):
            out.append(t)
            continue
        if t.is_local:
            matches = sorted(p for p in glob.glob(t.output_dir) if Path(p).is_dir())
        else:
            matches = _expand_remote_glob(t)
        if matches:
            out.extend(replace(t, output_dir=m) for m in matches)
        else:
            out.append(t)
    return out


def load_config(path: str) -> dict[str, Any]:
    return json.loads(Path(path).read_text(encoding="utf-8"))


def resolve_targets(
    specs: list[str],
    *,
    config: dict[str, Any] | None,
    ssh_opts: list[str],
    connect_timeout: int,
) -> list[Target]:
    targets: list[Target] = []
    if config:
        for entry in config.get("targets", []):
            if isinstance(entry, str):
                targets.append(parse_target(entry, ssh_opts=ssh_opts, connect_timeout=connect_timeout))
            elif isinstance(entry, dict):
                targets.append(
                    Target(
                        output_dir=entry["path"],
                        host=entry.get("host", "local"),
                        port=entry.get("port"),
                        ssh_opts=entry.get("ssh_opts", list(ssh_opts)),
                        connect_timeout=entry.get("connect_timeout", connect_timeout),
                    )
                )
    for spec in specs:
        targets.append(parse_target(spec, ssh_opts=ssh_opts, connect_timeout=connect_timeout))
    return targets


# ============================================================================
# cli — entry point: one-shot / --watch, fleet, exit codes, on-change hook.
# ============================================================================


def detect_change(report: Report, prior: dict[str, Any] | None) -> dict[str, Any] | None:
    """A change worth alerting on vs the previous run: verdict flip or new crashes.

    Returns ``None`` on the first-ever run (no baseline) or when nothing changed.
    """
    if prior is None:
        return None
    changes: list[str] = []
    if report.overall != prior.get("verdict"):
        changes.append(f"verdict {prior.get('verdict')} -> {report.overall}")
    prev_crashes = int(prior.get("crashes") or 0)
    if report.snapshot.crashes.total > prev_crashes:
        changes.append(f"new crashes {prev_crashes} -> {report.snapshot.crashes.total}")
    if not changes:
        return None
    return {
        "campaign_id": report.snapshot.campaign_id,
        "change": changes,
        "overall": report.overall,
        "prev_overall": prior.get("verdict"),
        "crashes": report.snapshot.crashes.total,
        "prev_crashes": prev_crashes,
        "generated_at": report.snapshot.generated_at,
    }


def run_hook(cmd: str, event: dict[str, Any]) -> None:
    """Run the user's --on-change hook, piping the change event as JSON on stdin."""
    with contextlib.suppress(subprocess.TimeoutExpired, OSError):
        subprocess.run(
            cmd,
            shell=True,
            input=json.dumps(event),
            text=True,
            capture_output=True,
            timeout=30,
            check=False,
        )


def run_once(
    targets: list[Target],
    *,
    store: TrendStore | None = None,
    expected_eps: float | None = None,
    now: int | None = None,
    on_change: str | None = None,
    no_pids: bool = False,
) -> list[Report]:
    # Collection is the slow, I/O-bound (ssh) step → parallelize it; analysis and the
    # (single-threaded) store run on the main thread afterwards.
    workers = min(8, max(1, len(targets)))
    with ThreadPoolExecutor(max_workers=workers) as pool:
        raws = list(pool.map(lambda t: collect(t, now_override=now, no_pids=no_pids), targets))

    reports: list[Report] = []
    for raw in raws:
        snap = build_snapshot(raw, no_pids=no_pids)
        prior = store.prior(snap.campaign_id) if store else None
        report = analyze(snap, expected_eps=expected_eps, prior=prior)
        if store:
            store.save(report)
        event = detect_change(report, prior)
        if event and on_change:
            run_hook(on_change, event)
        reports.append(report)
    return reports


def _ssh_opts(raw_opts: list[str]) -> list[str]:
    out: list[str] = []
    for opt in raw_opts:
        out += ["-o", opt]
    return out


def _default_state_dir() -> Path:
    base = os.environ.get("XDG_STATE_HOME") or str(Path.home() / ".local" / "state")
    return Path(base) / "afl-health"


class _HelpFormatter(argparse.HelpFormatter):
    def __init__(self, prog, *args, **kwargs):
        super().__init__(prog, *args, **kwargs)
        self.add_text("health, trend, and action analysis tool for AFL++ campaigns")
        self.add_text("")
        self.add_text("%s [ options ] [TARGET ...]" % prog)


def _build_parser() -> argparse.ArgumentParser:
    p = argparse.ArgumentParser(
        prog="afl-health",
        formatter_class=_HelpFormatter,
        description="Health, trend, and action analysis for AFL++/cargo-afl/ziggy campaigns (local or over ssh).",
    )
    p.add_argument(
        "targets",
        nargs="*",
        metavar="TARGET",
        help="local /path/to/out, or [user@]host:/path/to/out for ssh; "
        "glob patterns (e.g. host:/runs/*/afl) expand to one campaign per match (remotely for ssh)",
    )
    p.add_argument("--config", help="JSON config defining targets / ssh opts / state dir")
    p.add_argument("--json", action="store_true", help="machine report (incl. action directives)")
    p.add_argument(
        "--summary",
        action="store_true",
        help="condensed one-line per campaign (fleet roll-up for many); a degraded or stalled campaign "
        "names its degradation reason(s): " + " / ".join(DEGRADED_REASON_TEXT),
    )
    g = p.add_mutually_exclusive_group()
    g.add_argument("-r", "--running", "-a", "--alive", action="store_true", help="show only running/alive campaigns (>=1 instance alive)")
    g.add_argument("-d", "--dead", action="store_true", help="show only dead campaigns (no instance alive)")
    p.add_argument("--watch", type=float, metavar="SECONDS", help="live refresh every N seconds")
    p.add_argument(
        "--on-change", metavar="CMD", help="run CMD when a target changes state; event JSON is piped on stdin"
    )
    p.add_argument("--expect-eps", type=float, help="expected execs/sec (sharpens speed judgment)")
    p.add_argument("--state-dir", help="trend history location (default: XDG state dir)")
    p.add_argument("--no-store", action="store_true", help="do not persist/read trend history")
    p.add_argument("--ssh-opt", action="append", default=[], metavar="KEY=VAL", help="extra ssh -o option (repeatable)")
    p.add_argument("--fail-on", choices=sorted(EXIT), help="exit nonzero only at/above this verdict")
    p.add_argument("--connect-timeout", type=int, default=12, help="ssh connect timeout (seconds)")
    p.add_argument(
        "-n",
        "--no-pids",
        action="store_true",
        help=f"judge liveness by fuzzer_stats freshness (<{NO_PIDS_STALE_SECONDS}s) instead of the pid; "
        "for output dirs copied off the host that ran them",
    )
    p.add_argument("--quiet", action="store_true", help="suppress human output (for scripting)")
    p.add_argument("--now", type=int, help=argparse.SUPPRESS)  # deterministic clock for tests
    p.add_argument(
        "--version", action="version", version=f"afl-health {__version__} (license: {__license__})"
    )
    return p


def _filter_reports(reports: list[Report], args: argparse.Namespace) -> list[Report]:
    if args.running:
        return [r for r in reports if r.snapshot.alive_count > 0]
    if args.dead:
        return [r for r in reports if r.snapshot.alive_count == 0]
    return reports


def _emit(reports: list[Report], args: argparse.Namespace, now: int) -> None:
    if args.json:
        print(render_json(reports, now=now, version=__version__))
        return
    if args.quiet:
        return
    if not reports:
        print("No running campaigns." if args.running else "No dead campaigns." if args.dead else "No campaigns.")
        return
    if args.summary and len(reports) > 1:
        print(_fleet_summary(reports))
        return
    print("\n".join(render_text(r, summary=args.summary) for r in reports), end="")


def _fleet_summary(reports: list[Report]) -> str:
    lines = [f"FLEET verdict: {fleet_verdict(reports).upper()}  ({len(reports)} campaigns)", ""]
    for r in reports:
        s = r.snapshot
        if r.overall in NOTHING_MEASURED:
            reason = r.verdict_explainer.split(" — ")[0].split("(")[0].strip()
            lines.append(f"  {r.overall.upper():<14} {s.campaign_id}  (nothing measured: {reason})")
            continue
        why = f", degraded: {degraded_tag(r.degraded_reasons)}" if r.degraded_reasons else ""
        stab_vals = [i.stability for i in s.instances if i.stability > 0]
        stability = f"{min(stab_vals):.1f}%" if stab_vals else r.headline.get("stability", "?")
        cov_age = "n/a" if s.trend.last_cov_entry_age_s is None else _hm(s.trend.last_cov_entry_age_s)
        lines.append(
            f"  {r.overall.upper():<14} {s.campaign_id}  "
            f"({s.alive_count}/{s.instance_count} alive{_dead_suffix(s)}, {s.bitmap_cvg_max():.2f}% cov, "
            f"{stability} stab, {s.pending_favs_total()} favs, {s.crashes.total} crashes, "
            f"{cov_age} +cov{why})"
        )
    return "\n".join(lines)


def _final_rc(reports: list[Report], args: argparse.Namespace) -> int:
    rc = exit_code(reports)
    if args.fail_on:
        return rc if rc >= EXIT[args.fail_on] else 0
    return rc


def _watch(targets: list[Target], args: argparse.Namespace, store: TrendStore | None, now: int | None) -> int:
    reports: list[Report] = []
    try:
        while True:
            reports = run_once(
                targets,
                store=store,
                expected_eps=args.expect_eps,
                now=now,
                on_change=args.on_change,
                no_pids=args.no_pids,
            )
            reports = _filter_reports(reports, args)
            sys.stdout.write("\x1b[2J\x1b[H")  # clear screen
            _emit(reports, args, now or int(time.time()))
            sys.stdout.flush()
            time.sleep(args.watch)
    except KeyboardInterrupt:
        return _final_rc(reports, args) if reports else 0


def main(argv: list[str] | None = None) -> int:
    parser = _build_parser()
    args = parser.parse_args(argv)
    now = args.now
    if now is None and os.environ.get("AFL_HEALTH_NOW"):
        now = int(os.environ["AFL_HEALTH_NOW"])

    config = load_config(args.config) if args.config else None
    ssh_opts = _ssh_opts(args.ssh_opt)
    targets = resolve_targets(args.targets, config=config, ssh_opts=ssh_opts, connect_timeout=args.connect_timeout)
    if not targets:
        parser.print_help()  # no target given → behave like -h
        return 0
    targets = expand_globs(targets)

    store = (
        None
        if args.no_store
        else TrendStore((Path(args.state_dir) if args.state_dir else _default_state_dir()) / "history.sqlite")
    )

    if args.watch:
        return _watch(targets, args, store, now)

    reports = run_once(
        targets, store=store, expected_eps=args.expect_eps, now=now, on_change=args.on_change, no_pids=args.no_pids
    )
    reports = _filter_reports(reports, args)
    _emit(reports, args, now or int(time.time()))
    return _final_rc(reports, args)


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