fix(gateway): stop slow restart redelivery loops

gateway/run.py

@@ -7867,13 +7867,14 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
             await asyncio.sleep(1.0)
 
         # Notify the chat that initiated /restart that the gateway is back.
+        chat_restart_notification_pending = _restart_notification_pending()
         planned_restart_notification_pending = _planned_restart_notification_pending()
         # Capture, before _send_restart_notification() unlinks the marker,
         # whether this process booted from a chat-originated /restart. Used as
         # a one-shot signal by the /restart redelivery guard so a missing
         # dedup marker only suppresses a /restart when we KNOW we just came out
         # of a restart cycle (see _is_stale_restart_redelivery).
-        if _restart_notification_pending() or planned_restart_notification_pending:
+        if chat_restart_notification_pending:
             self._booted_from_restart = True
         await self._send_restart_notification()
 
@@ -13815,8 +13816,8 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
         The previous gateway wrote ``.restart_last_processed.json`` with the
         triggering platform + update_id when it processed the /restart.  If
         we now see a /restart on the same platform with an update_id <= that
-        recorded value AND the marker is recent (< 5 minutes), it's a
-        redelivery and should be ignored.
+        recorded value, it is a redelivery when this process booted from that
+        restart. Otherwise the marker must still be recent (< 5 minutes).
 
         Only applies to Telegram today (the only platform that exposes a
         numeric cross-session update ordering); other platforms return False.
@@ -13868,6 +13869,19 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
         recorded_uid = data.get("update_id")
         if not isinstance(recorded_uid, int):
             return False
+        if event.platform_update_id > recorded_uid:
+            return False
+
+        # A service-managed restart can legitimately take longer than the
+        # marker's normal five-minute trust window while adapters, cron, and
+        # in-flight deliveries drain. If this process booted from the recorded
+        # chat restart, the first same-or-older update is still that restart's
+        # redelivery regardless of elapsed wall time. Consume the boot signal
+        # one-shot so a later genuine command is evaluated normally.
+        if getattr(self, "_booted_from_restart", False):
+            self._booted_from_restart = False
+            return True
+
         # Staleness guard: ignore markers older than 5 minutes.  A legitimately
         # old marker (e.g. crash recovery where notify never fired) should not
         # swallow a fresh /restart from the user.
@@ -13875,7 +13889,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
         if isinstance(requested_at, (int, float)):
             if time.time() - requested_at > 300:
                 return False
-        return event.platform_update_id <= recorded_uid
+        return True
 
 
 

tests/gateway/test_restart_redelivery_dedup.py

@@ -148,6 +148,43 @@ async def test_stale_marker_older_than_5min_does_not_block(tmp_path, monkeypatch
     runner.request_restart.assert_called_once()
 
 
+@pytest.mark.asyncio
+async def test_slow_service_restart_still_ignores_same_update(tmp_path, monkeypatch):
+    """A slow drain must not outlive dedup when this boot came from /restart.
+
+    Service-managed shutdown can take more than five minutes while in-flight
+    gateway work drains. The replacement process still knows it booted from
+    the recorded chat restart, so the first same update must be suppressed
+    instead of requesting exit 75 again and entering a supervisor loop.
+    """
+    monkeypatch.setattr(gateway_run, "_hermes_home", tmp_path)
+    monkeypatch.setenv("INVOCATION_ID", "systemd-test")
+
+    marker = tmp_path / ".restart_last_processed.json"
+    marker.write_text(
+        json.dumps(
+            {
+                "platform": "telegram",
+                "update_id": 12345,
+                "requested_at": time.time() - 1200,
+            }
+        )
+    )
+
+    runner, _adapter = make_restart_runner()
+    request_restart = MagicMock()
+    monkeypatch.setattr(runner, "request_restart", request_restart)
+    runner._booted_from_restart = True
+
+    result = await runner._handle_restart_command(
+        _make_restart_event(update_id=12345)
+    )
+
+    assert result == ""
+    request_restart.assert_not_called()
+    assert runner._booted_from_restart is False
+
+
 @pytest.mark.asyncio
 async def test_no_marker_file_allows_restart(tmp_path, monkeypatch):
     """Clean gateway start (no prior marker) processes /restart normally."""

tests/hermes_cli/test_gateway.py

@@ -1,8 +1,12 @@
 """Tests for hermes_cli.gateway."""
 
 import argparse
+import os
+import pty
 import signal
+import subprocess
 import sys
+import textwrap
 from types import ModuleType, SimpleNamespace
 
 import pytest
@@ -78,6 +82,87 @@ def test_run_gateway_exits_nonzero_when_start_gateway_reports_failure(monkeypatc
     assert calls == [(True, None)]
 
 
+@pytest.mark.skipif(sys.platform == "win32", reason="POSIX PTY coverage")
+@pytest.mark.parametrize(
+    ("stdin_is_tty", "outcome", "expected_exit"),
+    [
+        (True, "systemexit:75", 75),
+        (False, "systemexit:75", 75),
+        (False, "systemexit:78", 78),
+        (False, "failure", 1),
+    ],
+)
+def test_gateway_run_subprocess_preserves_daemon_exit_codes(
+    tmp_path, stdin_is_tty, outcome, expected_exit
+):
+    """TTY state must not rewrite the gateway's process-level exit contract.
+
+    Exit 75 is the intentional systemd/launchd restart handoff, exit 78 is a
+    fatal configuration error, and a false startup result is a generic failure.
+    In particular, a non-TTY daemon launch must not blanket-catch SystemExit,
+    because doing so would hide genuine startup/configuration failures.
+    """
+    script = textwrap.dedent(
+        """
+        import os
+        import sys
+        import types
+
+        import hermes_cli.gateway as gateway_cli
+
+        outcome = os.environ["HERMES_TEST_GATEWAY_OUTCOME"]
+
+        async def start_gateway(*, replace, verbosity):
+            if outcome == "failure":
+                return False
+            raise SystemExit(int(outcome.split(":", 1)[1]))
+
+        fake_run = types.ModuleType("gateway.run")
+        fake_run.start_gateway = start_gateway
+        sys.modules["gateway.run"] = fake_run
+
+        gateway_cli._guard_official_docker_root_gateway = lambda: None
+        gateway_cli._guard_named_profile_under_multiplexer = lambda force=False: None
+        gateway_cli._guard_supervised_gateway_conflict = lambda force=False: None
+        gateway_cli._guard_existing_gateway_process_conflict = lambda replace=False: None
+        gateway_cli.supports_systemd_services = lambda: False
+        gateway_cli.run_gateway()
+        """
+    )
+    env = {
+        **os.environ,
+        "HERMES_HOME": str(tmp_path),
+        "HERMES_GATEWAY_EXIT_DIAG": "0",
+        "HERMES_TEST_GATEWAY_OUTCOME": outcome,
+        "INVOCATION_ID": "systemd-test",
+    }
+
+    master_fd = slave_fd = None
+    try:
+        if stdin_is_tty:
+            master_fd, slave_fd = pty.openpty()
+            stdin = slave_fd
+        else:
+            stdin = subprocess.DEVNULL
+        completed = subprocess.run(
+            [sys.executable, "-c", script],
+            stdin=stdin,
+            stdout=subprocess.PIPE,
+            stderr=subprocess.PIPE,
+            text=True,
+            env=env,
+            timeout=30,
+            check=False,
+        )
+    finally:
+        if slave_fd is not None:
+            os.close(slave_fd)
+        if master_fd is not None:
+            os.close(master_fd)
+
+    assert completed.returncode == expected_exit, completed.stderr
+
+
 def test_run_gateway_refuses_root_in_official_docker(monkeypatch, tmp_path, capsys):
     project_root = tmp_path / "opt" / "hermes"
     (project_root / "docker").mkdir(parents=True)