diff --git a/opax-mcp/server.py b/opax-mcp/server.py index 7174cae..cde624a 100644 --- a/opax-mcp/server.py +++ b/opax-mcp/server.py @@ -19,6 +19,18 @@ from fastapi.middleware.cors import CORSMiddleware import asyncio import secrets +# --- OPAX Imports --- +from persistence.firestore_store import FirestoreMemoryStore +from persistence.memory_store import MemoryStore +from contracts.common import ( + CallerContext, + CallEmmaRequest, + CallEmmaResponse, + AuditEvent as EmmaAuditEvent, +) +from policy.caller_context import derive_caller_context +from persistence.memory_scope import calculate_effective_memory_scope + # --- Tool Imports (with fault tolerance) --- try: from tyr.tools.scan_tyr_surface import scan_tyr_surface @@ -770,6 +782,57 @@ async def _gitea_put(path: str, body: dict) -> Any: return r.json() +# --------------------------------------------------------------------------- +# Phase 2C: Local-only, identity-gated helpers for call_emma +# --------------------------------------------------------------------------- + +def is_call_emma_local_mode_enabled() -> bool: + """Checks if the server is configured to run call_emma in local test mode.""" + return ( + os.getenv("ENABLE_CALL_EMMA_TOOL", "").lower() == "true" + and bool(os.getenv("FIRESTORE_EMULATOR_HOST")) + ) + +def get_emma_store() -> MemoryStore: + """Factory to get the appropriate memory store for call_emma.""" + if not is_call_emma_local_mode_enabled(): + raise RuntimeError( + "call_emma is enabled only for local Firestore emulator mode." + ) + + project_id = os.getenv("GOOGLE_CLOUD_PROJECT", "opax-phase2c-local") + return FirestoreMemoryStore(project_id=project_id) + +def get_call_emma_caller_context() -> CallerContext: + """ + Gets the caller context for call_emma, using a server-injected test identity. + This is for local/test environments only. + """ + if not is_call_emma_local_mode_enabled(): + raise RuntimeError( + "Test caller context is available only in local emulator mode." + ) + + test_caller_id = os.getenv("OPAX_EMMA_TEST_CALLER_ID") + if not test_caller_id: + raise ValueError( + "Configuration error: OPAX_EMMA_TEST_CALLER_ID must be set " + "for local emulator tests." + ) + + return derive_caller_context(test_caller_id) + + +async def _local_deterministic_emma_chat(model: str, prompt: str, system: str = "") -> dict: + """A local, deterministic, non-network chat function for testing.""" + return { + "model": "local-test-model", + "response": "This is a deterministic local reply.", + "done": True, + "total_duration_ms": 1, + } + + # --------------------------------------------------------------------------- # Tool implementations (refaktorert til å bruke _agent_get/_agent_post) # --------------------------------------------------------------------------- @@ -908,6 +971,145 @@ async def build_and_deploy_service(p: dict) -> dict: return result + +async def handle_call_emma( + arguments: Dict[str, Any], + caller: CallerContext, + store: MemoryStore, +) -> CallEmmaResponse: + """ + Handles the call_emma tool, managing conversation state, memory, and auditing + for a single turn of conversation with the canonical Emma agent. + """ + correlation_id = str(uuid.uuid4()) + conversation_id = None + status = "failure" + error_type = None + model_name = None + history_message_count = 0 + memory_count = 0 + proposed_action_count = 0 + + try: + request = CallEmmaRequest(**arguments) + conversation_id = request.conversation_id + + if conversation_id: + conversation = await store.get_conversation(conversation_id, caller.owner_id) + if not conversation: + raise ValueError(f"Conversation {conversation_id} not found or access denied.") + else: + conversation = await store.create_conversation( + owner_id=caller.owner_id, + workspace_id=caller.workspace_id, + created_by=caller.caller_id + ) + conversation_id = conversation.conversation_id + + user_message = { + "message_id": str(uuid.uuid4()), + "conversation_id": conversation_id, + "owner_id": caller.owner_id, + "workspace_id": caller.workspace_id, + "role": "user", + "content": request.prompt, + "created_at": datetime.now(timezone.utc), + "caller_id": caller.caller_id, + "correlation_id": correlation_id, + } + await store.append_message(conversation_id, user_message) + + history = await store.list_messages(conversation_id, limit=20) + history_message_count = len(history) + + # Per rule #10, recall_memory is not called as it requires an embedding. + recalled_memories = [] + memory_count = 0 + + if is_call_emma_local_mode_enabled(): + local_emma = CanonicalEmma( + chat_function=_local_deterministic_emma_chat, + model="local-test-model" + ) + emma_for_call = local_emma + else: + emma_for_call = canonical_emma + + emma_response = await emma_for_call.run( + prompt=request.prompt, + history=history + ) + model_name = emma_response.get("model") + + assistant_message_content = emma_response.get("response", "") + assistant_message = { + "message_id": str(uuid.uuid4()), + "conversation_id": conversation_id, + "owner_id": caller.owner_id, + "workspace_id": caller.workspace_id, + "role": "assistant", + "content": assistant_message_content, + "created_at": datetime.now(timezone.utc), + "caller_id": "agent:emma", + "correlation_id": correlation_id, + "model": model_name, + } + await store.append_message(conversation_id, assistant_message) + + status = "success" + + # Proposed actions are not implemented yet + proposed_actions = [] + proposed_action_count = len(proposed_actions) + + return CallEmmaResponse( + reply_text=assistant_message_content, + model=model_name or "unknown", + conversation_id=conversation_id, + correlation_id=correlation_id, + proposed_actions=proposed_actions, + ) + + except Exception as e: + error_type = type(e).__name__ + logger.error( + "[handle_call_emma ERROR] correlation_id=%s error_type=%s", + correlation_id, + error_type, + ) + raise RuntimeError( + "An internal error occurred in handle_call_emma. " + f"Correlation ID: {correlation_id}" + ) from None + + finally: + try: + event_type = "TOOL_CALL" if status == "success" else "TOOL_CALL_FAILED" + audit_details = { + "tool_name": "call_emma", + "conversation_id": conversation_id, + "correlation_id": correlation_id, + "status": status, + "model": model_name, + "history_message_count": history_message_count, + "memory_count": memory_count, + "proposed_action_count": proposed_action_count, + "error_type": error_type, + } + audit_event = EmmaAuditEvent( + event_type=event_type, + caller_id=caller.caller_id, + details={k: v for k, v in audit_details.items() if v is not None}, + workspace_id=caller.workspace_id, + ) + await store.record_audit_event(audit_event) + except Exception as audit_e: + logger.error( + "[handle_call_emma AUDIT FAILED] correlation_id=%s audit_error_type=%s", + correlation_id, + type(audit_e).__name__, + ) + # --------------------------------------------------------------------------- # Tool registry + MCP schema # --------------------------------------------------------------------------- @@ -992,6 +1194,10 @@ TOOLS = { "write_memory_bank": (write_memory_bank, "Writes a file to the project memory bank.", {"type": "object", "properties": {"file_name": {"type": "string"}, "content": {"type": "string"}}, "required": ["file_name", "content"]}), "build_and_deploy_service": (build_and_deploy_service, "Triggers the production build & deployment pipeline.", {"type": "object", "properties": {"branch": {"type": "string"}}, "required": []}), } + +if is_call_emma_local_mode_enabled(): + TOOLS["call_emma"] = (handle_call_emma, "Calls the canonical Emma agent with managed persistence (EMULATOR ONLY)", {"type": "object", "properties": {"prompt": {"type": "string"}}, "required": ["prompt"]}) + if provision_new_mcp_module: TOOLS["provision_new_mcp_module"] = (provision_new_mcp_module, "Provisions a new MCP module", {"type":"object","properties":{"module_name":{"type":"string"},"tool_name":{"type":"string"},"tool_spec":{"type":"object"}},"required":["module_name","tool_name","tool_spec"]}) @@ -1024,9 +1230,35 @@ async def mcp_handler(request: Request): if not entry: return JSONResponse(_jsonrpc_err(req_id, -32601, f"Unknown tool: {tool_name}")) handler, _, _ = entry try: - result = await handler(tool_args) - return JSONResponse(_jsonrpc_ok(req_id, {"content": [{"type": "text", "text": json.dumps(result, ensure_ascii=False)}]})) + if tool_name == "call_emma" and is_call_emma_local_mode_enabled(): + caller = get_call_emma_caller_context() + store = get_emma_store() + result = await handler(tool_args, caller, store) + else: + result = await handler(tool_args) + + # For call_emma, the result is already the final response object. + # For other tools, we wrap it. + if tool_name == "call_emma": + return JSONResponse(_jsonrpc_ok(req_id, result.model_dump(mode='json'))) + else: + return JSONResponse(_jsonrpc_ok(req_id, {"content": [{"type": "text", "text": json.dumps(result, ensure_ascii=False)}]})) except Exception as e: + if tool_name == "call_emma": + error_type = type(e).__name__ + logger.error( + "[MCP HANDLER ERROR] tool=%s error_type=%s", + tool_name, + error_type, + ) + return JSONResponse( + _jsonrpc_err( + req_id, + -32000, + f"Server error in tool 'call_emma': {error_type}", + ) + ) + logger.error(f"[MCP HANDLER ERROR] tool={tool_name} {type(e).__name__}: {e}", exc_info=True) return JSONResponse(_jsonrpc_err(req_id, -32000, str(e))) if method.startswith("notifications/"): return JSONResponse(status_code=202, content={}) diff --git a/opax-mcp/test_phase2c_real_integration.py b/opax-mcp/test_phase2c_real_integration.py new file mode 100644 index 0000000..604f9ef --- /dev/null +++ b/opax-mcp/test_phase2c_real_integration.py @@ -0,0 +1,203 @@ + +import os +import unittest +import httpx +import asyncio +from google.cloud import firestore +import subprocess +import time +import uuid +import secrets +from policy.caller_context import derive_caller_context +from contracts.common import CallerContext +from server import is_call_emma_local_mode_enabled + +# This test requires a running Firestore emulator and the server to be started. +# It makes REAL HTTP requests to the server, which in turn talks to the emulator. + +class TestPhase2CRealIntegration(unittest.IsolatedAsyncioTestCase): + _server_process = None + _expected_caller: CallerContext = None + + def __init__(self, *args, **kwargs): + super().__init__(*args, **kwargs) + self.created_conversation_ids = [] + self.created_correlation_ids = [] + + @classmethod + def setUpClass(cls): + # 1. Check environment prerequisites + assert os.getenv("FIRESTORE_EMULATOR_HOST"), "FIRESTORE_EMULATOR_HOST must be set" + + # 2. Define and start the MCP server with a fixed, isolated environment + test_env = os.environ.copy() + test_env["ENABLE_CALL_EMMA_TOOL"] = "true" + test_env["OPAX_EMMA_TEST_CALLER_ID"] = "agent:perplexity" + test_env["MCP_SECRET"] = secrets.token_hex(16) + test_env["GOOGLE_CLOUD_PROJECT"] = "opax-phase2c-local" + + cls.mcp_secret = test_env["MCP_SECRET"] + + # Use the import path confirmed by the smoke test + cls._server_process = subprocess.Popen( + ["opax-mcp/.venv/bin/uvicorn", "opax-mcp.server:app", "--host", "127.0.0.1", "--port", "8088"], + env=test_env, + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + ) + # Wait for the server to start up + time.sleep(5) + if cls._server_process.poll() is not None: + raise RuntimeError("MCP test server exited during startup.") + + # 3. Create a real Firestore client for test verification + # Client is created in asyncSetUp to share the event loop + + # 4. Derive the expected identity for assertions + cls._expected_caller = derive_caller_context(test_env["OPAX_EMMA_TEST_CALLER_ID"]) + + @classmethod + def tearDownClass(cls): + if cls._server_process: + cls._server_process.terminate() + try: + cls._server_process.wait(timeout=5) + except subprocess.TimeoutExpired: + cls._server_process.kill() + cls._server_process.wait(timeout=5) + + async def asyncSetUp(self): + self._firestore_client = firestore.AsyncClient(project="opax-phase2c-local") + + async def asyncTearDown(self): + # Clean up only the specific documents created during this test run + for convo_id in self.created_conversation_ids: + messages_ref = self._firestore_client.collection("emma_conversations").document(convo_id).collection("messages") + messages = [doc async for doc in messages_ref.stream()] + for msg in messages: + await msg.reference.delete() + await self._firestore_client.collection("emma_conversations").document(convo_id).delete() + + for corr_id in self.created_correlation_ids: + query = self._firestore_client.collection("emma_audit_events").where("details.correlation_id", "==", corr_id) + docs = [doc async for doc in query.stream()] + for doc in docs: + await doc.reference.delete() + + if self._firestore_client: + self._firestore_client.close() + + async def _call_tool(self, name, args, headers=None): + json_payload = { + "jsonrpc": "2.0", + "id": str(uuid.uuid4()), + "method": "tools/call", + "params": { + "name": name, + "arguments": args + } + } + default_headers = {"api-key": self.mcp_secret} + if headers: + default_headers.update(headers) + + async with httpx.AsyncClient() as client: + response = await client.post("http://127.0.0.1:8088/", json=json_payload, headers=default_headers) + return response.json() + + def test_00_local_mode_gate_requires_both_flags(self): + """Tests the is_call_emma_local_mode_enabled helper directly.""" + original_enable = os.environ.get("ENABLE_CALL_EMMA_TOOL") + original_host = os.environ.get("FIRESTORE_EMULATOR_HOST") + try: + # Case 1: Both missing + if "ENABLE_CALL_EMMA_TOOL" in os.environ: del os.environ["ENABLE_CALL_EMMA_TOOL"] + if "FIRESTORE_EMULATOR_HOST" in os.environ: del os.environ["FIRESTORE_EMULATOR_HOST"] + self.assertFalse(is_call_emma_local_mode_enabled()) + + # Case 2: Host missing + os.environ["ENABLE_CALL_EMMA_TOOL"] = "true" + self.assertFalse(is_call_emma_local_mode_enabled()) + + # Case 3: Enable flag missing + if "ENABLE_CALL_EMMA_TOOL" in os.environ: del os.environ["ENABLE_CALL_EMMA_TOOL"] + os.environ["FIRESTORE_EMULATOR_HOST"] = "localhost:8686" + self.assertFalse(is_call_emma_local_mode_enabled()) + + # Case 4: Both present + os.environ["ENABLE_CALL_EMMA_TOOL"] = "true" + self.assertTrue(is_call_emma_local_mode_enabled()) + finally: + if original_enable is not None: os.environ["ENABLE_CALL_EMMA_TOOL"] = original_enable + else: os.environ.pop("ENABLE_CALL_EMMA_TOOL", None) + if original_host is not None: os.environ["FIRESTORE_EMULATOR_HOST"] = original_host + else: os.environ.pop("FIRESTORE_EMULATOR_HOST", None) + + async def test_01_new_request_creates_conversation(self): + """Tests a new request creates a conversation and messages with the correct identity.""" + test_prompt = f"This is an integration test. Run ID: {uuid.uuid4()}" + + response = await self._call_tool("call_emma", {"prompt": test_prompt}) + + self.assertIn("result", response) + result = response["result"] + self.assertIn("conversation_id", result) + self.assertIn("correlation_id", result) + conversation_id = result["conversation_id"] + correlation_id = result["correlation_id"] + + # Track IDs for cleanup + self.created_conversation_ids.append(conversation_id) + self.created_correlation_ids.append(correlation_id) + + # Verify deterministic test model response + self.assertEqual(result.get("model"), "local-test-model") + self.assertIn("deterministic local reply", result.get("reply_text", "")) + self.assertNotIn(test_prompt, result.get("reply_text", "")) + + # Verify conversation document exists + convo_doc = await self._firestore_client.collection("emma_conversations").document(conversation_id).get() + self.assertTrue(convo_doc.exists) + self.assertEqual(convo_doc.to_dict().get("owner_id"), self._expected_caller.owner_id) + self.assertEqual(convo_doc.to_dict().get("workspace_id"), self._expected_caller.workspace_id) + + # Verify messages subcollection contains two messages (user and assistant) + messages_ref = self._firestore_client.collection("emma_conversations").document(conversation_id).collection("messages") + messages = [doc async for doc in messages_ref.stream()] + self.assertEqual(len(messages), 2) + + # Verify audit log was written + audit_ref = self._firestore_client.collection("emma_audit_events") + query = audit_ref.where("details.correlation_id", "==", correlation_id) + audit_docs = [doc async for doc in query.stream()] + self.assertEqual(len(audit_docs), 1) + self.assertEqual(audit_docs[0].to_dict()["details"]["status"], "success") + + async def test_02_client_identity_is_ignored(self): + """Tests that a client-supplied user_id in the arguments is ignored.""" + malicious_args = { + "prompt": "test", + "client_context": { + "user_id": "user:chris.christiansen@vauco.no" + } + } + response = await self._call_tool("call_emma", malicious_args) + self.assertIn("result", response) + conversation_id = response["result"]["conversation_id"] + correlation_id = response["result"]["correlation_id"] + + # Track IDs for cleanup + self.created_conversation_ids.append(conversation_id) + self.created_correlation_ids.append(correlation_id) + + # Verify deterministic test model response + self.assertEqual(response["result"].get("model"), "local-test-model") + self.assertIn("deterministic local reply", response["result"].get("reply_text", "")) + + # Verify the created conversation belongs to the SERVER-INJECTED identity, not the client-supplied one + convo_doc = await self._firestore_client.collection("emma_conversations").document(conversation_id).get() + self.assertTrue(convo_doc.exists) + self.assertEqual(convo_doc.to_dict().get("owner_id"), self._expected_caller.owner_id) + +if __name__ == "__main__": + unittest.main()