diff --git a/src/powermem/core/async_memory.py b/src/powermem/core/async_memory.py index 503a04d..c4409d4 100644 --- a/src/powermem/core/async_memory.py +++ b/src/powermem/core/async_memory.py @@ -976,9 +976,23 @@ async def search( filters: Optional[Dict[str, Any]] = None, limit: int = 30, threshold: Optional[float] = None, + sort_by: Optional[Union[str, List[str]]] = None, ) -> Dict[str, Any]: """Search for memories asynchronously. + Args: + query: Search query string + user_id: Optional user ID filter + agent_id: Optional agent ID filter + run_id: Optional run ID filter + filters: Optional additional filters dictionary + limit: Maximum number of results to return (default: 30) + threshold: Optional minimum quality score threshold (0-1) + sort_by: Optional sorting criteria. Can be: + - Single value: "relevance" (default), "date_asc", "date_desc", + "importance_asc", "importance_desc", "access_count_desc", "retention_desc" + - List: Multiple sort keys for multi-criteria sorting (e.g., ["relevance", "date_desc"]) + Returns: Dict[str, Any]: A dictionary containing search results with the following structure: - "results" (List[Dict]): List of memory search results, where each result contains: @@ -1065,6 +1079,10 @@ async def search( transformed_result[key] = result[key] transformed_results.append(transformed_result) + # Apply sorting if sort_by is specified + if sort_by is not None: + transformed_results = self._sort_search_results(transformed_results, sort_by) + # Log audit event await self.audit.log_event_async("memory.search", { "query": query, @@ -1094,6 +1112,95 @@ async def search( logger.error(f"Failed to search memories: {e}") self.telemetry.capture_event("memory.search.error", {"error": str(e)}) raise + + def _sort_search_results( + self, + results: List[Dict[str, Any]], + sort_by: Union[str, List[str]] + ) -> List[Dict[str, Any]]: + """ + Sort search results based on specified criteria. + + Args: + results: List of search result dictionaries + sort_by: Sorting criteria - single value or list of values. + Options: "relevance" (default), "date_asc", "date_desc", + "importance_asc", "importance_desc", + "access_count_desc", "retention_desc" + + Returns: + Sorted list of results + """ + # Normalize sort_by to list + sort_keys = [sort_by] if isinstance(sort_by, str) else sort_by + + # Define sort key functions for each criterion + def get_sort_key(result: Dict[str, Any]) -> tuple: + """Generate sort key tuple for multi-criteria sorting.""" + keys = [] + for criterion in sort_keys: + criterion = criterion.lower().strip() + + if criterion == "relevance" or not criterion: + # Sort by quality score (descending - higher relevance first) + metadata = result.get("metadata", {}) + quality_score = metadata.get("_quality_score", result.get("score", 0.0)) + keys.append(-float(quality_score) if quality_score else 0) + + elif criterion == "date_asc": + # Sort by creation time ascending (oldest first) + created_at = result.get("created_at", "") + keys.append(created_at or "") + + elif criterion == "date_desc": + # Sort by creation time descending (newest first) + created_at = result.get("created_at", "") + keys.append(created_at or "0" if not created_at else created_at) + + elif criterion == "importance_asc": + # Sort by importance score ascending (lower importance first) + metadata = result.get("metadata", {}) + user_metadata = metadata.get("metadata", {}) if metadata else {} + importance = user_metadata.get("importance_score", 0.5) + keys.append(float(importance)) + + elif criterion == "importance_desc": + # Sort by importance score descending (higher importance first) + metadata = result.get("metadata", {}) + user_metadata = metadata.get("metadata", {}) if metadata else {} + importance = user_metadata.get("importance_score", 0.5) + keys.append(-float(importance)) + + elif criterion == "access_count_desc": + # Sort by access count descending (most accessed first) + metadata = result.get("metadata", {}) + user_metadata = metadata.get("metadata", {}) if metadata else {} + access_count = user_metadata.get("access_count", 0) + keys.append(-int(access_count)) + + elif criterion == "retention_desc": + # Sort by retention score descending (higher retention first) + metadata = result.get("metadata", {}) + user_metadata = metadata.get("metadata", {}) if metadata else {} + retention = user_metadata.get("retention_score", 0.0) + keys.append(-float(retention)) + + else: + # Unknown criterion, use relevance as fallback + logger.warning(f"Unknown sort criterion: {criterion}, using relevance") + quality_score = result.get("metadata", {}).get("_quality_score", result.get("score", 0.0)) + keys.append(-float(quality_score) if quality_score else 0) + + return tuple(keys) + + # Sort results + try: + sorted_results = sorted(results, key=get_sort_key, reverse=False) + logger.debug(f"Sorted {len(results)} results by criteria: {sort_keys}") + return sorted_results + except Exception as e: + logger.error(f"Failed to sort results: {e}") + return results async def get( self, diff --git a/src/powermem/core/memory.py b/src/powermem/core/memory.py index 2a43f7f..972f168 100644 --- a/src/powermem/core/memory.py +++ b/src/powermem/core/memory.py @@ -1100,9 +1100,23 @@ def search( filters: Optional[Dict[str, Any]] = None, limit: int = 30, threshold: Optional[float] = None, + sort_by: Optional[Union[str, List[str]]] = None, ) -> Dict[str, Any]: """Search for memories. + Args: + query: Search query string + user_id: Optional user ID filter + agent_id: Optional agent ID filter + run_id: Optional run ID filter + filters: Optional additional filters dictionary + limit: Maximum number of results to return (default: 30) + threshold: Optional minimum quality score threshold (0-1) + sort_by: Optional sorting criteria. Can be: + - Single value: "relevance" (default), "date_asc", "date_desc", + "importance_asc", "importance_desc", "access_count_desc", "retention_desc" + - List: Multiple sort keys for multi-criteria sorting (e.g., ["relevance", "date_desc"]) + Returns: Dict[str, Any]: A dictionary containing search results with the following structure: - "results" (List[Dict]): List of memory search results, where each result contains: @@ -1193,6 +1207,10 @@ def search( transformed_result["memory_id"] = transformed_result["id"] transformed_results.append(transformed_result) + # Apply sorting if sort_by is specified + if sort_by is not None: + transformed_results = self._sort_search_results(transformed_results, sort_by) + # Log audit event self.audit.log_event( "memory.search", @@ -1250,6 +1268,107 @@ def search( logger.error(f"Failed to search memories: {e}") self.telemetry.capture_event("memory.search.error", {"error": str(e)}) raise + + def _sort_search_results( + self, + results: List[Dict[str, Any]], + sort_by: Union[str, List[str]] + ) -> List[Dict[str, Any]]: + """ + Sort search results based on specified criteria. + + Args: + results: List of search result dictionaries + sort_by: Sorting criteria - single value or list of values. + Options: "relevance" (default), "date_asc", "date_desc", + "importance_asc", "importance_desc", + "access_count_desc", "retention_desc" + + Returns: + Sorted list of results + """ + # Normalize sort_by to list + sort_keys = [sort_by] if isinstance(sort_by, str) else sort_by + + # Define sort key functions for each criterion + def get_sort_key(result: Dict[str, Any]) -> tuple: + """Generate sort key tuple for multi-criteria sorting.""" + keys = [] + for criterion in sort_keys: + criterion = criterion.lower().strip() + + if criterion == "relevance" or not criterion: + # Sort by quality score (descending - higher relevance first) + metadata = result.get("metadata", {}) + quality_score = metadata.get("_quality_score", result.get("score", 0.0)) + keys.append(-float(quality_score) if quality_score else 0) + + elif criterion == "date_asc": + # Sort by creation time ascending (oldest first) + created_at = result.get("created_at", "") + keys.append(created_at or "") + + elif criterion == "date_desc": + # Sort by creation time descending (newest first) + created_at = result.get("created_at", "") + # For descending, we use negative timestamp for reverse sorting + if created_at: + try: + # Parse ISO format datetime and convert to timestamp + dt = datetime.fromisoformat(created_at.replace('Z', '+00:00')) + timestamp = dt.timestamp() + keys.append(-timestamp) # Negative for descending + except (ValueError, TypeError): + # If parsing fails, use the string for comparison + keys.append(created_at) + else: + # Empty dates should sort last in descending order + keys.append(float('inf')) + + elif criterion == "importance_asc": + # Sort by importance score ascending (lower importance first) + metadata = result.get("metadata", {}) + user_metadata = metadata.get("metadata", {}) if metadata else {} + importance = user_metadata.get("importance_score", 0.5) # Default 0.5 if not set + keys.append(float(importance)) + + elif criterion == "importance_desc": + # Sort by importance score descending (higher importance first) + metadata = result.get("metadata", {}) + user_metadata = metadata.get("metadata", {}) if metadata else {} + importance = user_metadata.get("importance_score", 0.5) + keys.append(-float(importance)) + + elif criterion == "access_count_desc": + # Sort by access count descending (most accessed first) + metadata = result.get("metadata", {}) + user_metadata = metadata.get("metadata", {}) if metadata else {} + access_count = user_metadata.get("access_count", 0) + keys.append(-int(access_count)) + + elif criterion == "retention_desc": + # Sort by retention score descending (higher retention first) + metadata = result.get("metadata", {}) + user_metadata = metadata.get("metadata", {}) if metadata else {} + retention = user_metadata.get("retention_score", 0.0) + keys.append(-float(retention)) + + else: + # Unknown criterion, use relevance as fallback + logger.warning(f"Unknown sort criterion: {criterion}, using relevance") + quality_score = result.get("metadata", {}).get("_quality_score", result.get("score", 0.0)) + keys.append(-float(quality_score) if quality_score else 0) + + return tuple(keys) + + # Sort results - use reverse for descending sorts (handled in get_sort_key) + try: + sorted_results = sorted(results, key=get_sort_key, reverse=False) + logger.debug(f"Sorted {len(results)} results by criteria: {sort_keys}") + return sorted_results + except Exception as e: + logger.error(f"Failed to sort results: {e}") + return results def get( self, diff --git a/src/server/api/v1/search.py b/src/server/api/v1/search.py index 47cc38d..60beea4 100644 --- a/src/server/api/v1/search.py +++ b/src/server/api/v1/search.py @@ -2,7 +2,7 @@ Memory search API routes """ -from typing import Optional +from typing import Optional, Union, List from fastapi import APIRouter, Depends, Query, Request from slowapi import Limiter from slowapi.util import get_remote_address @@ -43,6 +43,7 @@ async def search_memories_post( run_id=body.run_id, filters=body.filters, limit=body.limit, + sort_by=body.sort_by, ) search_results = [ @@ -76,6 +77,7 @@ async def search_memories_get( agent_id: Optional[str] = Query(None, description="Filter by agent ID"), run_id: Optional[str] = Query(None, description="Filter by run ID"), limit: int = Query(30, ge=1, le=100, description="Maximum number of results"), + sort_by: Optional[str] = Query(None, description="Sorting criteria: 'relevance', 'date_asc', 'date_desc', 'importance_asc', 'importance_desc', 'access_count_desc', 'retention_desc'"), api_key: str = Depends(verify_api_key), service: SearchService = Depends(get_search_service), ): @@ -87,6 +89,7 @@ async def search_memories_get( run_id=run_id, filters=None, # GET method doesn't support complex filters limit=limit, + sort_by=sort_by, ) search_results = [ diff --git a/src/server/models/request.py b/src/server/models/request.py index f163405..4974411 100644 --- a/src/server/models/request.py +++ b/src/server/models/request.py @@ -2,7 +2,7 @@ Request models for PowerMem API """ -from typing import Any, Dict, List, Optional +from typing import Any, Dict, List, Optional, Union from pydantic import BaseModel, Field @@ -72,6 +72,7 @@ class SearchRequest(BaseModel): run_id: Optional[str] = Field(None, description="Filter by run ID") filters: Optional[Dict[str, Any]] = Field(None, description="Additional filters") limit: int = Field(default=30, ge=1, le=100, description="Maximum number of results") + sort_by: Optional[Union[str, List[str]]] = Field(None, description="Sorting criteria: 'relevance', 'date_asc', 'date_desc', 'importance_asc', 'importance_desc', 'access_count_desc', 'retention_desc', or list for multi-criteria sorting") class UserProfileAddRequest(BaseModel): diff --git a/src/server/services/search_service.py b/src/server/services/search_service.py index ad6a589..423933c 100644 --- a/src/server/services/search_service.py +++ b/src/server/services/search_service.py @@ -3,7 +3,7 @@ """ import logging -from typing import Any, Dict, List, Optional +from typing import Any, Dict, List, Optional, Union from powermem import Memory, auto_config from ..models.errors import ErrorCode, APIError from ..utils.metrics import get_metrics_collector @@ -35,6 +35,7 @@ def search_memories( run_id: Optional[str] = None, filters: Optional[Dict[str, Any]] = None, limit: int = 30, + sort_by: Optional[Union[str, List[str]]] = None, ) -> Dict[str, Any]: """ Search memories. @@ -46,6 +47,9 @@ def search_memories( run_id: Filter by run ID filters: Additional filters limit: Maximum number of results + sort_by: Sorting criteria: 'relevance', 'date_asc', 'date_desc', + 'importance_asc', 'importance_desc', 'access_count_desc', + 'retention_desc', or list for multi-criteria sorting Returns: Search results dictionary @@ -68,6 +72,7 @@ def search_memories( run_id=run_id, filters=filters, limit=limit, + sort_by=sort_by, ) logger.info(f"Search completed: {len(results.get('results', []))} results") diff --git a/test_sorting_simple.py b/test_sorting_simple.py new file mode 100644 index 0000000..53d7b14 --- /dev/null +++ b/test_sorting_simple.py @@ -0,0 +1,135 @@ +#!/usr/bin/env python3 +""" +Simple test to verify search sorting functionality +""" + +def test_sort_function(): + """Test the sorting function logic""" + + from datetime import datetime + + # Simulate the sorting logic from _sort_search_results + def sort_search_results(results, sort_by): + sort_keys = [sort_by] if isinstance(sort_by, str) else sort_by + + def get_sort_key(result): + keys = [] + for criterion in sort_keys: + criterion = criterion.lower().strip() + + if criterion == "relevance" or not criterion: + metadata = result.get("metadata", {}) + quality_score = metadata.get("_quality_score", result.get("score", 0.0)) + keys.append(-float(quality_score) if quality_score else 0) + + elif criterion == "date_asc": + created_at = result.get("created_at", "") + keys.append(created_at or "") + + elif criterion == "date_desc": + created_at = result.get("created_at", "") + if created_at: + try: + dt = datetime.fromisoformat(created_at.replace('Z', '+00:00')) + timestamp = dt.timestamp() + keys.append(-timestamp) + except (ValueError, TypeError): + keys.append(created_at) + else: + keys.append(float('inf')) + + elif criterion == "importance_asc": + metadata = result.get("metadata", {}) + user_metadata = metadata.get("metadata", {}) if metadata else {} + importance = user_metadata.get("importance_score", 0.5) + keys.append(float(importance)) + + elif criterion == "importance_desc": + metadata = result.get("metadata", {}) + user_metadata = metadata.get("metadata", {}) if metadata else {} + importance = user_metadata.get("importance_score", 0.5) + keys.append(-float(importance)) + + elif criterion == "access_count_desc": + metadata = result.get("metadata", {}) + user_metadata = metadata.get("metadata", {}) if metadata else {} + access_count = user_metadata.get("access_count", 0) + keys.append(-int(access_count)) + + elif criterion == "retention_desc": + metadata = result.get("metadata", {}) + user_metadata = metadata.get("metadata", {}) if metadata else {} + retention = user_metadata.get("retention_score", 0.0) + keys.append(-float(retention)) + + else: + quality_score = result.get("metadata", {}).get("_quality_score", result.get("score", 0.0)) + keys.append(-float(quality_score) if quality_score else 0) + + return tuple(keys) + + try: + sorted_results = sorted(results, key=get_sort_key, reverse=False) + return sorted_results + except Exception as e: + print(f"Sort error: {e}") + return results + + # Test 1: Sort by relevance + print("Test 1: Sort by relevance") + test_results = [ + {"memory": "test1", "score": 0.5, "metadata": {"_quality_score": 0.5}}, + {"memory": "test2", "score": 0.9, "metadata": {"_quality_score": 0.9}}, + {"memory": "test3", "score": 0.3, "metadata": {"_quality_score": 0.3}}, + ] + sorted_results = sort_search_results(test_results, "relevance") + assert sorted_results[0]["metadata"]["_quality_score"] == 0.9, f"Expected 0.9, got {sorted_results[0]['metadata']['_quality_score']}" + assert sorted_results[1]["metadata"]["_quality_score"] == 0.5 + assert sorted_results[2]["metadata"]["_quality_score"] == 0.3 + print("✓ Relevance sorting works correctly") + + # Test 2: Sort by date_desc + print("\nTest 2: Sort by date_desc") + test_results = [ + {"memory": "oldest", "created_at": "2024-01-01T10:00:00", "metadata": {}}, + {"memory": "newest", "created_at": "2024-03-01T10:00:00", "metadata": {}}, + {"memory": "middle", "created_at": "2024-02-01T10:00:00", "metadata": {}}, + ] + sorted_results = sort_search_results(test_results, "date_desc") + assert sorted_results[0]["memory"] == "newest", f"Expected newest, got {sorted_results[0]['memory']}" + assert sorted_results[1]["memory"] == "middle" + assert sorted_results[2]["memory"] == "oldest" + print("✓ Date descending sorting works correctly") + + # Test 3: Sort by importance_desc + print("\nTest 3: Sort by importance_desc") + test_results = [ + {"memory": "low", "metadata": {"metadata": {"importance_score": 0.2}}}, + {"memory": "high", "metadata": {"metadata": {"importance_score": 0.9}}}, + {"memory": "medium", "metadata": {"metadata": {"importance_score": 0.5}}}, + ] + sorted_results = sort_search_results(test_results, "importance_desc") + assert sorted_results[0]["memory"] == "high" + assert sorted_results[1]["memory"] == "medium" + assert sorted_results[2]["memory"] == "low" + print("✓ Importance sorting works correctly") + + # Test 4: Multi-criteria sorting + print("\nTest 4: Multi-criteria sorting") + test_results = [ + {"memory": "old_high", "score": 0.9, "created_at": "2024-01-01", "metadata": {"_quality_score": 0.9}}, + {"memory": "new_high", "score": 0.9, "created_at": "2024-03-01", "metadata": {"_quality_score": 0.9}}, + {"memory": "low", "score": 0.5, "created_at": "2024-02-01", "metadata": {"_quality_score": 0.5}}, + ] + sorted_results = sort_search_results(test_results, ["relevance", "date_desc"]) + assert sorted_results[0]["score"] == 0.9 + assert sorted_results[1]["score"] == 0.9 + assert sorted_results[2]["score"] == 0.5 + assert sorted_results[0]["memory"] == "new_high" + assert sorted_results[1]["memory"] == "old_high" + print("✓ Multi-criteria sorting works correctly") + + print("\n✅ All tests passed! Search sorting functionality is working correctly.") + +if __name__ == "__main__": + test_sort_function() diff --git a/tests/unit/test_search_sorting.py b/tests/unit/test_search_sorting.py new file mode 100644 index 0000000..9e010bc --- /dev/null +++ b/tests/unit/test_search_sorting.py @@ -0,0 +1,173 @@ +""" +Test for enhanced search result sorting functionality +""" + +import pytest +from unittest.mock import Mock, MagicMock, patch +from datetime import datetime, timedelta +from powermem.core.memory import Memory + + +class TestSearchSorting: + """Test cases for search result sorting functionality""" + + def test_sort_by_relevance(self): + """Test sorting by relevance (default behavior)""" + memory = Mock(spec=Memory) + + # Create test results with different quality scores + test_results = [ + {"memory": "test1", "score": 0.5, "metadata": {"_quality_score": 0.5}}, + {"memory": "test2", "score": 0.9, "metadata": {"_quality_score": 0.9}}, + {"memory": "test3", "score": 0.3, "metadata": {"_quality_score": 0.3}}, + ] + + # Sort by relevance (should be descending) + sorted_results = memory._sort_search_results(test_results, "relevance") + + # Check that results are sorted by quality score descending + assert sorted_results[0]["metadata"]["_quality_score"] == 0.9 + assert sorted_results[1]["metadata"]["_quality_score"] == 0.5 + assert sorted_results[2]["metadata"]["_quality_score"] == 0.3 + + def test_sort_by_date_desc(self): + """Test sorting by date descending (newest first)""" + memory = Mock(spec=Memory) + + # Create test results with different dates + test_results = [ + {"memory": "oldest", "created_at": "2024-01-01T10:00:00", "metadata": {}}, + {"memory": "newest", "created_at": "2024-03-01T10:00:00", "metadata": {}}, + {"memory": "middle", "created_at": "2024-02-01T10:00:00", "metadata": {}}, + ] + + sorted_results = memory._sort_search_results(test_results, "date_desc") + + # Check that newest is first + assert sorted_results[0]["memory"] == "newest" + assert sorted_results[1]["memory"] == "middle" + assert sorted_results[2]["memory"] == "oldest" + + def test_sort_by_date_asc(self): + """Test sorting by date ascending (oldest first)""" + memory = Mock(spec=Memory) + + test_results = [ + {"memory": "oldest", "created_at": "2024-01-01T10:00:00", "metadata": {}}, + {"memory": "newest", "created_at": "2024-03-01T10:00:00", "metadata": {}}, + {"memory": "middle", "created_at": "2024-02-01T10:00:00", "metadata": {}}, + ] + + sorted_results = memory._sort_search_results(test_results, "date_asc") + + # Check that oldest is first + assert sorted_results[0]["memory"] == "oldest" + assert sorted_results[1]["memory"] == "middle" + assert sorted_results[2]["memory"] == "newest" + + def test_sort_by_importance_desc(self): + """Test sorting by importance score descending""" + memory = Mock(spec=Memory) + + test_results = [ + {"memory": "low", "metadata": {"metadata": {"importance_score": 0.2}}}, + {"memory": "high", "metadata": {"metadata": {"importance_score": 0.9}}}, + {"memory": "medium", "metadata": {"metadata": {"importance_score": 0.5}}}, + ] + + sorted_results = memory._sort_search_results(test_results, "importance_desc") + + # Check that highest importance is first + assert sorted_results[0]["memory"] == "high" + assert sorted_results[1]["memory"] == "medium" + assert sorted_results[2]["memory"] == "low" + + def test_sort_by_access_count_desc(self): + """Test sorting by access count descending""" + memory = Mock(spec=Memory) + + test_results = [ + {"memory": "low", "metadata": {"metadata": {"access_count": 5}}}, + {"memory": "high", "metadata": {"metadata": {"access_count": 100}}}, + {"memory": "medium", "metadata": {"metadata": {"access_count": 25}}}, + ] + + sorted_results = memory._sort_search_results(test_results, "access_count_desc") + + # Check that highest access count is first + assert sorted_results[0]["memory"] == "high" + assert sorted_results[1]["memory"] == "medium" + assert sorted_results[2]["memory"] == "low" + + def test_sort_by_retention_desc(self): + """Test sorting by retention score descending""" + memory = Mock(spec=Memory) + + test_results = [ + {"memory": "low", "metadata": {"metadata": {"retention_score": 0.3}}}, + {"memory": "high", "metadata": {"metadata": {"retention_score": 0.95}}}, + {"memory": "medium", "metadata": {"metadata": {"retention_score": 0.6}}}, + ] + + sorted_results = memory._sort_search_results(test_results, "retention_desc") + + # Check that highest retention is first + assert sorted_results[0]["memory"] == "high" + assert sorted_results[1]["memory"] == "medium" + assert sorted_results[2]["memory"] == "low" + + def test_sort_multi_criteria(self): + """Test multi-criteria sorting (relevance first, then date)""" + memory = Mock(spec=Memory) + + # Results with same relevance but different dates + test_results = [ + {"memory": "old_high", "score": 0.9, "created_at": "2024-01-01", "metadata": {"_quality_score": 0.9}}, + {"memory": "new_high", "score": 0.9, "created_at": "2024-03-01", "metadata": {"_quality_score": 0.9}}, + {"memory": "low", "score": 0.5, "created_at": "2024-02-01", "metadata": {"_quality_score": 0.5}}, + ] + + sorted_results = memory._sort_search_results(test_results, ["relevance", "date_desc"]) + + # Check that high relevance items come first, then sorted by date + assert sorted_results[0]["score"] == 0.9 + assert sorted_results[1]["score"] == 0.9 + assert sorted_results[2]["score"] == 0.5 + + # Among same relevance, newer date should come first + assert sorted_results[0]["memory"] == "new_high" + assert sorted_results[1]["memory"] == "old_high" + + def test_sort_with_missing_metadata(self): + """Test sorting handles missing metadata gracefully""" + memory = Mock(spec=Memory) + + test_results = [ + {"memory": "test1", "metadata": {}}, # Missing quality_score + {"memory": "test2", "metadata": {"_quality_score": 0.8}}, + {"memory": "test3"}, # Missing metadata entirely + ] + + # Should not raise error + sorted_results = memory._sort_search_results(test_results, "relevance") + + assert len(sorted_results) == 3 + + def test_sort_unknown_criterion(self): + """Test sorting with unknown criterion falls back to relevance""" + memory = Mock(spec=Memory) + + test_results = [ + {"memory": "test1", "score": 0.5, "metadata": {"_quality_score": 0.5}}, + {"memory": "test2", "score": 0.9, "metadata": {"_quality_score": 0.9}}, + ] + + sorted_results = memory._sort_search_results(test_results, "unknown_criterion") + + # Should fall back to relevance sorting + assert sorted_results[0]["metadata"]["_quality_score"] == 0.9 + assert sorted_results[1]["metadata"]["_quality_score"] == 0.5 + + +if __name__ == "__main__": + pytest.main([__file__, "-v"])