Add self-maintaining Athena platform context MCP

2026-08-23 17:31:05 +02:00
parent 64d36ad838
commit 3dbb66037d
22 changed files with 1071 additions and 9 deletions
@@ -0,0 +1,113 @@
+#!/usr/bin/env python3
+"""Create a bounded, payload-free Athena runtime snapshot for the context MCP."""
+
+from __future__ import annotations
+
+import hashlib
+import json
+import os
+import pathlib
+import subprocess
+import tempfile
+import time
+
+
+OUTPUT = pathlib.Path("/var/lib/mike-ai-platform-context/runtime.json")
+STACK = pathlib.Path("/opt/mike-ai/stack")
+
+
+def command(*args: str) -> str:
+    try:
+        return subprocess.run(args, check=True, text=True, capture_output=True, timeout=15).stdout.strip()
+    except (OSError, subprocess.SubprocessError):
+        return ""
+
+
+def docs_hash() -> str | None:
+    digest = hashlib.sha256()
+    files = sorted((STACK / "docs").glob("*.md"))
+    if not files:
+        return None
+    for path in files:
+        digest.update(path.name.encode())
+        digest.update(b"\0")
+        digest.update(path.read_bytes())
+        digest.update(b"\0")
+    return digest.hexdigest()
+
+
+def containers() -> list[dict[str, str]]:
+    raw = command("docker", "ps", "--filter", "name=mike-ai-", "--format", "{{.Names}}|{{.Image}}|{{.Status}}")
+    result = []
+    for line in raw.splitlines():
+        fields = line.split("|", 2)
+        if len(fields) == 3:
+            result.append({"name": fields[0], "image": fields[1], "status": fields[2]})
+    return sorted(result, key=lambda item: item["name"])
+
+
+def gpus() -> list[dict[str, object]]:
+    raw = command("nvidia-smi", "--query-gpu=uuid,name,memory.total,memory.used,driver_version", "--format=csv,noheader,nounits")
+    result = []
+    for line in raw.splitlines():
+        fields = [field.strip() for field in line.split(",")]
+        if len(fields) == 5:
+            result.append({"uuid": fields[0], "name": fields[1], "memory_total_mib": int(fields[2]), "memory_used_mib": int(fields[3]), "driver": fields[4]})
+    return result
+
+
+def filesystems() -> list[dict[str, object]]:
+    raw = command("df", "-B1", "--output=target,fstype,size,used,avail,pcent", "/", "/data")
+    result = []
+    for line in raw.splitlines()[1:]:
+        fields = line.split()
+        if len(fields) == 6:
+            result.append({"mount": fields[0], "fstype": fields[1], "size_bytes": int(fields[2]), "used_bytes": int(fields[3]), "available_bytes": int(fields[4]), "used_percent": fields[5]})
+    return result
+
+
+def recovery_status() -> dict[str, object]:
+    link = pathlib.Path("/data/mike-ai-recovery-kit")
+    if not link.exists():
+        return {"present": False}
+    target = link.resolve()
+    checksums = target / "SHA256SUMS"
+    return {"present": True, "target": str(target), "modified_unix": int(target.stat().st_mtime), "checksums_present": checksums.is_file()}
+
+
+def main() -> None:
+    generated = int(time.time())
+    active = [item["name"].removeprefix("mike-ai-llama-") for item in containers() if item["name"].startswith("mike-ai-llama-")]
+    commit_file = STACK / ".mike-ai-source-commit"
+    data = {
+        "generated_unix": generated,
+        "generated_at": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime(generated)),
+        "hostname": command("hostname"),
+        "os_release": command("sh", "-c", ". /etc/os-release && printf '%s %s' \"$ID\" \"$VERSION_ID\""),
+        "kernel": command("uname", "-r"),
+        "uptime_seconds": float(pathlib.Path("/proc/uptime").read_text().split()[0]),
+        "memory": {"summary": command("free", "-b", "--si").splitlines()[1] if command("free", "-b", "--si") else ""},
+        "filesystems": filesystems(),
+        "gpus": gpus(),
+        "containers": containers(),
+        "active_inference_profiles": active,
+        "source_commit": commit_file.read_text().strip() if commit_file.is_file() else None,
+        "documentation_tree_sha256": docs_hash(),
+        "recovery_kit": recovery_status(),
+        "privacy_scope": "No logs, prompts, chats, container environment values, file contents outside versioned docs, or secrets are collected.",
+    }
+    OUTPUT.parent.mkdir(parents=True, exist_ok=True)
+    fd, temporary = tempfile.mkstemp(prefix=".runtime.", dir=OUTPUT.parent)
+    try:
+        with os.fdopen(fd, "w", encoding="utf-8") as handle:
+            json.dump(data, handle, ensure_ascii=False, indent=2)
+            handle.write("\n")
+        os.chmod(temporary, 0o644)
+        os.replace(temporary, OUTPUT)
+    finally:
+        if os.path.exists(temporary):
+            os.unlink(temporary)
+
+
+if __name__ == "__main__":
+    main()