leocrm/app/plugins/builtins/graph_rag/plugin.py

"""GraphRAG plugin — entity relationship graph with BFS traversal and search provider."""

from __future__ import annotations

from app.plugins.base import BasePlugin
from app.plugins.manifest import PluginManifest, PluginRouteDef


class GraphRAGPlugin(BasePlugin):
    """GraphRAG plugin for entity relationship graph with BFS traversal."""

    manifest = PluginManifest(
        name="graph_rag",
        version="1.0.0",
        display_name="GraphRAG",
        description="Entity relationship graph with BFS traversal and unified search integration. Stores typed relationships between entities for knowledge graph exploration.",
        dependencies=["unified_search"],
        routes=[
            PluginRouteDef(
                path="/api/v1/graph",
                module="app.plugins.builtins.graph_rag.routes",
                router_attr="router",
            ),
        ],
        events=[],
        migrations=["0001_initial.sql"],
        permissions=[
            "graph:read",
            "graph:write",
        ],
        is_core=True,
        author="LeoCRM Team",
        min_app_version="1.0.0",
        contract_version="1.0.0",
    )

    def get_entity_models(self) -> dict[str, type]:
        from app.plugins.builtins.graph_rag.models import EntityRelationship
        return {"entity_relationship": EntityRelationship}

    async def on_activate(self, db, service_container, event_bus) -> None:
        """Activate plugin: register GraphRAG search provider."""
        # Base class first (event subscriptions etc.), then own registrations
        # so a failure in our code cannot leave the base half-initialized (ARCH-037).
        await super().on_activate(db, service_container, event_bus)

        from app.plugins.builtins.graph_rag.provider import GraphRAGSearchProvider
        from app.plugins.builtins.unified_search.contracts import get_search_registry

        registry = get_search_registry()
        try:
            registry.register(GraphRAGSearchProvider())
        except Exception:
            import logging
            logging.getLogger(__name__).exception("Failed to register GraphRAGSearchProvider")

    async def on_deactivate(self, db, service_container, event_bus) -> None:
        """Deactivate plugin: unregister search provider and contract."""
        from app.plugins.builtins.unified_search.contracts import get_search_registry
        get_search_registry().unregister("graph_relationship")

        from app.plugins.builtins.contracts import get_contract_registry
        get_contract_registry().unregister(self.manifest.name)
        await super().on_deactivate(db, service_container, event_bus)