diff --git a/core/cognition/pipeline.py b/core/cognition/pipeline.py index e7e30af6..ee5f8c0e 100644 --- a/core/cognition/pipeline.py +++ b/core/cognition/pipeline.py @@ -482,13 +482,54 @@ class CognitiveTurnPipeline: ) surface = resolved.surface articulation_surface = resolved.articulation_surface + authority_source = resolved.authority + + # === LOGOS MORPH AUTHORITY (bulk live seam) === + # Same pure decision function as teaching store + four-arm ablation. + # Executable morph may force abstain/refuse; never soft-pass a + # certified singular-exclusivity claim against observed plural HE. + # English-only turns with no HE surface → no-op (None). + logos_decision_kind = "" + logos_decision_reason = "" + logos_rule_id = "" + logos_constraint_id = "" + logos_decision = None + try: + from generate.observed_he_morph_v0.authority import ( + decision_as_coherence_refusal, + evaluate_logos_on_text, + first_logos_constraint, + logos_blocks_certified_answer, + ) + + logos_decision = evaluate_logos_on_text(text=text, mode="executable") + if logos_decision is not None: + logos_decision_kind = logos_decision.kind.value + logos_decision_reason = logos_decision.reason + logos_rule_id = logos_decision.rule_id or "" + lc = first_logos_constraint(logos_decision) + if lc is not None: + logos_constraint_id = lc.constraint_id + if logos_blocks_certified_answer(logos_decision): + refusal = decision_as_coherence_refusal(logos_decision) + surface = refusal.surface_message or refusal.message + articulation_surface = surface + authority_source = "logos_morph_constraint" + except Exception: + # Pack load / catalog failure must not crash the turn spine; + # English path continues without morph authority. + logos_decision = None # SUBSTRATE_BYPASS_HAZARD telemetry (data-driven roadmap). # Only populated when a graph existed yet substrate did not win. # This is *observability only* — never used to change control flow # after the fact, never folded into trace_hash in Phase A. substrate_hazard: tuple[str, ...] = () - if effective_graph is not None and resolved.authority not in ("substrate_realizer", "realizer"): + if effective_graph is not None and authority_source not in ( + "substrate_realizer", + "realizer", + "logos_morph_constraint", + ): reasons: list[str] = [] if not effective_graph.is_fully_grounded(): reasons.append("unfilled_pending_slots") @@ -501,6 +542,11 @@ class CognitiveTurnPipeline: if not _is_useful_surface(realized_plan.surface): reasons.append("realizer_surface_not_useful") substrate_hazard = tuple(reasons) + if logos_decision is not None and logos_blocks_certified_answer(logos_decision): + substrate_hazard = substrate_hazard + ( + f"logos_morph_{logos_decision.kind.value}", + logos_decision.reason, + ) # Phase C (Geometric Anti-Unification) — read-only telemetry instrumentation. # Captures the graph-structural context around any OOV/pending "hole" @@ -713,6 +759,10 @@ class CognitiveTurnPipeline: # recognition wins (earlier-fail boundary) over generation. _generation_refusal_reason = getattr(response, "refusal_reason", "") or "" refusal_reason = _recognition_refusal_reason or _generation_refusal_reason + # Logos morph abstain/fail-closed is answer-authority: surface the reason + # when morph blocked certification (does not override recognition refuse). + if not refusal_reason and logos_decision_kind in ("abstain", "fail_closed"): + refusal_reason = logos_decision_reason # Phase B — Quantized Topological Hashing for the cognitive spine. # Include the discrete (string-only) topological form of the @@ -825,13 +875,19 @@ class CognitiveTurnPipeline: # authority_source makes the winner of the substrate vs legacy # decision first-class evidence (visible in trace, workbench, # evals). substrate_hazard is the precise bypass signal. - authority_source=resolved.authority, + # Logos morph may override to "logos_morph_constraint" when it + # blocks certified answers (same decision fn as teaching/ablation). + authority_source=authority_source, substrate_hazard=substrate_hazard, oov_geometric_context=oov_geometric_context, # 3-lang depth unification: surface the same data at top level on result # (extracted from pre-computed node_depths var or oov_geometric_context to keep single source) node_depths=node_depths if node_depths else None, graph_anti_unify=(oov_geometric_context or {}).get("graph_anti_unify") if oov_geometric_context else None, + logos_decision_kind=logos_decision_kind, + logos_decision_reason=logos_decision_reason, + logos_rule_id=logos_rule_id, + logos_constraint_id=logos_constraint_id, ) # ------------------------------------------------------------------ diff --git a/core/cognition/result.py b/core/cognition/result.py index 55bd599a..4b2d511f 100644 --- a/core/cognition/result.py +++ b/core/cognition/result.py @@ -207,3 +207,13 @@ class CognitiveTurnResult: # Never folded into trace_hash (observational only, like oov_geometric_context). node_depths: dict | None = None graph_anti_unify: dict | None = None + + # --- Logos morph authority (observational + outcome-affecting when non-pass) --- + # Populated when observed HE surface is present in the turn input and the + # shared ``evaluate_logos_on_text`` path runs. Empty kind == not consulted. + # Never folded into trace_hash as a separate key (surface/refusal already + # capture the user-visible effect when morph blocks certification). + logos_decision_kind: str = "" + logos_decision_reason: str = "" + logos_rule_id: str = "" + logos_constraint_id: str = "" diff --git a/core/semantic_primitives/__init__.py b/core/semantic_primitives/__init__.py index 99b9481b..e9ae715d 100644 --- a/core/semantic_primitives/__init__.py +++ b/core/semantic_primitives/__init__.py @@ -15,6 +15,7 @@ from core.semantic_primitives.model import ( Entity, Event, IdentityBridge, + LogosConstraint, MissingReferent, Operator, OperatorClass, @@ -36,6 +37,7 @@ __all__ = [ "Entity", "Event", "IdentityBridge", + "LogosConstraint", "MissingReferent", "Operator", "OperatorClass", diff --git a/core/semantic_primitives/model.py b/core/semantic_primitives/model.py index 7bc77593..83aceb07 100644 --- a/core/semantic_primitives/model.py +++ b/core/semantic_primitives/model.py @@ -86,6 +86,62 @@ class ProvenanceSpan: raise ValidationError("ProvenanceSpan requires 0 <= start <= end") +@dataclass(frozen=True, slots=True) +class LogosConstraint: + """Language-independent morph→constraint record with full provenance. + + Used by the live Logos authority path (teaching + CognitiveTurn). Fields + are typed — not a free-form meaning dict. rule_id + morphology_id + + source_pack_id + source_span make the constraint falsifiable and replayable. + """ + + constraint_id: str + kind: str + rule_id: str + lemma: str + root: str + surface: str + morphology_id: str + source_pack_id: str + source_span: ProvenanceSpan + language: str = "he" + + def __post_init__(self) -> None: + _require_nonempty(self.constraint_id, "LogosConstraint.constraint_id") + _require_nonempty(self.kind, "LogosConstraint.kind") + _require_nonempty(self.rule_id, "LogosConstraint.rule_id") + _require_nonempty(self.morphology_id, "LogosConstraint.morphology_id") + _require_nonempty(self.source_pack_id, "LogosConstraint.source_pack_id") + if not isinstance(self.source_span, ProvenanceSpan): + raise ValidationError( + "LogosConstraint.source_span must be a ProvenanceSpan" + ) + _reject_dict_blob(self.kind, "LogosConstraint.kind") + + def as_dict(self) -> dict[str, Any]: + return { + "constraint_id": self.constraint_id, + "kind": self.kind, + "rule_id": self.rule_id, + "lemma": self.lemma, + "root": self.root, + "surface": self.surface, + "morphology_id": self.morphology_id, + "source_pack_id": self.source_pack_id, + "source_span": [self.source_span.start, self.source_span.end], + "language": self.language, + } + + @property + def provenance_complete(self) -> bool: + return bool( + self.rule_id + and self.morphology_id + and self.source_pack_id + and self.source_span.end >= self.source_span.start + ) + + @dataclass(frozen=True, slots=True) class Entity: """Persistent identity anchor with typed id, name, type, and frame bindings.""" diff --git a/docs/analysis/logos-bulk-live-authority-residual-2026-07-20.md b/docs/analysis/logos-bulk-live-authority-residual-2026-07-20.md new file mode 100644 index 00000000..ecd72176 --- /dev/null +++ b/docs/analysis/logos-bulk-live-authority-residual-2026-07-20.md @@ -0,0 +1,35 @@ +# Logos bulk live-authority residual (honest accounting) + +**Date:** 2026-07-20 +**Branch:** `feat/logos-bulk-live-authority` +**Claim:** bulk Logos **impact** landed on live seams with sealed four-arm proof — **not** full Logos complete. + +## What now has live authority + +| Seam | Module | Decision function | Outcome effect | +|------|--------|-------------------|----------------| +| Teaching store `add` | `teaching/store.py` → `authority.evaluate_logos_on_text` | `apply_he_morph_constraint` | ABSTAIN/FAIL_CLOSED → proposal `CONTESTED` | +| CognitiveTurn answer path | `core/cognition/pipeline.py` after `resolve_surface` | same | ABSTAIN/FAIL_CLOSED → typed abstention surface, `authority_source=logos_morph_constraint` | +| Four-arm ablation | `generate/observed_he_morph_v0/ablation.py` | same | sealed metrics | + +Shared typed IR: `CanonicalConstraint` → `LogosConstraint` (`core.semantic_primitives`) with rule_id, morphology_id, source_pack_id, source_span — no free meaning dict on the exercised path. + +## Sealed improvement proof (expected signals) + +- `metadata_bit_identical_to_canonical=true` — morph observation does not smuggle outcomes +- `executable_changed_decision=true` — plural HE blocks singular-exclusivity claims +- `wrong_count=0`, `refusal_correct=true`, `provenance_complete=true` +- Dual-run digests identical (determinism) + +## Residual dual-system debt (not closed) + +1. **`generate/linguistic_pipeline` cue tables** (English “sold”/“mkr” hand maps) remain a **parallel** constraint producer for the governance demo path; they do **not** consult `packs/data/he_*` morphology and are **not** answer authority over field/fail-closed. +2. **Lexicon breadth** — only `he_logos_micro_v1` observed surfaces; no full HE/GRC tables, no binyan universals, no GRC case→role. +3. **Legacy math / meaning_graph IR** — still parallel outside the Logos morph seam; not fully migrated onto `semantic_primitives`. +4. **Holonomy crown** — still not robust; no LIVE claim. +5. **Sense disambiguation / frames** — pack frames largely disconnected; first-match pack semantics unchanged outside this morph rule. +6. **One rule type** — `he_morph_v0.plural_abstain` only; construct-state / prep+case / aspect deferred. + +## What “bulk impact” means here + +Logos now **changes closed outcomes** on production teaching + turn paths under provenance, and **refuses** adversarial OOV/missing surfaces — the bulk of the novel design’s *authority* contribution for this vertical. Remaining work is **breadth + IR consolidation**, not reopening fail-closed architecture. diff --git a/generate/observed_he_morph_v0/__init__.py b/generate/observed_he_morph_v0/__init__.py index a7894397..bd2bfdba 100644 --- a/generate/observed_he_morph_v0/__init__.py +++ b/generate/observed_he_morph_v0/__init__.py @@ -1,12 +1,21 @@ -"""Observed-Hebrew morph → canonical constraint vertical slice (Stage 4). +"""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. +Compiled pack data only; no English-to-Hebrew pseudo-morphology. +Live authority is shared via ``authority.evaluate_logos_on_text`` for +teaching store and CognitiveTurn (same pure decision function as ablation). """ from generate.observed_he_morph_v0.ablation import run_four_arm_ablation +from generate.observed_he_morph_v0.authority import ( + decision_as_coherence_refusal, + evaluate_logos_on_text, + first_logos_constraint, + logos_blocks_certified_answer, + scan_observed_he_surface, +) from generate.observed_he_morph_v0.consumer import ( ConstraintDecision, + DecisionKind, apply_he_morph_constraint, ) from generate.observed_he_morph_v0.records import ( @@ -21,9 +30,15 @@ __all__ = [ "AuthoredMappingRule", "CanonicalConstraint", "ConstraintDecision", + "DecisionKind", "ObservedHebrewSurface", "PLURAL_ABSTAIN_RULE_V0", "apply_he_morph_constraint", + "decision_as_coherence_refusal", + "evaluate_logos_on_text", + "first_logos_constraint", "load_observed_morphology", + "logos_blocks_certified_answer", "run_four_arm_ablation", + "scan_observed_he_surface", ] diff --git a/generate/observed_he_morph_v0/ablation.py b/generate/observed_he_morph_v0/ablation.py index 3cb9331d..8b7e23aa 100644 --- a/generate/observed_he_morph_v0/ablation.py +++ b/generate/observed_he_morph_v0/ablation.py @@ -145,14 +145,18 @@ def run_four_arm_ablation( if not adv_ok: wrong += 1 - # Provenance: executable arm carries constraint with source_span + # Provenance: executable arm carries typed constraint (rule_id, morph, pack, 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 + c0 = d_exec.constraints[0] + provenance_ok = bool( + getattr(c0, "provenance_complete", False) + or ( + c0.morphology_id + and c0.source_pack_id + and c0.rule_id + and c0.source_span + ) ) return AblationReport( diff --git a/generate/observed_he_morph_v0/authority.py b/generate/observed_he_morph_v0/authority.py new file mode 100644 index 00000000..8a055c71 --- /dev/null +++ b/generate/observed_he_morph_v0/authority.py @@ -0,0 +1,200 @@ +"""Shared Logos morph authority — single decision entry for live seams + ablation. + +Teaching store and CognitiveTurn consult the same pure decision function +(``apply_he_morph_constraint``) via helpers here. Modes: + + * canonical / metadata / executable / adversarial + +Morph may force abstain/refuse; it never soft-passes a certified singular +exclusivity claim against observed plural HE morphology. +""" + +from __future__ import annotations + +from typing import Sequence + +from core.cognition.fail_closed import ( + CoherenceRefusal, + FailureClass, + ResidualState, +) +from core.semantic_primitives import LogosConstraint, ProvenanceSpan +from generate.observed_he_morph_v0.consumer import ( + ConstraintDecision, + DecisionKind, + apply_he_morph_constraint, +) +from generate.observed_he_morph_v0.records import ( + CanonicalConstraint, + ObservedHebrewSurface, + load_observed_morphology, +) +from generate.observed_he_morph_v0.rules import PLURAL_ABSTAIN_RULE_V0 + +_DEFAULT_PACK = "he_logos_micro_v1" +_CATALOG: tuple[ObservedHebrewSurface, ...] | None = None +_LOAD_ATTEMPTED = False + + +def load_default_catalog() -> tuple[ObservedHebrewSurface, ...] | None: + """Load compiled HE morphology once; None if pack missing (fail closed).""" + global _CATALOG, _LOAD_ATTEMPTED + if _LOAD_ATTEMPTED: + return _CATALOG + _LOAD_ATTEMPTED = True + try: + _CATALOG = load_observed_morphology(_DEFAULT_PACK) + except Exception: + _CATALOG = None + return _CATALOG + + +def reset_catalog_cache() -> None: + """Test-only: clear process-local catalog cache.""" + global _CATALOG, _LOAD_ATTEMPTED + _CATALOG = None + _LOAD_ATTEMPTED = False + + +def scan_observed_he_surface( + text: str, + catalog: Sequence[ObservedHebrewSurface] | None = None, +) -> ObservedHebrewSurface | None: + """Longest exact observed HE surface substring present in ``text``.""" + cat = catalog if catalog is not None else load_default_catalog() + if not cat or not text: + return None + hits = [s for s in cat if s.surface and s.surface in text] + if not hits: + return None + hits.sort(key=lambda s: len(s.surface), reverse=True) + return hits[0] + + +def evaluate_logos_on_text( + *, + text: str, + mode: str = "executable", + catalog: Sequence[ObservedHebrewSurface] | None = None, + lemma_key: str | None = None, + he_surface: str | None = None, +) -> ConstraintDecision | None: + """Evaluate Logos morph authority on free text. + + Returns None when no HE surface is present and mode is not adversarial + (no-op for English-only turns). Otherwise returns the shared + ``ConstraintDecision`` from ``apply_he_morph_constraint``. + """ + cat = catalog if catalog is not None else load_default_catalog() + if cat is None: + if mode.strip().lower() == "adversarial": + return ConstraintDecision( + kind=DecisionKind.FAIL_CLOSED, + reason="morph_catalog_unavailable", + ) + return None + + surface_row = None + if he_surface is not None: + surface = he_surface + else: + surface_row = scan_observed_he_surface(text, cat) + if surface_row is None: + if mode.strip().lower() == "adversarial": + return apply_he_morph_constraint( + proposal_text=text, + lemma_key=lemma_key or "", + observed_catalog=cat, + mode="adversarial", + he_surface=None, + ) + return None + surface = surface_row.surface + + lemma = lemma_key or (surface_row.lemma if surface_row is not None else "") + if not lemma and surface: + # Recover lemma from catalog when only surface was supplied. + from generate.observed_he_morph_v0.records import lookup_surface + + hits = lookup_surface(cat, surface) + if hits: + lemma = hits[0].lemma + + return apply_he_morph_constraint( + proposal_text=text, + lemma_key=lemma or surface, + observed_catalog=cat, + rule=PLURAL_ABSTAIN_RULE_V0, + mode=mode, + he_surface=surface, + ) + + +def logos_blocks_certified_answer(decision: ConstraintDecision | None) -> bool: + """True when morph authority forbids a certified fluent answer.""" + if decision is None: + return False + return decision.kind in (DecisionKind.ABSTAIN, DecisionKind.FAIL_CLOSED) + + +def decision_as_coherence_refusal( + decision: ConstraintDecision, +) -> CoherenceRefusal: + """Map morph ConstraintDecision → typed fail-closed CoherenceRefusal.""" + if decision.kind is DecisionKind.FAIL_CLOSED: + fclass = FailureClass.AMBIGUITY + if "oov" in decision.reason or "missing" in decision.reason: + fclass = FailureClass.MISSING_REFERENT + elif "invalid" in decision.reason: + fclass = FailureClass.CONSTRAINT + return CoherenceRefusal( + failure_class=fclass, + violated_condition=f"logos_morph:{decision.reason}", + residual_state=ResidualState( + detail=decision.reason, + unresolved_hazards=(decision.reason,), + ), + refusal_reason=decision.reason, + surface_message=( + f"Abstaining: Logos morph constraint ({decision.reason})." + ), + ) + # ABSTAIN + return CoherenceRefusal( + failure_class=FailureClass.CONSTRAINT, + violated_condition=f"logos_morph:{decision.reason}", + residual_state=ResidualState( + detail=decision.reason, + unresolved_hazards=(decision.reason,), + ), + refusal_reason=decision.reason, + surface_message=( + f"Abstaining: observed HE morphology blocks claim " + f"({decision.rule_id or decision.reason})." + ), + ) + + +def constraint_to_logos(constraint: CanonicalConstraint) -> LogosConstraint: + """Project live CanonicalConstraint into shared semantic_primitives IR.""" + return constraint.to_logos_constraint() + + +def first_logos_constraint( + decision: ConstraintDecision | None, +) -> LogosConstraint | None: + if decision is None or not decision.constraints: + return None + return decision.constraints[0].to_logos_constraint() + + +__all__ = [ + "constraint_to_logos", + "decision_as_coherence_refusal", + "evaluate_logos_on_text", + "first_logos_constraint", + "load_default_catalog", + "logos_blocks_certified_answer", + "reset_catalog_cache", + "scan_observed_he_surface", +] diff --git a/generate/observed_he_morph_v0/records.py b/generate/observed_he_morph_v0/records.py index 4766d7d8..8401de7d 100644 --- a/generate/observed_he_morph_v0/records.py +++ b/generate/observed_he_morph_v0/records.py @@ -5,7 +5,9 @@ from __future__ import annotations import json from dataclasses import dataclass from pathlib import Path -from typing import Any, Sequence +from typing import Any, Mapping, Sequence + +from core.semantic_primitives import LogosConstraint, ProvenanceSpan, ValidationError _REPO_ROOT = Path(__file__).resolve().parents[2] _DEFAULT_PACK = "he_logos_micro_v0" # may not exist @@ -41,19 +43,88 @@ class ObservedHebrewSurface: @dataclass(frozen=True, slots=True) class CanonicalConstraint: - """Language-independent constraint shared across consumers.""" + """Language-independent constraint with typed provenance (no free meaning dict). + + Fields are explicit; ``payload`` is a *derived read-only view* for + transitional digests only — new code must use typed attributes or + ``to_logos_constraint()``. + """ constraint_id: str kind: str # e.g. plurality_marked - payload: dict[str, Any] + rule_id: str + lemma: str + root: str + surface: str + morphology_id: str + source_pack_id: str + source_span: tuple[int, int] + language: str = "he" + + def __post_init__(self) -> None: + if not self.constraint_id: + raise ValidationError("CanonicalConstraint.constraint_id required") + if not self.kind: + raise ValidationError("CanonicalConstraint.kind required") + if not self.rule_id: + raise ValidationError("CanonicalConstraint.rule_id required") + if not self.morphology_id: + raise ValidationError("CanonicalConstraint.morphology_id required") + if not self.source_pack_id: + raise ValidationError("CanonicalConstraint.source_pack_id required") + if len(self.source_span) != 2: + raise ValidationError("CanonicalConstraint.source_span must be (start, end)") def as_dict(self) -> dict[str, Any]: return { "constraint_id": self.constraint_id, "kind": self.kind, - "payload": dict(self.payload), + "rule_id": self.rule_id, + "lemma": self.lemma, + "root": self.root, + "surface": self.surface, + "morphology_id": self.morphology_id, + "source_pack_id": self.source_pack_id, + "source_span": list(self.source_span), + "language": self.language, } + @property + def payload(self) -> Mapping[str, Any]: + """Derived provenance view (legacy digests / transitional access).""" + return { + "lemma": self.lemma, + "root": self.root, + "surface": self.surface, + "morphology_id": self.morphology_id, + "source_span": list(self.source_span), + "source_pack_id": self.source_pack_id, + "rule_id": self.rule_id, + "language": self.language, + } + + def to_logos_constraint(self) -> LogosConstraint: + return LogosConstraint( + constraint_id=self.constraint_id, + kind=self.kind, + rule_id=self.rule_id, + lemma=self.lemma, + root=self.root, + surface=self.surface, + morphology_id=self.morphology_id, + source_pack_id=self.source_pack_id, + source_span=ProvenanceSpan( + start=int(self.source_span[0]), + end=int(self.source_span[1]), + text=self.surface, + ), + language=self.language, + ) + + @property + def provenance_complete(self) -> bool: + return self.to_logos_constraint().provenance_complete + @dataclass(frozen=True, slots=True) class AuthoredMappingRule: diff --git a/generate/observed_he_morph_v0/rules.py b/generate/observed_he_morph_v0/rules.py index 50c82d68..3a943064 100644 --- a/generate/observed_he_morph_v0/rules.py +++ b/generate/observed_he_morph_v0/rules.py @@ -45,14 +45,14 @@ class PluralAbstainRuleV0(AuthoredMappingRule): 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, - }, + rule_id=self.rule_id, + lemma=surface.lemma, + root=surface.root, + surface=surface.surface, + morphology_id=surface.morphology_id, + source_pack_id=surface.source_pack_id, + source_span=surface.source_span, + language=surface.language, ) diff --git a/teaching/store.py b/teaching/store.py index 051fe2be..4087612a 100644 --- a/teaching/store.py +++ b/teaching/store.py @@ -137,50 +137,20 @@ def _proposal_id(candidate: CorrectionCandidate) -> str: return hashlib.sha256(payload.encode("utf-8")).hexdigest()[:16] -# Cached compiled HE morphology for the live teaching consumer (Stage 4). -_HE_MORPH_CATALOG: tuple | None = None -_HE_MORPH_LOAD_ATTEMPTED = False - - -def _he_morph_catalog(): - """Load compiled HE morphology once; fail closed on missing pack (None).""" - global _HE_MORPH_CATALOG, _HE_MORPH_LOAD_ATTEMPTED - if _HE_MORPH_LOAD_ATTEMPTED: - return _HE_MORPH_CATALOG - _HE_MORPH_LOAD_ATTEMPTED = True - try: - from generate.observed_he_morph_v0.records import load_observed_morphology - - _HE_MORPH_CATALOG = load_observed_morphology("he_logos_micro_v1") - except Exception: - _HE_MORPH_CATALOG = None - return _HE_MORPH_CATALOG - - def _auto_he_morph_decision(proposal: PackMutationProposal): """Apply executable HE morph rule when correction text cites a catalog surface. - Live production path for Stage 4 — not optional test-only wiring. + Live production path — shares ``evaluate_logos_on_text`` with CognitiveTurn + and the four-arm ablation (same pure decision function). Returns None when no HE surface is present (no-op, English corrections unchanged). """ - catalog = _he_morph_catalog() - if not catalog: - return None - text = proposal.correction_text or "" - # Prefer longest surface match present in the correction text. - hits = [s for s in catalog if s.surface and s.surface in text] - if not hits: - return None - hits.sort(key=lambda s: len(s.surface), reverse=True) - surface = hits[0] - from generate.observed_he_morph_v0.consumer import apply_he_morph_constraint + from generate.observed_he_morph_v0.authority import evaluate_logos_on_text - return apply_he_morph_constraint( - proposal_text=text, - lemma_key=str(proposal.subject or surface.lemma), - observed_catalog=catalog, + text = proposal.correction_text or "" + return evaluate_logos_on_text( + text=text, mode="executable", - he_surface=surface.surface, + lemma_key=str(proposal.subject or "") or None, ) diff --git a/tests/test_observed_he_morph_constraint_v0.py b/tests/test_observed_he_morph_constraint_v0.py index 7cfca2a4..9e2fd493 100644 --- a/tests/test_observed_he_morph_constraint_v0.py +++ b/tests/test_observed_he_morph_constraint_v0.py @@ -79,7 +79,15 @@ def test_metadata_only_inert_vs_executable_effect(): 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"] + # Typed IR — no free meaning dict; provenance on fields + LogosConstraint. + c0 = exe.constraints[0] + assert c0.morphology_id + assert c0.source_pack_id + assert c0.rule_id == PLURAL_ABSTAIN_RULE_V0.rule_id + assert c0.provenance_complete is True + logos = c0.to_logos_constraint() + assert logos.constraint_id == c0.constraint_id + assert logos.provenance_complete is True def test_oov_and_invalid_fail_closed(): @@ -156,3 +164,77 @@ def test_vault_promotion_default_requires_geometric_unitarity(): raw=0.05, energy_class=EnergyClass.E0, coherence_residual=1e-9 ) assert policy.decide(tight).promote is True + + +def test_shared_authority_four_modes_and_typed_ir(): + """Live authority entry shares apply_he_morph_constraint; modes diverge correctly.""" + from generate.observed_he_morph_v0.authority import ( + evaluate_logos_on_text, + first_logos_constraint, + logos_blocks_certified_answer, + ) + + catalog = load_observed_morphology("he_logos_micro_v1") + plural = next(s for s in catalog if s.number == "plural") + claim = ( + f"{plural.surface} ({plural.lemma}) must be singular only — " + f"exclusive singular identity" + ) + can = evaluate_logos_on_text(text=claim, mode="canonical", catalog=catalog) + meta = evaluate_logos_on_text(text=claim, mode="metadata", catalog=catalog) + exe = evaluate_logos_on_text(text=claim, mode="executable", catalog=catalog) + oov = evaluate_logos_on_text( + text="zzz_not_a_surface must be singular only", + mode="adversarial", + catalog=catalog, + he_surface="zzz_not_a_surface", + ) + assert can is not None and meta is not None and exe is not None and oov is not None + assert can.kind is DecisionKind.PASS + assert meta.kind is DecisionKind.PASS + assert can.as_dict() == meta.as_dict() + assert exe.kind is DecisionKind.ABSTAIN + assert logos_blocks_certified_answer(exe) is True + assert oov.kind is DecisionKind.FAIL_CLOSED + lc = first_logos_constraint(exe) + assert lc is not None + assert lc.rule_id == PLURAL_ABSTAIN_RULE_V0.rule_id + assert lc.source_pack_id == "he_logos_micro_v1" + assert lc.morphology_id == plural.morphology_id + + +def test_cognitive_turn_logos_authority_abstains_on_plural_singular_claim(): + """Live CognitiveTurnPipeline.run path: executable morph blocks certified answer.""" + from chat.runtime import ChatRuntime + from core.cognition.pipeline import CognitiveTurnPipeline + + catalog = load_observed_morphology("he_logos_micro_v1") + plural = next(s for s in catalog if s.number == "plural") + claim = ( + f"{plural.surface} ({plural.lemma}) must be singular only — " + f"exclusive singular identity" + ) + pipe = CognitiveTurnPipeline(runtime=ChatRuntime()) + result = pipe.run(claim) + assert result.logos_decision_kind == "abstain" + assert result.logos_decision_reason == "plural_morph_blocks_singular_exclusivity" + assert result.logos_rule_id == PLURAL_ABSTAIN_RULE_V0.rule_id + assert result.logos_constraint_id + assert result.authority_source == "logos_morph_constraint" + assert "Abstaining" in result.surface or "abstain" in result.surface.lower() + assert result.refusal_reason == "plural_morph_blocks_singular_exclusivity" + # Hazard ledger records morph block (improvement observability). + assert any("logos_morph" in h for h in result.substrate_hazard) + + +def test_cognitive_turn_english_only_no_logos_noop(): + """English-only turn does not invent HE morph or force logos authority.""" + from chat.runtime import ChatRuntime + from core.cognition.pipeline import CognitiveTurnPipeline + + pipe = CognitiveTurnPipeline(runtime=ChatRuntime()) + result = pipe.run("What is light?") + assert result.logos_decision_kind in ("", "pass") + assert result.authority_source != "logos_morph_constraint" + # Must not be a morph abstention surface when no HE present. + assert "Logos morph" not in result.surface