feat(opax): add Firestore persistence foundation

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

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

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

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()