diff --git a/.gitea/workflows/ci.yml b/.gitea/workflows/ci.yml index 83a68d9..2b56870 100644 --- a/.gitea/workflows/ci.yml +++ b/.gitea/workflows/ci.yml @@ -66,6 +66,11 @@ jobs: env: GITEA_TOKEN: ${{ secrets.DEPLOY_TOKEN }} run: | + # Jobs that never install uv (image builds, infra checks) still + # need it here — otherwise the filer dies with "uv: command not + # found" and the failure is never filed. + command -v uv >/dev/null 2>&1 || curl -LsSf https://astral.sh/uv/install.sh | sh + export PATH="$HOME/.local/bin:$PATH" uv sync --no-dev --quiet 2>/dev/null || true uv run python -m api.diag.ci \ --workflow "CI" --job "test" \ @@ -108,6 +113,11 @@ jobs: env: GITEA_TOKEN: ${{ secrets.DEPLOY_TOKEN }} run: | + # Jobs that never install uv (image builds, infra checks) still + # need it here — otherwise the filer dies with "uv: command not + # found" and the failure is never filed. + command -v uv >/dev/null 2>&1 || curl -LsSf https://astral.sh/uv/install.sh | sh + export PATH="$HOME/.local/bin:$PATH" uv sync --no-dev --quiet 2>/dev/null || true uv run python -m api.diag.ci \ --workflow "CI" --job "notebooks-smoke" \ diff --git a/.gitea/workflows/deploy.yml b/.gitea/workflows/deploy.yml index ee7ea44..d269c8e 100644 --- a/.gitea/workflows/deploy.yml +++ b/.gitea/workflows/deploy.yml @@ -50,6 +50,11 @@ jobs: env: GITEA_TOKEN: ${{ secrets.DEPLOY_TOKEN }} run: | + # Jobs that never install uv (image builds, infra checks) still + # need it here — otherwise the filer dies with "uv: command not + # found" and the failure is never filed. + command -v uv >/dev/null 2>&1 || curl -LsSf https://astral.sh/uv/install.sh | sh + export PATH="$HOME/.local/bin:$PATH" uv sync --no-dev --quiet 2>/dev/null || true uv run python -m api.diag.ci \ --workflow "Deploy" --job "notebooks" \ @@ -98,6 +103,11 @@ jobs: env: GITEA_TOKEN: ${{ secrets.DEPLOY_TOKEN }} run: | + # Jobs that never install uv (image builds, infra checks) still + # need it here — otherwise the filer dies with "uv: command not + # found" and the failure is never filed. + command -v uv >/dev/null 2>&1 || curl -LsSf https://astral.sh/uv/install.sh | sh + export PATH="$HOME/.local/bin:$PATH" uv sync --no-dev --quiet 2>/dev/null || true uv run python -m api.diag.ci \ --workflow "Deploy" --job "zotero" \ @@ -146,6 +156,11 @@ jobs: env: GITEA_TOKEN: ${{ secrets.DEPLOY_TOKEN }} run: | + # Jobs that never install uv (image builds, infra checks) still + # need it here — otherwise the filer dies with "uv: command not + # found" and the failure is never filed. + command -v uv >/dev/null 2>&1 || curl -LsSf https://astral.sh/uv/install.sh | sh + export PATH="$HOME/.local/bin:$PATH" uv sync --no-dev --quiet 2>/dev/null || true uv run python -m api.diag.ci \ --workflow "Deploy" --job "docs" \ @@ -195,6 +210,11 @@ jobs: env: GITEA_TOKEN: ${{ secrets.DEPLOY_TOKEN }} run: | + # Jobs that never install uv (image builds, infra checks) still + # need it here — otherwise the filer dies with "uv: command not + # found" and the failure is never filed. + command -v uv >/dev/null 2>&1 || curl -LsSf https://astral.sh/uv/install.sh | sh + export PATH="$HOME/.local/bin:$PATH" uv sync --no-dev --quiet 2>/dev/null || true uv run python -m api.diag.ci \ --workflow "Deploy" --job "api" \ @@ -244,6 +264,11 @@ jobs: env: GITEA_TOKEN: ${{ secrets.DEPLOY_TOKEN }} run: | + # Jobs that never install uv (image builds, infra checks) still + # need it here — otherwise the filer dies with "uv: command not + # found" and the failure is never filed. + command -v uv >/dev/null 2>&1 || curl -LsSf https://astral.sh/uv/install.sh | sh + export PATH="$HOME/.local/bin:$PATH" uv sync --no-dev --quiet 2>/dev/null || true uv run python -m api.diag.ci \ --workflow "Deploy" --job "llm" \ @@ -286,6 +311,11 @@ jobs: env: GITEA_TOKEN: ${{ secrets.DEPLOY_TOKEN }} run: | + # Jobs that never install uv (image builds, infra checks) still + # need it here — otherwise the filer dies with "uv: command not + # found" and the failure is never filed. + command -v uv >/dev/null 2>&1 || curl -LsSf https://astral.sh/uv/install.sh | sh + export PATH="$HOME/.local/bin:$PATH" uv sync --no-dev --quiet 2>/dev/null || true uv run python -m api.diag.ci \ --workflow "Deploy" --job "mc" \ diff --git a/.gitea/workflows/harden.yml b/.gitea/workflows/harden.yml index 65489ba..993a480 100644 --- a/.gitea/workflows/harden.yml +++ b/.gitea/workflows/harden.yml @@ -117,6 +117,11 @@ jobs: env: GITEA_TOKEN: ${{ secrets.DEPLOY_TOKEN }} run: | + # Jobs that never install uv (image builds, infra checks) still + # need it here — otherwise the filer dies with "uv: command not + # found" and the failure is never filed. + command -v uv >/dev/null 2>&1 || curl -LsSf https://astral.sh/uv/install.sh | sh + export PATH="$HOME/.local/bin:$PATH" uv sync --no-dev --quiet 2>/dev/null || true uv run python -m api.diag.ci \ --workflow "Harden" --job "build-scan-report" \ diff --git a/.gitea/workflows/infra-ci.yml b/.gitea/workflows/infra-ci.yml index de79b05..7d6464f 100644 --- a/.gitea/workflows/infra-ci.yml +++ b/.gitea/workflows/infra-ci.yml @@ -64,6 +64,11 @@ jobs: env: GITEA_TOKEN: ${{ secrets.DEPLOY_TOKEN }} run: | + # Jobs that never install uv (image builds, infra checks) still + # need it here — otherwise the filer dies with "uv: command not + # found" and the failure is never filed. + command -v uv >/dev/null 2>&1 || curl -LsSf https://astral.sh/uv/install.sh | sh + export PATH="$HOME/.local/bin:$PATH" uv sync --no-dev --quiet 2>/dev/null || true uv run python -m api.diag.ci \ --workflow "Infra CI" --job "notebooks" \ @@ -92,6 +97,11 @@ jobs: env: GITEA_TOKEN: ${{ secrets.DEPLOY_TOKEN }} run: | + # Jobs that never install uv (image builds, infra checks) still + # need it here — otherwise the filer dies with "uv: command not + # found" and the failure is never filed. + command -v uv >/dev/null 2>&1 || curl -LsSf https://astral.sh/uv/install.sh | sh + export PATH="$HOME/.local/bin:$PATH" uv sync --no-dev --quiet 2>/dev/null || true uv run python -m api.diag.ci \ --workflow "Infra CI" --job "zotero" \ @@ -120,6 +130,11 @@ jobs: env: GITEA_TOKEN: ${{ secrets.DEPLOY_TOKEN }} run: | + # Jobs that never install uv (image builds, infra checks) still + # need it here — otherwise the filer dies with "uv: command not + # found" and the failure is never filed. + command -v uv >/dev/null 2>&1 || curl -LsSf https://astral.sh/uv/install.sh | sh + export PATH="$HOME/.local/bin:$PATH" uv sync --no-dev --quiet 2>/dev/null || true uv run python -m api.diag.ci \ --workflow "Infra CI" --job "docs" \ @@ -148,6 +163,11 @@ jobs: env: GITEA_TOKEN: ${{ secrets.DEPLOY_TOKEN }} run: | + # Jobs that never install uv (image builds, infra checks) still + # need it here — otherwise the filer dies with "uv: command not + # found" and the failure is never filed. + command -v uv >/dev/null 2>&1 || curl -LsSf https://astral.sh/uv/install.sh | sh + export PATH="$HOME/.local/bin:$PATH" uv sync --no-dev --quiet 2>/dev/null || true uv run python -m api.diag.ci \ --workflow "Infra CI" --job "api" \ @@ -176,6 +196,11 @@ jobs: env: GITEA_TOKEN: ${{ secrets.DEPLOY_TOKEN }} run: | + # Jobs that never install uv (image builds, infra checks) still + # need it here — otherwise the filer dies with "uv: command not + # found" and the failure is never filed. + command -v uv >/dev/null 2>&1 || curl -LsSf https://astral.sh/uv/install.sh | sh + export PATH="$HOME/.local/bin:$PATH" uv sync --no-dev --quiet 2>/dev/null || true uv run python -m api.diag.ci \ --workflow "Infra CI" --job "llm" \ @@ -204,6 +229,11 @@ jobs: env: GITEA_TOKEN: ${{ secrets.DEPLOY_TOKEN }} run: | + # Jobs that never install uv (image builds, infra checks) still + # need it here — otherwise the filer dies with "uv: command not + # found" and the failure is never filed. + command -v uv >/dev/null 2>&1 || curl -LsSf https://astral.sh/uv/install.sh | sh + export PATH="$HOME/.local/bin:$PATH" uv sync --no-dev --quiet 2>/dev/null || true uv run python -m api.diag.ci \ --workflow "Infra CI" --job "mc" \ diff --git a/.gitea/workflows/llm-golden.yml b/.gitea/workflows/llm-golden.yml index 4a0e8d5..6d10662 100644 --- a/.gitea/workflows/llm-golden.yml +++ b/.gitea/workflows/llm-golden.yml @@ -54,6 +54,11 @@ jobs: env: GITEA_TOKEN: ${{ secrets.DEPLOY_TOKEN }} run: | + # Jobs that never install uv (image builds, infra checks) still + # need it here — otherwise the filer dies with "uv: command not + # found" and the failure is never filed. + command -v uv >/dev/null 2>&1 || curl -LsSf https://astral.sh/uv/install.sh | sh + export PATH="$HOME/.local/bin:$PATH" uv sync --no-dev --quiet 2>/dev/null || true uv run python -m api.diag.ci \ --workflow "LLM Golden" --job "llm-golden" \ diff --git a/.gitea/workflows/notebooks-integration.yml b/.gitea/workflows/notebooks-integration.yml index 2f76491..935eee6 100644 --- a/.gitea/workflows/notebooks-integration.yml +++ b/.gitea/workflows/notebooks-integration.yml @@ -49,6 +49,11 @@ jobs: env: GITEA_TOKEN: ${{ secrets.DEPLOY_TOKEN }} run: | + # Jobs that never install uv (image builds, infra checks) still + # need it here — otherwise the filer dies with "uv: command not + # found" and the failure is never filed. + command -v uv >/dev/null 2>&1 || curl -LsSf https://astral.sh/uv/install.sh | sh + export PATH="$HOME/.local/bin:$PATH" uv sync --no-dev --quiet 2>/dev/null || true uv run python -m api.diag.ci \ --workflow "Notebooks Integration" --job "notebooks-integration" \ diff --git a/.gitea/workflows/rebuild-all.yml b/.gitea/workflows/rebuild-all.yml index c3997ac..7eb6a93 100644 --- a/.gitea/workflows/rebuild-all.yml +++ b/.gitea/workflows/rebuild-all.yml @@ -115,6 +115,11 @@ jobs: env: GITEA_TOKEN: ${{ secrets.DEPLOY_TOKEN }} run: | + # Jobs that never install uv (image builds, infra checks) still + # need it here — otherwise the filer dies with "uv: command not + # found" and the failure is never filed. + command -v uv >/dev/null 2>&1 || curl -LsSf https://astral.sh/uv/install.sh | sh + export PATH="$HOME/.local/bin:$PATH" uv sync --no-dev --quiet 2>/dev/null || true uv run python -m api.diag.ci \ --workflow "Rebuild All" --job "build-scan-report" \ diff --git a/.gitea/workflows/release.yml b/.gitea/workflows/release.yml index 2488685..90e6ec1 100644 --- a/.gitea/workflows/release.yml +++ b/.gitea/workflows/release.yml @@ -34,6 +34,11 @@ jobs: env: GITEA_TOKEN: ${{ secrets.DEPLOY_TOKEN }} run: | + # Jobs that never install uv (image builds, infra checks) still + # need it here — otherwise the filer dies with "uv: command not + # found" and the failure is never filed. + command -v uv >/dev/null 2>&1 || curl -LsSf https://astral.sh/uv/install.sh | sh + export PATH="$HOME/.local/bin:$PATH" uv sync --no-dev --quiet 2>/dev/null || true uv run python -m api.diag.ci \ --workflow "Release" --job "release" \ diff --git a/.gitea/workflows/zotero-sync.yml b/.gitea/workflows/zotero-sync.yml index 0c71171..690ed2e 100644 --- a/.gitea/workflows/zotero-sync.yml +++ b/.gitea/workflows/zotero-sync.yml @@ -51,6 +51,11 @@ jobs: env: GITEA_TOKEN: ${{ secrets.DEPLOY_TOKEN }} run: | + # Jobs that never install uv (image builds, infra checks) still + # need it here — otherwise the filer dies with "uv: command not + # found" and the failure is never filed. + command -v uv >/dev/null 2>&1 || curl -LsSf https://astral.sh/uv/install.sh | sh + export PATH="$HOME/.local/bin:$PATH" uv sync --no-dev --quiet 2>/dev/null || true uv run python -m api.diag.ci \ --workflow "Zotero Sync" --job "zotero-sync" \ diff --git a/dev/scripts/backends/gitea.py b/dev/scripts/backends/gitea.py index ae10d4b..24ae924 100644 --- a/dev/scripts/backends/gitea.py +++ b/dev/scripts/backends/gitea.py @@ -109,6 +109,11 @@ def _failure_step(workflow_name: str, job_name: str) -> str: env: GITEA_TOKEN: ${{{{ secrets.DEPLOY_TOKEN }}}} run: | + # Jobs that never install uv (image builds, infra checks) still + # need it here — otherwise the filer dies with "uv: command not + # found" and the failure is never filed. + command -v uv >/dev/null 2>&1 || curl -LsSf https://astral.sh/uv/install.sh | sh + export PATH="$HOME/.local/bin:$PATH" uv sync --no-dev --quiet 2>/dev/null || true uv run python -m api.diag.ci \\ --workflow "{workflow_name}" --job "{job_name}" \\ diff --git a/infra/images/mc.Dockerfile b/infra/images/mc.Dockerfile index c31a6e1..78e7d37 100644 --- a/infra/images/mc.Dockerfile +++ b/infra/images/mc.Dockerfile @@ -1,7 +1,10 @@ # syntax=docker/dockerfile:1 # MinIO Client (mc) sidecar — least-privilege admin helper for RustFS. +# The image moved off Docker Hub (docker.io/minio/mc now 404s — pull access +# denied) to quay.io; pinned to a release tag so a base-image change is a +# reviewed diff, not a silent CI break. # Runs as non-root, read-only filesystem, no capabilities. -FROM minio/mc:latest AS mc +FROM quay.io/minio/mc:RELEASE.2025-08-13T08-35-41Z AS mc FROM alpine:3 RUN addgroup -g 10002 -S mc && adduser -u 10002 -S mc -G mc diff --git a/notebooks/llm_search.py b/notebooks/llm_search.py new file mode 100644 index 0000000..23077c7 --- /dev/null +++ b/notebooks/llm_search.py @@ -0,0 +1,215 @@ +import marimo + +__generated_with = "0.23.13" +app = marimo.App(width="medium") + + +@app.cell(hide_code=True) +def _(): + import marimo as mo + + return (mo,) + + +@app.cell(hide_code=True) +def _(mo): + mo.md(""" + # Semantic search over a docket + + A notebook-side look at what `LlmClient` (`llm.client`, #573) gives you: the + same pgvector similarity search the chat UI's `GET /search` endpoint runs, + called directly from a marimo cell over plain httpx — no `langchain`, + `sqlalchemy` or `psycopg` required, which matters because the notebooks + container installs the workspace without the `llm` extra (#720). Pick a + docket and a query below; the semantic-search hits sit beside the DuckDB + replica's own comment-analysis table for the same docket, so you can see + both views of "what commenters said" at once. + """) + return + + +@app.cell(hide_code=True) +def _(): + # ── Setup ── + import polars as pl + + from conf import connect + from llm.client import LlmClient + + NOTES = {} + + def _open_replica(): + try: + return connect.duckdb("aco", read_only=True) + except Exception as e: # noqa: BLE001 — degrade, never crash the page + NOTES["replica"] = f"replica unavailable: {e}" + return None + + def _open_bib(): + try: + return connect.bib() + except Exception as e: # noqa: BLE001 + NOTES["bib"] = f"bibliography unavailable: {e}" + return None + + con = _open_replica() + store = _open_bib() + client = LlmClient() + + def _service_up(): + try: + return client.health() + except Exception: # noqa: BLE001 — health() never raises; belt & suspenders + return False + + llm_up = _service_up() + if not llm_up: + NOTES["llm"] = f"llm service unreachable at {client.base_url}" + + return NOTES, client, con, llm_up, pl, store + + +@app.cell(hide_code=True) +def _(mo, store): + # ── Docket + query pickers ── + _DEFAULT_DOCKETS = ["CMS-2023-0121", "CMS-2025-0304"] + + def _docket_options(): + if store is not None: + try: + ids = sorted({d.id for d in store.dockets()}) + except Exception: # noqa: BLE001 — fall back to the hard-coded list + ids = [] + if ids: + return ids + return list(_DEFAULT_DOCKETS) + + _options = _docket_options() + _default = "CMS-2025-0304" if "CMS-2025-0304" in _options else _options[0] + docket_picker = mo.ui.dropdown(options=_options, value=_default, label="Docket") + query_box = mo.ui.text( + value="telehealth", + label="Query", + full_width=True, + placeholder="a phrase to search for", + ) + mo.vstack([docket_picker, query_box]) + return docket_picker, query_box + + +@app.cell(hide_code=True) +def _(NOTES, client, docket_picker, llm_up, mo, pl, query_box): + # ── Semantic search (LlmClient.search) ── + _docket = docket_picker.value + _q = (query_box.value or "").strip() + + search_results = pl.DataFrame() + if not llm_up: + search_view = mo.md( + f"_llm service unreachable at `{client.base_url}` — search skipped. " + "Set `LLM_URL` or run this inside the notebooks container " + "(`gateway` network) to reach it._" + ) + elif not _q: + search_view = mo.md("_Type a query above to search._") + else: + try: + _hits = client.search(_q, collection="comments", docket=_docket, limit=10) + except Exception as e: # noqa: BLE001 — degrade, never crash the page + NOTES["search"] = f"search failed: {e}" + _hits = [] + if _hits: + search_results = pl.DataFrame( + [ + { + "label": h.get("label", ""), + "date": h.get("date", ""), + "snippet": h.get("snippet", ""), + "url": h.get("url", ""), + } + for h in _hits + ] + ) + search_view = mo.ui.table( + search_results, label=f"Semantic search — {_docket} ({len(_hits)} hits)" + ) + else: + search_view = mo.md( + f"_No semantic-search hits for `{_q}` in {_docket} — either " + "nothing matches, or the docket isn't indexed yet " + "(`stack llm index`)._" + ) + return search_results, search_view + + +@app.cell(hide_code=True) +def _(NOTES, con, docket_picker, mo, pl): + # ── DuckDB comments view (skin_subs.rulemaking_comments) ── + _docket = docket_picker.value + + duckdb_comments = pl.DataFrame() + if con is None: + duckdb_view = mo.md("_replica unavailable — DuckDB comments view skipped._") + else: + try: + duckdb_comments = con.execute( + "SELECT comment_id, posted_date, organization, position, " + "text_length, is_form_letter FROM skin_subs.rulemaking_comments " + "WHERE docket_id = ? ORDER BY posted_date DESC", + [_docket], + ).pl() + except Exception as e: # noqa: BLE001 — a missing table is "not built yet" + NOTES[f"duckdb:{_docket}"] = str(e) + if duckdb_comments.is_empty(): + duckdb_view = mo.md( + f"_No `skin_subs.rulemaking_comments` rows for {_docket} yet — " + "this table only covers dockets run through " + "`uv run python dev/scripts/classify_comments.py --docket " + f"{_docket}` (the skin-substitute-relevant subset of a " + "docket's comments, not every comment on it)._" + ) + else: + duckdb_view = mo.ui.table( + duckdb_comments, + label=( + f"DuckDB skin_subs.rulemaking_comments — {_docket} " + f"({duckdb_comments.height} rows)" + ), + ) + return (duckdb_view,) + + +@app.cell(hide_code=True) +def _(docket_picker, duckdb_view, mo, search_view): + # ── Side by side ── + mo.hstack( + [ + mo.vstack( + [mo.md(f"### Semantic search — {docket_picker.value}"), search_view] + ), + mo.vstack( + [mo.md(f"### DuckDB comments — {docket_picker.value}"), duckdb_view] + ), + ], + widths="equal", + gap=2, + ) + return + + +@app.cell(hide_code=True) +def _(NOTES, mo): + # ── Notes ── + mo.md( + "### Notes\n\n" + + ( + "\n".join(f"- {k}: {v}" for k, v in NOTES.items()) + if NOTES + else "_All sources available._" + ) + ) + return + + +if __name__ == "__main__": + app.run() diff --git a/src/bib/dockets.py b/src/bib/dockets.py index 087d1ea..629f9f5 100644 --- a/src/bib/dockets.py +++ b/src/bib/dockets.py @@ -66,7 +66,10 @@ def fingerprint_files(paths: Iterable[Path]) -> str: try: st = p.stat() rows.append(f"{p.name}\x00{st.st_size}\x00{st.st_mtime_ns}") - except FileNotFoundError: + except OSError: + # Missing (FileNotFoundError), unreadable (PermissionError), + # or a broken symlink/removed parent dir — any of these + # means "not there" for fingerprinting purposes. rows.append(f"{p.name}\x00-1\x000") rows.sort() return hashlib.sha256("\n".join(rows).encode()).hexdigest() diff --git a/src/bib/regulations_gov.py b/src/bib/regulations_gov.py index 32491b3..ed14725 100644 --- a/src/bib/regulations_gov.py +++ b/src/bib/regulations_gov.py @@ -391,6 +391,42 @@ class Client: # ── Bib integration ──────────────────────────────────────────── +def _sealed_backfill_skip( + store: Store, docket: str, force: bool +) -> dict[str, int] | None: + """Sealed-docket short-circuit shared by :func:`backfill_details` and + :func:`backfill_from_mirror`: both return the same zero-work stats + dict, unchanged, when *docket* is sealed and *force* is false. + + Returns the stats dict to return immediately, or ``None`` when the + caller should proceed (no docket given, ``force``, unknown docket, + or a known-but-open docket). + """ + if not docket or force: + return None + d = store.docket_get(docket) + if d is None or not d.sealed: + return None + log.info( + "%s sealed %s (%s); backfill skipped", + docket, + d.sealed_at[:10], + d.seal_reason, + ) + print( + f"{docket}: sealed {d.sealed_at[:10]} ({d.seal_reason}); skipped — use --force", + flush=True, + ) + return { + "skipped_sealed": 1, + "seen": 0, + "enriched": 0, + "created": 0, + "attached": 0, + "errors": 0, + } + + def backfill_details( store: Store, client: Client, @@ -418,27 +454,9 @@ def backfill_details( so we stop hammering it. Commits land every ``commit_every`` items so a crash loses at most that many items of work. """ - if docket and not force: - d = store.docket_get(docket) - if d is not None and d.sealed: - log.info( - "%s sealed %s (%s); backfill skipped", - docket, - d.sealed_at[:10], - d.seal_reason, - ) - print( - f"{docket}: sealed {d.sealed_at[:10]} ({d.seal_reason}); skipped — use --force", - flush=True, - ) - return { - "skipped_sealed": 1, - "seen": 0, - "enriched": 0, - "created": 0, - "attached": 0, - "errors": 0, - } + skip = _sealed_backfill_skip(store, docket, force) + if skip is not None: + return skip con = store._con() # noqa: SLF001 # Resume rule: treat ``enriched:ok`` as the completion marker. # Using a tag (not just the abstract) lets us handle "see attached" @@ -711,27 +729,9 @@ def backfill_from_mirror( S3 fetches fan out over a thread pool; every Store/sqlite call stays on this thread. Idempotent: ``enriched:ok`` items are skipped. """ - if docket and not force: - d = store.docket_get(docket) - if d is not None and d.sealed: - log.info( - "%s sealed %s (%s); backfill skipped", - docket, - d.sealed_at[:10], - d.seal_reason, - ) - print( - f"{docket}: sealed {d.sealed_at[:10]} ({d.seal_reason}); skipped — use --force", - flush=True, - ) - return { - "skipped_sealed": 1, - "seen": 0, - "enriched": 0, - "created": 0, - "attached": 0, - "errors": 0, - } + skip = _sealed_backfill_skip(store, docket, force) + if skip is not None: + return skip con = store._con() # noqa: SLF001 existing: dict[str, tuple[str, bool]] = {} # cid -> (key, enriched) for key, url, enriched in con.execute( @@ -1030,18 +1030,25 @@ def walk_docket( def docket_counts(store: Store, docket_id: str) -> dict[str, int]: - """``{"comments": n, "enriched": n}`` from SQL only (no filesystem).""" + """``{"comments": n, "enriched": n}`` from SQL only (no filesystem). + + One grouped scan of ``items`` (a left join against the tagged-item + set) rather than two separate ``url LIKE`` scans. + """ con = store._con() # noqa: SLF001 like = f"https://www.regulations.gov/comment/{docket_id}-%" - comments = con.execute( - "SELECT count(*) FROM items WHERE url LIKE ?", (like,) - ).fetchone()[0] - enriched = con.execute( - """SELECT count(*) FROM items i WHERE i.url LIKE ? AND i.id IN ( - SELECT item_id FROM item_tags WHERE tag_id IN (SELECT id FROM tags WHERE name='enriched:ok'))""", + row = con.execute( + """SELECT count(*), + SUM(CASE WHEN et.item_id IS NOT NULL THEN 1 ELSE 0 END) + FROM items i + LEFT JOIN ( + SELECT item_id FROM item_tags + WHERE tag_id IN (SELECT id FROM tags WHERE name = 'enriched:ok') + ) et ON et.item_id = i.id + WHERE i.url LIKE ?""", (like,), - ).fetchone()[0] - return {"comments": comments, "enriched": enriched} + ).fetchone() + return {"comments": row[0], "enriched": row[1] or 0} # ── Internals ────────────────────────────────────────────────── diff --git a/src/bib/store.py b/src/bib/store.py index f28e21a..c73fa52 100644 --- a/src/bib/store.py +++ b/src/bib/store.py @@ -86,7 +86,15 @@ class Store: # The (item_id, filename) unique index can only exist once the # dedupe migration has run (dev/scripts/dedupe_attachments.py); # until then we leave it off rather than fail to open the store. + # Once it exists, sqlite_master says so directly — skip the + # GROUP BY scan (which otherwise runs on every Store() open). con = self._con() + indexed = con.execute( + "SELECT 1 FROM sqlite_master WHERE type = 'index' " + "AND name = 'idx_attachments_item_filename'" + ).fetchone() + if indexed is not None: + return dup = con.execute( "SELECT 1 FROM attachments GROUP BY item_id, filename HAVING count(*) > 1 LIMIT 1" ).fetchone() @@ -266,9 +274,11 @@ class Store: * columns: an empty incoming value keeps the stored one (a list-walk row carries no body, and must not blank an enriched - abstract). ``extra_json`` is exempt — it is always a full - serialization, so an incoming ``{}`` is a real value and - replaces what is stored. + abstract). ``extra_json`` is merged per key instead of + column-wise: a stored key survives unless the incoming + payload sets that key to a non-empty value, so a list-walk + row's empty subclass fields (``{"effective_date": "", ...}``) + never blank an already-stored value. * tags and collections: union-merged with what is stored, never replaced. Removal goes through :meth:`remove_tag`. """ @@ -301,11 +311,22 @@ class Store: cur_row = current.to_row() # Merge, don't replace: an empty incoming value carries no # information (a list-walk row has no body), so the stored value - # wins. extra_json is always a full serialization and is exempt. + # wins. for c in self._COMPARE_COLS: if c != "extra_json" and not new_row.get(c) and cur_row.get(c): new_row[c] = cur_row[c] setattr(item, c, cur_row[c]) + + # extra_json is a full serialization of subclass-specific fields + # every time, so a column-wise empty check can't apply — merge + # the decoded dict per key instead: a stored key survives unless + # the incoming payload sets it to a non-empty value. + cur_extra = json.loads(cur_row.get("extra_json") or "{}") + new_extra = json.loads(new_row.get("extra_json") or "{}") + merged_extra = {**cur_extra, **{k: v for k, v in new_extra.items() if v}} + merged_extra_json = json.dumps(merged_extra) + new_row["extra_json"] = merged_extra_json + same_cols = all( new_row.get(c, "") == cur_row.get(c, "") for c in self._COMPARE_COLS ) @@ -319,6 +340,7 @@ class Store: item.stamp_access() row = item.to_row() row.pop("key", None) + row["extra_json"] = merged_extra_json row["tags"] = merged_tags row["collections"] = merged_cols self.update(ekey, **row) diff --git a/src/cli/bib.py b/src/cli/bib.py index 80a0ed8..04d07b9 100644 --- a/src/cli/bib.py +++ b/src/cli/bib.py @@ -220,7 +220,14 @@ def fetch_pfs_comments( docket: str = typer.Option( "", "--docket", - help="Only pull this reg.gov docket (e.g. CMS-2026-2377); other dockets are skipped before any API call once known.", + help=( + "Only pull this reg.gov docket (e.g. CMS-2026-2377). Filtered " + "before any API call ONLY once the docket is known (a `dockets` " + "row already exists for its CMS id) — a CMS id with no `dockets` " + "row yet still costs one Federal Register rule-metadata fetch " + "plus one resolve_docket call, because the reg.gov docket id " + "isn't known until those calls resolve it." + ), ), force: bool = typer.Option( False, @@ -232,7 +239,10 @@ def fetch_pfs_comments( on each docket from its stored watermark. Sealed dockets (comment period closed + quiet period + an empty pull) - cost nothing: no rule-metadata fetch, no resolve call, no walk. + cost nothing: no rule-metadata fetch, no resolve call, no walk. The + same is true for a docket already known via `--docket` that resolves + to a different id. A CMS id with no `dockets` row at all is not free + even under `--docket`: see that option's help. """ from bib import connect from bib.federalregister import pfs_rules, split_docket_ids @@ -279,6 +289,12 @@ def fetch_pfs_comments( rule.add_tag("module:pfs") for cid in cms_ids: rule.add_tag(f"cms-rule:{cid}") + # Upsert now, before any docket resolution: a CMS id whose + # docket never resolves (resolve_docket returns nothing) must + # not cost the rule item its only upsert. Each docket below + # re-upserts to add its reg-docket tag; upsert_status makes + # the extra call a no-op when nothing changed. + store.upsert(rule) for cms_id in todo: d = known[cms_id] diff --git a/src/cli/comments.py b/src/cli/comments.py index 2818a72..7c8532f 100644 --- a/src/cli/comments.py +++ b/src/cli/comments.py @@ -239,22 +239,32 @@ def extract_ocr( return body_lookup = _bib_lookup_factory(True) - engine = RapidOcrEngine() - done = errors = 0 - for i, comment_dir in enumerate(pending, 1): - try: - extract_comment( - comment_dir, - inline_body=body_lookup(comment_dir.name), - force=True, - ocr_engine=engine, - ) - done += 1 - except Exception as e: # noqa: BLE001 - log.warning("ocr extract failed for %s: %s", comment_dir.name, e) - errors += 1 - if i % 25 == 0: - typer.echo(f" {i}/{len(pending)} ok={done} errors={errors}", err=False) + try: + engine = RapidOcrEngine() + done = errors = 0 + for i, comment_dir in enumerate(pending, 1): + try: + extract_comment( + comment_dir, + inline_body=body_lookup(comment_dir.name), + force=True, + ocr_engine=engine, + ) + done += 1 + except Exception as e: # noqa: BLE001 + log.warning("ocr extract failed for %s: %s", comment_dir.name, e) + errors += 1 + if i % 25 == 0: + typer.echo( + f" {i}/{len(pending)} ok={done} errors={errors}", err=False + ) + finally: + # _bib_lookup_factory opens its own read connection to bib.sqlite + # when the store isn't :memory: (see _build_bib_helpers); close + # it here the same way extract() closes its own. + close = getattr(body_lookup, "close", None) + if close is not None: + close() typer.echo(f" done: {done} re-extracted, {errors} errors") diff --git a/src/cli/pfs.py b/src/cli/pfs.py index 75d4d6d..1325e51 100644 --- a/src/cli/pfs.py +++ b/src/cli/pfs.py @@ -41,7 +41,7 @@ from pfs.codetables import ( write_reaction, ) from pfs.cpt_load import ingest as cpt_ingest -from pfs.extract import extract_code +from pfs.extract import extract_codes from pfs.families import ( FAMILIES, HAND_FAMILIES, @@ -117,8 +117,13 @@ def _engine() -> Any: def _run_elements( con: Any, store: Any, targets: list[str], classify: Any, *, write: bool ) -> None: + # #698: one inverted pass over fr_anchors (plus one query each over + # pfs.cpt_code / terminology.hcpcs_level_2 / pfs.rvu) for the whole + # target list — the per-code form re-scanned the 193k-row, unindexed + # fr_anchors table twice per code, ~15-25 min for --all-payable. + by_code = extract_codes(store, con, targets, classify=classify) for c in targets: - x = extract_code(store, con, c, classify=classify) + x = by_code[c] if write: write_elements(con, c, x.rows, x.reviews) confirmed = sum(1 for r in x.rows if r.confirmed_by) @@ -134,10 +139,13 @@ def _codes_for( *, warn_slow: bool = True, ) -> list[str]: - """*warn_slow*: the "~20 min of SQL before any model call" warning - describes ``elements``' per-code LLM classification pass — it does - NOT apply to ``lineage --all-payable`` (one inverted pass over - ``fr_anchors``, seconds not minutes), which passes ``warn_slow=False``.""" + """*warn_slow*: log how many codes ``--all-payable`` resolved to, so a + run that is about to make one model call per unplaced line for + thousands of codes says so up front. ``lineage --all-payable`` is + pure SQL/regex and needs no such note, so it passes + ``warn_slow=False``. (Both commands now sweep ``fr_anchors`` in one + inverted pass — #698 — so neither is the ~20 min of per-code SQL the + warning used to describe.)""" out: list[str] = [c.upper() for c in codes] if families: refresh_from(con) @@ -157,11 +165,7 @@ def _codes_for( "AND year = (SELECT max(year) FROM pfs.rvu) ORDER BY hcpcs" ).fetchall() if warn_slow: - log.warning( - "--all-payable is experimental: targeting %d codes (~20 min of SQL " - "before any model call)", - len(rows), - ) + log.warning("--all-payable: targeting %d codes", len(rows)) out.extend(r[0] for r in rows) if not out: raise typer.BadParameter("pass --code, --family or --all-payable") @@ -178,8 +182,8 @@ def elements( False, "--all-payable", help=( - "Every A/R/T code in the newest RVU year (experimental: ~20 min " - "of SQL before any model call)." + "Every A/R/T code in the newest RVU year, one inverted pass over " + "fr_anchors." ), ), no_llm: bool = typer.Option( diff --git a/src/llm/client.py b/src/llm/client.py new file mode 100644 index 0000000..8294e36 --- /dev/null +++ b/src/llm/client.py @@ -0,0 +1,170 @@ +"""``LlmClient``: an httpx-only client for the llm chat/search service (#573). + +Notebook-side counterpart to the retrieval and chat endpoints exposed by +``llm.api`` (``GET /search``, ``GET /similar/{key}``, ``POST /chat``, +``GET /health``) — the same shape used by the chat UI, but callable from a +marimo cell without pulling in the ``llm`` extra (langchain/sqlalchemy/ +psycopg). The notebooks container installs the workspace *without* that +extra (#720); this module only needs ``httpx``, which it always has. + +Base URL resolution, in order: + 1. the *base_url* constructor argument + 2. the ``LLM_URL`` environment variable + 3. ``http://llm:8000`` — the compose service name, reachable because + the notebooks container sits on the same ``gateway`` network as + ``llm`` (compose.yml) + 4. ``http://localhost:8000`` — outside compose (llm publishes no host + port, so this only works when something else forwards it) + +Usage:: + + from llm.client import LlmClient + + client = LlmClient() + if client.health(): + hits = client.search("telehealth", docket="CMS-2025-0304") +""" + +from __future__ import annotations + +import json +import os +import socket +from typing import Any, Iterator + +import httpx + +_DEFAULT_TIMEOUT = 30.0 +_COMPOSE_URL = "http://llm:8000" +_LOCAL_URL = "http://localhost:8000" + + +def _on_gateway_network() -> bool: + """Whether the ``llm`` compose service name resolves from here — true + inside a container on the ``gateway`` network (e.g. ``notebooks``), + false on the host, where nothing publishes that name.""" + try: + socket.gethostbyname("llm") + return True + except OSError: + return False + + +def _resolve_base_url(base_url: str | None) -> str: + if base_url: + return base_url.rstrip("/") + env = os.environ.get("LLM_URL", "").strip() + if env: + return env.rstrip("/") + return _COMPOSE_URL if _on_gateway_network() else _LOCAL_URL + + +class LlmClient: + """Thin httpx client for the llm service's retrieval and chat endpoints. + + ``search``/``similar``/``chat`` raise ``httpx.HTTPError`` (with the + request context in the message) on a network failure or non-2xx + response. ``health`` never raises — it is meant for a notebook guard + cell that wants a plain yes/no before doing anything else. + """ + + def __init__( + self, + base_url: str | None = None, + *, + timeout: float = _DEFAULT_TIMEOUT, + ) -> None: + self.base_url = _resolve_base_url(base_url) + self.timeout = timeout + + def _client(self) -> httpx.Client: + return httpx.Client(base_url=self.base_url, timeout=self.timeout) + + def _get(self, path: str, *, params: dict[str, Any]) -> dict: + try: + with self._client() as http: + resp = http.get(path, params=params) + resp.raise_for_status() + return resp.json() + except httpx.HTTPError as exc: + raise httpx.HTTPError(f"GET {self.base_url}{path} failed: {exc}") from exc + + def health(self) -> bool: + """``True`` iff ``GET /health`` responds ok. Never raises.""" + try: + with self._client() as http: + resp = http.get("/health") + resp.raise_for_status() + return bool(resp.json().get("status") == "ok") + except (httpx.HTTPError, ValueError): + return False + + def search( + self, + q: str, + *, + collection: str = "all", + docket: str | None = None, + item_key: str | None = None, + year: str | int | None = None, + kind: str | None = None, + limit: int = 10, + offset: int = 0, + ) -> list[dict]: + """``GET /search`` — metadata-filtered similarity search. + + Mirrors ``llm.api.search_endpoint``'s query parameters; returns + the ``results`` list (each a source dict: ``label``, ``date``, + ``url``, ``snippet``, ``docket``, ``item_key``, ``score``, ...). + """ + params: dict[str, Any] = { + "q": q, + "collection": collection, + "limit": limit, + "offset": offset, + } + if docket: + params["docket"] = docket + if item_key: + params["item_key"] = item_key + if year: + params["year"] = str(year) + if kind: + params["kind"] = kind + return self._get("/search", params=params)["results"] + + def similar( + self, key: str, *, collection: str = "all", limit: int = 10 + ) -> list[dict]: + """``GET /similar/{key}`` — nearest neighbours of an already-indexed + bib item's own chunk(s). Returns the ``results`` list.""" + params = {"collection": collection, "limit": limit} + return self._get(f"/similar/{key}", params=params)["results"] + + def chat( + self, question: str, *, mode: str = "auto", since: str | None = None + ) -> Iterator[dict]: + """``POST /chat`` — stream the SSE events the chat UI consumes. + + Yields each ``data:`` line's JSON payload (``{"type": ..., ...}``) + as the server sends it; the caller decides what to do with + ``token``/``source``/``error``/etc. event types. + """ + payload: dict[str, Any] = {"question": question, "mode": mode} + if since: + payload["since"] = since + try: + with ( + self._client() as http, + http.stream("POST", "/chat", json=payload) as resp, + ): + resp.raise_for_status() + for line in resp.iter_lines(): + if not line.startswith("data:"): + continue + data = line[len("data:") :].strip() + if not data: + continue + yield json.loads(data) + except httpx.HTTPError as exc: + raise httpx.HTTPError(f"POST {self.base_url}/chat failed: {exc}") from exc diff --git a/src/llm/lineage.py b/src/llm/lineage.py index 99978fa..7f53fbe 100644 --- a/src/llm/lineage.py +++ b/src/llm/lineage.py @@ -829,6 +829,14 @@ def _collect_guidance( url = cfrlink.url(cfrlink.parse_cite(r.locator)) except Exception as e: # noqa: BLE001 log.warning("lineage CFR url unresolved (%s): %s", r.locator, e) + elif r.item_key: + # #705: a resolved IOM chapter / MLN product links to the bib + # item's own URL (the CMS PDF), plus ``#page=N`` when the + # guidance build located the section heading in the chapter. + item = _cached_item(store, r.item_key, mtime) + url = (getattr(item, "url", "") or "") if item is not None else "" + if url and r.page: + url = f"{url}#page={r.page}" # IOM section heading, #705 item 2 out.append( GuidanceRef( kind=r.kind, diff --git a/src/llm/pages.py b/src/llm/pages.py index 9b1bba0..860c269 100644 --- a/src/llm/pages.py +++ b/src/llm/pages.py @@ -58,6 +58,32 @@ def locate(pages: list[str], probe: str) -> int: return 0 +#: An IOM chapter heading in the body — "30.6.4 - Evaluation and +#: Management ... (Rev. 12345; Issued: ...)" — is always followed by its +#: revision line; the table of contents on the first pages lists the +#: same "30.6.4 - ..." entry without one, which is how the two are told +#: apart (#705 item 2). +_REV_WINDOW = 240 + + +def locate_section(pages: list[str], section: str) -> int: + """1-based page whose text carries IOM ``section`` ("30.6.4") as a + body heading — the number, a dash, the title, then "(Rev." within + ``_REV_WINDOW`` characters; 0 when no page does (a TOC-only hit is + not a location).""" + section = section.strip().rstrip(".") + if not section: + return 0 + rx = re.compile( + rf"(? list[Chunk]: """Stamp ``attachment`` (+ ``page`` for PDFs) on chunks whose ``section`` names one of ``doc.files``. Identity when ``doc.files`` diff --git a/src/llm/source.py b/src/llm/source.py index 532c397..a149007 100644 --- a/src/llm/source.py +++ b/src/llm/source.py @@ -87,6 +87,26 @@ def _attachment_paths(store: Store, item_key: str) -> list[Path]: return [Path(r[0]) for r in rows] +def _attachment_paths_by_key(store: Store) -> dict[str, list[Path]]: + """Every item's attachment paths, grouped by key, in one scan. + + ``iter_rule_refs``/``iter_corpus_refs`` used to call + :func:`_attachment_paths` once per item just to build the + fingerprint — one query per rule/corpus item on every run, before + any change is even known. One table scan replaces that N+1.""" + rows = ( + store._con() + .execute( + "SELECT i.key, a.storage_path FROM attachments a JOIN items i ON i.id = a.item_id" + ) + .fetchall() + ) + by_key: dict[str, list[Path]] = {} + for key, path in rows: + by_key.setdefault(key, []).append(Path(path)) + return by_key + + # ── comments ── @@ -183,12 +203,12 @@ def iter_comment_refs( _, body = parse_combined(combined.read_text()) if body.strip(): return Doc(key=key, text=body, metadata=meta, files=files) - row = ( + item_row = ( store._con() .execute("SELECT abstract FROM items WHERE key = ?", (key,)) .fetchone() ) - abstract = (row["abstract"] if row else "") or "" + abstract = (item_row["abstract"] if item_row else "") or "" return ( Doc(key=key, text=abstract, metadata=meta) if abstract.strip() else None ) @@ -208,6 +228,9 @@ def iter_comment_docs( yield doc +# ── rules ── + + def _attachment_text(store: Store, item_key: str) -> str: from rex.comments.combine import extract_attachment @@ -265,9 +288,6 @@ def _anchor_doc(store: Store, item_key: str) -> tuple[str, str]: return (row[0], str(row[1])) if row else ("", "") -# ── rules ── - - def iter_rule_refs( store: Store, *, keys: tuple[str, ...] = (), tag: str = "" ) -> Iterator[DocRef]: @@ -291,6 +311,13 @@ def iter_rule_refs( ) .fetchall() ) + # Attachment paths are only needed for rows without an anchor sha; + # one batched scan replaces one _attachment_paths query per such row. + paths_by_key = ( + _attachment_paths_by_key(store) + if any(not r["anchor_sha"] for r in rows) + else {} + ) for row in rows: key = row["key"] if keys and key not in keys: @@ -300,11 +327,7 @@ def iter_rule_refs( # item's own metadata (title, cms-rule: tag, date). fp = f"anchors:{row['anchor_sha']}|{row['updated_at']}" else: - fp = ( - fingerprint_files(_attachment_paths(store, key)) - + "|" - + row["updated_at"] - ) + fp = fingerprint_files(paths_by_key.get(key, [])) + "|" + row["updated_at"] def _load(key=key) -> Doc | None: return _build_rule_doc(store, store.get(key)) @@ -524,11 +547,16 @@ def iter_corpus_refs( ) + " ORDER BY i.id" ) - for row in store._con().execute(sql, (tag,) if tag else ()).fetchall(): + rows = store._con().execute(sql, (tag,) if tag else ()).fetchall() + # One batched scan replaces one _attachment_paths query per item — + # the fingerprint step used to cost a query per corpus item before + # any change was even known. + paths_by_key = _attachment_paths_by_key(store) + for row in rows: key = row["key"] if keys and key not in keys: continue - fp = row["updated_at"] + "|" + fingerprint_files(_attachment_paths(store, key)) + fp = row["updated_at"] + "|" + fingerprint_files(paths_by_key.get(key, [])) def _load(key=key) -> Doc | None: return _build_corpus_doc(store, store.get(key), zotero) diff --git a/src/pfs/codetables.py b/src/pfs/codetables.py index b075d30..cb04a76 100644 --- a/src/pfs/codetables.py +++ b/src/pfs/codetables.py @@ -119,7 +119,11 @@ class GuidanceRow: most CFR/IOM citations have no matching library item on hand); ``item_key_src``/``p_id_src``/``page_src`` are the citing paragraph's own provenance (an FR paragraph, or the CPT edition - item with ``p_id_src=0``, ``page_src=0``).""" + item with ``p_id_src=0``, ``page_src=0``); ``page`` is the 1-based + page inside the resolved item's PDF where an IOM section heading + sits (#705 item 2 — ``pfs.guidance.iom_section_page``), ``0`` when + unlocated or not applicable. Trailing default so positional + constructors written before it existed keep working.""" family: str code: str @@ -129,6 +133,7 @@ class GuidanceRow: item_key_src: str p_id_src: int page_src: int + page: int = 0 @dataclass(frozen=True) @@ -261,7 +266,7 @@ CREATE TABLE IF NOT EXISTS pfs.code_family ( item_key VARCHAR, p_id INTEGER, note VARCHAR); CREATE TABLE IF NOT EXISTS pfs.code_guidance ( family VARCHAR, code VARCHAR, kind VARCHAR, locator VARCHAR, item_key VARCHAR, - item_key_src VARCHAR, p_id_src INTEGER, page_src INTEGER); + item_key_src VARCHAR, p_id_src INTEGER, page_src INTEGER, page INTEGER); CREATE TABLE IF NOT EXISTS pfs.code_reaction ( family VARCHAR, period VARCHAR, period_kind VARCHAR, year INTEGER, n_items INTEGER, n_total INTEGER, stance_support INTEGER, stance_oppose INTEGER, @@ -320,6 +325,8 @@ def ensure_tables(con: Any) -> None: con.execute( "ALTER TABLE pfs.code_element_review ADD COLUMN IF NOT EXISTS source VARCHAR" ) + # #705 item 2: the located IOM section page. + con.execute("ALTER TABLE pfs.code_guidance ADD COLUMN IF NOT EXISTS page INTEGER") def _insert(con: Any, table: str, rows: Sequence[Any]) -> int: @@ -429,7 +436,8 @@ def read_guidance(con: Any, family: str) -> list[GuidanceRow]: "SELECT * FROM pfs.code_guidance WHERE family = ? ORDER BY code, kind, locator", [family], ).fetchall() - return [GuidanceRow(*r) for r in rows] + # Rows written before ``page`` existed read back NULL for it. + return [GuidanceRow(*r[:-1], r[-1] or 0) for r in rows] def write_reaction(con: Any, family: str, rows: Sequence[ReactionRow]) -> int: diff --git a/src/pfs/descriptors.py b/src/pfs/descriptors.py index 7184162..859d02d 100644 --- a/src/pfs/descriptors.py +++ b/src/pfs/descriptors.py @@ -20,7 +20,7 @@ from __future__ import annotations import re from dataclasses import dataclass -from typing import Any +from typing import Any, Sequence from pfs.elements import is_element_paragraph @@ -157,6 +157,118 @@ def descriptor_runs( return runs +#: A code token as the codes we target are shaped: five alphanumerics +#: ("99490", "G0556", "0075T"). ``_stem_pattern_any`` uses it as the +#: generic stand-in for "any target code" so one regex finds every code's +#: stem in a paragraph; a target of any other length falls back to its +#: own literal in the same alternation. +_CODE_TOKEN = re.compile(r"[0-9A-Za-z]{5}\Z") + + +def _stem_pattern_any(codes: Sequence[str]) -> re.Pattern[str]: + """One pattern matching *any* of *codes* in descriptor-stem position. + + ``_stem_pattern`` compiled per code turns an N-code sweep into N + passes over every paragraph. The per-code pattern differs between + codes only in the code literal, and every target is a fixed-width + token, so substituting the token's character class for the literal + matches at exactly the positions the per-code patterns do — the class + can only stand for the whole five characters that precede the opening + parenthesis, so no other code can slip in. The caller re-checks + ``group(1)`` against its target set. Odd-length targets, if any, are + added as explicit literals so they keep matching too.""" + odd = sorted(c for c in codes if not _CODE_TOKEN.match(c)) + token = "[0-9A-Z]{5}" + if odd: + token = "(?:" + token + "|" + "|".join(re.escape(c) for c in odd) + ")" + return re.compile(rf"\b({token})\s*\(\s*(?!codes?\b)[A-Za-z]", re.I) + + +@dataclass +class _OpenRun: + """A stem whose element paragraphs are still arriving in the stream.""" + + item_key: str + rule_year: int + stem: Para + sort_key: tuple[str, int] + elements: list[Para] + + +def descriptor_runs_bucketed( + store: Any, codes: Sequence[str], *, max_elements: int = 40 +) -> dict[str, list[DescriptorRun]]: + """``descriptor_runs`` for every one of *codes* in ONE streaming pass + over ``fr_anchors`` instead of two unindexed scans per code (a + ``LIKE '%CODE%'`` stem query plus a follow-up query per stem found) — + the whole point for an 8.7k-code universe over ~193k paragraphs + (#698). + + Rows arrive ordered by ``(item_key, p_id)``, which is exactly the + order the per-code follow-up query walks, so element paragraphs are + collected by keeping each stem's run open until a paragraph fails + ``is_element_paragraph``, closes with a parenthesis, or *max_elements* + have been taken. Each code's runs are then ordered by + ``(date_published, p_id)`` to match ``descriptor_runs``' own + ``ORDER BY``, so ``descriptor_runs_bucketed(store, [c])[c] == + descriptor_runs(store, c)`` for any single code.""" + targets = {c.upper() for c in codes} + if not targets: + return {} + pat = _stem_pattern_any(targets) + pending: dict[str, list[_OpenRun]] = {c: [] for c in targets} + open_runs: list[_OpenRun] = [] + item = None + for r in store._con().execute( + "SELECT a.item_key, a.p_id, a.page, a.text, i.title, i.date_published " + "FROM fr_anchors a JOIN items i ON i.key = a.item_key " + "ORDER BY a.item_key, a.p_id" + ): + text = r["text"] + if r["item_key"] != item: + item, open_runs = r["item_key"], [] + if open_runs: + if is_element_paragraph(text): + para = Para(r["item_key"], r["p_id"], r["page"], text) + closes = text.rstrip().endswith((")", ").")) + still: list[_OpenRun] = [] + for run in open_runs: + run.elements.append(para) + if not closes and len(run.elements) < max_elements: + still.append(run) + open_runs = still + else: + open_runs = [] + hits = { + m.group(1).upper() + for m in pat.finditer(text) + if m.group(1).upper() in targets + } + for code in sorted(hits): + run = _OpenRun( + r["item_key"], + rule_year_of(r["title"], r["date_published"]), + Para(r["item_key"], r["p_id"], r["page"], descriptor_span(text, code)), + (r["date_published"] or "", r["p_id"]), + [], + ) + pending[code].append(run) + open_runs.append(run) + return { + code: [ + DescriptorRun( + code=code, + item_key=run.item_key, + rule_year=run.rule_year, + stem=run.stem, + elements=tuple(run.elements), + ) + for run in sorted(runs, key=lambda run: run.sort_key) + ] + for code, runs in pending.items() + } + + def hcpcs_long_description(con: Any, code: str) -> str: row = con.execute( "SELECT long_description FROM terminology.hcpcs_level_2 WHERE hcpcs = ? ORDER BY CAST(seqnum AS INTEGER) DESC, recid DESC LIMIT 1", @@ -176,3 +288,49 @@ def rvu_descriptions(con: Any, code: str) -> list[tuple[int, str, str]]: [code.upper()], ).fetchall() return [(int(y), s or "", d or "") for y, s, d in rows] + + +#: DuckDB has no bound-list ``IN`` — unnesting a ``VARCHAR[]`` parameter +#: is the supported way to filter one query by every target code at once, +#: and it keeps the code list out of the SQL text (no 8.7k-literal string +#: to build or re-plan). +IN_TARGETS = "IN (SELECT unnest(CAST(? AS VARCHAR[])))" + + +def hcpcs_long_descriptions(con: Any, codes: Sequence[str]) -> dict[str, str]: + """``hcpcs_long_description`` for every one of *codes* in one query. + Codes with no ``terminology.hcpcs_level_2`` row are simply absent — + callers read them back with ``.get(code, "")``, the same empty string + the per-code helper returns.""" + targets = sorted({c.upper() for c in codes}) + if not targets: + return {} + rows = con.execute( + "SELECT hcpcs, long_description FROM terminology.hcpcs_level_2 " + f"WHERE hcpcs {IN_TARGETS} QUALIFY row_number() OVER " + "(PARTITION BY hcpcs ORDER BY CAST(seqnum AS INTEGER) DESC, recid DESC) = 1", + [targets], + ).fetchall() + return {h: (d or "") for h, d in rows} + + +def rvu_descriptions_bucketed( + con: Any, codes: Sequence[str] +) -> dict[str, list[tuple[int, str, str]]]: + """``rvu_descriptions`` for every one of *codes* in one query — + partitioning by ``(hcpcs, year)`` is the same base-row pick the + per-code helper makes with ``PARTITION BY year`` inside its own + ``WHERE hcpcs = ?``.""" + targets = sorted({c.upper() for c in codes}) + if not targets: + return {} + out: dict[str, list[tuple[int, str, str]]] = {} + for h, y, st, d in con.execute( + "SELECT hcpcs, year, status_code, description FROM pfs.rvu " + f"WHERE hcpcs {IN_TARGETS} AND (mod IS NULL OR mod = '') " + "QUALIFY row_number() OVER (PARTITION BY hcpcs, year ORDER BY mod NULLS FIRST) = 1 " + "ORDER BY hcpcs, year", + [targets], + ).fetchall(): + out.setdefault(h, []).append((int(y), st or "", d or "")) + return out diff --git a/src/pfs/extract.py b/src/pfs/extract.py index d57fcf6..a09d6a6 100644 --- a/src/pfs/extract.py +++ b/src/pfs/extract.py @@ -11,7 +11,10 @@ winning over HCPCS/RVU on a duplicate ``(type, value, detail)``. The losing source isn't just dropped: it's recorded on the winning row's ``confirmed_by`` as a ``source:item_key`` token (#703) — an independent transcription of the same element cross-checks the winner without -doubling the row count. No I/O here beyond what the caller hands in. +doubling the row count. ``extract_codes`` is ``extract_code`` for a whole +target list with the source lookups inverted — one pass over the corpus +rather than one per code (#698). No I/O here beyond what the caller hands +in. """ from __future__ import annotations @@ -21,11 +24,15 @@ from typing import Any, Callable, Sequence from pfs.codetables import ElementRow, ReviewRow, is_missing_table_error from pfs.descriptors import ( + IN_TARGETS, DescriptorRun, Para, descriptor_runs, + descriptor_runs_bucketed, hcpcs_long_description, + hcpcs_long_descriptions, rvu_descriptions, + rvu_descriptions_bucketed, ) from pfs.elements import VOCAB, Element, ElementType, parse_descriptor @@ -158,6 +165,18 @@ def _cpt_elements( raise if row is None: return [], [] + return _cpt_elements_of_row(code, row, classify=classify) + + +def _cpt_elements_of_row( + code: str, + row: Sequence[Any], + *, + classify: Classifier | None = None, +) -> tuple[list[ElementRow], list[ReviewRow]]: + """The parsing half of ``_cpt_elements``, over one already-fetched + ``(edition_year, item_key, stem, elements, tail)`` row, so the + per-code query and the bucketed one share it exactly.""" edition_year, item_key, stem, elements, tail = row stem_text = (stem or "").strip() tail_text = (tail or "").strip() @@ -185,26 +204,56 @@ def _cpt_elements( return rows, reviews -def extract_code( - store: Any, con: Any, code: str, *, classify: Classifier | None = None +def _cpt_elements_bucketed( + con: Any, codes: Sequence[str], *, classify: Classifier | None = None +) -> dict[str, tuple[list[ElementRow], list[ReviewRow]]]: + """``_cpt_elements`` for every one of *codes* in one query — the same + newest-edition pick (the per-code ``ORDER BY edition_year DESC LIMIT + 1``, expressed as a ``QUALIFY row_number()``) and the same + missing-``pfs.cpt_code`` tolerance (I4).""" + targets = sorted({c.upper() for c in codes}) + if not targets: + return {} + try: + rows = con.execute( + "SELECT code, edition_year, item_key, stem, elements, tail " + f"FROM pfs.cpt_code WHERE code {IN_TARGETS} QUALIFY row_number() OVER " + "(PARTITION BY code ORDER BY edition_year DESC) = 1", + [targets], + ).fetchall() + except Exception as exc: + if is_missing_table_error(exc): + return {} + raise + return {r[0]: _cpt_elements_of_row(r[0], r[1:], classify=classify) for r in rows} + + +def _assemble( + code: str, + runs: Sequence[DescriptorRun], + cpt: tuple[list[ElementRow], list[ReviewRow]], + long_desc: str, + years: Sequence[tuple[int, str, str]], + *, + classify: Classifier | None = None, ) -> Extraction: - """All sources for *code*; FR rows win over CPT, and CPT wins over - HCPCS/RVU rows for the same element — each source contributes its own - year.""" - code = code.upper() + """Merge one code's already-fetched sources into an ``Extraction``. + + The single home of the precedence rules (FR > CPT > HCPCS/RVU, the + loser recorded on the winner's ``confirmed_by``), shared by the + per-code ``extract_code`` and the bucketed ``extract_codes`` so the + two paths cannot drift apart (#698).""" merged: dict[tuple[str, str, str], ElementRow] = {} reviews: list[ReviewRow] = [] - for run in descriptor_runs(store, code): + for run in runs: x = extract_run(run, classify=classify) for r in x.rows: _merge_or_confirm(merged, r) reviews.extend(x.reviews) - cpt_rows, cpt_reviews = _cpt_elements(con, code, classify=classify) + cpt_rows, cpt_reviews = cpt for r in cpt_rows: _merge_or_confirm(merged, r) reviews.extend(cpt_reviews) - long_desc = hcpcs_long_description(con, code) - years = rvu_descriptions(con, code) if long_desc: y = years[-1][0] if years else 0 for r in extract_text(code, long_desc, year=y, source="hcpcs").rows: @@ -214,3 +263,51 @@ def extract_code( _merge_or_confirm(merged, r) ordered = sorted(merged.values(), key=lambda r: (r.type, r.value, r.detail)) return Extraction(code, tuple(ordered), tuple(reviews)) + + +def extract_code( + store: Any, con: Any, code: str, *, classify: Classifier | None = None +) -> Extraction: + """All sources for *code*; FR rows win over CPT, and CPT wins over + HCPCS/RVU rows for the same element — each source contributes its own + year.""" + code = code.upper() + return _assemble( + code, + descriptor_runs(store, code), + _cpt_elements(con, code, classify=classify), + hcpcs_long_description(con, code), + rvu_descriptions(con, code), + classify=classify, + ) + + +def extract_codes( + store: Any, con: Any, codes: Sequence[str], *, classify: Classifier | None = None +) -> dict[str, Extraction]: + """``extract_code`` for every one of *codes*, with the source lookups + inverted: ONE streaming pass over ``fr_anchors`` and ONE query each + over ``pfs.cpt_code``, ``terminology.hcpcs_level_2`` and ``pfs.rvu``, + instead of two unindexed ``fr_anchors`` scans plus three replica + queries per code (#698). ``extract_codes(store, con, [c])[c] == + extract_code(store, con, c)``: the bucketed lookups select the same + rows per code, and ``_assemble`` does the merging for both paths. + + Keyed in ``sorted(set(...))`` order, one entry per distinct code — a + code no source mentions still gets an empty ``Extraction``.""" + targets = sorted({c.upper() for c in codes}) + runs = descriptor_runs_bucketed(store, targets) + cpt = _cpt_elements_bucketed(con, targets, classify=classify) + long_descs = hcpcs_long_descriptions(con, targets) + years = rvu_descriptions_bucketed(con, targets) + return { + code: _assemble( + code, + runs.get(code, ()), + cpt.get(code, ([], [])), + long_descs.get(code, ""), + years.get(code, ()), + classify=classify, + ) + for code in targets + } diff --git a/src/pfs/guidance.py b/src/pfs/guidance.py index 6ff3c5d..a9ac53c 100644 --- a/src/pfs/guidance.py +++ b/src/pfs/guidance.py @@ -16,7 +16,7 @@ Two sources, one extraction pass over each hit's text: Both funnel through ``_extract``, which runs three regex families (CFR, IOM, MLN) over one paragraph/guideline's text and resolves each -hit against the bib (``resolve_cfr``/``resolve_iom``) — unresolved +hit against the bib (``resolve_cfr``/``resolve_iom``/``resolve_mln``) — unresolved references keep an empty ``item_key`` (the canonical locator is kept either way, Resolutions §fr_anchors). Rows are written by ``pfs.codetables.write_guidance``/``read_guidance`` into @@ -31,6 +31,7 @@ bare CFR forms firing on the same span) collapses to one. from __future__ import annotations import re +from pathlib import Path from typing import Any, Sequence from bib.cfrlink import canonical, item_for, parse_cite @@ -109,10 +110,38 @@ IOM_RE = re.compile( r"(?:[^.]{0,120}?[Ss]ection\s*([\d.]+))?" ) -#: "MLN907166" / "MLN 907166" booklet numbers (#689: optional/best-effort -#: — no title match against bib items, just the number surfaced as a -#: locator; always unresolved). -MLN_RE = re.compile(r"\bMLN\s*(\d{6,7})\b", re.IGNORECASE) +#: "MLN907166" / "MLN 907166" / "ICN 909289" / "ICN MLN909188" — an MLN +#: product number (booklet, fact sheet, web-based training). CMS renamed +#: the "ICN" (Internet Content Number) prefix to "MLN" in 2020 without +#: renumbering, so both prefixes name the same product series and the +#: stored locator is always the "MLN " form (#705 item 3). +MLN_RE = re.compile(r"\b(?:ICN\s*)?(?:MLN|ICN)\s*(\d{6,7})\b", re.IGNORECASE) +#: "MLN Matters® Number MM9603" / "MLN Matters article SE1316" / "MLN +#: Matters SE 1316" / "MLN Matters article 11268" (a bare number is a +#: change-request article, prefix "MM"). Product numbers (``MLN_RE``) +#: never follow the "Matters" keyword, so the two patterns are disjoint. +MLN_MATTERS_RE = re.compile( + r"\bMLN\s+Matters\W{0,3}(?:(?:article|number|no\.?|#)\s*)*" + r"(MM|SE)?\s?(\d{4,5})\b", + re.IGNORECASE, +) + + +def mln_refs(text: str) -> list[str]: + """Every MLN locator in *text* — ``"MLN 909188"`` for product numbers, + ``"MLN Matters MM9603"`` / ``"MLN Matters SE1316"`` for articles — + in order of appearance, first occurrence wins.""" + out: list[str] = [] + for m in MLN_RE.finditer(text): + loc = f"MLN {m.group(1)}" + if loc not in out: + out.append(loc) + for m in MLN_MATTERS_RE.finditer(text): + prefix = (m.group(1) or "MM").upper() + loc = f"MLN Matters {prefix}{m.group(2)}" + if loc not in out: + out.append(loc) + return out def resolve_cfr(store: Any, title: str, section: str) -> str: @@ -134,6 +163,89 @@ def resolve_iom(store: Any, pub: str, chapter: str) -> str: return "" +def resolve_mln(store: Any, locator: str) -> str: + """The bib item whose URL or title carries the MLN identifier in + *locator* (``"MLN 909188"`` -> ``MLN909188`` or the pre-2020 + ``ICN909188`` spelling; ``"MLN Matters MM9603"`` -> ``MM9603``), or + ``""`` when unresolved. CMS files its products under the number + (``.../MLNProducts/Downloads/eval-mgmt-serv-guide-ICN006764.pdf``, + ``.../MLNMattersArticles/downloads/MM9603.pdf``) and ``bib`` titles + curated booklets with it ("... (MLN909188, June 2025)"), so a + word-bounded match on either column is the whole resolver.""" + tail = locator.split()[-1] if locator else "" + if not tail: + return "" + if tail.isdigit(): + idents = (f"MLN{tail}", f"ICN{tail}") + else: + idents = (tail,) + con = store._con() # noqa: SLF001 — same pattern as ``harvest`` + like = " OR ".join("url LIKE ? OR title LIKE ?" for _ in idents) + params = [p for i in idents for p in (f"%{i}%", f"%{i}%")] + rows = con.execute( + f"SELECT key, title, url FROM items WHERE {like} ORDER BY id", params + ).fetchall() + bounded = [ + re.compile(rf"(? located page; storage path -> normalized page +#: texts. Both live for the process — one ``stack pfs guidance --write`` +#: run locates the same chapter's sections for several families. +_PAGE_CACHE: dict[tuple[str, str], int] = {} +_PDF_CACHE: dict[str, list[str]] = {} + + +def iom_section_page(store: Any, item_key: str, section: str) -> int: + """1-based page of IOM *section* ("30.6.4") inside the chapter PDF + attached to bib item *item_key* (``bib.iom.download_attachments`` + stores one PDF per chapter), or 0 when the item has no PDF on hand, + the section has no body heading, or the PDF toolkit is missing + (#705 item 2). Located at build time on the host so the chat, whose + image has neither the storage tree nor pymupdf, only reads the + stored number.""" + if not item_key or not section: + return 0 + key = (item_key, section) + if key in _PAGE_CACHE: + return _PAGE_CACHE[key] + try: + from llm.pages import locate_section, pdf_pages + except ImportError: # pragma: no cover — llm package always ships with pfs + return 0 + con = store._con() # noqa: SLF001 — same pattern as ``harvest`` + rows = con.execute( + "SELECT a.storage_path FROM attachments a JOIN items i ON i.id = a.item_id " + "WHERE i.key = ? ORDER BY a.id", + (item_key,), + ).fetchall() + page = 0 + for (path,) in rows: + if not path or not path.lower().endswith(".pdf"): + continue + if path not in _PDF_CACHE: + _PDF_CACHE[path] = pdf_pages(Path(path)) + page = locate_section(_PDF_CACHE[path], section) + if page: + break + _PAGE_CACHE[key] = page + return page + + +def _guidance_page(store: Any, kind: str, locator: str, item_key: str) -> int: + """``iom_section_page`` for a resolved IOM locator that names a + section ("100-04 ch.12 §30.6.4"); 0 for everything else.""" + if kind != "iom" or not item_key or "§" not in locator: + return 0 + return iom_section_page(store, item_key, locator.rsplit("§", 1)[1]) + + def _cfr_continuations(text: str, pos: int, pattern: re.Pattern[str]) -> list[str]: """Every list-continuation token starting at *pos* (Ruling A10) — ``pattern`` is ``_CFR_CONT_SECTION_RE`` or ``_CFR_CONT_PART_RE``. @@ -217,8 +329,8 @@ def _extract(text: str, store: Any) -> list[tuple[str, str, str]]: for pub, chapter, section in _iom_refs(text): locator = f"{pub} ch.{chapter}" + (f" §{section}" if section else "") out.append(("iom", locator, resolve_iom(store, pub, chapter))) - for m in MLN_RE.finditer(text): - out.append(("mln", f"MLN {m.group(1)}", "")) + for locator in mln_refs(text): + out.append(("mln", locator, resolve_mln(store, locator))) return out @@ -256,10 +368,19 @@ def harvest(store: Any, codes: Sequence[str], *, family: str) -> list[GuidanceRo if not present: continue for kind, locator, resolved in _extract(text, store): + located = _guidance_page(store, kind, locator, resolved) for code in present: out.append( GuidanceRow( - family, code, kind, locator, resolved, item_key, p_id, page + family, + code, + kind, + locator, + resolved, + item_key, + p_id, + page, + located, ) ) return dedupe(out) @@ -271,8 +392,8 @@ def harvest_cpt( """The newest ingested CPT edition's guideline text and ``see``/ ``other`` instructions for *codes* -> the CFR/IOM/MLN references they cite, anchored to the CPT edition item (``p_id_src=0``, - ``page_src=0`` — no FR paragraph; locating a page inside the PDF - chapter is out of scope, #689). ``[]`` on a replica with no + ``page_src=0`` — no FR paragraph). A resolved IOM section carries + the chapter-PDF page it was located on (``page``, #705 item 2). ``[]`` on a replica with no ``pfs.cpt_*`` tables yet (I4: not a bug, just nothing ingested).""" from pfs.codetables import ( cpt_years, @@ -306,10 +427,19 @@ def harvest_cpt( if section is None or not section.guideline: continue for kind, locator, resolved in _extract(section.guideline, store): + located = _guidance_page(store, kind, locator, resolved) for code in sorted(fam_codes): out.append( GuidanceRow( - family, code, kind, locator, resolved, section.item_key, 0, 0 + family, + code, + kind, + locator, + resolved, + section.item_key, + 0, + 0, + located, ) ) @@ -319,9 +449,18 @@ def harvest_cpt( if not any(k in instr.text for k in ("CFR", "Medicare", "Chapter", "chapter")): continue for kind, locator, resolved in _extract(instr.text, store): + located = _guidance_page(store, kind, locator, resolved) out.append( GuidanceRow( - family, instr.code, kind, locator, resolved, instr.item_key, 0, 0 + family, + instr.code, + kind, + locator, + resolved, + instr.item_key, + 0, + 0, + located, ) ) return dedupe(out) diff --git a/src/zot/schema.py b/src/zot/schema.py index 4a61465..8e6a409 100644 --- a/src/zot/schema.py +++ b/src/zot/schema.py @@ -32,6 +32,11 @@ def create_db(path: str = ":memory:") -> sqlite3.Connection: close it afterward; for ``:memory:`` keep it alive. """ con = sqlite3.connect(path) - con.executescript(_load_sql()) + # One transaction for the whole ~2,500-statement script. executescript + # otherwise autocommits every CREATE/INSERT — thousands of fsyncs — and + # on a file-backed path under disk load that took 15–20 s per fresh + # DB (every prisma test that builds a zotero.sqlite in tmp_path paid + # it; the suite looked hung). As one transaction it is ~instant. + con.executescript("BEGIN;\n" + _load_sql() + "\nCOMMIT;") con.row_factory = sqlite3.Row return con diff --git a/tests/bib/test_dockets.py b/tests/bib/test_dockets.py index 8aded44..6b5b7d1 100644 --- a/tests/bib/test_dockets.py +++ b/tests/bib/test_dockets.py @@ -96,6 +96,18 @@ class TestFingerprintFiles: assert fingerprint_files([]) == fingerprint_files([]) assert len(fingerprint_files([])) == 64 + def test_non_missing_oserror_is_recorded_not_fatal(self, tmp_path: Path): + """refs #680: widened from FileNotFoundError to OSError — a path + under a non-directory (NotADirectoryError, not + FileNotFoundError) must not raise, same as a missing file.""" + f = tmp_path / "not_a_dir.pdf" + f.write_bytes(b"x") + bogus = f / "nested.pdf" # stat() raises NotADirectoryError + assert fingerprint_files([bogus]) == fingerprint_files([bogus]) + assert fingerprint_files([bogus]) == fingerprint_files( + [tmp_path / "nested.pdf"] # same name, both "missing" + ) + class TestQuietDays: def test_default_when_section_missing(self, monkeypatch): diff --git a/tests/bib/test_regulations_gov_exercise.py b/tests/bib/test_regulations_gov_exercise.py index 772f327..66166dc 100644 --- a/tests/bib/test_regulations_gov_exercise.py +++ b/tests/bib/test_regulations_gov_exercise.py @@ -869,3 +869,47 @@ class TestBackfillAddTagGone: stats = backfill_details(store, api, limit=1, scratch_root=tmp_path) assert stats["enriched"] == 1 assert stats["attached"] == 0 + + +class TestDocketCounts: + def test_one_grouped_query(self, tmp_path): + """docket_counts counts comments and enriched ones from one scan + of items (refs #680) — not two separate ``url LIKE`` queries.""" + from bib.item import Source + from bib.regulations_gov import docket_counts + from bib.store import Store + + store = Store(tmp_path / "bib.sqlite") + k1 = store.create( + Source( + title="Comment 1", + url="https://www.regulations.gov/comment/CMS-2023-0001-0001", + ) + ) + k2 = store.create( + Source( + title="Comment 2", + url="https://www.regulations.gov/comment/CMS-2023-0001-0002", + ) + ) + store.create( + Source( + title="Other docket", + url="https://www.regulations.gov/comment/CMS-2099-9999-0001", + ) + ) + store.add_tags(k1, ["enriched:ok"]) + + assert k1 and k2 # both created without error + counts = docket_counts(store, "CMS-2023-0001") + assert counts == {"comments": 2, "enriched": 1} + store.close() + + def test_no_matches(self, tmp_path): + from bib.regulations_gov import docket_counts + from bib.store import Store + + store = Store(tmp_path / "bib.sqlite") + counts = docket_counts(store, "CMS-9999-9999") + assert counts == {"comments": 0, "enriched": 0} + store.close() diff --git a/tests/bib/test_store_attach_idempotent.py b/tests/bib/test_store_attach_idempotent.py index 403439c..f78642d 100644 --- a/tests/bib/test_store_attach_idempotent.py +++ b/tests/bib/test_store_attach_idempotent.py @@ -28,3 +28,28 @@ def test_different_filename_creates_second_row(tmp_path: Path): f2 = tmp_path / "b.pdf" f2.write_bytes(b"b") assert s.attach_file(key, f1) != s.attach_file(key, f2) + + +def test_explicit_title_dedupes_across_different_source_files(tmp_path: Path): + """refs #680: the dedupe key is (item_id, filename) where filename + is ``title or path.name`` — an explicit title must dedupe on its + own value even when the two calls point at physically different + source files (different path.name, different bytes).""" + s = Store(str(tmp_path / "bib.sqlite"), storage_dir=tmp_path / "storage") + key = s.create(Source(title="T", url="https://x/1")) + f1 = tmp_path / "original_name_1.pdf" + f1.write_bytes(b"first version") + f2 = tmp_path / "totally_different_name.pdf" + f2.write_bytes(b"second version, different bytes and path.name") + + k1 = s.attach_file(key, f1, title="Comment text") + k2 = s.attach_file(key, f2, title="Comment text") + + assert k1 == k2 + n = s._con().execute("SELECT count(*) FROM attachments").fetchone()[0] + assert n == 1 + # only the first file's bytes were ever copied into storage + assert len(list((tmp_path / "storage").iterdir())) == 1 + stored = next((tmp_path / "storage").iterdir()) + copied = next(stored.iterdir()) + assert copied.read_bytes() == b"first version" diff --git a/tests/bib/test_store_upsert_noop.py b/tests/bib/test_store_upsert_noop.py index a75fb8e..9acfa3f 100644 --- a/tests/bib/test_store_upsert_noop.py +++ b/tests/bib/test_store_upsert_noop.py @@ -102,3 +102,44 @@ def test_non_empty_incoming_still_updates(): key, _ = s.upsert_status(_item(abstract="v1")) _, status = s.upsert_status(_item(abstract="v2")) assert status == "updated" and s.get(key).abstract == "v2" + + +# ── extra_json merges per key (refs #680) ──────────────────────────── + + +def _rule(**kw): + from bib.item import Rule + + it = Rule(title="T", url="https://example.com/rule-1") + it.cms_id = kw.get("cms_id", "") + it.effective_date = kw.get("effective_date", "") + it.fr_page = kw.get("fr_page", "") + return it + + +def test_empty_incoming_extra_json_key_keeps_stored_key(): + """A list-walk Rule with a blank effective_date must not blank an + already-stored one — extra_json is merged per key, not replaced + wholesale.""" + s = _store() + key, _ = s.upsert_status(_rule(cms_id="CMS-1848-P", effective_date="2026-01-01")) + _, status = s.upsert_status(_rule(cms_id="CMS-1848-P", effective_date="")) + assert status == "unchanged" + stored = s.get(key) + assert stored.effective_date == "2026-01-01" + assert stored.cms_id == "CMS-1848-P" + + +def test_non_empty_incoming_extra_json_key_updates_that_key_only(): + s = _store() + key, _ = s.upsert_status(_rule(cms_id="CMS-1848-P", effective_date="2026-01-01")) + _, status = s.upsert_status( + _rule(cms_id="CMS-1848-P", effective_date="", fr_page="12345") + ) + assert status == "updated" + stored = s.get(key) + # New key applied... + assert stored.fr_page == "12345" + # ...but the blank incoming effective_date did not blank the stored one. + assert stored.effective_date == "2026-01-01" + assert stored.cms_id == "CMS-1848-P" diff --git a/tests/cli/test_bib_exercise.py b/tests/cli/test_bib_exercise.py index 911a95c..cedfdd9 100644 --- a/tests/cli/test_bib_exercise.py +++ b/tests/cli/test_bib_exercise.py @@ -287,6 +287,7 @@ class TestFetchPfsComments: @patch("bib.translate.federal_register") @patch("bib.regulations_gov.Client") def test_no_docket(self, mc_client_cls, mc_translate, mc_split, mc_pfs, mc_connect): + from bib.item import Rule from bib.store import Store store = Store(":memory:", storage_dir="/tmp/nope") @@ -299,7 +300,7 @@ class TestFetchPfsComments: doc.html_url = "https://example.com" mc_pfs.return_value = [doc] - rule = MagicMock() + rule = Rule(title="CY2024 PFS NPRM", url="https://example.com") mc_translate.return_value = rule api = MagicMock() @@ -312,6 +313,50 @@ class TestFetchPfsComments: assert result.exit_code == 0 assert "skip" in result.output + @patch("bib.connect") + @patch("bib.federalregister.pfs_rules") + @patch("bib.federalregister.split_docket_ids", return_value=["CMS-1676-P"]) + @patch("bib.translate.federal_register") + @patch("bib.regulations_gov.Client") + def test_unresolvable_docket_still_upserts_rule( + self, mc_client_cls, mc_translate, mc_split, mc_pfs, mc_connect + ): + """refs #680: a rule whose only CMS id never resolves to a reg.gov + docket must still be upserted — the upsert used to live only + inside the per-docket loop, so it was skipped entirely here.""" + from bib.item import Rule + from bib.store import Store + + store = Store(":memory:", storage_dir="/tmp/nope") + mc_connect.return_value = store + + doc = MagicMock() + doc.type = "Proposed Rule" + doc.publication_date = "2023-01-01" + doc.dockets = ["CMS-1676-P"] + doc.html_url = "https://example.com/unresolvable" + mc_pfs.return_value = [doc] + + rule = Rule(title="CY2024 PFS NPRM", url="https://example.com/unresolvable") + mc_translate.return_value = rule + + api = MagicMock() + api.__enter__ = MagicMock(return_value=api) + api.__exit__ = MagicMock(return_value=False) + mc_client_cls.return_value = api + api.resolve_docket.return_value = None + + result = runner.invoke(app, ["fetch-pfs-comments"]) + assert result.exit_code == 0 + + con = store._con() # noqa: SLF001 + row = con.execute( + "SELECT title FROM items WHERE url = ?", + ("https://example.com/unresolvable",), + ).fetchone() + assert row is not None + assert row["title"] == "CY2024 PFS NPRM" + class TestIngestMail: @patch("bib.connect") diff --git a/tests/cli/test_comments.py b/tests/cli/test_comments.py index 59a467e..b93b39e 100644 --- a/tests/cli/test_comments.py +++ b/tests/cli/test_comments.py @@ -48,6 +48,33 @@ def test_extract_ocr_empty_queue(tmp_path: Path): assert "ocr queue: 0" in result.output +def test_extract_ocr_closes_bib_lookup_connection(tmp_path: Path, monkeypatch): + """refs #680: _bib_lookup_factory opens its own read connection to + bib.sqlite (via a `.close` attribute on the returned callable — + the same contract `extract()` honours in its `finally`); extract_ocr + used to never call it, leaking the connection.""" + cdir = tmp_path / "CMS-2024-0001" / "CMS-2024-0001-0001" + cdir.mkdir(parents=True) + (cdir / "combined.md").write_text( + "---\nattachments:\n- status: ocr_needed\n---\nbody\n" + ) + + closed = {"called": False} + + def fake_lookup(comment_id: str) -> str: + return "" + + fake_lookup.close = lambda: closed.__setitem__("called", True) + + monkeypatch.setattr("cli.comments._bib_lookup_factory", lambda use_bib: fake_lookup) + monkeypatch.setattr("rex.comments.combine.extract_comment", lambda *a, **k: None) + monkeypatch.setattr("rex.comments.ocr.RapidOcrEngine", lambda: lambda p: "") + + result = runner.invoke(app, ["extract-ocr", "--root", str(tmp_path)]) + assert result.exit_code == 0, result.output + assert closed["called"] is True + + def test_stats_after_extract(tmp_path: Path): _seed(tmp_path) runner.invoke( @@ -127,3 +154,69 @@ def test_extract_attaches_combined_md_as_note(tmp_path: Path, monkeypatch): runner.invoke(app, ["extract", "--root", str(tmp_path), "--workers", "1"]) assert len(notes()) == 1 assert md_attachments() == [] + + +def test_extract_reattach_repairs_missing_note_for_current_dir( + tmp_path: Path, monkeypatch +): + """refs #680: --reattach at the CLI level is the repair path — a dir + already "current" (combined.md up to date) is normally skipped + without touching bib at all, so a note lost from under it (e.g. a + bad Zotero sync) never comes back on a plain re-run. --reattach + re-fires the attach callback for every skipped dir.""" + from bib import connect + from bib.item import Source + + bib_db = tmp_path / "bib.sqlite" + monkeypatch.setattr("conf.path", lambda _: bib_db) + + cdir = _seed(tmp_path) + comment_id = cdir.name + + store = connect(str(bib_db)) + item = Source( + title=f"Comment {comment_id}", + url=f"https://www.regulations.gov/comment/{comment_id}", + ) + item_key = store.upsert(item) + store.close() + + def note_count() -> int: + store = connect(str(bib_db)) + n = ( + store._con() + .execute( + "SELECT count(*) FROM notes n JOIN items i ON n.item_id = i.id " + "WHERE i.key = ?", + (item_key,), + ) + .fetchone()[0] + ) + store.close() + return n + + result = runner.invoke(app, ["extract", "--root", str(tmp_path), "--workers", "1"]) + assert result.exit_code == 0, result.output + assert note_count() == 1 + + # The note vanishes without combined.md changing at all — the dir + # stays "current" and a plain re-run must skip it untouched. + store = connect(str(bib_db)) + store._con().execute( + "DELETE FROM notes WHERE item_id = (SELECT id FROM items WHERE key = ?)", + (item_key,), + ) + store._con().commit() + store.close() + assert note_count() == 0 + + result = runner.invoke(app, ["extract", "--root", str(tmp_path), "--workers", "1"]) + assert result.exit_code == 0, result.output + assert "skipped: 1" in result.output.lower() + assert note_count() == 0 # not repaired without --reattach + + result = runner.invoke( + app, ["extract", "--root", str(tmp_path), "--workers", "1", "--reattach"] + ) + assert result.exit_code == 0, result.output + assert note_count() == 1 # repaired diff --git a/tests/cli/test_pfs_cli.py b/tests/cli/test_pfs_cli.py index 72af1ab..a424bf3 100644 --- a/tests/cli/test_pfs_cli.py +++ b/tests/cli/test_pfs_cli.py @@ -99,29 +99,38 @@ def con(monkeypatch, restore_families): c.close() +def _empty_extractions(store, con, codes, *, classify=None): + """``pfs.extract.extract_codes``'s shape with nothing extracted — the + stand-in for tests that only care which codes the CLI targeted.""" + return {code: Extraction(code, (), ()) for code in codes} + + class TestElements: def test_writes_rows_and_publishes(self, con, monkeypatch): - def fake_extract(store, con_, code, *, classify=None): - return Extraction( - code, - ( - ElementRow( - code, - 2026, - "activity", - "consent", - "", - "Consent;", - "K", - 1, - 2, - "fr", + def fake_extract(store, con_, codes, *, classify=None): + return { + code: Extraction( + code, + ( + ElementRow( + code, + 2026, + "activity", + "consent", + "", + "Consent;", + "K", + 1, + 2, + "fr", + ), ), - ), - (), - ) + (), + ) + for code in codes + } - monkeypatch.setattr(pfs_cli, "extract_code", fake_extract) + monkeypatch.setattr(pfs_cli, "extract_codes", fake_extract) res = runner.invoke(app, ["pfs", "elements", "--code", "g0556", "--no-llm"]) assert res.exit_code == 0, res.output assert "G0556: 1 elements, 0 for review" in res.output @@ -131,40 +140,43 @@ class TestElements: def test_summary_line_reports_confirmed_count_when_nonzero(self, con, monkeypatch): # #703: the elements summary surfaces confirmed_by, but only when # there's something to report. - def fake_extract(store, con_, code, *, classify=None): - return Extraction( - code, - ( - ElementRow( - code, - 2026, - "activity", - "consent", - "", - "Consent;", - "K", - 1, - 2, - "fr", - "cpt:GQGTPGYV", + def fake_extract(store, con_, codes, *, classify=None): + return { + code: Extraction( + code, + ( + ElementRow( + code, + 2026, + "activity", + "consent", + "", + "Consent;", + "K", + 1, + 2, + "fr", + "cpt:GQGTPGYV", + ), + ElementRow( + code, + 2026, + "activity", + "24-7-access", + "", + "24/7;", + "K", + 3, + 2, + "fr", + ), ), - ElementRow( - code, - 2026, - "activity", - "24-7-access", - "", - "24/7;", - "K", - 3, - 2, - "fr", - ), - ), - (), - ) + (), + ) + for code in codes + } - monkeypatch.setattr(pfs_cli, "extract_code", fake_extract) + monkeypatch.setattr(pfs_cli, "extract_codes", fake_extract) res = runner.invoke(app, ["pfs", "elements", "--code", "g0556", "--no-llm"]) assert res.exit_code == 0, res.output assert "G0556: 2 elements, 0 for review, 1 confirmed by a second source" in ( @@ -172,21 +184,13 @@ class TestElements: ) def test_summary_line_omits_confirmed_when_none(self, con, monkeypatch): - monkeypatch.setattr( - pfs_cli, - "extract_code", - lambda s, c, code, *, classify=None: Extraction(code, (), ()), - ) + monkeypatch.setattr(pfs_cli, "extract_codes", _empty_extractions) res = runner.invoke(app, ["pfs", "elements", "--code", "g0556", "--no-llm"]) assert res.exit_code == 0, res.output assert "confirmed" not in res.output def test_dry_run_does_not_write(self, con, monkeypatch): - monkeypatch.setattr( - pfs_cli, - "extract_code", - lambda s, c, code, *, classify=None: Extraction(code, (), ()), - ) + monkeypatch.setattr(pfs_cli, "extract_codes", _empty_extractions) def fail_batch(): raise AssertionError("--dry-run must not open a RW duckdb_batch connection") @@ -213,9 +217,9 @@ class TestElements: seen = [] monkeypatch.setattr( pfs_cli, - "extract_code", - lambda s, c, code, *, classify=None: ( - seen.append(code) or Extraction(code, (), ()) + "extract_codes", + lambda s, c, codes, *, classify=None: ( + seen.extend(codes) or _empty_extractions(s, c, codes) ), ) res = runner.invoke( @@ -230,9 +234,9 @@ class TestElements: seen = [] monkeypatch.setattr( pfs_cli, - "extract_code", - lambda s, c, code, *, classify=None: ( - seen.append(code) or Extraction(code, (), ()) + "extract_codes", + lambda s, c, codes, *, classify=None: ( + seen.extend(codes) or _empty_extractions(s, c, codes) ), ) res = runner.invoke( @@ -256,17 +260,17 @@ class TestElements: class TestCodesForWarnSlow: - """``_codes_for``'s "~20 min of SQL" warning describes the - per-code LLM classification pass ``elements`` runs — it does not - apply to ``lineage --all-payable`` (one inverted pass, seconds).""" + """``_codes_for``'s target-count note is for ``elements``, whose + ``--all-payable`` run makes one model call per unplaced line — not for + ``lineage --all-payable``, which is pure SQL/regex.""" def test_warns_by_default(self, con, caplog): pfs_cli._codes_for(con, [], [], True) - assert "20 min" in caplog.text + assert "targeting 1 codes" in caplog.text def test_silent_when_warn_slow_is_false(self, con, caplog): pfs_cli._codes_for(con, [], [], True, warn_slow=False) - assert "20 min" not in caplog.text + assert "targeting" not in caplog.text class TestLineage: diff --git a/tests/dev/test_dedupe_attachments.py b/tests/dev/test_dedupe_attachments.py index ed7db3e..b95d007 100644 --- a/tests/dev/test_dedupe_attachments.py +++ b/tests/dev/test_dedupe_attachments.py @@ -98,6 +98,27 @@ def test_refuses_group_with_differing_sizes(tmp_path: Path): assert rep.rows_removed == 0 and rep.skipped_conflicts == 1 +def test_conflict_when_kept_file_is_missing(tmp_path: Path): + """refs #680: the kept row's file being gone (e.g. an earlier + partial cleanup, or a manual delete) must not be treated as "no + conflict" just because a missing file's _size() sentinel (-1) + happens to also mean "ignore mismatches" for individual dupes — a + duplicate with real bytes on disk while the kept copy has none is + exactly the size-mismatch case this guard exists for.""" + s, _ = _seed(tmp_path) + (tmp_path / "storage" / "AAAAAAAA" / "attachment_1.pdf").unlink() + groups = mod.plan(s._con()) + assert len(groups) == 1 + assert groups[0].conflict is True + rep = mod.apply(s._con(), groups) + assert rep.rows_removed == 0 + assert rep.skipped_conflicts == 1 + # nothing touched: all three rows and both surviving files remain + assert s._con().execute("SELECT count(*) FROM attachments").fetchone()[0] == 3 + assert (tmp_path / "storage" / "BBBBBBBB" / "attachment_1.pdf").is_file() + assert (tmp_path / "storage" / "CCCCCCCC" / "attachment_1.pdf").is_file() + + def test_unique_index_created_only_when_clean(tmp_path: Path): s, _ = _seed(tmp_path) s.close() @@ -117,6 +138,63 @@ def test_unique_index_created_only_when_clean(tmp_path: Path): assert "idx_attachments_item_filename" in idx +class _ConnSpy: + """Wraps a real sqlite3.Connection to record GROUP BY calls. + + ``sqlite3.Connection`` is a C type and can't be monkeypatched + in-place, so this proxies everything through ``__getattr__`` except + ``execute`` (recorded) and attribute assignment (forwarded — needed + for ``row_factory``, which ``Store._con`` sets directly).""" + + def __init__(self, real: sqlite3.Connection) -> None: + object.__setattr__(self, "_real", real) + object.__setattr__(self, "group_by_calls", []) + + def execute(self, sql, *args, **kwargs): + if "GROUP BY item_id, filename" in sql: + self.group_by_calls.append(sql) + return self._real.execute(sql, *args, **kwargs) + + def __getattr__(self, name): + return getattr(self._real, name) + + def __setattr__(self, name, value): + setattr(self._real, name, value) + + +def test_reopen_skips_group_by_scan_once_indexed(tmp_path: Path, monkeypatch): + """refs #680: Store._init_schema short-circuits via sqlite_master + once the unique index exists — a re-open must not re-run the + duplicate-attachments GROUP BY scan at all.""" + s, _ = _seed(tmp_path) + mod.apply(s._con(), mod.plan(s._con())) + s.close() + + # First reopen creates the unique index (no duplicates left). + s_setup = Store(str(tmp_path / "bib.sqlite"), storage_dir=tmp_path / "storage") + s_setup.close() + + import bib.store as store_mod + + real_connect = store_mod.sqlite3.connect + spies: list[_ConnSpy] = [] + + def fake_connect(path, *args, **kwargs): + spy = _ConnSpy(real_connect(path, *args, **kwargs)) + spies.append(spy) + return spy + + monkeypatch.setattr(store_mod.sqlite3, "connect", fake_connect) + + # A further reopen must skip the GROUP BY scan now that the index + # already exists. + s2 = Store(str(tmp_path / "bib.sqlite"), storage_dir=tmp_path / "storage") + s2.close() + + assert len(spies) == 1 + assert spies[0].group_by_calls == [] + + class _FailAfterFirstDelete: """Connection proxy that dies partway through the transaction.""" diff --git a/tests/llm/test_client.py b/tests/llm/test_client.py new file mode 100644 index 0000000..df77ad8 --- /dev/null +++ b/tests/llm/test_client.py @@ -0,0 +1,272 @@ +"""``llm.client.LlmClient`` — httpx-only client for the llm service (#573). + +All requests go through ``httpx.MockTransport``; no real network I/O. +""" + +from __future__ import annotations + +import json + +import httpx +import pytest + +from llm.client import LlmClient, _resolve_base_url + + +def _client(handler, **kw) -> LlmClient: + """An LlmClient whose internal httpx.Client is wired to *handler*.""" + c = LlmClient(base_url="http://llm-test:8000", **kw) + transport = httpx.MockTransport(handler) + c._client = lambda: httpx.Client( # noqa: SLF001 — test seam + base_url=c.base_url, timeout=c.timeout, transport=transport + ) + return c + + +# ── base URL resolution ────────────────────────────────────────────── + + +def test_explicit_base_url_wins(): + assert _resolve_base_url("http://example:9000/") == "http://example:9000" + + +def test_env_var_used_when_no_explicit_base_url(monkeypatch): + monkeypatch.setenv("LLM_URL", "http://from-env:1234/") + assert _resolve_base_url(None) == "http://from-env:1234" + + +def test_falls_back_to_compose_name_when_it_resolves(monkeypatch): + monkeypatch.delenv("LLM_URL", raising=False) + monkeypatch.setattr("llm.client.socket.gethostbyname", lambda host: "192.168.5.9") + assert _resolve_base_url(None) == "http://llm:8000" + + +def test_falls_back_to_localhost_when_compose_name_does_not_resolve(monkeypatch): + import socket + + monkeypatch.delenv("LLM_URL", raising=False) + + def _raise(host): + raise socket.gaierror("not found") + + monkeypatch.setattr("llm.client.socket.gethostbyname", _raise) + assert _resolve_base_url(None) == "http://localhost:8000" + + +def test_default_constructor_resolves_base_url(monkeypatch): + monkeypatch.delenv("LLM_URL", raising=False) + monkeypatch.setattr( + "llm.client.socket.gethostbyname", + lambda host: (_ for _ in ()).throw(OSError("no")), + ) + client = LlmClient() + assert client.base_url == "http://localhost:8000" + assert client.timeout == 30.0 + + +# ── health ──────────────────────────────────────────────────────────── + + +def test_health_true_on_ok(): + def handler(request: httpx.Request) -> httpx.Response: + assert request.url.path == "/health" + return httpx.Response(200, json={"status": "ok"}) + + assert _client(handler).health() is True + + +def test_health_false_on_bad_status(): + def handler(request: httpx.Request) -> httpx.Response: + return httpx.Response(503, json={"status": "down"}) + + assert _client(handler).health() is False + + +def test_health_false_on_connect_error(): + def handler(request: httpx.Request) -> httpx.Response: + raise httpx.ConnectError("refused", request=request) + + assert _client(handler).health() is False + + +def test_health_false_on_bad_json(): + def handler(request: httpx.Request) -> httpx.Response: + return httpx.Response(200, text="not json") + + assert _client(handler).health() is False + + +# ── search ──────────────────────────────────────────────────────────── + + +def test_search_sends_expected_query_params_and_returns_results(): + seen = {} + + def handler(request: httpx.Request) -> httpx.Response: + seen["path"] = request.url.path + seen["params"] = dict(request.url.params) + return httpx.Response( + 200, + json={ + "query": "telehealth", + "filters": {}, + "total": 1, + "results": [ + { + "label": "CMS-2025-0304-1", + "url": "https://x", + "date": "2025-01-01", + "snippet": "hi", + } + ], + }, + ) + + results = _client(handler).search( + "telehealth", + collection="comments", + docket="CMS-2025-0304", + item_key="ABC123", + year=2025, + kind="comment", + limit=5, + offset=2, + ) + assert seen["path"] == "/search" + assert seen["params"] == { + "q": "telehealth", + "collection": "comments", + "limit": "5", + "offset": "2", + "docket": "CMS-2025-0304", + "item_key": "ABC123", + "year": "2025", + "kind": "comment", + } + assert results == [ + { + "label": "CMS-2025-0304-1", + "url": "https://x", + "date": "2025-01-01", + "snippet": "hi", + } + ] + + +def test_search_omits_unset_optional_filters(): + seen = {} + + def handler(request: httpx.Request) -> httpx.Response: + seen["params"] = dict(request.url.params) + return httpx.Response(200, json={"results": []}) + + _client(handler).search("q") + assert seen["params"] == { + "q": "q", + "collection": "all", + "limit": "10", + "offset": "0", + } + + +def test_search_raises_httperror_on_bad_status(): + def handler(request: httpx.Request) -> httpx.Response: + return httpx.Response(400, json={"detail": "unknown collection"}) + + with pytest.raises(httpx.HTTPError): + _client(handler).search("q", collection="nope") + + +def test_search_raises_httperror_on_connect_error(): + def handler(request: httpx.Request) -> httpx.Response: + raise httpx.ConnectError("refused", request=request) + + with pytest.raises(httpx.HTTPError): + _client(handler).search("q") + + +# ── similar ─────────────────────────────────────────────────────────── + + +def test_similar_hits_expected_path_and_returns_results(): + seen = {} + + def handler(request: httpx.Request) -> httpx.Response: + seen["path"] = request.url.path + seen["params"] = dict(request.url.params) + return httpx.Response( + 200, json={"key": "ABC123", "total": 1, "results": [{"label": "x"}]} + ) + + results = _client(handler).similar("ABC123", collection="rules", limit=3) + assert seen["path"] == "/similar/ABC123" + assert seen["params"] == {"collection": "rules", "limit": "3"} + assert results == [{"label": "x"}] + + +def test_similar_raises_httperror_on_404(): + def handler(request: httpx.Request) -> httpx.Response: + return httpx.Response(404, json={"detail": "no indexed chunks"}) + + with pytest.raises(httpx.HTTPError): + _client(handler).similar("nope") + + +# ── chat (SSE) ──────────────────────────────────────────────────────── + + +def _sse_body(*events: dict) -> bytes: + return "".join(f"data: {json.dumps(e)}\n\n" for e in events).encode() + + +def test_chat_parses_sse_events_in_order(): + events = [ + {"type": "token", "text": "Hel"}, + {"type": "token", "text": "lo"}, + {"type": "done"}, + ] + + def handler(request: httpx.Request) -> httpx.Response: + assert request.method == "POST" + assert json.loads(request.content) == {"question": "hi?", "mode": "auto"} + return httpx.Response(200, content=_sse_body(*events)) + + got = list(_client(handler).chat("hi?")) + assert got == events + + +def test_chat_includes_since_and_mode_when_given(): + seen = {} + + def handler(request: httpx.Request) -> httpx.Response: + seen["body"] = json.loads(request.content) + return httpx.Response(200, content=_sse_body({"type": "done"})) + + list(_client(handler).chat("q", mode="timeline", since="2025-01-01")) + assert seen["body"] == {"question": "q", "mode": "timeline", "since": "2025-01-01"} + + +def test_chat_ignores_blank_and_non_data_lines(): + raw = b': comment line\n\ndata: {"type": "done"}\n\n\n' + + def handler(request: httpx.Request) -> httpx.Response: + return httpx.Response(200, content=raw) + + got = list(_client(handler).chat("q")) + assert got == [{"type": "done"}] + + +def test_chat_raises_httperror_on_bad_status(): + def handler(request: httpx.Request) -> httpx.Response: + return httpx.Response(400, json={"detail": "empty question"}) + + with pytest.raises(httpx.HTTPError): + list(_client(handler).chat("")) + + +def test_chat_raises_httperror_on_connect_error(): + def handler(request: httpx.Request) -> httpx.Response: + raise httpx.ConnectError("refused", request=request) + + with pytest.raises(httpx.HTTPError): + list(_client(handler).chat("q")) diff --git a/tests/llm/test_init_lazy.py b/tests/llm/test_init_lazy.py index 0e39c9e..6d706a3 100644 --- a/tests/llm/test_init_lazy.py +++ b/tests/llm/test_init_lazy.py @@ -2,7 +2,11 @@ ``llm`` extra (no langchain, no sqlalchemy). ``llm.config``, ``llm.links`` and ``llm.lineage`` must import there anyway — the pool exports are resolved lazily, and ``llm.evidence`` only touches -sqlalchemy when a pgvector query actually runs.""" +sqlalchemy when a pgvector query actually runs. + +#573: ``llm.client`` (the notebook-side ``LlmClient``) is httpx-only and +must import there too — it's the one module notebooks are expected to +import directly.""" from __future__ import annotations @@ -21,8 +25,9 @@ def _fake(name, *a, **k): return _real(name, *a, **k) builtins.__import__ = _fake import llm.config, llm.links, llm.lineage, llm.evidence # must not need langchain/sqlalchemy +import llm.client # #573: notebook-side LlmClient — httpx only import llm -print('ok', llm.LlmConfig.__name__) +print('ok', llm.LlmConfig.__name__, llm.client.LlmClient.__name__) """ @@ -31,7 +36,7 @@ def test_config_links_lineage_evidence_import_without_langchain_or_sqlalchemy(): [sys.executable, "-c", _PROBE], capture_output=True, text=True, check=False ) assert r.returncode == 0, r.stderr[-800:] - assert "ok LlmConfig" in r.stdout + assert "ok LlmConfig LlmClient" in r.stdout def test_pool_exports_resolve_lazily(): diff --git a/tests/llm/test_lineage.py b/tests/llm/test_lineage.py index 551df3b..4f1fcb3 100644 --- a/tests/llm/test_lineage.py +++ b/tests/llm/test_lineage.py @@ -113,6 +113,12 @@ class _FakeStore: "2026-07-16", "https://www.federalregister.gov/d/2026-2027doc", ), + ( + "PV8APQ4A", + "Chronic Care Management Services (MLN909188, June 2025)", + "2025-06-01", + "https://www.cms.gov/files/document/chroniccaremanagement.pdf", + ), ], ) self.con.executemany( @@ -167,11 +173,13 @@ class _FakeStore: def get(self, key: str) -> SimpleNamespace: row = self.con.execute( - "SELECT title, date_published FROM items WHERE key = ?", (key,) + "SELECT title, date_published, url FROM items WHERE key = ?", (key,) ).fetchone() if row is None: raise KeyError(key) - return SimpleNamespace(title=row["title"], date_published=row["date_published"]) + return SimpleNamespace( + title=row["title"], date_published=row["date_published"], url=row["url"] + ) def close(self): self.con.close() @@ -219,9 +227,11 @@ def _el(code, year, type_, value, item_key="", p_id=0, source="fr") -> ElementRo def _gd( - family, code, kind, locator, item_key_src, p_id_src, item_key="" + family, code, kind, locator, item_key_src, p_id_src, item_key="", page=0 ) -> GuidanceRow: - return GuidanceRow(family, code, kind, locator, item_key, item_key_src, p_id_src, 0) + return GuidanceRow( + family, code, kind, locator, item_key, item_key_src, p_id_src, 0, page + ) class TestKindRank: @@ -636,7 +646,7 @@ class TestGuidance: out = lineage._collect_guidance(c, store, ["APCM"], ["APCM"], 1) assert out == () - def test_iom_mln_have_no_url(self, con, store): + def test_unresolved_iom_mln_have_no_url(self, con, store): c, _ = con write_guidance( c, @@ -647,6 +657,40 @@ class TestGuidance: assert out[0].url == "" assert out[0].label == "CY2021 PFS final 85 FR 84547 ¶686" + def test_resolved_mln_links_to_the_bib_item_url(self, con, store): + # #705 item 3: pfs.guidance.resolve_mln fills item_key; the chat + # link is the booklet PDF itself. #705 item 2: a located IOM + # section adds ``#page=N`` to the chapter PDF's URL. + c, _ = con + write_guidance( + c, + "CCM", + [ + _gd("CCM", "99490", "mln", "MLN 909188", "YBM4IZUS", 686, "PV8APQ4A"), + _gd("CCM", "99490", "iom", "100-04 ch.12", "YBM4IZUS", 686, "GONE1234"), + _gd( + "CCM", + "99490", + "iom", + "100-04 ch.12 §30.6.4", + "YBM4IZUS", + 686, + "PV8APQ4A", + 39, + ), + ], + ) + out = lineage._collect_guidance(c, store, ["CCM"], [], 1) + by = {g.locator: g for g in out} + assert ( + by["MLN 909188"].url + == "https://www.cms.gov/files/document/chroniccaremanagement.pdf" + ) + assert by["100-04 ch.12"].url == "" # item_key set but not in the bib + assert by["100-04 ch.12 §30.6.4"].url == ( + "https://www.cms.gov/files/document/chroniccaremanagement.pdf#page=39" + ) + 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) == () diff --git a/tests/llm/test_pages.py b/tests/llm/test_pages.py index 453b9eb..d9ee772 100644 --- a/tests/llm/test_pages.py +++ b/tests/llm/test_pages.py @@ -3,6 +3,7 @@ import fitz import pytest +from llm import pages as pages_mod from llm.chunk import Chunk, Doc from llm.pages import enrich_pdf_pages, locate, pdf_pages @@ -113,3 +114,34 @@ class TestEnrich: ) (out,) = enrich_pdf_pages(doc, [c]) assert out.metadata["page"] == "9" + + +# ── locate_section (#705 item 2: IOM section headings) ────────────── + + +class TestLocateSection: + TOC = ( + "Table of Contents (Rev. 12780) 30.6.3 - Payment for Immunosuppressive " + "Therapy Management 30.6.4 - Evaluation and Management (E/M) Services " + "Furnished Incident to Physician's Service 30.6.5 - Physicians in Group" + ) + BODY = ( + "visit is for immunosuppressive therapy. 30.6.4 - Evaluation and " + "Management (E/M) Services Furnished Incident to Physician's Service by " + "Nonphysician Practitioners (Rev. 11288; Issued: 03-31-22) A. General" + ) + + def test_body_heading_beats_the_table_of_contents(self): + assert pages_mod.locate_section([self.TOC, "filler", self.BODY], "30.6.4") == 3 + + def test_toc_only_is_not_a_location(self): + assert pages_mod.locate_section([self.TOC], "30.6.4") == 0 + + def test_prefix_numbers_do_not_match(self): + # "30.6.4" must not fire on "130.6.4" or "30.6.4.1" + body = "130.6.4 - Other (Rev. 1) 30.6.4.1 - Sub (Rev. 2)" + assert pages_mod.locate_section([body], "30.6.4") == 0 + + def test_trailing_period_and_blank_section(self): + assert pages_mod.locate_section([self.BODY], "30.6.4.") == 1 + assert pages_mod.locate_section([self.BODY], "") == 0 diff --git a/tests/notebooks/test_llm_search_nb.py b/tests/notebooks/test_llm_search_nb.py new file mode 100644 index 0000000..e8999a9 --- /dev/null +++ b/tests/notebooks/test_llm_search_nb.py @@ -0,0 +1,107 @@ +"""notebooks/llm_search.py — structure and headless degradation (#573).""" + +from __future__ import annotations + +import ast +import importlib.util +from pathlib import Path + +import pytest + +NB = Path(__file__).resolve().parents[2] / "notebooks" / "llm_search.py" + + +def _load(): + spec = importlib.util.spec_from_file_location("llm_search_nb", NB) + mod = importlib.util.module_from_spec(spec) + spec.loader.exec_module(mod) + return mod + + +def test_notebook_is_a_marimo_app(): + mod = _load() + assert mod.app.__class__.__name__ == "App" + + +def test_cells_are_anonymous(): + src = NB.read_text() + tree = ast.parse(src) + names = [n.name for n in tree.body if isinstance(n, ast.FunctionDef)] + assert names and set(names) == {"_"} + + +def test_headless_run_degrades_without_service_or_data(monkeypatch): + """With the llm service unreachable and no replica/bib, every guard cell + should render a note instead of raising.""" + mod = _load() + + import conf.connect as cc + from llm.client import LlmClient + + monkeypatch.setattr(LlmClient, "health", lambda self: False) + monkeypatch.setattr( + cc, + "duckdb", + lambda *a, **k: (_ for _ in ()).throw(FileNotFoundError("no replica")), + ) + monkeypatch.setattr( + cc, "bib", lambda *a, **k: (_ for _ in ()).throw(FileNotFoundError("no bib")) + ) + + outputs, _defs = mod.app.run() + assert outputs is not None + rendered = "\n".join(o._repr_html_() for o in outputs if hasattr(o, "_repr_html_")) + assert "Traceback" not in rendered + assert "llm service unreachable" in rendered + assert "replica unavailable" in rendered + assert "bibliography unavailable" in rendered + + +def test_headless_run_still_renders_with_service_up_but_no_data(monkeypatch): + """Health true but no replica/bib and no real search results (network + still stubbed out) — the search cell should degrade to "no hits", not + raise, and the duckdb cell should still show "replica unavailable".""" + mod = _load() + + import conf.connect as cc + from llm.client import LlmClient + + monkeypatch.setattr(LlmClient, "health", lambda self: True) + monkeypatch.setattr( + LlmClient, + "search", + lambda self, *a, **k: (_ for _ in ()).throw( + __import__("httpx").ConnectError("refused") + ), + ) + monkeypatch.setattr( + cc, + "duckdb", + lambda *a, **k: (_ for _ in ()).throw(FileNotFoundError("no replica")), + ) + monkeypatch.setattr( + cc, "bib", lambda *a, **k: (_ for _ in ()).throw(FileNotFoundError("no bib")) + ) + + outputs, _defs = mod.app.run() + rendered = "\n".join(o._repr_html_() for o in outputs if hasattr(o, "_repr_html_")) + assert "Traceback" not in rendered + assert "replica unavailable" in rendered + + +_ROOT = Path(__file__).resolve().parents[2] +_HAS_DATA = (_ROOT / "data" / "replica" / "aco.ro.duckdb").exists() and ( + _ROOT / "data" / "bib.sqlite" +).exists() + + +@pytest.mark.skipif(not _HAS_DATA, reason="needs the live replica and bib") +def test_docket_picker_offers_known_dockets_against_real_bib(monkeypatch): + """Live smoke test: with the real bib store, the docket dropdown's + options come from ``store.dockets()`` and are non-empty.""" + from llm.client import LlmClient + + monkeypatch.setattr(LlmClient, "health", lambda self: False) + mod = _load() + outputs, _defs = mod.app.run() + assert outputs is not None diff --git a/tests/pfs/test_descriptors.py b/tests/pfs/test_descriptors.py index f293f04..d378b84 100644 --- a/tests/pfs/test_descriptors.py +++ b/tests/pfs/test_descriptors.py @@ -9,10 +9,14 @@ import pytest from pfs.descriptors import ( DescriptorRun, + _stem_pattern_any, descriptor_runs, + descriptor_runs_bucketed, hcpcs_long_description, + hcpcs_long_descriptions, rule_year_of, rvu_descriptions, + rvu_descriptions_bucketed, ) @@ -293,3 +297,164 @@ class TestDescriptorSpan: ] assert run.stem.text.startswith("G0502 (Initial psychiatric") assert "Chronic care management prose" not in run.stem.text + + +@pytest.fixture +def wide_store(): + """Three rule items naming several codes, so the inverted pass has to + reset its open runs at an item boundary, carry two stems through the + same element paragraphs, and order one code's runs across items.""" + s = _Store() + s.con.executemany( + "INSERT INTO items VALUES (?,?,?)", + [ + ("AAAAAAAA", "Medicare Program; CY 2015 PFS Final Rule", "2014-11-13"), + ("BBBBBBBB", "Medicare Program; CY 2021 PFS Final Rule", "2020-12-28"), + ("CCCCCCCC", "Revisions to Payment Policies", "2016-11-15"), + ], + ) + rows = [ + # AAAAAAAA — a plain stem run, then an unrelated prose paragraph + ( + "AAAAAAAA", + 10, + 100, + "Comment: commenters noted the CPT panel created a code.", + ), + ( + "AAAAAAAA", + 11, + 100, + "We use the new CPT code 99490 (Chronic care management services, at least " + "20 minutes of clinical staff time directed by a physician or other " + "qualified health care professional, per calendar month, with the " + "following required elements:", + ), + ("AAAAAAAA", 12, 100, "Consent;"), + ( + "AAAAAAAA", + 13, + 101, + "Comprehensive care plan established, implemented, revised, or monitored).", + ), + ("AAAAAAAA", 14, 101, "Response: it is our preference to use CPT codes."), + # BBBBBBBB — two stems back to back, so 99487's run must absorb the + # element paragraphs that follow 99489's stem too + ( + "BBBBBBBB", + 20, + 200, + "CPT code 99487 (Complex chronic care management services, with the " + "following required elements:", + ), + ( + "BBBBBBBB", + 21, + 200, + "99489 (Each additional 30 minutes of clinical staff time, per calendar " + "month, with the following required elements:", + ), + ("BBBBBBBB", 22, 200, "Consent;"), + ( + "BBBBBBBB", + 23, + 201, + "Provide 24/7 access for urgent needs to care team/practitioner;", + ), + ("BBBBBBBB", 24, 201, "We received many comments on this proposal."), + ( + "BBBBBBBB", + 25, + 201, + "( 9) 99439 (code for non-complex chronic care management).", + ), + # CCCCCCCC — an earlier-dated item mentioning 99490 again, so the + # per-code ORDER BY date_published puts it first + ( + "CCCCCCCC", + 30, + 300, + "We finalized 99490 (Chronic care management services, per calendar " + "month, with the following required elements:", + ), + ("CCCCCCCC", 31, 300, "Consent;"), + # an odd-length target code, exercising the literal branch of + # `_stem_pattern_any` + ( + "CCCCCCCC", + 32, + 300, + "And HCPCS code G0556X (Advanced primary care management services, per " + "calendar month, with the following elements:", + ), + ("CCCCCCCC", 33, 300, "Consent;"), + ] + s.con.executemany( + "INSERT INTO fr_anchors VALUES (?,?,?,?,?)", + [(k, pid, pg, pid, t) for k, pid, pg, t in rows], + ) + yield s + s.close() + + +class TestDescriptorRunsBucketed: + """#698: one streaming pass over ``fr_anchors`` must reproduce + ``descriptor_runs`` exactly, code for code.""" + + _CODES = ("99490", "99487", "99489", "99439", "G0556X", "00000") + + def test_matches_descriptor_runs_per_code(self, wide_store): + bucketed = descriptor_runs_bucketed(wide_store, self._CODES) + assert sorted(bucketed) == sorted(self._CODES) + for c in self._CODES: + assert bucketed[c] == descriptor_runs(wide_store, c), c + + def test_finds_the_runs_the_fixture_plants(self, wide_store): + bucketed = descriptor_runs_bucketed(wide_store, self._CODES) + # 99490 stems in two items, ordered by the rule's publication date + assert [r.item_key for r in bucketed["99490"]] == ["AAAAAAAA", "CCCCCCCC"] + # 99487's run keeps taking element paragraphs past 99489's stem + assert [p.p_id for p in bucketed["99487"][0].elements] == [21, 22, 23] + assert [p.p_id for p in bucketed["99489"][0].elements] == [22, 23] + # an enumeration line is not a stem, and an unmentioned code is empty + assert bucketed["99439"] == [] and bucketed["00000"] == [] + # the odd-length literal branch of `_stem_pattern_any` still matches + assert [r.stem.p_id for r in bucketed["G0556X"]] == [32] + + def test_max_elements_cap_matches(self, wide_store): + bucketed = descriptor_runs_bucketed(wide_store, ["99487"], max_elements=2) + assert bucketed["99487"] == descriptor_runs(wide_store, "99487", max_elements=2) + assert [p.p_id for p in bucketed["99487"][0].elements] == [21, 22] + + def test_no_codes_is_an_empty_result(self, wide_store): + assert descriptor_runs_bucketed(wide_store, []) == {} + + def test_generic_token_only_when_every_target_is_five_wide(self): + # The all-five-wide fast path compiles the character class alone; + # an odd-length target adds its own literal alternative. + assert "|" not in _stem_pattern_any(["99490", "G0556"]).pattern + assert "G0556X" in _stem_pattern_any(["99490", "G0556X"]).pattern + + +class TestBucketedReplicaLookups: + """The DuckDB half of the inversion: one query per table for every + target code, matching the per-code helpers row for row.""" + + def test_hcpcs_long_descriptions_matches_per_code(self, con): + codes = ["G0556", "G9999"] + bucketed = hcpcs_long_descriptions(con, codes) + for c in codes: + assert bucketed.get(c, "") == hcpcs_long_description(con, c), c + assert bucketed["G0556"].startswith("Advanced primary care") + assert "G9999" not in bucketed + + def test_rvu_descriptions_matches_per_code(self, con): + codes = ["99490", "99999"] + bucketed = rvu_descriptions_bucketed(con, codes) + for c in codes: + assert bucketed.get(c, []) == rvu_descriptions(con, c), c + assert [y for y, _s, _d in bucketed["99490"]] == [2015, 2022] + + def test_no_codes_is_an_empty_result(self, con): + assert hcpcs_long_descriptions(con, []) == {} + assert rvu_descriptions_bucketed(con, []) == {} diff --git a/tests/pfs/test_extract.py b/tests/pfs/test_extract.py index d926288..29c3966 100644 --- a/tests/pfs/test_extract.py +++ b/tests/pfs/test_extract.py @@ -12,11 +12,14 @@ from pfs.descriptors import DescriptorRun, Para, descriptor_runs from pfs.extract import ( Extraction, _cpt_elements, + _cpt_elements_bucketed, _merge_or_confirm, extract_code, + extract_codes, extract_run, extract_text, ) +from pfs.families import HAND_FAMILIES STEM = Para( "JJ6AM5HJ", @@ -534,3 +537,192 @@ class TestMergeOrConfirm: _merge_or_confirm(merged, _el(source="hcpcs")) (row,) = merged.values() assert row.confirmed_by == "cpt:GQGTPGYV,hcpcs:" + + +#: The 17 hand-family codes, the fixture universe #698 is measured +#: against: whatever `extract_codes` does in one inverted pass must match +#: `extract_code`'s per-code path for every one of them. +FIXTURE_CODES = tuple(c for fam in HAND_FAMILIES.values() for c in fam.codes) + + +def _wide_store_con(): + con = _sqlite_store_con() + con.executemany( + "INSERT INTO items VALUES (?,?,?)", + [ + ("AAAAAAAA", "Medicare Program; CY 2015 PFS Final Rule", "2014-11-13"), + ("BBBBBBBB", "Medicare Program; CY 2025 PFS Final Rule", "2024-11-01"), + ], + ) + rows = [ + ("AAAAAAAA", 10, 100, "Comment: commenters noted the CPT panel created codes."), + ( + "AAAAAAAA", + 11, + 100, + "We adopted CPT code 99490 (Chronic care management services, at least 20 " + "minutes of clinical staff time directed by a physician or other " + "qualified health care professional, per calendar month, with the " + "following required elements:", + ), + ("AAAAAAAA", 12, 100, "Consent;"), + ( + "AAAAAAAA", + 13, + 101, + "Chronic conditions place the patient at significant risk of death, acute " + "exacerbation/decompensation, or functional decline;", + ), + ("AAAAAAAA", 14, 101, "A line the vocabulary does not know about at all;"), + ( + "AAAAAAAA", + 15, + 101, + "Comprehensive care plan established, implemented, revised, or monitored).", + ), + ("AAAAAAAA", 16, 101, "Response: we agree with the commenters."), + ( + "BBBBBBBB", + 20, + 200, + "HCPCS code G0556 ( Advanced primary care management services provided by " + "clinical staff and directed by a physician, per calendar month, with the " + "following elements, as appropriate:", + ), + ("BBBBBBBB", 21, 200, "Consent;"), + ( + "BBBBBBBB", + 22, + 200, + "Provide 24/7 access for urgent needs to care team/practitioner;", + ), + ( + "BBBBBBBB", + 23, + 201, + "CPT code 99495 (Transitional care management services, with the following " + "required elements:", + ), + ("BBBBBBBB", 24, 201, "Consent;"), + ("BBBBBBBB", 25, 201, "( 9) 99439 (code for non-complex chronic care)."), + ] + con.executemany( + "INSERT INTO fr_anchors VALUES (?,?,?,?,?)", + [(k, pid, pg, pid, t) for k, pid, pg, t in rows], + ) + return con + + +class TestExtractCodesBucketed: + """#698: `extract_codes` inverts the per-code loop — one streaming + pass over `fr_anchors` plus one query each over `pfs.cpt_code`, + `terminology.hcpcs_level_2` and `pfs.rvu` — and must yield exactly + what `extract_code` yields, code for code.""" + + @pytest.fixture + def store(self): + con = _wide_store_con() + yield _Store(con) + con.close() + + @pytest.fixture + def con(self): + c = _duckdb_con() + c.executemany( + "INSERT INTO pfs.rvu VALUES (?,?,?,?,?,?)", + [ + ("99490", "", "Chron care mgmt srvc 20 min", "A", 1.0, 2015), + ("99490", None, "Chrnc care mgmt staff 1st 20", "A", 1.2, 2026), + ("99490", "26", "ignored modifier row", "A", 1.0, 2026), + ("G0556", "", "Adv prim care mgmt lvl 1, consent", "A", 2.0, 2026), + ("99497", "", "Advncd care plan, per calendar month", "A", 1.0, 2026), + ], + ) + c.executemany( + "INSERT INTO terminology.hcpcs_level_2 VALUES (?,?,?,?)", + [ + ("G0556", "… per calendar month, consent …", "10", "3"), + ("G0556", "a lower-seqnum transcription", "5", "1"), + ("G0557", "… 24/7 access, per calendar month …", "1", "1"), + ], + ) + _insert_cpt_code( + c, + 2022, + "OLDEDITN", + "99490", + stem="An older edition's stem", + elements=["Consent;"], + tail="per calendar month.", + ) + _insert_cpt_code( + c, + 2024, + "GQGTPGYV", + "99490", + stem="Chronic care management services", + elements=["Consent;", "A line the vocabulary does not know at all;"], + tail="first 20 minutes, per calendar month.", + ) + _insert_cpt_code( + c, + 2024, + "TCMENTRY", + "99495", + stem="Transitional care management services", + elements=["Consent;"], + tail="within 14 days of discharge.", + ) + yield c + c.close() + + def test_matches_extract_code_for_every_fixture_code(self, store, con): + bucketed = extract_codes(store, con, FIXTURE_CODES) + assert sorted(bucketed) == sorted(set(FIXTURE_CODES)) + for c in FIXTURE_CODES: + assert bucketed[c] == extract_code(store, con, c), c + # the fixture really exercises all four sources, so the equality + # above is not a comparison of 17 empty extractions + sources = {r.source for x in bucketed.values() for r in x.rows} + assert sources == {"fr", "cpt", "hcpcs", "rvu"} + assert any(x.reviews for x in bucketed.values()) + + def test_matches_extract_code_with_a_classifier(self, store, con): + classify = lambda text, choices: ( # noqa: E731 + "community-coordination" if "vocabulary" in text else None + ) + bucketed = extract_codes(store, con, FIXTURE_CODES, classify=classify) + for c in FIXTURE_CODES: + assert bucketed[c] == extract_code(store, con, c, classify=classify), c + assert any( + r.value == "community-coordination" + for x in bucketed.values() + for r in x.rows + ) + + def test_lowercase_input_is_normalised_and_deduped(self, store, con): + assert list(extract_codes(store, con, ["g0556", "G0556", "99490"])) == [ + "99490", + "G0556", + ] + + def test_no_codes_is_an_empty_result(self, store, con): + assert extract_codes(store, con, []) == {} + + def test_replica_with_no_cpt_tables_at_all_does_not_raise(self, store): + # Same I4 tolerance the per-code path has: `elements --dry-run` + # never calls ensure_tables, so a pre-cpt-ingest replica must + # degrade to no CPT rows, not a CatalogException. + con = _duckdb_con_no_cpt() + try: + bucketed = extract_codes(store, con, ["99490"]) + finally: + con.close() + assert not any(r.source == "cpt" for r in bucketed["99490"].rows) + + def test_non_missing_table_error_propagates(self): + with pytest.raises(RuntimeError, match="disk I/O error"): + _cpt_elements_bucketed(TestCptElementsErrors._RaisingCon(), ["99490"]) + + def test_no_codes_never_touches_the_replica(self): + assert _cpt_elements_bucketed(TestCptElementsErrors._RaisingCon(), []) == {} diff --git a/tests/pfs/test_guidance.py b/tests/pfs/test_guidance.py index 9597d51..5f19baf 100644 --- a/tests/pfs/test_guidance.py +++ b/tests/pfs/test_guidance.py @@ -25,13 +25,17 @@ from pfs.guidance import ( MLN_RE, _cfr_refs, _extract, + _guidance_page, _iom_refs, build, dedupe, harvest, harvest_cpt, + iom_section_page, + mln_refs, resolve_cfr, resolve_iom, + resolve_mln, ) # ── real fr_anchors snippets ────────────────────────────────────────── @@ -601,3 +605,196 @@ class TestBuild: assert harvest_cpt(bare, s, ("99490",), family="CCM") == [] finally: bare.close() + + +# ── MLN (#705 item 3) ─────────────────────────────────────────────── + + +class TestMlnRefs: + def test_icn_prefix_and_icn_mln_combo_normalise_to_mln(self): + text = ( + "MLN Booklet “Chronic Care Management Services” (ICN MLN909188, July " + "2019); see also “Advance Care Planning” (ICN 909289, August 2016)." + ) + assert mln_refs(text) == ["MLN 909188", "MLN 909289"] + + def test_matters_article_forms(self): + assert mln_refs("MLN Matters® Number MM9603: https://x") == [ + "MLN Matters MM9603" + ] + assert mln_refs("as stated in MLN Matters article SE1316, issued") == [ + "MLN Matters SE1316" + ] + assert mln_refs("documentation is available in MLN Matters SE 1316.") == [ + "MLN Matters SE1316" + ] + # a bare number after "Matters" is a change-request article + assert mln_refs("consistent with MLN Matters article 11268.") == [ + "MLN Matters MM11268" + ] + + def test_prose_mentions_without_a_number_yield_nothing(self): + assert ( + mln_refs("We will issue an MLN Matters article once the CR is out.") == [] + ) + assert mln_refs("Medicare Learning Network (MLN) Matters® article,") == [] + + def test_first_occurrence_wins(self): + assert mln_refs("MLN006764 ... again MLN 006764") == ["MLN 006764"] + + +@pytest.fixture +def mln_store(): + s = Store(":memory:") + booklet = s.create( + Manual( + title="Chronic Care Management Services (MLN909188, June 2025) — CCM", + url="https://www.cms.gov/files/document/chroniccaremanagement.pdf", + ) + ) + em_guide = s.create( + Manual( + title="Evaluation and Management Services Guide", + url=( + "https://www.cms.gov/Outreach-and-Education/Medicare-Learning-" + "Network-MLN/MLNProducts/Downloads/eval-mgmt-serv-guide-ICN006764.pdf" + ), + ) + ) + article = s.create( + Manual( + title="MLN Matters MM9603", + url=( + "https://www.cms.gov/Outreach-and-Education/Medicare-Learning-" + "Network-MLN/MLNMattersArticles/Downloads/MM9603.pdf" + ), + ) + ) + decoy = s.create( + Manual(title="Unrelated", url="https://www.cms.gov/files/MM96030.pdf") + ) + yield s, booklet, em_guide, article, decoy + s.close() + + +class TestResolveMln: + def test_title_carries_the_number(self, mln_store): + s, booklet, *_ = mln_store + assert resolve_mln(s, "MLN 909188") == booklet + + def test_pre_2020_icn_spelling_in_url(self, mln_store): + s, _b, em_guide, *_ = mln_store + assert resolve_mln(s, "MLN 006764") == em_guide + + def test_matters_article_by_url_word_bounded(self, mln_store): + s, _b, _e, article, _decoy = mln_store + # MM9603 must not match the decoy's MM96030 + assert resolve_mln(s, "MLN Matters MM9603") == article + assert resolve_mln(s, "MLN Matters MM9604") == "" + + def test_unresolved_and_empty(self, mln_store): + s, *_ = mln_store + assert resolve_mln(s, "MLN 907166") == "" + assert resolve_mln(s, "") == "" + + +class TestExtractMlnResolved: + def test_extract_resolves_against_store(self, mln_store): + s, booklet, *_ = mln_store + rows = _extract("Refer to the CCM booklet (ICN MLN909188, July 2019).", s) + assert rows == [("mln", "MLN 909188", booklet)] + + +# ── IOM section page locating (#705 item 2) ───────────────────────── + + +class TestIomSectionPage: + TOC = "Table of Contents 30.6.4 - E/M Services Incident to 30.6.5 - Groups" + BODY = "30.6.4 - E/M Services Incident to (Rev. 11288; Issued: 03-31-22) A." + + @pytest.fixture(autouse=True) + def _fresh_caches(self, monkeypatch): + monkeypatch.setattr(guidance_mod, "_PAGE_CACHE", {}) + monkeypatch.setattr(guidance_mod, "_PDF_CACHE", {}) + + @pytest.fixture + def chapter(self, tmp_path, monkeypatch): + # Own store: an in-memory Store's default storage dir is ./storage + # relative to the cwd, which attach_file would create in the repo. + s = Store(":memory:", storage_dir=tmp_path / "storage") + manual_key = s.create( + Manual(title="Medicare Claims Processing Manual — Chapter 12: X") + ) + pdf = tmp_path / "clm104c12.pdf" + pdf.write_bytes(b"%PDF-1.4 stub") + s.attach_file(manual_key, pdf) + calls: list[str] = [] + + def fake_pages(path): + calls.append(str(path)) + return [self.TOC, "filler", self.BODY] + + monkeypatch.setattr("llm.pages.pdf_pages", fake_pages) + yield s, manual_key, calls + s.close() + + def test_locates_body_heading_past_the_toc(self, chapter): + s, key, _ = chapter + assert iom_section_page(s, key, "30.6.4") == 3 + + def test_pdf_read_once_and_page_cached(self, chapter): + s, key, calls = chapter + assert iom_section_page(s, key, "30.6.4") == 3 + assert iom_section_page(s, key, "30.6.4") == 3 + assert iom_section_page(s, key, "30.6.5") == 0 # TOC-only + assert len(calls) == 1 + + def test_no_pdf_attachment_or_blank_args(self, store): + s, _sec, manual_key = store + assert iom_section_page(s, manual_key, "30.6.4") == 0 + assert iom_section_page(s, "", "30.6.4") == 0 + assert iom_section_page(s, manual_key, "") == 0 + + def test_guidance_page_only_for_resolved_iom_sections(self, chapter): + s, key, _ = chapter + assert _guidance_page(s, "iom", "100-04 ch.12 §30.6.4", key) == 3 + assert _guidance_page(s, "iom", "100-04 ch.12", key) == 0 + assert _guidance_page(s, "iom", "100-04 ch.12 §30.6.4", "") == 0 + assert _guidance_page(s, "cfr", "42 CFR 410.78", key) == 0 + + def test_harvest_carries_the_located_page(self, store_with_anchors, monkeypatch): + s, _item, _sec, manual_key = store_with_anchors + monkeypatch.setattr( + guidance_mod, + "iom_section_page", + lambda st, k, sec: 39 if k == manual_key else 0, + ) + rows = harvest(s, ("99490", "99439"), family="CCM") + iom = [r for r in rows if r.kind == "iom"] + assert iom and all(r.page == 39 for r in iom) + assert all(r.page == 0 for r in rows if r.kind != "iom") + + +class TestGuidanceTablePage: + def test_page_round_trips(self, con): + row = GuidanceRow( + "CCM", "99490", "iom", "100-04 ch.12 §30.6.4", "K", "S", 1, 2, 39 + ) + write_guidance(con, "CCM", [row]) + assert read_guidance(con, "CCM") == [row] + + def test_legacy_table_without_page_column_reads_zero(self): + c = duckdb.connect(":memory:") + c.execute("CREATE SCHEMA pfs") + c.execute( + "CREATE TABLE pfs.code_guidance (family VARCHAR, code VARCHAR, kind VARCHAR, " + "locator VARCHAR, item_key VARCHAR, item_key_src VARCHAR, p_id_src INTEGER, " + "page_src INTEGER)" + ) + c.execute( + "INSERT INTO pfs.code_guidance VALUES ('CCM','99490','iom','x','','S',1,2)" + ) + ensure_tables(c) # adds the column in place; the old row reads NULL + got = read_guidance(c, "CCM") + assert got == [GuidanceRow("CCM", "99490", "iom", "x", "", "S", 1, 2, 0)] + c.close()