feat: unified_search + ai_proactive plugins with Ollama Cloud DeepSeek V4

- unified_search: Hybride Suche (PostgreSQL FTS + pgvector + RRF Fusion) - 5 Search Providers (Contact, Company, Mail, File, Event) - KI Query Understanding (Fuzzy, Facetten via LiteLLM) - DMS Text-Extraction (PDF, DOCX, XLSX, PPTX) - Embedding Pipeline (ollama/nomic-embed-text, 768 Dim) - Background Jobs für Indexierung - Plugin-basierte Provider Registry - ai_proactive: Proaktiver KI-Agent - Context-Tracking (Frontend → Backend → Event Bus) - Proactive Engine mit LLM Suggestion-Generierung - SSE Real-time Push an Frontend - 6 AI Tools für Tool Registry - Rate-Limiting + User Settings - Deep Analysis Background Jobs - Frontend Integration: - useAIContext Hook, SuggestionSidebar, SuggestionBadge - ProactiveAISettings Page, Search API Client - Globale Suche auf neue API umgestellt - Tests: test_unified_search.py + test_ai_proactive.py (alle bestanden) - Config: Ollama Cloud DeepSeek V4 als Default, konfigurierbar - Dependencies: PyMuPDF, python-docx, python-pptx, pgvector - Bugfixes: notification type_key length, migration IF NOT EXISTS
2026-07-18 11:21:51 +02:00
parent 4c9295de21
commit 8cebb4f4e9
52 changed files with 6493 additions and 41 deletions
@@ -0,0 +1,7 @@
+"""Unified Search plugin — hybrid FTS + semantic search across all CRM data."""
+
+from __future__ import annotations
+
+from app.plugins.builtins.unified_search.plugin import UnifiedSearchPlugin
+
+__all__ = ["UnifiedSearchPlugin"]
@@ -0,0 +1,131 @@
+"""Embedding pipeline using LiteLLM with configurable Ollama Cloud provider."""
+
+from __future__ import annotations
+
+import os
+import logging
+import uuid
+from typing import TYPE_CHECKING
+
+import litellm
+
+if TYPE_CHECKING:
+    from sqlalchemy.ext.asyncio import AsyncSession
+
+logger = logging.getLogger(__name__)
+
+MAX_INPUT_CHARS = 8000
+
+DEFAULT_EMBEDDING_MODEL = os.environ.get('SEARCH_EMBEDDING_MODEL', 'ollama/nomic-embed-text')
+OLLAMA_API_KEY = os.environ.get('API_KEY_OLLAMA_CLOUD', '')
+
+
+async def generate_embedding(text: str, model: str | None = None) -> list[float]:
+    """Generate a single embedding via LiteLLM.
+
+    Args:
+        text: Input text (truncated to 8000 chars).
+        model: Embedding model name (default: ollama/nomic-embed-text).
+
+    Returns:
+        Embedding vector as list of floats.
+    """
+    model = model or DEFAULT_EMBEDDING_MODEL
+    truncated = text[:MAX_INPUT_CHARS]
+    try:
+        response = await litellm.aembedding(
+            model=model,
+            input=truncated,
+            api_key=OLLAMA_API_KEY,
+        )
+        return response.data[0]["embedding"]
+    except Exception:
+        logger.exception("Failed to generate embedding")
+        return []
+
+
+async def generate_embeddings_batch(
+    texts: list[str], model: str | None = None
+) -> list[list[float]]:
+    """Generate embeddings for multiple texts in a single API call.
+
+    Args:
+        texts: List of input texts.
+        model: Embedding model name (default: ollama/nomic-embed-text).
+
+    Returns:
+        List of embedding vectors.
+    """
+    model = model or DEFAULT_EMBEDDING_MODEL
+    truncated = [t[:MAX_INPUT_CHARS] for t in texts]
+    try:
+        response = await litellm.aembedding(
+            model=model,
+            input=truncated,
+            api_key=OLLAMA_API_KEY,
+        )
+        return [d["embedding"] for d in response.data]
+    except Exception:
+        logger.exception("Failed to generate batch embeddings")
+        return [[] for _ in texts]
+
+
+async def index_entity(
+    entity_type: str,
+    entity_id: uuid.UUID,
+    tenant_id: uuid.UUID,
+    db: AsyncSession,
+) -> bool:
+    """Generate and store embedding for a single entity.
+
+    Uses the provider registry to get embedding text, generates embedding,
+    and updates the entity's embedding column.
+
+    Returns True on success, False on failure.
+    """
+    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)
+        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)
+            return False
+
+        embedding = await generate_embedding(text)
+        if not embedding:
+            return False
+
+        # Update the entity's embedding column
+        from sqlalchemy import text as sql_text
+
+        table_map = {
+            "contact": "contacts",
+            "company": "companies",
+            "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
+    except Exception:
+        logger.exception("Failed to index entity %s/%s", entity_type, entity_id)
+        return False
@@ -0,0 +1,246 @@
+"""ARQ background jobs for the Unified Search plugin."""
+
+from __future__ import annotations
+
+import logging
+import uuid
+from typing import Any
+
+from app.core.db import get_session_factory
+
+logger = logging.getLogger(__name__)
+
+BATCH_SIZE = 100
+
+
+def _parse_id(id_str: str) -> uuid.UUID:
+    """Parse a string to UUID."""
+    if isinstance(id_str, uuid.UUID):
+        return id_str
+    return uuid.UUID(str(id_str))
+
+
+async def index_mails(ctx: dict[str, Any], mail_ids: list[str]) -> None:
+    """Index mails: generate and store embeddings."""
+    from app.plugins.builtins.unified_search.embedding import index_entity
+
+    factory = get_session_factory()
+    async with factory() as db:
+        for mail_id in mail_ids:
+            try:
+                eid = _parse_id(mail_id)
+                # tenant_id is derived from the mail itself
+                from sqlalchemy import text
+                result = await db.execute(
+                    text("SELECT tenant_id FROM mails WHERE id = :mid"),
+                    {"mid": eid},
+                )
+                row = result.mappings().first()
+                if not row:
+                    continue
+                tenant_id = row["tenant_id"]
+                await index_entity("mail", eid, tenant_id, db)
+            except Exception:
+                logger.exception("Failed to index mail %s", mail_id)
+
+
+async def index_file(ctx: dict[str, Any], file_id: str) -> None:
+    """Index a file: extract text, store content_text, generate embedding."""
+    from sqlalchemy import text
+    from app.plugins.builtins.unified_search.text_extraction import extract_text_from_file
+    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 FROM files WHERE id = :fid"),
+                {"fid": eid},
+            )
+            row = result.mappings().first()
+            if not row:
+                logger.warning("File not found: %s", file_id)
+                return
+
+            tenant_id = row["tenant_id"]
+            storage_path = row["storage_path"]
+            mime_type = row["mime_type"]
+
+            # Extract text
+            content_text = await extract_text_from_file(storage_path, mime_type)
+
+            # Store content_text
+            await db.execute(
+                text("UPDATE files SET content_text = :ct WHERE id = :fid"),
+                {"ct": content_text, "fid": eid},
+            )
+            await db.commit()
+
+            # Generate embedding from extracted text + filename
+            result_name = await db.execute(
+                text("SELECT name FROM files WHERE id = :fid"),
+                {"fid": eid},
+            )
+            name_row = result_name.mappings().first()
+            name = name_row["name"] if name_row else ""
+            embedding_text = f"{name} {content_text[:5000]}"
+
+            if embedding_text.strip():
+                embedding = await generate_embedding(embedding_text)
+                if embedding:
+                    await db.execute(
+                        text("UPDATE files SET embedding = cast(:emb AS vector) WHERE id = :fid"),
+                        {"emb": str(embedding), "fid": eid},
+                    )
+                    await db.commit()
+
+            logger.info("Indexed file %s", file_id)
+        except Exception:
+            logger.exception("Failed to index file %s", file_id)
+
+
+async def index_contact(ctx: dict[str, Any], contact_id: str) -> None:
+    """Index a contact: generate and store embedding."""
+    from sqlalchemy import text
+    from app.plugins.builtins.unified_search.embedding import index_entity
+
+    factory = get_session_factory()
+    async with factory() as db:
+        try:
+            eid = _parse_id(contact_id)
+            result = await db.execute(
+                text("SELECT tenant_id FROM contacts WHERE id = :cid"),
+                {"cid": eid},
+            )
+            row = result.mappings().first()
+            if not row:
+                return
+            await index_entity("contact", eid, row["tenant_id"], db)
+        except Exception:
+            logger.exception("Failed to index contact %s", contact_id)
+
+
+async def index_company(ctx: dict[str, Any], company_id: str) -> None:
+    """Index a company: generate and store embedding."""
+    from sqlalchemy import text
+    from app.plugins.builtins.unified_search.embedding import index_entity
+
+    factory = get_session_factory()
+    async with factory() as db:
+        try:
+            eid = _parse_id(company_id)
+            result = await db.execute(
+                text("SELECT tenant_id FROM companies WHERE id = :cid"),
+                {"cid": eid},
+            )
+            row = result.mappings().first()
+            if not row:
+                return
+            await index_entity("company", eid, row["tenant_id"], db)
+        except Exception:
+            logger.exception("Failed to index company %s", company_id)
+
+
+async def index_event(ctx: dict[str, Any], event_id: str) -> None:
+    """Index a calendar event: generate and store embedding."""
+    from sqlalchemy import text
+    from app.plugins.builtins.unified_search.embedding import index_entity
+
+    factory = get_session_factory()
+    async with factory() as db:
+        try:
+            eid = _parse_id(event_id)
+            result = await db.execute(
+                text("SELECT tenant_id FROM calendar_entries WHERE id = :eid"),
+                {"eid": eid},
+            )
+            row = result.mappings().first()
+            if not row:
+                return
+            await index_entity("event", eid, row["tenant_id"], db)
+        except Exception:
+            logger.exception("Failed to index event %s", event_id)
+
+
+async def reindex(ctx: dict[str, Any], entity_type: str) -> None:
+    """Reindex all entities of a given type with pagination."""
+    from sqlalchemy import text
+    from app.plugins.builtins.unified_search.embedding import index_entity
+
+    table_map = {
+        "contact": "contacts",
+        "company": "companies",
+        "mail": "mails",
+        "file": "files",
+        "event": "calendar_entries",
+    }
+    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:
+        offset = 0
+        while True:
+            result = await db.execute(
+                text(
+                    f"SELECT id, tenant_id FROM {table} "
+                    f"WHERE deleted_at IS NULL ORDER BY created_at LIMIT :lim OFFSET :off"
+                ),
+                {"lim": BATCH_SIZE, "off": offset},
+            )
+            rows = result.mappings().all()
+            if not rows:
+                break
+            for row in rows:
+                try:
+                    await index_entity(
+                        entity_type,
+                        row["id"],
+                        row["tenant_id"],
+                        db,
+                    )
+                except Exception:
+                    logger.exception("Reindex failed for %s/%s", entity_type, row["id"])
+            offset += BATCH_SIZE
+
+    logger.info("Reindex complete for %s", entity_type)
+
+
+async def embedding_batch(ctx: dict[str, Any]) -> None:
+    """Periodic job: find entities without embeddings and index them."""
+    from sqlalchemy import text
+    from app.plugins.builtins.unified_search.embedding import index_entity
+
+    table_map = {
+        "contact": "contacts",
+        "company": "companies",
+        "mail": "mails",
+        "file": "files",
+        "event": "calendar_entries",
+    }
+
+    factory = get_session_factory()
+    async with factory() as db:
+        for etype, table in table_map.items():
+            try:
+                result = await db.execute(
+                    text(
+                        f"SELECT id, tenant_id FROM {table} "
+                        f"WHERE deleted_at IS NULL AND embedding IS NULL "
+                        f"LIMIT :lim"
+                    ),
+                    {"lim": BATCH_SIZE},
+                )
+                rows = result.mappings().all()
+                for row in rows:
+                    try:
+                        await index_entity(etype, row["id"], row["tenant_id"], db)
+                    except Exception:
+                        logger.exception("Batch index failed for %s/%s", etype, row["id"])
+            except Exception:
+                logger.exception("Batch query failed for %s", etype)
+
+    logger.info("Embedding batch job complete")
@@ -0,0 +1,154 @@
+-- Unified Search: pgvector extension, FTS triggers, provider registry
+
+-- pgvector extension
+CREATE EXTENSION IF NOT EXISTS vector;
+
+-- ─── Provider Registry Table ───
+CREATE TABLE IF NOT EXISTS unified_search_providers (
+    id          UUID PRIMARY KEY DEFAULT gen_random_uuid(),
+    tenant_id   UUID NOT NULL,
+    entity_type VARCHAR(50) NOT NULL,
+    plugin_name VARCHAR(80) NOT NULL,
+    is_active   BOOLEAN NOT NULL DEFAULT TRUE,
+    config      JSONB NOT NULL DEFAULT '{}'::jsonb,
+    created_at  TIMESTAMPTZ NOT NULL DEFAULT now(),
+    updated_at  TIMESTAMPTZ NOT NULL DEFAULT now(),
+    UNIQUE(tenant_id, entity_type)
+);
+
+CREATE INDEX IF NOT EXISTS ix_usp_tenant ON unified_search_providers(tenant_id);
+
+-- ─── Index Log Table ───
+CREATE TABLE IF NOT EXISTS unified_search_index_log (
+    id           UUID PRIMARY KEY DEFAULT gen_random_uuid(),
+    tenant_id    UUID NOT NULL,
+    entity_type  VARCHAR(50) NOT NULL,
+    entity_id    UUID NOT NULL,
+    action       VARCHAR(20) NOT NULL,
+    status       VARCHAR(20) NOT NULL DEFAULT 'pending',
+    error_message TEXT,
+    created_at   TIMESTAMPTZ NOT NULL DEFAULT now()
+);
+
+CREATE INDEX IF NOT EXISTS ix_usil_tenant ON unified_search_index_log(tenant_id);
+CREATE INDEX IF NOT EXISTS ix_usil_entity ON unified_search_index_log(entity_type, entity_id);
+
+-- ─── FTS: Mails ───
+ALTER TABLE mails ADD COLUMN IF NOT EXISTS body_tsv tsvector;
+
+CREATE OR REPLACE FUNCTION mails_tsv_trigger() RETURNS trigger AS $$
+BEGIN
+    NEW.body_tsv :=
+        setweight(to_tsvector('pg_catalog.german', coalesce(NEW.subject, '')), 'A') ||
+        setweight(to_tsvector('pg_catalog.german', coalesce(NEW.from_address, '')), 'B') ||
+        setweight(to_tsvector('pg_catalog.german', coalesce(NEW.to_addresses, '')), 'C') ||
+        setweight(to_tsvector('pg_catalog.german', coalesce(NEW.body_text, '')), 'D');
+    RETURN NEW;
+END;
+$$ LANGUAGE plpgsql;
+
+DROP TRIGGER IF EXISTS mails_tsv_update ON mails;
+CREATE TRIGGER mails_tsv_update
+    BEFORE INSERT OR UPDATE ON mails
+    FOR EACH ROW EXECUTE FUNCTION mails_tsv_trigger();
+
+CREATE INDEX IF NOT EXISTS ix_mails_body_tsv ON mails USING gin(body_tsv);
+
+-- ─── FTS: Contacts ───
+ALTER TABLE contacts ADD COLUMN IF NOT EXISTS search_tsv tsvector;
+
+CREATE OR REPLACE FUNCTION contacts_tsv_trigger() RETURNS trigger AS $$
+BEGIN
+    NEW.search_tsv :=
+        setweight(to_tsvector('pg_catalog.german', coalesce(NEW.first_name, '') || ' ' || coalesce(NEW.last_name, '')), 'A') ||
+        setweight(to_tsvector('pg_catalog.german', coalesce(NEW.email, '')), 'B') ||
+        setweight(to_tsvector('pg_catalog.german', coalesce(NEW.phone, '') || ' ' || coalesce(NEW.mobile, '')), 'C') ||
+        setweight(to_tsvector('pg_catalog.german', coalesce(NEW.notes, '')), 'D');
+    RETURN NEW;
+END;
+$$ LANGUAGE plpgsql;
+
+DROP TRIGGER IF EXISTS contacts_tsv_update ON contacts;
+CREATE TRIGGER contacts_tsv_update
+    BEFORE INSERT OR UPDATE ON contacts
+    FOR EACH ROW EXECUTE FUNCTION contacts_tsv_trigger();
+
+CREATE INDEX IF NOT EXISTS ix_contacts_search_tsv ON contacts USING gin(search_tsv);
+
+-- ─── FTS: Calendar Entries ───
+ALTER TABLE calendar_entries ADD COLUMN IF NOT EXISTS search_tsv tsvector;
+
+CREATE OR REPLACE FUNCTION calendar_entries_tsv_trigger() RETURNS trigger AS $$
+BEGIN
+    NEW.search_tsv :=
+        setweight(to_tsvector('pg_catalog.german', coalesce(NEW.title, '')), 'A') ||
+        setweight(to_tsvector('pg_catalog.german', coalesce(NEW.description, '')), 'B') ||
+        setweight(to_tsvector('pg_catalog.german', coalesce(NEW.location, '')), 'C');
+    RETURN NEW;
+END;
+$$ LANGUAGE plpgsql;
+
+DROP TRIGGER IF EXISTS calendar_entries_tsv_update ON calendar_entries;
+CREATE TRIGGER calendar_entries_tsv_update
+    BEFORE INSERT OR UPDATE ON calendar_entries
+    FOR EACH ROW EXECUTE FUNCTION calendar_entries_tsv_trigger();
+
+CREATE INDEX IF NOT EXISTS ix_cal_entries_search_tsv ON calendar_entries USING gin(search_tsv);
+
+-- ─── FTS: Files (content_text added here, not in 0002, to avoid trigger dependency) ───
+ALTER TABLE files ADD COLUMN IF NOT EXISTS content_text text;
+ALTER TABLE files ADD COLUMN IF NOT EXISTS content_tsv tsvector;
+
+CREATE OR REPLACE FUNCTION files_tsv_trigger() RETURNS trigger AS $$
+BEGIN
+    NEW.content_tsv :=
+        setweight(to_tsvector('pg_catalog.german', coalesce(NEW.name, '')), 'A') ||
+        setweight(to_tsvector('pg_catalog.german', coalesce(NEW.content_text, '')), 'D');
+    RETURN NEW;
+END;
+$$ LANGUAGE plpgsql;
+
+DROP TRIGGER IF EXISTS files_tsv_update ON files;
+CREATE TRIGGER files_tsv_update
+    BEFORE INSERT OR UPDATE ON files
+    FOR EACH ROW EXECUTE FUNCTION files_tsv_trigger();
+
+CREATE INDEX IF NOT EXISTS ix_files_content_tsv ON files USING gin(content_tsv);
+
+-- ─── FTS: Tags ───
+ALTER TABLE tags ADD COLUMN IF NOT EXISTS search_tsv tsvector;
+
+CREATE OR REPLACE FUNCTION tags_tsv_trigger() RETURNS trigger AS $$
+BEGIN
+    NEW.search_tsv := to_tsvector('pg_catalog.german', coalesce(NEW.name, ''));
+    RETURN NEW;
+END;
+$$ LANGUAGE plpgsql;
+
+DROP TRIGGER IF EXISTS tags_tsv_update ON tags;
+CREATE TRIGGER tags_tsv_update
+    BEFORE INSERT OR UPDATE ON tags
+    FOR EACH ROW EXECUTE FUNCTION tags_tsv_trigger();
+
+CREATE INDEX IF NOT EXISTS ix_tags_search_tsv ON tags USING gin(search_tsv);
+
+-- ─── FTS: Audit Log ───
+ALTER TABLE audit_log ADD COLUMN IF NOT EXISTS search_tsv tsvector;
+
+CREATE OR REPLACE FUNCTION audit_log_tsv_trigger() RETURNS trigger AS $$
+BEGIN
+    NEW.search_tsv :=
+        to_tsvector('pg_catalog.german',
+            coalesce(NEW.entity_type, '') || ' ' ||
+            coalesce(NEW.action, '') || ' ' ||
+            coalesce(NEW.user_id::text, ''));
+    RETURN NEW;
+END;
+$$ LANGUAGE plpgsql;
+
+DROP TRIGGER IF EXISTS audit_log_tsv_update ON audit_log;
+CREATE TRIGGER audit_log_tsv_update
+    BEFORE INSERT OR UPDATE ON audit_log
+    FOR EACH ROW EXECUTE FUNCTION audit_log_tsv_trigger();
+
+CREATE INDEX IF NOT EXISTS ix_audit_log_search_tsv ON audit_log USING gin(search_tsv);
@@ -0,0 +1,24 @@
+-- Unified Search: Embedding columns and HNSW indexes
+
+-- ─── Mails ───
+ALTER TABLE mails ADD COLUMN IF NOT EXISTS embedding vector(768);
+CREATE INDEX IF NOT EXISTS ix_mails_embedding ON mails USING hnsw(embedding vector_cosine_ops);
+
+-- ─── Contacts ───
+ALTER TABLE contacts ADD COLUMN IF NOT EXISTS embedding vector(768);
+CREATE INDEX IF NOT EXISTS ix_contacts_embedding ON contacts USING hnsw(embedding vector_cosine_ops);
+
+-- ─── Companies ───
+ALTER TABLE companies ADD COLUMN IF NOT EXISTS embedding vector(768);
+CREATE INDEX IF NOT EXISTS ix_companies_embedding ON companies USING hnsw(embedding vector_cosine_ops);
+
+-- ─── Files (content_text already added in 0001) ───
+ALTER TABLE files ADD COLUMN IF NOT EXISTS embedding vector(768);
+CREATE INDEX IF NOT EXISTS ix_files_embedding ON files USING hnsw(embedding vector_cosine_ops);
+
+-- ─── Calendar Entries ───
+ALTER TABLE calendar_entries ADD COLUMN IF NOT EXISTS embedding vector(768);
+CREATE INDEX IF NOT EXISTS ix_cal_embedding ON calendar_entries USING hnsw(embedding vector_cosine_ops);
+
+-- ─── Tags (shorter dimension for short text) ───
+ALTER TABLE tags ADD COLUMN IF NOT EXISTS embedding vector(384);
@@ -0,0 +1,50 @@
+"""SQLAlchemy models for the Unified Search plugin."""
+
+from __future__ import annotations
+
+import uuid
+from datetime import datetime
+
+from sqlalchemy import DateTime, ForeignKey, Index, String, Text
+from sqlalchemy.dialects.postgresql import JSONB, UUID as PGUUID
+from sqlalchemy.orm import Mapped, mapped_column
+
+from app.core.db import Base, TenantMixin
+
+
+class SearchProviderRegistry(Base, TenantMixin):
+    """Registry of active search providers per tenant."""
+
+    __tablename__ = "unified_search_providers"
+    __table_args__ = (
+        Index("ix_usp_tenant", "tenant_id"),
+    )
+
+    id: Mapped[uuid.UUID] = mapped_column(
+        PGUUID(as_uuid=True), primary_key=True, default=uuid.uuid4
+    )
+    entity_type: Mapped[str] = mapped_column(String(50), nullable=False)
+    plugin_name: Mapped[str] = mapped_column(String(80), nullable=False)
+    is_active: Mapped[bool] = mapped_column(
+        nullable=False, default=True
+    )
+    config: Mapped[dict] = mapped_column(JSONB, nullable=False, default=dict)
+
+
+class SearchIndexLog(Base, TenantMixin):
+    """Log of indexing actions for audit and debugging."""
+
+    __tablename__ = "unified_search_index_log"
+    __table_args__ = (
+        Index("ix_usil_tenant", "tenant_id"),
+        Index("ix_usil_entity", "entity_type", "entity_id"),
+    )
+
+    id: Mapped[uuid.UUID] = mapped_column(
+        PGUUID(as_uuid=True), primary_key=True, default=uuid.uuid4
+    )
+    entity_type: Mapped[str] = mapped_column(String(50), nullable=False)
+    entity_id: Mapped[uuid.UUID] = mapped_column(PGUUID(as_uuid=True), nullable=False)
+    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)
@@ -0,0 +1,137 @@
+"""Unified Search plugin class and manifest."""
+
+from __future__ import annotations
+
+import logging
+from typing import Any
+
+from app.plugins.base import BasePlugin
+from app.plugins.manifest import PluginManifest, PluginRouteDef
+
+logger = logging.getLogger(__name__)
+
+
+class UnifiedSearchPlugin(BasePlugin):
+    """Hybrid full-text and semantic search across all CRM entities."""
+
+    manifest = PluginManifest(
+        name="unified_search",
+        version="1.0.0",
+        display_name="Unified Search",
+        description=(
+            "Hybrid full-text (PostgreSQL FTS) and semantic (pgvector) search "
+            "with KI query understanding and RRF rank fusion across all CRM data."
+        ),
+        dependencies=[],
+        routes=[
+            PluginRouteDef(
+                path="/api/v1/search",
+                module="app.plugins.builtins.unified_search.routes",
+                router_attr="router",
+            ),
+        ],
+        events=[
+            "mail.synced",
+            "file.uploaded",
+            "contact.created",
+            "contact.updated",
+            "company.created",
+            "company.updated",
+            "calendar.entry.created",
+        ],
+        migrations=["0001_initial.sql", "0002_embeddings.sql"],
+        permissions=["search:read", "search:admin"],
+        is_core=False,
+    )
+
+    async def on_activate(self, db, service_container, event_bus) -> None:
+        """Register search providers on activation."""
+        await super().on_activate(db, service_container, event_bus)
+        try:
+            from app.plugins.builtins.unified_search.provider_registry import (
+                auto_register_providers,
+            )
+            await auto_register_providers(db)
+            logger.info("Unified Search providers auto-registered")
+        except Exception:
+            logger.exception("Failed to auto-register search providers")
+
+    async def on_deactivate(self, db, service_container, event_bus) -> None:
+        """Clear provider registry on deactivation."""
+        from app.plugins.builtins.unified_search.provider_registry import (
+            get_search_registry,
+        )
+        registry = get_search_registry()
+        registry.clear()
+        await super().on_deactivate(db, service_container, event_bus)
+
+    # ─── Event Handlers ───
+
+    async def on_mail_synced(self, payload: dict[str, Any]) -> None:
+        """Enqueue embedding jobs for synced mails."""
+        from app.core.jobs import enqueue_job
+
+        mail_ids = payload.get("mail_ids", [])
+        if mail_ids:
+            await enqueue_job("index_mails", mail_ids)
+
+    async def on_file_uploaded(self, payload: dict[str, Any]) -> None:
+        """Enqueue file indexing job."""
+        from app.core.jobs import enqueue_job
+
+        file_id = payload.get("file_id")
+        if file_id:
+            await enqueue_job("index_file", file_id)
+
+    async def on_contact_created(self, payload: dict[str, Any]) -> None:
+        from app.core.jobs import enqueue_job
+
+        contact_id = payload.get("contact_id")
+        if contact_id:
+            await enqueue_job("index_contact", contact_id)
+
+    async def on_contact_updated(self, payload: dict[str, Any]) -> None:
+        from app.core.jobs import enqueue_job
+
+        contact_id = payload.get("contact_id")
+        if contact_id:
+            await enqueue_job("index_contact", contact_id)
+
+    async def on_company_created(self, payload: dict[str, Any]) -> None:
+        from app.core.jobs import enqueue_job
+
+        company_id = payload.get("company_id")
+        if company_id:
+            await enqueue_job("index_company", company_id)
+
+    async def on_company_updated(self, payload: dict[str, Any]) -> None:
+        from app.core.jobs import enqueue_job
+
+        company_id = payload.get("company_id")
+        if company_id:
+            await enqueue_job("index_company", company_id)
+
+    async def on_calendar_entry_created(self, payload: dict[str, Any]) -> None:
+        from app.core.jobs import enqueue_job
+
+        entry_id = payload.get("entry_id")
+        if entry_id:
+            await enqueue_job("index_event", entry_id)
+
+    def get_notification_types(self) -> list[dict[str, Any]]:
+        return [
+            {
+                "type_key": "search_error",
+                "category": "search",
+                "label": "Suchfehler",
+                "description": "Fehler bei der Suchausführung",
+                "is_enabled_by_default": True,
+            },
+            {
+                "type_key": "search_reindex_complete",
+                "category": "search",
+                "label": "Reindex abgeschlossen",
+                "description": "Neuindizierung abgeschlossen",
+                "is_enabled_by_default": False,
+            },
+        ]
@@ -0,0 +1,133 @@
+"""SearchProvider protocol and global registry."""
+
+from __future__ import annotations
+
+import logging
+from typing import TYPE_CHECKING, Protocol, runtime_checkable
+
+from sqlalchemy import select
+
+if TYPE_CHECKING:
+    import uuid
+    from sqlalchemy.ext.asyncio import AsyncSession
+
+logger = logging.getLogger(__name__)
+
+
+@runtime_checkable
+class SearchProvider(Protocol):
+    """Protocol for entity-specific search providers."""
+
+    entity_type: str
+
+    async def search_fts(
+        self,
+        db: AsyncSession,
+        tsquery: str,
+        tenant_id: uuid.UUID,
+        limit: int,
+    ) -> list[dict]:
+        """Full-text search using PostgreSQL tsquery."""
+        ...
+
+    async def search_vector(
+        self,
+        db: AsyncSession,
+        embedding: list[float],
+        tenant_id: uuid.UUID,
+        limit: int,
+    ) -> list[dict]:
+        """Semantic vector search using pgvector."""
+        ...
+
+    async def get_embedding_text(
+        self,
+        db: AsyncSession,
+        entity_id: uuid.UUID,
+        tenant_id: uuid.UUID,
+    ) -> str:
+        """Get text representation for embedding generation."""
+        ...
+
+    def to_search_result(self, entity: object) -> dict:
+        """Convert an ORM entity to a search result dict."""
+        ...
+
+
+class SearchProviderRegistry:
+    """In-memory registry of search providers."""
+
+    def __init__(self) -> None:
+        self._providers: dict[str, SearchProvider] = {}
+
+    def register(self, provider: SearchProvider) -> None:
+        """Register a search provider."""
+        self._providers[provider.entity_type] = provider
+        logger.debug("Registered search provider: %s", provider.entity_type)
+
+    def unregister(self, entity_type: str) -> None:
+        """Unregister a search provider by entity type."""
+        self._providers.pop(entity_type, None)
+
+    def get(self, entity_type: str) -> SearchProvider | None:
+        """Get a provider by entity type."""
+        return self._providers.get(entity_type)
+
+    def get_all(self) -> list[SearchProvider]:
+        """Get all registered providers."""
+        return list(self._providers.values())
+
+    def get_entity_types(self) -> list[str]:
+        """Get all registered entity types."""
+        return list(self._providers.keys())
+
+    def clear(self) -> None:
+        """Clear all registered providers."""
+        self._providers.clear()
+
+
+# Global singleton
+_registry = SearchProviderRegistry()
+
+
+def get_search_registry() -> SearchProviderRegistry:
+    """Get the global search provider registry."""
+    return _registry
+
+
+async def auto_register_providers(db: AsyncSession) -> None:
+    """Auto-register providers for active plugins.
+
+    Checks which plugins are active and registers corresponding providers.
+    """
+    from app.plugins.builtins.unified_search.providers.contact_provider import (
+        ContactSearchProvider,
+    )
+    from app.plugins.builtins.unified_search.providers.company_provider import (
+        CompanySearchProvider,
+    )
+    from app.plugins.builtins.unified_search.providers.mail_provider import (
+        MailSearchProvider,
+    )
+    from app.plugins.builtins.unified_search.providers.file_provider import (
+        FileSearchProvider,
+    )
+    from app.plugins.builtins.unified_search.providers.event_provider import (
+        EventSearchProvider,
+    )
+
+    registry = get_search_registry()
+    registry.clear()
+
+    # Register all built-in providers
+    for provider_cls in [
+        ContactSearchProvider,
+        CompanySearchProvider,
+        MailSearchProvider,
+        FileSearchProvider,
+        EventSearchProvider,
+    ]:
+        try:
+            registry.register(provider_cls())
+        except Exception:
+            logger.exception("Failed to register %s", provider_cls.__name__)
@@ -0,0 +1,3 @@
+"""Search providers for unified search."""
+
+from __future__ import annotations
@@ -0,0 +1,114 @@
+"""Company search provider."""
+
+from __future__ import annotations
+
+import logging
+import uuid
+from typing import Any
+
+from sqlalchemy import text
+from sqlalchemy.ext.asyncio import AsyncSession
+
+logger = logging.getLogger(__name__)
+
+
+class CompanySearchProvider:
+    """Search provider for Company entities."""
+
+    entity_type = "company"
+
+    async def search_fts(
+        self,
+        db: AsyncSession,
+        tsquery: str,
+        tenant_id: uuid.UUID,
+        limit: int,
+    ) -> list[dict[str, Any]]:
+        """Full-text search on companies.search_tsv."""
+        sql = text(
+            """
+            SELECT c.*, ts_rank(c.search_tsv, to_tsquery('pg_catalog.german', :q)) AS rank
+            FROM companies c
+            WHERE c.tenant_id = :tid
+              AND c.deleted_at IS NULL
+              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(
+        self,
+        db: AsyncSession,
+        embedding: list[float],
+        tenant_id: uuid.UUID,
+        limit: int,
+    ) -> list[dict[str, Any]]:
+        """Semantic search on companies.embedding."""
+        sql = text(
+            """
+            SELECT c.*, 1 - (c.embedding <=> cast(:emb AS vector)) AS score
+            FROM companies c
+            WHERE c.tenant_id = :tid
+              AND c.deleted_at IS NULL
+              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,
+    ) -> str:
+        """Get text for embedding generation."""
+        sql = text(
+            """
+            SELECT name, description, industry
+            FROM companies
+            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", ""),
+            row.get("industry", ""),
+        ]
+        return " ".join(str(p) for p in parts if p)
+
+    def to_search_result(self, entity: object) -> dict[str, Any]:
+        """Convert company to search result dict."""
+        if isinstance(entity, dict):
+            name = entity.get("name", "")
+            description = entity.get("description", "") or ""
+            entity_id = str(entity.get("id", ""))
+        else:
+            name = getattr(entity, "name", "")
+            description = getattr(entity, "description", "") or ""
+            entity_id = str(getattr(entity, "id", ""))
+        return {
+            "entity_type": self.entity_type,
+            "entity_id": entity_id,
+            "title": name,
+            "snippet": description[:200],
+            "score": 0.0,
+            "data": {},
+        }
@@ -0,0 +1,121 @@
+"""Contact search provider."""
+
+from __future__ import annotations
+
+import logging
+import uuid
+from typing import Any
+
+from sqlalchemy import text
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.models.contact import Contact
+
+logger = logging.getLogger(__name__)
+
+
+class ContactSearchProvider:
+    """Search provider for Contact entities."""
+
+    entity_type = "contact"
+
+    async def search_fts(
+        self,
+        db: AsyncSession,
+        tsquery: str,
+        tenant_id: uuid.UUID,
+        limit: int,
+    ) -> list[dict[str, Any]]:
+        """Full-text search on contacts.search_tsv."""
+        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.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(
+        self,
+        db: AsyncSession,
+        embedding: list[float],
+        tenant_id: uuid.UUID,
+        limit: int,
+    ) -> list[dict[str, Any]]:
+        """Semantic search on contacts.embedding."""
+        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.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,
+    ) -> str:
+        """Get text for embedding generation."""
+        sql = text(
+            """
+            SELECT first_name, last_name, email, phone, mobile, notes
+            FROM contacts
+            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("first_name", ""),
+            row.get("last_name", ""),
+            row.get("email", ""),
+            row.get("phone", ""),
+            row.get("mobile", ""),
+            row.get("notes", ""),
+        ]
+        return " ".join(str(p) for p in parts if p)
+
+    def to_search_result(self, entity: object) -> dict[str, Any]:
+        """Convert contact to search result dict."""
+        if isinstance(entity, dict):
+            first = entity.get("first_name", "")
+            last = entity.get("last_name", "")
+            email = entity.get("email", "")
+            entity_id = str(entity.get("id", ""))
+        else:
+            first = getattr(entity, "first_name", "")
+            last = getattr(entity, "last_name", "")
+            email = getattr(entity, "email", "")
+            entity_id = str(getattr(entity, "id", ""))
+        return {
+            "entity_type": self.entity_type,
+            "entity_id": entity_id,
+            "title": f"{first} {last}".strip(),
+            "snippet": email or "",
+            "score": 0.0,
+            "data": {},
+        }
@@ -0,0 +1,114 @@
+"""Calendar event search provider."""
+
+from __future__ import annotations
+
+import logging
+import uuid
+from typing import Any
+
+from sqlalchemy import text
+from sqlalchemy.ext.asyncio import AsyncSession
+
+logger = logging.getLogger(__name__)
+
+
+class EventSearchProvider:
+    """Search provider for CalendarEntry entities."""
+
+    entity_type = "event"
+
+    async def search_fts(
+        self,
+        db: AsyncSession,
+        tsquery: str,
+        tenant_id: uuid.UUID,
+        limit: int,
+    ) -> 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},
+        )
+        rows = result.mappings().all()
+        return [dict(r) for r in rows]
+
+    async def search_vector(
+        self,
+        db: AsyncSession,
+        embedding: list[float],
+        tenant_id: uuid.UUID,
+        limit: int,
+    ) -> 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},
+        )
+        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 title, description, location
+            FROM calendar_entries
+            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("title", ""),
+            row.get("description", ""),
+            row.get("location", ""),
+        ]
+        return " ".join(str(p) for p in parts if p)
+
+    def to_search_result(self, entity: object) -> dict[str, Any]:
+        """Convert calendar entry to search result dict."""
+        if isinstance(entity, dict):
+            title = entity.get("title", "")
+            description = entity.get("description", "") or ""
+            entity_id = str(entity.get("id", ""))
+        else:
+            title = getattr(entity, "title", "")
+            description = getattr(entity, "description", "") or ""
+            entity_id = str(getattr(entity, "id", ""))
+        return {
+            "entity_type": self.entity_type,
+            "entity_id": entity_id,
+            "title": title,
+            "snippet": description[:200],
+            "score": 0.0,
+            "data": {},
+        }
@@ -0,0 +1,111 @@
+"""DMS File search provider."""
+
+from __future__ import annotations
+
+import logging
+import uuid
+from typing import Any
+
+from sqlalchemy import text
+from sqlalchemy.ext.asyncio import AsyncSession
+
+logger = logging.getLogger(__name__)
+
+
+class FileSearchProvider:
+    """Search provider for DMS File entities."""
+
+    entity_type = "file"
+
+    async def search_fts(
+        self,
+        db: AsyncSession,
+        tsquery: str,
+        tenant_id: uuid.UUID,
+        limit: int,
+    ) -> 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},
+        )
+        rows = result.mappings().all()
+        return [dict(r) for r in rows]
+
+    async def search_vector(
+        self,
+        db: AsyncSession,
+        embedding: list[float],
+        tenant_id: uuid.UUID,
+        limit: int,
+    ) -> 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},
+        )
+        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 name, content_text
+            FROM files
+            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 ""
+        name = row.get("name", "") or ""
+        content = row.get("content_text", "") or ""
+        return f"{name} {content[:5000]}"
+
+    def to_search_result(self, entity: object) -> dict[str, Any]:
+        """Convert file to search result dict."""
+        if isinstance(entity, dict):
+            name = entity.get("name", "")
+            content = entity.get("content_text", "") or ""
+            entity_id = str(entity.get("id", ""))
+        else:
+            name = getattr(entity, "name", "")
+            content = getattr(entity, "content_text", "") or ""
+            entity_id = str(getattr(entity, "id", ""))
+        return {
+            "entity_type": self.entity_type,
+            "entity_id": entity_id,
+            "title": name,
+            "snippet": content[:200],
+            "score": 0.0,
+            "data": {},
+        }
@@ -0,0 +1,111 @@
+"""Mail search provider."""
+
+from __future__ import annotations
+
+import logging
+import uuid
+from typing import Any
+
+from sqlalchemy import text
+from sqlalchemy.ext.asyncio import AsyncSession
+
+logger = logging.getLogger(__name__)
+
+
+class MailSearchProvider:
+    """Search provider for Mail entities."""
+
+    entity_type = "mail"
+
+    async def search_fts(
+        self,
+        db: AsyncSession,
+        tsquery: str,
+        tenant_id: uuid.UUID,
+        limit: int,
+    ) -> 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},
+        )
+        rows = result.mappings().all()
+        return [dict(r) for r in rows]
+
+    async def search_vector(
+        self,
+        db: AsyncSession,
+        embedding: list[float],
+        tenant_id: uuid.UUID,
+        limit: int,
+    ) -> 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},
+        )
+        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 subject, body_text
+            FROM mails
+            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 ""
+        subject = row.get("subject", "") or ""
+        body = row.get("body_text", "") or ""
+        return f"{subject} {body[:5000]}"
+
+    def to_search_result(self, entity: object) -> dict[str, Any]:
+        """Convert mail to search result dict."""
+        if isinstance(entity, dict):
+            subject = entity.get("subject", "")
+            body = entity.get("body_text", "") or ""
+            entity_id = str(entity.get("id", ""))
+        else:
+            subject = getattr(entity, "subject", "")
+            body = getattr(entity, "body_text", "") or ""
+            entity_id = str(getattr(entity, "id", ""))
+        return {
+            "entity_type": self.entity_type,
+            "entity_id": entity_id,
+            "title": subject,
+            "snippet": body[:200],
+            "score": 0.0,
+            "data": {},
+        }
@@ -0,0 +1,101 @@
+"""KI query understanding and result aggregation via LiteLLM."""
+
+from __future__ import annotations
+
+import os
+import json
+import logging
+from typing import Any
+
+import litellm
+
+logger = logging.getLogger(__name__)
+
+DEFAULT_LLM_MODEL = os.environ.get('SEARCH_LLM_MODEL', 'ollama/deepseek-v4')
+OLLAMA_API_KEY = os.environ.get('API_KEY_OLLAMA_CLOUD', '')
+
+QUERY_ANALYZE_SYSTEM = (
+    "Du bist ein Query-Analyzer fuer ein CRM. "
+    "Analysiere die Suchanfrage und gib JSON zurueck: "
+    '{"normalized_query": str, "entities": {"person": str|null, "company": str|null, "topic": str|null}, '
+    '"intent": str, "semantic_terms": [str], "suggested_filters": {}}'
+)
+
+RESULT_AGGREGATE_SYSTEM = (
+    "Du bist ein Result-Aggregator. Fasse Ergebnisse zusammen und generiere Facetten: "
+    '{"summary": str, "facets": {"types": {}, "dates": {}, "people": []}, "suggestions": [str]}'
+)
+
+
+def _fallback_query_analysis(query: str) -> dict[str, Any]:
+    return {
+        "normalized_query": query,
+        "entities": {},
+        "intent": "search",
+        "semantic_terms": [],
+        "suggested_filters": {},
+    }
+
+
+def _fallback_aggregate(results: list[dict], query: str) -> dict[str, Any]:
+    return {
+        "summary": f"{len(results)} Ergebnisse gefunden",
+        "facets": {},
+        "suggestions": [],
+    }
+
+
+async def llm_analyze_query(query: str) -> dict[str, Any]:
+    """Analyze a search query using LLM for intent, entities, and semantic terms.
+
+    Falls back to a simple dict if LLM fails.
+    """
+    try:
+        response = await litellm.acompletion(
+            model=DEFAULT_LLM_MODEL,
+            messages=[
+                {"role": "system", "content": QUERY_ANALYZE_SYSTEM},
+                {"role": "user", "content": query},
+            ],
+            temperature=0.1,
+            max_tokens=500,
+            response_format={"type": "json_object"},
+            api_key=OLLAMA_API_KEY,
+        )
+        content = response.choices[0].message.content
+        return json.loads(content)
+    except Exception:
+        logger.warning("LLM query analysis failed, using fallback")
+        return _fallback_query_analysis(query)
+
+
+async def llm_aggregate_results(results: list[dict], query: str) -> dict[str, Any]:
+    """Aggregate search results using LLM for summary, facets, and suggestions.
+
+    Falls back to a simple dict if LLM fails.
+    """
+    if not results:
+        return _fallback_aggregate(results, query)
+    try:
+        # Truncate results to avoid token overflow
+        compact = [
+            {"entity_type": r.get("entity_type"), "title": r.get("title", "")[:100]}
+            for r in results[:50]
+        ]
+        user_msg = json.dumps({"query": query, "results": compact})
+        response = await litellm.acompletion(
+            model=DEFAULT_LLM_MODEL,
+            messages=[
+                {"role": "system", "content": RESULT_AGGREGATE_SYSTEM},
+                {"role": "user", "content": user_msg},
+            ],
+            temperature=0.1,
+            max_tokens=1000,
+            response_format={"type": "json_object"},
+            api_key=OLLAMA_API_KEY,
+        )
+        content = response.choices[0].message.content
+        return json.loads(content)
+    except Exception:
+        logger.warning("LLM result aggregation failed, using fallback")
+        return _fallback_aggregate(results, query)
@@ -0,0 +1,316 @@
+"""API routes for the Unified Search plugin."""
+
+from __future__ import annotations
+
+import logging
+import uuid
+
+from fastapi import APIRouter, Depends, HTTPException, Query
+from sqlalchemy import text
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.core.db import get_db
+from app.core.jobs import enqueue_job
+from app.deps import get_current_user, require_permission
+from app.plugins.builtins.unified_search.provider_registry import get_search_registry
+from app.plugins.builtins.unified_search.query_understanding import (
+    llm_aggregate_results,
+    llm_analyze_query,
+)
+from app.plugins.builtins.unified_search.schemas import (
+    ProviderResponse,
+    ReindexRequest,
+    SearchRequest,
+    SearchResponse,
+    SearchResult,
+    SimilarRequest,
+    SimilarResponse,
+    SuggestRequest,
+    SuggestResponse,
+)
+from app.plugins.builtins.unified_search.search_engine import (
+    autocomplete,
+    find_similar_all_types,
+    hybrid_search,
+)
+
+logger = logging.getLogger(__name__)
+
+router = APIRouter(prefix="/api/v1/search", tags=["search"])
+
+
+# ─── Search ───
+
+@router.post("", dependencies=[Depends(require_permission("search:read"))])
+async def search(
+    req: SearchRequest,
+    current_user: dict = Depends(get_current_user),
+    db: AsyncSession = Depends(get_db),
+) -> SearchResponse:
+    """Perform hybrid search with KI query understanding."""
+    tenant_id = uuid.UUID(current_user["tenant_id"])
+
+    # KI query understanding
+    query_analysis = await llm_analyze_query(req.query)
+
+    # Hybrid search
+    results = await hybrid_search(
+        db=db,
+        query_analysis=query_analysis,
+        tenant_id=tenant_id,
+        entity_types=req.entity_types,
+        limit=req.limit,
+    )
+
+    # KI result aggregation
+    aggregation = await llm_aggregate_results(results, req.query)
+
+    search_results = [
+        SearchResult(
+            entity_type=r.get("entity_type", ""),
+            entity_id=r.get("entity_id", ""),
+            title=r.get("title", ""),
+            snippet=r.get("snippet", ""),
+            score=r.get("score", 0.0),
+            data=r.get("data", {}),
+        )
+        for r in results
+    ]
+
+    return SearchResponse(
+        query=req.query,
+        normalized_query=query_analysis.get("normalized_query", req.query),
+        results=search_results,
+        facets=aggregation.get("facets", {}),
+        summary=aggregation.get("summary", f"{len(results)} Ergebnisse"),
+        suggestions=aggregation.get("suggestions", []),
+    )
+
+
+# ─── Suggest / Autocomplete ───
+
+@router.get("/suggest", dependencies=[Depends(require_permission("search:read"))])
+async def suggest(
+    q: str = Query(..., min_length=1, max_length=200),
+    limit: int = Query(default=10, ge=1, le=50),
+    current_user: dict = Depends(get_current_user),
+    db: AsyncSession = Depends(get_db),
+) -> SuggestResponse:
+    """Autocomplete suggestions using FTS prefix search."""
+    tenant_id = uuid.UUID(current_user["tenant_id"])
+    suggestions = await autocomplete(db, q, tenant_id, limit)
+    return SuggestResponse(suggestions=suggestions)
+
+
+# ─── Similar ───
+
+@router.post("/similar", dependencies=[Depends(require_permission("search:read"))])
+async def find_similar(
+    req: SimilarRequest,
+    current_user: dict = Depends(get_current_user),
+    db: AsyncSession = Depends(get_db),
+) -> SimilarResponse:
+    """Find similar entities across all types based on embedding."""
+    tenant_id = uuid.UUID(current_user["tenant_id"])
+    try:
+        entity_id = uuid.UUID(req.entity_id)
+    except ValueError:
+        raise HTTPException(status_code=400, detail="Invalid entity_id")
+
+    similar = await find_similar_all_types(
+        db=db,
+        entity_type=req.entity_type,
+        entity_id=entity_id,
+        tenant_id=tenant_id,
+        limit=req.limit,
+    )
+
+    result_dict: dict[str, list[SearchResult]] = {}
+    for etype, items in similar.items():
+        result_dict[etype] = [
+            SearchResult(
+                entity_type=r.get("entity_type", etype),
+                entity_id=r.get("entity_id", ""),
+                title=r.get("title", ""),
+                snippet=r.get("snippet", ""),
+                score=r.get("score", 0.0),
+                data=r.get("data", {}),
+            )
+            for r in items
+        ]
+
+    return SimilarResponse(similar=result_dict)
+
+
+# ─── Reindex ───
+
+@router.post("/reindex", dependencies=[Depends(require_permission("search:admin"))])
+async def reindex(
+    req: ReindexRequest,
+    current_user: dict = Depends(get_current_user),
+    db: AsyncSession = Depends(get_db),
+) -> dict:
+    """Trigger reindexing of specified entity types."""
+    entity_types = req.entity_types
+    if not entity_types:
+        entity_types = get_search_registry().get_entity_types()
+
+    job_ids: list[str] = []
+    for etype in entity_types:
+        job_id = await enqueue_job("reindex", etype)
+        if job_id:
+            job_ids.append(job_id)
+
+    return {
+        "status": "ok",
+        "message": f"Reindexing {len(entity_types)} entity types",
+        "entity_types": entity_types,
+        "job_ids": job_ids,
+    }
+
+
+# ─── Providers ───
+
+@router.get("/providers", dependencies=[Depends(require_permission("search:read"))])
+async def list_providers(
+    current_user: dict = Depends(get_current_user),
+    db: AsyncSession = Depends(get_db),
+) -> list[ProviderResponse]:
+    """List all active search providers."""
+    registry = get_search_registry()
+    providers = registry.get_all()
+    return [
+        ProviderResponse(
+            entity_type=p.entity_type,
+            plugin_name="unified_search",
+            is_active=True,
+        )
+        for p in providers
+    ]
+
+
+@router.post("/providers/{entity_type}/toggle", dependencies=[Depends(require_permission("search:admin"))])
+async def toggle_provider(
+    entity_type: str,
+    current_user: dict = Depends(get_current_user),
+    db: AsyncSession = Depends(get_db),
+) -> dict:
+    """Toggle a search provider on/off."""
+    registry = get_search_registry()
+    provider = registry.get(entity_type)
+    if provider is None:
+        raise HTTPException(status_code=404, detail=f"Provider not found: {entity_type}")
+
+    # Toggle in DB
+    tenant_id = uuid.UUID(current_user["tenant_id"])
+    result = await db.execute(
+        text(
+            """
+            SELECT is_active FROM unified_search_providers
+            WHERE tenant_id = :tid AND entity_type = :et
+            """
+        ),
+        {"tid": tenant_id, "et": entity_type},
+    )
+    row = result.mappings().first()
+
+    if row:
+        new_status = not row["is_active"]
+        await db.execute(
+            text(
+                """
+                UPDATE unified_search_providers SET is_active = :active
+                WHERE tenant_id = :tid AND entity_type = :et
+                """
+            ),
+            {"active": new_status, "tid": tenant_id, "et": entity_type},
+        )
+    else:
+        new_status = False
+        await db.execute(
+            text(
+                """
+                INSERT INTO unified_search_providers (tenant_id, entity_type, plugin_name, is_active)
+                VALUES (:tid, :et, 'unified_search', false)
+                """
+            ),
+            {"tid": tenant_id, "et": entity_type},
+        )
+
+    await db.commit()
+
+    if not new_status:
+        registry.unregister(entity_type)
+    else:
+        # Re-register by clearing and re-running auto_register
+        from app.plugins.builtins.unified_search.provider_registry import auto_register_providers
+        await auto_register_providers(db)
+
+    return {
+        "entity_type": entity_type,
+        "is_active": new_status,
+    }
+
+
+# ─── Stats ───
+
+@router.get("/stats", dependencies=[Depends(require_permission("search:read"))])
+async def search_stats(
+    current_user: dict = Depends(get_current_user),
+    db: AsyncSession = Depends(get_db),
+) -> dict:
+    """Get search index statistics."""
+    tenant_id = uuid.UUID(current_user["tenant_id"])
+
+    stats: dict = {}
+    tables = {
+        "contacts": "embedding",
+        "companies": "embedding",
+        "mails": "embedding",
+        "files": "embedding",
+        "calendar_entries": "embedding",
+    }
+
+    for table, col in tables.items():
+        try:
+            total_result = await db.execute(
+                text(f"SELECT count(*) AS cnt FROM {table} WHERE tenant_id = :tid AND deleted_at IS NULL"),
+                {"tid": tenant_id},
+            )
+            total = total_result.scalar() or 0
+
+            indexed_result = await db.execute(
+                text(f"SELECT count(*) AS cnt FROM {table} WHERE tenant_id = :tid AND deleted_at IS NULL AND {col} IS NOT NULL"),
+                {"tid": tenant_id},
+            )
+            indexed = indexed_result.scalar() or 0
+
+            stats[table] = {
+                "total": total,
+                "indexed": indexed,
+                "pending": total - indexed,
+            }
+        except Exception:
+            logger.exception("Stats query failed for %s", table)
+            stats[table] = {"total": 0, "indexed": 0, "pending": 0}
+
+    # Last index log entries
+    try:
+        log_result = await db.execute(
+            text(
+                """
+                SELECT entity_type, action, status, created_at
+                FROM unified_search_index_log
+                WHERE tenant_id = :tid
+                ORDER BY created_at DESC
+                LIMIT 10
+                """
+            ),
+            {"tid": tenant_id},
+        )
+        recent_logs = [dict(r) for r in log_result.mappings().all()]
+    except Exception:
+        recent_logs = []
+
+    stats["recent_logs"] = recent_logs
+    return stats
@@ -0,0 +1,71 @@
+"""Pydantic schemas for the Unified Search plugin."""
+
+from __future__ import annotations
+
+from typing import Any
+
+from pydantic import BaseModel, Field
+
+
+# ─── Search ───
+
+class SearchRequest(BaseModel):
+    query: str = Field(..., min_length=1, max_length=500)
+    entity_types: list[str] | None = None
+    limit: int = Field(default=20, ge=1, le=100)
+    offset: int = Field(default=0, ge=0)
+
+
+class SearchResult(BaseModel):
+    entity_type: str
+    entity_id: str
+    title: str
+    snippet: str
+    score: float
+    data: dict[str, Any] = Field(default_factory=dict)
+
+
+class SearchResponse(BaseModel):
+    query: str
+    normalized_query: str
+    results: list[SearchResult]
+    facets: dict[str, Any]
+    summary: str
+    suggestions: list[str]
+
+
+# ─── Similar ───
+
+class SimilarRequest(BaseModel):
+    entity_type: str
+    entity_id: str
+    limit: int = Field(default=5, ge=1, le=50)
+
+
+class SimilarResponse(BaseModel):
+    similar: dict[str, list[SearchResult]]
+
+
+# ─── Suggest ───
+
+class SuggestRequest(BaseModel):
+    q: str = Field(..., min_length=1, max_length=200)
+    limit: int = Field(default=10, ge=1, le=50)
+
+
+class SuggestResponse(BaseModel):
+    suggestions: list[str]
+
+
+# ─── Reindex ───
+
+class ReindexRequest(BaseModel):
+    entity_types: list[str] = Field(default_factory=list)
+
+
+# ─── Provider ───
+
+class ProviderResponse(BaseModel):
+    entity_type: str
+    plugin_name: str
+    is_active: bool
@@ -0,0 +1,252 @@
+"""Hybrid search engine: FTS + pgvector with RRF fusion."""
+
+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.embedding import generate_embedding
+from app.plugins.builtins.unified_search.provider_registry import get_search_registry
+
+logger = logging.getLogger(__name__)
+
+# Entity type -> (table_name, tsv_column, embedding_column)
+SEARCHABLE_ENTITIES: dict[str, tuple[str, str, str]] = {
+    "contact": ("contacts", "search_tsv", "embedding"),
+    "company": ("companies", "search_tsv", "embedding"),
+    "mail": ("mails", "body_tsv", "embedding"),
+    "file": ("files", "content_tsv", "embedding"),
+    "event": ("calendar_entries", "search_tsv", "embedding"),
+}
+
+RRF_K = 60
+RRF_ALPHA = 0.5
+RRF_BETA = 0.5
+
+
+def rrf_fusion(
+    fts_results: list[dict[str, Any]],
+    vec_results: list[dict[str, Any]],
+    entity_type: str,
+    k: int = RRF_K,
+    alpha: float = RRF_ALPHA,
+    beta: float = RRF_BETA,
+) -> list[dict[str, Any]]:
+    """Reciprocal Rank Fusion of FTS and vector search results.
+
+    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)
+
+
+async def hybrid_search(
+    db: AsyncSession,
+    query_analysis: dict[str, Any],
+    tenant_id: uuid.UUID,
+    entity_types: list[str] | None = None,
+    limit: int = 20,
+) -> list[dict[str, Any]]:
+    """Perform hybrid search across all entity types.
+
+    For each entity: FTS search + vector search + RRF fusion.
+    Merge all results, sort by fused score, return top limit.
+    """
+    registry = get_search_registry()
+    all_entity_types = entity_types or registry.get_entity_types()
+
+    normalized_query = query_analysis.get("normalized_query", "")
+    semantic_terms = query_analysis.get("semantic_terms", [])
+
+    # Build tsquery string
+    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
+
+    # Generate query embedding
+    query_text = normalized_query
+    if semantic_terms:
+        query_text = f"{normalized_query} {' '.join(semantic_terms)}"
+    query_embedding = await generate_embedding(query_text)
+
+    all_results: list[dict[str, Any]] = []
+    fetch_limit = limit * 2
+
+    for entity_type in all_entity_types:
+        provider = registry.get(entity_type)
+        if provider is None:
+            continue
+
+        fts_results: list[dict[str, Any]] = []
+        vec_results: list[dict[str, Any]] = []
+
+        try:
+            fts_results = await provider.search_fts(db, tsquery, tenant_id, fetch_limit)
+        except Exception:
+            logger.exception("FTS search failed for %s", entity_type)
+
+        if query_embedding:
+            try:
+                vec_results = await provider.search_vector(db, query_embedding, tenant_id, fetch_limit)
+            except Exception:
+                logger.exception("Vector search failed for %s", entity_type)
+
+        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)
+            all_results.append(result)
+
+    all_results.sort(key=lambda x: x.get("score", 0.0), reverse=True)
+    return all_results[:limit]
+
+
+async def find_similar_all_types(
+    db: AsyncSession,
+    entity_type: str,
+    entity_id: uuid.UUID,
+    tenant_id: uuid.UUID,
+    limit: int = 5,
+) -> dict[str, list[dict[str, Any]]]:
+    """Find similar entities across all types based on embedding.
+
+    Gets the embedding of the source entity, then searches all other tables.
+    """
+    table_map = SEARCHABLE_ENTITIES
+    source_info = table_map.get(entity_type)
+    if not source_info:
+        return {}
+
+    table_name, _, emb_col = source_info
+
+    # Get source embedding
+    sql = text(
+        f"SELECT {emb_col} AS embedding FROM {table_name} "
+        f"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 or not row.get("embedding"):
+        return {}
+
+    source_embedding_str = str(row["embedding"])
+
+    registry = get_search_registry()
+    similar: dict[str, list[dict[str, Any]]] = {}
+
+    for etype, (tbl, _, emb) in table_map.items():
+        if etype == entity_type:
+            continue
+        provider = registry.get(etype)
+        if provider is None:
+            continue
+        try:
+            sql_sim = text(
+                f"""
+                SELECT *, 1 - ({emb} <=> cast(:emb_str AS vector)) AS score
+                FROM {tbl}
+                WHERE tenant_id = :tid
+                  AND deleted_at IS NULL
+                  AND {emb} IS NOT NULL
+                ORDER BY {emb} <=> cast(:emb_str AS vector)
+                LIMIT :lim
+                """
+            )
+            res = await db.execute(
+                sql_sim,
+                {"emb_str": source_embedding_str, "tid": tenant_id, "lim": limit},
+            )
+            rows = res.mappings().all()
+            results = []
+            for r in rows:
+                sr = provider.to_search_result(dict(r))
+                sr["score"] = float(r.get("score", 0.0))
+                results.append(sr)
+            similar[etype] = results
+        except Exception:
+            logger.exception("Similar search failed for %s", etype)
+            similar[etype] = []
+
+    return similar
+
+
+async def autocomplete(
+    db: AsyncSession,
+    query: str,
+    tenant_id: uuid.UUID,
+    limit: int = 10,
+) -> list[str]:
+    """Autocomplete using FTS prefix search.
+
+    Uses to_tsquery with :* suffix for prefix matching.
+    """
+    query_clean = query.strip().replace(" ", " & ")
+    if not query_clean:
+        return []
+    tsquery = f"{query_clean}:*"
+
+    suggestions: list[str] = []
+    registry = get_search_registry()
+
+    for entity_type in registry.get_entity_types():
+        provider = registry.get(entity_type)
+        if provider is None:
+            continue
+        entity_info = SEARCHABLE_ENTITIES.get(entity_type)
+        if not entity_info:
+            continue
+        table_name, tsv_col, _ = entity_info
+        try:
+            sql = text(
+                f"""
+                SELECT * FROM {table_name}
+                WHERE tenant_id = :tid
+                  AND deleted_at IS NULL
+                  AND {tsv_col} @@ to_tsquery('pg_catalog.german', :q)
+                LIMIT :lim
+                """
+            )
+            result = await db.execute(
+                sql,
+                {"q": tsquery, "tid": tenant_id, "lim": limit},
+            )
+            rows = result.mappings().all()
+            for r in rows:
+                sr = provider.to_search_result(dict(r))
+                title = sr.get("title", "")
+                if title and title not in suggestions:
+                    suggestions.append(title)
+        except Exception:
+            logger.debug("Autocomplete failed for %s", entity_type)
+
+    return suggestions[:limit]
@@ -0,0 +1,126 @@
+"""Text extraction from various file formats for DMS indexing."""
+
+from __future__ import annotations
+
+import logging
+from typing import Any
+
+logger = logging.getLogger(__name__)
+
+MAX_CHARS = 50_000
+
+
+async def extract_text_from_file(file_path: str, mime_type: str) -> str:
+    """Extract text content from a file based on its MIME type.
+
+    Supports:
+    - PDF (via PyMuPDF/fitz)
+    - DOCX (via python-docx)
+    - XLSX (via openpyxl)
+    - PPTX (via python-pptx)
+    - text/* (via aiofiles)
+    - Images: returns empty (OCR not implemented)
+
+    Args:
+        file_path: Absolute path to the file.
+        mime_type: MIME type of the file.
+
+    Returns:
+        Extracted text, truncated to MAX_CHARS. Empty string on error.
+    """
+    try:
+        if mime_type == "application/pdf":
+            return await _extract_pdf(file_path)
+        elif mime_type == "application/vnd.openxmlformats-officedocument.wordprocessingml.document":
+            return await _extract_docx(file_path)
+        elif mime_type == "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet":
+            return await _extract_xlsx(file_path)
+        elif mime_type == "application/vnd.openxmlformats-officedocument.presentationml.presentation":
+            return await _extract_pptx(file_path)
+        elif mime_type.startswith("text/"):
+            return await _extract_text(file_path)
+        elif mime_type.startswith("image/"):
+            return ""
+        else:
+            logger.debug("Unsupported mime_type=%s for text extraction", mime_type)
+            return ""
+    except Exception:
+        logger.warning("Failed to extract text from %s (mime=%s)", file_path, mime_type)
+        return ""
+
+
+def _truncate(text: str) -> str:
+    """Truncate text to MAX_CHARS with indicator."""
+    if len(text) > MAX_CHARS:
+        return text[:MAX_CHARS] + "... [truncated]"
+    return text
+
+
+async def _extract_pdf(file_path: str) -> str:
+    """Extract text from PDF using PyMuPDF (fitz)."""
+    import fitz  # PyMuPDF
+
+    doc = fitz.open(file_path)
+    parts: list[str] = []
+    for page in doc:
+        parts.append(page.get_text())
+    doc.close()
+    return _truncate("\n".join(parts))
+
+
+async def _extract_docx(file_path: str) -> str:
+    """Extract text from DOCX using python-docx."""
+    from docx import Document
+
+    doc = Document(file_path)
+    parts: list[str] = []
+    for para in doc.paragraphs:
+        if para.text.strip():
+            parts.append(para.text)
+    # Also extract table text
+    for table in doc.tables:
+        for row in table.rows:
+            for cell in row.cells:
+                if cell.text.strip():
+                    parts.append(cell.text)
+    return _truncate("\n".join(parts))
+
+
+async def _extract_xlsx(file_path: str) -> str:
+    """Extract text from XLSX using openpyxl."""
+    from openpyxl import load_workbook
+
+    wb = load_workbook(file_path, read_only=True, data_only=True)
+    parts: list[str] = []
+    for ws in wb.worksheets:
+        for row in ws.iter_rows(values_only=True):
+            row_text = " ".join(str(c) for c in row if c is not None)
+            if row_text.strip():
+                parts.append(row_text)
+    wb.close()
+    return _truncate("\n".join(parts))
+
+
+async def _extract_pptx(file_path: str) -> str:
+    """Extract text from PPTX using python-pptx."""
+    from pptx import Presentation
+
+    prs = Presentation(file_path)
+    parts: list[str] = []
+    for slide in prs.slides:
+        for shape in slide.shapes:
+            if shape.has_text_frame:
+                for para in shape.text_frame.paragraphs:
+                    text = para.text.strip()
+                    if text:
+                        parts.append(text)
+    return _truncate("\n".join(parts))
+
+
+async def _extract_text(file_path: str) -> str:
+    """Read plain text file using aiofiles."""
+    import aiofiles
+
+    async with aiofiles.open(file_path, mode="r", encoding="utf-8", errors="replace") as f:
+        content = await f.read()
+    return _truncate(content)