feat: sync Python Grok hook integration

This commit is contained in:
ZacharyZhang-NY committed 2026-07-19 07:38:00 -04:00
1 parent dfad7d45fa
commit 17fee3d1fa
6 files changed
+901 -1

No files matched your search

+1 -1
View File
@@ -259,7 +259,7 @@
}, },
"adapters": { "adapters": {
"node": true, "node": true,
"python": false, "python": true,
"go": false "go": false
}, },
"delivery_state": "partial", "delivery_state": "partial",
+3
View File
@@ -77,6 +77,8 @@ Codex performs a one-time trust review for user hooks. Run `/hooks` in Codex aft
Kimi installation writes the strict hook contract to each detected runtime: Kimi Code's `$KIMI_CODE_HOME/config.toml` (default `~/.kimi-code/config.toml`) and the migrating Python CLI's `~/.kimi/config.toml`. A fresh installation targets current Kimi Code, avoiding cross-runtime migration markers. Both runtimes load the hooks globally; run `/hooks` to inspect them. Kimi installation writes the strict hook contract to each detected runtime: Kimi Code's `$KIMI_CODE_HOME/config.toml` (default `~/.kimi-code/config.toml`) and the migrating Python CLI's `~/.kimi/config.toml`. A fresh installation targets current Kimi Code, avoiding cross-runtime migration markers. Both runtimes load the hooks globally; run `/hooks` to inspect them.
Grok Build installation writes native global hooks to `$GROK_HOME/hooks/elydora-audit.json` (default `~/.grok/hooks/elydora-audit.json`). Project hooks still follow Grok's `/hooks-trust` workflow; Elydora leaves project, Claude Code, and Cursor compatibility files unchanged.
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
@@ -95,6 +97,7 @@ Kiro CLI installation covers both runtime contracts. Kiro CLI v2 uses the genera
| Claude Code | `claudecode` | | Claude Code | `claudecode` |
| OpenAI Codex | `codex` | | OpenAI Codex | `codex` |
| Kimi Code | `kimi` | | Kimi Code | `kimi` |
| Grok Build | `grok` |
| Copilot CLI | `copilot` | | Copilot CLI | `copilot` |
| Cursor | `cursor` | | Cursor | `cursor` |
| Gemini CLI | `gemini` | | Gemini CLI | `gemini` |
+2
View File
@@ -23,6 +23,7 @@ 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
from .plugins.grok import GrokPlugin
from .plugins.kirocli import KiroCliPlugin from .plugins.kirocli import KiroCliPlugin
from .plugins.kiroide import KiroIdePlugin from .plugins.kiroide import KiroIdePlugin
from .plugins.kimi import KimiPlugin from .plugins.kimi import KimiPlugin
@@ -36,6 +37,7 @@ PLUGIN_MAP: Dict[str, Type[AgentPlugin]] = {
"copilot": CopilotPlugin, "copilot": CopilotPlugin,
"cursor": CursorPlugin, "cursor": CursorPlugin,
"gemini": GeminiPlugin, "gemini": GeminiPlugin,
"grok": GrokPlugin,
"kirocli": KiroCliPlugin, "kirocli": KiroCliPlugin,
"kiroide": KiroIdePlugin, "kiroide": KiroIdePlugin,
"kimi": KimiPlugin, "kimi": KimiPlugin,
+423
View File
@@ -0,0 +1,423 @@
"""Grok Build native user-hook integration."""
from __future__ import annotations
from dataclasses import dataclass
import math
import os
import shlex
# subprocess provides argument quoting only; this module starts no process.
import subprocess # nosec B404
import sys
from typing import Any, Dict, List, Optional, Set, Tuple
from ._file_io import (
read_json as _read_json,
regular_file_exists as _regular_file_exists,
remove_file as _remove_file,
require_runtime as _require_runtime,
write_json_atomic as _write_json_atomic,
write_text_atomic as _write_text_atomic,
)
from .base import AgentPlugin, InstallConfig, PluginStatus
from .hook_template import generate_hook_script
AGENT_KEY = "grok"
GUARD_SCRIPT = "guard.py"
AUDIT_SCRIPT = "hook.py"
HOOK_TIMEOUT_SECONDS = 10
CONFIG_FILE = "elydora-audit.json"
ELYDORA_DIR = os.path.join(os.path.expanduser("~"), ".elydora")
JsonObject = Dict[str, Any]
GrokHooks = Dict[str, List[JsonObject]]
@dataclass(frozen=True)
class GrokDocument:
exists: bool
config_path: str
root: JsonObject
hooks: GrokHooks
@dataclass(frozen=True)
class RuntimeContract:
agent_id: str
guard_path: str
audit_path: str
def _home_dir() -> str:
return os.path.expanduser("~")
def _resolve_config_path() -> str:
grok_home = os.environ.get("GROK_HOME") or os.path.join(_home_dir(), ".grok")
return os.path.join(grok_home, "hooks", CONFIG_FILE)
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 _build_handler(script_path: str) -> JsonObject:
return {
"type": "command",
"command": _build_command(script_path),
"timeout": HOOK_TIMEOUT_SECONDS,
}
def _parse_generated_command(command: str) -> Optional[Tuple[str, str]]:
try:
arguments = shlex.split(command, posix=os.name != "nt")
except ValueError:
return None
if os.name == "nt":
arguments = [
value[1:-1]
if len(value) >= 2 and value.startswith('"') and value.endswith('"')
else value
for value in arguments
]
if len(arguments) != 2 or not arguments[0] or not arguments[1]:
return None
return arguments[0], arguments[1]
def _same_path(left: str, right: str) -> bool:
return os.path.normcase(os.path.abspath(left)) == os.path.normcase(
os.path.abspath(right)
)
def _same_agent_id(left: str, right: str) -> bool:
return os.path.normcase(left) == os.path.normcase(right)
def _managed_agent_id(handler: JsonObject, script_name: str) -> Optional[str]:
if (
handler.get("type") != "command"
or handler.get("timeout") != HOOK_TIMEOUT_SECONDS
or not isinstance(handler.get("command"), str)
):
return None
parsed = _parse_generated_command(str(handler["command"]))
if parsed is None:
return None
script_path = parsed[1]
if os.path.normcase(os.path.basename(script_path)) != os.path.normcase(script_name):
return None
agent_dir = os.path.dirname(script_path)
if not _same_path(os.path.dirname(agent_dir), ELYDORA_DIR):
return None
agent_id = os.path.basename(agent_dir)
return agent_id if agent_id not in {"", ".", ".."} else None
def _validate_handler(
value: Any,
event: str,
group_index: int,
handler_index: int,
) -> JsonObject:
label = (
f"Grok hooks config handler hooks.{event}[{group_index}]"
f".hooks[{handler_index}]"
)
if not isinstance(value, dict):
raise ValueError(f"{label} must be an object")
handler = dict(value)
handler_type = handler.get("type")
if handler_type not in ("command", "http"):
raise ValueError(f'{label} has unsupported type "{handler_type}"')
if handler_type == "command" and (
not isinstance(handler.get("command"), str) or not handler["command"]
):
raise ValueError(f"{label} requires a non-empty command")
if handler_type == "http" and (
not isinstance(handler.get("url"), str) or not handler["url"]
):
raise ValueError(f"{label} requires a non-empty url")
timeout = handler.get("timeout")
if timeout is not None and (
isinstance(timeout, bool)
or not isinstance(timeout, (int, float))
or not math.isfinite(timeout)
or timeout <= 0
):
raise ValueError(f"{label} timeout must be a positive number")
return handler
def _validate_group(value: Any, event: str, group_index: int) -> JsonObject:
label = f"Grok hooks config group hooks.{event}[{group_index}]"
if not isinstance(value, dict):
raise ValueError(f"{label} must be an object")
group = dict(value)
matcher = group.get("matcher")
if "matcher" in group and not isinstance(matcher, str):
raise ValueError(f"{label} matcher must be a string")
handlers = group.get("hooks")
if not isinstance(handlers, list):
raise ValueError(f"{label} must contain a hooks array")
group["hooks"] = [
_validate_handler(handler, event, group_index, handler_index)
for handler_index, handler in enumerate(handlers)
]
return group
def _read_hooks(root: JsonObject) -> GrokHooks:
if "hooks" not in root:
return {}
raw_hooks = root["hooks"]
if not isinstance(raw_hooks, dict):
raise ValueError('Grok hooks config field "hooks" must be an object')
hooks: GrokHooks = {}
for event, groups in raw_hooks.items():
if not isinstance(groups, list):
raise ValueError(
f'Grok hooks config field "hooks.{event}" must be an array'
)
hooks[event] = [
_validate_group(group, event, group_index)
for group_index, group in enumerate(groups)
]
return hooks
def _read_config() -> GrokDocument:
config_path = _resolve_config_path()
root = _read_json(config_path, "Grok hooks config")
if root is None:
return GrokDocument(False, config_path, {}, {})
return GrokDocument(True, config_path, root, _read_hooks(root))
def _remove_managed(
groups: List[JsonObject],
script_name: str,
agent_id: str = "",
) -> Tuple[List[JsonObject], bool]:
result: List[JsonObject] = []
changed = False
for group in groups:
if "matcher" in group:
result.append(group)
continue
handlers = group["hooks"]
kept: List[JsonObject] = []
for handler in handlers:
managed_id = _managed_agent_id(handler, script_name)
remove = managed_id is not None and (
not agent_id or _same_agent_id(managed_id, agent_id)
)
if remove:
changed = True
else:
kept.append(handler)
if kept:
result.append(group if len(kept) == len(handlers) else {**group, "hooks": kept})
return result, changed
def _remove_managed_hooks(
hooks: GrokHooks,
agent_id: str = "",
) -> Tuple[GrokHooks, bool]:
result = dict(hooks)
changed = False
for event, script_name in (
("PreToolUse", GUARD_SCRIPT),
("PostToolUse", AUDIT_SCRIPT),
):
groups, event_changed = _remove_managed(
result.get(event, []), script_name, agent_id
)
if not event_changed:
continue
changed = True
if groups:
result[event] = groups
else:
result.pop(event, None)
return result, changed
def _managed_ids(groups: List[JsonObject], script_name: str) -> Dict[str, str]:
result: Dict[str, str] = {}
for group in groups:
if "matcher" in group:
continue
for handler in group["hooks"]:
agent_id = _managed_agent_id(handler, script_name)
if agent_id:
result[os.path.normcase(agent_id)] = agent_id
return result
def _runtime_contracts(hooks: GrokHooks) -> List[RuntimeContract]:
guards = _managed_ids(hooks.get("PreToolUse", []), GUARD_SCRIPT)
audits = _managed_ids(hooks.get("PostToolUse", []), AUDIT_SCRIPT)
contracts: List[RuntimeContract] = []
for normalized_id in sorted(set(guards) & set(audits)):
agent_id = guards[normalized_id]
contracts.append(RuntimeContract(
agent_id=agent_id,
guard_path=os.path.join(ELYDORA_DIR, agent_id, GUARD_SCRIPT),
audit_path=os.path.join(ELYDORA_DIR, agent_id, AUDIT_SCRIPT),
))
return contracts
def _runtime_scripts_exist(contracts: List[RuntimeContract]) -> 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 contract in contracts:
entry = next(
(
candidate
for candidate in entries
if _same_agent_id(candidate.name, contract.agent_id)
),
None,
)
if entry is None:
continue
try:
if not entry.is_dir(follow_symlinks=False):
continue
except OSError as error:
raise OSError(f"Read Elydora runtime entry at {entry.path}: {error}") from error
runtime_config_path = os.path.join(entry.path, "config.json")
runtime_config = _read_json(runtime_config_path, "Elydora runtime config")
if runtime_config is None:
continue
agent_name = runtime_config.get("agent_name")
if not isinstance(agent_name, str):
raise ValueError(
f'Elydora runtime config at {runtime_config_path} field '
'"agent_name" must be a string'
)
if agent_name != AGENT_KEY:
continue
guard_path = os.path.join(entry.path, GUARD_SCRIPT)
audit_path = os.path.join(entry.path, AUDIT_SCRIPT)
return _regular_file_exists(
guard_path, "Elydora guard runtime"
) and _regular_file_exists(audit_path, "Elydora audit runtime")
return False
class GrokPlugin(AgentPlugin):
"""Install Elydora into Grok Build's native global user hooks."""
def install(self, config: InstallConfig) -> None:
agent_id = config.get("agent_id", "")
if not agent_id:
raise ValueError("agent_id is required")
source = _read_config()
guard_path = config.get("guard_script_path", "")
_require_runtime(guard_path, "Elydora guard runtime")
agent_dir = os.path.join(ELYDORA_DIR, agent_id)
audit_path = os.path.join(agent_dir, AUDIT_SCRIPT)
cleaned, _ = _remove_managed_hooks(source.hooks)
hooks: GrokHooks = {
**cleaned,
"PreToolUse": [
*cleaned.get("PreToolUse", []),
{"hooks": [_build_handler(guard_path)]},
],
"PostToolUse": [
*cleaned.get("PostToolUse", []),
{"hooks": [_build_handler(audit_path)]},
],
}
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": AGENT_KEY,
}
audit_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(audit_path, audit_script, 0o700, "Elydora audit runtime")
_write_json_atomic(
source.config_path,
{**source.root, "hooks": hooks},
0o600,
"Grok hooks config",
)
print("Grok Build: global PreToolUse and PostToolUse hooks installed.")
def uninstall(self, agent_id: str = "") -> None:
source = _read_config()
if not source.exists:
return
hooks, changed = _remove_managed_hooks(source.hooks, agent_id)
if not changed:
return
root = dict(source.root)
if hooks:
root["hooks"] = hooks
else:
root.pop("hooks", None)
if root:
_write_json_atomic(
source.config_path,
root,
0o600,
"Grok hooks config",
)
else:
_remove_file(source.config_path, "Grok hooks config")
def status(self) -> PluginStatus:
source = _read_config()
contracts = _runtime_contracts(source.hooks)
if not contracts:
return PluginStatus(
installed=False,
agent=AGENT_KEY,
details="Not installed",
)
installed = _runtime_scripts_exist(contracts)
details = (
f"Config: {source.config_path}"
if installed
else f"Configured at {source.config_path}; runtime scripts missing"
)
return PluginStatus(installed=installed, agent=AGENT_KEY, details=details)
+5
View File
@@ -31,6 +31,11 @@ SUPPORTED_AGENTS: Dict[str, Dict[str, str]] = {
"hook_event": "AfterTool", "hook_event": "AfterTool",
"config_path": "~/.gemini/settings.json", "config_path": "~/.gemini/settings.json",
}, },
"grok": {
"name": "Grok Build",
"hook_event": "PreToolUse/PostToolUse",
"config_path": "~/.grok/hooks/elydora-audit.json",
},
"kirocli": { "kirocli": {
"name": "Kiro CLI", "name": "Kiro CLI",
"hook_event": "PreToolUse/PostToolUse", "hook_event": "PreToolUse/PostToolUse",
+467
View File
@@ -0,0 +1,467 @@
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 import cli
from elydora.plugins import grok
from elydora.plugins.base import InstallConfig
from elydora.plugins.registry import SUPPORTED_AGENTS
AGENT_ID = "agent-1"
MISSING = object()
@dataclass(frozen=True)
class GrokFixture:
plugin: grok.GrokPlugin
config: InstallConfig
home_dir: Path
grok_home: Path
agent_dir: Path
config_path: Path
guard_path: Path
hook_path: Path
runtime_config_path: Path
private_key_path: Path
def prepare_fixture(
monkeypatch: pytest.MonkeyPatch,
tmp_path: Path,
*,
existing_config: object = MISSING,
explicit_grok_home: bool = True,
create_guard: bool = True,
) -> GrokFixture:
home_dir = tmp_path / "home with spaces"
agent_dir = home_dir / ".elydora" / AGENT_ID
grok_home = (
home_dir / "custom grok"
if explicit_grok_home
else home_dir / ".grok"
)
config_path = grok_home / "hooks" / "elydora-audit.json"
guard_path = agent_dir / "guard.py"
hook_path = agent_dir / "hook.py"
agent_dir.mkdir(parents=True)
if create_guard:
guard_path.write_text(
"import sys\nsys.stdin.read()\n"
"sys.stderr.write('Agent is frozen by Elydora.')\n"
"raise SystemExit(2)\n",
encoding="utf-8",
)
if existing_config is not MISSING:
config_path.parent.mkdir(parents=True, exist_ok=True)
raw = (
str(existing_config)
if isinstance(existing_config, str)
else json.dumps(existing_config, indent=2)
)
config_path.write_text(raw, encoding="utf-8")
monkeypatch.setattr(grok, "ELYDORA_DIR", str(agent_dir.parent))
monkeypatch.setattr(grok, "_home_dir", lambda: str(home_dir))
if explicit_grok_home:
monkeypatch.setenv("GROK_HOME", str(grok_home))
else:
monkeypatch.delenv("GROK_HOME", raising=False)
config: InstallConfig = {
"agent_id": AGENT_ID,
"agent_name": "grok",
"org_id": "org-1",
"private_key": "test-key",
"kid": "kid-1",
"base_url": "https://api.elydora.test",
"guard_script_path": str(guard_path),
}
return GrokFixture(
plugin=grok.GrokPlugin(),
config=config,
home_dir=home_dir,
grok_home=grok_home,
agent_dir=agent_dir,
config_path=config_path,
guard_path=guard_path,
hook_path=hook_path,
runtime_config_path=agent_dir / "config.json",
private_key_path=agent_dir / "private.key",
)
def managed_handler(
settings: dict[str, Any], event: str, script_name: str
) -> dict[str, Any]:
for group in settings["hooks"][event]:
if "matcher" in group:
continue
for handler in group["hooks"]:
if script_name in str(handler.get("command")):
return handler
raise AssertionError(f"managed {event} handler not found")
def run_command(
command: str,
home_dir: Path,
payload: str,
) -> subprocess.CompletedProcess[str]:
return subprocess.run(
command,
shell=True,
capture_output=True,
check=False,
env={**os.environ, "HOME": str(home_dir), "USERPROFILE": str(home_dir)},
input=payload,
text=True,
)
def test_grok_is_registered_in_sdk_and_cli() -> None:
assert SUPPORTED_AGENTS["grok"] == {
"name": "Grok Build",
"hook_event": "PreToolUse/PostToolUse",
"config_path": "~/.grok/hooks/elydora-audit.json",
}
assert cli.PLUGIN_MAP["grok"] is grok.GrokPlugin
def test_install_preserves_native_config_and_is_idempotent(
monkeypatch: pytest.MonkeyPatch,
tmp_path: Path,
capsys: pytest.CaptureFixture[str],
) -> None:
existing = {
"schemaVersion": 1,
"hooks": {
"SessionStart": [{
"matcher": "startup",
"hooks": [{
"type": "http",
"url": "https://example.test/hook",
"timeout": 5,
"headers": {"x": "keep"},
}],
"label": "keep group metadata",
}],
"PreToolUse": [{
"matcher": "Bash",
"hooks": [{
"type": "command",
"command": "existing-command",
"timeout": 5,
}],
}],
},
}
fixture = prepare_fixture(
monkeypatch, tmp_path, existing_config=existing
)
fixture.plugin.install(fixture.config)
fixture.plugin.install(fixture.config)
assert "global PreToolUse and PostToolUse hooks installed" in capsys.readouterr().out
settings = json.loads(fixture.config_path.read_text(encoding="utf-8"))
assert settings["schemaVersion"] == 1
assert settings["hooks"]["SessionStart"] == existing["hooks"]["SessionStart"]
assert settings["hooks"]["PreToolUse"][0] == existing["hooks"]["PreToolUse"][0]
assert len(settings["hooks"]["PreToolUse"]) == 2
assert len(settings["hooks"]["PostToolUse"]) == 1
assert set(settings["hooks"]["PreToolUse"][1]) == {"hooks"}
for event, script_name in (
("PreToolUse", "guard.py"),
("PostToolUse", "hook.py"),
):
handler = managed_handler(settings, event, script_name)
assert set(handler) == {"type", "command", "timeout"}
assert handler["type"] == "command"
assert handler["timeout"] == 10
assert json.loads(fixture.runtime_config_path.read_text())["agent_name"] == "grok"
assert fixture.private_key_path.read_text(encoding="utf-8") == "test-key"
assert (fixture.home_dir / ".claude" / "settings.json").exists() is False
assert (fixture.home_dir / ".cursor" / "hooks.json").exists() is False
assert (
fixture.home_dir / ".grok" / "hooks" / "elydora-audit.json"
).exists() is False
def test_empty_home_override_uses_official_default(
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
) -> None:
fixture = prepare_fixture(
monkeypatch, tmp_path, explicit_grok_home=False
)
fixture.plugin.install(fixture.config)
monkeypatch.setenv("GROK_HOME", "")
assert fixture.plugin.status()["installed"] is True
def test_commands_block_and_forward_official_payload_byte_for_byte(
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
) -> None:
fixture = prepare_fixture(monkeypatch, tmp_path)
fixture.plugin.install(fixture.config)
capture_path = tmp_path / "captured-event.json"
fixture.hook_path.write_text(
"from pathlib import Path\nimport sys\n"
f"Path({str(capture_path)!r}).write_text(sys.stdin.read(), encoding='utf-8')\n",
encoding="utf-8",
)
settings = json.loads(fixture.config_path.read_text(encoding="utf-8"))
pre_payload = json.dumps({
"hookEventName": "PreToolUse",
"sessionId": "session-1",
"cwd": str(fixture.home_dir),
"workspaceRoot": str(fixture.home_dir),
"toolName": "Bash",
"toolInput": {"command": "echo test"},
}, separators=(",", ":"))
guard = managed_handler(settings, "PreToolUse", "guard.py")
guard_result = run_command(guard["command"], fixture.home_dir, pre_payload)
assert guard_result.returncode == 2
assert "Agent is frozen by Elydora" in guard_result.stderr
post_payload = json.dumps({
"hookEventName": "PostToolUse",
"sessionId": "session-1",
"cwd": str(fixture.home_dir),
"workspaceRoot": str(fixture.home_dir),
"toolName": "Bash",
"toolInput": {"command": "echo test"},
"toolResult": {"output": "test"},
}, separators=(",", ":"))
audit = managed_handler(settings, "PostToolUse", "hook.py")
audit_result = run_command(audit["command"], fixture.home_dir, post_payload)
assert audit_result.returncode == 0
assert capture_path.read_text(encoding="utf-8") == post_payload
def test_status_requires_complete_pair_and_both_runtime_files(
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
) -> None:
fixture = prepare_fixture(monkeypatch, tmp_path)
fixture.plugin.install(fixture.config)
assert fixture.plugin.status()["installed"] is True
settings = json.loads(fixture.config_path.read_text(encoding="utf-8"))
del settings["hooks"]["PostToolUse"]
fixture.config_path.write_text(json.dumps(settings), encoding="utf-8")
assert fixture.plugin.status()["installed"] is False
fixture.plugin.install(fixture.config)
fixture.guard_path.unlink()
status = fixture.plugin.status()
assert status["installed"] is False
assert "runtime scripts missing" in status["details"]
def test_uninstall_removes_exact_ownership_and_preserves_mixed_handlers(
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
) -> None:
fixture = prepare_fixture(
monkeypatch,
tmp_path,
existing_config={"owner": "user", "hooks": {"Notification": []}},
)
fixture.plugin.install(fixture.config)
settings = json.loads(fixture.config_path.read_text(encoding="utf-8"))
settings["hooks"]["PreToolUse"][-1]["hooks"].append({
"type": "command",
"command": "user-command",
"timeout": 10,
})
settings["hooks"]["PreToolUse"].append({"hooks": [{
"type": "command",
"command": grok._build_command(str(fixture.agent_dir / "guard.py.backup")),
"timeout": 10,
}]})
settings["hooks"]["PreToolUse"].append({"hooks": [{
"type": "command",
"command": grok._build_command(str(
fixture.agent_dir.parent / "agent-10" / "guard.py"
)),
"timeout": 10,
}]})
fixture.config_path.write_text(json.dumps(settings), encoding="utf-8")
uninstall_id = "AGENT-1" if os.name == "nt" else AGENT_ID
fixture.plugin.uninstall(uninstall_id)
remaining = json.loads(fixture.config_path.read_text(encoding="utf-8"))
assert remaining["owner"] == "user"
assert remaining["hooks"]["Notification"] == []
assert len(remaining["hooks"]["PreToolUse"]) == 3
assert remaining["hooks"]["PreToolUse"][0]["hooks"] == [{
"type": "command",
"command": "user-command",
"timeout": 10,
}]
raw = fixture.config_path.read_text(encoding="utf-8")
assert "guard.py.backup" in raw
assert "agent-10" in raw
assert "PostToolUse" not in remaining["hooks"]
def test_install_replaces_stale_handlers_for_every_agent(
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"))
for event, script_name in (
("PreToolUse", "guard.py"),
("PostToolUse", "hook.py"),
):
settings["hooks"][event].append({"hooks": [{
"type": "command",
"command": grok._build_command(str(
fixture.agent_dir.parent / "agent-old" / script_name
)),
"timeout": 10,
}]})
fixture.config_path.write_text(json.dumps(settings), encoding="utf-8")
fixture.plugin.install(fixture.config)
raw = fixture.config_path.read_text(encoding="utf-8")
current = json.loads(raw)
assert "agent-old" not in raw
assert len(current["hooks"]["PreToolUse"]) == 1
assert len(current["hooks"]["PostToolUse"]) == 1
def test_uninstall_preserves_untouched_empty_native_event(
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
) -> None:
fixture = prepare_fixture(
monkeypatch, tmp_path, existing_config={"owner": "user"}
)
fixture.plugin.install(fixture.config)
settings = json.loads(fixture.config_path.read_text(encoding="utf-8"))
settings["hooks"]["PreToolUse"] = []
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 remaining["hooks"]["PreToolUse"] == []
assert "PostToolUse" not in remaining["hooks"]
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_install_preserves_malformed_json_for_recovery(
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
) -> None:
fixture = prepare_fixture(
monkeypatch, tmp_path, existing_config="{ malformed"
)
with pytest.raises(ValueError, match="parse Grok 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
@pytest.mark.parametrize(
"existing",
[
{"hooks": None},
{"hooks": {"PreToolUse": None}},
{"hooks": {"PreToolUse": [None]}},
{"hooks": {"PreToolUse": [{"matcher": None, "hooks": []}]}},
{"hooks": {"PreToolUse": [{"matcher": 1, "hooks": []}]}},
{"hooks": {"PreToolUse": [{"hooks": None}]}},
{"hooks": {"PreToolUse": [{"hooks": [
{"type": "command", "command": ""}
]}]}},
{"hooks": {"PreToolUse": [{"hooks": [
{"type": "file", "command": "x"}
]}]}},
{"hooks": {"PreToolUse": [{"hooks": [
{"type": {"future": True}, "command": "x"}
]}]}},
{"hooks": {"PreToolUse": [{"hooks": [
{"type": "http", "url": ""}
]}]}},
{"hooks": {"PreToolUse": [{"hooks": [
{"type": "command", "command": "x", "timeout": 0}
]}]}},
{"hooks": {"PreToolUse": [{"hooks": [
{"type": "command", "command": "x", "timeout": float("inf")}
]}]}},
],
)
def test_install_rejects_invalid_native_shapes_before_writes(
monkeypatch: pytest.MonkeyPatch,
tmp_path: Path,
existing: dict[str, Any],
) -> None:
fixture = prepare_fixture(
monkeypatch, tmp_path, existing_config=existing
)
original = fixture.config_path.read_text(encoding="utf-8")
with pytest.raises(ValueError):
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
def test_install_rejects_missing_guard_before_creating_files(
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_atomic_writes_leave_no_temporary_files(
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
) -> None:
fixture = prepare_fixture(monkeypatch, tmp_path)
fixture.plugin.install(fixture.config)
for directory in (fixture.agent_dir, fixture.config_path.parent):
assert all(path.suffix != ".tmp" for path in directory.iterdir())