diff --git a/README.md b/README.md index 2e7a85e..031a986 100644 --- a/README.md +++ b/README.md @@ -158,7 +158,7 @@ Hooks: - `.codex/hooks/stop.py` MCP: -- `.codex/mcp/server.py` +- registered with Codex as `ix-memory`, served by the Ix CLI (`ix mcp`) — no file is installed ### Home-local install @@ -172,7 +172,6 @@ This writes: - `~/.codex/config.toml` - `~/.codex/hooks.json` - `~/.codex/hooks/*.py` -- `~/.codex/mcp/server.py` ### Repo-local install @@ -186,7 +185,6 @@ This writes: - `/path/to/project/.codex/config.toml` - `/path/to/project/.codex/hooks.json` - `/path/to/project/.codex/hooks/*.py` -- `/path/to/project/.codex/mcp/server.py` ### Symlink mode for local development diff --git a/install.sh b/install.sh index c4bc7d1..60c7e9a 100755 --- a/install.sh +++ b/install.sh @@ -22,7 +22,7 @@ Examples: Flags: --plugin Copy/register the ix-memory Codex plugin in a local marketplace --hooks Install the .codex hook bundle (session, prompt, pre/post tool, stop) - --mcp Install the ix-memory MCP server and print the codex mcp add command + --mcp Register the Ix CLI's MCP server (`ix mcp`) with Codex Notes: - If none of --plugin, --hooks, or --mcp is passed, the installer defaults to diff --git a/mcp/ix_llm.py b/mcp/ix_llm.py deleted file mode 100644 index 8286c74..0000000 --- a/mcp/ix_llm.py +++ /dev/null @@ -1,236 +0,0 @@ -"""ix `--format llm` fast-path, gated on the installed CLI's version. - -Every read tool in `server.py` calls `_json()` and hands the result to `_ok()`, -which re-serialises it with `indent=2`. So the JSON is parsed only to be printed -straight back to the model — and printed *larger* than it arrived. That is -exactly the overhead `--format llm` exists to remove: a token-minimal, -newline-delimited rendering, typically 2-4x smaller than `json` on tree- and -table-shaped output. - -The port target is ix-cursor-plugin's `mcp/lib/llm.ts`, which has run this -design since v0.7.0. Two things had to change for this surface. - -## Why the floor is per-command, not global - -`--format llm` did not arrive all at once, and the cursor plugin's single -`MIN_LLM_VERSION = 0.7.0` is correct only because none of its twelve tools -touches a command that landed later. This server exposes two that do. - -| commands | renderer landed in | -|---|---| -| Tier 1-4 — `map` `subsystems` `impact` `smells` `overview` `stats` `inventory` `rank` `depends` `trace` `callers` `callees` `imports` `imported-by` `text` `history` `locate` `diff` | **v0.7.0** | -| Tier 5 — `explain` `read` | **v0.9.2** | - -## Why a wrong floor fails silently - -`ix` does not validate `--format`. Every renderer is -`if json: ... elif llm: ... else: text`, so an unrecognised value falls through -to **human text** and exits 0 (verified against `formatPatches` and its -siblings). An old CLI therefore answers `--format llm` with a rendered table -rather than an error — no exception to catch, no non-zero exit, just prose where -records were expected. That is the failure this table prevents, and it is why -the floors are stated per command instead of being probed. - -The same property is what makes the whole change safe: there is no version of -`ix` on which asking for `llm` breaks. - -## Pro commands are excluded outright - -`briefing` and `decisions` come from `@ix/pro`, whose commands declare only -`text|json` — there is no `llm` renderer at any version. They are absent from -the table below and must stay absent: a version gate cannot help, because no -future release of the *OSS* CLI adds them. - -## Invariants - -Every fall-through returns None so the caller runs its unchanged JSON path, -which keeps behaviour byte-identical to before this module existed: - - - the command is not in the table, or the CLI is older than its floor - - the CLI version cannot be determined - - any error, timeout, or empty output - - `IX_DISABLE_LLM_FORMAT=1` (kill switch) - -This module never *parses* llm output. It forwards it. -""" - -from __future__ import annotations - -import os -import re -from typing import Callable, Optional - -# command -> (major, minor, patch) of the release whose renderer it needs. -# Derived from docs/llm-format.md in the Ix repo and confirmed against the tags -# that contain each tier's commit. -LLM_MIN_VERSION: dict[str, tuple[int, int, int]] = { - # Tier 1 - # `map` and `smells` are absent on purpose: neither goes through _read. - # ix_map ingests and keeps _run; ix_smells filters client-side on parsed - # candidates and so needs records, not prose. Listing a command that cannot - # consult the table reads as support that was considered and granted. - "subsystems": (0, 7, 0), - "impact": (0, 7, 0), - "overview": (0, 7, 0), - "stats": (0, 7, 0), - # Tier 2 - "inventory": (0, 7, 0), - "rank": (0, 7, 0), - "depends": (0, 7, 0), - "trace": (0, 7, 0), - "callers": (0, 7, 0), - "callees": (0, 7, 0), - "imports": (0, 7, 0), - "imported-by": (0, 7, 0), - # Tier 3 - "text": (0, 7, 0), - "history": (0, 7, 0), - # Tier 4 - "locate": (0, 7, 0), - # `diff` is deliberately absent, not merely un-versioned. Its renderer has - # branches with no llm arm at any version -- the textual-changes path - # (diff.ts: graph reports no change but the file text differs) drops to the - # `else` and prints " modified ( textual changes -- not captured by - # parser)" at exit 0. `ix_diff` reaches it with the arguments it already - # sends, and prose at exit 0 is exactly what this module forwards as records. - # A version floor cannot fix a per-code-path gap: there is no release that - # makes it right, so the fast-path must never ask. - # Tier 5 — these are the reason the table is per-command. - "explain": (0, 9, 2), - "read": (0, 9, 2), -} - -SemVer = tuple[int, int, int] - -# Decoration is fine; ambiguity is not. `ix --version` prints a bare number -# today and could grow a suffix, so "ix 0.9.2 (linux-amd64)" must still parse. -# But if the output carries more than one version-shaped token there is no -# way to tell which one is ix's -- "node v20.11.0 / ix 0.7.0" would otherwise -# gate on node's. Reading the wrong version is the one error that can turn -# the fast-path on for a CLI that renders prose, so that case refuses. -_SEMVER_RE = re.compile(r"v?(\d+)\.(\d+)\.(\d+)") - -# Process-lifetime memo. A server probes the CLI version at most once; the -# probe is deliberately not cached to disk so the gate stays deterministic and -# resettable in tests. -_version_cache: Optional[SemVer] = None -_version_probed = False - - -def reset_version_cache() -> None: - """Forget the probed version. For tests.""" - global _version_cache, _version_probed - _version_cache = None - _version_probed = False - - -def parse_semver(value: str) -> Optional[SemVer]: - matches = _SEMVER_RE.findall(value or "") - if len(matches) != 1: - return None - major, minor, patch = matches[0] - return (int(major), int(minor), int(patch)) - - -def llm_disabled() -> bool: - return (os.environ.get("IX_DISABLE_LLM_FORMAT") or "").strip().lower() in {"1", "true", "yes"} - - -def detect_version(run: Callable[..., tuple[bool, str, str]]) -> Optional[SemVer]: - """Probe `ix --version`, memoised for the life of the process. - - A generous timeout on purpose: `ix` kicks off a non-awaited update check - that delays process exit, and the probe resolves on exit. A tight timeout - would fail the probe intermittently and silently disable the fast-path for - the rest of the session. Failing to determine a version disables it too, so - the cost of being wrong here is only lost savings, never wrong output. - """ - global _version_cache, _version_probed - if _version_probed: - return _version_cache - - _version_probed = True - try: - ok, stdout, _ = run(["--version"], timeout=20) - except Exception: - _version_cache = None - return None - - _version_cache = parse_semver(stdout.strip()) if ok else None - return _version_cache - - -# Flag combinations that route to text even on a current CLI, documented as -# deliberate exceptions in docs/llm-format.md: `diff --content` emits verbatim -# hunks, which have no record form. `ix_diff` does not pass `--content` today, -# so this guards a future edit rather than a live case — but the failure it -# prevents is the silent kind, text forwarded as though it were records. -_TEXT_ONLY_FLAGS: dict[str, frozenset[str]] = { - "diff": frozenset({"--content"}), -} - - -def supports_llm(command: str, run: Callable[..., tuple[bool, str, str]], args: Optional[list[str]] = None) -> bool: - if llm_disabled(): - return False - floor = LLM_MIN_VERSION.get(command) - if floor is None: - return False - if args and (blocked := _TEXT_ONLY_FLAGS.get(command)): - # Prefix match, not set intersection: `--content=x` is the same flag as - # `--content x` and an exact-token check misses it. This guard exists to - # survive a future edit, so the spelling that edit might use is exactly - # the one it has to catch. - if any( - arg == flag or arg.startswith(flag + "=") - for arg in args - for flag in blocked - ): - return False - version = detect_version(run) - return version is not None and version >= floor - - -def is_llm_error_line(text: str) -> bool: - """`ix` reports some failures as a record on stdout, with exit 0. - - `error code= message="..."` is part of the llm format by design. A - caller that only checked the exit status would forward that line to the - model as a successful result, so it is detected here and deferred to the - JSON path, where the error is formatted exactly as it always was. No success - record for any command in the table begins with `error code=`. - """ - return text.lstrip().startswith("error code=") - - -def try_llm( - args: list[str], - run: Callable[..., tuple[bool, str, str]], - timeout: int = 15, -) -> Optional[str]: - """Return `--format llm` text for `args`, or None to use the JSON path. - - `args[0]` is the ix subcommand; the gate is keyed on it. - """ - if not args: - return None - if not supports_llm(args[0], run, args): - return None - - try: - ok, stdout, _ = run([*args, "--format", "llm"], timeout=timeout) - except Exception: - return None - if not ok: - return None - - if not (stdout or "").strip(): - return None - # Exactly one trailing newline, not .strip(). `read` emits - # `content lines=` followed by n raw source lines, so stripping trailing - # whitespace deletes blank final lines and leaves the count over-reporting - # what follows it. Leading whitespace can be meaningful for the same reason. - text = re.sub(r"\r?\n\Z", "", stdout) - if is_llm_error_line(text): - return None - return text diff --git a/mcp/server.py b/mcp/server.py deleted file mode 100644 index 28aaafb..0000000 --- a/mcp/server.py +++ /dev/null @@ -1,472 +0,0 @@ -#!/usr/bin/env python3 -"""ix-memory MCP server — stdio transport, MCP Python SDK. - -23 tools mirroring the Cursor plugin tool set. All tools call the ix CLI -directly (same fallback strategy as the other hooks). Future work will -migrate each to call_runtime() once the v2 runtime API is live. - -Registration (run once after install): - codex mcp add ix-memory -- python3 /path/to/.codex/mcp/server.py -""" -from __future__ import annotations - -import json -import os -import shutil -import subprocess -import threading -from typing import Optional - -try: - from ix_llm import try_llm -except ImportError: # pragma: no cover - degraded but functional - # `ix_llm.py` is installed beside this file by install_mcp(). If it is - # absent — a hand-copied server.py, a partial install, an older layout — - # every read tool falls back to the JSON path it used before the fast-path - # existed. An ImportError here would instead take down all 23 tools to save - # tokens on some of them, which is not a trade worth making. - def try_llm(*_args: object, **_kwargs: object) -> Optional[str]: - return None - -# The MCP Python SDK renamed FastMCP to MCPServer in 2.0.0 and moved it from -# mcp.server.fastmcp to mcp.server.mcpserver. Nothing in this repo installs or -# pins the SDK — the installer copies this file and prints a `codex mcp add` -# line — so whichever version pip last resolved is what this has to run on, and -# both lines are in the wild. On a fresh `pip install mcp` the v1 import raises -# ModuleNotFoundError, and all Codex surfaces is that the ix-memory client -# "failed to start". -# -# Only the constructor moved. Everything below is unchanged across the two: -# @mcp.tool() keeps its name and signature, the tools stay plain functions -# returning str, and run(transport="stdio") is the same call. None of them use -# the get_context() that v2 removed, so there is nothing to adapt per tool. -try: # mcp >= 2.0.0 - from mcp.server.mcpserver import MCPServer as _Server -except ImportError: # mcp < 2.0.0 - try: - from mcp.server.fastmcp import FastMCP as _Server - except ImportError as exc: # the SDK is missing outright, not merely older - raise SystemExit( - "ix-memory MCP server requires the MCP Python SDK, which is not " - "installed for this interpreter.\n" - " pip install mcp" - ) from exc - -mcp = _Server("ix-memory") - - -# ── Helpers ─────────────────────────────────────────────────────────────────── - -# Resolving the executable is necessary on Windows (see _run) but it is not free: -# npm ships no `ix.exe`, so shutil.which returns `ix.CMD`, and CreateProcess runs -# a .cmd/.bat by handing the command line to `cmd.exe /c`. Every argument is then -# parsed by a shell before the CLI sees it -- and subprocess only quotes -# arguments containing whitespace, so `Widget&whoami` splits into two commands -# while `Widget & whoami` does not. `%VAR%` is expanded either way, and a literal -# `"` escapes the quoting that protects the rest. -# -# These arguments come from tool calls the model makes, so the content is not -# trusted. Quoting correctly for cmd.exe is a well-known trap -- it is what -# CVE-2024-24576 was -- so this refuses the input instead of trying to escape it. -# The cost is that a symbol named with one of these cannot be queried on Windows, -# which beats running it. -_CMD_SHIM_SUFFIXES = (".cmd", ".bat") -# `!` is here for shims that enable delayed expansion: it cannot split a command -# (expansion happens after tokenisation) but it does substitute an environment -# value into the query, which is the same disclosure `%VAR%` gives. -_CMD_METACHARACTERS = frozenset('&|<>^"%!\r\n') - - -def _unsafe_for_cmd_shim(executable: str, args: list[str]) -> str: - """The cmd.exe metacharacters in `args`, if this executable routes through one.""" - if not executable.lower().endswith(_CMD_SHIM_SUFFIXES): - return "" - found: set[str] = set() - # The executable too: list2cmdline leaves an unquoted path alone, so an `ix` - # installed under `C:\Users\a&b\` splits at the `&`. Not attacker-controlled, - # but it fails silently and is one line to catch. - for arg in (executable, *args): - found |= set(arg) & _CMD_METACHARACTERS - return " ".join(repr(c) for c in sorted(found)) - - -# Why the reason a call failed is stashed rather than returned: _json hands its -# 21 callers bare data, so stderr had nowhere to go and every failure read -# "failed without output" -- including "ix CLI not found" and the refusal above, -# the two a user most needs to see. Threading a second value through 21 call -# sites is the tidier fix and a much larger diff; this is the small one. -# -# It is thread-local and written by *every* _run, including the early returns. -# A module-level string would be wrong twice over: mcp >= 2.0.0 dispatches each -# message with anyio.to_thread.run_sync, so two tools really can be in flight at -# once; and even single-threaded, ix_map and ix_health call _run directly, so a -# stale value from an earlier tool's _json would be reported under theirs -- -# which for ix_read means one tool's file content surfacing in another's error. -_call_state = threading.local() - - -def _record_stderr(stderr: str) -> None: - _call_state.stderr = stderr - - -def _recorded_stderr() -> str: - return getattr(_call_state, "stderr", "") - - -def _run(args: list[str], timeout: int = 15) -> tuple[bool, str, str]: - """Run `ix` with the given subcommand and flags. - - The executable is prepended here rather than written out at each of the 23 - tools, because that is exactly what went wrong: every tool but ix_health - passed only the subcommand, so Python tried to execute programs named - `stats`, `locate` and `map`. One tool spelling it correctly is what let the - server look alive while nothing else in it worked. - - It is resolved with shutil.which rather than passed as the bare string "ix". - On Windows the installed CLI is `ix.CMD`, and CreateProcess does not consult - PATHEXT the way the shell does -- `subprocess.run(["ix", ...])` raises - FileNotFoundError there however well-formed the rest of the argv is. Getting - the argv right and still naming the executable in a way Windows cannot - resolve would have left all 23 tools returning an error on the platform this - is meant to fix. - """ - # shutil.which searches the *current directory first* on Windows unless - # NoDefaultCurrentDirectoryInExePath is set, which it is not by default -- so - # a repository shipping an `ix.bat` at its root would own this server - # outright, no metacharacters required. Passing `path=` does NOT prevent it: - # CPython inserts os.curdir whenever the command has no directory part, - # explicit path or not (verified on 3.13 -- it still returns `.\ix.BAT`). - # - # What does prevent it is refusing the result: a PATH hit is absolute, the - # curdir hit is not. - executable = shutil.which("ix", path=os.environ.get("PATH")) - if executable is not None and not os.path.isabs(executable): - _record_stderr( - msg := ( - f"refusing to run {executable!r}: resolved from the working " - "directory rather than PATH. Remove it, or put the real ix ahead " - "of it on PATH." - ) - ) - return False, "", msg - if executable is None: - _record_stderr(msg := "ix CLI not found. Install it and ensure it is on PATH.") - return False, "", msg - unsafe = _unsafe_for_cmd_shim(executable, args) - if unsafe: - _record_stderr( - msg := ( - f"refusing to run `ix {args[0] if args else ''}`: an argument " - f"contains {unsafe}, which the Windows command processor would act " - "on rather than pass to the CLI. Re-run without those characters " - "-- for a symbol, name it without them; for a search, simplify the " - "pattern." - ) - ) - return False, "", msg - try: - r = subprocess.run( - [executable, *args], - capture_output=True, - text=True, - timeout=timeout, - check=False, - ) - _record_stderr(r.stderr) - return r.returncode == 0, r.stdout, r.stderr - # ValueError too: an embedded NUL raises it out of subprocess on every - # platform, and it is not an OSError. - except (OSError, ValueError, subprocess.SubprocessError) as exc: - _record_stderr(detail := str(exc)) - return False, "", detail - - -def _parse(text: str) -> object: - text = text.strip() - for i, ch in enumerate(text): - if ch in "{[": - try: - return json.loads(text[i:]) - except json.JSONDecodeError: - pass - return None - - -def _json(args: list[str], timeout: int = 15) -> object: - """Run an ix query and parse its output as JSON. - - `--format json` is requested rather than assumed. This parsed stdout as JSON - while asking for nothing, so it was reading the human-oriented text output -- - _parse would find the first `{` or `[` in a rendered table and either fail or, - worse, succeed on a fragment. Every subcommand reached from here accepts the - flag; the ones that do not take --format at all (config, init, reset, upgrade, - view, watch) are not exposed as tools. - """ - ok, stdout, _ = _run([*args, "--format", "json"], timeout) - return _parse(stdout) if ok else None - - -def _ok(data: object) -> str: - return json.dumps(data, indent=2) if data is not None else json.dumps({}) - - -def _err(tool: str, cmd: str, stderr: str = "") -> str: - detail = (stderr or _recorded_stderr()).strip() - msg = f"{cmd} failed: {detail}" if detail else f"{cmd} failed without output" - return json.dumps({"error": msg, "tool": tool}) - - -def _read(tool: str, cmd: str, args: list[str], timeout: int = 15) -> str: - """Run a read command, preferring `--format llm` where the CLI supports it. - - These tools parsed JSON only to hand it straight back to the model through - `_ok`, which re-serialises it with `indent=2` — so the round trip made the - payload *larger* than it arrived. `--format llm` is the same content - rendered 2-4x smaller. - - The fast-path is gated per command on the installed CLI's version, because - `--format llm` landed tier by tier and an older `ix` answers it with human - text rather than an error (see mcp/ix_llm.py). Anything that is not a - confident success — unsupported command, old CLI, failed probe, empty - output, an `error code=` record — returns None there and lands on the JSON - path below, unchanged. - - Pro commands do come through here -- `ix_decisions` calls this -- and are - kept off the fast-path by their absence from the version table, not by - avoiding this helper. `@ix/pro` ships no llm renderer at any version, so - the omission is permanent rather than a floor waiting to be met. - """ - fast = try_llm(args, _run, timeout=timeout) - if fast is not None: - return fast - data = _json(args, timeout) - if data is None: - return _err(tool, cmd) - return _ok(data) - - -# ── Tools ───────────────────────────────────────────────────────────────────── - -@mcp.tool() -def ix_health() -> str: - """Check whether the ix CLI is available and the graph is ready.""" - if not shutil.which("ix"): - return json.dumps({"error": "ix CLI not found. Install it and ensure it is on PATH.", "graph_ready": False}) - # No "ix" here any more: _run prepends it. This tool was the only one that - # passed it, which is why it was the only one that worked. - ok, stdout, stderr = _run(["--version"], timeout=5) - if not ok: - return json.dumps({"error": f"ix health probe failed: {stderr.strip()}", "graph_ready": False}) - version = stdout.strip().split()[0] if stdout.strip() else "unknown" - return json.dumps({"version": version, "graph_ready": True}) - - -@mcp.tool() -def ix_briefing() -> str: - """Load the ix Pro session briefing for current goals, plans, and recent decisions.""" - data = _json(["briefing"]) - if data is None: - return json.dumps({"error": "ix briefing failed or requires ix Pro"}) - return _ok(data) - - -@mcp.tool() -def ix_locate(symbol: str) -> str: - """Resolve a symbol to its canonical graph-backed target.""" - return _read("ix_locate", f"ix locate {symbol}", ["locate", symbol]) - - -@mcp.tool() -def ix_text( - pattern: str, - limit: int = 20, - path: Optional[str] = None, - language: Optional[str] = None, -) -> str: - """Search text across the indexed repository and return ranked hits.""" - args = ["text", pattern, "--limit", str(limit)] - if path: - args += ["--path", path] - if language: - args += ["--language", language] - return _read("ix_text", f"ix text {pattern}", args) - - -@mcp.tool() -def ix_impact(target: str) -> str: - """Analyze the blast radius of a symbol or file — risk level and what depends on it. - - Field names are deliberately not promised here. On a current CLI this returns - the compact llm rendering, which spells risk as `risk=` and omits a - recommended-action field entirely; on an older one it returns the JSON - object with `risk_level`/`dependents`/`recommended_action`. A docstring is - the model's contract, so naming keys only one of the two paths produces is a - promise this tool cannot keep. - """ - return _read("ix_impact", f"ix impact {target}", ["impact", target]) - - -@mcp.tool() -def ix_map(file: Optional[str] = None) -> str: - """Ingest a file into the graph (ix map ) or run a full architecture map (ix map).""" - # --format json like every other tool. Without it this asked for the rendered - # view -- which for `map` is explicitly truncated (--max-items defaults to 10) - # -- and then ran _parse over the ASCII art, so it reliably fell through to - # {"raw": ...} and handed the model unstructured, silently partial text. It - # was the last tool still doing what #14 was filed about. - args = ["map", file, "--format", "json"] if file else ["map", "--format", "json"] - ok, stdout, stderr = _run(args, timeout=60) - if not ok: - return _err("ix_map", f"ix map{' ' + file if file else ''}", stderr) - return _ok(_parse(stdout) or {"raw": stdout.strip()}) - - -@mcp.tool() -def ix_overview(target: str) -> str: - """Return a structural overview of a symbol or file — its shape and where it sits. - - The llm rendering is a compressed record form; it does not reproduce every - key the JSON object carries. - """ - return _read("ix_overview", f"ix overview {target}", ["overview", target]) - - -@mcp.tool() -def ix_read(symbol: str) -> str: - """Read the source content of a symbol via the graph (bounds raw file reads to graph-known symbols).""" - return _read("ix_read", f"ix read {symbol}", ["read", symbol]) - - -@mcp.tool() -def ix_diff( - from_rev: int, - to_rev: int, - target: Optional[str] = None, - summary: bool = False, -) -> str: - """Show the structural diff between two graph revisions, optionally scoped to a file or symbol.""" - args = ["diff", str(from_rev), str(to_rev)] - if target: - args.append(target) - if summary: - args.append("--summary") - return _read("ix_diff", f"ix diff {from_rev}..{to_rev}", args) - - -@mcp.tool() -def ix_callers(symbol: str) -> str: - """List entities that call a symbol (incoming call edges).""" - return _read("ix_callers", f"ix callers {symbol}", ["callers", symbol]) - - -@mcp.tool() -def ix_callees(symbol: str) -> str: - """List entities called by a symbol (outgoing call edges).""" - return _read("ix_callees", f"ix callees {symbol}", ["callees", symbol]) - - -@mcp.tool() -def ix_imported_by(symbol: str) -> str: - """List files or symbols that import a given symbol (incoming import edges).""" - return _read("ix_imported_by", f"ix imported-by {symbol}", ["imported-by", symbol]) - - -@mcp.tool() -def ix_imports(symbol: str) -> str: - """List symbols or files imported by a given symbol (outgoing import edges).""" - return _read("ix_imports", f"ix imports {symbol}", ["imports", symbol]) - - -@mcp.tool() -def ix_depends(symbol: str, depth: int = 2) -> str: - """Show the downstream dependency graph for a symbol up to a given depth (default 2).""" - return _read("ix_depends", f"ix depends {symbol}", ["depends", symbol, "--depth", str(depth)]) - - -@mcp.tool() -def ix_trace(symbol: str, to: Optional[str] = None) -> str: - """Trace execution paths through a symbol — upstream callers and downstream callees.""" - args = ["trace", symbol] - if to: - args += ["--to", to] - return _read("ix_trace", f"ix trace {symbol}", args) - - -@mcp.tool() -def ix_explain(symbol: str) -> str: - """Explain a symbol's role, importance, and callers using graph data. - - Callees are reported as a count rather than a list: the llm rendering emits - `edges callees=`, and the names are only available under `--raw`, which - this does not pass. - """ - return _read("ix_explain", f"ix explain {symbol}", ["explain", symbol]) - - -@mcp.tool() -def ix_rank( - by: str = "dependents", - kind: str = "class", - top: int = 10, - path: Optional[str] = None, -) -> str: - """Rank symbols by a quality metric (dependents, callers, importers, members) to surface hotspots.""" - args = ["rank", "--by", by, "--kind", kind, "--top", str(top)] - if path: - args += ["--path", path] - return _read("ix_rank", "ix rank", args) - - -@mcp.tool() -def ix_inventory(path: str, kind: str = "file") -> str: - """List files or symbols within a repository path scope.""" - return _read("ix_inventory", f"ix inventory {path}", ["inventory", "--kind", kind, "--path", path]) - - -@mcp.tool() -def ix_smells(path: Optional[str] = None, limit: int = 50) -> str: - """Detect code quality smells across the graph — orphan files, high coupling, etc.""" - args = ["smells"] - if path: - args += ["--path", path] - data = _json(args) - if data is None: - return _err("ix_smells", "ix smells") - if isinstance(data, dict) and isinstance(data.get("candidates"), list): - data["candidates"] = data["candidates"][:limit] - return _ok(data) - - -@mcp.tool() -def ix_stats() -> str: - """Return graph-wide ix statistics. - - The llm rendering reports node and edge counts and drops zero-valued - categories, so this is a summary rather than the full JSON breakdown. - """ - return _read("ix_stats", "ix stats", ["stats"]) - - -@mcp.tool() -def ix_subsystems() -> str: - """List graph-derived subsystems for top-level repository orientation.""" - return _read("ix_subsystems", "ix subsystems", ["subsystems"]) - - -@mcp.tool() -def ix_decisions(path: Optional[str] = None) -> str: - """List architecture decisions recorded in the graph, optionally scoped to a path.""" - args = ["decisions"] - if path: - args += ["--path", path] - return _read("ix_decisions", "ix decisions", args) - - -@mcp.tool() -def ix_history(target: str) -> str: - """Show the provenance/patch history for a file or symbol.""" - return _read("ix_history", f"ix history {target}", ["history", target]) - - -if __name__ == "__main__": - mcp.run(transport="stdio") diff --git a/scripts/install_codex_integration.py b/scripts/install_codex_integration.py index 00e0093..34b0e50 100644 --- a/scripts/install_codex_integration.py +++ b/scripts/install_codex_integration.py @@ -63,7 +63,7 @@ def parse_args() -> argparse.Namespace: parser.add_argument( "--mcp", action="store_true", - help="Install the ix-memory MCP server and print the codex mcp add registration command", + help="Register the Ix CLI's MCP server (`ix mcp`) with Codex", ) parser.add_argument( "--mode", @@ -784,19 +784,52 @@ def install_plugin( return installed +MIN_IX_VERSION_FOR_MCP = (0, 9, 3) + + +def _ix_executable() -> str | None: + """The `ix` on PATH, as an absolute path, or None if there is none. + + Resolved rather than spawned by name because this script runs on Windows, + where npm ships no `ix.exe` — only `ix.CMD` — and CreateProcess consults no + PATHEXT, so `subprocess.run(["ix", ...])` raises FileNotFoundError however + well-formed the rest of the argv is. `shutil.which` DOES apply PATHEXT, so it + finds the shim. This is Ix#383's inner half, which `.codex/hooks/common.py` + already fixes for the hooks; every new `ix` call site has to do the same. + """ + executable = shutil.which("ix", path=os.environ.get("PATH")) + if executable is None or not os.path.isabs(executable): + return None + return executable + + +def _ix_version(executable: str) -> tuple[int, ...] | None: + """The installed CLI's version, or None if it could not be read. + + Takes the resolved path rather than resolving its own, so the version that + gates the registration and the binary that performs it cannot be two + different installs. + """ + try: + out = subprocess.run( + [executable, "--version"], capture_output=True, text=True, timeout=30 + ).stdout + except (OSError, ValueError, subprocess.SubprocessError): + return None + match = re.search(r"(\d+)\.(\d+)\.(\d+)", out) + return tuple(int(g) for g in match.groups()) if match else None + + def install_mcp(target_root: Path, mode: str, force: bool) -> list[Path]: - installed: list[Path] = [] - mcp_dest_dir = target_root / ".codex" / "mcp" - mcp_dest_dir.mkdir(parents=True, exist_ok=True) - # server.py imports `ix_llm` as a sibling, so both have to land. The import - # is guarded on the server side, so a stale install that has only server.py - # loses the `--format llm` fast-path rather than failing to start — but - # shipping the pair is what makes the fast-path available at all. - for name in ("server.py", "ix_llm.py"): - dest = mcp_dest_dir / name - install_file(repo_root() / "mcp" / name, dest, mode, force) - installed.append(dest) - return installed + """Register the CLI's own MCP server rather than shipping one. + + This plugin used to install `mcp/server.py`, a FastMCP server whose 23 tools + each shelled out to the `ix` CLI. The CLI now serves the same tools itself + via `ix mcp`, in-process, so the copy here was a second implementation of + one surface. Nothing is installed now; the CLI is pointed at instead. + """ + del target_root, mode, force # nothing is copied any more + return [] def main() -> None: @@ -834,9 +867,23 @@ def main() -> None: if args.hooks: print("Restart Codex so it reloads .codex/config.toml and hooks.json.") if args.mcp: - mcp_path = target_root / ".codex" / "mcp" / "server.py" - print(f"Register the MCP server with Codex:") - print(f" codex mcp add ix-memory -- python3 {mcp_path}") + executable = _ix_executable() + version = _ix_version(executable) if executable is not None else None + if version is None: + print("Could not run `ix --version`; install the Ix CLI, then re-run with --mcp.") + elif version < MIN_IX_VERSION_FOR_MCP: + wanted = ".".join(str(part) for part in MIN_IX_VERSION_FOR_MCP) + got = ".".join(str(part) for part in version) + print(f"The MCP server needs Ix CLI >= {wanted} (found {got}). Run `ix upgrade`.") + else: + # `ix mcp install` owns the per-host detail — including resolving the + # launcher for the seven hosts it registers, on the Windows where npm + # ships ix.CMD and no ix.exe. Reaching it has the same problem one + # level up, so the resolved path is what is spawned: the bare name + # would raise FileNotFoundError before `check=False` had any say, and + # the registration this flag exists to write would never happen. + print("Registering the Ix MCP server with Codex:") + subprocess.run([executable, "mcp", "install", "--host", "codex"], check=False) if __name__ == "__main__": diff --git a/test-local.sh b/test-local.sh index 7c707af..37a5a89 100755 --- a/test-local.sh +++ b/test-local.sh @@ -57,10 +57,8 @@ python3 -m py_compile \ "$REPO/.codex/hooks/pre_tool_use.py" \ "$REPO/.codex/hooks/post_tool_use.py" \ "$REPO/.codex/hooks/stop.py" \ - "$REPO/mcp/ix_llm.py" \ "$REPO/.codex/hooks/_launch.py" \ - "$REPO/scripts/install_codex_integration.py" \ - "$REPO/mcp/server.py" >/dev/null \ + "$REPO/scripts/install_codex_integration.py" >/dev/null \ && ok "Python files compile" || fail "Python compile failed" # Show the failures rather than swallowing them: `>/dev/null 2>&1` leaves a @@ -119,51 +117,25 @@ POST_OUT="$(printf '{"cwd":"%s","tool_input":{"command":"echo hello > /tmp/ix-te ok "post_tool_use: dry-run completed without error" echo "" -echo "-- MCP server checks --" +echo "-- MCP registration checks --" +# This plugin no longer ships an MCP server: `ix mcp` in the CLI serves the same +# tools, so there is nothing here to introspect. What still has to hold is that +# the installer delegates instead of copying a server, and that it refuses to +# register against a CLI too old to have the subcommand. python3 -c " -import sys -sys.path.insert(0, '$REPO/.codex/hooks') -# The SDK must be importable for the server to work, under either of its names. -# 2.0.0 renamed FastMCP to MCPServer and moved it to mcp.server.mcpserver; this -# check pinned the v1 path, so it reported a missing package on a machine where -# the package was installed and current. -try: - from mcp.server.mcpserver import MCPServer - print(' [ok] mcp package importable (>= 2.0.0)') -except ImportError: - try: - from mcp.server.fastmcp import FastMCP - print(' [ok] mcp package importable (< 2.0.0)') - except ImportError: - print(' [FAIL] mcp package not installed (pip install mcp)') - sys.exit(1) +import importlib.util, sys +spec = importlib.util.spec_from_file_location('inst', '$REPO/scripts/install_codex_integration.py') +mod = importlib.util.module_from_spec(spec); spec.loader.exec_module(mod) + +if mod.install_mcp(None, 'copy', False) != []: + print(' [FAIL] install_mcp still installs files'); sys.exit(1) +print(' [ok] install_mcp copies no server of its own') -# Verify server.py registers at least 20 tools -import asyncio, importlib.util -spec = importlib.util.spec_from_file_location('mcp_server', '$REPO/mcp/server.py') -mod = importlib.util.module_from_spec(spec) -spec.loader.exec_module(mod) -# list_tools(), not the _tool_manager._tools it used to read: a private -# attribute is exactly what a major version is free to move, and this test -# exists to catch that class of break rather than take part in it. -tool_count = len(asyncio.run(mod.mcp.list_tools())) -if tool_count >= 20: - print(f' [ok] MCP server registers {tool_count} tools') -else: - print(f' [FAIL] MCP server only registers {tool_count} tools (expected >= 20)') - sys.exit(1) -" 2>/dev/null && true || fail "MCP server check failed" - -# Registering tools says nothing about whether they can run. Every tool but -# ix_health used to omit the `ix` executable and try to exec a program named -# after the subcommand, and _json parsed stdout as JSON without ever asking for -# it. Both are invisible to a count, so tests/ drives all 23 tools against a -# stub `ix` on PATH that reports the argv it was handed, under both MCP SDK -# major versions. -python3 "$REPO/tests/test_mcp_cli_invocation.py" >/dev/null 2>&1 \ - && ok "MCP tools invoke the ix CLI and request a machine format" \ - || fail "MCP tool invocation check failed" +if mod.MIN_IX_VERSION_FOR_MCP < (0, 9, 3): + print(' [FAIL] version floor predates the ix mcp subcommand'); sys.exit(1) +print(f' [ok] requires ix >= {\".\".join(str(p) for p in mod.MIN_IX_VERSION_FOR_MCP)}') +" && ok "installer delegates MCP to the ix CLI" || fail "MCP delegation check failed" echo "" @@ -185,14 +157,6 @@ python3 "$REPO/tests/test_windows_hook_launch.py" >/dev/null 2>&1 \ && ok "hooks launch without a shell on Windows" \ || fail "Windows hook-launch check failed" -# `ix` does not validate --format: an unknown value falls through to human text -# and exits 0. So asking an old CLI for `llm` never errors -- it silently -# answers with prose. The per-command version floors are the only thing standing -# between that and records being expected. Tier 5 (explain, read) landed in -# 0.9.2, everything else in 0.7.0, and Pro commands have no llm renderer at all. -python3 "$REPO/tests/test_llm_fastpath.py" >/dev/null 2>&1 \ - && ok "llm fast-path is gated on the CLI version" \ - || fail "llm fast-path gate check failed" echo "" echo "-- _scrub_secrets unit tests --" diff --git a/tests/test_ix_argv_resolution.py b/tests/test_ix_argv_resolution.py index b201398..f30657d 100644 --- a/tests/test_ix_argv_resolution.py +++ b/tests/test_ix_argv_resolution.py @@ -16,15 +16,23 @@ from __future__ import annotations +import ast import importlib.util +import io import os import subprocess import sys +import tempfile import unittest +from contextlib import redirect_stdout from pathlib import Path from unittest.mock import patch HOOKS_DIR = Path(__file__).resolve().parents[1] / ".codex" / "hooks" +# Scanned by the argv guard too: the installer spawns `ix` as well, and the +# guard covering only the hooks is exactly how the --mcp path came to spawn +# the bare name. +INSTALLER_PATH = Path(__file__).resolve().parents[1] / "scripts" / "install_codex_integration.py" def _load_common(): @@ -49,6 +57,15 @@ def _load_common(): ) +def load_installer(): + spec = importlib.util.spec_from_file_location("ix_installer_argv", INSTALLER_PATH) + assert spec and spec.loader + module = importlib.util.module_from_spec(spec) + sys.modules["ix_installer_argv"] = module + spec.loader.exec_module(module) + return module + + class IxArgvResolutionTest(unittest.TestCase): def setUp(self) -> None: self.common = _load_common() @@ -212,18 +229,91 @@ def test_every_hook_argv_starts_with_ix(self) -> None: If a caller ever builds an argv some other way, it silently skips the resolver and regresses on Windows only. """ - sources = list(HOOKS_DIR.glob("*.py")) + sources = list(HOOKS_DIR.glob("*.py")) + [INSTALLER_PATH] self.assertTrue(sources, "no hook sources found") offenders = [] for path in sources: - for lineno, line in enumerate(path.read_text().splitlines(), 1): - stripped = line.strip() - if stripped.startswith("#"): + tree = ast.parse(path.read_text()) + for node in ast.walk(tree): + # The real shape, not a line-grep. Grepping for `"ix"` and + # `subprocess.` on one line reads prose as code — a docstring + # explaining that `subprocess.run(["ix", ...])` is wrong scored as + # an offender — and reads code as prose the moment a call is split + # across lines, which is the direction that actually costs + # something. + if not isinstance(node, ast.Call): + continue + func = node.func + if not (isinstance(func, ast.Attribute) and isinstance(func.value, ast.Name)): + continue + if func.value.id != "subprocess": + continue + if not node.args: + continue + argv = node.args[0] + if not isinstance(argv, (ast.List, ast.Tuple)) or not argv.elts: continue - if '"ix"' in stripped and "subprocess." in stripped: - offenders.append(f"{path.name}:{lineno}") + head = argv.elts[0] + if isinstance(head, ast.Constant) and head.value == "ix": + offenders.append(f"{path.name}:{node.lineno}") self.assertEqual([], offenders, "argv passed straight to subprocess without resolve_ix_argv") +class InstallerIxLaunchTest(unittest.TestCase): + """`--mcp` registers by running the CLI, so it has to reach it on Windows. + + The registration itself is right to delegate — `ix mcp install` resolves the + launcher for each host it writes. But reaching *that* has the same problem one + level up: a bare "ix" dies in CreateProcess before `check=False` is consulted, + so the flag prints its banner and then raises, having registered nothing. + """ + + def setUp(self) -> None: + self.installer = load_installer() + + def _run_mcp(self, which_returns: str | None, version_stdout: str = "ix 0.9.3"): + """Drive `main()` down the --mcp path, returning the spawned argvs.""" + calls: list[list[str]] = [] + + def fake_run(argv, *args, **kwargs): + calls.append(list(argv)) + return subprocess.CompletedProcess(argv, 0, stdout=version_stdout, stderr="") + + with tempfile.TemporaryDirectory() as target: + argv = ["install_codex_integration.py", "--repo", target, "--mcp"] + with patch.object(self.installer.shutil, "which", return_value=which_returns), \ + patch.object(self.installer.subprocess, "run", fake_run), \ + patch.object(sys, "argv", argv), \ + redirect_stdout(io.StringIO()) as out: + self.installer.main() + return calls, out.getvalue() + + def test_registers_with_the_resolved_path_not_the_bare_name(self) -> None: + calls, _ = self._run_mcp(WINDOWS_SHIM) + register = [c for c in calls if "mcp" in c] + self.assertEqual(1, len(register), f"expected one registration, got {calls}") + self.assertEqual( + [WINDOWS_SHIM, "mcp", "install", "--host", "codex"], + register[0], + "the bare name never resolves through CreateProcess on Windows", + ) + + def test_the_version_gate_and_the_registration_use_one_install(self) -> None: + """Two resolutions could gate on one `ix` and then register with another.""" + calls, _ = self._run_mcp(WINDOWS_SHIM) + self.assertTrue(calls, "nothing was spawned") + self.assertEqual({WINDOWS_SHIM}, {c[0] for c in calls}) + + def test_says_so_rather_than_raising_when_ix_is_not_on_path(self) -> None: + calls, printed = self._run_mcp(None) + self.assertEqual([], [c for c in calls if "mcp" in c]) + self.assertIn("install the Ix CLI", printed) + + def test_refuses_a_cli_below_the_floor(self) -> None: + calls, printed = self._run_mcp(WINDOWS_SHIM, version_stdout="ix 0.9.2") + self.assertEqual([], [c for c in calls if "mcp" in c]) + self.assertIn("0.9.3", printed) + + if __name__ == "__main__": sys.exit(0 if unittest.main(exit=False, verbosity=2).result.wasSuccessful() else 1) diff --git a/tests/test_llm_fastpath.py b/tests/test_llm_fastpath.py deleted file mode 100644 index 31e6007..0000000 --- a/tests/test_llm_fastpath.py +++ /dev/null @@ -1,319 +0,0 @@ -#!/usr/bin/env python3 -"""The `--format llm` fast-path and, mostly, the version gate in front of it. - -`ix` does not validate `--format`. Every renderer is -`if json: ... elif llm: ... else: text`, so an unrecognised value falls through -to human-readable text and exits 0. That is what makes the whole change safe — -there is no version of `ix` on which asking for `llm` breaks — and it is also -what makes a wrong gate dangerous: the CLI answers with prose instead of an -error, and nothing raises. Most of what follows pins that boundary. -""" - -from __future__ import annotations - -import importlib.util -import sys -import unittest -from pathlib import Path - -REPO_ROOT = Path(__file__).resolve().parents[1] - - -def load_module(path: Path, name: str): - spec = importlib.util.spec_from_file_location(name, path) - assert spec and spec.loader - module = importlib.util.module_from_spec(spec) - sys.modules[name] = module - spec.loader.exec_module(module) - return module - - -ix_llm = load_module(REPO_ROOT / "mcp" / "ix_llm.py", "ix_llm") - - -def fake_run(version: str, output: str = "", ok: bool = True): - """A `_run` stand-in that reports `version` and returns `output` otherwise.""" - calls: list[list[str]] = [] - - def run(args, timeout=15): - calls.append(list(args)) - if args == ["--version"]: - return True, version, "" - return ok, output, "" if ok else "boom" - - run.calls = calls # type: ignore[attr-defined] - return run - - -class Semver(unittest.TestCase): - def test_parses_a_plain_version(self) -> None: - self.assertEqual((0, 9, 2), ix_llm.parse_semver("0.9.2")) - - def test_parses_a_decorated_version(self) -> None: - # `ix --version` has printed a bare number, and could print more. - self.assertEqual((0, 9, 2), ix_llm.parse_semver("ix 0.9.2 (linux-amd64)")) - - def test_ignores_a_prerelease_suffix(self) -> None: - # 0.9.0-rc.1 is treated as 0.9.0. It predates every floor that matters, - # and the alternative — full semver precedence — buys nothing here. - self.assertEqual((0, 9, 0), ix_llm.parse_semver("0.9.0-rc.1")) - - def test_returns_none_for_junk(self) -> None: - self.assertIsNone(ix_llm.parse_semver("unknown")) - self.assertIsNone(ix_llm.parse_semver("")) - - -class Gate(unittest.TestCase): - def setUp(self) -> None: - ix_llm.reset_version_cache() - - def tearDown(self) -> None: - ix_llm.reset_version_cache() - - def test_tier_one_command_is_allowed_from_070(self) -> None: - self.assertTrue(ix_llm.supports_llm("stats", fake_run("0.7.0"))) - - def test_tier_one_command_is_refused_below_070(self) -> None: - self.assertFalse(ix_llm.supports_llm("stats", fake_run("0.6.0"))) - - def test_tier_five_command_is_refused_between_070_and_092(self) -> None: - """The reason the floors are per-command. - - `explain` and `read` only grew renderers in 0.9.2. On 0.9.1 they accept - `--format llm` and return *text* — no error, exit 0 — so a single - 0.7.0 gate would forward prose to the model as though it were records. - """ - for version in ("0.7.0", "0.8.1", "0.9.0", "0.9.1"): - ix_llm.reset_version_cache() - self.assertFalse( - ix_llm.supports_llm("explain", fake_run(version)), - f"explain must not use llm on {version}", - ) - ix_llm.reset_version_cache() - self.assertFalse(ix_llm.supports_llm("read", fake_run(version))) - - def test_tier_five_command_is_allowed_from_092(self) -> None: - self.assertTrue(ix_llm.supports_llm("explain", fake_run("0.9.2"))) - ix_llm.reset_version_cache() - self.assertTrue(ix_llm.supports_llm("read", fake_run("0.9.3"))) - - def test_pro_commands_are_never_allowed(self) -> None: - """`@ix/pro` declares only text|json — there is no llm renderer at any - version, so no gate can let these through and none should.""" - for command in ("briefing", "decisions", "goals", "plan", "truth"): - ix_llm.reset_version_cache() - self.assertFalse(ix_llm.supports_llm(command, fake_run("99.0.0")), command) - - def test_an_unknown_command_is_refused(self) -> None: - ix_llm.reset_version_cache() - self.assertFalse(ix_llm.supports_llm("nonesuch", fake_run("99.0.0"))) - - def test_an_unreadable_version_disables_the_fast_path(self) -> None: - def run(args, timeout=15): - return True, "not a version", "" - - self.assertFalse(ix_llm.supports_llm("stats", run)) - - def test_a_failed_probe_disables_the_fast_path(self) -> None: - def run(args, timeout=15): - return False, "", "ix not found" - - self.assertFalse(ix_llm.supports_llm("stats", run)) - - def test_a_raising_probe_disables_the_fast_path(self) -> None: - def run(args, timeout=15): - raise OSError("no such executable") - - self.assertFalse(ix_llm.supports_llm("stats", run)) - - def test_the_version_is_probed_once(self) -> None: - run = fake_run("0.9.2") - for _ in range(5): - ix_llm.supports_llm("stats", run) - probes = [c for c in run.calls if c == ["--version"]] - self.assertEqual(1, len(probes)) - - def test_kill_switch(self) -> None: - import os - - os.environ["IX_DISABLE_LLM_FORMAT"] = "1" - try: - self.assertFalse(ix_llm.supports_llm("stats", fake_run("0.9.2"))) - finally: - del os.environ["IX_DISABLE_LLM_FORMAT"] - - def test_diff_never_uses_the_fast_path(self) -> None: - """Not a version question, which is why no floor can answer it. - - `diff`'s renderer has branches with no llm arm at any version. The - textual-changes path -- graph reports no change but the file text - differs -- falls to the text `else` and prints " modified ( - textual changes ...)" at exit 0, and ix_diff reaches it with the - arguments it already sends. Prose at exit 0 is precisely what this - module would forward as records. - """ - self.assertNotIn("diff", ix_llm.LLM_MIN_VERSION) - for args in (["diff", "1", "5"], ["diff", "1", "5", "--content"]): - with self.subTest(args): - ix_llm.reset_version_cache() - self.assertFalse(ix_llm.supports_llm("diff", fake_run("9.9.9"), args)) - - def test_a_text_only_flag_still_defers(self) -> None: - """The _TEXT_ONLY_FLAGS mechanism, which `diff` no longer exercises. - - Both spellings: `--content x` and `--content=x` are the same flag, and - an exact-token check sees only the first. - """ - original_flags = dict(ix_llm._TEXT_ONLY_FLAGS) - original_floors = dict(ix_llm.LLM_MIN_VERSION) - - def restore() -> None: - # One callable, because addCleanup runs LIFO: registering - # clear() and update() separately ran them in the wrong order and - # left the table empty for every test after this one. - ix_llm._TEXT_ONLY_FLAGS.clear() - ix_llm._TEXT_ONLY_FLAGS.update(original_flags) - ix_llm.LLM_MIN_VERSION.clear() - ix_llm.LLM_MIN_VERSION.update(original_floors) - - self.addCleanup(restore) - ix_llm.LLM_MIN_VERSION["probe"] = (0, 7, 0) - ix_llm._TEXT_ONLY_FLAGS["probe"] = frozenset({"--content"}) - - for args, allowed in ( - (["probe", "x"], True), - (["probe", "--content", "x"], False), - (["probe", "--content=x"], False), - ): - with self.subTest(args): - ix_llm.reset_version_cache() - self.assertEqual( - allowed, ix_llm.supports_llm("probe", fake_run("0.9.2"), args) - ) - - - def test_a_double_digit_minor_is_newer_not_older(self) -> None: - """Tuples, not strings: "0.10.0" < "0.9.2" lexically, and 0.10.0 is - the newer release. A string compare would silently refuse the - fast-path on every version past 0.9.x.""" - for version in ("0.10.0", "0.9.10", "1.0.0"): - with self.subTest(version): - ix_llm.reset_version_cache() - self.assertTrue(ix_llm.supports_llm("explain", fake_run(version))) - - def test_a_probe_that_exits_non_zero_disables_the_fast_path(self) -> None: - """Even when its stdout still looks like a version.""" - def run(args, timeout=15): - if args == ["--version"]: - return False, "0.9.2", "boom" - return True, "stats nodes=1", "" - - ix_llm.reset_version_cache() - self.assertIsNone(ix_llm.detect_version(run)) - self.assertFalse(ix_llm.supports_llm("stats", run)) - - - def test_a_trailing_blank_line_survives(self) -> None: - """`read` emits `content lines=` then n raw lines. - - .strip() deleted blank final lines and left the count over-reporting - what followed it; leading whitespace is significant for the same - reason. Exactly one trailing newline comes off, no more. - """ - payload = "content lines=3" + chr(10) + " indented" + chr(10) + chr(10) + chr(10) - ix_llm.reset_version_cache() - text = ix_llm.try_llm(["read"], fake_run("0.9.2", output=payload)) - self.assertEqual(payload[:-1], text) - self.assertEqual(3, len(text.split(chr(10))) - 1) - - def test_an_ambiguous_version_string_disables_the_fast_path(self) -> None: - """Decoration is fine; two version-shaped tokens are not. - - Reading the wrong one is the single error that turns the fast-path on - for a CLI that answers with prose, so it refuses instead of guessing. - """ - self.assertEqual((0, 9, 2), ix_llm.parse_semver("ix 0.9.2 (linux-amd64)")) - self.assertIsNone(ix_llm.parse_semver("node v20.11.0 / ix 0.7.0")) - ix_llm.reset_version_cache() - self.assertFalse( - ix_llm.supports_llm("stats", fake_run("node v20.11.0 / ix 0.7.0")) - ) - - -class TryLlm(unittest.TestCase): - def setUp(self) -> None: - ix_llm.reset_version_cache() - - def tearDown(self) -> None: - ix_llm.reset_version_cache() - - def test_returns_text_and_asks_for_the_llm_format(self) -> None: - run = fake_run("0.9.2", output="stats nodes=98979 edges=354283\n") - text = ix_llm.try_llm(["stats"], run) - self.assertEqual("stats nodes=98979 edges=354283", text) - self.assertIn(["stats", "--format", "llm"], run.calls) - - def test_defers_when_the_cli_is_too_old(self) -> None: - run = fake_run("0.6.0", output="stats nodes=1") - self.assertIsNone(ix_llm.try_llm(["stats"], run)) - # And did not even try the command. - self.assertNotIn(["stats", "--format", "llm"], run.calls) - - def test_defers_on_a_failed_invocation(self) -> None: - """With output, so it is the exit code that defers and not emptiness. - - `fake_run(ok=False)` also returns empty stdout, so this passed with the - `if not ok` guard deleted -- the emptiness check caught it instead and - the exit-code branch had no coverage at all. - """ - run = fake_run("0.9.2", output="stats nodes=1 edges=2", ok=False) - self.assertIsNone(ix_llm.try_llm(["stats"], run)) - - def test_defers_on_empty_output(self) -> None: - self.assertIsNone(ix_llm.try_llm(["stats"], fake_run("0.9.2", output=" \n"))) - - def test_defers_on_an_error_record(self) -> None: - """ix reports some failures as `error code=...` on stdout with exit 0. - - Forwarding that verbatim would hand the model an error line dressed as - a result. Deferring keeps the error contract identical to the JSON path. - """ - run = fake_run("0.9.2", output='error code=unknown_target message="No entity named X"') - self.assertIsNone(ix_llm.try_llm(["locate", "X"], run)) - - def test_defers_on_an_empty_argv(self) -> None: - self.assertIsNone(ix_llm.try_llm([], fake_run("0.9.2"))) - - def test_does_not_parse_the_output(self) -> None: - # The contract is passthrough. Anything llm-shaped comes back verbatim, - # including content that is not valid JSON and never has to be. - payload = 'region id=cli label="Cli / Client" level=2\nregion id=srv parent=cli' - run = fake_run("0.9.2", output=payload + "\n") - self.assertEqual(payload, ix_llm.try_llm(["subsystems"], run)) - - -class ServerWiring(unittest.TestCase): - """The server must degrade rather than die if the module is missing.""" - - def test_server_imports_try_llm_defensively(self) -> None: - source = (REPO_ROOT / "mcp" / "server.py").read_text(encoding="utf-8") - self.assertIn("from ix_llm import try_llm", source) - self.assertIn("except ImportError", source) - - def test_installer_ships_both_files(self) -> None: - # server.py imports ix_llm as a sibling; installing only server.py would - # silently drop the fast-path on every install. - source = (REPO_ROOT / "scripts" / "install_codex_integration.py").read_text( - encoding="utf-8" - ) - self.assertIn('"server.py", "ix_llm.py"', source) - - def test_pro_tools_do_not_reach_the_fast_path(self) -> None: - # ix_decisions routes through _read like the rest; it is safe only - # because `decisions` is absent from the table. Pin that. - self.assertNotIn("decisions", ix_llm.LLM_MIN_VERSION) - self.assertNotIn("briefing", ix_llm.LLM_MIN_VERSION) - - -if __name__ == "__main__": - unittest.main(verbosity=2) diff --git a/tests/test_mcp_cli_invocation.py b/tests/test_mcp_cli_invocation.py deleted file mode 100644 index d88ac12..0000000 --- a/tests/test_mcp_cli_invocation.py +++ /dev/null @@ -1,517 +0,0 @@ -#!/usr/bin/env python3 -"""Every MCP tool must actually reach the `ix` CLI, and ask for JSON before parsing it. - -Registering 23 tools says nothing about whether they run. Every tool but -`ix_health` used to pass only the subcommand to `subprocess.run`, so Python -tried to exec programs named `stats`, `locate` and `map`; each died in the -OSError branch and returned an error blob. `ix_health` was the one tool that -spelled the executable correctly, which is exactly what let the server look -alive while nothing in it worked. A tool count cannot see any of that. - -So this drives all 23 through a fake `ix` on PATH that echoes the argv it was -handed, and asserts the full argv list. Two properties fall out of doing it -this way rather than by mocking `subprocess.run`: - - * `--format json` is *requested* rather than assumed. `_parse` looks for the - first `{` or `[` in the output, so against human-oriented text it would - either fail or, worse, succeed on a fragment of a rendered table. - * arguments reach the CLI as argv rather than as text a shell re-parses. That - property is platform-specific and is checked separately, in - test_shell_metacharacters_never_reach_a_shell — on Windows there is no - `ix.exe`, so the resolved `.CMD` runs via cmd.exe and the server refuses - metacharacters rather than passing them. - -Originally contributed by @Hiro-Chiba in #16; folded in here because #13 -changes the same call path. -""" - -from __future__ import annotations - -import importlib.util -import json -import os -import shutil -import subprocess -import sys -import tempfile -import types -import unittest -from pathlib import Path -from unittest.mock import patch - -REPO_ROOT = Path(__file__).resolve().parents[1] - -# The same module object server.py imports, so resetting its memoised version -# here actually affects the server under test. -_llm_spec = importlib.util.spec_from_file_location( - "ix_llm", REPO_ROOT / "mcp" / "ix_llm.py" -) -ix_llm = importlib.util.module_from_spec(_llm_spec) -sys.modules["ix_llm"] = ix_llm -_llm_spec.loader.exec_module(ix_llm) - - -class _FakeServer: - """Stands in for FastMCP / MCPServer — only the constructor and .tool() matter.""" - - def __init__(self, name: str) -> None: - self.name = name - - def tool(self): - return lambda function: function - - -def _load_server(sdk: str): - """Import mcp/server.py against a faked MCP SDK. - - `sdk` selects which import line the server should find, so both branches of - the version shim get exercised: 2.0.0 renamed FastMCP to MCPServer and moved - it to mcp.server.mcpserver, and both spellings are in the wild because - nothing in this repo pins the SDK. - """ - mcp_module = types.ModuleType("mcp") - mcp_server_module = types.ModuleType("mcp.server") - # Pin our ix_llm instance for the exec: test_llm_fastpath loads the same - # file under the same name, so whichever module was imported last owns - # sys.modules["ix_llm"] and the server would otherwise bind an instance - # whose memoised version this file never resets -- making the expected - # format token depend on test file ordering. - modules = {"mcp": mcp_module, "mcp.server": mcp_server_module, "ix_llm": ix_llm} - - if sdk == "v2": - mcpserver_module = types.ModuleType("mcp.server.mcpserver") - mcpserver_module.MCPServer = _FakeServer - modules["mcp.server.mcpserver"] = mcpserver_module - elif sdk == "v1": - fastmcp_module = types.ModuleType("mcp.server.fastmcp") - fastmcp_module.FastMCP = _FakeServer - modules["mcp.server.fastmcp"] = fastmcp_module - # Absent mcp.server.mcpserver must fall through to the v1 import. - modules["mcp.server.mcpserver"] = None - else: # pragma: no cover - programming error - raise ValueError(sdk) - - spec = importlib.util.spec_from_file_location( - f"ix_mcp_server_{sdk}", REPO_ROOT / "mcp" / "server.py" - ) - assert spec is not None and spec.loader is not None - module = importlib.util.module_from_spec(spec) - with patch.dict(sys.modules, modules): - spec.loader.exec_module(module) - return module - - -# (attribute, args, kwargs, expected argv). `ix_health` probes --version and is -# here because being the only correct tool is what hid the bug. - -CASES: list[tuple[str, tuple, dict, list]] = [ - ("ix_health", (), {}, ["--version"]), - ("ix_briefing", (), {}, ["briefing", "--format", "json"]), - # A symbol with a space, which is enough to prove the argument survives as one - # argv element. Shell metacharacters are a separate matter and get their own - # test — folding them in here made the sweep depend on a payload that has to - # be space-free, which is not what these 23 cases are checking. - ("ix_locate", ("User Service",), {}, ["locate", "User Service", "--format", "json"]), - ("ix_text", ("needle",), {"limit": 7, "path": "src", "language": "python"}, - ["text", "needle", "--limit", "7", "--path", "src", "--language", "python", "--format", "json"]), - ("ix_impact", ("Widget",), {}, ["impact", "Widget", "--format", "json"]), - # Was pinned as bare ["map"] — locking in the one tool still parsing rendered - # text as JSON, which is the defect the rest of this file exists to catch. - ("ix_map", (), {}, ["map", "--format", "json"]), - ("ix_overview", ("Widget",), {}, ["overview", "Widget", "--format", "json"]), - ("ix_read", ("Widget",), {}, ["read", "Widget", "--format", "json"]), - ("ix_diff", (3, 8), {"target": "Widget", "summary": True}, - ["diff", "3", "8", "Widget", "--summary", "--format", "json"]), - ("ix_callers", ("Widget",), {}, ["callers", "Widget", "--format", "json"]), - ("ix_callees", ("Widget",), {}, ["callees", "Widget", "--format", "json"]), - ("ix_imported_by", ("Widget",), {}, ["imported-by", "Widget", "--format", "json"]), - ("ix_imports", ("Widget",), {}, ["imports", "Widget", "--format", "json"]), - ("ix_depends", ("Widget",), {"depth": 2}, ["depends", "Widget", "--depth", "2", "--format", "json"]), - ("ix_trace", ("Widget",), {"to": "render"}, ["trace", "Widget", "--to", "render", "--format", "json"]), - ("ix_explain", ("Widget",), {}, ["explain", "Widget", "--format", "json"]), - ("ix_rank", (), {"by": "callers", "kind": "function", "top": 5, "path": "src"}, - ["rank", "--by", "callers", "--kind", "function", "--top", "5", "--path", "src", "--format", "json"]), - ("ix_inventory", ("src",), {"kind": "function"}, - ["inventory", "--kind", "function", "--path", "src", "--format", "json"]), - # limit is applied client-side to the parsed candidates, so it is deliberately - # not forwarded to the CLI. - ("ix_smells", (), {"path": "src", "limit": 10}, ["smells", "--path", "src", "--format", "json"]), - ("ix_stats", (), {}, ["stats", "--format", "json"]), - ("ix_subsystems", (), {}, ["subsystems", "--format", "json"]), - ("ix_decisions", (), {"path": "src"}, ["decisions", "--path", "src", "--format", "json"]), - ("ix_history", ("Widget",), {}, ["history", "Widget", "--format", "json"]), -] - -# The version the fake CLI reports. 0.9.1 is deliberate: it is above the Tier -# 1-4 floor and below Tier 5 (explain, read landed in 0.9.2), so one sweep -# exercises both sides of the per-command gate. -FAKE_IX_VERSION = "0.9.1" -FAKE_IX_SEMVER = (0, 9, 1) - - -FAKE_IX = """#!/usr/bin/env python3 -import json -import os -import sys - -with open(os.environ["FAKE_IX_LOG"], "a", encoding="utf-8") as log: - log.write(json.dumps(sys.argv[1:]) + "\\n") - -if sys.argv[1:] == ["--version"]: - print("__IX_VERSION__") -else: - print(json.dumps({"argv": sys.argv[1:]})) -""" - - -# FAKE_IX is a plain literal (it contains JSON braces), so patch the token in. -FAKE_IX = FAKE_IX.replace("__IX_VERSION__", FAKE_IX_VERSION) -assert FAKE_IX_VERSION in FAKE_IX, "version placeholder was not substituted" - -FAILING_IX = """#!/usr/bin/env python3 -import sys - -sys.stderr.write("graph not ingested; run ix init\\n") -sys.exit(2) -""" - - -def _write_fake_ix(directory: Path, source: str = FAKE_IX) -> None: - """Put a fake `ix` on PATH that this platform can actually execute. - - A bare `ix` carrying a `#!` line works only on POSIX. Windows cannot exec a - shebang script at all, so the stub never ran, nothing was ever logged, and - both argv tests died reading a file that was never created — on the one PR - whose subject is making this work on Windows. `.bat` is in the default - PATHEXT, so `shutil.which("ix")` (what the server uses to resolve it) finds - this and hands off to the interpreter running the tests. - """ - if os.name == "nt": - impl = directory / "ix_impl.py" - impl.write_text(source, encoding="utf-8") - (directory / "ix.bat").write_text( - f'@echo off\r\n"{sys.executable}" "{impl}" %*\r\n', encoding="utf-8" - ) - return - fake_ix = directory / "ix" - fake_ix.write_text(source, encoding="utf-8") - fake_ix.chmod(0o755) - - -class McpCliInvocationTest(unittest.TestCase): - def setUp(self) -> None: - """Reset the fast-path probe so ordering cannot decide what is asserted. - - The llm fast-path rewrites the format token, and whether it engages - depends on a version memoised in ix_llm for the life of the process. - Without this, these assertions inherited whatever ran before them: alone - the cache was already primed and the JSON argv was asserted, under - `unittest discover` test_llm_fastpath had just reset it, the probe ran - against this file's own fake `ix`, and every expected `json` arrived as - `llm`. - - Reset rather than disabled. Switching the feature off here would leave - the shipped path -- the one a current CLI actually takes -- with no - end-to-end coverage anywhere, since test_llm_fastpath only drives - ix_llm against a stub. The fake reports FAKE_IX_VERSION, so the expected - token is derived from the same table production consults. - """ - ix_llm.reset_version_cache() - self.addCleanup(ix_llm.reset_version_cache) - - @staticmethod - def _expected_argv(argv: list[str]) -> list[str]: - """`argv` with its format token set to whatever the gate would choose.""" - if "--format" not in argv: - return argv - floor = ix_llm.LLM_MIN_VERSION.get(argv[0]) - if floor is None or FAKE_IX_SEMVER < floor: - return argv - swapped = list(argv) - swapped[swapped.index("--format") + 1] = "llm" - return swapped - - def _drive_all_tools(self, sdk: str): - server = _load_server(sdk) - - with tempfile.TemporaryDirectory() as temp_dir: - temp_path = Path(temp_dir) - log_path = temp_path / "argv.jsonl" - marker_path = temp_path / "shell-was-invoked" - _write_fake_ix(temp_path) - - def resolve(value): - return value - - environment = { - "FAKE_IX_LOG": str(log_path), - "PATH": f"{temp_path}{os.pathsep}{os.environ.get('PATH', '')}", - } - - with patch.dict(os.environ, environment): - # Prime the version probe before measuring. Production runs it - # once per process; leaving it inside the loop would land an - # extra ["--version"] in the log at whichever case first - # consults the gate. - ix_llm.detect_version(server._run) - log_path.unlink(missing_ok=True) - - results = [] - with patch.dict(os.environ, environment): - for name, args, kwargs, _expected in CASES: - tool = getattr(server, name) - results.append(tool(*(resolve(a) for a in args), **kwargs)) - - expected = [ - self._expected_argv([resolve(part) for part in argv]) - for _n, _a, _k, argv in CASES - ] - actual = [json.loads(line) for line in log_path.read_text().splitlines()] - return results, expected, actual, marker_path - - def test_v1_sdk_all_tools_invoke_ix_with_expected_arguments(self) -> None: - results, expected, actual, marker = self._drive_all_tools("v1") - self.assertEqual(expected, actual) - self.assertEqual(len(CASES), len(results)) - self.assertEqual(23, len(CASES), "every registered tool should be covered here") - self.assertFalse(marker.exists(), "arguments must reach ix as argv, never via a shell") - self.assertTrue(all("error" not in json.loads(r) for r in results)) - - def test_v2_sdk_all_tools_invoke_ix_with_expected_arguments(self) -> None: - results, expected, actual, marker = self._drive_all_tools("v2") - self.assertEqual(expected, actual) - self.assertFalse(marker.exists()) - self.assertTrue(all("error" not in json.loads(r) for r in results)) - - def test_shell_metacharacters_never_reach_a_shell(self) -> None: - """A space-free payload — the only kind that can reach one. - - `subprocess` quotes an argument only when it contains whitespace, so - `Widget & touch x` is protected by that quoting and passes this check - whether or not a shell is in the chain. It has to be space-free to test - anything, which is why the control below runs first: if the payload - cannot fire even through an explicit shell, the assertion that follows - means nothing. - - The two platforms are legitimately different. On POSIX the fake `ix` is - exec'd directly and the payload arrives verbatim as one argv element. On - Windows there is no `ix.exe` to exec — `shutil.which` finds a `.CMD`, and - CreateProcess runs those through `cmd.exe` — so the server refuses the - argument instead, and it must not reach the CLI at all. - """ - server = _load_server("v1") - with tempfile.TemporaryDirectory() as temp_dir: - temp_path = Path(temp_dir) - log_path = temp_path / "argv.jsonl" - marker_path = temp_path / "shell-was-invoked" - self.assertNotIn(" ", str(marker_path), "payload must stay space-free") - _write_fake_ix(temp_path) - payload = f"Widget&echo>{marker_path}" - environment = { - "FAKE_IX_LOG": str(log_path), - "PATH": f"{temp_path}{os.pathsep}{os.environ.get('PATH', '')}", - } - - # Positive control, in the shape the bug actually has: the payload - # handed to subprocess as an argv element, exactly as _run does it, - # with nothing guarding it. Running it through `shell=True` instead - # would prove only that a shell is a shell -- it is insensitive to - # the quoting rule that makes this payload dangerous and that one - # safe, so re-adding spaces would leave the control green. - with patch.dict(os.environ, environment): - resolved = shutil.which("ix", path=str(temp_path)) - self.assertIsNotNone(resolved) - subprocess.run( - [resolved, "locate", payload], capture_output=True - ) - if os.name == "nt": - self.assertTrue( - marker_path.exists(), - "control failed: an unguarded argv call did not fire the " - "payload, so the assertion below would pass for the wrong " - "reason -- check the payload is still space-free", - ) - marker_path.unlink() - else: - # No shim, no shell: nothing to escape, and the control is that - # the CLI receives it whole. - self.assertFalse(marker_path.exists()) - log_path.unlink(missing_ok=True) - - with patch.dict(os.environ, environment): - result = json.loads(server.ix_locate(payload)) - - self.assertFalse(marker_path.exists(), "the payload reached a shell") - logged = ( - [json.loads(x) for x in log_path.read_text().splitlines()] - if log_path.exists() - else [] - ) - # The gate probes `ix --version` on first use, and where that lands - # depends on what primed the cache earlier in the process. It is a - # legitimate call carrying no payload, so it is not what this test - # measures -- and leaving it in made the result order-dependent. - logged = [argv for argv in logged if argv != ["--version"]] - if os.name == "nt": - self.assertIn("error", result) - self.assertIn("refusing to run", result["error"]) - # Not "the log is empty": the gate probes `ix --version` before - # the refusal, and that call is legitimate. The property is that - # the payload itself never reaches the CLI. - self.assertEqual( - [], - [argv for argv in logged if any(payload in part for part in argv)], - "the argument must not reach the CLI", - ) - else: - # Format token from the same table production consults: the - # fake reports a version above `locate`'s floor, so the shipped - # path asks for llm here. - self.assertEqual( - [self._expected_argv(["locate", payload, "--format", "json"])], - logged, - ) - - @unittest.skipUnless(os.name == "nt", "cmd.exe shim only exists on Windows") - def test_every_refused_character_is_refused(self) -> None: - """One payload carrying several metacharacters cannot pin the set. - - With `Widget&echo>MARKER` as the only case, dropping either `&` or `>` - from the refused set leaves the test green, and `%` is never exercised - at all — so a later "fix" for the `Vec` false positive could quietly - reopen the hole. - """ - server = _load_server("v1") - for char in "&|<>^\"%!": - with self.subTest(char=char), tempfile.TemporaryDirectory() as temp_dir: - temp_path = Path(temp_dir) - _write_fake_ix(temp_path) - environment = { - "FAKE_IX_LOG": str(temp_path / "argv.jsonl"), - "PATH": f"{temp_path}{os.pathsep}{os.environ.get('PATH', '')}", - } - # Run from a scratch directory so the `>` case redirects there - # rather than into the repo if the guard ever breaks -- that is - # how a file called `x` once got committed. It must NOT be the - # directory holding the fake ix: shutil.which searches the - # working directory first on Windows, the resolved path would - # then be relative, and the isabs refusal would answer before - # the metacharacter check ever ran -- leaving this green for a - # reason that has nothing to do with the character under test. - with tempfile.TemporaryDirectory() as elsewhere: - cwd = os.getcwd() - os.chdir(elsewhere) - try: - with patch.dict(os.environ, environment): - result = json.loads(server.ix_locate(f"Widget{char}x")) - finally: - os.chdir(cwd) - self.assertIn("refusing to run", result.get("error", "")) - self.assertIn(repr(char), result["error"]) - - @unittest.skipUnless(os.name == "nt", "cmd.exe shim only exists on Windows") - def test_a_newline_in_an_argument_is_refused(self) -> None: - """CR and LF end a command line as surely as `&` splits one.""" - server = _load_server("v1") - with tempfile.TemporaryDirectory() as temp_dir: - temp_path = Path(temp_dir) - _write_fake_ix(temp_path) - environment = { - "FAKE_IX_LOG": str(temp_path / "argv.jsonl"), - "PATH": f"{temp_path}{os.pathsep}{os.environ.get('PATH', '')}", - } - for char in ("\r", "\n"): - with self.subTest(char=char), patch.dict(os.environ, environment): - result = json.loads(server.ix_locate(f"Widget{char}whoami")) - self.assertIn("refusing to run", result.get("error", "")) - - @unittest.skipUnless(os.name == "nt", "cmd.exe shim only exists on Windows") - def test_a_metacharacter_in_the_resolved_path_is_refused(self) -> None: - """list2cmdline leaves an unquoted path alone, so `C:\\a&b\\ix.cmd` splits.""" - server = _load_server("v1") - with tempfile.TemporaryDirectory() as temp_dir: - bad_dir = Path(temp_dir) / "a&b" - bad_dir.mkdir() - _write_fake_ix(bad_dir) - environment = { - "FAKE_IX_LOG": str(bad_dir / "argv.jsonl"), - "PATH": f"{bad_dir}{os.pathsep}{os.environ.get('PATH', '')}", - } - with patch.dict(os.environ, environment): - result = json.loads(server.ix_locate("Widget")) - self.assertIn("refusing to run", result.get("error", "")) - - @unittest.skipUnless( - os.name == "nt", "CPython only inserts os.curdir into the search on Windows" - ) - def test_a_working_directory_ix_is_not_executed(self) -> None: - """shutil.which prefers the CWD on Windows, `path=` notwithstanding.""" - server = _load_server("v1") - with tempfile.TemporaryDirectory() as temp_dir: - temp_path = Path(temp_dir) - _write_fake_ix(temp_path) - # PATH must be a real directory that simply has no `ix`: an empty - # PATH makes shutil.which give up before it ever consults the - # working directory, which would make this pass for the wrong - # reason. With a populated PATH it still returns `.\ix.bat`. - elsewhere = tempfile.mkdtemp() - self.addCleanup(os.rmdir, elsewhere) - cwd = os.getcwd() - os.chdir(temp_path) - try: - with patch.dict( - os.environ, - {"PATH": elsewhere, "FAKE_IX_LOG": str(temp_path / "l")}, - clear=False, - ): - os.environ.pop("NoDefaultCurrentDirectoryInExePath", None) - self.assertIsNotNone( - shutil.which("ix", path=elsewhere), - "control: shutil.which should still find the CWD copy, " - "otherwise this test proves nothing", - ) - result = json.loads(server.ix_locate("Widget")) - finally: - # Before the TemporaryDirectory is torn down: Windows cannot - # remove a directory that is some process's working directory. - os.chdir(cwd) - self.assertIn("error", result) - self.assertFalse( - (temp_path / "l").exists(), "the working-directory ix was executed" - ) - - def test_the_reason_a_call_failed_reaches_the_caller(self) -> None: - """_json hands its callers bare data, so stderr had nowhere to go. - - Both routes: the early returns in _run, and a CLI that actually ran and - exited non-zero. The second is the one a user hits normally. - """ - server = _load_server("v1") - with patch.dict(os.environ, {"PATH": ""}): - result = json.loads(server.ix_locate("Widget")) - self.assertIn("ix CLI not found", result.get("error", "")) - - with tempfile.TemporaryDirectory() as temp_dir: - temp_path = Path(temp_dir) - _write_fake_ix(temp_path, source=FAILING_IX) - environment = { - "FAKE_IX_LOG": str(temp_path / "argv.jsonl"), - "PATH": f"{temp_path}{os.pathsep}{os.environ.get('PATH', '')}", - } - with patch.dict(os.environ, environment): - result = json.loads(server.ix_locate("Widget")) - self.assertIn("graph not ingested", result.get("error", "")) - - def test_every_registered_tool_is_covered(self) -> None: - """A tool added without a case here would otherwise go untested silently.""" - server = _load_server("v1") - exported = { - name - for name in dir(server) - if name.startswith("ix_") and callable(getattr(server, name)) - } - self.assertEqual(exported, {name for name, _a, _k, _e in CASES}) - - -if __name__ == "__main__": - unittest.main(verbosity=2)