feat(ops): /operations — מוני-תור אחידים, "מה רץ עכשיו", וניהול-תהליכים

הדף הציג את התורים באופן לא-אחיד (by_status גולמי), בלי הבחנה בין "ממתין"
(בקלוג: status=pending) ל"בתור" (התור הפעיל: requested_at IS NOT NULL), בלי
הצגת הפריט שרץ כרגע, ובלי שום שליטה בתהליכים.

מה נוסף:
1. כרטיסי-תור אחידים — בתור / ממתין(בקלוג) / בעיבוד / הושלם / נכשל + "רץ עכשיו"
   (citation/case_number של הפריט בעיבוד) לכל drain (אחזור-פסיקה, מטא-דאטה,
   הלכות, יומונים). שערי-אנוש (אישור-הלכות, פסיקה-חסרה) נשארים מוני-סטטוס.
2. פאנל ניהול-תהליכים בסגנון "שירותי Windows":
   - דמון (court-fetch-service/xvfb/chat/reaper): הפעל-מחדש / עצור / הפעל.
   - cron drain: "הרץ עכשיו" (pm2 restart) + מתג הפעל/כבה תזמון.
3. כל תגי-הסטטוס מתורגמים לעברית.

מנגנון:
- הפעל/כבה תזמון = דגל ב-DB (טבלה drain_controls). pm2 cron_restart מחיה תהליך
  שעוצר ב-stop, לכן ה"כיבוי" האמין הוא דגל שכל drain בודק ב-startup (no-op מיידי
  כשכבוי). הקונטיינר כותב/קורא ישירות מ-DB.
- הרץ-עכשיו + restart/stop/start = proxy ל-pm2 דרך endpoint חדש בגשר-המארח
  (court_fetch_service /pm2/control), מאובטח Bearer + whitelist ל-legal-* בלבד.
- יומונים: drain_digests הועבר מ-crontab ל-pm2 (legal-digest-drain.config.cjs)
  כדי שיופיע ויהיה שליט כמו כל drain. drain_halacha_queue.py הובא לבקרת-גרסאות.

Invariants: מקיים G2 (הרחבת /operations + הגשר הקיים, לא מסלול מקביל) ו-G1
(drain_controls = מקור-אמת יחיד לכיבוי, נורמליזציה במקור ולא תיקון-בקריאה).
אין בליעת שגיאות שקטה (הגשר מחזיר {ok,error}; המוטציות מציגות toast).

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

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

@@ -1401,6 +1401,21 @@ UPDATE digests SET digest_kind =
 CREATE INDEX IF NOT EXISTS idx_digests_kind ON digests(digest_kind);
 """
 
+SCHEMA_V33_SQL = """
+-- drain_controls: a per-drain "startup type" switch for the /operations
+-- dashboard's process-management panel. pm2 cron_restart resurrects a stopped
+-- cron job at the next tick, so `pm2 stop` is NOT a durable "disable" for the
+-- drains. Instead each drain checks this flag at startup and no-ops when
+-- disabled (like a Windows service set to Disabled). The container writes it
+-- directly (no host roundtrip); the drains read it. name = the pm2 process
+-- name (e.g. 'legal-metadata-drain').
+CREATE TABLE IF NOT EXISTS drain_controls (
+    name        TEXT PRIMARY KEY,
+    disabled    BOOLEAN NOT NULL DEFAULT false,
+    updated_at  TIMESTAMPTZ NOT NULL DEFAULT now()
+);
+"""
+
 
 async def _run_schema_migrations(pool: asyncpg.Pool) -> None:
     async with pool.acquire() as conn:
@@ -1437,7 +1452,8 @@ async def _run_schema_migrations(pool: asyncpg.Pool) -> None:
         await conn.execute(SCHEMA_V30_SQL)
         await conn.execute(SCHEMA_V31_SQL)
         await conn.execute(SCHEMA_V32_SQL)
-    logger.info("Database schema initialized (v1-v32)")
+        await conn.execute(SCHEMA_V33_SQL)
+    logger.info("Database schema initialized (v1-v33)")
 
 
 async def init_schema() -> None:
@@ -6144,3 +6160,34 @@ async def court_fetch_job_list(status: str | None = None, limit: int = 100) -> l
                 limit,
             )
     return [_row_to_court_fetch_job(r) for r in rows]
+
+
+# ── Drain controls (/operations process-management panel) ──────────────────
+async def is_drain_disabled(name: str) -> bool:
+    """True if the named drain is switched off (drains check this at startup)."""
+    pool = await get_pool()
+    async with pool.acquire() as conn:
+        val = await conn.fetchval(
+            "SELECT disabled FROM drain_controls WHERE name = $1", name
+        )
+    return bool(val)
+
+
+async def set_drain_disabled(name: str, disabled: bool) -> None:
+    """Switch a drain on/off (upsert). name = pm2 process name."""
+    pool = await get_pool()
+    async with pool.acquire() as conn:
+        await conn.execute(
+            "INSERT INTO drain_controls (name, disabled, updated_at) "
+            "VALUES ($1, $2, now()) "
+            "ON CONFLICT (name) DO UPDATE SET disabled = $2, updated_at = now()",
+            name, disabled,
+        )
+
+
+async def get_drain_controls() -> dict[str, bool]:
+    """Map of drain name → disabled flag (only rows that were ever toggled)."""
+    pool = await get_pool()
+    async with pool.acquire() as conn:
+        rows = await conn.fetch("SELECT name, disabled FROM drain_controls")
+    return {r["name"]: bool(r["disabled"]) for r in rows}