Add dashboard Hermes console websocket

hermes_cli/dashboard_auth/routes.py

@@ -608,7 +608,8 @@ async def api_auth_ws_ticket(request: Request):
 
     Browsers cannot set ``Authorization`` on a WebSocket upgrade, so in
     gated mode the SPA POSTs this endpoint to get a ``?ticket=`` value to
-    append to ``/api/pty``, ``/api/ws``, ``/api/pub``, or ``/api/events``.
+    append to ``/api/pty``, ``/api/console``, ``/api/ws``, ``/api/pub``, or
+    ``/api/events``.
 
     The ticket has a 30-second TTL and is single-use. Calling this endpoint
     multiple times in quick succession (e.g. one ticket per WS) is the

hermes_cli/web_server.py

@@ -12628,6 +12628,548 @@ def _ws_close_reason(text: str) -> str:
     return encoded[:120].decode("utf-8", "ignore") + "..."
 
 
+# ---------------------------------------------------------------------------
+# /api/console — safe Hermes Console command WebSocket.
+#
+# Unlike /api/pty, this endpoint never spawns a PTY, shell, or full Hermes CLI
+# subprocess. It runs the curated console engine in-process and exchanges
+# structured JSON frames with the dashboard xterm overlay.
+# ---------------------------------------------------------------------------
+
+_CONSOLE_PROMPT = "hermes> "
+_CONSOLE_COMMAND_TIMEOUT_SECONDS = 60.0
+_CONSOLE_OUTPUT_LIMIT = 50000
+
+
+def _dashboard_console_context() -> str:
+    """Choose local vs hosted command policy for the dashboard console."""
+    return "hosted" if _default_hermes_root_is_opt_data() else "local"
+
+
+def _console_profile_from_ws(ws: WebSocket) -> Optional[str]:
+    profile = (ws.query_params.get("profile") or "").strip()
+    return profile or None
+
+
+def _execute_console_line(
+    engine: Any,
+    line: str,
+    *,
+    confirmed: bool,
+    profile: Optional[str],
+) -> Any:
+    # _profile_scope swaps process-global skill module paths; keep it inside
+    # the worker thread and never hold it across awaits.
+    with _profile_scope(profile):
+        return engine.execute(line, confirmed=confirmed)
+
+
+async def _console_send(
+    ws: WebSocket,
+    send_lock: asyncio.Lock,
+    payload: Dict[str, Any],
+) -> None:
+    async with send_lock:
+        await ws.send_json(payload)
+
+
+async def _console_send_result(
+    ws: WebSocket,
+    send_lock: asyncio.Lock,
+    result: Any,
+    *,
+    command_id: int,
+) -> None:
+    command = result.command or ""
+    status = result.status
+    if status == "ok":
+        if result.output:
+            await _console_send(
+                ws,
+                send_lock,
+                {
+                    "type": "output",
+                    "id": command_id,
+                    "stream": "stdout",
+                    "data": result.output,
+                    "command": command,
+                },
+            )
+        await _console_send(
+            ws,
+            send_lock,
+            {
+                "type": "complete",
+                "id": command_id,
+                "status": "ok",
+                "command": command,
+                "prompt": _CONSOLE_PROMPT,
+            },
+        )
+        return
+
+    if status == "error":
+        await _console_send(
+            ws,
+            send_lock,
+            {
+                "type": "error",
+                "id": command_id,
+                "message": result.output or "Command failed.",
+                "command": command,
+            },
+        )
+        await _console_send(
+            ws,
+            send_lock,
+            {
+                "type": "complete",
+                "id": command_id,
+                "status": "error",
+                "command": command,
+                "prompt": _CONSOLE_PROMPT,
+            },
+        )
+        return
+
+    if status == "confirm_required":
+        await _console_send(
+            ws,
+            send_lock,
+            {
+                "type": "confirm_required",
+                "id": command_id,
+                "command": command,
+                "message": result.confirmation_message or f"Run `{command}`?",
+                "prompt": _CONSOLE_PROMPT,
+            },
+        )
+        await _console_send(
+            ws,
+            send_lock,
+            {
+                "type": "complete",
+                "id": command_id,
+                "status": "confirm_required",
+                "command": command,
+                "prompt": _CONSOLE_PROMPT,
+            },
+        )
+        return
+
+    if status == "clear":
+        await _console_send(ws, send_lock, {"type": "clear", "id": command_id})
+        await _console_send(
+            ws,
+            send_lock,
+            {
+                "type": "complete",
+                "id": command_id,
+                "status": "clear",
+                "command": command,
+                "prompt": _CONSOLE_PROMPT,
+            },
+        )
+        return
+
+    if status == "exit":
+        await _console_send(
+            ws,
+            send_lock,
+            {
+                "type": "complete",
+                "id": command_id,
+                "status": "exit",
+                "command": command,
+                "prompt": "",
+            },
+        )
+        return
+
+    await _console_send(
+        ws,
+        send_lock,
+        {
+            "type": "error",
+            "id": command_id,
+            "message": f"Unknown console result status: {status}",
+            "command": command,
+        },
+    )
+
+
+def _console_json_payload(msg: Any) -> tuple[Optional[dict[str, Any]], Optional[str]]:
+    raw: str | bytes | None = msg.get("text")
+    if raw is None:
+        raw = msg.get("bytes")
+    if raw is None:
+        return None, None
+    if isinstance(raw, bytes):
+        try:
+            raw = raw.decode("utf-8")
+        except UnicodeDecodeError:
+            return None, "Console frames must be UTF-8 JSON."
+    try:
+        payload = json.loads(raw)
+    except json.JSONDecodeError:
+        return None, "Console frames must be JSON objects."
+    if not isinstance(payload, dict):
+        return None, "Console frames must be JSON objects."
+    return payload, None
+
+
+@app.websocket("/api/console")
+async def console_ws(ws: WebSocket) -> None:
+    peer = ws.client.host if ws.client else "?"
+
+    if not _DASHBOARD_EMBEDDED_CHAT_ENABLED:
+        _log.info("console refused: embedded chat disabled peer=%s", peer)
+        await ws.close(code=4404, reason="embedded chat disabled")
+        return
+
+    auth_reason, cred = _ws_auth_reason(ws)
+    mode = _ws_auth_mode()
+    if auth_reason is not None:
+        _log.warning(
+            "console auth rejected reason=%s mode=%s cred=%s peer=%s",
+            auth_reason, mode, cred, peer,
+        )
+        await ws.close(code=4401, reason=_ws_close_reason(f"auth: {auth_reason}"))
+        return
+
+    host_origin_reason = _ws_host_origin_reason(ws)
+    if host_origin_reason is not None:
+        _log.warning("console refused: %s peer=%s", host_origin_reason, peer)
+        await ws.close(code=4403, reason=_ws_close_reason(host_origin_reason))
+        return
+
+    client_reason = _ws_client_reason(ws)
+    if client_reason is not None:
+        _log.warning("console refused: %s", client_reason)
+        await ws.close(code=4408, reason=_ws_close_reason(client_reason))
+        return
+
+    await ws.accept()
+
+    profile = _console_profile_from_ws(ws)
+    context = _dashboard_console_context()
+    send_lock = asyncio.Lock()
+
+    try:
+        from hermes_cli.console_engine import HermesConsoleEngine
+
+        engine = HermesConsoleEngine(
+            output_limit=_CONSOLE_OUTPUT_LIMIT,
+            context=context,  # type: ignore[arg-type]
+        )
+        if profile and profile.lower() != "current":
+            _resolve_profile_dir(profile)
+    except HTTPException as exc:
+        await _console_send(
+            ws,
+            send_lock,
+            {
+                "type": "error",
+                "message": str(exc.detail),
+                "prompt": "",
+            },
+        )
+        await ws.close(code=4400, reason=_ws_close_reason(str(exc.detail)))
+        return
+    except Exception as exc:
+        _log.exception("console failed to initialize")
+        await _console_send(
+            ws,
+            send_lock,
+            {
+                "type": "error",
+                "message": f"Console unavailable: {exc}",
+                "prompt": "",
+            },
+        )
+        await ws.close(code=1011)
+        return
+
+    _log.info(
+        "console accepted peer=%s mode=%s cred=%s context=%s profile=%s",
+        peer,
+        mode,
+        cred,
+        context,
+        profile or "current",
+    )
+    await _console_send(
+        ws,
+        send_lock,
+        {
+            "type": "ready",
+            "context": context,
+            "profile": profile or "current",
+            "prompt": _CONSOLE_PROMPT,
+        },
+    )
+
+    active_task: asyncio.Task | None = None
+    pending_confirmation: Optional[str] = None
+    command_generation = 0
+
+    async def run_command(line: str, *, confirmed: bool, command_id: int) -> None:
+        nonlocal active_task, pending_confirmation, command_generation
+        try:
+            result = await asyncio.wait_for(
+                asyncio.to_thread(
+                    _execute_console_line,
+                    engine,
+                    line,
+                    confirmed=confirmed,
+                    profile=profile,
+                ),
+                timeout=_CONSOLE_COMMAND_TIMEOUT_SECONDS,
+            )
+        except asyncio.CancelledError:
+            raise
+        except asyncio.TimeoutError:
+            if command_id == command_generation:
+                pending_confirmation = None
+                await _console_send(
+                    ws,
+                    send_lock,
+                    {
+                        "type": "error",
+                        "id": command_id,
+                        "message": (
+                            "Command timed out. Hermes Console returned to the prompt."
+                        ),
+                        "command": line,
+                    },
+                )
+                await _console_send(
+                    ws,
+                    send_lock,
+                    {
+                        "type": "complete",
+                        "id": command_id,
+                        "status": "timeout",
+                        "command": line,
+                        "prompt": _CONSOLE_PROMPT,
+                    },
+                )
+        except Exception as exc:
+            if command_id == command_generation:
+                pending_confirmation = None
+                _log.exception("console command failed")
+                await _console_send(
+                    ws,
+                    send_lock,
+                    {
+                        "type": "error",
+                        "id": command_id,
+                        "message": str(exc) or exc.__class__.__name__,
+                        "command": line,
+                    },
+                )
+                await _console_send(
+                    ws,
+                    send_lock,
+                    {
+                        "type": "complete",
+                        "id": command_id,
+                        "status": "error",
+                        "command": line,
+                        "prompt": _CONSOLE_PROMPT,
+                    },
+                )
+        else:
+            if command_id != command_generation:
+                return
+            pending_confirmation = (
+                result.command if result.status == "confirm_required" else None
+            )
+            await _console_send_result(
+                ws,
+                send_lock,
+                result,
+                command_id=command_id,
+            )
+            if result.status == "exit":
+                await ws.close(code=1000)
+        finally:
+            if command_id == command_generation:
+                active_task = None
+
+    async def start_command(line: str, *, confirmed: bool = False) -> None:
+        nonlocal active_task, command_generation
+        command_generation += 1
+        command_id = command_generation
+        active_task = asyncio.create_task(
+            run_command(line, confirmed=confirmed, command_id=command_id)
+        )
+
+    try:
+        while True:
+            try:
+                msg = await ws.receive()
+            except RuntimeError:
+                break
+            msg_type = msg.get("type")
+            if msg_type == "websocket.disconnect":
+                break
+
+            payload, error = _console_json_payload(msg)
+            if error:
+                await _console_send(
+                    ws,
+                    send_lock,
+                    {
+                        "type": "error",
+                        "message": error,
+                        "prompt": _CONSOLE_PROMPT,
+                    },
+                )
+                continue
+            if payload is None:
+                continue
+
+            frame_type = str(payload.get("type") or "").strip().lower()
+            if frame_type == "ping":
+                await _console_send(
+                    ws,
+                    send_lock,
+                    {
+                        "type": "pong",
+                        "prompt": _CONSOLE_PROMPT,
+                    },
+                )
+                continue
+
+            if frame_type == "cancel":
+                if active_task and not active_task.done():
+                    command_generation += 1
+                    active_task.cancel()
+                    active_task = None
+                    pending_confirmation = None
+                    await _console_send(
+                        ws,
+                        send_lock,
+                        {
+                            "type": "complete",
+                            "status": "cancelled",
+                            "prompt": _CONSOLE_PROMPT,
+                        },
+                    )
+                elif pending_confirmation:
+                    pending_confirmation = None
+                    await _console_send(
+                        ws,
+                        send_lock,
+                        {
+                            "type": "complete",
+                            "status": "cancelled",
+                            "prompt": _CONSOLE_PROMPT,
+                        },
+                    )
+                else:
+                    await _console_send(
+                        ws,
+                        send_lock,
+                        {
+                            "type": "complete",
+                            "status": "idle",
+                            "prompt": _CONSOLE_PROMPT,
+                        },
+                    )
+                continue
+
+            if active_task and not active_task.done():
+                await _console_send(
+                    ws,
+                    send_lock,
+                    {
+                        "type": "error",
+                        "message": "A console command is already running.",
+                        "prompt": _CONSOLE_PROMPT,
+                    },
+                )
+                continue
+
+            if frame_type == "confirm":
+                command = str(payload.get("command") or pending_confirmation or "").strip()
+                if not pending_confirmation:
+                    await _console_send(
+                        ws,
+                        send_lock,
+                        {
+                            "type": "error",
+                            "message": "No command is waiting for confirmation.",
+                            "prompt": _CONSOLE_PROMPT,
+                        },
+                    )
+                    continue
+                if command != pending_confirmation:
+                    await _console_send(
+                        ws,
+                        send_lock,
+                        {
+                            "type": "error",
+                            "message": "Confirmation does not match the pending command.",
+                            "prompt": _CONSOLE_PROMPT,
+                        },
+                    )
+                    continue
+                pending_confirmation = None
+                await start_command(command, confirmed=True)
+                continue
+
+            if frame_type in {"input", "command"}:
+                line = str(payload.get("line") or payload.get("command") or "").strip()
+                if not line:
+                    await _console_send(
+                        ws,
+                        send_lock,
+                        {
+                            "type": "complete",
+                            "status": "ok",
+                            "prompt": _CONSOLE_PROMPT,
+                        },
+                    )
+                    continue
+                if pending_confirmation:
+                    await _console_send(
+                        ws,
+                        send_lock,
+                        {
+                            "type": "error",
+                            "message": (
+                                "Confirm or cancel the pending command before "
+                                "running another one."
+                            ),
+                            "prompt": _CONSOLE_PROMPT,
+                        },
+                    )
+                    continue
+                await start_command(line)
+                continue
+
+            await _console_send(
+                ws,
+                send_lock,
+                {
+                    "type": "error",
+                    "message": f"Unsupported console frame: {frame_type or '?'}",
+                    "prompt": _CONSOLE_PROMPT,
+                },
+            )
+    except WebSocketDisconnect:
+        pass
+    finally:
+        if active_task and not active_task.done():
+            active_task.cancel()
+            try:
+                await active_task
+            except (asyncio.CancelledError, Exception):
+                pass
+
+
 @app.websocket("/api/pty")
 async def pty_ws(ws: WebSocket) -> None:
     peer = ws.client.host if ws.client else "?"

tests/hermes_cli/test_dashboard_auth_ws_auth.py

@@ -1,9 +1,9 @@
 """Tests for the WS-upgrade auth helper (Phase 5 task 5.2).
 
-The dashboard's four WS endpoints (``/api/pty``, ``/api/ws``, ``/api/pub``,
-``/api/events``) share an auth gate: ``_ws_auth_ok``. In loopback mode it
-accepts ``?token=<_SESSION_TOKEN>``; in gated mode it accepts a single-use
-``?ticket=`` minted by ``POST /api/auth/ws-ticket``.
+The dashboard's WS endpoints (``/api/pty``, ``/api/console``, ``/api/ws``,
+``/api/pub``, ``/api/events``) share an auth gate: ``_ws_auth_ok``. In
+loopback mode it accepts ``?token=<_SESSION_TOKEN>``; in gated mode it accepts
+a single-use ``?ticket=`` minted by ``POST /api/auth/ws-ticket``.
 
 These tests exercise the helper at the unit level (no actual WS upgrade)
 plus the ticket-mint endpoint under realistic gated-mode setup. We don't
@@ -315,9 +315,10 @@ class TestWsRequestIsAllowedGated:
     (intended only for unauthenticated loopback dev) must not also reject
     those upgrades: the OAuth gate + single-use ticket is the auth.
 
-    Regression coverage: every WS endpoint (``/api/pty``, ``/api/ws``,
-    ``/api/pub``, ``/api/events``) calls ``_ws_request_is_allowed`` after
-    ``_ws_auth_ok``. If the peer-IP check rejects gated mode, the chat
+    Regression coverage: every WS endpoint (``/api/pty``, ``/api/console``,
+    ``/api/ws``, ``/api/pub``, ``/api/events``) calls
+    ``_ws_request_is_allowed`` after ``_ws_auth_ok``. If the peer-IP check
+    rejects gated mode, the chat
     tab + sidebar tool feed silently fail to connect even after a
     successful OAuth login.
     """

tests/hermes_cli/test_web_server_console_ws.py

@@ -0,0 +1,134 @@
+"""Dashboard Hermes Console websocket tests."""
+
+from __future__ import annotations
+
+import time
+from urllib.parse import urlencode
+
+import pytest
+from starlette.testclient import TestClient
+from starlette.websockets import WebSocketDisconnect
+
+from hermes_cli import web_server
+
+
+@pytest.fixture
+def console_client(monkeypatch, _isolate_hermes_home):
+    previous_auth_required = getattr(web_server.app.state, "auth_required", None)
+    previous_bound_host = getattr(web_server.app.state, "bound_host", None)
+    web_server.app.state.auth_required = False
+    web_server.app.state.bound_host = None
+    monkeypatch.setattr(web_server, "_DASHBOARD_EMBEDDED_CHAT_ENABLED", True)
+
+    client = TestClient(web_server.app)
+    try:
+        yield client
+    finally:
+        close = getattr(client, "close", None)
+        if close is not None:
+            close()
+        if previous_auth_required is None:
+            if hasattr(web_server.app.state, "auth_required"):
+                delattr(web_server.app.state, "auth_required")
+        else:
+            web_server.app.state.auth_required = previous_auth_required
+        if previous_bound_host is None:
+            if hasattr(web_server.app.state, "bound_host"):
+                delattr(web_server.app.state, "bound_host")
+        else:
+            web_server.app.state.bound_host = previous_bound_host
+
+
+def _url(token: str | None = None, **params: str) -> str:
+    query = {"token": web_server._SESSION_TOKEN, **params}
+    if token is not None:
+        query["token"] = token
+    return f"/api/console?{urlencode(query)}"
+
+
+def _recv_until(conn, frame_type: str, *, status: str | None = None) -> dict:
+    deadline = time.monotonic() + 5.0
+    while time.monotonic() < deadline:
+        frame = conn.receive_json()
+        if frame.get("type") != frame_type:
+            continue
+        if status is not None and frame.get("status") != status:
+            continue
+        return frame
+    raise AssertionError(f"Timed out waiting for {frame_type} frame")
+
+
+def test_console_ws_rejects_missing_or_bad_token(console_client):
+    with pytest.raises(WebSocketDisconnect) as exc:
+        with console_client.websocket_connect("/api/console"):
+            pass
+    assert exc.value.code == 4401
+
+    with pytest.raises(WebSocketDisconnect) as exc:
+        with console_client.websocket_connect(_url(token="wrong")):
+            pass
+    assert exc.value.code == 4401
+
+
+def test_console_ws_runs_read_only_command(console_client):
+    with console_client.websocket_connect(_url()) as conn:
+        ready = conn.receive_json()
+        assert ready["type"] == "ready"
+        assert ready["context"] == "local"
+        assert ready["prompt"] == "hermes> "
+
+        conn.send_json({"type": "input", "line": "help"})
+
+        output = _recv_until(conn, "output")
+        assert "Hermes Console" in output["data"]
+        complete = _recv_until(conn, "complete", status="ok")
+        assert complete["prompt"] == "hermes> "
+
+
+def test_console_ws_confirmed_command_executes_after_confirmation(console_client):
+    from hermes_cli.config import load_config
+
+    with console_client.websocket_connect(_url()) as conn:
+        assert conn.receive_json()["type"] == "ready"
+        conn.send_json({"type": "input", "line": "config set display.interface cli"})
+
+        confirmation = _recv_until(conn, "confirm_required")
+        assert confirmation["command"] == "config set display.interface cli"
+        assert confirmation["message"]
+
+        conn.send_json({"type": "confirm", "command": confirmation["command"]})
+        _recv_until(conn, "complete", status="ok")
+
+    assert load_config()["display"]["interface"] == "cli"
+
+
+def test_console_ws_uses_hosted_context_for_opt_data_policy(console_client, monkeypatch):
+    monkeypatch.setattr(web_server, "_default_hermes_root_is_opt_data", lambda: True)
+
+    with console_client.websocket_connect(_url()) as conn:
+        ready = conn.receive_json()
+        assert ready["type"] == "ready"
+        assert ready["context"] == "hosted"
+
+        conn.send_json({"type": "input", "line": "profile create nope"})
+
+        error = _recv_until(conn, "error")
+        assert "hosted Hermes Console" in error["message"]
+
+
+def test_console_ws_cancel_returns_to_prompt(console_client, monkeypatch):
+    from hermes_cli.console_engine import ConsoleResult, HermesConsoleEngine
+
+    def slow_execute(self, line: str, *, confirmed: bool = False):
+        time.sleep(0.5)
+        return ConsoleResult("ok", output="late", command=line)
+
+    monkeypatch.setattr(HermesConsoleEngine, "execute", slow_execute)
+
+    with console_client.websocket_connect(_url()) as conn:
+        assert conn.receive_json()["type"] == "ready"
+        conn.send_json({"type": "input", "line": "status"})
+        conn.send_json({"type": "cancel"})
+
+        complete = _recv_until(conn, "complete", status="cancelled")
+        assert complete["prompt"] == "hermes> "