feat: sync Python Cline hook integration

This commit is contained in:
ZacharyZhang-NY committed 2026-07-19 10:00:33 -04:00
1 parent 619fe1659a
commit 008999a0c1
10 files changed
+1120 -5

No files matched your search

+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.
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` |
+2
View File
@@ -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,
+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,
)
@@ -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"
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()
+5
View File
@@ -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",
+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")) == []