Phase 21: container terminal (web exec), local + agent (0.27.0)

Interactive shell into a compose-managed container over WebSocket + xterm.js,
opened from the container card on the stack Overview tab. Admin-only (non-admin
handshake rejected with 4403); only containers with the compose project label
are reachable.

- backend services/exec_service.py: create/start/resize exec + a shared
  bidirectional pump_exec (recv/sendall on sock._sock, executor thread,
  resize control frames, exit-code frame).
- routers/ws.py: _authorize_admin + /ws/exec/{container_id} and the
  /ws/agent-exec/{agent_id}/{container_id} proxy (forwards BOTH directions).
- agent_app.py: /agent/ws/exec/{container_id}.
- frontend: @xterm/xterm + @xterm/addon-fit; ContainerTerminal modal (shell
  picker, fit/resize, exit/error handling) + a Terminal button on ContainerCard.

Live-verified (TestClient): local happy/exit/guard/4403/4401, agent happy/4401,
proxy bidirectional round-trip.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

backend/routers/ws.py

@@ -17,7 +17,7 @@ from auth import decode_token
 from database import engine
 from models.agent import Agent
 from models.setting import EVENT_STACK_ERROR, EVENT_STACK_START
-from services import audit_service, compose_service, notify_service
+from services import audit_service, compose_service, exec_service, notify_service
 
 logger = logging.getLogger("stackpilot.ws")
 
@@ -37,6 +37,24 @@ async def _authorize(websocket: WebSocket, token: str | None) -> bool:
     return True
 
 
+async def _authorize_admin(websocket: WebSocket, token: str | None) -> bool:
+    """Like _authorize but also requires the admin role (exec is root-equivalent).
+
+    Closes 4401 on a missing/invalid token, 4403 on a valid non-admin token."""
+    if not token:
+        await websocket.close(code=4401)
+        return False
+    try:
+        payload = decode_token(token, "access")
+    except (JWTError, Exception):  # noqa: BLE001
+        await websocket.close(code=4401)
+        return False
+    if payload.get("role") != "admin":
+        await websocket.close(code=4403)
+        return False
+    return True
+
+
 async def _stream_logs(websocket: WebSocket, stack_id: str, service: str | None):
     """Stream `docker compose logs -f` output to the client."""
     args = ["logs", "--no-color", "--tail", "200", "--timestamps", "-f"]
@@ -348,3 +366,133 @@ async def ws_events(
     finally:
         stop.set()
         task.cancel()
+
+
+@router.websocket("/ws/exec/{container_id}")
+async def ws_exec(
+    websocket: WebSocket,
+    container_id: str,
+    token: str | None = Query(default=None),
+    cmd: str | None = Query(default=None),
+):
+    """Interactive shell into a compose-managed container (admin only)."""
+    await websocket.accept()
+    if not await _authorize_admin(websocket, token):
+        return
+    username = decode_token(token, "access").get("sub", "unknown") if token else "unknown"
+    shell = cmd or exec_service.DEFAULT_SHELL
+
+    try:
+        exec_id = exec_service.create_exec(container_id, [shell])
+        holder, raw = exec_service.start_exec(exec_id)
+    except Exception as exc:  # noqa: BLE001
+        with contextlib.suppress(Exception):
+            await websocket.send_text(json.dumps({"type": "error", "detail": str(exc)}))
+        with contextlib.suppress(Exception):
+            await websocket.close()
+        return
+
+    with contextlib.suppress(Exception):
+        with Session(engine) as session:
+            audit_service.record(
+                session, user=username, action="container.exec",
+                target=container_id[:12], detail=shell, ip="ws",
+            )
+
+    try:
+        await exec_service.pump_exec(websocket, exec_id, holder, raw)
+    except WebSocketDisconnect:
+        pass
+    finally:
+        with contextlib.suppress(Exception):
+            await websocket.close()
+
+
+@router.websocket("/ws/agent-exec/{agent_id}/{container_id}")
+async def ws_agent_exec(
+    websocket: WebSocket,
+    agent_id: int,
+    container_id: str,
+    token: str | None = Query(default=None),
+    cmd: str | None = Query(default=None),
+):
+    """Proxy an interactive exec session to a remote agent (admin only).
+
+    Unlike the log/deploy proxies this forwards in BOTH directions so keystrokes
+    reach the container and its output streams back."""
+    await websocket.accept()
+    if not await _authorize_admin(websocket, token):
+        return
+
+    with Session(engine) as session:
+        agent = session.get(Agent, agent_id)
+    if not agent:
+        await websocket.send_text(json.dumps({"type": "error", "detail": "agent not found"}))
+        await websocket.close()
+        return
+
+    base = agent.url.rstrip("/")
+    ws_url = ("wss://" + base[8:] if base.startswith("https://")
+              else "ws://" + base[7:] if base.startswith("http://")
+              else "ws://" + base)
+    ws_url += f"/agent/ws/exec/{container_id}?token={urllib.parse.quote(agent.token, safe='')}"
+    if cmd:
+        ws_url += f"&cmd={urllib.parse.quote(cmd, safe='')}"
+
+    async def _err(detail: str) -> None:
+        with contextlib.suppress(Exception):
+            await websocket.send_text(json.dumps({"type": "error", "detail": detail}))
+
+    try:
+        upstream = await websockets.connect(ws_url, open_timeout=10, ping_interval=20)
+    except websockets.InvalidStatus as exc:
+        code = getattr(getattr(exc, "response", None), "status_code", None)
+        hint = " — the agent may be running an old version without terminal support; update it." if code == 404 else ""
+        logger.warning("Agent exec proxy: handshake to %s failed (%s)", agent.name, code)
+        await _err(f"Agent '{agent.name}' rejected the terminal (HTTP {code}){hint}")
+        with contextlib.suppress(Exception):
+            await websocket.close()
+        return
+    except Exception as exc:  # noqa: BLE001
+        logger.warning("Agent exec proxy: cannot reach %s at %s: %s", agent.name, agent.url, exc)
+        await _err(f"Could not connect to agent '{agent.name}' at {agent.url}: {exc}")
+        with contextlib.suppress(Exception):
+            await websocket.close()
+        return
+
+    with contextlib.suppress(Exception):
+        with Session(engine) as session:
+            username = decode_token(token, "access").get("sub", "unknown") if token else "unknown"
+            audit_service.record(
+                session, user=username, action="agent.container.exec",
+                target=f"{agent.name}/{container_id[:12]}", ip="ws",
+            )
+
+    async def browser_to_agent() -> None:
+        try:
+            while True:
+                msg = await websocket.receive_text()
+                await upstream.send(msg)
+        except (WebSocketDisconnect, websockets.ConnectionClosed):
+            pass
+
+    async def agent_to_browser() -> None:
+        try:
+            async for message in upstream:
+                await websocket.send_text(
+                    message if isinstance(message, str) else message.decode("utf-8", "replace")
+                )
+        except (WebSocketDisconnect, websockets.ConnectionClosed):
+            pass
+
+    b2a = asyncio.create_task(browser_to_agent())
+    a2b = asyncio.create_task(agent_to_browser())
+    done, pending = await asyncio.wait({b2a, a2b}, return_when=asyncio.FIRST_COMPLETED)
+    for task in pending:
+        task.cancel()
+    with contextlib.suppress(Exception):
+        await asyncio.gather(*pending, return_exceptions=True)
+    with contextlib.suppress(Exception):
+        await upstream.close()
+    with contextlib.suppress(Exception):
+        await websocket.close()