Add live capabilities and safety map (#40)

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",

tests/test_app.py

@@ -46,6 +46,33 @@ def test_health_exposes_current_version():
     assert response.json()["version"] == app.VERSION == "2.3.5"
 
 
+def test_capabilities_is_live_machine_readable_safety_map():
+    with TestClient(app.app) as client:
+        response = client.get(
+            "/capabilities",
+            headers={"Authorization": "Bearer test-token"},
+        )
+    assert response.status_code == 200
+    payload = response.json()
+    by_operation = {(item["method"], item["path"]): item for item in payload["operations"]}
+    assert ("GET", "/network/wireguard/{host}") in by_operation
+    assert by_operation[("GET", "/network/wireguard/{host}")]["mode"] == "read_only"
+    removal = by_operation[("DELETE", "/network/wireguard/{host}/peer")]
+    assert removal["mode"] == "mutation"
+    assert removal["dry_run"] is True
+    assert removal["critical"] is True
+    assert by_operation[("DELETE", "/vm/destroy/{vmid}")]["dry_run"] is True
+    assert all(item["path"] != "/{service}/{path}" for item in payload["operations"])
+    assert payload["model_contract"]["instruction"].startswith("Prefer read_only")
+
+
+def test_info_advertises_capabilities_endpoint():
+    with TestClient(app.app) as client:
+        response = client.get("/info", headers={"Authorization": "Bearer test-token"})
+    assert response.status_code == 200
+    assert response.json()["endpoints"]["capabilities"] == "/capabilities"
+
+
 def test_wireguard_status_returns_redacted_live_state(monkeypatch):
     payload = {
         "interface": "wg0",