Chaim
c7156b575e
Make the effort-calibration recommendation style-clean (style acquisition is
the goal — write like chair Dafna Tamir):
- recommend_effort ranks by anti_pattern_total (PRIMARY, the clean style-rule
violation count) → golden_ratio_deviation_pp (SECONDARY, structural-style,
None=worst) → distance (faint final tiebreak). change_percent is EXCLUDED from
the ranking (it mixes style with content-completeness per 07-learning §0.7 —
noise for STYLE) and kept in the report as context only.
- Add a confidence flag ("clear"/"weak"): a pick whose anti-pattern lead over the
runner-up is within noise (margin < epsilon = max(0.5, 0.20·spread)) is flagged
⚠️ weak, so an over-claimed pick (the old confident block-zayin "low") is never
shown as confident. Rendered as a confidence column in the report.
- Add --rerank <report.json>: re-apply the new key to a saved calibration JSON
offline (NO LLM, NO DB), print + write *-reranked.md. Re-ranks today's 8-final
run without regenerating.
- recommend_effort stays a single shared pure function reused by
scripts/calibrate_block_yod.py (G2); both --self-test suites pass.
- Spec note added (07-learning §0.7) + SCRIPTS.md updated.
Invariants: G8 (style-clean empirical ranking), G2 (single shared recommend_effort,
reused by the block-yod harness — no parallel metric path), INV-LRN5 (style-only
signal; change% reported-not-ranked).
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
678 lines
35 KiB
Python
678 lines
35 KiB
Python
#!/usr/bin/env python3
|
||
"""#208 (WS5 / Q1, INV-G8 eval-harness) — model×effort calibration vs the finals.
|
||
|
||
Empirically picks the per-block reasoning `effort` whose regenerated block lands
|
||
CLOSEST to עו"ד דפנה תמיר's signed final, over the EXISTING `draft_final_pairs`
|
||
ledger. This is the calibration INV-G8 / 07-learning §0.7 ask for: stop choosing
|
||
the per-block effort defaults (#204: ה=medium, ו=medium, ז=high, ח=medium, ט=high)
|
||
"by feel".
|
||
|
||
WHAT IT MEASURES (per (case, block, effort) cell — the A/B grid):
|
||
Regenerate `block` for `case` via block_writer.write_block(effort_override=…)
|
||
(the PRODUCTION generation path → claude_session.query → `claude -p`, pinned
|
||
Opus 4.8, local-only), then score the regenerated block against the matching
|
||
SECTION of the final via services.style_distance.block_distance_to_final:
|
||
• change_percent — word-diff regen↔final-section (compute_diff_stats)
|
||
• anti_pattern_total — lessons.ANTI_PATTERNS hits in the regen (cleanest
|
||
style signal — see §0.7 warning below)
|
||
• golden_ratio_deviation_pp — structural-weight gap vs the final
|
||
• distance — normalized composite (lower = closer to Dafna)
|
||
No parallel metric path: it reuses style_distance + learning_loop (G2).
|
||
|
||
RECOMMENDATION (#213 — STYLE-CLEAN, anti-primary): for each block, the effort is
|
||
ranked by `anti_pattern_total` FIRST (the clean Dafna-style-rule-violation count),
|
||
then `golden_ratio_deviation_pp` (structural-style), and `distance` only as a faint
|
||
final tiebreak. `change_percent` is REPORTED-NOT-RANKED: per 07-learning §0.7 it
|
||
mixes style with content-completeness (a case's missing facts the model can't know)
|
||
→ noise for STYLE. The pick carries a `confidence` flag ("clear"/"weak"): a pick
|
||
whose anti-pattern lead over the runner-up is within noise is flagged "weak" so an
|
||
over-claimed pick (the old block-zayin "low") is never shown as confident. Reported
|
||
next to the #204 current default so a regression/improvement is visible.
|
||
|
||
⚠️ SAMPLE-SIZE CAVEAT (honored, not hidden): very few cases have an uploaded final
|
||
(draft_final_pairs.final_text non-empty). The report prints n_finals PROMINENTLY and
|
||
labels the output DIRECTIONAL EVIDENCE, not a regression. With n<3 per block the
|
||
recommendation is advisory only; the chair/operator decides whether to adopt.
|
||
|
||
⚠️ change_percent mixes style with content completeness (07-learning §0.7): the chair
|
||
sometimes doubles length for missing substance. anti_pattern_total is the cleaner
|
||
style signal — the report surfaces both, and the composite down-weights neither
|
||
silently.
|
||
|
||
GENERATION PATH (do not violate — reference_claude_generation_path / claude_session
|
||
docstring): write_block → claude_session.query → `claude -p` uses the local claude.ai
|
||
session. It runs ONLY on the host where the `claude` CLI exists (NOT the legal-ai
|
||
container, NOT a symlinked worktree without CLI access). Hence the live A/B is
|
||
host-only; --self-test proves the measurement logic offline with zero model calls.
|
||
|
||
Usage (mcp-server venv; live needs POSTGRES + the `claude` CLI on the host):
|
||
PY=/home/chaim/legal-ai/mcp-server/.venv/bin/python
|
||
$PY scripts/calibrate_effort.py --self-test # offline proof, no DB/CLI
|
||
POSTGRES_PASSWORD=… POSTGRES_HOST=127.0.0.1 POSTGRES_PORT=5433 \
|
||
$PY scripts/calibrate_effort.py # live A/B over finals
|
||
… --efforts low,medium,high,xhigh # override the effort grid
|
||
… --blocks block-he,block-vav # restrict to some blocks
|
||
… --case 8137-11-24 # a single case
|
||
… --repeats 2 # avg N gens/cell (noise)
|
||
… --dry-run # plan the grid, no model calls
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import asyncio
|
||
import json
|
||
import logging
|
||
import os
|
||
import sys
|
||
from datetime import datetime
|
||
from pathlib import Path
|
||
from zoneinfo import ZoneInfo
|
||
from statistics import mean
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
REPO_ROOT = Path(__file__).resolve().parent.parent
|
||
sys.path.insert(0, str(REPO_ROOT / "mcp-server" / "src"))
|
||
|
||
if "POSTGRES_URL" not in os.environ:
|
||
os.environ["POSTGRES_URL"] = (
|
||
f"postgres://{os.environ.get('POSTGRES_USER','legal_ai')}:"
|
||
f"{os.environ.get('POSTGRES_PASSWORD','')}@"
|
||
f"{os.environ.get('POSTGRES_HOST','127.0.0.1')}:"
|
||
f"{os.environ.get('POSTGRES_PORT','5433')}/"
|
||
f"{os.environ.get('POSTGRES_DB','legal_ai')}"
|
||
)
|
||
|
||
OUT_DIR = REPO_ROOT / "data" / "eval"
|
||
|
||
# Only the AI blocks that map to a golden-ratio section can be scored against the
|
||
# final's matching section (split_final_by_section). Template blocks (א-ד, יב) are
|
||
# deterministic template-fill — no effort knob. block-yod (discussion) is EXCLUDED
|
||
# from the defaults: it requires an approved direction (brainstorm → approve_direction)
|
||
# that calibration cases lack, so block_writer.write_block raises and it is NOT
|
||
# calibratable standalone — and it's out of WS5's interim-draft scope. It stays in
|
||
# VALID_BLOCKS so a user can still force it (--blocks block-yod) at their own risk;
|
||
# the per-cell guard in _run() keeps such a failure non-fatal.
|
||
CALIBRATABLE_BLOCKS = ["block-he", "block-vav", "block-zayin", "block-chet", "block-tet", "block-yod-alef"]
|
||
VALID_BLOCKS = CALIBRATABLE_BLOCKS + ["block-yod"]
|
||
DEFAULT_EFFORTS = ["low", "medium", "high", "xhigh"]
|
||
VALID_EFFORTS = {"low", "medium", "high", "xhigh", "max"}
|
||
|
||
|
||
# ── pure helpers (offline-testable) ──────────────────────────────────────────
|
||
# Style-clean ranking epsilon (#213). Confidence in a pick = how far the best
|
||
# effort's anti_pattern_total beats the 2nd-best. We call a pick "clear" only when
|
||
# that margin clears a floor that is itself the LARGER of:
|
||
# • an absolute floor (≥0.5 avg anti-patterns — a half-violation per case), and
|
||
# • a fraction of the spread across efforts (best can't claim a lead that is mere
|
||
# noise on a block whose efforts barely differ).
|
||
# Below that → "weak" (within noise; nominal pick still reported, but flagged so a
|
||
# within-noise pick is never shown as confident). Documented in the docstring.
|
||
CONFIDENCE_ABS_FLOOR = 0.5 # avg anti-patterns (a half-violation per case)
|
||
CONFIDENCE_SPREAD_FRACTION = 0.20 # of the across-effort anti spread
|
||
|
||
|
||
def _anti_confidence(rows: list[dict]) -> tuple[str, float, float]:
|
||
"""Confidence that the anti-primary pick is real, not within-noise.
|
||
|
||
Returns (confidence, margin, epsilon) where confidence ∈ {"clear","weak"}.
|
||
margin = (2nd-best anti) − (best anti); epsilon = the bar margin must clear.
|
||
A single effort (no rival) is trivially "clear". Pure → unit-tested.
|
||
"""
|
||
if len(rows) < 2:
|
||
return "clear", float("inf"), 0.0
|
||
antis = sorted(r["anti_pattern_total"] for r in rows)
|
||
best, second = antis[0], antis[1]
|
||
margin = second - best
|
||
spread = antis[-1] - antis[0]
|
||
epsilon = max(CONFIDENCE_ABS_FLOOR, CONFIDENCE_SPREAD_FRACTION * spread)
|
||
return ("clear" if margin >= epsilon else "weak"), round(margin, 4), round(epsilon, 4)
|
||
|
||
|
||
def recommend_effort(cells: list[dict]) -> dict | None:
|
||
"""Pick the STYLE-CLEANEST effort for ONE block from its scored cells (#213).
|
||
|
||
cells: [{"effort","distance","anti_pattern_total","change_percent",
|
||
"golden_ratio_deviation_pp","n"}].
|
||
|
||
Ranking key (style-clean, G8 / 07-learning §0.7) — change_percent is EXCLUDED
|
||
from the ranking (it mixes style with content-completeness; reported only):
|
||
1. PRIMARY anti_pattern_total (↑ fewer Dafna-style violations = better)
|
||
2. SECONDARY golden_ratio_deviation_pp (↑ structural-style; None → +inf worst)
|
||
3. TIEBREAK distance (↑ faint final tiebreak only)
|
||
|
||
Returns the winning row dict (so callers reading ["effort"] keep working) with
|
||
two added keys: ``confidence`` ∈ {"clear","weak"} and ``confidence_margin`` —
|
||
so a within-noise pick is never presented as confident. Pure → unit-tested.
|
||
"""
|
||
if not cells:
|
||
return None
|
||
ranked = sorted(
|
||
cells,
|
||
key=lambda c: (
|
||
c["anti_pattern_total"],
|
||
c["golden_ratio_deviation_pp"] if c.get("golden_ratio_deviation_pp") is not None else float("inf"),
|
||
c["distance"],
|
||
),
|
||
)
|
||
best = dict(ranked[0])
|
||
conf, margin, _eps = _anti_confidence(cells)
|
||
best["confidence"] = conf
|
||
best["confidence_margin"] = margin
|
||
return best
|
||
|
||
|
||
def aggregate_cell(per_run: list[dict]) -> dict:
|
||
"""Mean each metric across repeated generations of the same (case, block, effort)."""
|
||
if not per_run:
|
||
return {"distance": 1.0, "anti_pattern_total": 0.0, "change_percent": 100.0,
|
||
"golden_ratio_deviation_pp": None, "n": 0}
|
||
ratios = [r["golden_ratio_deviation_pp"] for r in per_run if r.get("golden_ratio_deviation_pp") is not None]
|
||
return {
|
||
"distance": round(mean(r["distance"] for r in per_run), 4),
|
||
"anti_pattern_total": round(mean(r["anti_pattern_total"] for r in per_run), 2),
|
||
"change_percent": round(mean(r["change_percent"] for r in per_run), 2),
|
||
"golden_ratio_deviation_pp": round(mean(ratios), 2) if ratios else None,
|
||
"n": len(per_run),
|
||
}
|
||
|
||
|
||
def _current_default(block_id: str) -> str | None:
|
||
from legal_mcp.services.block_writer import BLOCK_CONFIG, DEFAULT_EFFORT
|
||
cfg = BLOCK_CONFIG.get(block_id, {})
|
||
if cfg.get("model") != "ai":
|
||
return None
|
||
return cfg.get("effort", DEFAULT_EFFORT)
|
||
|
||
|
||
# ── re-rank a saved report (offline — NO LLM, NO DB) ─────────────────────────
|
||
def rerank_saved(saved: dict) -> dict:
|
||
"""Re-apply the CURRENT (#213) recommend_effort to a saved calibration report.
|
||
|
||
Reads the per-(block,effort) rows that the run already persisted
|
||
(distance/anti/change%/ratioΔ/n) and recomputes recommended + confidence — no
|
||
regeneration, no model, no DB. Lets a past run be re-ranked under the new
|
||
style-clean key. Returns {"block_id": {old, new, confidence, confidence_margin,
|
||
current_default, efforts}}. Pure → unit-tested in --self-test.
|
||
"""
|
||
out: dict[str, dict] = {}
|
||
for block_id, bd in (saved.get("by_block") or {}).items():
|
||
rows = bd.get("efforts") or []
|
||
rec = recommend_effort(rows)
|
||
out[block_id] = {
|
||
"section": bd.get("section"),
|
||
"current_default": bd.get("current_default"),
|
||
"old_recommended": bd.get("recommended"),
|
||
"new_recommended": rec["effort"] if rec else None,
|
||
"confidence": rec["confidence"] if rec else None,
|
||
"confidence_margin": rec.get("confidence_margin") if rec else None,
|
||
"efforts": rows,
|
||
}
|
||
return out
|
||
|
||
|
||
def _render_rerank_md(saved: dict, reranked: dict, src: Path) -> str:
|
||
g = saved.get("grid", {})
|
||
finals = g.get("finals", [])
|
||
lines = [
|
||
f"# #213 — re-rank של {src.name} (style-clean, anti-primary · OFFLINE)\n",
|
||
f"> מקור: `{src.name}` · {g.get('n_finals','?')} סופיים "
|
||
f"({', '.join(finals) or '—'}) · ללא LLM/DB — דירוג-מחדש בלבד.\n",
|
||
"## old pick → new pick (+confidence)\n",
|
||
"| block | section | current | old pick | **new pick** | confidence | n |",
|
||
"|---|---|---|---|---|---|---|",
|
||
]
|
||
for b, rd in reranked.items():
|
||
old = rd.get("old_recommended") or "—"
|
||
new = rd.get("new_recommended") or "—"
|
||
mark = "" if new == old else " ⬅︎"
|
||
conf = rd.get("confidence")
|
||
conf_cell = {"clear": "clear", "weak": "⚠️ weak (within-noise)"}.get(conf, "—")
|
||
n_b = rd["efforts"][0]["n"] if rd.get("efforts") else 0
|
||
lines.append(
|
||
f"| {b} | {rd.get('section') or '—'} | {rd.get('current_default') or '—'} | "
|
||
f"{old} | **{new}**{mark} | {conf_cell} | {n_b} |")
|
||
lines.append("")
|
||
for b, rd in reranked.items():
|
||
if not rd.get("efforts"):
|
||
continue
|
||
lines += [f"### {b} ({rd.get('section') or '—'})\n",
|
||
"| effort | distance | anti_total | change% (reported) | ratioΔpp | n |",
|
||
"|---|---|---|---|---|---|"]
|
||
for r in rd["efforts"]:
|
||
star = " ⭐" if r["effort"] == rd.get("new_recommended") else ""
|
||
ratio = r.get("golden_ratio_deviation_pp")
|
||
lines.append(
|
||
f"| {r['effort']}{star} | {r['distance']:.4f} | {r['anti_pattern_total']} | "
|
||
f"{r['change_percent']} | {ratio if ratio is not None else '—'} | {r['n']} |")
|
||
lines.append("")
|
||
lines.append("> דירוג **style-clean** (#213): anti_total ראשי → ratioΔ → distance (tiebreak); "
|
||
"**change% מדווח-לא-מדורג** (07-learning §0.7); "
|
||
"confidence=⚠️weak ⇒ הובלת-anti < epsilon = max(0.5, 0.20·spread).\n")
|
||
return "\n".join(lines) + "\n"
|
||
|
||
|
||
def _do_rerank(path: Path) -> int:
|
||
"""--rerank entry: load a saved report, re-rank offline, print + write *-reranked.md."""
|
||
if not path.exists():
|
||
print(f"rerank: file not found: {path}", file=sys.stderr)
|
||
return 2
|
||
saved = json.loads(path.read_text(encoding="utf-8"))
|
||
reranked = rerank_saved(saved)
|
||
md = _render_rerank_md(saved, reranked, path)
|
||
print(md)
|
||
out = path.with_name(path.stem + "-reranked.md")
|
||
out.write_text(md, encoding="utf-8")
|
||
print(f" reranked report written: {out}", file=sys.stderr)
|
||
return 0
|
||
|
||
|
||
# ── self-test (no DB, no model) ──────────────────────────────────────────────
|
||
def _self_test() -> int:
|
||
ok = True
|
||
|
||
def chk(name, cond):
|
||
nonlocal ok
|
||
ok = ok and cond
|
||
print(f" {name:42} {'ok' if cond else 'FAIL'}")
|
||
|
||
# block_distance_to_final: regen identical to final-section ⇒ change ~0,
|
||
# distance dominated by anti-patterns (0 here) ⇒ ~0.
|
||
from legal_mcp.services.style_distance import (
|
||
block_distance_to_final, split_final_by_section,
|
||
)
|
||
final_section = "לפנינו ערר על החלטת הוועדה המקומית. " * 40
|
||
d_same = block_distance_to_final("block-he", final_section, final_section, "rejection",
|
||
section_target_total_words=len(final_section.split()))
|
||
chk("identical regen ⇒ change_percent==0", d_same["change_percent"] == 0.0)
|
||
chk("identical regen ⇒ anti==0", d_same["anti_pattern_total"] == 0)
|
||
chk("identical regen ⇒ distance small", d_same["distance"] < 0.05)
|
||
|
||
# A regen full of anti-patterns (markdown headers / bullet lists) scores worse
|
||
# than clean continuous narrative, holding the final fixed.
|
||
clean = "אנו סבורים כי דין הערר להידחות. כידוע, הלכה פסוקה היא. " * 20
|
||
dirty = "## כותרת\n- נקודה ראשונה\n- נקודה שנייה\n* עוד נקודה\n### תת\n" * 10
|
||
d_clean = block_distance_to_final("block-yod", clean, final_section, "rejection",
|
||
section_target_total_words=len(final_section.split()))
|
||
d_dirty = block_distance_to_final("block-yod", dirty, final_section, "rejection",
|
||
section_target_total_words=len(final_section.split()))
|
||
chk("dirty regen has more anti-patterns", d_dirty["anti_pattern_total"] > d_clean["anti_pattern_total"])
|
||
chk("dirty regen ⇒ larger distance", d_dirty["distance"] > d_clean["distance"])
|
||
|
||
# ratio deviation is None when no total provided (never fabricated)
|
||
d_noratio = block_distance_to_final("block-he", clean, final_section, "rejection")
|
||
chk("no total ⇒ ratio deviation None", d_noratio["golden_ratio_deviation_pp"] is None)
|
||
|
||
# split_final_by_section returns mapped golden sections only.
|
||
sample_final = (
|
||
"רקע עובדתי\nהמקרקעין נשוא הערר. " * 10 + "\n\n"
|
||
"תמצית טענות הצדדים\nהעוררים טוענים כי. " * 10 + "\n\n"
|
||
"דיון והכרעה\nאנו סבורים כי. " * 10
|
||
)
|
||
secs = split_final_by_section(sample_final)
|
||
chk("split maps to golden sections", set(secs).issubset(
|
||
{"background", "claims", "discussion", "summary"}))
|
||
chk("split found ≥1 section", len(secs) >= 1)
|
||
|
||
# recommend_effort (#213, style-clean): anti_pattern_total is PRIMARY — the
|
||
# effort with the FEWEST style violations wins, even if its distance is higher.
|
||
rec = recommend_effort([
|
||
{"effort": "low", "distance": 0.20, "anti_pattern_total": 5, "change_percent": 22,
|
||
"golden_ratio_deviation_pp": 3.0, "n": 1},
|
||
{"effort": "high", "distance": 0.40, "anti_pattern_total": 2, "change_percent": 40,
|
||
"golden_ratio_deviation_pp": 5.0, "n": 1},
|
||
{"effort": "xhigh", "distance": 0.50, "anti_pattern_total": 1, "change_percent": 50,
|
||
"golden_ratio_deviation_pp": 9.0, "n": 1},
|
||
])
|
||
chk("recommend: anti-primary (not distance)", rec["effort"] == "xhigh")
|
||
chk("recommend: change_percent NOT ranked", rec["distance"] == 0.50) # worst distance still won
|
||
chk("recommend empty ⇒ None", recommend_effort([]) is None)
|
||
|
||
# SECONDARY = golden_ratio_deviation_pp when anti ties; distance only as last tiebreak.
|
||
rec_tie = recommend_effort([
|
||
{"effort": "low", "distance": 0.10, "anti_pattern_total": 3, "change_percent": 10,
|
||
"golden_ratio_deviation_pp": 8.0, "n": 1},
|
||
{"effort": "high", "distance": 0.90, "anti_pattern_total": 3, "change_percent": 90,
|
||
"golden_ratio_deviation_pp": 2.0, "n": 1},
|
||
])
|
||
chk("recommend: ratio breaks anti-tie", rec_tie["effort"] == "high")
|
||
rec_ratio_tie = recommend_effort([
|
||
{"effort": "low", "distance": 0.30, "anti_pattern_total": 3, "change_percent": 10,
|
||
"golden_ratio_deviation_pp": 5.0, "n": 1},
|
||
{"effort": "high", "distance": 0.10, "anti_pattern_total": 3, "change_percent": 90,
|
||
"golden_ratio_deviation_pp": 5.0, "n": 1},
|
||
])
|
||
chk("recommend: distance breaks final tie", rec_ratio_tie["effort"] == "high")
|
||
# None ratio is treated as worst (never beats a real deviation on the secondary key).
|
||
rec_none = recommend_effort([
|
||
{"effort": "low", "distance": 0.10, "anti_pattern_total": 2, "change_percent": 10,
|
||
"golden_ratio_deviation_pp": None, "n": 1},
|
||
{"effort": "high", "distance": 0.90, "anti_pattern_total": 2, "change_percent": 90,
|
||
"golden_ratio_deviation_pp": 4.0, "n": 1},
|
||
])
|
||
chk("recommend: None ratio is worst", rec_none["effort"] == "high")
|
||
|
||
# confidence flag (#213): a clear anti lead ⇒ "clear"; within-noise ⇒ "weak".
|
||
rec_clear = recommend_effort([
|
||
{"effort": "low", "distance": 0.5, "anti_pattern_total": 1.0, "change_percent": 50,
|
||
"golden_ratio_deviation_pp": 5.0, "n": 5},
|
||
{"effort": "high", "distance": 0.5, "anti_pattern_total": 4.0, "change_percent": 50,
|
||
"golden_ratio_deviation_pp": 5.0, "n": 5},
|
||
])
|
||
chk("confidence: big anti lead ⇒ clear", rec_clear["confidence"] == "clear")
|
||
rec_weak = recommend_effort([
|
||
{"effort": "low", "distance": 0.4, "anti_pattern_total": 3.6, "change_percent": 93,
|
||
"golden_ratio_deviation_pp": 6.7, "n": 5},
|
||
{"effort": "xhigh", "distance": 0.5, "anti_pattern_total": 3.6, "change_percent": 102,
|
||
"golden_ratio_deviation_pp": 6.6, "n": 5},
|
||
{"effort": "high", "distance": 0.53, "anti_pattern_total": 3.8, "change_percent": 104,
|
||
"golden_ratio_deviation_pp": 5.7, "n": 5},
|
||
])
|
||
chk("confidence: anti tie ⇒ weak", rec_weak["confidence"] == "weak")
|
||
rec_single = recommend_effort([
|
||
{"effort": "low", "distance": 0.4, "anti_pattern_total": 2.0, "change_percent": 50,
|
||
"golden_ratio_deviation_pp": 5.0, "n": 5}])
|
||
chk("confidence: single effort ⇒ clear", rec_single["confidence"] == "clear")
|
||
|
||
# aggregate_cell: means + ratio drops Nones, keeps n.
|
||
agg = aggregate_cell([
|
||
{"distance": 0.2, "anti_pattern_total": 2, "change_percent": 20, "golden_ratio_deviation_pp": 3.0},
|
||
{"distance": 0.4, "anti_pattern_total": 4, "change_percent": 30, "golden_ratio_deviation_pp": None},
|
||
])
|
||
chk("aggregate distance mean", agg["distance"] == 0.3)
|
||
chk("aggregate n counted", agg["n"] == 2)
|
||
chk("aggregate ratio skips None", agg["golden_ratio_deviation_pp"] == 3.0)
|
||
|
||
# rerank_saved: re-applies the new key to a saved report's stored rows, no I/O.
|
||
# Mimics the real 8-final block-vav rows: low has fewest anti by a clear margin.
|
||
saved = {"grid": {"n_finals": 7, "finals": ["x"]}, "by_block": {
|
||
"block-vav": {"section": "background", "current_default": "medium",
|
||
"recommended": "low", "efforts": [
|
||
{"effort": "low", "distance": 0.4454, "anti_pattern_total": 1.57, "change_percent": 87.39, "golden_ratio_deviation_pp": 6.33, "n": 7},
|
||
{"effort": "medium", "distance": 0.4749, "anti_pattern_total": 2.29, "change_percent": 87.76, "golden_ratio_deviation_pp": 6.37, "n": 7},
|
||
{"effort": "high", "distance": 0.5381, "anti_pattern_total": 4.14, "change_percent": 90.69, "golden_ratio_deviation_pp": 5.99, "n": 7},
|
||
{"effort": "xhigh", "distance": 0.5413, "anti_pattern_total": 4.57, "change_percent": 89.53, "golden_ratio_deviation_pp": 6.41, "n": 7},
|
||
]},
|
||
}}
|
||
rr = rerank_saved(saved)
|
||
chk("rerank: block-vav stays low", rr["block-vav"]["new_recommended"] == "low")
|
||
chk("rerank: block-vav clear (anti lead)", rr["block-vav"]["confidence"] == "clear")
|
||
chk("rerank: keeps old pick for diff", rr["block-vav"]["old_recommended"] == "low")
|
||
chk("rerank: renders markdown", "new pick" in _render_rerank_md(saved, rr, Path("x.json")))
|
||
|
||
print("ALL PASS" if ok else "*** FAILURES ***")
|
||
return 0 if ok else 1
|
||
|
||
|
||
# ── live A/B (host-only — needs DB + `claude` CLI) ───────────────────────────
|
||
async def _finals_for_calibration(case_filter: str | None) -> list[dict]:
|
||
"""draft_final_pairs whose final_text is populated (the held-out comparison set)."""
|
||
from legal_mcp.services import db
|
||
pairs = await db.list_draft_final_pairs(limit=500)
|
||
out: list[dict] = []
|
||
for p in pairs:
|
||
if case_filter and p.get("case_number") != case_filter:
|
||
continue
|
||
full = await db.get_draft_final_pair(p["id"])
|
||
if full and (full.get("final_text") or "").strip():
|
||
out.append(full)
|
||
return out
|
||
|
||
|
||
async def _score_cell(case_id, block_id: str, effort: str, final_section: str,
|
||
final_total_words: int, outcome: str, repeats: int) -> dict:
|
||
"""Generate `block_id` at `effort` `repeats` times; score each vs the final section."""
|
||
from legal_mcp.services import block_writer
|
||
from legal_mcp.services.style_distance import block_distance_to_final
|
||
runs: list[dict] = []
|
||
for _ in range(repeats):
|
||
res = await block_writer.write_block(case_id, block_id, effort_override=effort)
|
||
scored = block_distance_to_final(
|
||
block_id, res.get("content", ""), final_section, outcome,
|
||
section_target_total_words=final_total_words,
|
||
)
|
||
runs.append(scored)
|
||
agg = aggregate_cell(runs)
|
||
agg["effort"] = effort
|
||
agg["runs"] = runs
|
||
return agg
|
||
|
||
|
||
async def _run(args, ts: str) -> dict:
|
||
from uuid import UUID
|
||
from legal_mcp.services import db
|
||
from legal_mcp.services.lessons import canonical_outcome
|
||
from legal_mcp.services.style_distance import split_final_by_section, _BLOCK_TO_SECTION
|
||
|
||
efforts = args.efforts
|
||
blocks = args.blocks
|
||
finals = await _finals_for_calibration(args.case)
|
||
|
||
cases_meta = []
|
||
for f in finals:
|
||
case = await db.get_case_by_number(f["case_number"]) if f.get("case_number") else None
|
||
if not case:
|
||
continue
|
||
decision = await db.get_decision_by_case(UUID(case["id"]))
|
||
outcome = canonical_outcome((decision or {}).get("outcome", "rejection"))
|
||
sections = split_final_by_section(f.get("final_text", ""))
|
||
final_total_words = len((f.get("final_text", "") or "").split())
|
||
cases_meta.append({
|
||
"case_number": f["case_number"], "case_id": case["id"],
|
||
"outcome": outcome, "sections": sections, "final_total_words": final_total_words,
|
||
})
|
||
|
||
# grid plan: (block → cases that have its section)
|
||
plan: dict[str, list[dict]] = {}
|
||
for block_id in blocks:
|
||
section = _BLOCK_TO_SECTION.get(block_id)
|
||
plan[block_id] = [c for c in cases_meta if section and c["sections"].get(section)]
|
||
|
||
total_cells = sum(len(plan[b]) for b in blocks) * len(efforts) * args.repeats
|
||
grid_summary = {
|
||
"n_finals": len(cases_meta),
|
||
"finals": [c["case_number"] for c in cases_meta],
|
||
"blocks": blocks, "efforts": efforts, "repeats": args.repeats,
|
||
"total_generations": total_cells,
|
||
"per_block_n": {b: len(plan[b]) for b in blocks},
|
||
}
|
||
|
||
if args.dry_run:
|
||
return {"dry_run": True, "grid": grid_summary, "by_block": {}}
|
||
|
||
by_block: dict[str, dict] = {}
|
||
for block_id in blocks:
|
||
section = _BLOCK_TO_SECTION.get(block_id)
|
||
per_effort_runs: dict[str, list[dict]] = {e: [] for e in efforts}
|
||
per_case: list[dict] = []
|
||
for c in plan[block_id]:
|
||
final_section = c["sections"][section]
|
||
case_cells = []
|
||
for effort in efforts:
|
||
# Per-cell guard: ANY failure of a single (case, block, effort) cell
|
||
# — e.g. block-yod raising "ללא כיוון מאושר", or a transient rate-limit
|
||
# mid-run — is logged and skipped, NOT allowed to kill the whole run
|
||
# (INV-G8 eval-harness robustness). Partial results still persist below.
|
||
try:
|
||
cell = await _score_cell(
|
||
UUID(c["case_id"]), block_id, effort, final_section,
|
||
c["final_total_words"], c["outcome"], args.repeats,
|
||
)
|
||
except Exception as exc: # noqa: BLE001 — harness must survive any cell failure
|
||
logger.warning(
|
||
"calibration cell skipped: case=%s block=%s effort=%s — %s",
|
||
c["case_number"], block_id, effort, exc,
|
||
)
|
||
continue
|
||
per_effort_runs[effort].append(cell)
|
||
case_cells.append({k: cell[k] for k in
|
||
("effort", "distance", "anti_pattern_total",
|
||
"change_percent", "golden_ratio_deviation_pp", "n")})
|
||
per_case.append({"case_number": c["case_number"], "cells": case_cells})
|
||
|
||
# mean across cases for each effort → one comparable row per effort
|
||
effort_rows = []
|
||
for effort in efforts:
|
||
rows = per_effort_runs[effort]
|
||
if not rows:
|
||
continue
|
||
ratios = [r["golden_ratio_deviation_pp"] for r in rows if r.get("golden_ratio_deviation_pp") is not None]
|
||
effort_rows.append({
|
||
"effort": effort,
|
||
"distance": round(mean(r["distance"] for r in rows), 4),
|
||
"anti_pattern_total": round(mean(r["anti_pattern_total"] for r in rows), 2),
|
||
"change_percent": round(mean(r["change_percent"] for r in rows), 2),
|
||
"golden_ratio_deviation_pp": round(mean(ratios), 2) if ratios else None,
|
||
"n": len(rows),
|
||
})
|
||
rec = recommend_effort(effort_rows)
|
||
by_block[block_id] = {
|
||
"section": section,
|
||
"current_default": _current_default(block_id),
|
||
"recommended": rec["effort"] if rec else None,
|
||
"confidence": rec["confidence"] if rec else None,
|
||
"confidence_margin": rec.get("confidence_margin") if rec else None,
|
||
"efforts": effort_rows,
|
||
"per_case": per_case,
|
||
}
|
||
|
||
# Incremental persistence (INV-G8): flush the report after every completed
|
||
# block so a crash later in the grid never loses blocks already generated.
|
||
# Blocks not yet done are simply absent from by_block; _write_report tolerates
|
||
# partial results. main() does the final flush once the loop finishes.
|
||
try:
|
||
_write_report({"dry_run": False, "grid": grid_summary, "by_block": by_block}, ts)
|
||
except Exception as exc: # noqa: BLE001 — a write hiccup must not abort the run
|
||
logger.warning("incremental report write failed after block=%s — %s", block_id, exc)
|
||
|
||
return {"dry_run": False, "grid": grid_summary, "by_block": by_block}
|
||
|
||
|
||
IL_TZ = ZoneInfo("Asia/Jerusalem")
|
||
|
||
|
||
def _ts() -> str:
|
||
"""Filename-safe report stamp in Israel time (INV-UI9: human-facing display).
|
||
|
||
This script runs on the host, which is UTC, so a process-clock stamp would
|
||
mislead the chair. We pin Asia/Jerusalem and suffix ``IL`` (instead of a bare
|
||
``Z``) so the displayed/filename time is unambiguously Israel-local.
|
||
"""
|
||
return datetime.now(IL_TZ).strftime("%Y%m%dT%H%M%S-IL")
|
||
|
||
|
||
def _write_report(result: dict, ts: str) -> tuple[Path, Path]:
|
||
OUT_DIR.mkdir(parents=True, exist_ok=True)
|
||
jp = OUT_DIR / f"effort-calibration-{ts}.json"
|
||
mp = OUT_DIR / f"effort-calibration-{ts}.md"
|
||
jp.write_text(json.dumps(result, ensure_ascii=False, indent=2), encoding="utf-8")
|
||
|
||
g = result["grid"]
|
||
n = g["n_finals"]
|
||
il_now = datetime.now(IL_TZ).strftime("%Y-%m-%d %H:%M")
|
||
lines = [
|
||
f"# #208 — כיול model×effort מול הסופיים — {ts}\n",
|
||
f"> נוצר: {il_now} (שעון ישראל · Asia/Jerusalem)\n",
|
||
f"> ⚠️ **גודל-מדגם: {n} סופיים** ({', '.join(g['finals']) or '—'}). "
|
||
"זוהי **עדות-כיוון, לא רגרסיה** — מעט תיקים בעלי סופי-עלוי. "
|
||
"ההמלצה אדוויזורית; ההכרעה בידי היו\"ר/המפעיל.\n",
|
||
f"- בלוקים: {', '.join(g['blocks'])}",
|
||
f"- efforts: {', '.join(g['efforts'])} · repeats/cell: {g['repeats']}",
|
||
f"- סך ייצורי-מודל: {g['total_generations']}",
|
||
"",
|
||
]
|
||
if result.get("dry_run"):
|
||
lines += ["## DRY-RUN — תכנון הגריד בלבד (ללא ייצור)\n",
|
||
"| block | #cases | current default |", "|---|---|---|"]
|
||
for b in g["blocks"]:
|
||
lines.append(f"| {b} | {g['per_block_n'].get(b,0)} | {_current_default(b) or '—'} |")
|
||
mp.write_text("\n".join(lines) + "\n", encoding="utf-8")
|
||
return jp, mp
|
||
|
||
lines += ["## המלצה per-בלוק (style-clean: anti_pattern_total ראשי — #213)\n",
|
||
"| block | section | current | **recommended** | confidence | n |",
|
||
"|---|---|---|---|---|---|"]
|
||
for b, bd in result["by_block"].items():
|
||
rec = bd.get("recommended") or "—"
|
||
mark = "" if rec == bd.get("current_default") else " ⬅︎"
|
||
conf = bd.get("confidence")
|
||
conf_cell = {"clear": "clear", "weak": "⚠️ weak (within-noise)"}.get(conf, "—")
|
||
n_b = bd["efforts"][0]["n"] if bd.get("efforts") else 0
|
||
lines.append(
|
||
f"| {b} | {bd.get('section','')} | {bd.get('current_default') or '—'} | "
|
||
f"**{rec}**{mark} | {conf_cell} | {n_b} |")
|
||
lines.append("")
|
||
for b, bd in result["by_block"].items():
|
||
lines += [f"### {b} ({bd.get('section','')})\n",
|
||
"| effort | distance | anti_total | change% | ratioΔpp | n |",
|
||
"|---|---|---|---|---|---|"]
|
||
for r in bd.get("efforts", []):
|
||
star = " ⭐" if r["effort"] == bd.get("recommended") else ""
|
||
ratio = r["golden_ratio_deviation_pp"]
|
||
lines.append(
|
||
f"| {r['effort']}{star} | {r['distance']:.4f} | {r['anti_pattern_total']} | "
|
||
f"{r['change_percent']} | {ratio if ratio is not None else '—'} | {r['n']} |")
|
||
lines.append("")
|
||
lines.append("> דירוג-ההמלצה **style-clean** (#213): anti_total ראשי → ratioΔ → distance (tiebreak). "
|
||
"**change% מדווח-לא-מדורג** — מערבב סגנון עם שלמות-תוכן (07-learning §0.7), "
|
||
"anti_total הוא הסיגנל הנקי-לסגנון. confidence=⚠️weak ⇒ הבחירה בתוך-הרעש "
|
||
"(ההובלה ב-anti קטנה מ-epsilon = max(0.5, 0.20·spread)).\n")
|
||
mp.write_text("\n".join(lines) + "\n", encoding="utf-8")
|
||
return jp, mp
|
||
|
||
|
||
async def main() -> int:
|
||
ap = argparse.ArgumentParser(description="#208 model/effort calibration harness")
|
||
ap.add_argument("--self-test", action="store_true", help="offline measurement-logic proof (no DB/CLI)")
|
||
ap.add_argument("--rerank", metavar="REPORT.json", default=None,
|
||
help="re-rank a saved calibration JSON under the current style-clean key (no LLM/DB)")
|
||
ap.add_argument("--dry-run", action="store_true", help="plan the A/B grid over existing finals, no model calls")
|
||
ap.add_argument("--efforts", default=",".join(DEFAULT_EFFORTS),
|
||
help=f"comma effort grid (default {','.join(DEFAULT_EFFORTS)})")
|
||
ap.add_argument("--blocks", default=",".join(CALIBRATABLE_BLOCKS),
|
||
help="comma block ids to calibrate")
|
||
ap.add_argument("--case", default=None, help="restrict to a single case_number")
|
||
ap.add_argument("--repeats", type=int, default=1, help="generations per cell (avg out gen noise)")
|
||
args = ap.parse_args()
|
||
|
||
logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
|
||
|
||
if args.self_test:
|
||
return _self_test()
|
||
|
||
if args.rerank:
|
||
return _do_rerank(Path(args.rerank))
|
||
|
||
args.efforts = [e.strip() for e in args.efforts.split(",") if e.strip()]
|
||
bad = [e for e in args.efforts if e not in VALID_EFFORTS]
|
||
if bad:
|
||
print(f"invalid effort(s): {bad}. valid: {sorted(VALID_EFFORTS)}", file=sys.stderr)
|
||
return 2
|
||
args.blocks = [b.strip() for b in args.blocks.split(",") if b.strip()]
|
||
bad_b = [b for b in args.blocks if b not in VALID_BLOCKS]
|
||
if bad_b:
|
||
print(f"non-calibratable block(s): {bad_b}. valid: {VALID_BLOCKS}", file=sys.stderr)
|
||
return 2
|
||
|
||
ts = _ts()
|
||
result = await _run(args, ts)
|
||
jp, mp = _write_report(result, ts)
|
||
|
||
g = result["grid"]
|
||
print(f"CALIBRATION: {g['n_finals']} finals — DIRECTIONAL EVIDENCE, not a regression")
|
||
if g["n_finals"] == 0:
|
||
print(" no finals with final_text in draft_final_pairs — upload a signed final first.")
|
||
elif result.get("dry_run"):
|
||
print(f" dry-run: {g['total_generations']} generations planned across {len(g['blocks'])} blocks")
|
||
else:
|
||
for b, bd in result["by_block"].items():
|
||
print(f" {b:16} current={bd.get('current_default') or '—':6} "
|
||
f"→ recommended={bd.get('recommended') or '—':6} "
|
||
f"[{bd.get('confidence') or '—'}]")
|
||
print(f" report: {mp}")
|
||
return 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
sys.exit(asyncio.run(main()))
|