fix: align Kimi hooks with official contracts

This commit is contained in:
ZacharyZhang-NY committed 2026-07-20 04:25:38 -04:00
1 parent cb6173e6ee
commit 978e15e251
13 files changed
+2105 -745

No files matched your search

+217 -300
View File
@@ -1,423 +1,340 @@
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 tomlkit import dumps, parse
from elydora import cli
from elydora.plugins import kimi
from elydora.plugins.base import InstallConfig
from elydora.plugins.registry import SUPPORTED_AGENTS
from kimi_support import (
AGENT_ID,
MISSING,
assert_managed_triple,
assert_no_transaction_files,
legacy_command,
parsed_hooks,
prepare_fixture,
)
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,
)
STABLE_USER_CONFIG = (
'# stable user config\ndefault_model = "kimi-code/k3"\n\n'
'[[hooks]]\nevent = "SessionStart"\ncommand = "existing-stable"\n'
"timeout = 30 # keep stable hook\n"
)
LEGACY_USER_CONFIG = (
"# legacy user config\ntelemetry = false\n\n"
'[[hooks]]\nevent = "SessionEnd"\ncommand = "existing-legacy"\n'
)
def test_kimi_is_registered_in_sdk_and_cli() -> None:
assert SUPPORTED_AGENTS["kimi"] == {
"name": "Kimi Code",
"hook_event": "PreToolUse/PostToolUse",
"hook_event": "PreToolUse/PostToolUse/PostToolUseFailure",
"config_path": "~/.kimi-code/config.toml",
}
assert cli.PLUGIN_MAP["kimi"] is kimi.KimiPlugin
def test_install_preserves_both_configs_and_is_idempotent(
def test_install_writes_exact_triples_to_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,
stable_config=STABLE_USER_CONFIG,
legacy_config=LEGACY_USER_CONFIG,
)
fixture.plugin.install(fixture.config)
fixture.plugin.install(fixture.config)
fixture.install()
first_sources = {
path: path.read_text(encoding="utf-8")
for path in (fixture.stable_path, fixture.legacy_path)
}
fixture.install()
assert "Kimi Code and kimi-cli" in capsys.readouterr().out
for path, comment, command in (
(fixture.modern_path, "# modern user config", "existing-modern"),
(fixture.stable_path, "# stable user config", "existing-stable"),
(fixture.legacy_path, "# legacy user config", "existing-legacy"),
):
raw = path.read_text(encoding="utf-8")
hooks = parsed_hooks(path)
assert raw == first_sources[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
assert any(hook.get("command") == command for hook in hooks)
assert len(hooks) == 4
assert_managed_triple(hooks)
assert not (fixture.home_dir / ".kimi-code" / "config.toml").exists()
def test_install_preserves_inline_hook_array_style(
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
) -> None:
modern = (
source = (
'# inline user hook\nhooks = [{ event = "SessionStart", '
'command = "existing-inline" }] # keep array\n'
)
fixture = prepare_fixture(
monkeypatch,
tmp_path,
modern_config=modern,
legacy_installed=False,
stable_config=source,
legacy_detected=False,
)
fixture.plugin.install(fixture.config)
fixture.plugin.install(fixture.config)
fixture.install()
fixture.install()
raw = fixture.modern_path.read_text(encoding="utf-8")
raw = fixture.stable_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
hooks = parsed_hooks(fixture.stable_path)
assert len(hooks) == 4
assert_managed_triple(hooks)
def test_modern_install_avoids_false_legacy_migration_source(
def test_uninstall_preserves_user_owned_empty_inline_hook_array(
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
) -> None:
fixture = prepare_fixture(monkeypatch, tmp_path, legacy_installed=False)
fixture.plugin.install(fixture.config)
source = "# user container\nhooks = [] # keep empty array\n"
fixture = prepare_fixture(
monkeypatch,
tmp_path,
stable_config=source,
legacy_detected=False,
)
fixture.install()
fixture.plugin.uninstall(AGENT_ID)
assert len(parsed_hooks(fixture.modern_path)) == 2
assert fixture.legacy_path.exists() is False
raw = fixture.stable_path.read_text(encoding="utf-8")
assert "# user container" in raw
assert "# keep empty array" in raw
assert parse(raw)["hooks"].unwrap() == []
def test_empty_home_override_uses_official_default(
def test_fresh_stable_install_leaves_legacy_source_absent(
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
) -> None:
fixture = prepare_fixture(monkeypatch, tmp_path, legacy_detected=False)
fixture.install()
hooks = parsed_hooks(fixture.stable_path)
assert len(hooks) == 3
assert_managed_triple(hooks)
assert not fixture.legacy_path.exists()
def test_empty_home_override_uses_documented_default(
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
) -> None:
fixture = prepare_fixture(
monkeypatch,
tmp_path,
explicit_kimi_home=False,
legacy_installed=False,
legacy_detected=False,
)
fixture.plugin.install(fixture.config)
fixture.install()
monkeypatch.setenv("KIMI_CODE_HOME", "")
assert fixture.plugin.status()["installed"] is True
assert_managed_triple(parsed_hooks(fixture.stable_path))
def test_legacy_install_avoids_premature_modern_migration_target(
def test_legacy_only_install_leaves_stable_target_absent(
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,
explicit_kimi_home=False,
stable_detected=False,
)
fixture.plugin.install(fixture.config)
fixture.install()
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,
}]
assert_managed_triple(parsed_hooks(fixture.legacy_path))
assert not fixture.stable_path.exists()
assert fixture.plugin.status()["installed"] is True
def test_uninstall_removes_configs_created_by_elydora(
def test_install_parses_every_selected_config_before_runtime_writes(
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,
stable_config=STABLE_USER_CONFIG,
legacy_config="[malformed",
)
with pytest.raises(ValueError, match="parse kimi-cli legacy hooks config"):
fixture.plugin.install(fixture.config)
fixture.install()
assert fixture.modern_path.read_text(encoding="utf-8") == modern
assert fixture.stable_path.read_text(encoding="utf-8") == STABLE_USER_CONFIG
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
assert not fixture.agent_dir.exists()
def test_install_rejects_invalid_hook_shape_without_writes(
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
@pytest.mark.parametrize(
("stable_config", "legacy_config", "pattern"),
[
(
'[[hooks]]\nevent = "PreToolUse"\ncommand = "existing"\n'
'cwd = "/tmp"\n',
None,
'unsupported field "cwd"',
),
(
None,
'[[hooks]]\nevent = "Interrupt"\ncommand = "existing"\n',
'unsupported event "Interrupt"',
),
(
'[[hooks]]\nevent = "PreToolUse"\ncommand = "existing"\n'
"timeout = 601\n",
None,
"integer from 1 to 600",
),
],
)
def test_install_rejects_fields_and_events_outside_each_contract(
monkeypatch: pytest.MonkeyPatch,
tmp_path: Path,
stable_config: str | None,
legacy_config: str | None,
pattern: str,
) -> None:
modern = (
'[[hooks]]\nevent = "PreToolUse"\ncommand = "existing-command"\n'
'cwd = "/tmp"\n'
)
fixture = prepare_fixture(
monkeypatch,
tmp_path,
modern_config=modern,
legacy_installed=False,
stable_config=stable_config if stable_config is not None else MISSING,
legacy_config=legacy_config if legacy_config is not None else MISSING,
)
with pytest.raises(ValueError, match='unsupported field "cwd"'):
fixture.plugin.install(fixture.config)
with pytest.raises(ValueError, match=pattern):
fixture.install()
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
assert not fixture.agent_dir.exists()
def test_install_rejects_missing_guard_before_creating_files(
def test_install_migrates_exact_legacy_commands_and_preserves_lookalikes(
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
) -> None:
fixture = prepare_fixture(monkeypatch, tmp_path, create_guard=False)
fixture = prepare_fixture(monkeypatch, tmp_path, legacy_detected=False)
lookalike = f"{legacy_command(fixture.guard_path)} --inspect"
source = (
'[[hooks]]\nevent = "PreToolUse"\n'
f"command = {json.dumps(legacy_command(fixture.guard_path))}\n"
"timeout = 10\n\n"
'[[hooks]]\nevent = "PostToolUse"\n'
f"command = {json.dumps(legacy_command(fixture.hook_path))}\n"
"timeout = 10\n\n"
'[[hooks]]\nevent = "PreToolUse"\n'
f"command = {json.dumps(lookalike)}\ntimeout = 10\n"
)
fixture.stable_path.parent.mkdir(parents=True, exist_ok=True)
fixture.stable_path.write_text(source, encoding="utf-8")
with pytest.raises(FileNotFoundError, match="guard runtime is missing"):
fixture.plugin.install(fixture.config)
fixture.install()
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
hooks = parsed_hooks(fixture.stable_path)
assert len(hooks) == 4
assert any(hook.get("command") == lookalike for hook in hooks)
assert_managed_triple(hooks)
fixture.plugin.uninstall(AGENT_ID)
assert parsed_hooks(fixture.stable_path) == [
{"event": "PreToolUse", "command": lookalike, "timeout": 10}
]
def test_status_surfaces_malformed_runtime_metadata(
def test_uninstall_preserves_user_configs_and_removes_managed_configs(
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")
fixture = prepare_fixture(
monkeypatch,
tmp_path,
stable_config=STABLE_USER_CONFIG,
legacy_config=LEGACY_USER_CONFIG,
)
fixture.install()
fixture.plugin.uninstall(AGENT_ID)
with pytest.raises(ValueError, match="parse Elydora runtime config"):
assert "# keep stable hook" in fixture.stable_path.read_text(encoding="utf-8")
assert "# legacy user config" in fixture.legacy_path.read_text(encoding="utf-8")
assert [hook["command"] for hook in parsed_hooks(fixture.stable_path)] == [
"existing-stable"
]
assert [hook["command"] for hook in parsed_hooks(fixture.legacy_path)] == [
"existing-legacy"
]
managed = prepare_fixture(
monkeypatch,
tmp_path / "managed",
)
managed.install()
managed.plugin.uninstall(AGENT_ID)
assert not managed.stable_path.exists()
assert not managed.legacy_path.exists()
def test_status_requires_complete_triple_runtime_identity_and_private_key(
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
) -> None:
fixture = prepare_fixture(monkeypatch, tmp_path, legacy_detected=False)
fixture.install()
assert fixture.plugin.status()["installed"] is True
document = parse(fixture.stable_path.read_text(encoding="utf-8"))
hooks = document["hooks"]
for index in range(len(hooks) - 1, -1, -1):
if hooks[index]["event"] == "PostToolUseFailure":
del hooks[index]
fixture.stable_path.write_text(dumps(document), encoding="utf-8")
assert fixture.plugin.status()["installed"] is False
fixture.install()
fixture.private_key_path.write_text("invalid", encoding="utf-8")
with pytest.raises(ValueError, match="private key.*canonical 32-byte"):
fixture.plugin.status()
def test_atomic_writes_leave_no_temporary_files(
@pytest.mark.parametrize(
"runtime_name",
["guard.py", "hook.py", "config.json", "private.key"],
)
def test_status_requires_every_runtime_file(
monkeypatch: pytest.MonkeyPatch,
tmp_path: Path,
runtime_name: str,
) -> None:
fixture = prepare_fixture(monkeypatch, tmp_path, legacy_detected=False)
fixture.install()
(fixture.agent_dir / runtime_name).unlink()
assert fixture.plugin.status()["installed"] is False
def test_installation_leaves_no_transaction_files(
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
) -> None:
fixture = prepare_fixture(monkeypatch, tmp_path)
fixture.plugin.install(fixture.config)
fixture.install()
for directory in (fixture.modern_path.parent, fixture.legacy_path.parent):
assert all(path.suffix != ".tmp" for path in directory.iterdir())
assert_no_transaction_files(fixture.home_dir)
if os.name != "nt":
for path in (
fixture.stable_path,
fixture.legacy_path,
fixture.runtime_config_path,
fixture.private_key_path,
):
assert path.stat().st_mode & 0o777 == 0o600