feat(tests): add Blockbuster runtime gate for event-loop blocking IO (#3229)

* feat(tests): add Blockbuster runtime gate for event-loop blocking IO

Adds a strict runtime gate that fails CI when sync blocking IO calls run
on the asyncio event loop thread through DeerFlow business code.

Components:
- backend/tests/support/detectors/blocking_io_runtime.py — Blockbuster
  context scoped to `app.*` and `deerflow.*` so test infrastructure,
  pytest internals, and third-party libraries stay silent.
- backend/tests/blocking_io/conftest.py — pytest_runtest_protocol
  hookwrapper that wraps every item (setup + call + teardown) with the
  strict context. Respects `@pytest.mark.allow_blocking_io` opt-out.
- backend/tests/blocking_io/test_skills_load.py — regression anchor for
  the #1917 fix (asyncio.to_thread offload around
  LocalSkillStorage.load_skills).
- backend/tests/blocking_io/test_sqlite_lifespan.py — regression anchor
  for the #1912 fix (asyncio.to_thread offload around
  ensure_sqlite_parent_dir).
- backend/tests/blocking_io/test_gate_smoke.py — meta-test asserting the
  gate actually catches unoffloaded blocking IO and that the
  `@pytest.mark.allow_blocking_io` opt-out works.
- backend/Makefile — `make test-blocking-io` target.
- .github/workflows/backend-blocking-io-tests.yml — hard-fail PR gate on
  ubuntu-latest. Windows matrix deferred to follow-up.

Dependencies:
- blockbuster>=1.5.26,<1.6 added to dev group.

Coverage boundary (called out in PR body): the gate only catches blocking
IO on code paths the test suite actually exercises. Static AST inventory
(separate, informational) is the complementary coverage tool. Three blind
spot categories — untested paths, mocked-away paths, env-mismatched paths
— are documented in the PR description.

Findings surfaced while authoring this PR:
- resolve_sqlite_conn_str in runtime/store/_sqlite_utils.py:19 does sync
  Path.resolve() -> os.path.abspath on the lifespan loop thread, ahead of
  the #1912 fix. Not addressed here; tracked as follow-up.

Tests: 4 passed locally (`make test-blocking-io`).
Lint/format: clean (`ruff check` and `ruff format --check`).

* fix(tests): scope Blockbuster gate to blocking-io suite

* fix(tests): harden Blockbuster runtime gate

* test(blocking-io): add project rule extension point

* test(blocking-io): address review cleanup

backend/tests/blocking_io/conftest.py

@@ -0,0 +1,37 @@
+"""Pytest conftest for the strict Blockbuster runtime gate.
+
+Activates `detect_blocking_io_strict()` around the entire pytest item
+protocol (setup + call + teardown) so blocking IO in async fixtures and
+lifespan code is also caught, not just blocking IO inside the test body.
+
+Scope: only applies to items whose path is under `backend/tests/blocking_io/`.
+Pytest registers conftest hookwrappers globally once the file is loaded,
+so an explicit path filter is required to keep the strict gate from
+firing on unrelated tests when the full suite is collected.
+
+Opt-out: mark a test with `@pytest.mark.allow_blocking_io` to skip the gate.
+"""
+
+from __future__ import annotations
+
+from collections.abc import Generator
+from pathlib import Path
+
+import pytest
+from support.detectors.blocking_io_runtime import detect_blocking_io_strict
+
+_BLOCKING_IO_TEST_ROOT = Path(__file__).resolve().parent
+
+
+@pytest.hookimpl(hookwrapper=True)
+def pytest_runtest_protocol(item: pytest.Item, nextitem: pytest.Item | None) -> Generator[None, None, None]:
+    if not _is_blocking_io_item(item) or item.get_closest_marker("allow_blocking_io") is not None:
+        yield
+        return
+
+    with detect_blocking_io_strict():
+        yield
+
+
+def _is_blocking_io_item(item: pytest.Item) -> bool:
+    return Path(item.path).resolve().is_relative_to(_BLOCKING_IO_TEST_ROOT)

backend/tests/blocking_io/test_gate_smoke.py

@@ -0,0 +1,55 @@
+"""Smoke test: the strict Blockbuster gate is wired up and actively catching.
+
+Independent of any specific production code path, asserts that calling a
+known blocking IO function directly from an `async def` (without an
+`asyncio.to_thread` wrapper) raises `BlockingError`. If this test ever
+stops raising, the gate machinery itself is broken — typical causes are
+`scanned_modules` misconfiguration, accidental removal of the Blockbuster
+dev dependency, or the conftest hookwrapper no longer firing.
+
+This is the meta-test that protects every other test in this directory
+from silent regressions (a green gate that no longer catches anything is
+worse than no gate at all).
+"""
+
+from __future__ import annotations
+
+import os
+from pathlib import Path
+
+import pytest
+from blockbuster import BlockingError
+from support.detectors.blocking_io_runtime import detect_blocking_io_strict
+
+pytestmark = pytest.mark.asyncio
+
+
+async def test_gate_catches_unoffloaded_blocking_io_in_deerflow_module(tmp_path: Path) -> None:
+    from deerflow.runtime.store._sqlite_utils import ensure_sqlite_parent_dir
+
+    db_file = tmp_path / "subdir" / "store.db"
+
+    with pytest.raises(BlockingError):
+        ensure_sqlite_parent_dir(str(db_file))
+
+
+async def test_gate_restores_blockbuster_patches_after_exceptions() -> None:
+    original_stat = os.stat
+
+    with pytest.raises(RuntimeError, match="boom"):
+        with detect_blocking_io_strict():
+            raise RuntimeError("boom")
+
+    assert os.stat is original_stat
+
+
+@pytest.mark.allow_blocking_io
+async def test_allow_blocking_io_marker_opts_out_of_gate(tmp_path: Path) -> None:
+    """Verify the @pytest.mark.allow_blocking_io opt-out actually disables the gate."""
+    from deerflow.runtime.store._sqlite_utils import ensure_sqlite_parent_dir
+
+    db_file = tmp_path / "subdir" / "store.db"
+
+    ensure_sqlite_parent_dir(str(db_file))
+
+    assert db_file.parent.exists()

backend/tests/blocking_io/test_skills_load.py

@@ -0,0 +1,102 @@
+"""Regression test: skill loading must remain releasable to a worker thread.
+
+Anchors the production offload from `subagents/executor.py:_load_skills`,
+where both `get_or_new_skill_storage` and the sync `storage.load_skills(...)`
+method are dispatched via `asyncio.to_thread`. That fix addressed #1917,
+where `os.walk` inside `load_skills` blocked the LangGraph async event loop.
+
+This test invokes the production `_load_skills()` call path under the strict
+Blockbuster context against a real `LocalSkillStorage` instance pointed at
+a tmp directory. If the production `asyncio.to_thread` offload is removed,
+Blockbuster raises `BlockingError` and this test fails.
+"""
+
+from __future__ import annotations
+
+import importlib
+import sys
+from collections.abc import Iterator
+from contextlib import contextmanager
+from pathlib import Path
+from types import SimpleNamespace
+from unittest.mock import MagicMock
+
+import pytest
+
+pytestmark = pytest.mark.asyncio
+
+_MISSING = object()
+_EXECUTOR_IMPORT_MOCKS = (
+    "deerflow.agents",
+    "deerflow.agents.thread_state",
+    "deerflow.models",
+)
+
+
+def _seed_skill(skills_root: Path) -> None:
+    skill = skills_root / "public" / "demo"
+    skill.mkdir(parents=True, exist_ok=True)
+    (skill / "SKILL.md").write_text(
+        "---\nname: demo\ndescription: regression-test skill\n---\n# demo\n",
+        encoding="utf-8",
+    )
+
+
+@contextmanager
+def _real_subagent_executor() -> Iterator[type]:
+    """Import the real executor despite the suite-level circular-import mock."""
+    original_modules = {name: sys.modules.get(name, _MISSING) for name in _EXECUTOR_IMPORT_MOCKS}
+    original_executor = sys.modules.get("deerflow.subagents.executor", _MISSING)
+    parent_module = sys.modules.get("deerflow.subagents")
+    original_parent_executor = getattr(parent_module, "executor", _MISSING) if parent_module is not None else _MISSING
+
+    sys.modules.pop("deerflow.subagents.executor", None)
+    for name in _EXECUTOR_IMPORT_MOCKS:
+        sys.modules[name] = MagicMock()
+
+    try:
+        executor_module = importlib.import_module("deerflow.subagents.executor")
+        yield executor_module.SubagentExecutor
+    finally:
+        if original_executor is _MISSING:
+            sys.modules.pop("deerflow.subagents.executor", None)
+        else:
+            sys.modules["deerflow.subagents.executor"] = original_executor
+
+        if parent_module is not None:
+            if original_parent_executor is _MISSING:
+                try:
+                    delattr(parent_module, "executor")
+                except AttributeError:
+                    pass
+            else:
+                parent_module.executor = original_parent_executor
+
+        for name, module in original_modules.items():
+            if module is _MISSING:
+                sys.modules.pop(name, None)
+            else:
+                sys.modules[name] = module
+
+
+async def test_load_skills_via_to_thread_does_not_block_event_loop(tmp_path: Path) -> None:
+    from deerflow.config.skills_config import SkillsConfig
+    from deerflow.subagents.config import SubagentConfig
+
+    _seed_skill(tmp_path)
+
+    with _real_subagent_executor() as SubagentExecutor:
+        executor = SubagentExecutor(
+            config=SubagentConfig(
+                name="demo",
+                description="Loads skills through the production async path.",
+            ),
+            tools=[],
+            app_config=SimpleNamespace(skills=SkillsConfig(path=str(tmp_path))),
+            parent_model="test-model",
+        )
+
+        skills = await executor._load_skills()
+
+    assert isinstance(skills, list)
+    assert any(s.name == "demo" for s in skills)

backend/tests/blocking_io/test_sqlite_lifespan.py

@@ -0,0 +1,52 @@
+"""Regression test: sqlite path setup must run off the event loop.
+
+Anchors the production offload from
+`runtime/checkpointer/async_provider.py:_async_checkpointer`, where SQLite
+path resolution and `ensure_sqlite_parent_dir` are dispatched via
+`await asyncio.to_thread(...)`.
+That fix addressed #1912, where the sync `Path.mkdir` / `os.mkdir` inside
+`ensure_sqlite_parent_dir` ran on the FastAPI lifespan event loop thread
+and blocked startup.
+
+This test invokes the production `_async_checkpointer()` path under the
+strict Blockbuster context. The target path's parent does not yet exist, so
+the underlying path resolution and `os.mkdir` both execute. If either step is
+regressed to run directly on the event loop, Blockbuster raises
+`BlockingError` and this test fails.
+"""
+
+from __future__ import annotations
+
+import sys
+from pathlib import Path
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+
+pytestmark = pytest.mark.asyncio
+
+
+async def test_async_checkpointer_sqlite_setup_does_not_block_event_loop(tmp_path: Path) -> None:
+    from deerflow.config.checkpointer_config import CheckpointerConfig
+    from deerflow.runtime.checkpointer.async_provider import _async_checkpointer
+
+    db_file = tmp_path / "subdir" / "store.db"
+
+    mock_saver = AsyncMock()
+    mock_context_manager = AsyncMock()
+    mock_context_manager.__aenter__.return_value = mock_saver
+    mock_context_manager.__aexit__.return_value = False
+
+    mock_saver_cls = MagicMock()
+    mock_saver_cls.from_conn_string.return_value = mock_context_manager
+
+    mock_module = MagicMock()
+    mock_module.AsyncSqliteSaver = mock_saver_cls
+
+    with patch.dict(sys.modules, {"langgraph.checkpoint.sqlite.aio": mock_module}):
+        async with _async_checkpointer(CheckpointerConfig(type="sqlite", connection_string=str(db_file))) as saver:
+            assert saver is mock_saver
+
+    assert db_file.parent.exists()
+    mock_saver_cls.from_conn_string.assert_called_once_with(str(db_file.resolve()))
+    mock_saver.setup.assert_awaited_once()