diff --git a/AGENTS.md b/AGENTS.md index b396458..b781b19 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -28,7 +28,7 @@ This repository publishes the open-source server, console, integration catalog, - Write Auggie hooks only to `~/.augment/settings.json`; keep system and workspace settings read-only. Generate `.cmd` wrappers on Windows and `.sh` wrappers on Unix because Auggie dispatches supported script paths, and express hook timeouts in milliseconds. - Validate Auggie matcher syntax during installation with Node.js `new RegExp`; keep status and uninstall independent from the JavaScript validator so recovery remains available offline. - Write Cline file hooks to `$CLINE_DIR/hooks` with `~/.cline/hooks` as the default; keep Documents and workspace hook roots read-only. Use supported event filenames, preserve official hook input byte-for-byte, use `process.execPath` in Node and Go wrappers, and resolve Python runtimes from absolute shebangs. Translate guard exit code `2` into JSON stdout with `cancel: true`. Treat hook errors, timeouts, and invalid control JSON as fail-open. -- Preserve Factory Droid's scope-root `hooks.json` precedence, legacy nested hook source, and per-event `settings.json` fallback. Scope-root and legacy files store event keys at the document root; settings stores the same map under `hooks`. Edit only effective user sources; keep project and organization sources read-only. Preserve JSONC comments through syntax-tree edits and require `/hooks` review after external changes. +- Preserve Factory Droid's scope-root `hooks.json` precedence, legacy nested hook source, and per-event `settings.json` fallback. Scope-root and legacy files store event keys at the document root; settings stores the same map under `hooks`. Edit only effective user sources; keep project and organization sources read-only. Preserve JSONC comments through syntax-tree edits, use SDK-native absolute runtime commands, commit Python runtime artifacts and hook sources as one recoverable transaction, and require `/hooks` review after external changes. ## Mirror Workflow diff --git a/integrations/catalog.json b/integrations/catalog.json index ff1b0e9..c5d82e6 100644 --- a/integrations/catalog.json +++ b/integrations/catalog.json @@ -186,7 +186,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://docs.factory.ai/reference/hooks-reference", "contract_variants": [ diff --git a/sdks/python/README.md b/sdks/python/README.md index 7af256a..e519df7 100644 --- a/sdks/python/README.md +++ b/sdks/python/README.md @@ -83,6 +83,8 @@ Grok Build installation writes native global hooks to `$GROK_HOME/hooks/elydora- Auggie installation writes user-level hooks to `~/.augment/settings.json` and creates the `.cmd` or `.sh` wrappers required by its command runner. System and workspace settings remain unchanged. Sessions started with `--augment-cache-dir` load settings from that alternate directory. +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. + 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 @@ -106,6 +108,7 @@ Kiro CLI installation covers both runtime contracts. Kiro CLI v2 uses the genera | Grok Build | `grok` | | Copilot CLI | `copilot` | | Cursor | `cursor` | +| Factory Droid | `droid` | | Gemini CLI | `gemini` | | Kiro CLI | `kirocli` | | Kiro IDE | `kiroide` | diff --git a/sdks/python/elydora/cli.py b/sdks/python/elydora/cli.py index 765b375..f8cf5f2 100644 --- a/sdks/python/elydora/cli.py +++ b/sdks/python/elydora/cli.py @@ -24,6 +24,7 @@ from .plugins.codex import CodexPlugin from .plugins.cline import ClinePlugin from .plugins.copilot import CopilotPlugin from .plugins.cursor import CursorPlugin +from .plugins.droid import DroidPlugin from .plugins.gemini import GeminiPlugin from .plugins.grok import GrokPlugin from .plugins.kirocli import KiroCliPlugin @@ -40,6 +41,7 @@ PLUGIN_MAP: Dict[str, Type[AgentPlugin]] = { "cline": ClinePlugin, "copilot": CopilotPlugin, "cursor": CursorPlugin, + "droid": DroidPlugin, "gemini": GeminiPlugin, "grok": GrokPlugin, "kirocli": KiroCliPlugin, diff --git a/sdks/python/elydora/plugins/droid.py b/sdks/python/elydora/plugins/droid.py new file mode 100644 index 0000000..9cc2950 --- /dev/null +++ b/sdks/python/elydora/plugins/droid.py @@ -0,0 +1,190 @@ +"""Factory Droid native global user-hook integration.""" + +from __future__ import annotations + +import json +import os +from typing import List, Optional + +from .base import AgentPlugin, InstallConfig, PluginStatus +from .droid_config import ( + additions_for, + installation_targets, + render_document, + unique_documents, +) +from .droid_contract import ( + AGENT_KEY, + AUDIT_SCRIPT, + GUARD_SCRIPT, + TOOL_EVENTS, + build_group, + elydora_dir, + merge_hook_settings, + runtime_contracts, + 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 + + +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 DroidPlugin(AgentPlugin): + """Install Elydora into Factory Droid's user-level hook sources.""" + + def install(self, config: InstallConfig) -> None: + agent_id = config.get("agent_id", "") + agent_directory = _agent_directory(agent_id) + sources = read_sources() + validate_javascript_regexes([ + sources.primary.hooks if sources.primary is not None else {}, + sources.settings.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) + + selected = installation_targets(sources) + groups = { + "PreToolUse": build_group(guard_path), + "PostToolUse": build_group(audit_path), + } + documents = unique_documents([ + sources.primary, + sources.settings if sources.settings.has_hooks_container else None, + *(selected.targets[event] for event in TOOL_EVENTS), + ]) + rendered = [ + render_document( + document, + None, + additions_for(document, selected.targets, groups), + ) + for document in documents + ] + + 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, + )) + for item in rendered: + _append_change(changes, rendered_change(item)) + write_changes(changes, "Write Factory Droid installation") + + locations = ", ".join( + f"{event}: {selected.targets[event].file_path}" + for event in TOOL_EVENTS + ) + print(f"Factory Droid: {locations}") + print("Factory Droid: run /hooks to review the Elydora hook changes.") + + def uninstall(self, agent_id: str = "") -> None: + sources = read_sources() + documents = unique_documents([ + sources.primary, + sources.settings if sources.settings.has_hooks_container else None, + ]) + changes: List[FileChange] = [] + for document in documents: + rendered = render_document(document, agent_id or None, {}) + _append_change(changes, rendered_change(rendered)) + write_changes(changes, "Write Factory Droid hook sources") + + def status(self) -> PluginStatus: + sources = read_sources() + effective = merge_hook_settings( + sources.primary.hooks if sources.primary is not None else None, + sources.settings.hooks, + ) + contracts = runtime_contracts(effective) + config_path = display_config_path(sources) + if effective.get("hooksDisabled") is True: + return PluginStatus( + installed=False, + agent=AGENT_KEY, + details=f"Configured hooks are disabled: {config_path}", + ) + if not contracts: + return PluginStatus( + installed=False, + agent=AGENT_KEY, + details="Not installed", + ) + installed = runtime_files_exist(contracts) + details = ( + f"Config: {config_path}" + if installed + else f"Configured at {config_path}; runtime scripts missing" + ) + return PluginStatus( + installed=installed, + agent=AGENT_KEY, + details=details, + ) diff --git a/sdks/python/elydora/plugins/droid_config.py b/sdks/python/elydora/plugins/droid_config.py new file mode 100644 index 0000000..42bc73d --- /dev/null +++ b/sdks/python/elydora/plugins/droid_config.py @@ -0,0 +1,262 @@ +"""Factory Droid source selection and source-preserving rendering.""" + +from __future__ import annotations + +from dataclasses import dataclass +from typing import Dict, List, Optional, Sequence, Tuple + +from .droid_contract import ( + TOOL_EVENTS, + DroidHookSettings, + JsonObject, + has_own, + managed_removals, + read_hook_settings, +) +from .droid_jsonc import JsonPathPart, JsoncEditor + + +OWNED_FILE_MARKER = "// Managed by Elydora" + + +@dataclass(frozen=True) +class DroidDocument: + kind: str + file_path: str + exists: bool + raw: str + root: JsonObject + hooks: DroidHookSettings + base_path: Tuple[JsonPathPart, ...] + has_hooks_container: bool + owned_file: bool + + +@dataclass(frozen=True) +class DroidSources: + root_path: str + primary: Optional[DroidDocument] + settings: DroidDocument + + +@dataclass(frozen=True) +class InstallationTargets: + targets: Dict[str, DroidDocument] + created_root: Optional[DroidDocument] + + +@dataclass(frozen=True) +class RenderedDocument: + document: DroidDocument + changed: bool + next_source: Optional[str] + + +def _label(kind: str, file_path: str) -> str: + if kind == "settings": + return f"Factory Droid settings at {file_path}" + if kind == "legacy": + return f"Factory Droid legacy hooks at {file_path}" + return f"Factory Droid hooks at {file_path}" + + +def parse_document( + *, + exists: bool, + file_path: str, + kind: str, + raw: str, +) -> DroidDocument: + label = _label(kind, file_path) + editor = JsoncEditor(raw, label) + if not isinstance(editor.value, dict): + raise ValueError(f"{label} must contain a JSON object") + root = editor.value + if kind == "settings": + has_container = has_own(root, "hooks") + hooks = ( + read_hook_settings(root["hooks"], f'{label} field "hooks"') + if has_container + else {} + ) + return DroidDocument( + kind, + file_path, + exists, + raw, + root, + hooks, + ("hooks",), + has_container, + False, + ) + return DroidDocument( + kind, + file_path, + exists, + raw, + root, + read_hook_settings(root, label), + (), + True, + raw.startswith(OWNED_FILE_MARKER), + ) + + +def create_settings_document(file_path: str) -> DroidDocument: + return parse_document( + exists=False, + file_path=file_path, + kind="settings", + raw="{}\n", + ) + + +def create_owned_hook_document(file_path: str) -> DroidDocument: + return parse_document( + exists=False, + file_path=file_path, + kind="hooks", + raw=f"{OWNED_FILE_MARKER}\n{{}}\n", + ) + + +def _hooks_from_editor( + editor: JsoncEditor, + document: DroidDocument, +) -> DroidHookSettings: + root = editor.value + if not isinstance(root, dict): + raise ValueError(f"{_label(document.kind, document.file_path)} must be an object") + if document.kind == "settings": + hooks = root.get("hooks") + return read_hook_settings( + hooks, + f'{_label(document.kind, document.file_path)} field "hooks"', + ) + return read_hook_settings(root, _label(document.kind, document.file_path)) + + +def _event_path(document: DroidDocument, event: str) -> Tuple[JsonPathPart, ...]: + return (*document.base_path, event) + + +def _remove_managed( + editor: JsoncEditor, + document: DroidDocument, + 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 = (*_event_path(document, 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)) + current = _hooks_from_editor(editor, document) + if document.owned_file and event in current and not current[event]: + editor.delete(_event_path(document, event)) + + +def render_document( + document: DroidDocument, + agent_id: Optional[str], + additions: Dict[str, JsonObject], +) -> RenderedDocument: + editor = JsoncEditor( + document.raw, + _label(document.kind, document.file_path), + ) + _remove_managed(editor, document, agent_id) + for event in TOOL_EVENTS: + group = additions.get(event) + if group is None: + continue + current = _hooks_from_editor(editor, document) + if event in current: + editor.append(_event_path(document, event), group) + else: + editor.add_property(document.base_path, event, [group]) + next_document = parse_document( + exists=document.exists, + file_path=document.file_path, + kind=document.kind, + raw=editor.raw, + ) + if ( + not additions + and document.owned_file + and document.kind != "settings" + and not next_document.hooks + ): + return RenderedDocument(document, True, None) + return RenderedDocument( + document, + editor.raw != document.raw, + editor.raw, + ) + + +def _event_target( + event: str, + sources: DroidSources, + create_root: "_RootFactory", +) -> DroidDocument: + if sources.primary is not None and event in sources.primary.hooks: + return sources.primary + if sources.settings.has_hooks_container and event in sources.settings.hooks: + return sources.settings + if sources.primary is not None: + return sources.primary + if sources.settings.has_hooks_container: + return sources.settings + return create_root.get() + + +class _RootFactory: + def __init__(self, file_path: str) -> None: + self.file_path = file_path + self.document: Optional[DroidDocument] = None + + def get(self) -> DroidDocument: + if self.document is None: + self.document = create_owned_hook_document(self.file_path) + return self.document + + +def installation_targets(sources: DroidSources) -> InstallationTargets: + root = _RootFactory(sources.root_path) + targets = { + event: _event_target(event, sources, root) + for event in TOOL_EVENTS + } + return InstallationTargets(targets, root.document) + + +def additions_for( + document: DroidDocument, + targets: Dict[str, DroidDocument], + groups: Dict[str, JsonObject], +) -> Dict[str, JsonObject]: + return { + event: groups[event] + for event in TOOL_EVENTS + if targets[event].file_path == document.file_path + } + + +def unique_documents( + documents: Sequence[Optional[DroidDocument]], +) -> List[DroidDocument]: + unique: Dict[str, DroidDocument] = {} + for document in documents: + if document is not None: + unique[document.file_path] = document + return list(unique.values()) diff --git a/sdks/python/elydora/plugins/droid_contract.py b/sdks/python/elydora/plugins/droid_contract.py new file mode 100644 index 0000000..36bdc97 --- /dev/null +++ b/sdks/python/elydora/plugins/droid_contract.py @@ -0,0 +1,373 @@ +"""Factory Droid hook schema, command construction, and 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 = "droid" +GUARD_SCRIPT = "guard.py" +AUDIT_SCRIPT = "hook.py" +HOOK_TIMEOUT_SECONDS = 10 +TOOL_EVENTS = ("PreToolUse", "PostToolUse") + +_EVENT_NAMES = { + "PreToolUse", + "PostToolUse", + "Notification", + "UserPromptSubmit", + "Stop", + "SubagentStop", + "PreCompact", + "SessionStart", + "SessionEnd", +} +_FLAG_NAMES = {"hooksDisabled", "showHookOutput"} +_HANDLER_KEYS = {"command", "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] +DroidHookSettings = 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 is_object(value: Any) -> bool: + return isinstance(value, dict) + + +def has_own(value: JsonObject, key: str) -> bool: + return key in value + + +def elydora_dir() -> str: + return os.path.join(os.path.expanduser("~"), ".elydora") + + +def _quote_windows(value: str) -> str: + return '"' + value.replace('"', '\\"') + '"' + + +def _quote_posix(value: str) -> str: + return "'" + value.replace("'", "'\"'\"'") + "'" + + +def build_command(script_path: str) -> str: + quote = _quote_windows if os.name == "nt" else _quote_posix + return f"{quote(sys.executable)} {quote(script_path)}" + + +def build_group(script_path: str) -> JsonObject: + return { + "matcher": "*", + "hooks": [{ + "type": "command", + "command": build_command(script_path), + "timeout": HOOK_TIMEOUT_SECONDS, + }], + } + + +def _validate_handler(value: Any, label: str) -> JsonObject: + if not is_object(value): + raise ValueError(f"{label} must be an object") + handler = dict(value) + if handler.get("type") != "command": + raise ValueError(f'{label} type must be "command"') + if not isinstance(handler.get("command"), str): + raise ValueError(f"{label} command must be a string") + timeout = handler.get("timeout") + if timeout is not None and ( + isinstance(timeout, bool) + or not isinstance(timeout, (int, float)) + or not math.isfinite(timeout) + ): + raise ValueError(f"{label} timeout must be a finite number") + return handler + + +def _validate_group(value: Any, label: str) -> JsonObject: + if not is_object(value): + raise ValueError(f"{label} must be an object") + group = dict(value) + matcher = group.get("matcher") + if "matcher" in group and not isinstance(matcher, str): + raise ValueError(f"{label} matcher must be a string") + command_regex = group.get("commandRegex") + if "commandRegex" in group and not isinstance(command_regex, str): + raise ValueError(f"{label} commandRegex must be a string") + 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) -> DroidHookSettings: + if not is_object(value): + raise ValueError(f"{label} must contain a JSON object") + settings: DroidHookSettings = dict(value) + for key, item in value.items(): + if key in _FLAG_NAMES: + if not isinstance(item, bool): + raise ValueError(f'{label} field "{key}" must be a boolean') + continue + if key not in _EVENT_NAMES: + raise ValueError(f'{label} contains unsupported field "{key}"') + if not isinstance(item, list): + raise ValueError(f'{label} field "{key}" must be an array') + settings[key] = [ + _validate_group(group, f'{label} field "{key}"[{index}]') + for index, group in enumerate(item) + ] + return settings + + +def _regex_entries(settings: DroidHookSettings) -> List[JsonObject]: + entries: List[JsonObject] = [] + for event, groups in settings.items(): + if event in _FLAG_NAMES: + continue + for index, group in enumerate(groups): + matcher = group.get("matcher") + if isinstance(matcher, str) and matcher not in {"", "*"}: + entries.append({ + "label": f'Factory Droid hooks field "{event}"[{index}] matcher', + "pattern": matcher, + }) + command_regex = group.get("commandRegex") + if isinstance(command_regex, str): + entries.append({ + "label": ( + f'Factory Droid hooks field "{event}"[{index}] commandRegex' + ), + "pattern": command_regex, + }) + return entries + + +def validate_javascript_regexes(settings: Sequence[DroidHookSettings]) -> 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 Factory Droid 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( + "Factory Droid hook expression validation timed out after " + f"{_REGEX_TIMEOUT_SECONDS} seconds" + ) from error + except OSError as error: + raise OSError(f"Run Node.js Factory Droid 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( + "Factory Droid matcher must be a valid JavaScript regular expression: " + f"{message}" + ) + + +def _read_windows_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:index + 2] == '\\"': + value += '"' + index += 2 + continue + if command[index] == '"': + return value, index + 1 + value += command[index] + index += 1 + return None + + +_POSIX_APOSTROPHE = "'\"'\"'" + + +def _read_posix_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.startswith(_POSIX_APOSTROPHE, index): + value += "'" + index += len(_POSIX_APOSTROPHE) + 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]]: + reader = _read_windows_argument if os.name == "nt" else _read_posix_argument + executable = reader(command, 0) + if executable is None or executable[1] >= len(command): + return None + if command[executable[1]] != " ": + return None + script = reader(command, executable[1] + 1) + if script is None or script[1] != len(command): + return None + if 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 + if handler.get("type") != "command" or handler.get("timeout") != 10: + return None + command = handler.get("command") + if not isinstance(command, str): + return None + parsed = _parse_generated_command(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: + hooks = group["hooks"] + return ( + set(group) == _GROUP_KEYS + and group.get("matcher") == "*" + and bool(indexes) + and len(indexes) == len(hooks) + ) + + +def managed_removals( + settings: DroidHookSettings, + agent_id: Optional[str] = None, +) -> List[ManagedRemoval]: + removals: List[ManagedRemoval] = [] + 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: DroidHookSettings) -> 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): + if not any(same_agent_id(agent_id, audit_id) for audit_id in audits): + continue + 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 + + +def merge_hook_settings( + primary: Optional[DroidHookSettings], + fallback: Optional[DroidHookSettings], +) -> DroidHookSettings: + return {**(fallback or {}), **(primary or {})} diff --git a/sdks/python/elydora/plugins/droid_io.py b/sdks/python/elydora/plugins/droid_io.py new file mode 100644 index 0000000..9cd39f5 --- /dev/null +++ b/sdks/python/elydora/plugins/droid_io.py @@ -0,0 +1,371 @@ +"""Factory Droid source discovery and transactional multi-file I/O.""" + +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 .droid_config import ( + DroidSources, + RenderedDocument, + create_settings_document, + parse_document, +) +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 ( + os.path.join(directory, "hooks.json"), + os.path.join(directory, "hooks", "hooks.json"), + os.path.join(directory, "settings.json"), + ) + + +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") + primary = None + if root_raw is not None: + primary = parse_document( + exists=True, + file_path=root_path, + kind="hooks", + raw=root_raw, + ) + else: + legacy_raw = _read_optional(legacy_path, "Factory Droid legacy hooks") + if legacy_raw is not None: + primary = parse_document( + exists=True, + file_path=legacy_path, + kind="legacy", + raw=legacy_raw, + ) + settings = ( + create_settings_document(settings_path) + if settings_raw is None + else parse_document( + exists=True, + file_path=settings_path, + kind="settings", + raw=settings_raw, + ) + ) + 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 + if rendered.document.kind == "settings": + label = "Factory Droid settings" + elif rendered.document.kind == "legacy": + label = "Factory Droid legacy hooks" + else: + label = "Factory Droid hooks" + return FileChange( + 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") + 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: Sequence[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")) + config_agent_id = config.get("agent_id") if config else None + if ( + config is None + 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 + if regular_file_exists( + contract.guard_path, "Elydora guard runtime" + ) and regular_file_exists(contract.audit_path, "Elydora audit runtime"): + return True + return False + + +def display_config_path(sources: DroidSources) -> str: + if sources.primary is not None: + return sources.primary.file_path + if sources.settings.exists and sources.settings.has_hooks_container: + return sources.settings.file_path + return sources.root_path diff --git a/sdks/python/elydora/plugins/droid_jsonc.py b/sdks/python/elydora/plugins/droid_jsonc.py new file mode 100644 index 0000000..93e5ccd --- /dev/null +++ b/sdks/python/elydora/plugins/droid_jsonc.py @@ -0,0 +1,387 @@ +"""Strict JSONC parsing and source-preserving edits for Factory Droid.""" + +from __future__ import annotations + +from dataclasses import dataclass +import json +import re +from typing import Any, Dict, List, Optional, Sequence, Tuple, Union + + +JsonObject = Dict[str, Any] +JsonPathPart = Union[str, int] +_NUMBER = re.compile(r"-?(?:0|[1-9]\d*)(?:\.\d+)?(?:[eE][+-]?\d+)?") + + +@dataclass(frozen=True) +class JsoncItem: + node: "JsoncNode" + comma: Optional[int] + + +@dataclass(frozen=True) +class JsoncMember: + key: str + start: int + end: int + value_node: "JsoncNode" + comma: Optional[int] + + +@dataclass(frozen=True) +class JsoncNode: + kind: str + start: int + end: int + value: Any + members: Tuple[JsoncMember, ...] = () + items: Tuple[JsoncItem, ...] = () + + +class _Parser: + def __init__(self, raw: str, label: str) -> None: + self.raw = raw + self.label = label + self.position = 0 + + def _error(self, message: str, position: Optional[int] = None) -> ValueError: + offset = self.position if position is None else position + return ValueError(f"Failed to parse {self.label}: {message} at offset {offset}") + + def _skip_trivia(self) -> None: + while self.position < len(self.raw): + if self.position == 0 and self.raw.startswith("\ufeff"): + self.position += 1 + continue + character = self.raw[self.position] + if character in " \t\r\n": + self.position += 1 + continue + if self.raw.startswith("//", self.position): + newline = self.raw.find("\n", self.position + 2) + self.position = len(self.raw) if newline < 0 else newline + 1 + continue + if self.raw.startswith("/*", self.position): + end = self.raw.find("*/", self.position + 2) + if end < 0: + raise self._error("unterminated block comment") + self.position = end + 2 + continue + return + + def _expect(self, character: str) -> None: + if self.position >= len(self.raw) or self.raw[self.position] != character: + raise self._error(f'expected "{character}"') + self.position += 1 + + def _parse_string(self) -> Tuple[str, int, int]: + start = self.position + self._expect('"') + escaped = False + while self.position < len(self.raw): + character = self.raw[self.position] + self.position += 1 + if escaped: + escaped = False + continue + if character == "\\": + escaped = True + continue + if character == '"': + token = self.raw[start:self.position] + try: + value = json.loads(token) + except json.JSONDecodeError as error: + raise self._error(error.msg, start + error.pos) from error + return value, start, self.position + if ord(character) < 0x20: + raise self._error("unescaped control character", self.position - 1) + raise self._error("unterminated string", start) + + def _parse_object(self) -> JsoncNode: + start = self.position + self.position += 1 + members: List[JsoncMember] = [] + value: JsonObject = {} + seen = set() + self._skip_trivia() + if self.position < len(self.raw) and self.raw[self.position] == "}": + self.position += 1 + return JsoncNode("object", start, self.position, value) + while True: + self._skip_trivia() + if self.position >= len(self.raw) or self.raw[self.position] != '"': + raise self._error("object key must be a string") + key, member_start, _ = self._parse_string() + if key in seen: + raise self._error(f'duplicate field "{key}"', member_start) + seen.add(key) + self._skip_trivia() + self._expect(":") + node = self._parse_value() + value[key] = node.value + member_end = node.end + self._skip_trivia() + comma: Optional[int] = None + if self.position < len(self.raw) and self.raw[self.position] == ",": + comma = self.position + self.position += 1 + self._skip_trivia() + members.append(JsoncMember( + key, member_start, member_end, node, comma + )) + if self.position < len(self.raw) and self.raw[self.position] == "}": + self.position += 1 + return JsoncNode( + "object", start, self.position, value, tuple(members) + ) + continue + members.append(JsoncMember(key, member_start, member_end, node, None)) + if self.position >= len(self.raw) or self.raw[self.position] != "}": + raise self._error('expected "," or "}"') + self.position += 1 + return JsoncNode("object", start, self.position, value, tuple(members)) + + def _parse_array(self) -> JsoncNode: + start = self.position + self.position += 1 + items: List[JsoncItem] = [] + values: List[Any] = [] + self._skip_trivia() + if self.position < len(self.raw) and self.raw[self.position] == "]": + self.position += 1 + return JsoncNode("array", start, self.position, values) + while True: + node = self._parse_value() + values.append(node.value) + self._skip_trivia() + comma: Optional[int] = None + if self.position < len(self.raw) and self.raw[self.position] == ",": + comma = self.position + self.position += 1 + self._skip_trivia() + items.append(JsoncItem(node, comma)) + if self.position < len(self.raw) and self.raw[self.position] == "]": + self.position += 1 + return JsoncNode( + "array", start, self.position, values, items=tuple(items) + ) + continue + items.append(JsoncItem(node, None)) + if self.position >= len(self.raw) or self.raw[self.position] != "]": + raise self._error('expected "," or "]"') + self.position += 1 + return JsoncNode( + "array", start, self.position, values, items=tuple(items) + ) + + def _parse_value(self) -> JsoncNode: + self._skip_trivia() + if self.position >= len(self.raw): + raise self._error("expected a value") + start = self.position + character = self.raw[self.position] + if character == "{": + return self._parse_object() + if character == "[": + return self._parse_array() + if character == '"': + value, _, end = self._parse_string() + return JsoncNode("string", start, end, value) + for token, literal in (("true", True), ("false", False), ("null", None)): + if self.raw.startswith(token, self.position): + self.position += len(token) + return JsoncNode("scalar", start, self.position, literal) + match = _NUMBER.match(self.raw, self.position) + if match: + self.position = match.end() + token = match.group(0) + number = float(token) if any(marker in token for marker in ".eE") else int(token) + return JsoncNode("number", start, self.position, number) + raise self._error("invalid value") + + def parse(self) -> JsoncNode: + node = self._parse_value() + self._skip_trivia() + if self.position != len(self.raw): + raise self._error("unexpected content") + return node + + +def parse_jsonc(raw: str, label: str) -> Any: + """Parse strict JSON plus comments and trailing commas.""" + return _Parser(raw, label).parse().value + + +class JsoncEditor: + """Apply narrow path edits while retaining untouched source bytes.""" + + def __init__(self, raw: str, label: str) -> None: + self.raw = raw + self.label = label + self._refresh() + + @property + def value(self) -> Any: + return self.root.value + + def _refresh(self) -> None: + self.root = _Parser(self.raw, self.label).parse() + + def _resolve(self, path: Sequence[JsonPathPart]) -> JsoncNode: + node = self.root + for part in path: + if isinstance(part, str) and node.kind == "object": + member = next((item for item in node.members if item.key == part), None) + if member is None: + raise KeyError(f"Missing JSONC path: {list(path)}") + node = member.value_node + continue + if isinstance(part, int) and node.kind == "array": + if part < 0 or part >= len(node.items): + raise IndexError(f"Invalid JSONC path: {list(path)}") + node = node.items[part].node + continue + raise TypeError(f"Invalid JSONC path: {list(path)}") + return node + + def _apply(self, ranges: Sequence[Tuple[int, int, str]]) -> None: + next_raw = self.raw + for start, end, replacement in sorted(ranges, reverse=True): + next_raw = next_raw[:start] + replacement + next_raw[end:] + self.raw = next_raw + self._refresh() + + def delete(self, path: Sequence[JsonPathPart]) -> None: + if not path: + raise ValueError("Cannot delete the JSONC root") + parent = self._resolve(path[:-1]) + part = path[-1] + ranges: List[Tuple[int, int, str]] = [] + if isinstance(part, str) and parent.kind == "object": + index = next( + (position for position, item in enumerate(parent.members) if item.key == part), + -1, + ) + if index < 0: + raise KeyError(f"Missing JSONC path: {list(path)}") + member = parent.members[index] + ranges.append((member.start, member.end, "")) + if member.comma is not None: + ranges.append((member.comma, member.comma + 1, "")) + elif index > 0 and parent.members[index - 1].comma is not None: + comma = parent.members[index - 1].comma + if comma is None: + raise RuntimeError("JSONC object separator metadata is missing") + ranges.append((comma, comma + 1, "")) + elif isinstance(part, int) and parent.kind == "array": + if part < 0 or part >= len(parent.items): + raise IndexError(f"Invalid JSONC path: {list(path)}") + item = parent.items[part] + ranges.append((item.node.start, item.node.end, "")) + if item.comma is not None: + ranges.append((item.comma, item.comma + 1, "")) + elif part > 0 and parent.items[part - 1].comma is not None: + comma = parent.items[part - 1].comma + if comma is None: + raise RuntimeError("JSONC array separator metadata is missing") + ranges.append((comma, comma + 1, "")) + else: + raise TypeError(f"Invalid JSONC path: {list(path)}") + self._apply(ranges) + + def _eol(self) -> str: + return "\r\n" if "\r\n" in self.raw else "\n" + + def _indent_unit(self) -> str: + match = re.search(r"\r?\n([ \t]+)\S", self.raw) + if match is None: + return " " + indentation = match.group(1) + return "\t" if "\t" in indentation else " " * len(indentation) + + def _line_indent(self, position: int) -> str: + line_start = self.raw.rfind("\n", 0, position) + 1 + prefix = self.raw[line_start:position] + match = re.match(r"[ \t]*", prefix) + if match is None: + raise RuntimeError("JSONC indentation parser failed") + return match.group(0) + + def _json_lines(self, value: Any) -> List[str]: + unit = self._indent_unit() + lines = json.dumps(value, ensure_ascii=False, indent=1).splitlines() + converted = [] + for line in lines: + depth = len(line) - len(line.lstrip(" ")) + converted.append(unit * depth + line[depth:]) + return converted + + def _insert( + self, + parent: JsoncNode, + lines: List[str], + had_trailing_comma: bool, + ) -> None: + close = parent.end - 1 + eol = self._eol() + unit = self._indent_unit() + 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) + 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] + if close_line_start > parent.start and not close_prefix.strip(): + insertion_at = close_line_start + parent_indent = close_prefix + prefix = "" + suffix = eol + else: + insertion_at = close + parent_indent = self._line_indent(parent.start) + prefix = eol + suffix = eol + parent_indent + child_indent = parent_indent + unit + content = eol.join(child_indent + line for line in lines) + if had_trailing_comma: + content += "," + insertion = prefix + content + suffix + else: + insertion_at = close + content = " ".join(lines) + if had_trailing_comma: + content += "," + insertion = (" " if has_existing else "") + content + ranges.append((insertion_at, insertion_at, insertion)) + self._apply(ranges) + + def add_property( + self, + path: Sequence[JsonPathPart], + key: str, + value: Any, + ) -> None: + parent = self._resolve(path) + if parent.kind != "object": + raise TypeError(f"JSONC path must be an object: {list(path)}") + if any(member.key == key for member in parent.members): + raise KeyError(f'JSONC field "{key}" already exists') + value_lines = self._json_lines(value) + key_text = json.dumps(key, ensure_ascii=False) + ": " + lines = [key_text + value_lines[0], *value_lines[1:]] + trailing = bool(parent.members and parent.members[-1].comma is not None) + self._insert(parent, lines, trailing) + + def append(self, path: Sequence[JsonPathPart], value: Any) -> None: + parent = self._resolve(path) + if parent.kind != "array": + raise TypeError(f"JSONC path must be an array: {list(path)}") + trailing = bool(parent.items and parent.items[-1].comma is not None) + self._insert(parent, self._json_lines(value), trailing) diff --git a/sdks/python/elydora/plugins/registry.py b/sdks/python/elydora/plugins/registry.py index a71a372..273d6c9 100644 --- a/sdks/python/elydora/plugins/registry.py +++ b/sdks/python/elydora/plugins/registry.py @@ -36,6 +36,11 @@ SUPPORTED_AGENTS: Dict[str, Dict[str, str]] = { "hook_event": "PostToolUse", "config_path": "~/.cursor/hooks.json", }, + "droid": { + "name": "Factory Droid", + "hook_event": "PreToolUse/PostToolUse", + "config_path": "~/.factory/hooks.json", + }, "gemini": { "name": "Gemini CLI", "hook_event": "AfterTool", diff --git a/sdks/python/tests/test_droid_plugin.py b/sdks/python/tests/test_droid_plugin.py new file mode 100644 index 0000000..47e0243 --- /dev/null +++ b/sdks/python/tests/test_droid_plugin.py @@ -0,0 +1,446 @@ +from __future__ import annotations + +import base64 +import json +import os +from pathlib import Path +import subprocess # nosec B404 +from typing import Any, Dict, Optional + +import pytest + +from elydora import cli +from elydora.plugins import droid, droid_io, droid_jsonc +from elydora.plugins.base import InstallConfig +from elydora.plugins.registry import SUPPORTED_AGENTS + + +AGENT_ID = "agent-1" +JsonObject = Dict[str, Any] + + +class DroidFixture: + def __init__( + self, + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, + *, + hooks: Any = None, + legacy_hooks: Any = None, + settings: Any = None, + create_guard: bool = True, + ) -> None: + self.home_dir = tmp_path / "home with spaces and 'quote" + self.workspace_dir = self.home_dir / "workspace" + self.factory_dir = self.home_dir / ".factory" + self.config_path = self.factory_dir / "hooks.json" + self.legacy_path = self.factory_dir / "hooks" / "hooks.json" + self.settings_path = self.factory_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\nsys.stdin.read()\n" + "sys.stderr.write('Agent is frozen by Elydora.\\n')\n" + "raise SystemExit(2)\n", + encoding="utf-8", + ) + self._write(self.config_path, hooks) + self._write(self.legacy_path, legacy_hooks) + self._write(self.settings_path, settings) + monkeypatch.setenv("HOME", str(self.home_dir)) + monkeypatch.setenv("USERPROFILE", str(self.home_dir)) + private_key = base64.urlsafe_b64encode(bytes([1]) * 32).rstrip(b"=").decode() + self.config: InstallConfig = { + "agent_id": AGENT_ID, + "agent_name": "droid", + "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 = droid.DroidPlugin() + + @staticmethod + def _write(path: Path, value: Any) -> None: + if value is None: + return + 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) + + +def _load(path: Path) -> JsonObject: + value = droid_jsonc.parse_jsonc(path.read_text(encoding="utf-8"), str(path)) + assert isinstance(value, dict) + return value + + +def _managed_handler(groups: Any, script_path: Path) -> Optional[JsonObject]: + for group in groups or []: + for handler in group["hooks"]: + if str(script_path) in handler.get("command", ""): + return handler + return None + + +def _run_command(command: str, payload: str) -> subprocess.CompletedProcess[str]: + return subprocess.run( # nosec B602 + command, + shell=True, + input=payload, + text=True, + capture_output=True, + check=False, + timeout=10, + ) + + +def test_droid_is_registered_in_the_sdk_and_cli() -> None: + assert SUPPORTED_AGENTS["droid"] == { + "name": "Factory Droid", + "hook_event": "PreToolUse/PostToolUse", + "config_path": "~/.factory/hooks.json", + } + assert cli.PLUGIN_MAP["droid"] is droid.DroidPlugin + + +def test_install_preserves_jsonc_and_uses_per_event_precedence( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, + capsys: pytest.CaptureFixture[str], +) -> None: + hooks = """{ + // root hook source + "PreToolUse": [ + // keep root group comment + { "matcher": "Read", "hooks": [{ "type": "command", "command": "root-user" }] } + ], + "Notification": [] +} +""" + settings = """{ + // general setting + "theme": "dark", + "hooks": { + // settings fallback event + "PostToolUse": [ + // keep settings group comment + { "matcher": "Edit", "hooks": [{ "type": "command", "command": "settings-user" }] } + ], + "showHookOutput": true, + }, +} +""" + fixture = DroidFixture(monkeypatch, tmp_path, hooks=hooks, settings=settings) + fixture.install() + fixture.install() + output = capsys.readouterr().out + root_raw = fixture.config_path.read_text(encoding="utf-8") + settings_raw = fixture.settings_path.read_text(encoding="utf-8") + root = _load(fixture.config_path) + user_settings = _load(fixture.settings_path) + assert "root hook source" in root_raw + assert "keep root group comment" in root_raw + assert "general setting" in settings_raw + assert "settings fallback event" in settings_raw + assert "keep settings group comment" in settings_raw + assert len(root["PreToolUse"]) == 2 + assert root["PreToolUse"][0]["hooks"][0]["command"] == "root-user" + assert "PostToolUse" not in root + assert user_settings["theme"] == "dark" + assert len(user_settings["hooks"]["PostToolUse"]) == 2 + assert "PreToolUse" not in user_settings["hooks"] + for groups, path in ( + (root["PreToolUse"], fixture.guard_path), + (user_settings["hooks"]["PostToolUse"], fixture.audit_path), + ): + handler = _managed_handler(groups, path) + assert handler is not None + assert sorted(handler) == ["command", "timeout", "type"] + assert handler["type"] == "command" + assert handler["timeout"] == 10 + assert "run /hooks" in output + assert not (fixture.workspace_dir / ".factory" / "hooks.json").exists() + + +def test_install_keeps_active_legacy_source( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + fixture = DroidFixture( + monkeypatch, + tmp_path, + legacy_hooks={"PreToolUse": []}, + settings={"hooks": {"PostToolUse": []}}, + ) + fixture.install() + assert not fixture.config_path.exists() + assert _managed_handler(_load(fixture.legacy_path)["PreToolUse"], fixture.guard_path) + assert _managed_handler( + _load(fixture.settings_path)["hooks"]["PostToolUse"], + fixture.audit_path, + ) + + +def test_install_reuses_settings_container_and_formatting( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + source = '{\r\n\t"owner": "user",\r\n\t"hooks": {}\r\n}\r\n' + fixture = DroidFixture(monkeypatch, tmp_path, settings=source) + fixture.install() + assert not fixture.config_path.exists() + with open(fixture.settings_path, "r", encoding="utf-8", newline="") as file: + raw = file.read() + settings = _load(fixture.settings_path) + assert "\r\n\t\t\"PreToolUse\"" in raw + assert "\r\n\t\t\"PostToolUse\"" in raw + assert settings["owner"] == "user" + + +def test_commands_block_and_forward_official_input_byte_for_byte( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + fixture = DroidFixture(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", + ) + root = _load(fixture.config_path) + guard = _managed_handler(root["PreToolUse"], fixture.guard_path) + audit = _managed_handler(root["PostToolUse"], fixture.audit_path) + assert guard is not None and audit is not None + pre_payload = json.dumps({ + "session_id": "session-1", + "transcript_path": str(tmp_path / "transcript.jsonl"), + "cwd": str(fixture.workspace_dir), + "permission_mode": "auto-high", + "hook_event_name": "PreToolUse", + "tool_name": "Execute", + "tool_input": {"command": "echo test"}, + }) + guard_result = _run_command(guard["command"], pre_payload) + assert guard_result.returncode == 2 + assert "Agent is frozen by Elydora" in guard_result.stderr + post_payload = json.dumps({ + **json.loads(pre_payload), + "hook_event_name": "PostToolUse", + "tool_response": {"output": "test", "success": True}, + }) + audit_result = _run_command(audit["command"], 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_runtime_files( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + fixture = DroidFixture( + monkeypatch, + tmp_path, + hooks={"PreToolUse": []}, + settings={"hooks": {"PostToolUse": []}}, + ) + fixture.install() + assert fixture.plugin.status()["installed"] is True + fixture.audit_path.unlink() + assert fixture.plugin.status()["installed"] is False + root = _load(fixture.config_path) + root["hooksDisabled"] = True + fixture.config_path.write_text(json.dumps(root), encoding="utf-8") + status = fixture.plugin.status() + assert status["installed"] is False + assert "disabled" in status["details"].lower() + + +def test_uninstall_preserves_user_sources_and_exact_ownership( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + hooks = '{\n // keep root comment\n "PreToolUse": []\n}\n' + settings = ( + '{\n "theme": "dark",\n "hooks": {\n' + ' // keep settings comment\n "PostToolUse": []\n }\n}\n' + ) + fixture = DroidFixture(monkeypatch, tmp_path, hooks=hooks, settings=settings) + fixture.install() + fixture.plugin.uninstall("agent-10") + assert _managed_handler(_load(fixture.config_path)["PreToolUse"], fixture.guard_path) + uninstall_id = "AGENT-1" if os.name == "nt" else AGENT_ID + fixture.plugin.uninstall(uninstall_id) + root_raw = fixture.config_path.read_text(encoding="utf-8") + settings_raw = fixture.settings_path.read_text(encoding="utf-8") + assert "keep root comment" in root_raw + assert "keep settings comment" in settings_raw + assert _load(fixture.config_path)["PreToolUse"] == [] + current_settings = _load(fixture.settings_path) + assert current_settings["hooks"]["PostToolUse"] == [] + assert current_settings["theme"] == "dark" + + +def test_uninstall_deletes_only_an_owned_empty_hook_file( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + fixture = DroidFixture(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() + + +def test_uninstall_preserves_mixed_groups_and_lookalikes( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + fixture = DroidFixture(monkeypatch, tmp_path) + fixture.install() + root = _load(fixture.config_path) + group = next( + item for item in root["PreToolUse"] + if _managed_handler([item], fixture.guard_path) + ) + command = group["hooks"][0]["command"] + group["hooks"].append({"type": "command", "command": "user-command"}) + group["owner"] = "user" + root["PreToolUse"].extend([ + { + "matcher": "*", + "hooks": [{ + "type": "command", + "command": command.replace("guard.py", "guard.py.backup"), + "timeout": 10, + }], + }, + { + "matcher": "*", + "hooks": [{ + "type": "command", + "command": command.replace("agent-1", "agent-10"), + "timeout": 10, + }], + }, + ]) + fixture.config_path.write_text(json.dumps(root, indent=2) + "\n", encoding="utf-8") + fixture.plugin.uninstall(AGENT_ID) + remaining = _load(fixture.config_path) + raw = json.dumps(remaining) + assert "user-command" in raw + assert "guard.py.backup" in raw + assert "agent-10" in raw + assert remaining["PostToolUse"] == [] + + +@pytest.mark.parametrize( + ("hooks", "legacy", "settings", "pattern"), + [ + ("{ malformed", None, None, "parse Factory Droid hooks"), + ("[]", None, None, "JSON object"), + ('{ "PreToolUse": [], "PreToolUse": [] }', None, None, "duplicate"), + ({"PreToolUse": None}, None, None, "must be an array"), + ({"PreToolUse": [None]}, None, None, "must be an object"), + ({"PreToolUse": [{"matcher": "[", "hooks": []}]}, None, None, "regular expression"), + ({"PreToolUse": [{"hooks": [{"type": "command", "command": 1}]}]}, None, None, "string"), + (None, "{ malformed", None, "legacy hooks"), + ({"PreToolUse": []}, None, "{ malformed", "settings"), + (None, None, {"hooks": None}, "must contain a JSON object"), + ], +) +def test_install_preserves_every_malformed_source_before_writes( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, + hooks: Any, + legacy: Any, + settings: Any, + pattern: str, +) -> None: + fixture = DroidFixture( + monkeypatch, + tmp_path, + hooks=hooks, + legacy_hooks=legacy, + settings=settings, + ) + targets = [ + (fixture.config_path, hooks), + (fixture.legacy_path, legacy), + (fixture.settings_path, settings), + ] + originals = { + path: path.read_text(encoding="utf-8") for path, value in targets if value is not None + } + with pytest.raises((ValueError, RuntimeError), match=pattern): + fixture.install() + for path, raw in originals.items(): + assert path.read_text(encoding="utf-8") == raw + assert not fixture.audit_path.exists() + assert not fixture.runtime_config.exists() + assert not fixture.private_key_path.exists() + + +def test_install_rejects_missing_guard_before_creating_files( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + fixture = DroidFixture(monkeypatch, tmp_path, create_guard=False) + with pytest.raises(FileNotFoundError, match="guard runtime is missing"): + fixture.install() + assert not fixture.config_path.exists() + assert not fixture.audit_path.exists() + assert not fixture.runtime_config.exists() + + +def test_transaction_rolls_back_all_files_and_cleans_staging( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + fixture = DroidFixture( + monkeypatch, + tmp_path, + hooks={"PreToolUse": []}, + settings={"hooks": {"PostToolUse": []}}, + ) + original_root = fixture.config_path.read_text(encoding="utf-8") + original_settings = fixture.settings_path.read_text(encoding="utf-8") + real_replace = os.replace + failed = False + + def fail_settings_commit(source: Any, destination: Any) -> None: + nonlocal failed + if not failed and Path(destination) == fixture.settings_path: + failed = True + raise OSError("simulated settings commit failure") + real_replace(source, destination) + + monkeypatch.setattr(droid_io.os, "replace", fail_settings_commit) + with pytest.raises(OSError, match="Write Factory Droid installation"): + fixture.install() + assert failed is True + assert fixture.config_path.read_text(encoding="utf-8") == original_root + assert fixture.settings_path.read_text(encoding="utf-8") == original_settings + assert not fixture.audit_path.exists() + assert not fixture.runtime_config.exists() + assert not fixture.private_key_path.exists() + staging = [ + path for path in fixture.home_dir.rglob("*") + if path.suffix in {".tmp", ".rollback"} + ] + assert staging == []