From 909f3dd1062e56cac1df56a9cb7dac2204e3d209 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E9=9B=B7=E7=94=B5=E8=8A=BD=E8=A1=A3?= Date: Mon, 20 Jul 2026 18:42:04 -0400 Subject: [PATCH] fix(letta): harden official Python hook integration --- AGENTS.md | 1 + README.md | 2 + elydora/_runtime_paths.py | 5 + elydora/cli.py | 19 +- elydora/plugins/letta.py | 244 ++++------------ elydora/plugins/letta_command.py | 175 ++++++++++++ elydora/plugins/letta_config.py | 195 +++++++++++++ elydora/plugins/letta_contract.py | 277 ++++++++++++++++++ elydora/plugins/letta_installation.py | 205 +++++++++++++ elydora/plugins/letta_io.py | 201 +++++++++++++ elydora/plugins/letta_sources.py | 159 +++++++++++ elydora/plugins/registry.py | 2 +- tests/letta_support.py | 289 +++++++++++++++++++ tests/test_cli_security.py | 2 - tests/test_letta_official.py | 180 ++++++++++++ tests/test_letta_plugin.py | 395 ++++++++++++++++++++++++++ tests/test_letta_runtime.py | 147 ++++++++++ tests/test_letta_transaction.py | 111 ++++++++ tests/test_settings_hook_detection.py | 46 ++- 19 files changed, 2443 insertions(+), 212 deletions(-) create mode 100644 elydora/plugins/letta_command.py create mode 100644 elydora/plugins/letta_config.py create mode 100644 elydora/plugins/letta_contract.py create mode 100644 elydora/plugins/letta_installation.py create mode 100644 elydora/plugins/letta_io.py create mode 100644 elydora/plugins/letta_sources.py create mode 100644 tests/letta_support.py create mode 100644 tests/test_letta_official.py create mode 100644 tests/test_letta_plugin.py create mode 100644 tests/test_letta_runtime.py create mode 100644 tests/test_letta_transaction.py diff --git a/AGENTS.md b/AGENTS.md index 4ea17df..12f3c11 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -36,6 +36,7 @@ This repository owns the published `elydora` package, its CLI, synchronous and a - Resolve Factory Droid 0.175.0 user hooks through `~/.factory/hooks.json`, the legacy `~/.factory/hooks/hooks.json`, then the user `settings.local.json` and `settings.json` hook containers. Current hook files use a top-level `hooks` object; retain direct event maps as an explicit legacy read contract until Factory migrates them. Register exact `PreToolUse` and `PostToolUse` groups with ten-second absolute command handlers in one active source, preserve the native snake_case payload, and propagate freeze or revocation through exit code `2`. Use PowerShell's call operator with single-quoted absolute arguments and explicit `$LASTEXITCODE` propagation on Windows, and migrate the legacy double-quoted command form through exact ownership checks. Remove exact Elydora ownership from inactive user sources while preserving JSONC comments and unknown event arrays. Keep project, folder, plugin, and organization hook definitions read-only. Resolve local settings over their matching base file, then apply system, project, folder, and user `hooksDisabled` through Factory's extension-only scalar precedence; reject system-managed `allowManagedHooksOnly`. Commit user hook sources, generated runtimes, runtime config, and private key through one rollback-capable transaction, precondition every parsed source and policy file, require exact generated runtime sources during status checks, and require `/hooks` review for the session snapshot and remote organization policy. - Resolve Qwen Code 0.20.0 user settings through explicit `QWEN_HOME`, then the initial Qwen home `.env`, then the OS home `.env`; when bootstrap discovers a redirected home, read that home's `.env` as well. Treat empty overrides as unset while preserving their process-level ownership semantics. Read system defaults, user settings, the active trusted workspace, system overrides, and trusted-folder rules with Qwen's documented precedence; write only the user source. Validate the current twenty-one event names and command, HTTP, and prompt handler schemas while preserving future event payloads. Register exact matchless `PreToolUse`, `PostToolUse`, and `PostToolUseFailure` groups named `elydora-guard` and `elydora-audit` with ten-second millisecond timeouts. Preserve the complete native snake_case payload, propagate freeze and revocation through exit code `2`, and use PowerShell's call operator with explicit `$LASTEXITCODE` propagation on Windows. - Commit Qwen Code user settings, generated runtimes, runtime config, and private key through one rollback-capable transaction, with snapshot preconditions for every consulted routing, policy, trust, and settings source. Status requires physical files, exact runtime sources, a canonical private key, matching runtime identity, and an effective enabled state. Verify discovery with the official 0.20.0 loader, review `/hooks`, and restart active Qwen Code sessions. +- Resolve Letta Code 0.28.13 global hooks through `${HOME:-~}/.letta/settings.json`, matching Letta's `HOME || homedir()` behavior. Keep Elydora runtimes under the shared CLI root from `os.path.expanduser("~")`, independent from Letta's configurable settings home. Keep the active workspace `.letta/settings.json` and `.letta/settings.local.json` sources read-only, deduplicate the project source when the workspace is the home directory, and apply the official `hooks.disabled` precedence: an explicit global `false` enables hooks, global `true` disables them, then project and project-local `true` values disable them. Validate all eleven current events plus command and prompt handler shapes while preserving future event payloads. Register exact `PreToolUse`, `PostToolUse`, and `PostToolUseFailure` groups with `*` matchers and ten-second millisecond timeouts. Preserve the complete native snake_case payload, propagate frozen or revoked state through exit code `2`, and surface audit delivery failures through stderr, owner-only `error.log`, and exit code `1`; Letta post events retain their documented nonblocking semantics. Use PowerShell's call operator with explicit `$LASTEXITCODE` propagation on Windows and exact two-argument shell commands on Unix. Commit global settings, generated runtimes, runtime config, and private key through one rollback-capable transaction guarded by every consulted source snapshot. Migrate only exact legacy Python command ownership, require physical sources, canonical private keys, exact runtime sources, executable identity, matching runtime identity, and an effective enabled state during status checks, then run `/hooks` and restart active sessions. ## Code Quality diff --git a/README.md b/README.md index 939c0c0..f860f31 100644 --- a/README.md +++ b/README.md @@ -102,6 +102,8 @@ Qwen Code 0.20.0 installation writes exact, named `PreToolUse`, `PostToolUse`, a Kiro CLI installation covers both runtime contracts. Kiro CLI v2 uses the generated custom agent through `kiro-cli --agent elydora-audit`. Kiro CLI v3 loads the global standalone hooks when started with `kiro-cli --v3`. +Letta Code 0.28.13 installation writes exact `PreToolUse`, `PostToolUse`, and `PostToolUseFailure` groups to `${HOME:-~}/.letta/settings.json`, following the [official hook source and event contract](https://docs.letta.com/letta-agent/hooks). Elydora runtime files stay under the shared CLI root resolved by `os.path.expanduser("~")`, so Windows `HOME` overrides affect Letta settings while `USERPROFILE` retains Elydora runtime ownership. Workspace `.letta/settings.json` and `.letta/settings.local.json` remain read-only and participate in effective `hooks.disabled` evaluation. An explicit global `false` enables hooks; global `true`, project `true`, and project-local `true` disable them according to Letta's precedence. Managed command handlers use ten-second millisecond timeouts, the current absolute Python executable, PowerShell exit-code propagation on Windows, and complete native snake_case payloads. Frozen and revoked state propagates through exit code `2`; audit delivery failures remain visible through stderr, owner-only `error.log`, and exit code `1`, while Letta post events remain nonblocking. Global settings, generated runtimes, runtime config, and private key commit through one rollback-capable transaction guarded by snapshots of all three settings sources. Status requires an effective enabled state, exact hook and executable identity, physical files, canonical private keys, and matching runtime identity. Run `/hooks` and restart active Letta Code sessions after installation. + ### Commands | Command | Description | diff --git a/elydora/_runtime_paths.py b/elydora/_runtime_paths.py index 1dc25e7..485a15e 100644 --- a/elydora/_runtime_paths.py +++ b/elydora/_runtime_paths.py @@ -14,6 +14,11 @@ _WINDOWS_DEVICE_NAME = re.compile( ) +def runtime_root() -> str: + """Return the shared per-user root for Elydora agent runtimes.""" + return os.path.join(os.path.expanduser("~"), ".elydora") + + def resolve_agent_directory(root: str, agent_id: str) -> str: """Resolve one cross-platform-safe agent directory directly under *root*.""" if ( diff --git a/elydora/cli.py b/elydora/cli.py index 315d435..0d83d01 100644 --- a/elydora/cli.py +++ b/elydora/cli.py @@ -20,6 +20,7 @@ from ._runtime_paths import ( require_physical_directory, require_physical_file, resolve_agent_directory, + runtime_root, ) from .crypto import get_public_key_base64url from .plugins.base import AgentPlugin, InstallConfig @@ -67,10 +68,6 @@ LEGACY_SECRET_OPTIONS = { } -def _runtime_root() -> str: - return os.path.join(os.path.expanduser("~"), ".elydora") - - def _exit_with_error(message: str) -> NoReturn: print(f"Error: {message}", file=sys.stderr) raise SystemExit(1) @@ -89,7 +86,7 @@ def _reject_legacy_secret_arguments(arguments: list[str]) -> None: def _resolve_agent_directory_or_exit(agent_id: str) -> str: try: - return resolve_agent_directory(_runtime_root(), agent_id) + return resolve_agent_directory(runtime_root(), agent_id) except ValueError as error: _exit_with_error(str(error)) @@ -149,13 +146,13 @@ def _read_installed_agent_or_exit(agent_id: str) -> Optional[_InstalledAgent]: def _discover_installed_agents_or_exit() -> list[_InstalledAgent]: - runtime_root = _runtime_root() - if not _require_physical_directory_or_exit(runtime_root): + root = runtime_root() + if not _require_physical_directory_or_exit(root): return [] installed_agents: list[_InstalledAgent] = [] try: - with os.scandir(runtime_root) as entries: + with os.scandir(root) as entries: for entry in entries: if entry.is_symlink(): _exit_with_error( @@ -167,7 +164,7 @@ def _discover_installed_agents_or_exit() -> list[_InstalledAgent]: if installed_agent is not None: installed_agents.append(installed_agent) except OSError as error: - _exit_with_error(f"Scan agent runtime root at {runtime_root}: {error}") + _exit_with_error(f"Scan agent runtime root at {root}: {error}") return installed_agents @@ -219,7 +216,7 @@ def cmd_install(args: argparse.Namespace) -> None: plugin.preflight_install(config) # Create per-agent directory under ~/.elydora/{agent_id}/ - ensure_private_directory(_runtime_root()) + ensure_private_directory(runtime_root()) ensure_private_directory(agent_dir) # Generate and write the guard script for adapters that use the shared runtime. @@ -239,7 +236,7 @@ def cmd_install(args: argparse.Namespace) -> None: def cmd_uninstall(args: argparse.Namespace) -> None: """Handle the 'uninstall' subcommand.""" explicit_agent_id = getattr(args, "agent_id", None) - elydora_dir = _runtime_root() + elydora_dir = runtime_root() agent_id = explicit_agent_id if agent_id: agent_dir = _resolve_agent_directory_or_exit(agent_id) diff --git a/elydora/plugins/letta.py b/elydora/plugins/letta.py index 25885c2..58cc589 100644 --- a/elydora/plugins/letta.py +++ b/elydora/plugins/letta.py @@ -1,208 +1,78 @@ -"""Letta Code plugin — merges PreToolUse/PostToolUse hooks into ~/.letta/settings.json.""" +"""Letta Code 0.28 native global hook integration.""" from __future__ import annotations -import json -import os -import sys +from typing import Dict -from ._file_io import write_json_atomic, write_text_atomic from .base import AgentPlugin, InstallConfig, PluginStatus -from .hook_template import generate_hook_script +from .letta_config import render_letta_document +from .letta_contract import ( + AGENT_KEY, + JsonObject, + build_letta_group, + letta_runtime_contracts, +) +from .letta_installation import ( + LettaRuntimePaths, + commit_letta_installation, + commit_letta_uninstall, + preflight_letta_installation, + prepare_letta_installation, + prepare_letta_uninstall, +) +from .letta_io import letta_runtime_files_exist +from .letta_sources import read_letta_sources -SETTINGS_PATH = os.path.join(os.path.expanduser("~"), ".letta", "settings.json") -ELYDORA_DIR = os.path.join(os.path.expanduser("~"), ".elydora") +def _installed_groups(paths: LettaRuntimePaths) -> Dict[str, JsonObject]: + return { + "PreToolUse": build_letta_group(paths.guard_path), + "PostToolUse": build_letta_group(paths.audit_path), + "PostToolUseFailure": build_letta_group(paths.audit_path), + } class LettaPlugin(AgentPlugin): - """Install/uninstall Elydora audit hook for Letta Code.""" + """Install Elydora into Letta Code's native global settings.""" - @staticmethod - def _hook_path_for(agent_id: str) -> str: - return os.path.join(ELYDORA_DIR, agent_id, "hook.py") + manages_guard_runtime = True + + def preflight_install(self, config: InstallConfig) -> None: + sources = read_letta_sources() + preflight_letta_installation(config, sources) def install(self, config: InstallConfig) -> None: - agent_id = config.get("agent_id", "") - agent_name = config.get("agent_name", "") - - # Create per-agent directory - agent_dir = os.path.join(ELYDORA_DIR, agent_id) - os.makedirs(agent_dir, exist_ok=True) - - # Write config.json - config_data = { - "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_name, - } - config_path = os.path.join(agent_dir, "config.json") - write_json_atomic( - config_path, - config_data, - 0o600, - "Elydora runtime config", + sources = read_letta_sources() + paths = preflight_letta_installation(config, sources) + rendered = render_letta_document( + sources.global_settings, + None, + _installed_groups(paths), ) - - # Write private key - private_key_path = os.path.join(agent_dir, "private.key") - write_text_atomic( - private_key_path, - config.get("private_key", ""), - 0o600, - "Elydora private key", - ) - - script = generate_hook_script( - org_id=config.get("org_id", ""), - agent_id=agent_id, - kid=config.get("kid", ""), - base_url=config.get("base_url", "https://api.elydora.com"), - ) - hook_path = self._hook_path_for(agent_id) - write_text_atomic( - hook_path, - script, - 0o700, - "Elydora audit runtime", - ) - - guard_script_path = config.get("guard_script_path", "") - python_exe = sys.executable - - settings = _load_json(SETTINGS_PATH) - hooks = settings.setdefault("hooks", {}) - - # --- PreToolUse (guard — freeze enforcement, PascalCase with matcher, no timeout_ms) --- - pre_tool_use = hooks.setdefault("PreToolUse", []) - pre_tool_use[:] = [h for h in pre_tool_use if not _is_elydora_hook(h)] - if guard_script_path: - pre_tool_use.append({ - "matcher": "*", - "hooks": [ - { - "type": "command", - "command": f'"{python_exe}" {guard_script_path}', - } - ], - }) - - # --- PostToolUse (audit logging, PascalCase with matcher, no timeout_ms) --- - post_tool_use = hooks.setdefault("PostToolUse", []) - - post_tool_use[:] = [h for h in post_tool_use if not _is_elydora_hook(h)] - - post_tool_use.append({ - "matcher": "*", - "hooks": [ - { - "type": "command", - "command": hook_path, - } - ], - }) - - _save_json(SETTINGS_PATH, settings) - print("Elydora hook installed for Letta Code.") - print(f" Hook script: {hook_path}") - print(f" Settings: {SETTINGS_PATH}") + changes = prepare_letta_installation(config, paths, rendered) + commit_letta_installation(changes, sources) + print(f"Letta Code hooks installed at {sources.global_settings.file_path}") + print("Letta Code verification: run /hooks and restart active sessions.") def uninstall(self, agent_id: str = "") -> None: - if os.path.exists(SETTINGS_PATH): - settings = _load_json(SETTINGS_PATH) - hooks = settings.get("hooks", {}) - changed = False - - # Remove PreToolUse entries - pre_tool_use = hooks.get("PreToolUse", []) - pre_filtered = [h for h in pre_tool_use if not _is_elydora_hook(h, agent_id)] - if len(pre_filtered) != len(pre_tool_use): - hooks["PreToolUse"] = pre_filtered - if not pre_filtered: - del hooks["PreToolUse"] - changed = True - - # Remove PostToolUse entries - post_tool_use = hooks.get("PostToolUse", []) - post_filtered = [h for h in post_tool_use if not _is_elydora_hook(h, agent_id)] - if len(post_filtered) != len(post_tool_use): - hooks["PostToolUse"] = post_filtered - if not post_filtered: - del hooks["PostToolUse"] - changed = True - - if changed: - if not hooks: - del settings["hooks"] - _save_json(SETTINGS_PATH, settings) - - # Hook script removal is handled by cli.py cmd_uninstall (rmtree of agent dir) - print("Elydora hook uninstalled from Letta Code.") + sources = read_letta_sources() + rendered = render_letta_document( + sources.global_settings, agent_id or None, {} + ) + commit_letta_uninstall(prepare_letta_uninstall(rendered), sources) def status(self) -> PluginStatus: - # Scan ~/.elydora/*/hook.py for any installed hook - import glob as _glob - hook_pattern = os.path.join(ELYDORA_DIR, "*", "hook.py") - hook_files = _glob.glob(hook_pattern) - hook_exists = len(hook_files) > 0 - - settings_configured = False - if os.path.exists(SETTINGS_PATH): - settings = _load_json(SETTINGS_PATH) - hooks = settings.get("hooks", {}) - pre_tool_use = hooks.get("PreToolUse", []) - post_tool_use = hooks.get("PostToolUse", []) - pre_configured = any(_is_elydora_hook(h) for h in pre_tool_use) - post_configured = any(_is_elydora_hook(h) for h in post_tool_use) - settings_configured = pre_configured and post_configured - - installed = hook_exists and settings_configured + sources = read_letta_sources() + contracts = letta_runtime_contracts(sources.global_settings.hooks) + configured = not sources.disable_control.disabled and bool(contracts) + installed = configured and letta_runtime_files_exist(contracts) + file_path = sources.global_settings.file_path if installed: - details = f"Found {len(hook_files)} agent(s): {', '.join(hook_files)}" - elif hook_exists: - details = "Hook script exists but not configured in settings" - elif settings_configured: - details = "Configured in settings but hook script missing" + details = f"Config: {file_path}" + elif contracts and sources.disable_control.disabled: + details = f"Configured hooks are disabled: {file_path}" + elif contracts: + details = f"Configured at {file_path}; managed contract incomplete" else: details = "Not installed" - - return PluginStatus(installed=installed, agent="letta", details=details) - - -def _is_elydora_hook(entry: dict, agent_id: str = "") -> bool: - # Collect all command strings from the entry - commands: list[str] = [] - inner_hooks = entry.get("hooks") - if isinstance(inner_hooks, list): - for hook in inner_hooks: - if isinstance(hook, dict) and isinstance(hook.get("command"), str): - commands.append(hook["command"]) - else: - command = entry.get("command") - if isinstance(command, str): - commands.append(command) - - for cmd in commands: - cmd_lower = cmd.lower() - if "elydora" not in cmd_lower: - continue - # If agent_id is specified, only match hooks for that specific agent - if agent_id and agent_id in cmd: - return True - if not agent_id: - return True - return False - - -def _load_json(path: str) -> dict: - if not os.path.exists(path): - return {} - with open(path, "r", encoding="utf-8") as f: - return json.load(f) - - -def _save_json(path: str, data: dict) -> None: - write_json_atomic(path, data, 0o600, "Letta settings") + return PluginStatus(installed=installed, agent=AGENT_KEY, details=details) diff --git a/elydora/plugins/letta_command.py b/elydora/plugins/letta_command.py new file mode 100644 index 0000000..4f557cd --- /dev/null +++ b/elydora/plugins/letta_command.py @@ -0,0 +1,175 @@ +"""Letta Code command construction and exact runtime ownership.""" + +from __future__ import annotations + +from dataclasses import dataclass +import os +import re +import sys +from typing import Optional, Tuple + +from elydora._runtime_paths import runtime_root + + +@dataclass(frozen=True) +class LettaRuntimeReference: + agent_id: str + script_path: str + executable_path: Optional[str] = None + + +def same_letta_path(left: str, right: str) -> bool: + return os.path.normcase(os.path.abspath(left)) == os.path.normcase( + os.path.abspath(right) + ) + + +def same_letta_agent_id(left: str, right: str) -> bool: + return os.path.normcase(left) == os.path.normcase(right) + + +def _quote_posix(value: str) -> str: + return "'" + value.replace("'", "'\"'\"'") + "'" + + +def _quote_powershell(value: str) -> str: + return "'" + value.replace("'", "''") + "'" + + +def build_letta_command(script_path: str) -> str: + if not os.path.isabs(sys.executable) or not os.path.isabs(script_path): + raise ValueError( + "Letta Code hook commands require absolute executable and script paths" + ) + invocation = ( + f"{_quote_powershell(sys.executable)} {_quote_powershell(script_path)}" + if os.name == "nt" + else f"{_quote_posix(sys.executable)} {_quote_posix(script_path)}" + ) + return f"& {invocation}; exit $LASTEXITCODE" if os.name == "nt" else invocation + + +def _read_posix_argument(command: str, start: int) -> Optional[Tuple[str, int]]: + if start >= len(command) or command[start] != "'": + return None + apostrophe = "'\"'\"'" + value = "" + index = start + 1 + while index < len(command): + if command.startswith(apostrophe, index): + value += "'" + index += len(apostrophe) + continue + if command[index] == "'": + return value, index + 1 + value += command[index] + index += 1 + return None + + +def _parse_posix_command(command: str) -> Optional[Tuple[str, str]]: + executable = _read_posix_argument(command, 0) + if executable is None or executable[1] >= len(command): + return None + if command[executable[1]] != " ": + return None + script = _read_posix_argument(command, executable[1] + 1) + if script is None or script[1] != len(command): + return None + return executable[0], script[0] + + +def _read_powershell_argument( + command: str, start: int +) -> Optional[Tuple[str, int]]: + if start >= len(command) or command[start] != "'": + return None + value = "" + index = start + 1 + while index < len(command): + if command[index] != "'": + value += command[index] + index += 1 + continue + if index + 1 < len(command) and command[index + 1] == "'": + value += "'" + index += 2 + continue + return value, index + 1 + return None + + +def _parse_powershell_command(command: str) -> Optional[Tuple[str, str]]: + if not command.startswith("& "): + return None + executable = _read_powershell_argument(command, 2) + if executable is None or executable[1] >= len(command): + return None + if command[executable[1]] != " ": + return None + script = _read_powershell_argument(command, executable[1] + 1) + if script is None or command[script[1] :] != "; exit $LASTEXITCODE": + return None + return executable[0], script[0] + + +def _is_python_executable(file_path: str) -> bool: + return bool(re.fullmatch(r"python(?:3(?:\.\d+)?)?(?:\.exe)?", os.path.basename(file_path), re.IGNORECASE)) + + +def _runtime_reference( + script_path: str, + script_name: str, + executable_path: Optional[str] = None, +) -> Optional[LettaRuntimeReference]: + if not os.path.isabs(script_path) or os.path.basename(script_path) != script_name: + return None + agent_directory = os.path.dirname(script_path) + if not same_letta_path(os.path.dirname(agent_directory), runtime_root()): + return None + agent_id = os.path.basename(agent_directory) + if agent_id in {"", ".", ".."}: + return None + return LettaRuntimeReference(agent_id, script_path, executable_path) + + +def letta_runtime_reference( + command: str, script_name: str +) -> Optional[LettaRuntimeReference]: + parsed = ( + _parse_powershell_command(command) + if os.name == "nt" + else _parse_posix_command(command) + ) + if ( + parsed is None + or not os.path.isabs(parsed[0]) + or not _is_python_executable(parsed[0]) + ): + return None + return _runtime_reference(parsed[1], script_name, parsed[0]) + + +def letta_legacy_guard_reference( + command: str, script_name: str +) -> Optional[LettaRuntimeReference]: + if not command.startswith('"'): + return None + closing = command.find('"', 1) + if closing < 0 or closing + 1 >= len(command) or command[closing + 1] != " ": + return None + executable = command[1:closing] + script_path = command[closing + 2 :] + if ( + not os.path.isabs(executable) + or not _is_python_executable(executable) + or not script_path + ): + return None + return _runtime_reference(script_path, script_name, executable) + + +def letta_legacy_audit_reference( + command: str, script_name: str +) -> Optional[LettaRuntimeReference]: + return _runtime_reference(command, script_name) diff --git a/elydora/plugins/letta_config.py b/elydora/plugins/letta_config.py new file mode 100644 index 0000000..2c107a4 --- /dev/null +++ b/elydora/plugins/letta_config.py @@ -0,0 +1,195 @@ +"""Strict Letta Code settings parsing and source-preserving edits.""" + +from __future__ import annotations + +from dataclasses import dataclass +from typing import Dict, Optional + +from ._jsonc import JsoncEditor +from ._managed_files import FileSnapshot +from ._strict_json import parse_json_object +from .letta_contract import ( + MANAGED_EVENTS, + JsonObject, + LettaHooks, + managed_letta_removals, + read_letta_hooks, +) + + +@dataclass(frozen=True) +class LettaDocument: + kind: str + file_path: str + exists: bool + raw: str + snapshot: Optional[FileSnapshot] + root: JsonObject + hooks: LettaHooks + has_hooks_container: bool + + +@dataclass(frozen=True) +class RenderedLettaDocument: + document: LettaDocument + changed: bool + next_source: Optional[str] + + +def letta_source_label(kind: str) -> str: + labels = { + "global": "Letta Code global settings", + "project": "Letta Code project settings", + "project-local": "Letta Code project-local settings", + } + try: + return labels[kind] + except KeyError as error: + raise ValueError(f"Unsupported Letta Code document kind: {kind}") from error + + +def letta_document_label(document: LettaDocument) -> str: + return letta_source_label(document.kind) + + +def parse_letta_document( + *, + kind: str, + exists: bool, + file_path: str, + raw: str, + snapshot: Optional[FileSnapshot] = None, +) -> LettaDocument: + label = f"{letta_source_label(kind)} at {file_path}" + root = parse_json_object(raw, label) + return LettaDocument( + kind, + file_path, + exists, + raw, + snapshot, + root, + read_letta_hooks(root["hooks"]) if "hooks" in root else {}, + "hooks" in root, + ) + + +def create_letta_document(kind: str, file_path: str) -> LettaDocument: + return parse_letta_document( + kind=kind, + exists=False, + file_path=file_path, + raw="{}\n", + ) + + +def _current_document( + document: LettaDocument, editor: JsoncEditor +) -> LettaDocument: + return parse_letta_document( + kind=document.kind, + exists=document.exists, + file_path=document.file_path, + raw=editor.raw, + snapshot=document.snapshot, + ) + + +def _event_groups(hooks: LettaHooks, event: str) -> list[JsonObject]: + value = hooks.get(event) + return value if isinstance(value, list) else [] + + +def _remove_managed( + editor: JsoncEditor, + document: LettaDocument, + agent_id: Optional[str], +) -> None: + removals = managed_letta_removals(document.hooks, agent_id) + for event in MANAGED_EVENTS: + event_removals = sorted( + (item for item in removals if item.event == event), + key=lambda item: item.group_index, + reverse=True, + ) + for removal in event_removals: + group_path = ("hooks", event, removal.group_index) + if removal.remove_group: + editor.delete(group_path) + continue + for handler_index in sorted(removal.handler_indexes, reverse=True): + editor.delete((*group_path, "hooks", handler_index)) + if event_removals: + current = _current_document(document, editor) + if event in current.hooks and not _event_groups(current.hooks, event): + editor.delete(("hooks", event)) + current = _current_document(document, editor) + if current.has_hooks_container and not current.hooks: + editor.delete(("hooks",)) + + +def _append_group( + editor: JsoncEditor, + document: LettaDocument, + event: str, + group: JsonObject, +) -> None: + current = _current_document(document, editor) + groups = _event_groups(current.hooks, event) + if not current.has_hooks_container: + editor.add_property((), "hooks", {event: [group]}) + elif groups: + editor.append(("hooks", event), group) + elif event in current.hooks: + editor.append(("hooks", event), group) + else: + editor.add_property(("hooks",), event, [group]) + + +def _already_installed( + document: LettaDocument, additions: Dict[str, JsonObject] +) -> bool: + if set(additions) != set(MANAGED_EVENTS): + return False + removals = managed_letta_removals(document.hooks) + for event in MANAGED_EVENTS: + expected = additions[event] + event_removals = [item for item in removals if item.event == event] + exact_groups = [ + group for group in _event_groups(document.hooks, event) + if group == expected + ] + if ( + len(event_removals) != 1 + or not event_removals[0].remove_group + or len(exact_groups) != 1 + ): + return False + return True + + +def render_letta_document( + document: LettaDocument, + agent_id: Optional[str], + additions: Dict[str, JsonObject], +) -> RenderedLettaDocument: + if agent_id is None and _already_installed(document, additions): + return RenderedLettaDocument(document, False, document.raw) + editor = JsoncEditor( + document.raw, + f"{letta_document_label(document)} at {document.file_path}", + allow_trailing_commas=False, + ) + _remove_managed(editor, document, agent_id) + for event in MANAGED_EVENTS: + group = additions.get(event) + if group is not None: + _append_group(editor, document, event, group) + current = _current_document(document, editor) + if not additions and not current.root: + return RenderedLettaDocument(document, True, None) + return RenderedLettaDocument( + document, + editor.raw != document.raw, + editor.raw, + ) diff --git a/elydora/plugins/letta_contract.py b/elydora/plugins/letta_contract.py new file mode 100644 index 0000000..f9344db --- /dev/null +++ b/elydora/plugins/letta_contract.py @@ -0,0 +1,277 @@ +"""Letta Code 0.28 hook schema and exact ownership rules.""" + +from __future__ import annotations + +from dataclasses import dataclass +import math +import os +import sys +from typing import Any, Dict, List, Optional, Tuple + +from elydora._runtime_paths import runtime_root + +from .letta_command import ( + LettaRuntimeReference, + build_letta_command, + letta_legacy_audit_reference, + letta_legacy_guard_reference, + letta_runtime_reference, + same_letta_agent_id, + same_letta_path, +) + + +AGENT_KEY = "letta" +GUARD_SCRIPT = "guard.py" +AUDIT_SCRIPT = "hook.py" +HOOK_TIMEOUT_MILLISECONDS = 10_000 +MANAGED_EVENTS = ("PreToolUse", "PostToolUse", "PostToolUseFailure") + +TOOL_EVENTS = { + "PreToolUse", + "PostToolUse", + "PostToolUseFailure", + "PermissionRequest", +} +SIMPLE_EVENTS = { + "UserPromptSubmit", + "Notification", + "Stop", + "SubagentStop", + "PreCompact", + "SessionStart", + "SessionEnd", +} +KNOWN_EVENTS = TOOL_EVENTS | SIMPLE_EVENTS + +JsonObject = Dict[str, Any] +LettaHooks = Dict[str, Any] + + +@dataclass(frozen=True) +class LettaRuntimeContract: + agent_id: str + guard_path: str + audit_path: str + + +@dataclass(frozen=True) +class ManagedRemoval: + event: str + group_index: int + handler_indexes: Tuple[int, ...] + remove_group: bool + + +def elydora_dir() -> str: + return runtime_root() + + +def _validate_timeout(value: JsonObject, label: str) -> None: + if "timeout" not in value: + return + timeout = value["timeout"] + try: + finite = math.isfinite(timeout) + except (TypeError, OverflowError): + finite = False + if ( + isinstance(timeout, bool) + or not isinstance(timeout, (int, float)) + or not finite + or timeout < 0 + ): + raise ValueError(f"{label} timeout must be a non-negative finite number") + + +def _validate_handler(value: Any, label: str) -> JsonObject: + if not isinstance(value, dict): + raise ValueError(f"{label} must be an object") + handler = dict(value) + kind = handler.get("type") + if kind not in {"command", "prompt"}: + raise ValueError(f'{label} has unsupported type "{kind}"') + _validate_timeout(handler, label) + if "quiet" in handler and not isinstance(handler["quiet"], bool): + raise ValueError(f"{label} quiet must be a boolean") + if kind == "command": + command = handler.get("command") + if not isinstance(command, str) or not command: + raise ValueError(f"{label} requires a non-empty command") + else: + prompt = handler.get("prompt") + if not isinstance(prompt, str) or not prompt: + raise ValueError(f"{label} requires a non-empty prompt") + if "model" in handler and not isinstance(handler["model"], str): + raise ValueError(f"{label} model must be a string") + return handler + + +def _validate_group(value: Any, event: str, index: int) -> JsonObject: + label = f"Letta Code settings group hooks.{event}[{index}]" + if not isinstance(value, dict): + raise ValueError(f"{label} must be an object") + group = dict(value) + if event in TOOL_EVENTS: + if not isinstance(group.get("matcher"), str): + raise ValueError(f"{label} matcher must be a string") + elif "matcher" in group: + raise ValueError(f"{label} matcher is unsupported for {event}") + handlers = group.get("hooks") + if not isinstance(handlers, list): + raise ValueError(f"{label} must contain a hooks array") + group["hooks"] = [ + _validate_handler(handler, f"{label}.hooks[{handler_index}]") + for handler_index, handler in enumerate(handlers) + ] + return group + + +def read_letta_hooks(value: Any) -> LettaHooks: + if not isinstance(value, dict): + raise ValueError('Letta Code settings field "hooks" must be an object') + if "disabled" in value and not isinstance(value["disabled"], bool): + raise ValueError( + 'Letta Code settings field "hooks.disabled" must be a boolean' + ) + hooks: LettaHooks = {} + for event, groups in value.items(): + if event == "disabled" or event not in KNOWN_EVENTS: + hooks[event] = groups + continue + if not isinstance(groups, list): + raise ValueError( + f'Letta Code settings field "hooks.{event}" must be an array' + ) + hooks[event] = [ + _validate_group(group, event, index) + for index, group in enumerate(groups) + ] + return hooks + + +def build_letta_group(script_path: str) -> JsonObject: + return { + "matcher": "*", + "hooks": [{ + "type": "command", + "command": build_letta_command(script_path), + "timeout": HOOK_TIMEOUT_MILLISECONDS, + }], + } + + +def _exact_group(group: JsonObject) -> bool: + return set(group) == {"matcher", "hooks"} and group.get("matcher") == "*" + + +def _current_reference( + handler: JsonObject, script_name: str +) -> Optional[LettaRuntimeReference]: + if ( + set(handler) == {"type", "command", "timeout"} + and handler.get("type") == "command" + and handler.get("timeout") == HOOK_TIMEOUT_MILLISECONDS + and isinstance(handler.get("command"), str) + ): + return letta_runtime_reference(handler["command"], script_name) + return None + + +def _legacy_reference( + event: str, handler: JsonObject, script_name: str +) -> Optional[LettaRuntimeReference]: + if ( + set(handler) != {"type", "command"} + or handler.get("type") != "command" + or not isinstance(handler.get("command"), str) + ): + return None + if event == "PreToolUse": + return letta_legacy_guard_reference(handler["command"], script_name) + return letta_legacy_audit_reference(handler["command"], script_name) + + +def _event_groups(hooks: LettaHooks, event: str) -> List[JsonObject]: + value = hooks.get(event) + return value if isinstance(value, list) else [] + + +_EVENT_CONTRACTS = ( + ("PreToolUse", GUARD_SCRIPT), + ("PostToolUse", AUDIT_SCRIPT), + ("PostToolUseFailure", AUDIT_SCRIPT), +) + + +def managed_letta_removals( + hooks: LettaHooks, agent_id: Optional[str] = None +) -> List[ManagedRemoval]: + removals: List[ManagedRemoval] = [] + for event, script_name in _EVENT_CONTRACTS: + for group_index, group in enumerate(_event_groups(hooks, event)): + indexes = tuple( + index + for index, handler in enumerate(group["hooks"]) + if ( + reference := ( + _current_reference(handler, script_name) + or _legacy_reference(event, handler, script_name) + ) + ) + is not None + and ( + agent_id is None + or same_letta_agent_id(reference.agent_id, agent_id) + ) + ) + if indexes: + removals.append(ManagedRemoval( + event, + group_index, + indexes, + _exact_group(group) and len(indexes) == len(group["hooks"]), + )) + return removals + + +def _references_for_event( + groups: List[JsonObject], script_name: str +) -> Dict[str, List[LettaRuntimeReference]]: + references: Dict[str, List[LettaRuntimeReference]] = {} + for group in groups: + if not _exact_group(group): + continue + for handler in group["hooks"]: + reference = _current_reference(handler, script_name) + if ( + reference is None + or reference.executable_path is None + or not same_letta_path(reference.executable_path, sys.executable) + ): + continue + key = os.path.normcase(reference.agent_id) + references.setdefault(key, []).append(reference) + return references + + +def letta_runtime_contracts(hooks: LettaHooks) -> List[LettaRuntimeContract]: + guards = _references_for_event(_event_groups(hooks, "PreToolUse"), GUARD_SCRIPT) + posts = _references_for_event(_event_groups(hooks, "PostToolUse"), AUDIT_SCRIPT) + failures = _references_for_event( + _event_groups(hooks, "PostToolUseFailure"), AUDIT_SCRIPT + ) + contracts: List[LettaRuntimeContract] = [] + for key, guard in guards.items(): + post = posts.get(key, []) + failure = failures.get(key, []) + if len(guard) != 1 or len(post) != 1 or len(failure) != 1: + continue + if not same_letta_path(post[0].script_path, failure[0].script_path): + continue + contracts.append(LettaRuntimeContract( + guard[0].agent_id, + guard[0].script_path, + post[0].script_path, + )) + return contracts diff --git a/elydora/plugins/letta_installation.py b/elydora/plugins/letta_installation.py new file mode 100644 index 0000000..47fa5bc --- /dev/null +++ b/elydora/plugins/letta_installation.py @@ -0,0 +1,205 @@ +"""Transactional Letta Code hook and runtime installation.""" + +from __future__ import annotations + +from dataclasses import dataclass +import json +import os +from typing import List, Optional, Sequence + +from elydora._runtime_paths import resolve_agent_directory + +from ._managed_files import MAX_CONFIG_BYTES, MAX_SECRET_BYTES, MAX_SOURCE_BYTES +from ._transaction import ( + FileChange, + FilePrecondition, + file_change, + source_change, + write_changes, +) +from .base import InstallConfig +from .guard_template import generate_guard_script +from .hook_template import generate_hook_script +from .letta_command import same_letta_path +from .letta_config import RenderedLettaDocument, letta_document_label +from .letta_contract import AGENT_KEY, AUDIT_SCRIPT, GUARD_SCRIPT, elydora_dir +from .letta_io import validate_api_origin, validate_private_key, validate_runtime_tree +from .letta_sources import LettaSources, require_letta_hooks_enabled + + +@dataclass(frozen=True) +class LettaRuntimePaths: + agent_id: str + agent_directory: str + guard_path: str + audit_path: str + + +def _runtime_config(config: InstallConfig, agent_id: str) -> dict: + value = { + "org_id": config.get("org_id", ""), + "agent_id": agent_id, + "kid": config.get("kid", ""), + "base_url": config.get("base_url", "https://api.elydora.com"), + "agent_name": AGENT_KEY, + } + token = config.get("token") + if token: + value["token"] = token + return value + + +def _json_source(value: dict) -> str: + return json.dumps(value, indent=2) + "\n" + + +def _agent_paths(config: InstallConfig) -> LettaRuntimePaths: + agent_id = config.get("agent_id", "") + if not agent_id: + raise ValueError("agent_id is required") + agent_directory = resolve_agent_directory(elydora_dir(), agent_id) + guard_path = os.path.join(agent_directory, GUARD_SCRIPT) + audit_path = os.path.join(agent_directory, AUDIT_SCRIPT) + if not same_letta_path(config.get("guard_script_path", ""), guard_path): + raise ValueError( + "Elydora guard runtime must use the managed agent directory: " + f"{guard_path}" + ) + return LettaRuntimePaths(agent_id, agent_directory, guard_path, audit_path) + + +def _validate_install_config(config: InstallConfig) -> None: + for field in ("org_id", "agent_id", "kid", "private_key", "base_url"): + value = config.get(field) + if not isinstance(value, str) or not value: + raise ValueError(f"{field} is required") + if config.get("agent_name") != AGENT_KEY: + raise ValueError( + f"Letta Code installation requires agent_name {AGENT_KEY}" + ) + validate_private_key(config["private_key"]) + validate_api_origin(config["base_url"]) + if "token" in config: + token = config["token"] + if not isinstance(token, str) or not token: + raise ValueError("token must be a non-empty string when provided") + _agent_paths(config) + + +def preflight_letta_installation( + config: InstallConfig, sources: LettaSources +) -> LettaRuntimePaths: + require_letta_hooks_enabled(sources) + _validate_install_config(config) + paths = _agent_paths(config) + validate_runtime_tree(paths.agent_directory, paths.agent_id) + return paths + + +def _runtime_change( + file_path: str, + label: str, + next_source: str, + mode: int, + maximum_bytes: int = MAX_SOURCE_BYTES, +) -> Optional[FileChange]: + return file_change(file_path, label, next_source, mode, maximum_bytes) + + +def _rendered_change( + rendered: RenderedLettaDocument, +) -> Optional[FileChange]: + if not rendered.changed: + return None + return source_change( + rendered.document.file_path, + letta_document_label(rendered.document), + rendered.document.raw if rendered.document.exists else None, + rendered.next_source, + 0o600, + MAX_SOURCE_BYTES, + expected_snapshot=rendered.document.snapshot, + ) + + +def _present(changes: Sequence[Optional[FileChange]]) -> List[FileChange]: + return [change for change in changes if change is not None] + + +def prepare_letta_installation( + config: InstallConfig, + paths: LettaRuntimePaths, + rendered: RenderedLettaDocument, +) -> List[FileChange]: + guard_script = generate_guard_script(AGENT_KEY, paths.agent_id) + audit_script = generate_hook_script( + org_id=config.get("org_id", ""), + agent_id=paths.agent_id, + kid=config.get("kid", ""), + base_url=config.get("base_url", "https://api.elydora.com"), + fail_closed=True, + native_payload=True, + agent_name=AGENT_KEY, + ) + return _present([ + _runtime_change( + paths.guard_path, "Elydora guard runtime", guard_script, 0o700 + ), + _runtime_change( + os.path.join(paths.agent_directory, "config.json"), + "Elydora runtime config", + _json_source(_runtime_config(config, paths.agent_id)), + 0o600, + MAX_CONFIG_BYTES, + ), + _runtime_change( + os.path.join(paths.agent_directory, "private.key"), + "Elydora private key", + config.get("private_key", ""), + 0o600, + MAX_SECRET_BYTES, + ), + _runtime_change( + paths.audit_path, "Elydora audit runtime", audit_script, 0o700 + ), + _rendered_change(rendered), + ]) + + +def _read_only_preconditions( + sources: LettaSources, changed_path: Optional[str] +) -> List[FilePrecondition]: + return [ + condition + for condition in sources.preconditions + if changed_path is None + or not same_letta_path(condition.file_path, changed_path) + ] + + +def commit_letta_installation( + changes: List[FileChange], sources: LettaSources +) -> None: + write_changes( + changes, + "Install Letta Code hooks", + _read_only_preconditions(sources, sources.global_settings.file_path), + ) + + +def prepare_letta_uninstall( + rendered: RenderedLettaDocument, +) -> Optional[FileChange]: + return _rendered_change(rendered) + + +def commit_letta_uninstall( + change: Optional[FileChange], sources: LettaSources +) -> None: + if change is None: + return + write_changes( + [change], + "Uninstall Letta Code hooks", + _read_only_preconditions(sources, sources.global_settings.file_path), + ) diff --git a/elydora/plugins/letta_io.py b/elydora/plugins/letta_io.py new file mode 100644 index 0000000..5254fb8 --- /dev/null +++ b/elydora/plugins/letta_io.py @@ -0,0 +1,201 @@ +"""Strict Letta Code managed runtime inspection.""" + +from __future__ import annotations + +import base64 +import os +from typing import Any, List, Optional +import urllib.parse + +from ._managed_files import ( + MAX_CONFIG_BYTES, + MAX_SECRET_BYTES, + physical_directory_exists, + physical_file_exists, + read_physical_file, +) +from ._strict_json import JsonObject, parse_json_object +from .guard_template import generate_guard_script +from .hook_template import generate_hook_script +from .letta_command import same_letta_agent_id, same_letta_path +from .letta_contract import AGENT_KEY, LettaRuntimeContract, elydora_dir + + +def validate_api_origin(value: str, label: str = "base_url") -> None: + try: + parsed = urllib.parse.urlsplit(value) + hostname = parsed.hostname + parsed.port + except ValueError as error: + raise ValueError(f"{label} must be an absolute HTTP or HTTPS URL") from error + invalid_character = "\\" in value or any( + character.isspace() or ord(character) < 32 for character in value + ) + if ( + parsed.scheme not in ("http", "https") + or not parsed.netloc + or hostname is None + or invalid_character + ): + raise ValueError(f"{label} must be an absolute HTTP or HTTPS URL") + if ( + parsed.username is not None + or parsed.password is not None + or parsed.query + or parsed.fragment + ): + raise ValueError( + f"{label} must exclude credentials, query parameters, and fragments" + ) + + +def validate_private_key(value: str, label: str = "private_key") -> None: + try: + padded = value + "=" * ((4 - len(value) % 4) % 4) + seed = base64.b64decode( + padded.replace("-", "+").replace("_", "/"), validate=True + ) + canonical = base64.urlsafe_b64encode(seed).rstrip(b"=").decode("ascii") + except (ValueError, UnicodeEncodeError) as error: + raise ValueError( + f"{label} must be a canonical 32-byte base64url value" + ) from error + if len(seed) != 32 or canonical != value: + raise ValueError(f"{label} must be a canonical 32-byte base64url value") + + +def _require_non_empty_string( + value: Any, field: str, config_path: str +) -> str: + if not isinstance(value, str) or not value.strip(): + raise ValueError( + f"Elydora runtime config {field} is invalid: {config_path}" + ) + return value + + +def _validate_runtime_config( + config: JsonObject, expected_agent_id: str, config_path: str +) -> None: + supported = {"org_id", "agent_id", "kid", "base_url", "token", "agent_name"} + extra = next((key for key in config if key not in supported), None) + if extra is not None: + raise ValueError( + f'Elydora runtime config has unsupported field "{extra}": {config_path}' + ) + _require_non_empty_string(config.get("org_id"), "org_id", config_path) + _require_non_empty_string(config.get("kid"), "kid", config_path) + agent_id = _require_non_empty_string( + config.get("agent_id"), "agent_id", config_path + ) + if ( + not same_letta_agent_id(agent_id, expected_agent_id) + or config.get("agent_name") != AGENT_KEY + ): + raise ValueError( + "Elydora runtime identity does not match Letta Code hooks: " + f"{config_path}" + ) + if "token" in config: + _require_non_empty_string(config.get("token"), "token", config_path) + base_url = _require_non_empty_string( + config.get("base_url"), "base_url", config_path + ) + validate_api_origin(base_url, "Elydora runtime config base_url") + + +def _read_runtime_config(file_path: str) -> Optional[JsonObject]: + snapshot = read_physical_file( + file_path, "Elydora runtime config", MAX_CONFIG_BYTES + ) + if snapshot is None: + return None + return parse_json_object( + snapshot.contents, f"Elydora runtime config at {file_path}" + ) + + +def validate_runtime_tree(agent_directory: str, agent_id: str) -> None: + root = elydora_dir() + if not physical_directory_exists(root, "Elydora runtime directory"): + return + if not physical_directory_exists( + agent_directory, "Elydora agent runtime directory" + ): + return + config_path = os.path.join(agent_directory, "config.json") + config = _read_runtime_config(config_path) + artifacts = ( + ("private.key", "Elydora private key"), + ("guard.py", "Elydora guard runtime"), + ("hook.py", "Elydora audit runtime"), + ("chain-state.json", "Elydora chain state"), + ("status-cache.json", "Elydora status cache"), + ("error.log", "Elydora error log"), + ) + artifact_exists = any( + physical_file_exists(os.path.join(agent_directory, name), label) + for name, label in artifacts + ) + if config is None: + if artifact_exists: + raise ValueError( + "Elydora runtime identity cannot be verified without config.json: " + f"{agent_directory}" + ) + return + configured_id = config.get("agent_id") + if ( + not isinstance(configured_id, str) + or not same_letta_agent_id(configured_id, agent_id) + or config.get("agent_name") != AGENT_KEY + ): + raise ValueError( + "Elydora runtime config identity does not match Letta Code agent " + f"{agent_id}: {config_path}" + ) + + +def _runtime_contract_exists(contract: LettaRuntimeContract) -> bool: + root = elydora_dir() + agent_directory = os.path.dirname(contract.guard_path) + if ( + not same_letta_path(os.path.dirname(agent_directory), root) + or not same_letta_path( + contract.audit_path, os.path.join(agent_directory, "hook.py") + ) + ): + return False + if not physical_directory_exists(root, "Elydora runtime directory"): + return False + if not physical_directory_exists( + agent_directory, "Elydora agent runtime directory" + ): + return False + config_path = os.path.join(agent_directory, "config.json") + key_path = os.path.join(agent_directory, "private.key") + config = _read_runtime_config(config_path) + key = read_physical_file(key_path, "Elydora private key", MAX_SECRET_BYTES) + guard = read_physical_file(contract.guard_path, "Elydora guard runtime") + audit = read_physical_file(contract.audit_path, "Elydora audit runtime") + if config is None or key is None or guard is None or audit is None: + return False + _validate_runtime_config(config, contract.agent_id, config_path) + validate_private_key(key.contents, "Elydora private key") + expected_guard = generate_guard_script(AGENT_KEY, contract.agent_id) + expected_audit = generate_hook_script( + org_id=str(config["org_id"]), + agent_id=contract.agent_id, + kid=str(config["kid"]), + base_url=str(config["base_url"]), + fail_closed=True, + native_payload=True, + agent_name=AGENT_KEY, + ) + return guard.contents == expected_guard and audit.contents == expected_audit + + +def letta_runtime_files_exist( + contracts: List[LettaRuntimeContract], +) -> bool: + return any(_runtime_contract_exists(contract) for contract in contracts) diff --git a/elydora/plugins/letta_sources.py b/elydora/plugins/letta_sources.py new file mode 100644 index 0000000..b046c53 --- /dev/null +++ b/elydora/plugins/letta_sources.py @@ -0,0 +1,159 @@ +"""Letta Code 0.28 settings discovery and effective hook controls.""" + +from __future__ import annotations + +from dataclasses import dataclass +import os +from typing import Dict, Optional, Tuple + +from ._managed_files import ( + MAX_SOURCE_BYTES, + physical_directory_exists, + read_physical_file, +) +from ._transaction import FilePrecondition +from .letta_command import same_letta_path +from .letta_config import ( + LettaDocument, + create_letta_document, + letta_document_label, + letta_source_label, + parse_letta_document, +) + + +@dataclass(frozen=True) +class LettaDisableControl: + disabled: bool + source: Optional[LettaDocument] + + +@dataclass(frozen=True) +class LettaSources: + home_directory: str + global_settings: LettaDocument + project: LettaDocument + project_local: LettaDocument + project_active: bool + disable_control: LettaDisableControl + preconditions: Tuple[FilePrecondition, ...] + + +def _home_directory() -> str: + home = os.environ.get("HOME") or os.path.expanduser("~") + return os.path.abspath(home) + + +def _comparison_path(file_path: str) -> str: + return os.path.normcase(os.path.abspath(file_path)) + + +def _canonical_path(file_path: str) -> str: + return os.path.realpath(file_path) + + +def _read_document(kind: str, file_path: str) -> LettaDocument: + snapshot = read_physical_file( + file_path, letta_source_label(kind), MAX_SOURCE_BYTES + ) + if snapshot is None: + return create_letta_document(kind, file_path) + return parse_letta_document( + kind=kind, + exists=True, + file_path=file_path, + raw=snapshot.contents, + snapshot=snapshot, + ) + + +def _source_precondition(document: LettaDocument) -> FilePrecondition: + return FilePrecondition( + document.file_path, + letta_document_label(document), + document.snapshot, + MAX_SOURCE_BYTES, + ) + + +def _deduplicate_preconditions( + values: list[FilePrecondition], +) -> Tuple[FilePrecondition, ...]: + result: Dict[str, FilePrecondition] = {} + for value in values: + result.setdefault(_comparison_path(value.file_path), value) + return tuple(result.values()) + + +def _effective_disable( + global_settings: LettaDocument, + project: LettaDocument, + project_local: LettaDocument, + project_active: bool, +) -> LettaDisableControl: + if global_settings.hooks.get("disabled") is False: + return LettaDisableControl(False, global_settings) + if global_settings.hooks.get("disabled") is True: + return LettaDisableControl(True, global_settings) + if project_active and project.hooks.get("disabled") is True: + return LettaDisableControl(True, project) + if project_local.hooks.get("disabled") is True: + return LettaDisableControl(True, project_local) + return LettaDisableControl(False, None) + + +def read_letta_sources() -> LettaSources: + home_directory = _home_directory() + workspace = os.path.abspath(os.getcwd()) + global_directory = os.path.join(home_directory, ".letta") + project_directory = os.path.join(workspace, ".letta") + physical_directory_exists(home_directory, "Letta Code home directory") + physical_directory_exists( + global_directory, "Letta Code global configuration directory" + ) + if not same_letta_path(project_directory, global_directory): + physical_directory_exists( + project_directory, "Letta Code project configuration directory" + ) + global_path = os.path.join(global_directory, "settings.json") + project_path = os.path.join(project_directory, "settings.json") + local_path = os.path.join(project_directory, "settings.local.json") + project_active = _comparison_path(_canonical_path(workspace)) != _comparison_path( + _canonical_path(home_directory) + ) + global_settings = _read_document("global", global_path) + project = ( + _read_document("project", project_path) + if project_active + else create_letta_document("project", project_path) + ) + project_local = _read_document("project-local", local_path) + preconditions = [_source_precondition(global_settings)] + if project_active: + preconditions.append(_source_precondition(project)) + preconditions.append(_source_precondition(project_local)) + return LettaSources( + home_directory, + global_settings, + project, + project_local, + project_active, + _effective_disable( + global_settings, project, project_local, project_active + ), + _deduplicate_preconditions(preconditions), + ) + + +def require_letta_hooks_enabled(sources: LettaSources) -> None: + if not sources.disable_control.disabled: + return + source = sources.disable_control.source + location = ( + f"{letta_document_label(source)} at {source.file_path}" + if source is not None + else "effective settings" + ) + raise ValueError( + f"Letta Code hooks are disabled by hooks.disabled in {location}" + ) diff --git a/elydora/plugins/registry.py b/elydora/plugins/registry.py index d902d4d..c1bc6da 100644 --- a/elydora/plugins/registry.py +++ b/elydora/plugins/registry.py @@ -68,7 +68,7 @@ SUPPORTED_AGENTS: Dict[str, Dict[str, str]] = { }, "letta": { "name": "Letta Code", - "hook_event": "PostToolUse", + "hook_event": "PreToolUse/PostToolUse/PostToolUseFailure", "config_path": "~/.letta/settings.json", }, "opencode": { diff --git a/tests/letta_support.py b/tests/letta_support.py new file mode 100644 index 0000000..379ca47 --- /dev/null +++ b/tests/letta_support.py @@ -0,0 +1,289 @@ +from __future__ import annotations + +from dataclasses import dataclass +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +import json +import ntpath +import os +from pathlib import Path +import subprocess # nosec B404 +import sys +from threading import Thread +from typing import Any, Mapping, Optional + +import pytest + +from elydora.plugins.letta import LettaPlugin +from elydora.utils import base64url_encode + + +AGENT_ID = "agent-1" +MISSING = object() +VALID_PRIVATE_KEY = base64url_encode(bytes(range(32))) + + +@dataclass +class LettaFixture: + root_dir: Path + home_dir: Path + project_dir: Path + global_path: Path + project_path: Path + local_path: Path + agent_dir: Path + guard_path: Path + audit_path: Path + runtime_config_path: Path + private_key_path: Path + plugin: LettaPlugin + config: dict[str, str] + + def install(self) -> None: + self.plugin.install(self.config) + + def source(self, file_path: Optional[Path] = None) -> str: + path = file_path or self.global_path + with path.open("r", encoding="utf-8", newline="") as file: + return file.read() + + def settings(self, file_path: Optional[Path] = None) -> dict[str, Any]: + return json.loads(self.source(file_path)) + + +class ElydoraApiHandler(BaseHTTPRequestHandler): + agent_status: Any = "active" + operation_status = 201 + requests: list[dict[str, Any]] = [] + + @classmethod + def reset(cls, status: Any, operation_status: int) -> None: + cls.agent_status = status + cls.operation_status = operation_status + cls.requests = [] + + def _respond(self, status: int, value: object) -> None: + body = json.dumps(value).encode() + self.send_response(status) + self.send_header("Content-Type", "application/json") + self.send_header("Content-Length", str(len(body))) + self.end_headers() + self.wfile.write(body) + + def do_GET(self) -> None: + type(self).requests.append({ + "method": "GET", + "path": self.path, + "authorization": self.headers.get("Authorization"), + }) + self._respond(200, {"agent": {"status": type(self).agent_status}}) + + def do_POST(self) -> None: + length = int(self.headers.get("Content-Length", "0")) + raw = self.rfile.read(length) + type(self).requests.append({ + "method": "POST", + "path": self.path, + "authorization": self.headers.get("Authorization"), + "raw": raw, + "json": json.loads(raw), + }) + self._respond( + type(self).operation_status, + {"operation": {"accepted": True}}, + ) + + def log_message(self, _format: str, *_args: object) -> None: + return + + +def start_api_server( + *, status: Any = "active", operation_status: int = 201 +) -> ThreadingHTTPServer: + ElydoraApiHandler.reset(status, operation_status) + server = ThreadingHTTPServer(("127.0.0.1", 0), ElydoraApiHandler) + Thread(target=server.serve_forever, daemon=True).start() + return server + + +def server_base_url(server: ThreadingHTTPServer) -> str: + return f"http://127.0.0.1:{server.server_port}" + + +def write_text(path: Path, source: str, mode: int = 0o600) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(source, encoding="utf-8", newline="") + os.chmod(path, mode) + + +def write_json(path: Path, value: object, mode: int = 0o600) -> None: + source = value if isinstance(value, str) else json.dumps(value, indent=2) + "\n" + write_text(path, source, mode) + + +def prepare_fixture( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, + *, + global_settings: object = MISSING, + project_settings: object = MISSING, + local_settings: object = MISSING, + base_url: str = "http://127.0.0.1:9", +) -> LettaFixture: + home_dir = tmp_path / "home with spaces and 'quote $HOME %HOME%" + project_dir = tmp_path / "project with spaces" + project_dir.mkdir(parents=True) + monkeypatch.setenv("HOME", str(home_dir)) + monkeypatch.setenv("USERPROFILE", str(home_dir)) + monkeypatch.chdir(project_dir) + global_path = home_dir / ".letta" / "settings.json" + project_path = project_dir / ".letta" / "settings.json" + local_path = project_dir / ".letta" / "settings.local.json" + agent_dir = home_dir / ".elydora" / AGENT_ID + guard_path = agent_dir / "guard.py" + audit_path = agent_dir / "hook.py" + runtime_config_path = agent_dir / "config.json" + private_key_path = agent_dir / "private.key" + for path, value in ( + (global_path, global_settings), + (project_path, project_settings), + (local_path, local_settings), + ): + if value is not MISSING: + write_json(path, value) + return LettaFixture( + root_dir=tmp_path, + home_dir=home_dir, + project_dir=project_dir, + global_path=global_path, + project_path=project_path, + local_path=local_path, + agent_dir=agent_dir, + guard_path=guard_path, + audit_path=audit_path, + runtime_config_path=runtime_config_path, + private_key_path=private_key_path, + plugin=LettaPlugin(), + config={ + "org_id": "org-1", + "agent_id": AGENT_ID, + "agent_name": "letta", + "private_key": VALID_PRIVATE_KEY, + "kid": "kid-1", + "token": "token-1", + "base_url": base_url, + "guard_script_path": str(guard_path), + }, + ) + + +def managed_handler( + settings: Mapping[str, Any], event: str +) -> dict[str, Any]: + for group in settings.get("hooks", {}).get(event, []): + if set(group) != {"matcher", "hooks"} or group.get("matcher") != "*": + continue + for handler in group["hooks"]: + if ( + set(handler) == {"type", "command", "timeout"} + and handler.get("type") == "command" + and handler.get("timeout") == 10_000 + ): + return handler + raise AssertionError(f"managed handler for {event!r} not found") + + +def assert_managed_triple(settings: Mapping[str, Any]) -> dict[str, Any]: + guard = managed_handler(settings, "PreToolUse") + audit = managed_handler(settings, "PostToolUse") + failure = managed_handler(settings, "PostToolUseFailure") + assert audit["command"] == failure["command"] + return {"guard": guard, "audit": audit, "failure": failure} + + +def legacy_group(script_path: Path, event: str) -> dict[str, Any]: + command = ( + f'"{sys.executable}" {script_path}' + if event == "PreToolUse" + else str(script_path) + ) + return { + "matcher": "*", + "hooks": [{"type": "command", "command": command}], + } + + +def run_handler( + handler: Mapping[str, Any], payload: bytes, fixture: LettaFixture +) -> subprocess.CompletedProcess[bytes]: + environment = { + **os.environ, + "HOME": str(fixture.home_dir), + "USERPROFILE": str(fixture.home_dir), + } + if os.name == "nt": + powershell = ntpath.join( + os.environ.get("SystemRoot", r"C:\Windows"), + "System32", + "WindowsPowerShell", + "v1.0", + "powershell.exe", + ) + command = [ + powershell, + "-NoLogo", + "-NoProfile", + "-NonInteractive", + "-Command", + str(handler["command"]), + ] + else: + command = ["/bin/bash", "-c", str(handler["command"])] + return subprocess.run( # nosec B603 + command, + input=payload, + capture_output=True, + check=False, + cwd=fixture.project_dir, + env=environment, + timeout=15, + ) + + +def official_input(fixture: LettaFixture, event: str) -> dict[str, Any]: + payload: dict[str, Any] = { + "event_type": event, + "working_directory": str(fixture.project_dir), + "session_id": "session-1", + "tool_name": "Bash", + "tool_input": {"command": "echo test"}, + "tool_call_id": "call-1", + "agent_id": "letta-agent-1", + } + if event == "PostToolUse": + payload["tool_result"] = {"status": "success", "output": "test"} + if event == "PostToolUseFailure": + payload.update({ + "error_message": "Command failed", + "error_type": "ProcessError", + }) + return payload + + +def run_cli( + fixture: LettaFixture, arguments: list[str] +) -> subprocess.CompletedProcess[str]: + environment = dict(os.environ) + repository = str(Path(__file__).resolve().parents[1]) + existing = environment.get("PYTHONPATH") + environment["PYTHONPATH"] = ( + repository + os.pathsep + existing if existing else repository + ) + return subprocess.run( # nosec B603 + [sys.executable, "-m", "elydora.cli", *arguments], + text=True, + capture_output=True, + check=False, + cwd=fixture.project_dir, + env=environment, + timeout=30, + ) diff --git a/tests/test_cli_security.py b/tests/test_cli_security.py index e1d0940..3049bc5 100644 --- a/tests/test_cli_security.py +++ b/tests/test_cli_security.py @@ -18,7 +18,6 @@ from elydora.plugins import ( copilot, cursor, kiroide, - letta, opencode, ) from elydora.plugins.hook_template import generate_hook_script @@ -375,7 +374,6 @@ def test_atomic_secret_write_preserves_existing_file_on_replace_failure( (copilot, copilot.CopilotPlugin, "copilot"), (cursor, cursor.CursorPlugin, "cursor"), (kiroide, kiroide.KiroIdePlugin, "kiroide"), - (letta, letta.LettaPlugin, "letta"), (opencode, opencode.OpenCodePlugin, "opencode"), ], ) diff --git a/tests/test_letta_official.py b/tests/test_letta_official.py new file mode 100644 index 0000000..d4919f1 --- /dev/null +++ b/tests/test_letta_official.py @@ -0,0 +1,180 @@ +from __future__ import annotations + +import json +import os +from pathlib import Path +import shutil +import subprocess # nosec B404 + +import pytest + +from letta_support import ( + ElydoraApiHandler, + managed_handler, + prepare_fixture, + server_base_url, + start_api_server, +) + + +SOURCE_ROOT = os.environ.get("ELYDORA_LETTA_SOURCE") +BUN_BINARY = os.environ.get("ELYDORA_BUN_BINARY") or shutil.which("bun") + + +def _module_url(relative_path: str) -> str: + if SOURCE_ROOT is None: + raise RuntimeError("ELYDORA_LETTA_SOURCE is required") + return (Path(SOURCE_ROOT) / relative_path).resolve().as_uri() + + +def _result_json(stdout: str) -> dict: + return json.loads(stdout.strip().splitlines()[-1]) + + +def _run_official_probe(fixture: object, mode: str) -> subprocess.CompletedProcess[str]: + if SOURCE_ROOT is None or BUN_BINARY is None: + raise RuntimeError("official Letta source and Bun are required") + global_path = getattr(fixture, "global_path") + settings = json.loads(Path(global_path).read_text(encoding="utf-8")) + commands = { + "guard": managed_handler(settings, "PreToolUse")["command"], + "audit": managed_handler(settings, "PostToolUse")["command"], + "failure": managed_handler(settings, "PostToolUseFailure")["command"], + } + source = f""" +import {{ settingsManager }} from {json.dumps(_module_url('src/settings-manager.ts'))}; +import {{ getHooksForEvent }} from {json.dumps(_module_url('src/hooks/loader.ts'))}; +import {{ executeCommandHook }} from {json.dumps(_module_url('src/hooks/executor.ts'))}; +import {{ runPostToolUseHooks }} from {json.dumps(_module_url('src/hooks/index.ts'))}; +const workspace = process.env.ELYDORA_OFFICIAL_WORKSPACE; +const commands = JSON.parse(process.env.ELYDORA_OFFICIAL_COMMANDS); +await settingsManager.initialize(); +await settingsManager.loadProjectSettings(workspace); +await settingsManager.loadLocalProjectSettings(workspace); +const pre = await getHooksForEvent('PreToolUse', 'Bash', workspace); +const post = await getHooksForEvent('PostToolUse', 'Bash', workspace); +const failure = await getHooksForEvent('PostToolUseFailure', 'Bash', workspace); +const managed = (hooks, command) => hooks.find((hook) => ( + hook.type === 'command' && hook.command === command +)); +if (process.env.ELYDORA_OFFICIAL_MODE === 'active') {{ + const guardResult = await executeCommandHook( + managed(pre, commands.guard), + {{ + event_type: 'PreToolUse', + working_directory: workspace, + tool_name: 'Bash', + tool_input: {{ command: 'echo official' }}, + tool_call_id: 'official-pre', + agent_id: 'letta-agent', + }}, + workspace, + ); + const postResult = await executeCommandHook( + managed(post, commands.audit), + {{ + event_type: 'PostToolUse', + working_directory: workspace, + tool_name: 'Bash', + tool_input: {{ command: 'echo official' }}, + tool_result: {{ status: 'success', output: 'official' }}, + tool_call_id: 'official-post', + }}, + workspace, + ); + const failureResult = await executeCommandHook( + managed(failure, commands.failure), + {{ + event_type: 'PostToolUseFailure', + working_directory: workspace, + tool_name: 'Bash', + tool_input: {{ command: 'exit 1' }}, + error_message: 'official failure', + error_type: 'ProcessError', + tool_call_id: 'official-failure', + }}, + workspace, + ); + console.log(JSON.stringify({{ + counts: [pre.length, post.length, failure.length], + exits: [guardResult.exitCode, postResult.exitCode, failureResult.exitCode], + }})); +}} else {{ + const result = await runPostToolUseHooks( + 'Bash', + {{ command: 'echo unavailable' }}, + {{ status: 'success', output: 'unavailable' }}, + 'official-post-error', + workspace, + ); + console.log(JSON.stringify({{ + blocked: result.blocked, + errored: result.errored, + exits: result.results.map((entry) => entry.exitCode), + }})); +}} +await settingsManager.flush(); +""" + environment = { + **os.environ, + "HOME": str(getattr(fixture, "home_dir")), + "USERPROFILE": str(getattr(fixture, "home_dir")), + "LETTA_SKIP_KEYCHAIN_CHECK": "1", + "ELYDORA_OFFICIAL_COMMANDS": json.dumps(commands), + "ELYDORA_OFFICIAL_MODE": mode, + "ELYDORA_OFFICIAL_WORKSPACE": str(getattr(fixture, "project_dir")), + } + return subprocess.run( # nosec B603 + [BUN_BINARY, "--silent", "--eval", source], + text=True, + capture_output=True, + check=False, + cwd=SOURCE_ROOT, + env=environment, + timeout=60, + ) + + +@pytest.mark.skipif( + SOURCE_ROOT is None or BUN_BINARY is None, + reason="official Letta source and Bun are required", +) +def test_official_letta_code_loads_and_executes_all_managed_hooks( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + assert SOURCE_ROOT is not None + manifest = json.loads( + (Path(SOURCE_ROOT) / "package.json").read_text(encoding="utf-8") + ) + assert manifest["version"] == "0.28.13" + server = start_api_server() + fixture = prepare_fixture( + monkeypatch, tmp_path, base_url=server_base_url(server) + ) + server_closed = False + try: + fixture.install() + active = _run_official_probe(fixture, "active") + assert active.returncode == 0, active.stderr + assert _result_json(active.stdout) == { + "counts": [1, 1, 1], + "exits": [0, 0, 0], + } + assert len([ + request for request in ElydoraApiHandler.requests + if request["method"] == "POST" + ]) == 2 + server.shutdown() + server.server_close() + server_closed = True + unavailable = _run_official_probe(fixture, "unavailable") + assert unavailable.returncode == 0, unavailable.stderr + assert _result_json(unavailable.stdout) == { + "blocked": False, + "errored": True, + "exits": [1], + } + finally: + if not server_closed: + server.shutdown() + server.server_close() diff --git a/tests/test_letta_plugin.py b/tests/test_letta_plugin.py new file mode 100644 index 0000000..46b6152 --- /dev/null +++ b/tests/test_letta_plugin.py @@ -0,0 +1,395 @@ +from __future__ import annotations + +import json +import os +from pathlib import Path +from typing import Any + +import pytest + +from elydora import cli +from elydora.plugins import letta +from elydora.plugins.registry import SUPPORTED_AGENTS +from letta_support import ( + AGENT_ID, + VALID_PRIVATE_KEY, + assert_managed_triple, + legacy_group, + prepare_fixture, + run_cli, + write_json, + write_text, +) + + +def test_letta_is_registered_in_the_sdk_and_cli() -> None: + assert SUPPORTED_AGENTS["letta"] == { + "name": "Letta Code", + "hook_event": "PreToolUse/PostToolUse/PostToolUseFailure", + "config_path": "~/.letta/settings.json", + } + assert cli.PLUGIN_MAP["letta"] is letta.LettaPlugin + + +def test_install_preserves_every_source_and_is_idempotent( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, + capsys: pytest.CaptureFixture[str], +) -> None: + global_source = ( + "{\r\n" + ' "theme": "dark",\r\n' + ' "hooks": {\r\n' + ' "FutureEvent": [null],\r\n' + ' "PreToolUse": [{ "matcher": "Bash", "hooks": ' + '[{ "type": "command", "command": "user-hook", "quiet": true }] }]\r\n' + " }\r\n" + "}\r\n" + ) + fixture = prepare_fixture( + monkeypatch, + tmp_path, + global_settings=global_source, + project_settings={"owner": "project"}, + local_settings={"owner": "local"}, + ) + project_before = fixture.source(fixture.project_path) + local_before = fixture.source(fixture.local_path) + + fixture.install() + first_source = fixture.source() + settings = fixture.settings() + assert "\r\n" in first_source + assert settings["theme"] == "dark" + assert settings["hooks"]["FutureEvent"] == [None] + assert settings["hooks"]["PreToolUse"][0]["hooks"][0]["command"] == "user-hook" + assert_managed_triple(settings) + assert fixture.source(fixture.project_path) == project_before + assert fixture.source(fixture.local_path) == local_before + for file_path in ( + fixture.guard_path, + fixture.audit_path, + fixture.runtime_config_path, + fixture.private_key_path, + ): + assert file_path.is_file() + + fixture.install() + assert fixture.source() == first_source + output = capsys.readouterr().out + assert "run /hooks" in output + assert "restart active sessions" in output + + +def test_letta_validates_known_schemas_and_preserves_future_events( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + fixture = prepare_fixture(monkeypatch, tmp_path, global_settings={ + "hooks": { + "FutureEvent": [None], + "PreToolUse": [{ + "matcher": "[", + "hooks": [{ + "type": "command", + "command": "user-command", + "timeout": 0, + "quiet": False, + "futureField": True, + }], + }], + "Stop": [{ + "hooks": [{ + "type": "prompt", + "prompt": "Evaluate $ARGUMENTS", + "model": "fast", + "timeout": 30, + }], + }], + }, + }) + fixture.install() + settings = fixture.settings() + assert settings["hooks"]["FutureEvent"] == [None] + assert settings["hooks"]["PreToolUse"][0]["matcher"] == "[" + assert settings["hooks"]["PreToolUse"][0]["hooks"][0]["futureField"] is True + assert settings["hooks"]["Stop"][0]["hooks"][0]["type"] == "prompt" + + +@pytest.mark.parametrize( + "source", + [ + "{ malformed", + "[]", + '{ "theme": true, }', + '{ "theme": true // comment\n }', + '{ "hooks": {}, "hooks": {} }', + '{ "hooks": null }', + '{ "hooks": [] }', + '{ "hooks": { "disabled": "yes" } }', + '{ "hooks": { "PreToolUse": null } }', + '{ "hooks": { "PreToolUse": [null] } }', + '{ "hooks": { "PreToolUse": [{ "hooks": [] }] } }', + '{ "hooks": { "Stop": [{ "matcher": "*", "hooks": [] }] } }', + '{ "hooks": { "PreToolUse": [{ "matcher": "*" }] } }', + '{ "hooks": { "Stop": [{ "hooks": [null] }] } }', + '{ "hooks": { "Stop": [{ "hooks": [{ "type": "http" }] }] } }', + '{ "hooks": { "PreToolUse": [{ "matcher": "*", "hooks": [{ "type": "command" }] }] } }', + '{ "hooks": { "Stop": [{ "hooks": [{ "type": "prompt" }] }] } }', + '{ "hooks": { "Stop": [{ "hooks": [{ "type": "prompt", "prompt": "x", "model": 1 }] }] } }', + '{ "hooks": { "Stop": [{ "hooks": [{ "type": "command", "command": "x", "quiet": 1 }] }] } }', + '{ "hooks": { "Stop": [{ "hooks": [{ "type": "command", "command": "x", "timeout": -1 }] }] } }', + ], +) +def test_install_rejects_malformed_global_settings_before_every_write( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path, source: str +) -> None: + fixture = prepare_fixture( + monkeypatch, tmp_path, global_settings=source + ) + with pytest.raises((OSError, ValueError), match="Letta Code"): + fixture.install() + assert fixture.source() == source + assert not fixture.agent_dir.exists() + + +@pytest.mark.parametrize("kind", ["project", "local"]) +def test_install_surfaces_malformed_read_only_sources( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path, kind: str +) -> None: + options = ( + {"project_settings": "{ malformed"} + if kind == "project" + else {"local_settings": "{ malformed"} + ) + fixture = prepare_fixture(monkeypatch, tmp_path, **options) + label = "project-local" if kind == "local" else kind + with pytest.raises(ValueError, match=f"Letta Code {label}"): + fixture.install() + assert not fixture.global_path.exists() + assert not fixture.agent_dir.exists() + + +@pytest.mark.parametrize( + ("global_settings", "project_settings", "local_settings", "enabled"), + [ + ({"hooks": {"disabled": True}}, {}, {}, False), + ({}, {"hooks": {"disabled": True}}, {}, False), + ({}, {}, {"hooks": {"disabled": True}}, False), + ( + {"hooks": {"disabled": False}}, + {"hooks": {"disabled": True}}, + {"hooks": {"disabled": True}}, + True, + ), + ], +) +def test_hooks_disabled_uses_official_precedence( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, + global_settings: object, + project_settings: object, + local_settings: object, + enabled: bool, +) -> None: + fixture = prepare_fixture( + monkeypatch, + tmp_path, + global_settings=global_settings, + project_settings=project_settings, + local_settings=local_settings, + ) + if enabled: + fixture.install() + assert_managed_triple(fixture.settings()) + else: + with pytest.raises(ValueError, match="hooks.disabled"): + fixture.install() + assert not fixture.agent_dir.exists() + + +def test_install_migrates_exact_legacy_and_preserves_lookalikes( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + fixture = prepare_fixture(monkeypatch, tmp_path) + guard_legacy = legacy_group(fixture.guard_path, "PreToolUse") + guard_lookalike = legacy_group(fixture.guard_path, "PreToolUse") + guard_lookalike["hooks"][0]["quiet"] = True + audit_legacy = legacy_group(fixture.audit_path, "PostToolUse") + write_json(fixture.global_path, { + "owner": "user", + "hooks": { + "PreToolUse": [guard_legacy, guard_lookalike], + "PostToolUse": [audit_legacy], + }, + }) + fixture.install() + settings = fixture.settings() + assert_managed_triple(settings) + assert any( + group["hooks"][0].get("quiet") is True + for group in settings["hooks"]["PreToolUse"] + ) + managed = next( + group for group in settings["hooks"]["PreToolUse"] + if group["hooks"][0].get("timeout") == 10_000 + ) + managed["userField"] = "preserve-group" + managed["hooks"].append({"type": "command", "command": "user-command"}) + write_json(fixture.global_path, settings) + + fixture.plugin.uninstall(AGENT_ID) + remaining = fixture.settings() + assert remaining["owner"] == "user" + assert any( + group["hooks"][0].get("quiet") is True + for group in remaining["hooks"]["PreToolUse"] + ) + assert any( + handler.get("command") == "user-command" + for group in remaining["hooks"]["PreToolUse"] + for handler in group["hooks"] + ) + assert "PostToolUse" not in remaining["hooks"] + assert "PostToolUseFailure" not in remaining["hooks"] + + +def test_uninstall_removes_owned_file_and_preserves_user_settings( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + user = prepare_fixture( + monkeypatch, + tmp_path / "user", + global_settings={"theme": "dark", "hooks": {"Notification": []}}, + ) + user.install() + user.plugin.uninstall(AGENT_ID) + assert user.settings() == { + "theme": "dark", + "hooks": {"Notification": []}, + } + + owned = prepare_fixture(monkeypatch, tmp_path / "owned") + owned.install() + owned.plugin.uninstall(AGENT_ID) + assert not owned.global_path.exists() + + +def test_status_requires_effective_contract_and_strict_runtime_identity( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + fixture = prepare_fixture(monkeypatch, tmp_path, project_settings={}) + fixture.install() + assert fixture.plugin.status()["installed"] is True + write_json(fixture.project_path, {"hooks": {"disabled": True}}) + assert fixture.plugin.status()["installed"] is False + write_json(fixture.project_path, {}) + settings = fixture.settings() + settings["hooks"]["PostToolUseFailure"].append( + settings["hooks"]["PostToolUseFailure"][-1] + ) + write_json(fixture.global_path, settings) + assert fixture.plugin.status()["installed"] is False + + fixture.install() + write_text(fixture.private_key_path, "invalid") + with pytest.raises(ValueError, match="private key"): + fixture.plugin.status() + fixture.install() + write_text(fixture.audit_path, "tampered") + assert fixture.plugin.status()["installed"] is False + fixture.install() + runtime = json.loads(fixture.runtime_config_path.read_text()) + runtime["agent_name"] = "other" + write_json(fixture.runtime_config_path, runtime) + with pytest.raises(ValueError, match="runtime identity"): + fixture.plugin.status() + + +def test_preflight_rejects_unverifiable_runtime_and_invalid_agent_id( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + identity = prepare_fixture(monkeypatch, tmp_path / "identity") + write_text(identity.audit_path, "unverified") + with pytest.raises(ValueError, match="without config.json"): + identity.install() + assert not identity.global_path.exists() + + invalid = prepare_fixture(monkeypatch, tmp_path / "invalid") + invalid.config["agent_id"] = "../escape" + with pytest.raises(ValueError, match="Invalid agent ID"): + invalid.install() + assert not invalid.global_path.exists() + + +def test_letta_cli_install_status_and_uninstall_end_to_end( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + fixture = prepare_fixture( + monkeypatch, tmp_path, global_settings={"theme": "dark"} + ) + key_file = tmp_path / "install-private.key" + token_file = tmp_path / "install-token.txt" + write_text(key_file, VALID_PRIVATE_KEY + "\n") + write_text(token_file, "token-1\n") + install = run_cli(fixture, [ + "install", + "--agent", "letta", + "--org_id", "org-1", + "--agent_id", AGENT_ID, + "--kid", "kid-1", + "--private_key_file", str(key_file), + "--token_file", str(token_file), + "--base_url", "http://127.0.0.1:9", + ]) + assert install.returncode == 0, install.stderr + assert_managed_triple(fixture.settings()) + status = run_cli(fixture, ["status"]) + assert status.returncode == 0, status.stderr + assert "Letta Code" in status.stdout and "[installed]" in status.stdout + uninstall = run_cli(fixture, [ + "uninstall", "--agent", "letta", "--agent_id", AGENT_ID + ]) + assert uninstall.returncode == 0, uninstall.stderr + assert fixture.settings() == {"theme": "dark"} + assert not fixture.agent_dir.exists() + + +@pytest.mark.skipif( + os.name != "nt", reason="Windows resolves HOME and USERPROFILE separately" +) +def test_windows_cli_separates_letta_home_from_runtime_home( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + fixture = prepare_fixture( + monkeypatch, tmp_path, global_settings={"theme": "dark"} + ) + runtime_home = tmp_path / "windows profile" + runtime_agent_dir = runtime_home / ".elydora" / AGENT_ID + monkeypatch.setenv("USERPROFILE", str(runtime_home)) + key_file = tmp_path / "install-private.key" + write_text(key_file, VALID_PRIVATE_KEY + "\n") + + install = run_cli(fixture, [ + "install", + "--agent", "letta", + "--org_id", "org-1", + "--agent_id", AGENT_ID, + "--kid", "kid-1", + "--private_key_file", str(key_file), + "--base_url", "http://127.0.0.1:9", + ]) + assert install.returncode == 0, install.stderr + assert_managed_triple(fixture.settings()) + assert (runtime_agent_dir / "guard.py").is_file() + assert (runtime_agent_dir / "hook.py").is_file() + + status = run_cli(fixture, ["status"]) + assert status.returncode == 0, status.stderr + assert "Letta Code" in status.stdout and "[installed]" in status.stdout + + uninstall = run_cli(fixture, [ + "uninstall", "--agent", "letta", "--agent_id", AGENT_ID + ]) + assert uninstall.returncode == 0, uninstall.stderr + assert fixture.settings() == {"theme": "dark"} + assert not runtime_agent_dir.exists() diff --git a/tests/test_letta_runtime.py b/tests/test_letta_runtime.py new file mode 100644 index 0000000..f87a858 --- /dev/null +++ b/tests/test_letta_runtime.py @@ -0,0 +1,147 @@ +from __future__ import annotations + +import json +import os +from pathlib import Path + +import pytest + +from letta_support import ( + ElydoraApiHandler, + assert_managed_triple, + official_input, + prepare_fixture, + run_handler, + server_base_url, + start_api_server, +) + + +def _payload(value: object) -> bytes: + return json.dumps(value).encode() + + +def test_guard_accepts_active_agents_with_native_exit_semantics( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + server = start_api_server(status="active") + fixture = prepare_fixture( + monkeypatch, tmp_path, base_url=server_base_url(server) + ) + try: + fixture.install() + guard = assert_managed_triple(fixture.settings())["guard"] + result = run_handler( + guard, + _payload(official_input(fixture, "PreToolUse")), + fixture, + ) + assert result.returncode == 0, result.stderr + assert result.stdout == b"" + assert len([ + request for request in ElydoraApiHandler.requests + if request["method"] == "GET" + ]) == 1 + finally: + server.shutdown() + server.server_close() + + +@pytest.mark.parametrize("status", ["frozen", "revoked"]) +def test_guard_propagates_blocking_states_through_exit_code_two( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path, status: str +) -> None: + server = start_api_server(status=status) + fixture = prepare_fixture( + monkeypatch, tmp_path, base_url=server_base_url(server) + ) + try: + fixture.install() + guard = assert_managed_triple(fixture.settings())["guard"] + result = run_handler( + guard, + _payload(official_input(fixture, "PreToolUse")), + fixture, + ) + assert result.returncode == 2, result.stderr + assert result.stdout == b"" + assert status.encode() in result.stderr.lower() + finally: + server.shutdown() + server.server_close() + + +def test_audit_forwards_complete_success_and_failure_payloads( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + server = start_api_server() + fixture = prepare_fixture( + monkeypatch, tmp_path, base_url=server_base_url(server) + ) + try: + fixture.install() + handlers = assert_managed_triple(fixture.settings()) + for handler_name, event in ( + ("audit", "PostToolUse"), + ("failure", "PostToolUseFailure"), + ): + payload = { + **official_input(fixture, event), + "future_provider_field": {"preserved": True}, + } + result = run_handler( + handlers[handler_name], _payload(payload), fixture + ) + assert result.returncode == 0, result.stderr + assert result.stdout == b"" + posts = [ + request for request in ElydoraApiHandler.requests + if request["method"] == "POST" + ] + assert len(posts) == 2 + assert posts[0]["json"]["payload"] == { + **official_input(fixture, "PostToolUse"), + "future_provider_field": {"preserved": True}, + } + assert posts[1]["json"]["payload"] == { + **official_input(fixture, "PostToolUseFailure"), + "future_provider_field": {"preserved": True}, + } + finally: + server.shutdown() + server.server_close() + + +def test_audit_failure_is_observable_and_fails_closed( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + fixture = prepare_fixture(monkeypatch, tmp_path) + fixture.install() + audit = assert_managed_triple(fixture.settings())["audit"] + result = run_handler( + audit, + _payload(official_input(fixture, "PostToolUse")), + fixture, + ) + assert result.returncode == 1 + assert b"Elydora audit" in result.stderr + error_log = fixture.agent_dir / "error.log" + source = error_log.read_text(encoding="utf-8") + assert "[elydora-hook]" in source + assert "URLError" in source + + +@pytest.mark.skipif(os.name == "nt", reason="POSIX mode bits apply on Unix") +def test_runtime_artifacts_use_private_modes( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + fixture = prepare_fixture(monkeypatch, tmp_path) + fixture.install() + for file_path, expected in ( + (fixture.runtime_config_path, 0o600), + (fixture.private_key_path, 0o600), + (fixture.guard_path, 0o700), + (fixture.audit_path, 0o700), + (fixture.global_path, 0o600), + ): + assert file_path.stat().st_mode & 0o777 == expected diff --git a/tests/test_letta_transaction.py b/tests/test_letta_transaction.py new file mode 100644 index 0000000..064c75f --- /dev/null +++ b/tests/test_letta_transaction.py @@ -0,0 +1,111 @@ +from __future__ import annotations + +import json +import os +from pathlib import Path +from typing import Any + +import pytest + +from elydora.plugins import _transaction +from elydora.plugins.letta_config import render_letta_document +from elydora.plugins.letta_contract import build_letta_group +from elydora.plugins.letta_installation import ( + commit_letta_installation, + preflight_letta_installation, + prepare_letta_installation, +) +from elydora.plugins.letta_sources import read_letta_sources +from letta_support import prepare_fixture, write_json + + +def _groups(guard_path: str, audit_path: str) -> dict: + return { + "PreToolUse": build_letta_group(guard_path), + "PostToolUse": build_letta_group(audit_path), + "PostToolUseFailure": build_letta_group(audit_path), + } + + +def test_transaction_aborts_when_read_only_source_changes( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + fixture = prepare_fixture( + monkeypatch, tmp_path, project_settings={"owner": "before"} + ) + sources = read_letta_sources() + paths = preflight_letta_installation(fixture.config, sources) + rendered = render_letta_document( + sources.global_settings, + None, + _groups(paths.guard_path, paths.audit_path), + ) + changes = prepare_letta_installation(fixture.config, paths, rendered) + write_json(fixture.project_path, {"owner": "after"}) + with pytest.raises(OSError, match="project settings changed"): + commit_letta_installation(changes, sources) + assert json.loads(fixture.project_path.read_text()) == {"owner": "after"} + assert not fixture.global_path.exists() + assert not fixture.agent_dir.exists() + + +def test_transaction_rolls_back_every_runtime_and_settings_change( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + fixture = prepare_fixture( + monkeypatch, tmp_path, global_settings={"owner": "user"} + ) + original = fixture.source() + real_replace = _transaction.os.replace + failed = False + + def fail_settings_commit(source: Any, destination: Any) -> None: + nonlocal failed + if not failed and Path(destination) == fixture.global_path: + failed = True + raise OSError("simulated Letta settings commit failure") + real_replace(source, destination) + + monkeypatch.setattr(_transaction.os, "replace", fail_settings_commit) + with pytest.raises(OSError, match="Install Letta Code hooks"): + fixture.install() + assert failed + assert fixture.source() == original + for file_path in ( + fixture.guard_path, + fixture.audit_path, + fixture.runtime_config_path, + fixture.private_key_path, + ): + assert not file_path.exists() + assert [ + path + for path in fixture.home_dir.rglob("*") + if path.suffix in {".tmp", ".rollback"} + ] == [] + + +def test_install_rejects_linked_configuration_directory( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + fixture = prepare_fixture(monkeypatch, tmp_path) + redirected = tmp_path / "redirected-letta" + redirected.mkdir() + fixture.home_dir.mkdir(parents=True) + try: + os.symlink(redirected, fixture.global_path.parent, target_is_directory=True) + except (NotImplementedError, OSError): + pytest.skip("directory links require platform privileges") + with pytest.raises(OSError, match="global configuration directory"): + fixture.install() + assert not fixture.agent_dir.exists() + + +def test_install_rejects_directory_at_settings_path( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + fixture = prepare_fixture(monkeypatch, tmp_path) + fixture.global_path.mkdir(parents=True) + with pytest.raises(OSError, match="global settings path"): + fixture.install() + assert not fixture.agent_dir.exists() diff --git a/tests/test_settings_hook_detection.py b/tests/test_settings_hook_detection.py index 9633a7f..b71f1b2 100644 --- a/tests/test_settings_hook_detection.py +++ b/tests/test_settings_hook_detection.py @@ -1,15 +1,39 @@ from __future__ import annotations -from elydora.plugins.letta import _is_elydora_hook as is_letta_hook +from pathlib import Path +import sys + +import pytest + +from elydora.plugins.letta_command import build_letta_command +from elydora.plugins.letta_contract import managed_letta_removals -def test_hook_detection_handles_only_string_commands() -> None: - assert is_letta_hook({"command": 42}, "") is False - assert is_letta_hook({"hooks": [{"command": None}]}, "") is False - assert is_letta_hook( - {"hooks": [{"command": "run .elydora/agent-1/hook.py"}]}, "" - ) is True - assert is_letta_hook( - {"hooks": [{"command": "run .elydora/agent-1/hook.py"}]}, - "agent-2", - ) is False +def test_letta_hook_detection_requires_exact_owned_commands( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + home = tmp_path / "home" + guard = home / ".elydora" / "agent-1" / "guard.py" + monkeypatch.setenv("HOME", str(home)) + monkeypatch.setenv("USERPROFILE", str(home)) + hooks = { + "PreToolUse": [{ + "matcher": "*", + "hooks": [{ + "type": "command", + "command": build_letta_command(str(guard)), + "timeout": 10_000, + }], + }], + } + assert len(managed_letta_removals(hooks)) == 1 + + hooks["PreToolUse"][0]["hooks"][0]["quiet"] = True + assert managed_letta_removals(hooks) == [] + + hooks["PreToolUse"][0]["hooks"][0] = { + "type": "command", + "command": f'"{sys.executable}" {guard}', + } + assert len(managed_letta_removals(hooks, "agent-1")) == 1 + assert managed_letta_removals(hooks, "agent-2") == []