feat: web terminal + Headlamp-style K8s browser

agent/shipper.py

@@ -21,13 +21,15 @@
 """
 
 import argparse
+import base64
 import hashlib
 import json
 import logging
 import logging.handlers
 import os
 import re
 import socket
+import ssl
 import subprocess
 import sys
 import time
@@ -76,6 +78,12 @@
 EXEC_RESULT_URL = f"{API_URL}/api/v1/exec/result"
 EXEC_POLL_INTERVAL = 3
 
+# Interactive terminal (PTY relayed to the browser). Opt-out with PYXIS_TERMINAL_ENABLED=0.
+TERMINAL_ENABLED = os.environ.get("PYXIS_TERMINAL_ENABLED", "1") not in ("0", "false", "no")
+_WS_BASE = API_URL.replace("https://", "wss://", 1).replace("http://", "ws://", 1)
+TERMINAL_WS_URL = f"{_WS_BASE}/api/v1/terminal/ws-agent"
+TERMINAL_SHELL = os.environ.get("PYXIS_TERMINAL_SHELL", os.environ.get("SHELL", "/bin/bash"))
+
 SHIPPER_PATH = os.environ.get("PYXIS_SHIPPER_PATH", "/opt/pyxis/shipper.py")
 CONFIG_PATH  = os.environ.get("PYXIS_CONFIG_PATH", "/opt/pyxis/config.json")
 AUTOUPDATE_INTERVAL = int(os.environ.get("PYXIS_AUTOUPDATE_INTERVAL", "300"))  # 5 min
@@ -720,6 +728,213 @@ def _exec_allowed(cmd: str) -> tuple[bool, str]:
     return False, "not in allowlist"
 
 
+# ── Interactive terminal (PTY ↔ WebSocket) ────────────────────────────────────
+#
+# Self-hosted, zero third-party. The agent dials the backend over a WebSocket
+# (hand-rolled here to keep this file stdlib-only, like the rest of the agent),
+# spawns a PTY shell, and pumps bytes both ways. The shell runs as whatever user
+# the agent runs as (the restricted 'pyxis' user in the hardened install), so a
+# browser terminal is no more privileged than the agent itself.
+
+
+class _WSClient:
+    """Minimal RFC6455 WebSocket client (text frames). Client frames are masked."""
+
+    def __init__(self, url: str):
+        from urllib.parse import urlparse
+
+        u = urlparse(url)
+        self.secure = u.scheme == "wss"
+        self.host = u.hostname or "localhost"
+        self.port = u.port or (443 if self.secure else 80)
+        self.path = u.path or "/"
+        self.sock: socket.socket | None = None
+        self._buf = b""
+
+    def connect(self, timeout: float = 10.0) -> None:
+        raw = socket.create_connection((self.host, self.port), timeout=timeout)
+        if self.secure:
+            ctx = ssl.create_default_context()
+            raw = ctx.wrap_socket(raw, server_hostname=self.host)
+        self.sock = raw
+        key = base64.b64encode(os.urandom(16)).decode()
+        req = (
+            f"GET {self.path} HTTP/1.1\r\n"
+            f"Host: {self.host}:{self.port}\r\n"
+            "Upgrade: websocket\r\nConnection: Upgrade\r\n"
+            f"Sec-WebSocket-Key: {key}\r\nSec-WebSocket-Version: 13\r\n\r\n"
+        )
+        self.sock.sendall(req.encode())
+        # Read until end of headers
+        while b"\r\n\r\n" not in self._buf:
+            chunk = self.sock.recv(4096)
+            if not chunk:
+                raise ConnectionError("WS handshake: connection closed")
+            self._buf += chunk
+        head, self._buf = self._buf.split(b"\r\n\r\n", 1)
+        if b"101" not in head.split(b"\r\n", 1)[0]:
+            raise ConnectionError(f"WS handshake failed: {head[:80]!r}")
+
+    def send_text(self, data: str) -> None:
+        payload = data.encode()
+        header = bytearray([0x81])  # FIN + text
+        n = len(payload)
+        if n < 126:
+            header.append(0x80 | n)
+        elif n < 65536:
+            header.append(0x80 | 126)
+            header += n.to_bytes(2, "big")
+        else:
+            header.append(0x80 | 127)
+            header += n.to_bytes(8, "big")
+        mask = os.urandom(4)
+        header += mask
+        masked = bytes(b ^ mask[i % 4] for i, b in enumerate(payload))
+        assert self.sock is not None
+        self.sock.sendall(bytes(header) + masked)
+
+    def _read_exact(self, n: int) -> bytes:
+        assert self.sock is not None
+        while len(self._buf) < n:
+            chunk = self.sock.recv(65536)
+            if not chunk:
+                raise ConnectionError("WS closed")
+            self._buf += chunk
+        out, self._buf = self._buf[:n], self._buf[n:]
+        return out
+
+    def recv_text(self) -> str | None:
+        """Return the next text message, or None on close. Handles ping/pong."""
+        while True:
+            b0, b1 = self._read_exact(2)
+            opcode = b0 & 0x0F
+            masked = b1 & 0x80
+            length = b1 & 0x7F
+            if length == 126:
+                length = int.from_bytes(self._read_exact(2), "big")
+            elif length == 127:
+                length = int.from_bytes(self._read_exact(8), "big")
+            mask = self._read_exact(4) if masked else b""
+            payload = self._read_exact(length)
+            if masked:
+                payload = bytes(b ^ mask[i % 4] for i, b in enumerate(payload))
+            if opcode == 0x8:        # close
+                return None
+            if opcode == 0x9:        # ping → pong
+                self._send_control(0x8A, payload)
+                continue
+            if opcode == 0xA:        # pong
+                continue
+            return payload.decode("utf-8", "replace")
+
+    def _send_control(self, op: int, payload: bytes) -> None:
+        mask = os.urandom(4)
+        masked = bytes(b ^ mask[i % 4] for i, b in enumerate(payload))
+        assert self.sock is not None
+        self.sock.sendall(bytes([op, 0x80 | len(payload)]) + mask + masked)
+
+    def close(self) -> None:
+        try:
+            if self.sock:
+                self._send_control(0x88, b"")
+                self.sock.close()
+        except Exception:
+            pass
+
+
+def run_pty_session(session_id: str, argv: list | None = None) -> None:
+    """
+    Spawn a PTY and bridge it to the backend over a WebSocket.
+    argv = [] → the node's default shell; otherwise exec that command (e.g.
+    `kubectl exec -it ...` for a pod shell). argv[0] is resolved via PATH.
+    """
+    if os.name != "posix":
+        log.warning("terminal: PTY not supported on this platform")
+        return
+    import pty
+    import fcntl
+    import termios
+    import struct
+    import select
+
+    command = list(argv) if argv else [TERMINAL_SHELL]
+    ws = _WSClient(f"{TERMINAL_WS_URL}/{session_id}")
+    pid = master = -1
+    try:
+        ws.connect()
+        ws.send_text(json.dumps({"api_key": API_KEY, "node_name": NODE_NAME}))
+        log.info("terminal: session %s attaching PTY (%s)", session_id[:8], command[0])
+
+        pid, master = pty.fork()
+        if pid == 0:  # child → exec the command
+            os.environ["TERM"] = "xterm-256color"
+            try:
+                os.execvp(command[0], command)
+            except Exception:
+                pass
+            os._exit(127)
+
+        # Parent: pump master fd ↔ websocket. A reader thread handles ws→pty
+        # so the main loop can select() on the pty for pty→ws.
+        stop = threading.Event()
+
+        def ws_reader() -> None:
+            while not stop.is_set():
+                try:
+                    raw = ws.recv_text()
+                except Exception:
+                    break
+                if raw is None:
+                    break
+                try:
+                    msg = json.loads(raw)
+                except Exception:
+                    continue
+                t = msg.get("t")
+                if t == "d":
+                    os.write(master, base64.b64decode(msg["b"]))
+                elif t == "resize":
+                    winsize = struct.pack("HHHH", int(msg.get("rows", 24)),
+                                          int(msg.get("cols", 80)), 0, 0)
+                    fcntl.ioctl(master, termios.TIOCSWINSZ, winsize)
+                elif t == "close":
+                    break
+            stop.set()
+
+        threading.Thread(target=ws_reader, daemon=True).start()
+
+        while not stop.is_set():
+            r, _, _ = select.select([master], [], [], 0.5)
+            if master in r:
+                try:
+                    data = os.read(master, 65536)
+                except OSError:
+                    break
+                if not data:
+                    break
+                ws.send_text(json.dumps({"t": "d", "b": base64.b64encode(data).decode()}))
+        try:
+            ws.send_text(json.dumps({"t": "exit", "code": 0}))
+        except Exception:
+            pass
+    except Exception as e:
+        log.warning("terminal: session %s error: %s", session_id[:8], e)
+    finally:
+        try:
+            if master >= 0:
+                os.close(master)
+        except Exception:
+            pass
+        try:
+            if pid > 0:
+                os.kill(pid, 9)
+                os.waitpid(pid, 0)
+        except Exception:
+            pass
+        ws.close()
+        log.info("terminal: session %s closed", session_id[:8])
+
+
 def exec_loop() -> None:
     """Poll for commands from the backend and execute them."""
     while True:
@@ -733,6 +948,21 @@ def exec_loop() -> None:
             with urllib.request.urlopen(req, timeout=10) as resp:
                 data = json.loads(resp.read())
 
+            # Interactive terminal sessions requested from the browser.
+            if TERMINAL_ENABLED:
+                for item in data.get("pty", []) or []:
+                    try:
+                        spec = json.loads(item)
+                    except Exception:
+                        spec = {"sid": item, "argv": []}
+                    sid = spec.get("sid")
+                    if sid:
+                        threading.Thread(
+                            target=run_pty_session,
+                            args=(sid, spec.get("argv") or []),
+                            daemon=True,
+                        ).start()
+
             cmd_id = data.get("cmd_id")
             cmd = data.get("cmd")
             if cmd_id and cmd:
@@ -1446,6 +1676,41 @@ def _kubectl_get(resource: str, all_namespaces: bool = False) -> list:
     return []
 
 
+# Resource kinds surfaced in the cluster browser. (kind -> kubectl resource name)
+_K8S_RESOURCE_KINDS = {
+    "services":          "services",
+    "configmaps":        "configmaps",
+    "secrets":           "secrets",
+    "statefulsets":      "statefulsets",
+    "daemonsets":        "daemonsets",
+    "replicasets":       "replicasets",
+    "jobs":              "jobs",
+    "cronjobs":          "cronjobs",
+    "ingresses":         "ingresses",
+    "endpoints":         "endpoints",
+    "networkpolicies":   "networkpolicies",
+    "pvcs":              "persistentvolumeclaims",
+    "pvs":               "persistentvolumes",
+    "serviceaccounts":   "serviceaccounts",
+    "events":            "events",
+}
+
+
+def _collect_k8s_resources() -> dict:
+    """Collect the extended resource set for the cluster browser."""
+    out: dict[str, list] = {}
+    for kind, resource in _K8S_RESOURCE_KINDS.items():
+        items = _kubectl_get(resource, all_namespaces=True)
+        if kind == "secrets":
+            # Never ship secret values — keep metadata + key names only.
+            for it in items:
+                data = it.get("data") or {}
+                it["data"] = {k: None for k in data}
+                it.pop("stringData", None)
+        out[kind] = items
+    return out
+
+
 def k8s_monitor_loop() -> None:
     """Snapshot cluster state and push to backend every K8S_MONITOR_INTERVAL seconds."""
     # Check kubectl is available
@@ -1457,10 +1722,15 @@ def k8s_monitor_loop() -> None:
     while True:
         try:
             state = {
+                # Which agent host has kubectl access — backend routes pod-exec here.
+                "agent_node":  NODE_NAME,
                 "nodes":       _kubectl_get("nodes"),
                 "pods":        _kubectl_get("pods",        all_namespaces=True),
                 "deployments": _kubectl_get("deployments", all_namespaces=True),
                 "namespaces":  _kubectl_get("namespaces"),
+                # Headlamp-style resource browser. Full -o json kept so the UI can
+                # render each item's YAML offline. Secret values are stripped below.
+                "resources":   _collect_k8s_resources(),
             }
             payload = json.dumps(state).encode()
             req = urllib.request.Request(

backend/.env.example

@@ -1,5 +1,14 @@
 DATABASE_URL=postgresql+asyncpg://pyxis:pyxis@localhost:5432/pyxis
 REDIS_URL=redis://localhost:6379
 ANTHROPIC_API_KEY=sk-ant-...
+# Model tiers (optional overrides). FAST=chat/extraction, MODEL=balanced, DEEP=incident RCA.
+CLAUDE_MODEL=claude-sonnet-4-6
+CLAUDE_MODEL_FAST=claude-haiku-4-5-20251001
+CLAUDE_MODEL_DEEP=claude-opus-4-8
 SECRET_KEY=change-me-in-production-use-32-random-bytes
 DEBUG=false
+# Interactive browser terminal (PTY via the agent). Set false to disable entirely.
+TERMINAL_ENABLED=true
+TERMINAL_IDLE_TIMEOUT=600
+TERMINAL_MAX_DURATION=3600
+TERMINAL_MAX_PER_TENANT=5

backend/app/ai/engine.py

@@ -271,7 +271,7 @@ async def _run_rca(
         for p in past_incidents
     ) or "(no prior incidents)"
 
-    log.info("_run_rca: calling Claude for incident %s (model=%s)", incident.id, settings.CLAUDE_MODEL)
+    log.info("_run_rca: calling Claude for incident %s (model=%s)", incident.id, settings.CLAUDE_MODEL_DEEP)
 
     # 7. Call Claude
     system_prompt = (
@@ -333,7 +333,7 @@ async def _run_rca(
 
     try:
         message = await _anthropic.messages.create(
-            model=settings.CLAUDE_MODEL,
+            model=settings.CLAUDE_MODEL_DEEP,
             max_tokens=2048,
             system=system_prompt,
             messages=[{"role": "user", "content": user_prompt}],

backend/app/api/routes/assistant.py

@@ -66,7 +66,7 @@ async def chat(
 
     try:
         response = await _anthropic.messages.create(
-            model=settings.CLAUDE_MODEL,
+            model=settings.CLAUDE_MODEL_FAST,
             max_tokens=1024,
             system=system,
             messages=messages,

backend/app/api/routes/deploy_events.py

@@ -1,5 +1,5 @@
 import uuid
-from datetime import datetime
+from datetime import datetime, timezone
 from typing import Any
 
 from fastapi import APIRouter, Depends
@@ -50,7 +50,7 @@ async def create_deploy_event(
         version=payload.version,
         deployed_by=payload.deployed_by,
         environment=payload.environment,
-        deployed_at=payload.deployed_at or datetime.utcnow(),
+        deployed_at=payload.deployed_at or datetime.now(timezone.utc),
         meta=payload.meta,
     )
     db.add(event)