Timmy-time-dashboard/src/timmy/tools_delegation/__init__.py

"""Timmy's delegation tools — submit tasks and list agents.

Reads agent roster from agents.yaml via the loader module.
No hardcoded agent lists.
"""

import logging
from typing import Any

logger = logging.getLogger(__name__)


def delegate_task(
    agent_name: str, task_description: str, priority: str = "normal"
) -> dict[str, Any]:
    """Record a delegation intent to another agent.

    Args:
        agent_name: Name or ID of the agent to delegate to
        task_description: What you want the agent to do
        priority: Task priority - "low", "normal", "high"

    Returns:
        Dict with agent, status, and message
    """
    from timmy.agents.loader import list_agents

    agent_name = agent_name.lower().strip()

    # Build valid agents map from YAML config
    available = {a["id"]: a["role"] for a in list_agents()}

    if agent_name not in available:
        return {
            "success": False,
            "error": f"Unknown agent: {agent_name}. Valid agents: {', '.join(sorted(available))}",
            "task_id": None,
        }

    valid_priorities = ["low", "normal", "high"]
    if priority not in valid_priorities:
        priority = "normal"

    logger.info(
        "Delegation intent: %s → %s (priority=%s)", agent_name, task_description[:80], priority
    )

    return {
        "success": True,
        "task_id": None,
        "agent": agent_name,
        "role": available[agent_name],
        "status": "noted",
        "message": f"Delegation to {agent_name} ({available[agent_name]}): {task_description[:100]}",
    }


def list_swarm_agents() -> dict[str, Any]:
    """List all available sub-agents and their roles.

    Reads from agents.yaml — no hardcoded roster.

    Returns:
        Dict with agent list
    """
    try:
        from timmy.agents.loader import list_agents

        agents = list_agents()
        return {
            "success": True,
            "agents": [
                {
                    "name": a["name"],
                    "id": a["id"],
                    "role": a["role"],
                    "status": a.get("status", "available"),
                    "capabilities": ", ".join(a.get("tools", [])),
                }
                for a in agents
            ],
        }
    except Exception as e:
        logger.debug("Agent list unavailable: %s", e)
        return {
            "success": False,
            "error": str(e),
            "agents": [],
        }


def _find_kimi_cli() -> str | None:
    """Return the path to the kimi CLI binary, or None if not installed."""
    import shutil

    return shutil.which("kimi")


def _resolve_workdir(working_directory: str) -> str | dict[str, Any]:
    """Return a validated working directory path, or an error dict."""
    from pathlib import Path

    from config import settings

    workdir = working_directory or settings.repo_root
    if not Path(workdir).is_dir():
        return {
            "success": False,
            "error": f"Working directory does not exist: {workdir}",
        }
    return workdir


def _run_kimi(cmd: list[str], workdir: str) -> dict[str, Any]:
    """Execute the kimi subprocess and return a result dict."""
    import subprocess

    try:
        result = subprocess.run(
            cmd,
            capture_output=True,
            text=True,
            timeout=300,  # 5 minute timeout for coding tasks
            cwd=workdir,
        )

        output = result.stdout.strip()
        if result.returncode != 0 and result.stderr:
            output += "\n\nSTDERR:\n" + result.stderr.strip()

        return {
            "success": result.returncode == 0,
            "output": output[-4000:] if len(output) > 4000 else output,
            "return_code": result.returncode,
        }
    except subprocess.TimeoutExpired:
        return {
            "success": False,
            "error": "Kimi timed out after 300s. Task may be too broad — try breaking it into smaller pieces.",
        }
    except Exception as exc:
        return {
            "success": False,
            "error": f"Failed to run Kimi: {exc}",
        }


def delegate_to_kimi(task: str, working_directory: str = "") -> dict[str, Any]:
    """Delegate a coding task to Kimi, the external coding agent.

    Kimi has 262K context and is optimized for code tasks: writing,
    debugging, refactoring, test writing. Timmy thinks and plans,
    Kimi executes bulk code changes.

    Args:
        task: Clear, specific coding task description. Include file paths
              and expected behavior. Good: "Fix the bug in src/timmy/session.py
              where sessions don't persist." Bad: "Fix all bugs."
        working_directory: Directory for Kimi to work in. Defaults to repo root.

    Returns:
        Dict with success status and Kimi's output or error.
    """
    kimi_path = _find_kimi_cli()
    if not kimi_path:
        return {
            "success": False,
            "error": "kimi CLI not found on PATH. Install with: pip install kimi-cli",
        }

    workdir = _resolve_workdir(working_directory)
    if isinstance(workdir, dict):
        return workdir

    logger.info("Delegating to Kimi: %s (cwd=%s)", task[:80], workdir)
    return _run_kimi([kimi_path, "--print", "-p", task], workdir)