feat(arguments): הפרדת-משיבים בצבירה לפי כתב-תשובה (#224)

הצבירה קיבצה את כל המשיבים תחת מפלגה אחת "respondent", כך שכתבי-תשובה
נפרדים (משיבות 2-3 מול משיבים 4-6) עם עמדות שעלולות להיות מנוגדות נבלעו
לרשימה אחת. התגלה בתיק 1043-02-26.

השינוי מפצל את הצדדים הרב-משתתפים (respondent/permit_applicant) לפי
source_document — כל כתב-תשובה משותף = יחידת-ליטיגציה קוהרנטית נפרדת —
ומאחסן את התווית ב-legal_arguments.party_name (מיגרציה V50). עוררים/ועדה
נשארים קבוצה יחידה. בונוס: פיצול ה-129 טענות ל-קריאות-Claude קטנות מתקן
את הכשל-בשקט המקורי (קריאה של 100+ פרופוזיציות החזירה non-JSON והפילה את
כל הצד).

- db.py: SCHEMA_V50_SQL — legal_arguments.party_name (ADD COLUMN IF NOT EXISTS, אידמפוטנטי)
- argument_aggregator: קיבוץ (party, party_name); _build_prompt מקבל שם-כתב; SELECT מחזיר party_name
- UI: PartySection מתת-קבץ לפי כתב-תשובה; טיפוס LegalArgument.party_name

טווח v1: הקיבוץ לפי source_document; אכלוס party_name ב-extractor וזיהוי
עמדות-מנוגדות בבלוק ז/י — המשך ב-#224.

Invariants: G1 (נרמול-במקור — party_name בסכמה, לא תיקון-בקריאה) · G2 (אין
מסלול-צבירה מקביל, אותו pipeline) · §6 (אין בליעה שקטה — הכשל-בשקט תוקן).

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

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

@@ -31,6 +31,10 @@ logger = logging.getLogger(__name__)
 
 # Allowed enum values mirror the DB CHECK constraints.
 ALLOWED_PARTIES = {"appellant", "respondent", "committee", "permit_applicant", "unknown"}
+# Sides that may comprise multiple distinct litigants with potentially opposing
+# positions — aggregated per source pleading (party_name) rather than collapsed
+# into one bucket (#224). Appellant/committee speak with a single voice.
+SPLIT_PARTIES = {"respondent", "permit_applicant"}
 ALLOWED_PRIORITIES = {"threshold", "substantive", "procedural", "relief"}
 
 # Hebrew labels for the prompt (Claude needs context in the same
@@ -81,7 +85,7 @@ AGGREGATE_PROMPT_TEMPLATE = """אתה מנתח כתבי טענות בתחום ת
 """
 
 
-def _build_prompt(party: str, propositions: list[dict]) -> str:
+def _build_prompt(party: str, propositions: list[dict], party_name: str = "") -> str:
     """Compose the per-party aggregation prompt."""
     n = len(propositions)
     # Conservative target: ~1 argument per 2-3 propositions, clamped 4-12.
@@ -89,6 +93,11 @@ def _build_prompt(party: str, propositions: list[dict]) -> str:
     target_max = max(target_min + 1, min(12, n // 2 + 1))
 
     party_he = PARTY_LABELS_HE.get(party, party)
+    # For a split side (e.g. a specific respondent brief), name the brief so
+    # Claude aggregates only that litigant's position and does not conflate it
+    # with a co-respondent's separate — possibly opposing — pleading (#224).
+    if party_name:
+        party_he = f"{party_he} — {party_name}"
     # Strip noise from propositions for the prompt — Claude only needs
     # the id and the text to do the grouping.
     compact = [
@@ -139,12 +148,17 @@ def _normalize_argument(raw: dict, fallback_topic: str = "") -> dict | None:
 
 
 async def _aggregate_party(
-    party: str, propositions: list[dict],
+    party: str, propositions: list[dict], party_name: str = "",
 ) -> list[dict]:
-    """Ask Claude to group one party's propositions; return normalized rows."""
+    """Ask Claude to group one party's propositions; return normalized rows.
+
+    ``party_name`` names the specific pleading when this is a split side
+    (respondent / permit_applicant brief), so the prompt scopes to that
+    litigant's position only (#224).
+    """
     if not propositions:
         return []
-    prompt = _build_prompt(party, propositions)
+    prompt = _build_prompt(party, propositions, party_name=party_name)
 
     try:
         raw_result = await claude_session.query_json(prompt, tools="")  # no tool_use → no error_max_turns
@@ -221,14 +235,25 @@ async def aggregate_claims_to_arguments(
             "total": 0,
         }
 
-    # Group propositions by party.
-    by_party: dict[str, list[dict]] = {}
+    # Group propositions by party — and, for the multi-party sides (respondent /
+    # permit_applicant), further by their source pleading so opposing joint
+    # briefs (e.g. "כתב תשובה משיבות 2-3" vs "משיבים 4-6") aggregate into
+    # SEPARATE argument sets instead of collapsing into one "respondent" bucket
+    # (#224). Appellant/committee stay single-group (party_name=""). Splitting
+    # the large respondent set per-brief also keeps each Claude call small enough
+    # to succeed — a single 100+ proposition call previously returned non-JSON
+    # and silently dropped the whole side.
+    by_group: dict[tuple[str, str], list[dict]] = {}
     for r in rows:
         party = r["party_role"]
         # Map deprecated 'appraiser' or unknown labels to 'unknown'.
         if party not in ALLOWED_PARTIES:
             party = "unknown"
-        by_party.setdefault(party, []).append(dict(r))
+        party_name = (
+            (r["source_document"] or "").strip()
+            if party in SPLIT_PARTIES else ""
+        )
+        by_group.setdefault((party, party_name), []).append(dict(r))
 
     # Valid claim_ids for this case == the ids of the claims we just fetched.
     # The LLM is asked to echo back supporting claim_ids, but it may hallucinate
@@ -243,9 +268,11 @@ async def aggregate_claims_to_arguments(
     inserted = 0
     errors: list[str] = []
 
-    for party, props in by_party.items():
+    for (party, party_name), props in by_group.items():
+        # Display key for the per-side summary: keep opposing briefs distinct.
+        group_key = f"{party}·{party_name}" if party_name else party
         try:
-            arguments = await _aggregate_party(party, props)
+            arguments = await _aggregate_party(party, props, party_name=party_name)
         except RuntimeError as e:
             # Most likely cause: Claude CLI not installed (running from
             # the container). Don't crash — record the gap and continue.
@@ -259,11 +286,11 @@ async def aggregate_claims_to_arguments(
                     ),
                     "total": 0,
                 }
-            errors.append(f"{party}: {msg}")
+            errors.append(f"{group_key}: {msg}")
             continue
 
         if not arguments:
-            party_counts[party] = 0
+            party_counts[group_key] = 0
             continue
 
         async with pool.acquire() as conn:
@@ -271,12 +298,13 @@ async def aggregate_claims_to_arguments(
                 for idx, arg in enumerate(arguments):
                     arg_id = await conn.fetchval(
                         """INSERT INTO legal_arguments
-                            (case_id, party, argument_index, argument_title,
-                             argument_body, legal_topic, priority)
-                           VALUES ($1, $2, $3, $4, $5, $6, $7)
+                            (case_id, party, party_name, argument_index,
+                             argument_title, argument_body, legal_topic, priority)
+                           VALUES ($1, $2, $3, $4, $5, $6, $7, $8)
                            RETURNING id""",
                         case_id,
                         party,
+                        party_name,
                         idx + 1,
                         arg["title"],
                         arg["body"],
@@ -313,7 +341,7 @@ async def aggregate_claims_to_arguments(
                                 cid, arg_id, e,
                             )
                     inserted += 1
-        party_counts[party] = len(arguments)
+        party_counts[group_key] = len(arguments)
 
     result: dict = {
         "status": "completed",
@@ -338,9 +366,9 @@ async def get_legal_arguments(
     async with pool.acquire() as conn:
         if party and party in ALLOWED_PARTIES:
             rows = await conn.fetch(
-                """SELECT id, case_id, party, argument_index, argument_title,
-                          argument_body, legal_topic, priority, cited_precedents,
-                          created_at, updated_at
+                """SELECT id, case_id, party, party_name, argument_index,
+                          argument_title, argument_body, legal_topic, priority,
+                          cited_precedents, created_at, updated_at
                    FROM legal_arguments
                    WHERE case_id = $1 AND party = $2
                    ORDER BY priority, argument_index""",
@@ -348,12 +376,12 @@ async def get_legal_arguments(
             )
         else:
             rows = await conn.fetch(
-                """SELECT id, case_id, party, argument_index, argument_title,
-                          argument_body, legal_topic, priority, cited_precedents,
-                          created_at, updated_at
+                """SELECT id, case_id, party, party_name, argument_index,
+                          argument_title, argument_body, legal_topic, priority,
+                          cited_precedents, created_at, updated_at
                    FROM legal_arguments
                    WHERE case_id = $1
-                   ORDER BY party, priority, argument_index""",
+                   ORDER BY party, party_name, priority, argument_index""",
                 case_id,
             )
 

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

@@ -1888,6 +1888,18 @@ CREATE INDEX IF NOT EXISTS idx_documents_claims_pending
 """
 
 
+# V50 (#224): per-respondent-brief separation. The aggregator groups respondent /
+# permit_applicant claims by their source pleading — each joint response brief
+# (e.g. "כתב תשובה משיבות 2-3" vs "משיבים 4-6") is one coherent litigation
+# position, so opposing briefs no longer collapse into a single "respondent"
+# bucket. ``party_name`` carries the brief label; appellant/committee stay ''.
+SCHEMA_V50_SQL = """
+ALTER TABLE legal_arguments ADD COLUMN IF NOT EXISTS party_name TEXT NOT NULL DEFAULT '';
+CREATE INDEX IF NOT EXISTS idx_legal_arguments_party_name
+    ON legal_arguments(case_id, party, party_name);
+"""
+
+
 # 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
@@ -1959,6 +1971,7 @@ async def _apply_schema_ddl(conn: asyncpg.Connection) -> None:
         await conn.execute(SCHEMA_V47_SQL)
         await conn.execute(SCHEMA_V48_SQL)
         await conn.execute(SCHEMA_V49_SQL)
+        await conn.execute(SCHEMA_V50_SQL)
 
 
 async def init_schema() -> None: