legal-ai/mcp-server/src/legal_mcp/services/protocol_analyzer.py

"""ניתוח פרוטוקול-דיון השוואתי (WS4 / #203).

מנתח פרוטוקול דיון של ועדת הערר מול כתבי-הטענות הכתובים: אילו טענות **ירדו**
(נזנחו בדיון), אילו **חוזקו**, ואילו **עלו חדשות** — ומחדד את השאלות המשפטיות
לקראת פרק הדיון. התוצאה נכנסת ל"ידע-התיק" (טבלת protocol_analysis), זמינה
לסוכני הניתוח והכתיבה.

בנוסף, מחלץ את **נתוני הכותרת (א–ד)** מהפרוטוקול בפורמט-ההזנה המוכר —
הרכב הוועדה, תאריך הדיון, והצדדים שהופיעו — ומזין אותם חזרה לעמודות הקנוניות
(`cases.hearing_date`, panel דרך decisions; G2: לא כפילות).

הפרדת-אחריות (claude_session.py): כל קריאת-LLM כאן רצה רק מה-MCP server המקומי
(אין claude CLI בקונטיינר). הקריאה היחידה היא דרך claude_session, מעוגנת ל-
Opus 4.8 + effort=high כמתבקש בניתוח-השוואתי (reference_claude_generation_path).
"""

from __future__ import annotations

import json
import logging
from uuid import UUID

from legal_mcp.services import claude_session, db

logger = logging.getLogger(__name__)

# Generation pinned per reference_claude_generation_path: Opus 4.8 rejects
# temperature/top_p/top_k (400) — the only knob is effort. Comparative legal
# reasoning over the protocol vs. pleadings is high-effort.
ANALYSIS_MODEL = "claude-opus-4-8"
ANALYSIS_EFFORT = "high"
HEADER_EFFORT = "medium"  # א–ד is mechanical extraction, not reasoning

# Valid change verdicts — mirror the DB CHECK on protocol_analysis.change_type.
VALID_CHANGE_TYPES = {"dropped", "strengthened", "newly_raised"}
VALID_PARTY_ROLES = {"appellant", "respondent", "committee", "permit_applicant", ""}

PARTY_LABELS_HE = {
    "appellant": "עוררים",
    "respondent": "משיבים",
    "committee": "ועדה מקומית",
    "permit_applicant": "מבקשי היתר",
}


# ── Comparative analysis: pleadings vs. protocol ──────────────────

COMPARE_PROMPT = """אתה מנתח משפטי בכיר בועדת ערר לתכנון ובנייה. לפניך **פרוטוקול דיון**
ולצדו **הטיעונים המשפטיים שהוגשו בכתב** לפני הדיון. תפקידך: ניתוח השוואתי מדויק
של מה שקרה בדיון ביחס לכתבי-הטענות.

## מה לזהות (לכל צד בנפרד):
1. **dropped (ירדה)** — טענה שהוגשה בכתב אך הצד **זנח** אותה בדיון: לא חזר עליה,
   הודה שאינה עומדת, או ויתר עליה במפורש.
2. **strengthened (חוזקה)** — טענה כתובה שהצד **חיזק** בדיון: הוסיף נימוק, אסמכתא,
   הבהרה, או דגש שהפך אותה למרכזית.
3. **newly_raised (חדשה)** — טענה/סוגיה ש**עלתה לראשונה בדיון** ולא הופיעה בכתבי
   הטענות כלל.

## כללים קריטיים (anti-hallucination — חובה):
- **ציטוט-או-הימנעות:** לכל קביעה חייב להיות `evidence_quote` — ציטוט **מילולי**
  מהפרוטוקול (עד 200 תווים) שמבסס אותה. אם אין ציטוט תומך — אל תכלול את הרשומה.
- אל תמציא טענות שלא נאמרו. אל תסיק "ירדה" רק כי לא הוזכרה — רק אם יש ראיה פוזיטיבית
  לזניחה/ויתור, או שהיא טענה כתובה מרכזית שכלל לא עלתה והצד עסק בנושא.
- **חידוד-השאלה:** לכל רשומה, נסח את `sharpened_question` — השאלה המשפטית הממוקדת
  שהשינוי בדיון מחדד (לטובת פרק הדיון). אם לא רלוונטי — השאר ריק.
- שייך כל רשומה ל-`party_role` הנכון: appellant / respondent / committee / permit_applicant.
- אם רשומה תואמת טיעון כתוב קיים — החזר את ה-`argument_id` שלו מהרשימה למטה. לטענה חדשה
  (newly_raised) — argument_id ריק.

## פלט:
החזר JSON array בלבד — ללא markdown, ללא הסברים:
[
  {{
    "change_type": "dropped" | "strengthened" | "newly_raised",
    "party_role": "appellant" | "respondent" | "committee" | "permit_applicant",
    "argument_id": "uuid-של-הטיעון-הכתוב או null",
    "argument_title": "כותרת קצרה של הטענה/הסוגיה",
    "summary": "מה השתנה בדיון, במשפט-שניים",
    "sharpened_question": "השאלה המשפטית שהשינוי מחדד, או ריק",
    "evidence_quote": "ציטוט מילולי מהפרוטוקול",
    "page_number": null
  }}
]
אם אין שינויים בני-ביסוס — החזר [].

## הטיעונים שהוגשו בכתב:
{arguments_json}

## פרוטוקול הדיון:
--- תחילת פרוטוקול ---
{protocol_text}
--- סוף פרוטוקול ---
"""


# ── Header (א–ד) extraction in the known feed format ──────────────

HEADER_PROMPT = """אתה מחלץ נתוני-כותרת מפרוטוקול דיון של ועדת ערר לתכנון ובנייה.
חלץ אך-ורק עובדות מנהליות המופיעות במפורש בפרוטוקול — אל תמציא ואל תסיק.

## פלט:
החזר JSON object בלבד (ללא markdown):
{{
  "hearing_date": "YYYY-MM-DD אם תאריך הדיון מופיע, אחרת ריק",
  "panel_members": ["שמות חברי ההרכב כפי שמופיעים, כולל תוארם; ריק אם לא צוין"],
  "appellants_present": ["שמות העוררים/באי-כוחם שהופיעו בדיון"],
  "respondents_present": ["שמות המשיבים/באי-כוחם שהופיעו בדיון"]
}}
שדה שלא צוין בפרוטוקול — החזר ריק ([] או "").

## פרוטוקול:
--- תחילת פרוטוקול ---
{protocol_text}
--- סוף פרוטוקול ---
"""

# A single protocol rarely exceeds the model's context, but cap defensively so a
# pathological OCR dump doesn't blow the prompt budget.
MAX_PROTOCOL_CHARS = 120_000


def _find_protocol(docs: list[dict]) -> dict | None:
    """The protocol document for the case, if present (doc_type or title)."""
    for d in docs:
        if d.get("doc_type") == "protocol":
            return d
    for d in docs:
        if "פרוטוקול" in (d.get("title") or ""):
            return d
    return None


def _compact_arguments(arguments: list[dict]) -> list[dict]:
    """Strip aggregated arguments down to what the comparison needs."""
    out = []
    for a in arguments:
        out.append({
            "argument_id": str(a["id"]),
            "party_role": a.get("party", ""),
            "title": a.get("argument_title", ""),
            "body": a.get("argument_body", ""),
            "topic": a.get("legal_topic", ""),
        })
    return out


def _normalize_change(raw: dict, valid_argument_ids: set[str]) -> dict | None:
    """Validate & normalize one comparative-analysis row from Claude.

    Returns None for unusable rows (missing change_type/summary, or — per the
    anti-hallucination gate — no supporting evidence_quote).
    """
    if not isinstance(raw, dict):
        return None
    change_type = (raw.get("change_type") or "").strip()
    if change_type not in VALID_CHANGE_TYPES:
        return None
    summary = (raw.get("summary") or "").strip()
    evidence_quote = (raw.get("evidence_quote") or "").strip()
    # INV-AH (quote-or-retract): drop any verdict the model couldn't anchor to a
    # verbatim protocol excerpt — at source, so unfounded rows never reach the DB.
    if not summary or not evidence_quote:
        return None

    party_role = (raw.get("party_role") or "").strip()
    if party_role not in VALID_PARTY_ROLES:
        party_role = ""

    # argument_id only kept when it actually belongs to this case's arguments;
    # a newly_raised point has none, and a hallucinated id is dropped (FK safety).
    argument_id = None
    raw_aid = raw.get("argument_id")
    if raw_aid and str(raw_aid) in valid_argument_ids:
        try:
            argument_id = UUID(str(raw_aid))
        except (ValueError, TypeError):
            argument_id = None

    page = raw.get("page_number")
    if not isinstance(page, int):
        page = None

    return {
        "change_type": change_type,
        "party_role": party_role,
        "argument_id": argument_id,
        "argument_title": (raw.get("argument_title") or "").strip(),
        "summary": summary,
        "sharpened_question": (raw.get("sharpened_question") or "").strip(),
        "evidence_quote": evidence_quote[:200],
        "page_number": page,
    }


async def _extract_header(protocol_text: str, case_id: UUID) -> dict:
    """Extract א–ד header data and write it back to canonical case fields.

    Returns the raw extracted feed (for provenance) plus a record of what was
    written back. Header data is NOT stored in protocol_analysis — it lives in
    the canonical `cases`/`decisions` columns (G2: single source of truth).
    """
    prompt = HEADER_PROMPT.format(protocol_text=protocol_text[:MAX_PROTOCOL_CHARS])
    try:
        feed = await claude_session.query_json(
            prompt, model=ANALYSIS_MODEL, effort=HEADER_EFFORT, tools="",
        )
    except Exception as e:  # noqa: BLE001 — surface, don't swallow (§6 no silent swallow)
        logger.warning("protocol header extraction failed for case %s: %s", case_id, e)
        return {"status": "header_failed", "error": str(e)}

    if not isinstance(feed, dict):
        return {"status": "header_no_data"}

    applied: dict = {}
    # hearing_date → canonical cases.hearing_date (only if the case lacks one,
    # so a chair-entered date is never overwritten by extraction).
    hearing_date = (feed.get("hearing_date") or "").strip()
    if hearing_date:
        from datetime import date as date_type
        try:
            parsed = date_type.fromisoformat(hearing_date)
            case = await db.get_case(case_id)
            if case and not case.get("hearing_date"):
                await db.update_case(case_id, hearing_date=parsed)
                applied["hearing_date"] = hearing_date
        except ValueError:
            logger.info("protocol header: unparseable hearing_date %r", hearing_date)

    return {
        "status": "ok",
        "feed": {
            "hearing_date": hearing_date,
            "panel_members": feed.get("panel_members") or [],
            "appellants_present": feed.get("appellants_present") or [],
            "respondents_present": feed.get("respondents_present") or [],
        },
        "applied_to_case": applied,
    }


async def analyze_protocol(case_id: UUID) -> dict:
    """Comparative analysis of the case's hearing protocol vs. its pleadings.

    1. Locates the protocol document.
    2. Pulls the aggregated legal_arguments (the written-pleadings baseline).
    3. Asks Claude (Opus 4.8, effort=high) to classify each as dropped /
       strengthened, and to surface newly-raised arguments + sharpened questions.
    4. Stores the verdicts in protocol_analysis (case-knowledge, idempotent).
    5. Extracts the א–ד header feed and writes hearing_date back to the case.

    Returns a serializable summary dict.
    """
    docs = await db.list_documents(case_id)
    protocol = _find_protocol(docs)
    if not protocol:
        return {
            "status": "no_protocol",
            "message": "לא נמצא פרוטוקול דיון בתיק (doc_type='protocol'). העלה פרוטוקול והרץ שוב.",
        }

    document_id = UUID(protocol["id"])
    protocol_text = await db.get_document_text(document_id)
    if not protocol_text or not protocol_text.strip():
        return {
            "status": "no_text",
            "message": "לפרוטוקול אין טקסט מחולץ. ודא שהעיבוד הסתיים והרץ שוב.",
            "document_id": str(document_id),
        }

    # Use the aggregator's read path (single source of truth — no parallel
    # query for the same legal_arguments data, G2).
    from legal_mcp.services import argument_aggregator
    arguments = await argument_aggregator.get_legal_arguments(case_id)

    if not arguments:
        return {
            "status": "no_arguments",
            "message": (
                "אין טיעונים מאוגדים להשוואה. הרץ extract_claims + "
                "aggregate_claims_to_arguments על כתבי-הטענות תחילה."
            ),
            "document_id": str(document_id),
        }

    compact_args = _compact_arguments(arguments)
    valid_argument_ids = {a["argument_id"] for a in compact_args}

    prompt = COMPARE_PROMPT.format(
        arguments_json=json.dumps(compact_args, ensure_ascii=False, indent=2),
        protocol_text=protocol_text[:MAX_PROTOCOL_CHARS],
    )

    try:
        raw_result = await claude_session.query_json(
            prompt, model=ANALYSIS_MODEL, effort=ANALYSIS_EFFORT, tools="",
        )
    except RuntimeError as e:
        msg = str(e)
        if "Claude CLI not found" in msg:
            return {
                "status": "llm_unavailable",
                "message": (
                    "Claude CLI לא זמין. הניתוח ההשוואתי חייב לרוץ מה-MCP server "
                    "המקומי, לא מהקונטיינר."
                ),
                "document_id": str(document_id),
            }
        return {"status": "error", "message": msg, "document_id": str(document_id)}

    if not isinstance(raw_result, list):
        logger.warning(
            "analyze_protocol: Claude returned non-list (%s) for case %s",
            type(raw_result).__name__, case_id,
        )
        raw_result = []

    rows: list[dict] = []
    for entry in raw_result:
        norm = _normalize_change(entry, valid_argument_ids)
        if norm:
            rows.append(norm)

    stored = await db.replace_protocol_analysis(case_id, document_id, rows)

    # Extract header (א–ד) and write hearing_date back to the canonical case.
    header = await _extract_header(protocol_text, case_id)

    by_change: dict[str, int] = {}
    for r in rows:
        by_change[r["change_type"]] = by_change.get(r["change_type"], 0) + 1

    return {
        "status": "completed",
        "document_id": str(document_id),
        "protocol_title": protocol.get("title", ""),
        "total": stored,
        "by_change": by_change,
        "arguments_compared": len(compact_args),
        "header": header,
    }