From 97421374c2bfe664b74953a48f69ebc1607441e3 Mon Sep 17 00:00:00 2001 From: Chris Christiansen Date: Fri, 18 Sep 2026 15:58:04 +0000 Subject: [PATCH] feat(emma): bridge capability status into runtime --- opax-mcp/capability_bridge.py | 56 ++++++++++ opax-mcp/emma_adapter.py | 12 ++- opax-mcp/server.py | 3 + opax-mcp/test_capability_bridge.py | 165 +++++++++++++++++++++++++++++ opax-mcp/test_emma_adapter.py | 47 ++++++++ 5 files changed, 280 insertions(+), 3 deletions(-) create mode 100644 opax-mcp/capability_bridge.py create mode 100644 opax-mcp/test_capability_bridge.py diff --git a/opax-mcp/capability_bridge.py b/opax-mcp/capability_bridge.py new file mode 100644 index 0000000..fcdcda5 --- /dev/null +++ b/opax-mcp/capability_bridge.py @@ -0,0 +1,56 @@ +""" +Builds a user-safe, read-only capability summary for the Emma system prompt. +""" + +from typing import List, Dict +from capability_registry import Availability, Capability, list_capabilities + + +def build_capability_system_context() -> str: + """ + Builds a deterministic, user-safe, plain-text summary of capabilities. + + The output is structured for inclusion in a system prompt and must not leak + internal details. + """ + categorized: Dict[Availability, List[str]] = { + Availability.ACTIVE: [], + Availability.PLANNED: [], + Availability.FORBIDDEN: [], + } + + for capability in list_capabilities(): + if capability.availability == Availability.FORBIDDEN: + if capability.id == "terminal.arbitrary_shell": + categorized[Availability.FORBIDDEN].append( + "- Arbitrary terminal access is not available." + ) + continue + + if capability.availability in categorized: + categorized[capability.availability].append( + f"- {capability.display_name}: {capability.description}" + ) + + output_lines = ["CURRENT CAPABILITY STATUS FOR THIS CHAT"] + + if categorized[Availability.ACTIVE]: + output_lines.append("\nAvailable now:") + output_lines.extend(categorized[Availability.ACTIVE]) + + if categorized[Availability.PLANNED]: + output_lines.append("\nPlanned, but not active in this chat:") + output_lines.extend(categorized[Availability.PLANNED]) + + if categorized[Availability.FORBIDDEN]: + output_lines.append("\nUnavailable:") + output_lines.extend(categorized[Availability.FORBIDDEN]) + + output_lines.append( + "\nRules:\n" + "- Planned capabilities are not available in this chat and must not be claimed as active.\n" + "- No tool, repository, deployment, memory, document retrieval, external service or state-changing action is activated by this capability summary.\n" + "- Future write or high-impact capabilities require a concrete, explicit, bound user approval before execution." + ) + + return "\n".join(output_lines) diff --git a/opax-mcp/emma_adapter.py b/opax-mcp/emma_adapter.py index ac305e6..2db740c 100644 --- a/opax-mcp/emma_adapter.py +++ b/opax-mcp/emma_adapter.py @@ -22,7 +22,7 @@ class CanonicalEmma: self._model = model self._system_prompt = system_prompt or get_runtime_system_prompt() - async def run(self, prompt, history=None): + async def run(self, prompt, history=None, system_context: str | None = None): """ Runs the Emma agent with the given prompt. @@ -30,15 +30,21 @@ class CanonicalEmma: ensuring the canonical system prompt is used. History is ignored for now. Args: - prompt: The user's prompt. + prompt: The user\'s prompt. history: The conversation history (currently ignored). + system_context: Optional context to append to the system prompt. Returns: The raw dictionary response from the chat_function. """ # Phase 1 does not use history or memory. + + composed_system_prompt = self._system_prompt + if system_context: + composed_system_prompt = f"{self._system_prompt}\n\n{system_context.strip()}" + return await self._chat_function( self._model, prompt, - self._system_prompt, + composed_system_prompt, ) diff --git a/opax-mcp/server.py b/opax-mcp/server.py index cde624a..a7d87d3 100644 --- a/opax-mcp/server.py +++ b/opax-mcp/server.py @@ -58,6 +58,7 @@ except Exception as e: print(f"Failed to load provision_new_mcp_module: {e}") provision_new_mcp_module = None from emma_adapter import CanonicalEmma +from capability_bridge import build_capability_system_context from email.mime.text import MIMEText from datetime import datetime, timezone, timedelta from typing import Any, Optional, Dict, List @@ -874,9 +875,11 @@ async def run_jason(p): return await _agent_post("/run", {"message": p.get("prompt", p.get("message", "")), "user_id": p.get("user_id", "opax"), "session_id": p.get("session_id", "mcp"), "mode": p.get("mode", "light")}) async def run_emma(p: dict) -> dict: """Kaller den kanoniske Emma-agenten med en prompt.""" + capability_system_context = build_capability_system_context() return await canonical_emma.run( prompt=p.get("prompt", p.get("message", "")), history=[], + system_context=capability_system_context, ) async def run_emma_fast(p): return await _ollama_chat(EMMA_FAST_MODEL, p.get("prompt", p.get("message", "")), p.get("system", "Du er en rask og konsis AI-assistent...")) async def run_qwen(p): return await _ollama_chat(EMMA_LIGHT_MODEL, p.get("prompt", p.get("message", ""))) diff --git a/opax-mcp/test_capability_bridge.py b/opax-mcp/test_capability_bridge.py new file mode 100644 index 0000000..071e674 --- /dev/null +++ b/opax-mcp/test_capability_bridge.py @@ -0,0 +1,165 @@ +import unittest +import ast +from pathlib import Path +import sys + +# Ensure the project root is in the path for imports +REPO_ROOT = Path(__file__).resolve().parents[1] +sys.path.insert(0, str(REPO_ROOT / "opax-mcp")) + +from capability_bridge import build_capability_system_context +from capability_registry import ( + list_capabilities, + Capability, + Availability, +) + + +class TestCapabilityBridge(unittest.TestCase): + + def setUp(self): + self.context_output = build_capability_system_context() + + def test_output_is_deterministic(self): + self.assertEqual(self.context_output, build_capability_system_context()) + + def test_output_has_all_three_headings(self): + self.assertIn("\nAvailable now:", self.context_output) + self.assertIn("\nPlanned, but not active in this chat:", self.context_output) + self.assertIn("\nUnavailable:", self.context_output) + + def test_available_section_contains_correct_capabilities(self): + active_capabilities = [ + c for c in list_capabilities() if c.availability == Availability.ACTIVE + ] + self.assertIn("- Emma chat: Conversational reasoning through Emma.", self.context_output) + self.assertIn("- Analyze user-provided text: Analyze, summarize, and structure text provided in the current request.", self.context_output) + + # Isolate the "Available now" section to count items accurately + start_index = self.context_output.find("Available now:") + end_index = self.context_output.find("\n\nPlanned, but not active in this chat:") + available_section = self.context_output[start_index:end_index] + + self.assertEqual(available_section.count("- "), len(active_capabilities)) + + def test_planned_section_contains_all_planned_capabilities(self): + planned_capabilities = [ + c for c in list_capabilities() if c.availability == Availability.PLANNED + ] + self.assertEqual(len(planned_capabilities), 16) + for capability in planned_capabilities: + self.assertIn(f"- {capability.display_name}: {capability.description}", self.context_output) + + def test_unavailable_section_is_correct(self): + self.assertIn("- Arbitrary terminal access is not available.", self.context_output) + # Ensure no other forbidden capabilities are listed by name + self.assertNotIn("Arbitrary terminal shell:", self.context_output) + + def test_bridge_text_does_not_contain_forbidden_internal_details(self): + forbidden_substrings = [ + # IDs + "emma.chat", "terminal.arbitrary_shell", + # Backend tool names + "run_emma", "list_commits", "get_file", "create_issue", "push_file", + "build_and_deploy_service", + # Internal metadata + "required_actor_scope", "approval_binding", "audit_required", + "rollback_required", "self_approval_forbidden", + # Target names / infrastructure + "git.vauco.no", "opax-mcp", "propane-will-491900-m5", "us-central1", + ] + for substring in forbidden_substrings: + self.assertNotIn(substring, self.context_output) + + def test_bridge_module_has_no_prohibited_imports(self): + bridge_path = REPO_ROOT / "opax-mcp" / "capability_bridge.py" + with open(bridge_path, "r", encoding="utf-8") as f: + tree = ast.parse(f.read()) + + forbidden_imports = { + "os", "subprocess", "requests", "httpx", "google", "gitea", + "firebase", "firestore", + } + + for node in ast.walk(tree): + if isinstance(node, ast.Import): + for alias in node.names: + self.assertNotIn(alias.name.split('.')[0], forbidden_imports) + elif isinstance(node, ast.ImportFrom): + if node.module: + self.assertNotIn(node.module.split('.')[0], forbidden_imports) + + def test_bridge_does_not_mutate_registry(self): + # This is implicitly tested by the deterministic output and other tests, + # but a direct check confirms the source objects are not modified. + initial_capabilities = list_capabilities() + build_capability_system_context() + self.assertEqual(initial_capabilities, list_capabilities()) + + def test_bridge_output_includes_all_rules(self): + rules_block = ( + "Rules:\n" + "- Planned capabilities are not available in this chat and must not be claimed as active.\n" + "- No tool, repository, deployment, memory, document retrieval, external service or state-changing action is activated by this capability summary.\n" + "- Future write or high-impact capabilities require a concrete, explicit, bound user approval before execution." + ) + self.assertIn(rules_block, self.context_output) + + def test_bridge_does_not_expose_backend_tool_names(self): + all_capabilities = list_capabilities() + tool_names = {c.backend_tool for c in all_capabilities if c.backend_tool} + for tool_name in tool_names: + self.assertNotIn(tool_name, self.context_output) + + +class TestServerIntegrationContract(unittest.TestCase): + + def test_run_emma_integration_in_server_py(self): + server_path = REPO_ROOT / "opax-mcp" / "server.py" + with open(server_path, "r", encoding="utf-8") as f: + tree = ast.parse(f.read()) + + run_emma_func = None + for node in ast.walk(tree): + if isinstance(node, ast.AsyncFunctionDef) and node.name == "run_emma": + run_emma_func = node + break + + self.assertIsNotNone(run_emma_func, "async def run_emma not found in server.py") + + # 1. References build_capability_system_context + call_to_build_context = any( + isinstance(n, ast.Call) and isinstance(n.func, ast.Name) and n.func.id == 'build_capability_system_context' + for n in ast.walk(run_emma_func) + ) + self.assertTrue(call_to_build_context, "run_emma does not call build_capability_system_context") + + # 2. Passes system_context to canonical_emma.run + call_to_run = None + for node in ast.walk(run_emma_func): + if isinstance(node, ast.Call) and hasattr(node.func, 'value') and hasattr(node.func.value, 'id') and node.func.value.id == 'canonical_emma' and node.func.attr == 'run': + call_to_run = node + break + + self.assertIsNotNone(call_to_run, "run_emma does not call canonical_emma.run") + + has_system_context_kw = any( + kw.arg == 'system_context' for kw in call_to_run.keywords + ) + self.assertTrue(has_system_context_kw, "system_context is not passed to canonical_emma.run") + + # 3. Preserves history=[] + has_history_kw = any( + kw.arg == 'history' and isinstance(kw.value, ast.List) and not kw.value.elts + for kw in call_to_run.keywords + ) + self.assertTrue(has_history_kw, "history=[] is not preserved in call to canonical_emma.run") + + # 4. Preserves prompt argument structure + prompt_kw = next((kw for kw in call_to_run.keywords if kw.arg == 'prompt'), None) + self.assertIsNotNone(prompt_kw, "prompt argument missing in call to canonical_emma.run") + # A basic check to ensure it's calling p.get, not a simple variable + self.assertIsInstance(prompt_kw.value, ast.Call, "prompt argument is not a function call") + +if __name__ == "__main__": + unittest.main() diff --git a/opax-mcp/test_emma_adapter.py b/opax-mcp/test_emma_adapter.py index ff43351..33ff07b 100644 --- a/opax-mcp/test_emma_adapter.py +++ b/opax-mcp/test_emma_adapter.py @@ -134,5 +134,52 @@ class TestCanonicalEmmaPhase1B(unittest.IsolatedAsyncioTestCase): ) self.assertIn("Ingen handling utføres.", normalized_prompt) + async def test_run_without_system_context(self): + """(New) Verifies run() without context passes the original system prompt.""" + mock_chat_fn = AsyncMock() + model = "test-model-no-context" + prompt = "test prompt no context" + mock_response = {"response": "ok"} + mock_chat_fn.return_value = mock_response + + adapter = CanonicalEmma(chat_function=mock_chat_fn, model=model) + result = await adapter.run(prompt, system_context=None) + + self.assertEqual(result, mock_response) + mock_chat_fn.assert_awaited_once() + + call_args = mock_chat_fn.call_args + self.assertEqual(len(call_args.args), 3) + self.assertEqual(call_args.args[0], model) + self.assertEqual(call_args.args[1], prompt) + self.assertEqual(call_args.args[2], get_runtime_system_prompt()) + + async def test_run_with_system_context(self): + """(New) Verifies run() with context appends it correctly.""" + mock_chat_fn = AsyncMock() + model = "test-model-with-context" + prompt = "test prompt with context" + bridge_context = "CURRENT CAPABILITY STATUS\n\nAvailable: Yes" + mock_response = {"response": "ok"} + mock_chat_fn.return_value = mock_response + + adapter = CanonicalEmma(chat_function=mock_chat_fn, model=model) + result = await adapter.run(prompt, system_context=bridge_context) + + self.assertEqual(result, mock_response) + mock_chat_fn.assert_awaited_once() + + call_args = mock_chat_fn.call_args + self.assertEqual(len(call_args.args), 3) + self.assertEqual(call_args.args[0], model) + self.assertEqual(call_args.args[1], prompt) # User prompt is unchanged + + composed_prompt = call_args.args[2] + canonical_prompt = get_runtime_system_prompt() + + self.assertTrue(composed_prompt.startswith(canonical_prompt)) + self.assertIn(f"\n\n{bridge_context.strip()}", composed_prompt) + + if __name__ == '__main__': unittest.main()