From 4eacc756b04bf8dbca01501db4e8168143f917c4 Mon Sep 17 00:00:00 2001 From: Chris Christiansen Date: Fri, 18 Sep 2026 15:34:46 +0000 Subject: [PATCH] feat(emma): add canonical capability registry --- opax-mcp/capability_registry.py | 246 +++++++++++++++++++++++++++ opax-mcp/test_capability_registry.py | 140 +++++++++++++++ 2 files changed, 386 insertions(+) create mode 100644 opax-mcp/capability_registry.py create mode 100644 opax-mcp/test_capability_registry.py diff --git a/opax-mcp/capability_registry.py b/opax-mcp/capability_registry.py new file mode 100644 index 0000000..a3d4599 --- /dev/null +++ b/opax-mcp/capability_registry.py @@ -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 + ) diff --git a/opax-mcp/test_capability_registry.py b/opax-mcp/test_capability_registry.py new file mode 100644 index 0000000..7a28841 --- /dev/null +++ b/opax-mcp/test_capability_registry.py @@ -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()