diff --git a/integrations/catalog.json b/integrations/catalog.json index 1158194..b69cce1 100644 --- a/integrations/catalog.json +++ b/integrations/catalog.json @@ -30,7 +30,7 @@ "mechanism": "exit_code_2", "failure_mode": "fail_open" }, - "adapters": { "node": true, "python": false, "go": false }, + "adapters": { "node": true, "python": true, "go": false }, "delivery_state": "partial", "source_url": "https://docs.augmentcode.com/cli/hooks" }, diff --git a/sdks/python/README.md b/sdks/python/README.md index b1a76e5..3014f20 100644 --- a/sdks/python/README.md +++ b/sdks/python/README.md @@ -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` | diff --git a/sdks/python/elydora/cli.py b/sdks/python/elydora/cli.py index de87600..98840e2 100644 --- a/sdks/python/elydora/cli.py +++ b/sdks/python/elydora/cli.py @@ -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, diff --git a/sdks/python/elydora/plugins/augment.py b/sdks/python/elydora/plugins/augment.py new file mode 100644 index 0000000..92e5628 --- /dev/null +++ b/sdks/python/elydora/plugins/augment.py @@ -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, + ) diff --git a/sdks/python/elydora/plugins/augment_contract.py b/sdks/python/elydora/plugins/augment_contract.py new file mode 100644 index 0000000..403cd78 --- /dev/null +++ b/sdks/python/elydora/plugins/augment_contract.py @@ -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 diff --git a/sdks/python/elydora/plugins/registry.py b/sdks/python/elydora/plugins/registry.py index 4627b1e..5024aba 100644 --- a/sdks/python/elydora/plugins/registry.py +++ b/sdks/python/elydora/plugins/registry.py @@ -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", diff --git a/sdks/python/tests/test_augment_plugin.py b/sdks/python/tests/test_augment_plugin.py new file mode 100644 index 0000000..8ad7a97 --- /dev/null +++ b/sdks/python/tests/test_augment_plugin.py @@ -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())