feat: sync Python Kimi hook integration

sdks/python/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` |

sdks/python/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,
 }

sdks/python/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

sdks/python/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)

sdks/python/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",

sdks/python/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'",
 ]
 

sdks/python/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())