fix(qwen): mirror Qwen Code 0.20 Python hooks

sdks/python/README.md

@@ -98,7 +98,7 @@ Auggie installation commits `~/.augment/settings.json`, both platform wrappers,
 
 Factory Droid 0.175.0 installation resolves one active user hook source in this order: `~/.factory/hooks.json`, the legacy `~/.factory/hooks/hooks.json`, then the `hooks` field from user `settings.local.json` and `settings.json`. Current hook files use the documented top-level `hooks` container; legacy direct event maps remain readable until Factory migrates them. Elydora writes exact `PreToolUse` and `PostToolUse` groups to that source, removes exact Elydora ownership from inactive user sources, and keeps project, folder, plugin, and organization hook definitions read-only. System, project, folder, and user `hooksDisabled` values follow Factory's extension-only precedence; system-managed `allowManagedHooksOnly` blocks a user-hook installation. Hook sources, generated runtimes, runtime config, and private key commit through one rollback-capable transaction. Managed commands preserve the complete native snake_case payload, propagate frozen or revoked state through exit code `2`, and use Factory's native PowerShell call operator with explicit `$LASTEXITCODE` propagation on Windows. Run `/hooks` after installation to refresh Droid's session snapshot and verify remote organization policy.
 
-Qwen Code installation resolves its user settings through explicit `QWEN_HOME`, `~/.qwen/.env`, then `~/.env`. Elydora preserves comments and unrelated settings, leaves workspace settings unchanged, and uses Qwen's explicit Bash or PowerShell command contract with millisecond timeouts. Run `/hooks` after installation to review the changes.
+Qwen Code 0.20.0 installation writes exact, named `PreToolUse`, `PostToolUse`, and `PostToolUseFailure` command hooks to `$QWEN_HOME/settings.json` (default `~/.qwen/settings.json`). Explicit environment values lead Qwen's home bootstrap, followed by `<QWEN_HOME>/.env` or `~/.qwen/.env`, then `~/.env`; a discovered home contributes its own `.env`. System defaults, trusted workspace settings, system overrides, and trust rules remain read-only and participate in effective `disableAllHooks` evaluation. User settings, generated runtimes, runtime config, and private key commit through one rollback-capable transaction guarded by snapshots of every consulted source. Managed handlers use ten-second millisecond timeouts, preserve Qwen's complete native snake_case payload for success and failure events, and propagate frozen or revoked state through exit code `2`. Run `/hooks` after installation and restart active Qwen Code sessions.
 
 Kiro CLI installation covers both runtime contracts. Kiro CLI v2 uses the generated custom agent through `kiro-cli --agent elydora-audit`. Kiro CLI v3 loads the global standalone hooks when started with `kiro-cli --v3`.
 

sdks/python/elydora/plugins/qwen.py

@@ -1,161 +1,86 @@
-"""Qwen Code native global user-hook integration."""
+"""Qwen Code 0.20 native global user-hook integration."""
 
 from __future__ import annotations
 
-import json
+from typing import Dict
-import os
-from typing import List, Optional
 
-from ._transaction import FileChange, file_change, write_changes
 from .base import AgentPlugin, InstallConfig, PluginStatus
-from .hook_template import generate_hook_script
+from .qwen_config import render_qwen_document
-from .qwen_config import render_document
 from .qwen_contract import (
     AGENT_KEY,
-    AUDIT_SCRIPT,
+    AUDIT_HOOK_NAME,
-    GUARD_SCRIPT,
+    GUARD_HOOK_NAME,
-    build_group,
+    JsonObject,
-    elydora_dir,
+    build_qwen_group,
-    runtime_contracts,
+    qwen_runtime_contracts,
-    validate_javascript_regexes,
 )
-from .qwen_io import (
+from .qwen_installation import (
-    read_document,
+    QwenRuntimePaths,
-    rendered_change,
+    commit_qwen_installation,
-    require_runtime,
+    commit_qwen_uninstall,
-    runtime_files_exist,
+    preflight_qwen_installation,
+    prepare_qwen_installation,
+    prepare_qwen_uninstall,
 )
+from .qwen_io import qwen_runtime_files_exist
+from .qwen_sources import read_qwen_sources
 
 
-def _agent_directory(agent_id: str) -> str:
+def _installed_groups(paths: QwenRuntimePaths) -> Dict[str, JsonObject]:
-    if (
+    return {
-        not agent_id
+        "PreToolUse": build_qwen_group(paths.guard_path, GUARD_HOOK_NAME),
-        or agent_id in {".", ".."}
+        "PostToolUse": build_qwen_group(paths.audit_path, AUDIT_HOOK_NAME),
-        or os.path.basename(agent_id) != agent_id
+        "PostToolUseFailure": build_qwen_group(
-        or os.path.isabs(agent_id)
+            paths.audit_path, AUDIT_HOOK_NAME
-    ):
+        ),
-        raise ValueError("agent_id must be a single non-empty path segment")
+    }
-    return os.path.join(elydora_dir(), agent_id)
-
-
-def _same_path(left: str, right: str) -> bool:
-    return os.path.normcase(os.path.abspath(left)) == os.path.normcase(
-        os.path.abspath(right)
-    )
-
-
-def _append_change(changes: List[FileChange], change: Optional[FileChange]) -> None:
-    if change is not None:
-        changes.append(change)
 
 
 class QwenPlugin(AgentPlugin):
-    """Install Elydora into Qwen Code's global user settings."""
+    """Install Elydora into Qwen Code's native user settings."""
+
+    manages_guard_runtime = True
+
+    def preflight_install(self, config: InstallConfig) -> None:
+        sources = read_qwen_sources()
+        preflight_qwen_installation(config, sources)
 
     def install(self, config: InstallConfig) -> None:
-        agent_id = config.get("agent_id", "")
+        sources = read_qwen_sources()
-        agent_directory = _agent_directory(agent_id)
+        paths = preflight_qwen_installation(config, sources)
-        document = read_document()
+        rendered = render_qwen_document(
-        validate_javascript_regexes([document.hooks])
+            sources.user,
-
-        guard_path = config.get("guard_script_path", "")
-        expected_guard_path = os.path.join(agent_directory, GUARD_SCRIPT)
-        if not _same_path(guard_path, expected_guard_path):
-            raise ValueError(
-                "Elydora guard runtime must use the managed agent directory: "
-                f"{expected_guard_path}"
-            )
-        require_runtime(guard_path, "Elydora guard runtime")
-        audit_path = os.path.join(agent_directory, AUDIT_SCRIPT)
-        rendered = render_document(
-            document,
             None,
-            {
+            _installed_groups(paths),
-                "PreToolUse": build_group(guard_path),
-                "PostToolUse": build_group(audit_path),
-            },
         )
-
+        changes = prepare_qwen_installation(config, paths, rendered)
-        runtime_config = {
+        commit_qwen_installation(changes, sources)
-            "org_id": config.get("org_id", ""),
+        print(f"Qwen Code hooks installed at {sources.user.file_path}")
-            "agent_id": agent_id,
+        print("Qwen Code verification: run /hooks and restart active sessions.")
-            "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"),
-        )
-        changes: List[FileChange] = []
-        _append_change(
-            changes,
-            file_change(
-                os.path.join(agent_directory, "config.json"),
-                "Elydora runtime config",
-                json.dumps(runtime_config, indent=2) + "\n",
-                0o600,
-            ),
-        )
-        _append_change(
-            changes,
-            file_change(
-                os.path.join(agent_directory, "private.key"),
-                "Elydora private key",
-                config.get("private_key", ""),
-                0o600,
-            ),
-        )
-        _append_change(
-            changes,
-            file_change(
-                audit_path,
-                "Elydora audit runtime",
-                audit_script,
-                0o700,
-            ),
-        )
-        _append_change(changes, rendered_change(rendered))
-        write_changes(changes, "Write Qwen Code installation")
-
-        print(f"Qwen Code: user hooks installed at {document.file_path}")
-        print("Qwen Code: run /hooks to review the Elydora hook changes.")
 
     def uninstall(self, agent_id: str = "") -> None:
-        document = read_document()
+        sources = read_qwen_sources()
-        if not document.exists:
+        rendered = render_qwen_document(
-            return
+            sources.user, agent_id or None, {}
-        rendered = render_document(document, agent_id or None, {})
-        change = rendered_change(rendered)
-        write_changes(
-            [change] if change is not None else [],
-            "Write Qwen Code settings",
         )
+        commit_qwen_uninstall(prepare_qwen_uninstall(rendered), sources)
 
     def status(self) -> PluginStatus:
-        document = read_document()
+        sources = read_qwen_sources()
-        contracts = runtime_contracts(document.hooks)
+        contracts = qwen_runtime_contracts(sources.user.hooks)
-        if document.hooks_disabled:
+        configured = not sources.disable_control.disabled and bool(contracts)
-            return PluginStatus(
+        installed = configured and qwen_runtime_files_exist(contracts)
-                installed=False,
+        if installed:
-                agent=AGENT_KEY,
+            details = f"Config: {sources.user.file_path}"
-                details=f"Configured hooks are disabled: {document.file_path}",
+        elif contracts and sources.disable_control.disabled:
-            )
+            details = f"Configured hooks are disabled: {sources.user.file_path}"
-        if not contracts:
+        elif contracts:
-            return PluginStatus(
-                installed=False,
-                agent=AGENT_KEY,
-                details="Not installed",
-            )
-        installed = runtime_files_exist(contracts)
             details = (
-            f"Config: {document.file_path}"
+                f"Configured at {sources.user.file_path}; "
-            if installed
+                "managed contract incomplete"
-            else f"Configured at {document.file_path}; runtime scripts missing"
             )
+        else:
+            details = "Not installed"
         return PluginStatus(
             installed=installed,
             agent=AGENT_KEY,

sdks/python/elydora/plugins/qwen_command.py

@@ -0,0 +1,134 @@
+"""Qwen Code command construction and exact runtime ownership."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+import os
+import sys
+from typing import Optional, Tuple
+
+
+@dataclass(frozen=True)
+class QwenRuntimeReference:
+    agent_id: str
+    script_path: str
+
+
+def same_qwen_path(left: str, right: str) -> bool:
+    return os.path.normcase(os.path.abspath(left)) == os.path.normcase(
+        os.path.abspath(right)
+    )
+
+
+def same_qwen_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 build_qwen_command(script_path: str) -> str:
+    if not os.path.isabs(sys.executable) or not os.path.isabs(script_path):
+        raise ValueError(
+            "Qwen Code hook commands require absolute executable and script paths"
+        )
+    invocation = (
+        f"{_quote_powershell(sys.executable)} {_quote_powershell(script_path)}"
+        if os.name == "nt"
+        else f"{_quote_posix(sys.executable)} {_quote_posix(script_path)}"
+    )
+    return f"& {invocation}; exit $LASTEXITCODE" if os.name == "nt" else invocation
+
+
+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_command(command: str) -> Optional[Tuple[str, str]]:
+    if not command.startswith("& "):
+        return None
+    executable = _read_powershell_argument(command, 2)
+    if executable is None or executable[1] >= len(command):
+        return None
+    if command[executable[1]] != " ":
+        return None
+    script = _read_powershell_argument(command, executable[1] + 1)
+    if script is None or command[script[1] :] != "; exit $LASTEXITCODE":
+        return None
+    return executable[0], script[0]
+
+
+def qwen_runtime_reference(
+    command: str, script_name: str
+) -> Optional[QwenRuntimeReference]:
+    parsed = (
+        _parse_powershell_command(command)
+        if os.name == "nt"
+        else _parse_posix_command(command)
+    )
+    if (
+        parsed is None
+        or not same_qwen_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_qwen_path(os.path.dirname(agent_directory), runtime_root):
+        return None
+    agent_id = os.path.basename(agent_directory)
+    if agent_id in {"", ".", ".."}:
+        return None
+    return QwenRuntimeReference(agent_id, parsed[1])

sdks/python/elydora/plugins/qwen_config.py

@@ -1,4 +1,4 @@
-"""Qwen Code settings parsing and source-preserving rendering."""
+"""Qwen Code JSONC parsing and source-preserving hook edits."""
 
 from __future__ import annotations
 
@@ -6,12 +6,13 @@ from dataclasses import dataclass
 from typing import Dict, Optional
 
 from ._jsonc import JsoncEditor
+from ._managed_files import FileSnapshot
 from .qwen_contract import (
-    TOOL_EVENTS,
+    MANAGED_EVENTS,
     JsonObject,
-    QwenHookSettings,
+    QwenHooks,
-    managed_removals,
+    managed_qwen_removals,
-    read_hook_settings,
+    read_qwen_hooks,
 )
 
 
@@ -20,83 +21,121 @@ OWNED_FILE_MARKER = "// Managed by Elydora"
 
 @dataclass(frozen=True)
 class QwenDocument:
+    kind: str
     file_path: str
     exists: bool
     raw: str
+    snapshot: Optional[FileSnapshot]
     root: JsonObject
-    hooks: QwenHookSettings
+    hooks: QwenHooks
     has_hooks_container: bool
-    hooks_disabled: bool
+    disable_all_hooks: Optional[bool]
+    folder_trust_enabled: Optional[bool]
     owned_file: bool
 
 
 @dataclass(frozen=True)
-class RenderedDocument:
+class RenderedQwenDocument:
     document: QwenDocument
     changed: bool
     next_source: Optional[str]
 
 
-def _label(file_path: str) -> str:
+def qwen_source_label(kind: str) -> str:
-    return f"Qwen Code settings at {file_path}"
+    labels = {
+        "system-defaults": "Qwen Code system defaults",
+        "user": "Qwen Code user settings",
+        "workspace": "Qwen Code workspace settings",
+        "system": "Qwen Code system override settings",
+    }
+    try:
+        return labels[kind]
+    except KeyError as error:
+        raise ValueError(f"Unsupported Qwen Code document kind: {kind}") from error
 
 
-def parse_document(
+def qwen_document_label(document: QwenDocument) -> str:
+    return qwen_source_label(document.kind)
+
+
+def parse_qwen_jsonc_object(raw: str, label: str) -> JsonObject:
+    editor = JsoncEditor(raw, label, allow_trailing_commas=False)
+    if not isinstance(editor.value, dict):
+        raise ValueError(f"{label} must contain a JSON object")
+    return editor.value
+
+
+def _folder_trust_enabled(
+    root: JsonObject, label: str
+) -> Optional[bool]:
+    if "security" not in root:
+        return None
+    security = root["security"]
+    if not isinstance(security, dict):
+        raise ValueError(f'{label} field "security" must be an object')
+    if "folderTrust" not in security:
+        return None
+    folder_trust = security["folderTrust"]
+    if not isinstance(folder_trust, dict):
+        raise ValueError(
+            f'{label} field "security.folderTrust" must be an object'
+        )
+    enabled = folder_trust.get("enabled")
+    if enabled is not None and not isinstance(enabled, bool):
+        raise ValueError(
+            f'{label} field "security.folderTrust.enabled" must be a boolean'
+        )
+    return enabled
+
+
+def parse_qwen_document(
     *,
+    kind: str,
     exists: bool,
     file_path: str,
     raw: str,
+    snapshot: Optional[FileSnapshot] = None,
 ) -> QwenDocument:
-    label = _label(file_path)
+    label = f"{qwen_source_label(kind)} at {file_path}"
-    editor = JsoncEditor(
+    root = parse_qwen_jsonc_object(raw, label)
-        raw,
-        label,
-        allow_trailing_commas=False,
-    )
-    if not isinstance(editor.value, dict):
-        raise ValueError(f"{label} must contain a JSON object")
-    root = editor.value
     disabled = root.get("disableAllHooks")
     if "disableAllHooks" in root and not isinstance(disabled, bool):
         raise ValueError(f'{label} field "disableAllHooks" must be a boolean')
-    has_container = "hooks" in root
+    has_hooks = "hooks" in root
-    hooks = (
-        read_hook_settings(root["hooks"], f'{label} field "hooks"')
-        if has_container
-        else {}
-    )
     return QwenDocument(
-        file_path,
+        kind=kind,
-        exists,
+        file_path=file_path,
-        raw,
+        exists=exists,
-        root,
+        raw=raw,
-        hooks,
+        snapshot=snapshot,
-        has_container,
+        root=root,
-        disabled is True,
+        hooks=read_qwen_hooks(root["hooks"]) if has_hooks else {},
-        raw.startswith(OWNED_FILE_MARKER),
+        has_hooks_container=has_hooks,
+        disable_all_hooks=disabled if isinstance(disabled, bool) else None,
+        folder_trust_enabled=_folder_trust_enabled(root, label),
+        owned_file=kind == "user" and raw.startswith(OWNED_FILE_MARKER),
     )
 
 
-def create_owned_document(file_path: str) -> QwenDocument:
+def create_qwen_document(kind: str, file_path: str) -> QwenDocument:
-    return parse_document(
+    raw = f"{OWNED_FILE_MARKER}\n{{}}\n" if kind == "user" else "{}\n"
+    return parse_qwen_document(
+        kind=kind,
         exists=False,
         file_path=file_path,
-        raw=f"{OWNED_FILE_MARKER}\n{{}}\n",
+        raw=raw,
     )
 
 
-def _settings_from_editor(
+def _current_document(
-    editor: JsoncEditor,
+    document: QwenDocument, editor: JsoncEditor
-    document: QwenDocument,
+) -> QwenDocument:
-) -> QwenHookSettings:
+    return parse_qwen_document(
-    root = editor.value
+        kind=document.kind,
-    if not isinstance(root, dict):
+        exists=document.exists,
-        raise ValueError(f"{_label(document.file_path)} must contain a JSON object")
+        file_path=document.file_path,
-    if "hooks" not in root:
+        raw=editor.raw,
-        return {}
+        snapshot=document.snapshot,
-    return read_hook_settings(
-        root["hooks"],
-        f'{_label(document.file_path)} field "hooks"',
     )
 
 
@@ -105,8 +144,8 @@ def _remove_managed(
     document: QwenDocument,
     agent_id: Optional[str],
 ) -> None:
-    removals = managed_removals(document.hooks, agent_id)
+    removals = managed_qwen_removals(document.hooks, agent_id)
-    for event in TOOL_EVENTS:
+    for event in MANAGED_EVENTS:
         event_removals = sorted(
             (item for item in removals if item.event == event),
             key=lambda item: item.group_index,
@@ -120,13 +159,11 @@ def _remove_managed(
             for handler_index in sorted(removal.handler_indexes, reverse=True):
                 editor.delete((*group_path, "hooks", handler_index))
         if event_removals:
-            current = _settings_from_editor(editor, document)
+            current = _current_document(document, editor)
-            if event in current and not current[event]:
+            if event in current.hooks and not current.hooks[event]:
                 editor.delete(("hooks", event))
-    root = editor.value
+    current = _current_document(document, editor)
-    if isinstance(root, dict) and "hooks" in root:
+    if current.has_hooks_container and not current.hooks:
-        current = _settings_from_editor(editor, document)
-        if not current:
         editor.delete(("hooks",))
 
 
@@ -136,42 +173,57 @@ def _append_group(
     event: str,
     group: JsonObject,
 ) -> None:
-    root = editor.value
+    current = _current_document(document, editor)
-    if not isinstance(root, dict):
+    if not current.has_hooks_container:
-        raise ValueError(f"{_label(document.file_path)} must contain a JSON object")
-    if "hooks" not in root:
         editor.add_property((), "hooks", {event: [group]})
-        return
+    elif event in current.hooks:
-    current = _settings_from_editor(editor, document)
-    if event in current:
         editor.append(("hooks", event), group)
     else:
         editor.add_property(("hooks",), event, [group])
 
 
-def render_document(
+def _already_installed(
+    document: QwenDocument, additions: Dict[str, JsonObject]
+) -> bool:
+    if set(additions) != set(MANAGED_EVENTS):
+        return False
+    removals = managed_qwen_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_qwen_document(
     document: QwenDocument,
     agent_id: Optional[str],
     additions: Dict[str, JsonObject],
-) -> RenderedDocument:
+) -> RenderedQwenDocument:
+    if agent_id is None and _already_installed(document, additions):
+        return RenderedQwenDocument(document, False, document.raw)
     editor = JsoncEditor(
         document.raw,
-        _label(document.file_path),
+        f"{qwen_document_label(document)} at {document.file_path}",
         allow_trailing_commas=False,
     )
     _remove_managed(editor, document, agent_id)
-    for event in TOOL_EVENTS:
+    for event in MANAGED_EVENTS:
         group = additions.get(event)
         if group is not None:
             _append_group(editor, document, event, group)
-    next_document = parse_document(
+    current = _current_document(document, editor)
-        exists=document.exists,
+    if not additions and document.owned_file and not current.root:
-        file_path=document.file_path,
+        return RenderedQwenDocument(document, True, None)
-        raw=editor.raw,
+    return RenderedQwenDocument(
-    )
-    if not additions and document.owned_file and not next_document.root:
-        return RenderedDocument(document, True, None)
-    return RenderedDocument(
         document,
         editor.raw != document.raw,
         editor.raw,

sdks/python/elydora/plugins/qwen_contract.py

@@ -1,4 +1,4 @@
-"""Qwen Code hook schema, commands, and exact ownership rules."""
+"""Qwen Code 0.20 hook schema and exact ownership rules."""
 
 from __future__ import annotations
 
@@ -8,17 +8,26 @@ import math
 import os
 import shutil
 import subprocess  # nosec B404
-import sys
+from typing import Any, Dict, List, Optional, Sequence, Tuple
-from typing import Any, Dict, List, Optional, Sequence, Set, Tuple
+
+from .qwen_command import (
+    QwenRuntimeReference,
+    build_qwen_command,
+    qwen_runtime_reference,
+    same_qwen_agent_id,
+    same_qwen_path,
+)
 
 
 AGENT_KEY = "qwen"
 GUARD_SCRIPT = "guard.py"
 AUDIT_SCRIPT = "hook.py"
+GUARD_HOOK_NAME = "elydora-guard"
+AUDIT_HOOK_NAME = "elydora-audit"
 HOOK_TIMEOUT_MILLISECONDS = 10_000
-TOOL_EVENTS = ("PreToolUse", "PostToolUse")
+MANAGED_EVENTS = ("PreToolUse", "PostToolUse", "PostToolUseFailure")
 
-_EVENT_NAMES = {
+KNOWN_EVENTS = {
     "PreToolUse",
     "PostToolUse",
     "PostToolUseFailure",
@@ -41,9 +50,25 @@ _EVENT_NAMES = {
     "TodoCompleted",
     "InstructionsLoaded",
 }
-_CONFIG_FIELDS = {"enabled", "disabled", "notifications"}
+
-_HANDLER_KEYS = {"command", "shell", "timeout", "type"}
+REGEX_MATCHER_EVENTS = {
-_GROUP_KEYS = {"hooks", "matcher"}
+    "PreToolUse",
+    "PostToolUse",
+    "PostToolUseFailure",
+    "PermissionRequest",
+    "PermissionDenied",
+    "SubagentStart",
+    "SubagentStop",
+    "PreCompact",
+    "PostCompact",
+    "SessionStart",
+    "SessionEnd",
+    "StopFailure",
+    "Notification",
+    "InstructionsLoaded",
+    "UserPromptExpansion",
+}
+
 _REGEX_TIMEOUT_SECONDS = 10
 _REGEX_VALIDATOR = """import fs from 'node:fs';
 const entries = JSON.parse(fs.readFileSync(0, 'utf8'));
@@ -59,11 +84,11 @@ for (const entry of entries) {
 """
 
 JsonObject = Dict[str, Any]
-QwenHookSettings = Dict[str, Any]
+QwenHooks = Dict[str, Any]
 
 
 @dataclass(frozen=True)
-class RuntimeContract:
+class QwenRuntimeContract:
     agent_id: str
     guard_path: str
     audit_path: str
@@ -81,67 +106,37 @@ def elydora_dir() -> str:
     return os.path.join(os.path.expanduser("~"), ".elydora")
 
 
-def _quote_posix(value: str) -> str:
+def _optional_string(value: JsonObject, field: str, label: str) -> None:
-    return "'" + value.replace("'", "'\"'\"'") + "'"
+    if field in value and not isinstance(value[field], str):
+        raise ValueError(f'{label} field "{field}" must be a string')
 
 
-def _quote_powershell(value: str) -> str:
+def _optional_boolean(value: JsonObject, field: str, label: str) -> None:
-    return "'" + value.replace("'", "''") + "'"
+    if field in value and not isinstance(value[field], bool):
+        raise ValueError(f'{label} field "{field}" must be a boolean')
 
 
-def build_command(script_path: str) -> str:
+def _optional_string_map(value: JsonObject, field: str, label: str) -> None:
-    if os.name == "nt":
+    if field not in value:
-        invocation = (
-            f"{_quote_powershell(sys.executable)} {_quote_powershell(script_path)}"
-        )
-        return f"& {invocation}; exit $LASTEXITCODE"
-    return f"{_quote_posix(sys.executable)} {_quote_posix(script_path)}"
-
-
-def build_group(script_path: str) -> JsonObject:
-    return {
-        "matcher": "*",
-        "hooks": [
-            {
-                "type": "command",
-                "command": build_command(script_path),
-                "shell": "powershell" if os.name == "nt" else "bash",
-                "timeout": HOOK_TIMEOUT_MILLISECONDS,
-            }
-        ],
-    }
-
-
-def _optional_string(value: JsonObject, key: str, label: str) -> None:
-    if key in value and not isinstance(value[key], str):
-        raise ValueError(f'{label} field "{key}" must be a string')
-
-
-def _optional_boolean(value: JsonObject, key: str, label: str) -> None:
-    if key in value and not isinstance(value[key], bool):
-        raise ValueError(f'{label} field "{key}" must be a boolean')
-
-
-def _optional_string_map(value: JsonObject, key: str, label: str) -> None:
-    if key not in value:
         return
-    item = value[key]
+    item = value[field]
     if not isinstance(item, dict) or any(
         not isinstance(entry, str) for entry in item.values()
     ):
-        raise ValueError(f'{label} field "{key}" must contain string values')
+        raise ValueError(f'{label} field "{field}" must map names to strings')
 
 
 def _validate_timeout(value: JsonObject, label: str) -> None:
     if "timeout" not in value:
         return
     timeout = value["timeout"]
-    if (
+    try:
-        isinstance(timeout, bool)
+        finite = math.isfinite(timeout)
-        or not isinstance(timeout, (int, float))
+    except (TypeError, OverflowError):
-        or not math.isfinite(timeout)
+        finite = False
-        or timeout < 0
+    if isinstance(timeout, bool) or not isinstance(timeout, (int, float)) or not finite:
-    ):
+        raise ValueError(f"{label} timeout must be a non-negative finite number")
+    if timeout < 0:
         raise ValueError(f"{label} timeout must be a non-negative finite number")
 
 
@@ -151,20 +146,20 @@ def _validate_handler(value: Any, label: str) -> JsonObject:
     handler = dict(value)
     kind = handler.get("type")
     if kind not in {"command", "http", "prompt"}:
-        raise ValueError(f'{label} type must be "command", "http", or "prompt"')
+        raise ValueError(f'{label} has unsupported type "{kind}"')
     _validate_timeout(handler, label)
-    for key in ("name", "description", "statusMessage", "source"):
+    for field in ("name", "description", "statusMessage", "source"):
-        _optional_string(handler, key, label)
+        _optional_string(handler, field, label)
     if kind == "command":
         if not isinstance(handler.get("command"), str) or not handler["command"]:
-            raise ValueError(f"{label} command must be a non-empty string")
+            raise ValueError(f"{label} requires a non-empty command")
         _optional_string_map(handler, "env", label)
         _optional_boolean(handler, "async", label)
         if "shell" in handler and handler["shell"] not in {"bash", "powershell"}:
             raise ValueError(f'{label} shell must be "bash" or "powershell"')
     elif kind == "http":
         if not isinstance(handler.get("url"), str) or not handler["url"]:
-            raise ValueError(f"{label} url must be a non-empty string")
+            raise ValueError(f"{label} requires a non-empty url")
         _optional_string_map(handler, "headers", label)
         _optional_boolean(handler, "once", label)
         _optional_string(handler, "if", label)
@@ -176,12 +171,13 @@ def _validate_handler(value: Any, label: str) -> JsonObject:
             raise ValueError(f"{label} allowedEnvVars must be an array of strings")
     else:
         if not isinstance(handler.get("prompt"), str) or not handler["prompt"]:
-            raise ValueError(f"{label} prompt must be a non-empty string")
+            raise ValueError(f"{label} requires a non-empty prompt")
         _optional_string(handler, "model", label)
     return handler
 
 
-def _validate_group(value: Any, label: str) -> JsonObject:
+def _validate_group(value: Any, event: str, index: int) -> JsonObject:
+    label = f"Qwen Code settings group hooks.{event}[{index}]"
     if not isinstance(value, dict):
         raise ValueError(f"{label} must be an object")
     group = dict(value)
@@ -193,49 +189,47 @@ def _validate_group(value: Any, label: str) -> JsonObject:
     if not isinstance(handlers, list):
         raise ValueError(f"{label} must contain a hooks array")
     group["hooks"] = [
-        _validate_handler(handler, f"{label}.hooks[{index}]")
+        _validate_handler(handler, f"{label}.hooks[{handler_index}]")
-        for index, handler in enumerate(handlers)
+        for handler_index, handler in enumerate(handlers)
     ]
     return group
 
 
-def read_hook_settings(value: Any, label: str) -> QwenHookSettings:
+def read_qwen_hooks(value: Any) -> QwenHooks:
     if not isinstance(value, dict):
-        raise ValueError(f"{label} must contain a JSON object")
+        raise ValueError('Qwen Code settings field "hooks" must be an object')
-    settings = dict(value)
+    hooks: QwenHooks = {}
     for event, groups in value.items():
-        if event in _CONFIG_FIELDS:
+        if event not in KNOWN_EVENTS:
+            hooks[event] = groups
             continue
-        if event not in _EVENT_NAMES:
-            raise ValueError(f'{label} contains unsupported field "{event}"')
         if not isinstance(groups, list):
-            raise ValueError(f'{label} field "{event}" must be an array')
+            raise ValueError(
-        settings[event] = [
+                f'Qwen Code settings field "hooks.{event}" must be an array'
-            _validate_group(group, f'{label} field "{event}"[{index}]')
+            )
+        hooks[event] = [
+            _validate_group(group, event, index)
             for index, group in enumerate(groups)
         ]
-    return settings
+    return hooks
 
 
-def _regex_entries(settings: QwenHookSettings) -> List[JsonObject]:
+def _matcher_entries(sources: Sequence[QwenHooks]) -> List[JsonObject]:
-    entries = []
+    entries: List[JsonObject] = []
-    for event, groups in settings.items():
+    for hooks in sources:
-        if event in _CONFIG_FIELDS:
+        for event in REGEX_MATCHER_EVENTS:
-            continue
+            for index, group in enumerate(hooks.get(event, [])):
-        for index, group in enumerate(groups):
                 matcher = group.get("matcher")
                 if isinstance(matcher, str) and matcher.strip() not in {"", "*"}:
-                entries.append(
+                    entries.append({
-                    {
+                        "label": f"Qwen Code settings group hooks.{event}[{index}]",
-                        "label": f'Qwen Code hooks field "{event}"[{index}] matcher',
                         "pattern": matcher,
-                    }
+                    })
-                )
     return entries
 
 
-def validate_javascript_regexes(settings: Sequence[QwenHookSettings]) -> None:
+def validate_javascript_matchers(sources: Sequence[QwenHooks]) -> None:
-    entries = [entry for source in settings for entry in _regex_entries(source)]
+    entries = _matcher_entries(sources)
     if not entries:
         return
     node_path = shutil.which("node")
@@ -267,143 +261,148 @@ def validate_javascript_regexes(settings: Sequence[QwenHookSettings]) -> None:
         or f"Node.js exited with code {result.returncode}"
     )
     raise ValueError(
-        f"Qwen Code matcher must be a valid JavaScript regular expression: {message}"
+        "Qwen Code matcher must be a valid JavaScript regular expression: "
+        + message
     )
 
 
-def _read_quoted_argument(command: str, start: int) -> Optional[Tuple[str, int]]:
+def build_qwen_group(script_path: str, name: str) -> JsonObject:
-    if start >= len(command) or command[start] != "'":
-        return None
-    value = ""
-    index = start + 1
-    while index < len(command):
-        if os.name == "nt" and command[index : index + 2] == "''":
-            value += "'"
-            index += 2
-            continue
-        if os.name != "nt" and command[index : index + 5] == "'\"'\"'":
-            value += "'"
-            index += 5
-            continue
-        if command[index] == "'":
-            return value, index + 1
-        value += command[index]
-        index += 1
-    return None
-
-
-def _parse_generated_command(command: str) -> Optional[Tuple[str, str]]:
-    if os.name == "nt" and not command.startswith("& "):
-        return None
-    start = 2 if os.name == "nt" else 0
-    executable = _read_quoted_argument(command, start)
-    if executable is None or executable[1] >= len(command):
-        return None
-    if command[executable[1]] != " ":
-        return None
-    script = _read_quoted_argument(command, executable[1] + 1)
-    if script is None:
-        return None
-    suffix = "; exit $LASTEXITCODE" if os.name == "nt" else ""
-    if command[script[1] :] != suffix or not executable[0] or not script[0]:
-        return None
-    return executable[0], script[0]
-
-
-def _same_path(left: str, right: str) -> bool:
-    return os.path.normcase(os.path.abspath(left)) == os.path.normcase(
-        os.path.abspath(right)
-    )
-
-
-def same_agent_id(left: str, right: str) -> bool:
-    return os.path.normcase(left) == os.path.normcase(right)
-
-
-def managed_agent_id(handler: JsonObject, script_name: str) -> Optional[str]:
-    if set(handler) != _HANDLER_KEYS:
-        return None
-    expected_shell = "powershell" if os.name == "nt" else "bash"
-    if (
-        handler.get("type") != "command"
-        or handler.get("timeout") != HOOK_TIMEOUT_MILLISECONDS
-        or handler.get("shell") != expected_shell
-        or not isinstance(handler.get("command"), str)
-    ):
-        return None
-    parsed = _parse_generated_command(handler["command"])
-    if parsed is None or not _same_path(parsed[0], sys.executable):
-        return None
-    script_path = parsed[1]
-    if os.path.basename(script_path) != script_name:
-        return None
-    agent_directory = os.path.dirname(script_path)
-    if not _same_path(os.path.dirname(agent_directory), elydora_dir()):
-        return None
-    agent_id = os.path.basename(agent_directory)
-    return agent_id if agent_id not in {"", ".", ".."} else None
-
-
-def _exact_owned_group(group: JsonObject, indexes: Tuple[int, ...]) -> bool:
-    return (
-        set(group) == _GROUP_KEYS
-        and group.get("matcher") == "*"
-        and bool(indexes)
-        and len(indexes) == len(group["hooks"])
-    )
-
-
-def managed_removals(
-    settings: QwenHookSettings,
-    agent_id: Optional[str] = None,
-) -> List[ManagedRemoval]:
-    removals = []
-    for event, script_name in (
-        ("PreToolUse", GUARD_SCRIPT),
-        ("PostToolUse", AUDIT_SCRIPT),
-    ):
-        for group_index, group in enumerate(settings.get(event, [])):
-            indexes = tuple(
-                index
-                for index, handler in enumerate(group["hooks"])
-                if (
-                    (managed_id := managed_agent_id(handler, script_name)) is not None
-                    and (agent_id is None or same_agent_id(managed_id, agent_id))
-                )
-            )
-            if indexes:
-                removals.append(
-                    ManagedRemoval(
-                        event,
-                        group_index,
-                        indexes,
-                        _exact_owned_group(group, indexes),
-                    )
-                )
-    return removals
-
-
-def _managed_ids(groups: List[JsonObject], script_name: str) -> Set[str]:
     return {
-        agent_id
+        "hooks": [{
-        for group in groups
+            "type": "command",
-        for handler in group["hooks"]
+            "name": name,
-        if (agent_id := managed_agent_id(handler, script_name)) is not None
+            "command": build_qwen_command(script_path),
+            "shell": "powershell" if os.name == "nt" else "bash",
+            "timeout": HOOK_TIMEOUT_MILLISECONDS,
+        }]
     }
 
 
-def runtime_contracts(settings: QwenHookSettings) -> List[RuntimeContract]:
+def _exact_current_group(group: JsonObject) -> bool:
-    guards = _managed_ids(settings.get("PreToolUse", []), GUARD_SCRIPT)
+    return set(group) == {"hooks"}
-    audits = _managed_ids(settings.get("PostToolUse", []), AUDIT_SCRIPT)
+
-    contracts = []
+
-    for agent_id in sorted(guards, key=os.path.normcase):
+def _exact_legacy_group(group: JsonObject) -> bool:
-        if any(same_agent_id(agent_id, audit_id) for audit_id in audits):
+    return set(group) == {"hooks", "matcher"} and group.get("matcher") == "*"
-            root = os.path.join(elydora_dir(), agent_id)
+
-            contracts.append(
+
-                RuntimeContract(
+def _current_reference(
-                    agent_id,
+    handler: JsonObject, script_name: str, hook_name: str
-                    os.path.join(root, GUARD_SCRIPT),
+) -> Optional[QwenRuntimeReference]:
-                    os.path.join(root, AUDIT_SCRIPT),
+    if (
+        set(handler) == {"command", "name", "shell", "timeout", "type"}
+        and handler.get("type") == "command"
+        and handler.get("name") == hook_name
+        and handler.get("shell") == ("powershell" if os.name == "nt" else "bash")
+        and handler.get("timeout") == HOOK_TIMEOUT_MILLISECONDS
+        and isinstance(handler.get("command"), str)
+    ):
+        return qwen_runtime_reference(handler["command"], script_name)
+    return None
+
+
+def _legacy_reference(
+    handler: JsonObject, script_name: str
+) -> Optional[QwenRuntimeReference]:
+    if (
+        set(handler) == {"command", "shell", "timeout", "type"}
+        and handler.get("type") == "command"
+        and handler.get("shell") == ("powershell" if os.name == "nt" else "bash")
+        and handler.get("timeout") == HOOK_TIMEOUT_MILLISECONDS
+        and isinstance(handler.get("command"), str)
+    ):
+        return qwen_runtime_reference(handler["command"], script_name)
+    return None
+
+
+def _managed_reference(
+    handler: JsonObject,
+    script_name: str,
+    hook_name: str,
+    include_legacy: bool,
+) -> Optional[QwenRuntimeReference]:
+    current = _current_reference(handler, script_name, hook_name)
+    if current is not None or not include_legacy:
+        return current
+    return _legacy_reference(handler, script_name)
+
+
+_EVENT_CONTRACTS = (
+    ("PreToolUse", GUARD_SCRIPT, GUARD_HOOK_NAME),
+    ("PostToolUse", AUDIT_SCRIPT, AUDIT_HOOK_NAME),
+    ("PostToolUseFailure", AUDIT_SCRIPT, AUDIT_HOOK_NAME),
+)
+
+
+def managed_qwen_removals(
+    hooks: QwenHooks, agent_id: Optional[str] = None
+) -> List[ManagedRemoval]:
+    removals: List[ManagedRemoval] = []
+    for event, script_name, hook_name in _EVENT_CONTRACTS:
+        for group_index, group in enumerate(hooks.get(event, [])):
+            handlers = group["hooks"]
+            indexes = tuple(
+                index
+                for index, handler in enumerate(handlers)
+                if (
+                    reference := _managed_reference(
+                        handler, script_name, hook_name, True
                     )
                 )
+                is not None
+                and (
+                    agent_id is None
+                    or same_qwen_agent_id(reference.agent_id, agent_id)
+                )
+            )
+            if indexes:
+                removals.append(ManagedRemoval(
+                    event,
+                    group_index,
+                    indexes,
+                    (_exact_current_group(group) or _exact_legacy_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[QwenRuntimeReference]]:
+    references: Dict[str, List[QwenRuntimeReference]] = {}
+    for group in groups:
+        if not _exact_current_group(group):
+            continue
+        for handler in group["hooks"]:
+            reference = _current_reference(handler, script_name, hook_name)
+            if reference is None:
+                continue
+            key = os.path.normcase(reference.agent_id)
+            references.setdefault(key, []).append(reference)
+    return references
+
+
+def qwen_runtime_contracts(hooks: QwenHooks) -> List[QwenRuntimeContract]:
+    guards = _references_for_event(
+        hooks.get("PreToolUse", []), GUARD_SCRIPT, GUARD_HOOK_NAME
+    )
+    posts = _references_for_event(
+        hooks.get("PostToolUse", []), AUDIT_SCRIPT, AUDIT_HOOK_NAME
+    )
+    failures = _references_for_event(
+        hooks.get("PostToolUseFailure", []), AUDIT_SCRIPT, AUDIT_HOOK_NAME
+    )
+    contracts: List[QwenRuntimeContract] = []
+    for key, guard in guards.items():
+        post = posts.get(key, [])
+        failure = failures.get(key, [])
+        if len(guard) != 1 or len(post) != 1 or len(failure) != 1:
+            continue
+        if not same_qwen_path(post[0].script_path, failure[0].script_path):
+            continue
+        contracts.append(QwenRuntimeContract(
+            guard[0].agent_id,
+            guard[0].script_path,
+            post[0].script_path,
+        ))
     return contracts

sdks/python/elydora/plugins/qwen_installation.py

@@ -0,0 +1,198 @@
+"""Transactional Qwen Code hook and runtime installation."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+import json
+import os
+from typing import List, Optional, Sequence
+
+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, FilePrecondition, file_change, source_change, write_changes
+from .base import InstallConfig
+from .guard_template import generate_guard_script
+from .hook_template import generate_hook_script
+from .qwen_command import same_qwen_path
+from .qwen_config import RenderedQwenDocument, qwen_document_label
+from .qwen_contract import AGENT_KEY, AUDIT_SCRIPT, GUARD_SCRIPT, elydora_dir
+from .qwen_io import validate_api_origin, validate_private_key, validate_runtime_tree
+from .qwen_sources import QwenSources, require_qwen_hooks_enabled
+
+
+@dataclass(frozen=True)
+class QwenRuntimePaths:
+    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) -> QwenRuntimePaths:
+    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_qwen_path(config.get("guard_script_path", ""), guard_path):
+        raise ValueError(
+            "Elydora guard runtime must use the managed agent directory: "
+            f"{guard_path}"
+        )
+    return QwenRuntimePaths(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"Qwen Code 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_qwen_installation(
+    config: InstallConfig, sources: QwenSources
+) -> QwenRuntimePaths:
+    require_qwen_hooks_enabled(sources)
+    _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 _rendered_change(
+    rendered: RenderedQwenDocument,
+) -> Optional[FileChange]:
+    if not rendered.changed:
+        return None
+    return source_change(
+        rendered.document.file_path,
+        qwen_document_label(rendered.document),
+        rendered.document.raw if rendered.document.exists else None,
+        rendered.next_source,
+        0o600,
+        MAX_SOURCE_BYTES,
+        expected_snapshot=rendered.document.snapshot,
+    )
+
+
+def _present(changes: Sequence[Optional[FileChange]]) -> List[FileChange]:
+    return [change for change in changes if change is not None]
+
+
+def prepare_qwen_installation(
+    config: InstallConfig,
+    paths: QwenRuntimePaths,
+    rendered: RenderedQwenDocument,
+) -> List[FileChange]:
+    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,
+    )
+    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 _read_only_preconditions(
+    sources: QwenSources, changed_path: Optional[str]
+) -> List[FilePrecondition]:
+    return [
+        condition
+        for condition in sources.preconditions
+        if changed_path is None
+        or not same_qwen_path(condition.file_path, changed_path)
+    ]
+
+
+def commit_qwen_installation(
+    changes: List[FileChange], sources: QwenSources
+) -> None:
+    write_changes(
+        changes,
+        "Install Qwen Code hooks",
+        _read_only_preconditions(sources, sources.user.file_path),
+    )
+
+
+def prepare_qwen_uninstall(
+    rendered: RenderedQwenDocument,
+) -> Optional[FileChange]:
+    return _rendered_change(rendered)
+
+
+def commit_qwen_uninstall(
+    change: Optional[FileChange], sources: QwenSources
+) -> None:
+    if change is None:
+        return
+    write_changes(
+        [change],
+        "Uninstall Qwen Code hooks",
+        _read_only_preconditions(sources, sources.user.file_path),
+    )

sdks/python/elydora/plugins/qwen_io.py

@@ -1,139 +1,200 @@
-"""Qwen Code home discovery, settings I/O, and runtime inspection."""
+"""Strict Qwen Code managed runtime inspection."""
 
 from __future__ import annotations
 
-import json
+import base64
 import os
-import re
+from typing import Any, List, Optional
-import tempfile
+import urllib.parse
-from typing import Optional
 
-from ._dotenv import parse_dotenv
+from ._managed_files import (
-from ._transaction import (
+    MAX_CONFIG_BYTES,
-    FileChange,
+    MAX_SECRET_BYTES,
-    read_optional,
+    physical_directory_exists,
-    regular_file_exists,
+    physical_file_exists,
-    require_runtime,
+    read_physical_file,
-    source_change,
 )
-from .qwen_config import (
+from ._strict_json import JsonObject, parse_json_object
-    QwenDocument,
+from .guard_template import generate_guard_script
-    RenderedDocument,
+from .hook_template import generate_hook_script
-    create_owned_document,
+from .qwen_command import same_qwen_agent_id, same_qwen_path
-    parse_document,
+from .qwen_contract import AGENT_KEY, QwenRuntimeContract, elydora_dir
-)
-from .qwen_contract import AGENT_KEY, JsonObject, RuntimeContract, same_agent_id
 
 
-def _default_qwen_home() -> str:
+def validate_api_origin(value: str, label: str = "base_url") -> None:
-    home_dir = os.path.expanduser("~")
-    if not home_dir or home_dir == "~":
-        home_dir = tempfile.gettempdir()
-    return os.path.join(home_dir, ".qwen")
-
-
-def _resolve_storage_path(value: str) -> str:
-    resolved = value
-    if value == "~":
-        resolved = os.path.expanduser("~")
-    elif value.startswith(("~/", "~\\")):
-        segments = [item for item in re.split(r"[/\\]+", value[2:]) if item]
-        resolved = os.path.join(os.path.expanduser("~"), *segments)
-    return resolved if os.path.isabs(resolved) else os.path.abspath(resolved)
-
-
-def _qwen_home_from_env_file(file_path: str) -> Optional[str]:
-    raw = read_optional(file_path, "Qwen home environment")
-    if raw is None:
-        return None
-    value = parse_dotenv(raw).get("QWEN_HOME")
-    return value or None
-
-
-def resolve_qwen_home() -> str:
-    initial_value = os.environ.get("QWEN_HOME")
-    initial_home = (
-        _resolve_storage_path(initial_value) if initial_value else _default_qwen_home()
-    )
-    if "QWEN_HOME" in os.environ:
-        return initial_home
-    candidates = (
-        os.path.join(initial_home, ".env"),
-        os.path.join(os.path.dirname(initial_home), ".env"),
-    )
-    for candidate in candidates:
-        discovered = _qwen_home_from_env_file(candidate)
-        if discovered:
-            return _resolve_storage_path(discovered)
-    return initial_home
-
-
-def read_document() -> QwenDocument:
-    config_path = os.path.join(resolve_qwen_home(), "settings.json")
-    raw = read_optional(config_path, "Qwen Code settings")
-    if raw is None:
-        return create_owned_document(config_path)
-    return parse_document(exists=True, file_path=config_path, raw=raw)
-
-
-def rendered_change(rendered: RenderedDocument) -> Optional[FileChange]:
-    if not rendered.changed:
-        return None
-    return source_change(
-        rendered.document.file_path,
-        "Qwen Code settings",
-        rendered.document.raw if rendered.document.exists else None,
-        rendered.next_source,
-        0o600,
-    )
-
-
-def _read_runtime_config(file_path: str) -> Optional[JsonObject]:
-    raw = read_optional(file_path, "Elydora runtime config")
-    if raw is None:
-        return None
     try:
-        value = json.loads(raw)
+        parsed = urllib.parse.urlsplit(value)
-    except json.JSONDecodeError as error:
+        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"Failed to parse Elydora runtime config at {file_path}: {error}"
+            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 not isinstance(value, dict):
+    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 at {file_path} must contain a JSON object"
+            f"Elydora runtime config {field} is invalid: {config_path}"
         )
     return value
 
 
-def runtime_files_exist(contracts: list[RuntimeContract]) -> bool:
+def _validate_runtime_config(
-    for contract in contracts:
+    config: JsonObject, expected_agent_id: str, config_path: str
-        agent_directory = os.path.dirname(contract.guard_path)
+) -> None:
-        runtime_config = _read_runtime_config(
+    supported = {"org_id", "agent_id", "kid", "base_url", "token", "agent_name"}
-            os.path.join(agent_directory, "config.json")
+    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
     )
-        config_agent_id = runtime_config.get("agent_id") if runtime_config else None
     if (
-            runtime_config is None
+        not same_qwen_agent_id(agent_id, expected_agent_id)
-            or runtime_config.get("agent_name") != AGENT_KEY
+        or config.get("agent_name") != AGENT_KEY
-            or not isinstance(config_agent_id, str)
-            or not same_agent_id(config_agent_id, contract.agent_id)
     ):
-            continue
+        raise ValueError(
-        if regular_file_exists(
+            "Elydora runtime identity does not match Qwen Code hooks: "
-            contract.guard_path,
+            f"{config_path}"
-            "Elydora guard runtime",
+        )
-        ) and regular_file_exists(
+    if "token" in config:
-            contract.audit_path,
+        _require_non_empty_string(config.get("token"), "token", config_path)
-            "Elydora audit runtime",
+    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_qwen_agent_id(configured_id, agent_id)
+        or config.get("agent_name") != AGENT_KEY
+    ):
+        raise ValueError(
+            "Elydora runtime config identity does not match Qwen Code agent "
+            f"{agent_id}: {config_path}"
+        )
+
+
+def _runtime_contract_exists(contract: QwenRuntimeContract) -> bool:
+    root = elydora_dir()
+    agent_directory = os.path.dirname(contract.guard_path)
+    if (
+        not same_qwen_path(os.path.dirname(agent_directory), root)
+        or not same_qwen_path(
+            contract.audit_path, os.path.join(agent_directory, "hook.py")
+        )
     ):
-            return True
         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")
+    expected_guard = generate_guard_script(AGENT_KEY, contract.agent_id)
+    expected_audit = generate_hook_script(
+        org_id=str(config["org_id"]),
+        agent_id=contract.agent_id,
+        kid=str(config["kid"]),
+        base_url=str(config["base_url"]),
+        native_payload=True,
+        agent_name=AGENT_KEY,
+    )
+    return guard.contents == expected_guard and audit.contents == expected_audit
 
 
-__all__ = [
+def qwen_runtime_files_exist(
-    "read_document",
+    contracts: List[QwenRuntimeContract],
-    "rendered_change",
+) -> bool:
-    "require_runtime",
+    return any(_runtime_contract_exists(contract) for contract in contracts)
-    "resolve_qwen_home",
-    "runtime_files_exist",
-]

sdks/python/elydora/plugins/qwen_sources.py

@@ -0,0 +1,373 @@
+"""Qwen Code 0.20 settings discovery, trust, and effective controls."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+import os
+import re
+import sys
+import tempfile
+from typing import Dict, List, Optional, Set, Tuple
+
+from ._dotenv import parse_dotenv
+from ._managed_files import MAX_SOURCE_BYTES, FileSnapshot, read_physical_file
+from ._transaction import FilePrecondition
+from .qwen_command import same_qwen_path
+from .qwen_config import (
+    QwenDocument,
+    create_qwen_document,
+    parse_qwen_document,
+    parse_qwen_jsonc_object,
+    qwen_document_label,
+    qwen_source_label,
+)
+from .qwen_contract import validate_javascript_matchers
+
+
+_HOME_ENV_KEYS = ("QWEN_HOME", "QWEN_RUNTIME_DIR")
+
+
+@dataclass(frozen=True)
+class QwenDisableControl:
+    disabled: bool
+    source: Optional[QwenDocument]
+
+
+@dataclass(frozen=True)
+class QwenSources:
+    qwen_home: str
+    system_defaults: QwenDocument
+    user: QwenDocument
+    workspace: QwenDocument
+    system: QwenDocument
+    workspace_active: bool
+    workspace_trusted: bool
+    disable_control: QwenDisableControl
+    preconditions: Tuple[FilePrecondition, ...]
+
+
+@dataclass(frozen=True)
+class _RoutingResult:
+    qwen_home: str
+    preconditions: Tuple[FilePrecondition, ...]
+
+
+def _default_qwen_home() -> str:
+    home = os.path.expanduser("~")
+    if not home or home == "~":
+        home = tempfile.gettempdir()
+    return os.path.join(home, ".qwen")
+
+
+def _resolve_config_path(value: str) -> str:
+    resolved = value
+    if value == "~":
+        resolved = os.path.expanduser("~")
+    elif value.startswith(("~/", "~\\")):
+        segments = [item for item in re.split(r"[/\\]+", value[2:]) if item]
+        resolved = os.path.join(os.path.expanduser("~"), *segments)
+    return resolved if os.path.isabs(resolved) else os.path.abspath(resolved)
+
+
+def _environment_value(key: str) -> Tuple[bool, Optional[str]]:
+    return key in os.environ, os.environ.get(key)
+
+
+def _resolve_qwen_routing() -> _RoutingResult:
+    values: Dict[str, Optional[str]] = {}
+    owned: Set[str] = set()
+    for key in _HOME_ENV_KEYS:
+        present, value = _environment_value(key)
+        values[key] = value
+        if present:
+            owned.add(key)
+    if all(values.get(key) for key in _HOME_ENV_KEYS):
+        return _RoutingResult(
+            _resolve_config_path(values["QWEN_HOME"] or ""), ()
+        )
+
+    initial_qwen_home = values.get("QWEN_HOME")
+    initial_directory = (
+        _resolve_config_path(initial_qwen_home)
+        if initial_qwen_home
+        else _default_qwen_home()
+    )
+    candidates = [os.path.join(initial_directory, ".env")]
+    if not initial_qwen_home:
+        candidates.append(os.path.join(os.path.dirname(initial_directory), ".env"))
+    preconditions: List[FilePrecondition] = []
+    visited: Set[str] = set()
+
+    def read_candidate(file_path: str) -> None:
+        resolved = os.path.abspath(file_path)
+        key = os.path.normcase(resolved)
+        if key in visited:
+            return
+        visited.add(key)
+        snapshot = read_physical_file(
+            resolved, "Qwen Code home environment", MAX_SOURCE_BYTES
+        )
+        preconditions.append(FilePrecondition(
+            resolved,
+            "Qwen Code home environment",
+            snapshot,
+            MAX_SOURCE_BYTES,
+        ))
+        if snapshot is None:
+            return
+        parsed = parse_dotenv(snapshot.contents)
+        for env_key in _HOME_ENV_KEYS:
+            value = parsed.get(env_key)
+            if value and env_key not in owned:
+                values[env_key] = value
+                owned.add(env_key)
+
+    for candidate in candidates:
+        read_candidate(candidate)
+    discovered_home = values.get("QWEN_HOME")
+    if discovered_home and discovered_home != initial_qwen_home:
+        discovered_directory = _resolve_config_path(discovered_home)
+        if not same_qwen_path(discovered_directory, initial_directory):
+            read_candidate(os.path.join(discovered_directory, ".env"))
+    resolved_home = values.get("QWEN_HOME")
+    return _RoutingResult(
+        _resolve_config_path(resolved_home) if resolved_home else _default_qwen_home(),
+        tuple(preconditions),
+    )
+
+
+def resolve_qwen_home() -> str:
+    return _resolve_qwen_routing().qwen_home
+
+
+def _system_settings_path() -> str:
+    configured = os.environ.get("QWEN_CODE_SYSTEM_SETTINGS_PATH")
+    if configured:
+        return os.path.abspath(configured)
+    if sys.platform == "darwin":
+        return "/Library/Application Support/QwenCode/settings.json"
+    if os.name == "nt":
+        return r"C:\ProgramData\qwen-code\settings.json"
+    return "/etc/qwen-code/settings.json"
+
+
+def _system_defaults_path(system_path: str) -> str:
+    configured = os.environ.get("QWEN_CODE_SYSTEM_DEFAULTS_PATH")
+    return (
+        os.path.abspath(configured)
+        if configured
+        else os.path.join(os.path.dirname(system_path), "system-defaults.json")
+    )
+
+
+def _read_document(kind: str, file_path: str) -> QwenDocument:
+    snapshot = read_physical_file(
+        file_path, qwen_source_label(kind), MAX_SOURCE_BYTES
+    )
+    if snapshot is None:
+        return create_qwen_document(kind, file_path)
+    return parse_qwen_document(
+        kind=kind,
+        exists=True,
+        file_path=file_path,
+        raw=snapshot.contents,
+        snapshot=snapshot,
+    )
+
+
+def _canonical_path(file_path: str) -> str:
+    try:
+        return os.path.realpath(file_path)
+    except OSError:
+        return os.path.abspath(file_path)
+
+
+def _comparison_path(file_path: str) -> str:
+    return os.path.normcase(os.path.abspath(file_path))
+
+
+def _is_within(child: str, parent: str) -> bool:
+    try:
+        relative = os.path.relpath(_comparison_path(child), _comparison_path(parent))
+    except ValueError:
+        return False
+    return relative == "." or (
+        relative != ".."
+        and not relative.startswith(".." + os.sep)
+        and not os.path.isabs(relative)
+    )
+
+
+def _workspace_trust(
+    system: QwenDocument,
+    user: QwenDocument,
+    qwen_home: str,
+    workspace_path: str,
+) -> Tuple[bool, Optional[FilePrecondition]]:
+    enabled = (
+        user.folder_trust_enabled
+        if user.folder_trust_enabled is not None
+        else system.folder_trust_enabled
+    )
+    if enabled is not True:
+        return True, None
+    configured = os.environ.get("QWEN_CODE_TRUSTED_FOLDERS_PATH")
+    file_path = (
+        os.path.abspath(configured)
+        if configured
+        else os.path.join(qwen_home, "trustedFolders.json")
+    )
+    snapshot = read_physical_file(
+        file_path, "Qwen Code trusted folders", MAX_SOURCE_BYTES
+    )
+    precondition = FilePrecondition(
+        file_path,
+        "Qwen Code trusted folders",
+        snapshot,
+        MAX_SOURCE_BYTES,
+    )
+    if snapshot is None:
+        return True, precondition
+    rules = parse_qwen_jsonc_object(
+        snapshot.contents, f"Qwen Code trusted folders at {file_path}"
+    )
+    allowed_levels = {"TRUST_FOLDER", "TRUST_PARENT", "DO_NOT_TRUST"}
+    for rule_path, level in rules.items():
+        if level not in allowed_levels:
+            raise ValueError(
+                "Qwen Code trusted folders has invalid trust level for "
+                f'"{rule_path}"'
+            )
+    workspace = _canonical_path(workspace_path)
+    for rule_path, level in rules.items():
+        canonical_rule = _canonical_path(rule_path)
+        trust_root = (
+            os.path.dirname(canonical_rule)
+            if level == "TRUST_PARENT"
+            else canonical_rule
+        )
+        if level in {"TRUST_FOLDER", "TRUST_PARENT"} and _is_within(
+            workspace, trust_root
+        ):
+            return True, precondition
+    for rule_path, level in rules.items():
+        if level == "DO_NOT_TRUST" and _comparison_path(
+            workspace
+        ) == _comparison_path(_canonical_path(rule_path)):
+            return False, precondition
+    return True, precondition
+
+
+def _effective_disable(
+    system_defaults: QwenDocument,
+    user: QwenDocument,
+    workspace: QwenDocument,
+    system: QwenDocument,
+    use_workspace: bool,
+) -> QwenDisableControl:
+    disabled = False
+    source: Optional[QwenDocument] = None
+    documents = [system_defaults, user]
+    if use_workspace:
+        documents.append(workspace)
+    documents.append(system)
+    for document in documents:
+        if document.disable_all_hooks is None:
+            continue
+        disabled = document.disable_all_hooks
+        source = document
+    return QwenDisableControl(disabled, source)
+
+
+def _source_precondition(document: QwenDocument) -> FilePrecondition:
+    return FilePrecondition(
+        document.file_path,
+        qwen_document_label(document),
+        document.snapshot,
+        MAX_SOURCE_BYTES,
+    )
+
+
+def _deduplicate_preconditions(
+    values: List[FilePrecondition],
+) -> Tuple[FilePrecondition, ...]:
+    result: Dict[str, FilePrecondition] = {}
+    for value in values:
+        result.setdefault(_comparison_path(value.file_path), value)
+    return tuple(result.values())
+
+
+def read_qwen_sources() -> QwenSources:
+    routing = _resolve_qwen_routing()
+    system_path = _system_settings_path()
+    workspace_path = os.path.join(os.getcwd(), ".qwen", "settings.json")
+    system = _read_document("system", system_path)
+    system_defaults = _read_document(
+        "system-defaults", _system_defaults_path(system_path)
+    )
+    user = _read_document(
+        "user", os.path.join(routing.qwen_home, "settings.json")
+    )
+    canonical_workspace = _canonical_path(os.getcwd())
+    canonical_home = _canonical_path(os.path.expanduser("~"))
+    workspace_active = _comparison_path(canonical_workspace) != _comparison_path(
+        canonical_home
+    )
+    workspace = (
+        _read_document("workspace", workspace_path)
+        if workspace_active
+        else create_qwen_document("workspace", workspace_path)
+    )
+    if workspace_active:
+        workspace_trusted, trust_precondition = _workspace_trust(
+            system, user, routing.qwen_home, canonical_workspace
+        )
+    else:
+        workspace_trusted, trust_precondition = False, None
+    validate_javascript_matchers([
+        system_defaults.hooks,
+        user.hooks,
+        workspace.hooks,
+        system.hooks,
+    ])
+    preconditions = list(routing.preconditions)
+    preconditions.extend([
+        _source_precondition(system_defaults),
+        _source_precondition(user),
+    ])
+    if workspace_active:
+        preconditions.append(_source_precondition(workspace))
+    preconditions.append(_source_precondition(system))
+    if trust_precondition is not None:
+        preconditions.append(trust_precondition)
+    return QwenSources(
+        qwen_home=routing.qwen_home,
+        system_defaults=system_defaults,
+        user=user,
+        workspace=workspace,
+        system=system,
+        workspace_active=workspace_active,
+        workspace_trusted=workspace_trusted,
+        disable_control=_effective_disable(
+            system_defaults,
+            user,
+            workspace,
+            system,
+            workspace_active and workspace_trusted,
+        ),
+        preconditions=_deduplicate_preconditions(preconditions),
+    )
+
+
+def require_qwen_hooks_enabled(sources: QwenSources) -> None:
+    if not sources.disable_control.disabled:
+        return
+    source = sources.disable_control.source
+    location = (
+        f"{qwen_document_label(source)} at {source.file_path}"
+        if source is not None
+        else "effective settings"
+    )
+    raise ValueError(
+        f"Qwen Code hooks are disabled by disableAllHooks in {location}"
+    )

sdks/python/elydora/plugins/registry.py

@@ -78,7 +78,7 @@ SUPPORTED_AGENTS: Dict[str, Dict[str, str]] = {
     },
     "qwen": {
         "name": "Qwen Code",
-        "hook_event": "PreToolUse/PostToolUse",
+        "hook_event": "PreToolUse/PostToolUse/PostToolUseFailure",
         "config_path": "~/.qwen/settings.json",
     },
 }

sdks/python/tests/qwen_support.py

@@ -1,134 +1,309 @@
 from __future__ import annotations
 
-import base64
+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 typing import Any, Dict, Optional
+from threading import Thread
+from typing import Any, Mapping, Optional
 
 import pytest
 
-from elydora.plugins import qwen
 from elydora.plugins._jsonc import parse_jsonc
-from elydora.plugins.base import InstallConfig
+from elydora.plugins.qwen import QwenPlugin
+from elydora.plugins.qwen_command import build_qwen_command
+from elydora.utils import base64url_encode
 
 
 AGENT_ID = "agent-1"
 MISSING = object()
-JsonObject = Dict[str, Any]
+VALID_PRIVATE_KEY = base64url_encode(bytes(range(32)))
+GUARD_NAME = "elydora-guard"
+AUDIT_NAME = "elydora-audit"
 
 
-def generated_command(script_path: Path) -> str:
+@dataclass
-    if os.name == "nt":
+class QwenFixture:
-        executable = "'" + sys.executable.replace("'", "''") + "'"
+    root_dir: Path
-        script = "'" + str(script_path).replace("'", "''") + "'"
+    home_dir: Path
-        return f"& {executable} {script}; exit $LASTEXITCODE"
+    project_dir: Path
-    executable = "'" + sys.executable.replace("'", "'\"'\"'") + "'"
+    config_path: Path
-    script = "'" + str(script_path).replace("'", "'\"'\"'") + "'"
+    system_defaults_path: Path
-    return f"{executable} {script}"
+    system_path: Path
+    trusted_folders_path: Path
+    agent_dir: Path
+    guard_path: Path
+    audit_path: Path
+    runtime_config_path: Path
+    private_key_path: Path
+    plugin: QwenPlugin
+    config: dict[str, str]
 
+    def install(self) -> None:
+        self.plugin.install(self.config)
 
-def parse_settings(path: Path) -> JsonObject:
+    def source(self, file_path: Optional[Path] = None) -> str:
+        path = file_path or self.config_path
+        with path.open("r", encoding="utf-8", newline="") as file:
+            return file.read()
+
+    def settings(self, file_path: Optional[Path] = None) -> dict[str, Any]:
+        path = file_path or self.config_path
         value = parse_jsonc(
-        path.read_text(encoding="utf-8"),
+            self.source(path),
-        str(path),
+            f"Qwen Code test settings at {path}",
             allow_trailing_commas=False,
         )
         assert isinstance(value, dict)
         return value
 
 
-def managed_handler(
+class ElydoraApiHandler(BaseHTTPRequestHandler):
-    settings: JsonObject,
+    agent_status: Any = "active"
-    event: str,
+    operation_status = 201
-    script_path: Path,
+    requests: list[dict[str, Any]] = []
-) -> Optional[JsonObject]:
-    command = generated_command(script_path)
-    for group in settings.get("hooks", {}).get(event, []):
-        for handler in group["hooks"]:
-            if handler.get("command") == command:
-                return handler
-    return None
 
+    @classmethod
+    def reset(cls, status: Any, operation_status: int) -> None:
+        cls.agent_status = status
+        cls.operation_status = operation_status
+        cls.requests = []
 
-def run_handler(
+    def _respond(self, status: int, value: object) -> None:
-    handler: JsonObject,
+        body = json.dumps(value).encode()
-    payload: str,
+        self.send_response(status)
-) -> subprocess.CompletedProcess[str]:
+        self.send_header("Content-Type", "application/json")
-    if handler["shell"] == "powershell":
+        self.send_header("Content-Length", str(len(body)))
-        command = [
+        self.end_headers()
-            "powershell",
+        self.wfile.write(body)
-            "-NoProfile",
+
-            "-NonInteractive",
+    def do_GET(self) -> None:
-            "-Command",
+        type(self).requests.append({
-            handler["command"],
+            "method": "GET",
-        ]
+            "path": self.path,
-    else:
+            "authorization": self.headers.get("Authorization"),
-        command = ["bash", "-c", handler["command"]]
+        })
-    return subprocess.run(  # nosec B603
+        self._respond(200, {"agent": {"status": type(self).agent_status}})
-        command,
+
-        input=payload,
+    def do_POST(self) -> None:
-        text=True,
+        length = int(self.headers.get("Content-Length", "0"))
-        capture_output=True,
+        raw = self.rfile.read(length)
-        check=False,
+        type(self).requests.append({
-        timeout=10,
+            "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
 
-class QwenFixture:
+
-    def __init__(
+def start_api_server(
-        self,
+    *, 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 prepare_fixture(
     monkeypatch: pytest.MonkeyPatch,
     tmp_path: Path,
     *,
-        existing_settings: Any = MISSING,
+    existing_settings: object = MISSING,
-        create_guard: bool = True,
+    base_url: str = "http://127.0.0.1:9",
-    ) -> None:
+) -> QwenFixture:
-        self.home_dir = tmp_path / "home with spaces and 'quote"
+    home_dir = tmp_path / "home with spaces and 'quote $QWEN_CWD %QWEN_CWD%"
-        self.workspace_dir = self.home_dir / "workspace"
+    project_dir = tmp_path / "project with spaces"
-        self.qwen_dir = self.home_dir / ".qwen"
+    project_dir.mkdir(parents=True)
-        self.config_path = self.qwen_dir / "settings.json"
+    monkeypatch.setenv("HOME", str(home_dir))
-        self.agent_dir = self.home_dir / ".elydora" / AGENT_ID
+    monkeypatch.setenv("USERPROFILE", str(home_dir))
-        self.guard_path = self.agent_dir / "guard.py"
-        self.audit_path = self.agent_dir / "hook.py"
-        self.runtime_config = self.agent_dir / "config.json"
-        self.private_key_path = self.agent_dir / "private.key"
-        self.workspace_dir.mkdir(parents=True)
-        self.agent_dir.mkdir(parents=True)
-        if create_guard:
-            self.guard_path.write_text(
-                "import sys\n"
-                "sys.stdin.buffer.read()\n"
-                "sys.stderr.write('Agent is frozen by Elydora.\\n')\n"
-                "raise SystemExit(2)\n",
-                encoding="utf-8",
-            )
-        if existing_settings is not MISSING:
-            self._write(self.config_path, existing_settings)
-        monkeypatch.setenv("HOME", str(self.home_dir))
-        monkeypatch.setenv("USERPROFILE", str(self.home_dir))
     monkeypatch.delenv("QWEN_HOME", raising=False)
-        monkeypatch.chdir(self.workspace_dir)
+    monkeypatch.delenv("QWEN_RUNTIME_DIR", raising=False)
-        private_key = base64.urlsafe_b64encode(bytes([1]) * 32).rstrip(b"=").decode()
+    monkeypatch.chdir(project_dir)
-        self.config: InstallConfig = {
+    config_path = home_dir / ".qwen" / "settings.json"
+    system_defaults_path = tmp_path / "system-defaults.json"
+    system_path = tmp_path / "system-settings.json"
+    trusted_folders_path = tmp_path / "trusted-folders.json"
+    monkeypatch.setenv(
+        "QWEN_CODE_SYSTEM_DEFAULTS_PATH", str(system_defaults_path)
+    )
+    monkeypatch.setenv("QWEN_CODE_SYSTEM_SETTINGS_PATH", str(system_path))
+    monkeypatch.setenv(
+        "QWEN_CODE_TRUSTED_FOLDERS_PATH", str(trusted_folders_path)
+    )
+    agent_dir = home_dir / ".elydora" / AGENT_ID
+    guard_path = agent_dir / "guard.py"
+    audit_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 QwenFixture(
+        root_dir=tmp_path,
+        home_dir=home_dir,
+        project_dir=project_dir,
+        config_path=config_path,
+        system_defaults_path=system_defaults_path,
+        system_path=system_path,
+        trusted_folders_path=trusted_folders_path,
+        agent_dir=agent_dir,
+        guard_path=guard_path,
+        audit_path=audit_path,
+        runtime_config_path=runtime_config_path,
+        private_key_path=private_key_path,
+        plugin=QwenPlugin(),
+        config={
+            "org_id": "org-1",
             "agent_id": AGENT_ID,
             "agent_name": "qwen",
-            "org_id": "org-1",
+            "private_key": VALID_PRIVATE_KEY,
-            "private_key": private_key,
             "kid": "kid-1",
-            "base_url": "https://api.elydora.test",
+            "token": "token-1",
-            "guard_script_path": str(self.guard_path),
+            "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.get("hooks", {}).get(event, []):
+        if set(group) != {"hooks"}:
+            continue
+        for handler in group["hooks"]:
+            if (
+                set(handler)
+                == {"type", "name", "command", "shell", "timeout"}
+                and handler.get("type") == "command"
+                and handler.get("name") == name
+                and handler.get("shell")
+                == ("powershell" if os.name == "nt" else "bash")
+                and handler.get("timeout") == 10_000
+            ):
+                return handler
+    raise AssertionError(f"managed handler for {event!r} not found")
+
+
+def assert_managed_triple(settings: Mapping[str, Any]) -> dict[str, Any]:
+    guard = managed_handler(settings, "PreToolUse", GUARD_NAME)
+    audit = managed_handler(settings, "PostToolUse", AUDIT_NAME)
+    failure = managed_handler(settings, "PostToolUseFailure", AUDIT_NAME)
+    assert audit["command"] == failure["command"]
+    return {"guard": guard, "audit": audit, "failure": failure}
+
+
+def legacy_group(script_path: Path) -> dict[str, Any]:
+    return {
+        "matcher": "*",
+        "hooks": [{
+            "type": "command",
+            "command": build_qwen_command(str(script_path)),
+            "shell": "powershell" if os.name == "nt" else "bash",
+            "timeout": 10_000,
+        }],
     }
-        self.plugin = qwen.QwenPlugin()
 
-    @staticmethod
-    def _write(path: Path, value: Any) -> None:
-        path.parent.mkdir(parents=True, exist_ok=True)
-        raw = value if isinstance(value, str) else json.dumps(value, indent=2) + "\n"
-        with open(path, "w", encoding="utf-8", newline="") as file:
-            file.write(raw)
 
-    def install(self) -> None:
+def run_handler(
-        self.plugin.install(self.config)
+    handler: Mapping[str, Any], payload: bytes, fixture: QwenFixture
+) -> subprocess.CompletedProcess[bytes]:
+    environment = {
+        **os.environ,
+        "HOME": str(fixture.home_dir),
+        "USERPROFILE": str(fixture.home_dir),
+    }
+    if os.name == "nt":
+        powershell = ntpath.join(
+            os.environ.get("SystemRoot", r"C:\Windows"),
+            "System32",
+            "WindowsPowerShell",
+            "v1.0",
+            "powershell.exe",
+        )
+        command = [
+            powershell,
+            "-NoLogo",
+            "-NoProfile",
+            "-NonInteractive",
+            "-Command",
+            str(handler["command"]),
+        ]
+    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,
+        timeout=15,
+    )
+
+
+def official_input(fixture: QwenFixture, event: str) -> dict[str, Any]:
+    payload: dict[str, Any] = {
+        "session_id": "session-1",
+        "transcript_path": str(fixture.project_dir / "transcript.jsonl"),
+        "cwd": str(fixture.project_dir),
+        "hook_event_name": event,
+        "timestamp": "2026-07-19T00:00:00.000Z",
+        "permission_mode": "default",
+        "tool_name": "run_shell_command",
+        "tool_input": {"command": "echo test"},
+        "tool_use_id": "toolu_1",
+        "tool_call_id": "call_1",
+    }
+    if event == "PostToolUse":
+        payload["tool_response"] = {"output": "test", "error": None}
+    if event == "PostToolUseFailure":
+        payload.update({"error": "command failed", "is_interrupt": False})
+    return payload
+
+
+def run_cli(
+    fixture: QwenFixture, arguments: list[str]
+) -> subprocess.CompletedProcess[str]:
+    environment = dict(os.environ)
+    repository = str(Path(__file__).resolve().parents[1])
+    existing = environment.get("PYTHONPATH")
+    environment["PYTHONPATH"] = (
+        repository + os.pathsep + existing if existing else repository
+    )
+    return subprocess.run(  # nosec B603
+        [sys.executable, "-m", "elydora.cli", *arguments],
+        text=True,
+        capture_output=True,
+        check=False,
+        cwd=fixture.project_dir,
+        env=environment,
+        timeout=30,
+    )

sdks/python/tests/test_qwen_plugin.py

@@ -9,30 +9,41 @@ import pytest
 
 from elydora import cli
 from elydora.plugins import _transaction, qwen
+from elydora.plugins.qwen_config import render_qwen_document
+from elydora.plugins.qwen_contract import (
+    AUDIT_HOOK_NAME,
+    GUARD_HOOK_NAME,
+    build_qwen_group,
+)
+from elydora.plugins.qwen_installation import (
+    commit_qwen_installation,
+    preflight_qwen_installation,
+    prepare_qwen_installation,
+)
+from elydora.plugins.qwen_sources import read_qwen_sources
 from elydora.plugins.registry import SUPPORTED_AGENTS
 from qwen_support import (
     AGENT_ID,
-    QwenFixture,
+    VALID_PRIVATE_KEY,
-    generated_command,
+    assert_managed_triple,
-    managed_handler,
+    legacy_group,
-    parse_settings,
+    prepare_fixture,
-    run_handler,
+    run_cli,
+    write_json,
+    write_text,
 )
 
 
-EXPECTED_SHELL = "powershell" if os.name == "nt" else "bash"
-
-
 def test_qwen_is_registered_in_the_sdk_and_cli() -> None:
     assert SUPPORTED_AGENTS["qwen"] == {
         "name": "Qwen Code",
-        "hook_event": "PreToolUse/PostToolUse",
+        "hook_event": "PreToolUse/PostToolUse/PostToolUseFailure",
         "config_path": "~/.qwen/settings.json",
     }
     assert cli.PLUGIN_MAP["qwen"] is qwen.QwenPlugin
 
 
-def test_install_preserves_jsonc_and_is_idempotent(
+def test_install_preserves_all_sources_and_is_idempotent(
     monkeypatch: pytest.MonkeyPatch,
     tmp_path: Path,
     capsys: pytest.CaptureFixture[str],
@@ -42,208 +53,132 @@ def test_install_preserves_jsonc_and_is_idempotent(
         "  // Keep this user preference.\r\n"
         '  "theme": "GitHub",\r\n'
         '  "hooks": {\r\n'
-        '    "SessionStart": [{ "hooks": [{ "type": "command", "command": "session-hook" }] }],\r\n'
+        '    "FutureEvent": [null],\r\n'
-        '    "PreToolUse": [{ "matcher": "read_file", "hooks": [{ "type": "command", "command": "user-hook" }] }]\r\n'
+        '    "PreToolUse": [{ "matcher": "read_file", "hooks": '
+        '[{ "type": "command", "command": "user-hook" }] }]\r\n'
         "  }\r\n"
         "}\r\n"
     )
-    fixture = QwenFixture(monkeypatch, tmp_path, existing_settings=source)
+    fixture = prepare_fixture(
-    workspace_settings = fixture.workspace_dir / ".qwen" / "settings.json"
+        monkeypatch, tmp_path, existing_settings=source
-    workspace_settings.parent.mkdir(parents=True)
+    )
-    workspace_settings.write_text('{ "owner": "workspace" }\n', encoding="utf-8")
+    workspace_path = fixture.project_dir / ".qwen" / "settings.json"
+    write_json(workspace_path, {"owner": "workspace"})
+    write_json(fixture.system_defaults_path, {"owner": "defaults"})
+    write_json(fixture.system_path, {"owner": "system"})
+
     fixture.install()
-    fixture.install()
+    first_source = fixture.source()
-    output = capsys.readouterr().out
+    first = fixture.settings()
-    with open(fixture.config_path, "r", encoding="utf-8", newline="") as file:
+    assert "Keep this user preference" in first_source
-        raw = file.read()
+    assert "\r\n" in first_source
-    settings = parse_settings(fixture.config_path)
+    assert first["theme"] == "GitHub"
-    assert "Keep this user preference" in raw
+    assert first["hooks"]["FutureEvent"] == [None]
-    assert "\r\n" in raw
+    assert first["hooks"]["PreToolUse"][0]["hooks"][0]["command"] == "user-hook"
-    assert settings["theme"] == "GitHub"
+    assert_managed_triple(first)
-    assert settings["hooks"]["SessionStart"][0]["hooks"][0]["command"] == "session-hook"
+    for file_path in (
-    assert len(settings["hooks"]["PreToolUse"]) == 2
+        fixture.guard_path,
-    assert len(settings["hooks"]["PostToolUse"]) == 1
+        fixture.audit_path,
-    for event, script_path in (
+        fixture.runtime_config_path,
-        ("PreToolUse", fixture.guard_path),
+        fixture.private_key_path,
-        ("PostToolUse", fixture.audit_path),
     ):
-        handler = managed_handler(settings, event, script_path)
+        assert file_path.is_file()
-        assert handler is not None
+
-        assert sorted(handler) == ["command", "shell", "timeout", "type"]
+    fixture.install()
-        assert handler["shell"] == EXPECTED_SHELL
+    assert fixture.source() == first_source
-        assert handler["timeout"] == 10_000
+    assert json.loads(workspace_path.read_text(encoding="utf-8")) == {
-    assert parse_settings(workspace_settings) == {"owner": "workspace"}
+        "owner": "workspace"
-    assert (
+    }
-        json.loads(fixture.runtime_config.read_text(encoding="utf-8"))["agent_name"]
+    assert json.loads(fixture.system_defaults_path.read_text()) == {
-        == "qwen"
+        "owner": "defaults"
-    )
+    }
-    assert fixture.private_key_path.is_file()
+    assert json.loads(fixture.system_path.read_text()) == {"owner": "system"}
-    assert fixture.audit_path.is_file()
+    assert "run /hooks" in capsys.readouterr().out
-    assert "run /hooks" in output
 
 
-def test_qwen_home_uses_official_user_env_precedence(
+def test_qwen_home_uses_complete_official_bootstrap_precedence(
-    monkeypatch: pytest.MonkeyPatch,
+    monkeypatch: pytest.MonkeyPatch, tmp_path: Path
-    tmp_path: Path,
 ) -> None:
-    fixture = QwenFixture(monkeypatch, tmp_path)
+    fixture = prepare_fixture(monkeypatch, tmp_path)
     first_home = tmp_path / "first # qwen home"
     second_home = tmp_path / "second qwen home"
-    fixture.qwen_dir.mkdir(parents=True)
+    write_text(
-    (fixture.qwen_dir / ".env").write_text(
+        fixture.home_dir / ".qwen" / ".env",
         f'export QWEN_HOME = "{first_home}" # selected by Qwen\n',
-        encoding="utf-8",
-    )
-    (fixture.home_dir / ".env").write_text(
-        f"QWEN_HOME={second_home}\n", encoding="utf-8"
     )
+    write_text(fixture.home_dir / ".env", f"QWEN_HOME={second_home}\n")
+    write_text(first_home / ".env", "QWEN_RUNTIME_DIR=runtime\n")
+
     fixture.install()
     selected = first_home / "settings.json"
-    assert managed_handler(parse_settings(selected), "PreToolUse", fixture.guard_path)
+    assert_managed_triple(fixture.settings(selected))
     assert not (second_home / "settings.json").exists()
     assert not fixture.config_path.exists()
     assert str(selected) in fixture.plugin.status()["details"]
 
 
 @pytest.mark.parametrize(
-    ("value", "relative"),
+    ("value", "expected"),
     [
-        ("relative-qwen", "relative-qwen"),
+        ("relative-qwen", "project"),
-        ("~/custom-qwen", "custom-qwen"),
+        ("~/custom-qwen", "home"),
-        ("", ".qwen"),
+        ("", "default"),
     ],
 )
-def test_explicit_qwen_home_supports_relative_tilde_and_empty_values(
+def test_explicit_qwen_home_preserves_process_ownership_semantics(
     monkeypatch: pytest.MonkeyPatch,
     tmp_path: Path,
     value: str,
-    relative: str,
+    expected: str,
 ) -> None:
-    fixture = QwenFixture(monkeypatch, tmp_path / (relative.replace(".", "default")))
+    fixture = prepare_fixture(monkeypatch, tmp_path)
-    fixture.qwen_dir.mkdir(parents=True)
+    ignored = tmp_path / "ignored-home"
-    ignored = tmp_path / "ignored-qwen-home"
+    write_text(
-    (fixture.qwen_dir / ".env").write_text(f"QWEN_HOME={ignored}\n", encoding="utf-8")
+        fixture.home_dir / ".qwen" / ".env", f"QWEN_HOME={ignored}\n"
+    )
     monkeypatch.setenv("QWEN_HOME", value)
     fixture.install()
-    if value.startswith("relative"):
+    paths = {
-        selected = fixture.workspace_dir / relative / "settings.json"
+        "project": fixture.project_dir / "relative-qwen" / "settings.json",
-    else:
+        "home": fixture.home_dir / "custom-qwen" / "settings.json",
-        selected = fixture.home_dir / relative / "settings.json"
+        "default": fixture.config_path,
-    assert "hooks" in parse_settings(selected)
+    }
+    assert_managed_triple(fixture.settings(paths[expected]))
     assert not (ignored / "settings.json").exists()
 
 
-def test_commands_block_and_forward_official_input_byte_for_byte(
+def test_qwen_validates_current_schemas_and_preserves_future_events(
-    monkeypatch: pytest.MonkeyPatch,
+    monkeypatch: pytest.MonkeyPatch, tmp_path: Path
-    tmp_path: Path,
 ) -> None:
-    fixture = QwenFixture(monkeypatch, tmp_path)
+    fixture = prepare_fixture(monkeypatch, tmp_path, existing_settings={
+        "hooks": {
+            "FutureEvent": [None],
+            "MessageDisplay": [{
+                "matcher": "[",
+                "hooks": [{
+                    "type": "http",
+                    "url": "http://127.0.0.1:8080/hook",
+                    "headers": {"Authorization": "Bearer ${TOKEN}"},
+                    "allowedEnvVars": ["TOKEN"],
+                    "timeout": 10,
+                    "once": True,
+                }],
+            }],
+            "Stop": [{
+                "hooks": [{
+                    "type": "prompt",
+                    "prompt": "Evaluate $ARGUMENTS",
+                    "model": "fast",
+                    "timeout": 30,
+                }],
+            }],
+        },
+    })
     fixture.install()
-    capture_path = tmp_path / "captured-event.json"
+    settings = fixture.settings()
-    fixture.audit_path.write_text(
+    assert settings["hooks"]["FutureEvent"] == [None]
-        "import pathlib, sys\n"
+    assert settings["hooks"]["MessageDisplay"][0]["matcher"] == "["
-        f"pathlib.Path({str(capture_path)!r}).write_bytes(sys.stdin.buffer.read())\n",
+    assert settings["hooks"]["Stop"][0]["hooks"][0]["type"] == "prompt"
-        encoding="utf-8",
-    )
-    settings = parse_settings(fixture.config_path)
-    guard = managed_handler(settings, "PreToolUse", fixture.guard_path)
-    audit = managed_handler(settings, "PostToolUse", fixture.audit_path)
-    assert guard is not None and audit is not None
-    payload = json.dumps(
-        {
-            "session_id": "session-1",
-            "transcript_path": "transcript.jsonl",
-            "cwd": str(fixture.workspace_dir),
-            "hook_event_name": "PreToolUse",
-            "timestamp": "2026-07-19T00:00:00.000Z",
-            "tool_name": "run_shell_command",
-            "tool_input": {"command": "echo test"},
-        }
-    )
-    guard_result = run_handler(guard, payload)
-    assert guard_result.returncode == 2
-    assert "Agent is frozen by Elydora" in guard_result.stderr
-    post_payload = payload.replace("PreToolUse", "PostToolUse")
-    audit_result = run_handler(audit, post_payload)
-    assert audit_result.returncode == 0, audit_result.stderr
-    assert capture_path.read_text(encoding="utf-8") == post_payload
-
-
-def test_status_requires_enabled_pair_and_complete_runtime(
-    monkeypatch: pytest.MonkeyPatch,
-    tmp_path: Path,
-) -> None:
-    fixture = QwenFixture(monkeypatch, tmp_path)
-    fixture.install()
-    assert fixture.plugin.status()["installed"] is True
-    settings = parse_settings(fixture.config_path)
-    settings["disableAllHooks"] = True
-    fixture.config_path.write_text(json.dumps(settings), encoding="utf-8")
-    status = fixture.plugin.status()
-    assert status["installed"] is False
-    assert "disabled" in status["details"].lower()
-    settings["disableAllHooks"] = False
-    fixture.config_path.write_text(json.dumps(settings), encoding="utf-8")
-    fixture.guard_path.unlink()
-    status = fixture.plugin.status()
-    assert status["installed"] is False
-    assert "runtime" in status["details"].lower()
-
-
-def test_uninstall_preserves_external_mutations_and_exact_ownership(
-    monkeypatch: pytest.MonkeyPatch,
-    tmp_path: Path,
-) -> None:
-    fixture = QwenFixture(
-        monkeypatch,
-        tmp_path,
-        existing_settings='{"$version":4,"owner":"user"}',
-    )
-    fixture.install()
-    settings = parse_settings(fixture.config_path)
-    managed_group = settings["hooks"]["PreToolUse"][-1]
-    managed_group["hooks"].append({"type": "command", "command": "user-command"})
-    lookalike = fixture.home_dir / ".elydora" / "agent-10" / "guard.py"
-    settings["hooks"]["PreToolUse"].append(
-        {
-            "matcher": "*",
-            "hooks": [
-                {
-                    "type": "command",
-                    "command": generated_command(lookalike),
-                    "shell": EXPECTED_SHELL,
-                    "timeout": 10_000,
-                }
-            ],
-        }
-    )
-    fixture.config_path.write_text(
-        json.dumps(settings, indent=2) + "\n", encoding="utf-8"
-    )
-    before = fixture.config_path.read_text(encoding="utf-8")
-    fixture.plugin.uninstall("other-agent")
-    assert fixture.config_path.read_text(encoding="utf-8") == before
-    uninstall_id = "AGENT-1" if os.name == "nt" else AGENT_ID
-    fixture.plugin.uninstall(uninstall_id)
-    remaining = parse_settings(fixture.config_path)
-    assert remaining["$version"] == 4
-    assert remaining["owner"] == "user"
-    assert remaining["hooks"]["PreToolUse"][0]["hooks"][0]["command"] == "user-command"
-    assert "agent-10" in json.dumps(remaining)
-    assert "PostToolUse" not in remaining.get("hooks", {})
-
-
-def test_uninstall_deletes_an_empty_elydora_owned_settings_file(
-    monkeypatch: pytest.MonkeyPatch,
-    tmp_path: Path,
-) -> None:
-    fixture = QwenFixture(monkeypatch, tmp_path)
-    fixture.install()
-    assert fixture.config_path.read_text(encoding="utf-8").startswith(
-        "// Managed by Elydora"
-    )
-    fixture.plugin.uninstall(AGENT_ID)
-    assert not fixture.config_path.exists()
 
 
 @pytest.mark.parametrize(
@@ -254,81 +189,132 @@ def test_uninstall_deletes_an_empty_elydora_owned_settings_file(
         '{ "owner": true, }',
         '{ "hooks": {}, "hooks": {} }',
         '{ "disableAllHooks": "yes" }',
+        '{ "hooks": null }',
         '{ "hooks": [] }',
-        '{ "hooks": { "UnknownEvent": [] } }',
         '{ "hooks": { "PreToolUse": null } }',
         '{ "hooks": { "PreToolUse": [null] } }',
+        '{ "hooks": { "PreToolUse": [{ "matcher": 1, "hooks": [] }] } }',
         '{ "hooks": { "PreToolUse": [{ "matcher": "[", "hooks": [] }] } }',
-        '{ "hooks": { "PreToolUse": [{ "sequential": "yes", "hooks": [] }] } }',
+        '{ "hooks": { "Notification": [{ "matcher": "[", "hooks": [] }] } }',
-        '{ "hooks": { "PreToolUse": [{ "hooks": null }] } }',
+        '{ "hooks": { "PreToolUse": [{ "sequential": 1, "hooks": [] }] } }',
+        '{ "hooks": { "PreToolUse": [{}] } }',
+        '{ "hooks": { "PreToolUse": [{ "hooks": [null] }] } }',
+        '{ "hooks": { "PreToolUse": [{ "hooks": [{ "type": "function" }] }] } }',
         '{ "hooks": { "PreToolUse": [{ "hooks": [{ "type": "command" }] }] } }',
-        '{ "hooks": { "PreToolUse": [{ "hooks": [{ "type": "http" }] }] }',
+        '{ "hooks": { "PreToolUse": [{ "hooks": [{ "type": "http" }] }] } }',
-        '{ "hooks": { "PreToolUse": [{ "hooks": [{ "type": "function", "command": "x" }] }] } }',
+        '{ "hooks": { "Stop": [{ "hooks": [{ "type": "prompt" }] }] } }',
-        '{ "hooks": { "PreToolUse": [{ "hooks": [{ "type": "command", "command": "x", "timeout": "ten" }] }] } }',
+        '{ "hooks": { "PreToolUse": [{ "hooks": [{ "type": "command", "command": "x", "timeout": -1 }] }] } }',
+        '{ "security": { "folderTrust": { "enabled": "yes" } } }',
     ],
 )
-def test_install_rejects_malformed_settings_before_every_write(
+def test_install_rejects_malformed_sources_before_every_write(
-    monkeypatch: pytest.MonkeyPatch,
+    monkeypatch: pytest.MonkeyPatch, tmp_path: Path, source: str
-    tmp_path: Path,
-    source: str,
 ) -> None:
-    fixture = QwenFixture(monkeypatch, tmp_path, existing_settings=source)
+    fixture = prepare_fixture(
-    with pytest.raises((ValueError, RuntimeError), match="Qwen"):
+        monkeypatch, tmp_path, existing_settings=source
-        fixture.install()
-    assert fixture.config_path.read_text(encoding="utf-8") == source
-    assert not fixture.audit_path.exists()
-    assert not fixture.runtime_config.exists()
-    assert not fixture.private_key_path.exists()
-
-
-def test_install_fails_on_unreadable_env_and_missing_guard(
-    monkeypatch: pytest.MonkeyPatch,
-    tmp_path: Path,
-) -> None:
-    env_fixture = QwenFixture(monkeypatch, tmp_path / "env")
-    (env_fixture.qwen_dir / ".env").mkdir(parents=True)
-    with pytest.raises(OSError, match="Qwen home environment"):
-        env_fixture.install()
-    assert not env_fixture.config_path.exists()
-    assert not env_fixture.runtime_config.exists()
-
-    runtime_fixture = QwenFixture(
-        monkeypatch,
-        tmp_path / "runtime",
-        create_guard=False,
     )
-    with pytest.raises(FileNotFoundError, match="guard runtime is missing"):
+    with pytest.raises((OSError, ValueError), match="Qwen"):
-        runtime_fixture.install()
-    assert not runtime_fixture.config_path.exists()
-    assert not runtime_fixture.runtime_config.exists()
-
-
-def test_status_surfaces_malformed_runtime_metadata_and_cleans_staging(
-    monkeypatch: pytest.MonkeyPatch,
-    tmp_path: Path,
-) -> None:
-    fixture = QwenFixture(monkeypatch, tmp_path)
         fixture.install()
-    assert [
+    assert fixture.source() == source
-        path
+    assert not fixture.agent_dir.exists()
-        for path in fixture.home_dir.rglob("*")
-        if path.suffix in {".tmp", ".rollback"}
-    ] == []
-    fixture.runtime_config.write_text("{ malformed", encoding="utf-8")
-    with pytest.raises(ValueError, match="parse Elydora runtime config"):
-        fixture.plugin.status()
 
 
-def test_transaction_rolls_back_runtime_and_settings(
+def test_install_surfaces_malformed_read_only_sources_and_routing(
-    monkeypatch: pytest.MonkeyPatch,
+    monkeypatch: pytest.MonkeyPatch, tmp_path: Path
-    tmp_path: Path,
 ) -> None:
-    fixture = QwenFixture(
+    system = prepare_fixture(monkeypatch, tmp_path / "system")
+    write_text(system.system_path, "{ malformed")
+    with pytest.raises(ValueError, match="system override"):
+        system.install()
+    assert not system.config_path.exists()
+    assert not system.agent_dir.exists()
+
+    routing = prepare_fixture(monkeypatch, tmp_path / "routing")
+    (routing.home_dir / ".qwen" / ".env").mkdir(parents=True)
+    with pytest.raises(OSError, match="Qwen Code home environment"):
+        routing.install()
+    assert not routing.config_path.exists()
+    assert not routing.agent_dir.exists()
+
+
+def test_disable_precedence_respects_workspace_trust(
+    monkeypatch: pytest.MonkeyPatch, tmp_path: Path
+) -> None:
+    user = prepare_fixture(
         monkeypatch,
-        tmp_path,
+        tmp_path / "user",
-        existing_settings={"owner": "user"},
+        existing_settings={"disableAllHooks": False},
     )
-    original = fixture.config_path.read_text(encoding="utf-8")
+    write_json(user.system_defaults_path, {"disableAllHooks": True})
+    user.install()
+
+    system = prepare_fixture(monkeypatch, tmp_path / "system")
+    write_json(system.system_path, {"disableAllHooks": True})
+    with pytest.raises(ValueError, match="system override"):
+        system.install()
+
+    untrusted = prepare_fixture(
+        monkeypatch,
+        tmp_path / "untrusted",
+        existing_settings={"security": {"folderTrust": {"enabled": True}}},
+    )
+    write_json(
+        untrusted.project_dir / ".qwen" / "settings.json",
+        {"disableAllHooks": True},
+    )
+    write_json(
+        untrusted.trusted_folders_path,
+        {str(untrusted.project_dir): "DO_NOT_TRUST"},
+    )
+    untrusted.install()
+
+    trusted = prepare_fixture(
+        monkeypatch,
+        tmp_path / "trusted",
+        existing_settings={"security": {"folderTrust": {"enabled": True}}},
+    )
+    write_json(
+        trusted.project_dir / ".qwen" / "settings.json",
+        {"disableAllHooks": True},
+    )
+    write_json(
+        trusted.trusted_folders_path,
+        {str(trusted.project_dir): "TRUST_FOLDER"},
+    )
+    with pytest.raises(ValueError, match="workspace settings"):
+        trusted.install()
+
+
+def test_transaction_aborts_when_read_only_source_changes(
+    monkeypatch: pytest.MonkeyPatch, tmp_path: Path
+) -> None:
+    fixture = prepare_fixture(monkeypatch, tmp_path)
+    write_json(fixture.system_path, {"owner": "before"})
+    sources = read_qwen_sources()
+    paths = preflight_qwen_installation(fixture.config, sources)
+    rendered = render_qwen_document(sources.user, None, {
+        "PreToolUse": build_qwen_group(paths.guard_path, GUARD_HOOK_NAME),
+        "PostToolUse": build_qwen_group(paths.audit_path, AUDIT_HOOK_NAME),
+        "PostToolUseFailure": build_qwen_group(
+            paths.audit_path, AUDIT_HOOK_NAME
+        ),
+    })
+    changes = prepare_qwen_installation(fixture.config, paths, rendered)
+    write_json(fixture.system_path, {"owner": "after"})
+    with pytest.raises(OSError, match="system override settings changed"):
+        commit_qwen_installation(changes, sources)
+    assert json.loads(fixture.system_path.read_text()) == {"owner": "after"}
+    assert not fixture.config_path.exists()
+    assert not fixture.agent_dir.exists()
+
+
+def test_transaction_rolls_back_every_runtime_and_settings_change(
+    monkeypatch: pytest.MonkeyPatch, tmp_path: Path
+) -> None:
+    fixture = prepare_fixture(
+        monkeypatch, tmp_path, existing_settings={"owner": "user"}
+    )
+    original = fixture.source()
     real_replace = _transaction.os.replace
     failed = False
 
@@ -340,13 +326,17 @@ def test_transaction_rolls_back_runtime_and_settings(
         real_replace(source, destination)
 
     monkeypatch.setattr(_transaction.os, "replace", fail_settings_commit)
-    with pytest.raises(OSError, match="Write Qwen Code installation"):
+    with pytest.raises(OSError, match="Install Qwen Code hooks"):
         fixture.install()
-    assert failed is True
+    assert failed
-    assert fixture.config_path.read_text(encoding="utf-8") == original
+    assert fixture.source() == original
-    assert not fixture.audit_path.exists()
+    for file_path in (
-    assert not fixture.runtime_config.exists()
+        fixture.guard_path,
-    assert not fixture.private_key_path.exists()
+        fixture.audit_path,
+        fixture.runtime_config_path,
+        fixture.private_key_path,
+    ):
+        assert not file_path.exists()
     assert [
         path
         for path in fixture.home_dir.rglob("*")
@@ -354,12 +344,135 @@ def test_transaction_rolls_back_runtime_and_settings(
     ] == []
 
 
-def test_install_rejects_invalid_agent_id_before_writes(
+def test_install_migrates_legacy_and_uninstall_preserves_lookalikes(
-    monkeypatch: pytest.MonkeyPatch,
+    monkeypatch: pytest.MonkeyPatch, tmp_path: Path
-    tmp_path: Path,
 ) -> None:
-    fixture = QwenFixture(monkeypatch, tmp_path)
+    fixture = prepare_fixture(monkeypatch, tmp_path)
-    fixture.config["agent_id"] = "../escape"
+    lookalike = legacy_group(fixture.guard_path)
-    with pytest.raises(ValueError, match="single non-empty path segment"):
+    lookalike["hooks"][0]["timeout"] = 9_000
+    write_json(fixture.config_path, {
+        "owner": "user",
+        "hooks": {
+            "PreToolUse": [legacy_group(fixture.guard_path), lookalike],
+            "PostToolUse": [legacy_group(fixture.audit_path)],
+        },
+    })
     fixture.install()
+    settings = fixture.settings()
+    assert_managed_triple(settings)
+    managed = settings["hooks"]["PreToolUse"][-1]
+    managed["userField"] = "preserve-group"
+    managed["hooks"].append({"type": "command", "command": "user-command"})
+    write_json(fixture.config_path, settings)
+
+    fixture.plugin.uninstall(AGENT_ID)
+    remaining = fixture.settings()
+    assert remaining["owner"] == "user"
+    assert any(
+        group["hooks"][0].get("timeout") == 9_000
+        for group in remaining["hooks"]["PreToolUse"]
+    )
+    assert any(
+        handler.get("command") == "user-command"
+        for group in remaining["hooks"]["PreToolUse"]
+        for handler in group["hooks"]
+    )
+    assert "PostToolUse" not in remaining["hooks"]
+    assert "PostToolUseFailure" not in remaining["hooks"]
+
+
+def test_uninstall_removes_owned_file_and_preserves_user_settings(
+    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()
+    assert owned.source().startswith("// Managed by Elydora")
+    owned.plugin.uninstall(AGENT_ID)
+    assert not owned.config_path.exists()
+
+
+def test_status_requires_exact_contract_and_runtime_identity(
+    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"]["PostToolUseFailure"].append(
+        settings["hooks"]["PostToolUseFailure"][-1]
+    )
+    write_json(fixture.config_path, settings)
+    assert fixture.plugin.status()["installed"] is False
+
+    fixture.install()
+    write_text(fixture.private_key_path, "invalid")
+    with pytest.raises(ValueError, match="private key"):
+        fixture.plugin.status()
+    fixture.install()
+    write_text(fixture.guard_path, "tampered")
+    assert fixture.plugin.status()["installed"] is False
+    fixture.install()
+    runtime = json.loads(fixture.runtime_config_path.read_text())
+    runtime["agent_name"] = "other"
+    write_json(fixture.runtime_config_path, runtime)
+    with pytest.raises(ValueError, match="runtime identity"):
+        fixture.plugin.status()
+
+
+def test_preflight_rejects_unverifiable_runtime_and_invalid_agent_id(
+    monkeypatch: pytest.MonkeyPatch, tmp_path: Path
+) -> None:
+    identity = prepare_fixture(monkeypatch, tmp_path / "identity")
+    write_text(identity.audit_path, "unverified")
+    with pytest.raises(ValueError, match="without config.json"):
+        identity.install()
+    assert not identity.config_path.exists()
+
+    invalid = prepare_fixture(monkeypatch, tmp_path / "invalid")
+    invalid.config["agent_id"] = "../escape"
+    with pytest.raises(ValueError, match="Invalid agent ID"):
+        invalid.install()
+    assert not invalid.config_path.exists()
+
+
+def test_qwen_cli_install_status_and_uninstall_end_to_end(
+    monkeypatch: pytest.MonkeyPatch, tmp_path: Path
+) -> None:
+    fixture = prepare_fixture(monkeypatch, tmp_path)
+    key_file = tmp_path / "install-private.key"
+    token_file = tmp_path / "install-token.txt"
+    write_text(key_file, VALID_PRIVATE_KEY + "\n")
+    write_text(token_file, "token-1\n")
+    install = run_cli(fixture, [
+        "install",
+        "--agent", "qwen",
+        "--org_id", "org-1",
+        "--agent_id", AGENT_ID,
+        "--kid", "kid-1",
+        "--private_key_file", str(key_file),
+        "--token_file", str(token_file),
+        "--base_url", "http://127.0.0.1:9",
+    ])
+    assert install.returncode == 0, install.stderr
+    assert_managed_triple(fixture.settings())
+    status = run_cli(fixture, ["status"])
+    assert status.returncode == 0, status.stderr
+    assert "Qwen Code" in status.stdout and "[installed]" in status.stdout
+    uninstall = run_cli(fixture, [
+        "uninstall", "--agent", "qwen", "--agent_id", AGENT_ID
+    ])
+    assert uninstall.returncode == 0, uninstall.stderr
     assert not fixture.config_path.exists()
+    assert not fixture.agent_dir.exists()

sdks/python/tests/test_qwen_runtime.py

@@ -0,0 +1,141 @@
+from __future__ import annotations
+
+import json
+import os
+from pathlib import Path
+import stat
+
+import pytest
+
+from qwen_support import (
+    ElydoraApiHandler,
+    assert_managed_triple,
+    official_input,
+    prepare_fixture,
+    run_handler,
+    server_base_url,
+    start_api_server,
+)
+
+
+def _close_server(server: object) -> None:
+    server.shutdown()  # type: ignore[attr-defined]
+    server.server_close()  # type: ignore[attr-defined]
+
+
+def test_qwen_guard_accepts_active_agents_with_native_exit_semantics(
+    monkeypatch: pytest.MonkeyPatch, tmp_path: Path
+) -> None:
+    server = start_api_server(status="active")
+    fixture = prepare_fixture(
+        monkeypatch, tmp_path, base_url=server_base_url(server)
+    )
+    try:
+        fixture.install()
+        guard = assert_managed_triple(fixture.settings())["guard"]
+        payload = json.dumps(
+            official_input(fixture, "PreToolUse")
+        ).encode()
+        result = run_handler(guard, payload, fixture)
+        assert result.returncode == 0, result.stderr.decode()
+        assert result.stdout == b""
+        assert len([
+            request
+            for request in ElydoraApiHandler.requests
+            if request["method"] == "GET"
+        ]) == 1
+    finally:
+        _close_server(server)
+
+
+@pytest.mark.parametrize("status", ["frozen", "revoked"])
+def test_qwen_guard_propagates_blocking_states_through_exit_code_two(
+    monkeypatch: pytest.MonkeyPatch, tmp_path: Path, status: str
+) -> None:
+    server = start_api_server(status=status)
+    fixture = prepare_fixture(
+        monkeypatch, tmp_path, base_url=server_base_url(server)
+    )
+    try:
+        fixture.install()
+        guard = assert_managed_triple(fixture.settings())["guard"]
+        result = run_handler(
+            guard,
+            json.dumps(official_input(fixture, "PreToolUse")).encode(),
+            fixture,
+        )
+        assert result.returncode == 2
+        assert result.stdout == b""
+        assert status.encode() in result.stderr.lower()
+    finally:
+        _close_server(server)
+
+
+def test_qwen_audit_forwards_complete_success_and_failure_payloads(
+    monkeypatch: pytest.MonkeyPatch, tmp_path: Path
+) -> None:
+    server = start_api_server()
+    fixture = prepare_fixture(
+        monkeypatch, tmp_path, base_url=server_base_url(server)
+    )
+    try:
+        fixture.install()
+        handlers = assert_managed_triple(fixture.settings())
+        expected = []
+        for handler_name, event in (
+            ("audit", "PostToolUse"),
+            ("failure", "PostToolUseFailure"),
+        ):
+            payload = official_input(fixture, event)
+            payload["future_provider_field"] = {"preserved": True}
+            expected.append(payload)
+            result = run_handler(
+                handlers[handler_name], json.dumps(payload).encode(), fixture
+            )
+            assert result.returncode == 0, result.stderr.decode()
+            assert result.stdout == b""
+        posts = [
+            request
+            for request in ElydoraApiHandler.requests
+            if request["method"] == "POST"
+        ]
+        assert len(posts) == 2
+        assert posts[0]["json"]["payload"] == expected[0]
+        assert posts[1]["json"]["payload"] == expected[1]
+    finally:
+        _close_server(server)
+
+
+def test_qwen_audit_keeps_delivery_failures_observable_and_fail_open(
+    monkeypatch: pytest.MonkeyPatch, tmp_path: Path
+) -> None:
+    fixture = prepare_fixture(monkeypatch, tmp_path)
+    fixture.install()
+    audit = assert_managed_triple(fixture.settings())["audit"]
+    result = run_handler(
+        audit,
+        json.dumps(official_input(fixture, "PostToolUse")).encode(),
+        fixture,
+    )
+    assert result.returncode == 0, result.stderr.decode()
+    assert result.stdout == b""
+    error_log = (fixture.agent_dir / "error.log").read_text(encoding="utf-8").lower()
+    assert "elydora-hook" in error_log and "urlerror" in error_log
+
+
+@pytest.mark.skipif(
+    os.name == "nt", reason="POSIX mode bits are not authoritative on Windows"
+)
+def test_qwen_runtime_artifacts_use_private_modes(
+    monkeypatch: pytest.MonkeyPatch, tmp_path: Path
+) -> None:
+    fixture = prepare_fixture(monkeypatch, tmp_path)
+    fixture.install()
+    for file_path, expected in (
+        (fixture.runtime_config_path, 0o600),
+        (fixture.private_key_path, 0o600),
+        (fixture.guard_path, 0o700),
+        (fixture.audit_path, 0o700),
+        (fixture.config_path, 0o600),
+    ):
+        assert stat.S_IMODE(file_path.stat().st_mode) == expected