feat: add native Codex hook integration

This commit is contained in:
ZacharyZhang-NY committed 2026-07-19 05:34:47 -04:00
1 parent ea550154b5
commit 104efbc720
5 files changed
+840

No files matched your search

+3
View File
@@ -73,6 +73,8 @@ elydora install \
--kid agent-456-key-v1 --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`. 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
@@ -89,6 +91,7 @@ Kiro CLI installation covers both runtime contracts. Kiro CLI v2 uses the genera
| Agent | Key | | Agent | Key |
|-------|-----| |-------|-----|
| Claude Code | `claudecode` | | Claude Code | `claudecode` |
| OpenAI Codex | `codex` |
| Copilot CLI | `copilot` | | Copilot CLI | `copilot` |
| Cursor | `cursor` | | Cursor | `cursor` |
| Gemini CLI | `gemini` | | Gemini CLI | `gemini` |
+2
View File
@@ -19,6 +19,7 @@ from .plugins.base import AgentPlugin, InstallConfig
from .plugins.registry import SUPPORTED_AGENTS, get_agent_names from .plugins.registry import SUPPORTED_AGENTS, get_agent_names
from .plugins.hook_template import generate_guard_script from .plugins.hook_template import generate_guard_script
from .plugins.claudecode import ClaudeCodePlugin from .plugins.claudecode import ClaudeCodePlugin
from .plugins.codex import CodexPlugin
from .plugins.copilot import CopilotPlugin from .plugins.copilot import CopilotPlugin
from .plugins.cursor import CursorPlugin from .plugins.cursor import CursorPlugin
from .plugins.gemini import GeminiPlugin from .plugins.gemini import GeminiPlugin
@@ -30,6 +31,7 @@ from .plugins.opencode import OpenCodePlugin
PLUGIN_MAP: Dict[str, Type[AgentPlugin]] = { PLUGIN_MAP: Dict[str, Type[AgentPlugin]] = {
"claudecode": ClaudeCodePlugin, "claudecode": ClaudeCodePlugin,
"codex": CodexPlugin,
"copilot": CopilotPlugin, "copilot": CopilotPlugin,
"cursor": CursorPlugin, "cursor": CursorPlugin,
"gemini": GeminiPlugin, "gemini": GeminiPlugin,
+419
View File
@@ -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,
)
+5
View File
@@ -11,6 +11,11 @@ SUPPORTED_AGENTS: Dict[str, Dict[str, str]] = {
"hook_event": "PostToolUse", "hook_event": "PostToolUse",
"config_path": "~/.claude/settings.json", "config_path": "~/.claude/settings.json",
}, },
"codex": {
"name": "OpenAI Codex",
"hook_event": "PreToolUse/PostToolUse",
"config_path": "~/.codex/hooks.json",
},
"copilot": { "copilot": {
"name": "Copilot CLI", "name": "Copilot CLI",
"hook_event": "PostToolUse", "hook_event": "PostToolUse",
+411
View File
@@ -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