"""Delegation engine — spawn child AgentRunner instances for complex parallel tasks.""" from __future__ import annotations import asyncio import json import logging import os logger = logging.getLogger("shira.delegate") MAX_CONCURRENT_CHILDREN = 3 BLOCKED_TOOLS = {"delegate_task", "update_user_profile", "update_case_memory", "skill_manage"} async def run_child_task( goal: str, context: dict, espocrm_url: str, espocrm_api_key: str, allowed_toolsets: list[str] | None = None, ) -> str: """Run a single child agent conversation with a focused goal.""" from api.services.agent_runner import AgentRunner runner = AgentRunner( ai_gateway_url=os.environ.get("AI_GATEWAY_URL", "http://localhost:3000"), ai_gateway_api_key=os.environ.get("AI_GATEWAY_API_KEY", ""), model=os.environ.get("CLAUDE_MODEL", "sonnet"), max_tokens=int(os.environ.get("MAX_OUTPUT_TOKENS", "4096")), max_iterations=8, ) system_prompt = ( "You are a focused sub-agent working on a specific delegated task.\n" "Complete the task thoroughly and return a clear, concise result.\n" "Always respond in Hebrew.\n\n" f"YOUR TASK:\n{goal}\n" ) messages = [{"role": "user", "content": goal}] try: result = await runner.run( system_prompt=system_prompt, messages=messages, context=context, espocrm_url=espocrm_url, espocrm_api_key=espocrm_api_key, depth=1, blocked_tools=BLOCKED_TOOLS, allowed_toolsets=allowed_toolsets, ) return result except Exception as e: logger.error("Child task failed: %s — %s", goal[:50], e) return f"❌ משימה נכשלה: {e}" async def delegate_tasks( tasks: list[dict], context: dict, espocrm_url: str, espocrm_api_key: str, ) -> str: """Run multiple child tasks concurrently. Returns combined results.""" if len(tasks) > MAX_CONCURRENT_CHILDREN: tasks = tasks[:MAX_CONCURRENT_CHILDREN] logger.warning("Truncated to %d concurrent children", MAX_CONCURRENT_CHILDREN) coros = [ run_child_task( goal=t["goal"], context=context, espocrm_url=espocrm_url, espocrm_api_key=espocrm_api_key, allowed_toolsets=t.get("toolsets"), ) for t in tasks ] results = await asyncio.gather(*coros, return_exceptions=True) parts = [] for i, (task, result) in enumerate(zip(tasks, results), 1): if isinstance(result, Exception): parts.append(f"### משימה {i}: {task['goal']}\n❌ שגיאה: {result}") else: parts.append(f"### משימה {i}: {task['goal']}\n{result}") return "\n\n".join(parts) def register_delegate_tool( tools: dict, context: dict, espocrm_url: str, espocrm_api_key: str, ) -> None: """Register the delegate_task tool.""" async def delegate_task(goal: str, context_extra: str = "", toolsets: str = "") -> str: """Delegate a single complex task to a focused sub-agent.""" allowed = toolsets.split(",") if toolsets else None result = await run_child_task( goal=f"{goal}\n\n{context_extra}" if context_extra else goal, context=context, espocrm_url=espocrm_url, espocrm_api_key=espocrm_api_key, allowed_toolsets=allowed, ) return result tools["delegate_task"] = { "description": ( "Delegate a complex task to a focused sub-agent that runs independently. " "Use for: parallel document analysis, multi-step research, report preparation. " "The sub-agent has access to CRM and document tools but cannot delegate further or modify memory/skills." ), "parameters": { "type": "object", "properties": { "goal": {"type": "string", "description": "Clear description of the task for the sub-agent"}, "context_extra": {"type": "string", "description": "Additional context to help the sub-agent"}, "toolsets": {"type": "string", "description": "Comma-separated tool categories to allow (crm,documents,legal). Empty = all."}, }, "required": ["goal"], }, "handler": delegate_task, }