diff --git a/CHANGES b/CHANGES index bd33a90..fde4762 100644 --- a/CHANGES +++ b/CHANGES @@ -16,6 +16,14 @@ or as a bare name carrying only its type. ### Development +**Safer pull-request testing with `mcp_swap.py`** + +`use-local --pr N` points installed agent CLIs at a pull request without a +checkout and verifies the server before changing their configuration. Swaps +preserve unrelated config text, file permissions, and symlink targets, while +retaining the original recovery data across failed or concurrent updates. +(#115) + #### CI actions updated to current majors Workflow actions moved to their current major releases: `actions/checkout` v7, diff --git a/scripts/README.md b/scripts/README.md index 08601f7..1588f3c 100644 --- a/scripts/README.md +++ b/scripts/README.md @@ -46,6 +46,43 @@ This matches Claude's conventional dev form and takes advantage of `uv run`'s automatic editable install — source edits flow through on the next invocation with no reinstall step. +### `--pr N` — point every CLI at a pull request + +Review a branch across your agents without checking it out: + +```console +$ uv run scripts/mcp_swap.py use-local --pr 114 +``` + +Each CLI's entry becomes: + +``` +command = "uvx" +args = ["--from", "git+@refs/pull/114/head", "libtmux-mcp"] +``` + +`uv` resolves the ref itself, so nothing lands on disk to refresh or +prune and `revert` restores the config with no extra cleanup. GitHub +publishes `refs/pull/N/head` on the base repository, so a pull request +from a fork needs no special handling. + +Before writing anything, the swap launches the resolved command once and +completes an MCP `initialize` round trip. A ref that does not exist, or a +dependency that cannot resolve, fails there — rather than landing in +every CLI's config and surfacing later as an opaque startup error inside +each agent. Pass `--no-preflight` to skip the probe when offline. + +`gh` confirms the number exists and labels the output; resolution does +not depend on it, so an unauthenticated `gh` degrades to an unlabelled +swap rather than a failure. + +A branch whose dependencies need resolver flags can carry them as +environment, the same way any other setting travels: + +```console +$ uv run scripts/mcp_swap.py use-local --pr 114 --env UV_NO_CONFIG=1 +``` + ### `--scope {user,project}` (Claude only) Claude's `~/.claude.json` supports two config scopes for MCP servers: diff --git a/scripts/mcp_swap.py b/scripts/mcp_swap.py index 5b4355e..6139b10 100755 --- a/scripts/mcp_swap.py +++ b/scripts/mcp_swap.py @@ -8,7 +8,8 @@ Use when you want every installed agent CLI to run a local checkout of an MCP server (editable) instead of a pinned release. ``use-local`` rewrites each CLI's config to invoke the checkout via ``uv --directory run -``; ``revert`` restores from the timestamped backup the swap wrote. +``, or a pull request's head via ``uvx`` with ``--pr``; ``revert`` +restores from the timestamped backup the swap wrote. Swapping a layer that is already swapped keeps that first backup rather than taking a new one, so ``revert`` always lands on the pre-swap config. @@ -25,6 +26,7 @@ $ uv run scripts/mcp_swap.py status $ uv run scripts/mcp_swap.py use-local --dry-run $ uv run scripts/mcp_swap.py use-local +$ uv run scripts/mcp_swap.py use-local --pr 115 $ uv run scripts/mcp_swap.py revert ``` @@ -71,12 +73,17 @@ from __future__ import annotations import argparse +import contextlib import dataclasses import difflib +import fcntl import json import os import pathlib +import re import shutil +import stat +import subprocess import sys import tempfile import time @@ -235,6 +242,12 @@ class CLIInfo: } +#: A ``--from`` argument pointing at a pull request's head commit. +#: GitHub publishes ``refs/pull//head`` on the *base* repository, so +#: one URL serves same-repo and fork pull requests alike. +PR_REF_RE = re.compile(r"git\+(?P.+?)@refs/pull/(?P\d+)/head") + + @dataclasses.dataclass class McpServerSpec: """The portable shape shared across CLI configs.""" @@ -275,6 +288,16 @@ def local_repo_path(self) -> pathlib.Path | None: return None return pathlib.Path(self.args[i + 1]) + def pr_ref(self) -> tuple[str, int] | None: + """Return ``(repo_url, pr_number)`` for a ``uvx`` pull-request spec.""" + if self.command != "uvx": + return None + for arg in self.args: + match = PR_REF_RE.fullmatch(arg) + if match: + return match.group("url"), int(match.group("number")) + return None + @dataclasses.dataclass class SwapEntry: @@ -297,6 +320,15 @@ class SwapEntry: #: ``Lib/sched.py`` uses to break ties on ``Event(time, priority, #: sequence, …)``. seq_no: int + #: Exact destination changed by the swap. ``config_path`` may be a + #: symlink that is later repointed, so it is not sufficient recovery + #: identity. Older state entries omit this field and fall back to + #: ``config_path`` during revert. + target_path: str | None = None + + +class SwapStateError(RuntimeError): + """Swap state is unsafe to use for a mutating operation.""" # --------------------------------------------------------------------------- @@ -317,22 +349,67 @@ def load_config(info: CLIInfo) -> t.Any: return tomlkit.parse(raw.decode()) -def dump_config_bytes(info: CLIInfo, config: t.Any) -> bytes: - """Serialize an edited config back to bytes in its original format.""" - if info.fmt == "json": - return (json.dumps(config, indent=2) + "\n").encode() - return tomlkit.dumps(config).encode() +def _json_trailer(original: bytes) -> str: + """Return the newline a rewritten JSON config should end with. + + Claude writes ``~/.claude.json`` without a trailing newline, so + appending one unconditionally grows the file by a byte on every swap + and shows as a diff hunk in a region the swap never touched. Empty + bytes mean a file being seeded, which gets the conventional newline. + """ + if not original: + return "\n" + return "\n" if original.endswith(b"\n") else "" + + +def dump_config_bytes(info: CLIInfo, config: t.Any, *, original: bytes) -> bytes: + """Serialize an edited config back to bytes in its original format. + + ``original`` is the file's pre-edit bytes, or empty when seeding a + new one. The parsed structure does not record the byte-level + conventions of the file it came from, so they are carried over from + the source instead. Required rather than defaulted: a caller that + omitted it would silently start rewriting regions it never touched, + which is the defect this parameter exists to prevent. tomlkit + preserves those conventions itself; only the JSON writer needs it. + """ + if info.fmt != "json": + return tomlkit.dumps(config).encode() + trailer = _json_trailer(original) + # ensure_ascii would re-escape every non-ASCII character in the file, + # including config text the swap never read. + text = json.dumps(config, indent=2, ensure_ascii=False) + trailer + try: + return text.encode() + except UnicodeEncodeError: + # A lone surrogate — a JS writer slicing a string mid-pair — has no + # UTF-8 encoding. Escaping the document is then the only form that + # can be written at all. + return (json.dumps(config, indent=2) + trailer).encode() def atomic_write(path: pathlib.Path, data: bytes) -> None: - """Write bytes to ``path`` via tempfile + ``os.replace`` to avoid partial writes.""" - path.parent.mkdir(parents=True, exist_ok=True) - fd, tmp_name = tempfile.mkstemp(prefix=path.name + ".", dir=str(path.parent)) + """Write bytes to ``path`` without replacing a symlinked config. + + Parameters + ---------- + path : pathlib.Path + Destination path. A symlink resolves to its final target so the + write preserves every link in the chain. + data : bytes + Bytes to write atomically. + """ + target = path.resolve() if path.is_symlink() else path + target.parent.mkdir(parents=True, exist_ok=True) + mode = stat.S_IMODE(target.stat().st_mode) if target.exists() else None + fd, tmp_name = tempfile.mkstemp(prefix=target.name + ".", dir=str(target.parent)) tmp = pathlib.Path(tmp_name) try: with os.fdopen(fd, "wb") as fh: + if mode is not None: + os.fchmod(fh.fileno(), mode) fh.write(data) - tmp.replace(path) + tmp.replace(target) except Exception: tmp.unlink(missing_ok=True) raise @@ -665,12 +742,167 @@ def build_local_spec(repo: pathlib.Path, entry: str) -> McpServerSpec: ) +def build_pr_spec(repo_url: str, pr: int, entry: str) -> McpServerSpec: + """Build the ``uvx --from git+@refs/pull//head `` spec. + + Nothing is checked out: ``uv`` resolves the ref itself, so a swap + leaves no worktree to refresh or prune and ``revert`` needs no + cleanup beyond restoring the config. + """ + return McpServerSpec( + command="uvx", + args=["--from", f"git+{repo_url}@refs/pull/{pr}/head", entry], + ) + + +def _run_text(argv: list[str], cwd: pathlib.Path | None = None) -> str: + """Run ``argv`` and return stdout, raising on a non-zero exit.""" + return subprocess.run( + argv, + cwd=None if cwd is None else str(cwd), + capture_output=True, + text=True, + check=True, + ).stdout + + +def remote_https_url(repo: pathlib.Path, remote: str = "origin") -> str: + """Return ``https:////`` for a repo's git remote. + + Normalizes the spellings git accepts — ``git@host:owner/name.git``, + an ``ssh://`` or ``git+ssh://`` scheme, an embedded user, a trailing + ``.git`` — because the pull-request ref is fetched over https however + the working copy was cloned. + """ + try: + raw = _run_text(["git", "-C", str(repo), "remote", "get-url", remote]) + except (OSError, subprocess.CalledProcessError) as exc: + msg = f"cannot read git remote {remote!r} in {repo}" + raise RuntimeError(msg) from exc + return _normalize_remote_url(raw.strip()) + + +def _normalize_remote_url(url: str) -> str: + """Rewrite any git remote spelling as a plain https URL. + + Examples + -------- + >>> _normalize_remote_url("git+ssh://git@github.com/o/n.git") + 'https://github.com/o/n' + >>> _normalize_remote_url("git@github.com:o/n.git") + 'https://github.com/o/n' + >>> _normalize_remote_url("https://github.com/o/n") + 'https://github.com/o/n' + """ + url = url.removeprefix("git+") + if url.startswith("ssh://"): + url = "https://" + url.removeprefix("ssh://") + elif "://" not in url and ":" in url: + host, _, path = url.partition(":") + url = f"https://{host}/{path}" + scheme, sep, rest = url.partition("://") + authority, slash, path = rest.partition("/") + return f"{scheme}{sep}{authority.rpartition('@')[2]}{slash}{path}".removesuffix( + ".git" + ) + + +def gh_pr_summary(repo: pathlib.Path, pr: int) -> dict[str, t.Any] | None: + """Return ``gh``'s view of a pull request, or ``None`` when unreadable. + + Used to confirm the number exists and to label output. Resolution + does not depend on it: the ref and URL come from git, so a missing + or unauthenticated ``gh`` degrades to an unlabelled swap rather than + a failure. + """ + try: + out = _run_text( + [ + "gh", + "pr", + "view", + str(pr), + "--json", + "number,title,state,headRefName,isCrossRepository", + ], + cwd=repo, + ) + except (OSError, subprocess.CalledProcessError): + return None + try: + loaded = json.loads(out) + except json.JSONDecodeError: + return None + return loaded if isinstance(loaded, dict) else None + + +#: One MCP ``initialize`` request, newline-framed for stdio. +_INITIALIZE_FRAME = ( + json.dumps( + { + "jsonrpc": "2.0", + "id": 1, + "method": "initialize", + "params": { + "protocolVersion": "2025-06-18", + "capabilities": {}, + "clientInfo": {"name": "mcp_swap-preflight", "version": "1"}, + }, + } + ) + + "\n" +) + + +def preflight_spec(spec: McpServerSpec, *, timeout: float = 300.0) -> str | None: + """Launch ``spec`` and complete one MCP ``initialize`` round trip. + + Returns ``None`` when the server answered, otherwise a reason to + show the operator. A pull-request spec resolves its dependencies at + launch time, inside whichever agent starts it, so an unresolvable + ref would otherwise land in every config and surface later as an + opaque startup failure in each one. + + Closing stdin after the frame lets a well-behaved stdio server exit + on its own, which keeps this free of signal handling. + """ + try: + proc = subprocess.Popen( + [spec.command, *spec.args], + stdin=subprocess.PIPE, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + env={**os.environ, **spec.env}, + text=True, + ) + except OSError as exc: + return f"could not launch {spec.command}: {exc}" + + try: + out, err = proc.communicate(_INITIALIZE_FRAME, timeout=timeout) + except subprocess.TimeoutExpired: + proc.kill() + proc.communicate() + return f"no MCP response within {timeout:.0f}s" + + for line in out.splitlines(): + try: + message = json.loads(line) + except json.JSONDecodeError: + continue + if isinstance(message, dict) and message.get("id") == 1 and "result" in message: + return None + + tail = "\n".join(err.strip().splitlines()[-3:]) + return tail or "server exited without answering initialize" + + # --------------------------------------------------------------------------- # State file # --------------------------------------------------------------------------- -def load_state() -> dict[tuple[CLIName, Scope], SwapEntry]: +def load_state(*, strict: bool = False) -> dict[tuple[CLIName, Scope], SwapEntry]: """Read the swap-state file, returning an empty mapping when absent. The state file's schema is internal — no compatibility contract — @@ -678,23 +910,68 @@ def load_state() -> dict[tuple[CLIName, Scope], SwapEntry]: (those that don't parse as ``cli:scope``) and entries with a non-coercible ``seq_no`` or missing required fields are dropped silently so a hand-edited file cannot crash the script. + + A file that will not parse at all is reported rather than dropped + silently: it means the record of every swap is gone, so ``revert`` + is about to say there is nothing to unwind while swapped configs + and their backups sit on disk. Saying so is what lets the operator + go find those backups. Mutating callers pass ``strict=True`` so an + unreadable or malformed record blocks changes instead of being + overwritten as empty state. """ if not STATE_FILE.exists(): return {} - raw = json.loads(STATE_FILE.read_text()) + try: + raw = json.loads(STATE_FILE.read_text()) + except (OSError, ValueError) as exc: + message = f"swap state unreadable ({STATE_FILE}): {exc}" + print(message, file=sys.stderr) + if strict: + raise SwapStateError(message) from exc + return {} + if not isinstance(raw, dict): + if strict: + message = f"swap state has invalid shape: {STATE_FILE}" + print(message, file=sys.stderr) + raise SwapStateError(message) + return {} entries = raw.get("entries", {}) + if not isinstance(entries, dict): + if strict: + message = f"swap state has invalid entries: {STATE_FILE}" + print(message, file=sys.stderr) + raise SwapStateError(message) + entries = {} out: dict[tuple[CLIName, Scope], SwapEntry] = {} for k, v in entries.items(): parsed = _parse_state_key(k) if parsed is None: + if strict: + message = f"swap state has invalid key {k!r}: {STATE_FILE}" + print(message, file=sys.stderr) + raise SwapStateError(message) continue entry = _parse_state_entry(v) if entry is None: + if strict: + message = f"swap state has invalid entry {k!r}: {STATE_FILE}" + print(message, file=sys.stderr) + raise SwapStateError(message) continue out[parsed] = entry return out +@contextlib.contextmanager +def _state_lock() -> t.Iterator[None]: + """Serialize config mutations that share the swap state file.""" + STATE_DIR.mkdir(parents=True, exist_ok=True) + fd = os.open(STATE_DIR / "state.lock", os.O_RDWR | os.O_CREAT, 0o600) + with os.fdopen(fd, "a+b") as lock_file: + fcntl.flock(lock_file.fileno(), fcntl.LOCK_EX) + yield + + def save_state(entries: dict[tuple[CLIName, Scope], SwapEntry]) -> None: """Write the swap-state file atomically.""" STATE_DIR.mkdir(parents=True, exist_ok=True) @@ -707,13 +984,10 @@ def save_state(entries: dict[tuple[CLIName, Scope], SwapEntry]) -> None: atomic_write(STATE_FILE, (json.dumps(payload, indent=2) + "\n").encode("utf-8")) -def clear_state(keys: t.Iterable[tuple[CLIName, Scope]]) -> None: - """Remove the given ``(cli, scope)`` keys; delete the file if empty.""" - current = load_state() - for key in keys: - current.pop(key, None) - if current: - save_state(current) +def _save_or_clear_state(entries: dict[tuple[CLIName, Scope], SwapEntry]) -> None: + """Persist ``entries``, removing the state file when the mapping is empty.""" + if entries: + save_state(entries) elif STATE_FILE.exists(): STATE_FILE.unlink() @@ -838,27 +1112,44 @@ def cmd_status(args: argparse.Namespace) -> int: print( f"[{cli}] {server} = {spec.command} {' '.join(spec.args)} ({tag})" ) - except RuntimeError as exc: + except (RuntimeError, ValueError, OSError) as exc: print(f"[{cli}] {exc}", file=sys.stderr) continue return 0 def _describe_spec(spec: McpServerSpec, repo: pathlib.Path) -> str: - """Return a short label classifying a spec (local/pypi-pin/other).""" + """Return a short label classifying a spec (local/PR/pypi-pin/other).""" if spec.is_local_uv_directory(): local = spec.local_repo_path() if local and local.resolve() == repo.resolve(): return "local: this repo" return f"local: {local}" + pr = spec.pr_ref() + if pr is not None: + # Checked before the pin branch below: a PR ref contains `@`, + # which that branch would report as a version pin. + return f"PR #{pr[1]}: {pr[0]}" if spec.command == "uvx": pinned = next((a for a in spec.args if "==" in a or "@" in a), None) return f"pypi pin: {pinned}" if pinned else "pypi (unpinned)" return "other" -def cmd_use_local(args: argparse.Namespace) -> int: - """Rewrite each target CLI's config to run the repo's checkout via ``uv``. +def _points_at( + current: McpServerSpec, target: McpServerSpec, repo: pathlib.Path +) -> bool: + """Return True when ``current`` already runs what ``target`` describes.""" + if target.pr_ref() is not None: + return current.pr_ref() == target.pr_ref() + return current.is_local_uv_directory() and current.local_repo_path() == repo + + +def _cmd_use_local(args: argparse.Namespace) -> int: + """Rewrite each target CLI's config to run the repo, or a pull request. + + Without ``--pr`` the entry runs the repo's checkout via ``uv``; with + it, the pull request's head via ``uvx``. The optional ``--scope`` flag selects Claude's user-level fallback vs. per-project override; see :data:`Scope`. The flag is silently @@ -868,9 +1159,30 @@ def cmd_use_local(args: argparse.Namespace) -> int: server, default_entry = resolve_repo_meta(repo) server = args.server or server entry = args.entry or default_entry - spec = build_local_spec(repo, entry) extra_env = dict(args.env or []) + pr = getattr(args, "pr", None) + if pr is None: + spec = build_local_spec(repo, entry) + else: + try: + spec = build_pr_spec(remote_https_url(repo), pr, entry) + except RuntimeError as exc: + print(exc, file=sys.stderr) + return 1 + spec = dataclasses.replace(spec, env=dict(extra_env)) + summary = gh_pr_summary(repo, pr) + if summary is None: + print(f"PR #{pr}: gh could not read it — swapping anyway", file=sys.stderr) + else: + fork = " (fork)" if summary.get("isCrossRepository") else "" + print( + f"PR #{summary.get('number', pr)} [{summary.get('state', '?')}]" + f"{fork} {summary.get('headRefName', '?')} — " + f"{summary.get('title', '')}", + file=sys.stderr, + ) + hint = _naming_hint(repo, server) if hint: print(hint, file=sys.stderr) @@ -880,8 +1192,17 @@ def cmd_use_local(args: argparse.Namespace) -> int: print("no CLIs detected — nothing to do", file=sys.stderr) return 1 + # Runs under --dry-run too: resolving the ref is the only signal a + # dry run can give about whether the swap would actually start. + if pr is not None and not args.no_preflight: + print(f"preflight: {spec.command} {' '.join(spec.args)}", file=sys.stderr) + failure = preflight_spec(spec) + if failure is not None: + print(f"preflight failed, nothing written:\n{failure}", file=sys.stderr) + return 1 + ts = time.strftime("%Y%m%d%H%M%S") - state = load_state() + state = load_state(strict=True) had_error = 0 for cli in targets: scope = _normalize_scope(cli, args.scope) @@ -889,23 +1210,28 @@ def cmd_use_local(args: argparse.Namespace) -> int: info = CLIS[cli] if not info.config_path.exists(): print(f"[{label}] skip — config not found at {info.config_path}") + had_error = 1 continue - # Wrap the read + shape-guarded mutation in try/except RuntimeError - # so a malformed Claude config (top-level mcpServers / projects not a - # mapping) surfaces as a clean per-CLI error instead of an uncaught - # traceback. Same per-CLI continuation pattern the inner write-failure - # handler below uses. + target_path = info.config_path.resolve() + target_info = dataclasses.replace(info, config_path=target_path) + # Wrap the read + shape-guarded mutation so an unreadable config + # surfaces as a clean per-CLI error instead of an uncaught traceback. + # The three arms are the three ways it fails: a shape this script + # rejects raises RuntimeError, an unparseable one raises ValueError + # (JSON, TOML and UTF-8 decode errors all derive from it), and an + # unopenable one raises OSError. Same trio ``doctor`` catches, and + # the same per-CLI continuation the write-failure handler below uses. try: - original_bytes = info.config_path.read_bytes() - config = load_config(info) + original_bytes = target_path.read_bytes() + config = load_config(target_info) current = get_server(cli, config, server, repo, scope=scope) if ( current - and current.is_local_uv_directory() - and current.local_repo_path() == repo + and _points_at(current, spec, repo) and all(current.env.get(k) == v for k, v in extra_env.items()) ): - print(f"[{label}] already local (this repo) — no change") + where = "local (this repo)" if pr is None else f"PR #{pr}" + print(f"[{label}] already {where} — no change") continue # Preserve the existing entry's env on replacement. ``build_local_spec`` # writes an empty env, so without this merge a swap would silently drop @@ -920,8 +1246,8 @@ def cmd_use_local(args: argparse.Namespace) -> int: else spec ) action = set_server(cli, config, server, cli_spec, repo, scope=scope) - new_bytes = dump_config_bytes(info, config) - except RuntimeError as exc: + new_bytes = dump_config_bytes(info, config, original=original_bytes) + except (RuntimeError, ValueError, OSError) as exc: print(f"[{label}] {exc}", file=sys.stderr) had_error = 1 continue @@ -966,22 +1292,21 @@ def cmd_use_local(args: argparse.Namespace) -> int: backup_suffix = f"{BACKUP_SUFFIX_PREFIX}{ts}" if cli == "claude": backup_suffix += f"-{scope}" - backup_path = write_new_backup( - info.config_path.with_suffix(info.config_path.suffix + backup_suffix), - original_bytes, - ) + # A backup that cannot be written must abort this CLI rather + # than degrade into a swap with nothing to revert to — an + # unwritable directory is the case that produces both. + try: + backup_path = write_new_backup( + info.config_path.with_suffix( + info.config_path.suffix + backup_suffix + ), + original_bytes, + ) + except OSError as exc: + print(f"[{label}] cannot write backup: {exc}", file=sys.stderr) + had_error = 1 + continue backup_note = f"backup: {backup_path}" - try: - atomic_write(info.config_path, new_bytes) - _revalidate(info) - except Exception as exc: - atomic_write(info.config_path, original_bytes) - print( - f"[{label}] write failed ({exc}); backup at {backup_path}", - file=sys.stderr, - ) - had_error = 1 - continue if prior is not None and backup_path == prior_backup: # ``swapped_at`` mirrors the timestamp in the backup filename # and ``seq_no`` fixes the backup's place in the unwind @@ -990,27 +1315,76 @@ def cmd_use_local(args: argparse.Namespace) -> int: else: seq_no = max((e.seq_no for e in state.values()), default=-1) + 1 swapped_at = ts - state[(cli, scope)] = SwapEntry( + next_state = dict(state) + next_state[(cli, scope)] = SwapEntry( config_path=str(info.config_path), backup_path=str(backup_path), server=server, action=action, swapped_at=swapped_at, seq_no=seq_no, + target_path=str(target_path), ) + try: + save_state(next_state) + except OSError as exc: + print( + f"[{label}] cannot save recovery state ({exc}); config unchanged; " + f"backup at {backup_path}", + file=sys.stderr, + ) + had_error = 1 + continue + previous_state = state + state = next_state + try: + atomic_write(target_path, new_bytes) + _revalidate(target_info) + except Exception as exc: + try: + atomic_write(target_path, original_bytes) + except Exception as rollback_exc: + rollback_note = f"; rollback failed ({rollback_exc})" + else: + rollback_note = "; original config restored" + try: + _save_or_clear_state(previous_state) + except OSError as state_exc: + rollback_note += f"; recovery state cleanup failed ({state_exc})" + else: + state = previous_state + print( + f"[{label}] write failed ({exc}){rollback_note}; " + f"backup at {backup_path}", + file=sys.stderr, + ) + had_error = 1 + continue print(f"[{label}] {action}; {backup_note}") - if not args.dry_run: - save_state(state) return had_error +def cmd_use_local(args: argparse.Namespace) -> int: + """Run :func:`_cmd_use_local` under the shared mutation lock.""" + if args.dry_run: + return _cmd_use_local(args) + try: + with _state_lock(): + return _cmd_use_local(args) + except SwapStateError: + return 1 + except OSError as exc: + print(f"swap state unavailable: {exc}", file=sys.stderr) + return 1 + + def _revalidate(info: CLIInfo) -> None: """Re-parse the file after writing; raise on failure.""" load_config(info) -def cmd_revert(args: argparse.Namespace) -> int: +def _cmd_revert(args: argparse.Namespace) -> int: """Restore each target CLI's config from the backup recorded in the state file. Without ``--scope``, every recorded entry for the targeted CLIs is @@ -1019,14 +1393,14 @@ def cmd_revert(args: argparse.Namespace) -> int: the matching scope is reverted; the parameter is silently coerced to ``"user"`` for non-Claude CLIs. """ - state = load_state() + state = load_state(strict=True) # Without --cli, revert every CLI that has any recorded swap. targets = list(args.cli) if args.cli else list({cli for cli, _scope in state}) if not targets: print("no recorded swaps — nothing to revert", file=sys.stderr) return 1 - reverted: list[tuple[CLIName, Scope]] = [] + had_error = 0 for cli in targets: if args.scope is not None: wanted_scopes: tuple[Scope, ...] = (_normalize_scope(cli, args.scope),) @@ -1059,28 +1433,56 @@ def cmd_revert(args: argparse.Namespace) -> int: entry = state[key] label = f"{sc_cli}:{sc_scope}" if sc_cli == "claude" else sc_cli backup = pathlib.Path(entry.backup_path) - dest = pathlib.Path(entry.config_path) + dest = pathlib.Path(entry.target_path or entry.config_path) if not backup.exists(): print(f"[{label}] backup missing: {backup}", file=sys.stderr) - continue + had_error = 1 + break if args.dry_run: print(f"[{label}] would restore {dest} from {backup}") continue - atomic_write(dest, backup.read_bytes()) - # Backup served its purpose; LIFO unwind for this layer is - # complete. Delete on success, keep on error — same idiom - # CPython's ``tempfile.NamedTemporaryFile`` uses - # (Lib/tempfile.py:614-618). If ``atomic_write`` had raised, - # this line wouldn't run and the backup would survive for - # post-mortem; on success the backup is redundant and would - # otherwise accumulate forever across swap/revert cycles. - backup.unlink() + try: + atomic_write(dest, backup.read_bytes()) + except OSError as exc: + print(f"[{label}] restore failed: {exc}", file=sys.stderr) + had_error = 1 + break + next_state = dict(state) + next_state.pop(key) + try: + _save_or_clear_state(next_state) + except OSError as exc: + print( + f"[{label}] restored, but recovery state could not be updated: " + f"{exc}", + file=sys.stderr, + ) + had_error = 1 + break + state = next_state + try: + backup.unlink() + except OSError as exc: + print( + f"[{label}] restored; backup cleanup failed: {exc}", file=sys.stderr + ) + had_error = 1 print(f"[{label}] restored from {backup}") - reverted.append(key) + return had_error - if not args.dry_run and reverted: - clear_state(reverted) - return 0 + +def cmd_revert(args: argparse.Namespace) -> int: + """Run :func:`_cmd_revert` under the shared mutation lock.""" + if args.dry_run: + return _cmd_revert(args) + try: + with _state_lock(): + return _cmd_revert(args) + except SwapStateError: + return 1 + except OSError as exc: + print(f"swap state unavailable: {exc}", file=sys.stderr) + return 1 # --------------------------------------------------------------------------- @@ -1110,6 +1512,19 @@ def _env_pair(raw: str) -> tuple[str, str]: return key, value +def _pr_number(raw: str) -> int: + """Parse a ``--pr`` argument as a pull-request number, or raise for argparse.""" + try: + number = int(raw) + except ValueError: + msg = f"--pr expects a number, got {raw!r}" + raise argparse.ArgumentTypeError(msg) from None + if number < 1: + msg = f"--pr expects a positive number, got {number}" + raise argparse.ArgumentTypeError(msg) + return number + + def _config_present_clis() -> list[CLIName]: """CLIs whose config file exists — enough to *read* entries (no binary needed). @@ -1318,8 +1733,30 @@ def build_parser() -> argparse.ArgumentParser: ) ps.set_defaults(func=cmd_status) - pu = sub.add_parser("use-local", help="rewrite configs to run this checkout") + pu = sub.add_parser( + "use-local", help="rewrite configs to run this checkout, or a pull request" + ) pu.add_argument("--repo", default=".", help="repo root (default: .)") + pu.add_argument( + "--pr", + type=_pr_number, + metavar="N", + help=( + "Point the CLIs at pull request N instead of the working copy. " + "Writes 'uvx --from git+@refs/pull/N/head ', so " + "nothing is checked out and 'revert' needs no cleanup. The ref " + "lives on the base repo, so fork PRs work unchanged." + ), + ) + pu.add_argument( + "--no-preflight", + action="store_true", + help=( + "Skip the MCP initialize round trip --pr runs before writing. " + "The probe resolves the ref once so a bad PR fails here instead " + "of inside every agent; skip it when offline or already warm." + ), + ) pu.add_argument( "--server", help="MCP server name (default: derived from pyproject.toml)" ) diff --git a/tests/test_mcp_swap.py b/tests/test_mcp_swap.py index a54139b..5291f63 100644 --- a/tests/test_mcp_swap.py +++ b/tests/test_mcp_swap.py @@ -7,10 +7,14 @@ from __future__ import annotations +import argparse import importlib.util import json +import os import pathlib import sys +import threading +import time import types import typing as t @@ -849,6 +853,58 @@ def test_save_state_writes_atomically(fake_home: pathlib.Path) -> None: assert leftovers == [], f"unexpected tempfile leftovers: {leftovers}" +def test_use_local_serializes_the_full_state_transaction( + fake_home: pathlib.Path, + fake_repo: pathlib.Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Concurrent commands cannot lose one another's recovery records.""" + for cli in ("cursor", "gemini"): + _write_json( + mcp_swap.CLIS[cli].config_path, + {"mcpServers": {"libtmux": _pinned_json_entry()}}, + ) + parser = mcp_swap.build_parser() + real_save_state = mcp_swap.save_state + guard = threading.Lock() + gate = threading.Barrier(3) + active = 0 + overlapped = False + results: list[int] = [] + + def slow_save_state(entries: dict[t.Any, t.Any]) -> None: + nonlocal active, overlapped + with guard: + active += 1 + overlapped = overlapped or active > 1 + try: + time.sleep(0.1) + real_save_state(entries) + finally: + with guard: + active -= 1 + + def swap(cli: str) -> None: + args = parser.parse_args(["use-local", "--repo", str(fake_repo), "--cli", cli]) + gate.wait() + results.append(mcp_swap.cmd_use_local(args)) + + monkeypatch.setattr(mcp_swap, "save_state", slow_save_state) + threads = [ + threading.Thread(target=swap, args=(cli,)) for cli in ("cursor", "gemini") + ] + for thread in threads: + thread.start() + gate.wait() + for thread in threads: + thread.join(timeout=2) + + assert results == [0, 0] + assert not overlapped + assert set(mcp_swap.load_state()) == {("cursor", "user"), ("gemini", "user")} + assert all(not thread.is_alive() for thread in threads) + + # --------------------------------------------------------------------------- # McpServerSpec helpers # --------------------------------------------------------------------------- @@ -1142,20 +1198,20 @@ def test_load_state_drops_entries_with_missing_required_fields( assert state == {} -def test_revert_with_corrupt_seq_no_does_not_crash( +def test_revert_with_corrupt_seq_no_preserves_every_recovery_layer( fake_home: pathlib.Path, fake_repo: pathlib.Path, ) -> None: - """Same-CLI two-scope state with one corrupt ``seq_no`` does not raise TypeError. + """Same-file recovery stops when one layer has a corrupt ``seq_no``. Regression: the LIFO sort at ``cmd_revert`` would compare ``int`` vs ``str`` (``int < str`` raises in Python 3) when two same-CLI entries existed and one had a hand-edited corrupt counter. Cross-CLI buckets are length-1 and never invoke comparison — making the failure mode asymmetric, only triggering on Claude - project + user. Validating at load time eliminates the - asymmetry: the corrupt entry is dropped before it reaches the - sort, so the well-formed entry's revert applies normally. + project + user. Dropping that layer and applying the other backup + would violate LIFO order, so mutation is refused while all recovery + material remains intact. """ info = mcp_swap.CLIS["claude"] _write_json( @@ -1195,15 +1251,16 @@ def test_revert_with_corrupt_seq_no_does_not_crash( == 0 ) - # Hand-edit one of the two entries to corrupt seq_no. + before_config = info.config_path.read_bytes() raw = json.loads(mcp_swap.STATE_FILE.read_text()) raw["entries"]["claude:user"]["seq_no"] = "not-an-int" mcp_swap.STATE_FILE.write_text(json.dumps(raw)) + corrupt_state = mcp_swap.STATE_FILE.read_bytes() - # Revert must NOT raise TypeError. The corrupt entry is silently - # dropped at load time; the well-formed entry's revert applies. rc = mcp_swap.cmd_revert(parser.parse_args(["revert", "--cli", "claude"])) - assert rc == 0 + assert rc == 1 + assert info.config_path.read_bytes() == before_config + assert mcp_swap.STATE_FILE.read_bytes() == corrupt_state # --------------------------------------------------------------------------- @@ -1273,6 +1330,40 @@ def test_revert_dry_run_keeps_backup( assert backup.exists() +def test_revert_returns_failure_when_the_recorded_backup_is_missing( + fake_home: pathlib.Path, fake_repo: pathlib.Path +) -> None: + """Automation receives a nonzero status when recovery cannot complete.""" + info = mcp_swap.CLIS["cursor"] + _write_json(info.config_path, {"mcpServers": {"libtmux": _pinned_json_entry()}}) + parser = mcp_swap.build_parser() + assert ( + mcp_swap.cmd_use_local( + parser.parse_args( + ["use-local", "--repo", str(fake_repo), "--cli", "cursor"] + ) + ) + == 0 + ) + backup = pathlib.Path(mcp_swap.load_state()["cursor", "user"].backup_path) + backup.unlink() + + assert mcp_swap.cmd_revert(parser.parse_args(["revert", "--cli", "cursor"])) == 1 + assert ("cursor", "user") in mcp_swap.load_state() + + +def test_explicit_missing_config_returns_failure_without_creating_state( + fake_home: pathlib.Path, fake_repo: pathlib.Path +) -> None: + """An explicitly requested absent config is an error, not a successful no-op.""" + args = mcp_swap.build_parser().parse_args( + ["use-local", "--repo", str(fake_repo), "--cli", "cursor"] + ) + + assert mcp_swap.cmd_use_local(args) == 1 + assert not mcp_swap.STATE_FILE.exists() + + # --------------------------------------------------------------------------- # `status --scope` filter — completes symmetry with use-local / revert. # --------------------------------------------------------------------------- @@ -1969,3 +2060,887 @@ def swap(value: str) -> None: assert "recorded backup is gone" in capsys.readouterr().err fresh = pathlib.Path(mcp_swap.load_state()[("cursor", "user")].backup_path) assert fresh.read_bytes() == swapped_bytes + + +# --------------------------------------------------------------------------- +# Pull-request targeting +# --------------------------------------------------------------------------- + + +class RemoteURLFixture(t.NamedTuple): + """One git remote spelling and the https URL it normalizes to. + + Attributes + ---------- + test_id : str + Identifier shown in the parametrized test name. + remote : str + A URL as ``git remote get-url`` may report it. + expected : str + The https form the pull-request ref is fetched from. + """ + + test_id: str + remote: str + expected: str + + +REMOTE_URL_FIXTURES: list[RemoteURLFixture] = [ + RemoteURLFixture( + "git_ssh_scheme", "git+ssh://git@github.com/o/n.git", "https://github.com/o/n" + ), + RemoteURLFixture( + "ssh_scheme", "ssh://git@github.com/o/n.git", "https://github.com/o/n" + ), + RemoteURLFixture( + "scp_shorthand", "git@github.com:o/n.git", "https://github.com/o/n" + ), + RemoteURLFixture( + "https_dotgit", "https://github.com/o/n.git", "https://github.com/o/n" + ), + RemoteURLFixture("https_plain", "https://github.com/o/n", "https://github.com/o/n"), + RemoteURLFixture( + "self_hosted", + "git@git.example.com:team/n.git", + "https://git.example.com/team/n", + ), +] + + +@pytest.mark.parametrize( + RemoteURLFixture._fields, + REMOTE_URL_FIXTURES, + ids=[f.test_id for f in REMOTE_URL_FIXTURES], +) +def test_normalize_remote_url(test_id: str, remote: str, expected: str) -> None: + """Every spelling git accepts resolves to the same https URL.""" + assert test_id + assert mcp_swap._normalize_remote_url(remote) == expected + + +def test_build_pr_spec_round_trips_through_pr_ref() -> None: + """A built pull-request spec is recognized by the reader that parses it.""" + spec = mcp_swap.build_pr_spec("https://github.com/o/n", 114, "libtmux-mcp") + + assert spec.command == "uvx" + assert spec.args == [ + "--from", + "git+https://github.com/o/n@refs/pull/114/head", + "libtmux-mcp", + ] + assert spec.pr_ref() == ("https://github.com/o/n", 114) + assert spec.is_local_uv_directory() is False + + +def test_pr_ref_ignores_non_pr_specs() -> None: + """A local checkout and a version pin are not pull-request specs.""" + local = mcp_swap.McpServerSpec( + command="uv", args=["--directory", "/tmp", "run", "x"] + ) + pinned = mcp_swap.McpServerSpec(command="uvx", args=["libtmux-mcp==0.1.0a2"]) + branch = mcp_swap.McpServerSpec( + command="uvx", args=["--from", "git+https://github.com/o/n@main", "x"] + ) + + assert local.pr_ref() is None + assert pinned.pr_ref() is None + assert branch.pr_ref() is None + + +def test_describe_spec_labels_a_pr_before_the_version_pin_branch( + tmp_path: pathlib.Path, +) -> None: + """A pull-request ref is described as a PR, not as a version pin. + + The ref carries an ``@``, which the pin branch would otherwise report + as ``pypi pin: git+...@refs/pull/114/head``. + """ + spec = mcp_swap.build_pr_spec("https://github.com/o/n", 114, "libtmux-mcp") + + assert mcp_swap._describe_spec(spec, tmp_path) == "PR #114: https://github.com/o/n" + + +def test_points_at_distinguishes_pr_numbers(tmp_path: pathlib.Path) -> None: + """A swap to one pull request is not treated as already pointing at another.""" + target = mcp_swap.build_pr_spec("https://github.com/o/n", 114, "x") + same = mcp_swap.build_pr_spec("https://github.com/o/n", 114, "x") + other = mcp_swap.build_pr_spec("https://github.com/o/n", 115, "x") + local = mcp_swap.build_local_spec(tmp_path, "x") + + assert mcp_swap._points_at(same, target, tmp_path) is True + assert mcp_swap._points_at(other, target, tmp_path) is False + assert mcp_swap._points_at(local, target, tmp_path) is False + assert mcp_swap._points_at(local, local, tmp_path) is True + + +def test_preflight_accepts_a_server_that_answers_initialize( + tmp_path: pathlib.Path, +) -> None: + """A stdio server that replies to ``initialize`` passes preflight.""" + server = tmp_path / "server.py" + server.write_text( + "import json, sys\n" + "line = sys.stdin.readline()\n" + "req = json.loads(line)\n" + 'print(json.dumps({"jsonrpc": "2.0", "id": req["id"], "result": {}}))\n', + encoding="utf-8", + ) + spec = mcp_swap.McpServerSpec(command=sys.executable, args=[str(server)]) + + assert mcp_swap.preflight_spec(spec, timeout=60) is None + + +def test_preflight_reports_stderr_when_the_server_never_answers( + tmp_path: pathlib.Path, +) -> None: + """A server that dies is reported with the tail of its stderr.""" + server = tmp_path / "server.py" + server.write_text( + 'import sys\nsys.stderr.write("could not resolve ref\\n")\nsys.exit(1)\n', + encoding="utf-8", + ) + spec = mcp_swap.McpServerSpec(command=sys.executable, args=[str(server)]) + + assert mcp_swap.preflight_spec(spec, timeout=60) == "could not resolve ref" + + +def test_preflight_reports_a_command_that_cannot_launch() -> None: + """A missing binary is named rather than raising.""" + spec = mcp_swap.McpServerSpec(command="mcp-swap-no-such-binary", args=[]) + + failure = mcp_swap.preflight_spec(spec, timeout=60) + + assert failure is not None + assert "mcp-swap-no-such-binary" in failure + + +def test_preflight_passes_spec_env_to_the_process(tmp_path: pathlib.Path) -> None: + """``spec.env`` reaches the launched server. + + The cooldown bypass a prerelease branch needs travels this way, so a + preflight that dropped it would reject a spec that works in an agent. + """ + server = tmp_path / "server.py" + server.write_text( + "import json, os, sys\n" + "req = json.loads(sys.stdin.readline())\n" + 'if os.environ.get("MCP_SWAP_PROBE") != "1":\n' + " sys.exit(2)\n" + 'print(json.dumps({"jsonrpc": "2.0", "id": req["id"], "result": {}}))\n', + encoding="utf-8", + ) + spec = mcp_swap.McpServerSpec( + command=sys.executable, args=[str(server)], env={"MCP_SWAP_PROBE": "1"} + ) + + assert mcp_swap.preflight_spec(spec, timeout=60) is None + + +# --------------------------------------------------------------------------- +# JSON writer fidelity +# +# The swap edits one entry inside a file the user owns, so bytes it did +# not set out to change must survive the rewrite. ``load_config`` -> +# ``dump_config_bytes`` is the whole write path, so an unmodified config +# has to come back byte-identical. +# +# Out of scope, and normalized rather than preserved: indent width, CRLF, +# `\/` and `\uXXXX` escapes of characters that need none, duplicate keys, +# and number spelling (`1e5` -> `100000.0`). None appear in what the JSON +# CLIs write — they all emit `JSON.stringify(x, null, 2)` — and none +# change what a CLI reads, only the bytes a dotfile diff shows. +# --------------------------------------------------------------------------- + + +class JSONFidelityCase(t.NamedTuple): + """A JSON config body whose exact bytes survive a no-op rewrite. + + Attributes + ---------- + test_id : str + Identifier shown in the parametrized test name. + body : str + The config file's text, written to disk verbatim. + """ + + test_id: str + body: str + + +PRESERVED_JSON: list[JSONFidelityCase] = [ + JSONFidelityCase( + "mcp_servers_block", + '{\n "mcpServers": {\n "libtmux": {\n "command": "uvx",\n' + ' "args": [\n "libtmux-mcp==0.1.0a2"\n ]\n }\n }\n}\n', + ), + JSONFidelityCase( + "non_ascii_model_label", + '{\n "model": "Fable 5 · Most capable…",\n "mcpServers": {}\n}\n', + ), + JSONFidelityCase( + "emoji_and_cjk", '{\n "history": [\n "🙂 日本語 café"\n ]\n}\n' + ), + JSONFidelityCase("escaped_lone_surrogate", '{\n "truncated": "\\ud800"\n}\n'), + JSONFidelityCase("unsorted_keys", '{\n "zeta": 1,\n "alpha": 2\n}\n'), + JSONFidelityCase( + "claude_shape_without_trailing_newline", + '{\n "model": "Fable 5 · Most capable…",\n "projects": {\n' + ' "/home/someone/repo": {\n "mcpServers": {}\n }\n }\n}', + ), +] + + +def _json_config(tmp_path: pathlib.Path, body: str) -> tuple[t.Any, bytes]: + """Write ``body`` verbatim and return its ``CLIInfo`` and exact bytes.""" + path = tmp_path / "config.json" + raw = body.encode() + path.write_bytes(raw) + info = mcp_swap.CLIInfo( + name="cursor", binary="cursor-agent", config_path=path, fmt="json" + ) + return info, raw + + +@pytest.mark.parametrize( + JSONFidelityCase._fields, + PRESERVED_JSON, + ids=[c.test_id for c in PRESERVED_JSON], +) +def test_untouched_json_config_round_trips_byte_identical( + tmp_path: pathlib.Path, test_id: str, body: str +) -> None: + """Parsing a config and writing it back unmodified changes nothing. + + Every case is a shape the JavaScript agent CLIs actually emit: + two-space indent, literal non-ASCII, escapes only below ``0x20`` plus + lone surrogates, and no terminating newline. + """ + assert test_id + info, raw = _json_config(tmp_path, body) + + assert ( + mcp_swap.dump_config_bytes(info, mcp_swap.load_config(info), original=raw) + == raw + ) + + +def test_dump_config_bytes_ends_a_seeded_file_with_a_newline( + tmp_path: pathlib.Path, +) -> None: + """With no original to match, a JSON config gets the conventional newline.""" + info, _ = _json_config(tmp_path, "") + + assert ( + mcp_swap.dump_config_bytes(info, {"mcpServers": {}}, original=b"") + == b'{\n "mcpServers": {}\n}\n' + ) + + +def test_dump_config_bytes_escapes_a_config_it_cannot_encode( + tmp_path: pathlib.Path, +) -> None: + r"""A lone surrogate has no UTF-8 form, so the document is escaped instead. + + JavaScript writes a string sliced through a surrogate pair as + ``"\ud800"``, which parses to a Python string ``str.encode`` rejects. + Escaping the whole document is what keeps the file writable at all. + """ + config = {"truncated": "\ud800", "label": "café"} + + with pytest.raises(UnicodeEncodeError): + json.dumps(config, indent=2, ensure_ascii=False).encode() + + info, _ = _json_config(tmp_path, "") + written = mcp_swap.dump_config_bytes(info, config, original=b"") + + assert written == b'{\n "truncated": "\\ud800",\n "label": "caf\\u00e9"\n}\n' + assert json.loads(written.decode()) == config + + +def test_swap_leaves_non_ascii_elsewhere_in_the_config_alone( + fake_home: pathlib.Path, fake_repo: pathlib.Path +) -> None: + """A real swap does not re-escape config text it never read. + + Claude stores model labels and prompt history alongside the MCP + entries, so escaping on write turns a one-entry edit into a diff + spanning the file. + """ + info = mcp_swap.CLIS["claude"] + label = "Fable 5 · Most capable…" + _write_json( + info.config_path, + { + "model": label, + "projects": { + str(fake_repo.resolve()): { + "mcpServers": {"libtmux": _pinned_claude_entry()}, + "history": ["café ☕"], + } + }, + }, + ) + args = mcp_swap.build_parser().parse_args( + ["use-local", "--repo", str(fake_repo), "--cli", "claude"] + ) + + assert mcp_swap.cmd_use_local(args) == 0 + + after = info.config_path.read_text() + assert f'"model": "{label}"' in after + assert '"café ☕"' in after + assert "\\u" not in after + + +def test_swap_does_not_append_a_newline_the_cli_never_wrote( + fake_home: pathlib.Path, fake_repo: pathlib.Path +) -> None: + """Claude's config has no trailing newline, and swapping must not add one.""" + info = mcp_swap.CLIS["claude"] + body = json.dumps( + { + "projects": { + str(fake_repo.resolve()): { + "mcpServers": {"libtmux": _pinned_claude_entry()} + } + } + }, + indent=2, + ) + info.config_path.parent.mkdir(parents=True, exist_ok=True) + info.config_path.write_text(body) + + args = mcp_swap.build_parser().parse_args( + ["use-local", "--repo", str(fake_repo), "--cli", "claude"] + ) + assert mcp_swap.cmd_use_local(args) == 0 + + assert not info.config_path.read_bytes().endswith(b"\n") + + +class UnreadableConfigCase(t.NamedTuple): + """A config body that cannot be parsed, and the error it provokes. + + Attributes + ---------- + test_id : str + Identifier shown in the parametrized test name. + body : bytes + Exact bytes written to the config file. + """ + + test_id: str + body: bytes + + +UNREADABLE_CONFIGS: list[UnreadableConfigCase] = [ + UnreadableConfigCase("malformed_json", b"{ this is not json"), + UnreadableConfigCase("truncated_json", b'{"mcpServers": {'), + UnreadableConfigCase("invalid_utf8", b'{"a": "\xff\xfe"}'), +] + + +@pytest.mark.parametrize( + UnreadableConfigCase._fields, + UNREADABLE_CONFIGS, + ids=[c.test_id for c in UNREADABLE_CONFIGS], +) +def test_unreadable_config_reports_instead_of_crashing( + fake_home: pathlib.Path, + fake_repo: pathlib.Path, + capsys: pytest.CaptureFixture[str], + test_id: str, + body: bytes, +) -> None: + """A config that will not parse is reported and skipped, not raised through. + + ``load_config`` raises ``ValueError`` for every unparseable form — + JSON, TOML and UTF-8 decode errors all derive from it — which the + per-CLI handler has to catch for the run to survive one bad file. + """ + assert test_id + info = mcp_swap.CLIS["cursor"] + info.config_path.parent.mkdir(parents=True, exist_ok=True) + info.config_path.write_bytes(body) + args = mcp_swap.build_parser().parse_args( + ["use-local", "--repo", str(fake_repo), "--cli", "cursor"] + ) + + assert mcp_swap.cmd_use_local(args) == 1 + + assert "cursor" in capsys.readouterr().err + assert info.config_path.read_bytes() == body + + +def test_unreadable_config_does_not_stop_the_other_clis( + fake_home: pathlib.Path, + fake_repo: pathlib.Path, +) -> None: + """One bad config does not prevent the remaining CLIs from swapping.""" + bad = mcp_swap.CLIS["cursor"] + bad.config_path.parent.mkdir(parents=True, exist_ok=True) + bad.config_path.write_bytes(b"{ not json") + good = mcp_swap.CLIS["gemini"] + _write_json(good.config_path, {"mcpServers": {}}) + args = mcp_swap.build_parser().parse_args( + ["use-local", "--repo", str(fake_repo), "--cli", "cursor", "--cli", "gemini"] + ) + + assert mcp_swap.cmd_use_local(args) == 1 + + written = json.loads(good.config_path.read_text()) + assert "libtmux" in written["mcpServers"] + + +class CorruptStateCase(t.NamedTuple): + """A swap-state file body that cannot yield entries. + + Attributes + ---------- + test_id : str + Identifier shown in the parametrized test name. + body : str + Exact text written to the state file. + """ + + test_id: str + body: str + + +CORRUPT_STATE: list[CorruptStateCase] = [ + CorruptStateCase("not_json", "{ not json at all"), + CorruptStateCase("empty_file", ""), + CorruptStateCase("json_but_a_list", "[1, 2, 3]"), + CorruptStateCase("entries_not_a_mapping", '{"entries": "nope"}'), +] + + +@pytest.mark.parametrize( + CorruptStateCase._fields, + CORRUPT_STATE, + ids=[c.test_id for c in CORRUPT_STATE], +) +def test_corrupt_swap_state_is_reported_not_raised( + fake_home: pathlib.Path, + test_id: str, + body: str, +) -> None: + """A state file that yields no entries degrades to empty, never raises. + + ``revert`` and ``doctor`` both read this file before doing anything, + so a hand-edited or truncated one would otherwise take down every + command that consults it. + """ + assert test_id + mcp_swap.STATE_FILE.parent.mkdir(parents=True, exist_ok=True) + mcp_swap.STATE_FILE.write_text(body, encoding="utf-8") + + assert mcp_swap.load_state() == {} + + +def test_unparseable_swap_state_names_the_file( + fake_home: pathlib.Path, + capsys: pytest.CaptureFixture[str], +) -> None: + """An unreadable state file says so, because backups are now orphaned. + + Returning empty silently would let ``revert`` report nothing to do + while swapped configs and their backups sit on disk. + """ + mcp_swap.STATE_FILE.parent.mkdir(parents=True, exist_ok=True) + mcp_swap.STATE_FILE.write_text("{ not json", encoding="utf-8") + + mcp_swap.load_state() + + assert str(mcp_swap.STATE_FILE) in capsys.readouterr().err + + +def test_revert_survives_a_corrupt_state_file( + fake_home: pathlib.Path, + fake_repo: pathlib.Path, +) -> None: + """``revert`` reports nothing to unwind rather than crashing.""" + info = mcp_swap.CLIS["cursor"] + _write_json(info.config_path, {"mcpServers": {}}) + args = mcp_swap.build_parser().parse_args( + ["use-local", "--repo", str(fake_repo), "--cli", "cursor"] + ) + assert mcp_swap.cmd_use_local(args) == 0 + mcp_swap.STATE_FILE.write_text("{ corrupted", encoding="utf-8") + + revert_args = mcp_swap.build_parser().parse_args(["revert", "--cli", "cursor"]) + + assert mcp_swap.cmd_revert(revert_args) in (0, 1) + + +def test_corrupt_state_blocks_a_new_swap_without_touching_the_config( + fake_home: pathlib.Path, fake_repo: pathlib.Path +) -> None: + """Unreadable recovery bookkeeping is never overwritten as empty state.""" + info = mcp_swap.CLIS["cursor"] + _write_json(info.config_path, {"mcpServers": {"libtmux": _pinned_json_entry()}}) + original = info.config_path.read_bytes() + corrupt = b"{ not json" + mcp_swap.STATE_FILE.parent.mkdir(parents=True, exist_ok=True) + mcp_swap.STATE_FILE.write_bytes(corrupt) + args = mcp_swap.build_parser().parse_args( + ["use-local", "--repo", str(fake_repo), "--cli", "cursor"] + ) + + assert mcp_swap.cmd_use_local(args) == 1 + assert info.config_path.read_bytes() == original + assert mcp_swap.STATE_FILE.read_bytes() == corrupt + + +def test_unwritable_directory_aborts_before_swapping( + fake_home: pathlib.Path, + fake_repo: pathlib.Path, + capsys: pytest.CaptureFixture[str], +) -> None: + """A config whose backup cannot be written is left alone. + + The backup is the only copy of the pre-swap config, so a swap that + could not take one would leave nothing to revert to. Aborting the + CLI is the safe half of that trade. + """ + if os.geteuid() == 0: + pytest.skip("root ignores directory permissions") + info = mcp_swap.CLIS["grok"] + info.config_path.parent.mkdir(parents=True, exist_ok=True) + original = '[mcp_servers.other]\ncommand = "x"\n' + info.config_path.write_text(original, encoding="utf-8") + info.config_path.parent.chmod(0o500) + args = mcp_swap.build_parser().parse_args( + ["use-local", "--repo", str(fake_repo), "--cli", "grok"] + ) + + try: + assert mcp_swap.cmd_use_local(args) == 1 + assert "backup" in capsys.readouterr().err + assert info.config_path.read_text() == original + finally: + info.config_path.parent.chmod(0o700) + + +def test_unwritable_directory_does_not_stop_the_other_clis( + fake_home: pathlib.Path, + fake_repo: pathlib.Path, +) -> None: + """One unwritable config directory does not abort the whole run.""" + if os.geteuid() == 0: + pytest.skip("root ignores directory permissions") + blocked = mcp_swap.CLIS["grok"] + blocked.config_path.parent.mkdir(parents=True, exist_ok=True) + blocked.config_path.write_text('[mcp_servers.o]\ncommand = "x"\n', encoding="utf-8") + blocked.config_path.parent.chmod(0o500) + reachable = mcp_swap.CLIS["cursor"] + _write_json(reachable.config_path, {"mcpServers": {}}) + args = mcp_swap.build_parser().parse_args( + ["use-local", "--repo", str(fake_repo), "--cli", "grok", "--cli", "cursor"] + ) + + try: + assert mcp_swap.cmd_use_local(args) == 1 + written = json.loads(reachable.config_path.read_text()) + assert "libtmux" in written["mcpServers"] + finally: + blocked.config_path.parent.chmod(0o700) + + +def test_state_write_failure_leaves_the_config_unchanged( + fake_home: pathlib.Path, + fake_repo: pathlib.Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """A swap is not applied until its recovery record is durable.""" + info = mcp_swap.CLIS["cursor"] + _write_json(info.config_path, {"mcpServers": {"libtmux": _pinned_json_entry()}}) + original = info.config_path.read_bytes() + real_atomic_write = mcp_swap.atomic_write + write_error = PermissionError("state is read-only") + + def fail_state_write(path: pathlib.Path, data: bytes) -> None: + if path == mcp_swap.STATE_FILE: + raise write_error + real_atomic_write(path, data) + + monkeypatch.setattr(mcp_swap, "atomic_write", fail_state_write) + args = mcp_swap.build_parser().parse_args( + ["use-local", "--repo", str(fake_repo), "--cli", "cursor"] + ) + + assert mcp_swap.cmd_use_local(args) == 1 + assert info.config_path.read_bytes() == original + assert not mcp_swap.STATE_FILE.exists() + + +def test_swap_write_failure_keeps_recovery_state_without_raising( + fake_home: pathlib.Path, + fake_repo: pathlib.Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """A failed config write remains recoverable even when rollback also fails.""" + info = mcp_swap.CLIS["cursor"] + _write_json(info.config_path, {"mcpServers": {"libtmux": _pinned_json_entry()}}) + original = info.config_path.read_bytes() + target = info.config_path.resolve() + real_atomic_write = mcp_swap.atomic_write + write_error = PermissionError("config is read-only") + + def fail_config_write(path: pathlib.Path, data: bytes) -> None: + if path == target: + raise write_error + real_atomic_write(path, data) + + monkeypatch.setattr(mcp_swap, "atomic_write", fail_config_write) + args = mcp_swap.build_parser().parse_args( + ["use-local", "--repo", str(fake_repo), "--cli", "cursor"] + ) + + assert mcp_swap.cmd_use_local(args) == 1 + state = mcp_swap.load_state() + assert info.config_path.read_bytes() == original + assert pathlib.Path(state["cursor", "user"].backup_path).exists() + + +def test_revert_write_failure_returns_failure_and_keeps_recovery_files( + fake_home: pathlib.Path, + fake_repo: pathlib.Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """An unwritable destination does not crash or discard recovery material.""" + info = mcp_swap.CLIS["cursor"] + _write_json(info.config_path, {"mcpServers": {"libtmux": _pinned_json_entry()}}) + parser = mcp_swap.build_parser() + assert ( + mcp_swap.cmd_use_local( + parser.parse_args( + ["use-local", "--repo", str(fake_repo), "--cli", "cursor"] + ) + ) + == 0 + ) + state = mcp_swap.load_state() + backup = pathlib.Path(state["cursor", "user"].backup_path) + target = info.config_path.resolve() + real_atomic_write = mcp_swap.atomic_write + write_error = PermissionError("config is read-only") + + def fail_config_write(path: pathlib.Path, data: bytes) -> None: + if path == target: + raise write_error + real_atomic_write(path, data) + + monkeypatch.setattr(mcp_swap, "atomic_write", fail_config_write) + + assert mcp_swap.cmd_revert(parser.parse_args(["revert", "--cli", "cursor"])) == 1 + assert backup.exists() + assert ("cursor", "user") in mcp_swap.load_state() + + +def test_revert_state_failure_keeps_recovery_files( + fake_home: pathlib.Path, + fake_repo: pathlib.Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """A restored config keeps its backup until state cleanup is durable.""" + info = mcp_swap.CLIS["cursor"] + _write_json(info.config_path, {"mcpServers": {"libtmux": _pinned_json_entry()}}) + original = info.config_path.read_bytes() + parser = mcp_swap.build_parser() + assert ( + mcp_swap.cmd_use_local( + parser.parse_args( + ["use-local", "--repo", str(fake_repo), "--cli", "cursor"] + ) + ) + == 0 + ) + state_bytes = mcp_swap.STATE_FILE.read_bytes() + backup = pathlib.Path(mcp_swap.load_state()["cursor", "user"].backup_path) + state_error = PermissionError("state is read-only") + + def fail_state_update(_entries: dict[t.Any, t.Any]) -> None: + raise state_error + + monkeypatch.setattr(mcp_swap, "_save_or_clear_state", fail_state_update) + + assert mcp_swap.cmd_revert(parser.parse_args(["revert", "--cli", "cursor"])) == 1 + assert info.config_path.read_bytes() == original + assert mcp_swap.STATE_FILE.read_bytes() == state_bytes + assert backup.exists() + + +@pytest.mark.parametrize("raw", ["0", "-5", "notanumber", "1.5", ""]) +def test_pr_number_rejects_what_is_not_a_pull_request(raw: str) -> None: + """``--pr`` takes a positive number; anything else stops at the parser. + + Pull requests are numbered from one, so a non-positive value can only + be a typo. Catching it here keeps it out of the ref the swap builds. + """ + with pytest.raises(argparse.ArgumentTypeError): + mcp_swap._pr_number(raw) + + +def test_pr_number_accepts_a_pull_request_number() -> None: + """A positive number parses to an int.""" + assert mcp_swap._pr_number("115") == 115 + + +@pytest.mark.parametrize("raw", ["0", "-5", "notanumber"]) +def test_parser_rejects_a_bad_pr_argument(raw: str) -> None: + """The parser exits rather than building a ref from a bad number.""" + with pytest.raises(SystemExit): + mcp_swap.build_parser().parse_args(["use-local", "--pr", raw]) + + +# --------------------------------------------------------------------------- +# Atomic writes through symlinked configs +# --------------------------------------------------------------------------- + + +def _build_symlink_chain( + root: pathlib.Path, hops: int +) -> tuple[pathlib.Path, pathlib.Path, list[pathlib.Path]]: + """Create ``hops`` links ending at an existing config file. + + Parameters + ---------- + root : pathlib.Path + Empty directory where the link and target trees are created. + hops : int + Number of links in the chain. + + Returns + ------- + tuple of pathlib.Path, pathlib.Path, list of pathlib.Path + Entry path, final target, and each link in the chain. + """ + target = root / "dotfiles" / "mcp.json" + target.parent.mkdir(parents=True) + target.write_bytes(b"original\n") + link_dir = root / "home" + link_dir.mkdir() + links: list[pathlib.Path] = [] + entry = target + for hop in range(hops): + link = link_dir / f"hop-{hop}.json" + link.symlink_to(entry) + links.append(link) + entry = link + return entry, target, links + + +@pytest.mark.parametrize("hops", [1, 3], ids=["single", "chain"]) +def test_atomic_write_updates_the_symlink_target( + tmp_path: pathlib.Path, hops: int +) -> None: + """The final target receives the bytes and every link survives.""" + entry, target, links = _build_symlink_chain(tmp_path, hops) + + mcp_swap.atomic_write(entry, b"swapped\n") + + assert all(link.is_symlink() for link in links) + assert target.read_bytes() == b"swapped\n" + assert entry.read_bytes() == b"swapped\n" + + +def test_atomic_write_stages_beside_the_symlink_target( + tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """The temp file shares the final target's filesystem for atomic rename.""" + entry, target, _links = _build_symlink_chain(tmp_path, 1) + real_mkstemp = mcp_swap.tempfile.mkstemp + staged_in: list[str | None] = [] + + def recording_mkstemp(*args: t.Any, **kwargs: t.Any) -> tuple[int, str]: + staged_in.append(kwargs.get("dir")) + return t.cast(tuple[int, str], real_mkstemp(*args, **kwargs)) + + monkeypatch.setattr(mcp_swap.tempfile, "mkstemp", recording_mkstemp) + + mcp_swap.atomic_write(entry, b"swapped\n") + + assert staged_in == [str(target.parent)] + + +def test_atomic_write_preserves_the_target_mode(tmp_path: pathlib.Path) -> None: + """Replacing a config does not silently narrow its permission bits.""" + target = tmp_path / "mcp.json" + target.write_bytes(b"original\n") + target.chmod(0o640) + + mcp_swap.atomic_write(target, b"swapped\n") + + assert target.stat().st_mode & 0o777 == 0o640 + + +def test_symlinked_config_swap_and_revert_round_trip( + fake_home: pathlib.Path, fake_repo: pathlib.Path +) -> None: + """Swap and revert update the target without replacing the config link.""" + info = mcp_swap.CLIS["cursor"] + target = fake_home / "dotfiles" / "cursor" / "mcp.json" + _write_json(target, {"mcpServers": {"libtmux": _pinned_json_entry()}}) + original = target.read_bytes() + info.config_path.parent.mkdir(parents=True) + info.config_path.symlink_to(target) + parser = mcp_swap.build_parser() + + assert ( + mcp_swap.cmd_use_local( + parser.parse_args( + ["use-local", "--repo", str(fake_repo), "--cli", "cursor"] + ) + ) + == 0 + ) + state = mcp_swap.load_state()["cursor", "user"] + backup = pathlib.Path(state.backup_path) + assert info.config_path.is_symlink() + assert backup.parent == info.config_path.parent + assert json.loads(target.read_text())["mcpServers"]["libtmux"]["command"] == "uv" + + assert mcp_swap.cmd_revert(parser.parse_args(["revert", "--cli", "cursor"])) == 0 + assert info.config_path.is_symlink() + assert target.read_bytes() == original + assert not backup.exists() + + +@pytest.mark.parametrize("replacement_kind", ["symlink", "file"]) +def test_revert_uses_the_original_target_when_a_config_link_is_replaced( + fake_home: pathlib.Path, + fake_repo: pathlib.Path, + replacement_kind: str, +) -> None: + """Repointing or replacing a link cannot redirect recovery into a new file.""" + info = mcp_swap.CLIS["cursor"] + original_target = fake_home / "dotfiles" / "original.json" + new_target = fake_home / "dotfiles" / "replacement.json" + _write_json(original_target, {"mcpServers": {"libtmux": _pinned_json_entry()}}) + _write_json(new_target, {"sentinel": "leave me alone"}) + original = original_target.read_bytes() + replacement = new_target.read_bytes() + info.config_path.parent.mkdir(parents=True) + info.config_path.symlink_to(original_target) + parser = mcp_swap.build_parser() + + assert ( + mcp_swap.cmd_use_local( + parser.parse_args( + ["use-local", "--repo", str(fake_repo), "--cli", "cursor"] + ) + ) + == 0 + ) + info.config_path.unlink() + if replacement_kind == "symlink": + info.config_path.symlink_to(new_target) + replacement_path = new_target + else: + info.config_path.write_bytes(replacement) + replacement_path = info.config_path + + assert mcp_swap.cmd_revert(parser.parse_args(["revert", "--cli", "cursor"])) == 0 + assert original_target.read_bytes() == original + assert replacement_path.read_bytes() == replacement