fix(gateway): clean service restart notifications

tests/gateway/restart_test_helpers.py

@@ -69,6 +69,7 @@ def make_restart_runner(
     runner._restart_task_started = False
     runner._restart_detached = False
     runner._restart_via_service = False
+    runner._restart_command_source = None
     runner._restart_drain_timeout = DEFAULT_GATEWAY_RESTART_DRAIN_TIMEOUT
     runner._stop_task = None
     runner._busy_input_mode = "interrupt"

tests/gateway/test_gateway_shutdown.py

@@ -3,6 +3,8 @@ from unittest.mock import AsyncMock, MagicMock, patch
 
 import pytest
 
+import gateway.run as gateway_run
+from gateway.config import HomeChannel, Platform
 from gateway.platforms.base import MessageEvent
 from gateway.restart import GATEWAY_SERVICE_RESTART_EXIT_CODE
 from gateway.session import build_session_key
@@ -132,16 +134,127 @@ async def test_gateway_stop_interrupts_after_drain_timeout():
 
 
 @pytest.mark.asyncio
-async def test_gateway_stop_service_restart_sets_named_exit_code():
+async def test_gateway_stop_systemd_service_restart_exits_cleanly(tmp_path, monkeypatch):
+    monkeypatch.setattr(gateway_run, "_hermes_home", tmp_path)
     runner, adapter = make_restart_runner()
     adapter.disconnect = AsyncMock()
+    monkeypatch.setenv("INVOCATION_ID", "systemd-test")
+    runner._launch_systemd_restart_shortcut = MagicMock()
 
     with patch("gateway.status.remove_pid_file"), patch("gateway.status.write_runtime_status"):
         await runner.stop(restart=True, service_restart=True)
 
+    runner._launch_systemd_restart_shortcut.assert_called_once_with()
+    assert runner._exit_code == 0
+    assert (tmp_path / ".restart_pending.json").exists()
+
+
+@pytest.mark.asyncio
+async def test_gateway_stop_launchd_service_restart_keeps_nonzero_exit(tmp_path, monkeypatch):
+    monkeypatch.setattr(gateway_run, "_hermes_home", tmp_path)
+    runner, adapter = make_restart_runner()
+    adapter.disconnect = AsyncMock()
+
+    with patch("gateway.run.sys.platform", "darwin"), patch(
+        "gateway.status.remove_pid_file"
+    ), patch("gateway.status.write_runtime_status"):
+        await runner.stop(restart=True, service_restart=True)
+
     assert runner._exit_code == GATEWAY_SERVICE_RESTART_EXIT_CODE
 
 
+@pytest.mark.asyncio
+async def test_restart_shutdown_warning_uses_restart_command_reply_anchor_for_active_session():
+    runner, adapter = make_restart_runner()
+    source = make_restart_source(thread_id="42")
+    session_key = build_session_key(source)
+    runner._running_agents = {session_key: MagicMock()}
+    runner._cache_session_source(session_key, source)
+    restart_source = make_restart_source(thread_id="42")
+    restart_source.message_id = "restart-command"
+    runner._restart_requested = True
+    runner._restart_command_source = restart_source
+    runner.config.platforms[Platform.TELEGRAM].home_channel = HomeChannel(
+        platform=Platform.TELEGRAM,
+        chat_id=source.chat_id,
+        name="Telegram",
+        thread_id=source.thread_id,
+    )
+
+    await runner._notify_active_sessions_of_shutdown()
+
+    assert len(adapter.sent_calls) == 1
+    chat_id, message, metadata = adapter.sent_calls[0]
+    assert chat_id == source.chat_id
+    assert "Gateway restarting" in message
+    assert metadata["thread_id"] == source.thread_id
+    assert metadata["telegram_dm_topic_reply_fallback"] is True
+    assert metadata["direct_messages_topic_id"] == source.thread_id
+    assert metadata["telegram_reply_to_message_id"] == "restart-command"
+
+
+@pytest.mark.asyncio
+async def test_in_chat_restart_skips_home_shutdown_even_with_active_session():
+    runner, adapter = make_restart_runner()
+    source = make_restart_source(thread_id="42")
+    session_key = build_session_key(source)
+    runner._running_agents = {session_key: MagicMock()}
+    runner._cache_session_source(session_key, source)
+    restart_source = make_restart_source(thread_id="42")
+    restart_source.message_id = "restart-command"
+    runner._restart_requested = True
+    runner._restart_command_source = restart_source
+    runner.config.platforms[Platform.TELEGRAM].home_channel = HomeChannel(
+        platform=Platform.TELEGRAM,
+        chat_id="home-chat",
+        name="Telegram Home",
+    )
+
+    await runner._notify_active_sessions_of_shutdown()
+
+    assert len(adapter.sent_calls) == 1
+    chat_id, message, metadata = adapter.sent_calls[0]
+    assert chat_id == source.chat_id
+    assert "Gateway restarting" in message
+    assert metadata["telegram_reply_to_message_id"] == "restart-command"
+
+
+@pytest.mark.asyncio
+async def test_idle_in_chat_restart_does_not_send_interruption_warning():
+    runner, adapter = make_restart_runner()
+    source = make_restart_source(thread_id="42")
+    source.message_id = "restart-command"
+    runner._restart_requested = True
+    runner._restart_command_source = source
+    runner.config.platforms[Platform.TELEGRAM].home_channel = HomeChannel(
+        platform=Platform.TELEGRAM,
+        chat_id=source.chat_id,
+        name="Telegram",
+        thread_id=source.thread_id,
+    )
+
+    await runner._notify_active_sessions_of_shutdown()
+
+    assert adapter.sent_calls == []
+
+
+@pytest.mark.asyncio
+async def test_in_chat_restart_does_not_write_home_startup_marker(tmp_path, monkeypatch):
+    monkeypatch.setattr(gateway_run, "_hermes_home", tmp_path)
+    runner, adapter = make_restart_runner()
+    adapter.disconnect = AsyncMock()
+    source = make_restart_source(thread_id="42")
+    source.message_id = "restart-command"
+    runner._restart_command_source = source
+    runner._launch_systemd_restart_shortcut = MagicMock()
+    monkeypatch.setenv("INVOCATION_ID", "systemd-test")
+
+    with patch("gateway.status.remove_pid_file"), patch("gateway.status.write_runtime_status"):
+        await runner.stop(restart=True, service_restart=True)
+
+    assert not (tmp_path / ".restart_pending.json").exists()
+
+
 @pytest.mark.asyncio
 async def test_drain_active_agents_throttles_status_updates():
     runner, _adapter = make_restart_runner()

tests/gateway/test_restart_notification.py

@@ -32,6 +32,19 @@ def test_restart_notification_pending_true_with_marker(tmp_path, monkeypatch):
     assert gateway_run._restart_notification_pending() is True
 
 
+def test_planned_restart_notification_pending_roundtrip(tmp_path, monkeypatch):
+    monkeypatch.setattr(gateway_run, "_hermes_home", tmp_path)
+    marker = tmp_path / ".restart_pending.json"
+
+    assert gateway_run._planned_restart_notification_pending() is False
+    marker.write_text("{}")
+    assert gateway_run._planned_restart_notification_pending() is True
+
+    gateway_run._clear_planned_restart_notification()
+
+    assert gateway_run._planned_restart_notification_pending() is False
+
+
 # ── _handle_restart_command writes .restart_notify.json ──────────────────
 
 
@@ -60,6 +73,7 @@ async def test_restart_command_writes_notify_file(tmp_path, monkeypatch):
     assert data["platform"] == "telegram"
     assert data["chat_id"] == "42"
     assert data["chat_type"] == "dm"
+    assert data["message_id"] == "m1"
     assert "thread_id" not in data  # no thread → omitted
 
 
@@ -127,6 +141,7 @@ async def test_restart_command_preserves_thread_id(tmp_path, monkeypatch):
     data = json.loads((tmp_path / ".restart_notify.json").read_text())
     assert data["chat_type"] == "dm"
     assert data["thread_id"] == "777"
+    assert data["message_id"] == "m2"
 
 
 @pytest.mark.asyncio
@@ -390,6 +405,7 @@ async def test_send_restart_notification_with_thread(tmp_path, monkeypatch):
         "chat_id": "99",
         "chat_type": "dm",
         "thread_id": "777",
+        "message_id": "m2",
     }))
 
     runner, adapter = make_restart_runner()
@@ -403,6 +419,7 @@ async def test_send_restart_notification_with_thread(tmp_path, monkeypatch):
         "thread_id": "777",
         "telegram_dm_topic_reply_fallback": True,
         "direct_messages_topic_id": "777",
+        "telegram_reply_to_message_id": "m2",
     }
     assert not notify_path.exists()
 
@@ -642,3 +659,28 @@ async def test_shutdown_notifications_use_cached_live_thread_source_when_origin_
         "⚠️ Gateway shutting down — Your current task will be interrupted.",
         metadata={"thread_id": "topic-7"},
     )
+
+
+@pytest.mark.asyncio
+async def test_restart_shutdown_notification_anchors_telegram_dm_topic():
+    runner, adapter = make_restart_runner()
+    runner._restart_requested = True
+    source = make_restart_source(chat_id="123456", thread_id="20197")
+    source.message_id = "462"
+    session_key = build_session_key(source)
+
+    runner._running_agents[session_key] = object()
+    runner.session_store._entries[session_key] = MagicMock(origin=source)
+    adapter.send = AsyncMock(return_value=SendResult(success=True, message_id="shutdown"))
+
+    await runner._notify_active_sessions_of_shutdown()
+
+    call = adapter.send.await_args
+    assert call.args[0] == "123456"
+    assert "Gateway restarting" in call.args[1]
+    assert call.kwargs["metadata"] == {
+        "thread_id": "20197",
+        "telegram_dm_topic_reply_fallback": True,
+        "direct_messages_topic_id": "20197",
+        "telegram_reply_to_message_id": "462",
+    }

tests/gateway/test_telegram_thread_fallback.py

@@ -597,6 +597,35 @@ async def test_send_uses_reply_fallback_for_hermes_dm_topics():
     assert "direct_messages_topic_id" not in call_log[0]
 
 
+@pytest.mark.asyncio
+async def test_send_uses_reply_anchor_when_direct_topic_fallback_metadata_exists():
+    """Restart/update replay metadata keeps the anchor authoritative when present."""
+    adapter = _make_adapter()
+    call_log = []
+
+    async def mock_send_message(**kwargs):
+        call_log.append(kwargs)
+        return SimpleNamespace(message_id=777)
+
+    adapter._bot = SimpleNamespace(send_message=mock_send_message)
+
+    result = await adapter.send(
+        chat_id="123",
+        content="test message",
+        metadata={
+            "thread_id": "20197",
+            "telegram_dm_topic_reply_fallback": True,
+            "direct_messages_topic_id": "20197",
+            "telegram_reply_to_message_id": "462",
+        },
+    )
+
+    assert result.success is True
+    assert call_log[0]["reply_to_message_id"] == 462
+    assert call_log[0]["message_thread_id"] == 20197
+    assert "direct_messages_topic_id" not in call_log[0]
+
+
 @pytest.mark.asyncio
 async def test_send_created_private_topic_uses_message_thread_without_anchor():
     """Topics created via createForumTopic are addressable by message_thread_id directly."""

tests/gateway/test_update_command.py

@@ -189,6 +189,7 @@ class TestHandleUpdateCommand:
         """Writes .update_pending.json with correct platform and chat info."""
         runner = _make_runner()
         event = _make_event(platform=Platform.TELEGRAM, chat_id="99999")
+        event.message_id = "m-update"
 
         fake_root = tmp_path / "project"
         fake_root.mkdir()
@@ -211,6 +212,7 @@ class TestHandleUpdateCommand:
         assert data["platform"] == "telegram"
         assert data["chat_id"] == "99999"
         assert data["chat_type"] == "dm"
+        assert data["message_id"] == "m-update"
         assert "timestamp" in data
         assert not (hermes_home / ".update_exit_code").exists()
 
@@ -223,6 +225,7 @@ class TestHandleUpdateCommand:
             chat_id="99999",
             thread_id="777",
         )
+        event.message_id = "m-update-thread"
 
         fake_root = tmp_path / "project"
         fake_root.mkdir()
@@ -241,6 +244,7 @@ class TestHandleUpdateCommand:
 
         data = json.loads((hermes_home / ".update_pending.json").read_text())
         assert data["thread_id"] == "777"
+        assert data["message_id"] == "m-update-thread"
 
     @pytest.mark.asyncio
     async def test_spawns_setsid(self, tmp_path):
@@ -472,6 +476,7 @@ class TestSendUpdateNotification:
             "chat_id": "67890",
             "chat_type": "dm",
             "thread_id": "777",
+            "message_id": "m-update-thread",
             "user_id": "12345",
         }
         (hermes_home / ".update_pending.json").write_text(json.dumps(pending))
@@ -488,6 +493,7 @@ class TestSendUpdateNotification:
             "thread_id": "777",
             "telegram_dm_topic_reply_fallback": True,
             "direct_messages_topic_id": "777",
+            "telegram_reply_to_message_id": "m-update-thread",
         }
 
     @pytest.mark.asyncio