"""Drain the halacha extraction queue for the incoming batch. Calls the canonical process_pending_extractions(kind='halacha') in small batches until the queue is empty (two consecutive zero-progress rounds). Serial + global advisory-lock coordinated inside the service — avoids concurrent Claude load spikes. NIGHT-WINDOW: halacha extraction is slow (Opus, ~10 min/case) and token-heavy, so the backlog drain runs ONLY in an off-hours window (default 23:00–05:00 Israel time) — it never competes with daytime interactive work or other agents. A tick that starts at 23:00 keeps going until the queue empties OR the window closes (checked before every round); whatever's left resumes the next night (FIFO + per-chunk checkpoint → no lost or duplicated work). Single-case extraction requested by the chair goes through the CEO immediately and is NOT gated here. Window is DST-safe (zoneinfo) — the host runs in UTC. Env overrides: HALACHA_DRAIN_WINDOW_START / _END (hours, 0–23) · HALACHA_DRAIN_TZ. Run: mcp-server/.venv/bin/python scripts/drain_halacha_queue.py """ import asyncio import os import sys from datetime import datetime from zoneinfo import ZoneInfo sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "mcp-server", "src")) from legal_mcp.services import db from legal_mcp.services import precedent_library as pl _TZ = ZoneInfo(os.environ.get("HALACHA_DRAIN_TZ", "Asia/Jerusalem")) _WINDOW_START = int(os.environ.get("HALACHA_DRAIN_WINDOW_START", "23")) _WINDOW_END = int(os.environ.get("HALACHA_DRAIN_WINDOW_END", "5")) def _in_window() -> bool: """True iff the current Israel-time hour is inside [START, END). Handles the midnight wrap (e.g. 23→5): the window is the union of [START,24) and [0,END). If START == END the window is treated as 'always'. """ if _WINDOW_START == _WINDOW_END: return True hour = datetime.now(_TZ).hour if _WINDOW_START < _WINDOW_END: # same-day window, e.g. 1→5 return _WINDOW_START <= hour < _WINDOW_END return hour >= _WINDOW_START or hour < _WINDOW_END # wraps midnight, e.g. 23→5 async def main(): # /operations "disable" switch — no-op immediately if turned off (pm2 # cron_restart can still fire a stopped job, so the gate lives in the DB). if await db.is_drain_disabled("legal-halacha-drain"): print("===SKIP=== legal-halacha-drain disabled via /operations", flush=True) return if not _in_window(): now = datetime.now(_TZ).strftime("%H:%M %Z") print(f"===SKIP=== outside drain window {_WINDOW_START:02d}:00–" f"{_WINDOW_END:02d}:00 (now {now})", flush=True) return total = 0 empty_rounds = 0 rnd = 0 while empty_rounds < 2: if not _in_window(): print(f"===STOP=== drain window closed ({_WINDOW_END:02d}:00) — " f"{total} cases this run; rest resumes next night", flush=True) break rnd += 1 out = await pl.process_pending_extractions(kind="halacha", limit=4) processed = out.get("processed", 0) total_pending = out.get("total_pending", 0) total += processed print(f"[round {rnd}] processed={processed} total_pending={total_pending} status={out.get('status')}", flush=True) for r in out.get("results", []): print(f" {r.get('case_number')}: {r.get('status')} stored={r.get('stored')} retry={r.get('retry_attempts')}", flush=True) if processed == 0: empty_rounds += 1 await asyncio.sleep(5) else: empty_rounds = 0 print(f"\n===DONE=== total halachot rounds processed; cases handled cumulatively={total}", flush=True) if __name__ == "__main__": asyncio.run(main())