From a9c8f79dfefc0d4ad52af187d947bc5ec58ff833 Mon Sep 17 00:00:00 2001 From: XuRuuuy <2948622198@qq.com> Date: Tue, 6 Oct 2026 07:52:47 +0800 Subject: [PATCH] fix(gateway): resolve the agent factory off the event loop (#6206) * fix(gateway): resolve the agent factory off the event loop start_run() called resolve_agent_factory() inline, which lazily imports the full lead-agent assembly stack (middlewares, authz, MCP, jsonschema; measured at ~8s cold on Windows, multi-second on slower filesystems). start_run is the single choke point every run-creation path flows through, so the first run after Gateway startup blocked the event loop for the whole import, stalling every other request, stream and channel on the same loop - the loop-stall family issue #5172 fixed for agent construction itself (the checkpoint-state path already resolves through the assembly pool). Resolve the factory via asyncio.to_thread: the first call pays one worker-thread hop for the import, later runs hit the cached module. Regression test pins the contract with a probe module whose module-level __getattr__ records the thread that performed the from-import. * fix(gateway): isolate factory resolution in the assembly pool Move import regressions into the strict blocking-I/O suite and cover default-executor isolation, request context, and failure-before-admission ordering. --------- Co-authored-by: Willem Jiang --- README.md | 4 + backend/app/gateway/services.py | 5 +- .../packages/harness/deerflow/utils/AGENTS.md | 5 + .../test_agent_factory_resolve_offloop.py | 185 ++++++++++++++++++ 4 files changed, 198 insertions(+), 1 deletion(-) create mode 100644 backend/tests/blocking_io/test_agent_factory_resolve_offloop.py diff --git a/README.md b/README.md index a0a7549a65..03baafab01 100644 --- a/README.md +++ b/README.md @@ -574,6 +574,10 @@ is opt-in: it fails fast when `frontend/.next` has no completed build. Gateway owns `/api/langgraph/*` and translates those public LangGraph-compatible paths to its native `/api/*` routers behind nginx. +Cold agent imports during run creation use a dedicated worker pool, keeping +unrelated Gateway requests responsive while the agent stack loads. A failed +factory import prevents the run from being admitted. + For a read-only demo without the Gateway, run `make build-static` from `frontend/`, then `HOSTNAME=127.0.0.1 PORT=3000 node --env-file=.env .next/standalone/server.js` from the same directory. The build includes public demo assets and resolves diff --git a/backend/app/gateway/services.py b/backend/app/gateway/services.py index c84e827ca7..06402eea16 100644 --- a/backend/app/gateway/services.py +++ b/backend/app/gateway/services.py @@ -1815,7 +1815,10 @@ async def start_run( # Validate even when resume takes precedence, so ignored input cannot # appear to have been admitted or persist as unchecked run audit data. normalized_input = normalize_input(body.input, trusted_internal=is_internal_caller) - agent_factory = resolve_agent_factory(body.assistant_id) + # Cold resolution imports the lead-agent stack. Use the assembly pool + # so import-lock waiters cannot starve unrelated default-executor work. + # Keep this before admission so import failures cannot create a run. + agent_factory = await run_assembly(resolve_agent_factory, body.assistant_id) command = getattr(body, "command", None) if command and command.get("resume") is not None: graph_input = Command(resume=command["resume"]) diff --git a/backend/packages/harness/deerflow/utils/AGENTS.md b/backend/packages/harness/deerflow/utils/AGENTS.md index bf7e250f95..10f9fdbbe9 100644 --- a/backend/packages/harness/deerflow/utils/AGENTS.md +++ b/backend/packages/harness/deerflow/utils/AGENTS.md @@ -23,6 +23,11 @@ lives in `tests/test_utils_llm_text.py`. ### Agent / Tool Assembly Off-Load +`start_run` also resolves the factory through `run_assembly` before admission: +cold imports cannot occupy the loop or its default pool, and import errors +cannot admit a run. `tests/blocking_io/test_agent_factory_resolve_offloop.py` +pins offload, executor isolation, request context, and failure ordering. + Tool and agent assembly re-enters `get_available_tools()` and may block on MCP discovery, so the four async assembly entry points — `run_agent`'s `agent_factory` call, `task_tool`, durable batch `_execute_item`, and `abuild_checkpoint_state_accessor` — dispatch through `deerflow.utils.assembly_io.run_assembly`, a dedicated ContextVar-preserving bounded executor (`DEER_FLOW_ASSEMBLY_WORKERS`, default 8) rather than the loop's default executor; a hung MCP server therefore parks an assembly worker instead of queueing unrelated default-executor work, and the pool logs a warning when pending assemblies exceed the worker count. `tests/blocking_io/test_tool_assembly_offloop.py` pins all four offloads plus the ContextVar propagation. ### Uploaded Document Summaries diff --git a/backend/tests/blocking_io/test_agent_factory_resolve_offloop.py b/backend/tests/blocking_io/test_agent_factory_resolve_offloop.py new file mode 100644 index 0000000000..c80ef5ba1c --- /dev/null +++ b/backend/tests/blocking_io/test_agent_factory_resolve_offloop.py @@ -0,0 +1,185 @@ +"""Regression tests: the lead-agent assembly import must stay off the event loop. + +``resolve_agent_factory`` lazily imports ``deerflow.agents.lead_agent.agent``, +which transitively imports the middlewares, authz, MCP and jsonschema chains — +multiple seconds on a cold start. ``start_run`` is the single choke point every +run-creation path flows through, so resolving it on the Gateway event loop +stalls every other request until the import finishes (the same loop-stall +family as issue #5172). ``start_run`` must therefore resolve the factory on a +dedicated assembly worker; these tests pin that contract with a probe module +whose ``__getattr__`` performs real file IO under the strict Blockbuster gate. +""" + +from __future__ import annotations + +import asyncio +import contextvars +import sys +import threading +import types +from concurrent.futures import ThreadPoolExecutor +from pathlib import Path +from types import SimpleNamespace +from unittest.mock import AsyncMock, patch + +import pytest +from blockbuster import BlockingError +from langgraph.checkpoint.memory import InMemorySaver +from langgraph.store.memory import InMemoryStore + +from app.gateway.run_models import RunCreateRequest +from app.gateway.services import resolve_agent_factory, start_run +from deerflow.config.app_config import AppConfig, reset_app_config, set_app_config +from deerflow.persistence.thread_meta.memory import MemoryThreadMetaStore +from deerflow.runtime import RunManager +from deerflow.runtime.events.store.memory import MemoryRunEventStore +from deerflow.runtime.runs.store.memory import MemoryRunStore + +LEAD_AGENT_MODULE = "deerflow.agents.lead_agent.agent" + + +@pytest.fixture +def _stub_app_config(): + """Keep run tests independent from a developer-local config.yaml.""" + set_app_config(AppConfig.model_validate({"sandbox": {"use": "deerflow.sandbox.local:LocalSandboxProvider"}})) + yield + reset_app_config() + + +def _install_probe_module(monkeypatch: pytest.MonkeyPatch, probe_file: Path) -> list[int]: + """Replace the lead-agent module with a probe whose import records the thread. + + ``from deerflow.agents.lead_agent.agent import assemble_lead_agent`` resolves + attributes through the module-level ``__getattr__`` (PEP 562), so whatever + thread calls it is the thread that pays the import. + """ + threads: list[int] = [] + + def _getattr(name: str) -> object: + if name == "assemble_lead_agent": + assert probe_file.read_text(encoding="utf-8") == "import probe" + threads.append(threading.get_ident()) + return lambda **kwargs: SimpleNamespace(graph=object()) + raise AttributeError(name) + + probe = types.ModuleType(LEAD_AGENT_MODULE) + probe.__dict__["__getattr__"] = _getattr + monkeypatch.setitem(sys.modules, LEAD_AGENT_MODULE, probe) + return threads + + +def _make_start_run_request(run_manager: RunManager) -> SimpleNamespace: + return SimpleNamespace( + headers={}, + state=SimpleNamespace(auth_source=None), + app=SimpleNamespace( + state=SimpleNamespace( + stream_bridge=SimpleNamespace(), + run_manager=run_manager, + checkpointer=InMemorySaver(), + store=InMemoryStore(), + run_event_store=MemoryRunEventStore(), + run_events_config=None, + thread_store=MemoryThreadMetaStore(InMemoryStore()), + ) + ), + ) + + +@pytest.mark.asyncio +async def test_start_run_resolves_agent_factory_off_the_event_loop(_stub_app_config, monkeypatch, tmp_path) -> None: + """``start_run`` must resolve the agent factory on a worker thread.""" + probe_file = tmp_path / "import-probe.txt" + await asyncio.to_thread(probe_file.write_text, "import probe", encoding="utf-8") + probe_threads = _install_probe_module(monkeypatch, probe_file) + run_manager = RunManager(store=MemoryRunStore()) + request = _make_start_run_request(run_manager) + body = RunCreateRequest(input={"messages": [{"role": "user", "content": "hi"}]}) + + async def fake_run_agent(*args, **kwargs): + await asyncio.sleep(0) + + with patch("app.gateway.services.run_agent", side_effect=fake_run_agent): + record = await start_run(body, "thread-resolve-offloop", request) + assert record.task is not None + await asyncio.wait_for(record.task, timeout=5) + + assert probe_threads, "resolve_agent_factory did not import the lead-agent module" + assert all(thread_id != threading.get_ident() for thread_id in probe_threads), "resolve_agent_factory imported the lead-agent stack on the event loop" + + +@pytest.mark.asyncio +async def test_inline_factory_resolution_trips_the_blocking_io_gate(monkeypatch, tmp_path) -> None: + """The import probe must fail if production resolution returns to the loop.""" + probe_file = tmp_path / "import-probe.txt" + await asyncio.to_thread(probe_file.write_text, "import probe", encoding="utf-8") + _install_probe_module(monkeypatch, probe_file) + + with pytest.raises(BlockingError): + resolve_agent_factory(None) + + +@pytest.mark.asyncio +async def test_slow_factory_resolution_leaves_default_executor_available(_stub_app_config) -> None: + """A cold import cannot occupy the default pool used by unrelated requests.""" + loop = asyncio.get_running_loop() + # One worker makes default-executor starvation deterministic. The test's + # event loop owns this executor and shuts it down during fixture teardown. + loop.set_default_executor(ThreadPoolExecutor(max_workers=1)) + resolver_started = asyncio.Event() + release_resolver = threading.Event() + request_context = contextvars.ContextVar("factory-request-context", default=None) + token = request_context.set("request-owner") + factory = object() + observed_context = [] + + def slow_resolver(assistant_id): + observed_context.append(request_context.get()) + loop.call_soon_threadsafe(resolver_started.set) + release_resolver.wait(timeout=10) + return factory + + run_manager = RunManager(store=MemoryRunStore()) + request = _make_start_run_request(run_manager) + body = RunCreateRequest(input={"messages": [{"role": "user", "content": "hi"}]}) + + with ( + patch("app.gateway.services.resolve_agent_factory", side_effect=slow_resolver) as resolver, + patch("app.gateway.services.run_agent", new_callable=AsyncMock) as worker, + ): + admission = asyncio.create_task(start_run(body, "thread-resolve-isolation", request)) + try: + await asyncio.wait_for(resolver_started.wait(), timeout=5) + assert await asyncio.wait_for(asyncio.to_thread(lambda: "unrelated-work"), timeout=5) == "unrelated-work" + assert not admission.done(), "factory resolution must finish before admitting the run" + finally: + release_resolver.set() + try: + record = await asyncio.wait_for(admission, timeout=5) + assert record.task is not None + await asyncio.wait_for(record.task, timeout=5) + finally: + request_context.reset(token) + + resolver.assert_called_once_with(body.assistant_id) + assert observed_context == ["request-owner"] + assert worker.await_args.kwargs["agent_factory"] is factory + + +@pytest.mark.asyncio +async def test_factory_resolution_failure_prevents_run_admission(_stub_app_config) -> None: + """Import errors propagate before any durable run or worker is created.""" + run_manager = RunManager(store=MemoryRunStore()) + request = _make_start_run_request(run_manager) + body = RunCreateRequest(input={"messages": [{"role": "user", "content": "hi"}]}) + + with ( + patch("app.gateway.services.resolve_agent_factory", side_effect=ImportError("factory import failed")), + patch.object(run_manager, "create_or_reject", new_callable=AsyncMock) as admit, + patch("app.gateway.services.run_agent", new_callable=AsyncMock) as worker, + ): + with pytest.raises(ImportError, match="factory import failed"): + await start_run(body, "thread-resolve-failure", request) + + admit.assert_not_awaited() + worker.assert_not_awaited()