#!/usr/bin/env python3 """PMOS portable kit — balanced metrics over the file-mode state layer (D54, D39). Reads pmos/state/{runs,evals,acceptances}.jsonl and prints: B1 north star — human corrections per ACCEPTED run (lower is better; D10, counted over accepted runs only so leniency can't deflate it) B2 throughput — accepted runs (guards against winning B1 by doing less) -- eval pass rate — share of evals passing the Review-Gate rule (see review_gate_pass) B4 calibration — evaluator verdict vs PM label, joined by run_id: raw agreement, TPR, TNR, Cohen's kappa (the D40 classifier method, computed locally). κ is broken out PER evaluator_model, with a mixed-model warning — a κ is a property of the evaluator model and is not inherited across a model swap (harness-model-agnosticism) -- Goodhart tripwire — fires when eval_pass_rate >= 0.8 while avg corrections >= 1.5 (the pass-happy-judge signal; anti-optimism: read the red first) No network, no secrets. Exit 0 always (a reporting tool, not a gate). Anti-optimism: reds print first. Import-safe (D64 Track D1): every helper is module-level and the report runs only under __main__, so kit/fixtures/test_parsers.py can import review_gate_pass + leading_frontmatter and prove the Python parsers agree with the JavaScript ones in workspace/lib/. """ import datetime, glob, json, os, re, sys KIT = os.environ.get("KIT_ROOT") or os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) STATE = os.path.join(KIT, "state") # ── CANONICAL Review-Gate pass rule — the MIRROR of workspace/lib/review-gate.js ────────────── # Rule: verdict ILIKE 'pass%' OR score >= 0.85. This is the single source of truth's Python half; # the JS half is workspace/lib/review-gate.js. kit/fixtures/test_parsers.py runs BOTH against # kit/fixtures/review-gate-pass-rule.json and fails if they ever disagree — the mirror can't drift. PASS_VERDICT_PREFIX = "pass" SCORE_THRESHOLD = 0.85 def review_gate_pass(verdict, score): """A recorded verdict is AUTHORITATIVE (pass iff it starts with "pass"); an explicit FAIL is never overridden by a high score. Only when NO verdict was recorded does the score decide. AMENDED 2026-07-26: was `verdict ILIKE 'pass%' OR score >= 0.85`, which let a FAIL at score 0.93 through the gate (blind-plant round 2, specimen be58). Mirror of workspace/lib/review-gate.js isReviewPass; agreement enforced by kit/fixtures/test_parsers.py.""" v = str("" if verdict is None else verdict).strip().lower() if v: return v.startswith(PASS_VERDICT_PREFIX) try: return float(score) >= SCORE_THRESHOLD except (TypeError, ValueError): return False def leading_frontmatter(path): """The LEADING frontmatter block only — body text cannot fake a declaration. Mirror of workspace/lib/file-mode-parse.js parseLeadingFrontmatter (agreement enforced by the fixture test). The value group is `(.*?)`, NOT `(.+?)`. A key written bare (`prd_path:` with nothing after it) is legal YAML for a null value; under `.+?` the line does not match, so it reads as the first body line, the loop breaks, and every key below it is dropped. Agreement with the JS parser was never sufficient to catch this — both carried the same wrong regex, so the fixture compared two identical mistakes and passed. See kit/fixtures/state/initiatives/epsilon-empty-value.md. The closing --- fence is TERMINAL, never skipped: stepping over it absorbed a key-shaped first body line into the block (zeta-fence-fake.md pins this). Fenceless files parse from line 1 and stop at the first non-key line. First value wins, and ONE pair of surrounding double quotes is stripped, exactly as the JS parsers do (eta-quoted-value.md pins the quoted case). """ fm = {} fence_open = False with open(path, encoding="utf-8") as fh: for raw in fh: line = raw.rstrip("\n") if line.startswith("---"): # Only the FIRST ---, before any key, opens the block; every later --- ends it. if fence_open or fm: break fence_open = True continue m = re.match(r"^([a-z_]+):\s*(.*?)\s*$", line) if m: val = m.group(2) if len(val) >= 2 and val.startswith('"') and val.endswith('"'): val = val[1:-1] fm.setdefault(m.group(1), val) elif line.strip(): break # frontmatter is the LEADING block only return fm def load(name): path = os.path.join(STATE, name) rows = [] if os.path.exists(path): for i, line in enumerate(open(path, encoding="utf-8"), 1): line = line.strip() if not line: continue try: rows.append(json.loads(line)) except json.JSONDecodeError: print(f"⚠ {name}:{i} is not valid JSON — skipped (append-only files must hold one JSON object per line)") return rows def confusion(rows, label_by_run): """(tp, tn, fp, fn) over evals joined to a pass/fail PM label by run_id, using the canonical rule.""" tp = tn = fp = fn = 0 for e in rows: lab = label_by_run.get(e.get("run_id")) if lab not in ("pass", "fail"): continue pred_pass = review_gate_pass(e.get("verdict"), e.get("score")) if pred_pass and lab == "pass": tp += 1 elif pred_pass and lab == "fail": fp += 1 elif not pred_pass and lab == "fail": tn += 1 else: fn += 1 return tp, tn, fp, fn def kappa_of(tp, tn, fp, fn): n = tp + tn + fp + fn if n == 0: return None po = (tp + tn) / n pe = (((tp + fp) * (tp + fn)) + ((fn + tn) * (fp + tn))) / (n * n) return None if pe == 1 else (po - pe) / (1 - pe) def main(): runs = load("runs.jsonl") evals = load("evals.jsonl") acc = load("acceptances.jsonl") accepted = [a for a in acc if a.get("accepted") is True] corrections_by_run = {r.get("run_id"): int(r.get("human_corrections") or 0) for r in runs} acc_corr = [corrections_by_run.get(a.get("run_id"), 0) for a in accepted] n_pass = sum(1 for e in evals if review_gate_pass(e.get("verdict"), e.get("score"))) pass_rate = (n_pass / len(evals)) if evals else None # --- B4: verdict-vs-PM-label confusion matrix, joined by run_id (D40) --- label_by_run = {a.get("run_id"): str(a.get("pm_label", "")).lower() for a in acc if a.get("pm_label")} tp, tn, fp, fn = confusion(evals, label_by_run) n = tp + tn + fp + fn k = kappa_of(tp, tn, fp, fn) avg_corr = (sum(corrections_by_run.values()) / len(corrections_by_run)) if corrections_by_run else None tripwire = pass_rate is not None and avg_corr is not None and pass_rate >= 0.8 and avg_corr >= 1.5 print(f"PMOS kit metrics — {len(runs)} run(s), {len(evals)} eval(s), {len(acc)} acceptance record(s)\n") if tripwire: print("🔴 GOODHART TRIPWIRE FIRED — evaluator passes >=80% while humans average >=1.5 corrections/run.") print(" The north star is being gamed; treat the Review Gate as NOT calibrated (D40).\n") if not accepted: print("🔴 B1 north star: UNMEASURABLE — no accepted runs yet (record acceptances to pmos/state/acceptances.jsonl).") else: ns = sum(acc_corr) / len(accepted) print(f"B1 north star: {ns:.2f} human corrections per accepted run (over {len(accepted)} accepted run(s); lower is better)") print(f"B2 throughput: {len(accepted)} accepted run(s)") print(f" eval pass rate: {'n/a — no evals yet' if pass_rate is None else f'{pass_rate:.0%} ({n_pass}/{len(evals)})'}") if n == 0: print("🔴 B4 calibration: UNMEASURABLE — no eval joined to a PM-labelled acceptance yet; the Review Gate is uncalibrated.") else: tpr = tp / (tp + fn) if (tp + fn) else None tnr = tn / (tn + fp) if (tn + fp) else None print(f"B4 calibration (n={n}): raw agreement {(tp + tn) / n:.0%}, " f"TPR {'n/a' if tpr is None else f'{tpr:.0%}'}, " f"TNR {'n/a — no PM fails seen (leniency blind spot)' if tnr is None else f'{tnr:.0%}'}, " f"kappa {'undefined' if k is None else f'{k:.2f}'} (target >= 0.60 — verdict ADVISORY below it, D40)") # --- per-evaluator-model calibration (harness-model-agnosticism) ------------- labeled_evals = [e for e in evals if label_by_run.get(e.get("run_id")) in ("pass", "fail")] by_model = {} for e in labeled_evals: by_model.setdefault(e.get("evaluator_model") or "unknown", []).append(e) if len(by_model) > 1: print(f"\n🔴 MIXED-MODEL CALIBRATION — labeled evals span {len(by_model)} evaluator models; the " "aggregate κ above POOLS them and is not any single evaluator's calibration (κ is per " "evaluator model, D40). Per model:") for m in sorted(by_model): c = confusion(by_model[m], label_by_run) km = kappa_of(*c) print(f" - {m}: n={sum(c)}, kappa {'undefined' if km is None else f'{km:.2f}'} " f"(tp{c[0]} tn{c[1]} fp{c[2]} fn{c[3]})") elif by_model and set(by_model) == {"unknown"}: print("\n⚠ evaluator_model unrecorded on every labeled eval — κ cannot be attributed to a model " "and silently pools across any model swaps. Pass the evaluator model as the last arg to " "`pmos-log.sh eval` (harness-model-agnosticism).") # --- D57 outcome checks (accepted-but-unvalidated; due prints red) ----------- checks = load("outcome_checks.jsonl") latest_disp = {} for c in checks: if c.get("initiative_id"): latest_disp[c["initiative_id"]] = str(c.get("disposition", "")).lower() declared = [] for path in sorted(glob.glob(os.path.join(STATE, "initiatives", "*.md"))): fm = leading_frontmatter(path) if not fm.get("id"): if fm.get("outcome_metric"): print(f"⚠ {os.path.basename(path)} declares an outcome but has no id: — it can never be disposed; fix the frontmatter") continue if fm.get("outcome_metric"): declared.append(fm) if declared: today = datetime.date.today().isoformat() undisposed = [f for f in declared if f.get("id") not in latest_disp] # a missing horizon errs RED (counts as due): a bet with no judgment date can never come due # by itself — horizon-gaming-by-omission (mirrors the hosted CHECK requiring a horizon). due = [f for f in undisposed if str(f.get("outcome_horizon", "0000-00-00")) < today] pending = [f for f in undisposed if f not in due] print() if due: print(f"🔴 D57 outcome checks DUE ({len(due)}) — accepted bets past horizon, never validated:") for f in due: print(f" - {f.get('id')}: \"{f.get('outcome_metric')}\" (horizon {f.get('outcome_horizon', '?')})" " -> dispose validated|refuted|inconclusive to state/outcome_checks.jsonl") refuted = sum(1 for v in latest_disp.values() if v == "refuted") print(f"D57 outcomes: {len(declared)} declared — {len(pending)} pending, {len(due)} due, " f"{len(latest_disp)} disposed" + (f" ({refuted} refuted — recorded, not hidden)" if refuted else "")) if __name__ == "__main__": main()