shira-hermes/api/services/delegate.py

"""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,
    }