131 lines
5.0 KiB
Python
131 lines
5.0 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import mimetypes
|
|
import sys
|
|
import threading
|
|
from http import HTTPStatus
|
|
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
|
from pathlib import Path
|
|
from typing import Any, Callable
|
|
from urllib.parse import parse_qs, urlparse
|
|
|
|
|
|
StatusProvider = Callable[[], dict[str, Any]]
|
|
|
|
|
|
def _json_bytes(payload: Any) -> bytes:
|
|
return json.dumps(payload, ensure_ascii=False, indent=2).encode("utf-8")
|
|
|
|
|
|
def make_handler(status_provider: StatusProvider):
|
|
class AgentLocalServiceHandler(BaseHTTPRequestHandler):
|
|
server_version = "SystemTestAgentLocalService/0.1"
|
|
|
|
def log_message(self, fmt: str, *args: Any) -> None:
|
|
sys.stdout.write(
|
|
"%s - - [%s] %s\n"
|
|
% (self.client_address[0], self.log_date_time_string(), fmt % args)
|
|
)
|
|
|
|
def _send_json(self, status: int, payload: Any) -> None:
|
|
body = _json_bytes(payload)
|
|
self.send_response(status)
|
|
self.send_header("Content-Type", "application/json; charset=utf-8")
|
|
self.send_header("Content-Length", str(len(body)))
|
|
self.send_header("Access-Control-Allow-Origin", "*")
|
|
self.send_header("Access-Control-Allow-Headers", "Content-Type, Authorization")
|
|
self.send_header("Access-Control-Allow-Methods", "GET, OPTIONS")
|
|
self.end_headers()
|
|
self.wfile.write(body)
|
|
|
|
def _send_file(self, file_path: Path) -> None:
|
|
body = file_path.read_bytes()
|
|
content_type = mimetypes.guess_type(str(file_path))[0] or "application/octet-stream"
|
|
self.send_response(HTTPStatus.OK)
|
|
self.send_header("Content-Type", content_type)
|
|
self.send_header("Content-Length", str(len(body)))
|
|
self.send_header("Access-Control-Allow-Origin", "*")
|
|
self.end_headers()
|
|
self.wfile.write(body)
|
|
|
|
def _send_screenshot(self, raw_path: str) -> None:
|
|
status = status_provider()
|
|
memory_path = ((status.get("agent_core") or {}).get("memory_path") or "")
|
|
if not memory_path:
|
|
self._send_json(HTTPStatus.NOT_FOUND, {"ok": False, "error": "Memory path is not available"})
|
|
return
|
|
screenshot_root = (Path(memory_path).parent / "screenshots").resolve()
|
|
requested = Path(raw_path).resolve()
|
|
try:
|
|
requested.relative_to(screenshot_root)
|
|
except ValueError:
|
|
self._send_json(HTTPStatus.FORBIDDEN, {"ok": False, "error": "Screenshot path is outside evidence directory"})
|
|
return
|
|
if not requested.exists() or not requested.is_file():
|
|
self._send_json(HTTPStatus.NOT_FOUND, {"ok": False, "error": "Screenshot not found"})
|
|
return
|
|
self._send_file(requested)
|
|
|
|
def do_OPTIONS(self) -> None:
|
|
self._send_json(HTTPStatus.OK, {"ok": True})
|
|
|
|
def do_GET(self) -> None:
|
|
parsed = urlparse(self.path)
|
|
path = parsed.path
|
|
if path in {"/", "/health"}:
|
|
self._send_json(HTTPStatus.OK, status_provider())
|
|
return
|
|
if path == "/agent/tools":
|
|
status = status_provider()
|
|
self._send_json(
|
|
HTTPStatus.OK,
|
|
{
|
|
"ok": True,
|
|
"tools": ((status.get("agent_core") or {}).get("tools") or []),
|
|
},
|
|
)
|
|
return
|
|
if path == "/agent/memory":
|
|
status = status_provider()
|
|
agent_core = status.get("agent_core") or {}
|
|
self._send_json(
|
|
HTTPStatus.OK,
|
|
{
|
|
"ok": True,
|
|
"memory_path": agent_core.get("memory_path"),
|
|
"memory": agent_core.get("memory"),
|
|
},
|
|
)
|
|
return
|
|
if path == "/agent/screenshot":
|
|
query = parse_qs(parsed.query)
|
|
raw_path = (query.get("path") or [""])[0]
|
|
if not raw_path:
|
|
self._send_json(HTTPStatus.BAD_REQUEST, {"ok": False, "error": "Missing path"})
|
|
return
|
|
self._send_screenshot(raw_path)
|
|
return
|
|
self._send_json(
|
|
HTTPStatus.NOT_FOUND,
|
|
{"ok": False, "error": "Route not found"},
|
|
)
|
|
|
|
return AgentLocalServiceHandler
|
|
|
|
|
|
class LocalService:
|
|
def __init__(self, host: str, port: int, status_provider: StatusProvider):
|
|
self.host = host
|
|
self.port = port
|
|
self._server = ThreadingHTTPServer((host, port), make_handler(status_provider))
|
|
self._thread = threading.Thread(target=self._server.serve_forever, daemon=True)
|
|
|
|
def start(self) -> None:
|
|
self._thread.start()
|
|
|
|
def stop(self) -> None:
|
|
self._server.shutdown()
|
|
self._server.server_close()
|
|
self._thread.join(timeout=2)
|