Skills/explore/scripts/server.py

#!/usr/bin/env python3
"""explore — localhost server behind the /explore skill.

Serves <workdir>/page.html, records marker-click events into
<workdir>/events.jsonl (one JSON object per line), and pushes an SSE
"reload" event whenever page.html changes on disk. The Claude session
arms a Monitor (`tail -F -n 0 <workdir>/events.jsonl`) as the
back-channel; rewriting page.html is the reply channel.

Usage: server.py <workdir> [--port N]

Stdlib only. Binds to localhost.
"""
import argparse
import json
import time
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path

# The single place the default port is configured (--port overrides per run).
PORT = 8000

# SSE tuning: how often to poll the page mtime, and how many idle polls
# between keep-alive comments (0.5s * 30 = one ping every 15s).
POLL_S = 0.5
PINGS_EVERY = 30

# Set in main() from the workdir argument.
PAGE = None
EVENTS = None


class Handler(BaseHTTPRequestHandler):
    protocol_version = "HTTP/1.1"

    def do_GET(self):
        if self.path in ("/", "/index.html"):
            self._serve_page()
        elif self.path == "/stream":
            self._serve_stream()
        else:
            self.send_error(404)

    def do_POST(self):
        if self.path == "/event":
            self._record_event()
        else:
            self.send_error(404)

    def _serve_page(self):
        # Read fresh from disk every time — Claude rewrites this file.
        try:
            body = PAGE.read_bytes()
        except OSError:
            self.send_error(503, "page.html missing")
            return
        self.send_response(200)
        self.send_header("Content-Type", "text/html; charset=utf-8")
        self.send_header("Cache-Control", "no-store")
        self.send_header("Content-Length", str(len(body)))
        self.end_headers()
        self.wfile.write(body)

    def _serve_stream(self):
        # Long-lived SSE connection: emit "reload" whenever page.html
        # changes on disk, keep-alive comments in between.
        self.send_response(200)
        self.send_header("Content-Type", "text/event-stream")
        self.send_header("Cache-Control", "no-store")
        self.end_headers()
        try:
            last = PAGE.stat().st_mtime_ns
        except OSError:
            last = 0
        idle = 0
        try:
            while True:
                time.sleep(POLL_S)
                try:
                    cur = PAGE.stat().st_mtime_ns
                except OSError:
                    continue  # page briefly missing mid-rewrite
                if cur != last:
                    last = cur
                    self.wfile.write(b"event: reload\ndata: 1\n\n")
                    self.wfile.flush()
                    idle = 0
                else:
                    idle += 1
                    if idle >= PINGS_EVERY:
                        self.wfile.write(b": ping\n\n")
                        self.wfile.flush()
                        idle = 0
        except (BrokenPipeError, ConnectionResetError):
            return  # client went away — normal

    def _record_event(self):
        try:
            length = int(self.headers.get("Content-Length", 0))
        except ValueError:
            self.send_error(411)
            return
        raw = self.rfile.read(length)
        try:
            event = json.loads(raw)
            if not isinstance(event, dict):
                raise ValueError("expected a JSON object")
        except (ValueError, UnicodeDecodeError):
            self.send_error(400, "invalid JSON")
            return
        event["ts"] = time.strftime("%Y-%m-%dT%H:%M:%S")
        # Single-line appends in "a" mode are atomic enough here.
        with EVENTS.open("a", encoding="utf-8") as f:
            f.write(json.dumps(event, ensure_ascii=False) + "\n")
        self.send_response(204)
        self.end_headers()

    def log_message(self, *args):
        pass  # keep stdout clean — it is the background-task log


def main():
    global PAGE, EVENTS
    ap = argparse.ArgumentParser(description=__doc__)
    ap.add_argument("workdir", type=Path, help="directory holding page.html and events.jsonl")
    ap.add_argument("--port", type=int, default=PORT, help=f"listen port (default {PORT})")
    a = ap.parse_args()
    workdir = a.workdir.resolve()
    workdir.mkdir(parents=True, exist_ok=True)
    PAGE = workdir / "page.html"
    EVENTS = workdir / "events.jsonl"
    EVENTS.touch()
    srv = ThreadingHTTPServer(("127.0.0.1", a.port), Handler)
    print(f"explore serving {workdir} on http://127.0.0.1:{a.port}", flush=True)
    srv.serve_forever()


if __name__ == "__main__":
    main()