From eded89ace26f29ababa098baf8f69147038bf9b9 Mon Sep 17 00:00:00 2001 From: Seppe Gadeyne Date: Wed, 15 Jul 2026 23:42:55 +0200 Subject: [PATCH 1/8] test(mcp): cover parked shutdown drain path --- tests/tools/test_mcp_tool.py | 73 ++++++++++++++++++++++++++++++++++++ 1 file changed, 73 insertions(+) diff --git a/tests/tools/test_mcp_tool.py b/tests/tools/test_mcp_tool.py index a2cf61e04a5..cc0c3e4089f 100644 --- a/tests/tools/test_mcp_tool.py +++ b/tests/tools/test_mcp_tool.py @@ -1795,6 +1795,79 @@ class TestShutdown: assert len(_servers) == 0 mock_server.shutdown.assert_called_once() + def test_shutdown_drains_parked_server_after_bounded_wait_expires(self): + """The public shutdown path drains a parked server if graceful shutdown stalls. + + This exercises the production ownership path: a real ``MCPServerTask`` + is registered, its parked waiter owns child tasks on the shared loop, + and the bounded wait for the scheduled shutdown expires. The loop owner + must still cancel and drain that waiter before closing the loop. + """ + import tools.mcp_tool as mcp_mod + from tools.mcp_tool import MCPServerTask, shutdown_mcp_servers + + shutdown_started = threading.Event() + parked_task_done = threading.Event() + scheduled_shutdown = {} + + class StalledShutdownServer(MCPServerTask): + async def shutdown(self): + shutdown_started.set() + await asyncio.Event().wait() + + server = StalledShutdownServer("parked") + with mcp_mod._lock: + mcp_mod._servers.clear() + mcp_mod._server_connecting.clear() + mcp_mod._ensure_mcp_loop() + with mcp_mod._lock: + loop = mcp_mod._mcp_loop + assert loop is not None + + async def install_parked_waiter(): + task = asyncio.create_task(server._wait_for_reconnect_or_shutdown()) + server._task = task + task.add_done_callback(lambda _task: parked_task_done.set()) + await asyncio.sleep(0) + return task + + parked_task = asyncio.run_coroutine_threadsafe( + install_parked_waiter(), loop + ).result(timeout=2) + with mcp_mod._lock: + mcp_mod._servers[server.name] = server + + def schedule_then_report_timeout(coro, target_loop, **_kwargs): + future = asyncio.run_coroutine_threadsafe(coro, target_loop) + scheduled_shutdown["future"] = future + + class TimedOutFuture: + def result(self, timeout): + assert timeout > 0 + assert shutdown_started.wait(timeout=2) + raise TimeoutError("simulated bounded MCP shutdown timeout") + + return TimedOutFuture() + + try: + with patch( + "agent.async_utils.safe_schedule_threadsafe", + side_effect=schedule_then_report_timeout, + ): + shutdown_mcp_servers() + + assert loop.is_closed() + assert parked_task_done.is_set(), ( + "parked MCPServerTask was not drained before its loop closed" + ) + assert parked_task.done() + assert scheduled_shutdown["future"].done() + finally: + with mcp_mod._lock: + mcp_mod._servers.clear() + mcp_mod._server_connecting.clear() + mcp_mod._stop_mcp_loop() + def test_shutdown_deregisters_registered_tools(self): """shutdown_mcp_servers removes MCP tools and their raw alias.""" import tools.mcp_tool as mcp_mod From cc21c4f78ff661e4709f1b6b3d96bf468b16bb4e Mon Sep 17 00:00:00 2001 From: shady Date: Fri, 17 Jul 2026 10:01:41 +0300 Subject: [PATCH 2/8] fix(mcp): drain pending tasks before closing the MCP loop MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit _stop_mcp_loop() stopped and closed the background loop without reaping the tasks still on it. A task left suspended is resumed later by the GC, whose finalizer drives its cleanup against the now-closed loop: Exception ignored in: File "tools/mcp_tool.py", line 2947, in run parked = await self._wait_for_reconnect_or_shutdown( File "tools/mcp_tool.py", line 2161, in _wait_for_reconnect_or_shutdown t.cancel() RuntimeError: Event loop is closed shutdown_mcp_servers() only reaps servers held in _servers, so a server that parked after exhausting its initial-connect budget — never inserted there, because start() raises _error before the caller registers it — has no owner to signal it and stays suspended until the loop is gone. Drain the loop the way asyncio.run() does: cancel the remaining tasks and gather them while the loop is still open, so each runs its own finally. Cancel alone is not enough — Task.cancel() only schedules the throw. This resolves the reported traceback, but not the ownership bug that strands the task in the first place; that needs a follow-up. Deliberately not using "Fixes" so #60197 stays open for it. Addresses #60197 Addresses #66113 Co-Authored-By: Claude Opus 4.8 --- tests/tools/test_mcp_stability.py | 72 +++++++++++++++++++++++++++++++ tools/mcp_tool.py | 35 +++++++++++++++ 2 files changed, 107 insertions(+) diff --git a/tests/tools/test_mcp_stability.py b/tests/tools/test_mcp_stability.py index f204615aca0..b843439fff7 100644 --- a/tests/tools/test_mcp_stability.py +++ b/tests/tools/test_mcp_stability.py @@ -736,3 +736,75 @@ class TestMCPInitialConnectionRetry: await task asyncio.get_event_loop().run_until_complete(_run()) + + +# --------------------------------------------------------------------------- +# Fix: drain pending tasks before closing the MCP loop +# --------------------------------------------------------------------------- + +class TestMCPLoopDrainOnStop: + """_stop_mcp_loop reaps pending tasks while the loop is still open.""" + + def test_pending_task_cleanup_runs_before_close(self): + """A task left on the loop must finish its cleanup before close. + + Regression for #60197: a task still suspended when the loop closes is + resumed later by the GC, and its ``finally`` then drives ``cancel()`` + -> ``call_soon()`` against the closed loop, surfacing as an ignored + ``RuntimeError: Event loop is closed``. Servers absent from + ``_servers`` (e.g. parked after an initial-connect failure) never get + ``shutdown()``, so this drain is their only reaper. + """ + import time + import tools.mcp_tool as mcp_mod + + state = { + "started": False, + "cleanup_ran": False, + "cleanup_error": None, + "task": None, + } + + async def _parked(): + state["task"] = asyncio.current_task() + state["started"] = True + try: + await asyncio.sleep(3600) + finally: + # Mirrors _wait_for_reconnect_or_shutdown's finally: cancelling + # a helper task needs call_soon(), which a closed loop rejects. + try: + helper = asyncio.ensure_future(asyncio.sleep(0)) + helper.cancel() + state["cleanup_ran"] = True + except BaseException as exc: + state["cleanup_error"] = exc + + with mcp_mod._lock: + mcp_mod._servers.clear() + mcp_mod._server_connecting.clear() + try: + mcp_mod._ensure_mcp_loop() + with mcp_mod._lock: + loop = mcp_mod._mcp_loop + assert loop is not None + asyncio.run_coroutine_threadsafe(_parked(), loop) + + deadline = time.monotonic() + 5 + while not state["started"] and time.monotonic() < deadline: + time.sleep(0.01) + assert state["started"], "task never started on the MCP loop" + + mcp_mod._stop_mcp_loop() + + assert state["task"] is not None + assert state["task"].done(), "task left pending when the loop closed" + assert state["cleanup_error"] is None, ( + f"cleanup ran against a closed loop: {state['cleanup_error']!r}" + ) + assert state["cleanup_ran"], "task cleanup never ran" + finally: + with mcp_mod._lock: + mcp_mod._servers.clear() + mcp_mod._server_connecting.clear() + mcp_mod._stop_mcp_loop() diff --git a/tools/mcp_tool.py b/tools/mcp_tool.py index af7fc6b20ec..bc5a8d7f273 100644 --- a/tools/mcp_tool.py +++ b/tools/mcp_tool.py @@ -6618,6 +6618,23 @@ def _stop_mcp_loop_if_idle() -> bool: return _stop_mcp_loop(only_if_idle=True) +async def _drain_mcp_loop_tasks() -> None: + """Cancel every task still pending on the MCP loop and reap it. + + Cancelling is not enough on its own: ``Task.cancel()`` only schedules the + throw, so the task must be awaited before the loop goes away. Gather them + here — while the loop is still open — so each one runs its own ``finally``. + """ + current = asyncio.current_task() + pending = [t for t in asyncio.all_tasks() if t is not current and not t.done()] + if not pending: + return + logger.debug("Draining %d pending task(s) from the MCP loop", len(pending)) + for task in pending: + task.cancel() + await asyncio.gather(*pending, return_exceptions=True) + + def _stop_mcp_loop(*, only_if_idle: bool = False) -> bool: """Stop the background event loop and join its thread.""" global _mcp_loop, _mcp_thread @@ -6630,6 +6647,24 @@ def _stop_mcp_loop(*, only_if_idle: bool = False) -> bool: _mcp_loop = None _mcp_thread = None if loop is not None: + # Drain before stopping: closing the loop with tasks still suspended + # leaves their coroutines for the GC, whose finalizer then resumes them + # to run cleanup against a loop that is already closed -> "Event loop + # is closed" (#60197). ``shutdown_mcp_servers`` only reaps servers held + # in ``_servers``, so anything else left on this loop ends up here. + if loop.is_running(): + from agent.async_utils import safe_schedule_threadsafe + + future = safe_schedule_threadsafe( + _drain_mcp_loop_tasks(), loop, + logger=logger, + log_message="MCP loop drain: failed to schedule", + ) + if future is not None: + try: + future.result(timeout=5) + except BaseException as exc: + logger.debug("Error draining MCP loop tasks: %s", exc) loop.call_soon_threadsafe(loop.stop) if thread is not None: thread.join(timeout=5) From cac74e06c85c1d1eadca82d088cc4f425e6a859d Mon Sep 17 00:00:00 2001 From: Seppe Gadeyne Date: Fri, 17 Jul 2026 10:06:54 +0200 Subject: [PATCH 3/8] fix(mcp): bound loop-owned shutdown drain --- tests/tools/test_mcp_stability.py | 89 ++++++++++++++++++++++++++++++- tests/tools/test_mcp_tool.py | 7 +++ tools/mcp_tool.py | 52 ++++++++++++++---- 3 files changed, 138 insertions(+), 10 deletions(-) diff --git a/tests/tools/test_mcp_stability.py b/tests/tools/test_mcp_stability.py index b843439fff7..712a80c04eb 100644 --- a/tests/tools/test_mcp_stability.py +++ b/tests/tools/test_mcp_stability.py @@ -795,7 +795,23 @@ class TestMCPLoopDrainOnStop: time.sleep(0.01) assert state["started"], "task never started on the MCP loop" - mcp_mod._stop_mcp_loop() + stop_saw_cleanup = [] + call_soon_threadsafe = loop.call_soon_threadsafe + + def record_stop_order(callback, *args, **kwargs): + if ( + getattr(callback, "__self__", None) is loop + and getattr(callback, "__name__", None) == "stop" + ): + stop_saw_cleanup.append(state["cleanup_ran"]) + return call_soon_threadsafe(callback, *args, **kwargs) + + with patch.object( + loop, + "call_soon_threadsafe", + side_effect=record_stop_order, + ): + mcp_mod._stop_mcp_loop() assert state["task"] is not None assert state["task"].done(), "task left pending when the loop closed" @@ -803,8 +819,79 @@ class TestMCPLoopDrainOnStop: f"cleanup ran against a closed loop: {state['cleanup_error']!r}" ) assert state["cleanup_ran"], "task cleanup never ran" + assert stop_saw_cleanup == [True], "loop.stop ran before task cleanup" finally: with mcp_mod._lock: mcp_mod._servers.clear() mcp_mod._server_connecting.clear() mcp_mod._stop_mcp_loop() + + def test_drain_is_bounded_when_task_ignores_cancellation(self, caplog): + """A cancellation-resistant task must not hang final MCP shutdown.""" + import tools.mcp_tool as mcp_mod + + async def _run(): + release = asyncio.Event() + + async def cancellation_resistant(): + try: + await asyncio.Event().wait() + except asyncio.CancelledError: + await release.wait() + + task = asyncio.create_task(cancellation_resistant()) + await asyncio.sleep(0) + try: + with caplog.at_level("WARNING", logger=mcp_mod.logger.name): + await mcp_mod._drain_mcp_loop_tasks(timeout=0.01) + assert not task.done(), "drain waited indefinitely for resistant task" + finally: + release.set() + if not task.done() and task.cancelling() == 0: + task.cancel() + await task + + asyncio.run(_run()) + + assert any( + "still pending after" in record.getMessage() + for record in caplog.records + ) + + def test_stop_warns_when_drain_wait_times_out(self, caplog): + """A failed final drain must be visible instead of silently no-oping.""" + import tools.mcp_tool as mcp_mod + + class TimedOutFuture: + def result(self, timeout): + assert timeout > 0 + raise TimeoutError("simulated drain timeout") + + def cancel(self): + return True + + def report_timeout(coro, _loop, **_kwargs): + coro.close() + return TimedOutFuture() + + with mcp_mod._lock: + mcp_mod._servers.clear() + mcp_mod._server_connecting.clear() + mcp_mod._ensure_mcp_loop() + try: + with caplog.at_level("WARNING", logger=mcp_mod.logger.name): + with patch( + "agent.async_utils.safe_schedule_threadsafe", + side_effect=report_timeout, + ): + mcp_mod._stop_mcp_loop() + finally: + with mcp_mod._lock: + mcp_mod._servers.clear() + mcp_mod._server_connecting.clear() + mcp_mod._stop_mcp_loop() + + assert any( + "Timed out waiting for MCP loop drain" in record.getMessage() + for record in caplog.records + ) diff --git a/tests/tools/test_mcp_tool.py b/tests/tools/test_mcp_tool.py index cc0c3e4089f..35eb365c9ad 100644 --- a/tests/tools/test_mcp_tool.py +++ b/tests/tools/test_mcp_tool.py @@ -1809,6 +1809,7 @@ class TestShutdown: shutdown_started = threading.Event() parked_task_done = threading.Event() scheduled_shutdown = {} + schedule_count = 0 class StalledShutdownServer(MCPServerTask): async def shutdown(self): @@ -1838,7 +1839,12 @@ class TestShutdown: mcp_mod._servers[server.name] = server def schedule_then_report_timeout(coro, target_loop, **_kwargs): + nonlocal schedule_count + schedule_count += 1 future = asyncio.run_coroutine_threadsafe(coro, target_loop) + if schedule_count > 1: + return future + scheduled_shutdown["future"] = future class TimedOutFuture: @@ -1862,6 +1868,7 @@ class TestShutdown: ) assert parked_task.done() assert scheduled_shutdown["future"].done() + assert schedule_count == 2 finally: with mcp_mod._lock: mcp_mod._servers.clear() diff --git a/tools/mcp_tool.py b/tools/mcp_tool.py index bc5a8d7f273..a5eb720a24a 100644 --- a/tools/mcp_tool.py +++ b/tools/mcp_tool.py @@ -363,6 +363,11 @@ def _jittered(seconds: float) -> float: _DEFAULT_KEEPALIVE_INTERVAL = 180 # seconds between liveness pings _MIN_KEEPALIVE_INTERVAL = 5 # clamp floor for configured intervals +# Final shutdown gives pending MCP-loop tasks one bounded cancellation cycle +# before closing their owning loop. Cooperative parked/reconnect waiters finish +# immediately; cancellation-resistant tasks must not hang process exit. +_MCP_LOOP_DRAIN_TIMEOUT = 3.0 + # Environment variables that are safe to pass to stdio subprocesses _SAFE_ENV_KEYS = frozenset({ "PATH", "HOME", "USER", "LANG", "LC_ALL", "TERM", "SHELL", "TMPDIR", @@ -6618,12 +6623,16 @@ def _stop_mcp_loop_if_idle() -> bool: return _stop_mcp_loop(only_if_idle=True) -async def _drain_mcp_loop_tasks() -> None: +async def _drain_mcp_loop_tasks( + *, + timeout: float = _MCP_LOOP_DRAIN_TIMEOUT, +) -> None: """Cancel every task still pending on the MCP loop and reap it. Cancelling is not enough on its own: ``Task.cancel()`` only schedules the - throw, so the task must be awaited before the loop goes away. Gather them - here — while the loop is still open — so each one runs its own ``finally``. + throw, so tasks need a cancellation cycle before the loop goes away. Wait + for them here — on their owning loop — but keep the final drain bounded so + a task that suppresses cancellation cannot hang process exit indefinitely. """ current = asyncio.current_task() pending = [t for t in asyncio.all_tasks() if t is not current and not t.done()] @@ -6632,7 +6641,23 @@ async def _drain_mcp_loop_tasks() -> None: logger.debug("Draining %d pending task(s) from the MCP loop", len(pending)) for task in pending: task.cancel() - await asyncio.gather(*pending, return_exceptions=True) + + done, still_pending = await asyncio.wait(pending, timeout=timeout) + for task in done: + if task.cancelled(): + continue + try: + task.exception() + except asyncio.CancelledError: + pass + except Exception as exc: + logger.debug("Pending MCP loop task ended during shutdown: %s", exc) + + if still_pending: + logger.warning( + "%d MCP loop task(s) still pending after %.1fs drain", + len(still_pending), timeout, + ) def _stop_mcp_loop(*, only_if_idle: bool = False) -> bool: @@ -6656,22 +6681,31 @@ def _stop_mcp_loop(*, only_if_idle: bool = False) -> bool: from agent.async_utils import safe_schedule_threadsafe future = safe_schedule_threadsafe( - _drain_mcp_loop_tasks(), loop, + _drain_mcp_loop_tasks(timeout=_MCP_LOOP_DRAIN_TIMEOUT), loop, logger=logger, log_message="MCP loop drain: failed to schedule", + log_level=logging.WARNING, ) if future is not None: try: - future.result(timeout=5) + future.result(timeout=_MCP_LOOP_DRAIN_TIMEOUT + 1) + except TimeoutError: + future.cancel() + logger.warning( + "Timed out waiting for MCP loop drain after %.1fs", + _MCP_LOOP_DRAIN_TIMEOUT + 1, + ) except BaseException as exc: - logger.debug("Error draining MCP loop tasks: %s", exc) + logger.warning("Error draining MCP loop tasks: %s", exc) loop.call_soon_threadsafe(loop.stop) if thread is not None: thread.join(timeout=5) + if thread.is_alive(): + logger.warning("MCP event loop thread did not stop within 5.0s") try: loop.close() - except Exception: - pass + except Exception as exc: + logger.warning("Unable to close MCP event loop cleanly: %s", exc) # After closing the loop, any stdio subprocesses that survived the # graceful shutdown are now orphaned — include active PIDs too # since the loop is gone and no session can still be in flight. From ab0d3fac3d4cde14333d4107d8b5ae1475399a04 Mon Sep 17 00:00:00 2001 From: Seppe Gadeyne Date: Fri, 17 Jul 2026 10:18:01 +0200 Subject: [PATCH 4/8] fix(mcp): keep drain and stop on loop thread --- tests/tools/test_mcp_stability.py | 81 ++++++++++++++++++++++--------- tools/mcp_tool.py | 31 ++++++++++-- 2 files changed, 86 insertions(+), 26 deletions(-) diff --git a/tests/tools/test_mcp_stability.py b/tests/tools/test_mcp_stability.py index 712a80c04eb..cf43a0d5785 100644 --- a/tests/tools/test_mcp_stability.py +++ b/tests/tools/test_mcp_stability.py @@ -796,20 +796,32 @@ class TestMCPLoopDrainOnStop: assert state["started"], "task never started on the MCP loop" stop_saw_cleanup = [] + call_soon = loop.call_soon call_soon_threadsafe = loop.call_soon_threadsafe - def record_stop_order(callback, *args, **kwargs): + def record_stop_order(schedule, callback, *args, **kwargs): if ( getattr(callback, "__self__", None) is loop and getattr(callback, "__name__", None) == "stop" ): stop_saw_cleanup.append(state["cleanup_ran"]) - return call_soon_threadsafe(callback, *args, **kwargs) + return schedule(callback, *args, **kwargs) - with patch.object( - loop, - "call_soon_threadsafe", - side_effect=record_stop_order, + with ( + patch.object( + loop, + "call_soon", + side_effect=lambda callback, *args, **kwargs: record_stop_order( + call_soon, callback, *args, **kwargs + ), + ), + patch.object( + loop, + "call_soon_threadsafe", + side_effect=lambda callback, *args, **kwargs: record_stop_order( + call_soon_threadsafe, callback, *args, **kwargs + ), + ), ): mcp_mod._stop_mcp_loop() @@ -843,7 +855,8 @@ class TestMCPLoopDrainOnStop: await asyncio.sleep(0) try: with caplog.at_level("WARNING", logger=mcp_mod.logger.name): - await mcp_mod._drain_mcp_loop_tasks(timeout=0.01) + async with asyncio.timeout(0.5): + await mcp_mod._drain_mcp_loop_tasks(timeout=0.01) assert not task.done(), "drain waited indefinitely for resistant task" finally: release.set() @@ -858,34 +871,56 @@ class TestMCPLoopDrainOnStop: for record in caplog.records ) - def test_stop_warns_when_drain_wait_times_out(self, caplog): - """A failed final drain must be visible instead of silently no-oping.""" + def test_outer_timeout_still_allows_loop_owned_drain_before_stop( + self, caplog, monkeypatch + ): + """A blocked loop must drain after it resumes, not stop ahead of the drain.""" + import threading import tools.mcp_tool as mcp_mod - class TimedOutFuture: - def result(self, timeout): - assert timeout > 0 - raise TimeoutError("simulated drain timeout") + parked_started = threading.Event() + cleanup_ran = threading.Event() + blocker_started = threading.Event() + release_blocker = threading.Event() - def cancel(self): - return True + async def parked_task(): + parked_started.set() + try: + await asyncio.Event().wait() + finally: + cleanup_ran.set() - def report_timeout(coro, _loop, **_kwargs): - coro.close() - return TimedOutFuture() + def block_loop(): + blocker_started.set() + release_blocker.wait(timeout=5) with mcp_mod._lock: mcp_mod._servers.clear() mcp_mod._server_connecting.clear() mcp_mod._ensure_mcp_loop() + with mcp_mod._lock: + loop = mcp_mod._mcp_loop + assert loop is not None + + future = asyncio.run_coroutine_threadsafe(parked_task(), loop) + assert parked_started.wait(timeout=2) + loop.call_soon_threadsafe(block_loop) + assert blocker_started.wait(timeout=2) + + monkeypatch.setattr(mcp_mod, "_MCP_LOOP_DRAIN_TIMEOUT", 0.01) + release_timer = threading.Timer(1.2, release_blocker.set) + release_timer.start() try: with caplog.at_level("WARNING", logger=mcp_mod.logger.name): - with patch( - "agent.async_utils.safe_schedule_threadsafe", - side_effect=report_timeout, - ): - mcp_mod._stop_mcp_loop() + mcp_mod._stop_mcp_loop() + + assert cleanup_ran.is_set(), "drain was overtaken by loop.stop" + assert future.done(), "parked task remained pending after loop resumed" + assert loop.is_closed() finally: + release_timer.cancel() + release_blocker.set() + future.cancel() with mcp_mod._lock: mcp_mod._servers.clear() mcp_mod._server_connecting.clear() diff --git a/tools/mcp_tool.py b/tools/mcp_tool.py index a5eb720a24a..dd3f353a906 100644 --- a/tools/mcp_tool.py +++ b/tools/mcp_tool.py @@ -6660,6 +6660,21 @@ async def _drain_mcp_loop_tasks( ) +async def _drain_and_stop_mcp_loop() -> None: + """Drain pending tasks, then stop the loop from its owning thread. + + Keeping both operations in one loop-owned sequence matters when the caller + times out waiting for a blocked loop. Queuing ``loop.stop`` separately from + the caller can overtake the scheduled drain before it receives a loop cycle, + leaving the drain coroutine itself pending when the loop is closed. + """ + loop = asyncio.get_running_loop() + try: + await _drain_mcp_loop_tasks(timeout=_MCP_LOOP_DRAIN_TIMEOUT) + finally: + loop.call_soon(loop.stop) + + def _stop_mcp_loop(*, only_if_idle: bool = False) -> bool: """Stop the background event loop and join its thread.""" global _mcp_loop, _mcp_thread @@ -6677,27 +6692,37 @@ def _stop_mcp_loop(*, only_if_idle: bool = False) -> bool: # to run cleanup against a loop that is already closed -> "Event loop # is closed" (#60197). ``shutdown_mcp_servers`` only reaps servers held # in ``_servers``, so anything else left on this loop ends up here. + stop_owned_by_loop = False if loop.is_running(): from agent.async_utils import safe_schedule_threadsafe future = safe_schedule_threadsafe( - _drain_mcp_loop_tasks(timeout=_MCP_LOOP_DRAIN_TIMEOUT), loop, + _drain_and_stop_mcp_loop(), loop, logger=logger, log_message="MCP loop drain: failed to schedule", log_level=logging.WARNING, ) if future is not None: + stop_owned_by_loop = True try: future.result(timeout=_MCP_LOOP_DRAIN_TIMEOUT + 1) except TimeoutError: - future.cancel() logger.warning( "Timed out waiting for MCP loop drain after %.1fs", _MCP_LOOP_DRAIN_TIMEOUT + 1, ) except BaseException as exc: logger.warning("Error draining MCP loop tasks: %s", exc) - loop.call_soon_threadsafe(loop.stop) + elif not loop.is_closed(): + try: + loop.run_until_complete( + _drain_mcp_loop_tasks(timeout=_MCP_LOOP_DRAIN_TIMEOUT) + ) + except BaseException as exc: + logger.warning("Error draining stopped MCP loop tasks: %s", exc) + + if not stop_owned_by_loop and loop.is_running(): + loop.call_soon_threadsafe(loop.stop) if thread is not None: thread.join(timeout=5) if thread.is_alive(): From fd5397d69afc15dc0af4e8768017d23759587d8d Mon Sep 17 00:00:00 2001 From: Seppe Gadeyne Date: Fri, 17 Jul 2026 10:33:56 +0200 Subject: [PATCH 5/8] chore(release): map shady2k contributor email --- scripts/release.py | 1 + 1 file changed, 1 insertion(+) diff --git a/scripts/release.py b/scripts/release.py index 0df1a1b70d5..0a6c65c08f0 100755 --- a/scripts/release.py +++ b/scripts/release.py @@ -47,6 +47,7 @@ LEGACY_AUTHOR_MAP = { "declanbatesmith@outlook.com": "cat-thats-fat", # PR #60489 (desktop: first-run remote connection option) "drbs2004@me.com": "cat-thats-fat", # PR #60489 (desktop: first-run remote connection option; historical merge email) "122438640+ragingbulld@users.noreply.github.com": "ragingbulld", # PR #65606 salvage (non-finite API wait deadlines; #65746) + "shady2k@gmail.com": "shady2k", # PR #60104 salvage of #66143 (MCP loop-owned shutdown drain) "zzpigpinggai@users.noreply.github.com": "zzpigpinggai", # PR #66017 salvage of #63617 (OpenRouter explicit-provider picker visibility) "stellarisw@users.noreply.github.com": "StellarisW", # PR #66222 salvage (Discord WebSocket liveness + systemd watchdog; #26656 follow-up) "wx.xw@bytedance.com": "wxy-nlp", # PR #66222 salvage (systemd event-loop watchdog co-author) From c00a1d58d505123a64fef7e2f0500f6e483e8193 Mon Sep 17 00:00:00 2001 From: Stepan Zadolia Date: Fri, 10 Jul 2026 13:21:39 +0300 Subject: [PATCH 6/8] fix(mcp): retain parked startup tasks for clean shutdown --- .../test_mcp_initial_connect_shutdown.py | 273 ++++++++++++++++++ tools/mcp_tool.py | 101 ++++++- 2 files changed, 362 insertions(+), 12 deletions(-) create mode 100644 tests/tools/test_mcp_initial_connect_shutdown.py diff --git a/tests/tools/test_mcp_initial_connect_shutdown.py b/tests/tools/test_mcp_initial_connect_shutdown.py new file mode 100644 index 00000000000..049e319bef6 --- /dev/null +++ b/tests/tools/test_mcp_initial_connect_shutdown.py @@ -0,0 +1,273 @@ +"""Regression tests for initial MCP failure ownership and teardown.""" + +import asyncio +import json +import threading +from types import SimpleNamespace + +import pytest + + +def _reset_mcp_state(mcp_tool) -> None: + mcp_tool.shutdown_mcp_servers() + with mcp_tool._lock: + mcp_tool._servers.clear() + mcp_tool._server_connecting.clear() + mcp_tool._server_connect_errors.clear() + + +def _cleanup_mcp_state(mcp_tool, extra_servers=()) -> None: + with mcp_tool._lock: + loop = mcp_tool._mcp_loop + if loop is not None and loop.is_running(): + for server in extra_servers: + task = getattr(server, "_task", None) + if task is not None and not task.done(): + mcp_tool._run_on_mcp_loop(server.shutdown, timeout=5) + mcp_tool.shutdown_mcp_servers() + with mcp_tool._lock: + mcp_tool._servers.clear() + mcp_tool._server_connecting.clear() + mcp_tool._server_connect_errors.clear() + + +def test_initial_connect_failure_is_registry_owned_and_reaped(monkeypatch, tmp_path): + """Normal discovery must retain the parked task for clean shutdown.""" + monkeypatch.setenv("HERMES_HOME", str(tmp_path)) + + from tools import mcp_tool + + _reset_mcp_state(mcp_tool) + created = [] + + class _FailingServerTask(mcp_tool.MCPServerTask): + def __init__(self, name): + super().__init__(name) + created.append(self) + + async def _run_stdio(self, config): + raise ConnectionError("deterministic initial failure") + + monkeypatch.setattr(mcp_tool, "MCPServerTask", _FailingServerTask) + monkeypatch.setattr(mcp_tool, "_MCP_AVAILABLE", True) + monkeypatch.setattr(mcp_tool, "_MAX_INITIAL_CONNECT_RETRIES", 0) + monkeypatch.setattr(mcp_tool, "_PARKED_RETRY_INTERVAL", 3600) + + real_stop = mcp_tool._stop_mcp_loop + pending_at_stop = [] + + async def _pending_tasks(): + current = asyncio.current_task() + return sorted( + task.get_coro().__qualname__ + for task in asyncio.all_tasks() + if task is not current and not task.done() + ) + + def _observed_stop(*, only_if_idle=False): + pending_at_stop.extend( + mcp_tool._run_on_mcp_loop(_pending_tasks, timeout=5) + ) + return real_stop(only_if_idle=only_if_idle) + + monkeypatch.setattr(mcp_tool, "_stop_mcp_loop", _observed_stop) + + try: + assert mcp_tool.register_mcp_servers({ + "initial-failure": {"command": "unused", "connect_timeout": 5} + }) == [] + + assert len(created) == 1 + server = created[0] + with mcp_tool._lock: + assert mcp_tool._servers["initial-failure"] is server + assert "deterministic initial failure" in ( + mcp_tool._server_connect_errors["initial-failure"] + ) + assert server._task is not None + assert not server._task.done(), "recoverable initial failure was not parked" + + mcp_tool.shutdown_mcp_servers() + + assert pending_at_stop == [], ( + "shutdown left MCP tasks pending at loop stop: " + f"{pending_at_stop!r}" + ) + assert server._task.done() + with mcp_tool._lock: + assert mcp_tool._mcp_loop is None + assert mcp_tool._mcp_thread is None + finally: + monkeypatch.setattr(mcp_tool, "_stop_mcp_loop", real_stop) + _cleanup_mcp_state(mcp_tool, created) + + +def test_initial_connect_failure_revives_same_registered_server(monkeypatch, tmp_path): + """A cached parked failure must revive through register_mcp_servers().""" + monkeypatch.setenv("HERMES_HOME", str(tmp_path)) + + from tools import mcp_tool + from tools.registry import ToolRegistry + import tools.registry as registry_module + + _reset_mcp_state(mcp_tool) + created = [] + backend_up = threading.Event() + revived = threading.Event() + state = {"transport_calls": 0, "tool_calls": 0} + mock_registry = ToolRegistry() + + class _Session: + async def call_tool(self, name, arguments): + state["tool_calls"] += 1 + return SimpleNamespace( + isError=False, + content=[SimpleNamespace(text=f"revived:{arguments['value']}")], + structuredContent=None, + ) + + class _RecoveringServerTask(mcp_tool.MCPServerTask): + def __init__(self, name): + super().__init__(name) + created.append(self) + + async def _run_stdio(self, config): + assert mcp_tool._connect_server_claim.get() is None + state["transport_calls"] += 1 + if not backend_up.is_set(): + raise ConnectionError("backend still booting") + + self.session = _Session() + self._tools = [SimpleNamespace( + name="ping", + description="Return a deterministic revival result", + inputSchema={ + "type": "object", + "properties": {"value": {"type": "string"}}, + "required": ["value"], + }, + )] + # Match the real transports: discovery runs before _ready is set. + self._register_discovered_tools_if_needed() + self._ready.set() + revived.set() + return await self._wait_for_lifecycle_event() + + monkeypatch.setattr(mcp_tool, "MCPServerTask", _RecoveringServerTask) + monkeypatch.setattr(mcp_tool, "_MCP_AVAILABLE", True) + monkeypatch.setattr(mcp_tool, "_MAX_INITIAL_CONNECT_RETRIES", 0) + monkeypatch.setattr(mcp_tool, "_PARKED_RETRY_INTERVAL", 3600) + monkeypatch.setattr(registry_module, "registry", mock_registry) + + config = { + "recovering": {"command": "unused", "connect_timeout": 5} + } + + try: + assert mcp_tool.register_mcp_servers(config) == [] + assert len(created) == 1 + server = created[0] + with mcp_tool._lock: + assert mcp_tool._servers["recovering"] is server + assert "backend still booting" in ( + mcp_tool._server_connect_errors["recovering"] + ) + assert not server._task.done() + + backend_up.set() + mcp_tool.register_mcp_servers(config) + + assert revived.wait(timeout=5), "cached parked server did not revive" + assert len(created) == 1, "revival created a duplicate server task" + with mcp_tool._lock: + assert mcp_tool._servers["recovering"] is server + assert "recovering" not in mcp_tool._server_connect_errors + assert state["transport_calls"] == 2 + assert server.session is not None + assert server._error is None + + entry = mock_registry.get_entry("mcp__recovering__ping") + assert entry is not None + assert entry.check_fn() is True + assert json.loads(entry.handler({"value": "ok"})) == { + "result": "revived:ok" + } + assert state["tool_calls"] == 1 + finally: + _cleanup_mcp_state(mcp_tool, created) + + +def test_terminal_initial_failure_is_not_retained(monkeypatch, tmp_path): + """A non-recoverable startup error must not leave a dead cache entry.""" + monkeypatch.setenv("HERMES_HOME", str(tmp_path)) + + from tools import mcp_tool + + _reset_mcp_state(mcp_tool) + created = [] + + class _AuthFailingServerTask(mcp_tool.MCPServerTask): + def __init__(self, name): + super().__init__(name) + created.append(self) + + async def _run_stdio(self, config): + raise PermissionError("terminal authentication failure") + + monkeypatch.setattr(mcp_tool, "MCPServerTask", _AuthFailingServerTask) + monkeypatch.setattr(mcp_tool, "_MCP_AVAILABLE", True) + monkeypatch.setattr(mcp_tool, "_is_auth_error", lambda exc: True) + + try: + assert mcp_tool.register_mcp_servers({ + "auth-failure": {"command": "unused", "connect_timeout": 5} + }) == [] + assert len(created) == 1 + assert created[0]._task.done() + with mcp_tool._lock: + assert "auth-failure" not in mcp_tool._servers + assert "terminal authentication failure" in ( + mcp_tool._server_connect_errors["auth-failure"] + ) + finally: + _cleanup_mcp_state(mcp_tool, created) + + +def test_standalone_failed_connect_is_reaped_without_global_owner(monkeypatch, tmp_path): + """Probe-only _connect_server failures must not publish parked servers.""" + monkeypatch.setenv("HERMES_HOME", str(tmp_path)) + + from tools import mcp_tool + + _reset_mcp_state(mcp_tool) + created = [] + + class _ProbeServerTask(mcp_tool.MCPServerTask): + def __init__(self, name): + super().__init__(name) + created.append(self) + + async def _run_stdio(self, config): + raise ConnectionError("probe target unavailable") + + monkeypatch.setattr(mcp_tool, "MCPServerTask", _ProbeServerTask) + monkeypatch.setattr(mcp_tool, "_MAX_INITIAL_CONNECT_RETRIES", 0) + monkeypatch.setattr(mcp_tool, "_PARKED_RETRY_INTERVAL", 3600) + mcp_tool._ensure_mcp_loop() + + try: + with pytest.raises(ConnectionError, match="probe target unavailable"): + mcp_tool._run_on_mcp_loop( + lambda: mcp_tool._connect_server( + "probe-only", {"command": "unused"} + ), + timeout=5, + ) + + assert len(created) == 1 + assert created[0]._task.done() + with mcp_tool._lock: + assert "probe-only" not in mcp_tool._servers + assert "probe-only" not in mcp_tool._server_connect_errors + finally: + _cleanup_mcp_state(mcp_tool, created) diff --git a/tools/mcp_tool.py b/tools/mcp_tool.py index dd3f353a906..9895da2bc6f 100644 --- a/tools/mcp_tool.py +++ b/tools/mcp_tool.py @@ -3021,7 +3021,7 @@ class MCPServerTask: self._register_discovered_tools_if_needed() def _register_discovered_tools_if_needed(self) -> None: - """Re-register tools after a post-ready reconnect if needed. + """Re-register tools after an owned server reconnects if needed. Initial registration is performed by ``_discover_and_register_server`` after ``start()`` completes. During a later reconnect, outage handling @@ -3029,6 +3029,9 @@ class MCPServerTask: A managed server can still be identified by its entry in ``_servers``; publish its freshly discovered tools before transport readiness is restored so a successful revival cannot come back with zero tools. + A server retained after a recoverable initial failure is likewise + registry-owned before its first successful session, so ownership also + authorizes its first publication. """ if self._registered_tool_names: return @@ -3039,6 +3042,12 @@ class MCPServerTask: self._registered_tool_names = _register_server_tools( self.name, self, self._config ) + # A retained initial-failure server that just published tools has + # recovered: drop its stale connect error so status surfaces stop + # reporting it as failed. + with _lock: + if _servers.get(self.name) is self: + _server_connect_errors.pop(self.name, None) async def run(self, config: dict): """Long-lived coroutine: connect, discover tools, wait, disconnect. @@ -3523,6 +3532,12 @@ class MCPServerTask: _servers: Dict[str, MCPServerTask] = {} _server_connecting: set[str] = set() _server_connect_errors: Dict[str, str] = {} +# Discovery installs a task-local claim before calling ``_connect_server`` so +# it can retain a recoverable parked task without making standalone probe calls +# publish failed servers into module-global ownership. +_connect_server_claim: contextvars.ContextVar[ + Optional[Callable[[MCPServerTask], None]] +] = contextvars.ContextVar("mcp_connect_server_claim", default=None) # Connection-retry cooldown (per-server isolation against restart storms). # @@ -3632,9 +3647,8 @@ def _signal_reconnect(server: Any) -> bool: The tool handlers run on caller threads, while the server task and its ``_reconnect_event`` live on the background MCP loop. Setting an asyncio.Event from another thread must go through - ``loop.call_soon_threadsafe``; only fall back to a direct ``.set()`` - when the loop isn't running (e.g. unit tests that drive the handler - synchronously). + ``loop.call_soon_threadsafe``; non-async adapters and tests without a + running loop can use a direct ``.set()``. Returns True if a reconnect signal was delivered, False if the server has no reconnect machinery (nothing to revive). @@ -3643,7 +3657,11 @@ def _signal_reconnect(server: Any) -> bool: if event is None: return False loop = _mcp_loop - if loop is not None and loop.is_running(): + if ( + isinstance(event, asyncio.Event) + and loop is not None + and loop.is_running() + ): loop.call_soon_threadsafe(event.set) else: event.set() @@ -4616,7 +4634,26 @@ async def _connect_server(name: str, config: dict) -> MCPServerTask: Exception: on connection or initialization failure. """ server = MCPServerTask(name) - await server.start(config) + claim = _connect_server_claim.get() + claim_token = None + if claim is not None: + claim(server) + # ``start()`` creates the long-lived run task by copying this context. + # The ownership callback is only for this connection attempt; do not + # retain its discovery closure for the server's lifetime. + claim_token = _connect_server_claim.set(None) + try: + await server.start(config) + except BaseException: + # Discovery owns claimed tasks and decides whether a failed start is a + # live recoverable park or a terminal failure. Standalone probes have + # no revival owner, so they must reap their failed task locally. + if claim is None: + await server.shutdown() + raise + finally: + if claim_token is not None: + _connect_server_claim.reset(claim_token) return server @@ -5788,10 +5825,42 @@ async def _discover_and_register_server(name: str, config: dict) -> List[str]: Returns list of registered tool names. """ connect_timeout = config.get("connect_timeout", _DEFAULT_CONNECT_TIMEOUT) - server = await asyncio.wait_for( - _connect_server(name, config), - timeout=connect_timeout, - ) + server: Optional[MCPServerTask] = None + + def _claim_server(created: MCPServerTask) -> None: + nonlocal server + server = created + + claim_token = _connect_server_claim.set(_claim_server) + try: + server = await asyncio.wait_for( + _connect_server(name, config), + timeout=connect_timeout, + ) + except BaseException: + task = server._task if server is not None else None + task_cancelling = ( + task.cancelling() + if task is not None and hasattr(task, "cancelling") + else 0 + ) + recoverable_park = ( + server is not None + and server._error is not None + and task is not None + and not task.done() + and not task_cancelling + ) + if recoverable_park: + with _lock: + _servers[name] = server + elif server is not None: + await server.shutdown() + raise + finally: + _connect_server_claim.reset(claim_token) + + assert server is not None with _lock: _server_connecting.discard(name) _server_connect_errors.pop(name, None) @@ -5932,7 +6001,11 @@ def register_mcp_servers(servers: Dict[str, dict]) -> List[str]: # Log a summary so ACP callers get visibility into what was registered. with _lock: - connected = [n for n in new_servers if n in _servers] + connected = [ + n + for n in new_servers + if n in _servers and n not in _server_connect_errors + ] new_tool_count = sum( len(getattr(_servers[n], "_registered_tool_names", [])) for n in connected @@ -6005,7 +6078,11 @@ def discover_mcp_tools() -> List[str]: return tool_names with _lock: - connected_server_names = [name for name in new_server_names if name in _servers] + connected_server_names = [ + name + for name in new_server_names + if name in _servers and name not in _server_connect_errors + ] new_tool_count = sum( len(getattr(_servers[name], "_registered_tool_names", [])) for name in connected_server_names From 1f70ba6bca71caa2c440607b06b74e596969e1b1 Mon Sep 17 00:00:00 2001 From: kshitijk4poor <82637225+kshitijk4poor@users.noreply.github.com> Date: Wed, 29 Jul 2026 17:11:15 +0500 Subject: [PATCH 7/8] fix(mcp): propagate cancellation untouched in _connect_server orphan reap Follow-up to the salvaged #62026 ownership fix, folding in #72054's CancelledError rule by @adurham: start() already cancels/reaps its own run task when the caller's connect timeout cancels start() itself, so _connect_server() must propagate cancellation without awaiting a redundant shutdown() inside a cancelled context. Non-cancellation failures on the unclaimed (standalone probe) path still reap the parked task, now with the reap failure logged instead of raising over the real error. Also maps mrz@mrzlab630.pw for the attribution check. Co-authored-by: Adam Durham --- contributors/emails/amdnative@gmail.com | 1 + contributors/emails/mrz@mrzlab630.pw | 1 + tools/mcp_tool.py | 30 ++++++++++++++++++------- 3 files changed, 24 insertions(+), 8 deletions(-) create mode 100644 contributors/emails/amdnative@gmail.com create mode 100644 contributors/emails/mrz@mrzlab630.pw diff --git a/contributors/emails/amdnative@gmail.com b/contributors/emails/amdnative@gmail.com new file mode 100644 index 00000000000..269ad2f1eb6 --- /dev/null +++ b/contributors/emails/amdnative@gmail.com @@ -0,0 +1 @@ +adurham diff --git a/contributors/emails/mrz@mrzlab630.pw b/contributors/emails/mrz@mrzlab630.pw new file mode 100644 index 00000000000..93523b9e3cb --- /dev/null +++ b/contributors/emails/mrz@mrzlab630.pw @@ -0,0 +1 @@ +mrzlab630 diff --git a/tools/mcp_tool.py b/tools/mcp_tool.py index 9895da2bc6f..fe5ab9cfdfa 100644 --- a/tools/mcp_tool.py +++ b/tools/mcp_tool.py @@ -4644,12 +4644,23 @@ async def _connect_server(name: str, config: dict) -> MCPServerTask: claim_token = _connect_server_claim.set(None) try: await server.start(config) + except asyncio.CancelledError: + # start() already cancels/reaps server._task on external cancellation + # (see the comment there) -- awaiting a redundant shutdown() inside a + # cancelled context would only risk swallowing the cancellation. + raise except BaseException: # Discovery owns claimed tasks and decides whether a failed start is a # live recoverable park or a terminal failure. Standalone probes have # no revival owner, so they must reap their failed task locally. if claim is None: - await server.shutdown() + try: + await server.shutdown() + except Exception as shutdown_exc: # noqa: BLE001 -- best-effort reap, don't mask the real error + logger.debug( + "MCP server '%s' shutdown during orphan-reap failed: %s", + name, shutdown_exc, + ) raise finally: if claim_token is not None: @@ -5825,11 +5836,13 @@ async def _discover_and_register_server(name: str, config: dict) -> List[str]: Returns list of registered tool names. """ connect_timeout = config.get("connect_timeout", _DEFAULT_CONNECT_TIMEOUT) - server: Optional[MCPServerTask] = None + # List-based claim (not a ``nonlocal`` rebind): the claim callback runs + # inside ``_connect_server`` while this frame is suspended, and appending + # keeps type narrowing intact for the module's other ``server`` locals. + claimed: List[MCPServerTask] = [] def _claim_server(created: MCPServerTask) -> None: - nonlocal server - server = created + claimed.append(created) claim_token = _connect_server_claim.set(_claim_server) try: @@ -5838,20 +5851,22 @@ async def _discover_and_register_server(name: str, config: dict) -> List[str]: timeout=connect_timeout, ) except BaseException: + server = claimed[0] if claimed else None task = server._task if server is not None else None task_cancelling = ( task.cancelling() if task is not None and hasattr(task, "cancelling") else 0 ) - recoverable_park = ( + if ( server is not None and server._error is not None and task is not None and not task.done() and not task_cancelling - ) - if recoverable_park: + ): + # Recoverable park: the run task deliberately stays alive to + # self-probe, so adopt it into the registry for shutdown/revival. with _lock: _servers[name] = server elif server is not None: @@ -5860,7 +5875,6 @@ async def _discover_and_register_server(name: str, config: dict) -> List[str]: finally: _connect_server_claim.reset(claim_token) - assert server is not None with _lock: _server_connecting.discard(name) _server_connect_errors.pop(name, None) From 28524adb0ef62145e8e31d3f2c2f749765e3e5ca Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Wed, 29 Jul 2026 12:18:06 -0700 Subject: [PATCH 8/8] =?UTF-8?q?fix(tests):=20eliminate=20flaky/broken=20te?= =?UTF-8?q?sts=20=E2=80=94=20shadow=20sys.path=20inserts,=20unmocked=20net?= =?UTF-8?q?work=20in=20compressor=20tests,=20stale-SDK=20feishu=20pin=20gu?= =?UTF-8?q?ard,=20quadratic=20redact=20regexes?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Remove tests/-shadowing sys.path.insert(dirname/'..') from 11 test files: it prepended the tests/ dir itself to sys.path, so 'import agent' / 'import hermes_cli' resolved to the test packages and collection died with ModuleNotFoundError depending on import order (2 files failed in every full-suite run; 9 more were latent). - Patch call_llm in 5 context-compressor tests that called compress() unmocked: each burned ~50s attempting live LLM traffic through the relay before falling back (572s file — the slowest in the suite, and flaky under the 300s per-file timeout). File now runs in ~5s. - agent/redact.py: fix two catastrophically-backtracking regexes hit by the compressor's redaction pass on large payloads — _STRICT_URL_USERINFO_RE anchors on the mandatory '//' (optional-scheme prefix backtracked O(n^2): ~55s on a 320KB payload, now sub-ms; output-equivalence fuzz-verified on 20k random strings), and the _CFG_DOTTED_RE/_CFG_ANCHORED_RE subs gain an exact linear keyword pre-gate so secret-free text skips the quadratic pattern entirely. - tests/gateway/test_feishu.py: version-guard the extra_ua_tags SDK signature check; the repo pins lark-oapi==1.6.8 but stale local installs (1.5.3) fail the assertion — skip below the pin. - tests/tools/test_managed_browserbase_and_modal.py: stub agent.redact + agent.credential_persistence in the fake agent package (empty __path__ blocks all real agent.* imports added since the fake was written). - tests/gateway/test_startup_restart_race.py: raise wait_for timeouts 2s -> 30s; 2s wall-clock on a loaded 40-worker box flaked in the baseline run (passes instantly when the box is quiet). --- agent/redact.py | 26 +++++++++++++++++-- tests/agent/test_context_compressor.py | 25 +++++++++++------- tests/agent/test_model_metadata_local_ctx.py | 3 --- tests/agent/test_model_metadata_ssl.py | 3 --- tests/agent/test_probe_cache_followups.py | 3 --- tests/cli/test_cli_init.py | 1 - tests/cli/test_cli_user_message_preview.py | 2 -- tests/cli/test_resume_display.py | 3 --- tests/cli/test_tool_progress_scrollback.py | 2 -- tests/gateway/test_feishu.py | 19 +++++++++++++- tests/gateway/test_startup_restart_race.py | 8 +++--- tests/hermes_cli/test_auth_ssl_macos.py | 2 -- tests/hermes_cli/test_voice_wrapper.py | 3 --- tests/test_ctx_halving_fix.py | 3 --- tests/test_model_picker_scroll.py | 3 --- .../test_managed_browserbase_and_modal.py | 11 ++++++++ 16 files changed, 73 insertions(+), 44 deletions(-) diff --git a/agent/redact.py b/agent/redact.py index bff23934da6..b104ad4c22f 100644 --- a/agent/redact.py +++ b/agent/redact.py @@ -140,6 +140,11 @@ _ENV_ASSIGN_RE = re.compile( # The colon-form URL guard (skip when ``://`` present) lives at the call site. _SECRET_CFG_NAMES = r"(?:api[ _.\-]?key|token|secret|passwd|password|credential|auth)" _CFG_VALUE = r"(['\"]?)([^\s&]+?)\2(?=[\s&]|$)" +# Linear pre-gate for the _CFG_*_RE subs below: a text with no secret keyword +# can never match either pattern, so the (potentially backtrack-heavy) subs +# are skipped entirely for such text. See the call site in +# redact_sensitive_text(). +_CFG_SECRET_WORD_RE = re.compile(_SECRET_CFG_NAMES, re.IGNORECASE) # Programmatic env lookups (``os.getenv(...)``, ``os.environ[...]``, # ``os.environ.get(...)``, ``process.env.X``, ``$ENV{X}``) reference variable @@ -377,8 +382,17 @@ _STRICT_URL_PARAM_RE = re.compile( # Match userinfo in both absolute (``scheme://user:pass@host``) and # network-path (``//user:pass@host``) references. The authority boundary stops # at path/query/fragment delimiters so an ``@`` elsewhere in a URL is ignored. +# +# Anchored on the mandatory ``//`` rather than an optional scheme prefix: the +# scheme sits outside the match either way (replacement callbacks re-emit +# group(1), so ``https:`` stays untouched in the surrounding text), and the +# old optional-scheme prefix ``(?:[A-Za-z][A-Za-z0-9+.-]*:)?`` backtracked +# catastrophically (O(n²)) on long unbroken alphanumeric runs — a 320KB +# synthetic compaction payload spent ~55s inside this pattern per sub() call. +# Output-equivalence to the old pattern was fuzz-verified (20k random strings +# plus targeted URL forms). _STRICT_URL_USERINFO_RE = re.compile( - r"((?:[A-Za-z][A-Za-z0-9+.-]*:)?//)([^/\s?#@]+)@" + r"(//)([^/\s?#@]+)@" ) # HTTP access logs often use a relative request target rather than a full URL: @@ -706,7 +720,15 @@ def redact_sensitive_text( # web-URL query params are intentionally passed through (see note # near the bottom of this function); _DB_CONNSTR_RE still guards # connection-string passwords. - if "://" not in text: + # + # Extra gate: every _CFG_*_RE match requires a secret keyword in + # the key, so a text without any secret keyword cannot match — + # skipping is exact. This matters because _CFG_DOTTED_RE + # backtracks quadratically on long unbroken [A-Za-z0-9_.\-] runs + # (e.g. base64/hex blobs in compaction payloads); the linear + # keyword scan prevents that pathological path on secret-free + # text. + if "://" not in text and _CFG_SECRET_WORD_RE.search(text): text = _CFG_DOTTED_RE.sub(_redact_env, text) text = _CFG_ANCHORED_RE.sub(_redact_env, text) diff --git a/tests/agent/test_context_compressor.py b/tests/agent/test_context_compressor.py index 88c694515d1..3d08ad46932 100644 --- a/tests/agent/test_context_compressor.py +++ b/tests/agent/test_context_compressor.py @@ -174,7 +174,8 @@ class TestCompress: def test_too_few_messages_returns_unchanged(self, compressor): msgs = self._make_messages(4) # protect_first=2 + protect_last=2 + 1 = 5 needed - result = compressor.compress(msgs) + with patch("agent.context_compressor.call_llm", side_effect=RuntimeError("no provider")): + result = compressor.compress(msgs) assert result == msgs def test_truncation_fallback_no_client(self, compressor): @@ -380,15 +381,19 @@ class TestCompress: def test_compression_increments_count(self, compressor): msgs = self._make_messages(10) # Default config (abort_on_summary_failure=False) — fallback path - # increments the count even on summary failure. - compressor.compress(msgs) - assert compressor.compression_count == 1 - compressor.compress(msgs) - assert compressor.compression_count == 2 + # increments the count even on summary failure. Patch call_llm so the + # summary attempt fails fast instead of attempting live network I/O + # (unpatched, each compress() call burned ~50s in connect/retry). + with patch("agent.context_compressor.call_llm", side_effect=RuntimeError("no provider")): + compressor.compress(msgs) + assert compressor.compression_count == 1 + compressor.compress(msgs) + assert compressor.compression_count == 2 def test_protects_first_and_last(self, compressor): msgs = self._make_messages(10) - result = compressor.compress(msgs) + with patch("agent.context_compressor.call_llm", side_effect=RuntimeError("no provider")): + result = compressor.compress(msgs) # First 2 messages should be preserved (protect_first_n=2) # Last 2 messages should be preserved (protect_last_n=2) assert result[-1]["content"] == msgs[-1]["content"] @@ -607,7 +612,8 @@ class TestGenerateSummaryNoneContent: {"role": "user" if i % 2 == 0 else "assistant", "content": f"msg {i}"} for i in range(10) ] - result = c.compress(msgs) + with patch("agent.context_compressor.call_llm", side_effect=RuntimeError("no provider")): + result = c.compress(msgs) assert len(result) < len(msgs) @@ -2682,7 +2688,8 @@ class TestSummaryTargetRatio: + [{"role": "user" if i % 2 == 0 else "assistant", "content": f"msg {i}"} for i in range(8)] ) - result = c.compress(msgs) + with patch("agent.context_compressor.call_llm", side_effect=RuntimeError("no provider")): + result = c.compress(msgs) # System prompt (msg[0]) survives as head assert result[0]["role"] == "system" assert result[0]["content"].startswith("System prompt") diff --git a/tests/agent/test_model_metadata_local_ctx.py b/tests/agent/test_model_metadata_local_ctx.py index 1fcc24ecf82..a1fa3c40dc1 100644 --- a/tests/agent/test_model_metadata_local_ctx.py +++ b/tests/agent/test_model_metadata_local_ctx.py @@ -4,13 +4,10 @@ get_model_context_length. All tests use synthetic inputs — no filesystem or live server required. """ -import sys -import os from unittest.mock import MagicMock, patch import pytest -sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..")) @pytest.fixture(autouse=True) diff --git a/tests/agent/test_model_metadata_ssl.py b/tests/agent/test_model_metadata_ssl.py index 6859fd309cd..f54bd9a7777 100644 --- a/tests/agent/test_model_metadata_ssl.py +++ b/tests/agent/test_model_metadata_ssl.py @@ -9,11 +9,8 @@ No filesystem or network I/O required — we use tmp_path to create real CA bundle stand-in files and monkeypatch env vars. """ -import os -import sys from pathlib import Path -sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..")) import pytest diff --git a/tests/agent/test_probe_cache_followups.py b/tests/agent/test_probe_cache_followups.py index 7d87516a407..b6659db86a6 100644 --- a/tests/agent/test_probe_cache_followups.py +++ b/tests/agent/test_probe_cache_followups.py @@ -7,13 +7,10 @@ Covers: from __future__ import annotations -import os -import sys from unittest.mock import MagicMock, patch import pytest -sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..")) @pytest.fixture(autouse=True) diff --git a/tests/cli/test_cli_init.py b/tests/cli/test_cli_init.py index c3915777da0..0f50ff6da22 100644 --- a/tests/cli/test_cli_init.py +++ b/tests/cli/test_cli_init.py @@ -6,7 +6,6 @@ import sys from types import SimpleNamespace from unittest.mock import MagicMock, patch -sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..")) def _make_cli(env_overrides=None, config_overrides=None, **kwargs): diff --git a/tests/cli/test_cli_user_message_preview.py b/tests/cli/test_cli_user_message_preview.py index f3e84759eed..b3829587d06 100644 --- a/tests/cli/test_cli_user_message_preview.py +++ b/tests/cli/test_cli_user_message_preview.py @@ -1,9 +1,7 @@ import importlib -import os import sys from unittest.mock import MagicMock, patch -sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..")) _cli_mod = None diff --git a/tests/cli/test_resume_display.py b/tests/cli/test_resume_display.py index 6f36ace71ae..8a314746fda 100644 --- a/tests/cli/test_resume_display.py +++ b/tests/cli/test_resume_display.py @@ -5,14 +5,11 @@ Verifies that resuming a session shows a compact recap of the previous conversation with correct formatting, truncation, and config behavior. """ -import os -import sys from io import StringIO from unittest.mock import MagicMock, patch import cli as cli_mod -sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..")) def _make_cli(config_overrides=None, env_overrides=None, **kwargs): diff --git a/tests/cli/test_tool_progress_scrollback.py b/tests/cli/test_tool_progress_scrollback.py index 2f5f6a8952d..00033bb4b11 100644 --- a/tests/cli/test_tool_progress_scrollback.py +++ b/tests/cli/test_tool_progress_scrollback.py @@ -5,12 +5,10 @@ persistent lines to scrollback on tool.completed, restoring the stacked tool history that was lost when the TUI switched to a single-line spinner. """ -import os import sys import importlib from unittest.mock import MagicMock, patch -sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..")) # Module-level reference to the cli module (set by _make_cli on first call) _cli_mod = None diff --git a/tests/gateway/test_feishu.py b/tests/gateway/test_feishu.py index d887bc94f38..480b6a65dce 100644 --- a/tests/gateway/test_feishu.py +++ b/tests/gateway/test_feishu.py @@ -176,9 +176,26 @@ class TestFeishuMessageNormalization(unittest.TestCase): class TestFeishuAdapterMessaging(unittest.TestCase): @unittest.skipUnless(_HAS_LARK_OAPI, "lark-oapi not installed") def test_websocket_sdk_accepts_channel_ua_tag(self): - """The shipped SDK must support the Channel signaling argument.""" + """The shipped SDK must support the Channel signaling argument. + + Guarded on the pinned version: the repo pins lark-oapi==1.6.8 (the + first release with ``extra_ua_tags``). Dev machines can carry an + older lazy-installed lark-oapi that predates the argument — that is + an environment artifact, not a product regression, so skip rather + than fail there. Environments installing the pin (the feishu extra) + still exercise the real assertion. + """ import inspect + from importlib.metadata import version as _pkg_version + + installed = tuple(int(p) for p in _pkg_version("lark-oapi").split(".")[:3] if p.isdigit()) + if installed < (1, 6, 8): + self.skipTest( + f"lark-oapi {_pkg_version('lark-oapi')} predates extra_ua_tags; " + "repo pin is 1.6.8 — stale local install" + ) + from lark_oapi.ws import Client as FeishuWSClient signature = inspect.signature(FeishuWSClient) diff --git a/tests/gateway/test_startup_restart_race.py b/tests/gateway/test_startup_restart_race.py index bdf5bbc42a4..3c1666ac830 100644 --- a/tests/gateway/test_startup_restart_race.py +++ b/tests/gateway/test_startup_restart_race.py @@ -136,7 +136,7 @@ async def test_startup_aborts_when_restart_requested_before_start(tmp_path, monk runner.request_restart(detached=False, via_service=True) runner._create_adapter = MagicMock() - result = await asyncio.wait_for(runner.start(), timeout=2) + result = await asyncio.wait_for(runner.start(), timeout=30) assert result is True runner._create_adapter.assert_not_called() @@ -167,7 +167,7 @@ async def test_startup_aborts_when_restart_begins_during_platform_connect(tmp_pa telegram.disconnect = disconnect_and_release runner._create_adapter = MagicMock(side_effect=[telegram, slack]) - result = await asyncio.wait_for(runner.start(), timeout=2) + result = await asyncio.wait_for(runner.start(), timeout=30) assert result is True assert telegram.disconnected is True @@ -202,7 +202,7 @@ async def test_startup_abort_waits_for_existing_stop_task(tmp_path): result = await asyncio.wait_for( runner._abort_startup_if_shutdown_requested(adapter, Platform.TELEGRAM), - timeout=2, + timeout=30, ) assert result is True @@ -227,7 +227,7 @@ async def test_startup_aborts_after_registered_adapter_restart(tmp_path, monkeyp runner._update_platform_runtime_status = MagicMock(side_effect=update_platform_runtime_status) - result = await asyncio.wait_for(runner.start(), timeout=2) + result = await asyncio.wait_for(runner.start(), timeout=30) assert result is True assert telegram.connected is True diff --git a/tests/hermes_cli/test_auth_ssl_macos.py b/tests/hermes_cli/test_auth_ssl_macos.py index a6ebb316814..8e6c0d062f5 100644 --- a/tests/hermes_cli/test_auth_ssl_macos.py +++ b/tests/hermes_cli/test_auth_ssl_macos.py @@ -10,7 +10,6 @@ fixture because `ssl.create_default_context(cafile=...)` parses the bundle and refuses stubs. """ -import os import shutil import ssl import subprocess @@ -19,7 +18,6 @@ from pathlib import Path import pytest -sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..")) from hermes_cli.auth import _default_verify, _resolve_verify diff --git a/tests/hermes_cli/test_voice_wrapper.py b/tests/hermes_cli/test_voice_wrapper.py index a403d3c2892..eb0b90d9bbf 100644 --- a/tests/hermes_cli/test_voice_wrapper.py +++ b/tests/hermes_cli/test_voice_wrapper.py @@ -9,12 +9,9 @@ and ``speak_text`` tolerates empty input without touching the provider stack. """ -import os -import sys import pytest -sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..")) class TestPublicAPI: diff --git a/tests/test_ctx_halving_fix.py b/tests/test_ctx_halving_fix.py index 63c965ac965..6cbe4af8851 100644 --- a/tests/test_ctx_halving_fix.py +++ b/tests/test_ctx_halving_fix.py @@ -23,11 +23,8 @@ These are different and the old code conflated them; the fix keeps them separate. """ -import sys -import os from unittest.mock import MagicMock -sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..")) diff --git a/tests/test_model_picker_scroll.py b/tests/test_model_picker_scroll.py index f37a82fe611..3a30580299d 100644 --- a/tests/test_model_picker_scroll.py +++ b/tests/test_model_picker_scroll.py @@ -12,10 +12,7 @@ is always within the visible window. These tests exercise that logic in isolation without requiring a real TTY. """ -import sys -import os -sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..")) # --------------------------------------------------------------------------- diff --git a/tests/tools/test_managed_browserbase_and_modal.py b/tests/tools/test_managed_browserbase_and_modal.py index 96ab53ae090..f86d8d748b6 100644 --- a/tests/tools/test_managed_browserbase_and_modal.py +++ b/tests/tools/test_managed_browserbase_and_modal.py @@ -101,6 +101,17 @@ def _install_fake_tools_package(): sys.modules["agent.auxiliary_client"] = types.SimpleNamespace( call_llm=lambda *args, **kwargs: "", ) + # The fake `agent` package has an empty __path__, so every real + # agent.* submodule that production code imports needs an explicit + # stand-in here. tools.browser_tool imports redact_cdp_url; + # hermes_cli.auth (imported transitively by nous_account / + # tool_backend_helpers) imports sanitize_borrowed_credential_payload. + sys.modules["agent.redact"] = types.SimpleNamespace( + redact_cdp_url=lambda value: str(value), + ) + sys.modules["agent.credential_persistence"] = types.SimpleNamespace( + sanitize_borrowed_credential_payload=lambda entry, provider_id=None: entry, + ) # Stubs for the browser-provider plugin layer introduced in PR #25214. # The fake `agent` package has an empty __path__ so real submodules