From 1e1c124c9f71fa868e6a08dd10053b2942209d05 Mon Sep 17 00:00:00 2001 From: user Date: Fri, 17 Jul 2026 08:24:37 +0200 Subject: [PATCH] feat: switch dynamic MCP container targets --- rvctl.py | 216 +++++++++++++++++++++++++++++++++++++++++- tests/test_stemctl.py | 45 +++++++++ 2 files changed, 259 insertions(+), 2 deletions(-) diff --git a/rvctl.py b/rvctl.py index c5f6d98..d2dd114 100755 --- a/rvctl.py +++ b/rvctl.py @@ -11,6 +11,7 @@ import os import re import shlex import shutil +import stat import subprocess import sys from dataclasses import dataclass, replace @@ -3833,7 +3834,17 @@ def run_probe(config: WorkspaceConfig, args: argparse.Namespace) -> None: def run_mcp(config: WorkspaceConfig, args: argparse.Namespace) -> None: - """Replace stemctl with the MCP server bound to one verified instance.""" + """Manage the dynamic MCP target or serve one explicitly bound instance.""" + if args.mcp_kind == "list": + print_mcp_instances(config) + return + if args.mcp_kind == "select": + select_mcp_instance(config, args.selector) + return + if args.mcp_kind == "status": + print_mcp_selection(config) + return + env_script = ensure_env_tool(config) wrapper = env_script.parent / "scripts" / f"mcp-{args.mcp_kind}.sh" if not wrapper.is_file(): @@ -3848,6 +3859,201 @@ def run_mcp(config: WorkspaceConfig, args: argparse.Namespace) -> None: os.execvpe(str(wrapper), [str(wrapper), args.instance], env) +def mcp_selection_root(config: WorkspaceConfig) -> Path: + """Return the short, stable host path used by generic MCP providers.""" + return config.socket_root / "mcp-selected" + + +def mcp_registry_records(config: WorkspaceConfig) -> list[dict]: + registry_root = config.socket_root / "registry" + records: list[dict] = [] + for path in sorted(registry_root.glob("*.json")): + try: + record = json.loads(path.read_text(encoding="utf-8")) + except (OSError, json.JSONDecodeError): + continue + if isinstance(record, dict): + record["_registry_path"] = str(path) + records.append(record) + return records + + +def is_unix_socket(path: Path) -> bool: + try: + return stat.S_ISSOCK(path.stat().st_mode) + except OSError: + return False + + +def probe_mcp_socket_servers(record: dict, nvim_socket: Path, tmux_socket: Path) -> None: + runtime = str(record["runtime"]) + container_name = str(record["container_name"]) + probes = ( + ( + "Neovim", + [runtime, "exec", container_name, "nvim", "--server", str(nvim_socket), "--remote-expr", "1"], + ), + ( + "tmux", + [runtime, "exec", container_name, "tmux", "-S", str(tmux_socket), "list-sessions"], + ), + ) + failures: list[str] = [] + for name, command in probes: + try: + result = subprocess.run( + command, + check=False, + capture_output=True, + text=True, + timeout=3, + ) + except subprocess.TimeoutExpired: + failures.append(f"{name}: connection timed out") + continue + if result.returncode != 0: + detail = (result.stderr or result.stdout).strip().splitlines() + failures.append(f"{name}: {detail[-1] if detail else 'connection failed'}") + if failures: + raise SystemExit("MCP socket files exist, but their servers are not live:\n" + "\n".join(failures)) + + +def validated_mcp_record(record: dict) -> tuple[dict, Path, Path]: + required = ("runtime", "container_name", "container_id", "instance_key", "socket_dir") + missing = [key for key in required if not str(record.get(key, "")).strip()] + if missing: + raise SystemExit(f"Incomplete MCP registry ({', '.join(missing)}): {record.get('_registry_path', '?')}") + + runtime = str(record["runtime"]) + container_name = str(record["container_name"]) + expected_id = str(record["container_id"]) + expected_instance_key = str(record["instance_key"]) + if shutil.which(runtime) is None: + raise SystemExit(f"Recorded container runtime is unavailable: {runtime}") + + inspected = subprocess.run( + [runtime, "inspect", container_name], + check=False, + capture_output=True, + text=True, + ) + if inspected.returncode != 0: + raise SystemExit(f"Container is not available: {container_name}") + try: + inspection = json.loads(inspected.stdout)[0] + except (IndexError, KeyError, TypeError, json.JSONDecodeError) as error: + raise SystemExit(f"Cannot inspect container identity: {container_name}") from error + actual_id = str(inspection.get("Id", "")) + labels = inspection.get("Config", {}).get("Labels") or {} + actual_instance_key = str(labels.get("edu.stem.instance-key", "")) + if actual_id != expected_id or actual_instance_key != expected_instance_key: + raise SystemExit(f"Stale MCP registry rejected for {container_name} (container identity changed).") + + socket_dir = Path(str(record["socket_dir"])) + nvim_socket = socket_dir / "n.sock" + tmux_socket = socket_dir / "t.sock" + missing_sockets = [str(path) for path in (nvim_socket, tmux_socket) if not is_unix_socket(path)] + if missing_sockets: + raise SystemExit("MCP sockets are not live:\n" + "\n".join(missing_sockets)) + probe_mcp_socket_servers(record, nvim_socket, tmux_socket) + return record, nvim_socket, tmux_socket + + +def resolve_mcp_record(config: WorkspaceConfig, selector: str) -> dict: + selector = selector.strip() + if not selector: + raise SystemExit("MCP selector cannot be empty.") + matches: list[dict] = [] + for record in mcp_registry_records(config): + container_id = str(record.get("container_id", "")) + if selector in {str(record.get("instance", "")), str(record.get("container_name", ""))}: + matches.append(record) + elif len(selector) >= 12 and container_id.startswith(selector): + matches.append(record) + unique = {str(record.get("container_id", "")): record for record in matches} + if not unique: + raise SystemExit( + f"No registered MCP container matches {selector!r}. " + "Use 'stemctl mcp list' to see available instances." + ) + if len(unique) > 1: + names = ", ".join(sorted(str(record.get("instance", "?")) for record in unique.values())) + raise SystemExit(f"Ambiguous MCP selector {selector!r}: {names}") + return next(iter(unique.values())) + + +def select_mcp_instance(config: WorkspaceConfig, selector: str) -> None: + record, nvim_socket, tmux_socket = validated_mcp_record(resolve_mcp_record(config, selector)) + container_id = str(record["container_id"]) + cid12 = container_id[:12] + selection_root = mcp_selection_root(config) + target_root = selection_root / "targets" / cid12 + target_root.mkdir(parents=True, exist_ok=True) + os.chmod(selection_root, 0o700) + os.chmod(selection_root / "targets", 0o700) + os.chmod(target_root, 0o700) + + for link, target in ((target_root / "n.sock", nvim_socket), (target_root / "t.sock", tmux_socket)): + temporary = link.with_name(f".{link.name}.{os.getpid()}") + temporary.unlink(missing_ok=True) + temporary.symlink_to(target) + os.replace(temporary, link) + + metadata = { + key: record.get(key, "") + for key in ("runtime", "container_name", "container_id", "profile", "target", "instance", "instance_key", "thread_key", "socket_dir") + } + metadata_path = target_root / "selection.json" + temporary_metadata = metadata_path.with_name(f".{metadata_path.name}.{os.getpid()}") + temporary_metadata.write_text(json.dumps(metadata, indent=2) + "\n", encoding="utf-8") + os.replace(temporary_metadata, metadata_path) + + current = selection_root / "current" + temporary_current = selection_root / f".current.{os.getpid()}" + temporary_current.unlink(missing_ok=True) + temporary_current.symlink_to(Path("targets") / cid12) + os.replace(temporary_current, current) + + print(f"instance\t{record.get('instance', '')}") + print(f"container\t{record.get('container_name', '')}") + print(f"container_id\t{container_id}") + print(f"nvim_socket\t{current / 'n.sock'}") + print(f"tmux_socket\t{current / 't.sock'}") + + +def print_mcp_selection(config: WorkspaceConfig) -> None: + current = mcp_selection_root(config) / "current" + metadata_path = current / "selection.json" + try: + record = json.loads(metadata_path.read_text(encoding="utf-8")) + except (OSError, json.JSONDecodeError) as error: + raise SystemExit("No dynamic MCP container is selected. Use 'stemctl mcp select INSTANCE'.") from error + record, _, _ = validated_mcp_record(record) + print(f"instance\t{record.get('instance', '')}") + print(f"container\t{record.get('container_name', '')}") + print(f"container_id\t{record.get('container_id', '')}") + print(f"nvim_socket\t{current / 'n.sock'}") + print(f"tmux_socket\t{current / 't.sock'}") + + +def print_mcp_instances(config: WorkspaceConfig) -> None: + print("container_id\tinstance\tprofile\tcontainer\tsockets") + for record in sorted(mcp_registry_records(config), key=lambda item: str(item.get("instance", ""))): + socket_dir = Path(str(record.get("socket_dir", ""))) + live = False + if is_unix_socket(socket_dir / "n.sock") and is_unix_socket(socket_dir / "t.sock"): + try: + validated_mcp_record(record) + live = True + except SystemExit: + pass + print( + f"{str(record.get('container_id', ''))[:12]}\t{record.get('instance', '')}\t" + f"{record.get('profile', '')}\t{record.get('container_name', '')}\t" + f"{'ready' if live else 'not-ready'}" + ) + + def run_profile_action(config: WorkspaceConfig, args: argparse.Namespace, action: str) -> None: profile = normalize_env_profile(args.profile) selector_values, pane_target = split_pane_selector(args.selector, getattr(args, "pane", None)) @@ -4301,11 +4507,17 @@ def build_parser() -> argparse.ArgumentParser: probe_list_parser = probe_subparsers.add_parser("list", help="List usable probe/target devices without sudo.") probe_list_parser.add_argument("--json", action="store_true", help="Emit machine-readable JSON.") - mcp_parser = subparsers.add_parser("mcp", help="Run the container-bundled MCP server for one logical instance.") + mcp_parser = subparsers.add_parser("mcp", help="Select an MCP container or run its bundled MCP server.") mcp_subparsers = mcp_parser.add_subparsers(dest="mcp_kind", required=True) for kind in ("tmux", "nvim"): kind_parser = mcp_subparsers.add_parser(kind, help=f"Serve {kind} tools over MCP stdio.") kind_parser.add_argument("instance", help="Stable logical instance name (required to avoid cross-container selection).") + mcp_select_parser = mcp_subparsers.add_parser( + "select", help="Atomically point dynamic tmux and Neovim MCP sockets at one container." + ) + mcp_select_parser.add_argument("selector", help="Logical instance, container name, or at least 12 ID characters.") + mcp_subparsers.add_parser("status", help="Show and validate the dynamically selected MCP container.") + mcp_subparsers.add_parser("list", help="List registered container instances and socket readiness.") tasks_parser = subparsers.add_parser("tasks", help="Shortcut for tasks in the default card.") tasks_subparsers = tasks_parser.add_subparsers(dest="tasks_command", required=True) diff --git a/tests/test_stemctl.py b/tests/test_stemctl.py index a15c62c..c807ced 100644 --- a/tests/test_stemctl.py +++ b/tests/test_stemctl.py @@ -3,6 +3,7 @@ from __future__ import annotations import json import io import os +import socket import subprocess import tempfile import unittest @@ -289,6 +290,50 @@ class StemctlContractTests(unittest.TestCase): self.assertEqual(args.mcp_kind, "nvim") self.assertEqual(args.instance, "hazard3-inf-bss-task1") + def test_mcp_select_accepts_instance_or_container_id(self) -> None: + parser = rvctl.build_parser() + by_instance = parser.parse_args(["mcp", "select", "rp2350-pointers-final"]) + by_container = parser.parse_args(["mcp", "select", "3aca2c1c4c7a"]) + self.assertEqual(by_instance.mcp_kind, "select") + self.assertEqual(by_instance.selector, "rp2350-pointers-final") + self.assertEqual(by_container.selector, "3aca2c1c4c7a") + + def test_mcp_select_switches_both_sockets_with_one_current_symlink(self) -> None: + with tempfile.TemporaryDirectory() as temporary: + root = Path(temporary) + socket_dir = root / "source" + socket_dir.mkdir() + nvim_server = socket.socket(socket.AF_UNIX) + tmux_server = socket.socket(socket.AF_UNIX) + try: + nvim_server.bind(str(socket_dir / "n.sock")) + tmux_server.bind(str(socket_dir / "t.sock")) + config = SimpleNamespace(socket_root=root) + record = { + "runtime": "podman", + "container_name": "stem-rp2350-example", + "container_id": "3aca2c1c4c7a" + "0" * 52, + "profile": "rp2350", + "target": "rp2350-rv", + "instance": "rp2350-example", + "instance_key": "example000001", + "thread_key": "thread000001", + "socket_dir": str(socket_dir), + } + with mock.patch.object(rvctl, "resolve_mcp_record", return_value=record), mock.patch.object( + rvctl, + "validated_mcp_record", + return_value=(record, socket_dir / "n.sock", socket_dir / "t.sock"), + ), redirect_stdout(io.StringIO()): + rvctl.select_mcp_instance(config, "rp2350-example") + current = root / "mcp-selected" / "current" + self.assertEqual(os.readlink(current), "targets/3aca2c1c4c7a") + self.assertEqual((current / "n.sock").resolve(), socket_dir / "n.sock") + self.assertEqual((current / "t.sock").resolve(), socket_dir / "t.sock") + finally: + nvim_server.close() + tmux_server.close() + if __name__ == "__main__": unittest.main()