6 Commits

Author SHA1 Message Date
1e43c12418 feat(analysis): דגל "לא-נותח" + ניתוח-מחדש מאחד + בדיקת-השפעה (#201)
All checks were successful
G12 Leak-Guard / leak-guard (pull_request) Successful in 3s
Lint — undefined names / undefined-names (pull_request) Successful in 10s
WS2b של עיצוב-מחדש זרימת-העבודה. מאפשר ניתוח כתב-הערר לבד ואז איחוד
מסמך-עיקרי שנוסף מאוחר, בלי force-delete גורף, עם diff ליו"ר.

- SCHEMA_V49: documents.claims_extracted_at + claims_extraction_status
  (אירוע-חילוץ per-מסמך, לא נגזר מ-doc_type), + אינדקס חלקי
  idx_documents_claims_pending. idempotent (ADD COLUMN IF NOT EXISTS).
  V48 שמור ל-#357.
- claims_extractor חותם את המסמך אחרי שמירת/אי-מציאת טענות.
- db.primary_docs_not_analyzed (is_primary V47 + claims_extracted_at IS NULL)
  + mark_document_claims_extracted; _row_to_doc חושף claims_analyzed.
- reanalyze_claims (כלי-MCP): snapshot→חילוץ-מאחד רק למסמכים חדשים/לא-נותחו
  (store_claims מחליף per-source ⇒ טענות אחרות נשמרות)→aggregate force=True
  (מסלול קנוני, מוחק רק legal_arguments)→_impact_diff before↔after ליו"ר.
- workflow_status חושף primary_docs_not_analyzed + next-step.
- ספ: 02-data-model §2ג (דגל לא-נותח), 04-analysis-writing §1.3.
- 5 בדיקות-יחידה ל-_impact_diff/_snapshot (פונקציות טהורות).

Invariants:
- G1 (נרמול-במקור): claims_extracted_at נחתם בנקודת-החילוץ, claims_analyzed
  נגזר ממנו — לא תיקון-בקריאה.
- G2 (מסלול קנוני יחיד / merge-not-fork): reanalyze מרחיב את מסלול
  claims_extractor+argument_aggregator הקיים, לא forks; האיחוד דרך
  store_claims per-source; aggregate force=True מוחק רק נגזר (legal_arguments).
- INV-DM (מודל-נתונים): דגל-אירוע per-מסמך, אינדקס חלקי, מקור-אמת יחיד.
- G10 (שער-אנושי): בדיקת-ההשפעה מוצגת ליו"ר, לא מוחלת אוטומטית.
- INV-TOOL idempotency: SCHEMA_V49 idempotent; ניתוח-מחדש חוזר על מסמך
  כבר-נותח הוא refresh נקי (store_methods replace per-source).

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-30 12:18:07 +00:00
eb0653fb80 Merge pull request 'feat(calibration): כיול-אמפירי model×effort מול הסופיים (#208)' (#360) from worktree-agent-a5a22be0318670871 into main
All checks were successful
Build & Deploy / build-and-deploy (push) Successful in 1m27s
G12 Leak-Guard / leak-guard (push) Successful in 5s
Lint — undefined names / undefined-names (push) Successful in 10s
2026-06-30 12:15:46 +00:00
f935f166a9 feat(calibration): כיול-אמפירי model×effort מול הסופיים (#208)
All checks were successful
G12 Leak-Guard / leak-guard (pull_request) Successful in 3s
Lint — undefined names / undefined-names (pull_request) Successful in 10s
A/B harness שמכייל את ה-effort הנעוץ per-בלוק (#204) מול הסופיים של דפנה:
מייצר מחדש כל בלוק דרך מסלול-הייצור (write_block(effort_override=…) →
claude_session.query → claude -p, Opus 4.8, מקומי-בלבד) ומודד מול הסקשן
המתאים בסופי דרך style_distance.block_distance_to_final (change_percent,
anti_pattern_total, golden-ratio deviation, composite distance). ממליץ
per-בלוק על ה-effort הקרוב-ביותר לסופי.

- scripts/calibrate_effort.py — ההארנס (מודל eval_retrieval.py): --self-test
  (offline, מוכיח מדידה+המלצה, אפס DB/CLI) · --dry-run · --efforts/--blocks/
  --case/--repeats. דוח data/eval/effort-calibration-<ts>.{json,md} עם
  גודל-מדגם בולט — עדות-כיוון, לא רגרסיה (מעט סופיים-עלויים).
- style_distance.py — block_distance_to_final + split_final_by_section
  (מקור-מדידה יחיד, G2; reuse compute_diff_stats/count_anti_patterns/chunker).
- block_writer.py — write_block(effort_override=) להזרקת effort per-קריאה
  בלי לדרוס את ברירות-המחדל הנעוצות; רושם את ה-effort האפקטיבי.
- scripts/SCRIPTS.md — ערך חדש.

Invariants: G8 (eval-harness — מדידה אמפירית, לא הנחה) · G2 (reuse של
style_distance/learning_loop — אין מסלול-מדד מקביל) · claude_session
local-only (reference_claude_generation_path) · INV-LRN4/5 (השוואה מול
הסופי, מדידת-סגנון; אין מהות-תיק נגררת). אומת: --self-test 14/14 PASS.
הריצה החיה host-only (claude CLI) — לא ניתנת-להרצה ב-worktree/קונטיינר.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-30 12:13:43 +00:00
33b36e38ee Merge pull request 'feat(analysis): analyze_protocol — ניתוח פרוטוקול-דיון השוואתי (WS4 #203)' (#357) from worktree-agent-aefd5594d532d9f06 into main
All checks were successful
Build & Deploy / build-and-deploy (push) Successful in 1m32s
G12 Leak-Guard / leak-guard (push) Successful in 4s
Lint — undefined names / undefined-names (push) Successful in 10s
2026-06-30 12:12:05 +00:00
8d704e6e40 Merge remote-tracking branch 'origin/main' into rebase-357
All checks were successful
G12 Leak-Guard / leak-guard (pull_request) Successful in 4s
Lint — undefined names / undefined-names (pull_request) Successful in 10s
# Conflicts:
#	mcp-server/src/legal_mcp/services/db.py
2026-06-30 12:09:49 +00:00
24e3e2fe80 feat(analysis): analyze_protocol — comparative hearing-protocol analysis (WS4 #203)
All checks were successful
G12 Leak-Guard / leak-guard (pull_request) Successful in 4s
Lint — undefined names / undefined-names (pull_request) Successful in 10s
מנתח פרוטוקול-דיון מול כתבי-הטענות: אילו טענות ירדו/חוזקו/עלו-חדשות + חידוד
שאלות משפטיות, ומחלץ נתוני כותרת (א–ד). התוצאה נכנסת לידע-התיק.

- 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>
2026-06-30 11:21:13 +00:00
16 changed files with 1630 additions and 18 deletions

View File

@@ -95,6 +95,19 @@ WS2 (#200) מבחין בין **מסמך-עיקרי** (מסמך-מהות שהיו
`document_list` / `case_get` / API-המסמכים. אינדקס חלקי `idx_documents_primary(case_id) WHERE
is_primary` משרת את שאילתת "מסמכים-עיקריים שטרם-נותחו" (#201).
**דגל "לא-נותח" — per-מסמך (#201, SCHEMA_V49).** שני שדות מתעדים את *אירוע* חילוץ-הטענות
של המסמך (לא נגזרים מ-`doc_type`): `claims_extracted_at TIMESTAMPTZ` (נחתם ע"י
`claims_extractor` **אחרי** ששמרנו את טענות המסמך) ו-`claims_extraction_status TEXT`
(`pending`/`completed`/`no_claims`/`failed`). `_row_to_doc` חושף `claims_analyzed`
(`claims_extracted_at IS NOT NULL`). מקור-האמת היחיד ל"האם המסמך נכלל בניתוח" הוא
`claims_extracted_at` — לא משוחזר ממקום אחר. אינדקס חלקי `idx_documents_claims_pending(case_id)
WHERE is_primary AND claims_extracted_at IS NULL` (יחד עם `is_primary` של V47) הופך את
"מסמך-עיקרי שנוסף-אחרי-ניתוח ולא-נכלל" (`db.primary_docs_not_analyzed`) לשאילתה זולה.
נחשף ב-`workflow_status` (`primary_docs_not_analyzed` + next-step). הניתוח-מחדש המאחד
(`reanalyze_claims`, [04-analysis-writing §1.3](04-analysis-writing.md)) מחלץ-מחדש רק את
המסמכים-העיקריים החדשים/לא-נותחו, **מאחד** עם הטענות הקיימות, ומריץ צבירה-מחדש (G2 — מסלול
קנוני אחד, לא fork).
### 2ד. ישויות-נגזרות (אחסון-ניתוחים)
מעבר לישויות-המקור, המערכת **שומרת ניתוחים נגזרים** — תוצרי-חילוץ של LLM/קוד. אלו כפופים לכללי

View File

@@ -47,6 +47,54 @@
> מקוריות מהשלמות). הסינון לפי `party_role` מאפשר לזהות את הצד המפסיד ולוודא שכל טיעון
> שלו מקבל מענה בבלוק י.
### 1.3 ניתוח-פרוטוקול — ניתוח פרוטוקול-דיון השוואתי (WS4 / #203)
> **בעלוּת תת-סעיף:** §1.3 (זה) שייך ל-WS4 (ניתוח-פרוטוקול). תת-סעיף סיכום-טענות-הצדדים
> (PR-אחות #358) הוא **תת-סעיף נפרד** תחת §1 — אין חפיפת-מספור בין השניים.
אחרי הדיון, פרוטוקול (`doc_type='protocol'`) מנותח **מול הטיעונים המאוגדים**
`analyze_protocol(case_number)` (→ `services/protocol_analyzer.py`). הניתוח מסווג כל
טיעון כתוב כ**ירד** (`dropped` — נזנח/ויתר בדיון), **חוזק** (`strengthened`), או מזהה
טענה ש**עלתה לראשונה בדיון** (`newly_raised`), ולכל רשומה מנסח את ה**שאלה המשפטית
המחודדת** לטובת בלוק י. התוצאה נשמרת ב**ידע-התיק** — טבלת `protocol_analysis` (case-
knowledge נגזר; G2: מקור-האמת הוא הפרוטוקול + `legal_arguments`, הרשומה היא השוואה
מטוריאליזת בת-שחזור) — וזמינה ל-`get_protocol_analysis(case_number, change_type="")`
(סימטריית extract/get, [INV-TOOL4](X9-mcp-tool-contract.md)).
- **קלט:** הטיעונים המאוגדים (§1.2) הם קו-הבסיס; ללא כינוס אין מול-מה להשוות.
- **שער anti-hallucination ([INV-AH](../anti-hallucination-gate.md)):** כל רשומה חייבת
`evidence_quote` — ציטוט **מילולי** מהפרוטוקול. רשומה ללא ציטוט-מבסס נדחית **במקור**
(`_normalize_change`) ולא מגיעה ל-DB (quote-or-retract).
- **טענות-הדיון בנפרד מבלוק ז:** טענות שעלו בדיון מתויגות `claim_type='protocol'`
ב-`extract_claims` ו**מוחרגות** מהקשר בלוק ז (`block_writer._build_claims_context`) —
בלוק ז נשאר **טענות-כתב מקוריות בלבד** ([INV-WR4](#inv-wr4-בלוק-ז--טענות-מקוריות-בלבד));
טענות-הדיון שייכות לבלוק ח (הליכים) ולידע-התיק ההשוואתי.
- **נתוני כותרת (א–ד):** הניתוח מחלץ גם את הפיד המוכר (הרכב, תאריך-דיון, צדדים שהופיעו)
ומזין את `hearing_date` חזרה לעמודה הקנונית `cases.hearing_date` (G1; לא נכתב אם
היו"ר כבר מילא תאריך — לא דורסים קלט-יו"ר).
- **ייצור:** קריאת-ה-LLM ההשוואתית עוברת `claude_session` (מקומי בלבד), מעוגנת
`model="claude-opus-4-8"` + `effort="high"` (ראה `reference_claude_generation_path`).
### 1.4 ניתוח-מחדש מאחד אחרי מסמך-עיקרי חדש (#201)
תיק מנותח לרוב מכתב-הערר לבד, ומסמך-עיקרי (תשובה/התנגדות/פרוטוקול…) מגיע מאוחר יותר.
כל מסמך נושא **דגל "לא-נותח"** (`claims_extracted_at`, [02-data-model §2ג](02-data-model.md)),
ו-`workflow_status` מסמן "מסמך-עיקרי שטרם-נכלל בניתוח". `reanalyze_claims(case_number,
reanalyze_all_primary=False)` סוגר את הפער **בלי force-delete גורף**:
1. **snapshot לפני** — הטיעונים המאוגדים הנוכחיים (בסיס בדיקת-ההשפעה).
2. **חילוץ-מאחד** — מחלץ רק את המסמכים-העיקריים החדשים/לא-נותחו (`db.primary_docs_not_analyzed`)
דרך **אותו** `claims_extractor.extract_and_store_claims`; `store_claims` מחליף רק את טענות
*אותו* מסמך (לפי `source_document`), כך שטענות ממסמכים שכבר-נותחו **נשמרות** (האיחוד).
3. **צבירה-מחדש**`aggregate_claims_to_arguments(force=True)` מחשב את הטיעונים מחדש מתוך
מערך-הטענות המאוחד השלם. זהו **מסלול-החישוב הקנוני** ([G2](00-constitution.md#inv-g2-מקור-אמת-יחיד--אין-מסלולים-מקבילים-מתפצלים)),
לא מסלול-עיבוד מקביל; הוא מוחק רק `legal_arguments` (נגזר), לעולם לא `claims`.
4. **בדיקת-השפעה ליו"ר** — snapshot אחרי + `_impact_diff` מחזיר per-צד אילו טיעונים
נוספו/הוסרו ואיך השתנה תמהיל-העדיפויות ("מאזן-ההמלצה"), עם דגל `changed`. הפלט **מוצג**
ליו"ר ולא מוחל אוטומטית ([G10](00-constitution.md#inv-g10-המערכת-מסייעת--שערים-אנושיים-הם-invariant)).
ראה `tools/legal_arguments.py` (`reanalyze_claims`, `_snapshot`, `_impact_diff`).
---
## 2. ארכיטקטורת 12 הבלוקים (סיכום)

View File

@@ -572,6 +572,18 @@ async def get_legal_arguments(
return await la_tools.get_legal_arguments(case_number, party)
@mcp.tool()
async def reanalyze_claims(
case_number: str,
reanalyze_all_primary: bool = False,
) -> str:
"""ניתוח-מחדש מאחד אחרי הוספת מסמך-עיקרי: מחלץ ממסמכים-עיקריים חדשים/לא-נותחו בלבד,
מאחד עם הטענות הקיימות (לא מוחק הכול), מריץ צבירה-מחדש, ומחזיר בדיקת-השפעה
(diff טיעונים/המלצה לפני↔אחרי) ליו"ר. reanalyze_all_primary=True = כל המסמכים-העיקריים.
"""
return await la_tools.reanalyze_claims(case_number, reanalyze_all_primary)
# References
@mcp.tool()
async def extract_references(
@@ -744,6 +756,27 @@ async def get_appraiser_facts(case_number: str) -> str:
return await drafting.get_appraiser_facts(case_number)
# ── Protocol comparative analysis (WS4 / #203) — ניתוח פרוטוקול ────
@mcp.tool()
async def analyze_protocol(case_number: str) -> str:
"""ניתוח השוואתי של פרוטוקול-דיון מול כתבי-הטענות: אילו טענות ירדו/חוזקו/עלו-חדשות + חידוד-שאלות + חילוץ א–ד.
מזין ל"ידע-התיק" (protocol_analysis); דורש פרוטוקול doc_type='protocol' +
טיעונים מאוגדים. Claude מקומי Opus 4.8 effort=high; re-run מחליף (idempotent).
"""
return await drafting.analyze_protocol(case_number)
@mcp.tool()
async def get_protocol_analysis(case_number: str, change_type: str = "") -> str:
"""קריאת ניתוח-הפרוטוקול שכבר חולץ — ללא ניתוח-מחדש. ה-get המקביל ל-analyze_protocol.
change_type: dropped/strengthened/newly_raised (ריק=הכל).
"""
return await drafting.get_protocol_analysis(case_number, change_type)
# ── Planning-schemes registry (V38) — מרשם-התכניות ─────────────────
# SSOT לזהות+תוקף של תכנית, נעשה שימוש חוזר בין תיקים (G2). פלט-LLM נכנס
# pending_review וממתין לאישור-יו"ר (plan_review, G10) לפני שמשמש בבלוק ט.

View File

@@ -355,6 +355,7 @@ async def write_block(
case_id: UUID,
block_id: str,
instructions: str = "",
effort_override: str | None = None,
) -> dict:
"""כתיבת בלוק יחיד בהחלטה.
@@ -362,6 +363,11 @@ async def write_block(
case_id: מזהה התיק
block_id: מזהה הבלוק (block-alef, block-he, block-yod, ...)
instructions: הנחיות נוספות
effort_override: optional per-call reasoning effort (low/medium/high/
xhigh/max). When set, overrides BLOCK_CONFIG[block_id].effort for
THIS call only — used by the #208 model/effort calibration harness
to A/B efforts without mutating the pinned defaults. Production
callers leave it None and get the deterministic per-block effort.
Returns:
dict עם content, word_count, block_id, generation_type
@@ -472,7 +478,7 @@ async def write_block(
# reasoning effort so generation is structurally deterministic — these were
# previously NOT forwarded (the source of inconsistency). model/effort flow
# through claude_session.query → `claude -p --model … --effort …`.
effort = block_cfg.get("effort", DEFAULT_EFFORT)
effort = effort_override or block_cfg.get("effort", DEFAULT_EFFORT)
timeout = claude_session.LONG_TIMEOUT if effort in _LONG_EFFORTS else claude_session.DEFAULT_TIMEOUT
content = await claude_session.query(
prompt,
@@ -485,6 +491,10 @@ async def write_block(
sources = await _collect_block_sources(case_id, block_id)
sources["case_law_ids"] = _precedent_case_law_ids
result = _build_result(block_id, content, block_cfg)
# Record the EFFECTIVE effort (override wins) so the harness can attribute
# the measured distance to the effort that actually produced the text.
if result.get("effort") is not None:
result["effort"] = effort
result["sources"] = sources
return result
@@ -587,11 +597,19 @@ async def _build_claims_context(case_id: UUID) -> str:
return "(לא חולצו טענות)"
# Filter out claims from block-zayin (decision summary) — use only
# claims extracted from original pleadings (appeal, response, etc.)
source_claims = [c for c in claims if c.get("source_document", "") != "block-zayin"]
# claims extracted from original pleadings (appeal, response, etc.).
# Also drop claim_type='protocol' (oral arguments raised at the hearing):
# block ז carries ORIGINAL written pleadings only (INV-WR4); hearing-raised
# arguments belong to block ח (proceedings) and to analyze_protocol's
# comparative case-knowledge, not to the parties'-claims summary.
source_claims = [
c for c in claims
if c.get("source_document", "") != "block-zayin"
and c.get("claim_type", "claim") != "protocol"
]
if not source_claims:
# Fallback to all claims if no source claims exist
source_claims = claims
# Fallback to all non-block-zayin claims if no source claims exist.
source_claims = [c for c in claims if c.get("source_document", "") != "block-zayin"] or claims
lines = []
current_role = ""

View File

@@ -227,8 +227,15 @@ def _infer_claim_type(doc_type: str, source_name: str) -> str:
- 'claim' = from appeal documents (כתב ערר)
- 'response' = from original response documents (כתב תשובה)
- 'reply' = from supplementary responses (תגובה, השלמת טיעון)
- 'protocol' = oral arguments raised at the hearing (פרוטוקול דיון)
"""
name_lower = source_name.lower() if source_name else ""
# A hearing protocol carries oral arguments — tagged distinctly so the
# comparative protocol analysis (analyze_protocol) and block-chet
# (proceedings) can tell them apart from the original written pleadings
# (INV-WR4: block ז stays original-pleadings-only).
if doc_type == "protocol" or "פרוטוקול" in name_lower:
return "protocol"
if doc_type == "appeal" or "כתב ערר" in name_lower:
return "claim"
if "כתב תשובה" in name_lower:
@@ -353,6 +360,10 @@ async def extract_and_store_claims(
claims = await extract_claims_with_ai(text, doc_type, party_hint)
if not claims:
# Stamp the document as analysed even with zero claims, so it is not
# perpetually flagged "not-analysed" (#201 — the not-analysed flag is
# about whether extraction *ran*, not whether it found anything).
await db.mark_document_claims_extracted(document_id, status="no_claims")
return {"status": "no_claims", "total": 0, "source": source_name}
# Determine claim_type from document type and title
@@ -361,6 +372,10 @@ async def extract_and_store_claims(
c["claim_type"] = claim_type
stored = await db.store_claims(case_id, claims, source_document=source_name)
# Mark this document analysed (WS2 / #201). store_claims already replaced
# this document's prior claims by source_document — so re-running is a clean
# per-document refresh that does NOT touch other documents' claims (the merge).
await db.mark_document_claims_extracted(document_id, status="completed")
# Summarize by role
role_counts: dict[str, int] = {}

View File

@@ -1835,6 +1835,58 @@ CREATE INDEX IF NOT EXISTS idx_documents_primary
ON documents(case_id) WHERE is_primary;
"""
# ── V48: 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_V48_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);
"""
# V49 (WS2 / task #201): per-document "claims analysed" tracking — the "not-analysed"
# flag. `claims_extracted_at` is stamped by claims_extractor AFTER a document's claims
# are stored; `claims_extraction_status` records the outcome
# ('pending' | 'completed' | 'no_claims' | 'failed'). These describe the per-document
# extraction *event* (the single source of truth for "has this doc been analysed") —
# NOT derived from doc_type. They let us detect a PRIMARY document (is_primary, V47)
# added after the analysis ran and not yet included → "primary doc not yet analysed".
# The partial index pairs with idx_documents_primary so that query is cheap. Idempotent:
# ADD COLUMN IF NOT EXISTS; re-running is a no-op. (V48 = protocol_analysis, #357.)
SCHEMA_V49_SQL = """
ALTER TABLE documents ADD COLUMN IF NOT EXISTS claims_extracted_at TIMESTAMPTZ;
ALTER TABLE documents ADD COLUMN IF NOT EXISTS claims_extraction_status TEXT
NOT NULL DEFAULT 'pending';
CREATE INDEX IF NOT EXISTS idx_documents_claims_pending
ON documents(case_id) WHERE is_primary AND claims_extracted_at IS NULL;
"""
# Stable, arbitrary key for the session-level advisory lock that serialises
# schema DDL across processes. Every short-lived process (cron drains, services)
@@ -1905,6 +1957,8 @@ async def _apply_schema_ddl(conn: asyncpg.Connection) -> None:
await conn.execute(SCHEMA_V45_SQL)
await conn.execute(SCHEMA_V46_SQL)
await conn.execute(SCHEMA_V47_SQL)
await conn.execute(SCHEMA_V48_SQL)
await conn.execute(SCHEMA_V49_SQL)
async def init_schema() -> None:
@@ -2285,6 +2339,14 @@ def _row_to_doc(row: asyncpg.Record) -> dict:
else is_primary_doc_type(d.get("doc_type"))
d["is_primary"] = is_primary
d["doc_category"] = "primary" if is_primary else "secondary"
# Per-document claims-analysis state (WS2 / #201). `claims_analyzed` is the
# human-facing "has this doc been included in the claims analysis" flag —
# True iff claims extraction has been stamped for it. Single source of truth:
# claims_extracted_at. Defaults keep pre-migration rows well-formed.
extracted_at = d.get("claims_extracted_at")
if "claims_extraction_status" not in d or d.get("claims_extraction_status") is None:
d["claims_extraction_status"] = "pending"
d["claims_analyzed"] = extracted_at is not None
return d
@@ -2334,6 +2396,47 @@ async def get_claims(case_id: UUID, party_role: str | None = None) -> list[dict]
return [dict(r) for r in rows]
# ── Per-document claims-analysis tracking (WS2 / #201) ──────────────
async def mark_document_claims_extracted(
doc_id: UUID, status: str = "completed",
) -> None:
"""Stamp a document as having had its claims extracted (WS2 / #201).
Sets ``claims_extracted_at = now()`` and records the outcome in
``claims_extraction_status`` ('completed' | 'no_claims' | 'failed').
This is the single source of truth for the per-document "not-analysed"
flag — it is written at the point claims are stored, not derived later.
"""
pool = await get_pool()
async with pool.acquire() as conn:
await conn.execute(
"UPDATE documents SET claims_extracted_at = now(), "
"claims_extraction_status = $2 WHERE id = $1",
doc_id, status,
)
async def primary_docs_not_analyzed(case_id: UUID) -> list[dict]:
"""Return PRIMARY documents in a case that have NOT yet been analysed.
"Primary" = ``is_primary`` (the V47 generated column, single source of
truth = PRIMARY_DOC_TYPES). "Not analysed" = ``claims_extracted_at IS NULL``.
Backed by the partial index ``idx_documents_claims_pending``. Surfaces the
"there is a primary document not yet included in the analysis" condition
(#201). Returns the same shape as ``list_documents`` rows.
"""
pool = await get_pool()
async with pool.acquire() as conn:
rows = await conn.fetch(
"SELECT * FROM documents "
"WHERE case_id = $1 AND is_primary AND claims_extracted_at IS NULL "
"ORDER BY created_at",
case_id,
)
return [_row_to_doc(r) for r in rows]
# ── Decisions ──────────────────────────────────────────────────────
async def create_decision(
@@ -4029,6 +4132,83 @@ async def detect_appraiser_conflicts(case_id: UUID) -> list[dict]:
return conflicts
# ── Protocol comparative analysis (V48 / 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.

View File

@@ -0,0 +1,341 @@
"""ניתוח פרוטוקול-דיון השוואתי (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,
}

View File

@@ -15,6 +15,7 @@ import re
from uuid import UUID
from legal_mcp.services import db
from legal_mcp.services.learning_loop import compute_diff_stats
from legal_mcp.services.lessons import ANTI_PATTERNS, GOLDEN_RATIOS, canonical_outcome
logger = logging.getLogger(__name__)
@@ -122,6 +123,93 @@ def golden_ratio_adherence(block_word_counts: dict[str, int], outcome: str) -> d
return {"outcome": outcome, "total_words": total, "sections": sections, "max_deviation": max_dev}
def split_final_by_section(final_text: str) -> dict[str, str]:
"""Group a signed final decision into golden-ratio sections (#208 calibration).
Reuses the SAME structure-aware splitter as measure_corpus_ratios
(chunker._split_into_sections + _CHUNK_SECTION_TO_GOLDEN) — no parallel
parsing path (G2). Returns {golden_section: concatenated_text} for the
sections that map to an AI block (background/claims/discussion/summary).
A section type that does not map (e.g. headers) is dropped, never silently
folded into another section. Best-effort: an unsplittable final returns {}.
"""
from legal_mcp.services.chunker import _split_into_sections
by_section: dict[str, list[str]] = {}
for stype, stext in _split_into_sections(final_text or ""):
g = _CHUNK_SECTION_TO_GOLDEN.get(stype)
if g and stext.strip():
by_section.setdefault(g, []).append(stext.strip())
return {sec: "\n\n".join(parts) for sec, parts in by_section.items()}
def block_distance_to_final(
block_id: str,
regenerated_text: str,
final_section_text: str,
outcome: str,
section_target_total_words: int | None = None,
) -> dict:
"""Distance of ONE regenerated block from the chair's matching final section.
The per-(block, effort) measurement cell for the #208 model/effort
calibration harness. Pure/deterministic (no LLM, no DB) — reuses the
existing style-distance primitives so the harness has no parallel metric
path (G2 / INV-G8 eval-harness):
• change_percent — compute_diff_stats(regen, final_section)
(learning_loop, the SAME diff the pairing
ledger stores). Lower ⇒ the draft already
reads like the final ⇒ less chair rewriting.
• anti_pattern_total — count_anti_patterns(regen) (lessons.ANTI_PATTERNS).
Lower ⇒ closer to Dafna's continuous-narrative
voice; the CLEANEST style signal (07-learning §0.7).
• golden_ratio_deviation_pp — |regen %-of-total final %-of-total| for this
block's section. 0 ⇒ same structural weight as
the final. Requires the final's total words
(section_target_total_words); otherwise None
(we never fabricate a denominator).
Returns the three metrics + a single composite `distance` (normalized,
lower=closer) the harness ranks efforts by.
"""
outcome = canonical_outcome(outcome)
diff = compute_diff_stats(regenerated_text or "", final_section_text or "")
change_percent = diff["change_percent"]
anti_total = count_anti_patterns(regenerated_text or "")["total"]
section = _BLOCK_TO_SECTION.get(block_id)
regen_words = len((regenerated_text or "").split())
final_words = len((final_section_text or "").split())
ratio_dev: float | None = None
if section and section_target_total_words and section_target_total_words > 0:
# Replace the final's own block contribution with the regen's, holding
# the rest of the final constant, to compare structural weight fairly.
regen_total = section_target_total_words - final_words + regen_words
if regen_total > 0:
regen_pct = regen_words / regen_total * 100
final_pct = final_words / section_target_total_words * 100
ratio_dev = round(abs(regen_pct - final_pct), 1)
# Composite: normalize each component to ~[0,1] and average the present ones.
# change_percent/100, anti_total/10 (10+ hits is already very bad), ratio/20.
comps: list[float] = [min(change_percent / 100.0, 1.0), min(anti_total / 10.0, 1.0)]
if ratio_dev is not None:
comps.append(min(ratio_dev / 20.0, 1.0))
distance = round(sum(comps) / len(comps), 4)
return {
"block_id": block_id,
"section": section,
"regen_words": regen_words,
"final_words": final_words,
"change_percent": change_percent,
"anti_pattern_total": anti_total,
"golden_ratio_deviation_pp": ratio_dev,
"distance": distance,
}
async def style_distance(case_number: str) -> dict:
"""Assemble the 3 style-distance components for one case (T7)."""
case = await db.get_case_by_number(case_number)

View File

@@ -350,11 +350,15 @@ async def extract_claims(
if not docs:
return empty(f"אין מסמכים בתיק {case_number}.")
# Filter to claims documents (appeal, response) or specific doc
# Filter to claims documents or a specific doc. protocol carries oral
# arguments raised at the hearing — extracted with claim_type='protocol'
# (claims_extractor._infer_claim_type) so analyze_protocol can compare them
# against the written pleadings, while block ז (write_block_zayin) excludes
# them to stay original-pleadings-only (INV-WR4).
if doc_title:
docs = [d for d in docs if doc_title.lower() in d["title"].lower()]
else:
docs = [d for d in docs if d["doc_type"] in ("appeal", "response", "objection")]
docs = [d for d in docs if d["doc_type"] in ("appeal", "response", "objection", "protocol")]
if not docs:
return empty("לא נמצאו כתבי טענות בתיק.")

View File

@@ -580,6 +580,73 @@ async def get_appraiser_facts(case_number: str) -> str:
return err(str(e))
async def analyze_protocol(case_number: str) -> str:
"""ניתוח השוואתי של פרוטוקול-דיון מול כתבי-הטענות (WS4 / #203).
מזהה אילו טענות **ירדו** (נזנחו בדיון), אילו **חוזקו**, ואילו **עלו חדשות**,
מחדד את השאלות המשפטיות, ומחלץ את נתוני הכותרת (א–ד) מהפרוטוקול. התוצאה נכנסת
ל"ידע-התיק" (טבלת protocol_analysis), זמינה לסוכני הניתוח והכתיבה; תאריך-הדיון
מוזן חזרה לעמודה הקנונית cases.hearing_date.
דורש פרוטוקול מתויג doc_type='protocol' + טיעונים מאוגדים
(aggregate_claims_to_arguments). רץ עם Claude מקומי (Opus 4.8, effort=high);
re-run מחליף את הניתוח הקודם לאותו פרוטוקול (idempotent).
Args:
case_number: מספר תיק הערר
"""
from legal_mcp.services import protocol_analyzer
case = await db.get_case_by_number(case_number)
if not case:
return err(f"תיק {case_number} לא נמצא.")
case_id = UUID(case["id"])
try:
result = await protocol_analyzer.analyze_protocol(case_id)
await audit.log_action_safe(
"analyze_protocol", case_id=case_id,
details={"status": result.get("status"), "total": result.get("total", 0)},
)
return ok(result)
except Exception as e:
return err(str(e))
async def get_protocol_analysis(case_number: str, change_type: str = "") -> str:
"""קריאת ניתוח-הפרוטוקול שכבר חולץ — ללא הרצת ניתוח-מחדש (INV-TOOL4).
ה-get המקביל ל-analyze_protocol: מחזיר את רשומות ידע-התיק השמורות
(ירדה/חוזקה/חדשה + שאלה-מחודדת + ציטוט-מבסס), בלי קריאת-LLM יקרה.
מחזיר רשימה ריקה אם הניתוח טרם רץ (status=ok, count=0) — לא שגיאה.
Args:
case_number: מספר תיק הערר
change_type: סינון (dropped/strengthened/newly_raised). ריק = הכל.
"""
from legal_mcp.services import protocol_analyzer
case = await db.get_case_by_number(case_number)
if not case:
return err(f"תיק {case_number} לא נמצא.")
if change_type and change_type not in protocol_analyzer.VALID_CHANGE_TYPES:
return err(
f"change_type לא תקין: {change_type}",
data={"allowed": sorted(protocol_analyzer.VALID_CHANGE_TYPES)},
)
case_id = UUID(case["id"])
try:
rows = await db.list_protocol_analysis(
case_id, change_type=change_type or None,
)
return ok({
"case_number": case_number,
"count": len(rows),
"analysis": rows,
})
except Exception as e:
return err(str(e))
async def write_interim_draft(case_number: str, instructions: str = "") -> str:
"""כתיבת ארבעת הבלוקים לטיוטת ביניים: רקע (ו), תכניות+היתרים (ט),
טענות הצדדים (ז), הליכים (ח). אם לא חולצו עובדות שמאיות עדיין —

View File

@@ -4,9 +4,18 @@ from __future__ import annotations
from uuid import UUID
from legal_mcp.services import argument_aggregator, db
from legal_mcp.services import argument_aggregator, claims_extractor, db
from legal_mcp.tools.envelope import empty, err, ok # GAP-48: SSoT envelope
# Party labels reused across this module (display + impact diff).
_PARTY_HE = {
"appellant": "עוררים",
"respondent": "משיבים",
"committee": "ועדה מקומית",
"permit_applicant": "מבקשי היתר",
"unknown": "צד לא מזוהה",
}
async def aggregate_claims_to_arguments(
case_number: str,
@@ -55,16 +64,9 @@ async def get_legal_arguments(
)
# Group by party for nicer display.
party_he = {
"appellant": "עוררים",
"respondent": "משיבים",
"committee": "ועדה מקומית",
"permit_applicant": "מבקשי היתר",
"unknown": "צד לא מזוהה",
}
by_party: dict[str, list[dict]] = {}
for a in args:
label = party_he.get(a["party"], a["party"])
label = _PARTY_HE.get(a["party"], a["party"])
by_party.setdefault(label, []).append(a)
return ok({
@@ -72,3 +74,147 @@ async def get_legal_arguments(
"total": len(args),
"by_party": by_party,
})
# ── Re-analysis with merge + impact diff (WS2 / #201) ───────────────
def _snapshot(args: list[dict]) -> dict:
"""Build a comparable snapshot of aggregated arguments for the impact diff.
Captures, per party: the set of argument titles and a priority histogram —
enough to show the chair what changed without dumping full bodies. The
priority mix is the deterministic "recommendation" signal: it summarises the
balance of threshold/substantive/procedural/relief arguments per side.
"""
by_party: dict[str, dict] = {}
for a in args:
party = a.get("party", "unknown")
bucket = by_party.setdefault(party, {"titles": [], "priorities": {}})
bucket["titles"].append((a.get("argument_title") or "").strip())
pr = a.get("priority", "substantive")
bucket["priorities"][pr] = bucket["priorities"].get(pr, 0) + 1
return {"total": len(args), "by_party": by_party}
def _impact_diff(before: dict, after: dict) -> dict:
"""Diff two argument snapshots → a chair-facing before↔after summary.
Reports, per party, which argument titles were added/removed and how the
priority mix (the "recommendation" balance) shifted. ``changed`` is a quick
boolean the UI/chair can gate on. Nothing here is auto-applied — it is purely
surfaced for the human gate (G10).
"""
parties = sorted(set(before["by_party"]) | set(after["by_party"]))
per_party: list[dict] = []
any_change = False
for party in parties:
b = before["by_party"].get(party, {"titles": [], "priorities": {}})
a = after["by_party"].get(party, {"titles": [], "priorities": {}})
b_titles, a_titles = set(b["titles"]), set(a["titles"])
added = sorted(a_titles - b_titles)
removed = sorted(b_titles - a_titles)
prio_before = dict(sorted(b["priorities"].items()))
prio_after = dict(sorted(a["priorities"].items()))
changed = bool(added or removed or prio_before != prio_after)
any_change = any_change or changed
per_party.append({
"party": party,
"party_he": _PARTY_HE.get(party, party),
"count_before": len(b["titles"]),
"count_after": len(a["titles"]),
"added_arguments": added,
"removed_arguments": removed,
"priority_before": prio_before,
"priority_after": prio_after,
"changed": changed,
})
return {
"changed": any_change,
"total_before": before["total"],
"total_after": after["total"],
"by_party": per_party,
}
async def reanalyze_claims(
case_number: str,
reanalyze_all_primary: bool = False,
) -> str:
"""ניתוח-מחדש מאחד של טענות — מחלץ ממסמכים-עיקריים חדשים/לא-נותחו בלבד,
מאחד עם הטענות הקיימות (לא מוחק הכול), מריץ צבירה-מחדש, ומחזיר בדיקת-השפעה
(diff טיעונים/המלצה לפני↔אחרי) ליו"ר.
Args:
case_number: מספר תיק הערר.
reanalyze_all_primary: True = לחלץ מחדש מכל המסמכים-העיקריים (ולא רק
מאלה שטרם-נותחו). ברירת-מחדל False = רק מסמכים-עיקריים חדשים/לא-נותחו.
מנגנון האיחוד (לא force-delete): כל מסמך מחולץ דרך
``claims_extractor.extract_and_store_claims``, ש-``store_claims`` שלו מחליף רק
את טענות *אותו* מסמך (לפי ``source_document``) — כך טענות ממסמכים שכבר-נותחו
נשמרות. הצבירה-מחדש (``aggregate_claims_to_arguments(force=True)``) מחשבת את
הטיעונים מחדש מתוך **מערך-הטענות המאוחד השלם** — מסלול-החישוב הקנוני, לא מסלול
מקביל (G2).
"""
case = await db.get_case_by_number(case_number)
if not case:
return err(f"תיק {case_number} לא נמצא.")
case_id = UUID(case["id"])
# 1. BEFORE snapshot — current aggregated arguments (the impact baseline).
before_args = await argument_aggregator.get_legal_arguments(case_id)
before = _snapshot(before_args)
# 2. Select PRIMARY documents to (re)extract. Default: only those not yet
# analysed (the not-analysed flag). reanalyze_all_primary widens to every
# primary doc. Source of truth for "primary" = is_primary (V47).
if reanalyze_all_primary:
all_docs = await db.list_documents(case_id)
targets = [d for d in all_docs if d.get("is_primary")]
else:
targets = await db.primary_docs_not_analyzed(case_id)
if not targets:
return ok({
"case_number": case_number,
"status": "no_pending_documents",
"message": "אין מסמכים-עיקריים חדשים/לא-נותחו. לא בוצע ניתוח-מחדש.",
"documents_analyzed": [],
"impact": _impact_diff(before, before),
})
# 3. Extract+merge claims from the selected documents only. Each call replaces
# only that document's claims (merge), never the whole case (no force-delete).
analyzed: list[dict] = []
for doc in targets:
text = await db.get_document_text(UUID(doc["id"]))
if not text:
await db.mark_document_claims_extracted(UUID(doc["id"]), status="no_claims")
analyzed.append({"document": doc["title"], "status": "empty_text", "total": 0})
continue
res = await claims_extractor.extract_and_store_claims(
case_id=case_id,
document_id=UUID(doc["id"]),
text=text,
doc_type=doc["doc_type"],
)
analyzed.append({"document": doc["title"], **res})
# 4. Re-aggregate from the now-complete merged claim set. force=True is the
# canonical recompute (deletes only legal_arguments, NOT claims) — same
# path aggregate_claims_to_arguments always uses, not a fork.
agg = await argument_aggregator.aggregate_claims_to_arguments(case_id, force=True)
# 5. AFTER snapshot + impact diff for the chair.
after_args = await argument_aggregator.get_legal_arguments(case_id)
after = _snapshot(after_args)
impact = _impact_diff(before, after)
return ok({
"case_number": case_number,
"status": "completed",
"documents_analyzed": analyzed,
"aggregation": agg,
"impact": impact,
})

View File

@@ -46,8 +46,19 @@ async def workflow_status(case_number: str) -> str:
"extraction": doc["extraction_status"],
"chunks": chunk_map.get(doc["id"], 0),
"pages": doc.get("page_count"),
# WS2 / #201: primary/secondary + whether claims were analysed.
"is_primary": doc.get("is_primary", False),
"claims_analyzed": doc.get("claims_analyzed", False),
})
# WS2 / #201: PRIMARY documents added after the analysis ran and not yet
# included → the "not-analysed" condition that should prompt a merging
# re-analysis (reanalyze_claims). Derived from is_primary + claims_extracted_at.
pending_primary = await db.primary_docs_not_analyzed(case_id)
primary_not_analyzed = [
{"title": d["title"], "type": d["doc_type"]} for d in pending_primary
]
# Check draft status
from pathlib import Path
from legal_mcp import config
@@ -64,15 +75,19 @@ async def workflow_status(case_number: str) -> str:
"documents": doc_status,
"total_documents": len(docs),
"total_chunks": sum(chunk_map.values()),
"primary_docs_not_analyzed": primary_not_analyzed,
"has_draft": has_draft,
"draft_size_bytes": draft_size,
"next_steps": _suggest_next_steps(case, docs, has_draft),
"next_steps": _suggest_next_steps(case, docs, has_draft, primary_not_analyzed),
}
return ok(status)
def _suggest_next_steps(case: dict, docs: list, has_draft: bool) -> list[str]:
def _suggest_next_steps(
case: dict, docs: list, has_draft: bool,
primary_not_analyzed: list | None = None,
) -> list[str]:
"""Suggest next steps based on case state."""
steps = []
doc_types = {d["doc_type"] for d in docs}
@@ -85,6 +100,13 @@ def _suggest_next_steps(case: dict, docs: list, has_draft: bool) -> list[str]:
if "response" not in doc_types:
steps.append("העלה תשובת ועדה/משיבים")
# WS2 / #201: a primary document was added but not yet analysed.
if primary_not_analyzed:
titles = ", ".join(d["title"] for d in primary_not_analyzed)
steps.append(
f"מסמך-עיקרי לא-נותח ({titles}) — הרץ ניתוח-מחדש מאחד (reanalyze_claims)"
)
pending = [d for d in docs if d["extraction_status"] == "pending"]
if pending:
steps.append(f"עיבוד {len(pending)} מסמכים ממתינים")

View File

@@ -0,0 +1,141 @@
"""Tests for the comparative protocol analyzer (WS4 / #203).
Covers the pure (non-LLM, non-DB) logic: the anti-hallucination normalization
gate, protocol-document discovery, claim_type tagging for protocols, and the
block-ז filter that keeps oral hearing arguments out of the original-pleadings
summary (INV-WR4).
"""
from __future__ import annotations
import pytest
from legal_mcp.services import protocol_analyzer as pa
from legal_mcp.services.claims_extractor import _infer_claim_type
# ── claim_type tagging — protocol distinct from written pleadings ───────────
@pytest.mark.parametrize("doc_type,title,expected", [
("protocol", "פרוטוקול דיון", "protocol"),
("appeal", "פרוטוקול הדיון מיום 1.1", "protocol"), # title-based
("appeal", "כתב ערר", "claim"),
("response", "כתב תשובה", "response"),
("response", "תגובת המשיבה", "reply"),
])
def test_infer_claim_type(doc_type, title, expected):
assert _infer_claim_type(doc_type, title) == expected
# ── _find_protocol ─────────────────────────────────────────────────────────
def test_find_protocol_by_doc_type():
docs = [
{"id": "1", "doc_type": "appeal", "title": "כתב ערר"},
{"id": "2", "doc_type": "protocol", "title": "פרוטוקול"},
]
assert pa._find_protocol(docs)["id"] == "2"
def test_find_protocol_by_title_fallback():
docs = [
{"id": "1", "doc_type": "reference", "title": "פרוטוקול הדיון"},
{"id": "2", "doc_type": "appeal", "title": "כתב ערר"},
]
assert pa._find_protocol(docs)["id"] == "1"
def test_find_protocol_none():
docs = [{"id": "1", "doc_type": "appeal", "title": "כתב ערר"}]
assert pa._find_protocol(docs) is None
# ── _normalize_change — anti-hallucination gate (INV-AH) ───────────────────
_AID = "11111111-1111-1111-1111-111111111111"
def test_normalize_change_valid():
row = pa._normalize_change(
{
"change_type": "dropped",
"party_role": "appellant",
"argument_id": _AID,
"argument_title": "טענת השיהוי",
"summary": "העורר ויתר על הטענה בדיון.",
"sharpened_question": "האם נותרה טענת סף?",
"evidence_quote": "ב\"כ העורר: איננו עומדים על טענת השיהוי.",
"page_number": 3,
},
valid_argument_ids={_AID},
)
assert row is not None
assert row["change_type"] == "dropped"
assert str(row["argument_id"]) == _AID
assert row["page_number"] == 3
def test_normalize_change_drops_row_without_evidence_quote():
# quote-or-retract: no verbatim support → row is rejected at source.
row = pa._normalize_change(
{"change_type": "strengthened", "summary": "חוזקה", "evidence_quote": ""},
valid_argument_ids=set(),
)
assert row is None
def test_normalize_change_drops_row_without_summary():
row = pa._normalize_change(
{"change_type": "dropped", "summary": "", "evidence_quote": "ציטוט"},
valid_argument_ids=set(),
)
assert row is None
def test_normalize_change_rejects_bad_change_type():
row = pa._normalize_change(
{"change_type": "modified", "summary": "x", "evidence_quote": "y"},
valid_argument_ids=set(),
)
assert row is None
def test_normalize_change_drops_hallucinated_argument_id():
# An argument_id that isn't one of this case's arguments must be dropped
# (FK safety) — but the row itself, being newly_raised-shaped, survives.
row = pa._normalize_change(
{
"change_type": "newly_raised",
"party_role": "respondent",
"argument_id": "99999999-9999-9999-9999-999999999999",
"summary": "סוגיה חדשה שעלתה בדיון.",
"evidence_quote": "ב\"כ המשיבה העלה לראשונה את שאלת הסמכות.",
},
valid_argument_ids={_AID},
)
assert row is not None
assert row["argument_id"] is None
assert row["change_type"] == "newly_raised"
def test_normalize_change_invalid_party_role_blanked():
row = pa._normalize_change(
{
"change_type": "strengthened",
"party_role": "judge", # not a valid party
"summary": "חוזקה הטענה.",
"evidence_quote": "ציטוט מבסס.",
},
valid_argument_ids=set(),
)
assert row is not None
assert row["party_role"] == ""
def test_normalize_change_truncates_quote():
long_quote = "א" * 500
row = pa._normalize_change(
{"change_type": "dropped", "summary": "x", "evidence_quote": long_quote},
valid_argument_ids=set(),
)
assert row is not None
assert len(row["evidence_quote"]) == 200

View File

@@ -0,0 +1,76 @@
"""Unit tests for the re-analysis impact diff — WS2 / task #201.
`_snapshot` + `_impact_diff` are pure functions (no DB) that produce the
before↔after summary surfaced to the chair after a merging re-analysis. They
must report per-party argument titles added/removed and the shift in the
priority ("recommendation balance") mix, and set ``changed`` correctly.
"""
from __future__ import annotations
from legal_mcp.tools.legal_arguments import _impact_diff, _snapshot
def _arg(party, title, priority="substantive"):
return {"party": party, "argument_title": title, "priority": priority}
def test_no_change_is_flagged_false():
args = [_arg("appellant", "זכות עמידה"), _arg("committee", "שיהוי")]
snap = _snapshot(args)
diff = _impact_diff(snap, snap)
assert diff["changed"] is False
assert diff["total_before"] == diff["total_after"] == 2
# Every party entry reports changed=False.
assert all(p["changed"] is False for p in diff["by_party"])
def test_added_argument_detected():
before = _snapshot([_arg("appellant", "זכות עמידה")])
after = _snapshot([_arg("appellant", "זכות עמידה"), _arg("appellant", "תחולת תמא 38")])
diff = _impact_diff(before, after)
assert diff["changed"] is True
assert diff["total_before"] == 1
assert diff["total_after"] == 2
appellant = next(p for p in diff["by_party"] if p["party"] == "appellant")
assert appellant["added_arguments"] == ["תחולת תמא 38"]
assert appellant["removed_arguments"] == []
assert appellant["party_he"] == "עוררים"
def test_removed_argument_detected():
before = _snapshot([_arg("committee", "שיהוי"), _arg("committee", "סמכות")])
after = _snapshot([_arg("committee", "שיהוי")])
diff = _impact_diff(before, after)
committee = next(p for p in diff["by_party"] if p["party"] == "committee")
assert committee["removed_arguments"] == ["סמכות"]
assert committee["count_before"] == 2
assert committee["count_after"] == 1
assert diff["changed"] is True
def test_priority_mix_shift_is_a_change():
# Same title, but priority moved threshold -> substantive: the recommendation
# balance shifted, so the chair must see a change even with identical titles.
before = _snapshot([_arg("appellant", "סמכות", priority="threshold")])
after = _snapshot([_arg("appellant", "סמכות", priority="substantive")])
diff = _impact_diff(before, after)
appellant = next(p for p in diff["by_party"] if p["party"] == "appellant")
assert appellant["added_arguments"] == []
assert appellant["removed_arguments"] == []
assert appellant["priority_before"] == {"threshold": 1}
assert appellant["priority_after"] == {"substantive": 1}
assert appellant["changed"] is True
assert diff["changed"] is True
def test_new_party_appears():
before = _snapshot([_arg("appellant", "זכות עמידה")])
after = _snapshot([_arg("appellant", "זכות עמידה"), _arg("respondent", "מענה")])
diff = _impact_diff(before, after)
parties = {p["party"] for p in diff["by_party"]}
assert parties == {"appellant", "respondent"}
respondent = next(p for p in diff["by_party"] if p["party"] == "respondent")
assert respondent["count_before"] == 0
assert respondent["count_after"] == 1
assert respondent["added_arguments"] == ["מענה"]

View File

@@ -76,6 +76,7 @@
| `calibrate_halacha_dedup.py` | python | **#82.1** — כיול ספי ה-dedup הלקסיקלי (#82.3) מול gold-set הניקוי. קורא `halacha-cleanup-manifest-*.csv` (זוגות duplicate↔survivor מתויגי-אדם), טוען טקסט-survivor מה-DB, ו-sweep של (jaccard_min × levenshtein_min) עם P/R/F1, מסמן את נקודת-העבודה המוגדרת. אימת ש-(0.55, 0.70) → **precision 1.0** (אפס false-merge), recall 0.30 — מתאים לאיתות-משני שחוסם auto-approve. `--manifest <path>`. רץ עם venv של mcp-server | חד-פעמי — כיול (בוצע 2026-06-06) |
| `ab_halacha_opus48.py` | python | **A/B לא-הרסני לחילוץ הלכות (Claude)** — מריץ מחדש חילוץ הלכות על פסק-דין בודד דרך מודל/effort נבחרים (`AB_MODEL`/`AB_EFFORT`, ברירת-מחדל `claude-opus-4-8`/`xhigh`) ומשווה לסטטיסטיקות ההלכות הקיימות ב-DB **בלי למחוק/לכתוב כלום**. משכפל את `halacha_extractor.extract()` (אותם פרומפטים, בחירת-צ'אנקים, אימות-ציטוט) ומחליף רק את קריאת ה-LLM ב-`claude -p --model --effort`. מפיק `data/ab_halacha_<case>_<effort>.json`. הרצה: `DOTENV_PATH=/home/chaim/.env DATA_DIR=.../data .venv/bin/python scripts/ab_halacha_opus48.py <case_law_id>`. **ממצא 2026-05-31 (שטיין 1128-08-20):** Opus 4.8@xhigh חילץ 51 מול 124 בייצור (100% quote-verified מול 96%) אך ביטחון מכויל-נמוך יותר (חציון 0.75 מול 0.82) — ולכן **לא** מקטין את תור-האישור-הידני תחת sweep אוטו-אישור conf≥0.78 (26 מול 24). שיפור איכות, לא צמצום-תור. | ידני (החלטת מודל-חילוץ) |
| `ab_halacha_codex.py` | python | **A/B לא-הרסני לחילוץ הלכות (Codex/gpt-5.5)** — עמית ל-`ab_halacha_opus48` אך מחליף את `claude -p` ב-`codex exec --model gpt-5.5` (אימות ChatGPT, ללא OPENAI_API_KEY). אותם פרומפטים ואותו הסקת quote-verification. הפלט האחרון של הסוכן (`-o FILE`) נפענח כ-JSON. `AB_MODEL` (default `gpt-5.5`), `AB_REASONING` low/medium/high/xhigh (default `medium`), `AB_CONCURRENCY` (default 1), `CODEX_BIN`. מפיק `data/ab_halacha_codex_<case>_<model>_<reasoning>.json`. הרצה: `DOTENV_PATH=/home/chaim/.env DATA_DIR=.../data mcp-server/.venv/bin/python scripts/ab_halacha_codex.py <case_law_id>`. **ממצא 2026-06-17 (8181-21 האוניברסיטה העברית):** gpt-5.5@medium חילץ 27 מול 28 של Opus (quote-verified 100%/100%), ביטחון חציון 0.86 מול 0.78 — אך **0 פריטים מתחת ל-0.7** (לעומת 9/28 של Opus = 32%), דבר המצביע על over-confidence. holding↑ (12 מול 7), procedural↓ (4 מול 7). **מסקנה: ריאלי כ-fallback חירום; לא מוכן לייצור ללא כיול-ביטחון.** | ידני (בנצ'מרק מודל codex) |
| `calibrate_effort.py` | python | **#208 (WS5/Q1, INV-G8 eval-harness) — כיול model×effort של הכותב מול הסופיים.** A/B per-(תיק,בלוק,effort) על `draft_final_pairs` בעלי `final_text`: מייצר מחדש כל בלוק דרך מסלול-הייצור (`block_writer.write_block(effort_override=…)``claude_session.query``claude -p`, Opus 4.8 נעוץ, **מקומי-בלבד**) ומודד מול ה**סקשן** המתאים בסופי דרך `services/style_distance.block_distance_to_final` (מקור-מדידה יחיד, G2): `change_percent` (compute_diff_stats) · `anti_pattern_total` (`lessons.ANTI_PATTERNS`, הסיגנל הנקי-לסגנון) · `golden_ratio_deviation_pp` · `distance` מרוכב. **ממליץ** per-בלוק על ה-effort בעל ה-distance-הממוצע-הנמוך (לצד ברירת-המחדל מ-#204). מפיק `data/eval/effort-calibration-<ts>.{json,md}`. ⚠️ **גודל-מדגם מודפס בראש הדוח — עדות-כיוון, לא רגרסיה** (מעט סופיים-עלויים). `--self-test` (offline, אפס DB/CLI — מוכיח את לוגיקת-המדידה) · `--dry-run` (תכנון-גריד) · `--efforts`/`--blocks`/`--case`/`--repeats`. **חובה מקומי** (claude CLI; לא בקונטיינר/worktree-ללא-CLI). הרצה: `POSTGRES_PASSWORD=… mcp-server/.venv/bin/python scripts/calibrate_effort.py`. | ידני — לכיול ברירות-effort של הכותב |
| `monitor_halacha_quality.py` | python | מנטר איכות חילוץ הלכות. בודק drift של `avg(confidence)` בין baseline היסטורי לחלון אחרון. מחזיר JSON מטריקות + alert ב-stderr אם drift > threshold (ברירת מחדל 5%). 2 סדרות: trusted (approved+published) ו-all_extracted. תומך `--window N` / `--threshold X` / `--min-sample N` / `--silent` / `--exit-on-alert`. רץ ב-container או מקומית עם `mcp-server/.venv` (אין תלות ב-LLM, רק SQL). **תזמון מומלץ**: `0 8 * * 1` (יום ראשון 08:00, שבועי) | `0 8 * * 1` (לתזמן) |
| `audit_training_corpus.py` | python | audit של `style_corpus` — לכל החלטה: שדות מטא-דאטה מאוכלסים (`summary`/`outcome`/`key_principles`/`appeal_subtype`/`subject_categories`), קישור ל-`documents` (FK + chunks + embeddings). מפיק `data/audit/corpus-YYYY-MM-DD.json` + summary בקונסול. דרוש `POSTGRES_URL` או POSTGRES_*. אין תלויות חיצוניות מלבד asyncpg. **רץ מהמכונה המקומית** (לא קונטיינר) — חיבור ישיר ל-Postgres :5433 | ידני / קדם-עבודה לפני enrichment של מטא-דאטה |
| `backfill_style_exemplars.py` | python | **T1 (style-acquisition)** — מאכלס `style_exemplars` מקורפוס דפנה (`style_corpus` + `internal_committee` chair=דפנה): מפצל לסעיפים (`chunker._split_into_sections`) → פסקאות (25-450 מילים) → embed (Voyage) → שמירה עם `section`/`outcome`/`practice_area`. מאפשר לכותב לאחזר פסקאות-בלוק אמיתיות של דפנה (T2/T3). מקור-סגנון בלבד (INV-LRN5). אידמפוטנטי (מנקה per-decision). `--dry-run` (default) / `--apply`. דורש POSTGRES_URL + Voyage. **רץ מקומית** (venv). | ידני (`python scripts/backfill_style_exemplars.py --apply`) |

419
scripts/calibrate_effort.py Normal file
View File

@@ -0,0 +1,419 @@
#!/usr/bin/env python3
"""#208 (WS5 / Q1, INV-G8 eval-harness) — model×effort calibration vs the finals.
Empirically picks the per-block reasoning `effort` whose regenerated block lands
CLOSEST to עו"ד דפנה תמיר's signed final, over the EXISTING `draft_final_pairs`
ledger. This is the calibration INV-G8 / 07-learning §0.7 ask for: stop choosing
the per-block effort defaults (#204: ה=medium, ו=medium, ז=high, ח=medium, ט=high)
"by feel".
WHAT IT MEASURES (per (case, block, effort) cell — the A/B grid):
Regenerate `block` for `case` via block_writer.write_block(effort_override=…)
(the PRODUCTION generation path → claude_session.query → `claude -p`, pinned
Opus 4.8, local-only), then score the regenerated block against the matching
SECTION of the final via services.style_distance.block_distance_to_final:
• change_percent — word-diff regen↔final-section (compute_diff_stats)
• anti_pattern_total — lessons.ANTI_PATTERNS hits in the regen (cleanest
style signal — see §0.7 warning below)
• golden_ratio_deviation_pp — structural-weight gap vs the final
• distance — normalized composite (lower = closer to Dafna)
No parallel metric path: it reuses style_distance + learning_loop (G2).
RECOMMENDATION: for each block, the effort with the lowest MEAN composite distance
across cases (ties → fewer anti-patterns → lower change_percent). Reported next to
the #204 current default so a regression/improvement is visible.
⚠️ SAMPLE-SIZE CAVEAT (honored, not hidden): very few cases have an uploaded final
(draft_final_pairs.final_text non-empty). The report prints n_finals PROMINENTLY and
labels the output DIRECTIONAL EVIDENCE, not a regression. With n<3 per block the
recommendation is advisory only; the chair/operator decides whether to adopt.
⚠️ change_percent mixes style with content completeness (07-learning §0.7): the chair
sometimes doubles length for missing substance. anti_pattern_total is the cleaner
style signal — the report surfaces both, and the composite down-weights neither
silently.
GENERATION PATH (do not violate — reference_claude_generation_path / claude_session
docstring): write_block → claude_session.query → `claude -p` uses the local claude.ai
session. It runs ONLY on the host where the `claude` CLI exists (NOT the legal-ai
container, NOT a symlinked worktree without CLI access). Hence the live A/B is
host-only; --self-test proves the measurement logic offline with zero model calls.
Usage (mcp-server venv; live needs POSTGRES + the `claude` CLI on the host):
PY=/home/chaim/legal-ai/mcp-server/.venv/bin/python
$PY scripts/calibrate_effort.py --self-test # offline proof, no DB/CLI
POSTGRES_PASSWORD=… POSTGRES_HOST=127.0.0.1 POSTGRES_PORT=5433 \
$PY scripts/calibrate_effort.py # live A/B over finals
… --efforts low,medium,high,xhigh # override the effort grid
… --blocks block-he,block-vav # restrict to some blocks
… --case 8137-11-24 # a single case
… --repeats 2 # avg N gens/cell (noise)
… --dry-run # plan the grid, no model calls
"""
from __future__ import annotations
import argparse
import asyncio
import json
import os
import sys
from datetime import datetime, timezone
from pathlib import Path
from statistics import mean
REPO_ROOT = Path(__file__).resolve().parent.parent
sys.path.insert(0, str(REPO_ROOT / "mcp-server" / "src"))
if "POSTGRES_URL" not in os.environ:
os.environ["POSTGRES_URL"] = (
f"postgres://{os.environ.get('POSTGRES_USER','legal_ai')}:"
f"{os.environ.get('POSTGRES_PASSWORD','')}@"
f"{os.environ.get('POSTGRES_HOST','127.0.0.1')}:"
f"{os.environ.get('POSTGRES_PORT','5433')}/"
f"{os.environ.get('POSTGRES_DB','legal_ai')}"
)
OUT_DIR = REPO_ROOT / "data" / "eval"
# Only the AI blocks that map to a golden-ratio section can be scored against the
# final's matching section (split_final_by_section). Template blocks (א-ד, יב) are
# deterministic template-fill — no effort knob. block-yod (discussion) is out of
# WS5's interim scope but mappable, so it's included when present.
CALIBRATABLE_BLOCKS = ["block-he", "block-vav", "block-zayin", "block-chet", "block-tet", "block-yod", "block-yod-alef"]
DEFAULT_EFFORTS = ["low", "medium", "high", "xhigh"]
VALID_EFFORTS = {"low", "medium", "high", "xhigh", "max"}
# ── pure helpers (offline-testable) ──────────────────────────────────────────
def recommend_effort(cells: list[dict]) -> dict | None:
"""Pick the best effort for ONE block from its scored cells.
cells: [{"effort","distance","anti_pattern_total","change_percent","n"}].
Lowest mean composite distance wins; ties broken by fewer anti-patterns,
then lower change_percent. Pure → unit-tested in --self-test.
"""
if not cells:
return None
ranked = sorted(
cells,
key=lambda c: (c["distance"], c["anti_pattern_total"], c["change_percent"]),
)
return ranked[0]
def aggregate_cell(per_run: list[dict]) -> dict:
"""Mean each metric across repeated generations of the same (case, block, effort)."""
if not per_run:
return {"distance": 1.0, "anti_pattern_total": 0.0, "change_percent": 100.0,
"golden_ratio_deviation_pp": None, "n": 0}
ratios = [r["golden_ratio_deviation_pp"] for r in per_run if r.get("golden_ratio_deviation_pp") is not None]
return {
"distance": round(mean(r["distance"] for r in per_run), 4),
"anti_pattern_total": round(mean(r["anti_pattern_total"] for r in per_run), 2),
"change_percent": round(mean(r["change_percent"] for r in per_run), 2),
"golden_ratio_deviation_pp": round(mean(ratios), 2) if ratios else None,
"n": len(per_run),
}
def _current_default(block_id: str) -> str | None:
from legal_mcp.services.block_writer import BLOCK_CONFIG, DEFAULT_EFFORT
cfg = BLOCK_CONFIG.get(block_id, {})
if cfg.get("model") != "ai":
return None
return cfg.get("effort", DEFAULT_EFFORT)
# ── self-test (no DB, no model) ──────────────────────────────────────────────
def _self_test() -> int:
ok = True
def chk(name, cond):
nonlocal ok
ok = ok and cond
print(f" {name:42} {'ok' if cond else 'FAIL'}")
# block_distance_to_final: regen identical to final-section ⇒ change ~0,
# distance dominated by anti-patterns (0 here) ⇒ ~0.
from legal_mcp.services.style_distance import (
block_distance_to_final, split_final_by_section,
)
final_section = "לפנינו ערר על החלטת הוועדה המקומית. " * 40
d_same = block_distance_to_final("block-he", final_section, final_section, "rejection",
section_target_total_words=len(final_section.split()))
chk("identical regen ⇒ change_percent==0", d_same["change_percent"] == 0.0)
chk("identical regen ⇒ anti==0", d_same["anti_pattern_total"] == 0)
chk("identical regen ⇒ distance small", d_same["distance"] < 0.05)
# A regen full of anti-patterns (markdown headers / bullet lists) scores worse
# than clean continuous narrative, holding the final fixed.
clean = "אנו סבורים כי דין הערר להידחות. כידוע, הלכה פסוקה היא. " * 20
dirty = "## כותרת\n- נקודה ראשונה\n- נקודה שנייה\n* עוד נקודה\n### תת\n" * 10
d_clean = block_distance_to_final("block-yod", clean, final_section, "rejection",
section_target_total_words=len(final_section.split()))
d_dirty = block_distance_to_final("block-yod", dirty, final_section, "rejection",
section_target_total_words=len(final_section.split()))
chk("dirty regen has more anti-patterns", d_dirty["anti_pattern_total"] > d_clean["anti_pattern_total"])
chk("dirty regen ⇒ larger distance", d_dirty["distance"] > d_clean["distance"])
# ratio deviation is None when no total provided (never fabricated)
d_noratio = block_distance_to_final("block-he", clean, final_section, "rejection")
chk("no total ⇒ ratio deviation None", d_noratio["golden_ratio_deviation_pp"] is None)
# split_final_by_section returns mapped golden sections only.
sample_final = (
"רקע עובדתי\nהמקרקעין נשוא הערר. " * 10 + "\n\n"
"תמצית טענות הצדדים\nהעוררים טוענים כי. " * 10 + "\n\n"
"דיון והכרעה\nאנו סבורים כי. " * 10
)
secs = split_final_by_section(sample_final)
chk("split maps to golden sections", set(secs).issubset(
{"background", "claims", "discussion", "summary"}))
chk("split found ≥1 section", len(secs) >= 1)
# recommend_effort: lowest distance wins; tie → fewer anti-patterns.
rec = recommend_effort([
{"effort": "low", "distance": 0.40, "anti_pattern_total": 5, "change_percent": 40, "n": 1},
{"effort": "high", "distance": 0.20, "anti_pattern_total": 2, "change_percent": 25, "n": 1},
{"effort": "xhigh", "distance": 0.20, "anti_pattern_total": 1, "change_percent": 22, "n": 1},
])
chk("recommend picks lowest distance", rec["distance"] == 0.20)
chk("recommend tie → fewer anti-patterns", rec["effort"] == "xhigh")
chk("recommend empty ⇒ None", recommend_effort([]) is None)
# aggregate_cell: means + ratio drops Nones, keeps n.
agg = aggregate_cell([
{"distance": 0.2, "anti_pattern_total": 2, "change_percent": 20, "golden_ratio_deviation_pp": 3.0},
{"distance": 0.4, "anti_pattern_total": 4, "change_percent": 30, "golden_ratio_deviation_pp": None},
])
chk("aggregate distance mean", agg["distance"] == 0.3)
chk("aggregate n counted", agg["n"] == 2)
chk("aggregate ratio skips None", agg["golden_ratio_deviation_pp"] == 3.0)
print("ALL PASS" if ok else "*** FAILURES ***")
return 0 if ok else 1
# ── live A/B (host-only — needs DB + `claude` CLI) ───────────────────────────
async def _finals_for_calibration(case_filter: str | None) -> list[dict]:
"""draft_final_pairs whose final_text is populated (the held-out comparison set)."""
from legal_mcp.services import db
pairs = await db.list_draft_final_pairs(limit=500)
out: list[dict] = []
for p in pairs:
if case_filter and p.get("case_number") != case_filter:
continue
full = await db.get_draft_final_pair(p["id"])
if full and (full.get("final_text") or "").strip():
out.append(full)
return out
async def _score_cell(case_id, block_id: str, effort: str, final_section: str,
final_total_words: int, outcome: str, repeats: int) -> dict:
"""Generate `block_id` at `effort` `repeats` times; score each vs the final section."""
from legal_mcp.services import block_writer
from legal_mcp.services.style_distance import block_distance_to_final
runs: list[dict] = []
for _ in range(repeats):
res = await block_writer.write_block(case_id, block_id, effort_override=effort)
scored = block_distance_to_final(
block_id, res.get("content", ""), final_section, outcome,
section_target_total_words=final_total_words,
)
runs.append(scored)
agg = aggregate_cell(runs)
agg["effort"] = effort
agg["runs"] = runs
return agg
async def _run(args) -> dict:
from uuid import UUID
from legal_mcp.services import db
from legal_mcp.services.lessons import canonical_outcome
from legal_mcp.services.style_distance import split_final_by_section, _BLOCK_TO_SECTION
efforts = args.efforts
blocks = args.blocks
finals = await _finals_for_calibration(args.case)
cases_meta = []
for f in finals:
case = await db.get_case_by_number(f["case_number"]) if f.get("case_number") else None
if not case:
continue
decision = await db.get_decision_by_case(UUID(case["id"]))
outcome = canonical_outcome((decision or {}).get("outcome", "rejection"))
sections = split_final_by_section(f.get("final_text", ""))
final_total_words = len((f.get("final_text", "") or "").split())
cases_meta.append({
"case_number": f["case_number"], "case_id": case["id"],
"outcome": outcome, "sections": sections, "final_total_words": final_total_words,
})
# grid plan: (block → cases that have its section)
plan: dict[str, list[dict]] = {}
for block_id in blocks:
section = _BLOCK_TO_SECTION.get(block_id)
plan[block_id] = [c for c in cases_meta if section and c["sections"].get(section)]
total_cells = sum(len(plan[b]) for b in blocks) * len(efforts) * args.repeats
grid_summary = {
"n_finals": len(cases_meta),
"finals": [c["case_number"] for c in cases_meta],
"blocks": blocks, "efforts": efforts, "repeats": args.repeats,
"total_generations": total_cells,
"per_block_n": {b: len(plan[b]) for b in blocks},
}
if args.dry_run:
return {"dry_run": True, "grid": grid_summary, "by_block": {}}
by_block: dict[str, dict] = {}
for block_id in blocks:
section = _BLOCK_TO_SECTION.get(block_id)
per_effort_runs: dict[str, list[dict]] = {e: [] for e in efforts}
per_case: list[dict] = []
for c in plan[block_id]:
final_section = c["sections"][section]
case_cells = []
for effort in efforts:
cell = await _score_cell(
UUID(c["case_id"]), block_id, effort, final_section,
c["final_total_words"], c["outcome"], args.repeats,
)
per_effort_runs[effort].append(cell)
case_cells.append({k: cell[k] for k in
("effort", "distance", "anti_pattern_total",
"change_percent", "golden_ratio_deviation_pp", "n")})
per_case.append({"case_number": c["case_number"], "cells": case_cells})
# mean across cases for each effort → one comparable row per effort
effort_rows = []
for effort in efforts:
rows = per_effort_runs[effort]
if not rows:
continue
ratios = [r["golden_ratio_deviation_pp"] for r in rows if r.get("golden_ratio_deviation_pp") is not None]
effort_rows.append({
"effort": effort,
"distance": round(mean(r["distance"] for r in rows), 4),
"anti_pattern_total": round(mean(r["anti_pattern_total"] for r in rows), 2),
"change_percent": round(mean(r["change_percent"] for r in rows), 2),
"golden_ratio_deviation_pp": round(mean(ratios), 2) if ratios else None,
"n": len(rows),
})
rec = recommend_effort(effort_rows)
by_block[block_id] = {
"section": section,
"current_default": _current_default(block_id),
"recommended": rec["effort"] if rec else None,
"efforts": effort_rows,
"per_case": per_case,
}
return {"dry_run": False, "grid": grid_summary, "by_block": by_block}
def _ts() -> str:
return datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ")
def _write_report(result: dict, ts: str) -> tuple[Path, Path]:
OUT_DIR.mkdir(parents=True, exist_ok=True)
jp = OUT_DIR / f"effort-calibration-{ts}.json"
mp = OUT_DIR / f"effort-calibration-{ts}.md"
jp.write_text(json.dumps(result, ensure_ascii=False, indent=2), encoding="utf-8")
g = result["grid"]
n = g["n_finals"]
lines = [
f"# #208 — כיול model×effort מול הסופיים — {ts}\n",
f"> ⚠️ **גודל-מדגם: {n} סופיים** ({', '.join(g['finals']) or ''}). "
"זוהי **עדות-כיוון, לא רגרסיה** — מעט תיקים בעלי סופי-עלוי. "
"ההמלצה אדוויזורית; ההכרעה בידי היו\"ר/המפעיל.\n",
f"- בלוקים: {', '.join(g['blocks'])}",
f"- efforts: {', '.join(g['efforts'])} · repeats/cell: {g['repeats']}",
f"- סך ייצורי-מודל: {g['total_generations']}",
"",
]
if result.get("dry_run"):
lines += ["## DRY-RUN — תכנון הגריד בלבד (ללא ייצור)\n",
"| block | #cases | current default |", "|---|---|---|"]
for b in g["blocks"]:
lines.append(f"| {b} | {g['per_block_n'].get(b,0)} | {_current_default(b) or ''} |")
mp.write_text("\n".join(lines) + "\n", encoding="utf-8")
return jp, mp
lines += ["## המלצה per-בלוק (distance נמוך = קרוב יותר לדפנה)\n",
"| block | section | current | **recommended** | n |", "|---|---|---|---|---|"]
for b, bd in result["by_block"].items():
rec = bd.get("recommended") or ""
mark = "" if rec == bd.get("current_default") else " ⬅︎"
n_b = bd["efforts"][0]["n"] if bd.get("efforts") else 0
lines.append(f"| {b} | {bd.get('section','')} | {bd.get('current_default') or ''} | **{rec}**{mark} | {n_b} |")
lines.append("")
for b, bd in result["by_block"].items():
lines += [f"### {b} ({bd.get('section','')})\n",
"| effort | distance | anti_total | change% | ratioΔpp | n |",
"|---|---|---|---|---|---|"]
for r in bd.get("efforts", []):
star = "" if r["effort"] == bd.get("recommended") else ""
ratio = r["golden_ratio_deviation_pp"]
lines.append(
f"| {r['effort']}{star} | {r['distance']:.4f} | {r['anti_pattern_total']} | "
f"{r['change_percent']} | {ratio if ratio is not None else ''} | {r['n']} |")
lines.append("")
lines.append("> change% מערבב סגנון עם שלמות-תוכן (07-learning §0.7); "
"anti_total הוא הסיגנל הנקי-יותר לסגנון.\n")
mp.write_text("\n".join(lines) + "\n", encoding="utf-8")
return jp, mp
async def main() -> int:
ap = argparse.ArgumentParser(description="#208 model/effort calibration harness")
ap.add_argument("--self-test", action="store_true", help="offline measurement-logic proof (no DB/CLI)")
ap.add_argument("--dry-run", action="store_true", help="plan the A/B grid over existing finals, no model calls")
ap.add_argument("--efforts", default=",".join(DEFAULT_EFFORTS),
help=f"comma effort grid (default {','.join(DEFAULT_EFFORTS)})")
ap.add_argument("--blocks", default=",".join(CALIBRATABLE_BLOCKS),
help="comma block ids to calibrate")
ap.add_argument("--case", default=None, help="restrict to a single case_number")
ap.add_argument("--repeats", type=int, default=1, help="generations per cell (avg out gen noise)")
args = ap.parse_args()
if args.self_test:
return _self_test()
args.efforts = [e.strip() for e in args.efforts.split(",") if e.strip()]
bad = [e for e in args.efforts if e not in VALID_EFFORTS]
if bad:
print(f"invalid effort(s): {bad}. valid: {sorted(VALID_EFFORTS)}", file=sys.stderr)
return 2
args.blocks = [b.strip() for b in args.blocks.split(",") if b.strip()]
bad_b = [b for b in args.blocks if b not in CALIBRATABLE_BLOCKS]
if bad_b:
print(f"non-calibratable block(s): {bad_b}. valid: {CALIBRATABLE_BLOCKS}", file=sys.stderr)
return 2
result = await _run(args)
ts = _ts()
jp, mp = _write_report(result, ts)
g = result["grid"]
print(f"CALIBRATION: {g['n_finals']} finals — DIRECTIONAL EVIDENCE, not a regression")
if g["n_finals"] == 0:
print(" no finals with final_text in draft_final_pairs — upload a signed final first.")
elif result.get("dry_run"):
print(f" dry-run: {g['total_generations']} generations planned across {len(g['blocks'])} blocks")
else:
for b, bd in result["by_block"].items():
print(f" {b:16} current={bd.get('current_default') or '':6} "
f"→ recommended={bd.get('recommended') or ''}")
print(f" report: {mp}")
return 0
if __name__ == "__main__":
sys.exit(asyncio.run(main()))