fix: windows async MCP tool execution, restore graph state after interruptions (#290)

Co-authored-by: z00827015 <zhoulun1@huawei.com>

EvoScientist/_winloop.py

@@ -0,0 +1,50 @@
+"""Windows asyncio event-loop policy compatibility.
+
+On Windows, MCP stdio servers are launched as subprocesses by the MCP SDK's
+stdio transport, which uses ``anyio.open_process`` → ``asyncio`` async
+subprocess support. ``asyncio``'s *Selector* event loop does **not** implement
+async subprocess creation, so on a Selector loop the stdio transport falls back
+to a synchronous ``subprocess.Popen`` inside an ``async`` function. Under
+``langgraph dev`` (which enables ``blockbuster`` by default to police blocking
+I/O) that synchronous call is flagged as a ``BlockingError`` — see issue #283.
+
+The *Proactor* loop supports async subprocesses natively, so the fallback never
+happens and ``blockbuster`` allows the (now genuinely async) spawn.
+
+Windows has defaulted to ``WindowsProactorEventLoopPolicy`` since Python 3.8, so
+this is normally a no-op. We set it explicitly anyway as a safeguard: a
+dependency, IDE, or notebook host may have installed a Selector policy earlier
+in the process, and the ``langgraph dev`` subprocess in particular runs code we
+don't fully control. Calling this at each process entrypoint — **before any
+event loop is created** — guarantees the MCP subprocess path stays async.
+
+This must run at import/startup time, ahead of the first ``asyncio.run`` /
+``new_event_loop`` call; once a loop exists, swapping the policy does not change
+the already-running loop.
+"""
+
+from __future__ import annotations
+
+import sys
+
+
+def ensure_proactor_event_loop_policy() -> bool:
+    """Install ``WindowsProactorEventLoopPolicy`` on Windows if needed.
+
+    Returns ``True`` if a Proactor policy is in effect afterwards (always
+    ``False`` off Windows, where the concept doesn't apply). Safe and idempotent
+    to call multiple times; a no-op on non-Windows platforms.
+    """
+    if sys.platform != "win32":
+        return False
+
+    import asyncio
+
+    proactor_policy = getattr(asyncio, "WindowsProactorEventLoopPolicy", None)
+    if proactor_policy is None:  # pragma: no cover - non-Windows / stripped build
+        return False
+
+    current = asyncio.get_event_loop_policy()
+    if not isinstance(current, proactor_policy):
+        asyncio.set_event_loop_policy(proactor_policy())
+    return True

EvoScientist/cli/__init__.py

@@ -69,6 +69,12 @@ def main():
     import os
     import warnings
 
+    # Keep MCP stdio subprocess spawning async on Windows (see #283). Must run
+    # before any event loop is created, hence at the very top of the entrypoint.
+    from .._winloop import ensure_proactor_event_loop_policy
+
+    ensure_proactor_event_loop_policy()
+
     warnings.filterwarnings("ignore", message=".*not known to support tools.*")
     warnings.filterwarnings(
         "ignore", message=".*type is unknown and inference may fail.*"

EvoScientist/langgraph_dev/__init__.py

@@ -15,3 +15,15 @@ package only owns the *deployment* concern. Adding a new async sub-agent
 takes three steps: flip the yaml flag, add a one-line binding in
 ``graphs.py``, and register it in ``langgraph.json``.
 """
+
+# ``langgraph dev`` imports this package (to resolve the graphs listed in
+# ``langgraph.json``) inside its own subprocess, before it creates the event
+# loop that serves runs. Forcing the Proactor loop policy here — at import,
+# ahead of loop creation — keeps MCP stdio subprocess spawning async on
+# Windows so it isn't flagged as a blocking call by the dev runtime's
+# ``blockbuster`` guard (see #283). This is the entrypoint where that error
+# actually surfaces; the CLI entrypoint sets the same policy for its own
+# process.
+from .._winloop import ensure_proactor_event_loop_policy
+
+ensure_proactor_event_loop_policy()

EvoScientist/mcp/client.py

@@ -21,6 +21,81 @@ import yaml
 
 logger = logging.getLogger(__name__)
 
+
+# =============================================================================
+# Windows MCP SDK patch — drop the blocking os.access from stdio command resolution
+# =============================================================================
+#
+# On Windows, ``mcp.client.stdio`` resolves the server command on every stdio
+# session open via ``get_windows_executable_command()``, which calls
+# ``shutil.which()`` → ``os.access()`` — a blocking syscall. ``langgraph dev``
+# enables ``blockbuster`` by default, which flags that ``os.access()`` as an
+# illegal blocking call inside the event loop (one of the two root causes of
+# issue #283; the other is the Selector-loop subprocess fallback handled by
+# ``EvoScientist._winloop``).
+#
+# ``_resolve_command()`` (below) already resolves bare command names to their
+# absolute paths at connection-build time — a sync context where blocking is
+# fine. So by the time the stdio transport asks for the executable, the command
+# is already absolute and there is nothing left to look up. We short-circuit the
+# SDK's resolver for absolute paths, avoiding the ``os.access()`` entirely; bare
+# commands (which shouldn't reach here, but might via a transport we don't build)
+# still fall through to the original SDK behaviour.
+
+
+def _patch_mcp_windows_command_resolver() -> None:
+    """Make the MCP SDK's stdio command resolver skip ``os.access`` for absolute
+    paths.
+
+    Idempotent. A no-op when the MCP SDK is absent (it is an optional
+    dependency). If the SDK is present but its resolver can't be located, we log
+    a warning rather than failing silently — a silent no-op here would let the
+    blocking ``os.access`` quietly return after an SDK refactor.
+    """
+    try:
+        import mcp.client.stdio as _stdio_mod
+    except ImportError:
+        return  # MCP SDK not installed — nothing to patch.
+
+    original = getattr(_stdio_mod, "get_windows_executable_command", None)
+    if not callable(original):
+        logger.warning(
+            "MCP SDK layout changed: mcp.client.stdio.get_windows_executable_command "
+            "is missing; the Windows os.access fast-path was NOT applied. MCP stdio "
+            "tool calls may trip blocking-call detection (blockbuster) on Windows."
+        )
+        return
+
+    if getattr(original, "_evosci_absolute_fast_path", False):
+        return  # Already patched.
+
+    def _patched_get_windows_executable_command(command: str) -> str:
+        # Absolute path → already resolved, no filesystem probe needed.
+        if os.path.isabs(command):
+            return command
+        return original(command)
+
+    _patched_get_windows_executable_command._evosci_absolute_fast_path = True  # type: ignore[attr-defined]
+
+    # Patch the name in the stdio module (the call site resolves it from this
+    # module's globals) and, best-effort, the origin module in case anything
+    # imports it from there directly.
+    _stdio_mod.get_windows_executable_command = _patched_get_windows_executable_command
+    try:
+        import mcp.os.win32.utilities as _win32_utils
+
+        _win32_utils.get_windows_executable_command = (
+            _patched_get_windows_executable_command
+        )
+    except ImportError:
+        pass  # Origin module path differs in this SDK version; stdio patch suffices.
+
+    logger.debug("Applied MCP Windows command-resolver os.access fast-path patch")
+
+
+_patch_mcp_windows_command_resolver()
+
+
 # =============================================================================
 # Constants
 # =============================================================================

EvoScientist/sessions.py

@@ -1410,6 +1410,10 @@ def _api_workspace_dir() -> str:
     ``start_langgraph_dev`` injects ``EVOSCIENTIST_WORKSPACE_DIR`` and sets
     the subprocess cwd to the workspace, so either source identifies the
     workspace this server instance is serving.
+
+    NOTE: ``Path.resolve()`` / ``Path.cwd()`` call ``os.getcwd()``, a blocking
+    syscall. Call this from a sync context or via ``asyncio.to_thread`` — never
+    directly from an async function on the dev runtime (blockbuster will flag it).
     """
     import os
 
@@ -1419,6 +1423,16 @@ def _api_workspace_dir() -> str:
     return str(Path.cwd().resolve())
 
 
+async def _api_workspace_dir_async() -> str:
+    """Async wrapper for :func:`_api_workspace_dir`.
+
+    Offloads the blocking ``os.getcwd()`` (via ``Path.cwd()/resolve()``) to a
+    thread so it never runs on the event loop — the dev runtime's blockbuster
+    guard flags it otherwise. Use this from any async call site.
+    """
+    return await asyncio.to_thread(_api_workspace_dir)
+
+
 class _ApiPruningCheckpointer(PruningCheckpointer):
     """``PruningCheckpointer`` that stamps CLI-compatible ownership metadata.
 
@@ -1442,7 +1456,13 @@ class _ApiPruningCheckpointer(PruningCheckpointer):
         if isinstance(metadata, dict) and metadata.get("graph_id") == AGENT_NAME:
             metadata = dict(metadata)
             metadata.setdefault("agent_name", AGENT_NAME)
-            metadata.setdefault("workspace_dir", _api_workspace_dir())
+            # _api_workspace_dir() calls Path.resolve()/Path.cwd() -> os.getcwd(),
+            # a blocking syscall flagged by the dev runtime's blockbuster guard.
+            # Run it in a thread, and only when actually needed — ``setdefault``
+            # would evaluate the argument eagerly on every write even when the
+            # key is already present.
+            if "workspace_dir" not in metadata:
+                metadata["workspace_dir"] = await _api_workspace_dir_async()
             metadata["updated_at"] = datetime.now(UTC).isoformat()
         return await super().aput(config, checkpoint, metadata, new_versions)
 
@@ -1543,7 +1563,7 @@ async def _restore_webui_threads_to_global_store() -> None:
         # graph_id='evomemory-*' and is excluded by the first clause even
         # though it also stamps agent_name. Rows predating stamping have no
         # workspace_dir and are deliberately excluded.
-        current_workspace = _api_workspace_dir()
+        current_workspace = await _api_workspace_dir_async()
         sqlite_data: dict[uuid.UUID, tuple[str | None, str | None, str]] = {}
         titles: dict[uuid.UUID, str] = {}
         db_path = str(get_db_path())

EvoScientist/stream/events.py

@@ -13,6 +13,7 @@ from dataclasses import dataclass
 from typing import Any
 
 from langchain_core.messages import AIMessage, AIMessageChunk, BaseMessage, ToolMessage
+from langgraph.graph import END
 from langgraph.types import Command, Interrupt
 
 from ..memory.worker_activity import clear_memory_worker_saved_counts
@@ -45,6 +46,83 @@ def _is_interrupt_error_message(message: object) -> bool:
     return stripped.startswith("(Interrupt(value=")
 
 
+def _snapshot_has_pending_interrupt(snapshot: Any) -> bool:
+    """True if the graph is parked at a genuine human-in-the-loop interrupt.
+
+    A non-empty ``next`` alone can't distinguish "crashed mid-node" from
+    "legitimately waiting at ``interrupt()`` for a ``Command(resume=...)``".
+    The presence of pending interrupts — surfaced both on the snapshot and on
+    its tasks — is what tells them apart.
+    """
+    if getattr(snapshot, "interrupts", None):
+        return True
+    for task in getattr(snapshot, "tasks", None) or ():
+        if getattr(task, "interrupts", None):
+            return True
+    return False
+
+
+async def _clear_interrupted_graph_state(
+    agent: Any,
+    config: dict[str, Any],
+) -> None:
+    """Force the graph back to a clean (non-interrupted) state after an error.
+
+    When an exception occurs mid-run the LangGraph checkpoint can be left with a
+    non-empty ``next`` tuple — the graph is stuck waiting to resume at a specific
+    node. On the next invocation with a fresh user message LangGraph tries to
+    **resume** that interrupted step rather than starting a new turn: it ignores
+    the new human message and replays the broken step, which typically produces
+    no output and leaves the messages channel unchanged. From the user's side the
+    conversation looks like it lost all history because the agent stops responding.
+
+    The fix: ``aupdate_state(config, None, as_node=END)`` clears all pending tasks
+    and writes a checkpoint whose ``next`` is the empty tuple, without touching
+    any channel values (message history is preserved).
+
+    Critically, this only runs when the stuck state is *not* a legitimate
+    human-in-the-loop interrupt. The agent pauses via ``interrupt()`` /
+    ``Command(resume=...)`` for ask-user flows, which also leaves ``next``
+    non-empty; clearing those would silently discard a pending question the user
+    still needs to answer. ``_snapshot_has_pending_interrupt`` distinguishes the
+    two.
+
+    Best-effort: any failure is logged at DEBUG and swallowed so it never shadows
+    the original exception that triggered recovery.
+    """
+    import logging
+
+    _log = logging.getLogger(__name__)
+    try:
+        snapshot = await agent.aget_state(config)
+        # Only act when the graph is genuinely stuck (non-empty next tuple)...
+        if not snapshot or not getattr(snapshot, "next", None):
+            return
+        # ...and not parked at a real human-in-the-loop interrupt.
+        if _snapshot_has_pending_interrupt(snapshot):
+            _log.debug(
+                "Leaving interrupted graph state intact for thread %s: "
+                "pending human-in-the-loop interrupt (next=%s)",
+                config.get("configurable", {}).get("thread_id", "?"),
+                snapshot.next,
+            )
+            return
+
+        stuck_at = snapshot.next
+        await agent.aupdate_state(config, None, as_node=END)
+        _log.debug(
+            "Cleared interrupted graph state for thread %s (was stuck at: %s)",
+            config.get("configurable", {}).get("thread_id", "?"),
+            stuck_at,
+        )
+    except Exception as exc:  # pragma: no cover — best-effort recovery
+        _log.debug(
+            "Could not clear interrupted graph state: %s",
+            exc,
+            exc_info=True,
+        )
+
+
 @dataclass(frozen=True)
 class _SubagentInfo:
     path: tuple[str, ...]
@@ -654,6 +732,7 @@ async def stream_agent_events(
 
     stream: Any | None = None
     producers: list[asyncio.Task[Any]] = []
+    _run_raised: bool = False
     try:
         from langgraph.stream.transformers import UpdatesTransformer
 
@@ -765,6 +844,7 @@ async def stream_agent_events(
                 raise item
             yield item
     except Exception as e:
+        _run_raised = True
         yield emitter.error(str(e)).data
         raise
     finally:
@@ -780,5 +860,11 @@ async def stream_agent_events(
                 task.cancel()
         if producers:
             await asyncio.gather(*producers, return_exceptions=True)
+        # When the run ended with an exception the LangGraph checkpoint may be
+        # left interrupted (``next`` non-empty). Clear it — unless it's a real
+        # human-in-the-loop pause — so the next user message starts a fresh turn
+        # instead of replaying the broken step (which would look like lost history).
+        if _run_raised:
+            await _clear_interrupted_graph_state(agent, config)
 
     yield emitter.done(processor.full_response).data

tests/test_stream_recovery.py

@@ -0,0 +1,94 @@
+"""Targeted tests for interrupted-graph-state recovery.
+
+These run against a real compiled LangGraph graph with a checkpointer,
+so they actually verify the two claims the recovery rests on:
+
+1. After a mid-run crash, ``aupdate_state(config, None, as_node=END)`` clears the
+   stuck ``next`` tuple while preserving channel values.
+2. A legitimate human-in-the-loop ``interrupt()`` (also a non-empty ``next``) is
+   left intact, so a pending question is never silently discarded.
+"""
+
+from typing import TypedDict
+
+from langgraph.checkpoint.memory import InMemorySaver
+from langgraph.graph import END, START, StateGraph
+from langgraph.types import interrupt
+
+from EvoScientist.stream.events import _clear_interrupted_graph_state
+from tests.conftest import run_async as _run
+
+
+class _S(TypedDict):
+    x: int
+
+
+def _crashing_app():
+    # Node 'b' crashes once, then succeeds — so a post-recovery run can complete
+    # and prove the graph is genuinely unstuck (not replaying the dead step).
+    crashed = {"v": False}
+
+    def a(state):
+        return {"x": state["x"] + 1}
+
+    def b(state):
+        if not crashed["v"]:
+            crashed["v"] = True
+            raise RuntimeError("boom")
+        return {"x": state["x"] + 100}
+
+    g = StateGraph(_S)
+    g.add_node("a", a)
+    g.add_node("b", b)
+    g.add_edge(START, "a")
+    g.add_edge("a", "b")
+    g.add_edge("b", END)
+    return g.compile(checkpointer=InMemorySaver())
+
+
+def _interrupting_app():
+    def ask(state):
+        interrupt({"question": "continue?"})
+        return {"x": state["x"] + 1}
+
+    g = StateGraph(_S)
+    g.add_node("ask", ask)
+    g.add_edge(START, "ask")
+    g.add_edge("ask", END)
+    return g.compile(checkpointer=InMemorySaver())
+
+
+def test_recovery_clears_stuck_state_after_crash():
+    app = _crashing_app()
+    cfg = {"configurable": {"thread_id": "t1"}}
+    try:
+        app.invoke({"x": 0}, cfg)
+    except Exception:
+        pass  # LangGraph re-raises the node error (wrapped); we only care about state
+    # The crash left the graph frozen at node 'b'.
+    assert app.get_state(cfg).next == ("b",)
+
+    _run(_clear_interrupted_graph_state(app, cfg))
+
+    snap = app.get_state(cfg)
+    assert snap.next == ()  # stuck state actually cleared
+    assert snap.values == {"x": 1}  # channel values (history) preserved
+
+    # And the graph is genuinely unstuck: a fresh run completes (a: +1, b: +100)
+    # instead of replaying the dead node.
+    assert app.invoke({"x": 41}, cfg)["x"] == 142
+
+
+def test_recovery_preserves_pending_hitl_interrupt():
+    app = _interrupting_app()
+    cfg = {"configurable": {"thread_id": "t1"}}
+    app.invoke({"x": 0}, cfg)  # parks at interrupt()
+    before = app.get_state(cfg)
+    assert before.next == ("ask",)
+    assert before.interrupts
+
+    _run(_clear_interrupted_graph_state(app, cfg))
+
+    after = app.get_state(cfg)
+    assert after.next == ("ask",)  # interrupt left intact, still resumable
+    assert after.interrupts