feat: support Kiro CLI v2 and v3 hooks

This commit is contained in:
ZacharyZhang-NY committed 2026-07-19 04:40:40 -04:00
1 parent 80bb8b4c12
commit d9cb1f0d93
5 files changed
+787 -159

No files matched your search

+50
View File
@@ -0,0 +1,50 @@
# Elydora Python SDK Engineering Contract
## Scope
This repository owns the published `elydora` package, its CLI, synchronous and asynchronous API clients, local signing behavior, generated hook runtimes, and Python-specific agent adapters.
## Integration Sources
- Verify every agent hook contract against current official provider documentation before changing an adapter.
- Treat `../Elydora-Open-Source/integrations/catalog.json` as the cross-product provider inventory.
- Keep adapter delivery claims aligned with executable tests in this repository.
- Synchronize completed provider behavior into the Node SDK, Go SDK, Open Source distribution, Console, Docs, and landing page through separate reviewed commits.
## Hook Adapter Invariants
- Preserve unrelated user configuration and remove only Elydora-owned entries.
- Parse every affected user configuration before the first write.
- Surface malformed or unreadable configuration with contextual errors and leave the original file intact.
- Write configuration through a same-directory temporary file followed by an atomic replace.
- Quote the current Python executable and generated script paths for the host shell.
- Forward official hook JSON from STDIN without reshaping provider fields.
- 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.
## Code Quality
- Support every Python version declared in `pyproject.toml`.
- Keep production source files at or below 500 lines.
- Keep functions focused on one ownership boundary.
- Propagate unexpected errors to the CLI boundary.
- Use documented defaults only for genuinely optional configuration.
- Avoid compatibility shims without a named public or user configuration contract.
## Verification
Run the focused adapter test during development, then execute all release gates before commit:
```powershell
py -3 -m pytest tests/test_<provider>_plugin.py -q
py -3 -m pytest -q
py -3 -m mypy elydora
py -3 -m pip check
py -3 -m pip wheel . --no-deps --wheel-dir <temporary-directory>
git diff --check
```
Provider adapter tests must cover installation, idempotency, official event forwarding, blocking behavior, status, missing runtime files, uninstall ownership, and malformed configuration preservation.
Commit and push one root issue before starting the next one.
+2
View File
@@ -73,6 +73,8 @@ elydora install \
--kid agent-456-key-v1 --kid agent-456-key-v1
``` ```
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 ### Commands
| Command | Description | | Command | Description |
+377 -157
View File
@@ -1,211 +1,431 @@
"""Kiro CLI plugin — merges PreToolUse/PostToolUse hooks into ~/.kiro/settings.json.""" """Kiro CLI v2 custom-agent and v3 standalone-hook integration."""
from __future__ import annotations from __future__ import annotations
import json import json
import os import os
import shlex
import stat import stat
import subprocess
import sys import sys
import tempfile
from typing import Any, Dict, List, Optional, Tuple
from .base import AgentPlugin, InstallConfig, PluginStatus from .base import AgentPlugin, InstallConfig, PluginStatus
from .hook_template import generate_hook_script from .hook_template import generate_hook_script
SETTINGS_PATH = os.path.join(os.path.expanduser("~"), ".kiro", "settings.json") AGENT_KEY = "kirocli"
V2_AGENT_NAME = "elydora-audit"
V2_DESCRIPTION = "Kiro CLI with Elydora audit and freeze enforcement"
V3_GUARD_NAME = "elydora-guard"
V3_AUDIT_NAME = "elydora-audit"
ELYDORA_DIR = os.path.join(os.path.expanduser("~"), ".elydora") ELYDORA_DIR = os.path.join(os.path.expanduser("~"), ".elydora")
V2_AGENT_PATH = os.path.join(
os.path.expanduser("~"), ".kiro", "agents", f"{V2_AGENT_NAME}.json"
)
V3_HOOKS_PATH = os.path.join(
os.path.expanduser("~"), ".kiro", "hooks", "elydora-audit.json"
)
JsonObject = Dict[str, Any]
HookContract = Tuple[str, str, str]
class KiroCliPlugin(AgentPlugin): class KiroCliPlugin(AgentPlugin):
"""Install/uninstall Elydora audit hook for Kiro CLI.""" """Install Elydora into both supported Kiro CLI hook generations."""
@staticmethod
def _hook_path_for(agent_id: str) -> str:
return os.path.join(ELYDORA_DIR, agent_id, "hook.py")
def install(self, config: InstallConfig) -> None: def install(self, config: InstallConfig) -> None:
agent_id = config.get("agent_id", "") agent_id = config.get("agent_id", "")
agent_name = config.get("agent_name", "") guard_path = config.get("guard_script_path", "")
if not agent_id:
raise ValueError("agent_id is required")
if not guard_path:
raise ValueError("guard_script_path is required")
if not _regular_file_exists(guard_path, "Elydora guard runtime"):
raise FileNotFoundError(f"Elydora guard runtime is missing: {guard_path}")
v2_settings = _read_json(V2_AGENT_PATH, "Kiro CLI v2 agent config") or {}
v3_settings = _read_json(V3_HOOKS_PATH, "Kiro CLI v3 hooks config") or {}
v2_hooks = _hooks_object(v2_settings, "Kiro CLI v2 agent config")
current_v3_hooks = _v3_hooks(v3_settings)
v2_hooks["preToolUse"] = [
*_without_v2_hooks(
_hook_entries(v2_hooks, "preToolUse", "Kiro CLI v2 agent config")
),
_build_v2_hook(guard_path),
]
# Create per-agent directory
agent_dir = os.path.join(ELYDORA_DIR, agent_id) agent_dir = os.path.join(ELYDORA_DIR, agent_id)
os.makedirs(agent_dir, exist_ok=True) hook_path = os.path.join(agent_dir, "hook.py")
v2_hooks["postToolUse"] = [
*_without_v2_hooks(
_hook_entries(v2_hooks, "postToolUse", "Kiro CLI v2 agent config")
),
_build_v2_hook(hook_path),
]
# Write config.json next_v2_settings: JsonObject = {
config_data = { "name": V2_AGENT_NAME,
"description": V2_DESCRIPTION,
"tools": ["*"],
"includeMcpJson": True,
**v2_settings,
"hooks": v2_hooks,
}
next_v3_settings: JsonObject = {
**v3_settings,
"version": "v1",
"hooks": [
*[hook for hook in current_v3_hooks if not _is_managed_v3_hook(hook)],
_build_v3_hook(
V3_GUARD_NAME,
"Block tool use when the Elydora agent is frozen",
"PreToolUse",
guard_path,
),
_build_v3_hook(
V3_AUDIT_NAME,
"Record tool use in the Elydora audit trail",
"PostToolUse",
hook_path,
),
],
}
runtime_config = {
"org_id": config.get("org_id", ""), "org_id": config.get("org_id", ""),
"agent_id": agent_id, "agent_id": agent_id,
"kid": config.get("kid", ""), "kid": config.get("kid", ""),
"base_url": config.get("base_url", "https://api.elydora.com"), "base_url": config.get("base_url", "https://api.elydora.com"),
"token": config.get("token", ""), "token": config.get("token", ""),
"agent_name": agent_name, "agent_name": config.get("agent_name", AGENT_KEY),
} }
config_path = os.path.join(agent_dir, "config.json") hook_script = generate_hook_script(
with open(config_path, "w", encoding="utf-8") as f:
json.dump(config_data, f, indent=2)
f.write("\n")
# Write private key
private_key_path = os.path.join(agent_dir, "private.key")
with open(private_key_path, "w", encoding="utf-8") as f:
f.write(config.get("private_key", ""))
try:
os.chmod(private_key_path, stat.S_IRUSR | stat.S_IWUSR)
except Exception:
pass # chmod may fail on Windows
script = generate_hook_script(
org_id=config.get("org_id", ""), org_id=config.get("org_id", ""),
agent_id=agent_id, agent_id=agent_id,
private_key=config.get("private_key", ""), private_key=config.get("private_key", ""),
kid=config.get("kid", ""), kid=config.get("kid", ""),
base_url=config.get("base_url", "https://api.elydora.com"), base_url=config.get("base_url", "https://api.elydora.com"),
) )
hook_path = self._hook_path_for(agent_id) _write_json_atomic(os.path.join(agent_dir, "config.json"), runtime_config, 0o600)
with open(hook_path, "w", encoding="utf-8") as f: _write_text_atomic(
f.write(script) os.path.join(agent_dir, "private.key"),
try: config.get("private_key", ""),
os.chmod(hook_path, stat.S_IRWXU) 0o600,
except Exception: )
pass # chmod may fail on Windows _write_text_atomic(hook_path, hook_script, 0o700)
_write_json_atomic(V2_AGENT_PATH, next_v2_settings, 0o600)
_write_json_atomic(V3_HOOKS_PATH, next_v3_settings, 0o600)
guard_script_path = config.get("guard_script_path", "") print('Kiro CLI v2: start with "kiro-cli --agent elydora-audit".')
python_exe = sys.executable print('Kiro CLI v3: start with "kiro-cli --v3"; global hooks load automatically.')
settings = _load_json(SETTINGS_PATH)
hooks = settings.setdefault("hooks", {})
# --- PreToolUse (guard — freeze enforcement, PascalCase with matcher) ---
pre_tool_use = hooks.setdefault("PreToolUse", [])
pre_tool_use[:] = [h for h in pre_tool_use if not _is_elydora_hook(h)]
if guard_script_path:
pre_tool_use.append({
"matcher": "*",
"hooks": [
{
"type": "command",
"command": f'"{python_exe}" {guard_script_path}',
"timeout_ms": 5000,
}
],
})
# --- PostToolUse (audit logging, PascalCase with matcher) ---
post_tool_use = hooks.setdefault("PostToolUse", [])
post_tool_use[:] = [h for h in post_tool_use if not _is_elydora_hook(h)]
post_tool_use.append({
"matcher": "*",
"hooks": [
{
"type": "command",
"command": hook_path,
"timeout_ms": 5000,
}
],
})
_save_json(SETTINGS_PATH, settings)
print(f"Elydora hook installed for Kiro CLI.")
print(f" Hook script: {hook_path}")
print(f" Settings: {SETTINGS_PATH}")
def uninstall(self, agent_id: str = "") -> None: def uninstall(self, agent_id: str = "") -> None:
if os.path.exists(SETTINGS_PATH): v2_settings = _read_json(V2_AGENT_PATH, "Kiro CLI v2 agent config")
settings = _load_json(SETTINGS_PATH) v3_settings = _read_json(V3_HOOKS_PATH, "Kiro CLI v3 hooks config")
hooks = settings.get("hooks", {})
changed = False
# Remove PreToolUse entries next_v2: Optional[Tuple[JsonObject, bool]] = None
pre_tool_use = hooks.get("PreToolUse", []) if v2_settings is not None:
pre_filtered = [h for h in pre_tool_use if not _is_elydora_hook(h, agent_id)] v2_hooks = _hooks_object(v2_settings, "Kiro CLI v2 agent config")
if len(pre_filtered) != len(pre_tool_use): current_pre = _hook_entries(v2_hooks, "preToolUse", "Kiro CLI v2 agent config")
hooks["PreToolUse"] = pre_filtered current_post = _hook_entries(v2_hooks, "postToolUse", "Kiro CLI v2 agent config")
if not pre_filtered: next_pre = _without_v2_hooks(current_pre, agent_id)
del hooks["PreToolUse"] next_post = _without_v2_hooks(current_post, agent_id)
changed = True if next_pre != current_pre or next_post != current_post:
v2_hooks["preToolUse"] = next_pre
v2_hooks["postToolUse"] = next_post
next_v2 = (
{**v2_settings, "hooks": v2_hooks},
_is_owned_v2(v2_settings, v2_hooks),
)
# Remove PostToolUse entries next_v3: Optional[Tuple[JsonObject, bool]] = None
post_tool_use = hooks.get("PostToolUse", []) if v3_settings is not None:
post_filtered = [h for h in post_tool_use if not _is_elydora_hook(h, agent_id)] current_hooks = _v3_hooks(v3_settings)
if len(post_filtered) != len(post_tool_use): v3_hooks = [
hooks["PostToolUse"] = post_filtered hook for hook in current_hooks if not _is_managed_v3_hook(hook, agent_id)
if not post_filtered: ]
del hooks["PostToolUse"] if v3_hooks != current_hooks:
changed = True owned = all(key in {"version", "hooks"} for key in v3_settings)
next_v3 = ({**v3_settings, "hooks": v3_hooks}, owned and not v3_hooks)
if changed: if next_v2 is not None:
if not hooks: settings, remove = next_v2
del settings["hooks"] if remove:
_save_json(SETTINGS_PATH, settings) _remove_file(V2_AGENT_PATH, "Kiro CLI v2 agent config")
else:
_write_json_atomic(V2_AGENT_PATH, settings, 0o600)
# Hook script removal is handled by cli.py cmd_uninstall (rmtree of agent dir) if next_v3 is not None:
print("Elydora hook uninstalled from Kiro CLI.") settings, remove = next_v3
if remove:
_remove_file(V3_HOOKS_PATH, "Kiro CLI v3 hooks config")
else:
_write_json_atomic(V3_HOOKS_PATH, settings, 0o600)
print("Elydora hooks uninstalled from Kiro CLI.")
def status(self) -> PluginStatus: def status(self) -> PluginStatus:
# Scan ~/.elydora/*/hook.py for any installed hook v2_settings = _read_json(V2_AGENT_PATH, "Kiro CLI v2 agent config")
import glob as _glob v3_settings = _read_json(V3_HOOKS_PATH, "Kiro CLI v3 hooks config")
hook_pattern = os.path.join(ELYDORA_DIR, "*", "hook.py") contracts: List[HookContract] = []
hook_files = _glob.glob(hook_pattern)
hook_exists = len(hook_files) > 0
settings_configured = False if v2_settings is not None:
if os.path.exists(SETTINGS_PATH): v2_hooks = _hooks_object(v2_settings, "Kiro CLI v2 agent config")
settings = _load_json(SETTINGS_PATH) guard = _find_v2_command(
hooks = settings.get("hooks", {}) _hook_entries(v2_hooks, "preToolUse", "Kiro CLI v2 agent config"),
pre_tool_use = hooks.get("PreToolUse", []) "guard.py",
post_tool_use = hooks.get("PostToolUse", []) )
pre_configured = any(_is_elydora_hook(h) for h in pre_tool_use) audit = _find_v2_command(
post_configured = any(_is_elydora_hook(h) for h in post_tool_use) _hook_entries(v2_hooks, "postToolUse", "Kiro CLI v2 agent config"),
settings_configured = pre_configured and post_configured "hook.py",
)
if guard and audit:
contracts.append((guard, audit, V2_AGENT_PATH))
installed = hook_exists and settings_configured if v3_settings is not None:
if installed: v3_hooks = _v3_hooks(v3_settings)
details = f"Found {len(hook_files)} agent(s): {', '.join(hook_files)}" guard = _find_v3_command(v3_hooks, V3_GUARD_NAME, "guard.py")
elif hook_exists: audit = _find_v3_command(v3_hooks, V3_AUDIT_NAME, "hook.py")
details = "Hook script exists but not configured in settings" if guard and audit:
elif settings_configured: contracts.append((guard, audit, V3_HOOKS_PATH))
details = "Configured in settings but hook script missing"
else:
details = "Not installed"
return PluginStatus(installed=installed, agent="kirocli", details=details) if not contracts:
return PluginStatus(installed=False, agent=AGENT_KEY, details="Not installed")
config_path = contracts[-1][2]
def _is_elydora_hook(entry: dict, agent_id: str = "") -> bool: installed = _runtime_scripts_exist(contracts)
# Collect all command strings from the entry details = (
commands = [] f"Config: {config_path}"
inner_hooks = entry.get("hooks") if installed
if isinstance(inner_hooks, list): else f"Configured at {config_path}; runtime scripts missing"
commands.extend(
h.get("command", "")
for h in inner_hooks
if isinstance(h, dict)
) )
else: return PluginStatus(installed=installed, agent=AGENT_KEY, details=details)
commands.append(entry.get("command", ""))
for cmd in commands:
cmd_lower = cmd.lower() def _read_json(path: str, label: str) -> Optional[JsonObject]:
if "elydora" not in cmd_lower: try:
continue with open(path, "r", encoding="utf-8") as file:
# If agent_id is specified, only match hooks for that specific agent raw = file.read()
if agent_id and agent_id in cmd: except FileNotFoundError:
return True return None
if not agent_id: except OSError as error:
return True 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
def _write_text_atomic(path: str, content: str, mode: int) -> None:
directory = os.path.dirname(path) or "."
os.makedirs(directory, exist_ok=True)
descriptor, temporary_path = tempfile.mkstemp(
dir=directory,
prefix=f".{os.path.basename(path)}.",
suffix=".tmp",
text=True,
)
try:
with os.fdopen(descriptor, "w", encoding="utf-8", newline="\n") as file:
file.write(content)
os.chmod(temporary_path, mode)
os.replace(temporary_path, path)
except BaseException as write_error:
try:
os.remove(temporary_path)
except FileNotFoundError:
pass
except OSError as cleanup_error:
raise RuntimeError(
f"Write {path} failed and temporary file cleanup also failed: {cleanup_error}"
) from write_error
raise
def _write_json_atomic(path: str, value: JsonObject, mode: int) -> None:
_write_text_atomic(path, json.dumps(value, indent=2) + "\n", mode)
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 _build_command(script_path: str) -> str:
arguments = [sys.executable, script_path]
return subprocess.list2cmdline(arguments) if os.name == "nt" else shlex.join(arguments)
def _hooks_object(settings: JsonObject, label: str) -> JsonObject:
if "hooks" not in settings:
return {}
value = settings["hooks"]
if not isinstance(value, dict):
raise ValueError(f'{label} field "hooks" must be an object')
return dict(value)
def _hook_entries(hooks: JsonObject, event: str, label: str) -> List[JsonObject]:
value = hooks.get(event)
if value is None:
return []
if not isinstance(value, list) or not all(isinstance(item, dict) for item in value):
raise ValueError(f'{label} field "hooks.{event}" must be an array of objects')
return value
def _is_managed_command(command: Any, script_name: str, agent_id: str = "") -> bool:
if not isinstance(command, str):
return False
normalized = command.lower()
if ".elydora" not in normalized or script_name not in normalized:
return False
return not agent_id or agent_id in command
def _without_v2_hooks(
entries: List[JsonObject], agent_id: str = ""
) -> List[JsonObject]:
return [
hook
for hook in entries
if not _is_managed_command(hook.get("command"), "guard.py", agent_id)
and not _is_managed_command(hook.get("command"), "hook.py", agent_id)
]
def _build_v2_hook(script_path: str) -> JsonObject:
return {
"matcher": "*",
"command": _build_command(script_path),
"timeout_ms": 5000,
}
def _is_owned_v2(settings: JsonObject, hooks: JsonObject) -> bool:
config_keys = {"name", "description", "tools", "includeMcpJson", "hooks"}
hook_keys = {"preToolUse", "postToolUse"}
return (
all(key in config_keys for key in settings)
and settings.get("name") == V2_AGENT_NAME
and settings.get("description") == V2_DESCRIPTION
and settings.get("tools") == ["*"]
and settings.get("includeMcpJson") is True
and all(key in hook_keys for key in hooks)
and all(isinstance(value, list) and not value for value in hooks.values())
)
def _v3_hooks(settings: JsonObject) -> List[JsonObject]:
if "version" in settings and settings["version"] != "v1":
raise ValueError('Kiro CLI v3 hooks config field "version" must be "v1"')
if "hooks" not in settings:
return []
hooks = settings["hooks"]
if not isinstance(hooks, list) or not all(isinstance(item, dict) for item in hooks):
raise ValueError('Kiro CLI v3 hooks config field "hooks" must be an array of objects')
return hooks
def _v3_action_command(hook: JsonObject) -> Any:
action = hook.get("action")
return action.get("command") if isinstance(action, dict) else None
def _is_managed_v3_hook(hook: JsonObject, agent_id: str = "") -> bool:
if hook.get("name") == V3_GUARD_NAME:
return _is_managed_command(_v3_action_command(hook), "guard.py", agent_id)
if hook.get("name") == V3_AUDIT_NAME:
return _is_managed_command(_v3_action_command(hook), "hook.py", agent_id)
return False return False
def _load_json(path: str) -> dict: def _build_v3_hook(
if not os.path.exists(path): name: str,
return {} description: str,
with open(path, "r", encoding="utf-8") as f: trigger: str,
return json.load(f) script_path: str,
) -> JsonObject:
return {
"name": name,
"description": description,
"trigger": trigger,
"matcher": ".*",
"action": {"type": "command", "command": _build_command(script_path)},
"timeout": 5,
"enabled": True,
}
def _save_json(path: str, data: dict) -> None: def _find_v2_command(entries: List[JsonObject], script_name: str) -> Optional[str]:
os.makedirs(os.path.dirname(path), exist_ok=True) for hook in entries:
with open(path, "w", encoding="utf-8") as f: command = hook.get("command")
json.dump(data, f, indent=2) if _is_managed_command(command, script_name):
f.write("\n") return command
return None
def _find_v3_command(
entries: List[JsonObject], name: str, script_name: str
) -> Optional[str]:
for hook in entries:
command = _v3_action_command(hook)
if hook.get("name") == name and _is_managed_command(command, script_name):
return command
return None
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 _runtime_scripts_exist(contracts: List[HookContract]) -> 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:
is_directory = entry.is_dir(follow_symlinks=False)
except OSError as error:
raise OSError(f"Read Elydora runtime entry at {entry.path}: {error}") from error
if not is_directory:
continue
guard_path = os.path.join(entry.path, "guard.py")
hook_path = os.path.join(entry.path, "hook.py")
references_runtime = any(
guard_path in guard and hook_path in audit for guard, audit, _path in contracts
)
if not references_runtime:
continue
config_path = os.path.join(entry.path, "config.json")
config = _read_json(config_path, "Elydora runtime config")
if config is None or config.get("agent_name") != AGENT_KEY:
continue
return _regular_file_exists(
guard_path, "Elydora guard runtime"
) and _regular_file_exists(hook_path, "Elydora audit runtime")
return False
+2 -2
View File
@@ -28,8 +28,8 @@ SUPPORTED_AGENTS: Dict[str, Dict[str, str]] = {
}, },
"kirocli": { "kirocli": {
"name": "Kiro CLI", "name": "Kiro CLI",
"hook_event": "PostToolUse", "hook_event": "PreToolUse/PostToolUse",
"config_path": "~/.kiro/settings.json", "config_path": "~/.kiro/hooks/elydora-audit.json",
}, },
"kiroide": { "kiroide": {
"name": "Kiro IDE", "name": "Kiro IDE",
+356
View File
@@ -0,0 +1,356 @@
from __future__ import annotations
from dataclasses import dataclass
import json
import os
from pathlib import Path
import subprocess
import sys
from typing import Any
import pytest
from elydora.plugins import kirocli
from elydora.plugins.base import InstallConfig
from elydora.plugins.registry import SUPPORTED_AGENTS
AGENT_ID = "agent-1"
MISSING = object()
@dataclass(frozen=True)
class KiroFixture:
plugin: kirocli.KiroCliPlugin
config: InstallConfig
home_dir: Path
guard_path: Path
hook_path: Path
runtime_config_path: Path
v2_path: Path
v3_path: Path
def write_json_or_text(path: Path, value: Any) -> None:
path.parent.mkdir(parents=True, exist_ok=True)
content = value if isinstance(value, str) else json.dumps(value, indent=2)
path.write_text(content, encoding="utf-8")
def prepare_fixture(
monkeypatch: pytest.MonkeyPatch,
tmp_path: Path,
*,
existing_v2: Any = MISSING,
existing_v3: Any = MISSING,
) -> KiroFixture:
home_dir = tmp_path / "home with spaces"
elydora_dir = home_dir / ".elydora"
agent_dir = elydora_dir / AGENT_ID
v2_path = home_dir / ".kiro" / "agents" / "elydora-audit.json"
v3_path = home_dir / ".kiro" / "hooks" / "elydora-audit.json"
agent_dir.mkdir(parents=True)
guard_path = agent_dir / "guard.py"
guard_path.write_text(
"import sys\nsys.stderr.write('Agent is frozen by Elydora.')\nraise SystemExit(2)\n",
encoding="utf-8",
)
if existing_v2 is not MISSING:
write_json_or_text(v2_path, existing_v2)
if existing_v3 is not MISSING:
write_json_or_text(v3_path, existing_v3)
monkeypatch.setattr(kirocli, "ELYDORA_DIR", str(elydora_dir))
monkeypatch.setattr(kirocli, "V2_AGENT_PATH", str(v2_path))
monkeypatch.setattr(kirocli, "V3_HOOKS_PATH", str(v3_path))
config: InstallConfig = {
"agent_id": AGENT_ID,
"agent_name": "kirocli",
"org_id": "org-1",
"private_key": "test-key",
"kid": "kid-1",
"base_url": "https://api.elydora.test",
"guard_script_path": str(guard_path),
}
return KiroFixture(
plugin=kirocli.KiroCliPlugin(),
config=config,
home_dir=home_dir,
guard_path=guard_path,
hook_path=agent_dir / "hook.py",
runtime_config_path=agent_dir / "config.json",
v2_path=v2_path,
v3_path=v3_path,
)
def find_v3_hook(config: dict[str, Any], name: str) -> dict[str, Any]:
return next(hook for hook in config["hooks"] if hook.get("name") == name)
def run_command(
command: str,
home_dir: Path,
payload: dict[str, Any],
) -> subprocess.CompletedProcess[str]:
env = {
**os.environ,
"HOME": str(home_dir),
"USERPROFILE": str(home_dir),
}
return subprocess.run(
command,
shell=True,
capture_output=True,
check=False,
env=env,
input=json.dumps(payload),
text=True,
)
def test_kirocli_registry_points_at_v3_global_hook_contract() -> None:
assert SUPPORTED_AGENTS["kirocli"] == {
"name": "Kiro CLI",
"hook_event": "PreToolUse/PostToolUse",
"config_path": "~/.kiro/hooks/elydora-audit.json",
}
def test_install_preserves_user_hooks_and_writes_idempotent_v2_v3_contracts(
monkeypatch: pytest.MonkeyPatch,
tmp_path: Path,
capsys: pytest.CaptureFixture[str],
) -> None:
fixture = prepare_fixture(
monkeypatch,
tmp_path,
existing_v2={
"description": "User Kiro agent",
"tools": ["read"],
"hooks": {
"agentSpawn": [{"command": "existing-spawn"}],
"preToolUse": [{"matcher": "read", "command": "existing-v2"}],
},
},
existing_v3={
"version": "v1",
"hooks": [{
"name": "existing-v3",
"trigger": "SessionStart",
"action": {"type": "command", "command": "existing-command"},
}],
},
)
fixture.plugin.install(fixture.config)
fixture.plugin.install(fixture.config)
output = capsys.readouterr().out
assert "kiro-cli --agent elydora-audit" in output
assert "kiro-cli --v3" in output
v2 = json.loads(fixture.v2_path.read_text(encoding="utf-8"))
assert v2["description"] == "User Kiro agent"
assert v2["tools"] == ["read"]
assert v2["hooks"]["agentSpawn"] == [{"command": "existing-spawn"}]
assert len(v2["hooks"]["preToolUse"]) == 2
assert len(v2["hooks"]["postToolUse"]) == 1
assert v2["hooks"]["preToolUse"][1]["matcher"] == "*"
assert v2["hooks"]["preToolUse"][1]["timeout_ms"] == 5000
v3 = json.loads(fixture.v3_path.read_text(encoding="utf-8"))
assert v3["version"] == "v1"
assert len(v3["hooks"]) == 3
assert v3["hooks"][0]["name"] == "existing-v3"
assert find_v3_hook(v3, "elydora-guard") == {
"name": "elydora-guard",
"description": "Block tool use when the Elydora agent is frozen",
"trigger": "PreToolUse",
"matcher": ".*",
"action": {
"type": "command",
"command": find_v3_hook(v3, "elydora-guard")["action"]["command"],
},
"timeout": 5,
"enabled": True,
}
assert find_v3_hook(v3, "elydora-audit")["trigger"] == "PostToolUse"
def test_hook_commands_block_frozen_agents_and_forward_official_event_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(
"import pathlib, sys\n"
f"pathlib.Path({str(capture_path)!r}).write_text(sys.stdin.read(), encoding='utf-8')\n",
encoding="utf-8",
)
v3 = json.loads(fixture.v3_path.read_text(encoding="utf-8"))
payload = {
"hook_event_name": "PreToolUse",
"cwd": str(fixture.home_dir),
"session_id": "session-1",
"tool_name": "execute_bash",
"tool_input": {"command": "echo test"},
}
guard_result = run_command(
find_v3_hook(v3, "elydora-guard")["action"]["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_response"] = {"success": True, "result": "test"}
audit_result = run_command(
find_v3_hook(v3, "elydora-audit")["action"]["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_runtime_scripts(
monkeypatch: pytest.MonkeyPatch,
tmp_path: Path,
) -> None:
fixture = prepare_fixture(monkeypatch, tmp_path)
fixture.plugin.install(fixture.config)
assert fixture.plugin.status() == {
"installed": True,
"agent": "kirocli",
"details": f"Config: {fixture.v3_path}",
}
fixture.v3_path.unlink()
assert fixture.plugin.status()["details"] == f"Config: {fixture.v2_path}"
fixture.plugin.install(fixture.config)
fixture.v2_path.unlink()
assert fixture.plugin.status()["details"] == f"Config: {fixture.v3_path}"
fixture.guard_path.unlink()
status = fixture.plugin.status()
assert status["installed"] is False
assert status["details"] == f"Configured at {fixture.v3_path}; runtime scripts missing"
def test_status_surfaces_malformed_referenced_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_uninstall_preserves_unrelated_v2_and_v3_hooks(
monkeypatch: pytest.MonkeyPatch,
tmp_path: Path,
) -> None:
fixture = prepare_fixture(
monkeypatch,
tmp_path,
existing_v2={
"hooks": {"preToolUse": [{"matcher": "read", "command": "existing-v2"}]},
},
existing_v3={
"version": "v1",
"hooks": [{
"name": "existing-v3",
"trigger": "SessionStart",
"action": {"type": "command", "command": "existing-command"},
}],
},
)
fixture.plugin.install(fixture.config)
fixture.plugin.uninstall(AGENT_ID)
v2 = json.loads(fixture.v2_path.read_text(encoding="utf-8"))
assert v2["hooks"]["preToolUse"] == [{"matcher": "read", "command": "existing-v2"}]
assert v2["hooks"]["postToolUse"] == []
v3 = json.loads(fixture.v3_path.read_text(encoding="utf-8"))
assert v3["hooks"] == [{
"name": "existing-v3",
"trigger": "SessionStart",
"action": {"type": "command", "command": "existing-command"},
}]
def test_uninstall_removes_configs_owned_entirely_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.v2_path.exists() is False
assert fixture.v3_path.exists() is False
@pytest.mark.parametrize("malformed_contract", ["v2", "v3"])
def test_install_preserves_malformed_configs_before_any_runtime_write(
monkeypatch: pytest.MonkeyPatch,
tmp_path: Path,
malformed_contract: str,
) -> None:
fixture = prepare_fixture(
monkeypatch,
tmp_path,
existing_v2="{ malformed" if malformed_contract == "v2" else MISSING,
existing_v3="{ malformed" if malformed_contract == "v3" else MISSING,
)
malformed_path = fixture.v2_path if malformed_contract == "v2" else fixture.v3_path
with pytest.raises(ValueError, match=f"parse Kiro CLI {malformed_contract}"):
fixture.plugin.install(fixture.config)
assert malformed_path.read_text(encoding="utf-8") == "{ malformed"
assert fixture.hook_path.exists() is False
assert fixture.runtime_config_path.exists() is False
@pytest.mark.parametrize(
("existing_v2", "existing_v3", "error_pattern"),
[
({"hooks": None}, MISSING, 'field "hooks" must be an object'),
(MISSING, {"version": None, "hooks": []}, 'field "version" must be "v1"'),
(MISSING, {"version": "v1", "hooks": None}, 'field "hooks" must be an array'),
],
)
def test_install_rejects_invalid_contract_shapes_before_any_runtime_write(
monkeypatch: pytest.MonkeyPatch,
tmp_path: Path,
existing_v2: Any,
existing_v3: Any,
error_pattern: str,
) -> None:
fixture = prepare_fixture(
monkeypatch,
tmp_path,
existing_v2=existing_v2,
existing_v3=existing_v3,
)
with pytest.raises(ValueError, match=error_pattern):
fixture.plugin.install(fixture.config)
assert fixture.hook_path.exists() is False
assert fixture.runtime_config_path.exists() is False