feat: sync Python Auggie hook integration

This commit is contained in:
ZacharyZhang-NY committed 2026-07-19 08:38:41 -04:00
1 parent c32fa72a68
commit e2f319914b
7 files changed
+1099 -1

No files matched your search

+3
View File
@@ -79,6 +79,8 @@ Kimi installation writes the strict hook contract to each detected runtime: Kimi
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.
Auggie installation writes user-level hooks to `~/.augment/settings.json` and creates the `.cmd` or `.sh` wrappers required by its command runner. System and workspace settings remain unchanged. Sessions started with `--augment-cache-dir` load settings from that alternate directory.
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
@@ -94,6 +96,7 @@ Kiro CLI installation covers both runtime contracts. Kiro CLI v2 uses the genera
| Agent | Key |
|-------|-----|
| Augment Code CLI | `augment` |
| Claude Code | `claudecode` |
| OpenAI Codex | `codex` |
| Kimi Code | `kimi` |
+2
View File
@@ -18,6 +18,7 @@ from .crypto import get_public_key_base64url
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.augment import AugmentPlugin
from .plugins.claudecode import ClaudeCodePlugin
from .plugins.codex import CodexPlugin
from .plugins.copilot import CopilotPlugin
@@ -32,6 +33,7 @@ from .plugins.opencode import OpenCodePlugin
PLUGIN_MAP: Dict[str, Type[AgentPlugin]] = {
"augment": AugmentPlugin,
"claudecode": ClaudeCodePlugin,
"codex": CodexPlugin,
"copilot": CopilotPlugin,
+227
View File
@@ -0,0 +1,227 @@
"""Augment Code CLI native user-hook integration."""
from __future__ import annotations
import os
from typing import List
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 .augment_contract import (
AGENT_KEY,
AUDIT_SCRIPT,
AugmentDocument,
JsonObject,
RuntimeContract,
build_handler,
build_wrapper,
elydora_dir,
read_hooks,
remove_managed_hooks,
resolve_config_path,
runtime_contracts,
wrapper_paths,
)
from .base import AgentPlugin, InstallConfig, PluginStatus
from .hook_template import generate_hook_script
def _read_config() -> AugmentDocument:
config_path = resolve_config_path()
root = _read_json(config_path, "Auggie settings")
if root is None:
return AugmentDocument(False, config_path, {}, {})
return AugmentDocument(True, config_path, root, read_hooks(root))
def _same_agent_id(left: str, right: str) -> bool:
return os.path.normcase(left) == os.path.normcase(right)
def _runtime_files_exist(contracts: List[RuntimeContract]) -> bool:
root = elydora_dir()
try:
with os.scandir(root) 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 {root}: {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
files = (
(contract.guard_path, "Elydora guard runtime"),
(contract.audit_path, "Elydora audit runtime"),
(contract.guard_wrapper_path, "Auggie guard wrapper"),
(contract.audit_wrapper_path, "Auggie audit wrapper"),
)
return all(_regular_file_exists(file_path, label) for file_path, label in files)
return False
class AugmentPlugin(AgentPlugin):
"""Install Elydora into Auggie's global user settings."""
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_directory = os.path.join(elydora_dir(), agent_id)
audit_path = os.path.join(agent_directory, AUDIT_SCRIPT)
wrappers = wrapper_paths(agent_id)
cleaned, _ = remove_managed_hooks(source.hooks)
hooks = {
**cleaned,
"PreToolUse": [
*cleaned.get("PreToolUse", []),
{
"matcher": ".*",
"hooks": [build_handler(wrappers.guard_path)],
},
],
"PostToolUse": [
*cleaned.get("PostToolUse", []),
{
"matcher": ".*",
"hooks": [build_handler(wrappers.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_directory, "config.json"),
runtime_config,
0o600,
"Elydora runtime config",
)
_write_text_atomic(
os.path.join(agent_directory, "private.key"),
config.get("private_key", ""),
0o600,
"Elydora private key",
)
_write_text_atomic(
audit_path,
audit_script,
0o700,
"Elydora audit runtime",
)
_write_text_atomic(
wrappers.guard_path,
build_wrapper(guard_path),
0o700,
"Auggie guard wrapper",
)
_write_text_atomic(
wrappers.audit_path,
build_wrapper(audit_path),
0o700,
"Auggie audit wrapper",
)
_write_json_atomic(
source.config_path,
{**source.root, "hooks": hooks},
0o600,
"Auggie settings",
)
print("Auggie: user-level 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,
"Auggie settings",
)
else:
_remove_file(source.config_path, "Auggie settings")
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_files_exist(contracts)
details = (
f"Config: {source.config_path}"
if installed
else (
f"Configured at {source.config_path}; "
"runtime scripts or wrappers missing"
)
)
return PluginStatus(
installed=installed,
agent=AGENT_KEY,
details=details,
)
@@ -0,0 +1,363 @@
"""Auggie hook contract parsing and exact Elydora ownership."""
from __future__ import annotations
from dataclasses import dataclass
import math
import os
import re
import sys
from typing import Any, Dict, List, Optional, Set, Tuple
AGENT_KEY = "augment"
GUARD_SCRIPT = "guard.py"
AUDIT_SCRIPT = "hook.py"
HOOK_TIMEOUT_MILLISECONDS = 10_000
_WRAPPER_EXTENSION = ".cmd" if os.name == "nt" else ".sh"
GUARD_WRAPPER = f"augment-guard{_WRAPPER_EXTENSION}"
AUDIT_WRAPPER = f"augment-hook{_WRAPPER_EXTENSION}"
_TOOL_EVENTS = {"PreToolUse", "PostToolUse"}
_SESSION_EVENTS = {
"Stop",
"SessionStart",
"SessionEnd",
"Notification",
"PromptSubmit",
}
JsonObject = Dict[str, Any]
AugmentHooks = Dict[str, List[JsonObject]]
@dataclass(frozen=True)
class AugmentDocument:
exists: bool
config_path: str
root: JsonObject
hooks: AugmentHooks
@dataclass(frozen=True)
class WrapperPaths:
guard_path: str
audit_path: str
@dataclass(frozen=True)
class RuntimeContract:
agent_id: str
guard_path: str
audit_path: str
guard_wrapper_path: str
audit_wrapper_path: str
def home_dir() -> str:
return os.path.expanduser("~")
def elydora_dir() -> str:
return os.path.join(home_dir(), ".elydora")
def resolve_config_path() -> str:
return os.path.join(home_dir(), ".augment", "settings.json")
def wrapper_paths(agent_id: str) -> WrapperPaths:
agent_directory = os.path.join(elydora_dir(), agent_id)
return WrapperPaths(
guard_path=os.path.join(agent_directory, GUARD_WRAPPER),
audit_path=os.path.join(agent_directory, AUDIT_WRAPPER),
)
def _quote_posix(value: str) -> str:
return "'" + value.replace("'", "'\"'\"'") + "'"
def _quote_windows(value: str) -> str:
return '"' + value.replace('"', '\\"') + '"'
def _quote_batch(value: str) -> str:
return '"' + value.replace("%", "%%") + '"'
def build_command(wrapper_path: str) -> str:
quote = _quote_windows if os.name == "nt" else _quote_posix
return quote(wrapper_path)
def build_wrapper(runtime_path: str) -> str:
if os.name == "nt":
return (
"@echo off\r\n"
f"{_quote_batch(sys.executable)} {_quote_batch(runtime_path)}\r\n"
"exit /b %errorlevel%\r\n"
)
return (
f"#!/bin/sh\nexec {_quote_posix(sys.executable)} {_quote_posix(runtime_path)}\n"
)
def build_handler(wrapper_path: str) -> JsonObject:
return {
"type": "command",
"command": build_command(wrapper_path),
"timeout": HOOK_TIMEOUT_MILLISECONDS,
}
def _parse_windows_argument(command: str) -> Optional[str]:
if len(command) < 2 or command[0] != '"':
return None
value = ""
index = 1
while index < len(command):
if command[index] == "\\" and index + 1 < len(command):
if command[index + 1] == '"':
value += '"'
index += 2
continue
if command[index] == '"':
return value if index == len(command) - 1 and value else None
value += command[index]
index += 1
return None
def _parse_posix_argument(command: str) -> Optional[str]:
if len(command) < 2 or command[0] != "'":
return None
value = ""
index = 1
apostrophe = "'\"'\"'"
while index < len(command):
if command.startswith(apostrophe, index):
value += "'"
index += len(apostrophe)
continue
if command[index] == "'":
return value if index == len(command) - 1 and value else None
value += command[index]
index += 1
return None
def _parse_wrapper_command(command: str) -> Optional[str]:
parser = _parse_windows_argument if os.name == "nt" else _parse_posix_argument
return parser(command)
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, wrapper_name: str) -> Optional[str]:
if (
handler.get("type") != "command"
or handler.get("timeout") != HOOK_TIMEOUT_MILLISECONDS
or not isinstance(handler.get("command"), str)
or "args" in handler
):
return None
wrapper_path = _parse_wrapper_command(str(handler["command"]))
if wrapper_path is None:
return None
if os.path.normcase(os.path.basename(wrapper_path)) != os.path.normcase(
wrapper_name
):
return None
agent_directory = os.path.dirname(wrapper_path)
if not _same_path(os.path.dirname(agent_directory), elydora_dir()):
return None
agent_id = os.path.basename(agent_directory)
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"Auggie settings handler hooks.{event}[{group_index}].hooks[{handler_index}]"
)
if not isinstance(value, dict):
raise ValueError(f"{label} must be an object")
handler = dict(value)
if handler.get("type") != "command":
raise ValueError(f'{label} type must be "command"')
if not isinstance(handler.get("command"), str) or not handler["command"]:
raise ValueError(f"{label} requires a non-empty command")
arguments = handler.get("args")
if "args" in handler and (
not isinstance(arguments, list)
or not all(isinstance(argument, str) for argument in arguments)
):
raise ValueError(f"{label} args must be an array of strings")
timeout = handler.get("timeout")
if "timeout" in handler 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 finite number")
return handler
def _validate_metadata(value: Any, label: str) -> None:
if not isinstance(value, dict):
raise ValueError(f"{label} metadata must be an object")
for key in (
"includeConversationData",
"includeMCPMetadata",
"includeUserContext",
):
if key in value and not isinstance(value[key], bool):
raise ValueError(f"{label} metadata.{key} must be a boolean")
def _validate_group(value: Any, event: str, group_index: int) -> JsonObject:
label = f"Auggie settings group hooks.{event}[{group_index}]"
if not isinstance(value, dict):
raise ValueError(f"{label} must be an object")
group = dict(value)
if event in _SESSION_EVENTS and "matcher" in group:
raise ValueError(f"{label} matcher is only supported for tool events")
matcher = group.get("matcher")
if "matcher" in group:
if not isinstance(matcher, str):
raise ValueError(f"{label} matcher must be a string")
try:
re.compile(matcher)
except re.error as error:
raise ValueError(
f"{label} matcher must be a valid regular expression: {error}"
) from error
if "metadata" in group:
_validate_metadata(group["metadata"], label)
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) -> AugmentHooks:
if "hooks" not in root:
return {}
raw_hooks = root["hooks"]
if not isinstance(raw_hooks, dict):
raise ValueError('Auggie settings field "hooks" must be an object')
hooks: AugmentHooks = {}
for event, groups in raw_hooks.items():
if event not in _TOOL_EVENTS and event not in _SESSION_EVENTS:
raise ValueError(
f'Auggie settings field "hooks.{event}" uses an unsupported event'
)
if not isinstance(groups, list):
raise ValueError(f'Auggie settings 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 _remove_managed(
groups: List[JsonObject],
wrapper_name: str,
agent_id: str = "",
) -> Tuple[List[JsonObject], bool]:
result: List[JsonObject] = []
changed = False
for group in groups:
kept: List[JsonObject] = []
group_changed = False
for handler in group["hooks"]:
managed_id = _managed_agent_id(handler, wrapper_name)
remove = managed_id is not None and (
not agent_id or _same_agent_id(managed_id, agent_id)
)
if remove:
changed = True
group_changed = True
else:
kept.append(handler)
if not group_changed:
result.append(group)
elif kept:
result.append({**group, "hooks": kept})
return result, changed
def remove_managed_hooks(
hooks: AugmentHooks, agent_id: str = ""
) -> Tuple[AugmentHooks, bool]:
result = dict(hooks)
changed = False
for event, wrapper_name in (
("PreToolUse", GUARD_WRAPPER),
("PostToolUse", AUDIT_WRAPPER),
):
groups, event_changed = _remove_managed(
result.get(event, []), wrapper_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], wrapper_name: str) -> Set[str]:
result: Set[str] = set()
for group in groups:
for handler in group["hooks"]:
agent_id = _managed_agent_id(handler, wrapper_name)
if agent_id:
result.add(agent_id)
return result
def runtime_contracts(hooks: AugmentHooks) -> List[RuntimeContract]:
guards = _managed_ids(hooks.get("PreToolUse", []), GUARD_WRAPPER)
audits = _managed_ids(hooks.get("PostToolUse", []), AUDIT_WRAPPER)
contracts: List[RuntimeContract] = []
for guard_id in sorted(guards, key=os.path.normcase):
audit_id = next(
(candidate for candidate in audits if _same_agent_id(candidate, guard_id)),
None,
)
if audit_id is None:
continue
agent_directory = os.path.join(elydora_dir(), guard_id)
contracts.append(
RuntimeContract(
agent_id=guard_id,
guard_path=os.path.join(agent_directory, GUARD_SCRIPT),
audit_path=os.path.join(agent_directory, AUDIT_SCRIPT),
guard_wrapper_path=os.path.join(agent_directory, GUARD_WRAPPER),
audit_wrapper_path=os.path.join(agent_directory, AUDIT_WRAPPER),
)
)
return contracts
+5
View File
@@ -6,6 +6,11 @@ from typing import Dict
SUPPORTED_AGENTS: Dict[str, Dict[str, str]] = {
"augment": {
"name": "Augment Code CLI",
"hook_event": "PreToolUse/PostToolUse",
"config_path": "~/.augment/settings.json",
},
"claudecode": {
"name": "Claude Code",
"hook_event": "PostToolUse",
+498
View File
@@ -0,0 +1,498 @@
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 augment
from elydora.plugins import augment_contract as contract
from elydora.plugins.base import InstallConfig
from elydora.plugins.registry import SUPPORTED_AGENTS
AGENT_ID = "agent-1"
MISSING = object()
@dataclass(frozen=True)
class AugmentFixture:
plugin: augment.AugmentPlugin
config: InstallConfig
home_dir: Path
workspace_dir: Path
agent_dir: Path
config_path: Path
guard_path: Path
hook_path: Path
guard_wrapper_path: Path
audit_wrapper_path: Path
runtime_config_path: Path
private_key_path: Path
def prepare_fixture(
monkeypatch: pytest.MonkeyPatch,
tmp_path: Path,
*,
existing_settings: object = MISSING,
create_guard: bool = True,
) -> AugmentFixture:
home_dir = tmp_path / "home with spaces and 'quote"
workspace_dir = home_dir / "workspace"
agent_dir = home_dir / ".elydora" / AGENT_ID
config_path = home_dir / ".augment" / "settings.json"
guard_path = agent_dir / contract.GUARD_SCRIPT
hook_path = agent_dir / contract.AUDIT_SCRIPT
guard_wrapper_path = agent_dir / contract.GUARD_WRAPPER
audit_wrapper_path = agent_dir / contract.AUDIT_WRAPPER
workspace_dir.mkdir(parents=True)
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_settings is not MISSING:
config_path.parent.mkdir(parents=True, exist_ok=True)
raw = (
str(existing_settings)
if isinstance(existing_settings, str)
else json.dumps(existing_settings, indent=2)
)
config_path.write_text(raw, encoding="utf-8")
monkeypatch.setattr(contract, "home_dir", lambda: str(home_dir))
config: InstallConfig = {
"agent_id": AGENT_ID,
"agent_name": "augment",
"org_id": "org-1",
"private_key": "test-key",
"kid": "kid-1",
"base_url": "https://api.elydora.test",
"guard_script_path": str(guard_path),
}
return AugmentFixture(
plugin=augment.AugmentPlugin(),
config=config,
home_dir=home_dir,
workspace_dir=workspace_dir,
agent_dir=agent_dir,
config_path=config_path,
guard_path=guard_path,
hook_path=hook_path,
guard_wrapper_path=guard_wrapper_path,
audit_wrapper_path=audit_wrapper_path,
runtime_config_path=agent_dir / "config.json",
private_key_path=agent_dir / "private.key",
)
def managed_handler(
settings: dict[str, Any], event: str, wrapper_path: Path
) -> dict[str, Any]:
command = contract.build_command(str(wrapper_path))
for group in settings["hooks"][event]:
for handler in group["hooks"]:
if handler.get("command") == 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_augment_is_registered_in_sdk_and_cli() -> None:
assert SUPPORTED_AGENTS["augment"] == {
"name": "Augment Code CLI",
"hook_event": "PreToolUse/PostToolUse",
"config_path": "~/.augment/settings.json",
}
assert cli.PLUGIN_MAP["augment"] is augment.AugmentPlugin
def test_install_preserves_user_settings_and_is_idempotent(
monkeypatch: pytest.MonkeyPatch,
tmp_path: Path,
capsys: pytest.CaptureFixture[str],
) -> None:
existing = {
"telemetryEnabled": False,
"hooks": {
"SessionStart": [
{
"hooks": [
{
"type": "command",
"command": "existing-command",
"args": ["one"],
"timeout": 5_000,
}
],
"metadata": {"includeUserContext": True},
"label": "keep group metadata",
}
],
"PreToolUse": [
{
"matcher": "launch-process",
"hooks": [
{
"type": "command",
"command": "user-command",
}
],
}
],
},
}
fixture = prepare_fixture(monkeypatch, tmp_path, existing_settings=existing)
fixture.plugin.install(fixture.config)
fixture.plugin.install(fixture.config)
assert "user-level PreToolUse and PostToolUse" in capsys.readouterr().out
settings = json.loads(fixture.config_path.read_text(encoding="utf-8"))
assert settings["telemetryEnabled"] is False
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 settings["hooks"]["PreToolUse"][1]["matcher"] == ".*"
for event, wrapper_path in (
("PreToolUse", fixture.guard_wrapper_path),
("PostToolUse", fixture.audit_wrapper_path),
):
handler = managed_handler(settings, event, wrapper_path)
assert set(handler) == {"type", "command", "timeout"}
assert handler["type"] == "command"
assert handler["timeout"] == 10_000
guard_wrapper = fixture.guard_wrapper_path.read_text(encoding="utf-8")
audit_wrapper = fixture.audit_wrapper_path.read_text(encoding="utf-8")
assert fixture.guard_path.name in guard_wrapper
assert fixture.hook_path.name in audit_wrapper
if os.name == "nt":
assert guard_wrapper.startswith("@echo off\n")
assert "exit /b %errorlevel%" in guard_wrapper
else:
assert fixture.guard_wrapper_path.stat().st_mode & 0o111
assert guard_wrapper.startswith("#!/bin/sh\nexec ")
assert (
json.loads(fixture.runtime_config_path.read_text(encoding="utf-8"))[
"agent_name"
]
== "augment"
)
assert fixture.private_key_path.read_text(encoding="utf-8") == "test-key"
assert (fixture.workspace_dir / ".augment" / "settings.json").exists() is False
assert (
fixture.workspace_dir / ".augment" / "settings.local.json"
).exists() is False
def test_wrappers_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(
{
"hook_event_name": "PreToolUse",
"conversation_id": "conversation-1",
"workspace_roots": [str(fixture.workspace_dir)],
"tool_name": "launch-process",
"tool_input": {"command": "echo test"},
"is_mcp_tool": False,
},
separators=(",", ":"),
)
guard = managed_handler(settings, "PreToolUse", fixture.guard_wrapper_path)
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(
{
"hook_event_name": "PostToolUse",
"conversation_id": "conversation-1",
"workspace_roots": [str(fixture.workspace_dir)],
"tool_name": "launch-process",
"tool_input": {"command": "echo test"},
"tool_output": "test",
"is_mcp_tool": False,
},
separators=(",", ":"),
)
audit = managed_handler(settings, "PostToolUse", fixture.audit_wrapper_path)
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_core_runtimes_and_wrappers(
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
fixture.guard_wrapper_path.unlink()
status = fixture.plugin.status()
assert status["installed"] is False
assert "wrappers missing" in status["details"]
fixture.plugin.install(fixture.config)
fixture.hook_path.unlink()
assert fixture.plugin.status()["installed"] is False
fixture.plugin.install(fixture.config)
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
def test_uninstall_removes_exact_ownership_and_preserves_user_groups(
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
) -> None:
fixture = prepare_fixture(
monkeypatch,
tmp_path,
existing_settings={"owner": "user", "hooks": {"Notification": []}},
)
fixture.plugin.install(fixture.config)
settings = json.loads(fixture.config_path.read_text(encoding="utf-8"))
settings["hooks"]["PreToolUse"].insert(
0, {"hooks": [], "label": "keep empty group"}
)
settings["hooks"]["PreToolUse"][1]["hooks"].append(
{
"type": "command",
"command": "user-command",
}
)
settings["hooks"]["PreToolUse"].append(
{
"hooks": [
{
"type": "command",
"command": contract.build_command(
str(fixture.guard_wrapper_path) + ".backup"
),
"timeout": 10_000,
}
]
}
)
settings["hooks"]["PreToolUse"].append(
{
"hooks": [
{
"type": "command",
"command": contract.build_command(
str(
fixture.agent_dir.parent
/ "agent-10"
/ contract.GUARD_WRAPPER
)
),
"timeout": 10_000,
}
]
}
)
fixture.config_path.write_text(json.dumps(settings, indent=2), 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"]) == 4
assert remaining["hooks"]["PreToolUse"][0] == {
"hooks": [],
"label": "keep empty group",
}
assert remaining["hooks"]["PreToolUse"][1]["hooks"] == [
{
"type": "command",
"command": "user-command",
}
]
raw = fixture.config_path.read_text(encoding="utf-8")
assert "augment-guard" in raw and "backup" in raw
assert "agent-10" in raw
assert "PostToolUse" not in remaining["hooks"]
def test_install_replaces_stale_handlers_and_preserves_empty_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"].insert(
0, {"hooks": [], "label": "keep empty group"}
)
for event, wrapper_name in (
("PreToolUse", contract.GUARD_WRAPPER),
("PostToolUse", contract.AUDIT_WRAPPER),
):
settings["hooks"][event].append(
{
"hooks": [
{
"type": "command",
"command": contract.build_command(
str(fixture.agent_dir.parent / "agent-old" / wrapper_name)
),
"timeout": 10_000,
}
]
}
)
fixture.config_path.write_text(json.dumps(settings, indent=2), encoding="utf-8")
fixture.plugin.install(fixture.config)
current = json.loads(fixture.config_path.read_text(encoding="utf-8"))
assert "agent-old" not in json.dumps(current)
assert current["hooks"]["PreToolUse"][0] == {
"hooks": [],
"label": "keep empty group",
}
assert len(current["hooks"]["PreToolUse"]) == 2
assert len(current["hooks"]["PostToolUse"]) == 1
def test_uninstall_removes_settings_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
@pytest.mark.parametrize(
"existing",
[
"{ malformed",
"null",
"[]",
{"hooks": None},
{"hooks": {"UnknownEvent": []}},
{"hooks": {"PreToolUse": None}},
{"hooks": {"PreToolUse": [None]}},
{"hooks": {"SessionStart": [{"matcher": ".*", "hooks": []}]}},
{"hooks": {"PreToolUse": [{"matcher": "[", "hooks": []}]}},
{"hooks": {"PreToolUse": [{"hooks": None}]}},
{"hooks": {"PreToolUse": [{"hooks": [{"type": "http", "command": "x"}]}]}},
{
"hooks": {
"PreToolUse": [
{"hooks": [{"type": "command", "command": "", "args": []}]}
]
}
},
{
"hooks": {
"PreToolUse": [
{"hooks": [{"type": "command", "command": "x", "args": [1]}]}
]
}
},
{
"hooks": {
"PreToolUse": [
{"hooks": [{"type": "command", "command": "x", "timeout": 0}]}
]
}
},
{
"hooks": {
"PreToolUse": [{"hooks": [], "metadata": {"includeUserContext": "yes"}}]
}
},
],
)
def test_install_rejects_malformed_settings_before_writes(
monkeypatch: pytest.MonkeyPatch,
tmp_path: Path,
existing: object,
) -> None:
fixture = prepare_fixture(monkeypatch, tmp_path, existing_settings=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
assert fixture.private_key_path.exists() is False
assert fixture.guard_wrapper_path.exists() is False
assert fixture.audit_wrapper_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
assert fixture.guard_wrapper_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())