Completes the Wave M B3.5 consolidation slice (b–e), built on #728. B3.5-b — calibration as a first-class evidence subject (`calibration_class`, address `calibration:<class_name>`): RightInspector projection + Evidence Chain Rail semantics (serving-discipline evidence, not runtime truth). B3.5-c / B4a — nullable `LeewayEvidence` read model threaded through turn, replay, cognition-proposal, and math-proposal surfaces, with a shared absence-honest card. B4 is gated correctly: the tuple exists in typed data but no producer populates it, so the card renders absence (verified: no non-null producer in workbench/core/chat). B3.5-d/e — UI-UX-GUIDE.md, b4-leeway-feasibility-gate.md, phase-a-residue-ledger.md. Practice artifact — earn-it-for-real (runner-reproducible). The committed `report.json` (additive earns PROPOSE @0.861, 95/5/50) is now emitted by a deterministic runner rather than copied from the queue. `propose_runner` gains `regenerate_practice_artifacts()`, which runs ONE sealed `resolve_pooled` practice pass and writes BOTH report.json (the per-class ledger the calibration reader consumes) and ratification_queue.json — two projections of one ledger, coherent by construction and byte-reproducible. `runner.main()` delegates to it (lazy import, no cycle), so both entry points produce the identical pair. This closes the gap where a hand-copied report.json agreed with the queue but no runner produced it. `resolve_pooled` is the aggressive sealed PROPOSE-regime scorer (proposal-only/HITL, unsafe for serving, legitimate for attempt-and-eliminate); wrong=5 is the sealed-practice learning signal, NOT the serving wrong=0. No serving/derivation/reliability_gate source touched; the practice lane is not in the serving-frozen SHA gate. Validated: - python -m pytest tests/test_workbench_{calibration,journal,replay,schemas}.py -> 31 passed - python -m pytest tests/ -k "workbench or propose or learning_arena or practice" -> 190 passed (3 failing tests in test_adr_0175_phase2_practice_lane.py are PRE-EXISTING reds on clean origin/main: stale 4/0/46 assertions on build_report, which this change does not touch) - report.json + ratification_queue.json: deterministic (run1==run2) and reproduced byte-identically by both `python -m ...runner` and `...propose_runner` - pnpm build green; 144 UI tests across calibration/leeway/evidence/replay/ doctrine-gates/routes-docs-drift all pass
113 lines
4.3 KiB
Python
113 lines
4.3 KiB
Python
"""Wave M Phase B — calibration / serving-discipline readers (ADR-0175).
|
|
|
|
The load-bearing obligation: the workbench re-implements none of the engine's
|
|
calibration math. These tests prove the reader's numbers come from
|
|
``core.reliability_gate`` (``conservative_floor`` / ``license_for``), and that
|
|
the serving counts are read from the committed reports unchanged.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from core.reliability_gate import conservative_floor
|
|
from workbench import calibration
|
|
from workbench.api import WorkbenchApi
|
|
from workbench.readers import EvidenceUnavailableError
|
|
|
|
|
|
def _write_practice_report(tmp_path: Path, per_class: dict) -> Path:
|
|
path = tmp_path / "report.json"
|
|
path.write_text(
|
|
json.dumps({"adr": "0175", "regime": "practice", "per_class": per_class}),
|
|
encoding="utf-8",
|
|
)
|
|
return path
|
|
|
|
|
|
def test_serving_metrics_read_committed_counts_unchanged() -> None:
|
|
metrics = {m.lane: m for m in calibration.read_serving_metrics()}
|
|
assert "train_sample" in metrics
|
|
# The live invariant: the committed serving lane commits zero wrong answers.
|
|
assert metrics["train_sample"].wrong == 0
|
|
assert metrics["train_sample"].correct >= 0
|
|
assert metrics["train_sample"].source_digest.startswith("sha256:")
|
|
|
|
|
|
def test_calibration_classes_over_committed_report_are_honest() -> None:
|
|
rows = calibration.read_calibration_classes()
|
|
assert rows, "expected the committed per_class ledger to yield rows"
|
|
earned = [row for row in rows if row.propose_licensed or row.serve_licensed]
|
|
assert earned, "committed practice evidence must show a class crossing theta"
|
|
additive = next(row for row in rows if row.class_name == "additive")
|
|
assert additive.correct == 95
|
|
assert additive.wrong == 5
|
|
assert additive.committed == 100
|
|
assert additive.propose_licensed is True
|
|
assert additive.serve_licensed is False
|
|
assert additive.source_path == "evals/gsm8k_math/practice/v1/report.json"
|
|
assert additive.source_digest.startswith("sha256:")
|
|
|
|
queue = json.loads(
|
|
(calibration.PRACTICE_REPORT.parent / "ratification_queue.json").read_text(
|
|
encoding="utf-8"
|
|
)
|
|
)
|
|
proposal = queue["proposals"][0]
|
|
assert proposal["class_name"] == additive.class_name
|
|
assert proposal["correct"] == additive.correct
|
|
assert proposal["wrong"] == additive.wrong
|
|
assert proposal["committed"] == additive.committed
|
|
|
|
|
|
def test_reader_uses_the_engine_math_not_its_own(tmp_path) -> None:
|
|
# A class that has earned PROPOSE (0.86 >= 0.85) but not SERVE (< 0.99).
|
|
report = _write_practice_report(
|
|
tmp_path,
|
|
{
|
|
"additive": {"correct": 95, "wrong": 5, "refused": 50},
|
|
"novice": {"correct": 0, "wrong": 0, "refused": 4},
|
|
},
|
|
)
|
|
rows = {r.class_name: r for r in calibration.read_calibration_classes(report)}
|
|
|
|
earned = rows["additive"]
|
|
# The reader's reliability is the engine's own Wilson floor, to the digit.
|
|
assert earned.reliability_floor == round(conservative_floor(95, 100), 9)
|
|
assert earned.committed == 100
|
|
assert earned.propose_required == 0.85 and earned.propose_licensed is True
|
|
assert earned.serve_required == 0.99 and earned.serve_licensed is False
|
|
|
|
novice = rows["novice"]
|
|
assert novice.reliability_floor == 0.0 # below N_MIN
|
|
assert novice.propose_licensed is False
|
|
|
|
|
|
def test_calibration_classes_are_failures_first(tmp_path) -> None:
|
|
report = _write_practice_report(
|
|
tmp_path,
|
|
{
|
|
"earned": {"correct": 95, "wrong": 5, "refused": 0},
|
|
"unearned": {"correct": 0, "wrong": 0, "refused": 9},
|
|
},
|
|
)
|
|
rows = calibration.read_calibration_classes(report)
|
|
# Un-licensed / lowest-reliability comes first.
|
|
assert rows[0].class_name == "unearned"
|
|
assert rows[-1].class_name == "earned"
|
|
|
|
|
|
def test_endpoints_return_items() -> None:
|
|
api = WorkbenchApi()
|
|
r1 = api.handle("GET", "/calibration/classes", b"")
|
|
assert r1.status == 200 and isinstance(r1.payload["data"]["items"], list)
|
|
r2 = api.handle("GET", "/serving/metrics", b"")
|
|
assert r2.status == 200 and isinstance(r2.payload["data"]["items"], list)
|
|
|
|
|
|
def test_missing_practice_report_is_evidence_unavailable(tmp_path) -> None:
|
|
with pytest.raises(EvidenceUnavailableError):
|
|
calibration.read_calibration_classes(tmp_path / "nope.json")
|