xc_agent_platform_demo/src/server.py

"""Minimal HTTP server for the adaptation agent (stdlib only).

Listens on PORT (default 8080). Handles SIGTERM and exits the main loop
within ~1s (platform window documented as ≤30s).

Startup: require STRATEGY_ID (Phase1 product plan — fail if missing).
"""

from __future__ import annotations

import os
import signal
import sys
import threading
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from typing import Optional

from .health import handle_get
from .strategy import require_strategy_id
from .gate import live_enabled
from .live_cycle import run_cycle

_shutdown = threading.Event()
_httpd: Optional[ThreadingHTTPServer] = None


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

    def log_message(self, fmt: str, *args) -> None:  # noqa: A003
        # Avoid noisy default logging; never log env secrets.
        sys.stderr.write("[http] request handled (details omitted)\n")

    def do_GET(self) -> None:  # noqa: N802
        code, headers, body = handle_get(self.path)
        self.send_response(code)
        for k, v in headers.items():
            self.send_header(k, v)
        self.send_header("Content-Length", str(len(body)))
        self.end_headers()
        self.wfile.write(body)

    def do_POST(self) -> None:  # noqa: N802
        body = b'{"error":"method_not_allowed"}'
        self.send_response(405)
        self.send_header("Content-Type", "application/json; charset=utf-8")
        self.send_header("Content-Length", str(len(body)))
        self.end_headers()
        self.wfile.write(body)


def _on_sigterm(signum: int, frame) -> None:  # noqa: ARG001
    sys.stderr.write(f"Received signal {signum}; shutting down...\n")
    _shutdown.set()
    httpd = _httpd
    if httpd is not None:
        # wake serve_forever
        threading.Thread(target=httpd.shutdown, daemon=True).start()


def _live_worker():
    from pathlib import Path
    from .ledger import rebuild
    try:
        _, _, _, sizes = rebuild(Path(os.environ.get('LIVE_STATE_FILE', '/tmp/live-submitted.json')))
        sys.stderr.write('[live] ledger rebuilt remote=%d baked=%d persisted=%d\n' % sizes)
    except Exception:
        sys.stderr.write('[live] hold reason=ledger_rebuild_failed\n')
    try:
        interval = max(60, int(os.environ.get('LIVE_CYCLE_SECONDS', '600')))
    except ValueError:
        interval = 600
    while not _shutdown.is_set():
        try:
            run_cycle()
        except Exception:
            sys.stderr.write('[live] hold reason=cycle_error\n')
        if _shutdown.wait(interval):
            break


def main() -> None:
    global _httpd

    require_strategy_id()  # exit 1 if missing

    port = int(os.environ.get("PORT", "8080"))
    host = os.environ.get("HOST", "0.0.0.0")

    signal.signal(signal.SIGTERM, _on_sigterm)
    signal.signal(signal.SIGINT, _on_sigterm)

    httpd = ThreadingHTTPServer((host, port), AgentHandler)
    httpd.daemon_threads = True
    _httpd = httpd

    if live_enabled():
        threading.Thread(target=_live_worker, daemon=True).start()

    sys.stderr.write(f"Listening on {host}:{port} (GET /health, GET /status)\n")
    try:
        httpd.serve_forever(poll_interval=0.5)
    finally:
        httpd.server_close()
        sys.stderr.write("Server stopped.\n")


if __name__ == "__main__":
    main()