diff --git a/generate/discourse_planner.py b/generate/discourse_planner.py index ad3ac424..eaf63b0e 100644 --- a/generate/discourse_planner.py +++ b/generate/discourse_planner.py @@ -252,28 +252,270 @@ class DiscoursePlan: return json.dumps(self.as_dict(), sort_keys=True, separators=(",", ":")) +def _move_budget(mode: ResponseMode) -> tuple[int, int]: + """Return ``(min_moves, max_moves)`` for *mode*. + + BRIEF → exactly 1 (ANCHOR only) so flag-on rendering of a + single-sentence pack-grounded surface stays at parity + with the existing string composer. + EXPLAIN → up to 3 (ANCHOR + SUPPORT + RELATION). + PARAGRAPH → up to 5 (ANCHOR + SUPPORT + RELATION + TRANSITION + + CLOSURE). + EXAMPLE → up to 3 (ANCHOR + RELATION + CLOSURE) — instance-shape + surfacing through the reverse-chain view. + WALKTHROUGH→ deferred (needs operator-chain semantics), capped at 1. + """ + + return _MODE_BUDGETS.get(mode, (1, 1)) + + +_MODE_BUDGETS: dict[ResponseMode, tuple[int, int]] = { + ResponseMode.BRIEF: (1, 1), + ResponseMode.EXPLAIN: (1, 3), + ResponseMode.PARAGRAPH: (1, 5), + ResponseMode.EXAMPLE: (1, 3), + ResponseMode.WALKTHROUGH: (1, 1), +} + + +def _select_anchor( + intent: DialogueIntent, + bundle: GroundingBundle, +) -> GroundedFact | None: + """Pick the anchor fact: a pack ``is_defined_as`` for the subject if + available, otherwise the first canonical pack fact, otherwise the + first canonical fact of any source. + """ + + if bundle.is_empty(): + return None + subject = intent.subject.strip().lower() + pack_facts = bundle.facts_by_source(FactSource.PACK) + # Prefer is_defined_as on the subject (carries the gloss). + for fact in pack_facts: + if fact.subject == subject and fact.predicate == "is_defined_as": + return fact + # Fall back to the first canonical pack fact on the subject. + for fact in pack_facts: + if fact.subject == subject: + return fact + # Fall back to the first canonical fact of any source. + for fact in bundle.sorted_facts(): + return fact + return None + + +def _select_support( + anchor: GroundedFact, + bundle: GroundingBundle, +) -> GroundedFact | None: + """Pick a SUPPORT fact distinct from the anchor: a pack ``belongs_to`` + on the anchor's subject if available. + """ + + for fact in bundle.facts_by_source(FactSource.PACK): + if fact == anchor: + continue + if fact.subject != anchor.subject: + continue + if fact.predicate == "belongs_to": + return fact + # Any other pack fact on the same subject. + for fact in bundle.facts_by_source(FactSource.PACK): + if fact == anchor or fact.subject != anchor.subject: + continue + return fact + return None + + +def _select_relation( + anchor: GroundedFact, + bundle: GroundingBundle, + *, + exclude: frozenset[tuple[int, str, str, str, str]] = frozenset(), +) -> GroundedFact | None: + """Pick a RELATION fact: a teaching/cross-pack chain rooted on the + anchor's subject. + """ + + for fact in bundle.facts_by_source(FactSource.TEACHING): + if fact.sort_key() in exclude: + continue + if fact.subject == anchor.subject: + return fact + return None + + +def _select_transition( + relation: GroundedFact, + bundle: GroundingBundle, + *, + exclude: frozenset[tuple[int, str, str, str, str]] = frozenset(), +) -> GroundedFact | None: + """Pick a TRANSITION fact: a teaching/cross-pack chain rooted on the + RELATION's object (the topic shifts to the chain's tail). + """ + + target = relation.obj.strip().lower() + if not target: + return None + for fact in bundle.facts_by_source(FactSource.TEACHING): + if fact.sort_key() in exclude: + continue + if fact.subject == target: + return fact + # No same-source continuation — try any pack fact on the new topic + # (lets the closure step still describe the transitioned topic). + for fact in bundle.facts_by_source(FactSource.PACK): + if fact.sort_key() in exclude: + continue + if fact.subject == target: + return fact + return None + + def plan_discourse( intent: DialogueIntent, mode: ResponseMode, bundle: GroundingBundle, ) -> DiscoursePlan: - """Pure planner function — contract-only signature in this landing. + """Deterministic discourse planner. - Same ``(intent, mode, bundle)`` must produce the same plan on every - invocation: no I/O, no clock reads, no module-level mutable state. + Selects ordered moves from *bundle* according to *mode*'s budget + and the canonical anchor/support/relation/transition/closure + vocabulary. Pure: same ``(intent, mode, bundle)`` always produces + the same plan; no I/O, no clock reads, no module-level state. - The implementation is intentionally deferred: a follow-up ADR will - fill in the move-selection rules (anchor → support → relation → - transition → closure) per ``ResponseMode``. Landing the signature - first locks the contract callers can target without committing to - the heuristics that will populate it. + Empty bundles produce an empty plan rather than raising — callers + fall through to the existing single-sentence composer path so the + runtime is always safe to call with the flag on. + + Mode rules: + + * ``BRIEF`` — ANCHOR only. Equivalent to today's single- + sentence pack-grounded surface. + * ``EXPLAIN`` — ANCHOR + SUPPORT + RELATION (up to 3 moves). + * ``PARAGRAPH`` — ANCHOR + SUPPORT + RELATION + TRANSITION + + CLOSURE (up to 5 moves). + * ``EXAMPLE`` — ANCHOR + RELATION + CLOSURE (up to 3 moves). + The relation is selected from the reverse-chain + view via the bundle (callers supply + cross-pack `include_object_view=True`). + * ``WALKTHROUGH`` — deferred to a follow-up ADR; falls back to + BRIEF shape so the planner is total. """ - _ = (intent, mode, bundle) - raise NotImplementedError( - "plan_discourse is contract-only in this landing; " - "move-selection rules will land in a follow-up ADR." + if bundle.is_empty(): + return DiscoursePlan(intent=intent, mode=mode, moves=()) + + anchor_fact = _select_anchor(intent, bundle) + if anchor_fact is None: + return DiscoursePlan(intent=intent, mode=mode, moves=()) + + moves: list[DiscourseMove] = [ + DiscourseMove( + kind=DiscourseMoveKind.ANCHOR, + topic=anchor_fact.subject, + given=(), + new=(anchor_fact.subject,), + relation_to_previous=None, + fact=anchor_fact, + ) + ] + 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: + return DiscoursePlan(intent=intent, mode=mode, moves=tuple(moves)) + + given_lemmas: list[str] = [anchor_fact.subject] + last_topic = anchor_fact.subject + + # SUPPORT (EXPLAIN, PARAGRAPH — not EXAMPLE which goes anchor→relation). + if mode in (ResponseMode.EXPLAIN, ResponseMode.PARAGRAPH): + support_fact = _select_support(anchor_fact, bundle) + if support_fact is not None: + 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()) + given_lemmas.append(support_fact.obj) + last_topic = support_fact.subject + if len(moves) >= max_moves: + return DiscoursePlan( + intent=intent, mode=mode, moves=tuple(moves) + ) + + # RELATION. + relation_fact = _select_relation( + anchor_fact, bundle, exclude=frozenset(used) ) + if relation_fact is not None: + moves.append( + DiscourseMove( + kind=DiscourseMoveKind.RELATION, + topic=relation_fact.subject, + given=tuple(given_lemmas), + new=(relation_fact.obj,), + relation_to_previous=Relation.CAUSE, + fact=relation_fact, + ) + ) + used.add(relation_fact.sort_key()) + given_lemmas.append(relation_fact.obj) + last_topic = relation_fact.subject + if len(moves) >= max_moves: + return DiscoursePlan( + intent=intent, mode=mode, moves=tuple(moves) + ) + + # TRANSITION (PARAGRAPH only). + transition_fact: GroundedFact | None = None + if mode is ResponseMode.PARAGRAPH and relation_fact is not None: + transition_fact = _select_transition( + relation_fact, bundle, exclude=frozenset(used) + ) + if transition_fact is not None: + moves.append( + DiscourseMove( + kind=DiscourseMoveKind.TRANSITION, + topic=transition_fact.subject, + given=tuple(given_lemmas), + new=(transition_fact.obj,), + relation_to_previous=Relation.SEQUENCE, + fact=transition_fact, + ) + ) + used.add(transition_fact.sort_key()) + given_lemmas.append(transition_fact.obj) + last_topic = transition_fact.subject + if len(moves) >= max_moves: + return DiscoursePlan( + intent=intent, mode=mode, moves=tuple(moves) + ) + + # CLOSURE (PARAGRAPH, EXAMPLE) — summarize the latest topic. No + # new fact (fact=None); closure carries the prior given lemmas + # forward without introducing new content. + if mode in (ResponseMode.PARAGRAPH, ResponseMode.EXAMPLE): + moves.append( + DiscourseMove( + kind=DiscourseMoveKind.CLOSURE, + topic=last_topic, + given=tuple(given_lemmas), + new=(), + relation_to_previous=Relation.ELABORATION, + fact=None, + ) + ) + + return DiscoursePlan(intent=intent, mode=mode, moves=tuple(moves)) __all__ = [ diff --git a/tests/test_discourse_planner_behavior.py b/tests/test_discourse_planner_behavior.py new file mode 100644 index 00000000..faaa4401 --- /dev/null +++ b/tests/test_discourse_planner_behavior.py @@ -0,0 +1,299 @@ +"""Behavior tests for ``plan_discourse`` (step 4). + +These tests pin move-selection rules per ``ResponseMode``: + +* BRIEF → exactly one ANCHOR. +* EXPLAIN → ANCHOR + (SUPPORT) + (RELATION). +* PARAGRAPH → ANCHOR + (SUPPORT) + (RELATION) + (TRANSITION) + (CLOSURE). +* EXAMPLE → ANCHOR + (RELATION) + (CLOSURE). +* WALKTHROUGH → falls back to BRIEF (deferred). + +Determinism is verified end-to-end: ``(intent, mode, bundle)`` ⇒ same +plan ⇒ same canonical JSON. This is the precondition for the future +ADR that folds ``DiscoursePlan`` into ``compute_trace_hash``. + +Bundles are constructed by hand so the planner's selection rules are +tested in isolation from the live pack/teaching corpora. Integration +with ``grounding_bundle_for`` is verified in step 5 once the runtime +flag is wired up. +""" + +from __future__ import annotations + +import pytest + +from generate.discourse_planner import ( + DialogueIntent, + DiscourseMoveKind, + FactSource, + GroundedFact, + GroundingBundle, + IntentTag, + ResponseMode, + plan_discourse, +) + + +def _intent(tag: IntentTag = IntentTag.DEFINITION, subject: str = "truth") -> DialogueIntent: + return DialogueIntent(tag=tag, subject=subject) + + +def _full_bundle() -> GroundingBundle: + """Bundle with anchor + support + relation + transition pieces. + + Topic: ``truth`` + ANCHOR : truth is_defined_as ... + SUPPORT : truth belongs_to epistemic_domain + RELATION : truth reveals knowledge (teaching) + TRANSITION : knowledge requires evidence (teaching, new topic) + """ + + return GroundingBundle( + facts=( + GroundedFact( + subject="truth", + predicate="is_defined_as", + obj="that which corresponds to reality", + 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", + ), + 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", + ), + ) + ) + + +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", + ), + ) + ) + + +# --------------------------------------------------------------------------- +# BRIEF +# --------------------------------------------------------------------------- + + +class TestBriefMode: + def test_brief_emits_anchor_only(self) -> None: + plan = plan_discourse(_intent(), ResponseMode.BRIEF, _full_bundle()) + kinds = [m.kind for m in plan.moves] + assert kinds == [DiscourseMoveKind.ANCHOR] + + def test_brief_anchor_prefers_is_defined_as(self) -> None: + plan = plan_discourse(_intent(), ResponseMode.BRIEF, _full_bundle()) + anchor = plan.anchor() + assert anchor is not None + assert anchor.fact is not None + assert anchor.fact.predicate == "is_defined_as" + + +# --------------------------------------------------------------------------- +# EXPLAIN +# --------------------------------------------------------------------------- + + +class TestExplainMode: + def test_explain_emits_anchor_support_relation(self) -> None: + plan = plan_discourse(_intent(), ResponseMode.EXPLAIN, _full_bundle()) + kinds = [m.kind for m in plan.moves] + assert kinds == [ + DiscourseMoveKind.ANCHOR, + DiscourseMoveKind.SUPPORT, + DiscourseMoveKind.RELATION, + ] + + def test_explain_support_is_pack_belongs_to(self) -> None: + plan = plan_discourse(_intent(), ResponseMode.EXPLAIN, _full_bundle()) + support = next( + m for m in plan.moves if m.kind is DiscourseMoveKind.SUPPORT + ) + assert support.fact is not None + assert support.fact.predicate == "belongs_to" + assert support.fact.source is FactSource.PACK + + def test_explain_relation_is_teaching_chain(self) -> None: + plan = plan_discourse(_intent(), ResponseMode.EXPLAIN, _full_bundle()) + relation = next( + m for m in plan.moves if m.kind is DiscourseMoveKind.RELATION + ) + assert relation.fact is not None + assert relation.fact.source is FactSource.TEACHING + + def test_explain_collapses_when_only_pack_facts(self) -> None: + plan = plan_discourse( + _intent(), ResponseMode.EXPLAIN, _pack_only_bundle() + ) + kinds = [m.kind for m in plan.moves] + assert kinds == [DiscourseMoveKind.ANCHOR, DiscourseMoveKind.SUPPORT] + + +# --------------------------------------------------------------------------- +# PARAGRAPH +# --------------------------------------------------------------------------- + + +class TestParagraphMode: + def test_paragraph_emits_full_five_moves(self) -> None: + plan = plan_discourse(_intent(), ResponseMode.PARAGRAPH, _full_bundle()) + kinds = [m.kind for m in plan.moves] + assert kinds == [ + DiscourseMoveKind.ANCHOR, + DiscourseMoveKind.SUPPORT, + DiscourseMoveKind.RELATION, + DiscourseMoveKind.TRANSITION, + DiscourseMoveKind.CLOSURE, + ] + + def test_paragraph_transition_changes_topic(self) -> None: + plan = plan_discourse(_intent(), ResponseMode.PARAGRAPH, _full_bundle()) + transition = next( + m for m in plan.moves if m.kind is DiscourseMoveKind.TRANSITION + ) + # The transition's topic must differ from the anchor's topic. + assert transition.topic != "truth" + assert transition.topic == "knowledge" + + def test_paragraph_closure_has_no_new_content(self) -> None: + plan = plan_discourse(_intent(), ResponseMode.PARAGRAPH, _full_bundle()) + closure = next( + m for m in plan.moves if m.kind is DiscourseMoveKind.CLOSURE + ) + assert closure.fact is None + assert closure.new == () + # Closure carries the chain of given lemmas forward. + assert len(closure.given) > 0 + + def test_paragraph_topics_cover_full_chain(self) -> None: + plan = plan_discourse(_intent(), ResponseMode.PARAGRAPH, _full_bundle()) + assert plan.topics() == ("truth", "knowledge") + + def test_paragraph_facts_are_unique(self) -> None: + plan = plan_discourse(_intent(), ResponseMode.PARAGRAPH, _full_bundle()) + used = [m.fact for m in plan.moves if m.fact is not None] + keys = [f.sort_key() for f in used] + assert len(keys) == len(set(keys)) + + +# --------------------------------------------------------------------------- +# EXAMPLE +# --------------------------------------------------------------------------- + + +class TestExampleMode: + def test_example_emits_anchor_relation_closure(self) -> None: + plan = plan_discourse(_intent(), ResponseMode.EXAMPLE, _full_bundle()) + kinds = [m.kind for m in plan.moves] + assert kinds == [ + DiscourseMoveKind.ANCHOR, + DiscourseMoveKind.RELATION, + DiscourseMoveKind.CLOSURE, + ] + + def test_example_skips_support(self) -> None: + plan = plan_discourse(_intent(), ResponseMode.EXAMPLE, _full_bundle()) + kinds = {m.kind for m in plan.moves} + assert DiscourseMoveKind.SUPPORT not in kinds + + +# --------------------------------------------------------------------------- +# WALKTHROUGH (deferred) +# --------------------------------------------------------------------------- + + +class TestWalkthroughMode: + def test_walkthrough_falls_back_to_brief_shape(self) -> None: + plan = plan_discourse(_intent(), ResponseMode.WALKTHROUGH, _full_bundle()) + kinds = [m.kind for m in plan.moves] + assert kinds == [DiscourseMoveKind.ANCHOR] + + +# --------------------------------------------------------------------------- +# Empty / degenerate inputs +# --------------------------------------------------------------------------- + + +class TestDegenerateInputs: + @pytest.mark.parametrize("mode", list(ResponseMode)) + def test_empty_bundle_returns_empty_plan(self, mode: ResponseMode) -> None: + plan = plan_discourse(_intent(), mode, GroundingBundle()) + assert plan.is_empty() + assert plan.intent.subject == "truth" + assert plan.mode is mode + + def test_anchor_falls_back_to_first_canonical_fact_when_no_pack_subject_match( + self, + ) -> None: + # Bundle only has teaching facts whose subject differs from the + # intent subject — anchor falls back to the first canonical fact. + bundle = GroundingBundle( + facts=( + GroundedFact( + subject="memory", predicate="requires", obj="recall", + source=FactSource.TEACHING, + source_id="cognition_chains_v1#cause_memory_requires_recall", + ), + ) + ) + plan = plan_discourse( + _intent(subject="nonexistent"), + ResponseMode.BRIEF, + bundle, + ) + assert not plan.is_empty() + anchor = plan.anchor() + assert anchor is not None + assert anchor.fact is not None + assert anchor.fact.subject == "memory" + + +# --------------------------------------------------------------------------- +# Determinism +# --------------------------------------------------------------------------- + + +class TestPlannerDeterminism: + @pytest.mark.parametrize("mode", list(ResponseMode)) + def test_plan_is_byte_stable_across_calls(self, mode: ResponseMode) -> None: + bundle = _full_bundle() + intent = _intent() + encoded = [ + plan_discourse(intent, mode, bundle).to_json() for _ in range(8) + ] + assert len(set(encoded)) == 1 + + def test_plan_is_pure_function(self) -> None: + # Same inputs ⇒ equal plans (including positional move order). + a = plan_discourse(_intent(), ResponseMode.PARAGRAPH, _full_bundle()) + b = plan_discourse(_intent(), ResponseMode.PARAGRAPH, _full_bundle()) + assert a == b diff --git a/tests/test_discourse_planner_contract.py b/tests/test_discourse_planner_contract.py index 8d461ec9..9dab13c3 100644 --- a/tests/test_discourse_planner_contract.py +++ b/tests/test_discourse_planner_contract.py @@ -303,13 +303,17 @@ class TestPlannerSignature: assert hints["bundle"] is GroundingBundle assert hints["return"] is DiscoursePlan - def test_plan_discourse_is_contract_only(self) -> None: - with pytest.raises(NotImplementedError): - plan_discourse( - _make_intent(), - ResponseMode.PARAGRAPH, - GroundingBundle(facts=_make_facts()), - ) + def test_plan_discourse_handles_empty_bundle(self) -> None: + # Empty bundle ⇒ empty plan (planner is total; callers fall + # through to the existing single-sentence composer path). + plan = plan_discourse( + _make_intent(), + ResponseMode.PARAGRAPH, + GroundingBundle(), + ) + assert plan.is_empty() + assert plan.intent == _make_intent() + assert plan.mode is ResponseMode.PARAGRAPH def test_no_runtime_imports(self) -> None: import generate.discourse_planner as dp