Fix #103961: Drain embedded Kanban workers before gateway restart

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

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()

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.