From dccec10fea43c3aacc8cf343807a3eeccefd9803 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E9=9B=B7=E7=94=B5=E8=8A=BD=E8=A1=A3?= Date: Mon, 20 Jul 2026 09:57:11 -0400 Subject: [PATCH] fix(python): mirror transactional Auggie hooks --- AGENTS.md | 2 +- sdks/python/README.md | 2 +- sdks/python/elydora/plugins/augment.py | 246 ++------- .../elydora/plugins/augment_contract.py | 42 +- .../elydora/plugins/augment_installation.py | 203 ++++++++ sdks/python/elydora/plugins/augment_io.py | 266 ++++++++++ sdks/python/tests/augment_support.py | 201 ++++++++ sdks/python/tests/test_augment_official.py | 62 +++ sdks/python/tests/test_augment_plugin.py | 485 ++++++------------ sdks/python/tests/test_augment_runtime.py | 194 +++++++ sdks/python/tests/test_augment_transaction.py | 259 ++++++++++ 11 files changed, 1452 insertions(+), 510 deletions(-) create mode 100644 sdks/python/elydora/plugins/augment_installation.py create mode 100644 sdks/python/elydora/plugins/augment_io.py create mode 100644 sdks/python/tests/augment_support.py create mode 100644 sdks/python/tests/test_augment_official.py create mode 100644 sdks/python/tests/test_augment_runtime.py create mode 100644 sdks/python/tests/test_augment_transaction.py diff --git a/AGENTS.md b/AGENTS.md index 3e024c0..f9619e1 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -29,7 +29,7 @@ This repository publishes the open-source server, console, integration catalog, - Resolve Claude Code user hooks through `$CLAUDE_CONFIG_DIR/settings.json` with `~/.claude/settings.json` when the variable is absent. Match Claude Code's path resolution exactly: relative and empty values resolve from the current working directory, and literal tildes remain path segments. Keep project, local, managed, plugin, skill, and agent hook sources read-only. Validate the complete shipped handler schema, including command exec form, async rewake metadata, HTTP, MCP tool, prompt, and agent handlers. Register exact matchless `PreToolUse`, `PostToolUse`, and `PostToolUseFailure` groups with ten-second exec-form command handlers. Preserve the complete native snake_case payload and propagate freeze and revocation through exit code `2`. Commit each SDK's Claude Code user settings, generated runtimes, runtime config, and private key through one rollback-capable transaction. Reject installation while the user source sets `disableAllHooks`, then require `/hooks` and `claude doctor` verification for higher-scope policy effects. - Resolve Gemini CLI user hooks through `$GEMINI_CLI_HOME/.gemini/settings.json` with `~/.gemini/settings.json` when the variable is absent or empty. Preserve relative and literal-tilde home values as Gemini CLI does. Keep workspace, system defaults, system overrides, and extension hooks read-only. Preserve JSON comments while rejecting trailing commas and duplicate keys. Register exact matchless `BeforeTool` and `AfterTool` groups named `elydora-guard` and `elydora-audit` with ten-second command handlers. Preserve the complete native snake_case payload, emit valid JSON on exit code `0`, and propagate freeze and revocation through exit code `2`. Use encoded PowerShell commands on Windows. Commit each SDK's Gemini CLI user settings, generated runtimes, runtime config, and private key through one rollback-capable transaction. Respect `hooksConfig.enabled` and `hooksConfig.disabled`, then require `/hooks list` verification for higher-scope policy effects. - Resolve Auggie user hooks through `~/.augment/settings.json`; keep system, workspace, local workspace, and alternate `--augment-cache-dir` settings read-only. Validate the shipped Auggie 0.33 hook schema: `PreToolUse`, `PostToolUse`, `Stop`, `SessionStart`, `SessionEnd`, `Notification`, and `PromptSubmit`; regex matchers belong to tool events, command handlers use string-array arguments and positive millisecond timeouts, and metadata flags are booleans. Generate `.cmd` wrappers on Windows and `.sh` wrappers on Unix, preserve the complete native snake_case payload, and propagate freeze and revocation through exit code `2`. -- The Node SDK commits Auggie settings, both wrappers, generated runtimes, runtime config, and private key through one rollback-capable transaction. Node status requires exact runtime identity, canonical private keys, physical files, and exact wrapper sources, followed by effective configuration verification with `auggie tools list`. Python and Go validate matcher syntax with their native regex engines and keep status and uninstall available for recovery. +- The Node and Python SDKs commit Auggie settings, both wrappers, generated runtimes, runtime config, and private key through one rollback-capable transaction. Their status checks require exact runtime identity, canonical private keys, physical files, and exact wrapper sources, followed by effective configuration verification with `auggie tools list`. Go validates matcher syntax with its native regex engine and keeps status and uninstall available for recovery. - Write Cline file hooks to `$CLINE_DIR/hooks` with `~/.cline/hooks` as the default; keep Documents and workspace hook roots read-only. Use supported event filenames, preserve official hook input byte-for-byte, use `process.execPath` in Node and Go wrappers, and resolve Python runtimes from absolute shebangs. Translate guard exit code `2` into JSON stdout with `cancel: true`. Treat hook errors, timeouts, and invalid control JSON as fail-open. - Resolve Codex user hooks through `$CODEX_HOME/hooks.json` with `~/.codex/hooks.json` as the default, matching Codex's existing-directory canonicalization rule. Model user JSON, inline TOML layers, project files, managed `requirements.toml`, and plugin-bundled hooks as additive stable sources. SDK adapters write only the user JSON source, register exact `PreToolUse` and `PostToolUse` match-all command groups, preserve the complete native payload, propagate freeze and revocation through exit code `2`, keep guard lookup and audit delivery fail-open with observable errors, commit user hooks and all four runtime artifacts through one rollback-capable transaction, and require `/hooks` trust review for both definition hashes. - Write GitHub Copilot CLI hooks only to `$COPILOT_HOME/hooks/elydora-audit.json` with `~/.copilot/hooks/elydora-audit.json` as the default. Migrate exact Elydora-owned entries from project `.github/hooks/hooks.json`, preserve native camelCase input, require `disableAllHooks` to be false, and commit runtime plus hook documents in one rollback-capable transaction. diff --git a/sdks/python/README.md b/sdks/python/README.md index 0ebdc66..c93af73 100644 --- a/sdks/python/README.md +++ b/sdks/python/README.md @@ -92,7 +92,7 @@ Kimi installation writes exact `PreToolUse`, `PostToolUse`, and `PostToolUseFail Grok Build installation writes native global `PreToolUse`, `PostToolUse`, and `PostToolUseFailure` hooks to `$GROK_HOME/hooks/elydora-audit.json` (default `~/.grok/hooks/elydora-audit.json`). The hook file, generated runtimes, runtime config, and private key commit as one rollback-capable transaction. Managed commands preserve Grok's native camelCase JSON, return `{"decision":"deny","reason":"..."}` with exit code `2` for frozen or revoked agents, and use encoded PowerShell on Windows. Project hooks follow Grok's `/hooks-trust` workflow; Elydora leaves project, plugin, `hooks-paths`, Claude Code, and Cursor compatibility sources unchanged. Run `grok inspect --json` to inspect the effective hook inventory. -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. +Auggie installation commits `~/.augment/settings.json`, both platform wrappers, generated runtimes, runtime config, and private key as one rollback-capable transaction. The managed `PreToolUse` guard propagates exit code `2`; `PostToolUse` preserves Auggie's complete native hook payload. System, workspace, local workspace, and alternate `--augment-cache-dir` settings remain unchanged. Run `auggie tools list` to validate the effective user configuration. Factory Droid installation follows its active user source per event: `~/.factory/hooks.json`, the legacy `~/.factory/hooks/hooks.json` fallback, or hooks nested in `~/.factory/settings.json`. Elydora preserves JSONC comments and formatting, leaves project and organization hooks unchanged, and runs both managed commands with absolute Python and script paths. Run `/hooks` after installation to review the changes. diff --git a/sdks/python/elydora/plugins/augment.py b/sdks/python/elydora/plugins/augment.py index b0ddf88..0a8e86c 100644 --- a/sdks/python/elydora/plugins/augment.py +++ b/sdks/python/elydora/plugins/augment.py @@ -2,37 +2,27 @@ from __future__ import annotations -import os import shutil import subprocess # nosec B404 -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, AugmentHooks, - JsonObject, - RuntimeContract, build_handler, - build_wrapper, - elydora_dir, - read_hooks, remove_managed_hooks, - resolve_config_path, + render_augment_document, runtime_contracts, - wrapper_paths, ) +from .augment_installation import ( + AugmentRuntimePaths, + commit_augment_installation, + commit_augment_uninstall, + preflight_augment_installation, + prepare_augment_installation, + prepare_augment_uninstall, +) +from .augment_io import augment_runtime_files_exist, read_augment_document from .base import AgentPlugin, InstallConfig, PluginStatus -from .hook_template import generate_hook_script _REGEX_VALIDATION_TIMEOUT_SECONDS = 10 @@ -105,196 +95,70 @@ def _validate_matchers(hooks: AugmentHooks) -> None: _validate_matcher(node_path, label, matcher) -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 +def _installed_hooks(hooks: AugmentHooks, paths: AugmentRuntimePaths) -> AugmentHooks: + cleaned, _ = remove_managed_hooks(hooks) + return { + **cleaned, + "PreToolUse": [ + *cleaned.get("PreToolUse", []), + { + "matcher": ".*", + "hooks": [build_handler(paths.guard_wrapper_path)], + }, + ], + "PostToolUse": [ + *cleaned.get("PostToolUse", []), + { + "matcher": ".*", + "hooks": [build_handler(paths.audit_wrapper_path)], + }, + ], + } class AugmentPlugin(AgentPlugin): """Install Elydora into Auggie's global user settings.""" + manages_guard_runtime = True + + def preflight_install(self, config: InstallConfig) -> None: + document = read_augment_document() + _validate_matchers(document.hooks) + preflight_augment_installation(config, document) + 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() - _validate_matchers(source.hooks) - 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, - 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", + document = read_augment_document() + _validate_matchers(document.hooks) + paths = preflight_augment_installation(config, document) + rendered = render_augment_document( + document, _installed_hooks(document.hooks, paths) ) + changes = prepare_augment_installation(config, paths, rendered) + commit_augment_installation(changes) print("Auggie: user-level PreToolUse and PostToolUse hooks installed.") def uninstall(self, agent_id: str = "") -> None: - source = _read_config() - if not source.exists: + document = read_augment_document() + if not document.exists: return - hooks, changed = remove_managed_hooks(source.hooks, agent_id) + hooks, changed = remove_managed_hooks(document.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") + rendered = render_augment_document(document, hooks) + commit_augment_uninstall(prepare_augment_uninstall(rendered)) 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) + document = read_augment_document() + contracts = runtime_contracts(document.hooks) + configured = bool(contracts) + installed = configured and augment_runtime_files_exist(contracts) details = ( - f"Config: {source.config_path}" + f"Config: {document.config_path}" if installed - else ( - f"Configured at {source.config_path}; " - "runtime scripts or wrappers missing" - ) + else f"Configured at {document.config_path}; managed contract incomplete" ) return PluginStatus( installed=installed, agent=AGENT_KEY, - details=details, + details=details if configured else "Not installed", ) diff --git a/sdks/python/elydora/plugins/augment_contract.py b/sdks/python/elydora/plugins/augment_contract.py index cc6edfe..a40a621 100644 --- a/sdks/python/elydora/plugins/augment_contract.py +++ b/sdks/python/elydora/plugins/augment_contract.py @@ -3,11 +3,14 @@ from __future__ import annotations from dataclasses import dataclass +import json import math import os import sys from typing import Any, Dict, List, Optional, Set, Tuple +from ._strict_json import parse_json_object + AGENT_KEY = "augment" GUARD_SCRIPT = "guard.py" @@ -37,6 +40,14 @@ class AugmentDocument: config_path: str root: JsonObject hooks: AugmentHooks + raw: Optional[str] = None + + +@dataclass(frozen=True) +class RenderedAugmentDocument: + document: AugmentDocument + changed: bool + next_source: Optional[str] = None @dataclass(frozen=True) @@ -66,8 +77,7 @@ 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) +def wrapper_paths(agent_directory: str) -> WrapperPaths: return WrapperPaths( guard_path=os.path.join(agent_directory, GUARD_WRAPPER), audit_path=os.path.join(agent_directory, AUDIT_WRAPPER), @@ -273,6 +283,34 @@ def read_hooks(root: JsonObject) -> AugmentHooks: return hooks +def parse_augment_document(config_path: str, raw: str) -> AugmentDocument: + root = parse_json_object(raw, f"Auggie user settings at {config_path}") + return AugmentDocument(True, config_path, root, read_hooks(root), raw) + + +def create_augment_document(config_path: str) -> AugmentDocument: + return AugmentDocument(False, config_path, {}, {}) + + +def render_augment_document( + document: AugmentDocument, hooks: AugmentHooks +) -> RenderedAugmentDocument: + if hooks == document.hooks: + return RenderedAugmentDocument(document, False) + if not document.exists and not hooks: + return RenderedAugmentDocument(document, False) + root = dict(document.root) + if hooks: + root["hooks"] = hooks + else: + root.pop("hooks", None) + if not root: + return RenderedAugmentDocument(document, True) + next_source = json.dumps(root, indent=2, ensure_ascii=False) + "\n" + parse_augment_document(document.config_path, next_source) + return RenderedAugmentDocument(document, next_source != document.raw, next_source) + + def _remove_managed( groups: List[JsonObject], wrapper_name: str, diff --git a/sdks/python/elydora/plugins/augment_installation.py b/sdks/python/elydora/plugins/augment_installation.py new file mode 100644 index 0000000..86c6c10 --- /dev/null +++ b/sdks/python/elydora/plugins/augment_installation.py @@ -0,0 +1,203 @@ +"""Transactional Auggie hook and runtime installation.""" + +from __future__ import annotations + +from dataclasses import dataclass +import json +import os +from typing import List, Optional + +from elydora._runtime_paths import resolve_agent_directory + +from ._managed_files import MAX_CONFIG_BYTES, MAX_SECRET_BYTES, MAX_SOURCE_BYTES +from ._transaction import FileChange, file_change, write_changes +from .augment_contract import ( + AGENT_KEY, + AUDIT_SCRIPT, + GUARD_SCRIPT, + AugmentDocument, + RenderedAugmentDocument, + build_wrapper, + elydora_dir, + wrapper_paths, +) +from .augment_io import ( + rendered_change, + validate_api_origin, + validate_private_key, + validate_runtime_tree, +) +from .base import InstallConfig +from .guard_template import generate_guard_script +from .hook_template import generate_hook_script + + +@dataclass(frozen=True) +class AugmentRuntimePaths: + agent_id: str + agent_directory: str + guard_path: str + audit_path: str + guard_wrapper_path: str + audit_wrapper_path: str + + +def _same_path(left: str, right: str) -> bool: + return os.path.normcase(os.path.abspath(left)) == os.path.normcase( + os.path.abspath(right) + ) + + +def _runtime_config(config: InstallConfig, agent_id: str) -> dict: + value = { + "org_id": config.get("org_id", ""), + "agent_id": agent_id, + "kid": config.get("kid", ""), + "base_url": config.get("base_url", "https://api.elydora.com"), + "agent_name": AGENT_KEY, + } + token = config.get("token") + if token: + value["token"] = token + return value + + +def _json_source(value: dict) -> str: + return json.dumps(value, indent=2) + "\n" + + +def _agent_paths(config: InstallConfig) -> AugmentRuntimePaths: + agent_id = config.get("agent_id", "") + if not agent_id: + raise ValueError("agent_id is required") + agent_directory = resolve_agent_directory(elydora_dir(), agent_id) + guard_path = os.path.join(agent_directory, GUARD_SCRIPT) + audit_path = os.path.join(agent_directory, AUDIT_SCRIPT) + if not _same_path(config.get("guard_script_path", ""), guard_path): + raise ValueError( + f"Elydora guard runtime must use the managed agent directory: {guard_path}" + ) + wrappers = wrapper_paths(agent_directory) + return AugmentRuntimePaths( + agent_id, + agent_directory, + guard_path, + audit_path, + wrappers.guard_path, + wrappers.audit_path, + ) + + +def _validate_install_config(config: InstallConfig) -> None: + for field in ("org_id", "agent_id", "kid", "private_key", "base_url"): + value = config.get(field) + if not isinstance(value, str) or not value: + raise ValueError(f"{field} is required") + if config.get("agent_name") != AGENT_KEY: + raise ValueError( + f"Augment Code CLI installation requires agent_name {AGENT_KEY}" + ) + validate_private_key(config["private_key"]) + validate_api_origin(config["base_url"]) + if "token" in config: + token = config["token"] + if not isinstance(token, str) or not token: + raise ValueError("token must be a non-empty string when provided") + _agent_paths(config) + + +def preflight_augment_installation( + config: InstallConfig, document: AugmentDocument +) -> AugmentRuntimePaths: + if not document.config_path: + raise ValueError("Augment Code CLI installation requires a settings path") + _validate_install_config(config) + paths = _agent_paths(config) + validate_runtime_tree(paths.agent_directory, paths.agent_id) + return paths + + +def _runtime_change( + file_path: str, + label: str, + next_source: str, + mode: int, + maximum_bytes: int = MAX_SOURCE_BYTES, +) -> Optional[FileChange]: + return file_change(file_path, label, next_source, mode, maximum_bytes) + + +def _present(changes: List[Optional[FileChange]]) -> List[FileChange]: + return [change for change in changes if change is not None] + + +def prepare_augment_installation( + config: InstallConfig, + paths: AugmentRuntimePaths, + rendered: RenderedAugmentDocument, +) -> List[FileChange]: + if not rendered.changed and rendered.document.raw is None: + raise ValueError("Auggie hook installation did not produce a settings document") + validate_runtime_tree(paths.agent_directory, paths.agent_id) + guard_script = generate_guard_script(AGENT_KEY, paths.agent_id) + audit_script = generate_hook_script( + org_id=config.get("org_id", ""), + agent_id=paths.agent_id, + kid=config.get("kid", ""), + base_url=config.get("base_url", "https://api.elydora.com"), + native_payload=True, + agent_name=AGENT_KEY, + ) + changes = _present( + [ + _runtime_change( + paths.guard_path, "Elydora guard runtime", guard_script, 0o700 + ), + _runtime_change( + os.path.join(paths.agent_directory, "config.json"), + "Elydora runtime config", + _json_source(_runtime_config(config, paths.agent_id)), + 0o600, + MAX_CONFIG_BYTES, + ), + _runtime_change( + os.path.join(paths.agent_directory, "private.key"), + "Elydora private key", + config.get("private_key", ""), + 0o600, + MAX_SECRET_BYTES, + ), + _runtime_change( + paths.audit_path, "Elydora audit runtime", audit_script, 0o700 + ), + _runtime_change( + paths.guard_wrapper_path, + "Auggie guard wrapper", + build_wrapper(paths.guard_path), + 0o700, + ), + _runtime_change( + paths.audit_wrapper_path, + "Auggie audit wrapper", + build_wrapper(paths.audit_path), + 0o700, + ), + rendered_change(rendered), + ] + ) + validate_runtime_tree(paths.agent_directory, paths.agent_id) + return changes + + +def commit_augment_installation(changes: List[FileChange]) -> None: + write_changes(changes, "Install Augment Code CLI hooks") + + +def prepare_augment_uninstall( + rendered: RenderedAugmentDocument, +) -> List[FileChange]: + return _present([rendered_change(rendered)]) + + +def commit_augment_uninstall(changes: List[FileChange]) -> None: + write_changes(changes, "Uninstall Augment Code CLI hooks") diff --git a/sdks/python/elydora/plugins/augment_io.py b/sdks/python/elydora/plugins/augment_io.py new file mode 100644 index 0000000..706946d --- /dev/null +++ b/sdks/python/elydora/plugins/augment_io.py @@ -0,0 +1,266 @@ +"""Auggie settings discovery and managed runtime I/O.""" + +from __future__ import annotations + +import base64 +import os +from typing import Any, List, Optional +import urllib.parse + +from ._managed_files import ( + MAX_CONFIG_BYTES, + MAX_SECRET_BYTES, + physical_directory_exists, + physical_file_exists, + read_physical_file, +) +from ._strict_json import JsonObject, parse_json_object +from ._transaction import FileChange, source_change, write_changes +from .augment_contract import ( + AGENT_KEY, + AUDIT_SCRIPT, + AUDIT_WRAPPER, + GUARD_SCRIPT, + GUARD_WRAPPER, + AugmentDocument, + RenderedAugmentDocument, + RuntimeContract, + build_wrapper, + create_augment_document, + elydora_dir, + parse_augment_document, + resolve_config_path, +) + + +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 read_augment_document() -> AugmentDocument: + config_path = resolve_config_path() + physical_directory_exists( + os.path.dirname(config_path), "Auggie configuration directory" + ) + snapshot = read_physical_file(config_path, "Auggie user settings", MAX_CONFIG_BYTES) + if snapshot is None: + return create_augment_document(config_path) + return parse_augment_document(config_path, snapshot.contents) + + +def rendered_change( + rendered: RenderedAugmentDocument, +) -> Optional[FileChange]: + if not rendered.changed: + return None + return source_change( + rendered.document.config_path, + "Auggie user settings", + rendered.document.raw, + rendered.next_source, + 0o600, + MAX_CONFIG_BYTES, + ) + + +def write_augment_document(rendered: RenderedAugmentDocument) -> None: + change = rendered_change(rendered) + if change is not None: + write_changes([change], "Uninstall Augment Code CLI hooks") + + +def validate_api_origin(value: str, label: str = "base_url") -> None: + try: + parsed = urllib.parse.urlsplit(value) + hostname = parsed.hostname + parsed.port + except ValueError as error: + raise ValueError(f"{label} must be an absolute HTTP or HTTPS URL") from error + invalid_character = "\\" in value or any( + character.isspace() or ord(character) < 32 for character in value + ) + if ( + parsed.scheme not in ("http", "https") + or not parsed.netloc + or hostname is None + or invalid_character + ): + raise ValueError(f"{label} must be an absolute HTTP or HTTPS URL") + if ( + parsed.username is not None + or parsed.password is not None + or parsed.query + or parsed.fragment + ): + raise ValueError( + f"{label} must exclude credentials, query parameters, and fragments" + ) + + +def validate_private_key(value: str, label: str = "private_key") -> None: + try: + padded = value + "=" * ((4 - len(value) % 4) % 4) + seed = base64.b64decode( + padded.replace("-", "+").replace("_", "/"), validate=True + ) + canonical = base64.urlsafe_b64encode(seed).rstrip(b"=").decode("ascii") + except (ValueError, UnicodeEncodeError) as error: + raise ValueError( + f"{label} must be a canonical 32-byte base64url value" + ) from error + if len(seed) != 32 or canonical != value: + raise ValueError(f"{label} must be a canonical 32-byte base64url value") + + +def _require_non_empty_string(value: Any, field: str, config_path: str) -> str: + if not isinstance(value, str) or not value.strip(): + raise ValueError(f"Elydora runtime config {field} is invalid: {config_path}") + return value + + +def _validate_runtime_config( + config: JsonObject, expected_agent_id: str, config_path: str +) -> None: + supported = {"org_id", "agent_id", "kid", "base_url", "token", "agent_name"} + extra = next((key for key in config if key not in supported), None) + if extra is not None: + raise ValueError( + f'Elydora runtime config has unsupported field "{extra}": {config_path}' + ) + _require_non_empty_string(config.get("org_id"), "org_id", config_path) + _require_non_empty_string(config.get("kid"), "kid", config_path) + agent_id = _require_non_empty_string( + config.get("agent_id"), "agent_id", config_path + ) + if ( + not _same_agent_id(agent_id, expected_agent_id) + or config.get("agent_name") != AGENT_KEY + ): + raise ValueError( + f"Elydora runtime identity does not match Auggie hooks: {config_path}" + ) + if "token" in config: + _require_non_empty_string(config.get("token"), "token", config_path) + base_url = _require_non_empty_string( + config.get("base_url"), "base_url", config_path + ) + validate_api_origin(base_url, "Elydora runtime config base_url") + + +def _read_runtime_config(file_path: str) -> Optional[JsonObject]: + snapshot = read_physical_file(file_path, "Elydora runtime config", MAX_CONFIG_BYTES) + if snapshot is None: + return None + return parse_json_object( + snapshot.contents, f"Elydora runtime config at {file_path}" + ) + + +def validate_runtime_tree(agent_directory: str, agent_id: str) -> None: + root = elydora_dir() + if not physical_directory_exists(root, "Elydora runtime directory"): + return + if not physical_directory_exists( + agent_directory, "Elydora agent runtime directory" + ): + return + config_path = os.path.join(agent_directory, "config.json") + config = _read_runtime_config(config_path) + artifacts = ( + ("private.key", "Elydora private key"), + (GUARD_SCRIPT, "Elydora guard runtime"), + (AUDIT_SCRIPT, "Elydora audit runtime"), + (GUARD_WRAPPER, "Auggie guard wrapper"), + (AUDIT_WRAPPER, "Auggie audit wrapper"), + ("chain-state.json", "Elydora chain state"), + ("status-cache.json", "Elydora status cache"), + ("error.log", "Elydora error log"), + ) + artifact_exists = any( + physical_file_exists(os.path.join(agent_directory, name), label) + for name, label in artifacts + ) + if config is None: + if artifact_exists: + raise ValueError( + "Elydora runtime identity cannot be verified without config.json: " + f"{agent_directory}" + ) + return + configured_id = config.get("agent_id") + if ( + not isinstance(configured_id, str) + or not _same_agent_id(configured_id, agent_id) + or config.get("agent_name") != AGENT_KEY + ): + raise ValueError( + "Elydora runtime config identity does not match Auggie agent " + f"{agent_id}: {config_path}" + ) + + +def _runtime_contract_exists(contract: RuntimeContract) -> bool: + root = elydora_dir() + agent_directory = os.path.dirname(contract.guard_path) + if ( + not _same_path(os.path.dirname(agent_directory), root) + or not _same_path( + contract.audit_path, os.path.join(agent_directory, AUDIT_SCRIPT) + ) + or not _same_path( + contract.guard_wrapper_path, + os.path.join(agent_directory, GUARD_WRAPPER), + ) + or not _same_path( + contract.audit_wrapper_path, + os.path.join(agent_directory, AUDIT_WRAPPER), + ) + ): + return False + if not physical_directory_exists(root, "Elydora runtime directory"): + return False + if not physical_directory_exists( + agent_directory, "Elydora agent runtime directory" + ): + return False + config_path = os.path.join(agent_directory, "config.json") + key_path = os.path.join(agent_directory, "private.key") + config = _read_runtime_config(config_path) + key = read_physical_file(key_path, "Elydora private key", MAX_SECRET_BYTES) + guard = read_physical_file(contract.guard_path, "Elydora guard runtime") + audit = read_physical_file(contract.audit_path, "Elydora audit runtime") + guard_wrapper = read_physical_file( + contract.guard_wrapper_path, "Auggie guard wrapper" + ) + audit_wrapper = read_physical_file( + contract.audit_wrapper_path, "Auggie audit wrapper" + ) + if any( + item is None + for item in (config, key, guard, audit, guard_wrapper, audit_wrapper) + ): + return False + assert config is not None + assert key is not None + assert guard is not None + assert audit is not None + assert guard_wrapper is not None + assert audit_wrapper is not None + _validate_runtime_config(config, contract.agent_id, config_path) + validate_private_key(key.contents, "Elydora private key") + return ( + bool(guard.contents) + and bool(audit.contents) + and guard_wrapper.contents == build_wrapper(contract.guard_path) + and audit_wrapper.contents == build_wrapper(contract.audit_path) + ) + + +def augment_runtime_files_exist(contracts: List[RuntimeContract]) -> bool: + return any(_runtime_contract_exists(contract) for contract in contracts) diff --git a/sdks/python/tests/augment_support.py b/sdks/python/tests/augment_support.py new file mode 100644 index 0000000..57b4616 --- /dev/null +++ b/sdks/python/tests/augment_support.py @@ -0,0 +1,201 @@ +from __future__ import annotations + +from dataclasses import dataclass +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +import json +import os +from pathlib import Path +import subprocess # nosec B404 +from threading import Thread +from typing import Any, Mapping + +import pytest + +from elydora.plugins import augment_contract as contract +from elydora.plugins.augment import AugmentPlugin +from elydora.plugins.base import InstallConfig +from elydora.utils import base64url_encode + + +AGENT_ID = "agent-1" +MISSING = object() +VALID_PRIVATE_KEY = base64url_encode(bytes(range(32))) + + +@dataclass +class AugmentFixture: + root_dir: Path + home_dir: Path + project_dir: Path + config_path: Path + agent_dir: Path + guard_path: Path + hook_path: Path + guard_wrapper_path: Path + audit_wrapper_path: Path + runtime_config_path: Path + private_key_path: Path + plugin: AugmentPlugin + config: InstallConfig + + def install(self) -> None: + self.plugin.install(self.config) + + def settings(self) -> dict[str, Any]: + return json.loads(self.config_path.read_text(encoding="utf-8")) + + +class ElydoraApiHandler(BaseHTTPRequestHandler): + agent_status: Any = "active" + operation_status = 201 + requests: list[dict[str, Any]] = [] + + @classmethod + def reset(cls, status: Any, operation_status: int) -> None: + cls.agent_status = status + cls.operation_status = operation_status + cls.requests = [] + + def _respond(self, status: int, value: object) -> None: + body = json.dumps(value).encode() + self.send_response(status) + self.send_header("Content-Type", "application/json") + self.send_header("Content-Length", str(len(body))) + self.end_headers() + self.wfile.write(body) + + def do_GET(self) -> None: + type(self).requests.append( + { + "method": "GET", + "path": self.path, + "authorization": self.headers.get("Authorization"), + } + ) + self._respond(200, {"agent": {"status": type(self).agent_status}}) + + def do_POST(self) -> None: + length = int(self.headers.get("Content-Length", "0")) + raw = self.rfile.read(length) + type(self).requests.append( + { + "method": "POST", + "path": self.path, + "authorization": self.headers.get("Authorization"), + "raw": raw, + "json": json.loads(raw), + } + ) + self._respond( + type(self).operation_status, + {"operation": {"accepted": True}}, + ) + + def log_message(self, _format: str, *_args: object) -> None: + return + + +def start_api_server( + *, status: Any = "active", operation_status: int = 201 +) -> ThreadingHTTPServer: + ElydoraApiHandler.reset(status, operation_status) + server = ThreadingHTTPServer(("127.0.0.1", 0), ElydoraApiHandler) + Thread(target=server.serve_forever, daemon=True).start() + return server + + +def server_base_url(server: ThreadingHTTPServer) -> str: + return f"http://127.0.0.1:{server.server_port}" + + +def write_text(path: Path, source: str, mode: int = 0o600) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(source, encoding="utf-8") + os.chmod(path, mode) + + +def write_json(path: Path, value: object, mode: int = 0o600) -> None: + source = value if isinstance(value, str) else json.dumps(value, indent=2) + "\n" + write_text(path, source, mode) + + +def prepare_fixture( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, + *, + existing_settings: object = MISSING, + base_url: str = "http://127.0.0.1:9", +) -> AugmentFixture: + home_dir = tmp_path / "home with spaces and 'quote %AUGGIE%" + project_dir = tmp_path / "project with spaces" + project_dir.mkdir(parents=True) + monkeypatch.setenv("HOME", str(home_dir)) + monkeypatch.setenv("USERPROFILE", str(home_dir)) + monkeypatch.chdir(project_dir) + monkeypatch.setattr(contract, "home_dir", lambda: str(home_dir)) + config_path = home_dir / ".augment" / "settings.json" + agent_dir = home_dir / ".elydora" / AGENT_ID + guard_path = agent_dir / contract.GUARD_SCRIPT + hook_path = agent_dir / contract.AUDIT_SCRIPT + if existing_settings is not MISSING: + write_json(config_path, existing_settings) + return AugmentFixture( + root_dir=tmp_path, + home_dir=home_dir, + project_dir=project_dir, + config_path=config_path, + agent_dir=agent_dir, + guard_path=guard_path, + hook_path=hook_path, + guard_wrapper_path=agent_dir / contract.GUARD_WRAPPER, + audit_wrapper_path=agent_dir / contract.AUDIT_WRAPPER, + runtime_config_path=agent_dir / "config.json", + private_key_path=agent_dir / "private.key", + plugin=AugmentPlugin(), + config={ + "org_id": "org-1", + "agent_id": AGENT_ID, + "agent_name": "augment", + "private_key": VALID_PRIVATE_KEY, + "kid": "key-1", + "token": "token-1", + "base_url": base_url, + "guard_script_path": str(guard_path), + }, + ) + + +def managed_handler( + settings: Mapping[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 handler for {event!r} not found") + + +def run_handler( + handler: Mapping[str, Any], payload: bytes, fixture: AugmentFixture +) -> subprocess.CompletedProcess[bytes]: + return subprocess.run( # nosec B602 + str(handler["command"]), + shell=True, + input=payload, + capture_output=True, + check=False, + cwd=fixture.project_dir, + env={ + **os.environ, + "HOME": str(fixture.home_dir), + "USERPROFILE": str(fixture.home_dir), + }, + ) + + +def symlink_or_skip(target: Path, link: Path, *, directory: bool = False) -> None: + try: + link.symlink_to(target, target_is_directory=directory) + except OSError as error: + pytest.skip(f"Symbolic links unavailable: {error}") diff --git a/sdks/python/tests/test_augment_official.py b/sdks/python/tests/test_augment_official.py new file mode 100644 index 0000000..ebea62d --- /dev/null +++ b/sdks/python/tests/test_augment_official.py @@ -0,0 +1,62 @@ +from __future__ import annotations + +import os +from pathlib import Path +import shutil +import subprocess # nosec B404 + +import pytest + +from augment_support import prepare_fixture + + +AUGGIE_ENTRY = os.environ.get("ELYDORA_AUGGIE_ENTRY") + + +@pytest.mark.skipif( + AUGGIE_ENTRY is None, + reason="set ELYDORA_AUGGIE_ENTRY to the official Auggie entry file", +) +def test_official_auggie_accepts_installed_user_hook_contract( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + node = shutil.which("node") + assert node is not None + assert AUGGIE_ENTRY is not None + fixture = prepare_fixture(monkeypatch, tmp_path) + fixture.install() + environment = { + **os.environ, + "HOME": str(fixture.home_dir), + "USERPROFILE": str(fixture.home_dir), + } + version = subprocess.run( # nosec B603 + [node, AUGGIE_ENTRY, "--version"], + capture_output=True, + check=False, + cwd=fixture.project_dir, + env=environment, + text=True, + timeout=10, + ) + assert version.returncode == 0, version.stderr + assert "0.33.0" in version.stdout + + load = subprocess.run( # nosec B603 + [node, AUGGIE_ENTRY, "tools", "list"], + capture_output=True, + check=False, + cwd=fixture.project_dir, + env=environment, + text=True, + timeout=15, + ) + assert load.returncode == 0, load.stderr + output = f"{load.stdout}\n{load.stderr}".lower() + for error in ( + "invalid settings", + "settings validation", + "failed to parse", + "hook configuration error", + ): + assert error not in output diff --git a/sdks/python/tests/test_augment_plugin.py b/sdks/python/tests/test_augment_plugin.py index 2397828..df35831 100644 --- a/sdks/python/tests/test_augment_plugin.py +++ b/sdks/python/tests/test_augment_plugin.py @@ -1,141 +1,37 @@ from __future__ import annotations -from dataclasses import dataclass import json import os from pathlib import Path -import subprocess -from typing import Any +import subprocess # nosec B404 +import sys import pytest +from augment_support import ( + AGENT_ID, + VALID_PRIVATE_KEY, + managed_handler, + prepare_fixture, + write_text, +) 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: +def test_augment_is_registered_and_owns_its_complete_runtime() -> 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 + assert augment.AugmentPlugin.manages_guard_runtime is True -def test_install_preserves_user_settings_and_is_idempotent( +def test_install_preserves_official_settings_and_is_idempotent( monkeypatch: pytest.MonkeyPatch, tmp_path: Path, capsys: pytest.CaptureFixture[str], @@ -153,34 +49,42 @@ def test_install_preserves_user_settings_and_is_idempotent( "timeout": 5_000, } ], - "metadata": {"includeUserContext": True}, + "metadata": { + "includeConversationData": True, + "includeMCPMetadata": False, + "includeUserContext": True, + }, "label": "keep group metadata", } ], + "PromptSubmit": [ + {"hooks": [{"type": "command", "command": "prompt-hook"}]} + ], + "Notification": [ + {"hooks": [{"type": "command", "command": "notification-hook"}]} + ], "PreToolUse": [ { "matcher": "(?launch-process)", - "hooks": [ - { - "type": "command", - "command": "user-command", - } - ], + "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) + fixture.install() + first = fixture.config_path.read_text(encoding="utf-8") + fixture.install() + assert fixture.config_path.read_text(encoding="utf-8") == first assert "user-level PreToolUse and PostToolUse" in capsys.readouterr().out - settings = json.loads(fixture.config_path.read_text(encoding="utf-8")) + + settings = fixture.settings() assert settings["telemetryEnabled"] is False - assert settings["hooks"]["SessionStart"] == existing["hooks"]["SessionStart"] + for event in ("SessionStart", "PromptSubmit", "Notification"): + assert settings["hooks"][event] == existing["hooks"][event] 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), @@ -189,97 +93,67 @@ def test_install_preserves_user_settings_and_is_idempotent( assert set(handler) == {"type", "command", "timeout"} assert handler["type"] == "command" assert handler["timeout"] == 10_000 + + assert json.loads(fixture.runtime_config_path.read_text(encoding="utf-8")) == { + "org_id": "org-1", + "agent_id": AGENT_ID, + "kid": "key-1", + "base_url": "http://127.0.0.1:9", + "agent_name": "augment", + "token": "token-1", + } + assert fixture.private_key_path.read_text(encoding="utf-8") == VALID_PRIVATE_KEY + assert "NATIVE_PAYLOAD = True" in fixture.hook_path.read_text(encoding="utf-8") 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 + assert fixture.hook_path.name in fixture.audit_wrapper_path.read_text( + encoding="utf-8" + ) 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 + assert not (fixture.project_dir / ".augment" / "settings.json").exists() -def test_wrappers_block_and_forward_official_payload_byte_for_byte( +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) - 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", + fixture.install() + settings = fixture.settings() + settings["hooks"]["PreToolUse"].insert( + 0, {"hooks": [], "label": "keep empty group"} ) - 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 + 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.install() + current = fixture.settings() + 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_exact_ownership_and_preserves_user_groups( @@ -290,16 +164,10 @@ def test_uninstall_removes_exact_ownership_and_preserves_user_groups( 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", - } + fixture.install() + settings = fixture.settings() + settings["hooks"]["PreToolUse"][0]["hooks"].append( + {"type": "command", "command": "user-command"} ) settings["hooks"]["PreToolUse"].append( { @@ -334,19 +202,11 @@ def test_uninstall_removes_exact_ownership_and_preserves_user_groups( 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")) + remaining = fixture.settings() 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", - } + assert remaining["hooks"]["PreToolUse"][0]["hooks"] == [ + {"type": "command", "command": "user-command"} ] raw = fixture.config_path.read_text(encoding="utf-8") assert "augment-guard" in raw and "backup" in raw @@ -354,51 +214,81 @@ def test_uninstall_removes_exact_ownership_and_preserves_user_groups( 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.install() fixture.plugin.uninstall(AGENT_ID) - assert fixture.config_path.exists() is False + assert not fixture.config_path.exists() + + +def test_cli_install_status_and_uninstall_end_to_end( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + fixture = prepare_fixture(monkeypatch, tmp_path) + key_path = tmp_path / "install-private.key" + token_path = tmp_path / "install-token.txt" + write_text(key_path, VALID_PRIVATE_KEY + "\n") + write_text(token_path, "token-1\n") + environment = { + **os.environ, + "HOME": str(fixture.home_dir), + "USERPROFILE": str(fixture.home_dir), + } + root = Path(__file__).parents[1] + base = [sys.executable, "-m", "elydora.cli"] + install = subprocess.run( # nosec B603 + [ + *base, + "install", + "--agent", + "augment", + "--org_id", + "org-1", + "--agent_id", + AGENT_ID, + "--private_key_file", + str(key_path), + "--token_file", + str(token_path), + "--kid", + "key-1", + "--base_url", + "http://127.0.0.1:9", + ], + capture_output=True, + check=False, + cwd=root, + env=environment, + text=True, + ) + assert install.returncode == 0, install.stderr + assert "PreToolUse and PostToolUse hooks installed" in install.stdout + + status = subprocess.run( # nosec B603 + [*base, "status"], + capture_output=True, + check=False, + cwd=root, + env=environment, + text=True, + ) + assert status.returncode == 0, status.stderr + assert "Augment Code CLI" in status.stdout + assert "[installed]" in status.stdout + + uninstall = subprocess.run( # nosec B603 + [*base, "uninstall", "--agent", "augment", "--agent_id", AGENT_ID], + capture_output=True, + check=False, + cwd=root, + env=environment, + text=True, + ) + assert uninstall.returncode == 0, uninstall.stderr + assert not fixture.config_path.exists() + assert not fixture.agent_dir.exists() @pytest.mark.parametrize( @@ -407,6 +297,7 @@ def test_uninstall_removes_settings_owned_entirely_by_elydora( "{ malformed", "null", "[]", + '{"hooks":{},"hooks":{}}', {"hooks": None}, {"hooks": {"UnknownEvent": []}}, {"hooks": {"PreToolUse": None}}, @@ -443,56 +334,20 @@ def test_uninstall_removes_settings_owned_entirely_by_elydora( }, ], ) -def test_install_rejects_malformed_settings_before_writes( - monkeypatch: pytest.MonkeyPatch, - tmp_path: Path, - existing: object, +def test_install_rejects_invalid_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) - + with pytest.raises((FileNotFoundError, ValueError)): + fixture.install() 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()) + for path in ( + fixture.guard_path, + fixture.hook_path, + fixture.runtime_config_path, + fixture.private_key_path, + fixture.guard_wrapper_path, + fixture.audit_wrapper_path, + ): + assert not path.exists() diff --git a/sdks/python/tests/test_augment_runtime.py b/sdks/python/tests/test_augment_runtime.py new file mode 100644 index 0000000..8a25ee9 --- /dev/null +++ b/sdks/python/tests/test_augment_runtime.py @@ -0,0 +1,194 @@ +from __future__ import annotations + +import json +from pathlib import Path + +import pytest + +from augment_support import ( + ElydoraApiHandler, + managed_handler, + prepare_fixture, + run_handler, + server_base_url, + start_api_server, +) + + +def encoded(value: object) -> bytes: + return json.dumps(value, separators=(",", ":")).encode() + + +def official_payload(event: str, **overrides: object) -> dict[str, object]: + value: dict[str, object] = { + "hook_event_name": event, + "conversation_id": "conversation-1", + "workspace_roots": ["/workspace"], + "tool_name": "launch-process", + "tool_input": {"command": "pytest", "nested": {"preserve": True}}, + "is_mcp_tool": False, + "conversation_data": [{"role": "user", "content": "preserve this"}], + "mcp_metadata": {"server": "local", "transport": "stdio"}, + "user_context": {"account": "user-1"}, + "future_field": {"survives": ["exactly", 2]}, + } + value.update(overrides) + return value + + +@pytest.mark.parametrize("status", ["frozen", "revoked"]) +def test_guard_propagates_official_blocking_exit_code( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, + status: str, +) -> None: + server = start_api_server(status=status) + try: + fixture = prepare_fixture( + monkeypatch, tmp_path, base_url=server_base_url(server) + ) + fixture.install() + handler = managed_handler( + fixture.settings(), "PreToolUse", fixture.guard_wrapper_path + ) + result = run_handler(handler, encoded(official_payload("PreToolUse")), fixture) + finally: + server.shutdown() + server.server_close() + + assert result.returncode == 2 + assert result.stdout == b"" + assert status.encode() in result.stderr.lower() + assert b"Tool execution blocked" in result.stderr + + +def test_audit_preserves_complete_native_hook_payload( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + server = start_api_server() + payload = official_payload( + "PostToolUse", + tool_output={"stdout": "passed", "stderr": "", "exit_code": 0}, + ) + try: + fixture = prepare_fixture( + monkeypatch, tmp_path, base_url=server_base_url(server) + ) + fixture.install() + handler = managed_handler( + fixture.settings(), "PostToolUse", fixture.audit_wrapper_path + ) + result = run_handler(handler, encoded(payload), fixture) + finally: + server.shutdown() + server.server_close() + + assert result.returncode == 0 + request = next( + item for item in ElydoraApiHandler.requests if item["method"] == "POST" + ) + assert request["json"]["payload"] == payload + assert request["authorization"] == "Bearer token-1" + + +def test_status_requires_exact_runtime_identity_key_scripts_and_wrappers( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + fixture = prepare_fixture(monkeypatch, tmp_path) + fixture.install() + assert fixture.plugin.status()["installed"] is True + + fixture.guard_wrapper_path.unlink() + assert fixture.plugin.status()["installed"] is False + fixture.install() + + fixture.hook_path.unlink() + assert fixture.plugin.status()["installed"] is False + fixture.install() + + fixture.audit_wrapper_path.write_text("tampered wrapper\n", encoding="utf-8") + assert fixture.plugin.status()["installed"] is False + fixture.install() + + fixture.private_key_path.write_text("invalid", encoding="utf-8") + with pytest.raises(ValueError, match="private key.*canonical 32-byte"): + fixture.plugin.status() + + +@pytest.mark.parametrize( + ("source", "pattern"), + [ + ("{ malformed", "parse Elydora runtime config"), + ('{"agent_name":"augment","agent_name":"augment"}', "duplicate field"), + ( + json.dumps( + { + "org_id": "org-1", + "agent_id": "other-agent", + "kid": "key-1", + "base_url": "https://api.elydora.com", + "agent_name": "augment", + } + ), + "identity does not match", + ), + ( + json.dumps( + { + "org_id": "org-1", + "agent_id": "agent-1", + "kid": "key-1", + "base_url": "https://api.elydora.com", + "agent_name": "augment", + "hidden": True, + } + ), + "unsupported field", + ), + ], +) +def test_status_surfaces_invalid_runtime_metadata( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, + source: str, + pattern: str, +) -> None: + fixture = prepare_fixture(monkeypatch, tmp_path) + fixture.install() + fixture.runtime_config_path.write_text(source, encoding="utf-8") + with pytest.raises(ValueError, match=pattern): + fixture.plugin.status() + + +def test_status_ignores_incomplete_managed_hook_pairs( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + fixture = prepare_fixture(monkeypatch, tmp_path) + fixture.install() + settings = fixture.settings() + del settings["hooks"]["PostToolUse"] + fixture.config_path.write_text(json.dumps(settings), encoding="utf-8") + status = fixture.plugin.status() + assert status["installed"] is False + assert status["details"] == "Not installed" + + +def test_runtime_failures_remain_observable_and_fail_open( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + fixture = prepare_fixture(monkeypatch, tmp_path) + fixture.install() + settings = fixture.settings() + guard = managed_handler(settings, "PreToolUse", fixture.guard_wrapper_path) + audit = managed_handler(settings, "PostToolUse", fixture.audit_wrapper_path) + guard_result = run_handler(guard, encoded(official_payload("PreToolUse")), fixture) + audit_result = run_handler(audit, encoded(official_payload("PostToolUse")), fixture) + malformed = run_handler(audit, b"{ malformed", fixture) + + assert guard_result.returncode == 0 + assert b"Failed to resolve agent status" in guard_result.stderr + assert audit_result.returncode == 0 + assert malformed.returncode == 0 + log = (fixture.agent_dir / "error.log").read_text(encoding="utf-8") + assert "invalid JSON" in log + assert "refused" in log.lower() or "urlopen" in log.lower() diff --git a/sdks/python/tests/test_augment_transaction.py b/sdks/python/tests/test_augment_transaction.py new file mode 100644 index 0000000..6a7d3d3 --- /dev/null +++ b/sdks/python/tests/test_augment_transaction.py @@ -0,0 +1,259 @@ +from __future__ import annotations + +import os +from pathlib import Path +import subprocess # nosec B404 +import sys + +import pytest + +from augment_support import ( + AGENT_ID, + VALID_PRIVATE_KEY, + prepare_fixture, + symlink_or_skip, + write_json, + write_text, +) +from elydora.plugins import _transaction, augment, augment_installation +from elydora.plugins.augment_contract import render_augment_document +from elydora.plugins.augment_io import read_augment_document + + +RUNTIME_NAMES = ( + "guard.py", + "config.json", + "private.key", + "hook.py", + "augment-guard.cmd" if os.name == "nt" else "augment-guard.sh", + "augment-hook.cmd" if os.name == "nt" else "augment-hook.sh", +) + + +def prepare_installation(fixture: object) -> list[_transaction.FileChange]: + document = read_augment_document() + paths = augment_installation.preflight_augment_installation( + fixture.config, document + ) + rendered = render_augment_document( + document, augment._installed_hooks(document.hooks, paths) + ) + return augment_installation.prepare_augment_installation( + fixture.config, paths, rendered + ) + + +def assert_no_transaction_files(root: Path) -> None: + assert not list(root.rglob("*.tmp")) + assert not list(root.rglob("*.rollback")) + + +def test_transaction_rolls_back_all_seven_files( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + original = '{"telemetryEnabled":true}\n' + fixture = prepare_fixture(monkeypatch, tmp_path, existing_settings=original) + real_replace = _transaction.os.replace + failed = False + + def fail_settings_once(source: str, destination: str) -> None: + nonlocal failed + if os.path.abspath(destination) == str(fixture.config_path) and not failed: + failed = True + raise OSError("injected Auggie settings failure") + real_replace(source, destination) + + monkeypatch.setattr(_transaction.os, "replace", fail_settings_once) + with pytest.raises(OSError, match="injected Auggie settings failure"): + fixture.install() + + assert fixture.config_path.read_text(encoding="utf-8") == original + for name in RUNTIME_NAMES: + assert not (fixture.agent_dir / name).exists() + assert_no_transaction_files(fixture.home_dir) + + +def test_concurrent_settings_change_is_detected_before_runtime_commits( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + fixture = prepare_fixture( + monkeypatch, tmp_path, existing_settings={"telemetryEnabled": True} + ) + changes = prepare_installation(fixture) + concurrent = '{"telemetryEnabled":false,"hooks":{"Notification":[]}}\n' + fixture.config_path.write_text(concurrent, encoding="utf-8") + + with pytest.raises(OSError, match="changed during installation"): + augment_installation.commit_augment_installation(changes) + + assert fixture.config_path.read_text(encoding="utf-8") == concurrent + for name in RUNTIME_NAMES: + assert not (fixture.agent_dir / name).exists() + assert_no_transaction_files(fixture.home_dir) + + +def test_stale_settings_are_rejected_before_staging( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + fixture = prepare_fixture(monkeypatch, tmp_path) + document = read_augment_document() + paths = augment_installation.preflight_augment_installation( + fixture.config, document + ) + rendered = render_augment_document( + document, augment._installed_hooks(document.hooks, paths) + ) + write_text(fixture.config_path, '{"hooks":{"Notification":[]}}\n') + with pytest.raises(OSError, match="changed before staging"): + augment_installation.prepare_augment_installation( + fixture.config, paths, rendered + ) + assert not fixture.agent_dir.exists() + assert_no_transaction_files(fixture.home_dir) + + +@pytest.mark.parametrize( + "runtime_name", + [ + "private.key", + "guard.py", + "hook.py", + "augment-guard.cmd" if os.name == "nt" else "augment-guard.sh", + "augment-hook.cmd" if os.name == "nt" else "augment-hook.sh", + "chain-state.json", + "status-cache.json", + "error.log", + ], +) +def test_orphan_runtime_artifacts_fail_before_settings_writes( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, + runtime_name: str, +) -> None: + fixture = prepare_fixture(monkeypatch, tmp_path) + artifact = fixture.agent_dir / runtime_name + write_text(artifact, "orphan\n") + with pytest.raises(ValueError, match="identity cannot be verified"): + fixture.install() + assert artifact.read_text(encoding="utf-8") == "orphan\n" + assert not fixture.config_path.exists() + + +def test_mismatched_runtime_identity_fails_before_writes( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + fixture = prepare_fixture(monkeypatch, tmp_path) + write_json( + fixture.runtime_config_path, + { + "agent_id": "another-agent", + "agent_name": "augment", + }, + ) + original = fixture.runtime_config_path.read_text(encoding="utf-8") + with pytest.raises(ValueError, match="identity does not match"): + fixture.install() + assert fixture.runtime_config_path.read_text(encoding="utf-8") == original + assert not fixture.config_path.exists() + + +@pytest.mark.parametrize("kind", ["configuration", "settings", "runtime", "wrapper"]) +def test_linked_settings_runtime_and_wrapper_paths_are_rejected( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, + kind: str, +) -> None: + fixture = prepare_fixture(monkeypatch, tmp_path) + target = tmp_path / f"{kind}-target" + if kind == "settings": + write_text(target, "{}\n") + fixture.config_path.parent.mkdir(parents=True) + symlink_or_skip(target, fixture.config_path) + elif kind == "configuration": + target.mkdir() + fixture.config_path.parent.parent.mkdir(parents=True) + symlink_or_skip(target, fixture.config_path.parent, directory=True) + elif kind == "runtime": + target.mkdir() + fixture.agent_dir.parent.parent.mkdir(parents=True) + symlink_or_skip(target, fixture.agent_dir.parent, directory=True) + else: + fixture.install() + write_text(target, "external wrapper\n") + fixture.guard_wrapper_path.unlink() + symlink_or_skip(target, fixture.guard_wrapper_path) + + with pytest.raises(OSError, match="physical (file|directory)"): + fixture.install() + + +@pytest.mark.parametrize( + ("field", "value", "pattern"), + [ + ("agent_name", "codex", "requires agent_name augment"), + ("agent_id", "../escape", "Invalid agent ID"), + ("private_key", "invalid", "canonical 32-byte"), + ("token", "", "non-empty string"), + ("base_url", "https://api.elydora.com?q=1", "query parameters"), + ("guard_script_path", "outside", "managed agent directory"), + ], +) +def test_install_validates_runtime_inputs_before_writes( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, + field: str, + value: str, + pattern: str, +) -> None: + fixture = prepare_fixture(monkeypatch, tmp_path) + fixture.config[field] = value + with pytest.raises(ValueError, match=pattern): + fixture.install() + assert not fixture.config_path.exists() + assert not fixture.agent_dir.exists() + + +def test_installation_leaves_no_transaction_artifacts( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + fixture = prepare_fixture(monkeypatch, tmp_path) + fixture.install() + assert_no_transaction_files(fixture.home_dir) + + +def test_cli_preflight_preserves_malformed_settings(tmp_path: Path) -> None: + home_dir = tmp_path / "cli-home" + settings_path = home_dir / ".augment" / "settings.json" + key_path = tmp_path / "private.key" + write_text(settings_path, "{ malformed") + write_text(key_path, VALID_PRIVATE_KEY) + environment = {**os.environ, "HOME": str(home_dir), "USERPROFILE": str(home_dir)} + result = subprocess.run( # nosec B603 + [ + sys.executable, + "-m", + "elydora.cli", + "install", + "--agent", + "augment", + "--org_id", + "org-1", + "--agent_id", + AGENT_ID, + "--private_key_file", + str(key_path), + "--kid", + "key-1", + "--base_url", + "https://api.elydora.com", + ], + capture_output=True, + check=False, + cwd=Path(__file__).parents[1], + env=environment, + text=True, + ) + assert result.returncode == 1 + assert "parse Auggie user settings" in result.stderr + assert settings_path.read_text(encoding="utf-8") == "{ malformed" + assert not (home_dir / ".elydora").exists()