diff --git a/scripts/mcp_hooks/__init__.py b/scripts/mcp_hooks/__init__.py new file mode 100644 index 00000000..0ab5636d --- /dev/null +++ b/scripts/mcp_hooks/__init__.py @@ -0,0 +1,32 @@ +""" +CodeLens MCP Hooks — auto-trigger scan/check after AI agent file writes. + +Implements the hook system described in issue #47 (Phase 1: post_tool hook). + +Hooks are *opt-in*: every hook defaults to ``enabled: false`` in +``.codelens/hooks.json``. The MCP server (:class:`mcp_server.HookManager`) +loads that config on first use, creates the default file if it is missing, +and runs each enabled hook non-blocking in a ThreadPoolExecutor so a hook +failure can never crash the server or stall a tool response. + +Public API +---------- +- :func:`post_tool.run_post_tool_hook` — single entry point used by + :class:`mcp_server.HookManager`. +""" + +from .post_tool import ( + run_post_tool_hook, + PostToolHookResult, + DEFAULT_CONFIG, + SEVERITY_ORDER, +) + +__all__ = [ + "run_post_tool_hook", + "PostToolHookResult", + "DEFAULT_CONFIG", + "SEVERITY_ORDER", +] + +__version__ = "1.0.0" diff --git a/scripts/mcp_hooks/post_tool.py b/scripts/mcp_hooks/post_tool.py new file mode 100644 index 00000000..d67a8912 --- /dev/null +++ b/scripts/mcp_hooks/post_tool.py @@ -0,0 +1,386 @@ +""" +CodeLens post_tool MCP hook — auto-scan a file after an AI agent writes it. + +Implements Phase 1 of issue #47. When an MCP agent (Claude Desktop, Cursor, +Continue.dev, etc.) writes a file via *any* MCP tool, the agent previously +had to remember to call ``codelens_scan`` followed by ``codelens_check`` to +discover newly-introduced security smells, secrets, or dead code. In +practice agents forgot ~70 % of the time, so high-severity findings shipped +to production unnoticed. + +This hook closes that gap. After every CodeLens MCP tool call whose +arguments reference a specific file, the hook (when enabled): + +1. Resolves the file path relative to the workspace. +2. Runs an incremental scan of the workspace so the registry reflects the + just-written file (``scan --incremental``). +3. Asks the smell / secrets / complexity engines whether the file produced + any new ``critical`` or ``high`` findings. +4. Returns a :class:`PostToolHookResult`. The :class:`HookManager` in + ``mcp_server.py`` surfaces that result to the agent either as a + JSON-RPC notification (``notifications/message``) or as a ``_hooks`` + field on the next ``tools/call`` response. + +Performance target: <500 ms added latency to the original tool call. The +hook itself runs entirely in a ThreadPoolExecutor so the tool response is +returned to the agent immediately; the only synchronous cost is the +argument inspection that decides whether to fire the hook at all (<<1 ms). + +Configuration (``.codelens/hooks.json``): + +.. code-block:: json + + { + "hooks": { + "post_tool": { + "enabled": false, + "severity_threshold": "high" + } + } + } + +All hooks default to ``enabled: false`` — users must opt in explicitly. +""" + +from __future__ import annotations + +import os +import sys +from dataclasses import dataclass, field, asdict +from typing import Any, Dict, List, Optional, Tuple + +# ─── Module-level constants ─────────────────────────────────────────── + +#: Severity ordering — lower index = more severe. Used both for the +#: ``severity_threshold`` comparison and to bucket findings. +SEVERITY_ORDER: Dict[str, int] = { + "critical": 0, + "high": 1, + "medium": 2, + "low": 3, + "info": 4, +} + +#: Default config block written into ``.codelens/hooks.json`` when the +#: file does not exist yet. Every hook defaults to *disabled* — opt-in. +DEFAULT_CONFIG: Dict[str, Any] = { + "hooks": { + "post_tool": { + "enabled": False, + "severity_threshold": "high", + } + } +} + +#: Engine commands the hook consults to surface new findings. Each entry +#: is (command_name, result_key, severity_for_match). The hook invokes the +#: command's ``execute`` function directly (in-process, no subprocess) to +#: stay well under the 500 ms latency budget. +_FINDING_SOURCES: List[Tuple[str, str, str]] = [ + # (engine, key_in_result, default_severity) + ("smell", "by_category", "high"), + ("secrets", "findings", "critical"), + ("complexity", "high_complexity", "high"), + ("dead-code", "by_category", "medium"), + ("debug-leak", "by_category", "low"), +] + + +@dataclass +class PostToolHookResult: + """Structured result returned by :func:`run_post_tool_hook`. + + Attributes + ---------- + triggered: + ``True`` if the hook actually ran (file resolved, scan executed). + ``False`` for early-exit cases (hook disabled, no file in args). + file_path: + Absolute path of the file that was inspected, or ``None``. + workspace: + Workspace root the scan was run against. + severity_threshold: + Severity level used to filter findings. + findings: + List of finding dicts (one per matching finding) for debugging. + critical_count / high_count: + Bucketed counts of the two severities the hook surfaces to the + agent. Lower severities are intentionally ignored. + message: + Human-readable message intended for the agent. Empty when nothing + was found or the hook did not run. + error: + ``None`` on success, otherwise a short description of why the hook + could not complete (file missing, scan failed, …). Errors never + raise — the hook is non-blocking and must not crash the MCP + server. + elapsed_ms: + Wall-clock time spent inside the hook, measured in milliseconds. + """ + + triggered: bool = False + file_path: Optional[str] = None + workspace: Optional[str] = None + severity_threshold: str = "high" + findings: List[Dict[str, Any]] = field(default_factory=list) + critical_count: int = 0 + high_count: int = 0 + message: str = "" + error: Optional[str] = None + elapsed_ms: float = 0.0 + + def to_dict(self) -> Dict[str, Any]: + """Return a JSON-serialisable dict (dataclasses.asdict wrapper).""" + return asdict(self) + + +# ─── Argument inspection ───────────────────────────────────────────── + + +def extract_file_path(arguments: Dict[str, Any], workspace: str) -> Optional[str]: + """Best-effort extraction of a file path from MCP tool call arguments. + + The hook is generic over every CodeLens MCP tool, so we inspect the + common argument names that carry a file path: ``file``, ``path``, + ``file_path``. Both absolute and workspace-relative paths are + accepted; relative paths are resolved against ``workspace``. + + Returns ``None`` when no file argument is present or the resolved + path does not exist on disk. Returning ``None`` is the signal to the + HookManager that the hook should not fire for this tool call. + """ + if not isinstance(arguments, dict): + return None + for key in ("file", "path", "file_path", "filename"): + value = arguments.get(key) + if not value or not isinstance(value, str): + continue + candidate = value + if not os.path.isabs(candidate) and workspace: + candidate = os.path.join(workspace, candidate) + candidate = os.path.normpath(candidate) + if os.path.isfile(candidate): + return candidate + # Some commands accept a file substring filter rather than a real + # path. Try to resolve it inside the workspace. + if workspace: + alt = os.path.join(workspace, value) + if os.path.isfile(alt): + return os.path.normpath(alt) + return None + + +# ─── Finding collection ────────────────────────────────────────────── + + +def _matches_file(finding: Any, file_path: str, workspace: str) -> bool: + """Return ``True`` if a finding struct references ``file_path``.""" + if not isinstance(finding, dict): + return False + haystack_keys = ("file", "path", "file_path", "filename", "source", "location") + rel_path = "" + try: + rel_path = os.path.relpath(file_path, workspace).replace(os.sep, "/") + except ValueError: + pass + norm_path = file_path.replace(os.sep, "/") + base_name = os.path.basename(file_path) + for key in haystack_keys: + val = finding.get(key) + if not isinstance(val, str): + continue + val_norm = val.replace(os.sep, "/") + if val_norm == norm_path or val_norm == rel_path or val_norm.endswith("/" + base_name): + return True + # Substring match — many engines store relative paths. + if rel_path and rel_path in val_norm: + return True + if base_name and base_name == os.path.basename(val_norm): + return True + return False + + +def _severity_of(finding: Dict[str, Any], default: str) -> str: + """Resolve a finding's severity, falling back to ``default``.""" + sev = finding.get("severity") or finding.get("level") or default + if isinstance(sev, str): + sev = sev.lower().strip() + if sev in SEVERITY_ORDER: + return sev + return default + + +def collect_findings_for_file( + workspace: str, + file_path: str, + severity_threshold: str = "high", +) -> List[Dict[str, Any]]: + """Run the lightweight analysis engines and return matching findings. + + Only findings whose severity is at or above ``severity_threshold`` AND + whose ``file`` field references ``file_path`` are returned. The + engines are imported lazily so importing :mod:`mcp_hooks.post_tool` + has no cost when hooks are disabled. + """ + threshold_idx = SEVERITY_ORDER.get( + str(severity_threshold).lower(), SEVERITY_ORDER["high"] + ) + matched: List[Dict[str, Any]] = [] + + # Lazy import — keeps the module import-cheap for disabled-hook case. + from commands import get_all_commands # type: ignore + + try: + registry = get_all_commands() + except Exception: + return matched + + # Reuse the _ArgsNamespace shim from the MCP server so we can invoke + # each command's execute() with the right shape. + from mcp_server import _ArgsNamespace # type: ignore + + args_ns = _ArgsNamespace({"file": file_path}, workspace) + # Some engines look at args.file; others scan the whole workspace and + # let us filter findings by file afterwards. We do both. + + for cmd_name, result_key, default_sev in _FINDING_SOURCES: + cmd_info = registry.get(cmd_name) + if not cmd_info or "execute" not in cmd_info: + continue + try: + result = cmd_info["execute"](args_ns, workspace) + except Exception: + # Engine failures must never break the hook. + continue + if not isinstance(result, dict): + continue + bucket = result.get(result_key) + items: List[Any] = [] + if isinstance(bucket, list): + items = bucket + elif isinstance(bucket, dict): + for sub in bucket.values(): + if isinstance(sub, list): + items.extend(sub) + elif result_key == "by_category" and isinstance(bucket, dict): + items = [i for sub in bucket.values() if isinstance(sub, list) for i in sub] + for item in items: + if not isinstance(item, dict): + continue + sev = _severity_of(item, default_sev) + if SEVERITY_ORDER.get(sev, 99) > threshold_idx: + continue + if _matches_file(item, file_path, workspace): + item_copy = dict(item) + item_copy.setdefault("severity", sev) + item_copy.setdefault("source_engine", cmd_name) + matched.append(item_copy) + + return matched + + +# ─── Public entry point ────────────────────────────────────────────── + + +def run_post_tool_hook( + arguments: Dict[str, Any], + workspace: str, + severity_threshold: str = "high", +) -> PostToolHookResult: + """Run the post_tool hook for a single MCP tool call. + + This function is *the* contract between :class:`HookManager` and the + hook. It is intentionally synchronous — the HookManager wraps the + call in a ThreadPoolExecutor so callers never block. + + Parameters + ---------- + arguments: + The MCP tool call arguments dict (``params.arguments``). + workspace: + Absolute path to the workspace root. Used to resolve relative + file paths and as the scan target. + severity_threshold: + Findings below this severity are ignored. Defaults to ``high``. + + Returns + ------- + PostToolHookResult + Always returns — never raises. Any internal error is captured in + ``PostToolHookResult.error``. + """ + import time + + start = time.monotonic() + result = PostToolHookResult( + workspace=workspace, + severity_threshold=str(severity_threshold).lower(), + ) + + if not workspace or not os.path.isdir(workspace): + result.error = "workspace not found" + result.elapsed_ms = (time.monotonic() - start) * 1000.0 + return result + + file_path = extract_file_path(arguments, workspace) + if not file_path: + # No file in args → nothing to do. Not an error. + result.elapsed_ms = (time.monotonic() - start) * 1000.0 + return result + + result.file_path = file_path + result.triggered = True + + try: + # 1. Incremental scan so the registry reflects the just-written file. + # The scan command does not accept a ``--file`` flag, so we run an + # incremental workspace scan (cheap when only one file changed). + from commands.scan import cmd_scan # type: ignore + + cmd_scan(workspace, incremental=True) + + # 2. Collect findings for the target file from the lightweight engines. + findings = collect_findings_for_file( + workspace, file_path, result.severity_threshold + ) + result.findings = findings + for f in findings: + sev = _severity_of(f, "medium") + if sev == "critical": + result.critical_count += 1 + elif sev == "high": + result.high_count += 1 + + # 3. Build the agent-facing message. Only critical/high counts are + # surfaced to keep the message short (issue #47 spec). + if result.critical_count or result.high_count: + short = os.path.basename(file_path) + parts = [] + if result.critical_count: + parts.append(f"{result.critical_count} critical") + if result.high_count: + parts.append(f"{result.high_count} high") + result.message = ( + f"⚠️ post_tool hook: {' and '.join(parts)} " + f"finding(s) in {short}" + ) + except Exception as exc: # pragma: no cover — defensive guard + # Hook failures must never propagate to the MCP server. + result.error = f"{type(exc).__name__}: {exc}" + try: + print(f"[CodeLens MCP] post_tool hook error: {exc}", file=sys.stderr) + except Exception: + pass + + result.elapsed_ms = (time.monotonic() - start) * 1000.0 + return result + + +__all__ = [ + "PostToolHookResult", + "run_post_tool_hook", + "extract_file_path", + "collect_findings_for_file", + "DEFAULT_CONFIG", + "SEVERITY_ORDER", +] diff --git a/scripts/mcp_server.py b/scripts/mcp_server.py index 8bf1307a..da26552a 100644 --- a/scripts/mcp_server.py +++ b/scripts/mcp_server.py @@ -34,7 +34,8 @@ import threading import time import traceback -from typing import Any, Dict, List, Optional, Tuple +from concurrent.futures import ThreadPoolExecutor +from typing import Any, Callable, Dict, List, Optional, Tuple from datetime import datetime, timezone # Add scripts directory to path @@ -1115,6 +1116,11 @@ def __init__(self, watch: bool = False): self._client_info = {} self._command_registry = None # Lazy loaded self._workspace_registries: Dict[str, Dict[str, Any]] = {} # In-memory registry cache + # Hook manager (issue #47 Phase 1) — defaults to all hooks disabled. + # Bound to a workspace lazily on first tool call so the manager + # picks up the auto-detected workspace. + self._hook_manager: Optional[HookManager] = None + self._hook_workspace: Optional[str] = None # ─── Lifecycle ──────────────────────────────────────── @@ -1199,6 +1205,14 @@ def _shutdown(self) -> None: self._watcher.stop() except Exception: pass + # Shut down the hook manager (issue #47) so pending hook tasks + # are cancelled and the worker threads exit cleanly. + if self._hook_manager is not None: + try: + self._hook_manager.shutdown() + except Exception: + pass + self._hook_manager = None elapsed = time.time() - self._start_time stats = self._cache.stats() print( @@ -1444,7 +1458,7 @@ def _handle_tools_call(self, params: Dict[str, Any]) -> Dict[str, Any]: if cmd_name not in ("scan", "init"): cached = self._cache.get(cache_key) if cached is not None: - return { + response = { "content": [{ "type": "text", "text": json.dumps(cached, ensure_ascii=False) @@ -1452,6 +1466,11 @@ def _handle_tools_call(self, params: Dict[str, Any]) -> Dict[str, Any]: "isError": False, "_cached": True } + # Surface any hook notifications that have arrived since the + # last tool call (issue #47). Even cached responses must + # drain the queue so the agent doesn't miss warnings. + self._attach_pending_hooks(response) + return response # Execute the command try: @@ -1466,7 +1485,7 @@ def _handle_tools_call(self, params: Dict[str, Any]) -> Dict[str, Any]: if cmd_name == "scan" and workspace: self._load_registry_to_memory(workspace) - return { + response = { "content": [{ "type": "text", "text": json.dumps(result, ensure_ascii=False, default=str) @@ -1474,9 +1493,21 @@ def _handle_tools_call(self, params: Dict[str, Any]) -> Dict[str, Any]: "isError": False } + # ─── Post-tool hook (issue #47) ────────────────────────── + # Schedule the (opt-in) post_tool hook non-blocking. Even when + # disabled this call is <1 ms — it does an enabled-flag check + # and returns. Failures here must never break the response. + self._maybe_run_post_tool_hook(tool_name, arguments, workspace) + # Attach any *previously* queued hook notifications to this + # response. (The hook we just scheduled will surface on the + # next call — that's by design: hooks are non-blocking.) + self._attach_pending_hooks(response) + + return response + except Exception as e: traceback.print_exc(file=sys.stderr) - return { + response = { "content": [{ "type": "text", "text": json.dumps({ @@ -1488,6 +1519,103 @@ def _handle_tools_call(self, params: Dict[str, Any]) -> Dict[str, Any]: }], "isError": True } + # Even on errors we drain pending hook notifications so the + # agent doesn't lose them — issue #47 spec calls for hook + # results to be surfaced via "next tool response". + self._attach_pending_hooks(response) + return response + + # ─── Hook integration (issue #47) ───────────────────── + + def _get_hook_manager(self, workspace: str) -> "Optional[HookManager]": + """Lazily create the HookManager bound to ``workspace``. + + The manager is bound to the first workspace we see and re-used for + the lifetime of the server. ``.codelens/hooks.json`` lives at the + workspace root, so switching workspaces would silently change the + config — we treat the first workspace as authoritative to keep + behavior predictable for the agent. + """ + if not workspace: + return None + workspace = os.path.abspath(workspace) + if self._hook_manager is None: + try: + self._hook_manager = HookManager( + workspace=workspace, + send_notification=self._send_hook_notification, + ) + self._hook_workspace = workspace + except Exception as exc: + # Hook manager construction must never break the server. + print( + f"[CodeLens MCP] failed to construct HookManager: {exc}", + file=sys.stderr, + ) + return None + return self._hook_manager + + def _maybe_run_post_tool_hook( + self, + tool_name: str, + arguments: Dict[str, Any], + workspace: str, + ) -> None: + """Fire the post_tool hook if enabled (issue #47). + + Wrapped in try/except so any failure inside the hook subsystem + (config parse error, executor closed, hook bug) leaves the MCP + tool response untouched. + """ + try: + manager = self._get_hook_manager(workspace) + if manager is None: + return + manager.after_tool_call(tool_name, arguments, workspace) + except Exception as exc: + try: + print( + f"[CodeLens MCP] post_tool hook dispatch failed: {exc}", + file=sys.stderr, + ) + except Exception: + pass + + def _attach_pending_hooks(self, response: Dict[str, Any]) -> None: + """Drain queued hook notifications and attach them to ``response``. + + Adds a ``_hooks`` field to the response payload (in addition to + the standard MCP fields). The MCP spec does not define this field, + but the issue #47 spec explicitly allows hook results to be + "added to next tool response" — agents ignore unknown top-level + fields by spec. + """ + try: + manager = self._hook_manager + if manager is None: + return + pending = manager.drain_pending() + if pending: + response["_hooks"] = pending + except Exception: + pass + + def _send_hook_notification(self, notification: Dict[str, Any]) -> None: + """Push a hook notification to the agent via stdout JSON-RPC. + + The MCP spec allows the server to send unsolicited + ``notifications/message`` payloads at any time. We write them + inline to stdout (the same channel used for responses) — agents + route them by ``method``. + """ + try: + payload = json.dumps(notification, ensure_ascii=False) + sys.stdout.write(payload + "\n") + sys.stdout.flush() + except Exception: + # If stdout is closed (e.g. during shutdown), silently drop. + pass + def _handle_resources_list(self) -> Dict[str, Any]: """Handle resources/list request. Expose codebase registries as resources.""" @@ -1806,6 +1934,371 @@ def __repr__(self): return f"_ArgsNamespace(workspace={self.workspace!r}, args={self._arguments!r})" +# ─── MCP Hook Manager (issue #47, Phase 1: post_tool) ──────────────── + + +class HookManager: + """Manages opt-in MCP hooks that auto-trigger scan/check after AI writes. + + Implements Phase 1 of issue #47. The manager: + + 1. Reads ``.codelens/hooks.json`` from the workspace on construction, + creating the file with default content (all hooks ``enabled: false``) + if it does not exist yet. + 2. After every MCP tool call that *might* modify a file, the + :meth:`after_tool_call` method schedules the ``post_tool`` hook + non-blocking in a :class:`ThreadPoolExecutor`. The hook itself is + implemented in :mod:`mcp_hooks.post_tool`. + 3. When the hook surfaces a new critical/high finding, the manager + enqueues an MCP ``notifications/message`` payload that the caller + can either emit immediately via stdout or attach to the *next* + ``tools/call`` response as a ``_hooks`` field. + + Design constraints (issue #47): + + - **Default disabled**: every hook defaults to ``enabled: false`` — the + user must opt in explicitly via ``.codelens/hooks.json``. + - **Non-blocking**: ``after_tool_call`` returns in <<1 ms (it only + schedules the executor). Hook failures are caught, logged to stderr, + and never propagated to the caller. The MCP server stays alive. + - **Performance target**: <500 ms added latency to the original tool + call. Because the hook runs entirely off the request thread, the + only synchronous cost is argument inspection that decides whether + to fire the hook at all. + + The class is intentionally usable in isolation (no MCPServer required) + so unit tests can exercise config loading, hook dispatch, and + notification buffering without booting the full server. + """ + + #: Path of the hooks config file, relative to the workspace root. + CONFIG_PATH = os.path.join(".codelens", "hooks.json") + + def __init__( + self, + workspace: Optional[str] = None, + max_workers: int = 2, + send_notification: Optional[Callable[[Dict[str, Any]], None]] = None, + ): + """Create a hook manager bound to ``workspace``. + + Parameters + ---------- + workspace: + Workspace root used to locate ``.codelens/hooks.json`` and to + pass through to the hook implementation. Falls back to the + current working directory if ``None``. + max_workers: + Size of the underlying ThreadPoolExecutor. Default 2 is enough + for the post_tool hook — hooks share the pool. + send_notification: + Optional callback invoked with each MCP notification dict the + hook produces. The MCPServer wires this to its stdout writer so + notifications reach the agent in real time. If ``None``, the + notifications are buffered in :attr:`_pending` and surfaced via + :meth:`drain_pending` (useful for tests and for the + "attach to next response" mode). + """ + self._workspace = workspace or os.getcwd() + self._send_notification = send_notification + self._lock = threading.Lock() + self._pending: List[Dict[str, Any]] = [] + self._executor: Optional[ThreadPoolExecutor] = None + try: + self._executor = ThreadPoolExecutor( + max_workers=max_workers, + thread_name_prefix="codelens-hook", + ) + except RuntimeError: + # Interpreter shutdown in progress — leave executor as None, + # hooks will be no-ops. Server is going down anyway. + self._executor = None + self._config: Dict[str, Any] = self._load_or_create_config(self._workspace) + + # ─── Config ──────────────────────────────────────────── + + @staticmethod + def _default_config() -> Dict[str, Any]: + """Return a fresh deep copy of the default (all-disabled) config.""" + from mcp_hooks.post_tool import DEFAULT_CONFIG + + # json round-trip for a robust deep copy across Python versions. + return json.loads(json.dumps(DEFAULT_CONFIG)) + + @classmethod + def _load_or_create_config(cls, workspace: str) -> Dict[str, Any]: + """Load ``.codelens/hooks.json`` from ``workspace``. + + If the file does not exist, create it with the default config + (all hooks disabled — opt-in). If the file exists but is + unreadable or malformed, fall back to defaults silently so a + corrupted config never breaks the MCP server. + + Missing per-hook keys are backfilled from the defaults so the rest + of the manager can rely on the schema being complete. + """ + default = cls._default_config() + if not workspace or not os.path.isdir(workspace): + return default + config_path = os.path.join(workspace, cls.CONFIG_PATH) + try: + if not os.path.isfile(config_path): + os.makedirs(os.path.dirname(config_path), exist_ok=True) + with open(config_path, "w", encoding="utf-8") as f: + json.dump(default, f, indent=2, ensure_ascii=False) + return default + + with open(config_path, "r", encoding="utf-8") as f: + loaded = json.load(f) + if not isinstance(loaded, dict): + return default + + # Merge: start from defaults so newly-added hooks appear. + merged = default + hooks_block = loaded.get("hooks") + if isinstance(hooks_block, dict): + merged_hooks = dict(default.get("hooks", {})) + for name, cfg in hooks_block.items(): + if not isinstance(cfg, dict): + continue + base = dict(merged_hooks.get(name, {"enabled": False})) + base.update(cfg) + merged_hooks[name] = base + merged = dict(default) + merged["hooks"] = merged_hooks + return merged + except (OSError, json.JSONDecodeError, TypeError): + return default + + def reload_config(self) -> None: + """Re-read the config file from disk (call after the user edits it).""" + with self._lock: + self._config = self._load_or_create_config(self._workspace) + + @property + def config(self) -> Dict[str, Any]: + """Snapshot of the merged config dict.""" + with self._lock: + # Return a shallow copy so callers can't mutate the live view. + return dict(self._config) + + @property + def workspace(self) -> str: + return self._workspace + + def is_enabled(self, hook_name: str) -> bool: + """Return ``True`` only when ``hook_name`` is explicitly enabled. + + Defaults to ``False`` for unknown hooks — opt-in. + """ + with self._lock: + hooks_block = self._config.get("hooks", {}) + if not isinstance(hooks_block, dict): + return False + cfg = hooks_block.get(hook_name, {}) + if not isinstance(cfg, dict): + return False + # Only an explicit truthy bool enables a hook. + return bool(cfg.get("enabled", False)) + + def severity_threshold(self, hook_name: str) -> str: + """Return the configured severity threshold for ``hook_name``. + + Falls back to ``"high"`` for unknown hooks / malformed entries — + matches the issue #47 default schema. + """ + with self._lock: + hooks_block = self._config.get("hooks", {}) + if not isinstance(hooks_block, dict): + return "high" + cfg = hooks_block.get(hook_name, {}) + if not isinstance(cfg, dict): + return "high" + sev = cfg.get("severity_threshold", "high") + if not isinstance(sev, str) or not sev.strip(): + return "high" + return sev.strip().lower() + + # ─── Hook dispatch ──────────────────────────────────── + + def after_tool_call( + self, + tool_name: str, + arguments: Dict[str, Any], + workspace: Optional[str] = None, + on_complete: Optional[Callable[["PostToolHookResult"], None]] = None, + ) -> None: + """Schedule the post_tool hook non-blocking. + + This is the *only* method :class:`MCPServer` calls after every + tool call. It never raises: any internal error is logged to stderr + and swallowed so the MCP server stays alive. + + ``on_complete`` is invoked from the worker thread once the hook + finishes (success or failure). Tests use it to wait for the + async hook deterministically. + + Parameters + ---------- + tool_name: + The MCP tool name (e.g. ``codelens_query``). + arguments: + The MCP tool call arguments dict (``params.arguments``). + Snapshotted before submission so later mutations don't race. + workspace: + Optional workspace override. Defaults to the manager's + workspace. + on_complete: + Optional callback invoked with the :class:`PostToolHookResult` + when the hook finishes. Failures in the callback are swallowed. + """ + if not self.is_enabled("post_tool"): + # Hook disabled — fast path with zero side effects. + return + if self._executor is None: + # Server is shutting down — silently no-op. + return + ws = workspace or self._workspace or "" + if not ws: + return + if not isinstance(arguments, dict): + arguments = {} + # Shallow-copy arguments so the worker thread sees a stable snapshot + # even if the caller mutates the dict after we return. + args_snapshot = dict(arguments) + severity = self.severity_threshold("post_tool") + try: + self._executor.submit( + self._run_post_tool_blocking, + tool_name, + args_snapshot, + ws, + severity, + on_complete, + ) + except RuntimeError: + # Executor was shut down between the None check and submit. + pass + + def _run_post_tool_blocking( + self, + tool_name: str, + arguments: Dict[str, Any], + workspace: str, + severity_threshold: str, + on_complete: Optional[Callable[["PostToolHookResult"], None]], + ) -> None: + """Worker entry point — runs the hook and buffers/forwards the result. + + Wrapped in a try/except so a bug in the hook implementation can + never take down the executor thread (which would leak the pool). + """ + try: + from mcp_hooks.post_tool import run_post_tool_hook + + result = run_post_tool_hook(arguments, workspace, severity_threshold) + except Exception as exc: # pragma: no cover — defensive guard + try: + print( + f"[CodeLens MCP] post_tool hook crashed: {exc}", + file=sys.stderr, + ) + except Exception: + pass + return + + # Surface critical/high findings to the agent. + notification = self._result_to_notification(result, tool_name) + if notification is not None: + self._emit_notification(notification) + + if on_complete is not None: + try: + on_complete(result) + except Exception: + # Callback failures must not break the worker thread. + pass + + def _result_to_notification( + self, + result: "PostToolHookResult", + tool_name: str, + ) -> Optional[Dict[str, Any]]: + """Build an MCP ``notifications/message`` payload from a hook result. + + Returns ``None`` when the hook did not surface anything worth + notifying the agent about (no critical/high findings, hook did + not trigger, or hook errored without findings). + """ + if not getattr(result, "triggered", False): + return None + if not (result.critical_count or result.high_count): + return None + return { + "jsonrpc": "2.0", + "method": "notifications/message", + "params": { + "level": "warning", + "data": { + "source": "codelens.post_tool_hook", + "tool": tool_name, + "file": result.file_path, + "workspace": result.workspace, + "severity_threshold": result.severity_threshold, + "critical_count": result.critical_count, + "high_count": result.high_count, + "message": result.message, + "elapsed_ms": round(result.elapsed_ms, 2), + "error": result.error, + }, + }, + } + + def _emit_notification(self, notification: Dict[str, Any]) -> None: + """Either forward the notification via the callback or buffer it.""" + if self._send_notification is not None: + try: + self._send_notification(notification) + return + except Exception: + # Forwarding failed — fall through and buffer so the + # notification still surfaces via the next tool response. + pass + with self._lock: + self._pending.append(notification) + + # ─── Pending notification queue ─────────────────────── + + def drain_pending(self) -> List[Dict[str, Any]]: + """Atomically return and clear all queued hook notifications.""" + with self._lock: + pending = list(self._pending) + self._pending.clear() + return pending + + def pending_notifications(self) -> List[Dict[str, Any]]: + """Return a copy of the queued hook notifications (does not clear).""" + with self._lock: + return list(self._pending) + + # ─── Lifecycle ──────────────────────────────────────── + + def shutdown(self) -> None: + """Cancel queued hook tasks and shut down the executor. + + Called by :class:`MCPServer` during graceful shutdown. + """ + executor = self._executor + self._executor = None + if executor is None: + return + try: + # cancel_futures added in Python 3.9 — fall back gracefully. + executor.shutdown(wait=False, cancel_futures=True) + except TypeError: # pragma: no cover — Python <3.9 fallback + executor.shutdown(wait=False) + + # ─── HTTP/SSE Transport (Optional) ──────────────────────────────────── def start_http_server(mcp_server: MCPServer, port: int = 8080) -> None: diff --git a/tests/test_mcp_hooks.py b/tests/test_mcp_hooks.py new file mode 100644 index 00000000..70cfcd25 --- /dev/null +++ b/tests/test_mcp_hooks.py @@ -0,0 +1,816 @@ +"""Tests for the MCP post_tool hook (issue #47, Phase 1). + +These tests cover the full hook subsystem: + +- ``scripts/mcp_hooks/__init__.py`` and ``scripts/mcp_hooks/post_tool.py`` + provide the hook implementation. +- :class:`mcp_server.HookManager` reads ``.codelens/hooks.json`` (creating + it with all hooks disabled when missing) and dispatches the post_tool + hook non-blocking in a ThreadPoolExecutor. +- :class:`mcp_server.MCPServer` wires the manager into ``tools/call`` so + every successful tool invocation triggers the hook (when enabled) and + attaches any queued hook notifications to the response. + +The two contract tests required by the issue #47 spec are: + +1. **hook disabled → no side effect**: when the config has ``enabled: false``, + calling ``after_tool_call`` must not invoke the hook implementation at all. +2. **hook enabled → scan called**: when the config has ``enabled: true``, + calling ``after_tool_call`` must invoke ``run_post_tool_hook`` with the + right arguments. + +The remaining tests cover config loading/creation, notification buffering, +non-blocking behavior, error isolation, and the MCPServer integration. +""" + +from __future__ import annotations + +import json +import os +import sys +import tempfile +import threading +import time +from unittest import mock + +import pytest + +# ─── Path setup (mirrors test_cli.py / test_compact_format.py) ──────── + +SCRIPT_DIR = os.path.join( + os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "scripts" +) +if SCRIPT_DIR not in sys.path: + sys.path.insert(0, SCRIPT_DIR) + +from mcp_hooks.post_tool import ( # noqa: E402 (after sys.path tweak) + DEFAULT_CONFIG, + SEVERITY_ORDER, + PostToolHookResult, + extract_file_path, + run_post_tool_hook, +) +from mcp_server import HookManager, MCPServer # noqa: E402 + + +# ─── Fixtures ───────────────────────────────────────────────────────── + + +@pytest.fixture +def tmp_workspace(): + """Fresh empty workspace directory; cleans up after the test.""" + with tempfile.TemporaryDirectory() as ws: + yield ws + + +@pytest.fixture +def enabled_workspace(tmp_workspace): + """Workspace with ``.codelens/hooks.json`` post_tool enabled.""" + codelens_dir = os.path.join(tmp_workspace, ".codelens") + os.makedirs(codelens_dir, exist_ok=True) + with open(os.path.join(codelens_dir, "hooks.json"), "w", encoding="utf-8") as f: + json.dump( + { + "hooks": { + "post_tool": { + "enabled": True, + "severity_threshold": "high", + } + } + }, + f, + ) + return tmp_workspace + + +@pytest.fixture +def disabled_workspace(tmp_workspace): + """Workspace with ``.codelens/hooks.json`` post_tool disabled (default).""" + codelens_dir = os.path.join(tmp_workspace, ".codelens") + os.makedirs(codelens_dir, exist_ok=True) + with open(os.path.join(codelens_dir, "hooks.json"), "w", encoding="utf-8") as f: + json.dump(DEFAULT_CONFIG, f) + return tmp_workspace + + +def _make_result(**overrides) -> PostToolHookResult: + """Build a :class:`PostToolHookResult` with sensible test defaults.""" + defaults = dict( + triggered=True, + file_path="/tmp/example.py", + workspace="/tmp", + severity_threshold="high", + findings=[], + critical_count=0, + high_count=0, + message="", + error=None, + elapsed_ms=12.3, + ) + defaults.update(overrides) + return PostToolHookResult(**defaults) + + +# ─── 1. Default config & schema ─────────────────────────────────────── + + +class TestDefaultConfig: + """The shipped default config must be opt-in (all hooks disabled).""" + + def test_default_config_has_post_tool_disabled(self): + assert DEFAULT_CONFIG["hooks"]["post_tool"]["enabled"] is False + + def test_default_config_severity_threshold_is_high(self): + assert DEFAULT_CONFIG["hooks"]["post_tool"]["severity_threshold"] == "high" + + def test_severity_order_critical_is_most_severe(self): + # The hook uses SEVERITY_ORDER to bucket findings; critical must be + # the most severe so a critical finding always surfaces. + assert SEVERITY_ORDER["critical"] == 0 + assert SEVERITY_ORDER["critical"] < SEVERITY_ORDER["high"] + + +# ─── 2. HookManager: config loading & file creation ────────────────── + + +class TestHookManagerConfig: + """HookManager creates ``.codelens/hooks.json`` when missing and loads it.""" + + def test_creates_hooks_json_when_missing(self, tmp_workspace): + # No .codelens directory yet. + assert not os.path.exists(os.path.join(tmp_workspace, ".codelens")) + + HookManager(workspace=tmp_workspace) + + config_path = os.path.join(tmp_workspace, ".codelens", "hooks.json") + assert os.path.isfile(config_path) + with open(config_path, "r", encoding="utf-8") as f: + written = json.load(f) + assert written == DEFAULT_CONFIG + # Hard constraint: post_tool must be disabled by default. + assert written["hooks"]["post_tool"]["enabled"] is False + + def test_loads_existing_enabled_config(self, enabled_workspace): + mgr = HookManager(workspace=enabled_workspace) + assert mgr.is_enabled("post_tool") is True + assert mgr.severity_threshold("post_tool") == "high" + + def test_loads_existing_disabled_config(self, disabled_workspace): + mgr = HookManager(workspace=disabled_workspace) + assert mgr.is_enabled("post_tool") is False + + def test_unknown_hook_defaults_to_disabled(self, enabled_workspace): + mgr = HookManager(workspace=enabled_workspace) + # Any hook not explicitly enabled must be disabled — opt-in. + assert mgr.is_enabled("pre_tool") is False + assert mgr.is_enabled("nonexistent_hook") is False + + def test_malformed_config_falls_back_to_defaults(self, tmp_workspace): + codelens_dir = os.path.join(tmp_workspace, ".codelens") + os.makedirs(codelens_dir, exist_ok=True) + # Write invalid JSON. + with open(os.path.join(codelens_dir, "hooks.json"), "w") as f: + f.write("{ this is not valid json") + + mgr = HookManager(workspace=tmp_workspace) + # Hook stays disabled (default) — malformed config never enables. + assert mgr.is_enabled("post_tool") is False + assert mgr.severity_threshold("post_tool") == "high" + + def test_reload_picks_up_config_changes(self, tmp_workspace): + mgr = HookManager(workspace=tmp_workspace) + assert mgr.is_enabled("post_tool") is False + + # Flip the config to enabled. + config_path = os.path.join(tmp_workspace, ".codelens", "hooks.json") + with open(config_path, "r", encoding="utf-8") as f: + cfg = json.load(f) + cfg["hooks"]["post_tool"]["enabled"] = True + with open(config_path, "w", encoding="utf-8") as f: + json.dump(cfg, f) + + mgr.reload_config() + assert mgr.is_enabled("post_tool") is True + + def test_missing_workspace_falls_back_to_defaults(self): + # Non-existent path → no crash, defaults returned. + mgr = HookManager(workspace="/path/that/does/not/exist") + assert mgr.is_enabled("post_tool") is False + assert mgr.config == DEFAULT_CONFIG + + +# ─── 3. CONTRACT TEST: hook disabled → no side effect ──────────────── + + +class TestHookDisabledNoSideEffect: + """When ``post_tool`` is disabled, ``after_tool_call`` must do nothing. + + This is the first required contract test from issue #47. + """ + + def test_disabled_hook_does_not_call_run_post_tool_hook( + self, disabled_workspace + ): + mgr = HookManager(workspace=disabled_workspace) + assert mgr.is_enabled("post_tool") is False + + with mock.patch( + "mcp_hooks.post_tool.run_post_tool_hook" + ) as mock_hook: + mgr.after_tool_call( + "codelens_query", + {"file": "app.py", "workspace": disabled_workspace}, + ) + + # Give the (non-existent) worker thread a moment to prove it + # never fires. + time.sleep(0.05) + mock_hook.assert_not_called() + + def test_disabled_hook_does_not_enqueue_notifications( + self, disabled_workspace + ): + mgr = HookManager(workspace=disabled_workspace) + mgr.after_tool_call( + "codelens_query", + {"file": "app.py", "workspace": disabled_workspace}, + ) + # Drain after a brief wait — queue must stay empty. + time.sleep(0.05) + assert mgr.drain_pending() == [] + + def test_disabled_hook_does_not_create_executor_work( + self, disabled_workspace + ): + mgr = HookManager(workspace=disabled_workspace) + # Spy on the executor's submit — must not be invoked at all when + # the hook is disabled (fast-path early return). + with mock.patch.object(mgr._executor, "submit") as mock_submit: + mgr.after_tool_call( + "codelens_query", + {"file": "app.py", "workspace": disabled_workspace}, + ) + mock_submit.assert_not_called() + + +# ─── 4. CONTRACT TEST: hook enabled → scan called ──────────────────── + + +class TestHookEnabledScanCalled: + """When ``post_tool`` is enabled, ``after_tool_call`` must invoke the hook. + + This is the second required contract test from issue #47. We mock the + underlying ``run_post_tool_hook`` so the test is fast and deterministic + — the contract under test is "the hook is *invoked*", not "the scan + engine produces findings". + """ + + def test_enabled_hook_calls_run_post_tool_hook(self, enabled_workspace): + mgr = HookManager(workspace=enabled_workspace) + assert mgr.is_enabled("post_tool") is True + + done = threading.Event() + captured = {} + + def fake_run_post_tool_hook(arguments, workspace, severity_threshold): + captured["arguments"] = arguments + captured["workspace"] = workspace + captured["severity_threshold"] = severity_threshold + done.set() + return _make_result() + + with mock.patch( + "mcp_hooks.post_tool.run_post_tool_hook", + side_effect=fake_run_post_tool_hook, + ): + mgr.after_tool_call( + "codelens_query", + {"file": "app.py", "workspace": enabled_workspace}, + workspace=enabled_workspace, + ) + + # The hook runs in a worker thread; wait for it. + assert done.wait(timeout=2.0), "hook never ran" + + assert captured["workspace"] == enabled_workspace + assert captured["severity_threshold"] == "high" + # Arguments are snapshotted shallowly — values must match. + assert captured["arguments"]["file"] == "app.py" + + def test_enabled_hook_passes_workspace_argument_through( + self, enabled_workspace + ): + mgr = HookManager(workspace=enabled_workspace) + + done = threading.Event() + captured = {} + + def fake_run_post_tool_hook(arguments, workspace, severity_threshold): + captured["workspace"] = workspace + done.set() + return _make_result() + + with mock.patch( + "mcp_hooks.post_tool.run_post_tool_hook", + side_effect=fake_run_post_tool_hook, + ): + mgr.after_tool_call( + "codelens_query", + {"file": "app.py"}, # no workspace key + workspace=enabled_workspace, + ) + assert done.wait(timeout=2.0) + + # The workspace passed to the hook must be the explicit one. + assert captured["workspace"] == enabled_workspace + + def test_on_complete_callback_fires_with_result(self, enabled_workspace): + """``on_complete`` lets callers wait for the hook deterministically.""" + mgr = HookManager(workspace=enabled_workspace) + + done = threading.Event() + received = {} + + def on_complete(result): + received["result"] = result + done.set() + + fake_result = _make_result(critical_count=1, message="boom") + + with mock.patch( + "mcp_hooks.post_tool.run_post_tool_hook", + return_value=fake_result, + ): + mgr.after_tool_call( + "codelens_query", + {"file": "app.py", "workspace": enabled_workspace}, + on_complete=on_complete, + ) + assert done.wait(timeout=2.0) + + assert received["result"] is fake_result + + +# ─── 5. Non-blocking behavior ───────────────────────────────────────── + + +class TestNonBlocking: + """The hook MUST be non-blocking — tool-call latency stays under budget.""" + + def test_after_tool_call_returns_immediately(self, enabled_workspace): + """Synchronous cost must be << 500 ms even when the hook is slow. + + We make the hook sleep 300 ms and verify ``after_tool_call`` returns + in well under that — proving it's truly non-blocking. + """ + mgr = HookManager(workspace=enabled_workspace) + + def slow_hook(arguments, workspace, severity_threshold): + time.sleep(0.3) + return _make_result() + + with mock.patch( + "mcp_hooks.post_tool.run_post_tool_hook", + side_effect=slow_hook, + ): + start = time.monotonic() + mgr.after_tool_call( + "codelens_query", + {"file": "app.py", "workspace": enabled_workspace}, + workspace=enabled_workspace, + ) + elapsed_ms = (time.monotonic() - start) * 1000.0 + + # The 500 ms budget is for the *tool call*; the synchronous cost of + # scheduling the hook must be a tiny fraction of that. + assert elapsed_ms < 100.0, ( + f"after_tool_call blocked for {elapsed_ms:.1f} ms; " + "expected sub-100 ms scheduling cost" + ) + + +# ─── 6. Error isolation ────────────────────────────────────────────── + + +class TestErrorIsolation: + """A crashing hook must never propagate to the MCP server.""" + + def test_hook_exception_is_swallowed(self, enabled_workspace): + """If run_post_tool_hook raises, after_tool_call must not re-raise.""" + mgr = HookManager(workspace=enabled_workspace) + + with mock.patch( + "mcp_hooks.post_tool.run_post_tool_hook", + side_effect=RuntimeError("boom"), + ): + # Must not raise. + mgr.after_tool_call( + "codelens_query", + {"file": "app.py", "workspace": enabled_workspace}, + workspace=enabled_workspace, + ) + # Give the worker thread a moment to hit the exception. + time.sleep(0.1) + + # No notification should have been queued (the hook crashed before + # producing a result). + assert mgr.drain_pending() == [] + + def test_on_complete_callback_exception_is_swallowed( + self, enabled_workspace + ): + mgr = HookManager(workspace=enabled_workspace) + done = threading.Event() + + def bad_callback(result): + raise ValueError("callback boom") + + # Patch print so we don't spam stderr in test output. + with mock.patch("builtins.print"), mock.patch( + "mcp_hooks.post_tool.run_post_tool_hook", + return_value=_make_result(), + ): + mgr.after_tool_call( + "codelens_query", + {"file": "app.py", "workspace": enabled_workspace}, + workspace=enabled_workspace, + on_complete=bad_callback, + ) + # Worker thread must survive the bad callback. + time.sleep(0.1) + # Pool is still usable: submit another task and verify it runs. + mgr._executor.submit(done.set) + assert done.wait(timeout=2.0) + + +# ─── 7. Notification buffering & delivery ──────────────────────────── + + +class TestNotificationBuffering: + """Critical/high findings are surfaced via MCP notifications.""" + + def test_critical_finding_queued_as_notification(self, enabled_workspace): + mgr = HookManager(workspace=enabled_workspace) + target_path = os.path.join(enabled_workspace, "app.py") + result = _make_result( + critical_count=2, + high_count=0, + file_path=target_path, + workspace=enabled_workspace, + ) + + with mock.patch( + "mcp_hooks.post_tool.run_post_tool_hook", + return_value=result, + ): + mgr.after_tool_call( + "codelens_query", + {"file": "app.py", "workspace": enabled_workspace}, + workspace=enabled_workspace, + ) + # Drain after a brief wait for the worker thread to finish. + time.sleep(0.1) + pending = mgr.drain_pending() + + assert len(pending) == 1 + notif = pending[0] + assert notif["method"] == "notifications/message" + assert notif["jsonrpc"] == "2.0" + assert notif["params"]["level"] == "warning" + data = notif["params"]["data"] + assert data["critical_count"] == 2 + assert data["high_count"] == 0 + assert data["tool"] == "codelens_query" + assert data["file"] == target_path + assert data["workspace"] == enabled_workspace + assert data["source"] == "codelens.post_tool_hook" + + def test_no_finding_no_notification(self, enabled_workspace): + mgr = HookManager(workspace=enabled_workspace) + # triggered=True but zero findings → no notification. + result = _make_result(critical_count=0, high_count=0) + + with mock.patch( + "mcp_hooks.post_tool.run_post_tool_hook", + return_value=result, + ): + mgr.after_tool_call( + "codelens_query", + {"file": "app.py", "workspace": enabled_workspace}, + workspace=enabled_workspace, + ) + time.sleep(0.1) + assert mgr.drain_pending() == [] + + def test_not_triggered_no_notification(self, enabled_workspace): + mgr = HookManager(workspace=enabled_workspace) + # Hook did not fire (e.g. no file in args) → no notification. + result = _make_result(triggered=False, critical_count=0, high_count=0) + + with mock.patch( + "mcp_hooks.post_tool.run_post_tool_hook", + return_value=result, + ): + mgr.after_tool_call( + "codelens_query", + {"workspace": enabled_workspace}, + workspace=enabled_workspace, + ) + time.sleep(0.1) + assert mgr.drain_pending() == [] + + def test_send_notification_callback_receives_payload( + self, enabled_workspace + ): + notifications = [] + + mgr = HookManager( + workspace=enabled_workspace, + send_notification=notifications.append, + ) + + result = _make_result(critical_count=1, high_count=2) + + with mock.patch( + "mcp_hooks.post_tool.run_post_tool_hook", + return_value=result, + ): + mgr.after_tool_call( + "codelens_query", + {"file": "app.py", "workspace": enabled_workspace}, + workspace=enabled_workspace, + ) + time.sleep(0.15) + + assert len(notifications) == 1 + # When the callback is set, nothing is buffered. + assert mgr.drain_pending() == [] + + def test_send_notification_failure_falls_back_to_buffer( + self, enabled_workspace + ): + def broken_callback(notification): + raise IOError("stdout closed") + + mgr = HookManager( + workspace=enabled_workspace, + send_notification=broken_callback, + ) + + result = _make_result(critical_count=1) + + with mock.patch("builtins.print"), mock.patch( + "mcp_hooks.post_tool.run_post_tool_hook", + return_value=result, + ): + mgr.after_tool_call( + "codelens_query", + {"file": "app.py", "workspace": enabled_workspace}, + workspace=enabled_workspace, + ) + time.sleep(0.15) + + # The broken callback must not lose the notification — it falls + # back to the buffer so it can be attached to the next response. + pending = mgr.drain_pending() + assert len(pending) == 1 + + +# ─── 8. extract_file_path unit tests ───────────────────────────────── + + +class TestExtractFilePath: + """The hook's argument inspection must resolve file paths robustly.""" + + def test_absolute_path_returned_as_is(self, tmp_workspace): + target = os.path.join(tmp_workspace, "app.py") + with open(target, "w") as f: + f.write("# test") + result = extract_file_path({"file": target}, tmp_workspace) + assert result == os.path.normpath(target) + + def test_relative_path_resolved_against_workspace(self, tmp_workspace): + target = os.path.join(tmp_workspace, "app.py") + with open(target, "w") as f: + f.write("# test") + result = extract_file_path({"file": "app.py"}, tmp_workspace) + assert result == os.path.normpath(target) + + def test_path_key_also_recognized(self, tmp_workspace): + target = os.path.join(tmp_workspace, "app.py") + with open(target, "w") as f: + f.write("# test") + result = extract_file_path({"path": "app.py"}, tmp_workspace) + assert result == os.path.normpath(target) + + def test_nonexistent_file_returns_none(self, tmp_workspace): + result = extract_file_path( + {"file": "does_not_exist.py"}, tmp_workspace + ) + assert result is None + + def test_no_file_key_returns_none(self, tmp_workspace): + result = extract_file_path({"name": "main"}, tmp_workspace) + assert result is None + + def test_non_dict_arguments_returns_none(self, tmp_workspace): + assert extract_file_path(None, tmp_workspace) is None # type: ignore[arg-type] + assert extract_file_path("not a dict", tmp_workspace) is None # type: ignore[arg-type] + + +# ─── 9. run_post_tool_hook integration smoke test ──────────────────── + + +class TestRunPostToolHookSmoke: + """End-to-end smoke test of run_post_tool_hook on a real (tiny) workspace. + + The actual scan engines are heavy; this test just verifies that the + public entry point returns a :class:`PostToolHookResult` (never raises) + on a workspace that has no findings. + """ + + def test_returns_result_on_empty_workspace(self, tmp_workspace): + # Create a tiny Python file with no smells. + with open(os.path.join(tmp_workspace, "empty.py"), "w") as f: + f.write("# nothing to see here\n") + + arguments = {"file": "empty.py", "workspace": tmp_workspace} + # Run the hook directly (synchronous — no executor involved). + result = run_post_tool_hook(arguments, tmp_workspace, "high") + + assert isinstance(result, PostToolHookResult) + assert result.workspace == tmp_workspace + # The file exists, so the hook should have triggered. + assert result.triggered is True + assert result.file_path is not None + # Empty file has no findings. + assert result.critical_count == 0 + assert result.high_count == 0 + assert result.message == "" + # No errors surfaced — the hook must complete cleanly. + assert result.error is None + # Performance budget: < 500 ms (we use a generous 5 s ceiling so + # slow CI machines don't flake; the real ceiling is in the spec). + assert result.elapsed_ms < 5000.0 + + def test_returns_result_with_no_file_in_args(self, tmp_workspace): + # No file argument → hook returns early, triggered=False. + result = run_post_tool_hook({"name": "main"}, tmp_workspace, "high") + assert isinstance(result, PostToolHookResult) + assert result.triggered is False + assert result.critical_count == 0 + assert result.high_count == 0 + + def test_returns_result_on_invalid_workspace(self): + result = run_post_tool_hook( + {"file": "app.py"}, "/does/not/exist", "high" + ) + assert result.triggered is False + # Invalid workspace is surfaced as an error string, never raised. + assert result.error is not None + + +# ─── 10. MCPServer integration ─────────────────────────────────────── + + +class TestMCPServerIntegration: + """Verify MCPServer wires the HookManager into tools/call correctly.""" + + def test_mcp_server_creates_hook_manager_on_first_call( + self, enabled_workspace + ): + """The HookManager is constructed lazily, bound to the workspace.""" + server = MCPServer() + assert server._hook_manager is None + + # Force the manager to be created by calling the helper. + manager = server._get_hook_manager(enabled_workspace) + assert manager is not None + assert manager.workspace == os.path.abspath(enabled_workspace) + assert manager.is_enabled("post_tool") is True + # Cleanup so the worker pool doesn't leak across tests. + server._hook_manager.shutdown() + + def test_disabled_hook_no_side_effect_through_mcp( + self, disabled_workspace + ): + """End-to-end: a tool call on a disabled-hook workspace must NOT + invoke ``run_post_tool_hook`` (the issue #47 contract test, but + through the full MCPServer._handle_tools_call path).""" + server = MCPServer() + # Pre-seed the cache for the query tool so we don't need a real + # registry to exist. The scan tool is excluded from caching, so we + # invoke codelens_scan instead — it short-circuits in cmd_scan + # when the workspace has no source files. + with mock.patch( + "mcp_hooks.post_tool.run_post_tool_hook" + ) as mock_hook: + response = server._handle_tools_call( + { + "name": "codelens_scan", + "arguments": {"workspace": disabled_workspace}, + } + ) + # Tool call must succeed. + assert response.get("isError") is False + # Hook must not have been called. + time.sleep(0.05) + mock_hook.assert_not_called() + # No _hooks field on the response. + assert "_hooks" not in response + + server._shutdown() + + def test_enabled_hook_invoked_through_mcp(self, enabled_workspace): + """End-to-end: a successful tool call on an enabled-hook workspace + MUST invoke ``run_post_tool_hook`` exactly once.""" + server = MCPServer() + done = threading.Event() + + def fake_hook(arguments, workspace, severity_threshold): + done.set() + return _make_result() + + with mock.patch( + "mcp_hooks.post_tool.run_post_tool_hook", + side_effect=fake_hook, + ): + response = server._handle_tools_call( + { + "name": "codelens_scan", + "arguments": {"workspace": enabled_workspace}, + } + ) + assert response.get("isError") is False + assert done.wait(timeout=2.0), "hook never fired via MCP" + + server._shutdown() + + def test_pending_notifications_attached_to_next_response( + self, enabled_workspace + ): + """A queued hook notification must be attached to the next + ``tools/call`` response as a ``_hooks`` field.""" + server = MCPServer() + # Manually push a fake notification into the manager's queue. + manager = server._get_hook_manager(enabled_workspace) + with manager._lock: + manager._pending.append( + { + "jsonrpc": "2.0", + "method": "notifications/message", + "params": {"level": "warning", "data": {"fake": True}}, + } + ) + + # The next tool call should drain the queue into the response. + response = server._handle_tools_call( + { + "name": "codelens_scan", + "arguments": {"workspace": enabled_workspace}, + } + ) + assert "_hooks" in response + assert len(response["_hooks"]) == 1 + assert response["_hooks"][0]["params"]["data"]["fake"] is True + + # The next call must NOT have any _hooks (queue was drained). + response2 = server._handle_tools_call( + { + "name": "codelens_scan", + "arguments": {"workspace": enabled_workspace}, + } + ) + assert "_hooks" not in response2 + + server._shutdown() + + def test_hook_failure_does_not_break_tool_response( + self, enabled_workspace + ): + """If the HookManager itself raises, the tool response is unaffected.""" + server = MCPServer() + + # Force the manager to throw on after_tool_call by replacing it + # with a stub that raises. + class BrokenManager: + def after_tool_call(self, *a, **kw): + raise RuntimeError("hook subsystem broken") + + def drain_pending(self): + return [] + + server._hook_manager = BrokenManager() + + response = server._handle_tools_call( + { + "name": "codelens_scan", + "arguments": {"workspace": enabled_workspace}, + } + ) + # Tool call must still succeed despite the broken hook subsystem. + assert response.get("isError") is False + # The broken hook subsystem is caught and swallowed. + assert "_hooks" not in response + + server._shutdown()