feat: Stage A finalizers + #35/#36/#37 — critical-gap closure

Four parallel sub-agents closed the remaining critical gaps from the
26/05 Stage A/B sprint. Each block independently tested; aggregated here.

## #30/#31 finalizers (sub-agent A)
* Auto-derive practice_area in case_create from case_number prefix
  (1xxx→rishuy_uvniya, 8xxx→betterment_levy, 9xxx→compensation_197);
  default for CaseCreateRequest is now "" (the DB constraint catches
  any stray "appeals_committee").
* practice_area.py: derive_subtype now handles axis-B domain values
  (rishuy_uvniya/betterment_levy/compensation_197) without parsing the
  case number; new helper derive_domain_practice_area().
* Halacha re-extraction verified unnecessary — all 6 reclassified
  records already had is_binding=false and approved halachot.
* Regression tests: 6 cases in tests/test_corpus_constraints.py
  covering practice_area enum, internal-committee chair/district,
  external-upload arar prefix, MCP guard.
* UI: district input → Select dropdown (7 districts) in
  precedent-edit-sheet.tsx, preserving legacy free-text values.

## #37 בל"מ subtypes (sub-agent B)
* 3 new appeal_subtypes: extension_request_{building_permit,
  betterment_levy,compensation}. APPEALS_COMMITTEE_SUBTYPES extended,
  SUBTYPES_BY_AREA mappings added.
* New helpers: is_blam_subject(), is_blam_subtype(),
  derive_subtype_with_blam(case_number, subject, practice_area).
  case_create now uses it to auto-detect "בקשה להארכת מועד" subjects.
* 3 methodology templates under docs/methodology/extension-request-*.md.
* paperclip_client.py mapping updated for the 3 new subtypes
  (extension_request_building_permit→CMP, the other two→CMPA).
* Frontend: bilingual "בל"מ" badge + filter dropdown on cases list +
  detail header; appeal-type-bars collapseBlam() merges בל"מ into its
  parent domain for aggregate bars.
* Wizard auto-detects בל"מ from subject during case creation.
* 3 Berlinger cases (1017/1018/1019-03-26) migrated to
  appeal_subtype=extension_request_building_permit via psql.

## #35 missing_precedents feature (sub-agent C)
* Schema V13: missing_precedents table (citation, case_id, party,
  legal_topic, status, linked_case_law_id, claim_quote, ...) +
  FK constraints + 3 indexes. Applied via psql + idempotent migration.
* 6 db.py service functions, 3 MCP tools, 6 FastAPI endpoints
  (POST/GET/PATCH/DELETE/upload — upload routes by citation prefix
  to ingest_internal_decision or ingest_precedent).
* Next.js page /missing-precedents with 5 status tabs + filters +
  sidebar badge counter + detail drawer with metadata edit + smart
  upload form that switches fields per committee/court.
* Bootstrap: 7 rows imported from the JSON file
  (3 citations × cases, all status=closed with linked_case_law_id).
* legal-researcher.md: new §2ב.5 with missing_precedent_create
  usage + dedup semantics + tool grant.

## #36 legal_arguments aggregation (sub-agent D)
* Schema V14: legal_arguments + legal_argument_propositions M:M.
  Applied via psql.
* New service argument_aggregator.py with two functions —
  aggregate_claims_to_arguments() (Claude CLI / claude_session) and
  get_legal_arguments(). Graceful llm_unavailable handling when CLI
  is missing (containers).
* 2 MCP tools + 2 API endpoints (POST .../aggregate-arguments as
  BackgroundTask, GET .../legal-arguments).
* Frontend: shadcn Accordion + new legal-arguments-panel.tsx with
  hierarchical (party → priority badge → arguments) display, "טיעונים"
  tab on the case page, "חשב/חשב מחדש" buttons.
* scripts/backfill_legal_arguments.py + SCRIPTS.md entry — dry-run
  found 8 candidate cases including 1017/1018/1019.

## Open follow-ups (intentionally deferred)
* npm run api:types in web-ui (CLAUDE.md flow) — recommended before
  the next UI commit; not required for backend deployment.
* Run backfill_legal_arguments.py --apply once the container picks up
  the new aggregator service.
* webhook on missing-precedents upload-close to Paperclip (optional).

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

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

@@ -745,6 +745,84 @@ CREATE INDEX IF NOT EXISTS idx_halachot_tsv
 """
 
 
+# ── V13: Missing precedents log ───────────────────────────────────
+# Track citations that the parties brought up but which are NOT yet in
+# the precedent_library. Created by the researcher (auto or chair)
+# whenever a citation can't be found in the corpus; closed by uploading
+# the actual decision via internal_decision_upload or
+# precedent_library_upload, at which point linked_case_law_id points to
+# the new case_law row and status flips to 'closed'.
+SCHEMA_V13_SQL = """
+CREATE TABLE IF NOT EXISTS missing_precedents (
+    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
+    citation TEXT NOT NULL,
+    case_name TEXT,
+    cited_in_case_id UUID REFERENCES cases(id) ON DELETE CASCADE,
+    cited_in_document_id UUID REFERENCES documents(id) ON DELETE SET NULL,
+    cited_by_party TEXT CHECK (cited_by_party IN (
+        'appellant', 'respondent', 'committee', 'permit_applicant', 'unknown'
+    )),
+    cited_by_party_name TEXT,
+    legal_topic TEXT,
+    legal_issue TEXT,
+    claim_quote TEXT,
+    status TEXT DEFAULT 'open' CHECK (status IN (
+        'open', 'uploaded', 'closed', 'irrelevant'
+    )),
+    linked_case_law_id UUID REFERENCES case_law(id) ON DELETE SET NULL,
+    closed_at TIMESTAMPTZ,
+    created_at TIMESTAMPTZ DEFAULT NOW(),
+    updated_at TIMESTAMPTZ DEFAULT NOW(),
+    notes TEXT
+);
+
+CREATE INDEX IF NOT EXISTS idx_missing_precedents_case
+    ON missing_precedents(cited_in_case_id);
+CREATE INDEX IF NOT EXISTS idx_missing_precedents_status
+    ON missing_precedents(status);
+CREATE INDEX IF NOT EXISTS idx_missing_precedents_citation
+    ON missing_precedents(citation);
+"""
+
+
+# ── V14: Legal arguments (aggregated propositions) ────────────────
+# After ``claims_extractor`` extracts raw propositions (rows in ``claims``)
+# the LLM-driven aggregator groups them into ~6-12 distinct legal arguments
+# per party. ``legal_arguments`` holds the consolidated argument; the M:M
+# join table ``legal_argument_propositions`` links back to the source
+# propositions for traceability ("which raw claims feed this argument?").
+SCHEMA_V14_SQL = """
+CREATE TABLE IF NOT EXISTS legal_arguments (
+    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
+    case_id UUID NOT NULL REFERENCES cases(id) ON DELETE CASCADE,
+    party TEXT NOT NULL CHECK (party IN (
+        'appellant', 'respondent', 'committee', 'permit_applicant', 'unknown'
+    )),
+    argument_index INTEGER NOT NULL,
+    argument_title TEXT NOT NULL,
+    argument_body TEXT NOT NULL,
+    legal_topic TEXT,
+    priority TEXT DEFAULT 'substantive' CHECK (priority IN (
+        'threshold', 'substantive', 'procedural', 'relief'
+    )),
+    cited_precedents TEXT[],
+    created_at TIMESTAMPTZ DEFAULT NOW(),
+    updated_at TIMESTAMPTZ DEFAULT NOW()
+);
+CREATE INDEX IF NOT EXISTS idx_legal_arguments_case
+    ON legal_arguments(case_id);
+CREATE INDEX IF NOT EXISTS idx_legal_arguments_party
+    ON legal_arguments(case_id, party);
+
+-- M:M back to ``claims`` (raw propositions).
+CREATE TABLE IF NOT EXISTS legal_argument_propositions (
+    argument_id UUID NOT NULL REFERENCES legal_arguments(id) ON DELETE CASCADE,
+    claim_id UUID NOT NULL REFERENCES claims(id) ON DELETE CASCADE,
+    PRIMARY KEY (argument_id, claim_id)
+);
+"""
+
+
 async def _run_schema_migrations(pool: asyncpg.Pool) -> None:
     async with pool.acquire() as conn:
         await conn.execute(SCHEMA_SQL)
@@ -760,7 +838,9 @@ async def _run_schema_migrations(pool: asyncpg.Pool) -> None:
         await conn.execute(SCHEMA_V10_SQL)
         await conn.execute(SCHEMA_V11_SQL)
         await conn.execute(SCHEMA_V12_SQL)
-    logger.info("Database schema initialized (v1-v12)")
+        await conn.execute(SCHEMA_V13_SQL)
+        await conn.execute(SCHEMA_V14_SQL)
+    logger.info("Database schema initialized (v1-v14)")
 
 
 async def init_schema() -> None:
@@ -782,7 +862,10 @@ async def create_case(
     hearing_date: date | None = None,
     notes: str = "",
     expected_outcome: str = "",
-    practice_area: str = "appeals_committee",
+    # Default "" — DB CHECK constraint accepts empty, the upstream tool
+    # (cases.case_create) is responsible for deriving the domain value
+    # from the case_number prefix before calling here.
+    practice_area: str = "",
     appeal_subtype: str = "",
 ) -> dict:
     pool = await get_pool()
@@ -3106,3 +3189,228 @@ async def search_precedent_library_hybrid(
         merged.append(d)
     merged.sort(key=lambda x: -x["score"])
     return merged[:limit]
+
+
+# ── Missing precedents (V13) ───────────────────────────────────────
+# Track citations from party briefs that aren't yet in the corpus.
+# Lifecycle: 'open' → researcher logs gap → chair uploads decision
+# → status='uploaded' (file ingested) → status='closed' (linked to
+# case_law row). 'irrelevant' = chair decided the citation isn't worth
+# adding to the library.
+
+ALLOWED_MP_PARTIES = {
+    "appellant", "respondent", "committee", "permit_applicant", "unknown",
+}
+ALLOWED_MP_STATUS = {"open", "uploaded", "closed", "irrelevant"}
+
+
+def _row_to_missing_precedent(row: asyncpg.Record) -> dict:
+    d = dict(row)
+    d["id"] = str(d["id"])
+    if d.get("cited_in_case_id") is not None:
+        d["cited_in_case_id"] = str(d["cited_in_case_id"])
+    if d.get("cited_in_document_id") is not None:
+        d["cited_in_document_id"] = str(d["cited_in_document_id"])
+    if d.get("linked_case_law_id") is not None:
+        d["linked_case_law_id"] = str(d["linked_case_law_id"])
+    return d
+
+
+async def create_missing_precedent(
+    citation: str,
+    case_name: str | None = None,
+    cited_in_case_id: UUID | None = None,
+    cited_in_document_id: UUID | None = None,
+    cited_by_party: str | None = None,
+    cited_by_party_name: str | None = None,
+    legal_topic: str | None = None,
+    legal_issue: str | None = None,
+    claim_quote: str | None = None,
+    notes: str | None = None,
+) -> dict:
+    """Create a new missing-precedent row (status='open' by default)."""
+    if not citation.strip():
+        raise ValueError("citation is required")
+    if cited_by_party and cited_by_party not in ALLOWED_MP_PARTIES:
+        raise ValueError(
+            f"cited_by_party must be one of {sorted(ALLOWED_MP_PARTIES)}"
+        )
+    pool = await get_pool()
+    async with pool.acquire() as conn:
+        row = await conn.fetchrow(
+            """INSERT INTO missing_precedents (
+                citation, case_name, cited_in_case_id, cited_in_document_id,
+                cited_by_party, cited_by_party_name, legal_topic, legal_issue,
+                claim_quote, notes
+            )
+            VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10)
+            RETURNING *""",
+            citation.strip(), case_name, cited_in_case_id, cited_in_document_id,
+            cited_by_party, cited_by_party_name, legal_topic, legal_issue,
+            claim_quote, notes,
+        )
+    return _row_to_missing_precedent(row)
+
+
+async def list_missing_precedents(
+    status: str | None = None,
+    case_id: UUID | None = None,
+    legal_topic: str | None = None,
+    limit: int = 200,
+    offset: int = 0,
+) -> list[dict]:
+    """List missing precedents, joining the cited-in case_number for display."""
+    pool = await get_pool()
+    conditions: list[str] = []
+    params: list = []
+    idx = 1
+    if status:
+        conditions.append(f"mp.status = ${idx}")
+        params.append(status)
+        idx += 1
+    if case_id:
+        conditions.append(f"mp.cited_in_case_id = ${idx}")
+        params.append(case_id)
+        idx += 1
+    if legal_topic:
+        conditions.append(f"mp.legal_topic ILIKE ${idx}")
+        params.append(f"%{legal_topic}%")
+        idx += 1
+    where = f"WHERE {' AND '.join(conditions)}" if conditions else ""
+    params.append(limit)
+    params.append(offset)
+    sql = f"""
+        SELECT mp.*,
+               c.case_number AS cited_in_case_number,
+               cl.case_number AS linked_case_law_number,
+               cl.case_name AS linked_case_law_name
+        FROM missing_precedents mp
+        LEFT JOIN cases c ON c.id = mp.cited_in_case_id
+        LEFT JOIN case_law cl ON cl.id = mp.linked_case_law_id
+        {where}
+        ORDER BY
+            CASE mp.status
+                WHEN 'open' THEN 0
+                WHEN 'uploaded' THEN 1
+                WHEN 'closed' THEN 2
+                WHEN 'irrelevant' THEN 3
+            END,
+            mp.created_at DESC
+        LIMIT ${idx} OFFSET ${idx + 1}
+    """
+    async with pool.acquire() as conn:
+        rows = await conn.fetch(sql, *params)
+    return [_row_to_missing_precedent(r) for r in rows]
+
+
+async def get_missing_precedent(mp_id: UUID) -> dict | None:
+    pool = await get_pool()
+    async with pool.acquire() as conn:
+        row = await conn.fetchrow(
+            """
+            SELECT mp.*,
+                   c.case_number AS cited_in_case_number,
+                   cl.case_number AS linked_case_law_number,
+                   cl.case_name AS linked_case_law_name
+            FROM missing_precedents mp
+            LEFT JOIN cases c ON c.id = mp.cited_in_case_id
+            LEFT JOIN case_law cl ON cl.id = mp.linked_case_law_id
+            WHERE mp.id = $1
+            """,
+            mp_id,
+        )
+    return _row_to_missing_precedent(row) if row else None
+
+
+async def update_missing_precedent(mp_id: UUID, **fields) -> dict | None:
+    """Patch a missing-precedent row. Allowed fields: legal_topic,
+    legal_issue, notes, cited_by_party, cited_by_party_name, case_name,
+    status, linked_case_law_id, closed_at."""
+    if not fields:
+        return await get_missing_precedent(mp_id)
+    allowed = {
+        "legal_topic", "legal_issue", "notes", "cited_by_party",
+        "cited_by_party_name", "case_name", "status", "linked_case_law_id",
+        "closed_at", "claim_quote", "citation",
+    }
+    clean = {k: v for k, v in fields.items() if k in allowed}
+    if not clean:
+        return await get_missing_precedent(mp_id)
+    if "status" in clean and clean["status"] not in ALLOWED_MP_STATUS:
+        raise ValueError(
+            f"status must be one of {sorted(ALLOWED_MP_STATUS)}"
+        )
+    if "cited_by_party" in clean and clean["cited_by_party"] and \
+            clean["cited_by_party"] not in ALLOWED_MP_PARTIES:
+        raise ValueError(
+            f"cited_by_party must be one of {sorted(ALLOWED_MP_PARTIES)}"
+        )
+    set_clauses = []
+    values = []
+    for i, (key, val) in enumerate(clean.items(), start=2):
+        set_clauses.append(f"{key} = ${i}")
+        values.append(val)
+    set_clauses.append("updated_at = now()")
+    sql = (
+        f"UPDATE missing_precedents SET {', '.join(set_clauses)} "
+        f"WHERE id = $1 RETURNING *"
+    )
+    pool = await get_pool()
+    async with pool.acquire() as conn:
+        row = await conn.fetchrow(sql, mp_id, *values)
+    return _row_to_missing_precedent(row) if row else None
+
+
+async def close_missing_precedent(
+    mp_id: UUID,
+    linked_case_law_id: UUID | None = None,
+    notes: str | None = None,
+    status: str = "closed",
+) -> dict | None:
+    """Mark a missing-precedent row as closed (or 'uploaded'/'irrelevant')
+    and link it to a case_law row if provided."""
+    if status not in ALLOWED_MP_STATUS:
+        raise ValueError(
+            f"status must be one of {sorted(ALLOWED_MP_STATUS)}"
+        )
+    pool = await get_pool()
+    async with pool.acquire() as conn:
+        sets = ["status = $2", "closed_at = now()", "updated_at = now()"]
+        params: list = [mp_id, status]
+        idx = 3
+        if linked_case_law_id is not None:
+            sets.append(f"linked_case_law_id = ${idx}")
+            params.append(linked_case_law_id)
+            idx += 1
+        if notes is not None:
+            sets.append(f"notes = ${idx}")
+            params.append(notes)
+            idx += 1
+        sql = (
+            f"UPDATE missing_precedents SET {', '.join(sets)} "
+            f"WHERE id = $1 RETURNING *"
+        )
+        row = await conn.fetchrow(sql, *params)
+    return _row_to_missing_precedent(row) if row else None
+
+
+async def find_missing_precedent_by_citation(
+    citation: str,
+    case_id: UUID | None = None,
+) -> dict | None:
+    """Look up an existing row by citation string (exact match) and optionally
+    cited-in case_id. Used to deduplicate auto-creation by the researcher."""
+    pool = await get_pool()
+    async with pool.acquire() as conn:
+        if case_id is not None:
+            row = await conn.fetchrow(
+                "SELECT * FROM missing_precedents "
+                "WHERE citation = $1 AND cited_in_case_id = $2 LIMIT 1",
+                citation.strip(), case_id,
+            )
+        else:
+            row = await conn.fetchrow(
+                "SELECT * FROM missing_precedents WHERE citation = $1 LIMIT 1",
+                citation.strip(),
+            )
+    return _row_to_missing_precedent(row) if row else None