merge: P48 LLM chat code valuation — RVUs, CFs and national payments beside the references (refs #682)

compose.yml

@@ -679,10 +679,13 @@ services:
       - storage
       - data
       - observability
+    volumes:
+      - ./data/replica:/app/data/replica:ro   # DuckDB read replica only (directory mount survives replica re-publish)
     environment:
       - LLM_OLLAMA_HOSTS=${LLM_OLLAMA_HOSTS_IN_CONTAINER:-http://ollama:11434}
       - LLM_PG_HOST=postgres
       - LLM_DB_PASSWORD=${LLM_DB_PASSWORD}
+      - LLM_DUCKDB_REPLICA=/app/data/replica/aco.ro.duckdb
       - OTEL_SERVICE_NAME=llm
       - OTEL_EXPORTER_OTLP_ENDPOINT=http://otel-collector:4317
     depends_on:

docs/docs/duckdb-concurrency.md

@@ -48,7 +48,7 @@ with duckdb_batch("aco") as con:      # retries, then names the lock holder
 ## Notebooks read a replica, not the primary
 
 Read-only opens through `conf.connect.duckdb()` resolve to the snapshot
-`data/aco.ro.duckdb` when it exists — so notebook kernels never hold the
+`data/replica/aco.ro.duckdb` when it exists — so notebook kernels never hold the
 primary's lock at all, and ingests stop caring how many notebooks are open.
 
 - **Refresh cadence**: every ingest republishes the snapshot as its last step

src/conf/connect.py

@@ -30,8 +30,10 @@ from conf import ROOT, cfg, path
 
 
 def _replica_path(name: str) -> Any:
-    """data/<name>.ro.duckdb next to the primary."""
-    return path(f"db.{name}").with_suffix(".ro.duckdb")
+    """data/replica/<name>.ro.duckdb — its own directory, so a container
+    can mount the snapshots without seeing the primary databases."""
+    primary = path(f"db.{name}")
+    return primary.parent / "replica" / f"{primary.stem}.ro.duckdb"
 
 
 def duckdb(
@@ -46,7 +48,8 @@ def duckdb(
     read_only
         Open in read-only mode (default ``True``).
     replica
-        Read-only opens resolve to the ``<name>.ro.duckdb`` snapshot
+        Read-only opens resolve to the ``replica/<name>.ro.duckdb``
+        snapshot
         when it exists — long-running readers (notebook kernels) then
         never hold the primary's single-writer lock, so ingests stop
         failing (#510). Staleness is bounded by the refresh cadence:
@@ -136,7 +139,7 @@ def duckdb_batch(
 
 
 def publish_replica(name: str = "aco") -> Any:
-    """Publish/refresh the read-only snapshot ``<name>.ro.duckdb``.
+    """Publish/refresh the read-only snapshot ``replica/<name>.ro.duckdb``.
 
     Called after each ingest so notebook readers (see :func:`duckdb`)
     work from a lock-free snapshot instead of contending with writers
@@ -152,7 +155,9 @@ def publish_replica(name: str = "aco") -> Any:
     import shutil
 
     src = path(f"db.{name}")
-    dst = _replica_path(name).resolve()
+    dst = _replica_path(name)
+    dst.parent.mkdir(parents=True, exist_ok=True)
+    dst = dst.resolve()
     tmp = dst.with_suffix(".tmp")
     with duckdb_batch(name) as con:
         con.execute("CHECKPOINT")

src/llm/chunk.py

@@ -14,6 +14,8 @@ import hashlib
 import re
 from dataclasses import dataclass
 
+from pfs.families import find_codes
+
 
 @dataclass(frozen=True)
 class Paragraph:
@@ -143,6 +145,10 @@ def chunk_doc(
                 "item_key": doc.key,
                 "seq": str(seq),
                 "section": heading,
+                # The heading is scanned with the piece: a section titled
+                # "## G0556 — APCM" is about G0556 all the way down, but
+                # only its first chunk repeats the code.
+                "codes": " ".join(find_codes(f"{heading} {piece}")),
             },
         )
         for seq, (heading, piece) in enumerate(pieces)
@@ -184,6 +190,7 @@ def _chunk_paragraphs(doc: Doc, target: int, overlap: int) -> list[Chunk]:
                 "item_key": doc.key,
                 "seq": str(seq),
                 "section": "",
+                "codes": " ".join(find_codes(text)),
                 "p_id": str(first.p_id),
                 "p_id_last": str(last.p_id),
                 "page": str(first.page),

src/llm/config.py

@@ -7,6 +7,7 @@ Env contract:
     LLM_PG_HOST          beats [llm].pg_host (containers set this to "postgres")
     LLM_DB_PASSWORD      required for pg_url(); lives in .env, never stack.toml
     LLM_BUILD_ANN_INDEX  beats [llm].build_ann_index ("1"/"true"/"yes" = True)
+    LLM_DUCKDB_REPLICA   beats [llm].duckdb_replica (containers set /app/data/replica/aco.ro.duckdb)
 
 Ollama has no GPU-size API, so VRAM is *declared* per host; liveness is
 probed at request time (``HostPool.check``). Together they implement
@@ -43,6 +44,9 @@ class LlmConfig:
         default_factory=lambda: dict(_DEFAULT_K_PER_KIND)
     )
     top_n: int = 8
+    duckdb_replica: str = "data/replica/aco.ro.duckdb"
+    valuation_years: int = 4
+    code_cited_per_code: int = 3
 
 
 def parse_hosts(spec: str) -> tuple[tuple[str, float], ...]:
@@ -97,6 +101,10 @@ def load() -> LlmConfig:
         recency_weight=float(_opt(section, "recency_weight", 0.3)),
         k_per_kind=k_per_kind,
         top_n=int(_opt(section, "top_n", 8)),
+        duckdb_replica=os.environ.get("LLM_DUCKDB_REPLICA")
+        or str(_opt(section, "duckdb_replica", "data/replica/aco.ro.duckdb")),
+        valuation_years=int(_opt(section, "valuation_years", 4)),
+        code_cited_per_code=int(_opt(section, "code_cited_per_code", 3)),
         embed_dim=int(section.embed_dim),
         build_ann_index=build_ann_index,
         pg_host=os.environ.get("LLM_PG_HOST", str(section.pg_host)),

src/llm/evidence.py

@@ -0,0 +1,229 @@
+"""Structured evidence for the chat: code valuations beside the excerpts.
+
+``valuation_evidence`` detects HCPCS/CPT codes (and registered families)
+in the question, reads their RVUs from the DuckDB read replica and
+attaches conversion factors from ``pfs.rules``. ``code_cited_sources``
+pulls rule chunks whose ``codes`` metadata mention those codes straight
+from pgvector so the answer can cite where CMS valued them. Both degrade
+to "no evidence" with a warning — the chat never fails because of them.
+"""
+
+from __future__ import annotations
+
+import logging
+import os
+import threading
+from dataclasses import asdict, dataclass
+from pathlib import Path
+from typing import Any, Sequence
+
+import duckdb
+from sqlalchemy import text
+
+from llm.config import LlmConfig
+from llm.links import as_source
+from pfs.families import detect_codes
+from pfs.valuation import ValuationRow, valuation
+
+log = logging.getLogger(__name__)
+
+_HEADER = (
+    "Valuation (authoritative for RVUs, conversion factors and payments; "
+    "national unadjusted, no GPCI):"
+)
+
+
+def _n(v: float | None, digits: int = 2, money: bool = False) -> str:
+    if v is None:
+        return "n/a"
+    s = f"{v:.{digits}f}"
+    return f"${s}" if money else s
+
+
+@dataclass(frozen=True)
+class ValuationEvidence:
+    codes: tuple[str, ...]
+    families: tuple[str, ...]
+    rows: tuple[ValuationRow, ...]
+    unpriced: tuple[str, ...]
+
+    def prompt_block(self) -> str:
+        lines = [_HEADER]
+        for r in self.rows:
+            # The status note says why an unpaid code shows no dollars, and
+            # the CF note says which of the year's CFs the money came from.
+            status = f"{r.status or '?'}" + (
+                f"; {r.status_note}" if r.status_note else ""
+            )
+            cf = _n(r.cf, 4, money=True) + (f" [{r.cf_note}]" if r.cf_note else "")
+            lines.append(
+                f'{r.label} {r.code} "{r.description}" — status {status}; '
+                f"work {_n(r.work)}; PE {_n(r.pe_nf)} non-fac / {_n(r.pe_f)} fac; "
+                f"MP {_n(r.mp)}; total {_n(r.total_nf)} non-fac / {_n(r.total_f)} fac; "
+                f"CF {cf} → payment {_n(r.pay_nf, money=True)} non-fac / "
+                f"{_n(r.pay_f, money=True)} fac"
+            )
+        if self.unpriced:
+            lines.append(
+                "Not priced in the indexed fee schedule: " + ", ".join(self.unpriced)
+            )
+        return "\n".join(lines)
+
+    def payload(self) -> dict[str, Any]:
+        provenance: list[dict[str, str]] = []
+        seen: set[str] = set()
+        for r in self.rows:
+            if r.label not in seen:
+                seen.add(r.label)
+                provenance.append(
+                    {
+                        "label": r.label,
+                        "vintage": r.vintage,
+                        "citation": r.citation,
+                        "url": r.url,
+                    }
+                )
+        return {
+            "type": "valuation",
+            "codes": list(self.codes),
+            "families": list(self.families),
+            "rows": [asdict(r) for r in self.rows],
+            "provenance": provenance,
+            "unpriced": list(self.unpriced),
+        }
+
+
+#: The open replica, keyed by (path, mtime_ns). DuckDB pays a real cost
+#: to open a database, and every chat turn that mentions a code would
+#: otherwise pay it. ``publish_replica`` swaps the file with os.replace,
+#: so the handle stays valid on the old inode until we notice the new
+#: mtime and reopen.
+_REPLICA: tuple[str, int, Any] | None = None
+_REPLICA_LOCK = threading.Lock()
+
+
+def _mtime(path: str) -> int:
+    try:
+        return os.stat(path).st_mtime_ns
+    except OSError:
+        return -1
+
+
+def _connect(path: str) -> Any:
+    """The cached read-only connection to *path*, reopened when the
+    replica file has been republished since we opened it."""
+    global _REPLICA
+    mtime = _mtime(path)
+    with _REPLICA_LOCK:  # two chat threads racing the first open would leak one
+        if _REPLICA is not None:
+            cached_path, cached_mtime, con = _REPLICA
+            if (cached_path, cached_mtime) == (path, mtime):
+                return con
+            _REPLICA = None
+            try:
+                con.close()
+            except Exception as e:  # noqa: BLE001 — a stale handle must not break the chat
+                log.warning("stale replica handle not closed: %s", e)
+        con = duckdb.connect(path, read_only=True)
+        _REPLICA = (path, mtime, con)
+        return con
+
+
+def _replica_path(cfg: LlmConfig) -> str:
+    p = Path(cfg.duckdb_replica)
+    if p.is_absolute():
+        return str(p)
+    from conf import ROOT
+
+    return str(ROOT / p)
+
+
+def valuation_evidence(question: str, cfg: LlmConfig) -> ValuationEvidence | None:
+    det = detect_codes(question)
+    if not det.codes:
+        return None
+    path = _replica_path(cfg)
+    try:
+        # A chat turn runs in a threadpool worker, and DuckDB gives each
+        # thread its own cursor off the shared connection — without one,
+        # concurrent turns read each other's result sets.
+        cur = _connect(path).cursor()
+    except Exception as e:  # noqa: BLE001 — duckdb raises several types
+        log.warning("valuation evidence skipped (%s): %s", path, e)
+        return None
+    # A whole family is six codes; showing four vintages of each would
+    # crowd the excerpts out of the prompt.
+    years = cfg.valuation_years if len(det.codes) <= 3 else 2
+    try:
+        rows, unpriced = valuation(cur, list(det.codes), years=years)
+    except Exception as e:  # noqa: BLE001
+        log.warning("valuation evidence skipped (query): %s", e)
+        return None
+    finally:
+        cur.close()  # the cached parent connection stays open
+    return ValuationEvidence(det.codes, det.families, tuple(rows), tuple(unpriced))
+
+
+_CITED_SQL = text(
+    "SELECT document, cmetadata FROM ( "
+    "SELECT e.document, e.cmetadata, "
+    "row_number() OVER ( "
+    "PARTITION BY w.code "
+    "ORDER BY e.cmetadata->>'date' DESC NULLS LAST, e.cmetadata->>'item_key', e.cmetadata->>'p_id' "
+    ") AS rn "
+    "FROM langchain_pg_embedding e "
+    "JOIN unnest(CAST(:codes AS text[])) AS w(code) "
+    "ON w.code = ANY(string_to_array(COALESCE(e.cmetadata->>'codes', ''), ' ')) "
+    "WHERE e.collection_id = (SELECT uuid FROM langchain_pg_collection "
+    "WHERE name = :collection) "
+    ") t "
+    "WHERE rn <= :window "
+    "ORDER BY cmetadata->>'date' DESC NULLS LAST, cmetadata->>'item_key', cmetadata->>'p_id'"
+)
+
+
+def code_cited_sources(
+    engine: Any, codes: Sequence[str], *, per_code: int, collection: str = "rules"
+) -> list[dict]:
+    """Chunks in *collection* whose ``codes`` metadata mention any of
+    *codes* — at most *per_code* per code, newest rule first, deduped on
+    (item_key, p_id). Scores are 0.0: these are additive, not ranked."""
+    wanted = [c.upper() for c in codes]
+    if not wanted:
+        return []
+    try:
+        with engine.begin() as conn:
+            rows = conn.execute(
+                _CITED_SQL,
+                {
+                    "collection": collection,
+                    "codes": wanted,
+                    "window": per_code * 2,
+                },
+            ).fetchall()
+    except Exception as e:  # noqa: BLE001
+        log.warning("code-cited sources skipped: %s", e)
+        return []
+    out: list[dict] = []
+    seen: set[tuple[str, str]] = set()
+    count: dict[str, int] = {c: 0 for c in wanted}
+    for document, md in rows:
+        md = {k: str(v) for k, v in (md or {}).items()}
+        key = (md.get("item_key", ""), md.get("p_id", ""))
+        if key in seen:
+            continue
+        hit_codes = [
+            c for c in md.get("codes", "").split() if c in count and count[c] < per_code
+        ]
+        if not hit_codes:
+            continue
+        seen.add(key)
+        for c in hit_codes:
+            count[c] += 1
+        out.append(as_source(md, document[:250], 0.0))
+    return out
+
+
+def merge_sources(retrieved: list[dict], extra: list[dict]) -> list[dict]:
+    labels = {s["label"] for s in retrieved}
+    return retrieved + [s for s in extra if s["label"] not in labels]

src/llm/links.py

@@ -19,6 +19,7 @@ from bib.frlink import text_fragment
 _COMMENT = "https://www.regulations.gov/comment/{cid}"
 _DOWNLOAD = "https://downloads.regulations.gov/{cid}/{name}"
 _TITLE_MAX = 60
+_SNIPPET_CHARS = 500
 
 
 def _short(title: str) -> str:
@@ -84,3 +85,21 @@ def for_source(md: dict[str, str], snippet: str) -> tuple[str, str]:
     if kind == "rule":
         return _rule(md, snippet)
     return _corpus(md)
+
+
+def as_source(md: dict[str, str], text: str, score: float) -> dict:
+    """The source dict the chat sends to the prompt and the UI."""
+    snippet = text[:_SNIPPET_CHARS].strip()
+    url, label = for_source(md, snippet)
+    return {
+        "id": label,
+        "label": label,
+        "kind": md.get("kind", ""),
+        "url": url,
+        "title": md.get("title", ""),
+        "date": md.get("date", ""),
+        "docket": md.get("docket", ""),
+        "comment_id": md.get("comment_id", ""),
+        "snippet": snippet,
+        "score": round(score, 4),
+    }

src/llm/migrate.py

@@ -48,6 +48,14 @@ _HNSW_DDL = [
     "ALTER TABLE langchain_pg_embedding ALTER COLUMN embedding TYPE vector({dim})",
     "CREATE INDEX IF NOT EXISTS ix_embedding_hnsw "
     "ON langchain_pg_embedding USING hnsw (embedding vector_cosine_ops)",
+    # Code-cited sources (llm.evidence) match a code against the space-
+    # separated cmetadata->>'codes' inside one collection; without these
+    # two the query is a sequential scan of every embedding.
+    "CREATE INDEX IF NOT EXISTS ix_embedding_codes "
+    "ON langchain_pg_embedding USING gin "
+    "(string_to_array(coalesce(cmetadata->>'codes',''), ' '))",
+    "CREATE INDEX IF NOT EXISTS ix_embedding_collection "
+    "ON langchain_pg_embedding (collection_id)",
 ]
 
 

src/llm/rag.py

@@ -7,7 +7,11 @@ its own. Retrieval embeds the question once and searches the
 merged and re-ranked by similarity × recency (``llm.rerank``), then each
 source gets a deep link (``llm.links``). Generation streams from the
 largest live Ollama host (``HostPool.acquire_generation``) with the
-model tier that host can hold (``pick_model``).
+model tier that host can hold (``pick_model``). When the question
+cites HCPCS/CPT codes, ``llm.evidence`` supplies authoritative
+valuation rows and rule chunks that cite those codes; the SSE event
+order is ``valuation`` (only when evidence is found) → ``token``* →
+``sources`` → ``done``.
 """
 
 from __future__ import annotations
@@ -19,14 +23,20 @@ from typing import Iterator
 import httpx
 
 from llm.config import LlmConfig
-from llm.links import for_source
+from llm.evidence import (
+    ValuationEvidence,
+    code_cited_sources,
+    merge_sources,
+    valuation_evidence,
+)
+from llm.index import _engine
+from llm.links import as_source
 from llm.pool import HostPool, PoolEmbeddings, pick_model
 from llm.rerank import Hit, blend, filter_since
 
 _TIMEOUT = httpx.Timeout(300.0, connect=5.0)
 _COLLECTIONS = {"comment": "comments", "rule": "rules", "corpus": "corpus"}
 _OVERFETCH = 3
-_SNIPPET_CHARS = 500
 
 _SYSTEM = (
     "You answer questions about CMS rulemaking using ONLY the excerpts "
@@ -41,6 +51,12 @@ _SYSTEM = (
     "statement comes from when it matters. If the excerpts do not contain "
     "the answer, say you don't have information on that in the indexed "
     "library — do not invent facts."
+    " A Valuation section may follow the excerpts. Its rows are "
+    "authoritative for RVUs, conversion factors and payment amounts; when "
+    "you state any of those numbers, cite the row's bracketed label exactly, "
+    "e.g. [PFS CY2026 Addendum B]. Do not compute, extrapolate or convert "
+    "numbers beyond what the rows show; if a code is listed as not priced, "
+    "say so."
 )
 
 
@@ -65,21 +81,7 @@ def _hits(question_vec: list[float], *, cfg: LlmConfig, pool: HostPool) -> list[
 
 
 def _source(hit: Hit) -> dict:
-    md = hit.metadata
-    snippet = hit.text[:_SNIPPET_CHARS].strip()
-    url, label = for_source(md, snippet)
-    return {
-        "id": label,
-        "label": label,
-        "kind": md.get("kind", ""),
-        "url": url,
-        "title": md.get("title", ""),
-        "date": md.get("date", ""),
-        "docket": md.get("docket", ""),
-        "comment_id": md.get("comment_id", ""),
-        "snippet": snippet,
-        "score": round(hit.score, 4),
-    }
+    return as_source(hit.metadata, hit.text, hit.score)
 
 
 def retrieve(
@@ -106,8 +108,12 @@ def retrieve(
     return [_source(h) for h in ranked]
 
 
-def build_messages(question: str, sources: list[dict]) -> list[dict]:
-    """Grounded chat messages: system rules + question with excerpts."""
+def build_messages(
+    question: str, sources: list[dict], evidence: ValuationEvidence | None = None
+) -> list[dict]:
+    """Grounded chat messages: system rules + question with excerpts and,
+    when available, a Valuation block between the excerpts and the
+    question."""
     if sources:
         context = "\n\n".join(
             f"[{s['label']}] ({s.get('kind', '')}, {s.get('date', '') or 'undated'}) "
@@ -116,7 +122,11 @@ def build_messages(question: str, sources: list[dict]) -> list[dict]:
         )
     else:
         context = "(no relevant excerpts found)"
-    user = f"Excerpts:\n\n{context}\n\nQuestion: {question}"
+    parts = [f"Excerpts:\n\n{context}"]
+    if evidence is not None:
+        parts.append(evidence.prompt_block())
+    parts.append(f"Question: {question}")
+    user = "\n\n".join(parts)
     return [
         {"role": "system", "content": _SYSTEM},
         {"role": "user", "content": user},
@@ -128,13 +138,23 @@ def stream_answer(
 ) -> Iterator[dict]:
     """Retrieve, then stream a grounded answer from the largest live host.
 
-    Yields ``{"type":"token","text":…}`` events as the model generates,
-    then one ``{"type":"sources", "sources": […], "model": …, "host": …}``
-    and a final ``{"type":"done"}``.
+    When the question cites HCPCS/CPT codes, yields a ``valuation`` event
+    first (evidence's ``payload()``) and merges rule chunks that cite
+    those codes into ``sources``. Then yields ``{"type":"token","text":…}``
+    events as the model generates, then one
+    ``{"type":"sources", "sources": […], "model": …, "host": …}`` and a
+    final ``{"type":"done"}``.
     """
     sources = retrieve(question, cfg=cfg, pool=pool, since=since)
+    evidence = valuation_evidence(question, cfg)
+    if evidence is not None:
+        cited = code_cited_sources(
+            _engine(cfg), evidence.codes, per_code=cfg.code_cited_per_code
+        )
+        sources = merge_sources(sources, cited)
+        yield evidence.payload()
     pool.check(cfg.instruct_model)
-    messages = build_messages(question, sources)
+    messages = build_messages(question, sources, evidence)
     with pool.acquire_generation() as host, httpx.Client(timeout=_TIMEOUT) as client:
         model = pick_model(cfg, pool, host)
         with client.stream(

src/llm/web/chat.html

@@ -66,6 +66,24 @@
       background: var(--secondary); padding: 1px 5px; border-radius: 2px;
       white-space: nowrap;
     }
+    table.valuation {
+      border-collapse: collapse; width: 100%; font-size: 13px; margin-top: 4px;
+      font-variant-numeric: tabular-nums;
+    }
+    table.valuation th, table.valuation td {
+      border: 1px solid var(--border); padding: 4px 6px; text-align: right; white-space: nowrap;
+    }
+    table.valuation th { font-family: var(--font-mono); font-size: 10px; letter-spacing: .08em;
+      text-transform: uppercase; color: var(--muted-fg); }
+    table.valuation td:nth-child(-n+3), table.valuation th:nth-child(-n+3) { text-align: left; }
+    table.valuation tr.proposed td { font-style: italic; opacity: .85; }
+    table.valuation caption {
+      caption-side: bottom; text-align: left; font-size: 11px;
+      color: var(--muted-fg); padding: 4px 2px 0;
+    }
+    .valuation-wrap { overflow-x: auto; }
+    .provenance { font-size: 12px; color: var(--muted-fg); margin-top: 4px; }
+    .provenance .cite { margin-right: 4px; }
     details.sources { margin-top: 2px; font-size: 14px; }
     details.sources > summary {
       cursor: pointer; color: var(--muted-fg); font-family: var(--font-mono);
@@ -177,6 +195,79 @@
       if (last < text.length) el.appendChild(document.createTextNode(text.slice(last)));
     }
 
+    const VAL_COLS = [
+      ['code', 'Code'], ['description', 'Description'], ['vintage', 'Vintage'], ['status', 'St'],
+      ['work', 'Work'], ['pe_nf', 'PE NF'], ['pe_f', 'PE F'], ['mp', 'MP'],
+      ['total_nf', 'Total NF'], ['total_f', 'Total F'], ['cf', 'CF (non-APM)'],
+      ['pay_nf', 'Pay NF'], ['pay_f', 'Pay F'],
+    ];
+    function fmt(key, v) {
+      if (v === null || v === undefined) return '—';
+      if (key === 'cf') return '$' + Number(v).toFixed(4);
+      if (key === 'pay_nf' || key === 'pay_f') return '$' + Number(v).toFixed(2);
+      if (typeof v === 'number') return v.toFixed(2);
+      return String(v);
+    }
+    function renderValuation(wrap, ev) {
+      const rows = ev.rows || [];
+      const box = document.createElement('div');
+      box.className = 'valuation-wrap';
+      if (rows.length) {
+        const t = document.createElement('table');
+        t.className = 'valuation';
+        t.createCaption().textContent =
+          'National unadjusted amounts (total RVUs \u00d7 CF); no geographic (GPCI) adjustment.';
+        const thead = t.createTHead().insertRow();
+        for (const [, name] of VAL_COLS) { const th = document.createElement('th'); th.textContent = name; thead.appendChild(th); }
+        const tb = t.createTBody();
+        for (const r of rows) {
+          const tr = tb.insertRow();
+          if (r.proposed) tr.className = 'proposed';
+          for (const [key] of VAL_COLS) {
+            const td = tr.insertCell();
+            td.textContent = fmt(key, r[key]);
+            // why a status is not paid / which of the year's CFs this is
+            if (key === 'status' && r.status_note) td.title = r.status_note;
+            if (key === 'cf' && r.cf_note) td.title = r.cf_note;
+          }
+        }
+        box.appendChild(t);
+      }
+      const prov = document.createElement('div');
+      prov.className = 'provenance';
+      for (const p of ev.provenance || []) {
+        const c = document.createElement('span'); c.className = 'cite'; c.textContent = p.label;
+        prov.appendChild(c);
+        let cite;
+        if (p.url) {
+          cite = document.createElement('a');
+          cite.href = p.url; cite.target = '_blank'; cite.rel = 'noopener';
+        } else {
+          cite = document.createElement('span');  // an empty href reloads the chat
+        }
+        cite.textContent = p.citation || p.vintage;
+        prov.appendChild(cite);
+        prov.appendChild(document.createTextNode(' '));
+      }
+      if ((ev.unpriced || []).length) {
+        prov.appendChild(document.createTextNode('not priced: ' + ev.unpriced.join(', ')));
+      }
+      const notes = [];
+      for (const r of rows) {
+        if (!r.status_note) continue;
+        const line = r.code + ' (' + (r.status || '?') + '): ' + r.status_note;
+        if (!notes.includes(line)) notes.push(line);
+      }
+      for (const line of notes) {
+        const d = document.createElement('div');
+        d.textContent = line;
+        prov.appendChild(d);
+      }
+      box.appendChild(prov);
+      wrap.appendChild(box);
+      log.scrollTop = log.scrollHeight;
+    }
+
     function renderSources(wrap, ev) {
       const sources = ev.sources || [];
       if (ev.model) {
@@ -252,6 +343,7 @@
             if (!line) continue;
             const ev = JSON.parse(line);
             if (ev.type === 'token') { answer += ev.text; paint(b, answer); log.scrollTop = log.scrollHeight; }
+            else if (ev.type === 'valuation') renderValuation(wrap, ev);
             else if (ev.type === 'sources') renderSources(wrap, ev);
             else if (ev.type === 'error') { b.className = 'bubble err'; b.textContent = 'Error: ' + ev.message; }
           }

src/pfs/families.py

@@ -0,0 +1,109 @@
+"""Code families and deterministic code detection for chat questions.
+
+The registry used to live in a notebook cell
+(``notebooks/palliative_care_rfi.py``); this is its home in ``src`` so
+the chat, notebooks and scripts share one list. Pure: no I/O, no
+DuckDB, no narwhals — it must import inside the ``llm`` container.
+"""
+
+from __future__ import annotations
+
+import re
+from dataclasses import dataclass
+
+# HCPCS level II (letter + 4 digits) or CPT (5 digits). Validation against
+# the fee schedule happens at lookup time (pfs.valuation), not here.
+CODE_RE = re.compile(r"\b(?:[A-Z]\d{4}|\d{5})\b", re.IGNORECASE)
+
+# "91 FR 43842" — an FR page number is five digits and so reads as a CPT
+# code; citations are removed before scanning so they never become one.
+_FR_CITE_RE = re.compile(r"\b\d{1,3}\s+FR\s+\d{3,6}\b", re.IGNORECASE)
+
+
+def find_codes(text: str) -> tuple[str, ...]:
+    """Sorted unique codes literally present in *text*, upper-cased.
+
+    Federal Register citations are stripped first: their page numbers are
+    not codes, and a chat answer that priced "43842" out of "91 FR 43842"
+    would be inventing money.
+    """
+    text = _FR_CITE_RE.sub(" ", text)
+    return tuple(sorted({m.group(0).upper() for m in CODE_RE.finditer(text)}))
+
+
+@dataclass(frozen=True)
+class Family:
+    key: str
+    name: str
+    codes: tuple[str, ...]
+    synonyms: tuple[str, ...]  # lower-case; matched on word boundaries
+
+
+FAMILIES: dict[str, Family] = {
+    "ACP": Family(
+        "ACP",
+        "Advance Care Planning",
+        ("99497", "99498"),
+        ("acp", "advance care planning"),
+    ),
+    "CCM": Family(
+        "CCM",
+        "Chronic Care Management",
+        ("99437", "99439", "99487", "99489", "99490", "99491"),
+        ("ccm", "chronic care management"),
+    ),
+    "PCM": Family(
+        "PCM",
+        "Principal Care Management",
+        ("99424", "99425", "99426", "99427"),
+        ("pcm", "principal care management"),
+    ),
+    "TCM": Family(
+        "TCM",
+        "Transitional Care Management",
+        ("99495", "99496"),
+        ("tcm", "transitional care management"),
+    ),
+    "APCM": Family(
+        "APCM",
+        "Advanced Primary Care Management",
+        ("G0556", "G0557", "G0558"),
+        ("apcm", "advanced primary care management"),
+    ),
+}
+
+_CODE_TO_FAMILY: dict[str, Family] = {c: f for f in FAMILIES.values() for c in f.codes}
+
+
+def family_of(code: str) -> Family | None:
+    return _CODE_TO_FAMILY.get(code.upper())
+
+
+@dataclass(frozen=True)
+class Detection:
+    codes: tuple[str, ...]  # sorted unique: explicit + family-expanded
+    families: tuple[str, ...]  # family keys, sorted
+    explicit: tuple[str, ...]  # codes literally present in the text
+
+
+def detect_codes(text: str) -> Detection:
+    """Codes a question is about: explicit codes plus every code of any
+    family named (by synonym) or touched (by one member code)."""
+    explicit = find_codes(text)
+    lowered = text.lower()
+    families: set[str] = set()
+    for fam in FAMILIES.values():
+        if any(re.search(rf"\b{re.escape(s)}\b", lowered) for s in fam.synonyms):
+            families.add(fam.key)
+    for code in explicit:
+        fam = family_of(code)
+        if fam is not None:
+            families.add(fam.key)
+    codes = set(explicit)
+    for key in families:
+        codes.update(FAMILIES[key].codes)
+    return Detection(
+        codes=tuple(sorted(codes)),
+        families=tuple(sorted(families)),
+        explicit=explicit,
+    )

src/pfs/valuation.py

@@ -0,0 +1,273 @@
+"""RVUs, conversion factor and national unadjusted payment per code and
+vintage, read from the DuckDB replica (``pfs.rvu`` final years,
+``pfs.rvu_proposed`` for the newest NPRM). Conversion factors come from
+``pfs.rules`` — never from ``pfs.rvu.conv_factor``. Pure DuckDB SQL; no
+narwhals (this module runs inside the ``llm`` container).
+"""
+
+from __future__ import annotations
+
+import logging
+from dataclasses import dataclass
+from typing import Any, Sequence
+
+from pfs.nprm import NPRM_SOURCES
+from pfs.rules import RULES, ProposedRule, RuleYear, proposed_for
+
+log = logging.getLogger(__name__)
+
+
+@dataclass(frozen=True)
+class ValuationRow:
+    code: str
+    description: str
+    vintage: str  # "CY2026 final" | "CY2027 proposed"
+    year: int
+    proposed: bool
+    status: str
+    work: float | None
+    pe_nf: float | None
+    pe_f: float | None
+    mp: float | None
+    total_nf: float | None
+    total_f: float | None
+    cf: float
+    pay_nf: float | None
+    pay_f: float | None
+    label: str  # "[PFS CY2026 Addendum B]"
+    citation: str  # "90 FR 49266"
+    url: str
+    status_note: str = ""  # why a non-A/R/T status is not paid ("" when paid)
+    cf_note: str = ""  # which of the year's several CFs this one is
+
+
+def vintage_label(year: int, proposed: bool) -> str:
+    return f"[CY{year} NPRM Addendum B]" if proposed else f"[PFS CY{year} Addendum B]"
+
+
+def fr_citation_url(cite: str) -> str:
+    parts = cite.split()
+    if len(parts) != 3 or parts[1].upper() != "FR":
+        return ""
+    return f"https://www.federalregister.gov/citation/{parts[0]}-FR-{parts[2]}"
+
+
+#: Payment status codes paid as total RVUs × CF. Every other status is
+#: priced some other way (or not at all), so multiplying its RVUs by the
+#: CF would put a number on screen that Medicare never pays.
+PAID_STATUSES = frozenset({"A", "R", "T"})
+
+_STATUS_NOTES = {
+    "B": "bundled — no separate PFS payment",
+    "C": "carrier priced by the MAC",
+    "J": "anesthesia — paid on base units × the anesthesia CF",
+    "I": "not valid for Medicare",
+    "N": "non-covered",
+    "E": "excluded by regulation",
+    "M": "measurement code — not paid",
+    "X": "statutory exclusion",
+    "P": "bundled/excluded",
+}
+
+
+def status_note(status: str) -> str:
+    """Why *status* is not paid from RVUs × CF — ``""`` when it is."""
+    code = (status or "").strip().upper()
+    if code in PAID_STATUSES:
+        return ""
+    if code in _STATUS_NOTES:
+        return _STATUS_NOTES[code]
+    if not code:
+        return "status not stated — no RVU-based payment"
+    return f"status {code} — no RVU-based payment"
+
+
+def final_cf_note(rule: RuleYear) -> str:
+    """Which of a final year's several conversion factors ``cf`` is.
+
+    Since CY2026 CMS publishes separate QP (APM) and non-QP standard
+    CFs, and a mid-year legislative update can split a year in two — an
+    unlabelled dollar figure would hide both.
+    """
+    parts = ["non-APM standard CF"]
+    if rule.cf_qp is not None and rule.cf_qp != rule.conversion_factor:
+        parts.append(f"QP CF ${rule.cf_qp:.4f}")
+    segs = sorted(rule.cf_segments, key=lambda s: s.start)
+    if len(segs) > 1:
+        earlier = next(
+            (
+                (seg, nxt)
+                for seg, nxt in zip(segs, segs[1:])
+                if seg.cf != rule.conversion_factor
+            ),
+            None,
+        )
+        if earlier is None:
+            parts.append("mid-year split — dominant CF shown")
+        else:
+            seg, nxt = earlier
+            parts.append(
+                f"mid-year split — dominant CF shown (${seg.cf:.4f} before {nxt.start})"
+            )
+    return "; ".join(parts)
+
+
+def proposed_cf_note(prop: ProposedRule) -> str:
+    """Which CF a proposed row shows (the NPRM's non-QP standard one)."""
+    parts = ["proposed; non-QP CF"]
+    if prop.cf_qp is not None and prop.cf_qp != prop.conversion_factor:
+        parts.append(f"QP CF ${prop.cf_qp:.4f}")
+    return "; ".join(parts)
+
+
+def _pay(total: float | None, cf: float, status: str) -> float | None:
+    """National unadjusted payment, or ``None`` when the status is not
+    paid from RVUs (see :data:`PAID_STATUSES`)."""
+    if total is None or (status or "").strip().upper() not in PAID_STATUSES:
+        return None
+    return round(total * cf, 2)
+
+
+_FINAL_SQL = """
+WITH base AS (
+    SELECT hcpcs, year, description, status_code, work_rvu, non_fac_pe_rvu,
+           fac_pe_rvu, mp_rvu, non_fac_total, fac_total
+    FROM pfs.rvu
+    WHERE hcpcs IN (SELECT UNNEST(?::VARCHAR[]))
+      AND year IN (SELECT UNNEST(?::INTEGER[]))
+    QUALIFY row_number() OVER (
+        PARTITION BY hcpcs, year ORDER BY (mod IS NULL OR mod = '') DESC, mod
+    ) = 1
+)
+SELECT * FROM base
+QUALIFY dense_rank() OVER (PARTITION BY hcpcs ORDER BY year DESC) <= ?
+ORDER BY hcpcs, year
+"""
+
+_PROPOSED_SQL = """
+SELECT hcpcs, description, status_code, work_rvu, non_fac_pe_rvu, fac_pe_rvu, mp_rvu
+FROM pfs.rvu_proposed
+WHERE cms_rule_id = ? AND hcpcs IN (SELECT UNNEST(?::VARCHAR[]))
+QUALIFY row_number() OVER (
+    PARTITION BY hcpcs ORDER BY (mod IS NULL OR mod = '') DESC, mod
+) = 1
+ORDER BY hcpcs
+"""
+
+
+_UNKNOWN_YEARS_SQL = """
+SELECT DISTINCT year FROM pfs.rvu
+WHERE hcpcs IN (SELECT UNNEST(?::VARCHAR[]))
+  AND year NOT IN (SELECT UNNEST(?::INTEGER[]))
+ORDER BY year
+"""
+
+
+def _sum(*vals: float | None) -> float | None:
+    if any(v is None for v in vals):
+        return None
+    return round(sum(vals), 4)  # type: ignore[arg-type]
+
+
+def valuation(
+    con: Any, codes: Sequence[str], *, years: int = 4
+) -> tuple[list[ValuationRow], list[str]]:
+    """Rows for *codes* (last *years* final vintages + the newest NPRM),
+    ordered by code then year with the proposed row last per code, and the
+    codes that produced no rows at all."""
+    wanted = sorted({c.upper() for c in codes})
+    if not wanted:
+        return [], []
+    by_code: dict[str, list[ValuationRow]] = {c: [] for c in wanted}
+    rule_years = sorted(RULES)
+
+    skipped = [
+        int(y)
+        for (y,) in con.execute(_UNKNOWN_YEARS_SQL, [wanted, rule_years]).fetchall()
+    ]
+    if skipped:
+        log.warning(
+            "valuation: %s priced in pfs.rvu year(s) %s with no pfs.rules entry "
+            "— skipped (no conversion factor)",
+            ", ".join(wanted),
+            ", ".join(str(y) for y in skipped),
+        )
+
+    for hcpcs, year, desc, status, work, pe_nf, pe_f, mp, tot_nf, tot_f in con.execute(
+        _FINAL_SQL, [wanted, rule_years, years]
+    ).fetchall():
+        rule = RULES[int(year)]
+        cf = rule.conversion_factor
+        cite = rule.federal_register_citation
+        by_code[hcpcs].append(
+            ValuationRow(
+                code=hcpcs,
+                description=desc or "",
+                vintage=f"CY{year} final",
+                year=int(year),
+                proposed=False,
+                status=status or "",
+                status_note=status_note(status or ""),
+                work=work,
+                pe_nf=pe_nf,
+                pe_f=pe_f,
+                mp=mp,
+                total_nf=tot_nf,
+                total_f=tot_f,
+                cf=cf,
+                pay_nf=_pay(tot_nf, cf, status or ""),
+                pay_f=_pay(tot_f, cf, status or ""),
+                cf_note=final_cf_note(rule),
+                label=vintage_label(int(year), False),
+                citation=cite,
+                url=fr_citation_url(cite),
+            )
+        )
+
+    nprm_year, _tag, rule_id, _pin = max(NPRM_SOURCES, key=lambda s: s[0])
+    prop = proposed_for(nprm_year)
+    if prop is None:
+        log.warning(
+            "valuation: no ProposedRule for CY%s — the newest NPRM vintage is "
+            "omitted from the table",
+            nprm_year,
+        )
+    else:
+        cf = prop.conversion_factor
+        cite = prop.federal_register_citation
+        for hcpcs, desc, status, work, pe_nf, pe_f, mp in con.execute(
+            _PROPOSED_SQL, [rule_id, wanted]
+        ).fetchall():
+            tot_nf, tot_f = _sum(work, pe_nf, mp), _sum(work, pe_f, mp)
+            by_code[hcpcs].append(
+                ValuationRow(
+                    code=hcpcs,
+                    description=desc or "",
+                    vintage=f"CY{nprm_year} proposed",
+                    year=nprm_year,
+                    proposed=True,
+                    status=status or "",
+                    status_note=status_note(status or ""),
+                    work=work,
+                    pe_nf=pe_nf,
+                    pe_f=pe_f,
+                    mp=mp,
+                    total_nf=tot_nf,
+                    total_f=tot_f,
+                    cf=cf,
+                    pay_nf=_pay(tot_nf, cf, status or ""),
+                    pay_f=_pay(tot_f, cf, status or ""),
+                    cf_note=proposed_cf_note(prop),
+                    label=vintage_label(nprm_year, True),
+                    citation=cite,
+                    url=fr_citation_url(cite),
+                )
+            )
+
+    rows = [
+        r
+        for c in wanted
+        for r in sorted(by_code[c], key=lambda r: (r.year, r.proposed))
+    ]
+    unpriced = [c for c in wanted if not by_code[c]]
+    return rows, unpriced

stack.toml

@@ -124,6 +124,9 @@ pg_user = "llm"
 recency_half_life_days = 365
 recency_weight = 0.3
 top_n = 8
+duckdb_replica = "data/replica/aco.ro.duckdb"   # read-only DuckDB replica the chat reads valuations from
+valuation_years = 4                      # final-rule vintages shown per code (plus the newest NPRM)
+code_cited_per_code = 3                  # rule paragraphs literally citing each detected code
 
 [llm.k_per_kind]                       # over-fetched ×3 per kind, then re-ranked
 comment = 8

tests/conf/test_connect.py

@@ -289,6 +289,8 @@ class TestReplica:
 
         dst = connect.publish_replica("custom")
         assert dst.name == "custom.ro.duckdb"
+        # its own directory, so the container mount cannot see the primary
+        assert dst.parent == db_file.parent / "replica"
         import duckdb as _duckdb
 
         con = _duckdb.connect(str(dst), read_only=True)
@@ -344,17 +346,16 @@ class TestReplica:
         self._patch_path(monkeypatch, db_file)
 
         target = shared / "custom.ro.duckdb"
+        link = tmp_path / "replica" / "custom.ro.duckdb"
         connect.publish_replica("custom")  # seed a real file to link to
-        (tmp_path / "custom.ro.duckdb").rename(target)
-        (tmp_path / "custom.ro.duckdb").symlink_to(target)
+        link.rename(target)
+        link.symlink_to(target)
 
         with connect.duckdb_batch("custom") as con:
             con.execute("UPDATE t SET x = 7")
         dst = connect.publish_replica("custom")
 
-        assert (tmp_path / "custom.ro.duckdb").is_symlink(), (
-            "publish detached the symlink into a standalone copy"
-        )
+        assert link.is_symlink(), "publish detached the symlink into a standalone copy"
         import duckdb as _duckdb
 
         con = _duckdb.connect(str(target), read_only=True)

tests/llm/test_api.py

@@ -24,6 +24,22 @@ class TestIndex:
         assert "Library Chat" in r.text
         assert 'id="form"' in r.text
 
+    def test_serves_chat_page_with_valuation_renderer(self):
+        r = client.get("/")
+        assert r.status_code == 200
+        html = r.text
+        assert "function renderValuation" in html
+        assert "ev.type === 'valuation'" in html
+        assert "table.valuation" in html
+        # the money on screen is national and unadjusted — say so
+        assert "no geographic (GPCI) adjustment" in html
+        assert "createCaption" in html
+        # notes explain an unpaid status and which CF is shown
+        assert "r.status_note" in html and "r.cf_note" in html
+        assert "'cf', 'CF (non-APM)'" in html
+        # a citation without a URL is not an empty <a href="">
+        assert "if (p.url)" in html
+
 
 class TestWhoami:
     def test_reads_forwarded_header(self):

tests/llm/test_chunk.py

@@ -151,3 +151,36 @@ class TestParagraphChunks:
         a = chunk_doc(Doc(key="R", text="same", metadata={}, paragraphs=paras))
         b = chunk_doc(Doc(key="R", text="same", metadata={}, paragraphs=paras))
         assert [c.id for c in a] == [c.id for c in b]
+
+
+class TestCodesMetadata:
+    def test_codes_stamped_per_chunk(self):
+        text = "We propose G0556 and G0557.\n\n## Other\n\nNo codes here.\n\n## More\n\n99490 applies."
+        chunks = chunk_doc(_doc(text), target_chars=60, overlap_chars=10)
+        found = {c.metadata["codes"] for c in chunks}
+        assert "G0556 G0557" in found
+        assert "99490" in found
+        assert "" in found
+
+    def test_heading_codes_stamped_on_every_piece_of_the_section(self):
+        """A section headed by the code it discusses ("## G0556 — APCM")
+        splits into several chunks; the ones past the first never repeat
+        the code, and used to be unfindable by it."""
+        text = "## G0556 — APCM\n\n" + "We propose to value the service. " * 20
+        chunks = chunk_doc(_doc(text), target_chars=120, overlap_chars=10)
+        assert len(chunks) > 1
+        assert all(c.metadata["codes"] == "G0556" for c in chunks)
+
+    def test_rule_paragraph_chunks_get_codes(self):
+        from llm.chunk import Paragraph
+
+        doc = Doc(
+            key="R1",
+            text="x",
+            metadata={"kind": "rule"},
+            paragraphs=(
+                Paragraph(1, 100, 1, "APCM code G0556 is valued at 0.25 work RVUs."),
+            ),
+        )
+        (chunk,) = chunk_doc(doc)
+        assert chunk.metadata["codes"] == "G0556"

tests/llm/test_config.py

@@ -87,3 +87,14 @@ class TestNewKnobs:
         cfg = llm_config.load()
         assert cfg.ollama_hosts == ("http://rig:11434", "http://laptop:11434")
         assert cfg.host_vram == {"http://rig:11434": 24.0, "http://laptop:11434": 12.0}
+
+    def test_valuation_knobs_defaults(self, monkeypatch):
+        monkeypatch.delenv("LLM_DUCKDB_REPLICA", raising=False)
+        cfg = llm_config.load()
+        assert cfg.duckdb_replica == "data/replica/aco.ro.duckdb"
+        assert cfg.valuation_years == 4
+        assert cfg.code_cited_per_code == 3
+
+    def test_duckdb_replica_env_override(self, monkeypatch):
+        monkeypatch.setenv("LLM_DUCKDB_REPLICA", "/app/data/replica/aco.ro.duckdb")
+        assert llm_config.load().duckdb_replica == "/app/data/replica/aco.ro.duckdb"

tests/llm/test_evidence.py

@@ -0,0 +1,404 @@
+"""llm.evidence — valuation prompt block, SSE payload, code-cited rule sources."""
+
+from __future__ import annotations
+
+from concurrent.futures import ThreadPoolExecutor
+from dataclasses import replace
+from unittest.mock import MagicMock, patch
+
+import duckdb
+import pytest
+
+from llm import evidence
+from llm.config import LlmConfig
+from llm.evidence import (
+    ValuationEvidence,
+    code_cited_sources,
+    merge_sources,
+    valuation_evidence,
+)
+from pfs.valuation import ValuationRow
+
+CFG = LlmConfig(
+    ollama_hosts=("http://h1:11434",),
+    embed_model="e",
+    instruct_model="c",
+    embed_dim=768,
+    build_ann_index=False,
+    pg_host="x",
+    pg_port=5432,
+    pg_db="llm",
+    pg_user="llm",
+    duckdb_replica="/nonexistent/aco.ro.duckdb",
+    valuation_years=4,
+    code_cited_per_code=2,
+)
+
+
+@pytest.fixture(autouse=True)
+def _clear_replica_cache():
+    """The replica handle is process-global; no test may inherit another's."""
+    evidence._REPLICA = None
+    yield
+    evidence._REPLICA = None
+
+
+def _row(**kw) -> ValuationRow:
+    base = dict(
+        code="G0556",
+        description="Adv prim care mgmt lvl 1",
+        vintage="CY2026 final",
+        year=2026,
+        proposed=False,
+        status="A",
+        work=0.25,
+        pe_nf=0.22,
+        pe_f=0.06,
+        mp=0.02,
+        total_nf=0.49,
+        total_f=0.33,
+        cf=33.4009,
+        pay_nf=16.37,
+        pay_f=11.02,
+        label="[PFS CY2026 Addendum B]",
+        citation="90 FR 49266",
+        url="https://www.federalregister.gov/citation/90-FR-49266",
+    )
+    base.update(kw)
+    return ValuationRow(**base)
+
+
+class TestEvidenceText:
+    def test_prompt_block_exact(self):
+        ev = ValuationEvidence(
+            codes=("G0556",), families=("APCM",), rows=(_row(),), unpriced=("Z9999",)
+        )
+        assert ev.prompt_block() == (
+            "Valuation (authoritative for RVUs, conversion factors and payments; "
+            "national unadjusted, no GPCI):\n"
+            '[PFS CY2026 Addendum B] G0556 "Adv prim care mgmt lvl 1" — status A; work 0.25; '
+            "PE 0.22 non-fac / 0.06 fac; MP 0.02; total 0.49 non-fac / 0.33 fac; "
+            "CF $33.4009 → payment $16.37 non-fac / $11.02 fac\n"
+            "Not priced in the indexed fee schedule: Z9999"
+        )
+
+    def test_prompt_block_labels_the_cf_and_an_unpaid_status(self):
+        ev = ValuationEvidence(
+            ("G0559",),
+            (),
+            (
+                _row(
+                    code="G0559",
+                    status="B",
+                    status_note="bundled — no separate PFS payment",
+                    cf_note="non-APM standard CF; QP CF $33.5675",
+                    pay_nf=None,
+                    pay_f=None,
+                ),
+            ),
+            (),
+        )
+        line = ev.prompt_block().splitlines()[1]
+        assert "— status B; bundled — no separate PFS payment; work" in line
+        assert "CF $33.4009 [non-APM standard CF; QP CF $33.5675] → payment" in line
+        assert "payment n/a non-fac / n/a fac" in line
+
+    def test_prompt_block_null_components(self):
+        ev = ValuationEvidence(
+            ("G9999",),
+            (),
+            (
+                _row(
+                    code="G9999",
+                    status="I",
+                    work=None,
+                    pe_nf=None,
+                    pe_f=None,
+                    mp=None,
+                    total_nf=None,
+                    total_f=None,
+                    pay_nf=None,
+                    pay_f=None,
+                ),
+            ),
+            (),
+        )
+        line = ev.prompt_block().splitlines()[1]
+        assert (
+            "status I; work n/a; PE n/a non-fac / n/a fac" in line
+            and "payment n/a" in line
+        )
+
+    def test_payload_shape(self):
+        ev = ValuationEvidence(
+            ("G0556",),
+            ("APCM",),
+            (
+                _row(),
+                _row(
+                    year=2027,
+                    proposed=True,
+                    vintage="CY2027 proposed",
+                    label="[CY2027 NPRM Addendum B]",
+                    citation="91 FR 43842",
+                    url="https://www.federalregister.gov/citation/91-FR-43842",
+                ),
+            ),
+            (),
+        )
+        p = ev.payload()
+        assert (
+            p["type"] == "valuation"
+            and p["codes"] == ["G0556"]
+            and p["families"] == ["APCM"]
+        )
+        assert p["rows"][0]["pay_nf"] == 16.37 and p["rows"][1]["proposed"] is True
+        assert p["rows"][0]["status_note"] == "" and p["rows"][0]["cf_note"] == ""
+        assert p["provenance"] == [
+            {
+                "label": "[PFS CY2026 Addendum B]",
+                "vintage": "CY2026 final",
+                "citation": "90 FR 49266",
+                "url": "https://www.federalregister.gov/citation/90-FR-49266",
+            },
+            {
+                "label": "[CY2027 NPRM Addendum B]",
+                "vintage": "CY2027 proposed",
+                "citation": "91 FR 43842",
+                "url": "https://www.federalregister.gov/citation/91-FR-43842",
+            },
+        ]
+        assert p["unpriced"] == []
+
+
+class TestValuationEvidence:
+    def test_none_when_no_codes(self):
+        assert (
+            valuation_evidence("what did commenters say about telehealth?", CFG) is None
+        )
+
+    @patch("llm.evidence.valuation", return_value=([_row()], []))
+    @patch("llm.evidence.duckdb.connect")
+    def test_detects_and_queries_replica_read_only(self, mock_connect, mock_val):
+        ev = valuation_evidence("How is APCM valued?", CFG)
+        mock_connect.assert_called_once_with(
+            "/nonexistent/aco.ro.duckdb", read_only=True
+        )
+        assert mock_val.call_args.args[1] == ["G0556", "G0557", "G0558"]
+        assert mock_val.call_args.kwargs == {"years": 4}
+        assert (
+            ev is not None
+            and ev.codes == ("G0556", "G0557", "G0558")
+            and ev.families == ("APCM",)
+        )
+        mock_connect.return_value.close.assert_not_called()  # cached, not closed
+
+    @patch("llm.evidence.duckdb.connect", side_effect=OSError("no replica"))
+    def test_unopenable_replica_yields_none(self, _c, caplog):
+        assert valuation_evidence("G0556?", CFG) is None
+        assert "valuation evidence skipped" in caplog.text
+
+    @patch("llm.evidence.valuation", return_value=([_row()], []))
+    @patch("llm.evidence.duckdb.connect")
+    def test_replica_connection_is_reused(self, mock_connect, _val):
+        valuation_evidence("G0556?", CFG)
+        valuation_evidence("G0557?", CFG)
+        mock_connect.assert_called_once()
+        mock_connect.return_value.close.assert_not_called()
+
+    @patch("llm.evidence.valuation", return_value=([_row()], []))
+    @patch("llm.evidence.duckdb.connect")
+    def test_republished_replica_reopens_and_closes_the_old_handle(
+        self, mock_connect, _val, monkeypatch
+    ):
+        first, second = MagicMock(), MagicMock()
+        mock_connect.side_effect = [first, second]
+        mtimes = iter([1, 2])
+        monkeypatch.setattr(evidence, "_mtime", lambda _p: next(mtimes))
+        valuation_evidence("G0556?", CFG)
+        valuation_evidence("G0557?", CFG)
+        assert mock_connect.call_count == 2
+        first.close.assert_called_once()
+        second.close.assert_not_called()
+
+    @patch("llm.evidence.valuation", return_value=([], []))
+    @patch("llm.evidence.duckdb.connect")
+    def test_many_codes_shorten_the_window(self, _c, mock_val):
+        valuation_evidence("How is CCM valued?", CFG)  # six-code family
+        assert mock_val.call_args.kwargs == {"years": 2}
+
+
+RVU_COLS = (
+    "hcpcs VARCHAR, mod VARCHAR, description VARCHAR, status_code VARCHAR, "
+    "work_rvu DOUBLE, non_fac_pe_rvu DOUBLE, fac_pe_rvu DOUBLE, mp_rvu DOUBLE, "
+    "non_fac_total DOUBLE, fac_total DOUBLE, conv_factor DOUBLE, year INTEGER"
+)
+PROPOSED_COLS = (
+    "hcpcs VARCHAR, mod VARCHAR, description VARCHAR, status_code VARCHAR, "
+    "work_rvu DOUBLE, non_fac_pe_rvu DOUBLE, fac_pe_rvu DOUBLE, mp_rvu DOUBLE, "
+    "cms_rule_id VARCHAR"
+)
+
+
+class TestConcurrentChats:
+    """/chat streams run in Starlette's threadpool, so several turns share
+    the cached replica handle. Without a per-call cursor their result sets
+    cross-talk and the broad except swallows it as "no valuation"."""
+
+    @pytest.fixture
+    def replica(self, tmp_path):
+        db = tmp_path / "aco.ro.duckdb"
+        con = duckdb.connect(str(db))
+        con.execute("CREATE SCHEMA pfs")
+        con.execute(f"CREATE TABLE pfs.rvu ({RVU_COLS})")
+        con.execute(f"CREATE TABLE pfs.rvu_proposed ({PROPOSED_COLS})")
+        # two unfamilied codes with different row counts, so a swapped
+        # result set is visible as a wrong count
+        con.executemany(
+            "INSERT INTO pfs.rvu VALUES (?,?,?,?,?,?,?,?,?,?,?,?)",
+            [
+                (
+                    "99213",
+                    None,
+                    "Office visit",
+                    "A",
+                    1.0,
+                    1.0,
+                    0.5,
+                    0.05,
+                    2.05,
+                    1.55,
+                    30.0,
+                    2026,
+                ),
+                *[
+                    (
+                        "99214",
+                        None,
+                        "Office visit",
+                        "A",
+                        1.5,
+                        1.0,
+                        0.5,
+                        0.05,
+                        2.55,
+                        2.05,
+                        30.0,
+                        y,
+                    )
+                    for y in (2024, 2025, 2026)
+                ],
+            ],
+        )
+        con.close()
+        return db
+
+    def test_parallel_turns_share_one_handle_without_cross_talk(self, replica):
+        cfg = replace(CFG, duckdb_replica=str(replica))
+        real_connect, opens = duckdb.connect, []
+
+        def counting_connect(*a, **kw):
+            opens.append(a)
+            return real_connect(*a, **kw)
+
+        expected = {"99213": 1, "99214": 3}
+        questions = ["What does 99213 pay?", "What does 99214 pay?"]
+
+        def one(i):
+            q = questions[i % 2]
+            ev = valuation_evidence(q, cfg)
+            assert ev is not None, "valuation dropped under concurrency"
+            (code,) = ev.codes
+            return code, len(ev.rows)
+
+        with patch("llm.evidence.duckdb.connect", counting_connect):
+            with ThreadPoolExecutor(max_workers=8) as pool:
+                results = list(pool.map(one, range(160)))
+
+        assert all(n == expected[code] for code, n in results), sorted(set(results))
+        assert len(opens) == 1  # one handle, shared
+
+    def test_cursor_is_closed_but_the_handle_is_kept(self, replica):
+        cfg = replace(CFG, duckdb_replica=str(replica))
+        assert valuation_evidence("99213?", cfg) is not None
+        assert evidence._REPLICA is not None
+        # the cached parent still answers — it was never closed
+        assert valuation_evidence("99214?", cfg) is not None
+
+
+def _engine(rows):
+    engine = MagicMock()
+    conn = engine.begin.return_value.__enter__.return_value
+    conn.execute.return_value.fetchall.return_value = rows
+    return engine, conn
+
+
+def _rule_md(p_id, codes, date="2024-12-09"):
+    return {
+        "kind": "rule",
+        "item_key": "R1",
+        "p_id": str(p_id),
+        "page": "97710",
+        "ordinal": "1",
+        "html_url": "https://www.federalregister.gov/d/2024-25382",
+        "fr_volume": "89",
+        "date": date,
+        "title": "CY2025 PFS final rule",
+        "codes": codes,
+    }
+
+
+class TestCodeCitedSources:
+    def test_sql_and_per_code_cap_and_dedupe(self):
+        engine, conn = _engine(
+            [
+                ("para one G0556", _rule_md(1, "G0556")),
+                ("para two G0556", _rule_md(2, "G0556")),
+                ("para three G0556", _rule_md(3, "G0556")),  # over the cap for G0556
+                (
+                    "para four G0557",
+                    _rule_md(4, "G0557 G0556"),
+                ),  # G0557 still under cap
+                ("para one again", _rule_md(1, "G0556")),  # duplicate (item_key, p_id)
+            ]
+        )
+        out = code_cited_sources(engine, ["G0556", "G0557"], per_code=2)
+        sql = str(conn.execute.call_args.args[0])
+        assert "string_to_array" in sql and "langchain_pg_collection" in sql
+        assert "PARTITION BY" in sql
+        # the collection is a scalar subquery, not a join to filter after
+        assert "JOIN langchain_pg_collection" not in sql
+        assert "e.collection_id = (SELECT uuid FROM langchain_pg_collection" in sql
+        assert conn.execute.call_args.args[1]["collection"] == "rules"
+        assert conn.execute.call_args.args[1]["codes"] == ["G0556", "G0557"]
+        assert conn.execute.call_args.args[1]["window"] == 4
+        assert [s["snippet"] for s in out] == [
+            "para one G0556",
+            "para two G0556",
+            "para four G0557",
+        ]
+        assert all(s["kind"] == "rule" and s["score"] == 0.0 for s in out)
+
+    def test_snippets_are_capped(self):
+        engine, _ = _engine([("x" * 900, _rule_md(1, "G0556"))])
+        (out,) = code_cited_sources(engine, ["G0556"], per_code=2)
+        assert len(out["snippet"]) == 250
+
+    def test_engine_error_yields_empty(self, caplog):
+        engine = MagicMock()
+        engine.begin.side_effect = RuntimeError("pg down")
+        assert code_cited_sources(engine, ["G0556"], per_code=2) == []
+        assert "code-cited sources skipped" in caplog.text
+
+    def test_no_codes_returns_empty_without_querying(self):
+        engine = MagicMock()
+        assert code_cited_sources(engine, [], per_code=2) == []
+        engine.begin.assert_not_called()
+
+
+class TestMergeSources:
+    def test_keeps_order_and_dedupes_by_label(self):
+        a = [{"label": "X", "score": 0.9}, {"label": "Y", "score": 0.8}]
+        b = [{"label": "Y", "score": 0.0}, {"label": "Z", "score": 0.0}]
+        assert [s["label"] for s in merge_sources(a, b)] == ["X", "Y", "Z"]
+        assert merge_sources(a, b)[1]["score"] == 0.8

tests/llm/test_links.py

@@ -127,3 +127,34 @@ class TestCorpus:
 def test_unknown_kind_treated_as_corpus():
     url, label = for_source({"url": "https://u.test", "title": "T"}, "s")
     assert (url, label) == ("https://u.test", "T")
+
+
+class TestAsSource:
+    def test_shape_matches_rag_source(self):
+        from llm.links import as_source
+
+        s = as_source(
+            {
+                "kind": "comment",
+                "comment_id": "CMS-2026-2377-1",
+                "date": "2026-08-19",
+                "title": "t",
+                "docket": "CMS-2026-2377",
+            },
+            "body " * 200,
+            0.123456,
+        )
+        assert set(s) == {
+            "id",
+            "label",
+            "kind",
+            "url",
+            "title",
+            "date",
+            "docket",
+            "comment_id",
+            "snippet",
+            "score",
+        }
+        assert s["label"] == "CMS-2026-2377-1" and s["kind"] == "comment"
+        assert len(s["snippet"]) <= 500 and s["score"] == 0.1235

tests/llm/test_migrate.py

@@ -24,6 +24,15 @@ class TestDdl:
         executed = " ".join(str(call.args[0]) for call in conn.execute.call_args_list)
         assert "vector(768)" in executed
         assert "USING hnsw" in executed
+        # code-cited lookups scan cmetadata->>'codes' inside one collection
+        assert (
+            "ix_embedding_codes" in executed
+            and "USING gin (string_to_array" in executed
+        )
+        assert (
+            "ix_embedding_collection" in executed
+            and "langchain_pg_embedding (collection_id)" in executed
+        )
 
     def test_fingerprint_column_and_docket_state(self):
         assert "fingerprint" in migrate.INDEX_STATE_DDL

tests/llm/test_rag.py

@@ -247,6 +247,46 @@ class TestBuildMessages:
         msgs = build_messages("q", [])
         assert "no relevant excerpts" in msgs[1]["content"].lower()
 
+    def test_evidence_block_between_excerpts_and_question(self):
+        from llm.evidence import ValuationEvidence
+        from pfs.valuation import ValuationRow
+
+        row = ValuationRow(
+            code="G0556",
+            description="d",
+            vintage="CY2026 final",
+            year=2026,
+            proposed=False,
+            status="A",
+            work=0.25,
+            pe_nf=0.22,
+            pe_f=0.06,
+            mp=0.02,
+            total_nf=0.49,
+            total_f=0.33,
+            cf=33.4009,
+            pay_nf=16.37,
+            pay_f=11.02,
+            label="[PFS CY2026 Addendum B]",
+            citation="90 FR 49266",
+            url="u",
+        )
+        ev = ValuationEvidence(("G0556",), ("APCM",), (row,), ())
+        msgs = build_messages("how much?", [], evidence=ev)
+        user = msgs[1]["content"]
+        assert (
+            user.index("Excerpts:")
+            < user.index("Valuation (authoritative")
+            < user.index("Question: how much?")
+        )
+        assert "[PFS CY2026 Addendum B] G0556" in user
+        assert "valuation" in msgs[0]["content"].lower()
+        assert "do not compute" in msgs[0]["content"].lower()
+
+    def test_no_evidence_prompt_unchanged(self):
+        assert build_messages("q", []) == build_messages("q", [], evidence=None)
+        assert "Valuation" not in build_messages("q", [])[1]["content"]
+
 
 class TestStreamAnswer:
     def _pool(self, vram=24.0, serves_big=True):
@@ -256,9 +296,13 @@ class TestStreamAnswer:
         pool.serves.return_value = serves_big
         return pool
 
+    @patch("llm.rag._engine")
+    @patch("llm.rag.valuation_evidence", return_value=None)
     @patch("llm.rag.httpx.Client")
     @patch("llm.rag.retrieve")
-    def test_yields_tokens_then_sources_then_done(self, mock_retrieve, MockClient):
+    def test_yields_tokens_then_sources_then_done(
+        self, mock_retrieve, MockClient, _ev, mock_engine
+    ):
         src = {
             "id": "C1",
             "label": "C1",
@@ -281,6 +325,8 @@ class TestStreamAnswer:
         events = list(stream_answer("q", cfg=CFG, pool=pool))
 
         pool.check.assert_called_once_with("chat")
+        # no codes in the question — pgvector is never asked for cited rules
+        mock_engine.assert_not_called()
         mock_retrieve.assert_called_once_with("q", cfg=CFG, pool=pool, since="")
         assert events[0] == {"type": "token", "text": "Doc"}
         assert events[1] == {"type": "token", "text": "tors"}
@@ -296,9 +342,10 @@ class TestStreamAnswer:
         assert body["options"] == {"num_ctx": 8192}
         assert client.stream.call_args.args[1] == "http://h1:11434/api/chat"
 
+    @patch("llm.rag.valuation_evidence", return_value=None)
     @patch("llm.rag.httpx.Client")
     @patch("llm.rag.retrieve")
-    def test_small_host_uses_baseline_model(self, mock_retrieve, MockClient):
+    def test_small_host_uses_baseline_model(self, mock_retrieve, MockClient, _ev):
         mock_retrieve.return_value = []
         client = MockClient.return_value.__enter__.return_value
         resp = client.stream.return_value.__enter__.return_value
@@ -307,9 +354,10 @@ class TestStreamAnswer:
         assert client.stream.call_args.kwargs["json"]["model"] == "chat"
         assert events[-2]["model"] == "chat"
 
+    @patch("llm.rag.valuation_evidence", return_value=None)
     @patch("llm.rag.httpx.Client")
     @patch("llm.rag.retrieve")
-    def test_since_forwarded(self, mock_retrieve, MockClient):
+    def test_since_forwarded(self, mock_retrieve, MockClient, _ev):
         mock_retrieve.return_value = []
         client = MockClient.return_value.__enter__.return_value
         resp = client.stream.return_value.__enter__.return_value
@@ -317,12 +365,55 @@ class TestStreamAnswer:
         list(stream_answer("q", cfg=CFG, pool=self._pool(), since="2025-09-01"))
         assert mock_retrieve.call_args.kwargs["since"] == "2025-09-01"
 
+    @patch("llm.rag.valuation_evidence", return_value=None)
     @patch("llm.rag.httpx.Client")
     @patch("llm.rag.retrieve")
-    def test_http_error_propagates(self, mock_retrieve, MockClient):
+    def test_http_error_propagates(self, mock_retrieve, MockClient, _ev):
         mock_retrieve.return_value = []
         client = MockClient.return_value.__enter__.return_value
         resp = client.stream.return_value.__enter__.return_value
         resp.raise_for_status.side_effect = RuntimeError("ollama down")
         with pytest.raises(RuntimeError, match="ollama down"):
             list(stream_answer("q", cfg=CFG, pool=self._pool()))
+
+    @patch("llm.rag._engine")
+    @patch("llm.rag.code_cited_sources")
+    @patch("llm.rag.valuation_evidence")
+    @patch("llm.rag.httpx.Client")
+    @patch("llm.rag.retrieve")
+    def test_valuation_event_first_and_sources_merged(
+        self, mock_retrieve, MockClient, mock_ev, mock_cited, mock_engine
+    ):
+        from llm.evidence import ValuationEvidence
+
+        src = {
+            "id": "C1",
+            "label": "C1",
+            "kind": "comment",
+            "snippet": "s",
+            "score": 0.1,
+        }
+        cited = {
+            "id": "89 FR 97710 ¶3",
+            "label": "89 FR 97710 ¶3",
+            "kind": "rule",
+            "snippet": "G0556",
+            "score": 0.0,
+        }
+        mock_retrieve.return_value = [src]
+        mock_cited.return_value = [cited]
+        mock_ev.return_value = ValuationEvidence(("G0556",), ("APCM",), (), ())
+        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}'])
+
+        events = list(stream_answer("APCM?", cfg=CFG, pool=self._pool()))
+
+        assert events[0]["type"] == "valuation" and events[0]["codes"] == ["G0556"]
+        assert events[1] == {"type": "token", "text": "x"}
+        assert events[-2]["sources"] == [src, cited]
+        mock_cited.assert_called_once_with(
+            mock_engine.return_value, ("G0556",), per_code=CFG.code_cited_per_code
+        )
+        body = client.stream.call_args.kwargs["json"]
+        assert "Valuation (authoritative" in body["messages"][1]["content"]

tests/pfs/test_families.py

@@ -0,0 +1,76 @@
+"""pfs.families — code-family registry + deterministic code detection."""
+
+from __future__ import annotations
+
+from pfs.families import FAMILIES, Detection, detect_codes, family_of, find_codes
+
+
+class TestFindCodes:
+    def test_hcpcs_and_cpt(self):
+        assert find_codes("Codes G0556 and 99490 apply; see g0557.") == (
+            "99490",
+            "G0556",
+            "G0557",
+        )
+
+    def test_lower_case_codes_are_normalised(self):
+        assert find_codes("what is g0556?") == ("G0556",)
+
+    def test_no_false_hits_on_years_and_fr_pages(self):
+        # A Federal Register citation is stripped before scanning, so its page
+        # number (43842 — a well-formed CPT code) never reads as a code.
+        assert find_codes("91 FR 43842, CY2026") == ()
+
+    def test_fr_citation_alone_yields_nothing(self):
+        assert find_codes("What does 91 FR 43842 say?") == ()
+
+    def test_dedupes_and_sorts(self):
+        assert find_codes("G0558 G0556 G0558") == ("G0556", "G0558")
+
+    def test_empty(self):
+        assert find_codes("") == ()
+
+
+class TestRegistry:
+    def test_apcm_family(self):
+        f = FAMILIES["APCM"]
+        assert f.codes == ("G0556", "G0557", "G0558")
+        assert "advanced primary care management" in f.synonyms
+        assert family_of("G0557") is f
+        assert family_of("00000") is None
+
+    def test_all_families_present(self):
+        assert set(FAMILIES) == {"ACP", "CCM", "PCM", "TCM", "APCM"}
+
+
+class TestDetectCodes:
+    def test_family_name_expands_to_codes(self):
+        d = detect_codes("What is APCM and how is it valued?")
+        assert d == Detection(
+            codes=("G0556", "G0557", "G0558"), families=("APCM",), explicit=()
+        )
+
+    def test_synonym_case_insensitive_word_boundary(self):
+        d = detect_codes("Tell me about Advanced Primary Care Management.")
+        assert d.families == ("APCM",)
+        assert detect_codes("the apcmx code").families == ()
+
+    def test_single_member_code_expands_family(self):
+        d = detect_codes("How much does G0557 pay?")
+        assert d.explicit == ("G0557",)
+        assert d.families == ("APCM",)
+        assert d.codes == ("G0556", "G0557", "G0558")
+
+    def test_unregistered_code_stays_alone(self):
+        d = detect_codes("value of 99213")
+        assert d == Detection(codes=("99213",), families=(), explicit=("99213",))
+
+    def test_multiple_families(self):
+        d = detect_codes("compare CCM and TCM")
+        assert d.families == ("CCM", "TCM")
+        assert d.codes == tuple(sorted(FAMILIES["CCM"].codes + FAMILIES["TCM"].codes))
+
+    def test_nothing(self):
+        assert detect_codes("what did commenters say about telehealth?") == Detection(
+            (), (), ()
+        )

tests/pfs/test_valuation.py

@@ -0,0 +1,295 @@
+"""pfs.valuation — RVUs + national payment by vintage from a DuckDB replica."""
+
+from __future__ import annotations
+
+import duckdb
+import pytest
+
+from pfs.rules import RULES, proposed_for
+from pfs.valuation import ValuationRow, fr_citation_url, valuation, vintage_label
+
+RVU_COLS = (
+    "hcpcs VARCHAR, mod VARCHAR, description VARCHAR, status_code VARCHAR, "
+    "work_rvu DOUBLE, non_fac_pe_rvu DOUBLE, fac_pe_rvu DOUBLE, mp_rvu DOUBLE, "
+    "non_fac_total DOUBLE, fac_total DOUBLE, conv_factor DOUBLE, year INTEGER"
+)
+PROPOSED_COLS = (
+    "hcpcs VARCHAR, mod VARCHAR, description VARCHAR, status_code VARCHAR, "
+    "work_rvu DOUBLE, non_fac_pe_rvu DOUBLE, fac_pe_rvu DOUBLE, mp_rvu DOUBLE, cms_rule_id VARCHAR"
+)
+
+
+@pytest.fixture
+def con():
+    c = duckdb.connect(":memory:")
+    c.execute("CREATE SCHEMA pfs")
+    c.execute(f"CREATE TABLE pfs.rvu ({RVU_COLS})")
+    c.execute(f"CREATE TABLE pfs.rvu_proposed ({PROPOSED_COLS})")
+    rows = [
+        # G0556: 2025 + 2026 final, with a modifier row that must be ignored
+        (
+            "G0556",
+            None,
+            "Adv prim care mgmt lvl 1",
+            "A",
+            0.25,
+            0.20,
+            0.10,
+            0.02,
+            0.47,
+            0.37,
+            32.3465,
+            2025,
+        ),
+        (
+            "G0556",
+            None,
+            "Adv prim care mgmt lvl 1",
+            "A",
+            0.25,
+            0.22,
+            0.06,
+            0.02,
+            0.49,
+            0.33,
+            33.4009,
+            2026,
+        ),
+        (
+            "G0556",
+            "26",
+            "Adv prim care mgmt lvl 1",
+            "A",
+            9.0,
+            9.0,
+            9.0,
+            9.0,
+            27.0,
+            27.0,
+            33.4009,
+            2026,
+        ),
+        # 99490: six years so the last-N cut applies (2021..2026)
+        *[
+            (
+                "99490",
+                "",
+                "Chrnc care mgmt srvc 20 min",
+                "A",
+                1.0,
+                1.0,
+                0.5,
+                0.05,
+                2.05,
+                1.55,
+                30.0,
+                y,
+            )
+            for y in range(2021, 2027)
+        ],
+        # A vintage older than the first year pfs.rules knows (RULES starts
+        # at 2014): it has no CF, so it must not consume a window slot.
+        (
+            "G0556",
+            None,
+            "Adv prim care mgmt lvl 1",
+            "A",
+            0.25,
+            0.20,
+            0.10,
+            0.02,
+            0.47,
+            0.37,
+            30.0,
+            2013,
+        ),
+        # B-status code with full RVUs — bundled, so never paid
+        (
+            "G0559",
+            None,
+            "Bundled service",
+            "B",
+            0.30,
+            0.20,
+            0.10,
+            0.02,
+            0.52,
+            0.42,
+            33.4009,
+            2026,
+        ),
+        # I-status code with NULL components
+        (
+            "G9999",
+            None,
+            "Not priced",
+            "I",
+            None,
+            None,
+            None,
+            None,
+            None,
+            None,
+            33.4009,
+            2026,
+        ),
+    ]
+    c.executemany("INSERT INTO pfs.rvu VALUES (?,?,?,?,?,?,?,?,?,?,?,?)", rows)
+    c.executemany(
+        "INSERT INTO pfs.rvu_proposed VALUES (?,?,?,?,?,?,?,?,?)",
+        [
+            (
+                "G0556",
+                None,
+                "Adv prim care mgmt lvl 1",
+                "A",
+                0.25,
+                0.23,
+                0.07,
+                0.02,
+                "CMS-1848-P",
+            ),
+            (
+                "G0556",
+                "26",
+                "Adv prim care mgmt lvl 1",
+                "A",
+                9.0,
+                9.0,
+                9.0,
+                9.0,
+                "CMS-1848-P",
+            ),
+            ("G0556", None, "older nprm", "A", 0.1, 0.1, 0.1, 0.1, "CMS-1832-P"),
+        ],
+    )
+    yield c
+    c.close()
+
+
+class TestHelpers:
+    def test_vintage_label(self):
+        assert vintage_label(2026, False) == "[PFS CY2026 Addendum B]"
+        assert vintage_label(2027, True) == "[CY2027 NPRM Addendum B]"
+
+    def test_fr_citation_url(self):
+        assert (
+            fr_citation_url("90 FR 49266")
+            == "https://www.federalregister.gov/citation/90-FR-49266"
+        )
+        assert fr_citation_url("") == ""
+
+
+class TestValuation:
+    def test_final_and_proposed_rows_with_cf_and_payment(self, con):
+        rows, unpriced = valuation(con, ["G0556"], years=4)
+        assert unpriced == []
+        assert [(r.year, r.proposed) for r in rows] == [
+            (2025, False),
+            (2026, False),
+            (2027, True),
+        ]
+        r26 = rows[1]
+        assert isinstance(r26, ValuationRow)
+        assert (r26.code, r26.status, r26.work, r26.pe_nf, r26.pe_f, r26.mp) == (
+            "G0556",
+            "A",
+            0.25,
+            0.22,
+            0.06,
+            0.02,
+        )
+        assert (r26.total_nf, r26.total_f) == (0.49, 0.33)
+        assert r26.cf == RULES[2026].conversion_factor
+        assert r26.pay_nf == round(0.49 * RULES[2026].conversion_factor, 2)
+        assert r26.pay_f == round(0.33 * RULES[2026].conversion_factor, 2)
+        assert r26.label == "[PFS CY2026 Addendum B]"
+        assert r26.citation == RULES[2026].federal_register_citation
+        assert r26.url == fr_citation_url(RULES[2026].federal_register_citation)
+        assert r26.vintage == "CY2026 final"
+        p = rows[2]
+        assert p.label == "[CY2027 NPRM Addendum B]" and p.vintage == "CY2027 proposed"
+        assert p.cf == proposed_for(2027).conversion_factor
+        assert p.total_nf == round(0.25 + 0.23 + 0.02, 4) and p.total_f == round(
+            0.25 + 0.07 + 0.02, 4
+        )
+        assert p.pay_nf == round(p.total_nf * p.cf, 2)
+        assert p.citation == proposed_for(2027).federal_register_citation
+
+    def test_modifier_rows_are_ignored(self, con):
+        rows, _ = valuation(con, ["G0556"])
+        assert all(r.work == 0.25 for r in rows)
+
+    def test_last_n_years_only(self, con):
+        rows, _ = valuation(con, ["99490"], years=4)
+        assert [r.year for r in rows if not r.proposed] == [2023, 2024, 2025, 2026]
+
+    def test_unpriced_codes_reported(self, con):
+        rows, unpriced = valuation(con, ["G0556", "Z9999"])
+        assert unpriced == ["Z9999"]
+        assert {r.code for r in rows} == {"G0556"}
+
+    def test_null_components_pass_through(self, con):
+        rows, unpriced = valuation(con, ["G9999"])
+        assert unpriced == []
+        (r,) = rows
+        assert r.status == "I" and r.work is None and r.pay_nf is None
+
+    def test_bundled_status_is_not_paid_and_is_annotated(self, con):
+        rows, _ = valuation(con, ["G0559"])
+        (r,) = rows
+        assert (r.status, r.total_nf, r.total_f) == ("B", 0.52, 0.42)
+        assert r.pay_nf is None and r.pay_f is None
+        assert r.status_note == "bundled — no separate PFS payment"
+
+    def test_paid_statuses_have_no_note(self, con):
+        rows, _ = valuation(con, ["G0556"], years=1)
+        r = rows[0]
+        assert r.status == "A" and r.status_note == ""
+        assert r.pay_nf == round(r.total_nf * r.cf, 2)
+
+    def test_unknown_status_note_names_the_code(self, con):
+        con.execute(
+            "INSERT INTO pfs.rvu VALUES "
+            "('G0560',NULL,'Odd','Q',0.1,0.1,0.1,0.0,0.2,0.2,33.4009,2026)"
+        )
+        (r,) = valuation(con, ["G0560"])[0]
+        assert r.pay_nf is None
+        assert r.status_note == "status Q — no RVU-based payment"
+
+    def test_window_counts_only_years_pfs_rules_knows(self, con, caplog):
+        """A vintage with no RuleYear must not eat a slot in the last-N
+        window — otherwise asking for one year returns nothing at all."""
+        with caplog.at_level("WARNING"):
+            rows, _ = valuation(con, ["G0556"], years=1)
+        assert [(r.year, r.proposed) for r in rows] == [(2026, False), (2027, True)]
+        assert "2013" in caplog.text and "G0556" in caplog.text
+
+    def test_cf_note_final_year(self, con):
+        rows, _ = valuation(con, ["G0556"], years=1)
+        r = rows[0]
+        assert r.cf_note == (f"non-APM standard CF; QP CF ${RULES[2026].cf_qp:.4f}")
+
+    def test_cf_note_flags_a_mid_year_split(self, con):
+        rows, _ = valuation(con, ["99490"], years=3)
+        r24 = next(r for r in rows if r.year == 2024)
+        assert r24.cf_note.startswith("non-APM standard CF; mid-year split")
+        assert "$32.7442 before 2024-03-09" in r24.cf_note
+
+    def test_cf_note_proposed(self, con):
+        rows, _ = valuation(con, ["G0556"], years=1)
+        p = rows[-1]
+        assert p.proposed
+        assert p.cf_note == (
+            f"proposed; non-QP CF; QP CF ${proposed_for(2027).cf_qp:.4f}"
+        )
+
+    def test_order_is_code_then_year(self, con):
+        rows, _ = valuation(con, ["G0556", "99490"], years=2)
+        assert [(r.code, r.year) for r in rows] == [
+            ("99490", 2025),
+            ("99490", 2026),
+            ("G0556", 2025),
+            ("G0556", 2026),
+            ("G0556", 2027),
+        ]