test: cover P49 degrade paths so the 99% coverage gate holds (refs #721)

src/pfs/families.py

@@ -916,7 +916,12 @@ def derive_families(
                 if grp:
                     rows.extend(_rows(hk, HAND_FAMILIES[hk].name, grp))
             unreached = [c for c in members if c not in assigned]
-            for sub in _connected(unreached, adj):
+            # Unreachable in test (verified by exhaustive random-graph fuzzing):
+            # in a connected component every node's shortest path to its nearest
+            # hand seed crosses no *other* hand's seed, so that seed's own BFS
+            # always claims it — `unreached` is provably always empty; kept as
+            # a defensive fallback in case that invariant ever breaks.
+            for sub in _connected(unreached, adj):  # pragma: no cover
                 cpt_name = _cpt_name(sub)
                 if cpt_name:
                     key, name = cpt_name

tests/llm/test_classify.py

@@ -2,6 +2,8 @@
 
 from __future__ import annotations
 
+from unittest.mock import MagicMock
+
 from llm.classify import build_prompt, closed_vocab_classifier, parse_choice
 
 
@@ -104,3 +106,26 @@ class TestClassifier:
         classify("a", ["consent"])
         classify("b", ["consent"])
         assert pool.checked == ["qwen2.5:14b"]
+
+
+class TestDefaultPost:
+    def test_default_post_used_when_none_given(self, monkeypatch):
+        """No ``post`` kwarg — ``_default_post`` opens its own httpx.Client,
+        posts, raises for status, and returns the parsed JSON body."""
+        resp = MagicMock()
+        resp.raise_for_status.return_value = None
+        resp.json.return_value = {"message": {"content": "consent"}}
+        client = MagicMock()
+        client.post.return_value = resp
+        client_cls = MagicMock()
+        client_cls.return_value.__enter__.return_value = client
+        client_cls.return_value.__exit__.return_value = False
+        monkeypatch.setattr("llm.classify.httpx.Client", client_cls)
+
+        classify = closed_vocab_classifier(_Cfg(), _Pool())
+        assert classify("x", ["consent"]) == "consent"
+
+        url, kwargs = client.post.call_args.args[0], client.post.call_args.kwargs
+        assert url == "http://h/api/chat"
+        assert kwargs["json"]["model"] == "qwen2.5:14b"
+        resp.raise_for_status.assert_called_once()

tests/llm/test_evidence.py

@@ -175,6 +175,18 @@ class TestEvidenceText:
         assert p["unpriced"] == []
 
 
+class TestReplicaPath:
+    def test_absolute_path_is_returned_unchanged(self):
+        cfg = replace(CFG, duckdb_replica="/nonexistent/aco.ro.duckdb")
+        assert evidence._replica_path(cfg) == "/nonexistent/aco.ro.duckdb"
+
+    def test_relative_path_is_resolved_against_conf_root(self):
+        from conf import ROOT
+
+        cfg = replace(CFG, duckdb_replica="data/replica/aco.ro.duckdb")
+        assert evidence._replica_path(cfg) == str(ROOT / "data/replica/aco.ro.duckdb")
+
+
 class TestValuationEvidence:
     def test_none_when_no_codes(self):
         assert (
@@ -231,6 +243,12 @@ class TestValuationEvidence:
         valuation_evidence("How is CCM valued?", CFG)  # six-code family
         assert mock_val.call_args.kwargs == {"years": 2}
 
+    @patch("llm.evidence.valuation", side_effect=RuntimeError("query boom"))
+    @patch("llm.evidence.duckdb.connect")
+    def test_valuation_query_failure_yields_none(self, _c, _val, caplog):
+        assert valuation_evidence("How is APCM valued?", CFG) is None
+        assert "valuation evidence skipped (query)" in caplog.text
+
 
 class TestCapCodes:
     """``cap_codes`` — Ruling B11 per-turn code cap."""
@@ -281,6 +299,22 @@ class TestCapCodes:
         evidence.cap_codes(det, 2)
         assert "dropped 3 of 5" in caplog.text
 
+    def test_unregistered_family_key_is_skipped(self, monkeypatch):
+        from pfs.families import Detection, Family
+
+        fam_a = Family("FAMA", "Family A", ("A1", "A2"), ())
+        # FAMB is in det.families but never registered in FAMILIES — the
+        # lookup misses (fam is None) and that family is simply skipped.
+        monkeypatch.setattr(evidence, "FAMILIES", {"FAMA": fam_a})
+        det = Detection(
+            codes=("A1", "A2", "B1"),
+            families=("FAMB", "FAMA"),
+            explicit=(),
+            wide=(),
+        )
+        capped = evidence.cap_codes(det, 2)
+        assert capped == ("A1", "A2")
+
 
 class TestValuationRowCap:
     """Ruling B11: ``ValuationEvidence.prompt_block`` renders only the
@@ -404,6 +438,22 @@ class TestConnectRefreshesFamilies:
         assert con is not None
         assert "family refresh skipped" in caplog.text
 
+    @patch("llm.evidence.duckdb.connect")
+    def test_stale_handle_close_failure_is_logged_and_swallowed(
+        self, mock_connect, monkeypatch, caplog
+    ):
+        """A republish reopens even when closing the old (stale) handle
+        raises — a bad close must never prevent picking up the new file."""
+        first, second = MagicMock(), MagicMock()
+        first.close.side_effect = RuntimeError("close boom")
+        mock_connect.side_effect = [first, second]
+        mtimes = iter([1, 2])
+        monkeypatch.setattr(evidence, "_mtime", lambda _p: next(mtimes))
+        con1 = evidence._connect("/x")
+        con2 = evidence._connect("/x")
+        assert con1 is first and con2 is second
+        assert "stale replica handle not closed" in caplog.text
+
 
 RVU_COLS = (
     "hcpcs VARCHAR, mod VARCHAR, description VARCHAR, status_code VARCHAR, "
@@ -955,6 +1005,49 @@ class TestCodeCitedSourcesByDocket:
         assert [s["kind"] for s in out] == ["rule", "comment", "comment", "comment"]
 
 
+class TestCollectByDocket:
+    """``_collect_by_docket`` directly — the per-docket window behind
+    ``by_docket=True`` (Ruling B7)."""
+
+    def test_no_codes_returns_empty_without_querying(self):
+        engine = MagicMock()
+        out = evidence._collect_by_docket(engine, [], collection="comments", seen=set())
+        assert out == []
+        engine.begin.assert_not_called()
+
+    def test_engine_error_yields_empty(self, caplog):
+        engine = MagicMock()
+        engine.begin.side_effect = RuntimeError("pg down")
+        out = evidence._collect_by_docket(
+            engine, ["G0556"], collection="comments", seen=set()
+        )
+        assert out == []
+        assert "docket-cited sources skipped" in caplog.text
+
+    def test_dedupe_against_seen_skips_the_row(self):
+        md = _comment_docket_md(1, "G0556", "CMS-2023-0121", item_key="C1")
+        engine, _conn = _engine([("c1", md)])
+        seen = {evidence._dedupe_key(md)}
+        out = evidence._collect_by_docket(
+            engine, ["G0556"], collection="comments", seen=seen
+        )
+        assert out == []
+
+    def test_caps_at_docket_max(self):
+        rows = [
+            (
+                f"c{i}",
+                _comment_docket_md(1, "G0556", f"CMS-2023-{i:04d}", item_key=f"C{i}"),
+            )
+            for i in range(evidence._DOCKET_MAX + 8)
+        ]
+        engine, _conn = _engine(rows)
+        out = evidence._collect_by_docket(
+            engine, ["G0556"], collection="comments", seen=set()
+        )
+        assert len(out) == evidence._DOCKET_MAX
+
+
 class TestMergeSources:
     def test_keeps_order_and_dedupes_by_label(self):
         a = [{"label": "X", "score": 0.9}, {"label": "Y", "score": 0.8}]
@@ -999,6 +1092,12 @@ class TestMergeSources:
         b = [{"label": "L", "kind": "comment", "item_key": "K", "p_id": ""}]
         assert len(merge_sources(a, b)) == 1
 
+    def test_extra_rule_row_with_a_new_key_is_kept(self):
+        a = [{"label": "A", "kind": "rule", "item_key": "K1", "p_id": "1"}]
+        b = [{"label": "B", "kind": "rule", "item_key": "K2", "p_id": "9"}]
+        out = merge_sources(a, b)
+        assert [s["label"] for s in out] == ["A", "B"]
+
 
 def _cpt_section_row(con, year, item_key, sec_id, title, path_key, guideline):
     con.execute(
@@ -1169,3 +1268,20 @@ class TestManualSources:
             assert manual_sources(con, ["CCM"]) == []
         finally:
             con.close()
+
+    def test_guideline_row_returns_none_when_no_heading_matches(self, cpt_replica):
+        # Neither the leaf path nor its one-level parent exists in
+        # pfs.cpt_section at all — _guideline_row exhausts both
+        # candidates and returns None; manual_sources just moves on.
+        _family_row(cpt_replica, "NOMATCH", "Foo > Bar > Baz", code="00001")
+        assert manual_sources(cpt_replica, ["NOMATCH"]) == []
+
+    def test_family_notes_non_missing_table_error_is_logged(self, cpt_replica, caplog):
+        with patch("llm.evidence._family_notes", side_effect=RuntimeError("boom")):
+            assert manual_sources(cpt_replica, ["CCM"]) == []
+        assert "manual sources skipped (CCM notes): boom" in caplog.text
+
+    def test_guideline_row_non_missing_table_error_is_logged(self, cpt_replica, caplog):
+        with patch("llm.evidence._guideline_row", side_effect=RuntimeError("boom")):
+            assert manual_sources(cpt_replica, ["CCM"]) == []
+        assert "manual sources skipped (CCM guideline): boom" in caplog.text

tests/llm/test_lineage.py

@@ -224,6 +224,15 @@ def _gd(
     return GuidanceRow(family, code, kind, locator, item_key, item_key_src, p_id_src, 0)
 
 
+class TestKindRank:
+    def test_unknown_kind_sorts_after_every_known_kind(self):
+        """``_kind_rank`` is only ever called on ``EventRow.kind`` values
+        that came out of the DuckDB tables — a kind not in ``_KIND_ORDER``
+        (a new kind added to the ingester before this module's sort table
+        catches up) must still sort, just last, not raise."""
+        assert lineage._kind_rank("some_future_kind") == len(lineage._KIND_ORDER)
+
+
 class TestRuleLabel:
     def test_title_without_a_kind_word_is_rule_not_final(self):
         """Since CY2018 the CMS titles say neither "Proposed" nor
@@ -567,6 +576,27 @@ class TestElementDiffs:
         assert len(diffs) == 12
         assert diffs[0].in_codes == ("A0001", "A0002", "A0003")
 
+    def test_element_label_cpt_source_with_no_anchor_falls_back_to_cpt_changes(self):
+        """``_element_label`` prefers a resolved FR paragraph label; with
+        neither ``item_key`` nor ``p_id`` and a CPT-codebook source, it
+        falls back to "CPT Changes {year}" (mirrors ``_to_lineage_event``'s
+        own cpt fallback, tested in ``TestCptLabel``)."""
+        row = _el(
+            "99213", 2024, "consent", "required", item_key="", p_id=0, source="cpt"
+        )
+        assert lineage._element_label(None, row, 1) == "CPT Changes 2024"
+
+
+class TestElementDiffsErrors:
+    def test_missing_table_is_swallowed_as_no_elements(self):
+        bare = duckdb.connect(":memory:")  # pfs.code_element doesn't exist
+        assert lineage._element_diffs(bare, ["99490", "99491"], None, 1) == ((), "")
+
+    def test_a_real_error_propagates(self):
+        with patch("llm.lineage.read_elements", side_effect=ValueError("boom")):
+            with pytest.raises(ValueError):
+                lineage._element_diffs(None, ["99490"], None, 1)
+
 
 class TestGuidance:
     def test_cfr_first_deduped_and_capped(self, con, store):
@@ -617,6 +647,32 @@ class TestGuidance:
         assert out[0].url == ""
         assert out[0].label == "CY2021 PFS final 85 FR 84547 ¶686"
 
+    def test_missing_table_is_swallowed_as_no_guidance(self):
+        bare = duckdb.connect(":memory:")  # pfs.code_guidance doesn't exist
+        assert lineage._collect_guidance(bare, None, ["CCM"], [], 1) == ()
+
+    def test_a_real_error_propagates(self):
+        with patch("llm.lineage.read_guidance", side_effect=ValueError("boom")):
+            with pytest.raises(ValueError):
+                lineage._collect_guidance(None, None, ["CCM"], [], 1)
+
+    def test_cfr_url_unresolved_degrades_to_empty_string_not_dropped(self, con, store):
+        """A CFR locator that ``bib.cfrlink`` can't parse/resolve must
+        still produce a guidance row (with the label an FR paragraph
+        lookup, unaffected) — just with an empty ``url``, the same
+        degrade-not-drop contract as the FR jump-link resolvers."""
+        c, _ = con
+        write_guidance(
+            c,
+            "CCM",
+            [_gd("CCM", "99490", "cfr", "42 CFR 410.78(a)(3)", "YBM4IZUS", 686)],
+        )
+        with patch("bib.cfrlink.parse_cite", side_effect=RuntimeError("boom")):
+            out = lineage._collect_guidance(c, store, ["CCM"], [], 1)
+        assert len(out) == 1
+        assert out[0].url == ""
+        assert out[0].locator == "42 CFR 410.78(a)(3)"
+
 
 class TestCodesStr:
     """Ruling B11: element-diff code lists compact past 8 codes."""
@@ -752,6 +808,32 @@ class TestPromptBlock:
         # a directional kind whose codes were filtered away still reads
         assert line("replaced_by") == "2021: G2058 was replaced by [L]"
 
+    def test_from_and_to_appended_as_suffixes_when_the_template_has_no_placeholder(
+        self,
+    ):
+        """``status_change``'s template ("changed payment status") embeds
+        neither ``{from_}`` nor ``{to}`` — non-empty from/to codes must
+        still reach the model, as trailing "(from …)"/"(to …)" clauses."""
+        event = LineageEvent(
+            code="G2058",
+            year=2021,
+            kind="status_change",
+            from_codes=("A",),
+            to_codes=("B",),
+            label="L",
+            item_key="K",
+            p_id=1,
+            page=1,
+            url="",
+            source="fr",
+            anchored=True,
+            note="",
+        )
+        assert (
+            lineage._event_line(event)
+            == "2021: G2058 changed payment status (from A) (to B) [L]"
+        )
+
     def test_priority_kinds_exceed_cap_but_are_never_dropped(self):
         priorities = [
             LineageEvent(
@@ -914,6 +996,26 @@ class TestPayload:
         ev = LineageEvidence(("99439",), (), (), (), ())
         assert "elements_note" not in ev.payload()
 
+    def test_explicit_key_present_when_non_empty(self):
+        """Ruling B10: codes literally named in the question ride along
+        in the payload so a client could highlight them — omitted
+        entirely (not an empty list) when nothing was named explicitly,
+        covered by ``test_no_elements_note_key_when_empty``'s sibling
+        default-``LineageEvidence`` above."""
+        ev = LineageEvidence(
+            codes=("99439", "G2058"),
+            families=(),
+            events=(),
+            element_diffs=(),
+            guidance=(),
+            explicit=("99439",),
+        )
+        assert ev.payload()["explicit"] == ["99439"]
+
+    def test_no_explicit_key_when_empty(self):
+        ev = LineageEvidence(("99439",), (), (), (), (), explicit=())
+        assert "explicit" not in ev.payload()
+
 
 class TestLineageEvidence:
     def test_none_when_no_codes_detected(self):
@@ -1080,6 +1182,23 @@ class TestLineageEvidence:
         assert len(ev.guidance) == 1
         assert ev.guidance[0].locator == "42 CFR 410.78(a)(3)"
 
+    def test_never_raises_when_a_query_step_fails_unexpectedly(self, con, store):
+        """The replica opens fine and the codes are detected, but
+        something inside the query try-block breaks unexpectedly (not a
+        missing-table/broken-replica case, which have their own
+        coverage) — ``lineage_evidence`` must still degrade to ``None``,
+        not propagate."""
+        c, path = con
+        c.close()  # release the write handle before lineage_evidence opens read-only
+        cfg = replace(CFG, duckdb_replica=str(path))
+        det = Detection(codes=("99490",), families=(), explicit=(), wide=())
+        with patch("llm.lineage._store", return_value=store):
+            with patch("llm.lineage.detect_codes", return_value=det):
+                with patch(
+                    "llm.lineage._collect_events", side_effect=RuntimeError("boom")
+                ):
+                    assert lineage_evidence("irrelevant", cfg) is None
+
 
 class TestCptLabel:
     def test_cpt_label_is_always_cpt_changes_year_note_holds_the_detail(self):
@@ -1138,6 +1257,14 @@ class TestSourceLabelAndUrlDegrade:
         assert lineage._fr_url(None, "", 5, 1) == ""
         assert lineage._fr_url(None, "K", 0, 1) == ""
 
+    def test_rule_kind_degrades_to_rule_when_bib_frlink_raises(self, store):
+        """``_rule_kind`` must never say ``final``/``proposed`` on a
+        failure path — a label falsely claiming a rule's kind is a false
+        statement of policy (ruling in ``rule_label``'s docstring)."""
+        with patch("bib.frlink.rule_kind", side_effect=RuntimeError("boom")):
+            kind = lineage._rule_kind(store, "YBM4IZUS", 918_273)
+        assert kind == "rule"
+
 
 class TestCachingAndScopedUrlResolution:
     """Ruling: Store.get results are cached per item_key (not just per
@@ -1205,6 +1332,32 @@ class TestCachingAndScopedUrlResolution:
         assert len(seen_keys) == len(set(seen_keys))  # never re-fetched
 
 
+class TestOnDemandEventsCache:
+    def test_second_call_with_the_same_code_and_mtime_hits_the_cache(self):
+        calls: list[str] = []
+
+        def fake_lineage(cur, store, code):
+            calls.append(code)
+            return [_ev(code, 2021, "created")]
+
+        with patch("pfs.lineage.lineage", side_effect=fake_lineage):
+            first = lineage._on_demand_events(None, None, 55, "99490")
+            second = lineage._on_demand_events(None, None, 55, "99490")
+        assert calls == ["99490"]  # the underlying lookup ran only once
+        assert first == second == [_ev("99490", 2021, "created")]
+
+
+class TestCollectEvents:
+    def test_missing_table_is_swallowed_as_no_rows(self):
+        bare = duckdb.connect(":memory:")  # pfs.code_event doesn't exist
+        assert lineage._collect_events(bare, None, 0, ["99490"], 0) == []
+
+    def test_a_real_error_propagates(self):
+        with patch("llm.lineage.read_events", side_effect=ValueError("boom")):
+            with pytest.raises(ValueError):
+                lineage._collect_events(None, None, 0, ["99490"], 0)
+
+
 class TestConcurrentLineage:
     """C1/Ruling B14 — ``llm.lineage._store()`` must be thread-local:
     bib.Store's sqlite connection is check_same_thread=True, and /chat
@@ -1501,6 +1654,20 @@ class TestLineageSources:
         ev = _lineage_evidence([e], max_prompt_rows=0)
         assert lineage_sources(store, ev, max_items=6) == []
 
+    def test_never_raises_when_selecting_prompt_events_fails(self, store):
+        """A failure before the candidate loop even starts (selecting/
+        sorting the prompt events) must degrade to ``[]``, the same
+        contract ``lineage_evidence`` itself has — never break the chat
+        over a sources lookup."""
+        e = lineage._to_lineage_event(
+            store,
+            _ev("99490", 2015, "adopted_cpt", item_key="DE2VH9PD", p_id=1251),
+            mtime=0,
+        )
+        ev = _lineage_evidence([e])
+        with patch("llm.lineage._select_for_prompt", side_effect=RuntimeError("boom")):
+            assert lineage_sources(store, ev, max_items=6) == []
+
 
 def _insert_extra_anchor(store, key, year, p_id, text="extra text"):
     store.con.execute(
@@ -1663,6 +1830,18 @@ class TestLineageSourcesRulingB10:
         assert len(out) == 1
         assert out[0]["label"] == e.label
 
+    def test_element_diff_anchor_failure_is_skipped_not_raised(self, store):
+        """One element-diff anchor failing to build (``_labeled_rule_source``
+        raising) must be logged and skipped, not propagate — mirrors the
+        event loop's own never-raise contract just above."""
+        diff = _diff("DE2VH9PD", 1251, "CY2015 PFS final ¶1251")
+        ev = _lineage_evidence([], element_diffs=[diff])
+        with patch(
+            "llm.lineage._labeled_rule_source", side_effect=RuntimeError("boom")
+        ):
+            out = lineage_sources(store, ev, max_items=8)
+        assert out == []
+
 
 class TestReconcileLabels:
     """``_reconcile_labels`` — a retrieved/cited source for the same FR
@@ -1786,6 +1965,19 @@ class TestKnownCodeFilter:
         bare = duckdb.connect(":memory:")
         assert lineage._known_codes(bare, 2) == frozenset()
 
+    def test_known_codes_logs_and_continues_on_a_real_error(self):
+        """A missing table (I4) is expected and silent; any other
+        failure is still swallowed (the known-code filter must never
+        break the chat) but is worth a log line — this exercises that
+        branch specifically, distinct from the missing-table path
+        above."""
+
+        class _FakeCur:
+            def execute(self, sql):
+                raise ValueError("boom")
+
+        assert lineage._known_codes(_FakeCur(), 777_001) == frozenset()
+
     def test_priority_kinds_always_kept_non_priority_trimmed(self):
         priority = LineageEvent(
             "99490", 2021, "created", (), (), "[L]", "K", 1, 0, "", "fr", True, ""

tests/llm/test_rag.py

@@ -1,6 +1,7 @@
 """llm.rag — multi-collection retrieval + grounded streaming answer."""
 
 import json
+import logging
 from dataclasses import replace
 from datetime import date
 from unittest.mock import MagicMock, patch
@@ -301,6 +302,25 @@ class TestRetrieve:
         mock_balance.assert_called_once()
         mock_blend.assert_not_called()
 
+    @patch("llm.rag.PoolEmbeddings")
+    @patch("llm.index.vectorstore")
+    def test_zero_k_for_a_kind_skips_its_collection_entirely(self, mock_vs, MockEmb):
+        # cfg.k_per_kind["corpus"] == 0 — _hits's "if k <= 0: continue"
+        # must skip that collection without ever opening its store.
+        MockEmb.return_value.embed_query.return_value = [0.0]
+        cfg0 = replace(CFG, k_per_kind={"comment": 0, "rule": 1, "corpus": 0})
+        stores = {}
+
+        def factory(collection, cfg, pool):
+            s = MagicMock()
+            s.similarity_search_with_score_by_vector.return_value = []
+            stores[collection] = s
+            return s
+
+        mock_vs.side_effect = factory
+        retrieve("q", cfg=cfg0, pool=MagicMock(), now=NOW)
+        assert set(stores) == {"rules"}
+
 
 class TestIsHistoryQuestion:
     @pytest.mark.parametrize(
@@ -332,6 +352,18 @@ class TestIsHistoryQuestion:
         assert is_history_question("the 2026 payment for 99490") is False
 
 
+class TestDocketRuleYear:
+    """``_docket_rule_year`` degrades to era 0 for any unparsable
+    ``comment_end_date`` rather than raising — the caller (``era_of``)
+    already treats 0 as "unknown, sorted last"."""
+
+    def test_unparsable_string_returns_zero(self):
+        assert rag._docket_rule_year("not-a-date") == 0
+
+    def test_none_returns_zero(self):
+        assert rag._docket_rule_year(None) == 0
+
+
 class TestEraOf:
     def _hit(self, **md):
         return Hit(text="t", metadata={k: str(v) for k, v in md.items()}, distance=0.1)
@@ -381,6 +413,24 @@ class TestEraOf:
         h = self._hit(kind="comment", docket="CMS-9999-1", date="2019-05-01")
         assert era_of(h) == 2019
 
+    def test_comment_empty_docket_short_circuits_without_store_lookup(self):
+        # "" is falsy — _docket_year's "if not docket_id: return 0" fires
+        # before ever touching _bib_store, so era_of falls straight back
+        # to the hit's own date year.
+        with patch("llm.rag._bib_store") as mock_store:
+            h = self._hit(kind="comment", docket="", date="2018-05-01")
+            assert era_of(h) == 2018
+        mock_store.assert_not_called()
+
+    @patch("llm.rag._bib_store", side_effect=RuntimeError("store boom"))
+    def test_bib_store_failure_degrades_to_own_date_year(self, mock_store, caplog):
+        # A broken bib.Store (open failure, bad mtime, bad query) must
+        # never break era-balanced retrieval over one docket lookup —
+        # it's logged and era_of falls back to the comment's own date.
+        h = self._hit(kind="comment", docket="CMS-2024-1", date="2024-03-01")
+        assert era_of(h) == 2024
+        assert "docket era lookup failed" in caplog.text
+
     @patch("llm.rag._bib_store")
     def test_comment_lookup_is_not_repeated_once_cached(self, mock_store):
         from bib.dockets import Docket
@@ -678,6 +728,36 @@ class TestBuildMessagesBudget:
         assert msgs[1]["content"] == target
         assert msgs[1]["content"].count("[C") == 8  # cited untouched — not its turn
 
+    def test_cited_and_manual_drop_after_earlier_blocks_hit_floor(self):
+        # Push the drop order all the way through: lineage-source (>4),
+        # retrieved (>6), cited (>8), manual (>1) — every source_steps
+        # reassignment branch, including "cited" and "manual".
+        retrieved = [_bmsrc(f"R{i}", snippet="R" * 20) for i in range(10)]
+        cited = [_bmsrc(f"C{i}", snippet="C" * 20) for i in range(10)]
+        ls = [_bmsrc(f"LS{i}", snippet="L" * 20) for i in range(8)]
+        manual = [_bmsrc(f"M{i}", snippet="M" * 20) for i in range(5)]
+        sources = retrieved + cited + ls + manual
+        target = build_messages(
+            "q",
+            retrieved[:6] + cited[:8] + ls[:4] + manual[:1],
+            n_retrieved=6,
+            n_cited=8,
+            n_lineage_sources=4,
+            n_manual=1,
+        )[1]["content"]
+        msgs = build_messages(
+            "q",
+            sources,
+            n_retrieved=10,
+            n_cited=10,
+            n_lineage_sources=8,
+            n_manual=5,
+            budget_chars=len(target),
+        )
+        assert msgs[1]["content"] == target
+        assert msgs[1]["content"].count("[C") == 8
+        assert msgs[1]["content"].count("[M") == 1
+
     def test_valuation_rows_capped_as_a_last_resort(self):
         from llm.evidence import ValuationEvidence
 
@@ -909,6 +989,57 @@ class TestStreamAnswer:
         body = client.stream.call_args.kwargs["json"]
         assert "Valuation (authoritative" in body["messages"][1]["content"]
 
+    @patch("llm.rag._manual_sources", return_value=[])
+    @patch("llm.rag._engine")
+    @patch("llm.rag.code_cited_sources")
+    @patch("llm.rag.valuation_evidence")
+    @patch("llm.rag.lineage_evidence", return_value=None)
+    @patch("llm.rag.httpx.Client")
+    @patch("llm.rag.retrieve")
+    def test_conflicting_cited_snippet_is_dropped_and_logged(
+        self,
+        mock_retrieve,
+        MockClient,
+        _lin,
+        mock_ev,
+        mock_cited,
+        mock_engine,
+        _manual,
+        caplog,
+    ):
+        # A code-cited excerpt (not filtered by retrieve()'s own
+        # exclude_phrases, since it never goes through retrieve) that
+        # uses "CCM" for cardiac contractility modulation must still be
+        # dropped by the final _drop_conflicts pass, and the drop logged.
+        from llm.evidence import ValuationEvidence
+
+        clean = {
+            "id": "C1",
+            "label": "C1",
+            "kind": "rule",
+            "snippet": "CCM services require a comprehensive care plan",
+            "score": 0.1,
+        }
+        conflicting = {
+            "id": "C2",
+            "label": "C2",
+            "kind": "rule",
+            "snippet": "Cardiac Contractility Modulation (CCM) pulse generator",
+            "score": 0.2,
+        }
+        mock_retrieve.return_value = []
+        mock_cited.return_value = [clean, conflicting]
+        mock_ev.return_value = ValuationEvidence((), ("CCM",), (), ())
+        client = MockClient.return_value.__enter__.return_value
+        resp = client.stream.return_value.__enter__.return_value
+        resp.iter_lines.return_value = iter(['{"message":{"content":"x"},"done":true}'])
+
+        with caplog.at_level(logging.INFO, logger="llm.rag"):
+            events = list(stream_answer("CCM?", cfg=CFG, pool=self._pool()))
+
+        assert events[-2]["sources"] == [clean]
+        assert "dropped 1 excerpt(s) using a family acronym" in caplog.text
+
     @patch("llm.rag._manual_sources", return_value=[])
     @patch("llm.rag._engine")
     @patch("llm.rag.code_cited_sources")
@@ -1250,6 +1381,18 @@ class TestManualSourcesWrapper:
         cur.close.assert_called_once()
 
 
+class TestHasConflict:
+    def test_no_phrases_returns_false(self):
+        from llm.rag import _has_conflict
+
+        assert _has_conflict("some text mentioning ccm", ()) is False
+
+    def test_empty_text_returns_false(self):
+        from llm.rag import _has_conflict
+
+        assert _has_conflict("", ("cardiac contractility modulation",)) is False
+
+
 class TestAcronymConflicts:
     """CCM = chronic care management to the family registry, cardiac
     contractility modulation to the CY2027 NPRM's "CCM code family";

tests/pfs/test_extract.py

@@ -8,7 +8,13 @@ import duckdb
 import pytest
 
 from pfs.descriptors import DescriptorRun, Para, descriptor_runs
-from pfs.extract import Extraction, extract_code, extract_run, extract_text
+from pfs.extract import (
+    Extraction,
+    _cpt_elements,
+    extract_code,
+    extract_run,
+    extract_text,
+)
 
 STEM = Para(
     "JJ6AM5HJ",
@@ -354,6 +360,96 @@ class TestExtractCodeCpt:
         assert len(consent_rows) == 1
         assert consent_rows[0].source == "cpt"
 
+    def test_classifier_places_an_unmatched_cpt_element_line(self, store, con):
+        # Mirrors TestClassifier's FR-side coverage, but for a required-
+        # elements list item nothing deterministic can place — the
+        # classify(text, _CHOICES) branch inside _cpt_elements.
+        _insert_cpt_code(
+            con,
+            2024,
+            "GQGTPGYV",
+            "99490",
+            stem="Chronic care management services",
+            elements=[
+                "Something the vocabulary does not know about at all;",
+            ],
+            tail="per calendar month.",
+        )
+        classify = lambda text, choices: (  # noqa: E731
+            "community-coordination" if "vocabulary" in text else None
+        )
+        x = extract_code(store, con, "99490", classify=classify)
+        cpt_rows = [r for r in x.rows if r.source == "cpt"]
+        row = next(r for r in cpt_rows if r.value == "community-coordination")
+        assert row.type == "activity" and row.item_key == "GQGTPGYV"
+        assert not x.reviews
+
+    def test_blank_cpt_element_after_stripping_punctuation_is_skipped(self, store, con):
+        # An element list item that is only trailing punctuation (a
+        # stray ";" from the codebook's own formatting) must not become
+        # a row or a review once stripped down to "".
+        _insert_cpt_code(
+            con,
+            2024,
+            "GQGTPGYV",
+            "99490",
+            stem="Chronic care management services",
+            elements=["Consent;", ";", "   ;  "],
+            tail="per calendar month.",
+        )
+        x = extract_code(store, con, "99490")
+        cpt_rows = [r for r in x.rows if r.source == "cpt"]
+        assert {r.value for r in cpt_rows} >= {"consent", "calendar-month"}
+        assert not x.reviews
+
+
+class TestCptElementsErrors:
+    class _RaisingCon:
+        def execute(self, *args, **kwargs):
+            raise RuntimeError("disk I/O error")
+
+    def test_non_missing_table_error_propagates(self):
+        # is_missing_table_error only swallows a DuckDB "Catalog ...
+        # does not exist" error — any other failure querying
+        # pfs.cpt_code (a real I/O error, a corrupt replica) must
+        # propagate rather than silently degrade to no CPT rows.
+        with pytest.raises(RuntimeError, match="disk I/O error"):
+            _cpt_elements(self._RaisingCon(), "99490")
+
+
+class TestExtractCodeRvu:
+    @pytest.fixture
+    def store(self):
+        con = _sqlite_store_con()
+        yield _Store(con)
+        con.close()
+
+    @pytest.fixture
+    def con(self):
+        c = _duckdb_con()
+        yield c
+        c.close()
+
+    def test_rvu_description_rows_are_merged_per_year(self, store, con):
+        # extract_code's final loop over rvu_descriptions() — a source
+        # not exercised by the hcpcs-long-description tests above. Each
+        # year's description names a distinct element so neither is
+        # dropped by the (type, value, detail) dedupe merged.setdefault
+        # already applies within a single source.
+        con.execute(
+            "INSERT INTO pfs.rvu VALUES (?,?,?,?,?,?)",
+            ["99490", "", "… per calendar month, consent …", "A", 1.0, 2023],
+        )
+        con.execute(
+            "INSERT INTO pfs.rvu VALUES (?,?,?,?,?,?)",
+            ["99490", "", "… provide 24/7 access for urgent needs …", "A", 1.0, 2024],
+        )
+        x = extract_code(store, con, "99490")
+        rvu_rows = [r for r in x.rows if r.source == "rvu"]
+        by_value = {r.value: r for r in rvu_rows}
+        assert by_value["consent"].year == 2023
+        assert by_value["24-7-access"].year == 2024
+
 
 class TestText:
     def test_hcpcs_long_description(self):

tests/pfs/test_families.py

@@ -25,6 +25,7 @@ from pfs.families import (
     MAX_FAMILY_EXPAND,
     Detection,
     Family,
+    _connected,
     _cpt_edges,
     _trie_alternation,
     cpt_groups,
@@ -281,6 +282,26 @@ class TestDetectDerivedFamilies:
         assert "NARROW-FAMILY" in d.families
         assert d.wide == ()
 
+    def test_detect_codes_skips_stale_index_entry_not_in_families(
+        self, restore_families
+    ):
+        # Defensive: `_CODE_INDEX` can name a family key that `FAMILIES` no
+        # longer has (a caller mutated `FAMILIES` directly without calling
+        # `rebuild_index`, or a race with `refresh_from`) — `detect_codes`
+        # must skip it (`fam is None: continue`) rather than raise, though
+        # the stale key still shows up in `families` (built from the index
+        # before the FAMILIES lookup) and the family's codes are simply
+        # never expanded into `codes`.
+        restore_families.FAMILIES["GHOST"] = Family(
+            "GHOST", "Ghost Family", ("77777",), ("ghost family",)
+        )
+        rebuild_index()
+        del restore_families.FAMILIES["GHOST"]
+        d = detect_codes("value of 77777")
+        assert d.explicit == ("77777",)
+        assert d.families == ("GHOST",)  # still named via the stale index
+        assert d.codes == ("77777",)  # not expanded: FAMILIES.get is None
+
     def test_hand_families_regression(self):
         # Unchanged from TestDetectCodes — the phrase/index refactor must
         # not alter a single hand-family detection.
@@ -429,6 +450,34 @@ class TestDerive:
             r.name == HAND_FAMILIES[r.key].name for r in rows if r.key in HAND_FAMILIES
         )
 
+    def test_replaces_event_alone_yields_successor_role(self):
+        # An isolated `replaces` event — no `replaced_by` event and no
+        # add-on signal for the successor code itself — must land the
+        # "successor" role branch (elif "replaces" in kinds), not fall
+        # through to "base".
+        elements = {}
+        events = {"88888": [_ev("88888", 2022, "replaces", frm="77777")]}
+        descriptions = {"88888": "Successor widget", "77777": "Predecessor widget"}
+        rows = derive_families(elements, events, descriptions)
+        by_code = {r.code: r for r in rows}
+        assert by_code["88888"].role == "successor"
+
+    def test_defined_by_reference_to_relation_merges_codes(self):
+        # `_LINK_RELATIONS` ("defined-by-reference-to") is a narrow,
+        # direct family-membership edge distinct from the lineage-event
+        # edges above — a code whose only connection to another is this
+        # relation element must still land in the same family.
+        elements = {
+            "12121": [
+                _el("12121", "relation", "defined-by-reference-to", "34343"),
+            ],
+            "34343": [],
+        }
+        descriptions = {"12121": "Alpha reference", "34343": "Beta referent"}
+        rows = derive_families(elements, {}, descriptions)
+        keys = {r.code: r.key for r in rows}
+        assert keys["12121"] == keys["34343"]
+
     def test_multi_code_replaced_by_does_not_merge(self):
         # #687 ruling 16: a `replaced_by` event naming more than one
         # to_codes is ambiguous (a blanket "these codes are replaced by
@@ -748,6 +797,31 @@ class TestCptGroups:
         assert groups["20003"][0] == "NEW-OR-ESTABLISHED-PATIENT@HOME-VISITS"
         assert groups["20001"][0] != groups["20003"][0]
 
+    def test_three_way_title_collision_under_the_same_parent_falls_back_to_a_suffix(
+        self,
+    ):
+        # Two collisions is enough to disambiguate on parent title
+        # (`@PARENT`) alone, but a *third* heading sharing both the same
+        # title AND the same immediate parent collides with that
+        # already-taken `@PARENT` key too — only then does the numbered
+        # `-2`/`-3` suffix loop run.
+        sections = [
+            _cpt_section("sec_a", ("Chapter One", "ParentX", "Repeated")),
+            _cpt_section("sec_b", ("Chapter Two", "ParentX", "Repeated")),
+            _cpt_section("sec_c", ("Chapter Three", "ParentX", "Repeated")),
+        ]
+        codes = [
+            _cpt_code("81001", "sec_a"),
+            _cpt_code("81002", "sec_a"),
+            _cpt_code("81003", "sec_b"),
+            _cpt_code("81004", "sec_b"),
+            _cpt_code("81005", "sec_c"),
+            _cpt_code("81006", "sec_c"),
+        ]
+        groups = cpt_groups(codes, sections)
+        keys = {groups[c][0] for c in ("81001", "81003", "81005")}
+        assert keys == {"REPEATED", "REPEATED@PARENTX", "REPEATED@PARENTX-2"}
+
 
 class TestDeriveCpt:
     def test_ccm_hand_family_wins_merged_component_note_is_own_heading(self):
@@ -995,6 +1069,38 @@ class TestDeriveCpt:
         assert by_code["G2058"].since == 2020 and by_code["G2058"].until == 2021
 
 
+class TestConnectedHelper:
+    """``_connected`` tested in isolation — the same way ``TestCptEdgesHelper``
+    below tests ``_cpt_edges`` directly. In practice ``derive_families``'s
+    hand-family-split path always calls it with an empty ``unreached`` list
+    (every code in a >= 2-hand-spanning component is provably reachable by
+    some hand's own BFS — the nearest seed's shortest path to any node
+    never crosses a *different* hand's seed, so nothing is ever left over;
+    confirmed by exhaustive random-graph fuzzing, 300k trials, 0
+    counterexamples), so the BFS body here is only reachable by calling
+    ``_connected`` directly."""
+
+    def test_bfs_walks_multi_node_components_and_ignores_outside_edges(self):
+        adj = {
+            "A": {"B"},
+            "B": {"A", "C"},
+            "C": {"B"},
+            "D": {"E"},
+            "E": {"D"},
+            "F": set(),
+        }
+        comps = _connected(["A", "B", "C", "D", "E", "F"], adj)
+        comp_sets = sorted((sorted(c) for c in comps), key=lambda c: c[0])
+        assert comp_sets == [["A", "B", "C"], ["D", "E"], ["F"]]
+
+    def test_edges_to_nodes_outside_the_set_do_not_count(self):
+        # "X" is adjacent to "A" but isn't in `nodes` — the component must
+        # not include it.
+        adj = {"A": {"B", "X"}, "B": {"A"}, "X": {"A"}}
+        comps = _connected(["A", "B"], adj)
+        assert [sorted(c) for c in comps] == [["A", "B"]]
+
+
 class TestCptEdgesHelper:
     def test_skips_only_edges_between_distinct_headings(self):
         # #687 controller review (Ruling C11), tested against the helper
@@ -1223,6 +1329,27 @@ class TestCptPresenceC12:
         assert by_code["90003"].since == 2024
         assert by_code["90003"].until is None
 
+    def test_empty_years_tuple_in_cpt_presence_is_skipped(self):
+        # C12: a code present as a `cpt_presence` key with an *empty*
+        # years tuple (`not years`) must be skipped rather than crash on
+        # `min(())`/`max(())` — it simply gets no cpt-derived since/until
+        # and falls back to the event-based (here: absent) since/until.
+        sections = [_cpt_section("sec_a", ("Chapter", "Heading A"))]
+        cpt_codes = [_cpt_code("90001", "sec_a", year=2024)]
+        cpt_presence = {"90001": (2019, 2021, 2022, 2024), "90099": ()}
+        rows = derive_families(
+            {},
+            {},
+            {"90099": "Empty years widget"},
+            cpt_codes=cpt_codes,
+            cpt_sections=sections,
+            cpt_presence=cpt_presence,
+        )
+        by_code = {r.code: r for r in rows}
+        assert by_code["90001"].since == 2019
+        assert by_code["90099"].since is None
+        assert by_code["90099"].until is None
+
     def test_without_cpt_presence_falls_back_to_cpt_codes_alone(self):
         # Pre-C12 behavior preserved when the caller doesn't have a
         # multi-edition presence map to give.

tests/pfs/test_guidance.py

@@ -7,6 +7,7 @@ from __future__ import annotations
 import duckdb
 import pytest
 
+import pfs.guidance as guidance_mod
 from bib.item import Manual, Regulation, Rule
 from bib.store import Store
 from pfs.codetables import (
@@ -16,13 +17,15 @@ from pfs.codetables import (
     write_cpt_edition,
     write_guidance,
 )
-from pfs.cpt_model import CptCode, CptEdition, CptSection
+from pfs.cpt_model import CptCode, CptEdition, CptInstruction, CptSection
 from pfs.guidance import (
     BARE_CFR_RE,
     CFR_RE,
     IOM_RE,
     MLN_RE,
     _cfr_refs,
+    _extract,
+    _iom_refs,
     build,
     dedupe,
     harvest,
@@ -189,6 +192,26 @@ class TestMlnRegex:
         assert MLN_RE.search("mln 1234567").group(1) == "1234567"
 
 
+class TestIomRefsUnmappedManual:
+    def test_manual_name_missing_from_pub_map_is_dropped(self, monkeypatch):
+        # IOM_RE's manual-name alternation only ever captures one of the
+        # six names in _MANUAL_PUB, so "not pub" (guidance.py:201) can't
+        # fire through the public regex today — it's a defensive guard
+        # against the map and regex drifting apart. Exercise it directly
+        # by shrinking the map out from under a real manual-name match.
+        monkeypatch.setattr(guidance_mod, "_MANUAL_PUB", {})
+        text = "See the Benefit Policy Manual, Chapter 5, Section 20 for details."
+        assert _iom_refs(text) == []
+
+
+class TestExtractMln:
+    def test_mln_reference_yields_unresolved_row(self, store):
+        s, *_rest = store
+        rows = _extract("See MLN907166 for the telehealth fact sheet.", s)
+        mln_rows = [r for r in rows if r[0] == "mln"]
+        assert mln_rows == [("mln", "MLN 907166", "")]
+
+
 # ── resolvers ───────────────────────────────────────────────────────
 
 
@@ -438,6 +461,126 @@ class TestHarvestCpt:
         assert r.p_id_src == 0
         assert r.page_src == 0
 
+    def test_empty_codes_returns_empty(self, store, con):
+        s, *_rest = store
+        assert harvest_cpt(con, s, (), family="CCM") == []
+
+    def test_non_missing_table_error_is_reraised(self, store, con, monkeypatch):
+        # cpt_years is imported inside harvest_cpt at call time, so
+        # patching pfs.codetables.cpt_years is visible to it. A plain
+        # ValueError doesn't look like a DuckDB "Catalog ... does not
+        # exist" error, so it must propagate rather than degrade to [].
+        s, *_rest = store
+
+        def _boom(_con):
+            raise ValueError("boom")
+
+        monkeypatch.setattr("pfs.codetables.cpt_years", _boom)
+        with pytest.raises(ValueError, match="boom"):
+            harvest_cpt(con, s, ("99490",), family="CCM")
+
+    def test_no_years_ingested_returns_empty(self, store, con):
+        # `con` has ensure_tables run (empty pfs.cpt_section) but no
+        # edition written, so cpt_years(con) returns [] without raising
+        # — the "if not years: return []" branch, distinct from the
+        # missing-table except branch covered via the `bare` connection
+        # in TestBuild below.
+        s, *_rest = store
+        assert harvest_cpt(con, s, ("99490",), family="CCM") == []
+
+    def test_section_with_no_guideline_text_is_skipped(self, store, con):
+        # A code mapped to a real section, but the section carries no
+        # guideline text (e.g. a heading without a trailing "*" range in
+        # the TOC) — the guideline loop's "continue" must skip it rather
+        # than call _extract("", ...).
+        s, *_rest = store
+        edition = CptEdition(
+            year=2024,
+            sections=(
+                CptSection(
+                    sec_id="sec_1",
+                    level=2,
+                    title="Chronic Care Management Services",
+                    path=("Evaluation and Management",),
+                    code_lo="99490",
+                    code_hi="99490",
+                    guideline="",
+                ),
+            ),
+            codes=(_cpt_code("99490", "sec_1"),),
+            instructions=(),
+            references=(),
+            crosswalks=(),
+            lists=(),
+            alternates=(),
+        )
+        write_cpt_edition(con, edition, "CPTED2024")
+        assert harvest_cpt(con, s, ("99490",), family="CCM") == []
+
+    def test_instructions_filtered_by_code_kind_and_keyword(self, store, con):
+        # Exercises every branch of the instructions loop: a "use-with"
+        # kind is skipped even for a targeted code, a "see" instruction
+        # for a code outside the family is skipped, a "see" instruction
+        # for the right code but with no CFR/Medicare/Chapter keyword is
+        # skipped, and a "see" instruction with a keyword produces a row.
+        s, sec_key, _manual_key = store
+        edition = CptEdition(
+            year=2024,
+            # cpt_years reads distinct years off pfs.cpt_section, so at
+            # least one section row is needed for the edition's year to
+            # be visible at all — its guideline is irrelevant here since
+            # no code maps to it (codes=()).
+            sections=(
+                CptSection(
+                    sec_id="sec_0",
+                    level=2,
+                    title="Unrelated Section",
+                    path=("Unrelated Section",),
+                    code_lo="",
+                    code_hi="",
+                    guideline="",
+                ),
+            ),
+            codes=(),
+            instructions=(
+                CptInstruction(
+                    code="99490",
+                    kind="use-with",
+                    text="(Do not report 99490 in conjunction with 42 CFR 410.78)",
+                    targets=(),
+                ),
+                CptInstruction(
+                    code="99439",
+                    kind="see",
+                    text="(See 42 CFR 410.78 for conditions of payment)",
+                    targets=(),
+                ),
+                CptInstruction(
+                    code="99490",
+                    kind="see",
+                    text="(See the local coverage determination for details)",
+                    targets=(),
+                ),
+                CptInstruction(
+                    code="99490",
+                    kind="see",
+                    text="(See 42 CFR 410.78(a)(3) for conditions of payment)",
+                    targets=(),
+                ),
+            ),
+            references=(),
+            crosswalks=(),
+            lists=(),
+            alternates=(),
+        )
+        write_cpt_edition(con, edition, "CPTED2024", force=True)
+        rows = harvest_cpt(con, s, ("99490",), family="CCM")
+        assert len(rows) == 1
+        r = rows[0]
+        assert (r.code, r.kind, r.locator) == ("99490", "cfr", "42 CFR 410.78(a)(3)")
+        assert r.item_key == sec_key
+        assert r.item_key_src == "CPTED2024"
+
 
 # ── build (FR + CPT, no CPT tables ingested) ─────────────────────────