fix(gateway): coerce plaintext "restart gateway" DMs to /restart

Narrow plaintext shortcut that rewrites a tiny set of admin phrases
("restart gateway", "restart the gateway", "restart hermes") into the
/restart slash command, but only in DMs. Scope is intentionally tight:

- DM text messages only — group chats keep natural-language semantics
- Exact restart-style phrases only
- Skips anything already starting with "/"

Without this, the LLM can receive "restart gateway" as a user turn and
try to satisfy it via the terminal tool (systemctl restart ...). That
kills the gateway while the originating agent is still running, which
leaves systemd in "draining" state waiting on a process it's about to
kill. Routing the phrase to the slash-command dispatcher bypasses the
agent loop and uses the existing restart machinery (request_restart).

Called once, at the adapter level in BasePlatformAdapter.handle_message,
so every platform gets it for free and pending-message reinjection is
covered by the same call site.

Adds 2 Telegram-parametrized e2e tests: DM routes to request_restart,
group chats fall through to the normal agent path.

tests/e2e/conftest.py

@@ -125,13 +125,13 @@ from gateway.platforms.slack import SlackAdapter  # noqa: E402
 
 # Platform-generic factories
 
-def make_source(platform: Platform, chat_id: str = "e2e-chat-1", user_id: str = "e2e-user-1") -> SessionSource:
+def make_source(platform: Platform, chat_id: str = "e2e-chat-1", user_id: str = "e2e-user-1", chat_type: str = "dm") -> SessionSource:
     return SessionSource(
         platform=platform,
         chat_id=chat_id,
         user_id=user_id,
         user_name="e2e_tester",
-        chat_type="dm",
+        chat_type=chat_type,
     )
 
 
@@ -147,10 +147,16 @@ def make_session_entry(platform: Platform, source: SessionSource = None) -> Sess
     )
 
 
-def make_event(platform: Platform, text: str = "/help", chat_id: str = "e2e-chat-1", user_id: str = "e2e-user-1") -> MessageEvent:
+def make_event(
+    platform: Platform,
+    text: str = "/help",
+    chat_id: str = "e2e-chat-1",
+    user_id: str = "e2e-user-1",
+    chat_type: str = "dm",
+) -> MessageEvent:
     return MessageEvent(
         text=text,
-        source=make_source(platform, chat_id, user_id),
+        source=make_source(platform, chat_id, user_id, chat_type),
         message_id=f"msg-{uuid.uuid4().hex[:8]}",
     )
 
@@ -185,6 +191,23 @@ def make_runner(platform: Platform, session_entry: SessionEntry = None) -> "Gate
     runner._running_agents = {}
     runner._pending_messages = {}
     runner._pending_approvals = {}
+    runner._shutdown_event = asyncio.Event()
+    runner._exit_reason = None
+    runner._exit_code = None
+    runner._background_tasks = set()
+    runner._draining = False
+    runner._restart_requested = False
+    runner._restart_task_started = False
+    runner._restart_detached = False
+    runner._restart_via_service = False
+    from gateway.restart import DEFAULT_GATEWAY_RESTART_DRAIN_TIMEOUT
+    runner._restart_drain_timeout = DEFAULT_GATEWAY_RESTART_DRAIN_TIMEOUT
+    runner._stop_task = None
+    runner._busy_input_mode = "interrupt"
+    runner._running_agents_ts = {}
+    runner._pending_model_notes = {}
+    runner._update_prompt_pending = {}
+    runner._voice_mode = {}
     runner._session_db = None
     runner._reasoning_config = None
     runner._provider_routing = {}
@@ -193,6 +216,7 @@ def make_runner(platform: Platform, session_entry: SessionEntry = None) -> "Gate
 
     runner._is_user_authorized = lambda _source: True
     runner._set_session_env = lambda _context: None
+    runner._handle_message_with_agent = AsyncMock(return_value="agent-handled-default")
     runner._should_send_voice_reply = lambda *_a, **_kw: False
     runner._send_voice_reply = AsyncMock()
     runner._capture_gateway_honcho_if_configured = lambda *a, **kw: None

tests/e2e/test_platform_commands.py

@@ -11,10 +11,11 @@ Tests are parametrized over platforms via the ``platform`` fixture in conftest.
 """
 
 import asyncio
-from unittest.mock import AsyncMock
+from unittest.mock import AsyncMock, MagicMock
 
 import pytest
 
+from gateway.config import Platform
 from gateway.platforms.base import SendResult
 from tests.e2e.conftest import make_event, send_and_capture
 
@@ -82,6 +83,37 @@ class TestSlashCommands:
         # Either shows the mode cycle or tells user to enable it in config
         assert "verbose" in response_text.lower() or "tool_progress" in response_text
 
+    @pytest.mark.asyncio
+    async def test_plaintext_restart_gateway_routes_to_safe_restart_command(self, adapter, runner, platform, monkeypatch):
+        if platform != Platform.TELEGRAM:
+            pytest.skip("Plaintext restart shortcut is intentionally DM/Telegram-focused")
+
+        monkeypatch.setenv("INVOCATION_ID", "e2e-systemd")
+        runner.request_restart = MagicMock(return_value=True)
+
+        send = await send_and_capture(adapter, "restart gateway", platform)
+
+        send.assert_called_once()
+        response_text = send.call_args[1].get("content") or send.call_args[0][1]
+        assert "restart" in response_text.lower() or "draining" in response_text.lower()
+        runner.request_restart.assert_called_once_with(detached=False, via_service=True)
+
+    @pytest.mark.asyncio
+    async def test_plaintext_restart_gateway_in_group_stays_plain_text(self, adapter, runner, platform, monkeypatch):
+        if platform != Platform.TELEGRAM:
+            pytest.skip("Shortcut scope is only verified for Telegram here")
+
+        monkeypatch.setenv("INVOCATION_ID", "e2e-systemd")
+        runner.request_restart = MagicMock(return_value=True)
+        runner._handle_message_with_agent = AsyncMock(return_value="agent-handled")
+
+        send = await send_and_capture(adapter, "restart gateway", platform, chat_id="group-chat-1", user_id="u1", chat_type="group")
+
+        send.assert_called_once()
+        response_text = send.call_args[1].get("content") or send.call_args[0][1]
+        assert response_text == "agent-handled"
+        runner.request_restart.assert_not_called()
+
     @pytest.mark.asyncio
     async def test_personality_lists_options(self, adapter, platform):
         send = await send_and_capture(adapter, "/personality", platform)