feat(update): add opt-out for gateway shutdown notices

apps/desktop/src/app/settings/constants.ts

@@ -423,7 +423,8 @@ export const FIELD_LABELS: Record<string, string> = defineFieldCopy({
     reasoningEffort: 'Subagent Reasoning Effort'
   },
   updates: {
-    nonInteractiveLocalChanges: 'In-App Update Local Changes'
+    nonInteractiveLocalChanges: 'In-App Update Local Changes',
+    gatewayShutdownNotification: 'Update Shutdown Notifications'
   }
 })
 
@@ -494,7 +495,9 @@ export const FIELD_DESCRIPTIONS: Record<string, string> = defineFieldCopy({
   },
   updates: {
     nonInteractiveLocalChanges:
-      'When Hermes updates itself from the app (no terminal prompt), keep local source edits (stash) or throw them away (discard). Terminal updates always ask.'
+      'When Hermes updates itself from the app (no terminal prompt), keep local source edits (stash) or throw them away (discard). Terminal updates always ask.',
+    gatewayShutdownNotification:
+      'Send gateway shutdown warnings when Desktop or CLI updates pause messaging gateways. Other shutdown and restart notices are unchanged.'
   }
 })
 
@@ -630,6 +633,7 @@ export const SECTIONS: DesktopConfigSection[] = [
       'delegation.max_concurrent_children',
       'delegation.child_timeout_seconds',
       'delegation.reasoning_effort',
+      'updates.gateway_shutdown_notification',
       'updates.non_interactive_local_changes'
     ]
   }

apps/desktop/src/app/settings/helpers.test.ts

@@ -2,6 +2,7 @@ import { describe, expect, it } from 'vitest'
 
 import type { HermesConfigRecord } from '@/types/hermes'
 
+import { FIELD_DESCRIPTIONS, FIELD_LABELS, SECTIONS } from './constants'
 import { defineFieldCopy, fieldCopyForSchemaKey, schemaKeyToFieldCopyKey } from './field-copy'
 import { enumOptionsFor, getNested, providerGroup, setNested, stripToolsetLabel, toolsetDisplayLabel } from './helpers'
 
@@ -48,6 +49,9 @@ describe('settings helpers', () => {
       expect(schemaKeyToFieldCopyKey('updates.non_interactive_local_changes')).toBe(
         'updates.nonInteractiveLocalChanges'
       )
+      expect(schemaKeyToFieldCopyKey('updates.gateway_shutdown_notification')).toBe(
+        'updates.gatewayShutdownNotification'
+      )
     })
 
     it('looks up camelCase field copy by schema key with legacy fallback', () => {
@@ -82,6 +86,18 @@ describe('settings helpers', () => {
     })
   })
 
+  it('surfaces update shutdown notifications in Advanced settings', () => {
+    const advanced = SECTIONS.find(section => section.id === 'advanced')
+
+    expect(advanced?.keys).toContain('updates.gateway_shutdown_notification')
+    expect(fieldCopyForSchemaKey(FIELD_LABELS, 'updates.gateway_shutdown_notification')).toBe(
+      'Update Shutdown Notifications'
+    )
+    expect(fieldCopyForSchemaKey(FIELD_DESCRIPTIONS, 'updates.gateway_shutdown_notification')).toContain(
+      'Other shutdown and restart notices are unchanged.'
+    )
+  })
+
   it('reads and writes nested config paths', () => {
     const config: HermesConfigRecord = { display: { theme: 'mono' } }
     const next = setNested(config, 'display.theme', 'slate')

gateway/run.py

@@ -2878,6 +2878,10 @@ def __init__(self, config: Optional[GatewayConfig] = None):
         # request -> poll -> proceed loop.
         self._external_drain_active = False
         self._restart_requested = False
+        # Update-specific planned-stop markers set this before stop() starts.
+        # It suppresses only user-facing shutdown pings; drain, resume_pending,
+        # interruption, cleanup, and restart behavior remain unchanged.
+        self._suppress_shutdown_notifications = False
         # Set by shutdown_signal_handler when a SIGTERM/SIGINT arrived
         # WITHOUT a planned-stop / takeover marker — i.e. an unexpected
         # external signal (container/s6 SIGTERM on `docker restart` or
@@ -5718,6 +5722,10 @@ async def _notify_active_sessions_of_shutdown(self) -> None:
         messages can be delivered. Best-effort: individual send failures are
         logged and swallowed so they never block the shutdown sequence.
         """
+        if getattr(self, "_suppress_shutdown_notifications", False):
+            logger.info("Shutdown notifications suppressed for planned update")
+            return
+
         active = self._snapshot_running_agents()
         restart_source = self._restart_command_source if self._restart_requested else None
 
@@ -20613,15 +20621,29 @@ def shutdown_signal_handler(received_signal=None):
         # external kill unless the CLI marks it first. SIGINT comes from an
         # interactive Ctrl+C and is likewise an intentional foreground stop.
         planned_stop = False
+        suppress_planned_stop_notification = False
         if received_signal == signal.SIGINT:
             planned_stop = True
         elif not planned_takeover:
+            try:
+                from gateway.status import (
+                    planned_stop_notification_suppressed_for_self,
+                )
+
+                suppress_planned_stop_notification = (
+                    planned_stop_notification_suppressed_for_self()
+                )
+            except Exception as e:
+                logger.debug("Planned stop notification check failed: %s", e)
             try:
                 from gateway.status import consume_planned_stop_marker_for_self
                 planned_stop = consume_planned_stop_marker_for_self()
             except Exception as e:
                 logger.debug("Planned stop marker check failed: %s", e)
 
+        if planned_stop and suppress_planned_stop_notification:
+            runner._suppress_shutdown_notifications = True
+
         # Fast (<10ms) snapshot of who's asking us to shut down — runs
         # synchronously inside the asyncio signal handler, so we keep it
         # purely stdlib + /proc reads, no subprocesses.  See PR #15826

gateway/status.py

@@ -1428,6 +1428,21 @@ def planned_stop_marker_targets_self() -> bool:
     return True
 
 
+def planned_stop_notification_suppressed_for_self() -> bool:
+    """Return whether this process's live planned stop requests quiet shutdown.
+
+    Only update-specific writers set ``suppress_notification=true``. Legacy
+    markers and normal ``hermes gateway stop`` / service-stop markers omit the
+    field, preserving their existing user-visible shutdown notifications.
+    Validation is delegated to :func:`planned_stop_marker_targets_self` so a
+    stale or foreign marker can never silence this gateway.
+    """
+    if not planned_stop_marker_targets_self():
+        return False
+    record = _read_json_file(_get_planned_stop_marker_path())
+    return bool(record and record.get("suppress_notification") is True)
+
+
 def clear_planned_stop_marker() -> None:
     """Remove the planned-stop marker unconditionally."""
     try:

hermes_cli/config.py

@@ -3085,6 +3085,11 @@ def _ensure_hermes_home_managed(home: Path):
         #               ignored paths — node_modules, venv, build outputs —
         #               are never touched.
         "non_interactive_local_changes": "stash",
+        # Send the normal gateway shutdown/interruption notice when a Desktop or
+        # CLI update intentionally pauses messaging gateways. True preserves the
+        # historical behavior. Set false for quiet update maintenance; manual
+        # stops, restarts, and unexpected shutdowns remain visible.
+        "gateway_shutdown_notification": True,
         # Refresh an already-installed cua-driver during `hermes update`.
         # The refresh is best-effort and macOS-only. Turn this off if the
         # upstream installer is not appropriate for the machine, for example

hermes_cli/main.py

@@ -8783,6 +8783,30 @@ def _run_pre_update_backup(args) -> None:
     print()
 
 
+def _update_gateway_shutdown_notification_enabled(profile_path: Path) -> bool:
+    """Read the per-profile update shutdown-notification preference.
+
+    Missing, malformed, or non-boolean values preserve the historical visible
+    behavior. Failing loud is safer than letting a damaged config silently mute
+    manual-looking interruption warnings.
+    """
+    config_path = Path(profile_path) / "config.yaml"
+    if not config_path.exists():
+        return True
+    try:
+        from utils import fast_safe_load
+
+        with open(config_path, encoding="utf-8") as config_file:
+            config = fast_safe_load(config_file) or {}
+        updates = config.get("updates") if isinstance(config, dict) else None
+        if not isinstance(updates, dict):
+            return True
+        value = updates.get("gateway_shutdown_notification", True)
+        return value if isinstance(value, bool) else True
+    except Exception:
+        return True
+
+
 def _write_update_planned_stop_marker(profile_path: Path, pid: int) -> bool:
     """Write a planned-stop marker into a specific profile home."""
     try:
@@ -8796,6 +8820,11 @@ def _write_update_planned_stop_marker(profile_path: Path, pid: int) -> bool:
             "target_start_time": _get_process_start_time(pid),
             "stopper_pid": os.getpid(),
             "written_at": datetime.now(timezone.utc).isoformat(),
+            # Only quiet the update when this profile explicitly opted out.
+            # The default remains user-visible for backward compatibility.
+            "suppress_notification": not (
+                _update_gateway_shutdown_notification_enabled(profile_path)
+            ),
         }
         atomic_json_write(
             Path(profile_path) / ".gateway-planned-stop.json",

tests/gateway/test_restart_drain.py

@@ -1,4 +1,5 @@
 import asyncio
+import os
 import shutil
 import subprocess
 from datetime import datetime
@@ -477,6 +478,131 @@ async def test_shutdown_notification_send_failure_does_not_block():
     await runner._notify_active_sessions_of_shutdown()
 
 
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    ("marker_kind", "expected_notifications"),
+    [
+        pytest.param("update-suppressed", 0, id="update-opt-out"),
+        pytest.param("update-visible", 1, id="update-default"),
+        pytest.param("legacy", 1, id="legacy-marker"),
+    ],
+)
+async def test_planned_stop_marker_flows_through_signal_handler_to_notification_delivery(
+    marker_kind,
+    expected_notifications,
+    monkeypatch,
+    tmp_path,
+):
+    """Exercise config -> marker -> signal handler -> notification delivery.
+
+    This is fully in-process: it captures the registered signal callback and
+    replaces ``runner.stop`` with a notification-only coroutine. No OS signal
+    is sent and no live gateway or Hermes session is stopped.
+    """
+    from gateway import status as status_mod
+    from hermes_cli import main as cli_main
+
+    marker = tmp_path / ".gateway-planned-stop.json"
+    monkeypatch.setenv("HERMES_HOME", str(tmp_path))
+    monkeypatch.setattr(gateway_run, "_hermes_home", tmp_path)
+    monkeypatch.setattr(status_mod, "_get_planned_stop_marker_path", lambda: marker)
+    monkeypatch.setattr(status_mod, "_get_process_start_time", lambda _pid: 42)
+
+    if marker_kind == "legacy":
+        assert status_mod.write_planned_stop_marker(target_pid=os.getpid()) is True
+    else:
+        enabled = marker_kind == "update-visible"
+        (tmp_path / "config.yaml").write_text(
+            f"updates:\n  gateway_shutdown_notification: {str(enabled).lower()}\n",
+            encoding="utf-8",
+        )
+        assert cli_main._write_update_planned_stop_marker(tmp_path, os.getpid()) is True
+
+    runner, adapter = make_restart_runner()
+    runner._running_agents["agent:main:telegram:dm:999"] = MagicMock()
+    runner._exit_cleanly = False
+    runner._exit_with_failure = False
+    stopped = asyncio.Event()
+    registered_signal_handlers = {}
+
+    async def notification_only_stop():
+        await runner._notify_active_sessions_of_shutdown()
+        runner._running = False
+        stopped.set()
+
+    async def start_and_invoke_registered_handler():
+        callback, args = registered_signal_handlers[gateway_run.signal.SIGTERM]
+        callback(*args)
+        await asyncio.wait_for(stopped.wait(), timeout=1)
+        runner._exit_cleanly = True
+        return True
+
+    runner.stop = notification_only_stop
+    runner.start = start_and_invoke_registered_handler
+
+    # Keep start_gateway's real signal-handler wiring while replacing every
+    # process-, lock-, network-, diagnostic-, and live-service boundary.
+    monkeypatch.setattr(gateway_run, "GatewayRunner", lambda _config: runner)
+    monkeypatch.setattr("gateway.code_skew.record_boot_fingerprint", lambda: None)
+    monkeypatch.setattr(
+        "hermes_cli.security_audit_startup.log_startup_security_warnings",
+        lambda **_kwargs: None,
+    )
+    monkeypatch.setattr("tools.skills_sync.sync_skills", lambda quiet=True: None)
+    monkeypatch.setattr(
+        "hermes_logging.setup_logging", lambda hermes_home, mode: tmp_path
+    )
+    monkeypatch.setattr(status_mod, "get_running_pid", lambda: None)
+    monkeypatch.setattr(status_mod, "acquire_gateway_runtime_lock", lambda: True)
+    monkeypatch.setattr(status_mod, "write_pid_file", lambda: None)
+    monkeypatch.setattr(status_mod, "remove_pid_file", lambda: None)
+    monkeypatch.setattr(status_mod, "release_gateway_runtime_lock", lambda: None)
+    monkeypatch.setattr("atexit.register", lambda *_args, **_kwargs: None)
+    monkeypatch.setattr(
+        "hermes_cli.nous_auth_keepalive.start_nous_auth_keepalive", lambda: None
+    )
+    monkeypatch.setattr("tools.mcp_tool.discover_mcp_tools", lambda: None)
+    monkeypatch.setattr(
+        gateway_run, "_ensure_windows_gateway_venv_imports", lambda: None
+    )
+    monkeypatch.setattr(
+        gateway_run, "_run_planned_stop_watcher", lambda *_args, **_kwargs: None
+    )
+    monkeypatch.setattr(
+        "gateway.shutdown_forensics.snapshot_shutdown_context",
+        lambda _signal: {"signal": "SIGTERM"},
+    )
+    monkeypatch.setattr(
+        "gateway.shutdown_forensics.format_context_for_log", lambda _context: "test"
+    )
+    monkeypatch.setattr(
+        "gateway.shutdown_forensics.spawn_async_diagnostic",
+        lambda *_args, **_kwargs: None,
+    )
+
+    loop = asyncio.get_running_loop()
+    monkeypatch.setattr(
+        loop,
+        "add_signal_handler",
+        lambda sig, callback, *args: registered_signal_handlers.__setitem__(
+            sig, (callback, args)
+        ),
+    )
+    monkeypatch.setattr(loop, "set_exception_handler", lambda _handler: None)
+
+    assert (
+        await gateway_run.start_gateway(
+            config=runner.config,
+            replace=False,
+            verbosity=None,
+        )
+        is True
+    )
+
+    assert not marker.exists(), "the signal handler must consume the marker"
+    assert len(adapter.sent) == expected_notifications
+
+
 @pytest.mark.asyncio
 async def test_shutdown_notification_suppressed_when_flag_disabled():
     """Active-session ping is muted when gateway_restart_notification=False on the platform."""

tests/gateway/test_status.py

@@ -1311,12 +1311,40 @@ def test_write_marker_records_target_identity(self, tmp_path, monkeypatch):
         assert payload["target_start_time"] == 42
         assert payload["stopper_pid"] == os.getpid()
         assert "written_at" in payload
+        assert payload.get("suppress_notification") is not True
+
+    def test_update_marker_suppresses_notification_only_for_target_self(
+        self, tmp_path, monkeypatch
+    ):
+        from datetime import datetime, timezone
+
+        monkeypatch.setenv("HERMES_HOME", str(tmp_path))
+        monkeypatch.setattr(status, "_get_process_start_time", lambda pid: 42)
+        marker = tmp_path / ".gateway-planned-stop.json"
+        marker.write_text(
+            json.dumps({
+                "target_pid": os.getpid(),
+                "target_start_time": 42,
+                "stopper_pid": 99999,
+                "written_at": datetime.now(timezone.utc).isoformat(),
+                "suppress_notification": True,
+            })
+        )
+
+        assert status.planned_stop_notification_suppressed_for_self() is True
+        assert marker.exists(), "suppression probe must be non-destructive"
+
+        payload = json.loads(marker.read_text())
+        payload["target_pid"] = os.getpid() + 9999
+        marker.write_text(json.dumps(payload))
+        assert status.planned_stop_notification_suppressed_for_self() is False
 
     def test_consume_returns_true_when_marker_names_self(self, tmp_path, monkeypatch):
         monkeypatch.setenv("HERMES_HOME", str(tmp_path))
         monkeypatch.setattr(status, "_get_process_start_time", lambda pid: 100)
         ok = status.write_planned_stop_marker(target_pid=os.getpid())
         assert ok is True
+        assert status.planned_stop_notification_suppressed_for_self() is False
 
         result = status.consume_planned_stop_marker_for_self()