feat(precedents): UI button queues extraction for local MCP worker

The chair wanted a one-click "extract metadata" button on the edit sheet.
The constraint stays the same — claude_session needs the local CLI which
the container doesn't have, so the button can't run the extractor itself.
Compromise: button stamps a queue marker; the local MCP server drains the
queue on demand.

DB (V8): two nullable timestamps on case_law,
metadata_extraction_requested_at and halacha_extraction_requested_at,
with partial indexes for cheap "find pending" scans.

API:
  POST /api/precedent-library/{id}/request-metadata   → stamp the row
  POST /api/precedent-library/{id}/request-halachot   → same for halacha
  GET  /api/precedent-library/queue/pending?kind=...  → read-only view

UI: Sparkles button in the edit sheet header. Click → toast tells the
chair what to run from Claude Code. The button never triggers the
extractor directly from the container.

MCP tool: precedent_process_pending(kind, limit) — runs from Claude Code
with the local CLI, picks up everything stamped, calls the extractor for
each, clears the timestamp on success. Failures keep the timestamp so the
next invocation retries them.

Architectural rule (claude_session local-only) is preserved end-to-end
and called out in the new endpoint comment + tool docstring.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

mcp-server/src/legal_mcp/services/db.py

@@ -603,6 +603,26 @@ CREATE INDEX IF NOT EXISTS idx_halachot_vec
 """
 
 
+# ── V8: Extraction request queue ─────────────────────────────────
+# Web UI buttons ("Sparkles" = request metadata extraction; "Refresh" =
+# request halacha extraction) run inside the FastAPI container, which has
+# no `claude` CLI. They can't run the LLM extractor directly. Instead they
+# stamp a request timestamp here, and the chair (or me) runs the MCP tool
+# `precedent_process_pending_extractions` from local Claude Code, where the
+# CLI is available, to drain the queue. See claude_session.py for the rule.
+
+SCHEMA_V8_SQL = """
+ALTER TABLE case_law ADD COLUMN IF NOT EXISTS metadata_extraction_requested_at TIMESTAMPTZ;
+ALTER TABLE case_law ADD COLUMN IF NOT EXISTS halacha_extraction_requested_at TIMESTAMPTZ;
+CREATE INDEX IF NOT EXISTS idx_case_law_metadata_requested
+    ON case_law(metadata_extraction_requested_at)
+    WHERE metadata_extraction_requested_at IS NOT NULL;
+CREATE INDEX IF NOT EXISTS idx_case_law_halacha_requested
+    ON case_law(halacha_extraction_requested_at)
+    WHERE halacha_extraction_requested_at IS NOT NULL;
+"""
+
+
 async def init_schema() -> None:
     pool = await get_pool()
     async with pool.acquire() as conn:
@@ -614,7 +634,8 @@ async def init_schema() -> None:
         await conn.execute(SCHEMA_V5_SQL)
         await conn.execute(SCHEMA_V6_SQL)
         await conn.execute(SCHEMA_V7_SQL)
-    logger.info("Database schema initialized (v1-v7)")
+        await conn.execute(SCHEMA_V8_SQL)
+    logger.info("Database schema initialized (v1-v8)")
 
 
 # ── Case CRUD ───────────────────────────────────────────────────────
@@ -2191,3 +2212,79 @@ async def precedent_library_stats() -> dict:
         "halachot_pending": int(halachot_pending or 0),
         "halachot_approved": int(halachot_approved or 0),
     }
+
+
+# ── V8: extraction request queue helpers ─────────────────────────
+
+
+async def request_metadata_extraction(case_law_id: UUID) -> bool:
+    """Stamp ``metadata_extraction_requested_at`` for the local MCP worker
+    to pick up. Returns False if the row is missing."""
+    pool = await get_pool()
+    result = await pool.execute(
+        "UPDATE case_law SET metadata_extraction_requested_at = now() "
+        "WHERE id = $1 AND source_kind = 'external_upload'",
+        case_law_id,
+    )
+    return result == "UPDATE 1"
+
+
+async def request_halacha_extraction(case_law_id: UUID) -> bool:
+    """Same but for halacha extraction."""
+    pool = await get_pool()
+    result = await pool.execute(
+        "UPDATE case_law SET halacha_extraction_requested_at = now() "
+        "WHERE id = $1 AND source_kind = 'external_upload'",
+        case_law_id,
+    )
+    return result == "UPDATE 1"
+
+
+async def list_pending_extraction_requests(
+    kind: str = "metadata",  # 'metadata' | 'halacha'
+    limit: int = 20,
+) -> list[dict]:
+    """Return rows requesting extraction, oldest request first.
+
+    The MCP worker drains the queue in order: process → clear timestamp.
+    """
+    col = (
+        "metadata_extraction_requested_at"
+        if kind == "metadata"
+        else "halacha_extraction_requested_at"
+    )
+    pool = await get_pool()
+    rows = await pool.fetch(
+        f"""SELECT id, case_number, case_name, court, date,
+                  practice_area, is_binding, {col} AS requested_at
+            FROM case_law
+            WHERE {col} IS NOT NULL
+              AND source_kind = 'external_upload'
+            ORDER BY {col} ASC
+            LIMIT $1""",
+        limit,
+    )
+    out = []
+    for r in rows:
+        d = dict(r)
+        if d.get("date") is not None:
+            d["date"] = d["date"].isoformat()
+        if d.get("requested_at") is not None:
+            d["requested_at"] = d["requested_at"].isoformat()
+        out.append(d)
+    return out
+
+
+async def clear_extraction_request(
+    case_law_id: UUID, kind: str = "metadata",
+) -> None:
+    col = (
+        "metadata_extraction_requested_at"
+        if kind == "metadata"
+        else "halacha_extraction_requested_at"
+    )
+    pool = await get_pool()
+    await pool.execute(
+        f"UPDATE case_law SET {col} = NULL WHERE id = $1",
+        case_law_id,
+    )