feat(sleep): add kimicode backend (Kimi Code CLI)

skillopt_sleep/__main__.py

@@ -101,12 +101,13 @@ def _add_common(p: argparse.ArgumentParser) -> None:
     p.add_argument("--scope", default="", choices=["", "all", "invoked"])
     p.add_argument("--backend", default="",
                    choices=["", "mock", "claude", "codex", "copilot", "cursor", "pi",
-                            "opencode", "handoff", "azure_openai"])
+                            "opencode", "kimicode", "handoff", "azure_openai"])
     p.add_argument("--model", default="")
     p.add_argument("--codex-path", default="", help="path to the real @openai/codex binary")
     p.add_argument("--cursor-path", default="", help="path to the Cursor Agent CLI")
     p.add_argument("--pi-path", default="", help="path to the Pi coding-agent CLI")
     p.add_argument("--opencode-path", default="", help="path to the OpenCode CLI")
+    p.add_argument("--kimi-path", default="", help="path to the Kimi Code CLI (`kimi`)")
     p.add_argument(
         "--opencode-tool-replay",
         action="store_true",
@@ -170,6 +171,8 @@ def _cfg_from_args(args, task_meta: Dict[str, Any] | None = None) -> Any:
         overrides["cursor_path"] = os.path.abspath(os.path.expanduser(args.cursor_path))
     if getattr(args, "opencode_path", ""):
         overrides["opencode_path"] = os.path.abspath(os.path.expanduser(args.opencode_path))
+    if getattr(args, "kimi_path", ""):
+        overrides["kimi_path"] = os.path.abspath(os.path.expanduser(args.kimi_path))
     if getattr(args, "opencode_tool_replay", False):
         overrides["opencode_tool_replay"] = True
     if getattr(args, "claude_home", ""):

skillopt_sleep/backend.py

@@ -1153,6 +1153,97 @@ def _prepare_opencode_replay_project(work: str, tools: List[str]) -> _OpenCodeRe
         raise OpenCodeError("OpenCode CLI tool replay could not be prepared") from None
 
 
+class KimiCliBackend(CliBackend):
+    """Drives the authenticated `kimi` CLI (Kimi Code): kimi -p --output-format text."""
+
+    name = "kimicode"
+
+    def __init__(self, model: str = "", kimi_path: str = "kimi", timeout: int = 0) -> None:
+        # Cold starts run ~60s unloaded but can exceed 180s when the Kimi
+        # desktop app is busy on the same account/machine; allow an env
+        # override and default to a generous 600s rather than the 180s the
+        # other backends use.
+        if not timeout:
+            timeout = int(os.environ.get("SKILLOPT_SLEEP_KIMI_TIMEOUT", "600") or "600")
+        super().__init__(model=model or os.environ.get("SKILLOPT_SLEEP_KIMI_MODEL", ""),
+                         timeout=timeout)
+        import shutil as _shutil
+        resolved = _shutil.which(kimi_path)
+        if not resolved and kimi_path in ("kimi", "kimi.exe", "kimi.cmd"):
+            # Only the bare default name falls back to the desktop install
+            # location; an explicit custom path is kept as-is even if missing
+            # (better a clear spawn error than a silent wrong binary).
+            candidate = os.path.join(os.path.expanduser("~"), ".kimi-code", "bin",
+                                     "kimi.exe" if os.name == "nt" else "kimi")
+            resolved = candidate if os.path.exists(candidate) else kimi_path
+        self.kimi_path = resolved or kimi_path
+
+    _CLI_ERROR_MARKERS = (
+        "not logged in",
+        "please run /login",
+        "authentication required",
+        "invalid api key",
+        "unauthorized",
+        "401",
+    )
+
+    def _call(self, prompt: str, *, max_tokens: int = 1024) -> str:
+        # Isolated cwd so no project AGENTS.md leaks into the optimizer call.
+        # Note: the user's global ~/.kimi-code/AGENTS.md is still injected by
+        # the CLI (there is no --bare equivalent); treat that as ambient memory.
+        import tempfile
+        # NOTE: kimi's -p/--prompt takes the prompt INLINE (unlike claude -p
+        # which reads stdin); `--output-format text` alone would be swallowed
+        # as the prompt value and the real prompt would error as a command.
+        cmd = [self.kimi_path, "--prompt", prompt, "--output-format", "text"]
+        if self.model:
+            cmd += ["-m", self.model]
+        clean_cwd = tempfile.mkdtemp(prefix="skillopt_sleep_kimi_")
+        try:
+            proc = subprocess.run(
+                cmd, capture_output=True, creationflags=_NO_WINDOW,
+                timeout=self.timeout, cwd=clean_cwd,
+            )
+        except subprocess.TimeoutExpired:
+            import logging
+            self.last_call_error = f"Kimi CLI timed out after {self.timeout}s"
+            logging.getLogger("skillopt_sleep").warning(
+                "Kimi CLI call timed out after %ss; consider SKILLOPT_SLEEP_KIMI_TIMEOUT",
+                self.timeout,
+            )
+            return ""
+        except Exception as exc:
+            import logging
+            self.last_call_error = f"Kimi CLI spawn failed: {exc}"
+            logging.getLogger("skillopt_sleep").warning(
+                "Kimi CLI could not be executed: %s", exc,
+            )
+            return ""
+        finally:
+            try:
+                import shutil
+                shutil.rmtree(clean_cwd, ignore_errors=True)
+            except OSError:
+                pass
+        # The CLI emits UTF-8 regardless of the Windows console codepage;
+        # decoding with the locale codec (gbk) crashes on multibyte output.
+        out = (proc.stdout or b"").decode("utf-8", errors="replace").strip()
+        stderr_text = (proc.stderr or b"").decode("utf-8", errors="replace")
+        check = out if len(out) < 300 else ""
+        combined = check + "\n" + stderr_text
+        for marker in self._CLI_ERROR_MARKERS:
+            if marker in combined.lower():
+                from skillopt_sleep.staging import redact_secrets
+                import logging
+                self.last_call_error = combined[:500]
+                logging.getLogger("skillopt_sleep").warning(
+                    "Kimi CLI returned a likely auth error: %s",
+                    redact_secrets(combined[:200].replace("\n", " ")),
+                )
+                break
+        return out
+
+
 def _opencode_temporary_workspace(prefix: str, error: str) -> tempfile.TemporaryDirectory:
     try:
         return tempfile.TemporaryDirectory(prefix=prefix, ignore_cleanup_errors=True)
@@ -2715,6 +2806,7 @@ def get_backend(
     pi_path: str = "",
     cursor_path: str = "",
     opencode_path: str = "",
+    kimi_path: str = "",
     opencode_tool_replay: bool = False,
     azure_endpoint: str = "",
     project_dir: str = "",
@@ -2735,6 +2827,8 @@ def get_backend(
         return CopilotCliBackend(model=model)
     if n in {"cursor", "cursor_agent", "cursor_cli"}:
         return CursorCliBackend(model=model, cursor_path=cursor_path)
+    if n in {"kimi", "kimicode", "kimi_code", "kimi-cli", "kimi_cli"}:
+        return KimiCliBackend(model=model, kimi_path=kimi_path or "kimi")
     if n in {"opencode", "opencode_cli", "opencode-cli"}:
         return OpenCodeCliBackend(
             model=model,
@@ -2763,6 +2857,7 @@ def build_backend(
     pi_path: str = "",
     cursor_path: str = "",
     opencode_path: str = "",
+    kimi_path: str = "",
     opencode_tool_replay: bool = False,
     azure_endpoint: str = "",
     preferences: str = "",
@@ -2784,6 +2879,7 @@ def build_backend(
             pi_path=pi_path,
             cursor_path=cursor_path,
             opencode_path=opencode_path,
+            kimi_path=kimi_path,
             opencode_tool_replay=opencode_tool_replay,
             azure_endpoint=azure_endpoint,
             project_dir=project_dir,

skillopt_sleep/config.py

@@ -62,6 +62,7 @@
     "pi_path": "",                # "" => use `pi` on PATH
     "cursor_path": "",            # "" => auto-detect the Cursor Agent CLI
     "opencode_path": "",          # "" => SKILLOPT_SLEEP_OPENCODE_PATH, then `opencode` on PATH/PATHEXT
+    "kimi_path": "",              # "" => `kimi` on PATH, then ~/.kimi-code/bin/kimi
     "opencode_tool_replay": False,  # explicit opt-in for OpenCode tool-aware replay
     "edit_budget": 4,             # textual learning rate (max edits/night)
     "preferences": "",            # free-text house rules injected into reflect as a prior

skillopt_sleep/cycle.py

@@ -139,6 +139,7 @@ def _make_model_key(cfg: SleepConfig) -> str:
             pi_path=cfg.get("pi_path", ""),
             cursor_path=cfg.get("cursor_path", ""),
             opencode_path=cfg.get("opencode_path", ""),
+            kimi_path=cfg.get("kimi_path", ""),
             opencode_tool_replay=cfg.get("opencode_tool_replay", False),
             azure_endpoint=cfg.get("azure_endpoint", ""),
             project_dir=cfg.get("invoked_project", "") or os.getcwd(),
@@ -691,6 +692,7 @@ def run_sleep_cycle(
         pi_path=cfg.get("pi_path", ""),
         cursor_path=cfg.get("cursor_path", ""),
         opencode_path=cfg.get("opencode_path", ""),
+        kimi_path=cfg.get("kimi_path", ""),
         opencode_tool_replay=cfg.get("opencode_tool_replay", False),
         azure_endpoint=cfg.get("azure_endpoint", ""),
         preferences=cfg.get("preferences", ""),

tests/test_backend_kimi.py

@@ -0,0 +1,122 @@
+"""Tests for the Kimi Code CLI backend (`--backend kimicode`)."""
+from __future__ import annotations
+
+import os
+import subprocess
+from unittest import mock
+
+from skillopt_sleep.backend import (
+    _NO_WINDOW,
+    KimiCliBackend,
+    build_backend,
+    get_backend,
+)
+
+
+class _FakeProc:
+    def __init__(self, stdout: bytes, stderr: bytes = b"", returncode: int = 0):
+        self.stdout = stdout
+        self.stderr = stderr
+        self.returncode = returncode
+
+
+def test_get_backend_kimi_aliases():
+    for alias in ("kimi", "kimicode", "kimi_code", "kimi-cli", "kimi_cli", "KIMICODE"):
+        be = get_backend(alias, model="k2")
+        assert isinstance(be, KimiCliBackend), alias
+        assert be.name == "kimicode"
+        assert be.model == "k2"
+
+
+def test_get_backend_kimi_path_override():
+    be = get_backend("kimicode", kimi_path="/opt/kimi/bin/kimi")
+    assert isinstance(be, KimiCliBackend)
+    assert be.kimi_path == "/opt/kimi/bin/kimi"
+
+
+def test_default_model_from_env(monkeypatch):
+    monkeypatch.setenv("SKILLOPT_SLEEP_KIMI_MODEL", "kimi-for-coding")
+    be = KimiCliBackend()
+    assert be.model == "kimi-for-coding"
+
+
+def test_path_falls_back_to_desktop_install(monkeypatch):
+    # PATH lookup fails -> the desktop install location is used when it exists
+    with mock.patch("shutil.which", return_value=None):
+        with mock.patch("os.path.exists", return_value=True):
+            be = KimiCliBackend(kimi_path="kimi")
+    assert be.kimi_path.endswith(os.path.join(".kimi-code", "bin", "kimi.exe"))
+
+
+def test_call_builds_isolated_command_and_decodes_utf8():
+    be = KimiCliBackend(model="k2", kimi_path="/usr/local/bin/kimi")
+    captured = {}
+
+    def fake_run(cmd, **kwargs):
+        captured["cmd"] = cmd
+        captured.update(kwargs)
+        return _FakeProc("• OK".encode("utf-8"))
+
+    with mock.patch("skillopt_sleep.backend.subprocess.run", side_effect=fake_run):
+        out = be._call("Reply with exactly: OK")
+
+    assert out == "• OK"
+    cmd = captured["cmd"]
+    assert cmd[0] == "/usr/local/bin/kimi"
+    # kimi's --prompt takes the prompt INLINE (unlike claude -p via stdin):
+    # --output-format must not be swallowed as the prompt value.
+    assert "--prompt" in cmd
+    assert cmd[cmd.index("--prompt") + 1] == "Reply with exactly: OK"
+    assert cmd[cmd.index("--output-format") + 1] == "text"
+    assert cmd[cmd.index("-m") + 1] == "k2"
+    # isolated cwd, no console window, no stdin prompt
+    assert captured["cwd"] and "skillopt_sleep_kimi_" in captured["cwd"]
+    assert captured["creationflags"] == _NO_WINDOW
+    assert "input" not in captured
+
+
+def test_call_decodes_utf8_despite_gbk_locale():
+    be = KimiCliBackend(kimi_path="/usr/local/bin/kimi")
+    with mock.patch(
+        "skillopt_sleep.backend.subprocess.run",
+        return_value=_FakeProc("中文回答".encode("utf-8")),
+    ):
+        out = be._call("question")
+    assert out == "中文回答"
+
+
+def test_call_flags_auth_error(monkeypatch):
+    be = KimiCliBackend(kimi_path="/usr/local/bin/kimi")
+    with mock.patch(
+        "skillopt_sleep.backend.subprocess.run",
+        return_value=_FakeProc(b"", stderr="Error: Invalid API key".encode("utf-8")),
+    ):
+        out = be._call("question")
+    assert out == ""
+    assert "Invalid API key" in be.last_call_error
+
+
+def test_build_backend_routes_kimi_path():
+    be = build_backend(backend="kimicode", kimi_path="/opt/kimi/bin/kimi")
+    assert isinstance(be, KimiCliBackend)
+    assert be.kimi_path == "/opt/kimi/bin/kimi"
+
+
+def test_default_timeout_is_generous_and_env_overridable(monkeypatch):
+    monkeypatch.delenv("SKILLOPT_SLEEP_KIMI_TIMEOUT", raising=False)
+    be = KimiCliBackend(kimi_path="kimi")
+    assert be.timeout == 600  # cold starts under desktop-app load exceed 180s
+    monkeypatch.setenv("SKILLOPT_SLEEP_KIMI_TIMEOUT", "900")
+    be = KimiCliBackend(kimi_path="kimi")
+    assert be.timeout == 900
+
+
+def test_timeout_is_reported_distinctly(monkeypatch):
+    be = KimiCliBackend(kimi_path="/usr/local/bin/kimi")
+    with mock.patch(
+        "skillopt_sleep.backend.subprocess.run",
+        side_effect=subprocess.TimeoutExpired(cmd="kimi", timeout=be.timeout),
+    ):
+        out = be._call("question")
+    assert out == ""
+    assert "timed out" in be.last_call_error