fix(acp): add bounded wait + late-refresh for configured MCP servers
ACP entry.py fires MCP discovery in a background daemon thread, but _make_agent snapshots tools once at build and never re-reads the registry. Unlike CLI/TUI, ACP had no bounded wait before the snapshot and no late-refresh for configured (config.yaml) MCP servers — a reachable-but- slow server that finished after agent build was invisible for the whole session. Changes: - acp_adapter/session.py (_make_agent): call wait_for_mcp_discovery() before AIAgent construction, bounded by mcp_discovery_timeout (default ~1.5s). A dead server can't block; servers that miss the bound are picked up by the late-refresh below. - acp_adapter/server.py (_schedule_mcp_late_refresh): new method on HermesACPAgent — if discovery is still in flight after session creation, spawns an off-critical-path daemon that joins it (bounded 30s), then rebuilds the tool snapshot via the shared refresh_agent_mcp_tools helper. Cache-safe: only runs pre-first-turn (_user_turn_count/_api_call_count both 0); once the user has sent a message the snapshot is frozen, exactly as TUI PR #48403 does. - Called from new_session, load_session, resume_session. - Mirrors the TUI pattern (tui_gateway _schedule_mcp_late_refresh, PR #48403) and the CLI pattern (get_tool_definitions → wait_for_mcp_discovery). Tests: - Replace the AST-based test (source-text inspection) with three behavioral regression tests in tests/acp_adapter/test_acp_mcp_discovery.py: 1. Blocked discovery does not block startup (non-blocking contract) 2. Delayed discovery lands tools via late-refresh (pre-first-turn) 3. Late-refresh is cache-safe: skips rebuild after first turn Addresses teknium1 review on PR #32811.
This commit is contained in:
Sergey Prontsevich
kshitij
parent
89f0b63da4
commit
640de6562e
|
|
@ -1037,6 +1037,84 @@ class HermesACPAgent(acp.Agent):
|
|||
exc_info=True,
|
||||
)
|
||||
|
||||
def _schedule_mcp_late_refresh(self, state: SessionState) -> None:
|
||||
"""Refresh the agent's tool snapshot when background MCP discovery lands late.
|
||||
|
||||
ACP entry.py starts MCP tool discovery in a background daemon thread so a
|
||||
slow/dead configured server can't block ``asyncio.run()``. ``_make_agent``
|
||||
briefly joins that thread (``wait_for_mcp_discovery``, bounded ~1.5s) so
|
||||
already-spawning fast servers land in the snapshot — but a server slower
|
||||
than the bound lands *after* the agent is built, leaving its tools absent
|
||||
for the whole session.
|
||||
|
||||
This schedules an off-critical-path daemon that waits for discovery to
|
||||
finish (bounded 30s), then rebuilds the snapshot via the shared
|
||||
``refresh_agent_mcp_tools`` helper — the same rebuild ``/reload-mcp``
|
||||
performs, but automatic. Mirrors the TUI late-refresh (PR #48403).
|
||||
|
||||
Cache safety: the rebuild only runs while the session is still
|
||||
pre-first-turn (no API call made yet → nothing cached to invalidate).
|
||||
Once the user has sent a message we leave the snapshot frozen rather
|
||||
than break the cached prompt prefix mid-conversation; late tools then
|
||||
require an explicit ``/reload-mcp`` (user-consented), exactly as today.
|
||||
No-op when discovery already finished, when the join times out, when the
|
||||
registry was unchanged, or when the session was closed while waiting.
|
||||
"""
|
||||
try:
|
||||
from hermes_cli.mcp_startup import mcp_discovery_in_flight
|
||||
except Exception:
|
||||
return
|
||||
if not mcp_discovery_in_flight():
|
||||
return
|
||||
|
||||
import threading
|
||||
|
||||
agent = state.agent
|
||||
session_id = state.session_id
|
||||
|
||||
def _wait_then_refresh() -> None:
|
||||
try:
|
||||
from hermes_cli.mcp_startup import join_mcp_discovery
|
||||
|
||||
if not join_mcp_discovery(timeout=30.0):
|
||||
return
|
||||
|
||||
# Session may have been closed while we waited.
|
||||
current = self.session_manager.get_session(session_id)
|
||||
if current is None or current.agent is not agent:
|
||||
return
|
||||
|
||||
# Cache safety: never rebuild the tool list once the conversation
|
||||
# has started — that would invalidate the cached prompt prefix.
|
||||
if (
|
||||
int(getattr(agent, "_user_turn_count", 0) or 0) > 0
|
||||
or int(getattr(agent, "_api_call_count", 0) or 0) > 0
|
||||
):
|
||||
return
|
||||
|
||||
from tools.mcp_tool import refresh_agent_mcp_tools
|
||||
|
||||
added = refresh_agent_mcp_tools(agent)
|
||||
if added:
|
||||
logger.info(
|
||||
"Session %s: late MCP refresh added %d tools: %s",
|
||||
session_id,
|
||||
len(added),
|
||||
", ".join(sorted(added)),
|
||||
)
|
||||
except Exception:
|
||||
logger.debug(
|
||||
"Session %s: late MCP refresh failed",
|
||||
session_id,
|
||||
exc_info=True,
|
||||
)
|
||||
|
||||
threading.Thread(
|
||||
target=_wait_then_refresh,
|
||||
name=f"acp-mcp-late-refresh-{session_id}",
|
||||
daemon=True,
|
||||
).start()
|
||||
|
||||
# ---- ACP lifecycle ------------------------------------------------------
|
||||
|
||||
async def initialize(
|
||||
|
|
@ -1343,6 +1421,7 @@ class HermesACPAgent(acp.Agent):
|
|||
) -> NewSessionResponse:
|
||||
state = self.session_manager.create_session(cwd=cwd)
|
||||
await self._register_session_mcp_servers(state, mcp_servers)
|
||||
self._schedule_mcp_late_refresh(state)
|
||||
logger.info("New session %s (cwd=%s)", state.session_id, cwd)
|
||||
self._schedule_available_commands_update(state.session_id)
|
||||
self._schedule_usage_update(state)
|
||||
|
|
@ -1367,6 +1446,7 @@ class HermesACPAgent(acp.Agent):
|
|||
logger.warning("load_session: session %s not found", session_id)
|
||||
return None
|
||||
await self._register_session_mcp_servers(state, mcp_servers)
|
||||
self._schedule_mcp_late_refresh(state)
|
||||
logger.info("Loaded session %s", session_id)
|
||||
# Per ACP spec, `session/load` must stream the prior conversation back
|
||||
# to the client via `session/update` notifications BEFORE responding,
|
||||
|
|
@ -1414,6 +1494,7 @@ class HermesACPAgent(acp.Agent):
|
|||
logger.warning("resume_session: session %s not found, creating new", session_id)
|
||||
state = self.session_manager.create_session(cwd=cwd)
|
||||
await self._register_session_mcp_servers(state, mcp_servers)
|
||||
self._schedule_mcp_late_refresh(state)
|
||||
logger.info("Resumed session %s", state.session_id)
|
||||
# See `load_session` above for the spec rationale — replay must
|
||||
# complete before the response so clients receive the full transcript
|
||||
|
|
|
|||
|
|
@ -648,6 +648,23 @@ class SessionManager:
|
|||
logger.debug("ACP session falling back to default provider resolution", exc_info=True)
|
||||
|
||||
_register_task_cwd(session_id, cwd)
|
||||
|
||||
# Bounded wait for background MCP discovery so already-spawning fast
|
||||
# servers land in the agent's tool snapshot. ACP entry.py fires
|
||||
# discovery in a background daemon thread (start_background_mcp_discovery);
|
||||
# the agent snapshots tools once at build (run_agent/agent_init) and
|
||||
# never re-reads the registry, so without this join a reachable-but-
|
||||
# slow configured server would be invisible for the whole session.
|
||||
# Bounded by ``mcp_discovery_timeout`` (config.yaml, default ~1.5s) so a
|
||||
# dead server can't block — servers that miss the bound are picked up
|
||||
# by the automatic late-refresh (see HermesACPAgent._schedule_mcp_late_refresh).
|
||||
try:
|
||||
from hermes_cli.mcp_startup import wait_for_mcp_discovery
|
||||
|
||||
wait_for_mcp_discovery()
|
||||
except Exception:
|
||||
logger.debug("ACP: bounded MCP discovery wait failed", exc_info=True)
|
||||
|
||||
agent = AIAgent(**kwargs)
|
||||
# Codex app-server sessions are spawned lazily on the first turn. Stamp
|
||||
# the ACP workspace onto the agent so the Codex runtime starts from the
|
||||
|
|
|
|||
|
|
@ -0,0 +1,264 @@
|
|||
"""Behavioral regression tests for ACP background MCP discovery + late-refresh.
|
||||
|
||||
These replace the previous AST-based test that only inspected source text.
|
||||
They verify the *behavior*: (1) a blocked discovery doesn't block startup, and
|
||||
(2) a delayed-but-reachable MCP server's tools land in the agent's snapshot
|
||||
via the automatic late-refresh, cache-safely (pre-first-turn only).
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import sys
|
||||
import threading
|
||||
import time
|
||||
import types
|
||||
from contextlib import nullcontext
|
||||
from types import ModuleType, SimpleNamespace
|
||||
|
||||
import pytest
|
||||
|
||||
from acp_adapter.server import HermesACPAgent
|
||||
from acp_adapter.session import SessionManager, SessionState
|
||||
from hermes_cli import mcp_startup
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Shared helpers
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class FakeAgent:
|
||||
"""Minimal stand-in for AIAgent with the attributes late-refresh touches."""
|
||||
|
||||
def __init__(self):
|
||||
self.model = "fake-model"
|
||||
self.provider = "fake-provider"
|
||||
self.enabled_toolsets = ["hermes-acp"]
|
||||
self.disabled_toolsets = []
|
||||
self.tools = []
|
||||
self.valid_tool_names = set()
|
||||
self._user_turn_count = 0
|
||||
self._api_call_count = 0
|
||||
|
||||
|
||||
class NoopDb:
|
||||
def get_session(self, *_a, **_k):
|
||||
return None
|
||||
|
||||
def create_session(self, *_a, **_k):
|
||||
return None
|
||||
|
||||
def update_session(self, *_a, **_k):
|
||||
return None
|
||||
|
||||
|
||||
def _mod(name: str, **attrs) -> ModuleType:
|
||||
module = ModuleType(name)
|
||||
for key, value in attrs.items():
|
||||
setattr(module, key, value)
|
||||
return module
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def _reset_mcp_startup_state():
|
||||
"""Ensure each test starts with a clean discovery thread state."""
|
||||
saved_started = mcp_startup._mcp_discovery_started
|
||||
saved_thread = mcp_startup._mcp_discovery_thread
|
||||
mcp_startup._mcp_discovery_started = False
|
||||
mcp_startup._mcp_discovery_thread = None
|
||||
yield
|
||||
thread = mcp_startup._mcp_discovery_thread
|
||||
if thread is not None and thread.is_alive():
|
||||
thread.join(timeout=2.0)
|
||||
mcp_startup._mcp_discovery_started = saved_started
|
||||
mcp_startup._mcp_discovery_thread = saved_thread
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Test 1 — blocked discovery does not block startup
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_acp_background_discovery_does_not_block_startup(monkeypatch):
|
||||
"""start_background_mcp_discovery must return immediately even if discovery hangs."""
|
||||
block = threading.Event()
|
||||
|
||||
def _blocking_discover():
|
||||
block.wait(timeout=5.0)
|
||||
|
||||
monkeypatch.setitem(
|
||||
sys.modules,
|
||||
"hermes_cli.config",
|
||||
_mod(
|
||||
"hermes_cli.config",
|
||||
read_raw_config=lambda: {"mcp_servers": {"slow": {"url": "https://mcp.example.test"}}},
|
||||
),
|
||||
)
|
||||
monkeypatch.setitem(
|
||||
sys.modules,
|
||||
"tools.mcp_oauth",
|
||||
_mod("tools.mcp_oauth", suppress_interactive_oauth=lambda: nullcontext()),
|
||||
)
|
||||
monkeypatch.setitem(
|
||||
sys.modules,
|
||||
"tools.mcp_tool",
|
||||
_mod("tools.mcp_tool", discover_mcp_tools=_blocking_discover),
|
||||
)
|
||||
|
||||
start = time.monotonic()
|
||||
mcp_startup.start_background_mcp_discovery(
|
||||
logger=SimpleNamespace(debug=lambda *_a, **_k: None),
|
||||
thread_name="test-acp-discovery",
|
||||
)
|
||||
elapsed = time.monotonic() - start
|
||||
|
||||
assert elapsed < 0.2, "start_background_mcp_discovery blocked for {:.3f}s".format(elapsed)
|
||||
assert mcp_startup._mcp_discovery_thread is not None
|
||||
assert mcp_startup._mcp_discovery_thread.is_alive()
|
||||
block.set()
|
||||
mcp_startup._mcp_discovery_thread.join(timeout=2.0)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Test 2 — delayed discovery lands tools via late-refresh (pre-first-turn)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_acp_late_refresh_adds_tools_when_discovery_lands_after_build(monkeypatch):
|
||||
"""A slow MCP server that finishes after agent build must still appear in tools."""
|
||||
|
||||
discovery_block = threading.Event()
|
||||
discovery_done = threading.Event()
|
||||
|
||||
def _slow_discover():
|
||||
discovery_block.wait(timeout=5.0)
|
||||
discovery_done.set()
|
||||
|
||||
monkeypatch.setitem(
|
||||
sys.modules,
|
||||
"hermes_cli.config",
|
||||
_mod(
|
||||
"hermes_cli.config",
|
||||
read_raw_config=lambda: {"mcp_servers": {"slow": {"url": "https://mcp.example.test"}}},
|
||||
),
|
||||
)
|
||||
monkeypatch.setitem(
|
||||
sys.modules,
|
||||
"tools.mcp_oauth",
|
||||
_mod("tools.mcp_oauth", suppress_interactive_oauth=lambda: nullcontext()),
|
||||
)
|
||||
monkeypatch.setitem(
|
||||
sys.modules,
|
||||
"tools.mcp_tool",
|
||||
_mod("tools.mcp_tool", discover_mcp_tools=_slow_discover),
|
||||
)
|
||||
|
||||
mcp_startup.start_background_mcp_discovery(
|
||||
logger=SimpleNamespace(debug=lambda *_a, **_k: None),
|
||||
thread_name="test-acp-late",
|
||||
)
|
||||
|
||||
# Build the session immediately — discovery is still in flight.
|
||||
fake = FakeAgent()
|
||||
manager = SessionManager(agent_factory=lambda **_k: fake, db=NoopDb())
|
||||
acp_agent = HermesACPAgent(session_manager=manager)
|
||||
state = manager.create_session(cwd=".")
|
||||
|
||||
# Discovery is blocked, so it must still be in flight.
|
||||
assert not discovery_done.is_set(), "discovery finished too early for this test"
|
||||
|
||||
# Track refresh_agent_mcp_tools calls.
|
||||
refreshed = []
|
||||
|
||||
def _fake_refresh(agent, **_kw):
|
||||
agent.tools = [{"function": {"name": "mcp_slow_tool"}}]
|
||||
agent.valid_tool_names = {"mcp_slow_tool"}
|
||||
refreshed.append(agent)
|
||||
return {"mcp_slow_tool"}
|
||||
|
||||
monkeypatch.setitem(
|
||||
sys.modules,
|
||||
"tools.mcp_tool",
|
||||
_mod("tools.mcp_tool", refresh_agent_mcp_tools=_fake_refresh),
|
||||
)
|
||||
|
||||
# Trigger late-refresh.
|
||||
acp_agent._schedule_mcp_late_refresh(state)
|
||||
|
||||
# Release discovery so the late-refresh daemon can proceed.
|
||||
discovery_block.set()
|
||||
|
||||
# Wait for the late-refresh daemon to finish.
|
||||
deadline = time.monotonic() + 5.0
|
||||
while not refreshed and time.monotonic() < deadline:
|
||||
time.sleep(0.01)
|
||||
|
||||
assert refreshed, "late-refresh daemon did not call refresh_agent_mcp_tools"
|
||||
assert refreshed[0] is fake
|
||||
assert "mcp_slow_tool" in fake.valid_tool_names
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Test 3 — late-refresh is cache-safe: skips after first turn
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_acp_late_refresh_skips_after_first_turn(monkeypatch):
|
||||
"""Once the user has sent a message, late-refresh must NOT rebuild tools."""
|
||||
|
||||
discovery_block = threading.Event()
|
||||
|
||||
def _slow_discover():
|
||||
discovery_block.wait(timeout=5.0)
|
||||
|
||||
monkeypatch.setitem(
|
||||
sys.modules,
|
||||
"hermes_cli.config",
|
||||
_mod(
|
||||
"hermes_cli.config",
|
||||
read_raw_config=lambda: {"mcp_servers": {"slow": {"url": "https://mcp.example.test"}}},
|
||||
),
|
||||
)
|
||||
monkeypatch.setitem(
|
||||
sys.modules,
|
||||
"tools.mcp_oauth",
|
||||
_mod("tools.mcp_oauth", suppress_interactive_oauth=lambda: nullcontext()),
|
||||
)
|
||||
monkeypatch.setitem(
|
||||
sys.modules,
|
||||
"tools.mcp_tool",
|
||||
_mod("tools.mcp_tool", discover_mcp_tools=_slow_discover),
|
||||
)
|
||||
|
||||
mcp_startup.start_background_mcp_discovery(
|
||||
logger=SimpleNamespace(debug=lambda *_a, **_k: None),
|
||||
thread_name="test-acp-cache",
|
||||
)
|
||||
|
||||
fake = FakeAgent()
|
||||
fake._api_call_count = 1 # simulate: user already sent a message
|
||||
manager = SessionManager(agent_factory=lambda **_k: fake, db=NoopDb())
|
||||
acp_agent = HermesACPAgent(session_manager=manager)
|
||||
state = manager.create_session(cwd=".")
|
||||
|
||||
refreshed = []
|
||||
|
||||
def _fake_refresh(agent, **_kw):
|
||||
refreshed.append(agent)
|
||||
return set()
|
||||
|
||||
monkeypatch.setitem(
|
||||
sys.modules,
|
||||
"tools.mcp_tool",
|
||||
_mod("tools.mcp_tool", refresh_agent_mcp_tools=_fake_refresh),
|
||||
)
|
||||
|
||||
acp_agent._schedule_mcp_late_refresh(state)
|
||||
|
||||
# Release discovery so the daemon can proceed (if it were going to).
|
||||
discovery_block.set()
|
||||
|
||||
# Give the daemon time to run (if it were going to).
|
||||
time.sleep(0.5)
|
||||
|
||||
assert not refreshed, "late-refresh rebuilt tools after the first turn — cache broken!"
|
||||
Loading…
Reference in New Issue