feat: complete backend thread-boundary inventory (#4562)

* feat: complete backend thread-boundary inventory

Classify asyncio, dedicated executor, AnyIO, event-loop, and dynamic boundaries in the backend detector.

Add configured LangChain tool and model fallback inspection, versioned JSON output, concise summaries, focused tests, and uv-based Make integration.

* fix: classify dedicated executor helper wrappers

Pre-register ThreadPoolExecutor targets before helper discovery so run_in_executor wrappers retain dedicated executor ownership.

Add regression coverage and synchronize the detector CLI help assertion.

Makefile

@@ -26,7 +26,7 @@ help:
 	@echo "  make config          - Generate local config files (aborts if config already exists)"
 	@echo "  make config-upgrade  - Merge new fields from config.example.yaml into config.yaml"
 	@echo "  make check           - Check if all required tools are installed"
-	@echo "  make detect-thread-boundaries - Inventory async/thread boundary points"
+	@echo "  make detect-thread-boundaries - Inventory backend executor/thread/event-loop boundaries"
 	@echo "  make detect-blocking-io        - Inventory blocking IO that may block the backend event loop"
 	@echo "  make install         - Install all dependencies (frontend + backend + pre-commit hooks)"
 	@echo "  make setup-sandbox   - Pre-pull sandbox container image (recommended)"
@@ -62,7 +62,7 @@ support-bundle:
 	@$(BACKEND_UV_RUN) python ../scripts/support_bundle.py --include-doctor
 
 detect-thread-boundaries:
-	@$(PYTHON) ./scripts/detect_thread_boundaries.py
+	@$(BACKEND_UV_RUN) python ../scripts/detect_thread_boundaries.py --json-output ../.deer-flow/thread-boundary-inventory.json
 
 detect-blocking-io:
 	@$(MAKE) -C backend detect-blocking-io

backend/AGENTS.md

@@ -86,6 +86,7 @@ When making code changes, you MUST update the relevant documentation:
 ```bash
 make check      # Check system requirements
 make install    # Install all dependencies (frontend + backend)
+make detect-thread-boundaries  # Inventory backend executor/thread/event-loop boundaries
 make dev        # Start all services (Gateway + Frontend + Nginx), with config.yaml preflight
 make start      # Start production services locally
 make stop       # Stop all services
@@ -104,6 +105,38 @@ make format             # Format code with ruff
 make migrate-rev MSG="..."  # Autogenerate a new alembic revision (see Schema Migrations section)
 ```
 
+The root `detect-thread-boundaries` target statically inventories execution
+boundaries under `backend/app/` and `backend/packages/harness/deerflow/`. It
+prints a concise count by execution domain and writes the complete, versioned
+JSON payload to `.deer-flow/thread-boundary-inventory.json`. Every finding has
+a stable `boundary_kind`: `asyncio_default_executor`, `dedicated_executor`,
+`anyio_worker_thread`, `direct_event_loop_blocking`, `separate_event_loop`, or
+`unresolved_dynamic_boundary`.
+
+The AST inventory covers `asyncio.to_thread`, default and explicit
+`run_in_executor` submissions, imported aliases, simple same-module helper
+wrappers (after pre-registering dedicated executor targets), `set_default_executor`,
+`ThreadPoolExecutor` construction/submission,
+additional event loops, synchronous LangChain tools, and direct
+`BaseChatModel` fallback inheritance. It remains read-only and does not alter
+executor routing or sizing.
+
+To supplement the static scan with configured runtime types, run:
+
+```bash
+python scripts/detect_thread_boundaries.py \
+  --runtime-config config.yaml \
+  --json-output .deer-flow/thread-boundary-inventory.json
+```
+
+Runtime inspection imports configured tool objects and model classes so it can
+record concrete tool names/types/modules, sync functions, async coroutines,
+and `_agenerate`/`_astream` ownership. It does not invoke tools, instantiate
+models, or call external services; import failures remain in the JSON as
+`unresolved_dynamic_boundary` records. The detector implementation and focused
+coverage live in `tests/support/detectors/thread_boundaries.py` and
+`tests/test_detect_thread_boundaries.py`.
+
 The `detect-blocking-io` target parses `app/`, `packages/harness/deerflow/`,
 and `scripts/` with AST. By default it reports only blocking IO candidates that
 are inside async code, reachable from async code in the same file, or reachable

backend/tests/test_detect_thread_boundaries.py

@@ -2,8 +2,13 @@ from __future__ import annotations
 
 import json
 import textwrap
+from dataclasses import dataclass
 from pathlib import Path
 
+from langchain_core.language_models.chat_models import BaseChatModel
+from langchain_core.messages import AIMessage
+from langchain_core.outputs import ChatGeneration, ChatResult
+from langchain_core.tools import StructuredTool
 from support.detectors import thread_boundaries as detector
 
 
@@ -65,6 +70,165 @@ def test_scan_file_detects_async_thread_and_tool_boundaries(tmp_path):
     assert "ASYNC_ONLY_TOOL_FACTORY" in categories
 
 
+def test_scan_file_classifies_default_dedicated_anyio_and_loop_boundaries(tmp_path):
+    source_file = _write_python(
+        tmp_path / "sample.py",
+        """
+        import asyncio as aio
+        import anyio
+        from concurrent.futures import ThreadPoolExecutor as Pool
+        from langchain_core.runnables import run_in_executor as dispatch
+        from starlette.concurrency import run_in_threadpool as starlette_worker
+
+        async def offload():
+            loop = aio.get_running_loop()
+            pool = Pool(max_workers=2)
+            await aio.to_thread(str, "direct")
+            await loop.run_in_executor(None, str, "stdlib-default")
+            await aio.get_event_loop().run_in_executor(None, str, "chained-default")
+            await dispatch(None, str, "langchain-default")
+            await loop.run_in_executor(pool, str, "dedicated")
+            await anyio.to_thread.run_sync(str, "anyio")
+            await starlette_worker(str, "starlette")
+            loop.set_default_executor(pool)
+            pool.submit(str, "submitted")
+            other = aio.new_event_loop()
+            other.run_until_complete(aio.sleep(0))
+        """,
+    )
+
+    findings = detector.scan_file(source_file, repo_root=tmp_path)
+    by_category: dict[str, list[detector.BoundaryFinding]] = {}
+    for finding in findings:
+        by_category.setdefault(finding.category, []).append(finding)
+
+    assert len(by_category["ASYNC_THREAD_OFFLOAD"]) == 1
+    assert len(by_category["RUN_IN_DEFAULT_EXECUTOR"]) == 3
+    assert len(by_category["RUN_IN_DEDICATED_EXECUTOR"]) == 1
+    assert len(by_category["ANYIO_WORKER_THREAD"]) == 2
+    assert len(by_category["SET_DEFAULT_EXECUTOR"]) == 1
+    assert len(by_category["EXECUTOR_SUBMIT"]) == 1
+    assert len(by_category["NEW_EVENT_LOOP"]) == 1
+    assert {finding.boundary_kind for finding in by_category["RUN_IN_DEFAULT_EXECUTOR"]} == {detector.ASYNCIO_DEFAULT_EXECUTOR}
+    assert {finding.boundary_kind for finding in by_category["ANYIO_WORKER_THREAD"]} == {detector.ANYIO_WORKER_THREAD}
+    assert by_category["RUN_IN_DEDICATED_EXECUTOR"][0].boundary_kind == detector.DEDICATED_EXECUTOR
+    assert by_category["NEW_EVENT_LOOP"][0].boundary_kind == detector.SEPARATE_EVENT_LOOP
+
+
+def test_scan_file_discovers_same_module_thread_helper_wrappers(tmp_path):
+    source_file = _write_python(
+        tmp_path / "sample.py",
+        """
+        from asyncio import to_thread as offload
+
+        async def _worker(func, *args):
+            return await offload(func, *args)
+
+        async def caller():
+            return await _worker(str, "wrapped")
+        """,
+    )
+
+    findings = detector.scan_file(source_file, repo_root=tmp_path)
+    helper = next(finding for finding in findings if finding.category == "THREAD_HELPER_CALL")
+
+    assert helper.function == "caller"
+    assert helper.symbol == "_worker"
+    assert helper.boundary_kind == detector.ASYNCIO_DEFAULT_EXECUTOR
+
+
+def test_scan_file_classifies_dedicated_executor_helper_wrappers(tmp_path):
+    source_file = _write_python(
+        tmp_path / "sample.py",
+        """
+        import asyncio
+        from concurrent.futures import ThreadPoolExecutor as Pool
+
+        pool = Pool(max_workers=1)
+
+        async def _worker(func, *args):
+            loop = asyncio.get_running_loop()
+            return await loop.run_in_executor(pool, func, *args)
+
+        async def caller():
+            return await _worker(str, "wrapped")
+        """,
+    )
+
+    findings = detector.scan_file(source_file, repo_root=tmp_path)
+    helper = next(finding for finding in findings if finding.category == "THREAD_HELPER_CALL")
+
+    assert helper.function == "caller"
+    assert helper.symbol == "_worker"
+    assert helper.boundary_kind == detector.DEDICATED_EXECUTOR
+
+
+def test_scan_file_identifies_sync_langchain_tool_fallbacks(tmp_path):
+    source_file = _write_python(
+        tmp_path / "sample.py",
+        """
+        from langchain_core.tools import BaseTool, StructuredTool, tool
+
+        @tool
+        def sync_tool(value: str) -> str:
+            return value
+
+        class SyncOnlyTool(BaseTool):
+            def _run(self, value: str) -> str:
+                return value
+
+        def factory():
+            return StructuredTool.from_function(
+                func=sync_tool,
+                name="factory_tool",
+                description="sync only",
+            )
+        """,
+    )
+
+    findings = detector.scan_file(source_file, repo_root=tmp_path)
+    fallback_categories = {finding.category: finding.boundary_kind for finding in findings if finding.category in {"SYNC_TOOL_DEFINITION", "SYNC_TOOL_CLASS_FALLBACK", "SYNC_ONLY_TOOL_FACTORY"}}
+
+    assert fallback_categories == {
+        "SYNC_TOOL_DEFINITION": detector.ASYNCIO_DEFAULT_EXECUTOR,
+        "SYNC_TOOL_CLASS_FALLBACK": detector.ASYNCIO_DEFAULT_EXECUTOR,
+        "SYNC_ONLY_TOOL_FACTORY": detector.ASYNCIO_DEFAULT_EXECUTOR,
+    }
+
+
+def test_scan_file_identifies_base_chat_model_executor_fallbacks(tmp_path):
+    source_file = _write_python(
+        tmp_path / "sample.py",
+        """
+        from langchain_core.language_models.chat_models import BaseChatModel as ChatBase
+
+        class SyncModel(ChatBase):
+            def _generate(self, messages, stop=None, run_manager=None, **kwargs):
+                raise NotImplementedError
+
+        class AsyncModel(ChatBase):
+            def _generate(self, messages, stop=None, run_manager=None, **kwargs):
+                raise NotImplementedError
+
+            async def _agenerate(self, messages, stop=None, run_manager=None, **kwargs):
+                raise NotImplementedError
+
+            async def _astream(self, messages, stop=None, run_manager=None, **kwargs):
+                if False:
+                    yield
+        """,
+    )
+
+    findings = detector.scan_file(source_file, repo_root=tmp_path)
+    model_fallbacks = [finding for finding in findings if finding.category in {"MODEL_AGENERATE_FALLBACK", "MODEL_ASTREAM_FALLBACK"}]
+
+    assert [(finding.function, finding.category) for finding in model_fallbacks] == [
+        ("SyncModel", "MODEL_AGENERATE_FALLBACK"),
+        ("SyncModel", "MODEL_ASTREAM_FALLBACK"),
+    ]
+    assert {finding.boundary_kind for finding in model_fallbacks} == {detector.ASYNCIO_DEFAULT_EXECUTOR}
+
+
 def test_scan_file_ignores_unqualified_threads_and_generic_method_names(tmp_path):
     source_file = _write_python(
         tmp_path / "sample.py",
@@ -163,8 +327,146 @@ def test_json_output_and_min_severity_filter(tmp_path, capsys):
 
     assert exit_code == 0
     payload = json.loads(capsys.readouterr().out)
-    categories = {finding["category"] for finding in payload}
+    categories = {finding["category"] for finding in payload["static_findings"]}
     assert categories == {"SYNC_INVOKE_IN_ASYNC"}
+    assert payload["schema_version"] == 1
+    assert payload["summary"]["static_findings"] == 1
+    assert payload["runtime"] is None
+
+
+def test_summary_output_is_concise_and_grouped_by_boundary_kind(tmp_path, capsys):
+    source_file = _write_python(
+        tmp_path / "sample.py",
+        """
+        import asyncio
+        import anyio
+
+        async def handler():
+            await asyncio.to_thread(str, "x")
+            await anyio.to_thread.run_sync(str, "y")
+        """,
+    )
+
+    exit_code = detector.main(["--format", "summary", str(source_file)])
+
+    assert exit_code == 0
+    output = capsys.readouterr().out
+    assert "Thread-boundary inventory: 2 static findings" in output
+    assert "asyncio_default_executor: 1" in output
+    assert "anyio_worker_thread: 1" in output
+    assert "code:" not in output
+
+
+def _sync_tool(value: str) -> str:
+    return value
+
+
+async def _async_tool(value: str) -> str:
+    return value
+
+
+class _SyncFallbackModel(BaseChatModel):
+    @property
+    def _llm_type(self) -> str:
+        return "sync-fallback"
+
+    def _generate(self, messages, stop=None, run_manager=None, **kwargs):
+        return ChatResult(generations=[ChatGeneration(message=AIMessage(content="ok"))])
+
+
+class _NativeAsyncModel(_SyncFallbackModel):
+    async def _agenerate(self, messages, stop=None, run_manager=None, **kwargs):
+        return self._generate(messages, stop=stop, run_manager=None, **kwargs)
+
+    async def _astream(self, messages, stop=None, run_manager=None, **kwargs):
+        if False:
+            yield
+
+
+def test_runtime_inventory_identifies_tool_and_model_fallbacks_without_invocation():
+    sync_tool = StructuredTool.from_function(
+        func=_sync_tool,
+        name="sync_tool",
+        description="sync",
+    )
+    async_tool = StructuredTool.from_function(
+        coroutine=_async_tool,
+        name="async_tool",
+        description="async",
+    )
+
+    inventory = detector.inspect_runtime_components(
+        tools=[sync_tool, async_tool],
+        models=[("sync-model", _SyncFallbackModel), ("async-model", _NativeAsyncModel)],
+    )
+
+    assert [(tool.name, tool.async_requires_default_executor) for tool in inventory.tools] == [
+        ("sync_tool", True),
+        ("async_tool", False),
+    ]
+    assert inventory.tools[0].synchronous_function.endswith("._sync_tool")
+    assert inventory.tools[0].async_coroutine is None
+    assert inventory.tools[1].async_coroutine.endswith("._async_tool")
+    assert [(model.name, model.agenerate_overridden, model.astream_overridden) for model in inventory.models] == [
+        ("sync-model", False, False),
+        ("async-model", True, True),
+    ]
+    assert inventory.models[0].inherits_base_chat_model_executor_fallback is True
+    assert inventory.models[1].inherits_base_chat_model_executor_fallback is False
+
+
+@dataclass
+class _ConfiguredTool:
+    name: str
+    use: str
+
+
+@dataclass
+class _ConfiguredModel:
+    name: str
+    use: str
+
+
+@dataclass
+class _ConfiguredComponents:
+    tools: list[_ConfiguredTool]
+    models: list[_ConfiguredModel]
+
+
+def test_configured_runtime_inventory_resolves_symbols_but_never_invokes_them():
+    calls: list[tuple[str, str]] = []
+    sync_tool = StructuredTool.from_function(
+        func=_sync_tool,
+        name="resolved_tool",
+        description="sync",
+    )
+
+    def resolve_tool(path, expected_type):
+        calls.append(("tool", path))
+        return sync_tool
+
+    def resolve_model(path, base_class):
+        calls.append(("model", path))
+        return _SyncFallbackModel
+
+    config = _ConfiguredComponents(
+        tools=[_ConfiguredTool(name="configured-tool", use="package.tools:tool")],
+        models=[_ConfiguredModel(name="configured-model", use="package.models:Model")],
+    )
+
+    inventory = detector.inspect_configured_components(
+        config,
+        resolve_tool=resolve_tool,
+        resolve_model=resolve_model,
+    )
+
+    assert calls == [
+        ("tool", "package.tools:tool"),
+        ("model", "package.models:Model"),
+    ]
+    assert inventory.tools[0].configured_name == "configured-tool"
+    assert inventory.models[0].name == "configured-model"
+    assert inventory.unresolved == []
 
 
 def test_parse_errors_are_reported_as_findings(tmp_path):

backend/tests/test_detector_repo_root.py

@@ -44,7 +44,7 @@ def test_cli_shims_delegate_to_their_detectors(capsys: pytest.CaptureFixture[str
 
     for shim, description_fragment in (
         (detect_blocking_io_static, "Statically inventory blocking IO calls"),
-        (detect_thread_boundaries, "Detect async/thread boundary points"),
+        (detect_thread_boundaries, "Inventory backend thread/executor/event-loop boundaries"),
         (scan_changed_blocking_io, "blocking-IO candidates this change introduces"),
     ):
         with pytest.raises(SystemExit) as excinfo: