"""
AutoDev - Web UI
Serves a live dashboard showing LLM activity, file tree, and file contents.
Uses SSE (Server-Sent Events) for real-time updates. No external dependencies.
"""
import http.server
import json
import os
import queue
import threading
import urllib.parse
from . import config
# Global event system โ broadcast to all connected SSE clients
_clients: list = [] # list of queue.Queue, one per connected client
_clients_lock = threading.Lock()
_event_history: list = [] # buffer of all past events for new clients
_workdir: str = ""
_shutdown: bool = False
_server = None
def push_event(event_type: str, data: dict):
"""Push an event to all connected web clients."""
msg = {"type": event_type, "data": data}
_event_history.append(msg)
with _clients_lock:
for q in _clients:
try:
q.put_nowait(msg)
except queue.Full:
pass
class WebHandler(http.server.BaseHTTPRequestHandler):
def log_message(self, format, *args):
pass # Suppress default logging
def do_GET(self):
try:
parsed = urllib.parse.urlparse(self.path)
path = parsed.path
params = urllib.parse.parse_qs(parsed.query)
if path == "/":
self._serve_html()
elif path == "/events":
self._serve_sse()
elif path == "/api/files":
self._serve_file_tree()
elif path == "/api/file":
filepath = params.get("path", [""])[0]
self._serve_file_content(filepath)
else:
self.send_error(404)
except (BrokenPipeError, ConnectionResetError, OSError):
pass
def _serve_html(self):
self.send_response(200)
self.send_header("Content-Type", "text/html; charset=utf-8")
self.end_headers()
self.wfile.write(HTML_PAGE.encode("utf-8"))
def _serve_sse(self):
self.send_response(200)
self.send_header("Content-Type", "text/event-stream")
self.send_header("Cache-Control", "no-cache")
self.send_header("Connection", "keep-alive")
self.send_header("Access-Control-Allow-Origin", "*")
self.end_headers()
client_queue = queue.Queue(maxsize=5000)
# Replay event history for late-connecting clients
for past_event in _event_history:
client_queue.put_nowait(past_event)
with _clients_lock:
_clients.append(client_queue)
try:
while not _shutdown:
try:
event = client_queue.get(timeout=1)
line = f"data: {json.dumps(event)}\n\n"
self.wfile.write(line.encode("utf-8"))
self.wfile.flush()
except queue.Empty:
self.wfile.write(b": keepalive\n\n")
self.wfile.flush()
except (BrokenPipeError, ConnectionResetError, OSError):
pass
finally:
with _clients_lock:
if client_queue in _clients:
_clients.remove(client_queue)
def _serve_file_tree(self):
files = []
for root, dirs, filenames in os.walk(_workdir):
# Skip hidden dirs and autodev backups
dirs[:] = [d for d in dirs if not d.startswith(".") and d != "__pycache__"]
for fname in sorted(filenames):
if fname.startswith("."):
continue
full = os.path.join(root, fname)
rel = os.path.relpath(full, _workdir)
try:
size = os.path.getsize(full)
except OSError:
size = 0
files.append({"path": rel, "size": size})
self.send_response(200)
self.send_header("Content-Type", "application/json")
self.end_headers()
self.wfile.write(json.dumps(files).encode("utf-8"))
def _serve_file_content(self, filepath: str):
if not filepath:
self.send_error(400, "Missing path parameter")
return
full = os.path.realpath(os.path.join(_workdir, filepath))
if not full.startswith(os.path.realpath(_workdir)):
self.send_error(403, "Path outside workspace")
return
try:
with open(full, "r") as f:
content = f.read()
self.send_response(200)
self.send_header("Content-Type", "application/json")
self.end_headers()
self.wfile.write(json.dumps({"path": filepath, "content": content}).encode("utf-8"))
except (IOError, UnicodeDecodeError):
self.send_error(404, "File not found or not readable")
def stop_web_server():
global _shutdown
_shutdown = True
if _server:
threading.Thread(target=_server.shutdown, daemon=True).start()
def start_web_server(port: int, workdir: str):
"""Start the web UI server in a background daemon thread."""
global _workdir
_workdir = workdir
class ThreadedHTTPServer(http.server.HTTPServer):
allow_reuse_address = True
daemon_threads = True
def process_request(self, request, client_address):
t = threading.Thread(target=self._handle, args=(request, client_address),
daemon=True)
t.start()
def _handle(self, request, client_address):
try:
self.finish_request(request, client_address)
except Exception:
pass
try:
self.shutdown_request(request)
except Exception:
pass
server = ThreadedHTTPServer(("0.0.0.0", port), WebHandler)
global _server
_server = server
t = threading.Thread(target=server.serve_forever, daemon=True)
t.start()
return server
HTML_PAGE = """
AutoDev โ Live Dashboard
๐ Plan Progress
Waiting for plan...
๐ File Content
Select a file
"""