diff --git a/core/cognition/geometric_coherence.py b/core/cognition/geometric_coherence.py new file mode 100644 index 00000000..18510f81 --- /dev/null +++ b/core/cognition/geometric_coherence.py @@ -0,0 +1,118 @@ +"""Turn-level geometric coherence verdict (Master Blueprint Stage 3A). + +Ownership model (deliberate dual taxonomy — do NOT collapse names): + +* ``teaching.epistemic.EpistemicStatus`` — vault / pack durable standing + (SPECULATIVE | COHERENT | CONTESTED | FALSIFIED). Mutation only via + ``vault/store.py`` (INV-29). +* ``core.epistemic_state.EpistemicState`` — turn / surface taxonomy for + dialogue observability (perceived, verified, decoded, …). **No** member + named COHERENT — vault COHERENT maps to DECODED via + ``epistemic_state_for_vault_status``. + +This module adds a third, geometry-native axis: whether the *field* closed +under versor + GoldTether residual checks on this turn. It never renames +EpistemicState and never stamps vault COHERENT by itself. +""" + +from __future__ import annotations + +from dataclasses import dataclass +from enum import Enum +from typing import Any + +import numpy as np + +from algebra.backend import versor_condition +from core.physics.goldtether import coherence_residual + +_CLOSURE = 1e-6 + + +class GeometricCoherenceStatus(str, Enum): + """Turn-level geometric standing — orthogonal to vault EpistemicStatus.""" + + GEOMETRICALLY_VERIFIED = "geometrically_verified" + UNVERIFIED = "unverified" + REFUSED = "refused" + + +@dataclass(frozen=True, slots=True) +class GeometricCoherenceVerdict: + """Pipeline-visible geometric coherence for one turn. + + ``GEOMETRICALLY_VERIFIED`` requires: + * field present + * versor_condition(F) < 1e-6 + * R_GoldTether ≤ 1e-6 + Identity leakage flags are observational while identity_wave_gate is off + (ADR-0244 honest scope) and do not alone refuse this verdict unless + ``boundary_violations`` is non-empty. + """ + + status: GeometricCoherenceStatus + versor_condition: float + goldtether_residual: float + field_present: bool + identity_boundary_breach: bool = False + detail: str = "" + + @property + def closed(self) -> bool: + return self.status is GeometricCoherenceStatus.GEOMETRICALLY_VERIFIED + + def as_dict(self) -> dict[str, Any]: + return { + "status": self.status.value, + "closed": self.closed, + "versor_condition": float(self.versor_condition), + "goldtether_residual": float(self.goldtether_residual), + "field_present": bool(self.field_present), + "identity_boundary_breach": bool(self.identity_boundary_breach), + "detail": self.detail, + } + + +def evaluate_geometric_coherence( + F, + *, + identity_score=None, + epsilon: float = _CLOSURE, +) -> GeometricCoherenceVerdict: + """Compute turn geometric coherence from the live field versor.""" + if F is None: + return GeometricCoherenceVerdict( + status=GeometricCoherenceStatus.REFUSED, + versor_condition=float("inf"), + goldtether_residual=float("inf"), + field_present=False, + detail="missing_wave_field", + ) + arr = np.asarray(F, dtype=np.float64) + vc = float(versor_condition(arr)) + r_gt = float(coherence_residual(arr)) + boundary = bool(getattr(identity_score, "boundary_violations", ()) or ()) + if boundary: + return GeometricCoherenceVerdict( + status=GeometricCoherenceStatus.REFUSED, + versor_condition=vc, + goldtether_residual=r_gt, + field_present=True, + identity_boundary_breach=True, + detail="identity_boundary_breach", + ) + if vc >= float(epsilon) or r_gt > float(epsilon): + return GeometricCoherenceVerdict( + status=GeometricCoherenceStatus.UNVERIFIED, + versor_condition=vc, + goldtether_residual=r_gt, + field_present=True, + detail=f"versor_condition={vc:.3e}; R_GoldTether={r_gt:.3e}", + ) + return GeometricCoherenceVerdict( + status=GeometricCoherenceStatus.GEOMETRICALLY_VERIFIED, + versor_condition=vc, + goldtether_residual=r_gt, + field_present=True, + detail="versor_and_goldtether_closed", + ) diff --git a/core/cognition/inductive_closure.py b/core/cognition/inductive_closure.py new file mode 100644 index 00000000..f18d941b --- /dev/null +++ b/core/cognition/inductive_closure.py @@ -0,0 +1,223 @@ +"""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. + +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. +""" + +from __future__ import annotations + +from dataclasses import dataclass +from typing import Any, Iterable, Sequence + +_DEFAULT_BUDGET = 16 + + +def _norm(token: str) -> str: + return token.strip().lower() + + +@dataclass(frozen=True, slots=True) +class DerivedRelation: + """One promoted (or base) triple with provenance path.""" + + head: str + relation: str + tail: str + path: tuple[str, ...] # entity path head … tail + step: int # 0 = base fact from store; >0 = derived at fixed-point step + admissible: bool = True + contradiction: bool = False + + def as_triple(self) -> tuple[str, str, str]: + return (self.head, self.relation, self.tail) + + def as_dict(self) -> dict[str, Any]: + return { + "head": self.head, + "relation": self.relation, + "tail": self.tail, + "path": list(self.path), + "step": self.step, + "admissible": self.admissible, + "contradiction": self.contradiction, + } + + +@dataclass(frozen=True, slots=True) +class InductiveClosureResult: + """Result of bounded fixed-point expansion.""" + + base: tuple[DerivedRelation, ...] + derived: tuple[DerivedRelation, ...] + contradictions: tuple[DerivedRelation, ...] + steps_taken: int + budget: int + fixed_point: bool + truncated: bool + + def as_dict(self) -> dict[str, Any]: + return { + "base": [r.as_dict() for r in self.base], + "derived": [r.as_dict() for r in self.derived], + "contradictions": [r.as_dict() for r in self.contradictions], + "steps_taken": self.steps_taken, + "budget": self.budget, + "fixed_point": self.fixed_point, + "truncated": self.truncated, + "n_derived": len(self.derived), + } + + +def expand_relation_closure( + triples: Sequence[tuple[str, str, str]], + *, + budget: int = _DEFAULT_BUDGET, + relations: Iterable[str] | None = None, +) -> InductiveClosureResult: + """Compute same-relation transitive closure with budget and contradictions. + + Rules: + * Base facts: each input triple (normalized). + * Step k+1: if (a,r,b) and (b,r,c) known and a≠c and (a,r,c) unknown, + 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). + * 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") + + 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() + + for h, r, t in triples: + hn, rn, tn = _norm(h), _norm(r), _norm(t) + if not hn or not rn or not tn: + continue + if rel_filter is not None and rn not in rel_filter: + continue + key3 = (hn, rn, tn) + if key3 in known: + continue + known.add(key3) + edge_map.setdefault((hn, rn), set()).add(tn) + base_list.append( + DerivedRelation( + head=hn, + relation=rn, + tail=tn, + path=(hn, tn), + step=0, + admissible=True, + ) + ) + + # Mark base contradictions + contradictions: list[DerivedRelation] = [] + for (h, r), tails in edge_map.items(): + if len(tails) > 1: + for dr in base_list: + if dr.head == h and dr.relation == r: + contradictions.append( + DerivedRelation( + head=dr.head, + relation=dr.relation, + tail=dr.tail, + path=dr.path, + step=0, + admissible=False, + contradiction=True, + ) + ) + + derived: list[DerivedRelation] = [] + # Working set of edges as (h,r,t) for composition + work = set(known) + steps_taken = 0 + fixed_point = False + truncated = False + + 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) + + for (a, r), mids in by_hr.items(): + for b in mids: + for c in by_hr.get((b, r), ()): + if a == c: + continue # cycle / identity + key3 = (a, r, c) + if key3 in work: + continue + # path reconstruction (bounded) + path = (a, b, c) + new_edges.append( + DerivedRelation( + head=a, + relation=r, + tail=c, + path=path, + step=step, + admissible=True, + ) + ) + + if not new_edges: + fixed_point = True + 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: + continue + work.add(key3) + 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) + for (a, r), mids in by_hr.items(): + for b in mids: + for c in by_hr.get((b, r), ()): + if a != c and (a, r, c) not in work: + truncated = True + break + + return InductiveClosureResult( + base=tuple(base_list), + derived=tuple(derived), + contradictions=tuple(contradictions), + steps_taken=steps_taken, + budget=budget, + fixed_point=fixed_point and not truncated, + truncated=truncated, + ) diff --git a/core/cognition/pipeline.py b/core/cognition/pipeline.py index 325e2a4e..b2477d14 100644 --- a/core/cognition/pipeline.py +++ b/core/cognition/pipeline.py @@ -24,6 +24,8 @@ import numpy as np from algebra.backend import versor_condition from algebra.cl41 import geometric_product, reverse, scalar_part from field.state import FieldState +from core.cognition.geometric_coherence import evaluate_geometric_coherence +from core.cognition.inductive_closure import expand_relation_closure from core.cognition.leeway import build_leeway_record from core.cognition.result import CognitiveTurnResult from core.cognition.surface_resolution import resolve_surface @@ -439,6 +441,10 @@ 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) + entailment_trace = self._maybe_entailment_trace(intent, triples) # === SHADOW COHERENCE GATE WIRING === @@ -645,10 +651,34 @@ class CognitiveTurnPipeline: entailment_serialised = CognitiveTurnPipeline._serialize_entailment_trace( entailment_trace ) - # Deterministic concatenation: walk, compose, then entailment. Empty - # strings are dropped so unaffected turns keep existing trace bytes. + # Stage 3C — inductive fixed-point provenance (before hash so replay + # includes multi-step derivation when teaching store has chains). + inductive_serialised = "" + if inductive_closure.derived or inductive_closure.contradictions: + inductive_serialised = json.dumps( + { + "inductive_closure": { + "n_derived": len(inductive_closure.derived), + "n_contradictions": len(inductive_closure.contradictions), + "steps_taken": inductive_closure.steps_taken, + "fixed_point": inductive_closure.fixed_point, + "truncated": inductive_closure.truncated, + "derived": [d.as_dict() for d in inductive_closure.derived[:8]], + } + }, + sort_keys=True, + ensure_ascii=False, + ) + # Deterministic concatenation: walk, compose, entailment, inductive. + # Empty strings are dropped so unaffected turns keep existing trace bytes. operator_invocation = "|".join( - s for s in (walk_serialised, compose_serialised, entailment_serialised) + s + for s in ( + walk_serialised, + compose_serialised, + entailment_serialised, + inductive_serialised, + ) if s ) # ADR-0023 — admissibility trace + ratification provenance. @@ -724,6 +754,15 @@ class CognitiveTurnPipeline: license_decision=getattr(accrual, "license", None), ) + # Stage 3A — turn-level geometric coherence (orthogonal to vault COHERENT). + geo_F = F_gate + if geo_F is None and field_state_after is not None: + geo_F = getattr(field_state_after, "F", None) + geometric_coherence = evaluate_geometric_coherence( + geo_F, + identity_score=response.identity_score, + ) + return CognitiveTurnResult( input_text=text, input_tokens=raw_tokens, @@ -762,6 +801,7 @@ class CognitiveTurnPipeline: dispatch_trace=getattr(response, "dispatch_trace", None), dropped_compound_clauses=dropped_compound_clauses, versor_condition=response.versor_condition, + geometric_coherence=geometric_coherence, trace_hash=trace_hash, leeway=leeway, # Phase A — Shadow Coherence Gate observability. diff --git a/core/cognition/result.py b/core/cognition/result.py index 751dd5b6..55bd599a 100644 --- a/core/cognition/result.py +++ b/core/cognition/result.py @@ -10,6 +10,7 @@ from __future__ import annotations from dataclasses import dataclass +from core.cognition.geometric_coherence import GeometricCoherenceVerdict from core.cognition.leeway import LeewayRecord from field.state import FieldState from generate.articulation import ArticulationPlan @@ -150,6 +151,9 @@ class CognitiveTurnResult: # --- invariant bookkeeping --- versor_condition: float = 0.0 # must be < 1e-6 + # Stage 3A — geometry-native turn coherence (orthogonal to vault EpistemicStatus). + # None only on pre-Stage-3 artifacts; live pipeline always populates. + geometric_coherence: GeometricCoherenceVerdict | None = None trace_hash: str = "" # SHA-256 over deterministic key fields # --- response-governance leeway evidence (B4; observational, not in trace_hash) --- diff --git a/docs/adr/epistemic-taxonomy-ownership-stage3.md b/docs/adr/epistemic-taxonomy-ownership-stage3.md new file mode 100644 index 00000000..8a15fa54 --- /dev/null +++ b/docs/adr/epistemic-taxonomy-ownership-stage3.md @@ -0,0 +1,26 @@ +# Epistemic taxonomy ownership (Master Blueprint Stage 3A) + +**Status**: Binding ownership note (not an ADR renumber) +**Date**: 2026-07-20 +**Related**: `core/epistemic_state.py`, `teaching/epistemic.py`, `vault/store.py`, `core/cognition/geometric_coherence.py` + +## Decision + +CORE keeps **three orthogonal axes**. They must not be collapsed into one enum. + +| Axis | Type | Owner module | Purpose | +|------|------|--------------|---------| +| Vault / pack standing | `EpistemicStatus` | `teaching/epistemic.py` | Durable SPECULATIVE / COHERENT / CONTESTED / FALSIFIED | +| Turn / dialogue taxonomy | `EpistemicState` | `core/epistemic_state.py` | Observability (perceived, verified, decoded, …) — **no COHERENT member** | +| Field geometric closure | `GeometricCoherenceVerdict` | `core/cognition/geometric_coherence.py` | Turn-level versor + GoldTether closed vs unverified | + +## Mapping + +- Vault `EpistemicStatus.COHERENT` → surface `EpistemicState.DECODED` via `epistemic_state_for_vault_status` (existing). +- Turn `GeometricCoherenceVerdict.GEOMETRICALLY_VERIFIED` does **not** auto-promote vault rows. +- Vault COHERENT promotion remains `VaultStore.store` / `apply_certified_promotion` / `promote_eligible_entries` only (INV-29). + +## Forbidden + +- Adding `EpistemicState.COHERENT` as a duplicate label without this ownership split. +- Using a single scalar score or bookkeeping flag as “coherent” without geometric checks on the field axis. diff --git a/tests/test_stage3_epistemic_inductive.py b/tests/test_stage3_epistemic_inductive.py new file mode 100644 index 00000000..69d0cf13 --- /dev/null +++ b/tests/test_stage3_epistemic_inductive.py @@ -0,0 +1,106 @@ +"""Stage 3A/C — geometric coherence taxonomy + inductive fixed-point.""" + +from __future__ import annotations + +import numpy as np + +from algebra.rotor import make_rotor_from_angle +from core.cognition.geometric_coherence import ( + GeometricCoherenceStatus, + evaluate_geometric_coherence, +) +from core.cognition.inductive_closure import expand_relation_closure +from chat.runtime import ChatRuntime +from core.cognition import CognitiveTurnPipeline + + +def test_geometric_coherence_verified_on_unit_versor(): + R = make_rotor_from_angle(0.25) + v = evaluate_geometric_coherence(R) + assert v.status is GeometricCoherenceStatus.GEOMETRICALLY_VERIFIED + assert v.closed is True + assert v.field_present is True + + +def test_geometric_coherence_refuses_missing_field(): + v = evaluate_geometric_coherence(None) + assert v.status is GeometricCoherenceStatus.REFUSED + assert v.closed is False + + +def test_geometric_coherence_unverified_on_dirty_field(): + dirty = np.zeros(32, dtype=np.float64) + dirty[0] = 0.5 + dirty[1] = 0.5 + v = evaluate_geometric_coherence(dirty) + assert v.status is GeometricCoherenceStatus.UNVERIFIED + assert v.closed is False + + +def test_pipeline_populates_geometric_coherence(): + rt = ChatRuntime() + p = CognitiveTurnPipeline(rt) + result = p.run("what is light", max_tokens=6) + assert result.geometric_coherence is not None + assert result.geometric_coherence.field_present is True + # Live fields after chat are typically closed versors. + assert result.geometric_coherence.status in { + GeometricCoherenceStatus.GEOMETRICALLY_VERIFIED, + GeometricCoherenceStatus.UNVERIFIED, + GeometricCoherenceStatus.REFUSED, + } + # Dual taxonomy: no EpistemicState.COHERENT + from core.epistemic_state import EpistemicState + from teaching.epistemic import EpistemicStatus + + assert not hasattr(EpistemicState, "COHERENT") + assert EpistemicStatus.COHERENT.value == "coherent" + + +def test_inductive_closure_derives_two_hop(): + triples = ( + ("a", "is", "b"), + ("b", "is", "c"), + ) + res = expand_relation_closure(triples, budget=8) + 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" + + +def test_inductive_closure_detects_contradiction(): + triples = ( + ("a", "is", "b"), + ("a", "is", "c"), + ) + res = expand_relation_closure(triples, budget=4) + assert any(c.contradiction for c in res.contradictions) + assert len(res.contradictions) >= 2 + + +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) + assert res.truncated or res.steps_taken <= 2 + # With larger budget, multi-hop appears + res2 = expand_relation_closure(triples, budget=16) + assert any(d.head == "a0" and d.tail == "a2" for d in res2.derived) or any( + d.head == "a0" for d in res2.derived + ) + + +def test_inductive_closure_cycle_safe(): + triples = ( + ("a", "r", "b"), + ("b", "r", "a"), + ) + res = expand_relation_closure(triples, budget=8) + # 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) diff --git a/tests/test_stage3_oov_egress_authority.py b/tests/test_stage3_oov_egress_authority.py new file mode 100644 index 00000000..7ca77bbd --- /dev/null +++ b/tests/test_stage3_oov_egress_authority.py @@ -0,0 +1,58 @@ +"""Stage 3D — OOV conformal probe + horosphere egress authority pins.""" + +from __future__ import annotations + +import numpy as np + +from algebra.cga import cga_inner +from algebra.versor import unitize_versor +from chat.runtime import ChatRuntime +from core.cognition import CognitiveTurnPipeline +from vocab.manifold import VocabManifold + + +def test_oov_geometric_context_carries_conformal_neighbors_or_topology(): + """Live pipeline OOV/pending path records geometric context, not lexical only.""" + rt = ChatRuntime() + p = CognitiveTurnPipeline(rt) + # Force an OOV-shaped prompt (unlikely pack lemma). + result = p.run("what is xyzzyplugh_oov_token_zzz", max_tokens=6) + ctx = result.oov_geometric_context + # Either OOV path fired with conformal note, or graph topology was recorded. + if ctx is not None: + assert "conformal_neighbors" in ctx or "unresolved_topology" in ctx or "node_depths" in ctx + if "note" in ctx: + assert "cga_inner" in ctx["note"] or "Conformal" in ctx["note"] or "depth" in ctx["note"] + + +def test_vocab_nearest_is_cga_inner_not_cosine(): + """Horosphere egress ranking is exact cga_inner argmax (Blueprint B.3).""" + rng = np.random.default_rng(11) + m = VocabManifold() + for w in ("alpha", "beta", "gamma"): + m.add(w, unitize_versor(rng.standard_normal(32).astype(np.float64))) + query = unitize_versor( + m.get_versor("beta").astype(np.float64) + 0.08 * rng.standard_normal(32) + ) + word, idx = m.nearest(query) + scores = [float(cga_inner(query, m.get_versor_at(i))) for i in range(len(m))] + assert idx == int(np.argmax(scores)) + assert word == m.get_word_at(idx) + + +def test_pipeline_does_not_use_vault_hits_as_gate_for_surface(): + """vault_hits remains telemetry; surface authority is geometric/resolution.""" + rt = ChatRuntime() + p = CognitiveTurnPipeline(rt) + result = p.run("what is light", max_tokens=6) + assert isinstance(result.vault_hits, int) + assert result.authority_source in { + "runtime_canonical", + "runtime_pre_decoration", + "runtime", + "realizer", + "substrate_realizer", + "", + } + # Geometric coherence is first-class and independent of vault_hits. + assert result.geometric_coherence is not None