feat(logos): bulk live morph authority on turn + teaching seams

Wire observed-HE morph constraint into CognitiveTurn answer path and
unify teaching store with the same pure decision function used by the
four-arm ablation. Typed LogosConstraint IR carries rule/morph/pack/span
provenance without free meaning dicts. Executable mode abstains on
plural-vs-singular exclusivity; metadata remains bit-identical to
canonical; adversarial OOV/missing fails closed. Residual dual-system
debt documented honestly (cue-table linguistic_pipeline still parallel).

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,
         )
 
     # ------------------------------------------------------------------

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 = ""

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",

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."""

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.

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
+Compiled pack data only; no English-to-Hebrew pseudo-morphology.
-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",
 ]

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
+        c0 = d_exec.constraints[0]
-        provenance_ok = (
+        provenance_ok = bool(
-            "source_span" in payload
+            getattr(c0, "provenance_complete", False)
-            and "morphology_id" in payload
+            or (
-            and "source_pack_id" in payload
+                c0.morphology_id
+                and c0.source_pack_id
+                and c0.rule_id
+                and c0.source_span
+            )
         )
 
     return AblationReport(

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",
+]

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:

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={
+            rule_id=self.rule_id,
-                "lemma": surface.lemma,
+            lemma=surface.lemma,
-                "root": surface.root,
+            root=surface.root,
-                "surface": surface.surface,
+            surface=surface.surface,
-                "morphology_id": surface.morphology_id,
+            morphology_id=surface.morphology_id,
-                "source_span": list(surface.source_span),
+            source_pack_id=surface.source_pack_id,
-                "source_pack_id": surface.source_pack_id,
+            source_span=surface.source_span,
-            },
+            language=surface.language,
         )
 
 

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()
+    from generate.observed_he_morph_v0.authority import evaluate_logos_on_text
-    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
 
-    return apply_he_morph_constraint(
+    text = proposal.correction_text or ""
-        proposal_text=text,
+    return evaluate_logos_on_text(
-        lemma_key=str(proposal.subject or surface.lemma),
+        text=text,
-        observed_catalog=catalog,
         mode="executable",
-        he_surface=surface.surface,
+        lemma_key=str(proposal.subject or "") or None,
     )
 
 

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