From e839c77a4b8b94c237cc088675b4b88f7d224caa 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 09:56:50 -0400 Subject: [PATCH] feat: add native Cline hook integration --- AGENTS.md | 1 + README.md | 3 + elydora/cli.py | 2 + elydora/plugins/cline.py | 126 ++++++++ elydora/plugins/cline_contract.py | 288 ++++++++++++++++++ elydora/plugins/cline_io.py | 194 ++++++++++++ elydora/plugins/hook_template.py | 20 +- elydora/plugins/registry.py | 5 + tests/test_cline_plugin.py | 483 ++++++++++++++++++++++++++++++ 9 files changed, 1119 insertions(+), 3 deletions(-) create mode 100644 elydora/plugins/cline.py create mode 100644 elydora/plugins/cline_contract.py create mode 100644 elydora/plugins/cline_io.py create mode 100644 tests/test_cline_plugin.py diff --git a/AGENTS.md b/AGENTS.md index f74ecf1..6a7922d 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -26,6 +26,7 @@ This repository owns the published `elydora` package, its CLI, synchronous and a - Keep Grok Build writes inside its native global `$GROK_HOME/hooks/*.json` contract. Treat Claude Code and Cursor compatibility files plus project `.grok/hooks` as read-only integration sources. - 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 hooks only to `$CLINE_DIR/hooks` with `~/.cline/hooks` as the default; keep Documents and workspace hook roots read-only. Preserve official input byte-for-byte, resolve Python runtimes from their absolute shebangs, and translate guard exit code `2` into Cline's JSON stdout cancellation control. ## Code Quality diff --git a/README.md b/README.md index bda320d..2d14699 100644 --- a/README.md +++ b/README.md @@ -75,6 +75,8 @@ elydora install \ Codex performs a one-time trust review for user hooks. Run `/hooks` in Codex after installation and trust the Elydora `PreToolUse` and `PostToolUse` definitions. +Cline installation writes `PreToolUse.mjs` and `PostToolUse.mjs` to `$CLINE_DIR/hooks` (default `~/.cline/hooks`). Elydora leaves the Documents and workspace hook roots unchanged. The wrappers preserve Cline's official payload and translate a frozen guard into JSON stdout cancellation control. + Kimi installation writes the strict hook contract to each detected runtime: Kimi Code's `$KIMI_CODE_HOME/config.toml` (default `~/.kimi-code/config.toml`) and the migrating Python CLI's `~/.kimi/config.toml`. A fresh installation targets current Kimi Code, avoiding cross-runtime migration markers. Both runtimes load the hooks globally; run `/hooks` to inspect them. Grok Build installation writes native global hooks to `$GROK_HOME/hooks/elydora-audit.json` (default `~/.grok/hooks/elydora-audit.json`). Project hooks still follow Grok's `/hooks-trust` workflow; Elydora leaves project, Claude Code, and Cursor compatibility files unchanged. @@ -99,6 +101,7 @@ Kiro CLI installation covers both runtime contracts. Kiro CLI v2 uses the genera | Augment Code CLI | `augment` | | Claude Code | `claudecode` | | OpenAI Codex | `codex` | +| Cline | `cline` | | Kimi Code | `kimi` | | Grok Build | `grok` | | Copilot CLI | `copilot` | diff --git a/elydora/cli.py b/elydora/cli.py index 8686304..c694265 100644 --- a/elydora/cli.py +++ b/elydora/cli.py @@ -21,6 +21,7 @@ from .plugins.hook_template import generate_guard_script from .plugins.augment import AugmentPlugin from .plugins.claudecode import ClaudeCodePlugin from .plugins.codex import CodexPlugin +from .plugins.cline import ClinePlugin from .plugins.copilot import CopilotPlugin from .plugins.cursor import CursorPlugin from .plugins.gemini import GeminiPlugin @@ -36,6 +37,7 @@ PLUGIN_MAP: Dict[str, Type[AgentPlugin]] = { "augment": AugmentPlugin, "claudecode": ClaudeCodePlugin, "codex": CodexPlugin, + "cline": ClinePlugin, "copilot": CopilotPlugin, "cursor": CursorPlugin, "gemini": GeminiPlugin, diff --git a/elydora/plugins/cline.py b/elydora/plugins/cline.py new file mode 100644 index 0000000..bb1b524 --- /dev/null +++ b/elydora/plugins/cline.py @@ -0,0 +1,126 @@ +"""Cline native user-level file-hook integration.""" + +from __future__ import annotations + +import os + +from ._file_io import write_json_atomic, write_text_atomic +from .base import AgentPlugin, InstallConfig, PluginStatus +from .cline_contract import ( + AGENT_KEY, + AUDIT_SCRIPT, + HookFile, + build_metadata, + build_wrapper, + elydora_dir, + resolve_hook_files, + runtime_contract, +) +from .cline_io import ( + PendingWrite, + read_hook_file, + remove_owned_hooks, + require_available_hook_file, + require_runtime, + runtime_files_exist, + write_hook_pair, +) +from .hook_template import generate_hook_script + + +class ClinePlugin(AgentPlugin): + """Install Elydora into Cline's global file-hook directory.""" + + def install(self, config: InstallConfig) -> None: + agent_id = config.get("agent_id", "") + if not agent_id: + raise ValueError("agent_id is required") + paths = resolve_hook_files() + guard_state = read_hook_file(paths.guard_path) + audit_state = read_hook_file(paths.audit_path) + require_available_hook_file(guard_state) + require_available_hook_file(audit_state) + + guard_path = config.get("guard_script_path", "") + require_runtime(guard_path, "Elydora guard runtime") + agent_directory = os.path.join(elydora_dir(), agent_id) + audit_path = os.path.join(agent_directory, AUDIT_SCRIPT) + guard_metadata = build_metadata("guard", agent_id, guard_path) + audit_metadata = build_metadata("audit", agent_id, audit_path) + guard_source = build_wrapper(guard_metadata) + audit_source = build_wrapper(audit_metadata) + runtime_contract( + HookFile(True, paths.guard_path, guard_source, guard_metadata), + HookFile(True, paths.audit_path, audit_source, audit_metadata), + ) + + 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"), + ) + write_json_atomic( + os.path.join(agent_directory, "config.json"), + runtime_config, + 0o600, + "Elydora runtime config", + ) + write_text_atomic( + os.path.join(agent_directory, "private.key"), + config.get("private_key", ""), + 0o600, + "Elydora private key", + ) + write_text_atomic( + audit_path, + audit_script, + 0o700, + "Elydora audit runtime", + ) + write_hook_pair( + PendingWrite(guard_state, guard_source), + PendingWrite(audit_state, audit_source), + ) + print("Cline: user-level PreToolUse and PostToolUse hooks installed.") + + def uninstall(self, agent_id: str = "") -> None: + paths = resolve_hook_files() + guard_state = read_hook_file(paths.guard_path) + audit_state = read_hook_file(paths.audit_path) + remove_owned_hooks((guard_state, audit_state), agent_id) + + def status(self) -> PluginStatus: + paths = resolve_hook_files() + guard_state = read_hook_file(paths.guard_path) + audit_state = read_hook_file(paths.audit_path) + contract = runtime_contract(guard_state, audit_state) + if contract is None: + return PluginStatus( + installed=False, + agent=AGENT_KEY, + details="Not installed", + ) + installed = runtime_files_exist(contract) + details = ( + f"Config: {paths.hooks_directory}" + if installed + else ( + f"Configured at {paths.hooks_directory}; " + "runtime scripts missing" + ) + ) + return PluginStatus( + installed=installed, + agent=AGENT_KEY, + details=details, + ) diff --git a/elydora/plugins/cline_contract.py b/elydora/plugins/cline_contract.py new file mode 100644 index 0000000..81be90c --- /dev/null +++ b/elydora/plugins/cline_contract.py @@ -0,0 +1,288 @@ +"""Cline file-hook contract and exact Elydora ownership metadata.""" + +from __future__ import annotations + +import base64 +from dataclasses import dataclass +import json +import os +from typing import Any, Dict, Optional + + +AGENT_KEY = "cline" +GUARD_FILE_NAME = "PreToolUse.mjs" +AUDIT_FILE_NAME = "PostToolUse.mjs" +GUARD_SCRIPT = "guard.py" +AUDIT_SCRIPT = "hook.py" + +_METADATA_MARKER = "// @elydora-cline-hook " +_METADATA_VERSION = 1 + +JsonObject = Dict[str, Any] + + +@dataclass(frozen=True) +class HookMetadata: + version: int + kind: str + agent_id: str + runtime_path: str + + +@dataclass(frozen=True) +class HookFile: + exists: bool + file_path: str + source: Optional[str] = None + metadata: Optional[HookMetadata] = None + + +@dataclass(frozen=True) +class HookPaths: + hooks_directory: str + guard_path: str + audit_path: str + + +@dataclass(frozen=True) +class RuntimeContract: + agent_id: str + agent_directory: str + guard_path: str + audit_path: str + + +def home_dir() -> str: + return os.path.expanduser("~") + + +def elydora_dir() -> str: + return os.path.join(home_dir(), ".elydora") + + +def same_agent_id(left: str, right: str) -> bool: + return os.path.normcase(left) == os.path.normcase(right) + + +def _same_path(left: str, right: str) -> bool: + return os.path.normcase(os.path.abspath(left)) == os.path.normcase( + os.path.abspath(right) + ) + + +def resolve_hooks_directory() -> str: + configured = os.environ.get("CLINE_DIR", "").strip() + cline_directory = ( + os.path.abspath(configured) + if configured + else os.path.join(home_dir(), ".cline") + ) + return os.path.join(cline_directory, "hooks") + + +def resolve_hook_files() -> HookPaths: + hooks_directory = resolve_hooks_directory() + return HookPaths( + hooks_directory=hooks_directory, + guard_path=os.path.join(hooks_directory, GUARD_FILE_NAME), + audit_path=os.path.join(hooks_directory, AUDIT_FILE_NAME), + ) + + +def build_metadata(kind: str, agent_id: str, runtime_path: str) -> HookMetadata: + if kind not in {"guard", "audit"}: + raise ValueError('Cline hook kind must be "guard" or "audit"') + if not agent_id: + raise ValueError("agent_id is required") + if not os.path.isabs(runtime_path): + raise ValueError(f"Cline {kind} runtime path must be absolute") + return HookMetadata(_METADATA_VERSION, kind, agent_id, runtime_path) + + +def _metadata_object(metadata: HookMetadata) -> JsonObject: + return { + "version": metadata.version, + "kind": metadata.kind, + "agentId": metadata.agent_id, + "runtimePath": metadata.runtime_path, + } + + +def _encode_metadata(metadata: HookMetadata) -> str: + raw = json.dumps(_metadata_object(metadata), separators=(",", ":")) + return base64.urlsafe_b64encode(raw.encode("utf-8")).rstrip(b"=").decode("ascii") + + +def build_wrapper(metadata: HookMetadata) -> str: + template = """#!/usr/bin/env node +__METADATA_MARKER____METADATA__ +import { spawn } from 'node:child_process'; +import { readFileSync, statSync } from 'node:fs'; +import { isAbsolute } from 'node:path'; + +const hookKind = __HOOK_KIND__; +const runtimePath = __RUNTIME_PATH__; + +function resolveRuntimeExecutable() { + const firstLine = readFileSync(runtimePath, 'utf-8').split(/\\r?\\n/, 1)[0] ?? ''; + if (!firstLine.startsWith('#!')) throw new Error('runtime is missing an absolute shebang'); + const executable = firstLine.slice(2).trim(); + if (!isAbsolute(executable)) throw new Error('runtime shebang executable must be absolute'); + try { + if (!statSync(executable).isFile()) throw new Error('path is not a file'); + } catch (error) { + const message = error instanceof Error ? error.message : String(error); + throw new Error('runtime shebang executable is unavailable: ' + message, { cause: error }); + } + return executable; +} + +function runRuntime(input) { + return new Promise((resolve, reject) => { + const child = spawn(resolveRuntimeExecutable(), [runtimePath], { + stdio: ['pipe', 'pipe', 'pipe'], + windowsHide: true, + }); + const stdout = []; + const stderr = []; + let stdinError; + child.stdout.on('data', (chunk) => stdout.push(chunk)); + child.stderr.on('data', (chunk) => stderr.push(chunk)); + child.stdin.on('error', (error) => { + if (error?.code !== 'EPIPE') stdinError = error; + }); + child.once('error', reject); + child.once('close', (code, signal) => { + if (stdinError) { + reject(stdinError); + return; + } + resolve({ + code, + signal, + stdout: Buffer.concat(stdout).toString('utf-8'), + stderr: Buffer.concat(stderr).toString('utf-8'), + }); + }); + child.stdin.end(input); + }); +} + +async function main() { + const chunks = []; + for await (const chunk of process.stdin) chunks.push(chunk); + const result = await runRuntime(Buffer.concat(chunks)); + if (result.stdout) process.stderr.write(result.stdout); + if (result.stderr) process.stderr.write(result.stderr); + + if (hookKind === 'guard' && result.code === 2) { + const errorMessage = result.stderr.trim() || 'Agent is frozen by Elydora.'; + process.stdout.write('HOOK_CONTROL\\t' + JSON.stringify({ cancel: true, errorMessage }) + '\\n'); + return; + } + if (result.signal) throw new Error('runtime terminated by signal ' + result.signal); + if (result.code !== 0) throw new Error('runtime exited with code ' + result.code); +} + +main().catch((error) => { + const message = error instanceof Error ? error.stack || error.message : String(error); + process.stderr.write('Elydora Cline ' + hookKind + ' wrapper failed: ' + message + '\\n'); + process.exitCode = 1; +}); +""" + return ( + template.replace("__METADATA_MARKER__", _METADATA_MARKER) + .replace("__METADATA__", _encode_metadata(metadata)) + .replace("__HOOK_KIND__", json.dumps(metadata.kind)) + .replace("__RUNTIME_PATH__", json.dumps(metadata.runtime_path)) + ) + + +def _decode_metadata(encoded: str) -> JsonObject: + padding = "=" * (-len(encoded) % 4) + try: + raw = base64.urlsafe_b64decode((encoded + padding).encode("ascii")) + value = json.loads(raw.decode("utf-8")) + except (ValueError, UnicodeError, json.JSONDecodeError) as error: + raise ValueError(f"invalid encoded metadata: {error}") from error + if not isinstance(value, dict): + raise ValueError("metadata must be an object") + return value + + +def _validate_metadata(value: JsonObject) -> HookMetadata: + if set(value) != {"version", "kind", "agentId", "runtimePath"}: + raise ValueError("metadata contains an unexpected field set") + version = value["version"] + kind = value["kind"] + agent_id = value["agentId"] + runtime_path = value["runtimePath"] + if isinstance(version, bool) or version != _METADATA_VERSION: + raise ValueError("metadata version must be 1") + if kind not in {"guard", "audit"}: + raise ValueError('metadata kind must be "guard" or "audit"') + if not isinstance(agent_id, str) or not agent_id: + raise ValueError("metadata agentId must be a non-empty string") + if not isinstance(runtime_path, str) or not os.path.isabs(runtime_path): + raise ValueError("metadata runtimePath must be an absolute path") + return HookMetadata(version, kind, agent_id, runtime_path) + + +def parse_metadata(file_path: str, source: str) -> Optional[HookMetadata]: + lines = source.splitlines() + marker_line = lines[1] if len(lines) > 1 else "" + if not marker_line.startswith(_METADATA_MARKER): + return None + try: + encoded = marker_line[len(_METADATA_MARKER):] + return _validate_metadata(_decode_metadata(encoded)) + except ValueError as error: + raise ValueError( + f"Failed to parse Elydora Cline hook metadata at {file_path}: {error}" + ) from error + + +def assert_wrapper_integrity(file: HookFile) -> None: + if file.metadata is None or file.source is None: + return + if file.source != build_wrapper(file.metadata): + raise ValueError( + f"Elydora Cline hook at {file.file_path} does not match the managed template" + ) + + +def _validate_agent_segment(agent_id: str) -> None: + if agent_id in {".", ".."} or os.path.basename(agent_id) != agent_id: + raise ValueError("Elydora Cline hook metadata contains an invalid agentId") + + +def runtime_contract( + guard_file: HookFile, + audit_file: HookFile, +) -> Optional[RuntimeContract]: + if guard_file.metadata is None or audit_file.metadata is None: + return None + assert_wrapper_integrity(guard_file) + assert_wrapper_integrity(audit_file) + guard = guard_file.metadata + audit = audit_file.metadata + if guard.kind != "guard" or audit.kind != "audit": + raise ValueError("Elydora Cline hook files contain mismatched event metadata") + if not same_agent_id(guard.agent_id, audit.agent_id): + raise ValueError("Elydora Cline hook files reference different agents") + _validate_agent_segment(guard.agent_id) + agent_directory = os.path.join(elydora_dir(), guard.agent_id) + expected_guard = os.path.join(agent_directory, GUARD_SCRIPT) + expected_audit = os.path.join(agent_directory, AUDIT_SCRIPT) + if not _same_path(guard.runtime_path, expected_guard) or not _same_path( + audit.runtime_path, expected_audit + ): + raise ValueError( + "Elydora Cline hook metadata references an unexpected runtime path" + ) + return RuntimeContract( + agent_id=guard.agent_id, + agent_directory=agent_directory, + guard_path=expected_guard, + audit_path=expected_audit, + ) diff --git a/elydora/plugins/cline_io.py b/elydora/plugins/cline_io.py new file mode 100644 index 0000000..b678d4d --- /dev/null +++ b/elydora/plugins/cline_io.py @@ -0,0 +1,194 @@ +"""Fail-fast Cline hook I/O and runtime validation.""" + +from __future__ import annotations + +from dataclasses import dataclass +import os +import tempfile +from typing import List, Sequence, Tuple + +from ._file_io import ( + read_json, + regular_file_exists, + remove_file, + require_runtime, + write_text_atomic, +) +from .cline_contract import ( + AGENT_KEY, + HookFile, + RuntimeContract, + parse_metadata, + same_agent_id, +) + + +@dataclass(frozen=True) +class PendingWrite: + state: HookFile + source: str + + +@dataclass(frozen=True) +class StagedFile: + state: HookFile + temporary_path: str + + +def read_hook_file(file_path: str) -> HookFile: + try: + with open(file_path, "r", encoding="utf-8") as file: + source = file.read() + except FileNotFoundError: + return HookFile(False, file_path) + except OSError as error: + raise OSError(f"Read Cline hook at {file_path}: {error}") from error + return HookFile( + exists=True, + file_path=file_path, + source=source, + metadata=parse_metadata(file_path, source), + ) + + +def require_available_hook_file(file: HookFile) -> None: + if file.exists and file.metadata is None: + raise ValueError( + f"Cline hook at {file.file_path} already exists and is owned " + "by another integration" + ) + + +def _remove_temporary(path: str) -> None: + try: + os.remove(path) + except FileNotFoundError: + return + + +def _stage_file(write: PendingWrite) -> StagedFile: + directory = os.path.dirname(write.state.file_path) + try: + os.makedirs(directory, mode=0o700, exist_ok=True) + except OSError as error: + raise OSError(f"Create Cline hooks directory at {directory}: {error}") from error + descriptor = -1 + temporary_path = "" + try: + descriptor, temporary_path = tempfile.mkstemp( + prefix=f".{os.path.basename(write.state.file_path)}.", + suffix=".tmp", + dir=directory, + text=True, + ) + with os.fdopen(descriptor, "w", encoding="utf-8", newline="") as file: + descriptor = -1 + file.write(write.source) + file.flush() + os.fsync(file.fileno()) + os.chmod(temporary_path, 0o700) + return StagedFile(write.state, temporary_path) + except Exception as error: + if descriptor >= 0: + try: + os.close(descriptor) + except OSError as close_error: + _remove_temporary(temporary_path) + raise OSError( + f"Stage Cline hook at {write.state.file_path}: {error}; " + f"close failed: {close_error}" + ) from error + _remove_temporary(temporary_path) + raise OSError( + f"Stage Cline hook at {write.state.file_path}: {error}" + ) from error + + +def _rollback_file(state: HookFile) -> None: + if state.exists: + if state.source is None: + raise RuntimeError("Existing Cline hook rollback state is missing source") + write_text_atomic( + state.file_path, + state.source, + 0o700, + "Cline hook rollback", + ) + else: + remove_file(state.file_path, "Cline hook rollback") + + +def write_hook_pair(guard: PendingWrite, audit: PendingWrite) -> None: + staged: List[StagedFile] = [] + try: + staged.append(_stage_file(guard)) + staged.append(_stage_file(audit)) + except Exception as error: + cleanup_errors: List[str] = [] + for item in staged: + try: + _remove_temporary(item.temporary_path) + except OSError as cleanup_error: + cleanup_errors.append(str(cleanup_error)) + suffix = f"; cleanup failed: {'; '.join(cleanup_errors)}" if cleanup_errors else "" + raise OSError(f"Stage Cline hook pair: {error}{suffix}") from error + + committed: List[StagedFile] = [] + try: + for item in staged: + os.replace(item.temporary_path, item.state.file_path) + committed.append(item) + except Exception as error: + recovery_errors: List[str] = [] + for item in reversed(committed): + try: + _rollback_file(item.state) + except (OSError, RuntimeError) as rollback_error: + recovery_errors.append(str(rollback_error)) + for item in staged[len(committed):]: + try: + _remove_temporary(item.temporary_path) + 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"Write Cline hook pair: {error}{suffix}") from error + + +def remove_owned_hooks(files: Sequence[HookFile], agent_id: str = "") -> None: + for file in files: + owned_agent_id = file.metadata.agent_id if file.metadata else "" + if not owned_agent_id or ( + agent_id and not same_agent_id(owned_agent_id, agent_id) + ): + continue + remove_file(file.file_path, "Cline hook") + + +def runtime_files_exist(contract: RuntimeContract) -> bool: + config_path = os.path.join(contract.agent_directory, "config.json") + config = read_json(config_path, "Elydora runtime config") + if config is None: + return False + config_agent_id = config.get("agent_id") + if ( + config.get("agent_name") != AGENT_KEY + or not isinstance(config_agent_id, str) + or not same_agent_id(config_agent_id, contract.agent_id) + ): + return False + files: Tuple[Tuple[str, str], ...] = ( + (contract.guard_path, "Elydora guard runtime"), + (contract.audit_path, "Elydora audit runtime"), + ) + return all(regular_file_exists(path, label) for path, label in files) + + +__all__ = [ + "PendingWrite", + "read_hook_file", + "remove_owned_hooks", + "require_available_hook_file", + "require_runtime", + "runtime_files_exist", + "write_hook_pair", +] diff --git a/elydora/plugins/hook_template.py b/elydora/plugins/hook_template.py index 9e59879..0857ed2 100644 --- a/elydora/plugins/hook_template.py +++ b/elydora/plugins/hook_template.py @@ -206,9 +206,23 @@ def main(): session_id = "unknown" if isinstance(event, dict): - tool_name = event.get("tool_name", event.get("toolName", event.get("name", "unknown"))) - tool_input = event.get("tool_input", event.get("toolInput", event.get("input", {{}}))) - session_id = event.get("session_id", event.get("sessionId", event.get("session", "unknown"))) + cline_event = event.get("tool_result") + if not isinstance(cline_event, dict): + cline_event = event.get("tool_call") + if not isinstance(cline_event, dict): + cline_event = {{}} + tool_name = event.get( + "tool_name", + event.get("toolName", event.get("name", cline_event.get("name", "unknown"))), + ) + tool_input = event.get( + "tool_input", + event.get("toolInput", event.get("input", cline_event.get("input", {{}}))), + ) + session_id = event.get( + "session_id", + event.get("sessionId", event.get("session", event.get("taskId", "unknown"))), + ) # Read chain state prev_chain_hash = read_chain_state() diff --git a/elydora/plugins/registry.py b/elydora/plugins/registry.py index 5024aba..a71a372 100644 --- a/elydora/plugins/registry.py +++ b/elydora/plugins/registry.py @@ -21,6 +21,11 @@ SUPPORTED_AGENTS: Dict[str, Dict[str, str]] = { "hook_event": "PreToolUse/PostToolUse", "config_path": "~/.codex/hooks.json", }, + "cline": { + "name": "Cline", + "hook_event": "PreToolUse/PostToolUse", + "config_path": "~/.cline/hooks/PreToolUse.mjs", + }, "copilot": { "name": "Copilot CLI", "hook_event": "PostToolUse", diff --git a/tests/test_cline_plugin.py b/tests/test_cline_plugin.py new file mode 100644 index 0000000..6c175e7 --- /dev/null +++ b/tests/test_cline_plugin.py @@ -0,0 +1,483 @@ +from __future__ import annotations + +import base64 +from dataclasses import dataclass +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +import json +import os +from pathlib import Path +from queue import Queue +import shutil +import subprocess # nosec B404 +import sys +from threading import Thread +from typing import Any, Optional + +import pytest + +from elydora import cli +from elydora.plugins import cline +from elydora.plugins import cline_contract, cline_io +from elydora.plugins.base import InstallConfig +from elydora.plugins.registry import SUPPORTED_AGENTS + + +AGENT_ID = "agent-1" +MISSING = object() + + +@dataclass(frozen=True) +class ClineFixture: + plugin: cline.ClinePlugin + config: InstallConfig + home_dir: Path + workspace_dir: Path + cline_dir: Path + hooks_dir: Path + agent_dir: Path + guard_path: Path + audit_path: Path + guard_wrapper: Path + audit_wrapper: Path + runtime_config: Path + + +def _python_script(body: str) -> str: + return f"#!{sys.executable}\n{body}" + + +def prepare_fixture( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, + *, + auto_install: bool = True, + create_guard: bool = True, + existing_audit: Any = MISSING, + existing_guard: Any = MISSING, + guard_source: Optional[str] = None, +) -> ClineFixture: + home_dir = tmp_path / "home with spaces" + workspace_dir = tmp_path / "workspace" + cline_dir = tmp_path / "custom-cline-home" + hooks_dir = cline_dir / "hooks" + agent_dir = home_dir / ".elydora" / AGENT_ID + guard_path = agent_dir / "guard.py" + audit_path = agent_dir / "hook.py" + guard_wrapper = hooks_dir / "PreToolUse.mjs" + audit_wrapper = hooks_dir / "PostToolUse.mjs" + workspace_dir.mkdir(parents=True) + agent_dir.mkdir(parents=True) + if create_guard: + guard_path.write_text( + guard_source + or _python_script( + "import sys\n" + "sys.stdin.read()\n" + "sys.stderr.write('Agent is frozen by Elydora.\\n')\n" + "raise SystemExit(2)\n" + ), + encoding="utf-8", + ) + if existing_guard is not MISSING: + hooks_dir.mkdir(parents=True, exist_ok=True) + guard_wrapper.write_text(str(existing_guard), encoding="utf-8") + if existing_audit is not MISSING: + hooks_dir.mkdir(parents=True, exist_ok=True) + audit_wrapper.write_text(str(existing_audit), encoding="utf-8") + + monkeypatch.setenv("HOME", str(home_dir)) + monkeypatch.setenv("USERPROFILE", str(home_dir)) + monkeypatch.setenv("CLINE_DIR", str(cline_dir)) + private_key = base64.urlsafe_b64encode(bytes([1]) * 32).rstrip(b"=").decode() + config: InstallConfig = { + "agent_id": AGENT_ID, + "agent_name": "cline", + "org_id": "org-1", + "private_key": private_key, + "kid": "kid-1", + "base_url": "https://api.elydora.test", + "guard_script_path": str(guard_path), + } + fixture = ClineFixture( + plugin=cline.ClinePlugin(), + config=config, + home_dir=home_dir, + workspace_dir=workspace_dir, + cline_dir=cline_dir, + hooks_dir=hooks_dir, + agent_dir=agent_dir, + guard_path=guard_path, + audit_path=audit_path, + guard_wrapper=guard_wrapper, + audit_wrapper=audit_wrapper, + runtime_config=agent_dir / "config.json", + ) + if auto_install: + fixture.plugin.install(config) + return fixture + + +def run_wrapper( + wrapper: Path, + fixture: ClineFixture, + payload: dict[str, Any], +) -> subprocess.CompletedProcess[str]: + node = shutil.which("node") + assert node is not None + return subprocess.run( # nosec B603 + [node, str(wrapper)], + cwd=fixture.workspace_dir, + env={ + **os.environ, + "HOME": str(fixture.home_dir), + "USERPROFILE": str(fixture.home_dir), + }, + input=json.dumps(payload), + text=True, + capture_output=True, + check=False, + timeout=10, + ) + + +def test_cline_is_registered_in_the_sdk_and_cli() -> None: + assert SUPPORTED_AGENTS["cline"] == { + "name": "Cline", + "hook_event": "PreToolUse/PostToolUse", + "config_path": "~/.cline/hooks/PreToolUse.mjs", + } + assert cli.PLUGIN_MAP["cline"] is cline.ClinePlugin + + +def test_install_writes_only_native_global_hooks_and_is_idempotent( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + fixture = prepare_fixture(monkeypatch, tmp_path) + guard_source = fixture.guard_wrapper.read_text(encoding="utf-8") + audit_source = fixture.audit_wrapper.read_text(encoding="utf-8") + + fixture.plugin.install(fixture.config) + + assert fixture.guard_wrapper.read_text(encoding="utf-8") == guard_source + assert fixture.audit_wrapper.read_text(encoding="utf-8") == audit_source + assert guard_source.startswith("#!/usr/bin/env node\n// @elydora-cline-hook ") + assert list(fixture.hooks_dir.glob("*.tmp")) == [] + assert not (fixture.home_dir / "Documents" / "Cline" / "Hooks").exists() + assert not (fixture.workspace_dir / ".cline" / "hooks").exists() + assert not (fixture.workspace_dir / ".clinerules" / "hooks").exists() + + +def test_install_uses_official_default_without_cline_dir( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + fixture = prepare_fixture(monkeypatch, tmp_path, auto_install=False) + monkeypatch.delenv("CLINE_DIR") + + fixture.plugin.install(fixture.config) + + default_hooks = fixture.home_dir / ".cline" / "hooks" + assert (default_hooks / "PreToolUse.mjs").is_file() + assert (default_hooks / "PostToolUse.mjs").is_file() + assert not fixture.guard_wrapper.exists() + + +def test_wrappers_translate_freezes_and_forward_payload_byte_for_byte( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + fixture = prepare_fixture(monkeypatch, tmp_path) + capture_path = tmp_path / "captured-event.json" + fixture.audit_path.write_text( + _python_script( + "from pathlib import Path\n" + "import os, sys\n" + f"Path({str(capture_path)!r}).write_text(" + "json.dumps({'cwd': os.getcwd(), 'input': sys.stdin.read()}), " + "encoding='utf-8')\n" + ).replace("import os, sys", "import json, os, sys"), + encoding="utf-8", + ) + pre_payload = { + "clineVersion": "3.0.46", + "hookName": "tool_call", + "taskId": "task-1", + "tool_call": {"name": "read_file", "input": {"path": "README.md"}}, + } + guard = run_wrapper(fixture.guard_wrapper, fixture, pre_payload) + assert guard.returncode == 0 + assert "Agent is frozen by Elydora" in guard.stderr + marker, control = guard.stdout.strip().split("\t", 1) + assert marker == "HOOK_CONTROL" + assert json.loads(control) == { + "cancel": True, + "errorMessage": "Agent is frozen by Elydora.", + } + + post_payload = { + "clineVersion": "3.0.46", + "hookName": "tool_result", + "taskId": "task-1", + "tool_result": {"name": "read_file", "input": {"path": "README.md"}}, + } + audit = run_wrapper(fixture.audit_wrapper, fixture, post_payload) + assert audit.returncode == 0 + captured = json.loads(capture_path.read_text(encoding="utf-8")) + assert captured == {"cwd": str(fixture.workspace_dir), "input": json.dumps(post_payload)} + + +def test_wrappers_keep_passes_quiet_and_surface_runtime_failures( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + passing = prepare_fixture( + monkeypatch, + tmp_path / "passing", + guard_source=_python_script("import sys\nsys.stdin.read()\n"), + ) + result = run_wrapper(passing.guard_wrapper, passing, {}) + assert result.returncode == 0 + assert result.stdout == "" + + failing = prepare_fixture(monkeypatch, tmp_path / "failing") + failing.audit_path.write_text( + _python_script( + "import sys\n" + "sys.stderr.write('audit failed\\n')\n" + "raise SystemExit(7)\n" + ), + encoding="utf-8", + ) + result = run_wrapper(failing.audit_wrapper, failing, {}) + assert result.returncode == 1 + assert "audit failed" in result.stderr + assert "exited with code 7" in result.stderr + + +def test_wrapper_rejects_a_non_absolute_runtime_shebang( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + fixture = prepare_fixture( + monkeypatch, + tmp_path, + guard_source="#!python.exe\nimport sys\nsys.stdin.read()\n", + ) + + result = run_wrapper(fixture.guard_wrapper, fixture, {}) + + assert result.returncode == 1 + assert result.stdout == "" + assert "runtime shebang executable must be absolute" in result.stderr + + +class _OperationHandler(BaseHTTPRequestHandler): + operations: Queue[dict[str, Any]] + + def do_POST(self) -> None: # noqa: N802 + length = int(self.headers["Content-Length"]) + body = json.loads(self.rfile.read(length).decode("utf-8")) + self.operations.put(body) + self.send_response(201) + self.end_headers() + self.wfile.write(b"{}") + + def log_message(self, format: str, *args: Any) -> None: + return + + +def test_audit_runtime_maps_official_nested_fields_into_the_operation( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + operations: Queue[dict[str, Any]] = Queue() + _OperationHandler.operations = operations + server = ThreadingHTTPServer(("127.0.0.1", 0), _OperationHandler) + thread = Thread(target=server.serve_forever, daemon=True) + thread.start() + fixture = prepare_fixture(monkeypatch, tmp_path, auto_install=False) + fixture.config["base_url"] = f"http://127.0.0.1:{server.server_port}" + try: + fixture.plugin.install(fixture.config) + payload = { + "hookName": "tool_result", + "taskId": "task-1", + "tool_result": { + "name": "read_file", + "input": {"path": "README.md"}, + "output": "ok", + }, + } + result = run_wrapper(fixture.audit_wrapper, fixture, payload) + assert result.returncode == 0 + operation = operations.get(timeout=3) + assert operation["payload"] == { + "tool_name": "read_file", + "tool_input": {"path": "README.md"}, + "session_id": "task-1", + } + assert operation["action"] == {"tool": "read_file"} + assert operation["subject"] == {"session_id": "task-1"} + finally: + server.shutdown() + server.server_close() + thread.join(timeout=3) + + +def test_status_requires_intact_hooks_and_runtimes( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + fixture = prepare_fixture(monkeypatch, tmp_path) + assert fixture.plugin.status() == { + "installed": True, + "agent": "cline", + "details": f"Config: {fixture.hooks_dir}", + } + fixture.guard_path.unlink() + assert fixture.plugin.status()["installed"] is False + fixture.guard_path.write_text(_python_script("pass\n"), encoding="utf-8") + fixture.audit_wrapper.unlink() + assert fixture.plugin.status() == { + "installed": False, + "agent": "cline", + "details": "Not installed", + } + + +def test_status_surfaces_corrupt_hooks_and_runtime_metadata( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + fixture = prepare_fixture(monkeypatch, tmp_path) + fixture.guard_wrapper.write_text( + fixture.guard_wrapper.read_text(encoding="utf-8") + "\n// tampered\n", + encoding="utf-8", + ) + with pytest.raises(ValueError, match="managed template"): + fixture.plugin.status() + + fixture.plugin.install(fixture.config) + fixture.runtime_config.write_text("{ malformed", encoding="utf-8") + with pytest.raises(ValueError, match="parse Elydora runtime config"): + fixture.plugin.status() + + +@pytest.mark.parametrize("collision", ["guard", "audit"]) +def test_install_rejects_user_filename_collisions_before_writes( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, + collision: str, +) -> None: + existing_guard = "// user PreToolUse hook\n" if collision == "guard" else MISSING + existing_audit = "// user PostToolUse hook\n" if collision == "audit" else MISSING + fixture = prepare_fixture( + monkeypatch, + tmp_path, + auto_install=False, + existing_guard=existing_guard, + existing_audit=existing_audit, + ) + with pytest.raises(ValueError, match="owned by another integration"): + fixture.plugin.install(fixture.config) + assert fixture.audit_path.exists() is False + assert fixture.runtime_config.exists() is False + if collision == "guard": + assert fixture.guard_wrapper.read_text(encoding="utf-8") == existing_guard + assert fixture.audit_wrapper.exists() is False + else: + assert fixture.audit_wrapper.read_text(encoding="utf-8") == existing_audit + assert fixture.guard_wrapper.exists() is False + + +def test_install_preserves_corrupt_metadata_for_recovery( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + corrupt = "#!/usr/bin/env node\n// @elydora-cline-hook invalid\n" + fixture = prepare_fixture( + monkeypatch, + tmp_path, + auto_install=False, + existing_guard=corrupt, + ) + with pytest.raises(ValueError, match="parse Elydora Cline hook metadata"): + fixture.plugin.install(fixture.config) + assert fixture.guard_wrapper.read_text(encoding="utf-8") == corrupt + assert fixture.audit_wrapper.exists() is False + assert fixture.audit_path.exists() is False + + +def test_install_rejects_missing_guard_before_creating_files( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + fixture = prepare_fixture( + monkeypatch, + tmp_path, + auto_install=False, + create_guard=False, + ) + with pytest.raises(FileNotFoundError, match="guard runtime is missing"): + fixture.plugin.install(fixture.config) + assert fixture.guard_wrapper.exists() is False + assert fixture.audit_wrapper.exists() is False + assert fixture.audit_path.exists() is False + assert fixture.runtime_config.exists() is False + + +def test_uninstall_removes_exact_ownership_and_preserves_other_hooks( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + fixture = prepare_fixture(monkeypatch, tmp_path) + user_hook = fixture.hooks_dir / "PreToolUse.py" + user_hook.write_text("# user hook\n", encoding="utf-8") + + fixture.plugin.uninstall("agent-10") + assert fixture.guard_wrapper.is_file() + assert fixture.audit_wrapper.is_file() + fixture.plugin.uninstall(AGENT_ID) + assert fixture.guard_wrapper.exists() is False + assert fixture.audit_wrapper.exists() is False + assert user_hook.read_text(encoding="utf-8") == "# user hook\n" + + +def test_hook_pair_restores_the_first_hook_when_the_second_commit_fails( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + fixture = prepare_fixture(monkeypatch, tmp_path) + original_guard = fixture.guard_wrapper.read_text(encoding="utf-8") + original_audit = fixture.audit_wrapper.read_text(encoding="utf-8") + guard_state = cline_io.read_hook_file(str(fixture.guard_wrapper)) + audit_state = cline_io.read_hook_file(str(fixture.audit_wrapper)) + replacement_guard = cline_contract.build_wrapper( + cline_contract.build_metadata("guard", "agent-2", str(fixture.guard_path)) + ) + replacement_audit = cline_contract.build_wrapper( + cline_contract.build_metadata("audit", "agent-2", str(fixture.audit_path)) + ) + real_replace = os.replace + failure_injected = False + + def fail_audit_commit(source: Any, destination: Any) -> None: + nonlocal failure_injected + if not failure_injected and Path(destination) == fixture.audit_wrapper: + failure_injected = True + raise OSError("simulated audit commit failure") + real_replace(source, destination) + + monkeypatch.setattr(os, "replace", fail_audit_commit) + + with pytest.raises(OSError, match="Write Cline hook pair"): + cline_io.write_hook_pair( + cline_io.PendingWrite(guard_state, replacement_guard), + cline_io.PendingWrite(audit_state, replacement_audit), + ) + + assert failure_injected is True + assert fixture.guard_wrapper.read_text(encoding="utf-8") == original_guard + assert fixture.audit_wrapper.read_text(encoding="utf-8") == original_audit + assert list(fixture.hooks_dir.glob("*.tmp")) == []