Compare commits

...

1 Commits

Author SHA1 Message Date
kimi
bf57da87b3 refactor: break up delegate_to_kimi into helpers
Extract _find_kimi_cli, _resolve_workdir, and _run_kimi from the 68-line
delegate_to_kimi function, reducing it to ~15 lines.

Fixes #635

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
2026-03-20 16:54:41 -04:00

View File

@@ -89,52 +89,41 @@ def list_swarm_agents() -> dict[str, Any]:
} }
def delegate_to_kimi(task: str, working_directory: str = "") -> dict[str, Any]: def _find_kimi_cli() -> str | None:
"""Delegate a coding task to Kimi, the external coding agent. """Return the path to the kimi CLI, or None if not installed."""
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 shutil
import subprocess
return shutil.which("kimi")
def _resolve_workdir(working_directory: str) -> str | dict[str, Any]:
"""Resolve and validate the working directory.
Returns the resolved path string, or an error dict if invalid.
"""
from pathlib import Path from pathlib import Path
from config import settings 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 workdir = working_directory or settings.repo_root
if not Path(workdir).is_dir(): if not Path(workdir).is_dir():
return { return {
"success": False, "success": False,
"error": f"Working directory does not exist: {workdir}", "error": f"Working directory does not exist: {workdir}",
} }
return workdir
cmd = [kimi_path, "--print", "-p", task]
logger.info("Delegating to Kimi: %s (cwd=%s)", task[:80], workdir) def _run_kimi(cmd: list[str], workdir: str) -> dict[str, Any]:
"""Execute the kimi subprocess and return a result dict."""
import subprocess
try: try:
result = subprocess.run( result = subprocess.run(
cmd, cmd,
capture_output=True, capture_output=True,
text=True, text=True,
timeout=300, # 5 minute timeout for coding tasks timeout=300,
cwd=workdir, cwd=workdir,
) )
@@ -157,3 +146,34 @@ def delegate_to_kimi(task: str, working_directory: str = "") -> dict[str, Any]:
"success": False, "success": False,
"error": f"Failed to run Kimi: {exc}", "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)