feat(emma): add canonical capability registry
This commit is contained in:
parent
e48f8390ed
commit
4eacc756b0
246
opax-mcp/capability_registry.py
Normal file
246
opax-mcp/capability_registry.py
Normal file
|
|
@ -0,0 +1,246 @@
|
|||
"""Canonical, declarative Capability Registry v1 for Emma."""
|
||||
|
||||
from dataclasses import dataclass
|
||||
from enum import Enum
|
||||
from types import MappingProxyType
|
||||
from typing import Mapping, Tuple
|
||||
|
||||
|
||||
class RiskLevel(str, Enum):
|
||||
READ = "read"
|
||||
WRITE = "write"
|
||||
HIGH_IMPACT = "high_impact"
|
||||
FORBIDDEN = "forbidden"
|
||||
|
||||
|
||||
class Availability(str, Enum):
|
||||
ACTIVE = "active"
|
||||
PLANNED = "planned"
|
||||
FORBIDDEN = "forbidden"
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class Capability:
|
||||
id: str
|
||||
display_name: str
|
||||
description: str
|
||||
category: str
|
||||
risk_level: RiskLevel
|
||||
availability: Availability
|
||||
execution_owner: str
|
||||
backend_tool: str | None
|
||||
required_actor_scope: Tuple[str, ...]
|
||||
input_schema: Mapping[str, str]
|
||||
output_schema: Mapping[str, str]
|
||||
approval_required: bool
|
||||
approval_binding: str | None
|
||||
audit_required: bool
|
||||
rollback_required: bool
|
||||
self_approval_forbidden: bool
|
||||
allowed_targets: Tuple[str, ...]
|
||||
external_source_policy: str | None
|
||||
|
||||
|
||||
_EMPTY_SCHEMA: Mapping[str, str] = MappingProxyType({})
|
||||
|
||||
|
||||
def _capability(
|
||||
capability_id: str,
|
||||
display_name: str,
|
||||
description: str,
|
||||
category: str,
|
||||
risk_level: RiskLevel,
|
||||
availability: Availability,
|
||||
execution_owner: str,
|
||||
backend_tool: str | None,
|
||||
required_actor_scope: Tuple[str, ...],
|
||||
approval_required: bool,
|
||||
approval_binding: str | None,
|
||||
audit_required: bool,
|
||||
rollback_required: bool,
|
||||
self_approval_forbidden: bool,
|
||||
allowed_targets: Tuple[str, ...],
|
||||
external_source_policy: str | None,
|
||||
) -> Capability:
|
||||
return Capability(
|
||||
id=capability_id,
|
||||
display_name=display_name,
|
||||
description=description,
|
||||
category=category,
|
||||
risk_level=risk_level,
|
||||
availability=availability,
|
||||
execution_owner=execution_owner,
|
||||
backend_tool=backend_tool,
|
||||
required_actor_scope=required_actor_scope,
|
||||
input_schema=_EMPTY_SCHEMA,
|
||||
output_schema=_EMPTY_SCHEMA,
|
||||
approval_required=approval_required,
|
||||
approval_binding=approval_binding,
|
||||
audit_required=audit_required,
|
||||
rollback_required=rollback_required,
|
||||
self_approval_forbidden=self_approval_forbidden,
|
||||
allowed_targets=allowed_targets,
|
||||
external_source_policy=external_source_policy,
|
||||
)
|
||||
|
||||
|
||||
_CAPABILITIES_TUPLE: Tuple[Capability, ...] = (
|
||||
_capability(
|
||||
"emma.chat", "Emma chat", "Conversational reasoning through Emma.",
|
||||
"conversation", RiskLevel.READ, Availability.ACTIVE, "opax-mcp", "run_emma",
|
||||
("authenticated_user",), False, None, True, False, False,
|
||||
("opax.vauco.no",), "none",
|
||||
),
|
||||
_capability(
|
||||
"emma.analyze_user_text", "Analyze user-provided text",
|
||||
"Analyze, summarize, and structure text provided in the current request.",
|
||||
"conversation", RiskLevel.READ, Availability.ACTIVE, "opax-mcp", "run_emma",
|
||||
("authenticated_user",), False, None, True, False, False,
|
||||
("user_provided_text",), "user_provided_only",
|
||||
),
|
||||
_capability(
|
||||
"vauco.context.retrieve", "Retrieve approved VAUCO context",
|
||||
"Retrieve approved VAUCO internal context.", "context", RiskLevel.READ,
|
||||
Availability.PLANNED, "opax-mcp", None,
|
||||
("authenticated_user", "vauco_context_read"), False, None, True, False, False,
|
||||
("approved_vauco_sources",), "approved_internal_only",
|
||||
),
|
||||
_capability(
|
||||
"conversation.session_history", "Retrieve active session history",
|
||||
"Retrieve bounded history for the active user session.", "context", RiskLevel.READ,
|
||||
Availability.PLANNED, "opax-web", None,
|
||||
("authenticated_user", "conversation_read"), False, None, True, False, False,
|
||||
("active_user_session",), "session_scoped",
|
||||
),
|
||||
_capability(
|
||||
"memory.morphic.read", "Read scoped Morphic memory",
|
||||
"Read actor-scoped Morphic memory.", "memory", RiskLevel.READ,
|
||||
Availability.PLANNED, "opax-mcp", None,
|
||||
("authenticated_user", "morphic_memory_read"), False, None, True, False, False,
|
||||
("actor_scoped_memory",), "actor_scoped",
|
||||
),
|
||||
_capability(
|
||||
"gitea.read_commits", "Read Gitea commits",
|
||||
"Read commit history from the authoritative Gitea service.", "gitea", RiskLevel.READ,
|
||||
Availability.PLANNED, "opax-mcp", "list_commits",
|
||||
("authenticated_user", "gitea_read"), False, None, True, False, False,
|
||||
("git.vauco.no",), "internal_authoritative_only",
|
||||
),
|
||||
_capability(
|
||||
"gitea.read_file", "Read Gitea file",
|
||||
"Read a file from the authoritative Gitea service.", "gitea", RiskLevel.READ,
|
||||
Availability.PLANNED, "opax-mcp", "get_file",
|
||||
("authenticated_user", "gitea_read"), False, None, True, False, False,
|
||||
("git.vauco.no",), "internal_authoritative_only",
|
||||
),
|
||||
_capability(
|
||||
"cloudbuild.read_status", "Read Cloud Build status",
|
||||
"Read Cloud Build status in the approved VAUCO project.", "cloudbuild", RiskLevel.READ,
|
||||
Availability.PLANNED, "opax-mcp", None,
|
||||
("authenticated_user", "cloudbuild_read"), False, None, True, False, False,
|
||||
("propane-will-491900-m5",), "internal_project_only",
|
||||
),
|
||||
_capability(
|
||||
"cloudrun.read_status", "Read Cloud Run status",
|
||||
"Read Cloud Run status in the approved VAUCO project.", "cloudrun", RiskLevel.READ,
|
||||
Availability.PLANNED, "opax-mcp", None,
|
||||
("authenticated_user", "cloudrun_read"), False, None, True, False, False,
|
||||
("propane-will-491900-m5", "us-central1"), "internal_project_only",
|
||||
),
|
||||
_capability(
|
||||
"github.intake.read", "Read reviewed external GitHub repository",
|
||||
"Read an explicitly requested external repository in read-only quarantine.",
|
||||
"github_intake", RiskLevel.READ, Availability.PLANNED, "opax-mcp", None,
|
||||
("authenticated_user", "github_intake_read"), False, None, True, False, False,
|
||||
("explicitly_requested_repository",), "explicit_read_only_quarantine",
|
||||
),
|
||||
_capability(
|
||||
"gitea.create_issue", "Create Gitea issue",
|
||||
"Create an approved Gitea issue.", "gitea", RiskLevel.WRITE,
|
||||
Availability.PLANNED, "opax-mcp", "create_issue",
|
||||
("authenticated_user", "gitea_write"), True, "exact_repository_title_body", True,
|
||||
False, True, ("git.vauco.no",), "internal_authoritative_only",
|
||||
),
|
||||
_capability(
|
||||
"gitea.push_change", "Push approved Gitea change",
|
||||
"Push an approved change to Gitea.", "gitea", RiskLevel.WRITE,
|
||||
Availability.PLANNED, "opax-mcp", "push_file",
|
||||
("authenticated_user", "gitea_write"), True,
|
||||
"exact_repository_branch_path_content_sha", True, True, True,
|
||||
("git.vauco.no",), "internal_authoritative_only",
|
||||
),
|
||||
_capability(
|
||||
"cloudbuild.trigger", "Trigger approved Cloud Build",
|
||||
"Trigger an approved Cloud Build.", "cloudbuild", RiskLevel.HIGH_IMPACT,
|
||||
Availability.PLANNED, "opax-mcp", "trigger_build",
|
||||
("authenticated_user", "cloudbuild_execute"), True,
|
||||
"exact_build_source_config_substitutions", True, False, True,
|
||||
("propane-will-491900-m5",), "internal_project_only",
|
||||
),
|
||||
_capability(
|
||||
"cloudrun.deploy_digest", "Deploy approved Cloud Run image digest",
|
||||
"Deploy an approved immutable image digest to Cloud Run.", "cloudrun",
|
||||
RiskLevel.HIGH_IMPACT, Availability.PLANNED, "opax-mcp", "build_and_deploy_service",
|
||||
("authenticated_user", "cloudrun_deploy"), True,
|
||||
"exact_service_region_image_digest_rollback_revision", True, True, True,
|
||||
("propane-will-491900-m5", "us-central1"), "internal_project_only",
|
||||
),
|
||||
_capability(
|
||||
"emma.update_ui", "Update Emma UI",
|
||||
"Update Emma UI through the approved self-update flow.", "self_update",
|
||||
RiskLevel.HIGH_IMPACT, Availability.PLANNED, "opax-mcp", None,
|
||||
("authenticated_user", "emma_self_update"), True,
|
||||
"exact_repository_branch_diff_tests_image_digest_service", True, True, True,
|
||||
("git.vauco.no", "opax.vauco.no"), "internal_authoritative_only",
|
||||
),
|
||||
_capability(
|
||||
"emma.update_backend", "Update Emma backend",
|
||||
"Update Emma backend through the approved self-update flow.", "self_update",
|
||||
RiskLevel.HIGH_IMPACT, Availability.PLANNED, "opax-mcp", None,
|
||||
("authenticated_user", "emma_self_update"), True,
|
||||
"exact_repository_branch_diff_tests_image_digest_service", True, True, True,
|
||||
("git.vauco.no", "opax-mcp"), "internal_authoritative_only",
|
||||
),
|
||||
_capability(
|
||||
"emma.update_model_config", "Update Emma model configuration",
|
||||
"Update model configuration through the approved self-update flow.", "self_update",
|
||||
RiskLevel.HIGH_IMPACT, Availability.PLANNED, "opax-mcp", None,
|
||||
("authenticated_user", "emma_self_update"), True,
|
||||
"exact_model_config_diff_tests_target", True, True, True,
|
||||
("opax-mcp", "os-vauco-agent"), "internal_project_only",
|
||||
),
|
||||
_capability(
|
||||
"emma.cli", "Emma CLI",
|
||||
"Use Emma through a CLI governed by the same policy as the UI.", "interface",
|
||||
RiskLevel.READ, Availability.PLANNED, "opax-mcp", None,
|
||||
("authenticated_user",), False, None, True, False, False,
|
||||
("approved_vauco_operator_environment",), "same_policy_as_emma_ui",
|
||||
),
|
||||
_capability(
|
||||
"terminal.arbitrary_shell", "Arbitrary terminal shell",
|
||||
"Arbitrary terminal execution is prohibited.", "terminal", RiskLevel.FORBIDDEN,
|
||||
Availability.FORBIDDEN, "none", None, (), False, None, True, False, True,
|
||||
(), "prohibited",
|
||||
),
|
||||
)
|
||||
|
||||
|
||||
CAPABILITIES: Mapping[str, Capability] = MappingProxyType(
|
||||
{capability.id: capability for capability in _CAPABILITIES_TUPLE}
|
||||
)
|
||||
|
||||
|
||||
def get_capability(capability_id: str) -> Capability | None:
|
||||
return CAPABILITIES.get(capability_id)
|
||||
|
||||
|
||||
def list_capabilities() -> Tuple[Capability, ...]:
|
||||
return _CAPABILITIES_TUPLE
|
||||
|
||||
|
||||
def list_active_capabilities() -> Tuple[Capability, ...]:
|
||||
return tuple(
|
||||
capability
|
||||
for capability in _CAPABILITIES_TUPLE
|
||||
if capability.availability == Availability.ACTIVE
|
||||
)
|
||||
140
opax-mcp/test_capability_registry.py
Normal file
140
opax-mcp/test_capability_registry.py
Normal file
|
|
@ -0,0 +1,140 @@
|
|||
import ast
|
||||
import sys
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
|
||||
REPO_ROOT = Path(__file__).resolve().parents[1]
|
||||
sys.path.insert(0, str(REPO_ROOT / "opax-mcp"))
|
||||
|
||||
from capability_registry import (
|
||||
CAPABILITIES,
|
||||
Availability,
|
||||
RiskLevel,
|
||||
get_capability,
|
||||
list_active_capabilities,
|
||||
list_capabilities,
|
||||
)
|
||||
|
||||
|
||||
class TestCapabilityRegistry(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.capabilities = list_capabilities()
|
||||
|
||||
def test_registry_contains_exactly_nineteen_capabilities(self):
|
||||
self.assertEqual(len(self.capabilities), 19)
|
||||
self.assertEqual(len(CAPABILITIES), 19)
|
||||
|
||||
def test_identifiers_are_unique(self):
|
||||
identifiers = [capability.id for capability in self.capabilities]
|
||||
self.assertEqual(len(identifiers), len(set(identifiers)))
|
||||
|
||||
def test_order_is_deterministic_and_starts_with_active_capabilities(self):
|
||||
self.assertEqual(
|
||||
[capability.id for capability in self.capabilities[:2]],
|
||||
["emma.chat", "emma.analyze_user_text"],
|
||||
)
|
||||
self.assertEqual(
|
||||
[capability.id for capability in self.capabilities],
|
||||
list(CAPABILITIES.keys()),
|
||||
)
|
||||
|
||||
def test_active_capabilities_are_exactly_the_two_permitted_capabilities(self):
|
||||
active = list_active_capabilities()
|
||||
self.assertEqual(
|
||||
[capability.id for capability in active],
|
||||
["emma.chat", "emma.analyze_user_text"],
|
||||
)
|
||||
for capability in active:
|
||||
self.assertEqual(capability.risk_level, RiskLevel.READ)
|
||||
self.assertFalse(capability.approval_required)
|
||||
self.assertFalse(capability.self_approval_forbidden)
|
||||
|
||||
def test_write_and_high_impact_capabilities_are_planned_and_gated(self):
|
||||
gated = [
|
||||
capability
|
||||
for capability in self.capabilities
|
||||
if capability.risk_level in {RiskLevel.WRITE, RiskLevel.HIGH_IMPACT}
|
||||
]
|
||||
self.assertTrue(gated)
|
||||
for capability in gated:
|
||||
self.assertEqual(capability.availability, Availability.PLANNED)
|
||||
self.assertTrue(capability.approval_required)
|
||||
self.assertTrue(capability.audit_required)
|
||||
self.assertTrue(capability.self_approval_forbidden)
|
||||
|
||||
def test_forbidden_shell_capability_has_no_execution_target(self):
|
||||
capability = get_capability("terminal.arbitrary_shell")
|
||||
self.assertIsNotNone(capability)
|
||||
self.assertEqual(capability.risk_level, RiskLevel.FORBIDDEN)
|
||||
self.assertEqual(capability.availability, Availability.FORBIDDEN)
|
||||
self.assertIsNone(capability.backend_tool)
|
||||
self.assertFalse(capability.approval_required)
|
||||
self.assertTrue(capability.self_approval_forbidden)
|
||||
self.assertEqual(capability.allowed_targets, ())
|
||||
|
||||
def test_unknown_capability_returns_none(self):
|
||||
self.assertIsNone(get_capability("unknown"))
|
||||
|
||||
def test_public_registry_mapping_is_immutable(self):
|
||||
with self.assertRaises(TypeError):
|
||||
CAPABILITIES["new.capability"] = get_capability("emma.chat")
|
||||
|
||||
def test_schema_mappings_are_immutable(self):
|
||||
for capability in self.capabilities:
|
||||
with self.assertRaises(TypeError):
|
||||
capability.input_schema["unexpected"] = "value"
|
||||
with self.assertRaises(TypeError):
|
||||
capability.output_schema["unexpected"] = "value"
|
||||
|
||||
def test_registry_module_has_no_external_execution_imports(self):
|
||||
module_path = REPO_ROOT / "opax-mcp" / "capability_registry.py"
|
||||
tree = ast.parse(module_path.read_text(encoding="utf-8"))
|
||||
forbidden_roots = {
|
||||
"os",
|
||||
"subprocess",
|
||||
"requests",
|
||||
"httpx",
|
||||
"google",
|
||||
"gitea",
|
||||
"firebase",
|
||||
"firestore",
|
||||
}
|
||||
imported_roots = set()
|
||||
for node in ast.walk(tree):
|
||||
if isinstance(node, ast.Import):
|
||||
imported_roots.update(alias.name.split(".")[0] for alias in node.names)
|
||||
if isinstance(node, ast.ImportFrom) and node.module:
|
||||
imported_roots.add(node.module.split(".")[0])
|
||||
self.assertFalse(imported_roots & forbidden_roots)
|
||||
|
||||
def test_active_and_planned_capabilities_have_required_metadata(self):
|
||||
for capability in self.capabilities:
|
||||
if capability.availability == Availability.FORBIDDEN:
|
||||
continue
|
||||
self.assertTrue(capability.id)
|
||||
self.assertTrue(capability.display_name)
|
||||
self.assertTrue(capability.description)
|
||||
self.assertTrue(capability.category)
|
||||
self.assertTrue(capability.execution_owner)
|
||||
self.assertTrue(capability.allowed_targets)
|
||||
|
||||
def test_planned_write_and_high_impact_bindings_and_rollbacks(self):
|
||||
rollback_required_ids = {
|
||||
"gitea.push_change",
|
||||
"cloudrun.deploy_digest",
|
||||
"emma.update_ui",
|
||||
"emma.update_backend",
|
||||
"emma.update_model_config",
|
||||
}
|
||||
for capability in self.capabilities:
|
||||
if capability.risk_level not in {RiskLevel.WRITE, RiskLevel.HIGH_IMPACT}:
|
||||
continue
|
||||
self.assertTrue(capability.approval_binding)
|
||||
self.assertEqual(
|
||||
capability.rollback_required,
|
||||
capability.id in rollback_required_ids,
|
||||
)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Loading…
Reference in New Issue
Block a user