diff --git a/backends/advanced/src/advanced_omi_backend/services/memory/providers/chronicle.py b/backends/advanced/src/advanced_omi_backend/services/memory/providers/chronicle.py index a0974e21..d43036d3 100644 --- a/backends/advanced/src/advanced_omi_backend/services/memory/providers/chronicle.py +++ b/backends/advanced/src/advanced_omi_backend/services/memory/providers/chronicle.py @@ -290,6 +290,62 @@ async def count_memories(self, user_id: str) -> Optional[int]: memory_logger.error(f"Count memories failed: {e}") return None + async def get_memory(self, memory_id: str, user_id: Optional[str] = None) -> Optional[MemoryEntry]: + """Get a specific memory by ID. + + Retrieves a single memory from Qdrant by its ID and verifies user ownership. + + Args: + memory_id: Unique identifier of the memory to retrieve + user_id: Optional user ID for access control + + Returns: + MemoryEntry object if found and authorized, None otherwise + """ + if not self._initialized: + await self.initialize() + + try: + # Retrieve the point from Qdrant + points = await self.vector_store.client.retrieve( + collection_name=self.vector_store.collection_name, + ids=[memory_id], + with_payload=True, + with_vectors=False + ) + + if not points: + memory_logger.debug(f"Memory {memory_id} not found in vector store") + return None + + point = points[0] + payload = point.payload + + # Check user ownership if user_id is provided + if user_id: + memory_user_id = payload.get("metadata", {}).get("user_id") + if memory_user_id != user_id: + memory_logger.warning( + f"Access denied: Memory {memory_id} belongs to {memory_user_id}, requested by {user_id}" + ) + return None + + # Convert to MemoryEntry + memory_entry = MemoryEntry( + id=str(point.id), + content=payload.get("content", ""), + metadata=payload.get("metadata", {}), + created_at=payload.get("created_at"), + embedding=None # Don't return embeddings for single memory retrieval + ) + + memory_logger.info(f"📖 Retrieved memory {memory_id} for user {user_id}") + return memory_entry + + except Exception as e: + memory_logger.error(f"Get memory {memory_id} failed: {e}", exc_info=True) + return None + async def delete_memory(self, memory_id: str, user_id: Optional[str] = None, user_email: Optional[str] = None) -> bool: """Delete a specific memory by ID. @@ -428,7 +484,7 @@ def _create_memory_entries( "extraction_enabled": self.config.extraction_enabled, }, embedding=embedding, - created_at=str(int(time.time())), + created_at=int(time.time()), ) ) @@ -626,7 +682,7 @@ async def _apply_memory_actions( if emb is None: memory_logger.warning(f"Skipping ADD action due to missing embedding: {action_text}") continue - + memory_id = str(uuid.uuid4()) memory_entries.append( MemoryEntry( @@ -634,7 +690,7 @@ async def _apply_memory_actions( content=action_text, metadata=base_metadata, embedding=emb, - created_at=str(int(time.time())), + created_at=int(time.time()), ) ) memory_logger.info(f"➕ Added new memory: {memory_id} - {action_text[:50]}...") diff --git a/tests/setup/test_env.py b/tests/setup/test_env.py index 929e83e2..7e3ca983 100644 --- a/tests/setup/test_env.py +++ b/tests/setup/test_env.py @@ -1,25 +1,26 @@ # Test Environment Configuration import os from pathlib import Path +from dotenv import load_dotenv # Load .env file from backends/advanced directory if it exists # This allows tests to work when run from VSCode or command line -def load_env_file(): - """Load environment variables from .env file if it exists.""" - # Look for .env in backends/advanced directory - env_file = Path(__file__).parent.parent.parent / "backends" / "advanced" / ".env" - if env_file.exists(): - with open(env_file) as f: - for line in f: - line = line.strip() - if line and not line.startswith('#') and '=' in line: - key, value = line.split('=', 1) - # Only set if not already in environment (CI takes precedence) - if key not in os.environ: - os.environ[key] = value +# def load_env_file(): +# """Load environment variables from .env file if it exists.""" +# # Look for .env in backends/advanced directory +# env_file = Path(__file__).parent.parent.parent / "backends" / "advanced" / ".env" +# if env_file.exists(): +# with open(env_file) as f: +# for line in f: +# line = line.strip() +# if line and not line.startswith('#') and '=' in line: +# key, value = line.split('=', 1) +# # Only set if not already in environment (CI takes precedence) +# if key not in os.environ: +# os.environ[key] = value # Load .env file (CI environment variables take precedence) -load_env_file() +# load_env_file() # Load .env from backends/advanced directory to get COMPOSE_PROJECT_NAME backend_env_path = Path(__file__).resolve().parents[2] / "backends" / "advanced" / ".env"