Add reboot and version slash commands with permission guard

server/app/auth_service.py

@@ -44,6 +44,7 @@ PERMISSIONS: tuple[str, ...] = (
     "user.change_role",
     "role.manage",
     "server.manage_settings",
+    "server.allow_reboot",
 )
 
 PERMISSION_DESCRIPTIONS: dict[str, str] = {
@@ -63,6 +64,7 @@ PERMISSION_DESCRIPTIONS: dict[str, str] = {
     "user.change_role": "Allow assigning user roles.",
     "role.manage": "Allow creating, editing, and deleting roles.",
     "server.manage_settings": "Allow changing server settings.",
+    "server.allow_reboot": "Allow scheduling a server reboot from chat command.",
 }
 
 DEFAULT_ROLE_PERMISSIONS: dict[str, set[str]] = {

server/app/server.py

@@ -14,6 +14,7 @@ import logging
 import os
 import random
 import re
+import signal
 import ssl
 import time
 import uuid
@@ -185,6 +186,7 @@ class SignalingServer:
         self._clock_top_of_hour_markers: dict[str, str] = {}
         self._clock_alarm_markers: dict[str, str] = {}
         self._started_at_monotonic = time.monotonic()
+        self._pending_reboot_task: asyncio.Task[None] | None = None
 
     @staticmethod
     def _resolve_server_version() -> str:
@@ -1218,6 +1220,11 @@ class SignalingServer:
             ):
                 await asyncio.Future()
         finally:
+            if self._pending_reboot_task is not None:
+                self._pending_reboot_task.cancel()
+                with suppress(asyncio.CancelledError):
+                    await self._pending_reboot_task
+                self._pending_reboot_task = None
             if self._clock_announce_task is not None:
                 self._clock_announce_task.cancel()
                 with suppress(asyncio.CancelledError):
@@ -1598,6 +1605,24 @@ class SignalingServer:
         elapsed_seconds = int(max(0.0, time.monotonic() - self._started_at_monotonic))
         return self._format_duration(elapsed_seconds)
 
+    async def _run_delayed_reboot(self, requested_by: str, message: str) -> None:
+        """Wait for reboot delay, then terminate process for supervisor restart."""
+
+        try:
+            await asyncio.sleep(5)
+        except asyncio.CancelledError:
+            return
+        LOGGER.warning("server reboot requested by=%s message=%s", requested_by, message)
+        os.kill(os.getpid(), signal.SIGTERM)
+
+    def _schedule_reboot(self, requested_by: str, message: str) -> bool:
+        """Schedule one delayed reboot; return False when one is already pending."""
+
+        if self._pending_reboot_task is not None and not self._pending_reboot_task.done():
+            return False
+        self._pending_reboot_task = asyncio.create_task(self._run_delayed_reboot(requested_by, message))
+        return True
+
     async def _handle_chat_command(self, client: ClientConnection, message: str) -> bool:
         """Handle slash commands in chat input; return True when handled."""
 
@@ -1638,6 +1663,49 @@ class SignalingServer:
                 ),
             )
             return True
+        if command == "version":
+            await self._send(
+                client.websocket,
+                BroadcastChatMessagePacket(
+                    type="chat_message",
+                    message=f"Server version: {self.server_version}",
+                    system=True,
+                ),
+            )
+            return True
+        if command == "reboot":
+            if not self._client_has_permission(client, "server.allow_reboot"):
+                await self._send(
+                    client.websocket,
+                    BroadcastChatMessagePacket(
+                        type="chat_message",
+                        message="Not authorized to reboot server.",
+                        system=True,
+                    ),
+                )
+                return True
+            reboot_message = remainder if separator else ""
+            if not self._schedule_reboot(client.username or client.nickname, reboot_message):
+                await self._send(
+                    client.websocket,
+                    BroadcastChatMessagePacket(
+                        type="chat_message",
+                        message="Server reboot already scheduled.",
+                        system=True,
+                    ),
+                )
+                return True
+            announcement = "Server rebooting in 5 seconds."
+            if reboot_message:
+                announcement = f"{announcement} {reboot_message}"
+            await self._broadcast(
+                BroadcastChatMessagePacket(
+                    type="chat_message",
+                    message=announcement,
+                    system=True,
+                )
+            )
+            return True
         await self._send(
             client.websocket,
             BroadcastChatMessagePacket(

server/tests/test_server_message_handling.py

@@ -497,3 +497,89 @@ async def test_chat_command_requires_leading_slash(monkeypatch: pytest.MonkeyPat
     assert packet.system is False
     assert packet.action is False
     assert packet.message == " /up"
+
+
+@pytest.mark.asyncio
+async def test_chat_version_command_is_sender_only(monkeypatch: pytest.MonkeyPatch) -> None:
+    server = SignalingServer("127.0.0.1", 8765, None, None)
+    ws = _fake_ws()
+    client = ClientConnection(websocket=ws, id="u1", nickname="Tester")
+    server.clients[ws] = client
+    server.server_version = "2026.02.27 R293"
+
+    broadcast_payloads: list[object] = []
+    send_payloads: list[object] = []
+
+    async def fake_broadcast(packet: object, exclude: ServerConnection | None = None) -> None:
+        broadcast_payloads.append(packet)
+
+    async def fake_send(websocket: ServerConnection, packet: object) -> None:
+        send_payloads.append(packet)
+
+    monkeypatch.setattr(server, "_broadcast", fake_broadcast)
+    monkeypatch.setattr(server, "_send", fake_send)
+
+    await server._handle_message(client, json.dumps({"type": "chat_message", "message": "/version"}))
+
+    assert broadcast_payloads == []
+    assert len(send_payloads) == 1
+    packet = send_payloads[0]
+    assert getattr(packet, "type", "") == "chat_message"
+    assert packet.system is True
+    assert packet.message == "Server version: 2026.02.27 R293"
+
+
+@pytest.mark.asyncio
+async def test_chat_reboot_requires_permission(monkeypatch: pytest.MonkeyPatch) -> None:
+    server = SignalingServer("127.0.0.1", 8765, None, None)
+    ws = _fake_ws()
+    client = ClientConnection(websocket=ws, id="u1", nickname="Tester", authenticated=True, user_id="1", permissions={"chat.send"})
+    server.clients[ws] = client
+
+    send_payloads: list[object] = []
+
+    async def fake_send(websocket: ServerConnection, packet: object) -> None:
+        send_payloads.append(packet)
+
+    monkeypatch.setattr(server, "_send", fake_send)
+    monkeypatch.setattr(server, "_schedule_reboot", lambda _requested_by, _message: True)
+
+    await server._handle_message(client, json.dumps({"type": "chat_message", "message": "/reboot patching"}))
+
+    assert send_payloads
+    packet = send_payloads[-1]
+    assert getattr(packet, "type", "") == "chat_message"
+    assert packet.system is True
+    assert "not authorized" in packet.message.lower()
+
+
+@pytest.mark.asyncio
+async def test_chat_reboot_schedules_and_broadcasts_message(monkeypatch: pytest.MonkeyPatch) -> None:
+    server = SignalingServer("127.0.0.1", 8765, None, None)
+    ws = _fake_ws()
+    client = ClientConnection(
+        websocket=ws,
+        id="u1",
+        nickname="Tester",
+        authenticated=True,
+        user_id="1",
+        username="tester",
+        permissions={"chat.send", "server.allow_reboot"},
+    )
+    server.clients[ws] = client
+
+    broadcast_payloads: list[object] = []
+
+    async def fake_broadcast(packet: object, exclude: ServerConnection | None = None) -> None:
+        broadcast_payloads.append(packet)
+
+    monkeypatch.setattr(server, "_broadcast", fake_broadcast)
+    monkeypatch.setattr(server, "_schedule_reboot", lambda requested_by, message: requested_by == "tester" and message == "maintenance")
+
+    await server._handle_message(client, json.dumps({"type": "chat_message", "message": "/reboot maintenance"}))
+
+    assert len(broadcast_payloads) == 1
+    packet = broadcast_payloads[0]
+    assert getattr(packet, "type", "") == "chat_message"
+    assert packet.system is True
+    assert packet.message == "Server rebooting in 5 seconds. maintenance"