core-labs/core
1
0
Fork 0
Code Issues Pull requests 1 Projects Releases Packages Wiki Activity Actions

feat(adr-0175-phase2): sealed practice lane over GSM8K train

Browse source 
ADR-0175 Phase 2 — a NEW lane (evals/gsm8k_math/practice/v1/), separate from the
wrong=0-pinned serving runner which is NOT modified. Runs the 50 cases in
practice mode: scores correct/wrong/refused as practice metrics, feeds per-class
counts into the Phase 1 ledger, diagnoses every refusal (§8), emits an
elimination record per wrong.

- classify_operation: gold-derived primary op class {multiplicative,divisive,
  additive} from <<a*b=c>> calc annotations (Tier-1 checkable in practice).
- diagnose_refusal (§8): skill_gap / knowledge_gap / genuine_ambiguity router.
- EliminationRecord (§9): wrong attempt gold caught -> pruning signal.
- PracticeReport: counts + per-class ledger + diagnoses + eliminations; as_dict.
- run_practice(cases, scorer=...): injectable scorer for tests; defaults to the
  candidate-graph scorer (read-only — never alters serving).

Live result mirrors serving (3 correct / 0 wrong / 47 refused of 50) because the
engine still refuses rather than guesses — attempts/eliminations go live in
Phase 3. But the diagnosis is already actionable: 35 skill_gap / 12 knowledge_gap
/ 0 genuine_ambiguity — 74% of refusals are skill gaps (Phase 3's search target),
quantifying the skill-vs-knowledge split.

Invariants: #1 seal (serving still 3/47/0; no generate/chat import of the lane),
#3 determinism (report byte-identical across runs). Elimination + wrong-tolerance
paths unit-tested via injected scorer (no live wrongs yet).

Verified: Phase 1+2 53/53, serving train_sample tests 4/4 (seal), smoke 67/67,
ruff clean.
This commit is contained in:
Shay 2026-05-28 15:12:33 -07:00
parent 8775765881
commit d90887b80f
5 changed files with 508 additions and 0 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

0
evals/gsm8k_math/practice/__init__.py Normal file
View file

0
evals/gsm8k_math/practice/v1/__init__.py Normal file
View file

91
evals/gsm8k_math/practice/v1/report.json Normal file
View file

@ -0,0 +1,91 @@
{
"adr": "0175",
"counts": {
"correct": 3,
"refused": 47,
"wrong": 0
},
"diagnosis_counts": {
"genuine_ambiguity": 0,
"knowledge_gap": 12,
"skill_gap": 35
},
"elimination_records": [],
"per_class": {
"additive": {
"committed": 0,
"correct": 0,
"coverage": 0.0,
"refused": 4,
"reliability": 0.0,
"wrong": 0
},
"divisive": {
"committed": 0,
"correct": 0,
"coverage": 0.0,
"refused": 6,
"reliability": 0.0,
"wrong": 0
},
"multiplicative": {
"committed": 3,
"correct": 3,
"coverage": 0.075,
"refused": 37,
"reliability": 0.0,
"wrong": 0
}
},
"refusal_diagnoses": {
"gsm8k-train-sample-v1-0001": "skill_gap",
"gsm8k-train-sample-v1-0002": "skill_gap",
"gsm8k-train-sample-v1-0003": "skill_gap",
"gsm8k-train-sample-v1-0004": "knowledge_gap",
"gsm8k-train-sample-v1-0005": "knowledge_gap",
"gsm8k-train-sample-v1-0006": "skill_gap",
"gsm8k-train-sample-v1-0007": "knowledge_gap",
"gsm8k-train-sample-v1-0008": "knowledge_gap",
"gsm8k-train-sample-v1-0009": "knowledge_gap",
"gsm8k-train-sample-v1-0010": "knowledge_gap",
"gsm8k-train-sample-v1-0011": "skill_gap",
"gsm8k-train-sample-v1-0012": "skill_gap",
"gsm8k-train-sample-v1-0013": "skill_gap",
"gsm8k-train-sample-v1-0015": "skill_gap",
"gsm8k-train-sample-v1-0016": "skill_gap",
"gsm8k-train-sample-v1-0017": "skill_gap",
"gsm8k-train-sample-v1-0019": "skill_gap",
"gsm8k-train-sample-v1-0020": "skill_gap",
"gsm8k-train-sample-v1-0021": "skill_gap",
"gsm8k-train-sample-v1-0022": "skill_gap",
"gsm8k-train-sample-v1-0023": "skill_gap",
"gsm8k-train-sample-v1-0024": "skill_gap",
"gsm8k-train-sample-v1-0025": "knowledge_gap",
"gsm8k-train-sample-v1-0026": "knowledge_gap",
"gsm8k-train-sample-v1-0027": "skill_gap",
"gsm8k-train-sample-v1-0028": "skill_gap",
"gsm8k-train-sample-v1-0029": "skill_gap",
"gsm8k-train-sample-v1-0030": "knowledge_gap",
"gsm8k-train-sample-v1-0031": "skill_gap",
"gsm8k-train-sample-v1-0032": "skill_gap",
"gsm8k-train-sample-v1-0033": "skill_gap",
"gsm8k-train-sample-v1-0034": "skill_gap",
"gsm8k-train-sample-v1-0035": "knowledge_gap",
"gsm8k-train-sample-v1-0036": "skill_gap",
"gsm8k-train-sample-v1-0037": "skill_gap",
"gsm8k-train-sample-v1-0038": "skill_gap",
"gsm8k-train-sample-v1-0039": "skill_gap",
"gsm8k-train-sample-v1-0040": "skill_gap",
"gsm8k-train-sample-v1-0041": "skill_gap",
"gsm8k-train-sample-v1-0043": "skill_gap",
"gsm8k-train-sample-v1-0044": "skill_gap",
"gsm8k-train-sample-v1-0045": "skill_gap",
"gsm8k-train-sample-v1-0046": "skill_gap",
"gsm8k-train-sample-v1-0047": "skill_gap",
"gsm8k-train-sample-v1-0048": "knowledge_gap",
"gsm8k-train-sample-v1-0049": "skill_gap",
"gsm8k-train-sample-v1-0050": "knowledge_gap"
},
"regime": "practice",
"schema_version": 1
}

206
evals/gsm8k_math/practice/v1/runner.py Normal file
View file

@ -0,0 +1,206 @@
"""ADR-0175 Phase 2 — sealed practice lane over the GSM8K train sample.
Separate from the wrong=0-pinned serving runner (``train_sample/v1/runner.py``),
which is **never modified**. Runs the 47 cases in *practice* mode: scores
correct/wrong/refused as practice metrics (wrong is tolerated — it is the
learning signal, not a lane failure), feeds per-class counts into the Phase 1
reliability ledger, diagnoses every refusal (§8 skill/knowledge/ambiguity), and
emits an elimination record for each wrong.
The seal (invariant #1): this lane writes only its own ``report.json``; no
serving path reads it and no serving module imports this runner. A wrong here
never becomes a served answer.
On the current refuse-preferring pipeline the engine still declines rather than
guesses, so the live practice ledger mirrors serving (3/47/0) and zero
eliminations fire — the attempt-generating grounded search is Phase 3. Phase 2
proves the *regime*: lane, ledger wiring, diagnosis, elimination schema, seal.
"""
from __future__ import annotations
import json
import re
from dataclasses import dataclass
from pathlib import Path
from typing import Any, Callable, Mapping
from core.reliability_gate import ClassTally
from evals.gsm8k_math.runner import _score_one_candidate_graph
from evals.gsm8k_math.train_sample.v1.runner import _CASES_PATH, _adapt, _load_cases
OPERATION_CLASSES: tuple[str, ...] = ("multiplicative", "divisive", "additive")
REFUSAL_DIAGNOSES: tuple[str, ...] = ("skill_gap", "knowledge_gap", "genuine_ambiguity")
_HERE = Path(__file__).resolve().parent
_REPORT_PATH = _HERE / "report.json"
_CALC_RE = re.compile(r"<<([^=>]+)=")
def classify_operation(answer_expression: str) -> str:
"""Primary gold operation class from GSM8K ``<<a*b=c>>`` calc annotations.
``multiplicative`` if any ``*``; else ``divisive`` if any ``/``; else
``additive``. Gold-derived — legitimate in practice (Tier-1 checkable).
"""
has_mul = has_div = False
for step in _CALC_RE.findall(answer_expression or ""):
if "*" in step:
has_mul = True
if "/" in step:
has_div = True
if has_mul:
return "multiplicative"
if has_div:
return "divisive"
return "additive"
def diagnose_refusal(reason: str) -> str:
"""§8 router — name the missing piece behind a refusal.
First cut from the current refusal-reason vocabulary; refined in Phase 3
when the grounded search makes "skill" precise ("no grounded derivation
found"). Defaults conservatively to ``knowledge_gap`` (assume a missing
piece) rather than silently dropping a refusal.
"""
low = (reason or "").lower()
if "branches disagree" in low:
return "genuine_ambiguity"
if "produced no injection" in low or "no branch produced a solvable" in low:
return "skill_gap"
if "no admissible candidate" in low or "expected exactly one question" in low:
return "knowledge_gap"
return "knowledge_gap"
@dataclass(frozen=True, slots=True)
class EliminationRecord:
"""A wrong practice attempt that gold caught — the pruning signal (§9)."""
case_id: str
class_name: str
attempted: float | None
gold: float
reason: str
@dataclass(frozen=True, slots=True)
class PracticeReport:
counts: Mapping[str, int]
ledger: Mapping[str, ClassTally]
refusal_diagnoses: Mapping[str, str]
elimination_records: tuple[EliminationRecord, ...]
def as_dict(self) -> dict[str, Any]:
return {
"schema_version": 1,
"adr": "0175",
"regime": "practice",
"counts": dict(self.counts),
"per_class": {
cls: {
"correct": t.correct,
"wrong": t.wrong,
"refused": t.refused,
"committed": t.committed,
"reliability": t.reliability,
"coverage": t.coverage,
}
for cls, t in sorted(self.ledger.items())
},
"refusal_diagnoses": dict(sorted(self.refusal_diagnoses.items())),
"diagnosis_counts": _bucket_counts(self.refusal_diagnoses),
"elimination_records": [
{
"case_id": r.case_id,
"class_name": r.class_name,
"attempted": r.attempted,
"gold": r.gold,
"reason": r.reason,
}
for r in self.elimination_records
],
}
def _bucket_counts(diagnoses: Mapping[str, str]) -> dict[str, int]:
out = {d: 0 for d in REFUSAL_DIAGNOSES}
for d in diagnoses.values():
out[d] = out.get(d, 0) + 1
return out
def run_practice(
cases: list[dict[str, Any]],
*,
scorer: Callable[[dict[str, Any]], Any] | None = None,
) -> PracticeReport:
"""Run the cases in practice mode and build the report.
``scorer`` is injectable for testing; it defaults to the candidate-graph
scorer :func:`evals.gsm8k_math.runner._score_one_candidate_graph`. The
practice lane only *reads* the engine's outcome — it never alters the
serving path.
"""
score = scorer if scorer is not None else _score_one_candidate_graph
counts = {"correct": 0, "wrong": 0, "refused": 0}
ledger: dict[str, ClassTally] = {}
diagnoses: dict[str, str] = {}
elims: list[EliminationRecord] = []
for raw in cases:
cls = classify_operation(raw.get("answer_expression", ""))
outcome = score(_adapt(raw))
verdict = outcome.outcome
counts[verdict] = counts.get(verdict, 0) + 1
tally = ledger.get(cls) or ClassTally(cls)
if verdict == "correct":
tally = tally.record(correct=1)
elif verdict == "wrong":
tally = tally.record(wrong=1)
elims.append(
EliminationRecord(
case_id=outcome.case_id,
class_name=cls,
attempted=getattr(outcome, "actual_answer", None),
gold=float(raw["answer_numeric"]),
reason=outcome.reason or "",
)
)
else: # refused
tally = tally.record(refused=1)
diagnoses[outcome.case_id] = diagnose_refusal(outcome.reason or "")
ledger[cls] = tally
return PracticeReport(
counts=counts,
ledger=ledger,
refusal_diagnoses=diagnoses,
elimination_records=tuple(elims),
)
def build_report() -> PracticeReport:
return run_practice(_load_cases(_CASES_PATH))
def write_report(report: PracticeReport, path: Path = _REPORT_PATH) -> None:
path.write_text(
json.dumps(report.as_dict(), indent=2, sort_keys=True) + "\n",
encoding="utf-8",
)
def main() -> int:
"""Run the practice lane. Never fails on wrong — practice records it."""
write_report(build_report())
return 0
if __name__ == "__main__":
import sys
sys.exit(main())

211
tests/test_adr_0175_phase2_practice_lane.py Normal file
View file

@ -0,0 +1,211 @@
"""ADR-0175 Phase 2 — sealed practice lane.
A NEW lane (never the wrong=0-pinned train_sample serving runner) that runs the
47 train cases in *practice* mode: scores correct/wrong/refused as practice
metrics, feeds per-class counts into the Phase 1 ledger, diagnoses every refusal
(§8 skill/knowledge/ambiguity), and emits elimination records for wrongs.
On the current pipeline the engine still refuses rather than guesses, so the
practice ledger mirrors serving (3 correct / 0 wrong / 47 refused, of 50) and
zero eliminations fire live — the attempt-generating search is Phase 3. Phase 2
proves the *regime*: the lane, the ledger wiring, the diagnosis, the elimination
schema, and the seal.
Invariants exercised by failing-under-violation tests:
- #1 seal -> TestSealInvariant
- #3 determinism -> TestDeterminismInvariant
"""
from __future__ import annotations
import dataclasses
import pytest
from core.reliability_gate import conservative_floor
from evals.gsm8k_math.practice.v1.runner import (
OPERATION_CLASSES,
REFUSAL_DIAGNOSES,
EliminationRecord,
build_report,
classify_operation,
diagnose_refusal,
run_practice,
)
# A stub CaseOutcome shape compatible with what the practice lane reads.
@dataclasses.dataclass(frozen=True)
class _StubOutcome:
case_id: str
outcome: str
reason: str | None = None
actual_answer: float | None = None
def _case(cid: str, expr: str, gold: float, q: str = "Q?") -> dict:
return {
"case_id": cid,
"question": q,
"answer_numeric": gold,
"answer_expression": expr,
}
# ---------------------------------------------------------------------------
# classify_operation — gold-derived primary operation class
# ---------------------------------------------------------------------------
class TestClassifyOperation:
def test_multiplicative_when_star_present(self) -> None:
assert classify_operation("a x b = <<15*10=150>>150 then <<150*3=450>>450") == "multiplicative"
def test_divisive_when_only_division(self) -> None:
assert classify_operation("<<20/5=4>>4 and <<4+0=4>>4") == "divisive"
def test_additive_when_no_mul_or_div(self) -> None:
assert classify_operation("<<2+4=6>>6 and <<6-1=5>>5") == "additive"
def test_classes_are_closed_set(self) -> None:
for expr in ["<<1*2=2>>", "<<4/2=2>>", "<<1+1=2>>", "no annotations here"]:
assert classify_operation(expr) in OPERATION_CLASSES
# ---------------------------------------------------------------------------
# diagnose_refusal — §8 skill / knowledge / ambiguity router
# ---------------------------------------------------------------------------
class TestDiagnoseRefusal:
def test_empty_injection_is_skill_gap(self) -> None:
r = "candidate_graph: recognizer matched but produced no injection for statement: 'X.' (category=discrete_count_statement)"
assert diagnose_refusal(r) == "skill_gap"
def test_no_admissible_statement_is_knowledge_gap(self) -> None:
assert diagnose_refusal("candidate_graph: no admissible candidate for statement: 'X.'") == "knowledge_gap"
def test_branches_disagree_is_genuine_ambiguity(self) -> None:
assert diagnose_refusal("candidate_graph: branches disagree on answer (distinct values: [1, 2])") == "genuine_ambiguity"
def test_every_diagnosis_in_closed_set(self) -> None:
for r in ["anything", "", "no admissible candidate for question: 'Q?'", "no branch produced a solvable graph"]:
assert diagnose_refusal(r) in REFUSAL_DIAGNOSES
# ---------------------------------------------------------------------------
# run_practice — ledger wiring + diagnosis + elimination records
# ---------------------------------------------------------------------------
class TestRunPractice:
def test_ledger_is_per_class_and_totals_match(self) -> None:
cases = [
_case("m1", "<<2*3=6>>", 6.0),
_case("m2", "<<2*3=6>>", 6.0),
_case("a1", "<<2+3=5>>", 5.0),
]
def scorer(adapted): # all refuse
return _StubOutcome(adapted["id"], "refused", reason="no admissible candidate for statement: 'x'")
rep = run_practice(cases, scorer=scorer)
assert sum(t.attempted for t in rep.ledger.values()) == 3
assert rep.ledger["multiplicative"].refused == 2
assert rep.ledger["additive"].refused == 1
def test_reliability_flows_from_phase1(self) -> None:
cases = [_case(f"m{i}", "<<2*3=6>>", 6.0) for i in range(40)]
def scorer(adapted):
return _StubOutcome(adapted["id"], "correct", actual_answer=6.0)
rep = run_practice(cases, scorer=scorer)
tally = rep.ledger["multiplicative"]
assert tally.correct == 40
assert tally.reliability == conservative_floor(40, 40)
assert tally.reliability >= 0.85 # 40 clean commitments clears propose
def test_wrong_emits_elimination_record(self) -> None:
cases = [_case("w1", "<<2*3=6>>", 6.0)]
def scorer(adapted):
return _StubOutcome(adapted["id"], "wrong", reason="got 5", actual_answer=5.0)
rep = run_practice(cases, scorer=scorer)
assert rep.counts["wrong"] == 1
assert len(rep.elimination_records) == 1
rec = rep.elimination_records[0]
assert rec.case_id == "w1"
assert rec.class_name == "multiplicative"
assert rec.attempted == 5.0
assert rec.gold == 6.0
def test_refusal_diagnosed(self) -> None:
cases = [_case("r1", "<<2*3=6>>", 6.0)]
def scorer(adapted):
return _StubOutcome(adapted["id"], "refused", reason="candidate_graph: branches disagree on answer (distinct values: [1, 2])")
rep = run_practice(cases, scorer=scorer)
assert rep.refusal_diagnoses["r1"] == "genuine_ambiguity"
def test_practice_tolerates_wrong_no_exit_failure(self) -> None:
# practice does NOT gate on wrong==0 (that is serving's contract)
cases = [_case("w1", "<<2*3=6>>", 6.0)]
def scorer(adapted):
return _StubOutcome(adapted["id"], "wrong", reason="x", actual_answer=99.0)
rep = run_practice(cases, scorer=scorer)
assert rep.counts["wrong"] == 1 # recorded, not rejected
assert rep.as_dict()["regime"] == "practice"
def test_elimination_record_is_frozen(self) -> None:
rec = EliminationRecord("c", "multiplicative", 1.0, 2.0, "r")
with pytest.raises(dataclasses.FrozenInstanceError):
rec.gold = 9.0 # type: ignore[misc]
# ---------------------------------------------------------------------------
# Live lane over the real 47 — mirrors serving on the current pipeline
# ---------------------------------------------------------------------------
class TestLiveLane:
def test_live_practice_mirrors_serving_today(self) -> None:
# With the refuse-preferring engine, practice == serving (3/47/0).
# Attempts/eliminations go live in Phase 3.
rep = build_report()
assert rep.counts == {"correct": 3, "wrong": 0, "refused": 47}
assert len(rep.elimination_records) == 0 # no wrongs yet
def test_every_refusal_is_diagnosed(self) -> None:
rep = build_report()
assert len(rep.refusal_diagnoses) == 47
assert all(d in REFUSAL_DIAGNOSES for d in rep.refusal_diagnoses.values())
# ---------------------------------------------------------------------------
# Invariant #1 — the seal (nothing leaks to serving)
# ---------------------------------------------------------------------------
class TestSealInvariant:
def test_practice_does_not_change_serving_score(self) -> None:
from evals.gsm8k_math.train_sample.v1.runner import (
_CASES_PATH,
_load_cases,
build_report as serving_build_report,
)
build_report() # run practice
serving = serving_build_report(_load_cases(_CASES_PATH))
assert serving["counts"] == {"correct": 3, "wrong": 0, "refused": 47}
def test_no_serving_module_imports_the_practice_lane(self) -> None:
import subprocess
from pathlib import Path
repo = Path(__file__).resolve().parents[1]
# the engine/serving path must not import the practice lane
out = subprocess.run(
["grep", "-rl", "practice.v1.runner", "--include=*.py", "generate", "chat"],
cwd=repo, capture_output=True, text=True,
)
assert out.stdout.strip() == ""
# ---------------------------------------------------------------------------
# Invariant #3 — determinism / replay
# ---------------------------------------------------------------------------
class TestDeterminismInvariant:
def test_report_byte_identical_across_runs(self) -> None:
import json
a = json.dumps(build_report().as_dict(), sort_keys=True)
b = json.dumps(build_report().as_dict(), sort_keys=True)
assert a == b