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

Merge pull request #432 from AssetOverflow/feat/adr-0175-phase1-reliability-gate

Browse source 
ADR-0175 Phase 1: reliability ledger + attempt/refuse gate substrate
This commit is contained in:
Shay 2026-05-28 15:35:40 -07:00 committed by GitHub
commit 1a73d4bab9
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
6 changed files with 608 additions and 0 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

30
core/reliability_gate/__init__.py Normal file
View file

@ -0,0 +1,30 @@
"""ADR-0175 Phase 1 — reliability ledger + attempt/refuse gate substrate.
Standalone, deterministic, replay-stable. NOT wired into the serving/eval path
(invariant #1: zero serving change). NOT the `calibration/` module (that is a
grid-search hyperparameter tuner; this is the per-class reliability ledger).
Public surface:
- :func:`conservative_floor`, :data:`WILSON_Z`, :data:`N_MIN` — the pinned floor.
- :class:`ClassTally` — per-class counted ledger; reliability = commitment precision.
- :class:`Action`, :class:`Ceilings` — human-set θ ceilings (engine never mutates).
- :func:`license_for`, :class:`LicenseDecision` — the deterministic gate.
"""
from __future__ import annotations
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
__all__ = [
"Action",
"Ceilings",
"ClassTally",
"LicenseDecision",
"N_MIN",
"WILSON_Z",
"conservative_floor",
"license_for",
]

66
core/reliability_gate/ceilings.py Normal file
View file

@ -0,0 +1,66 @@
"""ADR-0175 §3 — human-set required-reliability ceilings (θ).
``θ`` is the per-class, per-action reliability an action *demands* before it is
licensed. These are the **human autonomy dial**, distinct from the global
estimator skepticism :data:`core.reliability_gate.floor.WILSON_Z`.
Invariant #4 — the engine never raises its own ceiling. Enforced structurally:
:class:`Ceilings` is frozen (no in-place mutation), and "raising a ceiling" via
:meth:`with_override` returns a NEW instance — an explicit reconstruction that
models a human editing config, never engine self-authorization.
"""
from __future__ import annotations
from dataclasses import dataclass
from enum import Enum
from typing import Final, Mapping
class Action(Enum):
"""The blast-radius of what a passed gate licenses (ADR-0175 §3)."""
PRACTICE = "practice" # sealed; θ = 0 -> always attempt
PROPOSE = "propose" # emit a ratifiable proposal
SERVE = "serve" # touch a served answer
# Global default ceilings. PRACTICE = 0.0 (sealed practice always attempts).
_DEFAULTS: Final[Mapping[Action, float]] = {
Action.PRACTICE: 0.0,
Action.PROPOSE: 0.85,
Action.SERVE: 0.99,
}
@dataclass(frozen=True, slots=True)
class Ceilings:
"""Immutable θ table: global defaults + explicit per-(class, action) overrides."""
overrides: tuple[tuple[str, Action, float], ...] = ()
def required(self, class_name: str, action: Action) -> float:
"""θ for this class+action — an override if present, else the global default."""
for cls, act, theta in self.overrides:
if cls == class_name and act == action:
return theta
return _DEFAULTS[action]
def with_override(self, class_name: str, action: Action, theta: float) -> "Ceilings":
"""Return a NEW Ceilings with this ceiling set (immutable; not in-place).
θ must be in ``[0.0, 1.0)`` — a ceiling of 1.0 is unreachable by the
floor (no finite record proves perfection), so it is rejected as a
configuration error rather than silently making an action impossible.
"""
if not (0.0 <= theta < 1.0):
raise ValueError("θ must be in [0.0, 1.0)")
kept = tuple(
o for o in self.overrides if not (o[0] == class_name and o[1] == action)
)
return Ceilings(kept + ((class_name, action, theta),))
@classmethod
def default(cls) -> "Ceilings":
"""The global-default ceilings with no per-class overrides."""
return cls()

71
core/reliability_gate/floor.py Normal file
View file

@ -0,0 +1,71 @@
"""ADR-0175 §4a — pinned conservative reliability floor (one-sided Wilson lower bound).
A deterministic, replay-stable lower bound on a success proportion given integer
counts. No learned weights, no stochastic sampling — the only "statistics" here is
a fixed closed-form bound over counts.
Two independent dials (do not conflate):
- ``WILSON_Z`` — how skeptical the *estimator* is. Pinned, global. The single
caution knob. The engine never touches it.
- per-class ``θ`` ceilings (see :mod:`core.reliability_gate.ceilings`) — how much
reliability an *action* demands. Human-set, per class. Also untouchable by the
engine (invariant #4).
"""
from __future__ import annotations
import math
from typing import Final
# Single global pessimism constant (~99% one-sided). ADR-0175 §4a.
WILSON_Z: Final[float] = 2.576
# Minimum committed trials before any reliability is claimed.
N_MIN: Final[int] = 10
# Replay rounding: half-to-even to this many decimals (determinism contract).
_ROUND_DECIMALS: Final[int] = 9
# Largest value strictly below 1.0 at the pinned precision — honours the [0,1)
# range invariant unconditionally (the Wilson lower bound is < 1 for all finite
# committed; this only ever binds at absurd counts, e.g. committed > ~6.6e9).
_MAX_BELOW_ONE: Final[float] = 1.0 - 10.0 ** (-_ROUND_DECIMALS)
def conservative_floor(successes: int, committed: int) -> float:
"""Lower bound on the success proportion of ``successes`` out of ``committed``.
Pinned one-sided Wilson lower bound (ADR-0175 §4a). Returns a value in
``[0.0, 1.0)`` — never exactly ``1.0`` (no finite record proves perfection).
Below :data:`N_MIN` committed trials returns ``0.0`` (insufficient evidence).
Deterministic and replay-stable: IEEE-754 float64 with the result rounded
half-to-even to ``1e-9`` so the downstream gate comparison is byte-stable
across backends.
For a perfect record (``successes == committed``) the bound reduces to
``committed / (committed + z²)`` — reliability is *earned by volume*, never
granted by a lucky streak.
"""
if not isinstance(successes, int) or not isinstance(committed, int):
raise TypeError("successes and committed must be int counts")
if committed < 0 or successes < 0:
raise ValueError("counts must be non-negative")
if successes > committed:
raise ValueError(
f"successes ({successes}) cannot exceed committed ({committed})"
)
if committed < N_MIN:
return 0.0
p = successes / committed
z2 = WILSON_Z * WILSON_Z
denom = 1.0 + z2 / committed
center = (p + z2 / (2.0 * committed)) / denom
margin = (WILSON_Z / denom) * math.sqrt(
p * (1.0 - p) / committed + z2 / (4.0 * committed * committed)
)
lower = center - margin
if lower < 0.0:
lower = 0.0
rounded = round(lower, _ROUND_DECIMALS)
# Honour [0.0, 1.0) as a hard invariant regardless of rounding.
return rounded if rounded < 1.0 else _MAX_BELOW_ONE

72
core/reliability_gate/gate.py Normal file
View file

@ -0,0 +1,72 @@
"""ADR-0175 §3 — the deterministic attempt/refuse gate.
``license_for`` decides whether an action is permitted for a class:
measured_reliability(class) / θ_required(action, class) ≥ 1
Equivalently ``measured ≥ required`` (and ``required == 0`` -> always licensed),
which avoids a divide-by-zero on the sealed-practice ceiling. The ratio is kept
as an inspectable field. Pure and deterministic: never mutates or emits the
``Ceilings`` it is given (invariant #4), identical inputs -> identical decision
(invariant #3).
"""
from __future__ import annotations
from dataclasses import dataclass
from core.reliability_gate.ceilings import Action, Ceilings
from core.reliability_gate.ledger import ClassTally
# Which earned number gates an action.
_CHECKERS: frozenset[str] = frozenset({"reliability", "t2_precision"})
@dataclass(frozen=True, slots=True)
class LicenseDecision:
"""Inspectable result of the gate. Carries the numbers behind the verdict."""
class_name: str
action: Action
checker: str
measured: float
required: float
ratio: float # measured / required; +inf when required == 0
licensed: bool
def license_for(
tally: ClassTally,
action: Action,
ceilings: Ceilings,
*,
checker: str = "reliability",
) -> LicenseDecision:
"""Decide whether ``action`` is licensed for ``tally``'s class.
``checker`` selects the earned number that gates this action:
``"reliability"`` (commitment precision) or ``"t2_precision"`` (Tier-2
self-verification trust, used for widening past gold).
"""
if checker not in _CHECKERS:
raise ValueError(f"unknown checker {checker!r}; expected one of {sorted(_CHECKERS)}")
measured = tally.reliability if checker == "reliability" else tally.t2_precision
required = ceilings.required(tally.class_name, action)
if required <= 0.0:
licensed = True
ratio = float("inf")
else:
# measured / required ≥ 1 ⟺ measured ≥ required (both pre-rounded to 1e-9)
licensed = round(measured, 9) >= round(required, 9)
ratio = round(measured / required, 9)
return LicenseDecision(
class_name=tally.class_name,
action=action,
checker=checker,
measured=measured,
required=required,
ratio=ratio,
licensed=licensed,
)

91
core/reliability_gate/ledger.py Normal file
View file

@ -0,0 +1,91 @@
"""ADR-0175 §4 — the per-class calibration ledger.
A replayable tally of *counted* outcomes per class (= capability axis: G1..G5,
S1, ...). Nothing learned, nothing stochastic — every figure is an integer count.
Reliability is **commitment precision** (``correct / (correct + wrong)`` via the
pinned :func:`conservative_floor`): refusals are excluded from the denominator on
purpose. Refusing is always safe, so a high refusal rate is a *coverage* fact
(``coverage``), never a *reliability* penalty.
"""
from __future__ import annotations
from dataclasses import dataclass
from core.reliability_gate.floor import conservative_floor
@dataclass(frozen=True, slots=True)
class ClassTally:
"""Immutable per-class outcome counts.
``class_name`` is a capability-axis id. All mutation is via :meth:`record`,
which returns a new tally (immutability rule).
"""
class_name: str
correct: int = 0
wrong: int = 0
refused: int = 0
t2_verified: int = 0
t2_agrees_gold: int = 0
def __post_init__(self) -> None:
for value in (
self.correct,
self.wrong,
self.refused,
self.t2_verified,
self.t2_agrees_gold,
):
if not isinstance(value, int) or value < 0:
raise ValueError("tally counts must be non-negative ints")
if self.t2_agrees_gold > self.t2_verified:
raise ValueError(
f"t2_agrees_gold ({self.t2_agrees_gold}) cannot exceed "
f"t2_verified ({self.t2_verified})"
)
@property
def committed(self) -> int:
"""Attempts where the engine committed to an answer (excludes refusals)."""
return self.correct + self.wrong
@property
def attempted(self) -> int:
return self.correct + self.wrong + self.refused
@property
def reliability(self) -> float:
"""Conservative lower bound on commitment precision (ADR-0175 §4/§4a)."""
return conservative_floor(self.correct, self.committed)
@property
def t2_precision(self) -> float:
"""How trustworthy Tier-2 self-verification is on this class (vs gold)."""
return conservative_floor(self.t2_agrees_gold, self.t2_verified)
@property
def coverage(self) -> float:
"""Commit rate = committed / attempted. A coverage fact, not reliability."""
return round(self.committed / self.attempted, 9) if self.attempted else 0.0
def record(
self,
*,
correct: int = 0,
wrong: int = 0,
refused: int = 0,
t2_verified: int = 0,
t2_agrees_gold: int = 0,
) -> "ClassTally":
"""Return a new tally with the given outcomes added (immutable update)."""
return ClassTally(
class_name=self.class_name,
correct=self.correct + correct,
wrong=self.wrong + wrong,
refused=self.refused + refused,
t2_verified=self.t2_verified + t2_verified,
t2_agrees_gold=self.t2_agrees_gold + t2_agrees_gold,
)

278
tests/test_adr_0175_phase1_reliability_gate.py Normal file
View file

@ -0,0 +1,278 @@
"""ADR-0175 Phase 1 — ledger + gate substrate.
Proves the pinned `conservative_floor` (§4a), the per-class `ClassTally`
ledger (§4), the human-set `Ceilings` (§3), and the deterministic `license`
gate (§3). Each ADR-0175 invariant is exercised by a test that *fails* under
the violation it names (CLAUDE.md §Schema-Defined Proof Obligations):
- #3 determinism/replay -> TestDeterminismInvariant
- #4 no self-authorization -> TestNoSelfAuthorizationInvariant
This substrate is standalone — nothing in the serving/eval path imports it
(invariant #1, zero serving change, is satisfied by non-wiring; asserted in
TestZeroServingCoupling).
"""
from __future__ import annotations
import dataclasses
import pytest
from core.reliability_gate import (
Action,
Ceilings,
ClassTally,
LicenseDecision,
conservative_floor,
license_for,
N_MIN,
WILSON_Z,
)
# ---------------------------------------------------------------------------
# conservative_floor (§4a)
# ---------------------------------------------------------------------------
class TestConservativeFloor:
def test_below_n_min_is_zero(self) -> None:
for k in range(0, N_MIN):
assert conservative_floor(k, k) == 0.0
# exactly N_MIN with a perfect record is the first non-zero
assert conservative_floor(N_MIN, N_MIN) > 0.0
def test_zero_committed_is_zero(self) -> None:
assert conservative_floor(0, 0) == 0.0
def test_perfect_record_matches_closed_form(self) -> None:
# For a perfect record (s == k) the Wilson lower bound reduces to
# k / (k + z²). Pin exact values via the closed form (no hand-rounding).
z2 = WILSON_Z * WILSON_Z
for k in (10, 38, 60, 100, 657):
expected = round(k / (k + z2), 9)
assert conservative_floor(k, k) == expected
def test_range_is_zero_to_below_one(self) -> None:
for s, k in [(10, 10), (38, 40), (100, 100), (5, 20), (657, 657)]:
v = conservative_floor(s, k)
assert 0.0 <= v < 1.0 # never exactly 1.0 — no finite record proves perfection
def test_perfect_record_is_monotonic_in_k(self) -> None:
# more clean evidence -> higher earned floor
ks = [10, 20, 40, 80, 160, 657]
vals = [conservative_floor(k, k) for k in ks]
assert vals == sorted(vals)
assert all(a < b for a, b in zip(vals, vals[1:]))
def test_cost_to_clear_propose_ceiling(self) -> None:
# ADR worked example: ~38 clean commitments to clear θ_propose = 0.85
assert conservative_floor(37, 37) < 0.85
assert conservative_floor(38, 38) >= 0.85
def test_one_wrong_drops_below_perfect_and_below_propose(self) -> None:
# ADR asymmetry example: one wrong in 40 -> ~0.818, below a 0.85 gate
perfect = conservative_floor(40, 40)
one_wrong = conservative_floor(39, 40)
assert one_wrong < perfect
assert abs(one_wrong - 0.8177) < 1e-3
assert one_wrong < 0.85
def test_serving_is_expensive(self) -> None:
# θ_serve = 0.99 needs hundreds of clean commitments; 100 is not enough
assert conservative_floor(100, 100) < 0.99
assert conservative_floor(657, 657) >= 0.99
def test_rejects_invalid_counts(self) -> None:
with pytest.raises(ValueError):
conservative_floor(5, 4) # successes > committed
with pytest.raises(ValueError):
conservative_floor(-1, 10)
with pytest.raises(ValueError):
conservative_floor(3, -1)
# ---------------------------------------------------------------------------
# ClassTally ledger (§4) — reliability is COMMITMENT precision
# ---------------------------------------------------------------------------
class TestClassTally:
def test_reliability_uses_committed_not_total(self) -> None:
# 40 clean commitments + a pile of refusals: refusals must NOT lower
# reliability (refusing is safe; high refusal is a coverage fact).
no_refusals = ClassTally("G1", correct=40, wrong=0, refused=0)
many_refusals = ClassTally("G1", correct=40, wrong=0, refused=500)
assert no_refusals.reliability == many_refusals.reliability
assert no_refusals.reliability >= 0.85
def test_refusal_only_class_has_zero_reliability(self) -> None:
# No commitments -> no demonstrated reliability -> 0 (cannot serve/propose)
t = ClassTally("G2", correct=0, wrong=0, refused=50)
assert t.committed == 0
assert t.reliability == 0.0
def test_t2_precision_over_anchor_set(self) -> None:
t = ClassTally("G1", t2_verified=40, t2_agrees_gold=40)
assert t.t2_precision >= 0.85
def test_coverage_tracks_commit_rate(self) -> None:
t = ClassTally("G1", correct=8, wrong=2, refused=90)
assert t.coverage == round(10 / 100, 9)
def test_record_is_immutable_returns_new(self) -> None:
t0 = ClassTally("G1")
t1 = t0.record(correct=1)
assert t0.correct == 0 # original untouched
assert t1.correct == 1
assert t1 is not t0
def test_tally_is_frozen(self) -> None:
t = ClassTally("G1", correct=1)
with pytest.raises(dataclasses.FrozenInstanceError):
t.correct = 99 # type: ignore[misc]
def test_rejects_inconsistent_counts(self) -> None:
with pytest.raises(ValueError):
ClassTally("G1", t2_verified=3, t2_agrees_gold=5)
with pytest.raises(ValueError):
ClassTally("G1", correct=-1)
# ---------------------------------------------------------------------------
# Ceilings (§3) + invariant #4 (no self-authorization)
# ---------------------------------------------------------------------------
class TestCeilings:
def test_practice_ceiling_is_zero(self) -> None:
c = Ceilings.default()
assert c.required("G1", Action.PRACTICE) == 0.0
def test_default_propose_and_serve(self) -> None:
c = Ceilings.default()
assert c.required("G1", Action.PROPOSE) == 0.85
assert c.required("G1", Action.SERVE) == 0.99
def test_override_is_per_class(self) -> None:
c = Ceilings.default().with_override("G3", Action.SERVE, 0.95)
assert c.required("G3", Action.SERVE) == 0.95
assert c.required("G1", Action.SERVE) == 0.99 # other classes unchanged
def test_override_rejects_out_of_range(self) -> None:
with pytest.raises(ValueError):
Ceilings.default().with_override("G1", Action.SERVE, 1.0)
with pytest.raises(ValueError):
Ceilings.default().with_override("G1", Action.SERVE, -0.1)
class TestNoSelfAuthorizationInvariant:
"""Invariant #4 — the engine never raises its own ceiling."""
def test_ceilings_are_frozen(self) -> None:
c = Ceilings.default()
with pytest.raises(dataclasses.FrozenInstanceError):
c.overrides = () # type: ignore[misc]
def test_raising_a_ceiling_produces_a_new_object_not_mutation(self) -> None:
c0 = Ceilings.default()
c1 = c0.with_override("G1", Action.SERVE, 0.90)
assert c1 is not c0
# the original is unchanged — there is no in-place "lower the bar"
assert c0.required("G1", Action.SERVE) == 0.99
assert c1.required("G1", Action.SERVE) == 0.90
def test_gate_never_emits_or_mutates_ceilings(self) -> None:
# license_for() returns a LicenseDecision, never a Ceilings, and does not
# mutate the ceilings it was given.
c = Ceilings.default()
before = c.overrides
d = license_for(ClassTally("G1", correct=40), Action.PROPOSE, c)
assert isinstance(d, LicenseDecision)
assert not isinstance(d, Ceilings)
assert c.overrides == before
# ---------------------------------------------------------------------------
# license gate (§3)
# ---------------------------------------------------------------------------
class TestLicenseGate:
def test_practice_always_licensed_even_with_no_evidence(self) -> None:
# θ_practice = 0 -> sealed practice may always attempt
d = license_for(ClassTally("G9", correct=0, wrong=0, refused=0), Action.PRACTICE, Ceilings.default())
assert d.licensed is True
assert d.required == 0.0
def test_propose_licensed_at_38_clean_commitments(self) -> None:
d = license_for(ClassTally("G1", correct=38), Action.PROPOSE, Ceilings.default())
assert d.licensed is True
assert d.measured >= 0.85
def test_propose_denied_with_one_wrong_in_40(self) -> None:
d = license_for(ClassTally("G1", correct=39, wrong=1), Action.PROPOSE, Ceilings.default())
assert d.licensed is False
assert d.measured < 0.85
def test_perfect_100_proposes_but_does_not_serve(self) -> None:
t = ClassTally("G1", correct=100)
assert license_for(t, Action.PROPOSE, Ceilings.default()).licensed is True
assert license_for(t, Action.SERVE, Ceilings.default()).licensed is False
def test_t2_precision_checker_gates_widening(self) -> None:
t = ClassTally("G1", correct=5, wrong=0, t2_verified=40, t2_agrees_gold=40)
d = license_for(t, Action.PROPOSE, Ceilings.default(), checker="t2_precision")
assert d.checker == "t2_precision"
assert d.licensed is True
def test_decision_carries_inspectable_ratio(self) -> None:
d = license_for(ClassTally("G1", correct=100), Action.PROPOSE, Ceilings.default())
assert d.ratio == round(d.measured / d.required, 9)
assert d.ratio >= 1.0
def test_unknown_checker_rejected(self) -> None:
with pytest.raises(ValueError):
license_for(ClassTally("G1", correct=40), Action.PROPOSE, Ceilings.default(), checker="vibes")
# ---------------------------------------------------------------------------
# Invariant #3 — determinism / replay
# ---------------------------------------------------------------------------
class TestDeterminismInvariant:
def test_floor_is_idempotent_and_pre_rounded(self) -> None:
for s, k in [(10, 10), (38, 40), (5, 10), (657, 657), (39, 40)]:
a = conservative_floor(s, k)
b = conservative_floor(s, k)
assert a == b # pure
assert a == round(a, 9) # already at 1e-9 (replay-stable)
def test_gate_decision_is_deterministic(self) -> None:
t = ClassTally("G1", correct=38, wrong=1, refused=12)
c = Ceilings.default()
d1 = license_for(t, Action.PROPOSE, c)
d2 = license_for(t, Action.PROPOSE, c)
assert d1 == d2
def test_reliability_stable_across_recompute(self) -> None:
t = ClassTally("G1", correct=38, wrong=2)
assert t.reliability == ClassTally("G1", correct=38, wrong=2).reliability
# ---------------------------------------------------------------------------
# Invariant #1 (proxy) — zero serving coupling
# ---------------------------------------------------------------------------
class TestZeroServingCoupling:
def test_package_does_not_import_serving_runtime(self) -> None:
# The substrate must not pull in the parse/solve/eval serving path.
import core.reliability_gate as rg
mod_file = rg.__file__
assert mod_file is not None
# none of the serving-path modules should be a (transitive) hard import
# of the gate package's own modules — assert the package modules don't
# name them at import time.
import importlib
for name in ("floor", "ledger", "ceilings", "gate"):
m = importlib.import_module(f"core.reliability_gate.{name}")
src = (m.__doc__ or "")
# structural: these modules import only stdlib + sibling gate modules
assert "generate.math_candidate_graph" not in src