diff --git a/active-runtime/app/aios_v2/app.py b/active-runtime/app/aios_v2/app.py new file mode 100644 index 0000000..dddf423 --- /dev/null +++ b/active-runtime/app/aios_v2/app.py @@ -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()