diff --git a/CHANGELOG.md b/CHANGELOG.md index 28ebcc1e..6e6a0576 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,9 +2,18 @@ Status of the `main` branch. Changes prior to the next official version change will appear here. +* General: + - Add Grok Build support (context `grok`, setup CLI, hooks) + * Language Servers: - Add `python_basedpyright` as an alternative Python language server +* Hooks: + - Add `serena-hooks --client=grok`, including Grok-native PreToolUse allow/deny output. + - PreToolUse remind hook: coerce non-string shell command values instead of failing, and recognize + `target_file`/`targetFile` file-path keys (shared payload parsing, applies to all hook clients). + + # v1.6.1 (2026-07-21) * General: diff --git a/docs/02-usage/030_clients.md b/docs/02-usage/030_clients.md index d8e30397..3a7f2980 100644 --- a/docs/02-usage/030_clients.md +++ b/docs/02-usage/030_clients.md @@ -380,6 +380,88 @@ The hooks will: The `PreToolUse` matcher is intentionally restricted to `Bash`. The Serena reminder hook for Codex tracks shell-based grep and code-file reads, so running it for every tool call is unnecessary. +## Grok + +Serena provides native support for xAI's Grok Build CLI. To set up the Serena MCP server for Grok, +simply run: + + serena setup grok + +### Manual Setup + +**Global Configuration**. To add the Serena MCP server for all your projects, use Grok's user-level configuration and the `--project-from-cwd` flag: + +```bash +grok mcp add --scope user serena -- serena start-mcp-server --context=grok --project-from-cwd +``` + +Alternatively, add the following to `~/.grok/config.toml`: + +```toml +[mcp_servers.serena] +command = "serena" +args = ["start-mcp-server", "--project-from-cwd", "--context=grok"] +``` + +**Project-Level Configuration**. To add the Serena MCP server for a single project only: + +```bash +grok mcp add --scope project serena -- serena start-mcp-server --context=grok --project "$(pwd)" +``` + +**Verification.** +Run `grok inspect` and verify that Serena is listed as an MCP server. You can also use Grok's `/mcps` +modal to inspect, refresh, enable, or disable configured MCP servers. + +Grok can also load existing Claude Code MCP configuration. If you previously ran `serena setup claude-code`, +Serena may already appear in Grok, but it will use the `claude-code` context instead of the dedicated `grok` context. + +### Hooks + +Grok supports lifecycle hooks; see Grok's bundled hooks documentation for details. To enable Serena's hooks +for Grok globally, create `~/.grok/hooks/serena-hooks.json` with the following content. For a single project, +create `.grok/hooks/serena-hooks.json` in that project instead; note that Grok loads project-scoped hooks +only after you have trusted the project for hook execution (via Grok's `/hooks-trust` command). + +```json +{ + "hooks": { + "PreToolUse": [ + { + "matcher": "grep|read_file|run_terminal_command", + "hooks": [ + { + "type": "command", + "command": "serena-hooks remind --client=grok", + "timeout": 5 + } + ] + } + ], + "Stop": [ + { + "hooks": [ + { + "type": "command", + "command": "serena-hooks cleanup --client=grok", + "timeout": 5 + } + ] + } + ] + } +} +``` + +The hooks will: + +- **`remind`**: Nudge the agent to use Serena's symbolic tools when it makes too many consecutive + code-search or code-file-read calls without using Serena tools in between. +- **`cleanup`**: Clean up hook session data when the agent turn ends. + +Grok ignores stdout from passive hooks such as `SessionStart`, so the Grok hook setup intentionally +uses only `PreToolUse` for reminders and `Stop` for cleanup. + ## Claude Desktop On Windows and macOS, there are official [Claude Desktop applications by Anthropic](https://claude.ai/download); for Linux, there is an [open-source diff --git a/docs/02-usage/050_configuration.md b/docs/02-usage/050_configuration.md index ab9d6611..b47e4d46 100644 --- a/docs/02-usage/050_configuration.md +++ b/docs/02-usage/050_configuration.md @@ -73,6 +73,7 @@ Serena comes with pre-defined contexts: The full set of Serena's tools is provided, as the application is assumed to have no prior coding-specific capabilities. * `claude-code`: Optimized for use with Claude Code, it disables tools that would duplicate Claude Code's built-in capabilities. * `codex`: Optimized for use with OpenAI Codex. +* `grok`: Optimized for use with xAI's Grok Build CLI. * `ide`: Generic context for IDE assistants/coding agents, e.g. VSCode, Cursor, or Cline, focusing on augmenting existing capabilities. Basic file operations and shell execution are assumed to be handled by the assistant's own capabilities. * `agent`: Designed for scenarios where Serena acts as a more autonomous agent, for example, when used with Agno. @@ -81,7 +82,7 @@ Choose the context that best matches the type of integration you are using. Find the concrete definitions of the above contexts [here](https://github.com/oraios/serena/tree/main/src/serena/resources/config/contexts). -Note that the contexts `ide` and `claude-code` are **single-project contexts** (defining `single_project: true`). +Note that the contexts `ide`, `claude-code`, and `grok` are **single-project contexts** (defining `single_project: true`). For such contexts, if a project is provided at startup, the set of tools is limited to those required by the project's concrete configuration, and other tools are excluded completely, allowing the set of tools to be minimal. Tools explicitly disabled by the project will not be available at all. Since changing the active project diff --git a/scripts/live_test_grok.py b/scripts/live_test_grok.py new file mode 100644 index 00000000..505a88d7 --- /dev/null +++ b/scripts/live_test_grok.py @@ -0,0 +1,1194 @@ +""" +Live smoke test for Serena's Grok Build integration. + +Verifies that the Grok support (``serena setup grok``, the ``grok`` context and the native +``serena-hooks --client=grok`` protocol) works against a *real* ``grok`` CLI installation — +exercising the exact code paths that the unit tests cover with mocks +(``test/serena/config/test_client_setup.py``, ``test/serena/test_hooks.py``, ...). + +The test is designed to be safe and free: + +* **Zero inference cost.** No Grok session is ever started and no prompt is sent; only + configuration, discovery, stdio hook invocations and raw MCP handshakes (including one + read-only Serena tool call) are used. +* **State-preserving.** The Grok user configuration is backed up before the first mutation, + every ``grok mcp add`` is paired with a removal, and the baseline is verified (and restored + from backup if necessary) at the end — also on abort. Hook state is written to an isolated, + temporary ``SERENA_HOME``. The script refuses to run if a ``serena`` MCP server is already + registered in Grok, so it never clobbers a real setup. +* **Credential-safe.** The config backup (which may carry tokens) is written 0600 inside a private, + owner-only work directory (a fresh ``mkdtemp`` by default; a supplied ``--work-dir`` is validated + as non-symlink, owner-owned and not group/world-accessible) and is deleted once the baseline is + confirmed intact. ``--hooks-only`` never mutates the config and therefore never backs it up. + +Per-check evidence files and a Markdown report are written to the work directory (printed at +startup; default: a fresh private 0700 directory created via ``mkdtemp`` under the system tmp, +prefix ``serena-grok-live-``). The exit code is 0 iff no check FAILed. + +Usage:: + + uv run python scripts/live_test_grok.py # full run + uv run python scripts/live_test_grok.py --hooks-only # pure-local checks only (no Grok config changes) + uv run python scripts/live_test_grok.py --skip-unit # skip the pytest smoke run + uv run python scripts/live_test_grok.py --help # all options +""" + +import argparse +import json +import os +import queue +import shutil +import subprocess +import sys +import tempfile +import threading +import time +from dataclasses import dataclass +from datetime import datetime +from enum import Enum +from pathlib import Path +from typing import IO, Optional + +CONTEXT_EXCLUDED_TOOLS = ("create_text_file", "execute_shell_command", "find_file", "list_dir", "read_file", "search_for_pattern") +"""tools excluded by the ``grok`` context (mirrors ``src/serena/resources/config/contexts/grok.yml``)""" + +REQUIRED_SYMBOLIC_TOOLS = ("find_symbol", "get_symbols_overview", "replace_symbol_body", "find_referencing_symbols") +"""symbolic tools that must be exposed by the MCP server in project mode""" + +UNIT_TEST_FILES = ( + "test/serena/test_hooks.py", + "test/serena/test_cli_setup.py", + "test/serena/config/test_client_setup.py", + "test/serena/config/test_context_mode.py", + "test/serena/config/test_grok_docs.py", +) +"""unit-test files whose mocked assumptions this live test validates""" + +LIVE_MCP_SERVER_NAME = "serena-live" +"""name of the temporary MCP server entry registered for connectivity checks""" + +STRUCTURED_OUTPUT_PROBE_TOOL = "initial_instructions" +"""read-only project-mode tool (returns a plain string) used to probe the structured-output wire shape""" + +HOOKS_JSON = { + "hooks": { + "PreToolUse": [ + { + "matcher": "grep|read_file|run_terminal_command", + "hooks": [{"type": "command", "command": "serena-hooks remind --client=grok", "timeout": 5}], + } + ], + "Stop": [{"hooks": [{"type": "command", "command": "serena-hooks cleanup --client=grok", "timeout": 5}]}], + } +} +"""the hooks configuration recommended in docs/02-usage/030_clients.md (Grok section)""" + + +class Status(Enum): + """The outcome category of a single live check.""" + + PASS = "PASS" + FAIL = "FAIL" + WARN = "WARN" + INFO = "INFO" + SKIP = "SKIP" + + +@dataclass +class CheckResult: + """The recorded outcome of a single live check.""" + + check_id: str + status: Status + note: str + + +@dataclass +class CommandResult: + """The captured outcome of an executed subprocess.""" + + return_code: Optional[int] + stdout: str + stderr: str + timed_out: bool = False + + @property + def output(self) -> str: + return self.stdout + ("\n" + self.stderr if self.stderr else "") + + +class AbortError(Exception): + """Raised when the environment makes continuing pointless or unsafe.""" + + +def run_command( + argv: list[str], + timeout: float = 60, + cwd: Optional[Path] = None, + env_overlay: Optional[dict[str, str]] = None, + stdin_text: Optional[str] = None, +) -> CommandResult: + """ + Runs a subprocess without a shell, capturing output and never raising on non-zero exit. + + :param argv: the command and its arguments + :param timeout: seconds after which the process is killed + :param cwd: the working directory for the process + :param env_overlay: environment variables to add on top of the current environment + :param stdin_text: text to pass on standard input + :return: the captured result; on timeout, ``timed_out`` is set and ``return_code`` is None + """ + env = None + if env_overlay is not None: + env = {**os.environ, **env_overlay} + try: + completed = subprocess.run(argv, capture_output=True, text=True, timeout=timeout, cwd=cwd, env=env, input=stdin_text, check=False) + return CommandResult(completed.returncode, completed.stdout, completed.stderr) + except subprocess.TimeoutExpired as e: + stdout = e.stdout.decode() if isinstance(e.stdout, bytes) else (e.stdout or "") + stderr = e.stderr.decode() if isinstance(e.stderr, bytes) else (e.stderr or "") + return CommandResult(None, stdout, stderr, timed_out=True) + except FileNotFoundError: + return CommandResult(127, "", f"executable not found: {argv[0]}") + + +class McpStdioProbe: + """ + A minimal MCP client that spawns a stdio server, performs the initialize handshake and + lists the exposed tools. Uses a reader thread (rather than ``select``) for portability. + """ + + def __init__(self, argv: list[str], cwd: Path, stderr_log: IO[str]) -> None: + """ + :param argv: the server command line + :param cwd: the working directory for the server process + :param stderr_log: an open file receiving the server's stderr (Serena logs to stderr) + """ + self._process = subprocess.Popen( + argv, stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=stderr_log, text=True, bufsize=1, cwd=cwd + ) + self._messages: queue.Queue[dict] = queue.Queue() + self._reader = threading.Thread(target=self._read_loop, daemon=True) + self._reader.start() + + def _read_loop(self) -> None: + assert self._process.stdout is not None + for line in self._process.stdout: + line = line.strip() + if not line: + continue + try: + self._messages.put(json.loads(line)) + except json.JSONDecodeError: + continue + + def _send(self, message: dict) -> None: + assert self._process.stdin is not None + self._process.stdin.write(json.dumps(message) + "\n") + self._process.stdin.flush() + + def request(self, request_id: int, method: str, params: dict, timeout: float) -> dict: + """ + Sends a JSON-RPC request and waits for the response with the matching id. + + :param request_id: the JSON-RPC id to send and match on + :param method: the JSON-RPC method name + :param params: the JSON-RPC params object + :param timeout: seconds to wait before giving up + :return: the raw response message + :raises AbortError: if the server exits prematurely or the response does not arrive in time + """ + self._send({"jsonrpc": "2.0", "id": request_id, "method": method, "params": params}) + deadline = time.monotonic() + timeout + while time.monotonic() < deadline: + if self._process.poll() is not None: + raise AbortError(f"MCP server exited prematurely with code {self._process.returncode}") + try: + message = self._messages.get(timeout=1.0) + except queue.Empty: + continue + if message.get("id") == request_id: + return message + raise AbortError(f"timed out waiting for MCP response to '{method}'") + + def notify(self, method: str) -> None: + self._send({"jsonrpc": "2.0", "method": method, "params": {}}) + + def close(self) -> None: + self._process.terminate() + try: + self._process.wait(timeout=10) + except subprocess.TimeoutExpired: + self._process.kill() + + +@dataclass +class LiveTestConfig: + """The resolved configuration of a live test run.""" + + repo_root: Path + serena_bin: Path + serena_hooks_bin: Path + python_bin: Path + grok_bin: str + grok_config_path: Path + work_dir: Path + hooks_only: bool = False + skip_unit: bool = False + grok_config_explicit: bool = False + """whether --grok-config was supplied on the command line (enables the config-path/CLI consistency check)""" + + @property + def evidence_dir(self) -> Path: + return self.work_dir / "evidence" + + @property + def hook_home(self) -> Path: + """An isolated ``SERENA_HOME`` so hook state never touches the user's real one""" + return self.work_dir / "serena-home" + + @property + def config_backup_path(self) -> Path: + return self.work_dir / "config.toml.bak" + + @staticmethod + def _venv_executable(repo_root: Path, name: str) -> Path: + if sys.platform == "win32": + return repo_root / ".venv" / "Scripts" / f"{name}.exe" + return repo_root / ".venv" / "bin" / name + + @classmethod + def from_args(cls, args: argparse.Namespace) -> "LiveTestConfig": + """ + :param args: the parsed command-line arguments + :return: the resolved configuration, with defaults derived from the repository layout + :raises AbortError: if a required executable cannot be resolved + """ + repo_root = Path(args.repo_root).resolve() if args.repo_root else Path(__file__).resolve().parents[1] + + # resolve the executables under test, defaulting to the repository's venv + serena_bin = Path(args.serena_bin) if args.serena_bin else cls._venv_executable(repo_root, "serena") + serena_hooks_bin = Path(args.serena_hooks_bin) if args.serena_hooks_bin else cls._venv_executable(repo_root, "serena-hooks") + python_bin = cls._venv_executable(repo_root, "python") + for executable in (serena_bin, serena_hooks_bin, python_bin): + if not executable.exists(): + raise AbortError(f"executable not found: {executable} — run 'uv sync' first (or pass --serena-bin/--serena-hooks-bin)") + + # resolve the grok CLI and its user-level configuration file + grok_bin = args.grok_bin or shutil.which("grok") or "" + if not grok_bin: + raise AbortError("the 'grok' CLI was not found on PATH (override with --grok-bin)") + grok_config_path = Path(args.grok_config) if args.grok_config else Path.home() / ".grok" / "config.toml" + + # Default to a private, unpredictable work dir (0700, owner-only): it holds a backup of the + # user's grok config, which may carry credentials. A fixed shared-tmp path would expose those + # to other local users and invites symlink/pre-creation attacks. + work_dir = Path(args.work_dir) if args.work_dir else Path(tempfile.mkdtemp(prefix="serena-grok-live-")) + return cls( + repo_root=repo_root, + serena_bin=serena_bin, + serena_hooks_bin=serena_hooks_bin, + python_bin=python_bin, + grok_bin=grok_bin, + grok_config_path=grok_config_path, + work_dir=work_dir, + hooks_only=args.hooks_only, + skip_unit=args.skip_unit, + grok_config_explicit=bool(args.grok_config), + ) + + +class GrokLiveTest: + """Orchestrates the live checks, evidence capture, cleanup and reporting.""" + + def __init__(self, config: LiveTestConfig) -> None: + self.config = config + self.results: list[CheckResult] = [] + self.findings: list[str] = [] + self._registered_server_names: list[str] = [] + self._created_global_hooks_file: Optional[Path] = None + self._cleanup_done = False + self._baseline_mcp_list = "" + self._environment_summary = "" + + # ------------------------------------------------------------------ infrastructure + + def _record(self, check_id: str, status: Status, note: str) -> None: + self.results.append(CheckResult(check_id, status, note)) + print(f"[{status.value}] {check_id} — {note}") + + def _finding(self, text: str) -> None: + self.findings.append(text) + print(f"FINDING: {text}") + + @staticmethod + def _section(title: str) -> None: + print(f"\n=== {title} ===") + + def _write_evidence(self, name: str, content: str) -> None: + # Evidence can embed other MCP servers' env/args (a common place for tokens); keep it owner-only. + path = self.config.evidence_dir / name + path.write_text(content, encoding="utf-8") + os.chmod(path, 0o600) + + def _open_evidence(self, name: str) -> IO[str]: + """Open an evidence file for writing with owner-only (0600) permissions (e.g. server stderr logs).""" + path = self.config.evidence_dir / name + path.touch(mode=0o600) + return open(path, "w", encoding="utf-8") + + def _grok(self, *args: str, timeout: float = 60) -> CommandResult: + return run_command([self.config.grok_bin, *args], timeout=timeout) + + def _run_hook(self, subcommand: str, payload: dict) -> CommandResult: + argv = [str(self.config.serena_hooks_bin), subcommand, "--client", "grok"] + return run_command(argv, timeout=30, env_overlay={"SERENA_HOME": str(self.config.hook_home)}, stdin_text=json.dumps(payload)) + + @staticmethod + def _pre_tool_use_payload(session_id: str, tool_name: str, tool_input: dict, permission_mode: str = "bypassPermissions") -> dict: + """:return: a PreToolUse payload in Grok's camelCase envelope (mirrors ``test_hooks.py::_grok_input``)""" + return { + "hookEventName": "pre_tool_use", + "sessionId": session_id, + "toolName": tool_name, + "toolInput": tool_input, + "toolInputTruncated": False, + "permissionMode": permission_mode, + } + + def _read_grok_mcp_server_section(self, server_name: str) -> dict: + """:return: the ``[mcp_servers.]`` table from the Grok configuration, or an empty dict""" + import tomllib + + try: + with open(self.config.grok_config_path, "rb") as f: + config = tomllib.load(f) + return config.get("mcp_servers", {}).get(server_name, {}) + except (OSError, tomllib.TOMLDecodeError): + return {} + + def _verify_config_path_matches_cli(self, mcp_list_output: str) -> None: + """Abort if ``--grok-config`` points at a file the grok CLI is not actually using. + + The grok CLI offers no alternate-config option (``grok mcp`` knows only the user/project + scopes), so ``--grok-config`` exists solely for nonstandard Grok homes. If it named a file + the CLI does not use, every file-level safety measure in this script (backup, restore, + baseline byte-comparison) would silently protect the wrong file while ``grok mcp + add/remove`` mutates the real configuration. Cross-check: the ``[mcp_servers.*]`` table + names in the file must equal the server names reported by ``grok mcp list``. + + :param mcp_list_output: the raw output of ``grok mcp list`` captured for the baseline + :raises AbortError: on any mismatch, before any mutation has occurred + """ + if not self.config.grok_config_explicit: + return + import tomllib + + try: + with open(self.config.grok_config_path, "rb") as f: + file_servers = set(tomllib.load(f).get("mcp_servers", {}).keys()) + except (OSError, tomllib.TOMLDecodeError): + file_servers = set() + import re + + cli_servers = set() + for line in mcp_list_output.splitlines(): + m = re.match(r"^\s*([\w-]+):\s", line) + if m: + cli_servers.add(m.group(1)) + if file_servers != cli_servers: + self._record("P3", Status.FAIL, "--grok-config does not match the configuration the grok CLI actually uses") + raise AbortError( + f"--grok-config {self.config.grok_config_path} is not the config the grok CLI uses " + f"(file declares mcp_servers {sorted(file_servers)}, 'grok mcp list' reports {sorted(cli_servers)}). " + "This flag is for nonstandard Grok homes only — it cannot isolate the test from the real " + "configuration, because 'grok mcp' has no alternate-config option. Aborting before any mutation." + ) + + def _insert_startup_timeout_sec(self, seconds: int) -> bool: + """Add ``startup_timeout_sec`` to the live server's ``[mcp_servers.]`` table, located structurally. + + The table is confirmed to exist (and to lack the key) by parsing the config with ``tomllib``, then the + key is inserted immediately after the exact table-header line. This avoids a blind substring replace, + which could match ``[mcp_servers.]`` inside a comment or string value elsewhere in the user's + config and corrupt it. The rest of the file (comments, formatting) is preserved by editing in place. + + :return: True if the key was inserted, False if the table is absent, already has the key, or is unparsable. + """ + section = f"[mcp_servers.{LIVE_MCP_SERVER_NAME}]" + existing = self._read_grok_mcp_server_section(LIVE_MCP_SERVER_NAME) + if not existing or "startup_timeout_sec" in existing: + return False + lines = self.config.grok_config_path.read_text(encoding="utf-8").splitlines() + for i, line in enumerate(lines): + if line.strip() == section: + lines.insert(i + 1, f"startup_timeout_sec = {seconds}") + self.config.grok_config_path.write_text("\n".join(lines) + "\n", encoding="utf-8") + return True + return False + + def _cleanup_grok_state(self) -> None: + """Removes every Grok-side artifact this run created (idempotent, best-effort).""" + for server_name in list(self._registered_server_names): + self._grok("mcp", "remove", "--scope", "user", server_name) + self._registered_server_names.remove(server_name) + if self._created_global_hooks_file is not None: + self._created_global_hooks_file.unlink(missing_ok=True) + try: + self._created_global_hooks_file.parent.rmdir() + except OSError: + pass + self._created_global_hooks_file = None + + # ------------------------------------------------------------------ phase P: preflight + + def check_p1_environment(self) -> None: + self._section("P1 — repository, binaries, local-source resolution") + + # gather the facts that every later check relies on + branch = run_command(["git", "rev-parse", "--abbrev-ref", "HEAD"], cwd=self.config.repo_root).stdout.strip() + commit = run_command(["git", "log", "--oneline", "-1"], cwd=self.config.repo_root).stdout.strip() + grok_version = self._grok("--version").stdout.strip() + module_path = run_command([str(self.config.python_bin), "-c", "import serena; print(serena.__file__)"]).stdout.strip() + setup_help = run_command([str(self.config.serena_bin), "setup", "--help"]).output + remind_help = run_command([str(self.config.serena_hooks_bin), "remind", "--help"]).output + self._environment_summary = f"branch {branch} ({commit}); {grok_version}" + + # validate that the grok CLI works and the serena under test actually has Grok support + problems = [] + if not grok_version.lower().startswith("grok "): + problems.append(f"grok CLI not functional: {grok_version!r}") + expected_module = self.config.repo_root / "src" / "serena" / "__init__.py" + if Path(module_path) != expected_module: + problems.append(f"serena resolves to {module_path}, not the repository source ({expected_module}) — run 'uv sync'") + if "grok" not in setup_help: + problems.append("'serena setup' does not offer the grok client — is Grok support present on this branch?") + if "grok" not in remind_help: + problems.append("'serena-hooks remind --client' does not offer grok") + + if problems: + self._record("P1", Status.FAIL, "; ".join(problems)) + raise AbortError("environment not usable for the live test") + self._record("P1", Status.PASS, self._environment_summary) + + @staticmethod + def _require_private_dir(path: Path) -> None: + """Ensure ``path`` is a real, owner-owned, non-group/world-accessible directory. + + Guards a user-supplied --work-dir before any credential-bearing backup is written into it, + and rejects a symlink or another user's pre-created directory. + """ + info = path.lstat() + import stat as _stat + + if _stat.S_ISLNK(info.st_mode): + raise AbortError(f"work dir {path} is a symlink; refusing to write a config backup through it") + if not path.is_dir(): + raise AbortError(f"work dir {path} exists but is not a directory") + if info.st_uid != os.getuid(): + raise AbortError(f"work dir {path} is not owned by the current user; refusing to use it") + if info.st_mode & (_stat.S_IRWXG | _stat.S_IRWXO): + raise AbortError(f"work dir {path} is group/world-accessible; run 'chmod 700 {path}' or omit --work-dir") + + def _backup_config_private(self) -> None: + """Copy the grok config into the work dir with 0600 from creation (no world-readable window).""" + dst = self.config.config_backup_path + content = self.config.grok_config_path.read_bytes() if self.config.grok_config_path.exists() else b"" + fd = os.open(str(dst), os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600) + with os.fdopen(fd, "wb") as out: + out.write(content) + + def check_p3_baseline(self) -> None: + self._section("P3 — work directory, baseline snapshot, config backup") + + # The work dir holds a backup of the possibly-credential-bearing grok config, so it must be + # private. A default work dir is a fresh 0700 mkdtemp; a user-supplied one is validated. + if self.config.work_dir.exists(): + self._require_private_dir(self.config.work_dir) + else: + self.config.work_dir.mkdir(mode=0o700, parents=True) + self.config.evidence_dir.mkdir(mode=0o700, parents=True, exist_ok=True) + self.config.hook_home.mkdir(mode=0o700, parents=True, exist_ok=True) + + # Back up the grok configuration before anything may mutate it — but only when we might mutate + # it. --hooks-only never touches the config, so it must not copy the credentials at all. + if self.config.hooks_only: + self._baseline_mcp_list = self._grok("mcp", "list").output + self._write_evidence("P3-baseline-mcp-list.txt", self._baseline_mcp_list) + self._record("P3", Status.PASS, "hooks-only: baseline captured, config left untouched (no backup)") + return + + self._backup_config_private() + + # snapshot the MCP baseline and refuse to run against a pre-existing serena registration + baseline = self._grok("mcp", "list") + self._baseline_mcp_list = baseline.output + self._write_evidence("P3-baseline-mcp-list.txt", self._baseline_mcp_list) + print(self._baseline_mcp_list.strip()) + self._verify_config_path_matches_cli(self._baseline_mcp_list) + if "serena" in self._baseline_mcp_list: + self._record("P3", Status.FAIL, "a serena MCP entry is already registered in Grok — refusing to touch it") + raise AbortError( + "remove it first if it is a leftover: grok mcp remove --scope user serena " + f"(and/or {LIVE_MCP_SERVER_NAME}); if it is your real setup, run with --hooks-only instead" + ) + self._record("P3", Status.PASS, f"baseline captured, config backed up to {self.config.config_backup_path}") + + def check_p2_unit_smoke(self) -> None: + self._section("P2 — unit-test smoke (anchors live results to the mocked assumptions)") + if self.config.skip_unit: + self._record("P2", Status.SKIP, "--skip-unit") + return + + result = run_command([str(self.config.python_bin), "-m", "pytest", *UNIT_TEST_FILES, "-q"], timeout=900, cwd=self.config.repo_root) + agent_result = run_command( + [str(self.config.python_bin), "-m", "pytest", "test/serena/test_serena_agent.py", "-k", "grok", "-q"], + timeout=900, + cwd=self.config.repo_root, + ) + self._write_evidence("P2-pytest.txt", result.output + "\n" + agent_result.output) + if result.return_code == 0 and agent_result.return_code == 0: + self._record("P2", Status.PASS, "all Grok-related unit tests pass") + else: + self._record("P2", Status.FAIL, "unit tests failed — live results would be meaningless (see evidence/P2-pytest.txt)") + raise AbortError("fix the unit tests before live-testing") + + def check_p4_context_loads(self) -> None: + self._section("P4 — grok context loads from local source") + code = ( + "from serena.config.context_mode import SerenaAgentContext\n" + "ctx = SerenaAgentContext.from_name('grok')\n" + "print(ctx.name); print(ctx.single_project); print(','.join(sorted(ctx.excluded_tools)))" + ) + result = run_command([str(self.config.python_bin), "-c", code]) + self._write_evidence("P4-context.txt", result.output) + lines = result.stdout.strip().splitlines() + expected = [ + "grok", + "True", + ",".join(sorted(CONTEXT_EXCLUDED_TOOLS)), + ] + if result.return_code == 0 and lines == expected: + self._record("P4", Status.PASS, "name, single_project and tool exclusions as expected") + else: + self._record("P4", Status.FAIL, "context mismatch (see evidence/P4-context.txt)") + + # ------------------------------------------------------------------ phase H: hook protocol + + def _check_remind_burst( + self, check_id: str, title: str, deny_on_call: int, reason_fragment: str, calls: list[tuple[str, dict]] + ) -> None: + """ + Feeds a burst of PreToolUse payloads to the remind hook and asserts that only the + ``deny_on_call``-th call produces a native Grok deny containing ``reason_fragment``. + """ + self._section(f"{check_id} — {title}") + evidence_lines = [] + problem = None + for index, (tool_name, tool_input) in enumerate(calls, start=1): + result = self._run_hook("remind", self._pre_tool_use_payload(f"live-{check_id}", tool_name, tool_input)) + evidence_lines.append(f"--- call {index} (tool={tool_name}) rc={result.return_code} ---\n{result.stdout}") + output = result.stdout.strip() + + # every call must exit cleanly; only the final call may (and must) produce a deny + if result.return_code != 0: + problem = f"call {index} exited {result.return_code}" + elif index < deny_on_call and output: + problem = f"premature output on call {index}" + elif index == deny_on_call: + problem = self._validate_native_deny(output, reason_fragment) + if problem: + break + self._write_evidence(f"{check_id}.txt", "\n".join(evidence_lines)) + if problem is None: + self._record(check_id, Status.PASS, f"deny on call {deny_on_call} with expected native shape") + else: + self._record(check_id, Status.FAIL, f"{problem} (see evidence/{check_id}.txt)") + + @staticmethod + def _validate_native_deny(output: str, reason_fragment: str) -> Optional[str]: + """:return: a problem description if ``output`` is not a native Grok deny containing ``reason_fragment``, else None""" + try: + decision = json.loads(output) + except json.JSONDecodeError: + return f"deny output is not valid JSON: {output!r}" + if decision.get("decision") != "deny": + return f"expected a deny, got {output!r}" + if reason_fragment not in decision.get("reason", ""): + return f"deny reason lacks {reason_fragment!r}" + if "hookSpecificOutput" in decision: + return "Claude-shaped output leaked into the Grok format" + return None + + def check_h_hook_protocol(self) -> None: + grep_call = ("grep", {"pattern": "foo", "path": "."}) + read_call = ("read_file", {"target_file": "src/foo.py"}) + shell_grep_call = ("run_terminal_command", {"command": "rg -n foo src/bar.py"}) + + self._check_remind_burst("H1", "grep burst → native deny on 3rd call", 3, "Too many consecutive grep calls", [grep_call] * 3) + + self._section("H2 — deny rate-limit window (same session, immediately after H1)") + result = self._run_hook("remind", self._pre_tool_use_payload("live-H1", "grep", {"pattern": "bar", "path": "."})) + self._write_evidence("H2.txt", f"rc={result.return_code}\n{result.stdout}") + if result.return_code == 0 and not result.stdout.strip(): + self._record("H2", Status.PASS, "hook is a no-op within the 120s post-deny window") + else: + self._record("H2", Status.FAIL, f"expected silence, got rc={result.return_code} out={result.stdout.strip()!r}") + + self._check_remind_burst( + "H3", "read_file burst (target_file) → deny on 3rd call", 3, "Too many consecutive read calls", [read_call] * 3 + ) + self._check_remind_burst( + "H4", + "run_terminal_command + rg classified as grep → deny on 3rd call", + 3, + "Too many consecutive grep calls", + [shell_grep_call] * 3, + ) + self._check_remind_burst( + "H5", + "mixed grep/read burst → combined deny on 4th call", + 4, + "mixed grep and read", + [grep_call, read_call, grep_call, read_call], + ) + + self._section("H6 — neutral tools (edit/list) never trigger") + neutral_calls = [ + ("search_replace", {"target_file": "src/a.py", "instructions": "x"}), + ("list_dir", {"target_directory": "."}), + ("search_replace", {"target_file": "src/b.py"}), + ] + neutral_outputs = [ + self._run_hook("remind", self._pre_tool_use_payload("live-H6", name, tool_input)) for name, tool_input in neutral_calls + ] + self._write_evidence("H6.txt", "\n".join(f"rc={r.return_code} out={r.stdout!r}" for r in neutral_outputs)) + if all(r.return_code == 0 and not r.stdout.strip() for r in neutral_outputs): + self._record("H6", Status.PASS, "search_replace/list_dir stay silent") + else: + self._record("H6", Status.FAIL, "a neutral tool produced output or an error (see evidence/H6.txt)") + + self._section("H7 — auto-approve emits native allow") + result = self._run_hook( + "auto-approve", self._pre_tool_use_payload("live-H7", "serena__find_symbol", {}, permission_mode="acceptEdits") + ) + self._write_evidence("H7.txt", f"rc={result.return_code}\n{result.stdout}") + if result.return_code == 0 and result.stdout.strip() == '{"decision": "allow"}': + self._record("H7", Status.PASS, 'exact native allow: {"decision": "allow"}') + else: + self._record("H7", Status.FAIL, f"expected exact native allow, got {result.stdout.strip()!r}") + + self._section("H8 — activate accepts camelCase sessionId (informational)") + result = self._run_hook("activate", {"sessionId": "live-H8"}) + self._write_evidence("H8.txt", f"rc={result.return_code}\n{result.stdout}") + if result.return_code == 0 and '"hookEventName": "SessionStart"' in result.stdout: + self._record( + "H8", Status.INFO, "runs clean; emits the client-agnostic SessionStart shape (Grok setup omits this hook by design)" + ) + elif result.return_code == 0: + self._record("H8", Status.WARN, "ran clean but produced unexpected output (see evidence/H8.txt)") + else: + self._record("H8", Status.FAIL, f"activate crashed (rc={result.return_code})") + + self._section("H9 — cleanup removes exactly its session's state") + hook_data_dir = self.config.hook_home / "hook_data" + if not hook_data_dir.is_dir(): + self._record("H9", Status.FAIL, "no hook state was persisted by the preceding checks — nothing to clean up") + return + before = sorted(p.name for p in hook_data_dir.iterdir()) + result = self._run_hook("cleanup", {"sessionId": "live-H1"}) + after = sorted(p.name for p in hook_data_dir.iterdir()) + self._write_evidence("H9.txt", f"before={before}\nafter={after}\nrc={result.return_code}") + if result.return_code == 0 and not result.stdout.strip() and "live-H1" not in after and "live-H3" in after: + self._record("H9", Status.PASS, "live-H1 state removed, other sessions untouched") + else: + self._record("H9", Status.FAIL, "cleanup did not behave as expected (see evidence/H9.txt)") + + # ------------------------------------------------------------------ phase S: client setup + + def check_s_client_setup(self) -> None: + self._section("S1 — serena setup grok") + expected_add_command = "grok mcp add --scope user serena -- serena start-mcp-server --context=grok --project-from-cwd" + result = run_command([str(self.config.serena_bin), "setup", "grok"], timeout=120, cwd=self.config.repo_root) + self._write_evidence("S1.txt", result.output) + expected_fragments = ( + expected_add_command, + "IMPORTANT: We additionally recommend to set up hooks for Grok", + "030_clients.html#grok", + "Serena has been successfully set up for grok.", + ) + if result.return_code == 0 and all(fragment in result.stdout for fragment in expected_fragments): + self._registered_server_names.append("serena") + self._record("S1", Status.PASS, "registered against the real grok CLI with all expected messaging") + else: + self._record("S1", Status.FAIL, f"rc={result.return_code}, output diverges (see evidence/S1.txt)") + + self._section("S2 — registration content is exact") + section = self._read_grok_mcp_server_section("serena") + list_result = self._grok("mcp", "list") + self._write_evidence("S2.txt", f"mcp list:\n{list_result.output}\n\n[mcp_servers.serena] = {section!r}") + registered_tokens = " ".join([str(section.get("command", "")), *section.get("args", [])]) + required_tokens = ("serena", "start-mcp-server", "--context=grok", "--project-from-cwd") + if "serena" in list_result.output and all(token in registered_tokens for token in required_tokens): + self._record("S2", Status.PASS, "config registers command 'serena' with the context and project flags") + else: + self._record("S2", Status.FAIL, f"registered entry incomplete: {section!r}") + + self._section("S3 — grok inspect discovers the server") + inspect_result = self._grok("inspect", "--json", timeout=60) + self._write_evidence("S3-inspect.json", inspect_result.output) + server_names = self._extract_inspect_mcp_server_names(inspect_result.stdout) + if "serena" in server_names: + self._record("S3", Status.PASS, "inspect --json lists serena under mcpServers") + else: + self._record("S3", Status.WARN, f"serena not among inspect's mcpServers {server_names} — review evidence/S3-inspect.json") + + self._section("S4 — idempotent re-run") + rerun = run_command([str(self.config.serena_bin), "setup", "grok"], timeout=120, cwd=self.config.repo_root) + self._write_evidence("S4.txt", rerun.output) + if rerun.return_code == 0 and self._read_grok_mcp_server_section("serena"): + self._record("S4", Status.PASS, "re-run succeeds and the entry remains registered exactly once") + else: + self._record("S4", Status.FAIL, f"re-run rc={rerun.return_code}") + + self._section("S5 — negative path: grok missing from PATH → clean refusal") + minimal_path = "C:\\Windows\\System32" if sys.platform == "win32" else "/usr/bin:/bin" + result = run_command([str(self.config.serena_bin), "setup", "grok"], env_overlay={"PATH": minimal_path}) + self._write_evidence("S5.txt", f"rc={result.return_code}\n{result.output}") + if result.return_code == 1 and "Cannot apply setup for client 'grok'" in result.stdout: + self._record("S5", Status.PASS, "is_applicable()==False path exits 1 with the documented message") + else: + self._record("S5", Status.FAIL, f"rc={result.return_code} (expected 1) or message mismatch (see evidence/S5.txt)") + + @staticmethod + def _extract_inspect_mcp_server_names(inspect_json: str) -> list[str]: + """:return: the MCP server names listed by ``grok inspect --json``, or an empty list if the output cannot be parsed""" + try: + data = json.loads(inspect_json) + return [server.get("name", "") for server in data.get("mcpServers", [])] + except (json.JSONDecodeError, AttributeError, TypeError): + return [] + + # ------------------------------------------------------------------ phase M: MCP server + + def _server_argv(self, with_project: bool, context: str = "grok") -> list[str]: + argv = [str(self.config.serena_bin), "start-mcp-server", "--context", context, "--enable-web-dashboard", "false"] + if with_project: + argv.insert(2, "--project-from-cwd") + return argv + + def _check_mcp_handshake( + self, check_id: str, title: str, cwd: Path, with_project: bool, required: tuple[str, ...], forbidden: tuple[str, ...] + ) -> None: + """ + Starts the MCP server over stdio, performs the initialize handshake, lists tools and + asserts the presence/absence of the given tool names. + """ + self._section(f"{check_id} — {title}") + with self._open_evidence(f"{check_id}-server.log") as stderr_log: + probe = McpStdioProbe(self._server_argv(with_project), cwd=cwd, stderr_log=stderr_log) + try: + init_params = { + "protocolVersion": "2024-11-05", + "capabilities": {}, + "clientInfo": {"name": "serena-live-test", "version": "0.0.0"}, + } + initialization = probe.request(1, "initialize", init_params, timeout=240) + probe.notify("notifications/initialized") + tools_response = probe.request(2, "tools/list", {}, timeout=240) + except AbortError as e: + self._record(check_id, Status.FAIL, f"{e} (see evidence/{check_id}-server.log)") + return + finally: + probe.close() + + # evaluate the exposed tool set against the context's contract + tools = sorted(tool["name"] for tool in tools_response["result"]["tools"]) + server_info = initialization["result"].get("serverInfo", {}) + self._write_evidence(f"{check_id}.txt", f"server: {server_info}\ntool count: {len(tools)}\ntools: {','.join(tools)}") + leaked = sorted(set(forbidden) & set(tools)) + missing = sorted(set(required) - set(tools)) + if not leaked and not missing: + self._record(check_id, Status.PASS, f"handshake ok, tool set as expected ({len(tools)} tools)") + else: + self._record(check_id, Status.FAIL, f"forbidden tools exposed: {leaked or 'none'}; required tools missing: {missing or 'none'}") + + def check_m_mcp_server(self) -> None: + no_project_dir = self.config.work_dir / "noproject" + no_project_dir.mkdir(exist_ok=True) + self._check_mcp_handshake( + "M1", + "MCP handshake, no project (context exclusions over the wire)", + no_project_dir, + with_project=False, + required=(), + forbidden=CONTEXT_EXCLUDED_TOOLS, + ) + self._check_mcp_handshake( + "M2", + "MCP handshake with --project-from-cwd (single-project mode)", + self.config.repo_root, + with_project=True, + required=REQUIRED_SYMBOLIC_TOOLS, + forbidden=(*CONTEXT_EXCLUDED_TOOLS, "activate_project"), + ) + + self._section("M3 — grok mcp doctor spawns the real server (absolute serena path)") + add_result = self._grok( + "mcp", + "add", + "--scope", + "user", + LIVE_MCP_SERVER_NAME, + "--", + str(self.config.serena_bin), + "start-mcp-server", + "--context=grok", + "--project-from-cwd", + "--enable-web-dashboard", + "false", + ) + if add_result.return_code != 0: + self._write_evidence("M3.txt", add_result.output) + self._record("M3", Status.FAIL, f"could not register the {LIVE_MCP_SERVER_NAME} entry (see evidence/M3.txt)") + else: + self._registered_server_names.append(LIVE_MCP_SERVER_NAME) + self._check_m3_doctor() + + self._section("M4 — readiness of the 'serena' command on PATH (informational)") + self._check_m4_path_serena() + + self._check_m5_structured_output() + + # the production and test entries are no longer needed past this point + self._cleanup_grok_state() + + def _doctor_looks_healthy(self, result: CommandResult) -> bool: + return result.return_code == 0 and not any( + marker in result.output.lower() for marker in ("timeout", "timed out", "failed", "error") + ) + + def _check_m3_doctor(self) -> None: + evidence = [] + + # first attempt, then a plain retry (grok may cache a slow first spawn) + doctor = self._grok("mcp", "doctor", LIVE_MCP_SERVER_NAME, "--json", timeout=180) + evidence.append(doctor.output) + if not self._doctor_looks_healthy(doctor): + doctor = self._grok("mcp", "doctor", LIVE_MCP_SERVER_NAME, "--json", timeout=180) + evidence.append("--- retry ---\n" + doctor.output) + + # if the spawn seems to time out, retry with a raised startup timeout (above grok's 30s default); + # 'grok mcp add' has no flag for it, so it must be inserted into the config directly + if not self._doctor_looks_healthy(doctor) and self._insert_startup_timeout_sec(60): + doctor = self._grok("mcp", "doctor", LIVE_MCP_SERVER_NAME, "--json", timeout=180) + evidence.append("--- with startup_timeout_sec=60 ---\n" + doctor.output) + if self._doctor_looks_healthy(doctor): + self._finding( + "grok's default MCP startup timeout (30s) was too low for Serena's cold start in this environment; " + "raising startup_timeout_sec (e.g. 60) in the TOML fixed it — worth noting for slow environments" + ) + + self._write_evidence("M3-doctor.txt", "\n".join(evidence)) + if self._doctor_looks_healthy(doctor): + self._record("M3", Status.PASS, "doctor reports the server connectable (heuristic — skim evidence/M3-doctor.txt)") + else: + self._record("M3", Status.FAIL, f"doctor rc={doctor.return_code} (see evidence/M3-doctor.txt)") + + def _check_m4_path_serena(self) -> None: + """Checks whether the ``serena`` resolved via PATH (what the production registration will spawn) supports the grok context.""" + path_serena = shutil.which("serena") + if path_serena is None: + self._record("M4", Status.INFO, "no 'serena' on PATH — the production registration requires one (e.g. via uv tool install)") + return + if Path(path_serena).resolve() == self.config.serena_bin.resolve(): + self._record( + "M4", Status.INFO, "PATH resolves 'serena' to the binary under test — production registration will behave like M1-M3" + ) + return + + # probe the foreign installation briefly: surviving the timeout means the context loaded + probe = run_command([path_serena, "start-mcp-server", "--context", "grok", "--enable-web-dashboard", "false"], timeout=15) + self._write_evidence( + "M4.txt", f"PATH serena: {path_serena}\nrc={probe.return_code} timed_out={probe.timed_out}\n{probe.output[-3000:]}" + ) + if probe.timed_out: + self._record( + "M4", Status.INFO, f"'serena' on PATH ({path_serena}) supports the grok context — production registration is functional" + ) + else: + self._record( + "M4", Status.INFO, f"'serena' on PATH ({path_serena}) cannot start with the grok context — likely a pre-Grok release" + ) + self._finding( + "the production registration spawns the bare 'serena' command; it only works once the installation on PATH " + "includes Grok support (see evidence/M4.txt for the observed failure)" + ) + + def _probe_tool_call(self, check_id: str, context: str) -> Optional[tuple[Optional[dict], dict]]: + """ + Starts the MCP server with the given context (project mode), calls the read-only probe tool + and returns its advertised output schema together with the raw ``tools/call`` result. + + :param check_id: the check id (used for evidence file names and FAIL records) + :param context: the Serena context to start the server with + :return: ``(output_schema, call_result)``, or None if the probe failed (a FAIL is recorded) + """ + with self._open_evidence(f"{check_id}-{context}-server.log") as stderr_log: + probe = McpStdioProbe(self._server_argv(with_project=True, context=context), cwd=self.config.repo_root, stderr_log=stderr_log) + try: + init_params = { + "protocolVersion": "2024-11-05", + "capabilities": {}, + "clientInfo": {"name": "serena-live-test", "version": "0.0.0"}, + } + probe.request(1, "initialize", init_params, timeout=240) + probe.notify("notifications/initialized") + tools_response = probe.request(2, "tools/list", {}, timeout=240) + call_response = probe.request(3, "tools/call", {"name": STRUCTURED_OUTPUT_PROBE_TOOL, "arguments": {}}, timeout=240) + except AbortError as e: + self._record(check_id, Status.FAIL, f"{context}: {e} (see evidence/{check_id}-{context}-server.log)") + return None + finally: + probe.close() + + entries = [tool for tool in tools_response["result"]["tools"] if tool["name"] == STRUCTURED_OUTPUT_PROBE_TOOL] + if not entries: + self._record(check_id, Status.FAIL, f"{context}: probe tool '{STRUCTURED_OUTPUT_PROBE_TOOL}' not exposed") + return None + return entries[0].get("outputSchema"), call_response.get("result", {}) + + def _check_m5_structured_output(self) -> None: + """ + Verifies the structured-tool-output wire shape of the grok context (auto default) against the + claude-code context (explicit ``false`` workaround): for a string-returning tool, the grok + context must advertise an output schema and return ``structuredContent`` while claude-code + must serve plain text only. + + This pins what Serena *serves* under each context; whether Grok Build unpacks + ``structuredContent`` client-side can only be observed in a real agent session, which is + out of scope for a zero-inference test. + """ + self._section("M5 — structured tool output over the wire (grok auto default vs claude-code workaround)") + grok_probe = self._probe_tool_call("M5", "grok") + if grok_probe is None: + return + claude_probe = self._probe_tool_call("M5", "claude-code") + if claude_probe is None: + return + grok_schema, grok_result = grok_probe + claude_schema, claude_result = claude_probe + + problems = [] + grok_structured = grok_result.get("structuredContent") + if grok_result.get("isError"): + problems.append("grok: probe tool call returned isError") + if not isinstance(grok_schema, dict): + problems.append("grok: no outputSchema advertised for the string-returning probe tool") + if not (isinstance(grok_structured, dict) and isinstance(grok_structured.get("result"), str)): + problems.append("grok: structuredContent missing or not of shape {'result': str}") + if claude_result.get("isError"): + problems.append("claude-code: probe tool call returned isError") + if claude_schema is not None: + problems.append("claude-code: unexpectedly advertises an outputSchema despite structured_tool_output=false") + if claude_result.get("structuredContent") is not None: + problems.append("claude-code: unexpectedly returns structuredContent despite structured_tool_output=false") + + self._write_evidence( + "M5.txt", + f"probe tool: {STRUCTURED_OUTPUT_PROBE_TOOL}\n" + f"grok outputSchema: {json.dumps(grok_schema)}\n" + f"grok structuredContent keys: {sorted(grok_structured) if isinstance(grok_structured, dict) else grok_structured!r}\n" + f"claude-code outputSchema: {json.dumps(claude_schema)}\n" + f"claude-code structuredContent: {claude_result.get('structuredContent')!r}\n", + ) + if problems: + self._record("M5", Status.FAIL, "; ".join(problems)) + else: + self._record( + "M5", + Status.PASS, + "grok serves outputSchema + structuredContent({'result': str}); claude-code serves plain text only " + "(auto-default divergence verified on the wire)", + ) + + # ------------------------------------------------------------------ phase D: hook discovery + + def check_d_hook_discovery(self) -> None: + hooks_json_text = json.dumps(HOOKS_JSON, indent=2) + + self._section("D1 — project-scoped .grok/hooks discovery") + hook_project = self.config.work_dir / "hookproj" + hooks_dir = hook_project / ".grok" / "hooks" + hooks_dir.mkdir(parents=True) + run_command(["git", "init", "-q", "."], cwd=hook_project) + (hooks_dir / "serena-hooks.json").write_text(hooks_json_text, encoding="utf-8") + inspect_result = run_command([self.config.grok_bin, "inspect", "--json"], timeout=60, cwd=hook_project) + self._write_evidence("D1-inspect.json", inspect_result.output) + if "serena-hooks" in inspect_result.output: + self._record("D1", Status.PASS, "grok inspect discovers the docs' hooks JSON in the project's .grok/hooks/") + else: + self._record( + "D1", Status.WARN, "project hooks not visible in inspect output — possibly gated on hook trust (evidence/D1-inspect.json)" + ) + self._finding( + "the docs' project-scoped hooks JSON was not surfaced by 'grok inspect --json' in an untrusted project; " + "Grok requires /hooks-trust before project hooks load (expected; documented in the Grok hooks docs section)" + ) + + self._section("D2 — global hooks directory next to the Grok config (informational)") + global_hooks_dir = self.config.grok_config_path.parent / "hooks" + global_hooks_file = global_hooks_dir / "serena-hooks.json" + if global_hooks_file.exists(): + self._record("D2", Status.SKIP, f"{global_hooks_file} already exists — not probing to avoid touching user state") + return + global_hooks_dir.mkdir(exist_ok=True) + global_hooks_file.write_text(hooks_json_text, encoding="utf-8") + self._created_global_hooks_file = global_hooks_file + inspect_result = run_command([self.config.grok_bin, "inspect", "--json"], timeout=60, cwd=self.config.work_dir / "noproject") + self._write_evidence("D2-inspect.json", inspect_result.output) + if "serena-hooks" in inspect_result.output: + self._record("D2", Status.INFO, "global hooks directory IS discovered — the docs' global recommendation works") + else: + self._record("D2", Status.INFO, "global hooks directory NOT discovered — the docs' 'globally' wording may need correction") + self._finding("the docs claim a global hooks file next to the Grok config works; 'grok inspect' did not surface it") + self._cleanup_grok_state() + + # ------------------------------------------------------------------ phase C: cleanup + + def check_c1_cleanup(self) -> None: + self._section("C1 — cleanup and baseline verification") + self._cleanup_grok_state() + self._cleanup_done = True + if self._baseline_mcp_list == "": + self._record("C1", Status.SKIP, "no baseline was captured (aborted before P3)") + return + + backup_exists = self.config.config_backup_path.exists() + final_list = self._grok("mcp", "list").output + self._write_evidence("C1-final-mcp-list.txt", final_list) + print(final_list.strip()) + if final_list == self._baseline_mcp_list: + if not backup_exists: + self._record("C1", Status.PASS, "mcp list matches the baseline; config was not modified (hooks-only, no backup)") + return + backup_matches = ( + self.config.grok_config_path.exists() + and self.config.grok_config_path.read_bytes() == self.config.config_backup_path.read_bytes() + ) + note = "byte-identical to the backup" if backup_matches else "differs from the backup only in formatting (mcp list matches)" + self._record("C1", Status.PASS, f"mcp list matches the baseline; config {note}") + self._discard_config_backup() + return + + # the removals did not return to baseline: restore the backup outright + if not backup_exists: + self._record("C1", Status.FAIL, "state diverged but no config backup exists to restore — MANUAL ATTENTION") + return + shutil.copy2(self.config.config_backup_path, self.config.grok_config_path) + restored_list = self._grok("mcp", "list").output + if restored_list == self._baseline_mcp_list: + self._record("C1", Status.PASS, "state diverged after removals; restored the config backup — baseline verified") + self._discard_config_backup() + else: + self._record( + "C1", Status.FAIL, f"could not restore the baseline — MANUAL ATTENTION, backup at {self.config.config_backup_path}" + ) + + def _discard_config_backup(self) -> None: + """Remove the config backup once the baseline is confirmed intact, so no credential copy persists.""" + try: + self.config.config_backup_path.unlink(missing_ok=True) + except OSError: + pass + + # ------------------------------------------------------------------ orchestration + + def run(self) -> int: + """ + Runs all phases, guaranteeing Grok-side cleanup even on abort. + + :return: the process exit code (0 iff no check failed) + """ + aborted: Optional[str] = None + try: + self.check_p1_environment() + self.check_p3_baseline() + self.check_p2_unit_smoke() + self.check_p4_context_loads() + self.check_h_hook_protocol() + if self.config.hooks_only: + for check_id in ("S1", "S2", "S3", "S4", "S5", "M1", "M2", "M3", "M4", "M5", "D1", "D2"): + self._record(check_id, Status.SKIP, "--hooks-only") + else: + self.check_s_client_setup() + self.check_m_mcp_server() + self.check_d_hook_discovery() + except AbortError as e: + aborted = str(e) + print(f"\nABORTED: {e}") + finally: + self.check_c1_cleanup() + + report_path = self._write_report(aborted) + print(f"\nreport written to {report_path}") + failed = sum(1 for result in self.results if result.status == Status.FAIL) + return 1 if failed or aborted else 0 + + def _write_report(self, aborted: Optional[str]) -> Path: + counts = {status: sum(1 for r in self.results if r.status == status) for status in Status} + lines = [ + "# Grok Live Test Report", + "", + f"- Date: {datetime.now().isoformat(timespec='seconds')}", + f"- Environment: {self._environment_summary or 'n/a'}", + f"- Mode: {'hooks-only' if self.config.hooks_only else 'full'}" + (", unit tests skipped" if self.config.skip_unit else ""), + f"- Totals: {counts[Status.PASS]} pass / {counts[Status.FAIL]} fail / {counts[Status.WARN]} warn" + f" / {counts[Status.INFO]} info / {counts[Status.SKIP]} skipped", + "- Cost: zero inference (no Grok session was started)", + ] + if aborted: + lines.append(f"- **ABORTED**: {aborted}") + lines += ["", "| ID | Status | Observation |", "|----|--------|-------------|"] + lines += [f"| {r.check_id} | {r.status.value} | {r.note} |" for r in self.results] + lines += ["", "## Findings", ""] + lines += [f"{i}. {finding}" for i, finding in enumerate(self.findings, start=1)] or ["None."] + lines += ["", f"Evidence: {self.config.evidence_dir}/", ""] + report_path = self.config.work_dir / "report.md" + report_path.write_text("\n".join(lines), encoding="utf-8") + os.chmod(report_path, 0o600) + return report_path + + +def main() -> int: + parser = argparse.ArgumentParser( + description="Live (zero-inference) smoke test of Serena's Grok Build integration against a real grok CLI.", + epilog="Backs up the Grok config before any change and restores the baseline afterwards; " + "refuses to run if a 'serena' MCP entry is already registered in Grok.", + ) + parser.add_argument("--hooks-only", action="store_true", help="run only the pure-local checks; never touches the Grok configuration") + parser.add_argument("--skip-unit", action="store_true", help="skip the pytest smoke run (phase P2)") + parser.add_argument("--work-dir", help="directory for evidence and reports (default: a fresh private mkdtemp under the system tmp)") + parser.add_argument("--repo-root", help="Serena repository root (default: derived from this script's location)") + parser.add_argument("--serena-bin", help="serena executable under test (default: /.venv)") + parser.add_argument("--serena-hooks-bin", help="serena-hooks executable under test (default: /.venv)") + parser.add_argument("--grok-bin", help="grok executable (default: resolved from PATH)") + parser.add_argument( + "--grok-config", + help="Grok user config file, for nonstandard Grok homes (default: ~/.grok/config.toml). NOT an isolation " + "mechanism: 'grok mcp' has no alternate-config option, so this must be the config the CLI actually uses — " + "verified against 'grok mcp list' at startup, mismatch aborts before any mutation", + ) + args = parser.parse_args() + + try: + config = LiveTestConfig.from_args(args) + except AbortError as e: + print(f"ABORTED: {e}") + return 1 + print(f"work directory: {config.work_dir}") + return GrokLiveTest(config).run() + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/src/serena/config/client_setup.py b/src/serena/config/client_setup.py index 933ff4a3..6091812e 100644 --- a/src/serena/config/client_setup.py +++ b/src/serena/config/client_setup.py @@ -109,4 +109,38 @@ class ClientSetupHandlerCodeBuddy(ClientSetupHandler): return is_success -client_setup_handlers = [ClientSetupHandlerClaudeCode(), ClientSetupHandlerCodeBuddy(), ClientSetupHandlerCodex()] +class ClientSetupHandlerGrok(ClientSetupHandler): + """ + Setup for xAI Grok Build. + """ + + def __init__(self) -> None: + super().__init__("grok") + + def is_applicable(self) -> bool: + version_result = execute_shell_command("grok --version", capture_stderr=True) + if version_result.return_code != 0 or not version_result.stdout.lower().startswith("grok "): + return False + + mcp_result = execute_shell_command("grok mcp add --help", capture_stderr=True) + return mcp_result.return_code == 0 and "Add or update an MCP server" in mcp_result.stdout + + def get_mcp_server_options(self) -> list[str]: + return ["--context=grok", "--project-from-cwd"] + + def apply(self) -> bool: + cmd = f"grok mcp add --scope user serena -- {self.get_mcp_server_command()}" + is_success = self._run_shell_command(cmd) + if is_success: + click.echo("\nIMPORTANT: We additionally recommend to set up hooks for Grok to ensure the best experience.") + click.echo(" Please read the instructions here:") + click.echo(" https://oraios.github.io/serena/02-usage/030_clients.html#grok") + return is_success + + +client_setup_handlers = [ + ClientSetupHandlerClaudeCode(), + ClientSetupHandlerCodeBuddy(), + ClientSetupHandlerCodex(), + ClientSetupHandlerGrok(), +] diff --git a/src/serena/hooks.py b/src/serena/hooks.py index e4c76769..3d53961a 100644 --- a/src/serena/hooks.py +++ b/src/serena/hooks.py @@ -25,6 +25,7 @@ class HookClient(Enum): CODEBUDDY = "codebuddy" VSCODE = "vscode" CODEX = "codex" + GROK = "grok" class Hook(ABC): @@ -88,6 +89,12 @@ class PreToolUseHook(Hook, ABC): additional_context: str = "" def to_json_string(self, client: HookClient) -> str: + if client == HookClient.GROK: + grok_output: dict[str, str] = {"decision": self.permission_decision} + if self.permission_decision == "deny": + grok_output["reason"] = self.permission_decision_reason + return json.dumps(grok_output) + hook_output = { "hookSpecificOutput": { "hookEventName": "PreToolUse", @@ -267,11 +274,11 @@ class PreToolUseRemindAboutSymbolicToolsHook(PreToolUseHook): #: are caught alongside ``read_file``, while ``write_file``/``edit_file`` are not. _READ_FILE_VERB_SUBSTRINGS: frozenset[str] = frozenset(("read", "view", "open", "show")) - #: Shell commands that perform grep-like search; used to classify Codex + #: Shell commands that perform grep-like search; used to classify Codex/Grok #: shell-command tool calls whose ``cmd`` or ``command`` field starts with one of these. _GREP_SHELL_COMMANDS: frozenset[str] = frozenset(("grep", "rg", "ag", "ack", "fgrep", "egrep", "search_for_pattern")) - #: Shell commands that perform file-read operations; used to classify Codex + #: Shell commands that perform file-read operations; used to classify Codex/Grok #: shell-command tool calls whose ``cmd`` or ``command`` field starts with one of these. _READ_SHELL_COMMANDS: frozenset[str] = frozenset(("cat", "head", "tail", "sed", "less", "more", "bat", "get-content", "gc")) @@ -345,7 +352,7 @@ class PreToolUseRemindAboutSymbolicToolsHook(PreToolUseHook): super().__init__(client) self._tool_call_counter = self.ToolUseCounter.load(self) - # extract a shell ``cmd``/``command`` field (Codex shell-command-style payloads): + # extract a shell ``cmd``/``command`` field (Codex/Grok shell-command-style payloads): # split into command name (basename, lowercased) and the remaining argument string # so both bare names (``rg``) and path-prefixed invocations (``/usr/bin/grep``) are # normalised the same way. Stays ``None`` when no shell command is present. @@ -356,18 +363,26 @@ class PreToolUseRemindAboutSymbolicToolsHook(PreToolUseHook): # payloads pass the target as ``file_path`` rather than via a shell command). self._file_path: str | None = None if self._tool_input is not None: - self._command = self._tool_input.get("cmd", self._tool_input.get("command", "")).strip() + self._command = str(self._tool_input.get("cmd", self._tool_input.get("command", ""))).strip() if self._command: cmd_split = self._command.split(maxsplit=1) if len(cmd_split) > 1: self._command_args_str = cmd_split[1] self._command_name = os.path.basename(cmd_split[0]).lower() - file_path = self._tool_input.get("file_path") or self._tool_input.get("filePath") or "" + file_path = ( + self._tool_input.get("file_path") + or self._tool_input.get("filePath") + or self._tool_input.get("target_file") + or self._tool_input.get("targetFile") + or "" + ) self._file_path = str(file_path).strip() or None def is_grep_call(self) -> bool: if self._client in (HookClient.CLAUDE_CODE, HookClient.CODEBUDDY): return self._tool_name == "grep" or "search_for_pattern" in self._tool_name + if self._client == HookClient.GROK: + return self._tool_name == "grep" or (self._is_shell_command_call() and self._command_name in self._GREP_SHELL_COMMANDS) if self._client == HookClient.CODEX and self._is_shell_command_call(): return self._command_name in self._GREP_SHELL_COMMANDS # heuristic for other clients @@ -376,6 +391,8 @@ class PreToolUseRemindAboutSymbolicToolsHook(PreToolUseHook): def is_read_call(self) -> bool: if self._client in (HookClient.CLAUDE_CODE, HookClient.CODEBUDDY): return self._tool_name == "read" or "read_file" in self._tool_name + if self._client == HookClient.GROK: + return self._tool_name == "read_file" or (self._is_shell_command_call() and self._command_name in self._READ_SHELL_COMMANDS) if self._client == HookClient.CODEX and self._is_shell_command_call(): return self._command_name in self._READ_SHELL_COMMANDS # heuristic for other clients @@ -400,7 +417,7 @@ class PreToolUseRemindAboutSymbolicToolsHook(PreToolUseHook): if self._file_path is not None: return self._is_code_file_path(self._file_path) - if self._client == HookClient.CODEX and self._command_args_str is not None: + if self._client in (HookClient.CODEX, HookClient.GROK) and self._command_args_str is not None: return any(self._is_code_file_path(argument) for argument in self._iter_shell_path_arguments()) return True diff --git a/src/serena/resources/config/contexts/grok.yml b/src/serena/resources/config/contexts/grok.yml new file mode 100644 index 00000000..ddd2e23e --- /dev/null +++ b/src/serena/resources/config/contexts/grok.yml @@ -0,0 +1,52 @@ +description: Grok Build (CLI agent where file operations, basic edits, etc. are already covered; single project mode) +prompt: | + You are running in a CLI coding agent context where file operations, basic (line-based) edits and reads + as well as shell commands are handled by your own, internal tools. + + You have access to Serena's code intelligence tools that exploit the symbolic + structure of the code and are much more efficient than your own tools for most coding scenarios. + If you are working on any coding task and if Serena's tools are deferred, you + should load them all immediately, before performing any read, grep or bash commands. + If Serena's tools can be used to achieve your task, you should prioritize them. + **CRITICAL**: If you haven't seen Serena's tools yet, you should use the tool search tool to load all of them + **right now**. + For exploring and reading code in a token-efficient manner, use Serena's code-centric overview and symbolic search tools. + For edits, prefer Serena's editing tools since they are more token efficient. + Especially when editing code you have read through Serena, use Serena's tools for editing to preserve the symbolic context you already gathered. + **CRITICAL**: Never use a tool before having read its schema via the tool-search tool! + + **CRITICAL**: You have a preference for relying on your own tools like grep, read_file and search_replace. It is of paramount + importance that you overcome this preference and instead develop a preference for using Serena's tools since they + are superior for most coding tasks. Follow the following rules: + + For any code files: + - read_file -> FORBIDDEN for discovery. Use get_symbols_overview, then find_symbol with include_body. + You may use read_file only if you already have an overview of the file and if reading + whole symbols is inappropriate (e.g. a few lines of code). + - list_dir (by name) -> Allowed for discovery only. + - grep (content) -> Allowed for discovery only; follow up reads or reference searches must be Serena. + - search_replace -> FORBIDDEN. Use replace_symbol_body / insert_*_symbol / replace_content. + + Disallowed reasoning. Do NOT use any of the following to justify read_file/search_replace on a code file: + - "I already know the path" + - "one read_file call is faster than three Serena calls" + - "the built-in tool description says to use read_file for known paths" + If you catch yourself reaching for one of these, that is the signal to switch to Serena. + +excluded_tools: + - create_text_file + - read_file + - execute_shell_command + - find_file + - list_dir + - search_for_pattern + +tool_description_overrides: {} + +# whether to assume that Serena shall only work on a single project in this context (provided that a project is given +# when Serena is started). +# If set to true and a project is provided at startup, the set of tools is limited to those required by the project's +# concrete configuration, and other tools are excluded completely, allowing the set of tools to be minimal. +# Tools explicitly disabled by the project will not be available at all. +# The `activate_project` tool is always disabled in this case, as project switching cannot be allowed. +single_project: true diff --git a/test/serena/config/test_client_setup.py b/test/serena/config/test_client_setup.py new file mode 100644 index 00000000..d9115a7e --- /dev/null +++ b/test/serena/config/test_client_setup.py @@ -0,0 +1,131 @@ +import pytest + +from serena.config.client_setup import ClientSetupHandlerGrok, client_setup_handlers +from serena.config.context_mode import SerenaAgentContext +from serena.util.shell import ShellCommandResult + + +def _result(command: str, return_code: int = 0, stdout: str = "") -> ShellCommandResult: + return ShellCommandResult(stdout=stdout, stderr="", return_code=return_code, cwd=".") + + +def test_grok_setup_handler_is_applicable_for_grok_build(monkeypatch): + commands: list[str] = [] + + def fake_execute_shell_command(command: str, capture_stderr: bool = False) -> ShellCommandResult: + commands.append(command) + if command == "grok --version": + return _result(command, stdout="grok 0.2.82 (6d0b07d2de) [stable]\n") + if command == "grok mcp add --help": + return _result(command, stdout="Add or update an MCP server\n") + return _result(command, return_code=1) + + monkeypatch.setattr("serena.config.client_setup.execute_shell_command", fake_execute_shell_command) + + assert ClientSetupHandlerGrok().is_applicable() is True + assert commands == ["grok --version", "grok mcp add --help"] + + +def test_grok_setup_handler_rejects_binary_without_mcp_add(monkeypatch): + def fake_execute_shell_command(command: str, capture_stderr: bool = False) -> ShellCommandResult: + if command == "grok --version": + return _result(command, stdout="grok 0.2.82 (6d0b07d2de) [stable]\n") + return _result(command, return_code=1, stdout="unknown command\n") + + monkeypatch.setattr("serena.config.client_setup.execute_shell_command", fake_execute_shell_command) + + assert ClientSetupHandlerGrok().is_applicable() is False + + +@pytest.mark.parametrize( + ("return_code", "stdout"), + [ + (1, ""), + (0, "0.0.34\n"), + (0, "my-grok wrapper 1.0\n"), + (0, "grok\n"), + (0, " grok 0.2.82"), + ], +) +def test_grok_setup_handler_short_circuits_when_version_probe_does_not_match(monkeypatch, return_code: int, stdout: str): + commands: list[str] = [] + + def fake_execute_shell_command(command: str, capture_stderr: bool = False) -> ShellCommandResult: + commands.append(command) + if command == "grok --version": + return _result(command, return_code=return_code, stdout=stdout) + return _result(command, stdout="Add or update an MCP server\n") + + monkeypatch.setattr("serena.config.client_setup.execute_shell_command", fake_execute_shell_command) + + assert ClientSetupHandlerGrok().is_applicable() is False + assert commands == ["grok --version"] + + +def test_grok_setup_handler_accepts_case_insensitive_grok_version(monkeypatch): + commands: list[str] = [] + + def fake_execute_shell_command(command: str, capture_stderr: bool = False) -> ShellCommandResult: + commands.append(command) + if command == "grok --version": + return _result(command, stdout="Grok 0.3.0 [stable]") + if command == "grok mcp add --help": + return _result(command, stdout="Add or update an MCP server\n") + return _result(command, return_code=1) + + monkeypatch.setattr("serena.config.client_setup.execute_shell_command", fake_execute_shell_command) + + assert ClientSetupHandlerGrok().is_applicable() is True + assert commands == ["grok --version", "grok mcp add --help"] + + +@pytest.mark.parametrize("help_stdout", ["Usage: grok mcp add\n", "add or update an mcp server\n"]) +def test_grok_setup_handler_rejects_mcp_add_help_without_expected_text(monkeypatch, help_stdout: str): + commands: list[str] = [] + + def fake_execute_shell_command(command: str, capture_stderr: bool = False) -> ShellCommandResult: + commands.append(command) + if command == "grok --version": + return _result(command, stdout="grok 0.2.82 (6d0b07d2de) [stable]\n") + if command == "grok mcp add --help": + return _result(command, stdout=help_stdout) + return _result(command, return_code=1) + + monkeypatch.setattr("serena.config.client_setup.execute_shell_command", fake_execute_shell_command) + + assert ClientSetupHandlerGrok().is_applicable() is False + assert commands == ["grok --version", "grok mcp add --help"] + + +def test_grok_setup_handler_apply_uses_grok_mcp_add(monkeypatch): + commands: list[str] = [] + + def fake_run_shell_command(self: ClientSetupHandlerGrok, command: str) -> bool: + commands.append(command) + return True + + monkeypatch.setattr(ClientSetupHandlerGrok, "_run_shell_command", fake_run_shell_command) + + assert ClientSetupHandlerGrok().apply() is True + assert commands == ["grok mcp add --scope user serena -- serena start-mcp-server --context=grok --project-from-cwd"] + + +def test_grok_setup_handler_apply_failure_skips_hook_recommendation(monkeypatch, capsys): + def fake_run_shell_command(self: ClientSetupHandlerGrok, command: str) -> bool: + return False + + monkeypatch.setattr(ClientSetupHandlerGrok, "_run_shell_command", fake_run_shell_command) + + assert ClientSetupHandlerGrok().apply() is False + assert "recommend" not in capsys.readouterr().out.lower() + + +def test_client_setup_handlers_use_resolvable_contexts(): + handler_names = [handler.name for handler in client_setup_handlers] + assert "grok" in handler_names + + for handler in client_setup_handlers: + context_options = [option for option in handler.get_mcp_server_options() if option.startswith("--context=")] + assert len(context_options) == 1 + context_name = context_options[0].removeprefix("--context=") + assert SerenaAgentContext.from_name(context_name).name == context_name diff --git a/test/serena/config/test_context_mode.py b/test/serena/config/test_context_mode.py new file mode 100644 index 00000000..2a88890f --- /dev/null +++ b/test/serena/config/test_context_mode.py @@ -0,0 +1,60 @@ +import pytest + +from interprompt.jinja_template import JinjaTemplate +from serena.config.context_mode import SerenaAgentContext + +GROK_EXCLUDED_TOOLS = { + "create_text_file", + "read_file", + "execute_shell_command", + "find_file", + "list_dir", + "search_for_pattern", +} + +BUILTIN_RUNTIME_CONTEXT_NAMES = [ + name for name in SerenaAgentContext.list_registered_context_names(include_user_contexts=False) if name != "context.template" +] + + +def _render_context_prompt(context: SerenaAgentContext) -> str: + """Render with the same template variables SerenaAgent._format_prompt provides. + + Note: context YAMLs (e.g. grok.yml) are currently static (no Jinja), but the variables + match exactly what _format_prompt supplies. The outer system prompt template uses them. + """ + # Provide non-empty values so any future Jinja in a context prompt would still be exercised. + return JinjaTemplate(context.prompt).render( + available_tools=["find_symbol", "get_symbols_overview", "replace_content"], + available_markers=["ToolMarkerSymbolicRead"], + tool_names={"find_symbol": "find_symbol", "get_symbols_overview": "get_symbols_overview"}, + ) + + +def test_grok_context_loads(): + context = SerenaAgentContext.from_name("grok") + + assert context.name == "grok" + assert context.single_project is True + assert context.structured_tool_output is None + assert set(context.excluded_tools) == GROK_EXCLUDED_TOOLS + + +def test_grok_context_prompt_renders(): + context = SerenaAgentContext.from_name("grok") + + rendered_prompt = _render_context_prompt(context) + + assert rendered_prompt.strip() + assert "Serena's code intelligence tools" in rendered_prompt + + +@pytest.mark.parametrize("context_name", BUILTIN_RUNTIME_CONTEXT_NAMES) +def test_builtin_contexts_load_and_prompt_templates_render(context_name: str): + context = SerenaAgentContext.from_name(context_name) + + rendered_prompt = _render_context_prompt(context) + + assert context.name == context_name + assert isinstance(context.excluded_tools, list) + assert rendered_prompt == "" or rendered_prompt.strip() diff --git a/test/serena/config/test_grok_docs.py b/test/serena/config/test_grok_docs.py new file mode 100644 index 00000000..a3f212f4 --- /dev/null +++ b/test/serena/config/test_grok_docs.py @@ -0,0 +1,67 @@ +import inspect +import json +import re +from io import StringIO +from pathlib import Path +from unittest.mock import patch + +from serena.config.client_setup import ClientSetupHandlerGrok +from serena.hooks import HookClient, PreToolUseRemindAboutSymbolicToolsHook + +PROJECT_ROOT = Path(__file__).resolve().parents[3] + + +def _extract_section(markdown: str, heading: str) -> str: + """Extract the content under a top-level ## heading (simple and sufficient for our docs).""" + pattern = rf"\n## {re.escape(heading)}\n(.*?)(?=\n## |\Z)" + match = re.search(pattern, markdown, re.DOTALL) + return match.group(1) if match else "" + + +def _grok_payload(tool_name: str, tool_input: dict) -> dict: + return { + "hookEventName": "pre_tool_use", + "sessionId": f"docs-{tool_name}", + "toolName": tool_name, + "toolInput": tool_input, + "toolInputTruncated": False, + "permissionMode": "bypassPermissions", + } + + +def test_grok_docs_are_consistent(): + clients_doc = (PROJECT_ROOT / "docs/02-usage/030_clients.md").read_text() + config_doc = (PROJECT_ROOT / "docs/02-usage/050_configuration.md").read_text() + handler_source = inspect.getsource(ClientSetupHandlerGrok.apply) + + assert "\n## Grok\n" in clients_doc + grok_section = _extract_section(clients_doc, "Grok") + + assert "030_clients.html#grok" in handler_source + assert "serena setup grok" in grok_section + assert "serena-hooks remind --client=grok" in grok_section + assert "serena-hooks cleanup --client=grok" in grok_section + + assert "* `grok`: Optimized for use with xAI's Grok Build CLI." in config_doc + assert "contexts `ide`, `claude-code`, and `grok` are **single-project contexts**" in config_doc + + +def test_grok_hook_matcher_docs_match_code(tmp_path: Path): + clients_doc = (PROJECT_ROOT / "docs/02-usage/030_clients.md").read_text() + grok_section = _extract_section(clients_doc, "Grok") + + matcher = re.search(r'"matcher": "([^"]+)"', grok_section) + assert matcher is not None + matcher_tools = set(matcher.group(1).split("|")) + assert matcher_tools == {"grep", "read_file", "run_terminal_command"} + + payloads = { + "grep": _grok_payload("grep", {"pattern": "foo", "path": "."}), + "read_file": _grok_payload("read_file", {"target_file": "src/foo.py"}), + "run_terminal_command": _grok_payload("run_terminal_command", {"command": "rg -n foo README.md"}), + } + for tool_name in matcher_tools: + with patch("sys.stdin", StringIO(json.dumps(payloads[tool_name]))), patch("serena.hooks.serena_home_dir", str(tmp_path)): + hook = PreToolUseRemindAboutSymbolicToolsHook(HookClient.GROK) + + assert hook.is_grep_call() or hook.is_read_call() or hook._is_shell_command_call() diff --git a/test/serena/test_cli_setup.py b/test/serena/test_cli_setup.py new file mode 100644 index 00000000..6f204f70 --- /dev/null +++ b/test/serena/test_cli_setup.py @@ -0,0 +1,91 @@ +from types import SimpleNamespace + +from click.testing import CliRunner + +from serena.cli import TopLevelCommands +from serena.util.shell import ShellCommandResult + +GROK_ADD_COMMAND = "grok mcp add --scope user serena -- serena start-mcp-server --context=grok --project-from-cwd" + + +def _result(command: str, return_code: int = 0, stdout: str = "", stderr: str = "") -> ShellCommandResult: + return ShellCommandResult(stdout=stdout, stderr=stderr, return_code=return_code, cwd=".") + + +def test_setup_grok_success(monkeypatch): + commands: list[str] = [] + + def fake_execute_shell_command(command: str, capture_stderr: bool = False) -> ShellCommandResult: + commands.append(command) + if command == "grok --version": + return _result(command, stdout="grok 0.2.82 (6d0b07d2de) [stable]\n") + if command == "grok mcp add --help": + return _result(command, stdout="Add or update an MCP server\n") + if command == GROK_ADD_COMMAND: + return _result(command) + return _result(command, return_code=1) + + monkeypatch.setattr("serena.config.client_setup.execute_shell_command", fake_execute_shell_command) + + result = CliRunner().invoke(TopLevelCommands.setup, ["grok"]) + + assert result.exit_code == 0, result.output + assert "successfully set up for grok" in result.output + assert "recommend" in result.output.lower() + assert "030_clients.html#grok" in result.output + assert commands == ["grok --version", "grok mcp add --help", GROK_ADD_COMMAND] + + +def test_setup_grok_not_applicable_exits_1(monkeypatch): + def fake_execute_shell_command(command: str, capture_stderr: bool = False) -> ShellCommandResult: + if command == "grok --version": + return _result(command, return_code=1) + return _result(command) + + monkeypatch.setattr("serena.config.client_setup.execute_shell_command", fake_execute_shell_command) + + result = CliRunner().invoke(TopLevelCommands.setup, ["grok"]) + + assert result.exit_code == 1 + assert "Cannot apply setup for client 'grok'" in result.output + + +def test_setup_grok_apply_failure_exits_1(monkeypatch): + def fake_execute_shell_command(command: str, capture_stderr: bool = False) -> ShellCommandResult: + if command == "grok --version": + return _result(command, stdout="grok 0.2.82 (6d0b07d2de) [stable]\n") + if command == "grok mcp add --help": + return _result(command, stdout="Add or update an MCP server\n") + if command == GROK_ADD_COMMAND: + return _result(command, return_code=1, stderr="boom") + return _result(command, return_code=1) + + monkeypatch.setattr("serena.config.client_setup.execute_shell_command", fake_execute_shell_command) + + result = CliRunner().invoke(TopLevelCommands.setup, ["grok"]) + + assert result.exit_code == 1 + assert "Failed to set up Serena for grok" in result.output + assert "recommend" not in result.output.lower() + assert "030_clients.html#grok" not in result.output + + +def test_init_suggests_grok_when_detected(monkeypatch): + def fake_execute_shell_command(command: str, capture_stderr: bool = False) -> ShellCommandResult: + if command == "grok --version": + return _result(command, stdout="grok 0.2.82 (6d0b07d2de) [stable]\n") + if command == "grok mcp add --help": + return _result(command, stdout="Add or update an MCP server\n") + return _result(command, return_code=1) + + def fake_init(**kwargs) -> SimpleNamespace: + return SimpleNamespace(config_file_path="/tmp/serena_config.yml") + + monkeypatch.setattr("serena.config.client_setup.execute_shell_command", fake_execute_shell_command) + monkeypatch.setattr("serena.cli.SerenaConfig.init", fake_init) + + result = CliRunner().invoke(TopLevelCommands.init, []) + + assert result.exit_code == 0, result.output + assert "serena setup grok" in result.output + assert "serena setup claude-code" not in result.output diff --git a/test/serena/test_hooks.py b/test/serena/test_hooks.py index d7287678..94bd23fb 100644 --- a/test/serena/test_hooks.py +++ b/test/serena/test_hooks.py @@ -11,6 +11,7 @@ from click.testing import CliRunner from serena.hooks import ( HookClient, PreToolUseAutoApproveSerenaHook, + PreToolUseHook, PreToolUseRemindAboutSymbolicToolsHook, SessionEndCleanupHook, hook_commands, @@ -35,6 +36,22 @@ def _base_input( } +def _grok_input( + tool_name: str, + tool_input: dict | None = None, + session_id: str = "grok-session-123", +) -> dict: + """Build a Grok PreToolUse payload using its real camelCase envelope.""" + return { + "hookEventName": "pre_tool_use", + "sessionId": session_id, + "toolName": tool_name, + "toolInput": tool_input if tool_input is not None else {}, + "toolInputTruncated": False, + "permissionMode": "bypassPermissions", + } + + def _read_input(tool_name: str = "read", session_id: str = "test-session-123", file_path: str = "src/foo.py") -> dict: """Build a hook payload for a read-style tool call against a file.""" return { @@ -44,6 +61,19 @@ def _read_input(tool_name: str = "read", session_id: str = "test-session-123", f } +def _execute_remind_hook(client: HookClient, payload: dict, tmp_path: Path) -> None: + with patch("sys.stdin", _make_stdin(payload)), patch("serena.hooks.serena_home_dir", str(tmp_path)): + PreToolUseRemindAboutSymbolicToolsHook(client).execute() + + +def _assert_grok_native_deny(output: str, reason_fragment: str) -> dict: + result = json.loads(output) + assert result["decision"] == "deny" + assert reason_fragment in result["reason"].lower() + assert "hookSpecificOutput" not in result + return result + + class TestHookClientDetection: """Tests for the --client option propagation.""" @@ -59,6 +89,12 @@ class TestHookClientDetection: hook = PreToolUseRemindAboutSymbolicToolsHook(HookClient.VSCODE) assert hook._client == HookClient.VSCODE + def test_grok_client(self, tmp_path: Path): + stdin_data = _grok_input("grep", {"pattern": "foo", "path": "."}) + with patch("sys.stdin", _make_stdin(stdin_data)), patch("serena.hooks.serena_home_dir", str(tmp_path)): + hook = PreToolUseRemindAboutSymbolicToolsHook(HookClient.GROK) + assert hook._client == HookClient.GROK + class TestPreToolUseRemindAboutSerenaHook: """Tests for the PreToolUse hook that nudges the agent toward symbolic tools.""" @@ -104,6 +140,37 @@ class TestPreToolUseRemindAboutSerenaHook: hook = PreToolUseRemindAboutSymbolicToolsHook(HookClient.CODEX) assert hook.is_grep_call() == expected, f"is_grep_tool() wrong for {tool_name} / {tool_input}" + def test_grep_tool_detection_grok(self, tmp_path: Path): + """Grok uses ``grep`` for its search tool and ``run_terminal_command`` for shell commands.""" + cases = [ + ("grep", {"pattern": "foo", "path": "."}, True), + ("grep_search", {"pattern": "foo", "path": "."}, False), + ("run_terminal_command", {"command": "rg -n foo README.md"}, True), + ("run_terminal_command", {"command": "cat README.md"}, False), + ] + for tool_name, tool_input, expected in cases: + with ( + patch("sys.stdin", _make_stdin(_grok_input(tool_name=tool_name, tool_input=tool_input))), + patch("serena.hooks.serena_home_dir", str(tmp_path)), + ): + hook = PreToolUseRemindAboutSymbolicToolsHook(HookClient.GROK) + assert hook.is_grep_call() == expected, f"is_grep_tool() wrong for {tool_name} / {tool_input}" + + def test_grok_camel_case_payload_detection(self, tmp_path: Path): + """Grok emits camelCase ``toolName`` / ``toolInput`` payload keys.""" + grep_payload = _grok_input("grep", {"pattern": "foo", "path": "."}) + with patch("sys.stdin", _make_stdin(grep_payload)), patch("serena.hooks.serena_home_dir", str(tmp_path)): + grep_hook = PreToolUseRemindAboutSymbolicToolsHook(HookClient.GROK) + assert grep_hook.is_grep_call() is True + assert grep_hook.is_read_file_call() is False + + read_payload = _grok_input("read_file", {"target_file": "src/foo.py"}) + with patch("sys.stdin", _make_stdin(read_payload)), patch("serena.hooks.serena_home_dir", str(tmp_path)): + read_hook = PreToolUseRemindAboutSymbolicToolsHook(HookClient.GROK) + assert read_hook.is_read_file_call() is True + assert read_hook.is_read_code_file_call() is True + assert read_hook.is_grep_call() is False + def test_read_file_tool_detection_claude_code(self, tmp_path: Path): """Claude Code uses the exact tool name ``Read`` (lowercased to ``read``).""" for name, expected in [("read", True), ("mcp__serena__read_file", True), ("grep", False), ("serena_search_for_pattern", False)]: @@ -177,12 +244,133 @@ class TestPreToolUseRemindAboutSerenaHook: hook = PreToolUseRemindAboutSymbolicToolsHook(HookClient.CODEX) assert hook.is_read_file_call() == expected, f"is_read_file_tool() wrong for {tool_name} / {tool_input}" + def test_read_file_tool_detection_grok(self, tmp_path: Path): + """Grok uses ``target_file`` for direct file reads and ``command`` for shell reads.""" + cases = [ + ("read_file", {"target_file": "src/foo.py"}, True, True), + ("read_file", {"target_file": "README.md"}, True, False), + ("run_terminal_command", {"command": "cat src/foo.py"}, True, True), + ("run_terminal_command", {"command": "cat README.md"}, True, False), + ("run_terminal_command", {"command": "rg -n foo README.md"}, False, False), + ] + for tool_name, tool_input, expected_read, expected_code_read in cases: + with ( + patch("sys.stdin", _make_stdin(_grok_input(tool_name=tool_name, tool_input=tool_input))), + patch("serena.hooks.serena_home_dir", str(tmp_path)), + ): + hook = PreToolUseRemindAboutSymbolicToolsHook(HookClient.GROK) + assert hook.is_read_file_call() == expected_read, f"is_read_file_tool() wrong for {tool_name} / {tool_input}" + assert hook.is_read_code_file_call() == expected_code_read, f"is_read_code_file_call() wrong for {tool_name} / {tool_input}" + + def test_grok_edit_and_list_payloads_do_not_count_as_search_or_read(self, tmp_path: Path, capsys: pytest.CaptureFixture[str]): + """Grok edit/list tools carry path fields but must not trigger the search/read reminder.""" + payloads = [ + _grok_input( + "search_replace", + {"file_path": "src/foo.py", "old_string": "old", "new_string": "new"}, + session_id="grok-edit-list", + ), + _grok_input("list_dir", {"target_directory": "."}, session_id="grok-edit-list"), + ] + for payload in payloads: + with patch("sys.stdin", _make_stdin(payload)), patch("serena.hooks.serena_home_dir", str(tmp_path)): + hook = PreToolUseRemindAboutSymbolicToolsHook(HookClient.GROK) + assert hook.is_grep_call() is False + assert hook.is_read_call() is False + assert hook.is_read_code_file_call() is False + + for _ in range(ToolUseCounter._NON_SYMBOLIC_USES_THRESHOLD): + for payload in payloads: + with patch("sys.stdin", _make_stdin(payload)), patch("serena.hooks.serena_home_dir", str(tmp_path)): + PreToolUseRemindAboutSymbolicToolsHook(HookClient.GROK).execute() + assert capsys.readouterr().out == "" + + @pytest.mark.parametrize( + ("payload", "expected_grep", "expected_read_file", "expected_read_code_file", "expected_shell"), + [ + ( + { + **_grok_input("run_terminal_command", session_id="grok-robust-string-input"), + "toolInput": "cat src/foo.py", + }, + False, + False, + False, + False, + ), + ( + { + key: value + for key, value in _grok_input("run_terminal_command", session_id="grok-robust-missing-input").items() + if key != "toolInput" + }, + False, + False, + False, + False, + ), + (_grok_input("run_terminal_command", {}, session_id="grok-robust-empty-input"), False, False, False, False), + ( + { + **_grok_input("grep", {"pattern": "foo", "path": "."}, session_id="grok-robust-truncated-input"), + "toolInputTruncated": True, + }, + True, + False, + False, + False, + ), + (_grok_input("run_terminal_command", {"command": 123}, session_id="grok-robust-numeric-command"), False, False, False, True), + ( + _grok_input("run_terminal_command", {"command": f"cat {'a' * 10000}.py"}, session_id="grok-robust-long-command"), + False, + True, + True, + True, + ), + ], + ids=[ + "string-tool-input", + "missing-tool-input", + "empty-tool-input", + "truncated-tool-input", + "numeric-command", + "long-command", + ], + ) + def test_grok_payload_robustness( + self, + payload: dict, + expected_grep: bool, + expected_read_file: bool, + expected_read_code_file: bool, + expected_shell: bool, + tmp_path: Path, + capsys: pytest.CaptureFixture[str], + ): + """Malformed or unusual Grok envelopes must not crash the hook.""" + with patch("sys.stdin", _make_stdin(payload)), patch("serena.hooks.serena_home_dir", str(tmp_path)): + hook = PreToolUseRemindAboutSymbolicToolsHook(HookClient.GROK) + assert hook.is_grep_call() is expected_grep + assert hook.is_read_file_call() is expected_read_file + assert hook.is_read_code_file_call() is expected_read_code_file + assert hook._is_shell_command_call() is expected_shell + hook.execute() + + assert capsys.readouterr().out == "" + def test_serena_tool_detection(self, tmp_path: Path): for name, expected in [("mcp_serena_find_symbol", True), ("serena_overview", True), ("grep_search", False)]: with patch("sys.stdin", _make_stdin(_base_input(tool_name=name))), patch("serena.hooks.serena_home_dir", str(tmp_path)): hook = PreToolUseRemindAboutSymbolicToolsHook(HookClient.CLAUDE_CODE) assert hook.is_serena_symbolic_tool() == expected, f"is_serena_tool() wrong for {name}" + def test_serena_tool_detection_grok_namespace(self, tmp_path: Path): + payload = _grok_input("serena__find_symbol") + with patch("sys.stdin", _make_stdin(payload)), patch("serena.hooks.serena_home_dir", str(tmp_path)): + hook = PreToolUseRemindAboutSymbolicToolsHook(HookClient.GROK) + assert hook.is_serena_symbolic_tool() is True + def test_no_output_below_threshold(self, tmp_path: Path, capsys: pytest.CaptureFixture[str]): """Below the threshold, the hook should produce no output (tool is allowed).""" for _ in range(ToolUseCounter._GREP_USES_THRESHOLD - 1): @@ -228,6 +416,92 @@ class TestPreToolUseRemindAboutSerenaHook: assert "additionalContext" not in hook_output assert "grep" in hook_output["permissionDecisionReason"].lower() + def test_deny_output_after_threshold_greps_grok(self, tmp_path: Path, capsys: pytest.CaptureFixture[str]): + """Grok expects native ``decision`` / ``reason`` JSON, not Claude-style hookSpecificOutput.""" + payload = _grok_input("grep", {"pattern": "foo", "path": "."}) + for _ in range(ToolUseCounter._GREP_USES_THRESHOLD): + with patch("sys.stdin", _make_stdin(payload)), patch("serena.hooks.serena_home_dir", str(tmp_path)): + PreToolUseRemindAboutSymbolicToolsHook(HookClient.GROK).execute() + + output = capsys.readouterr().out.strip() + result = json.loads(output) + assert result["decision"] == "deny" + assert "grep" in result["reason"].lower() + assert "hookSpecificOutput" not in result + + @pytest.mark.parametrize( + ("tool_name", "tool_input"), + [ + ("run_terminal_command", {"command": "cat src/foo.py"}), + ("read_file", {"target_file": "src/foo.py"}), + ], + ids=["shell-cat", "direct-read-file"], + ) + def test_deny_output_after_threshold_reads_grok( + self, + tool_name: str, + tool_input: dict, + tmp_path: Path, + capsys: pytest.CaptureFixture[str], + ): + """Grok code-read reminders are emitted in native ``decision`` / ``reason`` JSON.""" + payload = _grok_input(tool_name, tool_input, session_id=f"grok-read-{tool_name}") + for _ in range(ToolUseCounter._READ_FILE_USES_THRESHOLD): + _execute_remind_hook(HookClient.GROK, payload, tmp_path) + + _assert_grok_native_deny(capsys.readouterr().out.strip(), "read") + + _execute_remind_hook(HookClient.GROK, payload, tmp_path) + assert capsys.readouterr().out == "" + + def test_non_symbolic_deny_mixed_burst_grok(self, tmp_path: Path, capsys: pytest.CaptureFixture[str]): + """Alternating Grok grep/read calls trip the combined non-symbolic threshold only.""" + session_id = "grok-mixed-non-symbolic" + payloads = [ + _grok_input("grep", {"pattern": "foo", "path": "."}, session_id=session_id), + _grok_input("run_terminal_command", {"command": "cat src/foo.py"}, session_id=session_id), + _grok_input("grep", {"pattern": "bar", "path": "."}, session_id=session_id), + _grok_input("run_terminal_command", {"command": "cat src/bar.py"}, session_id=session_id), + ] + + for payload in payloads: + _execute_remind_hook(HookClient.GROK, payload, tmp_path) + + _assert_grok_native_deny(capsys.readouterr().out.strip(), "non-symbolic") + + _execute_remind_hook(HookClient.GROK, payloads[-1], tmp_path) + assert capsys.readouterr().out == "" + + def test_grok_native_allow_and_deny_output_json(self): + allow_output = PreToolUseHook.OutputData(permission_decision="allow", permission_decision_reason="allowed").to_json_string( + HookClient.GROK + ) + assert json.loads(allow_output) == {"decision": "allow"} + + deny_output = PreToolUseHook.OutputData(permission_decision="deny", permission_decision_reason="blocked").to_json_string( + HookClient.GROK + ) + assert json.loads(deny_output) == {"decision": "deny", "reason": "blocked"} + assert "hookSpecificOutput" not in json.loads(deny_output) + + def test_grok_serena_tool_resets_counters(self, tmp_path: Path, capsys: pytest.CaptureFixture[str]): + """Grok MCP tool names use ``serena__`` and must reset non-symbolic counters.""" + session_id = "grok-reset" + grep_payload = _grok_input("grep", {"pattern": "foo", "path": "."}, session_id=session_id) + serena_payload = _grok_input("serena__find_symbol", {"name_path": "Foo"}, session_id=session_id) + + for _ in range(ToolUseCounter._GREP_USES_THRESHOLD - 1): + with patch("sys.stdin", _make_stdin(grep_payload)), patch("serena.hooks.serena_home_dir", str(tmp_path)): + PreToolUseRemindAboutSymbolicToolsHook(HookClient.GROK).execute() + + with patch("sys.stdin", _make_stdin(serena_payload)), patch("serena.hooks.serena_home_dir", str(tmp_path)): + PreToolUseRemindAboutSymbolicToolsHook(HookClient.GROK).execute() + + with patch("sys.stdin", _make_stdin(grep_payload)), patch("serena.hooks.serena_home_dir", str(tmp_path)): + PreToolUseRemindAboutSymbolicToolsHook(HookClient.GROK).execute() + + assert capsys.readouterr().out == "" + def test_deny_output_after_threshold_reads(self, tmp_path: Path, capsys: pytest.CaptureFixture[str]): """After reaching the read file threshold, the hook should output a deny. @@ -693,6 +967,19 @@ class TestHookCli: assert result.exit_code == 0 assert not session_dir.exists() + def test_cleanup_command_grok_camelcase_session(self, tmp_path: Path): + session_id = "cli-grok-cleanup" + session_dir = tmp_path / "hook_data" / session_id + session_dir.mkdir(parents=True) + (session_dir / "somefile").write_text("data") + + stdin_json = json.dumps({"sessionId": session_id, "hookEventName": "stop"}) + runner = CliRunner() + with patch("serena.hooks.serena_home_dir", str(tmp_path)): + result = runner.invoke(hook_commands, ["cleanup", "--client", "grok"], input=stdin_json) + assert result.exit_code == 0 + assert not session_dir.exists() + def test_remind_command(self, tmp_path: Path): """Invoke the remind command enough times to trigger a deny.""" runner = CliRunner() @@ -705,6 +992,20 @@ class TestHookCli: output = json.loads(result.output) assert output["hookSpecificOutput"]["permissionDecision"] == "deny" + def test_remind_command_grok_emits_native_deny(self, tmp_path: Path): + """The documented Grok remind CLI emits Grok-native deny JSON.""" + runner = CliRunner() + payload = _grok_input("grep", {"pattern": "foo", "path": "."}, session_id="cli-grok-remind") + for _ in range(ToolUseCounter._GREP_USES_THRESHOLD): + with patch("serena.hooks.serena_home_dir", str(tmp_path)): + result = runner.invoke(hook_commands, ["remind", "--client", "grok"], input=json.dumps(payload)) + assert result.exit_code == 0 + + output = json.loads(result.output) + assert output["decision"] == "deny" + assert "grep" in output["reason"].lower() + assert "hookSpecificOutput" not in output + def test_auto_approve_command(self, tmp_path: Path): """The ``auto-approve`` CLI command emits an allow for a Serena tool in acceptEdits mode.""" stdin_json = json.dumps( @@ -755,6 +1056,60 @@ class TestHookCli: assert result.exit_code == 0 assert result.output == "" + def test_auto_approve_command_grok_uses_native_output(self, tmp_path: Path): + """The ``auto-approve`` CLI command accepts ``--client=grok`` and emits Grok-native JSON.""" + stdin_json = json.dumps( + { + "session_id": "cli-auto-approve-grok", + "toolName": "serena__find_symbol", + "toolInput": {}, + "permissionMode": "auto", + } + ) + runner = CliRunner() + with patch("serena.hooks.serena_home_dir", str(tmp_path)): + result = runner.invoke(hook_commands, ["auto-approve", "--client", "grok"], input=stdin_json) + assert result.exit_code == 0 + output = json.loads(result.output) + assert output == {"decision": "allow"} + + @pytest.mark.parametrize( + ("tool_name", "permission_mode", "expected_output"), + [ + ("serena__find_symbol", "auto", {"decision": "allow"}), + ("serena__find_symbol", "bypassPermissions", None), + ("serena__find_symbol", "dontAsk", None), + ("serena__find_symbol", "", None), + ("grep", "auto", None), + ], + ids=["auto-allows-serena", "bypass-silent", "dont-ask-silent", "empty-mode-silent", "non-serena-silent"], + ) + def test_auto_approve_command_grok_permission_modes( + self, + tool_name: str, + permission_mode: str, + expected_output: dict | None, + tmp_path: Path, + ): + """Grok auto-approve is intentionally active only for Serena tools in ``auto`` mode.""" + stdin_json = json.dumps( + { + "sessionId": f"cli-grok-auto-approve-{tool_name}-{permission_mode}", + "toolName": tool_name, + "toolInput": {}, + "permissionMode": permission_mode, + } + ) + runner = CliRunner() + with patch("serena.hooks.serena_home_dir", str(tmp_path)): + result = runner.invoke(hook_commands, ["auto-approve", "--client", "grok"], input=stdin_json) + + assert result.exit_code == 0 + if expected_output is None: + assert result.output == "" + else: + assert json.loads(result.output) == expected_output + def test_client_default_is_claude_code(self, tmp_path: Path): """When --client is omitted, it defaults to claude-code.""" stdin_json = json.dumps({"session_id": "cli-default"}) diff --git a/test/serena/test_serena_agent.py b/test/serena/test_serena_agent.py index 1199ebd3..e373fb8a 100644 --- a/test/serena/test_serena_agent.py +++ b/test/serena/test_serena_agent.py @@ -13,6 +13,7 @@ import pytest from _pytest.mark import Mark, MarkDecorator, ParameterSet from serena.agent import SerenaAgent +from serena.config.context_mode import SerenaAgentContext from serena.config.serena_config import ProjectConfig, RegisteredProject, SerenaConfig from serena.project import Project from serena.tools import ( @@ -39,6 +40,7 @@ from test.conftest import ( get_repo_path, language_tests_enabled, ) +from test.serena.config.test_context_mode import GROK_EXCLUDED_TOOLS @dataclass @@ -877,6 +879,25 @@ class TestSerenaAgent: serena_config = SerenaConfig().with_headless_mode_overrides() SerenaAgent(project=project, serena_config=serena_config) + @pytest.mark.python + @pytest.mark.skipif(not language_tests_enabled(Language.PYTHON), reason="python tests are disabled in this environment") + def test_grok_context_restricts_toolset_and_prompt(self, serena_config): + agent = SerenaAgent( + project="test_repo_python", + serena_config=serena_config, + context=SerenaAgentContext.from_name("grok"), + ) + agent.execute_task(lambda: None) + + try: + exposed = {tool.get_name() for tool in agent.get_exposed_tool_instances()} + assert exposed.isdisjoint(GROK_EXCLUDED_TOOLS) + assert "activate_project" not in exposed + assert {"find_symbol", "get_symbols_overview", "replace_symbol_body"} <= exposed + assert "Serena's code intelligence tools" in agent.create_system_prompt() + finally: + agent.on_shutdown(timeout=5) + def _symbol_matches_expected_name(self, symbol: dict, expected_name: str) -> bool: return ( symbol.get("name") == expected_name