diff --git a/chat/runtime.py b/chat/runtime.py index 44f83d2a..08f4b270 100644 --- a/chat/runtime.py +++ b/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) diff --git a/docs/decisions/ADR-0055-inter-session-memory-discovery-promotion.md b/docs/decisions/ADR-0055-inter-session-memory-discovery-promotion.md index e69c88e0..a7b40a37 100644 --- a/docs/decisions/ADR-0055-inter-session-memory-discovery-promotion.md +++ b/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 diff --git a/teaching/discovery.py b/teaching/discovery.py new file mode 100644 index 00000000..d3b633a4 --- /dev/null +++ b/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", +] diff --git a/teaching/discovery_sink.py b/teaching/discovery_sink.py new file mode 100644 index 00000000..778b8faf --- /dev/null +++ b/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 ``//.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 + ``//.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", +] diff --git a/tests/test_discovery_candidates.py b/tests/test_discovery_candidates.py new file mode 100644 index 00000000..3277616c --- /dev/null +++ b/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()