feat: integrate 3-core-language depth into PropositionGraph spine for bidirectional unification

- 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.
This commit is contained in:
Shay 2026-07-05 20:19:02 -07:00
parent 60b30f6cdd
commit 0c75be7c15
17 changed files with 1010 additions and 79 deletions

View file

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

View file

@ -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, "", "<pending>", "<prior>") 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 ("", "<pending>") 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,
)
# ------------------------------------------------------------------

View file

@ -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

View file

@ -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 <pending> 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 <pending>).
- 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

View file

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

View file

@ -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:

View file

@ -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

View file

@ -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 <pending>).
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 "<pending>" 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, "", "<pending>"):
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, "", "<pending>") 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 "<pending>",
source_intent=intent.tag,
)
right = GraphNode(
node_id="p1",
subject=intent.secondary_subject or "<pending>",
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 "<prior>",
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 "<pending>",
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 "<pending>",
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="<pending>",
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 "<prior>",
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="<pending>",
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 <pending> obj slots with recalled words from vault recall.
Each node whose obj is '<pending>' gets the next available recalled
word. If there are more nodes than words, remaining slots stay as
'<pending>'. 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 == "<pending>" 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,

View file

@ -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))

View file

@ -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:

View file

@ -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 ("<pending>", "<prior>") 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,

View file

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

View file

@ -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:

View file

@ -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)

View file

@ -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 "<pending>" sentinel and carries the
# expected content.
assert actual_obj != "<pending>"
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

View file

@ -72,6 +72,45 @@ class TestSemanticTemplates:
assert "<pending>" 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

View file

@ -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="<pending>",
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."