diff --git a/core/learning_arena/__init__.py b/core/learning_arena/__init__.py new file mode 100644 index 00000000..8bd4e393 --- /dev/null +++ b/core/learning_arena/__init__.py @@ -0,0 +1,38 @@ +"""ADR-0199 — cross-domain learning arena. + +The shared engine + interfaces every base subject plugs into. Domains live +outside this package (e.g. ``evals/gsm8k_math/practice``); this package never +imports a concrete domain. +""" + +from __future__ import annotations + +from core.learning_arena.engine import run_practice +from core.learning_arena.protocols import ( + Attempt, + BaseAttempt, + DomainProblem, + DomainSolver, + GoldTether, + Problem, +) +from core.learning_arena.report import ( + REFUSAL_DIAGNOSES, + EliminationRecord, + PracticeReport, + bucket_counts, +) + +__all__ = [ + "run_practice", + "Attempt", + "BaseAttempt", + "DomainProblem", + "DomainSolver", + "GoldTether", + "Problem", + "REFUSAL_DIAGNOSES", + "EliminationRecord", + "PracticeReport", + "bucket_counts", +] diff --git a/core/learning_arena/engine.py b/core/learning_arena/engine.py new file mode 100644 index 00000000..c5b16583 --- /dev/null +++ b/core/learning_arena/engine.py @@ -0,0 +1,98 @@ +"""ADR-0199 §2.2 — the domain-agnostic practice engine. + +This is the extraction of the GSM8K ``run_practice`` fold into a subject-neutral +core. It is the **only** new per-domain code path a subject needs to reach: a +subject supplies a :class:`DomainSolver` + :class:`GoldTether` and gets a +:class:`PracticeReport` whose ``.ledger`` is the ``dict[str, ClassTally]`` the +reliability gate (``propose_from_ledger``) consumes unchanged. + +Invariants (the load-bearing ADR-0199 mandates, enforced structurally here): + +- **L-1 (one floor).** Reliability is computed only via :class:`ClassTally` + (which calls the single pinned ``conservative_floor``). This module defines + no pessimism constant of its own. +- **L-3 (seal).** ``run_practice`` returns a report and mutates nothing. It + never touches a serving path or the active teaching corpus. Promotion is the + caller's separate ``propose_from_ledger`` step into the reviewed corridor. +- **L-4 (determinism).** Pure fold over the input order; identical + (problems, solver, tether, diagnose) -> identical report. +""" + +from __future__ import annotations + +from typing import Callable, Sequence + +from core.learning_arena.protocols import Attempt, DomainProblem, DomainSolver, GoldTether +from core.learning_arena.report import EliminationRecord, PracticeReport +from core.reliability_gate import ClassTally + + +def _default_diagnose(_reason: str) -> str: + """Conservative default: assume a missing piece (ADR-0175 §8). + + A domain supplies its own router (e.g. a refusal-reason vocabulary) via the + ``diagnose`` parameter; absent one, refusals are attributed to a knowledge + gap rather than silently dropped. + """ + return "knowledge_gap" + + +def run_practice( + problems: Sequence[DomainProblem], + solver: DomainSolver, + tether: GoldTether, + *, + diagnose: Callable[[str], str] = _default_diagnose, +) -> PracticeReport: + """Sealed practice: attempt -> gold-tether score -> per-class ledger. + + For each problem, in input order: the solver attempts it; the verdict is + ``refused`` when the attempt is uncommitted, else ``correct``/``wrong`` per + the tether's independent gold check. Counts and per-class :class:`ClassTally` + accumulate; each wrong yields an :class:`EliminationRecord`; each refusal is + routed by ``diagnose``. + """ + counts = {"correct": 0, "wrong": 0, "refused": 0} + ledger: dict[str, ClassTally] = {} + diagnoses: dict[str, str] = {} + elims: list[EliminationRecord] = [] + + for problem in problems: + cls = problem.class_name + attempt: Attempt = solver.attempt(problem) + + if not attempt.committed: + verdict = "refused" + elif tether.is_correct(attempt, problem): + verdict = "correct" + else: + verdict = "wrong" + + 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=attempt.case_id, + class_name=cls, + attempted=attempt.answer, + gold=tether.gold_answer(problem), + reason=attempt.reason or "", + ) + ) + else: # refused + tally = tally.record(refused=1) + diagnoses[attempt.case_id] = diagnose(attempt.reason or "") + + ledger[cls] = tally + + return PracticeReport( + counts=counts, + ledger=ledger, + refusal_diagnoses=diagnoses, + elimination_records=tuple(elims), + ) diff --git a/core/learning_arena/protocols.py b/core/learning_arena/protocols.py new file mode 100644 index 00000000..8ff1637a --- /dev/null +++ b/core/learning_arena/protocols.py @@ -0,0 +1,107 @@ +"""ADR-0199 §2.2 — the cross-domain learning-arena interfaces. + +A subject becomes a learning arena by supplying four domain-specific pieces +(``DomainSolver``, a gold anchor set, capability classes, a Tier-2 verifier) +and reusing the shared engine (:mod:`core.learning_arena.engine`) and the +shared reliability gate (:mod:`core.reliability_gate`) unchanged. + +These protocols are structural (PEP 544). A domain provides concrete classes; +the engine never imports a concrete domain. The first instance is GSM8K math +(``evals/gsm8k_math/practice/v1/runner.py``), re-expressed against this +contract with no behavior change. + +Note on the ADR's illustrative signatures: the ADR sketched +``is_correct(attempt, problem_id)``. We pass the whole ``DomainProblem`` (which +carries its ``problem_id``) so a tether can reach class/payload without a +separate lookup table — strictly more general, same contract. +""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from typing import Any, Protocol, runtime_checkable + + +@runtime_checkable +class DomainProblem(Protocol): + """One problem in a practice arena. + + ``class_name`` is the capability axis this problem exercises (the ledger + key); it is resolved up front by a domain adapter that may consult gold. + ``payload`` is opaque to the engine — only the domain's solver/tether read + it. + """ + + problem_id: str + class_name: str + payload: Any + + +@runtime_checkable +class Attempt(Protocol): + """The result of a single attempt. + + ``committed is False`` means the engine refused (always safe; excluded + from reliability's denominator per ADR-0175 §4). ``derivations`` are the + ≥2 structurally-distinct paths a Tier-2 verifier inspects; ``trace_sha256`` + is replayable provenance carrying no raw content beyond hashes. + """ + + committed: bool + answer: Any + reason: str + case_id: str + derivations: tuple[Any, ...] + trace_sha256: str + + +@runtime_checkable +class DomainSolver(Protocol): + """Attempts a grounded derivation over the subject's operations. + + This is where intelligence lives (ADR-0175 Pivot-2). The engine calls + :meth:`attempt` once per problem and never inspects how the answer was + reached beyond the :class:`Attempt` fields. + """ + + domain_id: str + + def attempt(self, problem: DomainProblem) -> Attempt: ... + + +@runtime_checkable +class GoldTether(Protocol): + """The Tier-1 truth anchor for a subject. + + ADR-0199 mandate 2: the truth ``is_correct`` consults must come from a + source **independent of the solver's derivation** (proof obligation L-2). + For dataset domains the gold is the dataset's own answer; for software it + is execution; etc. + """ + + domain_id: str + + def is_correct(self, attempt: Attempt, problem: DomainProblem) -> bool: ... + + def gold_answer(self, problem: DomainProblem) -> Any: ... + + +@dataclass(frozen=True, slots=True) +class Problem: + """Concrete :class:`DomainProblem` a domain adapter can build directly.""" + + problem_id: str + class_name: str + payload: Any = None + + +@dataclass(frozen=True, slots=True) +class BaseAttempt: + """Concrete :class:`Attempt` for domains that need no extra fields.""" + + committed: bool + answer: Any = None + reason: str = "" + case_id: str = "" + derivations: tuple[Any, ...] = field(default_factory=tuple) + trace_sha256: str = "" diff --git a/core/learning_arena/report.py b/core/learning_arena/report.py new file mode 100644 index 00000000..f4ff2d40 --- /dev/null +++ b/core/learning_arena/report.py @@ -0,0 +1,77 @@ +"""ADR-0199 / ADR-0175 — the domain-agnostic practice report. + +Extracted verbatim (schema-preserving) from +``evals/gsm8k_math/practice/v1/runner.py`` so every subject's arena emits the +same report shape. ``PracticeReport.as_dict`` is byte-stable with the original +GSM8K report so existing goldens and ``report.json`` are unaffected. + +The three refusal-diagnosis axes are the universal ADR-0175 §8 router +(skill / knowledge / ambiguity), not a domain quantity — so they live here. +""" + +from __future__ import annotations + +from dataclasses import dataclass +from typing import Any, Mapping + +from core.reliability_gate import ClassTally + +# ADR-0175 §8 — the universal "name the missing piece" axes. +REFUSAL_DIAGNOSES: tuple[str, ...] = ("skill_gap", "knowledge_gap", "genuine_ambiguity") + + +@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 + + +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 + + +@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 + ], + } diff --git a/evals/gsm8k_math/practice/v1/runner.py b/evals/gsm8k_math/practice/v1/runner.py index dde2a925..41b243b1 100644 --- a/evals/gsm8k_math/practice/v1/runner.py +++ b/evals/gsm8k_math/practice/v1/runner.py @@ -1,42 +1,52 @@ """ADR-0175 Phase 2 — sealed practice lane over the GSM8K train sample. +ADR-0199: this lane is now the **first instance** of the cross-domain learning +arena. The domain-agnostic fold lives in :mod:`core.learning_arena.engine`; this +module supplies only the GSM8K-specific pieces — the operation classifier +(capability classes from gold), the refusal-reason router, and the +solver/gold-tether adapters around the existing candidate-graph scorer. Behavior +is unchanged: the public surface (``run_practice(cases, scorer=...)``, +``build_report``, ``build_practice_report``, ``PracticeReport``, +``EliminationRecord``, ``classify_operation``, ``diagnose_refusal``) is +preserved byte-for-byte against the prior lane. + 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. +which is **never modified**. Runs the cases in *practice* mode: wrong is the +learning signal, not a lane failure. 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 dataclasses import dataclass, field from pathlib import Path -from typing import Any, Callable, Mapping +from typing import Any, Callable -from core.reliability_gate import ClassTally +from core.learning_arena.engine import run_practice as _engine_run_practice +from core.learning_arena.protocols import Problem +# Re-exported so existing callers/tests keep importing these from the lane. +from core.learning_arena.report import ( # noqa: F401 + REFUSAL_DIAGNOSES, + EliminationRecord, + PracticeReport, +) 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" _PRACTICE_CASES_PATH = _HERE / "cases.jsonl" _CALC_RE = re.compile(r"<<([^=>]+)=") +_DOMAIN_ID = "mathematics_logic" + def classify_operation(answer_expression: str) -> str: """Primary gold operation class from GSM8K ``<>`` calc annotations. @@ -75,61 +85,61 @@ def diagnose_refusal(reason: str) -> str: return "knowledge_gap" -@dataclass(frozen=True, slots=True) -class EliminationRecord: - """A wrong practice attempt that gold caught — the pruning signal (§9).""" +# --- GSM8K instance of the ADR-0199 DomainSolver / GoldTether ------------------ - case_id: str - class_name: str - attempted: float | None - gold: float + +@dataclass(frozen=True, slots=True) +class _GSM8KAttempt: + """Concrete Attempt that also carries the scorer's gold verdict. + + The candidate-graph scorer already decides correct/wrong/refused against the + dataset's ``expected_answer`` (gold independent of the engine's derivation — + ADR-0199 L-2). The tether reads that verdict via ``scorer_outcome`` so the + classification is reproduced exactly, not re-derived. + """ + + committed: bool + answer: Any reason: str + case_id: str + scorer_outcome: str + derivations: tuple[Any, ...] = field(default_factory=tuple) + trace_sha256: 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, ...] +class _GSM8KSolver: + score: Callable[[dict[str, Any]], Any] + domain_id: str = _DOMAIN_ID - 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 attempt(self, problem: Problem) -> _GSM8KAttempt: + outcome = self.score(_adapt(problem.payload)) + return _GSM8KAttempt( + committed=(outcome.outcome != "refused"), + answer=getattr(outcome, "actual_answer", None), + reason=outcome.reason or "", + case_id=outcome.case_id, + scorer_outcome=outcome.outcome, + ) -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 +@dataclass(frozen=True, slots=True) +class _GSM8KGoldTether: + domain_id: str = _DOMAIN_ID + + def is_correct(self, attempt: _GSM8KAttempt, problem: Problem) -> bool: + return attempt.scorer_outcome == "correct" + + def gold_answer(self, problem: Problem) -> float: + return float(problem.payload["answer_numeric"]) + + +def _to_problem(raw: dict[str, Any]) -> Problem: + return Problem( + problem_id=str(raw.get("id", raw.get("case_id", ""))), + class_name=classify_operation(raw.get("answer_expression", "")), + payload=raw, + ) def run_practice( @@ -139,49 +149,16 @@ def run_practice( ) -> 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. + Unchanged signature and behavior. ``scorer`` is injectable for testing; it + defaults to the candidate-graph scorer. The fold is delegated to the + domain-agnostic :func:`core.learning_arena.engine.run_practice` (ADR-0199); + this lane supplies the GSM8K solver/tether and the §8 diagnosis router. """ 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), - ) + solver = _GSM8KSolver(score) + tether = _GSM8KGoldTether() + problems = [_to_problem(raw) for raw in cases] + return _engine_run_practice(problems, solver, tether, diagnose=diagnose_refusal) def _load_practice_cases(path: Path = _PRACTICE_CASES_PATH) -> list[dict[str, Any]]: diff --git a/tests/test_adr_0199_learning_arena_engine.py b/tests/test_adr_0199_learning_arena_engine.py new file mode 100644 index 00000000..c75e9412 --- /dev/null +++ b/tests/test_adr_0199_learning_arena_engine.py @@ -0,0 +1,142 @@ +"""ADR-0199 PR-2 — the cross-domain learning-arena engine. + +Proves exactly the PR-2 gate: the extracted engine reuses the single pinned +floor (L-1), holds the seal (L-3), is a deterministic fold (L-4), and the GSM8K +math instance behaves byte-identically to before (the committed golden queue). +Tier-2 scoring / L-5 are deferred to the PR that wires t2 verification. +""" + +from __future__ import annotations + +import json +import subprocess +from dataclasses import dataclass +from pathlib import Path +from typing import Any + +from core.learning_arena import ( + BaseAttempt, + Problem, + run_practice, +) +from core.reliability_gate import conservative_floor + +_REPO = Path(__file__).resolve().parents[1] + + +# --- a tiny synthetic domain (no heavy deps) ---------------------------------- + + +@dataclass(frozen=True, slots=True) +class _StubSolver: + """Reads the verdict the synthetic payload declares; pure and total.""" + + domain_id: str = "stub" + + def attempt(self, problem: Problem) -> BaseAttempt: + p = problem.payload + return BaseAttempt( + committed=p["verdict"] != "refused", + answer=p.get("answer"), + reason=p.get("reason", ""), + case_id=problem.problem_id, + ) + + +@dataclass(frozen=True, slots=True) +class _StubTether: + domain_id: str = "stub" + + def is_correct(self, attempt: BaseAttempt, problem: Problem) -> bool: + return problem.payload["verdict"] == "correct" + + def gold_answer(self, problem: Problem) -> float: + return float(problem.payload["gold"]) + + +def _problems() -> list[Problem]: + return [ + Problem("c1", "alpha", {"verdict": "correct", "answer": 9.0, "gold": 9.0}), + Problem("c2", "alpha", {"verdict": "wrong", "answer": 7.0, "gold": 10.0, + "reason": "off by three"}), + Problem("c3", "beta", {"verdict": "refused", "reason": "branches disagree"}), + Problem("c4", "alpha", {"verdict": "correct", "answer": 4.0, "gold": 4.0}), + ] + + +def _diagnose(reason: str) -> str: + return "genuine_ambiguity" if "disagree" in reason else "knowledge_gap" + + +# --- L-4: deterministic fold + correct accounting ----------------------------- + + +def test_engine_counts_ledger_eliminations_diagnoses(): + rep = run_practice(_problems(), _StubSolver(), _StubTether(), diagnose=_diagnose) + assert rep.counts == {"correct": 2, "wrong": 1, "refused": 1} + + alpha = rep.ledger["alpha"] + assert (alpha.correct, alpha.wrong, alpha.refused) == (2, 1, 0) + beta = rep.ledger["beta"] + assert (beta.correct, beta.wrong, beta.refused) == (0, 0, 1) + + assert len(rep.elimination_records) == 1 + elim = rep.elimination_records[0] + assert (elim.case_id, elim.class_name, elim.attempted, elim.gold) == ( + "c2", "alpha", 7.0, 10.0, + ) + assert rep.refusal_diagnoses == {"c3": "genuine_ambiguity"} + + +def test_engine_is_deterministic(): + a = run_practice(_problems(), _StubSolver(), _StubTether(), diagnose=_diagnose) + b = run_practice(_problems(), _StubSolver(), _StubTether(), diagnose=_diagnose) + assert json.dumps(a.as_dict(), sort_keys=True) == json.dumps(b.as_dict(), sort_keys=True) + + +# --- L-1: one shared pinned floor; no per-arena pessimism constant ------------ + + +def test_engine_reliability_flows_through_shared_floor(): + rep = run_practice(_problems(), _StubSolver(), _StubTether(), diagnose=_diagnose) + alpha = rep.ledger["alpha"] + # reliability is exactly the shared conservative_floor over committed counts. + assert alpha.reliability == conservative_floor(alpha.correct, alpha.committed) + + +def test_learning_arena_defines_no_floor_constants(): + pkg = _REPO / "core" / "learning_arena" + for src in pkg.glob("*.py"): + text = src.read_text(encoding="utf-8") + assert "WILSON_Z" not in text, f"{src.name} must not redefine the floor" + assert "N_MIN" not in text, f"{src.name} must not redefine the floor" + + +# --- L-3: the seal — no serving module imports the arena ---------------------- + + +def test_seal_no_serving_imports_learning_arena(): + res = subprocess.run( + ["grep", "-rl", "learning_arena", "--include=*.py", "generate", "chat"], + cwd=_REPO, capture_output=True, text=True, + ) + assert res.stdout.strip() == "", ( + "a serving module imports the learning arena (seal violation):\n" + res.stdout + ) + + +# --- behavior parity: the GSM8K math instance reproduces its golden ----------- + + +def test_gsm8k_instance_reproduces_committed_queue(): + from evals.gsm8k_math.practice.v1.propose_runner import ( + build_ratification_queue, + resolve_pooled_scorer, + ) + + golden_path = ( + _REPO / "evals" / "gsm8k_math" / "practice" / "v1" / "ratification_queue.json" + ) + golden = json.loads(golden_path.read_text(encoding="utf-8")) + produced = build_ratification_queue(scorer=resolve_pooled_scorer) + assert json.dumps(produced, sort_keys=True) == json.dumps(golden, sort_keys=True)