837 lines
42 KiB
Python
Executable File
837 lines
42 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""Root-side executor for the single Athena Operator MCP.
|
|
|
|
The daemon exposes structured platform operations over a local Unix socket.
|
|
It offers both structured, confirmation-bound platform operations and one
|
|
bounded general terminal escape hatch. The latter keeps the platform useful
|
|
for unforeseen work while a small denylist protects remote reachability.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import difflib
|
|
import hashlib
|
|
import json
|
|
import os
|
|
import re
|
|
import shlex
|
|
import shutil
|
|
import socketserver
|
|
import stat
|
|
import subprocess
|
|
import tempfile
|
|
import threading
|
|
import time
|
|
import urllib.parse
|
|
import urllib.request
|
|
import uuid
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
|
|
VERSION = "3.1.0"
|
|
STACK = Path(os.environ.get("ATHENA_OPERATOR_STACK", "/opt/mike-ai/stack")).resolve()
|
|
REPOSITORY = Path(os.environ.get("ATHENA_OPERATOR_REPOSITORY", str(STACK))).resolve()
|
|
STATE = Path(os.environ.get("ATHENA_OPERATOR_STATE", "/data/mike-ai-operator/state")).resolve()
|
|
SOCKET = Path(os.environ.get("ATHENA_OPERATOR_SOCKET", "/run/mike-ai-operator/operator.sock"))
|
|
MODELS = Path(os.environ.get("ATHENA_OPERATOR_MODELS", "/data/models")).resolve()
|
|
TICKET_TTL = 1800
|
|
MAX_FILE_BYTES = 1_000_000
|
|
MAX_FILES = 24
|
|
MAX_OUTPUT = 30_000
|
|
LOCK = threading.RLock()
|
|
STAGING_ROOTS = tuple(
|
|
Path(value).resolve()
|
|
for value in os.environ.get(
|
|
"ATHENA_OPERATOR_STAGING_ROOTS",
|
|
"/data/mike-ai-operator/staging",
|
|
).split(":")
|
|
if value
|
|
)
|
|
|
|
SAFE_PATH = re.compile(r"^[A-Za-z0-9_.+/-]{1,240}$")
|
|
SAFE_NAME = re.compile(r"^[A-Za-z0-9_.-]{1,100}$")
|
|
SAFE_COMMIT = re.compile(r"^[A-Za-z0-9ÄÖÜäöüß _.,:;()+/\-]{5,120}$")
|
|
BLOCKED_PATH_PARTS = {".git", ".ssh", "secrets", "credentials", "authorized_keys"}
|
|
PROTECTED_CONTAINERS = {
|
|
"mike-ai-wireguard-gateway",
|
|
}
|
|
ALLOWED_OPERATIONS = {
|
|
"file_update", "patch_update", "mcp_release", "run_checks", "compose_deploy", "container_action",
|
|
"openwebui_sync", "git_publish", "model_download", "benchmark", "backup",
|
|
}
|
|
|
|
HUNK_HEADER = re.compile(r"^@@ -(\d+)(?:,(\d+))? \+(\d+)(?:,(\d+))? @@")
|
|
ALLOWED_CHECKS = {
|
|
"operator-tests": ["python3", "dev/test_athena_operator.py"],
|
|
"openwebui-filter-tests": ["python3", "dev/test_openwebui_filters.py"],
|
|
"platform-verify": ["bash", "platform/checks/verify-platform.sh"],
|
|
# Athena's Compose files deliberately require values from the root-only
|
|
# runtime environment. Validating without it reports required model/token
|
|
# variables as missing even though the deployed stack is valid.
|
|
"compose-main": ["docker", "compose", "--env-file", "/etc/mike-ai/stack.env", "-f", "compose.yaml", "config", "-q"],
|
|
"compose-mcp": ["docker", "compose", "--env-file", "/etc/mike-ai/stack.env", "-f", "compose.yaml", "config", "-q"],
|
|
}
|
|
|
|
# Athena is physically remote. The general terminal is intentionally broad,
|
|
# but these operations can strand the machine and therefore remain impossible
|
|
# through the AI operator. This is a reachability guard, not a general command
|
|
# allowlist: ordinary Docker, files, Git, HTTP, package, model and remote-SSH
|
|
# work stays available.
|
|
TERMINAL_BLOCK_PATTERNS = (
|
|
r"(?:^|[;&|()\s])(?:shutdown|poweroff|reboot|halt|kexec)(?:\s|$)",
|
|
r"(?:^|[;&|()\s])init\s+[06](?:\s|$)",
|
|
r"systemctl\s+(?:stop|restart|disable|mask|kill)\s+[^;&|]*(?:ssh|sshd|networking|networkmanager|systemd-networkd|wireguard|wg-quick)",
|
|
r"(?:^|[;&|()\s])(?:iptables|ip6tables|nft|ufw|firewall-cmd)(?:\s|$)",
|
|
r"(?:^|[;&|()\s])ip\s+(?:route|rule|link|addr(?:ess)?)(?:\s|$)",
|
|
r"(?:^|[;&|()\s])(?:nmcli|wg|wg-quick)(?:\s|$)",
|
|
r"(?:^|[;&|()\s])(?:mount|umount|fdisk|sfdisk|cfdisk|parted|mkfs(?:\.[a-z0-9]+)?|wipefs)(?:\s|$)",
|
|
r"(?:^|[;&|()\s])(?:grub-install|update-grub|update-initramfs|modprobe|rmmod|insmod)(?:\s|$)",
|
|
r"/(?:etc/(?:ssh|network|systemd/network|wireguard)|boot|proc/sys)(?:/|\b)",
|
|
r"docker\s+(?:stop|restart|rm|kill)\s+[^;&|]*mike-ai-wireguard-gateway",
|
|
)
|
|
|
|
|
|
def now() -> int:
|
|
return int(time.time())
|
|
|
|
|
|
def compact(text: str) -> str:
|
|
if len(text) <= MAX_OUTPUT:
|
|
return text
|
|
return text[:22_000] + f"\n...[truncated from {len(text)} chars]...\n" + text[-7_000:]
|
|
|
|
|
|
def run(argv: list[str], *, cwd: Path = STACK, timeout: int = 900, check: bool = False) -> dict[str, Any]:
|
|
# The systemd service deliberately exposes /root read-only. Docker Buildx
|
|
# still needs a tiny writable activity/config directory, so keep that
|
|
# disposable state inside the operator's dedicated state tree.
|
|
docker_config = STATE / "docker-config"
|
|
docker_config.mkdir(parents=True, exist_ok=True)
|
|
environment = {
|
|
"PATH": "/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin",
|
|
"LANG": "C.UTF-8",
|
|
"HOME": str(STATE),
|
|
"DOCKER_CONFIG": str(docker_config),
|
|
}
|
|
git_key = Path(os.environ.get("ATHENA_OPERATOR_GIT_SSH_KEY", "/etc/mike-ai/athena-operator-git"))
|
|
known_hosts = Path(os.environ.get("ATHENA_OPERATOR_GIT_KNOWN_HOSTS", "/etc/mike-ai/athena-operator-known-hosts"))
|
|
if git_key.is_file():
|
|
environment["GIT_SSH_COMMAND"] = (
|
|
f"ssh -i {shlex.quote(str(git_key))} -o IdentitiesOnly=yes "
|
|
f"-o UserKnownHostsFile={shlex.quote(str(known_hosts))} "
|
|
"-o StrictHostKeyChecking=yes -o BatchMode=yes -o ConnectTimeout=8"
|
|
)
|
|
completed = subprocess.run(
|
|
argv, cwd=cwd, env=environment,
|
|
stdin=subprocess.DEVNULL, stdout=subprocess.PIPE, stderr=subprocess.STDOUT,
|
|
text=True, errors="replace", timeout=timeout, check=False,
|
|
)
|
|
result = {"argv": argv, "exit_code": completed.returncode, "output": compact(completed.stdout)}
|
|
if check and completed.returncode != 0:
|
|
raise RuntimeError(json.dumps(result, ensure_ascii=False))
|
|
return result
|
|
|
|
|
|
def safe_relative(value: str) -> Path:
|
|
if not SAFE_PATH.fullmatch(value or "") or value.startswith("/"):
|
|
raise ValueError("invalid repository-relative path")
|
|
path = Path(value)
|
|
if ".." in path.parts or any(part.casefold() in BLOCKED_PATH_PARTS for part in path.parts):
|
|
raise PermissionError("protected path")
|
|
resolved = (REPOSITORY / path).resolve(strict=False)
|
|
resolved.relative_to(REPOSITORY)
|
|
return path
|
|
|
|
|
|
def source_file(root: Path, relative: Path) -> Path:
|
|
root = root.resolve(strict=False)
|
|
candidate = (root / relative).resolve(strict=False)
|
|
candidate.relative_to(root)
|
|
return candidate
|
|
|
|
|
|
def sha(data: bytes) -> str:
|
|
return hashlib.sha256(data).hexdigest()
|
|
|
|
|
|
def apply_unified_patch(before: str, patch: str) -> str:
|
|
"""Apply one ordinary unified diff without invoking a shell command."""
|
|
if not patch or len(patch.encode()) > MAX_FILE_BYTES:
|
|
raise ValueError("patch is empty or exceeds limit")
|
|
source = before.splitlines(keepends=True)
|
|
lines = patch.splitlines(keepends=True)
|
|
position = 0
|
|
output: list[str] = []
|
|
index = 0
|
|
if index < len(lines) and lines[index].startswith("--- "):
|
|
index += 1
|
|
if index < len(lines) and lines[index].startswith("+++ "):
|
|
index += 1
|
|
hunks = 0
|
|
while index < len(lines):
|
|
header = lines[index].rstrip("\r\n")
|
|
match = HUNK_HEADER.match(header)
|
|
if not match:
|
|
raise ValueError("patch must contain only standard unified-diff hunks")
|
|
old_start = int(match.group(1))
|
|
old_count = int(match.group(2) or "1")
|
|
new_count = int(match.group(4) or "1")
|
|
target = max(0, old_start - 1)
|
|
if target < position or target > len(source):
|
|
raise RuntimeError("patch hunk is outside the source file")
|
|
output.extend(source[position:target])
|
|
position = target
|
|
consumed = produced = 0
|
|
index += 1
|
|
while index < len(lines) and not lines[index].startswith("@@ "):
|
|
line = lines[index]
|
|
if line.startswith("\\ No newline at end of file"):
|
|
index += 1
|
|
continue
|
|
if not line or line[0] not in " +-":
|
|
raise ValueError("invalid unified-diff line")
|
|
marker, value = line[0], line[1:]
|
|
if marker in " -":
|
|
if position >= len(source) or source[position] != value:
|
|
raise RuntimeError("patch context does not match source")
|
|
if marker == " ":
|
|
output.append(source[position]); produced += 1
|
|
position += 1; consumed += 1
|
|
else:
|
|
output.append(value); produced += 1
|
|
index += 1
|
|
if consumed != old_count or produced != new_count:
|
|
raise RuntimeError("patch hunk line counts do not match its header")
|
|
hunks += 1
|
|
if not hunks:
|
|
raise ValueError("patch contains no hunks")
|
|
output.extend(source[position:])
|
|
return "".join(output)
|
|
|
|
|
|
def normalise_patches(files: Any) -> tuple[list[dict[str, Any]], str]:
|
|
if not isinstance(files, list) or not 1 <= len(files) <= MAX_FILES:
|
|
raise ValueError("files must contain 1..24 patch entries")
|
|
normal: list[dict[str, Any]] = []
|
|
previews: list[str] = []
|
|
for item in files:
|
|
if not isinstance(item, dict):
|
|
raise ValueError("each patch entry must be an object")
|
|
relative = safe_relative(str(item.get("path", "")))
|
|
target = source_file(REPOSITORY, relative)
|
|
before = target.read_text(encoding="utf-8", errors="strict") if target.is_file() else ""
|
|
before_sha = sha(before.encode())
|
|
expected = str(item.get("expected_sha256", ""))
|
|
if expected and expected != before_sha:
|
|
raise RuntimeError(f"source drift for {relative}")
|
|
patch = str(item.get("patch", ""))
|
|
after = apply_unified_patch(before, patch)
|
|
if len(after.encode()) > MAX_FILE_BYTES:
|
|
raise ValueError("patched file exceeds limit")
|
|
normal.append({"path": str(relative), "content": after, "before_sha256": before_sha, "after_sha256": sha(after.encode())})
|
|
previews.append(f"### {relative}\n{compact(patch)}")
|
|
return normal, "\n\n".join(previews)
|
|
|
|
|
|
def json_write(path: Path, value: Any) -> None:
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
temporary = path.with_suffix(path.suffix + ".tmp")
|
|
temporary.write_text(json.dumps(value, ensure_ascii=False, indent=2), encoding="utf-8")
|
|
os.replace(temporary, path)
|
|
|
|
|
|
def audit(event: str, **fields: Any) -> None:
|
|
record = {"time": now(), "event": event, **fields}
|
|
path = STATE / "audit.jsonl"
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
with path.open("a", encoding="utf-8") as handle:
|
|
handle.write(json.dumps(record, ensure_ascii=False, separators=(",", ":")) + "\n")
|
|
|
|
|
|
def ensure_repository() -> None:
|
|
if not (REPOSITORY / ".git").is_dir():
|
|
raise RuntimeError(
|
|
f"canonical Git checkout is missing at {REPOSITORY}; "
|
|
"restore /opt/mike-ai/stack from Git and run the documented restore command"
|
|
)
|
|
remote = os.environ.get("ATHENA_OPERATOR_GIT_REMOTE", "").strip()
|
|
if remote:
|
|
run(["git", "remote", "set-url", "origin", remote], cwd=REPOSITORY, check=True)
|
|
run(["git", "config", "user.name", os.environ.get("ATHENA_OPERATOR_GIT_NAME", "Athena Operator")], cwd=REPOSITORY, check=True)
|
|
run(["git", "config", "user.email", os.environ.get("ATHENA_OPERATOR_GIT_EMAIL", "athena-operator@localhost")], cwd=REPOSITORY, check=True)
|
|
|
|
|
|
def inspect(subject: str, arguments: dict[str, Any]) -> dict[str, Any]:
|
|
ensure_repository()
|
|
if subject == "guide":
|
|
guide = REPOSITORY / "ATHENA.md"
|
|
text = guide.read_text(encoding="utf-8", errors="replace")
|
|
return {"guide": text, "source": "ATHENA.md", "sha256": sha(guide.read_bytes())}
|
|
if subject == "overview":
|
|
return {
|
|
"version": VERSION,
|
|
"source_commit": run(["git", "rev-parse", "HEAD"], cwd=REPOSITORY, check=True)["output"].strip(),
|
|
"git": run(["git", "status", "--short", "--branch"], cwd=REPOSITORY),
|
|
"containers": run(["docker", "ps", "--format", "{{.Names}}\t{{.Status}}\t{{.Image}}"]),
|
|
"storage": run(["df", "-h", "/", "/data", str(MODELS)]),
|
|
"gpus": run(["nvidia-smi", "--query-gpu=name,memory.total,memory.used,utilization.gpu", "--format=csv,noheader"]),
|
|
"layout": {"worktree": str(REPOSITORY), "runtime_stack": str(STACK), "single_worktree": REPOSITORY == STACK},
|
|
"boundary": "No shutdown, reboot or Athena SSH/network/firewall/kernel/partition/mount changes",
|
|
}
|
|
if subject == "git_status":
|
|
return {"status": run(["git", "status", "--short", "--branch"], cwd=REPOSITORY), "diff": run(["git", "diff", "--stat"], cwd=REPOSITORY)}
|
|
if subject == "containers":
|
|
return {"containers": run(["docker", "ps", "-a", "--format", "{{.Names}}\t{{.Status}}\t{{.Image}}"])}
|
|
if subject == "models":
|
|
entries = []
|
|
if MODELS.is_dir():
|
|
for path in sorted(MODELS.rglob("*.gguf")):
|
|
entries.append({"path": str(path.relative_to(MODELS)), "bytes": path.stat().st_size})
|
|
return {"models": entries[:500], "count": len(entries)}
|
|
if subject == "jobs":
|
|
jobs = []
|
|
for path in sorted((STATE / "jobs").glob("*.json"), reverse=True)[:30]:
|
|
jobs.append(json.loads(path.read_text()))
|
|
return {"jobs": jobs}
|
|
raise ValueError("unsupported inspection subject")
|
|
|
|
|
|
def read_source(arguments: dict[str, Any]) -> dict[str, Any]:
|
|
ensure_repository()
|
|
relative = safe_relative(str(arguments.get("path", "")))
|
|
target = source_file(REPOSITORY, relative)
|
|
if not target.is_file():
|
|
return {"ok": False, "error": "source file not found", "path": str(relative), "retry": False}
|
|
text = target.read_text(encoding="utf-8", errors="replace")
|
|
start = max(1, int(arguments.get("start_line", 1)))
|
|
count = min(200, max(1, int(arguments.get("line_count", 80))))
|
|
lines = text.splitlines()
|
|
selected = lines[start - 1:start - 1 + count]
|
|
return {
|
|
"ok": True, "path": str(relative), "sha256": sha(target.read_bytes()),
|
|
"start_line": start, "end_line": start + len(selected) - 1 if selected else start - 1,
|
|
"content": "\n".join(selected), "total_lines": len(lines),
|
|
"truncated": start - 1 + len(selected) < len(lines),
|
|
}
|
|
|
|
|
|
def search_source(arguments: dict[str, Any]) -> dict[str, Any]:
|
|
ensure_repository()
|
|
query = str(arguments.get("query", ""))
|
|
if not query or len(query) > 200 or any(x in query for x in ("\x00", "\n", "\r")):
|
|
raise ValueError("invalid query")
|
|
if shutil.which("rg"):
|
|
result = run(["rg", "-n", "-F", "-m", "20", "--hidden", "--glob", "!.git/**", "--", query, "."], cwd=REPOSITORY, timeout=20)
|
|
lines = result["output"].splitlines()[:20]
|
|
return {"ok": True, "query": query, "matches": lines, "count": len(lines), "truncated": len(lines) == 20, "engine": "rg"}
|
|
pattern = re.compile(re.escape(query))
|
|
matches: list[str] = []
|
|
for path in sorted(REPOSITORY.rglob("*")):
|
|
if not path.is_file() or ".git" in path.parts or path.stat().st_size > MAX_FILE_BYTES:
|
|
continue
|
|
try:
|
|
lines = path.read_text(encoding="utf-8", errors="strict").splitlines()
|
|
except (UnicodeDecodeError, OSError):
|
|
continue
|
|
relative = path.relative_to(REPOSITORY)
|
|
for number, line in enumerate(lines, 1):
|
|
if pattern.search(line):
|
|
matches.append(f"{relative}:{number}:{line}")
|
|
if len(matches) >= 20:
|
|
return {"ok": True, "query": query, "matches": matches, "count": len(matches), "engine": "python", "truncated": True}
|
|
return {"ok": True, "query": query, "matches": matches, "count": len(matches), "engine": "python", "truncated": False}
|
|
|
|
|
|
def staged_file(item: dict[str, Any]) -> dict[str, Any]:
|
|
source = Path(str(item.get("source", "")))
|
|
if not source.is_absolute() or not source.is_file():
|
|
raise FileNotFoundError("staged source file not found")
|
|
resolved = source.resolve()
|
|
if not any(resolved.is_relative_to(root) for root in STAGING_ROOTS):
|
|
raise PermissionError("staged source is outside approved staging roots")
|
|
raw = resolved.read_bytes()
|
|
if len(raw) > MAX_FILE_BYTES:
|
|
raise ValueError("staged source exceeds limit")
|
|
expected = str(item.get("expected_source_sha256", ""))
|
|
actual = sha(raw)
|
|
if not expected or expected != actual:
|
|
raise RuntimeError("staged source checksum is missing or does not match")
|
|
try:
|
|
content = raw.decode("utf-8", errors="strict")
|
|
except UnicodeDecodeError as exc:
|
|
raise ValueError("staged source must be UTF-8 text") from exc
|
|
relative = safe_relative(str(item.get("path", "")))
|
|
target = source_file(REPOSITORY, relative)
|
|
before = target.read_text(encoding="utf-8", errors="strict") if target.is_file() else ""
|
|
before_sha = sha(before.encode())
|
|
expected_target = str(item.get("expected_target_sha256", ""))
|
|
if expected_target and expected_target != before_sha:
|
|
raise RuntimeError(f"source drift for {relative}")
|
|
return {
|
|
"path": str(relative), "content": content, "before_sha256": before_sha,
|
|
"after_sha256": actual, "staged_source": str(resolved),
|
|
}
|
|
|
|
|
|
def terminal(arguments: dict[str, Any]) -> dict[str, Any]:
|
|
command = str(arguments.get("command", "")).strip()
|
|
if not command or len(command) > 8000 or "\x00" in command:
|
|
raise ValueError("invalid terminal command")
|
|
lowered = command.casefold()
|
|
for pattern in TERMINAL_BLOCK_PATTERNS:
|
|
if re.search(pattern, lowered, flags=re.IGNORECASE):
|
|
raise PermissionError(
|
|
"command blocked because it could break Athena power or remote reachability"
|
|
)
|
|
|
|
cwd_value = str(arguments.get("cwd", str(STACK)))
|
|
cwd = Path(cwd_value)
|
|
if not cwd.is_absolute() or not cwd.is_dir():
|
|
raise ValueError("cwd must be an existing absolute directory")
|
|
timeout = min(3600, max(1, int(arguments.get("timeout_seconds", 300))))
|
|
output_limit = min(30000, max(1000, int(arguments.get("max_output_chars", 12000))))
|
|
result = run(["/bin/bash", "-lc", command], cwd=cwd, timeout=timeout)
|
|
output = result.get("output", "")
|
|
if len(output) > output_limit:
|
|
result["output"] = (
|
|
output[: int(output_limit * 0.72)]
|
|
+ f"\n...[terminal output truncated from {len(output)} chars]...\n"
|
|
+ output[-int(output_limit * 0.25) :]
|
|
)
|
|
result["cwd"] = str(cwd)
|
|
result["reachability_guard"] = "active"
|
|
audit("terminal", command_sha256=sha(command.encode()), cwd=str(cwd), exit_code=result["exit_code"])
|
|
return result
|
|
|
|
|
|
def normalise_operation(operation: str, payload: dict[str, Any]) -> tuple[dict[str, Any], str]:
|
|
if operation not in ALLOWED_OPERATIONS:
|
|
raise ValueError("unsupported operation")
|
|
if not isinstance(payload, dict):
|
|
raise ValueError("payload must be an object")
|
|
if operation == "file_update":
|
|
files = payload.get("files")
|
|
if not isinstance(files, list) or not 1 <= len(files) <= MAX_FILES:
|
|
raise ValueError("files must contain 1..24 entries")
|
|
normal = []
|
|
previews = []
|
|
for item in files:
|
|
relative = safe_relative(str(item.get("path", "")))
|
|
content = str(item.get("content", ""))
|
|
raw = content.encode()
|
|
if len(raw) > MAX_FILE_BYTES:
|
|
raise ValueError("file content exceeds limit")
|
|
target = source_file(REPOSITORY, relative)
|
|
before = target.read_text(encoding="utf-8", errors="replace") if target.is_file() else ""
|
|
expected = str(item.get("expected_sha256", ""))
|
|
before_sha = sha(before.encode())
|
|
if expected and expected != before_sha:
|
|
raise RuntimeError(f"source drift for {relative}")
|
|
diff = "".join(difflib.unified_diff(before.splitlines(True), content.splitlines(True), fromfile=f"a/{relative}", tofile=f"b/{relative}"))
|
|
normal.append({"path": str(relative), "content": content, "before_sha256": before_sha, "after_sha256": sha(raw)})
|
|
previews.append(compact(diff))
|
|
return {"files": normal}, "\n".join(previews)
|
|
if operation == "patch_update":
|
|
files, preview = normalise_patches(payload.get("files"))
|
|
return {"files": files}, preview
|
|
if operation == "mcp_release":
|
|
patch_items = payload.get("files") or []
|
|
import_items = payload.get("imports") or []
|
|
if not isinstance(patch_items, list) or not isinstance(import_items, list):
|
|
raise ValueError("files and imports must be lists")
|
|
if not patch_items and not import_items:
|
|
raise ValueError("mcp_release requires at least one patch or staged import")
|
|
files, patch_preview = normalise_patches(patch_items) if patch_items else ([], "")
|
|
imports = [staged_file(item) for item in import_items]
|
|
files.extend(imports)
|
|
import_preview = "\n".join(
|
|
f"IMPORT {item['staged_source']} -> {item['path']} sha256={item['after_sha256']}"
|
|
for item in imports
|
|
)
|
|
checks = payload.get("checks") or ["operator-tests", "compose-mcp"]
|
|
if not isinstance(checks, list) or not checks or any(name not in ALLOWED_CHECKS for name in checks):
|
|
raise ValueError("unknown release check suite")
|
|
compose_file = "compose.yaml"
|
|
services = payload.get("services") or []
|
|
if not isinstance(services, list) or not 1 <= len(services) <= 12 or any(not SAFE_NAME.fullmatch(str(x)) for x in services):
|
|
raise ValueError("invalid MCP service list")
|
|
message = str(payload.get("message", ""))
|
|
if not SAFE_COMMIT.fullmatch(message):
|
|
raise ValueError("invalid commit message")
|
|
paths = payload.get("paths") or [item["path"] for item in files]
|
|
selected = [str(safe_relative(str(path))) for path in paths]
|
|
if len(set(selected)) != len(selected) or not set(item["path"] for item in files).issubset(set(selected)):
|
|
raise ValueError("release paths must be unique and include every patched file")
|
|
normal = {
|
|
"files": files, "checks": checks, "compose_file": compose_file,
|
|
"services": [str(x) for x in services], "build": bool(payload.get("build", True)),
|
|
"openwebui_sync": bool(payload.get("openwebui_sync", True)),
|
|
"hermes_sync": bool(payload.get("hermes_sync", False)),
|
|
"message": message, "paths": selected,
|
|
}
|
|
preview = (
|
|
f"ONE MCP RELEASE\nServices: {', '.join(normal['services'])}\n"
|
|
f"Checks: {', '.join(checks)}\nOpenWebUI sync: {normal['openwebui_sync']}\n"
|
|
f"Hermes sync: {normal['hermes_sync']}\n"
|
|
f"Selective commit: {message}\nPaths: {', '.join(selected)}\n\n"
|
|
+ "\n".join(part for part in (import_preview, patch_preview) if part)
|
|
)
|
|
return normal, preview
|
|
if operation == "run_checks":
|
|
checks = payload.get("checks") or []
|
|
if not isinstance(checks, list) or not checks or any(name not in ALLOWED_CHECKS for name in checks):
|
|
raise ValueError("unknown check suite")
|
|
return {"checks": checks}, "Will run: " + ", ".join(checks)
|
|
if operation == "compose_deploy":
|
|
compose_file = str(payload.get("compose_file", ""))
|
|
if compose_file != "compose.yaml":
|
|
raise ValueError("unsupported compose file")
|
|
services = payload.get("services") or []
|
|
if not isinstance(services, list) or not 1 <= len(services) <= 12 or any(not SAFE_NAME.fullmatch(str(x)) for x in services):
|
|
raise ValueError("invalid compose service list")
|
|
return {"compose_file": compose_file, "services": services, "build": bool(payload.get("build", True))}, f"docker compose -f {compose_file} up -d {'--build ' if payload.get('build', True) else ''}{' '.join(services)}"
|
|
if operation == "container_action":
|
|
action = str(payload.get("action", ""))
|
|
containers = payload.get("containers") or []
|
|
if action not in {"start", "stop", "restart"}:
|
|
raise ValueError("invalid container action")
|
|
if not isinstance(containers, list) or not 1 <= len(containers) <= 12:
|
|
raise ValueError("invalid container list")
|
|
for name in containers:
|
|
if not SAFE_NAME.fullmatch(str(name)) or not str(name).startswith("mike-ai-") or name in PROTECTED_CONTAINERS:
|
|
raise PermissionError(f"container is outside operator boundary: {name}")
|
|
return {"action": action, "containers": containers}, f"docker {action} {' '.join(containers)}"
|
|
if operation == "openwebui_sync":
|
|
return {}, (
|
|
"Synchronise the versioned Open WebUI model profiles, filters, tool connections "
|
|
"and system prompt with the running Open WebUI instance."
|
|
)
|
|
if operation == "git_publish":
|
|
message = str(payload.get("message", ""))
|
|
if not SAFE_COMMIT.fullmatch(message):
|
|
raise ValueError("invalid commit message")
|
|
paths = payload.get("paths")
|
|
if not isinstance(paths, list) or not 1 <= len(paths) <= MAX_FILES:
|
|
raise ValueError("paths must contain 1..24 repository files")
|
|
selected = [str(safe_relative(str(path))) for path in paths]
|
|
if len(set(selected)) != len(selected):
|
|
raise ValueError("paths must be unique")
|
|
status = run(["git", "status", "--short", "--", *selected], cwd=REPOSITORY, check=True)["output"].strip()
|
|
if not status:
|
|
raise RuntimeError("selected repository paths have no changes to publish")
|
|
diff = run(["git", "diff", "--stat", "--", *selected], cwd=REPOSITORY, check=True)["output"]
|
|
preview = f"Commit message: {message}\nSelected paths: {', '.join(selected)}\n\nSelected changes:\n{status}\n\nDiff summary:\n{diff}"
|
|
return {"message": message, "paths": selected, "reviewed_status": status}, preview
|
|
if operation == "model_download":
|
|
url = str(payload.get("url", ""))
|
|
parsed = urllib.parse.urlparse(url)
|
|
if parsed.scheme != "https" or parsed.hostname not in {"huggingface.co", "cdn-lfs.huggingface.co", "hf.co"}:
|
|
raise PermissionError("only HTTPS Hugging Face downloads are allowed")
|
|
destination = safe_relative(str(payload.get("destination", "")))
|
|
expected = str(payload.get("sha256", ""))
|
|
if expected and not re.fullmatch(r"[0-9a-f]{64}", expected):
|
|
raise ValueError("invalid sha256")
|
|
return {"url": url, "destination": str(destination), "sha256": expected}, f"Download {url} to models/{destination}"
|
|
if operation == "benchmark":
|
|
script = safe_relative(str(payload.get("script", "")))
|
|
if not (str(script).startswith("dev/") or str(script).startswith("platform/bench")) or script.suffix not in {".py", ".sh"}:
|
|
raise PermissionError("benchmark script must be versioned below dev/ or platform/bench*")
|
|
args = payload.get("arguments") or []
|
|
if not isinstance(args, list) or len(args) > 20 or any(not isinstance(x, str) or len(x) > 200 or re.search(r"[\x00\n\r]", x) for x in args):
|
|
raise ValueError("invalid benchmark arguments")
|
|
return {"script": str(script), "arguments": args}, f"Run versioned benchmark {script} with {args!r}"
|
|
if operation == "backup":
|
|
return {}, "Create one Docker-data backup now in /data/docker-backups."
|
|
raise AssertionError(operation)
|
|
|
|
|
|
def prepare(arguments: dict[str, Any]) -> dict[str, Any]:
|
|
ensure_repository()
|
|
operation = str(arguments.get("operation", ""))
|
|
payload, preview = normalise_operation(operation, arguments.get("payload") or {})
|
|
ticket_id = uuid.uuid4().hex
|
|
record = {"id": ticket_id, "operation": operation, "payload": payload, "preview": preview, "created": now(), "expires": now() + TICKET_TTL, "status": "pending"}
|
|
record["binding"] = sha(json.dumps({"operation": operation, "payload": payload}, sort_keys=True, ensure_ascii=False).encode())
|
|
json_write(STATE / "pending" / f"{ticket_id}.json", record)
|
|
audit("prepared", ticket=ticket_id, operation=operation, binding=record["binding"])
|
|
return {"ticket": ticket_id, "operation": operation, "binding": record["binding"], "preview": preview, "expires_in_seconds": TICKET_TTL, "required_confirmation": f"EXECUTE {ticket_id}", "instruction": "Show the full preview to the user and wait for explicit confirmation in a later message."}
|
|
|
|
|
|
def sync_file(relative: Path, content: str, backup_root: Path) -> None:
|
|
for root in dict.fromkeys((REPOSITORY, STACK)):
|
|
target = source_file(root, relative)
|
|
target_mode = stat.S_IMODE(target.stat().st_mode) if target.exists() else 0o644
|
|
if target.exists():
|
|
backup = backup_root / root.name / relative
|
|
backup.parent.mkdir(parents=True, exist_ok=True)
|
|
shutil.copy2(target, backup)
|
|
target.parent.mkdir(parents=True, exist_ok=True)
|
|
fd, temp_name = tempfile.mkstemp(prefix=f".{target.name}.", dir=target.parent)
|
|
with os.fdopen(fd, "w", encoding="utf-8") as handle:
|
|
handle.write(content)
|
|
os.chmod(temp_name, target_mode)
|
|
os.replace(temp_name, target)
|
|
|
|
|
|
def start_job(ticket: str, operation: str, worker) -> dict[str, Any]:
|
|
job_id = uuid.uuid4().hex
|
|
job_path = STATE / "jobs" / f"{job_id}.json"
|
|
json_write(job_path, {"id": job_id, "ticket": ticket, "operation": operation, "status": "running", "started": now()})
|
|
def wrapped():
|
|
try:
|
|
result = worker()
|
|
json_write(job_path, {"id": job_id, "ticket": ticket, "operation": operation, "status": "completed", "started": json.loads(job_path.read_text())["started"], "finished": now(), "result": result})
|
|
audit("job_completed", job=job_id, operation=operation)
|
|
except Exception as exc:
|
|
json_write(job_path, {"id": job_id, "ticket": ticket, "operation": operation, "status": "failed", "finished": now(), "error": compact(str(exc))})
|
|
audit("job_failed", job=job_id, operation=operation)
|
|
threading.Thread(target=wrapped, daemon=True).start()
|
|
return {"job_id": job_id, "status": "running", "instruction": "Poll athena_operator_job until completed or failed."}
|
|
|
|
|
|
def apply_files(files: list[dict[str, Any]], backup: Path) -> list[str]:
|
|
for item in files:
|
|
relative = safe_relative(item["path"])
|
|
current = source_file(REPOSITORY, relative)
|
|
current_sha = sha(current.read_bytes()) if current.is_file() else sha(b"")
|
|
if current_sha != item["before_sha256"]:
|
|
raise RuntimeError(f"source drift after preview: {relative}")
|
|
for item in files:
|
|
sync_file(safe_relative(item["path"]), item["content"], backup)
|
|
return [item["path"] for item in files]
|
|
|
|
|
|
def restore_files(files: list[dict[str, Any]], backup: Path) -> None:
|
|
for item in files:
|
|
relative = safe_relative(item["path"])
|
|
for root in dict.fromkeys((REPOSITORY, STACK)):
|
|
target = source_file(root, relative)
|
|
saved = backup / root.name / relative
|
|
if saved.is_file():
|
|
target.parent.mkdir(parents=True, exist_ok=True)
|
|
shutil.copy2(saved, target)
|
|
elif target.exists():
|
|
target.unlink()
|
|
|
|
|
|
def publish_paths(message: str, selected: list[str]) -> dict[str, Any]:
|
|
run(["git", "add", "--", *selected], cwd=REPOSITORY, check=True)
|
|
run(["git", "diff", "--cached", "--check", "--", *selected], cwd=REPOSITORY, check=True)
|
|
commit = run(["git", "commit", "-m", message, "--", *selected], cwd=REPOSITORY, check=True)
|
|
pushed = run(["git", "push", "origin", "HEAD:main"], cwd=REPOSITORY, timeout=300, check=True)
|
|
head = run(["git", "rev-parse", "HEAD"], cwd=REPOSITORY, check=True)["output"].strip()
|
|
(STACK / ".mike-ai-source-commit").write_text(head + "\n")
|
|
return {"commit": head, "commit_output": commit, "push_output": pushed}
|
|
|
|
|
|
def execute_operation(ticket: str, operation: str, payload: dict[str, Any]) -> dict[str, Any]:
|
|
if operation in {"file_update", "patch_update"}:
|
|
backup = STATE / "backups" / f"{now()}-{ticket}"
|
|
changed = apply_files(payload["files"], backup)
|
|
return {"changed": changed, "backup": str(backup), "git_diff": run(["git", "diff", "--stat"], cwd=REPOSITORY)}
|
|
if operation == "mcp_release":
|
|
def release():
|
|
backup = STATE / "backups" / f"{now()}-{ticket}"
|
|
published = False
|
|
deployed = False
|
|
synced = False
|
|
hermes_synced = False
|
|
try:
|
|
changed = apply_files(payload["files"], backup)
|
|
checks = []
|
|
for name in payload["checks"]:
|
|
result = run(ALLOWED_CHECKS[name], cwd=REPOSITORY, timeout=1200, check=True)
|
|
checks.append({"name": name, **result})
|
|
run(["docker", "compose", "-f", payload["compose_file"], "config", "-q"], cwd=STACK, check=True)
|
|
argv = ["docker", "compose", "-f", payload["compose_file"], "up", "-d"]
|
|
if payload["build"]: argv.append("--build")
|
|
argv.extend(payload["services"])
|
|
deploy = run(argv, cwd=STACK, timeout=3600, check=True)
|
|
deployed = True
|
|
sync = None
|
|
if payload["openwebui_sync"]:
|
|
sync = run(["bash", str(STACK / "platform/openwebui/install-filters.sh")], cwd=STACK, timeout=1800, check=True)
|
|
synced = True
|
|
hermes_sync = None
|
|
if payload["hermes_sync"]:
|
|
hermes_sync = run(["bash", str(STACK / "platform/hermes/install-hermes.sh")], cwd=STACK, timeout=1800, check=True)
|
|
hermes_synced = True
|
|
publication = publish_paths(payload["message"], payload["paths"])
|
|
published = True
|
|
return {"changed": changed, "checks": checks, "deploy": deploy, "openwebui_sync": sync, "hermes_sync": hermes_sync, "publication": publication, "containers": run(["docker", "ps", "--format", "{{.Names}}\t{{.Status}}"])}
|
|
except Exception:
|
|
if not published:
|
|
restore_files(payload["files"], backup)
|
|
run(["git", "reset", "--", *payload["paths"]], cwd=REPOSITORY)
|
|
if deployed:
|
|
rollback_argv = ["docker", "compose", "-f", payload["compose_file"], "up", "-d", "--build", *payload["services"]]
|
|
run(rollback_argv, cwd=STACK, timeout=3600)
|
|
if synced:
|
|
run(["bash", str(STACK / "platform/openwebui/install-filters.sh")], cwd=STACK, timeout=1800)
|
|
if hermes_synced:
|
|
run(["bash", str(STACK / "platform/hermes/install-hermes.sh")], cwd=STACK, timeout=1800)
|
|
raise
|
|
return start_job(ticket, operation, release)
|
|
if operation == "run_checks":
|
|
return {"checks": [{"name": name, **run(ALLOWED_CHECKS[name], cwd=REPOSITORY, timeout=1200)} for name in payload["checks"]]}
|
|
if operation == "compose_deploy":
|
|
compose = payload["compose_file"]
|
|
run(["docker", "compose", "-f", compose, "config", "-q"], cwd=STACK, check=True)
|
|
argv = ["docker", "compose", "-f", compose, "up", "-d"]
|
|
if payload["build"]:
|
|
argv.append("--build")
|
|
argv.extend(payload["services"])
|
|
return {"deploy": run(argv, cwd=STACK, timeout=3600, check=True), "containers": run(["docker", "ps", "--format", "{{.Names}}\t{{.Status}}"])}
|
|
if operation == "container_action":
|
|
return {"action": run(["docker", payload["action"], *payload["containers"]], timeout=300, check=True)}
|
|
if operation == "openwebui_sync":
|
|
installer = STACK / "platform/openwebui/install-filters.sh"
|
|
if not installer.is_file():
|
|
raise FileNotFoundError("versioned Open WebUI synchronisation script is missing")
|
|
return {"sync": run(["bash", str(installer)], cwd=STACK, timeout=1800, check=True)}
|
|
if operation == "git_publish":
|
|
selected = payload["paths"]
|
|
current_status = run(["git", "status", "--short", "--", *selected], cwd=REPOSITORY, check=True)["output"].strip()
|
|
if current_status != payload["reviewed_status"]:
|
|
raise RuntimeError("selected repository paths changed after the Git publish preview")
|
|
return publish_paths(payload["message"], selected)
|
|
if operation == "model_download":
|
|
def download():
|
|
destination = source_file(MODELS, Path(payload["destination"]))
|
|
destination.parent.mkdir(parents=True, exist_ok=True)
|
|
partial = destination.with_suffix(destination.suffix + ".partial")
|
|
request = urllib.request.Request(payload["url"], headers={"User-Agent": "Athena-Operator/1"})
|
|
digest = hashlib.sha256()
|
|
total = 0
|
|
with urllib.request.urlopen(request, timeout=60) as response, partial.open("wb") as output:
|
|
while chunk := response.read(8 * 1024 * 1024):
|
|
output.write(chunk); digest.update(chunk); total += len(chunk)
|
|
actual = digest.hexdigest()
|
|
if payload["sha256"] and actual != payload["sha256"]:
|
|
partial.unlink(missing_ok=True); raise RuntimeError("model checksum mismatch")
|
|
os.replace(partial, destination)
|
|
return {"destination": str(destination), "bytes": total, "sha256": actual}
|
|
return start_job(ticket, operation, download)
|
|
if operation == "benchmark":
|
|
def benchmark():
|
|
script = source_file(REPOSITORY, safe_relative(payload["script"]))
|
|
interpreter = "python3" if script.suffix == ".py" else "bash"
|
|
return run([interpreter, str(script), *payload["arguments"]], cwd=REPOSITORY, timeout=86400)
|
|
return start_job(ticket, operation, benchmark)
|
|
if operation == "backup":
|
|
def backup():
|
|
result = run(["docker", "exec", "mike-ai-backup", "backup"], timeout=7200, check=True)
|
|
latest = Path("/data/docker-backups/athena-latest.tar.gz")
|
|
if not latest.is_file():
|
|
raise RuntimeError("backup completed without latest archive")
|
|
return {"backup": result, "archive": str(latest), "bytes": latest.stat().st_size}
|
|
return start_job(ticket, operation, backup)
|
|
raise AssertionError(operation)
|
|
|
|
|
|
def execute(arguments: dict[str, Any]) -> dict[str, Any]:
|
|
ticket = str(arguments.get("ticket", ""))
|
|
confirmation = str(arguments.get("confirmation", ""))
|
|
if not re.fullmatch(r"[0-9a-f]{32}", ticket) or confirmation != f"EXECUTE {ticket}":
|
|
raise PermissionError("exact ticket confirmation required")
|
|
path = STATE / "pending" / f"{ticket}.json"
|
|
if not path.is_file():
|
|
raise ValueError("ticket not found or already consumed")
|
|
with LOCK:
|
|
record = json.loads(path.read_text())
|
|
if record["status"] != "pending" or now() > record["expires"]:
|
|
raise PermissionError("ticket expired or unavailable")
|
|
expected = sha(json.dumps({"operation": record["operation"], "payload": record["payload"]}, sort_keys=True, ensure_ascii=False).encode())
|
|
if expected != record["binding"]:
|
|
raise RuntimeError("ticket binding mismatch")
|
|
record["status"] = "executing"
|
|
json_write(path, record)
|
|
try:
|
|
result = execute_operation(ticket, record["operation"], record["payload"])
|
|
record["status"] = "executed"
|
|
record["executed"] = now()
|
|
record["result_summary"] = compact(json.dumps(result, ensure_ascii=False))
|
|
completed = STATE / "completed" / path.name
|
|
json_write(completed, record)
|
|
path.unlink()
|
|
audit("executed", ticket=ticket, operation=record["operation"], binding=record["binding"])
|
|
return {"ticket": ticket, "operation": record["operation"], "result": result, "instruction": "Verify service health and Git state before declaring the work complete."}
|
|
except Exception:
|
|
record["status"] = "failed"
|
|
json_write(path, record)
|
|
raise
|
|
|
|
|
|
def change(arguments: dict[str, Any]) -> dict[str, Any]:
|
|
"""Apply one user-requested operation without a second ticket ceremony."""
|
|
ensure_repository()
|
|
operation = str(arguments.get("operation", ""))
|
|
payload, preview = normalise_operation(operation, arguments.get("payload") or {})
|
|
change_id = uuid.uuid4().hex
|
|
audit("change_started", change=change_id, operation=operation)
|
|
try:
|
|
result = execute_operation(change_id, operation, payload)
|
|
except Exception:
|
|
audit("change_failed", change=change_id, operation=operation)
|
|
raise
|
|
audit("change_completed", change=change_id, operation=operation)
|
|
return {
|
|
"change_id": change_id,
|
|
"operation": operation,
|
|
"summary": compact(preview),
|
|
"result": result,
|
|
"instruction": "Verify the focused service behavior before declaring completion.",
|
|
}
|
|
|
|
|
|
def job(arguments: dict[str, Any]) -> dict[str, Any]:
|
|
job_id = str(arguments.get("job_id", ""))
|
|
if not re.fullmatch(r"[0-9a-f]{32}", job_id):
|
|
raise ValueError("invalid job id")
|
|
path = STATE / "jobs" / f"{job_id}.json"
|
|
if not path.is_file():
|
|
raise FileNotFoundError("job not found")
|
|
return json.loads(path.read_text())
|
|
|
|
|
|
def dispatch(request: dict[str, Any]) -> dict[str, Any]:
|
|
action = request.get("action")
|
|
arguments = request.get("arguments") or {}
|
|
if action == "inspect": return inspect(str(arguments.get("subject", "overview")), arguments)
|
|
if action == "read_source": return read_source(arguments)
|
|
if action == "search_source": return search_source(arguments)
|
|
if action == "terminal": return terminal(arguments)
|
|
if action == "change": return change(arguments)
|
|
if action == "prepare": return prepare(arguments)
|
|
if action == "execute": return execute(arguments)
|
|
if action == "job": return job(arguments)
|
|
raise ValueError("unsupported operator action")
|
|
|
|
|
|
class Handler(socketserver.StreamRequestHandler):
|
|
def handle(self) -> None:
|
|
line = self.rfile.readline(2_000_000)
|
|
try:
|
|
request = json.loads(line)
|
|
result = dispatch(request)
|
|
response = {"ok": True, "result": result}
|
|
except Exception as exc:
|
|
response = {"ok": False, "error": compact(str(exc))}
|
|
self.wfile.write((json.dumps(response, ensure_ascii=False, separators=(",", ":")) + "\n").encode())
|
|
|
|
|
|
def main() -> None:
|
|
STATE.mkdir(parents=True, exist_ok=True)
|
|
SOCKET.parent.mkdir(parents=True, exist_ok=True)
|
|
SOCKET.unlink(missing_ok=True)
|
|
server = socketserver.ThreadingUnixStreamServer(str(SOCKET), Handler)
|
|
os.chmod(SOCKET, 0o660)
|
|
os.chown(SOCKET, 0, int(os.environ.get("ATHENA_OPERATOR_GID", "10003")))
|
|
audit("daemon_started", version=VERSION)
|
|
server.serve_forever()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|