feat: add delegate_to_kimi() tool for coding delegation (#67)

Timmy can now delegate coding tasks to Kimi CLI (262K context).
Includes timeout handling, workdir validation, output truncation.
Sovereign division of labor — Timmy plans, Kimi codes.
This commit is contained in:
2026-03-14 20:29:03 -04:00
parent c164d1736f
commit 453c9a0694
4 changed files with 173 additions and 1 deletions

View File

@@ -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.

View File

@@ -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)

View File

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

View File

@@ -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