diff --git a/README.md b/README.md index 3346abe..ea800cb 100644 --- a/README.md +++ b/README.md @@ -73,6 +73,8 @@ elydora install \ --kid agent-456-key-v1 ``` +Codex performs a one-time trust review for user hooks. Run `/hooks` in Codex after installation and trust the Elydora `PreToolUse` and `PostToolUse` definitions. + 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 @@ -89,6 +91,7 @@ Kiro CLI installation covers both runtime contracts. Kiro CLI v2 uses the genera | Agent | Key | |-------|-----| | Claude Code | `claudecode` | +| OpenAI Codex | `codex` | | Copilot CLI | `copilot` | | Cursor | `cursor` | | Gemini CLI | `gemini` | diff --git a/elydora/cli.py b/elydora/cli.py index 721bb99..899c12c 100644 --- a/elydora/cli.py +++ b/elydora/cli.py @@ -19,6 +19,7 @@ from .plugins.base import AgentPlugin, InstallConfig from .plugins.registry import SUPPORTED_AGENTS, get_agent_names from .plugins.hook_template import generate_guard_script from .plugins.claudecode import ClaudeCodePlugin +from .plugins.codex import CodexPlugin from .plugins.copilot import CopilotPlugin from .plugins.cursor import CursorPlugin from .plugins.gemini import GeminiPlugin @@ -30,6 +31,7 @@ from .plugins.opencode import OpenCodePlugin PLUGIN_MAP: Dict[str, Type[AgentPlugin]] = { "claudecode": ClaudeCodePlugin, + "codex": CodexPlugin, "copilot": CopilotPlugin, "cursor": CursorPlugin, "gemini": GeminiPlugin, diff --git a/elydora/plugins/codex.py b/elydora/plugins/codex.py new file mode 100644 index 0000000..0d1d1b9 --- /dev/null +++ b/elydora/plugins/codex.py @@ -0,0 +1,419 @@ +"""OpenAI Codex lifecycle-hook integration.""" + +from __future__ import annotations + +import json +import os +import shlex +import stat +# subprocess provides argument quoting only; this module starts no process. +import subprocess # nosec B404 +import sys +import tempfile +from typing import Any, Dict, List, Optional + +from .base import AgentPlugin, InstallConfig, PluginStatus +from .hook_template import generate_hook_script + + +AGENT_KEY = "codex" +OWNED_DESCRIPTION = "Elydora audit and freeze enforcement" +GUARD_STATUS = "Checking Elydora agent state" +AUDIT_STATUS = "Recording Elydora tool use" +GUARD_SCRIPT = "guard.py" +AUDIT_SCRIPT = "hook.py" +ELYDORA_DIR = os.path.join(os.path.expanduser("~"), ".elydora") +CONFIG_PATH = os.path.join(os.path.expanduser("~"), ".codex", "hooks.json") + +JsonObject = Dict[str, Any] + + +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 _cleanup_failed_write(path: str, label: str, cause: Exception) -> None: + 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="\n") 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: + encoded = json.dumps(value, indent=2) + "\n" + _write_text_atomic(path, encoded, 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 _build_handler(script_path: str, status_message: str) -> JsonObject: + return { + "type": "command", + "command": f"{shlex.quote(sys.executable)} {shlex.quote(script_path)}", + "commandWindows": subprocess.list2cmdline([sys.executable, script_path]), + "timeout": 10, + "statusMessage": status_message, + } + + +def _hooks_object(settings: JsonObject) -> JsonObject: + if "hooks" not in settings: + return {} + hooks = settings["hooks"] + if not isinstance(hooks, dict): + raise ValueError('Codex hooks config field "hooks" must be an object') + return dict(hooks) + + +def _event_groups(hooks: JsonObject, event: str) -> List[JsonObject]: + if event not in hooks: + return [] + groups = hooks[event] + if not isinstance(groups, list) or not all( + isinstance(group, dict) for group in groups + ): + raise ValueError( + f'Codex hooks config field "hooks.{event}" must be an array of objects' + ) + return groups + + +def _is_managed_command(command: Any, script_name: str, agent_id: str = "") -> bool: + if not isinstance(command, str): + return False + normalized = os.path.normcase(command) + if ".elydora" not in normalized or script_name.lower() not in normalized.lower(): + return False + if not agent_id: + return True + expected_path = os.path.normcase( + os.path.join(ELYDORA_DIR, agent_id, script_name) + ) + return expected_path in normalized + + +def _is_elydora_handler(handler: JsonObject, agent_id: str = "") -> bool: + status_message = handler.get("statusMessage") + if status_message == GUARD_STATUS: + script_name = GUARD_SCRIPT + elif status_message == AUDIT_STATUS: + script_name = AUDIT_SCRIPT + else: + return False + return any( + _is_managed_command(handler.get(key), script_name, agent_id) + for key in ("command", "commandWindows") + ) + + +def _without_elydora( + groups: List[JsonObject], + agent_id: str = "", +) -> List[JsonObject]: + filtered_groups: List[JsonObject] = [] + for group in groups: + handlers = group.get("hooks") + if not isinstance(handlers, list) or not all( + isinstance(handler, dict) for handler in handlers + ): + raise ValueError("Codex hook matcher group must contain a hooks array") + filtered = [ + handler + for handler in handlers + if not _is_elydora_handler(handler, agent_id) + ] + if filtered: + filtered_groups.append({**group, "hooks": filtered}) + return filtered_groups + + +def _find_handler( + groups: List[JsonObject], + status_message: str, +) -> Optional[JsonObject]: + for group in groups: + handlers = group.get("hooks") + if not isinstance(handlers, list) or not all( + isinstance(handler, dict) for handler in handlers + ): + raise ValueError("Codex hook matcher group must contain a hooks array") + for handler in handlers: + if ( + handler.get("statusMessage") == status_message + and _is_elydora_handler(handler) + ): + return handler + return None + + +def _command_references(handler: JsonObject, script_path: str) -> bool: + return any( + isinstance(handler.get(key), str) and script_path in handler[key] + for key in ("command", "commandWindows") + ) + + +def _is_owned_settings(settings: JsonObject, hooks: JsonObject) -> bool: + settings_keys = {"description", "hooks"} + hook_keys = {"PreToolUse", "PostToolUse"} + return ( + all(key in settings_keys for key in settings) + and settings.get("description") == OWNED_DESCRIPTION + and all(key in hook_keys for key in hooks) + and all(isinstance(value, list) and not value for value in hooks.values()) + ) + + +def _runtime_scripts_exist(guard: JsonObject, audit: JsonObject) -> 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_SCRIPT) + hook_path = os.path.join(entry.path, AUDIT_SCRIPT) + if not ( + _command_references(guard, guard_path) + and _command_references(audit, hook_path) + ): + continue + + config_path = os.path.join(entry.path, "config.json") + config = _read_json(config_path, "Elydora runtime config") + if config is None: + continue + agent_name = 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(hook_path, "Elydora audit runtime") + return False + + +class CodexPlugin(AgentPlugin): + """Install Elydora into Codex user lifecycle hooks.""" + + def install(self, config: InstallConfig) -> None: + agent_id = config.get("agent_id", "") + if not agent_id: + raise ValueError("agent_id is required") + + existing_settings = _read_json(CONFIG_PATH, "Codex hooks config") + settings = ( + existing_settings + if existing_settings is not None + else {"description": OWNED_DESCRIPTION} + ) + hooks = _hooks_object(settings) + pre_tool_use = _without_elydora(_event_groups(hooks, "PreToolUse")) + post_tool_use = _without_elydora(_event_groups(hooks, "PostToolUse")) + + guard_path = config.get("guard_script_path", "") + _require_runtime(guard_path, "Elydora guard runtime") + agent_dir = os.path.join(ELYDORA_DIR, agent_id) + hook_path = os.path.join(agent_dir, AUDIT_SCRIPT) + hooks["PreToolUse"] = [ + *pre_tool_use, + { + "matcher": "*", + "hooks": [_build_handler(guard_path, GUARD_STATUS)], + }, + ] + hooks["PostToolUse"] = [ + *post_tool_use, + { + "matcher": "*", + "hooks": [_build_handler(hook_path, AUDIT_STATUS)], + }, + ] + + runtime_config: JsonObject = { + "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": config.get("agent_name", AGENT_KEY) or AGENT_KEY, + } + 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"), + ) + _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( + hook_path, + hook_script, + 0o700, + "Elydora audit runtime", + ) + _write_json_atomic( + CONFIG_PATH, + {**settings, "hooks": hooks}, + 0o600, + "Codex hooks config", + ) + print("Codex: run /hooks to review and trust the Elydora hooks.") + + def uninstall(self, agent_id: str = "") -> None: + settings = _read_json(CONFIG_PATH, "Codex hooks config") + if settings is None: + return + hooks = _hooks_object(settings) + hooks["PreToolUse"] = _without_elydora( + _event_groups(hooks, "PreToolUse"), + agent_id, + ) + hooks["PostToolUse"] = _without_elydora( + _event_groups(hooks, "PostToolUse"), + agent_id, + ) + if _is_owned_settings(settings, hooks): + _remove_file(CONFIG_PATH, "Codex hooks config") + else: + _write_json_atomic( + CONFIG_PATH, + {**settings, "hooks": hooks}, + 0o600, + "Codex hooks config", + ) + + def status(self) -> PluginStatus: + settings = _read_json(CONFIG_PATH, "Codex hooks config") + guard: Optional[JsonObject] = None + audit: Optional[JsonObject] = None + if settings is not None: + hooks = _hooks_object(settings) + guard = _find_handler(_event_groups(hooks, "PreToolUse"), GUARD_STATUS) + audit = _find_handler(_event_groups(hooks, "PostToolUse"), AUDIT_STATUS) + + if guard is None or audit is None: + return PluginStatus( + installed=False, + agent=AGENT_KEY, + details="Not installed", + ) + installed = _runtime_scripts_exist(guard, audit) + details = ( + f"Config: {CONFIG_PATH}" + if installed + else f"Configured at {CONFIG_PATH}; runtime scripts missing" + ) + return PluginStatus( + installed=installed, + agent=AGENT_KEY, + details=details, + ) diff --git a/elydora/plugins/registry.py b/elydora/plugins/registry.py index 7982674..3b05e85 100644 --- a/elydora/plugins/registry.py +++ b/elydora/plugins/registry.py @@ -11,6 +11,11 @@ SUPPORTED_AGENTS: Dict[str, Dict[str, str]] = { "hook_event": "PostToolUse", "config_path": "~/.claude/settings.json", }, + "codex": { + "name": "OpenAI Codex", + "hook_event": "PreToolUse/PostToolUse", + "config_path": "~/.codex/hooks.json", + }, "copilot": { "name": "Copilot CLI", "hook_event": "PostToolUse", diff --git a/tests/test_codex_plugin.py b/tests/test_codex_plugin.py new file mode 100644 index 0000000..71c3493 --- /dev/null +++ b/tests/test_codex_plugin.py @@ -0,0 +1,411 @@ +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 elydora.plugins import codex +from elydora.plugins.base import InstallConfig +from elydora.plugins.registry import SUPPORTED_AGENTS + + +AGENT_ID = "agent-1" +MISSING = object() +GUARD_STATUS = "Checking Elydora agent state" +AUDIT_STATUS = "Recording Elydora tool use" + + +@dataclass(frozen=True) +class CodexFixture: + plugin: codex.CodexPlugin + config: InstallConfig + home_dir: Path + agent_dir: Path + config_path: Path + guard_path: Path + hook_path: Path + runtime_config_path: Path + private_key_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_settings: Any = MISSING, + create_guard: bool = True, +) -> CodexFixture: + home_dir = tmp_path / "home with spaces" + elydora_dir = home_dir / ".elydora" + agent_dir = elydora_dir / AGENT_ID + config_path = home_dir / ".codex" / "hooks.json" + agent_dir.mkdir(parents=True) + + guard_path = agent_dir / "guard.py" + if create_guard: + guard_path.write_text( + "import sys\nsys.stderr.write('Agent is frozen by Elydora.')\n" + "raise SystemExit(2)\n", + encoding="utf-8", + ) + if existing_settings is not MISSING: + write_json_or_text(config_path, existing_settings) + + monkeypatch.setattr(codex, "ELYDORA_DIR", str(elydora_dir)) + monkeypatch.setattr(codex, "CONFIG_PATH", str(config_path)) + config: InstallConfig = { + "agent_id": AGENT_ID, + "agent_name": "codex", + "org_id": "org-1", + "private_key": "test-key", + "kid": "kid-1", + "base_url": "https://api.elydora.test", + "guard_script_path": str(guard_path), + } + return CodexFixture( + plugin=codex.CodexPlugin(), + config=config, + home_dir=home_dir, + agent_dir=agent_dir, + config_path=config_path, + guard_path=guard_path, + hook_path=agent_dir / "hook.py", + runtime_config_path=agent_dir / "config.json", + private_key_path=agent_dir / "private.key", + ) + + +def find_handler( + settings: dict[str, Any], + event: str, + status_message: str, +) -> dict[str, Any]: + for group in settings["hooks"][event]: + for handler in group["hooks"]: + if handler.get("statusMessage") == status_message: + return handler + raise AssertionError(f"handler {status_message!r} not found") + + +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_codex_registry_points_at_global_hooks_contract() -> None: + assert SUPPORTED_AGENTS["codex"] == { + "name": "OpenAI Codex", + "hook_event": "PreToolUse/PostToolUse", + "config_path": "~/.codex/hooks.json", + } + + +def test_install_preserves_existing_hooks_and_is_idempotent( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, + capsys: pytest.CaptureFixture[str], +) -> None: + fixture = prepare_fixture( + monkeypatch, + tmp_path, + existing_settings={ + "description": "Workspace hooks", + "hooks": { + "SessionStart": [{ + "hooks": [{"type": "command", "command": "existing-command"}], + }], + }, + }, + ) + + fixture.plugin.install(fixture.config) + fixture.plugin.install(fixture.config) + + assert "run /hooks to review and trust" in capsys.readouterr().out + settings = json.loads(fixture.config_path.read_text(encoding="utf-8")) + assert settings["description"] == "Workspace hooks" + assert settings["hooks"]["SessionStart"][0]["hooks"][0]["command"] == "existing-command" + assert len(settings["hooks"]["PreToolUse"]) == 1 + assert len(settings["hooks"]["PostToolUse"]) == 1 + assert settings["hooks"]["PreToolUse"][0]["matcher"] == "*" + guard = find_handler(settings, "PreToolUse", GUARD_STATUS) + assert guard["type"] == "command" + assert guard["timeout"] == 10 + assert isinstance(guard["command"], str) + assert isinstance(guard["commandWindows"], str) + + +def test_commands_block_frozen_agents_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( + "import pathlib, sys\n" + f"pathlib.Path({str(capture_path)!r}).write_text(sys.stdin.read(), encoding='utf-8')\n", + encoding="utf-8", + ) + settings = json.loads(fixture.config_path.read_text(encoding="utf-8")) + command_key = "commandWindows" if os.name == "nt" else "command" + payload = { + "hook_event_name": "PreToolUse", + "session_id": "session-1", + "turn_id": "turn-1", + "transcript_path": None, + "cwd": str(fixture.home_dir), + "model": "gpt-5", + "permission_mode": "default", + "tool_name": "Bash", + "tool_use_id": "call-1", + "tool_input": {"command": "echo test"}, + } + + guard = find_handler(settings, "PreToolUse", GUARD_STATUS) + guard_result = run_command(guard[command_key], 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"] = {"output": "test"} + audit = find_handler(settings, "PostToolUse", AUDIT_STATUS) + audit_result = run_command(audit[command_key], fixture.home_dir, payload) + assert audit_result.returncode == 0 + assert json.loads(capture_path.read_text(encoding="utf-8")) == payload + + +def test_status_requires_both_runtimes_and_uninstall_preserves_other_hooks( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + existing_handler = {"type": "command", "command": "existing-command"} + fixture = prepare_fixture( + monkeypatch, + tmp_path, + existing_settings={ + "hooks": {"PreToolUse": [{"hooks": [existing_handler]}]}, + }, + ) + fixture.plugin.install(fixture.config) + assert fixture.plugin.status() == { + "installed": True, + "agent": "codex", + "details": f"Config: {fixture.config_path}", + } + + fixture.guard_path.unlink() + assert fixture.plugin.status() == { + "installed": False, + "agent": "codex", + "details": f"Configured at {fixture.config_path}; runtime scripts missing", + } + + fixture.plugin.uninstall(AGENT_ID) + settings = json.loads(fixture.config_path.read_text(encoding="utf-8")) + assert settings["hooks"]["PreToolUse"] == [{"hooks": [existing_handler]}] + assert settings["hooks"]["PostToolUse"] == [] + + +def test_install_preserves_user_handlers_with_matching_status_text( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + user_guard = { + "type": "command", + "command": "user-guard", + "statusMessage": GUARD_STATUS, + } + user_audit = { + "type": "command", + "command": "user-audit", + "statusMessage": AUDIT_STATUS, + } + fixture = prepare_fixture( + monkeypatch, + tmp_path, + existing_settings={ + "hooks": { + "PreToolUse": [{"matcher": "Bash", "hooks": [user_guard]}], + "PostToolUse": [{"matcher": "Bash", "hooks": [user_audit]}], + }, + }, + ) + + fixture.plugin.install(fixture.config) + + settings = json.loads(fixture.config_path.read_text(encoding="utf-8")) + assert settings["hooks"]["PreToolUse"][0]["hooks"][0] == user_guard + assert settings["hooks"]["PostToolUse"][0]["hooks"][0] == user_audit + assert len(settings["hooks"]["PreToolUse"]) == 2 + assert len(settings["hooks"]["PostToolUse"]) == 2 + + +def test_install_rejects_missing_guard_before_runtime_or_config_writes( + 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.config_path.exists() is False + 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_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_status_surfaces_malformed_matcher_groups( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + fixture = prepare_fixture(monkeypatch, tmp_path) + fixture.plugin.install(fixture.config) + settings = json.loads(fixture.config_path.read_text(encoding="utf-8")) + settings["hooks"]["PreToolUse"][0]["hooks"] = None + fixture.config_path.write_text(json.dumps(settings), encoding="utf-8") + + with pytest.raises(ValueError, match="matcher group must contain a hooks array"): + fixture.plugin.status() + + +def test_uninstall_removes_config_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.config_path.exists() is False + + +def test_uninstall_matches_the_exact_agent_runtime_directory( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + fixture = prepare_fixture(monkeypatch, tmp_path) + fixture.plugin.install(fixture.config) + settings = json.loads(fixture.config_path.read_text(encoding="utf-8")) + other_agent_dir = fixture.agent_dir.parent / "agent-10" + + for event, status in ( + ("PreToolUse", GUARD_STATUS), + ("PostToolUse", AUDIT_STATUS), + ): + handler = find_handler(settings, event, status) + other_handler = { + **handler, + "command": handler["command"].replace( + str(fixture.agent_dir), str(other_agent_dir) + ), + "commandWindows": handler["commandWindows"].replace( + str(fixture.agent_dir), str(other_agent_dir) + ), + } + settings["hooks"][event].append( + {"matcher": "*", "hooks": [other_handler]} + ) + fixture.config_path.write_text(json.dumps(settings), encoding="utf-8") + + fixture.plugin.uninstall(AGENT_ID) + + remaining = json.loads(fixture.config_path.read_text(encoding="utf-8")) + assert len(remaining["hooks"]["PreToolUse"]) == 1 + assert len(remaining["hooks"]["PostToolUse"]) == 1 + assert str(other_agent_dir) in find_handler( + remaining, "PreToolUse", GUARD_STATUS + )["command"] + + +@pytest.mark.parametrize( + ("existing_settings", "error_pattern"), + [ + ({"hooks": None}, 'field "hooks" must be an object'), + ({"hooks": {"PreToolUse": None}}, 'field "hooks.PreToolUse" must be an array'), + ( + {"hooks": {"PreToolUse": [{"hooks": None}]}}, + "matcher group must contain a hooks array", + ), + ], +) +def test_install_preserves_invalid_shapes_before_runtime_writes( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, + existing_settings: dict[str, Any], + error_pattern: str, +) -> None: + fixture = prepare_fixture( + monkeypatch, + tmp_path, + existing_settings=existing_settings, + ) + original = fixture.config_path.read_text(encoding="utf-8") + + with pytest.raises(ValueError, match=error_pattern): + fixture.plugin.install(fixture.config) + + assert fixture.config_path.read_text(encoding="utf-8") == original + 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_preserves_malformed_config_before_runtime_writes( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + fixture = prepare_fixture( + monkeypatch, + tmp_path, + existing_settings="{ malformed", + ) + + with pytest.raises(ValueError, match="parse Codex hooks config"): + fixture.plugin.install(fixture.config) + + assert fixture.config_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