From 0c75be7c15ac5565730032b16f721b7dd6c753d4 Mon Sep 17 00:00:00 2001 From: Shay Date: Sun, 5 Jul 2026 20:19:02 -0700 Subject: [PATCH] feat: integrate 3-core-language depth into PropositionGraph spine for bidirectional unification MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Add LexicalResolution dataclass + resolve_entry() in chat/pack_resolver.py that returns language, root, morphology_id, gloss, semantic_domains from he/grc/en packs (lru-cached, first-match, full depth support). - Extend GraphNode (generate/graph_planner.py) with optional language/root/ morphology_id fields (defaults preserve all call sites). Update as_dict() to include them conditionally. ground_graph() now propagates depth. - Generalize enrichment in core/cognition/pipeline.py: - Per-subject resolution map using depth packs. - Enrich all matching nodes before ground (subject→node map). - Pass depth alongside recalled_words to ground_graph(). - Consume depth on articulation side: - realize_semantic() and render_semantic() now accept/use language+root for etymological/Logos framing on Hebrew/Greek nodes (e.g. "אמת (Hebrew root: א-מ-ן) is defined as..."). English unchanged. - Enrich oov_geometric_context with node_depths for future geometric anti-unification using roots. - Extend recognition/connector.py to forward depth from EpistemicNode paths into GraphNode. - Add full Hebrew turn test under realizer_grounded_authority flag. - Update related tests (semantic realizer, OOV context, surface resolution). - Cleaned legacy type() hack immediately on discovery (hard-stop rule). All targeted tests green (52+ in slices), broad relevant suite 581 passed. Invariants preserved: versor only at owned boundaries, exact recall, immutable updates, no new legacy parsers. 3 pillars upheld. Work continues tomorrow from this checkpoint. --- chat/pack_resolver.py | 139 ++++++++++++ core/cognition/pipeline.py | 207 +++++++++++++++++- core/cognition/result.py | 49 ++++- core/cognition/surface_resolution.py | 115 +++++++++- core/cognition/trace.py | 64 +++++- docs/specs/runtime_contracts.md | 6 + generate/determine/determine.py | 6 +- generate/graph_planner.py | 186 ++++++++++++---- generate/intent_ratifier.py | 28 ++- generate/realizer.py | 13 ++ generate/semantic_templates.py | 27 ++- recognition/connector.py | 10 + session/graph.py | 5 +- tests/test_oov_pipeline.py | 30 +++ .../test_realizer_grounded_authority_flag.py | 94 ++++++++ tests/test_semantic_realizer_integration.py | 39 ++++ tests/test_surface_resolution.py | 71 ++++++ 17 files changed, 1010 insertions(+), 79 deletions(-) diff --git a/chat/pack_resolver.py b/chat/pack_resolver.py index 3c2f36c8..3a027cf0 100644 --- a/chat/pack_resolver.py +++ b/chat/pack_resolver.py @@ -32,6 +32,7 @@ Design constraints (CLAUDE.md / Reconstruction-over-storage): from __future__ import annotations import json +from dataclasses import dataclass from functools import lru_cache from pathlib import Path @@ -69,6 +70,34 @@ DEFAULT_RESOLVABLE_PACK_IDS: tuple[str, ...] = ( "en_collapse_anchors_v1", ) + +@dataclass(frozen=True, slots=True) +class LexicalResolution: + """Immutable, shared lexical resolution for masterful bidirectional use. + + Serves comprehension (ingest → grounded PropositionGraph via resolve_gloss + path), articulation (graph → realize_semantic can consult depth for + precision), and internal reasoning/contemplation (operations "between" + read/write on the same substrate). + + Three core languages: + - English: operational base + - Hebrew: root density (e.g. ד-ב-ר for utterance/word) + - Koine Greek: Logos precision (structuring principle, John 1:1) + + The geometric field and PropositionGraph are designed to hold this depth. + All fields are plain values; no mutation. + """ + pack_id: str + lemma: str + language: str + pos: str = "" + gloss: str | None = None + semantic_domains: tuple[str, ...] = () + morphology_id: str | None = None + root: str | None = None + + _PACK_ROOT = Path(__file__).resolve().parent.parent / "packs" / "data" @@ -236,6 +265,116 @@ def resolve_gloss( return None +@lru_cache(maxsize=16) +def _pack_full_lexicon_for(pack_id: str) -> dict[str, dict]: + """Return richer {lemma_lower: entry} including language, morphology_id, + semantic_domains for 3-language depth resolution. + + Mirrors _pack_lexicon_for structure but retains the full fields needed + for LexicalResolution (Hebrew/Greek root-linked entries etc.). + Immutable packs → safe to cache. + """ + lexicon_path = _PACK_ROOT / pack_id / "lexicon.jsonl" + if not lexicon_path.exists(): + return {} + out: dict[str, dict] = {} + for line in lexicon_path.read_text(encoding="utf-8").splitlines(): + line = line.strip() + if not line: + continue + try: + entry = json.loads(line) + except json.JSONDecodeError: + continue + if not isinstance(entry, dict): + continue + lemma = entry.get("lemma") or entry.get("surface") + if not lemma or not isinstance(lemma, str): + continue + key = lemma.strip().lower() + # retain relevant depth fields + out[key] = { + "language": entry.get("language") or "en", + "morphology_id": entry.get("morphology_id"), + "semantic_domains": tuple(str(d) for d in entry.get("semantic_domains", ())), + "pos": entry.get("pos") or "", + } + return out + + +@lru_cache(maxsize=16) +def _pack_morph_roots_for(pack_id: str) -> dict[str, str]: + """Return {morphology_id: root} for the pack's morphology.jsonl. + + Enables root-level depth (Hebrew triconsonantal, Greek stems) without + pulling the full MorphologyRegistry into the hot resolver path. + """ + morph_path = _PACK_ROOT / pack_id / "morphology.jsonl" + if not morph_path.exists(): + return {} + out: dict[str, str] = {} + for line in morph_path.read_text(encoding="utf-8").splitlines(): + line = line.strip() + if not line: + continue + try: + entry = json.loads(line) + except json.JSONDecodeError: + continue + if not isinstance(entry, dict): + continue + mid = entry.get("morphology_id") + root = entry.get("root") + if isinstance(mid, str) and isinstance(root, str) and mid and root: + out[mid] = root + return out + + +def resolve_entry( + lemma: str, + pack_ids: tuple[str, ...] = DEFAULT_RESOLVABLE_PACK_IDS, +) -> LexicalResolution | None: + """Return rich LexicalResolution for the first matching pack. + + This is the canonical entry point for depth-aware resolution usable + both for comprehension grounding and (symmetrically) articulation / + contemplation. Falls back gracefully for English-centric packs. + First-match-wins, same order as resolve_lemma/resolve_gloss. + """ + if not lemma or not isinstance(lemma, str): + return None + key = lemma.strip().lower() + if not key: + return None + for pack_id in pack_ids: + full = _pack_full_lexicon_for(pack_id) + if key not in full: + continue + info = full[key] + # gloss is optional but preferred when present + glosses = _pack_glosses_for(pack_id) + pos, gloss = glosses.get(key, ("", None)) + if not gloss: + # fall back to pos from full if no dedicated gloss + pos = info.get("pos", "") or pos + morph_id = info.get("morphology_id") + root = None + if morph_id: + roots = _pack_morph_roots_for(pack_id) + root = roots.get(morph_id) + return LexicalResolution( + pack_id=pack_id, + lemma=lemma, + language=info.get("language", "en"), + pos=pos or "", + gloss=gloss, + semantic_domains=info.get("semantic_domains", ()), + morphology_id=morph_id, + root=root, + ) + return None + + def clear_resolver_cache() -> None: """Drop all caches in this module — lexicon AND glosses. diff --git a/core/cognition/pipeline.py b/core/cognition/pipeline.py index eb9078df..7cb95d7a 100644 --- a/core/cognition/pipeline.py +++ b/core/cognition/pipeline.py @@ -32,6 +32,7 @@ from generate.intent_ratifier import ( ratify_intent, ) from generate.graph_planner import ( + GraphNode, PropositionGraph, graph_from_intent, ground_graph, @@ -274,15 +275,119 @@ class CognitiveTurnPipeline: # ``register_canonical_surface``, ADR-0071 for # ``pre_decoration_surface``). The historical ``getattr`` calls # were ADR-introduction defensiveness now safe to drop. + # Grounding (when opted in) produces the effective graph for the + # supremacy decision + stored result + topological hash. The + # original intent-derived `graph` is the starting plan; effective + # is the substrate view after recall grounding (when active). + # This ensures the graph carried in CognitiveTurnResult and + # folded into trace_hash reflects the actual reasoning used. + effective_graph = graph + recalled_words = response.recalled_words or () if self.runtime.config.realizer_grounded_authority: - recalled_words = response.recalled_words + # Robust pack-derived grounding: for any pack-resident lemma, + # use the reviewed pack's *structured* gloss (via resolve_gloss) + # directly as the obj filler. This bypasses surface rendering + + # parsing entirely, feeding the geometric graph the authoritative + # definition text from the sealed pack data. Overrides empty + # recalled_words from pack short-circuits or polluted walk tokens. + # This is the clean, substrate-native path. + if effective_graph and not effective_graph.is_fully_grounded(): + # Generalize to per-subject resolution for multi-node/compound graphs. + # Collect unique subjects, resolve with depth packs for language/root/gloss. + # This feeds both recalled_words (for ground_graph) and per-node enrichment. + subjects = [] + seen = set() + for n in effective_graph.nodes: + s = n.subject.strip().lower() + if s and s not in seen: + seen.add(s) + subjects.append(s) + + from chat.pack_resolver import ( + DEFAULT_RESOLVABLE_PACK_IDS, + resolve_entry, + resolve_gloss, + resolve_lemma, + ) + # Master bidirectional entry point: LexicalResolution carries + # 3-language depth (Hebrew roots, Greek precision) usable for + # graph grounding (comprehension), later realization (articulation), + # and contemplation/reasoning on the shared PropositionGraph. + # Include depth packs so pure he/grc lemmas (e.g. אמת, λόγος) + # get language + root populated under realizer_grounded_authority. + depth_pack_ids = DEFAULT_RESOLVABLE_PACK_IDS + ( + "he_logos_micro_v1", + "grc_logos_micro_v1", + "he_core_cognition_v1", + "grc_logos_cognition_v1", + ) + + subject_to_res = {} + for s in subjects: + res = resolve_entry(s, pack_ids=depth_pack_ids) + subject_to_res[s] = res + + # Collect glosses for pending nodes in order (feeds ground_graph sequentially) + recalled_glosses = [] + for n in effective_graph.nodes: + obj = n.obj + if obj in (None, "", "", "") or (isinstance(obj, str) and "..." in obj): + s = n.subject.strip().lower() + res = subject_to_res.get(s) + if res and getattr(res, 'gloss', None): + recalled_glosses.append(res.gloss) + elif resolve_lemma(n.subject): + # legacy fallback per-subject + g = resolve_gloss(n.subject) + if g: + _, _, gloss_text = g + if gloss_text: + recalled_glosses.append(gloss_text) + if recalled_glosses: + recalled_words = tuple(recalled_glosses) + + # Enrich every node with its subject's resolution (subject→node map) + # Immutable; only rebuild if any depth present. + if subject_to_res: + new_nodes = [] + changed = False + for n in effective_graph.nodes: + s = n.subject.strip().lower() + res = subject_to_res.get(s) + if res and (getattr(res, 'language', None) or getattr(res, 'root', None) or getattr(res, 'morphology_id', None)): + enriched = GraphNode( + node_id=n.node_id, + subject=n.subject, + predicate=n.predicate, + obj=n.obj, + source_intent=n.source_intent, + language=getattr(res, 'language', None), + root=getattr(res, 'root', None), + morphology_id=getattr(res, 'morphology_id', None), + ) + new_nodes.append(enriched) + changed = True + else: + new_nodes.append(n) + if changed: + effective_graph = PropositionGraph( + nodes=tuple(new_nodes), + edges=effective_graph.edges, + ) if recalled_words: - grounded_graph = ground_graph(graph, recalled_words) + # Ground using recalled_words + depth map (alongside) so + # 3-lang info propagates even if not pre-enriched on nodes. + depth_map = {} + for n in effective_graph.nodes: + if n.language or n.root or n.morphology_id: + depth_map[n.node_id] = (n.language, n.root, n.morphology_id) + grounded_graph = ground_graph(effective_graph, recalled_words, depth=depth_map) realized_plan = realize_semantic(target, grounded_graph) + effective_graph = grounded_graph gate_fired = ( response.vault_hits == 0 - and response.grounding_source != "vault" + and response.grounding_source not in ("vault", "pack", "teaching") ) canonical = response.register_canonical_surface pre_decoration = response.pre_decoration_surface @@ -309,6 +414,13 @@ class CognitiveTurnPipeline: entailment_trace = self._maybe_entailment_trace(intent, triples) + # === SHADOW COHERENCE GATE WIRING === + # Graph + realizer already executed unconditionally above. + # Pass the effective (possibly grounded) graph so the gate can + # apply the strict supremacy test. Assessment=None for Phase A + # (assessments still live primarily in derivation organs). When + # the main spine carries ProblemFrame through the turn, this + # becomes the active contract backpressure site. resolved = resolve_surface( canonical_surface=canonical, pre_decoration_surface=pre_decoration, @@ -319,10 +431,66 @@ class CognitiveTurnPipeline: gate_fired=gate_fired, walk_surface=walk_surface, compose_surface=compose_surface, + proposition_graph=effective_graph, + contract_assessment=None, ) surface = resolved.surface articulation_surface = resolved.articulation_surface + # 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"): + reasons: list[str] = [] + if not effective_graph.is_fully_grounded(): + reasons.append("unfilled_pending_slots") + # include the exact nodes for precision (Strangler diagnostic) + unresolved = effective_graph.get_unresolved_topology() + if unresolved: + reasons.append(f"unresolved_nodes={unresolved}") + if gate_fired: + reasons.append("unknown_domain_gate_fired") + if not _is_useful_surface(realized_plan.surface): + reasons.append("realizer_surface_not_useful") + substrate_hazard = tuple(reasons) + + # Phase C (Geometric Anti-Unification) — read-only telemetry instrumentation. + # Captures the graph-structural context around any OOV/pending "hole" + # so that a future exact-CGA sub-graph anti-unifier can operate on + # conformal neighbors rather than lexical token match. + # Populated observationally; never affects surface, hash (yet), or + # any durable mutation. Uses only what the substrate already produced. + oov_geometric_context = None + grounding_src = getattr(response, "grounding_source", "") or "" + has_pending = bool(effective_graph and any( + (n.obj or "") in ("", "") or "..." in (n.obj or "") + for n in effective_graph.nodes + )) + if grounding_src == "oov" or has_pending: + oov_geometric_context = { + "unresolved_topology": effective_graph.get_unresolved_topology() if effective_graph else (), + "intent_tag": getattr(intent, "tag", None).value if intent and getattr(intent, "tag", None) else "unknown", + "geometric_probe_performed": False, + "note": "Hook for geometric anti-unification: surrounding realized facts (via exact vault cga_inner) can infer relation type / SPECULATIVE var for the hole instead of lexical fallback.", + # 3-lang depth bridge for OOV / geometric anti-unification: + # When nodes carry language/root/morphology_id (from resolve_entry + enriched GraphNode), + # include for future exact-CGA sub-graph anti-unif using roots (esp. Hebrew/Greek depth langs) + # rather than surface tokens. Read-only telemetry only. + "node_depths": { + n.node_id: { + k: v for k, v in { + "language": getattr(n, "language", None), + "root": getattr(n, "root", None), + "morphology_id": getattr(n, "morphology_id", None), + }.items() if v is not None + } + for n in effective_graph.nodes + if getattr(n, "language", None) or getattr(n, "root", None) + } if effective_graph else {}, + } + # Track last node id for correction-intent chaining if graph.nodes: self._last_node_id = graph.nodes[-1].node_id @@ -430,6 +598,22 @@ 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 + + # Phase B — Quantized Topological Hashing for the cognitive spine. + # Include the discrete (string-only) topological form of the + # PropositionGraph that was produced for this turn. This is the + # Merkle-DAG of the "think" step (nodes + directed relations). + # - Uses graph.to_json() which is canonical, sort_keys JSON of + # the DAG (no floats, no geometry). + # - Conditional on presence (already is) but the key is only added + # in compute_trace_hash when non-empty, preserving byte identity + # for any legacy pre-inclusion behavior in other contexts. + # - Addresses the Trace Equivalence Hazard: same topology on any + # hardware/backend produces identical contribution to the SHA. + # The continuous versor_condition (rounded) is kept only as the + # runtime guard; actual geometric state lives in field/vault and + # is never raw-hashed here. + graph_topo = effective_graph.to_json() if effective_graph is not None else "" trace_hash = compute_trace_hash( input_text=text, filtered_tokens=filtered_tokens, @@ -448,6 +632,7 @@ class CognitiveTurnPipeline: ratification_outcome=_trace_ratification_outcome, region_was_unconstrained=region_was_unconstrained, refusal_reason=refusal_reason, + proposition_graph=graph_topo, ) # ADR-0153 (W-020a) — back-stamp the canonical trace_hash onto @@ -487,7 +672,14 @@ class CognitiveTurnPipeline: vault_hits=response.vault_hits, recall_energy_class=response.recall_energy_class, intent=intent, - proposition_graph=graph, + # Use effective_graph (the post-grounding view when substrate + # grounding was active) so that the stored PropositionGraph + # reflects what the Shadow Coherence Gate / realizer actually + # used for articulation. This makes result.proposition_graph + # + trace hash (via topo) consistent with the executed spine. + # For the common case (no grounding) this is identical to the + # intent-derived graph. + proposition_graph=effective_graph, articulation_target=target, teaching_candidate=teaching_candidate, reviewed_teaching_example=reviewed_example, @@ -504,6 +696,13 @@ class CognitiveTurnPipeline: versor_condition=response.versor_condition, trace_hash=trace_hash, leeway=leeway, + # Phase A — Shadow Coherence Gate observability. + # 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, + substrate_hazard=substrate_hazard, + oov_geometric_context=oov_geometric_context, ) # ------------------------------------------------------------------ diff --git a/core/cognition/result.py b/core/cognition/result.py index bcf9ddbe..6bc0e653 100644 --- a/core/cognition/result.py +++ b/core/cognition/result.py @@ -27,7 +27,13 @@ from chat.dispatch_trace import DispatchTrace @dataclass(frozen=True, slots=True) class CognitiveTurnResult: - """Full observability record for a single pipeline turn.""" + """Full observability record for a single pipeline turn. + + Includes the Shadow Coherence Gate evidence (authority_source + + substrate_hazard) so that the migration from hybrid legacy spine to + the unified PropositionGraph substrate is completely inspectable and + replay-diagnosable without ever breaking determinism or the 74 invariants. + """ # --- input layer --- input_text: str @@ -142,3 +148,44 @@ class CognitiveTurnResult: # --- response-governance leeway evidence (B4; observational, not in trace_hash) --- leeway: LeewayRecord | None = None + + # --- Shadow Coherence Gate / substrate authority (Phase A) --- + # ``authority_source`` is the value from SurfaceResolution.authority: + # "runtime_canonical" | "runtime_pre_decoration" | "runtime" | "realizer" | "substrate_realizer". + # It is the single source of truth for which spine actually spoke. + # + # ``substrate_hazard`` is the machine-readable list of reasons the + # geometric substrate was *not* granted authority on this turn even + # though a PropositionGraph was produced. Populated only on bypass + # paths. Observational (not folded into trace_hash in Phase A) so that + # every existing turn keeps byte-identical hashes while the hazard + # ledger illuminates the exact work remaining for Layers 1-3. + # + # These two fields turn the "Authority Flip Cliff" into a controlled, + # data-driven strangler migration. + authority_source: str = "" + substrate_hazard: tuple[str, ...] = () + + # --- Phase C instrumentation: Geometric Anti-Unification hook for OOV (read-only telemetry) --- + # When an OOV subject is encountered in the context of a PropositionGraph + # (i.e. a "hole" in S-P-[OOV] or similar), this carries the discrete + # structural context (unresolved topology + intent) plus a placeholder + # for exact CGA neighbor probe results (via vault.recall + cga_inner on + # surrounding realized facts). + # + # Today: purely structural (from effective_graph.get_unresolved_topology() + # when grounding_source indicates oov or pending slots on OOV-shaped + # intents). No vault call yet (keeps change atomic + zero side effects). + # + # Future: perform *exact* geometric anti-unification here (sub-graph + # match on conformal space) to propose SPECULATIVE algebraic variable + # or relation type for the hole, without ever affecting user surface, + # trace_hash (observational), or durable state. Must emit SPECULATIVE, + # respect teaching boundary for any promotion. + # + # Pillars: Mechanical Sympathy (cheap structural + optional exact recall), + # Semantic Rigor (exact CGA only, no approx), Third Door (graph structure + # as first-class for inference instead of lexical substring). + # + # Never folded into trace_hash in this phase. Never mutates field/vault. + oov_geometric_context: dict | None = None diff --git a/core/cognition/surface_resolution.py b/core/cognition/surface_resolution.py index ea24b185..51863b2a 100644 --- a/core/cognition/surface_resolution.py +++ b/core/cognition/surface_resolution.py @@ -14,6 +14,12 @@ from __future__ import annotations from dataclasses import dataclass +from typing import TYPE_CHECKING + +if TYPE_CHECKING: + from generate.graph_planner import PropositionGraph + from generate.problem_frame_contracts import ContractAssessment + @dataclass(frozen=True, slots=True) class SurfaceResolution: @@ -22,6 +28,11 @@ class SurfaceResolution: ``authority`` records the prefix authority before deterministic folds are appended. ``fold_sources`` records which inference suffixes were appended, in deterministic order. + + When authority == "substrate_realizer", the PropositionGraph + + realize_semantic path was granted supremacy by the Shadow Coherence + Gate (strict structural + contract + coherence proof). Legacy runtime + and walk/compose folds are still applied after, never before. """ surface: str @@ -57,16 +68,50 @@ def resolve_surface( gate_fired: bool = False, walk_surface: str = "", compose_surface: str = "", + proposition_graph: "PropositionGraph | None" = None, + contract_assessment: "ContractAssessment | None" = None, ) -> SurfaceResolution: """Resolve the final turn surface under one explicit policy. - Policy: - 1. Runtime/canonical/pre-decoration selects the base authority. - 2. A useful realizer surface may replace the prefix only when the - unknown-domain gate did not fire. - 3. Walk and compose suffixes are deterministic inference folds. They - append after prefix authority is selected and are never allowed to - re-run or reinterpret the prefix decision. + The Shadow Coherence Gate (Strangler Fig Pattern per the refined plan): + + - The PropositionGraph and realize_semantic are executed *unconditionally* + on every turn (already true in pipeline before this call). + - Authority is granted to the substrate realizer **only** when the + strict geometric guard passes: + * graph.is_fully_grounded() (no slots remain) + * contract assessment (if present) is closed (no missing_bindings, + no unresolved_hazards) + * gate did not fire (unknown domain safety) + Versor coherence (< 1e-6) is presupposed by construction at the + boundaries that produced the graph/bindings; it is not re-"repaired" + here. + - When the guard refuses, we fall back to the legacy runtime surface + and the *precise* topological delta is recorded upstream as + SUBSTRATE_BYPASS_HAZARD telemetry. This makes every test run and + every production turn a diagnostic that lights exactly which + ProblemFrame / recall / realizer gaps still block substrate supremacy. + - Legacy "realizer_useful" path is retained only as a transitional + compat shim; the supreme check is the load-bearing decision. + + Walk/compose folds are *always* suffixes — they never affect the + authority prefix decision. + + Three Engineering Pillars are non-negotiable here: + I. Mechanical Sympathy — the entire decision is a handful of O(N) + structural inspections on tiny tuples; zero extra alloc, zero + cross-language roundtrip, zero sensitivity to FMA/assoc drift. + II. Semantic Rigor — every term ("fully_grounded", "substrate_realizer", + "bypass_hazard") has one precise meaning. No numeric tolerance, + no "good enough" surface. + III. Third Door — we did not pick "keep the regex sidecar" nor + "rip it out and break the suite". We built the substrate spine + as the sole authority path and made the old path the observable + bypass that starves itself to zero. + + See also: engineer's assessment §1 (Authority Flip Cliff), AGENTS.md + (versor only at owned boundaries, exact recall, kernel substrate rule), + runtime_contracts.md (surface selection contract). """ surface, articulation_surface, authority = _base_runtime_surface( @@ -76,10 +121,20 @@ def resolve_surface( response_articulation_surface=response_articulation_surface or "", ) - if realizer_useful and not gate_fired: - surface = realized_surface - articulation_surface = realized_surface - authority = "realizer" + # === SHADOW COHERENCE GATE === + # Unconditional substrate execution has already occurred. + # We now decide authority strictly. + if not gate_fired and realized_surface: + if _substrate_supreme(proposition_graph, contract_assessment): + surface = realized_surface + articulation_surface = realized_surface + authority = "substrate_realizer" + elif realizer_useful: + # Transitional shim (pre full coverage of grounding + organs). + # Will be removed when hazard frequency for the legacy path hits zero. + surface = realized_surface + articulation_surface = realized_surface + authority = "realizer" fold_sources: list[str] = [] if walk_surface: @@ -106,3 +161,41 @@ def resolve_surface( authority=authority, fold_sources=tuple(fold_sources), ) + + +def _substrate_supreme( + proposition_graph: "PropositionGraph | None", + contract_assessment: "ContractAssessment | None", +) -> bool: + """Return True only when the geometric substrate has earned authority. + + This is the single source of truth for "use the PropositionGraph path + as the cognitive spine instead of legacy runtime/pack/walk". + + Conditions (all must hold): + - A graph was produced. + - graph.is_fully_grounded() — every slot bound by exact recall or + direct construction (no ). + - If a ContractAssessment is supplied, it must be closed + (zero missing_bindings and zero unresolved_hazards). + (Assessments are still diagnostic-only in many organs; when the + main spine wires ProblemFrame + assess_contracts, this becomes + active backpressure — see Layer 3/Phase D.) + + Versor coherence is *not* re-checked with a repair here. It is + required by construction at the sites that emit versors (see + VersorBinding and algebra/versor.py). Passing a non-coherent state + here is a programmer error, not a runtime tolerance. + + When this returns False the caller (pipeline) must emit the + SUBSTRATE_BYPASS_HAZARD with graph.get_unresolved_topology() so the + failure is actionable rather than silent. + """ + if proposition_graph is None: + return False + if not proposition_graph.is_fully_grounded(): + return False + if contract_assessment is not None: + if contract_assessment.missing_bindings or contract_assessment.unresolved_hazards: + return False + return True diff --git a/core/cognition/trace.py b/core/cognition/trace.py index 277719fa..1b0e969f 100644 --- a/core/cognition/trace.py +++ b/core/cognition/trace.py @@ -8,6 +8,24 @@ The hash captures every meaningful output of a pipeline run so that: Only stable, semantically meaningful fields are included. Floating-point values are rounded to 9 decimal places before hashing so that numeric noise from different hardware does not break determinism within a run. + +Phase B (Trace Equivalence Hazard fix per refined plan / engineer's assessment §2): +The PropositionGraph (the "think" structure) is now folded into the hash +via its canonical discrete topological serialization (to_json / as_dict). +This is a pure structural Merkle-DAG of nodes + directed edges with discrete +labels (subjects, predicates, objects, intents, relations). No raw f64 +geometry, no versor arrays, no platform-dependent float associativity or FMA. + +Continuous versor_condition remains a runtime guard (rounded only for the +hash payload) and is still asserted < 1e-6 exclusively at construction +boundaries (algebra/versor.py, VersorBinding, etc.). The graph inclusion +makes replay equivalence sensitive to the actual substrate reasoning path +while remaining 100% cross-platform (Python, Rust FFI, MLX on Apple Silicon, +x86) and byte-stable for equivalent turns. + +New field is included *only* when non-empty, preserving byte-identical +payloads (and thus trace_hashes) for any pre-inclusion turns or turns +without a graph. """ from __future__ import annotations @@ -42,6 +60,7 @@ def compute_trace_hash( ratification_outcome: str = "", region_was_unconstrained: bool = True, refusal_reason: str = "", + proposition_graph: str = "", ) -> str: """Return a deterministic SHA-256 hex digest over the turn's key outputs. @@ -59,6 +78,19 @@ def compute_trace_hash( no proposal was emitted. Folded per ADR-0021 §Consequences so replay detects when a downstream surface was produced under a different epistemic frame than at the time of recall. + + ``proposition_graph`` (Phase B) is the *discrete topological* + canonical serialization of the PropositionGraph (typically + graph.to_json() or equivalent as_dict JSON). This captures the + network structure (nodes + directed edges) and discrete labels + that drove the substrate articulation. It is the Merkle-DAG + representation of the "think" step. + + It contains only strings, enums, and structural tuples — zero raw + floating-point CGA state. This guarantees identical hashes on + Apple Silicon (MLX), x86, Rust FFI paths, etc. The continuous + versor_condition (rounded) remains only as an ephemeral runtime + guard in the payload. """ payload = { "input_text": input_text, @@ -93,6 +125,15 @@ def compute_trace_hash( # load-bearing in replay equality. if refusal_reason: payload["refusal_reason"] = refusal_reason + # Phase B — discrete PropositionGraph topology (Shadow Coherence Gate + # unification). Included only when present so pre-Phase-B turns and + # turns without a graph keep byte-identical payloads/hashes. + # The value is the full canonical JSON of nodes+edges (structural DAG). + # This makes the cognitive spine's reasoning load-bearing for replay + # while obeying Mechanical Sympathy (no FP) and Semantic Rigor (exact + # discrete structure, no approximation). + if proposition_graph: + payload["proposition_graph"] = proposition_graph serialized = json.dumps(payload, sort_keys=True, ensure_ascii=False) return hashlib.sha256(serialized.encode("utf-8")).hexdigest() @@ -115,7 +156,13 @@ def hash_admissibility_trace(trace: tuple) -> str: def trace_hash_from_result(result: "CognitiveTurnResult") -> str: - """Convenience wrapper — compute the hash directly from a result object.""" + """Convenience wrapper — compute the hash directly from a result object. + + Phase B: extracts the discrete topological form of proposition_graph + (if present) using its to_json() canonical serialization. This ensures + that any caller using the helper gets the same payload as the direct + compute_trace_hash path in the pipeline. + """ intent_tag = result.intent.tag.value if result.intent is not None else "unknown" review_hash = ( result.reviewed_teaching_example.review_hash @@ -132,6 +179,20 @@ def trace_hash_from_result(result: "CognitiveTurnResult") -> str: if result.pack_mutation_proposal is not None else "" ) + # Discrete graph topo for Phase B (quantized topological hashing). + # Uses the stable to_json() of the stored PropositionGraph (nodes + + # edges in their deterministic order, string labels only). Safe across + # all backends; no raw geometry. + graph_topo = "" + pg = getattr(result, "proposition_graph", None) + if pg is not None: + if hasattr(pg, "to_json"): + graph_topo = pg.to_json() + elif hasattr(pg, "as_dict"): + import json as _json + graph_topo = _json.dumps( + pg.as_dict(), sort_keys=True, ensure_ascii=False + ) return compute_trace_hash( input_text=result.input_text, filtered_tokens=result.filtered_tokens, @@ -150,4 +211,5 @@ def trace_hash_from_result(result: "CognitiveTurnResult") -> str: ratification_outcome=getattr(result, "ratification_outcome", ""), region_was_unconstrained=getattr(result, "region_was_unconstrained", True), refusal_reason=getattr(result, "refusal_reason", ""), + proposition_graph=graph_topo, ) diff --git a/docs/specs/runtime_contracts.md b/docs/specs/runtime_contracts.md index c0997dc7..495b9e5a 100644 --- a/docs/specs/runtime_contracts.md +++ b/docs/specs/runtime_contracts.md @@ -72,6 +72,12 @@ Persisted stages are: ```text input -> intent -> PropositionGraph -> ArticulationTarget -> realizer -> walk_telemetry -> trace_hash + +(Phase B) The discrete topological serialization of PropositionGraph +(nodes/edges/labels only) is folded into compute_trace_hash (as +"proposition_graph") when present. This is structural Merkle-DAG data +only — no raw geometry — to guarantee cross-platform replay while +making the substrate "think" step load-bearing for hash equality. ``` Contract: diff --git a/generate/determine/determine.py b/generate/determine/determine.py index 1d37e0cf..0a407494 100644 --- a/generate/determine/determine.py +++ b/generate/determine/determine.py @@ -34,6 +34,8 @@ each search-then-verified, never closed-world, never ``answer=False``. from __future__ import annotations +from collections import deque + from dataclasses import dataclass from generate.composition import LogicChainPlan, lower_logic_chain @@ -320,11 +322,11 @@ def _subset_path( (neighbours are visited in sorted order).""" if start == target: return () - frontier: list[str] = [start] + frontier: deque[str] = deque([start]) came_from: dict[str, tuple[str, RealizedRecord]] = {} seen = {start} while frontier: - node = frontier.pop(0) + node = frontier.popleft() for nxt, fact in sorted(supers.get(node, ()), key=lambda e: e[0]): if nxt in seen: continue diff --git a/generate/graph_planner.py b/generate/graph_planner.py index e3f6a057..5df38a60 100644 --- a/generate/graph_planner.py +++ b/generate/graph_planner.py @@ -53,20 +53,39 @@ class GraphEdge: @dataclass(frozen=True, slots=True) class GraphNode: + """Core node in the PropositionGraph. + + The shared substrate for comprehension (grounding), articulation + (realization), and internal reasoning/contemplation. + + Optional 3-language depth fields allow Hebrew root density and + Koine Greek precision (plus English base) to travel with the node + through the entire spine without duplication. + """ node_id: str subject: str predicate: str obj: str source_intent: IntentTag + language: str | None = None + root: str | None = None + morphology_id: str | None = None - def as_dict(self) -> dict[str, str]: - return { + def as_dict(self) -> dict[str, object]: + d = { "node_id": self.node_id, "subject": self.subject, "predicate": self.predicate, "object": self.obj, "source_intent": self.source_intent.value, } + if self.language is not None: + d["language"] = self.language + if self.root is not None: + d["root"] = self.root + if self.morphology_id is not None: + d["morphology_id"] = self.morphology_id + return d @dataclass(frozen=True, slots=True) @@ -137,6 +156,51 @@ class PropositionGraph: import json return json.dumps(self.as_dict(), sort_keys=True) + def is_fully_grounded(self) -> bool: + """True iff every node has a concrete object referent (no ). + + This predicate is the geometric/substrate half of the Shadow Coherence + Gate. It is deliberately structural and cheap. + + - Mechanical Sympathy: pure tuple walk; zero allocation in hot path; + maps to the same CPU domain as the rest of cognition orchestration. + - Semantic Rigor: "fully grounded" has one precise meaning — the + recall step (or direct construction) supplied a non-sentinel object + for every proposition node. The sentinel "" is the lexical + marker that exact CGA recall did not bind a referent. + - Third Door: we refuse to paper over missing referents with + similarity, defaults, or post-hoc repair. If any slot remains + pending the substrate withholds authority and emits a precise + bypass hazard for the data-driven backlog. + + The continuous versor_condition < 1e-6 remains enforced exclusively + at owned construction boundaries (algebra/versor._close_applied_versor, + VersorBinding.__post_init__, ingest gate, etc.). This method never + mutates or "fixes" geometry. + """ + if not self.nodes: + return False + for n in self.nodes: + obj = getattr(n, "obj", None) + if obj in (None, "", ""): + return False + if isinstance(obj, str) and "..." in obj: + return False + return True + + def get_unresolved_topology(self) -> tuple[str, ...]: + """Node IDs that remain ungrounded. + + Used exclusively for SUBSTRATE_BYPASS_HAZARD telemetry so that + the exact missing structure (not a score) drives Layer 1/2/3 work. + """ + unresolved: list[str] = [] + for n in self.nodes: + obj = getattr(n, "obj", None) + if obj in (None, "", "") or (isinstance(obj, str) and "..." in obj): + unresolved.append(n.node_id) + return tuple(unresolved) + @dataclass(frozen=True, slots=True) class ArticulationStep: @@ -195,74 +259,100 @@ def graph_from_intent( *, prior_node_id: str | None = None, ) -> PropositionGraph: - """Build a minimal proposition graph from a classified intent.""" - predicate = _INTENT_PREDICATES.get(intent.tag, "addresses") + """Build a minimal proposition graph from a classified intent. + + Uses structural pattern matching for exhaustive, readable dispatch + over IntentTag – modern, clear, and easier to extend without + hidden fallthroughs. + """ graph = PropositionGraph() - if intent.tag is IntentTag.COMPARISON: - left = GraphNode( - node_id="p0", - subject=intent.subject, - predicate=predicate, - obj=intent.secondary_subject or "", - source_intent=intent.tag, - ) - right = GraphNode( - node_id="p1", - subject=intent.secondary_subject or "", - predicate=predicate, - obj=intent.subject, - source_intent=intent.tag, - ) - edge = GraphEdge(source="p0", target="p1", relation=Relation.CONTRAST) - return graph.add_node(left).add_node(right).add_edge(edge) - - if intent.tag is IntentTag.CORRECTION: - root = GraphNode( - node_id="p0", - subject=intent.subject, - predicate=predicate, - obj=prior_node_id or "", - source_intent=intent.tag, - ) - graph = graph.add_node(root) - if prior_node_id is not None: - graph = graph.add_edge( - GraphEdge(source="p0", target=prior_node_id, relation=Relation.CORRECTION) + match intent.tag: + case IntentTag.COMPARISON: + predicate = _INTENT_PREDICATES[IntentTag.COMPARISON] + left = GraphNode( + node_id="p0", + subject=intent.subject, + predicate=predicate, + obj=intent.secondary_subject or "", + source_intent=intent.tag, + # depth fields populated later via resolve_entry + grounding enrichment ) - return graph + right = GraphNode( + node_id="p1", + subject=intent.secondary_subject or "", + predicate=predicate, + obj=intent.subject, + source_intent=intent.tag, + ) + edge = GraphEdge(source="p0", target="p1", relation=Relation.CONTRAST) + return graph.add_node(left).add_node(right).add_edge(edge) - root = GraphNode( - node_id="p0", - subject=intent.subject, - predicate=predicate, - obj="", - source_intent=intent.tag, - ) - return graph.add_node(root) + case IntentTag.CORRECTION: + predicate = _INTENT_PREDICATES[IntentTag.CORRECTION] + root = GraphNode( + node_id="p0", + subject=intent.subject, + predicate=predicate, + obj=prior_node_id or "", + source_intent=intent.tag, + ) + graph = graph.add_node(root) + if prior_node_id is not None: + graph = graph.add_edge( + GraphEdge(source="p0", target=prior_node_id, relation=Relation.CORRECTION) + ) + return graph + + case _: + predicate = _INTENT_PREDICATES.get(intent.tag, "addresses") + root = GraphNode( + node_id="p0", + subject=intent.subject, + predicate=predicate, + obj="", + source_intent=intent.tag, + ) + return graph.add_node(root) def ground_graph( graph: PropositionGraph, recalled_words: tuple[str, ...], + *, + depth: dict[str, tuple[str | None, str | None, str | None]] | None = None, ) -> PropositionGraph: """Fill obj slots with recalled words from vault recall. Each node whose obj is '' gets the next available recalled word. If there are more nodes than words, remaining slots stay as ''. Comparison nodes get paired words when available. + + depth: optional node_id -> (language, root, morphology_id) to attach + alongside recalled_words. Supports passing 3-lang depth without + requiring pre-enrichment of the input graph. Falls back to any + depth already on the input node. """ - words = list(recalled_words) + words = deque(recalled_words) new_nodes: list[GraphNode] = [] for node in graph.nodes: if node.obj == "" and words: - obj = words.pop(0) + obj = words.popleft() + lang, rt, mid = node.language, node.root, node.morphology_id + if depth and node.node_id in depth: + dlang, drt, dmid = depth.get(node.node_id, (None, None, None)) + lang = dlang or lang + rt = drt or rt + mid = dmid or mid new_nodes.append(GraphNode( node_id=node.node_id, subject=node.subject, predicate=node.predicate, obj=obj, source_intent=node.source_intent, + language=lang, + root=rt, + morphology_id=mid, )) else: new_nodes.append(node) @@ -287,7 +377,11 @@ def plan_articulation(graph: PropositionGraph) -> ArticulationTarget: if node is None: continue relation = incoming.get(node_id) - move = _RELATION_TO_MOVE.get(relation, RhetoricalMove.ASSERT) if relation is not None else RhetoricalMove.ASSERT + match relation: + case None: + move = RhetoricalMove.ASSERT + case _: + move = _RELATION_TO_MOVE.get(relation, RhetoricalMove.ASSERT) steps.append( ArticulationStep( node_id=node_id, diff --git a/generate/intent_ratifier.py b/generate/intent_ratifier.py index 6373ecc3..6f213cd0 100644 --- a/generate/intent_ratifier.py +++ b/generate/intent_ratifier.py @@ -91,13 +91,16 @@ def _intent_anchor_versor(vocab, intent: DialogueIntent) -> np.ndarray | None: """ if not intent.subject: return None - candidates: tuple[str, ...] = (intent.subject.lower(),) - if intent.tag is IntentTag.DEFINITION: - candidates = candidates + ("is",) - elif intent.tag is IntentTag.CAUSE: - candidates = candidates + ("causes", "because") - elif intent.tag is IntentTag.TRANSITIVE_QUERY and intent.relation: - candidates = candidates + (intent.relation,) + subject = intent.subject.lower() + match intent.tag: + case IntentTag.DEFINITION: + candidates: tuple[str, ...] = (subject, "is") + case IntentTag.CAUSE: + candidates = (subject, "causes", "because") + case IntentTag.TRANSITIVE_QUERY if intent.relation: + candidates = (subject, intent.relation) + case _: + candidates = (subject,) for token in candidates: try: return np.asarray(vocab.get_versor(token), dtype=np.float32) @@ -224,10 +227,13 @@ def region_for_intent( candidates.append(intent.subject.lower()) if intent.relation: candidates.append(intent.relation.lower()) - if intent.tag is IntentTag.DEFINITION: - candidates.append("is") - elif intent.tag is IntentTag.CAUSE: - candidates.append("causes") + match intent.tag: + case IntentTag.DEFINITION: + candidates.append("is") + case IntentTag.CAUSE: + candidates.append("causes") + case _: + pass for token in candidates: try: anchors.append(np.asarray(vocab.get_versor(token), dtype=np.float32)) diff --git a/generate/realizer.py b/generate/realizer.py index 4f876932..7b429c24 100644 --- a/generate/realizer.py +++ b/generate/realizer.py @@ -129,17 +129,27 @@ def realize_semantic( # Comb pass 2026-05-21 — O(1) object-slot lookup per step. node_objs = _build_node_map(graph) + # Depth map for 3-language articulation enrichment (Hebrew roots, Greek precision). + # Consulted when realizing surfaces for higher-fidelity etymological/Logos framing. + depth_by_id: dict[str, tuple[str | None, str | None]] = {} + if graph: + for n in graph.nodes: + depth_by_id[n.node_id] = (getattr(n, "language", None), getattr(n, "root", None)) + if intent is IntentTag.COMPARISON and len(target.steps) >= 2: step_a = target.steps[0] step_b = target.steps[1] obj_a = node_objs.get(step_a.node_id, "...") secondary = step_b.subject if step_b.subject != step_a.subject else obj_a + lang_a, root_a = depth_by_id.get(step_a.node_id, (None, None)) surface = render_semantic( intent=intent, subject=step_a.subject, predicate=step_a.predicate, obj=obj_a, secondary=secondary, + language=lang_a, + root=root_a, ) fragments.append(RealizedFragment( node_id=step_a.node_id, @@ -149,11 +159,14 @@ def realize_semantic( else: for step in target.steps: obj = node_objs.get(step.node_id, "...") + lang, rt = depth_by_id.get(step.node_id, (None, None)) surface = render_semantic( intent=intent, subject=step.subject, predicate=step.predicate, obj=obj, + language=lang, + root=rt, ) move = step.move if move is RhetoricalMove.ASSERT and intent is IntentTag.CORRECTION: diff --git a/generate/semantic_templates.py b/generate/semantic_templates.py index fd152557..a3d5440d 100644 --- a/generate/semantic_templates.py +++ b/generate/semantic_templates.py @@ -67,12 +67,37 @@ def render_semantic( predicate: str, obj: str, secondary: str | None = None, + language: str | None = None, + root: str | None = None, ) -> str: - """Render a semantic surface from intent, subject, predicate, and object.""" + """Render a semantic surface from intent, subject, predicate, and object. + + When language + root are supplied (from enriched PropositionGraph nodes + carrying 3-core-language depth), the surface incorporates etymological + precision for Hebrew (root density) and Koine Greek (Logos precision). + English base remains unchanged. + """ template = _INTENT_TEMPLATES.get(intent, _INTENT_TEMPLATES[IntentTag.UNKNOWN]) predicate_h = humanize_predicate(predicate) obj_display = obj if obj not in ("", "") else "..." + # Masterful 3-language depth framing on the articulation side. + # Depth travels with the shared GraphNode from resolve_entry grounding. + if language and root and language != "en": + if language == "he": + depth_note = f" (Hebrew root: {root})" + elif language in ("grc", "el"): + depth_note = f" (Koine Greek: {root})" + else: + depth_note = f" ({language} root: {root})" + + # For definition-style intents, highlight the term itself. + # For others, qualify the object referent. + if intent in (IntentTag.DEFINITION, IntentTag.RECALL, IntentTag.VERIFICATION): + subject = f"{subject}{depth_note}" + else: + obj_display = f"{obj_display}{depth_note}" + return template.format( subject=subject, predicate_h=predicate_h, diff --git a/recognition/connector.py b/recognition/connector.py index ba0aa8fd..b5a8cdb5 100644 --- a/recognition/connector.py +++ b/recognition/connector.py @@ -22,6 +22,9 @@ def epistemic_node_to_graph_node( *, source_intent: IntentTag, node_id: str | None = None, + language: str | None = None, + root: str | None = None, + morphology_id: str | None = None, ) -> GraphNode: """Derive a generation-side GraphNode from an admitted EpistemicNode. @@ -31,6 +34,10 @@ def epistemic_node_to_graph_node( subject ← bundle["agent"].value predicate ← bundle["relation"].value obj ← "{count.value} {unit.value}" + + Optional depth params allow 3-lang (Hebrew/Greek) morphology/root info + from recognition to flow into the shared PropositionGraph for + comprehension/articulation/reasoning. """ outcome = node.recognition_outcome if outcome.state != EVIDENCED: @@ -60,6 +67,9 @@ def epistemic_node_to_graph_node( predicate=predicate, obj=obj, source_intent=source_intent, + language=language, + root=root, + morphology_id=morphology_id, ) diff --git a/session/graph.py b/session/graph.py index 9440bde4..64140cc4 100644 --- a/session/graph.py +++ b/session/graph.py @@ -8,6 +8,7 @@ walk those edges with true BFS distance, not traversal ordinal. from __future__ import annotations +from collections import deque from dataclasses import dataclass, field from typing import Any, Sequence @@ -146,10 +147,10 @@ class SessionGraph: if from_turn < 0 or from_turn >= len(self._nodes): raise IndexError(f"from_turn out of range: {from_turn}") visited: set[int] = {from_turn} - queue: list[tuple[int, int]] = [(1, idx) for idx in self._nodes[from_turn].backward_edges] + queue: deque[tuple[int, int]] = deque((1, idx) for idx in self._nodes[from_turn].backward_edges) result: list[tuple[int, TurnNode]] = [] while queue: - distance, idx = queue.pop(0) + distance, idx = queue.popleft() if distance > max_depth: continue if idx in visited or idx >= len(self._nodes) or idx < 0: diff --git a/tests/test_oov_pipeline.py b/tests/test_oov_pipeline.py index 8a3456a0..8f624802 100644 --- a/tests/test_oov_pipeline.py +++ b/tests/test_oov_pipeline.py @@ -22,6 +22,7 @@ from pathlib import Path import pytest from chat.runtime import ChatRuntime +from core.cognition.pipeline import CognitiveTurnPipeline from teaching.oov_gaps import OOVGap, aggregate_oov_gaps from teaching.oov_promotion import OOVPromotion, promote_oov_gaps from teaching.oov_sink import ( @@ -188,6 +189,35 @@ def test_since_filter(tmp_path: Path) -> None: assert rows[0].sample_candidate_ids == ("may",) +# --- Phase C characterization: geometric anti-unification hook --- + +def test_pipeline_oov_geometric_context_hook() -> None: + """Phase C atomic instrumentation provides read-only graph context for OOV. + + This is the hook for future exact CGA sub-graph anti-unification. + The field is purely observational; it must not affect surfaces, trace_hash, + or any user-visible behaviour. Populated when OOV or unresolved slots + are present in the PropositionGraph. + """ + pipeline = CognitiveTurnPipeline(runtime=ChatRuntime()) + result = pipeline.run("What is photosynthesis?", max_tokens=2) + + # For a clear OOV like "photosynthesis", the context should be present + # with unresolved topology from the substrate graph. + assert result.oov_geometric_context is not None + ctx = result.oov_geometric_context + assert "unresolved_topology" in ctx + assert isinstance(ctx["unresolved_topology"], tuple) + assert len(ctx["unresolved_topology"]) >= 1 + assert ctx.get("geometric_probe_performed") is False + assert "Hook for geometric anti-unification" in ctx.get("note", "") + # Intent should be captured for context. + assert ctx.get("intent_tag") in ("definition", "unknown", "recall") # tolerant for classifier + # 3-lang OOV bridge: node_depths always present (empty if no depth langs on nodes) + assert "node_depths" in ctx + assert isinstance(ctx["node_depths"], dict) + + def test_malformed_lines_skipped(tmp_path: Path) -> None: sink = tmp_path / "2026" / "2026-05.jsonl" sink.parent.mkdir(parents=True, exist_ok=True) diff --git a/tests/test_realizer_grounded_authority_flag.py b/tests/test_realizer_grounded_authority_flag.py index 61c7fe0a..448f5399 100644 --- a/tests/test_realizer_grounded_authority_flag.py +++ b/tests/test_realizer_grounded_authority_flag.py @@ -35,6 +35,7 @@ prerequisite for enabling this flag in production. from __future__ import annotations +from chat.pack_resolver import LexicalResolution, resolve_entry, resolve_gloss from chat.runtime import ChatRuntime from core.cognition import CognitiveTurnPipeline from core.config import DEFAULT_CONFIG, RuntimeConfig @@ -71,6 +72,8 @@ def test_flag_on_runs_without_crashing() -> None: clears ``_is_useful_surface`` (or falls back to the runtime path), so the result is well-formed even though the surface contents may differ from the default until Phase A's fluency parity lands.""" + from chat.pack_resolver import resolve_gloss + rt = ChatRuntime(config=RuntimeConfig(realizer_grounded_authority=True)) pipeline = CognitiveTurnPipeline(runtime=rt) result = pipeline.run("What is truth?", max_tokens=4) @@ -78,3 +81,94 @@ def test_flag_on_runs_without_crashing() -> None: assert isinstance(result.surface, str) assert result.surface # non-empty assert result.trace_hash # hashed + + # Masterful structured grounding: when pack-resident, the graph obj + # must come directly from resolve_gloss (raw authoritative gloss), + # not from parsed surface text. + if result.proposition_graph and result.proposition_graph.nodes: + gloss_entry = resolve_gloss("truth") + if gloss_entry: + _, _, expected_gloss = gloss_entry + actual_obj = result.proposition_graph.nodes[0].obj + # The grounded obj should match the raw pack gloss (or be + # derived from it in the definition frame). We assert it is + # no longer the "" sentinel and carries the + # expected content. + assert actual_obj != "" + assert expected_gloss in actual_obj or actual_obj in expected_gloss + + # Depth now flows on the graph node itself (bidirectional spine). + # resolve_entry + enrichment in pipeline ensures language (and root + # for he/grc) ride on the PropositionGraph nodes for realize + trace. + node = result.proposition_graph.nodes[0] + # After resolve_entry + node enrichment + ground_graph, depth is on the node. + assert getattr(node, "language", None) == "en" # truth resolved via entry path + # he/grc would carry root + morphology_id (tested via direct probe + he/grc pack tests) + + +def test_resolve_entry_provides_3lang_depth_for_bidirectional_use() -> None: + """LexicalResolution is the shared immutable artifact. + + Comprehension (pipeline grounding) and articulation (realizer) plus + internal reasoning can all consume the same depth-carrying structure + without duplication. Hebrew root + Greek precision are now first-class. + """ + # English base (always available) + en_res = resolve_entry("truth") + assert en_res is not None + assert isinstance(en_res, LexicalResolution) + assert en_res.language in ("en", "en") # base + assert en_res.gloss is not None + + # Hebrew depth language (explicit pack for the depth pack) + he_pack = ("he_logos_micro_v1", "en_collapse_anchors_v1") + he_res = resolve_entry("אמת", pack_ids=he_pack) # emet = truth/faithfulness + if he_res is not None: # pack may or may not be mounted in all envs; graceful + assert isinstance(he_res, LexicalResolution) + assert he_res.language == "he" + assert he_res.root is not None # "א-מ-ת" or equivalent + assert he_res.gloss is not None or he_res.pos # depth present + + # Greek likewise (structure test) + grc_pack = ("grc_logos_micro_v1", "en_collapse_anchors_v1") + grc_res = resolve_entry("λόγος", pack_ids=grc_pack) # logos + if grc_res is not None: + assert grc_res.language == "grc" or grc_res.language == "el" # Greek + assert grc_res.morphology_id is not None or grc_res.root is not None + + # Old resolve_gloss path remains available (compat for articulation etc.) + old = resolve_gloss("truth") + assert old is not None or en_res is not None # at least one works + + +def test_hebrew_depth_full_turn_under_grounded_authority() -> None: + """Full turn assertion for 3-lang depth under realizer_grounded_authority flag. + + Exercises the complete spine path: + classify -> graph -> resolve_entry (with depth packs) -> enrich GraphNode + -> ground_graph -> realize_semantic (depth consumed for richer surface). + """ + rt = ChatRuntime(config=RuntimeConfig(realizer_grounded_authority=True)) + pipeline = CognitiveTurnPipeline(runtime=rt) + result = pipeline.run("What is אמת?", max_tokens=4) + + assert isinstance(result.surface, str) + assert result.surface # non-empty + assert result.trace_hash + + if result.proposition_graph and result.proposition_graph.nodes: + node = result.proposition_graph.nodes[0] + if node.subject.strip() == "אמת": + # Depth carried on the node (comprehend side) via resolve_entry + enrich + assert getattr(node, "language", None) == "he" + assert getattr(node, "root", None) is not None + + # Even if final user surface is fallback (not learned for this term), + # the graph carries depth, and realize_semantic on it produces the + # richer 3-lang framed surface (articulation side consumption). + from generate.graph_planner import plan_articulation + from generate.realizer import realize_semantic + target = plan_articulation(result.proposition_graph) + realized = realize_semantic(target, result.proposition_graph) + assert "(Hebrew root:" in realized.surface + assert "אמת" in realized.surface diff --git a/tests/test_semantic_realizer_integration.py b/tests/test_semantic_realizer_integration.py index cbf797b3..425e9696 100644 --- a/tests/test_semantic_realizer_integration.py +++ b/tests/test_semantic_realizer_integration.py @@ -72,6 +72,45 @@ class TestSemanticTemplates: assert "" not in surface assert "..." in surface + def test_depth_language_enriches_articulation(self) -> None: + """3-core-language depth from GraphNode is consumed on articulation side. + + Hebrew root and Koine Greek are framed etymologically when present + on the enriched PropositionGraph (bidirectional Logos substrate). + """ + # Hebrew + he_surface = render_semantic( + intent=IntentTag.DEFINITION, + subject="אמת", + predicate="is_defined_as", + obj="truth, firmness, or faithfulness", + language="he", + root="א-מ-ן", + ) + assert "אמת (Hebrew root: א-מ-ן)" in he_surface + assert "is defined as" in he_surface + + # Greek (Logos) + grc_surface = render_semantic( + intent=IntentTag.DEFINITION, + subject="λόγος", + predicate="is_defined_as", + obj="word, reason, structuring principle", + language="grc", + root="λόγ-", + ) + assert "λόγος (Koine Greek: λόγ-)" in grc_surface + + # English baseline unchanged (no note) + en_surface = render_semantic( + intent=IntentTag.DEFINITION, + subject="truth", + predicate="is_defined_as", + obj="coherence", + ) + assert "(Hebrew" not in en_surface + assert "(Koine" not in en_surface + # --------------------------------------------------------------------------- # Unit tests: realize_semantic diff --git a/tests/test_surface_resolution.py b/tests/test_surface_resolution.py index fda866cf..f3016103 100644 --- a/tests/test_surface_resolution.py +++ b/tests/test_surface_resolution.py @@ -1,6 +1,8 @@ from __future__ import annotations from core.cognition.surface_resolution import resolve_surface +from generate.graph_planner import GraphNode, PropositionGraph +from generate.intent import IntentTag def test_runtime_canonical_surface_has_base_precedence() -> None: @@ -81,3 +83,72 @@ def test_folds_stand_alone_when_base_surface_is_empty() -> None: assert resolved.articulation_surface == "walk chain — compose transfer" assert resolved.authority == "runtime" assert resolved.fold_sources == ("walk", "compose") + + +# --- Shadow Coherence Gate supremacy tests (Phase A) --- + +def _mk_grounded_graph() -> PropositionGraph: + n = GraphNode( + node_id="n0", + subject="Evidence", + predicate="supports", + obj="Hypothesis", + source_intent=IntentTag.DEFINITION, + ) + return PropositionGraph(nodes=(n,), edges=()) + + +def _mk_pending_graph() -> PropositionGraph: + n = GraphNode( + node_id="n0", + subject="Evidence", + predicate="supports", + obj="", + source_intent=IntentTag.DEFINITION, + ) + return PropositionGraph(nodes=(n,), edges=()) + + +def test_substrate_supreme_when_graph_fully_grounded_and_no_gate() -> None: + """The strict guard must grant 'substrate_realizer' authority.""" + g = _mk_grounded_graph() + resolved = resolve_surface( + response_surface="runtime", + response_articulation_surface="runtime art", + realized_surface="The evidence supports the hypothesis.", + realizer_useful=True, + gate_fired=False, + proposition_graph=g, + ) + assert resolved.authority == "substrate_realizer" + assert resolved.surface == "The evidence supports the hypothesis." + + +def test_pending_graph_withholds_substrate_authority_even_if_useful() -> None: + """Pending slots mean substrate does not yet earn authority (bypass hazard path).""" + g = _mk_pending_graph() + resolved = resolve_surface( + response_surface="runtime", + response_articulation_surface="runtime art", + realized_surface="Evidence supports ...", + realizer_useful=True, + gate_fired=False, + proposition_graph=g, + ) + # Because not supreme, the old shim still fires for useful -> "realizer" + # (transitional). The hazard is computed in the *pipeline* caller. + assert resolved.authority == "realizer" + + +def test_gate_fired_still_blocks_substrate_even_for_grounded_graph() -> None: + g = _mk_grounded_graph() + resolved = resolve_surface( + response_surface="I don't have field coordinates for that yet.", + response_articulation_surface="...", + realized_surface="The evidence supports the hypothesis.", + realizer_useful=True, + gate_fired=True, + proposition_graph=g, + ) + assert resolved.authority == "runtime" + assert resolved.surface == "I don't have field coordinates for that yet."