feat: add native Cline hook integration

This commit is contained in:
ZacharyZhang-NY committed 2026-07-19 09:56:50 -04:00
1 parent 86152c8179
commit e839c77a4b
9 files changed
+1119 -3

No files matched your search

+1
View File
@@ -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. - 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. - 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. - 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 ## Code Quality
+3
View File
@@ -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. 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. 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. 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` | | Augment Code CLI | `augment` |
| Claude Code | `claudecode` | | Claude Code | `claudecode` |
| OpenAI Codex | `codex` | | OpenAI Codex | `codex` |
| Cline | `cline` |
| Kimi Code | `kimi` | | Kimi Code | `kimi` |
| Grok Build | `grok` | | Grok Build | `grok` |
| Copilot CLI | `copilot` | | Copilot CLI | `copilot` |
+2
View File
@@ -21,6 +21,7 @@ from .plugins.hook_template import generate_guard_script
from .plugins.augment import AugmentPlugin from .plugins.augment import AugmentPlugin
from .plugins.claudecode import ClaudeCodePlugin from .plugins.claudecode import ClaudeCodePlugin
from .plugins.codex import CodexPlugin from .plugins.codex import CodexPlugin
from .plugins.cline import ClinePlugin
from .plugins.copilot import CopilotPlugin from .plugins.copilot import CopilotPlugin
from .plugins.cursor import CursorPlugin from .plugins.cursor import CursorPlugin
from .plugins.gemini import GeminiPlugin from .plugins.gemini import GeminiPlugin
@@ -36,6 +37,7 @@ PLUGIN_MAP: Dict[str, Type[AgentPlugin]] = {
"augment": AugmentPlugin, "augment": AugmentPlugin,
"claudecode": ClaudeCodePlugin, "claudecode": ClaudeCodePlugin,
"codex": CodexPlugin, "codex": CodexPlugin,
"cline": ClinePlugin,
"copilot": CopilotPlugin, "copilot": CopilotPlugin,
"cursor": CursorPlugin, "cursor": CursorPlugin,
"gemini": GeminiPlugin, "gemini": GeminiPlugin,
+126
View File
@@ -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,
)
+288
View File
@@ -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,
)
+194
View File
@@ -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",
]
+17 -3
View File
@@ -206,9 +206,23 @@ def main():
session_id = "unknown" session_id = "unknown"
if isinstance(event, dict): if isinstance(event, dict):
tool_name = event.get("tool_name", event.get("toolName", event.get("name", "unknown"))) cline_event = event.get("tool_result")
tool_input = event.get("tool_input", event.get("toolInput", event.get("input", {{}}))) if not isinstance(cline_event, dict):
session_id = event.get("session_id", event.get("sessionId", event.get("session", "unknown"))) 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 # Read chain state
prev_chain_hash = read_chain_state() prev_chain_hash = read_chain_state()
+5
View File
@@ -21,6 +21,11 @@ SUPPORTED_AGENTS: Dict[str, Dict[str, str]] = {
"hook_event": "PreToolUse/PostToolUse", "hook_event": "PreToolUse/PostToolUse",
"config_path": "~/.codex/hooks.json", "config_path": "~/.codex/hooks.json",
}, },
"cline": {
"name": "Cline",
"hook_event": "PreToolUse/PostToolUse",
"config_path": "~/.cline/hooks/PreToolUse.mjs",
},
"copilot": { "copilot": {
"name": "Copilot CLI", "name": "Copilot CLI",
"hook_event": "PostToolUse", "hook_event": "PostToolUse",
+483
View File
@@ -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")) == []