From e19c33fc2dee8b7e4f4c7c4207427957f3cdaa5f Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E9=9B=B7=E7=94=B5=E8=8A=BD=E8=A1=A3?= Date: Sun, 19 Jul 2026 12:48:35 -0400 Subject: [PATCH] feat: sync Python Qwen Code hook integration --- integrations/catalog.json | 2 +- sdks/python/README.md | 3 + sdks/python/elydora/cli.py | 2 + sdks/python/elydora/plugins/_dotenv.py | 32 ++ .../plugins/{droid_jsonc.py => _jsonc.py} | 46 +- sdks/python/elydora/plugins/_transaction.py | 273 ++++++++++++ sdks/python/elydora/plugins/droid.py | 5 +- sdks/python/elydora/plugins/droid_config.py | 2 +- sdks/python/elydora/plugins/droid_io.py | 273 +----------- sdks/python/elydora/plugins/qwen.py | 164 +++++++ sdks/python/elydora/plugins/qwen_config.py | 178 ++++++++ sdks/python/elydora/plugins/qwen_contract.py | 409 ++++++++++++++++++ sdks/python/elydora/plugins/qwen_io.py | 139 ++++++ sdks/python/elydora/plugins/registry.py | 5 + sdks/python/tests/qwen_support.py | 134 ++++++ sdks/python/tests/test_droid_plugin.py | 4 +- sdks/python/tests/test_qwen_plugin.py | 365 ++++++++++++++++ 17 files changed, 1761 insertions(+), 275 deletions(-) create mode 100644 sdks/python/elydora/plugins/_dotenv.py rename sdks/python/elydora/plugins/{droid_jsonc.py => _jsonc.py} (92%) create mode 100644 sdks/python/elydora/plugins/_transaction.py create mode 100644 sdks/python/elydora/plugins/qwen.py create mode 100644 sdks/python/elydora/plugins/qwen_config.py create mode 100644 sdks/python/elydora/plugins/qwen_contract.py create mode 100644 sdks/python/elydora/plugins/qwen_io.py create mode 100644 sdks/python/tests/qwen_support.py create mode 100644 sdks/python/tests/test_qwen_plugin.py diff --git a/integrations/catalog.json b/integrations/catalog.json index 4adc4ce..8819edf 100644 --- a/integrations/catalog.json +++ b/integrations/catalog.json @@ -479,7 +479,7 @@ "mechanism": "exit_code_2", "failure_mode": "fail_open" }, - "adapters": { "node": true, "python": false, "go": false }, + "adapters": { "node": true, "python": true, "go": false }, "delivery_state": "partial", "source_url": "https://qwenlm.github.io/qwen-code-docs/en/users/features/hooks/" } diff --git a/sdks/python/README.md b/sdks/python/README.md index e519df7..96bc22d 100644 --- a/sdks/python/README.md +++ b/sdks/python/README.md @@ -85,6 +85,8 @@ Auggie installation writes user-level hooks to `~/.augment/settings.json` and cr Factory Droid installation follows its active user source per event: `~/.factory/hooks.json`, the legacy `~/.factory/hooks/hooks.json` fallback, or hooks nested in `~/.factory/settings.json`. Elydora preserves JSONC comments and formatting, leaves project and organization hooks unchanged, and runs both managed commands with absolute Python and script paths. Run `/hooks` after installation to review the changes. +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. + Kiro CLI installation covers both runtime contracts. Kiro CLI v2 uses the generated custom agent through `kiro-cli --agent elydora-audit`. Kiro CLI v3 loads the global standalone hooks when started with `kiro-cli --v3`. ### Commands @@ -114,6 +116,7 @@ Kiro CLI installation covers both runtime contracts. Kiro CLI v2 uses the genera | Kiro IDE | `kiroide` | | Letta Code | `letta` | | OpenCode | `opencode` | +| Qwen Code | `qwen` | ## API Reference diff --git a/sdks/python/elydora/cli.py b/sdks/python/elydora/cli.py index f8cf5f2..8b1d194 100644 --- a/sdks/python/elydora/cli.py +++ b/sdks/python/elydora/cli.py @@ -32,6 +32,7 @@ from .plugins.kiroide import KiroIdePlugin from .plugins.kimi import KimiPlugin from .plugins.letta import LettaPlugin from .plugins.opencode import OpenCodePlugin +from .plugins.qwen import QwenPlugin PLUGIN_MAP: Dict[str, Type[AgentPlugin]] = { @@ -49,6 +50,7 @@ PLUGIN_MAP: Dict[str, Type[AgentPlugin]] = { "kimi": KimiPlugin, "letta": LettaPlugin, "opencode": OpenCodePlugin, + "qwen": QwenPlugin, } diff --git a/sdks/python/elydora/plugins/_dotenv.py b/sdks/python/elydora/plugins/_dotenv.py new file mode 100644 index 0000000..3584ac6 --- /dev/null +++ b/sdks/python/elydora/plugins/_dotenv.py @@ -0,0 +1,32 @@ +"""Read-only parser for the dotenv grammar used by Qwen Code.""" + +from __future__ import annotations + +import re +from typing import Dict + + +_LINE = re.compile( + r"^\s*(?:export\s+)?([\w.-]+)" + r"(?:\s*=\s*?|:\s+?)" + r"(\s*'(?:\\'|[^'])*'|\s*\"(?:\\\"|[^\"])*\"|" + r"\s*`(?:\\`|[^`])*`|[^#\r\n]+)?" + r"\s*(?:#.*)?$", + re.ASCII | re.MULTILINE, +) + + +def parse_dotenv(source: str) -> Dict[str, str]: + """Parse values with the same read-only rules as Node dotenv 17.""" + values: Dict[str, str] = {} + normalized = source.replace("\r\n", "\n").replace("\r", "\n") + for match in _LINE.finditer(normalized): + key = match.group(1) + value = (match.group(2) or "").strip() + quote = value[0] if value else "" + if len(value) >= 2 and quote in {"'", '"', "`"} and value[-1] == quote: + value = value[1:-1] + if quote == '"': + value = value.replace(r"\n", "\n").replace(r"\r", "\r") + values[key] = value + return values diff --git a/sdks/python/elydora/plugins/droid_jsonc.py b/sdks/python/elydora/plugins/_jsonc.py similarity index 92% rename from sdks/python/elydora/plugins/droid_jsonc.py rename to sdks/python/elydora/plugins/_jsonc.py index 93e5ccd..1d388e3 100644 --- a/sdks/python/elydora/plugins/droid_jsonc.py +++ b/sdks/python/elydora/plugins/_jsonc.py @@ -1,4 +1,4 @@ -"""Strict JSONC parsing and source-preserving edits for Factory Droid.""" +"""Strict JSON-with-comments parsing and source-preserving edits.""" from __future__ import annotations @@ -39,9 +39,15 @@ class JsoncNode: class _Parser: - def __init__(self, raw: str, label: str) -> None: + def __init__( + self, + raw: str, + label: str, + allow_trailing_commas: bool, + ) -> None: self.raw = raw self.label = label + self.allow_trailing_commas = allow_trailing_commas self.position = 0 def _error(self, message: str, position: Optional[int] = None) -> ValueError: @@ -131,6 +137,8 @@ class _Parser: key, member_start, member_end, node, comma )) if self.position < len(self.raw) and self.raw[self.position] == "}": + if not self.allow_trailing_commas: + raise self._error("trailing commas are not allowed", comma) self.position += 1 return JsoncNode( "object", start, self.position, value, tuple(members) @@ -162,6 +170,8 @@ class _Parser: self._skip_trivia() items.append(JsoncItem(node, comma)) if self.position < len(self.raw) and self.raw[self.position] == "]": + if not self.allow_trailing_commas: + raise self._error("trailing commas are not allowed", comma) self.position += 1 return JsoncNode( "array", start, self.position, values, items=tuple(items) @@ -208,17 +218,29 @@ class _Parser: return node -def parse_jsonc(raw: str, label: str) -> Any: - """Parse strict JSON plus comments and trailing commas.""" - return _Parser(raw, label).parse().value +def parse_jsonc( + raw: str, + label: str, + *, + allow_trailing_commas: bool = True, +) -> Any: + """Parse JSON with comments and an explicit trailing-comma policy.""" + return _Parser(raw, label, allow_trailing_commas).parse().value class JsoncEditor: """Apply narrow path edits while retaining untouched source bytes.""" - def __init__(self, raw: str, label: str) -> None: + def __init__( + self, + raw: str, + label: str, + *, + allow_trailing_commas: bool = True, + ) -> None: self.raw = raw self.label = label + self.allow_trailing_commas = allow_trailing_commas self._refresh() @property @@ -226,7 +248,11 @@ class JsoncEditor: return self.root.value def _refresh(self) -> None: - self.root = _Parser(self.raw, self.label).parse() + self.root = _Parser( + self.raw, + self.label, + self.allow_trailing_commas, + ).parse() def _resolve(self, path: Sequence[JsonPathPart]) -> JsoncNode: node = self.root @@ -329,12 +355,12 @@ class JsoncEditor: pretty = "\n" in self.raw or "\r" in self.raw ranges: List[Tuple[int, int, str]] = [] has_existing = bool(parent.members if parent.kind == "object" else parent.items) + previous_end: Optional[int] = None if has_existing and not had_trailing_comma: if parent.kind == "object": previous_end = parent.members[-1].end else: previous_end = parent.items[-1].node.end - ranges.append((previous_end, previous_end, ",")) if pretty: close_line_start = self.raw.rfind("\n", 0, close) + 1 close_prefix = self.raw[close_line_start:close] @@ -359,6 +385,10 @@ class JsoncEditor: if had_trailing_comma: content += "," insertion = (" " if has_existing else "") + content + if previous_end == insertion_at: + insertion = "," + insertion + elif previous_end is not None: + ranges.append((previous_end, previous_end, ",")) ranges.append((insertion_at, insertion_at, insertion)) self._apply(ranges) diff --git a/sdks/python/elydora/plugins/_transaction.py b/sdks/python/elydora/plugins/_transaction.py new file mode 100644 index 0000000..b3a8486 --- /dev/null +++ b/sdks/python/elydora/plugins/_transaction.py @@ -0,0 +1,273 @@ +"""Fail-fast transactional file changes shared by hook adapters.""" + +from __future__ import annotations + +from dataclasses import dataclass +import os +import stat +import tempfile +from typing import List, Optional, Sequence +from uuid import uuid4 + + +@dataclass(frozen=True) +class FileChange: + file_path: str + label: str + original: Optional[str] + next_source: Optional[str] + mode: int + original_mode: Optional[int] + + +@dataclass +class _StagedChange: + change: FileChange + temporary_path: Optional[str] + rollback_path: Optional[str] + committed: bool = False + + +def read_optional(file_path: str, label: str) -> Optional[str]: + try: + with open(file_path, "r", encoding="utf-8", newline="") as file: + return file.read() + except FileNotFoundError: + return None + except OSError as error: + raise OSError(f"Read {label} at {file_path}: {error}") from error + + +def _existing_mode(file_path: str) -> Optional[int]: + try: + return stat.S_IMODE(os.stat(file_path).st_mode) + except FileNotFoundError: + return None + except OSError as error: + raise OSError(f"Read file mode at {file_path}: {error}") from error + + +def source_change( + file_path: str, + label: str, + original: Optional[str], + next_source: Optional[str], + mode: int, +) -> Optional[FileChange]: + if original == next_source: + return None + return FileChange( + file_path, + label, + original, + next_source, + mode, + _existing_mode(file_path), + ) + + +def file_change( + file_path: str, + label: str, + next_source: Optional[str], + mode: int, +) -> Optional[FileChange]: + return source_change( + file_path, + label, + read_optional(file_path, label), + next_source, + mode, + ) + + +def _remove_optional(file_path: Optional[str]) -> None: + if not file_path: + return + try: + os.remove(file_path) + except FileNotFoundError: + return + + +def _cleanup_paths(paths: Sequence[Optional[str]]) -> List[str]: + errors = [] + for path in paths: + try: + _remove_optional(path) + except OSError as error: + errors.append(str(error)) + return errors + + +def _write_staged( + directory: str, + basename: str, + suffix: str, + content: str, + mode: int, + label: str, +) -> str: + descriptor = -1 + temporary_path = "" + try: + descriptor, temporary_path = tempfile.mkstemp( + prefix=f".{basename}.", + suffix=suffix, + dir=directory, + text=True, + ) + with os.fdopen(descriptor, "w", encoding="utf-8", newline="") as file: + descriptor = -1 + file.write(content) + file.flush() + os.fsync(file.fileno()) + os.chmod(temporary_path, mode) + return temporary_path + except Exception as error: + failures = [] + if descriptor >= 0: + try: + os.close(descriptor) + except OSError as close_error: + failures.append(f"close failed: {close_error}") + failures.extend(_cleanup_paths((temporary_path,))) + suffix = f"; cleanup failed: {'; '.join(failures)}" if failures else "" + raise OSError(f"Stage {label}: {error}{suffix}") from error + + +def _assert_unchanged(change: FileChange) -> None: + current = read_optional(change.file_path, change.label) + if current != change.original: + raise OSError(f"{change.label} changed during installation: {change.file_path}") + + +def _stage(change: FileChange) -> _StagedChange: + _assert_unchanged(change) + directory = os.path.dirname(change.file_path) + try: + os.makedirs(directory, mode=0o700, exist_ok=True) + except OSError as error: + raise OSError( + f"Create directory for {change.label} at {directory}: {error}" + ) from error + basename = os.path.basename(change.file_path) + temporary_path = None + rollback_path = None + try: + if change.next_source is not None: + temporary_path = _write_staged( + directory, + basename, + ".tmp", + change.next_source, + change.mode, + change.label, + ) + if change.original is not None and change.next_source is not None: + rollback_path = _write_staged( + directory, + basename, + ".rollback", + change.original, + change.original_mode or change.mode, + f"{change.label} rollback", + ) + elif change.original is not None: + rollback_path = os.path.join( + directory, + f".{basename}.{uuid4().hex}.rollback", + ) + if os.path.lexists(rollback_path): + raise FileExistsError(f"Rollback path already exists: {rollback_path}") + return _StagedChange(change, temporary_path, rollback_path) + except Exception as error: + failures = _cleanup_paths((temporary_path, rollback_path)) + suffix = f"; cleanup failed: {'; '.join(failures)}" if failures else "" + raise OSError(f"Stage {change.label}: {error}{suffix}") from error + + +def _commit(staged: _StagedChange) -> None: + _assert_unchanged(staged.change) + if staged.change.next_source is None: + if staged.rollback_path is None: + raise OSError(f"Missing rollback path for {staged.change.label}") + os.replace(staged.change.file_path, staged.rollback_path) + else: + if staged.temporary_path is None: + raise OSError(f"Missing staged file for {staged.change.label}") + os.replace(staged.temporary_path, staged.change.file_path) + staged.committed = True + + +def _rollback(staged: _StagedChange) -> None: + if not staged.committed: + return + if staged.change.original is None: + _remove_optional(staged.change.file_path) + return + if staged.rollback_path is None: + raise OSError(f"Missing rollback data for {staged.change.label}") + os.replace(staged.rollback_path, staged.change.file_path) + + +def _cleanup(staged: _StagedChange) -> None: + _remove_optional(staged.temporary_path) + _remove_optional(staged.rollback_path) + + +def write_changes(changes: Sequence[FileChange], label: str) -> None: + filtered = [change for change in changes if change.original != change.next_source] + if not filtered: + return + paths = [os.path.normcase(os.path.abspath(change.file_path)) for change in filtered] + if len(paths) != len(set(paths)): + raise ValueError(f"{label} contains duplicate file targets") + staged: List[_StagedChange] = [] + try: + for change in filtered: + staged.append(_stage(change)) + for item in staged: + _commit(item) + except Exception as error: + recovery_errors = [] + for item in reversed(staged): + try: + _rollback(item) + except OSError as rollback_error: + recovery_errors.append(str(rollback_error)) + for item in staged: + try: + _cleanup(item) + except OSError as cleanup_error: + recovery_errors.append(str(cleanup_error)) + suffix = ( + f"; recovery failed: {'; '.join(recovery_errors)}" + if recovery_errors + else "" + ) + raise OSError(f"{label}: {error}{suffix}") from error + cleanup_errors = [] + for item in staged: + try: + _cleanup(item) + except OSError as error: + cleanup_errors.append(str(error)) + if cleanup_errors: + raise OSError(f"{label} cleanup failed: {'; '.join(cleanup_errors)}") + + +def regular_file_exists(file_path: str, label: str) -> bool: + try: + return stat.S_ISREG(os.stat(file_path).st_mode) + except FileNotFoundError: + return False + except OSError as error: + raise OSError(f"Read {label} at {file_path}: {error}") from error + + +def require_runtime(file_path: str, label: str) -> None: + if not file_path: + raise ValueError(f"{label} path is required") + if not regular_file_exists(file_path, label): + raise FileNotFoundError(f"{label} is missing: {file_path}") diff --git a/sdks/python/elydora/plugins/droid.py b/sdks/python/elydora/plugins/droid.py index 9cc2950..a32affb 100644 --- a/sdks/python/elydora/plugins/droid.py +++ b/sdks/python/elydora/plugins/droid.py @@ -6,6 +6,7 @@ import json import os from typing import List, Optional +from ._transaction import FileChange, file_change, require_runtime, write_changes from .base import AgentPlugin, InstallConfig, PluginStatus from .droid_config import ( additions_for, @@ -25,14 +26,10 @@ from .droid_contract import ( validate_javascript_regexes, ) from .droid_io import ( - FileChange, display_config_path, - file_change, read_sources, rendered_change, - require_runtime, runtime_files_exist, - write_changes, ) from .hook_template import generate_hook_script diff --git a/sdks/python/elydora/plugins/droid_config.py b/sdks/python/elydora/plugins/droid_config.py index 42bc73d..8115a32 100644 --- a/sdks/python/elydora/plugins/droid_config.py +++ b/sdks/python/elydora/plugins/droid_config.py @@ -13,7 +13,7 @@ from .droid_contract import ( managed_removals, read_hook_settings, ) -from .droid_jsonc import JsonPathPart, JsoncEditor +from ._jsonc import JsonPathPart, JsoncEditor OWNED_FILE_MARKER = "// Managed by Elydora" diff --git a/sdks/python/elydora/plugins/droid_io.py b/sdks/python/elydora/plugins/droid_io.py index 9cd39f5..a1fb81d 100644 --- a/sdks/python/elydora/plugins/droid_io.py +++ b/sdks/python/elydora/plugins/droid_io.py @@ -1,15 +1,17 @@ -"""Factory Droid source discovery and transactional multi-file I/O.""" +"""Factory Droid source discovery and runtime inspection.""" from __future__ import annotations -from dataclasses import dataclass import json import os -import stat -import tempfile -from typing import List, Optional, Sequence -from uuid import uuid4 +from typing import Optional +from ._transaction import ( + FileChange, + read_optional, + regular_file_exists, + source_change, +) from .droid_config import ( DroidSources, RenderedDocument, @@ -19,24 +21,6 @@ from .droid_config import ( from .droid_contract import AGENT_KEY, JsonObject, RuntimeContract, same_agent_id -@dataclass(frozen=True) -class FileChange: - file_path: str - label: str - original: Optional[str] - next_source: Optional[str] - mode: int - original_mode: Optional[int] - - -@dataclass -class StagedChange: - change: FileChange - temporary_path: Optional[str] - rollback_path: Optional[str] - committed: bool = False - - def _factory_paths() -> tuple[str, str, str]: directory = os.path.join(os.path.expanduser("~"), ".factory") return ( @@ -46,20 +30,10 @@ def _factory_paths() -> tuple[str, str, str]: ) -def _read_optional(file_path: str, label: str) -> Optional[str]: - try: - with open(file_path, "r", encoding="utf-8", newline="") as file: - return file.read() - except FileNotFoundError: - return None - except OSError as error: - raise OSError(f"Read {label} at {file_path}: {error}") from error - - def read_sources() -> DroidSources: root_path, legacy_path, settings_path = _factory_paths() - root_raw = _read_optional(root_path, "Factory Droid hooks") - settings_raw = _read_optional(settings_path, "Factory Droid settings") + root_raw = read_optional(root_path, "Factory Droid hooks") + settings_raw = read_optional(settings_path, "Factory Droid settings") primary = None if root_raw is not None: primary = parse_document( @@ -69,7 +43,7 @@ def read_sources() -> DroidSources: raw=root_raw, ) else: - legacy_raw = _read_optional(legacy_path, "Factory Droid legacy hooks") + legacy_raw = read_optional(legacy_path, "Factory Droid legacy hooks") if legacy_raw is not None: primary = parse_document( exists=True, @@ -90,34 +64,6 @@ def read_sources() -> DroidSources: return DroidSources(root_path, primary, settings) -def _existing_mode(file_path: str) -> Optional[int]: - try: - return stat.S_IMODE(os.stat(file_path).st_mode) - except FileNotFoundError: - return None - except OSError as error: - raise OSError(f"Read file mode at {file_path}: {error}") from error - - -def file_change( - file_path: str, - label: str, - next_source: Optional[str], - mode: int, -) -> Optional[FileChange]: - original = _read_optional(file_path, label) - if original == next_source: - return None - return FileChange( - file_path, - label, - original, - next_source, - mode, - _existing_mode(file_path), - ) - - def rendered_change(rendered: RenderedDocument) -> Optional[FileChange]: if not rendered.changed: return None @@ -127,208 +73,17 @@ def rendered_change(rendered: RenderedDocument) -> Optional[FileChange]: label = "Factory Droid legacy hooks" else: label = "Factory Droid hooks" - return FileChange( + return source_change( rendered.document.file_path, label, rendered.document.raw if rendered.document.exists else None, rendered.next_source, 0o600, - _existing_mode(rendered.document.file_path), ) -def _remove_optional(file_path: Optional[str]) -> None: - if not file_path: - return - try: - os.remove(file_path) - except FileNotFoundError: - return - - -def _cleanup_paths(paths: Sequence[Optional[str]]) -> List[str]: - errors = [] - for path in paths: - try: - _remove_optional(path) - except OSError as error: - errors.append(str(error)) - return errors - - -def _write_staged( - directory: str, - basename: str, - suffix: str, - content: str, - mode: int, - label: str, -) -> str: - descriptor = -1 - temporary_path = "" - try: - descriptor, temporary_path = tempfile.mkstemp( - prefix=f".{basename}.", - suffix=suffix, - dir=directory, - text=True, - ) - with os.fdopen(descriptor, "w", encoding="utf-8", newline="") as file: - descriptor = -1 - file.write(content) - file.flush() - os.fsync(file.fileno()) - os.chmod(temporary_path, mode) - return temporary_path - except Exception as error: - failures = [] - if descriptor >= 0: - try: - os.close(descriptor) - except OSError as close_error: - failures.append(f"close failed: {close_error}") - failures.extend(_cleanup_paths((temporary_path,))) - suffix = f"; cleanup failed: {'; '.join(failures)}" if failures else "" - raise OSError(f"Stage {label}: {error}{suffix}") from error - - -def _assert_unchanged(change: FileChange) -> None: - current = _read_optional(change.file_path, change.label) - if current != change.original: - raise OSError(f"{change.label} changed during installation: {change.file_path}") - - -def _stage(change: FileChange) -> StagedChange: - _assert_unchanged(change) - directory = os.path.dirname(change.file_path) - try: - os.makedirs(directory, mode=0o700, exist_ok=True) - except OSError as error: - raise OSError(f"Create directory for {change.label} at {directory}: {error}") from error - basename = os.path.basename(change.file_path) - temporary_path = None - rollback_path = None - try: - if change.next_source is not None: - temporary_path = _write_staged( - directory, - basename, - ".tmp", - change.next_source, - change.mode, - change.label, - ) - if change.original is not None and change.next_source is not None: - rollback_path = _write_staged( - directory, - basename, - ".rollback", - change.original, - change.original_mode or change.mode, - f"{change.label} rollback", - ) - elif change.original is not None: - rollback_path = os.path.join( - directory, - f".{basename}.{uuid4().hex}.rollback", - ) - if os.path.lexists(rollback_path): - raise FileExistsError(f"Rollback path already exists: {rollback_path}") - return StagedChange(change, temporary_path, rollback_path) - except Exception as error: - failures = _cleanup_paths((temporary_path, rollback_path)) - suffix = f"; cleanup failed: {'; '.join(failures)}" if failures else "" - raise OSError(f"Stage {change.label}: {error}{suffix}") from error - - -def _commit(staged: StagedChange) -> None: - _assert_unchanged(staged.change) - if staged.change.next_source is None: - if staged.rollback_path is None: - raise OSError(f"Missing rollback path for {staged.change.label}") - os.replace(staged.change.file_path, staged.rollback_path) - else: - if staged.temporary_path is None: - raise OSError(f"Missing staged file for {staged.change.label}") - os.replace(staged.temporary_path, staged.change.file_path) - staged.committed = True - - -def _rollback(staged: StagedChange) -> None: - if not staged.committed: - return - if staged.change.original is None: - _remove_optional(staged.change.file_path) - return - if staged.rollback_path is None: - raise OSError(f"Missing rollback data for {staged.change.label}") - os.replace(staged.rollback_path, staged.change.file_path) - - -def _cleanup(staged: StagedChange) -> None: - _remove_optional(staged.temporary_path) - _remove_optional(staged.rollback_path) - - -def write_changes(changes: Sequence[FileChange], label: str) -> None: - filtered = [change for change in changes if change.original != change.next_source] - if not filtered: - return - paths = [os.path.normcase(os.path.abspath(change.file_path)) for change in filtered] - if len(paths) != len(set(paths)): - raise ValueError(f"{label} contains duplicate file targets") - staged: List[StagedChange] = [] - try: - for change in filtered: - staged.append(_stage(change)) - for item in staged: - _commit(item) - except Exception as error: - recovery_errors = [] - for item in reversed(staged): - try: - _rollback(item) - except OSError as rollback_error: - recovery_errors.append(str(rollback_error)) - for item in staged: - try: - _cleanup(item) - except OSError as cleanup_error: - recovery_errors.append(str(cleanup_error)) - suffix = ( - f"; recovery failed: {'; '.join(recovery_errors)}" - if recovery_errors - else "" - ) - raise OSError(f"{label}: {error}{suffix}") from error - cleanup_errors = [] - for item in staged: - try: - _cleanup(item) - except OSError as error: - cleanup_errors.append(str(error)) - if cleanup_errors: - raise OSError(f"{label} cleanup failed: {'; '.join(cleanup_errors)}") - - -def regular_file_exists(file_path: str, label: str) -> bool: - try: - return stat.S_ISREG(os.stat(file_path).st_mode) - except FileNotFoundError: - return False - except OSError as error: - raise OSError(f"Read {label} at {file_path}: {error}") from error - - -def require_runtime(file_path: str, label: str) -> None: - if not file_path: - raise ValueError(f"{label} path is required") - if not regular_file_exists(file_path, label): - raise FileNotFoundError(f"{label} is missing: {file_path}") - - def _read_runtime_config(file_path: str) -> Optional[JsonObject]: - raw = _read_optional(file_path, "Elydora runtime config") + raw = read_optional(file_path, "Elydora runtime config") if raw is None: return None try: @@ -344,7 +99,7 @@ def _read_runtime_config(file_path: str) -> Optional[JsonObject]: return value -def runtime_files_exist(contracts: Sequence[RuntimeContract]) -> bool: +def runtime_files_exist(contracts: list[RuntimeContract]) -> bool: for contract in contracts: agent_directory = os.path.dirname(contract.guard_path) config = _read_runtime_config(os.path.join(agent_directory, "config.json")) diff --git a/sdks/python/elydora/plugins/qwen.py b/sdks/python/elydora/plugins/qwen.py new file mode 100644 index 0000000..ed1fda3 --- /dev/null +++ b/sdks/python/elydora/plugins/qwen.py @@ -0,0 +1,164 @@ +"""Qwen Code native global user-hook integration.""" + +from __future__ import annotations + +import json +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_document +from .qwen_contract import ( + AGENT_KEY, + AUDIT_SCRIPT, + GUARD_SCRIPT, + build_group, + elydora_dir, + runtime_contracts, + validate_javascript_regexes, +) +from .qwen_io import ( + read_document, + rendered_change, + require_runtime, + runtime_files_exist, +) + + +def _agent_directory(agent_id: str) -> str: + if ( + not agent_id + or agent_id in {".", ".."} + or os.path.basename(agent_id) != agent_id + or os.path.isabs(agent_id) + ): + 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.""" + + def install(self, config: InstallConfig) -> None: + agent_id = config.get("agent_id", "") + agent_directory = _agent_directory(agent_id) + document = read_document() + validate_javascript_regexes([document.hooks]) + + 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, + { + "PreToolUse": build_group(guard_path), + "PostToolUse": build_group(audit_path), + }, + ) + + runtime_config = { + "org_id": config.get("org_id", ""), + "agent_id": agent_id, + "kid": config.get("kid", ""), + "base_url": config.get("base_url", "https://api.elydora.com"), + "token": config.get("token", ""), + "agent_name": AGENT_KEY, + } + audit_script = generate_hook_script( + org_id=config.get("org_id", ""), + agent_id=agent_id, + private_key=config.get("private_key", ""), + kid=config.get("kid", ""), + base_url=config.get("base_url", "https://api.elydora.com"), + ) + 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() + if not document.exists: + return + 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", + ) + + def status(self) -> PluginStatus: + document = read_document() + contracts = runtime_contracts(document.hooks) + if document.hooks_disabled: + return PluginStatus( + installed=False, + agent=AGENT_KEY, + details=f"Configured hooks are disabled: {document.file_path}", + ) + if not contracts: + return PluginStatus( + installed=False, + agent=AGENT_KEY, + details="Not installed", + ) + installed = runtime_files_exist(contracts) + details = ( + f"Config: {document.file_path}" + if installed + else f"Configured at {document.file_path}; runtime scripts missing" + ) + return PluginStatus( + installed=installed, + agent=AGENT_KEY, + details=details, + ) diff --git a/sdks/python/elydora/plugins/qwen_config.py b/sdks/python/elydora/plugins/qwen_config.py new file mode 100644 index 0000000..6529eb0 --- /dev/null +++ b/sdks/python/elydora/plugins/qwen_config.py @@ -0,0 +1,178 @@ +"""Qwen Code settings parsing and source-preserving rendering.""" + +from __future__ import annotations + +from dataclasses import dataclass +from typing import Dict, Optional + +from ._jsonc import JsoncEditor +from .qwen_contract import ( + TOOL_EVENTS, + JsonObject, + QwenHookSettings, + managed_removals, + read_hook_settings, +) + + +OWNED_FILE_MARKER = "// Managed by Elydora" + + +@dataclass(frozen=True) +class QwenDocument: + file_path: str + exists: bool + raw: str + root: JsonObject + hooks: QwenHookSettings + has_hooks_container: bool + hooks_disabled: bool + owned_file: bool + + +@dataclass(frozen=True) +class RenderedDocument: + document: QwenDocument + changed: bool + next_source: Optional[str] + + +def _label(file_path: str) -> str: + return f"Qwen Code settings at {file_path}" + + +def parse_document( + *, + exists: bool, + file_path: str, + raw: str, +) -> QwenDocument: + label = _label(file_path) + editor = JsoncEditor( + 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 + hooks = ( + read_hook_settings(root["hooks"], f'{label} field "hooks"') + if has_container + else {} + ) + return QwenDocument( + file_path, + exists, + raw, + root, + hooks, + has_container, + disabled is True, + raw.startswith(OWNED_FILE_MARKER), + ) + + +def create_owned_document(file_path: str) -> QwenDocument: + return parse_document( + exists=False, + file_path=file_path, + raw=f"{OWNED_FILE_MARKER}\n{{}}\n", + ) + + +def _settings_from_editor( + editor: JsoncEditor, + document: QwenDocument, +) -> QwenHookSettings: + root = editor.value + if not isinstance(root, dict): + raise ValueError(f"{_label(document.file_path)} must contain a JSON object") + if "hooks" not in root: + return {} + return read_hook_settings( + root["hooks"], + f'{_label(document.file_path)} field "hooks"', + ) + + +def _remove_managed( + editor: JsoncEditor, + document: QwenDocument, + agent_id: Optional[str], +) -> None: + removals = managed_removals(document.hooks, agent_id) + for event in TOOL_EVENTS: + event_removals = sorted( + (item for item in removals if item.event == event), + key=lambda item: item.group_index, + reverse=True, + ) + for removal in event_removals: + group_path = ("hooks", event, removal.group_index) + if removal.remove_group: + editor.delete(group_path) + continue + for handler_index in sorted(removal.handler_indexes, reverse=True): + editor.delete((*group_path, "hooks", handler_index)) + if event_removals: + current = _settings_from_editor(editor, document) + if event in current and not current[event]: + editor.delete(("hooks", event)) + root = editor.value + if isinstance(root, dict) and "hooks" in root: + current = _settings_from_editor(editor, document) + if not current: + editor.delete(("hooks",)) + + +def _append_group( + editor: JsoncEditor, + document: QwenDocument, + event: str, + group: JsonObject, +) -> None: + root = editor.value + if not isinstance(root, dict): + raise ValueError(f"{_label(document.file_path)} must contain a JSON object") + if "hooks" not in root: + editor.add_property((), "hooks", {event: [group]}) + return + 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( + document: QwenDocument, + agent_id: Optional[str], + additions: Dict[str, JsonObject], +) -> RenderedDocument: + editor = JsoncEditor( + document.raw, + _label(document.file_path), + allow_trailing_commas=False, + ) + _remove_managed(editor, document, agent_id) + for event in TOOL_EVENTS: + group = additions.get(event) + if group is not None: + _append_group(editor, document, event, group) + next_document = parse_document( + exists=document.exists, + file_path=document.file_path, + raw=editor.raw, + ) + 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, + ) diff --git a/sdks/python/elydora/plugins/qwen_contract.py b/sdks/python/elydora/plugins/qwen_contract.py new file mode 100644 index 0000000..ddfe4e6 --- /dev/null +++ b/sdks/python/elydora/plugins/qwen_contract.py @@ -0,0 +1,409 @@ +"""Qwen Code hook schema, commands, and exact ownership rules.""" + +from __future__ import annotations + +from dataclasses import dataclass +import json +import math +import os +import shutil +import subprocess # nosec B404 +import sys +from typing import Any, Dict, List, Optional, Sequence, Set, Tuple + + +AGENT_KEY = "qwen" +GUARD_SCRIPT = "guard.py" +AUDIT_SCRIPT = "hook.py" +HOOK_TIMEOUT_MILLISECONDS = 10_000 +TOOL_EVENTS = ("PreToolUse", "PostToolUse") + +_EVENT_NAMES = { + "PreToolUse", + "PostToolUse", + "PostToolUseFailure", + "PostToolBatch", + "Notification", + "UserPromptSubmit", + "UserPromptExpansion", + "SessionStart", + "Stop", + "MessageDisplay", + "SubagentStart", + "SubagentStop", + "PreCompact", + "PostCompact", + "SessionEnd", + "PermissionRequest", + "PermissionDenied", + "StopFailure", + "TodoCreated", + "TodoCompleted", + "InstructionsLoaded", +} +_CONFIG_FIELDS = {"enabled", "disabled", "notifications"} +_HANDLER_KEYS = {"command", "shell", "timeout", "type"} +_GROUP_KEYS = {"hooks", "matcher"} +_REGEX_TIMEOUT_SECONDS = 10 +_REGEX_VALIDATOR = """import fs from 'node:fs'; +const entries = JSON.parse(fs.readFileSync(0, 'utf8')); +for (const entry of entries) { + try { + new RegExp(entry.pattern); + } catch (error) { + const message = error instanceof Error ? error.message : String(error); + process.stderr.write(entry.label + ': ' + message); + process.exit(1); + } +} +""" + +JsonObject = Dict[str, Any] +QwenHookSettings = Dict[str, Any] + + +@dataclass(frozen=True) +class RuntimeContract: + agent_id: str + guard_path: str + audit_path: str + + +@dataclass(frozen=True) +class ManagedRemoval: + event: str + group_index: int + handler_indexes: Tuple[int, ...] + remove_group: bool + + +def elydora_dir() -> str: + return os.path.join(os.path.expanduser("~"), ".elydora") + + +def _quote_posix(value: str) -> str: + return "'" + value.replace("'", "'\"'\"'") + "'" + + +def _quote_powershell(value: str) -> str: + return "'" + value.replace("'", "''") + "'" + + +def build_command(script_path: str) -> str: + if os.name == "nt": + 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] + 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') + + +def _validate_timeout(value: JsonObject, label: str) -> None: + if "timeout" not in value: + return + timeout = value["timeout"] + if ( + isinstance(timeout, bool) + or not isinstance(timeout, (int, float)) + or not math.isfinite(timeout) + or timeout < 0 + ): + raise ValueError(f"{label} timeout must be a non-negative finite number") + + +def _validate_handler(value: Any, label: str) -> JsonObject: + if not isinstance(value, dict): + raise ValueError(f"{label} must be an object") + 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"') + _validate_timeout(handler, label) + for key in ("name", "description", "statusMessage", "source"): + _optional_string(handler, key, 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") + _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") + _optional_string_map(handler, "headers", label) + _optional_boolean(handler, "once", label) + _optional_string(handler, "if", label) + allowed = handler.get("allowedEnvVars") + if "allowedEnvVars" in handler and ( + not isinstance(allowed, list) + or any(not isinstance(item, str) for item in allowed) + ): + 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") + _optional_string(handler, "model", label) + return handler + + +def _validate_group(value: Any, label: str) -> JsonObject: + if not isinstance(value, dict): + raise ValueError(f"{label} must be an object") + group = dict(value) + if "matcher" in group and not isinstance(group["matcher"], str): + raise ValueError(f"{label} matcher must be a string") + if "sequential" in group and not isinstance(group["sequential"], bool): + raise ValueError(f"{label} sequential must be a boolean") + handlers = group.get("hooks") + if not isinstance(handlers, list): + raise ValueError(f"{label} must contain a hooks array") + group["hooks"] = [ + _validate_handler(handler, f"{label}.hooks[{index}]") + for index, handler in enumerate(handlers) + ] + return group + + +def read_hook_settings(value: Any, label: str) -> QwenHookSettings: + if not isinstance(value, dict): + raise ValueError(f"{label} must contain a JSON object") + settings = dict(value) + for event, groups in value.items(): + if event in _CONFIG_FIELDS: + 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') + settings[event] = [ + _validate_group(group, f'{label} field "{event}"[{index}]') + for index, group in enumerate(groups) + ] + return settings + + +def _regex_entries(settings: QwenHookSettings) -> List[JsonObject]: + entries = [] + for event, groups in settings.items(): + if event in _CONFIG_FIELDS: + continue + for index, group in enumerate(groups): + matcher = group.get("matcher") + if isinstance(matcher, str) and matcher.strip() not in {"", "*"}: + entries.append( + { + "label": f'Qwen Code hooks field "{event}"[{index}] matcher', + "pattern": matcher, + } + ) + return entries + + +def validate_javascript_regexes(settings: Sequence[QwenHookSettings]) -> None: + entries = [entry for source in settings for entry in _regex_entries(source)] + if not entries: + return + node_path = shutil.which("node") + if node_path is None: + raise FileNotFoundError( + "Node.js runtime is required to validate Qwen Code hook expressions" + ) + try: + result = subprocess.run( # nosec B603 + [node_path, "--input-type=module", "--eval", _REGEX_VALIDATOR], + input=json.dumps(entries), + text=True, + capture_output=True, + check=False, + timeout=_REGEX_TIMEOUT_SECONDS, + ) + except subprocess.TimeoutExpired as error: + raise TimeoutError( + "Qwen Code matcher validation timed out after " + f"{_REGEX_TIMEOUT_SECONDS} seconds" + ) from error + except OSError as error: + raise OSError(f"Run Node.js Qwen Code matcher validator: {error}") from error + if result.returncode == 0: + return + message = ( + result.stderr.strip() + or result.stdout.strip() + or f"Node.js exited with code {result.returncode}" + ) + raise ValueError( + f"Qwen Code matcher must be a valid JavaScript regular expression: {message}" + ) + + +def _read_quoted_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 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 + for group in groups + for handler in group["hooks"] + if (agent_id := managed_agent_id(handler, script_name)) is not None + } + + +def runtime_contracts(settings: QwenHookSettings) -> List[RuntimeContract]: + guards = _managed_ids(settings.get("PreToolUse", []), GUARD_SCRIPT) + audits = _managed_ids(settings.get("PostToolUse", []), AUDIT_SCRIPT) + contracts = [] + for agent_id in sorted(guards, key=os.path.normcase): + if any(same_agent_id(agent_id, audit_id) for audit_id in audits): + root = os.path.join(elydora_dir(), agent_id) + contracts.append( + RuntimeContract( + agent_id, + os.path.join(root, GUARD_SCRIPT), + os.path.join(root, AUDIT_SCRIPT), + ) + ) + return contracts diff --git a/sdks/python/elydora/plugins/qwen_io.py b/sdks/python/elydora/plugins/qwen_io.py new file mode 100644 index 0000000..32d10f7 --- /dev/null +++ b/sdks/python/elydora/plugins/qwen_io.py @@ -0,0 +1,139 @@ +"""Qwen Code home discovery, settings I/O, and runtime inspection.""" + +from __future__ import annotations + +import json +import os +import re +import tempfile +from typing import Optional + +from ._dotenv import parse_dotenv +from ._transaction import ( + FileChange, + read_optional, + regular_file_exists, + require_runtime, + source_change, +) +from .qwen_config import ( + QwenDocument, + RenderedDocument, + create_owned_document, + parse_document, +) +from .qwen_contract import AGENT_KEY, JsonObject, RuntimeContract, same_agent_id + + +def _default_qwen_home() -> str: + 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) + except json.JSONDecodeError as error: + raise ValueError( + f"Failed to parse Elydora runtime config at {file_path}: {error}" + ) from error + if not isinstance(value, dict): + raise ValueError( + f"Elydora runtime config at {file_path} must contain a JSON object" + ) + return value + + +def runtime_files_exist(contracts: list[RuntimeContract]) -> bool: + for contract in contracts: + agent_directory = os.path.dirname(contract.guard_path) + runtime_config = _read_runtime_config( + os.path.join(agent_directory, "config.json") + ) + config_agent_id = runtime_config.get("agent_id") if runtime_config else None + if ( + runtime_config is None + or runtime_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 + if regular_file_exists( + contract.guard_path, + "Elydora guard runtime", + ) and regular_file_exists( + contract.audit_path, + "Elydora audit runtime", + ): + return True + return False + + +__all__ = [ + "read_document", + "rendered_change", + "require_runtime", + "resolve_qwen_home", + "runtime_files_exist", +] diff --git a/sdks/python/elydora/plugins/registry.py b/sdks/python/elydora/plugins/registry.py index 273d6c9..45aa2bb 100644 --- a/sdks/python/elydora/plugins/registry.py +++ b/sdks/python/elydora/plugins/registry.py @@ -76,6 +76,11 @@ SUPPORTED_AGENTS: Dict[str, Dict[str, str]] = { "hook_event": "tool.execute.after", "config_path": "~/.config/opencode/plugins/", }, + "qwen": { + "name": "Qwen Code", + "hook_event": "PreToolUse/PostToolUse", + "config_path": "~/.qwen/settings.json", + }, } diff --git a/sdks/python/tests/qwen_support.py b/sdks/python/tests/qwen_support.py new file mode 100644 index 0000000..680bec1 --- /dev/null +++ b/sdks/python/tests/qwen_support.py @@ -0,0 +1,134 @@ +from __future__ import annotations + +import base64 +import json +import os +from pathlib import Path +import subprocess # nosec B404 +import sys +from typing import Any, Dict, Optional + +import pytest + +from elydora.plugins import qwen +from elydora.plugins._jsonc import parse_jsonc +from elydora.plugins.base import InstallConfig + + +AGENT_ID = "agent-1" +MISSING = object() +JsonObject = Dict[str, Any] + + +def generated_command(script_path: Path) -> str: + if os.name == "nt": + executable = "'" + sys.executable.replace("'", "''") + "'" + script = "'" + str(script_path).replace("'", "''") + "'" + return f"& {executable} {script}; exit $LASTEXITCODE" + executable = "'" + sys.executable.replace("'", "'\"'\"'") + "'" + script = "'" + str(script_path).replace("'", "'\"'\"'") + "'" + return f"{executable} {script}" + + +def parse_settings(path: Path) -> JsonObject: + value = parse_jsonc( + path.read_text(encoding="utf-8"), + str(path), + allow_trailing_commas=False, + ) + assert isinstance(value, dict) + return value + + +def managed_handler( + settings: JsonObject, + event: str, + script_path: Path, +) -> 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 + + +def run_handler( + handler: JsonObject, + payload: str, +) -> subprocess.CompletedProcess[str]: + if handler["shell"] == "powershell": + command = [ + "powershell", + "-NoProfile", + "-NonInteractive", + "-Command", + handler["command"], + ] + else: + command = ["bash", "-c", handler["command"]] + return subprocess.run( # nosec B603 + command, + input=payload, + text=True, + capture_output=True, + check=False, + timeout=10, + ) + + +class QwenFixture: + def __init__( + self, + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, + *, + existing_settings: Any = MISSING, + create_guard: bool = True, + ) -> None: + self.home_dir = tmp_path / "home with spaces and 'quote" + self.workspace_dir = self.home_dir / "workspace" + self.qwen_dir = self.home_dir / ".qwen" + self.config_path = self.qwen_dir / "settings.json" + self.agent_dir = self.home_dir / ".elydora" / AGENT_ID + 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) + private_key = base64.urlsafe_b64encode(bytes([1]) * 32).rstrip(b"=").decode() + self.config: InstallConfig = { + "agent_id": AGENT_ID, + "agent_name": "qwen", + "org_id": "org-1", + "private_key": private_key, + "kid": "kid-1", + "base_url": "https://api.elydora.test", + "guard_script_path": str(self.guard_path), + } + 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: + self.plugin.install(self.config) diff --git a/sdks/python/tests/test_droid_plugin.py b/sdks/python/tests/test_droid_plugin.py index 47e0243..a134c31 100644 --- a/sdks/python/tests/test_droid_plugin.py +++ b/sdks/python/tests/test_droid_plugin.py @@ -10,7 +10,7 @@ from typing import Any, Dict, Optional import pytest from elydora import cli -from elydora.plugins import droid, droid_io, droid_jsonc +from elydora.plugins import _jsonc, droid, droid_io from elydora.plugins.base import InstallConfig from elydora.plugins.registry import SUPPORTED_AGENTS @@ -81,7 +81,7 @@ class DroidFixture: def _load(path: Path) -> JsonObject: - value = droid_jsonc.parse_jsonc(path.read_text(encoding="utf-8"), str(path)) + value = _jsonc.parse_jsonc(path.read_text(encoding="utf-8"), str(path)) assert isinstance(value, dict) return value diff --git a/sdks/python/tests/test_qwen_plugin.py b/sdks/python/tests/test_qwen_plugin.py new file mode 100644 index 0000000..f3cb2f1 --- /dev/null +++ b/sdks/python/tests/test_qwen_plugin.py @@ -0,0 +1,365 @@ +from __future__ import annotations + +import json +import os +from pathlib import Path +from typing import Any + +import pytest + +from elydora import cli +from elydora.plugins import _transaction, qwen +from elydora.plugins.registry import SUPPORTED_AGENTS +from qwen_support import ( + AGENT_ID, + QwenFixture, + generated_command, + managed_handler, + parse_settings, + run_handler, +) + + +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", + "config_path": "~/.qwen/settings.json", + } + assert cli.PLUGIN_MAP["qwen"] is qwen.QwenPlugin + + +def test_install_preserves_jsonc_and_is_idempotent( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, + capsys: pytest.CaptureFixture[str], +) -> None: + source = ( + "{\r\n" + " // Keep this user preference.\r\n" + ' "theme": "GitHub",\r\n' + ' "hooks": {\r\n' + ' "SessionStart": [{ "hooks": [{ "type": "command", "command": "session-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) + workspace_settings = fixture.workspace_dir / ".qwen" / "settings.json" + workspace_settings.parent.mkdir(parents=True) + workspace_settings.write_text('{ "owner": "workspace" }\n', encoding="utf-8") + fixture.install() + fixture.install() + output = capsys.readouterr().out + with open(fixture.config_path, "r", encoding="utf-8", newline="") as file: + raw = file.read() + settings = parse_settings(fixture.config_path) + assert "Keep this user preference" in raw + assert "\r\n" in raw + assert settings["theme"] == "GitHub" + assert settings["hooks"]["SessionStart"][0]["hooks"][0]["command"] == "session-hook" + assert len(settings["hooks"]["PreToolUse"]) == 2 + assert len(settings["hooks"]["PostToolUse"]) == 1 + for event, script_path in ( + ("PreToolUse", fixture.guard_path), + ("PostToolUse", fixture.audit_path), + ): + handler = managed_handler(settings, event, script_path) + assert handler is not None + assert sorted(handler) == ["command", "shell", "timeout", "type"] + assert handler["shell"] == EXPECTED_SHELL + assert handler["timeout"] == 10_000 + assert parse_settings(workspace_settings) == {"owner": "workspace"} + assert ( + json.loads(fixture.runtime_config.read_text(encoding="utf-8"))["agent_name"] + == "qwen" + ) + assert fixture.private_key_path.is_file() + assert fixture.audit_path.is_file() + assert "run /hooks" in output + + +def test_qwen_home_uses_official_user_env_precedence( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + fixture = QwenFixture(monkeypatch, tmp_path) + first_home = tmp_path / "first # qwen home" + second_home = tmp_path / "second qwen home" + fixture.qwen_dir.mkdir(parents=True) + (fixture.qwen_dir / ".env").write_text( + 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" + ) + fixture.install() + selected = first_home / "settings.json" + assert managed_handler(parse_settings(selected), "PreToolUse", fixture.guard_path) + 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"), + [ + ("relative-qwen", "relative-qwen"), + ("~/custom-qwen", "custom-qwen"), + ("", ".qwen"), + ], +) +def test_explicit_qwen_home_supports_relative_tilde_and_empty_values( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, + value: str, + relative: str, +) -> None: + fixture = QwenFixture(monkeypatch, tmp_path / (relative.replace(".", "default"))) + fixture.qwen_dir.mkdir(parents=True) + ignored = tmp_path / "ignored-qwen-home" + (fixture.qwen_dir / ".env").write_text(f"QWEN_HOME={ignored}\n", encoding="utf-8") + monkeypatch.setenv("QWEN_HOME", value) + fixture.install() + if value.startswith("relative"): + selected = fixture.workspace_dir / relative / "settings.json" + else: + selected = fixture.home_dir / relative / "settings.json" + assert "hooks" in parse_settings(selected) + assert not (ignored / "settings.json").exists() + + +def test_commands_block_and_forward_official_input_byte_for_byte( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + fixture = QwenFixture(monkeypatch, tmp_path) + fixture.install() + capture_path = tmp_path / "captured-event.json" + fixture.audit_path.write_text( + "import pathlib, sys\n" + f"pathlib.Path({str(capture_path)!r}).write_bytes(sys.stdin.buffer.read())\n", + 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( + "source", + [ + "{ malformed", + "[]", + '{ "owner": true, }', + '{ "hooks": {}, "hooks": {} }', + '{ "disableAllHooks": "yes" }', + '{ "hooks": [] }', + '{ "hooks": { "UnknownEvent": [] } }', + '{ "hooks": { "PreToolUse": null } }', + '{ "hooks": { "PreToolUse": [null] } }', + '{ "hooks": { "PreToolUse": [{ "matcher": "[", "hooks": [] }] } }', + '{ "hooks": { "PreToolUse": [{ "sequential": "yes", "hooks": [] }] } }', + '{ "hooks": { "PreToolUse": [{ "hooks": null }] } }', + '{ "hooks": { "PreToolUse": [{ "hooks": [{ "type": "command" }] }] } }', + '{ "hooks": { "PreToolUse": [{ "hooks": [{ "type": "http" }] }] }', + '{ "hooks": { "PreToolUse": [{ "hooks": [{ "type": "function", "command": "x" }] }] } }', + '{ "hooks": { "PreToolUse": [{ "hooks": [{ "type": "command", "command": "x", "timeout": "ten" }] }] } }', + ], +) +def test_install_rejects_malformed_settings_before_every_write( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, + source: str, +) -> None: + fixture = QwenFixture(monkeypatch, tmp_path, existing_settings=source) + with pytest.raises((ValueError, RuntimeError), match="Qwen"): + 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"): + 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 [ + path + 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( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + fixture = QwenFixture( + monkeypatch, + tmp_path, + existing_settings={"owner": "user"}, + ) + original = fixture.config_path.read_text(encoding="utf-8") + real_replace = _transaction.os.replace + failed = False + + def fail_settings_commit(source: Any, destination: Any) -> None: + nonlocal failed + if not failed and Path(destination) == fixture.config_path: + failed = True + raise OSError("simulated settings commit failure") + real_replace(source, destination) + + monkeypatch.setattr(_transaction.os, "replace", fail_settings_commit) + with pytest.raises(OSError, match="Write Qwen Code installation"): + fixture.install() + assert failed is True + assert fixture.config_path.read_text(encoding="utf-8") == original + assert not fixture.audit_path.exists() + assert not fixture.runtime_config.exists() + assert not fixture.private_key_path.exists() + assert [ + path + for path in fixture.home_dir.rglob("*") + if path.suffix in {".tmp", ".rollback"} + ] == [] + + +def test_install_rejects_invalid_agent_id_before_writes( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + fixture = QwenFixture(monkeypatch, tmp_path) + fixture.config["agent_id"] = "../escape" + with pytest.raises(ValueError, match="single non-empty path segment"): + fixture.install() + assert not fixture.config_path.exists()