feat(style-acq T4+T5): פנקס-התאמה draft↔final + דיסטילציה אוטומטית דרך ה-curator

סוגר את לולאת-הלמידה (INV-LRN4): כל החלטה נסגרת מול הסופי, וכל סופי
מנותח מול הטיוטה. מזין את הטבלאות ש-T15 כבר קורא מהן.

T5 — פנקס-התאמה:
- SCHEMA_V26: טבלת draft_final_pairs (snapshot draft + final + diff + analysis + status).
- db: create/update/list_draft_final_pairs.
- mark-final (app.py): תופס snapshot של הטיוטה (decision_blocks) ברגע החתימה,
  לפני שאפשר לדרוס אותו, ופותח שורת-פנקס (status=final_received).

T4 — דיסטילציה אוטומטית:
- learning_loop.process_final_version: משתמש ב-snapshot (לא בבלוקים שאולי השתנו),
  מסווג style_method↔substance, שומר הצעה ב-pair (status=analyzed).
  **הוסר ה-auto-upsert של style_patterns** — ביטל את ה-bug שדרס את שער-היו"ר
  וזיהם סגנון במהות (INV-LRN1 + INV-LRN5).
- LESSONS_PROMPT: הפרדת style_method↔substance מפורשת + לקח מופשט בלבד.
- curator wake + hermes-curator.md: מריץ ingest_final_version ראשון; מציע רק
  style_method שלא תועד; substance→מסלול precedent.

INV-LRN1 (שער-יו"ר, אין auto-commit) · INV-LRN4 (ניגוד-אמת) · INV-LRN5 (טוהר).

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

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

@@ -1181,6 +1181,29 @@ ALTER TABLE precedent_chunks
     ADD COLUMN IF NOT EXISTS halacha_extracted_at TIMESTAMPTZ;
 """
 
+SCHEMA_V26_SQL = """
+-- draft_final_pairs (T5 / INV-LRN4): the reconciliation ledger.
+-- Every decision is "closed" only after it is compared against the chair's signed
+-- final. Captures an immutable snapshot of the AI draft at mark-final time (before
+-- it can be overwritten), paired with the final. The LLM distillation (curator)
+-- fills final_text + diff_stats + analysis later and advances status.
+CREATE TABLE IF NOT EXISTS draft_final_pairs (
+    id UUID PRIMARY KEY DEFAULT uuid_generate_v4(),
+    case_id UUID NOT NULL REFERENCES cases(id) ON DELETE CASCADE,
+    draft_text TEXT NOT NULL DEFAULT '',
+    final_path TEXT DEFAULT '',
+    final_text TEXT DEFAULT '',
+    diff_stats JSONB DEFAULT NULL,
+    analysis JSONB DEFAULT NULL,
+    -- final_received → analyzed → lessons_folded
+    status TEXT NOT NULL DEFAULT 'final_received',
+    created_at TIMESTAMPTZ DEFAULT now(),
+    updated_at TIMESTAMPTZ DEFAULT now()
+);
+CREATE INDEX IF NOT EXISTS idx_draft_final_pairs_case ON draft_final_pairs(case_id);
+CREATE INDEX IF NOT EXISTS idx_draft_final_pairs_status ON draft_final_pairs(status);
+"""
+
 
 async def _run_schema_migrations(pool: asyncpg.Pool) -> None:
     async with pool.acquire() as conn:
@@ -1210,7 +1233,8 @@ async def _run_schema_migrations(pool: asyncpg.Pool) -> None:
         await conn.execute(SCHEMA_V23_SQL)
         await conn.execute(SCHEMA_V24_SQL)
         await conn.execute(SCHEMA_V25_SQL)
-    logger.info("Database schema initialized (v1-v25)")
+        await conn.execute(SCHEMA_V26_SQL)
+    logger.info("Database schema initialized (v1-v26)")
 
 
 async def init_schema() -> None:
@@ -2241,6 +2265,70 @@ async def get_recent_decision_lessons(limit: int = 15, practice_area: str = "")
     return [dict(r) for r in rows]
 
 
+async def create_draft_final_pair(case_id: UUID, draft_text: str, final_path: str = "") -> str:
+    """Capture the draft↔final pairing at mark-final (T5 / INV-LRN4). Immutable draft
+    snapshot; final_text/diff_stats/analysis filled later by the curator distillation."""
+    pool = await get_pool()
+    async with pool.acquire() as conn:
+        row = await conn.fetchrow(
+            """INSERT INTO draft_final_pairs (case_id, draft_text, final_path, status)
+               VALUES ($1, $2, $3, 'final_received') RETURNING id""",
+            case_id, draft_text, final_path,
+        )
+    return str(row["id"])
+
+
+async def update_draft_final_pair(
+    pair_id: UUID,
+    final_text: str | None = None,
+    diff_stats: dict | None = None,
+    analysis: dict | None = None,
+    status: str | None = None,
+) -> None:
+    """Advance a pairing row (curator distillation): final_text → diff_stats → analysis → status."""
+    sets, params, idx = [], [], 1
+    if final_text is not None:
+        sets.append(f"final_text = ${idx}"); params.append(final_text); idx += 1
+    if diff_stats is not None:
+        sets.append(f"diff_stats = ${idx}::jsonb"); params.append(json.dumps(diff_stats, ensure_ascii=False)); idx += 1
+    if analysis is not None:
+        sets.append(f"analysis = ${idx}::jsonb"); params.append(json.dumps(analysis, ensure_ascii=False)); idx += 1
+    if status is not None:
+        sets.append(f"status = ${idx}"); params.append(status); idx += 1
+    if not sets:
+        return
+    sets.append("updated_at = now()")
+    params.append(pair_id)
+    pool = await get_pool()
+    async with pool.acquire() as conn:
+        await conn.execute(
+            f"UPDATE draft_final_pairs SET {', '.join(sets)} WHERE id = ${idx}", *params,
+        )
+
+
+async def list_draft_final_pairs(status: str | None = None, limit: int = 200) -> list[dict]:
+    """Reconciliation ledger: all decisions paired with their final + status."""
+    pool = await get_pool()
+    async with pool.acquire() as conn:
+        if status:
+            rows = await conn.fetch(
+                """SELECT p.id, p.case_id, c.case_number, c.title, p.status,
+                          p.diff_stats, p.created_at, p.updated_at
+                   FROM draft_final_pairs p LEFT JOIN cases c ON c.id = p.case_id
+                   WHERE p.status = $1 ORDER BY p.created_at DESC LIMIT $2""",
+                status, limit,
+            )
+        else:
+            rows = await conn.fetch(
+                """SELECT p.id, p.case_id, c.case_number, c.title, p.status,
+                          p.diff_stats, p.created_at, p.updated_at
+                   FROM draft_final_pairs p LEFT JOIN cases c ON c.id = p.case_id
+                   ORDER BY p.created_at DESC LIMIT $1""",
+                limit,
+            )
+    return [dict(r) for r in rows]
+
+
 async def upsert_style_pattern(
     pattern_type: str,
     pattern_text: str,