fix(python): mirror official Gemini hooks

sdks/python/README.md

@@ -84,6 +84,8 @@ Codex installation writes exact `PreToolUse` and `PostToolUse` command groups to
 
 Cursor installation writes native global `preToolUse`, `postToolUse`, and `postToolUseFailure` handlers to `~/.cursor/hooks.json`. Elydora forwards Cursor's native success and failure payloads, preserves unrelated user hooks, and leaves project and enterprise sources unchanged. The generated guard and audit handlers use a 10-second fail-closed boundary, exact freeze exit-code propagation, protected runtime credentials, and one rollback-capable transaction for all runtime and hook configuration files.
 
+Gemini CLI installation writes named, matchless `BeforeTool` and `AfterTool` command hooks to `$GEMINI_CLI_HOME/.gemini/settings.json` (`~/.gemini/settings.json` by default). Empty home overrides select the OS home; relative and literal-tilde values retain Gemini CLI's native path behavior. JSON comments and unrelated settings remain intact. Workspace, system-default, system-override, and extension hook sources remain unchanged and continue loading additively. User settings, generated runtimes, runtime config, and private key commit through one rollback-capable transaction. Managed handlers preserve the complete native snake_case payload, emit valid JSON on successful execution, propagate freeze and revocation through exit code `2`, and use encoded PowerShell commands on Windows. `hooksConfig.enabled` and `hooksConfig.disabled` remain authoritative. Run `/hooks list` after installation to inspect effective hooks from every source.
+
 Cline installation writes `PreToolUse.mjs` and `PostToolUse.mjs` to `$CLINE_DIR/hooks` (default `~/.cline/hooks`). Elydora leaves the Documents and workspace hook roots unchanged. The wrappers preserve Cline's official payload and translate a frozen guard into JSON stdout cancellation control.
 
 Kimi installation writes exact `PreToolUse`, `PostToolUse`, and `PostToolUseFailure` rules to every detected runtime: Kimi Code's `$KIMI_CODE_HOME/config.toml` (default `~/.kimi-code/config.toml`) and the legacy Python CLI's `~/.kimi/config.toml`. A fresh installation targets stable Kimi Code; an existing legacy home activates the migration contract. Empty `KIMI_CODE_HOME` values use the default. The selected TOML files, generated runtimes, runtime config, and private key commit as one rollback-capable transaction. Managed commands preserve Kimi's native snake_case event JSON, propagate freeze and revocation through exit code `2`, and use encoded PowerShell on Windows. Run `/hooks` to inspect the global rules.

sdks/python/elydora/plugins/gemini.py

@@ -1,206 +1,101 @@
-"""Gemini CLI plugin — merges AfterTool hook into ~/.gemini/settings.json."""
+"""Gemini CLI native global user-hook integration."""
 
 from __future__ import annotations
 
-import json
-import os
-import sys
-
-from ._file_io import write_json_atomic, write_text_atomic
 from .base import AgentPlugin, InstallConfig, PluginStatus
-from .hook_template import generate_hook_script
+from .gemini_config import GeminiDocument, render_gemini_document
+from .gemini_contract import (
+    AGENT_KEY,
+    AUDIT_HOOK_NAME,
+    GUARD_HOOK_NAME,
+    build_gemini_group,
+    disabled_managed_gemini_entries,
+    gemini_runtime_contracts,
+    managed_gemini_hooks_enabled,
+)
+from .gemini_installation import (
+    commit_gemini_installation,
+    preflight_gemini_installation,
+    prepare_gemini_installation,
+)
+from .gemini_io import (
+    gemini_runtime_files_exist,
+    read_gemini_document,
+    write_gemini_document,
+)
 
 
-SETTINGS_PATH = os.path.join(os.path.expanduser("~"), ".gemini", "settings.json")
+def _require_enabled_hooks(document: GeminiDocument) -> None:
-ELYDORA_DIR = os.path.join(os.path.expanduser("~"), ".elydora")
+    if not document.hook_controls.enabled:
+        raise ValueError(
+            "Gemini CLI hooks are disabled by hooksConfig.enabled: "
+            f"{document.file_path}"
+        )
+    disabled = disabled_managed_gemini_entries(document.hook_controls)
+    if disabled:
+        raise ValueError(
+            "Gemini CLI hooks are disabled by hooksConfig.disabled: "
+            + ", ".join(disabled)
+        )
 
 
 class GeminiPlugin(AgentPlugin):
-    """Install/uninstall Elydora audit hook for Gemini CLI."""
+    """Install Elydora into Gemini CLI's native user settings."""
 
-    @staticmethod
+    manages_guard_runtime = True
-    def _hook_path_for(agent_id: str) -> str:
+
-        return os.path.join(ELYDORA_DIR, agent_id, "hook.py")
+    def preflight_install(self, config: InstallConfig) -> None:
+        document = read_gemini_document()
+        _require_enabled_hooks(document)
+        preflight_gemini_installation(config, document)
 
     def install(self, config: InstallConfig) -> None:
-        agent_id = config.get("agent_id", "")
+        document = read_gemini_document()
-        agent_name = config.get("agent_name", "")
+        _require_enabled_hooks(document)
-
+        paths = preflight_gemini_installation(config, document)
-        # Create per-agent directory
+        rendered = render_gemini_document(
-        agent_dir = os.path.join(ELYDORA_DIR, agent_id)
+            document,
-        os.makedirs(agent_dir, exist_ok=True)
+            None,
-
-        # Write config.json
-        config_data = {
-            "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_name,
-        }
-        config_path = os.path.join(agent_dir, "config.json")
-        write_json_atomic(
-            config_path,
-            config_data,
-            0o600,
-            "Elydora runtime config",
-        )
-
-        # Write private key
-        private_key_path = os.path.join(agent_dir, "private.key")
-        write_text_atomic(
-            private_key_path,
-            config.get("private_key", ""),
-            0o600,
-            "Elydora private key",
-        )
-
-        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"),
-        )
-        hook_path = self._hook_path_for(agent_id)
-        write_text_atomic(
-            hook_path,
-            script,
-            0o700,
-            "Elydora audit runtime",
-        )
-
-        guard_script_path = config.get("guard_script_path", "")
-        python_exe = sys.executable
-
-        settings = _load_json(SETTINGS_PATH)
-        hooks = settings.setdefault("hooks", {})
-
-        # --- BeforeTool (guard — freeze enforcement) ---
-        before_tool = hooks.setdefault("BeforeTool", [])
-        before_tool[:] = [h for h in before_tool if not _is_elydora_hook(h)]
-        if guard_script_path:
-            before_tool.append({
-                "hooks": [
             {
-                        "type": "command",
+                "BeforeTool": build_gemini_group(
-                        "command": f'"{python_exe}" {guard_script_path}',
+                    paths.guard_path, GUARD_HOOK_NAME
-                    }
+                ),
-                ],
+                "AfterTool": build_gemini_group(
-            })
+                    paths.audit_path, AUDIT_HOOK_NAME
-
+                ),
-        # --- AfterTool (audit logging) ---
+            },
-        after_tool = hooks.setdefault("AfterTool", [])
+        )
-
+        changes = prepare_gemini_installation(config, paths, rendered)
-        after_tool[:] = [h for h in after_tool if not _is_elydora_hook(h)]
+        commit_gemini_installation(changes)
-
+        print(f"Gemini CLI hooks installed at {document.file_path}")
-        after_tool.append({
+        print("Gemini CLI verification: run /hooks list.")
-            "hooks": [
-                {
-                    "type": "command",
-                    "command": hook_path,
-                }
-            ],
-        })
-
-        _save_json(SETTINGS_PATH, settings)
-        print("Elydora hook installed for Gemini CLI.")
-        print(f"  Hook script: {hook_path}")
-        print(f"  Settings: {SETTINGS_PATH}")
 
     def uninstall(self, agent_id: str = "") -> None:
-        if os.path.exists(SETTINGS_PATH):
+        document = read_gemini_document()
-            settings = _load_json(SETTINGS_PATH)
+        write_gemini_document(
-            hooks = settings.get("hooks", {})
+            render_gemini_document(document, agent_id or None, {})
-            changed = False
+        )
-
-            # Remove BeforeTool entries
-            before_tool = hooks.get("BeforeTool", [])
-            before_filtered = [h for h in before_tool if not _is_elydora_hook(h, agent_id)]
-            if len(before_filtered) != len(before_tool):
-                hooks["BeforeTool"] = before_filtered
-                if not before_filtered:
-                    del hooks["BeforeTool"]
-                changed = True
-
-            # Remove AfterTool entries
-            after_tool = hooks.get("AfterTool", [])
-            after_filtered = [h for h in after_tool if not _is_elydora_hook(h, agent_id)]
-            if len(after_filtered) != len(after_tool):
-                hooks["AfterTool"] = after_filtered
-                if not after_filtered:
-                    del hooks["AfterTool"]
-                changed = True
-
-            if changed:
-                if not hooks:
-                    del settings["hooks"]
-                _save_json(SETTINGS_PATH, settings)
-
-        # Hook script removal is handled by cli.py cmd_uninstall (rmtree of agent dir)
-        print("Elydora hook uninstalled from Gemini CLI.")
 
     def status(self) -> PluginStatus:
-        # Scan ~/.elydora/*/hook.py for any installed hook
+        document = read_gemini_document()
-        import glob as _glob
+        contracts = gemini_runtime_contracts(document.hooks)
-        hook_pattern = os.path.join(ELYDORA_DIR, "*", "hook.py")
+        controls_enabled = managed_gemini_hooks_enabled(
-        hook_files = _glob.glob(hook_pattern)
+            document.hook_controls
-        hook_exists = len(hook_files) > 0
+        )
-
+        configured = controls_enabled and bool(contracts)
-        settings_configured = False
+        installed = configured and gemini_runtime_files_exist(contracts)
-        if os.path.exists(SETTINGS_PATH):
-            settings = _load_json(SETTINGS_PATH)
-            hooks = settings.get("hooks", {})
-            before_tool = hooks.get("BeforeTool", [])
-            after_tool = hooks.get("AfterTool", [])
-            before_configured = any(_is_elydora_hook(h) for h in before_tool)
-            after_configured = any(_is_elydora_hook(h) for h in after_tool)
-            settings_configured = before_configured and after_configured
-
-        installed = hook_exists and settings_configured
         if installed:
-            details = f"Found {len(hook_files)} agent(s): {', '.join(hook_files)}"
+            details = f"Config: {document.file_path}"
-        elif hook_exists:
+        elif contracts and not controls_enabled:
-            details = "Hook script exists but not configured in settings"
+            details = f"Configured hooks are disabled: {document.file_path}"
-        elif settings_configured:
+        elif contracts:
-            details = "Configured in settings but hook script missing"
+            details = (
+                f"Configured at {document.file_path}; managed contract incomplete"
+            )
         else:
             details = "Not installed"
-
+        return PluginStatus(
-        return PluginStatus(installed=installed, agent="gemini", details=details)
+            installed=installed,
-
+            agent=AGENT_KEY,
-
+            details=details,
-def _is_elydora_hook(entry: dict, agent_id: str = "") -> bool:
+        )
-    # Collect all command strings from the entry
-    commands: list[str] = []
-    inner_hooks = entry.get("hooks")
-    if isinstance(inner_hooks, list):
-        for hook in inner_hooks:
-            if isinstance(hook, dict) and isinstance(hook.get("command"), str):
-                commands.append(hook["command"])
-    else:
-        command = entry.get("command")
-        if isinstance(command, str):
-            commands.append(command)
-
-    for cmd in commands:
-        cmd_lower = cmd.lower()
-        if "elydora" not in cmd_lower:
-            continue
-        # If agent_id is specified, only match hooks for that specific agent
-        if agent_id and agent_id in cmd:
-            return True
-        if not agent_id:
-            return True
-    return False
-
-
-def _load_json(path: str) -> dict:
-    if not os.path.exists(path):
-        return {}
-    with open(path, "r", encoding="utf-8") as f:
-        return json.load(f)
-
-
-def _save_json(path: str, data: dict) -> None:
-    write_json_atomic(path, data, 0o600, "Gemini settings")

sdks/python/elydora/plugins/gemini_command.py

@@ -0,0 +1,206 @@
+"""Gemini CLI hook command construction and exact runtime ownership."""
+
+from __future__ import annotations
+
+import base64
+from dataclasses import dataclass
+import ntpath
+import os
+import re
+import sys
+from typing import Optional, Tuple
+
+
+@dataclass(frozen=True)
+class GeminiRuntimeReference:
+    agent_id: str
+    script_path: str
+
+
+def same_gemini_path(left: str, right: str) -> bool:
+    return os.path.normcase(os.path.abspath(left)) == os.path.normcase(
+        os.path.abspath(right)
+    )
+
+
+def same_gemini_agent_id(left: str, right: str) -> bool:
+    return os.path.normcase(left) == os.path.normcase(right)
+
+
+def _quote_posix(value: str) -> str:
+    return "'" + value.replace("'", "'\"'\"'") + "'"
+
+
+def _quote_powershell(value: str) -> str:
+    return "'" + value.replace("'", "''") + "'"
+
+
+def _windows_powershell_path() -> str:
+    configured = os.environ.get("SystemRoot")
+    system_root = (
+        configured
+        if configured
+        and ntpath.isabs(configured)
+        and re.search(r'["%\r\n]', configured) is None
+        else r"C:\Windows"
+    )
+    return ntpath.join(
+        system_root,
+        "System32",
+        "WindowsPowerShell",
+        "v1.0",
+        "powershell.exe",
+    )
+
+
+def _windows_command(script_path: str) -> str:
+    source = (
+        f"& {_quote_powershell(sys.executable)} "
+        f"{_quote_powershell(script_path)}; exit $LASTEXITCODE"
+    )
+    encoded = base64.b64encode(source.encode("utf-16le")).decode("ascii")
+    return (
+        f"& {_quote_powershell(_windows_powershell_path())} "
+        "-NoLogo -NoProfile -NonInteractive -EncodedCommand "
+        f"{encoded}"
+    )
+
+
+def build_gemini_command(script_path: str) -> str:
+    if not os.path.isabs(sys.executable) or not os.path.isabs(script_path):
+        raise ValueError(
+            "Gemini CLI hook commands require absolute executable and script paths"
+        )
+    if os.name == "nt":
+        return _windows_command(script_path)
+    return f"{_quote_posix(sys.executable)} {_quote_posix(script_path)}"
+
+
+def _read_posix_argument(command: str, start: int) -> Optional[Tuple[str, int]]:
+    if start >= len(command) or command[start] != "'":
+        return None
+    apostrophe = "'\"'\"'"
+    value = ""
+    index = start + 1
+    while index < len(command):
+        if command.startswith(apostrophe, index):
+            value += "'"
+            index += len(apostrophe)
+            continue
+        if command[index] == "'":
+            return value, index + 1
+        value += command[index]
+        index += 1
+    return None
+
+
+def _parse_posix_command(command: str) -> Optional[Tuple[str, str]]:
+    executable = _read_posix_argument(command, 0)
+    if executable is None or executable[1] >= len(command):
+        return None
+    if command[executable[1]] != " ":
+        return None
+    script = _read_posix_argument(command, executable[1] + 1)
+    if script is None or script[1] != len(command):
+        return None
+    return executable[0], script[0]
+
+
+def _read_powershell_argument(
+    command: str, start: int
+) -> Optional[Tuple[str, int]]:
+    if start >= len(command) or command[start] != "'":
+        return None
+    value = ""
+    index = start + 1
+    while index < len(command):
+        if command[index] != "'":
+            value += command[index]
+            index += 1
+            continue
+        if index + 1 < len(command) and command[index + 1] == "'":
+            value += "'"
+            index += 2
+            continue
+        return value, index + 1
+    return None
+
+
+def _parse_powershell_source(source: str) -> Optional[Tuple[str, str]]:
+    if not source.startswith("& "):
+        return None
+    executable = _read_powershell_argument(source, 2)
+    if executable is None or executable[1] >= len(source):
+        return None
+    if source[executable[1]] != " ":
+        return None
+    script = _read_powershell_argument(source, executable[1] + 1)
+    if script is None or source[script[1] :] != "; exit $LASTEXITCODE":
+        return None
+    return executable[0], script[0]
+
+
+def _parse_windows_command(command: str) -> Optional[Tuple[str, str]]:
+    if not command.startswith("& "):
+        return None
+    powershell = _read_powershell_argument(command, 2)
+    if (
+        powershell is None
+        or not ntpath.isabs(powershell[0])
+        or ntpath.basename(powershell[0]).lower() != "powershell.exe"
+    ):
+        return None
+    prefix = " -NoLogo -NoProfile -NonInteractive -EncodedCommand "
+    if not command.startswith(prefix, powershell[1]):
+        return None
+    encoded = command[powershell[1] + len(prefix) :]
+    if re.fullmatch(r"[A-Za-z0-9+/]+={0,2}", encoded) is None:
+        return None
+    try:
+        payload = base64.b64decode(encoded, validate=True)
+        if base64.b64encode(payload).decode("ascii") != encoded:
+            return None
+        if len(payload) % 2:
+            return None
+        source = payload.decode("utf-16le")
+    except (ValueError, UnicodeDecodeError):
+        return None
+    return _parse_powershell_source(source)
+
+
+def _parse_legacy_command(
+    command: str, script_name: str
+) -> Optional[Tuple[str, str]]:
+    if script_name == "guard.py":
+        match = re.fullmatch(r'"([^"\r\n]+)" ([^\r\n]+)', command)
+        if match is None or not same_gemini_path(match.group(1), sys.executable):
+            return None
+        return match.group(1), match.group(2)
+    if "\r" in command or "\n" in command:
+        return None
+    return sys.executable, command
+
+
+def gemini_runtime_reference(
+    command: str,
+    script_name: str,
+    include_legacy: bool = False,
+) -> Optional[GeminiRuntimeReference]:
+    parsed = _parse_posix_command(command) or _parse_windows_command(command)
+    if parsed is None and include_legacy:
+        parsed = _parse_legacy_command(command, script_name)
+    if (
+        parsed is None
+        or not same_gemini_path(parsed[0], sys.executable)
+        or not os.path.isabs(parsed[1])
+        or os.path.basename(parsed[1]) != script_name
+    ):
+        return None
+    agent_directory = os.path.dirname(parsed[1])
+    runtime_root = os.path.join(os.path.expanduser("~"), ".elydora")
+    if not same_gemini_path(os.path.dirname(agent_directory), runtime_root):
+        return None
+    agent_id = os.path.basename(agent_directory)
+    if agent_id in {"", ".", ".."}:
+        return None
+    return GeminiRuntimeReference(agent_id, parsed[1])

sdks/python/elydora/plugins/gemini_config.py

@@ -0,0 +1,181 @@
+"""Gemini CLI JSONC parsing and source-preserving hook edits."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import Dict, Optional
+
+from ._jsonc import JsoncEditor
+from .gemini_contract import (
+    MANAGED_EVENTS,
+    GeminiHookControls,
+    GeminiHooks,
+    JsonObject,
+    managed_gemini_removals,
+    read_gemini_hook_controls,
+    read_gemini_hooks,
+)
+
+
+OWNED_FILE_MARKER = "// Managed by Elydora"
+
+
+@dataclass(frozen=True)
+class GeminiDocument:
+    file_path: str
+    exists: bool
+    raw: str
+    root: JsonObject
+    hooks: GeminiHooks
+    hook_controls: GeminiHookControls
+    has_hooks_container: bool
+    owned_file: bool
+
+
+@dataclass(frozen=True)
+class RenderedGeminiDocument:
+    document: GeminiDocument
+    changed: bool
+    next_source: Optional[str]
+
+
+def _label(file_path: str) -> str:
+    return f"Gemini CLI user settings at {file_path}"
+
+
+def parse_gemini_document(
+    *, exists: bool, file_path: str, raw: str
+) -> GeminiDocument:
+    editor = JsoncEditor(
+        raw,
+        _label(file_path),
+        allow_trailing_commas=False,
+    )
+    if not isinstance(editor.value, dict):
+        raise ValueError(f"{_label(file_path)} must contain a JSON object")
+    root = editor.value
+    return GeminiDocument(
+        file_path=file_path,
+        exists=exists,
+        raw=raw,
+        root=root,
+        hooks=(read_gemini_hooks(root["hooks"]) if "hooks" in root else {}),
+        hook_controls=(
+            read_gemini_hook_controls(root["hooksConfig"])
+            if "hooksConfig" in root
+            else GeminiHookControls(True, ())
+        ),
+        has_hooks_container="hooks" in root,
+        owned_file=raw.startswith(OWNED_FILE_MARKER),
+    )
+
+
+def create_gemini_document(file_path: str) -> GeminiDocument:
+    return parse_gemini_document(
+        exists=False,
+        file_path=file_path,
+        raw=f"{OWNED_FILE_MARKER}\n{{}}\n",
+    )
+
+
+def _current_document(
+    document: GeminiDocument, editor: JsoncEditor
+) -> GeminiDocument:
+    return parse_gemini_document(
+        exists=document.exists,
+        file_path=document.file_path,
+        raw=editor.raw,
+    )
+
+
+def _remove_managed(
+    editor: JsoncEditor,
+    document: GeminiDocument,
+    agent_id: Optional[str],
+) -> None:
+    removals = managed_gemini_removals(document.hooks, agent_id)
+    for event in MANAGED_EVENTS:
+        event_removals = sorted(
+            (item for item in removals if item.event == event),
+            key=lambda item: item.group_index,
+            reverse=True,
+        )
+        for removal in event_removals:
+            group_path = ("hooks", event, removal.group_index)
+            if removal.remove_group:
+                editor.delete(group_path)
+                continue
+            for handler_index in sorted(
+                removal.handler_indexes, reverse=True
+            ):
+                editor.delete((*group_path, "hooks", handler_index))
+        if event_removals:
+            current = _current_document(document, editor)
+            if event in current.hooks and not current.hooks[event]:
+                editor.delete(("hooks", event))
+    current = _current_document(document, editor)
+    if current.has_hooks_container and not current.hooks:
+        editor.delete(("hooks",))
+
+
+def _append_group(
+    editor: JsoncEditor,
+    document: GeminiDocument,
+    event: str,
+    group: JsonObject,
+) -> None:
+    current = _current_document(document, editor)
+    if not current.has_hooks_container:
+        editor.add_property((), "hooks", {event: [group]})
+    elif event in current.hooks:
+        editor.append(("hooks", event), group)
+    else:
+        editor.add_property(("hooks",), event, [group])
+
+
+def _already_installed(
+    document: GeminiDocument, additions: Dict[str, JsonObject]
+) -> bool:
+    if set(additions) != set(MANAGED_EVENTS):
+        return False
+    removals = managed_gemini_removals(document.hooks)
+    for event in MANAGED_EVENTS:
+        expected = additions[event]
+        event_removals = [item for item in removals if item.event == event]
+        exact_groups = [
+            group for group in document.hooks.get(event, []) if group == expected
+        ]
+        if (
+            len(event_removals) != 1
+            or not event_removals[0].remove_group
+            or len(exact_groups) != 1
+        ):
+            return False
+    return True
+
+
+def render_gemini_document(
+    document: GeminiDocument,
+    agent_id: Optional[str],
+    additions: Dict[str, JsonObject],
+) -> RenderedGeminiDocument:
+    if agent_id is None and _already_installed(document, additions):
+        return RenderedGeminiDocument(document, False, document.raw)
+    editor = JsoncEditor(
+        document.raw,
+        _label(document.file_path),
+        allow_trailing_commas=False,
+    )
+    _remove_managed(editor, document, agent_id)
+    for event in MANAGED_EVENTS:
+        group = additions.get(event)
+        if group is not None:
+            _append_group(editor, document, event, group)
+    current = _current_document(document, editor)
+    if not additions and document.owned_file and not current.root:
+        return RenderedGeminiDocument(document, True, None)
+    return RenderedGeminiDocument(
+        document,
+        editor.raw != document.raw,
+        editor.raw,
+    )

sdks/python/elydora/plugins/gemini_contract.py

@@ -0,0 +1,344 @@
+"""Gemini CLI hook schema, controls, and exact ownership rules."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+import math
+import os
+from typing import Any, Dict, List, Optional, Tuple
+
+from .gemini_command import (
+    GeminiRuntimeReference,
+    build_gemini_command,
+    gemini_runtime_reference,
+    same_gemini_agent_id,
+)
+
+
+AGENT_KEY = "gemini"
+CONFIG_FILE = "settings.json"
+GUARD_SCRIPT = "guard.py"
+AUDIT_SCRIPT = "hook.py"
+GUARD_HOOK_NAME = "elydora-guard"
+AUDIT_HOOK_NAME = "elydora-audit"
+HOOK_TIMEOUT_MILLISECONDS = 10_000
+MANAGED_EVENTS = ("BeforeTool", "AfterTool")
+
+KNOWN_EVENTS = {
+    "BeforeTool",
+    "AfterTool",
+    "BeforeAgent",
+    "Notification",
+    "AfterAgent",
+    "SessionStart",
+    "SessionEnd",
+    "PreCompress",
+    "BeforeModel",
+    "AfterModel",
+    "BeforeToolSelection",
+}
+
+JsonObject = Dict[str, Any]
+GeminiHooks = Dict[str, List[JsonObject]]
+
+
+@dataclass(frozen=True)
+class GeminiHookControls:
+    enabled: bool
+    disabled: Tuple[str, ...]
+
+
+@dataclass(frozen=True)
+class GeminiRuntimeContract:
+    agent_id: str
+    guard_path: str
+    audit_path: str
+
+
+@dataclass(frozen=True)
+class ManagedRemoval:
+    event: str
+    group_index: int
+    handler_indexes: Tuple[int, ...]
+    remove_group: bool
+
+
+def elydora_dir() -> str:
+    return os.path.join(os.path.expanduser("~"), ".elydora")
+
+
+def _optional_string(value: JsonObject, field: str, label: str) -> None:
+    if field in value and not isinstance(value[field], str):
+        raise ValueError(f'{label} field "{field}" must be a string')
+
+
+def _validate_timeout(value: JsonObject, label: str) -> None:
+    if "timeout" not in value:
+        return
+    timeout = value["timeout"]
+    if (
+        isinstance(timeout, bool)
+        or not isinstance(timeout, (int, float))
+        or not math.isfinite(float(timeout))
+        or timeout < 0
+    ):
+        raise ValueError(f"{label} timeout must be a non-negative finite number")
+
+
+def _validate_environment(value: JsonObject, label: str) -> None:
+    if "env" not in value:
+        return
+    environment = value["env"]
+    if not isinstance(environment, dict) or any(
+        not isinstance(item, str) for item in environment.values()
+    ):
+        raise ValueError(f"{label} env must map names to strings")
+
+
+def _validate_handler(
+    value: Any, event: str, group_index: int, handler_index: int
+) -> JsonObject:
+    label = (
+        f"Gemini CLI settings handler hooks.{event}[{group_index}]"
+        f".hooks[{handler_index}]"
+    )
+    if not isinstance(value, dict):
+        raise ValueError(f"{label} must be an object")
+    handler = dict(value)
+    if handler.get("type") != "command":
+        raise ValueError(
+            f'{label} has unsupported type "{handler.get("type")}"'
+        )
+    for field in ("name", "description", "source"):
+        _optional_string(handler, field, label)
+    _validate_timeout(handler, label)
+    _validate_environment(handler, label)
+    if not isinstance(handler.get("command"), str) or not handler["command"]:
+        raise ValueError(f"{label} requires a non-empty command")
+    return handler
+
+
+def _validate_group(value: Any, event: str, index: int) -> JsonObject:
+    label = f"Gemini CLI settings group hooks.{event}[{index}]"
+    if not isinstance(value, dict):
+        raise ValueError(f"{label} must be an object")
+    group = dict(value)
+    if "matcher" in group and not isinstance(group["matcher"], str):
+        raise ValueError(f"{label} matcher must be a string")
+    if "sequential" in group and not isinstance(group["sequential"], bool):
+        raise ValueError(f"{label} sequential must be a boolean")
+    handlers = group.get("hooks")
+    if not isinstance(handlers, list):
+        raise ValueError(f"{label} must contain a hooks array")
+    group["hooks"] = [
+        _validate_handler(handler, event, index, handler_index)
+        for handler_index, handler in enumerate(handlers)
+    ]
+    return group
+
+
+def read_gemini_hooks(value: Any) -> GeminiHooks:
+    if not isinstance(value, dict):
+        raise ValueError('Gemini CLI settings field "hooks" must be an object')
+    hooks: GeminiHooks = {}
+    for event, groups in value.items():
+        if not isinstance(groups, list):
+            raise ValueError(
+                f'Gemini CLI settings field "hooks.{event}" must be an array'
+            )
+        hooks[event] = (
+            [
+                _validate_group(group, event, index)
+                for index, group in enumerate(groups)
+            ]
+            if event in KNOWN_EVENTS
+            else list(groups)
+        )
+    return hooks
+
+
+def read_gemini_hook_controls(value: Any) -> GeminiHookControls:
+    if not isinstance(value, dict):
+        raise ValueError(
+            'Gemini CLI settings field "hooksConfig" must be an object'
+        )
+    supported = {"enabled", "disabled", "notifications"}
+    extra = next((field for field in value if field not in supported), None)
+    if extra is not None:
+        raise ValueError(
+            "Gemini CLI settings field \"hooksConfig\" contains unsupported "
+            f'field "{extra}"'
+        )
+    if "enabled" in value and not isinstance(value["enabled"], bool):
+        raise ValueError(
+            'Gemini CLI settings field "hooksConfig.enabled" must be a boolean'
+        )
+    if "notifications" in value and not isinstance(
+        value["notifications"], bool
+    ):
+        raise ValueError(
+            "Gemini CLI settings field \"hooksConfig.notifications\" must be "
+            "a boolean"
+        )
+    disabled = value.get("disabled", [])
+    if not isinstance(disabled, list) or any(
+        not isinstance(item, str) for item in disabled
+    ):
+        raise ValueError(
+            "Gemini CLI settings field \"hooksConfig.disabled\" must be an "
+            "array of strings"
+        )
+    return GeminiHookControls(value.get("enabled") is not False, tuple(disabled))
+
+
+def managed_gemini_hooks_enabled(controls: GeminiHookControls) -> bool:
+    return (
+        controls.enabled
+        and GUARD_HOOK_NAME not in controls.disabled
+        and AUDIT_HOOK_NAME not in controls.disabled
+    )
+
+
+def disabled_managed_gemini_entries(
+    controls: GeminiHookControls,
+) -> Tuple[str, ...]:
+    return tuple(
+        entry
+        for entry in controls.disabled
+        if entry in {GUARD_HOOK_NAME, AUDIT_HOOK_NAME}
+        or gemini_runtime_reference(entry, GUARD_SCRIPT, True) is not None
+        or gemini_runtime_reference(entry, AUDIT_SCRIPT, True) is not None
+    )
+
+
+def build_gemini_group(script_path: str, name: str) -> JsonObject:
+    return {
+        "hooks": [{
+            "type": "command",
+            "name": name,
+            "command": build_gemini_command(script_path),
+            "timeout": HOOK_TIMEOUT_MILLISECONDS,
+        }]
+    }
+
+
+def _exact_managed_group(group: JsonObject) -> bool:
+    return set(group) == {"hooks"}
+
+
+def _current_managed_reference(
+    handler: JsonObject, script_name: str, hook_name: str
+) -> Optional[GeminiRuntimeReference]:
+    if (
+        set(handler) == {"command", "name", "timeout", "type"}
+        and handler.get("type") == "command"
+        and handler.get("name") == hook_name
+        and handler.get("timeout") == HOOK_TIMEOUT_MILLISECONDS
+        and isinstance(handler.get("command"), str)
+    ):
+        return gemini_runtime_reference(handler["command"], script_name)
+    return None
+
+
+def _legacy_managed_reference(
+    handler: JsonObject, script_name: str
+) -> Optional[GeminiRuntimeReference]:
+    if (
+        set(handler) == {"command", "type"}
+        and handler.get("type") == "command"
+        and isinstance(handler.get("command"), str)
+    ):
+        return gemini_runtime_reference(handler["command"], script_name, True)
+    return None
+
+
+def _managed_reference(
+    handler: JsonObject,
+    script_name: str,
+    hook_name: str,
+    include_legacy: bool,
+) -> Optional[GeminiRuntimeReference]:
+    current = _current_managed_reference(handler, script_name, hook_name)
+    if current is not None or not include_legacy:
+        return current
+    return _legacy_managed_reference(handler, script_name)
+
+
+def managed_gemini_removals(
+    hooks: GeminiHooks, agent_id: Optional[str] = None
+) -> List[ManagedRemoval]:
+    removals = []
+    for event, script_name, hook_name in (
+        ("BeforeTool", GUARD_SCRIPT, GUARD_HOOK_NAME),
+        ("AfterTool", AUDIT_SCRIPT, AUDIT_HOOK_NAME),
+    ):
+        for group_index, group in enumerate(hooks.get(event, [])):
+            handlers = group.get("hooks", [])
+            indexes = tuple(
+                index
+                for index, handler in enumerate(handlers)
+                if isinstance(handler, dict)
+                and (
+                    reference := _managed_reference(
+                        handler, script_name, hook_name, True
+                    )
+                )
+                is not None
+                and (
+                    agent_id is None
+                    or same_gemini_agent_id(reference.agent_id, agent_id)
+                )
+            )
+            if indexes:
+                removals.append(
+                    ManagedRemoval(
+                        event,
+                        group_index,
+                        indexes,
+                        _exact_managed_group(group)
+                        and len(indexes) == len(handlers),
+                    )
+                )
+    return removals
+
+
+def _references_for_event(
+    groups: List[JsonObject], script_name: str, hook_name: str
+) -> Dict[str, List[GeminiRuntimeReference]]:
+    result: Dict[str, List[GeminiRuntimeReference]] = {}
+    for group in groups:
+        if not _exact_managed_group(group):
+            continue
+        for handler in group["hooks"]:
+            reference = _current_managed_reference(
+                handler, script_name, hook_name
+            )
+            if reference is None:
+                continue
+            key = os.path.normcase(reference.agent_id)
+            result.setdefault(key, []).append(reference)
+    return result
+
+
+def gemini_runtime_contracts(
+    hooks: GeminiHooks,
+) -> List[GeminiRuntimeContract]:
+    guards = _references_for_event(
+        hooks.get("BeforeTool", []), GUARD_SCRIPT, GUARD_HOOK_NAME
+    )
+    audits = _references_for_event(
+        hooks.get("AfterTool", []), AUDIT_SCRIPT, AUDIT_HOOK_NAME
+    )
+    contracts = []
+    for key, guard in guards.items():
+        audit = audits.get(key, [])
+        if len(guard) != 1 or len(audit) != 1:
+            continue
+        contracts.append(
+            GeminiRuntimeContract(
+                guard[0].agent_id,
+                guard[0].script_path,
+                audit[0].script_path,
+            )
+        )
+    return contracts

sdks/python/elydora/plugins/gemini_installation.py

@@ -0,0 +1,161 @@
+"""Transactional Gemini CLI 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 .base import InstallConfig
+from .gemini_command import same_gemini_path
+from .gemini_config import GeminiDocument, RenderedGeminiDocument
+from .gemini_contract import (
+    AGENT_KEY,
+    AUDIT_SCRIPT,
+    GUARD_SCRIPT,
+    elydora_dir,
+)
+from .gemini_io import (
+    rendered_change,
+    validate_api_origin,
+    validate_private_key,
+    validate_runtime_tree,
+)
+from .hook_template import generate_guard_script, generate_hook_script
+
+
+@dataclass(frozen=True)
+class GeminiRuntimePaths:
+    agent_id: str
+    agent_directory: str
+    guard_path: str
+    audit_path: str
+
+
+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) -> GeminiRuntimePaths:
+    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_gemini_path(config.get("guard_script_path", ""), guard_path):
+        raise ValueError(
+            "Elydora guard runtime must use the managed agent directory: "
+            f"{guard_path}"
+        )
+    return GeminiRuntimePaths(agent_id, agent_directory, guard_path, 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"Gemini 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_gemini_installation(
+    config: InstallConfig, document: GeminiDocument
+) -> GeminiRuntimePaths:
+    if not document.file_path:
+        raise ValueError("Gemini 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_gemini_installation(
+    config: InstallConfig,
+    paths: GeminiRuntimePaths,
+    rendered: RenderedGeminiDocument,
+) -> List[FileChange]:
+    guard_script = generate_guard_script(
+        AGENT_KEY,
+        paths.agent_id,
+        success_output="{}\n",
+    )
+    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"),
+        success_output="{}\n",
+        native_payload=True,
+        agent_name=AGENT_KEY,
+    )
+    return _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
+        ),
+        rendered_change(rendered),
+    ])
+
+
+def commit_gemini_installation(changes: List[FileChange]) -> None:
+    write_changes(changes, "Install Gemini CLI hooks")

sdks/python/elydora/plugins/gemini_io.py

@@ -0,0 +1,250 @@
+"""Gemini CLI settings discovery and strict managed runtime inspection."""
+
+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,
+    MAX_SOURCE_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 .gemini_command import same_gemini_agent_id, same_gemini_path
+from .gemini_config import (
+    GeminiDocument,
+    RenderedGeminiDocument,
+    create_gemini_document,
+    parse_gemini_document,
+)
+from .gemini_contract import (
+    AGENT_KEY,
+    CONFIG_FILE,
+    GeminiRuntimeContract,
+    elydora_dir,
+)
+
+
+def gemini_configuration_directory() -> str:
+    configured_home = os.environ.get("GEMINI_CLI_HOME")
+    home = configured_home if configured_home else os.path.expanduser("~")
+    return os.path.join(home, ".gemini")
+
+
+def gemini_settings_path() -> str:
+    return os.path.join(gemini_configuration_directory(), CONFIG_FILE)
+
+
+def read_gemini_document() -> GeminiDocument:
+    file_path = gemini_settings_path()
+    physical_directory_exists(
+        os.path.dirname(file_path), "Gemini CLI configuration directory"
+    )
+    snapshot = read_physical_file(
+        file_path, "Gemini CLI user settings", MAX_SOURCE_BYTES
+    )
+    if snapshot is None:
+        return create_gemini_document(file_path)
+    return parse_gemini_document(
+        exists=True,
+        file_path=file_path,
+        raw=snapshot.contents,
+    )
+
+
+def rendered_change(
+    rendered: RenderedGeminiDocument,
+) -> Optional[FileChange]:
+    if not rendered.changed:
+        return None
+    return source_change(
+        rendered.document.file_path,
+        "Gemini CLI user settings",
+        rendered.document.raw if rendered.document.exists else None,
+        rendered.next_source,
+        0o600,
+        MAX_SOURCE_BYTES,
+    )
+
+
+def write_gemini_document(rendered: RenderedGeminiDocument) -> None:
+    change = rendered_change(rendered)
+    if change is not None:
+        write_changes([change], "Uninstall Gemini 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_gemini_agent_id(agent_id, expected_agent_id)
+        or config.get("agent_name") != AGENT_KEY
+    ):
+        raise ValueError(
+            "Elydora runtime identity does not match Gemini CLI hooks: "
+            f"{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.py", "Elydora guard runtime"),
+        ("hook.py", "Elydora audit runtime"),
+        ("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_gemini_agent_id(configured_id, agent_id)
+        or config.get("agent_name") != AGENT_KEY
+    ):
+        raise ValueError(
+            "Elydora runtime config identity does not match Gemini CLI agent "
+            f"{agent_id}: {config_path}"
+        )
+
+
+def _runtime_contract_exists(contract: GeminiRuntimeContract) -> bool:
+    root = elydora_dir()
+    agent_directory = os.path.dirname(contract.guard_path)
+    if (
+        not same_gemini_path(os.path.dirname(agent_directory), root)
+        or not same_gemini_path(
+            contract.audit_path, os.path.join(agent_directory, "hook.py")
+        )
+    ):
+        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")
+    if config is None or key is None or guard is None or audit is None:
+        return False
+    _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)
+
+
+def gemini_runtime_files_exist(
+    contracts: List[GeminiRuntimeContract],
+) -> bool:
+    return any(_runtime_contract_exists(contract) for contract in contracts)

sdks/python/elydora/plugins/registry.py

@@ -43,8 +43,8 @@ SUPPORTED_AGENTS: Dict[str, Dict[str, str]] = {
     },
     "gemini": {
         "name": "Gemini CLI",
-        "hook_event": "AfterTool",
+        "hook_event": "BeforeTool/AfterTool",
-        "config_path": "~/.gemini/settings.json",
+        "config_path": "$GEMINI_CLI_HOME/.gemini/settings.json",
     },
     "grok": {
         "name": "Grok Build",

sdks/python/tests/gemini_support.py

@@ -0,0 +1,279 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+import json
+import ntpath
+import os
+from pathlib import Path
+import subprocess  # nosec B404
+import sys
+from threading import Thread
+from typing import Any, Mapping
+
+import pytest
+
+from elydora.plugins._jsonc import parse_jsonc
+from elydora.plugins.gemini import GeminiPlugin
+from elydora.utils import base64url_encode
+
+
+AGENT_ID = "agent-1"
+MISSING = object()
+VALID_PRIVATE_KEY = base64url_encode(bytes(range(32)))
+
+
+@dataclass
+class GeminiFixture:
+    root_dir: Path
+    home_dir: Path
+    project_dir: Path
+    config_path: Path
+    agent_dir: Path
+    guard_path: Path
+    hook_path: Path
+    runtime_config_path: Path
+    private_key_path: Path
+    plugin: GeminiPlugin
+    config: dict[str, str]
+
+    def install(self) -> None:
+        self.plugin.install(self.config)
+
+    def source(self) -> str:
+        with self.config_path.open("r", encoding="utf-8", newline="") as file:
+            return file.read()
+
+    def settings(self) -> dict[str, Any]:
+        value = parse_jsonc(
+            self.source(),
+            f"Gemini CLI test settings at {self.config_path}",
+            allow_trailing_commas=False,
+        )
+        assert isinstance(value, dict)
+        return value
+
+
+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", newline="")
+    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 _settings_root(
+    home_dir: Path, project_dir: Path, override: object
+) -> Path:
+    if override is MISSING or override == "":
+        return home_dir
+    configured = str(home_dir) if override is None else str(override)
+    value = Path(configured)
+    return value if value.is_absolute() else project_dir / value
+
+
+def prepare_fixture(
+    monkeypatch: pytest.MonkeyPatch,
+    tmp_path: Path,
+    *,
+    existing_settings: object = MISSING,
+    base_url: str = "http://127.0.0.1:9",
+    gemini_home_override: object = None,
+) -> GeminiFixture:
+    home_dir = tmp_path / "home with spaces and 'quote $GEMINI_CWD %GEMINI_CWD%"
+    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)
+    if gemini_home_override is MISSING:
+        monkeypatch.delenv("GEMINI_CLI_HOME", raising=False)
+    else:
+        configured = (
+            str(home_dir)
+            if gemini_home_override is None
+            else str(gemini_home_override)
+        )
+        monkeypatch.setenv("GEMINI_CLI_HOME", configured)
+    config_path = (
+        _settings_root(home_dir, project_dir, gemini_home_override)
+        / ".gemini"
+        / "settings.json"
+    )
+    agent_dir = home_dir / ".elydora" / AGENT_ID
+    guard_path = agent_dir / "guard.py"
+    hook_path = agent_dir / "hook.py"
+    runtime_config_path = agent_dir / "config.json"
+    private_key_path = agent_dir / "private.key"
+    if existing_settings is not MISSING:
+        write_json(config_path, existing_settings)
+    return GeminiFixture(
+        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,
+        runtime_config_path=runtime_config_path,
+        private_key_path=private_key_path,
+        plugin=GeminiPlugin(),
+        config={
+            "org_id": "org-1",
+            "agent_id": AGENT_ID,
+            "agent_name": "gemini",
+            "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, name: str
+) -> dict[str, Any]:
+    for group in settings["hooks"][event]:
+        if set(group) != {"hooks"}:
+            continue
+        for handler in group["hooks"]:
+            if (
+                set(handler) == {"type", "name", "command", "timeout"}
+                and handler.get("type") == "command"
+                and handler.get("name") == name
+                and handler.get("timeout") == 10_000
+            ):
+                return handler
+    raise AssertionError(f"managed handler for {event!r} not found")
+
+
+def assert_managed_handler(
+    handler: Mapping[str, Any], name: str
+) -> None:
+    assert set(handler) == {"type", "name", "command", "timeout"}
+    assert handler["type"] == "command"
+    assert handler["name"] == name
+    assert handler["timeout"] == 10_000
+    assert isinstance(handler["command"], str)
+
+
+def run_handler(
+    handler: Mapping[str, Any], payload: bytes, fixture: GeminiFixture
+) -> subprocess.CompletedProcess[bytes]:
+    environment = {
+        **os.environ,
+        "HOME": str(fixture.home_dir),
+        "USERPROFILE": str(fixture.home_dir),
+        "GEMINI_CLI_HOME": str(fixture.home_dir),
+    }
+    if os.name == "nt":
+        powershell = ntpath.join(
+            os.environ.get("SystemRoot", r"C:\Windows"),
+            "System32",
+            "WindowsPowerShell",
+            "v1.0",
+            "powershell.exe",
+        )
+        source = (
+            f'{handler["command"]}; '
+            "if ($LASTEXITCODE -ne 0) { exit $LASTEXITCODE }"
+        )
+        command = [
+            powershell,
+            "-NoLogo",
+            "-NoProfile",
+            "-NonInteractive",
+            "-Command",
+            source,
+        ]
+    else:
+        command = ["/bin/bash", "-c", str(handler["command"])]
+    return subprocess.run(  # nosec B603
+        command,
+        input=payload,
+        capture_output=True,
+        check=False,
+        cwd=fixture.project_dir,
+        env=environment,
+    )
+
+
+def legacy_handler(script_path: Path) -> dict[str, str]:
+    command = (
+        f'"{sys.executable}" {script_path}'
+        if script_path.name == "guard.py"
+        else str(script_path)
+    )
+    return {"type": "command", "command": command}
+
+
+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_cli_security.py

@@ -17,7 +17,6 @@ from elydora.plugins._file_io import write_text_atomic
 from elydora.plugins import (
     copilot,
     cursor,
-    gemini,
     kiroide,
     letta,
     opencode,
@@ -375,7 +374,6 @@ def test_atomic_secret_write_preserves_existing_file_on_replace_failure(
     [
         (copilot, copilot.CopilotPlugin, "copilot"),
         (cursor, cursor.CursorPlugin, "cursor"),
-        (gemini, gemini.GeminiPlugin, "gemini"),
         (kiroide, kiroide.KiroIdePlugin, "kiroide"),
         (letta, letta.LettaPlugin, "letta"),
         (opencode, opencode.OpenCodePlugin, "opencode"),

sdks/python/tests/test_gemini_official.py

@@ -0,0 +1,64 @@
+from __future__ import annotations
+
+import os
+from pathlib import Path
+import shutil
+import subprocess  # nosec B404
+
+import pytest
+
+from gemini_support import prepare_fixture
+
+
+GEMINI_ENTRY = os.environ.get("ELYDORA_GEMINI_ENTRY")
+
+
+@pytest.mark.skipif(
+    not GEMINI_ENTRY,
+    reason="set ELYDORA_GEMINI_ENTRY to the official Gemini CLI entry file",
+)
+def test_official_gemini_cli_accepts_installed_user_hooks(
+    monkeypatch: pytest.MonkeyPatch, tmp_path: Path
+) -> None:
+    fixture = prepare_fixture(monkeypatch, tmp_path)
+    fixture.install()
+    node = shutil.which("node")
+    assert node is not None
+    environment = {
+        **os.environ,
+        "HOME": str(fixture.home_dir),
+        "USERPROFILE": str(fixture.home_dir),
+        "GEMINI_CLI_HOME": str(fixture.home_dir),
+        "GEMINI_API_KEY": "official-loader-test-key",
+        "GEMINI_TELEMETRY_ENABLED": "false",
+        "OTEL_SDK_DISABLED": "true",
+    }
+
+    version = subprocess.run(  # nosec B603
+        [node, str(GEMINI_ENTRY), "--version"],
+        capture_output=True,
+        check=False,
+        cwd=fixture.project_dir,
+        env=environment,
+        text=True,
+    )
+    loaded = subprocess.run(  # nosec B603
+        [node, str(GEMINI_ENTRY), "--skip-trust", "--list-extensions"],
+        capture_output=True,
+        check=False,
+        cwd=fixture.project_dir,
+        env=environment,
+        text=True,
+    )
+
+    assert version.returncode == 0, version.stderr
+    assert "0.51.0" in version.stdout
+    assert loaded.returncode == 0, loaded.stderr
+    output = (loaded.stdout + "\n" + loaded.stderr).lower()
+    for message in (
+        "invalid settings",
+        "settings validation",
+        "failed to parse",
+        "hook configuration error",
+    ):
+        assert message not in output

sdks/python/tests/test_gemini_plugin.py

@@ -0,0 +1,363 @@
+from __future__ import annotations
+
+import json
+import os
+from pathlib import Path
+import subprocess  # nosec B404
+import sys
+
+import pytest
+
+from gemini_support import (
+    AGENT_ID,
+    MISSING,
+    VALID_PRIVATE_KEY,
+    assert_managed_handler,
+    legacy_handler,
+    managed_handler,
+    prepare_fixture,
+    write_json,
+    write_text,
+)
+from elydora.plugins.registry import SUPPORTED_AGENTS
+
+
+GUARD_NAME = "elydora-guard"
+AUDIT_NAME = "elydora-audit"
+
+
+def assert_managed_pair(settings: dict, fixture: object) -> None:
+    guard = managed_handler(settings, "BeforeTool", GUARD_NAME)
+    audit = managed_handler(settings, "AfterTool", AUDIT_NAME)
+    assert_managed_handler(guard, GUARD_NAME)
+    assert_managed_handler(audit, AUDIT_NAME)
+    assert str(fixture.guard_path) not in ("", guard["command"])
+    assert str(fixture.hook_path) not in ("", audit["command"])
+
+
+def test_gemini_is_registered_with_the_official_contract() -> None:
+    assert SUPPORTED_AGENTS["gemini"] == {
+        "name": "Gemini CLI",
+        "hook_event": "BeforeTool/AfterTool",
+        "config_path": "$GEMINI_CLI_HOME/.gemini/settings.json",
+    }
+
+
+def test_install_preserves_jsonc_and_writes_one_exact_pair(
+    monkeypatch: pytest.MonkeyPatch,
+    tmp_path: Path,
+    capsys: pytest.CaptureFixture[str],
+) -> None:
+    existing = "\r\n".join([
+        "{",
+        "  // Keep this user preference.",
+        '  "theme": "GitHub",',
+        '  "hooks": {',
+        '    "FutureEvent": [null],',
+        '    "BeforeTool": [{ "matcher": "read_file", "hooks": '
+        '[{ "type": "command", "command": "user-hook" }] }]',
+        "  }",
+        "}",
+        "",
+    ])
+    fixture = prepare_fixture(
+        monkeypatch, tmp_path, existing_settings=existing
+    )
+    project_settings = fixture.project_dir / ".gemini" / "settings.json"
+    system_settings = tmp_path / "system-settings.json"
+    write_text(project_settings, '{ "owner": "project" }\n')
+    write_text(system_settings, '{ "owner": "system" }\n')
+    monkeypatch.setenv("GEMINI_CLI_SYSTEM_SETTINGS_PATH", str(system_settings))
+
+    fixture.install()
+    output = capsys.readouterr().out
+    assert "run /hooks list" in output
+    installed = fixture.settings()
+    assert "Keep this user preference" in fixture.source()
+    assert "\r\n" in fixture.source()
+    assert installed["theme"] == "GitHub"
+    assert installed["hooks"]["FutureEvent"] == [None]
+    assert installed["hooks"]["BeforeTool"][0]["hooks"][0]["command"] == "user-hook"
+    assert_managed_pair(installed, fixture)
+    first_source = fixture.source()
+
+    fixture.install()
+
+    assert fixture.source() == first_source
+    assert project_settings.read_text(encoding="utf-8") == '{ "owner": "project" }\n'
+    assert system_settings.read_text(encoding="utf-8") == '{ "owner": "system" }\n'
+
+
+@pytest.mark.parametrize(
+    ("override", "kind"),
+    [
+        (MISSING, "default"),
+        ("", "empty"),
+        ("relative gemini", "relative"),
+        ("~", "literal_tilde"),
+    ],
+)
+def test_gemini_home_matches_official_path_resolution(
+    monkeypatch: pytest.MonkeyPatch,
+    tmp_path: Path,
+    override: object,
+    kind: str,
+) -> None:
+    fixture = prepare_fixture(
+        monkeypatch, tmp_path, gemini_home_override=override
+    )
+
+    fixture.install()
+
+    expected = {
+        "default": fixture.home_dir / ".gemini" / "settings.json",
+        "empty": fixture.home_dir / ".gemini" / "settings.json",
+        "relative": fixture.project_dir / "relative gemini" / ".gemini" / "settings.json",
+        "literal_tilde": fixture.project_dir / "~" / ".gemini" / "settings.json",
+    }[kind]
+    assert fixture.config_path == expected
+    assert expected.is_file()
+
+
+INVALID_SETTINGS = [
+    ("{ malformed", "parse Gemini CLI user settings"),
+    ("[]", "must contain a JSON object"),
+    ('{ "theme": true, }', "trailing commas are not allowed"),
+    ('{ "hooks": {}, "hooks": {} }', 'duplicate field "hooks"'),
+    ('{ "hooks": null }', 'field "hooks" must be an object'),
+    ('{ "hooks": { "BeforeTool": null } }', "must be an array"),
+    ('{ "hooks": { "BeforeTool": [null] } }', "group.*must be an object"),
+    (
+        '{ "hooks": { "BeforeTool": [{ "matcher": 1, "hooks": [] }] } }',
+        "matcher must be a string",
+    ),
+    (
+        '{ "hooks": { "BeforeTool": [{ "sequential": 1, "hooks": [] }] } }',
+        "sequential must be a boolean",
+    ),
+    ('{ "hooks": { "BeforeTool": [{}] } }', "hooks array"),
+    ('{ "hooks": { "BeforeTool": [{ "hooks": [null] }] } }', "handler"),
+    (
+        '{ "hooks": { "BeforeTool": [{ "hooks": [{ "type": "http" }] }] } }',
+        "unsupported type",
+    ),
+    (
+        '{ "hooks": { "BeforeTool": [{ "hooks": [{ "type": "command", '
+        '"command": "" }] }] } }',
+        "non-empty command",
+    ),
+    (
+        '{ "hooks": { "BeforeTool": [{ "hooks": [{ "type": "command", '
+        '"command": "x", "timeout": -1 }] }] } }',
+        "non-negative finite number",
+    ),
+    (
+        '{ "hooks": { "BeforeTool": [{ "hooks": [{ "type": "command", '
+        '"command": "x", "env": { "A": 1 } }] }] } }',
+        "env must map names to strings",
+    ),
+    ('{ "hooksConfig": null }', "hooksConfig.*must be an object"),
+    ('{ "hooksConfig": { "future": true } }', 'unsupported field "future"'),
+    ('{ "hooksConfig": { "enabled": "yes" } }', "enabled.*must be a boolean"),
+    ('{ "hooksConfig": { "disabled": [1] } }', "array of strings"),
+    ('{ "hooksConfig": { "notifications": 1 } }', "notifications.*boolean"),
+]
+
+
+@pytest.mark.parametrize(("source", "pattern"), INVALID_SETTINGS)
+def test_invalid_official_settings_fail_before_every_write(
+    monkeypatch: pytest.MonkeyPatch,
+    tmp_path: Path,
+    source: str,
+    pattern: str,
+) -> None:
+    fixture = prepare_fixture(
+        monkeypatch, tmp_path, existing_settings=source
+    )
+
+    with pytest.raises(ValueError, match=pattern):
+        fixture.install()
+
+    assert fixture.source() == source
+    assert not fixture.agent_dir.exists()
+
+
+@pytest.mark.parametrize(
+    ("settings", "pattern"),
+    [
+        ({"hooksConfig": {"enabled": False}}, "hooksConfig.enabled"),
+        ({"hooksConfig": {"disabled": [GUARD_NAME]}}, GUARD_NAME),
+        ({"hooksConfig": {"disabled": [AUDIT_NAME]}}, AUDIT_NAME),
+    ],
+)
+def test_canonical_hook_controls_block_installation(
+    monkeypatch: pytest.MonkeyPatch,
+    tmp_path: Path,
+    settings: dict,
+    pattern: str,
+) -> None:
+    fixture = prepare_fixture(
+        monkeypatch, tmp_path, existing_settings=settings
+    )
+    with pytest.raises(ValueError, match=pattern):
+        fixture.install()
+    assert not fixture.agent_dir.exists()
+
+
+def test_disabled_legacy_command_blocks_installation(
+    monkeypatch: pytest.MonkeyPatch, tmp_path: Path
+) -> None:
+    fixture = prepare_fixture(monkeypatch, tmp_path)
+    write_json(fixture.config_path, {
+        "hooksConfig": {
+            "disabled": [legacy_handler(fixture.guard_path)["command"]]
+        }
+    })
+
+    with pytest.raises(ValueError, match="hooksConfig.disabled"):
+        fixture.install()
+
+    assert not fixture.agent_dir.exists()
+
+
+def test_exact_legacy_hooks_migrate_and_lookalikes_survive_uninstall(
+    monkeypatch: pytest.MonkeyPatch, tmp_path: Path
+) -> None:
+    fixture = prepare_fixture(monkeypatch, tmp_path)
+    lookalike = legacy_handler(fixture.guard_path)
+    lookalike["command"] += " --inspect"
+    write_json(fixture.config_path, {
+        "hooks": {
+            "BeforeTool": [
+                {"hooks": [legacy_handler(fixture.guard_path)]},
+                {"hooks": [lookalike]},
+            ],
+            "AfterTool": [{"hooks": [legacy_handler(fixture.hook_path)]}],
+        }
+    })
+
+    fixture.install()
+    settings = fixture.settings()
+    assert_managed_pair(settings, fixture)
+    assert any(
+        group["hooks"][0].get("command") == lookalike["command"]
+        for group in settings["hooks"]["BeforeTool"]
+    )
+    settings["hooks"]["BeforeTool"][-1]["hooks"].append({
+        "type": "command", "command": "user-command"
+    })
+    write_json(fixture.config_path, settings)
+
+    fixture.plugin.uninstall(AGENT_ID)
+
+    remaining = fixture.settings()["hooks"]
+    assert remaining["BeforeTool"] == [
+        {"hooks": [lookalike]},
+        {"hooks": [{"type": "command", "command": "user-command"}]},
+    ]
+    assert "AfterTool" not in remaining
+
+
+def test_uninstall_preserves_user_settings_and_removes_owned_file(
+    monkeypatch: pytest.MonkeyPatch, tmp_path: Path
+) -> None:
+    user = prepare_fixture(
+        monkeypatch,
+        tmp_path / "user",
+        existing_settings={"theme": "GitHub", "hooks": {"Notification": []}},
+    )
+    user.install()
+    user.plugin.uninstall(AGENT_ID)
+    assert user.settings() == {
+        "theme": "GitHub", "hooks": {"Notification": []}
+    }
+
+    owned = prepare_fixture(monkeypatch, tmp_path / "owned")
+    owned.install()
+    owned.plugin.uninstall(AGENT_ID)
+    assert not owned.config_path.exists()
+
+
+def test_status_requires_enabled_unique_pair_and_strict_runtime(
+    monkeypatch: pytest.MonkeyPatch, tmp_path: Path
+) -> None:
+    fixture = prepare_fixture(monkeypatch, tmp_path)
+    fixture.install()
+    assert fixture.plugin.status()["installed"] is True
+
+    settings = fixture.settings()
+    settings["hooks"]["AfterTool"].append(settings["hooks"]["AfterTool"][-1])
+    write_json(fixture.config_path, settings)
+    assert fixture.plugin.status()["installed"] is False
+
+    fixture.install()
+    settings = fixture.settings()
+    settings["hooksConfig"] = {"disabled": [AUDIT_NAME]}
+    write_json(fixture.config_path, settings)
+    assert fixture.plugin.status()["installed"] is False
+
+    settings["hooksConfig"] = {"disabled": []}
+    write_json(fixture.config_path, settings)
+    write_text(fixture.private_key_path, "invalid")
+    with pytest.raises(ValueError, match="canonical 32-byte"):
+        fixture.plugin.status()
+
+
+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),
+        "GEMINI_CLI_HOME": str(fixture.home_dir),
+    }
+    root = Path(__file__).parents[1]
+    base = [sys.executable, "-m", "elydora.cli"]
+
+    install = subprocess.run(  # nosec B603
+        [
+            *base, "install", "--agent", "gemini",
+            "--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 "run /hooks list" in install.stdout
+    assert_managed_pair(fixture.settings(), fixture)
+
+    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 "Gemini CLI" in status.stdout
+    assert "[installed]" in status.stdout
+
+    uninstall = subprocess.run(  # nosec B603
+        [*base, "uninstall", "--agent", "gemini", "--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()

sdks/python/tests/test_gemini_runtime.py

@@ -0,0 +1,165 @@
+from __future__ import annotations
+
+import json
+import os
+from pathlib import Path
+
+import pytest
+
+from gemini_support import (
+    ElydoraApiHandler,
+    managed_handler,
+    prepare_fixture,
+    run_handler,
+    server_base_url,
+    start_api_server,
+)
+
+
+GUARD_NAME = "elydora-guard"
+AUDIT_NAME = "elydora-audit"
+
+
+def official_payload(event: str, **overrides: object) -> dict[str, object]:
+    value: dict[str, object] = {
+        "session_id": "session-1",
+        "transcript_path": "/tmp/session-1.jsonl",
+        "cwd": "/tmp/project",
+        "hook_event_name": event,
+        "timestamp": "2026-07-19T00:00:00.000Z",
+        "tool_name": "run_shell_command",
+        "tool_input": {"command": "pytest"},
+    }
+    if event == "AfterTool":
+        value["tool_response"] = {"output": "passed", "error": None}
+    value.update(overrides)
+    return value
+
+
+def encoded(value: object) -> bytes:
+    return json.dumps(value, separators=(",", ":")).encode()
+
+
+def test_guard_accepts_active_agents_with_valid_json_stdout(
+    monkeypatch: pytest.MonkeyPatch, tmp_path: Path
+) -> None:
+    server = start_api_server()
+    try:
+        fixture = prepare_fixture(
+            monkeypatch, tmp_path, base_url=server_base_url(server)
+        )
+        fixture.install()
+        result = run_handler(
+            managed_handler(fixture.settings(), "BeforeTool", GUARD_NAME),
+            encoded(official_payload("BeforeTool")),
+            fixture,
+        )
+    finally:
+        server.shutdown()
+        server.server_close()
+
+    assert result.returncode == 0, result.stderr
+    assert json.loads(result.stdout) == {}
+    assert [request["method"] for request in ElydoraApiHandler.requests] == ["GET"]
+
+
+@pytest.mark.parametrize("status", ["frozen", "revoked"])
+def test_guard_propagates_official_exit_code_two(
+    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()
+        result = run_handler(
+            managed_handler(fixture.settings(), "BeforeTool", GUARD_NAME),
+            encoded(official_payload("BeforeTool")),
+            fixture,
+        )
+    finally:
+        server.shutdown()
+        server.server_close()
+
+    assert result.returncode == 2
+    assert result.stdout == b""
+    assert status.encode() in result.stderr.lower()
+
+
+def test_audit_preserves_complete_native_after_tool_payload(
+    monkeypatch: pytest.MonkeyPatch, tmp_path: Path
+) -> None:
+    server = start_api_server()
+    try:
+        fixture = prepare_fixture(
+            monkeypatch, tmp_path, base_url=server_base_url(server)
+        )
+        fixture.install()
+        payload = official_payload(
+            "AfterTool",
+            mcp_context={"server_name": "filesystem", "tool_name": "write_file"},
+            original_request_name="write_file",
+            future_provider_field={"preserved": True},
+        )
+        result = run_handler(
+            managed_handler(fixture.settings(), "AfterTool", AUDIT_NAME),
+            encoded(payload),
+            fixture,
+        )
+    finally:
+        server.shutdown()
+        server.server_close()
+
+    assert result.returncode == 0, result.stderr
+    assert json.loads(result.stdout) == {}
+    operation = next(
+        request["json"]
+        for request in ElydoraApiHandler.requests
+        if request["method"] == "POST"
+    )
+    assert operation["payload"] == payload
+    assert operation["subject"] == {"session_id": "session-1"}
+    assert operation["action"] == {"tool": "run_shell_command"}
+
+
+def test_runtime_failures_stay_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, "BeforeTool", GUARD_NAME)
+    audit = managed_handler(settings, "AfterTool", AUDIT_NAME)
+
+    invalid_guard = run_handler(guard, b"{ malformed", fixture)
+    audit_result = run_handler(
+        audit, encoded(official_payload("AfterTool")), fixture
+    )
+
+    assert invalid_guard.returncode == 0
+    assert json.loads(invalid_guard.stdout) == {}
+    assert b"invalid JSON" in invalid_guard.stderr
+    assert audit_result.returncode == 0
+    assert json.loads(audit_result.stdout) == {}
+    log = (fixture.agent_dir / "error.log").read_text(encoding="utf-8")
+    assert "[elydora-hook]" in log
+    assert "refused" in log.lower() or "urlerror" in log.lower()
+
+
+@pytest.mark.skipif(os.name == "nt", reason="POSIX mode bits apply on POSIX")
+def test_runtime_artifacts_use_private_modes(
+    monkeypatch: pytest.MonkeyPatch, tmp_path: Path
+) -> None:
+    fixture = prepare_fixture(monkeypatch, tmp_path)
+    fixture.install()
+    for path, expected in (
+        (fixture.runtime_config_path, 0o600),
+        (fixture.private_key_path, 0o600),
+        (fixture.guard_path, 0o700),
+        (fixture.hook_path, 0o700),
+        (fixture.config_path, 0o600),
+    ):
+        assert path.stat().st_mode & 0o777 == expected

sdks/python/tests/test_gemini_transaction.py

@@ -0,0 +1,254 @@
+from __future__ import annotations
+
+import os
+from pathlib import Path
+import shutil
+import subprocess  # nosec B404
+import sys
+
+import pytest
+
+from gemini_support import (
+    AGENT_ID,
+    VALID_PRIVATE_KEY,
+    prepare_fixture,
+    symlink_or_skip,
+    write_json,
+    write_text,
+)
+from elydora.plugins import _transaction, gemini_installation
+from elydora.plugins.gemini_config import render_gemini_document
+from elydora.plugins.gemini_contract import (
+    AUDIT_HOOK_NAME,
+    GUARD_HOOK_NAME,
+    build_gemini_group,
+)
+from elydora.plugins.gemini_io import read_gemini_document
+
+
+RUNTIME_NAMES = ("guard.py", "config.json", "private.key", "hook.py")
+
+
+def prepare_installation(fixture: object) -> list[_transaction.FileChange]:
+    document = read_gemini_document()
+    paths = gemini_installation.preflight_gemini_installation(
+        fixture.config, document
+    )
+    rendered = render_gemini_document(
+        document,
+        None,
+        {
+            "BeforeTool": build_gemini_group(
+                paths.guard_path, GUARD_HOOK_NAME
+            ),
+            "AfterTool": build_gemini_group(
+                paths.audit_path, AUDIT_HOOK_NAME
+            ),
+        },
+    )
+    return gemini_installation.prepare_gemini_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_five_files(
+    monkeypatch: pytest.MonkeyPatch, tmp_path: Path
+) -> None:
+    original = '{"theme":"GitHub"}\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 Gemini settings failure")
+        real_replace(source, destination)
+
+    monkeypatch.setattr(_transaction.os, "replace", fail_settings_once)
+
+    with pytest.raises(OSError, match="injected Gemini settings failure"):
+        fixture.install()
+
+    assert fixture.source() == 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_commit(
+    monkeypatch: pytest.MonkeyPatch, tmp_path: Path
+) -> None:
+    fixture = prepare_fixture(
+        monkeypatch, tmp_path, existing_settings={"theme": "GitHub"}
+    )
+    changes = prepare_installation(fixture)
+    concurrent = '{"theme":"Atom One","hooks":{"Notification":[]}}\n'
+    write_text(fixture.config_path, concurrent)
+
+    with pytest.raises(OSError, match="changed during installation"):
+        gemini_installation.commit_gemini_installation(changes)
+
+    assert fixture.source() == concurrent
+    for name in RUNTIME_NAMES:
+        assert not (fixture.agent_dir / name).exists()
+    assert_no_transaction_files(fixture.home_dir)
+
+
+@pytest.mark.parametrize(
+    "runtime_name",
+    [
+        "private.key",
+        "guard.py",
+        "hook.py",
+        "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": "gemini"
+    })
+    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"])
+def test_linked_configuration_and_runtime_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)
+    else:
+        target.mkdir()
+        link = (
+            fixture.config_path.parent
+            if kind == "configuration"
+            else fixture.agent_dir.parent
+        )
+        if kind == "runtime" and link.exists():
+            shutil.rmtree(link)
+        link.parent.mkdir(parents=True, exist_ok=True)
+        symlink_or_skip(target, link, directory=True)
+
+    with pytest.raises(OSError, match="physical (file|directory)"):
+        fixture.install()
+
+
+@pytest.mark.parametrize(
+    ("field", "value", "pattern"),
+    [
+        ("agent_name", "codex", "requires agent_name gemini"),
+        ("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_status_surfaces_malformed_runtime_metadata(
+    monkeypatch: pytest.MonkeyPatch, tmp_path: Path
+) -> None:
+    fixture = prepare_fixture(monkeypatch, tmp_path)
+    fixture.install()
+    write_text(fixture.runtime_config_path, "{ malformed")
+
+    with pytest.raises(ValueError, match="parse Elydora runtime config"):
+        fixture.plugin.status()
+
+
+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 / ".gemini" / "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),
+        "GEMINI_CLI_HOME": str(home_dir),
+    }
+
+    result = subprocess.run(  # nosec B603
+        [
+            sys.executable, "-m", "elydora.cli", "install",
+            "--agent", "gemini", "--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 Gemini CLI user settings" in result.stderr
+    assert settings_path.read_text(encoding="utf-8") == "{ malformed"
+    assert not (home_dir / ".elydora").exists()

sdks/python/tests/test_settings_hook_detection.py

@@ -1,22 +1,15 @@
 from __future__ import annotations
 
-from collections.abc import Callable
-
-import pytest
-
-from elydora.plugins.gemini import _is_elydora_hook as is_gemini_hook
 from elydora.plugins.letta import _is_elydora_hook as is_letta_hook
 
 
-HookDetector = Callable[[dict, str], bool]
+def test_hook_detection_handles_only_string_commands() -> None:
-
+    assert is_letta_hook({"command": 42}, "") is False
-
+    assert is_letta_hook({"hooks": [{"command": None}]}, "") is False
-@pytest.mark.parametrize("detector", [is_gemini_hook, is_letta_hook])
+    assert is_letta_hook(
-def test_hook_detection_handles_only_string_commands(detector: HookDetector) -> None:
+        {"hooks": [{"command": "run .elydora/agent-1/hook.py"}]}, ""
-    assert detector({"command": 42}, "") is False
+    ) is True
-    assert detector({"hooks": [{"command": None}]}, "") is False
+    assert is_letta_hook(
-    assert detector({"hooks": [{"command": "run .elydora/agent-1/hook.py"}]}, "") is True
-    assert detector(
         {"hooks": [{"command": "run .elydora/agent-1/hook.py"}]},
         "agent-2",
     ) is False