fix: drain embedded Kanban workers before gateway restart

gateway/kanban_watchers.py

@@ -230,6 +230,95 @@ def _kanban_dispatcher_boot(self) -> Optional[tuple]:
                            "on config control alone.", _lock_path)
         return _load_config, _kb, kanban_cfg
 
+    def _kanban_board_metadata(self):
+        """Return every non-archived board or fail closed."""
+        from hermes_cli import kanban_db as _kb
+        from hermes_cli import kanban_db_connect as _kbc
+        from hermes_cli import kanban_db_dispatch as _kbd
+
+        try:
+            boards = _kb.list_boards(include_archived=False)
+        except Exception as exc:
+            raise _kbd.KanbanWorkerScanError(
+                "kanban restart drain: board enumeration failed"
+            ) from exc
+        if not isinstance(boards, list):
+            raise _kbd.KanbanWorkerScanError(
+                "kanban restart drain: board enumeration returned a non-list"
+            )
+        return _kb, _kbc, _kbd, boards
+
+    def _kanban_running_workers(self, *, include_wedged: bool = True) -> list[dict]:
+        """Read live Kanban workers across every non-archived board.
+
+        Workers are child processes with durable PIDs in board databases, not
+        gateway turns. Any enumeration or per-board DB failure is fatal to this
+        scan: proceeding with a partial list could still leave a worker for
+        systemd's KillMode to SIGKILL.
+        """
+        _kb, _kbc, _kbd, boards = self._kanban_board_metadata()
+        workers: list[dict] = []
+        for board_meta in boards:
+            slug = board_meta.get("slug") or _kb.DEFAULT_BOARD
+            conn = None
+            try:
+                conn = _kbc.connect(board=slug)
+                workers.extend(
+                    _kbd.list_running_workers(conn, include_wedged=include_wedged)
+                )
+            except Exception as exc:
+                raise _kbd.KanbanWorkerScanError(
+                    f"kanban restart drain: worker scan failed on board {slug}"
+                ) from exc
+            finally:
+                if conn is not None:
+                    conn.close()
+        return workers
+
+    def _active_kanban_worker_count(self) -> int:
+        """Count live workers, including wedged workers for force cleanup."""
+        return len(self._kanban_running_workers(include_wedged=True))
+
+    def _wedged_kanban_worker_count(self) -> int:
+        """Count live workers excluded from the graceful restart wait."""
+        return sum(
+            1
+            for worker in self._kanban_running_workers(include_wedged=True)
+            if worker.get("wedged")
+        )
+
+    def _interrupt_kanban_workers_for_restart(self) -> int:
+        """Durably block and SIGTERM every worker still live at the deadline."""
+        # Preflight every board before changing any task. If any board cannot be
+        # inspected, propagate the scan error and keep the gateway draining.
+        self._kanban_running_workers(include_wedged=True)
+        _kb, _kbc, _kbd, boards = self._kanban_board_metadata()
+
+        interrupted = 0
+        for board_meta in boards:
+            slug = board_meta.get("slug") or _kb.DEFAULT_BOARD
+            conn = None
+            try:
+                conn = _kbc.connect(board=slug)
+                prepared = _kbd.prepare_workers_for_gateway_restart(
+                    conn, reason="gateway restart"
+                )
+                interrupted += len(prepared)
+                if prepared:
+                    logger.warning(
+                        "kanban restart drain: blocked and SIGTERM'd %d worker(s) on board %s",
+                        len(prepared),
+                        slug,
+                    )
+            except Exception as exc:
+                raise _kbd.KanbanWorkerScanError(
+                    f"kanban restart drain: force-interrupt failed on board {slug}"
+                ) from exc
+            finally:
+                if conn is not None:
+                    conn.close()
+        return interrupted
+
     async def _kanban_dispatcher_watcher(self) -> None:
         """Embedded kanban dispatcher — one tick every `dispatch_interval_seconds`.
 
@@ -271,8 +360,9 @@ 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():
+                if self._draining or not _kanban_dispatch_allowed():
                     bad_ticks = 0
+                    ready_pending = False
                 else:
                     # Re-read the auto-decompose toggle live so disabling it
                     # takes effect on the next tick, not on restart.

gateway/run_shutdown.py

@@ -131,11 +131,14 @@ def elapsed(self) -> float:
     # Active-work accounting
     def _active_work_count(self) -> int:
         """All agent work the gateway must expose and drain as one total."""
+        count_kanban = getattr(self, "_active_kanban_worker_count", None)
+        active_kanban = max(0, int(count_kanban())) if callable(count_kanban) else 0
         return (
             self._running_agent_count()
             + self._active_cron_job_count()
             + self._active_api_run_count()
             + self._active_deferred_agent_worker_count()
+            + active_kanban
         )
 
     @staticmethod
@@ -1329,65 +1332,154 @@ def _idle_seconds(agent: Any) -> Optional[float]:
         )
 
     def _awaitable_work_count(self) -> int:
-        """Active work minus wedged turns — what the restart wait waits on."""
-        return max(0, self._active_work_count() - self._wedged_agent_count())
+        """Active work minus wedged turns — what the restart wait waits on.
 
-    async def _await_active_work_before_restart(self) -> bool:
+        Kanban visibility is safety-critical: an unreadable board must not be
+        converted into zero workers and allow ``stop()`` to run.
+        """
+        count_wedged_kanban = getattr(self, "_wedged_kanban_worker_count", None)
+        wedged_kanban = (
+            max(0, int(count_wedged_kanban()))
+            if callable(count_wedged_kanban)
+            else 0
+        )
+        return max(0, self._active_work_count() - self._wedged_agent_count() - wedged_kanban)
+
+    async def _await_active_work_before_restart(self) -> Optional[bool]:
         """Wait for in-flight work before ``stop()`` so the requesting turn isn't force-interrupted.
 
-        Wedged turns are excluded (restart is their remedy). True when drained to zero, False when the
-        cap elapsed or only wedged work remains (caller proceeds to ``stop()``).
+        ``True`` means all work drained safely; ``False`` means the bounded wait
+        expired or only known wedged work remains and force cleanup is complete.
+        ``None`` means Kanban accounting was incomplete or force cleanup failed;
+        the caller must keep the gateway alive and draining rather than calling
+        ``stop()`` while systemd could SIGKILL an unaccounted worker.
         """
-        active = self._active_work_count()
+        def _interrupt_kanban_or_abort() -> bool:
+            interrupt_kanban = getattr(self, "_interrupt_kanban_workers_for_restart", None)
+            if not callable(interrupt_kanban):
+                return True
+            try:
+                interrupt_kanban()
+            except Exception:
+                logger.exception(
+                    "Restart drain cannot account for all Kanban boards; "
+                    "leaving gateway draining without calling stop()"
+                )
+                return False
+            return True
+
+        try:
+            active = self._active_work_count()
+        except Exception:
+            logger.exception(
+                "Restart drain cannot account for all Kanban boards; "
+                "leaving gateway draining without calling stop()"
+            )
+            return None
         if active <= 0:
             return True
-        if self._awaitable_work_count() <= 0:
+
+        try:
+            awaitable = self._awaitable_work_count()
+        except Exception:
+            logger.exception(
+                "Restart drain lost Kanban worker visibility; "
+                "leaving gateway draining without calling stop()"
+            )
+            return None
+        if awaitable <= 0:
             logger.warning(
                 "Restart requested with %d active work unit(s), all wedged "
                 "past the inactivity timeout; skipping the after-turn wait "
-                "and proceeding to stop()/drain which will interrupt them", active,
+                "and proceeding to stop()/drain which will interrupt them",
+                active,
             )
-            return False
+            return False if _interrupt_kanban_or_abort() else None
+
         timeout = float(getattr(self, "_restart_after_turn_timeout", 0.0) or 0.0)
         if timeout <= 0:
             logger.info(
                 "Restart requested with %d active work unit(s); "
-                "restart_after_turn_timeout=0 — entering stop()/drain immediately", active,
+                "restart_after_turn_timeout=0 — entering stop()/drain immediately",
+                active,
             )
-            return False
+            return False if _interrupt_kanban_or_abort() else None
+
         logger.info(
             "Restart requested with %d active work unit(s); "
             "deferring stop() until they finish (cap=%.0fs) so in-flight "
-            "turns are not amputated (#77184)", active, timeout,
+            "turns are not amputated (#77184)",
+            active,
+            timeout,
         )
         self._scale_to_zero_status("draining", "restart wait: status mark failed")
         loop = asyncio.get_running_loop()
         deadline = loop.time() + timeout
         last_status_at = 0.0
-        while self._awaitable_work_count() > 0:
+        while True:
+            try:
+                awaitable = self._awaitable_work_count()
+            except Exception:
+                logger.exception(
+                    "Restart drain lost Kanban worker visibility; "
+                    "leaving gateway draining without calling stop()"
+                )
+                return None
+            if awaitable <= 0:
+                break
             now = loop.time()
             if now >= deadline:
+                try:
+                    active_now = self._active_work_count()
+                except Exception:
+                    logger.exception(
+                        "Restart drain cannot account for all Kanban boards; "
+                        "leaving gateway draining without calling stop()"
+                    )
+                    return None
                 logger.warning(
                     "Restart after-turn wait timed out after %.0fs with %d "
                     "still active; proceeding to stop()/drain which may "
-                    "interrupt remaining work (#77184)", timeout, self._active_work_count(),
+                    "interrupt remaining work (#77184)",
+                    timeout,
+                    active_now,
                 )
-                return False
+                return False if _interrupt_kanban_or_abort() else None
             if (now - last_status_at) >= 30.0:
+                try:
+                    status_awaitable = self._awaitable_work_count()
+                except Exception:
+                    logger.exception(
+                        "Restart drain lost Kanban worker visibility; "
+                        "leaving gateway draining without calling stop()"
+                    )
+                    return None
                 logger.info(
                     "Restart deferred: waiting on %d active work unit(s) "
                     "(%d wedged and excluded; %.0fs remaining before force drain)",
-                    self._awaitable_work_count(), self._wedged_agent_count(), deadline - now,
+                    status_awaitable,
+                    self._wedged_agent_count(),
+                    deadline - now,
                 )
                 self._scale_to_zero_status("draining", "restart wait: status mark failed")
                 last_status_at = now
             await asyncio.sleep(0.1)
-        if self._active_work_count() > 0:
+
+        try:
+            active = self._active_work_count()
+        except Exception:
+            logger.exception(
+                "Restart drain cannot account for all Kanban boards; "
+                "leaving gateway draining without calling stop()"
+            )
+            return None
+        if active > 0:
             logger.warning(
                 "Restart deferred wait: %d wedged work unit(s) remain; "
-                "proceeding to stop()/drain which will interrupt them", self._active_work_count(),
+                "proceeding to stop()/drain which will interrupt them",
+                active,
             )
-            return False
+            return False if _interrupt_kanban_or_abort() else None
         logger.info("Restart deferred wait complete — active work drained; proceeding to stop()")
         return True
 
@@ -1402,7 +1494,18 @@ def request_restart(self, *, detached: bool = False, via_service: bool = False)
         self._draining = True
 
         async def _run_restart() -> None:
-            await self._await_active_work_before_restart()
+            drain_result = await self._await_active_work_before_restart()
+            if drain_result is None:
+                # A partial board scan is unsafe: keeping the gateway alive and
+                # draining is preferable to letting systemd KillMode amputate
+                # an unaccounted worker. A later operator restart request can
+                # retry after the board/runtime fault is repaired.
+                self._restart_task_started = False
+                logger.error(
+                    "Gateway restart deferred: Kanban worker accounting failed; "
+                    "gateway remains alive and draining"
+                )
+                return
             # Detached helper only AFTER the after-turn wait, or its drain_timeout+5 deadline fires mid-turn.
             if detached:
                 with _log_suppressed(logging.ERROR, "Failed to launch detached gateway restart helper: %s"):

hermes_cli/kanban_db.py

@@ -4188,9 +4188,12 @@ def latest_summaries(conn: sqlite3.Connection, task_ids: Iterable[str]) -> dict[
     DEFAULT_FAILURE_LIMIT,
     DEFAULT_RATE_LIMIT_COOLDOWN_SECONDS,
     DispatchResult,
+    KanbanWorkerScanError,
     _clear_failure_counter,
     _defer_reclaim_for_live_worker,
     _pid_alive,
+    list_running_workers,
+    prepare_workers_for_gateway_restart,
     _terminate_reclaimed_worker,
     _worker_survived_termination,
     _worker_terminal_timeout_env,