fix(gateway): guard stale /restart redelivery when dedup marker is missing (#56107)

When .restart_last_processed.json goes missing, a redelivered /restart from
Telegram polling can no longer be caught by the update_id comparison, so it
re-restarts the gateway forever (issue #18528, reported by @dontcallmejames
who hit it in production — gateway restarting every ~2min, zero messages
processed).

Fallback: on marker-missing, suppress the /restart only when we can confirm
we just came out of a restart cycle (_booted_from_restart, captured at startup
from .restart_notify.json before it is unlinked) AND the process is still
within a 60s post-boot window. Consumed one-shot. This closes the loop without
swallowing a genuine first /restart on a fresh boot — the flaw in the original
bare-uptime approach.

Credit to @dontcallmejames for the diagnosis and original patch.

tests/gateway/test_restart_redelivery_dedup.py

@@ -244,3 +244,74 @@ async def test_different_platform_bypasses_dedup(tmp_path, monkeypatch):
 
     assert "Restarting gateway" in result
     runner.request_restart.assert_called_once()
+
+
+@pytest.mark.asyncio
+async def test_marker_missing_but_booted_from_restart_ignores_redelivery(tmp_path, monkeypatch):
+    """Missing marker + just booted from a /restart + young process → treat as stale.
+
+    Reproduces the infinite-loop scenario (issue #18528): the dedup marker went
+    missing, so the update_id comparison can't run. Because this process booted
+    from a chat-originated /restart and is still within the post-boot window,
+    the redelivered /restart is suppressed instead of re-restarting the gateway.
+    """
+    monkeypatch.setattr(gateway_run, "_hermes_home", tmp_path)
+    monkeypatch.delenv("INVOCATION_ID", raising=False)
+
+    runner, _adapter = make_restart_runner()
+    runner.request_restart = MagicMock(return_value=True)
+    runner._booted_from_restart = True
+    runner._startup_time = time.time()
+
+    event = _make_restart_event(update_id=100)
+    result = await runner._handle_restart_command(event)
+
+    assert result == ""  # silently ignored
+    runner.request_restart.assert_not_called()
+    # One-shot: the flag is consumed so a later legitimate /restart is honored.
+    assert runner._booted_from_restart is False
+
+
+@pytest.mark.asyncio
+async def test_marker_missing_fresh_boot_allows_restart(tmp_path, monkeypatch):
+    """Missing marker on a genuine fresh boot (not from /restart) → /restart proceeds.
+
+    The guard must NOT swallow the first /restart a user sends shortly after a
+    normal (non-restart) startup: _booted_from_restart stays False, so the
+    fallback returns False and the restart goes through.
+    """
+    monkeypatch.setattr(gateway_run, "_hermes_home", tmp_path)
+    monkeypatch.delenv("INVOCATION_ID", raising=False)
+
+    runner, _adapter = make_restart_runner()
+    runner.request_restart = MagicMock(return_value=True)
+    runner._booted_from_restart = False
+    runner._startup_time = time.time()
+
+    event = _make_restart_event(update_id=100)
+    result = await runner._handle_restart_command(event)
+
+    assert "Restarting gateway" in result
+    runner.request_restart.assert_called_once()
+
+
+@pytest.mark.asyncio
+async def test_marker_missing_booted_from_restart_but_old_process_allows(tmp_path, monkeypatch):
+    """Missing marker + booted from /restart but past the window → /restart proceeds.
+
+    A /restart arriving long after boot is a genuine user action, not a boot-time
+    redelivery, so the uptime bound stops the guard from suppressing it forever.
+    """
+    monkeypatch.setattr(gateway_run, "_hermes_home", tmp_path)
+    monkeypatch.delenv("INVOCATION_ID", raising=False)
+
+    runner, _adapter = make_restart_runner()
+    runner.request_restart = MagicMock(return_value=True)
+    runner._booted_from_restart = True
+    runner._startup_time = time.time() - 120  # well past the 60s window
+
+    event = _make_restart_event(update_id=100)
+    result = await runner._handle_restart_command(event)
+
+    assert "Restarting gateway" in result
+    runner.request_restart.assert_called_once()