feat(bib): backfill-comments --mirror — bulk enrichment from the Mirrulations S3 mirror (refs #662)

docs/superpowers/plans/2026-08-19-mirrulations-backfill-p43.md

@@ -0,0 +1,54 @@
+# Mirrulations Mirror Backfill Implementation Plan
+
+> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.
+
+**Goal:** Replace the 7-day regulations.gov API crawl with a bulk pull from the Mirrulations public S3 mirror (AWS Open Data) — hours instead of days, zero API-key pressure (refs #662).
+
+**Architecture:** `s3://mirrulations` (public, HTTPS-listable, no auth) mirrors every docket: `raw-data/CMS/{D}/text-{D}/comments/{cid}.json` is byte-compatible with the v4 `/comments/{id}` detail response (`{"data": {...}}`, no `included` block), and `raw-data/CMS/{D}/binary-{D}/comments_attachments/{cid}_attachment_N.{ext}` holds the attachment binaries. The mirror is a superset of our farm (15,368 vs 13,245 for CMS-2018-0076; 3,433 vs 2,728 for the live docket). A new `backfill_from_mirror` in `bib/regulations_gov.py` reuses `_parse_comment` + `upsert_comment` (creation of mirror-only comments) and the enrichment steps of `backfill_details` (abstract, org tag, attach_file, `enriched:ok`), with S3 I/O fanned out over a thread pool and ALL sqlite work on the main thread (same pattern as `prisma/fetch.py`). VMs are unnecessary: the constraint was the per-key API cap, which the mirror sidesteps entirely; the reg.gov API remains only for freshness gap-fill (#615).
+
+**Tech Stack:** httpx (S3 REST list + GET), `concurrent.futures.ThreadPoolExecutor`, existing `bib.store.Store` / `Source` / `Tag` machinery; pytest + `httpx.MockTransport`.
+
+**Spec:** Issue #662 plus this plan.
+
+## Global Constraints
+
+- No git stash; explicit-path staging; `git status --short` pre-commit; ruff format+check first; commits in background (hook runs mapped tests).
+- No Claude co-author trailer.
+- All `Store`/sqlite calls on the main thread; worker threads do network + file I/O only.
+- Kill the running API backfill loop BEFORE launching the mirror run (two enrichers racing the same docket would double-attach).
+- Local attachment naming must match the existing layout: `.state/comments/{D}/{cid}/attachment_N.{ext}` (strip the `{cid}_` prefix from mirror basenames).
+- Mirror-only comments get the same tag set as farmed ones: `source:regulations-gov`, `doctype:comment`, `reg-docket:{D}`, `rule:{cms_id}` (derived from an existing item of the docket), `year:{posted[:4]}`, plus `org:` when present.
+
+---
+
+### Task 1: `Mirror` client + `backfill_from_mirror`
+
+**Files:**
+- Modify: `src/bib/regulations_gov.py`
+- Modify: `src/cli/bib.py` (add `--mirror` to `backfill-comments`)
+- Test: `tests/bib/test_regulations_gov_mirror.py` (new)
+
+**Interfaces:**
+- Consumes: `_parse_comment(row) -> Comment`, `upsert_comment(store, comment, cms_id=, extra_tags=)`, `Store.add_tag/attach_file/update`, `_slug`, `_docket_from_comment_id`.
+- Produces:
+  - `class Mirror` — `list_keys(prefix) -> list[str]` (paged, continuation-token), `comment_ids(docket) -> list[str]`, `attachment_keys(docket) -> dict[str, list[str]]`, `get_json(key) -> dict`, `download(key, dest) -> Path | None`. Base `https://mirrulations.s3.amazonaws.com`; constructor takes optional `httpx.Client` for testing.
+  - `backfill_from_mirror(store, mirror, *, docket, limit=None, log_path=None, commit_every=200, scratch_root=Path(".state/comments"), workers=12) -> dict[str, int]` with stats keys `{"created", "enriched", "attached", "errors", "seen"}`.
+  - CLI: `stack bib backfill-comments --docket D --mirror` routes to the mirror path (docket required with --mirror).
+
+- [ ] **Step 1: Write the failing tests** (`tests/bib/test_regulations_gov_mirror.py`) — a MockTransport that serves: a paged list response (two pages via `NextContinuationToken`), comment JSONs for one existing-stub cid and one mirror-only cid, one attachment binary (`%PDF...`). Assert: paging concatenates keys; `backfill_from_mirror` on a tmp `Store` (seed one stub item with `source:regulations-gov` + `reg-docket:` tags and the comment URL) → stub gets abstract + `enriched:ok`, mirror-only cid gets created with the full tag set, attachment lands at `scratch/{D}/{cid}/attachment_1.pdf` and is registered via `attach_file`, stats match.
+- [ ] **Step 2: Run to verify failure** — `uv run pytest tests/bib/test_regulations_gov_mirror.py -q` → ImportError.
+- [ ] **Step 3: Implement** `Mirror` + `backfill_from_mirror` per the interfaces; thread pool fetches `(cid) -> (json_dict, [downloaded attachment paths])` into scratch; main thread applies DB effects per completed future; `rule:` cms_id looked up once per docket from any existing item's `rule:` tag (empty string when the docket has none). Wire the CLI flag.
+- [ ] **Step 4: Tests green** — the new file plus `uv run pytest tests/bib -q` (background; large).
+- [ ] **Step 5: Live smoke** — `uv run stack bib backfill-comments --docket CMS-2017-0092 --mirror --limit 25` (smallest docket; verify enriched counts + attachment files).
+- [ ] **Step 6: Ruff, status check, commit** — `feat(bib): backfill-comments --mirror — bulk enrichment from the Mirrulations S3 mirror (refs #662)`.
+
+### Task 2: Swap the running crawl over to the mirror
+
+- [ ] **Step 1:** Kill the API loop (`pkill -f backfill-comments` — anchored; also the wrapping bash pid) after noting its progress.
+- [ ] **Step 2:** Launch the mirror loop (same docket order + extract + sweep chain, `.env` sourced), all ten dockets including CMS-2017-0092 and CMS-2026-2377 (mirror is fresher than our farm even there).
+- [ ] **Step 3:** Verify first-docket throughput; estimate completion (expect hours).
+- [ ] **Step 4:** Comment on #662 (mirror route live, API loop retired to gap-fill duty); leave open until the run + extraction + sweep complete.
+
+### Task 3: Close out
+
+- [ ] Update memory (`palliative_rfi_project.md` + the answer to "many VMs?": per-key limit, mirror route).

src/bib/regulations_gov.py

@@ -23,8 +23,10 @@ Reference endpoints::
 from __future__ import annotations
 
 import hashlib
+import json
 import logging
 import os
+import re
 import time
 from concurrent.futures import ThreadPoolExecutor, as_completed
 from dataclasses import dataclass, field
@@ -548,6 +550,232 @@ def _docket_from_comment_id(cid: str) -> str:
     return "-".join(parts[:3]) if len(parts) >= 3 else cid
 
 
+# ── Mirrulations S3 mirror ──────────────────────────────────────
+
+_MIRROR_BASE = "https://mirrulations.s3.amazonaws.com"
+
+
+class Mirror:
+    """Read-only client for the Mirrulations mirror (AWS Open Data).
+
+    ``s3://mirrulations`` publicly mirrors regulations.gov:
+    ``raw-data/{agency}/{D}/text-{D}/comments/{cid}.json`` is the v4
+    ``/comments/{id}`` detail response verbatim, and
+    ``raw-data/{agency}/{D}/binary-{D}/comments_attachments/`` holds the
+    attachment binaries. No auth, no rate cap — the bulk alternative to
+    the 1000-req/hr API crawl (#662).
+    """
+
+    def __init__(self, http: httpx.Client | None = None):
+        self._http = http or httpx.Client(
+            timeout=30, headers={"User-Agent": "stack-bib/1.0"}
+        )
+
+    def list_keys(self, prefix: str) -> list[str]:
+        """All keys under *prefix*, following continuation tokens."""
+        keys: list[str] = []
+        token: str | None = None
+        while True:
+            params = {"list-type": "2", "prefix": prefix, "max-keys": "1000"}
+            if token:
+                params["continuation-token"] = token
+            r = self._http.get(f"{_MIRROR_BASE}/", params=params)
+            r.raise_for_status()
+            keys.extend(m.group(1) for m in re.finditer(r"<Key>([^<]+)</Key>", r.text))
+            m = re.search(r"<NextContinuationToken>([^<]+)", r.text)
+            if not m:
+                return keys
+            token = m.group(1)
+
+    def _agency(self, docket: str) -> str:
+        return docket.split("-", 1)[0]
+
+    def comment_ids(self, docket: str) -> list[str]:
+        prefix = f"raw-data/{self._agency(docket)}/{docket}/text-{docket}/comments/"
+        return [
+            k.rsplit("/", 1)[-1].removesuffix(".json")
+            for k in self.list_keys(prefix)
+            if k.endswith(".json")
+        ]
+
+    def attachment_keys(self, docket: str) -> dict[str, list[str]]:
+        """comment_id → mirror keys of its attachment binaries."""
+        prefix = (
+            f"raw-data/{self._agency(docket)}/{docket}/"
+            f"binary-{docket}/comments_attachments/"
+        )
+        out: dict[str, list[str]] = {}
+        for k in self.list_keys(prefix):
+            name = k.rsplit("/", 1)[-1]
+            cid, sep, _ = name.partition("_attachment_")
+            if sep:
+                out.setdefault(cid, []).append(k)
+        return out
+
+    def get_json(self, docket: str, cid: str) -> dict:
+        r = self._http.get(
+            f"{_MIRROR_BASE}/raw-data/{self._agency(docket)}/{docket}/"
+            f"text-{docket}/comments/{cid}.json"
+        )
+        r.raise_for_status()
+        return json.loads(r.text)
+
+    def download(self, key: str, dest: Path) -> Path | None:
+        dest.parent.mkdir(parents=True, exist_ok=True)
+        try:
+            with self._http.stream("GET", f"{_MIRROR_BASE}/{key}") as r:
+                if r.status_code != 200:
+                    return None
+                with open(dest, "wb") as f:
+                    for chunk in r.iter_bytes(1 << 16):
+                        f.write(chunk)
+        except (httpx.HTTPError, OSError):
+            return None
+        return dest
+
+
+def _rule_tag_for_docket(store: Store, docket: str) -> str:
+    """The docket's ``rule:`` value from any existing item ('' if none)."""
+    row = (
+        store._con()  # noqa: SLF001
+        .execute(
+            """
+            SELECT t2.name FROM items i
+            JOIN item_tags it ON it.item_id = i.id
+            JOIN tags t ON t.id = it.tag_id AND t.name = ?
+            JOIN item_tags it2 ON it2.item_id = i.id
+            JOIN tags t2 ON t2.id = it2.tag_id AND t2.name LIKE 'rule:%'
+            LIMIT 1
+            """,
+            (f"reg-docket:{docket}",),
+        )
+        .fetchone()
+    )
+    return row[0].split(":", 1)[1] if row else ""
+
+
+def backfill_from_mirror(
+    store: Store,
+    mirror: Mirror,
+    *,
+    docket: str,
+    limit: int | None = None,
+    log_path: Path | None = None,
+    commit_every: int = 200,
+    scratch_root: Path = Path(".state/comments"),
+    workers: int = 12,
+) -> dict[str, int]:
+    """Enrich (and create) every comment of *docket* from the mirror.
+
+    The mirror is a superset of the API farm, so comments we never
+    listed get created via :func:`upsert_comment` with the same tag set.
+    S3 fetches fan out over a thread pool; every Store/sqlite call stays
+    on this thread. Idempotent: ``enriched:ok`` items are skipped.
+    """
+    con = store._con()  # noqa: SLF001
+    existing: dict[str, tuple[str, bool]] = {}  # cid -> (key, enriched)
+    for key, url, enriched in con.execute(
+        """
+        SELECT i.key, i.url,
+               EXISTS (
+                 SELECT 1 FROM item_tags it JOIN tags t ON t.id = it.tag_id
+                 WHERE it.item_id = i.id AND t.name = 'enriched:ok'
+               )
+        FROM items i
+        WHERE i.url LIKE ?
+        """,
+        (f"https://www.regulations.gov/comment/{docket}-%",),
+    ):
+        existing[url.rsplit("/", 1)[-1]] = (key, bool(enriched))
+
+    cids = mirror.comment_ids(docket)
+    atts = mirror.attachment_keys(docket)
+    todo = [c for c in cids if not existing.get(c, ("", False))[1]]
+    if limit:
+        todo = todo[: int(limit)]
+    cms_id = _rule_tag_for_docket(store, docket)
+
+    stats = {"created": 0, "enriched": 0, "attached": 0, "errors": 0, "seen": len(todo)}
+
+    def _log(msg: str) -> None:
+        log.info(msg)
+        print(msg, flush=True)
+        if log_path:
+            log_path.parent.mkdir(parents=True, exist_ok=True)
+            with open(log_path, "a") as f:
+                f.write(msg + "\n")
+
+    _log(f"mirror backfill start: {docket} — {len(todo)} of {len(cids)} pending")
+
+    def _fetch(cid: str) -> tuple[str, dict, list[Path]]:
+        data = mirror.get_json(docket, cid)
+        dest_dir = scratch_root / docket / cid
+        paths: list[Path] = []
+        for k in atts.get(cid, ()):
+            name = k.rsplit("/", 1)[-1].removeprefix(f"{cid}_")
+            dest = dest_dir / name
+            if dest.is_file() or mirror.download(k, dest):
+                paths.append(dest)
+        return cid, data, paths
+
+    done = 0
+    with ThreadPoolExecutor(max_workers=max(1, workers)) as pool:
+        futures = {pool.submit(_fetch, cid): cid for cid in todo}
+        for fut in as_completed(futures):
+            done += 1
+            try:
+                cid, data, paths = fut.result()
+            except Exception as e:  # noqa: BLE001
+                log.warning("mirror fetch failed for %s: %s", futures[fut], e)
+                stats["errors"] += 1
+                continue
+
+            row = data.get("data") or {}
+            attrs = row.get("attributes") or {}
+            if cid in existing:
+                key = existing[cid][0]
+            else:
+                key = upsert_comment(
+                    store,
+                    _parse_comment(row),
+                    cms_id=cms_id,
+                    extra_tags=[f"reg-docket:{docket}"],
+                )
+                stats["created"] += 1
+
+            body = (attrs.get("comment") or "")[:4000]
+            if body:
+                con.execute("UPDATE items SET abstract=? WHERE key=?", (body, key))
+            org = (attrs.get("organization") or "").strip()
+            if org:
+                try:
+                    store.add_tag(key, f"org:{_slug(org)}")
+                except Exception as e:  # noqa: BLE001
+                    log.warning("add_tag org failed for %s: %s", key, e)
+            for path in paths:
+                try:
+                    store.attach_file(key, path, title=path.name)
+                    stats["attached"] += 1
+                except Exception as e:  # noqa: BLE001
+                    log.debug("attach_file skipped for %s/%s: %s", key, path.name, e)
+            try:
+                store.add_tag(key, "enriched:ok")
+            except Exception as e:  # noqa: BLE001
+                log.warning("add_tag enriched:ok failed for %s: %s", key, e)
+            stats["enriched"] += 1
+            if done % commit_every == 0:
+                con.commit()
+                _log(
+                    f"  {done}/{len(todo)}  enriched={stats['enriched']} "
+                    f"created={stats['created']} attached={stats['attached']} "
+                    f"errors={stats['errors']}"
+                )
+
+    con.commit()
+    _log(f"mirror backfill done: {docket} {stats}")
+    return stats
+
+
 def upsert_comment(
     store: Store,
     comment: Comment,

src/cli/bib.py

@@ -363,6 +363,13 @@ def backfill_comments(
         "--log",
         help="Append-only progress log (survives restarts).",
     ),
+    mirror: bool = typer.Option(
+        False,
+        "--mirror",
+        help="Pull from the Mirrulations S3 mirror instead of the "
+        "reg.gov API — bulk speed, no rate cap, also ingests comments "
+        "the original farm never listed. Requires --docket.",
+    ),
 ) -> None:
     """Enrich every reg-gov comment stub with body + attachments.
 
@@ -372,19 +379,36 @@ def backfill_comments(
     it left off by skipping already-enriched items. At 1.3s/req the API
     calls run at ~2700/hr; attachment downloads are concurrent on the
     CDN and don't count against the rate limit.
+
+    ``--mirror`` swaps the API for the Mirrulations public S3 mirror
+    (#662): no API key, no cap, and mirror-only comments get created.
     """
     from bib import connect
-    from bib.regulations_gov import Client, backfill_details
 
     store = connect()
-    with Client(sleep=sleep) as api:
-        stats = backfill_details(
+    if mirror:
+        if not docket:
+            raise typer.BadParameter("--mirror requires --docket")
+        from bib.regulations_gov import Mirror, backfill_from_mirror
+
+        stats = backfill_from_mirror(
             store,
-            api,
+            Mirror(),
             docket=docket,
             limit=limit or None,
             log_path=log_path,
         )
+    else:
+        from bib.regulations_gov import Client, backfill_details
+
+        with Client(sleep=sleep) as api:
+            stats = backfill_details(
+                store,
+                api,
+                docket=docket,
+                limit=limit or None,
+                log_path=log_path,
+            )
     for k, v in stats.items():
         typer.echo(f"  {k}: {v}")
 

tests/bib/test_regulations_gov_mirror.py

@@ -0,0 +1,179 @@
+"""Tests for the Mirrulations S3 mirror backfill (bib/regulations_gov.py)."""
+
+from __future__ import annotations
+
+import json
+
+import httpx
+
+from bib.regulations_gov import (
+    Comment,
+    Mirror,
+    backfill_from_mirror,
+    upsert_comment,
+)
+from bib.store import Store
+
+_DOCKET = "CMS-2018-0076"
+_TEXT = f"raw-data/CMS/{_DOCKET}/text-{_DOCKET}/comments/"
+_BIN = f"raw-data/CMS/{_DOCKET}/binary-{_DOCKET}/comments_attachments/"
+
+
+def _detail(cid: str, comment: str, org: str = "") -> dict:
+    return {
+        "data": {
+            "id": cid,
+            "type": "comments",
+            "attributes": {
+                "title": f"Comment {cid}",
+                "docketId": _DOCKET,
+                "commentOnId": "0900006482921ba1",
+                "postedDate": "2018-09-10T00:00:00Z",
+                "receivedDate": "2018-09-09T00:00:00Z",
+                "organization": org,
+                "comment": comment,
+                "attachmentCount": 1,
+            },
+        }
+    }
+
+
+_PDF = b"%PDF-1.4 " + b"x" * 2000
+
+
+def _handler(request: httpx.Request) -> httpx.Response:
+    path = request.url.path.lstrip("/")
+    params = dict(request.url.params)
+    if path == "" and "list-type" in params:
+        prefix = params.get("prefix", "")
+        if prefix == _TEXT:
+            if "continuation-token" not in params:
+                return httpx.Response(
+                    200,
+                    text=(
+                        "<ListBucketResult>"
+                        f"<Contents><Key>{_TEXT}{_DOCKET}-0001.json</Key></Contents>"
+                        "<NextContinuationToken>tok1</NextContinuationToken>"
+                        "</ListBucketResult>"
+                    ),
+                )
+            return httpx.Response(
+                200,
+                text=(
+                    "<ListBucketResult>"
+                    f"<Contents><Key>{_TEXT}{_DOCKET}-0002.json</Key></Contents>"
+                    "</ListBucketResult>"
+                ),
+            )
+        if prefix == _BIN:
+            return httpx.Response(
+                200,
+                text=(
+                    "<ListBucketResult>"
+                    f"<Contents><Key>{_BIN}{_DOCKET}-0002_attachment_1.pdf</Key>"
+                    "</Contents></ListBucketResult>"
+                ),
+            )
+        return httpx.Response(200, text="<ListBucketResult></ListBucketResult>")
+    if path == f"{_TEXT}{_DOCKET}-0001.json":
+        return httpx.Response(
+            200, text=json.dumps(_detail(f"{_DOCKET}-0001", "inline body one"))
+        )
+    if path == f"{_TEXT}{_DOCKET}-0002.json":
+        return httpx.Response(
+            200,
+            text=json.dumps(_detail(f"{_DOCKET}-0002", "see attached", org="Test Org")),
+        )
+    if path == f"{_BIN}{_DOCKET}-0002_attachment_1.pdf":
+        return httpx.Response(
+            200, content=_PDF, headers={"content-type": "application/pdf"}
+        )
+    return httpx.Response(404)
+
+
+def _mirror() -> Mirror:
+    return Mirror(http=httpx.Client(transport=httpx.MockTransport(_handler)))
+
+
+class TestMirrorListing:
+    def test_paged_listing_concatenates(self):
+        ids = _mirror().comment_ids(_DOCKET)
+        assert ids == [f"{_DOCKET}-0001", f"{_DOCKET}-0002"]
+
+    def test_attachment_keys_by_cid(self):
+        m = _mirror().attachment_keys(_DOCKET)
+        assert list(m) == [f"{_DOCKET}-0002"]
+        assert m[f"{_DOCKET}-0002"] == [f"{_BIN}{_DOCKET}-0002_attachment_1.pdf"]
+
+
+class TestBackfillFromMirror:
+    def _store(self, tmp_path) -> Store:
+        store = Store(str(tmp_path / "bib.sqlite"))
+        # Seed one un-enriched stub the way the original farm did.
+        c = Comment(
+            id=f"{_DOCKET}-0001",
+            title="",
+            posted_date="2018-09-10",
+            received_date="2018-09-09",
+            docket_id=_DOCKET,
+            comment_on_id="x",
+        )
+        upsert_comment(
+            store, c, cms_id="CMS-1693-P", extra_tags=[f"reg-docket:{_DOCKET}"]
+        )
+        return store
+
+    def test_enriches_creates_and_attaches(self, tmp_path):
+        store = self._store(tmp_path)
+        stats = backfill_from_mirror(
+            store,
+            _mirror(),
+            docket=_DOCKET,
+            scratch_root=tmp_path / "scratch",
+            workers=2,
+        )
+        assert stats["enriched"] == 2
+        assert stats["created"] == 1
+        assert stats["attached"] == 1
+        assert stats["errors"] == 0
+
+        con = store._con()
+        # stub got its body + enriched:ok
+        row = con.execute(
+            "SELECT abstract FROM items WHERE url LIKE ?",
+            (f"%{_DOCKET}-0001",),
+        ).fetchone()
+        assert row[0] == "inline body one"
+        # mirror-only comment created with the farm tag set
+        row2 = con.execute(
+            "SELECT id FROM items WHERE url LIKE ?", (f"%{_DOCKET}-0002",)
+        ).fetchone()
+        assert row2 is not None
+        tags = {
+            r[0]
+            for r in con.execute(
+                "SELECT t.name FROM tags t JOIN item_tags it ON it.tag_id=t.id "
+                "WHERE it.item_id=?",
+                (row2[0],),
+            )
+        }
+        assert "source:regulations-gov" in tags
+        assert f"reg-docket:{_DOCKET}" in tags
+        assert "rule:CMS-1693-P" in tags
+        assert "enriched:ok" in tags
+        assert "org:test-org" in tags
+        # attachment landed with the layout name
+        att = tmp_path / "scratch" / _DOCKET / f"{_DOCKET}-0002" / "attachment_1.pdf"
+        assert att.is_file() and att.read_bytes().startswith(b"%PDF")
+
+    def test_idempotent_rerun(self, tmp_path):
+        store = self._store(tmp_path)
+        m = _mirror()
+        backfill_from_mirror(
+            store, m, docket=_DOCKET, scratch_root=tmp_path / "scratch", workers=2
+        )
+        stats2 = backfill_from_mirror(
+            store, m, docket=_DOCKET, scratch_root=tmp_path / "scratch", workers=2
+        )
+        assert stats2["enriched"] == 0
+        assert stats2["created"] == 0