fix(graph): normalize case_law subject_tags to underscore convention at write chokepoint

Topic hubs in /graph are built from case_law.subject_tags. The documented
extraction contract (precedent_library tool examples: קווי_בניין,
מועד_קביעת_שומה) and ~99% of the corpus store plain multi-word Hebrew tags
with underscores between words ("היטל_השבחה"). A single case (8126-03-25,
יעקב עמיאל) was tagged with spaces ("היטל השבחה"), which the graph renders as
a SECOND, distinct topic hub — a duplicate of the underscore form. The data
was normalized separately; this enforces the convention at the source so no
write path can re-introduce the split.

_normalize_subject_tags() is applied at the three (and only) case_law write
chokepoints in db.py — create_external_case_law, create_internal_committee_decision,
update_case_law — so the rule cannot be bypassed (G1: normalize at source, not
in the read/graph path). Tags carrying punctuation/digits/dashes
(e.g. "פטור מותנה — סעיף 19(ג)") are left untouched; only plain Hebrew word
phrases ([א-ת]+ separated by spaces) are converted. Also dedups post-normalize.

digests.subject_tags (TEXT[], a different graph layer) and canonical_halachot
are intentionally out of scope.

Invariants: maintains G1 (fix at source, not in the read projection),
G2 (graph_api stays a pure read projection — no parallel normalization there).

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

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

@@ -4261,6 +4261,34 @@ async def case_number_collides(case_number: str, exclude_id: UUID) -> bool:
     ))
 
 
+# Subject-tag convention for case_law topic hubs (powers /graph topic nodes).
+# A simple multi-word Hebrew phrase is stored with underscores between words
+# ("היטל_השבחה") — the documented extraction contract (precedent_library tool
+# examples: קווי_בניין, מועד_קביעת_שומה) that ~99% of the corpus already follows.
+# The same phrase arriving with spaces ("היטל השבחה") is the same topic in
+# disguise and spawns a duplicate graph hub. Normalize at the single case_law
+# write chokepoint (G1) so no path can re-introduce the split. Tags carrying
+# punctuation/digits/dashes (e.g. "פטור מותנה — סעיף 19(ג)") are left untouched —
+# the convention covers only plain Hebrew word phrases.
+_PURE_HEBREW_WORDS = re.compile(r"^[א-ת]+(?: [א-ת]+)+$")
+
+
+def _normalize_subject_tags(tags) -> list[str]:
+    """Strip, apply the underscore convention to plain Hebrew phrases, dedup."""
+    out: list[str] = []
+    seen: set[str] = set()
+    for raw in tags or []:
+        t = str(raw).strip()
+        if not t:
+            continue
+        if _PURE_HEBREW_WORDS.match(t):
+            t = t.replace(" ", "_")
+        if t not in seen:
+            seen.add(t)
+            out.append(t)
+    return out
+
+
 async def create_external_case_law(
     case_number: str,
     case_name: str,
@@ -4291,7 +4319,7 @@ async def create_external_case_law(
     if source_type == "appeals_committee":
         is_binding = False
     pool = await get_pool()
-    tags_json = json.dumps(subject_tags or [], ensure_ascii=False)
+    tags_json = json.dumps(_normalize_subject_tags(subject_tags), ensure_ascii=False)
     async with pool.acquire() as conn:
         # Atomic upsert on the V15 partial unique index
         # uq_case_law_external_number (case_number) WHERE source_kind <> 'internal_committee'.
@@ -4374,7 +4402,7 @@ async def create_internal_committee_decision(
     is_binding = False
     pool = await get_pool()
     case_number = _canonical_case_number(case_number)
-    tags_json = json.dumps(subject_tags or [], ensure_ascii=False)
+    tags_json = json.dumps(_normalize_subject_tags(subject_tags), ensure_ascii=False)
     async with pool.acquire() as conn:
         # Atomic upsert on V15 partial unique index
         # uq_case_law_internal_number_proc (case_number, proceeding_type)
@@ -4529,7 +4557,7 @@ async def update_case_law(case_law_id: UUID, **fields) -> dict | None:
     params: list = [case_law_id]
     for i, (k, v) in enumerate(updates.items(), start=2):
         if k == "subject_tags":
-            v = json.dumps(v or [], ensure_ascii=False)
+            v = json.dumps(_normalize_subject_tags(v), ensure_ascii=False)
         set_parts.append(f"{k} = ${i}")
         params.append(v)
     sql = f"UPDATE case_law SET {', '.join(set_parts)} WHERE id = $1 RETURNING *"