feat(pipeline): durable execution for final_halacha via LangGraph (P0, X16/INV-DUR1, #114)

scripts/_pipeline_runtime.py — runtime עמידות משותף: עוטף רשימת-צעדים async ב-LangGraph
StateGraph ליניארי עם AsyncSqliteSaver (checkpoint לכל צעד). קריסה/OOM ממשיכה מהצעד
שנכשל במקום להריץ הכל מחדש. degradation חיננית: ללא langgraph → ריצה ליניארית כמו קודם
(הכפתור לא נשבר). מימוש אחד לשני הפייפליינים (G2).

final_halacha_pipeline.py — 4 הצעדים ([0]extract [1]citations [2]corroboration [3]panel)
רצים דרך ה-runtime. CLI זהה + --fresh (ברירת-מחדל auto-resume). thread יציב לכל תיק;
dry-run = preview נפרד (תמיד fresh). קריסה בפאנל [3] → resume מ-[3] (steps 0-2 שמורים).

pyproject: extra "durable" (langgraph + langgraph-checkpoint-sqlite) — host-only,
optional. data/checkpoints/ ב-.gitignore.

גבול (X16 §1): LangGraph רק כמנוע-פנימי של הסקריפט — לא orchestrator (לא מסלול מקביל
ל-Paperclip; G2/G12). #108 (atomic extract) קדם לזה כתנאי.

אימות: test_pipeline_runtime.py — עם langgraph (venv-זמני): 3 passed (resume מדלג צעדים
שהושלמו · fresh מריץ-מחדש · linear). בלי langgraph (venv משותף): 1 passed + 2 skipped
(degradation). final_halacha מתקמפל ומיובא נקי בשני המצבים. הרצה end-to-end על הפייפליין
החי (DB+LLM) — לאחר `pip install -e ".[durable]"` בעץ הראשי.

Invariants: INV-DUR1 (עמידות), G2 (runtime יחיד), G3 (idempotency מחוזק).

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

scripts/final_halacha_pipeline.py

@@ -21,8 +21,16 @@ chair drives that from /precedents when a missing precedent is added.
 
 Local-only. Idempotent. The panel pass over the full pending queue can take minutes.
 
+Durable (X16 / INV-DUR1): the 4 steps run through scripts/_pipeline_runtime.py
+with a SQLite checkpoint per case (data/checkpoints/halacha.sqlite). A crash/OOM
+in the long panel [3] RESUMES from [3] on the next run instead of re-paying
+[0]–[2]. Default = auto-resume an interrupted run; ``--fresh`` forces a clean run
+from [0]. Requires the host extra ``pip install -e ".[durable]"`` (mcp-server);
+without it the steps run linearly (same as before) — the button never breaks.
+
     cd ~/legal-ai/mcp-server
     .venv/bin/python ../scripts/final_halacha_pipeline.py --case 8126-03-25
+    .venv/bin/python ../scripts/final_halacha_pipeline.py --case 8126-03-25 --fresh
 """
 from __future__ import annotations
 
@@ -35,6 +43,8 @@ from pathlib import Path
 
 sys.path.insert(0, str(Path(__file__).resolve().parent))
 
+import _pipeline_runtime  # noqa: E402 — durable runtime (X16); scripts/ on sys.path
+from legal_mcp import config  # noqa: E402
 from legal_mcp.services import corroboration, db  # noqa: E402
 from legal_mcp.tools.citations import extract_internal_citations  # noqa: E402
 from legal_mcp.tools.precedent_library import precedent_extract_halachot  # noqa: E402
@@ -59,54 +69,89 @@ async def main(args: argparse.Namespace) -> int:
         print(f"✗ תיק {case_number} לא נמצא")
         return 1
     chair = case.get("chair_name") or "דפנה תמיר"
-
-    # [0] extract the halachot the decision ITSELF states (its own row in case_law) —
-    # so they are not left pending. Idempotent: skip when already completed or on dry-run.
     row = await _decision_law_row(case_number)
-    if not row:
-        print(f"[0/4] ההחלטה {case_number} אינה ב-case_law עדיין — דילוג על חילוץ-הלכות")
-    elif row.get("halacha_extraction_status") == "completed":
-        print(f"[0/4] חילוץ-הלכות מההחלטה — דולג (כבר completed)")
-    elif args.dry_run:
-        print(f"[0/4] חילוץ-הלכות מההחלטה — מדולג (dry-run)")
-    else:
+
+    # The 4 steps as durable nodes (X16 / INV-DUR1): each is checkpointed the
+    # moment it finishes, so a crash/OOM in the long panel [3] resumes from [3]
+    # instead of re-paying [0]–[2]. Steps [0] and [2] stay non-fatal (record the
+    # error and continue); [1]/[3] may raise → the graph halts and the next run
+    # resumes there. All steps are idempotent, so a fresh re-run is also safe.
+
+    async def step_extract(results: dict) -> dict:
+        # [0] extract the halachot the decision ITSELF states (its own case_law row).
+        if not row:
+            print(f"[0/4] ההחלטה {case_number} אינה ב-case_law עדיין — דילוג על חילוץ-הלכות")
+            return {"extract": "skipped:not-enrolled"}
+        if row.get("halacha_extraction_status") == "completed":
+            print("[0/4] חילוץ-הלכות מההחלטה — דולג (כבר completed)")
+            return {"extract": "skipped:completed"}
+        if args.dry_run:
+            print("[0/4] חילוץ-הלכות מההחלטה — מדולג (dry-run)")
+            return {"extract": "skipped:dry-run"}
         print(f"[0/4] precedent_extract_halachot (החלטה {case_number})…", flush=True)
         try:
             raw0 = await precedent_extract_halachot(str(row["id"]))
             d0 = json.loads(raw0).get("data", {})
             print(f"  ✓ status={d0.get('status')} stored={d0.get('stored', d0.get('extracted'))}")
-        except Exception as e:
+            return {"extract": d0.get("status", "done")}
+        except Exception as e:  # non-fatal — record and continue
             print(f"  ⚠ halacha extraction failed (non-fatal): {e}")
+            return {"extract": f"error:{e}"}
 
-    # [1] citation graph
-    print(f"[1/4] extract_internal_citations (chair={chair})…", flush=True)
-    raw = await extract_internal_citations(chair_name=chair, limit=0)
-    try:
-        d = json.loads(raw).get("data", {})
-        print(f"  ✓ extracted {d.get('extracted')} · linked {d.get('linked')} "
-              f"· new {d.get('new')}")
-    except Exception:
-        print(f"  (citations returned: {str(raw)[:160]})")
+    async def step_citations(results: dict) -> dict:
+        # [1] citation graph
+        print(f"[1/4] extract_internal_citations (chair={chair})…", flush=True)
+        raw = await extract_internal_citations(chair_name=chair, limit=0)
+        try:
+            d = json.loads(raw).get("data", {})
+            print(f"  ✓ extracted {d.get('extracted')} · linked {d.get('linked')} "
+                  f"· new {d.get('new')}")
+            return {"citations": "done"}
+        except Exception:
+            print(f"  (citations returned: {str(raw)[:160]})")
+            return {"citations": "unparsed"}
 
-    # [2] corroboration signal + policy (whole corpus backfill) — skipped on dry-run
-    if args.dry_run:
-        print("[2/4] corroboration_rebuild — מדולג (dry-run)")
-    else:
+    async def step_corroboration(results: dict) -> dict:
+        # [2] corroboration signal + policy (whole corpus backfill) — skip on dry-run.
+        if args.dry_run:
+            print("[2/4] corroboration_rebuild — מדולג (dry-run)")
+            return {"corroboration": "skipped:dry-run"}
         print("[2/4] corroboration_rebuild (backfill)…", flush=True)
         try:
             cr = await corroboration.build_all()
             print(f"  ✓ {cr}")
-        except Exception as e:
+            return {"corroboration": "done"}
+        except Exception as e:  # non-fatal
             print(f"  ⚠ corroboration failed (non-fatal): {e}")
+            return {"corroboration": f"error:{e}"}
 
-    # [3] three-judge halacha panel
-    apply = not args.dry_run
-    print(f"[3/4] halacha_panel_approve {'--apply' if apply else '(dry-run)'} "
-          f"(Opus+DeepSeek+Gemini)…", flush=True)
-    import halacha_panel_approve as hpa
-    rc = await hpa.main(Namespace(limit=args.limit, concurrency=6, apply=apply))
+    async def step_panel(results: dict) -> dict:
+        # [3] three-judge halacha panel (the long step durability protects).
+        apply = not args.dry_run
+        print(f"[3/4] halacha_panel_approve {'--apply' if apply else '(dry-run)'} "
+              f"(Opus+DeepSeek+Gemini)…", flush=True)
+        import halacha_panel_approve as hpa
+        rc = await hpa.main(Namespace(limit=args.limit, concurrency=6, apply=apply))
+        return {"panel_rc": rc or 0}
+
+    steps = [
+        _pipeline_runtime.Step("extract_decision_halachot", step_extract),
+        _pipeline_runtime.Step("citations", step_citations),
+        _pipeline_runtime.Step("corroboration", step_corroboration),
+        _pipeline_runtime.Step("panel", step_panel),
+    ]
+    checkpoint_db = config.DATA_DIR / "checkpoints" / "halacha.sqlite"
+    # Stable thread per case → an interrupted real run resumes; dry-runs are
+    # previews (own thread, always fresh — never resume a stale preview).
+    thread_id = f"halacha:{case_number}" + (":dryrun" if args.dry_run else "")
+    results = await _pipeline_runtime.run_pipeline(
+        steps,
+        thread_id=thread_id,
+        checkpoint_db=checkpoint_db,
+        fresh=bool(args.fresh) or args.dry_run,
+    )
     print("\n✓ pipeline-אימות-הלכות הושלם" + (" (dry-run)" if args.dry_run else ""))
-    return rc or 0
+    return int(results.get("panel_rc", 0) or 0)
 
 
 if __name__ == "__main__":
@@ -117,4 +162,7 @@ if __name__ == "__main__":
                     help="cap pending halachot judged (0 = full queue)")
     ap.add_argument("--dry-run", dest="dry_run", action="store_true",
                     help="citations only; skip corroboration writes; panel in dry-run")
+    ap.add_argument("--fresh", action="store_true",
+                    help="ignore any incomplete checkpoint and run from step [0] "
+                         "(default: auto-resume an interrupted run; X16/INV-DUR1)")
     raise SystemExit(asyncio.run(main(ap.parse_args())))