feat(comments): LLM position/theme classifier for rulemaking comments (refs #254)

rex.comments.classify: relevance filter (skin-sub terms, CTP, Q41xx/
C527x code families), forced-tool-call classification via the
provider-agnostic prisma.llm layer — position (5-point scale on the
reclassification proposal), themes (9 tags), stakeholder type (#255's
per-comment half), rule provisions (#256), and commenter/organization
from letterhead/signature (seeds #416). Enum-validated so a model
glitch can't contaminate the table.

dev/scripts/classify_comments.py: per-docket driver — reads comment
text from bib (combined.md note preferred, inline abstract fallback),
filters to the relevant subset (CMS-2025-0304: 384 of 14,092),
classifies resumably into a JSONL cache, loads
skin_subs.rulemaking_comments via duckdb_batch (replacing the DEMO
pilot rows).

18 unit tests (fake provider — no network). Live run smoke-tested to
the API boundary; the at-scale run is blocked on Anthropic API
credits.

dev/scripts/classify_comments.py

@@ -0,0 +1,212 @@
+"""Classify skin-substitute-relevant rulemaking comments (#254).
+
+Pipeline per docket:
+  1. Pull every comment item for the docket from bib — full extracted
+     text (the "Comment text" note, i.e. combined.md) when present,
+     inline body (abstract) otherwise.
+  2. Keep the skin-substitute-relevant subset (rex.comments.classify
+     RELEVANCE_PAT) — OPPS rules draw thousands of comments on
+     unrelated provisions.
+  3. LLM-classify each relevant comment (position / themes /
+     stakeholder / provisions / commenter+org) via prisma.llm.
+     Resumable: results append to a JSONL cache keyed by comment key;
+     re-runs only classify new comments.
+  4. Load the results into skin_subs.rulemaking_comments (replacing
+     that docket's rows and any DEMO- pilot rows).
+
+Usage:
+    uv run python dev/scripts/classify_comments.py --docket CMS-2025-0304
+    uv run python dev/scripts/classify_comments.py --limit 5 --dry-run
+"""
+
+from __future__ import annotations
+
+import argparse
+import json
+import sqlite3
+from pathlib import Path
+
+ROOT = Path(__file__).resolve().parents[2]
+BIB_PATH = ROOT / "data" / "bib.sqlite"
+
+DEFAULT_DOCKET = "CMS-2025-0304"
+DEFAULT_MODEL = "claude-haiku-4-5"
+
+
+def load_comments(docket: str) -> list[dict]:
+    """One record per comment item: key, title, date, text, has_note."""
+    con = sqlite3.connect(BIB_PATH)
+    con.row_factory = sqlite3.Row
+    rows = con.execute(
+        """
+        SELECT i.id, i.key, i.title, i.date_published, i.abstract,
+               (SELECT content FROM notes n
+                WHERE n.item_id = i.id AND n.title = 'Comment text') AS note
+        FROM items i
+        JOIN item_tags it ON i.id = it.item_id
+        JOIN tags t ON it.tag_id = t.id
+        WHERE t.name = ?
+        """,
+        (f"reg-docket:{docket}",),
+    ).fetchall()
+    con.close()
+
+    from rex.comments.classify import strip_html
+
+    out = []
+    for r in rows:
+        text = strip_html(r["note"]) if r["note"] else (r["abstract"] or "")
+        out.append(
+            {
+                "key": r["key"],
+                "comment_id": r["title"] or r["key"],
+                "posted_date": r["date_published"] or "",
+                "text": text,
+                "has_attachments": bool(r["note"]),
+            }
+        )
+    return out
+
+
+def load_cache(path: Path) -> dict[str, dict]:
+    cache: dict[str, dict] = {}
+    if path.exists():
+        for line in path.read_text().splitlines():
+            if line.strip():
+                rec = json.loads(line)
+                cache[rec["key"]] = rec
+    return cache
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--docket", default=DEFAULT_DOCKET)
+    parser.add_argument("--model", default=DEFAULT_MODEL)
+    parser.add_argument("--limit", type=int, default=0, help="cap LLM calls this run")
+    parser.add_argument(
+        "--dry-run",
+        action="store_true",
+        help="relevance-filter and report only; no LLM calls, no DB write",
+    )
+    parser.add_argument(
+        "--no-load", action="store_true", help="classify but skip the DuckDB load"
+    )
+    args = parser.parse_args()
+
+    from rex.comments.classify import is_relevant
+
+    comments = load_comments(args.docket)
+    relevant = [c for c in comments if is_relevant(c["text"])]
+    print(f"{args.docket}: {len(comments)} comments, {len(relevant)} skin-sub relevant")
+    if args.dry_run:
+        for c in relevant[:10]:
+            print(f"  {c['comment_id'][:40]:40s} {len(c['text']):>7} chars")
+        return 0
+
+    cache_path = ROOT / "data" / "cms" / f"comments-classified-{args.docket}.jsonl"
+    cache_path.parent.mkdir(parents=True, exist_ok=True)
+    cache = load_cache(cache_path)
+    todo = [c for c in relevant if c["key"] not in cache]
+    if args.limit:
+        todo = todo[: args.limit]
+    print(f"cached: {len(cache)}, to classify: {len(todo)}")
+
+    if todo:
+        import os
+
+        os.environ["PRISMA_LLM_MODEL"] = args.model
+        from prisma.llm import make_provider
+        from rex.comments.classify import classify
+
+        provider = make_provider()
+        errors = 0
+        with cache_path.open("a") as fh:
+            for i, c in enumerate(todo, 1):
+                try:
+                    rec = classify(provider, c["text"], title=c["comment_id"])
+                except Exception as e:  # noqa: BLE001 — skip and continue the batch
+                    errors += 1
+                    print(f"  [{i}/{len(todo)}] {c['comment_id'][:36]} ERROR: {e}")
+                    continue
+                rec.update(
+                    key=c["key"],
+                    comment_id=c["comment_id"],
+                    docket_id=args.docket,
+                    posted_date=c["posted_date"],
+                    has_attachments=c["has_attachments"],
+                    text_length=len(c["text"]),
+                )
+                cache[c["key"]] = rec
+                fh.write(json.dumps(rec) + "\n")
+                fh.flush()
+                print(
+                    f"  [{i}/{len(todo)}] {c['comment_id'][:36]:36s} "
+                    f"{rec['position']:16s} {rec['stakeholder_type']}"
+                )
+        if errors:
+            print(f"{errors} comments failed to classify (rerun to retry)")
+
+    if args.no_load:
+        return 0
+
+    rows = [r for r in cache.values() if r.get("docket_id") == args.docket]
+    if not rows:
+        print("nothing to load")
+        return 0
+
+    from conf.connect import duckdb_batch
+
+    with duckdb_batch("aco") as con:
+        con.execute("CREATE SCHEMA IF NOT EXISTS skin_subs")
+        con.execute("""
+            CREATE TABLE IF NOT EXISTS skin_subs.rulemaking_comments (
+                comment_id VARCHAR, docket_id VARCHAR, commenter_name VARCHAR,
+                organization VARCHAR, posted_date VARCHAR, has_attachments BOOLEAN,
+                text_length BIGINT, position VARCHAR, position_score DOUBLE,
+                themes VARCHAR, stakeholder_type VARCHAR, coordination_group VARCHAR,
+                is_form_letter BOOLEAN, provisions VARCHAR
+            )
+        """)
+        # Replace this docket's rows; sweep any DEMO- pilot rows too.
+        con.execute(
+            "DELETE FROM skin_subs.rulemaking_comments "
+            "WHERE docket_id = ? OR comment_id LIKE 'DEMO-%'",
+            [args.docket],
+        )
+        con.executemany(
+            """
+            INSERT INTO skin_subs.rulemaking_comments
+              (comment_id, docket_id, commenter_name, organization, posted_date,
+               has_attachments, text_length, position, position_score, themes,
+               stakeholder_type, coordination_group, is_form_letter, provisions)
+            VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, NULL, NULL, ?)
+            """,
+            [
+                (
+                    r["comment_id"],
+                    r["docket_id"],
+                    r.get("commenter_name", ""),
+                    r.get("organization", ""),
+                    r.get("posted_date", ""),
+                    r.get("has_attachments", False),
+                    r.get("text_length", 0),
+                    r["position"],
+                    float(r["position_score"]),
+                    r.get("themes", ""),
+                    r.get("stakeholder_type", "unknown"),
+                    r.get("provisions", ""),
+                )
+                for r in rows
+            ],
+        )
+        n = con.execute(
+            "SELECT count(*) FROM skin_subs.rulemaking_comments WHERE docket_id = ?",
+            [args.docket],
+        ).fetchone()[0]
+        print(f"skin_subs.rulemaking_comments: {n} rows for {args.docket}")
+
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

src/rex/comments/classify.py

@@ -0,0 +1,207 @@
+"""Position + theme classification for rulemaking comments (#254).
+
+Classifies each skin-substitute-relevant comment on a 5-point position
+scale (about CMS's proposed reclassification/flat-rate payment), tags it
+with policy themes, maps it to rule provisions, and identifies the
+stakeholder type and commenter/organization from the letterhead or
+signature when the regulations.gov API fields are empty.
+
+Pure logic (relevance filter, output schema, prompt build, result
+validation) is separated from the LLM transport (``prisma.llm``
+Provider) so tests run without a network.
+"""
+
+from __future__ import annotations
+
+import html
+import re
+from typing import Any
+
+from prisma.llm import LLMCall, LLMMessage, LLMTool, Provider
+
+# ── Vocabulary (issues #254 / #255 / #256) ──────────────────────
+
+POSITIONS: dict[str, int] = {
+    "strongly_oppose": -2,
+    "oppose": -1,
+    "neutral": 0,
+    "support": 1,
+    "strongly_support": 2,
+}
+
+THEMES = (
+    "asp_methodology",
+    "patient_access",
+    "innovation_impact",
+    "fraud_waste_abuse",
+    "clinical_evidence",
+    "manufacturer_impact",
+    "provider_impact",
+    "geographic_disparities",
+    "flat_rate_design",
+)
+
+STAKEHOLDERS = (
+    "manufacturer",
+    "distributor",
+    "provider",
+    "wound_care_society",
+    "patient_advocacy",
+    "payer",
+    "government",
+    "academic",
+    "beneficiary",
+    "unknown",
+)
+
+PROVISIONS = (
+    "reclassification",
+    "flat_rate",
+    "hcpcs_c_codes",
+    "pass_through_elimination",
+    "documentation_requirements",
+    "transition_timeline",
+    "cost_category_removal",
+)
+
+# Skin-substitute relevance: product terms, the CTP acronym, and the
+# HCPCS code families (Q41xx biologics, C5271-78 application codes).
+RELEVANCE_PAT = re.compile(
+    r"skin\s+substitut"
+    r"|cellular.{0,15}tissue[- ]based"
+    r"|\bCTPs?\b"
+    r"|\bQ41\d\d\b"
+    r"|\bC527[1-8]\b",
+    re.IGNORECASE,
+)
+
+MAX_TEXT_CHARS = 24_000
+
+_TAG_RE = re.compile(r"<[^>]+>")
+
+
+def strip_html(text: str) -> str:
+    """Zotero-envelope note HTML → plain text (whitespace collapsed)."""
+    plain = html.unescape(_TAG_RE.sub(" ", text or ""))
+    return re.sub(r"[ \t]+", " ", plain).strip()
+
+
+def is_relevant(text: str) -> bool:
+    return bool(RELEVANCE_PAT.search(text or ""))
+
+
+# ── LLM call ────────────────────────────────────────────────────
+
+CLASSIFY_TOOL = LLMTool(
+    name="classify_comment",
+    description="Record the classification of one rulemaking comment.",
+    schema={
+        "type": "object",
+        "properties": {
+            "position": {
+                "type": "string",
+                "enum": list(POSITIONS),
+                "description": "Stance on CMS's proposed skin-substitute "
+                "reclassification (drugs/biologicals → incident-to supplies) "
+                "and flat-rate payment. oppose = wants ASP+6% preserved; "
+                "support = agrees with the proposal / wants stricter controls.",
+            },
+            "themes": {
+                "type": "array",
+                "items": {"type": "string", "enum": list(THEMES)},
+                "description": "Every policy theme the comment substantively raises.",
+            },
+            "stakeholder_type": {
+                "type": "string",
+                "enum": list(STAKEHOLDERS),
+                "description": "Who is commenting, judged from letterhead, "
+                "signature, and self-description.",
+            },
+            "provisions": {
+                "type": "array",
+                "items": {"type": "string", "enum": list(PROVISIONS)},
+                "description": "Specific rule provisions the comment addresses.",
+            },
+            "commenter_name": {
+                "type": "string",
+                "description": "Person who signed/submitted, from signature "
+                "line or letterhead. Empty string if not determinable.",
+            },
+            "organization": {
+                "type": "string",
+                "description": "Organization on whose behalf the comment is "
+                "made. Empty string if private individual / not determinable.",
+            },
+            "rationale": {
+                "type": "string",
+                "description": "One sentence: why this position label.",
+            },
+        },
+        "required": ["position", "themes", "stakeholder_type", "provisions"],
+    },
+)
+
+_SYSTEM = """\
+You classify public comments submitted on CMS's CY2026 OPPS proposed rule
+(CMS-1834-P, docket CMS-2025-0304), specifically regarding the proposal to
+reclassify skin substitutes (cellular and tissue-based products, CTPs) from
+drugs/biologicals paid at ASP+6% to incident-to supplies paid a flat packaged
+rate (~$127.28/cm²), replacing Q-codes with C5271–C5278.
+
+Position semantics:
+- strongly_oppose: wants ASP+6% preserved; rejects reclassification outright
+- oppose: against the proposal, citing access/innovation concerns
+- neutral: mixed — acknowledges the fraud/spending problem but fears overcorrection
+- support: agrees skin substitutes should be reclassified/paid flat
+- strongly_support: cites fraud evidence, urges the change or stricter controls
+
+Base every field only on the comment text given. Use the classify_comment tool."""
+
+
+def build_call(text: str, *, title: str = "") -> LLMCall:
+    body = text[:MAX_TEXT_CHARS]
+    user = f"Comment {title}:\n\n{body}" if title else body
+    return LLMCall(
+        messages=[
+            LLMMessage(role="system", content=_SYSTEM, cache=True),
+            LLMMessage(role="user", content=user),
+        ],
+        tools=[CLASSIFY_TOOL],
+        force_tool="classify_comment",
+        max_tokens=1024,
+    )
+
+
+def validate(raw: dict[str, Any]) -> dict[str, Any]:
+    """Coerce a tool-call payload into a clean classification record.
+
+    Unknown enum values are dropped (arrays) or rejected (position) so a
+    model glitch can't contaminate the table.
+    """
+    position = raw.get("position", "")
+    if position not in POSITIONS:
+        raise ValueError(f"invalid position: {position!r}")
+    themes = [t for t in raw.get("themes", []) if t in THEMES]
+    provisions = [p for p in raw.get("provisions", []) if p in PROVISIONS]
+    stakeholder = raw.get("stakeholder_type", "unknown")
+    if stakeholder not in STAKEHOLDERS:
+        stakeholder = "unknown"
+    return {
+        "position": position,
+        "position_score": POSITIONS[position],
+        "themes": "; ".join(dict.fromkeys(themes)),
+        "stakeholder_type": stakeholder,
+        "provisions": "; ".join(dict.fromkeys(provisions)),
+        "commenter_name": (raw.get("commenter_name") or "").strip()[:200],
+        "organization": (raw.get("organization") or "").strip()[:200],
+        "rationale": (raw.get("rationale") or "").strip()[:500],
+    }
+
+
+def classify(provider: Provider, text: str, *, title: str = "") -> dict[str, Any]:
+    """Classify one comment; raises ValueError on an unusable response."""
+    result = provider.complete(build_call(text, title=title))
+    for call in result.tool_calls:
+        if call["name"] == "classify_comment":
+            return validate(call["input"])
+    raise ValueError(f"no classify_comment tool call in response: {result.text[:200]}")

tests/rex/comments/test_classify.py

@@ -0,0 +1,152 @@
+"""Tests for rex.comments.classify — pure logic, no network.
+
+The LLM transport is exercised with a fake Provider; prompt/schema
+contents are asserted structurally so vocabulary drift breaks loudly.
+"""
+
+from __future__ import annotations
+
+import pytest
+
+from prisma.llm import LLMResult
+from rex.comments.classify import (
+    CLASSIFY_TOOL,
+    POSITIONS,
+    PROVISIONS,
+    STAKEHOLDERS,
+    THEMES,
+    build_call,
+    classify,
+    is_relevant,
+    strip_html,
+    validate,
+)
+
+
+class TestRelevance:
+    @pytest.mark.parametrize(
+        "text",
+        [
+            "We oppose the skin substitute payment change.",
+            "Skin Substitutes should stay at ASP+6%",
+            "cellular and tissue-based products (CTPs)",
+            "the new C5271 application code",
+            "HCPCS Q4186 pricing",
+        ],
+    )
+    def test_relevant(self, text):
+        assert is_relevant(text)
+
+    @pytest.mark.parametrize(
+        "text",
+        [
+            "I am concerned about telehealth reimbursement.",
+            "OT students deserve better PFS treatment",
+            "",
+        ],
+    )
+    def test_irrelevant(self, text):
+        assert not is_relevant(text)
+
+
+class TestStripHtml:
+    def test_strips_tags_and_entities(self):
+        raw = '<div class="zotero-note"><h2>Inline</h2><p>ASP&amp;6% &lt;rate&gt;</p></div>'
+        assert strip_html(raw) == "Inline ASP&6% <rate>"
+
+
+class TestValidate:
+    def _raw(self, **over):
+        base = {
+            "position": "oppose",
+            "themes": ["patient_access", "asp_methodology"],
+            "stakeholder_type": "provider",
+            "provisions": ["flat_rate"],
+            "commenter_name": "Dr. A",
+            "organization": "Clinic B",
+            "rationale": "cites access",
+        }
+        base.update(over)
+        return base
+
+    def test_happy_path(self):
+        rec = validate(self._raw())
+        assert rec["position_score"] == -1
+        assert rec["themes"] == "patient_access; asp_methodology"
+        assert rec["provisions"] == "flat_rate"
+
+    def test_invalid_position_rejected(self):
+        with pytest.raises(ValueError, match="invalid position"):
+            validate(self._raw(position="meh"))
+
+    def test_unknown_enum_values_dropped(self):
+        rec = validate(
+            self._raw(
+                themes=["patient_access", "made_up"],
+                provisions=["flat_rate", "nonsense"],
+                stakeholder_type="alien",
+            )
+        )
+        assert rec["themes"] == "patient_access"
+        assert rec["provisions"] == "flat_rate"
+        assert rec["stakeholder_type"] == "unknown"
+
+    def test_dedupes_preserving_order(self):
+        rec = validate(self._raw(themes=["flat_rate_design", "flat_rate_design"]))
+        assert rec["themes"] == "flat_rate_design"
+
+
+class TestBuildCall:
+    def test_shape(self):
+        call = build_call("some comment text", title="CMS-2025-0304-0004")
+        assert call.force_tool == "classify_comment"
+        assert call.tools == [CLASSIFY_TOOL]
+        assert call.messages[0].role == "system"
+        assert call.messages[0].cache is True
+        assert "CMS-2025-0304-0004" in call.messages[1].content
+
+    def test_truncates_long_text(self):
+        call = build_call("x" * 100_000)
+        assert len(call.messages[1].content) <= 30_000
+
+    def test_schema_vocab_in_sync(self):
+        props = CLASSIFY_TOOL.schema["properties"]
+        assert props["position"]["enum"] == list(POSITIONS)
+        assert props["themes"]["items"]["enum"] == list(THEMES)
+        assert props["stakeholder_type"]["enum"] == list(STAKEHOLDERS)
+        assert props["provisions"]["items"]["enum"] == list(PROVISIONS)
+
+
+class _FakeProvider:
+    def __init__(self, tool_calls):
+        self._tool_calls = tool_calls
+        self.calls = []
+
+    def complete(self, call):
+        self.calls.append(call)
+        return LLMResult(text="", tool_calls=self._tool_calls, usage={})
+
+
+class TestClassify:
+    def test_returns_validated_record(self):
+        provider = _FakeProvider(
+            [
+                {
+                    "name": "classify_comment",
+                    "input": {
+                        "position": "strongly_support",
+                        "themes": ["fraud_waste_abuse"],
+                        "stakeholder_type": "government",
+                        "provisions": ["reclassification"],
+                    },
+                }
+            ]
+        )
+        rec = classify(provider, "text", title="C-1")
+        assert rec["position_score"] == 2
+        assert provider.calls[0].force_tool == "classify_comment"
+
+    def test_no_tool_call_raises(self):
+        provider = _FakeProvider([])
+        with pytest.raises(ValueError, match="no classify_comment tool call"):
+            classify(provider, "text")