core/teaching/math_composition_ratification.py

397 lines
14 KiB
Python

"""ADR-0169 — CompositionClaim ratification into reviewed math composition mappings.
This module is the explicit post-review mutation boundary for math-domain
composition-pattern evidence. It edits only per-category source files
under ``language_packs/data/en_core_math_v1/compositions/``; it does not
regenerate the compiled lexicon and therefore does not rewrite the pack
manifest checksum.
Mirrors :mod:`teaching.math_frame_ratification` (F1) but lifts the safe
surface from bounded frame openers to allowlisted composition categories
over already-bound slot patterns (ADR-0169 §"Definition of a CompositionClaim").
Hard rules (ADR-0169 §"Decision"):
- ``SAFE_COMPOSITION_CATEGORIES`` is the only sanctioned surface — no other
categories may be ratified through this handler.
- case 0050 hazard pin: after any composition ratification, GSM8K
train-sample case 0050 must still refuse (no speculative pre-frame-filler
admission).
- evidence pointers MUST carry ``source="math_audit"`` — never ``"corpus"``.
- ``polarity in {"affirms", "falsifies"}``; the falsifies branch records the
pattern as non-composing and never appends an affirmative composition row.
"""
from __future__ import annotations
import hashlib
import json
import re
import string
from dataclasses import dataclass
from datetime import date
from pathlib import Path
from typing import Final
from teaching.math_evidence import MathReaderRefusalEvidence, from_audit_row
# ---------------------------------------------------------------------------
# Allowlist + constants
# ---------------------------------------------------------------------------
SAFE_COMPOSITION_CATEGORIES: Final[frozenset[str]] = frozenset(
{
"multiplicative_composition",
"additive_composition",
"subtractive_composition",
}
)
_VALID_POLARITIES: Final[frozenset[str]] = frozenset({"affirms", "falsifies"})
_REVIEWER_RE: Final[re.Pattern[str]] = re.compile(r"[^a-z0-9_-]+")
_TRIM_PUNCTUATION: Final[str] = string.punctuation
# ---------------------------------------------------------------------------
# Error hierarchy (mirror F1)
# ---------------------------------------------------------------------------
class RatificationError(ValueError):
"""Base class for composition ratification rejections."""
class WrongClaimSubType(RatificationError):
"""Raised when a non-composition evidence record reaches this handler."""
class WrongCompositionCategory(RatificationError):
"""Raised when ``composition_category`` is outside :data:`SAFE_COMPOSITION_CATEGORIES`.
Per ADR-0169.1 §"Trip-wires" #8, this is the explicit allowlist-enforcement
exception. Distinct from :class:`WrongZeroViolationCandidate` to give
operators a clearer remediation signal (the category is the problem; the
claim shape itself may be fine).
"""
class WrongZeroViolationCandidate(RatificationError):
"""Raised when a ratification could open a wrong>0 admission path."""
class AlreadyRatified(RatificationError):
"""Raised when the target claim is already covered by the source file."""
class EvidenceTampering(RatificationError):
"""Raised when the evidence hash no longer matches canonical bytes."""
class UnknownCategory(RatificationError):
"""Raised when the requested category is not a known composition source."""
class InvalidPolarity(RatificationError):
"""Raised when polarity is not in :data:`_VALID_POLARITIES`."""
class EvidenceLaundering(RatificationError):
"""Raised when audit evidence is presented as cognition corpus evidence.
ADR-0169.1 §"Evidence floor": math-audit evidence must never be
serialised with ``source="corpus"``; that would impersonate ADR-0057's
cognition corpus evidence floor.
"""
# ---------------------------------------------------------------------------
# Receipt
# ---------------------------------------------------------------------------
@dataclass(frozen=True, slots=True)
class CompositionRatificationReceipt:
target_file: str
surface_pattern: str
composition_category: str
polarity: str
provenance: str
file_sha256_before: str
file_sha256_after: str
evidence_hash: str
is_duplicate_evidence: bool
# ---------------------------------------------------------------------------
# Path / file helpers
# ---------------------------------------------------------------------------
def _repo_root() -> Path:
here = Path(__file__).resolve()
for parent in here.parents:
if (parent / "pyproject.toml").exists() or (parent / "setup.cfg").exists():
return parent
return here.parents[1]
def _default_pack_root() -> Path:
return _repo_root() / "language_packs" / "data" / "en_core_math_v1"
def _sha256_file(path: Path) -> str:
return hashlib.sha256(path.read_bytes()).hexdigest()
def _normalise_surface(surface: str) -> str:
return surface.lower().strip(_TRIM_PUNCTUATION)
def _reviewer_slug(reviewer: str) -> str:
slug = _REVIEWER_RE.sub("_", reviewer.lower()).strip("_-")
if not slug:
raise RatificationError("reviewer must contain at least one safe character")
return slug
def _read_entries(path: Path) -> list[dict[str, object]]:
entries: list[dict[str, object]] = []
for line_number, line in enumerate(
path.read_text(encoding="utf-8").splitlines(), start=1
):
if not line.strip():
continue
record = json.loads(line)
if not isinstance(record.get("surface_pattern"), str):
raise RatificationError(
f"{path}: line {line_number} missing string surface_pattern"
)
if not isinstance(record.get("composition_category"), str):
raise RatificationError(
f"{path}: line {line_number} missing string composition_category"
)
if not isinstance(record.get("polarity"), str):
raise RatificationError(
f"{path}: line {line_number} missing string polarity"
)
entries.append(record)
return entries
def _write_entries(path: Path, entries: list[dict[str, object]]) -> None:
ordered = sorted(entries, key=lambda item: str(item["surface_pattern"]))
lines = [
json.dumps(entry, ensure_ascii=False, sort_keys=False) for entry in ordered
]
path.write_text("\n".join(lines) + "\n", encoding="utf-8")
# ---------------------------------------------------------------------------
# Evidence validation
# ---------------------------------------------------------------------------
def _validate_evidence(claim: MathReaderRefusalEvidence) -> None:
"""Verify sub_type and tamper-resistance of evidence record."""
if claim.sub_type != "composition":
raise WrongClaimSubType(
f"Composition ratification requires sub_type='composition'; got {claim.sub_type!r}"
)
recomputed = from_audit_row(
claim.audit_row,
claim.sub_type,
claim_signature=claim.claim_signature,
)
if recomputed.evidence_hash != claim.evidence_hash:
raise EvidenceTampering(
"MathReaderRefusalEvidence hash does not match canonical audit row bytes"
)
def _check_no_corpus_laundering(evidence_source: str) -> None:
"""ADR-0169.1 §'Evidence floor' — audit evidence must declare math_audit.
Fail loudly if a caller hands us an evidence pointer that claims to be
cognition corpus evidence; that would impersonate the ADR-0057 floor.
"""
if evidence_source == "corpus":
raise EvidenceLaundering(
"audit evidence MUST NOT be laundered as source='corpus'; "
"use source='math_audit' (ADR-0169.1 §'Evidence floor')"
)
# ---------------------------------------------------------------------------
# Apply
# ---------------------------------------------------------------------------
def apply_composition_claim(
*,
claim: MathReaderRefusalEvidence,
composition_category: str,
polarity: str,
reviewer: str,
surface_pattern: str | None = None,
pack_root: Path | None = None,
evidence_source: str = "math_audit",
ratifier_kind: str = "cli",
) -> CompositionRatificationReceipt:
"""Apply a reviewed composition claim to a math composition source file.
Preconditions:
- ``claim.sub_type == "composition"`` (F1 analog of LexicalClaim sub_type pin).
- ``composition_category`` ∈ :data:`SAFE_COMPOSITION_CATEGORIES`.
- ``polarity`` ∈ ``{"affirms", "falsifies"}``.
- ``evidence_source == "math_audit"``; ``"corpus"`` raises
:class:`EvidenceLaundering` per ADR-0169.1.
- ``claim.evidence_hash`` matches a recomputation from ``claim.audit_row``.
The write target is
``{pack_root}/compositions/{composition_category}.jsonl``. Entries are
sorted alphabetically by ``surface_pattern``. Each entry records
``(surface_pattern, composition_category, polarity, provenance,
evidence_hashes)`` so the polarity decision is auditable and
falsification can supersede an earlier affirmation only through an
explicit reviewed re-apply.
``surface_pattern`` defaults to the normalized audit-row token when not
supplied — useful for tests and minimal first ratifications. Production
callers should pass an explicit bound-slot pattern (see ADR-0169
§"Definition of a CompositionClaim").
The compiled ``lexicon.jsonl`` and ``manifest.json`` are intentionally
not regenerated.
"""
# Trust-boundary checks (ADR-0169.1 §'Evidence floor' before any I/O).
_check_no_corpus_laundering(evidence_source)
if evidence_source != "math_audit":
raise EvidenceLaundering(
f"evidence_source must be 'math_audit'; got {evidence_source!r}"
)
if polarity not in _VALID_POLARITIES:
raise InvalidPolarity(
f"polarity must be one of {sorted(_VALID_POLARITIES)!r}; got {polarity!r}"
)
_validate_evidence(claim)
if composition_category not in SAFE_COMPOSITION_CATEGORIES:
raise WrongCompositionCategory(
f"composition_category {composition_category!r} is outside SAFE_COMPOSITION_CATEGORIES="
f"{sorted(SAFE_COMPOSITION_CATEGORIES)!r}; "
"CompositionClaim ratification is allowlist-only (ADR-0169 §'Initial safe category scope')"
)
root = (pack_root if pack_root is not None else _default_pack_root()).resolve()
source_dir = root / "compositions"
if not source_dir.exists():
raise UnknownCategory(
f"composition source directory does not exist: {source_dir}; "
"CompositionClaim handler requires a reviewed compositions/ tree"
)
target_file = source_dir / f"{composition_category}.jsonl"
if not target_file.exists():
# Initialise as empty source file rather than fail — the .gitkeep
# scaffold guarantees the parent directory; first ratification for a
# never-seen safe category seeds the file deterministically.
target_file.write_text("", encoding="utf-8")
if surface_pattern is None:
normalized_pattern = _normalise_surface(claim.audit_row.token_text)
else:
normalized_pattern = surface_pattern.lower().strip()
if not normalized_pattern:
raise WrongZeroViolationCandidate(
"empty composition surface_pattern cannot be ratified"
)
provenance = (
f"adr_0169_composition_ratified_{_reviewer_slug(reviewer)}_"
f"{date.today().isoformat()}"
)
before = _sha256_file(target_file)
entries = _read_entries(target_file)
target_relative = (
str(target_file.relative_to(_repo_root()))
if target_file.is_relative_to(_repo_root())
else str(target_file)
)
# Idempotency: identical (surface_pattern, composition_category, polarity,
# evidence_hash) is a no-op AlreadyRatified. Evidence-hash-only
# duplication (same claim ratified for a second time by *different*
# evidence) appends evidence per ADR-0169.1 §"Idempotency" path #1.
matching = [
e
for e in entries
if str(e["surface_pattern"]) == normalized_pattern
and str(e["composition_category"]) == composition_category
and str(e["polarity"]) == polarity
]
is_duplicate_evidence = False
if matching:
existing = matching[0]
existing_evidence: list[str] = list(existing.get("evidence_hashes", [])) # type: ignore[arg-type]
if claim.evidence_hash in existing_evidence:
raise AlreadyRatified(
f"composition claim ({normalized_pattern!r}, {composition_category!r}, "
f"{polarity!r}) is already ratified by evidence_hash={claim.evidence_hash}"
)
existing_evidence.append(claim.evidence_hash)
existing["evidence_hashes"] = sorted(set(existing_evidence))
existing["ratifier_kind"] = ratifier_kind
is_duplicate_evidence = True
else:
entries.append(
{
"surface_pattern": normalized_pattern,
"composition_category": composition_category,
"polarity": polarity,
"provenance": provenance,
"evidence_hashes": [claim.evidence_hash],
"ratifier_kind": ratifier_kind,
}
)
_write_entries(target_file, entries)
after = _sha256_file(target_file)
# RAT-1 — close the ratify→runtime gap: regenerate the compiled
# composition artifact and update the pack manifest's
# composition_checksum so the next runtime turn loads the new entry.
# Idempotent; identical source → identical compiled bytes.
from language_packs.compile_pack import compile_pack
compile_pack(root)
return CompositionRatificationReceipt(
target_file=target_relative,
surface_pattern=normalized_pattern,
composition_category=composition_category,
polarity=polarity,
provenance=provenance,
file_sha256_before=before,
file_sha256_after=after,
evidence_hash=claim.evidence_hash,
is_duplicate_evidence=is_duplicate_evidence,
)
__all__ = [
"AlreadyRatified",
"CompositionRatificationReceipt",
"EvidenceLaundering",
"EvidenceTampering",
"InvalidPolarity",
"RatificationError",
"SAFE_COMPOSITION_CATEGORIES",
"UnknownCategory",
"WrongClaimSubType",
"WrongCompositionCategory",
"WrongZeroViolationCandidate",
"apply_composition_claim",
]