From 677634f55a78aadfbe42ff70ccb98a55d3513b98 Mon Sep 17 00:00:00 2001 From: spuentesp Date: Mon, 5 Jan 2026 11:17:11 -0300 Subject: [PATCH 1/2] feat(data-layer): DL-10 - Vector Index Operations MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Implements full Qdrant vector storage and semantic search support with 6 MCP tools for embedding operations across scenes, memories, and snippets. ## Qdrant Client (db/qdrant.py) New QdrantClient with singleton pattern: - Auto-creates collections (scenes, memories, snippets) with 1536 dimensions - Supports both remote server and embedded local storage - Environment configuration (QDRANT_URL, QDRANT_API_KEY, QDRANT_PATH) - Lazy collection initialization with ensure_collection() - Cosine distance metric for semantic similarity ## Vector Schemas (schemas/vectors.py) New Pydantic schemas for vector operations: - VectorPoint/VectorUpsertRequest: Single vector storage - VectorBatchUpsertRequest: Batch operations - VectorSearchRequest/Response: Semantic search with filtering - VectorFilter: Multi-field filtering (story_id, scene_id, entity_id, type) - VectorDeleteRequest/ByFilterRequest: Point deletion - CollectionInfo: Collection metadata and stats ## Qdrant Tools (tools/qdrant_tools.py) 6 MCP tools for vector operations: - qdrant_upsert: Store single vector with payload - qdrant_upsert_batch: Efficient batch storage - qdrant_search: Semantic search with score threshold + filters - qdrant_delete: Delete by ID - qdrant_delete_by_filter: Batch delete by filter - qdrant_get_collection_info: Collection stats Helper function _build_qdrant_filter converts VectorFilter to Qdrant Filter DSL. ## Authority Rules (middleware/auth.py) All 6 operations have ["*"] authority (open access): - qdrant_upsert, qdrant_upsert_batch - qdrant_search - qdrant_delete, qdrant_delete_by_filter - qdrant_get_collection_info ## Tests (test_qdrant_tools.py) 15 comprehensive tests with proper mocking: - Upsert: single and batch operations, empty vector/points validation - Search: basic, with filters (story_id, entity_id, type), score threshold, multiple filters - Delete: by ID, by filter, empty filter validation - Collection info: metadata retrieval for populated and empty collections All 294 tests passing ✅ ## Design Decisions - Collections auto-created on first use (no manual setup) - UUID-to-string conversion at Qdrant boundary - Flexible filtering with custom filter support - Count before delete for accurate deletion counts - Pydantic validation catches schema errors early Implements: DL-10 Blocks: DL-7, Q-3, Q-5 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude Sonnet 4.5 --- .../src/monitor_data/db/__init__.py | 5 +- .../data-layer/src/monitor_data/db/qdrant.py | 213 ++++++++ .../src/monitor_data/middleware/auth.py | 12 +- .../src/monitor_data/schemas/__init__.py | 31 ++ .../src/monitor_data/schemas/vectors.py | 180 +++++++ .../src/monitor_data/tools/__init__.py | 16 +- .../src/monitor_data/tools/qdrant_tools.py | 492 +++++++++++++++++ .../tests/test_tools/test_qdrant_tools.py | 504 ++++++++++++++++++ 8 files changed, 1444 insertions(+), 9 deletions(-) create mode 100644 packages/data-layer/src/monitor_data/db/qdrant.py create mode 100644 packages/data-layer/src/monitor_data/schemas/vectors.py create mode 100644 packages/data-layer/src/monitor_data/tools/qdrant_tools.py create mode 100644 packages/data-layer/tests/test_tools/test_qdrant_tools.py diff --git a/packages/data-layer/src/monitor_data/db/__init__.py b/packages/data-layer/src/monitor_data/db/__init__.py index aab5c840..4df4ea00 100644 --- a/packages/data-layer/src/monitor_data/db/__init__.py +++ b/packages/data-layer/src/monitor_data/db/__init__.py @@ -16,8 +16,8 @@ from monitor_data.db.neo4j import Neo4jClient, get_neo4j_client from monitor_data.db.mongodb import MongoDBClient, get_mongodb_client +from monitor_data.db.qdrant import QdrantClient, get_qdrant_client -# from monitor_data.db.qdrant import QdrantClient # from monitor_data.db.minio import MinIOClient __all__ = [ @@ -25,6 +25,7 @@ "get_neo4j_client", "MongoDBClient", "get_mongodb_client", - # "QdrantClient", + "QdrantClient", + "get_qdrant_client", # "MinIOClient", ] diff --git a/packages/data-layer/src/monitor_data/db/qdrant.py b/packages/data-layer/src/monitor_data/db/qdrant.py new file mode 100644 index 00000000..18ea43dc --- /dev/null +++ b/packages/data-layer/src/monitor_data/db/qdrant.py @@ -0,0 +1,213 @@ +""" +Qdrant client for MONITOR Data Layer. + +LAYER: 1 (data-layer) +IMPORTS FROM: External libraries only (qdrant_client) +CALLED BY: qdrant_tools.py + +This client provides a thin wrapper around qdrant_client for: +- Vector storage and retrieval +- Semantic search across narrative content +- Collection management with auto-creation + +Collections: +- scenes: Scene embeddings for semantic search +- memories: Character and agent memory embeddings +- snippets: Document snippet embeddings +""" + +import os +from typing import Any, Dict, Optional +from qdrant_client import QdrantClient as QdrantClientLib +from qdrant_client.models import ( + Distance, + VectorParams, +) + + +# Default embedding dimension (OpenAI text-embedding-ada-002: 1536) +DEFAULT_VECTOR_SIZE = 1536 + +# Collection configurations +COLLECTION_CONFIGS: Dict[str, Dict[str, Any]] = { + "scenes": {"vector_size": DEFAULT_VECTOR_SIZE, "distance": Distance.COSINE}, + "memories": {"vector_size": DEFAULT_VECTOR_SIZE, "distance": Distance.COSINE}, + "snippets": {"vector_size": DEFAULT_VECTOR_SIZE, "distance": Distance.COSINE}, +} + + +class QdrantClient: + """ + Qdrant client for MONITOR vector storage and semantic search. + + Thread-safe singleton client for Qdrant operations. + Manages connection lifecycle and collection auto-creation. + """ + + def __init__( + self, + url: Optional[str] = None, + api_key: Optional[str] = None, + path: Optional[str] = None, + ): + """ + Initialize Qdrant client. + + Args: + url: Qdrant server URL (default: from QDRANT_URL env var) + api_key: Qdrant API key (default: from QDRANT_API_KEY env var, optional) + path: Local storage path for embedded Qdrant (default: from QDRANT_PATH env var) + + Note: + If neither url nor path is provided, defaults to http://localhost:6333 + """ + self.url: Optional[str] = url or os.getenv("QDRANT_URL") + self.api_key: Optional[str] = api_key or os.getenv("QDRANT_API_KEY") + self.path: Optional[str] = path or os.getenv("QDRANT_PATH") + + # If no configuration provided, default to localhost + if not self.url and not self.path: + self.url = "http://localhost:6333" + + self._client: Optional[QdrantClientLib] = None + self._collections_initialized: set = set() + + def connect(self) -> None: + """ + Establish connection to Qdrant. + + Creates client instance with appropriate configuration. + Collections are created lazily on first use. + """ + if self._client is None: + if self.path: + # Local embedded Qdrant + self._client = QdrantClientLib(path=self.path) + else: + # Remote Qdrant server + self._client = QdrantClientLib( + url=self.url, + api_key=self.api_key, + ) + + def close(self) -> None: + """Close Qdrant connection.""" + if self._client: + self._client.close() + self._client = None + self._collections_initialized.clear() + + def verify_connectivity(self) -> bool: + """ + Verify Qdrant connection is working. + + Returns: + True if connection is healthy, False otherwise + """ + try: + if self._client is None: + self.connect() + assert self._client is not None + # Try to get collection list to verify connectivity + self._client.get_collections() + return True + except Exception: + return False + + def get_client(self) -> QdrantClientLib: + """ + Get the Qdrant client object. + + Returns: + qdrant_client QdrantClient object + + Raises: + RuntimeError: If not connected + """ + if self._client is None: + raise RuntimeError("Qdrant client not connected. Call connect() first.") + return self._client + + def ensure_collection(self, collection_name: str) -> None: + """ + Ensure collection exists with correct configuration. + + Creates collection if it doesn't exist, using predefined configs. + + Args: + collection_name: Name of the collection (scenes, memories, snippets) + + Raises: + ValueError: If collection name is not in COLLECTION_CONFIGS + """ + if collection_name in self._collections_initialized: + return + + client = self.get_client() + + # Check if collection exists + try: + client.get_collection(collection_name) + self._collections_initialized.add(collection_name) + return + except Exception: + # Collection doesn't exist, create it + pass + + # Get configuration + if collection_name not in COLLECTION_CONFIGS: + raise ValueError( + f"Unknown collection '{collection_name}'. " + f"Valid collections: {', '.join(COLLECTION_CONFIGS.keys())}" + ) + + config = COLLECTION_CONFIGS[collection_name] + + # Create collection + client.create_collection( + collection_name=collection_name, + vectors_config=VectorParams( + size=int(config["vector_size"]), + distance=Distance(config["distance"]), + ), + ) + + self._collections_initialized.add(collection_name) + + +# ============================================================================= +# SINGLETON ACCESS +# ============================================================================= + +_qdrant_client_instance: Optional[QdrantClient] = None + + +def get_qdrant_client() -> QdrantClient: + """ + Get or create the singleton Qdrant client. + + Returns: + QdrantClient instance + + Thread-safe singleton pattern for database connections. + """ + global _qdrant_client_instance + + if _qdrant_client_instance is None: + _qdrant_client_instance = QdrantClient() + _qdrant_client_instance.connect() + + return _qdrant_client_instance + + +def reset_qdrant_client() -> None: + """ + Reset the Qdrant client singleton. + + Used for testing to ensure clean state between tests. + """ + global _qdrant_client_instance + + if _qdrant_client_instance is not None: + _qdrant_client_instance.close() + _qdrant_client_instance = None diff --git a/packages/data-layer/src/monitor_data/middleware/auth.py b/packages/data-layer/src/monitor_data/middleware/auth.py index 72448511..e55e427c 100644 --- a/packages/data-layer/src/monitor_data/middleware/auth.py +++ b/packages/data-layer/src/monitor_data/middleware/auth.py @@ -149,14 +149,14 @@ "mongodb_get_story_outline": ["*"], "mongodb_update_story_outline": ["Orchestrator"], # ========================================================================= - # QDRANT OPERATIONS - Vectors + # QDRANT OPERATIONS - Vectors (DL-10) # ========================================================================= - "qdrant_embed_scene": ["Indexer"], - "qdrant_embed_memory": ["Indexer"], - "qdrant_embed_snippet": ["Indexer"], + "qdrant_upsert": ["*"], + "qdrant_upsert_batch": ["*"], "qdrant_search": ["*"], - "qdrant_search_memories": ["*"], - "qdrant_delete_vectors": ["Indexer"], + "qdrant_delete": ["*"], + "qdrant_delete_by_filter": ["*"], + "qdrant_get_collection_info": ["*"], # ========================================================================= # NEO4J OPERATIONS - Parties (DL-15) # ========================================================================= diff --git a/packages/data-layer/src/monitor_data/schemas/__init__.py b/packages/data-layer/src/monitor_data/schemas/__init__.py index 00f5fe1a..dae0f3bf 100644 --- a/packages/data-layer/src/monitor_data/schemas/__init__.py +++ b/packages/data-layer/src/monitor_data/schemas/__init__.py @@ -78,6 +78,22 @@ StateTagUpdate, StateTagResponse, ) +from monitor_data.schemas.vectors import ( + VectorPoint, + VectorUpsertRequest, + VectorBatchUpsertRequest, + VectorUpsertResponse, + VectorFilter, + VectorSearchRequest, + ScoredVector, + VectorSearchResponse, + VectorDeleteRequest, + VectorDeleteByFilterRequest, + VectorDeleteResponse, + CollectionInfo, + CollectionInfoRequest, + CollectionInfoResponse, +) # from monitor_data.schemas.entities import * # from monitor_data.schemas.facts import * @@ -141,4 +157,19 @@ "RelationshipListResponse", "StateTagUpdate", "StateTagResponse", + # Vector schemas + "VectorPoint", + "VectorUpsertRequest", + "VectorBatchUpsertRequest", + "VectorUpsertResponse", + "VectorFilter", + "VectorSearchRequest", + "ScoredVector", + "VectorSearchResponse", + "VectorDeleteRequest", + "VectorDeleteByFilterRequest", + "VectorDeleteResponse", + "CollectionInfo", + "CollectionInfoRequest", + "CollectionInfoResponse", ] diff --git a/packages/data-layer/src/monitor_data/schemas/vectors.py b/packages/data-layer/src/monitor_data/schemas/vectors.py new file mode 100644 index 00000000..ca54ec46 --- /dev/null +++ b/packages/data-layer/src/monitor_data/schemas/vectors.py @@ -0,0 +1,180 @@ +""" +Vector embedding schemas for MONITOR Data Layer. + +LAYER: 1 (data-layer) +IMPORTS FROM: External libraries only (pydantic) +USED BY: qdrant_tools.py + +These schemas define the contracts for Qdrant vector operations. +""" + +from typing import Any, Dict, List, Optional +from uuid import UUID +from pydantic import BaseModel, Field + + +# ============================================================================= +# VECTOR POINT SCHEMAS +# ============================================================================= + + +class VectorPoint(BaseModel): + """Single vector point for storage.""" + + id: UUID = Field(description="Unique identifier for the vector point") + vector: List[float] = Field( + description="Embedding vector (typically 1536 dimensions for OpenAI)" + ) + payload: Dict[str, Any] = Field( + default_factory=dict, + description="Metadata payload (id, type, story_id, scene_id, entity_id, etc.)", + ) + + +class VectorUpsertRequest(BaseModel): + """Request to upsert a single vector.""" + + collection: str = Field(description="Collection name (scenes, memories, snippets)") + id: UUID = Field(description="Unique identifier for the vector point") + vector: List[float] = Field(description="Embedding vector") + payload: Dict[str, Any] = Field( + default_factory=dict, description="Metadata payload" + ) + + +class VectorBatchUpsertRequest(BaseModel): + """Request to upsert multiple vectors in batch.""" + + collection: str = Field(description="Collection name (scenes, memories, snippets)") + points: List[VectorPoint] = Field( + description="List of vector points to upsert", min_length=1 + ) + + +class VectorUpsertResponse(BaseModel): + """Response from vector upsert operation.""" + + success: bool = Field(description="Whether the operation succeeded") + collection: str = Field(description="Collection name") + upserted_count: int = Field(description="Number of points upserted") + ids: List[UUID] = Field(description="IDs of upserted points") + + +# ============================================================================= +# VECTOR SEARCH SCHEMAS +# ============================================================================= + + +class VectorFilter(BaseModel): + """Filter for vector search operations.""" + + story_id: Optional[UUID] = Field( + default=None, description="Filter by story_id in payload" + ) + scene_id: Optional[UUID] = Field( + default=None, description="Filter by scene_id in payload" + ) + entity_id: Optional[UUID] = Field( + default=None, description="Filter by entity_id in payload" + ) + type: Optional[str] = Field( + default=None, description="Filter by type in payload (scene, memory, snippet)" + ) + custom: Optional[Dict[str, Any]] = Field( + default=None, + description="Custom Qdrant filter conditions (for advanced filtering)", + ) + + +class VectorSearchRequest(BaseModel): + """Request to search for similar vectors.""" + + collection: str = Field(description="Collection name (scenes, memories, snippets)") + query_vector: List[float] = Field(description="Query embedding vector") + top_k: int = Field(default=10, description="Number of results to return", ge=1) + score_threshold: Optional[float] = Field( + default=None, + description="Minimum similarity score threshold (0-1 for cosine)", + ge=0.0, + le=1.0, + ) + filter: Optional[VectorFilter] = Field( + default=None, description="Optional payload filters" + ) + + +class ScoredVector(BaseModel): + """Single search result with score.""" + + id: UUID = Field(description="Vector point ID") + score: float = Field(description="Similarity score") + payload: Dict[str, Any] = Field( + default_factory=dict, description="Metadata payload" + ) + + +class VectorSearchResponse(BaseModel): + """Response from vector search operation.""" + + collection: str = Field(description="Collection name") + results: List[ScoredVector] = Field(description="Ranked search results") + count: int = Field(description="Number of results returned") + + +# ============================================================================= +# VECTOR DELETE SCHEMAS +# ============================================================================= + + +class VectorDeleteRequest(BaseModel): + """Request to delete a single vector by ID.""" + + collection: str = Field(description="Collection name (scenes, memories, snippets)") + id: UUID = Field(description="ID of the vector point to delete") + + +class VectorDeleteByFilterRequest(BaseModel): + """Request to delete vectors by filter.""" + + collection: str = Field(description="Collection name (scenes, memories, snippets)") + filter: VectorFilter = Field(description="Filter to match points for deletion") + + +class VectorDeleteResponse(BaseModel): + """Response from vector delete operation.""" + + success: bool = Field(description="Whether the operation succeeded") + collection: str = Field(description="Collection name") + deleted_count: int = Field(description="Number of points deleted") + + +# ============================================================================= +# COLLECTION INFO SCHEMAS +# ============================================================================= + + +class CollectionInfo(BaseModel): + """Information about a vector collection.""" + + name: str = Field(description="Collection name") + vector_size: int = Field(description="Dimension of vectors in the collection") + points_count: int = Field(description="Number of points in the collection") + indexed_vectors_count: Optional[int] = Field( + default=None, description="Number of indexed vectors" + ) + distance: str = Field( + description="Distance metric (Cosine, Dot, Euclidean, Manhattan)" + ) + status: str = Field(description="Collection status") + + +class CollectionInfoRequest(BaseModel): + """Request to get collection information.""" + + collection: str = Field(description="Collection name (scenes, memories, snippets)") + + +class CollectionInfoResponse(BaseModel): + """Response with collection information.""" + + collection: CollectionInfo = Field(description="Collection metadata and stats") diff --git a/packages/data-layer/src/monitor_data/tools/__init__.py b/packages/data-layer/src/monitor_data/tools/__init__.py index 85f9d514..06ab51d9 100644 --- a/packages/data-layer/src/monitor_data/tools/__init__.py +++ b/packages/data-layer/src/monitor_data/tools/__init__.py @@ -28,9 +28,16 @@ neo4j_delete_universe, neo4j_ensure_omniverse, ) +from monitor_data.tools.qdrant_tools import ( + qdrant_upsert, + qdrant_upsert_batch, + qdrant_search, + qdrant_delete, + qdrant_delete_by_filter, + qdrant_get_collection_info, +) # from monitor_data.tools.mongodb_tools import * -# from monitor_data.tools.qdrant_tools import * # from monitor_data.tools.composite_tools import * __all__ = [ @@ -43,4 +50,11 @@ "neo4j_update_universe", "neo4j_delete_universe", "neo4j_ensure_omniverse", + # Qdrant Vector tools + "qdrant_upsert", + "qdrant_upsert_batch", + "qdrant_search", + "qdrant_delete", + "qdrant_delete_by_filter", + "qdrant_get_collection_info", ] diff --git a/packages/data-layer/src/monitor_data/tools/qdrant_tools.py b/packages/data-layer/src/monitor_data/tools/qdrant_tools.py new file mode 100644 index 00000000..77a10f42 --- /dev/null +++ b/packages/data-layer/src/monitor_data/tools/qdrant_tools.py @@ -0,0 +1,492 @@ +""" +Qdrant MCP Tools for MONITOR Data Layer. + +LAYER: 1 (data-layer) +IMPORTS FROM: External libraries and data-layer modules only +CALLED BY: Agents (Layer 2) via MCP protocol + +These tools expose Qdrant vector operations via the MCP server. +Qdrant stores embeddings for semantic search across narrative content. +""" + +from typing import Optional +from uuid import UUID + +from qdrant_client.models import ( + PointStruct, + Filter, + FieldCondition, + MatchValue, +) + +from monitor_data.db.qdrant import get_qdrant_client +from monitor_data.schemas.vectors import ( + VectorUpsertRequest, + VectorBatchUpsertRequest, + VectorUpsertResponse, + VectorSearchRequest, + VectorSearchResponse, + ScoredVector, + VectorDeleteRequest, + VectorDeleteByFilterRequest, + VectorDeleteResponse, + CollectionInfoRequest, + CollectionInfoResponse, + CollectionInfo, + VectorFilter, +) + + +# ============================================================================= +# HELPER FUNCTIONS +# ============================================================================= + + +def _build_qdrant_filter(filter_params: VectorFilter) -> Optional[Filter]: + """ + Build Qdrant Filter from VectorFilter parameters. + + Args: + filter_params: VectorFilter with optional story_id, scene_id, entity_id, type + + Returns: + Qdrant Filter object or None if no filters specified + + Examples: + >>> filter = VectorFilter(story_id=uuid, type="scene") + >>> qdrant_filter = _build_qdrant_filter(filter) + >>> # Returns Filter with must conditions for story_id and type + """ + if not filter_params: + return None + + # Start with custom filter if provided + if filter_params.custom: + return Filter(**filter_params.custom) + + must_conditions = [] + + # Add story_id filter + if filter_params.story_id: + must_conditions.append( + FieldCondition( + key="story_id", + match=MatchValue(value=str(filter_params.story_id)), + ) + ) + + # Add scene_id filter + if filter_params.scene_id: + must_conditions.append( + FieldCondition( + key="scene_id", + match=MatchValue(value=str(filter_params.scene_id)), + ) + ) + + # Add entity_id filter + if filter_params.entity_id: + must_conditions.append( + FieldCondition( + key="entity_id", + match=MatchValue(value=str(filter_params.entity_id)), + ) + ) + + # Add type filter + if filter_params.type: + must_conditions.append( + FieldCondition( + key="type", + match=MatchValue(value=filter_params.type), + ) + ) + + if not must_conditions: + return None + + # Use cast to fix type variance issue + return Filter(must=must_conditions) # type: ignore[arg-type] + + +# ============================================================================= +# VECTOR UPSERT OPERATIONS +# ============================================================================= + + +def qdrant_upsert(params: VectorUpsertRequest) -> VectorUpsertResponse: + """ + Store a single vector with payload metadata in Qdrant. + + Creates collection if it doesn't exist. Upserts point (inserts or updates). + + Args: + params: VectorUpsertRequest with collection, id, vector, and payload + + Returns: + VectorUpsertResponse with success status and upserted IDs + + Raises: + ValueError: If collection name is invalid or vector is empty + Exception: If Qdrant operation fails + + Examples: + >>> params = VectorUpsertRequest( + ... collection="scenes", + ... id=scene_id, + ... vector=[0.1, 0.2, ...], # 1536 dims + ... payload={"type": "scene", "story_id": str(story_id)} + ... ) + >>> response = qdrant_upsert(params) + >>> assert response.success is True + >>> assert len(response.ids) == 1 + """ + if not params.vector: + raise ValueError("Vector cannot be empty") + + client = get_qdrant_client() + + # Ensure collection exists with correct configuration + client.ensure_collection(params.collection) + + # Get the underlying Qdrant client + qdrant = client.get_client() + + # Create point + point = PointStruct( + id=str(params.id), + vector=params.vector, + payload=params.payload, + ) + + # Upsert point + qdrant.upsert( + collection_name=params.collection, + points=[point], + ) + + return VectorUpsertResponse( + success=True, + collection=params.collection, + upserted_count=1, + ids=[params.id], + ) + + +def qdrant_upsert_batch(params: VectorBatchUpsertRequest) -> VectorUpsertResponse: + """ + Store multiple vectors in batch for efficient bulk operations. + + Creates collection if it doesn't exist. Upserts all points atomically. + + Args: + params: VectorBatchUpsertRequest with collection and list of points + + Returns: + VectorUpsertResponse with success status and upserted IDs + + Raises: + ValueError: If points list is empty or collection name is invalid + Exception: If Qdrant operation fails + + Examples: + >>> params = VectorBatchUpsertRequest( + ... collection="memories", + ... points=[ + ... VectorPoint(id=id1, vector=[...], payload={...}), + ... VectorPoint(id=id2, vector=[...], payload={...}), + ... ] + ... ) + >>> response = qdrant_upsert_batch(params) + >>> assert response.upserted_count == 2 + """ + if not params.points: + raise ValueError("Points list cannot be empty") + + client = get_qdrant_client() + + # Ensure collection exists + client.ensure_collection(params.collection) + + # Get the underlying Qdrant client + qdrant = client.get_client() + + # Convert to PointStruct list + qdrant_points = [ + PointStruct( + id=str(point.id), + vector=point.vector, + payload=point.payload, + ) + for point in params.points + ] + + # Batch upsert + qdrant.upsert( + collection_name=params.collection, + points=qdrant_points, + ) + + return VectorUpsertResponse( + success=True, + collection=params.collection, + upserted_count=len(params.points), + ids=[point.id for point in params.points], + ) + + +# ============================================================================= +# VECTOR SEARCH OPERATIONS +# ============================================================================= + + +def qdrant_search(params: VectorSearchRequest) -> VectorSearchResponse: + """ + Search for semantically similar vectors with optional filtering. + + Supports payload filtering (story_id, scene_id, entity_id, type) and + score thresholding for high-quality results. + + Args: + params: VectorSearchRequest with query vector, top_k, filters, threshold + + Returns: + VectorSearchResponse with ranked results and scores + + Raises: + ValueError: If collection doesn't exist or query vector is empty + Exception: If Qdrant search fails + + Examples: + >>> params = VectorSearchRequest( + ... collection="scenes", + ... query_vector=[0.1, 0.2, ...], + ... top_k=5, + ... score_threshold=0.7, + ... filter=VectorFilter(story_id=story_id) + ... ) + >>> response = qdrant_search(params) + >>> for result in response.results: + ... print(f"ID: {result.id}, Score: {result.score}") + """ + if not params.query_vector: + raise ValueError("Query vector cannot be empty") + + client = get_qdrant_client() + + # Ensure collection exists + client.ensure_collection(params.collection) + + # Get the underlying Qdrant client + qdrant = client.get_client() + + # Build filter if provided + qdrant_filter = None + if params.filter: + qdrant_filter = _build_qdrant_filter(params.filter) + + # Search + search_results = qdrant.search( # type: ignore[attr-defined] + collection_name=params.collection, + query_vector=params.query_vector, + limit=params.top_k, + query_filter=qdrant_filter, + score_threshold=params.score_threshold, + ) + + # Convert results to ScoredVector + results = [ + ScoredVector( + id=UUID(result.id), # Convert string ID back to UUID + score=result.score, + payload=result.payload or {}, + ) + for result in search_results + ] + + return VectorSearchResponse( + collection=params.collection, + results=results, + count=len(results), + ) + + +# ============================================================================= +# VECTOR DELETE OPERATIONS +# ============================================================================= + + +def qdrant_delete(params: VectorDeleteRequest) -> VectorDeleteResponse: + """ + Delete a single vector point by ID. + + Args: + params: VectorDeleteRequest with collection and point ID + + Returns: + VectorDeleteResponse with success status and deleted count + + Raises: + ValueError: If collection doesn't exist + Exception: If Qdrant operation fails + + Examples: + >>> params = VectorDeleteRequest( + ... collection="scenes", + ... id=scene_id + ... ) + >>> response = qdrant_delete(params) + >>> assert response.deleted_count == 1 + """ + client = get_qdrant_client() + + # Ensure collection exists + client.ensure_collection(params.collection) + + # Get the underlying Qdrant client + qdrant = client.get_client() + + # Delete point by ID + qdrant.delete( + collection_name=params.collection, + points_selector=[str(params.id)], + ) + + return VectorDeleteResponse( + success=True, + collection=params.collection, + deleted_count=1, # Single point deletion + ) + + +def qdrant_delete_by_filter( + params: VectorDeleteByFilterRequest, +) -> VectorDeleteResponse: + """ + Delete multiple vector points matching filter criteria. + + Useful for bulk deletions (e.g., all vectors for a story or scene). + + Args: + params: VectorDeleteByFilterRequest with collection and filter + + Returns: + VectorDeleteResponse with success status and deleted count + + Raises: + ValueError: If collection doesn't exist or filter is empty + Exception: If Qdrant operation fails + + Examples: + >>> params = VectorDeleteByFilterRequest( + ... collection="scenes", + ... filter=VectorFilter(story_id=story_id) + ... ) + >>> response = qdrant_delete_by_filter(params) + >>> print(f"Deleted {response.deleted_count} vectors") + """ + if not params.filter: + raise ValueError("Filter must be provided for delete_by_filter operation") + + client = get_qdrant_client() + + # Ensure collection exists + client.ensure_collection(params.collection) + + # Get the underlying Qdrant client + qdrant = client.get_client() + + # Build filter + qdrant_filter = _build_qdrant_filter(params.filter) + + if not qdrant_filter: + raise ValueError("Filter parameters resulted in empty filter") + + # First, count how many points match the filter + count_result = qdrant.count( + collection_name=params.collection, + count_filter=qdrant_filter, + ) + + deleted_count = count_result.count + + # Delete points matching filter + if deleted_count > 0: + qdrant.delete( + collection_name=params.collection, + points_selector=qdrant_filter, + ) + + return VectorDeleteResponse( + success=True, + collection=params.collection, + deleted_count=deleted_count, + ) + + +# ============================================================================= +# COLLECTION INFO OPERATIONS +# ============================================================================= + + +def qdrant_get_collection_info( + params: CollectionInfoRequest, +) -> CollectionInfoResponse: + """ + Get metadata and statistics about a vector collection. + + Returns collection configuration (dimension, distance metric) and + usage statistics (point count, index status). + + Args: + params: CollectionInfoRequest with collection name + + Returns: + CollectionInfoResponse with collection metadata and stats + + Raises: + ValueError: If collection doesn't exist + Exception: If Qdrant operation fails + + Examples: + >>> params = CollectionInfoRequest(collection="scenes") + >>> response = qdrant_get_collection_info(params) + >>> print(f"Collection: {response.collection.name}") + >>> print(f"Points: {response.collection.points_count}") + >>> print(f"Dimension: {response.collection.vector_size}") + """ + client = get_qdrant_client() + + # Ensure collection exists (creates if needed) + client.ensure_collection(params.collection) + + # Get the underlying Qdrant client + qdrant = client.get_client() + + # Get collection info + collection_info = qdrant.get_collection(params.collection) + + # Extract vector configuration + vectors_config = collection_info.config.params.vectors + + # Handle VectorParams (not dict) + if not hasattr(vectors_config, "size"): + raise ValueError( + f"Invalid vector configuration for collection {params.collection}" + ) + + vector_size = vectors_config.size # type: ignore[attr-defined,union-attr] + distance_name = vectors_config.distance.name # type: ignore[attr-defined,union-attr] + points_count = collection_info.points_count or 0 + + # Extract relevant information + info = CollectionInfo( + name=params.collection, + vector_size=vector_size, + points_count=points_count, + indexed_vectors_count=collection_info.indexed_vectors_count, + distance=distance_name, + status=collection_info.status.name, + ) + + return CollectionInfoResponse(collection=info) diff --git a/packages/data-layer/tests/test_tools/test_qdrant_tools.py b/packages/data-layer/tests/test_tools/test_qdrant_tools.py new file mode 100644 index 00000000..84f147a9 --- /dev/null +++ b/packages/data-layer/tests/test_tools/test_qdrant_tools.py @@ -0,0 +1,504 @@ +""" +Tests for Qdrant vector operations (DL-10). + +Tests all 6 Qdrant operations with proper mocking: +- qdrant_upsert: Store single vector +- qdrant_upsert_batch: Store multiple vectors +- qdrant_search: Semantic search with filtering +- qdrant_delete: Delete by ID +- qdrant_delete_by_filter: Batch delete +- qdrant_get_collection_info: Collection metadata +""" + +import pytest +from unittest.mock import Mock, patch +from uuid import uuid4 + +from monitor_data.tools.qdrant_tools import ( + qdrant_upsert, + qdrant_upsert_batch, + qdrant_search, + qdrant_delete, + qdrant_delete_by_filter, + qdrant_get_collection_info, +) +from monitor_data.schemas.vectors import ( + VectorUpsertRequest, + VectorBatchUpsertRequest, + VectorPoint, + VectorSearchRequest, + VectorFilter, + VectorDeleteRequest, + VectorDeleteByFilterRequest, + CollectionInfoRequest, +) + + +# ============================================================================= +# UPSERT TESTS +# ============================================================================= + + +@patch("monitor_data.tools.qdrant_tools.get_qdrant_client") +def test_upsert_success(mock_get_client: Mock): + """Test successful single vector upsert.""" + # Setup + vector_id = uuid4() + mock_client = Mock() + mock_qdrant = Mock() + mock_get_client.return_value = mock_client + mock_client.get_client.return_value = mock_qdrant + + params = VectorUpsertRequest( + collection="scenes", + id=vector_id, + vector=[0.1] * 1536, + payload={"type": "scene", "story_id": str(uuid4())}, + ) + + # Execute + result = qdrant_upsert(params) + + # Verify + assert result.success is True + assert result.collection == "scenes" + assert result.upserted_count == 1 + assert result.ids == [vector_id] + mock_client.ensure_collection.assert_called_once_with("scenes") + mock_qdrant.upsert.assert_called_once() + + +@patch("monitor_data.tools.qdrant_tools.get_qdrant_client") +def test_upsert_empty_vector(mock_get_client: Mock): + """Test upsert fails with empty vector.""" + params = VectorUpsertRequest( + collection="scenes", + id=uuid4(), + vector=[], # Empty vector + payload={}, + ) + + with pytest.raises(ValueError, match="Vector cannot be empty"): + qdrant_upsert(params) + + +@patch("monitor_data.tools.qdrant_tools.get_qdrant_client") +def test_upsert_batch_success(mock_get_client: Mock): + """Test successful batch vector upsert.""" + # Setup + id1, id2, id3 = uuid4(), uuid4(), uuid4() + mock_client = Mock() + mock_qdrant = Mock() + mock_get_client.return_value = mock_client + mock_client.get_client.return_value = mock_qdrant + + params = VectorBatchUpsertRequest( + collection="memories", + points=[ + VectorPoint(id=id1, vector=[0.1] * 1536, payload={"type": "memory"}), + VectorPoint(id=id2, vector=[0.2] * 1536, payload={"type": "memory"}), + VectorPoint(id=id3, vector=[0.3] * 1536, payload={"type": "memory"}), + ], + ) + + # Execute + result = qdrant_upsert_batch(params) + + # Verify + assert result.success is True + assert result.collection == "memories" + assert result.upserted_count == 3 + assert result.ids == [id1, id2, id3] + mock_client.ensure_collection.assert_called_once_with("memories") + mock_qdrant.upsert.assert_called_once() + + +def test_upsert_batch_empty_points(): + """Test batch upsert fails with empty points list.""" + # Pydantic validation will catch this before reaching the function + from pydantic import ValidationError + + with pytest.raises(ValidationError): + VectorBatchUpsertRequest( + collection="memories", + points=[], # Empty list - Pydantic min_length=1 + ) + + +# ============================================================================= +# SEARCH TESTS +# ============================================================================= + + +@patch("monitor_data.tools.qdrant_tools.get_qdrant_client") +def test_search_basic(mock_get_client: Mock): + """Test basic vector search without filters.""" + # Setup + id1, id2 = uuid4(), uuid4() + mock_client = Mock() + mock_qdrant = Mock() + mock_get_client.return_value = mock_client + mock_client.get_client.return_value = mock_qdrant + + # Mock search results + mock_result1 = Mock() + mock_result1.id = str(id1) + mock_result1.score = 0.95 + mock_result1.payload = {"type": "scene"} + + mock_result2 = Mock() + mock_result2.id = str(id2) + mock_result2.score = 0.87 + mock_result2.payload = {"type": "scene"} + + mock_qdrant.search.return_value = [mock_result1, mock_result2] + + params = VectorSearchRequest( + collection="scenes", + query_vector=[0.1] * 1536, + top_k=5, + ) + + # Execute + result = qdrant_search(params) + + # Verify + assert result.collection == "scenes" + assert result.count == 2 + assert len(result.results) == 2 + assert result.results[0].id == id1 + assert result.results[0].score == 0.95 + assert result.results[1].id == id2 + assert result.results[1].score == 0.87 + mock_client.ensure_collection.assert_called_once_with("scenes") + mock_qdrant.search.assert_called_once() + + +@patch("monitor_data.tools.qdrant_tools.get_qdrant_client") +def test_search_with_filter(mock_get_client: Mock): + """Test vector search with story_id filter.""" + # Setup + story_id = uuid4() + id1 = uuid4() + mock_client = Mock() + mock_qdrant = Mock() + mock_get_client.return_value = mock_client + mock_client.get_client.return_value = mock_qdrant + + mock_result1 = Mock() + mock_result1.id = str(id1) + mock_result1.score = 0.92 + mock_result1.payload = {"type": "scene", "story_id": str(story_id)} + + mock_qdrant.search.return_value = [mock_result1] + + params = VectorSearchRequest( + collection="scenes", + query_vector=[0.1] * 1536, + top_k=10, + filter=VectorFilter(story_id=story_id), + ) + + # Execute + result = qdrant_search(params) + + # Verify + assert result.count == 1 + assert result.results[0].id == id1 + assert result.results[0].payload["story_id"] == str(story_id) + + # Verify filter was passed to search + call_args = mock_qdrant.search.call_args + assert call_args[1]["query_filter"] is not None + + +@patch("monitor_data.tools.qdrant_tools.get_qdrant_client") +def test_search_with_threshold(mock_get_client: Mock): + """Test vector search with score threshold.""" + # Setup + id1 = uuid4() + mock_client = Mock() + mock_qdrant = Mock() + mock_get_client.return_value = mock_client + mock_client.get_client.return_value = mock_qdrant + + mock_result1 = Mock() + mock_result1.id = str(id1) + mock_result1.score = 0.85 + mock_result1.payload = {"type": "memory"} + + mock_qdrant.search.return_value = [mock_result1] + + params = VectorSearchRequest( + collection="memories", + query_vector=[0.2] * 1536, + top_k=5, + score_threshold=0.8, + ) + + # Execute + result = qdrant_search(params) + + # Verify + assert result.count == 1 + assert result.results[0].score >= 0.8 + + # Verify threshold was passed to search + call_args = mock_qdrant.search.call_args + assert call_args[1]["score_threshold"] == 0.8 + + +@patch("monitor_data.tools.qdrant_tools.get_qdrant_client") +def test_search_empty_vector(mock_get_client: Mock): + """Test search fails with empty query vector.""" + params = VectorSearchRequest( + collection="scenes", + query_vector=[], # Empty vector + top_k=5, + ) + + with pytest.raises(ValueError, match="Query vector cannot be empty"): + qdrant_search(params) + + +@patch("monitor_data.tools.qdrant_tools.get_qdrant_client") +def test_search_multiple_filters(mock_get_client: Mock): + """Test vector search with multiple filter conditions.""" + # Setup + story_id = uuid4() + entity_id = uuid4() + id1 = uuid4() + mock_client = Mock() + mock_qdrant = Mock() + mock_get_client.return_value = mock_client + mock_client.get_client.return_value = mock_qdrant + + mock_result1 = Mock() + mock_result1.id = str(id1) + mock_result1.score = 0.90 + mock_result1.payload = { + "type": "scene", + "story_id": str(story_id), + "entity_id": str(entity_id), + } + + mock_qdrant.search.return_value = [mock_result1] + + params = VectorSearchRequest( + collection="scenes", + query_vector=[0.1] * 1536, + top_k=5, + filter=VectorFilter( + story_id=story_id, + entity_id=entity_id, + type="scene", + ), + ) + + # Execute + result = qdrant_search(params) + + # Verify + assert result.count == 1 + assert result.results[0].payload["story_id"] == str(story_id) + assert result.results[0].payload["entity_id"] == str(entity_id) + assert result.results[0].payload["type"] == "scene" + + +# ============================================================================= +# DELETE TESTS +# ============================================================================= + + +@patch("monitor_data.tools.qdrant_tools.get_qdrant_client") +def test_delete_success(mock_get_client: Mock): + """Test successful vector deletion by ID.""" + # Setup + vector_id = uuid4() + mock_client = Mock() + mock_qdrant = Mock() + mock_get_client.return_value = mock_client + mock_client.get_client.return_value = mock_qdrant + + params = VectorDeleteRequest( + collection="scenes", + id=vector_id, + ) + + # Execute + result = qdrant_delete(params) + + # Verify + assert result.success is True + assert result.collection == "scenes" + assert result.deleted_count == 1 + mock_client.ensure_collection.assert_called_once_with("scenes") + mock_qdrant.delete.assert_called_once() + + +@patch("monitor_data.tools.qdrant_tools.get_qdrant_client") +def test_delete_by_filter_success(mock_get_client: Mock): + """Test successful batch deletion by filter.""" + # Setup + story_id = uuid4() + mock_client = Mock() + mock_qdrant = Mock() + mock_get_client.return_value = mock_client + mock_client.get_client.return_value = mock_qdrant + + # Mock count result + mock_count_result = Mock() + mock_count_result.count = 5 + mock_qdrant.count.return_value = mock_count_result + + params = VectorDeleteByFilterRequest( + collection="scenes", + filter=VectorFilter(story_id=story_id), + ) + + # Execute + result = qdrant_delete_by_filter(params) + + # Verify + assert result.success is True + assert result.collection == "scenes" + assert result.deleted_count == 5 + mock_client.ensure_collection.assert_called_once_with("scenes") + mock_qdrant.count.assert_called_once() + mock_qdrant.delete.assert_called_once() + + +@patch("monitor_data.tools.qdrant_tools.get_qdrant_client") +def test_delete_by_filter_no_matches(mock_get_client: Mock): + """Test batch deletion when no points match filter.""" + # Setup + story_id = uuid4() + mock_client = Mock() + mock_qdrant = Mock() + mock_get_client.return_value = mock_client + mock_client.get_client.return_value = mock_qdrant + + # Mock count result - no matches + mock_count_result = Mock() + mock_count_result.count = 0 + mock_qdrant.count.return_value = mock_count_result + + params = VectorDeleteByFilterRequest( + collection="scenes", + filter=VectorFilter(story_id=story_id), + ) + + # Execute + result = qdrant_delete_by_filter(params) + + # Verify + assert result.success is True + assert result.deleted_count == 0 + mock_qdrant.count.assert_called_once() + # Delete should not be called when count is 0 + mock_qdrant.delete.assert_not_called() + + +@patch("monitor_data.tools.qdrant_tools.get_qdrant_client") +def test_delete_by_filter_empty_filter(mock_get_client: Mock): + """Test delete by filter fails with empty filter.""" + mock_client = Mock() + mock_get_client.return_value = mock_client + + params = VectorDeleteByFilterRequest( + collection="scenes", + filter=VectorFilter(), # Empty filter + ) + + with pytest.raises(ValueError, match="Filter parameters resulted in empty filter"): + qdrant_delete_by_filter(params) + + +# ============================================================================= +# COLLECTION INFO TESTS +# ============================================================================= + + +@patch("monitor_data.tools.qdrant_tools.get_qdrant_client") +def test_get_collection_info_success(mock_get_client: Mock): + """Test successful collection info retrieval.""" + # Setup + mock_client = Mock() + mock_qdrant = Mock() + mock_get_client.return_value = mock_client + mock_client.get_client.return_value = mock_qdrant + + # Mock collection info + mock_collection_info = Mock() + mock_collection_info.points_count = 1500 + mock_collection_info.indexed_vectors_count = 1500 + mock_collection_info.status.name = "green" + + # Mock config structure + mock_vectors_config = Mock() + mock_vectors_config.size = 1536 + mock_vectors_config.distance.name = "COSINE" + + mock_params = Mock() + mock_params.vectors = mock_vectors_config + + mock_config = Mock() + mock_config.params = mock_params + + mock_collection_info.config = mock_config + + mock_qdrant.get_collection.return_value = mock_collection_info + + params = CollectionInfoRequest(collection="scenes") + + # Execute + result = qdrant_get_collection_info(params) + + # Verify + assert result.collection.name == "scenes" + assert result.collection.vector_size == 1536 + assert result.collection.points_count == 1500 + assert result.collection.indexed_vectors_count == 1500 + assert result.collection.distance == "COSINE" + assert result.collection.status == "green" + mock_client.ensure_collection.assert_called_once_with("scenes") + mock_qdrant.get_collection.assert_called_once_with("scenes") + + +@patch("monitor_data.tools.qdrant_tools.get_qdrant_client") +def test_get_collection_info_empty_collection(mock_get_client: Mock): + """Test collection info for empty collection.""" + # Setup + mock_client = Mock() + mock_qdrant = Mock() + mock_get_client.return_value = mock_client + mock_client.get_client.return_value = mock_qdrant + + # Mock empty collection info + mock_collection_info = Mock() + mock_collection_info.points_count = 0 + mock_collection_info.indexed_vectors_count = 0 + mock_collection_info.status.name = "green" + + mock_vectors_config = Mock() + mock_vectors_config.size = 1536 + mock_vectors_config.distance.name = "COSINE" + + mock_params = Mock() + mock_params.vectors = mock_vectors_config + + mock_config = Mock() + mock_config.params = mock_params + + mock_collection_info.config = mock_config + + mock_qdrant.get_collection.return_value = mock_collection_info + + params = CollectionInfoRequest(collection="memories") + + # Execute + result = qdrant_get_collection_info(params) + + # Verify + assert result.collection.points_count == 0 + assert result.collection.indexed_vectors_count == 0 From b09f71c239847be7073338e32ef9445b4c606b1f Mon Sep 17 00:00:00 2001 From: spuentesp Date: Mon, 5 Jan 2026 14:15:37 -0300 Subject: [PATCH 2/2] fix(data-layer): Address PR review comments for DL-10 MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Addressed all 9 review comments from PR #104: ## Thread Safety Fixes (P1) - Added thread-safe singleton pattern with double-checked locking - Added threading.Lock for _collections_initialized set - Fixed race conditions in ensure_collection() - Updated docstring to accurately reflect thread-safety ## Authority Restriction (Security) - Restricted qdrant_upsert, qdrant_upsert_batch to ["Indexer"] - Restricted qdrant_delete, qdrant_delete_by_filter to ["Indexer"] - Kept qdrant_search, qdrant_get_collection_info as ["*"] ## Delete Logic Fix - Removed conditional check before delete in delete_by_filter - Now always attempts delete (relies on filter match at delete time) - Count before AND after to calculate accurate deleted_count - Handles race conditions: new points added between operations won't be deleted ## Code Quality Improvements - Removed redundant Distance() conversion (already Distance enum) - Removed redundant empty points validation (Pydantic already enforces) - Added validation for custom filter case - Added error handling for invalid UUID conversion with informative messages ## Test Updates - Updated test_delete_by_filter_success: now counts twice (before/after) - Updated test_delete_by_filter_no_matches: delete always called All 294 tests passing ✅ 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude Sonnet 4.5 --- .../data-layer/src/monitor_data/db/qdrant.py | 78 +++++++++++-------- .../src/monitor_data/middleware/auth.py | 8 +- .../src/monitor_data/tools/qdrant_tools.py | 68 +++++++++++----- .../tests/test_tools/test_qdrant_tools.py | 24 +++--- 4 files changed, 111 insertions(+), 67 deletions(-) diff --git a/packages/data-layer/src/monitor_data/db/qdrant.py b/packages/data-layer/src/monitor_data/db/qdrant.py index 18ea43dc..7c061431 100644 --- a/packages/data-layer/src/monitor_data/db/qdrant.py +++ b/packages/data-layer/src/monitor_data/db/qdrant.py @@ -17,6 +17,7 @@ """ import os +import threading from typing import Any, Dict, Optional from qdrant_client import QdrantClient as QdrantClientLib from qdrant_client.models import ( @@ -40,7 +41,7 @@ class QdrantClient: """ Qdrant client for MONITOR vector storage and semantic search. - Thread-safe singleton client for Qdrant operations. + Client for Qdrant operations. Manages connection lifecycle and collection auto-creation. """ @@ -71,6 +72,7 @@ def __init__( self._client: Optional[QdrantClientLib] = None self._collections_initialized: set = set() + self._collection_lock = threading.Lock() def connect(self) -> None: """ @@ -133,6 +135,7 @@ def ensure_collection(self, collection_name: str) -> None: Ensure collection exists with correct configuration. Creates collection if it doesn't exist, using predefined configs. + Thread-safe using lock to prevent race conditions. Args: collection_name: Name of the collection (scenes, memories, snippets) @@ -140,39 +143,46 @@ def ensure_collection(self, collection_name: str) -> None: Raises: ValueError: If collection name is not in COLLECTION_CONFIGS """ + # Fast path: check without lock if collection_name in self._collections_initialized: return - client = self.get_client() - - # Check if collection exists - try: - client.get_collection(collection_name) - self._collections_initialized.add(collection_name) - return - except Exception: - # Collection doesn't exist, create it - pass - - # Get configuration - if collection_name not in COLLECTION_CONFIGS: - raise ValueError( - f"Unknown collection '{collection_name}'. " - f"Valid collections: {', '.join(COLLECTION_CONFIGS.keys())}" - ) + # Slow path: acquire lock for initialization + with self._collection_lock: + # Double-check after acquiring lock + if collection_name in self._collections_initialized: + return + + client = self.get_client() + + # Check if collection exists + try: + client.get_collection(collection_name) + self._collections_initialized.add(collection_name) + return + except Exception: + # Collection doesn't exist, create it + pass + + # Get configuration + if collection_name not in COLLECTION_CONFIGS: + raise ValueError( + f"Unknown collection '{collection_name}'. " + f"Valid collections: {', '.join(COLLECTION_CONFIGS.keys())}" + ) - config = COLLECTION_CONFIGS[collection_name] + config = COLLECTION_CONFIGS[collection_name] - # Create collection - client.create_collection( - collection_name=collection_name, - vectors_config=VectorParams( - size=int(config["vector_size"]), - distance=Distance(config["distance"]), - ), - ) + # Create collection + client.create_collection( + collection_name=collection_name, + vectors_config=VectorParams( + size=int(config["vector_size"]), + distance=config["distance"], # Already Distance enum + ), + ) - self._collections_initialized.add(collection_name) + self._collections_initialized.add(collection_name) # ============================================================================= @@ -180,6 +190,7 @@ def ensure_collection(self, collection_name: str) -> None: # ============================================================================= _qdrant_client_instance: Optional[QdrantClient] = None +_client_lock = threading.Lock() def get_qdrant_client() -> QdrantClient: @@ -189,13 +200,18 @@ def get_qdrant_client() -> QdrantClient: Returns: QdrantClient instance - Thread-safe singleton pattern for database connections. + Thread-safe singleton pattern using double-checked locking. """ global _qdrant_client_instance + # Fast path: check without lock if _qdrant_client_instance is None: - _qdrant_client_instance = QdrantClient() - _qdrant_client_instance.connect() + # Slow path: acquire lock for initialization + with _client_lock: + # Double-check after acquiring lock + if _qdrant_client_instance is None: + _qdrant_client_instance = QdrantClient() + _qdrant_client_instance.connect() return _qdrant_client_instance diff --git a/packages/data-layer/src/monitor_data/middleware/auth.py b/packages/data-layer/src/monitor_data/middleware/auth.py index e55e427c..12b444b2 100644 --- a/packages/data-layer/src/monitor_data/middleware/auth.py +++ b/packages/data-layer/src/monitor_data/middleware/auth.py @@ -151,11 +151,11 @@ # ========================================================================= # QDRANT OPERATIONS - Vectors (DL-10) # ========================================================================= - "qdrant_upsert": ["*"], - "qdrant_upsert_batch": ["*"], + "qdrant_upsert": ["Indexer"], + "qdrant_upsert_batch": ["Indexer"], "qdrant_search": ["*"], - "qdrant_delete": ["*"], - "qdrant_delete_by_filter": ["*"], + "qdrant_delete": ["Indexer"], + "qdrant_delete_by_filter": ["Indexer"], "qdrant_get_collection_info": ["*"], # ========================================================================= # NEO4J OPERATIONS - Parties (DL-15) diff --git a/packages/data-layer/src/monitor_data/tools/qdrant_tools.py b/packages/data-layer/src/monitor_data/tools/qdrant_tools.py index 77a10f42..3d80bcfa 100644 --- a/packages/data-layer/src/monitor_data/tools/qdrant_tools.py +++ b/packages/data-layer/src/monitor_data/tools/qdrant_tools.py @@ -62,7 +62,20 @@ def _build_qdrant_filter(filter_params: VectorFilter) -> Optional[Filter]: # Start with custom filter if provided if filter_params.custom: - return Filter(**filter_params.custom) + try: + custom_filter = Filter(**filter_params.custom) + # Validate the filter is not empty + if ( + not custom_filter.must + and not custom_filter.should + and not custom_filter.must_not + ): + return None + return custom_filter + except Exception as exc: + raise ValueError( + f"Invalid custom filter parameters: {filter_params.custom}" + ) from exc must_conditions = [] @@ -200,9 +213,7 @@ def qdrant_upsert_batch(params: VectorBatchUpsertRequest) -> VectorUpsertRespons >>> response = qdrant_upsert_batch(params) >>> assert response.upserted_count == 2 """ - if not params.points: - raise ValueError("Points list cannot be empty") - + # Note: Pydantic schema already enforces min_length=1 for points client = get_qdrant_client() # Ensure collection exists @@ -294,15 +305,23 @@ def qdrant_search(params: VectorSearchRequest) -> VectorSearchResponse: score_threshold=params.score_threshold, ) - # Convert results to ScoredVector - results = [ - ScoredVector( - id=UUID(result.id), # Convert string ID back to UUID - score=result.score, - payload=result.payload or {}, + # Convert results to ScoredVector with error handling + results = [] + for result in search_results: + try: + vector_id = UUID(str(result.id)) # Convert string ID back to UUID + except (ValueError, TypeError) as exc: + raise ValueError( + f"Invalid UUID '{result.id}' returned from Qdrant " + f"for collection '{params.collection}'." + ) from exc + results.append( + ScoredVector( + id=vector_id, + score=result.score, + payload=result.payload or {}, + ) ) - for result in search_results - ] return VectorSearchResponse( collection=params.collection, @@ -402,20 +421,27 @@ def qdrant_delete_by_filter( if not qdrant_filter: raise ValueError("Filter parameters resulted in empty filter") - # First, count how many points match the filter - count_result = qdrant.count( + # Count before delete for reporting + count_before = qdrant.count( # type: ignore[attr-defined] collection_name=params.collection, count_filter=qdrant_filter, + ).count + + # Delete points matching filter (always attempt; rely on filter match at delete time) + qdrant.delete( # type: ignore[attr-defined] + collection_name=params.collection, + points_selector=qdrant_filter, ) - deleted_count = count_result.count + # Count after delete to calculate actual deleted count + # This handles race conditions: if points were added between operations, + # they won't be deleted, and the count difference will be accurate + count_after = qdrant.count( # type: ignore[attr-defined] + collection_name=params.collection, + count_filter=qdrant_filter, + ).count - # Delete points matching filter - if deleted_count > 0: - qdrant.delete( - collection_name=params.collection, - points_selector=qdrant_filter, - ) + deleted_count = count_before - count_after return VectorDeleteResponse( success=True, diff --git a/packages/data-layer/tests/test_tools/test_qdrant_tools.py b/packages/data-layer/tests/test_tools/test_qdrant_tools.py index 84f147a9..0a74b431 100644 --- a/packages/data-layer/tests/test_tools/test_qdrant_tools.py +++ b/packages/data-layer/tests/test_tools/test_qdrant_tools.py @@ -346,10 +346,12 @@ def test_delete_by_filter_success(mock_get_client: Mock): mock_get_client.return_value = mock_client mock_client.get_client.return_value = mock_qdrant - # Mock count result - mock_count_result = Mock() - mock_count_result.count = 5 - mock_qdrant.count.return_value = mock_count_result + # Mock count results - before: 5, after: 0 (all deleted) + mock_count_before = Mock() + mock_count_before.count = 5 + mock_count_after = Mock() + mock_count_after.count = 0 + mock_qdrant.count.side_effect = [mock_count_before, mock_count_after] params = VectorDeleteByFilterRequest( collection="scenes", @@ -362,9 +364,9 @@ def test_delete_by_filter_success(mock_get_client: Mock): # Verify assert result.success is True assert result.collection == "scenes" - assert result.deleted_count == 5 + assert result.deleted_count == 5 # 5 - 0 = 5 deleted mock_client.ensure_collection.assert_called_once_with("scenes") - mock_qdrant.count.assert_called_once() + assert mock_qdrant.count.call_count == 2 # Called before and after mock_qdrant.delete.assert_called_once() @@ -378,7 +380,7 @@ def test_delete_by_filter_no_matches(mock_get_client: Mock): mock_get_client.return_value = mock_client mock_client.get_client.return_value = mock_qdrant - # Mock count result - no matches + # Mock count results - before: 0, after: 0 (nothing to delete) mock_count_result = Mock() mock_count_result.count = 0 mock_qdrant.count.return_value = mock_count_result @@ -393,10 +395,10 @@ def test_delete_by_filter_no_matches(mock_get_client: Mock): # Verify assert result.success is True - assert result.deleted_count == 0 - mock_qdrant.count.assert_called_once() - # Delete should not be called when count is 0 - mock_qdrant.delete.assert_not_called() + assert result.deleted_count == 0 # 0 - 0 = 0 deleted + assert mock_qdrant.count.call_count == 2 # Called before and after + # Delete is always called regardless of count + mock_qdrant.delete.assert_called_once() @patch("monitor_data.tools.qdrant_tools.get_qdrant_client")