feat(capability): implement ADR-0106 expert-demo promotion contract (#113)

Closes ADR-0106 acceptance evidence:

- ExpertDemoClaim dataclass + additive expert_demo_claims block on
  ReviewerRegistry (schema_version stays at 1; backward-compatible).
- New core/capability/expert_demo.py with derive_evidence_digest,
  evaluate_expert_demo, collect_domain_lanes, materialise_lane_results.
- core/capability/reporting.py: replaces the cognition-lane-only
  predicate (previous lines 418-433) with a domain-aware,
  reviewer-signed gate; ledger rows now also carry
  expert_demo_reason for operator legibility. Reviewer registry is
  fail-closed: an unloadable registry yields zero claims, so a broken
  registry never silently grants expert_demo=true.
- tests/test_expert_demo_contract.py covers all three ADR-0106
  invariants: requires_signature, domain_aware, replay_byte_equality;
  plus threshold + production-ledger-untouched gates. 12 new tests.
- tests/test_reviewer_registry.py extended with TestExpertDemoClaimsSchema
  covering omitted block, valid parse, unknown signer rejection,
  malformed digest rejection, duplicate domain rejection. 5 new tests.
- README index row + table preface updated to note expert_demo is
  contract-gated. Frontier list trimmed (ADR-0106 has landed).
- ADR-0106 Status flipped Proposed -> Accepted.

No domain row's expert_demo field flips by this PR -- only the contract
changes. Promotion of any ratified domain requires a follow-up ADR
(ADR-0107 reserved for mathematics_logic) plus a signed claim.
This commit is contained in:
Shay 2026-05-22 11:39:09 -07:00 committed by GitHub
parent 7bb8334fc5
commit 0493808215
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
7 changed files with 832 additions and 25 deletions

View file

@ -0,0 +1,243 @@
"""Domain-aware expert-demo promotion gate (ADR-0106).
Replaces the cognition-lane-only predicate previously embedded in
``core.capability.reporting``. A domain ``D`` is promoted to
``expert_demo=true`` iff:
1. ``D`` already passes the ``reasoning_capable`` predicate.
2. A signed ``ExpertDemoClaim`` exists in the reviewer registry for ``D``.
3. The reviewer named in ``claim.signed_by`` may review evals for ``D``
(ADR-0092 ``can_review`` check, scope ``"eval"``).
4. Every lane listed in ``claim.evidence_lanes`` is attached to at least
one of ``D``'s ratified packs (no cross-domain bleed).
5. Every named lane's threshold metrics meet the ADR-0106 §1 bar on both
``public`` and ``holdout``.
6. The canonical evidence-bundle digest reproduces ``claim.claim_digest``
byte-for-byte.
Any failure leaves the ledger row at ``reasoning-capable``.
"""
from __future__ import annotations
import hashlib
import json
from dataclasses import dataclass
from typing import Any, Callable, Mapping, Sequence
from core.capability.reviewers import ReviewerRegistry
_LaneResults = Mapping[str, Mapping[str, Mapping[str, Any]]]
SURFACE_GROUNDEDNESS_MIN: float = 0.95
TERM_CAPTURE_RATE_MIN: float = 0.85
INTENT_ACCURACY_MIN: float = 0.95
VERSOR_CLOSURE_RATE_MIN: float = 1.0
FABRICATION_PASSED_MIN: float = 1.0
FABRICATION_CONTROL_LANE: str = "fabrication_control"
@dataclass(frozen=True, slots=True)
class ExpertDemoVerdict:
passed: bool
reason: str
derived_digest: str | None
def derive_evidence_digest(
domain_id: str,
evidence_revision: str,
evidence_lanes: Sequence[str],
lane_results: _LaneResults,
) -> str:
"""Compute the canonical evidence-bundle SHA-256.
``lane_results`` maps ``lane_id -> split -> result_dict`` where the
result_dict is the raw JSON ``metrics`` block from
``evals/<lane>/results/v1_<split>.json``.
The bundle is deterministic in field order (sorted keys, compact
separators) so re-derivation reproduces the digest byte-for-byte.
"""
sorted_lanes = sorted(evidence_lanes)
bundle = {
"domain_id": domain_id,
"evidence_revision": evidence_revision,
"evidence_lanes": sorted_lanes,
"lane_metrics": {
lane: {
"public": dict(lane_results.get(lane, {}).get("public", {})),
"holdout": dict(lane_results.get(lane, {}).get("holdout", {})),
}
for lane in sorted_lanes
},
}
body = json.dumps(bundle, sort_keys=True, separators=(",", ":")).encode("utf-8")
return hashlib.sha256(body).hexdigest()
def _meets_thresholds(lane_id: str, metrics: Mapping[str, Any]) -> tuple[bool, str]:
if lane_id == FABRICATION_CONTROL_LANE:
passed = float(metrics.get("passed_rate", metrics.get("accuracy", 0)) or 0)
if passed < FABRICATION_PASSED_MIN:
return False, (
f"lane {lane_id!r} passed_rate={passed} below "
f"threshold {FABRICATION_PASSED_MIN}"
)
return True, ""
checks = (
("surface_groundedness", SURFACE_GROUNDEDNESS_MIN),
("term_capture_rate", TERM_CAPTURE_RATE_MIN),
("intent_accuracy", INTENT_ACCURACY_MIN),
("versor_closure_rate", VERSOR_CLOSURE_RATE_MIN),
)
for key, minimum in checks:
if key not in metrics:
return False, (
f"lane {lane_id!r} missing required metric {key!r}"
)
value = float(metrics[key] or 0)
if value < minimum:
return False, (
f"lane {lane_id!r} {key}={value} below threshold {minimum}"
)
return True, ""
def evaluate_expert_demo(
*,
domain_id: str,
reasoning_capable: bool,
registry: ReviewerRegistry,
domain_lanes: Sequence[str],
lane_results: Mapping[str, Mapping[str, Mapping[str, object]]],
) -> ExpertDemoVerdict:
"""Decide whether ``domain_id`` may carry ``expert_demo=true``.
``domain_lanes`` is the union of ``eval_lanes`` declared by the
ratified packs for ``domain_id`` the only lanes legal as evidence
sources for this promotion.
``lane_results`` is the materialised per-split metrics; the caller
is responsible for resolving each lane id to its on-disk result.
"""
if not reasoning_capable:
return ExpertDemoVerdict(
passed=False,
reason="domain not yet reasoning-capable",
derived_digest=None,
)
claim = registry.expert_demo_claim_for(domain_id)
if claim is None:
return ExpertDemoVerdict(
passed=False,
reason="no expert_demo_claims entry for this domain",
derived_digest=None,
)
if not registry.can_review(claim.signed_by, domain_id=domain_id, scope="eval"):
return ExpertDemoVerdict(
passed=False,
reason=(
f"signer {claim.signed_by!r} cannot review eval-scope "
f"artifacts for domain {domain_id!r}"
),
derived_digest=None,
)
domain_lane_set = set(domain_lanes)
cross_domain = [
lane for lane in claim.evidence_lanes if lane not in domain_lane_set
]
if cross_domain:
return ExpertDemoVerdict(
passed=False,
reason=(
f"claim cites lanes not attached to domain {domain_id!r}: "
f"{sorted(cross_domain)}"
),
derived_digest=None,
)
for lane_id in claim.evidence_lanes:
for split in ("public", "holdout"):
metrics = lane_results.get(lane_id, {}).get(split)
if not metrics:
return ExpertDemoVerdict(
passed=False,
reason=(
f"lane {lane_id!r} split {split!r} has no results"
),
derived_digest=None,
)
ok, why = _meets_thresholds(lane_id, metrics)
if not ok:
return ExpertDemoVerdict(
passed=False,
reason=f"{why} (split={split})",
derived_digest=None,
)
derived = derive_evidence_digest(
domain_id=domain_id,
evidence_revision=claim.evidence_revision,
evidence_lanes=claim.evidence_lanes,
lane_results=lane_results,
)
if derived != claim.claim_digest:
return ExpertDemoVerdict(
passed=False,
reason=(
"evidence-bundle digest does not match claim_digest "
"(replay drift)"
),
derived_digest=derived,
)
return ExpertDemoVerdict(
passed=True,
reason="all expert-demo predicates satisfied",
derived_digest=derived,
)
def collect_domain_lanes(
pack_manifests: Sequence[Mapping[str, Any]],
) -> tuple[str, ...]:
"""Return the union of ``eval_lanes[].lane`` across pack manifests."""
lanes: list[str] = []
seen: set[str] = set()
for manifest in pack_manifests:
entries = manifest.get("eval_lanes") or []
if not isinstance(entries, list):
continue
for entry in entries:
if not isinstance(entry, Mapping):
continue
lane_id = entry.get("lane")
if isinstance(lane_id, str) and lane_id not in seen:
seen.add(lane_id)
lanes.append(lane_id)
return tuple(lanes)
def materialise_lane_results(
lane_ids: Sequence[str],
*,
fetch_split: Callable[[str, str], Mapping[str, Any]],
) -> dict[str, dict[str, Mapping[str, Any]]]:
"""Materialise ``lane -> split -> metrics`` for the named lanes.
``fetch_split(lane_id, split)`` returns the parsed ``metrics``
sub-dict from the latest result file (or ``{}`` if absent).
"""
out: dict[str, dict[str, Mapping[str, object]]] = {}
for lane_id in lane_ids:
out[lane_id] = {
"public": dict(fetch_split(lane_id, "public") or {}),
"holdout": dict(fetch_split(lane_id, "holdout") or {}),
}
return out

View file

@ -17,7 +17,13 @@ from core.capability.domains import (
DOMAIN_OPERATOR_CLAIMS,
DOMAIN_PACKS,
)
from core.capability.expert_demo import (
collect_domain_lanes,
evaluate_expert_demo,
materialise_lane_results,
)
from core.capability.reviewers import (
ReviewerRegistry,
ReviewerRegistryError,
load_reviewer_registry,
)
@ -370,6 +376,7 @@ def ledger_report() -> dict[str, Any]:
missing_sources = [k for k, p in resolved.items() if not p.exists()]
gaps = _gap_registry()
chain_inventory = _chain_inventory()
_resolved_registry = _load_registry_for_expert_demo()
domain_rows: list[dict[str, Any]] = []
for domain, packs in DOMAIN_PACKS.items():
pack_metrics = [_pack_metrics(pack_id) for pack_id in packs]
@ -415,22 +422,23 @@ def ledger_report() -> dict[str, Any]:
and intent_shapes_present >= 3
and holdout_present
)
expert_demo = False
if reasoning_capable:
public = _latest_eval_result("cognition", "v1", "public").get("metrics", {})
holdout = _latest_eval_result("cognition", "v1", "holdout").get("metrics", {})
expert_demo = bool(
public
and holdout
and public.get("surface_groundedness", 0) >= 0.95
and holdout.get("surface_groundedness", 0) >= 0.95
and public.get("term_capture_rate", 0) >= 0.85
and holdout.get("term_capture_rate", 0) >= 0.85
and public.get("intent_accuracy", 0) >= 0.95
and holdout.get("intent_accuracy", 0) >= 0.95
and public.get("versor_closure_rate", 0) >= 1.0
and holdout.get("versor_closure_rate", 0) >= 1.0
)
manifests = [_manifest_for_pack(pack_id) for pack_id in packs]
domain_lanes = collect_domain_lanes(manifests)
lane_results = materialise_lane_results(
domain_lanes,
fetch_split=lambda lane, split: _latest_eval_result(
lane, "v1", split
).get("metrics", {}),
)
verdict = evaluate_expert_demo(
domain_id=domain,
reasoning_capable=reasoning_capable,
registry=_resolved_registry,
domain_lanes=domain_lanes,
lane_results=lane_results,
)
expert_demo = verdict.passed
expert_demo_reason = verdict.reason
if expert_demo:
status = "expert-demo"
elif reasoning_capable:
@ -474,6 +482,7 @@ def ledger_report() -> dict[str, Any]:
"reasoning_capable": reasoning_capable,
"expert_demo": expert_demo,
},
"expert_demo_reason": expert_demo_reason,
"known_gaps": domain_gaps,
"open_gaps": open_gaps,
"blocked_lift": blocked_lift,
@ -591,6 +600,23 @@ def evidence_plan_report() -> dict[str, Any]:
}
def _load_registry_for_expert_demo() -> ReviewerRegistry:
"""Load reviewer registry, returning an empty one on any failure.
Expert-demo gating is fail-closed: an unloadable registry yields a
registry with zero reviewers and zero claims, which makes
``evaluate_expert_demo`` refuse every domain. A broken registry
must never silently grant ``expert_demo=true``.
"""
path = _REPO_ROOT / LEDGER_SOURCES.reviewers
if not path.exists():
return ReviewerRegistry(schema_version=1, reviewers=(), expert_demo_claims=())
try:
return load_reviewer_registry(path)
except ReviewerRegistryError:
return ReviewerRegistry(schema_version=1, reviewers=(), expert_demo_claims=())
def _reviewer_registry_status() -> dict[str, Any]:
"""Report Reviewer Registry v1 validity for the capability ledger (ADR-0092).

View file

@ -17,10 +17,21 @@ from typing import Any, Mapping
import yaml
ALLOWED_TOP_LEVEL_KEYS: frozenset[str] = frozenset({"schema_version", "reviewers"})
ALLOWED_TOP_LEVEL_KEYS: frozenset[str] = frozenset(
{"schema_version", "reviewers", "expert_demo_claims"}
)
ALLOWED_REVIEWER_KEYS: frozenset[str] = frozenset(
{"reviewer_id", "display_name", "role", "domains", "review_scope", "provenance"}
)
ALLOWED_EXPERT_DEMO_CLAIM_KEYS: frozenset[str] = frozenset(
{
"domain_id",
"evidence_lanes",
"evidence_revision",
"signed_by",
"claim_digest",
}
)
ALLOWED_ROLES: frozenset[str] = frozenset({"primary", "domain"})
ALLOWED_SCOPES: frozenset[str] = frozenset({"pack", "proposal", "chain", "eval"})
SCHEMA_VERSION: int = 1
@ -41,10 +52,36 @@ class Reviewer:
provenance: str
@dataclass(frozen=True, slots=True)
class ExpertDemoClaim:
"""Reviewer-signed expert-demo promotion claim (ADR-0106).
A row in ``expert_demo_claims`` asserts that ``signed_by`` has
inspected the evidence at ``evidence_revision`` for ``domain_id``
across ``evidence_lanes`` and that the canonical evidence-bundle
SHA-256 equals ``claim_digest``. The reporting layer re-derives the
digest from the lane results on disk; a mismatch demotes the row
back to ``reasoning-capable``.
"""
domain_id: str
evidence_lanes: tuple[str, ...]
evidence_revision: str
signed_by: str
claim_digest: str
@dataclass(frozen=True, slots=True)
class ReviewerRegistry:
schema_version: int
reviewers: tuple[Reviewer, ...]
expert_demo_claims: tuple[ExpertDemoClaim, ...] = ()
def expert_demo_claim_for(self, domain_id: str) -> ExpertDemoClaim | None:
for claim in self.expert_demo_claims:
if claim.domain_id == domain_id:
return claim
return None
def resolve(self, reviewer_id: str) -> Reviewer | None:
for reviewer in self.reviewers:
@ -117,7 +154,89 @@ def load_reviewer_registry(path: Path) -> ReviewerRegistry:
seen_ids.add(reviewer.reviewer_id)
parsed.append(reviewer)
return ReviewerRegistry(schema_version=SCHEMA_VERSION, reviewers=tuple(parsed))
claims_raw = raw.get("expert_demo_claims", [])
if not isinstance(claims_raw, list):
raise ReviewerRegistryError(
"reviewer registry 'expert_demo_claims' must be a list when present"
)
parsed_claims: list[ExpertDemoClaim] = []
seen_domains: set[str] = set()
reviewer_ids = {reviewer.reviewer_id for reviewer in parsed}
for index, entry in enumerate(claims_raw):
claim = _parse_expert_demo_claim(entry, index=index, reviewer_ids=reviewer_ids)
if claim.domain_id in seen_domains:
raise ReviewerRegistryError(
f"expert_demo_claims contains duplicate domain_id "
f"{claim.domain_id!r}"
)
seen_domains.add(claim.domain_id)
parsed_claims.append(claim)
return ReviewerRegistry(
schema_version=SCHEMA_VERSION,
reviewers=tuple(parsed),
expert_demo_claims=tuple(parsed_claims),
)
def _parse_expert_demo_claim(
entry: Any, *, index: int, reviewer_ids: set[str]
) -> ExpertDemoClaim:
if not isinstance(entry, Mapping):
raise ReviewerRegistryError(
f"expert_demo_claims entry at index {index} must be a mapping"
)
unknown = set(entry.keys()) - ALLOWED_EXPERT_DEMO_CLAIM_KEYS
if unknown:
raise ReviewerRegistryError(
f"expert_demo_claims entry at index {index} has unknown fields: "
f"{sorted(unknown)}"
)
missing = ALLOWED_EXPERT_DEMO_CLAIM_KEYS - set(entry.keys())
if missing:
raise ReviewerRegistryError(
f"expert_demo_claims entry at index {index} missing required "
f"fields: {sorted(missing)}"
)
domain_id = _require_nonempty_str(
entry["domain_id"], field="domain_id", index=index
)
evidence_revision = _require_nonempty_str(
entry["evidence_revision"], field="evidence_revision", index=index
)
signed_by = _require_nonempty_str(
entry["signed_by"], field="signed_by", index=index
)
claim_digest = _require_nonempty_str(
entry["claim_digest"], field="claim_digest", index=index
)
if signed_by not in reviewer_ids:
raise ReviewerRegistryError(
f"expert_demo_claims entry at index {index} signed_by "
f"{signed_by!r} does not resolve to a registered reviewer"
)
if len(claim_digest) != 64 or any(
c not in "0123456789abcdef" for c in claim_digest
):
raise ReviewerRegistryError(
f"expert_demo_claims entry at index {index} claim_digest must be "
"a lowercase 64-char SHA-256 hex string"
)
evidence_lanes = _require_str_list(
entry["evidence_lanes"], field="evidence_lanes", index=index
)
if not evidence_lanes:
raise ReviewerRegistryError(
f"expert_demo_claims entry at index {index} has empty "
"'evidence_lanes' list"
)
return ExpertDemoClaim(
domain_id=domain_id,
evidence_lanes=evidence_lanes,
evidence_revision=evidence_revision,
signed_by=signed_by,
claim_digest=claim_digest,
)
def _parse_reviewer(entry: Any, *, index: int) -> Reviewer:

View file

@ -1,6 +1,6 @@
# ADR-0106 — Expert-Demo Promotion Contract
**Status:** Proposed
**Status:** Accepted
**Date:** 2026-05-22
**Author:** CORE agents + reviewers
**Depends on:** ADR-0091, ADR-0092, ADR-0093, ADR-0096, ADR-0097, ADR-0100, ADR-0101, ADR-0102

View file

@ -25,7 +25,7 @@ ADRs record significant architectural decisions: what was decided, why, what alt
| [ADR-0103](ADR-0103-fluency-lane-attachment-for-adr-0102.md) | Fluency Lane Attachment for ADR-0102 | Accepted (2026-05-22) |
| [ADR-0104](ADR-0104-curriculum-sourced-teaching-proposals.md) | Curriculum-Sourced Teaching Proposals | Accepted (2026-05-22) |
| [ADR-0105](ADR-0105-sealed-holdout-encryption.md) | Sealed Holdout Encryption via age | Accepted (2026-05-22) |
| [ADR-0106](ADR-0106-expert-demo-promotion-contract.md) | Expert-Demo Promotion Contract | Proposed (2026-05-22) |
| [ADR-0106](ADR-0106-expert-demo-promotion-contract.md) | Expert-Demo Promotion Contract | Accepted (2026-05-22) |
| [ADR-0108](ADR-0108-proposed-adr-sequencing.md) | Proposed-ADR Sequencing Post-ADR-0105 | Accepted (2026-05-22) |
---
@ -64,11 +64,10 @@ Seven lanes are SHA-pinned in `scripts/verify_lane_shas.py` and gated by the `la
Sequencing per ADR-0108. Listed in priority order:
1. **[ADR-0106](ADR-0106-expert-demo-promotion-contract.md) — Expert-Demo Promotion Contract.** Highest priority. Defines a domain-aware, reviewer-signed `expert_demo` predicate; closes the cognition-lane bleed at `core/capability/reporting.py:418`. Makes the four `reasoning-capable` ratifications externally legible.
2. **ADR-0107 (reserved) — `mathematics_logic` expert-demo promotion.** First worked promotion against ADR-0106. Smallest expert-demo proof surface across the four ratified domains.
3. **[ADR-0080](ADR-0080-contemplation-loop.md) — Contemplation Loop.** Sandboxed, read-only Phase 1 self-interrogation; emits `SPECULATIVE` findings from `frontier_compare` reports. Converts gap-finding from human-driven to system-emitted-and-reviewed.
4. **[ADR-0084](ADR-0084-definitional-layer.md) — Definitional Layer for Lexicon Packs.** Optional per-entry definitional block. Deferred pending ADR-0107 — value surfaces during a worked expert promotion that needs definitional depth.
5. **[ADR-0087](ADR-0087-rhetorical-style-axis.md) — Rhetorical Style Axis.** A third substantive selection axis sibling to anchor-lens. Lowest current priority — no active downstream consumer; register + anchor-lens already demonstrate the orthogonality pattern.
1. **ADR-0107 (reserved) — `mathematics_logic` expert-demo promotion.** First worked promotion against the now-Accepted ADR-0106 contract. Smallest expert-demo proof surface across the four ratified domains.
2. **[ADR-0080](ADR-0080-contemplation-loop.md) — Contemplation Loop.** Sandboxed, read-only Phase 1 self-interrogation; emits `SPECULATIVE` findings from `frontier_compare` reports. Converts gap-finding from human-driven to system-emitted-and-reviewed.
3. **[ADR-0084](ADR-0084-definitional-layer.md) — Definitional Layer for Lexicon Packs.** Optional per-entry definitional block. Deferred pending ADR-0107 — value surfaces during a worked expert promotion that needs definitional depth.
4. **[ADR-0087](ADR-0087-rhetorical-style-axis.md) — Rhetorical Style Axis.** A third substantive selection axis sibling to anchor-lens. Lowest current priority — no active downstream consumer; register + anchor-lens already demonstrate the orthogonality pattern.
### Open candidate directions (no ADR yet)
@ -78,6 +77,8 @@ Sequencing per ADR-0108. Listed in priority order:
## Accepted reasoning-capable domains
Per ADR-0106, `expert_demo` is **contract-gated**, not threshold-only: a domain row may carry `expert_demo=true` only when a reviewer-signed `expert_demo_claims` entry exists whose evidence-bundle digest reproduces byte-for-byte. All four currently-ratified domains sit at `expert_demo=false` until a worked promotion ADR (ADR-0107+) lands.
| Domain | Ratification ADR | Pack(s) | Evidence summary |
|---|---|---|---|
| `mathematics_logic` | ADR-0097 | `en_mathematics_logic_v1` | All nine ADR-0091 predicates pass; ledger row is `reasoning-capable`; `expert_demo` remains false; lanes include positive coverage, inference closure, and fabrication control. |

View file

@ -0,0 +1,352 @@
"""ADR-0106 — expert-demo promotion contract invariants.
Pins three load-bearing invariants:
1. ``expert_demo_requires_signature`` no domain row may carry
``expert_demo=true`` without a corresponding ``expert_demo_claims``
entry whose digest reproduces the on-disk evidence bundle.
2. ``expert_demo_domain_aware`` the reporting layer must consult only
lanes attached to a domain's ratified packs when computing
``expert_demo``. Cross-domain lane bleed (e.g. cognition-lane metrics
deciding a math promotion) is refused.
3. ``expert_demo_replay_byte_equality`` re-running every consulted
lane at ``evidence_revision`` must reproduce the exact JSON bytes
hashed into ``claim_digest``. Drift demotes the row back to
``reasoning-capable``.
The current main-branch state of the ledger has zero domains with
``expert_demo=true``; these tests synthesize fixtures to prove the gate
behaves as ADR-0106 specifies, without flipping any production row.
"""
from __future__ import annotations
from core.capability.expert_demo import (
derive_evidence_digest,
evaluate_expert_demo,
)
from core.capability.reviewers import (
ExpertDemoClaim,
Reviewer,
ReviewerRegistry,
)
_GOOD_METRICS = {
"surface_groundedness": 0.97,
"term_capture_rate": 0.90,
"intent_accuracy": 0.96,
"versor_closure_rate": 1.0,
}
_FAB_METRICS = {"passed_rate": 1.0}
def _primary_reviewer() -> Reviewer:
return Reviewer(
reviewer_id="shay-j",
display_name="Joshua Shay",
role="primary",
domains=("*",),
review_scope=("pack", "proposal", "chain", "eval"),
provenance="adr-0092:bootstrap:2026-05-21",
)
def _build_registry(
reviewers: tuple[Reviewer, ...],
claims: tuple[ExpertDemoClaim, ...] = (),
) -> ReviewerRegistry:
return ReviewerRegistry(
schema_version=1, reviewers=reviewers, expert_demo_claims=claims
)
def _good_lane_results(lanes: tuple[str, ...]) -> dict[str, dict[str, dict]]:
out: dict[str, dict[str, dict]] = {}
for lane in lanes:
metrics = _FAB_METRICS if lane == "fabrication_control" else _GOOD_METRICS
out[lane] = {"public": dict(metrics), "holdout": dict(metrics)}
return out
class TestExpertDemoRequiresSignature:
def test_no_claim_refuses_promotion(self) -> None:
registry = _build_registry((_primary_reviewer(),))
verdict = evaluate_expert_demo(
domain_id="mathematics_logic",
reasoning_capable=True,
registry=registry,
domain_lanes=("inference_closure", "fabrication_control"),
lane_results=_good_lane_results(
("inference_closure", "fabrication_control")
),
)
assert verdict.passed is False
assert "no expert_demo_claims entry" in verdict.reason
def test_unsigned_lanes_refuse_promotion(self) -> None:
registry = _build_registry(())
verdict = evaluate_expert_demo(
domain_id="mathematics_logic",
reasoning_capable=True,
registry=registry,
domain_lanes=("inference_closure",),
lane_results=_good_lane_results(("inference_closure",)),
)
assert verdict.passed is False
def test_not_reasoning_capable_refuses_promotion(self) -> None:
registry = _build_registry((_primary_reviewer(),))
verdict = evaluate_expert_demo(
domain_id="mathematics_logic",
reasoning_capable=False,
registry=registry,
domain_lanes=("inference_closure",),
lane_results=_good_lane_results(("inference_closure",)),
)
assert verdict.passed is False
assert "not yet reasoning-capable" in verdict.reason
def test_signer_without_eval_scope_refuses_promotion(self) -> None:
domain = "mathematics_logic"
reviewer = Reviewer(
reviewer_id="math-pack-reviewer",
display_name="Math Pack Reviewer",
role="domain",
domains=(domain,),
review_scope=("pack", "chain"),
provenance="adr-0092:bootstrap:2026-05-21",
)
lanes = ("inference_closure", "fabrication_control")
results = _good_lane_results(lanes)
digest = derive_evidence_digest(
domain_id=domain,
evidence_revision="abc123",
evidence_lanes=lanes,
lane_results=results,
)
claim = ExpertDemoClaim(
domain_id=domain,
evidence_lanes=lanes,
evidence_revision="abc123",
signed_by="math-pack-reviewer",
claim_digest=digest,
)
registry = _build_registry((reviewer,), (claim,))
verdict = evaluate_expert_demo(
domain_id=domain,
reasoning_capable=True,
registry=registry,
domain_lanes=lanes,
lane_results=results,
)
assert verdict.passed is False
assert "cannot review eval-scope" in verdict.reason
class TestExpertDemoDomainAware:
def test_cross_domain_lane_bleed_refused(self) -> None:
"""A claim that cites a cognition lane for a math promotion fails."""
domain = "mathematics_logic"
math_lanes = ("inference_closure", "fabrication_control")
bad_claim_lanes = ("cognition",)
results = _good_lane_results(bad_claim_lanes + math_lanes)
digest = derive_evidence_digest(
domain_id=domain,
evidence_revision="rev1",
evidence_lanes=bad_claim_lanes,
lane_results=results,
)
claim = ExpertDemoClaim(
domain_id=domain,
evidence_lanes=bad_claim_lanes,
evidence_revision="rev1",
signed_by="shay-j",
claim_digest=digest,
)
registry = _build_registry((_primary_reviewer(),), (claim,))
verdict = evaluate_expert_demo(
domain_id=domain,
reasoning_capable=True,
registry=registry,
domain_lanes=math_lanes,
lane_results=results,
)
assert verdict.passed is False
assert "lanes not attached to domain" in verdict.reason
def test_in_domain_lanes_accepted(self) -> None:
domain = "mathematics_logic"
lanes = ("inference_closure", "fabrication_control")
results = _good_lane_results(lanes)
digest = derive_evidence_digest(
domain_id=domain,
evidence_revision="rev1",
evidence_lanes=lanes,
lane_results=results,
)
claim = ExpertDemoClaim(
domain_id=domain,
evidence_lanes=lanes,
evidence_revision="rev1",
signed_by="shay-j",
claim_digest=digest,
)
registry = _build_registry((_primary_reviewer(),), (claim,))
verdict = evaluate_expert_demo(
domain_id=domain,
reasoning_capable=True,
registry=registry,
domain_lanes=lanes,
lane_results=results,
)
assert verdict.passed is True
assert verdict.derived_digest == digest
class TestExpertDemoReplayByteEquality:
def test_digest_is_stable_across_invocations(self) -> None:
lanes = ("inference_closure", "fabrication_control")
results = _good_lane_results(lanes)
first = derive_evidence_digest(
domain_id="mathematics_logic",
evidence_revision="rev1",
evidence_lanes=lanes,
lane_results=results,
)
second = derive_evidence_digest(
domain_id="mathematics_logic",
evidence_revision="rev1",
evidence_lanes=lanes,
lane_results=results,
)
assert first == second
def test_digest_is_order_independent_on_lanes(self) -> None:
results = _good_lane_results(("a", "b", "c"))
d1 = derive_evidence_digest(
domain_id="d",
evidence_revision="rev1",
evidence_lanes=("a", "b", "c"),
lane_results=results,
)
d2 = derive_evidence_digest(
domain_id="d",
evidence_revision="rev1",
evidence_lanes=("c", "a", "b"),
lane_results=results,
)
assert d1 == d2
def test_drift_in_results_demotes_row(self) -> None:
domain = "mathematics_logic"
lanes = ("inference_closure", "fabrication_control")
original = _good_lane_results(lanes)
original_digest = derive_evidence_digest(
domain_id=domain,
evidence_revision="rev1",
evidence_lanes=lanes,
lane_results=original,
)
claim = ExpertDemoClaim(
domain_id=domain,
evidence_lanes=lanes,
evidence_revision="rev1",
signed_by="shay-j",
claim_digest=original_digest,
)
drifted = _good_lane_results(lanes)
drifted["inference_closure"]["public"]["intent_accuracy"] = 0.99
registry = _build_registry((_primary_reviewer(),), (claim,))
verdict = evaluate_expert_demo(
domain_id=domain,
reasoning_capable=True,
registry=registry,
domain_lanes=lanes,
lane_results=drifted,
)
assert verdict.passed is False
assert "replay drift" in verdict.reason
assert verdict.derived_digest is not None
assert verdict.derived_digest != original_digest
class TestExpertDemoThresholds:
def test_below_threshold_metric_refuses(self) -> None:
domain = "mathematics_logic"
lanes = ("inference_closure", "fabrication_control")
results = _good_lane_results(lanes)
results["inference_closure"]["holdout"]["surface_groundedness"] = 0.50
digest = derive_evidence_digest(
domain_id=domain,
evidence_revision="rev1",
evidence_lanes=lanes,
lane_results=results,
)
claim = ExpertDemoClaim(
domain_id=domain,
evidence_lanes=lanes,
evidence_revision="rev1",
signed_by="shay-j",
claim_digest=digest,
)
registry = _build_registry((_primary_reviewer(),), (claim,))
verdict = evaluate_expert_demo(
domain_id=domain,
reasoning_capable=True,
registry=registry,
domain_lanes=lanes,
lane_results=results,
)
assert verdict.passed is False
assert "surface_groundedness" in verdict.reason
assert "below threshold" in verdict.reason
def test_fabrication_control_failure_refuses(self) -> None:
domain = "mathematics_logic"
lanes = ("inference_closure", "fabrication_control")
results = _good_lane_results(lanes)
results["fabrication_control"]["holdout"]["passed_rate"] = 0.8
digest = derive_evidence_digest(
domain_id=domain,
evidence_revision="rev1",
evidence_lanes=lanes,
lane_results=results,
)
claim = ExpertDemoClaim(
domain_id=domain,
evidence_lanes=lanes,
evidence_revision="rev1",
signed_by="shay-j",
claim_digest=digest,
)
registry = _build_registry((_primary_reviewer(),), (claim,))
verdict = evaluate_expert_demo(
domain_id=domain,
reasoning_capable=True,
registry=registry,
domain_lanes=lanes,
lane_results=results,
)
assert verdict.passed is False
assert "fabrication_control" in verdict.reason
class TestProductionLedgerUntouched:
"""ADR-0106 §Acceptance: no domain row's expert_demo field flips."""
def test_no_production_domain_promoted_by_this_adr(self) -> None:
from core.capability.reporting import ledger_report
report = ledger_report()
promoted = [
row["domain"]
for row in report.get("domains", [])
if row.get("predicates", {}).get("expert_demo")
]
assert promoted == [], (
f"ADR-0106 must not promote any domain by code change alone; "
f"got: {promoted}"
)

View file

@ -7,6 +7,7 @@ from pathlib import Path
import pytest
from core.capability.reviewers import (
ExpertDemoClaim,
ReviewerRegistry,
ReviewerRegistryError,
SCHEMA_VERSION,
@ -14,6 +15,9 @@ from core.capability.reviewers import (
)
_VALID_DIGEST = "a" * 64
def _write_yaml(tmp_path: Path, content: str) -> Path:
path = tmp_path / "reviewers.yaml"
path.write_text(content, encoding="utf-8")
@ -282,3 +286,65 @@ class TestProductionRegistry:
assert shay.role == "primary"
assert shay.domains == ("*",)
assert shay.provenance.startswith("adr-0092:bootstrap:")
class TestExpertDemoClaimsSchema:
"""ADR-0106 — additive ``expert_demo_claims`` block on the registry."""
def _yaml_with_claim(self, *, signed_by: str, digest: str = _VALID_DIGEST) -> str:
return f"""\
schema_version: 1
reviewers:
- reviewer_id: shay-j
display_name: "Joshua Shay"
role: primary
domains: ["*"]
review_scope: ["pack", "proposal", "chain", "eval"]
provenance: "adr-0092:bootstrap:2026-05-21"
expert_demo_claims:
- domain_id: mathematics_logic
evidence_lanes: ["inference_closure", "fabrication_control"]
evidence_revision: "abc1234"
signed_by: "{signed_by}"
claim_digest: "{digest}"
"""
def test_omitted_block_yields_empty_tuple(self, tmp_path: Path) -> None:
path = _write_yaml(tmp_path, _valid_primary_yaml())
registry = load_reviewer_registry(path)
assert registry.expert_demo_claims == ()
assert registry.expert_demo_claim_for("mathematics_logic") is None
def test_valid_claim_parses(self, tmp_path: Path) -> None:
path = _write_yaml(tmp_path, self._yaml_with_claim(signed_by="shay-j"))
registry = load_reviewer_registry(path)
assert len(registry.expert_demo_claims) == 1
claim = registry.expert_demo_claim_for("mathematics_logic")
assert isinstance(claim, ExpertDemoClaim)
assert claim.signed_by == "shay-j"
assert claim.claim_digest == _VALID_DIGEST
assert claim.evidence_lanes == ("inference_closure", "fabrication_control")
def test_signer_unknown_reviewer_id_rejected(self, tmp_path: Path) -> None:
path = _write_yaml(tmp_path, self._yaml_with_claim(signed_by="ghost"))
with pytest.raises(ReviewerRegistryError, match="does not resolve"):
load_reviewer_registry(path)
def test_malformed_digest_rejected(self, tmp_path: Path) -> None:
path = _write_yaml(
tmp_path, self._yaml_with_claim(signed_by="shay-j", digest="short")
)
with pytest.raises(ReviewerRegistryError, match="claim_digest"):
load_reviewer_registry(path)
def test_duplicate_domain_rejected(self, tmp_path: Path) -> None:
yaml = self._yaml_with_claim(signed_by="shay-j")
yaml += """ - domain_id: mathematics_logic
evidence_lanes: ["inference_closure"]
evidence_revision: "abc1234"
signed_by: "shay-j"
claim_digest: "{digest}"
""".format(digest=_VALID_DIGEST)
path = _write_yaml(tmp_path, yaml)
with pytest.raises(ReviewerRegistryError, match="duplicate domain_id"):
load_reviewer_registry(path)