From d62a09c84981d6604e3785ba749e6b5918aa011e Mon Sep 17 00:00:00 2001 From: Shay Date: Tue, 19 May 2026 11:06:13 -0700 Subject: [PATCH] feat(discourse): DiscoursePlan contract + determinism gate Contract-only landing for the typed multi-move discourse layer that will sit between grounding and graph construction: DialogueIntent + ResponseMode + GroundingBundle -> DiscoursePlan -> PropositionGraph -> ArticulationTarget -> RealizedPlan Adds frozen dataclasses (ResponseMode, FactSource, GroundedFact, GroundingBundle, DiscourseMoveKind, DiscourseMove, DiscoursePlan), canonical sort + as_dict + to_json serialization (sorted keys, no-whitespace separators), and the pure plan_discourse signature (raises NotImplementedError; move-selection rules deferred). 23 contract tests pin the determinism invariants required before DiscoursePlan can be folded into compute_trace_hash in a follow-up ADR: frozen-dataclass equality, canonical pack DiscoursePlan (this module's output) + -> PropositionGraph (graph_planner, downstream) + -> ArticulationTarget (existing) + -> RealizedPlan (existing) + -> surface / trace_hash (existing) + +Doctrine invariants this layer must satisfy: + +* Every fact carries a source tag (``pack / teaching / vault / operator``) + — no decorative prose, no ungrounded transitions. +* Canonical serialization (alphabetical keys, separators stripped) so + two equal plans hash to the same bytes. This is the precondition + for folding ``DiscoursePlan`` into ``compute_trace_hash`` in a later + ADR — and is asserted by the companion contract tests *before* any + trace-hash change lands. +* Frozen dataclasses with ``slots=True``; ``tuple[...]`` containers + rather than ``list[...]``. Equality is by value. +* The planner function is **pure**: no I/O, no module-level mutable + state, no clock reads. Same ``(intent, mode, bundle)`` → same plan. + +Reserved for follow-up ADRs (intentionally absent here): + +* Classification of ``ResponseMode`` from raw input. +* Structured grounding accessors (``pack_grounded_facts``, + ``teaching_grounded_chains``, ``cross_pack_grounded_chains``). +* Runtime wiring behind ``RuntimeConfig.discourse_planner=False``. +* Folding ``DiscoursePlan`` serialization into ``compute_trace_hash``. +""" + +from __future__ import annotations + +import json +from dataclasses import dataclass, field +from enum import Enum, unique + +from generate.graph_planner import Relation +from generate.intent import DialogueIntent, IntentTag + + +@unique +class ResponseMode(Enum): + """Presentation-depth axis, orthogonal to ``IntentTag``. + + A request like "Explain truth" vs "Tell me about truth" carries the + same ``IntentTag`` (DEFINITION/NARRATIVE) but a different mode. + Keeping mode separate from intent prevents corrupting the semantic + enum with presentation concerns (same lesson as ADR-0049's + syntactic subject extraction). + """ + + BRIEF = "brief" + EXPLAIN = "explain" + WALKTHROUGH = "walkthrough" + PARAGRAPH = "paragraph" + EXAMPLE = "example" + + +@unique +class FactSource(Enum): + """Provenance of a ``GroundedFact``. + + The enum order encodes canonical precedence used by + :func:`GroundingBundle.sorted_facts`: + + pack < teaching < vault < operator + + This mirrors in-pack precedence doctrine (ADR-0063 cross-pack + resolver: cognition pack consulted first) and the four-tier + inter-session memory architecture (ADR-0055: vault → audit → + reviewed corpus → ratified packs). + """ + + PACK = "pack" + TEACHING = "teaching" + VAULT = "vault" + OPERATOR = "operator" + + +_FACT_SOURCE_PRIORITY: dict[FactSource, int] = { + FactSource.PACK: 0, + FactSource.TEACHING: 1, + FactSource.VAULT: 2, + FactSource.OPERATOR: 3, +} + + +@unique +class DiscourseMoveKind(Enum): + """Five-move vocabulary the planner draws from. + + * ``ANCHOR`` — establish the topic (always position 0). + * ``SUPPORT`` — add a domain/definitional fact about the anchor. + * ``RELATION`` — add a cause/verification/chain fact. + * ``TRANSITION`` — move topic to a related node (introduces new + ``topic`` value distinct from prior move). + * ``CLOSURE`` — summarize endpoint or limitation. + """ + + ANCHOR = "anchor" + SUPPORT = "support" + RELATION = "relation" + TRANSITION = "transition" + CLOSURE = "closure" + + +@dataclass(frozen=True, slots=True) +class GroundedFact: + """Atomic, sourced, canonically-sortable fact triple. + + ``source_id`` is the provenance pointer inside ``source``: + pack lemma id, teaching chain id, vault entry hash, operator name. + It is preserved in the serialization so replay can re-locate the + fact deterministically. + """ + + subject: str + predicate: str + obj: str + source: FactSource + source_id: str + + def as_dict(self) -> dict[str, str]: + return { + "subject": self.subject, + "predicate": self.predicate, + "object": self.obj, + "source": self.source.value, + "source_id": self.source_id, + } + + def sort_key(self) -> tuple[int, str, str, str, str]: + return ( + _FACT_SOURCE_PRIORITY[self.source], + self.subject, + self.predicate, + self.obj, + self.source_id, + ) + + +@dataclass(frozen=True, slots=True) +class GroundingBundle: + """Collection of grounded facts available to the planner. + + The bundle is *unordered* at construction time; callers obtain a + canonical view via :meth:`sorted_facts`. This decouples the input + of structured grounding accessors (which may iterate corpora in any + order) from the planner's deterministic output. + """ + + facts: tuple[GroundedFact, ...] = () + + def sorted_facts(self) -> tuple[GroundedFact, ...]: + return tuple(sorted(self.facts, key=GroundedFact.sort_key)) + + def facts_by_source(self, source: FactSource) -> tuple[GroundedFact, ...]: + return tuple(f for f in self.sorted_facts() if f.source is source) + + def is_empty(self) -> bool: + return len(self.facts) == 0 + + def as_dict(self) -> dict[str, object]: + return {"facts": tuple(f.as_dict() for f in self.sorted_facts())} + + +@dataclass(frozen=True, slots=True) +class DiscourseMove: + """One step in a ``DiscoursePlan``. + + ``topic`` — the subject the move is *about* right now. + ``given`` — tuple of lemmas already established by + prior moves (information shared with the + reader). Empty for ``ANCHOR``. + ``new`` — lemmas introduced by this move. + ``relation_to_previous`` — ``None`` for ``ANCHOR``; otherwise the + rhetorical relation linking back to the + immediately-prior move. + ``fact`` — the ``GroundedFact`` this move surfaces; + ``None`` for ``CLOSURE`` moves that only + summarize prior facts. + """ + + kind: DiscourseMoveKind + topic: str + given: tuple[str, ...] = () + new: tuple[str, ...] = () + relation_to_previous: Relation | None = None + fact: GroundedFact | None = None + + def as_dict(self) -> dict[str, object]: + return { + "kind": self.kind.value, + "topic": self.topic, + "given": tuple(self.given), + "new": tuple(self.new), + "relation_to_previous": ( + self.relation_to_previous.value + if self.relation_to_previous is not None + else None + ), + "fact": self.fact.as_dict() if self.fact is not None else None, + } + + +@dataclass(frozen=True, slots=True) +class DiscoursePlan: + """Ordered, typed multi-move plan over a grounding bundle. + + Equality and serialization are positional in ``moves``: the planner + is responsible for emitting moves in canonical order, and consumers + must not reorder them. ``as_dict`` is byte-stable across runs; + ``to_json`` produces the exact bytes that a later ADR will hash into + ``compute_trace_hash``. + """ + + intent: DialogueIntent + mode: ResponseMode + moves: tuple[DiscourseMove, ...] = field(default_factory=tuple) + + def is_empty(self) -> bool: + return len(self.moves) == 0 + + def anchor(self) -> DiscourseMove | None: + for m in self.moves: + if m.kind is DiscourseMoveKind.ANCHOR: + return m + return None + + def topics(self) -> tuple[str, ...]: + seen: list[str] = [] + for m in self.moves: + if m.topic not in seen: + seen.append(m.topic) + return tuple(seen) + + def as_dict(self) -> dict[str, object]: + return { + "intent": { + "tag": self.intent.tag.value, + "subject": self.intent.subject, + "secondary_subject": self.intent.secondary_subject, + "relation": self.intent.relation, + "frame": self.intent.frame, + }, + "mode": self.mode.value, + "moves": tuple(m.as_dict() for m in self.moves), + } + + def to_json(self) -> str: + return json.dumps(self.as_dict(), sort_keys=True, separators=(",", ":")) + + +def plan_discourse( + intent: DialogueIntent, + mode: ResponseMode, + bundle: GroundingBundle, +) -> DiscoursePlan: + """Pure planner function — contract-only signature in this landing. + + Same ``(intent, mode, bundle)`` must produce the same plan on every + invocation: no I/O, no clock reads, no module-level mutable 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. + """ + + _ = (intent, mode, bundle) + raise NotImplementedError( + "plan_discourse is contract-only in this landing; " + "move-selection rules will land in a follow-up ADR." + ) + + +__all__ = [ + "DiscourseMove", + "DiscourseMoveKind", + "DiscoursePlan", + "DialogueIntent", + "FactSource", + "GroundedFact", + "GroundingBundle", + "IntentTag", + "Relation", + "ResponseMode", + "plan_discourse", +] diff --git a/tests/test_discourse_planner_contract.py b/tests/test_discourse_planner_contract.py new file mode 100644 index 00000000..8d461ec9 --- /dev/null +++ b/tests/test_discourse_planner_contract.py @@ -0,0 +1,324 @@ +"""Contract tests for ``generate/discourse_planner.py``. + +These tests pin the **serialization determinism** invariant that a +later ADR will rely on when folding ``DiscoursePlan`` into +``compute_trace_hash``. Adding the determinism gate *before* the +hash dependency is the explicit sequencing requirement: replay +regressions must surface as clean test failures, not as flaky +paragraph turns downstream. + +What this file is **not**: + +* Not a runtime/wiring test — nothing here imports ``chat.*`` or + exercises a live ``ChatRuntime``. At this stage the planner has + no runtime path; cognition-eval byte-identity is asserted by the + existing eval lane, not here. +* Not a planner-behavior test — ``plan_discourse`` is contract-only + in this landing and raises ``NotImplementedError``. Move-selection + rules will be tested in the follow-up ADR. +""" + +from __future__ import annotations + +import inspect +import json + +import pytest + +from generate.discourse_planner import ( + DialogueIntent, + DiscourseMove, + DiscourseMoveKind, + DiscoursePlan, + FactSource, + GroundedFact, + GroundingBundle, + IntentTag, + Relation, + ResponseMode, + plan_discourse, +) + + +def _make_intent() -> DialogueIntent: + return DialogueIntent(tag=IntentTag.DEFINITION, subject="truth") + + +def _make_facts() -> tuple[GroundedFact, ...]: + return ( + GroundedFact( + subject="truth", + predicate="reveals", + obj="knowledge", + source=FactSource.TEACHING, + source_id="cognition_chains_v1#cause_truth_reveals_knowledge", + ), + GroundedFact( + subject="truth", + predicate="is_defined_as", + obj="that which corresponds to reality", + source=FactSource.PACK, + source_id="en_core_cognition_v1:truth", + ), + GroundedFact( + subject="truth", + predicate="belongs_to", + obj="epistemic_domain", + source=FactSource.PACK, + source_id="en_core_cognition_v1:truth#domain", + ), + ) + + +def _make_plan() -> DiscoursePlan: + facts = _make_facts() + # pack fact[1] is the anchor (is_defined_as), pack fact[2] supports + # (belongs_to), teaching fact[0] is the relation. + moves = ( + DiscourseMove( + kind=DiscourseMoveKind.ANCHOR, + topic="truth", + given=(), + new=("truth",), + relation_to_previous=None, + fact=facts[1], + ), + DiscourseMove( + kind=DiscourseMoveKind.SUPPORT, + topic="truth", + given=("truth",), + new=("epistemic_domain",), + relation_to_previous=Relation.ELABORATION, + fact=facts[2], + ), + DiscourseMove( + kind=DiscourseMoveKind.RELATION, + topic="truth", + given=("truth", "epistemic_domain"), + new=("knowledge",), + relation_to_previous=Relation.CAUSE, + fact=facts[0], + ), + ) + return DiscoursePlan( + intent=_make_intent(), + mode=ResponseMode.PARAGRAPH, + moves=moves, + ) + + +# --------------------------------------------------------------------------- +# Immutability / frozen-dataclass invariants +# --------------------------------------------------------------------------- + + +class TestFrozenInvariants: + def test_grounded_fact_is_frozen(self) -> None: + fact = _make_facts()[0] + with pytest.raises((AttributeError, TypeError)): + fact.subject = "different" # type: ignore[misc] + + def test_grounding_bundle_is_frozen(self) -> None: + bundle = GroundingBundle(facts=_make_facts()) + with pytest.raises((AttributeError, TypeError)): + bundle.facts = () # type: ignore[misc] + + def test_discourse_move_is_frozen(self) -> None: + plan = _make_plan() + with pytest.raises((AttributeError, TypeError)): + plan.moves[0].topic = "lie" # type: ignore[misc] + + def test_discourse_plan_is_frozen(self) -> None: + plan = _make_plan() + with pytest.raises((AttributeError, TypeError)): + plan.mode = ResponseMode.BRIEF # type: ignore[misc] + + def test_value_equality(self) -> None: + assert _make_plan() == _make_plan() + assert hash(_make_facts()[0]) == hash(_make_facts()[0]) + + +# --------------------------------------------------------------------------- +# Canonical ordering invariants +# --------------------------------------------------------------------------- + + +class TestCanonicalOrdering: + def test_fact_source_priority_is_pack_first(self) -> None: + bundle = GroundingBundle( + facts=( + GroundedFact("a", "p", "b", FactSource.OPERATOR, "op:1"), + GroundedFact("a", "p", "b", FactSource.VAULT, "vault:1"), + GroundedFact("a", "p", "b", FactSource.TEACHING, "teach:1"), + GroundedFact("a", "p", "b", FactSource.PACK, "pack:1"), + ) + ) + sources = tuple(f.source for f in bundle.sorted_facts()) + assert sources == ( + FactSource.PACK, + FactSource.TEACHING, + FactSource.VAULT, + FactSource.OPERATOR, + ) + + def test_sort_is_total_within_same_source(self) -> None: + bundle = GroundingBundle( + facts=( + GroundedFact("zeta", "p", "o", FactSource.PACK, "id:2"), + GroundedFact("alpha", "p", "o", FactSource.PACK, "id:1"), + GroundedFact("alpha", "p", "o", FactSource.PACK, "id:0"), + ) + ) + ordered = bundle.sorted_facts() + assert ordered[0].subject == "alpha" + assert ordered[0].source_id == "id:0" + assert ordered[1].subject == "alpha" + assert ordered[1].source_id == "id:1" + assert ordered[2].subject == "zeta" + + def test_bundle_sort_is_idempotent(self) -> None: + bundle = GroundingBundle(facts=_make_facts()) + once = bundle.sorted_facts() + twice = GroundingBundle(facts=once).sorted_facts() + assert once == twice + + def test_facts_by_source_filters_and_orders(self) -> None: + bundle = GroundingBundle(facts=_make_facts()) + pack_only = bundle.facts_by_source(FactSource.PACK) + assert all(f.source is FactSource.PACK for f in pack_only) + assert pack_only == tuple( + sorted(pack_only, key=GroundedFact.sort_key) + ) + + +# --------------------------------------------------------------------------- +# Serialization determinism (the gate before trace_hash adoption) +# --------------------------------------------------------------------------- + + +class TestSerializationDeterminism: + def test_to_json_is_byte_stable_across_calls(self) -> None: + plan = _make_plan() + encoded = [plan.to_json() for _ in range(8)] + assert len(set(encoded)) == 1 + + def test_to_json_is_byte_stable_across_equal_plans(self) -> None: + a = _make_plan().to_json() + b = _make_plan().to_json() + assert a == b + + def test_to_json_uses_sorted_keys(self) -> None: + encoded = _make_plan().to_json() + decoded = json.loads(encoded) + # Re-encoding with sort_keys must round-trip byte-identical. + reencoded = json.dumps(decoded, sort_keys=True, separators=(",", ":")) + assert encoded == reencoded + + def test_to_json_has_no_whitespace(self) -> None: + encoded = _make_plan().to_json() + assert ", " not in encoded + assert ": " not in encoded + + def test_as_dict_round_trip_through_json(self) -> None: + plan = _make_plan() + decoded = json.loads(plan.to_json()) + assert decoded["intent"]["tag"] == "definition" + assert decoded["intent"]["subject"] == "truth" + assert decoded["mode"] == "paragraph" + assert len(decoded["moves"]) == 3 + assert decoded["moves"][0]["kind"] == "anchor" + assert decoded["moves"][0]["relation_to_previous"] is None + assert decoded["moves"][1]["relation_to_previous"] == "elaboration" + assert decoded["moves"][2]["relation_to_previous"] == "cause" + assert decoded["moves"][0]["fact"]["source"] == "pack" + + def test_grounded_fact_as_dict_has_object_key_not_obj(self) -> None: + fact = _make_facts()[0] + encoded = fact.as_dict() + assert "object" in encoded + assert "obj" not in encoded + + def test_bundle_as_dict_is_sorted(self) -> None: + unordered = ( + GroundedFact("z", "p", "o", FactSource.OPERATOR, "op:0"), + GroundedFact("a", "p", "o", FactSource.PACK, "pack:0"), + ) + bundle = GroundingBundle(facts=unordered) + encoded = bundle.as_dict() + facts_out = encoded["facts"] + assert isinstance(facts_out, tuple) + assert facts_out[0]["source"] == "pack" + assert facts_out[1]["source"] == "operator" + + +# --------------------------------------------------------------------------- +# Plan-level shape helpers +# --------------------------------------------------------------------------- + + +class TestPlanHelpers: + def test_empty_plan_reports_empty(self) -> None: + plan = DiscoursePlan( + intent=_make_intent(), + mode=ResponseMode.BRIEF, + ) + assert plan.is_empty() + assert plan.anchor() is None + assert plan.topics() == () + + def test_anchor_returns_first_anchor_move(self) -> None: + plan = _make_plan() + anchor = plan.anchor() + assert anchor is not None + assert anchor.kind is DiscourseMoveKind.ANCHOR + assert anchor.topic == "truth" + + def test_topics_preserves_first_introduction_order(self) -> None: + plan = _make_plan() + assert plan.topics() == ("truth",) + + def test_empty_bundle_helpers(self) -> None: + bundle = GroundingBundle() + assert bundle.is_empty() + assert bundle.sorted_facts() == () + assert bundle.facts_by_source(FactSource.PACK) == () + + +# --------------------------------------------------------------------------- +# Planner function signature is pure and contract-only +# --------------------------------------------------------------------------- + + +class TestPlannerSignature: + def test_plan_discourse_signature(self) -> None: + # ``from __future__ import annotations`` makes annotations strings; + # resolve them via ``get_type_hints`` for an identity comparison. + from typing import get_type_hints + + sig = inspect.signature(plan_discourse) + assert list(sig.parameters) == ["intent", "mode", "bundle"] + hints = get_type_hints(plan_discourse) + assert hints["intent"] is DialogueIntent + assert hints["mode"] is ResponseMode + 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_no_runtime_imports(self) -> None: + import generate.discourse_planner as dp + + src = inspect.getsource(dp) + assert "from chat" not in src + assert "import chat" not in src + # No clock reads, no env reads, no filesystem. + assert "time.time" not in src + assert "datetime.now" not in src + assert "os.environ" not in src + assert "open(" not in src