feat(engines): add Kilocode engine adapter (#417)

.env.example

@@ -15,6 +15,20 @@ ANTHROPIC_API_KEY=sk-ant-api03-...
 # Optional for Sprint 1 - only needed if using GPT-4 agents
 OPENAI_API_KEY=sk-...
 
+# ============================================================================
+# External Engine Configuration
+# ============================================================================
+
+# Kilocode engine (cf work start --engine kilocode)
+# Path to the kilo binary; defaults to 'kilo' on $PATH
+# KILOCODE_PATH=kilo
+
+# Optional model override passed to kilo --model
+# KILOCODE_MODEL=
+
+# Optional extra CLI flags passed to kilo (space-separated)
+# KILOCODE_FLAGS=
+
 # ============================================================================
 # Database Configuration
 # ============================================================================

codeframe/cli/app.py

@@ -2306,7 +2306,7 @@ def work_start(
     engine: Optional[str] = typer.Option(
         None,
         "--engine",
-        help="Agent engine: react (default), plan (legacy), claude-code, codex, opencode, or built-in",
+        help="Agent engine: react (default), plan (legacy), claude-code, codex, opencode, kilocode, or built-in",
     ),
     stall_timeout: int = typer.Option(
         300,
@@ -3537,7 +3537,7 @@ def batch_run(
     engine: Optional[str] = typer.Option(
         None,
         "--engine",
-        help="Agent engine: react (default), plan (legacy), claude-code, codex, opencode, or built-in",
+        help="Agent engine: react (default), plan (legacy), claude-code, codex, opencode, kilocode, or built-in",
     ),
     stall_timeout: int = typer.Option(
         300,

codeframe/core/adapters/kilocode.py

@@ -0,0 +1,126 @@
+"""Kilocode adapter for delegating task execution to the kilo CLI."""
+
+from __future__ import annotations
+
+import os
+import shlex
+import shutil
+from pathlib import Path
+
+from codeframe.core.adapters.agent_adapter import AgentResult
+from codeframe.core.adapters.subprocess_adapter import SubprocessAdapter
+
+# Exit code used by kilo when the timeout is exceeded
+_KILO_TIMEOUT_EXIT_CODE = 124
+
+
+class KilocodeAdapter(SubprocessAdapter):
+    """Adapter that delegates code execution to Kilocode CLI.
+
+    Invokes ``kilo run <prompt> --auto --workspace <path>`` for headless
+    non-interactive execution.  The prompt is passed as a positional argument
+    (not via stdin), matching Kilocode's CLI interface.
+
+    Note on prompt length: the prompt is passed as a single positional argument.
+    Linux supports up to ~2 MB per argument, but macOS caps individual arguments
+    at 256 KB. Very large task contexts assembled by TaskContextPackager may fail
+    on macOS. If Kilocode adds stdin support in a future release, prefer that path.
+
+    Exit codes:
+        0   — success
+        124 — timeout exceeded (mirrors the standard ``timeout(1)`` convention)
+        *   — execution error
+
+    Configuration via environment variables:
+        KILOCODE_PATH   — path to kilo binary (default: "kilo", resolved from $PATH)
+        KILOCODE_MODEL  — optional model override passed as ``--model``
+        KILOCODE_FLAGS  — optional extra CLI flags (shell-quoted, e.g. ``--flag "val"``)
+
+    Requires Kilocode to be installed:
+    https://kilocode.ai/
+    """
+
+    def __init__(
+        self,
+        *,
+        timeout_s: int | None = None,
+    ) -> None:
+        super().__init__(binary=self._resolve_binary(), timeout_s=timeout_s)
+
+    @property
+    def name(self) -> str:  # noqa: D102
+        return "kilocode"
+
+    @staticmethod
+    def _resolve_binary() -> str:
+        """Return the kilo binary path from env or default."""
+        return os.environ.get("KILOCODE_PATH") or "kilo"
+
+    @classmethod
+    def requirements(cls) -> dict[str, str]:
+        """Return environment variables recognised by ``cf engines check``."""
+        return {
+            "KILOCODE_PATH": "Path to kilo binary (optional — defaults to 'kilo' on $PATH)",
+        }
+
+    @classmethod
+    def check_ready(cls) -> dict[str, bool]:
+        """Check if the kilo binary is available on PATH."""
+        return {"kilo_binary": shutil.which(cls._resolve_binary()) is not None}
+
+    def build_command(self, prompt: str, workspace_path: Path) -> list[str]:
+        """Build the kilo CLI command.
+
+        Kilocode takes the prompt as a positional argument, with ``--auto``
+        for non-interactive execution and ``--workspace`` for the repo root.
+
+        Args:
+            prompt: The task prompt passed as a positional argument.
+            workspace_path: Workspace root passed as ``--workspace``.
+
+        Returns:
+            Command list for subprocess.Popen.
+        """
+        cmd = [
+            self._binary_path,
+            "run",
+            prompt,
+            "--auto",
+            "--workspace",
+            str(workspace_path),
+        ]
+
+        model = os.environ.get("KILOCODE_MODEL")
+        if model:
+            cmd.extend(["--model", model])
+
+        extra_flags_str = os.environ.get("KILOCODE_FLAGS", "").strip()
+        if extra_flags_str:
+            cmd.extend(shlex.split(extra_flags_str))
+
+        return cmd
+
+    def get_stdin(self, prompt: str) -> str | None:
+        """Return None — prompt is passed as a positional CLI argument, not stdin."""
+        return None
+
+    def _map_result(
+        self,
+        exit_code: int,
+        stdout: str,
+        stderr: str,
+        workspace_path: Path,
+    ) -> AgentResult:
+        """Map kilo exit codes to AgentResult.
+
+        Exit code 124 indicates a timeout (kilo's standard timeout sentinel),
+        which is surfaced as a failed result with a descriptive message.
+        All other non-zero codes use the base class logic for blocker detection.
+        """
+        if exit_code == _KILO_TIMEOUT_EXIT_CODE:
+            return AgentResult(
+                status="failed",
+                output=stdout,
+                error="Kilocode execution timed out (exit code 124)",
+            )
+        return super()._map_result(exit_code, stdout, stderr, workspace_path)

codeframe/core/engine_registry.py

@@ -15,6 +15,7 @@
     "claude-code",
     "codex",
     "opencode",
+    "kilocode",
     "built-in",  # Alias for "react"
 })
 
@@ -23,6 +24,7 @@
     "claude-code",
     "codex",
     "opencode",
+    "kilocode",
 })
 
 # Builtin engines that need workspace + LLM provider
@@ -96,6 +98,10 @@ def get_external_adapter(engine: str, **kwargs: Any) -> AgentAdapter:
         from codeframe.core.adapters.opencode import OpenCodeAdapter
 
         return OpenCodeAdapter()
+    elif engine == "kilocode":
+        from codeframe.core.adapters.kilocode import KilocodeAdapter
+
+        return KilocodeAdapter(**kwargs)
     else:
         raise ValueError(
             f"Unknown external engine '{engine}'. "
@@ -201,6 +207,9 @@ def _get_adapter_class(engine: str) -> type | None:
     elif engine == "opencode":
         from codeframe.core.adapters.opencode import OpenCodeAdapter
         return OpenCodeAdapter
+    elif engine == "kilocode":
+        from codeframe.core.adapters.kilocode import KilocodeAdapter
+        return KilocodeAdapter
     return None
 
 

tests/core/adapters/test_kilocode.py

@@ -0,0 +1,185 @@
+"""Tests for Kilocode adapter."""
+
+from pathlib import Path
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from codeframe.core.adapters.agent_adapter import AgentAdapter
+from codeframe.core.adapters.kilocode import KilocodeAdapter
+
+pytestmark = pytest.mark.v2
+
+_WHICH = "codeframe.core.adapters.subprocess_adapter.shutil.which"
+_WHICH_KILOCODE = "codeframe.core.adapters.kilocode.shutil.which"
+
+
+class TestKilocodeAdapter:
+    """Unit tests for KilocodeAdapter."""
+
+    @pytest.fixture(autouse=True)
+    def _no_git(self):
+        """Prevent _detect_modified_files from calling real git."""
+        with patch.object(KilocodeAdapter, "_detect_modified_files", return_value=[]):
+            yield
+
+    def test_name(self) -> None:
+        with patch(_WHICH, return_value="/usr/bin/kilo"):
+            adapter = KilocodeAdapter()
+            assert adapter.name == "kilocode"
+
+    def test_conforms_to_protocol(self) -> None:
+        with patch(_WHICH, return_value="/usr/bin/kilo"):
+            adapter = KilocodeAdapter()
+            assert isinstance(adapter, AgentAdapter)
+
+    def test_raises_if_kilo_not_installed(self) -> None:
+        with patch(_WHICH, return_value=None):
+            with pytest.raises(EnvironmentError, match="not found on PATH"):
+                KilocodeAdapter()
+
+    def test_build_command_includes_prompt_and_auto_flag(self) -> None:
+        with patch(_WHICH, return_value="/usr/bin/kilo"):
+            adapter = KilocodeAdapter()
+        cmd = adapter.build_command("do the thing", Path("/tmp/repo"))
+        assert cmd[0] == "/usr/bin/kilo"
+        assert cmd[1] == "run"
+        assert "do the thing" in cmd
+        assert "--auto" in cmd
+        assert "--workspace" in cmd
+        assert "/tmp/repo" in cmd
+
+    def test_prompt_is_not_sent_via_stdin(self) -> None:
+        with patch(_WHICH, return_value="/usr/bin/kilo"):
+            adapter = KilocodeAdapter()
+        assert adapter.get_stdin("my prompt") is None
+
+    def test_build_command_includes_model_when_env_set(self, monkeypatch) -> None:
+        monkeypatch.setenv("KILOCODE_MODEL", "claude-3-5-sonnet")
+        with patch(_WHICH, return_value="/usr/bin/kilo"):
+            adapter = KilocodeAdapter()
+        cmd = adapter.build_command("prompt", Path("/tmp/repo"))
+        assert "--model" in cmd
+        idx = cmd.index("--model")
+        assert cmd[idx + 1] == "claude-3-5-sonnet"
+
+    def test_build_command_extra_flags_uses_shlex(self, monkeypatch) -> None:
+        """KILOCODE_FLAGS must be split with shlex to handle quoted values."""
+        monkeypatch.setenv("KILOCODE_FLAGS", '--verbose --log-level "debug mode"')
+        with patch(_WHICH, return_value="/usr/bin/kilo"):
+            adapter = KilocodeAdapter()
+        cmd = adapter.build_command("prompt", Path("/tmp/repo"))
+        assert "--verbose" in cmd
+        assert "--log-level" in cmd
+        # shlex preserves quoted string as a single token
+        assert "debug mode" in cmd
+
+    def test_build_command_extra_flags_simple(self, monkeypatch) -> None:
+        monkeypatch.setenv("KILOCODE_FLAGS", "--verbose --log-level debug")
+        with patch(_WHICH, return_value="/usr/bin/kilo"):
+            adapter = KilocodeAdapter()
+        cmd = adapter.build_command("prompt", Path("/tmp/repo"))
+        assert "--verbose" in cmd
+        assert "--log-level" in cmd
+        assert "debug" in cmd
+
+    def test_custom_binary_from_env(self, monkeypatch) -> None:
+        monkeypatch.setenv("KILOCODE_PATH", "/opt/kilo/bin/kilo")
+        with patch(_WHICH, return_value="/opt/kilo/bin/kilo"):
+            adapter = KilocodeAdapter()
+        assert adapter._binary_path == "/opt/kilo/bin/kilo"
+
+    def test_resolve_binary_uses_env_var(self, monkeypatch) -> None:
+        monkeypatch.setenv("KILOCODE_PATH", "/custom/kilo")
+        assert KilocodeAdapter._resolve_binary() == "/custom/kilo"
+
+    def test_resolve_binary_defaults_to_kilo(self, monkeypatch) -> None:
+        monkeypatch.delenv("KILOCODE_PATH", raising=False)
+        assert KilocodeAdapter._resolve_binary() == "kilo"
+
+    def test_check_ready_when_binary_present(self) -> None:
+        with patch(_WHICH_KILOCODE, return_value="/usr/bin/kilo"):
+            result = KilocodeAdapter.check_ready()
+        assert result["kilo_binary"] is True
+
+    def test_check_ready_when_binary_missing(self) -> None:
+        with patch(_WHICH_KILOCODE, return_value=None):
+            result = KilocodeAdapter.check_ready()
+        assert result["kilo_binary"] is False
+
+    def test_requirements_returns_kilocode_path_key(self) -> None:
+        reqs = KilocodeAdapter.requirements()
+        assert "KILOCODE_PATH" in reqs
+
+    def test_successful_execution(self) -> None:
+        with patch(_WHICH, return_value="/usr/bin/kilo"):
+            adapter = KilocodeAdapter()
+
+        mock_process = MagicMock()
+        mock_process.stdout = iter(["Wrote src/foo.py\n"])
+        mock_process.stderr = MagicMock()
+        mock_process.stderr.read.return_value = ""
+        mock_process.stdin = None
+        mock_process.returncode = 0
+        mock_process.wait.return_value = None
+
+        with patch("subprocess.Popen", return_value=mock_process):
+            result = adapter.run("task-1", "implement foo", Path("/tmp/repo"))
+
+        assert result.status == "completed"
+        assert "Wrote src/foo.py" in result.output
+
+    def test_failed_execution_nonzero_exit(self) -> None:
+        with patch(_WHICH, return_value="/usr/bin/kilo"):
+            adapter = KilocodeAdapter()
+
+        mock_process = MagicMock()
+        mock_process.stdout = iter([])
+        mock_process.stderr = MagicMock()
+        mock_process.stderr.read.return_value = "kilo: fatal error"
+        mock_process.stdin = None
+        mock_process.returncode = 1
+        mock_process.wait.return_value = None
+
+        with patch("subprocess.Popen", return_value=mock_process):
+            result = adapter.run("task-1", "implement foo", Path("/tmp/repo"))
+
+        assert result.status == "failed"
+
+    def test_timeout_exit_code_124_maps_to_failed(self) -> None:
+        with patch(_WHICH, return_value="/usr/bin/kilo"):
+            adapter = KilocodeAdapter()
+
+        mock_process = MagicMock()
+        mock_process.stdout = iter([])
+        mock_process.stderr = MagicMock()
+        mock_process.stderr.read.return_value = ""
+        mock_process.stdin = None
+        mock_process.returncode = 124
+        mock_process.wait.return_value = None
+
+        with patch("subprocess.Popen", return_value=mock_process):
+            result = adapter.run("task-1", "implement foo", Path("/tmp/repo"))
+
+        assert result.status == "failed"
+        assert "timed out" in (result.error or "").lower()
+
+    def test_event_callback_receives_output_lines(self) -> None:
+        with patch(_WHICH, return_value="/usr/bin/kilo"):
+            adapter = KilocodeAdapter()
+
+        events: list = []
+        mock_process = MagicMock()
+        mock_process.stdout = iter(["step 1\n", "step 2\n"])
+        mock_process.stderr = MagicMock()
+        mock_process.stderr.read.return_value = ""
+        mock_process.stdin = None
+        mock_process.returncode = 0
+        mock_process.wait.return_value = None
+
+        with patch("subprocess.Popen", return_value=mock_process):
+            adapter.run("task-1", "do work", Path("/tmp/repo"), on_event=events.append)
+
+        assert len(events) == 2
+        assert events[0].data["line"] == "step 1"
+        assert events[1].data["line"] == "step 2"