feat(gateway): per-platform gateway_restart_notification flag

gateway/config.py

@@ -271,15 +271,23 @@ class PlatformConfig:
     # - "first": Only first chunk threads to user's message (default)
     # - "all": All chunks in multi-part replies thread to user's message
     reply_to_mode: str = "first"
-
+
+    # Whether the gateway is allowed to send "♻️ Gateway online" /
+    # "♻ Gateway restarted" lifecycle notifications on this platform.
+    # Default True preserves prior behavior. Set False on platforms used
+    # by end users (e.g. Slack) where operator-flavored restart pings are
+    # noise; keep True for back-channels where the operator wants them.
+    gateway_restart_notification: bool = True
+
     # Platform-specific settings
     extra: Dict[str, Any] = field(default_factory=dict)
-    
+
     def to_dict(self) -> Dict[str, Any]:
         result = {
             "enabled": self.enabled,
             "extra": self.extra,
             "reply_to_mode": self.reply_to_mode,
+            "gateway_restart_notification": self.gateway_restart_notification,
         }
         if self.token:
             result["token"] = self.token
@@ -288,19 +296,22 @@ def to_dict(self) -> Dict[str, Any]:
         if self.home_channel:
             result["home_channel"] = self.home_channel.to_dict()
         return result
-    
+
     @classmethod
     def from_dict(cls, data: Dict[str, Any]) -> "PlatformConfig":
         home_channel = None
         if "home_channel" in data:
             home_channel = HomeChannel.from_dict(data["home_channel"])
-        
+
         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(
+                data.get("gateway_restart_notification"), True
+            ),
             extra=data.get("extra", {}),
         )
 

gateway/run.py

@@ -2458,6 +2458,14 @@ async def _notify_active_sessions_of_shutdown(self) -> None:
                 if not adapter:
                     continue
 
+                platform_cfg = self.config.platforms.get(platform)
+                if platform_cfg is not None and not platform_cfg.gateway_restart_notification:
+                    logger.info(
+                        "Shutdown notification suppressed for active session: %s has gateway_restart_notification=false",
+                        platform_str,
+                    )
+                    continue
+
                 # Include thread_id if present so the message lands in the
                 # correct forum topic / thread.
                 metadata = {"thread_id": thread_id} if thread_id else None
@@ -2488,6 +2496,14 @@ async def _notify_active_sessions_of_shutdown(self) -> None:
             if not home or not home.chat_id:
                 continue
 
+            platform_cfg = self.config.platforms.get(platform)
+            if platform_cfg is not None and not platform_cfg.gateway_restart_notification:
+                logger.info(
+                    "Shutdown notification suppressed for home channel: %s has gateway_restart_notification=false",
+                    platform.value,
+                )
+                continue
+
             dedup_key = (platform.value, str(home.chat_id), str(home.thread_id) if home.thread_id else None)
             if dedup_key in notified:
                 continue
@@ -11386,6 +11402,14 @@ async def _send_restart_notification(self) -> Optional[tuple[str, str, Optional[
                 )
                 return None
 
+            platform_cfg = self.config.platforms.get(platform)
+            if platform_cfg is not None and not platform_cfg.gateway_restart_notification:
+                logger.info(
+                    "Restart notification suppressed: %s has gateway_restart_notification=false",
+                    platform_str,
+                )
+                return None
+
             metadata = {"thread_id": thread_id} if thread_id else None
             result = await adapter.send(
                 str(chat_id),
@@ -11437,6 +11461,14 @@ async def _send_home_channel_startup_notifications(
             if not home or not home.chat_id:
                 continue
 
+            platform_cfg = self.config.platforms.get(platform)
+            if platform_cfg is not None and not platform_cfg.gateway_restart_notification:
+                logger.info(
+                    "Home-channel startup notification suppressed: %s has gateway_restart_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

scripts/release.py

@@ -87,6 +87,7 @@
     "happy5318@users.noreply.github.com": "happy5318",
     "chengoak@users.noreply.github.com": "chengoak",
     "mrhanoi@outlook.com": "qxxaa",
+    "guillaume.meyer@outlook.com": "guillaumemeyer",
     "emelyanenko.kirill@gmail.com": "EmelyanenkoK",
     "lazycat.manatee@gmail.com": "manateelazycat",
     "bzarnitz13@gmail.com": "Beandon13",

tests/gateway/test_config.py

@@ -57,6 +57,19 @@ def test_from_dict_coerces_quoted_false_enabled(self):
         restored = PlatformConfig.from_dict({"enabled": "false"})
         assert restored.enabled is False
 
+    def test_gateway_restart_notification_defaults_true(self):
+        assert PlatformConfig().gateway_restart_notification is True
+        assert PlatformConfig.from_dict({}).gateway_restart_notification is True
+
+    def test_gateway_restart_notification_roundtrip_false(self):
+        pc = PlatformConfig(enabled=True, gateway_restart_notification=False)
+        restored = PlatformConfig.from_dict(pc.to_dict())
+        assert restored.gateway_restart_notification is False
+
+    def test_gateway_restart_notification_coerces_quoted_false(self):
+        restored = PlatformConfig.from_dict({"gateway_restart_notification": "false"})
+        assert restored.gateway_restart_notification is False
+
 
 class TestGetConnectedPlatforms:
     def test_returns_enabled_with_token(self):

tests/gateway/test_restart_drain.py

@@ -257,6 +257,40 @@ async def test_shutdown_notification_send_failure_does_not_block():
     await runner._notify_active_sessions_of_shutdown()
 
 
+@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."""
+    from gateway.config import Platform
+
+    runner, adapter = make_restart_runner()
+    runner._restart_requested = True
+    runner.config.platforms[Platform.TELEGRAM].gateway_restart_notification = False
+    session_key = "agent:main:telegram:dm:999"
+    runner._running_agents[session_key] = MagicMock()
+
+    await runner._notify_active_sessions_of_shutdown()
+
+    assert adapter.sent == []
+
+
+@pytest.mark.asyncio
+async def test_shutdown_notification_home_channel_suppressed_when_flag_disabled():
+    """Home-channel ping during shutdown is muted when the flag is False."""
+    from gateway.config import HomeChannel, Platform
+
+    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_restart_notification = False
+
+    await runner._notify_active_sessions_of_shutdown()
+
+    assert adapter.sent == []
+
+
 @pytest.mark.asyncio
 async def test_shutdown_notification_uses_persisted_origin_for_colon_ids():
     """Shutdown notifications should route from persisted origin, not reparsed keys."""

tests/gateway/test_restart_notification.py

@@ -496,6 +496,82 @@ async def test_send_restart_notification_logs_warning_on_sendresult_failure(
     assert not notify_path.exists()
 
 
+@pytest.mark.asyncio
+async def test_send_home_channel_startup_notification_skipped_when_flag_disabled(
+    tmp_path, monkeypatch
+):
+    """Per-platform opt-out: gateway_restart_notification=False mutes the home-channel ping."""
+    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_restart_notification = False
+    adapter.send = AsyncMock()
+
+    delivered = await runner._send_home_channel_startup_notifications()
+
+    assert delivered == set()
+    adapter.send.assert_not_called()
+
+
+@pytest.mark.asyncio
+async def test_send_home_channel_startup_notification_default_flag_true(
+    tmp_path, monkeypatch
+):
+    """Default behavior is unchanged: missing flag means notifications still fire."""
+    monkeypatch.setattr(gateway_run, "_hermes_home", tmp_path)
+
+    runner, adapter = make_restart_runner()
+    # Sanity-check the dataclass default — guards against future refactors
+    # silently flipping the default to False.
+    assert runner.config.platforms[Platform.TELEGRAM].gateway_restart_notification is True
+
+    runner.config.platforms[Platform.TELEGRAM].home_channel = HomeChannel(
+        platform=Platform.TELEGRAM,
+        chat_id="home-42",
+        name="Ops Home",
+    )
+    adapter.send = AsyncMock(return_value=SendResult(success=True, message_id="home"))
+
+    delivered = await runner._send_home_channel_startup_notifications()
+
+    assert delivered == {("telegram", "home-42", None)}
+    adapter.send.assert_called_once()
+
+
+@pytest.mark.asyncio
+async def test_send_restart_notification_skipped_when_flag_disabled(
+    tmp_path, monkeypatch
+):
+    """The /restart originator's notification also honors the per-platform flag.
+
+    Slack used by end users → flag off → no "Gateway restarted" message even
+    when an end user accidentally triggers /restart. The marker file is still
+    cleaned up so the notification doesn't leak into the next boot.
+    """
+    monkeypatch.setattr(gateway_run, "_hermes_home", tmp_path)
+
+    notify_path = tmp_path / ".restart_notify.json"
+    notify_path.write_text(json.dumps({
+        "platform": "telegram",
+        "chat_id": "42",
+    }))
+
+    runner, adapter = make_restart_runner()
+    runner.config.platforms[Platform.TELEGRAM].gateway_restart_notification = False
+    adapter.send = AsyncMock()
+
+    delivered_target = await runner._send_restart_notification()
+
+    assert delivered_target is None
+    adapter.send.assert_not_called()
+    assert not notify_path.exists()
+
+
 @pytest.mark.asyncio
 async def test_send_restart_notification_logs_info_on_sendresult_success(
     tmp_path, monkeypatch, caplog