"""Stop/drain/restart, scale-to-zero and active-work accounting methods for GatewayRunner.

Split out of ``gateway/run.py``; bound onto ``GatewayRunner`` via the MRO.
``gateway.run`` internals are imported lazily inside method bodies (import cycle),
so ``patch("gateway.run.X")`` keeps intercepting them at call time.
"""

from __future__ import annotations

import asyncio
import dataclasses
import json
import logging
import os
import shlex
import sys
import threading
import time
from contextlib import contextmanager, nullcontext, suppress
from contextvars import Context
from pathlib import Path
from typing import Any, Callable, Dict, Optional

from agent.i18n import t
from gateway.config import Platform
from gateway.restart import (
    DEFAULT_GATEWAY_CRON_DRAIN_TIMEOUT, GATEWAY_SERVICE_RESTART_EXIT_CODE,
    effective_stop_drain_timeout, effective_stop_watchdog_delay, resolve_cron_drain_budget
)
from gateway.run_common import _UNSET
from gateway.shutdown_watchdog import arm_shutdown_watchdog, resolve_shutdown_watchdog_delay

# Log-record parity with the origin module.
logger = logging.getLogger("gateway.run")


def _exit_with_failure_verdict(runner) -> bool:
    """True (after logging the reason) when the runner asked for a failure exit."""
    if not runner.should_exit_with_failure:
        return False
    if runner.exit_reason:
        logger.error("Gateway exiting with failure: %s", runner.exit_reason)
    return True


def _resolve_gateway_exit_verdict(runner, signal_initiated_shutdown: bool) -> bool:
    """Resolve the process verdict after either startup abort or normal shutdown."""
    if _exit_with_failure_verdict(runner):
        return False
    if runner.exit_code is not None:
        raise SystemExit(runner.exit_code)
    if signal_initiated_shutdown and not runner._restart_requested:
        logger.info(
            "Exiting with code 1 (signal-initiated shutdown without restart "
            "request) so the service manager can revive the gateway."
        )
        return False
    # Older restart paths may not set ``runner.exit_code``; retain the service-restart fallback.
    if runner._restart_via_service:
        logger.info(
            "Exiting with code %d (service-restart requested) so the service "
            "manager relaunches the gateway.",
            GATEWAY_SERVICE_RESTART_EXIT_CODE,
        )
        raise SystemExit(GATEWAY_SERVICE_RESTART_EXIT_CODE)
    return True

# Windows has no bash/setsid chain: a tiny detached Python watcher waits for the gateway PID to
# exit (bounded), then spawns ``hermes gateway restart``.
_WINDOWS_RESTART_WATCHER = """
import os, subprocess, sys, time
from hermes_cli._subprocess_compat import windows_detach_flags_without_breakaway
pid = int(sys.argv[1])
restart_after_s = float(sys.argv[2])
cmd = sys.argv[3:]
deadline = time.monotonic() + restart_after_s

def _alive(p):
    # On Windows, os.kill(pid, 0) is NOT a no-op — it maps to
    # GenerateConsoleCtrlEvent(0, pid) (bpo-14484). Use the
    # Win32 handle-based existence check instead.
    if os.name == 'nt':
        import ctypes
        k32 = ctypes.windll.kernel32
        k32.OpenProcess.restype = ctypes.c_void_p
        k32.WaitForSingleObject.restype = ctypes.c_uint
        k32.GetLastError.restype = ctypes.c_uint
        h = k32.OpenProcess(0x1000 | 0x100000, False, int(p))
        if not h:
            return k32.GetLastError() != 87
        try:
            return k32.WaitForSingleObject(h, 0) == 0x102
        finally:
            k32.CloseHandle(h)
    try:
        os.kill(int(p), 0)
        return True
    except ProcessLookupError:
        return False
    except PermissionError:
        return True
    except OSError:
        return False

while time.monotonic() < deadline:
    if not _alive(pid):
        break
    time.sleep(0.2)
subprocess.Popen(
    cmd,
    stdout=subprocess.DEVNULL,
    stderr=subprocess.DEVNULL,
    creationflags=windows_detach_flags_without_breakaway(),
)
""".strip()


@contextmanager
def _log_suppressed(level: int, msg: str, *args, exc_info: bool = False):
    """``suppress(Exception)`` that logs the swallowed exception on ``gateway.run``.

    Without ``exc_info`` the exception is appended as the last ``%s`` argument (``msg % (*args, exc)``);
    with it the traceback is attached instead. Best-effort seams use this everywhere a failure must be
    visible in the log but must never propagate.
    """
    try:
        yield
    except Exception as exc:
        if exc_info:
            logger.log(level, msg, *args, exc_info=(type(exc), exc, exc.__traceback__))
        else:
            logger.log(level, msg, *args, exc)


def _send_failed(result: Any) -> bool:
    """True when an adapter ``send()`` result explicitly reports failure."""
    return result is not None and getattr(result, "success", True) is False


def _send_error(result: Any) -> str:
    """Error text of a failed ``send()`` result (adapters may omit it)."""
    return getattr(result, "error", "send returned success=False")


def _notice_target_key(platform_value: str, chat_id, thread_id) -> tuple:
    """Dedup key for one notice destination: thread/topic platforms share a chat but route apart."""
    return (platform_value, str(chat_id), str(thread_id) if thread_id else None)


def _delivery_target_key(platform_value: str, chat_id, thread_id, *, profile: Optional[str] = None) -> tuple:
    """Dedupe key for one DELIVERED chat: profile-independent, except Telegram private chats.

    Two served profiles can share one home chat (one Telegram group for the whole host) and owe it
    ONE notice per host restart. A positive Telegram chat id names the USER, though: the same id
    under two bot tokens is two conversations, so those stay keyed per served profile (#118233).
    """
    from gateway.delivery import looks_like_telegram_private_chat_id
    if (profile and profile != "default" and platform_value == "telegram"
            and looks_like_telegram_private_chat_id(chat_id)):
        platform_value = f"{profile}:{platform_value}"
    return _notice_target_key(platform_value, chat_id, thread_id)


def _effective_watchdog_leash(runner: object) -> float:
    """Thread-watchdog leash for the stop in progress: effective drain + grace, clamped under
    launchd's live ``ExitTimeOut`` minus the dump margin. Lives here (not in gateway.restart)
    because restart.py cannot import shutdown_watchdog without a cycle."""
    return effective_stop_watchdog_delay(runner, resolve_shutdown_watchdog_delay(effective_stop_drain_timeout(runner)))


class GatewayShutdownMixin:
    """Stop/drain/restart, scale-to-zero and active-work accounting methods for GatewayRunner."""

    @dataclasses.dataclass
    class _StopContext:
        """State threaded through the ``_stop_*`` phases of one ``stop()`` run."""

        deferred_count: Callable[[], int]
        started_at: Optional[float] = None
        active_agents: dict = dataclasses.field(default_factory=dict)
        timed_out: bool = False
        drain_elapsed: float = 0.0
        # API-server runs still live when the adapters were released; the adapter map is empty by the
        # time the SessionDB close gate runs, so the count has to be taken before ``adapters.clear()``.
        api_live: int = 0

        def elapsed(self) -> float:
            return time.monotonic() - self.started_at

    # Active-work accounting
    def _active_work_count(self) -> int:
        """All agent work the gateway must expose and drain as one total."""
        return (
            self._running_agent_count()
            + self._active_cron_job_count()
            + self._active_api_run_count()
            + self._active_deferred_agent_worker_count()
        )

    @staticmethod
    def _running_cron_job_count() -> int:
        # The FULL work aggregate, not _running_agent_count(): cron jobs run on the scheduler's own thread
        # pool and API-server runs live on the adapter — both outside _running_agents (the #60432 blind
        # spot), so counting agents alone let a suspend land mid-cron-job. Fail-AWAKE accounting: the shared
        # shutdown-drain counters (_active_cron_job_count/_active_api_run_count) swallow exceptions to 0,
        # which is fine for a drain but unsafe for a suspend predicate — a transient read failure would make
        # live work look idle and reopen the mid-job freeze. Here an unreadable source counts as work
        # (sentinel 1) so the machine stays awake until the source is readable again.
        from cron.scheduler import get_running_job_ids
        return len(get_running_job_ids())

    def _active_cron_job_count(self) -> int:
        """Cron jobs currently executing — they run outside ``_running_agents``; 0 if cron can't import.

        Cron jobs run through a standalone ``AIAgent`` on the scheduler's own thread pool
        (``cron/scheduler.py::run_job``), entirely outside ``self._running_agents`` — the dict every OTHER
        active-work check on this class (``_running_agent_count``, ``_drain_active_agents``) reads. Without
        this, the shutdown drain is structurally blind to in-flight cron work: it can report
        ``active_at_start=0`` and proceed straight to killing tool subprocesses while a cron job's terminal
        command is still running (#60432). Best-effort: returns 0 if the cron module can't be imported (e.g.
        a minimal test double for this class).
        """
        try:
            return self._running_cron_job_count()
        except Exception:
            return 0

    def _api_server_hook(self, name: str, *args: Any) -> int:
        """Call the primary API-server adapter's ``name`` hook, clamped >= 0 (0 when the hook is absent).

        Only the primary API server owns the HTTP listener, so only it is a source of this work.
        """
        helper = getattr(getattr(self, "adapters", {}).get(Platform.API_SERVER), name, None)
        return max(0, int(helper(*args))) if callable(helper) else 0

    def _active_api_run_count(self) -> int:
        """API-server work that is outside ``_running_agents``."""
        try:
            return self._api_server_hook("active_agent_work_count")
        except Exception:
            return 0

    def _active_api_worker_count(self) -> int:
        """API-server executor threads still inside an agent turn (#116535).

        Module-level, like the cron registry above: the handler-side adapter count is already
        unreachable here (``adapters`` was cleared a phase earlier) and, worse, drops on handler
        cancellation while the worker thread lives on. Read live at the close gate instead of
        snapshotting.
        """
        try:
            from gateway.platforms.api_server_runs import api_worker_live_count
            return max(0, int(api_worker_live_count()))
        except Exception:
            return 0

    def _interrupt_api_server_runs(self, reason: str) -> int:
        """Interrupt API-server agents not in ``_running_agents`` (same set ``_active_api_run_count`` counts)."""
        try:
            return self._api_server_hook("interrupt_active_runs", reason)
        except Exception as exc:
            logger.debug("Failed interrupting api_server runs during shutdown: %s", exc)
            return 0

    def _mark_api_runs_shutdown_requested(self) -> int:
        """Persist the shutdown boundary on API runs before the drain can await."""
        try:
            return self._api_server_hook("mark_shutdown_requested")
        except Exception as exc:
            logger.debug("Failed marking api_server runs as shutdown-requested: %s", exc)
            return 0

    def _active_deferred_agent_worker_count(self) -> int:
        """Executor workers that outlived their gateway turn (e.g. a timed-out hygiene compression)."""
        workers = getattr(self, "_deferred_agent_workers", None)
        if not isinstance(workers, dict):
            return 0
        return sum(1 for future in list(workers) if not future.done())

    def _track_deferred_agent_worker(self, future: asyncio.Future, agent: Any) -> None:
        """Expose an executor worker to drain/interrupt until it really exits."""
        workers = getattr(self, "_deferred_agent_workers", None)
        if workers is None:
            workers = self._deferred_agent_workers = {}
        workers[future] = agent

        def _discard_worker(done_future: asyncio.Future) -> None:
            workers.pop(done_future, None)
            # Workers that outlive their starting coroutine have no later waiter: consume the
            # terminal exception so asyncio emits no unhandled-future warning.
            # See #98973.
            if not done_future.cancelled():
                with suppress(Exception):
                    done_future.exception()

        future.add_done_callback(_discard_worker)

    def _interrupt_deferred_agent_workers(self, reason: str) -> int:
        """Request cancellation of detached executor-backed agent work."""
        from gateway.run import _INTERRUPT_TOOL_REASON_GATEWAY_SHUTDOWN, request_hard_interrupt
        workers = getattr(self, "_deferred_agent_workers", None)
        if not isinstance(workers, dict):
            return 0
        interrupted = 0
        seen: set[int] = set()
        for future, agent in list(workers.items()):
            if future.done() or agent is None or id(agent) in seen:
                continue
            seen.add(id(agent))
            try:
                request_hard_interrupt(agent, reason, tool_reason=_INTERRUPT_TOOL_REASON_GATEWAY_SHUTDOWN)
                interrupted += 1
            except Exception as exc:
                logger.debug("Failed interrupting deferred agent worker during shutdown: %s", exc)
        return interrupted

    # Scale-to-zero idle detection / dormant-quiesce
    def _scale_to_zero_has_live_background_work(self) -> bool:
        """Live background work (delegations, processes, pending watchers) that must block a suspend.

        PERMANENT supervised watchers (_hermes_supervised_watcher, incl. the scale-to-zero watcher
        itself) are excluded, else this would be True forever and the gateway could never go dormant.
        """
        if any(
            not t.done() and not getattr(t, "_hermes_supervised_watcher", False)
            for t in self._background_tasks
        ):
            return True
        def _delegations_active() -> bool:
            from tools.async_delegation import active_count
            return active_count() > 0

        def _processes_active() -> bool:
            from tools.process_registry import process_registry
            return bool(process_registry.has_any_active() or process_registry.pending_watchers)

        for label, probe in (("async-delegation", _delegations_active), ("bg-work", _processes_active)):
            with _log_suppressed(logging.DEBUG, f"scale-to-zero {label} check failed", exc_info=True):
                if probe():
                    return True
        return False

    @staticmethod
    def _gateway_cfg_section(name: str) -> Optional[dict]:
        """``gateway.<name>`` from the user config when it is a dict, else None (never raises)."""
        from gateway.run import _load_gateway_config
        try:
            user_cfg = _load_gateway_config()
            gw = user_cfg.get("gateway") if isinstance(user_cfg, dict) else None
            section = gw.get(name) if isinstance(gw, dict) else None
        except Exception:  # noqa: BLE001
            return None
        return section if isinstance(section, dict) else None

    def _scale_to_zero_idle_timeout_seconds(self) -> float:
        from gateway.scale_to_zero import parse_idle_timeout_seconds
        stz = self._gateway_cfg_section("scale_to_zero")
        return parse_idle_timeout_seconds(stz.get("idle_timeout_minutes") if stz else None)

    def _restart_loop_guard_config(self) -> tuple:
        """``(max_restarts, window_seconds, max_gap_seconds)`` for the restart-loop breaker.

        ``max_restarts <= 0`` disables it; ``max_gap_seconds`` is the longest spacing between
        restart-interrupted boots that still counts as one loop.
        """
        from gateway import restart_loop_guard as _rlg
        rlg = self._gateway_cfg_section("restart_loop_guard") or {}

        def _int_or(key: str, default: int, positive: bool) -> int:
            value = rlg.get(key)
            if isinstance(value, int) and (value > 0 or not positive):
                return value
            return default

        return (
            _int_or("max_restarts", _rlg.DEFAULT_MAX_RESTARTS, positive=False),
            _int_or("window_seconds", _rlg.DEFAULT_WINDOW_SECONDS, positive=True),
            _int_or("max_gap_seconds", _rlg.DEFAULT_MAX_GAP_SECONDS, positive=True),
        )

    def _scale_to_zero_active_messaging_platforms(self) -> list:
        """Return every enabled or live messaging platform across served profiles.

        ``self.config`` belongs to the launch profile, while ``_profile_adapters`` holds
        live adapters for multiplexed secondary profiles. A direct secondary connection
        must block suspension just like a direct primary connection. A secondary adapter
        parked in ``_profile_failed_platforms`` (popped from ``_profile_adapters`` while a
        retryable fatal reconnects) is still served: suspending mid-reconnect would leave
        that reconnect unable to complete, so pending reconnects count as active too.

        config.platforms is pre-seeded with disabled placeholders, and the api_server is
        force-enabled on every hosted container (counting it silently disarmed the feature).
        """
        non_messaging = {Platform.LOCAL, Platform.API_SERVER, Platform.WEBHOOK}
        active = []

        def add_platform(platform: Platform) -> None:
            if platform not in non_messaging and platform not in active:
                active.append(platform)

        try:
            if self.config:
                for platform, platform_config in self.config.platforms.items():
                    if getattr(platform_config, "enabled", False):
                        add_platform(platform)
            for platform in getattr(self, "adapters", {}) or {}:
                add_platform(platform)
            for profile_adapters in (getattr(self, "_profile_adapters", {}) or {}).values():
                for platform in profile_adapters:
                    add_platform(platform)
            for profile_pending in (getattr(self, "_profile_failed_platforms", {}) or {}).values():
                for platform in profile_pending or {}:
                    add_platform(platform)
        except Exception:  # noqa: BLE001 - unreadable state must keep the gateway awake
            logger.debug(
                "scale-to-zero: active messaging platforms unreadable — staying awake",
                exc_info=True,
            )
            return ["<unavailable>"]
        return active

    @staticmethod
    def _relay_wake_url_or_none():
        from gateway.relay import relay_wake_url
        try:
            return relay_wake_url()
        except Exception:  # noqa: BLE001
            return None

    def _scale_to_zero_should_arm(self) -> bool:
        """Whether to start the idle watcher (D1/D11/§3.4(1))."""
        from gateway.scale_to_zero import messaging_is_relay_only_or_absent, scale_to_zero_enabled, should_arm
        return should_arm(
            enabled=scale_to_zero_enabled(),
            relay_only_or_absent=messaging_is_relay_only_or_absent(self._scale_to_zero_active_messaging_platforms()),
            wake_url=self._relay_wake_url_or_none(),
        )

    def _log_scale_to_zero_not_armed_reason(self) -> None:
        """One INFO line on why the idle watcher did NOT arm — only for an OPTED-IN instance."""
        from gateway.scale_to_zero import messaging_is_relay_only_or_absent, scale_to_zero_enabled
        try:
            if not scale_to_zero_enabled():
                return  # not opted in — normal, stay quiet
            active = [getattr(p, "value", p) for p in self._scale_to_zero_active_messaging_platforms()]
            logger.info(
                "scale-to-zero: NOT armed despite opt-in — relay_only_or_absent=%s (enabled platforms=%s), "
                "wake_url=%s. Need relay-only messaging + a registered wake URL.",
                messaging_is_relay_only_or_absent(active), active or "none",
                "set" if self._relay_wake_url_or_none() else "MISSING",
            )
        except Exception:  # noqa: BLE001 - diagnostics must never block startup
            logger.debug("scale-to-zero: not-armed reason logging failed", exc_info=True)

    def _scale_to_zero_is_idle(self) -> bool:
        from gateway.scale_to_zero import is_idle
        # FULL work aggregate with fail-AWAKE reads: the drain counters swallow errors to 0, which a
        # suspend predicate would read as idle, so an unreadable source counts as work here.

        def _read_or_awake(label: str, fn: Callable[[], Any], busy_sentinel: Any) -> Any:
            try:
                return fn()
            except Exception:  # noqa: BLE001 - unreadable source => assume busy
                logger.debug("scale-to-zero: %s unreadable — staying awake", label, exc_info=True)
                return busy_sentinel

        cron_count = _read_or_awake("cron work count", self._running_cron_job_count, 1)
        api_count = _read_or_awake("api work count", lambda: self._api_server_hook("active_agent_work_count"), 1)
        # An attached dashboard/desktop/TUI client (heartbeat mtime) is inbound activity — folded into
        # the inbound clock, not a conjunct, so a lingering marker cannot pin the box.
        last_inbound = self._last_inbound_at
        from gateway.scale_to_zero import dashboard_client_last_seen
        seen = _read_or_awake("dashboard heartbeat", dashboard_client_last_seen, time.time())
        if seen is not None and seen > last_inbound:
            last_inbound = seen
        return is_idle(
            active_work_count=self._running_agent_count() + cron_count + api_count,
            seconds_since_last_inbound=time.time() - last_inbound,
            idle_timeout_seconds=self._scale_to_zero_idle_timeout_seconds(),
            has_live_background_work=self._scale_to_zero_has_live_background_work(),
        )

    def _scale_to_zero_note_real_inbound(self) -> None:
        """Stamp real inbound and flip status back to running after a dormant wake.

        Internal completion/replay events deliberately do not call this (they must not keep an idle
        gateway awake).
        """
        self._last_inbound_at = time.time()
        if getattr(self, "_scale_to_zero_cooldown_until", 0.0) > 0:
            self._scale_to_zero_status(self._serving_state(), "scale-to-zero: status restore failed")
            self._scale_to_zero_cooldown_until = 0.0

    def _scale_to_zero_status(self, state: str, fail_msg: str) -> None:
        """Best-effort runtime status write; failures are debug-logged with ``fail_msg``."""
        try:
            self._update_runtime_status(state)
        except Exception:  # noqa: BLE001 - status is best-effort
            logger.debug(fail_msg, exc_info=True)

    def _relay_adapter_for_dormancy(self):
        """Return the connected RELAY adapter, if any (the one go_dormant targets)."""
        return self.adapters.get(Platform.RELAY)

    async def _scale_to_zero_watcher(self, interval: float = 30.0) -> None:
        """Watch for idle, drive the relay dormant, then self-suspend. On sustained idle: status
        `draining` (NOT _running=False), relay go_dormant() (socket close, NOT disconnect()), no
        mark_resume_pending (suspend preserves RAM), THEN suspend via the flaps socket — Fly autostop
        sees only INBOUND connections and would freeze mid-job. Without a flaps socket NAS brokers
        the stop through the stamped GATEWAY_RELAY_SLEEP_URL; with no lever at all the watcher
        abstains."""
        from gateway.scale_to_zero import messaging_is_relay_only_or_absent
        await asyncio.sleep(min(interval, 30.0))  # let startup settle
        while self._running:
            try:
                await asyncio.sleep(interval)
                if not self._running:
                    return
                if time.time() < self._scale_to_zero_cooldown_until or not self._scale_to_zero_is_idle():
                    continue
                # The arm gate ran once at boot. A direct adapter that came up since (profile
                # reconcile hot-adding a secondary, a re-enabled platform) owns a socket no wake
                # URL can revive, so the same gate is re-asked before every dormant sequence.
                active = self._scale_to_zero_active_messaging_platforms()
                if not messaging_is_relay_only_or_absent(active):
                    if not self._scale_to_zero_direct_platform_logged:
                        self._scale_to_zero_direct_platform_logged = True
                        logger.info(
                            "scale-to-zero: idle, but directly connected messaging platform(s) %s "
                            "hold a live socket that a suspended instance cannot wake from — staying "
                            "awake. Route them through the relay connector or disable them to allow "
                            "suspend.",
                            ", ".join(str(getattr(p, "value", p)) for p in active),
                        )
                    continue
                self._scale_to_zero_direct_platform_logged = False
                go_dormant = getattr(self._relay_adapter_for_dormancy(), "go_dormant", None)
                if not callable(go_dormant):
                    continue
                # Quiesce only when a suspend can follow: otherwise the re-dial after the socket
                # close just clears the flip again.
                from gateway.scale_to_zero import suspend_available
                if not suspend_available():
                    if not self._scale_to_zero_no_suspend_logged:
                        self._scale_to_zero_no_suspend_logged = True
                        logger.info(
                            "scale-to-zero: idle, but this platform offers no suspend lever (no "
                            "in-machine API and no brokered sleep URL); staying connected rather "
                            "than quiescing"
                        )
                    continue
                logger.info(
                    "scale-to-zero: gateway idle for >= %.0fs — going dormant "
                    "(relay buffered, socket closed) then self-suspending",
                    self._scale_to_zero_idle_timeout_seconds(),
                )
                self._scale_to_zero_status("draining", "scale-to-zero: status mark failed")
                # Both levers: the 1s dormant re-dial can beat either suspend and clear the flip.
                # Held BEFORE go_dormant, whose close arms it.
                if not self._scale_to_zero_hold_redial(True):
                    # Without the hold the re-dial can clear the flip before the stop lands, so
                    # refuse rather than suspend unprotected.
                    logger.warning(
                        "scale-to-zero: could not hold the relay re-dial — staying awake rather "
                        "than suspending unprotected"
                    )
                    self._scale_to_zero_abandon_suspend()
                    continue
                dormant_ok = True
                try:
                    result = go_dormant()
                    if asyncio.iscoroutine(result):
                        result = await result
                    # The going_idle ack. Without it inbound is NOT buffered, so suspending would
                    # freeze a live destination: the whole bug.
                    if result is not True:
                        dormant_ok = False
                        logger.warning(
                            "scale-to-zero: connector did not ack going_idle — staying awake "
                            "rather than freezing a live destination"
                        )
                except Exception:  # noqa: BLE001 - dormancy is best-effort
                    dormant_ok = False
                    logger.debug("scale-to-zero: go_dormant failed", exc_info=True)
                # After a wake the drained inbound updates _last_inbound_at; give it a window so we
                # don't immediately re-go-dormant on the same idle reading before traffic lands.
                self._scale_to_zero_cooldown_until = time.time() + max(interval, 60.0)
                # Suspend ONLY after an ACKED quiesce (else inbound black-holes while we sleep), and
                # re-check idle — inbound may have landed during the quiesce await.
                if not dormant_ok:
                    self._scale_to_zero_abandon_suspend()
                    continue
                if not self._scale_to_zero_is_idle():
                    logger.info("scale-to-zero: inbound arrived during quiesce — skipping suspend")
                    self._scale_to_zero_abandon_suspend()
                    continue
                await self._scale_to_zero_self_suspend()
            except asyncio.CancelledError:
                raise
            except Exception:  # noqa: BLE001 - the watcher must never crash the gateway
                logger.debug("scale-to-zero watcher iteration error", exc_info=True)

    async def _scale_to_zero_self_suspend(self) -> None:
        """Suspend this machine, in-guest where possible and via NAS otherwise (fail-awake).

        Called ONLY after a clean, acked go_dormant(), with the re-dial already held.
        """
        from gateway.scale_to_zero import (
            brokered_sleep_url, request_brokered_suspend, self_suspend_available, suspend_self
        )
        try:
            if self_suspend_available():
                accepted = await asyncio.to_thread(suspend_self)
                lever = "self-suspend"
                if accepted:
                    # flaps answers seconds BEFORE the kernel freezes, so the fence has to span
                    # that gap.
                    await self._scale_to_zero_await_freeze_gap()
                    self._scale_to_zero_hold_redial(False)
                else:
                    self._scale_to_zero_abandon_suspend()
            else:
                # No in-guest API (Azure ACA): NAS holds the credential for the stop verb and
                # brokers it for us.
                url = brokered_sleep_url()
                if not url:
                    # The watcher held on our behalf; nothing is coming to freeze the machine, so
                    # a held supervisor would just stay offline.
                    self._scale_to_zero_abandon_suspend()
                    logger.debug(
                        "scale-to-zero: no suspend lever available — dormant without platform suspend"
                    )
                    return
                # The watcher already holds the supervisor across this call.
                accepted = await asyncio.to_thread(request_brokered_suspend, url)
                lever = "brokered suspend"
                if not accepted:
                    self._scale_to_zero_abandon_suspend()
            if not accepted:
                logger.warning(
                    "scale-to-zero: %s not accepted — machine stays awake (fail-awake); will "
                    "retry on the next idle window", lever,
                )
        except Exception:  # noqa: BLE001 - suspend is best-effort, never crash
            logger.debug("scale-to-zero: self-suspend failed", exc_info=True)
            self._scale_to_zero_abandon_suspend()

    async def _scale_to_zero_await_freeze_gap(self) -> None:
        """Hold the re-dial fence across the flaps-2xx -> kernel-freeze gap.

        Sliced on the WALL clock rather than one ``asyncio.sleep`` because a Fly suspend stops
        CLOCK_MONOTONIC while CLOCK_REALTIME keeps tracking host time. Measured on a Fly machine
        (gru, 2026-09-03) across a 252.219s freeze: ``time.monotonic()`` advanced 0.501s,
        ``time.time()`` advanced 252.219s. ``asyncio.sleep`` runs on ``loop.time()`` (monotonic),
        so a single sleep would resume with its REMAINDER after the wake and delay the drain
        re-dial by exactly that much, on every wake of every Fly agent.

        On the wall clock the deadline is already past by the time we resume, so the fence costs
        nothing after a freeze while still spanning the full gap before one. That decoupling is
        what lets FLY_FREEZE_GRACE_S be sized for the slowest (largest-RAM) machine.
        """
        from gateway.scale_to_zero import FLY_FREEZE_GRACE_S, FLY_FREEZE_GRACE_TICK_S
        deadline = time.time() + FLY_FREEZE_GRACE_S
        while time.time() < deadline:
            await asyncio.sleep(FLY_FREEZE_GRACE_TICK_S)

    def _scale_to_zero_abandon_suspend(self) -> None:
        """Undo a quiesce we are not going to follow with a suspend.

        All three together: a released supervisor still advertising `draining` reads as
        mid-shutdown until the next real inbound event, and an abort that skips the cooldown
        re-runs on every tick.
        """
        self._scale_to_zero_hold_redial(False)
        # Same guard as _exit_external_drain: a real shutdown drain must win, so never resurrect
        # a stopping gateway to `running`.
        if not getattr(self, "_draining", False) and self._running:
            self._scale_to_zero_status(self._serving_state(), "scale-to-zero: status restore failed")
        # An abort before the cooldown is set would otherwise retry every tick.
        self._scale_to_zero_cooldown_until = max(
            self._scale_to_zero_cooldown_until, time.time() + 60.0
        )

    def _scale_to_zero_hold_redial(self, held: bool) -> bool:
        """Hold or release the relay's reconnect supervisor. Returns whether the transport actually
        took it, so the caller can refuse to suspend without the protection rather than fail open."""
        try:
            adapter = self._relay_adapter_for_dormancy()
            if adapter is None:
                return False
            method = getattr(adapter, "hold_redial" if held else "release_redial", None)
            if not callable(method):
                return False
            # Trust the adapter's answer rather than the absence of an exception: it deliberately
            # never raises, so "did not throw" proves nothing.
            return method() is True
        except Exception:  # noqa: BLE001 - never blocks the suspend it precedes
            logger.debug("scale-to-zero: redial hold toggle failed", exc_info=True)
            return False

    # External drain control: the dashboard writes/removes ``.drain_request.json`` (gateway/drain_control.py);
    # the watcher flips between accepting and refusing NEW turns WITHOUT exiting (reversible).
    def _enter_external_drain(self) -> None:
        """Begin external drain: refuse NEW turns (in-flight ones are NOT interrupted). Idempotent."""
        if self._external_drain_active:
            return
        self._external_drain_active = True
        logger.info(
            "External drain ENGAGED (.drain_request.json present) — refusing "
            "new turns; %d in-flight turn(s) will finish. Process stays up.", self._active_work_count(),
        )
        # Persist "draining" so /api/status tracks it; active_agents is read-merged, only state changes.
        self._update_runtime_status("draining")

    def _exit_external_drain(self) -> None:
        """Cancel external drain: re-accept new turns. Idempotent; never resurrects a stopping gateway."""
        if not self._external_drain_active:
            return
        self._external_drain_active = False
        if self._draining or not self._running:
            logger.info(
                "External drain marker cleared during shutdown — not reverting "
                "to running (shutdown takes precedence)."
            )
            return
        logger.info(
            "External drain RELEASED (.drain_request.json removed) — "
            "re-accepting new turns; gateway_state -> %s.", self._serving_state(),
        )
        self._update_runtime_status(self._serving_state())

    async def _drain_control_watcher(self, interval: float = 1.0) -> None:
        """Poll ``.drain_request.json`` at 1s: present -> enter drain, absent -> exit; a stale epoch = absent."""
        from gateway.drain_control import drain_requested
        while self._running:
            try:
                # Off-thread: a synchronous marker read at 1s cadence can stall 30s+ under host I/O
                # pressure and take every platform heartbeat down.
                if await asyncio.to_thread(drain_requested):
                    self._enter_external_drain()
                    # API and cron work live outside messaging's _running_agents map; refresh the
                    # aggregate while an external caller polls this reversible drain state.
                    self._persist_active_agents()
                else:
                    self._exit_external_drain()
            except asyncio.CancelledError:
                raise
            except Exception as exc:
                logger.debug("Drain-control watcher tick error: %s", exc, exc_info=True)
            await asyncio.sleep(interval)

    def _update_platform_runtime_status(
        self, platform: str, *, platform_state: Optional[str] = None,
        error_code: Optional[str] = None, error_message: Optional[str] = None,
        needs_attention: Optional[bool] = None, retrying_since: Any = _UNSET,
    ) -> None:
        from gateway.run import _write_runtime_status_quiet
        extra: Dict[str, Any] = {}
        if needs_attention is not None:
            extra["needs_attention"] = needs_attention
        if retrying_since is not _UNSET:
            extra["retrying_since"] = retrying_since
        _write_runtime_status_quiet(
            platform=platform, platform_state=platform_state, error_code=error_code,
            error_message=error_message, **extra,
        )

    # Per-platform circuit breaker (pause/resume): reconnect watcher + /platform pause|resume.
    def _pause_failed_platform(self, platform, *, reason: str = "") -> None:
        """Pause a queued platform (manual ``/platform pause`` only — the watcher never auto-pauses)."""
        info = getattr(self, "_failed_platforms", {}).get(platform)
        if info is None or info.get("paused"):
            return
        info["paused"] = True
        info["pause_reason"] = reason or "auto-paused after repeated failures"
        # next_retry=inf: a stale code path missing "paused" still never fires.
        info["next_retry"] = float("inf")
        self._update_platform_runtime_status(
            platform.value, platform_state="paused", error_code=None, error_message=info["pause_reason"],
        )
        logger.warning(
            "%s paused after %d consecutive failures (%s) — fix the underlying issue then run `/platform "
            "resume %s` to retry, or `hermes gateway restart` to restart the gateway.",
            platform.value, info.get("attempts", 0), info["pause_reason"], platform.value,
        )

    def _resume_paused_platform(self, platform) -> bool:
        """Unpause a platform (reset attempts, retry on the next watcher tick). True iff it was paused."""
        info = getattr(self, "_failed_platforms", {}).get(platform)
        if info is None or not info.get("paused"):
            return False
        info["paused"] = False
        info.pop("pause_reason", None)
        info["attempts"] = 0
        info["next_retry"] = time.monotonic()  # retry on next watcher tick
        self._update_platform_runtime_status(platform.value, platform_state="retrying")
        logger.info("%s resumed — retrying on next watcher tick", platform.value)
        return True

    # Drain / interrupt
    def _drain_work_counts(self) -> tuple:
        """``(agents, cron, api, deferred)`` — the four 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(),
        )

    async def _drain_active_agents(
        self, timeout: float, cron_timeout: Optional[float] = None
    ) -> tuple[Dict[str, Any], bool]:
        snapshot = self._snapshot_running_agents()
        loop = asyncio.get_running_loop()
        last_counts = self._drain_work_counts()
        last_status_at = 0.0

        def _maybe_update_status(force: bool = False) -> None:
            nonlocal last_counts, last_status_at
            now = loop.time()
            counts = self._drain_work_counts()
            if force or counts != last_counts or (now - last_status_at) >= 1.0:
                self._update_runtime_status("draining")
                last_counts, last_status_at = counts, now

        # Cron/API/deferred work lives outside ``_running_agents``; fold it in or it is killed unwarned.
        _cron0, _api0, _deferred0 = last_counts[1:]
        _maybe_update_status(force=True)
        if not self._running_agents and not (_cron0 or _api0 or _deferred0):
            return snapshot, False
        # Cron has its own deadline: a chat turn is announced+resumable; a killed cron run is a permanent failure.
        # ``timeout`` (``restart_drain_timeout``) defaults to 0 because interrupting a chat turn is
        # announced and resumable; a cron run killed mid-flight is recorded in jobs.json as a permanent
        # failure nobody is waiting on. Sharing one budget meant the default config could report
        # ``timed_out=True`` after 0.00s with a cron job in flight and kill it — the drain never even
        # entered this loop (#82161).
        started = loop.time()
        deadline = started + timeout
        cron_deadline = started + (timeout if cron_timeout is None else cron_timeout)

        def _still_draining() -> bool:
            now = loop.time()
            agents, cron, api, deferred = self._drain_work_counts()
            return bool(((agents or api or deferred) and now < deadline) or (cron and now < cron_deadline))

        # Both budgets at 0 = an expired deadline (loop unentered), so timed_out still comes from real state.
        while _still_draining():
            _maybe_update_status()
            await asyncio.sleep(0.1)
        timed_out = any(self._drain_work_counts())
        _maybe_update_status(force=True)
        return snapshot, timed_out

    def _interrupt_running_agents(self, reason: str) -> None:
        from gateway.run import _AGENT_PENDING_SENTINEL, _INTERRUPT_TOOL_REASON_GATEWAY_SHUTDOWN, request_hard_interrupt
        for session_key, agent in list(self._running_agents.items()):
            if agent is _AGENT_PENDING_SENTINEL:
                continue
            with _log_suppressed(logging.DEBUG, "Failed interrupting agent during shutdown: %s"):
                request_hard_interrupt(agent, reason, tool_reason=_INTERRUPT_TOOL_REASON_GATEWAY_SHUTDOWN)
                logger.debug("Interrupted running agent for session %s during shutdown", session_key)
        # API-server / desk turns are adapter-owned and never enter _running_agents, so the loop above
        # cannot see them even though _drain_active_agents() waited for them.
        for count, what in (
            (self._interrupt_api_server_runs(reason), "api_server run(s)"),
            (self._interrupt_deferred_agent_workers(reason), "deferred agent worker(s)"),
        ):
            if count:
                logger.debug("Interrupted %d %s during shutdown", count, what)

    def _shutdown_interrupt_reason(self) -> str:
        from gateway.run import _INTERRUPT_REASON_GATEWAY_RESTART, _INTERRUPT_REASON_GATEWAY_SHUTDOWN
        return _INTERRUPT_REASON_GATEWAY_RESTART if self._restart_requested else _INTERRUPT_REASON_GATEWAY_SHUTDOWN

    async def _mark_running_sessions_resume_pending(self, log_prefix: str) -> list:
        """Mark every non-pending running session resume_pending; returns the keys marked."""
        from gateway.run import _AGENT_PENDING_SENTINEL
        reason = "restart_timeout" if self._restart_requested else "shutdown_timeout"
        marked: list[str] = []
        # Pre-mark sessions as resume_pending BEFORE the drain wait. If the process is killed by the service
        # manager during the drain, the durable marker is already written so the next gateway boot can
        # recover in-flight sessions (#27856).
        for _sk, _agent in list(self._running_agents.items()):
            if _agent is _AGENT_PENDING_SENTINEL:
                continue
            with _log_suppressed(logging.DEBUG, "%s failed for %s: %s", log_prefix, _sk):
                await self.async_session_store.mark_resume_pending(_sk, reason)
                marked.append(_sk)
        return marked

    def _restart_notification_allowed(self, platform: Platform) -> bool:
        """False when the platform config sets ``gateway_restart_notification=false``."""
        platform_cfg = self.config.platforms.get(platform)
        return platform_cfg is None or bool(platform_cfg.gateway_restart_notification)

    def _notice_allowed(self, platform: Platform, what: str, platform_cfg=None) -> bool:
        """``_restart_notification_allowed`` with the INFO suppression line for shutdown notices.
        ``platform_cfg`` is a SERVED profile's own platform entry; ``self.config`` is the launch profile's."""
        allowed = (self._restart_notification_allowed(platform) if platform_cfg is None
                   else bool(platform_cfg.gateway_restart_notification))
        if allowed:
            return True
        logger.info(
            "Shutdown notification suppressed for %s: %s has gateway_restart_notification=false", what, platform.value,
        )
        return False

    async def _notify_interrupted_cron_jobs(self, job_ids) -> int:
        """Tell the owner of each just-interrupted cron job that its run died; returns notices sent.

        The cron worker can't (its thread reaches ``_deliver_result`` after teardown closed the
        transport), so this runs post-interrupt while adapters are still connected. Best-effort.

        Its thread reaches ``_deliver_result`` asynchronously, and by then ``_bounded_adapter_teardown`` has
        closed the transport — so the notice never leaves the process, and ``_consume_interrupted_flag``
        discards the resulting ``delivery_error`` along with it. The run's only trace is a line in jobs.json
        nobody reads (#82232).
        Must therefore be called from the post-interrupt phase, while adapters are still connected — the
        same window ``_notify_active_sessions_of_shutdown`` relies on for chat sessions, which is blind to
        cron work because cron runs on the scheduler's own thread pool rather than ``self._running_agents``
        (#60432).
        """
        if not job_ids:
            return 0
        try:
            from cron.jobs import get_job
            from cron.scheduler import _resolve_delivery_targets
        except Exception as e:
            logger.debug("Cron interrupt notification unavailable: %s", e)
            return 0
        action = t("gateway.shutdown.action_restarting" if self._restart_requested else "gateway.shutdown.action_shutting_down")
        notified: set = set()
        for job_id in job_ids:
            try:
                job = get_job(job_id)
                if not job:
                    continue
                # deliver=local / unresolvable-origin jobs resolve to zero targets and stay silent (no home-
                # channel fallback). Interrupted notices are failure-category status: honor failure_deliver.
                # See #43014.
                targets = _resolve_delivery_targets(job, for_failure=True)
            except Exception as e:
                logger.debug("Cron interrupt targets unresolved for %s: %s", job_id, e)
                continue
            job_name = job.get("name") or job_id
            msg = t("gateway.shutdown.cron_interrupted", job=job_name, action=action)
            for target in targets or ():
                try:
                    platform = Platform(str(target.get("platform", "")).lower())
                except Exception:
                    continue
                adapter = self.adapters.get(platform)
                if adapter is None or not self._restart_notification_allowed(platform):
                    continue
                chat_id = str(target.get("chat_id"))
                thread_id = target.get("thread_id")
                dedup_key = (job_id, *_notice_target_key(platform.value, chat_id, thread_id))
                if dedup_key in notified:
                    continue
                with _log_suppressed(logging.DEBUG, "Cron interrupt notice to %s:%s raised: %s", platform.value, chat_id):
                    metadata = self._thread_metadata_for_target(platform, chat_id, thread_id, adapter=adapter)
                    async def send_notice():
                        if await self._send_notice_logged(
                            adapter, chat_id, msg, platform.value, "Cron interrupt notice to %s:%s failed: %s",
                            "Cron interrupt notice to %s:%s raised: %s", metadata=metadata,
                        ):
                            notified.add(dedup_key)
                    from gateway.warning_notifications import present_notification
                    await present_notification(send_notice, platform=platform)
        if notified:
            logger.info("Shutdown: delivered %d interrupted-cron-job notice(s)", len(notified))
        return len(notified)

    async def _shutdown_notification_target(self, session_key: str):
        """``(source, platform_str, chat_id, thread_id, profile)``: persisted origin > cached source >
        parsed key. ``profile`` is the owning profile from the source or the ``agent:<profile>:`` key
        namespace (``None`` = default) so the notice leaves through that profile's bot."""
        from gateway.run import _parse_session_key
        source = None
        try:
            if getattr(self, "session_store", None) is not None:
                await self.async_session_store._ensure_loaded()
                entry = self.session_store._entries.get(session_key)
                source = getattr(entry, "origin", None) if entry else None
        except Exception as e:
            logger.debug("Failed to load session origin for shutdown notification %s: %s", session_key, e)
        if source is None:
            source = self._get_cached_session_source(session_key)
        if source is not None:
            return source, source.platform.value, str(source.chat_id), source.thread_id, getattr(source, "profile", None)
        _parsed = _parse_session_key(session_key)
        if not _parsed:
            return None
        return None, _parsed["platform"], _parsed["chat_id"], _parsed.get("thread_id"), _parsed.get("profile")

    async def _send_shutdown_notice(
        self, adapter, chat_id: str, msg: str, kind: str, platform_str: str, **send_kwargs
    ) -> bool:
        """Send one shutdown notice; True when delivered. Failures are debug-logged, never raised."""
        where = "home channel " if kind == "home channel" else ""
        fail_fmt = f"Failed to send shutdown notification to {where}%s:%s: %s"
        if not await self._send_notice_logged(adapter, chat_id, msg, platform_str, fail_fmt, **send_kwargs):
            return False
        logger.info("Sent shutdown notification to %s %s:%s", kind, platform_str, chat_id)
        return True

    @staticmethod
    async def _send_notice_logged(
        adapter, chat_id: str, msg: str, platform_str: str, fail_fmt: str, raise_fmt: Optional[str] = None, **kw
    ) -> bool:
        """``adapter.send`` whose failure is debug-logged as ``fmt % (platform, chat, error)`` — ``fail_fmt``
        for success=False, ``raise_fmt`` (default ``fail_fmt``) for a raise; True only on a delivered send.
        Every shutdown notice races live turns, so it always carries the interim marker (#98432)."""
        from gateway.run import _interim_metadata
        kw["metadata"] = _interim_metadata(kw.get("metadata"))
        try:
            result = await adapter.send(chat_id, msg, **kw)
        except Exception as e:
            logger.debug(raise_fmt or fail_fmt, platform_str, chat_id, e)
            return False
        if _send_failed(result):
            logger.debug(fail_fmt, platform_str, chat_id, _send_error(result))
            return False
        return True

    async def _notify_active_sessions_of_shutdown(self) -> None:
        """Send shutdown/restart notifications to active chats and home channels.

        Called at the start of stop() while adapters are connected; send failures never block shutdown.
        """
        restart_source = self._restart_command_source if self._restart_requested else None
        msg = t("gateway.shutdown.notice_restart" if self._restart_requested else "gateway.shutdown.notice_shutdown")
        restart_key = None
        if restart_source is not None:
            with suppress(Exception):
                restart_key = _notice_target_key(
                    restart_source.platform.value, restart_source.chat_id, restart_source.thread_id
                )
        notified: set[tuple[str, str, Optional[str]]] = set()
        for session_key in self._snapshot_running_agents():
            target = await self._shutdown_notification_target(session_key)
            if target is None:
                continue
            source, platform_str, chat_id, thread_id, profile = target
            dedup_key = _delivery_target_key(platform_str, chat_id, thread_id, profile=profile)
            if dedup_key in notified:
                continue
            try:
                platform = Platform(platform_str)
                # The session's OWN profile's bot (transport ref → profile map), never a bare
                # self.adapters hit: under multiplex that is the default bot, so a secondary session's
                # "Gateway shutting down" would land in the user's chat with the wrong bot.
                adapter = self._delivery_adapter_for(source) if source is not None else None
                if adapter is None:
                    adapter = self._authorization_adapter(platform, profile)
                if not adapter:
                    continue
                if not self._notice_allowed(platform, "active session"):
                    continue
                reply_to_message_id = getattr(source, "message_id", None)
                if reply_to_message_id is None and restart_key == dedup_key:
                    reply_to_message_id = getattr(restart_source, "message_id", None)
                metadata = self._thread_metadata_for_target(
                    platform, chat_id, thread_id, chat_type=getattr(source, "chat_type", None),
                    reply_to_message_id=reply_to_message_id, adapter=adapter,
                )
            except Exception as e:
                logger.debug("Failed to send shutdown notification to %s:%s: %s", platform_str, chat_id, e)
                continue
            # Automatic interrupt diagnostic, resolved under the session's own profile scope (same
            # shape as the stall watcher). The requester's own chat on an in-chat /restart is the
            # requested outcome of that command and is never suppressed.
            async def _send_active(adapter=adapter, chat_id=chat_id, platform_str=platform_str,
                                   metadata=metadata, dedup_key=dedup_key):
                if await self._send_shutdown_notice(
                    adapter, chat_id, msg, "active chat", platform_str, metadata=metadata
                ):
                    notified.add(dedup_key)
            from gateway.warning_notifications import present_notification
            from gateway.run import _async_profile_runtime_scope
            scope = (_async_profile_runtime_scope(self._resolve_profile_home_for_source(source))
                     if source is not None else nullcontext())
            async with scope:
                presented = await present_notification(_send_active, platform=platform, diagnostic=restart_key != dedup_key)
            if not presented:
                notified.add(dedup_key)  # suppressed: latch so the home-channel pass does not re-target it
        if self._restart_requested and restart_source is not None:
            logger.debug("Skipping home-channel shutdown notifications for in-chat restart")
            return
        # A quiet drain (routine fleet auto-update) suppresses ONLY the home-channel broadcast; per-session
        # pings above stay. Current-epoch marker only; a failing check fails toward the louder behaviour.
        with _log_suppressed(logging.DEBUG, "drain_notification_suppressed check failed: %s"):
            from gateway.drain_control import drain_notification_suppressed
            if drain_notification_suppressed():
                logger.info(
                    "Home-channel shutdown broadcast suppressed by drain marker (suppress_notification=true)"
                )
                return
        # EVERY served profile's home channel, through that profile's OWN bot: ``self.adapters`` and
        # ``self.config`` are the launch profile's alone, so iterating them left the secondaries'
        # channels silent (#118233). ``list(...)`` snapshots the adapter maps: adapter.send() can hit
        # a fatal path (_handle_fatal) that pops the adapter -> "dictionary changed size during iteration".
        profile_adapters = getattr(self, "_profile_adapters", None) or {}
        for profile, platform, platform_cfg in list(self._served_home_channel_configs()):
            home = platform_cfg.home_channel
            if not home or not home.chat_id:
                continue
            adapter = (self.adapters if profile is None else profile_adapters.get(profile) or {}).get(platform)
            if adapter is None:
                continue
            if not self._notice_allowed(platform, "home channel", platform_cfg):
                continue
            dedup_key = _delivery_target_key(platform.value, home.chat_id, home.thread_id, profile=profile)
            if dedup_key in notified:
                continue
            try:
                metadata = self._thread_metadata_for_target(platform, home.chat_id, home.thread_id, adapter=adapter)
            except Exception as e:
                logger.debug(
                    "Failed to send shutdown notification to home channel %s:%s: %s", platform.value, home.chat_id, e,
                )
                continue
            async def _send_home(adapter=adapter, home=home, platform=platform, metadata=metadata):
                if await self._send_shutdown_notice(
                    adapter, str(home.chat_id), msg, "home channel", platform.value, metadata=metadata,
                ):
                    notified.add(dedup_key)
            from gateway.warning_notifications import present_notification
            from gateway.run import _async_profile_runtime_scope
            # present_notification reads the ACTIVE profile's display settings: bind the served one's.
            profile_home = (getattr(self, "_served_profile_homes", None) or {}).get(profile) if profile else None
            async with _async_profile_runtime_scope(profile_home) if profile_home else nullcontext():
                await present_notification(_send_home, platform=platform)

    # Agent finalization / resource cleanup
    @staticmethod
    def _flush_agent_transcript_at_shutdown(agent: Any) -> None:
        """Persist an in-flight transcript before teardown.

        A force-interrupted agent may never reach finalize_turn (the only mid-turn flush), so its
        tool rounds would vanish on resume. Idempotent; gracefully finished agents re-flush nothing.
        """
        with _log_suppressed(logging.DEBUG, "Shutdown transcript flush failed: %s"):
            # Persist any in-flight transcript to the SQLite session store before teardown (#13121). An
            # agent forcibly interrupted by the drain-timeout escalation may never reach
            # ``turn_finalizer.finalize_turn`` (the only place that flushes the turn to state.db) — e.g. it
            # was blocked in a tool call that did not abort within the post-interrupt grace window. Its
            # in-flight tool rounds live only in the in-memory ``_session_messages`` (refreshed per tool
            # round in ``conversation_loop`` but never written to SQLite mid-turn), so the immediate
            # pre-restart turn is silently dropped from ``load_transcript()`` on resume. Flushing here
            # closes that gap; the resume_pending / fresh-tool-tail branches in
            # ``_handle_message_with_agent`` already expect a transcript whose tail may be a pending tool
            # result.
            _flush = getattr(agent, "_flush_messages_to_session_db", None)
            _session_messages = getattr(agent, "_session_messages", None)
            if not (callable(_flush) and isinstance(_session_messages, list) and _session_messages):
                return
            # Strip empty-response retry scaffolding from the tail first (as ``_persist_session``
            # does) so a resumed turn doesn't replay synthetic recovery nudges.
            _strip = getattr(agent, "_drop_trailing_empty_response_scaffolding", None)
            if callable(_strip):
                with suppress(Exception):
                    _strip(_session_messages)
            try:
                _flush(_session_messages)
            except Exception as _flush_err:
                # Transcript could not be persisted (e.g. FTS/SQLite corruption): dump the live history
                # to a JSON recovery snapshot rather than lose it. Non-fatal.
                logger.warning(
                    "Shutdown transcript flush failed (%s); preserving %d in-memory message(s) to recovery snapshot",
                    _flush_err, len(_session_messages),
                )
                from gateway.shutdown_flush import flush_agent_history_to_file
                flush_agent_history_to_file(getattr(agent, "session_id", None), _session_messages)

    async def _finalize_shutdown_agents(self, active_agents: Dict[str, Any]) -> None:
        for session_key, agent in active_agents.items():
            self._flush_agent_transcript_at_shutdown(agent)
            # Off-loop + bounded: plugin on_session_finalize hooks can do arbitrary synchronous work
            # (e.g. a full-session trace export) — same hang class as the memory provider below.
            await self._finalize_session_off_loop(
                session_id=getattr(agent, "session_id", None), platform="gateway", reason="shutdown",
                session_key=session_key,
            )
            # Off-loop + bounded: a wedged memory provider here used to hang the whole shutdown so
            # SIGTERM never completed.
            await self._cleanup_agent_resources_off_loop(agent, context="shutdown finalize", session_key=session_key)

    def _should_emit_long_running_notification(
        self, session_key: Optional[str], agent: Any, executor_task: Optional[Any],
    ) -> bool:
        """Emit the heartbeat only while this task still owns the live run (not after ``/new`` rebinds).

        Guards against a stale ``running: delegate_task`` heartbeat outliving the run that started it: stop
        once the executor finishes, the agent is gone, or the session key has been rebound to a different
        live agent (e.g. the user sent ``/new`` and a fresh agent took the slot mid-run, #12029).
        """
        if agent is None or (executor_task is not None and executor_task.done()):
            return False
        # Drain/restart already told the chat the task will be interrupted; a "still working"
        # heartbeat after that notice reads as a contradiction (#10990).
        if getattr(self, "_draining", False) or getattr(self, "_restart_requested", False):
            return False
        if session_key:
            _hb_state = self._peek_session_state(session_key)
            if (_hb_state.turn.agent if _hb_state else None) is not agent:
                return False
        return True

    def _defer_agent_cleanup_until_future_done(self, future: asyncio.Future, agent: Any, *, context: str) -> None:
        """Clean up ``agent`` only after its executor future finishes (it may still use the agent's clients)."""

        async def _cleanup_when_done() -> None:
            try:
                await asyncio.shield(future)
            except asyncio.CancelledError:
                # Loop shutdown can cancel this waiter while the executor still runs. Never turn
                # that cancellation into premature cleanup.
                return
            except Exception as exc:
                logger.debug(
                    "Deferred agent worker%s finished with an error: %s", f" ({context})" if context else "", exc,
                )
            await self._cleanup_agent_resources_off_loop(agent, context=context)

        self._track_deferred_agent_worker(future, agent)
        tasks = getattr(self, "_deferred_agent_cleanup_tasks", None)
        if tasks is None:
            tasks = self._deferred_agent_cleanup_tasks = set()
        self._track_task_in(tasks, asyncio.create_task(_cleanup_when_done()))

    async def _finalize_session_off_loop(
        self, *, session_id: Any, platform: str, reason: str, session_key: Optional[str] = None, **extra: Any,
    ) -> None:
        """Run hermes_cli.lifecycle.finalize_session off-loop, bounded; on timeout the worker is left alone.
        ``session_key`` lets an unscoped caller (shutdown) enter the owning profile's scope: plugin
        ``on_session_finalize`` observers and the Relay coordinator (``current_profile_key``) resolve
        profile state at call time."""

        def _call() -> None:
            from hermes_cli.lifecycle import finalize_session
            finalize_session(session_id=session_id, platform=platform, reason=reason, **extra)

        try:
            await asyncio.wait_for(
                self._run_housekeeping_in_executor(self._run_release_in_profile_scope, _call, (), session_key),
                timeout=self._FINALIZE_TIMEOUT_S,
            )
        except asyncio.TimeoutError:
            logger.warning(
                "Session finalize hooks (%s, reason=%s) exceeded %ss; proceeding without blocking the event loop "
                "(the worker thread is left to finish on its own).", session_id, reason, self._FINALIZE_TIMEOUT_S,
            )
        except Exception as finalize_exc:
            logger.debug("Session finalize hooks (%s, reason=%s) failed: %s", session_id, reason, finalize_exc)

    async def _cleanup_agent_resources_off_loop(
        self, agent: Any, *, context: str = "", session_key: Optional[str] = None,
    ) -> None:
        """Run _cleanup_agent_resources in a worker thread, bounded; on timeout the worker is left alone.

        The teardown fires the memory-provider lifecycle hooks (``flush_pending`` → ``on_session_end`` →
        ``shutdown`` → ``close``), which read credentials/home at call time. In-turn callers carry the
        profile scope through ``_run_housekeeping_in_executor``; shutdown does not (it runs on the main
        loop, outside any adapter handler), so under multiplexing ``on_session_end`` failed closed and the
        session tail was never committed (#110622). ``_run_release_in_profile_scope`` enters the OWNING
        profile's scope from ``session_key`` when the caller has none, exactly like cache eviction."""
        if agent is None:
            return
        if context.startswith("shutdown") or context == "session expiry":
            with suppress(Exception):
                agent._end_session_on_close = False
        ctx_label = f" ({context})" if context else ""
        try:
            await asyncio.wait_for(
                self._run_housekeeping_in_executor(
                    self._run_release_in_profile_scope, self._cleanup_agent_resources, (agent,), session_key,
                ),
                timeout=self._CLEANUP_TIMEOUT_S,
            )
        except asyncio.TimeoutError:
            logger.warning(
                "Agent resource cleanup%s exceeded %ss; proceeding without blocking the event loop (the worker "
                "thread is left to finish on its own). (#53175)", ctx_label, self._CLEANUP_TIMEOUT_S,
            )
        except Exception as cleanup_exc:
            logger.warning("Agent resource cleanup%s failed: %s (#53175)", ctx_label, cleanup_exc)

    def _cleanup_agent_resources(self, agent: Any) -> None:
        """Best-effort cleanup for temporary or cached agent instances."""
        if agent is None:
            return
        with suppress(Exception):
            if hasattr(agent, "shutdown_memory_provider"):
                # Drain queued memory writes BEFORE teardown (shutdown_all() gives the worker only ~5s, so a
                # /reset or rotation could drop them). Bounded; a failure never blocks teardown.
                # The memory manager persists per-turn sync and end-of-session extraction on a single
                # serialized background worker. shutdown_memory_provider() -> shutdown_all() only gives that
                # worker a ~5s bounded drain and abandons (cancels) anything still queued past it, so a
                # /reset — or any gateway session rotation that reaches this cleanup path — could silently
                # drop writes the session had already handed off. The next session then loads stale memory
                # (#73297). Give pending work a bounded head start through the manager's own barrier first,
                # mirroring the CLI exit path (cli.py). Best-effort: a flush failure must never block
                # teardown.
                _mm = getattr(agent, "_memory_manager", None)
                if _mm is not None and hasattr(_mm, "flush_pending"):
                    with suppress(Exception):
                        _mm.flush_pending(timeout=10)
                # Pass the real transcript so ``on_session_end`` hooks don't see the empty default.
                # ``_session_messages`` may be absent on ``object.__new__`` test stubs, hence getattr.
                # ``_session_messages`` is set on ``AIAgent`` (run_agent.py:1518) and refreshed at the end
                # of every ``run_conversation`` turn via ``_persist_session``; on an agent built through
                # ``object.__new__`` (test stubs) the attribute may be absent, so ``getattr`` with a
                # ``None`` default keeps the call signature-compatible with the pre-fix behaviour
                # (``shutdown_memory_provider(messages=None)``). See #15165.
                session_messages = getattr(agent, "_session_messages", None)
                if isinstance(session_messages, list):
                    agent.shutdown_memory_provider(session_messages)
                else:
                    agent.shutdown_memory_provider()
        # Close tool resources (sandboxes, browser daemons, background processes, httpx clients).
        with suppress(Exception):
            if hasattr(agent, "close"):
                agent.close()
        # Auxiliary async clients live in a process-global cache created from worker threads; drop
        # entries whose event loop is dead so httpx transports don't accumulate across turns.
        with suppress(Exception):
            from agent.auxiliary_client import cleanup_stale_async_clients
            cleanup_stale_async_clients()

    # Stuck-loop (restart failure) counters
    def _stuck_loop_counts_path(self) -> Path:
        from gateway.run import _hermes_home
        return _hermes_home / self._STUCK_LOOP_FILE

    @staticmethod
    def _read_json_counts(path: Path) -> Optional[dict]:
        """Parsed counter dict, or None when the file is missing/unreadable (no exists() pre-check needed)."""
        try:
            return json.loads(path.read_text(encoding="utf-8-sig"))
        except Exception:
            return None

    def _increment_restart_failure_counts(self, active_session_keys: set) -> None:
        """Increment persisted restart-failure counters for active sessions; drop the rest (loop broken)."""
        from utils import atomic_json_write
        path = self._stuck_loop_counts_path()
        counts = self._read_json_counts(path) or {}
        with suppress(Exception):
            atomic_json_write(path, {key: counts.get(key, 0) + 1 for key in active_session_keys}, indent=None)

    def _suspend_stuck_loop_sessions(self) -> int:
        """Suspend sessions active across too many restarts (startup, AFTER crash-turn recovery)."""
        path = self._stuck_loop_counts_path()
        if not path.exists():
            return 0
        counts = self._read_json_counts(path)
        if counts is None:
            return 0
        suspended = 0
        for session_key in [k for k, v in counts.items() if v >= self._STUCK_LOOP_THRESHOLD]:
            with suppress(Exception):
                entry = self.session_store._entries.get(session_key)
                if entry and not entry.suspended:
                    entry.suspended = True
                    suspended += 1
                    logger.warning(
                        "Auto-suspended stuck session %s (active across %d consecutive restarts — likely a stuck loop)",
                        session_key, counts[session_key],
                    )
        if suspended:
            with suppress(Exception):
                self.session_store._save()
        # Clear the file — counters start fresh after suspension
        with suppress(Exception):
            path.unlink(missing_ok=True)
        return suspended

    async def _clear_restart_failure_count(self, session_key: str) -> None:
        """Clear a completed session's restart-failure counter off-loop (atomic_json_write fsyncs)."""
        from utils import atomic_json_write
        path = self._stuck_loop_counts_path()
        if not path.exists():
            return
        # The whole read/mutate/write is guarded (as on main): a corrupt counters file
        # (non-dict JSON) must never raise out of a session-completion path.
        try:
            counts = self._read_json_counts(path) or {}
            if session_key in counts:
                del counts[session_key]
                if counts:
                    await asyncio.to_thread(atomic_json_write, path, counts, indent=None)
                else:
                    path.unlink(missing_ok=True)
        except Exception:
            pass

    # Restart orchestration
    @staticmethod
    def _restart_watcher_env() -> dict:
        """Watcher env minus ``_HERMES_GATEWAY`` (else the CLI's self-restart guard refuses; gateway stays down).

        The host multiplexer is respawned with ``host_gateway_child_env`` (default-root
        secrets via ``served_profile_child_env``, not ``os.environ.copy()``). A standalone
        named-profile gateway keeps that profile's home — only a multiplexer, or a process
        already on the default root, is the host.
        """
        from gateway.config_loader import drop_bridged_env
        from hermes_constants import get_default_hermes_root, get_hermes_home
        from tools.environments.local import host_gateway_child_env, served_profile_child_env

        home = get_hermes_home()
        try:
            on_default = home.resolve() == get_default_hermes_root().resolve()
        except Exception:
            on_default = False
        # ``resolve_multiplex_mode`` settles the default-on/unset decision before
        # restart. Carry that runtime identity instead of re-reading raw config:
        # ``None`` is the normal pre-resolution value for a named launcher.
        from agent.secret_scope import is_multiplex_active
        settled_multiplex = is_multiplex_active()
        multiplex = False
        if not on_default and not settled_multiplex:
            # Second settled source: the live host gateway's OWN published record
            # (its settled served set). Only when NO settled identity exists may the
            # raw config re-read stand — it reads the UNSET flag as False, which is
            # wrong exactly when this process IS the default-on host (#120305).
            try:
                from gateway import host_rendezvous as hr
                record = hr.read_record(hr.ROLE_GATEWAY)
                if record is not None and hr.liveness_is_proven(record) and len(record.profiles) > 1:
                    multiplex = True
            except Exception:
                multiplex = False
        if not on_default and not settled_multiplex and not multiplex:
            try:
                from gateway.config import load_gateway_config
                multiplex = bool(load_gateway_config().multiplex_profiles)
            except Exception:
                multiplex = False
        if on_default or settled_multiplex or multiplex:
            watcher_env = host_gateway_child_env()
        else:
            watcher_env = served_profile_child_env(
                target_home=home, inherit_credentials=True,
            )
        watcher_env = drop_bridged_env(watcher_env)
        watcher_env.pop("_HERMES_GATEWAY", None)
        return watcher_env

    @staticmethod
    def _spawn_windows_restart_watcher(hermes_cmd: list, current_pid: int, restart_after_s: float) -> None:
        """Spawn the detached Windows watcher (``python -c``), retrying once without job breakaway."""
        import subprocess
        from hermes_cli._subprocess_compat import (
            windows_detach_flags_without_breakaway, windows_detach_popen_kwargs
        )
        watcher_env = GatewayShutdownMixin._restart_watcher_env()
        # host_gateway_child_env does not copy the parent dotenv. The watcher
        # still has to run inside the venv this process is using, or the
        # respawn cannot import hermes.
        if not watcher_env.get("VIRTUAL_ENV"):
            inherited = os.environ.get("VIRTUAL_ENV")
            if inherited:
                watcher_env["VIRTUAL_ENV"] = inherited
        project_root = Path(__file__).resolve().parent.parent
        # Console python under CREATE_NO_WINDOW: nothing flashes. NOT pythonw.exe — a console-less
        # watcher makes every console-subsystem descendant allocate a visible conhost (#54220/#56747).
        # The watcher runs sys.executable (console python) under the CREATE_NO_WINDOW detach kwargs below:
        # it owns one hidden console, inherited by the `hermes gateway restart` child, so nothing flashes.
        # See #54220, #56747.
        from hermes_cli._launchers import runtime_command
        watcher_argv = runtime_command(project_root,
            [str(current_pid), str(restart_after_s), *hermes_cmd, "gateway", "restart"],
            code=_WINDOWS_RESTART_WATCHER)
        watcher_python = watcher_argv[0]
        popen_kwargs = dict(stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, env=watcher_env)
        # Break away from the parent CLI's job object or be reaped when the CLI exits; a job without
        # BREAKAWAY_OK rejects CREATE_BREAKAWAY_FROM_JOB (OSError) — retry once without the bit.
        try:
            subprocess.Popen(watcher_argv, **popen_kwargs, **windows_detach_popen_kwargs())
        except OSError:
            try:
                subprocess.Popen(
                    watcher_argv, **popen_kwargs, creationflags=windows_detach_flags_without_breakaway(),
                )
            except OSError as exc:
                # Both spawns failed. Log only the interpreter basename and numeric errno — never
                # argv, env, watcher source, or str(exc) (may carry a full path) — and return.
                winerror = getattr(exc, "winerror", None)
                logger.warning(
                    "Detached restart watcher was not started after the "
                    "no-breakaway retry (%s; %s=%r). The gateway will not "
                    "be respawned by this restart attempt.", os.path.basename(watcher_python),
                    "winerror" if winerror is not None else "errno",
                    winerror if winerror is not None else exc.errno,
                )

    async def _launch_detached_restart_command(self) -> None:
        from gateway.run import _resolve_hermes_bin
        import shutil
        import subprocess
        hermes_cmd = _resolve_hermes_bin()
        if not hermes_cmd:
            logger.error("Could not locate hermes binary for detached /restart")
            return
        if self._detached_restart_helper_started:
            return
        self._detached_restart_helper_started = True
        current_pid = os.getpid()
        restart_after_s = max(float(getattr(self, "_restart_drain_timeout", 0.0) or 0.0) + 5.0, 5.0)
        if sys.platform == "win32":
            GatewayShutdownMixin._spawn_windows_restart_watcher(hermes_cmd, current_pid, restart_after_s)
            return
        cmd = " ".join(shlex.quote(part) for part in hermes_cmd)
        shell_cmd = (
            f"deadline=$(( $(date +%s) + {int(restart_after_s)} )); "
            f"while kill -0 {current_pid} 2>/dev/null && [ $(date +%s) -lt $deadline ]; do sleep 0.2; done; "
            f"{cmd} gateway restart"
        )
        setsid_bin = shutil.which("setsid")
        argv = [setsid_bin, "bash", "-lc", shell_cmd] if setsid_bin else ["bash", "-lc", shell_cmd]
        subprocess.Popen(
            argv, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
            env=GatewayShutdownMixin._restart_watcher_env(), start_new_session=True,
        )

    def _wedged_agent_count(self) -> int:
        """Work units the restart wait may skip: chat agents idle past ``agent.gateway_timeout`` and
        cron runs older than the scheduler's stale-inflight allowance (#115469).

        API work has no activity clock and pending sentinels are brand-new, so neither counts.
        """
        return self._wedged_chat_agent_count() + self._wedged_cron_job_count()

    def _restart_wait_cron_counts(self) -> dict:
        """``cron.scheduler.get_restart_wait_cron_counts``, counted per profile-scoped run.

        Fail-soft toward waiting: if the split can't be read, every active cron run stays
        awaitable and nothing is excluded.
        """
        try:
            from cron.scheduler import get_restart_wait_cron_counts
            return get_restart_wait_cron_counts()
        except Exception:
            return {"awaitable": self._active_cron_job_count(), "wedged": 0, "restart_safe": 0}

    def _wedged_cron_job_count(self) -> int:
        """Cron runs past ``cron.scheduler.get_wedged_job_ids``'s allowance; 0 if cron can't import."""
        return self._restart_wait_cron_counts()["wedged"]

    def _restart_safe_cron_count(self) -> int:
        """Cron runs whose worker owns a restart-safe systemd scope; 0 if cron can't import.

        Such a worker runs outside the gateway cgroup, so neither the tool-process sweep nor
        ``mark_running_jobs_interrupted`` reaches it, and its final send rides the durable delivery
        queue for whichever gateway is live next. The restart after-turn wait must therefore not
        hold the gateway in ``draining`` for it: waiting buys the run nothing and refuses new turns
        for up to the whole cap (observed live: a 100-minute job held the gateway ~30 minutes).
        Degraded (no user bus) workers are NOT in this set — they share the cgroup and die with a
        systemd stop, so they keep holding the wait.
        """
        return self._restart_wait_cron_counts()["restart_safe"]

    def _wedged_chat_agent_count(self) -> int:
        """Running chat agents with no activity for ``agent.gateway_timeout`` (0 when disabled);
        an unreadable activity summary means "not wedged".
        """
        from gateway.run import _AGENT_PENDING_SENTINEL, _float_env
        timeout = _float_env("HERMES_AGENT_TIMEOUT", 1800)
        if timeout <= 0:
            return 0

        def _idle_seconds(agent: Any) -> Optional[float]:
            summary_fn = getattr(agent, "get_activity_summary", None)
            if not callable(summary_fn):
                return None
            try:
                summary = summary_fn()
                return float(summary.get("seconds_since_activity", 0.0)) if isinstance(summary, dict) else None
            except Exception:
                return None

        return sum(
            1
            for agent in list((getattr(self, "_running_agents", None) or {}).values())
            if agent is not None and agent is not _AGENT_PENDING_SENTINEL
            and (idle := _idle_seconds(agent)) is not None and idle >= timeout
        )

    def _awaitable_work_count(self) -> int:
        """Active work minus the units the restart wait must not hold for.

        Two disjoint exclusions: wedged turns (idle past ``agent.gateway_timeout`` / past the cron
        in-flight allowance — restart is their remedy, #115469) and cron runs executing in a
        restart-safe external worker (``_restart_safe_cron_count``), which outlives this process
        either way. Cron runs are counted per profile-scoped run by the scheduler, not by
        subtracting from the bare-ID active count: two profiles can run the same job ID, and one
        excluded run must not hide the other.
        """
        non_cron = (
            self._running_agent_count()
            + self._active_api_run_count()
            + self._active_deferred_agent_worker_count()
        )
        return (
            max(0, non_cron - self._wedged_chat_agent_count())
            + self._restart_wait_cron_counts()["awaitable"]
        )

    def _describe_active_work(self) -> list:
        """One dict per in-flight work unit the restart wait is holding for, so an observer
        (``hermes update``, ``hermes gateway status``) can name it instead of printing a bare count.

        ``kind`` ∈ ``chat`` (session turn), ``cron`` (job id + external worker pid when the run was
        handed to a restart-safe scope), ``api`` / ``deferred`` (count only — those sources expose
        no identity). Best-effort: a source that can't be read is omitted, never raises.
        """
        from gateway.run import _AGENT_PENDING_SENTINEL
        now = time.time()
        units: list = []
        for key, state in list(self._sessions_map().items()):
            agent = state.turn.agent
            if agent is None:
                continue
            unit: dict = {"kind": "chat", "session": key, "pid": os.getpid()}
            if state.turn.started_ts:
                unit["elapsed_s"] = round(now - state.turn.started_ts, 1)
            if agent is not _AGENT_PENDING_SENTINEL:
                unit["model"] = getattr(agent, "model", None)
                summary_fn = getattr(agent, "get_activity_summary", None)
                if callable(summary_fn):
                    with suppress(Exception):
                        summary = summary_fn()
                        unit["current_tool"] = summary.get("current_tool")
                        unit["idle_s"] = summary.get("seconds_since_activity")
            units.append(unit)
        with suppress(Exception):
            from cron.scheduler import get_running_job_details, get_wedged_job_ids
            wedged = get_wedged_job_ids()
            for job in get_running_job_details():
                units.append({"kind": "cron", "job_id": job["job_id"], "elapsed_s": job["elapsed_s"],
                              "pid": job["worker_pid"] or os.getpid(), "external": bool(job["worker_pid"]),
                              "wedged": job["job_id"] in wedged,
                              "restart_safe": bool(job.get("restart_safe"))})
        for kind, count in (("api", self._active_api_run_count()), ("deferred", self._active_deferred_agent_worker_count())):
            units.extend({"kind": kind, "pid": os.getpid()} for _ in range(count))
        return units

    async def _await_active_work_before_restart(self) -> 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()``).
        """
        active = self._active_work_count()
        if active <= 0:
            return True
        if self._awaitable_work_count() <= 0:
            logger.warning(
                "Restart requested with %d active work unit(s), none awaitable "
                "(%d wedged past the inactivity timeout, %d in restart-safe external cron "
                "workers that outlive this process); skipping the after-turn wait and "
                "proceeding to stop()/drain", active,
                self._wedged_agent_count(), self._restart_safe_cron_count(),
            )
            return False
        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,
            )
            return False
        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,
        )
        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:
            now = loop.time()
            if now >= deadline:
                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(),
                )
                return False
            if (now - last_status_at) >= 30.0:
                logger.info(
                    "Restart deferred: waiting on %d active work unit(s) "
                    "(%d wedged and excluded, %d restart-safe and excluded; "
                    "%.0fs remaining before force drain): %s",
                    self._awaitable_work_count(), self._wedged_agent_count(),
                    self._restart_safe_cron_count(), deadline - now,
                    self._describe_active_work(),
                )
                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:
            logger.warning(
                "Restart deferred wait: %d excluded work unit(s) remain "
                "(%d wedged, %d restart-safe external cron); proceeding to stop()/drain",
                self._active_work_count(), self._wedged_agent_count(),
                self._restart_safe_cron_count(),
            )
            return False
        logger.info("Restart deferred wait complete — active work drained; proceeding to stop()")
        return True

    def request_restart(self, *, detached: bool = False, via_service: bool = False) -> bool:
        if self._restart_task_started:
            return False
        self._restart_requested = True
        self._restart_detached = detached
        self._restart_via_service = via_service
        self._restart_task_started = True
        # Refuse new turns; keep ``_running`` True so the active turn can still deliver its final response.
        self._draining = True
        # The restart's after-turn wait is a drain window too: pollers of GET /v1/runs/{id} must see
        # the boundary from the moment new turns are refused, not only once stop() begins (#115133).
        self._mark_api_runs_shutdown_requested()

        async def _run_restart() -> None:
            await self._await_active_work_before_restart()
            # 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"):
                    await self._launch_detached_restart_command()
            await asyncio.sleep(0.05)
            await self.stop(restart=True, detached_restart=detached, service_restart=via_service)

        # NOT in _background_tasks: _stop_impl cancels those, which would skip _shutdown_event.set() / exit 75.
        # _run_restart is a short-lived self-terminating task (calls stop() then returns). Don't add it to
        # _background_tasks — _stop_impl cancels all entries in that set, which would cancel _run_restart
        # while it's awaiting _stop_task, propagating CancelledError into _stop_impl and preventing
        # _shutdown_event.set() / _exit_code = 75. See #12875. We still hold a strong reference in
        # self._restart_task: a bare asyncio.create_task() keeps only a weak reference, so the event loop
        # may garbage-collect a still-pending task mid-flight. The cancel loop in _stop_impl explicitly
        # skips _restart_task for the same reason it skips _stop_task.
        # Empty Context: /restart is handled inside the requester's profile scope, and a copied context
        # would run the HOST restart as that profile (watcher HERMES_HOME, stop()'s flushes).
        self._restart_task = Context().run(lambda: asyncio.create_task(_run_restart()))
        return True

    def _start_systemd_watchdog(self) -> bool:
        """Start sd_notify only after a configured gateway is truly running."""
        if not self._running or self.config.systemd_watchdog_seconds <= 0:
            return False
        if self._systemd_watchdog is not None:
            return True
        from gateway.systemd_notify import SystemdWatchdog
        watchdog = SystemdWatchdog(config_enabled=True)
        if not watchdog.start():
            return False
        self._systemd_watchdog = watchdog
        watchdog.ready("Hermes Gateway running")
        return True

    async def _stop_systemd_watchdog(self) -> None:
        """Stop heartbeats before any potentially long shutdown drain."""
        watchdog = self._systemd_watchdog
        if watchdog is None:
            return
        self._systemd_watchdog = None
        await watchdog.stop()

    # stop() phases. Invoked as ``GatewayRunner._stop_<phase>(self, ctx)`` so shutdown-path tests
    # can drive them from bare doubles that are not GatewayRunner instances.
    @staticmethod
    def _quiet_step(label: str, fn: Callable[[], Any]) -> Any:
        """Run one best-effort teardown step; a failure is debug-logged as ``"<label>: <exc>"``."""
        try:
            return fn()
        except Exception as _e:
            logger.debug("%s: %s", label, _e)
            return None

    @staticmethod
    def _stop_kill_tool_subprocesses(phase: str) -> list:
        """Kill tool subprocesses + terminal envs + browsers; returns cron job IDs marked interrupted.

        Called twice: after a drain timeout (reclaim children before systemd SIGKILLs) and as a final
        catch-all. Best-effort; one failing subsystem cannot block the rest.
        """

        def _step(label: str, fn: Callable[[], Any]) -> Any:
            return GatewayShutdownMixin._quiet_step(f"{label} ({phase}) error", fn)

        def _count_step(fmt: str, fn: Callable[[], int]) -> None:
            n = fn()
            if n:
                logger.info(fmt, phase, n)

        def _kill_processes() -> None:
            from tools.process_registry import process_registry
            # Host shutdown: kill even persist_on_release jobs or they become
            # PPID=1 orphans (#41225/#46778); an explicit source reaches them.
            _count_step(
                "Shutdown (%s): killed %d tool subprocess(es)",
                lambda: process_registry.kill_all(source="gateway_shutdown"))

        def _mark_cron_interrupted() -> list:
            # kill_all() is global: a cron job mid-dispatch lost its tool subprocess and its agent thread may
            # still emit a plausible response from truncated output — mark it interrupted, never success.
            # Any cron job still dispatched at this instant just had its tool subprocess killed above
            # (kill_all() has no per-job-ID targeting — it's a global sweep). No-op when no cron job is in
            # flight. See #60432.
            from cron.scheduler import mark_running_jobs_interrupted
            _interrupted = mark_running_jobs_interrupted(
                f"Gateway shutdown ({phase}) killed the job's tool subprocess before the run finished."
            )
            if _interrupted:
                logger.warning(
                    "Shutdown (%s): marked %d in-flight cron job(s) interrupted: %s",
                    phase, len(_interrupted), ", ".join(_interrupted),
                )
            return _interrupted

        def _interrupt_delegations() -> None:
            from tools.async_delegation import interrupt_all as _interrupt_async
            _count_step(
                "Shutdown (%s): interrupted %d background delegation(s)",
                lambda: _interrupt_async(reason=f"gateway shutdown ({phase})"),
            )

        _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)
        def _cleanup_environments() -> None:
            from tools.terminal_tool_lifecycle import cleanup_all_environments
            cleanup_all_environments()

        def _cleanup_browsers() -> None:
            from tools.browser_tool_lifecycle import cleanup_all_browsers
            cleanup_all_browsers()

        _step("cleanup_all_environments", _cleanup_environments)
        _step("cleanup_all_browsers", _cleanup_browsers)
        return _marked_cron_jobs

    @staticmethod
    async def _stop_kill_tool_subprocesses_off_loop(phase: str) -> list:
        """Run _stop_kill_tool_subprocesses in a worker thread; returns cron job IDs marked interrupted.

        ``kill_all`` fans out into per-target ``kill_process`` calls that do blocking work
        (registry checkpoint disk I/O, ``subprocess.run`` for systemd scopes, sandbox exec),
        so running the sweep inline would monopolize the gateway event loop (#116327).
        Offloaded with ``asyncio.to_thread`` — the loop's default executor, deliberately NOT
        the gateway-owned ``self._executor``, which ``_stop_quiesce_and_close_session_dbs``
        drains right after this phase. Phase order is preserved: callers await this before
        cron notices / adapter teardown. If the surrounding stop task is cancelled while the
        worker runs, the thread is left to finish on its own; the thread-based shutdown
        watchdog remains the hard backstop.
        """
        return await asyncio.to_thread(
            GatewayShutdownMixin._stop_kill_tool_subprocesses, phase
        )

    async def _stop_begin_teardown(self, ctx: "GatewayShutdownMixin._StopContext") -> None:
        """Flag teardown, stop room worker/watchdog, notify sessions."""
        logger.info("Stopping gateway%s...", " for restart" if self._restart_requested else "")
        ctx.started_at = time.monotonic()
        self._running = False
        self._clear_plugin_message_injector()
        self._draining = True
        self._mark_api_runs_shutdown_requested()
        # getattr-guards: shutdown-path test doubles may lack the room worker / systemd watchdog.
        stop_room_worker = getattr(self, "_stop_hosted_room_worker", None)
        if callable(stop_room_worker):
            try:
                if not await stop_room_worker(timeout=5.0):
                    logger.warning(
                        "Group Chat worker is still settling durable work; the next gateway start will recover it"
                    )
            except Exception:
                logger.warning(
                    "Group Chat worker could not stop cleanly; the next gateway start will recover durable work",
                    exc_info=True,
                )
        stop_watchdog = getattr(self, "_stop_systemd_watchdog", None)
        if callable(stop_watchdog):
            await stop_watchdog()
        await self._cancel_secondary_profile_reconnect_tasks()
        # Notify all chats with active agents BEFORE draining — adapters are still connected here.
        await self._notify_active_sessions_of_shutdown()
        logger.info("Shutdown phase: notify_active_sessions done at +%.2fs", ctx.elapsed())

    async def _stop_drain_active_work(self, timeout: float, ctx: "GatewayShutdownMixin._StopContext") -> None:
        """Pre-mark resume_pending, drain agents/cron/API work into ``ctx``."""
        from gateway.run import GatewayRunner
        # Pre-mark resume_pending BEFORE the drain so a mid-drain SIGKILL still leaves a durable marker.
        _pre_drain_keys = await GatewayRunner._mark_running_sessions_resume_pending(
            self, "pre-drain mark_resume_pending"
        )
        _cron_at_start = self._active_cron_job_count()
        _api_at_start = self._active_api_run_count()
        _deferred_at_start = ctx.deferred_count()
        # Cron floor clamped to the watchdog leash; getattr-guard for bare shutdown-path doubles.
        _cron_drain_cfg = getattr(self, "_cron_drain_timeout", DEFAULT_GATEWAY_CRON_DRAIN_TIMEOUT)
        # Under launchd the real leash is launchd's own exit timeout, not our watchdog
        # (drain + grace): a signal-driven stop that lets cron work push past it is SIGKILLed
        # before cleanup runs.
        # ``timeout`` is already the effective (launchd-capped) drain, so this is the same
        # leash the thread watchdog is armed with — dump margin included.
        _cron_leash = effective_stop_watchdog_delay(self, resolve_shutdown_watchdog_delay(timeout))
        _cron_timeout = resolve_cron_drain_budget(
            timeout, _cron_drain_cfg, watchdog_delay=_cron_leash, elapsed=ctx.elapsed(),
        )
        if _cron_at_start and _cron_timeout > timeout:
            logger.info(
                "Shutdown drain: %d in-flight cron job(s) — waiting up to "
                "%.0fs for them (cron_drain_timeout=%.0fs, restart_drain_timeout=%.0fs)",
                _cron_at_start, _cron_timeout, _cron_drain_cfg, timeout,
            )
        _drain_started_at = time.monotonic()
        ctx.active_agents, ctx.timed_out = await self._drain_active_agents(timeout, _cron_timeout)
        ctx.drain_elapsed = time.monotonic() - _drain_started_at
        logger.info(
            "Shutdown phase: drain done at +%.2fs (drain took %.2fs, timed_out=%s, active_at_start=%d, "
            "active_now=%d, cron_at_start=%d, cron_now=%d, api_at_start=%d, api_now=%d, "
            "deferred_at_start=%d, deferred_now=%d)", ctx.elapsed(), ctx.drain_elapsed,
            ctx.timed_out, len(ctx.active_agents), self._running_agent_count(), _cron_at_start,
            self._active_cron_job_count(), _api_at_start, self._active_api_run_count(),
            _deferred_at_start, ctx.deferred_count(),
        )
        if ctx.timed_out:
            return
        # Graceful drain: clear the pre-drain resume_pending markers so sessions that finished
        # during the drain window don't carry a stale flag.
        for _sk in _pre_drain_keys:
            if _sk not in self._running_agents:
                try:
                    await self.async_session_store.clear_resume_pending(_sk)
                except Exception as _e:
                    logger.debug("clear_resume_pending after drain failed for %s: %s", _sk, _e)

    async def _stop_interrupt_remaining_work(self, ctx: "GatewayShutdownMixin._StopContext") -> None:
        """Drain timed out: mark resume_pending, interrupt, settle, kill tool subprocesses, notify cron."""
        from gateway.run import GatewayRunner
        logger.warning(
            "Gateway drain timed out after %.1fs with %d active agent(s), "
            "%d in-flight cron job(s), %d api_server run(s), and %d deferred agent worker(s); "
            "interrupting remaining work.", ctx.drain_elapsed, self._running_agent_count(),
            self._active_cron_job_count(), self._active_api_run_count(), ctx.deferred_count(),
        )
        # Mark resume_pending BEFORE interrupting so the next message auto-resumes (stuck sessions
        # still escalate via .restart_failure_counts). CURRENT _running_agents, not the drain snapshot.
        await GatewayRunner._mark_running_sessions_resume_pending(self, "mark_resume_pending")
        reason = GatewayRunner._shutdown_interrupt_reason(self)
        self._interrupt_running_agents(reason)
        interrupt_grace_timeout = GatewayRunner._post_interrupt_grace_timeout(self)
        loop = asyncio.get_running_loop()
        interrupt_deadline = loop.time() + interrupt_grace_timeout
        logger.info("Shutdown phase: allowing %.1fs for interrupted agents to unwind", interrupt_grace_timeout)

        def _work_live() -> bool:
            return bool(self._running_agents or self._active_api_run_count() or ctx.deferred_count())

        # Wait on API-server work too, or an API turn's tool subprocesses are killed before it unwinds.
        while _work_live() and loop.time() < interrupt_deadline:
            self._update_runtime_status("draining")
            await asyncio.sleep(0.1)
        # Work can materialize AFTER the one-shot interrupt (/v1/runs registers on _create_agent return;
        # pending sentinels promote later). Re-signal for a cooperative interrupt, not a bare kill.
        if _work_live():
            self._interrupt_running_agents(reason)
            logger.debug("Re-signaled interrupt for work still live at settle-window exit")
        # Kill tool subprocesses NOW: deferring past adapter/DB teardown risks the systemd cgroup SIGKILL.
        # Off-loop: the sweep does blocking kills that must not monopolize the event loop (#116327).
        _interrupted_cron_jobs = await GatewayRunner._stop_kill_tool_subprocesses_off_loop("post-interrupt")
        logger.info("Shutdown phase: post-interrupt tool kill done at +%.2fs", ctx.elapsed())
        # Last window with the transport up (the cron worker's own notice arrives after teardown).
        with _log_suppressed(logging.DEBUG, "Cron interrupt notification failed: %s"):
            # The cron worker whose run we just killed will try to deliver its own "interrupted" notice, but
            # it gets there after the adapter teardown below and the message is lost (#82232).
            await self._notify_interrupted_cron_jobs(_interrupted_cron_jobs)
        logger.info("Shutdown phase: cron interrupt notices done at +%.2fs", ctx.elapsed())

    async def _stop_finalize_agents_and_adapters(self, ctx: "GatewayShutdownMixin._StopContext") -> None:
        """Detached restart launch, agent finalization, idle-cache cleanup, adapter teardown."""
        if self._restart_requested and self._restart_detached:
            with _log_suppressed(logging.ERROR, "Failed to launch detached gateway restart: %s"):
                await self._launch_detached_restart_command()
        await self._finalize_shutdown_agents(ctx.active_agents)
        # Idle cached agents too: their MemoryProviders may never have seen on_session_end().
        _cache_lock = getattr(self, "_agent_cache_lock", None)
        _cache = getattr(self, "_agent_cache", None)
        if _cache_lock is not None and _cache is not None:
            with _cache_lock:
                _idle_agents = list(_cache.items())
                _cache.clear()
            for _key, _entry in _idle_agents:
                # Bounded + off-loop: a wedged memory provider here once made SIGTERM hang forever.
                await self._cleanup_agent_resources_off_loop(
                    _entry[0] if isinstance(_entry, tuple) else _entry, context="shutdown idle-cache",
                    session_key=_key,
                )
        # Settle completion flush tasks while adapters are alive so every watcher gets a retryable result.
        cancel_completion_batches = getattr(self, "_cancel_process_completion_batch_tasks", None)
        if cancel_completion_batches is not None:
            await cancel_completion_batches()
        for platform, adapter in list(self.adapters.items()):
            await self._bounded_adapter_teardown(adapter, platform)
        # Disconnect secondary-profile adapters (multiplex mode).
        _profile_adapters = getattr(self, "_profile_adapters", {})
        for _prof, _amap in list(_profile_adapters.items()):
            for platform, adapter in list(_amap.items()):
                await self._bounded_adapter_teardown(adapter, platform, profile=_prof)
            _amap.clear()
        _profile_adapters.clear()
        logger.info("Shutdown phase: all adapters disconnected at +%.2fs", ctx.elapsed())

    async def _stop_release_runtime_state(self, ctx: "GatewayShutdownMixin._StopContext") -> None:
        """Cancel background tasks, flush pending messages, clear per-session state, final tool kill."""
        from gateway.run import GatewayRunner
        for _task in list(self._background_tasks):
            # _restart_task awaits _stop_task: cancelling it would tunnel into _stop_impl and skip _shutdown_event.set().
            if _task is self._stop_task or _task is self._restart_task:
                continue
            _task.cancel()
        # The restart orchestration task is awaiting _stop_task right now; cancelling it would propagate
        # CancelledError into this _stop_impl and skip _shutdown_event.set() / _exit_code = 75 (#12875). It
        # self-terminates anyway.
        self._background_tasks.clear()
        ctx.api_live = self._active_api_run_count()
        self.adapters.clear()
        for _session_key in list(self._running_agents):
            self._release_running_agent_state(_session_key)
        # Flush pending messages before clearing: under FTS5 corruption they are the only surviving copy.
        with suppress(Exception):
            from gateway.shutdown_flush import flush_pending_to_file
            flush_pending_to_file(dict(self._pending_messages), reason="shutdown")
        # The overflow FIFO tail lives in SessionState.conversation.queued_events — flush it too.
        with suppress(Exception):
            from gateway.shutdown_flush import flush_overflow_to_file
            flush_overflow_to_file(
                {_k: list(_v) for _k, _v in dict(getattr(self, "_queued_events", None) or {}).items() if _v},
                reason="shutdown",
            )
        # Live SessionState views: clear() resets one field per session (never a wholesale dict swap).
        self._running_agents.clear()
        self._running_agents_ts.clear()
        self._pending_messages.clear()
        self._pending_approvals.clear()
        for _attr in ("_active_session_leases", "_busy_ack_ts"):  # absent on bare shutdown-path doubles
            if hasattr(self, _attr):
                getattr(self, _attr).clear()
        self._shutdown_event.set()
        # Global catch-all subprocess kill (safe to repeat) for the graceful path and late respawns.
        # Off-loop: same blocking sweep as the post-interrupt kill (#116327).
        await GatewayRunner._stop_kill_tool_subprocesses_off_loop("final-cleanup")
        logger.info("Shutdown phase: final-cleanup tool kill done at +%.2fs", ctx.elapsed())
        # Reap the auxiliary-client cache: clients bound to dead worker-thread loops leak httpx transports.
        def _reap_aux_clients() -> None:
            # Reap the process-global auxiliary-client cache once at the very end of teardown. Per-turn
            # cleanup runs in _cleanup_agent_resources for each active agent, but clients bound to
            # worker-thread loops that died with their ThreadPoolExecutor (notably cron ticks) only get
            # swept here. Without this, long-running gateways accumulate async httpx transports until they
            # hit EMFILE on macOS's default RLIMIT_NOFILE=256. See #14210.
            from agent.auxiliary_client import shutdown_cached_clients
            shutdown_cached_clients()

        GatewayShutdownMixin._quiet_step("shutdown_cached_clients error", _reap_aux_clients)

    def _stop_quiesce_and_close_session_dbs(self, timeout: float, ctx: "GatewayShutdownMixin._StopContext") -> None:
        """Quiesce the executor, then close SessionDB handles only if no worker is still live."""
        from gateway.run import GatewayRunner, _EXECUTOR_QUIESCE_TIMEOUT
        # Quiesce the thread pool BEFORE closing session DBs: a late executor write after
        # SessionDB.close() checkpointed the WAL reopens the handle and splits the WAL generation
        # (close-time corruption). Clamped to the remaining watchdog leash minus 1s for the close.
        # This used to run *after* the close block below, which left two holes: (a) `_executor_closing` was
        # still False during the close, so any coroutine reaching `_run_in_executor_with_context` minted a
        # brand-new pool and ran more blocking DB work against handles that had just been closed; (b)
        # cancelling `self._background_tasks` above does not stop a `run_in_executor` future that already
        # started — the task dies, the worker thread keeps writing. Either way a write lands after
        # `SessionDB.close()`, which has already checkpointed the WAL and let SQLite unlink the sidecar. The
        # late write silently reopens the handle (#94736) and mints a fresh WAL generation behind that
        # checkpoint, so teardown checkpoints the same file a second time from a connection the shutdown log
        # never accounts for — the close-time page-write damage in #101093 and the split WAL generation in
        # #101064. The wait is bounded and clamped to what is left of the shutdown watchdog leash (minus a
        # second for the close itself), so a stuck worker can never cost us the post-close cleanup window
        # (#82161).
        _exec_quiesce_budget = max(
            0.0, min(_EXECUTOR_QUIESCE_TIMEOUT, resolve_shutdown_watchdog_delay(timeout) - ctx.elapsed() - 1.0),
        )
        _exec_live = GatewayRunner._shutdown_executor(self, drain_timeout=_exec_quiesce_budget)
        if _exec_live:
            # A live worker may be mid-write (the #101093 corruption sequence): skip the close and let
            # SQLite recover from its WAL on next open (at worst a transient "database is locked").
            logger.warning(
                "Shutdown phase: %d executor worker(s) still running after a %.2fs quiesce — skipping the "
                "SessionDB close/checkpoint to avoid racing a live write (#101093); handles are left "
                "open for SQLite to recover on next open", _exec_live, _exec_quiesce_budget,
            )
            return
        logger.info("Shutdown phase: executor quiesced at +%.2fs", ctx.elapsed())
        # Cron jobs (scheduler pool), API-server runs and deferred hygiene workers (both on the loop's
        # default executor) never touch self._executor, so the join above cannot see them. A writer that
        # outlived the drain is mid-write for the same #101093 reasons; the drain already spent its
        # budget, so no second wait — leave the handles open (#102198). The API count is the snapshot
        # taken before the adapters were released; a run whose handler task was cancelled at disconnect
        # has already left it, so that term under-counts — the live worker-scoped count below covers
        # the cancelled-handler case (#116535).
        _cron_live = self._active_cron_job_count()
        _api_live = ctx.api_live
        _api_worker_live = self._active_api_worker_count()
        _deferred_live = ctx.deferred_count()
        if _cron_live or _api_live or _api_worker_live or _deferred_live:
            logger.warning(
                "Shutdown phase: %d cron job(s) / %d API-server run(s) / %d API-server worker(s) / "
                "%d deferred worker(s) still running after the executor quiesce — skipping the SessionDB "
                "close/checkpoint, leaving state.db open for the live writer (#102198, #116535)",
                _cron_live, _api_live, _api_worker_live, _deferred_live,
            )
            return
        _step = GatewayShutdownMixin._quiet_step
        # Close SQLite session DBs so --replace's new gateway does not hit 'database is locked'.
        # ``_session_db`` is an AsyncSessionDB facade — unwrap; ``session_store`` holds ``_db``.
        _self_db = getattr(self, "_session_db", None)
        _self_db = getattr(_self_db, "_db", _self_db)
        store = getattr(self, "session_store", None)
        for _db in (_self_db, getattr(store, "_db", None)):
            if _db is not None and hasattr(_db, "close"):
                _step("SessionDB close error", _db.close)
        # Multiplexed session_store caches one SessionDB per profile; sweep the secondary WAL locks too.
        _sweep = getattr(store, "close_all_db_handles", None)
        if _sweep is not None:
            _step("SessionDB handle sweep error", _sweep)
        # Same sweep for the runner's own per-profile session_search handles.
        _step("Runner SessionDB handle sweep error", lambda: GatewayRunner.close_all_session_db_handles(self))

        def _close_shared() -> None:
            # Shared SessionDB instances still held by the process-wide registry (tools, cron, mirror).
            # This is the safety net that guarantees no WAL write lock survives past gateway shutdown
            # (#90837).
            from hermes_state_registry import close_all
            closed = close_all()
            if closed:
                logger.debug("Closed %d shared SessionDB instance(s) at shutdown", closed)

        _step("Shared SessionDB close error", _close_shared)
        logger.info("Shutdown phase: SessionDB close done at +%.2fs", ctx.elapsed())

    async def _stop_persist_exit_state(self, ctx: "GatewayShutdownMixin._StopContext") -> None:
        """PID/lock release, clean-shutdown marker, restart markers, terminal runtime status."""
        from gateway.run import _hermes_home, _planned_restart_notification_path, _shutdown_gateway_health_export
        from utils import atomic_json_write
        from gateway.status import remove_pid_file, release_gateway_runtime_lock
        remove_pid_file()
        release_gateway_runtime_lock()
        # Clean-shutdown marker skips crash-turn recovery next boot; a timed-out drain left
        # half-finished sessions, so no marker — the next startup recovers their turn markers.
        if not ctx.timed_out:
            with suppress(Exception):
                (_hermes_home / ".clean_shutdown").touch()
        else:
            logger.info(
                "Skipping .clean_shutdown marker — drain timed out with "
                "interrupted agents; next startup will recover their interrupted turns."
            )
        # Stuck-loop counter: sessions active across 3 consecutive restarts are auto-suspended next boot.
        if ctx.active_agents:
            self._increment_restart_failure_counts(set(ctx.active_agents.keys()))
        if self._restart_requested and self._restart_command_source is None:
            with _log_suppressed(logging.DEBUG, "Failed to write planned restart notification marker: %s"):
                atomic_json_write(
                    _planned_restart_notification_path(),
                    {
                        "requested_at": time.time(),
                        "via_service": bool(self._restart_via_service),
                        "detached": bool(self._restart_detached),
                    },
                    indent=None,
                )
        if self._restart_requested and self._restart_via_service:
            # Exit 75 + ``RestartForceExitStatus=75``: systemd replaces us without a racing helper.
            self._exit_code = GATEWAY_SERVICE_RESTART_EXIT_CODE
            self._exit_reason = self._exit_reason or "Gateway restart requested"
        self._draining = False
        # Terminal gateway_state: "stopped", or "running" on an UNEXPECTED signal (docker restart,
        # OOM) — container_boot.py only auto-starts gateways last seen "running".
        if getattr(self, "_signal_initiated_shutdown", False) and not self._restart_requested:
            logger.info(
                "Gateway stopped by an unexpected signal — persisting "
                "gateway_state=running so container_boot auto-starts on the next boot (issue #42675)"
            )
            self._update_runtime_status("running", self._exit_reason)
        else:
            self._update_runtime_status("stopped", self._exit_reason)
        try:
            from gateway.status import flush_runtime_status_async
            # Never outlive the launchd exit budget (``_launchd_exit_timeout_s`` is set by the
            # supervised-restart path when it exists; a plain 2 s bound otherwise). The cap only
            # applies to signal-driven stops — a programmatic stop is not racing the supervisor.
            flush_timeout = 2.0
            budget = getattr(self, "_launchd_exit_timeout_s", None)
            signal_stop = getattr(self, "_stop_requested_by_signal", False)
            if signal_stop and isinstance(budget, (int, float)) and budget > 0:
                flush_timeout = max(0.0, min(flush_timeout, budget - ctx.elapsed()))
            if not await flush_runtime_status_async(timeout=flush_timeout):
                logger.warning("Timed out flushing terminal gateway runtime status")
        except Exception:
            logger.debug("Failed to flush terminal gateway runtime status", exc_info=True)
        _shutdown_gateway_health_export(self)
        logger.info("Gateway stopped (total teardown %.2fs)", ctx.elapsed())

    def _shutdown_watchdog_snapshot(self, ctx: "GatewayShutdownMixin._StopContext") -> dict:
        """State dumped by the thread-based shutdown watchdog when teardown hangs."""
        return {
            "restart_requested": bool(self._restart_requested),
            "draining": bool(self._draining),
            "running": bool(self._running),
            "active_agents": self._running_agent_count(),
            "active_cron_jobs": self._active_cron_job_count(),
            "active_api_runs": self._active_api_run_count(),
            "active_deferred_agent_workers": ctx.deferred_count(),
            "restart_drain_timeout": self._restart_drain_timeout,
            "effective_drain_timeout": effective_stop_drain_timeout(self),
            "launchd_exit_timeout_s": getattr(self, "_launchd_exit_timeout_s", None),
            "watchdog_delay_s": _effective_watchdog_leash(self),
            "phase_elapsed_s": ctx.elapsed() if ctx.started_at is not None else None,
        }

    async def _stop_impl(self) -> None:
        """Run every ``_stop_*`` phase under the thread-based shutdown watchdog."""
        from gateway.run import GatewayRunner
        # Thread-based watchdog (asyncio timeouts cannot recover a frozen loop): dumps stacks and
        # os._exit past drain+grace so the service manager revives us. Skipped under pytest.
        # Arm a plain OS thread at the start of stop(); if teardown never finishes within drain+grace it
        # dumps faulthandler stacks and os._exit so KeepAlive/systemd can revive. Skip under pytest so
        # stop()-driving unit tests don't get a delayed hard-exit in the worker. See #66892.
        _watchdog_done = threading.Event()
        self._shutdown_watchdog_done = _watchdog_done
        # Shutdown-path doubles may lack the deferred-worker counter.
        ctx = GatewayShutdownMixin._StopContext(
            deferred_count=getattr(self, "_active_deferred_agent_worker_count", lambda: 0)
        )
        if not os.environ.get("PYTEST_CURRENT_TEST"):
            arm_shutdown_watchdog(
                _effective_watchdog_leash(self), done_event=_watchdog_done,
                snapshot_fn=lambda: GatewayRunner._shutdown_watchdog_snapshot(self, ctx), exit_code=1,
            )
        try:
            await GatewayRunner._stop_begin_teardown(self, ctx)
            timeout = effective_stop_drain_timeout(self)
            if timeout < self._restart_drain_timeout:
                logger.warning(
                    "Shutdown drain capped to %.0fs (configured %.0fs) to fit the live launchd exit "
                    "timeout of %.0fs — launchd SIGKILLs past it",
                    timeout, self._restart_drain_timeout, self._launchd_exit_timeout_s,
                )
            await GatewayRunner._stop_drain_active_work(self, timeout, ctx)
            if ctx.timed_out:
                await GatewayRunner._stop_interrupt_remaining_work(self, ctx)
            await GatewayRunner._stop_finalize_agents_and_adapters(self, ctx)
            await GatewayRunner._stop_release_runtime_state(self, ctx)
            GatewayRunner._stop_quiesce_and_close_session_dbs(self, timeout, ctx)
            await GatewayRunner._stop_persist_exit_state(self, ctx)
        finally:
            _watchdog_done.set()

    async def stop(
        self, *, restart: bool = False, detached_restart: bool = False, service_restart: bool = False
    ) -> None:
        """Stop the gateway and disconnect all adapters."""
        from gateway.run import GatewayRunner
        # getattr-guard: shutdown-path tests build bare runners via object.__new__ that lack the
        # liveness-guard machinery.
        _stop_guards = getattr(self, "_stop_loop_liveness_guards", None)
        if callable(_stop_guards):
            _stop_guards()
        if restart:
            self._restart_requested = True
            self._restart_detached = detached_restart
            self._restart_via_service = service_restart
        if self._stop_task is not None:
            await self._stop_task
            return
        self._stop_task = asyncio.create_task(GatewayRunner._stop_impl(self))
        await self._stop_task

    async def wait_for_shutdown(self) -> None:
        """Wait for shutdown signal."""
        await self._shutdown_event.wait()