686 lines
33 KiB
Python
686 lines
33 KiB
Python
#!/usr/bin/env python3
|
|
"""Bounded platform knowledge and documentation-maintenance MCP for Athena.
|
|
|
|
This service deliberately has no Docker socket, shell tool, network egress or
|
|
secret mounts. Live data is supplied by a root-owned, fixed-command snapshot
|
|
timer. Canonical documentation may only be changed through a preview/apply
|
|
workflow and only below docs/.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import difflib
|
|
import calendar
|
|
import json
|
|
import os
|
|
import re
|
|
import socket
|
|
import sys
|
|
import tempfile
|
|
import time
|
|
import uuid
|
|
import urllib.error
|
|
import urllib.request
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
|
|
SERVER_VERSION = "1.1.0"
|
|
REPO_ROOT = Path(os.environ.get("ATHENA_REPO_ROOT", "/knowledge/repo"))
|
|
DOCS_ROOT = Path(os.environ.get("ATHENA_DOCS_ROOT", "/workspace/docs"))
|
|
RUNTIME_FILE = Path(os.environ.get("ATHENA_RUNTIME_FILE", "/runtime/runtime.json"))
|
|
STATE_ROOT = Path(os.environ.get("ATHENA_CONTEXT_STATE", "/state"))
|
|
WRITE_MODE = os.environ.get("ATHENA_DOC_WRITE_MODE", "proposal-only")
|
|
MAX_DOCUMENT_CHARS = 24000
|
|
MAX_SEARCH_RESULTS = 8
|
|
MAX_UPDATE_CHARS = 120000
|
|
ALLOWED_TEXT_SUFFIXES = {".md", ".txt", ".yaml", ".yml", ".json", ".py", ".sh", ".service", ".timer", ".conf", ".example"}
|
|
EXCLUDED_PARTS = {".git", "__pycache__", "xtts-test-audio", ".venv", "node_modules"}
|
|
|
|
if hasattr(sys.stdin, "reconfigure"):
|
|
sys.stdin.reconfigure(encoding="utf-8", errors="replace")
|
|
if hasattr(sys.stdout, "reconfigure"):
|
|
sys.stdout.reconfigure(encoding="utf-8", errors="replace")
|
|
|
|
|
|
TOOLS = [
|
|
{
|
|
"name": "athena_get_overview",
|
|
"description": (
|
|
"USE FIRST when a request concerns Athena, MikeAI, its models, profiles, GPUs, "
|
|
"OpenWebUI, router, MCPs, TTS/STT, Vision, networking or recovery. Returns the "
|
|
"short authoritative architecture overview plus snapshot freshness. This is "
|
|
"read-only and contains no secrets. Runtime claims still require "
|
|
"athena_get_current_state or the relevant specialist MCP."
|
|
),
|
|
"inputSchema": {"type": "object", "properties": {}, "additionalProperties": False},
|
|
},
|
|
{
|
|
"name": "athena_get_current_state",
|
|
"description": (
|
|
"USE for the current bounded Athena runtime inventory: host, filesystems, GPUs, "
|
|
"active MikeAI containers, active inference profile, source commit and recovery "
|
|
"freshness. The snapshot contains no logs, prompts, chats, environment values or "
|
|
"secrets. If stale, state that explicitly. For detailed service diagnosis use the "
|
|
"specialist management tool instead of guessing."
|
|
),
|
|
"inputSchema": {"type": "object", "properties": {}, "additionalProperties": False},
|
|
},
|
|
{
|
|
"name": "athena_get_external_services",
|
|
"description": (
|
|
"USE before designing or installing an integration that may already run on Unraid "
|
|
"or elsewhere in the home network. Returns the versioned, secret-free service "
|
|
"catalog and performs only fixed bounded reachability checks for those catalogued "
|
|
"endpoints. It accepts no host, URL or port from the model and is not a scanner. "
|
|
"A failed check means unavailable or unverified; it never authorizes creating a "
|
|
"duplicate service. Use the listed specialist MCP for detailed current state."
|
|
),
|
|
"inputSchema": {"type": "object", "properties": {}, "additionalProperties": False},
|
|
},
|
|
{
|
|
"name": "athena_search_knowledge",
|
|
"description": (
|
|
"USE to find the relevant MikeAI documentation, Compose definition, installer, "
|
|
"profile, runbook or source file before planning a platform change. Returns bounded "
|
|
"matching excerpts and paths. Do not repeatedly rephrase the same search; follow up "
|
|
"with athena_read_source for the selected file."
|
|
),
|
|
"inputSchema": {
|
|
"type": "object",
|
|
"properties": {
|
|
"query": {"type": "string", "minLength": 2, "maxLength": 300},
|
|
"max_results": {"type": "integer", "minimum": 1, "maximum": 8, "default": 5},
|
|
},
|
|
"required": ["query"],
|
|
"additionalProperties": False,
|
|
},
|
|
},
|
|
{
|
|
"name": "athena_read_source",
|
|
"description": (
|
|
"USE after athena_search_knowledge to read a bounded section of one versioned "
|
|
"MikeAI source or documentation file. Secret files, .git and binary artifacts are "
|
|
"not accessible. Paths are relative to the repository, for example "
|
|
"docs/OPERATIONS.md or platform/mcp/compose.yaml."
|
|
),
|
|
"inputSchema": {
|
|
"type": "object",
|
|
"properties": {
|
|
"path": {"type": "string", "minLength": 3, "maxLength": 240},
|
|
"start_line": {"type": "integer", "minimum": 1, "default": 1},
|
|
"max_lines": {"type": "integer", "minimum": 1, "maximum": 300, "default": 160},
|
|
},
|
|
"required": ["path"],
|
|
"additionalProperties": False,
|
|
},
|
|
},
|
|
{
|
|
"name": "athena_get_change_workflow",
|
|
"description": (
|
|
"USE before adding or replacing a model, MCP, TTS/STT, Vision/image service, "
|
|
"OpenWebUI integration, network component or recovery behavior. Returns the source "
|
|
"files, safety gates, validation steps, documentation duties, Git duties and "
|
|
"recovery duties for that change type. It performs no change."
|
|
),
|
|
"inputSchema": {
|
|
"type": "object",
|
|
"properties": {
|
|
"change_type": {
|
|
"type": "string",
|
|
"enum": ["mcp", "model", "profile", "tts", "stt", "vision", "image", "openwebui", "network", "recovery", "other"],
|
|
}
|
|
},
|
|
"required": ["change_type"],
|
|
"additionalProperties": False,
|
|
},
|
|
},
|
|
{
|
|
"name": "athena_prepare_documentation_update",
|
|
"description": (
|
|
"USE only after a real platform change or verified documentation drift. Creates a "
|
|
"reviewable proposal; it does not alter canonical documentation. Each update must "
|
|
"target an existing or new Markdown file below docs/. Include only verified facts, "
|
|
"never secrets, prompts, chats or private content. After preview, wait for explicit "
|
|
"user approval before calling athena_apply_documentation_update."
|
|
),
|
|
"inputSchema": {
|
|
"type": "object",
|
|
"properties": {
|
|
"summary": {"type": "string", "minLength": 5, "maxLength": 500},
|
|
"evidence": {"type": "string", "minLength": 5, "maxLength": 2000},
|
|
"updates": {
|
|
"type": "array",
|
|
"minItems": 1,
|
|
"maxItems": 6,
|
|
"items": {
|
|
"type": "object",
|
|
"properties": {
|
|
"path": {"type": "string", "pattern": "^docs/[A-Za-z0-9_.-]+\\.md$"},
|
|
"content": {"type": "string", "minLength": 1, "maxLength": 120000},
|
|
},
|
|
"required": ["path", "content"],
|
|
"additionalProperties": False,
|
|
},
|
|
},
|
|
},
|
|
"required": ["summary", "evidence", "updates"],
|
|
"additionalProperties": False,
|
|
},
|
|
},
|
|
{
|
|
"name": "athena_apply_documentation_update",
|
|
"description": (
|
|
"WRITE TOOL. Use only after the user explicitly approved the exact proposal in the "
|
|
"current conversation. Applies an already prepared proposal atomically below docs/, "
|
|
"backs up prior files and appends an audit record. It cannot change code, Compose, "
|
|
"services, secrets, Git or recovery bundles. The result always lists required Git "
|
|
"commit/push and recovery refresh work; never claim those are complete unless their "
|
|
"separate tools verify them."
|
|
),
|
|
"inputSchema": {
|
|
"type": "object",
|
|
"properties": {
|
|
"proposal_id": {"type": "string", "pattern": "^[a-f0-9]{32}$"},
|
|
"confirmation": {"type": "string", "minLength": 38, "maxLength": 64},
|
|
},
|
|
"required": ["proposal_id", "confirmation"],
|
|
"additionalProperties": False,
|
|
},
|
|
},
|
|
{
|
|
"name": "athena_get_maintenance_status",
|
|
"description": (
|
|
"USE after documentation or platform work. Reports pending documentation proposals, "
|
|
"applied documentation changes awaiting Git/recovery handling, source commit and "
|
|
"recovery-kit freshness. It never commits, pushes or rebuilds recovery automatically."
|
|
),
|
|
"inputSchema": {"type": "object", "properties": {}, "additionalProperties": False},
|
|
},
|
|
{
|
|
"name": "athena_close_maintenance_record",
|
|
"description": (
|
|
"WRITE TOOL for maintenance metadata only. Use after separate tools have verified "
|
|
"that the documentation change was committed/pushed, deployed to Athena and followed "
|
|
"by a newer recovery kit. The server checks source commit and recovery timestamp before "
|
|
"moving the record to resolved. It changes no documentation, Git or recovery data."
|
|
),
|
|
"inputSchema": {
|
|
"type": "object",
|
|
"properties": {
|
|
"proposal_id": {"type": "string", "pattern": "^[a-f0-9]{32}$"},
|
|
"git_commit": {"type": "string", "pattern": "^[a-f0-9]{40}$"},
|
|
"verification": {"type": "string", "minLength": 10, "maxLength": 1000},
|
|
"confirmation": {"type": "string", "minLength": 38, "maxLength": 64},
|
|
},
|
|
"required": ["proposal_id", "git_commit", "verification", "confirmation"],
|
|
"additionalProperties": False,
|
|
},
|
|
},
|
|
]
|
|
|
|
|
|
def now_iso() -> str:
|
|
return time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())
|
|
|
|
|
|
def json_text(value: Any) -> str:
|
|
return json.dumps(value, ensure_ascii=False, separators=(",", ":"))
|
|
|
|
|
|
def allowed_text_file(path: Path) -> bool:
|
|
return (
|
|
path.suffix.lower() in ALLOWED_TEXT_SUFFIXES
|
|
or path.name.startswith("Dockerfile")
|
|
or path.name in {"LLAMA_CPP_COMMIT"}
|
|
)
|
|
|
|
|
|
def safe_repo_path(relative: str) -> Path:
|
|
if not relative or relative.startswith("/") or "\\" in relative:
|
|
raise ValueError("path must be repository-relative")
|
|
parts = Path(relative).parts
|
|
if ".." in parts or any(part in EXCLUDED_PARTS for part in parts):
|
|
raise ValueError("path is outside the allowed source tree")
|
|
lowered = relative.lower()
|
|
if any(token in lowered for token in ("secret", "authorized_keys", ".env", "agekey")):
|
|
raise ValueError("secret-bearing paths are not exposed")
|
|
path = (REPO_ROOT / relative).resolve()
|
|
root = REPO_ROOT.resolve()
|
|
if root not in path.parents and path != root:
|
|
raise ValueError("path escapes repository")
|
|
if not path.is_file() or not allowed_text_file(path):
|
|
raise ValueError("path is not an allowed text source")
|
|
return path
|
|
|
|
|
|
def safe_doc_path(relative: str) -> Path:
|
|
match = re.fullmatch(r"docs/([A-Za-z0-9_.-]+\.md)", relative)
|
|
if not match:
|
|
raise ValueError("documentation updates are limited to docs/*.md")
|
|
path = (DOCS_ROOT / match.group(1)).resolve()
|
|
root = DOCS_ROOT.resolve()
|
|
if root not in path.parents:
|
|
raise ValueError("documentation path escapes docs root")
|
|
if path.exists() and path.is_symlink():
|
|
raise ValueError("symbolic links are not writable")
|
|
return path
|
|
|
|
|
|
def read_runtime() -> dict[str, Any]:
|
|
try:
|
|
data = json.loads(RUNTIME_FILE.read_text(encoding="utf-8"))
|
|
except (OSError, json.JSONDecodeError) as exc:
|
|
return {"available": False, "error": str(exc), "instruction": "Do not infer current runtime state."}
|
|
generated = int(data.get("generated_unix", 0))
|
|
age = max(0, int(time.time()) - generated) if generated else None
|
|
data["available"] = True
|
|
data["age_seconds"] = age
|
|
data["stale"] = age is None or age > 180
|
|
return data
|
|
|
|
|
|
def overview() -> dict[str, Any]:
|
|
path = REPO_ROOT / "docs/PLATFORM_OVERVIEW.md"
|
|
text = path.read_text(encoding="utf-8")[:MAX_DOCUMENT_CHARS]
|
|
runtime = read_runtime()
|
|
return {
|
|
"source": "docs/PLATFORM_OVERVIEW.md",
|
|
"source_hierarchy": [
|
|
"current specialist-tool evidence",
|
|
"bounded Athena runtime snapshot",
|
|
"CURRENT_REFERENCE.md and STANDARD_PROFILE_MATRIX.md",
|
|
"versioned source and runbooks",
|
|
"chat memory only as an unverified hint",
|
|
],
|
|
"runtime_snapshot": {k: runtime.get(k) for k in ("available", "generated_at", "age_seconds", "stale", "source_commit")},
|
|
"content": text,
|
|
"instruction": "Search or read the relevant source before proposing a change; verify mutable claims with a current tool.",
|
|
}
|
|
|
|
|
|
def current_state() -> dict[str, Any]:
|
|
data = read_runtime()
|
|
data["scope"] = "bounded metadata only; no logs, prompts, chats, environment values or secrets"
|
|
if data.get("stale"):
|
|
data["instruction"] = "Snapshot is stale. Do not claim current service state until a specialist tool verifies it."
|
|
return data
|
|
|
|
|
|
def external_services() -> dict[str, Any]:
|
|
catalog_path = REPO_ROOT / "config/service-catalog.json"
|
|
try:
|
|
catalog = json.loads(catalog_path.read_text(encoding="utf-8"))
|
|
except (OSError, json.JSONDecodeError) as exc:
|
|
return {
|
|
"available": False,
|
|
"error": str(exc),
|
|
"instruction": "The service inventory is unavailable. Do not infer that a replacement service is needed.",
|
|
}
|
|
|
|
results = []
|
|
for service in catalog.get("services", [])[:24]:
|
|
item = {key: service.get(key) for key in (
|
|
"id", "name", "location", "host", "address", "port", "protocol",
|
|
"specialist_tool", "purpose",
|
|
)}
|
|
address = str(service.get("address", ""))
|
|
port = int(service.get("port", 0))
|
|
started = time.monotonic()
|
|
reachable = False
|
|
http_status = None
|
|
error = None
|
|
try:
|
|
with socket.create_connection((address, port), timeout=2):
|
|
reachable = True
|
|
if service.get("probe") == "http-head":
|
|
path = str(service.get("probe_path", "/"))
|
|
url = f"{service.get('protocol', 'http')}://{address}:{port}{path}"
|
|
request = urllib.request.Request(url, method="HEAD", headers={"User-Agent": "MikeAI-Service-Catalog/1"})
|
|
try:
|
|
with urllib.request.urlopen(request, timeout=3) as response:
|
|
http_status = response.status
|
|
except urllib.error.HTTPError as exc:
|
|
http_status = exc.code
|
|
except (OSError, ValueError, urllib.error.URLError) as exc:
|
|
error = type(exc).__name__
|
|
item["check"] = {
|
|
"reachable": reachable,
|
|
"http_status": http_status,
|
|
"elapsed_ms": round((time.monotonic() - started) * 1000),
|
|
"error_class": error,
|
|
}
|
|
results.append(item)
|
|
return {
|
|
"available": True,
|
|
"source": "config/service-catalog.json",
|
|
"catalog_version": catalog.get("version"),
|
|
"updated": catalog.get("updated"),
|
|
"services": results,
|
|
"instruction": (
|
|
"Existing catalog entries are architecture constraints, not disposable suggestions. "
|
|
"If a check or specialist tool fails, report the gap and ask for direction; do not plan a duplicate backend."
|
|
),
|
|
}
|
|
|
|
|
|
def candidate_files() -> list[Path]:
|
|
files: list[Path] = []
|
|
for path in REPO_ROOT.rglob("*"):
|
|
try:
|
|
rel = path.relative_to(REPO_ROOT)
|
|
except ValueError:
|
|
continue
|
|
if not path.is_file() or any(part in EXCLUDED_PARTS for part in rel.parts):
|
|
continue
|
|
if not allowed_text_file(path):
|
|
continue
|
|
lowered = str(rel).lower()
|
|
if any(token in lowered for token in ("secret", "authorized_keys", ".env", "agekey")):
|
|
continue
|
|
files.append(path)
|
|
return files
|
|
|
|
|
|
def search_knowledge(arguments: dict[str, Any]) -> dict[str, Any]:
|
|
query = str(arguments.get("query", "")).strip()
|
|
if len(query) < 2:
|
|
raise ValueError("query is too short")
|
|
limit = max(1, min(MAX_SEARCH_RESULTS, int(arguments.get("max_results", 5))))
|
|
terms = [term for term in re.findall(r"[a-zA-Z0-9_.-]{2,}", query.lower()) if term]
|
|
scored: list[tuple[int, str, int, str]] = []
|
|
for path in candidate_files():
|
|
try:
|
|
lines = path.read_text(encoding="utf-8", errors="replace").splitlines()
|
|
except OSError:
|
|
continue
|
|
rel = str(path.relative_to(REPO_ROOT))
|
|
for index, line in enumerate(lines):
|
|
lower = line.lower()
|
|
score = sum(3 if term in rel.lower() else 1 for term in terms if term in lower or term in rel.lower())
|
|
if score:
|
|
excerpt = "\n".join(lines[max(0, index - 2): min(len(lines), index + 4)])[:1800]
|
|
scored.append((score, rel, index + 1, excerpt))
|
|
scored.sort(key=lambda item: (-item[0], item[1], item[2]))
|
|
seen: set[tuple[str, int]] = set()
|
|
results = []
|
|
for score, rel, line, excerpt in scored:
|
|
key = (rel, line // 20)
|
|
if key in seen:
|
|
continue
|
|
seen.add(key)
|
|
results.append({"path": rel, "line": line, "score": score, "excerpt": excerpt})
|
|
if len(results) >= limit:
|
|
break
|
|
return {"query": query, "count": len(results), "results": results, "instruction": "Read selected sources; do not treat search excerpts as current runtime proof."}
|
|
|
|
|
|
def read_source(arguments: dict[str, Any]) -> dict[str, Any]:
|
|
relative = str(arguments.get("path", ""))
|
|
path = safe_repo_path(relative)
|
|
start = max(1, int(arguments.get("start_line", 1)))
|
|
max_lines = max(1, min(300, int(arguments.get("max_lines", 160))))
|
|
lines = path.read_text(encoding="utf-8", errors="replace").splitlines()
|
|
selected = lines[start - 1:start - 1 + max_lines]
|
|
content = "\n".join(f"{start + i}: {line}" for i, line in enumerate(selected))
|
|
return {"path": relative, "start_line": start, "end_line": start + len(selected) - 1, "total_lines": len(lines), "truncated": start - 1 + len(selected) < len(lines), "content": content[:MAX_DOCUMENT_CHARS]}
|
|
|
|
|
|
WORKFLOWS = {
|
|
"mcp": ["platform/mcp/compose.yaml", "platform/mcp/README.md", "platform/mcp/athena_operator_mcp.py", "platform/hermes/skills/athena-operator/SKILL.md", "compose.yaml", "docs/COMPONENTS.md", "docs/SECURITY.md", "docs/PLATFORM_CONTEXT_MCP.md", "docs/QWEN_OPERATOR_CONTEXT.md"],
|
|
"model": ["config/install.env.example", "platform/models/manifest.example.yaml", "platform/profiles/", "docs/STANDARD_PROFILE_MATRIX.md", "docs/QWEN_OPERATOR_CONTEXT.md"],
|
|
"profile": ["platform/profiles/", "router/router_profiles.json", "platform/openwebui/install-models.sh", "docs/STANDARD_PROFILE_MATRIX.md"],
|
|
"tts": ["compose.yaml", "router/xtts_worker.py", "platform/scripts/rollback-tts-production.sh", "docs/XTTS_EVALUATION_2026-08-23.md"],
|
|
"stt": ["compose.yaml", "router/stt_worker.py", "docs/COMPONENTS.md"],
|
|
"vision": ["compose.yaml", "router/ai_profile_router.py", "docs/STANDARD_PROFILE_MATRIX.md"],
|
|
"image": ["compose.yaml", "router/image_worker.py", "docs/OPERATIONS.md"],
|
|
"openwebui": ["compose.yaml", "platform/openwebui/", "docs/OPERATIONS.md", "docs/DISASTER_RECOVERY.md"],
|
|
"network": ["compose.yaml", "platform/host/", "docs/SECURITY.md", "docs/WIREGUARD_HOME_PEER.md", "docs/EMERGENCY_UNI_ACCESS.md"],
|
|
"recovery": ["platform/recovery/", "docs/BARE_METAL_RECOVERY.md", "docs/DISASTER_RECOVERY.md", "docs/RECOVERY_REQUIREMENTS.md"],
|
|
"other": ["docs/PLATFORM_OVERVIEW.md", "docs/QWEN_OPERATOR_CONTEXT.md", "docs/OPERATIONS.md"],
|
|
}
|
|
|
|
|
|
def change_workflow(arguments: dict[str, Any]) -> dict[str, Any]:
|
|
kind = str(arguments.get("change_type", "other"))
|
|
if kind not in WORKFLOWS:
|
|
raise ValueError("unsupported change_type")
|
|
return {
|
|
"change_type": kind,
|
|
"read_first": WORKFLOWS[kind],
|
|
"mandatory_sequence": [
|
|
"Capture current state with the narrowest specialist tool.",
|
|
"Read relevant versioned sources and identify documentation drift.",
|
|
"Define rollback and protect SSH, LAN, WireGuard and the active inference path.",
|
|
"Prepare and apply source changes with Athena Operator operation file_update. Never clone the repository inside the sandbox and never request or copy an SSH key.",
|
|
"Validate syntax/configuration with Athena Operator operation run_checks and run a bounded synthetic test.",
|
|
"Deploy only named services with Athena Operator operation compose_deploy.",
|
|
"Verify service health and remote reachability without reading chats or private payloads.",
|
|
"Update PLATFORM_OVERVIEW/CURRENT_REFERENCE/QWEN_OPERATOR_CONTEXT and the affected runbook.",
|
|
"Commit and push only the explicitly selected changed paths with Athena Operator operation git_publish.",
|
|
"Create and verify a new encrypted recovery bundle and self-contained data-disk kit with Athena Operator operation recovery.",
|
|
],
|
|
"hard_boundaries": [
|
|
"This context MCP does not modify services, Docker, networking, models or secrets.",
|
|
"The sandbox needs neither a Git clone nor an SSH key; Athena Operator owns the canonical repository and deploy credentials.",
|
|
"No shutdown, reboot, kernel/driver, SSH, firewall or VPN change without exact user approval and rollback.",
|
|
"Never claim Git or recovery is current until separately verified.",
|
|
],
|
|
}
|
|
|
|
|
|
def prepare_update(arguments: dict[str, Any]) -> dict[str, Any]:
|
|
summary = str(arguments.get("summary", "")).strip()
|
|
evidence = str(arguments.get("evidence", "")).strip()
|
|
updates = arguments.get("updates")
|
|
if len(summary) < 5 or len(evidence) < 5 or not isinstance(updates, list) or not updates:
|
|
raise ValueError("summary, evidence and at least one update are required")
|
|
normalized = []
|
|
total = 0
|
|
for update in updates[:6]:
|
|
relative = str(update.get("path", ""))
|
|
safe_doc_path(relative)
|
|
content = str(update.get("content", ""))
|
|
if not content or len(content) > MAX_UPDATE_CHARS:
|
|
raise ValueError("invalid documentation content size")
|
|
if re.search(r"(?i)(BEGIN [A-Z ]*PRIVATE KEY|github_pat_[A-Za-z0-9_]+|GITHUB_PERSONAL_ACCESS_TOKEN\s*=\s*\S+)", content):
|
|
raise ValueError("probable secret material detected")
|
|
total += len(content)
|
|
if total > MAX_UPDATE_CHARS * 2:
|
|
raise ValueError("proposal is too large")
|
|
target = safe_doc_path(relative)
|
|
previous = target.read_text(encoding="utf-8") if target.exists() else ""
|
|
diff = "\n".join(difflib.unified_diff(previous.splitlines(), content.splitlines(), fromfile=f"a/{relative}", tofile=f"b/{relative}", lineterm=""))
|
|
normalized.append({"path": relative, "content": content, "before_sha256": hashlib.sha256(previous.encode()).hexdigest(), "after_sha256": hashlib.sha256(content.encode()).hexdigest(), "before_chars": len(previous), "after_chars": len(content), "diff_preview": diff[:12000]})
|
|
proposal_id = uuid.uuid4().hex
|
|
proposal = {"proposal_id": proposal_id, "created_at": now_iso(), "summary": summary, "evidence": evidence, "updates": normalized, "status": "pending"}
|
|
pending = STATE_ROOT / "pending"
|
|
pending.mkdir(parents=True, exist_ok=True)
|
|
(pending / f"{proposal_id}.json").write_text(json.dumps(proposal, ensure_ascii=False, indent=2), encoding="utf-8")
|
|
return {"proposal_id": proposal_id, "summary": summary, "files": [{k: item[k] for k in ("path", "before_sha256", "after_sha256", "before_chars", "after_chars", "diff_preview")} for item in normalized], "canonical_files_changed": False, "required_confirmation": f"APPLY {proposal_id}", "instruction": "Show this proposal to the user and wait for explicit approval. Do not call apply in the same autonomous tool sequence."}
|
|
|
|
|
|
def apply_update(arguments: dict[str, Any]) -> dict[str, Any]:
|
|
proposal_id = str(arguments.get("proposal_id", ""))
|
|
confirmation = str(arguments.get("confirmation", ""))
|
|
if not re.fullmatch(r"[a-f0-9]{32}", proposal_id):
|
|
raise ValueError("invalid proposal_id")
|
|
if confirmation != f"APPLY {proposal_id}":
|
|
raise ValueError("confirmation does not match the exact proposal")
|
|
if WRITE_MODE != "enabled":
|
|
raise PermissionError("documentation writes are in proposal-only mode")
|
|
proposal_path = STATE_ROOT / "pending" / f"{proposal_id}.json"
|
|
if not proposal_path.is_file():
|
|
raise ValueError("proposal not found or already applied")
|
|
proposal = json.loads(proposal_path.read_text(encoding="utf-8"))
|
|
backup_root = STATE_ROOT / "backups" / f"{int(time.time())}-{proposal_id}"
|
|
backup_root.mkdir(parents=True, exist_ok=False)
|
|
changed = []
|
|
for item in proposal["updates"]:
|
|
target = safe_doc_path(item["path"])
|
|
current = target.read_text(encoding="utf-8") if target.exists() else ""
|
|
current_hash = hashlib.sha256(current.encode()).hexdigest()
|
|
if current_hash != item["before_sha256"]:
|
|
raise RuntimeError(f"documentation drift after preview: {item['path']}")
|
|
if target.exists():
|
|
(backup_root / target.name).write_text(current, encoding="utf-8")
|
|
target.parent.mkdir(parents=True, exist_ok=True)
|
|
fd, temporary = tempfile.mkstemp(prefix=f".{target.name}.", dir=target.parent)
|
|
try:
|
|
with os.fdopen(fd, "w", encoding="utf-8") as handle:
|
|
handle.write(item["content"])
|
|
handle.flush()
|
|
os.fsync(handle.fileno())
|
|
os.chmod(temporary, 0o664)
|
|
os.replace(temporary, target)
|
|
finally:
|
|
if os.path.exists(temporary):
|
|
os.unlink(temporary)
|
|
changed.append(item["path"])
|
|
applied = STATE_ROOT / "applied"
|
|
applied.mkdir(parents=True, exist_ok=True)
|
|
proposal["status"] = "applied_docs_only"
|
|
proposal["applied_at"] = now_iso()
|
|
proposal["backup_dir"] = str(backup_root)
|
|
destination = applied / proposal_path.name
|
|
destination.write_text(json.dumps(proposal, ensure_ascii=False, indent=2), encoding="utf-8")
|
|
proposal_path.unlink()
|
|
return {
|
|
"documentation_applied": True,
|
|
"changed_files": changed,
|
|
"backup_dir": str(backup_root),
|
|
"git_commit_complete": False,
|
|
"git_push_complete": False,
|
|
"recovery_refresh_complete": False,
|
|
"required_next_steps": [
|
|
"Use an authorized Git tool to apply the same documentation change to the private source repository, review diff, commit and push.",
|
|
"Deploy the committed source back to Athena so .mike-ai-source-commit matches.",
|
|
"Create and verify a new encrypted recovery bundle and self-contained /data recovery kit.",
|
|
"Run athena_get_maintenance_status and the platform verification checklist.",
|
|
],
|
|
"instruction": "Do not say the platform is fully documented or recoverable until all three false fields are separately verified.",
|
|
}
|
|
|
|
|
|
def maintenance_status() -> dict[str, Any]:
|
|
pending_dir = STATE_ROOT / "pending"
|
|
applied_dir = STATE_ROOT / "applied"
|
|
pending = sorted(path.stem for path in pending_dir.glob("*.json")) if pending_dir.exists() else []
|
|
applied = sorted(applied_dir.glob("*.json"), key=lambda path: path.stat().st_mtime, reverse=True) if applied_dir.exists() else []
|
|
runtime = read_runtime()
|
|
latest_applied = None
|
|
if applied:
|
|
data = json.loads(applied[0].read_text(encoding="utf-8"))
|
|
latest_applied = {"proposal_id": data.get("proposal_id"), "summary": data.get("summary"), "applied_at": data.get("applied_at"), "status": data.get("status")}
|
|
return {
|
|
"pending_proposals": pending,
|
|
"latest_applied_documentation_change": latest_applied,
|
|
"source_commit": runtime.get("source_commit"),
|
|
"documentation_tree_sha256": runtime.get("documentation_tree_sha256"),
|
|
"recovery_kit": runtime.get("recovery_kit"),
|
|
"attention_required": bool(pending or latest_applied),
|
|
"instruction": "Applied records mean Git and recovery may still be stale; verify them with their dedicated workflow before clearing the maintenance debt.",
|
|
}
|
|
|
|
|
|
def close_maintenance(arguments: dict[str, Any]) -> dict[str, Any]:
|
|
proposal_id = str(arguments.get("proposal_id", ""))
|
|
git_commit = str(arguments.get("git_commit", ""))
|
|
confirmation = str(arguments.get("confirmation", ""))
|
|
verification = str(arguments.get("verification", "")).strip()
|
|
if not re.fullmatch(r"[a-f0-9]{32}", proposal_id):
|
|
raise ValueError("invalid proposal_id")
|
|
if not re.fullmatch(r"[a-f0-9]{40}", git_commit):
|
|
raise ValueError("invalid git_commit")
|
|
if confirmation != f"CLOSE {proposal_id}":
|
|
raise ValueError("confirmation does not match the exact record")
|
|
if len(verification) < 10:
|
|
raise ValueError("verification summary is required")
|
|
record = STATE_ROOT / "applied" / f"{proposal_id}.json"
|
|
if not record.is_file():
|
|
raise ValueError("applied maintenance record not found")
|
|
data = json.loads(record.read_text(encoding="utf-8"))
|
|
runtime = read_runtime()
|
|
if runtime.get("stale"):
|
|
raise RuntimeError("runtime snapshot is stale")
|
|
if runtime.get("source_commit") != git_commit:
|
|
raise RuntimeError("deployed source commit does not match the verified Git commit")
|
|
applied_at = int(calendar.timegm(time.strptime(data["applied_at"], "%Y-%m-%dT%H:%M:%SZ")))
|
|
recovery = runtime.get("recovery_kit") or {}
|
|
if not recovery.get("present") or int(recovery.get("modified_unix", 0)) <= applied_at:
|
|
raise RuntimeError("recovery kit is absent or older than the documentation change")
|
|
data.update({"status": "resolved", "resolved_at": now_iso(), "git_commit": git_commit, "verification": verification, "recovery_kit": recovery})
|
|
resolved = STATE_ROOT / "resolved"
|
|
resolved.mkdir(parents=True, exist_ok=True)
|
|
destination = resolved / record.name
|
|
destination.write_text(json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8")
|
|
record.unlink()
|
|
return {"resolved": True, "proposal_id": proposal_id, "git_commit": git_commit, "recovery_kit": recovery.get("target"), "instruction": "Maintenance debt is closed because deployed Git and a newer recovery kit were both verified."}
|
|
|
|
|
|
def call_tool(name: str, arguments: dict[str, Any]) -> str:
|
|
if name == "athena_get_overview":
|
|
result = overview()
|
|
elif name == "athena_get_current_state":
|
|
result = current_state()
|
|
elif name == "athena_get_external_services":
|
|
result = external_services()
|
|
elif name == "athena_search_knowledge":
|
|
result = search_knowledge(arguments)
|
|
elif name == "athena_read_source":
|
|
result = read_source(arguments)
|
|
elif name == "athena_get_change_workflow":
|
|
result = change_workflow(arguments)
|
|
elif name == "athena_prepare_documentation_update":
|
|
result = prepare_update(arguments)
|
|
elif name == "athena_apply_documentation_update":
|
|
result = apply_update(arguments)
|
|
elif name == "athena_get_maintenance_status":
|
|
result = maintenance_status()
|
|
elif name == "athena_close_maintenance_record":
|
|
result = close_maintenance(arguments)
|
|
else:
|
|
raise ValueError(f"unknown tool: {name}")
|
|
return json_text(result)
|
|
|
|
|
|
def response(request_id: Any, result: Any = None, error: dict[str, Any] | None = None) -> None:
|
|
payload: dict[str, Any] = {"jsonrpc": "2.0", "id": request_id}
|
|
payload["error" if error is not None else "result"] = error if error is not None else result
|
|
sys.stdout.write(json_text(payload) + "\n")
|
|
sys.stdout.flush()
|
|
|
|
|
|
def handle(message: dict[str, Any]) -> None:
|
|
method = message.get("method")
|
|
request_id = message.get("id")
|
|
if method == "initialize":
|
|
response(request_id, {"protocolVersion": message.get("params", {}).get("protocolVersion", "2024-11-05"), "capabilities": {"tools": {"listChanged": False}}, "serverInfo": {"name": "mike-ai-platform-context", "version": SERVER_VERSION}})
|
|
elif method == "tools/list":
|
|
response(request_id, {"tools": TOOLS})
|
|
elif method == "tools/call":
|
|
params = message.get("params", {})
|
|
try:
|
|
text = call_tool(str(params.get("name", "")), params.get("arguments") or {})
|
|
response(request_id, {"content": [{"type": "text", "text": text}], "structuredContent": json.loads(text), "isError": False})
|
|
except Exception as exc:
|
|
response(request_id, {"content": [{"type": "text", "text": f"ERROR: {exc}"}], "isError": True})
|
|
elif request_id is not None:
|
|
response(request_id, error={"code": -32601, "message": f"Method not found: {method}"})
|
|
|
|
|
|
def main() -> None:
|
|
STATE_ROOT.mkdir(parents=True, exist_ok=True)
|
|
for line in sys.stdin:
|
|
try:
|
|
if line.strip():
|
|
handle(json.loads(line))
|
|
except Exception as exc:
|
|
sys.stderr.write(f"MCP input error: {exc}\n")
|
|
sys.stderr.flush()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|