fix(python): mirror transactional Auggie hooks

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.

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.
 

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:
+def _installed_hooks(hooks: AugmentHooks, paths: AugmentRuntimePaths) -> AugmentHooks:
-    config_path = resolve_config_path()
+    cleaned, _ = remove_managed_hooks(hooks)
-    root = _read_json(config_path, "Auggie settings")
+    return {
-    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()
-        _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)],
+                "hooks": [build_handler(paths.guard_wrapper_path)],
             },
         ],
         "PostToolUse": [
             *cleaned.get("PostToolUse", []),
             {
                 "matcher": ".*",
-                    "hooks": [build_handler(wrappers.audit_path)],
+                "hooks": [build_handler(paths.audit_wrapper_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"),
+class AugmentPlugin(AgentPlugin):
-            runtime_config,
+    """Install Elydora into Auggie's global user settings."""
-            0o600,
+
-            "Elydora runtime config",
+    manages_guard_runtime = True
-        )
+
-        _write_text_atomic(
+    def preflight_install(self, config: InstallConfig) -> None:
-            os.path.join(agent_directory, "private.key"),
+        document = read_augment_document()
-            config.get("private_key", ""),
+        _validate_matchers(document.hooks)
-            0o600,
+        preflight_augment_installation(config, document)
-            "Elydora private key",
+
-        )
+    def install(self, config: InstallConfig) -> None:
-        _write_text_atomic(
+        document = read_augment_document()
-            audit_path,
+        _validate_matchers(document.hooks)
-            audit_script,
+        paths = preflight_augment_installation(config, document)
-            0o700,
+        rendered = render_augment_document(
-            "Elydora audit runtime",
+            document, _installed_hooks(document.hooks, paths)
-        )
-        _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",
         )
+        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()
+        document = read_augment_document()
-        if not source.exists:
+        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)
+        rendered = render_augment_document(document, hooks)
-        if hooks:
+        commit_augment_uninstall(prepare_augment_uninstall(rendered))
-            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()
+        document = read_augment_document()
-        contracts = runtime_contracts(source.hooks)
+        contracts = runtime_contracts(document.hooks)
-        if not contracts:
+        configured = bool(contracts)
-            return PluginStatus(
+        installed = configured and augment_runtime_files_exist(contracts)
-                installed=False,
-                agent=AGENT_KEY,
-                details="Not installed",
-            )
-        installed = _runtime_files_exist(contracts)
         details = (
-            f"Config: {source.config_path}"
+            f"Config: {document.config_path}"
             if installed
-            else (
+            else f"Configured at {document.config_path}; managed contract incomplete"
-                f"Configured at {source.config_path}; "
-                "runtime scripts or wrappers missing"
-            )
         )
         return PluginStatus(
             installed=installed,
             agent=AGENT_KEY,
-            details=details,
+            details=details if configured else "Not installed",
         )

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:
+def wrapper_paths(agent_directory: 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),
@@ -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,

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")

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)

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}")

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

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
+import subprocess  # nosec B404
-from typing import Any
+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"
+def test_augment_is_registered_and_owns_its_complete_runtime() -> None:
-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
+    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": "(?<tool>launch-process)",
-                    "hooks": [
+                    "hooks": [{"type": "command", "command": "user-command"}],
-                        {
-                            "type": "command",
-                            "command": "user-command",
-                        }
-                    ],
                 }
             ],
         },
     }
     fixture = prepare_fixture(monkeypatch, tmp_path, existing_settings=existing)
-    fixture.plugin.install(fixture.config)
+    fixture.install()
-    fixture.plugin.install(fixture.config)
+    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 (
+    assert not (fixture.project_dir / ".augment" / "settings.json").exists()
-        json.loads(fixture.runtime_config_path.read_text(encoding="utf-8"))[
+
-            "agent_name"
+
+def test_install_replaces_stale_handlers_and_preserves_empty_groups(
+    monkeypatch: pytest.MonkeyPatch, tmp_path: Path
+) -> None:
+    fixture = prepare_fixture(monkeypatch, tmp_path)
+    fixture.install()
+    settings = fixture.settings()
+    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,
+                    }
                 ]
-        == "augment"
+            }
         )
-    assert fixture.private_key_path.read_text(encoding="utf-8") == "test-key"
+    fixture.config_path.write_text(json.dumps(settings, indent=2), encoding="utf-8")
-    assert (fixture.workspace_dir / ".augment" / "settings.json").exists() is False
+    fixture.install()
-    assert (
+    current = fixture.settings()
-        fixture.workspace_dir / ".augment" / "settings.local.json"
+    assert "agent-old" not in json.dumps(current)
-    ).exists() is False
+    assert current["hooks"]["PreToolUse"][0] == {
-
+        "hooks": [],
-
+        "label": "keep empty group",
-def test_wrappers_block_and_forward_official_payload_byte_for_byte(
+    }
-    monkeypatch: pytest.MonkeyPatch, tmp_path: Path
+    assert len(current["hooks"]["PreToolUse"]) == 2
-) -> None:
+    assert len(current["hooks"]["PostToolUse"]) == 1
-    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(
@@ -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)
+    fixture.install()
-    settings = json.loads(fixture.config_path.read_text(encoding="utf-8"))
+    settings = fixture.settings()
-    settings["hooks"]["PreToolUse"].insert(
+    settings["hooks"]["PreToolUse"][0]["hooks"].append(
-        0, {"hooks": [], "label": "keep empty group"}
+        {"type": "command", "command": "user-command"}
-    )
-    settings["hooks"]["PreToolUse"][1]["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"] == [
-    assert remaining["hooks"]["PreToolUse"][0] == {
+        {"type": "command", "command": "user-command"}
-        "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
@@ -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(
+def test_install_rejects_invalid_settings_before_writes(
-    monkeypatch: pytest.MonkeyPatch,
+    monkeypatch: pytest.MonkeyPatch, tmp_path: Path, existing: object
-    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((FileNotFoundError, ValueError)):
-    with pytest.raises(ValueError):
+        fixture.install()
-        fixture.plugin.install(fixture.config)
-
     assert fixture.config_path.read_text(encoding="utf-8") == original
-    assert fixture.hook_path.exists() is False
+    for path in (
-    assert fixture.runtime_config_path.exists() is False
+        fixture.guard_path,
-    assert fixture.private_key_path.exists() is False
+        fixture.hook_path,
-    assert fixture.guard_wrapper_path.exists() is False
+        fixture.runtime_config_path,
-    assert fixture.audit_wrapper_path.exists() is False
+        fixture.private_key_path,
-
+        fixture.guard_wrapper_path,
-
+        fixture.audit_wrapper_path,
-def test_install_rejects_missing_guard_before_creating_files(
+    ):
-    monkeypatch: pytest.MonkeyPatch, tmp_path: Path
+        assert not path.exists()
-) -> 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())

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()

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()