107 lines
3.5 KiB
Python
107 lines
3.5 KiB
Python
"""OpenClaw Cell lifecycle adapter for AIOS V2R1."""
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import asdict, dataclass, field
|
|
import json
|
|
from pathlib import Path
|
|
import subprocess
|
|
from typing import Any, Mapping
|
|
|
|
from .tenant_context import ValidationError, V2R1Error, require_ident
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class CellSpec:
|
|
tenant_id: str
|
|
image: str
|
|
runtime: str = "docker"
|
|
port: int | None = None
|
|
memory: str = "2g"
|
|
cpus: str = "2"
|
|
pids_limit: int = 512
|
|
network: str = "bridge"
|
|
env: tuple[str, ...] = field(default_factory=tuple)
|
|
|
|
def __post_init__(self) -> None:
|
|
require_ident("tenant_id", self.tenant_id, tenant=True)
|
|
if self.runtime not in {"docker", "podman"}:
|
|
raise ValidationError("unsupported runtime")
|
|
if not self.image:
|
|
raise ValidationError("image required")
|
|
|
|
|
|
class OpenClawFleet:
|
|
def __init__(self, binary: str = "openclaw") -> None:
|
|
self.binary = binary
|
|
|
|
def _run(self, args: list[str]) -> dict[str, Any]:
|
|
proc = subprocess.run(
|
|
[self.binary, *args],
|
|
check=False,
|
|
text=True,
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.PIPE,
|
|
)
|
|
if proc.returncode:
|
|
raise V2R1Error(proc.stderr.strip() or "openclaw command failed")
|
|
data = json.loads(proc.stdout or "{}")
|
|
if isinstance(data, dict):
|
|
return data
|
|
return {"result": data}
|
|
|
|
def create(self, spec: CellSpec) -> dict[str, Any]:
|
|
return self._run([
|
|
"fleet", "create", spec.tenant_id,
|
|
"--image", spec.image,
|
|
"--runtime", spec.runtime,
|
|
"--json",
|
|
])
|
|
|
|
def status(self, tenant_id: str) -> dict[str, Any]:
|
|
require_ident("tenant_id", tenant_id, tenant=True)
|
|
return self._run(["fleet", "status", tenant_id, "--json"])
|
|
|
|
def doctor(self, tenant_id: str | None = None) -> dict[str, Any]:
|
|
args = ["fleet", "doctor"]
|
|
if tenant_id:
|
|
require_ident("tenant_id", tenant_id, tenant=True)
|
|
args.append(tenant_id)
|
|
args.append("--json")
|
|
return self._run(args)
|
|
|
|
def backup(self, tenant_id: str, output: str | Path) -> dict[str, Any]:
|
|
return self._run(["fleet", "backup", tenant_id, "--out", str(output), "--json"])
|
|
|
|
def restore(self, tenant_id: str, archive: str | Path) -> dict[str, Any]:
|
|
return self._run(["fleet", "restore", tenant_id, "--from", str(archive), "--json"])
|
|
|
|
def upgrade(self, tenant_id: str, image: str) -> dict[str, Any]:
|
|
return self._run(["fleet", "upgrade", tenant_id, "--image", image, "--json"])
|
|
|
|
|
|
def load_inventory(path: str | Path) -> list[CellSpec]:
|
|
raw = json.loads(Path(path).read_text(encoding="utf-8"))
|
|
if raw.get("schema_version") != "aios.cells.v1":
|
|
raise ValidationError("invalid inventory schema")
|
|
result: list[CellSpec] = []
|
|
seen: set[str] = set()
|
|
for item in raw.get("cells", []):
|
|
spec = CellSpec(
|
|
tenant_id=item["tenant_id"],
|
|
image=item["image"],
|
|
runtime=item.get("runtime", "docker"),
|
|
port=item.get("port"),
|
|
env=tuple(item.get("env", [])),
|
|
)
|
|
if spec.tenant_id in seen:
|
|
raise ValidationError("duplicate tenant")
|
|
seen.add(spec.tenant_id)
|
|
result.append(spec)
|
|
if not result:
|
|
raise ValidationError("no cells")
|
|
return result
|
|
|
|
|
|
def inventory_as_dict(items: list[CellSpec]) -> dict[str, Any]:
|
|
return {"schema_version": "aios.cells.v1", "cells": [asdict(x) for x in items]}
|