fix(gateway): systemd KillMode=mixed + restart drain timeout + cron stop guard (#37454 #37453 #37858)

gateway/run.py

@@ -19301,10 +19301,10 @@ def _run_planned_stop_watcher(
         stop_event.wait(poll_interval)
 
 
-def _start_cron_ticker(stop_event: threading.Event, adapters=None, loop=None, interval: int = 60):
+def _start_cron_ticker(stop_event: threading.Event, adapters=None, loop=None, interval: int = 60, runner=None):
     """
     Background thread that ticks the cron scheduler at a regular interval.
-    
+
     Runs inside the gateway process so cronjobs fire automatically without
     needing a separate `hermes cron daemon` or system cron entry.
 
@@ -19314,6 +19314,17 @@ def _start_cron_ticker(stop_event: threading.Event, adapters=None, loop=None, in
     Also refreshes the channel directory every 5 minutes and prunes the
     image/audio/document cache + expired ``hermes debug share`` pastes
     once per hour.
+
+    Args:
+        stop_event: Set when the gateway is shutting down — causes the
+            ticker loop to exit cleanly.
+        adapters: Live platform adapters for in-process delivery.
+        loop: The gateway asyncio event loop.
+        interval: Seconds between scheduler ticks (default 60).
+        runner: Optional GatewayRunner reference.  When provided, each tick
+            checks ``runner._draining`` before invoking the scheduler so
+            that a cron job cannot start a new outbound agent API call
+            after shutdown has been initiated (issue #37858).
     """
     from cron.scheduler import tick as cron_tick
     from gateway.platforms.base import cleanup_image_cache, cleanup_document_cache
@@ -19328,7 +19339,13 @@ def _start_cron_ticker(stop_event: threading.Event, adapters=None, loop=None, in
     tick_count = 0
     while not stop_event.is_set():
         try:
-            cron_tick(verbose=False, adapters=adapters, loop=loop, sync=False)
+            # Issue #37858: skip cron tick when the gateway is draining so
+            # a pending job cannot fire an outbound agent API call after
+            # shutdown has been initiated.
+            if runner is not None and getattr(runner, "_draining", False):
+                logger.debug("Cron tick skipped — gateway is draining")
+            else:
+                cron_tick(verbose=False, adapters=adapters, loop=loop, sync=False)
         except Exception as e:
             logger.debug("Cron tick error: %s", e)
 
@@ -19755,11 +19772,13 @@ def restart_signal_handler():
 
     # Start background cron ticker so scheduled jobs fire automatically.
     # Pass the event loop so cron delivery can use live adapters (E2EE support).
+    # Pass runner so the ticker can skip ticks while the gateway is draining
+    # (issue #37858 — prevents outbound agent API calls after stop is initiated).
     cron_stop = threading.Event()
     cron_thread = threading.Thread(
         target=_start_cron_ticker,
         args=(cron_stop,),
-        kwargs={"adapters": runner.adapters, "loop": asyncio.get_running_loop()},
+        kwargs={"adapters": runner.adapters, "loop": asyncio.get_running_loop(), "runner": runner},
         daemon=True,
         name="cron-ticker",
     )

hermes_cli/gateway.py

@@ -6387,11 +6387,20 @@ def _gateway_command_inner(args):
                 print("  Fix the service, then retry: hermes gateway start")
                 sys.exit(1)
 
-            # Manual restart: stop only this profile's gateway
+            # Manual restart: stop only this profile's gateway.
+            # Issue #37453 — wait for the full drain timeout before force-
+            # killing so a simultaneous stop+start cannot race on the same
+            # port/socket.  Add 5s of headroom beyond the gateway's own
+            # drain_timeout so the drain loop can finish cleanly before the
+            # CLI escalates to SIGKILL.
             if stop_profile_gateway():
                 print("✓ Stopped gateway for this profile")
 
-            _wait_for_gateway_exit(timeout=10.0, force_after=5.0)
+            _drain_timeout = _get_restart_drain_timeout()
+            _wait_for_gateway_exit(
+                timeout=_drain_timeout + 5.0,
+                force_after=_drain_timeout,
+            )
 
             # Start fresh
             print("Starting gateway...")

tests/gateway/test_cron_drain_guard.py

@@ -0,0 +1,137 @@
+"""Tests for the cron-ticker drain guard (issue #37858).
+
+When ``hermes gateway stop`` initiates shutdown, the gateway sets
+``runner._draining = True`` before the drain loop runs.  If the cron ticker
+background thread fires a tick at that exact moment it can launch a new
+outbound agent API call — making an LLM request and (on cron-deliver jobs)
+sending a platform message — after the operator has already asked the gateway
+to stop.
+
+The fix passes the ``runner`` reference into ``_start_cron_ticker`` so the
+loop can skip ``cron_tick()`` calls whenever ``runner._draining`` is True.
+"""
+
+import threading
+import time
+from unittest.mock import MagicMock, patch
+
+
+from gateway.run import _start_cron_ticker
+
+
+class _FakeRunner:
+    """Minimal stand-in for GatewayRunner that exposes only the _draining flag."""
+
+    def __init__(self, *, draining: bool = False):
+        self._draining = draining
+
+
+def test_cron_ticker_skips_tick_when_runner_is_draining():
+    """While runner._draining is True the ticker must NOT call cron_tick()."""
+    tick_calls = []
+
+    stop_event = threading.Event()
+    runner = _FakeRunner(draining=True)
+
+    # Patch cron_tick so we can count invocations without running real jobs.
+    with patch("cron.scheduler.tick", side_effect=lambda **kw: tick_calls.append(1)) as _mock_tick:
+        # Run the ticker in a background thread for a couple of intervals.
+        thread = threading.Thread(
+            target=_start_cron_ticker,
+            args=(stop_event,),
+            kwargs={"interval": 0, "runner": runner},
+            daemon=True,
+        )
+        thread.start()
+        time.sleep(0.15)  # enough for several zero-interval ticks
+        stop_event.set()
+        thread.join(timeout=2.0)
+
+    assert not thread.is_alive(), "Ticker thread did not exit cleanly"
+    assert tick_calls == [], (
+        f"cron_tick() was called {len(tick_calls)} time(s) while runner._draining=True — "
+        "outbound agent calls must not fire after stop is initiated (#37858)"
+    )
+
+
+def test_cron_ticker_runs_tick_when_runner_is_not_draining():
+    """Normal operation: cron_tick() fires when the gateway is NOT draining."""
+    tick_calls = []
+
+    stop_event = threading.Event()
+    runner = _FakeRunner(draining=False)
+
+    with patch("cron.scheduler.tick", side_effect=lambda **kw: tick_calls.append(1)):
+        thread = threading.Thread(
+            target=_start_cron_ticker,
+            args=(stop_event,),
+            kwargs={"interval": 0, "runner": runner},
+            daemon=True,
+        )
+        thread.start()
+        time.sleep(0.15)
+        stop_event.set()
+        thread.join(timeout=2.0)
+
+    assert not thread.is_alive(), "Ticker thread did not exit cleanly"
+    assert tick_calls, (
+        "cron_tick() was never called when runner._draining=False — "
+        "normal tick operation is broken"
+    )
+
+
+def test_cron_ticker_skips_tick_without_runner():
+    """When runner=None (legacy call sites), the ticker must still call cron_tick()
+    unchanged — the drain guard is a no-op when no runner reference is provided."""
+    tick_calls = []
+
+    stop_event = threading.Event()
+
+    with patch("cron.scheduler.tick", side_effect=lambda **kw: tick_calls.append(1)):
+        thread = threading.Thread(
+            target=_start_cron_ticker,
+            args=(stop_event,),
+            kwargs={"interval": 0, "runner": None},
+            daemon=True,
+        )
+        thread.start()
+        time.sleep(0.15)
+        stop_event.set()
+        thread.join(timeout=2.0)
+
+    assert not thread.is_alive()
+    assert tick_calls, (
+        "cron_tick() was never called when runner=None — "
+        "backwards-compat with callers that omit runner is broken"
+    )
+
+
+def test_cron_ticker_resumes_after_drain_clears():
+    """Once runner._draining reverts to False, ticks should resume normally.
+
+    This covers the case where the gateway runner temporarily sets _draining
+    during a restart then clears it (edge-case drain flag lifecycle).
+    """
+    tick_calls = []
+    stop_event = threading.Event()
+    runner = _FakeRunner(draining=True)
+
+    with patch("cron.scheduler.tick", side_effect=lambda **kw: tick_calls.append(1)):
+        thread = threading.Thread(
+            target=_start_cron_ticker,
+            args=(stop_event,),
+            kwargs={"interval": 0, "runner": runner},
+            daemon=True,
+        )
+        thread.start()
+        time.sleep(0.1)
+        # Simulate drain completing (e.g. the runner resets the flag internally)
+        runner._draining = False
+        time.sleep(0.15)
+        stop_event.set()
+        thread.join(timeout=2.0)
+
+    assert not thread.is_alive()
+    assert tick_calls, (
+        "cron_tick() should fire once _draining is cleared, but never did"
+    )

tests/hermes_cli/test_gateway_service.py

@@ -2600,3 +2600,104 @@ def test_launchd_plist_keepalive_unconditional(self, tmp_path, monkeypatch):
         # The old conditional dict form must NOT appear
         assert "SuccessfulExit" not in plist
         assert "<key>KeepAlive</key>\n    <dict>" not in plist
+
+
+class TestSystemdKillMode:
+    """Issue #37454 — systemd unit must use KillMode=mixed so the main
+    process receives SIGTERM (drain-then-exit) before worker processes in
+    the cgroup are killed.  KillMode=control-group (the systemd default)
+    sends SIGKILL to the whole cgroup immediately, bypassing the drain.
+    """
+
+    def test_user_unit_has_kill_mode_mixed(self, monkeypatch):
+        monkeypatch.setattr(
+            gateway_cli,
+            "_get_restart_drain_timeout",
+            lambda: DEFAULT_GATEWAY_RESTART_DRAIN_TIMEOUT,
+        )
+        unit = gateway_cli.generate_systemd_unit(system=False)
+        assert "KillMode=mixed" in unit, (
+            "user-scope unit must set KillMode=mixed so SIGTERM reaches the "
+            "main process before workers are killed (#37454)"
+        )
+
+    def test_system_unit_has_kill_mode_mixed(self, monkeypatch):
+        monkeypatch.setattr(
+            gateway_cli,
+            "_get_restart_drain_timeout",
+            lambda: DEFAULT_GATEWAY_RESTART_DRAIN_TIMEOUT,
+        )
+        unit = gateway_cli.generate_systemd_unit(system=True)
+        assert "KillMode=mixed" in unit, (
+            "system-scope unit must set KillMode=mixed (#37454)"
+        )
+
+    def test_unit_does_not_use_control_group_kill_mode(self, monkeypatch):
+        monkeypatch.setattr(
+            gateway_cli,
+            "_get_restart_drain_timeout",
+            lambda: DEFAULT_GATEWAY_RESTART_DRAIN_TIMEOUT,
+        )
+        for system in (False, True):
+            unit = gateway_cli.generate_systemd_unit(system=system)
+            assert "KillMode=control-group" not in unit, (
+                f"{'system' if system else 'user'}-scope unit must NOT use "
+                "KillMode=control-group (would SIGKILL workers before drain)"
+            )
+
+
+class TestManualRestartDrainWait:
+    """Issue #37453 — manual `hermes gateway restart` (no systemd/launchd)
+    must wait for the full drain timeout before force-killing the old process
+    so a simultaneous stop+start cannot race on the same port/socket.
+    """
+
+    def test_manual_restart_waits_for_drain_timeout_not_hardcoded_10s(
+        self, monkeypatch, capsys
+    ):
+        """The wait before spawning the replacement must honour the configured
+        drain_timeout, not a hardcoded 10s value.  Without this, a 60s drain
+        that has not yet finished is force-killed after 5s, and the new
+        gateway process races on the same socket (#37453).
+        """
+        configured_drain = 45.0  # non-default — proves the value is respected
+
+        monkeypatch.setattr(gateway_cli, "supports_systemd_services", lambda: False)
+        monkeypatch.setattr(gateway_cli, "is_macos", lambda: False)
+        monkeypatch.setattr(gateway_cli, "is_windows", lambda: False)
+        monkeypatch.setattr(gateway_cli, "is_termux", lambda: False)
+        monkeypatch.setattr(gateway_cli, "is_container", lambda: False)
+        monkeypatch.setattr(gateway_cli, "_dispatch_via_service_manager_if_s6", lambda _: False)
+        monkeypatch.setattr(gateway_cli, "_dispatch_all_via_service_manager_if_s6", lambda _: False)
+        monkeypatch.setattr(gateway_cli, "_get_restart_drain_timeout", lambda: configured_drain)
+        monkeypatch.setattr(gateway_cli, "stop_profile_gateway", lambda: True)
+
+        wait_calls = []
+
+        def fake_wait_for_exit(timeout, force_after):
+            wait_calls.append((timeout, force_after))
+
+        monkeypatch.setattr(gateway_cli, "_wait_for_gateway_exit", fake_wait_for_exit)
+        monkeypatch.setattr(gateway_cli, "run_gateway", lambda verbose=0: None)
+
+        gateway_cli.gateway_command(
+            SimpleNamespace(
+                gateway_command="restart",
+                system=False,
+                **{"all": False},
+            )
+        )
+
+        assert wait_calls, "expected _wait_for_gateway_exit to be called"
+        timeout, force_after = wait_calls[0]
+        # timeout must be at least the drain_timeout so the process gets the
+        # full budget to finish draining before the CLI gives up.
+        assert timeout >= configured_drain, (
+            f"wait timeout ({timeout}s) must be >= drain_timeout ({configured_drain}s) "
+            "to avoid race-starting the new gateway while old one is still draining"
+        )
+        # force_after must also be >= drain_timeout so we don't SIGKILL before
+        # the drain window has expired.
+        assert force_after >= configured_drain, (
+            f"force_after ({force_after}s) must be >= drain_timeout ({configured_drain}s)"
+        )