feat(gateway): add /workspace command for multi-workspace sessions

gateway/run.py

@@ -9743,6 +9743,9 @@ async def _do_reset():
         if canonical == "status":
             return await self._handle_status_command(event)
 
+        if canonical == "workspace":
+            return await self._handle_workspace_command(event)
+
         if canonical == "agents":
             return await self._handle_agents_command(event)
 
@@ -10730,6 +10733,18 @@ async def _handle_message_with_agent(self, event, source, _quick_key: str, run_g
 
         session_entry = await self.async_session_store.get_or_create_session(source)
         session_key = session_entry.session_key
+        if session_entry.workspace_cwd:
+            # Task overrides are read dynamically by terminal + file tools, so
+            # this is prompt-cache safe even when the agent is reused. env_type
+            # is also an existing isolation signal, preventing concurrent
+            # gateway sessions from sharing mutable terminal cwd state.
+            from tools.terminal_tool import register_task_env_overrides
+            register_task_env_overrides(
+                session_entry.session_id, {
+                    "cwd": session_entry.workspace_cwd,
+                    "env_type": os.environ.get("TERMINAL_ENV", "local"),
+                }
+            )
         pinned_session_id = str(
             (getattr(event, "metadata", None) or {}).get("gateway_session_id") or ""
         ).strip()

gateway/session.py

@@ -718,6 +718,11 @@ class SessionEntry:
     # (see sanitize_model_override / SessionStore.set_model_override).
     model_override: Optional[Dict[str, str]] = None
 
+    # Gateway /workspace selection. The registry itself lives in config.yaml;
+    # only the selected name/path is session routing state.
+    workspace_name: Optional[str] = None
+    workspace_cwd: Optional[str] = None
+
     def to_dict(self) -> Dict[str, Any]:
         result = {
             "session_key": self.session_key,
@@ -753,6 +758,9 @@ def to_dict(self) -> Dict[str, Any]:
             # Defence-in-depth: strip credentials even if a caller stored an
             # unsanitized dict directly on the entry.
             result["model_override"] = sanitize_model_override(self.model_override)
+        if self.workspace_name and self.workspace_cwd:
+            result["workspace_name"] = self.workspace_name
+            result["workspace_cwd"] = self.workspace_cwd
         if self.origin:
             result["origin"] = self.origin.to_dict()
         return result
@@ -824,6 +832,8 @@ def from_dict(cls, data: Dict[str, Any]) -> "SessionEntry":
             auto_reset_reason=data.get("auto_reset_reason"),
             reset_had_activity=data.get("reset_had_activity", False),
             model_override=sanitize_model_override(data.get("model_override")),
+            workspace_name=data.get("workspace_name"),
+            workspace_cwd=data.get("workspace_cwd"),
         )
 
 
@@ -2071,6 +2081,21 @@ def get_model_override(self, session_key: str) -> Optional[Dict[str, str]]:
                 return None
             return dict(entry.model_override) if entry.model_override else None
 
+    def set_workspace(
+        self, session_key: str, name: Optional[str], cwd: Optional[str]
+    ) -> Optional[SessionEntry]:
+        """Persist the active workspace on an existing gateway session."""
+        with self._lock:
+            self._ensure_loaded_locked()
+            entry = self._entries.get(session_key)
+            if entry is None:
+                return None
+            entry.workspace_name = name
+            entry.workspace_cwd = cwd
+            entry.updated_at = _now()
+            self._save()
+            return entry
+
     def suspend_session(self, session_key: str) -> bool:
         """Mark a session as suspended so it auto-resets on next access.
 
@@ -2256,6 +2281,8 @@ def reset_session(self, session_key: str, display_name: Optional[str] = None) ->
                 platform=old_entry.platform,
                 chat_type=old_entry.chat_type,
                 is_fresh_reset=True,
+                workspace_name=old_entry.workspace_name,
+                workspace_cwd=old_entry.workspace_cwd,
             )
 
             self._entries[session_key] = new_entry

gateway/slash_commands.py

@@ -102,6 +102,112 @@ def _typed_command_prefix_for(self, platform) -> str:
         adapter = self.adapters.get(platform) if getattr(self, "adapters", None) else None
         return getattr(adapter, "typed_command_prefix", "/") if adapter is not None else "/"
 
+    async def _handle_workspace_command(self, event: MessageEvent) -> str:
+        """List and manage registered gateway working directories."""
+        try:
+            parts = shlex.split((event.text or "").strip())
+        except ValueError as exc:
+            return f"Invalid /workspace arguments: {exc}"
+        args = parts[1:]
+        action = args[0].lower() if args else "list"
+        if action not in {"list", "new", "switch", "remove"}:
+            args = ["switch", *args]
+            action = "switch"
+
+        source = event.source
+        entry = await self.async_session_store.get_or_create_session(source)
+        session_key = entry.session_key
+
+        def _operate() -> tuple[str, Optional[tuple[str, str]]]:
+            from hermes_cli.config import load_config, save_config
+            from hermes_constants import get_hermes_home
+
+            cfg = load_config()
+            gateway_cfg = cfg.setdefault("gateway", {})
+            raw_registry = gateway_cfg.setdefault("workspaces", {})
+            registry = raw_registry if isinstance(raw_registry, dict) else {}
+            if raw_registry is not registry:
+                gateway_cfg["workspaces"] = registry
+
+            if action == "list":
+                current_name = entry.workspace_name or "default"
+                current_cwd = entry.workspace_cwd or os.environ.get(
+                    "TERMINAL_CWD", str(Path.home())
+                )
+                lines = [f"Current: {current_name} ({current_cwd})", "Workspaces:"]
+                if not registry:
+                    lines.append("  (none registered)")
+                for name, raw_path in sorted(registry.items()):
+                    marker = "*" if name == entry.workspace_name else " "
+                    lines.append(f"{marker} {name}: {raw_path}")
+                if entry.workspace_name and entry.workspace_name not in registry:
+                    lines.append(f"* {entry.workspace_name}: {entry.workspace_cwd} (unregistered)")
+                if entry.workspace_name:
+                    lines.append("* = current")
+                return "\n".join(lines), None
+
+            if action == "new":
+                if len(args) not in {2, 3}:
+                    return "Usage: /workspace new <name> [path]", None
+                name = args[1]
+                if not re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9._-]{0,63}", name) or name in {".", ".."}:
+                    return "Workspace names may contain letters, numbers, '.', '_', and '-' (max 64).", None
+                if name in registry:
+                    return f"Workspace '{name}' is already registered at {registry[name]}", None
+                path = Path(args[2]).expanduser() if len(args) == 3 else get_hermes_home() / "workspaces" / name
+                try:
+                    path = path.resolve(strict=False)
+                    path.mkdir(parents=True, exist_ok=True)
+                except OSError as exc:
+                    return f"Could not create workspace '{name}': {exc}", None
+                registry[name] = str(path)
+                save_config(cfg)
+                return f"Registered workspace '{name}' at {path}", None
+
+            if len(args) != 2:
+                return f"Usage: /workspace {action} <name>", None
+            name = args[1]
+            if name not in registry:
+                return f"Unknown workspace '{name}'. Use /workspace list to see registered workspaces.", None
+
+            path = Path(str(registry[name])).expanduser().resolve(strict=False)
+            if action == "remove":
+                del registry[name]
+                save_config(cfg)
+                if entry.workspace_name == name:
+                    fallback = os.environ.get("TERMINAL_CWD", str(Path.home()))
+                    return f"Unregistered workspace '{name}'. Files were not deleted.", ("", fallback)
+                return f"Unregistered workspace '{name}'. Files were not deleted.", None
+
+            if not path.is_dir():
+                return f"Workspace '{name}' path does not exist or is not a directory: {path}", None
+            return f"Switched workspace to '{name}' ({path})", (name, str(path))
+
+        if getattr(getattr(self, "config", None), "multiplex_profiles", False):
+            from gateway.run import _profile_runtime_scope
+            with _profile_runtime_scope(self._resolve_profile_home_for_source(source)):
+                message, selection = await asyncio.to_thread(_operate)
+        else:
+            message, selection = await asyncio.to_thread(_operate)
+
+        if selection is not None:
+            name, cwd = selection
+            updated = await self.async_session_store.set_workspace(
+                session_key, name or None, cwd if name else None,
+            )
+            if updated is not None:
+                from tools.terminal_tool import register_task_env_overrides
+                register_task_env_overrides(updated.session_id, {
+                    "cwd": cwd,
+                    # Gateway workspaces may run concurrently. Mark this task
+                    # as isolated so two sessions never share a mutable env.cwd.
+                    "env_type": os.environ.get("TERMINAL_ENV", "local"),
+                })
+                db = getattr(self.session_store, "_db", None)
+                if db is not None:
+                    await asyncio.to_thread(db.update_session_cwd, updated.session_id, cwd)
+        return message
+
     async def _handle_reset_command(self, event: MessageEvent) -> Union[str, EphemeralReply]:
         """Handle /new or /reset command."""
         source = event.source

hermes_cli/commands.py

@@ -127,6 +127,9 @@ class CommandDef:
 
     # Configuration
     CommandDef("sessions", "Browse and resume previous sessions", "Session"),
+    CommandDef("workspace", "List, create, switch, or remove gateway workspaces", "Session",
+               gateway_only=True, args_hint="[list|new <name> [path]|switch <name>|remove <name>]",
+               subcommands=("list", "new", "switch", "remove")),
 
     # Configuration
     CommandDef("config", "Show current configuration", "Configuration",
@@ -1163,7 +1166,9 @@ def discord_skill_commands_by_category(
 #   - moa: high-cost slash mode, available through /hermes moa to avoid
 #     displacing existing native Slack slash commands at the 50-command cap.
 #   - debug: the log/report upload surface; reached via /hermes debug on Slack.
-_SLACK_VIA_HERMES_ONLY = frozenset({"credits", "billing", "moa", "debug"})
+_SLACK_VIA_HERMES_ONLY = frozenset(
+    {"credits", "billing", "moa", "debug", "workspace"}
+)
 
 
 def _sanitize_slack_name(raw: str) -> str:

tests/gateway/test_workspace_command.py

@@ -0,0 +1,120 @@
+"""Behavior contracts for gateway /workspace management."""
+
+from types import SimpleNamespace
+
+import pytest
+
+from gateway.config import GatewayConfig, Platform, PlatformConfig
+from gateway.platforms.base import MessageEvent
+from gateway.session import SessionSource, SessionStore
+from hermes_constants import reset_hermes_home_override, set_hermes_home_override
+
+
+def _source() -> SessionSource:
+    return SessionSource(
+        platform=Platform.TELEGRAM,
+        user_id="workspace-user",
+        chat_id="workspace-chat",
+        chat_type="dm",
+    )
+
+
+def _event(text: str) -> MessageEvent:
+    return MessageEvent(text=text, source=_source(), message_id="m1")
+
+
+@pytest.fixture
+def workspace_runner(tmp_path):
+    from gateway.run import GatewayRunner
+
+    home = tmp_path / ".hermes"
+    home.mkdir()
+    token = set_hermes_home_override(home)
+    config = GatewayConfig(
+        platforms={Platform.TELEGRAM: PlatformConfig(enabled=True, token="test")}
+    )
+    runner = object.__new__(GatewayRunner)
+    runner.config = config
+    runner.adapters = {}
+    runner.session_store = SessionStore(home / "gateway-sessions", config)
+    runner._async_session_store = None
+    runner._session_db = SimpleNamespace(_db=runner.session_store._db)
+    runner._agent_cache = {"unchanged": object()}
+
+    try:
+        yield runner, home
+    finally:
+        from tools.terminal_tool import clear_task_env_overrides
+
+        for entry in runner.session_store._entries.values():
+            clear_task_env_overrides(entry.session_id)
+        reset_hermes_home_override(token)
+
+
+@pytest.mark.asyncio
+async def test_workspace_switch_updates_tools_without_rebuilding_agent(workspace_runner):
+    runner, home = workspace_runner
+    target = home / "projects" / "alpha"
+    cached_agent = runner._agent_cache["unchanged"]
+
+    created = await runner._handle_workspace_command(
+        _event(f'/workspace new alpha "{target}"')
+    )
+    switched = await runner._handle_workspace_command(_event("/workspace alpha"))
+
+    entry = await runner.async_session_store.get_or_create_session(_source())
+    from tools.terminal_tool import resolve_task_overrides
+
+    assert "Registered workspace 'alpha'" in created
+    assert "Switched workspace to 'alpha'" in switched
+    assert target.is_dir()
+    assert entry.workspace_name == "alpha"
+    assert entry.workspace_cwd == str(target.resolve())
+    overrides = resolve_task_overrides(entry.session_id)
+    assert overrides["cwd"] == str(target.resolve())
+    assert overrides["env_type"] == "local"
+    assert runner._agent_cache["unchanged"] is cached_agent
+
+
+@pytest.mark.asyncio
+async def test_workspace_registry_persists_and_remove_never_deletes_files(workspace_runner):
+    runner, home = workspace_runner
+
+    await runner._handle_workspace_command(_event("/workspace new beta"))
+    await runner._handle_workspace_command(_event("/workspace switch beta"))
+    listed = await runner._handle_workspace_command(_event("/workspace list"))
+    target = home / "workspaces" / "beta"
+    sentinel = target / "keep.txt"
+    sentinel.write_text("keep", encoding="utf-8")
+    removed = await runner._handle_workspace_command(_event("/workspace remove beta"))
+
+    from hermes_cli.config import load_config
+
+    entry = await runner.async_session_store.get_or_create_session(_source())
+    assert f"Current: beta ({target.resolve()})" in listed
+    assert f"* beta: {target.resolve()}" in listed
+    assert "Files were not deleted" in removed
+    assert sentinel.read_text(encoding="utf-8") == "keep"
+    assert "beta" not in load_config().get("gateway", {}).get("workspaces", {})
+    assert entry.workspace_name is None
+    assert entry.workspace_cwd is None
+
+
+def test_workspace_selection_survives_session_routing_round_trip():
+    from datetime import datetime
+    from gateway.session import SessionEntry
+
+    entry = SessionEntry(
+        session_key="agent:main:telegram:dm:u:c",
+        session_id="session-1",
+        created_at=datetime.now(),
+        updated_at=datetime.now(),
+        workspace_name="alpha",
+        workspace_cwd="/tmp/alpha",
+    )
+
+    restored = SessionEntry.from_dict(entry.to_dict())
+    assert (restored.workspace_name, restored.workspace_cwd) == (
+        entry.workspace_name,
+        entry.workspace_cwd,
+    )

website/docs/reference/slash-commands.md

@@ -219,6 +219,7 @@ The messaging gateway supports the following built-in commands inside Telegram,
 | `/topic [off\|help\|session-id]` | **Telegram DM only.** Manage user-managed multi-session topic mode. `/topic` enables it or shows status; `/topic off` disables it and clears bindings; `/topic help` shows usage; `/topic <session-id>` inside a topic restores a previous session. See [Multi-session DM mode](/user-guide/messaging/telegram#multi-session-dm-mode-topic). |
 | `/title [name]` | Set or show the session title. |
 | `/resume [name]` | Resume a previously named session. |
+| `/workspace [list\|new <name> [path]\|switch <name>\|remove <name>]` | Register and switch per-session working directories without resetting the conversation. `/workspace <name>` is shorthand for `switch`; omitted paths default to `~/.hermes/workspaces/<name>`. Removing a workspace only unregisters it and never deletes files. |
 | `/usage` | Show token usage, estimated cost breakdown (input/output), context window state, session duration, and — when available from the active provider — an **Account limits** section with remaining quota / credits pulled live from the provider's API. |
 | `/credits` | Show your Nous credit balance and a top-up link that opens the portal billing page in a browser. |
 | `/insights [days]` | Show usage analytics. |
@@ -253,7 +254,7 @@ The messaging gateway supports the following built-in commands inside Telegram,
 - `/skin`, `/snapshot`, `/reload`, `/tools`, `/toolsets`, `/browser`, `/config`, `/cron`, `/platforms`, `/paste`, `/image`, `/statusbar`, `/plugins`, `/busy`, `/indicator`, `/redraw`, `/clear`, `/history`, `/save`, `/copy`, `/handoff`, `/billing`, and `/quit` are **CLI-only** commands.
 - `/skills` is **CLI-only for search/browse/install**; its write-approval review subcommands (`pending`, `approve`, `reject`, `diff`, `approval`) also work on messaging platforms when `skills.write_approval` is on. `/memory` works on **both** surfaces.
 - `/verbose` is **CLI-only by default**, but can be enabled for messaging platforms by setting `display.tool_progress_command: true` in `config.yaml`. When enabled, it cycles the `display.tool_progress` mode and saves to config.
-- `/sethome`, `/update`, `/restart`, `/approve`, `/deny`, `/topic`, `/platform`, and `/commands` are **messaging-only** commands.
+- `/sethome`, `/update`, `/restart`, `/approve`, `/deny`, `/topic`, `/platform`, `/workspace`, and `/commands` are **messaging-only** commands.
 - `/status`, `/version`, `/background`, `/queue`, `/steer`, `/voice`, `/reload-mcp`, `/reload-skills`, `/rollback`, `/debug`, `/fast`, `/footer`, `/curator`, `/kanban`, `/credits`, `/suggestions`, `/blueprint`, `/learn`, `/sessions`, and `/yolo` work in **both** the CLI and the messaging gateway.
 - `/voice join`, `/voice channel`, and `/voice leave` are only meaningful on Discord.
 - In the TUI, `/sessions` shows live sessions in the current TUI process. Use `/resume [name]` or `hermes --tui --resume <id-or-title>` for saved or closed transcripts.