From 3a2aae98aa424c4177deda9bb5e6b97ced6fc2be Mon Sep 17 00:00:00 2001 From: Chris Christiansen Date: Wed, 16 Sep 2026 16:46:24 +0000 Subject: [PATCH] feat(opax): add Firestore persistence foundation --- opax-mcp/persistence/firestore_store.py | 86 +++++++++++++++++++++ opax-mcp/persistence/memory_scope.py | 46 +++++++++++ opax-mcp/persistence/memory_store.py | 52 +++++++++---- opax-mcp/test_phase2b_firestore_store.py | 97 ++++++++++++++++++++++++ 4 files changed, 265 insertions(+), 16 deletions(-) create mode 100644 opax-mcp/persistence/firestore_store.py create mode 100644 opax-mcp/persistence/memory_scope.py create mode 100644 opax-mcp/test_phase2b_firestore_store.py diff --git a/opax-mcp/persistence/firestore_store.py b/opax-mcp/persistence/firestore_store.py new file mode 100644 index 0000000..513a7a1 --- /dev/null +++ b/opax-mcp/persistence/firestore_store.py @@ -0,0 +1,86 @@ +""" +Firestore-backed implementation of the MemoryStore interface. +""" +from typing import List, Dict, Any, Optional +from datetime import datetime, timezone + +from google.cloud import firestore + +from contracts.common import EmmaConversation, MemoryRecord, AuditEvent +from .memory_store import MemoryStore + +class FirestoreMemoryStore(MemoryStore): + """ + Implements the MemoryStore interface using Google Cloud Firestore. + """ + def __init__(self, project_id: str, client: Optional[firestore.AsyncClient] = None): + self._project_id = project_id + # Client can be injected for testing, otherwise it will be created on demand. + self._client = client + + def _get_client(self) -> firestore.AsyncClient: + if not self._client: + self._client = firestore.AsyncClient(project=self._project_id) + return self._client + + async def create_conversation(self, owner_id: str, workspace_id: str, created_by: str) -> EmmaConversation: + db = self._get_client() + convo = EmmaConversation(owner_id=owner_id, workspace_id=workspace_id, created_by=created_by) + doc_ref = db.collection("emma_conversations").document(convo.conversation_id) + await doc_ref.set(convo.model_dump(mode='json')) + return convo + + async def get_conversation(self, conversation_id: str, owner_id: str) -> Optional[EmmaConversation]: + db = self._get_client() + doc_ref = db.collection("emma_conversations").document(conversation_id) + doc = await doc_ref.get() + if not doc.exists: + return None + + convo = EmmaConversation(**doc.to_dict()) + # Security: Enforce ownership + if convo.owner_id != owner_id: + return None # Or raise an exception + return convo + + async def append_message(self, conversation_id: str, message: Dict[str, Any]) -> None: + db = self._get_client() + # Note: message_id should be auto-generated by Firestore for ordering + messages_ref = db.collection("emma_conversations").document(conversation_id).collection("messages") + await messages_ref.add(message) + # Update last_updated_at on parent conversation document + convo_ref = db.collection("emma_conversations").document(conversation_id) + await convo_ref.update({"last_updated_at": datetime.now(timezone.utc)}) + + async def list_messages(self, conversation_id: str, limit: int = 50) -> List[Dict[str, Any]]: + db = self._get_client() + messages_ref = db.collection("emma_conversations").document(conversation_id).collection("messages") + docs = messages_ref.order_by("created_at", direction=firestore.Query.DESCENDING).limit(limit).stream() + return [doc.to_dict() async for doc in docs][::-1] # Reverse to get chronological order + + async def record_audit_event(self, event: AuditEvent) -> str: + db = self._get_client() + doc_ref = db.collection("emma_audit_events").document(event.event_id) + # Use create() to ensure the document does not already exist, guaranteeing append-only. + await doc_ref.create(event.model_dump(mode='json')) + return event.event_id + + # --- Placeholder methods not fully implemented in Phase 2B --- + + async def save_conversation_summary(self, conversation_id: str, summary: str) -> None: + # In a real implementation, this would update a summary field + pass + + async def store_memory(self, record: MemoryRecord) -> None: + # This would write to the emma_memories collection + pass + + async def recall_memory(self, embedding: List[float], owner_id: str, scopes: List[str]) -> List[MemoryRecord]: + # This would perform a query against a vector database, which is out of scope. + return [] + + async def get_profile(self, owner_id: str) -> Optional[Dict[str, Any]]: + pass + + async def update_profile(self, owner_id: str, profile_data: Dict[str, Any]) -> None: + pass diff --git a/opax-mcp/persistence/memory_scope.py b/opax-mcp/persistence/memory_scope.py new file mode 100644 index 0000000..5990a65 --- /dev/null +++ b/opax-mcp/persistence/memory_scope.py @@ -0,0 +1,46 @@ +""" +Logic for calculating effective memory scope based on different policies. +""" +from typing import Set +from contracts.common import CallerContext, EmmaConversation + +def calculate_effective_memory_scope( + requested_scope: Set[str], + caller_context: CallerContext, + conversation_context: EmmaConversation +) -> Set[str]: + """ + Calculates the final, secure memory scope by intersecting requested scopes + with server-side policies. + + Args: + requested_scope: The scope the client is asking for. + caller_context: The server-derived context of the authenticated caller. + conversation_context: The context of the current conversation. + + Returns: + A set of strings representing the final, allowed memory scopes. + """ + # 1. Start with a default-deny principle + effective_scope = set() + + # 2. Define server-side allowable scopes based on caller profile + # This is a placeholder for a more complex policy engine. + if caller_context.profile == "admin": + allowed_by_caller = {"current_conversation", "owner_private_memory", "workspace_operational_memory"} + elif caller_context.profile == "operator": + allowed_by_caller = {"current_conversation", "workspace_operational_memory"} + else: # readonly / default + allowed_by_caller = {"current_conversation"} + + # 3. Intersect requested scope with what the caller is allowed to do + permitted_scope = requested_scope.intersection(allowed_by_caller) + + # 4. Filter based on data attributes (this would happen in the query) + # For now, we just return the permitted scope types. The query in the + # persistence layer would be responsible for adding the WHERE clauses. + # e.g., if "owner_private_memory" is in the scope, the query must add + # `WHERE owner_id == caller_context.owner_id`. + effective_scope = permitted_scope + + return effective_scope diff --git a/opax-mcp/persistence/memory_store.py b/opax-mcp/persistence/memory_store.py index 205ba10..1102df0 100644 --- a/opax-mcp/persistence/memory_store.py +++ b/opax-mcp/persistence/memory_store.py @@ -1,21 +1,28 @@ """ -Defines the abstract interface for Emma's memory systems. +Defines the abstract interface for Emma's memory systems and provides a +non-persistent, in-memory implementation for testing. """ from abc import ABC, abstractmethod from typing import List, Dict, Any, Optional - -from contracts.common import EmmaConversation, MemoryRecord - +from contracts.common import EmmaConversation, MemoryRecord, AuditEvent, CallEmmaRequest class MemoryStore(ABC): """Abstract Base Class for all memory store implementations.""" + @abstractmethod + async def create_conversation(self, owner_id: str, workspace_id: str, created_by: str) -> EmmaConversation: + pass + @abstractmethod async def get_conversation(self, conversation_id: str, owner_id: str) -> Optional[EmmaConversation]: pass @abstractmethod - async def save_conversation(self, conversation: EmmaConversation) -> None: + async def append_message(self, conversation_id: str, message: Dict[str, Any]) -> None: + pass + + @abstractmethod + async def list_messages(self, conversation_id: str, limit: int = 50) -> List[Dict[str, Any]]: pass @abstractmethod @@ -23,9 +30,12 @@ class MemoryStore(ABC): pass @abstractmethod - async def recall_memory(self, embedding: List[float], owner_id: str, scope: List[str]) -> List[MemoryRecord]: + async def recall_memory(self, embedding: List[float], owner_id: str, scopes: List[str]) -> List[MemoryRecord]: pass + @abstractmethod + async def record_audit_event(self, event: AuditEvent) -> str: + pass class InMemoryMemoryStore(MemoryStore): """ @@ -34,28 +44,38 @@ class InMemoryMemoryStore(MemoryStore): """ def __init__(self): self._conversations: Dict[str, EmmaConversation] = {} + self._messages: Dict[str, List[Dict[str, Any]]] = {} self._memories: List[MemoryRecord] = [] + self._audit: List[AuditEvent] = [] print("WARNING: InMemoryMemoryStore is active. Data will not be persisted.") + async def create_conversation(self, owner_id: str, workspace_id: str, created_by: str) -> EmmaConversation: + convo = EmmaConversation(owner_id=owner_id, workspace_id=workspace_id, created_by=created_by) + self._conversations[convo.conversation_id] = convo + self._messages[convo.conversation_id] = [] + return convo + async def get_conversation(self, conversation_id: str, owner_id: str) -> Optional[EmmaConversation]: convo = self._conversations.get(conversation_id) - # Enforce ownership check if convo and convo.owner_id == owner_id: return convo return None - async def save_conversation(self, conversation: EmmaConversation) -> None: - # Enforce ownership - if conversation.owner_id: - self._conversations[conversation.conversation_id] = conversation + async def append_message(self, conversation_id: str, message: Dict[str, Any]) -> None: + if conversation_id in self._messages: + self._messages[conversation_id].append(message) + + async def list_messages(self, conversation_id: str, limit: int = 50) -> List[Dict[str, Any]]: + return self._messages.get(conversation_id, [])[-limit:] async def store_memory(self, record: MemoryRecord) -> None: - # Enforce ownership if record.owner_id: self._memories.append(record) - async def recall_memory(self, embedding: List[float], owner_id: str, scope: List[str]) -> List[MemoryRecord]: - # Simple text match for testing, no real embedding search. - # Also filters on ownership. + async def recall_memory(self, embedding: List[float], owner_id: str, scopes: List[str]) -> List[MemoryRecord]: recalled = [mem for mem in self._memories if mem.owner_id == owner_id] - return recalled[:5] # Return top 5 for simplicity + return recalled[:5] + + async def record_audit_event(self, event: AuditEvent) -> str: + self._audit.append(event) + return event.event_id diff --git a/opax-mcp/test_phase2b_firestore_store.py b/opax-mcp/test_phase2b_firestore_store.py new file mode 100644 index 0000000..5a132e4 --- /dev/null +++ b/opax-mcp/test_phase2b_firestore_store.py @@ -0,0 +1,97 @@ +import unittest +from unittest.mock import MagicMock, AsyncMock, patch +from datetime import datetime, timezone +from pathlib import Path +import sys + +REPO_ROOT = Path(__file__).resolve().parents[1] +sys.path.insert(0, str(REPO_ROOT / "opax-mcp")) + +from contracts.common import ( + EmmaConversation, MemoryRecord, AuditEvent, ToolRiskLevel +) +from persistence.memory_store import InMemoryMemoryStore +from persistence.firestore_store import FirestoreMemoryStore +from persistence.memory_scope import calculate_effective_memory_scope +from policy.caller_context import derive_caller_context + + +class TestPhase2BFirestoreStore(unittest.IsolatedAsyncioTestCase): + + def setUp(self): + self.mock_firestore_client = MagicMock() + self.mock_firestore_client.collection.return_value.document.return_value.set = AsyncMock() + self.mock_firestore_client.collection.return_value.add = AsyncMock() + self.mock_firestore_client.collection.return_value.document.return_value.create = AsyncMock() + + self.store = FirestoreMemoryStore(project_id="test-project", client=self.mock_firestore_client) + + async def test_firestore_client_is_injected(self): + """(1) Verifies the Firestore client is used when provided.""" + self.assertIs(self.store._client, self.mock_firestore_client) + + async def test_create_conversation_uses_correct_path(self): + """(2) Verifies conversation is stored under the correct collection/document ID.""" + convo = await self.store.create_conversation("owner1", "ws1", "creator1") + + self.mock_firestore_client.collection.assert_called_with("emma_conversations") + collection_mock = self.mock_firestore_client.collection.return_value + collection_mock.document.assert_called_with(convo.conversation_id) + document_mock = collection_mock.document.return_value + document_mock.set.assert_awaited_once() + + async def test_append_message_uses_subcollection(self): + """(3) Verifies messages are stored as a subcollection.""" + convo_id = "convo-123" + message = {"role": "user", "content": "hello"} + await self.store.append_message(convo_id, message) + + # Check path to messages subcollection + self.mock_firestore_client.collection.assert_called_with("emma_conversations") + convo_collection_mock = self.mock_firestore_client.collection.return_value + convo_collection_mock.document.assert_called_with(convo_id) + doc_mock = convo_collection_mock.document.return_value + doc_mock.collection.assert_called_with("messages") + + # Check that add was called on the subcollection + messages_collection_mock = doc_mock.collection.return_value + messages_collection_mock.add.assert_awaited_once_with(message) + + async def test_audit_event_is_append_only(self): + """(E) Verifies audit events use create() to be append-only.""" + event = AuditEvent(event_type="test", caller_id="c1", details={}, workspace_id="ws1") + await self.store.record_audit_event(event) + + self.mock_firestore_client.collection.assert_called_with("emma_audit_events") + collection_mock = self.mock_firestore_client.collection.return_value + collection_mock.document.assert_called_with(event.event_id) + document_mock = collection_mock.document.return_value + document_mock.create.assert_awaited_once() # Must use create(), not set() + document_mock.set.assert_not_called() + + +class TestPhase2BMemoryScope(unittest.TestCase): + + def test_effective_scope_logic(self): + """(D) Tests the effective memory scope calculation.""" + admin_caller = derive_caller_context("user:chris.christiansen@vauco.no") + operator_caller = derive_caller_context("agent:perplexity") + + convo = MagicMock() + + # Admin requests all, gets all allowed for admin + requested = {"current_conversation", "owner_private_memory", "workspace_operational_memory"} + effective = calculate_effective_memory_scope(requested, admin_caller, convo) + self.assertEqual(effective, {"current_conversation", "owner_private_memory", "workspace_operational_memory"}) + + # Operator requests all, gets only what operator is allowed + effective = calculate_effective_memory_scope(requested, operator_caller, convo) + self.assertEqual(effective, {"current_conversation", "workspace_operational_memory"}) + + # Operator requests only conversation, gets only conversation + requested = {"current_conversation"} + effective = calculate_effective_memory_scope(requested, operator_caller, convo) + self.assertEqual(effective, {"current_conversation"}) + +if __name__ == '__main__': + unittest.main()