Merge pull request 'fix(corpus): re-link orphan citations when a cited ruling is uploaded (G1)' (#309) from worktree-citation-relink into main

precedent_internal_citations resolves cited_case_law_id only at extraction
time. When a decision cites a ruling that isn't in the corpus yet, the edge is
stored with cited_case_law_id=NULL. The ruling is often uploaded days later —
but the orphan edge was never re-resolved, so the citation graph permanently
under-counted real connectivity (same stale-derived-value pattern as the
searchable flag). Worse, external rulings store case_number WITHOUT the court
prefix ("3213/97") while citations carry it ("ע\"א 3213/97"), so a large share
of "missing" citations were actually present-but-unlinked.

Fix:
- new citation_extractor.relink_orphan_citations(case_law_id=None): re-resolves
  orphan edges via the canonical _resolve_case_law_id (no parallel matching, G2);
  scoped to one row, or full sweep when None. Idempotent.
- ingest_document calls it on every upload (non-fatal) so the graph self-heals.

Existing backlog already re-linked via a one-off sweep against the live DB:
linked 128 edges → in-corpus distinct rulings 67→178, unlinked distinct
262→143. So real "cited but not uploaded" ≈ 143, not the 262 the stale graph
implied.

Invariants: G1 (re-resolve the derived link at the write) · G2 (single resolver).

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

docs/spec/X8-field-provenance.md

@@ -35,14 +35,14 @@
 | `case_number` (citation) | CHAIR (חובה) | מפתח idempotency |
 | `full_text`, `extraction_status`, `source_kind` | DETERMINISTIC | — |
 | `case_name`, `court`, `date`, `headnote`, `summary`, `key_quote`, `subject_tags`, `appeal_subtype`, `precedent_level`, `source_type`, `citation_formatted` | CHAIR או OPUS | Opus ממלא רק אם ריק |
-| `is_binding` | CHAIR (default true) | קובע prompt-הלכה |
+| `is_binding` | CHAIR (default true) **אך** DETERMINISTIC-false ל-`source_type=appeals_committee` | קובע prompt-הלכה. סמכות מבנית ([INV-DM7](02-data-model.md#inv-dm7-סיווג-הלכה--סמכות-נגזרת--תפקיד-כלל-מסווג-שני-צירים-לא-enum-אחד)): מקור-ועדה הוא תמיד persuasive — נכפה ל-false ב-`create_external_case_law` ונאכף ע"י `case_law_committee_not_binding_check` |
 | chunks (`content`/`section_type`/`page_number`) | DETERMINISTIC | — |
 | `embedding` (chunks) | Voyage (לא-LLM-reasoning) | ⚠ לא-GENERATED ([gap-audit GAP-09](gap-audit.md)) |
 | כל `halachot` | OPUS | נכנס pending_review |
 
 ### 2ב. החלטה פנימית (`case_law`, source_kind=`internal_committee`)
 כמו 2א, ובנוסף: `case_number` **חובה**; `chair_name`/`district`/`proceeding_type` — CHAIR או OPUS או DERIVED;
-`source_type` = `appeals_committee` (DETERMINISTIC קבוע). placeholder `"(טרם חולץ)"` מסומן ל-chair_name/district
+`source_type` = `appeals_committee` (DETERMINISTIC קבוע); `is_binding` = `false` (DETERMINISTIC קבוע — נכפה ב-`create_internal_committee_decision`, [INV-DM7](02-data-model.md#inv-dm7-סיווג-הלכה--סמכות-נגזרת--תפקיד-כלל-מסווג-שני-צירים-לא-enum-אחד): החלטת-ועדה אינה מחייבת ועדה אחרת ולא את עצמה — תמיד persuasive). placeholder `"(טרם חולץ)"` מסומן ל-chair_name/district
 ריקים ומטופל כריק ע"י ה-extractor.
 
 ### 2ג. מסמך-תיק (`documents`)

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

@@ -432,3 +432,42 @@ async def get_cited_case_law_ids(source_case_law_ids: list[UUID]) -> dict[str, l
     for r in rows:
         out.setdefault(r["source_id"], []).append(r["cited_id"])
     return out
+
+
+async def relink_orphan_citations(case_law_id: "UUID | None" = None) -> int:
+    """Back-fill ``cited_case_law_id`` on orphan citation edges that now resolve.
+
+    Citations are resolved to a corpus row only at extraction time (see
+    ``extract_and_store``). When a cited ruling is uploaded LATER, its existing
+    orphan edges (``cited_case_law_id IS NULL``) are never re-resolved, so the
+    citation graph under-counts real connectivity. Call on every precedent
+    upload (``case_law_id`` set → link only edges that resolve to that row) or as
+    a one-off sweep (``case_law_id=None`` → re-resolve every orphan). Reuses the
+    canonical resolver ``_resolve_case_law_id`` (no parallel matching logic, G2);
+    idempotent — a second run links nothing new.
+    """
+    pool = await db.get_pool()
+    async with pool.acquire() as conn:
+        orphans = await conn.fetch(
+            "SELECT DISTINCT cited_case_number FROM precedent_internal_citations "
+            "WHERE cited_case_law_id IS NULL AND coalesce(cited_case_number, '') <> ''"
+        )
+    linked = 0
+    for r in orphans:
+        cited = await _resolve_case_law_id(_normalize_case_number(r["cited_case_number"]))
+        if cited is None:
+            continue
+        if case_law_id is not None and cited != case_law_id:
+            continue
+        async with pool.acquire() as conn:
+            res = await conn.execute(
+                "UPDATE precedent_internal_citations "
+                "SET cited_case_law_id = $1, confidence = GREATEST(confidence, 0.90) "
+                "WHERE cited_case_law_id IS NULL AND cited_case_number = $2",
+                cited, r["cited_case_number"],
+            )
+        try:
+            linked += int(res.split()[-1])
+        except (ValueError, IndexError):
+            pass
+    return linked

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

@@ -1669,6 +1669,40 @@ CREATE INDEX IF NOT EXISTS idx_hcc_canonical
 """
 
 
+SCHEMA_V42_SQL = """
+-- INV-DM7: authority (binding/persuasive) is STRUCTURAL for committee sources.
+-- An appeals-committee decision (source_kind='internal_committee', or any row
+-- with source_type='appeals_committee') is persuasive, NEVER binding — a
+-- committee's ruling does not bind another committee, nor itself. is_binding
+-- stays a stored column (it selects the halacha-extraction prompt), but for
+-- committee sources it is a faithful cache of the derived value, not a chair
+-- guess. Mirrors 02-data-model.md §INV-DM7 / X8-field-provenance.md.
+-- Step 1 — normalize legacy rows at the source (G1), idempotent.
+UPDATE case_law SET is_binding = FALSE
+    WHERE is_binding IS TRUE
+      AND (source_kind = 'internal_committee' OR source_type = 'appeals_committee');
+-- Step 2 — constrain so a binding committee row can never be written again.
+DO $$ BEGIN
+    ALTER TABLE case_law ADD CONSTRAINT case_law_committee_not_binding_check
+        CHECK (NOT ((source_kind = 'internal_committee'
+                     OR source_type = 'appeals_committee') AND is_binding));
+EXCEPTION WHEN duplicate_object THEN NULL; END $$;
+"""
+
+
+SCHEMA_V43_SQL = """
+-- חיפה (Haifa) is a distinct planning district, separate from הצפון (North).
+-- A bug in _district_from_court mapped "חיפה" -> "צפון", and the service-layer
+-- _VALID_DISTRICTS omitted "חיפה" entirely, so Haifa committee decisions were
+-- mis-filed under צפון. Both fixed in internal_decisions.py; this reclassifies
+-- the legacy rows at the source (G1) by their court text. Idempotent.
+UPDATE case_law SET district = 'חיפה'
+    WHERE source_kind = 'internal_committee'
+      AND district = 'צפון'
+      AND court ILIKE '%חיפה%';
+"""
+
+
 # Stable, arbitrary key for the session-level advisory lock that serialises
 # schema DDL across processes. Every short-lived process (cron drains, services)
 # re-runs the idempotent migrations on startup; without this lock two processes
@@ -1686,7 +1720,7 @@ async def _run_schema_migrations(pool: asyncpg.Pool) -> None:
             await _apply_schema_ddl(conn)
         finally:
             await conn.execute("SELECT pg_advisory_unlock($1)", _MIGRATION_LOCK_KEY)
-    logger.info("Database schema initialized (v1-v41)")
+    logger.info("Database schema initialized (v1-v43)")
 
 
 async def _apply_schema_ddl(conn: asyncpg.Connection) -> None:
@@ -1732,6 +1766,8 @@ async def _apply_schema_ddl(conn: asyncpg.Connection) -> None:
         await conn.execute(SCHEMA_V39_SQL)
         await conn.execute(SCHEMA_V40_SQL)
         await conn.execute(SCHEMA_V41_SQL)
+        await conn.execute(SCHEMA_V42_SQL)
+        await conn.execute(SCHEMA_V43_SQL)
 
 
 async def init_schema() -> None:
@@ -4249,6 +4285,11 @@ async def create_external_case_law(
     source_kind='cited_only' (auto-discovered), promote it to
     source_kind='external_upload' and fill in the missing fields.
     """
+    # INV-DM7: an appeals-committee source is persuasive even when uploaded via
+    # the external path — coerce to non-binding so it matches the committee
+    # invariant and case_law_committee_not_binding_check (G1).
+    if source_type == "appeals_committee":
+        is_binding = False
     pool = await get_pool()
     tags_json = json.dumps(subject_tags or [], ensure_ascii=False)
     async with pool.acquire() as conn:
@@ -4317,7 +4358,7 @@ async def create_internal_committee_decision(
     appeal_subtype: str = "",
     subject_tags: list[str] | None = None,
     summary: str = "",
-    is_binding: bool = True,
+    is_binding: bool = False,
     document_id: UUID | None = None,
     proceeding_type: str = "ערר",
 ) -> dict:
@@ -4327,6 +4368,10 @@ async def create_internal_committee_decision(
     exist as both 'ערר' and 'בל"מ' (an extension-of-time request can be
     filed against an existing appeal with the same number).
     """
+    # INV-DM7: authority is structural for this source — an appeals-committee
+    # decision is persuasive, never binding. Coerce regardless of caller so the
+    # value can never violate case_law_committee_not_binding_check (G1).
+    is_binding = False
     pool = await get_pool()
     case_number = _canonical_case_number(case_number)
     tags_json = json.dumps(subject_tags or [], ensure_ascii=False)
@@ -4470,6 +4515,12 @@ async def update_case_law(case_law_id: UUID, **fields) -> dict | None:
         "proceeding_type", "citation_formatted", "parties",
     }
     updates = {k: v for k, v in fields.items() if k in allowed}
+    # INV-DM7: reclassifying a row to an appeals-committee source makes it
+    # persuasive — coerce is_binding in the same patch so the UPDATE can't
+    # violate case_law_committee_not_binding_check (e.g. the Gemini metadata
+    # extractor relabels an external court_ruling as appeals_committee).
+    if updates.get("source_type") == "appeals_committee":
+        updates["is_binding"] = False
     if not updates:
         return await get_case_law(case_law_id)
 
@@ -4483,7 +4534,16 @@ async def update_case_law(case_law_id: UUID, **fields) -> dict | None:
         params.append(v)
     sql = f"UPDATE case_law SET {', '.join(set_parts)} WHERE id = $1 RETURNING *"
     row = await pool.fetchrow(sql, *params)
-    return _row_to_case_law(row) if row else None
+    if row is None:
+        return None
+    # `searchable` is a DERIVED completeness flag (INV-DM1). It is computed at
+    # ingest end and after the Gemini metadata extractor — but internal-committee
+    # decisions skip Gemini, so when their summary/tags/metadata are filled later
+    # (manual edit, deterministic enrichment) the flag would otherwise go stale
+    # and the row stays invisible to RAG. Recompute at this write so the derived
+    # value never drifts from the content (G1: normalize at the source).
+    await recompute_searchable(case_law_id)
+    return await get_case_law(case_law_id)
 
 
 async def set_case_law_extraction_status(case_law_id: UUID, status: str) -> None:

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

@@ -233,6 +233,17 @@ async def ingest_document(
             await db.request_metadata_extraction(case_law_id)
             await db.request_halacha_extraction(case_law_id)
             await db.recompute_searchable(case_law_id)
+            # Citations resolve to a corpus row only at extraction time. A ruling
+            # uploaded AFTER it was first cited leaves orphan edges (NULL link),
+            # so re-link them to this freshly-ingested row now — the citation
+            # graph self-heals instead of permanently under-counting (G1). Non-fatal.
+            try:
+                from legal_mcp.services import citation_extractor as _ce
+                relinked = await _ce.relink_orphan_citations(case_law_id)
+                if relinked:
+                    logger.info("relinked %d orphan citation(s) -> %s", relinked, case_law_id)
+            except Exception as e:  # noqa: BLE001 — relink is best-effort
+                logger.warning("citation relink failed (non-fatal): %s", e)
 
             await progress("completed", 100,
                            f"נקלט: {stored_chunks} chunks. חילוץ הלכות ומטא-דאטה ממתינים בתור.")

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

@@ -28,14 +28,14 @@ logger = logging.getLogger(__name__)
 INTERNAL_DECISIONS_DIR = Path(config.DATA_DIR) / "internal-decisions"
 
 _VALID_PRACTICE_AREAS = frozenset({"", "rishuy_uvniya", "betterment_levy", "compensation_197"})
-_VALID_DISTRICTS = frozenset({"", "ירושלים", "מרכז", "תל אביב", "צפון", "דרום", "ארצי"})
+_VALID_DISTRICTS = frozenset({"", "ירושלים", "מרכז", "תל אביב", "חיפה", "צפון", "דרום", "ארצי"})
 
 _COURT_TO_DISTRICT = [
     ("ירושלים", "ירושלים"),
     ("תל אביב", "תל אביב"),
     ('ת"א', "תל אביב"),
     ("מרכז", "מרכז"),
-    ("חיפה", "צפון"),
+    ("חיפה", "חיפה"),
     ("צפון", "צפון"),
     ("דרום", "דרום"),
     ("ארצי", "ארצי"),
@@ -77,7 +77,7 @@ async def _create_internal_record(**kw) -> dict:
         appeal_subtype=(kw.get("appeal_subtype") or "").strip(),
         subject_tags=list(kw.get("subject_tags") or []),
         summary=(kw.get("summary") or "").strip(),
-        is_binding=kw.get("is_binding", True),
+        is_binding=kw.get("is_binding", False),  # INV-DM7: committee = persuasive
         document_id=kw.get("document_id"),
         proceeding_type=kw.get("proceeding_type") or "ערר",
     )
@@ -108,7 +108,7 @@ async def ingest_internal_decision(
     appeal_subtype: str = "",
     subject_tags: list[str] | None = None,
     summary: str = "",
-    is_binding: bool = True,
+    is_binding: bool = False,  # INV-DM7: committee sources are persuasive
     file_path: str | Path | None = None,
     text: str | None = None,
     document_id: UUID | None = None,
@@ -229,6 +229,7 @@ async def migrate_from_external_corpus(dry_run: bool = False) -> dict:
             await conn.execute(
                 """UPDATE case_law
                    SET source_kind = 'internal_committee',
+                       is_binding = FALSE,  -- INV-DM7: committee = persuasive
                        district = CASE WHEN $2 <> '' THEN $2 ELSE district END
                    WHERE id = $1""",
                 row["id"], district,

web/app.py

@@ -7141,7 +7141,7 @@ async def internal_decisions_upload(
     practice_area: str = Form(""),
     appeal_subtype: str = Form(""),
     subject_tags: str = Form("[]"),
-    is_binding: bool = Form(True),
+    is_binding: bool = Form(False),  # INV-DM7: committee = persuasive (coerced at db too)
     summary: str = Form(""),
 ):
     """Upload a planning appeals-committee decision to the internal corpus.