fix: reload MCP launchers when runtime changes
This commit is contained in:
@@ -46,6 +46,12 @@ directory, add a trusted project `.codex/config.toml` override with both
|
|||||||
`cwd` and `NAPSEER_PROJECT_ROOT`. MCP processes inherit their launch directory;
|
`cwd` and `NAPSEER_PROJECT_ROOT`. MCP processes inherit their launch directory;
|
||||||
they cannot infer a later client workspace switch over stdio.
|
they cannot infer a later client workspace switch over stdio.
|
||||||
|
|
||||||
|
If `NAPSEER_MCP_WORKER_PATH` points to a stable launcher that loads a separate,
|
||||||
|
replaceable runtime, set `NAPSEER_MCP_WORKER_WATCH_PATHS` to the
|
||||||
|
`os.pathsep`-separated runtime paths. The supervisor then reloads the launcher
|
||||||
|
after any watched runtime is atomically replaced, preserving the same bounded,
|
||||||
|
non-replaying request lifecycle used for direct workers.
|
||||||
|
|
||||||
The public operator surface is intentionally small:
|
The public operator surface is intentionally small:
|
||||||
|
|
||||||
```text
|
```text
|
||||||
|
|||||||
@@ -12,7 +12,8 @@ session. This supervisor owns those client-facing pipes while treating
|
|||||||
notification to the worker;
|
notification to the worker;
|
||||||
- a worker exit fails every affected in-flight request once and never replays
|
- a worker exit fails every affected in-flight request once and never replays
|
||||||
it automatically;
|
it automatically;
|
||||||
- a replaced worker is reloaded at the next quiescent request boundary.
|
- a replaced worker, or an explicitly configured runtime dependency behind a
|
||||||
|
stable launcher, is reloaded at the next quiescent request boundary.
|
||||||
|
|
||||||
The supervisor never logs MCP payloads, tool results, auth state, or child
|
The supervisor never logs MCP payloads, tool results, auth state, or child
|
||||||
stderr. Protocol stdout contains JSON-RPC messages only.
|
stderr. Protocol stdout contains JSON-RPC messages only.
|
||||||
@@ -39,6 +40,11 @@ WORKER_PATH = pathlib.Path(
|
|||||||
pathlib.Path(__file__).resolve().with_name("napseer_mcp_server.py"),
|
pathlib.Path(__file__).resolve().with_name("napseer_mcp_server.py"),
|
||||||
)
|
)
|
||||||
).expanduser()
|
).expanduser()
|
||||||
|
WORKER_WATCH_PATHS = tuple(
|
||||||
|
pathlib.Path(value).expanduser()
|
||||||
|
for value in os.environ.get("NAPSEER_MCP_WORKER_WATCH_PATHS", "").split(os.pathsep)
|
||||||
|
if value.strip()
|
||||||
|
)
|
||||||
RESPONSE_TIMEOUT_SECONDS = max(
|
RESPONSE_TIMEOUT_SECONDS = max(
|
||||||
1,
|
1,
|
||||||
int(os.environ.get("NAPSEER_MCP_RESPONSE_TIMEOUT_SECONDS", "60")),
|
int(os.environ.get("NAPSEER_MCP_RESPONSE_TIMEOUT_SECONDS", "60")),
|
||||||
@@ -60,6 +66,15 @@ def source_identity(path: pathlib.Path) -> tuple[int, int, int]:
|
|||||||
return stat.st_ino, stat.st_size, stat.st_mtime_ns
|
return stat.st_ino, stat.st_size, stat.st_mtime_ns
|
||||||
|
|
||||||
|
|
||||||
|
def runtime_identity(
|
||||||
|
worker_path: pathlib.Path,
|
||||||
|
watch_paths: tuple[pathlib.Path, ...] = (),
|
||||||
|
) -> tuple[tuple[int, int, int], ...]:
|
||||||
|
"""Identify every source whose replacement changes the worker runtime."""
|
||||||
|
|
||||||
|
return tuple(source_identity(path) for path in (worker_path, *watch_paths))
|
||||||
|
|
||||||
|
|
||||||
def parse_message(message: bytes):
|
def parse_message(message: bytes):
|
||||||
try:
|
try:
|
||||||
payload = json.loads(message)
|
payload = json.loads(message)
|
||||||
@@ -121,10 +136,16 @@ class PendingRequest:
|
|||||||
|
|
||||||
|
|
||||||
class Worker:
|
class Worker:
|
||||||
def __init__(self, path: pathlib.Path, unsolicited: Callable[[bytes], None] | None = None):
|
def __init__(
|
||||||
|
self,
|
||||||
|
path: pathlib.Path,
|
||||||
|
unsolicited: Callable[[bytes], None] | None = None,
|
||||||
|
watch_paths: tuple[pathlib.Path, ...] = (),
|
||||||
|
):
|
||||||
self.path = path
|
self.path = path
|
||||||
|
self.watch_paths = tuple(watch_paths)
|
||||||
self.process: subprocess.Popen[bytes] | None = None
|
self.process: subprocess.Popen[bytes] | None = None
|
||||||
self.identity: tuple[int, int, int] | None = None
|
self.identity: tuple[tuple[int, int, int], ...] | None = None
|
||||||
self.generation = 0
|
self.generation = 0
|
||||||
self.unsolicited = unsolicited
|
self.unsolicited = unsolicited
|
||||||
self._lifecycle_lock = threading.RLock()
|
self._lifecycle_lock = threading.RLock()
|
||||||
@@ -237,7 +258,7 @@ class Worker:
|
|||||||
with self._lifecycle_lock:
|
with self._lifecycle_lock:
|
||||||
self._stop_locked()
|
self._stop_locked()
|
||||||
try:
|
try:
|
||||||
identity = source_identity(self.path)
|
identity = runtime_identity(self.path, self.watch_paths)
|
||||||
process = subprocess.Popen(
|
process = subprocess.Popen(
|
||||||
[sys.executable, "-u", str(self.path)],
|
[sys.executable, "-u", str(self.path)],
|
||||||
stdin=subprocess.PIPE,
|
stdin=subprocess.PIPE,
|
||||||
@@ -275,7 +296,7 @@ class Worker:
|
|||||||
def ensure_current(self) -> None:
|
def ensure_current(self) -> None:
|
||||||
with self._lifecycle_lock:
|
with self._lifecycle_lock:
|
||||||
try:
|
try:
|
||||||
current_identity = source_identity(self.path)
|
current_identity = runtime_identity(self.path, self.watch_paths)
|
||||||
except OSError as exc:
|
except OSError as exc:
|
||||||
self._stop_locked()
|
self._stop_locked()
|
||||||
raise WorkerUnavailable("Napseer MCP worker is missing") from exc
|
raise WorkerUnavailable("Napseer MCP worker is missing") from exc
|
||||||
@@ -361,7 +382,11 @@ def supervise() -> int:
|
|||||||
sys.stdout.buffer.write(response if response.endswith(b"\n") else response + b"\n")
|
sys.stdout.buffer.write(response if response.endswith(b"\n") else response + b"\n")
|
||||||
sys.stdout.buffer.flush()
|
sys.stdout.buffer.flush()
|
||||||
|
|
||||||
worker = Worker(WORKER_PATH, unsolicited=emit)
|
worker = Worker(
|
||||||
|
WORKER_PATH,
|
||||||
|
unsolicited=emit,
|
||||||
|
watch_paths=WORKER_WATCH_PATHS,
|
||||||
|
)
|
||||||
waiters = concurrent.futures.ThreadPoolExecutor(
|
waiters = concurrent.futures.ThreadPoolExecutor(
|
||||||
max_workers=MAX_IN_FLIGHT_REQUESTS,
|
max_workers=MAX_IN_FLIGHT_REQUESTS,
|
||||||
thread_name_prefix="napseer-mcp-request",
|
thread_name_prefix="napseer-mcp-request",
|
||||||
|
|||||||
@@ -70,9 +70,18 @@ for line in sys.stdin:
|
|||||||
|
|
||||||
|
|
||||||
class SupervisorClient:
|
class SupervisorClient:
|
||||||
def __init__(self, worker_path: pathlib.Path, project_root: pathlib.Path | None = None):
|
def __init__(
|
||||||
|
self,
|
||||||
|
worker_path: pathlib.Path,
|
||||||
|
project_root: pathlib.Path | None = None,
|
||||||
|
watch_paths: tuple[pathlib.Path, ...] = (),
|
||||||
|
):
|
||||||
environment = os.environ.copy()
|
environment = os.environ.copy()
|
||||||
environment["NAPSEER_MCP_WORKER_PATH"] = str(worker_path)
|
environment["NAPSEER_MCP_WORKER_PATH"] = str(worker_path)
|
||||||
|
if watch_paths:
|
||||||
|
environment["NAPSEER_MCP_WORKER_WATCH_PATHS"] = os.pathsep.join(
|
||||||
|
str(path) for path in watch_paths
|
||||||
|
)
|
||||||
environment["NAPSEER_MCP_RESPONSE_TIMEOUT_SECONDS"] = "5"
|
environment["NAPSEER_MCP_RESPONSE_TIMEOUT_SECONDS"] = "5"
|
||||||
environment["NAPSEER_TELEMETRY"] = "0"
|
environment["NAPSEER_TELEMETRY"] = "0"
|
||||||
if project_root is not None:
|
if project_root is not None:
|
||||||
@@ -272,6 +281,33 @@ def test_reloads_replaced_worker_between_requests(tmp_path):
|
|||||||
client.close()
|
client.close()
|
||||||
|
|
||||||
|
|
||||||
|
def test_reloads_stable_launcher_when_watched_runtime_is_replaced(tmp_path):
|
||||||
|
runtime_path = tmp_path / "runtime.py"
|
||||||
|
launcher_path = tmp_path / "launcher.py"
|
||||||
|
write_worker(runtime_path, 1)
|
||||||
|
launcher_path.write_text(
|
||||||
|
"import runpy\n"
|
||||||
|
f"runpy.run_path({str(runtime_path)!r}, run_name='__main__')\n",
|
||||||
|
encoding="utf-8",
|
||||||
|
)
|
||||||
|
client = SupervisorClient(launcher_path, watch_paths=(runtime_path,))
|
||||||
|
try:
|
||||||
|
first = client.request(1, "initialize")
|
||||||
|
launcher_identity = launcher_path.stat().st_mtime_ns
|
||||||
|
|
||||||
|
time.sleep(0.01)
|
||||||
|
write_worker(runtime_path, 2)
|
||||||
|
os.utime(runtime_path, None)
|
||||||
|
|
||||||
|
second = client.request(2, "tools/list")
|
||||||
|
|
||||||
|
assert first["result"]["version"] == 1
|
||||||
|
assert second["result"]["version"] == 2
|
||||||
|
assert launcher_path.stat().st_mtime_ns == launcher_identity
|
||||||
|
finally:
|
||||||
|
client.close()
|
||||||
|
|
||||||
|
|
||||||
def test_worker_crash_returns_error_without_closing_transport(tmp_path):
|
def test_worker_crash_returns_error_without_closing_transport(tmp_path):
|
||||||
worker_path = tmp_path / "worker.py"
|
worker_path = tmp_path / "worker.py"
|
||||||
write_worker(worker_path, 1)
|
write_worker(worker_path, 1)
|
||||||
|
|||||||
Reference in New Issue
Block a user