Add live Butler capabilities safety map

app.py

@@ -264,6 +264,7 @@ async def root():
         "openapi": "/openapi.json",
         "services": svc_list,
         "endpoints": {
+            "capabilities": "GET /capabilities - live machine-readable operation and safety map",
             "proxy": "GET/POST/PUT/DELETE /{service}/{path} - proxy to backend with auto-auth",
             "vm_list": "GET /vm/list",
             "vm_create": "POST /vm/create {node, ip, hostname, cores?, memory?, disk?}",
@@ -288,6 +289,80 @@ async def root():
 async def health():
     return {"status": "ok", "vault_items": len(_vault_cache), "services": len(SERVICES), "version": VERSION}
 
+
+def _schema_contains_property(node, property_name: str, components: dict, seen: set[str] | None = None) -> bool:
+    """Resolve local OpenAPI refs and look for a request property."""
+    seen = seen or set()
+    if isinstance(node, list):
+        return any(_schema_contains_property(item, property_name, components, seen) for item in node)
+    if not isinstance(node, dict):
+        return False
+    if node.get("name") == property_name or property_name in node.get("properties", {}):
+        return True
+    ref = node.get("$ref", "")
+    if ref.startswith("#/components/schemas/"):
+        name = ref.rsplit("/", 1)[-1]
+        if name in seen:
+            return False
+        return _schema_contains_property(components.get(name, {}), property_name, components, seen | {name})
+    return any(
+        _schema_contains_property(value, property_name, components, seen)
+        for key, value in node.items()
+        if key != "properties"
+    )
+
+
+@app.get("/capabilities")
+async def capabilities(_=Depends(_verify)):
+    """Live operation catalog with safety metadata for AI agents."""
+    schema = app.openapi()
+    components = schema.get("components", {}).get("schemas", {})
+    operations = []
+    for path, methods in schema.get("paths", {}).items():
+        if path == "/{service}/{path}" or path in {"/", "/health", "/openapi.json", "/docs", "/redoc"}:
+            continue
+        for method, operation in methods.items():
+            if method.upper() not in {"GET", "POST", "PUT", "PATCH", "DELETE"}:
+                continue
+            mode = "read_only" if method.upper() == "GET" else "mutation"
+            serialized = {"parameters": operation.get("parameters", []), "requestBody": operation.get("requestBody", {})}
+            dry_run = _schema_contains_property(serialized, "dry_run", components)
+            critical = path.startswith(("/network/wireguard", "/caddy/"))
+            destructive = method.upper() == "DELETE" or any(
+                marker in path for marker in ("/destroy/", "/cleanup/", "/break-lock/", "/restore/")
+            )
+            operations.append({
+                "method": method.upper(),
+                "path": path,
+                "summary": operation.get("summary", ""),
+                "description": operation.get("description", ""),
+                "mode": mode,
+                "dry_run": dry_run,
+                "critical": critical,
+                "destructive": destructive,
+                "confirmation_required": mode == "mutation",
+            })
+    operations.sort(key=lambda item: (item["path"], item["method"]))
+    counts = {
+        "total": len(operations),
+        "read_only": sum(item["mode"] == "read_only" for item in operations),
+        "mutations": sum(item["mode"] == "mutation" for item in operations),
+        "destructive": sum(item["destructive"] for item in operations),
+    }
+    return {
+        "schema_version": 1,
+        "service": "homelab-butler",
+        "version": VERSION,
+        "generated": datetime.now(timezone.utc).isoformat(),
+        "counts": counts,
+        "operations": operations,
+        "proxy": {"path": "/{service}/{path}", "note": "Generic backend proxy; inspect /info services and OpenAPI before use"},
+        "model_contract": {
+            "instruction": "Prefer read_only operations. Before every mutation inspect its schema, use dry_run when available, and obtain confirmation for critical or destructive actions.",
+            "source_of_truth": "/openapi.json",
+        },
+    }
+
 def _classify_http_status(status_code: int, expected: set[int]) -> str:
     """Return a deterministic service state suitable for small models."""
     if status_code in expected:
@@ -408,6 +483,7 @@ async def info(_=Depends(_verify)):
             for name, cfg in SERVICES.items()
         },
         "endpoints": {
+            "capabilities": "/capabilities",
             "status": "/status",
             "overview": "/overview?details=false",
             "audit": "/audit",