diff --git a/AGENTS.md b/AGENTS.md new file mode 100644 index 0000000..29d7fc3 --- /dev/null +++ b/AGENTS.md @@ -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__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 +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. diff --git a/README.md b/README.md index eac31b2..3346abe 100644 --- a/README.md +++ b/README.md @@ -73,6 +73,8 @@ elydora install \ --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 | Command | Description | diff --git a/elydora/plugins/kirocli.py b/elydora/plugins/kirocli.py index 214e1f9..54daec0 100644 --- a/elydora/plugins/kirocli.py +++ b/elydora/plugins/kirocli.py @@ -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 import json import os +import shlex import stat +import subprocess import sys +import tempfile +from typing import Any, Dict, List, Optional, Tuple from .base import AgentPlugin, InstallConfig, PluginStatus 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") +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): - """Install/uninstall Elydora audit hook for Kiro CLI.""" - - @staticmethod - def _hook_path_for(agent_id: str) -> str: - return os.path.join(ELYDORA_DIR, agent_id, "hook.py") + """Install Elydora into both supported Kiro CLI hook generations.""" def install(self, config: InstallConfig) -> None: 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) - 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 - config_data = { + next_v2_settings: JsonObject = { + "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", ""), "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_name, + "agent_name": config.get("agent_name", AGENT_KEY), } - config_path = os.path.join(agent_dir, "config.json") - 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( + hook_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"), ) - hook_path = self._hook_path_for(agent_id) - with open(hook_path, "w", encoding="utf-8") as f: - f.write(script) - try: - os.chmod(hook_path, stat.S_IRWXU) - except Exception: - pass # chmod may fail on Windows + _write_json_atomic(os.path.join(agent_dir, "config.json"), runtime_config, 0o600) + _write_text_atomic( + os.path.join(agent_dir, "private.key"), + config.get("private_key", ""), + 0o600, + ) + _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", "") - python_exe = sys.executable - - 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}") + print('Kiro CLI v2: start with "kiro-cli --agent elydora-audit".') + print('Kiro CLI v3: start with "kiro-cli --v3"; global hooks load automatically.') def uninstall(self, agent_id: str = "") -> None: - if os.path.exists(SETTINGS_PATH): - settings = _load_json(SETTINGS_PATH) - hooks = settings.get("hooks", {}) - changed = False + v2_settings = _read_json(V2_AGENT_PATH, "Kiro CLI v2 agent config") + v3_settings = _read_json(V3_HOOKS_PATH, "Kiro CLI v3 hooks config") - # Remove PreToolUse entries - pre_tool_use = hooks.get("PreToolUse", []) - pre_filtered = [h for h in pre_tool_use if not _is_elydora_hook(h, agent_id)] - if len(pre_filtered) != len(pre_tool_use): - hooks["PreToolUse"] = pre_filtered - if not pre_filtered: - del hooks["PreToolUse"] - changed = True + next_v2: Optional[Tuple[JsonObject, bool]] = None + if v2_settings is not None: + v2_hooks = _hooks_object(v2_settings, "Kiro CLI v2 agent config") + current_pre = _hook_entries(v2_hooks, "preToolUse", "Kiro CLI v2 agent config") + current_post = _hook_entries(v2_hooks, "postToolUse", "Kiro CLI v2 agent config") + next_pre = _without_v2_hooks(current_pre, agent_id) + next_post = _without_v2_hooks(current_post, agent_id) + 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 - post_tool_use = hooks.get("PostToolUse", []) - post_filtered = [h for h in post_tool_use if not _is_elydora_hook(h, agent_id)] - if len(post_filtered) != len(post_tool_use): - hooks["PostToolUse"] = post_filtered - if not post_filtered: - del hooks["PostToolUse"] - changed = True + next_v3: Optional[Tuple[JsonObject, bool]] = None + if v3_settings is not None: + current_hooks = _v3_hooks(v3_settings) + v3_hooks = [ + hook for hook in current_hooks if not _is_managed_v3_hook(hook, agent_id) + ] + if v3_hooks != current_hooks: + 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 not hooks: - del settings["hooks"] - _save_json(SETTINGS_PATH, settings) + if next_v2 is not None: + settings, remove = next_v2 + if remove: + _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) - print("Elydora hook uninstalled from Kiro CLI.") + if next_v3 is not None: + 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: - # Scan ~/.elydora/*/hook.py for any installed hook - import glob as _glob - hook_pattern = os.path.join(ELYDORA_DIR, "*", "hook.py") - hook_files = _glob.glob(hook_pattern) - hook_exists = len(hook_files) > 0 + v2_settings = _read_json(V2_AGENT_PATH, "Kiro CLI v2 agent config") + v3_settings = _read_json(V3_HOOKS_PATH, "Kiro CLI v3 hooks config") + contracts: List[HookContract] = [] - settings_configured = False - if os.path.exists(SETTINGS_PATH): - settings = _load_json(SETTINGS_PATH) - hooks = settings.get("hooks", {}) - pre_tool_use = hooks.get("PreToolUse", []) - post_tool_use = hooks.get("PostToolUse", []) - pre_configured = any(_is_elydora_hook(h) for h in pre_tool_use) - post_configured = any(_is_elydora_hook(h) for h in post_tool_use) - settings_configured = pre_configured and post_configured + if v2_settings is not None: + v2_hooks = _hooks_object(v2_settings, "Kiro CLI v2 agent config") + guard = _find_v2_command( + _hook_entries(v2_hooks, "preToolUse", "Kiro CLI v2 agent config"), + "guard.py", + ) + audit = _find_v2_command( + _hook_entries(v2_hooks, "postToolUse", "Kiro CLI v2 agent config"), + "hook.py", + ) + if guard and audit: + contracts.append((guard, audit, V2_AGENT_PATH)) - installed = hook_exists and settings_configured - if installed: - details = f"Found {len(hook_files)} agent(s): {', '.join(hook_files)}" - elif hook_exists: - details = "Hook script exists but not configured in settings" - elif settings_configured: - details = "Configured in settings but hook script missing" - else: - details = "Not installed" + if v3_settings is not None: + v3_hooks = _v3_hooks(v3_settings) + guard = _find_v3_command(v3_hooks, V3_GUARD_NAME, "guard.py") + audit = _find_v3_command(v3_hooks, V3_AUDIT_NAME, "hook.py") + if guard and audit: + contracts.append((guard, audit, V3_HOOKS_PATH)) - return PluginStatus(installed=installed, agent="kirocli", details=details) + if not contracts: + return PluginStatus(installed=False, agent=AGENT_KEY, details="Not installed") - -def _is_elydora_hook(entry: dict, agent_id: str = "") -> bool: - # Collect all command strings from the entry - commands = [] - inner_hooks = entry.get("hooks") - if isinstance(inner_hooks, list): - commands.extend( - h.get("command", "") - for h in inner_hooks - if isinstance(h, dict) + config_path = contracts[-1][2] + installed = _runtime_scripts_exist(contracts) + details = ( + f"Config: {config_path}" + if installed + else f"Configured at {config_path}; runtime scripts missing" ) - else: - commands.append(entry.get("command", "")) + return PluginStatus(installed=installed, agent=AGENT_KEY, details=details) - for cmd in commands: - cmd_lower = cmd.lower() - if "elydora" not in cmd_lower: - continue - # If agent_id is specified, only match hooks for that specific agent - if agent_id and agent_id in cmd: - return True - if not agent_id: - return True + +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 + + +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 -def _load_json(path: str) -> dict: - if not os.path.exists(path): - return {} - with open(path, "r", encoding="utf-8") as f: - return json.load(f) +def _build_v3_hook( + name: str, + description: str, + trigger: str, + 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: - os.makedirs(os.path.dirname(path), exist_ok=True) - with open(path, "w", encoding="utf-8") as f: - json.dump(data, f, indent=2) - f.write("\n") +def _find_v2_command(entries: List[JsonObject], script_name: str) -> Optional[str]: + for hook in entries: + command = hook.get("command") + if _is_managed_command(command, script_name): + 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 diff --git a/elydora/plugins/registry.py b/elydora/plugins/registry.py index 54de588..7982674 100644 --- a/elydora/plugins/registry.py +++ b/elydora/plugins/registry.py @@ -28,8 +28,8 @@ SUPPORTED_AGENTS: Dict[str, Dict[str, str]] = { }, "kirocli": { "name": "Kiro CLI", - "hook_event": "PostToolUse", - "config_path": "~/.kiro/settings.json", + "hook_event": "PreToolUse/PostToolUse", + "config_path": "~/.kiro/hooks/elydora-audit.json", }, "kiroide": { "name": "Kiro IDE", diff --git a/tests/test_kirocli_plugin.py b/tests/test_kirocli_plugin.py new file mode 100644 index 0000000..9e11761 --- /dev/null +++ b/tests/test_kirocli_plugin.py @@ -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