fix(gateway): format scoped MCP server names during reload

gateway/run_turn.py

@@ -2338,15 +2338,16 @@ async def _execute_mcp_reload(self, event: MessageEvent) -> str:
             from tools.mcp_tool_discovery import discover_mcp_tools
             from tools.mcp_tool import _servers, _lock, _server_visible_in_scope
             from tools.mcp_tool_agent import reprobe_tool_availability
+            from tools.mcp_tool_scope import _key_name
             from tools.registry import registry
 
             reload_scope = registry.current_scope_key() if multiplex else None
 
             def _scoped_server_names() -> set:
                 with _lock:
                     return {
-                        name for name in _servers
-                        if _server_visible_in_scope(name, reload_scope)
+                        _key_name(key) for key in _servers
+                        if _server_visible_in_scope(key, reload_scope)
                     }
 
             old_servers = _scoped_server_names()

tests/gateway/test_multiplex_mcp_discovery.py

@@ -102,6 +102,54 @@ def fake_discover() -> list[str]:
     assert "default-srv" not in result
 
 
+@pytest.mark.asyncio
+async def test_reload_mcp_formats_scoped_connection_keys_before_refreshing_cached_agents(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    """Connection-ledger tuple keys are internal; reload reports server names and completes refresh."""
+    from gateway.run import GatewayRunner
+    from tools import mcp_tool
+    from tools import mcp_tool_discovery as _mcp_discovery
+    from tools import mcp_tool_lifecycle as _mcp_lifecycle
+
+    launch_scope = hermes_home_key(tmp_path / "default")
+    worker_home = tmp_path / "profiles" / "worker"
+    worker_home.mkdir(parents=True)
+    worker_scope = hermes_home_key(worker_home)
+    launch_key = (launch_scope, "default-srv")
+    worker_key = (worker_scope, "worker-srv")
+
+    runner = GatewayRunner.__new__(GatewayRunner)
+    runner.config = GatewayConfig(multiplex_profiles=True)
+    runner._resolve_profile_home_for_source = MagicMock(return_value=worker_home)
+    runner._mcp_reload_refresh_cached_agents = MagicMock()
+    runner._async_session_store = SimpleNamespace(
+        get_or_create_session=MagicMock(side_effect=RuntimeError("skip transcript")),
+    )
+
+    monkeypatch.setattr(mcp_tool, "_servers", {launch_key: object(), worker_key: object()})
+    monkeypatch.setattr(
+        mcp_tool, "_server_scope_keys",
+        {launch_key: launch_scope, worker_key: worker_scope},
+    )
+    monkeypatch.setattr(_mcp_lifecycle, "shutdown_mcp_servers", lambda **_kwargs: None)
+    monkeypatch.setattr(_mcp_discovery, "discover_mcp_tools", lambda: [])
+
+    event = MessageEvent(
+        text="/reload-mcp", message_id="m1",
+        source=SessionSource(
+            platform=Platform.TELEGRAM, user_id="u1", chat_id="c1",
+            chat_type="dm", profile="worker",
+        ),
+    )
+    result = await runner._execute_mcp_reload(event)
+
+    assert "MCP reload failed" not in result
+    assert "worker-srv" in result
+    assert "default-srv" not in result
+    runner._mcp_reload_refresh_cached_agents.assert_called_once_with(True, "worker")
+
+
 @pytest.mark.asyncio
 async def test_reload_mcp_reports_a_shared_server_to_a_non_owner_profile(
     tmp_path: Path, monkeypatch: pytest.MonkeyPatch