#!/usr/bin/env python3 """ayf-server -- AY Framework HTTP feedback server. Bridges the HTML review packs (see /review-pack, task #51) with the agent swarm: a human opens a review pack in the browser, reads it, and submits a verdict that lands back on disk where the agents can read it. Stdlib only -- no pip dependencies. Runs anywhere Python 3.7+ is available. Routes ------ GET / Health check (JSON). GET /review-pack/ Serve the HTML review pack for task N from .ay/reviews/task--review.html. POST /verdict Accept JSON {task, verdict, notes} where verdict is 'pass' or 'needs-work'; write .ay/retros/verdict-.json. GET /status JSON list of every task that has a review pack and/or a recorded verdict, with its verdict status. Config ------ AYF_PORT Port to listen on (default 8742). --port overrides. AYF_REVIEWS_DIR Review-pack directory (default .ay/reviews). AYF_RETROS_DIR Verdict output directory (default .ay/retros). """ import argparse import datetime import json import os import re import sys from http.server import BaseHTTPRequestHandler, HTTPServer from pathlib import Path DEFAULT_PORT = 8742 VALID_VERDICTS = ("pass", "needs-work") # task--review.html / verdict-.json REVIEW_RE = re.compile(r"^task-(\d+)-review\.html$") VERDICT_RE = re.compile(r"^verdict-(\d+)\.json$") CORS_HEADERS = { "Access-Control-Allow-Origin": "*", "Access-Control-Allow-Methods": "GET, POST, OPTIONS", "Access-Control-Allow-Headers": "Content-Type", } def _utc_now_iso(): return datetime.datetime.now(datetime.timezone.utc).replace(microsecond=0).isoformat() def make_handler(reviews_dir: Path, retros_dir: Path): class AYFHandler(BaseHTTPRequestHandler): server_version = "ayf-server/1.0" # --- infra ------------------------------------------------------- def log_message(self, fmt, *args): sys.stderr.write(f"[ayf-server] {self.address_string()} - {fmt % args}\n") def _set_headers(self, code, content_type): self.send_response(code) self.send_header("Content-Type", content_type) for key, val in CORS_HEADERS.items(): self.send_header(key, val) self.end_headers() def _send_json(self, code, data): body = json.dumps(data, indent=2).encode("utf-8") self._set_headers(code, "application/json") self.wfile.write(body) def _send_html(self, code, body: bytes): self._set_headers(code, "text/html; charset=utf-8") self.wfile.write(body) def _read_body(self) -> bytes: try: length = int(self.headers.get("Content-Length", 0)) except (TypeError, ValueError): length = 0 return self.rfile.read(length) if length > 0 else b"" # --- routing ----------------------------------------------------- def do_OPTIONS(self): self._set_headers(204, "text/plain") def do_GET(self): path = self.path.split("?", 1)[0].rstrip("/") or "/" if path == "/": self._send_json(200, { "server": "ayf-server", "status": "ok", "reviews_dir": str(reviews_dir), "retros_dir": str(retros_dir), }) return if path == "/status": self._send_json(200, self._collect_status()) return if path.startswith("/review-pack/"): task_n = self._task_number(path[len("/review-pack/"):]) if task_n is None: self._send_json(400, {"error": "task must be a positive integer"}) return html_file = reviews_dir / f"task-{task_n}-review.html" if not html_file.is_file(): self._send_json(404, { "error": "review pack not found", "task": task_n, "expected": str(html_file), }) return self._send_html(200, html_file.read_bytes()) return self._send_json(404, {"error": "not found", "path": self.path}) def do_POST(self): path = self.path.split("?", 1)[0].rstrip("/") or "/" if path == "/verdict": self._handle_verdict() return self._send_json(404, {"error": "not found", "path": self.path}) # --- handlers ---------------------------------------------------- def _handle_verdict(self): raw = self._read_body() try: payload = json.loads(raw) if raw else {} except json.JSONDecodeError as e: self._send_json(400, {"error": f"invalid JSON: {e}"}) return if not isinstance(payload, dict): self._send_json(400, {"error": "body must be a JSON object"}) return task_n = self._task_number(payload.get("task")) if task_n is None: self._send_json(400, {"error": "'task' must be a positive integer"}) return verdict = payload.get("verdict") if verdict not in VALID_VERDICTS: self._send_json(400, { "error": "'verdict' must be one of " + " | ".join(VALID_VERDICTS), "got": verdict, }) return notes = payload.get("notes", "") if not isinstance(notes, str): self._send_json(400, {"error": "'notes' must be a string"}) return record = { "task": task_n, "verdict": verdict, "notes": notes, "received_at": _utc_now_iso(), } retros_dir.mkdir(parents=True, exist_ok=True) out_path = retros_dir / f"verdict-{task_n}.json" out_path.write_text(json.dumps(record, indent=2) + "\n", encoding="utf-8") self._send_json(200, {"saved": True, "task": task_n, "path": str(out_path)}) def _collect_status(self): tasks = {} # n -> {task, review_pack, verdict, notes} if reviews_dir.is_dir(): for f in reviews_dir.iterdir(): m = REVIEW_RE.match(f.name) if m: n = int(m.group(1)) tasks.setdefault(n, self._blank(n))["review_pack"] = True if retros_dir.is_dir(): for f in retros_dir.iterdir(): m = VERDICT_RE.match(f.name) if not m: continue n = int(m.group(1)) entry = tasks.setdefault(n, self._blank(n)) try: data = json.loads(f.read_text(encoding="utf-8")) entry["verdict"] = data.get("verdict") entry["notes"] = data.get("notes", "") entry["received_at"] = data.get("received_at") except (json.JSONDecodeError, OSError): entry["verdict"] = "unreadable" return {"tasks": [tasks[n] for n in sorted(tasks)]} @staticmethod def _blank(n): return {"task": n, "review_pack": False, "verdict": None, "notes": ""} @staticmethod def _task_number(raw): """Coerce a path segment or JSON value into a positive int, or None.""" if isinstance(raw, bool): return None if isinstance(raw, int): return raw if raw > 0 else None if isinstance(raw, str): s = raw.strip() if s.isdigit() and int(s) > 0: return int(s) return None return AYFHandler def resolve_port(cli_port): if cli_port is not None: return cli_port env_port = os.environ.get("AYF_PORT") if env_port: try: return int(env_port) except ValueError: sys.stderr.write(f"[ayf-server] invalid AYF_PORT={env_port!r}, using {DEFAULT_PORT}\n") return DEFAULT_PORT def main(argv=None): parser = argparse.ArgumentParser( description="ayf-server -- AY Framework stdlib HTTP feedback server", ) parser.add_argument("--port", type=int, default=None, help=f"Port to listen on (default: $AYF_PORT or {DEFAULT_PORT})") parser.add_argument("--reviews-dir", default=os.environ.get("AYF_REVIEWS_DIR", ".ay/reviews"), help="Directory holding task--review.html packs") parser.add_argument("--retros-dir", default=os.environ.get("AYF_RETROS_DIR", ".ay/retros"), help="Directory to write verdict-.json into") args = parser.parse_args(argv) port = resolve_port(args.port) reviews_dir = Path(args.reviews_dir) retros_dir = Path(args.retros_dir) host = "127.0.0.1" handler_class = make_handler(reviews_dir, retros_dir) server = HTTPServer((host, port), handler_class) url = f"http://{host}:{port}" sys.stderr.write(f"[ayf-server] AY Framework feedback server\n") sys.stderr.write(f"[ayf-server] Listening on {url}\n") sys.stderr.write(f"[ayf-server] review packs : {url}/review-pack/\n") sys.stderr.write(f"[ayf-server] submit verdict: POST {url}/verdict\n") sys.stderr.write(f"[ayf-server] task status : {url}/status\n") sys.stderr.write(f"[ayf-server] reviews-dir={reviews_dir} retros-dir={retros_dir}\n") sys.stderr.write("[ayf-server] Press Ctrl+C to stop.\n") sys.stderr.flush() try: server.serve_forever() except KeyboardInterrupt: sys.stderr.write("\n[ayf-server] Shutting down.\n") finally: server.server_close() if __name__ == "__main__": main()