"""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())