Agentswarm/agent/handoff_logic.py

"""Handoff decision logic for determining when to delegate tasks."""
import logging
from typing import Iterable, List, Optional
from dataclasses import dataclass

logger = logging.getLogger(__name__)


@dataclass
class HandoffDecision:
    """Result of handoff decision analysis."""
    should_handoff: bool
    reason: str
    target_capabilities: List[str]


CAPABILITY_ALIASES = {
    "code_generation": {
        "code-reading",
        "code-editing",
        "command-line",
        "credentials-handling",
        "editing",
        "environment-inspection",
        "environment-setup",
        "file-editing",
        "file-operations",
        "file-system",
        "filesystem",
        "read-only-operations",
        "reporting",
        "repository-inspection",
        "shell",
        "testing",
        "pytest",
        "quality-checking",
        "attention-to-detail",
        "error-handling",
        "technical-writing",
        "validation",
        "verification",
    },
    "python": {
        "pytest",
        "testing",
        "code-reading",
        "code-editing",
        "command-line",
        "editing",
        "file-editing",
        "file-operations",
        "file-system",
        "filesystem",
        "reporting",
        "shell",
        "quality-checking",
        "error-handling",
        "validation",
        "verification",
    },
    "general": {
        "general",
        "code-reading",
        "file-system",
        "filesystem",
        "quality-checking",
        "attention-to-detail",
    },
}


def _expanded_capabilities(capabilities: Iterable[str]) -> set[str]:
    """Return capabilities plus local aliases supported by this agent."""
    expanded = {"general", *capabilities}
    for capability in capabilities:
        expanded.update(CAPABILITY_ALIASES.get(capability, set()))
    return expanded


def should_handoff(
    subtask: dict,
    agent_id: str,
    agent_capabilities: Optional[List[str]] = None,
) -> HandoffDecision:
    """Determine if a subtask should be handed off to another agent.

    Args:
        subtask: Subtask definition with complexity, capabilities, etc.
        agent_id: Current agent ID
        agent_capabilities: Capabilities advertised by the current agent

    Returns:
        HandoffDecision: Decision with reasoning
    """
    complexity = subtask.get("complexity", "medium")
    required_capabilities = subtask.get("required_capabilities", ["general"])
    estimated_time = subtask.get("estimated_time", 30)
    current_capabilities = _expanded_capabilities(agent_capabilities or ["general"])

    # Decision criteria:
    # 1. High complexity tasks should be handed off to specialist agents
    # 2. Tasks requiring specialized capabilities should go to specialists
    # 3. Tasks estimated to take > 60 minutes should be broken down further

    # Check complexity threshold
    if complexity == "high":
        logger.info(f"High complexity subtask detected: {subtask['description']}")
        return HandoffDecision(
            should_handoff=True,
            reason="Task complexity exceeds agent capability threshold",
            target_capabilities=required_capabilities
        )

    # Check for specialized capabilities
    specialized_capabilities = [
        cap for cap in required_capabilities
        if cap not in current_capabilities
    ]

    if specialized_capabilities:
        logger.info(f"Specialized capabilities required: {specialized_capabilities}")
        return HandoffDecision(
            should_handoff=True,
            reason=f"Requires specialized capabilities: {', '.join(specialized_capabilities)}",
            target_capabilities=specialized_capabilities
        )

    # Check estimated time
    if estimated_time > 60:
        logger.info(f"Long-running task detected: {estimated_time} minutes")
        return HandoffDecision(
            should_handoff=True,
            reason=f"Task estimated to take {estimated_time} minutes (threshold: 60)",
            target_capabilities=required_capabilities
        )

    # No handoff needed
    return HandoffDecision(
        should_handoff=False,
        reason="Task within agent capability",
        target_capabilities=[]
    )


def select_target_agent(
    available_agents: List[dict],
    required_capabilities: List[str]
) -> str:
    """Select the best agent for a handoff based on capabilities.

    Args:
        available_agents: List of available agent metadata
        required_capabilities: Required capabilities for the task

    Returns:
        str: Selected agent ID or None if no suitable agent found
    """
    # Score each agent based on capability match
    best_agent = None
    best_score = -1

    for agent in available_agents:
        agent_capabilities = set(agent.get("capabilities", []))
        required_set = set(required_capabilities)

        # Calculate match score
        matches = len(agent_capabilities.intersection(required_set))
        total_required = len(required_set)

        if total_required > 0:
            score = matches / total_required
        else:
            score = 0

        # Prefer agents with exact capability match
        if score > best_score:
            best_score = score
            best_agent = agent

    if best_agent:
        logger.info(
            f"Selected agent {best_agent['agent_id']} "
            f"with score {best_score:.2f} for capabilities {required_capabilities}"
        )
        return best_agent["agent_id"]

    logger.warning(f"No suitable agent found for capabilities: {required_capabilities}")
    return None


def estimate_task_complexity(description: str) -> str:
    """Estimate task complexity based on description keywords.

    Args:
        description: Task description

    Returns:
        str: "low", "medium", or "high"
    """
    description_lower = description.lower()

    # High complexity indicators
    high_complexity_keywords = [
        "refactor", "redesign", "architecture", "migrate",
        "optimize", "performance", "security", "scale",
        "distributed", "concurrent", "async", "parallel"
    ]

    # Low complexity indicators
    low_complexity_keywords = [
        "fix typo", "update comment", "rename", "format",
        "add log", "simple", "trivial", "quick"
    ]

    # Check for high complexity
    if any(keyword in description_lower for keyword in high_complexity_keywords):
        return "high"

    # Check for low complexity
    if any(keyword in description_lower for keyword in low_complexity_keywords):
        return "low"

    # Default to medium
    return "medium"