feat(comprehension): deterministic multi-organ setup router (N3)
core/comprehension_attempt/router.py: route_setup tries R1 then R2, collects both attempts, and selects a setup ONLY when exactly one organ admits one. Zero->all_refused; >=2 with differing signatures->ambiguous (refuse, never pick); cross-organ signatures never coincide so two admitting organs always resolve to ambiguous. No scoring, no solving. wrong=0 invariant asserted: every routed R2 setup is byte-equal to the independent gold signature. Each gold problem routes to exactly its owning organ (R2 well-formed->r2; R1 admitted->r1; refused->all_refused); none is ambiguous. 4 router tests.
This commit is contained in:
parent
67c54d6813
commit
5ae516f4a2
3 changed files with 117 additions and 1 deletions
|
|
@ -8,5 +8,15 @@ from __future__ import annotations
|
|||
|
||||
from core.comprehension_attempt.classify import classify_r1, classify_r2
|
||||
from core.comprehension_attempt.model import ComprehensionAttempt, Organ, Outcome
|
||||
from core.comprehension_attempt.router import RouteResult, RouteStatus, route_setup
|
||||
|
||||
__all__ = ["ComprehensionAttempt", "Organ", "Outcome", "classify_r1", "classify_r2"]
|
||||
__all__ = [
|
||||
"ComprehensionAttempt",
|
||||
"Organ",
|
||||
"Outcome",
|
||||
"RouteResult",
|
||||
"RouteStatus",
|
||||
"classify_r1",
|
||||
"classify_r2",
|
||||
"route_setup",
|
||||
]
|
||||
|
|
|
|||
54
core/comprehension_attempt/router.py
Normal file
54
core/comprehension_attempt/router.py
Normal file
|
|
@ -0,0 +1,54 @@
|
|||
"""Deterministic multi-organ setup router (N3).
|
||||
|
||||
Boring on purpose: attempt the R1 and R2 setup compilers, collect their typed attempts, and
|
||||
select a setup ONLY when exactly one organ produced an admissible one. No dynamic "best"
|
||||
scoring, no priority heuristics.
|
||||
|
||||
```text
|
||||
exactly one setup_correct -> routed (use it)
|
||||
zero setup_correct -> all_refused (classify downstream)
|
||||
>= 2 setup_correct, signatures agree-> routed (organs concur)
|
||||
>= 2 setup_correct, signatures differ-> ambiguous (refuse — never pick)
|
||||
```
|
||||
|
||||
Cross-organ signatures are produced by different functions and never coincide, so in practice
|
||||
two admitting organs always resolve to ``ambiguous``. The router never solves and never emits
|
||||
``setup_wrong`` — that is an eval-only outcome; against gold the routed setup must match (the
|
||||
wrong=0 invariant, asserted by the router tests).
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from dataclasses import dataclass
|
||||
from typing import Literal
|
||||
|
||||
from core.comprehension_attempt.classify import classify_r1, classify_r2
|
||||
from core.comprehension_attempt.model import ComprehensionAttempt
|
||||
|
||||
RouteStatus = Literal["routed", "all_refused", "ambiguous"]
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class RouteResult:
|
||||
"""The outcome of routing one problem across the organs: every attempt, the selected setup
|
||||
(or ``None``), and the routing status."""
|
||||
|
||||
attempts: tuple[ComprehensionAttempt, ...]
|
||||
selected: ComprehensionAttempt | None
|
||||
status: RouteStatus
|
||||
|
||||
|
||||
def route_setup(text: str, *, case_id: str | None = None) -> RouteResult:
|
||||
"""Route *text* to the single organ that admits an honest setup, or refuse."""
|
||||
attempts = (classify_r1(text, case_id=case_id), classify_r2(text, case_id=case_id))
|
||||
correct = tuple(a for a in attempts if a.is_setup_correct)
|
||||
if len(correct) == 1:
|
||||
return RouteResult(attempts, correct[0], "routed")
|
||||
if not correct:
|
||||
return RouteResult(attempts, None, "all_refused")
|
||||
if len({a.setup_signature for a in correct}) == 1:
|
||||
return RouteResult(attempts, correct[0], "routed")
|
||||
return RouteResult(attempts, None, "ambiguous")
|
||||
|
||||
|
||||
__all__ = ["RouteResult", "RouteStatus", "route_setup"]
|
||||
52
tests/test_setup_router.py
Normal file
52
tests/test_setup_router.py
Normal file
|
|
@ -0,0 +1,52 @@
|
|||
"""Tests for the deterministic multi-organ setup router (N3).
|
||||
|
||||
Pins that each problem routes to exactly the organ that owns it (R1 relational, R2 constraint),
|
||||
that unreadable problems refuse, that no gold problem is read by both organs (no ambiguity), and
|
||||
— the wrong=0 invariant — that every routed R2 setup matches the independent gold signature.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from core.comprehension_attempt import route_setup
|
||||
from core.comprehension_attempt.classify import _r2_signature
|
||||
from evals.constraint_oracle.runner import _load_r2_gold, gold_to_problem
|
||||
from evals.setup_oracle.runner import _load_r1_gold
|
||||
|
||||
|
||||
def test_r2_well_formed_routes_to_r2_and_matches_gold_signature() -> None:
|
||||
for fx in _load_r2_gold():
|
||||
result = route_setup(fx["text"], case_id=fx["id"])
|
||||
if fx["expect"] in ("solved", "solver_refuses"):
|
||||
assert result.status == "routed", f"{fx['id']}: {result.status}"
|
||||
assert result.selected is not None and result.selected.organ == "r2_constraints"
|
||||
# wrong=0: the routed setup is byte-equal to the independent gold setup.
|
||||
assert result.selected.setup_signature == _r2_signature(gold_to_problem(fx)), fx["id"]
|
||||
else: # reader_refuses
|
||||
assert result.status == "all_refused" and result.selected is None
|
||||
|
||||
|
||||
def test_r1_admitted_routes_to_r1_refused_all_refused() -> None:
|
||||
routed = refused = 0
|
||||
for fx in _load_r1_gold():
|
||||
result = route_setup(fx["text"], case_id=fx["id"])
|
||||
if result.status == "routed":
|
||||
routed += 1
|
||||
assert result.selected.organ == "r1_quantitative", fx["id"]
|
||||
else:
|
||||
refused += 1
|
||||
assert result.status == "all_refused"
|
||||
assert routed == 7 and refused == 3
|
||||
|
||||
|
||||
def test_no_gold_problem_is_ambiguous() -> None:
|
||||
# Each organ is exclusive on the gold corpora — no text is admitted by both.
|
||||
for fx in _load_r2_gold() + _load_r1_gold():
|
||||
assert route_setup(fx["text"]).status != "ambiguous", fx.get("id")
|
||||
|
||||
|
||||
def test_router_never_selects_a_refusal() -> None:
|
||||
for fx in _load_r2_gold() + _load_r1_gold():
|
||||
result = route_setup(fx["text"])
|
||||
if result.selected is not None:
|
||||
assert result.selected.outcome == "setup_correct"
|
||||
assert len(result.attempts) == 2 # always one R1 + one R2 attempt
|
||||
Loading…
Reference in a new issue