arborist/bench/scripts/fivef_threshold_calibration.py

"""5S/5T/5F → v8 ForkScore threshold-calibration handoff (#000025 §10.14).

Closure deliverable for ticket #000025 §10.14 ("threshold calibration for
v8 selection acceptance handed off to #000012"). Pure measurement — runs
the canonical 5S / 5T / 5F sub-batteries (embedded packs + the 5F live
packs), builds the parent/child metrics bundle the ForkScore reads
(:func:`arborist.substrate.fork_score.bench_result_to_metrics` shape),
and reports:

  1. Per-pack baseline rate (the ``_BATTERY_RATE_KEYS`` metric the scorer
     consumes) + fixture count + 1/n observability granularity.
  2. The identity-fork verdict — ``fork_score(parent, parent)`` — which
     tells #000012 what "no change" scores to (it's MARGINAL, not
     ACCEPT, because every Δ-rate term is 0).
  3. Synthetic-perturbation verdicts: a single +``SIGNAL_FLOOR`` bump on
     one 5F sub-battery → ACCEPT; a single −``HARD_REGRESSION_FLOOR``
     drop → REJECT. Confirms the floor constants do what the docstrings
     claim.
  4. A recommendation block: are ``SIGNAL_FLOOR`` / ``HARD_REGRESSION_FLOOR``
     (both 0.05 today) appropriate given the observed pack granularity?

No mutation, no LLM call, no schema / governance-hash / canonicalization
change. The markdown report is the deliverable; #000012 cites it.
"""

from __future__ import annotations

import argparse
import statistics
import sys
from dataclasses import asdict
from datetime import datetime, timezone
from pathlib import Path

# bench/ is on sys.path when run as a module (python -m); when run as a
# script, add the repo root so `import bench...` and `import arborist...`
# resolve.
_REPO_ROOT = Path(__file__).resolve().parents[2]
if str(_REPO_ROOT) not in sys.path:
    sys.path.insert(0, str(_REPO_ROOT))

from bench.batteries import b_5f, b_5s, b_5t  # noqa: E402
from bench.batteries.base import BatteryResult  # noqa: E402
from arborist.substrate.fork_score import (  # noqa: E402
    HARD_REGRESSION_FLOOR,
    SIGNAL_FLOOR,
    _BATTERY_RATE_KEYS,
    bench_result_to_metrics,
    fork_score,
)
from arborist.substrate.weights import DELTA_AGGREGATORS, WeightSet  # noqa: E402

_FIX = _REPO_ROOT / "bench" / "fixtures"

# The canonical sub-battery → fixture-pack map the ForkScore Δ-rate terms
# read. Mirrors bench.batteries.runner._DEFAULT_FIXTURES but restricted
# to the keys present in fork_score._BATTERY_RATE_KEYS (so legacy 5t
# "transfer" is excluded — the scorer drops it). The "live" 5F packs are
# included as a separate column because their runners exercise real
# arborist surfaces on temp shards (Phase 1b.2).
_EMBEDDED_PACKS: dict[tuple[str, str], Path] = {
    ("5s", "syntax"): _FIX / "5s" / "syntax-v1.jsonl",
    ("5s", "semantics"): _FIX / "5s" / "semantics-v1.jsonl",
    ("5s", "syllogism"): _FIX / "5s" / "syllogism-v1.jsonl",
    ("5s", "synthesis"): _FIX / "5s" / "synthesis-v1.jsonl",
    ("5s", "semiotics"): _FIX / "5s" / "semiotics-v1.jsonl",
    ("5t", "transfer-learning"): _FIX / "5t" / "transfer-learning-v2.jsonl",
    ("5t", "triangulation"): _FIX / "5t" / "triangulation-v1.jsonl",
    ("5t", "truthtables"): _FIX / "5t" / "truthtables-v1.jsonl",
    ("5t", "transitivity"): _FIX / "5t" / "transitivity-v1.jsonl",
    ("5t", "time"): _FIX / "5t" / "time-v1.jsonl",
    ("5f", "function"): _FIX / "5f" / "function-v1.jsonl",
    ("5f", "finetuning"): _FIX / "5f" / "finetuning-v1.jsonl",
    ("5f", "falsification"): _FIX / "5f" / "falsification-v1.jsonl",
    ("5f", "formulate"): _FIX / "5f" / "formulate-v1.jsonl",
    ("5f", "feedback-loop"): _FIX / "5f" / "feedback-loop-v1.jsonl",
}

_LIVE_PACKS: dict[tuple[str, str], Path] = {
    ("5f", "function"): _FIX / "5f" / "function-live-v1.jsonl",
    ("5f", "finetuning"): _FIX / "5f" / "finetuning-live-v1.jsonl",
    ("5f", "falsification"): _FIX / "5f" / "falsification-live-v1.jsonl",
    ("5f", "formulate"): _FIX / "5f" / "formulate-live-v1.jsonl",
    ("5f", "feedback-loop"): _FIX / "5f" / "feedback-loop-live-v1.jsonl",
}

_SUB_RUNNERS = {"5s": b_5s.SUB_BATTERIES, "5t": b_5t.SUB_BATTERIES, "5f": b_5f.SUB_BATTERIES}


def _run_pack(battery: str, sub: str, path: Path) -> BatteryResult:
    return _SUB_RUNNERS[battery][sub](path)


def _results_payload(packs: dict[tuple[str, str], Path]) -> dict:
    """Run every pack and shape the output like bench.batteries.runner."""
    results = []
    for (battery, sub), path in packs.items():
        r = _run_pack(battery, sub, path)
        results.append(asdict(r))
    return {"schema_version": "bench-result-v1", "results": results}


def _rate_of(metrics: dict, battery: str, sub: str) -> float | None:
    key = _BATTERY_RATE_KEYS.get(battery, {}).get(sub)
    if key is None:
        return None
    v = metrics.get(key)
    return float(v) if v is not None else None


def _deepcopy_bundle(metrics_bundle: dict) -> dict:
    return {b: {s: dict(m) for s, m in subs.items()} for b, subs in metrics_bundle.items()}


def _set_rate(bundle: dict, battery: str, sub: str, value: float) -> None:
    key = _BATTERY_RATE_KEYS[battery][sub]
    bundle.setdefault(battery, {}).setdefault(sub, {})[key] = max(0.0, min(1.0, value))


def _shift_rate(bundle: dict, battery: str, sub: str, delta: float) -> None:
    key = _BATTERY_RATE_KEYS[battery][sub]
    base = float(bundle.get(battery, {}).get(sub, {}).get(key, 0.0))
    _set_rate(bundle, battery, sub, base + delta)


def _degraded_parent(parent: dict, knock_to: float = 0.90) -> dict:
    """A synthetic parent with every 5F sub-battery rate knocked down to
    ``knock_to`` — gives a child headroom to improve. The production
    parent is at ceiling (rate 1.0 on every pack), so threshold
    mechanics can only be exercised against a below-ceiling baseline."""
    out = _deepcopy_bundle(parent)
    for sub in _BATTERY_RATE_KEYS["5f"]:
        _set_rate(out, "5f", sub, knock_to)
    return out


def _fmt_verdict(sf) -> str:
    flags = f" — flags: {', '.join(sf.flags)}" if sf.flags else ""
    return f"{sf.verdict} (score {sf.score:+.4f}){flags}"


def build_report() -> str:
    now = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
    embedded = _results_payload(_EMBEDDED_PACKS)
    live = _results_payload(_LIVE_PACKS)
    parent = bench_result_to_metrics(embedded)
    live_metrics = bench_result_to_metrics(live)

    # ---- per-pack table rows -------------------------------------------------
    rows = []
    granularities: list[float] = []
    for r in embedded["results"]:
        battery, sub = r["battery"], r["sub_battery"]
        n = r["pass_count"] + r["fail_count"]
        rate = _rate_of(r["metrics"], battery, sub)
        gran = (1.0 / n) if n else float("nan")
        if n:
            granularities.append(gran)
        live_r = next(
            (lr for lr in live["results"] if lr["battery"] == battery and lr["sub_battery"] == sub),
            None,
        )
        live_rate = _rate_of(live_r["metrics"], battery, sub) if live_r else None
        rows.append((battery, sub, n, rate, gran, live_rate))

    coarsest = max(granularities) if granularities else float("nan")
    finest = min(granularities) if granularities else float("nan")

    # ---- fork-score sanity verdicts ------------------------------------------
    # The production parent is at ceiling (rate 1.0 everywhere), so a
    # "+SIGNAL_FLOOR" child clamps right back to 1.0 and shows nothing.
    # Exercise the threshold mechanics against a synthetic below-ceiling
    # parent (every 5F sub knocked to 0.90).
    identity = fork_score(parent, parent)
    knock = 0.90
    degraded = _degraded_parent(parent, knock)

    # Child recovers EVERY 5F sub uniformly by 0.06 (> SIGNAL_FLOOR each)
    # → mean Δ5F = 0.06 → score = γ·0.06 = 0.06 ≥ SIGNAL_FLOOR → ACCEPT.
    child_uniform = _deepcopy_bundle(degraded)
    for sub in _BATTERY_RATE_KEYS["5f"]:
        _shift_rate(child_uniform, "5f", sub, 0.06)
    accept_uniform = fork_score(degraded, child_uniform)

    # Child recovers only ONE 5F sub by 0.06 → mean Δ5F = 0.06/5 = 0.012
    # → score 0.012 < SIGNAL_FLOOR → MARGINAL. This is the 5× averaging
    # dilution: a single-sub gain is worth a fifth of its face value.
    child_single = _deepcopy_bundle(degraded)
    _shift_rate(child_single, "5f", "function", 0.06)
    marginal_single = fork_score(degraded, child_single)

    # Child drops one 5F sub by HARD_REGRESSION_FLOOR from the degraded
    # parent → hard regression → REJECT regardless of the other terms.
    child_regress = _deepcopy_bundle(degraded)
    _shift_rate(child_regress, "5f", "falsification", -HARD_REGRESSION_FLOOR)
    reject_regress = fork_score(degraded, child_regress)

    # ---- emit ----------------------------------------------------------------
    out: list[str] = []
    out.append("# 5S/5T/5F → v8 ForkScore threshold-calibration handoff")
    out.append("")
    out.append(f"**Date:** {now}")
    out.append("**Ticket:** #000025 §10.14 (closure deliverable) — handoff to #000012.")
    out.append(
        "**Method:** ran the canonical 5S/5T/5F sub-batteries (the packs "
        "`fork_score._BATTERY_RATE_KEYS` reads) plus the 5F live packs; "
        "computed baseline rates, observability granularity (1/n), and "
        "ran `fork_score` on the parent vs three synthetic child "
        "perturbations. Pure measurement."
    )
    out.append("")
    out.append("## 1. Baseline rates + granularity")
    out.append("")
    out.append(
        "Each row's *rate* is the single metric `fork_score` consumes for "
        "that sub-battery (`_BATTERY_RATE_KEYS`). *Granularity* = 1/n — "
        "the smallest rate change a single fixture flip can produce, i.e. "
        "the finest Δ the scorer could ever observe on that pack."
    )
    out.append("")
    out.append("| battery | sub-battery | fixtures (n) | embedded rate | live rate | granularity (1/n) |")
    out.append("|---|---|---|---|---|---|")
    for battery, sub, n, rate, gran, live_rate in rows:
        rate_s = f"{rate:.4f}" if rate is not None else "—"
        live_s = f"{live_rate:.4f}" if live_rate is not None else "—"
        out.append(f"| {battery} | {sub} | {n} | {rate_s} | {live_s} | {gran:.4f} ({gran*100:.1f}pp) |")
    out.append("")
    # Per-battery mean rate.
    for battery in ("5s", "5t", "5f"):
        vals = [rate for b, s, n, rate, g, lr in rows if b == battery and rate is not None]
        if vals:
            out.append(f"- **{battery.upper()} mean baseline rate:** {statistics.fmean(vals):.4f}")
    out.append(f"- **Coarsest pack granularity:** {coarsest:.4f} ({coarsest*100:.1f}pp)")
    out.append(f"- **Finest pack granularity:** {finest:.4f} ({finest*100:.1f}pp)")
    out.append("")
    out.append("## 2. Identity-fork verdict (no change)")
    out.append("")
    out.append(f"`fork_score(parent, parent)` → **{_fmt_verdict(identity)}**")
    out.append("")
    out.append(
        "Every 5S/5T/5F pack is at ceiling (rate 1.0) at HEAD, so every "
        "Δ-rate term is exactly 0 and the identity fork lands in the "
        "`[0, SIGNAL_FLOOR)` band → **MARGINAL**, not ACCEPT. Takeaway for "
        "#000012: an unchanged child is *not* auto-accepted; positive "
        "score has to come from a Δ-rate gain (needs harder fixtures, see "
        "§4), an efficiency-bonus increase (`adaptation_efficiency` / "
        "`feedback_efficiency`), or one of the non-bench terms "
        "(`selfmodel_calibration_gain`, `audit_completeness`, …)."
    )
    out.append("")
    out.append("## 3. Floor-constant sanity checks")
    out.append("")
    out.append(
        f"`SIGNAL_FLOOR = {SIGNAL_FLOOR}` · `HARD_REGRESSION_FLOOR = "
        f"{HARD_REGRESSION_FLOOR}` (current `arborist/substrate/fork_score.py`). "
        f"The production parent is at ceiling, so these are exercised against "
        f"a synthetic *degraded parent* — every 5F sub-battery rate knocked "
        f"to {knock:.2f}, giving a child headroom to improve."
    )
    out.append("")
    out.append("| scenario | verdict |")
    out.append("|---|---|")
    out.append(f"| identity — `fork_score(HEAD, HEAD)` | {_fmt_verdict(identity)} |")
    out.append(f"| degraded parent → child recovers **all 5** 5F subs by +0.06 | {_fmt_verdict(accept_uniform)} |")
    out.append(f"| degraded parent → child recovers **only 1** 5F sub by +0.06 | {_fmt_verdict(marginal_single)} |")
    out.append(f"| degraded parent → child drops one 5F sub by −{HARD_REGRESSION_FLOOR} | {_fmt_verdict(reject_regress)} |")
    out.append("")
    out.append(
        "Row 2 vs row 3 is the **5× averaging dilution**: `_delta_5f` means "
        "over all 5 sub-batteries, so the same +0.06 gain scores `1.0·0.06 = "
        "0.06` (ACCEPT) when applied to all five subs but only `1.0·(0.06/5) "
        "= 0.012` (MARGINAL) when applied to one. Each battery contributes a "
        "separate Δ-term at weight 1.0, so a uniform +`SIGNAL_FLOOR` across "
        "*every* sub of *all three* batteries would score `~0.15`."
    )
    out.append("")
    out.append("## 4. Recommendation for #000012")
    out.append("")
    out.append(
        f"- **Keep `SIGNAL_FLOOR = {SIGNAL_FLOOR}` and "
        f"`HARD_REGRESSION_FLOOR = {HARD_REGRESSION_FLOOR}`.** They match "
        "`docs/bench-maxing.md`'s 5-pp signal floor and the sanity checks "
        "in §3 behave as documented."
    )
    out.append(
        f"- **Granularity caveat (load-bearing):** the foundational 5S "
        f"packs are *coarser* than the floors — `syntax` (n=10 → 10.0pp) "
        f"and `semantics` (n=8 → 12.5pp). On those packs a single fixture "
        f"flip is ≥ {HARD_REGRESSION_FLOOR*100:.0f}pp, so *any* regression "
        f"there trips the hard-reject. That is the intended zero-tolerance "
        f"behaviour on the syntax/semantics base — not a bug — but #000012 "
        f"should document it: HARD_REGRESSION_FLOOR is not a tunable knob "
        f"on the small packs, it's effectively 'one fixture'. The n=30 / "
        f"n=50 / n=62 packs have 3.3pp / 2.0pp / 1.6pp granularity, so "
        f"there {HARD_REGRESSION_FLOOR*100:.0f}pp ≈ 1.5–3 fixture flips."
    )
    out.append(
        "- **5× averaging dilution (document this in #000012):** a single "
        "sub-battery's rate gain is worth a fifth of its face value because "
        "`_delta_5{s,t,f}` means over 5 subs. So 'a fork must improve "
        f"by `SIGNAL_FLOOR`' really means *one of*: ~`{SIGNAL_FLOOR*5:.2f}` "
        f"on a single sub, ~`{SIGNAL_FLOOR:.2f}` uniform across one "
        f"battery's 5 subs, or ~`{SIGNAL_FLOOR/3:.3f}` uniform across the "
        "whole 15-sub suite. If #000012 wants single-sub improvements to "
        "weigh equally it should switch `_delta_*` from mean to "
        "max-or-sum — but that's a #000012-owned design call, not a "
        "calibration finding."
    )
    out.append(
        "- **Ceiling saturation:** until harder fixtures drop a pack's "
        "baseline below 1.0, the `α·Δ5s + β·Δ5t + γ·Δ5f` terms can only be "
        "≤ 0. ForkScore acceptance at the current pack difficulty is "
        "driven by the efficiency bonuses + non-bench terms. If #000012 "
        "wants the bench Δ-rate terms to carry real positive signal, the "
        "5S/5T/5F packs need a harder tier (or a deliberately-degraded "
        "parent baseline) — track that as a #000025 follow-up, not a "
        "#000012 blocker."
    )
    out.append(
        "- **No constant change shipped by this calibration.** This is a "
        "handoff document; if #000012 decides to move a floor it owns that "
        "edit (and the resulting `governance_policy_hash` is unaffected — "
        "ForkScore constants don't fold into it; they're estimator "
        "parameters pinned by `ESTIMATOR_VERSION`)."
    )
    out.append("")

    # ---- §5. aggregator comparison on the #000046 below-ceiling pack --------
    out.append("## 5. `delta_aggregator` comparison (#000047, on the #000046 below-ceiling pack)")
    out.append("")
    hard_pack = _FIX / "5f" / "falsification-hard-v1.jsonl"
    if hard_pack.exists():
        hr = b_5f.run_falsification(hard_pack).metrics["error_detection_rate"]
        # parent = hard-pack rate on 5f/falsification; child = a verifier
        # tightening lifting it to 1.0 (a single-sub gain of (1 - hr)).
        p_hard = {"5s": {}, "5t": {}, "5f": {"falsification": {"error_detection_rate": hr}}}
        c_hard = {"5s": {}, "5t": {}, "5f": {"falsification": {"error_detection_rate": 1.0}}}
        out.append(
            f"Below-ceiling baseline: `5f/falsification` at the "
            f"`bench/fixtures/5f/falsification-hard-v1.jsonl` rate "
            f"**{hr:.4f}** ({hr*12:.0f}/12 — `verify_quotes` over-grounds "
            f"the rest); a child fork that tightens `verify_quotes` lifts "
            f"it toward 1.0 — a *single-sub* gain of {1.0-hr:.4f}. How that "
            f"single-sub gain scores under each aggregator:"
        )
        out.append("")
        out.append("| aggregator | `γ·Δ5f` | verdict | note |")
        out.append("|---|---|---|---|")
        notes = {
            "mean": "single-sub gain diluted 1/5 (still clears the floor here — the gain is large; a smaller fix would land MARGINAL, see §3)",
            "max": "single-sub gain at face value",
            "sum": "= max here (one sub); diverges from max only on a broad multi-sub gain",
        }
        for agg in DELTA_AGGREGATORS:
            sf = fork_score(p_hard, c_hard, weights=WeightSet(delta_aggregator=agg))
            out.append(
                f"| `{agg}` | {sf.breakdown['gamma_x_delta_5f']:+.4f} | "
                f"{sf.verdict} | {notes[agg]} |"
            )
        out.append("")
        out.append(
            "Reading: `mean` (default) makes a single-sub verifier fix worth "
            "≈ a fifth of its face value — so closing #000046 (lifting one "
            "5F sub) is rewarded modestly, while a broad cross-sub "
            "improvement is rewarded fully; `max`/`sum` flip that. The "
            "default stays `mean` (the conservative, noise-robust, "
            "regression-symmetric choice — `docs/bench-maxing.md`'s per-rate "
            "floor framing); #000012 can pick `max`/`sum` per-deployment via "
            "`WeightSet(delta_aggregator=...)` (recorded in "
            "`ScoredFork.weights`). #000047 ships the knob, not a default "
            "change."
        )
    else:
        out.append("_(hard pack not present — skipping; see #000046.)_")
    out.append("")
    return "\n".join(out)


def main(argv: list[str] | None = None) -> int:
    p = argparse.ArgumentParser(description=__doc__.splitlines()[0])
    p.add_argument(
        "--out",
        type=Path,
        default=None,
        help="Write the markdown report here (default: stdout).",
    )
    args = p.parse_args(argv)
    md = build_report()
    if args.out:
        args.out.parent.mkdir(parents=True, exist_ok=True)
        args.out.write_text(md, encoding="utf-8")
        print(f"wrote {args.out}")
    else:
        print(md)
    return 0


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