python/src/server/services/agents/dispatcher.py

"""
Phase 5.1.0: Agent Dispatcher & Strategy Pattern
Resolves the 'God Method' coupling in agent_service.py by encapsulating
different agent execution behaviors into independent strategies.
"""

from abc import ABC, abstractmethod
from typing import Any, cast

from src.server.config.logfire_config import get_logger
from src.server.config.model_ssot import SYSTEM_MODELS
from src.server.services.agent_registry import get_agent_config
from src.server.services.crawling.crawling_service import CrawlingService
from src.server.services.credential_service import credential_service
from src.server.services.llm.clients import get_llm_client
from src.server.services.projects.task_service import task_service
from src.server.services.shared_constants import AgentUUIDs
from src.server.services.system.rate_limiter import GlobalThrottler
from src.server.utils import get_supabase_client

logger = get_logger(__name__)


class BaseAgentStrategy(ABC):
    @abstractmethod
    async def execute(self, task_id: str, task_data: dict[str, Any], agent_id: str, agent_service: Any) -> None:
        """Executes the specific strategy for the given agent and task."""
        pass


class SupervisorStrategy(BaseAgentStrategy):
    """Routes the task to the isolated archon-agents WorkflowEngine container."""

    async def execute(self, task_id: str, task_data: dict[str, Any], agent_id: str, agent_service: Any) -> None:
        logger.info(f"🌉 Strategy: Routing task '{task_id}' to WorkflowEngine (Supervisor).")
        # Leverage the existing HTTP bridge method in AgentService for now
        await agent_service._run_workflow_engine_task(task_id, task_data, agent_id)


class LibrarianDirectStrategy(BaseAgentStrategy):
    """Bypasses LLM and directly triggers the crawling pipeline if the description is empty."""

    async def execute(self, task_id: str, task_data: dict[str, Any], agent_id: str, agent_service: Any) -> None:
        logger.info(f"[{agent_id}] Strategy: Direct crawler pipeline triggered for empty description")
        try:
            target_id = task_data["crawler_target_id"]
            supabase = get_supabase_client()
            res = supabase.table("archon_crawler_targets").select("*").eq("id", target_id).execute()
            if not res.data:
                raise ValueError(f"Crawler target {target_id} not found")
            target = res.data[0]

            crawler = CrawlingService()
            await crawler.orchestrate_crawl(
                {
                    "url": target["target_url"],
                    "max_depth": target.get("max_depth", 2),
                    "user_role": "system_admin",
                }
            )
            output_msg = f"Direct crawler pipeline started for {target['target_url']}"
            await task_service.update_task(task_id, {"status": "done"})
            await agent_service._award_agent_xp(agent_id, task_data, output_msg)
        except Exception as e:
            logger.error(f"Direct crawl failed: {e}")
            await task_service.update_task(task_id, {"status": "failed"})


class DefaultLLMStrategy(BaseAgentStrategy):
    """The traditional single-turn LLM execution via Google GenAI."""

    async def execute(self, task_id: str, task_data: dict[str, Any], agent_id: str, agent_service: Any) -> None:
        config = get_agent_config(agent_id)
        if not config:
            logger.error(f"Agent '{agent_id}' not found.")
            await task_service.update_task(task_id, {"status": "failed"})
            return

        task_desc = task_data.get("description", "No description provided.")
        messages = [
            {"role": "system", "content": config["system_prompt"]},
            {"role": "user", "content": f"Task: {task_data['title']}\n\nDetails: {task_desc}"},
        ]

        # Physical Synchronization: Fetch dynamic tools from MCP (Phase 4.6.19)
        all_mcp_tools: list[dict[str, Any]] = []
        if agent_service.mcp_client:
            all_mcp_tools = await agent_service.mcp_client.list_tools()
            logger.info(f"Dynamic Tool Discovery: Synced {len(all_mcp_tools)} tools from MCP.")

        agent_tools_list: list[str] = list(config.get("tools") or [])
        agent_tools = [t for t in all_mcp_tools if cast(dict, t["function"])["name"] in agent_tools_list]

        tools_param = agent_tools if agent_tools else None

        try:
            admin_api_key = await credential_service.get_credential(
                "GEMINI_API_KEY"
            ) or await credential_service.get_credential("GOOGLE_API_KEY")

            tier = config.get("model_tier", "lite")
            model_key = "DEFAULT_PRO" if tier == "pro" else "DEFAULT_TEXT"
            active_model = SYSTEM_MODELS[model_key]

            await GlobalThrottler.wait_for_capacity(tier=tier)

            async with get_llm_client(api_key=admin_api_key) as client:
                logger.info(
                    f"Generating content using SDK for model {active_model} with {len(agent_tools_list)} tools."
                )

                response = await client.chat.completions.create(
                    model=active_model, messages=messages, tools=tools_param
                )
                res_msg = response.choices[0].message
                if res_msg.tool_calls and agent_service.mcp_client:
                    messages.append(res_msg)
                    tool_results = await agent_service.tool_executor.handle_tool_calls(
                        res_msg.tool_calls, agent_id=agent_id
                    )
                    messages.extend(tool_results)
                    final_response = await client.chat.completions.create(
                        model=active_model, messages=messages, tools=tools_param
                    )
                    final_output = final_response.choices[0].message.content
                else:
                    final_output = res_msg.content

                # Save output & update status
                await task_service.update_task(task_id, {"status": "done"})

                # Phase 5.0.2 DB Compatibility: wrap in object
                save_payload = {"content": final_output}
                await task_service.save_agent_output(task_id, save_payload, agent_id)

                await agent_service._award_agent_xp(agent_id, task_data, final_output)

        except Exception as e:
            logger.error(f"Error executing DefaultLLMStrategy: {e}", exc_info=True)
            await task_service.update_task(task_id, {"status": "failed"})


class DraftFromLeadsStrategy(BaseAgentStrategy):
    """Phase 5.1.1: Executes the blog drafting pipeline from lead parameters."""

    async def execute(self, task_id: str, task_data: dict[str, Any], agent_id: str, agent_service: Any) -> None:
        logger.info(f"[{agent_id}] Strategy: Blog drafting pipeline triggered for task {task_id}")
        try:
            # 1. Extract lead_ids from metadata or description
            lead_ids = task_data.get("metadata", {}).get("lead_ids", [])
            if not lead_ids:
                # Fallback to description parsing if metadata column not present
                desc = task_data.get("description", "")
                if "[PARAM:LEAD_IDS:" in desc:
                    params = desc.split("[PARAM:LEAD_IDS:")[1].split("]")[0]
                    lead_ids = params.split(",")

            if not lead_ids:
                raise ValueError("No lead_ids found in task parameters.")

            # 2. Call the physical handler
            from src.server.services.marketing.content_handler import ContentHandler

            handler = ContentHandler(get_supabase_client())

            output_msg = await handler.draft_from_leads_physical(task_id, lead_ids)

            # 3. Update task
            await task_service.update_task(task_id, {"status": "done"})
            await agent_service._award_agent_xp(agent_id, task_data, output_msg)

        except Exception as e:
            logger.error(f"DraftFromLeadsStrategy failed: {e}")
            await task_service.update_task(task_id, {"status": "failed"})


class AgentDispatcher:
    """Routes tasks to the appropriate strategy based on agent_id and task_data."""

    def __init__(self):
        self._strategies: list[tuple[Any, BaseAgentStrategy]] = []
        self._default_strategy = DefaultLLMStrategy()

        # Register strategies (Condition function, Strategy instance)
        self.register_strategy(lambda a_id, t_data: a_id == AgentUUIDs.SUPERVISOR, SupervisorStrategy())
        self.register_strategy(
            lambda a_id, t_data: (
                a_id == AgentUUIDs.LIBRARIAN
                and t_data.get("crawler_target_id")
                and not t_data.get("description", "").strip()
            ),
            LibrarianDirectStrategy(),
        )
        self.register_strategy(
            lambda a_id, t_data: (
                a_id == AgentUUIDs.LIBRARIAN
                and t_data.get("crawler_target_id")
                and t_data.get("description", "").strip().lower() in ["periodic sync", "knowledge sync"]
            ),
            LibrarianDirectStrategy(),
        )
        self.register_strategy(
            lambda a_id, t_data: (
                a_id == AgentUUIDs.MARKET_BOT
                and (t_data.get("feature") == "blog_drafting" or "AI Draft from Leads" in t_data.get("title", ""))
            ),
            DraftFromLeadsStrategy(),
        )

    def register_strategy(self, condition_func: Any, strategy: BaseAgentStrategy):
        self._strategies.append((condition_func, strategy))

    def get_strategy(self, agent_id: str, task_data: dict[str, Any]) -> BaseAgentStrategy:
        for condition_func, strategy in self._strategies:
            if condition_func(agent_id, task_data):
                return strategy
        return self._default_strategy


agent_dispatcher = AgentDispatcher()