diff --git a/config/agents.yaml b/config/agents.yaml index 1faca4d..edc7cd3 100644 --- a/config/agents.yaml +++ b/config/agents.yaml @@ -112,6 +112,7 @@ agents: - memory_write - system_status - shell + - delegate_to_kimi prompt: | You are Timmy, a sovereign local AI orchestrator. Primary interface between the user and the agent swarm. diff --git a/src/timmy/tools.py b/src/timmy/tools.py index e9d53ca..f4c9090 100644 --- a/src/timmy/tools.py +++ b/src/timmy/tools.py @@ -590,9 +590,10 @@ def create_full_toolkit(base_dir: str | Path | None = None): # Inter-agent delegation - dispatch tasks to swarm agents try: - from timmy.tools_delegation import delegate_task, list_swarm_agents + from timmy.tools_delegation import delegate_task, delegate_to_kimi, list_swarm_agents toolkit.register(delegate_task, name="delegate_task") + toolkit.register(delegate_to_kimi, name="delegate_to_kimi") toolkit.register(list_swarm_agents, name="list_swarm_agents") except Exception as exc: logger.warning("Tool execution failed (Delegation tools registration): %s", exc) diff --git a/src/timmy/tools_delegation/__init__.py b/src/timmy/tools_delegation/__init__.py index 8221f08..dce03c5 100644 --- a/src/timmy/tools_delegation/__init__.py +++ b/src/timmy/tools_delegation/__init__.py @@ -87,3 +87,73 @@ def list_swarm_agents() -> dict[str, Any]: "error": str(e), "agents": [], } + + +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. + """ + import shutil + import subprocess + from pathlib import Path + + from config import settings + + kimi_path = shutil.which("kimi") + if not kimi_path: + return { + "success": False, + "error": "kimi CLI not found on PATH. Install with: pip install kimi-cli", + } + + workdir = working_directory or settings.repo_root + if not Path(workdir).is_dir(): + return { + "success": False, + "error": f"Working directory does not exist: {workdir}", + } + + cmd = [kimi_path, "--print", "-p", task] + + logger.info("Delegating to Kimi: %s (cwd=%s)", task[:80], workdir) + + 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}", + } diff --git a/tests/timmy/test_tools_delegation.py b/tests/timmy/test_tools_delegation.py index 7607db9..30cd8c8 100644 --- a/tests/timmy/test_tools_delegation.py +++ b/tests/timmy/test_tools_delegation.py @@ -50,3 +50,103 @@ class TestListSwarmAgents: assert "Seer" in agent_names assert "Forge" in agent_names assert "Timmy" in agent_names + + +class TestDelegateToKimi: + """Tests for delegate_to_kimi() — Timmy's Kimi delegation tool.""" + + def test_returns_dict(self, monkeypatch): + """delegate_to_kimi should always return a dict.""" + import shutil + + monkeypatch.setattr(shutil, "which", lambda x: None) + from timmy.tools_delegation import delegate_to_kimi + + result = delegate_to_kimi("test task") + assert isinstance(result, dict) + assert "success" in result + + def test_kimi_not_found_returns_error(self, monkeypatch): + """Should handle missing kimi CLI gracefully.""" + import shutil + + monkeypatch.setattr(shutil, "which", lambda x: None) + from timmy.tools_delegation import delegate_to_kimi + + result = delegate_to_kimi("test task") + assert result["success"] is False + assert "not found" in result["error"] + + def test_invalid_workdir_returns_error(self, monkeypatch): + """Should reject non-existent working directories.""" + import shutil + + monkeypatch.setattr(shutil, "which", lambda x: "/usr/bin/kimi") + from timmy.tools_delegation import delegate_to_kimi + + result = delegate_to_kimi("test", working_directory="/nonexistent/path") + assert result["success"] is False + assert "does not exist" in result["error"] + + def test_timeout_returns_error(self, monkeypatch): + """Should handle subprocess timeout gracefully.""" + import shutil + import subprocess + + monkeypatch.setattr(shutil, "which", lambda x: "/usr/bin/kimi") + + def timeout_run(*args, **kwargs): + raise subprocess.TimeoutExpired(cmd="kimi", timeout=300) + + monkeypatch.setattr(subprocess, "run", timeout_run) + from timmy.tools_delegation import delegate_to_kimi + + result = delegate_to_kimi("complex task") + assert result["success"] is False + assert "timed out" in result["error"].lower() + + def test_successful_delegation(self, monkeypatch): + """Should capture Kimi's output on success.""" + import shutil + import subprocess + + monkeypatch.setattr(shutil, "which", lambda x: "/usr/bin/kimi") + + def mock_run(*args, **kwargs): + return subprocess.CompletedProcess( + args=args[0] if args else [], + returncode=0, + stdout="Fixed the bug in session.py\n\nChanges:\n- Added null check", + stderr="", + ) + + monkeypatch.setattr(subprocess, "run", mock_run) + from config import settings + from timmy.tools_delegation import delegate_to_kimi + + result = delegate_to_kimi("fix session bug", working_directory=settings.repo_root) + assert result["success"] is True + assert "Fixed the bug" in result["output"] + assert result["return_code"] == 0 + + def test_default_workdir_uses_repo_root(self, monkeypatch): + """Empty working_directory should default to settings.repo_root.""" + import shutil + import subprocess + + calls = [] + monkeypatch.setattr(shutil, "which", lambda x: "/usr/bin/kimi") + + def capture_run(*args, **kwargs): + calls.append(kwargs.get("cwd", "")) + return subprocess.CompletedProcess( + args=args[0] if args else [], returncode=0, stdout="done", stderr="" + ) + + monkeypatch.setattr(subprocess, "run", capture_run) + from config import settings + from timmy.tools_delegation import delegate_to_kimi + + delegate_to_kimi("test task") + assert len(calls) == 1 + assert calls[0] == settings.repo_root