feat(corroboration): wire approval gate + backfill driver + rebuild tool (X11 Phase 2)

- db: approve_halacha_by_corroboration (pending_review→approved only),
  demote_halacha_overruled (approved→pending_review only), list_corroboration_grouped,
  precedents_with_halachot_and_incoming_citations
- corroboration: reconcile_approvals (INV-COR2/COR4/COR5), build_all backfill;
  build_for_precedent now returns approved/demoted counts
- mcp: corroboration_rebuild write tool (single precedent or full-corpus backfill)

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

mcp-server/src/legal_mcp/server.py

@@ -934,6 +934,17 @@ async def halacha_corroboration(halacha_id: str) -> dict:
     return {"halacha_id": halacha_id, "summary": agg, "citations": links}
 
 
+@mcp.tool()
+async def corroboration_rebuild(case_law_id: str = "") -> dict:
+    """בנה/רענן את ה-corroboration ויישם אישור-אוטומטי. ריק = כל הקורפוס (backfill);
+    מזהה-תקדים = תקדים בודד. כותב halacha_citation_corroboration ומעדכן review_status
+    (corroborated→approved, overruled→pending_review). X11 Phase 2."""
+    from legal_mcp.services import corroboration as cor
+    if case_law_id.strip():
+        return await cor.build_for_precedent(case_law_id.strip())
+    return await cor.build_all()
+
+
 def main():
     mcp.run(transport="stdio")
 

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

@@ -117,4 +117,49 @@ async def build_for_precedent(case_law_id: str | UUID) -> dict:
             treatment, best[1], ctx,
         )
         linked += 1
-    return {"citations": len(cits), "linked": linked}
+    appr = await reconcile_approvals(case_law_id)
+    return {"citations": len(cits), "linked": linked,
+            "approved": appr["approved"], "demoted": appr["demoted"]}
+
+
+async def reconcile_approvals(case_law_id: str | UUID) -> dict:
+    """Apply the corroboration→approval policy to every halacha of a precedent
+    (INV-COR2/COR4/COR5). No-op when the kill-switch is off. Idempotent: approve
+    only fires on ``pending_review``, demote only on ``approved``, so re-runs
+    converge."""
+    if not config.HALACHA_CORROBORATION_AUTO_APPROVE:
+        return {"approved": 0, "demoted": 0, "disabled": True}
+    if isinstance(case_law_id, str):
+        case_law_id = UUID(case_law_id)
+    grouped = await db.list_corroboration_grouped(case_law_id)
+    approved = demoted = 0
+    for halacha_id, links in grouped.items():
+        agg = aggregate(links)
+        has_overruled = any(l["treatment"] == "overruled" for l in links)
+        action = approval_action(agg, has_overruled)
+        if action == "approve":
+            if await db.approve_halacha_by_corroboration(
+                UUID(halacha_id), agg["positive_sources"],
+                config.HALACHA_CORROBORATION_MIN_CITES,
+            ):
+                approved += 1
+        elif action == "demote":
+            if await db.demote_halacha_overruled(UUID(halacha_id)):
+                demoted += 1
+    return {"approved": approved, "demoted": demoted, "disabled": False}
+
+
+async def build_all() -> dict:
+    """Backfill: build the signal + apply approvals for every precedent that has
+    halachot and incoming citations. Idempotent (link table ``ON CONFLICT`` +
+    state-gated transitions)."""
+    ids = await db.precedents_with_halachot_and_incoming_citations()
+    totals = {"precedents": 0, "citations": 0, "linked": 0,
+              "approved": 0, "demoted": 0}
+    for cid in ids:
+        r = await build_for_precedent(cid)
+        totals["precedents"] += 1
+        for k in ("citations", "linked", "approved", "demoted"):
+            totals[k] += r.get(k, 0)
+        logger.info("corroboration backfill %s: %s", cid, r)
+    return totals

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

@@ -3480,6 +3480,81 @@ async def update_halacha(
     return dict(row) if row else None
 
 
+async def approve_halacha_by_corroboration(
+    halacha_id: UUID, n_sources: int, min_cites: int,
+) -> bool:
+    """Approve a halacha on citation corroboration — ONLY if it is currently
+    awaiting the chair (``pending_review``). Never touches ``published`` /
+    ``rejected`` / already-``approved`` (INV-COR5: the chair gate is preserved for
+    everything else). The reviewer records the corroboration basis as provenance
+    (INV-COR6). Returns True iff a row actually transitioned."""
+    pool = await get_pool()
+    reviewer = f"corroborated ({n_sources} judicial citations ≥ {min_cites})"
+    row = await pool.fetchrow(
+        "UPDATE halachot SET review_status='approved', reviewer=$2, "
+        "reviewed_at=now(), updated_at=now() "
+        "WHERE id=$1 AND review_status='pending_review' RETURNING id",
+        halacha_id, reviewer,
+    )
+    return row is not None
+
+
+async def demote_halacha_overruled(halacha_id: UUID) -> bool:
+    """Demote an APPROVED halacha back to the chair gate because a later citing
+    court overruled it (INV-COR2). Acts only on ``approved`` → ``pending_review``;
+    leaves ``published`` / ``rejected`` / already-``pending_review`` untouched. The
+    reviewer note records why it re-entered the queue. Returns True iff a row
+    transitioned."""
+    pool = await get_pool()
+    row = await pool.fetchrow(
+        "UPDATE halachot SET review_status='pending_review', "
+        "reviewer='flagged: overruled by later citation (X11)', "
+        "reviewed_at=NULL, updated_at=now() "
+        "WHERE id=$1 AND review_status='approved' RETURNING id",
+        halacha_id,
+    )
+    return row is not None
+
+
+async def list_corroboration_grouped(case_law_id: UUID) -> dict[str, list[dict]]:
+    """Per-halacha corroboration links for a cited precedent, in the
+    ``{source_id, treatment}`` shape ``aggregate()`` consumes. The distinct citing
+    source is keyed by case_law/decision id (falling back to the citation row id
+    so two anonymous rows are not collapsed)."""
+    pool = await get_pool()
+    rows = await pool.fetch(
+        "SELECT hcc.halacha_id::text AS halacha_id, "
+        "  COALESCE(hcc.citing_case_law_id::text, hcc.citing_decision_id::text, "
+        "           hcc.source_citation_id::text) AS source_id, "
+        "  hcc.treatment "
+        "FROM halacha_citation_corroboration hcc "
+        "JOIN halachot h ON h.id = hcc.halacha_id "
+        "WHERE h.case_law_id = $1",
+        case_law_id,
+    )
+    out: dict[str, list[dict]] = {}
+    for r in rows:
+        out.setdefault(r["halacha_id"], []).append(
+            {"source_id": r["source_id"], "treatment": r["treatment"]}
+        )
+    return out
+
+
+async def precedents_with_halachot_and_incoming_citations() -> list[str]:
+    """case_law ids that have at least one halacha AND at least one incoming
+    citation (either graph) — the corroboration backfill target set."""
+    pool = await get_pool()
+    rows = await pool.fetch(
+        "SELECT c.id::text FROM case_law c "
+        "WHERE EXISTS (SELECT 1 FROM halachot h WHERE h.case_law_id=c.id) "
+        "  AND (EXISTS (SELECT 1 FROM precedent_internal_citations p "
+        "               WHERE p.cited_case_law_id=c.id) "
+        "    OR EXISTS (SELECT 1 FROM case_law_citations cc "
+        "               WHERE cc.case_law_id=c.id))",
+    )
+    return [r["id"] for r in rows]
+
+
 async def nearest_halacha_for_vector(case_law_id: UUID, vec: list[float]) -> tuple[str, float] | None:
     """Best-matching halacha of `case_law_id` for a context embedding (cosine)."""
     pool = await get_pool()