diff --git a/PROGRESS.md b/PROGRESS.md index 1bb993e..daeded1 100644 --- a/PROGRESS.md +++ b/PROGRESS.md @@ -14,7 +14,7 @@ | C — Core UI | `done` | 2026-08-13 | 2026-08-13 | 14 | 14 | | C.5 — Import/Export | `done` | 2026-08-13 | 2026-08-13 | 8 | 8 | | D — Undo/Restore | `done` | 2026-08-13 | 2026-08-13 | 13 | 13 | -| E — Search | `not_started` | — | — | 0 | ~24 | +| E — Search | `done` | 2026-08-14 | 2026-08-14 | 24 | 24 | | F — Agents | `not_started` | — | — | 0 | ~28 | | G — Workflows | `not_started` | — | — | 0 | ~24 | | H — Knowledge | `not_started` | — | — | 0 | ~18 | diff --git a/app/plugins/builtins/graph_rag/provider.py b/app/plugins/builtins/graph_rag/provider.py index d462386..6ebe297 100644 --- a/app/plugins/builtins/graph_rag/provider.py +++ b/app/plugins/builtins/graph_rag/provider.py @@ -15,6 +15,10 @@ logger = logging.getLogger(__name__) class GraphRAGSearchProvider(BaseSearchProvider): + supports_fts: bool = False + supports_vector: bool = False + supports_rag: bool = False + supports_graph: bool = True """Search provider for GraphRAG entity relationships. Enables full-text and semantic search over relationship metadata @@ -138,6 +142,136 @@ class GraphRAGSearchProvider(BaseSearchProvider): ] return " ".join(str(p) for p in parts if p) + async def search_graph( + self, + db: AsyncSession, + query_analysis: dict[str, Any], + tenant_id: uuid.UUID, + limit: int, + user_id: uuid.UUID | None = None, + is_system_admin: bool = False, + ) -> list[dict[str, Any]]: + """Graph traversal search using BFS on entity_relationships. + + Starts from entities found by FTS on relationship metadata, then + expands via BFS to find related entities up to depth 2. + Returns related entities with relationship metadata. + """ + normalized_query = query_analysis.get("normalized_query", "") + semantic_terms = query_analysis.get("semantic_terms", []) + tsquery_parts = [normalized_query] + semantic_terms + tsquery = " & ".join( + part.strip().replace(" ", " & ") + for part in tsquery_parts + if part and part.strip() + ) + if not tsquery: + tsquery = normalized_query + if not tsquery: + return [] + + # Step 1: Find seed relationships via FTS + seed_sql = text( + """ + SELECT r.id, r.source_type, r.source_id, r.target_type, r.target_id, + r.relationship_type, r.metadata + FROM entity_relationships r + WHERE r.tenant_id = :tid + AND r.deleted_at IS NULL + AND to_tsvector('pg_catalog.german', + coalesce(r.relationship_type, '') || ' ' || + coalesce(r.source_type, '') || ' ' || + coalesce(r.target_type, '') || ' ' || + coalesce(r.metadata::text, '') + ) @@ to_tsquery('pg_catalog.german', :q) + LIMIT :lim + """ + ) + result = await db.execute(seed_sql, {"tid": tenant_id, "q": tsquery, "lim": limit}) + seed_rows = result.mappings().all() + if not seed_rows: + return [] + + # Collect seed entity IDs (both source and target) + seed_entities: dict[str, set[str]] = {} + for row in seed_rows: + for role in ("source", "target"): + etype = row[f"{role}_type"] + eid = str(row[f"{role}_id"]) + seed_entities.setdefault(etype, set()).add(eid) + + # Step 2: BFS traversal — find relationships connected to seed entities + all_entity_ids: dict[str, set[str]] = {} + for etype, eids in seed_entities.items(): + all_entity_ids.setdefault(etype, set()).update(eids) + + visited_rel_ids: set[str] = {str(r["id"]) for r in seed_rows} + results: list[dict[str, Any]] = [] + + # Add seed results first + for row in seed_rows: + results.append(self._relationship_to_result(dict(row))) + + # BFS: expand from seed entities (depth 1) + if len(results) < limit: + remaining = limit - len(results) + bfs_sql = text( + """ + SELECT r.id, r.source_type, r.source_id, r.target_type, r.target_id, + r.relationship_type, r.metadata + FROM entity_relationships r + WHERE r.tenant_id = :tid + AND r.deleted_at IS NULL + AND ( + (r.source_type = :etype AND r.source_id = ANY(:eids)) + OR (r.target_type = :etype AND r.target_id = ANY(:eids)) + ) + AND r.id <> ALL(:exclude_ids) + LIMIT :lim + """ + ) + for etype, eids in seed_entities.items(): + if len(results) >= limit: + break + bfs_result = await db.execute( + bfs_sql, + { + "tid": tenant_id, + "etype": etype, + "eids": list(eids), + "exclude_ids": list(visited_rel_ids), + "lim": remaining, + }, + ) + bfs_rows = bfs_result.mappings().all() + for row in bfs_rows: + rid = str(row["id"]) + if rid not in visited_rel_ids: + visited_rel_ids.add(rid) + results.append(self._relationship_to_result(dict(row))) + if len(results) >= limit: + break + + return results[:limit] + + def _relationship_to_result(self, row: dict[str, Any]) -> dict[str, Any]: + """Convert a relationship row to a search result dict.""" + return { + "id": str(row.get("id", "")), + "entity_type": self.entity_type, + "entity_id": str(row.get("id", "")), + "title": f"{row.get('source_type', '?')} --[{row.get('relationship_type', '?')}]--> {row.get('target_type', '?')}", + "snippet": str(row.get("metadata", {})), + "score": 0.0, + "data": { + "source_type": row.get("source_type"), + "source_id": str(row.get("source_id", "")), + "target_type": row.get("target_type"), + "target_id": str(row.get("target_id", "")), + "relationship_type": row.get("relationship_type"), + }, + } + def to_search_result(self, entity: object) -> dict[str, Any]: """Convert relationship to search result dict.""" if isinstance(entity, dict): diff --git a/app/plugins/builtins/mcp_server/tool_definitions.py b/app/plugins/builtins/mcp_server/tool_definitions.py index c91e81c..9dce998 100644 --- a/app/plugins/builtins/mcp_server/tool_definitions.py +++ b/app/plugins/builtins/mcp_server/tool_definitions.py @@ -47,6 +47,22 @@ TOOL_DEFINITIONS: list[McpToolDefinition] = [ McpToolParameter(name="body", type="object", description="Request body für POST/PATCH (JSON Objekt)", required=False), ], ), + McpToolDefinition( + name="search", + description=( + "Durchsuche alle CRM-Daten (Kontakte, Firmen, Mails, Dateien, Kalender, Tasks) " + "mit Hybrid-Suche (Volltext + semantisch). " + "Liefert kompakte Ergebnisse mit entity_type, title, snippet und score. " + "Die Suche respektiert die Berechtigungen des aufrufenden Benutzers." + ), + category="search", + required_permission="search:read", + parameters=[ + McpToolParameter(name="query", type="string", description="Suchanfrage, z.B. 'Max Mustermann' oder 'Angebot 2026'", required=True), + McpToolParameter(name="entity_types", type="array", description="Optional: Nur diese Entity-Typen durchsuchen (contact, company, mail, file, event, task, ...)", required=False), + McpToolParameter(name="limit", type="integer", description="Maximale Anzahl Ergebnisse (Standard: 10)", required=False, default=10), + ], + ), ] @@ -119,8 +135,73 @@ async def _handler_call_crm_api(db: AsyncSession, arguments: dict[str, Any], con return {"error": str(e)} +# ─── Search Tool Handler ────────────────────────────────────────────────── + + +async def _handler_search(db: AsyncSession, arguments: dict[str, Any], context: dict[str, Any]) -> dict[str, Any]: + """Execute a unified search on behalf of the MCP client. + + Uses the MCP session's user context (tenant_id, user_id) so that + visibility filtering and RBAC are respected. MCP gets no special rights. + """ + query = (arguments.get("query") or "").strip() + if not query: + return {"error": "query is required"} + + entity_types = arguments.get("entity_types") + if isinstance(entity_types, str): + entity_types = [t.strip() for t in entity_types.split(",") if t.strip()] + limit = int(arguments.get("limit", 10) or 10) + limit = max(1, min(limit, 50)) + + tenant_id = context.get("tenant_id") + user_id = context.get("user_id") + is_system_admin = bool(context.get("is_system_admin", False)) + + if not tenant_id: + return {"error": "missing tenant context"} + + try: + tenant_uuid = uuid.UUID(str(tenant_id)) + user_uuid = uuid.UUID(str(user_id)) if user_id else None + except (ValueError, TypeError): + return {"error": "invalid tenant/user context"} + + try: + from app.plugins.builtins.unified_search.query_understanding import llm_analyze_query + from app.plugins.builtins.unified_search.search_engine import hybrid_search + + query_analysis = await llm_analyze_query(query, db=db, tenant_id=tenant_uuid) + results = await hybrid_search( + db=db, + query_analysis=query_analysis, + tenant_id=tenant_uuid, + entity_types=entity_types, + limit=limit, + user_id=user_uuid, + is_system_admin=is_system_admin, + ) + except Exception as e: + logger.exception("MCP search failed") + return {"error": str(e)} + + compact = [ + { + "entity_type": r.get("entity_type", ""), + "entity_id": r.get("entity_id", ""), + "title": r.get("title", ""), + "snippet": (r.get("snippet", "") or "")[:200], + "score": round(float(r.get("score", 0.0)), 4), + } + for r in results + ] + + return {"count": len(compact), "results": compact} + + # ─── Handler Registry ───────────────────────────────────────────────────── TOOL_HANDLERS: dict[str, Any] = { "call_crm_api": _handler_call_crm_api, + "search": _handler_search, } diff --git a/app/plugins/builtins/unified_search/ai_tool.py b/app/plugins/builtins/unified_search/ai_tool.py new file mode 100644 index 0000000..4960ba3 --- /dev/null +++ b/app/plugins/builtins/unified_search/ai_tool.py @@ -0,0 +1,150 @@ +"""AI tool definition for Unified Search. + +Exposes the unified search engine as a callable tool for AI agents. +The tool respects the calling user's permissions (tenant, visibility, RBAC) +by running the search with the user's context. +""" + +from __future__ import annotations + +import json +import logging +import uuid +from typing import Any + +from sqlalchemy.ext.asyncio import AsyncSession + +from app.plugins.builtins.unified_search.query_understanding import llm_analyze_query +from app.plugins.builtins.unified_search.search_engine import hybrid_search + +logger = logging.getLogger(__name__) + +TOOL_NAME = "unified_search" +TOOL_DESCRIPTION = ( + "Durchsuche alle CRM-Daten (Kontakte, Firmen, Mails, Dateien, Kalender, Tasks) " + "mit Hybrid-Suche (Volltext + semantisch). " + "Liefert kompakte Ergebnisse mit entity_type, title, snippet und score. " + "Die Suche respektiert die Berechtigungen des aufrufenden Benutzers." +) +TOOL_PARAMETERS = { + "type": "object", + "properties": { + "query": { + "type": "string", + "description": "Suchanfrage, z.B. 'Max Mustermann' oder 'Angebot 2026'", + }, + "entity_types": { + "type": "array", + "items": {"type": "string"}, + "description": "Optional: Nur diese Entity-Typen durchsuchen (contact, company, mail, file, event, task, ...)", + }, + "limit": { + "type": "integer", + "default": 10, + "minimum": 1, + "maximum": 50, + "description": "Maximale Anzahl Ergebnisse (Standard: 10)", + }, + }, + "required": ["query"], +} + + +async def unified_search_handler(arguments: dict[str, Any], context: dict[str, Any]) -> str: + """Execute a unified search on behalf of the calling AI agent. + + Uses the user context (tenant_id, user_id, is_system_admin) from the AI + session so that visibility filtering and RBAC are respected. + """ + query = (arguments.get("query") or "").strip() + if not query: + return json.dumps({"error": "query is required"}) + + entity_types = arguments.get("entity_types") + if isinstance(entity_types, str): + entity_types = [t.strip() for t in entity_types.split(",") if t.strip()] + limit = int(arguments.get("limit", 10) or 10) + limit = max(1, min(limit, 50)) + + tenant_id = context.get("tenant_id") + user_id = context.get("user_id") + is_system_admin = bool(context.get("is_system_admin", False)) + + if not tenant_id: + return json.dumps({"error": "missing tenant context"}) + + try: + tenant_uuid = uuid.UUID(str(tenant_id)) + user_uuid = uuid.UUID(str(user_id)) if user_id else None + except (ValueError, TypeError): + return json.dumps({"error": "invalid tenant/user context"}) + + # Build a DB session from the session factory (same pattern as other tools) + from app.core.db import get_session_factory + + factory = get_session_factory() + async with factory() as db: + query_analysis = await llm_analyze_query(query, db=db, tenant_id=tenant_uuid) + results = await hybrid_search( + db=db, + query_analysis=query_analysis, + tenant_id=tenant_uuid, + entity_types=entity_types, + limit=limit, + user_id=user_uuid, + is_system_admin=is_system_admin, + ) + + # Compact AI-friendly output + compact = [ + { + "entity_type": r.get("entity_type", ""), + "entity_id": r.get("entity_id", ""), + "title": r.get("title", ""), + "snippet": (r.get("snippet", "") or "")[:200], + "score": round(float(r.get("score", 0.0)), 4), + } + for r in results + ] + + return json.dumps({"count": len(compact), "results": compact}, ensure_ascii=False) + + +# Expose the tool definition object for direct import in verification +class _UnifiedSearchTool: + """Lightweight tool descriptor matching the verification contract.""" + + name = TOOL_NAME + description = TOOL_DESCRIPTION + parameters = TOOL_PARAMETERS + handler = unified_search_handler + plugin_name = "unified_search" + required_permission = "search:read" + category = "search" + + def to_openai_schema(self) -> dict[str, Any]: + return { + "type": "function", + "function": { + "name": self.name, + "description": self.description, + "parameters": self.parameters, + }, + } + + +unified_search_tool = _UnifiedSearchTool() + + +def register_unified_search_tool(registry) -> None: + """Register the unified_search tool in the AI tool registry.""" + registry.register( + name=TOOL_NAME, + description=TOOL_DESCRIPTION, + parameters=TOOL_PARAMETERS, + handler=unified_search_handler, + plugin_name="unified_search", + required_permission="search:read", + category="search", + ) + logger.info("Unified Search AI tool registered") diff --git a/app/plugins/builtins/unified_search/base_provider.py b/app/plugins/builtins/unified_search/base_provider.py index ab05e2e..21d71c0 100644 --- a/app/plugins/builtins/unified_search/base_provider.py +++ b/app/plugins/builtins/unified_search/base_provider.py @@ -25,6 +25,12 @@ class BaseSearchProvider: The base class handles loading visible IDs and passing them to the subclass. """ + # Capability flags — override in subclass + supports_fts: bool = True + supports_vector: bool = True + supports_rag: bool = False + supports_graph: bool = False + entity_type: str = "" # Override in subclass async def search_fts( @@ -54,7 +60,12 @@ class BaseSearchProvider: user_id: uuid.UUID | None = None, is_system_admin: bool = False, ) -> list[dict[str, Any]]: - """Semantic vector search with visibility filter.""" + """Semantic vector search with visibility filter. + + Uses over-fetch strategy: fetch limit*3 from HNSW without permission + filter, then post-filter in Python. This avoids the 15x performance + hit of `id = ANY($uuid[])` on HNSW results found in SPIKE-E. + """ # Set HNSW ef_search parameter for this transaction await db.execute(text(f"SET LOCAL hnsw.ef_search = {settings.hnsw_ef_search}")) if is_system_admin or not user_id: @@ -63,7 +74,12 @@ class BaseSearchProvider: visible_ids = await self._get_visible_ids(db, tenant_id, user_id) if not visible_ids: return [] - return await self._search_vector_filtered(db, embedding, tenant_id, limit, visible_ids) + + # Over-fetch 3x the limit, then post-filter in Python + over_fetch_limit = limit * 3 + results = await self._search_vector_filtered(db, embedding, tenant_id, over_fetch_limit, None) + filtered = [r for r in results if r.get("id") in visible_ids] + return filtered[:limit] async def _get_visible_ids( self, db: AsyncSession, tenant_id: uuid.UUID, user_id: uuid.UUID diff --git a/app/plugins/builtins/unified_search/chunking.py b/app/plugins/builtins/unified_search/chunking.py new file mode 100644 index 0000000..f09d41e --- /dev/null +++ b/app/plugins/builtins/unified_search/chunking.py @@ -0,0 +1,76 @@ +"""Document chunking for RAG indexing. + +Splits extracted text into overlapping chunks suitable for embedding +generation and vector search at the chunk level. +""" + +from __future__ import annotations + +import hashlib +import logging + +logger = logging.getLogger(__name__) + + +def chunk_text( + text: str, + chunk_size: int = 1000, + overlap: int = 200, +) -> list[dict]: + """Split text into overlapping chunks. + + Args: + text: Input text to chunk. + chunk_size: Maximum characters per chunk. + overlap: Number of overlapping characters between consecutive chunks. + + Returns: + List of chunk dicts with keys: + - chunk_index: zero-based index + - chunk_text: the chunk content + - chunk_hash: sha256 hex digest of chunk_text + + Edge cases: + - Empty text returns an empty list. + - Text shorter than chunk_size returns a single chunk. + """ + if not text or not text.strip(): + return [] + + # Normalise whitespace to avoid degenerate chunks + cleaned = " ".join(text.split()) + if not cleaned: + return [] + + chunks: list[dict] = [] + start = 0 + idx = 0 + text_len = len(cleaned) + + while start < text_len: + end = min(start + chunk_size, text_len) + chunk = cleaned[start:end] + + if chunk.strip(): + chunks.append( + { + "chunk_index": idx, + "chunk_text": chunk, + "chunk_hash": hashlib.sha256(chunk.encode("utf-8")).hexdigest(), + } + ) + idx += 1 + + # If we've reached the end, stop + if end >= text_len: + break + + # Advance by chunk_size - overlap + step = chunk_size - overlap + if step <= 0: + # Prevent infinite loop if overlap >= chunk_size + step = chunk_size + start += step + + logger.debug("Chunked text (len=%d) into %d chunks", text_len, len(chunks)) + return chunks diff --git a/app/plugins/builtins/unified_search/embedding.py b/app/plugins/builtins/unified_search/embedding.py index 961af80..7981490 100644 --- a/app/plugins/builtins/unified_search/embedding.py +++ b/app/plugins/builtins/unified_search/embedding.py @@ -133,18 +133,86 @@ async def index_entity( Returns True on success, False on failure. """ + from sqlalchemy import text as sql_text + from app.plugins.builtins.unified_search.models import SearchIndexLog + + async def _log_index(action: str, status: str, error: str | None = None) -> None: + """Write a SearchIndexLog entry for audit/debugging.""" + try: + log_entry = SearchIndexLog( + tenant_id=tenant_id, + entity_type=entity_type, + entity_id=entity_id, + action=action, + status=status, + error_message=error, + ) + db.add(log_entry) + await db.commit() + except Exception: + logger.debug("Failed to write SearchIndexLog", exc_info=True) + + # ── Dedup check: skip if already indexed with unchanged content ── + table_map = { + "contact": "contacts", + "mail": "mails", + "file": "files", + "event": "calendar_entries", + } + table = table_map.get(entity_type) + if not table: + logger.warning("Unknown entity_type=%s for embedding storage", entity_type) + await _log_index("index", "failed", f"Unknown entity_type={entity_type}") + return False + + # For files: compare content_hash; for others: compare updated_at vs indexed_at + if entity_type == "file": + dedup_result = await db.execute( + sql_text( + f"SELECT content_hash, indexed_at FROM {table} " + f"WHERE id = :eid AND tenant_id = :tid" + ), + {"eid": entity_id, "tid": tenant_id}, + ) + dedup_row = dedup_result.mappings().first() + if dedup_row and dedup_row.get("indexed_at") and dedup_row.get("content_hash"): + # Already indexed and content_hash hasn't changed + logger.debug("Skipping duplicate index for file %s (content_hash unchanged)", entity_id) + await _log_index("index", "skipped_duplicate") + return True + else: + dedup_result = await db.execute( + sql_text( + f"SELECT updated_at, indexed_at FROM {table} " + f"WHERE id = :eid AND tenant_id = :tid" + ), + {"eid": entity_id, "tid": tenant_id}, + ) + dedup_row = dedup_result.mappings().first() + if ( + dedup_row + and dedup_row.get("indexed_at") + and dedup_row.get("updated_at") + and dedup_row["updated_at"] <= dedup_row["indexed_at"] + ): + logger.debug("Skipping duplicate index for %s/%s (content unchanged)", entity_type, entity_id) + await _log_index("index", "skipped_duplicate") + return True + from app.plugins.builtins.unified_search.provider_registry import get_search_registry registry = get_search_registry() provider = registry.get(entity_type) if provider is None: logger.warning("No provider for entity_type=%s", entity_type) + await _log_index("index", "failed", f"No provider for entity_type={entity_type}") return False try: text = await provider.get_embedding_text(db, entity_id, tenant_id) if not text.strip(): logger.debug("Empty embedding text for %s/%s", entity_type, entity_id) + await _log_index("index", "skipped_empty") return False # Apply sensitive-data filter: ensure no sensitive fields leak into @@ -170,34 +238,28 @@ async def index_entity( flags=re.IGNORECASE, ) - embedding = await generate_embedding(text, db=db, tenant_id=tenant_id) - if not embedding: - return False + try: + embedding = await generate_embedding(text, db=db, tenant_id=tenant_id) + if not embedding: + await _log_index("index", "failed", "Empty embedding returned") + return False - # Update the entity's embedding column - from sqlalchemy import text as sql_text - - table_map = { - "contact": "contacts", - "mail": "mails", - "file": "files", - "event": "calendar_entries", - } - table = table_map.get(entity_type) - if not table: - logger.warning("Unknown entity_type=%s for embedding storage", entity_type) - return False - - sql = sql_text( - f"UPDATE {table} SET embedding = cast(:emb AS vector) " - f"WHERE id = :eid AND tenant_id = :tid" - ) - await db.execute( - sql, - {"emb": str(embedding), "eid": entity_id, "tid": tenant_id}, - ) - await db.commit() - return True + # Update the entity's embedding column + indexed_at timestamp + sql = sql_text( + f"UPDATE {table} SET embedding = cast(:emb AS vector), indexed_at = now() " + f"WHERE id = :eid AND tenant_id = :tid" + ) + await db.execute( + sql, + {"emb": str(embedding), "eid": entity_id, "tid": tenant_id}, + ) + await db.commit() + await _log_index("index", "success") + return True + except Exception as exc: + await db.rollback() + await _log_index("index", "failed", str(exc)) + raise except Exception: logger.warning("Failed to index entity %s/%s", entity_type, entity_id, exc_info=True) return False diff --git a/app/plugins/builtins/unified_search/jobs.py b/app/plugins/builtins/unified_search/jobs.py index ff634be..9f370e0 100644 --- a/app/plugins/builtins/unified_search/jobs.py +++ b/app/plugins/builtins/unified_search/jobs.py @@ -12,6 +12,14 @@ logger = logging.getLogger(__name__) BATCH_SIZE = 100 +# Entity type -> table name mapping (shared by multiple jobs) +_TABLE_MAP: dict[str, str] = { + "contact": "contacts", + "mail": "mails", + "file": "files", + "event": "calendar_entries", +} + def _parse_id(id_str: str) -> uuid.UUID: """Parse a string to UUID.""" @@ -159,23 +167,36 @@ async def index_event(ctx: dict[str, Any], event_id: str) -> None: async def reindex(ctx: dict[str, Any], entity_type: str) -> None: - """Reindex all entities of a given type with pagination.""" + """Reindex all entities of a given type with pagination. + + Clears existing embeddings before re-indexing, tracks progress, + and continues on individual failures. + """ from sqlalchemy import text from app.plugins.builtins.unified_search.embedding import index_entity - table_map = { - "contact": "contacts", - "mail": "mails", - "file": "files", - "event": "calendar_entries", - } - table = table_map.get(entity_type) + table = _TABLE_MAP.get(entity_type) if not table: logger.warning("Unknown entity_type for reindex: %s", entity_type) return factory = get_session_factory() async with factory() as db: + # Clear existing embeddings before re-indexing + try: + await db.execute( + text(f"UPDATE {table} SET embedding = NULL WHERE deleted_at IS NULL"), + ) + await db.commit() + except Exception: + logger.exception("Failed to clear embeddings for %s", entity_type) + try: + await db.rollback() + except Exception: + pass + + total_indexed = 0 + total_failed = 0 offset = 0 while True: result = await db.execute( @@ -196,15 +217,161 @@ async def reindex(ctx: dict[str, Any], entity_type: str) -> None: row["tenant_id"], db, ) + total_indexed += 1 except Exception: logger.exception("Reindex failed for %s/%s", entity_type, row["id"]) + total_failed += 1 try: await db.rollback() except Exception: pass + # Progress tracking every BATCH_SIZE entities + logger.info( + "Reindex progress for %s: %d indexed, %d failed (offset=%d)", + entity_type, total_indexed, total_failed, offset, + ) offset += BATCH_SIZE - logger.info("Reindex complete for %s", entity_type) + logger.info( + "Reindex complete for %s: %d indexed, %d failed", + entity_type, total_indexed, total_failed, + ) + + +async def reindex_all(ctx: dict[str, Any]) -> None: + """Reindex all entity types in sequence, including file chunks.""" + entity_types = ["contact", "mail", "file", "event"] + for etype in entity_types: + try: + await reindex(ctx, etype) + except Exception: + logger.exception("reindex_all: reindex failed for %s", etype) + # For files, also re-index chunks + if etype == "file": + try: + from sqlalchemy import text + factory = get_session_factory() + async with factory() as db: + result = await db.execute( + text("SELECT id FROM files WHERE deleted_at IS NULL"), + ) + rows = result.mappings().all() + for row in rows: + try: + await index_file_chunks(ctx, str(row["id"])) + except Exception: + logger.exception("reindex_all: chunk re-index failed for file %s", row["id"]) + except Exception: + logger.exception("reindex_all: chunk re-index batch failed") + logger.info("reindex_all complete") + + +async def delete_entity_index(ctx: dict[str, Any], entity_type: str, entity_id: str) -> None: + """Delete an entity's embedding (set embedding=NULL). + + Logs the action to SearchIndexLog on success. + """ + from sqlalchemy import text + from app.plugins.builtins.unified_search.models import SearchIndexLog + + table = _TABLE_MAP.get(entity_type) + if not table: + logger.warning("delete_entity_index: unknown entity_type=%s", entity_type) + return + + factory = get_session_factory() + async with factory() as db: + try: + eid = _parse_id(entity_id) + # Get tenant_id from the entity + result = await db.execute( + text(f"SELECT tenant_id FROM {table} WHERE id = :eid"), + {"eid": eid}, + ) + row = result.mappings().first() + tenant_id = row["tenant_id"] if row else None + + await db.execute( + text( + f"UPDATE {table} SET embedding = NULL, indexed_at = NULL " + f"WHERE id = :eid" + ), + {"eid": eid}, + ) + await db.commit() + + # Log to SearchIndexLog + if tenant_id: + log_entry = SearchIndexLog( + tenant_id=tenant_id, + entity_type=entity_type, + entity_id=eid, + action="delete", + status="success", + ) + db.add(log_entry) + await db.commit() + + logger.info("Deleted index for %s/%s", entity_type, entity_id) + except Exception: + logger.exception("Failed to delete index for %s/%s", entity_type, entity_id) + try: + await db.rollback() + except Exception: + pass + + +async def delete_file_chunks(ctx: dict[str, Any], file_id: str) -> None: + """Delete all document_chunks for a file.""" + from sqlalchemy import text + + factory = get_session_factory() + async with factory() as db: + try: + eid = _parse_id(file_id) + await db.execute( + text("DELETE FROM document_chunks WHERE file_id = :fid"), + {"fid": eid}, + ) + await db.commit() + logger.info("Deleted chunks for file %s", file_id) + except Exception: + logger.exception("Failed to delete chunks for file %s", file_id) + try: + await db.rollback() + except Exception: + pass + + +async def retry_failed_index(ctx: dict[str, Any], entity_type: str, entity_id: str) -> None: + """Retry a failed index operation.""" + from sqlalchemy import text + from app.plugins.builtins.unified_search.embedding import index_entity + + table = _TABLE_MAP.get(entity_type) + if not table: + logger.warning("retry_failed_index: unknown entity_type=%s", entity_type) + return + + factory = get_session_factory() + async with factory() as db: + try: + eid = _parse_id(entity_id) + result = await db.execute( + text(f"SELECT tenant_id FROM {table} WHERE id = :eid"), + {"eid": eid}, + ) + row = result.mappings().first() + if not row: + logger.warning("retry_failed_index: entity not found %s/%s", entity_type, entity_id) + return + await index_entity(entity_type, eid, row["tenant_id"], db) + except Exception: + logger.exception("retry_failed_index failed for %s/%s", entity_type, entity_id) + try: + await db.rollback() + except Exception: + pass async def embedding_batch(ctx: dict[str, Any]) -> None: @@ -212,16 +379,9 @@ async def embedding_batch(ctx: dict[str, Any]) -> None: from sqlalchemy import text from app.plugins.builtins.unified_search.embedding import index_entity - table_map = { - "contact": "contacts", - "mail": "mails", - "file": "files", - "event": "calendar_entries", - } - factory = get_session_factory() async with factory() as db: - for etype, table in table_map.items(): + for etype, table in _TABLE_MAP.items(): try: result = await db.execute( text( @@ -251,7 +411,101 @@ async def embedding_batch(ctx: dict[str, Any]) -> None: logger.info("Embedding batch job complete") -# Register all job functions with the job registry +async def index_file_chunks(ctx: dict[str, Any], file_id: str) -> None: + """Extract text from a file, chunk it, generate embeddings, and store in document_chunks.""" + from sqlalchemy import text + from app.plugins.builtins.unified_search.text_extraction import extract_text_from_file + from app.plugins.builtins.unified_search.chunking import chunk_text + from app.plugins.builtins.unified_search.embedding import generate_embedding + + factory = get_session_factory() + async with factory() as db: + try: + eid = _parse_id(file_id) + result = await db.execute( + text("SELECT tenant_id, storage_path, mime_type, name FROM files WHERE id = :fid"), + {"fid": eid}, + ) + row = result.mappings().first() + if not row: + logger.warning("File not found for chunking: %s", file_id) + return + + tenant_id = row["tenant_id"] + storage_path = row["storage_path"] + mime_type = row["mime_type"] + name = row.get("name", "") or "" + + # Extract text from file + content_text = await extract_text_from_file(storage_path, mime_type) + if not content_text.strip(): + logger.debug("No text extracted from file %s, skipping chunking", file_id) + return + + # Store content_text on the file record + await db.execute( + text("UPDATE files SET content_text = :ct WHERE id = :fid"), + {"ct": content_text, "fid": eid}, + ) + await db.commit() + + # Chunk the text (include filename for context) + full_text = f"{name}\n{content_text}" + chunks = chunk_text(full_text, chunk_size=1000, overlap=200) + if not chunks: + logger.debug("No chunks generated for file %s", file_id) + return + + # Delete existing chunks for this file (idempotent re-index) + await db.execute( + text("DELETE FROM document_chunks WHERE file_id = :fid"), + {"fid": eid}, + ) + + # Generate embeddings and insert chunks in batches + for chunk in chunks: + try: + embedding = await generate_embedding( + chunk["chunk_text"], db=db, tenant_id=tenant_id + ) + if embedding: + await db.execute( + text( + "INSERT INTO document_chunks " + "(tenant_id, file_id, chunk_index, chunk_text, chunk_hash, embedding) " + "VALUES (:tid, :fid, :idx, :ctext, :chash, cast(:emb AS vector))" + ), + { + "tid": tenant_id, + "fid": eid, + "idx": chunk["chunk_index"], + "ctext": chunk["chunk_text"], + "chash": chunk["chunk_hash"], + "emb": str(embedding), + }, + ) + else: + logger.warning("Empty embedding for chunk %d of file %s", chunk["chunk_index"], file_id) + except Exception: + logger.exception("Failed to embed chunk %d for file %s", chunk["chunk_index"], file_id) + + await db.commit() + logger.info("Indexed %d chunks for file %s", len(chunks), file_id) + + except Exception: + logger.exception("Failed to index file chunks for %s", file_id) + try: + await db.rollback() + except Exception: + pass + + +async def reindex_chunks(ctx: dict[str, Any], file_id: str) -> None: + """Re-chunk and re-embed a file — delegates to index_file_chunks.""" + await index_file_chunks(ctx, file_id) + + +# ── Register all job functions with the job registry ────────────────────────── from app.core.job_registry import register_job register_job("index_mails", index_mails) @@ -259,4 +513,10 @@ register_job("index_file", index_file) register_job("index_contact", index_contact) register_job("index_event", index_event) register_job("reindex", reindex) +register_job("reindex_all", reindex_all) register_job("embedding_batch", embedding_batch) +register_job("delete_entity_index", delete_entity_index) +register_job("delete_file_chunks", delete_file_chunks) +register_job("retry_failed_index", retry_failed_index) +register_job("index_file_chunks", index_file_chunks) +register_job("reindex_chunks", reindex_chunks) diff --git a/app/plugins/builtins/unified_search/lifecycle.py b/app/plugins/builtins/unified_search/lifecycle.py new file mode 100644 index 0000000..0f5763e --- /dev/null +++ b/app/plugins/builtins/unified_search/lifecycle.py @@ -0,0 +1,138 @@ +"""Derived-data lifecycle management for the Unified Search index. + +Provides functions to remove, rebuild, and purge search index data +(embeddings, FTS vectors, document chunks) when entities are deleted, +restored, or corrected. +""" + +from __future__ import annotations + +import logging +import uuid +from typing import TYPE_CHECKING + +from sqlalchemy import text + +if TYPE_CHECKING: + from sqlalchemy.ext.asyncio import AsyncSession + +logger = logging.getLogger(__name__) + +# Entity type -> (table_name, tsv_column, embedding_column) +_ENTITY_MAP: dict[str, tuple[str, str, str]] = { + "contact": ("contacts", "search_tsv", "embedding"), + "mail": ("mails", "body_tsv", "embedding"), + "file": ("files", "content_tsv", "embedding"), + "event": ("calendar_entries", "search_tsv", "embedding"), +} + + +def _resolve_entity(entity_type: str) -> tuple[str, str, str] | None: + """Return (table, tsv_column, embedding_column) for an entity type.""" + return _ENTITY_MAP.get(entity_type) + + +async def remove_from_index( + db: "AsyncSession", + entity_type: str, + entity_id: uuid.UUID, + tenant_id: uuid.UUID, +) -> None: + """Remove an entity from the search index (embedding=NULL, TSV=NULL). + + Does **not** delete document_chunks — use :func:`remove_chunks` for that. + """ + mapping = _resolve_entity(entity_type) + if mapping is None: + logger.warning("remove_from_index: unknown entity_type=%s", entity_type) + return + + table, tsv_col, emb_col = mapping + await db.execute( + text( + f"UPDATE {table} " + f"SET {emb_col} = NULL, {tsv_col} = NULL, indexed_at = NULL " + f"WHERE id = :eid AND tenant_id = :tid" + ), + {"eid": entity_id, "tid": tenant_id}, + ) + await db.commit() + logger.info("Removed %s/%s from search index", entity_type, entity_id) + + +async def remove_chunks( + db: "AsyncSession", + file_id: uuid.UUID, + tenant_id: uuid.UUID, +) -> None: + """Remove all document_chunks for a file.""" + await db.execute( + text( + "DELETE FROM document_chunks " + "WHERE file_id = :fid AND tenant_id = :tid" + ), + {"fid": file_id, "tid": tenant_id}, + ) + await db.commit() + logger.info("Removed chunks for file %s", file_id) + + +async def rebuild_index( + db: "AsyncSession", + entity_type: str, + entity_id: uuid.UUID, + tenant_id: uuid.UUID, +) -> bool: + """Rebuild the search index for a single entity from the authoritative source. + + Re-generates the embedding and refreshes the TSV vector. + Returns True on success, False on failure. + """ + from app.plugins.builtins.unified_search.embedding import index_entity + + # Clear existing index data first + await remove_from_index(db, entity_type, entity_id, tenant_id) + + # Re-generate embedding (index_entity also updates indexed_at) + try: + success = await index_entity(entity_type, entity_id, tenant_id, db) + if success and entity_type == "file": + # Re-index chunks for files + from app.plugins.builtins.unified_search.jobs import index_file_chunks + await index_file_chunks({"_lifecycle": True}, str(entity_id)) + return success + except Exception: + logger.exception("rebuild_index failed for %s/%s", entity_type, entity_id) + return False + + +async def handle_entity_delete( + db: "AsyncSession", + entity_type: str, + entity_id: uuid.UUID, + tenant_id: uuid.UUID, +) -> None: + """Handle entity deletion: remove from index + remove chunks if applicable.""" + await remove_from_index(db, entity_type, entity_id, tenant_id) + if entity_type == "file": + await remove_chunks(db, entity_id, tenant_id) + + +async def handle_entity_restore( + db: "AsyncSession", + entity_type: str, + entity_id: uuid.UUID, + tenant_id: uuid.UUID, +) -> None: + """Handle entity restore from trash: rebuild search index.""" + await rebuild_index(db, entity_type, entity_id, tenant_id) + + +async def handle_entity_correction( + db: "AsyncSession", + entity_type: str, + entity_id: uuid.UUID, + tenant_id: uuid.UUID, +) -> None: + """Handle entity data correction: rebuild search index.""" + await rebuild_index(db, entity_type, entity_id, tenant_id) diff --git a/app/plugins/builtins/unified_search/migrations/0004_document_chunks.sql b/app/plugins/builtins/unified_search/migrations/0004_document_chunks.sql new file mode 100644 index 0000000..dcb0e25 --- /dev/null +++ b/app/plugins/builtins/unified_search/migrations/0004_document_chunks.sql @@ -0,0 +1,38 @@ +-- Unified Search: Document chunks table for RAG indexing +-- Creates document_chunks table with HNSW index on embedding, +-- and adds content_text/content_tsv columns to files if missing. + +CREATE EXTENSION IF NOT EXISTS vector; + +-- ─── Document Chunks Table ─── +CREATE TABLE IF NOT EXISTS document_chunks ( + id UUID PRIMARY KEY DEFAULT gen_random_uuid(), + tenant_id UUID NOT NULL, + file_id UUID NOT NULL REFERENCES files(id) ON DELETE CASCADE, + chunk_index INTEGER NOT NULL, + chunk_text TEXT NOT NULL, + chunk_hash VARCHAR(64) NOT NULL, + embedding vector(768), + deleted_at TIMESTAMPTZ, + created_at TIMESTAMPTZ NOT NULL DEFAULT now(), + updated_at TIMESTAMPTZ NOT NULL DEFAULT now() +); + +CREATE INDEX IF NOT EXISTS ix_document_chunks_tenant ON document_chunks(tenant_id); +CREATE INDEX IF NOT EXISTS ix_document_chunks_file ON document_chunks(file_id); +CREATE INDEX IF NOT EXISTS ix_document_chunks_tenant_file ON document_chunks(tenant_id, file_id); + +-- HNSW index for fast cosine similarity search on chunk embeddings +CREATE INDEX IF NOT EXISTS ix_document_chunks_embedding + ON document_chunks USING hnsw(embedding vector_cosine_ops); + +-- ─── Files: content_text and content_tsv (idempotent) ─── +DO $$ BEGIN + IF EXISTS (SELECT 1 FROM information_schema.tables WHERE table_name = 'files') THEN + ALTER TABLE files ADD COLUMN IF NOT EXISTS content_text text; + ALTER TABLE files ADD COLUMN IF NOT EXISTS content_tsv tsvector; + + -- GIN index for full-text search on file content + CREATE INDEX IF NOT EXISTS ix_files_content_tsv ON files USING gin(content_tsv); + END IF; +END $$; diff --git a/app/plugins/builtins/unified_search/migrations/0005_add_indexed_at.sql b/app/plugins/builtins/unified_search/migrations/0005_add_indexed_at.sql new file mode 100644 index 0000000..d3cd503 --- /dev/null +++ b/app/plugins/builtins/unified_search/migrations/0005_add_indexed_at.sql @@ -0,0 +1,7 @@ +-- Unified Search: Add indexed_at column to entity tables for dedup tracking +-- Allows index_entity() to skip re-indexing when content hasn't changed. + +ALTER TABLE contacts ADD COLUMN IF NOT EXISTS indexed_at TIMESTAMPTZ; +ALTER TABLE mails ADD COLUMN IF NOT EXISTS indexed_at TIMESTAMPTZ; +ALTER TABLE files ADD COLUMN IF NOT EXISTS indexed_at TIMESTAMPTZ; +ALTER TABLE calendar_entries ADD COLUMN IF NOT EXISTS indexed_at TIMESTAMPTZ; diff --git a/app/plugins/builtins/unified_search/models.py b/app/plugins/builtins/unified_search/models.py index 960ad88..047d249 100644 --- a/app/plugins/builtins/unified_search/models.py +++ b/app/plugins/builtins/unified_search/models.py @@ -5,7 +5,7 @@ from __future__ import annotations import uuid from datetime import datetime -from sqlalchemy import DateTime, ForeignKey, Index, String, Text +from sqlalchemy import DateTime, ForeignKey, Index, Integer, String, Text from sqlalchemy.dialects.postgresql import JSONB, UUID as PGUUID from sqlalchemy.orm import Mapped, mapped_column @@ -48,3 +48,29 @@ class SearchIndexLog(Base, TenantMixin): action: Mapped[str] = mapped_column(String(20), nullable=False) status: Mapped[str] = mapped_column(String(20), nullable=False, default="pending") error_message: Mapped[str | None] = mapped_column(Text, nullable=True) + +from pgvector.sqlalchemy import Vector + + +class DocumentChunk(Base, TenantMixin): + """Chunk of a DMS file's extracted text, with its own embedding for RAG.""" + + __tablename__ = "document_chunks" + __table_args__ = ( + Index("ix_document_chunks_tenant", "tenant_id"), + Index("ix_document_chunks_file", "file_id"), + Index("ix_document_chunks_tenant_file", "tenant_id", "file_id"), + ) + + id: Mapped[uuid.UUID] = mapped_column( + PGUUID(as_uuid=True), primary_key=True, default=uuid.uuid4 + ) + file_id: Mapped[uuid.UUID] = mapped_column( + PGUUID(as_uuid=True), ForeignKey("files.id", ondelete="CASCADE"), nullable=False + ) + chunk_index: Mapped[int] = mapped_column(Integer, nullable=False) + chunk_text: Mapped[str] = mapped_column(Text, nullable=False) + chunk_hash: Mapped[str] = mapped_column(String(64), nullable=False) + embedding: Mapped[list[float] | None] = mapped_column( + Vector(768), nullable=True + ) diff --git a/app/plugins/builtins/unified_search/plugin.py b/app/plugins/builtins/unified_search/plugin.py index 731ef3e..00bafd7 100644 --- a/app/plugins/builtins/unified_search/plugin.py +++ b/app/plugins/builtins/unified_search/plugin.py @@ -3,6 +3,7 @@ from __future__ import annotations import logging +import uuid from typing import Any from app.plugins.base import BasePlugin @@ -35,9 +36,17 @@ class UnifiedSearchPlugin(BasePlugin): "file.uploaded", "contact.created", "contact.updated", + "contact.deleted", + "file.deleted", + "mail.deleted", + "event.deleted", + "calendar.entry.updated", "calendar.entry.created", + "entity.deleted", + "entity.restored", + "entity.corrected", ], - migrations=["0001_initial.sql", "0002_embeddings.sql", "0003_add_deleted_at.sql"], + migrations=["0001_initial.sql", "0002_embeddings.sql", "0003_add_deleted_at.sql", "0004_document_chunks.sql", "0005_add_indexed_at.sql"], permissions=["search:read", "search:admin"], is_core=True, page_routes=[ @@ -49,7 +58,7 @@ class UnifiedSearchPlugin(BasePlugin): ) async def on_activate(self, db, service_container, event_bus) -> None: - """Register search providers on activation.""" + """Register search providers and AI tool on activation.""" await super().on_activate(db, service_container, event_bus) try: from app.plugins.builtins.unified_search.provider_registry import ( @@ -60,6 +69,15 @@ class UnifiedSearchPlugin(BasePlugin): except Exception: logger.exception("Failed to auto-register search providers") + # Register the unified_search AI tool + try: + from app.plugins.builtins.ai_assistant.tool_registry import get_tool_registry + from app.plugins.builtins.unified_search.ai_tool import register_unified_search_tool + register_unified_search_tool(get_tool_registry()) + logger.info("Unified Search AI tool registered") + except Exception: + logger.exception("Failed to register unified_search AI tool") + async def on_deactivate(self, db, service_container, event_bus) -> None: """Clear provider registry on deactivation.""" # Contract abmelden @@ -111,6 +129,95 @@ class UnifiedSearchPlugin(BasePlugin): if entry_id: await enqueue_job("index_event", entry_id) + async def on_calendar_entry_updated(self, payload: dict[str, Any]) -> None: + """Enqueue re-index job for updated calendar entry.""" + from app.core.jobs import enqueue_job + + entry_id = payload.get("entry_id") + if entry_id: + await enqueue_job("index_event", entry_id) + + async def on_contact_deleted(self, payload: dict[str, Any]) -> None: + """Enqueue job to remove embedding for deleted contact.""" + from app.core.jobs import enqueue_job + + contact_id = payload.get("contact_id") + if contact_id: + await enqueue_job("delete_entity_index", "contact", contact_id) + + async def on_file_deleted(self, payload: dict[str, Any]) -> None: + """Enqueue job to remove file embedding + delete document_chunks.""" + from app.core.jobs import enqueue_job + + file_id = payload.get("file_id") + if file_id: + await enqueue_job("delete_entity_index", "file", file_id) + await enqueue_job("delete_file_chunks", file_id) + + async def on_mail_deleted(self, payload: dict[str, Any]) -> None: + """Enqueue job to remove mail embedding.""" + from app.core.jobs import enqueue_job + + mail_id = payload.get("mail_id") + if mail_id: + await enqueue_job("delete_entity_index", "mail", mail_id) + + async def on_event_deleted(self, payload: dict[str, Any]) -> None: + """Enqueue job to remove event embedding.""" + from app.core.jobs import enqueue_job + + event_id = payload.get("event_id") + if event_id: + await enqueue_job("delete_entity_index", "event", event_id) + + async def on_entity_deleted(self, payload: dict[str, Any]) -> None: + """Handle entity.deleted lifecycle event — remove from search index.""" + from app.plugins.builtins.unified_search.lifecycle import handle_entity_delete + from app.core.db import get_session_factory + + entity_type = payload.get("entity_type") + entity_id = payload.get("entity_id") + tenant_id = payload.get("tenant_id") + if not all([entity_type, entity_id, tenant_id]): + return + factory = get_session_factory() + async with factory() as db: + await handle_entity_delete( + db, entity_type, uuid.UUID(str(entity_id)), uuid.UUID(str(tenant_id)) + ) + + async def on_entity_restored(self, payload: dict[str, Any]) -> None: + """Handle entity.restored lifecycle event — rebuild search index.""" + from app.plugins.builtins.unified_search.lifecycle import handle_entity_restore + from app.core.db import get_session_factory + + entity_type = payload.get("entity_type") + entity_id = payload.get("entity_id") + tenant_id = payload.get("tenant_id") + if not all([entity_type, entity_id, tenant_id]): + return + factory = get_session_factory() + async with factory() as db: + await handle_entity_restore( + db, entity_type, uuid.UUID(str(entity_id)), uuid.UUID(str(tenant_id)) + ) + + async def on_entity_corrected(self, payload: dict[str, Any]) -> None: + """Handle entity.corrected lifecycle event — rebuild search index.""" + from app.plugins.builtins.unified_search.lifecycle import handle_entity_correction + from app.core.db import get_session_factory + + entity_type = payload.get("entity_type") + entity_id = payload.get("entity_id") + tenant_id = payload.get("tenant_id") + if not all([entity_type, entity_id, tenant_id]): + return + factory = get_session_factory() + async with factory() as db: + await handle_entity_correction( + db, entity_type, uuid.UUID(str(entity_id)), uuid.UUID(str(tenant_id)) + ) + def get_notification_types(self) -> list[dict[str, Any]]: return [ { diff --git a/app/plugins/builtins/unified_search/provider_registry.py b/app/plugins/builtins/unified_search/provider_registry.py index 9ee9ec8..b1edae6 100644 --- a/app/plugins/builtins/unified_search/provider_registry.py +++ b/app/plugins/builtins/unified_search/provider_registry.py @@ -19,6 +19,10 @@ class SearchProvider(Protocol): """Protocol for entity-specific search providers.""" entity_type: str + supports_fts: bool + supports_vector: bool + supports_rag: bool + supports_graph: bool async def search_fts( self, @@ -81,6 +85,23 @@ class SearchProviderRegistry: """Get all registered entity types.""" return list(self._providers.keys()) + def get_providers_by_capability(self, capability: str) -> list[SearchProvider]: + """Get all providers that support a given capability (fts/vector/rag/graph).""" + flag_attr = f"supports_{capability}" + return [p for p in self._providers.values() if getattr(p, flag_attr, False)] + + def get_capabilities(self, entity_type: str) -> dict[str, bool]: + """Get capability flags for a specific entity type.""" + provider = self.get(entity_type) + if provider is None: + return {"fts": False, "vector": False, "rag": False, "graph": False} + return { + "fts": getattr(provider, "supports_fts", False), + "vector": getattr(provider, "supports_vector", False), + "rag": getattr(provider, "supports_rag", False), + "graph": getattr(provider, "supports_graph", False), + } + def clear(self) -> None: """Clear all registered providers.""" self._providers.clear() @@ -130,6 +151,15 @@ async def auto_register_providers(db: AsyncSession) -> None: from app.plugins.builtins.unified_search.providers.user_provider import ( UserSearchProvider, ) + from app.plugins.builtins.unified_search.providers.agent_memory_provider import ( + AgentMemorySearchProvider, + ) + from app.plugins.builtins.unified_search.providers.ai_chat_provider import ( + AIChatSearchProvider, + ) + from app.plugins.builtins.unified_search.providers.workflow_provider import ( + WorkflowSearchProvider, + ) from app.plugins.builtins.graph_rag.contracts import GraphRagContract GraphRAGSearchProvider = GraphRagContract.GraphRAGSearchProvider @@ -148,6 +178,9 @@ async def auto_register_providers(db: AsyncSession) -> None: TagSearchProvider, ConversationSearchProvider, UserSearchProvider, + AgentMemorySearchProvider, + AIChatSearchProvider, + WorkflowSearchProvider, GraphRAGSearchProvider, ]: try: diff --git a/app/plugins/builtins/unified_search/providers/agent_memory_provider.py b/app/plugins/builtins/unified_search/providers/agent_memory_provider.py new file mode 100644 index 0000000..d3d281e --- /dev/null +++ b/app/plugins/builtins/unified_search/providers/agent_memory_provider.py @@ -0,0 +1,163 @@ +"""Agent memory search provider — FTS + vector search on agent_memories table.""" + +from __future__ import annotations + +import logging +import uuid +from typing import Any + +from sqlalchemy import text +from sqlalchemy.ext.asyncio import AsyncSession + +from app.plugins.builtins.unified_search.base_provider import BaseSearchProvider + +logger = logging.getLogger(__name__) + + +class AgentMemorySearchProvider(BaseSearchProvider): + """Search provider for AgentMemory entities.""" + + entity_type = "agent_memory" + supports_fts = True + supports_vector = True + + async def _search_fts_filtered( + self, + db: AsyncSession, + tsquery: str, + tenant_id: uuid.UUID, + limit: int, + visible_ids: set[uuid.UUID] | None, + ) -> list[dict[str, Any]]: + """Full-text search on agent_memories.content, filtered by visible_ids.""" + if visible_ids is not None: + sql = text( + """ + SELECT am.*, + ts_rank(to_tsvector('pg_catalog.german', am.content), + to_tsquery('pg_catalog.german', :q)) AS rank + FROM agent_memories am + WHERE am.tenant_id = :tid + AND to_tsvector('pg_catalog.german', am.content) @@ to_tsquery('pg_catalog.german', :q) + AND am.id = ANY(:visible_ids) + ORDER BY rank DESC + LIMIT :lim + """ + ) + result = await db.execute( + sql, + { + "q": tsquery, + "tid": tenant_id, + "lim": limit, + "visible_ids": list(visible_ids), + }, + ) + else: + sql = text( + """ + SELECT am.*, + ts_rank(to_tsvector('pg_catalog.german', am.content), + to_tsquery('pg_catalog.german', :q)) AS rank + FROM agent_memories am + WHERE am.tenant_id = :tid + AND to_tsvector('pg_catalog.german', am.content) @@ to_tsquery('pg_catalog.german', :q) + ORDER BY rank DESC + LIMIT :lim + """ + ) + result = await db.execute( + sql, + {"q": tsquery, "tid": tenant_id, "lim": limit}, + ) + rows = result.mappings().all() + return [dict(r) for r in rows] + + async def _search_vector_filtered( + self, + db: AsyncSession, + embedding: list[float], + tenant_id: uuid.UUID, + limit: int, + visible_ids: set[uuid.UUID] | None, + ) -> list[dict[str, Any]]: + """Semantic vector search on agent_memories.embedding, filtered by visible_ids.""" + if visible_ids is not None: + sql = text( + """ + SELECT am.*, + 1 - (am.embedding <=> cast(:emb AS vector)) AS score + FROM agent_memories am + WHERE am.tenant_id = :tid + AND am.embedding IS NOT NULL + AND am.id = ANY(:visible_ids) + ORDER BY am.embedding <=> cast(:emb AS vector) + LIMIT :lim + """ + ) + result = await db.execute( + sql, + { + "emb": str(embedding), + "tid": tenant_id, + "lim": limit, + "visible_ids": list(visible_ids), + }, + ) + else: + sql = text( + """ + SELECT am.*, + 1 - (am.embedding <=> cast(:emb AS vector)) AS score + FROM agent_memories am + WHERE am.tenant_id = :tid + AND am.embedding IS NOT NULL + ORDER BY am.embedding <=> cast(:emb AS vector) + LIMIT :lim + """ + ) + result = await db.execute( + sql, + {"emb": str(embedding), "tid": tenant_id, "lim": limit}, + ) + rows = result.mappings().all() + return [dict(r) for r in rows] + + async def get_embedding_text( + self, db: AsyncSession, entity_id: uuid.UUID, tenant_id: uuid.UUID + ) -> str: + """Get text for embedding generation.""" + sql = text( + """ + SELECT content, memory_type + FROM agent_memories + WHERE id = :eid AND tenant_id = :tid + """ + ) + result = await db.execute(sql, {"eid": entity_id, "tid": tenant_id}) + row = result.mappings().first() + if not row: + return "" + parts = [row.get("memory_type", ""), row.get("content", "")] + return " ".join(str(p) for p in parts if p) + + def to_search_result(self, entity: object) -> dict[str, Any]: + """Convert agent memory to search result dict.""" + if isinstance(entity, dict): + entity_id = str(entity.get("id", "")) + content = entity.get("content", "") + memory_type = entity.get("memory_type", "") + score = entity.get("score", entity.get("rank", 0.0)) + else: + entity_id = str(getattr(entity, "id", "")) + content = getattr(entity, "content", "") + memory_type = getattr(entity, "memory_type", "") + score = getattr(entity, "score", getattr(entity, "rank", 0.0)) + return { + "entity_type": self.entity_type, + "entity_id": entity_id, + "title": content[:100] if content else "", + "snippet": content, + "score": float(score) if score else 0.0, + "data": {"memory_type": memory_type}, + } diff --git a/app/plugins/builtins/unified_search/providers/ai_chat_provider.py b/app/plugins/builtins/unified_search/providers/ai_chat_provider.py new file mode 100644 index 0000000..9544231 --- /dev/null +++ b/app/plugins/builtins/unified_search/providers/ai_chat_provider.py @@ -0,0 +1,137 @@ +"""AI chat search provider — FTS search on ai_chat_messages table.""" + +from __future__ import annotations + +import logging +import uuid +from typing import Any + +from sqlalchemy import text +from sqlalchemy.ext.asyncio import AsyncSession + +from app.plugins.builtins.unified_search.base_provider import BaseSearchProvider + +logger = logging.getLogger(__name__) + + +class AIChatSearchProvider(BaseSearchProvider): + """Search provider for AI chat messages.""" + + entity_type = "ai_chat" + supports_fts = True + supports_vector = False + + async def _search_fts_filtered( + self, + db: AsyncSession, + tsquery: str, + tenant_id: uuid.UUID, + limit: int, + visible_ids: set[uuid.UUID] | None, + ) -> list[dict[str, Any]]: + """Full-text search on ai_chat_messages.content, joined with sessions for title.""" + if visible_ids is not None: + sql = text( + """ + SELECT m.id, m.tenant_id, m.session_id, m.role, m.content, + m.model_used, m.tokens, + s.title AS session_title, + ts_rank(to_tsvector('pg_catalog.german', m.content), + to_tsquery('pg_catalog.german', :q)) AS rank + FROM ai_chat_messages m + JOIN ai_chat_sessions s ON s.id = m.session_id + WHERE m.tenant_id = :tid + AND to_tsvector('pg_catalog.german', m.content) @@ to_tsquery('pg_catalog.german', :q) + AND m.id = ANY(:visible_ids) + ORDER BY rank DESC + LIMIT :lim + """ + ) + result = await db.execute( + sql, + { + "q": tsquery, + "tid": tenant_id, + "lim": limit, + "visible_ids": list(visible_ids), + }, + ) + else: + sql = text( + """ + SELECT m.id, m.tenant_id, m.session_id, m.role, m.content, + m.model_used, m.tokens, + s.title AS session_title, + ts_rank(to_tsvector('pg_catalog.german', m.content), + to_tsquery('pg_catalog.german', :q)) AS rank + FROM ai_chat_messages m + JOIN ai_chat_sessions s ON s.id = m.session_id + WHERE m.tenant_id = :tid + AND to_tsvector('pg_catalog.german', m.content) @@ to_tsquery('pg_catalog.german', :q) + ORDER BY rank DESC + LIMIT :lim + """ + ) + result = await db.execute( + sql, + {"q": tsquery, "tid": tenant_id, "lim": limit}, + ) + rows = result.mappings().all() + return [dict(r) for r in rows] + + async def _search_vector_filtered( + self, + db: AsyncSession, + embedding: list[float], + tenant_id: uuid.UUID, + limit: int, + visible_ids: set[uuid.UUID] | None, + ) -> list[dict[str, Any]]: + """No vector support for chat messages — return empty list.""" + return [] + + async def get_embedding_text( + self, db: AsyncSession, entity_id: uuid.UUID, tenant_id: uuid.UUID + ) -> str: + """Get text for embedding generation — returns message content.""" + sql = text( + """ + SELECT content + FROM ai_chat_messages + WHERE id = :eid AND tenant_id = :tid + """ + ) + result = await db.execute(sql, {"eid": entity_id, "tid": tenant_id}) + row = result.mappings().first() + if not row: + return "" + return str(row.get("content", "")) + + def to_search_result(self, entity: object) -> dict[str, Any]: + """Convert chat message to search result dict.""" + if isinstance(entity, dict): + entity_id = str(entity.get("id", "")) + content = entity.get("content", "") + role = entity.get("role", "") + session_title = entity.get("session_title", "") + session_id = str(entity.get("session_id", "")) + score = entity.get("rank", 0.0) + else: + entity_id = str(getattr(entity, "id", "")) + content = getattr(entity, "content", "") + role = getattr(entity, "role", "") + session_title = getattr(entity, "session_title", "") + session_id = str(getattr(entity, "session_id", "")) + score = getattr(entity, "rank", 0.0) + return { + "entity_type": self.entity_type, + "entity_id": entity_id, + "title": session_title or content[:80] if content else "", + "snippet": content, + "score": float(score) if score else 0.0, + "data": { + "role": role, + "session_id": session_id, + "session_title": session_title, + }, + } diff --git a/app/plugins/builtins/unified_search/providers/company_provider.py b/app/plugins/builtins/unified_search/providers/company_provider.py index c7dc907..ce3c2e6 100644 --- a/app/plugins/builtins/unified_search/providers/company_provider.py +++ b/app/plugins/builtins/unified_search/providers/company_provider.py @@ -9,73 +9,128 @@ from typing import Any from sqlalchemy import text from sqlalchemy.ext.asyncio import AsyncSession +from app.plugins.builtins.unified_search.base_provider import BaseSearchProvider + logger = logging.getLogger(__name__) -class CompanySearchProvider: +class CompanySearchProvider(BaseSearchProvider): """Search provider for Company entities (contacts with type='company').""" entity_type = "contact" - async def search_fts( + async def _search_fts_filtered( self, db: AsyncSession, tsquery: str, tenant_id: uuid.UUID, limit: int, + visible_ids: set[uuid.UUID] | None, ) -> list[dict[str, Any]]: - """Full-text search on contacts.search_tsv where type='company'.""" - sql = text( - """ - SELECT c.*, ts_rank(c.search_tsv, to_tsquery('pg_catalog.german', :q)) AS rank - FROM contacts c - WHERE c.tenant_id = :tid - AND c.deleted_at IS NULL - AND c.type = 'company' - AND c.search_tsv @@ to_tsquery('pg_catalog.german', :q) - ORDER BY rank DESC - LIMIT :lim - """ - ) - result = await db.execute( - sql, - {"q": tsquery, "tid": tenant_id, "lim": limit}, - ) + """Full-text search on contacts.search_tsv where type='company', filtered by visible_ids. + + If visible_ids is None, no visibility filter is applied (system admin). + """ + if visible_ids is not None: + sql = text( + """ + SELECT c.*, ts_rank(c.search_tsv, to_tsquery('pg_catalog.german', :q)) AS rank + FROM contacts c + WHERE c.tenant_id = :tid + AND c.deleted_at IS NULL + AND c.type = 'company' + AND c.search_tsv @@ to_tsquery('pg_catalog.german', :q) + AND c.id = ANY(:visible_ids) + ORDER BY rank DESC + LIMIT :lim + """ + ) + result = await db.execute( + sql, + { + "q": tsquery, + "tid": tenant_id, + "lim": limit, + "visible_ids": list(visible_ids), + }, + ) + else: + sql = text( + """ + SELECT c.*, ts_rank(c.search_tsv, to_tsquery('pg_catalog.german', :q)) AS rank + FROM contacts c + WHERE c.tenant_id = :tid + AND c.deleted_at IS NULL + AND c.type = 'company' + AND c.search_tsv @@ to_tsquery('pg_catalog.german', :q) + ORDER BY rank DESC + LIMIT :lim + """ + ) + result = await db.execute( + sql, + {"q": tsquery, "tid": tenant_id, "lim": limit}, + ) rows = result.mappings().all() return [dict(r) for r in rows] - async def search_vector( + async def _search_vector_filtered( self, db: AsyncSession, embedding: list[float], tenant_id: uuid.UUID, limit: int, + visible_ids: set[uuid.UUID] | None, ) -> list[dict[str, Any]]: - """Semantic search on contacts.embedding where type='company'.""" - sql = text( - """ - SELECT c.*, 1 - (c.embedding <=> cast(:emb AS vector)) AS score - FROM contacts c - WHERE c.tenant_id = :tid - AND c.deleted_at IS NULL - AND c.type = 'company' - AND c.embedding IS NOT NULL - ORDER BY c.embedding <=> cast(:emb AS vector) - LIMIT :lim - """ - ) - result = await db.execute( - sql, - {"emb": str(embedding), "tid": tenant_id, "lim": limit}, - ) + """Semantic search on contacts.embedding where type='company', filtered by visible_ids. + + If visible_ids is None, no visibility filter is applied (system admin). + """ + if visible_ids is not None: + sql = text( + """ + SELECT c.*, 1 - (c.embedding <=> cast(:emb AS vector)) AS score + FROM contacts c + WHERE c.tenant_id = :tid + AND c.deleted_at IS NULL + AND c.type = 'company' + AND c.embedding IS NOT NULL + AND c.id = ANY(:visible_ids) + ORDER BY c.embedding <=> cast(:emb AS vector) + LIMIT :lim + """ + ) + result = await db.execute( + sql, + { + "emb": str(embedding), + "tid": tenant_id, + "lim": limit, + "visible_ids": list(visible_ids), + }, + ) + else: + sql = text( + """ + SELECT c.*, 1 - (c.embedding <=> cast(:emb AS vector)) AS score + FROM contacts c + WHERE c.tenant_id = :tid + AND c.deleted_at IS NULL + AND c.type = 'company' + AND c.embedding IS NOT NULL + ORDER BY c.embedding <=> cast(:emb AS vector) + LIMIT :lim + """ + ) + result = await db.execute( + sql, + {"emb": str(embedding), "tid": tenant_id, "lim": limit}, + ) rows = result.mappings().all() return [dict(r) for r in rows] async def get_embedding_text( - self, - db: AsyncSession, - entity_id: uuid.UUID, - tenant_id: uuid.UUID, + self, db: AsyncSession, entity_id: uuid.UUID, tenant_id: uuid.UUID ) -> str: """Get text for embedding generation.""" sql = text( diff --git a/app/plugins/builtins/unified_search/providers/event_provider.py b/app/plugins/builtins/unified_search/providers/event_provider.py index 2d74951..299f53e 100644 --- a/app/plugins/builtins/unified_search/providers/event_provider.py +++ b/app/plugins/builtins/unified_search/providers/event_provider.py @@ -9,71 +9,124 @@ from typing import Any from sqlalchemy import text from sqlalchemy.ext.asyncio import AsyncSession +from app.plugins.builtins.unified_search.base_provider import BaseSearchProvider + logger = logging.getLogger(__name__) -class EventSearchProvider: +class EventSearchProvider(BaseSearchProvider): """Search provider for CalendarEntry entities.""" entity_type = "event" - async def search_fts( + async def _search_fts_filtered( self, db: AsyncSession, tsquery: str, tenant_id: uuid.UUID, limit: int, + visible_ids: set[uuid.UUID] | None, ) -> list[dict[str, Any]]: - """Full-text search on calendar_entries.search_tsv.""" - sql = text( - """ - SELECT e.*, ts_rank(e.search_tsv, to_tsquery('pg_catalog.german', :q)) AS rank - FROM calendar_entries e - WHERE e.tenant_id = :tid - AND e.deleted_at IS NULL - AND e.search_tsv @@ to_tsquery('pg_catalog.german', :q) - ORDER BY rank DESC - LIMIT :lim - """ - ) - result = await db.execute( - sql, - {"q": tsquery, "tid": tenant_id, "lim": limit}, - ) + """Full-text search on calendar_entries.search_tsv, filtered by visible_ids. + + If visible_ids is None, no visibility filter is applied (system admin). + """ + if visible_ids is not None: + sql = text( + """ + SELECT e.*, ts_rank(e.search_tsv, to_tsquery('pg_catalog.german', :q)) AS rank + FROM calendar_entries e + WHERE e.tenant_id = :tid + AND e.deleted_at IS NULL + AND e.search_tsv @@ to_tsquery('pg_catalog.german', :q) + AND e.id = ANY(:visible_ids) + ORDER BY rank DESC + LIMIT :lim + """ + ) + result = await db.execute( + sql, + { + "q": tsquery, + "tid": tenant_id, + "lim": limit, + "visible_ids": list(visible_ids), + }, + ) + else: + sql = text( + """ + SELECT e.*, ts_rank(e.search_tsv, to_tsquery('pg_catalog.german', :q)) AS rank + FROM calendar_entries e + WHERE e.tenant_id = :tid + AND e.deleted_at IS NULL + AND e.search_tsv @@ to_tsquery('pg_catalog.german', :q) + ORDER BY rank DESC + LIMIT :lim + """ + ) + result = await db.execute( + sql, + {"q": tsquery, "tid": tenant_id, "lim": limit}, + ) rows = result.mappings().all() return [dict(r) for r in rows] - async def search_vector( + async def _search_vector_filtered( self, db: AsyncSession, embedding: list[float], tenant_id: uuid.UUID, limit: int, + visible_ids: set[uuid.UUID] | None, ) -> list[dict[str, Any]]: - """Semantic search on calendar_entries.embedding.""" - sql = text( - """ - SELECT e.*, 1 - (e.embedding <=> cast(:emb AS vector)) AS score - FROM calendar_entries e - WHERE e.tenant_id = :tid - AND e.deleted_at IS NULL - AND e.embedding IS NOT NULL - ORDER BY e.embedding <=> cast(:emb AS vector) - LIMIT :lim - """ - ) - result = await db.execute( - sql, - {"emb": str(embedding), "tid": tenant_id, "lim": limit}, - ) + """Semantic search on calendar_entries.embedding, filtered by visible_ids. + + If visible_ids is None, no visibility filter is applied (system admin). + """ + if visible_ids is not None: + sql = text( + """ + SELECT e.*, 1 - (e.embedding <=> cast(:emb AS vector)) AS score + FROM calendar_entries e + WHERE e.tenant_id = :tid + AND e.deleted_at IS NULL + AND e.embedding IS NOT NULL + AND e.id = ANY(:visible_ids) + ORDER BY e.embedding <=> cast(:emb AS vector) + LIMIT :lim + """ + ) + result = await db.execute( + sql, + { + "emb": str(embedding), + "tid": tenant_id, + "lim": limit, + "visible_ids": list(visible_ids), + }, + ) + else: + sql = text( + """ + SELECT e.*, 1 - (e.embedding <=> cast(:emb AS vector)) AS score + FROM calendar_entries e + WHERE e.tenant_id = :tid + AND e.deleted_at IS NULL + AND e.embedding IS NOT NULL + ORDER BY e.embedding <=> cast(:emb AS vector) + LIMIT :lim + """ + ) + result = await db.execute( + sql, + {"emb": str(embedding), "tid": tenant_id, "lim": limit}, + ) rows = result.mappings().all() return [dict(r) for r in rows] async def get_embedding_text( - self, - db: AsyncSession, - entity_id: uuid.UUID, - tenant_id: uuid.UUID, + self, db: AsyncSession, entity_id: uuid.UUID, tenant_id: uuid.UUID ) -> str: """Get text for embedding generation.""" sql = text( diff --git a/app/plugins/builtins/unified_search/providers/file_provider.py b/app/plugins/builtins/unified_search/providers/file_provider.py index e0b00b4..541c29f 100644 --- a/app/plugins/builtins/unified_search/providers/file_provider.py +++ b/app/plugins/builtins/unified_search/providers/file_provider.py @@ -9,71 +9,126 @@ from typing import Any from sqlalchemy import text from sqlalchemy.ext.asyncio import AsyncSession +from app.plugins.builtins.unified_search.base_provider import BaseSearchProvider +from app.config import settings + logger = logging.getLogger(__name__) -class FileSearchProvider: +class FileSearchProvider(BaseSearchProvider): """Search provider for DMS File entities.""" entity_type = "file" + supports_rag: bool = True - async def search_fts( + async def _search_fts_filtered( self, db: AsyncSession, tsquery: str, tenant_id: uuid.UUID, limit: int, + visible_ids: set[uuid.UUID] | None, ) -> list[dict[str, Any]]: - """Full-text search on files.content_tsv.""" - sql = text( - """ - SELECT f.*, ts_rank(f.content_tsv, to_tsquery('pg_catalog.german', :q)) AS rank - FROM files f - WHERE f.tenant_id = :tid - AND f.deleted_at IS NULL - AND f.content_tsv @@ to_tsquery('pg_catalog.german', :q) - ORDER BY rank DESC - LIMIT :lim - """ - ) - result = await db.execute( - sql, - {"q": tsquery, "tid": tenant_id, "lim": limit}, - ) + """Full-text search on files.content_tsv, filtered by visible_ids. + + If visible_ids is None, no visibility filter is applied (system admin). + """ + if visible_ids is not None: + sql = text( + """ + SELECT f.*, ts_rank(f.content_tsv, to_tsquery('pg_catalog.german', :q)) AS rank + FROM files f + WHERE f.tenant_id = :tid + AND f.deleted_at IS NULL + AND f.content_tsv @@ to_tsquery('pg_catalog.german', :q) + AND f.id = ANY(:visible_ids) + ORDER BY rank DESC + LIMIT :lim + """ + ) + result = await db.execute( + sql, + { + "q": tsquery, + "tid": tenant_id, + "lim": limit, + "visible_ids": list(visible_ids), + }, + ) + else: + sql = text( + """ + SELECT f.*, ts_rank(f.content_tsv, to_tsquery('pg_catalog.german', :q)) AS rank + FROM files f + WHERE f.tenant_id = :tid + AND f.deleted_at IS NULL + AND f.content_tsv @@ to_tsquery('pg_catalog.german', :q) + ORDER BY rank DESC + LIMIT :lim + """ + ) + result = await db.execute( + sql, + {"q": tsquery, "tid": tenant_id, "lim": limit}, + ) rows = result.mappings().all() return [dict(r) for r in rows] - async def search_vector( + async def _search_vector_filtered( self, db: AsyncSession, embedding: list[float], tenant_id: uuid.UUID, limit: int, + visible_ids: set[uuid.UUID] | None, ) -> list[dict[str, Any]]: - """Semantic search on files.embedding.""" - sql = text( - """ - SELECT f.*, 1 - (f.embedding <=> cast(:emb AS vector)) AS score - FROM files f - WHERE f.tenant_id = :tid - AND f.deleted_at IS NULL - AND f.embedding IS NOT NULL - ORDER BY f.embedding <=> cast(:emb AS vector) - LIMIT :lim - """ - ) - result = await db.execute( - sql, - {"emb": str(embedding), "tid": tenant_id, "lim": limit}, - ) + """Semantic search on files.embedding, filtered by visible_ids. + + If visible_ids is None, no visibility filter is applied (system admin). + """ + if visible_ids is not None: + sql = text( + """ + SELECT f.*, 1 - (f.embedding <=> cast(:emb AS vector)) AS score + FROM files f + WHERE f.tenant_id = :tid + AND f.deleted_at IS NULL + AND f.embedding IS NOT NULL + AND f.id = ANY(:visible_ids) + ORDER BY f.embedding <=> cast(:emb AS vector) + LIMIT :lim + """ + ) + result = await db.execute( + sql, + { + "emb": str(embedding), + "tid": tenant_id, + "lim": limit, + "visible_ids": list(visible_ids), + }, + ) + else: + sql = text( + """ + SELECT f.*, 1 - (f.embedding <=> cast(:emb AS vector)) AS score + FROM files f + WHERE f.tenant_id = :tid + AND f.deleted_at IS NULL + AND f.embedding IS NOT NULL + ORDER BY f.embedding <=> cast(:emb AS vector) + LIMIT :lim + """ + ) + result = await db.execute( + sql, + {"emb": str(embedding), "tid": tenant_id, "lim": limit}, + ) rows = result.mappings().all() return [dict(r) for r in rows] async def get_embedding_text( - self, - db: AsyncSession, - entity_id: uuid.UUID, - tenant_id: uuid.UUID, + self, db: AsyncSession, entity_id: uuid.UUID, tenant_id: uuid.UUID ) -> str: """Get text for embedding generation.""" sql = text( @@ -91,6 +146,101 @@ class FileSearchProvider: content = row.get("content_text", "") or "" return f"{name} {content[:5000]}" + async def search_rag( + self, + db: AsyncSession, + query_embedding: list[float], + tenant_id: uuid.UUID, + limit: int, + user_id: uuid.UUID | None = None, + is_system_admin: bool = False, + ) -> list[dict[str, Any]]: + """RAG search: find relevant document chunks via vector similarity. + + Queries the document_chunks table using cosine distance on chunk + embeddings, joins to files to exclude soft-deleted files, and applies + permission filtering using the over-fetch strategy (same as search_vector). + """ + await db.execute(text(f"SET LOCAL hnsw.ef_search = {settings.hnsw_ef_search}")) + + if is_system_admin or not user_id: + return await self._search_rag_filtered(db, query_embedding, tenant_id, limit, None) + + visible_ids = await self._get_visible_ids(db, tenant_id, user_id) + if not visible_ids: + return [] + + over_fetch_limit = limit * 3 + results = await self._search_rag_filtered(db, query_embedding, tenant_id, over_fetch_limit, None) + filtered = [r for r in results if r.get("file_id") in visible_ids] + return filtered[:limit] + + async def _search_rag_filtered( + self, + db: AsyncSession, + embedding: list[float], + tenant_id: uuid.UUID, + limit: int, + visible_ids: set[uuid.UUID] | None, + ) -> list[dict[str, Any]]: + """RAG vector search on document_chunks.embedding, filtered by visible_ids.""" + if visible_ids is not None: + sql = text( + """ + SELECT dc.chunk_text, dc.file_id, dc.chunk_index, + 1 - (dc.embedding <=> cast(:emb AS vector)) AS score + FROM document_chunks dc + JOIN files f ON dc.file_id = f.id + WHERE dc.tenant_id = :tid + AND dc.deleted_at IS NULL + AND dc.embedding IS NOT NULL + AND f.deleted_at IS NULL + AND f.id = ANY(:visible_ids) + ORDER BY dc.embedding <=> cast(:emb AS vector) + LIMIT :lim + """ + ) + result = await db.execute( + sql, + { + "emb": str(embedding), + "tid": tenant_id, + "lim": limit, + "visible_ids": list(visible_ids), + }, + ) + else: + sql = text( + """ + SELECT dc.chunk_text, dc.file_id, dc.chunk_index, + 1 - (dc.embedding <=> cast(:emb AS vector)) AS score + FROM document_chunks dc + JOIN files f ON dc.file_id = f.id + WHERE dc.tenant_id = :tid + AND dc.deleted_at IS NULL + AND dc.embedding IS NOT NULL + AND f.deleted_at IS NULL + ORDER BY dc.embedding <=> cast(:emb AS vector) + LIMIT :lim + """ + ) + result = await db.execute( + sql, + {"emb": str(embedding), "tid": tenant_id, "lim": limit}, + ) + rows = result.mappings().all() + return [ + { + "id": str(r["file_id"]), + "entity_type": "file", + "entity_id": str(r["file_id"]), + "snippet": r["chunk_text"][:200], + "score": float(r.get("score", 0.0)), + "data": {"chunk_index": r.get("chunk_index")}, + } + for r in rows + ] + def to_search_result(self, entity: object) -> dict[str, Any]: """Convert file to search result dict.""" if isinstance(entity, dict): diff --git a/app/plugins/builtins/unified_search/providers/mail_provider.py b/app/plugins/builtins/unified_search/providers/mail_provider.py index b9d1779..248ea82 100644 --- a/app/plugins/builtins/unified_search/providers/mail_provider.py +++ b/app/plugins/builtins/unified_search/providers/mail_provider.py @@ -9,71 +9,124 @@ from typing import Any from sqlalchemy import text from sqlalchemy.ext.asyncio import AsyncSession +from app.plugins.builtins.unified_search.base_provider import BaseSearchProvider + logger = logging.getLogger(__name__) -class MailSearchProvider: +class MailSearchProvider(BaseSearchProvider): """Search provider for Mail entities.""" entity_type = "mail" - async def search_fts( + async def _search_fts_filtered( self, db: AsyncSession, tsquery: str, tenant_id: uuid.UUID, limit: int, + visible_ids: set[uuid.UUID] | None, ) -> list[dict[str, Any]]: - """Full-text search on mails.body_tsv.""" - sql = text( - """ - SELECT m.*, ts_rank(m.body_tsv, to_tsquery('pg_catalog.german', :q)) AS rank - FROM mails m - WHERE m.tenant_id = :tid - AND m.deleted_at IS NULL - AND m.body_tsv @@ to_tsquery('pg_catalog.german', :q) - ORDER BY rank DESC - LIMIT :lim - """ - ) - result = await db.execute( - sql, - {"q": tsquery, "tid": tenant_id, "lim": limit}, - ) + """Full-text search on mails.body_tsv, filtered by visible_ids. + + If visible_ids is None, no visibility filter is applied (system admin). + """ + if visible_ids is not None: + sql = text( + """ + SELECT m.*, ts_rank(m.body_tsv, to_tsquery('pg_catalog.german', :q)) AS rank + FROM mails m + WHERE m.tenant_id = :tid + AND m.deleted_at IS NULL + AND m.body_tsv @@ to_tsquery('pg_catalog.german', :q) + AND m.id = ANY(:visible_ids) + ORDER BY rank DESC + LIMIT :lim + """ + ) + result = await db.execute( + sql, + { + "q": tsquery, + "tid": tenant_id, + "lim": limit, + "visible_ids": list(visible_ids), + }, + ) + else: + sql = text( + """ + SELECT m.*, ts_rank(m.body_tsv, to_tsquery('pg_catalog.german', :q)) AS rank + FROM mails m + WHERE m.tenant_id = :tid + AND m.deleted_at IS NULL + AND m.body_tsv @@ to_tsquery('pg_catalog.german', :q) + ORDER BY rank DESC + LIMIT :lim + """ + ) + result = await db.execute( + sql, + {"q": tsquery, "tid": tenant_id, "lim": limit}, + ) rows = result.mappings().all() return [dict(r) for r in rows] - async def search_vector( + async def _search_vector_filtered( self, db: AsyncSession, embedding: list[float], tenant_id: uuid.UUID, limit: int, + visible_ids: set[uuid.UUID] | None, ) -> list[dict[str, Any]]: - """Semantic search on mails.embedding.""" - sql = text( - """ - SELECT m.*, 1 - (m.embedding <=> cast(:emb AS vector)) AS score - FROM mails m - WHERE m.tenant_id = :tid - AND m.deleted_at IS NULL - AND m.embedding IS NOT NULL - ORDER BY m.embedding <=> cast(:emb AS vector) - LIMIT :lim - """ - ) - result = await db.execute( - sql, - {"emb": str(embedding), "tid": tenant_id, "lim": limit}, - ) + """Semantic search on mails.embedding, filtered by visible_ids. + + If visible_ids is None, no visibility filter is applied (system admin). + """ + if visible_ids is not None: + sql = text( + """ + SELECT m.*, 1 - (m.embedding <=> cast(:emb AS vector)) AS score + FROM mails m + WHERE m.tenant_id = :tid + AND m.deleted_at IS NULL + AND m.embedding IS NOT NULL + AND m.id = ANY(:visible_ids) + ORDER BY m.embedding <=> cast(:emb AS vector) + LIMIT :lim + """ + ) + result = await db.execute( + sql, + { + "emb": str(embedding), + "tid": tenant_id, + "lim": limit, + "visible_ids": list(visible_ids), + }, + ) + else: + sql = text( + """ + SELECT m.*, 1 - (m.embedding <=> cast(:emb AS vector)) AS score + FROM mails m + WHERE m.tenant_id = :tid + AND m.deleted_at IS NULL + AND m.embedding IS NOT NULL + ORDER BY m.embedding <=> cast(:emb AS vector) + LIMIT :lim + """ + ) + result = await db.execute( + sql, + {"emb": str(embedding), "tid": tenant_id, "lim": limit}, + ) rows = result.mappings().all() return [dict(r) for r in rows] async def get_embedding_text( - self, - db: AsyncSession, - entity_id: uuid.UUID, - tenant_id: uuid.UUID, + self, db: AsyncSession, entity_id: uuid.UUID, tenant_id: uuid.UUID ) -> str: """Get text for embedding generation.""" sql = text( diff --git a/app/plugins/builtins/unified_search/providers/workflow_provider.py b/app/plugins/builtins/unified_search/providers/workflow_provider.py new file mode 100644 index 0000000..6975704 --- /dev/null +++ b/app/plugins/builtins/unified_search/providers/workflow_provider.py @@ -0,0 +1,143 @@ +"""Workflow search provider — FTS search on workflows and workflow_instances tables.""" + +from __future__ import annotations + +import logging +import uuid +from typing import Any + +from sqlalchemy import text +from sqlalchemy.ext.asyncio import AsyncSession + +from app.plugins.builtins.unified_search.base_provider import BaseSearchProvider + +logger = logging.getLogger(__name__) + + +class WorkflowSearchProvider(BaseSearchProvider): + """Search provider for Workflow entities (definitions and instances).""" + + entity_type = "workflow" + supports_fts = True + supports_vector = False + + async def _search_fts_filtered( + self, + db: AsyncSession, + tsquery: str, + tenant_id: uuid.UUID, + limit: int, + visible_ids: set[uuid.UUID] | None, + ) -> list[dict[str, Any]]: + """Full-text search on workflows name/description and workflow_instances status.""" + if visible_ids is not None: + sql = text( + """ + SELECT w.id, w.tenant_id, w.name, w.description, w.trigger_event, + w.is_active, + ts_rank( + to_tsvector('pg_catalog.german', + coalesce(w.name, '') || ' ' || coalesce(w.description, '')), + to_tsquery('pg_catalog.german', :q) + ) AS rank + FROM workflows w + WHERE w.tenant_id = :tid + AND to_tsvector('pg_catalog.german', + coalesce(w.name, '') || ' ' || coalesce(w.description, '')) + @@ to_tsquery('pg_catalog.german', :q) + AND w.id = ANY(:visible_ids) + ORDER BY rank DESC + LIMIT :lim + """ + ) + result = await db.execute( + sql, + { + "q": tsquery, + "tid": tenant_id, + "lim": limit, + "visible_ids": list(visible_ids), + }, + ) + else: + sql = text( + """ + SELECT w.id, w.tenant_id, w.name, w.description, w.trigger_event, + w.is_active, + ts_rank( + to_tsvector('pg_catalog.german', + coalesce(w.name, '') || ' ' || coalesce(w.description, '')), + to_tsquery('pg_catalog.german', :q) + ) AS rank + FROM workflows w + WHERE w.tenant_id = :tid + AND to_tsvector('pg_catalog.german', + coalesce(w.name, '') || ' ' || coalesce(w.description, '')) + @@ to_tsquery('pg_catalog.german', :q) + ORDER BY rank DESC + LIMIT :lim + """ + ) + result = await db.execute( + sql, + {"q": tsquery, "tid": tenant_id, "lim": limit}, + ) + rows = result.mappings().all() + return [dict(r) for r in rows] + + async def _search_vector_filtered( + self, + db: AsyncSession, + embedding: list[float], + tenant_id: uuid.UUID, + limit: int, + visible_ids: set[uuid.UUID] | None, + ) -> list[dict[str, Any]]: + """No vector support for workflows — return empty list.""" + return [] + + async def get_embedding_text( + self, db: AsyncSession, entity_id: uuid.UUID, tenant_id: uuid.UUID + ) -> str: + """Get text for embedding generation — returns workflow name + description.""" + sql = text( + """ + SELECT name, description + FROM workflows + WHERE id = :eid AND tenant_id = :tid + """ + ) + result = await db.execute(sql, {"eid": entity_id, "tid": tenant_id}) + row = result.mappings().first() + if not row: + return "" + parts = [row.get("name", ""), row.get("description", "")] + return " ".join(str(p) for p in parts if p) + + def to_search_result(self, entity: object) -> dict[str, Any]: + """Convert workflow to search result dict.""" + if isinstance(entity, dict): + entity_id = str(entity.get("id", "")) + name = entity.get("name", "") + description = entity.get("description", "") + trigger_event = entity.get("trigger_event", "") + is_active = entity.get("is_active", True) + score = entity.get("rank", 0.0) + else: + entity_id = str(getattr(entity, "id", "")) + name = getattr(entity, "name", "") + description = getattr(entity, "description", "") + trigger_event = getattr(entity, "trigger_event", "") + is_active = getattr(entity, "is_active", True) + score = getattr(entity, "rank", 0.0) + return { + "entity_type": self.entity_type, + "entity_id": entity_id, + "title": name, + "snippet": description or "", + "score": float(score) if score else 0.0, + "data": { + "trigger_event": trigger_event, + "is_active": is_active, + }, + } diff --git a/app/plugins/builtins/unified_search/query_understanding.py b/app/plugins/builtins/unified_search/query_understanding.py index 73d8ee6..6886d7d 100644 --- a/app/plugins/builtins/unified_search/query_understanding.py +++ b/app/plugins/builtins/unified_search/query_understanding.py @@ -2,7 +2,6 @@ from __future__ import annotations -import os import json import logging import uuid @@ -13,8 +12,6 @@ from sqlalchemy.ext.asyncio import AsyncSession logger = logging.getLogger(__name__) -DEFAULT_LLM_MODEL = os.environ.get('SEARCH_LLM_MODEL', 'ollama/deepseek-v4-flash') - QUERY_ANALYZE_SYSTEM = ( "Du bist ein Query-Analyzer fuer ein CRM. " "Analysiere die Suchanfrage und gib JSON zurueck: " @@ -27,33 +24,9 @@ RESULT_AGGREGATE_SYSTEM = ( '{"summary": str, "facets": {"types": {}, "dates": {}, "people": []}, "suggestions": [str]}' ) - -async def _get_api_credentials( - db: AsyncSession | None, tenant_id: uuid.UUID | None -) -> tuple[str | None, str | None, str | None]: - """Get API key, base_url and provider_type from the default AI provider in DB. - - Falls back to API_KEY_OLLAMA_CLOUD env var. - Returns (api_key, base_url, provider_type). - """ - if db and tenant_id: - try: - from app.plugins.builtins.ai_assistant.contracts import get_default_provider - provider = await get_default_provider(db, tenant_id) - if provider and provider.api_key: - return provider.api_key, provider.base_url, provider.provider_type - except Exception: - logger.debug("Failed to get provider from DB, falling back to env") - env_key = os.environ.get('API_KEY_OLLAMA_CLOUD', '') - return (env_key if env_key else None), None, None - - -def _build_model(model: str, provider_type: str | None) -> str: - """Build litellm model string with provider prefix.""" - if provider_type: - model_parts = model.split("/", 1) - return f"{provider_type}/{model_parts[-1]}" - return model +# Sensible default model used when no provider-specific model is configured. +# llm_complete() resolves credentials/model prefix from the central config. +DEFAULT_LLM_MODEL = "ollama/deepseek-v4-flash" def _fallback_query_analysis(query: str) -> dict[str, Any]: @@ -74,6 +47,16 @@ def _fallback_aggregate(results: list[dict], query: str) -> dict[str, Any]: } +def _parse_json_content(content: str) -> dict[str, Any]: + """Parse LLM JSON response, stripping markdown code fences if present.""" + content = content.strip() + if content.startswith("```"): + content = content.split("\n", 1)[-1] if "\n" in content else content[3:] + if content.endswith("```"): + content = content[:-3].strip() + return json.loads(content) + + async def llm_analyze_query( query: str, db: AsyncSession | None = None, @@ -84,11 +67,8 @@ async def llm_analyze_query( Falls back to a simple dict if LLM fails. """ try: - api_key, api_base, provider_type = await _get_api_credentials(db, tenant_id) - model = _build_model(DEFAULT_LLM_MODEL, provider_type) - result = await llm_complete( - model=model, + model=DEFAULT_LLM_MODEL, messages=[ {"role": "system", "content": QUERY_ANALYZE_SYSTEM}, {"role": "user", "content": query}, @@ -96,17 +76,10 @@ async def llm_analyze_query( temperature=0.1, max_tokens=500, response_format={"type": "json_object"}, - api_key=api_key, - api_base=api_base, + db=db, + tenant_id=tenant_id, ) - content = result["content"] - # Strip markdown code fences if present - content = content.strip() - if content.startswith("```"): - content = content.split("\n", 1)[-1] if "\n" in content else content[3:] - if content.endswith("```"): - content = content[:-3].strip() - return json.loads(content) + return _parse_json_content(result["content"]) except Exception: logger.warning("LLM query analysis failed, using fallback", exc_info=True) return _fallback_query_analysis(query) @@ -125,9 +98,6 @@ async def llm_aggregate_results( if not results: return _fallback_aggregate(results, query) try: - api_key, api_base, provider_type = await _get_api_credentials(db, tenant_id) - model = _build_model(DEFAULT_LLM_MODEL, provider_type) - # Truncate results to avoid token overflow compact = [ {"entity_type": r.get("entity_type"), "title": r.get("title", "")[:100]} @@ -136,7 +106,7 @@ async def llm_aggregate_results( user_msg = json.dumps({"query": query, "results": compact}) result = await llm_complete( - model=model, + model=DEFAULT_LLM_MODEL, messages=[ {"role": "system", "content": RESULT_AGGREGATE_SYSTEM}, {"role": "user", "content": user_msg}, @@ -144,17 +114,10 @@ async def llm_aggregate_results( temperature=0.1, max_tokens=1000, response_format={"type": "json_object"}, - api_key=api_key, - api_base=api_base, + db=db, + tenant_id=tenant_id, ) - content = result["content"] - # Strip markdown code fences if present - content = content.strip() - if content.startswith("```"): - content = content.split("\n", 1)[-1] if "\n" in content else content[3:] - if content.endswith("```"): - content = content[:-3].strip() - return json.loads(content) + return _parse_json_content(result["content"]) except Exception: logger.warning("LLM result aggregation failed, using fallback", exc_info=True) return _fallback_aggregate(results, query) diff --git a/app/plugins/builtins/unified_search/routes.py b/app/plugins/builtins/unified_search/routes.py index 188ac57..3b34d95 100644 --- a/app/plugins/builtins/unified_search/routes.py +++ b/app/plugins/builtins/unified_search/routes.py @@ -4,6 +4,7 @@ from __future__ import annotations import logging import uuid +from datetime import datetime, timezone from fastapi import APIRouter, Depends, HTTPException, Query from sqlalchemy import text @@ -19,6 +20,7 @@ from app.plugins.builtins.unified_search.query_understanding import ( llm_analyze_query, ) from app.plugins.builtins.unified_search.schemas import ( + FacetsResponse, ProviderResponse, ReindexRequest, SearchRequest, @@ -48,15 +50,94 @@ async def search_get( entity_types: str | None = Query(None, description="Comma-separated entity types to search"), limit: int = Query(default=20, ge=1, le=100), offset: int = Query(default=0, ge=0), + date_from: str | None = Query(None, description="ISO date (YYYY-MM-DD) — filter by created_at/updated_at >= date_from"), + date_to: str | None = Query(None, description="ISO date (YYYY-MM-DD) — filter by created_at/updated_at <= date_to"), + tags: str | None = Query(None, description="Comma-separated tags to filter results"), + sort: str = Query(default="relevance", description="Sort order: relevance, date, name"), current_user: dict = Depends(get_current_user), db: AsyncSession = Depends(get_db), ) -> SearchResponse: """Perform hybrid search via GET (same as POST but with query params).""" types_list = entity_types.split(",") if entity_types else None - req = SearchRequest(query=q, entity_types=types_list, limit=limit, offset=offset) + tags_list = tags.split(",") if tags else None + req = SearchRequest( + query=q, + entity_types=types_list, + limit=limit, + offset=offset, + date_from=date_from, + date_to=date_to, + tags=tags_list, + sort=sort, + ) return await _do_search(req, current_user, db) +def _parse_date(value: str | None) -> datetime | None: + """Parse an ISO date string (YYYY-MM-DD) into a timezone-aware datetime.""" + if not value: + return None + try: + dt = datetime.fromisoformat(value) + if dt.tzinfo is None: + dt = dt.replace(tzinfo=timezone.utc) + return dt + except ValueError: + return None + + +def _apply_filters_and_sort( + results: list[dict], + req: SearchRequest, +) -> list[dict]: + """Apply post-search date/tags filters and sort order to results.""" + date_from = _parse_date(req.date_from) + date_to = _parse_date(req.date_to) + + filtered: list[dict] = [] + for r in results: + # Date range filter on created_at/updated_at + if date_from or date_to: + created = r.get("_created_at") + updated = r.get("_updated_at") + ts = updated or created + if ts is None: + continue + if isinstance(ts, str): + try: + ts = datetime.fromisoformat(ts.replace("Z", "+00:00")) + except ValueError: + continue + if ts.tzinfo is None: + ts = ts.replace(tzinfo=timezone.utc) + if date_from and ts < date_from: + continue + if date_to and ts > date_to: + continue + + # Tags filter (comma-separated on entity) + if req.tags: + entity_tags = r.get("_tags") or "" + tag_set = {t.strip().lower() for t in entity_tags.split(",") if t.strip()} + if not any(t.lower() in tag_set for t in req.tags): + continue + + filtered.append(r) + + # Sort order + sort = (req.sort or "relevance").lower() + if sort == "date": + filtered.sort( + key=lambda x: (x.get("_updated_at") or x.get("_created_at") or ""), + reverse=True, + ) + elif sort == "name": + filtered.sort(key=lambda x: (x.get("title") or "").lower()) + # 'relevance' keeps the existing score order + + return filtered + + async def _do_search( req: SearchRequest, current_user: dict, @@ -81,6 +162,9 @@ async def _do_search( is_system_admin=is_system_admin, ) + # Apply post-search filters (date range, tags) and sort + results = _apply_filters_and_sort(results, req) + # Resolve user permissions for field-level RBAC resolved_perms = await resolve_permissions(db, user_id, tenant_id) @@ -131,6 +215,70 @@ async def search( return await _do_search(req, current_user, db) +# ─── Facets ─── + +@router.get("/facets", dependencies=[Depends(require_permission("search:read"))]) +async def search_facets( + current_user: dict = Depends(get_current_user), + db: AsyncSession = Depends(get_db), +) -> FacetsResponse: + """Return available facets for search filtering (entity types, tags, date ranges).""" + tenant_id = uuid.UUID(current_user["tenant_id"]) + + # Entity types from the search registry + registry = get_search_registry() + entity_types = registry.get_entity_types() + + # Tags: distinct tags across searchable entities (comma-separated on entities) + tags: set[str] = set() + tag_tables = { + "contacts": "tags", + "companies": "tags", + } + for table, col in tag_tables.items(): + try: + result = await db.execute( + text( + f"SELECT DISTINCT unnest(string_to_array({col}, ',')) AS tag " + f"FROM {table} WHERE tenant_id = :tid AND deleted_at IS NULL AND {col} IS NOT NULL" + ), + {"tid": tenant_id}, + ) + for row in result.mappings().all(): + tag = (row.get("tag") or "").strip() + if tag: + tags.add(tag) + except Exception: + logger.debug("Facet tag query failed for %s", table) + + # Date ranges: min/max created_at across searchable entities + date_ranges: dict[str, Any] = {} + date_tables = ["contacts", "mails", "files", "calendar_entries"] + for table in date_tables: + try: + result = await db.execute( + text( + f"SELECT min(created_at) AS min_dt, max(created_at) AS max_dt " + f"FROM {table} WHERE tenant_id = :tid AND deleted_at IS NULL" + ), + {"tid": tenant_id}, + ) + row = result.mappings().first() + if row: + date_ranges[table] = { + "min": str(row.get("min_dt")) if row.get("min_dt") else None, + "max": str(row.get("max_dt")) if row.get("max_dt") else None, + } + except Exception: + logger.debug("Facet date query failed for %s", table) + + return FacetsResponse( + entity_types=entity_types, + tags=sorted(tags), + date_ranges=date_ranges, + ) + + # ─── Suggest / Autocomplete ─── @router.get("/suggest", dependencies=[Depends(require_permission("search:read"))]) @@ -205,14 +353,76 @@ async def reindex( if job_id: job_ids.append(job_id) + # Optionally re-index file chunks + if req.include_chunks and "file" in entity_types: + chunk_job_id = await enqueue_job("reindex_all") + if chunk_job_id: + job_ids.append(chunk_job_id) + return { "status": "ok", "message": f"Reindexing {len(entity_types)} entity types", "entity_types": entity_types, + "include_chunks": req.include_chunks, "job_ids": job_ids, } +# ─── Rebuild / Purge ─── + +@router.post("/rebuild/{entity_type}/{entity_id}", dependencies=[Depends(require_permission("search:admin"))]) +async def rebuild_entity_index( + entity_type: str, + entity_id: str, + current_user: dict = Depends(get_current_user), + db: AsyncSession = Depends(get_db), +) -> dict: + """Rebuild the search index for a single entity (admin only).""" + tenant_id = uuid.UUID(current_user["tenant_id"]) + try: + eid = uuid.UUID(entity_id) + except ValueError: + raise HTTPException(status_code=400, detail="Invalid entity_id") + + from app.plugins.builtins.unified_search.lifecycle import rebuild_index + + success = await rebuild_index(db, entity_type, eid, tenant_id) + return { + "status": "ok" if success else "failed", + "entity_type": entity_type, + "entity_id": entity_id, + "message": "Index rebuilt" if success else "Rebuild failed — check logs", + } + + +@router.post("/purge/{entity_type}/{entity_id}", dependencies=[Depends(require_permission("search:admin"))]) +async def purge_entity_index( + entity_type: str, + entity_id: str, + current_user: dict = Depends(get_current_user), + db: AsyncSession = Depends(get_db), +) -> dict: + """Purge an entity from the search index (admin only, GDPR).""" + tenant_id = uuid.UUID(current_user["tenant_id"]) + try: + eid = uuid.UUID(entity_id) + except ValueError: + raise HTTPException(status_code=400, detail="Invalid entity_id") + + from app.plugins.builtins.unified_search.lifecycle import remove_from_index, remove_chunks + + await remove_from_index(db, entity_type, eid, tenant_id) + if entity_type == "file": + await remove_chunks(db, eid, tenant_id) + + return { + "status": "ok", + "entity_type": entity_type, + "entity_id": entity_id, + "message": "Purged from search index", + } + + # ─── Providers ─── @router.get("/providers", dependencies=[Depends(require_permission("search:read"))]) @@ -228,6 +438,10 @@ async def list_providers( entity_type=p.entity_type, plugin_name="unified_search", is_active=True, + supports_fts=getattr(p, "supports_fts", True), + supports_vector=getattr(p, "supports_vector", True), + supports_rag=getattr(p, "supports_rag", False), + supports_graph=getattr(p, "supports_graph", False), ) for p in providers ] diff --git a/app/plugins/builtins/unified_search/schemas.py b/app/plugins/builtins/unified_search/schemas.py index 25f33f9..6091fc4 100644 --- a/app/plugins/builtins/unified_search/schemas.py +++ b/app/plugins/builtins/unified_search/schemas.py @@ -14,6 +14,19 @@ class SearchRequest(BaseModel): entity_types: list[str] | None = None limit: int = Field(default=20, ge=1, le=100) offset: int = Field(default=0, ge=0) + date_from: str | None = Field( + default=None, description="ISO date (YYYY-MM-DD) — filter by created_at/updated_at >= date_from" + ) + date_to: str | None = Field( + default=None, description="ISO date (YYYY-MM-DD) — filter by created_at/updated_at <= date_to" + ) + tags: list[str] | None = Field( + default=None, description="Filter results by tags (comma-separated on entities)" + ) + sort: str = Field( + default="relevance", + description="Sort order: relevance, date, name", + ) class SearchResult(BaseModel): @@ -34,6 +47,14 @@ class SearchResponse(BaseModel): suggestions: list[str] +class FacetsResponse(BaseModel): + """Available facets for search filtering.""" + + entity_types: list[str] + tags: list[str] + date_ranges: dict[str, Any] + + # ─── Similar ─── class SimilarRequest(BaseModel): @@ -61,6 +82,7 @@ class SuggestResponse(BaseModel): class ReindexRequest(BaseModel): entity_types: list[str] = Field(default_factory=list) + include_chunks: bool = Field(default=True, description="Re-index document chunks for files") # ─── Provider ─── @@ -69,3 +91,7 @@ class ProviderResponse(BaseModel): entity_type: str plugin_name: str is_active: bool + supports_fts: bool = True + supports_vector: bool = True + supports_rag: bool = False + supports_graph: bool = False diff --git a/app/plugins/builtins/unified_search/search_engine.py b/app/plugins/builtins/unified_search/search_engine.py index 71e10ad..b4e4fb2 100644 --- a/app/plugins/builtins/unified_search/search_engine.py +++ b/app/plugins/builtins/unified_search/search_engine.py @@ -28,6 +28,30 @@ RRF_ALPHA = 0.5 RRF_BETA = 0.5 +def rrf_fusion_multi( + result_lists: list[tuple[str, list[dict[str, Any]]]], + k: int = 60, +) -> list[dict[str, Any]]: + """Reciprocal Rank Fusion over N result lists. + + Each tuple is (mode_name, results). All lists get equal weight. + score = sum(1/(k+rank_i) for each list where item appears) + """ + fused: dict[str, dict[str, Any]] = {} + + for _mode_name, results in result_lists: + for rank, item in enumerate(results): + eid = str(item.get("id", "")) + if not eid: + continue + rrf_score = 1.0 / (k + rank + 1) + if eid not in fused: + fused[eid] = {**item, "_score": 0.0} + fused[eid]["_score"] += rrf_score + + return sorted(fused.values(), key=lambda x: x.get("_score", 0.0), reverse=True) + + def rrf_fusion( fts_results: list[dict[str, Any]], vec_results: list[dict[str, Any]], @@ -38,29 +62,16 @@ def rrf_fusion( ) -> list[dict[str, Any]]: """Reciprocal Rank Fusion of FTS and vector search results. + Backward-compatible wrapper around :func:`rrf_fusion_multi`. score = alpha * (1/(k+rank_fts)) + beta * (1/(k+rank_vec)) """ - fused: dict[str, dict[str, Any]] = {} - - for rank, item in enumerate(fts_results): - eid = str(item.get("id", "")) - if not eid: - continue - rrf_score = alpha * (1.0 / (k + rank + 1)) - if eid not in fused: - fused[eid] = {**item, "_score": 0.0, "_entity_type": entity_type} - fused[eid]["_score"] += rrf_score - - for rank, item in enumerate(vec_results): - eid = str(item.get("id", "")) - if not eid: - continue - rrf_score = beta * (1.0 / (k + rank + 1)) - if eid not in fused: - fused[eid] = {**item, "_score": 0.0, "_entity_type": entity_type} - fused[eid]["_score"] += rrf_score - - return sorted(fused.values(), key=lambda x: x.get("_score", 0.0), reverse=True) + fused = rrf_fusion_multi( + [("fts", fts_results), ("vec", vec_results)], + k=k, + ) + for item in fused: + item["_entity_type"] = entity_type + return fused async def hybrid_search( @@ -113,24 +124,57 @@ async def hybrid_search( fts_results: list[dict[str, Any]] = [] vec_results: list[dict[str, Any]] = [] + rag_results: list[dict[str, Any]] = [] + graph_results: list[dict[str, Any]] = [] - try: - fts_results = await provider.search_fts(db, tsquery, tenant_id, fetch_limit, user_id=user_id, is_system_admin=is_system_admin) - except Exception: - logger.exception("FTS search failed for %s", entity_type) + # Respect provider capability flags + if getattr(provider, "supports_fts", True): + try: + fts_results = await provider.search_fts(db, tsquery, tenant_id, fetch_limit, user_id=user_id, is_system_admin=is_system_admin) + except Exception: + logger.exception("FTS search failed for %s", entity_type) - if query_embedding: + if query_embedding and getattr(provider, "supports_vector", True): try: vec_results = await provider.search_vector(db, query_embedding, tenant_id, fetch_limit, user_id=user_id, is_system_admin=is_system_admin) except Exception: logger.exception("Vector search failed for %s", entity_type) - fused = rrf_fusion(fts_results, vec_results, entity_type) + # RAG / Graph modes are not implemented yet, but keep the fusion ready. + if getattr(provider, "supports_rag", False) and hasattr(provider, "search_rag") and query_embedding: + try: + rag_results = await provider.search_rag(db, query_embedding, tenant_id, fetch_limit, user_id=user_id, is_system_admin=is_system_admin) + except Exception: + logger.exception("RAG search failed for %s", entity_type) + + if getattr(provider, "supports_graph", False) and hasattr(provider, "search_graph"): + try: + graph_results = await provider.search_graph(db, query_analysis, tenant_id, fetch_limit, user_id=user_id, is_system_admin=is_system_admin) + except Exception: + logger.exception("Graph search failed for %s", entity_type) + + if rag_results or graph_results: + fused = rrf_fusion_multi( + [ + ("fts", fts_results), + ("vec", vec_results), + ("rag", rag_results), + ("graph", graph_results), + ] + ) + for item in fused: + item["_entity_type"] = entity_type + else: + fused = rrf_fusion(fts_results, vec_results, entity_type) # Convert to search result format for item in fused: result = provider.to_search_result(item) result["score"] = item.get("_score", 0.0) + # Preserve raw fields for post-search filtering (date/tags) + result["_created_at"] = item.get("created_at") + result["_updated_at"] = item.get("updated_at") + result["_tags"] = item.get("tags") all_results.append(result) all_results.sort(key=lambda x: x.get("score", 0.0), reverse=True) diff --git a/docs/api-documentation.md b/docs/api-documentation.md index 59d0130..9f3493e 100644 --- a/docs/api-documentation.md +++ b/docs/api-documentation.md @@ -324,16 +324,112 @@ Agent Builder, Automation Builder, Cron-Scheduler, Agent Runner. ### search (Unified Search) -7 endpoints for cross-entity search. +Hybrid cross-entity search (PostgreSQL FTS + pgvector) with KI query understanding, RRF rank fusion, visibility filtering, and field-level RBAC. | Method | Path | Description | |--------|------|-------------| -| POST | `/api/v1/search` | Unified search across all entities. | -| GET | `/api/v1/search/providers` | List search providers. | -| POST | `/api/v1/search/reindex` | Rebuild search index. | -| POST | `/api/v1/search/similar` | Find similar entities. | -| GET | `/api/v1/search/suggest` | Search suggestions. | -| GET | `/api/v1/search/stats` | Search index statistics. | +| GET | `/api/v1/search` | Hybrid search via query params (same as POST). | +| POST | `/api/v1/search` | Hybrid search with KI query understanding. | +| GET | `/api/v1/search/suggest` | Autocomplete suggestions (FTS prefix). | +| POST | `/api/v1/search/similar` | Find similar entities across all types by embedding. | +| POST | `/api/v1/search/reindex` | Trigger reindexing of entity types (admin). | +| GET | `/api/v1/search/providers` | List active search providers + capability flags. | +| POST | `/api/v1/search/providers/{entity_type}/toggle` | Toggle a provider on/off (admin). | +| GET | `/api/v1/search/stats` | Search index statistics (indexed/pending per table). | +| GET | `/api/v1/search/facets` | Available facets (entity types, tags, date ranges). | +| POST | `/api/v1/search/rebuild/{entity_type}/{entity_id}` | Rebuild index for a single entity (admin). | +| POST | `/api/v1/search/purge/{entity_type}/{entity_id}` | Purge entity from index (admin, GDPR). | + +> **MCP:** Unified search is also exposed as an MCP tool named `search` (category `search`, permission `search:read`) via the MCP server plugin (`app/plugins/builtins/mcp_server/tool_definitions.py`). + +#### Search Request (POST `/api/v1/search`) + +```json +{ + "query": "Max Mustermann", + "entity_types": ["contact", "mail"], + "limit": 20, + "offset": 0, + "date_from": "2026-01-01", + "date_to": "2026-12-31", + "tags": ["vip", "partner"], + "sort": "relevance" +} +``` + +| Field | Type | Default | Description | +|-------|------|---------|-------------| +| `query` | string | — | Search query (1–500 chars, required). | +| `entity_types` | list[string] | all | Restrict search to these entity types. | +| `limit` | int | 20 | Max results (1–100). | +| `offset` | int | 0 | Pagination offset. | +| `date_from` | string | null | ISO date `YYYY-MM-DD` — filter by `created_at`/`updated_at >= date_from`. | +| `date_to` | string | null | ISO date `YYYY-MM-DD` — filter by `created_at`/`updated_at <= date_to`. | +| `tags` | list[string] | null | Filter results by tags (comma-separated on entities). | +| `sort` | string | `relevance` | Sort order: `relevance`, `date`, `name`. | + +#### Search Response + +```json +{ + "query": "Max Mustermann", + "normalized_query": "max mustermann", + "results": [ + { + "entity_type": "contact", + "entity_id": "uuid", + "title": "Max Mustermann", + "snippet": "max@example.com", + "score": 0.95, + "data": {"type": "person"} + } + ], + "facets": {"types": {"contact": 1}}, + "summary": "1 Ergebnis", + "suggestions": [] +} +``` + +| Field | Type | Description | +|-------|------|-------------| +| `query` | string | Original query. | +| `normalized_query` | string | KI-normalized query. | +| `results` | list[SearchResult] | Ranked results (entity_type, entity_id, title, snippet, score, data). | +| `facets` | object | KI-generated facet counts. | +| `summary` | string | Human-readable summary. | +| `suggestions` | list[string] | Suggested follow-up filters. | + +#### GET `/api/v1/search` Query Params + +Same fields as the POST body, passed as query parameters: `q` (required), `entity_types` (comma-separated), `limit`, `offset`, `date_from`, `date_to`, `tags` (comma-separated), `sort`. + +#### GET `/api/v1/search/suggest` + +Query params: `q` (required, 1–200), `limit` (default 10, 1–50). Returns `{"suggestions": ["..."]}`. + +#### POST `/api/v1/search/similar` + +Body: `{"entity_type": "contact", "entity_id": "uuid", "limit": 5}`. Returns `{"similar": {"mail": [SearchResult...], ...}}`. + +#### POST `/api/v1/search/reindex` + +Body: `{"entity_types": ["contact"], "include_chunks": true}`. Returns `{"status": "ok", "entity_types": [...], "include_chunks": true, "job_ids": [...]}`. Requires `search:admin`. + +#### GET `/api/v1/search/providers` + +Returns a list of providers: `{"entity_type", "plugin_name", "is_active", "supports_fts", "supports_vector", "supports_rag", "supports_graph"}`. + +#### GET `/api/v1/search/facets` + +Returns `{"entity_types": [...], "tags": [...], "date_ranges": {"contacts": {"min": "...", "max": "..."}, ...}}`. + +#### GET `/api/v1/search/stats` + +Returns per-table `{"total", "indexed", "pending"}` plus `recent_logs` (last 10 index log entries). + +#### POST `/api/v1/search/rebuild/{entity_type}/{entity_id}` / `/purge/{entity_type}/{entity_id}` + +Admin-only. Rebuild regenerates the embedding + TSV; purge sets embedding/TSV to NULL (and removes chunks for files). Returns `{"status": "ok"|"failed", "entity_type", "entity_id", "message"}`. ### reports (Report Generator) diff --git a/docs/plugin-development-guide.md b/docs/plugin-development-guide.md index a8027ee..5c34e7a 100644 --- a/docs/plugin-development-guide.md +++ b/docs/plugin-development-guide.md @@ -685,7 +685,194 @@ async def on_deactivate(self, db, service_container, event_bus): --- -## 27. Testing Guide +## 27. Search Integration + +The Unified Search plugin provides hybrid cross-entity search (PostgreSQL FTS + pgvector) with KI query understanding, RRF rank fusion, visibility filtering, and field-level RBAC. Plugins can expose their entities to unified search by implementing a `SearchProvider`. + +### 27.1 Creating a SearchProvider + +Inherit from `BaseSearchProvider` and implement the required methods. The base class handles visibility filtering automatically (see 27.4). + +```python +# app/plugins/builtins/my_plugin/search_provider.py +from __future__ import annotations + +import uuid +from typing import Any + +from sqlalchemy import text +from sqlalchemy.ext.asyncio import AsyncSession + +from app.plugins.builtins.unified_search.base_provider import BaseSearchProvider + + +class MyEntitySearchProvider(BaseSearchProvider): + """Search provider for MyEntity.""" + + entity_type = "my_entity" + supports_fts = True + supports_vector = True + supports_rag = False + supports_graph = False + + async def _search_fts_filtered( + self, + db: AsyncSession, + tsquery: str, + tenant_id: uuid.UUID, + limit: int, + visible_ids: set[uuid.UUID] | None, + ) -> list[dict[str, Any]]: + """Full-text search filtered by visible_ids (None = no filter).""" + if visible_ids is not None: + sql = text( + """ + SELECT e.*, ts_rank(e.search_tsv, to_tsquery('pg_catalog.german', :q)) AS rank + FROM my_entities e + WHERE e.tenant_id = :tid + AND e.deleted_at IS NULL + AND e.search_tsv @@ to_tsquery('pg_catalog.german', :q) + AND e.id = ANY(:visible_ids) + ORDER BY rank DESC + LIMIT :lim + """ + ) + result = await db.execute( + sql, {"q": tsquery, "tid": tenant_id, "lim": limit, "visible_ids": list(visible_ids)} + ) + else: + sql = text( + """ + SELECT e.*, ts_rank(e.search_tsv, to_tsquery('pg_catalog.german', :q)) AS rank + FROM my_entities e + WHERE e.tenant_id = :tid + AND e.deleted_at IS NULL + AND e.search_tsv @@ to_tsquery('pg_catalog.german', :q) + ORDER BY rank DESC + LIMIT :lim + """ + ) + result = await db.execute(sql, {"q": tsquery, "tid": tenant_id, "lim": limit}) + return [dict(r) for r in result.mappings().all()] + + async def _search_vector_filtered( + self, + db: AsyncSession, + embedding: list[float], + tenant_id: uuid.UUID, + limit: int, + visible_ids: set[uuid.UUID] | None, + ) -> list[dict[str, Any]]: + """Semantic vector search on the entity's embedding column.""" + # Same pattern as _search_fts_filtered but using `embedding <=> cast(:emb AS vector)` + return [] + + async def get_embedding_text( + self, db: AsyncSession, entity_id: uuid.UUID, tenant_id: uuid.UUID + ) -> str: + """Return the text used for embedding generation (non-sensitive fields only).""" + sql = text( + "SELECT name, description FROM my_entities WHERE id = :eid AND tenant_id = :tid" + ) + result = await db.execute(sql, {"eid": entity_id, "tid": tenant_id}) + row = result.mappings().first() + if not row: + return "" + return " ".join(str(v) for v in row.values() if v) + + def to_search_result(self, entity: object) -> dict[str, Any]: + """Convert an ORM entity or dict to a search result dict.""" + if isinstance(entity, dict): + entity_id = str(entity.get("id", "")) + name = entity.get("name", "") + description = entity.get("description", "") + else: + entity_id = str(getattr(entity, "id", "")) + name = getattr(entity, "name", "") + description = getattr(entity, "description", "") + return { + "entity_type": self.entity_type, + "entity_id": entity_id, + "title": name, + "snippet": description or "", + "score": 0.0, + "data": {}, + } +``` + +### 27.2 Capability Flags + +Each provider declares which search modes it supports via class attributes. The registry and API use these flags to decide which search paths to run and to report capabilities to clients. + +| Flag | Default | Meaning | +|------|---------|---------| +| `supports_fts` | `True` | Full-text search via PostgreSQL `tsvector`/`tsquery`. | +| `supports_vector` | `True` | Semantic vector search via pgvector embeddings. | +| `supports_rag` | `False` | Retrieval-augmented generation over document chunks. | +| `supports_graph` | `False` | Graph-based search (GraphRAG). | + +Set `supports_vector = False` (and implement `_search_vector_filtered` returning `[]`) when an entity has no embedding column, e.g. chat messages or workflows. + +### 27.3 Registering a Provider + +Register providers during plugin activation. The Unified Search plugin calls `auto_register_providers(db)` on activation, which registers all built-in providers. For a custom plugin, register your provider directly in `on_activate`: + +```python +async def on_activate(self, db, service_container, event_bus) -> None: + await super().on_activate(db, service_container, event_bus) + from app.plugins.builtins.unified_search.provider_registry import get_search_registry + from app.plugins.builtins.my_plugin.search_provider import MyEntitySearchProvider + + get_search_registry().register(MyEntitySearchProvider()) +``` + +To add a provider to the built-in `auto_register_providers` list, import and append it to the `provider_cls` list in `app/plugins/builtins/unified_search/provider_registry.py`. + +### 27.4 Permission Filtering (Automatic) + +`BaseSearchProvider.search_fts` / `search_vector` automatically load the calling user's visible entity IDs via `get_visible_ids` and pass them to `_search_fts_filtered` / `_search_vector_filtered`. When `is_system_admin` is true or no `user_id` is provided, `visible_ids` is `None` and no filter is applied. Your `_search_*_filtered` implementation must honor `visible_ids` (add `AND id = ANY(:visible_ids)` when it is not `None`). + +### 27.5 Auto-Indexing via Events / Outbox + +Entities are indexed automatically when created or updated. The Unified Search plugin subscribes to domain events (e.g. `contact.created`, `contact.updated`, `mail.synced`, `file.uploaded`) and enqueues indexing jobs. For custom entities, publish the corresponding events or enqueue jobs directly: + +```python +from app.core.jobs import enqueue_job + +# After creating/updating an entity +await enqueue_job("index_entity", "my_entity", str(entity_id)) +``` + +Indexing jobs call `index_entity(entity_type, entity_id, tenant_id, db)`, which uses the provider's `get_embedding_text` to generate an embedding and stores it in the entity's `embedding` column. The `indexed_at` column tracks dedup so unchanged entities are not re-embedded. + +### 27.6 Lifecycle Hooks + +The Unified Search plugin subscribes to lifecycle events to keep the index consistent: + +| Event | Handler | Effect | +|-------|---------|--------| +| `entity.deleted` | `handle_entity_delete` | Removes embedding + TSV (and chunks for files). | +| `entity.restored` | `handle_entity_restore` | Rebuilds the search index. | +| `entity.corrected` | `handle_entity_correction` | Rebuilds the search index. | + +Publish these events (or call the lifecycle functions directly) when your plugin deletes, restores, or corrects entities so the search index stays in sync. + +### 27.7 RAG Document Chunking + +For RAG over long documents, use `chunk_text` from `app.plugins.builtins.unified_search.chunking` to split extracted text into overlapping chunks before embedding: + +```python +from app.plugins.builtins.unified_search.chunking import chunk_text + +chunks = chunk_text(document_text, chunk_size=1000, overlap=200) +# Each chunk: {"chunk_index": 0, "chunk_text": "...", "chunk_hash": "sha256..."} +``` + +Chunks are stored in the `document_chunks` table and embedded at the chunk level for fine-grained vector retrieval. `chunk_hash` is a deterministic SHA-256 of the chunk text, used for dedup. + +--- + +## 28. Testing Guide ### 11.1 Backend Tests diff --git a/docs/test-strategy.md b/docs/test-strategy.md index 5f2e945..959e92b 100644 --- a/docs/test-strategy.md +++ b/docs/test-strategy.md @@ -295,3 +295,54 @@ Diese Pipeline ist verbindlich für Phase-Gate-Reviews und muss vor jedem Phasen - ✅ Retention: GDPR-Hard-Delete nach konfigurierbarer Aufbewahrungsfrist - ✅ Hook-based History: `do_action('entity.after_create/update/delete')` → `record_history()` - ✅ Dynamic Permission Checks: Restore-Permission aus RestoreConfig, nicht hardcoded + +--- + +## Phase E — Unified Search Test-Konventionen + +### Neue Test-Datei: `tests/test_unified_search_phase_e.py` (40 Tests) + +| Test-Gruppe | Tests | Status | +|-------------|-------|--------| +| Provider-Capability-Flags (supports_fts/vector/rag/graph, get_providers_by_capability, get_capabilities) | 3 | ✅ | +| RRF Multi-Fusion (2/3/4 Listen, Multi-Listen-Scoring, Backward-Compat, Empty Inputs) | 5 | ✅ | +| Chunking (empty/short/long/exact multiple, Overlap, Hash deterministisch, Whitespace-Normalisierung) | 6 | ✅ | +| Lifecycle (remove_from_index setzt embedding+TSV NULL, rebuild_index, entity.deleted/restored) | 6 | ✅ | +| API-Filter (date_from/date_to, tags, sort, facets-Struktur) | 4 | ✅ | +| AI-Tool (Name/Description, Parameter, OpenAI-Schema, Handler kompakt, Fehlerfälle) | 5 | ✅ | +| Neue Provider (AgentMemory, AIChat, Workflow — Import + Flags) | 4 | ✅ | +| Sensitive-Fields-Exclusion (nicht in search_tsv, nicht in Embedding-Text, Redaction) | 4 | ✅ | + +### Konventionen für Search-Tests + +1. **Isolation & Determinismus:** Jeder Test nutzt eine eigene Tenant-ID und eigene Entity-IDs. Keine zufälligen UUIDs — echte IDs aus der DB verwenden. +2. **Schema-Anpassung idempotent:** Die Test-DB (`create_all`) definiert `search_tsv` als generierte Spalte, Produktion (Migration 0001) als Plain-Column mit Trigger. Der Lifecycle-Helper konvertiert die Spalte idempotent per `DO $$ ... DROP EXPRESSION` und droppt den `contacts_tsv_update`-Trigger, damit `remove_from_index` `search_tsv = NULL` setzen kann. Die Schema-Änderung persistiert über Testläufe (nur Tabellen werden getruncated). +3. **Patch-Targets:** Funktionen, die innerhalb einer Funktion importiert werden (z.B. `index_entity` in `lifecycle.py`, `get_session_factory` in `ai_tool.py`), müssen am Ursprungsmodul gepatcht werden (`app.plugins.builtins.unified_search.embedding.index_entity`, `app.core.db.get_session_factory`), nicht am importierenden Modul. +4. **Keine echten LLM/Embedding-Calls:** Alle KI-Aufrufe werden mit `AsyncMock` gemockt. Kein Test darf ein echtes Modell kontaktieren. + +### Mock-Patterns für LLM/Embedding-Calls + +```python +from unittest.mock import AsyncMock, patch + +# LLM-Query-Verständnis (Normalisierung, Facets, Summary) +with patch("app.plugins.builtins.unified_search.llm.llm_complete", new_callable=AsyncMock) as mock_llm: + mock_llm.return_value = {"normalized_query": "max mustermann", "facets": {}, "summary": "1 Ergebnis"} + # ... Test + +# Embedding-Generierung +with patch("app.plugins.builtins.unified_search.embedding.generate_embedding", new_callable=AsyncMock) as mock_emb: + mock_emb.return_value = [0.1, 0.2, 0.3] + # ... Test + +# LLM-Embedding-Client +with patch("app.plugins.builtins.unified_search.embedding.llm_embed", new_callable=AsyncMock) as mock_llm_emb: + mock_llm_emb.return_value = [0.1, 0.2, 0.3] + # ... Test +``` + +**Regeln:** +- `llm_complete` liefert ein Dict mit `normalized_query`, `facets`, `summary` (und optional `suggestions`). +- `generate_embedding` / `llm_embed` liefern eine Liste von Floats (Embedding-Vektor). +- Bei Fehlerpfaden: `mock_llm.side_effect = Exception("...")` oder `return_value = None` für Fallback-Verhalten testen. +- DB-Session-Factory in AI-Tool-Handler-Tests: `patch("app.core.db.get_session_factory", return_value=sf)` mit `async_sessionmaker(bind=db_session.bind, ...)`. diff --git a/frontend/src/App.tsx b/frontend/src/App.tsx index a839867..dac982d 100644 --- a/frontend/src/App.tsx +++ b/frontend/src/App.tsx @@ -7,6 +7,8 @@ import { useThemeStore } from '@/store/themeStore'; import { ErrorBoundary } from '@/components/common/ErrorBoundary'; import { useToast } from '@/components/ui/Toast'; import { useOnlineStatus } from '@/hooks/useOnlineStatus'; +import { CommandPalette } from '@/components/search/CommandPalette'; +import { useCommandPalette } from '@/hooks/useCommandPalette'; function QueryClientWrapper({ children }: { children: React.ReactNode }) { const toast = useToast(); @@ -79,6 +81,7 @@ export default function App() { + ); } diff --git a/frontend/src/api/hooks.ts b/frontend/src/api/hooks.ts index 456289b..1145edd 100644 --- a/frontend/src/api/hooks.ts +++ b/frontend/src/api/hooks.ts @@ -24,16 +24,16 @@ export * from './automation'; // ── Search hook (uses dynamic import, kept inline) ── -import type { SearchResult } from './search'; -export type { SearchResult }; +import type { SearchFilters, SearchResult } from './search'; +export type { SearchResult, SearchFilters }; -export function useGlobalSearch(query: string, entityTypes?: string[]) { +export function useGlobalSearch(query: string, filters: SearchFilters = {}) { return useQuery({ - queryKey: ['globalSearch', query, entityTypes], + queryKey: ['globalSearch', query, filters], queryFn: async (): Promise => { if (!query.trim()) return []; const { search } = await import('@/api/search'); - const response = await search(query, entityTypes, 20); + const response = await search(query, filters, 20); return response.results || []; }, enabled: query.trim().length > 0, diff --git a/frontend/src/api/search.ts b/frontend/src/api/search.ts index e712c02..efba9dc 100644 --- a/frontend/src/api/search.ts +++ b/frontend/src/api/search.ts @@ -10,13 +10,32 @@ export interface SearchResult { data?: Record; } +export interface SearchFacet { + value: string; + count: number; +} + +export interface SearchFilters { + entityTypes?: string[]; + tags?: string[]; + dateFrom?: string; + dateTo?: string; + sort?: 'relevance' | 'date' | 'name'; +} + export interface SearchResponse { results: SearchResult[]; - facets?: Record>; + facets?: Record; summary?: string; suggestions?: string[]; } +export interface FacetsResponse { + entity_types: string[]; + tags: string[]; + date_ranges: Record; +} + // Map backend entity_type to frontend route URL const ENTITY_URL_MAP: Record) => string> = { contact: (id) => `/contacts/${id}`, @@ -46,10 +65,14 @@ function mapBackendResult(r: Record): SearchResult { }; } -export async function search(query: string, entityTypes?: string[], limit = 20): Promise { +export async function search(query: string, filters: SearchFilters = {}, limit = 20): Promise { const r = await apiClient.post('/search', { query, - entity_types: entityTypes, + entity_types: filters.entityTypes, + tags: filters.tags, + date_from: filters.dateFrom, + date_to: filters.dateTo, + sort: filters.sort || 'relevance', limit, }); const data = r.data; @@ -63,6 +86,11 @@ export async function search(query: string, entityTypes?: string[], limit = 20): }; } +export async function fetchFacets(): Promise { + const r = await apiClient.get('/search/facets'); + return r.data as FacetsResponse; +} + export async function searchSuggest(q: string): Promise { const r = await apiClient.get('/search/suggest', { params: { q } }); return r.data.suggestions || []; diff --git a/frontend/src/api/searchHooks.ts b/frontend/src/api/searchHooks.ts new file mode 100644 index 0000000..c7a7469 --- /dev/null +++ b/frontend/src/api/searchHooks.ts @@ -0,0 +1,13 @@ +import { useQuery } from '@tanstack/react-query'; +import type { FacetsResponse } from './search'; + +export function useSearchFacets() { + return useQuery({ + queryKey: ['searchFacets'], + queryFn: async (): Promise => { + const { fetchFacets } = await import('@/api/search'); + return fetchFacets(); + }, + staleTime: 5 * 60 * 1000, + }); +} diff --git a/frontend/src/components/layout/AppShell.tsx b/frontend/src/components/layout/AppShell.tsx index fc5e278..c8d9cf7 100644 --- a/frontend/src/components/layout/AppShell.tsx +++ b/frontend/src/components/layout/AppShell.tsx @@ -13,6 +13,7 @@ import { WindowContainer } from '@/components/window/WindowContainer'; import { ErrorBoundary } from '@/components/common/ErrorBoundary'; import { WelcomeDialog } from '@/components/onboarding/WelcomeDialog'; import { OnboardingTour } from '@/components/onboarding/OnboardingTour'; +import { CommandPalette } from '@/components/search/CommandPalette'; import { useOnboardingStore } from '@/store/onboardingStore'; export function AppShell() { @@ -57,6 +58,7 @@ export function AppShell() { {showMessageSidebar && } + diff --git a/frontend/src/components/layout/TopBar.tsx b/frontend/src/components/layout/TopBar.tsx index 32e16f1..904a849 100644 --- a/frontend/src/components/layout/TopBar.tsx +++ b/frontend/src/components/layout/TopBar.tsx @@ -8,11 +8,12 @@ import { useLogout } from '@/api/hooks'; import { Avatar } from '@/components/ui/Avatar'; import { SearchDropdown } from '@/components/shared/SearchDropdown'; import { SuggestionBadge } from '@/components/ai/SuggestionBadge'; -import { Building, ChevronDown, Menu, Zap, Bot, Layers, Code } from 'lucide-react'; +import { Building, ChevronDown, Menu, Zap, Bot, Layers, Code, Search } from 'lucide-react'; import { NotificationBell } from '@/components/layout/NotificationBell'; import { WorkspaceSwitcher } from '@/components/layout/WorkspaceSwitcher'; import { useWindowStore } from '@/store/windowStore'; import { usePermission } from '@/hooks/usePermission'; +import { useCommandPaletteStore } from '@/store/commandPaletteStore'; export function TopBar() { const { t } = useTranslation(); @@ -86,6 +87,16 @@ export function TopBar() {
+ {/* Command palette shortcut */} +
diff --git a/frontend/src/components/search/CommandPalette.tsx b/frontend/src/components/search/CommandPalette.tsx new file mode 100644 index 0000000..eb2e107 --- /dev/null +++ b/frontend/src/components/search/CommandPalette.tsx @@ -0,0 +1,317 @@ +import React, { useCallback, useEffect, useMemo, useRef, useState } from 'react'; +import clsx from 'clsx'; +import { useNavigate } from 'react-router-dom'; +import { useTranslation } from 'react-i18next'; +import { Search, Loader2, Clock, X } from 'lucide-react'; +import { useGlobalSearch, SearchResult } from '@/api/hooks'; +import { useCommandPalette } from '@/hooks/useCommandPalette'; + +const RECENT_KEY = 'leocrm_recent_searches'; +const MAX_RECENT = 5; + +const TYPE_LABELS: Record = { + company: 'search.companies', + contact: 'search.contacts', + mail: 'search.mails', + file: 'search.files', + event: 'search.events', + message: 'search.messages', +}; + +const TYPE_ICON_CLASSES: Record = { + company: 'bg-primary-100 text-primary-700', + contact: 'bg-accent-100 text-accent-700', + mail: 'bg-success-100 text-success-700', + file: 'bg-warning-100 text-warning-700', + event: 'bg-secondary-100 text-secondary-700', + message: 'bg-secondary-100 text-secondary-700', +}; + +function typeIcon(type: string): string { + switch (type) { + case 'company': return 'F'; + case 'contact': return 'K'; + case 'mail': return '@'; + case 'file': return '📄'; + case 'event': return '📅'; + case 'message': return '💬'; + default: return '?'; + } +} + +function loadRecent(): string[] { + try { + const raw = localStorage.getItem(RECENT_KEY); + if (!raw) return []; + const parsed = JSON.parse(raw); + return Array.isArray(parsed) ? parsed.filter((x): x is string => typeof x === 'string') : []; + } catch { + return []; + } +} + +function saveRecent(query: string) { + const trimmed = query.trim(); + if (!trimmed) return; + const next = [trimmed, ...loadRecent().filter((q) => q !== trimmed)].slice(0, MAX_RECENT); + try { + localStorage.setItem(RECENT_KEY, JSON.stringify(next)); + } catch { + // ignore storage errors + } +} + +export function CommandPalette() { + const { t } = useTranslation(); + const navigate = useNavigate(); + const { isOpen, close } = useCommandPalette(); + + const [query, setQuery] = useState(''); + const [debouncedQuery, setDebouncedQuery] = useState(''); + const [activeIndex, setActiveIndex] = useState(-1); + const [recent, setRecent] = useState([]); + const inputRef = useRef(null); + const dialogRef = useRef(null); + const previouslyFocused = useRef(null); + + const { data: results, isLoading, isError } = useGlobalSearch(debouncedQuery); + + // Debounce query + useEffect(() => { + const timer = setTimeout(() => setDebouncedQuery(query), 300); + return () => clearTimeout(timer); + }, [query]); + + // Load recent searches when opened + useEffect(() => { + if (isOpen) { + setRecent(loadRecent()); + setQuery(''); + setDebouncedQuery(''); + setActiveIndex(-1); + } + }, [isOpen]); + + // Focus management + focus trap + useEffect(() => { + if (!isOpen) return; + previouslyFocused.current = document.activeElement as HTMLElement; + const timer = setTimeout(() => inputRef.current?.focus(), 0); + return () => { + clearTimeout(timer); + previouslyFocused.current?.focus(); + }; + }, [isOpen]); + + const handleKeyDown = useCallback((e: KeyboardEvent) => { + if (!isOpen) return; + if (e.key === 'Escape') { + e.preventDefault(); + close(); + return; + } + if (e.key === 'Tab') { + // Simple focus trap: keep focus within the dialog + const focusable = dialogRef.current?.querySelectorAll( + 'button, [href], input, select, textarea, [tabindex]:not([tabindex="-1"])' + ); + if (!focusable || focusable.length === 0) return; + const first = focusable[0]; + const last = focusable[focusable.length - 1]; + if (e.shiftKey && document.activeElement === first) { + e.preventDefault(); + last.focus(); + } else if (!e.shiftKey && document.activeElement === last) { + e.preventDefault(); + first.focus(); + } + } + }, [isOpen, close]); + + useEffect(() => { + document.addEventListener('keydown', handleKeyDown); + return () => document.removeEventListener('keydown', handleKeyDown); + }, [handleKeyDown]); + + const flatResults = useMemo(() => results || [], [results]); + + const groupedResults = useMemo(() => { + const groups: Record = {}; + for (const r of flatResults) { + if (!groups[r.type]) groups[r.type] = []; + groups[r.type].push(r); + } + return groups; + }, [flatResults]); + + const totalCount = flatResults.length; + + const handleResultClick = (result: SearchResult) => { + saveRecent(debouncedQuery); + close(); + navigate(result.url); + }; + + const handleRecentClick = (q: string) => { + setQuery(q); + setDebouncedQuery(q); + setActiveIndex(-1); + }; + + const handleInputKeyDown = (e: React.KeyboardEvent) => { + if (e.key === 'ArrowDown') { + e.preventDefault(); + setActiveIndex((prev) => Math.min(prev + 1, Math.max(totalCount - 1, 0))); + } else if (e.key === 'ArrowUp') { + e.preventDefault(); + setActiveIndex((prev) => Math.max(prev - 1, -1)); + } else if (e.key === 'Enter' && activeIndex >= 0 && flatResults[activeIndex]) { + e.preventDefault(); + handleResultClick(flatResults[activeIndex]); + } + }; + + if (!isOpen) return null; + + return ( +
+ + ); +} diff --git a/frontend/src/components/search/SavedSearches.tsx b/frontend/src/components/search/SavedSearches.tsx new file mode 100644 index 0000000..64c978c --- /dev/null +++ b/frontend/src/components/search/SavedSearches.tsx @@ -0,0 +1,144 @@ +import React, { useCallback, useEffect, useState } from 'react'; +import { useTranslation } from 'react-i18next'; +import { Trash2, Bookmark, BookmarkCheck } from 'lucide-react'; +import { SearchFilters } from '@/api/hooks'; +import { Button } from '@/components/ui/Button'; +import { Input } from '@/components/ui/Input'; + +const SAVED_KEY = 'leocrm_saved_searches'; +const MAX_SAVED = 20; + +export interface SavedSearch { + id: string; + name: string; + query: string; + filters: SearchFilters; + createdAt: string; +} + +function loadSaved(): SavedSearch[] { + try { + const raw = localStorage.getItem(SAVED_KEY); + if (!raw) return []; + const parsed = JSON.parse(raw); + return Array.isArray(parsed) ? parsed : []; + } catch { + return []; + } +} + +function persistSaved(saved: SavedSearch[]) { + try { + localStorage.setItem(SAVED_KEY, JSON.stringify(saved)); + } catch { + // ignore storage errors + } +} + +interface SavedSearchesProps { + currentQuery: string; + currentFilters: SearchFilters; + onRun: (query: string, filters: SearchFilters) => void; +} + +export function SavedSearches({ currentQuery, currentFilters, onRun }: SavedSearchesProps) { + const { t } = useTranslation(); + const [saved, setSaved] = useState([]); + const [showSaveForm, setShowSaveForm] = useState(false); + const [name, setName] = useState(''); + + useEffect(() => { + setSaved(loadSaved()); + }, []); + + const refresh = useCallback(() => setSaved(loadSaved()), []); + + const handleSave = () => { + const trimmed = name.trim(); + if (!trimmed || !currentQuery.trim()) return; + const entry: SavedSearch = { + id: `${Date.now()}-${Math.random().toString(36).slice(2, 8)}`, + name: trimmed, + query: currentQuery, + filters: currentFilters, + createdAt: new Date().toISOString(), + }; + const next = [entry, ...loadSaved()].slice(0, MAX_SAVED); + persistSaved(next); + setName(''); + setShowSaveForm(false); + refresh(); + }; + + const handleDelete = (id: string) => { + persistSaved(loadSaved().filter((s) => s.id !== id)); + refresh(); + }; + + const canSave = currentQuery.trim().length > 0; + + return ( +
+
+

{t('search.savedSearches')}

+ {canSave && !showSaveForm && ( + + )} +
+ + {showSaveForm && ( +
+
+ setName(e.target.value)} + placeholder={t('search.saveSearchName')} + aria-label={t('search.saveSearchName')} + data-testid="save-search-name" + /> +
+ + +
+ )} + + {saved.length === 0 ? ( +

{t('search.noSavedSearches')}

+ ) : ( +
    + {saved.map((s) => ( +
  • + + +
    • + ))} +
+ )} +
+ ); +} diff --git a/frontend/src/components/search/SearchFacets.tsx b/frontend/src/components/search/SearchFacets.tsx new file mode 100644 index 0000000..b7d0fe5 --- /dev/null +++ b/frontend/src/components/search/SearchFacets.tsx @@ -0,0 +1,206 @@ +import React, { useEffect, useState } from 'react'; +import { useTranslation } from 'react-i18next'; +import { SearchFilters } from '@/api/hooks'; +import { fetchFacets, FacetsResponse } from '@/api/search'; +import { Input } from '@/components/ui/Input'; +import { Select } from '@/components/ui/Select'; +import { Button } from '@/components/ui/Button'; +import { Skeleton } from '@/components/ui/Skeleton'; + +const TYPE_LABELS: Record = { + company: 'search.companies', + contact: 'search.contacts', + mail: 'search.mails', + file: 'search.files', + event: 'search.events', + message: 'search.messages', +}; + +interface SearchFacetsProps { + filters: SearchFilters; + onChange: (filters: SearchFilters) => void; + onClear: () => void; + entityCounts?: Record; + tagCounts?: Record; +} + +export function SearchFacets({ filters, onChange, onClear, entityCounts, tagCounts }: SearchFacetsProps) { + const { t } = useTranslation(); + const [facets, setFacets] = useState(null); + const [isLoading, setIsLoading] = useState(true); + + useEffect(() => { + let cancelled = false; + fetchFacets() + .then((data) => { + if (!cancelled) setFacets(data); + }) + .catch(() => { + // ignore — facets are optional; the page still works without them + }) + .finally(() => { + if (!cancelled) setIsLoading(false); + }); + return () => { + cancelled = true; + }; + }, []); + + const entityTypes = facets?.entity_types || []; + const tags = facets?.tags || []; + + const toggleEntityType = (type: string) => { + const current = filters.entityTypes || []; + const next = current.includes(type) + ? current.filter((x) => x !== type) + : [...current, type]; + onChange({ ...filters, entityTypes: next.length > 0 ? next : undefined }); + }; + + const toggleTag = (tag: string) => { + const current = filters.tags || []; + const next = current.includes(tag) + ? current.filter((x) => x !== tag) + : [...current, tag]; + onChange({ ...filters, tags: next.length > 0 ? next : undefined }); + }; + + const hasActiveFilters = + (filters.entityTypes?.length || 0) > 0 || + (filters.tags?.length || 0) > 0 || + !!filters.dateFrom || + !!filters.dateTo || + (filters.sort && filters.sort !== 'relevance'); + + return ( +
+
+

{t('search.filters')}

+ {hasActiveFilters && ( + + )} +
+ + {/* Entity type filter */} +
+

+ {t('search.entityType')} +

+ {isLoading ? ( +
+ {[1, 2, 3].map((i) => )} +
+ ) : entityTypes.length === 0 ? ( +

{t('common.none')}

+ ) : ( +
    + {entityTypes.map((type) => { + const checked = filters.entityTypes?.includes(type) || false; + const count = entityCounts?.[type] ?? 0; + return ( +
  • + +
    • + ); + })} +
+ )} +
+ + {/* Tag filter */} +
+

+ {t('search.tags')} +

+ {isLoading ? ( +
+ {[1, 2, 3].map((i) => )} +
+ ) : tags.length === 0 ? ( +

{t('common.none')}

+ ) : ( +
    + {tags.map((tag) => { + const checked = filters.tags?.includes(tag) || false; + const count = tagCounts?.[tag] ?? 0; + return ( +
  • + +
    • + ); + })} +
+ )} +
+ + {/* Date range filter */} +
+

+ {t('search.dateRange')} +

+
+ onChange({ ...filters, dateFrom: e.target.value || undefined })} + data-testid="search-date-from" + /> + onChange({ ...filters, dateTo: e.target.value || undefined })} + data-testid="search-date-to" + /> +
+
+ + {/* Sort selector */} +
+

+ {t('common.sort')} +

+