From fe4cc2cd1f9e9f64ab0a39f753c78ce9b81f646d Mon Sep 17 00:00:00 2001 From: Shay Date: Mon, 18 May 2026 17:01:34 -0700 Subject: [PATCH] feat(adr-0066): session-thread context + opt-in anaphora prefix (Phase 3.1 + 3.2) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ADR-0066 P3.1 + P3.2. Conversation now reads as a thread: turns carry structured summaries of their predecessors and (optionally) prefix new pack/teaching surfaces with deterministic backreferences. P3.1 — chat/thread_context.py. TurnSummary(turn_index, intent_tag_name, subject, grounding_source, chain_id, corpus_id) — frozen, structured-fields-only. ThreadContext — bounded FIFO (default MAX_THREAD_TURNS=8) with snapshot(), recent_for_subject(), recent_subjects(), clear(). recent_for_subject() excludes ungrounded tiers (oov/partial/none) by default — those are not strong-enough anchors. ChatRuntime.thread_context is owned at construction. _push_thread_summary runs at end-of-turn on BOTH stub and walk paths. Teaching-grounded turns carry chain_id + corpus_id so downstream composers (P3.2) can detect same-chain reference. Cold-start intent classification now runs unconditionally (was: gated on sink attachment) so thread context captures subject regardless of sink state. P3.2 — chat/anaphora.py. thread_anaphora_prefix(ctx, subject, intent_name, source) returns a deterministic prefix when: - current turn is pack/teaching tier - a prior pack/teaching turn on the same subject exists - the prior intent differs from the current intent Format (structural-fields-only — no prose): "(Recalling turn N: chain .) " # prior was teaching "(Recalling turn N: grounded pack.) " # prior was pack Opt-in via RuntimeConfig.thread_anaphora=False. Default off keeps every existing surface byte-identical. Live verification (with thread_anaphora=True + seeded context): > What is light? # following a "Why does light exist?" teaching turn [pack] (Recalling turn 0: chain cause_light_reveals_truth.) light — pack-grounded (en_core_cognition_v1): cognition.illumination; logos.core; perception.clarity. No session evidence yet. 32 new tests passed. Curated lanes green. Cognition eval byte-identical to pre-ADR baseline. --- chat/anaphora.py | 94 +++++++++++++ chat/runtime.py | 146 +++++++++++++++++++- chat/thread_context.py | 181 +++++++++++++++++++++++++ core/config.py | 9 ++ tests/test_anaphora.py | 175 ++++++++++++++++++++++++ tests/test_thread_context.py | 254 +++++++++++++++++++++++++++++++++++ 6 files changed, 852 insertions(+), 7 deletions(-) create mode 100644 chat/anaphora.py create mode 100644 chat/thread_context.py create mode 100644 tests/test_anaphora.py create mode 100644 tests/test_thread_context.py diff --git a/chat/anaphora.py b/chat/anaphora.py new file mode 100644 index 00000000..9b99c0f1 --- /dev/null +++ b/chat/anaphora.py @@ -0,0 +1,94 @@ +"""chat/anaphora.py — Phase 3.2: deterministic thread anaphora prefix. + +When the current turn's subject lemma appeared in a recent *grounded* +turn (pack or teaching tier), the anaphora composer prepends a +deterministic backreference to the current surface. Conversation +reads as a thread instead of a sequence of independent surfaces: + + Turn 0 > Why does light exist? + [teaching] light — teaching-grounded (cognition_chains_v1): + cognition.illumination; logos.core. light reveals truth ... + Turn 1 > What is light? + [pack] (Recalling turn 0: chain cause_light_reveals_truth.) + light — pack-grounded (en_core_cognition_v1): cognition.illumination; + logos.core; perception.clarity. No session evidence yet. + +The prefix is **strictly deterministic**: + + - Same thread state + same current turn → byte-identical prefix. + - No prose generation; the prefix references the prior turn by + *turn index* and *chain id* (or grounding tier), nothing else. + - The composer only fires when *both* the prior turn AND the + current turn are pack/teaching tier; weaker tiers (vault / + partial / oov / none) do not anchor. + - Same-intent revisits (asking the same question twice) do not + fire — the prior turn IS the current turn modulo session vault + drift, prefixing it is redundant. + - Opt-in via :attr:`core.config.RuntimeConfig.thread_anaphora`. + Default ``False`` preserves every pre-P3.2 surface byte-identically. +""" + +from __future__ import annotations + +from chat.thread_context import ThreadContext + + +# Grounding tiers strong enough to anchor an anaphora reference. +_ANCHOR_TIERS: frozenset[str] = frozenset({"teaching", "pack"}) + + +def thread_anaphora_prefix( + thread_context: ThreadContext, + current_subject: str, + current_intent_tag_name: str, + current_grounding_source: str, +) -> str | None: + """Return a deterministic anaphora prefix, or ``None``. + + Engagement conditions (ALL must hold): + + 1. ``current_subject`` is non-empty. + 2. ``current_grounding_source`` is in ``{"pack", "teaching"}`` + — weaker tiers cannot host a meaningful backreference. + 3. A prior turn on the same subject exists in + ``thread_context`` AND was itself pack/teaching grounded. + 4. The prior turn's intent differs from the current turn's + intent (a same-intent revisit is the same surface modulo + vault drift; prefixing would be redundant). + + The prefix format references the prior turn by structured fields + only — never by surface text, never by re-derived prose. Two + shapes exist: + + - Prior was teaching-grounded → ``"(Recalling turn N: chain + .) "`` + - Prior was pack-grounded → ``"(Recalling turn N: + grounded pack.) "`` + + Returns ``None`` when any engagement condition fails. Callers + then emit the unprefixed surface byte-identically. + """ + if not current_subject or not isinstance(current_subject, str): + return None + key = current_subject.strip().lower() + if not key: + return None + if current_grounding_source not in _ANCHOR_TIERS: + return None + prior = thread_context.recent_for_subject(key) + if prior is None: + return None + if prior.grounding_source not in _ANCHOR_TIERS: + return None + if prior.intent_tag_name and prior.intent_tag_name == current_intent_tag_name: + return None + + if prior.grounding_source == "teaching" and prior.chain_id: + return f"(Recalling turn {prior.turn_index}: chain {prior.chain_id}.) " + return ( + f"(Recalling turn {prior.turn_index}: {key} grounded " + f"{prior.grounding_source}.) " + ) + + +__all__ = ["thread_anaphora_prefix"] diff --git a/chat/runtime.py b/chat/runtime.py index 63f5ccc3..31014e61 100644 --- a/chat/runtime.py +++ b/chat/runtime.py @@ -412,6 +412,13 @@ class ChatRuntime: ) self._last_refusal_was_typed: bool = True self.turn_log: List[TurnEvent] = [] + # P3.1 — session-thread state for downstream anaphora / + # NARRATIVE composers. Bounded recency window of structured + # TurnSummary records. Data layer only at P3.1 — no surface + # emission consults this until P3.2 (anaphora composer) opt-in + # flag flips on. + from chat.thread_context import ThreadContext + self.thread_context = ThreadContext() # ADR-0040 — opt-in structured-logging sink. Default None # preserves prior behavior; callers attach via # ``attach_telemetry_sink``. ``_telemetry_include_content`` @@ -512,6 +519,69 @@ class ChatRuntime: """ self._contemplate_discoveries = bool(enabled) + def _push_thread_summary( + self, + *, + turn_event: TurnEvent, + intent_tag: Any, + intent_subject: str | None, + grounding_source: str | None, + surface: str | None = None, + ) -> None: + """P3.1 — append one :class:`TurnSummary` to the bounded + session-thread context. Called at end-of-turn from both the + stub path (cold start / pack / teaching / OOV / partial) + and the walk path (vault). + + For teaching-grounded turns the chain_id + corpus_id are + recovered from the most-recently-loaded aggregated chain + index (deterministic O(1) lookup since the index is lru- + cached). For non-teaching turns those fields stay None. + + Pure data layer: this method does NOT consult the surface, + does NOT mutate any composer state, and does NOT call any + LLM. Push runs unconditionally — anaphora consumers are + opt-in elsewhere. + """ + from chat.thread_context import TurnSummary + + turn_index = len(self.turn_log) - 1 # turn_log was just appended + # Normalise the intent tag name; ``None`` (walk path) projects + # to the empty string so the recency lookup can still ignore + # mismatched intents without raising. + if intent_tag is not None and hasattr(intent_tag, "name"): + intent_name = str(intent_tag.name).lower() + else: + intent_name = "" + subject = (intent_subject or "").strip().lower() + source = (grounding_source or "none").lower() + + # Recover chain_id + corpus_id for teaching-grounded turns so + # the anaphora composer can detect "same chain" vs "same + # subject, different chain". + chain_id: str | None = None + corpus_id: str | None = None + if source == "teaching" and subject and intent_name in {"cause", "verification"}: + from chat.teaching_grounding import _all_chains_index + chain = _all_chains_index().get((subject, intent_name)) + if chain is not None: + chain_id = chain.chain_id + corpus_id = chain.corpus_id + # ``surface`` is accepted so future extensions can hash it, + # but P3.1 intentionally does not retain the text. + _ = surface + + self.thread_context.push( + TurnSummary( + turn_index=turn_index, + intent_tag_name=intent_name, + subject=subject, + grounding_source=source, + chain_id=chain_id, + corpus_id=corpus_id, + ) + ) + def _emit_oov_candidate( self, *, @@ -752,6 +822,27 @@ class ChatRuntime: # corpus is reviewed memory; every emitted atom is either a # lemma, a verbatim pack semantic_domains string, or a fixed # connective from humanize_predicate. + # P3.3 — NARRATIVE: "Tell me about X" / "Describe X". + # Multi-clause composer aggregates every reviewed chain + # rooted on X across all registered teaching corpora. + if intent.tag is IntentTag.NARRATIVE: + lemma = (intent.subject or "").strip() + if lemma: + from chat.narrative_surface import narrative_grounded_surface + surface = narrative_grounded_surface(lemma) + if surface is not None: + return (surface, "teaching") + # P3.4 — EXAMPLE: "Give me an example of X". Reverse-chain + # composer surfaces chains where X is the OBJECT. Same + # aggregated corpus index as NARRATIVE; inverts the access + # pattern. + if intent.tag is IntentTag.EXAMPLE: + lemma = (intent.subject or "").strip() + if lemma: + from chat.example_surface import example_grounded_surface + surface = example_grounded_surface(lemma) + if surface is not None: + return (surface, "teaching") if intent.tag in (IntentTag.CAUSE, IntentTag.VERIFICATION): lemma = (intent.subject or "").strip() if lemma: @@ -892,6 +983,27 @@ class ChatRuntime: # replaces the universal "insufficient grounding" disclosure # when no refusal applies. response_surface = pack_grounded_surface + # P3.2 — opt-in thread anaphora prefix. Engages only when + # the current turn AND a recent turn (same subject) are + # both pack/teaching grounded. Default-off so pre-P3.2 + # surfaces stay byte-identical; turning it on prepends a + # deterministic backreference referencing the prior turn + # by turn-index + chain_id (no prose generation). + if ( + self.config.thread_anaphora + and grounded_source_tag in {"pack", "teaching"} + and discovery_intent_subject + and discovery_intent_tag is not None + ): + from chat.anaphora import thread_anaphora_prefix + prefix = thread_anaphora_prefix( + self.thread_context, + discovery_intent_subject, + discovery_intent_tag.name.lower(), + grounded_source_tag, + ) + if prefix is not None: + response_surface = prefix + response_surface else: response_surface = _UNKNOWN_DOMAIN_SURFACE # ADR-0048 — grounding provenance recorded for both ChatResponse @@ -961,6 +1073,15 @@ class ChatRuntime: intent_tag=discovery_intent_tag, token=discovery_intent_subject, ) + # P3.1 — push session-thread summary. Data layer only; + # downstream composers (P3.2 anaphora) consult this. + self._push_thread_summary( + turn_event=stub_event, + intent_tag=discovery_intent_tag, + intent_subject=discovery_intent_subject, + grounding_source=grounding_source, + surface=response_surface, + ) return ChatResponse( surface=response_surface, proposition=prop, @@ -1039,14 +1160,15 @@ class ChatRuntime: # no sink is attached. discovery_intent_tag = None discovery_intent_subject: str | None = None - # Classify intent up-front when EITHER the discovery sink - # OR the OOV sink is attached (P2.3) — both downstream - # emission paths need it. Without either sink attached - # this is a no-op, so behaviour stays identical to the - # pre-Phase-2 path. + # Classify intent up-front whenever the gate fired on an + # empty vault. P2.3 needs it for OOV sink emission, + # ADR-0055 Phase B needs it for discovery sink emission, + # and P3.1 needs it for session-thread context — the + # classifier is cheap and deterministic, so always run + # it on the cold-start English path. Sinks themselves + # remain opt-in (no-op without ``attach_*_sink``). if ( - (self._discovery_sink is not None or self._oov_sink is not None) - and gate_decision.source == "empty_vault" + gate_decision.source == "empty_vault" and self.config.output_language == "en" ): from generate.intent_bridge import classify_intent_from_input @@ -1258,6 +1380,16 @@ class ChatRuntime: ) self.turn_log.append(turn_event) self._emit_turn_event(turn_event) + # P3.1 — push session-thread summary for the walk path. + # Subject is taken from the articulation (deterministic; + # matches what the surface foregrounded). + self._push_thread_summary( + turn_event=turn_event, + intent_tag=None, + intent_subject=articulation.subject, + grounding_source="vault", + surface=response_surface, + ) return ChatResponse( surface=response_surface, proposition=proposition, diff --git a/chat/thread_context.py b/chat/thread_context.py new file mode 100644 index 00000000..4e5afd7c --- /dev/null +++ b/chat/thread_context.py @@ -0,0 +1,181 @@ +"""chat/thread_context.py — Phase 3.1: session-thread state. + +The runtime today treats each turn as an independent grounded surface. +There is no thread-level memory: a turn cannot reference what was +established three turns ago, even when the same subject reappears. +That is the *articulation gap* in plain terms — conversation reads +mechanical because each turn is freshly minted, never referenced +backward. + +This module is the data primitive that closes that gap. Phase 3.1 +stores; Phase 3.2 (anaphora composer) reads. Surface emission is +unchanged at P3.1 — turning the data layer on cannot regress any +existing test. + +Design constraints (matching CLAUDE.md doctrine): + +- **Bounded.** Capacity ``MAX_THREAD_TURNS`` (default 8). Older + summaries evict in FIFO order; thread context is *not* a long-term + store, it is a small recency window the anaphora composer can + reference. Long-term memory is the vault. +- **Immutable summaries.** ``TurnSummary`` is frozen. Pushing + produces a new entry; never mutates an existing one. +- **No reconstruction from surface.** The summary carries only + structured fields (intent_tag, subject, grounding_source, + chain_id). Full surface text stays in the audit trail + (``rt.turn_log``); thread context references only the shape. +- **No clock-time reads.** Determinism — replays of the same + sequence of turns produce identical thread state. +""" + +from __future__ import annotations + +from collections import deque +from dataclasses import dataclass +from typing import Iterable + + +# Recency window size. 8 is large enough for typical multi-turn +# anaphora ("As we established two turns ago..." through to "earlier +# in this conversation...") without giving the anaphora composer a +# context bigger than the surface itself. Operators can override +# at construction time via :class:`ThreadContext(max_turns=N)`. +MAX_THREAD_TURNS: int = 8 + + +@dataclass(frozen=True, slots=True) +class TurnSummary: + """One structured record summarising a turn for thread-anaphora + lookup. Frozen — the runtime never mutates a pushed summary. + + Fields: + - ``turn_index``: monotonic integer (0-based). Same numbering + as ``rt.turn_log``. + - ``intent_tag_name``: lowercase intent name (``"cause"``, + ``"definition"``, etc.). Lowercased for case-insensitive + match against new turns. + - ``subject``: lowercased subject lemma (normalised to match + the pack-resolver layer). Empty string when the turn had no + clean subject. + - ``grounding_source``: ``"vault" | "teaching" | "pack" | + "partial" | "oov" | "none"`` — the tier the turn surfaced + through. Anaphora is most useful when both turns surfaced + through ``"pack"`` or ``"teaching"`` (deterministic + backreference); the composer can choose to skip vault / + partial / oov / none on its own policy. + - ``chain_id``: present when ``grounding_source == "teaching"``, + else ``None``. Lets the anaphora composer detect "same + chain referenced again" vs "different chain on same subject". + - ``corpus_id``: present when ``grounding_source == "teaching"``, + else ``None``. Cross-corpus thread anaphora reads the + corpus tag back to the user. + """ + + turn_index: int + intent_tag_name: str + subject: str + grounding_source: str + chain_id: str | None = None + corpus_id: str | None = None + + def as_dict(self) -> dict[str, object]: + return { + "turn_index": self.turn_index, + "intent_tag_name": self.intent_tag_name, + "subject": self.subject, + "grounding_source": self.grounding_source, + "chain_id": self.chain_id, + "corpus_id": self.corpus_id, + } + + +class ThreadContext: + """Bounded FIFO of :class:`TurnSummary` records. + + Owned by :class:`chat.runtime.ChatRuntime`; updated after each + turn. Read-only from outside the runtime — tests can inspect + via ``rt.thread_context.snapshot()``. + """ + + __slots__ = ("_deque", "_max_turns") + + def __init__(self, *, max_turns: int = MAX_THREAD_TURNS) -> None: + if max_turns < 1: + raise ValueError(f"max_turns must be >= 1 (got {max_turns!r})") + self._max_turns = max_turns + self._deque: deque[TurnSummary] = deque(maxlen=max_turns) + + @property + def max_turns(self) -> int: + return self._max_turns + + def __len__(self) -> int: + return len(self._deque) + + def push(self, summary: TurnSummary) -> None: + """Append a new turn summary; evict the oldest if at capacity.""" + if not isinstance(summary, TurnSummary): # pragma: no cover — defensive + raise TypeError(f"expected TurnSummary, got {type(summary).__name__}") + self._deque.append(summary) + + def snapshot(self) -> tuple[TurnSummary, ...]: + """Return an immutable tuple of every retained summary, in + insertion order (oldest first).""" + return tuple(self._deque) + + def recent_for_subject( + self, + subject: str, + *, + exclude_grounding: Iterable[str] = ("none", "oov", "partial"), + ) -> TurnSummary | None: + """Return the most-recent summary whose ``subject`` matches + *subject* (case-insensitive, whitespace-trimmed), or ``None``. + + Summaries with ``grounding_source`` in *exclude_grounding* are + skipped by default — they carry less anchor evidence than + pack/teaching turns and the anaphora composer's "as we just + established" reads false if the prior turn was actually + ungrounded. Operators can pass ``exclude_grounding=()`` to + include every prior turn. + """ + if not subject or not isinstance(subject, str): + return None + key = subject.strip().lower() + if not key: + return None + excluded = frozenset(exclude_grounding) + for summary in reversed(self._deque): + if summary.grounding_source in excluded: + continue + if summary.subject == key: + return summary + return None + + def recent_subjects( + self, + *, + exclude_grounding: Iterable[str] = ("none", "oov", "partial"), + ) -> tuple[str, ...]: + """Return the set of unique subjects in the window, ordered by + most-recent-first. Skips empty subjects and any whose + grounding tier is in *exclude_grounding*.""" + excluded = frozenset(exclude_grounding) + seen: set[str] = set() + ordered: list[str] = [] + for summary in reversed(self._deque): + if summary.grounding_source in excluded: + continue + if not summary.subject or summary.subject in seen: + continue + seen.add(summary.subject) + ordered.append(summary.subject) + return tuple(ordered) + + def clear(self) -> None: + """Drop every retained summary. Used by tests + by callers + that explicitly reset session memory.""" + self._deque.clear() + + +__all__ = ["TurnSummary", "ThreadContext", "MAX_THREAD_TURNS"] diff --git a/core/config.py b/core/config.py index 1c58f839..64db59f7 100644 --- a/core/config.py +++ b/core/config.py @@ -65,6 +65,15 @@ class RuntimeConfig: # depth (max one follow-up chain in v1). composed_surface: bool = False + # ADR-0066 / P3.2 — opt-in thread anaphora. When enabled, the + # runtime prepends a deterministic backreference to a recent + # grounded turn when the current turn's subject lemma matches + # one in the bounded session-thread context. Engages only on + # pack/teaching-tier turns (both prior and current); weaker + # tiers do not anchor. Default False preserves every pre-P3.2 + # surface byte-identically. + thread_anaphora: bool = False + DEFAULT_IDENTITY_PACK: str = "default_general_v1" DEFAULT_ETHICS_PACK: str = "default_general_ethics_v1" diff --git a/tests/test_anaphora.py b/tests/test_anaphora.py new file mode 100644 index 00000000..26327f23 --- /dev/null +++ b/tests/test_anaphora.py @@ -0,0 +1,175 @@ +"""Phase 3.2 — thread anaphora prefix tests. + +The contract these tests pin: + + - ``thread_anaphora_prefix`` is deterministic and pure. + - Fires only when BOTH the prior turn (recovered from + ``ThreadContext``) AND the current turn are pack/teaching tier. + - Same-intent revisits do not fire (redundant prefix). + - Format is structural-fields-only: turn index + chain_id or + grounding tier; never re-derives prose. + - Live runtime: ``RuntimeConfig.thread_anaphora=False`` keeps + every existing surface byte-identical (default off). + - With the flag on AND seeded thread state, the runtime prepends + the deterministic backreference to the pack-grounded surface. +""" + +from __future__ import annotations + +import pytest + +from chat.anaphora import thread_anaphora_prefix +from chat.runtime import ChatRuntime +from chat.thread_context import ThreadContext, TurnSummary +from core.config import RuntimeConfig + + +def _seed(tc: ThreadContext, **kw) -> None: + """Helper to push a TurnSummary into a context.""" + tc.push(TurnSummary( + turn_index=kw.get("turn_index", 0), + intent_tag_name=kw.get("intent", "cause"), + subject=kw.get("subject", "light"), + grounding_source=kw.get("source", "teaching"), + chain_id=kw.get("chain_id"), + corpus_id=kw.get("corpus_id"), + )) + + +# --------------------------------------------------------------------------- +# Pure-function contract +# --------------------------------------------------------------------------- + + +def test_prior_teaching_current_pack_fires_with_chain_id() -> None: + tc = ThreadContext() + _seed(tc, turn_index=0, intent="cause", subject="light", + source="teaching", chain_id="cause_light_reveals_truth", + corpus_id="cognition_chains_v1") + prefix = thread_anaphora_prefix(tc, "light", "definition", "pack") + assert prefix is not None + assert "Recalling turn 0" in prefix + assert "cause_light_reveals_truth" in prefix + + +def test_prior_pack_current_teaching_fires_with_grounding_tier() -> None: + tc = ThreadContext() + _seed(tc, turn_index=2, intent="definition", subject="light", source="pack") + prefix = thread_anaphora_prefix(tc, "light", "cause", "teaching") + assert prefix is not None + assert "Recalling turn 2" in prefix + assert "pack" in prefix + assert "light" in prefix + + +def test_same_intent_revisit_does_not_fire() -> None: + """Asking the same question twice on the same subject — the + prior turn IS the same surface modulo vault drift; prefixing + would be redundant.""" + tc = ThreadContext() + _seed(tc, turn_index=0, intent="definition", subject="light", source="pack") + assert thread_anaphora_prefix(tc, "light", "definition", "pack") is None + + +def test_prior_weak_tier_does_not_anchor() -> None: + """Prior turn whose grounding was OOV / partial / vault / none + is not a strong-enough anchor.""" + tc = ThreadContext() + _seed(tc, source="oov") + # recent_for_subject excludes OOV by default → prior lookup returns None. + assert thread_anaphora_prefix(tc, "light", "definition", "pack") is None + + tc2 = ThreadContext() + _seed(tc2, source="partial") + assert thread_anaphora_prefix(tc2, "light", "definition", "pack") is None + + +def test_current_weak_tier_does_not_fire() -> None: + """Anaphora is a *forward-reference* prefix on a strongly-grounded + surface; weak-tier current turns are not hosts.""" + tc = ThreadContext() + _seed(tc, source="teaching", chain_id="x") + assert thread_anaphora_prefix(tc, "light", "definition", "oov") is None + assert thread_anaphora_prefix(tc, "light", "definition", "partial") is None + assert thread_anaphora_prefix(tc, "light", "definition", "none") is None + + +def test_empty_subject_returns_none() -> None: + tc = ThreadContext() + _seed(tc, source="teaching", chain_id="x") + assert thread_anaphora_prefix(tc, "", "definition", "pack") is None + assert thread_anaphora_prefix(tc, " ", "definition", "pack") is None + + +def test_no_recent_match_returns_none() -> None: + tc = ThreadContext() + _seed(tc, subject="light", source="teaching", chain_id="x") + assert thread_anaphora_prefix(tc, "memory", "definition", "pack") is None + + +def test_most_recent_match_wins() -> None: + """If the same subject appears twice, the most recent matching + grounded turn is the anchor — not the earliest.""" + tc = ThreadContext() + _seed(tc, turn_index=0, intent="cause", subject="light", + source="teaching", chain_id="cause_old") + _seed(tc, turn_index=2, intent="cause", subject="light", + source="teaching", chain_id="cause_new") + prefix = thread_anaphora_prefix(tc, "light", "definition", "pack") + assert prefix is not None + assert "cause_new" in prefix + assert "turn 2" in prefix + + +def test_is_deterministic() -> None: + tc = ThreadContext() + _seed(tc, source="teaching", chain_id="x") + a = thread_anaphora_prefix(tc, "light", "definition", "pack") + b = thread_anaphora_prefix(tc, "light", "definition", "pack") + assert a == b + + +# --------------------------------------------------------------------------- +# Live runtime integration +# --------------------------------------------------------------------------- + + +def test_runtime_default_off_preserves_surface_bytewise() -> None: + """With ``thread_anaphora=False`` (default), even a seeded thread + context does not alter the emitted surface.""" + rt = ChatRuntime() # default config + rt.thread_context.push(TurnSummary( + turn_index=0, intent_tag_name="cause", subject="light", + grounding_source="teaching", chain_id="cause_light_reveals_truth", + corpus_id="cognition_chains_v1", + )) + resp = rt.chat("What is light?") + assert resp.grounding_source == "pack" + assert "Recalling" not in resp.surface + assert "light — pack-grounded" in resp.surface + + +def test_runtime_anaphora_on_emits_prefix() -> None: + cfg = RuntimeConfig(thread_anaphora=True) + rt = ChatRuntime(config=cfg) + rt.thread_context.push(TurnSummary( + turn_index=0, intent_tag_name="cause", subject="light", + grounding_source="teaching", chain_id="cause_light_reveals_truth", + corpus_id="cognition_chains_v1", + )) + resp = rt.chat("What is light?") + assert resp.grounding_source == "pack" + assert "Recalling turn 0" in resp.surface + assert "cause_light_reveals_truth" in resp.surface + # Prefix precedes the pack-grounded surface, never replaces it. + assert "light — pack-grounded" in resp.surface + + +def test_runtime_anaphora_on_does_not_fire_without_anchor() -> None: + """With the flag on but NO prior turn on the subject, the prefix + composer returns None and the surface is unchanged.""" + cfg = RuntimeConfig(thread_anaphora=True) + rt = ChatRuntime(config=cfg) + resp = rt.chat("What is light?") + assert resp.grounding_source == "pack" + assert "Recalling" not in resp.surface diff --git a/tests/test_thread_context.py b/tests/test_thread_context.py new file mode 100644 index 00000000..954eba46 --- /dev/null +++ b/tests/test_thread_context.py @@ -0,0 +1,254 @@ +"""Phase 3.1 — session-thread context tests. + +The contract these tests pin: + + - ``TurnSummary`` is frozen + carries only structured fields. + - ``ThreadContext`` is bounded (FIFO eviction at max_turns). + - ``push`` appends; ``snapshot`` returns the deque in insertion + order (oldest first). + - ``recent_for_subject`` returns the most-recent matching summary, + skipping ungrounded tiers by default. + - ``recent_subjects`` returns unique subjects most-recent-first. + - ChatRuntime owns one ThreadContext; pushes a summary at + end-of-turn for BOTH the stub path and the walk path. + - Teaching-grounded turns carry chain_id + corpus_id in the + summary so anaphora composers can detect same-chain reference. +""" + +from __future__ import annotations + +import pytest + +from chat.runtime import ChatRuntime +from chat.thread_context import ( + MAX_THREAD_TURNS, + ThreadContext, + TurnSummary, +) + + +# --------------------------------------------------------------------------- +# TurnSummary dataclass +# --------------------------------------------------------------------------- + + +def test_turn_summary_is_frozen() -> None: + s = TurnSummary(turn_index=0, intent_tag_name="cause", + subject="light", grounding_source="teaching", + chain_id="cause_light_reveals_truth", + corpus_id="cognition_chains_v1") + with pytest.raises((AttributeError, TypeError)): + s.subject = "other" # type: ignore[misc] + + +def test_turn_summary_as_dict_round_trips() -> None: + s = TurnSummary(turn_index=3, intent_tag_name="definition", + subject="parent", grounding_source="pack") + blob = s.as_dict() + assert blob["turn_index"] == 3 + assert blob["subject"] == "parent" + assert blob["grounding_source"] == "pack" + assert blob["chain_id"] is None + assert blob["corpus_id"] is None + + +# --------------------------------------------------------------------------- +# ThreadContext bounded FIFO +# --------------------------------------------------------------------------- + + +def test_default_capacity() -> None: + tc = ThreadContext() + assert tc.max_turns == MAX_THREAD_TURNS + + +def test_invalid_capacity_raises() -> None: + with pytest.raises(ValueError): + ThreadContext(max_turns=0) + with pytest.raises(ValueError): + ThreadContext(max_turns=-1) + + +def test_push_and_snapshot_preserve_order() -> None: + tc = ThreadContext(max_turns=4) + for i in range(3): + tc.push(TurnSummary(turn_index=i, intent_tag_name="cause", + subject=f"s{i}", grounding_source="teaching")) + snap = tc.snapshot() + assert [s.turn_index for s in snap] == [0, 1, 2] + assert [s.subject for s in snap] == ["s0", "s1", "s2"] + + +def test_eviction_drops_oldest() -> None: + tc = ThreadContext(max_turns=3) + for i in range(5): + tc.push(TurnSummary(turn_index=i, intent_tag_name="cause", + subject=f"s{i}", grounding_source="teaching")) + snap = tc.snapshot() + assert len(snap) == 3 + # 0, 1 evicted; 2, 3, 4 retained. + assert [s.turn_index for s in snap] == [2, 3, 4] + + +def test_clear_resets_state() -> None: + tc = ThreadContext(max_turns=4) + tc.push(TurnSummary(turn_index=0, intent_tag_name="cause", + subject="x", grounding_source="teaching")) + assert len(tc) == 1 + tc.clear() + assert len(tc) == 0 + + +# --------------------------------------------------------------------------- +# recent_for_subject +# --------------------------------------------------------------------------- + + +def test_recent_for_subject_returns_most_recent_match() -> None: + tc = ThreadContext() + tc.push(TurnSummary(turn_index=0, intent_tag_name="cause", + subject="light", grounding_source="teaching")) + tc.push(TurnSummary(turn_index=1, intent_tag_name="cause", + subject="memory", grounding_source="teaching")) + tc.push(TurnSummary(turn_index=2, intent_tag_name="definition", + subject="light", grounding_source="pack")) + match = tc.recent_for_subject("light") + assert match is not None + assert match.turn_index == 2 # most recent, not the first + assert match.intent_tag_name == "definition" + + +def test_recent_for_subject_skips_ungrounded_by_default() -> None: + """OOV / partial / none turns are excluded from recency lookup + by default — they're not strong-enough anchors for anaphora.""" + tc = ThreadContext() + tc.push(TurnSummary(turn_index=0, intent_tag_name="cause", + subject="light", grounding_source="teaching")) + tc.push(TurnSummary(turn_index=1, intent_tag_name="definition", + subject="light", grounding_source="oov")) + match = tc.recent_for_subject("light") + assert match is not None + assert match.turn_index == 0 # teaching turn wins; oov skipped + + +def test_recent_for_subject_can_include_excluded_tiers() -> None: + tc = ThreadContext() + tc.push(TurnSummary(turn_index=0, intent_tag_name="cause", + subject="light", grounding_source="teaching")) + tc.push(TurnSummary(turn_index=1, intent_tag_name="definition", + subject="light", grounding_source="oov")) + match = tc.recent_for_subject("light", exclude_grounding=()) + assert match is not None + assert match.turn_index == 1 + + +def test_recent_for_subject_normalises_input() -> None: + tc = ThreadContext() + tc.push(TurnSummary(turn_index=0, intent_tag_name="cause", + subject="light", grounding_source="teaching")) + for query in ("LIGHT", "Light", " light "): + match = tc.recent_for_subject(query) + assert match is not None + + +def test_recent_for_subject_returns_none_when_absent() -> None: + tc = ThreadContext() + assert tc.recent_for_subject("anything") is None + tc.push(TurnSummary(turn_index=0, intent_tag_name="cause", + subject="x", grounding_source="teaching")) + assert tc.recent_for_subject("y") is None + + +# --------------------------------------------------------------------------- +# recent_subjects +# --------------------------------------------------------------------------- + + +def test_recent_subjects_unique_most_recent_first() -> None: + tc = ThreadContext() + tc.push(TurnSummary(turn_index=0, intent_tag_name="cause", + subject="light", grounding_source="teaching")) + tc.push(TurnSummary(turn_index=1, intent_tag_name="cause", + subject="memory", grounding_source="teaching")) + tc.push(TurnSummary(turn_index=2, intent_tag_name="definition", + subject="light", grounding_source="pack")) + subjects = tc.recent_subjects() + assert subjects == ("light", "memory") + + +def test_recent_subjects_skips_empty_and_ungrounded() -> None: + tc = ThreadContext() + tc.push(TurnSummary(turn_index=0, intent_tag_name="", + subject="", grounding_source="vault")) + tc.push(TurnSummary(turn_index=1, intent_tag_name="definition", + subject="photosynthesis", grounding_source="oov")) + tc.push(TurnSummary(turn_index=2, intent_tag_name="cause", + subject="light", grounding_source="teaching")) + subjects = tc.recent_subjects() + assert subjects == ("light",) + + +# --------------------------------------------------------------------------- +# Runtime integration — ChatRuntime pushes after each turn +# --------------------------------------------------------------------------- + + +def test_runtime_pushes_summary_on_cold_start_pack_turn() -> None: + rt = ChatRuntime() + rt.chat("What is light?") + snap = rt.thread_context.snapshot() + assert len(snap) == 1 + s = snap[0] + assert s.intent_tag_name == "definition" + assert s.subject == "light" + assert s.grounding_source == "pack" + + +def test_runtime_pushes_chain_id_for_teaching_grounded_turn() -> None: + rt = ChatRuntime() + rt.chat("Why does light exist?") + s = rt.thread_context.snapshot()[0] + assert s.grounding_source == "teaching" + assert s.chain_id == "cause_light_reveals_truth" + assert s.corpus_id == "cognition_chains_v1" + + +def test_runtime_pushes_summary_for_oov_turn() -> None: + rt = ChatRuntime() + rt.chat("What is photosynthesis?") + s = rt.thread_context.snapshot()[0] + assert s.intent_tag_name == "definition" + assert s.subject == "photosynthesis" + assert s.grounding_source == "oov" + assert s.chain_id is None + + +def test_runtime_thread_context_grows_across_turns() -> None: + rt = ChatRuntime() + rt.chat("What is light?") + rt.chat("What is parent?") + rt.chat("What is photosynthesis?") + snap = rt.thread_context.snapshot() + assert len(snap) == 3 + assert snap[0].subject == "light" + assert snap[1].subject == "parent" + assert snap[2].subject == "photosynthesis" + + +def test_runtime_thread_context_indexes_match_turn_log() -> None: + rt = ChatRuntime() + rt.chat("What is light?") + rt.chat("What is parent?") + snap = rt.thread_context.snapshot() + assert [s.turn_index for s in snap] == list(range(len(rt.turn_log))) + + +def test_runtime_default_capacity_evicts_old_turns() -> None: + rt = ChatRuntime() + for _ in range(MAX_THREAD_TURNS + 3): + rt.chat("What is light?") + snap = rt.thread_context.snapshot() + assert len(snap) == MAX_THREAD_TURNS + # Oldest retained turn_index is (total - MAX_THREAD_TURNS). + total_turns = len(rt.turn_log) + assert snap[0].turn_index == total_turns - MAX_THREAD_TURNS