Coverage for astrocyte/harness/server.py: 91%
128 statements
« prev ^ index » next coverage.py v7.15.0, created at 2026-10-29 22:10 +0000
1"""Locating and probing the ``astrocyte-mcp`` server setup registers."""
3from __future__ import annotations
5import json
6import os
7import queue
8import re
9import shutil
10import subprocess
11import sys
12import threading
13import time
14from dataclasses import dataclass
15from pathlib import Path
17from .hosts import ServerSpec
20@dataclass(frozen=True)
21class ServerLocation:
22 command: str
23 ephemeral: bool # inside a cache uv/pipx may garbage-collect
26def locate_script(name: str) -> ServerLocation | None:
27 """A console script from *this* installation, as an absolute path.
29 The one beside the running interpreter wins over PATH: setup must register
30 commands from the same environment it is running in, not whichever
31 install happens to come first on PATH.
32 """
33 exe = f"{name}.exe" if os.name == "nt" else name
34 beside = Path(sys.executable).parent / exe
35 found = str(beside) if beside.exists() else shutil.which(name)
36 if not found:
37 return None
38 resolved = str(Path(found).absolute())
39 # `uvx astrocyte setup` runs from uv's archive cache, which uv prunes; a
40 # path registered from there breaks later with no obvious cause.
41 markers = (f"{os.sep}uv{os.sep}archive-", f"{os.sep}.cache{os.sep}uv{os.sep}", f"{os.sep}pipx{os.sep}.cache")
42 return ServerLocation(command=resolved, ephemeral=any(m in resolved for m in markers))
45def locate_mcp_server() -> ServerLocation | None:
46 """The interpreter of *this* installation, which runs the server.
48 Registrations name the interpreter rather than the ``astrocyte-mcp``
49 console script so they can pass ``-I`` (isolated mode): agents launch
50 servers and hooks with their own environment, and an inherited
51 ``PYTHONPATH`` made the installed scripts import a different checkout's
52 code (observed on first real install). ``-I`` also keeps the cwd off
53 ``sys.path``, so working inside an Astrocyte source tree can't shadow it.
54 The interpreter path is kept as-is, not resolved: in a venv it is a
55 symlink, and resolving it would escape the venv's site-packages.
56 """
57 if locate_script("astrocyte-mcp") is None:
58 return None # not installed in this environment
59 exe = str(Path(sys.executable).absolute())
60 markers = (f"{os.sep}uv{os.sep}archive-", f"{os.sep}.cache{os.sep}uv{os.sep}", f"{os.sep}pipx{os.sep}.cache")
61 return ServerLocation(command=exe, ephemeral=any(m in exe for m in markers))
64ISOLATED = "-I"
67def server_spec(python: str, config: Path) -> ServerSpec:
68 return ServerSpec(command=python, args=(ISOLATED, "-m", "astrocyte.mcp", "--config", str(config)))
71class HookPathError(ValueError):
72 """The interpreter path can't be written as a shell-neutral hook command."""
75# Characters every shell an agent may use on Windows (cmd, PowerShell, Git
76# Bash) reads literally in an unquoted word.
77_SHELL_NEUTRAL = re.compile(r"^[A-Za-z0-9_.~:/-]+$")
80def hook_prefix(python: str, *, windows: bool = os.name == "nt") -> str:
81 """Shell command prefix for ``astrocyte hook <event>`` registrations.
83 POSIX: shell-quoted. Windows: an agent runs hooks through cmd, PowerShell
84 or Git Bash depending on the machine (Claude Code: Git Bash when present,
85 else PowerShell), and no quoting means the same in all three. So the path
86 is written unquoted with forward slashes (bash keeps them; CreateProcess
87 accepts them) and without spaces, using its 8.3 short form if it has any.
88 """
89 if not windows:
90 import shlex
92 return f"{shlex.quote(python)} {ISOLATED} -m astrocyte.cli"
93 path = _short_path(python) if " " in python else python
94 path = path.replace("\\", "/")
95 if not _SHELL_NEUTRAL.match(path):
96 raise HookPathError(
97 f"{python} can't be written so that cmd, PowerShell and Git Bash all run it (it has spaces or shell "
98 "characters, and no 8.3 short name). Reinstall under a plain path, e.g. "
99 "uv tool install 'astrocyte[local]' with UV_TOOL_DIR=C:/astrocyte-tools."
100 )
101 return f"{path} {ISOLATED} -m astrocyte.cli"
104def _short_path(path: str) -> str:
105 """Windows' 8.3 alias of ``path`` (no spaces), or ``path`` where there is none."""
106 try:
107 import ctypes
108 from ctypes import wintypes
110 get = ctypes.windll.kernel32.GetShortPathNameW # type: ignore[attr-defined]
111 get.argtypes = [wintypes.LPCWSTR, wintypes.LPWSTR, wintypes.DWORD]
112 get.restype = wintypes.DWORD
113 size = get(path, None, 0)
114 if not size:
115 return path
116 buf = ctypes.create_unicode_buffer(size)
117 return buf.value if get(path, buf, size) else path
118 except (AttributeError, OSError): # not Windows, or no kernel32
119 return path
122@dataclass(frozen=True)
123class HandshakeResult:
124 ok: bool
125 detail: str
126 tools: tuple[str, ...] = ()
127 seconds: float = 0.0
130_EXCEPTION_LINE = re.compile(r"^\s*[A-Za-z_][\w.]*(?:Error|Exception|Exit|Interrupt)\b.*:")
133def _explain(stderr: str | None) -> str:
134 """The line of a crash's stderr that says what went wrong.
136 A traceback's last lines are often frames or source, and a chained
137 exception repeats; the final ``SomethingError: message`` line is the one
138 a user can act on. Without one, the last few lines as they are.
139 """
140 lines = [ln.rstrip() for ln in (stderr or "").splitlines() if ln.strip()]
141 errors = [ln.strip() for ln in lines if _EXCEPTION_LINE.match(ln)]
142 if errors:
143 return errors[-1]
144 return " / ".join(ln.strip() for ln in lines[-3:])
147def handshake(spec: ServerSpec, *, timeout: float = 90.0) -> HandshakeResult:
148 """Launch the server exactly as a harness would and complete an MCP
149 ``initialize`` + ``tools/list``. Catches what static checks cannot: a
150 missing dependency, a config the server rejects, a broken interpreter.
152 Each reply is awaited before the next request is sent and stdin stays open
153 until the end: a stdio server may exit on EOF before answering requests
154 that are still queued.
155 """
156 started = time.perf_counter()
157 deadline = started + timeout
158 try:
159 proc = subprocess.Popen(
160 [spec.command, *spec.args],
161 stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True,
162 )
163 except OSError as e:
164 return HandshakeResult(False, f"could not start {spec.command}: {e}")
166 lines: queue.Queue[str | None] = queue.Queue()
168 def pump() -> None:
169 assert proc.stdout is not None
170 for line in proc.stdout:
171 lines.put(line)
172 lines.put(None) # EOF
174 threading.Thread(target=pump, daemon=True).start()
176 def send(msg: dict) -> None:
177 assert proc.stdin is not None
178 proc.stdin.write(json.dumps(msg) + "\n")
179 proc.stdin.flush()
181 def await_reply(want: int) -> dict | None:
182 while (remaining := deadline - time.perf_counter()) > 0:
183 try:
184 line = lines.get(timeout=remaining)
185 except queue.Empty:
186 return None
187 if line is None:
188 return None
189 try:
190 msg = json.loads(line)
191 except json.JSONDecodeError:
192 continue
193 if isinstance(msg, dict) and msg.get("id") == want:
194 return msg
195 return None
197 def failure(default: str) -> HandshakeResult:
198 proc.kill()
199 _, stderr = proc.communicate(timeout=10)
200 return HandshakeResult(False, _explain(stderr) or default, seconds=time.perf_counter() - started)
202 try:
203 send({"jsonrpc": "2.0", "id": 1, "method": "initialize", "params": {
204 "protocolVersion": "2025-06-18", "capabilities": {},
205 "clientInfo": {"name": "astrocyte-doctor", "version": "1"}}})
206 if await_reply(1) is None:
207 return failure(f"no initialize reply within {timeout:.0f}s")
208 send({"jsonrpc": "2.0", "method": "notifications/initialized"})
209 send({"jsonrpc": "2.0", "id": 2, "method": "tools/list"})
210 listed = await_reply(2)
211 if listed is None or "result" not in listed:
212 return failure("server did not answer tools/list")
213 except (BrokenPipeError, OSError):
214 return failure("server exited during the handshake")
216 proc.kill()
217 proc.communicate(timeout=10)
218 tools = tuple(t.get("name", "") for t in listed["result"].get("tools", []))
219 elapsed = time.perf_counter() - started
220 if "memory_retain" not in tools:
221 return HandshakeResult(False, "server answered but exposes no memory_retain tool", tools, elapsed)
222 return HandshakeResult(True, f"{len(tools)} tools", tools, elapsed)