diff --git a/AGENTS.md b/AGENTS.md index 29d7fc3..6b3adf5 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -22,6 +22,7 @@ This repository owns the published `elydora` package, its CLI, synchronous and a - Use the provider's documented blocking mechanism. Command-hook providers that define exit code `2` must receive exit code `2` from the freeze guard. - Report installation as healthy only when a complete hook contract references both generated runtime scripts and both scripts exist. - Model stable, legacy, and early-access hook generations as explicit contracts. Keep their activation requirements visible in CLI output and README guidance. +- Select Kimi Code and legacy kimi-cli config targets from existing runtime evidence; a fresh installation targets current Kimi Code and keeps migration markers isolated to active runtimes. ## Code Quality diff --git a/README.md b/README.md index ea800cb..b3460ba 100644 --- a/README.md +++ b/README.md @@ -75,6 +75,8 @@ elydora install \ Codex performs a one-time trust review for user hooks. Run `/hooks` in Codex after installation and trust the Elydora `PreToolUse` and `PostToolUse` definitions. +Kimi installation writes the strict hook contract to each detected runtime: Kimi Code's `$KIMI_CODE_HOME/config.toml` (default `~/.kimi-code/config.toml`) and the migrating Python CLI's `~/.kimi/config.toml`. A fresh installation targets current Kimi Code, avoiding cross-runtime migration markers. Both runtimes load the hooks globally; run `/hooks` to inspect them. + Kiro CLI installation covers both runtime contracts. Kiro CLI v2 uses the generated custom agent through `kiro-cli --agent elydora-audit`. Kiro CLI v3 loads the global standalone hooks when started with `kiro-cli --v3`. ### Commands @@ -92,6 +94,7 @@ Kiro CLI installation covers both runtime contracts. Kiro CLI v2 uses the genera |-------|-----| | Claude Code | `claudecode` | | OpenAI Codex | `codex` | +| Kimi Code | `kimi` | | Copilot CLI | `copilot` | | Cursor | `cursor` | | Gemini CLI | `gemini` | diff --git a/elydora/cli.py b/elydora/cli.py index 899c12c..88e4f42 100644 --- a/elydora/cli.py +++ b/elydora/cli.py @@ -25,6 +25,7 @@ from .plugins.cursor import CursorPlugin from .plugins.gemini import GeminiPlugin from .plugins.kirocli import KiroCliPlugin from .plugins.kiroide import KiroIdePlugin +from .plugins.kimi import KimiPlugin from .plugins.letta import LettaPlugin from .plugins.opencode import OpenCodePlugin @@ -37,6 +38,7 @@ PLUGIN_MAP: Dict[str, Type[AgentPlugin]] = { "gemini": GeminiPlugin, "kirocli": KiroCliPlugin, "kiroide": KiroIdePlugin, + "kimi": KimiPlugin, "letta": LettaPlugin, "opencode": OpenCodePlugin, } diff --git a/elydora/plugins/_file_io.py b/elydora/plugins/_file_io.py new file mode 100644 index 0000000..aa98325 --- /dev/null +++ b/elydora/plugins/_file_io.py @@ -0,0 +1,109 @@ +"""Fail-fast, atomic file helpers shared by hook adapters.""" + +from __future__ import annotations + +import json +import os +import stat +import tempfile +from typing import Any, Dict, Optional + + +JsonObject = Dict[str, Any] + + +def _cleanup_failed_write(path: str, label: str, cause: Exception) -> None: + if not path: + return + try: + os.remove(path) + except FileNotFoundError: + return + except OSError as cleanup_error: + raise OSError( + f"Write {label} failed: {cause}; cleanup of {path} failed: {cleanup_error}" + ) from cause + + +def write_text_atomic(path: str, content: str, mode: int, label: str) -> None: + directory = os.path.dirname(path) + try: + os.makedirs(directory, mode=0o700, exist_ok=True) + except OSError as error: + raise OSError(f"Create directory for {label} at {directory}: {error}") from error + + descriptor = -1 + temporary_path = "" + try: + descriptor, temporary_path = tempfile.mkstemp( + prefix=f".{os.path.basename(path)}.", + suffix=".tmp", + dir=directory, + text=True, + ) + with os.fdopen(descriptor, "w", encoding="utf-8", newline="") as file: + descriptor = -1 + file.write(content) + file.flush() + os.fsync(file.fileno()) + os.chmod(temporary_path, mode) + os.replace(temporary_path, path) + except Exception as error: + if descriptor >= 0: + try: + os.close(descriptor) + except OSError as close_error: + _cleanup_failed_write(temporary_path, label, close_error) + raise OSError( + f"Write {label} at {path} failed: {error}; " + f"close failed: {close_error}" + ) from error + _cleanup_failed_write(temporary_path, label, error) + raise OSError(f"Write {label} at {path}: {error}") from error + + +def write_json_atomic(path: str, value: JsonObject, mode: int, label: str) -> None: + write_text_atomic(path, json.dumps(value, indent=2) + "\n", mode, label) + + +def remove_file(path: str, label: str) -> None: + try: + os.remove(path) + except FileNotFoundError: + return + except OSError as error: + raise OSError(f"Remove {label} at {path}: {error}") from error + + +def regular_file_exists(path: str, label: str) -> bool: + try: + metadata = os.stat(path) + except FileNotFoundError: + return False + except OSError as error: + raise OSError(f"Read {label} at {path}: {error}") from error + return stat.S_ISREG(metadata.st_mode) + + +def require_runtime(path: str, label: str) -> None: + if not path: + raise ValueError(f"{label} path is required") + if not regular_file_exists(path, label): + raise FileNotFoundError(f"{label} is missing: {path}") + + +def read_json(path: str, label: str) -> Optional[JsonObject]: + try: + with open(path, "r", encoding="utf-8") as file: + raw = file.read() + except FileNotFoundError: + return None + except OSError as error: + raise OSError(f"Read {label} at {path}: {error}") from error + try: + value = json.loads(raw) + except json.JSONDecodeError as error: + raise ValueError(f"Failed to parse {label} at {path}: {error}") from error + if not isinstance(value, dict): + raise ValueError(f"{label} at {path} must contain a JSON object") + return value diff --git a/elydora/plugins/kimi.py b/elydora/plugins/kimi.py new file mode 100644 index 0000000..d831106 --- /dev/null +++ b/elydora/plugins/kimi.py @@ -0,0 +1,474 @@ +"""Kimi Code and legacy kimi-cli lifecycle-hook integration.""" + +from __future__ import annotations + +import copy +from dataclasses import dataclass +import os +import shlex +import shutil +# subprocess provides argument quoting only; this module starts no process. +import subprocess # nosec B404 +import sys +from typing import Any, Dict, FrozenSet, List, Optional + +from tomlkit import aot, document, dumps, inline_table, parse, table +from tomlkit.items import AoT, Array + +from .base import AgentPlugin, InstallConfig, PluginStatus +from ._file_io import ( + read_json as _read_json, + regular_file_exists as _regular_file_exists, + remove_file as _remove_file, + require_runtime as _require_runtime, + write_json_atomic as _write_json_atomic, + write_text_atomic as _write_text_atomic, +) +from .hook_template import generate_hook_script + + +AGENT_KEY = "kimi" +GUARD_SCRIPT = "guard.py" +AUDIT_SCRIPT = "hook.py" +HOOK_TIMEOUT_SECONDS = 10 +ELYDORA_DIR = os.path.join(os.path.expanduser("~"), ".elydora") + +SHARED_EVENTS: FrozenSet[str] = frozenset({ + "PreToolUse", + "PostToolUse", + "PostToolUseFailure", + "UserPromptSubmit", + "Stop", + "StopFailure", + "SessionStart", + "SessionEnd", + "SubagentStart", + "SubagentStop", + "PreCompact", + "PostCompact", + "Notification", +}) +MODERN_EVENTS = SHARED_EVENTS | { + "PermissionRequest", + "PermissionResult", + "Interrupt", +} +LEGACY_EVENTS = SHARED_EVENTS +SUPPORTED_FIELDS = frozenset({"event", "matcher", "command", "timeout"}) + +TomlObject = Dict[str, Any] + + +@dataclass(frozen=True) +class KimiContract: + runtime_name: str + label: str + config_path: str + events: FrozenSet[str] + + +@dataclass(frozen=True) +class KimiDocument: + contract: KimiContract + exists: bool + raw: str + value: Any + hooks: List[TomlObject] + + +@dataclass(frozen=True) +class RuntimeContract: + guard: str + audit: str + config_path: str + + +@dataclass(frozen=True) +class ConfigMutation: + kind: str + source: KimiDocument + raw: str = "" + + +def _home_dir() -> str: + return os.path.expanduser("~") + + +def _path_exists(path: str, label: str) -> bool: + try: + os.stat(path) + return True + except (FileNotFoundError, NotADirectoryError): + return False + except OSError as error: + raise OSError(f"Inspect {label} at {path}: {error}") from error + + +def _legacy_cli_on_path() -> bool: + try: + return shutil.which("kimi-cli") is not None + except OSError as error: + raise OSError(f"Inspect kimi-cli executable on PATH: {error}") from error + + +def _resolve_contracts() -> List[KimiContract]: + home = _home_dir() + explicit_home = os.environ.get("KIMI_CODE_HOME") or None + kimi_home = explicit_home or os.path.join(home, ".kimi-code") + modern = KimiContract( + runtime_name="Kimi Code", + label="Kimi Code hooks config", + config_path=os.path.join(kimi_home, "config.toml"), + events=MODERN_EVENTS, + ) + legacy = KimiContract( + runtime_name="kimi-cli", + label="kimi-cli legacy hooks config", + config_path=os.path.join(home, ".kimi", "config.toml"), + events=LEGACY_EVENTS, + ) + modern_detected = explicit_home is not None or _path_exists( + kimi_home, "Kimi Code home" + ) + legacy_detected = _path_exists( + legacy.config_path, legacy.label + ) or _legacy_cli_on_path() + if legacy_detected and not modern_detected: + return [legacy] + return [modern, legacy] if legacy_detected else [modern] + + +def _unwrap(value: Any) -> Any: + unwrap = getattr(value, "unwrap", None) + return unwrap() if callable(unwrap) else value + + +def _validate_hook(value: Any, contract: KimiContract, index: int) -> TomlObject: + raw = _unwrap(value) + if not isinstance(raw, dict): + raise ValueError(f"{contract.label} hook {index + 1} must be a table") + hook = dict(raw) + unsupported = next((key for key in hook if key not in SUPPORTED_FIELDS), None) + if unsupported is not None: + raise ValueError( + f'{contract.label} hook {index + 1} has unsupported field "{unsupported}"' + ) + event = hook.get("event") + if not isinstance(event, str) or event not in contract.events: + raise ValueError( + f'{contract.label} hook {index + 1} has unsupported event "{event}"' + ) + command = hook.get("command") + if not isinstance(command, str) or not command: + raise ValueError( + f"{contract.label} hook {index + 1} requires a non-empty command" + ) + matcher = hook.get("matcher") + if matcher is not None and not isinstance(matcher, str): + raise ValueError(f"{contract.label} hook {index + 1} matcher must be a string") + timeout = hook.get("timeout") + if timeout is not None and ( + isinstance(timeout, bool) + or not isinstance(timeout, int) + or timeout < 1 + or timeout > 600 + ): + raise ValueError( + f"{contract.label} hook {index + 1} timeout must be an integer from 1 to 600" + ) + return hook + + +def _read_hooks(value: Any, contract: KimiContract) -> List[TomlObject]: + if "hooks" not in value: + return [] + container = value["hooks"] + raw = _unwrap(container) + if not isinstance(raw, list): + raise ValueError(f'{contract.label} field "hooks" must be an array') + return [_validate_hook(item, contract, index) for index, item in enumerate(raw)] + + +def _read_config(contract: KimiContract) -> KimiDocument: + try: + with open( + contract.config_path, + "r", + encoding="utf-8", + newline="", + ) as file: + raw = file.read() + except FileNotFoundError: + value = document() + return KimiDocument(contract, False, "", value, []) + except OSError as error: + raise OSError( + f"Read {contract.label} at {contract.config_path}: {error}" + ) from error + + try: + value = parse(raw) + except Exception as error: + raise ValueError( + f"Failed to parse {contract.label} at {contract.config_path}: {error}" + ) from error + return KimiDocument(contract, True, raw, value, _read_hooks(value, contract)) + + +def _read_all_configs() -> List[KimiDocument]: + return [_read_config(contract) for contract in _resolve_contracts()] + + +def _build_command(script_path: str) -> str: + arguments = [sys.executable, script_path] + return ( + subprocess.list2cmdline(arguments) + if os.name == "nt" + else shlex.join(arguments) + ) + + +def _build_hook(event: str, script_path: str) -> TomlObject: + return { + "event": event, + "command": _build_command(script_path), + "timeout": HOOK_TIMEOUT_SECONDS, + } + + +def _render_hooks( + source: KimiDocument, + keep_indices: List[int], + additions: List[TomlObject], +) -> str: + candidate = copy.deepcopy(source.value) + if "hooks" in candidate: + container: Any = candidate["hooks"] + if not isinstance(container, (AoT, Array)): + raise ValueError(f'{source.contract.label} field "hooks" must be an array') + keep = set(keep_indices) + for index in range(len(source.hooks) - 1, -1, -1): + if index not in keep: + del container[index] + else: + container = aot() + candidate.add("hooks", container) + + for hook in additions: + item = table() if isinstance(container, AoT) else inline_table() + for key, value in hook.items(): + item[key] = value + container.append(item) + if len(container) == 0: + del candidate["hooks"] + return dumps(candidate) + + +def _normalize(value: str) -> str: + return os.path.normcase(value) + + +def _command_references(command: str, path: str) -> bool: + return _normalize(path) in _normalize(command) + + +def _is_managed_hook( + hook: TomlObject, + event: str, + script_name: str, + agent_id: str = "", +) -> bool: + if hook.get("event") != event or not isinstance(hook.get("command"), str): + return False + command = str(hook["command"]) + if agent_id: + script_path = os.path.join(ELYDORA_DIR, agent_id, script_name) + return _command_references(command, script_path) + normalized = _normalize(command) + return ( + _normalize(f"{os.sep}.elydora{os.sep}") in normalized + and _normalize(f"{os.sep}{script_name}") in normalized + ) + + +def _kept_hook_indices(source: KimiDocument, agent_id: str = "") -> List[int]: + return [ + index + for index, hook in enumerate(source.hooks) + if not _is_managed_hook(hook, "PreToolUse", GUARD_SCRIPT, agent_id) + and not _is_managed_hook(hook, "PostToolUse", AUDIT_SCRIPT, agent_id) + ] + + +def _runtime_contract(source: KimiDocument) -> Optional[RuntimeContract]: + guard = next( + ( + hook + for hook in source.hooks + if _is_managed_hook(hook, "PreToolUse", GUARD_SCRIPT) + ), + None, + ) + audit = next( + ( + hook + for hook in source.hooks + if _is_managed_hook(hook, "PostToolUse", AUDIT_SCRIPT) + ), + None, + ) + if guard is None or audit is None: + return None + return RuntimeContract( + guard=str(guard["command"]), + audit=str(audit["command"]), + config_path=source.contract.config_path, + ) + + +def _runtime_scripts_exist(contracts: List[RuntimeContract]) -> bool: + try: + with os.scandir(ELYDORA_DIR) as iterator: + entries = sorted(iterator, key=lambda entry: entry.name) + except FileNotFoundError: + return False + except OSError as error: + raise OSError(f"Read Elydora runtime directory at {ELYDORA_DIR}: {error}") from error + + for entry in entries: + try: + if not entry.is_dir(follow_symlinks=False): + continue + except OSError as error: + raise OSError(f"Read Elydora runtime entry at {entry.path}: {error}") from error + guard_path = os.path.join(entry.path, GUARD_SCRIPT) + audit_path = os.path.join(entry.path, AUDIT_SCRIPT) + if not any( + _command_references(contract.guard, guard_path) + and _command_references(contract.audit, audit_path) + for contract in contracts + ): + continue + config_path = os.path.join(entry.path, "config.json") + runtime_config = _read_json(config_path, "Elydora runtime config") + if runtime_config is None: + continue + agent_name = runtime_config.get("agent_name") + if not isinstance(agent_name, str): + raise ValueError( + f'Elydora runtime config at {config_path} field "agent_name" ' + "must be a string" + ) + if agent_name != AGENT_KEY: + continue + return _regular_file_exists( + guard_path, "Elydora guard runtime" + ) and _regular_file_exists(audit_path, "Elydora audit runtime") + return False + + +def _apply_mutation(mutation: ConfigMutation) -> None: + path = mutation.source.contract.config_path + if mutation.kind == "write": + _write_text_atomic(path, mutation.raw, 0o600, mutation.source.contract.label) + elif mutation.kind == "remove": + _remove_file(path, mutation.source.contract.label) + + +class KimiPlugin(AgentPlugin): + """Install Elydora into detected Kimi user hook contracts.""" + + def install(self, config: InstallConfig) -> None: + agent_id = config.get("agent_id", "") + if not agent_id: + raise ValueError("agent_id is required") + sources = _read_all_configs() + guard_path = config.get("guard_script_path", "") + _require_runtime(guard_path, "Elydora guard runtime") + agent_dir = os.path.join(ELYDORA_DIR, agent_id) + audit_path = os.path.join(agent_dir, AUDIT_SCRIPT) + mutations = [ + ConfigMutation( + "write", + source, + _render_hooks( + source, + _kept_hook_indices(source), + [ + _build_hook("PreToolUse", guard_path), + _build_hook("PostToolUse", audit_path), + ], + ), + ) + for source in sources + ] + runtime_config: TomlObject = { + "org_id": config.get("org_id", ""), + "agent_id": agent_id, + "kid": config.get("kid", ""), + "base_url": config.get("base_url", "https://api.elydora.com"), + "token": config.get("token", ""), + "agent_name": AGENT_KEY, + } + audit_script = generate_hook_script( + org_id=config.get("org_id", ""), + agent_id=agent_id, + private_key=config.get("private_key", ""), + kid=config.get("kid", ""), + base_url=config.get("base_url", "https://api.elydora.com"), + ) + _write_json_atomic( + os.path.join(agent_dir, "config.json"), + runtime_config, + 0o600, + "Elydora runtime config", + ) + _write_text_atomic( + os.path.join(agent_dir, "private.key"), + config.get("private_key", ""), + 0o600, + "Elydora private key", + ) + _write_text_atomic(audit_path, audit_script, 0o700, "Elydora audit runtime") + for mutation in mutations: + _apply_mutation(mutation) + runtimes = " and ".join(source.contract.runtime_name for source in sources) + print(f"{runtimes}: global PreToolUse and PostToolUse hooks installed.") + + def uninstall(self, agent_id: str = "") -> None: + sources = _read_all_configs() + mutations: List[ConfigMutation] = [] + for source in sources: + keep_indices = _kept_hook_indices(source, agent_id) + if len(keep_indices) == len(source.hooks): + mutations.append(ConfigMutation("none", source)) + continue + raw = _render_hooks(source, keep_indices, []) + mutations.append( + ConfigMutation("write" if raw.strip() else "remove", source, raw) + ) + for mutation in mutations: + _apply_mutation(mutation) + + def status(self) -> PluginStatus: + sources = _read_all_configs() + contracts = [ + contract + for source in sources + for contract in [_runtime_contract(source)] + if contract is not None + ] + if not contracts: + return PluginStatus( + installed=False, + agent=AGENT_KEY, + details="Not installed", + ) + installed = _runtime_scripts_exist(contracts) + paths = ", ".join(contract.config_path for contract in contracts) + details = ( + f"Config: {paths}" + if installed + else f"Configured at {paths}; runtime scripts missing" + ) + return PluginStatus(installed=installed, agent=AGENT_KEY, details=details) diff --git a/elydora/plugins/registry.py b/elydora/plugins/registry.py index 3b05e85..e549e9a 100644 --- a/elydora/plugins/registry.py +++ b/elydora/plugins/registry.py @@ -41,6 +41,11 @@ SUPPORTED_AGENTS: Dict[str, Dict[str, str]] = { "hook_event": "PostToolUse", "config_path": "~/.kiro/hooks/", }, + "kimi": { + "name": "Kimi Code", + "hook_event": "PreToolUse/PostToolUse", + "config_path": "~/.kimi-code/config.toml", + }, "letta": { "name": "Letta Code", "hook_event": "PostToolUse", diff --git a/pyproject.toml b/pyproject.toml index 524f9b4..7add10c 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -25,6 +25,7 @@ dependencies = [ "requests>=2.28.0", "aiohttp>=3.8.0", "cryptography>=41.0.0", + "tomlkit>=0.15.1,<1.0.0", "typing_extensions>=4.0.0;python_version<'3.11'", ] diff --git a/tests/test_kimi_plugin.py b/tests/test_kimi_plugin.py new file mode 100644 index 0000000..87793c9 --- /dev/null +++ b/tests/test_kimi_plugin.py @@ -0,0 +1,423 @@ +from __future__ import annotations + +from dataclasses import dataclass +import json +import os +from pathlib import Path +import subprocess +from typing import Any + +import pytest +from tomlkit import parse + +from elydora import cli +from elydora.plugins import kimi +from elydora.plugins.base import InstallConfig +from elydora.plugins.registry import SUPPORTED_AGENTS + + +AGENT_ID = "agent-1" +MISSING = object() + + +@dataclass(frozen=True) +class KimiFixture: + plugin: kimi.KimiPlugin + config: InstallConfig + home_dir: Path + agent_dir: Path + modern_path: Path + legacy_path: Path + guard_path: Path + hook_path: Path + runtime_config_path: Path + private_key_path: Path + + +def write_optional(path: Path, value: object) -> None: + if value is MISSING: + return + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(str(value), encoding="utf-8") + + +def prepare_fixture( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, + *, + modern_config: object = MISSING, + legacy_config: object = MISSING, + legacy_installed: bool = True, + explicit_kimi_home: bool = True, + create_guard: bool = True, +) -> KimiFixture: + home_dir = tmp_path / "home with spaces" + agent_dir = home_dir / ".elydora" / AGENT_ID + kimi_home = ( + home_dir / "custom kimi code" + if explicit_kimi_home + else home_dir / ".kimi-code" + ) + modern_path = kimi_home / "config.toml" + legacy_path = home_dir / ".kimi" / "config.toml" + guard_path = agent_dir / "guard.py" + hook_path = agent_dir / "hook.py" + agent_dir.mkdir(parents=True) + if create_guard: + guard_path.write_text( + "import sys\nsys.stdin.read()\n" + "sys.stderr.write('Agent is frozen by Elydora.')\n" + "raise SystemExit(2)\n", + encoding="utf-8", + ) + write_optional(modern_path, modern_config) + write_optional(legacy_path, legacy_config) + + monkeypatch.setattr(kimi, "ELYDORA_DIR", str(agent_dir.parent)) + monkeypatch.setattr(kimi, "_home_dir", lambda: str(home_dir)) + monkeypatch.setattr( + kimi.shutil, + "which", + lambda command: str(home_dir / "bin" / command) + if legacy_installed and command == "kimi-cli" + else None, + ) + if explicit_kimi_home: + monkeypatch.setenv("KIMI_CODE_HOME", str(kimi_home)) + else: + monkeypatch.delenv("KIMI_CODE_HOME", raising=False) + + config: InstallConfig = { + "agent_id": AGENT_ID, + "agent_name": "kimi", + "org_id": "org-1", + "private_key": "test-key", + "kid": "kid-1", + "base_url": "https://api.elydora.test", + "guard_script_path": str(guard_path), + } + return KimiFixture( + plugin=kimi.KimiPlugin(), + config=config, + home_dir=home_dir, + agent_dir=agent_dir, + modern_path=modern_path, + legacy_path=legacy_path, + guard_path=guard_path, + hook_path=hook_path, + runtime_config_path=agent_dir / "config.json", + private_key_path=agent_dir / "private.key", + ) + + +def parsed_hooks(path: Path) -> list[dict[str, Any]]: + document = parse(path.read_text(encoding="utf-8")) + value = document["hooks"].unwrap() + assert isinstance(value, list) + return value + + +def managed_hook( + hooks: list[dict[str, Any]], event: str, script_name: str +) -> dict[str, Any]: + return next( + hook + for hook in hooks + if hook.get("event") == event and script_name in str(hook.get("command")) + ) + + +def assert_strict_hook(hook: dict[str, Any]) -> None: + assert set(hook) == {"event", "command", "timeout"} + assert hook["timeout"] == 10 + + +def run_command( + command: str, home_dir: Path, payload: dict[str, Any] +) -> subprocess.CompletedProcess[str]: + return subprocess.run( + command, + shell=True, + capture_output=True, + check=False, + env={**os.environ, "HOME": str(home_dir), "USERPROFILE": str(home_dir)}, + input=json.dumps(payload), + text=True, + ) + + +def test_kimi_is_registered_in_sdk_and_cli() -> None: + assert SUPPORTED_AGENTS["kimi"] == { + "name": "Kimi Code", + "hook_event": "PreToolUse/PostToolUse", + "config_path": "~/.kimi-code/config.toml", + } + assert cli.PLUGIN_MAP["kimi"] is kimi.KimiPlugin + + +def test_install_preserves_both_configs_and_is_idempotent( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, + capsys: pytest.CaptureFixture[str], +) -> None: + modern = ( + '# modern user config\ndefault_model = "kimi-code/k3"\n\n' + '[[hooks]]\nevent = "SessionStart"\ncommand = "existing-modern"\n' + "timeout = 30 # keep modern hook\n" + ) + legacy = ( + "# legacy user config\ntelemetry = false\n\n" + '[[hooks]]\nevent = "SessionEnd"\ncommand = "existing-legacy"\n' + ) + fixture = prepare_fixture( + monkeypatch, + tmp_path, + modern_config=modern, + legacy_config=legacy, + ) + + fixture.plugin.install(fixture.config) + fixture.plugin.install(fixture.config) + + assert "Kimi Code and kimi-cli" in capsys.readouterr().out + for path, comment, command in ( + (fixture.modern_path, "# modern user config", "existing-modern"), + (fixture.legacy_path, "# legacy user config", "existing-legacy"), + ): + raw = path.read_text(encoding="utf-8") + hooks = parsed_hooks(path) + assert comment in raw + assert command in raw + assert len(hooks) == 3 + assert_strict_hook(managed_hook(hooks, "PreToolUse", "guard.py")) + assert_strict_hook(managed_hook(hooks, "PostToolUse", "hook.py")) + assert (fixture.home_dir / ".kimi-code" / "config.toml").exists() is False + + +def test_install_preserves_inline_hook_array_style( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + modern = ( + '# inline user hook\nhooks = [{ event = "SessionStart", ' + 'command = "existing-inline" }] # keep array\n' + ) + fixture = prepare_fixture( + monkeypatch, + tmp_path, + modern_config=modern, + legacy_installed=False, + ) + + fixture.plugin.install(fixture.config) + fixture.plugin.install(fixture.config) + + raw = fixture.modern_path.read_text(encoding="utf-8") + assert "# inline user hook" in raw + assert "# keep array" in raw + assert "hooks = [" in raw + assert len(parsed_hooks(fixture.modern_path)) == 3 + + +def test_modern_install_avoids_false_legacy_migration_source( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + fixture = prepare_fixture(monkeypatch, tmp_path, legacy_installed=False) + fixture.plugin.install(fixture.config) + + assert len(parsed_hooks(fixture.modern_path)) == 2 + assert fixture.legacy_path.exists() is False + + +def test_empty_home_override_uses_official_default( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + fixture = prepare_fixture( + monkeypatch, + tmp_path, + explicit_kimi_home=False, + legacy_installed=False, + ) + fixture.plugin.install(fixture.config) + monkeypatch.setenv("KIMI_CODE_HOME", "") + + assert fixture.plugin.status()["installed"] is True + + +def test_legacy_install_avoids_premature_modern_migration_target( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + fixture = prepare_fixture(monkeypatch, tmp_path, explicit_kimi_home=False) + fixture.plugin.install(fixture.config) + + assert len(parsed_hooks(fixture.legacy_path)) == 2 + assert fixture.modern_path.exists() is False + monkeypatch.setattr(kimi.shutil, "which", lambda command: None) + assert fixture.plugin.status()["installed"] is True + + +def test_commands_block_and_forward_official_payload( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + fixture = prepare_fixture(monkeypatch, tmp_path) + fixture.plugin.install(fixture.config) + capture_path = tmp_path / "captured-event.json" + fixture.hook_path.write_text( + "from pathlib import Path\nimport sys\n" + f"Path({str(capture_path)!r}).write_text(sys.stdin.read(), encoding='utf-8')\n", + encoding="utf-8", + ) + hooks = parsed_hooks(fixture.modern_path) + payload = { + "hook_event_name": "PreToolUse", + "session_id": "session-1", + "cwd": str(fixture.home_dir), + "tool_name": "Bash", + "tool_input": {"command": "echo test"}, + "tool_call_id": "call-1", + } + + guard = managed_hook(hooks, "PreToolUse", "guard.py") + guard_result = run_command(guard["command"], fixture.home_dir, payload) + assert guard_result.returncode == 2 + assert "Agent is frozen by Elydora" in guard_result.stderr + + payload["hook_event_name"] = "PostToolUse" + payload["tool_output"] = {"output": "test"} + audit = managed_hook(hooks, "PostToolUse", "hook.py") + audit_result = run_command(audit["command"], fixture.home_dir, payload) + assert audit_result.returncode == 0 + assert json.loads(capture_path.read_text(encoding="utf-8")) == payload + + +def test_status_accepts_either_contract_and_requires_both_scripts( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + fixture = prepare_fixture(monkeypatch, tmp_path) + fixture.plugin.install(fixture.config) + fixture.modern_path.unlink() + assert fixture.plugin.status()["installed"] is True + + fixture.plugin.install(fixture.config) + fixture.legacy_path.unlink() + assert fixture.plugin.status()["installed"] is True + + fixture.guard_path.unlink() + assert fixture.plugin.status()["installed"] is False + + +def test_uninstall_preserves_user_hooks_in_both_contracts( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + user_hook = ( + "# user hook\n[[hooks]]\nevent = \"SessionStart\"\n" + 'command = "existing-command"\ntimeout = 30 # keep timeout\n' + ) + fixture = prepare_fixture( + monkeypatch, + tmp_path, + modern_config=user_hook, + legacy_config=user_hook, + ) + fixture.plugin.install(fixture.config) + + fixture.plugin.uninstall(AGENT_ID) + + for path in (fixture.modern_path, fixture.legacy_path): + raw = path.read_text(encoding="utf-8") + assert "# user hook" in raw + assert "# keep timeout" in raw + assert parsed_hooks(path) == [{ + "event": "SessionStart", + "command": "existing-command", + "timeout": 30, + }] + + +def test_uninstall_removes_configs_created_by_elydora( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + fixture = prepare_fixture(monkeypatch, tmp_path) + fixture.plugin.install(fixture.config) + + fixture.plugin.uninstall(AGENT_ID) + + assert fixture.modern_path.exists() is False + assert fixture.legacy_path.exists() is False + + +def test_install_parses_every_config_before_runtime_writes( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + modern = '# untouched modern\ndefault_model = "kimi-code/k3"\n' + fixture = prepare_fixture( + monkeypatch, + tmp_path, + modern_config=modern, + legacy_config="[malformed", + ) + + with pytest.raises(ValueError, match="parse kimi-cli legacy hooks config"): + fixture.plugin.install(fixture.config) + + assert fixture.modern_path.read_text(encoding="utf-8") == modern + assert fixture.legacy_path.read_text(encoding="utf-8") == "[malformed" + assert fixture.hook_path.exists() is False + assert fixture.runtime_config_path.exists() is False + assert fixture.private_key_path.exists() is False + + +def test_install_rejects_invalid_hook_shape_without_writes( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + modern = ( + '[[hooks]]\nevent = "PreToolUse"\ncommand = "existing-command"\n' + 'cwd = "/tmp"\n' + ) + fixture = prepare_fixture( + monkeypatch, + tmp_path, + modern_config=modern, + legacy_installed=False, + ) + + with pytest.raises(ValueError, match='unsupported field "cwd"'): + fixture.plugin.install(fixture.config) + + assert fixture.modern_path.read_text(encoding="utf-8") == modern + assert fixture.hook_path.exists() is False + assert fixture.runtime_config_path.exists() is False + + +def test_install_rejects_missing_guard_before_creating_files( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + fixture = prepare_fixture(monkeypatch, tmp_path, create_guard=False) + + with pytest.raises(FileNotFoundError, match="guard runtime is missing"): + fixture.plugin.install(fixture.config) + + assert fixture.modern_path.exists() is False + assert fixture.legacy_path.exists() is False + assert fixture.hook_path.exists() is False + assert fixture.runtime_config_path.exists() is False + + +def test_status_surfaces_malformed_runtime_metadata( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + fixture = prepare_fixture(monkeypatch, tmp_path) + fixture.plugin.install(fixture.config) + fixture.runtime_config_path.write_text("{ malformed", encoding="utf-8") + + with pytest.raises(ValueError, match="parse Elydora runtime config"): + fixture.plugin.status() + + +def test_atomic_writes_leave_no_temporary_files( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + fixture = prepare_fixture(monkeypatch, tmp_path) + fixture.plugin.install(fixture.config) + + for directory in (fixture.modern_path.parent, fixture.legacy_path.parent): + assert all(path.suffix != ".tmp" for path in directory.iterdir())