foxhunt/scripts/tier1_5_verdict.py

#!/usr/bin/env python3
"""Tier 1.5 mid-smoke behavioral verdict.

Reads `diag.jsonl` + `eval_diag.jsonl` + `eval_summary.json` from the
mid-smoke output directory and computes 7 behavioral signals + 1
per-step/eval_summary consistency check. Emits OK / KILL_<reason> /
WARN_<reason>.

Spec: docs/superpowers/specs/2026-06-02-fast-dev-cycle.md §3.3, §3.6
Linked: pearl_grwwh_eval_catastrophic_collapse (per-step vs eval_summary gap)

Per `feedback_kill_runs_on_anomaly_quickly`: any KILL_* response means
DO NOT submit cluster smoke. Fix locally first.

Usage:
    python3 scripts/tier1_5_verdict.py /tmp/foxhunt-mid-smoke
    python3 scripts/tier1_5_verdict.py /tmp/foxhunt-mid-smoke --quiet
    python3 scripts/tier1_5_verdict.py /tmp/foxhunt-mid-smoke --json

Exit codes:
    0  OK (all signals within tolerance)
    1  KILL (one or more behavioral red flags)
    2  Inputs missing / malformed (cannot verdict)
"""

from __future__ import annotations

import argparse
import json
import math
import statistics
import sys
from pathlib import Path
from typing import Any


# ── Verdict thresholds (per spec §3.3 table) ──────────────────────────
# Hard floors/ceilings are KILL bounds (catastrophic).
# Soft targets are reported but not gating (informational).
THRESHOLDS = {
    # action_entropy final: log(11) = 2.397; target [0.5×log(11), 0.85×log(11)]
    # KILL if < 0.6 (policy collapsed) or > 2.3 (essentially random).
    "action_entropy_kill_low": 0.6,
    "action_entropy_kill_high": 2.3,
    "action_entropy_target_low": 1.20,   # 0.5 × log(11)
    "action_entropy_target_high": 2.04,  # 0.85 × log(11)
    # q_pi_agree_ema final: ≥ 0.6 target, KILL if < 0.3 (Q/π decoupled)
    "q_pi_agree_kill": 0.3,
    "q_pi_agree_target": 0.6,
    # Pearson(rewards.sum, Δrealized_pnl) ≥ 0.5 target, KILL < 0.3
    "pearson_kill": 0.3,
    "pearson_target": 0.5,
    # popart.sigma CV (stddev/mean): KILL > 0.5 (wild oscillation)
    "popart_sigma_cv_kill": 0.5,
    # win_rate (train) ≥ 0.25 target, KILL < 0.15
    "wr_train_kill": 0.15,
    "wr_train_target": 0.25,
    # win_rate (eval) ≥ 0.20 target, KILL < 0.15
    "wr_eval_kill": 0.15,
    "wr_eval_target": 0.20,
    # Eval pnl > -$1M target, KILL < -$3M (catastrophic)
    "eval_pnl_kill": -3_000_000.0,
    "eval_pnl_target": -1_000_000.0,
    # Phase 0 reward-redesign (2026-06-02): per-step vs eval_summary now
    # read from the SAME `realised_pnl_usd_fp / 100` source — any gap > $50
    # (one $50/pt unit of float-summation slack) is a real source bug, KILL.
    "consistency_kill_usd": 50.0,
    # Min steps required for a meaningful verdict.
    "min_train_rows": 500,
    "min_eval_rows": 100,
    # Hold-step trend: linear regression slope over last N rows must be > 0.
    # KILL if slope is significantly negative.
    "hold_trend_window": 1500,
    "hold_trend_slope_kill": -0.01,  # tolerate small decline as noise
}


def parse_args() -> argparse.Namespace:
    p = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
    p.add_argument("out_dir", type=Path, help="Mid-smoke output directory (--out from local-mid-smoke.sh)")
    p.add_argument("--quiet", action="store_true", help="Suppress per-signal explanations")
    p.add_argument("--json", action="store_true", help="Emit machine-readable JSON verdict")
    return p.parse_args()


def load_jsonl(path: Path, *, max_rows: int | None = None) -> list[dict[str, Any]]:
    """Stream-parse a JSONL file; return list of decoded records."""
    rows: list[dict[str, Any]] = []
    with path.open("r") as f:
        for line in f:
            line = line.strip()
            if not line:
                continue
            try:
                rows.append(json.loads(line))
            except json.JSONDecodeError as exc:
                # Stop at first malformed line — typically means truncated run.
                print(f"WARN: malformed JSON at line {len(rows) + 1} of {path}: {exc}", file=sys.stderr)
                break
            if max_rows is not None and len(rows) >= max_rows:
                break
    return rows


def get_nested(obj: Any, dotted: str, default: Any = None) -> Any:
    """Dotted-path lookup: get_nested({'a':{'b':1}}, 'a.b') → 1."""
    cur = obj
    for key in dotted.split("."):
        if not isinstance(cur, dict) or key not in cur:
            return default
        cur = cur[key]
    return cur


def pearson(xs: list[float], ys: list[float]) -> float | None:
    """Return Pearson correlation, or None if input is degenerate."""
    n = len(xs)
    if n != len(ys) or n < 3:
        return None
    mx = statistics.fmean(xs)
    my = statistics.fmean(ys)
    sx2 = sum((x - mx) ** 2 for x in xs)
    sy2 = sum((y - my) ** 2 for y in ys)
    if sx2 == 0.0 or sy2 == 0.0:
        return None
    sxy = sum((x - mx) * (y - my) for x, y in zip(xs, ys))
    return sxy / math.sqrt(sx2 * sy2)


def linreg_slope(xs: list[float], ys: list[float]) -> float | None:
    """OLS slope of y over x; None if degenerate."""
    n = len(xs)
    if n != len(ys) or n < 3:
        return None
    mx = statistics.fmean(xs)
    my = statistics.fmean(ys)
    num = sum((x - mx) * (y - my) for x, y in zip(xs, ys))
    den = sum((x - mx) ** 2 for x in xs)
    if den == 0.0:
        return None
    return num / den


def coefficient_of_variation(xs: list[float]) -> float | None:
    """stdev / |mean|; None if mean ≈ 0 or fewer than 2 points."""
    if len(xs) < 2:
        return None
    m = statistics.fmean(xs)
    if abs(m) < 1e-12:
        return None
    sd = statistics.stdev(xs)
    return sd / abs(m)


# ── Signal computations ───────────────────────────────────────────────


class SignalResult:
    """A single verdict line — name, value, status, explanation."""

    __slots__ = ("name", "value", "status", "explain")

    STATUS_OK = "OK"
    STATUS_KILL = "KILL"
    STATUS_WARN = "WARN"
    STATUS_SKIP = "SKIP"

    def __init__(self, name: str, value: Any, status: str, explain: str) -> None:
        self.name = name
        self.value = value
        self.status = status
        self.explain = explain

    def to_dict(self) -> dict[str, Any]:
        return {
            "name": self.name,
            "value": self.value,
            "status": self.status,
            "explain": self.explain,
        }


def signal_action_entropy(train_rows: list[dict[str, Any]]) -> SignalResult:
    """Final action_entropy: KILL if collapsed (<0.6) or random (>2.3)."""
    val = get_nested(train_rows[-1], "action_entropy")
    if val is None or not isinstance(val, (int, float)):
        return SignalResult("action_entropy", None, SignalResult.STATUS_SKIP, "field missing from final train row")
    if val < THRESHOLDS["action_entropy_kill_low"]:
        return SignalResult("action_entropy", val, SignalResult.STATUS_KILL,
                            f"{val:.3f} < {THRESHOLDS['action_entropy_kill_low']} — policy collapsed to δ-function")
    if val > THRESHOLDS["action_entropy_kill_high"]:
        return SignalResult("action_entropy", val, SignalResult.STATUS_KILL,
                            f"{val:.3f} > {THRESHOLDS['action_entropy_kill_high']} — policy essentially uniform/random")
    if val < THRESHOLDS["action_entropy_target_low"]:
        return SignalResult("action_entropy", val, SignalResult.STATUS_WARN,
                            f"{val:.3f} below target [{THRESHOLDS['action_entropy_target_low']:.2f}, {THRESHOLDS['action_entropy_target_high']:.2f}]")
    if val > THRESHOLDS["action_entropy_target_high"]:
        return SignalResult("action_entropy", val, SignalResult.STATUS_WARN,
                            f"{val:.3f} above target [{THRESHOLDS['action_entropy_target_low']:.2f}, {THRESHOLDS['action_entropy_target_high']:.2f}]")
    return SignalResult("action_entropy", val, SignalResult.STATUS_OK,
                        f"{val:.3f} within target [{THRESHOLDS['action_entropy_target_low']:.2f}, {THRESHOLDS['action_entropy_target_high']:.2f}]")


def signal_q_pi_agree(train_rows: list[dict[str, Any]]) -> SignalResult:
    """Final q_pi_agree_ema: KILL if decoupled (<0.3)."""
    val = get_nested(train_rows[-1], "q_pi_agree_ema")
    if val is None or not isinstance(val, (int, float)):
        return SignalResult("q_pi_agree_ema", None, SignalResult.STATUS_SKIP, "field missing from final train row")
    if val < THRESHOLDS["q_pi_agree_kill"]:
        return SignalResult("q_pi_agree_ema", val, SignalResult.STATUS_KILL,
                            f"{val:.3f} < {THRESHOLDS['q_pi_agree_kill']} — Q and π decoupled (π is dead weight)")
    if val < THRESHOLDS["q_pi_agree_target"]:
        return SignalResult("q_pi_agree_ema", val, SignalResult.STATUS_WARN,
                            f"{val:.3f} below target ≥ {THRESHOLDS['q_pi_agree_target']}")
    return SignalResult("q_pi_agree_ema", val, SignalResult.STATUS_OK,
                        f"{val:.3f} ≥ {THRESHOLDS['q_pi_agree_target']}")


def signal_reward_pnl_pearson(train_rows: list[dict[str, Any]]) -> SignalResult:
    """Pearson(rewards.sum, Δrealized_pnl): KILL if anti-aligned (<0.3).

    Per `pearl_reward_signal_anti_aligned_with_pnl` (2026-06-01): the
    decisive bug behind 64 negative-eval commits — reward gradient
    systematically anti-aligned with pnl direction.
    """
    # Phase 0 reward-redesign (2026-06-02): `trading.realized_pnl_cum_usd`
    # was renamed to two distinct fields. We want the AUTHORITATIVE USD pnl
    # cumulative (`realized_pnl_usd_cum`), because the Pearson signal asks
    # whether the agent's gradient signal (`rewards.sum`) tracks the actual
    # USD pnl direction — not whether it tracks itself (which would be the
    # shaped-reward cumulative, which is mechanically the cumulative sum of
    # raw_rewards itself and would trivially correlate).
    rewards: list[float] = []
    dpnls: list[float] = []
    prev_pnl: float | None = None
    for row in train_rows:
        rs = get_nested(row, "rewards.sum")
        cur_pnl = get_nested(row, "trading.realized_pnl_usd_cum")
        if rs is None or cur_pnl is None:
            continue
        if not isinstance(rs, (int, float)) or not isinstance(cur_pnl, (int, float)):
            continue
        if prev_pnl is not None:
            rewards.append(float(rs))
            dpnls.append(float(cur_pnl) - float(prev_pnl))
        prev_pnl = float(cur_pnl)
    p = pearson(rewards, dpnls)
    if p is None:
        return SignalResult("pearson(rewards.sum, Δpnl)", None, SignalResult.STATUS_SKIP,
                            f"degenerate input (n={len(rewards)})")
    if p < THRESHOLDS["pearson_kill"]:
        return SignalResult("pearson(rewards.sum, Δpnl)", p, SignalResult.STATUS_KILL,
                            f"{p:.3f} < {THRESHOLDS['pearson_kill']} — reward gradient anti-aligned with pnl (n={len(rewards)})")
    if p < THRESHOLDS["pearson_target"]:
        return SignalResult("pearson(rewards.sum, Δpnl)", p, SignalResult.STATUS_WARN,
                            f"{p:.3f} below target ≥ {THRESHOLDS['pearson_target']} (n={len(rewards)})")
    return SignalResult("pearson(rewards.sum, Δpnl)", p, SignalResult.STATUS_OK,
                        f"{p:.3f} ≥ {THRESHOLDS['pearson_target']} (n={len(rewards)})")


def signal_hold_trend(train_rows: list[dict[str, Any]]) -> SignalResult:
    """avg_hold_steps trend over last N rows: KILL if significantly decreasing.

    The surfer pattern (wave-timescale edge per `pearl_edge_lives_at_wave_timescale_not_tick`)
    requires hold-steps growing through training.
    """
    window = THRESHOLDS["hold_trend_window"]
    tail = train_rows[-window:] if len(train_rows) > window else train_rows
    xs: list[float] = []
    ys: list[float] = []
    for row in tail:
        step = get_nested(row, "step")
        hold = get_nested(row, "trading.avg_hold_steps")
        if step is None or hold is None:
            continue
        if not isinstance(step, (int, float)) or not isinstance(hold, (int, float)):
            continue
        if math.isnan(float(hold)) or math.isinf(float(hold)):
            continue
        xs.append(float(step))
        ys.append(float(hold))
    if len(xs) < 50:
        return SignalResult("avg_hold_steps trend", None, SignalResult.STATUS_SKIP,
                            f"only {len(xs)} usable points in trend window")
    slope = linreg_slope(xs, ys)
    if slope is None:
        return SignalResult("avg_hold_steps trend", None, SignalResult.STATUS_SKIP, "degenerate regression")
    if slope < THRESHOLDS["hold_trend_slope_kill"]:
        return SignalResult("avg_hold_steps trend", slope, SignalResult.STATUS_KILL,
                            f"slope {slope:.5f}/step < {THRESHOLDS['hold_trend_slope_kill']} — hold-steps decreasing, surfer pattern absent")
    if slope <= 0.0:
        return SignalResult("avg_hold_steps trend", slope, SignalResult.STATUS_WARN,
                            f"slope {slope:.5f}/step — hold-steps not growing")
    return SignalResult("avg_hold_steps trend", slope, SignalResult.STATUS_OK,
                        f"slope +{slope:.5f}/step over last {len(xs)} rows")


def signal_popart_sigma_cv(train_rows: list[dict[str, Any]]) -> SignalResult:
    """popart.sigma coefficient of variation: KILL if wild oscillation (CV > 0.5)."""
    sigmas: list[float] = []
    for row in train_rows:
        s = get_nested(row, "popart.sigma")
        if s is None or not isinstance(s, (int, float)):
            continue
        if math.isnan(float(s)) or math.isinf(float(s)):
            continue
        sigmas.append(float(s))
    if len(sigmas) < 100:
        return SignalResult("popart.sigma CV", None, SignalResult.STATUS_SKIP,
                            f"only {len(sigmas)} usable samples")
    cv = coefficient_of_variation(sigmas)
    if cv is None:
        return SignalResult("popart.sigma CV", None, SignalResult.STATUS_SKIP,
                            "mean ≈ 0 — undefined CV")
    if cv > THRESHOLDS["popart_sigma_cv_kill"]:
        return SignalResult("popart.sigma CV", cv, SignalResult.STATUS_KILL,
                            f"{cv:.3f} > {THRESHOLDS['popart_sigma_cv_kill']} — controller wildly oscillating")
    return SignalResult("popart.sigma CV", cv, SignalResult.STATUS_OK,
                        f"{cv:.3f} ≤ {THRESHOLDS['popart_sigma_cv_kill']}")


def signal_wr_train(train_rows: list[dict[str, Any]]) -> SignalResult:
    """Final win_rate_ema (train): KILL if random/worse (<0.15)."""
    # Prefer kelly.win_rate_ema (EMA across all trades) over trading.win_rate
    # which is a point measure of last batch.
    val = get_nested(train_rows[-1], "isv_config.kelly.win_rate_ema")
    if val is None:
        val = get_nested(train_rows[-1], "trading.win_rate")
        source = "trading.win_rate"
    else:
        source = "isv_config.kelly.win_rate_ema"
    if val is None or not isinstance(val, (int, float)):
        return SignalResult(f"wr_train ({source})", None, SignalResult.STATUS_SKIP,
                            "field missing from final train row")
    if val < THRESHOLDS["wr_train_kill"]:
        return SignalResult(f"wr_train ({source})", val, SignalResult.STATUS_KILL,
                            f"{val:.3f} < {THRESHOLDS['wr_train_kill']} — random or worse")
    if val < THRESHOLDS["wr_train_target"]:
        return SignalResult(f"wr_train ({source})", val, SignalResult.STATUS_WARN,
                            f"{val:.3f} below target ≥ {THRESHOLDS['wr_train_target']}")
    return SignalResult(f"wr_train ({source})", val, SignalResult.STATUS_OK,
                        f"{val:.3f} ≥ {THRESHOLDS['wr_train_target']}")


def signal_wr_eval(eval_rows: list[dict[str, Any]]) -> SignalResult:
    """Final win_rate_ema (eval): KILL if < 0.15."""
    if not eval_rows:
        return SignalResult("wr_eval", None, SignalResult.STATUS_SKIP, "no eval rows")
    val = get_nested(eval_rows[-1], "isv_config.kelly.win_rate_ema")
    if val is None:
        val = get_nested(eval_rows[-1], "trading.win_rate")
        source = "trading.win_rate"
    else:
        source = "isv_config.kelly.win_rate_ema"
    if val is None or not isinstance(val, (int, float)):
        return SignalResult(f"wr_eval ({source})", None, SignalResult.STATUS_SKIP,
                            "field missing from final eval row")
    if val < THRESHOLDS["wr_eval_kill"]:
        return SignalResult(f"wr_eval ({source})", val, SignalResult.STATUS_KILL,
                            f"{val:.3f} < {THRESHOLDS['wr_eval_kill']} — random or worse")
    if val < THRESHOLDS["wr_eval_target"]:
        return SignalResult(f"wr_eval ({source})", val, SignalResult.STATUS_WARN,
                            f"{val:.3f} below target ≥ {THRESHOLDS['wr_eval_target']}")
    return SignalResult(f"wr_eval ({source})", val, SignalResult.STATUS_OK,
                        f"{val:.3f} ≥ {THRESHOLDS['wr_eval_target']}")


def signal_eval_pnl(eval_summary: dict[str, Any] | None,
                    eval_rows: list[dict[str, Any]]) -> SignalResult:
    """Final eval pnl: KILL if catastrophic (< -$3M).

    Prefer eval_summary.total_pnl_usd over per-step realized_pnl_usd_cum.
    Phase 0 reward-redesign (2026-06-02) made these two sources read the
    same per-trade `realised_pnl_usd_fp / 100` arithmetic — so they SHOULD
    now agree within $50 (enforced by `signal_consistency` below). Before
    Phase 0 they could disagree by hundreds of millions
    (`pearl_grwwh_eval_catastrophic_collapse`).
    """
    val: float | None = None
    source = "none"
    if eval_summary is not None and "total_pnl_usd" in eval_summary:
        val = float(eval_summary["total_pnl_usd"])
        source = "eval_summary.total_pnl_usd"
    elif eval_rows:
        # Phase 0 reward-redesign (2026-06-02): authoritative USD cum
        # (was `realized_pnl_cum_usd` — which actually wasn't USD).
        last = get_nested(eval_rows[-1], "trading.realized_pnl_usd_cum")
        if isinstance(last, (int, float)):
            val = float(last)
            source = "eval_diag last realized_pnl_usd_cum"
    if val is None:
        return SignalResult("eval_pnl", None, SignalResult.STATUS_SKIP,
                            "neither eval_summary.json nor eval_diag.jsonl has pnl")
    if val < THRESHOLDS["eval_pnl_kill"]:
        return SignalResult(f"eval_pnl ({source})", val, SignalResult.STATUS_KILL,
                            f"${val:,.0f} < ${THRESHOLDS['eval_pnl_kill']:,.0f} — catastrophic")
    if val < THRESHOLDS["eval_pnl_target"]:
        return SignalResult(f"eval_pnl ({source})", val, SignalResult.STATUS_WARN,
                            f"${val:,.0f} below target > ${THRESHOLDS['eval_pnl_target']:,.0f}")
    return SignalResult(f"eval_pnl ({source})", val, SignalResult.STATUS_OK,
                        f"${val:,.0f} > ${THRESHOLDS['eval_pnl_target']:,.0f}")


def signal_consistency(eval_rows: list[dict[str, Any]],
                       eval_summary: dict[str, Any] | None) -> SignalResult:
    """Per-step vs eval_summary consistency check (§3.6).

    Phase 0 reward-redesign (2026-06-02): KILL on $50+ disagreement between
    the per-step `trading.realized_pnl_usd_cum` (last eval row) and
    `eval_summary.total_pnl_usd`. Both sources are now computed from the
    SAME per-trade `realised_pnl_usd_fp / 100` arithmetic — the diag reads
    from `pnl_step_close_usd_d` (pnl_track.cu) and the summary reads
    `TradeRecord.realised_pnl_usd_fp` from the trade log. They MUST match
    within a single $0.01 unit; the $50 threshold is one $50/pt unit of
    slack for floating-point summation order across batches. A larger gap
    means one of the sources is broken — kill the run for investigation.
    """
    if eval_summary is None:
        return SignalResult("per-step vs eval_summary", None, SignalResult.STATUS_SKIP,
                            "eval_summary.json missing")
    if "total_pnl_usd" not in eval_summary:
        return SignalResult("per-step vs eval_summary", None, SignalResult.STATUS_SKIP,
                            "eval_summary.json has no total_pnl_usd")
    if not eval_rows:
        return SignalResult("per-step vs eval_summary", None, SignalResult.STATUS_SKIP,
                            "no eval rows")
    per_step = get_nested(eval_rows[-1], "trading.realized_pnl_usd_cum")
    if per_step is None or not isinstance(per_step, (int, float)):
        return SignalResult("per-step vs eval_summary", None, SignalResult.STATUS_SKIP,
                            "last eval row missing realized_pnl_usd_cum")
    summary = float(eval_summary["total_pnl_usd"])
    per_step = float(per_step)
    diff = abs(per_step - summary)
    detail = f"per_step=${per_step:,.2f} | eval_summary=${summary:,.2f} | gap=${diff:,.2f}"
    kill_usd = THRESHOLDS["consistency_kill_usd"]
    if diff > kill_usd:
        return SignalResult("per-step vs eval_summary", diff, SignalResult.STATUS_KILL,
                            f"{detail} > ${kill_usd:.0f} — eval pnl source disagreement, investigate")
    return SignalResult("per-step vs eval_summary", diff, SignalResult.STATUS_OK, detail)


# ── Driver ────────────────────────────────────────────────────────────


def run_verdict(out_dir: Path) -> tuple[str, list[SignalResult], dict[str, Any]]:
    """Compute all signals; return overall verdict + per-signal list + meta."""
    diag_path = out_dir / "diag.jsonl"
    eval_diag_path = out_dir / "eval_diag.jsonl"
    eval_summary_path = out_dir / "eval_summary.json"

    meta: dict[str, Any] = {
        "out_dir": str(out_dir),
        "diag_path": str(diag_path),
        "eval_diag_path": str(eval_diag_path),
        "eval_summary_path": str(eval_summary_path),
    }

    if not diag_path.exists():
        print(f"ERROR: {diag_path} not found", file=sys.stderr)
        return "INPUT_MISSING", [], meta

    train_rows = load_jsonl(diag_path)
    meta["n_train_rows"] = len(train_rows)
    if len(train_rows) < THRESHOLDS["min_train_rows"]:
        print(f"ERROR: only {len(train_rows)} train rows in {diag_path} (need ≥ {THRESHOLDS['min_train_rows']})", file=sys.stderr)
        return "INPUT_INSUFFICIENT", [], meta

    eval_rows = load_jsonl(eval_diag_path) if eval_diag_path.exists() else []
    meta["n_eval_rows"] = len(eval_rows)

    eval_summary: dict[str, Any] | None = None
    if eval_summary_path.exists():
        try:
            with eval_summary_path.open("r") as f:
                eval_summary = json.load(f)
        except (OSError, json.JSONDecodeError) as exc:
            print(f"WARN: failed to read {eval_summary_path}: {exc}", file=sys.stderr)
            eval_summary = None
    meta["eval_summary_present"] = eval_summary is not None

    signals: list[SignalResult] = [
        signal_action_entropy(train_rows),
        signal_q_pi_agree(train_rows),
        signal_reward_pnl_pearson(train_rows),
        signal_hold_trend(train_rows),
        signal_popart_sigma_cv(train_rows),
        signal_wr_train(train_rows),
        signal_wr_eval(eval_rows),
        signal_eval_pnl(eval_summary, eval_rows),
        signal_consistency(eval_rows, eval_summary),
    ]

    kill_reasons = [s.name for s in signals if s.status == SignalResult.STATUS_KILL]
    if kill_reasons:
        overall = "KILL_" + "+".join(s.replace(" ", "_") for s in kill_reasons)
    else:
        overall = "OK"
    return overall, signals, meta


def format_text(overall: str, signals: list[SignalResult], meta: dict[str, Any]) -> str:
    lines: list[str] = []
    lines.append("=" * 72)
    lines.append(f"Tier 1.5 verdict: {overall}")
    lines.append("=" * 72)
    lines.append(f"  out_dir:           {meta['out_dir']}")
    lines.append(f"  train rows:        {meta.get('n_train_rows', 0)}")
    lines.append(f"  eval rows:         {meta.get('n_eval_rows', 0)}")
    lines.append(f"  eval_summary:      {'present' if meta.get('eval_summary_present') else 'missing'}")
    lines.append("")
    lines.append(f"{'STATUS':<6}  {'SIGNAL':<42}  EXPLANATION")
    lines.append(f"{'-'*6}  {'-'*42}  {'-'*64}")
    for s in signals:
        marker = {
            SignalResult.STATUS_OK: "OK",
            SignalResult.STATUS_WARN: "WARN",
            SignalResult.STATUS_KILL: "KILL",
            SignalResult.STATUS_SKIP: "SKIP",
        }[s.status]
        lines.append(f"{marker:<6}  {s.name:<42}  {s.explain}")
    lines.append("")
    if overall == "OK":
        lines.append("[OK] No behavioral red flags. Cluster submit permitted.")
    elif overall.startswith("KILL_"):
        lines.append(f"[KILL] {overall} — DO NOT submit cluster smoke. Fix locally first.")
    else:
        lines.append(f"[{overall}] Inputs incomplete — re-run mid-smoke to generate full diag.")
    return "\n".join(lines)


def main() -> int:
    args = parse_args()
    overall, signals, meta = run_verdict(args.out_dir)

    if args.json:
        payload = {
            "overall": overall,
            "signals": [s.to_dict() for s in signals],
            "meta": meta,
        }
        print(json.dumps(payload, indent=2))
    elif args.quiet:
        print(overall)
    else:
        print(format_text(overall, signals, meta))

    if overall == "OK":
        return 0
    if overall.startswith("KILL_"):
        return 1
    return 2


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