Add /start, /stop, /restart session control commands across all channels

- /start: creates a new session or reports one is already active
- /stop: archives the current session (timestamped JSONL) and frees the slot
- /restart: replaces the process via os.execv for a full system reboot
- Commands intercepted in Orchestrator before reaching the agent (no LLM call)
- Command logic extracted to orchestrator/commands.py with COMMANDS set for future extensibility
- SessionStore.archive() added to rename active session file with timestamp suffix
- All channels (Telegram, Discord, Slack, Twitch) detect /start /stop /restart and set _command metadata
- Telegram /new kept as alias for /stop; Twitch uses / prefix instead of !
- 31 tests added covering archive, all command flows, routing, and edge cases

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

Author: Jeomon

operator_use/gateway/channels/discord.py

@@ -290,8 +290,6 @@ async def _on_message(self, message: discord.Message) -> None:
         content = "\n".join(content_parts) if content_parts else "[empty message]"
         str_channel_id = str(channel_id)
 
-        self._start_typing(channel_id)
-
         metadata = {
             "message_id": message.id,
             "user_id": author_id,
@@ -300,6 +298,22 @@ async def _on_message(self, message: discord.Message) -> None:
             "channel_type": type(message.channel).__name__,
         }
 
+        # Session control commands — no typing indicator, no agent call
+        if not media_paths and content.strip().lower() in ("/start", "/stop", "/restart"):
+            command = content.strip()[1:].lower()
+            incoming = IncomingMessage(
+                channel=self.name,
+                chat_id=str_channel_id,
+                parts=[TextPart(content=content.strip())],
+                user_id=str(author_id),
+                account_id=self._cfg("account_id") or "",
+                metadata={**metadata, "_command": command},
+            )
+            await self.receive(incoming)
+            return
+
+        self._start_typing(channel_id)
+
         if media_paths:
             content, parts = await self._process_media_to_parts(
                 content, media_paths, [self._get_media_type(p) for p in media_paths]

operator_use/gateway/channels/slack.py

@@ -368,6 +368,20 @@ async def _handle_webhook_event(self, event: dict) -> None:
             "user_id": sender_id,
         }
 
+        # Session control commands
+        if text and text.strip().lower() in ("/start", "/stop", "/restart"):
+            command = text.strip()[1:].lower()
+            incoming = IncomingMessage(
+                channel=self.name,
+                chat_id=channel_id,
+                parts=[TextPart(content=text.strip())],
+                user_id=sender_id,
+                account_id=self._cfg("account_id") or "",
+                metadata={**metadata, "_command": command},
+            )
+            await self.receive(incoming)
+            return
+
         # Create incoming message
         incoming = IncomingMessage(
             channel=self.name,
@@ -447,6 +461,20 @@ async def _on_socket_request(self, client, req: SocketModeRequest):
                 "user_id": sender_id,
             }
 
+            # Session control commands
+            if text and text.strip().lower() in ("/start", "/stop", "/restart"):
+                command = text.strip()[1:].lower()
+                incoming = IncomingMessage(
+                    channel=self.name,
+                    chat_id=channel_id,
+                    parts=[TextPart(content=text.strip())],
+                    user_id=sender_id,
+                    account_id=self._cfg("account_id") or "",
+                    metadata={**metadata, "_command": command},
+                )
+                await self.receive(incoming)
+                return
+
             incoming = IncomingMessage(
                 channel=self.name,
                 chat_id=channel_id,

operator_use/gateway/channels/telegram.py

@@ -158,9 +158,9 @@ class TelegramChannel(BaseChannel):
     """Telegram channel using long-polling or webhook."""
 
     BOT_COMMANDS = [
-        BotCommand("start", "Start the bot"),
-        BotCommand("new", "Start a new conversation"),
-        BotCommand("stop", "Stop the current task"),
+        BotCommand("start", "Start a new session"),
+        BotCommand("stop", "Stop and save the current session"),
+        BotCommand("restart", "Reboot the entire system"),
         BotCommand("help", "Show available commands"),
     ]
 
@@ -280,7 +280,9 @@ async def _post_init(app):
         self._app = builder.build()
 
         self._app.add_handler(CommandHandler("start", self._on_start))
-        self._app.add_handler(CommandHandler("new", self._on_new))
+        self._app.add_handler(CommandHandler("stop", self._on_stop))
+        self._app.add_handler(CommandHandler("restart", self._on_restart))
+        self._app.add_handler(CommandHandler("new", self._on_stop))  # /new kept as stop alias
         self._app.add_handler(MessageHandler(
             (
                 filters.TEXT|
@@ -848,7 +850,7 @@ async def _typing_loop(self, chat_id: str) -> None:
             logger.debug("Typing indicator stopped for %s: %s", chat_id, e)
 
     async def _on_start(self, update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
-        """Handle /start — welcome in DM, register group for setup-channels."""
+        """Handle /start — register group for setup-channels in groups, start session in DMs."""
         if not update.message or not update.effective_chat:
             return
         chat = update.effective_chat
@@ -870,14 +872,35 @@ async def _on_start(self, update: Update, context: ContextTypes.DEFAULT_TYPE) ->
             except Exception as e:
                 logger.error("Failed to register group: %s", e)
         else:
-            user = update.effective_user
-            name = user.first_name if user else "there"
-            await update.message.reply_text(
-                f"Hi {name}! I'm your AI agent. Send me a message to get started."
+            thread_id = update.message.message_thread_id
+            chat_id = self._build_chat_id(str(chat.id), thread_id)
+            incoming = IncomingMessage(
+                channel=self.name,
+                chat_id=chat_id,
+                parts=[TextPart(content="/start")],
+                user_id=str(update.effective_user.id) if update.effective_user else "",
+                metadata={"_command": "start", "thread_id": thread_id},
             )
+            await self.receive(incoming)
+
+    async def _on_stop(self, update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
+        """Handle /stop — archive the current session."""
+        if not update.message or not update.effective_chat:
+            return
+        chat = update.effective_chat
+        thread_id = update.message.message_thread_id
+        chat_id = self._build_chat_id(str(chat.id), thread_id)
+        incoming = IncomingMessage(
+            channel=self.name,
+            chat_id=chat_id,
+            parts=[TextPart(content="/stop")],
+            user_id=str(update.effective_user.id) if update.effective_user else "",
+            metadata={"_command": "stop", "thread_id": thread_id},
+        )
+        await self.receive(incoming)
 
-    async def _on_new(self, update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
-        """Handle /new — clear session and start fresh."""
+    async def _on_restart(self, update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
+        """Handle /restart (and /new alias) — clear session history and start fresh."""
         if not update.message or not update.effective_chat:
             return
         chat = update.effective_chat
@@ -886,9 +909,9 @@ async def _on_new(self, update: Update, context: ContextTypes.DEFAULT_TYPE) -> N
         incoming = IncomingMessage(
             channel=self.name,
             chat_id=chat_id,
-            parts=[TextPart(content="/new")],
+            parts=[TextPart(content="/restart")],
             user_id=str(update.effective_user.id) if update.effective_user else "",
-            metadata={"_command": "new", "thread_id": thread_id},
+            metadata={"_command": "restart", "thread_id": thread_id},
         )
         await self.receive(incoming)
 

operator_use/gateway/channels/twitch.py

@@ -144,6 +144,21 @@ async def _on_message(self, message: twitchio.Message) -> None:
 
         chat_id = channel_name  # use channel name as chat_id (Twitch has one chat per channel)
 
+        # Session control commands always use / prefix (e.g. /start, /stop, /restart)
+        lower_content = content.strip().lower()
+        for cmd in ("start", "stop", "restart"):
+            if lower_content == f"/{cmd}":
+                incoming = IncomingMessage(
+                    channel=self.name,
+                    chat_id=chat_id,
+                    parts=[TextPart(content=content.strip())],
+                    user_id=author_name,
+                    account_id=self._cfg("account_id") or "",
+                    metadata={"username": author_name, "channel": channel_name, "_command": cmd},
+                )
+                await self.receive(incoming)
+                return
+
         incoming = IncomingMessage(
             channel=self.name,
             chat_id=chat_id,

operator_use/orchestrator/__init__.py

@@ -1,5 +1,6 @@
 """Orchestrator: pipeline layer between channels and agents."""
 
 from operator_use.orchestrator.service import Orchestrator
+from operator_use.orchestrator.commands import COMMANDS
 
-__all__ = ["Orchestrator"]
+__all__ = ["Orchestrator", "COMMANDS"]

operator_use/orchestrator/commands.py

@@ -0,0 +1,79 @@
+"""Session and system control commands (/start, /stop, /restart, ...)."""
+
+import asyncio
+import logging
+import os
+import sys
+from typing import TYPE_CHECKING
+
+from operator_use.bus.views import IncomingMessage, OutgoingMessage, TextPart
+
+if TYPE_CHECKING:
+    from operator_use.bus import Bus
+    from operator_use.agent.service import Agent
+
+logger = logging.getLogger(__name__)
+
+# All recognised command names. Channels use this to detect commands.
+COMMANDS = {"start", "stop", "restart"}
+
+
+async def handle_command(
+    message: IncomingMessage,
+    agent: "Agent",
+    bus: "Bus",
+) -> None:
+    """Dispatch a session/system control command and publish the response."""
+    command = (message.metadata or {}).get("_command")
+    session_id = f"{message.channel}:{message.chat_id}"
+
+    if command == "start":
+        text = await _cmd_start(session_id, agent)
+    elif command == "stop":
+        text = await _cmd_stop(session_id, agent)
+    elif command == "restart":
+        text = await _cmd_restart()
+    else:
+        logger.warning("handle_command called with unknown command: %r", command)
+        return
+
+    await bus.publish_outgoing(
+        OutgoingMessage(
+            chat_id=message.chat_id,
+            channel=message.channel,
+            account_id=message.account_id,
+            parts=[TextPart(content=text)],
+            metadata=message.metadata,
+            reply=True,
+        )
+    )
+
+    if command == "restart":
+        asyncio.create_task(_deferred_process_restart())
+
+
+# ---------------------------------------------------------------------------
+# Individual command implementations
+# ---------------------------------------------------------------------------
+
+async def _cmd_start(session_id: str, agent: "Agent") -> str:
+    existing = agent.sessions.load(session_id)
+    if existing and existing.messages:
+        return "Session is already active. Use /stop to end the session."
+    return "Session started! Send me a message to get started."
+
+
+async def _cmd_stop(session_id: str, agent: "Agent") -> str:
+    agent.sessions.archive(session_id)
+    return "Session stopped and saved. Use /start to begin a new session."
+
+
+async def _cmd_restart() -> str:
+    return "Restarting system..."
+
+
+async def _deferred_process_restart(delay: float = 1.5) -> None:
+    """Wait for the response to be delivered, then replace the process."""
+    await asyncio.sleep(delay)
+    logger.info("Restarting process: %s %s", sys.executable, sys.argv)
+    os.execv(sys.executable, [sys.executable] + sys.argv)

operator_use/orchestrator/service.py

@@ -230,6 +230,12 @@ async def _build_outgoing_message(
     # Message handling
     # ------------------------------------------------------------------
 
+    async def _handle_command(self, message: IncomingMessage) -> None:
+        """Handle session/system control commands without running the agent."""
+        from operator_use.orchestrator.commands import handle_command
+        agent = self._resolve_agent(message)
+        await handle_command(message, agent, self.bus)
+
     async def _handle_message(self, request_message: IncomingMessage) -> None:
         """Process one incoming message end-to-end."""
         session_id = f"{request_message.channel}:{request_message.chat_id}"
@@ -376,6 +382,11 @@ async def ainvoke(self) -> None:
                     await agent._handle_reaction(request_message)
                     continue
 
+                # Session control commands: handle without running the agent
+                if request_message.metadata and request_message.metadata.get("_command") in ("start", "stop", "restart"):
+                    await self._handle_command(request_message)
+                    continue
+
                 # Pending reply: a tool is waiting for the user's next message
                 if session_id in self._pending_replies:
                     future = self._pending_replies.pop(session_id)

operator_use/session/service.py

@@ -91,6 +91,21 @@ def delete(self, session_id: str) -> bool:
             return True
         return False
 
+    def archive(self, session_id: str) -> bool:
+        """Archive a session by renaming its file with a timestamp suffix. Returns True if existed.
+
+        The active session slot is freed so the next get_or_create starts fresh.
+        """
+        path = self._sessions_path(session_id)
+        if session_id in self._sessions:
+            del self._sessions[session_id]
+        if path.exists():
+            timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
+            archive_name = f"{self._session_id_to_filename(session_id)}_archived_{timestamp}.jsonl"
+            path.rename(self.sessions_dir / archive_name)
+            return True
+        return False
+
     def list_sessions(self) -> list[dict[str, Any]]:
         """Load sessions from the sessions directory. Returns list of dicts with id, created_at, updated_at, path."""
         result: list[dict[str, Any]] = []