feat: add /resume command to gateway for switching to named sessions

Messaging users can now switch back to previously-named sessions:
- /resume My Project  — resolves the title (with auto-lineage) and
  restores that session's conversation history
- /resume (no args)   — lists recent titled sessions to choose from

Adds SessionStore.switch_session() which ends the current session and
points the session entry at the target session ID so the old transcript
is loaded on the next message. Running agents are cleared on switch.

Completes the session naming feature from PR #720 for gateway users.

8 new tests covering: name resolution, lineage auto-latest, already-on-
session check, nonexistent names, agent cleanup, no-DB fallback, and
listing titled sessions.

gateway/run.py

@@ -767,6 +767,9 @@ class GatewayRunner:
 
         if command == "title":
             return await self._handle_title_command(event)
+
+        if command == "resume":
+            return await self._handle_resume_command(event)
         
         # Skill slash commands: /skill-name loads the skill and sends to agent
         if command:
@@ -1306,6 +1309,7 @@ class GatewayRunner:
             "`/sethome` — Set this chat as the home channel",
             "`/compress` — Compress conversation context",
             "`/title [name]` — Set or show the session title",
+            "`/resume [name]` — Resume a previously-named session",
             "`/usage` — Show token usage for this session",
             "`/insights [days]` — Show usage insights and analytics",
             "`/reload-mcp` — Reload MCP servers from config",
@@ -1730,6 +1734,79 @@ class GatewayRunner:
             else:
                 return "No title set. Usage: `/title My Session Name`"
 
+    async def _handle_resume_command(self, event: MessageEvent) -> str:
+        """Handle /resume command — switch to a previously-named session."""
+        if not self._session_db:
+            return "Session database not available."
+
+        source = event.source
+        session_key = build_session_key(source)
+        name = event.get_command_args().strip()
+
+        if not name:
+            # List recent titled sessions for this user/platform
+            try:
+                user_source = source.platform.value if source.platform else None
+                sessions = self._session_db.list_sessions_rich(
+                    source=user_source, limit=10
+                )
+                titled = [s for s in sessions if s.get("title")]
+                if not titled:
+                    return (
+                        "No named sessions found.\n"
+                        "Use `/title My Session` to name your current session, "
+                        "then `/resume My Session` to return to it later."
+                    )
+                lines = ["📋 **Named Sessions**\n"]
+                for s in titled[:10]:
+                    title = s["title"]
+                    preview = s.get("preview", "")[:40]
+                    preview_part = f" — _{preview}_" if preview else ""
+                    lines.append(f"• **{title}**{preview_part}")
+                lines.append("\nUsage: `/resume <session name>`")
+                return "\n".join(lines)
+            except Exception as e:
+                logger.debug("Failed to list titled sessions: %s", e)
+                return f"Could not list sessions: {e}"
+
+        # Resolve the name to a session ID
+        target_id = self._session_db.resolve_session_by_title(name)
+        if not target_id:
+            return (
+                f"No session found matching '**{name}**'.\n"
+                "Use `/resume` with no arguments to see available sessions."
+            )
+
+        # Check if already on that session
+        current_entry = self.session_store.get_or_create_session(source)
+        if current_entry.session_id == target_id:
+            return f"📌 Already on session **{name}**."
+
+        # Flush memories for current session before switching
+        try:
+            asyncio.create_task(self._async_flush_memories(current_entry.session_id))
+        except Exception as e:
+            logger.debug("Memory flush on resume failed: %s", e)
+
+        # Clear any running agent for this session key
+        if session_key in self._running_agents:
+            del self._running_agents[session_key]
+
+        # Switch the session entry to point at the old session
+        new_entry = self.session_store.switch_session(session_key, target_id)
+        if not new_entry:
+            return "Failed to switch session."
+
+        # Get the title for confirmation
+        title = self._session_db.get_session_title(target_id) or name
+
+        # Count messages for context
+        history = self.session_store.load_transcript(target_id)
+        msg_count = len([m for m in history if m.get("role") == "user"]) if history else 0
+        msg_part = f" ({msg_count} message{'s' if msg_count != 1 else ''})" if msg_count else ""
+
+        return f"↻ Resumed session **{title}**{msg_part}. Conversation restored."
+
     async def _handle_usage_command(self, event: MessageEvent) -> str:
         """Handle /usage command -- show token usage for the session's last agent run."""
         source = event.source

gateway/session.py

@@ -593,7 +593,49 @@ class SessionStore:
                 logger.debug("Session DB operation failed: %s", e)
         
         return new_entry
-    
+
+    def switch_session(self, session_key: str, target_session_id: str) -> Optional[SessionEntry]:
+        """Switch a session key to point at an existing session ID.
+
+        Used by ``/resume`` to restore a previously-named session.
+        Ends the current session in SQLite (like reset), but instead of
+        generating a fresh session ID, re-uses ``target_session_id`` so the
+        old transcript is loaded on the next message.
+        """
+        self._ensure_loaded()
+
+        if session_key not in self._entries:
+            return None
+
+        old_entry = self._entries[session_key]
+
+        # Don't switch if already on that session
+        if old_entry.session_id == target_session_id:
+            return old_entry
+
+        # End the current session in SQLite
+        if self._db:
+            try:
+                self._db.end_session(old_entry.session_id, "session_switch")
+            except Exception as e:
+                logger.debug("Session DB end_session failed: %s", e)
+
+        now = datetime.now()
+        new_entry = SessionEntry(
+            session_key=session_key,
+            session_id=target_session_id,
+            created_at=now,
+            updated_at=now,
+            origin=old_entry.origin,
+            display_name=old_entry.display_name,
+            platform=old_entry.platform,
+            chat_type=old_entry.chat_type,
+        )
+
+        self._entries[session_key] = new_entry
+        self._save()
+        return new_entry
+
     def list_sessions(self, active_minutes: Optional[int] = None) -> List[SessionEntry]:
         """List all sessions, optionally filtered by activity."""
         self._ensure_loaded()

tests/gateway/test_resume_command.py

@@ -0,0 +1,200 @@
+"""Tests for /resume gateway slash command.
+
+Tests the _handle_resume_command handler (switch to a previously-named session)
+across gateway messenger platforms.
+"""
+
+from unittest.mock import MagicMock, AsyncMock
+
+import pytest
+
+from gateway.config import Platform
+from gateway.platforms.base import MessageEvent
+from gateway.session import SessionSource, build_session_key
+
+
+def _make_event(text="/resume", platform=Platform.TELEGRAM,
+                user_id="12345", chat_id="67890"):
+    """Build a MessageEvent for testing."""
+    source = SessionSource(
+        platform=platform,
+        user_id=user_id,
+        chat_id=chat_id,
+        user_name="testuser",
+    )
+    return MessageEvent(text=text, source=source)
+
+
+def _session_key_for_event(event):
+    """Get the session key that build_session_key produces for an event."""
+    return build_session_key(event.source)
+
+
+def _make_runner(session_db=None, current_session_id="current_session_001",
+                 event=None):
+    """Create a bare GatewayRunner with a mock session_store and optional session_db."""
+    from gateway.run import GatewayRunner
+    runner = object.__new__(GatewayRunner)
+    runner.adapters = {}
+    runner._session_db = session_db
+    runner._running_agents = {}
+
+    # Compute the real session key if an event is provided
+    session_key = build_session_key(event.source) if event else "agent:main:telegram:dm"
+
+    # Mock session_store that returns a session entry with a known session_id
+    mock_session_entry = MagicMock()
+    mock_session_entry.session_id = current_session_id
+    mock_session_entry.session_key = session_key
+    mock_store = MagicMock()
+    mock_store.get_or_create_session.return_value = mock_session_entry
+    mock_store.load_transcript.return_value = []
+    mock_store.switch_session.return_value = mock_session_entry
+    runner.session_store = mock_store
+
+    # Stub out memory flushing
+    runner._async_flush_memories = AsyncMock()
+
+    return runner
+
+
+# ---------------------------------------------------------------------------
+# _handle_resume_command
+# ---------------------------------------------------------------------------
+
+
+class TestHandleResumeCommand:
+    """Tests for GatewayRunner._handle_resume_command."""
+
+    @pytest.mark.asyncio
+    async def test_no_session_db(self):
+        """Returns error when session database is unavailable."""
+        runner = _make_runner(session_db=None)
+        event = _make_event(text="/resume My Project")
+        result = await runner._handle_resume_command(event)
+        assert "not available" in result.lower()
+
+    @pytest.mark.asyncio
+    async def test_list_named_sessions_when_no_arg(self, tmp_path):
+        """With no argument, lists recently titled sessions."""
+        from hermes_state import SessionDB
+        db = SessionDB(db_path=tmp_path / "state.db")
+        db.create_session("sess_001", "telegram")
+        db.create_session("sess_002", "telegram")
+        db.set_session_title("sess_001", "Research")
+        db.set_session_title("sess_002", "Coding")
+
+        event = _make_event(text="/resume")
+        runner = _make_runner(session_db=db, event=event)
+        result = await runner._handle_resume_command(event)
+        assert "Research" in result
+        assert "Coding" in result
+        assert "Named Sessions" in result
+        db.close()
+
+    @pytest.mark.asyncio
+    async def test_list_shows_usage_when_no_titled(self, tmp_path):
+        """With no arg and no titled sessions, shows instructions."""
+        from hermes_state import SessionDB
+        db = SessionDB(db_path=tmp_path / "state.db")
+        db.create_session("sess_001", "telegram")  # No title
+
+        event = _make_event(text="/resume")
+        runner = _make_runner(session_db=db, event=event)
+        result = await runner._handle_resume_command(event)
+        assert "No named sessions" in result
+        assert "/title" in result
+        db.close()
+
+    @pytest.mark.asyncio
+    async def test_resume_by_name(self, tmp_path):
+        """Resolves a title and switches to that session."""
+        from hermes_state import SessionDB
+        db = SessionDB(db_path=tmp_path / "state.db")
+        db.create_session("old_session_abc", "telegram")
+        db.set_session_title("old_session_abc", "My Project")
+        db.create_session("current_session_001", "telegram")
+
+        event = _make_event(text="/resume My Project")
+        runner = _make_runner(session_db=db, current_session_id="current_session_001",
+                              event=event)
+        result = await runner._handle_resume_command(event)
+
+        assert "Resumed" in result
+        assert "My Project" in result
+        # Verify switch_session was called with the old session ID
+        runner.session_store.switch_session.assert_called_once()
+        call_args = runner.session_store.switch_session.call_args
+        assert call_args[0][1] == "old_session_abc"
+        db.close()
+
+    @pytest.mark.asyncio
+    async def test_resume_nonexistent_name(self, tmp_path):
+        """Returns error for unknown session name."""
+        from hermes_state import SessionDB
+        db = SessionDB(db_path=tmp_path / "state.db")
+        db.create_session("current_session_001", "telegram")
+
+        event = _make_event(text="/resume Nonexistent Session")
+        runner = _make_runner(session_db=db, event=event)
+        result = await runner._handle_resume_command(event)
+        assert "No session found" in result
+        db.close()
+
+    @pytest.mark.asyncio
+    async def test_resume_already_on_session(self, tmp_path):
+        """Returns friendly message when already on the requested session."""
+        from hermes_state import SessionDB
+        db = SessionDB(db_path=tmp_path / "state.db")
+        db.create_session("current_session_001", "telegram")
+        db.set_session_title("current_session_001", "Active Project")
+
+        event = _make_event(text="/resume Active Project")
+        runner = _make_runner(session_db=db, current_session_id="current_session_001",
+                              event=event)
+        result = await runner._handle_resume_command(event)
+        assert "Already on session" in result
+        db.close()
+
+    @pytest.mark.asyncio
+    async def test_resume_auto_lineage(self, tmp_path):
+        """Asking for 'My Project' when 'My Project #2' exists gets the latest."""
+        from hermes_state import SessionDB
+        db = SessionDB(db_path=tmp_path / "state.db")
+        db.create_session("sess_v1", "telegram")
+        db.set_session_title("sess_v1", "My Project")
+        db.create_session("sess_v2", "telegram")
+        db.set_session_title("sess_v2", "My Project #2")
+        db.create_session("current_session_001", "telegram")
+
+        event = _make_event(text="/resume My Project")
+        runner = _make_runner(session_db=db, current_session_id="current_session_001",
+                              event=event)
+        result = await runner._handle_resume_command(event)
+
+        assert "Resumed" in result
+        # Should resolve to #2 (latest in lineage)
+        call_args = runner.session_store.switch_session.call_args
+        assert call_args[0][1] == "sess_v2"
+        db.close()
+
+    @pytest.mark.asyncio
+    async def test_resume_clears_running_agent(self, tmp_path):
+        """Switching sessions clears any cached running agent."""
+        from hermes_state import SessionDB
+        db = SessionDB(db_path=tmp_path / "state.db")
+        db.create_session("old_session", "telegram")
+        db.set_session_title("old_session", "Old Work")
+        db.create_session("current_session_001", "telegram")
+
+        event = _make_event(text="/resume Old Work")
+        runner = _make_runner(session_db=db, current_session_id="current_session_001",
+                              event=event)
+        # Simulate a running agent using the real session key
+        real_key = _session_key_for_event(event)
+        runner._running_agents[real_key] = MagicMock()
+
+        await runner._handle_resume_command(event)
+
+        assert real_key not in runner._running_agents
+        db.close()