diff --git a/apps/backend/agents/runtime/adapters/generic_edit.py b/apps/backend/agents/runtime/adapters/generic_edit.py index c5a62c804..3fd66686b 100644 --- a/apps/backend/agents/runtime/adapters/generic_edit.py +++ b/apps/backend/agents/runtime/adapters/generic_edit.py @@ -44,6 +44,11 @@ resolve_runtime_mcp_support, unavailable_mcp_action_result, ) +from ..plugin_tool_bridge import ( + decision_to_result, + to_canonical_result, + to_canonical_tool, +) from ..result import AgentRunResult from ..subagents import ( DEFAULT_SUBAGENT_MERGE_POLICY, @@ -65,6 +70,34 @@ logger = logging.getLogger(__name__) + +def _load_tool_hook_plugins(project_dir: Path) -> list[Any]: + """Load enabled plugins that can participate in tool hooks. + + Mirrors the Claude SDK path (``build_plugin_tool_hook_matchers``): keep only + enabled plugins whose capabilities include generic_edit/full_agent_runtime. + Any failure degrades to an empty list and a warning so a plugin problem + never breaks the Direct-API runtime loop. + """ + try: + from plugins.runtime import ( + can_use_tool_hooks, + load_enabled_runtime_plugins, + ) + except Exception as exc: # pragma: no cover - import guard + logger.warning("Plugin tool hooks unavailable: %s", exc) + return [] + try: + return [ + plugin + for plugin in load_enabled_runtime_plugins(project_dir) + if can_use_tool_hooks(plugin) + ] + except Exception as exc: + logger.warning("Failed to load plugin tool hooks: %s", exc) + return [] + + GENERIC_EDIT_CANCELLED_MESSAGE = "Generic edit runtime was cancelled." GENERIC_EDIT_PROMPT_TEMPLATE = """You are running in Auto Code generic_edit mode. @@ -676,6 +709,121 @@ def __init__( self._resume_metadata: dict[str, Any] | None = None self._active_batch_id: str | None = None self._batch_recovery_blockers: dict[str, list[str]] = {} + # Plugin tool hooks: the enabled hook-capable plugins are loaded once + # (project-scoped, lazy), and the per-run AgentContext is built in run() + # once spec_dir is known. Both stay []/None when no generic_edit or + # full_agent_runtime plugin is enabled, so the hook path is a cheap + # no-op on the common runtime. + self._tool_hook_plugins: list[Any] | None = None + self._plugin_hook_context: Any | None = None + + def _ensure_tool_hook_plugins(self) -> list[Any]: + if self._tool_hook_plugins is None: + self._tool_hook_plugins = _load_tool_hook_plugins( + self._executor.project_dir + ) + return self._tool_hook_plugins + + def _prepare_plugin_hook_context( + self, spec_dir: Path, subtask_id: str | None + ) -> None: + """Build the per-run plugin hook AgentContext once spec_dir is known. + + Leaves the context ``None`` (hook path becomes a no-op) when no + hook-capable plugin is enabled or context construction fails. + """ + self._plugin_hook_context = None + if not self._ensure_tool_hook_plugins(): + return + try: + from plugins.runtime import build_agent_context + + self._plugin_hook_context = build_agent_context( + self._executor.project_dir, + spec_dir, + self.agent_type or "", + metadata={"subtask_id": subtask_id} if subtask_id else None, + ) + except Exception as exc: + logger.warning("Failed to build plugin hook context: %s", exc) + self._plugin_hook_context = None + + async def _run_plugin_pre_tool_hook( + self, action: dict[str, Any], tool: str + ) -> ToolActionResult | None: + """Run enabled plugin ``pre_tool`` hooks for one native local action. + + Returns a blocked ``ToolActionResult`` when a plugin blocks the tool, + else ``None``. The native action is normalized to the canonical SDK + tool vocabulary first (see ``plugin_tool_bridge``) so hooks written + against the Claude SDK match; read-only actions are gated off. + """ + plugins = self._tool_hook_plugins + context = self._plugin_hook_context + if not plugins or context is None: + return None + canonical = to_canonical_tool(action, include_read_only=False) + if canonical is None: + return None + try: + from plugins.runtime import run_plugin_pre_tool_hooks + + decision = await run_plugin_pre_tool_hooks( + plugins, + context=context, + input_data={ + "tool_name": canonical.tool_name, + "tool_input": canonical.tool_input, + }, + ) + except Exception as exc: + logger.warning("Plugin pre_tool hook failed: %s", exc) + return None + if decision.get("decision") == "block": + return decision_to_result(decision, tool) + return None + + async def _run_plugin_post_tool_hook( + self, action: dict[str, Any], tool: str, result: ToolActionResult + ) -> None: + """Run enabled plugin ``post_tool`` hooks after a native action ran. + + Observational only: the action has already executed, so a post_tool + ``block`` cannot un-run it. A block is logged and annotated onto the + result (``result.data["plugin_post_tool_block"]``) rather than rolling + the mutation back — staged-snapshot rollback is a deliberate follow-up, + not part of this slice. + """ + plugins = self._tool_hook_plugins + context = self._plugin_hook_context + if not plugins or context is None: + return + canonical = to_canonical_tool(action, include_read_only=False) + if canonical is None: + return + try: + from plugins.runtime import run_plugin_post_tool_hooks + + decision = await run_plugin_post_tool_hooks( + plugins, + context=context, + input_data={ + "tool_name": canonical.tool_name, + "tool_input": canonical.tool_input, + "tool_result": to_canonical_result(result), + }, + ) + except Exception as exc: + logger.warning("Plugin post_tool hook failed: %s", exc) + return + if decision.get("decision") == "block": + reason = decision.get("reason") or "blocked by plugin runtime hook" + logger.warning( + "Plugin post_tool blocked %s after execution (observational): %s", + tool, + reason, + ) + result.data["plugin_post_tool_block"] = str(reason) @property def context_client(self) -> Any: @@ -710,6 +858,7 @@ async def run( project_dir=self._executor.project_dir, agent_type=self.agent_type, ) + self._prepare_plugin_hook_context(spec_dir, subtask_id) try: if self._supports_native_tool_calls(): @@ -750,6 +899,9 @@ async def resume( project_dir=self._executor.project_dir, agent_type=self.agent_type, ) + # Resumed runs drive _execute_action just like run(); build the plugin + # hook context here too, otherwise tool hooks stay a no-op on resume. + self._prepare_plugin_hook_context(spec_dir, subtask_id) checkpoint_path = resolve_generic_edit_resume_checkpoint_path( checkpoint_path=checkpoint_path, spec_dir=spec_dir, @@ -1937,6 +2089,15 @@ async def _execute_action( self._auto_open_changeset_batch(action) tool = action_tool(action) + + # Plugin pre_tool hooks — Direct-API parity with the Claude SDK + # PreToolUse path. A hook may block a tool before it runs; this sits + # after the scope/batch/isolation guards above so those stronger + # invariants still report first. + plugin_block = await self._run_plugin_pre_tool_hook(action, tool) + if plugin_block is not None: + return plugin_block + try: staged_workspace = self._materialize_staged_batch_workspace(action) except GenericEditRuntimeError as e: @@ -1988,6 +2149,7 @@ async def _execute_action( ) if restore_error is not None: return restore_error + await self._run_plugin_post_tool_hook(action, tool, result) return result def _materialize_staged_batch_workspace( diff --git a/apps/backend/agents/runtime/plugin_tool_bridge.py b/apps/backend/agents/runtime/plugin_tool_bridge.py new file mode 100644 index 000000000..ba3a1d546 --- /dev/null +++ b/apps/backend/agents/runtime/plugin_tool_bridge.py @@ -0,0 +1,296 @@ +"""Backend-neutral normalization between Direct-API local actions and the +plugin tool-hook vocabulary. + +Plugin ``pre_tool``/``post_tool`` hooks were written against the Claude Agent +SDK tool vocabulary (``Bash``, ``Edit``, ``Write``, ``Read`` ...) and the SDK +tool-input shapes (``file_path``, ``old_string`` ...). The Direct-API +in-process runtime speaks its own local-action vocabulary (``run_command``, +``replace_text``, ``write_file`` ...) with different field names. Without a +translation layer, wiring the hooks with native names silently no-ops: e.g. +skill-pack-runtime only blocks when ``tool_name.casefold() in {"bash", +"shell"}``, so a native ``run_command`` would sail straight past the guard. + +This module is the translation layer, and nothing more. It is pure and +side-effect-free: it does NOT load plugins, build agent context, or execute +actions. It only maps a native local-action dict to the canonical +``(tool_name, tool_input)`` a hook expects, exposes the post-hook result view, +and back-translates a hook block decision into a ``ToolActionResult``. Wiring +these into the execution loop is a separate slice. + +Design decisions (see the plugin-runtime cross-backend discussion): + +* Canonical vocabulary = SDK tool names where an SDK equivalent exists, and an + honest dedicated name (``Delete``/``Move``/``ApplyPatch``) where it does not. + We deliberately do NOT fabricate a synthetic ``Bash {command: "rm ..."}`` for + destructive local actions: a guard that believes it can rewrite a shell + command which will never run as a shell is worse than honest invisibility. +* Normalization is one-directional (native -> inspection view). Hooks may + ``block`` a tool on Direct-API, but input-rewrite is not supported because we + never reconstruct a native action from a canonical view. +* Read-only actions are mapped here but their hook invocation is gated behind + ``include_read_only`` (default off) so plugins do not run on every file read + unless a caller opts in (e.g. a secret-scanning / exfil guard). +""" + +from __future__ import annotations + +from collections.abc import Callable +from typing import Any, NamedTuple + +from .local_actions import ToolActionResult, action_tool + + +class CanonicalTool(NamedTuple): + """A native local action expressed in the plugin hook vocabulary.""" + + tool_name: str + tool_input: dict[str, Any] + + +# Native local actions that neither mutate the workspace nor run commands. +# They are mapped to canonical tools below, but hook invocation on them is +# opt-in via ``include_read_only`` so the common path stays hook-free. +READ_ONLY_TOOLS: frozenset[str] = frozenset( + { + "stat_path", + "list_files", + "search_text", + "read_file", + "read_file_range", + "read_many_files", + "git_status", + "git_diff", + } +) + +# Native local actions that are transaction/orchestration control-plane, not +# tool-use. They carry no meaningful tool semantics for a hook and are never +# translated (``to_canonical_tool`` returns ``None``). +NO_HOOK_TOOLS: frozenset[str] = frozenset( + { + "begin_batch", + "commit_batch", + "abort_batch", + "rollback_transaction", + "repair_mutation", + "resolve_subagent_conflict", + "run_subagents", + "finish", + } +) + + +def _include(target: dict[str, Any], key: str, value: Any) -> None: + """Set ``key`` on ``target`` only when ``value`` is present.""" + if value is not None: + target[key] = value + + +def _parse_patch_paths(patch: str) -> list[str]: + """Extract touched file paths from a unified diff, best-effort. + + Reads ``+++``/``---`` headers, strips ``a/``/``b/`` prefixes, and skips + ``/dev/null``. Returned sorted and de-duplicated so a file-scoped plugin can + match an ``apply_patch`` without parsing the diff itself. + """ + paths: set[str] = set() + for line in patch.splitlines(): + if not (line.startswith("+++ ") or line.startswith("--- ")): + continue + candidate = line[4:].strip() + # Drop a trailing tab-separated timestamp some diff tools append. + candidate = candidate.split("\t", 1)[0].strip() + if not candidate or candidate == "/dev/null": + continue + if candidate[:2] in ("a/", "b/"): + candidate = candidate[2:] + if candidate: + paths.add(candidate) + return sorted(paths) + + +def _bash(action: dict[str, Any]) -> CanonicalTool: + tool_input: dict[str, Any] = {} + _include(tool_input, "command", action.get("command")) + _include(tool_input, "timeout", action.get("timeout")) + return CanonicalTool("Bash", tool_input) + + +def _write(action: dict[str, Any]) -> CanonicalTool: + tool_input: dict[str, Any] = {} + _include(tool_input, "file_path", action.get("path")) + _include(tool_input, "content", action.get("content")) + return CanonicalTool("Write", tool_input) + + +def _edit(action: dict[str, Any]) -> CanonicalTool: + tool_input: dict[str, Any] = {} + _include(tool_input, "file_path", action.get("path")) + _include(tool_input, "old_string", action.get("old")) + _include(tool_input, "new_string", action.get("new")) + return CanonicalTool("Edit", tool_input) + + +def _apply_patch(action: dict[str, Any]) -> CanonicalTool: + patch = str(action.get("patch") or "") + tool_input: dict[str, Any] = { + "patch": patch, + "file_paths": _parse_patch_paths(patch), + } + return CanonicalTool("ApplyPatch", tool_input) + + +def _delete(action: dict[str, Any]) -> CanonicalTool: + tool_input: dict[str, Any] = {} + _include(tool_input, "file_path", action.get("path")) + return CanonicalTool("Delete", tool_input) + + +def _move(action: dict[str, Any]) -> CanonicalTool: + tool_input: dict[str, Any] = {} + _include(tool_input, "source", action.get("source")) + _include(tool_input, "destination", action.get("destination")) + return CanonicalTool("Move", tool_input) + + +def _read(action: dict[str, Any]) -> CanonicalTool: + tool_input: dict[str, Any] = {} + _include(tool_input, "file_path", action.get("path")) + return CanonicalTool("Read", tool_input) + + +def _read_range(action: dict[str, Any]) -> CanonicalTool: + tool_input: dict[str, Any] = {} + _include(tool_input, "file_path", action.get("path")) + _include(tool_input, "offset", action.get("start_line")) + _include(tool_input, "limit", action.get("max_lines")) + return CanonicalTool("Read", tool_input) + + +def _read_many(action: dict[str, Any]) -> CanonicalTool: + paths = action.get("paths") + tool_input: dict[str, Any] = {"file_paths": list(paths) if paths else []} + return CanonicalTool("Read", tool_input) + + +def _grep(action: dict[str, Any]) -> CanonicalTool: + tool_input: dict[str, Any] = {} + _include(tool_input, "pattern", action.get("query")) + _include(tool_input, "path", action.get("path")) + return CanonicalTool("Grep", tool_input) + + +def _ls(action: dict[str, Any]) -> CanonicalTool: + tool_input: dict[str, Any] = {} + _include(tool_input, "path", action.get("path")) + return CanonicalTool("LS", tool_input) + + +def _git_status(action: dict[str, Any]) -> CanonicalTool: + command = "git status" + path = action.get("path") + if path: + command += f" -- {path}" + tool_input: dict[str, Any] = {"command": command} + _include(tool_input, "include_untracked", action.get("include_untracked")) + return CanonicalTool("Bash", tool_input) + + +def _git_diff(action: dict[str, Any]) -> CanonicalTool: + command = "git diff" + if action.get("cached"): + command += " --cached" + if action.get("stat"): + command += " --stat" + path = action.get("path") + if path: + command += f" -- {path}" + tool_input: dict[str, Any] = {"command": command} + _include(tool_input, "max_chars", action.get("max_chars")) + return CanonicalTool("Bash", tool_input) + + +# native local-action name -> canonical builder. +_BUILDERS: dict[str, Callable[[dict[str, Any]], CanonicalTool]] = { + "run_command": _bash, + "write_file": _write, + "replace_text": _edit, + "apply_patch": _apply_patch, + "delete_file": _delete, + "move_file": _move, + "read_file": _read, + "read_file_range": _read_range, + "read_many_files": _read_many, + "search_text": _grep, + "list_files": _ls, + "stat_path": _ls, + "git_status": _git_status, + "git_diff": _git_diff, +} + + +def _mcp_passthrough(tool: str, action: dict[str, Any]) -> CanonicalTool: + """MCP tools already carry canonical ``mcp__server__name`` names. + + Pass the name through unchanged and expose the action's non-meta fields as + the tool input so a plugin sees the same shape it would on the SDK path. + """ + tool_input = { + key: value + for key, value in action.items() + if key not in ("tool", "name", "type") + } + return CanonicalTool(tool, tool_input) + + +def to_canonical_tool( + action: dict[str, Any], + *, + include_read_only: bool = False, +) -> CanonicalTool | None: + """Translate a native local action into the plugin hook vocabulary. + + Returns ``None`` when the action should not trigger a hook: control-plane + actions, unknown tools, or read-only actions when ``include_read_only`` is + false. MCP actions (``mcp__*``) pass through unchanged. + """ + tool = action_tool(action) + if not tool: + return None + if tool in NO_HOOK_TOOLS: + return None + if tool in READ_ONLY_TOOLS and not include_read_only: + return None + if tool.startswith("mcp__"): + return _mcp_passthrough(tool, action) + + builder = _BUILDERS.get(tool) + if builder is None: + return None + return builder(action) + + +def to_canonical_result(result: ToolActionResult) -> dict[str, Any]: + """Return the ``tool_result`` view a ``post_tool`` hook receives. + + Thin today (the native result dict is already a reasonable shape); kept as a + named seam so the post-hook result shape can evolve without touching the + execution loop. + """ + return result.to_dict() + + +def decision_to_result(decision: dict[str, Any], tool: str) -> ToolActionResult: + """Convert a plugin hook ``block`` decision into a failed action result. + + ``decision`` is the already-normalized hook response from + ``run_plugin_pre_tool_hooks`` — ``{"decision": "block", "reason": ...}`` — + whose reason is human-readable and plugin-attributed. + """ + reason = decision.get("reason") or "blocked by plugin runtime hook" + return ToolActionResult( + tool=tool or "unknown", + ok=False, + message=str(reason), + data={"blocked_by": "plugin_runtime"}, + ) diff --git a/apps/backend/plugins/runtime.py b/apps/backend/plugins/runtime.py index a382c5226..d57dffa79 100644 --- a/apps/backend/plugins/runtime.py +++ b/apps/backend/plugins/runtime.py @@ -64,6 +64,15 @@ def _can_use_tool_hooks(plugin: PluginBase) -> bool: ) +def can_use_tool_hooks(plugin: PluginBase) -> bool: + """Public predicate: whether an enabled plugin may participate in tool hooks. + + Backends that run their own tool loop (e.g. the Direct-API runtime) use this + to filter hook-capable plugins without reaching into private module state. + """ + return _can_use_tool_hooks(plugin) + + def build_agent_context( project_dir: Path, spec_dir: Path, diff --git a/docs/guides/plugin-development.md b/docs/guides/plugin-development.md index 45fc670ac..49fe93f9f 100644 --- a/docs/guides/plugin-development.md +++ b/docs/guides/plugin-development.md @@ -202,6 +202,65 @@ Check logs for your plugin's output. --- +## Tool Hooks and Backend Coverage + +Beyond the lifecycle hooks above, a plugin declaring the `generic_edit` or +`full_agent_runtime` capability can intercept individual tool calls: + +- `augment_prompt(context)` - contribute scoped instructions to the agent prompt +- `pre_tool(context, tool_name, tool_input)` - inspect a tool call before it + runs; return `ToolHookDecision.block(reason)` to prevent execution +- `post_tool(context, tool_name, tool_input, tool_result)` - observe a tool + call after it ran + +### Which hooks run on which backend + +Auto Code runs agents on more than one execution backend. Coverage differs, so +write hooks defensively: + +| Hook | Claude SDK (`full_autonomous`) | Direct-API in-process (`generic_edit`) | Codex / generic CLI | +|------|:---:|:---:|:---:| +| `before_session` / `after_session` / `on_message` | ✅ | ✅ | ✅ | +| `pre_tool` (block) | ✅ | ✅ | ❌ (opaque CLI loop) | +| `post_tool` (observe) | ✅ | ✅ (observational) | ❌ | +| `augment_prompt` | ✅ | see note | ❌ | + +Codex and other CLI backends run their own tool loop inside a separate process, +so per-tool interception is not possible there — only prompt-level and lifecycle +hooks can reach them. + +### Direct-API tool-name normalization + +The Direct-API runtime speaks its own local-action vocabulary (`run_command`, +`write_file`, `replace_text`, ...). Before your hook is called, each action is +normalized to the **canonical Claude SDK tool vocabulary** so a hook written +against SDK names matches on both backends. Match against the SDK name, not the +native one — e.g. a shell command is `Bash` with `{"command": ...}`, a full-file +write is `Write` with `{"file_path", "content"}`, an in-place edit is `Edit` with +`{"file_path", "old_string", "new_string"}`. Actions without an SDK equivalent +use a stable extension name: `Delete`, `Move`, `ApplyPatch`. The mapping lives in +`apps/backend/agents/runtime/plugin_tool_bridge.py`. + +### Direct-API limitations (design decisions, not bugs) + +1. **`pre_tool` is block-only.** On the SDK path a hook may return a *modified* + `tool_input`; on Direct-API the normalization is one-directional + (native → inspection view), so only `block`/allow is honored — input rewrite + is ignored. +2. **`post_tool` is observational.** The action has already executed, so a + `post_tool` block cannot un-run it. The block is logged and annotated onto + the result (`plugin_post_tool_block`) rather than rolled back. +3. **Read-only actions are gated off by default.** Reads/searches/listings + (`Read`, `Grep`, `LS`, git status/diff) do not trigger hooks on Direct-API + unless read-only hooking is explicitly enabled, so plugins don't run on every + file read. + +> **Note (`augment_prompt`)**: prompt augmentation currently reaches the Claude +> SDK backend only. Cross-backend prompt augmentation is tracked as separate +> work; do not rely on `augment_prompt` for Direct-API/Codex runs yet. + +--- + ## Plugin Manifest Every plugin requires a `plugin.json` manifest file: diff --git a/tests/test_generic_edit_plugin_hooks.py b/tests/test_generic_edit_plugin_hooks.py new file mode 100644 index 000000000..8606b2b10 --- /dev/null +++ b/tests/test_generic_edit_plugin_hooks.py @@ -0,0 +1,229 @@ +"""Integration tests for plugin pre_tool hooks on the Direct-API in-process +runtime (Slice 2 wiring in GenericEditRuntimeSession). + +These assert the wiring — native action -> canonical normalization -> plugin +pre_tool dispatch -> block translation -> read-only gating -> funnel through +``_execute_action`` — using a fake hook-capable plugin. The real skill-pack +guard behavior is covered separately (test_plugin_tool_bridge, test_skill_pack_runtime). +""" + +from __future__ import annotations + +import sys +from pathlib import Path +from types import SimpleNamespace + +import pytest + +REPO_ROOT = Path(__file__).resolve().parent.parent +sys.path.insert(0, str(REPO_ROOT / "apps" / "backend")) + +from agents.runtime.adapters.generic_edit import GenericEditRuntimeSession +from plugins.base import PluginCapability +from plugins.runtime import build_agent_context +from plugins.sdk.agent import ToolHookDecision + + +class _FakePlugin: + """Minimal hook-capable plugin: enabled + generic_edit capability.""" + + def __init__(self, predicate=None, *, post_predicate=None, name="fake-blocker"): + self.name = name + self.is_enabled = True + self.metadata = SimpleNamespace(capabilities=[PluginCapability.GENERIC_EDIT]) + self._predicate = predicate or (lambda _n, _i: False) + self._post_predicate = post_predicate + + def pre_tool(self, context, tool_name, tool_input): + if self._predicate(tool_name, tool_input): + return ToolHookDecision.block(f"{self.name} blocked {tool_name}") + return None + + def post_tool(self, context, tool_name, tool_input, tool_result): + if self._post_predicate and self._post_predicate( + tool_name, tool_input, tool_result + ): + return ToolHookDecision.block(f"{self.name} post-blocked {tool_name}") + return None + + +def _session(tmp_path: Path) -> GenericEditRuntimeSession: + return GenericEditRuntimeSession( + provider_name="openai", + agent_session=SimpleNamespace(), + project_dir=tmp_path, + agent_type="coder", + ) + + +def _wire( + session: GenericEditRuntimeSession, + tmp_path: Path, + predicate=None, + *, + post_predicate=None, +) -> None: + """Attach a hook-capable plugin + a real AgentContext, bypassing run().""" + session._tool_hook_plugins = [_FakePlugin(predicate, post_predicate=post_predicate)] + session._plugin_hook_context = build_agent_context( + tmp_path, tmp_path, "coder", metadata=None + ) + + +def _block_all(_tool_name, _tool_input) -> bool: + return True + + +def _block_skill_scripts(tool_name, tool_input) -> bool: + command = str(tool_input.get("command") or "") + return tool_name == "Bash" and "skills/" in command + + +async def test_pre_tool_blocks_mutating_action(tmp_path): + session = _session(tmp_path) + _wire(session, tmp_path, _block_all) + + result = await session._run_plugin_pre_tool_hook( + {"tool": "write_file", "path": "a.py", "content": "x"}, "write_file" + ) + + assert result is not None + assert result.ok is False + assert result.data == {"blocked_by": "plugin_runtime"} + assert "blocked Write" in result.message + + +async def test_pre_tool_blocks_skill_script_run_command(tmp_path): + session = _session(tmp_path) + _wire(session, tmp_path, _block_skill_scripts) + + blocked = await session._run_plugin_pre_tool_hook( + {"tool": "run_command", "command": "python skills/x/scripts/run.py"}, + "run_command", + ) + allowed = await session._run_plugin_pre_tool_hook( + {"tool": "run_command", "command": "pytest -q"}, "run_command" + ) + + assert blocked is not None and blocked.ok is False + assert allowed is None + + +async def test_pre_tool_gates_read_only_actions(tmp_path): + session = _session(tmp_path) + _wire(session, tmp_path, _block_all) # would block everything if reached + + # Read-only actions are gated off: the hook never runs, so no block. + assert ( + await session._run_plugin_pre_tool_hook( + {"tool": "read_file", "path": "a.py"}, "read_file" + ) + is None + ) + # A mutating action with the same plugin still blocks (control). + assert ( + await session._run_plugin_pre_tool_hook( + {"tool": "write_file", "path": "a.py", "content": "x"}, "write_file" + ) + is not None + ) + + +async def test_no_plugins_is_a_noop(tmp_path): + session = _session(tmp_path) + session._tool_hook_plugins = [] + session._plugin_hook_context = None + + assert ( + await session._run_plugin_pre_tool_hook( + {"tool": "write_file", "path": "a.py", "content": "x"}, "write_file" + ) + is None + ) + + +async def test_execute_action_funnels_block_before_executing(tmp_path): + session = _session(tmp_path) + _wire(session, tmp_path, _block_all) + target = tmp_path / "should_not_exist.py" + + result = await session._execute_action( + {"tool": "write_file", "path": "should_not_exist.py", "content": "x = 1"}, + loop="main", + iteration=0, + action_index=0, + spec_dir=tmp_path, + verbose=False, + phase=None, + subtask_id=None, + ) + + assert result.ok is False + assert result.data.get("blocked_by") == "plugin_runtime" + # Blocked before the executor ran -> the file was never written. + assert not target.exists() + + +async def test_post_tool_annotates_without_failing(tmp_path): + from agents.runtime.local_actions import ToolActionResult + + session = _session(tmp_path) + _wire(session, tmp_path, post_predicate=lambda _n, _i, _r: True) + result = ToolActionResult(tool="write_file", ok=True, message="wrote", data={}) + + await session._run_plugin_post_tool_hook( + {"tool": "write_file", "path": "a.py", "content": "x"}, "write_file", result + ) + + # Observational: success is preserved, the post-block is only annotated. + assert result.ok is True + assert "post-blocked Write" in result.data["plugin_post_tool_block"] + + +async def test_execute_action_runs_post_tool_on_success(tmp_path): + session = _session(tmp_path) + _wire(session, tmp_path, post_predicate=lambda _n, _i, _r: True) + target = tmp_path / "post_written.py" + + result = await session._execute_action( + {"tool": "write_file", "path": "post_written.py", "content": "x = 1"}, + loop="main", + iteration=0, + action_index=0, + spec_dir=tmp_path, + verbose=False, + phase=None, + subtask_id=None, + ) + + # The write succeeded AND post_tool observed it (mutation not rolled back). + assert result.ok is True + assert target.exists() + assert "plugin_post_tool_block" in result.data + + +async def test_resume_prepares_plugin_hook_context(tmp_path): + # resume() drives _execute_action like run(), so it must build the hook + # context too; otherwise tool hooks are a no-op on resumed runs. + session = _session(tmp_path) + seen = {} + + class _Sentinel(Exception): + pass + + def _spy(spec_dir, subtask_id): + seen["args"] = (spec_dir, subtask_id) + raise _Sentinel + + session._prepare_plugin_hook_context = _spy + + with pytest.raises(_Sentinel): + await session.resume( + checkpoint_path=tmp_path / "ckpt.json", + spec_dir=tmp_path, + verbose=False, + phase=None, + subtask_id="s1", + ) + + assert seen["args"] == (tmp_path, "s1") diff --git a/tests/test_plugin_tool_bridge.py b/tests/test_plugin_tool_bridge.py new file mode 100644 index 000000000..f2eace2d5 --- /dev/null +++ b/tests/test_plugin_tool_bridge.py @@ -0,0 +1,240 @@ +"""Unit tests for the Direct-API <-> plugin hook tool-name normalization layer. + +Slice 1: the module under test is pure (no plugin loading, no execution). These +tests lock the full native->canonical mapping table, the read-only gating flag, +the control-plane exclusions, and the block-decision translation. + +The regression anchor at the bottom is the reason this layer exists: it proves +that a native ``run_command`` invoking a skill script is only caught by +skill-pack-runtime's guard AFTER normalization to ``Bash`` — and that feeding +the guard the un-normalized native name silently fails to block. +""" + +from __future__ import annotations + +import sys +from pathlib import Path + +import pytest + +# Ensure apps/backend and the skill-pack plugin package root are importable. +REPO_ROOT = Path(__file__).resolve().parent.parent +sys.path.insert(0, str(REPO_ROOT / "apps" / "backend")) +sys.path.insert( + 0, + str(REPO_ROOT / "apps" / "backend" / "plugins" / "system" / "skill-pack-runtime"), +) + +from agents.runtime.local_actions import ToolActionResult +from agents.runtime.plugin_tool_bridge import ( + NO_HOOK_TOOLS, + READ_ONLY_TOOLS, + CanonicalTool, + decision_to_result, + to_canonical_result, + to_canonical_tool, +) +from skill_pack_runtime.runtime import SkillPackRuntime + +# --- mutating / command actions: always hooked ----------------------------- + +MUTATING_CASES = [ + ( + {"tool": "run_command", "command": "pytest -q", "timeout": 30}, + CanonicalTool("Bash", {"command": "pytest -q", "timeout": 30}), + ), + ( + {"tool": "run_command", "command": "pytest -q"}, + CanonicalTool("Bash", {"command": "pytest -q"}), + ), + ( + {"tool": "write_file", "path": "src/a.py", "content": "x = 1"}, + CanonicalTool("Write", {"file_path": "src/a.py", "content": "x = 1"}), + ), + ( + { + "tool": "replace_text", + "path": "src/a.py", + "old": "x", + "new": "y", + "count": 3, + }, + CanonicalTool( + "Edit", + {"file_path": "src/a.py", "old_string": "x", "new_string": "y"}, + ), + ), + ( + {"tool": "delete_file", "path": "src/gone.py"}, + CanonicalTool("Delete", {"file_path": "src/gone.py"}), + ), + ( + { + "tool": "move_file", + "source": "a.py", + "destination": "b.py", + "overwrite": True, + }, + CanonicalTool("Move", {"source": "a.py", "destination": "b.py"}), + ), +] + + +@pytest.mark.parametrize("action, expected", MUTATING_CASES) +def test_mutating_actions_map_to_canonical(action, expected): + assert to_canonical_tool(action) == expected + + +def test_replace_text_drops_count_from_view_only(): + # `count` is intentionally absent from the inspection view; the executor + # still honors it at execution time (normalization is inspection-only). + canonical = to_canonical_tool( + {"tool": "replace_text", "path": "a.py", "old": "x", "new": "y", "count": 5} + ) + assert "count" not in canonical.tool_input + + +def test_apply_patch_parses_touched_paths(): + patch = ( + "--- a/src/one.py\n" + "+++ b/src/one.py\n" + "@@ -1 +1 @@\n" + "-old\n+new\n" + "--- /dev/null\n" + "+++ b/src/two.py\n" + ) + canonical = to_canonical_tool({"tool": "apply_patch", "patch": patch}) + assert canonical.tool_name == "ApplyPatch" + assert canonical.tool_input["patch"] == patch + # /dev/null skipped, a//b/ prefixes stripped, sorted + de-duplicated. + assert canonical.tool_input["file_paths"] == ["src/one.py", "src/two.py"] + + +# --- read-only actions: gated behind include_read_only --------------------- + +READ_ONLY_CASES = [ + ( + {"tool": "read_file", "path": "a.py"}, + CanonicalTool("Read", {"file_path": "a.py"}), + ), + ( + {"tool": "read_file_range", "path": "a.py", "start_line": 10, "max_lines": 5}, + CanonicalTool("Read", {"file_path": "a.py", "offset": 10, "limit": 5}), + ), + ( + {"tool": "read_many_files", "paths": ["a.py", "b.py"]}, + CanonicalTool("Read", {"file_paths": ["a.py", "b.py"]}), + ), + ( + {"tool": "search_text", "query": "TODO", "path": "src"}, + CanonicalTool("Grep", {"pattern": "TODO", "path": "src"}), + ), + ({"tool": "list_files", "path": "src"}, CanonicalTool("LS", {"path": "src"})), + ({"tool": "stat_path", "path": "src"}, CanonicalTool("LS", {"path": "src"})), + ({"tool": "git_status"}, CanonicalTool("Bash", {"command": "git status"})), + ({"tool": "git_diff"}, CanonicalTool("Bash", {"command": "git diff"})), +] + + +@pytest.mark.parametrize("action, expected", READ_ONLY_CASES) +def test_read_only_actions_gated_off_by_default(action, expected): + assert to_canonical_tool(action) is None + assert to_canonical_tool(action, include_read_only=True) == expected + + +def test_read_only_set_matches_cases(): + assert {a["tool"] for a, _ in READ_ONLY_CASES} == set(READ_ONLY_TOOLS) + + +def test_git_actions_reflect_scope_in_command(): + # Scope (path/cached/stat) must survive into the canonical command so a hook + # consumer can tell a scoped/staged request from a bare one. + status = to_canonical_tool( + {"tool": "git_status", "path": "src", "include_untracked": True}, + include_read_only=True, + ) + assert status.tool_input["command"] == "git status -- src" + assert status.tool_input["include_untracked"] is True + + diff = to_canonical_tool( + {"tool": "git_diff", "path": "src", "cached": True, "stat": True}, + include_read_only=True, + ) + assert diff.tool_input["command"] == "git diff --cached --stat -- src" + + +# --- control-plane / unknown: never hooked --------------------------------- + + +@pytest.mark.parametrize("tool", sorted(NO_HOOK_TOOLS)) +def test_control_plane_actions_are_not_hooked(tool): + assert to_canonical_tool({"tool": tool}) is None + + +def test_unknown_and_empty_actions_are_not_hooked(): + assert to_canonical_tool({"tool": "totally_unknown"}) is None + assert to_canonical_tool({}) is None + + +def test_mcp_actions_pass_through_unchanged(): + action = {"tool": "mcp__ctx7__search", "query": "abc", "limit": 5} + canonical = to_canonical_tool(action) + assert canonical == CanonicalTool("mcp__ctx7__search", {"query": "abc", "limit": 5}) + + +# --- result / decision translation ----------------------------------------- + + +def test_to_canonical_result_reflects_action_result(): + result = ToolActionResult( + tool="write_file", ok=True, message="wrote", data={"n": 1} + ) + assert to_canonical_result(result) == result.to_dict() + + +def test_decision_to_result_produces_failed_result(): + decision = {"decision": "block", "reason": "Plugin X blocked tool use: nope"} + result = decision_to_result(decision, "run_command") + assert result.ok is False + assert result.tool == "run_command" + assert result.message == "Plugin X blocked tool use: nope" + assert result.data == {"blocked_by": "plugin_runtime"} + + +def test_decision_to_result_defaults_reason_and_tool(): + result = decision_to_result({"decision": "block"}, "") + assert result.ok is False + assert result.tool == "unknown" + assert result.message == "blocked by plugin runtime hook" + + +# --- regression anchor: normalization is what makes the guard fire ---------- + +_SKILL_SCRIPT_COMMAND = "python skills/my-skill/scripts/run.py" + + +def test_skill_pack_guard_blocks_only_after_normalization(): + """The whole point of Slice B: a native run_command that executes a skill + script must reach skill-pack-runtime's guard as ``Bash``. + """ + guard = SkillPackRuntime(REPO_ROOT, granted_permissions=[]) + action = {"tool": "run_command", "command": _SKILL_SCRIPT_COMMAND} + + # Un-normalized native name: guard does NOT match {"bash","shell"} -> bypass. + assert ( + guard.should_block_script_command( + "run_command", {"command": _SKILL_SCRIPT_COMMAND} + ) + is False + ) + + # Normalized: canonical name is Bash, guard blocks. + canonical = to_canonical_tool(action) + assert canonical.tool_name == "Bash" + assert guard.should_block_script_command(*canonical) is True + + +def test_skill_pack_guard_allows_benign_command_after_normalization(): + guard = SkillPackRuntime(REPO_ROOT, granted_permissions=[]) + canonical = to_canonical_tool({"tool": "run_command", "command": "pytest -q"}) + assert guard.should_block_script_command(*canonical) is False