feat: add native Auggie hook integration
This commit is contained in:
1 parent
bdcbc36002
commit
453804dff9
7 files changed
+1099
No files matched your search
@@ -24,6 +24,7 @@ This repository owns the published `elydora` package, its CLI, synchronous and a
|
|||||||
- Model stable, legacy, and early-access hook generations as explicit contracts. Keep their activation requirements visible in CLI output and README guidance.
|
- Model stable, legacy, and early-access hook generations as explicit contracts. Keep their activation requirements visible in CLI output and README guidance.
|
||||||
- Select Kimi Code and legacy kimi-cli config targets from existing runtime evidence; a fresh installation targets current Kimi Code and keeps migration markers isolated to active runtimes.
|
- Select Kimi Code and legacy kimi-cli config targets from existing runtime evidence; a fresh installation targets current Kimi Code and keeps migration markers isolated to active runtimes.
|
||||||
- Keep Grok Build writes inside its native global `$GROK_HOME/hooks/*.json` contract. Treat Claude Code and Cursor compatibility files plus project `.grok/hooks` as read-only integration sources.
|
- Keep Grok Build writes inside its native global `$GROK_HOME/hooks/*.json` contract. Treat Claude Code and Cursor compatibility files plus project `.grok/hooks` as read-only integration sources.
|
||||||
|
- Write Auggie hooks only to `~/.augment/settings.json`; keep system and workspace settings read-only. Generate `.cmd` wrappers on Windows and `.sh` wrappers on Unix because Auggie dispatches supported script paths, and express hook timeouts in milliseconds.
|
||||||
|
|
||||||
## Code Quality
|
## Code Quality
|
||||||
|
|
||||||
|
|||||||
@@ -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.
|
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`.
|
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
|
||||||
@@ -94,6 +96,7 @@ Kiro CLI installation covers both runtime contracts. Kiro CLI v2 uses the genera
|
|||||||
|
|
||||||
| Agent | Key |
|
| Agent | Key |
|
||||||
|-------|-----|
|
|-------|-----|
|
||||||
|
| Augment Code CLI | `augment` |
|
||||||
| Claude Code | `claudecode` |
|
| Claude Code | `claudecode` |
|
||||||
| OpenAI Codex | `codex` |
|
| OpenAI Codex | `codex` |
|
||||||
| Kimi Code | `kimi` |
|
| Kimi Code | `kimi` |
|
||||||
|
|||||||
@@ -18,6 +18,7 @@ from .crypto import get_public_key_base64url
|
|||||||
from .plugins.base import AgentPlugin, InstallConfig
|
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.augment import AugmentPlugin
|
||||||
from .plugins.claudecode import ClaudeCodePlugin
|
from .plugins.claudecode import ClaudeCodePlugin
|
||||||
from .plugins.codex import CodexPlugin
|
from .plugins.codex import CodexPlugin
|
||||||
from .plugins.copilot import CopilotPlugin
|
from .plugins.copilot import CopilotPlugin
|
||||||
@@ -32,6 +33,7 @@ from .plugins.opencode import OpenCodePlugin
|
|||||||
|
|
||||||
|
|
||||||
PLUGIN_MAP: Dict[str, Type[AgentPlugin]] = {
|
PLUGIN_MAP: Dict[str, Type[AgentPlugin]] = {
|
||||||
|
"augment": AugmentPlugin,
|
||||||
"claudecode": ClaudeCodePlugin,
|
"claudecode": ClaudeCodePlugin,
|
||||||
"codex": CodexPlugin,
|
"codex": CodexPlugin,
|
||||||
"copilot": CopilotPlugin,
|
"copilot": CopilotPlugin,
|
||||||
|
|||||||
@@ -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
|
||||||
@@ -6,6 +6,11 @@ from typing import Dict
|
|||||||
|
|
||||||
|
|
||||||
SUPPORTED_AGENTS: Dict[str, Dict[str, str]] = {
|
SUPPORTED_AGENTS: Dict[str, Dict[str, str]] = {
|
||||||
|
"augment": {
|
||||||
|
"name": "Augment Code CLI",
|
||||||
|
"hook_event": "PreToolUse/PostToolUse",
|
||||||
|
"config_path": "~/.augment/settings.json",
|
||||||
|
},
|
||||||
"claudecode": {
|
"claudecode": {
|
||||||
"name": "Claude Code",
|
"name": "Claude Code",
|
||||||
"hook_event": "PostToolUse",
|
"hook_event": "PostToolUse",
|
||||||
|
|||||||
@@ -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())
|
||||||
Reference in new issue
Block a user