diff --git a/generate/discourse_planner.py b/generate/discourse_planner.py index ea8eccd0..8ae15b3a 100644 --- a/generate/discourse_planner.py +++ b/generate/discourse_planner.py @@ -279,9 +279,18 @@ _MODE_BUDGETS: dict[ResponseMode, tuple[int, int]] = { ResponseMode.EXPLAIN: (1, 3), ResponseMode.PARAGRAPH: (1, 5), ResponseMode.EXAMPLE: (1, 3), - ResponseMode.WALKTHROUGH: (1, 1), + # WALKTHROUGH v1: ≤ 4 hops along the teaching-chain graph. The + # planner walks ``(subject, *, object) → (object, *, *)`` + # starting from the anchor and follows up to three additional + # hops (4 moves total including the anchor). When no chain is + # available the v1 implementation falls back to the expository + # plan shape (EXPLAIN budget) rather than fabricating steps — + # operator-chain WALKTHROUGH is deferred to a follow-up ADR. + ResponseMode.WALKTHROUGH: (1, 4), } +_WALKTHROUGH_MAX_HOPS = 3 # 3 hops after the anchor = 4 moves total + def _select_anchor( intent: DialogueIntent, @@ -379,6 +388,111 @@ def _select_transition( return None +def _plan_walkthrough( + intent: DialogueIntent, + mode: ResponseMode, + bundle: GroundingBundle, + anchor_fact: GroundedFact, + moves: list[DiscourseMove], + used: set[tuple[int, str, str, str, str]], +) -> DiscoursePlan: + """WALKTHROUGH v1 — sequential teaching-chain walk. + + Starting from the anchor's subject, follow up to + ``_WALKTHROUGH_MAX_HOPS`` hops along teaching-chain edges + ``(subject, *, object) → (object, *, *)``. Each hop is one + ``RELATION`` move; the final hop becomes a ``CLOSURE`` move. + + Cycle-safe: never re-emits a fact already in *used*. Bounded + depth. When the substrate has no chain rooted on the anchor (or + the walk stalls before any hop), the v1 implementation falls + back to the expository (EXPLAIN) plan shape rather than + fabricating walk steps. + """ + + given_lemmas: list[str] = [anchor_fact.subject] + current_subject = anchor_fact.subject + + walked_facts: list[GroundedFact] = [] + for _hop in range(_WALKTHROUGH_MAX_HOPS): + next_fact: GroundedFact | None = None + for fact in bundle.facts_by_source(FactSource.TEACHING): + if fact.sort_key() in used: + continue + if fact.subject == current_subject: + next_fact = fact + break + if next_fact is None: + break + walked_facts.append(next_fact) + used.add(next_fact.sort_key()) + current_subject = next_fact.obj.strip().lower() + + if not walked_facts: + # No teaching-chain substrate — fall back to expository plan + # rather than fabricating walk steps. Anchor + (SUPPORT) + + # (RELATION) shape preserves the "walkthrough" intent without + # claiming a process the substrate cannot support. + return _plan_walkthrough_fallback( + intent, bundle, anchor_fact, moves, used + ) + + # Emit walked facts as RELATION moves with the final one becoming + # CLOSURE so the rendered surface terminates explicitly. + for idx, fact in enumerate(walked_facts): + kind = ( + DiscourseMoveKind.CLOSURE + if idx == len(walked_facts) - 1 + else DiscourseMoveKind.RELATION + ) + moves.append( + DiscourseMove( + kind=kind, + topic=fact.subject, + given=tuple(given_lemmas), + new=(fact.obj,), + relation_to_previous=Relation.SEQUENCE, + fact=fact, + ) + ) + given_lemmas.append(fact.obj) + + return DiscoursePlan(intent=intent, mode=mode, moves=tuple(moves)) + + +def _plan_walkthrough_fallback( + intent: DialogueIntent, + bundle: GroundingBundle, + anchor_fact: GroundedFact, + moves: list[DiscourseMove], + used: set[tuple[int, str, str, str, str]], +) -> DiscoursePlan: + """Fallback shape when no teaching chain is available for + WALKTHROUGH. Emits an ANCHOR + (SUPPORT) plan — the + ``ResponseMode`` stays WALKTHROUGH on the resulting plan so + callers can tell the planner attempted a walkthrough but + degraded honestly. + """ + + given_lemmas: list[str] = [anchor_fact.subject] + support_fact = _select_support(anchor_fact, bundle) + if support_fact is not None and support_fact.sort_key() not in used: + moves.append( + DiscourseMove( + kind=DiscourseMoveKind.SUPPORT, + topic=support_fact.subject, + given=tuple(given_lemmas), + new=(support_fact.obj,), + relation_to_previous=Relation.ELABORATION, + fact=support_fact, + ) + ) + used.add(support_fact.sort_key()) + return DiscoursePlan( + intent=intent, mode=ResponseMode.WALKTHROUGH, moves=tuple(moves) + ) + + def plan_discourse( intent: DialogueIntent, mode: ResponseMode, @@ -441,7 +555,12 @@ def plan_discourse( ] used: set[tuple[int, str, str, str, str]] = {anchor_fact.sort_key()} _, max_moves = _move_budget(mode) - if max_moves <= 1 or mode is ResponseMode.WALKTHROUGH: + + # WALKTHROUGH v1 — sequential teaching-chain walk. + if mode is ResponseMode.WALKTHROUGH: + return _plan_walkthrough(intent, mode, bundle, anchor_fact, moves, used) + + if max_moves <= 1: return DiscoursePlan(intent=intent, mode=mode, moves=tuple(moves)) given_lemmas: list[str] = [anchor_fact.subject] diff --git a/generate/intent.py b/generate/intent.py index d389e7a3..480c9e3d 100644 --- a/generate/intent.py +++ b/generate/intent.py @@ -165,6 +165,18 @@ _RULES: tuple[tuple[re.Pattern[str], IntentTag], ...] = ( re.compile(r"^paragraph\s+(?:about|on)\s+", re.IGNORECASE), IntentTag.DEFINITION, ), + # WALKTHROUGH-shape requests — semantic intent is "describe X step + # by step". Routes to DEFINITION so the grounded substrate fires + # on X; ``ResponseMode.WALKTHROUGH`` carries the walk depth and + # selects the sequential teaching-chain plan budget at planning + # time. Same orthogonality discipline as the EXPLAIN rule. + ( + re.compile( + r"^walk\s+(?:me\s+)?through\s+", + re.IGNORECASE, + ), + IntentTag.DEFINITION, + ), (re.compile(r"^why\s+", re.IGNORECASE), IntentTag.CAUSE), # "What causes / triggers / enables / prevents / drives X?" — the # query is about what causes X, so the subject of the CAUSE intent diff --git a/tests/test_discourse_planner_behavior.py b/tests/test_discourse_planner_behavior.py index faaa4401..ccebe2f4 100644 --- a/tests/test_discourse_planner_behavior.py +++ b/tests/test_discourse_planner_behavior.py @@ -232,10 +232,16 @@ class TestExampleMode: class TestWalkthroughMode: - def test_walkthrough_falls_back_to_brief_shape(self) -> None: + def test_walkthrough_emits_chain_walk(self) -> None: + # WALKTHROUGH v1 — sequential teaching-chain walk. The + # _full_bundle has a 2-hop chain (truth→knowledge→evidence) + # plus pack anchor, so the walk emits ANCHOR + RELATION + + # CLOSURE. See test_discourse_planner_walkthrough.py for + # the dedicated suite. plan = plan_discourse(_intent(), ResponseMode.WALKTHROUGH, _full_bundle()) kinds = [m.kind for m in plan.moves] - assert kinds == [DiscourseMoveKind.ANCHOR] + assert kinds[0] is DiscourseMoveKind.ANCHOR + assert DiscourseMoveKind.CLOSURE in kinds or DiscourseMoveKind.RELATION in kinds # --------------------------------------------------------------------------- diff --git a/tests/test_discourse_planner_walkthrough.py b/tests/test_discourse_planner_walkthrough.py new file mode 100644 index 00000000..8a3cb2c8 --- /dev/null +++ b/tests/test_discourse_planner_walkthrough.py @@ -0,0 +1,239 @@ +"""Tests for ``WALKTHROUGH`` v1 — sequential teaching-chain walk. + +Pins: + +* ≤ 4 moves total (1 anchor + ≤3 hops) — the hop cap is structural. +* Each hop follows ``(subject, *, object) → (object, *, *)`` along + the teaching-chain graph; the final hop is a ``CLOSURE`` move. +* When no teaching chain is rooted on the anchor, the planner falls + back to the expository (ANCHOR + SUPPORT) shape rather than + fabricating walk steps. The fallback plan retains + ``mode=WALKTHROUGH`` so callers can tell the planner attempted a + walkthrough but degraded honestly. +* Cycle-safe: a teaching cycle ``A→B→A`` walks A→B→A only if the + facts are distinct; identical facts are never re-emitted. +""" + +from __future__ import annotations + +from generate.discourse_planner import ( + DiscourseMoveKind, + FactSource, + GroundedFact, + GroundingBundle, + plan_discourse, +) +from generate.intent import DialogueIntent, IntentTag, ResponseMode + + +def _intent(subject: str = "truth") -> DialogueIntent: + return DialogueIntent(tag=IntentTag.DEFINITION, subject=subject) + + +def _chain_bundle() -> GroundingBundle: + """4-link teaching chain: truth → knowledge → evidence → recall. + + Plus a pack anchor so ``_select_anchor`` has a definitional fact. + """ + return GroundingBundle( + facts=( + GroundedFact( + subject="truth", predicate="is_defined_as", + obj="reality-correspondence", source=FactSource.PACK, + source_id="en_core_cognition_v1:truth#gloss", + ), + GroundedFact( + subject="truth", predicate="reveals", obj="knowledge", + source=FactSource.TEACHING, + source_id="cognition_chains_v1#cause_truth_reveals_knowledge", + ), + GroundedFact( + subject="knowledge", predicate="requires", obj="evidence", + source=FactSource.TEACHING, + source_id="cognition_chains_v1#cause_knowledge_requires_evidence", + ), + GroundedFact( + subject="evidence", predicate="supports", obj="recall", + source=FactSource.TEACHING, + source_id="cognition_chains_v1#cause_evidence_supports_recall", + ), + ) + ) + + +def _pack_only_bundle() -> GroundingBundle: + return GroundingBundle( + facts=( + GroundedFact( + subject="truth", predicate="is_defined_as", + obj="reality-correspondence", source=FactSource.PACK, + source_id="en_core_cognition_v1:truth#gloss", + ), + GroundedFact( + subject="truth", predicate="belongs_to", + obj="epistemic_domain", source=FactSource.PACK, + source_id="en_core_cognition_v1:truth#domain:0", + ), + ) + ) + + +# --------------------------------------------------------------------------- +# Walk shape +# --------------------------------------------------------------------------- + + +class TestWalkthroughShape: + def test_full_chain_emits_anchor_relation_relation_closure(self) -> None: + plan = plan_discourse(_intent(), ResponseMode.WALKTHROUGH, _chain_bundle()) + kinds = [m.kind for m in plan.moves] + # 1 anchor + 3 hops; last hop is CLOSURE. + assert kinds == [ + DiscourseMoveKind.ANCHOR, + DiscourseMoveKind.RELATION, + DiscourseMoveKind.RELATION, + DiscourseMoveKind.CLOSURE, + ] + + def test_walk_follows_subject_to_object_to_subject(self) -> None: + plan = plan_discourse(_intent(), ResponseMode.WALKTHROUGH, _chain_bundle()) + # Walk invariant applies *across hops* — consecutive teaching + # facts on the chain. The anchor is a pack ``is_defined_as`` + # whose obj is a gloss string, not a graph node, so it's + # excluded. First hop starts on the anchor's *subject*. + teaching_moves = [m for m in plan.moves if m.fact is not None and m.fact.source is FactSource.TEACHING] + for prev, curr in zip(teaching_moves, teaching_moves[1:]): + assert curr.fact is not None and prev.fact is not None + assert curr.fact.subject == prev.fact.obj + # First teaching hop must start on the anchor's subject. + anchor = plan.anchor() + assert anchor is not None and anchor.fact is not None + assert teaching_moves[0].fact is not None + assert teaching_moves[0].fact.subject == anchor.fact.subject + + def test_hop_cap_at_four_moves(self) -> None: + # Build a chain longer than the cap. + long_facts = [ + GroundedFact( + subject="truth", predicate="is_defined_as", + obj="reality-correspondence", source=FactSource.PACK, + source_id="en_core_cognition_v1:truth#gloss", + ), + ] + # 6-link teaching chain. + chain_subjects = ["truth", "a", "b", "c", "d", "e"] + chain_objects = ["a", "b", "c", "d", "e", "f"] + for s, o in zip(chain_subjects, chain_objects): + long_facts.append( + GroundedFact( + subject=s, predicate="leads_to", obj=o, + source=FactSource.TEACHING, + source_id=f"chain#{s}_to_{o}", + ) + ) + bundle = GroundingBundle(facts=tuple(long_facts)) + plan = plan_discourse(_intent(), ResponseMode.WALKTHROUGH, bundle) + assert len(plan.moves) <= 4 + + def test_topics_walk_through_chain(self) -> None: + plan = plan_discourse(_intent(), ResponseMode.WALKTHROUGH, _chain_bundle()) + topics = plan.topics() + # Anchor topic + 3 hop topics. + assert topics == ("truth", "knowledge", "evidence") + + +# --------------------------------------------------------------------------- +# Fallback when no chain is rooted on the anchor +# --------------------------------------------------------------------------- + + +class TestWalkthroughFallback: + def test_no_chain_falls_back_to_expository(self) -> None: + plan = plan_discourse( + _intent(), ResponseMode.WALKTHROUGH, _pack_only_bundle() + ) + kinds = [m.kind for m in plan.moves] + # ANCHOR + SUPPORT, no fabricated walk steps. + assert kinds == [ + DiscourseMoveKind.ANCHOR, + DiscourseMoveKind.SUPPORT, + ] + + def test_fallback_plan_retains_walkthrough_mode(self) -> None: + plan = plan_discourse( + _intent(), ResponseMode.WALKTHROUGH, _pack_only_bundle() + ) + # Even though the planner degraded, the mode tag remains + # WALKTHROUGH so callers can detect "attempted walkthrough, + # degraded honestly". + assert plan.mode is ResponseMode.WALKTHROUGH + + def test_pack_only_no_support_returns_anchor_only(self) -> None: + # Anchor fact only, no support, no teaching chain ⇒ ANCHOR-only + # plan; mode still WALKTHROUGH. + bundle = GroundingBundle( + facts=( + GroundedFact( + subject="truth", predicate="is_defined_as", + obj="reality-correspondence", source=FactSource.PACK, + source_id="en_core_cognition_v1:truth#gloss", + ), + ) + ) + plan = plan_discourse(_intent(), ResponseMode.WALKTHROUGH, bundle) + kinds = [m.kind for m in plan.moves] + assert kinds == [DiscourseMoveKind.ANCHOR] + assert plan.mode is ResponseMode.WALKTHROUGH + + +# --------------------------------------------------------------------------- +# Cycle safety +# --------------------------------------------------------------------------- + + +class TestWalkthroughCycleSafety: + def test_cyclic_chain_does_not_re_emit_same_fact(self) -> None: + # truth → A → truth (cycle). Distinct facts, but if a third + # hop tried to re-walk truth→A, it would re-emit the first + # fact. The planner must not. + bundle = GroundingBundle( + facts=( + GroundedFact( + subject="truth", predicate="is_defined_as", + obj="reality-correspondence", source=FactSource.PACK, + source_id="en_core_cognition_v1:truth#gloss", + ), + GroundedFact( + subject="truth", predicate="produces", obj="echo", + source=FactSource.TEACHING, + source_id="chain#truth_produces_echo", + ), + GroundedFact( + subject="echo", predicate="returns_to", obj="truth", + source=FactSource.TEACHING, + source_id="chain#echo_returns_to_truth", + ), + ) + ) + plan = plan_discourse(_intent(), ResponseMode.WALKTHROUGH, bundle) + fact_keys = [m.fact.sort_key() for m in plan.moves if m.fact is not None] + assert len(fact_keys) == len(set(fact_keys)) + + +# --------------------------------------------------------------------------- +# Determinism +# --------------------------------------------------------------------------- + + +class TestWalkthroughDeterminism: + def test_walk_is_byte_stable(self) -> None: + encoded = [ + plan_discourse(_intent(), ResponseMode.WALKTHROUGH, _chain_bundle()).to_json() + for _ in range(8) + ] + assert len(set(encoded)) == 1 + + def test_walk_equality(self) -> None: + a = plan_discourse(_intent(), ResponseMode.WALKTHROUGH, _chain_bundle()) + b = plan_discourse(_intent(), ResponseMode.WALKTHROUGH, _chain_bundle()) + assert a == b