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 _protocol_scope(doc: dict) -> str:
    """Protocol scope from ``metadata.protocol_scope``.

    Absent/'' defaults to ``'appeal'`` (ועדת הערר), matching the
    document-type-editor convention where only the non-default ``'lower'``
    (ועדה מקומית/מחוזית) is persisted.
    """
    meta = doc.get("metadata") or {}
    return (meta.get("protocol_scope") or "").strip() or "appeal"


def _find_protocol(
    docs: list[dict], document_id: UUID | None = None,
) -> dict | None:
    """The ועדת-הערר hearing protocol to compare against the pleadings.

    - ``document_id`` given → return exactly that document (explicit target),
      so a case with several protocols can be pointed at the right hearing.
    - Otherwise prefer a protocol scoped to the appeals committee
      (``scope != 'lower'``). A ``'lower'`` protocol is the local/district
      committee's proceedings — it feeds background (block ו) only and must NOT
      drive the hearing-vs-pleadings comparison (#223). If every protocol is
      lower-scoped there is no ערר-hearing to compare → return None.
    - Among appeal-scoped protocols the most recent (by ``created_at``) wins —
      the operative hearing; ties fall back to list order.
    """
    if document_id is not None:
        return next(
            (d for d in docs if str(d.get("id")) == str(document_id)), None,
        )

    protocols = [d for d in docs if d.get("doc_type") == "protocol"]
    if not protocols:
        # Legacy fallback: untyped docs whose title says "פרוטוקול".
        protocols = [d for d in docs if "פרוטוקול" in (d.get("title") or "")]

    appeal_scoped = [d for d in protocols if _protocol_scope(d) != "lower"]
    if not appeal_scoped:
        return None
    appeal_scoped.sort(key=lambda d: d.get("created_at") or "", reverse=True)
    return appeal_scoped[0]


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, target_document_id: UUID | None = None,
) -> dict:
    """Comparative analysis of the case's hearing protocol vs. its pleadings.

    ``target_document_id`` pins the analysis to a specific protocol document —
    required when a case holds several protocols (e.g. a lower-committee annex
    plus the ועדת-הערר hearing) and the auto-pick would be ambiguous (#223).

    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, document_id=target_document_id)
    if not protocol:
        if target_document_id is not None:
            return {
                "status": "no_protocol",
                "message": f"מסמך {target_document_id} לא נמצא בתיק.",
            }
        return {
            "status": "no_protocol",
            "message": (
                "לא נמצא פרוטוקול ועדת-ערר בתיק (doc_type='protocol' עם "
                "protocol_scope שאינו 'lower'). העלה פרוטוקול-דיון, או תייג את "
                "הפרוטוקול הקיים כ-appeal, והרץ שוב."
            ),
        }

    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,
    }