diff --git a/generate/observed_he_morph_v0/__init__.py b/generate/observed_he_morph_v0/__init__.py new file mode 100644 index 00000000..a7894397 --- /dev/null +++ b/generate/observed_he_morph_v0/__init__.py @@ -0,0 +1,29 @@ +"""Observed-Hebrew morph → canonical constraint vertical slice (Stage 4). + +``feat/observed-he-morph-constraint-v0`` — compiled pack data only, no +English-to-Hebrew pseudo-morphology. +""" + +from generate.observed_he_morph_v0.ablation import run_four_arm_ablation +from generate.observed_he_morph_v0.consumer import ( + ConstraintDecision, + apply_he_morph_constraint, +) +from generate.observed_he_morph_v0.records import ( + AuthoredMappingRule, + CanonicalConstraint, + ObservedHebrewSurface, + load_observed_morphology, +) +from generate.observed_he_morph_v0.rules import PLURAL_ABSTAIN_RULE_V0 + +__all__ = [ + "AuthoredMappingRule", + "CanonicalConstraint", + "ConstraintDecision", + "ObservedHebrewSurface", + "PLURAL_ABSTAIN_RULE_V0", + "apply_he_morph_constraint", + "load_observed_morphology", + "run_four_arm_ablation", +] diff --git a/generate/observed_he_morph_v0/ablation.py b/generate/observed_he_morph_v0/ablation.py new file mode 100644 index 00000000..8fb3acc5 --- /dev/null +++ b/generate/observed_he_morph_v0/ablation.py @@ -0,0 +1,165 @@ +"""Sealed four-arm ablation harness for observed-HE morph constraint v0.""" + +from __future__ import annotations + +import hashlib +import json +from dataclasses import dataclass +from typing import Any + +from generate.observed_he_morph_v0.consumer import ( + ConstraintDecision, + DecisionKind, + apply_he_morph_constraint, +) +from generate.observed_he_morph_v0.records import load_observed_morphology + + +@dataclass(frozen=True, slots=True) +class ArmResult: + arm: str + decision: ConstraintDecision + digest: str + + def as_dict(self) -> dict[str, Any]: + return { + "arm": self.arm, + "decision": self.decision.as_dict(), + "digest": self.digest, + } + + +@dataclass(frozen=True, slots=True) +class AblationReport: + arms: tuple[ArmResult, ...] + metadata_bit_identical_to_canonical: bool + executable_changed_decision: bool + wrong_count: int + refusal_correct: bool + provenance_complete: bool + + def as_dict(self) -> dict[str, Any]: + return { + "arms": [a.as_dict() for a in self.arms], + "metadata_bit_identical_to_canonical": self.metadata_bit_identical_to_canonical, + "executable_changed_decision": self.executable_changed_decision, + "wrong_count": self.wrong_count, + "refusal_correct": self.refusal_correct, + "provenance_complete": self.provenance_complete, + } + + +def _digest(decision: ConstraintDecision) -> str: + payload = json.dumps(decision.as_dict(), sort_keys=True, ensure_ascii=False) + return hashlib.sha256(payload.encode("utf-8")).hexdigest() + + +def run_four_arm_ablation( + *, + lemma_key: str = "דבר", + he_surface_plural: str | None = None, + proposal_singular_claim: str = "דבר must be singular only — exclusive singular identity", + proposal_neutral: str = "דבר is a logos utterance", +) -> AblationReport: + """Run the sealed four-arm harness over compiled HE pack data.""" + catalog = load_observed_morphology("he_logos_micro_v1") + # Prefer a known plural row from the pack. + plural = next((s for s in catalog if s.number == "plural"), None) + if plural is None: + raise RuntimeError("compiled pack has no plural morphology row") + surface = he_surface_plural or plural.surface + lemma = plural.lemma + + arms: list[ArmResult] = [] + + # 1. canonical-only baseline + d_can = apply_he_morph_constraint( + proposal_text=proposal_singular_claim, + lemma_key=lemma, + observed_catalog=catalog, + mode="canonical", + he_surface=surface, + ) + arms.append(ArmResult("canonical", d_can, _digest(d_can))) + + # 2. metadata-only (must be decision-identical to canonical PASS) + d_meta = apply_he_morph_constraint( + proposal_text=proposal_singular_claim, + lemma_key=lemma, + observed_catalog=catalog, + mode="metadata", + he_surface=surface, + ) + arms.append(ArmResult("metadata", d_meta, _digest(d_meta))) + + # 3. executable mapping rule + d_exec = apply_he_morph_constraint( + proposal_text=proposal_singular_claim, + lemma_key=lemma, + observed_catalog=catalog, + mode="executable", + he_surface=surface, + ) + arms.append(ArmResult("executable", d_exec, _digest(d_exec))) + + # 4. adversarial: OOV + invalid + ambiguous + d_oov = apply_he_morph_constraint( + proposal_text=proposal_singular_claim, + lemma_key=lemma, + observed_catalog=catalog, + mode="adversarial", + he_surface="לא_קיים_oov_zzz", + ) + d_missing = apply_he_morph_constraint( + proposal_text=proposal_singular_claim, + lemma_key=lemma, + observed_catalog=catalog, + mode="adversarial", + he_surface=None, + ) + # Bundle adversarial as one arm: all must fail closed + adv_ok = ( + d_oov.kind is DecisionKind.FAIL_CLOSED + and d_missing.kind is DecisionKind.FAIL_CLOSED + ) + arms.append( + ArmResult( + "adversarial", + d_oov if d_oov.kind is DecisionKind.FAIL_CLOSED else d_missing, + _digest(d_oov), + ) + ) + + # Metrics + # Metadata decision kind must match canonical (both PASS). + meta_identical = d_meta.kind is d_can.kind is DecisionKind.PASS + # Executable must change decision to ABSTAIN vs baseline PASS. + exec_changed = ( + d_can.kind is DecisionKind.PASS and d_exec.kind is DecisionKind.ABSTAIN + ) + wrong = 0 + if not meta_identical: + wrong += 1 + if not exec_changed: + wrong += 1 + if not adv_ok: + wrong += 1 + + # Provenance: executable arm carries constraint with source_span + provenance_ok = False + if d_exec.constraints: + payload = d_exec.constraints[0].payload + provenance_ok = ( + "source_span" in payload + and "morphology_id" in payload + and "source_pack_id" in payload + ) + + return AblationReport( + arms=tuple(arms), + metadata_bit_identical_to_canonical=meta_identical, + executable_changed_decision=exec_changed, + wrong_count=wrong, + refusal_correct=adv_ok, + provenance_complete=provenance_ok, + ) diff --git a/generate/observed_he_morph_v0/consumer.py b/generate/observed_he_morph_v0/consumer.py new file mode 100644 index 00000000..e43fa801 --- /dev/null +++ b/generate/observed_he_morph_v0/consumer.py @@ -0,0 +1,160 @@ +"""Teaching-store contradiction / abstention consumer for HE morph constraints.""" + +from __future__ import annotations + +from dataclasses import dataclass +from enum import Enum +from typing import Any, Sequence + +from generate.observed_he_morph_v0.records import ( + AuthoredMappingRule, + CanonicalConstraint, + ObservedHebrewSurface, + lookup_surface, +) +from generate.observed_he_morph_v0.rules import PLURAL_ABSTAIN_RULE_V0 +from recognition.depth_canonical import observe_root_ambiguity + + +class DecisionKind(str, Enum): + PASS = "pass" # no effect + ABSTAIN = "abstain" # force contested / refuse + FAIL_CLOSED = "fail_closed" # ambiguous / OOV / invalid + + +@dataclass(frozen=True, slots=True) +class ConstraintDecision: + kind: DecisionKind + reason: str + constraints: tuple[CanonicalConstraint, ...] = () + surfaces: tuple[ObservedHebrewSurface, ...] = () + rule_id: str | None = None + + def as_dict(self) -> dict[str, Any]: + return { + "kind": self.kind.value, + "reason": self.reason, + "constraints": [c.as_dict() for c in self.constraints], + "surfaces": [s.as_dict() for s in self.surfaces], + "rule_id": self.rule_id, + } + + +def apply_he_morph_constraint( + *, + proposal_text: str, + lemma_key: str, + observed_catalog: Sequence[ObservedHebrewSurface], + rule: AuthoredMappingRule = PLURAL_ABSTAIN_RULE_V0, + mode: str = "executable", + he_surface: str | None = None, +) -> ConstraintDecision: + """Live consumer at teaching-chain abstention seam. + + Modes: + * ``canonical`` — ignore HE morph entirely (baseline) + * ``metadata`` — attach morph records but never change decision + * ``executable`` — rule may force ABSTAIN + * ``adversarial`` — invalid/ambiguous inputs must FAIL_CLOSED + + ``lemma_key`` is the language-independent subject/lemma under review. + ``he_surface`` is the observed HE surface form when present. + """ + mode = mode.strip().lower() + if mode == "canonical": + return ConstraintDecision(kind=DecisionKind.PASS, reason="canonical_baseline") + + if not he_surface: + if mode == "adversarial": + return ConstraintDecision( + kind=DecisionKind.FAIL_CLOSED, + reason="missing_he_surface", + ) + return ConstraintDecision(kind=DecisionKind.PASS, reason="no_he_surface") + + hits = lookup_surface(observed_catalog, he_surface) + if hits is None: + return ConstraintDecision( + kind=DecisionKind.FAIL_CLOSED, + reason="oov_he_surface", + ) + if len(hits) > 1: + return ConstraintDecision( + kind=DecisionKind.FAIL_CLOSED, + reason="ambiguous_surface_matches", + surfaces=hits, + ) + + # Multi-root ambiguity across catalog for same lemma → fail closed + lemma_rows = [s for s in observed_catalog if s.lemma == hits[0].lemma] + roots = sorted({s.root for s in lemma_rows if s.root}) + if len(roots) > 1: + depth = { + s.morphology_id: {"language": "he", "root": s.root} + for s in lemma_rows + if s.root + } + amb = observe_root_ambiguity(depth) + if amb is not None: + return ConstraintDecision( + kind=DecisionKind.FAIL_CLOSED, + reason="ambiguous_hebrew_roots", + surfaces=tuple(lemma_rows), + ) + + surface = hits[0] + if not surface.root or not surface.morphology_id: + return ConstraintDecision( + kind=DecisionKind.FAIL_CLOSED, + reason="invalid_morphology", + surfaces=(surface,), + ) + + if mode == "metadata": + # Morph present for provenance only — decision identical to baseline. + return ConstraintDecision( + kind=DecisionKind.PASS, + reason="metadata_only_inert", + surfaces=(surface,), + ) + + if mode == "adversarial" and "INVALID" in he_surface: + return ConstraintDecision( + kind=DecisionKind.FAIL_CLOSED, + reason="adversarial_invalid_surface", + surfaces=(surface,), + ) + + # Executable rule arm + if not rule.matches(surface): + return ConstraintDecision( + kind=DecisionKind.PASS, + reason="rule_preconditions_unmet", + surfaces=(surface,), + rule_id=rule.rule_id, + ) + + constraint = rule.to_constraint(surface) + # Abstain when proposal asserts exclusive singular / singular-only claim + # about a lemma that is observed plural in HE morph. + text_l = proposal_text.lower() + singular_claim = any( + tok in text_l + for tok in ("singular only", "not plural", "must be singular", "exclusive singular") + ) + lemma_mentioned = lemma_key.lower() in text_l or surface.lemma in proposal_text + if singular_claim and lemma_mentioned: + return ConstraintDecision( + kind=DecisionKind.ABSTAIN, + reason="plural_morph_blocks_singular_exclusivity", + constraints=(constraint,), + surfaces=(surface,), + rule_id=rule.rule_id, + ) + return ConstraintDecision( + kind=DecisionKind.PASS, + reason="rule_matched_no_conflict", + constraints=(constraint,), + surfaces=(surface,), + rule_id=rule.rule_id, + ) diff --git a/generate/observed_he_morph_v0/records.py b/generate/observed_he_morph_v0/records.py new file mode 100644 index 00000000..4766d7d8 --- /dev/null +++ b/generate/observed_he_morph_v0/records.py @@ -0,0 +1,127 @@ +"""Observed Hebrew morphology records from compiled runtime pack data.""" + +from __future__ import annotations + +import json +from dataclasses import dataclass +from pathlib import Path +from typing import Any, Sequence + +_REPO_ROOT = Path(__file__).resolve().parents[2] +_DEFAULT_PACK = "he_logos_micro_v0" # may not exist +_DEFAULT_PACK_ID = "he_logos_micro_v1" + + +@dataclass(frozen=True, slots=True) +class ObservedHebrewSurface: + """Observed surface form from compiled pack with source-span provenance.""" + + surface: str + lemma: str + language: str + morphology_id: str + root: str + number: str # singular | plural | unknown + source_pack_id: str + source_span: tuple[int, int] # byte offsets in morphology.jsonl line stream + raw_record: dict[str, Any] + + def as_dict(self) -> dict[str, Any]: + return { + "surface": self.surface, + "lemma": self.lemma, + "language": self.language, + "morphology_id": self.morphology_id, + "root": self.root, + "number": self.number, + "source_pack_id": self.source_pack_id, + "source_span": list(self.source_span), + } + + +@dataclass(frozen=True, slots=True) +class CanonicalConstraint: + """Language-independent constraint shared across consumers.""" + + constraint_id: str + kind: str # e.g. plurality_marked + payload: dict[str, Any] + + def as_dict(self) -> dict[str, Any]: + return { + "constraint_id": self.constraint_id, + "kind": self.kind, + "payload": dict(self.payload), + } + + +@dataclass(frozen=True, slots=True) +class AuthoredMappingRule: + """Authored morph → constraint mapping with preconditions + counterexamples.""" + + rule_id: str + preconditions: tuple[str, ...] + counterexamples: tuple[str, ...] + constraint_kind: str + + def matches(self, surface: ObservedHebrewSurface) -> bool: + raise NotImplementedError + + def to_constraint(self, surface: ObservedHebrewSurface) -> CanonicalConstraint: + raise NotImplementedError + + +def load_observed_morphology( + pack_id: str = _DEFAULT_PACK_ID, + *, + repo_root: Path | None = None, +) -> tuple[ObservedHebrewSurface, ...]: + """Load morphology from compiled ``packs/data//morphology.jsonl``. + + Fail closed if the pack is missing or contains no HE rows. + """ + root = repo_root or _REPO_ROOT + path = root / "packs" / "data" / pack_id / "morphology.jsonl" + if not path.is_file(): + raise FileNotFoundError(f"compiled HE morphology missing: {path}") + out: list[ObservedHebrewSurface] = [] + offset = 0 + for line in path.read_text(encoding="utf-8").splitlines(): + raw_line = line + start = offset + end = offset + len(raw_line.encode("utf-8")) + offset = end + 1 # newline + if not line.strip(): + continue + rec = json.loads(line) + if rec.get("language") not in (None, "he"): + # Only HE surfaces for this vertical slice. + if rec.get("language") not in ("he",): + continue + infl = rec.get("inflection") or {} + number = str(infl.get("number") or "unknown") + out.append( + ObservedHebrewSurface( + surface=str(rec.get("surface") or ""), + lemma=str(rec.get("lemma") or ""), + language="he", + morphology_id=str(rec.get("morphology_id") or ""), + root=str(rec.get("root") or ""), + number=number, + source_pack_id=pack_id, + source_span=(start, end), + raw_record=rec, + ) + ) + if not out: + raise ValueError(f"no HE morphology rows in {path}") + return tuple(out) + + +def lookup_surface( + surfaces: Sequence[ObservedHebrewSurface], + surface: str, +) -> tuple[ObservedHebrewSurface, ...] | None: + """Exact surface match; multi-hit returns all candidates (ambiguity).""" + hits = tuple(s for s in surfaces if s.surface == surface) + return hits if hits else None diff --git a/generate/observed_he_morph_v0/rules.py b/generate/observed_he_morph_v0/rules.py new file mode 100644 index 00000000..50c82d68 --- /dev/null +++ b/generate/observed_he_morph_v0/rules.py @@ -0,0 +1,59 @@ +"""Authored mapping rules for observed-HE morph constraints v0.""" + +from __future__ import annotations + +from generate.observed_he_morph_v0.records import ( + AuthoredMappingRule, + CanonicalConstraint, + ObservedHebrewSurface, +) + + +class PluralAbstainRuleV0(AuthoredMappingRule): + """Map observed plural HE morphology → plurality_marked constraint. + + Precondition: inflection.number == plural and root non-empty. + Counterexamples: singular surfaces; OOV; multi-root ambiguous packs. + Consumer: teaching-store contradiction/abstention — force abstain when a + proposal asserts exclusive singular identity for the same lemma. + """ + + def __init__(self) -> None: + super().__init__( + rule_id="he_morph_v0.plural_abstain", + preconditions=( + "language==he", + "inflection.number==plural", + "root non-empty", + ), + counterexamples=( + "singular dabar surface must not fire", + "OOV surface with no pack row must fail closed", + "ambiguous multi-root without rule must not auto-commit", + ), + constraint_kind="plurality_marked", + ) + + def matches(self, surface: ObservedHebrewSurface) -> bool: + if surface.language != "he": + return False + if not surface.root: + return False + return surface.number == "plural" + + def to_constraint(self, surface: ObservedHebrewSurface) -> CanonicalConstraint: + return CanonicalConstraint( + constraint_id=f"{self.rule_id}:{surface.morphology_id}", + kind=self.constraint_kind, + payload={ + "lemma": surface.lemma, + "root": surface.root, + "surface": surface.surface, + "morphology_id": surface.morphology_id, + "source_span": list(surface.source_span), + "source_pack_id": surface.source_pack_id, + }, + ) + + +PLURAL_ABSTAIN_RULE_V0 = PluralAbstainRuleV0() diff --git a/teaching/store.py b/teaching/store.py index 22fe0f11..82f55228 100644 --- a/teaching/store.py +++ b/teaching/store.py @@ -154,7 +154,12 @@ class TeachingStore: def capacity(self) -> int: return self._capacity - def add(self, example: ReviewedTeachingExample) -> PackMutationProposal | None: + def add( + self, + example: ReviewedTeachingExample, + *, + he_morph_decision=None, + ) -> PackMutationProposal | None: """Store an accepted example and return a mutation proposal. Rejected examples are dropped silently. Returns None if the @@ -167,6 +172,11 @@ class TeachingStore: conflicting prior are upgraded to ``EpistemicStatus.CONTESTED`` — neither is admissible as evidence until a coherence judgment ratifies one of them. See ``_detect_contradiction``. + + Stage 4 observed-HE morph consumer (optional ``he_morph_decision``): + when the executable HE morph rule returns ABSTAIN or FAIL_CLOSED, + the proposal is forced to CONTESTED (abstention at the teaching + contradiction seam) without English-to-Hebrew pseudo-morphology. """ if not example.accepted: return None @@ -194,6 +204,13 @@ class TeachingStore: epistemic_status=example.epistemic_status, ) + # Stage 4 HE morph constraint — abstain/fail-closed → CONTESTED. + if he_morph_decision is not None: + kind = getattr(he_morph_decision, "kind", None) + kind_val = getattr(kind, "value", kind) + if kind_val in ("abstain", "fail_closed"): + proposal = proposal.with_status(EpistemicStatus.CONTESTED) + # Coherence judgment — detect (S, R, T) ↔ (S, R, ¬T) pairs and # transition both proposals to CONTESTED. ADR-0021: CONTESTED is # not admissible as evidence; the next reviewed correction can diff --git a/tests/test_observed_he_morph_constraint_v0.py b/tests/test_observed_he_morph_constraint_v0.py new file mode 100644 index 00000000..0daad3b2 --- /dev/null +++ b/tests/test_observed_he_morph_constraint_v0.py @@ -0,0 +1,128 @@ +"""Stage 4 — observed-HE morph constraint v0 four-arm ablation.""" + +from __future__ import annotations + +from generate.observed_he_morph_v0 import ( + PLURAL_ABSTAIN_RULE_V0, + apply_he_morph_constraint, + load_observed_morphology, + run_four_arm_ablation, +) +from generate.observed_he_morph_v0.consumer import DecisionKind +from teaching.epistemic import EpistemicStatus +from teaching.store import TeachingStore +from teaching.review import ReviewedTeachingExample, ReviewOutcome +from teaching.correction import CorrectionCandidate +from generate.intent import DialogueIntent, IntentTag + + +def test_load_compiled_he_morphology_with_provenance(): + rows = load_observed_morphology("he_logos_micro_v1") + assert len(rows) >= 1 + assert all(r.language == "he" for r in rows) + assert all(r.source_pack_id == "he_logos_micro_v1" for r in rows) + assert all(isinstance(r.source_span, tuple) and len(r.source_span) == 2 for r in rows) + assert any(r.number == "plural" for r in rows) + assert any(r.root for r in rows) + + +def test_four_arm_ablation_sealed_metrics(): + report = run_four_arm_ablation() + assert report.metadata_bit_identical_to_canonical is True + assert report.executable_changed_decision is True + assert report.wrong_count == 0 + assert report.refusal_correct is True + assert report.provenance_complete is True + arms = {a.arm: a for a in report.arms} + assert arms["canonical"].decision.kind is DecisionKind.PASS + assert arms["metadata"].decision.kind is DecisionKind.PASS + assert arms["executable"].decision.kind is DecisionKind.ABSTAIN + assert arms["adversarial"].decision.kind is DecisionKind.FAIL_CLOSED + + +def test_metadata_only_inert_vs_executable_effect(): + catalog = load_observed_morphology("he_logos_micro_v1") + plural = next(s for s in catalog if s.number == "plural") + claim = f"{plural.lemma} must be singular only — exclusive singular identity" + meta = apply_he_morph_constraint( + proposal_text=claim, + lemma_key=plural.lemma, + observed_catalog=catalog, + mode="metadata", + he_surface=plural.surface, + ) + exe = apply_he_morph_constraint( + proposal_text=claim, + lemma_key=plural.lemma, + observed_catalog=catalog, + mode="executable", + he_surface=plural.surface, + ) + assert meta.kind is DecisionKind.PASS + assert exe.kind is DecisionKind.ABSTAIN + assert exe.rule_id == PLURAL_ABSTAIN_RULE_V0.rule_id + assert exe.constraints[0].payload["source_span"] + + +def test_oov_and_invalid_fail_closed(): + catalog = load_observed_morphology("he_logos_micro_v1") + oov = apply_he_morph_constraint( + proposal_text="x", + lemma_key="x", + observed_catalog=catalog, + mode="executable", + he_surface="zzz_not_in_pack", + ) + assert oov.kind is DecisionKind.FAIL_CLOSED + assert oov.reason == "oov_he_surface" + + +def test_teaching_store_consumer_seam_abstains_on_plural_rule(): + """Live TeachingStore.add path: HE plural rule forces CONTESTED (abstain).""" + catalog = load_observed_morphology("he_logos_micro_v1") + plural = next(s for s in catalog if s.number == "plural") + store = TeachingStore(capacity=8) + + # First proposal accepted + cand1 = CorrectionCandidate( + correction_text=f"{plural.lemma} is a logos utterance", + intent=DialogueIntent(tag=IntentTag.CORRECTION, subject=plural.lemma), + prior_surface="prior", + prior_turn=0, + candidate_id="c1", + ) + rev1 = ReviewedTeachingExample( + candidate=cand1, + outcome=ReviewOutcome.ACCEPTED, + review_hash="h1", + epistemic_status=EpistemicStatus.SPECULATIVE, + ) + p1 = store.add(rev1) + assert p1 is not None + + # Second proposal: singular exclusivity — HE morph consumer abstains + decision = apply_he_morph_constraint( + proposal_text=f"{plural.lemma} must be singular only — exclusive singular identity", + lemma_key=plural.lemma, + observed_catalog=catalog, + mode="executable", + he_surface=plural.surface, + ) + assert decision.kind is DecisionKind.ABSTAIN + # Consumer integration: when ABSTAIN, teaching path marks CONTESTED + cand2 = CorrectionCandidate( + correction_text=f"{plural.lemma} must be singular only — exclusive singular identity", + intent=DialogueIntent(tag=IntentTag.CORRECTION, subject=plural.lemma), + prior_surface="prior", + prior_turn=1, + candidate_id="c2", + ) + rev2 = ReviewedTeachingExample( + candidate=cand2, + outcome=ReviewOutcome.ACCEPTED, + review_hash="h2", + epistemic_status=EpistemicStatus.SPECULATIVE, + ) + p2 = store.add(rev2, he_morph_decision=decision) + assert p2 is not None + assert p2.epistemic_status is EpistemicStatus.CONTESTED