From 5cbb2e38f175fc69000dbed77f6975a5dfa5f158 Mon Sep 17 00:00:00 2001
From: Rohith Pariki <rohithpariki@gmail.com>
Date: Sun, 6 Sep 2026 17:27:52 +0530
Subject: [PATCH] Fix #103961: Drain embedded Kanban workers before gateway
 restart

---
 gateway/kanban_watchers.py       |  3 +-
 gateway/run_shutdown.py          | 22 ++++++++++-
 hermes_cli/kanban_db_dispatch.py | 66 ++++++++++++++++++++++++++++++++
 3 files changed, 89 insertions(+), 2 deletions(-)

diff --git a/gateway/kanban_watchers.py b/gateway/kanban_watchers.py
index c637086db4c0..37fef1fdffb7 100644
--- a/gateway/kanban_watchers.py
+++ b/gateway/kanban_watchers.py
@@ -271,7 +271,8 @@ async def _kanban_dispatcher_watcher(self) -> None:
             try:
                 # Emergency stop (`hermes pause`): no auto-decompose or
                 # dispatch while paused; running workers finish naturally.
-                if not _kanban_dispatch_allowed():
+                # Also pause during gateway drain.
+                if getattr(self, "_draining", False) or not _kanban_dispatch_allowed():
                     bad_ticks = 0
                 else:
                     # Re-read the auto-decompose toggle live so disabling it
diff --git a/gateway/run_shutdown.py b/gateway/run_shutdown.py
index 8f0281eabec1..295a6d1f57d7 100644
--- a/gateway/run_shutdown.py
+++ b/gateway/run_shutdown.py
@@ -677,12 +677,21 @@ def _resume_paused_platform(self, platform) -> bool:
 
     # Drain / interrupt
     def _drain_work_counts(self) -> tuple:
-        """``(agents, cron, api, deferred)`` — the four sources the drain waits on."""
+        """``(agents, cron, api, deferred, kanban)`` — the five sources the drain waits on."""
         return (
             self._running_agent_count(), self._active_cron_job_count(),
             self._active_api_run_count(), self._active_deferred_agent_worker_count(),
+            self._active_kanban_worker_count(),
         )
 
+    def _active_kanban_worker_count(self) -> int:
+        """Total running host-local Kanban tasks across all non-archived boards."""
+        try:
+            from hermes_cli import kanban_db_dispatch as _kbd
+            return _kbd.count_live_host_local_workers_all_boards()
+        except Exception:
+            return 0
+
     async def _drain_active_agents(
         self, timeout: float, cron_timeout: Optional[float] = None
     ) -> tuple[Dict[str, Any], bool]:
@@ -1498,9 +1507,20 @@ def _interrupt_delegations() -> None:
                 lambda: _interrupt_async(reason=f"gateway shutdown ({phase})"),
             )
 
+        def _stop_interrupt_kanban_workers() -> None:
+            try:
+                from hermes_cli import kanban_db_dispatch as _kbd
+                _count_step(
+                    "Shutdown (%s): durably blocked and SIGTERM'd %d kanban worker(s)",
+                    lambda: _kbd.interrupt_host_local_workers(reason="gateway restart"),
+                )
+            except Exception as _e:
+                logger.debug("Failed to interrupt kanban workers: %s", _e)
+
         _step("process_registry.kill_all", _kill_processes)
         _marked_cron_jobs = _step("mark_running_jobs_interrupted", _mark_cron_interrupted) or []
         _step("async interrupt_all", _interrupt_delegations)
+        _step("interrupt kanban workers", _stop_interrupt_kanban_workers)
         def _cleanup_environments() -> None:
             from tools.terminal_tool_lifecycle import cleanup_all_environments
             cleanup_all_environments()
diff --git a/hermes_cli/kanban_db_dispatch.py b/hermes_cli/kanban_db_dispatch.py
index 743e91dbb542..05d3af71427b 100644
--- a/hermes_cli/kanban_db_dispatch.py
+++ b/hermes_cli/kanban_db_dispatch.py
@@ -1396,6 +1396,72 @@ def count_running_tasks_other_boards(board: Optional[str] = None) -> int:
     return total
 
 
+def count_live_host_local_workers_all_boards() -> int:
+    """Total running Kanban tasks across all non-archived boards on this host."""
+    try:
+        boards = _kb.list_boards(include_archived=False)
+    except Exception:
+        return 0
+    total = 0
+    for meta in boards:
+        slug = meta.get("slug") or _kb.DEFAULT_BOARD
+        try:
+            path = _kb.kanban_db_path(board=slug).expanduser()
+            if not path.exists():
+                continue
+            other = _kbc.connect(board=slug)
+            try:
+                total += count_running_tasks(other)
+            finally:
+                with contextlib.suppress(Exception):
+                    other.close()
+        except Exception:
+            continue
+    return total
+
+
+def interrupt_host_local_workers(reason: str) -> int:
+    """Durably block all live host-local workers with `reason` and SIGTERM them."""
+    import signal
+    from hermes_cli._subprocess_compat import kill
+    try:
+        boards = _kb.list_boards(include_archived=False)
+    except Exception:
+        return 0
+    interrupted = 0
+    for meta in boards:
+        slug = meta.get("slug") or _kb.DEFAULT_BOARD
+        try:
+            path = _kb.kanban_db_path(board=slug).expanduser()
+            if not path.exists():
+                continue
+            other = _kbc.connect(board=slug)
+            try:
+                tasks = other.execute("SELECT id, current_run_id, worker_pid FROM tasks WHERE status = 'running'").fetchall()
+                for task in tasks:
+                    task_id = task["id"]
+                    run_id = task["current_run_id"]
+                    pid = task["worker_pid"]
+                    if not pid:
+                        continue
+                    if _kb.block_task(
+                        other, task_id, reason=reason,
+                        expected_run_id=run_id
+                    ):
+                        if _pid_alive(pid):
+                            try:
+                                kill(int(pid), getattr(signal, "SIGTERM", signal.SIGTERM))
+                            except Exception:
+                                pass
+                        interrupted += 1
+            finally:
+                with contextlib.suppress(Exception):
+                    other.close()
+        except Exception:
+            continue
+    return interrupted
+
+
 def _memory_pressure_level(sample: Optional[Mapping[str, Any]] = None) -> str:
     """Classify system memory pressure: ok/elevated/critical/unknown.