feat(analysis): analyze_protocol — comparative hearing-protocol analysis (WS4 #203)

מנתח פרוטוקול-דיון מול כתבי-הטענות: אילו טענות ירדו/חוזקו/עלו-חדשות + חידוד
שאלות משפטיות, ומחלץ נתוני כותרת (א–ד). התוצאה נכנסת לידע-התיק.

- protocol_analyzer.py — ניתוח השוואתי דרך claude_session (Opus 4.8, effort=high,
  מקומי בלבד); שער anti-hallucination (quote-or-retract) ב-_normalize_change;
  חילוץ א–ד + הזנת hearing_date חזרה ל-cases (לא דורס קלט-יו"ר).
- claims_extractor: protocol → claim_type='protocol'; extract_claims קולט protocol.
- block_writer._build_claims_context מחריג claim_type='protocol' מבלוק ז (INV-WR4).
- db: טבלת protocol_analysis (V47, idempotent per-doc) + replace/list helpers.
- tools/drafting + server: analyze_protocol + get_protocol_analysis (extract/get).
- ספ 04-analysis-writing §1.3 + 15 בדיקות.

Invariants: G1 (נרמול-במקור: hearing_date לעמודה הקנונית), G2 (ידע-תיק נגזר
ממקור-האמת protocol+legal_arguments, לא מסלול מקביל), INV-TOOL3 (idempotent
per document_id), INV-TOOL4 (extract/get symmetry), INV-AH (quote-or-retract),
INV-WR4 (בלוק ז = טענות-כתב מקוריות בלבד), claude_session local-only, G12 (נקי
מ-Paperclip — leak-guard עובר).

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

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

@@ -1782,6 +1782,41 @@ CREATE INDEX IF NOT EXISTS idx_decision_lessons_vec
     ON decision_lessons USING ivfflat (embedding vector_cosine_ops) WITH (lists = 30);
 """
 
+# ── V47: Protocol comparative analysis (WS4 / #203) ───────────────
+#
+# protocol_analysis: case-knowledge derived from a hearing-protocol document by
+# comparing the oral arguments raised at the hearing against the written
+# pleadings (legal_arguments). One row per (case_id, document_id) — idempotent
+# replace on re-run (INV-TOOL3). Each row holds the comparative verdict on a
+# single argument: was it DROPPED at the hearing, STRENGTHENED, or NEWLY_RAISED,
+# plus the sharpened legal question it bears on. This is *derived* knowledge —
+# the source of truth is the protocol document + legal_arguments; the row is a
+# materialized comparison, re-buildable from those (G2: no parallel store, this
+# is derived-from-source not a competing claims table).
+#
+# header_data (the א–ד feed: panel, hearing date, parties present) is written
+# back to the canonical `cases` columns (panel via decisions, hearing_date on
+# cases) — NOT duplicated here — and a copy of the raw extracted feed is kept on
+# the analysis row for provenance only (G9).
+SCHEMA_V47_SQL = """
+CREATE TABLE IF NOT EXISTS protocol_analysis (
+    id UUID PRIMARY KEY DEFAULT uuid_generate_v4(),
+    case_id UUID NOT NULL REFERENCES cases(id) ON DELETE CASCADE,
+    document_id UUID NOT NULL REFERENCES documents(id) ON DELETE CASCADE,
+    party_role TEXT NOT NULL DEFAULT '',          -- appellant/respondent/committee/permit_applicant/''
+    change_type TEXT NOT NULL CHECK (change_type IN ('dropped', 'strengthened', 'newly_raised')),
+    argument_id UUID REFERENCES legal_arguments(id) ON DELETE SET NULL,  -- the pleaded argument, when matched
+    argument_title TEXT NOT NULL DEFAULT '',      -- snapshot of the argument / new point
+    summary TEXT NOT NULL,                        -- what changed at the hearing, in prose
+    sharpened_question TEXT NOT NULL DEFAULT '',  -- the legal question this sharpens (for the discussion)
+    evidence_quote TEXT NOT NULL DEFAULT '',      -- verbatim protocol excerpt (INV-AH: quote-or-retract)
+    page_number INTEGER,
+    created_at TIMESTAMPTZ DEFAULT now()
+);
+CREATE INDEX IF NOT EXISTS idx_protocol_analysis_case ON protocol_analysis(case_id, change_type);
+CREATE INDEX IF NOT EXISTS idx_protocol_analysis_doc ON protocol_analysis(document_id);
+"""
+
 
 # Stable, arbitrary key for the session-level advisory lock that serialises
 # schema DDL across processes. Every short-lived process (cron drains, services)
@@ -1851,6 +1886,7 @@ async def _apply_schema_ddl(conn: asyncpg.Connection) -> None:
         await conn.execute(SCHEMA_V44_SQL)
         await conn.execute(SCHEMA_V45_SQL)
         await conn.execute(SCHEMA_V46_SQL)
+        await conn.execute(SCHEMA_V47_SQL)
 
 
 async def init_schema() -> None:
@@ -3967,6 +4003,83 @@ async def detect_appraiser_conflicts(case_id: UUID) -> list[dict]:
     return conflicts
 
 
+# ── Protocol comparative analysis (V47 / WS4 #203) ────────────────
+
+async def replace_protocol_analysis(
+    case_id: UUID,
+    document_id: UUID,
+    rows: list[dict],
+) -> int:
+    """Replace all protocol_analysis rows for a given protocol document (idempotent).
+
+    Each row dict: change_type ('dropped'|'strengthened'|'newly_raised'),
+    party_role, argument_id (UUID|None), argument_title, summary,
+    sharpened_question, evidence_quote, page_number (optional).
+
+    Idempotent on document_id (INV-TOOL3): re-running analyze_protocol replaces
+    the prior verdict for that protocol rather than appending duplicates.
+    """
+    pool = await get_pool()
+    async with pool.acquire() as conn:
+        async with conn.transaction():
+            await conn.execute(
+                "DELETE FROM protocol_analysis WHERE document_id = $1", document_id,
+            )
+            for r in rows:
+                await conn.execute(
+                    """INSERT INTO protocol_analysis
+                       (case_id, document_id, party_role, change_type, argument_id,
+                        argument_title, summary, sharpened_question, evidence_quote, page_number)
+                       VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10)""",
+                    case_id, document_id,
+                    r.get("party_role", "") or "",
+                    r["change_type"],
+                    r.get("argument_id"),
+                    r.get("argument_title", "") or "",
+                    r["summary"],
+                    r.get("sharpened_question", "") or "",
+                    r.get("evidence_quote", "") or "",
+                    r.get("page_number"),
+                )
+    return len(rows)
+
+
+async def list_protocol_analysis(
+    case_id: UUID,
+    change_type: str | None = None,
+) -> list[dict]:
+    """List protocol_analysis rows for a case, optionally filtered by change_type.
+
+    The read side of the extract/get symmetry (INV-TOOL4) for analyze_protocol.
+    """
+    pool = await get_pool()
+    async with pool.acquire() as conn:
+        if change_type:
+            rows = await conn.fetch(
+                """SELECT * FROM protocol_analysis
+                   WHERE case_id = $1 AND change_type = $2
+                   ORDER BY change_type, party_role, created_at""",
+                case_id, change_type,
+            )
+        else:
+            rows = await conn.fetch(
+                """SELECT * FROM protocol_analysis
+                   WHERE case_id = $1
+                   ORDER BY change_type, party_role, created_at""",
+                case_id,
+            )
+    results = []
+    for r in rows:
+        d = dict(r)
+        d["id"] = str(d["id"])
+        d["case_id"] = str(d["case_id"])
+        d["document_id"] = str(d["document_id"])
+        if d.get("argument_id") is not None:
+            d["argument_id"] = str(d["argument_id"])
+        results.append(d)
+    return results
+
+
 # ── Plans registry (V38) ──────────────────────────────────────────
 # Canonical registry of planning schemes (תכניות). SSOT for a plan's identity +
 # validity, reused across cases (G2). See SCHEMA_V38_SQL for the data contract.