tancouandClaude Fable 5.1
3b7b5cbe7c
Review gate on #7720 (two regressions reproduced against master with the installed Agent): 1. Deleting a profile's last MCP server left ADOPTED tools callable after /reload-mcp. The scoped teardown only stops connections the profile owns, and the Agent's discover_mcp_tools() returns before its reconcile step when the profile's config has no server, so an adopted shared connection was never judged against the config. /reload-mcp now calls register_mcp_servers({}) under the profile's scope before rediscovery: the Agent's overlay heal detaches only this profile's overlay for servers its config no longer has; the owner's connection keeps running (master's wildcard shutdown stopped the owner too). 2. Portable plugin MCP servers are merged by the Agent at runtime and never appear in the profile's raw ``mcp_servers``; the registry fallback required a config entry and dropped every plugin tool from /api/mcp/tools and the Notes inventory. The profile's own registry slot (registry_tool_owned_by_view) is the isolation check; the config is not an allowlist any more. Tests (red before, green after): agent-free unit tests for both (the fake agent now models the overlay heal and discovery's empty-config early return), and a real-Agent integration test for the adopted-then-deleted case (owner connection survives, adopter slot emptied, "Removed" reported). Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
650 lines
31 KiB
Python
650 lines
31 KiB
Python
"""Agent-free contract tests for profile-scoped MCP status, inventory and /reload-mcp.
|
|
|
|
``tests/test_mcp_runtime_profile_scope.py`` drives the real Hermes Agent MCP
|
|
runtime; CI has no Hermes Agent checkout, so this file models the small agent
|
|
surface WebUI relies on (context-local home override, routed-profile predicate,
|
|
scoped registry slots, scoped MCP ledger with adoption and connect cooldowns)
|
|
and pins the WebUI side of the contract: which profile is bound, which scope is
|
|
reset, what the root profile may see, and the fail-closed paths.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import sys
|
|
import threading
|
|
import types
|
|
from contextvars import ContextVar
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
from typing import Any, Optional
|
|
from unittest.mock import MagicMock
|
|
|
|
import pytest
|
|
import yaml
|
|
|
|
from api import profiles
|
|
|
|
|
|
class FakeAgent:
|
|
"""Minimal model of the hermes-agent MCP surface used by WebUI.
|
|
|
|
Mirrors the agent's visibility rules: a routed scope sees the connections it
|
|
owns or adopted; the launch profile's view (scope ``None``) is process-wide.
|
|
"""
|
|
|
|
def __init__(self, *, pin_supported=True, scoped=True):
|
|
self.override: ContextVar[Optional[str]] = ContextVar("fake_home_override", default=None)
|
|
self.secret: ContextVar[Optional[dict]] = ContextVar("fake_secret_scope", default=None)
|
|
self.pinned = None
|
|
self.servers = {} # ledger key -> connection
|
|
self.owners = {} # ledger key -> owner scope
|
|
self.adopters = {} # ledger key -> {adopter scope}
|
|
self.connecting = set()
|
|
self.errors = {} # ledger key -> error text
|
|
self.retry_after = {} # ledger key -> deadline (connect cooldown)
|
|
self.failures = {} # ledger key -> failure count
|
|
self.lazy = {} # ledger key -> cfg
|
|
self.global_tools = {} # tool name -> (toolset, schema)
|
|
self.scoped_tools = {} # scope -> {tool name -> (toolset, schema)}
|
|
self.calls = []
|
|
self.lock = threading.Lock()
|
|
self.modules = self._build_modules(pin_supported, scoped)
|
|
|
|
# hermes_constants
|
|
def home(self):
|
|
return self.override.get() or os.environ.get("HERMES_HOME", "")
|
|
|
|
@staticmethod
|
|
def key(path):
|
|
return str(Path(path).resolve())
|
|
|
|
def routed(self):
|
|
override = self.override.get()
|
|
anchor = self.pinned or os.environ.get("HERMES_HOME", "")
|
|
return override is not None and self.key(override) != self.key(anchor)
|
|
|
|
def scope(self):
|
|
return self.key(self.home()) if self.routed() else None
|
|
|
|
def _merged(self, scope):
|
|
return {**self.global_tools, **self.scoped_tools.get(scope, {})}
|
|
|
|
def _slot(self, scope):
|
|
return self.global_tools if scope is None else self.scoped_tools.setdefault(scope, {})
|
|
|
|
@staticmethod
|
|
def ledger_key(name, scope):
|
|
return name if scope is None else (scope, name)
|
|
|
|
def visible(self, key, scope):
|
|
"""The agent's ``_server_visible_in_scope``: process-wide for scope ``None``."""
|
|
if scope is None:
|
|
return True
|
|
return self.owners.get(key) == scope or scope in self.adopters.get(key, ())
|
|
|
|
def connect(self, name, tools, *, scope):
|
|
key = self.ledger_key(name, scope)
|
|
self.servers[key] = SimpleNamespace(name=name, session=object(), _registered_tool_names=list(tools))
|
|
self.owners[key] = scope
|
|
for tool in tools:
|
|
self._slot(scope)[tool] = (f"mcp-{name}", {"description": f"{tool} ({scope})"})
|
|
return self.servers[key]
|
|
|
|
def adopt(self, name, *, owner, adopter):
|
|
"""*adopter* shares *owner*'s identical connection (agent: ``_server_tool_scopes``)."""
|
|
key = self.ledger_key(name, owner)
|
|
self.adopters.setdefault(key, set()).add(adopter)
|
|
for tool in self.servers[key]._registered_tool_names:
|
|
self._slot(adopter)[tool] = (f"mcp-{name}", {"description": f"{tool} ({adopter})"})
|
|
|
|
def park(self, name, *, scope, error="spawn failed"):
|
|
"""A permanent spawn failure the agent parks: task retained in ``_servers``
|
|
without a session, error + backoff recorded under its key."""
|
|
key = self.ledger_key(name, scope)
|
|
self.servers[key] = SimpleNamespace(name=name, session=None, _registered_tool_names=[])
|
|
self.owners[key] = scope
|
|
self.fail(name, scope=scope, error=error)
|
|
return self.servers[key]
|
|
|
|
def fail(self, name, *, scope, error="spawn failed"):
|
|
"""A connect failure: never in ``_servers``; error + backoff recorded under its key."""
|
|
key = self.ledger_key(name, scope)
|
|
self.errors[key] = error
|
|
self.failures[key] = self.failures.get(key, 0) + 1
|
|
self.retry_after[key] = float("inf")
|
|
|
|
def _build_modules(self, pin_supported, scoped):
|
|
agent = self
|
|
hc: Any = types.ModuleType("hermes_constants")
|
|
hc.set_hermes_home_override = lambda path: agent.override.set(None if path is None else str(path))
|
|
hc.reset_hermes_home_override = lambda token: agent.override.reset(token)
|
|
hc.get_hermes_home_override = lambda: agent.override.get()
|
|
hc.hermes_home_key = lambda path=None: agent.key(path if path is not None else agent.home())
|
|
if pin_supported:
|
|
def pin(path):
|
|
agent.pinned = None if path is None else str(path)
|
|
hc.pin_process_hermes_home = pin
|
|
|
|
agent_pkg: Any = types.ModuleType("agent")
|
|
agent_pkg.__path__ = []
|
|
secret: Any = types.ModuleType("agent.secret_scope")
|
|
secret.set_secret_scope = lambda mapping: agent.secret.set(dict(mapping))
|
|
secret.reset_secret_scope = lambda token: agent.secret.reset(token)
|
|
secret.current_secret_scope = lambda: agent.secret.get()
|
|
if scoped:
|
|
secret.serves_routed_profile = agent.routed
|
|
|
|
tools_pkg: Any = types.ModuleType("tools")
|
|
tools_pkg.__path__ = []
|
|
registry_mod: Any = types.ModuleType("tools.registry")
|
|
|
|
class Registry:
|
|
@staticmethod
|
|
def current_scope_key():
|
|
return agent.key(agent.home())
|
|
|
|
def _view(self):
|
|
return agent._merged(agent.scope() if scoped else None)
|
|
|
|
def get_all_tool_names(self):
|
|
return sorted(self._view())
|
|
|
|
def get_toolset_for_tool(self, name):
|
|
return self._view().get(name, (None, None))[0]
|
|
|
|
def get_schema(self, name):
|
|
return self._view().get(name, (None, None))[1]
|
|
|
|
def snapshot_registration(self, name, *, scope=None):
|
|
slot = agent.global_tools if scope is None else agent.scoped_tools.get(scope, {})
|
|
return slot.get(name)
|
|
|
|
registry_mod.registry = Registry()
|
|
|
|
scope_mod: Any = types.ModuleType("tools.mcp_tool_scope")
|
|
scope_mod._key_name = lambda key: key[1] if isinstance(key, tuple) else key
|
|
scope_mod._key_scope = lambda key: key[0] if isinstance(key, tuple) else None
|
|
|
|
mcp_mod: Any = types.ModuleType("tools.mcp_tool")
|
|
mcp_mod._servers = agent.servers
|
|
mcp_mod._lock = agent.lock
|
|
mcp_mod._server_scope_keys = agent.owners
|
|
mcp_mod._server_tool_scopes = agent.adopters
|
|
mcp_mod._server_connecting = agent.connecting
|
|
mcp_mod._server_connect_errors = agent.errors
|
|
mcp_mod._server_connect_retry_after = agent.retry_after
|
|
mcp_mod._server_connect_failures = agent.failures
|
|
mcp_mod._lazy_server_configs = agent.lazy
|
|
|
|
def get_mcp_status(configured=None):
|
|
agent.calls.append(("status", agent.override.get(), dict(configured or {})))
|
|
scope = agent.scope() if scoped else None
|
|
with agent.lock:
|
|
# As the agent: keyed by bare name, last key wins (process-wide for scope None).
|
|
active = {scope_mod._key_name(k): s for k, s in agent.servers.items() if agent.visible(k, scope)}
|
|
errors = {scope_mod._key_name(k): e for k, e in agent.errors.items() if agent.visible(k, scope)}
|
|
rows = []
|
|
for name, cfg in (configured or {}).items():
|
|
enabled = cfg.get("enabled", True) is not False
|
|
server = active.get(name)
|
|
live = server is not None and server.session is not None
|
|
status = ("connected" if live else "disabled" if not enabled
|
|
else "failed" if name in errors else "configured")
|
|
row = {"name": name, "connected": live, "disabled": not enabled, "status": status,
|
|
"tools": len(server._registered_tool_names) if live else 0}
|
|
if status == "failed":
|
|
row["error"] = errors[name]
|
|
rows.append(row)
|
|
return rows
|
|
|
|
def shutdown_mcp_servers(*, scope=None, names=None, timeout=15.0):
|
|
agent.calls.append(("shutdown", scope, None if names is None else set(names)))
|
|
wildcard = scope is None and names is None
|
|
with agent.lock:
|
|
selected = [k for k in agent.servers
|
|
if wildcard or (agent.owners.get(k) == scope
|
|
and (names is None or scope_mod._key_name(k) in names))]
|
|
for key in selected:
|
|
del agent.servers[key]
|
|
agent.owners.pop(key, None)
|
|
agent.adopters.pop(key, None)
|
|
# Cooldown/error state: the wildcard drops everything, a scoped call
|
|
# only what it tore down (the agent's ``selected_status``).
|
|
for key in (list(agent.errors) if wildcard else selected):
|
|
agent.errors.pop(key, None)
|
|
agent.retry_after.pop(key, None)
|
|
agent.failures.pop(key, None)
|
|
|
|
def register_mcp_servers(servers):
|
|
"""The agent's overlay heal for the current scope: a server serving this
|
|
scope through adoption whose config entry is gone (or omitted from
|
|
``servers`` and absent from the profile's config) loses only this
|
|
profile's overlay; the owner's connection is untouched."""
|
|
agent.calls.append(("register", agent.override.get(), dict(servers)))
|
|
scope = agent.scope() if scoped else None
|
|
if scope is None:
|
|
return []
|
|
cfg = yaml.safe_load(Path(agent.home(), "config.yaml").read_text()) or {}
|
|
configured = cfg.get("mcp_servers") or {}
|
|
with agent.lock:
|
|
for key, adopters in list(agent.adopters.items()):
|
|
name = scope_mod._key_name(key)
|
|
if scope in adopters and name not in servers and name not in configured:
|
|
adopters.discard(scope)
|
|
slot = agent.scoped_tools.get(scope, {})
|
|
for tool in [t for t, (ts, _) in slot.items() if ts == f"mcp-{name}"]:
|
|
del slot[tool]
|
|
return sorted(agent._merged(scope))
|
|
|
|
def discover_mcp_tools():
|
|
home = agent.home()
|
|
agent.calls.append(("discover", agent.override.get()))
|
|
cfg = yaml.safe_load(Path(home, "config.yaml").read_text()) or {}
|
|
scope = agent.scope() if scoped else None
|
|
names = []
|
|
if not cfg.get("mcp_servers"):
|
|
return names # as the agent: nothing configured, no reconcile pass
|
|
for name, srv in (cfg.get("mcp_servers") or {}).items():
|
|
key = agent.ledger_key(name, scope)
|
|
served = any(scope_mod._key_name(k) == name and agent.visible(k, scope) and scope is not None
|
|
for k in agent.servers) or key in agent.servers
|
|
if served:
|
|
names += [t for k, s in agent.servers.items()
|
|
if scope_mod._key_name(k) == name for t in s._registered_tool_names]
|
|
continue
|
|
if key in agent.retry_after: # backoff honoured: no retry
|
|
continue
|
|
if srv.get("fake_fail"):
|
|
agent.fail(name, scope=scope, error=str(srv["fake_fail"]))
|
|
continue
|
|
agent.connect(name, srv.get("fake_tools", []), scope=scope)
|
|
names += srv.get("fake_tools", [])
|
|
return names
|
|
|
|
mcp_mod.get_mcp_status = get_mcp_status
|
|
mcp_mod.shutdown_mcp_servers = shutdown_mcp_servers
|
|
mcp_mod.discover_mcp_tools = discover_mcp_tools
|
|
mcp_mod.register_mcp_servers = register_mcp_servers
|
|
return {
|
|
"hermes_constants": hc,
|
|
"agent": agent_pkg,
|
|
"agent.secret_scope": secret,
|
|
"tools": tools_pkg,
|
|
"tools.registry": registry_mod,
|
|
"tools.mcp_tool_scope": scope_mod,
|
|
"tools.mcp_tool": mcp_mod,
|
|
"tools.mcp_tool_discovery": mcp_mod,
|
|
"tools.mcp_tool_lifecycle": mcp_mod,
|
|
}
|
|
|
|
|
|
READ = ["mcp__atlassian__jira_get_issue"]
|
|
WRITE = READ + ["mcp__atlassian__jira_create_issue"]
|
|
|
|
|
|
def _install(monkeypatch, tmp_path, **agent_kwargs):
|
|
agent = FakeAgent(**agent_kwargs)
|
|
for name, module in agent.modules.items():
|
|
monkeypatch.setitem(sys.modules, name, module)
|
|
base = tmp_path / "home"
|
|
for profile, tools in (("default", None), ("profile-read", READ), ("profile-write", WRITE)):
|
|
home = base if profile == "default" else base / "profiles" / profile
|
|
home.mkdir(parents=True, exist_ok=True)
|
|
servers = {} if tools is None else {"atlassian": {"command": "fake", "fake_tools": tools}}
|
|
(home / "config.yaml").write_text(yaml.safe_dump({"mcp_servers": servers}))
|
|
monkeypatch.setenv("HERMES_HOME", str(base))
|
|
monkeypatch.delenv("HERMES_CONFIG_PATH", raising=False)
|
|
monkeypatch.setattr(profiles, "_DEFAULT_HERMES_HOME", base)
|
|
monkeypatch.setattr(profiles, "_is_isolated_profile_mode", lambda: False)
|
|
monkeypatch.setattr(profiles, "_is_root_profile", lambda name: name in ("", "default"))
|
|
monkeypatch.setattr(profiles, "_secret_scope_available", None)
|
|
monkeypatch.setattr(profiles, "_PROCESS_PROFILE_HOME", None)
|
|
profiles._pin_process_profile_home(base)
|
|
agent.base = base
|
|
return agent
|
|
|
|
|
|
def _configure(agent, profile, servers):
|
|
home = agent.base if profile == "default" else agent.base / "profiles" / profile
|
|
(home / "config.yaml").write_text(yaml.safe_dump({"mcp_servers": servers}))
|
|
|
|
|
|
def _as_profile(profile, fn, *args):
|
|
profiles.set_request_profile(profile)
|
|
try:
|
|
return fn(*args)
|
|
finally:
|
|
profiles.clear_request_profile()
|
|
|
|
|
|
def _call(profile, handler_fn):
|
|
handler = MagicMock()
|
|
_as_profile(profile, handler_fn, handler)
|
|
return json.loads(handler.wfile.write.call_args[0][0])
|
|
|
|
|
|
def _reload(profile):
|
|
from api.commands import execute_agent_command
|
|
return _as_profile(profile, execute_agent_command, "/reload-mcp")
|
|
|
|
|
|
def _profile_scope(agent, profile):
|
|
return agent.key(agent.base / "profiles" / profile)
|
|
|
|
|
|
def _servers_by_name(payload):
|
|
return {row["name"]: row for row in payload["servers"]}
|
|
|
|
|
|
def test_init_pins_the_process_profile_home_for_the_agent(monkeypatch, tmp_path):
|
|
agent = _install(monkeypatch, tmp_path)
|
|
assert agent.pinned == str(agent.base)
|
|
assert profiles.get_process_profile_home() == agent.base
|
|
|
|
|
|
def test_process_wide_home_change_repins_in_the_same_step(monkeypatch, tmp_path):
|
|
"""``_set_hermes_home`` is the single writer of the process home: startup and
|
|
``switch_profile(process_wide=True)`` both go through it, so the MCP routing
|
|
anchor (WebUI and agent pin) can never drift from ``HERMES_HOME``."""
|
|
agent = _install(monkeypatch, tmp_path)
|
|
new_home = agent.base / "profiles" / "profile-write"
|
|
profiles._set_hermes_home(new_home)
|
|
assert os.environ["HERMES_HOME"] == str(new_home)
|
|
assert profiles.get_process_profile_home() == new_home
|
|
assert agent.pinned == str(new_home)
|
|
|
|
|
|
def test_status_and_inventory_bind_the_request_profile(monkeypatch, tmp_path):
|
|
from api.routes import _handle_mcp_servers_list, _handle_mcp_tools_list
|
|
|
|
agent = _install(monkeypatch, tmp_path)
|
|
agent.connect("atlassian", READ, scope=_profile_scope(agent, "profile-read"))
|
|
agent.connect("atlassian", WRITE, scope=_profile_scope(agent, "profile-write"))
|
|
|
|
servers = _call("profile-write", _handle_mcp_servers_list)
|
|
tools = _call("profile-write", _handle_mcp_tools_list)
|
|
assert servers["runtime_scope"] == tools["runtime_scope"] == "profile"
|
|
assert servers["servers"][0]["tool_count"] == 2
|
|
assert {t["name"] for t in tools["tools"]} == set(WRITE)
|
|
# get_mcp_status ran under the profile's home and was handed the displayed config.
|
|
status_calls = [c for c in agent.calls if c[0] == "status"]
|
|
assert all(c[1] == str(agent.base / "profiles" / "profile-write") for c in status_calls)
|
|
assert all(set(c[2]) == {"atlassian"} for c in status_calls)
|
|
# Nothing leaked into the request thread after the calls.
|
|
assert agent.override.get() is None and agent.secret.get() is None
|
|
|
|
read_tools = _call("profile-read", _handle_mcp_tools_list)
|
|
assert {t["name"] for t in read_tools["tools"]} == set(READ)
|
|
assert _call("default", _handle_mcp_tools_list)["total"] == 0
|
|
assert not [c for c in agent.calls if c[0] in ("discover", "shutdown")]
|
|
|
|
|
|
def test_inventory_excludes_the_process_profiles_global_slot(monkeypatch, tmp_path):
|
|
from api.routes import _handle_mcp_tools_list
|
|
|
|
agent = _install(monkeypatch, tmp_path)
|
|
agent.connect("atlassian", WRITE, scope=None) # launch profile's unscoped connection
|
|
agent.connect("atlassian", READ, scope=_profile_scope(agent, "profile-read"))
|
|
names = {t["name"] for t in _call("profile-read", _handle_mcp_tools_list)["tools"]}
|
|
assert names == set(READ)
|
|
|
|
|
|
def test_root_profile_status_hides_routed_profiles_runtime(monkeypatch, tmp_path):
|
|
"""The agent's launch-profile view is process-wide; WebUI narrows it to the
|
|
root profile's own (unscoped) connections, so a routed profile's live
|
|
connection or connect error never shows up as the root profile's."""
|
|
from api.routes import _handle_mcp_servers_list, _handle_mcp_tools_list
|
|
|
|
agent = _install(monkeypatch, tmp_path)
|
|
_configure(agent, "default", {"atlassian": {"command": "fake", "fake_tools": READ}})
|
|
agent.connect("atlassian", WRITE, scope=_profile_scope(agent, "profile-write"))
|
|
agent.fail("atlassian", scope=_profile_scope(agent, "profile-read"), error="bad token")
|
|
|
|
servers = _call("default", _handle_mcp_servers_list)
|
|
row = _servers_by_name(servers)["atlassian"]
|
|
assert servers["runtime_scope"] == "profile"
|
|
assert row["status"] == "configured" and row["active"] is False
|
|
assert not row["tool_count"]
|
|
assert _call("default", _handle_mcp_tools_list)["total"] == 0
|
|
|
|
# The root profile's own parked task (failed spawn, no session) must not borrow
|
|
# the routed profile's live same-named connection: the agent's process-wide
|
|
# view indexes by bare name and the last key wins.
|
|
agent.park("atlassian", scope=None, error="exec: not found")
|
|
row = _servers_by_name(_call("default", _handle_mcp_servers_list))["atlassian"]
|
|
assert row["status"] == "configured" and row["active"] is False and not row["tool_count"]
|
|
|
|
# The root profile's own connection is still reported normally.
|
|
agent.connect("atlassian", READ, scope=None)
|
|
row = _servers_by_name(_call("default", _handle_mcp_servers_list))["atlassian"]
|
|
assert row["status"] == "active" and row["tool_count"] == len(READ)
|
|
assert {t["name"] for t in _call("default", _handle_mcp_tools_list)["tools"]} == set(READ)
|
|
|
|
|
|
def test_adopted_connection_serves_status_inventory_and_reload(monkeypatch, tmp_path):
|
|
"""A profile that adopted another profile's identical connection sees it as
|
|
active; its reload counts it and never tears down the owner's connection."""
|
|
from api.routes import _handle_mcp_servers_list, _handle_mcp_tools_list
|
|
|
|
agent = _install(monkeypatch, tmp_path)
|
|
read_scope = _profile_scope(agent, "profile-read")
|
|
write_scope = _profile_scope(agent, "profile-write")
|
|
owner_conn = agent.connect("atlassian", READ, scope=read_scope)
|
|
agent.adopt("atlassian", owner=read_scope, adopter=write_scope)
|
|
|
|
row = _servers_by_name(_call("profile-write", _handle_mcp_servers_list))["atlassian"]
|
|
assert row["status"] == "active" and row["tool_count"] == len(READ)
|
|
assert {t["name"] for t in _call("profile-write", _handle_mcp_tools_list)["tools"]} == set(READ)
|
|
|
|
output = _reload("profile-write")
|
|
assert "Reconnected: atlassian" in output
|
|
assert "Added" not in output and "No MCP servers connected" not in output
|
|
assert ("shutdown", write_scope, {"atlassian"}) in agent.calls
|
|
assert agent.servers[(read_scope, "atlassian")] is owner_conn
|
|
assert write_scope in agent.adopters[(read_scope, "atlassian")]
|
|
|
|
|
|
def test_reload_with_an_empty_config_detaches_adopted_tools_but_keeps_the_owner(monkeypatch, tmp_path):
|
|
"""Deleting a profile's last server must stop its tools on reload, even when the
|
|
connection was adopted from another profile: the adopter's overlay goes, the
|
|
owner's connection stays (master's wildcard shutdown stopped the owner too)."""
|
|
from api.routes import _handle_mcp_tools_list
|
|
|
|
agent = _install(monkeypatch, tmp_path)
|
|
read_scope = _profile_scope(agent, "profile-read")
|
|
write_scope = _profile_scope(agent, "profile-write")
|
|
owner_conn = agent.connect("atlassian", READ, scope=read_scope)
|
|
agent.adopt("atlassian", owner=read_scope, adopter=write_scope)
|
|
assert {t["name"] for t in _call("profile-write", _handle_mcp_tools_list)["tools"]} == set(READ)
|
|
|
|
_configure(agent, "profile-write", {}) # the user deleted the profile's only server
|
|
output = _reload("profile-write")
|
|
|
|
assert "Removed: atlassian" in output and "Reconnected" not in output
|
|
assert _call("profile-write", _handle_mcp_tools_list)["total"] == 0
|
|
assert not agent.scoped_tools.get(write_scope)
|
|
assert write_scope not in agent.adopters.get((read_scope, "atlassian"), ())
|
|
# The owner is untouched: same connection, same tools.
|
|
assert agent.servers[(read_scope, "atlassian")] is owner_conn
|
|
assert {t["name"] for t in _call("profile-read", _handle_mcp_tools_list)["tools"]} == set(READ)
|
|
# The teardown stayed scoped to this profile (never the process-wide wildcard) and the
|
|
# overlay reconcile ran under this profile's home before rediscovery.
|
|
shutdowns = [c for c in agent.calls if c[0] == "shutdown"]
|
|
assert shutdowns == [("shutdown", write_scope, {"atlassian"})]
|
|
assert ("register", str(agent.base / "profiles" / "profile-write"), {}) in agent.calls
|
|
|
|
|
|
def test_portable_plugin_tools_are_listed_for_their_profile_without_a_config_entry(monkeypatch, tmp_path):
|
|
"""The agent merges plugin-provided (portable) MCP servers into the running config;
|
|
they never appear in the profile's raw ``mcp_servers``. Their tools must still be
|
|
listed for the profile whose registry slot holds them, and for no other profile."""
|
|
from api.routes import _handle_mcp_tools_list, _handle_notes_sources_list
|
|
|
|
agent = _install(monkeypatch, tmp_path)
|
|
write_scope = _profile_scope(agent, "profile-write")
|
|
agent.connect("notes-plugin", ["mcp__notes-plugin__search_notes"], scope=write_scope)
|
|
|
|
inventory = _call("profile-write", _handle_mcp_tools_list)
|
|
assert [(t["name"], t["server"]) for t in inventory["tools"]] == [
|
|
("mcp__notes-plugin__search_notes", "notes-plugin")
|
|
]
|
|
assert inventory["source"] == "tool_registry"
|
|
monkeypatch.setenv("HERMES_WEBUI_EXTERNAL_NOTES_SOURCES", "1")
|
|
notes = _call("profile-write", _handle_notes_sources_list)
|
|
assert [s["name"] for s in notes["sources"]] == ["notes-plugin"]
|
|
assert notes["sources"][0]["tool_count"] == 1
|
|
# Isolation still holds: another profile's slot does not list them.
|
|
assert _call("profile-read", _handle_mcp_tools_list)["total"] == 0
|
|
assert _call("default", _handle_mcp_tools_list)["total"] == 0
|
|
|
|
|
|
def test_skewed_routing_hides_runtime_and_refuses_reload(monkeypatch, tmp_path):
|
|
from api.routes import (
|
|
_handle_mcp_servers_list,
|
|
_handle_mcp_tools_list,
|
|
_handle_notes_sources_list,
|
|
)
|
|
|
|
agent = _install(monkeypatch, tmp_path, pin_supported=False)
|
|
_configure(agent, "profile-write", {
|
|
"atlassian": {"command": "fake", "fake_tools": WRITE},
|
|
"joplin": {"command": "fake", "fake_tools": []}, # a notes-drawer source
|
|
})
|
|
agent.connect("atlassian", READ, scope=None)
|
|
monkeypatch.setenv("HERMES_HOME", str(agent.base / "profiles" / "profile-write")) # turn mirror
|
|
|
|
servers = _call("profile-write", _handle_mcp_servers_list)
|
|
assert servers["runtime_scope"] == "unavailable"
|
|
assert servers["servers"][0]["status"] == "configured"
|
|
assert servers["servers"][0]["tool_count"] is None
|
|
assert _call("profile-write", _handle_mcp_tools_list)["total"] == 0
|
|
# The notes drawer reads the same inventory and must be told the runtime is withheld.
|
|
monkeypatch.setenv("HERMES_WEBUI_EXTERNAL_NOTES_SOURCES", "1")
|
|
notes = _call("profile-write", _handle_notes_sources_list)
|
|
assert notes["runtime_scope"] == "unavailable"
|
|
assert [s["name"] for s in notes["sources"]] == ["joplin"]
|
|
assert notes["sources"][0]["active"] is False
|
|
with pytest.raises(RuntimeError, match="could not be confirmed"):
|
|
_reload("profile-write")
|
|
assert not [c for c in agent.calls if c[0] in ("discover", "shutdown")]
|
|
assert "atlassian" in agent.servers
|
|
|
|
|
|
def test_reload_resets_only_the_requested_profile(monkeypatch, tmp_path):
|
|
agent = _install(monkeypatch, tmp_path)
|
|
read_scope = _profile_scope(agent, "profile-read")
|
|
write_scope = _profile_scope(agent, "profile-write")
|
|
read_conn = agent.connect("atlassian", READ, scope=read_scope)
|
|
agent.connect("atlassian", READ, scope=write_scope) # stale write-profile connection
|
|
|
|
output = _reload("profile-write")
|
|
assert "Reconnected: atlassian" in output
|
|
assert ("shutdown", write_scope, {"atlassian"}) in agent.calls
|
|
assert ("discover", str(agent.base / "profiles" / "profile-write")) in agent.calls
|
|
assert agent.servers[(read_scope, "atlassian")] is read_conn
|
|
assert agent.servers[(write_scope, "atlassian")]._registered_tool_names == WRITE
|
|
|
|
|
|
def test_reload_retries_the_profiles_failed_servers_only(monkeypatch, tmp_path):
|
|
"""A server that failed to spawn is under connect backoff and absent from the
|
|
ledger, so a scoped shutdown leaves its cooldown alone; the reload must still
|
|
retry it (as the process-wide wildcard did) without touching other owners'."""
|
|
agent = _install(monkeypatch, tmp_path)
|
|
write_scope = _profile_scope(agent, "profile-write")
|
|
agent.fail("atlassian", scope=write_scope, error="exec: not found")
|
|
agent.fail("atlassian", scope=None, error="root still broken") # root's own backoff
|
|
|
|
_configure(agent, "profile-write", {"atlassian": {"command": "fake", "fake_tools": WRITE}})
|
|
output = _reload("profile-write")
|
|
assert "Added: atlassian" in output
|
|
assert ("shutdown", write_scope, set()) in agent.calls
|
|
assert agent.servers[(write_scope, "atlassian")]._registered_tool_names == WRITE
|
|
assert (write_scope, "atlassian") not in agent.errors
|
|
assert "atlassian" in agent.retry_after and "atlassian" in agent.errors # root untouched
|
|
|
|
|
|
def test_root_profile_reload_never_uses_the_wildcard_shutdown(monkeypatch, tmp_path):
|
|
agent = _install(monkeypatch, tmp_path)
|
|
_configure(agent, "default", {"atlassian": {"command": "fake", "fake_tools": READ}})
|
|
agent.connect("atlassian", READ, scope=None)
|
|
write_conn = agent.connect("atlassian", WRITE, scope=_profile_scope(agent, "profile-write"))
|
|
|
|
assert "Reconnected: atlassian" in _reload("default")
|
|
shutdowns = [c for c in agent.calls if c[0] == "shutdown"]
|
|
assert shutdowns == [("shutdown", None, {"atlassian"})]
|
|
assert agent.servers[(_profile_scope(agent, "profile-write"), "atlassian")] is write_conn
|
|
|
|
|
|
def test_legacy_agent_keeps_process_wide_reload_but_reads_the_request_profile(monkeypatch, tmp_path):
|
|
from api.routes import _handle_mcp_servers_list
|
|
|
|
agent = _install(monkeypatch, tmp_path, pin_supported=False, scoped=False)
|
|
assert _call("profile-write", _handle_mcp_servers_list)["runtime_scope"] == "legacy_process"
|
|
_reload("profile-write")
|
|
assert ("shutdown", None, None) in agent.calls
|
|
assert ("discover", str(agent.base / "profiles" / "profile-write")) in agent.calls
|
|
|
|
|
|
def test_status_read_survives_an_uninspectable_get_mcp_status(monkeypatch, tmp_path):
|
|
"""Signature introspection failing must fall back to the bare call, not blank
|
|
every server's status."""
|
|
from api.routes import _handle_mcp_servers_list
|
|
|
|
agent = _install(monkeypatch, tmp_path)
|
|
agent.connect("atlassian", WRITE, scope=_profile_scope(agent, "profile-write"))
|
|
real = agent.modules["tools.mcp_tool"].get_mcp_status
|
|
|
|
class Uninspectable:
|
|
@property
|
|
def __signature__(self):
|
|
raise ValueError("no signature")
|
|
|
|
def __call__(self, *args, **kwargs):
|
|
assert not args and not kwargs, "bare call expected"
|
|
return real(configured={"atlassian": {"command": "fake"}})
|
|
|
|
monkeypatch.setattr(agent.modules["tools.mcp_tool"], "get_mcp_status", Uninspectable())
|
|
row = _servers_by_name(_call("profile-write", _handle_mcp_servers_list))["atlassian"]
|
|
assert row["status"] == "active" and row["tool_count"] == len(WRITE)
|
|
|
|
|
|
def test_ledger_key_helpers_prefer_the_agents_and_fall_back_without_them(monkeypatch, tmp_path):
|
|
from api import mcp_runtime
|
|
|
|
agent = _install(monkeypatch, tmp_path)
|
|
assert mcp_runtime.mcp_key_name(("scope-a", "atlassian")) == "atlassian"
|
|
assert mcp_runtime.mcp_key_scope(("scope-a", "atlassian")) == "scope-a"
|
|
assert mcp_runtime.mcp_key_name("atlassian") == "atlassian"
|
|
assert mcp_runtime.mcp_key_scope("atlassian") is None
|
|
monkeypatch.delitem(sys.modules, "tools.mcp_tool_scope")
|
|
assert mcp_runtime.mcp_key_name(("scope-a", "atlassian")) == "atlassian"
|
|
assert mcp_runtime.mcp_key_scope("atlassian") is None
|
|
assert mcp_runtime.accepts_keywords(len, "scope") is False
|
|
assert mcp_runtime.accepts_keywords(agent.modules["tools.mcp_tool"].shutdown_mcp_servers, "scope", "names")
|
|
|
|
|
|
def test_readonly_profile_scope_default_is_unchanged_and_root_is_opt_in(monkeypatch, tmp_path):
|
|
agent = _install(monkeypatch, tmp_path)
|
|
with profiles.profile_env_for_active_request_readonly("t") as bound:
|
|
assert bound is False and agent.override.get() is None
|
|
with profiles.profile_env_for_active_request_readonly("t", include_root=True) as bound:
|
|
assert bound is True and agent.override.get() == str(agent.base)
|
|
assert agent.override.get() is None
|
|
|
|
|
|
def test_runtime_scope_restores_context_after_exception(monkeypatch, tmp_path):
|
|
from api.mcp_runtime import mcp_runtime_scope
|
|
|
|
agent = _install(monkeypatch, tmp_path)
|
|
with pytest.raises(KeyError):
|
|
_as_profile("profile-write", lambda: _raise_inside(mcp_runtime_scope, agent))
|
|
assert agent.override.get() is None and agent.secret.get() is None
|
|
|
|
|
|
def _raise_inside(scope_factory, agent):
|
|
with scope_factory("t") as view:
|
|
assert view.trusted and agent.override.get() is not None and agent.secret.get() is not None
|
|
raise KeyError("boom")
|