feat(adr-0175): wire the PROPOSE step — autonomous loop closes (attempt->tether->ledger->propose)

The attempt/score/ledger half existed (run_practice -> ClassTally scored vs
gold); nothing consulted the gate to turn earned reliability into a ratifiable
proposal. Adds core/reliability_gate/propose.py (propose_from_ledger +
RatifiableProposal): for each class, license_for(PROPOSE) emits a proposal iff
its conservative Wilson floor (0 below N_MIN=10) clears theta=0.85. Refusals
never penalize; deterministic; PROPOSAL-ONLY (never a serving mutation).

propose_runner.py closes the loop end-to-end with an aggressive sealed scorer
(resolve_pooled): practice 95c/5w/50r -> ONE proposal (additive, reliability
0.8608>=0.85, 95/100); 5 wrongs tolerated but floor held; rest stayed sealed.
The gold-tethered autonomous contemplation: the engine earns the right to ASK,
not to SERVE. 11 failing-under-violation tests.
This commit is contained in:
Shay 2026-05-30 13:50:24 -07:00
parent 0770648257
commit b82897a0dd
7 changed files with 445 additions and 0 deletions

View file

@ -17,6 +17,7 @@ from core.reliability_gate.ceilings import Action, Ceilings
from core.reliability_gate.floor import N_MIN, WILSON_Z, conservative_floor
from core.reliability_gate.gate import LicenseDecision, license_for
from core.reliability_gate.ledger import ClassTally
from core.reliability_gate.propose import RatifiableProposal, propose_from_ledger
__all__ = [
"Action",
@ -24,7 +25,9 @@ __all__ = [
"ClassTally",
"LicenseDecision",
"N_MIN",
"RatifiableProposal",
"WILSON_Z",
"conservative_floor",
"license_for",
"propose_from_ledger",
]

View file

@ -0,0 +1,103 @@
"""ADR-0175 PROPOSE step — turn a practice ledger into ratifiable proposals.
This is the missing seam of the autonomous attempt-and-eliminate loop:
attempt (sealed practice) -> gold-tether score -> ClassTally ledger
-> **propose_from_ledger (this module)** -> HITL ratification queue
The attempt/score/ledger half already exists (``evals/gsm8k_math/practice``
populates a ``dict[str, ClassTally]`` of correct/wrong/refused scored against
gold the wrong=0 tether). What was missing is the gate consultation that
turns earned reliability into a *ratifiable proposal*: for each class, ask
:func:`license_for` whether the class has cleared its θ for the action.
Doctrine (ADR-0175):
- Reliability is the **conservative Wilson floor** on commitment precision
(``correct / (correct+wrong)``); it is **0 below N_MIN committed trials**, so
a class proposes only on sufficient evidence. Refusals never penalize it.
- A proposal is *proposal-only*. It NEVER touches the serving path it is a
queue entry for the reviewed teaching corridor to ratify. The engine earns
the right to *ask*, not to *serve*.
- Deterministic: proposals are sorted by class name; the same ledger yields
byte-identical proposals.
"""
from __future__ import annotations
from dataclasses import dataclass
from typing import Any, Mapping
from core.reliability_gate.ceilings import Action, Ceilings
from core.reliability_gate.gate import license_for
from core.reliability_gate.ledger import ClassTally
@dataclass(frozen=True, slots=True)
class RatifiableProposal:
"""A class whose earned practice reliability licenses a promotion proposal.
Inspectable by construction it carries the numbers behind the verdict so
a reviewer ratifies on evidence, not on a bare boolean.
"""
class_name: str
action: str # the licensed action ("propose" / "serve")
checker: str # which earned number gated it ("reliability" / "t2_precision")
measured: float # the conservative floor the class earned
required: float # the θ it cleared
correct: int
wrong: int
committed: int
def as_json(self) -> dict[str, Any]:
return {
"class_name": self.class_name,
"action": self.action,
"checker": self.checker,
"measured": self.measured,
"required": self.required,
"correct": self.correct,
"wrong": self.wrong,
"committed": self.committed,
}
def propose_from_ledger(
ledger: Mapping[str, ClassTally],
ceilings: Ceilings,
*,
action: Action = Action.PROPOSE,
checker: str = "reliability",
) -> tuple[RatifiableProposal, ...]:
"""Emit a ratifiable proposal for every class that clears the gate.
For each class tally, consult :func:`license_for`. A class earns a proposal
iff the gate licenses ``action`` i.e. its earned number (reliability, the
conservative Wilson floor that is 0 below ``N_MIN`` committed trials) is
``>=`` the class's θ for ``action``. ``action`` defaults to
:attr:`Action.PROPOSE` (θ=0.85); pass :attr:`Action.SERVE` (θ=0.99) to
query the stricter serve gate.
Returns a deterministic tuple sorted by class name. This is the loop's
output a ratification queue, never a serving mutation.
"""
proposals: list[RatifiableProposal] = []
for class_name in sorted(ledger):
tally = ledger[class_name]
decision = license_for(tally, action, ceilings, checker=checker)
if not decision.licensed:
continue
proposals.append(
RatifiableProposal(
class_name=class_name,
action=action.value,
checker=decision.checker,
measured=decision.measured,
required=decision.required,
correct=tally.correct,
wrong=tally.wrong,
committed=tally.committed,
)
)
return tuple(proposals)

View file

@ -431,3 +431,37 @@ serving yet.
capability axes G1G5, round-trip filter + disagreement rule, teaching-safety.
- **Anti-overfitting obligations:** ADR-0114a (perturbation / OOD / depth / adversarial axes apply to every flipped case).
- **Thesis:** [[thesis-decoding-not-generating]] — find, comprehend, rationalize; not a library of founds.
---
## Implemented — the PROPOSE step (autonomous loop closed)
The attempt→score→ledger half already existed (`evals/gsm8k_math/practice`
populates `dict[str, ClassTally]` scored against gold — the wrong=0 tether). The
**missing seam was the gate consultation that turns earned reliability into a
ratifiable proposal.** Nothing called `license_for` outside the gate itself.
Now wired:
```text
run_practice (attempt → gold-tether score → ClassTally ledger)
→ propose_from_ledger (core/reliability_gate/propose.py — the PROPOSE gate)
→ ratification_queue.json (HITL queue; NEVER a serving mutation)
```
- `propose_from_ledger(ledger, ceilings, action=PROPOSE)` emits a
`RatifiableProposal` for every class whose reliability (the conservative
Wilson floor, **0 below N_MIN=10 committed**) clears θ (PROPOSE=0.85). Refusals
never penalize; deterministic, sorted; proposal-only.
- `propose_runner.py` closes the loop end-to-end. With an **aggressive sealed
scorer** (`resolve_pooled`) over the 150-case practice set it produced, on
first run: practice `95 correct / 5 wrong / 50 refused`, and **one** ratifiable
proposal — `additive`, reliability `0.8608 ≥ 0.85` (95/100). The 5 wrongs were
tolerated (attempt-and-eliminate) but did not breach the floor; every other
class stayed sealed. This is the gold-tethered autonomous contemplation: the
engine earns the right to *ask*, not to *serve*.
Out of scope (next seams): ratification *consumption* (promote a ratified class
into serving — the ADR-0175 Phase-5 bridge), and using the frontier-shift
instrument to *prioritize* which classes to attempt. The PROPOSE gate is the
load-bearing piece that makes both safe.

View file

@ -0,0 +1,111 @@
"""ADR-0175 PROPOSE runner — close the autonomous loop end-to-end.
run_practice (attempt gold-tether score ClassTally ledger)
propose_from_ledger (the PROPOSE gate: reliability θ on N N_MIN)
ratification_queue.json (the HITL queue NEVER a serving change)
This is the wiring that was missing: the attempt/score/ledger half already
existed; nothing consulted the gate to turn earned reliability into a
ratifiable proposal. The output is *proposal-only* a queue for the reviewed
teaching corridor, not a serving mutation. No serving module imports this.
The attempt ``scorer`` is injectable. The default is the practice lane's own
scorer (the serving candidate-graph). Injecting an *aggressive* sealed scorer
(e.g. ``resolve_pooled``) is what makes this a real attempt-and-eliminate
regime the gate then filters classes the aggressive reader gets wrong while
proposing the ones it reads reliably.
"""
from __future__ import annotations
import json
from pathlib import Path
from typing import Any, Callable
from core.reliability_gate import Action, Ceilings, propose_from_ledger
from evals.gsm8k_math.practice.v1.runner import (
_load_practice_cases,
run_practice,
)
_HERE = Path(__file__).resolve().parent
_QUEUE_PATH = _HERE / "ratification_queue.json"
def resolve_pooled_scorer(adapted: dict[str, Any]) -> Any:
"""An AGGRESSIVE sealed scorer: attempt with the derivation reader
(``resolve_pooled``) and score against gold (the wrong=0 tether).
Unlike the conservative serving scorer, this reader *commits* on shapes
serving refuses so its practice ledger carries real ``wrong``. That is
the attempt-and-eliminate regime: the PROPOSE gate then proposes only the
classes it reads *reliably* and leaves the rest sealed.
"""
from evals.gsm8k_math.runner import CaseOutcome
from generate.derivation.pool import resolve_pooled
expected = float(adapted["expected_answer"])
unit = adapted.get("expected_unit", "") or ""
resolution = resolve_pooled(adapted["problem"])
if resolution is None:
return CaseOutcome(
case_id=adapted["id"], outcome="refused",
reason="resolve_pooled: no resolution", expected_answer=expected,
expected_unit=unit, actual_answer=None, actual_unit=None,
trace_hash=None, realized_prose=None,
)
value = float(resolution.answer)
correct = abs(value - expected) < 1e-6
return CaseOutcome(
case_id=adapted["id"], outcome="correct" if correct else "wrong",
reason="resolve_pooled", expected_answer=expected, expected_unit=unit,
actual_answer=value, actual_unit=getattr(resolution, "answer_unit", None),
trace_hash=None, realized_prose=None,
)
def build_ratification_queue(
*,
ceilings: Ceilings | None = None,
scorer: Callable[[dict[str, Any]], Any] | None = None,
cases: list[dict[str, Any]] | None = None,
) -> dict[str, Any]:
"""Run practice, consult the PROPOSE gate, return the ratification queue."""
ceilings = ceilings if ceilings is not None else Ceilings(())
report = run_practice(
cases if cases is not None else _load_practice_cases(), scorer=scorer
)
proposals = propose_from_ledger(report.ledger, ceilings, action=Action.PROPOSE)
return {
"schema_version": 1,
"adr": "0175",
"regime": "propose",
"note": "proposal-only ratification queue; never a serving mutation",
"practice_counts": dict(sorted(report.counts.items())),
"proposals": [p.as_json() for p in proposals],
"proposal_count": len(proposals),
}
def write_queue(queue: dict[str, Any], path: Path = _QUEUE_PATH) -> None:
path.write_text(json.dumps(queue, indent=2, sort_keys=True) + "\n", encoding="utf-8")
def main() -> int:
# Default the runnable demo to the aggressive sealed regime so the gate
# filters real wrong>0 — the point of attempt-and-eliminate.
queue = build_ratification_queue(scorer=resolve_pooled_scorer)
write_queue(queue)
print(f"practice counts: {queue['practice_counts']}")
print(f"ratifiable proposals: {queue['proposal_count']}")
for p in queue["proposals"]:
print(
f" {p['class_name']}: reliability={p['measured']:.4f} "
f">= {p['required']} (correct={p['correct']} wrong={p['wrong']} "
f"committed={p['committed']})"
)
return 0
if __name__ == "__main__":
raise SystemExit(main())

View file

@ -0,0 +1,24 @@
{
"adr": "0175",
"note": "proposal-only ratification queue; never a serving mutation",
"practice_counts": {
"correct": 95,
"refused": 50,
"wrong": 5
},
"proposal_count": 1,
"proposals": [
{
"action": "propose",
"checker": "reliability",
"class_name": "additive",
"committed": 100,
"correct": 95,
"measured": 0.86084162,
"required": 0.85,
"wrong": 5
}
],
"regime": "propose",
"schema_version": 1
}

View file

@ -0,0 +1,90 @@
"""ADR-0175 PROPOSE step — failing-under-violation guards.
The autonomous loop earns the right to *ask* (propose) only when a class clears
its θ on sufficient evidence. Each test fails if the gate is silently bypassed.
"""
from __future__ import annotations
from core.reliability_gate import (
Action,
Ceilings,
ClassTally,
N_MIN,
propose_from_ledger,
)
_CEILINGS = Ceilings(()) # global defaults: PROPOSE θ=0.85, SERVE θ=0.99
def _reliable() -> ClassTally:
# 50 committed, all correct → Wilson floor well above 0.85.
return ClassTally("reliable").record(correct=50)
def _below_theta() -> ClassTally:
# Enough evidence but imprecise → floor below 0.85.
return ClassTally("sloppy").record(correct=10).record(wrong=5)
def _insufficient() -> ClassTally:
# Perfect but under N_MIN committed → reliability is 0 (no evidence).
return ClassTally("thin").record(correct=N_MIN - 1)
def _all_refused() -> ClassTally:
return ClassTally("shy").record(refused=20)
def test_reliable_class_proposes() -> None:
props = propose_from_ledger({"reliable": _reliable()}, _CEILINGS)
assert [p.class_name for p in props] == ["reliable"]
p = props[0]
assert p.action == "propose"
assert p.measured >= p.required >= 0.85
assert p.correct == 50 and p.wrong == 0 and p.committed == 50
def test_below_theta_does_not_propose() -> None:
assert propose_from_ledger({"sloppy": _below_theta()}, _CEILINGS) == ()
def test_insufficient_evidence_does_not_propose() -> None:
# The N_MIN floor: a perfect-but-thin class earns NO proposal.
assert propose_from_ledger({"thin": _insufficient()}, _CEILINGS) == ()
def test_all_refused_does_not_propose() -> None:
# Refusals never earn a proposal (committed == 0 → reliability 0).
assert propose_from_ledger({"shy": _all_refused()}, _CEILINGS) == ()
def test_serve_gate_is_stricter_than_propose() -> None:
# A class licensed to PROPOSE (θ=0.85) need not be licensed to SERVE (θ=0.99).
ledger = {"reliable": _reliable()}
proposes = propose_from_ledger(ledger, _CEILINGS, action=Action.PROPOSE)
serves = propose_from_ledger(ledger, _CEILINGS, action=Action.SERVE)
assert proposes and not serves
def test_deterministic_and_sorted() -> None:
ledger = {
"zeta": _reliable(),
"alpha": ClassTally("alpha").record(correct=40),
"sloppy": _below_theta(),
}
a = propose_from_ledger(ledger, _CEILINGS)
b = propose_from_ledger(ledger, _CEILINGS)
assert a == b
assert [p.class_name for p in a] == ["alpha", "zeta"] # sorted; sloppy dropped
def test_mixed_ledger_only_reliable_classes_propose() -> None:
ledger = {
"reliable": _reliable(),
"sloppy": _below_theta(),
"thin": _insufficient(),
"shy": _all_refused(),
}
props = propose_from_ledger(ledger, _CEILINGS)
assert [p.class_name for p in props] == ["reliable"]

View file

@ -0,0 +1,80 @@
"""ADR-0175 PROPOSE runner — the loop closes: attempt → tether → ledger → propose.
Fast + deterministic via an injected scorer + cases (no heavy reader run).
"""
from __future__ import annotations
from typing import Any
from evals.gsm8k_math.practice.v1.propose_runner import build_ratification_queue
from evals.gsm8k_math.runner import CaseOutcome
def _cases(n: int) -> list[dict[str, Any]]:
# answer_expression carries a single '+' calc → one gold operation class.
return [
{
"case_id": f"c{i}",
"question": "irrelevant",
"answer_expression": "1+1 = <<1+1=2>>2",
"answer_numeric": 2,
}
for i in range(n)
]
def _scorer(outcomes: list[str]):
it = iter(outcomes)
def score(adapted: dict[str, Any]) -> CaseOutcome:
verdict = next(it)
gold = float(adapted["expected_answer"])
return CaseOutcome(
case_id=adapted["id"],
outcome=verdict,
reason="mock",
expected_answer=gold,
expected_unit="",
actual_answer=gold if verdict == "correct" else (gold + 1 if verdict == "wrong" else None),
actual_unit=None,
trace_hash=None,
realized_prose=None,
)
return score
def test_loop_proposes_a_reliable_class() -> None:
# 45 correct → the Wilson floor on 45/45 (≈0.87) clears θ=0.85. The floor is
# deliberately conservative at small N (30/30 ≈ 0.82 does NOT clear) — that
# strictness, demanding evidence before promotion, is the point.
q = build_ratification_queue(cases=_cases(45), scorer=_scorer(["correct"] * 45))
assert q["proposal_count"] == 1
p = q["proposals"][0]
assert p["correct"] == 45 and p["wrong"] == 0 and p["committed"] == 45
assert p["measured"] >= p["required"] >= 0.85
assert p["action"] == "propose"
def test_loop_filters_unreliable_class() -> None:
# 6 correct + 6 wrong → reliability far below 0.85 → no proposal.
q = build_ratification_queue(
cases=_cases(12), scorer=_scorer(["correct", "wrong"] * 6)
)
assert q["proposal_count"] == 0
assert q["practice_counts"]["wrong"] == 6
def test_loop_respects_n_min_floor() -> None:
# 9 correct (< N_MIN committed) → reliability 0 → no proposal.
q = build_ratification_queue(cases=_cases(9), scorer=_scorer(["correct"] * 9))
assert q["proposal_count"] == 0
def test_queue_is_deterministic_and_proposal_only() -> None:
a = build_ratification_queue(cases=_cases(45), scorer=_scorer(["correct"] * 45))
b = build_ratification_queue(cases=_cases(45), scorer=_scorer(["correct"] * 45))
assert a == b
assert a["regime"] == "propose"
assert "never a serving mutation" in a["note"]