feat(candidate): add production application entrypoint
This commit is contained in:
Codex
@@ -0,0 +1,349 @@
|
||||
"""Production-candidate AIOS V2 application entrypoint.
|
||||
|
||||
This process owns the public control-plane API and read-only web console. It
|
||||
loads durable control-plane state, dispatches approved work through runtime
|
||||
adapters, exposes tenant-scoped execution evidence, and never exposes runtime
|
||||
credentials to the browser.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import dataclasses
|
||||
import json
|
||||
import os
|
||||
import stat
|
||||
import urllib.parse
|
||||
from enum import Enum
|
||||
from http import HTTPStatus
|
||||
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
||||
from pathlib import Path
|
||||
from typing import Any, Mapping
|
||||
|
||||
from aios_hermes import HermesAdapter, HermesJobState, HermesResult
|
||||
from aios_openclaw import HTTPRuntimeTransport, OpenClawAdapter
|
||||
from aios_openclaw.callback import AuditCallbackSink, EventPublisher
|
||||
|
||||
from .console import CONSOLE_HTML
|
||||
from .control_plane import AuditEvent, RequestContext
|
||||
from .execution_tracking import ExecutionTracker
|
||||
from .gateway import GatewayApplication, error_response
|
||||
from .orchestrator import ExecutionOrchestrator
|
||||
from .persistence import DurableControlPlaneService, SQLiteSnapshotRepository
|
||||
|
||||
|
||||
def _json_value(value: Any) -> Any:
|
||||
if dataclasses.is_dataclass(value):
|
||||
return {
|
||||
item.name: _json_value(getattr(value, item.name))
|
||||
for item in dataclasses.fields(value)
|
||||
}
|
||||
if isinstance(value, Enum):
|
||||
return value.value
|
||||
if isinstance(value, Mapping):
|
||||
return {str(key): _json_value(item) for key, item in value.items()}
|
||||
if isinstance(value, (list, tuple, set, frozenset)):
|
||||
return [_json_value(item) for item in value]
|
||||
return value
|
||||
|
||||
|
||||
class FileSigningMaterialResolver:
|
||||
"""Resolve runtime signing material from a root-owned file reference."""
|
||||
|
||||
def __init__(self, path: str | Path) -> None:
|
||||
self.path = Path(path).resolve()
|
||||
|
||||
def __call__(self, tenant_id: str, key_id: str) -> bytes:
|
||||
del tenant_id, key_id
|
||||
metadata = self.path.stat()
|
||||
if not stat.S_ISREG(metadata.st_mode):
|
||||
raise RuntimeError("signing material reference is not a regular file")
|
||||
if metadata.st_mode & 0o077:
|
||||
raise RuntimeError("signing material file permissions must be 0600 or stricter")
|
||||
value = self.path.read_bytes().strip()
|
||||
if len(value) < 32:
|
||||
raise RuntimeError("signing material is too short")
|
||||
return value
|
||||
|
||||
|
||||
class ControlPlaneEventPublisher(EventPublisher):
|
||||
"""Append OpenClaw runtime events to the tenant audit stream."""
|
||||
|
||||
def __init__(self, service: DurableControlPlaneService) -> None:
|
||||
self.service = service
|
||||
|
||||
def publish(self, event: Mapping[str, Any]) -> None:
|
||||
tenant_id = str(event.get("tenant_id") or "").strip()
|
||||
trace_id = str(event.get("trace_id") or "").strip()
|
||||
command_id = str(event.get("command_id") or "").strip()
|
||||
event_type = str(event.get("event_type") or "OPENCLAW_EVENT").strip()
|
||||
if not tenant_id or not trace_id or not command_id:
|
||||
raise ValueError("runtime audit event is missing tenant identity")
|
||||
state_value = str(event.get("state") or "unknown").lower()
|
||||
outcome = "allowed" if state_value == "completed" else state_value
|
||||
audit = AuditEvent(
|
||||
event_id=f"runtime_{command_id}_{len(self.service.store.audit_events) + 1}",
|
||||
tenant_id=tenant_id,
|
||||
principal_id="openclaw-runtime",
|
||||
trace_id=trace_id,
|
||||
action=event_type,
|
||||
resource_type="openclaw_command",
|
||||
resource_id=command_id,
|
||||
outcome=outcome,
|
||||
policy_version="runtime-v1",
|
||||
details=dict(event.get("details") or {}),
|
||||
)
|
||||
with self.service.store.lock:
|
||||
self.service.store.audit_events.append(audit)
|
||||
self.service.repository.save(self.service._snapshot())
|
||||
|
||||
|
||||
class CandidateApplication:
|
||||
"""Compose durable control plane, runtimes, evidence and console routes."""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
gateway: GatewayApplication,
|
||||
orchestrator: ExecutionOrchestrator,
|
||||
tracker: ExecutionTracker,
|
||||
hermes: HermesAdapter,
|
||||
openclaw: OpenClawAdapter | None,
|
||||
) -> None:
|
||||
self.gateway = gateway
|
||||
self.orchestrator = orchestrator
|
||||
self.tracker = tracker
|
||||
self.hermes = hermes
|
||||
self.openclaw = openclaw
|
||||
|
||||
def context(self, headers: Mapping[str, str]) -> RequestContext:
|
||||
return self.gateway.context_from_headers(headers)
|
||||
|
||||
def handle_json(
|
||||
self,
|
||||
method: str,
|
||||
raw_path: str,
|
||||
headers: Mapping[str, str],
|
||||
body: Any,
|
||||
) -> tuple[int, Mapping[str, Any]]:
|
||||
parsed = urllib.parse.urlparse(raw_path)
|
||||
path = parsed.path.rstrip("/") or "/"
|
||||
method = method.upper()
|
||||
|
||||
if method == "GET" and path == "/api/v2/runtime":
|
||||
context = self.context(headers)
|
||||
runtimes: list[dict[str, Any]] = []
|
||||
if self.openclaw is None:
|
||||
runtimes.append(
|
||||
{
|
||||
"name": "openclaw",
|
||||
"status": "disabled",
|
||||
"details": {"reason": "runtime_not_configured"},
|
||||
}
|
||||
)
|
||||
else:
|
||||
try:
|
||||
health = self.openclaw.health()
|
||||
runtimes.append(
|
||||
{
|
||||
"name": "openclaw",
|
||||
"status": health.status,
|
||||
"details": _json_value(health),
|
||||
}
|
||||
)
|
||||
except Exception as exc:
|
||||
runtimes.append(
|
||||
{
|
||||
"name": "openclaw",
|
||||
"status": "unavailable",
|
||||
"details": {"message": str(exc)},
|
||||
}
|
||||
)
|
||||
hermes_health = self.hermes.health()
|
||||
runtimes.append(
|
||||
{
|
||||
"name": "hermes",
|
||||
"status": hermes_health.status,
|
||||
"details": _json_value(hermes_health),
|
||||
}
|
||||
)
|
||||
return HTTPStatus.OK, {
|
||||
"tenant_id": context.tenant_id,
|
||||
"runtimes": runtimes,
|
||||
}
|
||||
|
||||
if method == "GET" and path == "/api/v2/executions":
|
||||
context = self.context(headers)
|
||||
return HTTPStatus.OK, {
|
||||
"executions": _json_value(
|
||||
self.tracker.list(tenant_id=context.tenant_id)
|
||||
)
|
||||
}
|
||||
|
||||
dispatch_match = __import__("re").fullmatch(
|
||||
r"/api/v2/tasks/([^/]+)/dispatch",
|
||||
path,
|
||||
)
|
||||
if method == "POST" and dispatch_match:
|
||||
context = self.context(headers)
|
||||
task_id = urllib.parse.unquote(dispatch_match.group(1))
|
||||
payload = body if isinstance(body, Mapping) else {}
|
||||
timeout = payload.get("timeout_seconds")
|
||||
record = self.orchestrator.dispatch(
|
||||
context,
|
||||
task_id=task_id,
|
||||
timeout_seconds=float(timeout) if timeout is not None else None,
|
||||
)
|
||||
return HTTPStatus.ACCEPTED, {"dispatch": _json_value(record)}
|
||||
|
||||
if method == "POST" and path == "/api/v2/hermes/results/reconcile":
|
||||
context = self.context(headers)
|
||||
if not isinstance(body, Mapping):
|
||||
raise ValueError("JSON body must be an object")
|
||||
result = HermesResult(
|
||||
job_id=str(body["job_id"]),
|
||||
tenant_id=str(body["tenant_id"]),
|
||||
task_id=str(body["task_id"]),
|
||||
trace_id=str(body["trace_id"]),
|
||||
state=HermesJobState(str(body["state"])),
|
||||
worker_id=(
|
||||
str(body["worker_id"])
|
||||
if body.get("worker_id") is not None
|
||||
else None
|
||||
),
|
||||
output=body.get("output"),
|
||||
error=body.get("error"),
|
||||
started_at=body.get("started_at"),
|
||||
finished_at=body.get("finished_at"),
|
||||
)
|
||||
task = self.orchestrator.reconcile_hermes(context, result)
|
||||
return HTTPStatus.OK, {"task": _json_value(task)}
|
||||
|
||||
return self.gateway.handle(method, raw_path, headers, body)
|
||||
|
||||
|
||||
class CandidateRequestHandler(BaseHTTPRequestHandler):
|
||||
server_version = "AIOSV2Candidate/1.0"
|
||||
max_body_bytes = 1_048_576
|
||||
|
||||
@property
|
||||
def application(self) -> CandidateApplication:
|
||||
return self.server.application # type: ignore[attr-defined]
|
||||
|
||||
def _read_body(self) -> Any:
|
||||
length = int(self.headers.get("Content-Length", "0") or "0")
|
||||
if length < 0 or length > self.max_body_bytes:
|
||||
raise ValueError("request body is too large")
|
||||
if length == 0:
|
||||
return {}
|
||||
content_type = self.headers.get("Content-Type", "").split(";", 1)[0].strip()
|
||||
if content_type != "application/json":
|
||||
raise ValueError("Content-Type must be application/json")
|
||||
return json.loads(self.rfile.read(length).decode("utf-8"))
|
||||
|
||||
def _security_headers(self) -> None:
|
||||
self.send_header("Cache-Control", "no-store")
|
||||
self.send_header("X-Content-Type-Options", "nosniff")
|
||||
self.send_header("X-Frame-Options", "DENY")
|
||||
self.send_header("Referrer-Policy", "no-referrer")
|
||||
self.send_header(
|
||||
"Content-Security-Policy",
|
||||
"default-src 'self'; style-src 'unsafe-inline'; script-src 'unsafe-inline'; connect-src 'self'",
|
||||
)
|
||||
|
||||
def _send_json(self, status_code: int, payload: Mapping[str, Any]) -> None:
|
||||
encoded = json.dumps(
|
||||
_json_value(payload),
|
||||
ensure_ascii=False,
|
||||
separators=(",", ":"),
|
||||
).encode("utf-8")
|
||||
self.send_response(int(status_code))
|
||||
self.send_header("Content-Type", "application/json; charset=utf-8")
|
||||
self.send_header("Content-Length", str(len(encoded)))
|
||||
self._security_headers()
|
||||
self.end_headers()
|
||||
self.wfile.write(encoded)
|
||||
|
||||
def _send_console(self) -> None:
|
||||
encoded = CONSOLE_HTML.encode("utf-8")
|
||||
self.send_response(HTTPStatus.OK)
|
||||
self.send_header("Content-Type", "text/html; charset=utf-8")
|
||||
self.send_header("Content-Length", str(len(encoded)))
|
||||
self._security_headers()
|
||||
self.end_headers()
|
||||
self.wfile.write(encoded)
|
||||
|
||||
def _dispatch(self, method: str) -> None:
|
||||
path = urllib.parse.urlparse(self.path).path.rstrip("/") or "/"
|
||||
if method == "GET" and path in {"/", "/console"}:
|
||||
self._send_console()
|
||||
return
|
||||
try:
|
||||
body = self._read_body() if method == "POST" else {}
|
||||
status_code, payload = self.application.handle_json(
|
||||
method,
|
||||
self.path,
|
||||
{key: value for key, value in self.headers.items()},
|
||||
body,
|
||||
)
|
||||
except BaseException as exc:
|
||||
status_code, payload = error_response(exc)
|
||||
self._send_json(status_code, payload)
|
||||
|
||||
def do_GET(self) -> None: # noqa: N802
|
||||
self._dispatch("GET")
|
||||
|
||||
def do_POST(self) -> None: # noqa: N802
|
||||
self._dispatch("POST")
|
||||
|
||||
|
||||
class CandidateHTTPServer(ThreadingHTTPServer):
|
||||
daemon_threads = True
|
||||
allow_reuse_address = True
|
||||
|
||||
def __init__(self, address: tuple[str, int], application: CandidateApplication) -> None:
|
||||
self.application = application
|
||||
super().__init__(address, CandidateRequestHandler)
|
||||
|
||||
|
||||
def build_application() -> CandidateApplication:
|
||||
state_dir = Path(os.getenv("AIOS_V2_STATE_DIR", "/var/lib/aios-v2")).resolve()
|
||||
state_dir.mkdir(parents=True, exist_ok=True)
|
||||
repository = SQLiteSnapshotRepository(state_dir / "control-plane.sqlite3")
|
||||
service = DurableControlPlaneService(repository)
|
||||
tracker = ExecutionTracker(state_dir / "executions")
|
||||
hermes = HermesAdapter()
|
||||
|
||||
openclaw_url = os.getenv("AIOS_OPENCLAW_URL", "").strip()
|
||||
material_file = os.getenv("AIOS_OPENCLAW_SIGNING_MATERIAL_FILE", "").strip()
|
||||
openclaw = None
|
||||
resolver = None
|
||||
if openclaw_url and material_file:
|
||||
resolver = FileSigningMaterialResolver(material_file)
|
||||
transport = HTTPRuntimeTransport(
|
||||
openclaw_url,
|
||||
credential_ref=os.getenv("AIOS_OPENCLAW_CREDENTIAL_REF") or None,
|
||||
)
|
||||
openclaw = OpenClawAdapter(
|
||||
transport,
|
||||
resolver,
|
||||
callback_sink=AuditCallbackSink(ControlPlaneEventPublisher(service)),
|
||||
)
|
||||
|
||||
gateway = GatewayApplication(service)
|
||||
orchestrator = ExecutionOrchestrator(
|
||||
service,
|
||||
openclaw=openclaw,
|
||||
hermes=hermes,
|
||||
openclaw_signing_material_resolver=resolver,
|
||||
)
|
||||
return CandidateApplication(gateway, orchestrator, tracker, hermes, openclaw)
|
||||
|
||||
|
||||
def main() -> None:
|
||||
host = os.getenv("AIOS_V2_HOST", "127.0.0.1")
|
||||
port = int(os.getenv("AIOS_V2_PORT", "18920"))
|
||||
server = CandidateHTTPServer((host, port), build_application())
|
||||
print(f"AIOS V2 candidate listening on {host}:{port}")
|
||||
server.serve_forever()
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user