Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
90 changes: 89 additions & 1 deletion scripts/check_adjoint_prevalidation.py
Original file line number Diff line number Diff line change
Expand Up @@ -50,6 +50,13 @@
later deleted or reverted; this gate therefore certifies the current
surfaces, not erased history.

The `b0:` claim is re-verified the same way: when a dossier claims READY
with `b0: clear`, the gate runs the B.0 organ (`check_unaddressed_nits.py`)
and refuses the dossier if the organ still finds an unlifted remark, naming
each one. A green gate therefore no longer hides a red B.0. It still does not
dispense with reading the surfaces: the organ only sees its markers, and who
lifted a remark, when, and on what substance are read by hand (CLAUDE.md §B.0).

Exit codes -- dossier INTEGRITY and PR MERGEABILITY are two questions, and
conflating them is what this gate used to do (#16800):

Expand Down Expand Up @@ -565,6 +572,81 @@ class DWELL/FAIL lives there, as information for the reader, not in the
return contradictions


def b0_claim_contradictions(claim: str, result: dict[str, Any] | None) -> list[str]:
"""Re-verify a dossier's ``b0: clear`` claim against the live B.0 organ.

The ``checks:`` claim has been re-verified since #16957; ``b0:`` was
still taken on faith, and a READY dossier could attest ``b0: clear`` on a
pull request the organ blocks. Measured on 2026-09-24: two READY dossiers
(#16955 and #16987) declared ``b0: clear`` while ``check_unaddressed_nits.py``
exited 1 on an unlifted Hermes reserve. Only the coordinator's separate B.0
run caught them, and the gate's ``exit 0`` looked like a green light. Like
the checks claim, the b0 claim is now compared with what the organ measures
at evaluation time, and every unlifted remark is named.

``result`` is the dict returned by ``check_unaddressed_nits.analyse_pr``.
A claim other than ``clear`` is not refuted here, because a BLOCKED dossier
may say so.
"""
if claim != "clear" or not result or not result.get("blocked"):
return []
blocking = list(result.get("blocking") or [])
named = "; ".join(
f"{row.get('kind', '?')} by {row.get('author', '?')} via {row.get('src', '?')}"
for row in blocking[:5]
)
if len(blocking) > 5:
named += f" (+{len(blocking) - 5} more)"
return [
"b0 claim 'clear' is contradicted by the live B.0 organ "
f"(check_unaddressed_nits.py): {len(blocking)} unlifted remark(s)"
+ (f" -- {named}" if named else "")
]


def probe_b0(pr: int) -> dict[str, Any]:
"""Run the B.0 organ on ``pr``. A failure to measure is fail-closed.

The import is lazy because the probe runs only for a dossier that claims
READY: BLOCKED and absent dossiers never load the organ. A failure to
import it is a failure to measure like any other -- it surfaces as
``RuntimeError``, which ``main`` reports as UNKNOWN (exit 2), never as a
traceback.
"""
try:
try:
import check_unaddressed_nits
except ImportError: # charge via importlib dans les tests (hors scripts/)
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
import check_unaddressed_nits
return check_unaddressed_nits.analyse_pr(pr)
except Exception as exc: # noqa: BLE001 -- any failure means "not measured"
raise RuntimeError(f"B.0 organ could not measure PR #{pr}: {exc}") from exc


def refute_ready_b0(
pr: int,
verdict: str,
dossier: Dossier | None,
probe: Any = None,
) -> tuple[str, list[str], Dossier | None]:
"""Demote a READY verdict whose ``b0: clear`` claim the organ refutes.

The demotion is to "no dossier worth trusting" (exit 1), the same outcome
as a contradicted ``checks:`` claim: a dossier that attests a false
``clear`` cannot be trusted on its other fields either. Only READY is
probed, so the organ adds no API cost to BLOCKED or absent dossiers.
"""
if verdict != VERDICT_READY or dossier is None:
return verdict, [], dossier
refuted = b0_claim_contradictions(
dossier.fields.get("b0", ""), (probe or probe_b0)(pr)
)
if refuted:
return "", refuted, None
return verdict, [], dossier


def carrying_lane(snapshot: dict[str, Any]) -> str | None:
"""Return the lane that carries this pull request, from its `Grain:` tag.

Expand Down Expand Up @@ -999,7 +1081,11 @@ def load_snapshot(pr: int) -> dict[str, Any]:
# Fetched inside the before/after bracket: a check concluding during the
# read bumps updatedAt and aborts the snapshot (transient UNKNOWN, the
# caller retries), so the claim verification below never reads a state
# that was already stale when captured.
# that was already stale when captured. The B.0 probe (`probe_b0`) is NOT
# in this bracket: it runs after, and only on a READY dossier. A remark
# posted between the snapshot and the probe therefore makes the organ
# contradict a `b0: clear` claim -- a conservative refusal, which a rerun
# names as a changed discussion surface.
snapshot["checkRuns"] = _head_check_runs(snapshot["headRefOid"])
after = _pr_metadata(pr, with_rollup=True)
if _metadata_identity(before) != _metadata_identity(after):
Expand Down Expand Up @@ -1098,6 +1184,8 @@ def main() -> int:
)
return 0
verdict, errors, dossier = evaluate_with_dossier(snapshot)
if not errors:
verdict, errors, dossier = refute_ready_b0(args.pr, verdict, dossier)
except (
RuntimeError,
KeyError,
Expand Down
5 changes: 3 additions & 2 deletions scripts/coordination/merge_ready.py
Original file line number Diff line number Diff line change
Expand Up @@ -36,8 +36,9 @@
2 unknown, 3 blocked) sont des SKIPS nommes, pas des erreurs ;
4. champ ``b0:`` du dossier ACCEPTE par le gate egale ``clear`` --
grammaire du dossier relue via ``parse_dossier`` du gate lui-meme
(import, pas de duplication) ; le gate ne re-verifie pas b0, c'est
l'etape 5 qui le fait ;
(import, pas de duplication) ; depuis que le gate re-verifie une
claim ``b0: clear`` contre l'organe B.0, l'etape 5 fait double emploi
pour un READY : elle reste le filet si le gate change ;
5. organe B.0 ``check_unaddressed_nits.py <PR>`` exit 0 -- code de
retour capture DIRECTEMENT (subprocess.returncode, jamais a travers
un pipe) ;
Expand Down
125 changes: 125 additions & 0 deletions scripts/tests/test_check_adjoint_prevalidation.py
Original file line number Diff line number Diff line change
Expand Up @@ -1165,3 +1165,128 @@ def test_fingerprint_refusal_names_the_live_surface_landscape():
assert "threads=2 (1 non resolus)" in msg
assert "checks=1" in msg
assert "reviews=2" in msg


# --- b0 claim re-verified against the live B.0 organ -------------------------
# Measured 2026-09-24: READY dossiers on #16955 and #16987 declared `b0: clear`
# while check_unaddressed_nits.py exited 1. The gate answered exit 0 on both.


def _ready_dossier(**changes: str):
snapshot = _snapshot(_body(**changes))
verdict, errors, dossier = mod.evaluate_with_dossier(snapshot)
assert verdict == mod.VERDICT_READY, errors
return verdict, dossier


def _organ(blocked: bool, blocking: list | None = None):
calls = []

def probe(pr):
calls.append(pr)
return {"blocked": blocked, "blocking": blocking or []}

return probe, calls


def test_b0_clear_refuted_by_organ_demotes_ready_and_names_the_remark():
verdict, dossier = _ready_dossier()
probe, calls = _organ(
True,
[{"kind": "concern", "author": "jsboige", "src": "review 2026-09-22"}],
)
verdict, errors, dossier = mod.refute_ready_b0(123, verdict, dossier, probe)
assert calls == [123]
assert verdict == "" and dossier is None
assert len(errors) == 1
assert "b0 claim 'clear' is contradicted" in errors[0]
assert "concern by jsboige via review 2026-09-22" in errors[0]


def test_b0_clear_confirmed_by_organ_keeps_ready():
verdict, dossier = _ready_dossier()
probe, calls = _organ(False)
out = mod.refute_ready_b0(123, verdict, dossier, probe)
assert calls == [123]
assert out == (mod.VERDICT_READY, [], dossier)


def test_b0_probe_not_paid_for_blocked_or_absent_dossier():
probe, calls = _organ(True, [{"kind": "k", "author": "a", "src": "s"}])
assert mod.refute_ready_b0(123, mod.VERDICT_BLOCKED, None, probe)[0] == mod.VERDICT_BLOCKED
assert mod.refute_ready_b0(123, "", None, probe) == ("", [], None)
assert calls == []


def test_b0_contradictions_ignore_a_non_clear_claim_and_cap_the_list():
rows = [{"kind": f"k{i}", "author": "a", "src": "s"} for i in range(7)]
assert mod.b0_claim_contradictions("blocked", {"blocked": True, "blocking": rows}) == []
assert mod.b0_claim_contradictions("clear", None) == []
[error] = mod.b0_claim_contradictions("clear", {"blocked": True, "blocking": rows})
assert "7 unlifted remark(s)" in error and "(+2 more)" in error


def test_b0_probe_failure_is_fail_closed(monkeypatch):
class Broken:
@staticmethod
def analyse_pr(pr):
raise ValueError("network down")

monkeypatch.setitem(sys.modules, "check_unaddressed_nits", Broken)
with pytest.raises(RuntimeError, match="B.0 organ could not measure PR #123"):
mod.probe_b0(123)



def test_b0_probe_import_failure_is_fail_closed(monkeypatch):
"""An organ that cannot even be imported is 'not measured' (UNKNOWN), not a traceback."""
import builtins

real_import = builtins.__import__

def refuse(name, *args, **kwargs):
if name == "check_unaddressed_nits":
raise ImportError("organ missing")
return real_import(name, *args, **kwargs)

monkeypatch.delitem(sys.modules, "check_unaddressed_nits", raising=False)
monkeypatch.setattr(builtins, "__import__", refuse)
with pytest.raises(RuntimeError, match="B.0 organ could not measure PR #123"):
mod.probe_b0(123)


def test_main_exits_unknown_when_organ_cannot_be_imported(monkeypatch, capsys):
snapshot = _snapshot(_body())
monkeypatch.setattr(mod, "load_snapshot", lambda pr: snapshot)
monkeypatch.setattr(mod.gh_identity, "pin_gh_token", lambda: None)

def unmeasured(pr):
raise RuntimeError(f"B.0 organ could not measure PR #{pr}: organ missing")

monkeypatch.setattr(mod, "probe_b0", unmeasured)
monkeypatch.setattr(sys, "argv", ["check_adjoint_prevalidation.py", "123"])
assert mod.main() == mod.EXIT_UNKNOWN
assert "UNKNOWN" in capsys.readouterr().out

def test_main_exits_no_dossier_when_organ_refutes_b0(monkeypatch, capsys):
snapshot = _snapshot(_body())
monkeypatch.setattr(mod, "load_snapshot", lambda pr: snapshot)
monkeypatch.setattr(mod.gh_identity, "pin_gh_token", lambda: None)
monkeypatch.setattr(
mod,
"probe_b0",
lambda pr: {"blocked": True, "blocking": [{"kind": "nit", "author": "u", "src": "c"}]},
)
monkeypatch.setattr(sys, "argv", ["check_adjoint_prevalidation.py", "123"])
assert mod.main() == mod.EXIT_NO_DOSSIER
out = capsys.readouterr().out
assert "NO-DOSSIER" in out and "b0 claim 'clear' is contradicted" in out


def test_main_exits_ready_when_organ_agrees(monkeypatch, capsys):
snapshot = _snapshot(_body())
monkeypatch.setattr(mod, "load_snapshot", lambda pr: snapshot)
monkeypatch.setattr(mod.gh_identity, "pin_gh_token", lambda: None)
monkeypatch.setattr(mod, "probe_b0", lambda pr: {"blocked": False, "blocking": []})
monkeypatch.setattr(sys, "argv", ["check_adjoint_prevalidation.py", "123"])
assert mod.main() == mod.EXIT_READY
Loading