fix(gateway): skip home-channel shutdown broadcast when idle

README.md

@@ -158,6 +158,10 @@ Hermes has two entry points: start the terminal UI with `hermes`, or run the gat
 
 For the full command lists, see the [CLI guide](https://hermes-agent.nousresearch.com/docs/user-guide/cli) and the [Messaging Gateway guide](https://hermes-agent.nousresearch.com/docs/user-guide/messaging).
 
+When an external service restart stops an idle gateway, Hermes skips the home-channel
+interruption warning because no active task will be interrupted. Gateway-initiated
+restarts still announce the restart in the configured home channel.
+
 ---
 
 ## Documentation

gateway/run.py

@@ -8975,6 +8975,22 @@ async def _notify_active_sessions_of_shutdown(self) -> None:
             logger.debug("Skipping home-channel shutdown notifications for in-chat restart")
             return
 
+        if not active and not self._restart_requested:
+            # External stop/restart (systemd, cron auto-update) with no
+            # running agents: the "your current task will be interrupted"
+            # warning is false, so the home-channel ping is pure noise —
+            # scheduled restarts of an idle gateway hit this on every cycle
+            # (#20103, #29846). Gateway-initiated restarts keep the idle
+            # broadcast (test_restart_notifies_home_channel_even_without_
+            # active_sessions pins it) so the home channel hears the
+            # "I'll be back" hint, and the drain-marker gate below still
+            # covers force-interrupt drains, which reach here with active
+            # sessions.
+            logger.info(
+                "Skipping home-channel shutdown notification: no active sessions to interrupt"
+            )
+            return
+
         # Suppress ONLY the home-channel broadcast when the drain that is ending
         # in this shutdown asked us to be quiet (e.g. a NAS auto-update image
         # migration — drain-gated, then the machine is recreated). On the

tests/gateway/test_gateway_shutdown.py

@@ -120,6 +120,80 @@ async def test_in_chat_restart_skips_home_shutdown_even_with_active_session():
     assert metadata["telegram_reply_to_message_id"] == "restart-command"
 
 
+@pytest.mark.asyncio
+async def test_idle_in_chat_restart_does_not_send_interruption_warning():
+    runner, adapter = make_restart_runner()
+    source = make_restart_source(thread_id="42")
+    source.message_id = "restart-command"
+    runner._restart_requested = True
+    runner._restart_command_source = source
+    runner.config.platforms[Platform.TELEGRAM].home_channel = HomeChannel(
+        platform=Platform.TELEGRAM,
+        chat_id=source.chat_id,
+        name="Telegram",
+        thread_id=source.thread_id,
+    )
+
+    await runner._notify_active_sessions_of_shutdown()
+
+    assert adapter.sent_calls == []
+
+
+@pytest.mark.asyncio
+async def test_idle_external_shutdown_skips_home_channel_notification():
+    """External stop/restart (systemd, cron) with no running agents must not
+    broadcast the interruption warning to home channels (#20103, #29846)."""
+    runner, adapter = make_restart_runner()
+    runner.config.platforms[Platform.TELEGRAM].home_channel = HomeChannel(
+        platform=Platform.TELEGRAM,
+        chat_id="home-chat",
+        name="Telegram Home",
+    )
+
+    await runner._notify_active_sessions_of_shutdown()
+
+    assert adapter.sent_calls == []
+
+
+@pytest.mark.asyncio
+async def test_in_chat_restart_does_not_write_home_startup_marker(tmp_path, monkeypatch):
+    monkeypatch.setattr(gateway_run, "_hermes_home", tmp_path)
+    runner, adapter = make_restart_runner()
+    adapter.disconnect = AsyncMock()
+    source = make_restart_source(thread_id="42")
+    source.message_id = "restart-command"
+    runner._restart_command_source = source
+    runner._launch_systemd_restart_shortcut = MagicMock()
+    monkeypatch.setenv("INVOCATION_ID", "systemd-test")
+
+    with patch("gateway.status.remove_pid_file"), patch("gateway.status.write_runtime_status"):
+        await runner.stop(restart=True, service_restart=True)
+
+    assert not (tmp_path / ".restart_pending.json").exists()
+
+
+@pytest.mark.asyncio
+async def test_drain_active_agents_throttles_status_updates():
+    runner, _adapter = make_restart_runner()
+    runner._update_runtime_status = MagicMock()
+
+    runner._running_agents = {"a": MagicMock(), "b": MagicMock()}
+
+    async def finish_agents():
+        await asyncio.sleep(0.12)
+        runner._running_agents.pop("a")
+        await asyncio.sleep(0.12)
+        runner._running_agents.clear()
+
+    task = asyncio.create_task(finish_agents())
+    await runner._drain_active_agents(1.0)
+    await task
+
+    # Start, one count-change update, and final update. Allow one extra update
+    # if the loop observes the zero-agent state before exiting.
+    assert 3 <= runner._update_runtime_status.call_count <= 4
+
+
 @pytest.mark.asyncio
 async def test_gateway_stop_kills_tool_subprocesses_before_adapter_disconnect_on_timeout(monkeypatch):
     """On drain timeout, tool subprocesses must be killed BEFORE adapter
@@ -270,4 +344,3 @@ def status(self):
 
     assert status._pid_exists(4242) is False
 
-