diff --git a/core/cognition/inductive_closure.py b/core/cognition/inductive_closure.py index f18d941b..8014eedd 100644 --- a/core/cognition/inductive_closure.py +++ b/core/cognition/inductive_closure.py @@ -1,22 +1,26 @@ """Bounded multi-step inductive closure over teaching-store relations (Stage 3C). Fixed-point expansion of same-relation chains with explicit budgets, -cycle handling, contradiction detection, and replayable provenance. +cycle handling, contradiction detection, geometric admissibility, and +replayable provenance. Atom *identity* for entailment telemetry remains conformal (``CognitiveTurnPipeline._proof_atom``). This module expands the *relation graph* of surface triples from ``TeachingStore.triples()`` into a closed set under transitive composition of equal relation labels — not string -atom join as final authority for field identity. +atom join as final identity authority. """ from __future__ import annotations from dataclasses import dataclass -from typing import Any, Iterable, Sequence +from typing import Any, Callable, Iterable, Sequence _DEFAULT_BUDGET = 16 +# geometric_admissible(head, relation, tail) -> bool +GeometricAdmissibleFn = Callable[[str, str, str], bool] + def _norm(token: str) -> str: return token.strip().lower() @@ -71,14 +75,27 @@ class InductiveClosureResult: "fixed_point": self.fixed_point, "truncated": self.truncated, "n_derived": len(self.derived), + "n_admissible_derived": sum(1 for d in self.derived if d.admissible), } +def default_geometric_admissible(head: str, relation: str, tail: str) -> bool: + """Strict default: a derived relation is admissible only if endpoints are + non-empty distinct tokens and the relation is non-empty. Callers that + have a vocab/field resolver should pass a stronger callback that checks + closed Cl(4,1) versors (see pipeline wiring). + """ + del relation + h, t = head.strip(), tail.strip() + return bool(h) and bool(t) and h != t + + def expand_relation_closure( triples: Sequence[tuple[str, str, str]], *, budget: int = _DEFAULT_BUDGET, relations: Iterable[str] | None = None, + geometric_admissible: GeometricAdmissibleFn | None = None, ) -> InductiveClosureResult: """Compute same-relation transitive closure with budget and contradictions. @@ -88,22 +105,20 @@ def expand_relation_closure( derive (a,r,c) with path a…c. * Cycle: if path would revisit a node, skip (no infinite loop). * Contradiction: two different tails for the same (head, relation) - at the same fixed-point layer mark both as contradiction=True - (functional assumption for same-relation edges). + among base facts mark contradiction=True. + * Geometric admissibility: each *derived* relation is stamped + admissible only when ``geometric_admissible(h,r,t)`` is True. + Default requires non-empty distinct endpoints; pipeline supplies + versor-grounded checks when session vocab is available. * Termination: no new triples or budget exhausted. - - Geometric admissibility of *field* atoms is enforced by callers that - map surfaces through ``_proof_atom`` before using derived triples as - proof premises; this expander is total over the teaching-store graph. """ if budget < 1: raise ValueError("budget must be >= 1") + geom = geometric_admissible or default_geometric_admissible rel_filter = None if relations is None else {_norm(r) for r in relations} - # Base base_list: list[DerivedRelation] = [] - # key (h,r) -> set of tails for contradiction detection edge_map: dict[tuple[str, str], set[str]] = {} known: set[tuple[str, str, str]] = set() @@ -118,6 +133,7 @@ def expand_relation_closure( continue known.add(key3) edge_map.setdefault((hn, rn), set()).add(tn) + # Base facts are store-grounded; admissible if geometric check passes. base_list.append( DerivedRelation( head=hn, @@ -125,11 +141,10 @@ def expand_relation_closure( tail=tn, path=(hn, tn), step=0, - admissible=True, + admissible=bool(geom(hn, rn, tn)), ) ) - # Mark base contradictions contradictions: list[DerivedRelation] = [] for (h, r), tails in edge_map.items(): if len(tails) > 1: @@ -148,7 +163,6 @@ def expand_relation_closure( ) derived: list[DerivedRelation] = [] - # Working set of edges as (h,r,t) for composition work = set(known) steps_taken = 0 fixed_point = False @@ -157,7 +171,6 @@ def expand_relation_closure( for step in range(1, budget + 1): steps_taken = step new_edges: list[DerivedRelation] = [] - # Index tails by (h,r) by_hr: dict[tuple[str, str], list[str]] = {} for h, r, t in work: by_hr.setdefault((h, r), []).append(t) @@ -166,12 +179,12 @@ def expand_relation_closure( for b in mids: for c in by_hr.get((b, r), ()): if a == c: - continue # cycle / identity + continue key3 = (a, r, c) if key3 in work: continue - # path reconstruction (bounded) path = (a, b, c) + ok = bool(geom(a, r, c)) new_edges.append( DerivedRelation( head=a, @@ -179,7 +192,8 @@ def expand_relation_closure( tail=c, path=path, step=step, - admissible=True, + admissible=ok, + contradiction=False, ) ) @@ -188,9 +202,6 @@ def expand_relation_closure( steps_taken = step - 1 if step > 0 else 0 break - # Dedup new edges. Transitive multi-tails for the same (head, relation) - # are *not* contradictions — only base multi-tails (step 0) mark - # functional conflicts (recorded once above). for dr in new_edges: key3 = dr.as_triple() if key3 in work: @@ -199,9 +210,7 @@ def expand_relation_closure( edge_map.setdefault((dr.head, dr.relation), set()).add(dr.tail) derived.append(dr) else: - # Budget exhausted without fixed point truncated = True - # Check if more edges would exist by_hr = {} for h, r, t in work: by_hr.setdefault((h, r), []).append(t) diff --git a/core/cognition/pipeline.py b/core/cognition/pipeline.py index b2477d14..e7e30af6 100644 --- a/core/cognition/pipeline.py +++ b/core/cognition/pipeline.py @@ -442,8 +442,25 @@ class CognitiveTurnPipeline: compose_surface = CognitiveTurnPipeline._render_compose_surface(compose_result) # Stage 3C — bounded inductive closure over teaching-store relations. - # Provenance-preserving fixed-point; folded into operator_invocation only. - inductive_closure = expand_relation_closure(triples, budget=16) + # Provenance-preserving fixed-point; derived edges require geometric + # admissibility (closed Cl(4,1) versors when vocab can ground them). + def _geom_admissible(h: str, r: str, t: str) -> bool: + del r + hv = self._resolve_surface_versor(h) + tv = self._resolve_surface_versor(t) + if hv is None or tv is None: + # Ungrounded endpoints cannot be promoted as geometric facts. + return False + return ( + float(versor_condition(hv)) < 1e-6 + and float(versor_condition(tv)) < 1e-6 + ) + + inductive_closure = expand_relation_closure( + triples, + budget=16, + geometric_admissible=_geom_admissible, + ) entailment_trace = self._maybe_entailment_trace(intent, triples) diff --git a/core/physics/learning.py b/core/physics/learning.py index 9e6071e1..b0fd2845 100644 --- a/core/physics/learning.py +++ b/core/physics/learning.py @@ -15,9 +15,16 @@ class PromotionDecision: class VaultPromotionPolicy: - """Promote only settled, coherent regions into deep vault storage.""" + """Promote only settled, *geometrically* coherent regions to COHERENT. - def __init__(self, residual_threshold: float = 0.05) -> None: + Stage 3A / Master Blueprint: COHERENT standing requires the unitary + residual condition (``coherence_residual ≤ 1e-6`` by default), not a + soft energy-band threshold alone. Tests may pass a looser + ``residual_threshold`` explicitly for energy-class isolation fixtures; + production ``VaultPromotionPolicy()`` uses the geometric floor. + """ + + def __init__(self, residual_threshold: float = 1e-6) -> None: if residual_threshold < 0.0: raise ValueError("residual_threshold must be non-negative") self.residual_threshold = float(residual_threshold) @@ -27,6 +34,13 @@ class VaultPromotionPolicy: return PromotionDecision(False, "missing_energy_profile", EnergyClass.E2) if not energy.energy_class.vault_candidate: return PromotionDecision(False, "region_still_active", energy.energy_class) + # Full geometric unitarity gate for COHERENT promotion (Stage 3A). if energy.coherence_residual > self.residual_threshold: - return PromotionDecision(False, "coherence_residual_above_threshold", energy.energy_class) + return PromotionDecision( + False, "coherence_residual_above_threshold", energy.energy_class + ) + # Residual must also be finite and non-negative (typed safety). + r = float(energy.coherence_residual) + if not (r == r) or r < 0.0: # NaN or negative + return PromotionDecision(False, "coherence_residual_invalid", energy.energy_class) return PromotionDecision(True, "settled_coherent_region", energy.energy_class) diff --git a/generate/observed_he_morph_v0/ablation.py b/generate/observed_he_morph_v0/ablation.py index 8fb3acc5..3cb9331d 100644 --- a/generate/observed_he_morph_v0/ablation.py +++ b/generate/observed_he_morph_v0/ablation.py @@ -130,9 +130,9 @@ def run_four_arm_ablation( ) ) - # Metrics - # Metadata decision kind must match canonical (both PASS). - meta_identical = d_meta.kind is d_can.kind is DecisionKind.PASS + # Metrics — Stage 4 requires metadata bit-identical to canonical baseline + # (full SHA-256 of decision payload, not kind-only equality). + meta_identical = _digest(d_meta) == _digest(d_can) and d_meta.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 diff --git a/generate/observed_he_morph_v0/consumer.py b/generate/observed_he_morph_v0/consumer.py index e43fa801..c02f5700 100644 --- a/generate/observed_he_morph_v0/consumer.py +++ b/generate/observed_he_morph_v0/consumer.py @@ -61,8 +61,15 @@ def apply_he_morph_constraint( ``he_surface`` is the observed HE surface form when present. """ mode = mode.strip().lower() + # Canonical baseline decision (no morph authority). Metadata-only must + # return this *exact* decision object shape so digests are bit-identical + # to baseline (Stage 4 sealed harness). Morph provenance is tracked by + # the ablation harness separately, never folded into the decision payload + # on the metadata arm. + baseline = ConstraintDecision(kind=DecisionKind.PASS, reason="canonical_baseline") + if mode == "canonical": - return ConstraintDecision(kind=DecisionKind.PASS, reason="canonical_baseline") + return baseline if not he_surface: if mode == "adversarial": @@ -70,7 +77,8 @@ def apply_he_morph_constraint( kind=DecisionKind.FAIL_CLOSED, reason="missing_he_surface", ) - return ConstraintDecision(kind=DecisionKind.PASS, reason="no_he_surface") + # No HE surface → identical to baseline (including metadata arm). + return baseline hits = lookup_surface(observed_catalog, he_surface) if hits is None: @@ -111,12 +119,8 @@ def apply_he_morph_constraint( ) if mode == "metadata": - # Morph present for provenance only — decision identical to baseline. - return ConstraintDecision( - kind=DecisionKind.PASS, - reason="metadata_only_inert", - surfaces=(surface,), - ) + # Morph observed but must not change the consumer decision vs baseline. + return baseline if mode == "adversarial" and "INVALID" in he_surface: return ConstraintDecision( @@ -127,12 +131,7 @@ def apply_he_morph_constraint( # Executable rule arm if not rule.matches(surface): - return ConstraintDecision( - kind=DecisionKind.PASS, - reason="rule_preconditions_unmet", - surfaces=(surface,), - rule_id=rule.rule_id, - ) + return baseline constraint = rule.to_constraint(surface) # Abstain when proposal asserts exclusive singular / singular-only claim @@ -151,10 +150,5 @@ def apply_he_morph_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, - ) + # Rule matched but no singular-exclusivity conflict → same decision as baseline. + return baseline diff --git a/teaching/store.py b/teaching/store.py index 82f55228..051fe2be 100644 --- a/teaching/store.py +++ b/teaching/store.py @@ -137,6 +137,53 @@ 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. + 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 + + return apply_he_morph_constraint( + proposal_text=text, + lemma_key=str(proposal.subject or surface.lemma), + observed_catalog=catalog, + mode="executable", + he_surface=surface.surface, + ) + + class TeachingStore: """Bounded, append-only store for reviewed teaching examples. @@ -204,7 +251,10 @@ class TeachingStore: epistemic_status=example.epistemic_status, ) - # Stage 4 HE morph constraint — abstain/fail-closed → CONTESTED. + # Stage 4 HE morph constraint — live path auto-applies executable rule + # from compiled packs/data morphology (not tests-only optional wiring). + if he_morph_decision is None: + he_morph_decision = _auto_he_morph_decision(proposal) if he_morph_decision is not None: kind = getattr(he_morph_decision, "kind", None) kind_val = getattr(kind, "value", kind) diff --git a/tests/test_observed_he_morph_constraint_v0.py b/tests/test_observed_he_morph_constraint_v0.py index 0daad3b2..7cfca2a4 100644 --- a/tests/test_observed_he_morph_constraint_v0.py +++ b/tests/test_observed_he_morph_constraint_v0.py @@ -41,9 +41,19 @@ def test_four_arm_ablation_sealed_metrics(): def test_metadata_only_inert_vs_executable_effect(): + import hashlib + import json + 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" + can = apply_he_morph_constraint( + proposal_text=claim, + lemma_key=plural.lemma, + observed_catalog=catalog, + mode="canonical", + he_surface=plural.surface, + ) meta = apply_he_morph_constraint( proposal_text=claim, lemma_key=plural.lemma, @@ -58,6 +68,14 @@ def test_metadata_only_inert_vs_executable_effect(): mode="executable", he_surface=plural.surface, ) + # Bit-identical decision payloads (Stage 4 sealed requirement). + d_can = hashlib.sha256( + json.dumps(can.as_dict(), sort_keys=True).encode() + ).hexdigest() + d_meta = hashlib.sha256( + json.dumps(meta.as_dict(), sort_keys=True).encode() + ).hexdigest() + assert d_can == d_meta assert meta.kind is DecisionKind.PASS assert exe.kind is DecisionKind.ABSTAIN assert exe.rule_id == PLURAL_ABSTAIN_RULE_V0.rule_id @@ -78,12 +96,11 @@ def test_oov_and_invalid_fail_closed(): def test_teaching_store_consumer_seam_abstains_on_plural_rule(): - """Live TeachingStore.add path: HE plural rule forces CONTESTED (abstain).""" + """Live TeachingStore.add path: HE plural rule auto-applies (no test-only kwarg).""" 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), @@ -100,18 +117,13 @@ def test_teaching_store_consumer_seam_abstains_on_plural_rule(): 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 + # Correction text includes the observed HE plural surface so the live + # auto-path in TeachingStore.add finds the catalog row and abstains. cand2 = CorrectionCandidate( - correction_text=f"{plural.lemma} must be singular only — exclusive singular identity", + correction_text=( + f"{plural.surface} ({plural.lemma}) must be singular only — " + f"exclusive singular identity" + ), intent=DialogueIntent(tag=IntentTag.CORRECTION, subject=plural.lemma), prior_surface="prior", prior_turn=1, @@ -123,6 +135,24 @@ def test_teaching_store_consumer_seam_abstains_on_plural_rule(): review_hash="h2", epistemic_status=EpistemicStatus.SPECULATIVE, ) - p2 = store.add(rev2, he_morph_decision=decision) + # No he_morph_decision kwarg — production auto-path must fire. + p2 = store.add(rev2) assert p2 is not None assert p2.epistemic_status is EpistemicStatus.CONTESTED + + +def test_vault_promotion_default_requires_geometric_unitarity(): + """Production VaultPromotionPolicy() uses 1e-6 residual, not soft 0.05.""" + from core.physics.learning import VaultPromotionPolicy + from core.physics.energy import EnergyClass, EnergyProfile + + policy = VaultPromotionPolicy() # production default + assert policy.residual_threshold <= 1e-6 + soft = EnergyProfile( + raw=0.05, energy_class=EnergyClass.E0, coherence_residual=0.02 + ) + assert policy.decide(soft).promote is False + tight = EnergyProfile( + raw=0.05, energy_class=EnergyClass.E0, coherence_residual=1e-9 + ) + assert policy.decide(tight).promote is True diff --git a/tests/test_stage3_epistemic_inductive.py b/tests/test_stage3_epistemic_inductive.py index 69d0cf13..04f07bb2 100644 --- a/tests/test_stage3_epistemic_inductive.py +++ b/tests/test_stage3_epistemic_inductive.py @@ -62,15 +62,35 @@ def test_inductive_closure_derives_two_hop(): ("a", "is", "b"), ("b", "is", "c"), ) - res = expand_relation_closure(triples, budget=8) + # Explicit geometric_admissible always-true for pure graph composition tests. + res = expand_relation_closure( + triples, budget=8, geometric_admissible=lambda h, r, t: True + ) assert len(res.base) == 2 derived_tails = {(d.head, d.relation, d.tail) for d in res.derived} assert ("a", "is", "c") in derived_tails assert res.fixed_point is True assert res.steps_taken >= 1 - # Provenance path a_to_c = next(d for d in res.derived if d.head == "a" and d.tail == "c") assert a_to_c.path[0] == "a" and a_to_c.path[-1] == "c" + assert a_to_c.admissible is True + + +def test_inductive_derived_requires_geometric_admissibility(): + triples = ( + ("a", "is", "b"), + ("b", "is", "c"), + ) + + def refuse_all(h, r, t): + del h, r, t + return False + + res = expand_relation_closure( + triples, budget=8, geometric_admissible=refuse_all + ) + assert any(d.head == "a" and d.tail == "c" for d in res.derived) + assert all(d.admissible is False for d in res.derived) def test_inductive_closure_detects_contradiction(): @@ -78,7 +98,9 @@ def test_inductive_closure_detects_contradiction(): ("a", "is", "b"), ("a", "is", "c"), ) - res = expand_relation_closure(triples, budget=4) + res = expand_relation_closure( + triples, budget=4, geometric_admissible=lambda h, r, t: True + ) assert any(c.contradiction for c in res.contradictions) assert len(res.contradictions) >= 2 @@ -86,10 +108,10 @@ def test_inductive_closure_detects_contradiction(): def test_inductive_closure_budget_truncation(): # Long chain: a0->a1->...->a20 triples = tuple((f"a{i}", "r", f"a{i+1}") for i in range(20)) - res = expand_relation_closure(triples, budget=2) + ok = lambda h, r, t: True # noqa: E731 + res = expand_relation_closure(triples, budget=2, geometric_admissible=ok) assert res.truncated or res.steps_taken <= 2 - # With larger budget, multi-hop appears - res2 = expand_relation_closure(triples, budget=16) + res2 = expand_relation_closure(triples, budget=16, geometric_admissible=ok) assert any(d.head == "a0" and d.tail == "a2" for d in res2.derived) or any( d.head == "a0" for d in res2.derived ) @@ -100,7 +122,9 @@ def test_inductive_closure_cycle_safe(): ("a", "r", "b"), ("b", "r", "a"), ) - res = expand_relation_closure(triples, budget=8) + res = expand_relation_closure( + triples, budget=8, geometric_admissible=lambda h, r, t: True + ) # Must terminate without inventing infinite chain assert res.fixed_point or res.steps_taken <= 8 assert all(len(d.path) < 20 for d in res.derived)