forked from luopingyi/xc_agent_platform_demo
110 lines
3.3 KiB
Python
110 lines
3.3 KiB
Python
"""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', '3600')))
|
|
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()
|