"""Mobile web chat front-end for the operator bridge (`operator web`).
Turns a running operator daemon into a phone-friendly chat page. The daemon
already exposes the room as a local API over its AF_UNIX control socket
(`read` with long-poll, `say`, `roster`, `status`); this module is a thin,
**stdlib-only** HTTP shim in front of that socket so a browser can drive it:
GET / the single-page chat UI (self-contained HTML/JS)
GET /api/events?since=N long-poll: proxies {"op":"read","wait":true}
POST /api/say {text} proxies {"op":"say"}
GET /api/status proxies {"op":"status"} (roster + connection)
Why this exists: on Termux the tmux `read --wait` loop is read-only — replying
means a second shell typing `operator say …`, which is painful on a touch
keyboard. A browser page gives a real input box, scrollback and live updates
with zero extra dependencies (no server stack, no JS build — just http.server).
Binds 127.0.0.1 by default: the control socket grants room-send rights, so the
listener stays loopback-only unless you deliberately 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__
"""
class _Handler(BaseHTTPRequestHandler):
session_name = "default" # set on the class before serving
protocol_version = "HTTP/1.1"
# ── plumbing ─────────────────────────────────────────────────────────
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 == "/" or url.path == "/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/events":
q = parse_qs(url.query)
since = int((q.get("since") or ["0"])[0])
resp = self._bridge(
{"op": "read", "since": since, "wait": True, "timeout": _POLL_TIMEOUT},
read_timeout=_POLL_TIMEOUT + 5.0)
self._send_json(resp)
return
self._send_json({"ok": False, "error": "not found"}, code=404)
def do_POST(self) -> None:
url = urlparse(self.path)
if url.path != "/api/say":
self._send_json({"ok": False, "error": "not found"}, code=404)
return
length = int(self.headers.get("Content-Length") or 0)
raw = self.rfile.read(length) if length else b"{}"
try:
text = str(json.loads(raw.decode()).get("text", "")).strip()
except (ValueError, AttributeError):
self._send_json({"ok": False, "error": "bad json"}, code=400)
return
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))
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 UI for session '{sess.name}' → http://{shown}:{port} (Ctrl-C to stop)")
try:
httpd.serve_forever()
except KeyboardInterrupt:
print("\nweb UI stopped")
finally:
httpd.server_close()
return 0