feat(llm): stack llm restamp — metadata-only anchor backfill on pgvector (refs #688)

Backfills codes/families/elements onto the 1.15M chunks already in
pgvector without re-embedding: anchor_metadata is a pure function of
chunk text, so it's recomputed and merged into cmetadata with a jsonb
`||`, one transaction per batch, keyset-paginated on id. Idempotent —
a row whose three keys already match the freshly computed values is
skipped; a row missing a key (the pre-backfill norm) is patched even
when the computed value is "", since the acceptance check is key
presence (`cmetadata ? 'families'`), not value equality.

`stack llm restamp --collection all|comments|rules|corpus [--batch]
[--dry-run]` mirrors `index`'s config/engine setup, runs migrate() and
ensure_hnsw() first so the GIN indexes over families/elements exist,
refreshes the derived family registry from the DuckDB replica when
present, and builds the code->family index once per run via
pfs.anchors.code_family_index.

Live backfill (rig, pgvector at LLM_PG_HOST):
  rules:  scanned=67,340  updated=67,340  (168.4s, then 12,389 on the
          A5 re-run after excluding single-code families)
  corpus: scanned=235,140 updated=235,140 (626.1s, ~376 rows/s)
  comments (~918k rows) launched in the background per Ruling A3.

src/cli/llm.py

@@ -98,6 +98,62 @@ def index(
         )
 
 
+@app.command()
+def restamp(
+    collection: str = typer.Option(
+        "comments", help="Which collection: comments | rules | corpus | all."
+    ),
+    batch: int = typer.Option(2000, help="Rows per transaction."),
+    dry_run: bool = typer.Option(
+        False, "--dry-run", help="Scan and report without writing."
+    ),
+) -> None:
+    """Backfill codes/families/elements onto already-indexed chunks.
+
+    Metadata-only: computes `pfs.anchors.anchor_metadata` from each row's
+    stored document text and merges it into `cmetadata` with a jsonb `||`,
+    never touching the embedding — no GPU time, no re-chunking. Safe to
+    re-run: rows already stamped with the current values are skipped.
+    """
+    import os
+
+    from llm import config as llm_config
+    from llm.index import _engine
+    from llm.migrate import ensure_hnsw, migrate
+    from llm.restamp import restamp as restamp_collection
+    from pfs.anchors import code_family_index
+    from pfs.families import FAMILIES, refresh_from
+
+    targets = _COLLECTIONS if collection == "all" else (collection,)
+    if any(t not in _COLLECTIONS for t in targets):
+        raise typer.BadParameter("collection must be comments, rules, corpus or all")
+    cfg = llm_config.load()
+    engine = _engine(cfg)
+    migrate(engine)
+    ensure_hnsw(engine, cfg.embed_dim)
+    if os.path.exists(cfg.duckdb_replica):
+        import duckdb
+
+        con = duckdb.connect(cfg.duckdb_replica, read_only=True)
+        try:
+            refresh_from(con)
+        finally:
+            con.close()
+    code_index = code_family_index(FAMILIES)
+    for target in targets:
+        stats = restamp_collection(
+            engine,
+            collection=target,
+            batch=batch,
+            dry_run=dry_run,
+            code_index=code_index,
+        )
+        typer.echo(
+            f"{target}: scanned={stats['scanned']} updated={stats['updated']} "
+            f"seconds={stats['seconds']}"
+        )
+
+
 @app.command()
 def hosts() -> None:
     """Show the Ollama fleet: declared VRAM, liveness, models, and which

src/llm/restamp.py

@@ -0,0 +1,98 @@
+"""Backfill anchor metadata on already-indexed chunks without re-embedding.
+
+Re-indexing 1.15 M chunks costs days of GPU time; the anchors are a pure
+function of the chunk text, so they are recomputed in Python and merged
+into ``cmetadata`` with a jsonb concatenation. Keyset pagination on the
+primary key keeps memory flat; one transaction per batch keeps the lock
+short. Idempotent: rows whose three values already match are skipped.
+"""
+
+from __future__ import annotations
+
+import json
+import logging
+import time
+from typing import Any, Iterable, Mapping
+
+from sqlalchemy import text
+
+from pfs.anchors import anchor_metadata
+from pfs.families import Family
+
+log = logging.getLogger(__name__)
+
+KEYS = ("codes", "families", "elements")
+
+SELECT_SQL = (
+    "SELECT e.id, e.document, e.cmetadata FROM langchain_pg_embedding e "
+    "WHERE e.collection_id = (SELECT uuid FROM langchain_pg_collection WHERE name = :collection) "
+    "AND e.id > :after ORDER BY e.id LIMIT :batch"
+)
+UPDATE_SQL = "UPDATE langchain_pg_embedding SET cmetadata = cmetadata || CAST(:patch AS jsonb) WHERE id = :id"
+
+
+def plan_patches(
+    rows: Iterable[tuple[str, str, Mapping[str, Any] | None]],
+    *,
+    families: Mapping[str, Family] | None = None,
+    code_index: Mapping[str, str] | None = None,
+) -> list[tuple[str, str]]:
+    out: list[tuple[str, str]] = []
+    for id_, document, md in rows:
+        md = md or {}
+        want = anchor_metadata(document or "", families=families, code_index=code_index)
+        # A key that is entirely absent (the common case for a
+        # pre-backfill row) must still be patched even when its computed
+        # value happens to be "" — the acceptance check is key presence
+        # (`cmetadata ? 'families'`), not value equality, so falling back
+        # to a "" default for a missing key would make it look unchanged
+        # and the key would never get written.
+        if all(k in md and str(md[k]) == want[k] for k in KEYS):
+            continue
+        out.append((id_, json.dumps(want)))
+    return out
+
+
+def restamp(
+    engine: Any,
+    *,
+    collection: str,
+    batch: int = 2000,
+    dry_run: bool = False,
+    families: Mapping[str, Family] | None = None,
+    code_index: Mapping[str, str] | None = None,
+    log_every: int = 50_000,
+) -> dict:
+    scanned = updated = 0
+    after = ""
+    t0 = time.perf_counter()
+    while True:
+        with engine.begin() as conn:
+            rows = conn.execute(
+                text(SELECT_SQL),
+                {"collection": collection, "after": after, "batch": batch},
+            ).fetchall()
+            if not rows:
+                break
+            patches = plan_patches(rows, families=families, code_index=code_index)
+            if patches and not dry_run:
+                conn.execute(
+                    text(UPDATE_SQL), [{"id": i, "patch": p} for i, p in patches]
+                )
+        scanned += len(rows)
+        updated += len(patches)
+        after = rows[-1][0]
+        if scanned % log_every < batch:
+            log.info(
+                "restamp %s: scanned %d updated %d (%.0fs)",
+                collection,
+                scanned,
+                updated,
+                time.perf_counter() - t0,
+            )
+    return {
+        "collection": collection,
+        "scanned": scanned,
+        "updated": updated,
+        "seconds": round(time.perf_counter() - t0, 1),
+    }

tests/llm/test_restamp.py

@@ -0,0 +1,53 @@
+"""llm.restamp — metadata-only anchor backfill on already-indexed chunks."""
+
+import json
+import os
+
+import pytest
+
+from llm.restamp import SELECT_SQL, UPDATE_SQL, plan_patches, restamp
+
+
+def test_plan_skips_unchanged_and_patches_changed():
+    rows = [
+        (
+            "a",
+            "Use 99439 in conjunction with 99490; consent",
+            {
+                "codes": "99439 99490",
+                "families": "CCM",
+                "elements": "activity=consent relation=addon-of",
+            },
+        ),
+        (
+            "b",
+            "Use 99439 in conjunction with 99490; consent",
+            {"codes": "99439 99490"},
+        ),
+        ("c", "nothing", {}),
+    ]
+    patches = plan_patches(rows)
+    ids = [p[0] for p in patches]
+    assert ids == ["b", "c"]
+    assert json.loads(patches[0][1])["families"] == "CCM"
+    assert json.loads(patches[1][1]) == {"codes": "", "families": "", "elements": ""}
+
+
+def test_sql_shapes():
+    assert (
+        "ORDER BY e.id" in SELECT_SQL
+        and ":after" in SELECT_SQL
+        and ":batch" in SELECT_SQL
+    )
+    assert "cmetadata || CAST(:patch AS jsonb)" in UPDATE_SQL
+
+
+@pytest.mark.skipif(not os.environ.get("LLM_DB_PASSWORD"), reason="needs pgvector")
+def test_restamp_dry_run_scans_live_rules():
+    from llm import config as llm_config
+    from llm.index import _engine
+
+    cfg = llm_config.load()
+    engine = _engine(cfg)
+    stats = restamp(engine, collection="rules", batch=50, dry_run=True)
+    assert stats["scanned"] > 0