From 0b330308523b643f95632f172f202a22fcd5f4f6 Mon Sep 17 00:00:00 2001 From: Shay Date: Tue, 19 May 2026 11:19:59 -0700 Subject: [PATCH] feat(grounding): structured GroundedFact accessors for discourse planner MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Step 3 of the discourse-planner sequencing. Adds generate/grounding_accessors.py: * pack_grounded_facts(lemma) -> tuple[GroundedFact, ...] * teaching_grounded_chains(lemma) -> tuple[GroundedFact, ...] * cross_pack_grounded_chains(lemma) -> tuple[GroundedFact, ...] * grounding_bundle_for(lemma) -> GroundingBundle All four reuse the existing data substrate (chat.pack_resolver, chat.teaching_grounding._all_chains_index, chat.cross_pack_grounding chain accessors) — no new loader, no new I/O, no string composer touched. Pack facts emit one `is_defined_as` per gloss + one `belongs_to` per semantic_domain; teaching/cross-pack chains emit verbatim (subject, connective, object) triples; everything sorted by GroundedFact.sort_key for canonical determinism. 21 new tests pin: pack/teaching/cross-pack accessor shape, canonical sort order, verbatim object invariant (no synthesis), source_id points back into real artifact, bundle composition combines all three sources with pack-first priority, and doctrine invariants (no *_grounded_surface composer imported, no chat.runtime imported). Verification: * 21/21 new accessor tests pass. * smoke suite 67/67. * cognition eval byte-identical: public 100/100/91.7/100, holdout 100/100/83.3/100. --- generate/grounding_accessors.py | 221 ++++++++++++++++++++++++++++ tests/test_grounding_accessors.py | 236 ++++++++++++++++++++++++++++++ 2 files changed, 457 insertions(+) create mode 100644 generate/grounding_accessors.py create mode 100644 tests/test_grounding_accessors.py diff --git a/generate/grounding_accessors.py b/generate/grounding_accessors.py new file mode 100644 index 00000000..e152a5a7 --- /dev/null +++ b/generate/grounding_accessors.py @@ -0,0 +1,221 @@ +"""Structured grounding accessors for the discourse planner. + +Step 3 of the discourse-planner sequencing. These accessors convert +the existing grounding substrate (ratified packs, reviewed teaching +corpora, reviewed cross-pack corpora) into typed +:class:`generate.discourse_planner.GroundedFact` tuples that the +planner can compose into a :class:`DiscoursePlan`. + +Doctrine invariants this module respects: + +* **Reuse, do not reimplement.** Pack lemmas come from + ``chat.pack_resolver.resolve_lemma`` and ``resolve_gloss``; teaching + chains come from ``chat.teaching_grounding._all_chains_index``; + cross-pack chains come from + ``chat.cross_pack_grounding.cross_pack_chains_for_subject``. No new + loader, no new I/O path. +* **Existing string composers untouched.** This module does not + import from any ``pack_grounded_*`` / ``teaching_grounded_*`` / + ``cross_pack_grounded_*`` *surface* function — it consults only the + underlying data accessors that those composers already use. + ``chat/runtime.py`` is not imported. +* **Canonical ordering.** Returned facts are sorted by their + ``GroundedFact.sort_key`` so two equal calls produce byte-identical + tuples. This is the precondition that the source-order + characterization test (``test_grounding_source_characterization``) + pins for downstream determinism. +* **No content synthesis.** Every fact's ``obj`` is either a verbatim + pack ``semantic_domains`` string, a verbatim pack ``gloss`` string, + or a verbatim teaching/cross-pack chain object lemma. Never a + template, never an LLM string, never an approximation. +""" + +from __future__ import annotations + +from chat.cross_pack_grounding import ( + CROSS_PACK_CORPUS_ID, + cross_pack_chains_for_object, + cross_pack_chains_for_subject, +) +from chat.pack_resolver import ( + DEFAULT_RESOLVABLE_PACK_IDS, + resolve_gloss, + resolve_lemma, +) +from chat.teaching_grounding import _all_chains_index +from generate.discourse_planner import ( + FactSource, + GroundedFact, + GroundingBundle, +) + +# Canonical predicate vocabulary the accessors emit for pack facts. +# These mirror predicate forms already used in the existing +# pack-grounded composers and the semantic-templates module, so the +# planner's downstream graph mapping uses tokens the realizer knows. +_PACK_BELONGS_TO = "belongs_to" +_PACK_IS_DEFINED_AS = "is_defined_as" + + +def pack_grounded_facts( + lemma: str, + pack_ids: tuple[str, ...] = DEFAULT_RESOLVABLE_PACK_IDS, +) -> tuple[GroundedFact, ...]: + """Return canonical, sorted ``GroundedFact`` tuple for *lemma*. + + Emits one ``is_defined_as`` fact per pack gloss (when the pack + ships a gloss for the lemma) and one ``belongs_to`` fact per + ``semantic_domains`` entry. First-match-wins across *pack_ids* + matches ``resolve_lemma`` precedence (in-pack cognition first by + default), so the lemma is grounded in exactly one pack. + """ + + if not lemma or not isinstance(lemma, str): + return () + key = lemma.strip().lower() + if not key: + return () + resolved = resolve_lemma(key, pack_ids=pack_ids) + if resolved is None: + return () + pack_id, domains = resolved + facts: list[GroundedFact] = [] + gloss = resolve_gloss(key, pack_ids=(pack_id,)) + if gloss is not None: + _, _, gloss_text = gloss + facts.append( + GroundedFact( + subject=key, + predicate=_PACK_IS_DEFINED_AS, + obj=gloss_text, + source=FactSource.PACK, + source_id=f"{pack_id}:{key}#gloss", + ) + ) + for idx, domain in enumerate(domains): + facts.append( + GroundedFact( + subject=key, + predicate=_PACK_BELONGS_TO, + obj=str(domain), + source=FactSource.PACK, + source_id=f"{pack_id}:{key}#domain:{idx}", + ) + ) + return tuple(sorted(facts, key=GroundedFact.sort_key)) + + +def teaching_grounded_chains( + lemma: str, +) -> tuple[GroundedFact, ...]: + """Return canonical teaching chains rooted on *lemma* as facts. + + Pulls from the aggregated teaching index (every registered teaching + corpus, ADR-0064 cross-pack teaching). Both ``cause`` and + ``verification`` intents are surfaced so the discourse planner can + select either depending on response mode. + """ + + if not lemma or not isinstance(lemma, str): + return () + key = lemma.strip().lower() + if not key: + return () + aggregated = _all_chains_index() + facts: list[GroundedFact] = [] + for (subject, _intent), chain in aggregated.items(): + if subject != key: + continue + facts.append( + GroundedFact( + subject=chain.subject, + predicate=chain.connective, + obj=chain.object, + source=FactSource.TEACHING, + source_id=f"{chain.corpus_id}#{chain.chain_id}", + ) + ) + return tuple(sorted(facts, key=GroundedFact.sort_key)) + + +def cross_pack_grounded_chains( + lemma: str, + *, + include_object_view: bool = True, +) -> tuple[GroundedFact, ...]: + """Return canonical cross-pack chains touching *lemma* as facts. + + Surfaces chains where *lemma* is the subject (forward view) and, + when ``include_object_view`` is True (default), chains where *lemma* + is the object (reverse view used by EXAMPLE intent). + """ + + if not lemma or not isinstance(lemma, str): + return () + key = lemma.strip().lower() + if not key: + return () + raw: list[GroundedFact] = [] + for chain in cross_pack_chains_for_subject(key): + raw.append( + GroundedFact( + subject=chain.subject, + predicate=chain.connective, + obj=chain.object, + source=FactSource.TEACHING, + source_id=f"{CROSS_PACK_CORPUS_ID}#{chain.chain_id}", + ) + ) + if include_object_view: + for chain in cross_pack_chains_for_object(key): + raw.append( + GroundedFact( + subject=chain.subject, + predicate=chain.connective, + obj=chain.object, + source=FactSource.TEACHING, + source_id=f"{CROSS_PACK_CORPUS_ID}#{chain.chain_id}", + ) + ) + # Dedupe by sort_key — forward+reverse views can repeat the same + # chain when lemma == subject == object (rare but possible). + seen: set[tuple[int, str, str, str, str]] = set() + deduped: list[GroundedFact] = [] + for fact in raw: + sk = fact.sort_key() + if sk in seen: + continue + seen.add(sk) + deduped.append(fact) + return tuple(sorted(deduped, key=GroundedFact.sort_key)) + + +def grounding_bundle_for( + lemma: str, + pack_ids: tuple[str, ...] = DEFAULT_RESOLVABLE_PACK_IDS, +) -> GroundingBundle: + """Compose every grounding source into one bundle for the planner. + + Convenience constructor that the runtime adapter calls when + building a :class:`DiscoursePlan` input. Pack facts come first + (canonical priority), then aggregated teaching chains, then + cross-pack chains. The bundle's own ``sorted_facts`` re-sorts on + read, so callers always see a canonical view. + """ + + bundle = GroundingBundle( + facts=( + *pack_grounded_facts(lemma, pack_ids=pack_ids), + *teaching_grounded_chains(lemma), + *cross_pack_grounded_chains(lemma), + ) + ) + return bundle + + +__all__ = [ + "cross_pack_grounded_chains", + "grounding_bundle_for", + "pack_grounded_facts", + "teaching_grounded_chains", +] diff --git a/tests/test_grounding_accessors.py b/tests/test_grounding_accessors.py new file mode 100644 index 00000000..00023ab3 --- /dev/null +++ b/tests/test_grounding_accessors.py @@ -0,0 +1,236 @@ +"""Tests for ``generate/grounding_accessors.py`` (step 3). + +These tests pin the structured-grounding contract: + +* Every fact returned carries a canonical ``FactSource`` and a + ``source_id`` that points back into a real artifact (pack lemma, + teaching chain id, cross-pack chain id). +* Returned tuples are canonically sorted — equal calls produce + byte-identical tuples. +* No content synthesis: ``obj`` values are verbatim pack + ``semantic_domains`` strings, verbatim pack glosses, or verbatim + teaching/cross-pack chain object lemmas. +* The accessors do not import or call any ``*_grounded_surface`` + composer — they consult only the underlying data sources, so the + existing string composers stay untouched. + +The grounding-source characterization sidecar (step 1) covers the +underlying data substrate; this file pins the *adapter* layer that +converts that substrate into :class:`GroundedFact` tuples. +""" + +from __future__ import annotations + +import inspect + +from generate.discourse_planner import ( + FactSource, + GroundedFact, + GroundingBundle, +) +from generate.grounding_accessors import ( + cross_pack_grounded_chains, + grounding_bundle_for, + pack_grounded_facts, + teaching_grounded_chains, +) + + +# --------------------------------------------------------------------------- +# Pack accessor +# --------------------------------------------------------------------------- + + +class TestPackGroundedFacts: + def test_returns_facts_for_pack_lemma(self) -> None: + facts = pack_grounded_facts("truth") + assert len(facts) > 0 + assert all(f.source is FactSource.PACK for f in facts) + assert all(f.subject == "truth" for f in facts) + + def test_returns_empty_for_unknown_lemma(self) -> None: + assert pack_grounded_facts("definitely_not_a_lemma_xyz") == () + assert pack_grounded_facts("") == () + assert pack_grounded_facts(" ") == () + + def test_source_id_points_into_resolving_pack(self) -> None: + facts = pack_grounded_facts("truth") + assert all(":" in f.source_id for f in facts) + # Every source_id is namespaced by the resolving pack id; + # cognition pack comes first by default precedence so "truth" + # resolves there. + assert all( + f.source_id.startswith("en_core_cognition_v1:") + for f in facts + ) + + def test_facts_are_canonically_sorted(self) -> None: + facts = pack_grounded_facts("truth") + assert facts == tuple(sorted(facts, key=GroundedFact.sort_key)) + + def test_belongs_to_obj_is_verbatim_pack_domain(self) -> None: + # Every "belongs_to" obj must appear as a literal string in the + # pack's semantic_domains for the lemma — no synthesis. + from chat.pack_resolver import resolve_lemma + + facts = pack_grounded_facts("truth") + resolved = resolve_lemma("truth") + assert resolved is not None + _, domains = resolved + belongs_to_objs = { + f.obj for f in facts if f.predicate == "belongs_to" + } + assert belongs_to_objs <= set(domains) + + def test_predicates_are_canonical(self) -> None: + facts = pack_grounded_facts("truth") + predicates = {f.predicate for f in facts} + assert predicates <= {"belongs_to", "is_defined_as"} + + +# --------------------------------------------------------------------------- +# Teaching accessor +# --------------------------------------------------------------------------- + + +class TestTeachingGroundedChains: + def test_returns_chains_for_subject_with_reviewed_teaching(self) -> None: + facts = teaching_grounded_chains("knowledge") + assert len(facts) > 0 + assert all(f.source is FactSource.TEACHING for f in facts) + assert all(f.subject == "knowledge" for f in facts) + + def test_returns_empty_for_unknown_subject(self) -> None: + assert teaching_grounded_chains("definitely_not_a_subject_xyz") == () + assert teaching_grounded_chains("") == () + + def test_source_id_points_into_corpus(self) -> None: + facts = teaching_grounded_chains("knowledge") + # Format is "#" + assert all("#" in f.source_id for f in facts) + for f in facts: + corpus_id, chain_id = f.source_id.split("#", 1) + assert corpus_id != "" + assert chain_id != "" + + def test_facts_are_canonically_sorted(self) -> None: + facts = teaching_grounded_chains("knowledge") + assert facts == tuple(sorted(facts, key=GroundedFact.sort_key)) + + def test_obj_is_verbatim_chain_object(self) -> None: + from chat.teaching_grounding import _all_chains_index + + facts = teaching_grounded_chains("knowledge") + chains = _all_chains_index() + # Every fact's (subject, predicate, obj) must match exactly one + # aggregated chain. + chain_triples = { + (chain.subject, chain.connective, chain.object) + for chain in chains.values() + } + for f in facts: + assert (f.subject, f.predicate, f.obj) in chain_triples + + +# --------------------------------------------------------------------------- +# Cross-pack accessor +# --------------------------------------------------------------------------- + + +class TestCrossPackGroundedChains: + def test_returns_chains_for_subject_with_cross_pack_data(self) -> None: + facts = cross_pack_grounded_chains("parent") + assert len(facts) > 0 + assert all(f.source is FactSource.TEACHING for f in facts) + + def test_object_view_can_be_disabled(self) -> None: + forward_and_reverse = cross_pack_grounded_chains( + "memory", include_object_view=True + ) + forward_only = cross_pack_grounded_chains( + "memory", include_object_view=False + ) + assert len(forward_only) <= len(forward_and_reverse) + + def test_returns_empty_for_unknown_lemma(self) -> None: + assert cross_pack_grounded_chains("definitely_not_a_lemma_xyz") == () + + def test_facts_are_canonically_sorted_and_deduped(self) -> None: + facts = cross_pack_grounded_chains("parent") + assert facts == tuple(sorted(facts, key=GroundedFact.sort_key)) + # Dedupe by sort_key — no duplicate facts allowed. + keys = [f.sort_key() for f in facts] + assert len(keys) == len(set(keys)) + + +# --------------------------------------------------------------------------- +# Bundle composer +# --------------------------------------------------------------------------- + + +class TestGroundingBundleFor: + def test_combines_all_three_sources(self) -> None: + bundle = grounding_bundle_for("knowledge") + assert isinstance(bundle, GroundingBundle) + sources = {f.source for f in bundle.sorted_facts()} + # Knowledge is a cognition-pack lemma with reviewed teaching + # chains rooted on it. + assert FactSource.PACK in sources + assert FactSource.TEACHING in sources + + def test_pack_facts_come_before_teaching_in_sorted_view(self) -> None: + bundle = grounding_bundle_for("knowledge") + sources_in_order = [f.source for f in bundle.sorted_facts()] + # Pack < Teaching by canonical priority. First non-pack index + # must come AFTER the last pack index. + try: + first_teaching = sources_in_order.index(FactSource.TEACHING) + except ValueError: + return + for src in sources_in_order[:first_teaching]: + assert src is FactSource.PACK + + def test_empty_bundle_for_unknown_lemma(self) -> None: + bundle = grounding_bundle_for("definitely_not_a_lemma_xyz") + assert bundle.is_empty() + + def test_bundle_is_deterministic(self) -> None: + a = grounding_bundle_for("truth").sorted_facts() + b = grounding_bundle_for("truth").sorted_facts() + assert a == b + + +# --------------------------------------------------------------------------- +# Doctrine invariants +# --------------------------------------------------------------------------- + + +class TestAccessorDoctrine: + def test_no_string_composer_imports(self) -> None: + # The accessors must not pull in any *_grounded_surface + # composer — they consult only the underlying data accessors. + import generate.grounding_accessors as ga + + src = inspect.getsource(ga) + forbidden = ( + "pack_grounded_surface", + "pack_grounded_definition", + "pack_grounded_comparison", + "pack_grounded_correction", + "pack_grounded_procedure", + "teaching_grounded_surface", + "cross_pack_grounded_surface", + "narrative_grounded_surface", + "example_grounded_surface", + ) + for token in forbidden: + assert token not in src, ( + f"grounding_accessors must not depend on {token}" + ) + + def test_no_runtime_imports(self) -> None: + import generate.grounding_accessors as ga + + src = inspect.getsource(ga) + assert "chat.runtime" not in src + assert "ChatRuntime" not in src