feat(learning): אינדיקציית-תיק למצב למידת-קול + חילוץ-הלכות אחרי החלטה סופית

אחרי העלאת החלטה סופית והרצת שני הפייפליינים האוטומטיים (למידת-קול,
חילוץ/אימות-הלכות), התיק לא הציג אם כל תהליך בוצע/הצליח/למה-נכשל. במיוחד
תקלת chair_name ריק (2026-06-12) שמפילה בשקט את העתק-ה-case_law → חילוץ-הלכות
לא מתחיל בכלל, בלי שזה גלוי. כעת מוצגות שתי אינדיקציות ליד כפתורי-ההרצה.

Backend (גזירה ממקור-יחיד, ללא מסלול-מעקב מקביל):
- SCHEMA_V36: draft_final_pairs.learning_run (JSONB) — שדה-תיעוד על פנקס-ההתאמה
  (INV-LRN4), חותם את תוצאת-הריצה של פייפליין-הלמידה (succeeded/failed+סיבה+at).
- set_learning_run_outcome() — חיתום הצלחה/כישלון על ה-pair האחרון.
- case_learning_status() — גזירה read-only מ-draft_final_pairs/style_corpus/
  decision_lessons/case_law/halachot: בוצע? הצליח? למה-לא? כמה הלכות חולצו.
- final_learning_pipeline.py — חותם outcome בהצלחה וב-except (surfaced, לא בלוע).
- חשיפה: case_get מוסיף learning_status (→MCP + /api/cases/{case}/details) +
  endpoint ייעודי GET /api/cases/{case}/learning-status (אותה פונקציה — בלי כפילות).

UI (אושר דרך שער-העיצוב Claude Design — כרטיס 21-final-learning-status):
- useCaseLearningStatus (api/learning.ts) — hook + polling עדין בזמן in-flight.
- LearningStatusBadges — 2 שורות (למידת-קול / חילוץ-הלכות) עם badge + תת-שורה
  (מס' לקחים · רישום-קורפוס / מס' הלכות + פירוק אושרו/ממתינות/נדחו / סיבת-כישלון).
- שילוב ב-drafts-panel תחת "החלטה סופית של היו״ר" + אינוולידציה בכפתורי-ההרצה.

אומת מול ה-DB החי: הצליח+5 הלכות (8174-12-24) · נכנס-אך-pending (1200-12-25) ·
לא-נכנס-לקורפוס (8125-09-24) · round-trip חיתום-כישלון. tsc/eslint נקיים.

Invariants: G1 (נרמול-במקור — גזירה, לא טלאי), G2 (אין מסלול מקביל — שדה על
הפנקס הקיים + exposer יחיד), INV-LRN4 (פנקס-ההתאמה), INV-IA1 (מקור-אמת יחיד).

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

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

@@ -1488,6 +1488,19 @@ CREATE INDEX IF NOT EXISTS idx_panel_rounds_halacha ON halacha_panel_rounds(hala
 CREATE INDEX IF NOT EXISTS idx_panel_rounds_ts ON halacha_panel_rounds(round_ts);
 """
 
+SCHEMA_V36_SQL = """
+-- learning_run on draft_final_pairs: the voice-learning pipeline's RUN OUTCOME.
+-- The pair's `status` (final_received→analyzed→lessons_folded) records how far the
+-- DISTILLATION advanced, but not whether the run-learning button's pipeline
+-- (scripts/final_learning_pipeline.py) actually completed or crashed mid-way — a
+-- crash leaves status at final_received, indistinguishable from "never run". This
+-- column stamps the explicit outcome so the case can SHOW "succeeded / failed +
+-- reason / not-run". CAPTURE field on the existing INV-LRN4 ledger (not a parallel
+-- tracking path, G1/G2). Shape: {status:'succeeded'|'failed', error, at, steps}.
+-- NULL = the pipeline never recorded an outcome for this pair (treated as not-run).
+ALTER TABLE draft_final_pairs ADD COLUMN IF NOT EXISTS learning_run JSONB;
+"""
+
 
 async def _run_schema_migrations(pool: asyncpg.Pool) -> None:
     async with pool.acquire() as conn:
@@ -1527,7 +1540,8 @@ async def _run_schema_migrations(pool: asyncpg.Pool) -> None:
         await conn.execute(SCHEMA_V33_SQL)
         await conn.execute(SCHEMA_V34_SQL)
         await conn.execute(SCHEMA_V35_SQL)
-    logger.info("Database schema initialized (v1-v35)")
+        await conn.execute(SCHEMA_V36_SQL)
+    logger.info("Database schema initialized (v1-v36)")
 
 
 async def init_schema() -> None:
@@ -2627,6 +2641,158 @@ async def update_draft_final_pair(
         )
 
 
+async def set_learning_run_outcome(
+    case_id: UUID,
+    status: str,
+    error: str = "",
+    steps: dict | None = None,
+) -> bool:
+    """Stamp the voice-learning pipeline's RUN OUTCOME on the case's latest pair (SCHEMA_V36).
+
+    The pair's `status` says how far the distillation advanced; this says whether the
+    run-learning pipeline (scripts/final_learning_pipeline.py) completed or crashed —
+    so the case can show 'succeeded / failed + reason'. CAPTURE field on the INV-LRN4
+    ledger (not a parallel path). Returns False if no pair exists yet (nothing to stamp).
+    """
+    payload = {
+        "status": status,                       # 'succeeded' | 'failed'
+        "error": error or "",
+        "steps": steps or {},
+    }
+    pool = await get_pool()
+    async with pool.acquire() as conn:
+        # at = DB now() (no clock in the script path); written via the SQL expression.
+        row = await conn.fetchrow(
+            """UPDATE draft_final_pairs
+                   SET learning_run = ($2::jsonb || jsonb_build_object('at', now())),
+                       updated_at = now()
+                 WHERE id = (
+                     SELECT id FROM draft_final_pairs
+                      WHERE case_id = $1 ORDER BY created_at DESC LIMIT 1
+                 )
+             RETURNING id""",
+            UUID(str(case_id)), json.dumps(payload, ensure_ascii=False),
+        )
+    return row is not None
+
+
+def _as_obj(value) -> dict:
+    """JSONB columns come back as text (no codec registered) — parse defensively."""
+    if isinstance(value, dict):
+        return value
+    if isinstance(value, str) and value:
+        try:
+            parsed = json.loads(value)
+            return parsed if isinstance(parsed, dict) else {}
+        except (ValueError, TypeError):
+            return {}
+    return {}
+
+
+# Halacha-extraction status values that mean the extractor ran and could not produce
+# a usable result — surfaced (not swallowed) so the case shows WHY it didn't complete.
+_HALACHA_FAIL_STATUSES = {"failed", "partial", "extraction_failed", "no_chunks"}
+
+
+async def case_learning_status(case: dict) -> dict:
+    """Derived (read-only) status of the two post-final pipelines for one case.
+
+    Single source of truth — both MCP `case_get` and GET /api/cases/{case}/learning-status
+    call this; no parallel logic. Derives from EXISTING tables (draft_final_pairs,
+    style_corpus, decision_lessons, case_law, halachot); the only persisted addition is
+    draft_final_pairs.learning_run (SCHEMA_V36), the voice pipeline's explicit outcome.
+    """
+    case_id = UUID(str(case["id"]))
+    case_number = case.get("case_number", "")
+    pool = await get_pool()
+    async with pool.acquire() as conn:
+        pair = await conn.fetchrow(
+            """SELECT status, analysis, learning_run, updated_at
+                 FROM draft_final_pairs
+                WHERE case_id = $1 ORDER BY created_at DESC LIMIT 1""",
+            case_id,
+        )
+        corpus = await conn.fetchrow(
+            """SELECT id FROM style_corpus WHERE decision_number = $1 LIMIT 1""",
+            case_number,
+        )
+        lessons_proposed = await conn.fetchval(
+            """SELECT count(*) FROM decision_lessons dl
+                 JOIN style_corpus sc ON sc.id = dl.style_corpus_id
+                WHERE sc.decision_number = $1""",
+            case_number,
+        )
+        law = await conn.fetchrow(
+            """SELECT id, halacha_extraction_status FROM case_law
+                WHERE case_number = $1 AND source_kind = 'internal_committee'
+                ORDER BY created_at DESC LIMIT 1""",
+            case_number,
+        )
+        halacha_counts = {"total": 0, "approved": 0, "pending": 0, "rejected": 0}
+        if law:
+            crow = await conn.fetchrow(
+                """SELECT count(*) AS total,
+                          count(*) FILTER (WHERE review_status = 'approved')      AS approved,
+                          count(*) FILTER (WHERE review_status = 'pending_review') AS pending,
+                          count(*) FILTER (WHERE review_status = 'rejected')      AS rejected
+                     FROM halachot WHERE case_law_id = $1""",
+                law["id"],
+            )
+            halacha_counts = dict(crow)
+
+    # ── Voice learning ──────────────────────────────────────────────────
+    pair_status = pair["status"] if pair else ""
+    run = _as_obj(pair["learning_run"]) if pair else {}
+    if run:
+        outcome = run.get("status") or "not_run"      # explicit pipeline outcome (V36)
+        run_error = run.get("error") or ""
+    elif pair_status in ("analyzed", "lessons_folded"):
+        outcome = "succeeded"                          # distillation done (pre-V36 / direct ingest)
+        run_error = ""
+    else:
+        outcome = "not_run"
+        run_error = ""
+    analysis = _as_obj(pair["analysis"]) if pair else {}
+    voice = {
+        "ran": bool(run),
+        "outcome": outcome,                            # succeeded | failed | not_run
+        "error": run_error or None,
+        "pair_status": pair_status,
+        "lessons_count": len(analysis.get("changes", []) or []),
+        "style_corpus_enrolled": corpus is not None,
+        "lessons_proposed": int(lessons_proposed or 0),
+        "analyzed_at": pair["updated_at"].isoformat() if pair and pair["updated_at"] else None,
+    }
+
+    # ── Halacha extraction ──────────────────────────────────────────────
+    if not law:
+        halacha = {
+            "enrolled_in_corpus": False,
+            "not_enrolled_reason": "ההחלטה הסופית טרם נכנסה לקורפוס-הפסיקה הפנימי — אין ממה לחלץ הלכות",
+            "status": None,
+            "halachot_count": 0,
+            "approved": 0, "pending": 0, "rejected": 0,
+        }
+    else:
+        hstatus = law["halacha_extraction_status"] or "pending"
+        halacha = {
+            "enrolled_in_corpus": True,
+            "not_enrolled_reason": None,
+            "status": hstatus,
+            "failed": hstatus in _HALACHA_FAIL_STATUSES,
+            "halachot_count": int(halacha_counts.get("total", 0) or 0),
+            "approved": int(halacha_counts.get("approved", 0) or 0),
+            "pending": int(halacha_counts.get("pending", 0) or 0),
+            "rejected": int(halacha_counts.get("rejected", 0) or 0),
+        }
+
+    return {
+        "final_uploaded": pair is not None,
+        "voice_learning": voice,
+        "halacha_extraction": halacha,
+    }
+
+
 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()

mcp-server/src/legal_mcp/tools/cases.py

@@ -295,6 +295,15 @@ async def case_get(case_number: str) -> str:
 
     docs = await db.list_documents(UUID(case["id"]))
     case["documents"] = docs
+    # Derived post-final pipeline status (voice learning + halacha extraction) so the
+    # case shows whether each ran, succeeded, and how many halachot were extracted.
+    # Read-only derivation from existing tables (single source — same fn the
+    # /learning-status endpoint uses); best-effort, never fails the case fetch.
+    try:
+        case["learning_status"] = await db.case_learning_status(case)
+    except Exception as e:  # noqa: BLE001 — indicator is best-effort, must not 500 case_get
+        logger.warning("case_learning_status failed for %s: %s", case_number, e)
+        case["learning_status"] = None
     return ok(case)