| """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 `<body>`) |
| 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 |
|
|
| |
| |
| |
|
|
| 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")) |
| |
| |
| |
| BISECT_ENABLED = os.environ.get("BISECT_ENABLED", "1") not in ("0", "false", "False") |
|
|
|
|
| |
| |
| |
|
|
| _job_lock = threading.Lock() |
| _job_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, |
| } |
|
|
| |
| _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 <extra_args>` 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: |
| 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 <extra_args>` 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: |
| |
| 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 |
|
|
|
|
| |
| |
| |
|
|
|
|
| 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'<strong>{label}:</strong> running… \U0001f504 (started {started_at or "?"})' |
| if last_exit_code is None: |
| return f'<strong>{label}:</strong> idle' |
| if last_exit_code == 0: |
| return f'<strong>{label}:</strong> ok @ {last_finished_at}' |
| return f'<strong>{label}:</strong> 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 = ( |
| ' <form method="POST" action="/bisect" style="margin:0;display:inline-block;"' |
| f' title="Run spot_bisect --execute --max-buckets {BISECT_MAX_BUCKETS}. Requires GPU + transformers clone in container.">\n' |
| f' <button type="submit" {btn}>Bisect now</button>\n' |
| ' </form>\n' |
| ) |
|
|
| sep = " · " if bisect_summary else "" |
| return ( |
| '<div style="background:#fff8c5;border:1px solid #d4a72c;padding:8px 12px;' |
| 'margin-bottom:16px;border-radius:6px;font-size:13px;display:flex;' |
| 'justify-content:space-between;align-items:center;gap:12px;flex-wrap:wrap;">\n' |
| f' <span>{refresh_summary}{sep}{bisect_summary}</span>\n' |
| ' <span style="white-space:nowrap;">\n' |
| ' <form method="POST" action="/refresh" style="margin:0;display:inline-block;">\n' |
| f' <button type="submit" {btn}>Refresh now</button>\n' |
| ' </form>\n' |
| f'{bisect_form}' |
| ' </span>\n' |
| '</div>' |
| ) |
|
|
|
|
| def _inject_banner(html: bytes) -> bytes: |
| """Insert the banner right after the first `<body>` tag. |
| |
| String-replace at request time — `render.py` is not aware of the banner. |
| """ |
| banner = _format_banner().encode("utf-8") |
| |
| needle_simple = b"<body>" |
| 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:] |
| |
| lower = html.lower() |
| idx = lower.find(b"<body") |
| if idx != -1: |
| end = lower.find(b">", idx) |
| if end != -1: |
| insert_at = end + 1 |
| return html[:insert_at] + b"\n" + banner + b"\n" + html[insert_at:] |
| |
| return banner + b"\n" + html |
|
|
|
|
| |
| |
| |
|
|
|
|
| class TriageHandler(BaseHTTPRequestHandler): |
| server_version = "TriageServer/1.0" |
|
|
| |
| def log_message(self, format: str, *args) -> None: |
| sys.stderr.write( |
| "[%s] %s - - %s\n" |
| % (self.log_date_time_string(), self.address_string(), format % args) |
| ) |
|
|
| |
|
|
| def do_GET(self) -> None: |
| 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 |
| |
| 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: |
| |
| 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") |
| |
| 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() |
| |
| |
| self._send_json(HTTPStatus.OK, {**refresh_payload, "bisect": bisect_payload}) |
|
|
| |
|
|
| def do_POST(self) -> None: |
| 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: |
| |
| 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) |
|
|
| |
|
|
| 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) |
|
|
|
|
| |
| |
| |
|
|
|
|
| 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 |
| except Exception as exc: |
| 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: |
| 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) |
| |
| |
| 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) |
|
|
|
|
| |
| |
| |
|
|
|
|
| def main() -> None: |
| os.makedirs(OUTPUT_DIR, exist_ok=True) |
|
|
| |
| |
| |
| 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, |
| ) |
|
|
| |
| 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() |
|
|
