python/src/server/services/agent_service.py

# python/src/server/services/agent_service.py


from ..config.logfire_config import get_logger
from .agent_registry import get_agent_config
from .agent_tool_executor import AgentToolExecutor
from .dev_ops_agent_service import DevOpsAgentService
from .shared_constants import AI_AGENT_ROLES


class AgentService:
    """Service for handling business logic related to AI agents."""

    def __init__(self, mcp_client=None):
        self.tool_executor = AgentToolExecutor(mcp_client)
        self.dev_ops = DevOpsAgentService(self.tool_executor)

    @property
    def mcp_client(self):
        return self.tool_executor.mcp_client

    @mcp_client.setter
    def mcp_client(self, value):
        self.tool_executor.mcp_client = value

    async def get_assignable_agents(self, user_role: str | None = None) -> list[dict]:
        all_agents = []
        for role_name, agent_id in AI_AGENT_ROLES.items():
            all_agents.append(
                {"id": agent_id, "name": role_name, "role": role_name, "tools": [], "description": "AI Agent"}
            )

        from src.server.utils import get_supabase_client

        from .shared_constants import AgentUUIDs

        system_bots = [AgentUUIDs.PO_BOT, AgentUUIDs.CLOCKWORK]

        # 1. Try to load role mapping dynamically from database
        try:
            supabase = get_supabase_client()
            res = supabase.table("archon_role_agents").select("agent_key").eq("user_role", user_role or "").execute()
            if res.data:
                allowed_keys = {row["agent_key"] for row in res.data}

                from .agent_registry import get_agent_uuid
                allowed_ids = {get_agent_uuid(k) for k in allowed_keys if get_agent_uuid(k)}

                filtered = []
                for agent in all_agents:
                    agent_id = str(agent["id"])
                    if agent_id in allowed_ids:
                        config = get_agent_config(agent_id)
                        if config:
                            agent["tools"] = config.get("tools", [])
                            agent["description"] = config.get("system_prompt", "").split("\n")[0]
                        filtered.append(agent)
                return filtered
        except Exception:
            pass

        # 2. Fallback to static mapping in case database is down or not seeded
        if not user_role or user_role in ["admin", "system_admin", "manager"]:
            for agent in all_agents:
                agent_id = str(agent["id"])
                if agent_id in system_bots:
                    continue
                config = get_agent_config(agent_id)
                if config:
                    agent["tools"] = config.get("tools", [])
            return [a for a in all_agents if str(a["id"]) not in system_bots]

        filtered = []
        for agent in all_agents:
            agent_id = str(agent["id"])
            config = get_agent_config(agent_id)
            if config:
                agent["tools"] = config.get("tools", [])
                agent["description"] = config.get("system_prompt", "").split("\n")[0]
            if user_role == "sales" and agent_id == AgentUUIDs.MARKET_BOT:
                filtered.append(agent)
            elif user_role == "marketing" and agent_id in [AgentUUIDs.MARKET_BOT, AgentUUIDs.LIBRARIAN]:
                filtered.append(agent)
        return filtered

    async def run_agent_task(self, task_id: str, agent_id: str, immediate: bool = False):
        from ..services.projects.task_service import task_service

        logger = get_logger(__name__)

        if not immediate:
            logger.info(f"📥 Enqueuing task '{task_id}' for AI agent '{agent_id}'.")
            success, result = await task_service.update_task(task_id, {"status": "dispatched", "assignee": agent_id})
            if not success:
                logger.error(f"Failed to enqueue task: {result.get('error')}")
            return

        logger.info(f"🚀 AI agent '{agent_id}' starting physical work on task '{task_id}'.")

        # When immediate=True, we move to 'doing' (or it might be 'processing' from worker)
        success, result = await task_service.update_task(task_id, {"status": "doing", "assignee": agent_id})
        if not success:
            logger.error(f"Failed to update task status to doing: {result.get('error')}")
            return

        await self._run_general_agent_task(task_id, agent_id)

    async def _award_agent_xp(self, agent_id: str, task_data: dict, output_message: str):
        from .shared_constants import AgentUUIDs
        from .stats import stats_service

        # Physical Scoring instead of random (Phase 4.6.15)
        # We derive metadata from the task context
        meta = {
            "lint_passed": "Success" in output_message if output_message else False,  # Heuristic for self-healing
            "required_terms": ["Archon"] if agent_id == AgentUUIDs.LIBRARIAN else [],
        }


        score = stats_service.calculate_ai_score(output_message, meta)
        # Translate 0-100 score to 0-15 XP
        xp = int(score / 6.6)

        # Grounded ID check from registry (e.g. ai-dev-bot -> Archon DevBot)
        from .agent_registry import get_agent_config

        config = get_agent_config(agent_id)
        display_name = config["name"] if config else agent_id

        msg = f"Completed {display_name} task: {task_data.get('title', 'Unknown')}"

        await stats_service.add_agent_action_log(
            agent_name=display_name,
            agent_id=agent_id,
            xp_change=xp,
            message=msg,
            details={"task_id": task_data.get("id"), "score": score},
        )

    async def _run_workflow_engine_task(self, task_id: str, task_data: dict, agent_id: str):
        """Phase 5.0.2: Bridges the execution to the isolated archon-agents WorkflowEngine container."""
        import os

        import httpx

        from ..services.projects.task_service import task_service

        logger = get_logger(__name__)

        # 1. Determine task_type for dynamic prompt routing
        task_title = task_data.get("title", "")
        task_type = "General"
        # Temporary hack: Deduce task_type from title since UI lacks a dropdown
        if "Marketing Data Deep Dive" in task_title or "行銷數據" in task_title:
            task_type = "Marketing Data Deep Dive"
        elif "[Daily Report]" in task_title:
            task_type = "Daily Executive Summary"

        prompt = f"Task: {task_title}\n\nDetails: {task_data.get('description', '')}"

        # 2. Call WorkflowEngine via httpx
        agents_url = os.getenv("AGENTS_SERVICE_URL", "http://archon-agents:8052")
        try:
            # Group chats take time, set a safe timeout
            async with httpx.AsyncClient(timeout=300.0) as client:
                response = await client.post(
                    f"{agents_url}/agents/workflow/run",
                    json={"prompt": prompt, "context": {"task_type": task_type}},
                )
                response.raise_for_status()
                data = response.json()

                if data.get("success"):
                    await task_service.update_task(task_id, {"status": "done"})

                    # Milestone 2: Save the entire JSON state, not just final_result
                    messages = data.get("metadata", {}).get("messages", [])
                    final_result = data.get("result", "")

                    save_payload = {
                        "content": final_result,
                        "messages": messages,
                        "step_count": data.get("metadata", {}).get("step_count", 0),
                    }
                    await task_service.save_agent_output(task_id, save_payload, agent_id)
                    await self._award_agent_xp(agent_id, task_data, str(final_result))
                else:
                    logger.error(f"WorkflowEngine failed: {data.get('error')}")
                    await task_service.update_task(task_id, {"status": "failed"})

        except httpx.RequestError as e:
            logger.error(f"Network error calling WorkflowEngine: {e}")
            await task_service.update_task(task_id, {"status": "failed"})
        except Exception as e:
            logger.error(f"Unexpected error in WorkflowEngine execution: {e}")
            await task_service.update_task(task_id, {"status": "failed"})

    async def _run_general_agent_task(self, task_id: str, agent_id: str):
        from ..services.projects.task_service import task_service

        logger = get_logger(__name__)

        success, task_response = await task_service.get_task(task_id)
        if not (success and task_response and "task" in task_response):
            return
        task_data = task_response["task"]

        # Phase 5.1.0 Milestone 1: Delegate execution to the Agent Dispatcher (Strategy Pattern)
        from .agents.dispatcher import agent_dispatcher

        strategy = agent_dispatcher.get_strategy(agent_id, task_data)
        logger.info(f"🚀 Dispatching task '{task_id}' for agent '{agent_id}' using {strategy.__class__.__name__}")

        await strategy.execute(task_id, task_data, agent_id, self)


agent_service = AgentService()