feat(adr-0055): Phase B — DiscoveryCandidate emission from turn loop

Lands the first deterministic trigger of the discovery → reviewed-
memory loop. Candidates are structured evidence; emission is
opt-in via attach_discovery_sink and NEVER mutates the active
teaching corpus.

- teaching/discovery.py: DiscoveryCandidate dataclass + pure
  extract_discovery_candidates(turn_event, intent, subject) rule
  firing. Phase B fires only the would_have_grounded trigger:
    grounding_source == "none"
    AND intent ∈ {CAUSE, VERIFICATION}
    AND subject lemma in ratified cognition pack
    AND (subject, intent) NOT in active corpus
  candidate_id = SHA-256 of canonical JSON payload — replay-stable.
  Other DiscoveryTrigger literals (successful_comparison,
  hedge_acknowledged, oov_resolved_via_decomp) are reserved for
  later phases.

- teaching/discovery_sink.py: DiscoveryCandidateSink protocol,
  DiscoveryBufferSink (in-memory), DiscoveryMonthlyFileSink
  (append-only JSONL, <root>/<YYYY>/<YYYY-MM>.jsonl rollover,
  injectable clock).

- chat/runtime.py: opt-in attach_discovery_sink, post-turn
  emission inside _stub_response only when caller threads
  classified intent forward (gate-fire fall-through site).
  Intent classification at the call site reuses the same
  deterministic classifier already invoked by
  _maybe_pack_grounded_surface for the empty-vault English path.

Trust boundary: candidates write to a separate sink/file path
only; the active corpus on disk is never touched. Tests
explicitly assert corpus bytes are byte-identical before and
after a candidate-emitting turn.

Tests: tests/test_discovery_candidates.py — 24 tests covering
pure-predicate rule firing, every short-circuit path,
deterministic candidate_id, sink opt-in, runtime parity with no
sink, monthly rollover semantics, append-only behaviour, no
corpus mutation.

Lanes: smoke 67, cognition 121, runtime 19, teaching 17, packs 6
— all green. Cognition eval metrics unchanged on dev / public /
holdout splits. versor_condition < 1e-6 invariant untouched.

chat/runtime.py

@@ -5,7 +5,7 @@ import hashlib
 import json
 import re
 from collections.abc import Sequence
-from typing import List
+from typing import Any, List
 
 import numpy as np
 
@@ -28,6 +28,11 @@ from chat.refusal import (
 )
 from chat.telemetry import TurnEventSink, format_turn_event_jsonl
 from chat.verdicts import TurnVerdicts
+from teaching.discovery import (
+    extract_discovery_candidates,
+    format_candidate_jsonl,
+)
+from teaching.discovery_sink import DiscoveryCandidateSink
 from core.config import DEFAULT_CONFIG, DEFAULT_IDENTITY_PACK, RuntimeConfig
 from core.physics.drive import DriveGradientMap, GradientField
 from core.physics.energy import EnergyProfile
@@ -408,6 +413,11 @@ class ChatRuntime:
         # trust boundary.
         self._telemetry_sink: TurnEventSink | None = None
         self._telemetry_include_content: bool = False
+        # ADR-0055 Phase B — opt-in DiscoveryCandidate sink.  Default
+        # None preserves prior behavior; callers attach via
+        # ``attach_discovery_sink``.  Candidates are *evidence*, never
+        # mutate the corpus or runtime state.
+        self._discovery_sink: DiscoveryCandidateSink | None = None
         self._correction_pass = CorrectionPass()
         self._last_valence: float = 0.0
 
@@ -435,6 +445,44 @@ class ChatRuntime:
         self._telemetry_sink = sink
         self._telemetry_include_content = bool(include_content)
 
+    def attach_discovery_sink(
+        self,
+        sink: DiscoveryCandidateSink | None,
+    ) -> None:
+        """ADR-0055 Phase B — attach a DiscoveryCandidate sink.
+
+        After each turn, the runtime extracts zero-or-more candidates
+        from the most recent ``TurnEvent`` (deterministic rule firing
+        on the audit trail) and forwards each as one JSONL line.
+        Passing ``None`` detaches.
+
+        Candidates are **evidence**: emission never mutates the
+        active teaching corpus.  Phase C's ``TeachingChainProposal``
+        is the only path to corpus extension and runs through
+        review + replay.
+        """
+        self._discovery_sink = sink
+
+    def _emit_discovery_candidates(
+        self,
+        *,
+        turn_event: TurnEvent,
+        intent_tag: Any,
+        intent_subject: str | None,
+        grounding_source: str | None,
+    ) -> None:
+        sink = self._discovery_sink
+        if sink is None:
+            return
+        candidates = extract_discovery_candidates(
+            turn_event,
+            intent_tag,
+            intent_subject,
+            grounding_source=grounding_source,
+        )
+        for candidate in candidates:
+            sink.emit(format_candidate_jsonl(candidate))
+
     def _emit_turn_event(self, event: TurnEvent) -> None:
         """Internal — emit one serialised line for the current event.
 
@@ -612,6 +660,8 @@ class ChatRuntime:
         tokens: tuple[str, ...] = (),
         pack_grounded_surface: str | None = None,
         grounded_source_tag: str = "pack",
+        discovery_intent_tag: Any = None,
+        discovery_intent_subject: str | None = None,
     ) -> ChatResponse:
         zero = np.zeros(field_state.F.shape, dtype=np.float32)
         prop = Proposition(
@@ -723,6 +773,17 @@ class ChatRuntime:
             )
             self.turn_log.append(stub_event)
             self._emit_turn_event(stub_event)
+            # ADR-0055 Phase B — opt-in discovery candidate emission.
+            # Only meaningful when the caller threads classified
+            # intent forward (gate-fire / fall-through site).  Pure
+            # rule firing on the just-appended TurnEvent.
+            if discovery_intent_tag is not None:
+                self._emit_discovery_candidates(
+                    turn_event=stub_event,
+                    intent_tag=discovery_intent_tag,
+                    intent_subject=discovery_intent_subject,
+                    grounding_source=grounding_source,
+                )
         return ChatResponse(
             surface=response_surface,
             proposition=prop,
@@ -793,11 +854,30 @@ class ChatRuntime:
                     "grounding_source": pack_source_tag if pack_surface else "none",
                 },
             )
+            # ADR-0055 Phase B — thread classified intent forward only
+            # when a sink is attached.  Discovery emission is opt-in;
+            # the deterministic classification used here is the same
+            # call ``_maybe_pack_grounded_surface`` already ran for the
+            # empty-vault English path, so behaviour is identical when
+            # no sink is attached.
+            discovery_intent_tag = None
+            discovery_intent_subject: str | None = None
+            if (
+                self._discovery_sink is not None
+                and gate_decision.source == "empty_vault"
+                and self.config.output_language == "en"
+            ):
+                from generate.intent_bridge import classify_intent_from_input
+                _intent = classify_intent_from_input(text)
+                discovery_intent_tag = _intent.tag
+                discovery_intent_subject = _intent.subject
             return self._stub_response(
                 committed,
                 tokens=tuple(filtered),
                 pack_grounded_surface=pack_surface,
                 grounded_source_tag=pack_source_tag,
+                discovery_intent_tag=discovery_intent_tag,
+                discovery_intent_subject=discovery_intent_subject,
             )
 
         field_state = self._context.commit_ingest(filtered)

docs/decisions/ADR-0055-inter-session-memory-discovery-promotion.md

@@ -1,6 +1,6 @@
 # ADR-0055 — Inter-Session Memory: Reviewed Discovery Promotion
 
-**Status:** Proposed
+**Status:** Phase A + Phase B Accepted; Phases C–E Proposed
 **Date:** 2026-05-18
 **Author:** Shay
 

teaching/discovery.py

@@ -0,0 +1,232 @@
+"""ADR-0055 Phase B — DiscoveryCandidate emission from the turn loop.
+
+A ``DiscoveryCandidate`` is **structured evidence** — never a
+mutation.  When a turn's audit trail satisfies a deterministic
+predicate, an entry is emitted to the discovery candidate stream.
+Candidates **never** load into the active teaching corpus; the only
+path to corpus extension is the review-gated
+``TeachingChainProposal`` (Phase C, not yet built).
+
+Trigger set (Phase B lands the first; the others are stubbed in the
+``Literal`` so the structure is stable when later phases add them):
+
+  - ``would_have_grounded`` — the turn fell through to the universal
+    "insufficient grounding" disclosure, the classified intent was
+    ``CAUSE`` or ``VERIFICATION``, the subject lemma is in the
+    ratified cognition pack, and no active chain matched
+    ``(subject, intent)``.  A reviewed chain of that subject/intent
+    would have grounded the turn.
+
+  - ``successful_comparison`` — open question §5 in ADR-0055; not
+    fired in Phase B.
+
+  - ``hedge_acknowledged`` — open question §5 in ADR-0055; not
+    fired in Phase B.
+
+  - ``oov_resolved_via_decomp`` — not fired in Phase B.
+
+Determinism contract:
+
+  - ``extract_discovery_candidates`` is a pure function of its
+    inputs.
+  - ``candidate_id`` is a SHA-256 hash of a canonical JSON encoding
+    of the candidate's load-bearing fields; identical inputs always
+    produce the identical id.
+  - No LLM, no stochastic sampling, no clock-time read.
+
+Trust boundary:
+
+  - This module reads pack + corpus indices and a ``TurnEvent``.
+    It never writes to the corpus, the pack, or runtime state.
+  - The ``source_turn_trace`` is the upstream ``TurnEvent.trace_hash``
+    when present; absent that, the empty string.  Tying every
+    candidate to a replayable turn is the load-bearing audit
+    property.
+"""
+
+from __future__ import annotations
+
+import hashlib
+import json
+from dataclasses import dataclass
+from typing import Any, Literal
+
+from chat.pack_grounding import _pack_index
+from chat.teaching_grounding import _corpus_index
+from generate.intent import IntentTag
+
+
+DiscoveryTrigger = Literal[
+    "would_have_grounded",
+    "successful_comparison",
+    "hedge_acknowledged",
+    "oov_resolved_via_decomp",
+]
+
+
+@dataclass(frozen=True, slots=True)
+class DiscoveryCandidate:
+    """Structured evidence that a reviewed chain would have helped.
+
+    ``proposed_chain`` is *partial* by design: Phase B can only see
+    that a chain of a given ``(subject, intent)`` would have grounded
+    the turn — it cannot infer the connective or object.  Phase C's
+    ``TeachingChainProposal`` is the place where a complete proposed
+    entry is constructed and gated through review + replay.
+    """
+
+    candidate_id: str
+    proposed_chain: dict[str, Any]
+    trigger: DiscoveryTrigger
+    source_turn_trace: str
+    pack_consistent: bool
+    boundary_clean: bool
+    review_state: Literal["unreviewed"] = "unreviewed"
+
+    def as_dict(self) -> dict[str, Any]:
+        return {
+            "candidate_id": self.candidate_id,
+            "proposed_chain": self.proposed_chain,
+            "trigger": self.trigger,
+            "source_turn_trace": self.source_turn_trace,
+            "pack_consistent": self.pack_consistent,
+            "boundary_clean": self.boundary_clean,
+            "review_state": self.review_state,
+        }
+
+
+_TEACHING_INTENT_NAME: dict[IntentTag, str] = {
+    IntentTag.CAUSE: "cause",
+    IntentTag.VERIFICATION: "verification",
+}
+
+
+def _hash_candidate_id(payload: dict[str, Any]) -> str:
+    """Deterministic SHA-256 over a canonical JSON encoding.
+
+    Sorted keys + tight separators keep the hash stable across
+    Python runtimes and dict-insertion order.  This is the
+    ``candidate_id`` — used both as the on-disk JSONL line key and
+    by Phase C to look up the originating candidate.
+    """
+    blob = json.dumps(payload, sort_keys=True, separators=(",", ":"))
+    return hashlib.sha256(blob.encode("utf-8")).hexdigest()
+
+
+def _boundary_clean(turn_event: Any) -> bool:
+    """Return True iff the source turn produced no safety/ethics
+    refusal and no hedge injection.
+
+    Tolerates events that pre-date the bundled-verdicts era (ADR-0039
+    onward) by reading the canonical fields directly.
+    """
+    refusal_emitted = bool(getattr(turn_event, "refusal_emitted", False) or False)
+    hedge_injected = bool(getattr(turn_event, "hedge_injected", False) or False)
+    if refusal_emitted or hedge_injected:
+        return False
+    verdicts = getattr(turn_event, "verdicts", None)
+    if verdicts is not None:
+        if bool(getattr(verdicts, "refusal_emitted", False) or False):
+            return False
+        if bool(getattr(verdicts, "hedge_injected", False) or False):
+            return False
+    return True
+
+
+def _trace_hash(turn_event: Any) -> str:
+    value = getattr(turn_event, "trace_hash", "") or ""
+    return str(value)
+
+
+def extract_discovery_candidates(
+    turn_event: Any,
+    intent_tag: IntentTag | None,
+    intent_subject_lemma: str | None,
+    *,
+    grounding_source: str | None = None,
+) -> tuple[DiscoveryCandidate, ...]:
+    """Return zero or more DiscoveryCandidates for a single turn.
+
+    Phase B only fires the ``would_have_grounded`` trigger.  All
+    other triggers in the ``DiscoveryTrigger`` Literal are reserved
+    for later phases.
+
+    Fires when **every** condition holds (deterministic predicate):
+
+      1. ``grounding_source`` is ``"none"`` or absent — the turn
+         fell through to the universal disclosure.
+      2. ``intent_tag`` is ``CAUSE`` or ``VERIFICATION`` — the
+         intent set the teaching-grounded surface answers.
+      3. ``intent_subject_lemma`` is a non-empty pack lemma in the
+         ratified cognition pack.
+      4. ``(subject_lemma, intent_name)`` is **not** in the active
+         corpus — a chain of that shape would have grounded the
+         turn but does not exist.
+
+    Order of conditions matters for tests: short-circuit on the
+    cheapest predicate first.
+    """
+    source = (grounding_source or getattr(turn_event, "grounding_source", "none") or "none").lower()
+    if source != "none":
+        return ()
+    if intent_tag is None or intent_tag not in _TEACHING_INTENT_NAME:
+        return ()
+    if not intent_subject_lemma or not isinstance(intent_subject_lemma, str):
+        return ()
+    lemma = intent_subject_lemma.strip().lower()
+    if not lemma:
+        return ()
+
+    pack = _pack_index()
+    if lemma not in pack:
+        return ()
+
+    intent_name = _TEACHING_INTENT_NAME[intent_tag]
+    if (lemma, intent_name) in _corpus_index():
+        return ()
+
+    # The candidate's proposed_chain is intentionally partial: Phase B
+    # can only assert that a chain of this (subject, intent) would
+    # have helped.  Connective and object remain null; Phase C is
+    # where a complete proposed entry is constructed and review-gated.
+    proposed_chain = {
+        "subject": lemma,
+        "intent": intent_name,
+        "connective": None,
+        "object": None,
+    }
+
+    trace_hash = _trace_hash(turn_event)
+    boundary_clean = _boundary_clean(turn_event)
+    trigger: DiscoveryTrigger = "would_have_grounded"
+
+    hash_payload = {
+        "proposed_chain": proposed_chain,
+        "trigger": trigger,
+        "source_turn_trace": trace_hash,
+    }
+    candidate_id = _hash_candidate_id(hash_payload)
+
+    candidate = DiscoveryCandidate(
+        candidate_id=candidate_id,
+        proposed_chain=proposed_chain,
+        trigger=trigger,
+        source_turn_trace=trace_hash,
+        pack_consistent=True,  # subject is in pack; object is null pending Phase C
+        boundary_clean=boundary_clean,
+        review_state="unreviewed",
+    )
+    return (candidate,)
+
+
+def format_candidate_jsonl(candidate: DiscoveryCandidate) -> str:
+    """Serialise to one JSONL line (sorted keys, no trailing newline)."""
+    return json.dumps(candidate.as_dict(), sort_keys=True, separators=(",", ":"))
+
+
+__all__ = [
+    "DiscoveryCandidate",
+    "DiscoveryTrigger",
+    "extract_discovery_candidates",
+    "format_candidate_jsonl",
+]

teaching/discovery_sink.py

@@ -0,0 +1,105 @@
+"""ADR-0055 Phase B — sinks for DiscoveryCandidate emission.
+
+Mirrors the telemetry-sink shape from ADR-0040 / ADR-0041:
+
+  - Protocol ``DiscoveryCandidateSink.emit(line)``.
+  - ``DiscoveryBufferSink`` — in-memory, useful for tests and small-
+    volume audit where persistence is the caller's responsibility.
+  - ``DiscoveryMonthlyFileSink`` — append-only JSONL with per-month
+    rollover under ``<root>/<YYYY>/<YYYY-MM>.jsonl``.  Path computed
+    deterministically from an injected ``Clock`` so tests can pin
+    the rollover instant without monkey-patching ``datetime``.
+
+Trust boundary:
+
+  - Append-only.  No truncation, no rewrite.  Each ``emit()`` flushes
+    so a crashed runtime keeps its prior candidates durable on disk.
+  - Sink errors are NOT swallowed by the runtime — Phase B keeps
+    telemetry's fail-fast contract.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from datetime import datetime, timezone
+from pathlib import Path
+from typing import IO, Callable, Protocol
+
+
+class DiscoveryCandidateSink(Protocol):
+    """Minimal sink contract — one JSONL line per emission."""
+
+    def emit(self, line: str) -> None: ...
+
+
+@dataclass
+class DiscoveryBufferSink:
+    """In-memory sink that captures every emitted candidate line."""
+
+    lines: list[str] = field(default_factory=list)
+
+    def emit(self, line: str) -> None:
+        self.lines.append(line)
+
+
+Clock = Callable[[], datetime]
+
+
+def _utc_now() -> datetime:
+    return datetime.now(timezone.utc)
+
+
+class DiscoveryMonthlyFileSink:
+    """Append-only JSONL sink with monthly rollover.
+
+    Path is computed at each ``emit()`` from the injected clock as
+    ``<root>/<YYYY>/<YYYY-MM>.jsonl``.  The file handle is reopened
+    when the month rolls over; the previous month's file is closed
+    cleanly.
+
+    The clock is injected (default UTC) so tests can pin the
+    rollover instant without monkey-patching ``datetime``.
+    """
+
+    def __init__(self, root: str | Path, *, clock: Clock = _utc_now) -> None:
+        self._root = Path(root)
+        self._clock = clock
+        self._fh: IO[str] | None = None
+        self._current_path: Path | None = None
+
+    def _path_for_now(self) -> Path:
+        now = self._clock()
+        return self._root / f"{now.year:04d}" / f"{now.year:04d}-{now.month:02d}.jsonl"
+
+    def emit(self, line: str) -> None:
+        target = self._path_for_now()
+        if target != self._current_path:
+            if self._fh is not None:
+                self._fh.close()
+                self._fh = None
+            target.parent.mkdir(parents=True, exist_ok=True)
+            self._fh = target.open("a", encoding="utf-8")
+            self._current_path = target
+        assert self._fh is not None
+        self._fh.write(line)
+        self._fh.write("\n")
+        self._fh.flush()
+
+    def close(self) -> None:
+        if self._fh is not None:
+            self._fh.close()
+            self._fh = None
+            self._current_path = None
+
+    def __enter__(self) -> "DiscoveryMonthlyFileSink":
+        return self
+
+    def __exit__(self, *exc_info) -> None:
+        self.close()
+
+
+__all__ = [
+    "DiscoveryCandidateSink",
+    "DiscoveryBufferSink",
+    "DiscoveryMonthlyFileSink",
+]

tests/test_discovery_candidates.py

@@ -0,0 +1,396 @@
+"""ADR-0055 Phase B — DiscoveryCandidate emission + sink contracts.
+
+Pinned contracts:
+
+  - ``extract_discovery_candidates`` is pure and deterministic; same
+    inputs ⇒ identical ``candidate_id``.
+  - The ``would_have_grounded`` trigger fires only when ALL of:
+    fall-through (grounding_source == "none"), intent ∈ {CAUSE,
+    VERIFICATION}, subject lemma is a pack lemma, (subject, intent)
+    is NOT in the active corpus.
+  - Candidate emission **never** mutates the active teaching corpus
+    or runtime state.
+  - Sink is opt-in.  No sink attached ⇒ behaviour identical to pre-B.
+  - ``DiscoveryMonthlyFileSink`` rolls over by month; previous file
+    is closed cleanly.
+"""
+
+from __future__ import annotations
+
+import json
+from pathlib import Path
+from datetime import datetime, timezone
+import pytest
+
+from chat.runtime import ChatRuntime
+from generate.intent import IntentTag
+from teaching.discovery import (
+    DiscoveryCandidate,
+    extract_discovery_candidates,
+    format_candidate_jsonl,
+)
+from teaching.discovery_sink import (
+    DiscoveryBufferSink,
+    DiscoveryMonthlyFileSink,
+)
+
+
+# ---------------------------------------------------------------------------
+# Synthetic TurnEvent shim (avoids needing a full runtime turn just to test
+# the pure rule firing)
+# ---------------------------------------------------------------------------
+
+
+class _TE:
+    def __init__(
+        self,
+        *,
+        grounding_source: str = "none",
+        trace_hash: str = "abcd",
+        refusal_emitted: bool = False,
+        hedge_injected: bool = False,
+    ) -> None:
+        self.grounding_source = grounding_source
+        self.trace_hash = trace_hash
+        self.refusal_emitted = refusal_emitted
+        self.hedge_injected = hedge_injected
+        self.verdicts = None
+
+
+# ---------------------------------------------------------------------------
+# extract_discovery_candidates — pure predicate
+# ---------------------------------------------------------------------------
+
+
+def test_fires_on_would_have_grounded_for_cause_on_unknown_pack_lemma() -> None:
+    """``judgment`` is a pack lemma; (judgment, cause) is NOT in the
+    corpus today (the existing chains for ``judgment`` are
+    VERIFICATION + CAUSE→ordered_by_wisdom, but no ``(judgment,
+    cause)`` SUBJECT key exists)."""
+    # First check the pack actually has the lemma we're claiming.
+    from chat.pack_grounding import _pack_index
+    from chat.teaching_grounding import _corpus_index
+
+    assert "judgment" in _pack_index()
+    if ("judgment", "cause") in _corpus_index():
+        pytest.skip("judgment/cause is in the active corpus; pick another fixture")
+
+    cands = extract_discovery_candidates(
+        _TE(grounding_source="none"),
+        IntentTag.CAUSE,
+        "judgment",
+    )
+    assert len(cands) == 1
+    c = cands[0]
+    assert c.trigger == "would_have_grounded"
+    assert c.proposed_chain["subject"] == "judgment"
+    assert c.proposed_chain["intent"] == "cause"
+    assert c.proposed_chain["connective"] is None
+    assert c.proposed_chain["object"] is None
+    assert c.pack_consistent is True
+    assert c.boundary_clean is True
+    assert c.review_state == "unreviewed"
+
+
+def test_does_not_fire_when_grounding_source_is_pack() -> None:
+    cands = extract_discovery_candidates(
+        _TE(grounding_source="pack"),
+        IntentTag.CAUSE,
+        "judgment",
+    )
+    assert cands == ()
+
+
+def test_does_not_fire_when_grounding_source_is_teaching() -> None:
+    cands = extract_discovery_candidates(
+        _TE(grounding_source="teaching"),
+        IntentTag.CAUSE,
+        "judgment",
+    )
+    assert cands == ()
+
+
+def test_does_not_fire_for_definition_intent() -> None:
+    cands = extract_discovery_candidates(
+        _TE(grounding_source="none"),
+        IntentTag.DEFINITION,
+        "judgment",
+    )
+    assert cands == ()
+
+
+def test_does_not_fire_for_unknown_intent() -> None:
+    cands = extract_discovery_candidates(
+        _TE(grounding_source="none"),
+        IntentTag.UNKNOWN,
+        "judgment",
+    )
+    assert cands == ()
+
+
+def test_does_not_fire_for_non_pack_lemma() -> None:
+    cands = extract_discovery_candidates(
+        _TE(grounding_source="none"),
+        IntentTag.CAUSE,
+        "zzznotalemma",
+    )
+    assert cands == ()
+
+
+def test_does_not_fire_when_chain_already_in_corpus() -> None:
+    """``(light, cause)`` IS in the corpus today (``cause_light_reveals_truth``)
+    — the would_have_grounded predicate must not fire."""
+    cands = extract_discovery_candidates(
+        _TE(grounding_source="none"),
+        IntentTag.CAUSE,
+        "light",
+    )
+    assert cands == ()
+
+
+def test_does_not_fire_on_empty_lemma() -> None:
+    cands = extract_discovery_candidates(
+        _TE(grounding_source="none"),
+        IntentTag.CAUSE,
+        "",
+    )
+    assert cands == ()
+
+
+def test_does_not_fire_on_none_lemma() -> None:
+    cands = extract_discovery_candidates(
+        _TE(grounding_source="none"),
+        IntentTag.CAUSE,
+        None,
+    )
+    assert cands == ()
+
+
+def test_candidate_id_is_deterministic() -> None:
+    """Identical inputs MUST produce the identical candidate_id —
+    this is the load-bearing replay property."""
+    a = extract_discovery_candidates(
+        _TE(grounding_source="none", trace_hash="seed-1"),
+        IntentTag.CAUSE,
+        "judgment",
+    )
+    b = extract_discovery_candidates(
+        _TE(grounding_source="none", trace_hash="seed-1"),
+        IntentTag.CAUSE,
+        "judgment",
+    )
+    assert a[0].candidate_id == b[0].candidate_id
+
+
+def test_candidate_id_changes_with_trace_hash() -> None:
+    a = extract_discovery_candidates(
+        _TE(grounding_source="none", trace_hash="seed-1"),
+        IntentTag.CAUSE,
+        "judgment",
+    )
+    b = extract_discovery_candidates(
+        _TE(grounding_source="none", trace_hash="seed-2"),
+        IntentTag.CAUSE,
+        "judgment",
+    )
+    assert a[0].candidate_id != b[0].candidate_id
+
+
+def test_boundary_clean_false_when_refusal_emitted() -> None:
+    cands = extract_discovery_candidates(
+        _TE(grounding_source="none", refusal_emitted=True),
+        IntentTag.CAUSE,
+        "judgment",
+    )
+    # The trigger still fires (refusal does not block evidence), but
+    # boundary_clean flips to False so reviewers can filter these out
+    # downstream.
+    assert len(cands) == 1
+    assert cands[0].boundary_clean is False
+
+
+def test_boundary_clean_false_when_hedge_injected() -> None:
+    cands = extract_discovery_candidates(
+        _TE(grounding_source="none", hedge_injected=True),
+        IntentTag.CAUSE,
+        "judgment",
+    )
+    assert len(cands) == 1
+    assert cands[0].boundary_clean is False
+
+
+# ---------------------------------------------------------------------------
+# format_candidate_jsonl — stable on-disk shape
+# ---------------------------------------------------------------------------
+
+
+def test_format_jsonl_is_sorted_keys_compact() -> None:
+    cand = DiscoveryCandidate(
+        candidate_id="x",
+        proposed_chain={"subject": "judgment", "intent": "cause", "connective": None, "object": None},
+        trigger="would_have_grounded",
+        source_turn_trace="t",
+        pack_consistent=True,
+        boundary_clean=True,
+    )
+    line = format_candidate_jsonl(cand)
+    parsed = json.loads(line)
+    assert parsed["candidate_id"] == "x"
+    # Compact separators — no whitespace between key/value or fields.
+    assert ", " not in line
+    assert ": " not in line
+
+
+# ---------------------------------------------------------------------------
+# Runtime integration — sink opt-in + behaviour parity
+# ---------------------------------------------------------------------------
+
+
+def _find_pack_lemma_without_active_chain() -> tuple[str, IntentTag]:
+    """Pick a (lemma, intent) that lives in the pack but not in the
+    corpus.  Used so the integration tests are not coupled to a
+    specific lemma whose status might change as the corpus grows."""
+    from chat.pack_grounding import _pack_index
+    from chat.teaching_grounding import _corpus_index
+
+    pack = _pack_index()
+    corpus = _corpus_index()
+    for lemma in pack:
+        if (lemma, "cause") not in corpus:
+            return lemma, IntentTag.CAUSE
+        if (lemma, "verification") not in corpus:
+            return lemma, IntentTag.VERIFICATION
+    pytest.skip("Every pack lemma already has a corpus chain — no would-have-grounded candidate possible")
+
+
+def test_runtime_no_sink_no_emission() -> None:
+    """No sink attached ⇒ no error, no side effect."""
+    rt = ChatRuntime()
+    rt.chat("Why does judgment matter?")
+    # Just must not crash and turn_log must have grown.
+    assert len(rt.turn_log) >= 1
+
+
+def test_runtime_emits_to_buffer_sink_on_would_have_grounded() -> None:
+    """Wire a buffer sink, send a CAUSE prompt on a lemma without a
+    corpus chain, expect one JSONL line."""
+    lemma, intent_tag = _find_pack_lemma_without_active_chain()
+    if intent_tag is IntentTag.CAUSE:
+        prompt = f"Why does {lemma} matter?"
+    else:
+        prompt = f"Does {lemma} require evidence?"
+
+    rt = ChatRuntime()
+    sink = DiscoveryBufferSink()
+    rt.attach_discovery_sink(sink)
+    rt.chat(prompt)
+
+    assert len(sink.lines) == 1
+    payload = json.loads(sink.lines[0])
+    assert payload["trigger"] == "would_have_grounded"
+    assert payload["proposed_chain"]["subject"] == lemma
+    assert payload["review_state"] == "unreviewed"
+
+
+def test_runtime_does_not_emit_when_turn_is_grounded() -> None:
+    """``light`` + CAUSE grounds via the teaching corpus today; no
+    candidate should be emitted on a grounded turn."""
+    rt = ChatRuntime()
+    sink = DiscoveryBufferSink()
+    rt.attach_discovery_sink(sink)
+    rt.chat("Why does light matter?")
+    assert sink.lines == []
+
+
+def test_runtime_detach_sink_stops_emission() -> None:
+    lemma, _ = _find_pack_lemma_without_active_chain()
+    rt = ChatRuntime()
+    sink = DiscoveryBufferSink()
+    rt.attach_discovery_sink(sink)
+    rt.attach_discovery_sink(None)
+    rt.chat(f"Why does {lemma} matter?")
+    assert sink.lines == []
+
+
+def test_runtime_emission_does_not_mutate_corpus_on_disk() -> None:
+    """Phase B's core trust-boundary claim: emission writes nothing
+    to the corpus."""
+    from chat.teaching_grounding import _CORPUS_PATH
+
+    lemma, _ = _find_pack_lemma_without_active_chain()
+    before = _CORPUS_PATH.read_bytes()
+    rt = ChatRuntime()
+    rt.attach_discovery_sink(DiscoveryBufferSink())
+    rt.chat(f"Why does {lemma} matter?")
+    after = _CORPUS_PATH.read_bytes()
+    assert before == after
+
+
+# ---------------------------------------------------------------------------
+# DiscoveryMonthlyFileSink — rollover semantics
+# ---------------------------------------------------------------------------
+
+
+def test_monthly_sink_writes_jsonl_in_expected_month_path(tmp_path: Path) -> None:
+    fixed = datetime(2026, 5, 18, tzinfo=timezone.utc)
+    sink = DiscoveryMonthlyFileSink(tmp_path, clock=lambda: fixed)
+    sink.emit('{"a":1}')
+    sink.close()
+    written = tmp_path / "2026" / "2026-05.jsonl"
+    assert written.exists()
+    assert written.read_text(encoding="utf-8").splitlines() == ['{"a":1}']
+
+
+def test_monthly_sink_rolls_over_on_month_change(tmp_path: Path) -> None:
+    fake = {"now": datetime(2026, 5, 31, tzinfo=timezone.utc)}
+    sink = DiscoveryMonthlyFileSink(tmp_path, clock=lambda: fake["now"])
+    sink.emit('{"a":1}')
+    fake["now"] = datetime(2026, 6, 1, tzinfo=timezone.utc)
+    sink.emit('{"b":2}')
+    sink.close()
+    may = (tmp_path / "2026" / "2026-05.jsonl").read_text(encoding="utf-8")
+    jun = (tmp_path / "2026" / "2026-06.jsonl").read_text(encoding="utf-8")
+    assert may.splitlines() == ['{"a":1}']
+    assert jun.splitlines() == ['{"b":2}']
+
+
+def test_monthly_sink_append_only_existing_lines_preserved(tmp_path: Path) -> None:
+    target = tmp_path / "2026" / "2026-05.jsonl"
+    target.parent.mkdir(parents=True)
+    target.write_text('{"x":0}\n', encoding="utf-8")
+    fixed = datetime(2026, 5, 18, tzinfo=timezone.utc)
+    sink = DiscoveryMonthlyFileSink(tmp_path, clock=lambda: fixed)
+    sink.emit('{"y":1}')
+    sink.close()
+    lines = target.read_text(encoding="utf-8").splitlines()
+    assert lines == ['{"x":0}', '{"y":1}']
+
+
+def test_monthly_sink_context_manager_closes(tmp_path: Path) -> None:
+    fixed = datetime(2026, 5, 18, tzinfo=timezone.utc)
+    with DiscoveryMonthlyFileSink(tmp_path, clock=lambda: fixed) as sink:
+        sink.emit('{"a":1}')
+    # After exit, internal file handle is closed.
+    assert sink._fh is None  # type: ignore[attr-defined]
+
+
+# ---------------------------------------------------------------------------
+# Doctrine — no corpus mutation, no runtime state mutation
+# ---------------------------------------------------------------------------
+
+
+def test_pure_extract_does_not_load_or_mutate_runtime_state() -> None:
+    """Calling the extractor repeatedly must not change pack or
+    corpus internal state — a smoke test that the extractor is
+    a pure read."""
+    from chat.pack_grounding import _pack_index
+    from chat.teaching_grounding import _corpus_index
+
+    pack_before = dict(_pack_index())
+    corpus_before = dict(_corpus_index())
+    for _ in range(5):
+        extract_discovery_candidates(_TE(), IntentTag.CAUSE, "judgment")
+    pack_after = dict(_pack_index())
+    corpus_after = dict(_corpus_index())
+    assert pack_before.keys() == pack_after.keys()
+    assert corpus_before.keys() == corpus_after.keys()