feat(ADR-0168/F1): FrameClaim ratification handler (Tier 1.5) (#389)

Implements the FrameClaim ratification handler per ADR-0168 doctrine
and the ADR-0168.1 MathFrameClaimProposal adapter.  Mirrors the
ADR-0167 W2-D LexicalClaim template (apply_lexical_claim) but lifts
the safe surface from drain-class lexical entries to allowlisted
frame-opening categories — the next sub-type up the wrong=0 hazard
ladder.

teaching/math_frame_proposal.py
  + MathFrameClaimProposal dataclass (ADR-0168.1 §"Data shape")
  + MathReaderRefusalEvidencePointer with source pinned to "math_audit"
  + build_evidence_pointer() — only sanctioned constructor; rejects None
    missing_operator
  + build_frame_claim_proposal() — enforces:
      • surface_form non-empty after normalization
      • frame_category in the ADR-0168 allowlist
      • polarity in {affirms, falsifies}
      • ≥1 evidence pointer with source="math_audit"
      • source="corpus" rejected as schema-illegal (ADR-0168.1
        §"Evidence floor")
  + compute_claim_signature() / compute_proposal_id() / canonical_bytes()
    — deterministic identity per ADR-0168 §"Replay obligations" #1

teaching/math_frame_ratification.py
  + SAFE_FRAME_CATEGORIES = {increment_frame, decrement_frame,
    transfer_frame, remainder_frame} — no other categories
  + Error hierarchy: RatificationError, WrongClaimSubType,
    WrongZeroViolationCandidate, AlreadyRatified, EvidenceTampering,
    UnknownCategory, InvalidPolarity, EvidenceLaundering
  + FrameRatificationReceipt dataclass with before/after SHA + evidence_hash
  + apply_frame_claim(claim, frame_category, polarity, reviewer, pack_root,
    evidence_source="math_audit"):
      • rejects evidence_source != "math_audit" (ADR-0168.1 §"Evidence floor")
      • rejects polarity outside {affirms, falsifies}
      • rejects claim.sub_type != "frame"
      • rejects evidence_hash tampering (recomputes from audit_row)
      • rejects frame_category outside SAFE_FRAME_CATEGORIES
      • writes language_packs/data/en_core_math_v1/frames/{category}.jsonl
      • idempotent: same (surface, category, polarity, evidence_hash)
        raises AlreadyRatified
      • duplicate evidence appends evidence_hash to existing row
        (ADR-0168.1 §"Idempotency" path #1)
      • polarity=falsifies records non-opener; never appends to compiled
        lexicon or manifest

language_packs/data/en_core_math_v1/frames/.gitkeep
  Directory scaffold for the reviewed frame source files.

workbench/readers.py
  _HANDLER_DISPATCH gains "frame_reclassification" → "FrameClaim".
  GET /math-proposals/{id} detail and POST /math-proposals/{id}/ratify
  now return suggested_cli pointing at apply_frame_claim().

core/cli.py
  teaching test-suite tuple gains tests/test_math_frame_ratification.py.

tests/test_math_frame_ratification.py — 14 tests:
   1. SAFE_FRAME_CATEGORIES is exactly the ADR-0168 allowlist
   2. apply writes a frame entry for a safe category
   3. receipt records before/after sha + evidence_hash
   4. idempotent same-evidence → AlreadyRatified
   5. rejects non-frame sub_type → WrongClaimSubType
   6. rejects categories outside SAFE_FRAME_CATEGORIES → WrongZeroViolationCandidate
   7. rejects invalid polarity → InvalidPolarity
   8. rejects evidence_hash tampering → EvidenceTampering
   9. rejects source="corpus" → EvidenceLaundering (ADR-0168.1 §"Evidence floor")
  10. case 0050 hazard pin — after ratification, case 0050 still refuses
  11. polarity=falsifies branch records non-opener; affirms+falsifies coexist
  12. duplicate evidence appends evidence_hash, does not create a second row
  13. manifest.json checksum unchanged by frame ratification
  14. alphabetical sort by surface_form preserved across writes

Suite verification
  core test --suite teaching -q → 47 passed (was 33; +14 new)
  core test --suite runtime  -q → 20 passed
  tests/test_math_lexical_ratification.py → 15 passed (untouched, regression-clean)
  tests/test_adr_0172_w4_workbench_e2e.py → 7 passed (existing dispatch tests still hold)

Doctrine invariants preserved
  - wrong=0: case 0050 still refuses after ratification
  - replay equivalence: claim_signature and proposal_id are deterministic
    (sha256 of canonical identity, clock-time-independent)
  - refusal-first: no runtime mutation; handler is the only mutation
    boundary and writes only the reviewed frames/ source tree
  - ADR-0167 partition: math-audit evidence stays math-domain; corpus
    evidence is rejected loudly

Brief-correction note: the brief named the scaffold path
"packs/en_core_math_v1/frames/.gitkeep" but the existing math pack lives
at language_packs/data/en_core_math_v1/ (no top-level packs/en_core_math_v1
exists).  Scaffold placed at language_packs/data/en_core_math_v1/frames/
to mirror the existing lexicon/ source-tree convention; apply_frame_claim
defaults pack_root to that location.
This commit is contained in:
Shay 2026-05-27 14:10:43 -07:00 committed by GitHub
parent 73e52a128b
commit c6a9bb0096
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
6 changed files with 1185 additions and 0 deletions

View file

@ -67,6 +67,7 @@ _TEST_SUITES: dict[str, tuple[str, ...]] = {
"tests/test_epistemic_invariants.py",
"tests/test_adr_0172_w2_decomposer.py",
"tests/test_adr_0172_w5_inference_proposal.py",
"tests/test_math_frame_ratification.py",
),
"packs": (
"tests/test_core_semantic_seed_pack.py",

View file

@ -0,0 +1,342 @@
"""ADR-0168 / ADR-0168.1 — MathFrameClaimProposal adapter.
Defines the math-domain proposal type carrying FrameClaim evidence pointers.
Mirror of :mod:`teaching.math_contemplation_proposal` for the frame
admissibility sub-type.
Per ADR-0168.1 §"Evidence floor", a :class:`MathFrameClaimProposal` is
eligible only if every evidence pointer declares ``source="math_audit"``.
Audit evidence MUST NEVER be laundered as ``source="corpus"`` that would
falsely impersonate ADR-0057's cognition corpus evidence floor.
Trust boundary: schema-only module. No filesystem I/O, no teaching-store
writes, no runtime pipeline hooks. All validation lives in
:func:`build_frame_claim_proposal` and :func:`build_evidence_pointer`.
"""
from __future__ import annotations
import hashlib
import json
from dataclasses import dataclass, field
from typing import Any, Final, Literal
from teaching.math_evidence import MathReaderRefusalEvidence
# ---------------------------------------------------------------------------
# Constants
# ---------------------------------------------------------------------------
Polarity = Literal["affirms", "falsifies"]
_VALID_POLARITIES: Final[frozenset[str]] = frozenset({"affirms", "falsifies"})
# Mirrors SAFE_FRAME_CATEGORIES in :mod:`teaching.math_frame_ratification`.
# Held here as well so the proposal layer can reject illegal frame categories
# before the proposal is ever built — defense in depth at the schema boundary.
_ALLOWLISTED_FRAME_CATEGORIES: Final[frozenset[str]] = frozenset(
{
"increment_frame",
"decrement_frame",
"transfer_frame",
"remainder_frame",
}
)
# Forbidden evidence source values. ADR-0168.1 §"Evidence floor" requires
# math-audit evidence to carry ``source="math_audit"``. Any cognition-style
# value reaching this module indicates evidence laundering — fail loudly.
_FORBIDDEN_EVIDENCE_SOURCES: Final[frozenset[str]] = frozenset({"corpus"})
# ---------------------------------------------------------------------------
# Evidence pointer
# ---------------------------------------------------------------------------
@dataclass(frozen=True, slots=True)
class MathReaderRefusalEvidencePointer:
"""One pointer to a math-domain refusal-row evidence record.
Per ADR-0168.1, ``source`` is pinned to ``"math_audit"`` at construction.
The factory :func:`build_evidence_pointer` is the only sanctioned way to
create one direct instantiation works but is discouraged because the
factory ensures ``source`` cannot drift to a cognition value.
"""
source: Literal["math_audit"]
case_id: str
sentence_index: int
token_index: int
missing_operator: str
refusal_reason: str
evidence_hash: str
audit_row_digest: str
def _audit_row_digest(evidence: MathReaderRefusalEvidence) -> str:
"""SHA-256 over the evidence record's canonical bytes (excludes hash itself).
Provides the replay anchor: identical evidence yields an identical
digest across processes, runs, and reorderings.
"""
return hashlib.sha256(evidence.to_canonical_bytes()).hexdigest()
def build_evidence_pointer(
evidence: MathReaderRefusalEvidence,
) -> MathReaderRefusalEvidencePointer:
"""Build a math-audit evidence pointer from a refusal-row record.
Pins ``source="math_audit"`` per ADR-0168.1 §"Evidence floor".
"""
if evidence.missing_operator is None:
raise ValueError(
"MathFrameClaim evidence must declare a missing_operator; got None"
)
return MathReaderRefusalEvidencePointer(
source="math_audit",
case_id=evidence.case_id,
sentence_index=evidence.sentence_index,
token_index=evidence.token_index,
missing_operator=evidence.missing_operator,
refusal_reason=evidence.refusal_reason,
evidence_hash=evidence.evidence_hash,
audit_row_digest=_audit_row_digest(evidence),
)
# ---------------------------------------------------------------------------
# Proposal
# ---------------------------------------------------------------------------
@dataclass(frozen=True, slots=True)
class MathFrameClaimProposal:
"""A reviewed assertion that ``surface_form`` participates in ``frame_category``.
Per ADR-0168.1 the proposal carries:
- canonical identity (proposal_id, claim_signature)
- the structural claim (surface_form, frame_category, polarity)
- math-domain evidence pointers (source="math_audit" only)
- review state and operator note
The proposal is *not* a runtime frame registry entry. Operator
acceptance does not auto-mutate the runtime see
:mod:`teaching.math_frame_ratification` for the explicit mutation
boundary.
"""
proposal_id: str
claim_signature: str
surface_form: str
frame_category: str
polarity: Polarity
evidence: tuple[MathReaderRefusalEvidencePointer, ...]
review_state: Literal["pending", "accepted", "rejected", "withdrawn"]
operator_note: str = ""
domain: Literal["math"] = "math"
sub_type: Literal["frame"] = "frame"
def _serialise_pointer(p: MathReaderRefusalEvidencePointer) -> dict[str, Any]:
return {
"source": p.source,
"case_id": p.case_id,
"sentence_index": p.sentence_index,
"token_index": p.token_index,
"missing_operator": p.missing_operator,
"refusal_reason": p.refusal_reason,
"evidence_hash": p.evidence_hash,
"audit_row_digest": p.audit_row_digest,
}
def compute_claim_signature(
*,
surface_form: str,
frame_category: str,
polarity: str,
evidence: tuple[MathReaderRefusalEvidencePointer, ...],
) -> str:
"""Deterministic canonical signature for a frame claim.
Per ADR-0168 §"Replay obligations" #1, equivalent refusals must yield
identical signatures. We hash over:
- normalized surface form (lower, stripped)
- frame category (allowlisted string)
- polarity
- the sorted set of audit_row_digests from the evidence pointers
"""
surface_norm = surface_form.lower().strip()
digests = sorted(p.audit_row_digest for p in evidence)
payload = json.dumps(
{
"surface_form": surface_norm,
"frame_category": frame_category,
"polarity": polarity,
"audit_row_digests": digests,
},
ensure_ascii=False,
sort_keys=True,
separators=(",", ":"),
).encode("utf-8")
return hashlib.sha256(payload).hexdigest()
def compute_proposal_id(
*,
surface_form: str,
frame_category: str,
polarity: str,
evidence: tuple[MathReaderRefusalEvidencePointer, ...],
) -> str:
"""Stable proposal_id from canonical identity (ADR-0168.1 §"Idempotency").
``sha256(domain | subtype | surface_form | frame_category | polarity |
evidence_digest_set)`` clock-time-independent and identity-stable.
"""
surface_norm = surface_form.lower().strip()
digests = sorted(p.audit_row_digest for p in evidence)
payload = json.dumps(
{
"domain": "math",
"sub_type": "frame",
"surface_form": surface_norm,
"frame_category": frame_category,
"polarity": polarity,
"audit_row_digests": digests,
},
ensure_ascii=False,
sort_keys=True,
separators=(",", ":"),
).encode("utf-8")
return hashlib.sha256(payload).hexdigest()
def canonical_bytes(proposal: MathFrameClaimProposal) -> bytes:
"""Deterministic canonical bytes for replay / persistence.
Excludes ``proposal_id`` (the hash function input) and ``review_state``
(mutable transition surface).
"""
payload: dict[str, Any] = {
"domain": proposal.domain,
"sub_type": proposal.sub_type,
"claim_signature": proposal.claim_signature,
"surface_form": proposal.surface_form,
"frame_category": proposal.frame_category,
"polarity": proposal.polarity,
"evidence": sorted(
(_serialise_pointer(p) for p in proposal.evidence),
key=lambda d: d["audit_row_digest"],
),
"operator_note": proposal.operator_note,
}
return json.dumps(
payload, ensure_ascii=False, sort_keys=True, separators=(",", ":")
).encode("utf-8")
def build_frame_claim_proposal(
*,
surface_form: str,
frame_category: str,
polarity: str,
evidence: tuple[MathReaderRefusalEvidencePointer, ...],
operator_note: str = "",
review_state: Literal["pending", "accepted", "rejected", "withdrawn"] = "pending",
) -> MathFrameClaimProposal:
"""Build a :class:`MathFrameClaimProposal` with all invariants enforced.
Invariants (per ADR-0168 §"Decision" + ADR-0168.1 §"Evidence floor")
1. ``surface_form`` non-empty after normalization.
2. ``frame_category`` in the ADR-0168 allowlist (also enforced by the
handler, but rejected here too for defense-in-depth).
3. ``polarity in {"affirms", "falsifies"}``.
4. ``len(evidence) >= 1`` at least one audit/refusal evidence pointer.
5. every evidence pointer carries ``source="math_audit"`` corpus
evidence is rejected as schema-illegal.
6. proposal_id and claim_signature are derived deterministically.
Raises ``ValueError`` on any violation; the caller must fix the inputs.
"""
normalized = surface_form.lower().strip()
if not normalized:
raise ValueError(
f"surface_form must be non-empty after normalization; got {surface_form!r}"
)
if frame_category not in _ALLOWLISTED_FRAME_CATEGORIES:
raise ValueError(
f"frame_category {frame_category!r} is not in the ADR-0168 allowlist "
f"{sorted(_ALLOWLISTED_FRAME_CATEGORIES)!r}"
)
if polarity not in _VALID_POLARITIES:
raise ValueError(
f"polarity must be one of {sorted(_VALID_POLARITIES)!r}; got {polarity!r}"
)
if len(evidence) < 1:
raise ValueError(
"MathFrameClaimProposal requires at least one math-audit evidence pointer"
)
for idx, pointer in enumerate(evidence):
if pointer.source != "math_audit":
raise ValueError(
f"evidence[{idx}].source must be 'math_audit'; got {pointer.source!r}"
"audit evidence MUST NOT be laundered as cognition corpus evidence "
"(ADR-0168.1 §'Evidence floor')"
)
if pointer.source in _FORBIDDEN_EVIDENCE_SOURCES:
raise ValueError(
f"evidence[{idx}].source is forbidden: {pointer.source!r}"
)
claim_signature = compute_claim_signature(
surface_form=normalized,
frame_category=frame_category,
polarity=polarity,
evidence=evidence,
)
proposal_id = compute_proposal_id(
surface_form=normalized,
frame_category=frame_category,
polarity=polarity,
evidence=evidence,
)
return MathFrameClaimProposal(
proposal_id=proposal_id,
claim_signature=claim_signature,
surface_form=normalized,
frame_category=frame_category,
polarity=polarity, # type: ignore[arg-type]
evidence=tuple(evidence),
review_state=review_state,
operator_note=operator_note,
)
__all__ = [
"MathFrameClaimProposal",
"MathReaderRefusalEvidencePointer",
"Polarity",
"build_evidence_pointer",
"build_frame_claim_proposal",
"canonical_bytes",
"compute_claim_signature",
"compute_proposal_id",
]

View file

@ -0,0 +1,363 @@
"""ADR-0168 — FrameClaim ratification into reviewed math frame mappings.
This module is the explicit post-review mutation boundary for math-domain
frame-opener evidence. It edits only per-category source files under
``language_packs/data/en_core_math_v1/frames/``; it does not regenerate the
compiled lexicon and therefore does not rewrite the pack manifest checksum.
Mirrors :mod:`teaching.math_lexical_ratification` (W2-D) but lifts the safe
surface from drain-class lexical entries to allowlisted frame categories.
Hard rules (ADR-0168 §"Decision"):
- ``SAFE_FRAME_CATEGORIES`` is the only sanctioned surface no other
categories may be ratified through this handler.
- case 0050 hazard pin: after any frame 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
surface as a non-opener and never appends an opener lemma.
"""
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_FRAME_CATEGORIES: Final[frozenset[str]] = frozenset(
{
"increment_frame",
"decrement_frame",
"transfer_frame",
"remainder_frame",
}
)
_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 W2-D)
# ---------------------------------------------------------------------------
class RatificationError(ValueError):
"""Base class for frame ratification rejections."""
class WrongClaimSubType(RatificationError):
"""Raised when a non-frame evidence record reaches this handler."""
class WrongZeroViolationCandidate(RatificationError):
"""Raised when a ratification could open a wrong>0 admission path."""
class AlreadyRatified(RatificationError):
"""Raised when the target surface 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 frame 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-0168.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 FrameRatificationReceipt:
target_file: str
surface_form: str
frame_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_form"), str):
raise RatificationError(
f"{path}: line {line_number} missing string surface_form"
)
if not isinstance(record.get("frame_category"), str):
raise RatificationError(
f"{path}: line {line_number} missing string frame_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_form"]))
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 != "frame":
raise WrongClaimSubType(
f"Frame ratification requires sub_type='frame'; 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-0168.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-0168.1 §'Evidence floor')"
)
# ---------------------------------------------------------------------------
# Apply
# ---------------------------------------------------------------------------
def apply_frame_claim(
*,
claim: MathReaderRefusalEvidence,
frame_category: str,
polarity: str,
reviewer: str,
pack_root: Path | None = None,
evidence_source: str = "math_audit",
) -> FrameRatificationReceipt:
"""Apply a reviewed frame claim to a math frame source file.
Preconditions:
- ``claim.sub_type == "frame"`` (W2-D analog of LexicalClaim sub_type pin).
- ``frame_category`` :data:`SAFE_FRAME_CATEGORIES`.
- ``polarity`` ``{"affirms", "falsifies"}``.
- ``evidence_source == "math_audit"``; ``"corpus"`` raises
:class:`EvidenceLaundering` per ADR-0168.1.
- ``claim.evidence_hash`` matches a recomputation from ``claim.audit_row``.
The write target is ``{pack_root}/frames/{frame_category}.jsonl``.
Entries are sorted alphabetically by ``surface_form``. Each entry
records ``(surface_form, frame_category, polarity, provenance,
evidence_hash)`` so the polarity decision is auditable and
falsification can supersede an earlier affirmation only through an
explicit reviewed re-apply.
The compiled ``lexicon.jsonl`` and ``manifest.json`` are intentionally
not regenerated.
"""
# Trust-boundary checks (ADR-0168.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 frame_category not in SAFE_FRAME_CATEGORIES:
raise WrongZeroViolationCandidate(
f"frame_category {frame_category!r} is outside SAFE_FRAME_CATEGORIES="
f"{sorted(SAFE_FRAME_CATEGORIES)!r}; FrameClaim ratification is allowlist-only"
)
root = (pack_root if pack_root is not None else _default_pack_root()).resolve()
source_dir = root / "frames"
if not source_dir.exists():
raise UnknownCategory(
f"frame source directory does not exist: {source_dir}; "
"FrameClaim handler requires a reviewed frames/ tree"
)
target_file = source_dir / f"{frame_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")
surface_form = _normalise_surface(claim.audit_row.token_text)
if not surface_form:
raise WrongZeroViolationCandidate(
"empty frame surface cannot be ratified"
)
provenance = (
f"adr_0168_frame_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_form, frame_category, polarity,
# evidence_hash) is a no-op AlreadyRatified. Evidence-hash-only
# duplication (same surface ratified for a second time by *different*
# evidence) appends evidence per ADR-0168.1 §"Idempotency" path #1.
matching = [
e
for e in entries
if str(e["surface_form"]) == surface_form
and str(e["frame_category"]) == frame_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"frame claim ({surface_form!r}, {frame_category!r}, {polarity!r}) "
f"is already ratified by evidence_hash={claim.evidence_hash}"
)
existing_evidence.append(claim.evidence_hash)
existing["evidence_hashes"] = sorted(set(existing_evidence))
is_duplicate_evidence = True
else:
entries.append(
{
"surface_form": surface_form,
"frame_category": frame_category,
"polarity": polarity,
"provenance": provenance,
"evidence_hashes": [claim.evidence_hash],
}
)
_write_entries(target_file, entries)
after = _sha256_file(target_file)
return FrameRatificationReceipt(
target_file=target_relative,
surface_form=surface_form,
frame_category=frame_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",
"EvidenceLaundering",
"EvidenceTampering",
"FrameRatificationReceipt",
"InvalidPolarity",
"RatificationError",
"SAFE_FRAME_CATEGORIES",
"UnknownCategory",
"WrongClaimSubType",
"WrongZeroViolationCandidate",
"apply_frame_claim",
]

View file

@ -0,0 +1,465 @@
"""ADR-0168 — FrameClaim ratification handler tests.
Mirrors :mod:`tests.test_math_lexical_ratification` (W2-D) and pins:
1. SAFE_FRAME_CATEGORIES is exactly the ADR-0168 allowlist (4 entries).
2. apply_frame_claim writes a frame entry for a safe category.
3. receipt records before/after sha + evidence_hash.
4. idempotent same-evidence AlreadyRatified.
5. rejects non-frame sub_type.
6. rejects categories outside SAFE_FRAME_CATEGORIES (wrong=0 hazard).
7. rejects invalid polarity (must be affirms|falsifies).
8. rejects evidence tampering.
9. rejects evidence laundering (source='corpus' is forbidden).
10. case 0050 hazard pin after ratification, case 0050 still refuses.
11. polarity=falsifies branch records non-opener without admitting.
12. duplicate evidence on second apply appends evidence_hash, not new row.
13. manifest.json checksum is unchanged by frame ratification.
14. alphabetical sort by surface_form preserved across writes.
ADR-0168 §"Decision" + ADR-0168.1 §"Evidence floor" + hazard pins.
"""
from __future__ import annotations
import functools
import hashlib
import json
import shutil
from pathlib import Path
import pytest
from generate.comprehension import lexicon as comprehension_lexicon
from generate.comprehension import lifecycle
from generate.comprehension.audit import AuditRow, audit_problem
from generate.comprehension.state import ReaderRefusal
from teaching.math_evidence import from_audit_row
from teaching.math_frame_ratification import (
AlreadyRatified,
EvidenceLaundering,
EvidenceTampering,
InvalidPolarity,
SAFE_FRAME_CATEGORIES,
WrongClaimSubType,
WrongZeroViolationCandidate,
apply_frame_claim,
)
REPO_ROOT = Path(__file__).resolve().parents[1]
PACK_ROOT = REPO_ROOT / "language_packs" / "data" / "en_core_math_v1"
CASES_PATH = REPO_ROOT / "evals" / "gsm8k_math" / "train_sample" / "v1" / "cases.jsonl"
# ---------------------------------------------------------------------------
# Fixtures
# ---------------------------------------------------------------------------
@pytest.fixture()
def pack_copy(tmp_path: Path) -> Path:
target = tmp_path / "en_core_math_v1"
shutil.copytree(PACK_ROOT, target)
# Ensure the frames/ scaffold exists in the copy.
(target / "frames").mkdir(exist_ok=True)
comprehension_lexicon._CACHE.clear()
lifecycle._get_lexicon.cache_clear()
yield target
comprehension_lexicon._CACHE.clear()
lifecycle._get_lexicon.cache_clear()
def _row(
surface: str,
*,
missing_operator: str = "pre_frame_filler_sentence",
refusal_reason: str = "unexpected_category",
) -> AuditRow:
return AuditRow(
case_id=f"case-{surface}",
sentence_index=0,
token_index=2,
token_text=surface,
recognized_terms=("Mark", "does"),
skipped_frame=None,
missing_operator=missing_operator,
refusal_reason=refusal_reason,
refusal_detail=f"unexpected category for '{surface}'",
)
def _claim(surface: str, *, sub_type: str = "frame"):
return from_audit_row(_row(surface), sub_type) # type: ignore[arg-type]
def _entries(path: Path) -> list[dict]:
if not path.exists():
return []
return [
json.loads(line)
for line in path.read_text(encoding="utf-8").splitlines()
if line.strip()
]
def _manifest_sha(pack_root: Path) -> str:
return hashlib.sha256((pack_root / "manifest.json").read_bytes()).hexdigest()
def _use_pack_for_reader(monkeypatch: pytest.MonkeyPatch, pack_root: Path) -> None:
comprehension_lexicon._CACHE.clear()
@functools.cache
def _tmp_lexicon():
return comprehension_lexicon.load_lexicon(pack_root)
monkeypatch.setattr(lifecycle, "_get_lexicon", _tmp_lexicon)
# ---------------------------------------------------------------------------
# Test 1 — SAFE_FRAME_CATEGORIES is the ADR-0168 allowlist
# ---------------------------------------------------------------------------
def test_safe_frame_categories_is_adr_0168_allowlist() -> None:
"""The four ADR-0168 categories, no more, no less."""
assert SAFE_FRAME_CATEGORIES == frozenset(
{"increment_frame", "decrement_frame", "transfer_frame", "remainder_frame"}
)
# ---------------------------------------------------------------------------
# Test 2 — apply writes a frame entry for a safe category
# ---------------------------------------------------------------------------
def test_apply_frame_claim_writes_entry(pack_copy: Path) -> None:
receipt = apply_frame_claim(
claim=_claim("gives"),
frame_category="transfer_frame",
polarity="affirms",
reviewer="Ada",
pack_root=pack_copy,
)
assert receipt.surface_form == "gives"
assert receipt.frame_category == "transfer_frame"
assert receipt.polarity == "affirms"
assert receipt.is_duplicate_evidence is False
entry = next(
e
for e in _entries(pack_copy / "frames" / "transfer_frame.jsonl")
if e["surface_form"] == "gives"
)
assert entry["frame_category"] == "transfer_frame"
assert entry["polarity"] == "affirms"
assert receipt.evidence_hash in entry["evidence_hashes"]
# ---------------------------------------------------------------------------
# Test 3 — receipt records before/after sha
# ---------------------------------------------------------------------------
def test_receipt_records_before_after_sha(pack_copy: Path) -> None:
receipt = apply_frame_claim(
claim=_claim("gives"),
frame_category="transfer_frame",
polarity="affirms",
reviewer="Ada",
pack_root=pack_copy,
)
assert len(receipt.file_sha256_before) == 64
assert len(receipt.file_sha256_after) == 64
assert receipt.file_sha256_before != receipt.file_sha256_after
assert receipt.evidence_hash == _claim("gives").evidence_hash
# ---------------------------------------------------------------------------
# Test 4 — idempotent same-evidence raises AlreadyRatified
# ---------------------------------------------------------------------------
def test_idempotent_same_evidence_raises_already_ratified(pack_copy: Path) -> None:
claim = _claim("gives")
apply_frame_claim(
claim=claim,
frame_category="transfer_frame",
polarity="affirms",
reviewer="Ada",
pack_root=pack_copy,
)
with pytest.raises(AlreadyRatified, match="already ratified"):
apply_frame_claim(
claim=claim,
frame_category="transfer_frame",
polarity="affirms",
reviewer="Ada",
pack_root=pack_copy,
)
# ---------------------------------------------------------------------------
# Test 5 — rejects non-frame sub_type
# ---------------------------------------------------------------------------
def test_rejects_non_frame_sub_type(pack_copy: Path) -> None:
with pytest.raises(WrongClaimSubType):
apply_frame_claim(
claim=_claim("gives", sub_type="lexical"),
frame_category="transfer_frame",
polarity="affirms",
reviewer="Ada",
pack_root=pack_copy,
)
# ---------------------------------------------------------------------------
# Test 6 — rejects categories outside SAFE_FRAME_CATEGORIES (wrong=0 hazard)
# ---------------------------------------------------------------------------
def test_rejects_unsafe_frame_category(pack_copy: Path) -> None:
with pytest.raises(
WrongZeroViolationCandidate, match="SAFE_FRAME_CATEGORIES"
):
apply_frame_claim(
claim=_claim("gives"),
frame_category="comparison_frame", # not in allowlist
polarity="affirms",
reviewer="Ada",
pack_root=pack_copy,
)
# ---------------------------------------------------------------------------
# Test 7 — rejects invalid polarity
# ---------------------------------------------------------------------------
def test_rejects_invalid_polarity(pack_copy: Path) -> None:
with pytest.raises(InvalidPolarity, match="polarity must be one of"):
apply_frame_claim(
claim=_claim("gives"),
frame_category="transfer_frame",
polarity="maybe", # not affirms/falsifies
reviewer="Ada",
pack_root=pack_copy,
)
# ---------------------------------------------------------------------------
# Test 8 — rejects evidence tampering
# ---------------------------------------------------------------------------
def test_rejects_evidence_tampering(pack_copy: Path) -> None:
claim = _claim("gives")
object.__setattr__(claim, "evidence_hash", "0" * 64)
with pytest.raises(EvidenceTampering):
apply_frame_claim(
claim=claim,
frame_category="transfer_frame",
polarity="affirms",
reviewer="Ada",
pack_root=pack_copy,
)
# ---------------------------------------------------------------------------
# Test 9 — rejects evidence laundering as source='corpus'
# ---------------------------------------------------------------------------
def test_rejects_evidence_laundered_as_corpus(pack_copy: Path) -> None:
"""ADR-0168.1 §'Evidence floor': source='corpus' MUST be rejected."""
with pytest.raises(EvidenceLaundering, match="MUST NOT be laundered"):
apply_frame_claim(
claim=_claim("gives"),
frame_category="transfer_frame",
polarity="affirms",
reviewer="Ada",
pack_root=pack_copy,
evidence_source="corpus", # forbidden
)
# ---------------------------------------------------------------------------
# Test 10 — case 0050 hazard pin: still refused after ratification
# ---------------------------------------------------------------------------
def test_case_0050_remains_refused_after_frame_ratification(
pack_copy: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
"""After ratifying a safe transfer-frame verb, case 0050 must still refuse.
Case 0050 ("Mark does a gig every other day for 2 weeks. ..." the
period token at sentence_index=0 with missing_operator=
pre_frame_filler_sentence) is the prototype wrong=0 hazard: a
speculative frame opener here would admit a partial graph. The new
frames/ entries must not open that admission path.
"""
apply_frame_claim(
claim=_claim("gives"),
frame_category="transfer_frame",
polarity="affirms",
reviewer="Ada",
pack_root=pack_copy,
)
_use_pack_for_reader(monkeypatch, pack_copy)
cases = [json.loads(line) for line in CASES_PATH.read_text().splitlines()]
case = next(c for c in cases if c["case_id"] == "gsm8k-train-sample-v1-0050")
result, _rows = audit_problem(case["question"], case_id=case["case_id"])
assert isinstance(result, ReaderRefusal), (
"case 0050 must remain refused after FrameClaim ratification"
)
assert result.sentence_index == 0
# ---------------------------------------------------------------------------
# Test 11 — polarity=falsifies branch records non-opener
# ---------------------------------------------------------------------------
def test_polarity_falsifies_records_non_opener(pack_copy: Path) -> None:
"""A 'falsifies' ratification records the surface as NOT opening the frame.
This is the negative-evidence path: an operator reviewed the audit row
and decided the surface does not in fact open the frame category. The
entry is written to the same source file but with polarity=falsifies
so downstream registry builds can mark the surface as a known non-opener.
"""
receipt = apply_frame_claim(
claim=_claim("along"), # not actually a transfer verb
frame_category="transfer_frame",
polarity="falsifies",
reviewer="Ada",
pack_root=pack_copy,
)
assert receipt.polarity == "falsifies"
entry = next(
e
for e in _entries(pack_copy / "frames" / "transfer_frame.jsonl")
if e["surface_form"] == "along"
)
assert entry["polarity"] == "falsifies"
# affirms + falsifies of the same surface produce distinct entries
receipt2 = apply_frame_claim(
claim=_claim("gave"),
frame_category="transfer_frame",
polarity="affirms",
reviewer="Ada",
pack_root=pack_copy,
)
assert receipt2.polarity == "affirms"
polarities = {
e["polarity"]
for e in _entries(pack_copy / "frames" / "transfer_frame.jsonl")
}
assert {"affirms", "falsifies"} <= polarities
# ---------------------------------------------------------------------------
# Test 12 — duplicate evidence appends to existing row
# ---------------------------------------------------------------------------
def test_duplicate_surface_polarity_with_new_evidence_appends_hash(
pack_copy: Path,
) -> None:
"""ADR-0168.1 §'Idempotency': same claim + new evidence appends hash, not row."""
# First ratification
apply_frame_claim(
claim=_claim("gives"),
frame_category="transfer_frame",
polarity="affirms",
reviewer="Ada",
pack_root=pack_copy,
)
# Second piece of evidence for the same claim — different case_id, so
# the evidence_hash differs even though surface+polarity+category match.
second_row = AuditRow(
case_id="case-gives-2",
sentence_index=1,
token_index=4,
token_text="gives",
recognized_terms=("Bob", "buys"),
skipped_frame=None,
missing_operator="pre_frame_filler_sentence",
refusal_reason="unexpected_category",
refusal_detail="duplicate surface evidence",
)
second_claim = from_audit_row(second_row, "frame")
receipt = apply_frame_claim(
claim=second_claim,
frame_category="transfer_frame",
polarity="affirms",
reviewer="Ada",
pack_root=pack_copy,
)
assert receipt.is_duplicate_evidence is True
rows = [
e
for e in _entries(pack_copy / "frames" / "transfer_frame.jsonl")
if e["surface_form"] == "gives"
]
assert len(rows) == 1, "duplicate evidence must not create a second row"
assert len(rows[0]["evidence_hashes"]) == 2
# ---------------------------------------------------------------------------
# Test 13 — manifest.json checksum unchanged
# ---------------------------------------------------------------------------
def test_manifest_checksum_unchanged_by_frame_ratification(pack_copy: Path) -> None:
manifest_bytes_before = (pack_copy / "manifest.json").read_bytes()
manifest_sha_before = _manifest_sha(pack_copy)
declared_before = json.loads(manifest_bytes_before)["checksum"]
apply_frame_claim(
claim=_claim("gives"),
frame_category="transfer_frame",
polarity="affirms",
reviewer="Ada",
pack_root=pack_copy,
)
manifest_bytes_after = (pack_copy / "manifest.json").read_bytes()
assert manifest_bytes_after == manifest_bytes_before
assert _manifest_sha(pack_copy) == manifest_sha_before
assert json.loads(manifest_bytes_after)["checksum"] == declared_before
# ---------------------------------------------------------------------------
# Test 14 — alphabetical sort preserved across multiple writes
# ---------------------------------------------------------------------------
def test_alphabetical_sort_preserved(pack_copy: Path) -> None:
for surface in ("transfers", "gives", "ceded"):
apply_frame_claim(
claim=_claim(surface),
frame_category="transfer_frame",
polarity="affirms",
reviewer="Ada",
pack_root=pack_copy,
)
surfaces = [
e["surface_form"]
for e in _entries(pack_copy / "frames" / "transfer_frame.jsonl")
]
assert surfaces == sorted(surfaces)

View file

@ -54,6 +54,7 @@ _DEFAULT_MATH_AUDIT_PATH = (
# Handlers not listed here are not yet implemented.
_HANDLER_DISPATCH: dict[str, str] = {
"vocabulary_addition": "LexicalClaim",
"frame_reclassification": "FrameClaim",
}
@ -396,6 +397,13 @@ def read_math_proposal(
f"from teaching.math_lexical_ratification import apply_lexical_claim\n"
f"# apply_lexical_claim(claim=<evidence>, category='drain_token', reviewer='<you>')"
)
elif handler_name == "FrameClaim":
suggested_cli = (
f"# ratify via Python REPL (ADR-0168):\n"
f"from teaching.math_frame_ratification import apply_frame_claim\n"
f"# apply_frame_claim(claim=<evidence>, frame_category='increment_frame', "
f"polarity='affirms', reviewer='<you>')"
)
return MathProposalDetail(
proposal_id=proposal_id,
@ -452,6 +460,12 @@ def ratify_math_proposal(
f"from teaching.math_lexical_ratification import apply_lexical_claim\n"
f"# apply_lexical_claim(claim=<evidence>, category='drain_token', reviewer='<you>')"
)
elif handler_name == "FrameClaim":
suggested_cli = (
f"from teaching.math_frame_ratification import apply_frame_claim\n"
f"# apply_frame_claim(claim=<evidence>, frame_category='increment_frame', "
f"polarity='affirms', reviewer='<you>')"
)
return MathRatifyResult(
proposal_id=proposal_id,