Merge pull request 'feat(claims): אכלוס claims.party_name במקור + backfill — סגירת השדה-המת (#224)' (#403) from worktree-claims-party-name-populate into main

עד כה claims.party_name נשאר ריק והצבירה/הכתיבה גזרו את הפרדת-המשיבים
מ-source_document בזמן-ריצה (שדה-מת). עכשיו הוא מאוכלס במקור ונקרא ע"י הצרכנים.

- claims_extractor.extract_and_store_claims: מחתים party_name=source_document
  ל-SPLIT_PARTIES (respondent/permit_applicant); single-voice נשאר ''. אותו
  כלל SPLIT_PARTIES של הצבירה (G2, import לא-מעגלי).
- SCHEMA_V52: backfill idempotent ל-claims קיימים (רק split + party_name ריק).
- argument_aggregator + block_writer._build_claims_context: קוראים
  claims.party_name, עם fallback ל-source_document ל-claims מדור-קודם. התנהגות
  זהה (party_name==source_document) — אפס-רגרסיה, אך השדה כעת מקור-האמת (G1).

py_compile + טעינת-שרשרת-ייבוא נקיים. סוגר את הפתוח האחרון של #224.
Invariants: G1 (אכלוס במקור, לא גזירה-בקריאה), G2 (כלל-פיצול יחיד).

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

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

@@ -221,7 +221,7 @@ async def aggregate_claims_to_arguments(
 
         # Pull all claims for this case, grouped by party.
         rows = await conn.fetch(
-            """SELECT id, party_role, claim_text, claim_index, source_document
+            """SELECT id, party_role, claim_text, claim_index, source_document, party_name
                FROM claims
                WHERE case_id = $1
                ORDER BY party_role, claim_index""",
@@ -249,10 +249,13 @@ async def aggregate_claims_to_arguments(
         # Map deprecated 'appraiser' or unknown labels to 'unknown'.
         if party not in ALLOWED_PARTIES:
             party = "unknown"
-        party_name = (
+        # Prefer the stored party_name (stamped by the extractor, #224); fall
-            (r["source_document"] or "").strip()
+        # back to source_document for legacy claims predating the stamping.
-            if party in SPLIT_PARTIES else ""
+        # Single-voice sides (appellant/committee) stay ''.
-        )
+        if party in SPLIT_PARTIES:
+            party_name = (r["party_name"] or "").strip() or (r["source_document"] or "").strip()
+        else:
+            party_name = ""
         by_group.setdefault((party, party_name), []).append(dict(r))
 
     # Valid claim_ids for this case == the ids of the claims we just fetched.

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

@@ -645,7 +645,12 @@ async def _build_claims_context(case_id: UUID) -> str:
     groups: dict[tuple[str, str], list[dict]] = {}
     for c in source_claims:
         role = c.get("party_role", "") or ""
-        brief = (c.get("source_document", "") or "").strip() if role in SPLIT_PARTIES else ""
+        # Prefer the stored party_name (stamped by the extractor, #224); fall
+        # back to source_document for legacy claims predating the stamping.
+        if role in SPLIT_PARTIES:
+            brief = (c.get("party_name") or "").strip() or (c.get("source_document") or "").strip()
+        else:
+            brief = ""
         groups.setdefault((role, brief), []).append(c)
 
     def _sort_key(k: tuple[str, str]) -> tuple[int, str]:

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

@@ -368,8 +368,15 @@ async def extract_and_store_claims(
 
     # Determine claim_type from document type and title
     claim_type = _infer_claim_type(doc_type, source_name)
+    # Stamp party_name at the source (#224). For a multi-litigant side
+    # (respondent / permit_applicant) the brief label IS the source pleading, so
+    # opposing briefs (משיבות 2-3 vs משיבים 4-6) stay distinct downstream without
+    # the aggregator/block-writer having to re-derive it. SPLIT_PARTIES is the
+    # single rule shared with the aggregator (G2); single-voice sides stay ''.
+    from legal_mcp.services.argument_aggregator import SPLIT_PARTIES
     for c in claims:
         c["claim_type"] = claim_type
+        c["party_name"] = source_name if c.get("party_role") in SPLIT_PARTIES else ""
 
     stored = await db.store_claims(case_id, claims, source_document=source_name)
     # Mark this document analysed (WS2 / #201). store_claims already replaced

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

@@ -1918,6 +1918,20 @@ ALTER TABLE cases ADD COLUMN IF NOT EXISTS hearing_attendees JSONB NOT NULL DEFA
 """
 
 
+# V52 (#224): backfill claims.party_name for existing rows. The extractor now
+# stamps party_name = source_document for the multi-litigant sides at write time
+# (claims_extractor), but claims stored before that shipped have party_name=''.
+# For those, the brief label is the source pleading — same rule the aggregator
+# uses (respondent / permit_applicant split). Idempotent: only fills empties, so
+# re-running is a no-op. Single-voice sides (appellant/committee) stay ''.
+SCHEMA_V52_SQL = """
+UPDATE claims SET party_name = source_document
+WHERE party_role IN ('respondent', 'permit_applicant')
+  AND COALESCE(party_name, '') = ''
+  AND COALESCE(source_document, '') <> '';
+"""
+
+
 # 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
@@ -1991,6 +2005,7 @@ async def _apply_schema_ddl(conn: asyncpg.Connection) -> None:
         await conn.execute(SCHEMA_V49_SQL)
         await conn.execute(SCHEMA_V50_SQL)
         await conn.execute(SCHEMA_V51_SQL)
+        await conn.execute(SCHEMA_V52_SQL)
 
 
 async def init_schema() -> None: