"""Stdlib HTTP server for the integration-failure-triage Space. Replaces the old `python -m http.server` + bash refresh loop. Responsibilities: * Serve `output/index.html` and `output/state.json` like the previous setup. * Inject a small banner at the top of the served HTML (right after ``) with a "Refresh now" form button and the last-refresh timestamp. * `POST /refresh` → kicks `python app/refresh.py --no-history` in a subprocess. Returns 202 immediately + `Refresh-Job-Id` header. Only one concurrent refresh — second concurrent POST returns 409. * `GET /status` → JSON describing background-job state. * Background thread: synchronous initial render on startup, then loops every `REFRESH_INTERVAL_SECONDS` (default 21600 = 6h) running the full refresh (with history sweep). No third-party deps — `http.server` only. """ from __future__ import annotations import datetime import io import json import os import subprocess import sys import threading import time import uuid from http import HTTPStatus from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer # --------------------------------------------------------------------------- # Paths and config # --------------------------------------------------------------------------- APP_DIR = os.path.dirname(os.path.abspath(__file__)) REPO_DIR = os.path.dirname(APP_DIR) OUTPUT_DIR = os.environ.get("OUTPUT_DIR", os.path.join(REPO_DIR, "output")) REFRESH_SCRIPT = os.path.join(APP_DIR, "refresh.py") PORT = int(os.environ.get("PORT", "7860")) REFRESH_INTERVAL_SECONDS = int(os.environ.get("REFRESH_INTERVAL_SECONDS", "21600")) BISECT_MAX_BUCKETS = int(os.environ.get("BISECT_MAX_BUCKETS", "1")) # Whether to even expose the /bisect endpoint. The bisect step needs a GPU # upgrade + a local `transformers` clone — on cpu-basic it will exit ~125 (skip) # for every iteration. Enable explicitly once the Space is on suitable hardware. BISECT_ENABLED = os.environ.get("BISECT_ENABLED", "1") not in ("0", "false", "False") # --------------------------------------------------------------------------- # Job state — module-level globals, protected by _job_lock. # --------------------------------------------------------------------------- _job_lock = threading.Lock() _job_state: dict = { "running": False, "job_id": None, "started_at": None, # ISO-8601 UTC, set when a job starts "last_finished_at": None, # ISO-8601 UTC, set when a job ends "last_exit_code": None, # int, set when a job ends "last_job_id": None, "last_log_path": None, } # Parallel state machine for the bisect job (independent single-flight). _bisect_lock = threading.Lock() _bisect_state: dict = { "running": False, "job_id": None, "started_at": None, "last_finished_at": None, "last_exit_code": None, "last_job_id": None, "last_log_path": None, } def _utcnow_iso() -> str: return datetime.datetime.now(datetime.UTC).replace(microsecond=0).isoformat().replace("+00:00", "Z") def _public_status() -> dict: """Snapshot of job state suitable for JSON serialisation. Caller-locked.""" return { "running": _job_state["running"], "started_at": _job_state["started_at"], "last_finished_at": _job_state["last_finished_at"], "last_exit_code": _job_state["last_exit_code"], } def _public_bisect_status() -> dict: """Snapshot of bisect state. Caller holds `_bisect_lock`.""" return { "enabled": BISECT_ENABLED, "running": _bisect_state["running"], "started_at": _bisect_state["started_at"], "last_finished_at": _bisect_state["last_finished_at"], "last_exit_code": _bisect_state["last_exit_code"], } def _run_bisect_subprocess(job_id: str, extra_args: list[str]) -> None: """Run `python -m app.spot_bisect ` from REPO_DIR. -m form is required so the stdlib `bisect` module isn't shadowed by our `app/spot_bisect.py` file (the rename was the workaround). The script writes its cache to `OUTPUT_DIR/data/bisect_cache.json` (bucket-backed in prod). """ log_path = f"/tmp/bisect-{job_id}.log" cmd = [sys.executable, "-m", "app.spot_bisect", *extra_args] with _bisect_lock: _bisect_state["last_log_path"] = log_path try: with open(log_path, "wb") as logf: proc = subprocess.Popen( cmd, cwd=REPO_DIR, stdout=logf, stderr=subprocess.STDOUT, ) exit_code = proc.wait() except Exception as exc: # noqa: BLE001 try: with open(log_path, "ab") as logf: logf.write(f"\n[server.py] launching bisect failed: {exc!r}\n".encode()) except Exception: pass exit_code = -1 with _bisect_lock: _bisect_state["running"] = False _bisect_state["last_finished_at"] = _utcnow_iso() _bisect_state["last_exit_code"] = int(exit_code) _bisect_state["last_job_id"] = job_id _bisect_state["started_at"] = None _bisect_state["job_id"] = None def _try_start_bisect(extra_args: list[str]) -> str | None: """Start a bisect job (`spot_bisect --execute`) if none is in flight. Returns job_id on success, None if a bisect is already running. """ if not BISECT_ENABLED: return None with _bisect_lock: if _bisect_state["running"]: return None job_id = uuid.uuid4().hex[:12] _bisect_state["running"] = True _bisect_state["job_id"] = job_id _bisect_state["started_at"] = _utcnow_iso() t = threading.Thread( target=_run_bisect_subprocess, args=(job_id, extra_args), name=f"bisect-{job_id}", daemon=True, ) t.start() return job_id def _run_refresh_subprocess(job_id: str, extra_args: list[str]) -> None: """Run `python app/refresh.py ` and update state on exit.""" log_path = f"/tmp/refresh-{job_id}.log" cmd = [sys.executable, REFRESH_SCRIPT, *extra_args] with _job_lock: _job_state["last_log_path"] = log_path try: with open(log_path, "wb") as logf: proc = subprocess.Popen( cmd, cwd=REPO_DIR, stdout=logf, stderr=subprocess.STDOUT, ) exit_code = proc.wait() except Exception as exc: # noqa: BLE001 # Log the failure to the same file so users can debug it. try: with open(log_path, "ab") as logf: logf.write(f"\n[server.py] launching refresh failed: {exc!r}\n".encode()) except Exception: pass exit_code = -1 with _job_lock: _job_state["running"] = False _job_state["last_finished_at"] = _utcnow_iso() _job_state["last_exit_code"] = int(exit_code) _job_state["last_job_id"] = job_id _job_state["started_at"] = None _job_state["job_id"] = None def _try_start_refresh(extra_args: list[str]) -> str | None: """Try to start a refresh. Returns job_id on success, None if one is already running.""" with _job_lock: if _job_state["running"]: return None job_id = uuid.uuid4().hex[:12] _job_state["running"] = True _job_state["job_id"] = job_id _job_state["started_at"] = _utcnow_iso() t = threading.Thread( target=_run_refresh_subprocess, args=(job_id, extra_args), name=f"refresh-{job_id}", daemon=True, ) t.start() return job_id # --------------------------------------------------------------------------- # Banner injection # --------------------------------------------------------------------------- def _job_summary(state: dict, lock, label: str) -> str: """Return a one-line HTML span summarizing a job's state (refresh or bisect).""" with lock: running = state["running"] started_at = state["started_at"] last_finished_at = state["last_finished_at"] last_exit_code = state["last_exit_code"] if running: return f'{label}: running… \U0001f504 (started {started_at or "?"})' if last_exit_code is None: return f'{label}: idle' if last_exit_code == 0: return f'{label}: ok @ {last_finished_at}' return f'{label}: exit={last_exit_code} @ {last_finished_at}' def _format_banner() -> str: refresh_summary = _job_summary(_job_state, _job_lock, "Refresh") bisect_summary = _job_summary(_bisect_state, _bisect_lock, "Bisect") if BISECT_ENABLED else "" btn = ( 'style="background:#1f6feb;color:white;border:none;' 'padding:6px 12px;border-radius:5px;cursor:pointer;font-size:13px;margin-left:6px;"' ) bisect_form = "" if BISECT_ENABLED: bisect_form = ( ' \n' ) sep = " · " if bisect_summary else "" return ( '

\n' f' {refresh_summary}{sep}{bisect_summary}\n' ' \n' ' \n' f'{bisect_form}' ' \n' '

' ) def _inject_banner(html: bytes) -> bytes: """Insert the banner right after the first `` tag. String-replace at request time — `render.py` is not aware of the banner. """ banner = _format_banner().encode("utf-8") # Match `` and any `` variant, only first occurrence. needle_simple = b"" idx = html.find(needle_simple) if idx != -1: insert_at = idx + len(needle_simple) return html[:insert_at] + b"\n" + banner + b"\n" + html[insert_at:] # Fallback: try a regex-ish manual scan for ``. lower = html.lower() idx = lower.find(b"", idx) if end != -1: insert_at = end + 1 return html[:insert_at] + b"\n" + banner + b"\n" + html[insert_at:] # No body tag — prepend banner so it's still visible. return banner + b"\n" + html # --------------------------------------------------------------------------- # HTTP handler # --------------------------------------------------------------------------- class TriageHandler(BaseHTTPRequestHandler): server_version = "TriageServer/1.0" # Quieter logs — default BaseHTTPRequestHandler is noisy on stderr. def log_message(self, format: str, *args) -> None: # noqa: A002 sys.stderr.write( "[%s] %s - - %s\n" % (self.log_date_time_string(), self.address_string(), format % args) ) # ----- GET ----- def do_GET(self) -> None: # noqa: N802 path = self.path.split("?", 1)[0] if path == "/" or path == "/index.html": return self._serve_index() if path == "/state.json": return self._serve_static(os.path.join(OUTPUT_DIR, "state.json"), "application/json") if path == "/status": return self._serve_status() if path == "/healthz": return self._send_json(HTTPStatus.OK, {"ok": True}) if path == "/favicon.ico": self.send_response(HTTPStatus.NO_CONTENT) self.end_headers() return # Allow serving any other file from OUTPUT_DIR (e.g. /data/* if present). return self._serve_from_output(path) def _serve_index(self) -> None: html_path = os.path.join(OUTPUT_DIR, "index.html") if not os.path.isfile(html_path): return self._send_text( HTTPStatus.SERVICE_UNAVAILABLE, "Report not generated yet. Try again in a moment.\n", ) with open(html_path, "rb") as f: html = f.read() injected = _inject_banner(html) self.send_response(HTTPStatus.OK) self.send_header("Content-Type", "text/html; charset=utf-8") self.send_header("Content-Length", str(len(injected))) self.send_header("Cache-Control", "no-cache") self.end_headers() self.wfile.write(injected) def _serve_static(self, path: str, content_type: str) -> None: if not os.path.isfile(path): return self._send_text(HTTPStatus.NOT_FOUND, "not found\n") with open(path, "rb") as f: data = f.read() self.send_response(HTTPStatus.OK) self.send_header("Content-Type", content_type) self.send_header("Content-Length", str(len(data))) self.send_header("Cache-Control", "no-cache") self.end_headers() self.wfile.write(data) def _serve_from_output(self, url_path: str) -> None: # Strip leading slash, normalise, prevent traversal. rel = url_path.lstrip("/") abs_path = os.path.realpath(os.path.join(OUTPUT_DIR, rel)) if not abs_path.startswith(os.path.realpath(OUTPUT_DIR) + os.sep) and abs_path != os.path.realpath(OUTPUT_DIR): return self._send_text(HTTPStatus.FORBIDDEN, "forbidden\n") if os.path.isdir(abs_path): return self._send_text(HTTPStatus.NOT_FOUND, "not found\n") if not os.path.isfile(abs_path): return self._send_text(HTTPStatus.NOT_FOUND, "not found\n") # Cheap content-type guess ext = os.path.splitext(abs_path)[1].lower() ctype = { ".html": "text/html; charset=utf-8", ".json": "application/json", ".css": "text/css", ".js": "application/javascript", ".png": "image/png", ".jpg": "image/jpeg", ".svg": "image/svg+xml", ".txt": "text/plain; charset=utf-8", }.get(ext, "application/octet-stream") return self._serve_static(abs_path, ctype) def _serve_status(self) -> None: with _job_lock: refresh_payload = _public_status() with _bisect_lock: bisect_payload = _public_bisect_status() # Keep the old top-level keys (refresh state) for backwards compatibility # with anyone polling /status today; add a `bisect` sub-object. self._send_json(HTTPStatus.OK, {**refresh_payload, "bisect": bisect_payload}) # ----- POST ----- def do_POST(self) -> None: # noqa: N802 path = self.path.split("?", 1)[0] if path == "/refresh": return self._handle_post_job( kind="refresh", starter=lambda: _try_start_refresh(["--no-history"]), state=_job_state, lock=_job_lock, header_name="Refresh-Job-Id", disabled_reason=None, ) if path == "/bisect": if not BISECT_ENABLED: return self._send_json( HTTPStatus.SERVICE_UNAVAILABLE, {"error": "bisect endpoint disabled (set BISECT_ENABLED=1 on a GPU-capable Space)"}, ) return self._handle_post_job( kind="bisect", starter=lambda: _try_start_bisect([ "--execute", "--max-buckets", str(BISECT_MAX_BUCKETS), ]), state=_bisect_state, lock=_bisect_lock, header_name="Bisect-Job-Id", disabled_reason=None, ) return self._send_text(HTTPStatus.NOT_FOUND, "not found\n") def _handle_post_job(self, *, kind: str, starter, state: dict, lock, header_name: str, disabled_reason: str | None) -> None: # Drain any request body (forms send empty body but be polite). length = int(self.headers.get("Content-Length") or "0") if length: try: self.rfile.read(length) except Exception: pass if disabled_reason is not None: return self._send_json(HTTPStatus.SERVICE_UNAVAILABLE, {"error": disabled_reason}) job_id = starter() if job_id is None: with lock: current_id = state["job_id"] body = json.dumps({ "error": f"{kind} already in progress", "running_job_id": current_id, }).encode("utf-8") self.send_response(HTTPStatus.CONFLICT) self.send_header("Content-Type", "application/json") self.send_header("Content-Length", str(len(body))) if current_id: self.send_header(header_name, current_id) self.end_headers() self.wfile.write(body) return body = json.dumps({"job_id": job_id, "status": "accepted"}).encode("utf-8") self.send_response(HTTPStatus.ACCEPTED) self.send_header("Content-Type", "application/json") self.send_header("Content-Length", str(len(body))) self.send_header(header_name, job_id) accept = (self.headers.get("Accept") or "").lower() if "text/html" in accept: self.send_header("Location", "/") self.end_headers() self.wfile.write(body) # ----- helpers ----- def _send_text(self, status: HTTPStatus, text: str) -> None: data = text.encode("utf-8") self.send_response(status) self.send_header("Content-Type", "text/plain; charset=utf-8") self.send_header("Content-Length", str(len(data))) self.end_headers() self.wfile.write(data) def _send_json(self, status: HTTPStatus, payload: dict) -> None: data = json.dumps(payload).encode("utf-8") self.send_response(status) self.send_header("Content-Type", "application/json") self.send_header("Content-Length", str(len(data))) self.send_header("Cache-Control", "no-cache") self.end_headers() self.wfile.write(data) # --------------------------------------------------------------------------- # Background refresh loop (replaces the bash while-loop) # --------------------------------------------------------------------------- def _initial_refresh_sync() -> None: """Run a fast initial refresh in-process so the page is up immediately. Uses `refresh.main([...])` directly rather than a subprocess so that we block until the file is on disk before opening the HTTP port. Errors are swallowed and logged so the server still starts. """ sys.path.insert(0, APP_DIR) try: import refresh # type: ignore[import-not-found] except Exception as exc: # noqa: BLE001 print(f"[server.py] failed to import refresh: {exc!r}", file=sys.stderr, flush=True) return print("[server.py] initial render (--no-history)…", flush=True) t0 = time.time() job_id = uuid.uuid4().hex[:12] log_path = f"/tmp/refresh-{job_id}.log" with _job_lock: _job_state["running"] = True _job_state["job_id"] = job_id _job_state["started_at"] = _utcnow_iso() _job_state["last_log_path"] = log_path exit_code = 0 try: refresh.main(["--no-history", "--out-dir", OUTPUT_DIR]) except SystemExit as e: exit_code = int(e.code) if isinstance(e.code, int) else (0 if not e.code else 1) except Exception as exc: # noqa: BLE001 print(f"[server.py] initial refresh failed: {exc!r}", file=sys.stderr, flush=True) exit_code = 1 elapsed = time.time() - t0 print(f"[server.py] initial render done in {elapsed:.1f}s (exit={exit_code})", flush=True) with _job_lock: _job_state["running"] = False _job_state["last_finished_at"] = _utcnow_iso() _job_state["last_exit_code"] = exit_code _job_state["last_job_id"] = job_id _job_state["started_at"] = None _job_state["job_id"] = None def _scheduled_loop() -> None: """Cron-style loop: every REFRESH_INTERVAL_SECONDS, kick a full refresh.""" while True: time.sleep(REFRESH_INTERVAL_SECONDS) # Use the same single-flight gate; if a user-triggered refresh is in # flight, just skip this tick. job_id = _try_start_refresh(["--history-days", "90"]) if job_id is None: print("[server.py] scheduled refresh skipped — another in flight", flush=True) else: print(f"[server.py] scheduled refresh started job={job_id}", flush=True) # --------------------------------------------------------------------------- # Entrypoint # --------------------------------------------------------------------------- def main() -> None: os.makedirs(OUTPUT_DIR, exist_ok=True) # Only run the initial sync refresh if we don't already have an index.html # on disk (so reloads / local dev don't pay the fetch cost every start). # Toggle off via SKIP_INITIAL_REFRESH=1. skip_initial = os.environ.get("SKIP_INITIAL_REFRESH") == "1" have_index = os.path.isfile(os.path.join(OUTPUT_DIR, "index.html")) if not skip_initial and not have_index: _initial_refresh_sync() elif have_index: print( f"[server.py] reusing existing {OUTPUT_DIR}/index.html " "(set SKIP_INITIAL_REFRESH=0 + delete to force re-render)", flush=True, ) # Start the scheduled-refresh thread. t = threading.Thread(target=_scheduled_loop, name="refresh-scheduler", daemon=True) t.start() httpd = ThreadingHTTPServer(("0.0.0.0", PORT), TriageHandler) print(f"[server.py] serving on http://0.0.0.0:{PORT} (output={OUTPUT_DIR})", flush=True) try: httpd.serve_forever() except KeyboardInterrupt: print("[server.py] shutting down", flush=True) finally: httpd.server_close() if __name__ == "__main__": main()