feat(x11): treatment-aware citation authority wired into research agents (#154)

The internal citation graph fed only RANKING (raw in-degree), and the per-citation
TREATMENT was never classified — so a precedent distinguished N times got the same
authority boost as one followed N times (INV-COR2 violation), and the signal never
reached the agents' reasoning. Wires the full path:

Phase 1 — scripts/classify_citation_treatments.py: classify each linked edge's
  treatment (followed/distinguished/…) from its match_context via
  corroboration.classify_treatment (Opus 4.8 @ xhigh, local), filling
  precedent_internal_citations.treatment. Idempotent.
Phase 2 — db.refresh_verified_layer: count only NON-negative treatments toward
  verified/cite_count (INV-COR2/COR4). Unclassified counts as neutral-positive so
  the signal degrades gracefully before classification runs.
Phase 3 — db.citation_authority(ids): per-precedent {total, positive, negative,
  unclassified, by_treatment}. Surfaced as `cited_by` in search_precedent_library
  hits and precedent_library_get, and `treatment` per incoming citation.
Phase 4 — legal-researcher/analyst/writer prompts: weigh & ARGUE authority
  ("הלכה שאומצה ב-N החלטות ועדת-ערר"), flag distinguished/overruled, never invent
  the count (INV-AH; writer is read-only of the analyst).

Auto-approval stays kill-switched off (chair gate preserved, INV-G10). No schema
change (treatment column already existed). Operational: run the classifier +
refresh_verified_layer over the 379 edges, then sync agents across companies.

Invariants: G2 (one classifier + one authority query, reused), INV-COR2/COR3/COR4
(negative never corroborates; point-specific; ≥N), INV-G10 (no auto-approval),
INV-AH (no invented numbers).

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

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

@@ -399,6 +399,7 @@ async def list_citations_to_case_law(case_law_id: UUID) -> list[dict]:
                pic.cited_case_number,
                pic.match_context,
                pic.match_pattern,
+               pic.treatment,
                pic.confidence::float AS confidence,
                pic.created_at,
                cl.case_number AS source_case_number,

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

@@ -6536,8 +6536,13 @@ async def refresh_verified_layer() -> dict:
     """Recompute the verified/cite_count layer from chair citations (#153).
 
     'verified' = the principle's SOURCE precedent was cited by a chair (any
-    committee decision). 'cite_count' = # distinct chair decisions citing it. This
-    is the ONLY trust signal — never human review. Idempotent (full recompute).
+    committee decision) WITHOUT a negative treatment. 'cite_count' = # distinct
+    chair decisions citing it whose treatment is NOT negative (X11 §4 / INV-COR2:
+    a precedent *distinguished*/*criticized*/*questioned*/*overruled* must never
+    gain authority from those citations). Unclassified edges (treatment='') count
+    as neutral-positive until ``classify_citation_treatments.py`` labels them, so
+    the signal degrades gracefully before classification has run. This is the ONLY
+    trust signal — never human review. Idempotent (full recompute).
     Returns {verified_principles, verified_precedents}.
     """
     pool = await get_pool()
@@ -6554,6 +6559,8 @@ async def refresh_verified_layer() -> dict:
                 "  JOIN case_law src ON src.id = pic.source_case_law_id "
                 "  WHERE src.source_kind='internal_committee' "
                 "    AND pic.cited_case_law_id IS NOT NULL "
+                "    AND coalesce(pic.treatment,'') NOT IN "
+                "        ('distinguished','criticized','questioned','overruled') "
                 "  GROUP BY pic.cited_case_law_id) "
                 "UPDATE halachot h SET verified=true, cite_count=cc.n, updated_at=now() "
                 "FROM cc WHERE h.case_law_id = cc.id")
@@ -6564,6 +6571,55 @@ async def refresh_verified_layer() -> dict:
     return {"verified_principles": row["vp"], "verified_precedents": row["vc"]}
 
 
+# X11 §4 treatment buckets (mirrors corroboration.TREATMENT_POSITIVE/NEGATIVE) —
+# kept here so the SQL layer can label a breakdown without importing the service.
+_TREATMENT_POSITIVE = ("followed", "explained")
+_TREATMENT_NEGATIVE = ("distinguished", "criticized", "questioned", "overruled")
+
+
+async def citation_authority(case_law_ids: list["UUID"]) -> dict[str, dict]:
+    """Per-precedent incoming-citation breakdown by treatment (X11 Phase 2, #154).
+
+    For each precedent id → how many DISTINCT committee decisions cite it, split
+    into positive (followed/explained), negative (distinguished/criticized/
+    questioned/overruled) and unclassified (treatment not yet labelled). This is the
+    'cited_by N (X אומצו, Y אובחנו)' authority signal surfaced to research agents so
+    they can argue authority — and avoid leaning on a precedent that was repeatedly
+    distinguished/overruled. Counts distinct sources; a source with no treatment yet
+    falls in 'unclassified'. Returns {} for ids with no incoming committee citations.
+    """
+    if not case_law_ids:
+        return {}
+    pool = await get_pool()
+    rows = await pool.fetch(
+        "SELECT pic.cited_case_law_id::text AS id, "
+        "       coalesce(NULLIF(pic.treatment, ''), 'unclassified') AS t, "
+        "       count(DISTINCT pic.source_case_law_id) AS n "
+        "FROM precedent_internal_citations pic "
+        "JOIN case_law src ON src.id = pic.source_case_law_id "
+        "WHERE src.source_kind = 'internal_committee' "
+        "  AND pic.cited_case_law_id = ANY($1::uuid[]) "
+        "GROUP BY 1, 2",
+        case_law_ids,
+    )
+    out: dict[str, dict] = {}
+    for r in rows:
+        d = out.setdefault(r["id"], {
+            "total": 0, "positive": 0, "negative": 0, "unclassified": 0,
+            "by_treatment": {},
+        })
+        t, n = r["t"], int(r["n"])
+        d["by_treatment"][t] = d["by_treatment"].get(t, 0) + n
+        d["total"] += n
+        if t in _TREATMENT_POSITIVE:
+            d["positive"] += n
+        elif t in _TREATMENT_NEGATIVE:
+            d["negative"] += n
+        else:
+            d["unclassified"] += n
+    return out
+
+
 async def list_canonical_instances(canonical_id: "UUID") -> list[dict]:
     """List all halachot (instances) sharing a canonical_id — used by the UI accordion."""
     pool = await get_pool()

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

@@ -452,10 +452,19 @@ async def get_precedent(case_law_id: UUID | str) -> dict | None:
                 r["source_date"].isoformat()
                 if r.get("source_date") is not None else None
             ),
+            "treatment": r.get("treatment") or "",
             "confidence": r.get("confidence"),
         }
         for r in raw
     ]
+    # Authority signal (X11 Phase 2, #154): how the citing committee decisions
+    # TREATED this ruling (followed/distinguished/…) — surfaced so the chair (and
+    # research agents via the tool output) can argue authority and avoid leaning on
+    # a repeatedly-distinguished precedent.
+    record["cited_by"] = (await db.citation_authority([case_law_id])).get(
+        str(case_law_id),
+        {"total": 0, "positive": 0, "negative": 0, "unclassified": 0, "by_treatment": {}},
+    )
     return record
 
 

mcp-server/src/legal_mcp/tools/precedent_library.py

@@ -300,6 +300,20 @@ async def search_precedent_library(
         limit=limit,
         include_halachot=include_halachot,
     )
+    # X11 Phase 2 (#154): attach the incoming-citation authority breakdown so the
+    # research agent can WEIGH and ARGUE authority ("הלכה שאומצה ב-N החלטות ועדת-ערר")
+    # — and steer clear of a precedent that committees repeatedly distinguished /
+    # overruled. Batched: one query for the whole result page.
+    try:
+        ids = {str(r.get("case_law_id")) for r in results if r.get("case_law_id")}
+        if ids:
+            auth = await db.citation_authority([UUID(i) for i in ids])
+            for r in results:
+                cb = auth.get(str(r.get("case_law_id")))
+                if cb:
+                    r["cited_by"] = cb
+    except Exception:  # noqa: BLE001 — authority is an additive signal; never break search
+        pass
     elapsed_ms = int((time.perf_counter() - t0) * 1000)
     telemetry.log_search_bg(
         search_type="precedent_library",