feat(adr-0066): session-thread context + opt-in anaphora prefix (Phase 3.1 + 3.2)

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 <chain_id>.) "    # prior was teaching
    "(Recalling turn N: <subject> 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

@@ -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
+        <chain_id>.) "``
+      - Prior was pack-grounded → ``"(Recalling turn N: <subject>
+        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"]

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
+            # Classify intent up-front whenever the gate fired on an
-            # OR the OOV sink is attached (P2.3) — both downstream
+            # empty vault.  P2.3 needs it for OOV sink emission,
-            # emission paths need it.  Without either sink attached
+            # ADR-0055 Phase B needs it for discovery sink emission,
-            # this is a no-op, so behaviour stays identical to the
+            # and P3.1 needs it for session-thread context — the
-            # pre-Phase-2 path.
+            # 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)
+                gate_decision.source == "empty_vault"
-                and 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,

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"]

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"

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

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