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

"""Knowledge plugin — LLM-based entity/relationship extraction, ask-knowledge, review queue."""
from __future__ import annotations
import logging
import uuid
from typing import Any
from app.plugins.base import BasePlugin
from app.plugins.manifest import PluginManifest, PluginRouteDef

logger = logging.getLogger(__name__)

class KnowledgePlugin(BasePlugin):
    manifest = PluginManifest(
        name="knowledge",
        version="1.0.0",
        display_name="Knowledge",
        description="LLM-based knowledge extraction, ask-knowledge, review queue. Builds on graph_rag + unified_search.",
        dependencies=["permissions", "graph_rag", "unified_search"],
        routes=[
            PluginRouteDef(path="/api/v1/knowledge", module="app.plugins.builtins.knowledge.routes", router_attr="router"),
        ],
        permissions=["knowledge:read", "knowledge:write", "knowledge:admin"],
    )

    async def on_activate(self, db, service_container, event_bus) -> None:
        """Register event-driven extraction hooks on activation."""
        await super().on_activate(db, service_container, event_bus)
        try:
            from app.core.hooks import get_hook_registry
            from app.plugins.builtins.knowledge.services import extract_knowledge
            async def on_wiki_create(*args, **kwargs):
                article_id = kwargs.get("article_id") or kwargs.get("entity_id")
                tenant_id = kwargs.get("tenant_id")
                title = kwargs.get("title", "")
                content = kwargs.get("content", "")
                if article_id and tenant_id and content:
                    from app.core.db import get_worker_session_factory
                    factory = get_worker_session_factory()
                    async with factory() as session:
                        await extract_knowledge(
                            db=session, tenant_id=uuid.UUID(str(tenant_id)),
                            source_type="wiki_article", source_id=uuid.UUID(str(article_id)),
                            source_title=title, source_text=content,
                        )
            get_hook_registry().register_action(
                "wiki.article.created", on_wiki_create, priority=20, owner_tag="knowledge"
            )
            # H-DATA-LIFE: Re-extract when wiki article is updated
            async def on_wiki_update(*args, **kwargs):
                article_id = kwargs.get("article_id") or kwargs.get("entity_id")
                tenant_id = kwargs.get("tenant_id")
                title = kwargs.get("title", "")
                content = kwargs.get("content", "")
                if article_id and tenant_id and content:
                    from app.core.db import get_worker_session_factory
                    factory = get_worker_session_factory()
                    async with factory() as session:
                        await extract_knowledge(
                            db=session, tenant_id=uuid.UUID(str(tenant_id)),
                            source_type="wiki_article", source_id=uuid.UUID(str(article_id)),
                            source_title=title, source_text=content,
                        )
            get_hook_registry().register_action(
                "wiki.article.updated", on_wiki_update, priority=20, owner_tag="knowledge"
            )
            logger.info("Registered knowledge extraction hooks")
        except Exception:
            logger.exception("Failed to register knowledge hooks")
        # Register knowledge agent tools (I-AK: Agent→Knowledge)
        try:
            self._register_knowledge_agent_tools()
        except Exception:
            logger.exception("Failed to register knowledge agent tools")

    def _register_knowledge_agent_tools(self) -> None:
        """Register I-AK agent tools for asking and searching knowledge."""
        from app.ai.tool_registry import get_tool_registry
        registry = get_tool_registry()

        async def _ask_knowledge_handler(arguments: dict[str, Any], context: dict[str, Any]) -> dict[str, Any]:
            """Ask a knowledge question."""
            from app.plugins.builtins.knowledge.services import ask_knowledge
            from app.core.db import get_worker_session_factory
            question = arguments.get("question", "")
            tenant_id = context.get("tenant_id")
            if not question or not tenant_id:
                return {"error": "question and tenant_id required"}
            factory = get_worker_session_factory()
            async with factory() as db:
                result = await ask_knowledge(db=db, tenant_id=uuid.UUID(str(tenant_id)), question=question)
                return {"answer": result.get("answer", ""), "evidence_count": len(result.get("evidence", []))}

        registry.register(
            name="ask_knowledge",
            description="Ask a knowledge question. Searches wiki articles and graph relationships for context.",
            parameters={
                "type": "object",
                "properties": {
                    "question": {"type": "string", "description": "The question to ask"},
                },
                "required": ["question"],
            },
            handler=_ask_knowledge_handler,
            plugin_name=self.manifest.name,
            required_permission="wiki:read",
            category="knowledge",
        )

        async def _search_knowledge_handler(arguments: dict[str, Any], context: dict[str, Any]) -> dict[str, Any]:
            """Search wiki articles via unified search."""
            from app.plugins.builtins.unified_search.provider_registry import get_search_registry
            from app.core.db import get_worker_session_factory
            query = arguments.get("query", "")
            tenant_id = context.get("tenant_id")
            if not query or not tenant_id:
                return {"error": "query and tenant_id required"}
            factory = get_worker_session_factory()
            async with factory() as db:
                search_registry = get_search_registry()
                wiki_provider = search_registry.get("wiki_article")
                if not wiki_provider:
                    return {"error": "Wiki search provider not available"}
                results = await wiki_provider._search_fts_filtered(
                    db=db, tsquery=query, tenant_id=uuid.UUID(str(tenant_id)), limit=5, visible_ids=None
                )
                return {
                    "results": [
                        {"title": r.get("title", ""), "summary": (r.get("summary") or r.get("content", "")[:200] or "")}
                        for r in results
                    ],
                    "count": len(results),
                }

        registry.register(
            name="search_knowledge",
            description="Search wiki articles by keyword. Returns matching articles with title and summary.",
            parameters={
                "type": "object",
                "properties": {
                    "query": {"type": "string", "description": "Search query"},
                },
                "required": ["query"],
            },
            handler=_search_knowledge_handler,
            plugin_name=self.manifest.name,
            required_permission="wiki:read",
            category="knowledge",
        )
        logger.info("Registered knowledge agent tools: ask_knowledge, search_knowledge")

    def get_job_modules(self) -> list[str]:
        """ARQ job modules — the retention cleanup job lives with this plugin."""
        return ["app.plugins.builtins.knowledge.jobs"]

    async def on_deactivate(self, db, service_container, event_bus) -> None:
        """Clean up on deactivation."""
        from app.core.hooks import get_hook_registry
        get_hook_registry().unregister_all_for_plugin("knowledge")
        # Unregister knowledge agent tools from the core AI tool registry
        try:
            from app.ai.tool_registry import get_tool_registry
            get_tool_registry().unregister_plugin(self.manifest.name)
            logger.info("Unregistered AI agent tools for plugin '%s'", self.manifest.name)
        except Exception:
            logger.exception("Failed to unregister AI agent tools")
        await super().on_deactivate(db, service_container, event_bus)