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<teaching<vault<operator
ordering, byte-stable to_json across calls and equal plans, JSON
round-trip stability, and signature purity (no chat.* imports, no
clock/env/filesystem reads).
No runtime wiring; smoke suite 67/67; cognition eval byte-identical
(public 100/100/91.7/100, holdout 100/100/83.3/100).
This commit is contained in:
Shay
parent
e06fda5b8b
commit
d62a09c849
2 changed files with 633 additions and 0 deletions
309
generate/discourse_planner.py
Normal file
309
generate/discourse_planner.py
Normal file
|
|
@ -0,0 +1,309 @@
|
|||
"""Discourse planner contract — typed multi-move plan over grounded facts.
|
||||
|
||||
This module is **contract-only** in its initial landing: it defines the
|
||||
frozen dataclasses, enums, canonical serialization, and the pure planner
|
||||
function signature. It has **no runtime wiring**. Nothing in
|
||||
``chat/*`` or any live ``ChatRuntime`` path imports from here yet.
|
||||
|
||||
Architectural rationale (see memory: feedback-design-fix-upstream-not-beside):
|
||||
the existing ``realize_target`` already renders paragraph-scale output
|
||||
when fed a multi-node ``PropositionGraph`` (``evals/discourse_paragraph``
|
||||
passes at ``accuracy=1.0 / replay_determinism=1.0``). The bottleneck is
|
||||
upstream — ``graph_planner.build_target`` receives one-node graphs from
|
||||
runtime grounding. The fix is to lift grounding into a typed
|
||||
``DiscoursePlan`` *before* the graph is built, so the realizer is fed
|
||||
multi-node graphs from real runtime evidence rather than hand-authored
|
||||
fixtures.
|
||||
|
||||
Pipeline target:
|
||||
|
||||
DialogueIntent + ResponseMode + GroundingBundle
|
||||
-> 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",
|
||||
]
|
||||
324
tests/test_discourse_planner_contract.py
Normal file
324
tests/test_discourse_planner_contract.py
Normal file
|
|
@ -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
|
||||
Loading…
Reference in a new issue