Merge pull request 'core proposal-queue reseal — the ledger_reseal stage, which refuses to grant a license by accident (Phase D)' (#122) from feat/proposal-queue-reseal into feat/curriculum-practice-producer
This commit is contained in:
commit
e57b71a40c
4 changed files with 631 additions and 1 deletions
|
|
@ -139,6 +139,84 @@ def cmd_proposal_queue_ratify(args: argparse.Namespace) -> int:
|
||||||
f"{receipt.family_chains_before} -> {receipt.family_chains_after}")
|
f"{receipt.family_chains_before} -> {receipt.family_chains_after}")
|
||||||
if not args.dry_run:
|
if not args.dry_run:
|
||||||
print(f" still pending : {', '.join(receipt.pending_stages)}")
|
print(f" still pending : {', '.join(receipt.pending_stages)}")
|
||||||
|
# Wiring the receipt's `ledger_reseal` stage to the verb that performs
|
||||||
|
# it. Naming a pending stage without naming its command is how this one
|
||||||
|
# stayed unperformed since the ceremony was written.
|
||||||
|
if "ledger_reseal" in receipt.pending_stages:
|
||||||
|
print(
|
||||||
|
" next : core proposal-queue reseal curriculum_serve "
|
||||||
|
"--dry-run (the ledger is stale until you do)"
|
||||||
|
)
|
||||||
|
return 0
|
||||||
|
|
||||||
|
|
||||||
|
def cmd_proposal_queue_reseal(args: argparse.Namespace) -> int:
|
||||||
|
"""The ``ledger_reseal`` stage of the ceremony — explicit, never automatic.
|
||||||
|
|
||||||
|
``ratify_chain`` grows the corpus and names this stage in its receipt's
|
||||||
|
``pending_stages``; nothing performed it until now, so a sealed ledger went
|
||||||
|
stale the moment curriculum changed. This is the verb a human runs.
|
||||||
|
|
||||||
|
It refuses by default to grant a license. A band can clear θ_SERVE on correct
|
||||||
|
NON-COMMITMENTS alone, so "regenerate the artifact" and "make four bands
|
||||||
|
authoritative" would otherwise be the same keystroke.
|
||||||
|
"""
|
||||||
|
from teaching.ledger_reseal import ReadyToReseal, apply_reseal, plan_reseal
|
||||||
|
|
||||||
|
try:
|
||||||
|
plan = plan_reseal(args.capability)
|
||||||
|
except ReadyToReseal as exc:
|
||||||
|
print(f"REFUSED: {exc}")
|
||||||
|
return 2
|
||||||
|
|
||||||
|
payload: dict[str, object] = {
|
||||||
|
"capability": plan.capability,
|
||||||
|
"path": str(plan.path),
|
||||||
|
"licensed_before": list(plan.licensed_before),
|
||||||
|
"licensed_after": list(plan.licensed_after),
|
||||||
|
"newly_licensed": list(plan.newly_licensed),
|
||||||
|
"revoked": list(plan.revoked),
|
||||||
|
"committed": plan.committed,
|
||||||
|
"mix": plan.mix,
|
||||||
|
"is_housekeeping": plan.is_housekeeping,
|
||||||
|
"written": False,
|
||||||
|
}
|
||||||
|
if not args.json:
|
||||||
|
print(f"reseal plan for {plan.capability}:")
|
||||||
|
print(f" artifact : {plan.path}")
|
||||||
|
for band in sorted(plan.committed):
|
||||||
|
mix = ", ".join(f"{k}={v}" for k, v in sorted(plan.mix.get(band, {}).items()))
|
||||||
|
lic = "SERVE" if band in plan.licensed_after else " "
|
||||||
|
print(f" {lic} {band:44s} committed={plan.committed[band]:6d} {mix}")
|
||||||
|
print(f" licensed now : {len(plan.licensed_before)}")
|
||||||
|
print(f" licensed after: {len(plan.licensed_after)}")
|
||||||
|
if plan.newly_licensed:
|
||||||
|
print(f" NEWLY LICENSED: {', '.join(plan.newly_licensed)}")
|
||||||
|
if plan.revoked:
|
||||||
|
print(f" revoked : {', '.join(plan.revoked)}")
|
||||||
|
|
||||||
|
if args.dry_run:
|
||||||
|
if args.json:
|
||||||
|
print(json.dumps(payload, indent=2, sort_keys=True))
|
||||||
|
else:
|
||||||
|
print(" (dry run — nothing written)")
|
||||||
|
return 0
|
||||||
|
|
||||||
|
try:
|
||||||
|
path = apply_reseal(plan, allow_new_licenses=args.allow_new_licenses)
|
||||||
|
except ReadyToReseal as exc:
|
||||||
|
if args.json:
|
||||||
|
payload["refused"] = str(exc)
|
||||||
|
print(json.dumps(payload, indent=2, sort_keys=True))
|
||||||
|
else:
|
||||||
|
print(f"REFUSED: {exc}")
|
||||||
|
return 1
|
||||||
|
|
||||||
|
if args.json:
|
||||||
|
payload["written"] = True
|
||||||
|
print(json.dumps(payload, indent=2, sort_keys=True))
|
||||||
|
else:
|
||||||
|
print(f" SEALED -> {path}")
|
||||||
return 0
|
return 0
|
||||||
|
|
||||||
|
|
||||||
|
|
@ -198,7 +276,7 @@ def register(subparsers: argparse._SubParsersAction) -> None:
|
||||||
"curriculum loader admits the chain -> receipt. Refuses, and rolls "
|
"curriculum loader admits the chain -> receipt. Refuses, and rolls "
|
||||||
"back, if the appended row would be silently dropped. Does not "
|
"back, if the appended row would be silently dropped. Does not "
|
||||||
"write a ledger (bridge rule 1) or queue an arena entry; the "
|
"write a ledger (bridge rule 1) or queue an arena entry; the "
|
||||||
"receipt names those stages."
|
"receipt names those stages, and `reseal` performs the first."
|
||||||
),
|
),
|
||||||
)
|
)
|
||||||
ratify.add_argument("domain")
|
ratify.add_argument("domain")
|
||||||
|
|
@ -225,5 +303,32 @@ def register(subparsers: argparse._SubParsersAction) -> None:
|
||||||
ratify.add_argument("--json", action="store_true")
|
ratify.add_argument("--json", action="store_true")
|
||||||
ratify.set_defaults(func=cmd_proposal_queue_ratify)
|
ratify.set_defaults(func=cmd_proposal_queue_ratify)
|
||||||
|
|
||||||
|
from teaching.ledger_reseal import resealable_capabilities
|
||||||
|
|
||||||
|
reseal = sub.add_parser(
|
||||||
|
"reseal",
|
||||||
|
help="re-run sealed practice and rewrite a capability's committed ledger",
|
||||||
|
description=(
|
||||||
|
"The `ledger_reseal` stage `ratify_chain` names but never performs. "
|
||||||
|
"Ratifying a chain grows the corpus without re-running practice, so "
|
||||||
|
"the ledger the serving gate reads goes stale. This rebuilds it from "
|
||||||
|
"sealed practice. REFUSES by default if the reseal would newly "
|
||||||
|
"license a band: reliability is commitment precision, so a band can "
|
||||||
|
"clear theta_SERVE on correct NON-COMMITMENTS alone, and granting a "
|
||||||
|
"serving license is a ratification rather than housekeeping."
|
||||||
|
),
|
||||||
|
)
|
||||||
|
reseal.add_argument("capability", choices=resealable_capabilities())
|
||||||
|
reseal.add_argument(
|
||||||
|
"--dry-run", action="store_true",
|
||||||
|
help="report the license delta without writing",
|
||||||
|
)
|
||||||
|
reseal.add_argument(
|
||||||
|
"--allow-new-licenses", action="store_true",
|
||||||
|
help="authorize bands that would become newly licensed (a RATIFICATION)",
|
||||||
|
)
|
||||||
|
reseal.add_argument("--json", action="store_true")
|
||||||
|
reseal.set_defaults(func=cmd_proposal_queue_reseal)
|
||||||
|
|
||||||
|
|
||||||
__all__ = ["register"]
|
__all__ = ["register"]
|
||||||
|
|
|
||||||
|
|
@ -237,6 +237,7 @@ TEST_SUITES: dict[str, tuple[str, ...]] = {
|
||||||
"tests/test_curriculum_serve.py",
|
"tests/test_curriculum_serve.py",
|
||||||
"tests/test_curriculum_practice.py",
|
"tests/test_curriculum_practice.py",
|
||||||
"tests/test_curriculum_polarity.py",
|
"tests/test_curriculum_polarity.py",
|
||||||
|
"tests/test_ledger_reseal.py",
|
||||||
"tests/test_ratified_ledger_bridge.py",
|
"tests/test_ratified_ledger_bridge.py",
|
||||||
"tests/test_vocab_trigger_instrument.py",
|
"tests/test_vocab_trigger_instrument.py",
|
||||||
),
|
),
|
||||||
|
|
|
||||||
195
teaching/ledger_reseal.py
Normal file
195
teaching/ledger_reseal.py
Normal file
|
|
@ -0,0 +1,195 @@
|
||||||
|
"""teaching/ledger_reseal.py — the ``ledger_reseal`` stage of the ceremony.
|
||||||
|
|
||||||
|
``RatificationReceipt.pending_stages`` has named ``ledger_reseal`` since the
|
||||||
|
ratification ceremony was written, and nothing performed it. Ratifying a chain
|
||||||
|
grows the corpus; it does not re-run practice, so the sealed ledger the serving
|
||||||
|
gate reads goes stale the moment curriculum changes. This module is that stage,
|
||||||
|
and it is deliberately **not** reachable from ``ratify_chain``.
|
||||||
|
|
||||||
|
Why it is a separate, human-run act
|
||||||
|
-----------------------------------
|
||||||
|
``AGENTS.md`` forbids ratification automation, and an auto-reseal on every
|
||||||
|
``ratify`` would churn the ledger's ``content_sha256`` once per row. But the
|
||||||
|
stronger reason is what Phase C measured: **resealing can grant licenses.**
|
||||||
|
Reliability is commitment precision and a correct UNKNOWN is a commitment, so
|
||||||
|
any band with ≥657 routable atoms clears θ_SERVE on non-commitments alone. A
|
||||||
|
plain "regenerate the artifact" verb would therefore flip four curriculum bands
|
||||||
|
from disclosed to authoritative as a side effect of housekeeping, on evidence
|
||||||
|
that is ~99% non-entailed — which ADR-0262 §5.1 rules unacceptable
|
||||||
|
(``docs/research/curriculum-practice-producer-2026-07-26.md`` §1).
|
||||||
|
|
||||||
|
So :func:`plan_reseal` computes the license DELTA before writing anything, and
|
||||||
|
the caller must opt in explicitly to any band that would become newly licensed.
|
||||||
|
The default is refusal. Regenerating an artifact is housekeeping; changing what
|
||||||
|
the engine will assert authoritatively is a ratification, and the two must not
|
||||||
|
share a keystroke.
|
||||||
|
|
||||||
|
Scope: this module writes a ledger and nothing else. It has no opinion on
|
||||||
|
whether curriculum content is true, cannot ratify a chain, and cannot flip a
|
||||||
|
serving flag.
|
||||||
|
"""
|
||||||
|
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
from dataclasses import dataclass
|
||||||
|
from pathlib import Path
|
||||||
|
from typing import Any, Callable
|
||||||
|
|
||||||
|
from core.ratified_ledger import (
|
||||||
|
CAPABILITY_LEDGERS,
|
||||||
|
load_capability_ledger,
|
||||||
|
write_sealed_ledger,
|
||||||
|
)
|
||||||
|
from core.reliability_gate import Action, Ceilings, ClassTally, license_for
|
||||||
|
|
||||||
|
|
||||||
|
class ReadyToReseal(ValueError):
|
||||||
|
"""A reseal was requested that cannot proceed as asked."""
|
||||||
|
|
||||||
|
|
||||||
|
@dataclass(frozen=True, slots=True)
|
||||||
|
class ResealPlan:
|
||||||
|
"""What a reseal would do, computed before it does it."""
|
||||||
|
|
||||||
|
capability: str
|
||||||
|
path: Path
|
||||||
|
#: Bands licensed by the ledger currently on disk (empty when absent).
|
||||||
|
licensed_before: tuple[str, ...]
|
||||||
|
#: Bands the freshly-built artifact would license.
|
||||||
|
licensed_after: tuple[str, ...]
|
||||||
|
#: Per-band committed volume in the new artifact.
|
||||||
|
committed: dict[str, int]
|
||||||
|
#: Per-band gold verdict mix — the axis ``ClassTally`` cannot carry.
|
||||||
|
mix: dict[str, dict[str, int]]
|
||||||
|
artifact: dict[str, Any]
|
||||||
|
|
||||||
|
@property
|
||||||
|
def newly_licensed(self) -> tuple[str, ...]:
|
||||||
|
"""Bands this reseal would grant SERVE that do not hold it today.
|
||||||
|
|
||||||
|
The figure that decides whether the operation is housekeeping or a
|
||||||
|
ratification.
|
||||||
|
"""
|
||||||
|
before = set(self.licensed_before)
|
||||||
|
return tuple(b for b in self.licensed_after if b not in before)
|
||||||
|
|
||||||
|
@property
|
||||||
|
def revoked(self) -> tuple[str, ...]:
|
||||||
|
"""Bands that would LOSE a license — always allowed, never gated.
|
||||||
|
|
||||||
|
Losing a license is the gate becoming more conservative, which needs no
|
||||||
|
authorization. Only granting one does.
|
||||||
|
"""
|
||||||
|
after = set(self.licensed_after)
|
||||||
|
return tuple(b for b in self.licensed_before if b not in after)
|
||||||
|
|
||||||
|
@property
|
||||||
|
def is_housekeeping(self) -> bool:
|
||||||
|
return not self.newly_licensed
|
||||||
|
|
||||||
|
|
||||||
|
#: capability -> (build artifact, build ledger, build mix).
|
||||||
|
#:
|
||||||
|
#: ``deduction_serve`` is deliberately ABSENT. Its ledger is SHA-sealed,
|
||||||
|
#: ratified, and gating a live flag, and 21 of its 25 bands do not clear
|
||||||
|
#: θ_SERVE on distinct evidence (``tests/test_volume_honesty.py``). Re-sealing it
|
||||||
|
#: is Shay's ratification, not a CLI verb's, and a reseal that silently
|
||||||
|
#: regenerated it would erase the exposure this arc measured. ``estimation``
|
||||||
|
#: ships sealed with its gate and has no reason to move.
|
||||||
|
_RESEALABLE: dict[str, tuple[Callable[[], dict[str, Any]], Callable[[], dict[str, ClassTally]], Callable[[], dict[str, dict[str, int]]]]] = {}
|
||||||
|
|
||||||
|
|
||||||
|
def _register_curriculum() -> None:
|
||||||
|
from evals.curriculum_serve.practice.runner import (
|
||||||
|
build_ledger,
|
||||||
|
build_mix,
|
||||||
|
build_sealed_artifact,
|
||||||
|
)
|
||||||
|
|
||||||
|
_RESEALABLE["curriculum_serve"] = (build_sealed_artifact, build_ledger, build_mix)
|
||||||
|
|
||||||
|
|
||||||
|
def resealable_capabilities() -> tuple[str, ...]:
|
||||||
|
"""Capabilities this stage will rebuild, in a stable order."""
|
||||||
|
if not _RESEALABLE:
|
||||||
|
_register_curriculum()
|
||||||
|
return tuple(sorted(_RESEALABLE))
|
||||||
|
|
||||||
|
|
||||||
|
def _licensed_bands(ledger: dict[str, ClassTally], ceilings: Ceilings) -> tuple[str, ...]:
|
||||||
|
return tuple(
|
||||||
|
sorted(
|
||||||
|
band
|
||||||
|
for band, tally in ledger.items()
|
||||||
|
if license_for(tally, Action.SERVE, ceilings).licensed
|
||||||
|
)
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
def plan_reseal(capability: str, *, ceilings: Ceilings | None = None) -> ResealPlan:
|
||||||
|
"""Compute the reseal and its license delta — writing nothing.
|
||||||
|
|
||||||
|
The read of the CURRENT ledger goes through :func:`load_capability_ledger`,
|
||||||
|
the same production entry point the serving adapters use, so two properties
|
||||||
|
come for free. A hand-edited ledger refuses here rather than being silently
|
||||||
|
overwritten (losing the evidence that it was tampered with), and this stage
|
||||||
|
inherits the capability's DECLARED absence policy instead of choosing its own
|
||||||
|
— bridge rule 5: ``missing_ok`` is a fact about the capability, not about the
|
||||||
|
call, and no production path may pass it.
|
||||||
|
"""
|
||||||
|
if capability not in resealable_capabilities():
|
||||||
|
raise ReadyToReseal(
|
||||||
|
f"{capability!r} is not resealable — known: {list(resealable_capabilities())}. "
|
||||||
|
"Ledgers absent from that list are sealed by deliberate ratification only."
|
||||||
|
)
|
||||||
|
spec = CAPABILITY_LEDGERS[capability]
|
||||||
|
ceilings = ceilings if ceilings is not None else Ceilings.default()
|
||||||
|
build_artifact, build_ledger, build_mix = _RESEALABLE[capability]
|
||||||
|
|
||||||
|
current = load_capability_ledger(capability)
|
||||||
|
fresh_ledger = build_ledger()
|
||||||
|
artifact = build_artifact()
|
||||||
|
|
||||||
|
return ResealPlan(
|
||||||
|
capability=capability,
|
||||||
|
path=spec.path,
|
||||||
|
licensed_before=_licensed_bands(current, ceilings),
|
||||||
|
licensed_after=_licensed_bands(fresh_ledger, ceilings),
|
||||||
|
committed={band: t.committed for band, t in sorted(fresh_ledger.items())},
|
||||||
|
mix=build_mix(),
|
||||||
|
artifact=artifact,
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
def apply_reseal(plan: ResealPlan, *, allow_new_licenses: bool = False) -> Path:
|
||||||
|
"""Write the planned artifact, refusing to grant a license by accident.
|
||||||
|
|
||||||
|
``allow_new_licenses`` is the ratification. Without it, a plan that would
|
||||||
|
newly license any band refuses and names them — the operator then either
|
||||||
|
authorizes the grant or fixes the producer, but never discovers afterwards
|
||||||
|
that a serving surface became authoritative.
|
||||||
|
"""
|
||||||
|
if plan.newly_licensed and not allow_new_licenses:
|
||||||
|
detail = ", ".join(
|
||||||
|
f"{band} (committed={plan.committed.get(band, 0)}, "
|
||||||
|
f"entailed={plan.mix.get(band, {}).get('entailed', 0)})"
|
||||||
|
for band in plan.newly_licensed
|
||||||
|
)
|
||||||
|
raise ReadyToReseal(
|
||||||
|
f"{plan.capability}: this reseal would newly license {len(plan.newly_licensed)} "
|
||||||
|
f"band(s) — {detail}. That is a ratification, not a reseal: those bands "
|
||||||
|
"would begin serving authoritatively. Re-run with --allow-new-licenses "
|
||||||
|
"only if the grant is intended, and read the `entailed` counts first — a "
|
||||||
|
"band can clear theta_SERVE on correct NON-COMMITMENTS alone (ADR-0262 §5.1)."
|
||||||
|
)
|
||||||
|
write_sealed_ledger(plan.path, plan.artifact)
|
||||||
|
return plan.path
|
||||||
|
|
||||||
|
|
||||||
|
__all__ = [
|
||||||
|
"ReadyToReseal",
|
||||||
|
"ResealPlan",
|
||||||
|
"apply_reseal",
|
||||||
|
"plan_reseal",
|
||||||
|
"resealable_capabilities",
|
||||||
|
]
|
||||||
329
tests/test_ledger_reseal.py
Normal file
329
tests/test_ledger_reseal.py
Normal file
|
|
@ -0,0 +1,329 @@
|
||||||
|
"""Phase D — the ``ledger_reseal`` ceremony stage and its CLI verb.
|
||||||
|
|
||||||
|
`RatificationReceipt.pending_stages` has named `ledger_reseal` since the
|
||||||
|
ratification ceremony was written and nothing performed it. This pins the verb
|
||||||
|
that performs it, and — more importantly — the reason it refuses by default.
|
||||||
|
|
||||||
|
The load-bearing property is **not** "the ledger can be rewritten". It is that
|
||||||
|
rewriting it cannot grant a serving license by accident. Reliability is
|
||||||
|
commitment precision, so a curriculum band clears θ_SERVE on correct
|
||||||
|
NON-COMMITMENTS alone; without the gate below, `reseal` would be a housekeeping
|
||||||
|
verb that silently made four bands authoritative.
|
||||||
|
"""
|
||||||
|
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
import dataclasses
|
||||||
|
import json
|
||||||
|
from pathlib import Path
|
||||||
|
|
||||||
|
import pytest
|
||||||
|
|
||||||
|
from core.ratified_ledger import CAPABILITY_LEDGERS, LedgerSpec, load_sealed_ledger
|
||||||
|
from core.reliability_gate import ClassTally
|
||||||
|
from teaching.ledger_reseal import (
|
||||||
|
ReadyToReseal,
|
||||||
|
ResealPlan,
|
||||||
|
apply_reseal,
|
||||||
|
plan_reseal,
|
||||||
|
resealable_capabilities,
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.fixture(scope="module")
|
||||||
|
def plan() -> ResealPlan:
|
||||||
|
return plan_reseal("curriculum_serve")
|
||||||
|
|
||||||
|
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
# 1. What is resealable, and what must never be.
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
|
||||||
|
def test_only_curriculum_is_resealable() -> None:
|
||||||
|
"""`deduction_serve` must not be reachable from a CLI verb.
|
||||||
|
|
||||||
|
Its ledger is SHA-sealed, ratified, and gating a live flag, and 21 of its 25
|
||||||
|
bands do not clear θ_SERVE on distinct evidence
|
||||||
|
(`tests/test_volume_honesty.py`). A reseal verb that regenerated it would
|
||||||
|
erase a measured exposure as a side effect of housekeeping. `estimation`
|
||||||
|
ships sealed with its own gate and has no reason to move.
|
||||||
|
"""
|
||||||
|
assert resealable_capabilities() == ("curriculum_serve",)
|
||||||
|
assert "deduction_serve" not in resealable_capabilities()
|
||||||
|
assert "estimation" not in resealable_capabilities()
|
||||||
|
# ...and both are still registered capabilities, so this is an exclusion by
|
||||||
|
# decision rather than by them being unknown.
|
||||||
|
assert {"deduction_serve", "estimation"} <= set(CAPABILITY_LEDGERS)
|
||||||
|
|
||||||
|
|
||||||
|
def test_unregistered_capability_refuses() -> None:
|
||||||
|
with pytest.raises(ReadyToReseal, match="not resealable"):
|
||||||
|
plan_reseal("deduction_serve")
|
||||||
|
|
||||||
|
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
# 2. Planning writes nothing, and reports the license delta.
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
|
||||||
|
def test_planning_writes_nothing(plan: ResealPlan) -> None:
|
||||||
|
"""A plan is a read. The artifact must still be absent afterwards."""
|
||||||
|
assert not plan.path.exists(), (
|
||||||
|
"planning a reseal must not create the ledger — see "
|
||||||
|
"tests/test_curriculum_practice.py::test_curriculum_ledger_is_not_committed"
|
||||||
|
)
|
||||||
|
assert plan.capability == "curriculum_serve"
|
||||||
|
assert plan.artifact["schema"] == "curriculum_serve_ledger_v1"
|
||||||
|
|
||||||
|
|
||||||
|
def test_plan_reports_the_bands_a_reseal_would_newly_license(plan: ResealPlan) -> None:
|
||||||
|
"""The whole point of the stage being a separate act.
|
||||||
|
|
||||||
|
Nothing is licensed today (the ledger is absent), and a reseal would license
|
||||||
|
four bands — every band whose routable atom space reaches 657. Those are the
|
||||||
|
numbers `docs/research/curriculum-practice-producer-2026-07-26.md` §1 records.
|
||||||
|
"""
|
||||||
|
assert plan.licensed_before == ()
|
||||||
|
assert len(plan.licensed_after) == 4
|
||||||
|
assert set(plan.newly_licensed) == set(plan.licensed_after)
|
||||||
|
assert plan.revoked == ()
|
||||||
|
assert plan.is_housekeeping is False
|
||||||
|
|
||||||
|
|
||||||
|
def test_newly_licensed_bands_are_licensed_on_non_commitments(plan: ResealPlan) -> None:
|
||||||
|
"""The mix the operator must see before authorizing — `ClassTally` has no
|
||||||
|
verdict axis, so the gate itself cannot distinguish these from real
|
||||||
|
competence."""
|
||||||
|
for band in plan.newly_licensed:
|
||||||
|
mix = plan.mix[band]
|
||||||
|
assert mix.get("entailed", 0) <= 9
|
||||||
|
assert mix.get("unknown", 0) > 600
|
||||||
|
assert plan.committed[band] >= 657
|
||||||
|
|
||||||
|
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
# 3. Applying refuses to grant a license by accident. This is the unit.
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
|
||||||
|
def test_apply_refuses_to_grant_a_license_and_writes_nothing(
|
||||||
|
plan: ResealPlan, tmp_path: Path
|
||||||
|
) -> None:
|
||||||
|
target = tmp_path / "curriculum_serve_ledger.json"
|
||||||
|
redirected = dataclasses.replace(plan, path=target)
|
||||||
|
with pytest.raises(ReadyToReseal, match="would newly license 4 band"):
|
||||||
|
apply_reseal(redirected)
|
||||||
|
assert not target.exists(), "a refused reseal must not have written"
|
||||||
|
|
||||||
|
|
||||||
|
def test_refusal_names_the_bands_and_their_entailed_counts(
|
||||||
|
plan: ResealPlan, tmp_path: Path
|
||||||
|
) -> None:
|
||||||
|
"""A refusal an operator cannot act on is just an obstacle."""
|
||||||
|
with pytest.raises(ReadyToReseal) as exc:
|
||||||
|
apply_reseal(dataclasses.replace(plan, path=tmp_path / "l.json"))
|
||||||
|
message = str(exc.value)
|
||||||
|
for band in plan.newly_licensed:
|
||||||
|
assert band in message
|
||||||
|
assert "entailed=" in message
|
||||||
|
assert "--allow-new-licenses" in message
|
||||||
|
assert "NON-COMMITMENTS" in message
|
||||||
|
|
||||||
|
|
||||||
|
def test_apply_with_explicit_authorization_writes_a_verifying_artifact(
|
||||||
|
plan: ResealPlan, tmp_path: Path
|
||||||
|
) -> None:
|
||||||
|
"""The grant path works — into tmp, never into `chat/data/`."""
|
||||||
|
target = tmp_path / "curriculum_serve_ledger.json"
|
||||||
|
written = apply_reseal(
|
||||||
|
dataclasses.replace(plan, path=target), allow_new_licenses=True
|
||||||
|
)
|
||||||
|
assert written == target
|
||||||
|
tallies = load_sealed_ledger(target)
|
||||||
|
assert len(tallies) == 11
|
||||||
|
assert all(t.wrong == 0 for t in tallies.values())
|
||||||
|
artifact = json.loads(target.read_text(encoding="utf-8"))
|
||||||
|
assert artifact["provenance"] == "evals.curriculum_serve.practice.runner.seal_ledger"
|
||||||
|
|
||||||
|
|
||||||
|
def test_a_reseal_that_only_revokes_needs_no_authorization(tmp_path: Path) -> None:
|
||||||
|
"""Losing a license is the gate becoming MORE conservative.
|
||||||
|
|
||||||
|
Only granting is gated. A producer change that drops a band below the floor
|
||||||
|
must be applicable without an authorization flag, or the safe direction
|
||||||
|
becomes the hard one.
|
||||||
|
"""
|
||||||
|
plan = ResealPlan(
|
||||||
|
capability="curriculum_serve",
|
||||||
|
path=tmp_path / "l.json",
|
||||||
|
licensed_before=("band_a", "band_b"),
|
||||||
|
licensed_after=("band_a",),
|
||||||
|
committed={"band_a": 700, "band_b": 3},
|
||||||
|
mix={"band_a": {"entailed": 700}, "band_b": {"entailed": 3}},
|
||||||
|
artifact={"schema": "x", "classes": {}, "content_sha256": "y"},
|
||||||
|
)
|
||||||
|
assert plan.newly_licensed == ()
|
||||||
|
assert plan.revoked == ("band_b",)
|
||||||
|
assert plan.is_housekeeping is True
|
||||||
|
assert apply_reseal(plan) == plan.path # no flag needed
|
||||||
|
assert plan.path.exists()
|
||||||
|
|
||||||
|
|
||||||
|
def test_a_pure_housekeeping_reseal_is_allowed(tmp_path: Path) -> None:
|
||||||
|
"""Identical before/after — the ordinary case once volume is settled."""
|
||||||
|
plan = ResealPlan(
|
||||||
|
capability="curriculum_serve",
|
||||||
|
path=tmp_path / "l.json",
|
||||||
|
licensed_before=("band_a",),
|
||||||
|
licensed_after=("band_a",),
|
||||||
|
committed={"band_a": 700},
|
||||||
|
mix={"band_a": {"entailed": 700}},
|
||||||
|
artifact={"schema": "x", "classes": {}, "content_sha256": "y"},
|
||||||
|
)
|
||||||
|
assert plan.is_housekeeping is True
|
||||||
|
apply_reseal(plan)
|
||||||
|
assert plan.path.exists()
|
||||||
|
|
||||||
|
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
# 4. A tampered ledger on disk refuses rather than being silently overwritten.
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
|
||||||
|
def test_tampered_current_ledger_refuses_instead_of_being_overwritten(
|
||||||
|
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
||||||
|
) -> None:
|
||||||
|
"""Overwriting a hand-edited ledger would destroy the evidence of tampering.
|
||||||
|
|
||||||
|
`plan_reseal` reads the current ledger through the SAME verifier the serving
|
||||||
|
path uses, so a broken `content_sha256` surfaces here.
|
||||||
|
"""
|
||||||
|
target = tmp_path / "curriculum_serve_ledger.json"
|
||||||
|
target.write_text(
|
||||||
|
json.dumps(
|
||||||
|
{
|
||||||
|
"schema": "curriculum_serve_ledger_v1",
|
||||||
|
"classes": {"curriculum_physics_causal": {"correct": 999, "wrong": 0}},
|
||||||
|
"content_sha256": "0" * 64,
|
||||||
|
"note": "hand-edited",
|
||||||
|
"provenance": "not-sealed-practice",
|
||||||
|
},
|
||||||
|
indent=2,
|
||||||
|
),
|
||||||
|
encoding="utf-8",
|
||||||
|
)
|
||||||
|
import teaching.ledger_reseal as mod
|
||||||
|
|
||||||
|
monkeypatch.setitem(
|
||||||
|
mod.CAPABILITY_LEDGERS,
|
||||||
|
"curriculum_serve",
|
||||||
|
LedgerSpec(
|
||||||
|
capability="curriculum_serve", path=target, missing_ok=True, note="test"
|
||||||
|
),
|
||||||
|
)
|
||||||
|
with pytest.raises(Exception, match="content_sha256 mismatch"):
|
||||||
|
plan_reseal("curriculum_serve")
|
||||||
|
# The tampered bytes are still there to be investigated.
|
||||||
|
assert "hand-edited" in target.read_text(encoding="utf-8")
|
||||||
|
|
||||||
|
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
# 5. The CLI verb.
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
|
||||||
|
def _run_cli(*argv: str) -> tuple[int, str]:
|
||||||
|
import argparse
|
||||||
|
import io
|
||||||
|
import contextlib
|
||||||
|
|
||||||
|
from core.cli_proposal_queue import register
|
||||||
|
|
||||||
|
parser = argparse.ArgumentParser()
|
||||||
|
register(parser.add_subparsers(dest="command", required=True))
|
||||||
|
args = parser.parse_args(["proposal-queue", *argv])
|
||||||
|
buf = io.StringIO()
|
||||||
|
with contextlib.redirect_stdout(buf):
|
||||||
|
code = args.func(args)
|
||||||
|
return code, buf.getvalue()
|
||||||
|
|
||||||
|
|
||||||
|
def test_cli_dry_run_reports_without_writing() -> None:
|
||||||
|
code, out = _run_cli("reseal", "curriculum_serve", "--dry-run")
|
||||||
|
assert code == 0
|
||||||
|
assert "nothing written" in out
|
||||||
|
assert "NEWLY LICENSED" in out
|
||||||
|
assert not CAPABILITY_LEDGERS["curriculum_serve"].path.exists()
|
||||||
|
|
||||||
|
|
||||||
|
def test_cli_refuses_without_authorization() -> None:
|
||||||
|
code, out = _run_cli("reseal", "curriculum_serve")
|
||||||
|
assert code == 1
|
||||||
|
assert "REFUSED" in out
|
||||||
|
assert not CAPABILITY_LEDGERS["curriculum_serve"].path.exists()
|
||||||
|
|
||||||
|
|
||||||
|
def test_cli_json_is_machine_readable() -> None:
|
||||||
|
code, out = _run_cli("reseal", "curriculum_serve", "--dry-run", "--json")
|
||||||
|
assert code == 0
|
||||||
|
payload = json.loads(out)
|
||||||
|
assert payload["written"] is False
|
||||||
|
assert payload["is_housekeeping"] is False
|
||||||
|
assert len(payload["newly_licensed"]) == 4
|
||||||
|
assert payload["licensed_before"] == []
|
||||||
|
|
||||||
|
|
||||||
|
def test_reseal_verb_is_registered_under_proposal_queue() -> None:
|
||||||
|
"""The verb has to be reachable from `core proposal-queue`, not only importable."""
|
||||||
|
import argparse
|
||||||
|
|
||||||
|
from core.cli_proposal_queue import register
|
||||||
|
|
||||||
|
parser = argparse.ArgumentParser()
|
||||||
|
register(parser.add_subparsers(dest="command", required=True))
|
||||||
|
args = parser.parse_args(
|
||||||
|
["proposal-queue", "reseal", "curriculum_serve", "--dry-run"]
|
||||||
|
)
|
||||||
|
assert args.proposal_queue_command == "reseal"
|
||||||
|
assert args.capability == "curriculum_serve"
|
||||||
|
assert args.dry_run is True
|
||||||
|
assert args.allow_new_licenses is False
|
||||||
|
|
||||||
|
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
# 6. The receipt now names the command that performs its pending stage.
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
|
||||||
|
def test_ratify_output_names_the_reseal_command() -> None:
|
||||||
|
"""Naming a pending stage without naming its command is how `ledger_reseal`
|
||||||
|
stayed unperformed since the ceremony was written.
|
||||||
|
|
||||||
|
Asserted at SOURCE level deliberately. Exercising the branch needs a
|
||||||
|
successful non-dry-run ratification, which appends to the real committed
|
||||||
|
corpus — the test would either mutate `teaching/domain_chains/` or stub out
|
||||||
|
the very ceremony it claims to check. A source pin is the honest instrument
|
||||||
|
here, and it is stated as one rather than dressed up as a behavioural test.
|
||||||
|
"""
|
||||||
|
import inspect
|
||||||
|
|
||||||
|
from teaching.ratification import RatificationReceipt
|
||||||
|
|
||||||
|
import core.cli_proposal_queue as cli
|
||||||
|
|
||||||
|
assert "ledger_reseal" in RatificationReceipt.__dataclass_fields__[
|
||||||
|
"pending_stages"
|
||||||
|
].default
|
||||||
|
|
||||||
|
source = inspect.getsource(cli.cmd_proposal_queue_ratify)
|
||||||
|
assert 'if "ledger_reseal" in receipt.pending_stages' in source
|
||||||
|
assert "core proposal-queue reseal curriculum_serve" in source
|
||||||
|
|
||||||
|
|
||||||
|
def test_reseal_stage_is_not_reachable_from_ratify_chain() -> None:
|
||||||
|
"""Doctrine: no ratification automation. `ratify_chain` must not seal."""
|
||||||
|
import inspect
|
||||||
|
|
||||||
|
import teaching.ratification as ratification
|
||||||
|
|
||||||
|
source = inspect.getsource(ratification)
|
||||||
|
for forbidden in ("ledger_reseal", "seal_ledger", "write_sealed_ledger"):
|
||||||
|
assert f"import {forbidden}" not in source
|
||||||
|
assert f"{forbidden}(" not in source
|
||||||
|
tally = ClassTally("x", correct=1)
|
||||||
|
assert tally.committed == 1 # sanity: the gate module is importable here
|
||||||
Loading…
Reference in a new issue