feat: add opt-in gateway startup notifications

gateway/config.py

@@ -298,6 +298,12 @@ class PlatformConfig:
     # noise; keep True for back-channels where the operator wants them.
     gateway_restart_notification: bool = True
 
+    # Whether to send a home-channel "gateway is back" notification on every
+    # gateway startup, not just after a Hermes-initiated /restart. Default
+    # False avoids noisy pings for normal deploys and public/end-user bots;
+    # operators can opt in for back-channel health notices.
+    gateway_startup_notification: bool = False
+
     # Platform-specific settings
     extra: Dict[str, Any] = field(default_factory=dict)
 
@@ -307,6 +313,7 @@ def to_dict(self) -> Dict[str, Any]:
             "extra": self.extra,
             "reply_to_mode": self.reply_to_mode,
             "gateway_restart_notification": self.gateway_restart_notification,
+            "gateway_startup_notification": self.gateway_startup_notification,
         }
         if self.token:
             result["token"] = self.token
@@ -330,13 +337,18 @@ def from_dict(cls, data: Dict[str, Any]) -> "PlatformConfig":
         if _grn is None:
             _grn = data.get("extra", {}).get("gateway_restart_notification")
 
+        _gsn = data.get("gateway_startup_notification")
+        if _gsn is None:
+            _gsn = data.get("extra", {}).get("gateway_startup_notification")
+
         return cls(
             enabled=_coerce_bool(data.get("enabled"), False),
             token=data.get("token"),
             api_key=data.get("api_key"),
             home_channel=home_channel,
             reply_to_mode=data.get("reply_to_mode", "first"),
             gateway_restart_notification=_coerce_bool(_grn, True),
+            gateway_startup_notification=_coerce_bool(_gsn, False),
             extra=data.get("extra", {}),
         )
 
@@ -868,6 +880,8 @@ def load_gateway_config() -> GatewayConfig:
                         bridged["channel_prompts"] = channel_prompts
                 if "gateway_restart_notification" in platform_cfg:
                     bridged["gateway_restart_notification"] = platform_cfg["gateway_restart_notification"]
+                if "gateway_startup_notification" in platform_cfg:
+                    bridged["gateway_startup_notification"] = platform_cfg["gateway_startup_notification"]
                 enabled_was_explicit = "enabled" in platform_cfg
                 if not bridged and not enabled_was_explicit:
                     continue

gateway/run.py

@@ -538,6 +538,9 @@ def _restart_notification_pending() -> bool:
     return (_hermes_home / ".restart_notify.json").exists()
 
 
+_GATEWAY_STARTUP_NOTIFICATION_DEBOUNCE_SECONDS = 300.0
+
+
 # Mark this process as a gateway so cli.py's module-level load_cli_config()
 # knows not to clobber TERMINAL_CWD if lazily imported.
 os.environ["_HERMES_GATEWAY"] = "1"
@@ -4056,6 +4059,8 @@ async def start(self) -> bool:
             await self._send_home_channel_startup_notifications(
                 skip_targets=skip_home_targets,
             )
+        else:
+            await self._send_configured_gateway_startup_notifications()
 
         # Automatically continue fresh sessions that were interrupted by the
         # previous gateway restart/shutdown.  The resume_pending flag is cleared
@@ -14069,17 +14074,32 @@ async def _send_home_channel_startup_notifications(
         self,
         *,
         skip_targets: Optional[set[tuple[str, str, Optional[str]]]] = None,
+        require_startup_opt_in: bool = False,
+        debounce_seconds: Optional[float] = None,
     ) -> set[tuple[str, str, Optional[str]]]:
         """Notify configured home channels that the gateway is back online.
 
         The notification is best-effort and sent once per connected platform
         home channel. ``skip_targets`` lets startup avoid duplicate messages
         when a more specific restart notification is queued for the same chat.
+        ``require_startup_opt_in`` is used for ordinary process starts so
+        upstream defaults do not spam every configured home channel.
         """
         delivered: set[tuple[str, str, Optional[str]]] = set()
         skipped = skip_targets or set()
         message = "♻️ Gateway online — Hermes is back and ready."
 
+        debounce_path = _hermes_home / ".startup_notify_last.json"
+        now = time.time()
+        last_sent: dict[str, Any] = {}
+        if debounce_seconds is not None:
+            try:
+                loaded = json.loads(debounce_path.read_text()) if debounce_path.exists() else {}
+                if isinstance(loaded, dict):
+                    last_sent = loaded
+            except Exception:
+                last_sent = {}
+
         for platform, adapter in self.adapters.items():
             home = self.config.get_home_channel(platform)
             if not home or not home.chat_id:
@@ -14092,11 +14112,34 @@ async def _send_home_channel_startup_notifications(
                     platform.value,
                 )
                 continue
+            if require_startup_opt_in and not getattr(
+                platform_cfg, "gateway_startup_notification", False
+            ):
+                logger.debug(
+                    "Home-channel startup notification skipped: %s has gateway_startup_notification=false",
+                    platform.value,
+                )
+                continue
 
             target = (platform.value, str(home.chat_id), str(home.thread_id) if home.thread_id else None)
             if target in skipped or target in delivered:
                 continue
 
+            target_key = "|".join(part if part is not None else "" for part in target)
+            if debounce_seconds is not None:
+                try:
+                    previous = float(last_sent.get(target_key, 0))
+                except (TypeError, ValueError):
+                    previous = 0.0
+                elapsed = now - previous
+                if 0 <= elapsed < debounce_seconds:
+                    logger.info(
+                        "Home-channel startup notification debounced for %s:%s",
+                        platform.value,
+                        home.chat_id,
+                    )
+                    continue
+
             try:
                 metadata = {"thread_id": home.thread_id} if home.thread_id else None
                 if metadata:
@@ -14113,6 +14156,8 @@ async def _send_home_channel_startup_notifications(
                     continue
 
                 delivered.add(target)
+                if debounce_seconds is not None:
+                    last_sent[target_key] = now
                 logger.info(
                     "Sent home-channel startup notification to %s:%s",
                     platform.value,
@@ -14126,8 +14171,31 @@ async def _send_home_channel_startup_notifications(
                     exc,
                 )
 
+        if debounce_seconds is not None and delivered:
+            try:
+                atomic_json_write(debounce_path, last_sent)
+            except Exception as exc:
+                logger.debug("Failed to persist startup notification debounce state: %s", exc)
+
         return delivered
 
+    async def _send_configured_gateway_startup_notifications(
+        self,
+        *,
+        skip_targets: Optional[set[tuple[str, str, Optional[str]]]] = None,
+    ) -> set[tuple[str, str, Optional[str]]]:
+        """Send opt-in home-channel notifications after ordinary gateway startup.
+
+        This covers external restarts (systemd, host backup windows, process
+        recovery) where no /restart marker exists. The per-target debounce keeps
+        crash loops from spamming operator chats.
+        """
+        return await self._send_home_channel_startup_notifications(
+            skip_targets=skip_targets,
+            require_startup_opt_in=True,
+            debounce_seconds=_GATEWAY_STARTUP_NOTIFICATION_DEBOUNCE_SECONDS,
+        )
+
     def _set_session_env(self, context: SessionContext) -> list:
         """Set session context variables for the current async task.
 

tests/gateway/restart_test_helpers.py

@@ -100,6 +100,9 @@ def make_restart_runner(
     runner._send_home_channel_startup_notifications = (
         GatewayRunner._send_home_channel_startup_notifications.__get__(runner, GatewayRunner)
     )
+    runner._send_configured_gateway_startup_notifications = (
+        GatewayRunner._send_configured_gateway_startup_notifications.__get__(runner, GatewayRunner)
+    )
     runner._status_action_label = GatewayRunner._status_action_label.__get__(
         runner, GatewayRunner
     )

tests/gateway/test_config.py

@@ -70,6 +70,19 @@ def test_gateway_restart_notification_coerces_quoted_false(self):
         restored = PlatformConfig.from_dict({"gateway_restart_notification": "false"})
         assert restored.gateway_restart_notification is False
 
+    def test_gateway_startup_notification_defaults_false(self):
+        assert PlatformConfig().gateway_startup_notification is False
+        assert PlatformConfig.from_dict({}).gateway_startup_notification is False
+
+    def test_gateway_startup_notification_roundtrip_true(self):
+        pc = PlatformConfig(enabled=True, gateway_startup_notification=True)
+        restored = PlatformConfig.from_dict(pc.to_dict())
+        assert restored.gateway_startup_notification is True
+
+    def test_gateway_startup_notification_coerces_quoted_true(self):
+        restored = PlatformConfig.from_dict({"gateway_startup_notification": "true"})
+        assert restored.gateway_startup_notification is True
+
 
 class TestGetConnectedPlatforms:
     def test_returns_enabled_with_token(self):

tests/gateway/test_restart_notification.py

@@ -337,6 +337,77 @@ async def test_send_home_channel_startup_notification_ignores_false_send_result(
     adapter.send.assert_called_once()
 
 
+@pytest.mark.asyncio
+async def test_send_opt_in_gateway_startup_notification_to_home(
+    tmp_path, monkeypatch
+):
+    """External restarts can notify home channels when explicitly opted in."""
+    monkeypatch.setattr(gateway_run, "_hermes_home", tmp_path)
+
+    runner, adapter = make_restart_runner()
+    runner.config.platforms[Platform.TELEGRAM].home_channel = HomeChannel(
+        platform=Platform.TELEGRAM,
+        chat_id="home-42",
+        name="Ops Home",
+    )
+    runner.config.platforms[Platform.TELEGRAM].gateway_startup_notification = True
+    adapter.send = AsyncMock(return_value=SendResult(success=True, message_id="home"))
+
+    delivered = await runner._send_configured_gateway_startup_notifications()
+
+    assert delivered == {("telegram", "home-42", None)}
+    adapter.send.assert_called_once_with(
+        "home-42",
+        "♻️ Gateway online — Hermes is back and ready.",
+    )
+
+
+@pytest.mark.asyncio
+async def test_send_opt_in_gateway_startup_notification_skips_by_default(
+    tmp_path, monkeypatch
+):
+    """Every-startup pings are opt-in so upstream defaults do not get noisy."""
+    monkeypatch.setattr(gateway_run, "_hermes_home", tmp_path)
+
+    runner, adapter = make_restart_runner()
+    runner.config.platforms[Platform.TELEGRAM].home_channel = HomeChannel(
+        platform=Platform.TELEGRAM,
+        chat_id="home-42",
+        name="Ops Home",
+    )
+    adapter.send = AsyncMock()
+
+    delivered = await runner._send_configured_gateway_startup_notifications()
+
+    assert delivered == set()
+    adapter.send.assert_not_called()
+
+
+@pytest.mark.asyncio
+async def test_send_opt_in_gateway_startup_notification_debounces_target(
+    tmp_path, monkeypatch
+):
+    """Crash-loop protection: do not spam the same home target repeatedly."""
+    monkeypatch.setattr(gateway_run, "_hermes_home", tmp_path)
+    monkeypatch.setattr(gateway_run.time, "time", lambda: 1000.0)
+
+    runner, adapter = make_restart_runner()
+    runner.config.platforms[Platform.TELEGRAM].home_channel = HomeChannel(
+        platform=Platform.TELEGRAM,
+        chat_id="home-42",
+        name="Ops Home",
+    )
+    runner.config.platforms[Platform.TELEGRAM].gateway_startup_notification = True
+    adapter.send = AsyncMock(return_value=SendResult(success=True, message_id="home"))
+
+    first = await runner._send_configured_gateway_startup_notifications()
+    second = await runner._send_configured_gateway_startup_notifications()
+
+    assert first == {("telegram", "home-42", None)}
+    assert second == set()
+    adapter.send.assert_called_once()
+
+
 # ── _send_restart_notification ───────────────────────────────────────────