fix: notify active sessions on gateway shutdown + update health check

Three fixes for gateway lifecycle stability:

1. Notify active sessions before shutdown (#new)
   When the gateway receives SIGTERM or /restart, it now sends a
   notification to every chat with an active agent BEFORE starting
   the drain. Users see:
   - Shutdown: 'Gateway shutting down — your task will be interrupted.'
   - Restart: 'Gateway restarting — use /retry after restart to continue.'
   Deduplicates per-chat so group sessions with multiple users get
   one notification. Best-effort: send failures are logged and swallowed.

2. Skip .clean_shutdown marker when drain timed out
   Previously, a graceful SIGTERM always wrote .clean_shutdown, even if
   agents were force-interrupted when the drain timed out. This meant
   the next startup skipped session suspension, leaving interrupted
   sessions in a broken state (trailing tool response, no final message).
   Now the marker is only written if the drain completed without timeout,
   so interrupted sessions get properly suspended on next startup.

3. Post-restart health check for hermes update (#6631)
   cmd_update() now verifies the gateway actually survived after
   systemctl restart (sleep 3s + is-active check). If the service
   crashed immediately, it retries once. If still dead, prints
   actionable diagnostics (journalctl command, manual restart hint).

Also closes #8104 — already fixed on main (the /restart handler
correctly detects systemd via INVOCATION_ID and uses via_service=True).

Test plan:
- 6 new tests for shutdown notifications (dedup, restart vs shutdown
  messaging, sentinel filtering, send failure resilience)
- Existing restart drain + update tests pass (47 total)

tests/gateway/restart_test_helpers.py

@@ -93,6 +93,12 @@ def make_restart_runner(
     runner._running_agent_count = GatewayRunner._running_agent_count.__get__(
         runner, GatewayRunner
     )
+    runner._snapshot_running_agents = GatewayRunner._snapshot_running_agents.__get__(
+        runner, GatewayRunner
+    )
+    runner._notify_active_sessions_of_shutdown = (
+        GatewayRunner._notify_active_sessions_of_shutdown.__get__(runner, GatewayRunner)
+    )
     runner._launch_detached_restart_command = GatewayRunner._launch_detached_restart_command.__get__(
         runner, GatewayRunner
     )

tests/gateway/test_restart_drain.py

@@ -161,3 +161,84 @@ async def test_launch_detached_restart_command_uses_setsid(monkeypatch):
     assert kwargs["start_new_session"] is True
     assert kwargs["stdout"] is subprocess.DEVNULL
     assert kwargs["stderr"] is subprocess.DEVNULL
+
+
+# ── Shutdown notification tests ──────────────────────────────────────
+
+
+@pytest.mark.asyncio
+async def test_shutdown_notification_sent_to_active_sessions():
+    """Active sessions receive a notification when the gateway starts shutting down."""
+    runner, adapter = make_restart_runner()
+    source = make_restart_source(chat_id="999", chat_type="dm")
+    session_key = f"agent:main:telegram:dm:999"
+    runner._running_agents[session_key] = MagicMock()
+
+    await runner._notify_active_sessions_of_shutdown()
+
+    assert len(adapter.sent) == 1
+    assert "shutting down" in adapter.sent[0]
+    assert "interrupted" in adapter.sent[0]
+
+
+@pytest.mark.asyncio
+async def test_shutdown_notification_says_restarting_when_restart_requested():
+    """When _restart_requested is True, the message says 'restarting' and mentions /retry."""
+    runner, adapter = make_restart_runner()
+    runner._restart_requested = True
+    session_key = "agent:main:telegram:dm:999"
+    runner._running_agents[session_key] = MagicMock()
+
+    await runner._notify_active_sessions_of_shutdown()
+
+    assert len(adapter.sent) == 1
+    assert "restarting" in adapter.sent[0]
+    assert "/retry" in adapter.sent[0]
+
+
+@pytest.mark.asyncio
+async def test_shutdown_notification_deduplicates_per_chat():
+    """Multiple sessions in the same chat only get one notification."""
+    runner, adapter = make_restart_runner()
+    # Two sessions (different users) in the same chat
+    runner._running_agents["agent:main:telegram:group:chat1:u1"] = MagicMock()
+    runner._running_agents["agent:main:telegram:group:chat1:u2"] = MagicMock()
+
+    await runner._notify_active_sessions_of_shutdown()
+
+    assert len(adapter.sent) == 1
+
+
+@pytest.mark.asyncio
+async def test_shutdown_notification_skipped_when_no_active_agents():
+    """No notification is sent when there are no active agents."""
+    runner, adapter = make_restart_runner()
+
+    await runner._notify_active_sessions_of_shutdown()
+
+    assert len(adapter.sent) == 0
+
+
+@pytest.mark.asyncio
+async def test_shutdown_notification_ignores_pending_sentinels():
+    """Pending sentinels (not-yet-started agents) don't trigger notifications."""
+    from gateway.run import _AGENT_PENDING_SENTINEL
+
+    runner, adapter = make_restart_runner()
+    runner._running_agents["agent:main:telegram:dm:999"] = _AGENT_PENDING_SENTINEL
+
+    await runner._notify_active_sessions_of_shutdown()
+
+    assert len(adapter.sent) == 0
+
+
+@pytest.mark.asyncio
+async def test_shutdown_notification_send_failure_does_not_block():
+    """If sending a notification fails, the method still completes."""
+    runner, adapter = make_restart_runner()
+    adapter.send = AsyncMock(side_effect=Exception("network error"))
+    session_key = "agent:main:telegram:dm:999"
+    runner._running_agents[session_key] = MagicMock()
+
+    # Should not raise
+    await runner._notify_active_sessions_of_shutdown()