"""Mobile hack-house console for the operator bridge (`operator web`).
A phone-friendly browser mirror of a live hack-house room. The daemon already
exposes the room as a local API over its AF_UNIX control socket; this module is
a thin, **stdlib-only** HTTP shim in front of it so a browser gets the same
things the Rust TUI shows — the shared sandbox terminal, chat, roster — mobile
optimised:
GET / the single-page console (self-contained HTML/JS)
GET /api/status proxies {"op":"status"} (connection/roster/driver)
GET /api/events?since=N long-poll: proxies {"op":"read","wait":true}
POST /api/say {text} proxies {"op":"say"}
GET /api/screen?ansi=1 proxies {"op":"screen"} — the relayed shared PTY
POST /api/keys {tokens} proxies {"op":"keys"} — drive the shared PTY
Why this exists: on Termux the tmux `read --wait` loop is read-only and shows
none of the sandbox. A browser page gives the terminal view + a real input box
+ live updates with zero extra dependencies (no server stack, no JS build).
Binds 127.0.0.1 by default: the control socket grants room-send/drive rights,
so the listener stays loopback-only unless you pass a routable --bind.
"""
from __future__ import annotations
import json
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from urllib.parse import urlparse, parse_qs
from .cli_client import BridgeUnreachable, request
from .session import resolve
# Long-poll window the browser holds open per /api/events call. The socket read
# timeout must outlast it (see request(read_timeout=...)).
_POLL_TIMEOUT = 25.0
_PAGE = """
hack-house · __SESSION____SESSION__watching—
no shared sandbox yet — a room member runs /sbx and /grants this operator to drive it here.
"""
class _Handler(BaseHTTPRequestHandler):
session_name = "default" # set on the class before serving
protocol_version = "HTTP/1.1"
def log_message(self, *a): # silence per-request stderr spam
pass
def _sock(self):
return resolve(self.session_name).sock_path
def _send_json(self, obj: dict, code: int = 200) -> None:
body = json.dumps(obj).encode()
self.send_response(code)
self.send_header("Content-Type", "application/json")
self.send_header("Content-Length", str(len(body)))
self.end_headers()
self.wfile.write(body)
def _bridge(self, req: dict, read_timeout: float = 35.0) -> dict:
try:
return request(self._sock(), req, read_timeout=read_timeout)
except BridgeUnreachable as e:
return {"ok": False, "error": str(e), "offline": True}
# ── routes ───────────────────────────────────────────────────────────
def do_GET(self) -> None:
url = urlparse(self.path)
if url.path in ("/", "/index.html"):
page = _PAGE.replace("__SESSION__", resolve(self.session_name).name)
body = page.encode()
self.send_response(200)
self.send_header("Content-Type", "text/html; charset=utf-8")
self.send_header("Content-Length", str(len(body)))
self.end_headers()
self.wfile.write(body)
return
if url.path == "/api/status":
self._send_json(self._bridge({"op": "status"}, read_timeout=8))
return
if url.path == "/api/screen":
q = parse_qs(url.query)
ansi = (q.get("ansi") or ["0"])[0] not in ("0", "", "false")
tail = q.get("tail")
req = {"op": "screen", "ansi": ansi}
if tail:
req["tail"] = int(tail[0])
self._send_json(self._bridge(req, read_timeout=8))
return
if url.path == "/api/events":
q = parse_qs(url.query)
since = int((q.get("since") or ["0"])[0])
self._send_json(self._bridge(
{"op": "read", "since": since, "wait": True, "timeout": _POLL_TIMEOUT},
read_timeout=_POLL_TIMEOUT + 5.0))
return
self._send_json({"ok": False, "error": "not found"}, code=404)
def _read_body(self) -> dict:
length = int(self.headers.get("Content-Length") or 0)
raw = self.rfile.read(length) if length else b"{}"
try:
obj = json.loads(raw.decode())
return obj if isinstance(obj, dict) else {}
except ValueError:
return {}
def do_POST(self) -> None:
url = urlparse(self.path)
if url.path == "/api/say":
text = str(self._read_body().get("text", "")).strip()
if not text:
self._send_json({"ok": False, "error": "empty"}, code=400)
return
self._send_json(self._bridge({"op": "say", "text": text}, read_timeout=15))
return
if url.path == "/api/keys":
tokens = self._read_body().get("tokens")
if not isinstance(tokens, list) or not tokens:
self._send_json({"ok": False, "error": "no tokens"}, code=400)
return
self._send_json(self._bridge({"op": "keys", "keys": tokens}, read_timeout=10))
return
self._send_json({"ok": False, "error": "not found"}, code=404)
def serve(session_name: str, bind: str, port: int) -> int:
"""Run the web front-end until Ctrl-C. Returns a process exit code."""
sess = resolve(session_name)
if not sess.sock_path.exists():
print(f"no live bridge for session '{sess.name}' (run `up` first)")
return 2
_Handler.session_name = session_name
httpd = ThreadingHTTPServer((bind, port), _Handler)
httpd.daemon_threads = True
shown = bind if bind not in ("0.0.0.0", "::") else "127.0.0.1"
print(f"operator web console for session '{sess.name}' → http://{shown}:{port} (Ctrl-C to stop)")
try:
httpd.serve_forever()
except KeyboardInterrupt:
print("\nweb console stopped")
finally:
httpd.server_close()
return 0