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

feat(learning-arena): ADR-0199 PR-2 — extract domain-agnostic run_practice (#516)

Browse source
This commit is contained in:
Shay 2026-05-31 21:07:23 -07:00 committed by GitHub
parent 1d12c3b01e
commit f8b6f91627
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
6 changed files with 541 additions and 102 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

38
core/learning_arena/__init__.py Normal file
View file

@ -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",
]

98
core/learning_arena/engine.py Normal file
View file

@ -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),
)

107
core/learning_arena/protocols.py Normal file
View file

@ -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 = ""

77
core/learning_arena/report.py Normal file
View file

@ -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
],
}

181
evals/gsm8k_math/practice/v1/runner.py
View file

@ -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 ``<<a*b=c>>`` 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]]:

142
tests/test_adr_0199_learning_arena_engine.py Normal file
View file

@ -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)