diff --git a/EvoScientist/_winloop.py b/EvoScientist/_winloop.py new file mode 100644 index 0000000..2ce39b3 --- /dev/null +++ b/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 diff --git a/EvoScientist/cli/__init__.py b/EvoScientist/cli/__init__.py index 2b12d55..5fe976f 100644 --- a/EvoScientist/cli/__init__.py +++ b/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.*" diff --git a/EvoScientist/langgraph_dev/__init__.py b/EvoScientist/langgraph_dev/__init__.py index 462badc..ddfce88 100644 --- a/EvoScientist/langgraph_dev/__init__.py +++ b/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() diff --git a/EvoScientist/mcp/client.py b/EvoScientist/mcp/client.py index cc00e71..4aac7db 100644 --- a/EvoScientist/mcp/client.py +++ b/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 # ============================================================================= diff --git a/EvoScientist/sessions.py b/EvoScientist/sessions.py index 8835eec..11cefc9 100644 --- a/EvoScientist/sessions.py +++ b/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()) diff --git a/EvoScientist/stream/events.py b/EvoScientist/stream/events.py index 78f21f9..6c31ffc 100644 --- a/EvoScientist/stream/events.py +++ b/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 diff --git a/tests/test_stream_recovery.py b/tests/test_stream_recovery.py new file mode 100644 index 0000000..10e311d --- /dev/null +++ b/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