"""teaching/ratification.py — the curriculum ratification ceremony. The discovery loop is instrumented but not closed. Proposals flow into ``teaching/proposals/`` and a human can read them via ``core proposal-queue review``, but `review_log.jsonl` explicitly *never* ratifies — it records that someone looked. Nothing converts a reviewed proposal into a ratified chain, so the loop terminates in a human decision that produces no artifact. That gap is the binding constraint. Curriculum serving is OFF because no band can earn a license from present volume (ADR-0262 §5) — every one of the eleven live bands is 24×–73× short of the entailed-bucket floor — and the only way volume grows is one ratified chain at a time. This module is the code-bound ceremony: **a reviewed decision produces a chain record, a corpus commit, and a receipt that proves the chain is actually routable.** Why the last clause is the whole design --------------------------------------- ``teaching.curriculum_premises._ratified_rows`` DROPS every row it cannot admit — wrong ``review_status``, wrong domain, empty terms, a term absent from its declared pack, a connective outside ``CONNECTIVE_FAMILY``. It drops silently, by ``continue``, because at *serving* time that is right: a curriculum must never route on a row it cannot license. At *ratification* time the same silence is a trap. Appending a row that the loader will drop looks exactly like appending a row that works: the file grows, the commit lands, the band count does not move, and nobody learns why for weeks. So this ceremony refuses to call an append a ratification until it has re-read the curriculum through the real loader and observed the chain arrive. Validation is a pre-flight courtesy; **admission is the proof**. Scope, deliberately narrow -------------------------- Nothing here decides *whether* a proposal is true. It cannot auto-ratify, has no opinion on content, and requires a human-supplied reviewer identity and rationale on every call. It converts an already-made decision into the artifacts that decision implies — which is precisely what was missing. The two downstream stages of the full ceremony, arena queue entry and ledger reseal, are intentionally NOT performed here. Both are sealed-practice outputs, and bridge rule 1 is that nothing outside a practice run may write a ledger (``core.ratified_ledger``). :func:`ratify_chain` returns the receipt those stages consume; running them stays an explicit, separate operator step. """ from __future__ import annotations import json import re from dataclasses import dataclass, field from pathlib import Path from typing import Any from core.capability.domains import ( DOMAIN_CAPABILITY_CORPORA, DOMAIN_CORPORA, DOMAIN_PACKS, ) from teaching.curriculum_premises import CONNECTIVE_FAMILY, load_curriculum _REPO_ROOT = Path(__file__).resolve().parents[1] #: Chain ids are ``--``, zero-padded to three. _CHAIN_ID_RE = re.compile(r"^(?P[a-z_]+)-(?P[a-z]+)-(?P\d{3,})$") #: The committed row's field order. Chain corpora are reviewed as diffs, so the #: key order is part of the artifact, not an implementation detail. _ROW_FIELDS: tuple[str, ...] = ( "chain_id", "domain", "operator_family", "subject", "intent", "connective", "object", "subject_pack_id", "object_pack_id", "review_status", "provenance", ) class RatificationError(ValueError): """A proposed chain cannot be ratified, with the reason stated.""" @dataclass(frozen=True, slots=True) class ChainRecord: """One curriculum edge, in the committed on-disk shape.""" chain_id: str domain: str operator_family: str subject: str intent: str connective: str object: str subject_pack_id: str object_pack_id: str review_status: str provenance: str def as_row(self) -> dict[str, Any]: return {name: getattr(self, name) for name in _ROW_FIELDS} def as_jsonl_line(self) -> str: """Byte-compatible with the committed corpora: compact separators, insertion order preserved, one trailing newline.""" return json.dumps(self.as_row(), separators=(",", ":")) + "\n" @dataclass(frozen=True, slots=True) class RatificationReceipt: """Proof that a ratification actually changed what the engine can route.""" chain: ChainRecord corpus_id: str corpus_path: str chains_before: int chains_after: int family_chains_before: int family_chains_after: int #: Follow-on stages this receipt authorizes but does NOT perform. pending_stages: tuple[str, ...] = field( default=("arena_queue_entry", "ledger_reseal") ) @property def admitted(self) -> bool: return self.chains_after == self.chains_before + 1 def as_dict(self) -> dict[str, Any]: return { "chain": self.chain.as_row(), "corpus_id": self.corpus_id, "corpus_path": self.corpus_path, "chains_before": self.chains_before, "chains_after": self.chains_after, "family_chains_before": self.family_chains_before, "family_chains_after": self.family_chains_after, "admitted": self.admitted, "pending_stages": list(self.pending_stages), } def _pack_lemmas(pack_id: str) -> frozenset[str]: """Mirror of the loader's pack reader, so validation sees what it sees.""" path = _REPO_ROOT / "packs" / "data" / pack_id / "lexicon.jsonl" if not path.exists(): return frozenset() lemmas: set[str] = set() for line in path.read_text(encoding="utf-8").splitlines(): if not line.strip(): continue try: row = json.loads(line) except json.JSONDecodeError: continue if isinstance(row, dict): lemma = str(row.get("lemma") or "").strip() if lemma: lemmas.add(lemma) return frozenset(lemmas) def corpus_path_for(domain: str) -> tuple[str, Path]: """The single writable chain corpus for *domain*. A domain with several corpora (``philosophy_theology``) has no unambiguous append target, so the ceremony refuses rather than guessing which file a reviewer meant. """ corpora = DOMAIN_CORPORA.get(domain, ()) if not corpora: raise RatificationError(f"unknown domain: {domain!r}") writable = [c for c in corpora if c in DOMAIN_CAPABILITY_CORPORA] if len(writable) != 1: raise RatificationError( f"{domain}: expected exactly one writable chain corpus, found " f"{writable!r} — name the target explicitly before ratifying" ) corpus_id = writable[0] return corpus_id, _REPO_ROOT / DOMAIN_CAPABILITY_CORPORA[corpus_id] def existing_rows(domain: str) -> list[dict[str, Any]]: """Every row in *domain*'s corpus, admitted or not (raw file read).""" _, path = corpus_path_for(domain) if not path.exists(): return [] rows: list[dict[str, Any]] = [] for line in path.read_text(encoding="utf-8").splitlines(): if not line.strip(): continue try: row = json.loads(line) except json.JSONDecodeError: continue if isinstance(row, dict): rows.append(row) return rows def next_chain_id(domain: str, family: str) -> str: """The next free ``--NNN``, deterministic from the corpus.""" highest = 0 for row in existing_rows(domain): m = _CHAIN_ID_RE.match(str(row.get("chain_id") or "")) if m and m.group("domain") == domain and m.group("family") == family: highest = max(highest, int(m.group("seq"))) return f"{domain}-{family}-{highest + 1:03d}" def build_chain_record( *, domain: str, subject: str, connective: str, obj: str, reviewer: str, rationale: str, intent: str = "cause", chain_id: str | None = None, ) -> ChainRecord: """Construct the record a reviewed decision implies. ``reviewer`` and ``rationale`` are required and land in ``provenance``: a ratified chain must carry who ratified it and why, or the corpus loses the audit lineage that makes it reviewable as a diff. """ reviewer = reviewer.strip() rationale = rationale.strip() if not reviewer: raise RatificationError("ratification requires a reviewer identity") if not rationale: raise RatificationError("ratification requires a stated rationale") family = CONNECTIVE_FAMILY.get(connective.strip()) if family is None: raise RatificationError( f"connective {connective!r} is outside CONNECTIVE_FAMILY " f"{sorted(CONNECTIVE_FAMILY)} — the loader would drop this row" ) packs = DOMAIN_PACKS.get(domain, ()) if not packs: raise RatificationError(f"domain {domain!r} mounts no packs") subject, obj = subject.strip(), obj.strip() subject_pack = next((p for p in packs if subject in _pack_lemmas(p)), "") object_pack = next((p for p in packs if obj in _pack_lemmas(p)), "") if not subject_pack: raise RatificationError( f"subject {subject!r} is taught by no pack mounted for {domain} " f"({list(packs)}) — the loader would drop this row" ) if not object_pack: raise RatificationError( f"object {obj!r} is taught by no pack mounted for {domain} " f"({list(packs)}) — the loader would drop this row" ) return ChainRecord( chain_id=chain_id or next_chain_id(domain, family), domain=domain, operator_family=family, subject=subject, intent=intent, connective=connective.strip(), object=obj, subject_pack_id=subject_pack, object_pack_id=object_pack, review_status="reviewed", provenance=f"ratification-ceremony:{reviewer}:{rationale}", ) def validate_admissible(record: ChainRecord) -> None: """Pre-flight: every rule ``_ratified_rows`` drops on, raised LOUDLY. Serving is right to drop silently. Ratification is not: an append the loader will discard looks identical to one that works. """ if record.review_status != "reviewed": raise RatificationError( f"review_status={record.review_status!r} — only 'reviewed' is admitted" ) if not record.subject or not record.object: raise RatificationError("subject and object must both be non-empty") if CONNECTIVE_FAMILY.get(record.connective) is None: raise RatificationError(f"connective {record.connective!r} has no family") if record.subject_pack_id and record.subject not in _pack_lemmas(record.subject_pack_id): raise RatificationError( f"subject {record.subject!r} absent from pack {record.subject_pack_id}" ) if record.object_pack_id and record.object not in _pack_lemmas(record.object_pack_id): raise RatificationError( f"object {record.object!r} absent from pack {record.object_pack_id}" ) for row in existing_rows(record.domain): if str(row.get("chain_id")) == record.chain_id: raise RatificationError(f"chain_id {record.chain_id} already committed") if ( str(row.get("subject")) == record.subject and str(row.get("connective")) == record.connective and str(row.get("object")) == record.object ): raise RatificationError( f"duplicate edge: {row.get('chain_id')} already teaches " f"{record.subject} {record.connective} {record.object}" ) def _count_chains(domain: str, family: str) -> tuple[int, int]: load_curriculum.cache_clear() curriculum = load_curriculum(domain) return len(curriculum.chains), sum( 1 for c in curriculum.chains if c.family == family ) def ratify_chain(record: ChainRecord, *, dry_run: bool = False) -> RatificationReceipt: """The ceremony: validate → append → **confirm admission** → receipt. Raises :class:`RatificationError` if the appended row does not arrive in the reloaded curriculum. On that failure the append is rolled back, because a corpus carrying rows the engine silently ignores is worse than one that never grew: it makes the volume ledger lie. """ validate_admissible(record) corpus_id, path = corpus_path_for(record.domain) before_total, before_family = _count_chains(record.domain, record.operator_family) if dry_run: return RatificationReceipt( chain=record, corpus_id=corpus_id, corpus_path=str(path), chains_before=before_total, chains_after=before_total + 1, family_chains_before=before_family, family_chains_after=before_family + 1, ) original = path.read_text(encoding="utf-8") if path.exists() else "" if original and not original.endswith("\n"): original += "\n" path.parent.mkdir(parents=True, exist_ok=True) path.write_text(original + record.as_jsonl_line(), encoding="utf-8") after_total, after_family = _count_chains(record.domain, record.operator_family) if after_total != before_total + 1: path.write_text(original, encoding="utf-8") _count_chains(record.domain, record.operator_family) raise RatificationError( f"{record.chain_id} was appended but the curriculum loader did not " f"admit it ({before_total} -> {after_total}); append rolled back" ) return RatificationReceipt( chain=record, corpus_id=corpus_id, corpus_path=str(path), chains_before=before_total, chains_after=after_total, family_chains_before=before_family, family_chains_after=after_family, ) __all__ = [ "ChainRecord", "RatificationError", "RatificationReceipt", "build_chain_record", "corpus_path_for", "existing_rows", "next_chain_id", "ratify_chain", "validate_admissible", ]