From d3ce5de218853e7cca6136a9ed9ba63111787158 Mon Sep 17 00:00:00 2001 From: heart-scalpel Date: Wed, 29 Jul 2026 22:33:45 +0800 Subject: [PATCH] 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 | 4 +- backend/AGENTS.md | 33 + .../support/detectors/thread_boundaries.py | 719 ++++++++++++++++-- .../tests/test_detect_thread_boundaries.py | 304 +++++++- backend/tests/test_detector_repo_root.py | 2 +- 5 files changed, 1003 insertions(+), 59 deletions(-) diff --git a/Makefile b/Makefile index d8d69544b..6dfeb4c02 100644 --- a/Makefile +++ b/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 diff --git a/backend/AGENTS.md b/backend/AGENTS.md index c6b8f72f9..cf45bc219 100644 --- a/backend/AGENTS.md +++ b/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 diff --git a/backend/tests/support/detectors/thread_boundaries.py b/backend/tests/support/detectors/thread_boundaries.py index 42db69f9a..3853eb456 100644 --- a/backend/tests/support/detectors/thread_boundaries.py +++ b/backend/tests/support/detectors/thread_boundaries.py @@ -1,22 +1,26 @@ -"""Inventory async/thread boundary points for developer review. +"""Inventory backend async, executor, thread, and event-loop boundaries. -This detector is intentionally non-invasive: it parses Python source with AST -and reports places where code crosses sync/async/thread boundaries. Findings -are review evidence, not automatic bug decisions. +The static detector parses source without importing application modules. The +optional runtime inspector resolves configured tool objects and model classes, +but never invokes a tool, instantiates a model, or calls an external service. -Not directly executable: import as `support.detectors.thread_boundaries` or -run via the CLI shim `scripts/detect_thread_boundaries.py`. +Not directly executable: import as ``support.detectors.thread_boundaries`` or +run via ``scripts/detect_thread_boundaries.py``. """ from __future__ import annotations import argparse import ast +import inspect import json import os -from collections.abc import Iterable, Sequence +import sys +from collections import Counter +from collections.abc import Callable, Iterable, Sequence from dataclasses import asdict, dataclass from pathlib import Path +from typing import Any from support.detectors.repo_root import resolve_repo_root @@ -35,12 +39,32 @@ IGNORED_DIR_NAMES = { "node_modules", } SEVERITY_ORDER = {"INFO": 0, "WARN": 1, "FAIL": 2} +SCHEMA_VERSION = 1 + +# Stable machine-readable execution domains. Keep these values independent of +# category names: categories explain what happened, while boundary_kind tells +# later starvation tests which worker domain a finding consumes. +ASYNCIO_DEFAULT_EXECUTOR = "asyncio_default_executor" +DEDICATED_EXECUTOR = "dedicated_executor" +ANYIO_WORKER_THREAD = "anyio_worker_thread" +DIRECT_EVENT_LOOP_BLOCKING = "direct_event_loop_blocking" +SEPARATE_EVENT_LOOP = "separate_event_loop" +UNRESOLVED_DYNAMIC_BOUNDARY = "unresolved_dynamic_boundary" +BOUNDARY_KINDS = ( + ASYNCIO_DEFAULT_EXECUTOR, + DEDICATED_EXECUTOR, + ANYIO_WORKER_THREAD, + DIRECT_EVENT_LOOP_BLOCKING, + SEPARATE_EVENT_LOOP, + UNRESOLVED_DYNAMIC_BOUNDARY, +) @dataclass(frozen=True) class BoundaryFinding: severity: str category: str + boundary_kind: str path: str line: int column: int @@ -54,6 +78,63 @@ class BoundaryFinding: return asdict(self) +@dataclass(frozen=True) +class RuntimeToolFinding: + configured_name: str | None + name: str + type: str + source_module: str + synchronous_function: str | None + async_coroutine: str | None + async_requires_default_executor: bool + boundary_kind: str | None + source: str | None = None + + def to_dict(self) -> dict[str, object]: + return asdict(self) + + +@dataclass(frozen=True) +class RuntimeModelFinding: + name: str + type: str + source_module: str + agenerate_overridden: bool + astream_overridden: bool + inherits_base_chat_model_executor_fallback: bool + boundary_kind: str | None + source: str | None = None + + def to_dict(self) -> dict[str, object]: + return asdict(self) + + +@dataclass(frozen=True) +class UnresolvedComponent: + component_kind: str + name: str + source: str + error: str + boundary_kind: str = UNRESOLVED_DYNAMIC_BOUNDARY + + def to_dict(self) -> dict[str, object]: + return asdict(self) + + +@dataclass(frozen=True) +class RuntimeInventory: + tools: list[RuntimeToolFinding] + models: list[RuntimeModelFinding] + unresolved: list[UnresolvedComponent] + + def to_dict(self) -> dict[str, object]: + return { + "tools": [tool.to_dict() for tool in self.tools], + "models": [model.to_dict() for model in self.models], + "unresolved": [component.to_dict() for component in self.unresolved], + } + + @dataclass(frozen=True) class _FunctionContext: name: str @@ -64,6 +145,7 @@ class _FunctionContext: class _CallRule: severity: str category: str + boundary_kind: str message: str @@ -71,55 +153,108 @@ EXACT_CALL_RULES: dict[str, _CallRule] = { "asyncio.run": _CallRule( "WARN", "SYNC_ASYNC_BRIDGE", + SEPARATE_EVENT_LOOP, "Runs a coroutine from synchronous code by creating an event loop boundary.", ), "asyncio.to_thread": _CallRule( "INFO", "ASYNC_THREAD_OFFLOAD", - "Offloads synchronous work from an async context into a worker thread.", + ASYNCIO_DEFAULT_EXECUTOR, + "Offloads synchronous work into the asyncio default executor.", ), "deerflow.utils.file_io.run_file_io": _CallRule( "INFO", "ASYNC_FILE_IO_OFFLOAD", - "Offloads filesystem-oriented work from an async context into the dedicated file IO thread pool.", + DEDICATED_EXECUTOR, + "Offloads filesystem work into DeerFlow's dedicated file-IO executor.", + ), + "anyio.to_thread.run_sync": _CallRule( + "INFO", + "ANYIO_WORKER_THREAD", + ANYIO_WORKER_THREAD, + "Offloads synchronous work into AnyIO's worker-thread limiter domain.", + ), + "starlette.concurrency.run_in_threadpool": _CallRule( + "INFO", + "ANYIO_WORKER_THREAD", + ANYIO_WORKER_THREAD, + "Offloads synchronous work through Starlette into AnyIO worker threads.", ), "asyncio.new_event_loop": _CallRule( "WARN", "NEW_EVENT_LOOP", - "Creates a separate event loop; review resource ownership across loops.", + SEPARATE_EVENT_LOOP, + "Creates a separate event loop.", + ), + "asyncio.Runner": _CallRule( + "WARN", + "NEW_EVENT_LOOP", + SEPARATE_EVENT_LOOP, + "Creates an asyncio runner that owns a separate event loop.", ), "asyncio.run_coroutine_threadsafe": _CallRule( "WARN", "CROSS_THREAD_COROUTINE", + SEPARATE_EVENT_LOOP, "Submits a coroutine to an event loop from another thread.", ), "concurrent.futures.ThreadPoolExecutor": _CallRule( "INFO", "THREAD_POOL", - "Creates a thread pool boundary.", + DEDICATED_EXECUTOR, + "Creates a dedicated thread-pool executor.", ), "threading.Thread": _CallRule( "INFO", "RAW_THREAD", - "Creates a raw thread; ContextVar values do not propagate automatically.", + UNRESOLVED_DYNAMIC_BOUNDARY, + "Creates a raw thread outside the classified executor domains.", ), "threading.Timer": _CallRule( "INFO", "RAW_TIMER_THREAD", - "Creates a timer-backed raw thread; ContextVar values do not propagate automatically.", + UNRESOLVED_DYNAMIC_BOUNDARY, + "Creates a timer-backed raw thread outside the classified executor domains.", ), "make_sync_tool_wrapper": _CallRule( "INFO", "SYNC_TOOL_WRAPPER", - "Adapts an async tool coroutine for synchronous tool invocation.", + DEDICATED_EXECUTOR, + "Adapts an async tool for sync invocation through DeerFlow's dedicated tool executor.", + ), + "deerflow.tools.sync.make_sync_tool_wrapper": _CallRule( + "INFO", + "SYNC_TOOL_WRAPPER", + DEDICATED_EXECUTOR, + "Adapts an async tool for sync invocation through DeerFlow's dedicated tool executor.", ), } THREAD_POOL_CONSTRUCTORS = {"concurrent.futures.ThreadPoolExecutor"} +RUN_IN_EXECUTOR_CALLS = {"langchain_core.runnables.run_in_executor"} ASYNC_TOOL_FACTORY_CALLS = { "StructuredTool.from_function", "langchain.tools.StructuredTool.from_function", "langchain_core.tools.StructuredTool.from_function", } +TOOL_DECORATORS = { + "langchain.tools.tool", + "langchain_core.tools.tool", +} +BASE_TOOL_CLASSES = { + "langchain.tools.BaseTool", + "langchain_core.tools.BaseTool", + "langchain_core.tools.base.BaseTool", +} +STRUCTURED_TOOL_CLASSES = { + "langchain.tools.StructuredTool", + "langchain_core.tools.StructuredTool", + "langchain_core.tools.structured.StructuredTool", +} +BASE_CHAT_MODEL_CLASSES = { + "langchain.chat_models.BaseChatModel", + "langchain_core.language_models.BaseChatModel", + "langchain_core.language_models.chat_models.BaseChatModel", +} LANGCHAIN_INVOKE_RECEIVER_NAMES = { "agent", "chain", @@ -137,32 +272,36 @@ LANGCHAIN_INVOKE_RECEIVER_SUFFIXES = ( "_model", "_runnable", ) - ASYNC_BLOCKING_CALL_RULES: dict[str, _CallRule] = { "time.sleep": _CallRule( "WARN", "BLOCKING_CALL_IN_ASYNC", + DIRECT_EVENT_LOOP_BLOCKING, "Blocks the event loop when called directly inside async code.", ), "subprocess.run": _CallRule( "WARN", "BLOCKING_SUBPROCESS_IN_ASYNC", + DIRECT_EVENT_LOOP_BLOCKING, "Runs a blocking subprocess from async code.", ), "subprocess.check_call": _CallRule( "WARN", "BLOCKING_SUBPROCESS_IN_ASYNC", + DIRECT_EVENT_LOOP_BLOCKING, "Runs a blocking subprocess from async code.", ), "subprocess.check_output": _CallRule( "WARN", "BLOCKING_SUBPROCESS_IN_ASYNC", + DIRECT_EVENT_LOOP_BLOCKING, "Runs a blocking subprocess from async code.", ), "subprocess.Popen": _CallRule( "WARN", "BLOCKING_SUBPROCESS_IN_ASYNC", - "Starts a subprocess from async code; review whether it blocks later.", + DIRECT_EVENT_LOOP_BLOCKING, + "Starts a subprocess from async code; review whether later operations block.", ), } @@ -189,7 +328,7 @@ def is_none_node(node: ast.AST | None) -> bool: class BoundaryVisitor(ast.NodeVisitor): - def __init__(self, path: Path, relative_path: str, source_lines: Sequence[str]) -> None: + def __init__(self, path: Path, relative_path: str, source_lines: Sequence[str], tree: ast.Module) -> None: self.path = path self.relative_path = relative_path self.source_lines = source_lines @@ -197,6 +336,8 @@ class BoundaryVisitor(ast.NodeVisitor): self.function_stack: list[_FunctionContext] = [] self.import_aliases: dict[str, str] = {} self.executor_names: set[str] = set() + self.thread_helpers: dict[str, str] = {} + self._prepare(tree) @property def current_function(self) -> str: @@ -208,19 +349,45 @@ class BoundaryVisitor(ast.NodeVisitor): def in_async_context(self) -> bool: return bool(self.function_stack and self.function_stack[-1].is_async) - def visit_Import(self, node: ast.Import) -> None: + def _prepare(self, tree: ast.Module) -> None: + for node in ast.walk(tree): + if isinstance(node, ast.Import): + self._record_import(node) + elif isinstance(node, ast.ImportFrom): + self._record_import_from(node) + self._discover_executor_names(tree) + self._discover_thread_helpers(tree) + + def _discover_executor_names(self, tree: ast.Module) -> None: + for node in ast.walk(tree): + if isinstance(node, ast.Assign): + self._record_executor_targets(node.value, node.targets) + elif isinstance(node, ast.AnnAssign) and node.value is not None: + self._record_executor_targets(node.value, [node.target]) + elif isinstance(node, ast.With): + for item in node.items: + if item.optional_vars is not None: + self._record_executor_targets(item.context_expr, [item.optional_vars]) + + def _record_import(self, node: ast.Import) -> None: for alias in node.names: local_name = alias.asname or alias.name.split(".", 1)[0] canonical_name = alias.name if alias.asname else local_name self.import_aliases[local_name] = canonical_name - def visit_ImportFrom(self, node: ast.ImportFrom) -> None: + def _record_import_from(self, node: ast.ImportFrom) -> None: if node.module is None: return for alias in node.names: local_name = alias.asname or alias.name self.import_aliases[local_name] = f"{node.module}.{alias.name}" + def visit_Import(self, node: ast.Import) -> None: + self._record_import(node) + + def visit_ImportFrom(self, node: ast.ImportFrom) -> None: + self._record_import_from(node) + def visit_Assign(self, node: ast.Assign) -> None: self._record_executor_targets(node.value, node.targets) self.generic_visit(node) @@ -236,10 +403,21 @@ class BoundaryVisitor(ast.NodeVisitor): self._record_executor_targets(item.context_expr, [item.optional_vars]) self.generic_visit(node) + def visit_ClassDef(self, node: ast.ClassDef) -> None: + self.function_stack.append(_FunctionContext(node.name, is_async=False)) + try: + self._check_fallback_class(node) + self.generic_visit(node) + finally: + self.function_stack.pop() + def visit_FunctionDef(self, node: ast.FunctionDef) -> None: self.function_stack.append(_FunctionContext(node.name, is_async=False)) - self.generic_visit(node) - self.function_stack.pop() + try: + self._check_sync_tool_definition(node) + self.generic_visit(node) + finally: + self.function_stack.pop() def visit_AsyncFunctionDef(self, node: ast.AsyncFunctionDef) -> None: self.function_stack.append(_FunctionContext(node.name, is_async=True)) @@ -250,57 +428,195 @@ class BoundaryVisitor(ast.NodeVisitor): self.function_stack.pop() def visit_Call(self, node: ast.Call) -> None: - call_name = self._canonical_name(dotted_name(node.func)) + raw_name = dotted_name(node.func) + call_name = self._canonical_name(raw_name) if call_name: - self._check_call(node, call_name) + self._check_call(node, raw_name or call_name, call_name) self.generic_visit(node) + def _discover_thread_helpers(self, tree: ast.Module) -> None: + """Discover simple same-module wrappers around a known offload call.""" + for definition in tree.body: + if not isinstance(definition, (ast.FunctionDef, ast.AsyncFunctionDef)): + continue + kinds: set[str] = set() + for node in ast.walk(definition): + if not isinstance(node, ast.Call): + continue + call_name = self._canonical_name(dotted_name(node.func)) + if call_name == "asyncio.to_thread": + kinds.add(ASYNCIO_DEFAULT_EXECUTOR) + elif call_name in { + "anyio.to_thread.run_sync", + "starlette.concurrency.run_in_threadpool", + }: + kinds.add(ANYIO_WORKER_THREAD) + elif call_name == "deerflow.utils.file_io.run_file_io": + kinds.add(DEDICATED_EXECUTOR) + elif self._is_run_in_executor(call_name): + kinds.add(self._executor_boundary_kind(node)) + if len(kinds) == 1: + self.thread_helpers[definition.name] = kinds.pop() + + def _check_sync_tool_definition(self, node: ast.FunctionDef) -> None: + if self._has_tool_decorator(node): + self._emit( + node, + severity="INFO", + category="SYNC_TOOL_DEFINITION", + boundary_kind=ASYNCIO_DEFAULT_EXECUTOR, + symbol=self._tool_decorator_name(node) or "tool", + message="Defines a synchronous LangChain tool whose async fallback uses the default executor.", + ) + def _check_async_tool_definition(self, node: ast.AsyncFunctionDef) -> None: + decorator_name = self._tool_decorator_name(node) + if decorator_name: + self._emit( + node, + severity="INFO", + category="ASYNC_TOOL_DEFINITION", + boundary_kind=UNRESOLVED_DYNAMIC_BOUNDARY, + symbol=decorator_name, + message="Defines a native async LangChain tool; sync callers may still require a wrapper.", + ) + + def _has_tool_decorator(self, node: ast.FunctionDef | ast.AsyncFunctionDef) -> bool: + return self._tool_decorator_name(node) is not None + + def _tool_decorator_name(self, node: ast.FunctionDef | ast.AsyncFunctionDef) -> str | None: for decorator in node.decorator_list: decorator_call = decorator.func if isinstance(decorator, ast.Call) else decorator decorator_name = self._canonical_name(dotted_name(decorator_call)) - if decorator_name in {"langchain.tools.tool", "langchain_core.tools.tool"}: + if decorator_name in TOOL_DECORATORS: + return decorator_name + return None + + def _check_fallback_class(self, node: ast.ClassDef) -> None: + base_names = {self._canonical_name(dotted_name(base)) for base in node.bases} + method_names = {child.name for child in node.body if isinstance(child, (ast.FunctionDef, ast.AsyncFunctionDef))} + if base_names & (BASE_TOOL_CLASSES | STRUCTURED_TOOL_CLASSES): + if "_run" in method_names and "_arun" not in method_names: self._emit( node, severity="INFO", - category="ASYNC_TOOL_DEFINITION", - symbol=decorator_name, - message="Defines an async LangChain tool; sync clients need a wrapper before invoke().", + category="SYNC_TOOL_CLASS_FALLBACK", + boundary_kind=ASYNCIO_DEFAULT_EXECUTOR, + symbol=node.name, + message="A sync-only BaseTool subclass inherits the default-executor async fallback.", + ) + if base_names & BASE_CHAT_MODEL_CLASSES: + if "_agenerate" not in method_names: + self._emit( + node, + severity="INFO", + category="MODEL_AGENERATE_FALLBACK", + boundary_kind=ASYNCIO_DEFAULT_EXECUTOR, + symbol=node.name, + message="The model inherits BaseChatModel._agenerate's default-executor fallback.", + ) + if "_astream" not in method_names: + self._emit( + node, + severity="INFO", + category="MODEL_ASTREAM_FALLBACK", + boundary_kind=ASYNCIO_DEFAULT_EXECUTOR, + symbol=node.name, + message="The model inherits BaseChatModel._astream's default-executor fallback.", ) - return - def _check_call(self, node: ast.Call, call_name: str) -> None: + def _check_call(self, node: ast.Call, raw_name: str, call_name: str) -> None: rule = EXACT_CALL_RULES.get(call_name) if rule: self._emit_rule(node, call_name, rule) + if call_name.endswith(".new_event_loop") and call_name != "asyncio.new_event_loop": + self._emit( + node, + severity="WARN", + category="NEW_EVENT_LOOP", + boundary_kind=SEPARATE_EVENT_LOOP, + symbol=call_name, + message="Creates a separate event loop through an event-loop policy or compatible API.", + ) + if call_name.endswith(".run_until_complete"): self._emit( node, severity="WARN", category="RUN_UNTIL_COMPLETE", + boundary_kind=SEPARATE_EVENT_LOOP, symbol=call_name, - message="Drives an event loop from synchronous code; review nested-loop behavior.", + message="Drives an event loop synchronously.", ) - if self._is_executor_submit(node, call_name): + if self._is_run_in_executor(call_name): + boundary_kind = self._executor_boundary_kind(node) + category = { + ASYNCIO_DEFAULT_EXECUTOR: "RUN_IN_DEFAULT_EXECUTOR", + DEDICATED_EXECUTOR: "RUN_IN_DEDICATED_EXECUTOR", + UNRESOLVED_DYNAMIC_BOUNDARY: "RUN_IN_DYNAMIC_EXECUTOR", + }[boundary_kind] + self._emit( + node, + severity="INFO", + category=category, + boundary_kind=boundary_kind, + symbol=call_name, + message="Submits work through run_in_executor; the executor argument determines ownership.", + ) + + if call_name.endswith(".set_default_executor"): + self._emit( + node, + severity="INFO", + category="SET_DEFAULT_EXECUTOR", + boundary_kind=DEDICATED_EXECUTOR, + symbol=call_name, + message="Replaces an event loop's default executor with an explicitly owned executor.", + ) + + if self._is_executor_submit(node, raw_name): self._emit( node, severity="INFO", category="EXECUTOR_SUBMIT", + boundary_kind=DEDICATED_EXECUTOR, symbol=call_name, - message="Submits work to an executor; review context propagation and cancellation.", + message="Submits work directly to a dedicated ThreadPoolExecutor.", ) if call_name in ASYNC_TOOL_FACTORY_CALLS: - if any(keyword.arg == "coroutine" and not is_none_node(keyword.value) for keyword in node.keywords): + has_coroutine = any(keyword.arg == "coroutine" and not is_none_node(keyword.value) for keyword in node.keywords) + if has_coroutine: self._emit( node, severity="INFO", category="ASYNC_ONLY_TOOL_FACTORY", + boundary_kind=UNRESOLVED_DYNAMIC_BOUNDARY, symbol=call_name, - message="Creates a StructuredTool from a coroutine; sync clients need a wrapper.", + message="Creates a StructuredTool with a native async coroutine.", ) + elif self._factory_has_sync_function(node): + self._emit( + node, + severity="INFO", + category="SYNC_ONLY_TOOL_FACTORY", + boundary_kind=ASYNCIO_DEFAULT_EXECUTOR, + symbol=call_name, + message="Creates a sync-only StructuredTool whose async fallback uses the default executor.", + ) + + helper_kind = self.thread_helpers.get(raw_name) + if helper_kind is not None and raw_name != self.current_function.rsplit(".", 1)[-1]: + self._emit( + node, + severity="INFO", + category="THREAD_HELPER_CALL", + boundary_kind=helper_kind, + symbol=raw_name, + message="Calls a same-module helper that wraps a classified thread boundary.", + ) if self.in_async_context and call_name in ASYNC_BLOCKING_CALL_RULES: self._emit_rule(node, call_name, ASYNC_BLOCKING_CALL_RULES[call_name]) @@ -310,8 +626,9 @@ class BoundaryVisitor(ast.NodeVisitor): node, severity="WARN", category="SYNC_INVOKE_IN_ASYNC", + boundary_kind=DIRECT_EVENT_LOOP_BLOCKING, symbol=call_name, - message="Calls a synchronous invoke() from async code; review event-loop blocking.", + message="Calls synchronous invoke() directly from async code.", ) if not self.in_async_context and self._is_langchain_invoke(node, call_name, method_name="ainvoke"): @@ -319,8 +636,9 @@ class BoundaryVisitor(ast.NodeVisitor): node, severity="WARN", category="ASYNC_INVOKE_IN_SYNC", + boundary_kind=UNRESOLVED_DYNAMIC_BOUNDARY, symbol=call_name, - message="Calls async ainvoke() from sync code; review how the coroutine is awaited.", + message="Calls async ainvoke() from sync code; review how the coroutine is driven.", ) def _canonical_name(self, name: str | None) -> str | None: @@ -338,22 +656,43 @@ class BoundaryVisitor(ast.NodeVisitor): if call_name not in THREAD_POOL_CONSTRUCTORS: return for target in targets: - for name in self._target_names(target): - self.executor_names.add(name) + self.executor_names.update(self._target_names(target)) def _target_names(self, target: ast.AST) -> Iterable[str]: - if isinstance(target, ast.Name): - yield target.id + name = dotted_name(target) + if name: + yield name elif isinstance(target, (ast.Tuple, ast.List)): for element in target.elts: yield from self._target_names(element) - def _is_executor_submit(self, node: ast.Call, call_name: str) -> bool: - if not call_name.endswith(".submit"): + def _is_executor_submit(self, node: ast.Call, raw_name: str) -> bool: + if not raw_name.endswith(".submit"): return False receiver_name = call_receiver_name(node) return receiver_name in self.executor_names + def _is_run_in_executor(self, call_name: str | None) -> bool: + return bool(call_name and (call_name in RUN_IN_EXECUTOR_CALLS or call_name == "run_in_executor" or call_name.endswith(".run_in_executor"))) + + def _executor_boundary_kind(self, node: ast.Call) -> str: + executor_arg: ast.AST | None = node.args[0] if node.args else None + for keyword in node.keywords: + if keyword.arg == "executor": + executor_arg = keyword.value + break + if is_none_node(executor_arg): + return ASYNCIO_DEFAULT_EXECUTOR + executor_name = dotted_name(executor_arg) + if executor_name in self.executor_names: + return DEDICATED_EXECUTOR + return UNRESOLVED_DYNAMIC_BOUNDARY + + def _factory_has_sync_function(self, node: ast.Call) -> bool: + if node.args and not is_none_node(node.args[0]): + return True + return any(keyword.arg == "func" and not is_none_node(keyword.value) for keyword in node.keywords) + def _is_langchain_invoke(self, node: ast.Call, call_name: str, *, method_name: str) -> bool: if not call_name.endswith(f".{method_name}"): return False @@ -368,11 +707,21 @@ class BoundaryVisitor(ast.NodeVisitor): node, severity=rule.severity, category=rule.category, + boundary_kind=rule.boundary_kind, symbol=symbol, message=rule.message, ) - def _emit(self, node: ast.AST, *, severity: str, category: str, symbol: str, message: str) -> None: + def _emit( + self, + node: ast.AST, + *, + severity: str, + category: str, + boundary_kind: str, + symbol: str, + message: str, + ) -> None: line = getattr(node, "lineno", 0) column = getattr(node, "col_offset", 0) code = "" @@ -382,6 +731,7 @@ class BoundaryVisitor(ast.NodeVisitor): BoundaryFinding( severity=severity, category=category, + boundary_kind=boundary_kind, path=self.relative_path, line=line, column=column, @@ -414,6 +764,7 @@ def scan_file(path: Path, *, repo_root: Path = REPO_ROOT) -> list[BoundaryFindin BoundaryFinding( severity="WARN", category="PARSE_ERROR", + boundary_kind=UNRESOLVED_DYNAMIC_BOUNDARY, path=relative_path, line=line, column=max((exc.offset or 1) - 1, 0), @@ -425,7 +776,7 @@ def scan_file(path: Path, *, repo_root: Path = REPO_ROOT) -> list[BoundaryFindin ) ] - visitor = BoundaryVisitor(path, relative_path, source_lines) + visitor = BoundaryVisitor(path, relative_path, source_lines, tree) visitor.visit(tree) return visitor.findings @@ -449,25 +800,259 @@ def iter_python_files(paths: Iterable[Path]) -> Iterable[Path]: yield Path(dirpath) / filename -def scan_paths(paths: Iterable[Path], *, repo_root: Path = REPO_ROOT) -> list[BoundaryFinding]: +def scan_paths( + paths: Iterable[Path], + *, + repo_root: Path = REPO_ROOT, +) -> list[BoundaryFinding]: findings: list[BoundaryFinding] = [] for path in sorted(iter_python_files(paths)): findings.extend(scan_file(path, repo_root=repo_root)) - return sorted(findings, key=lambda finding: (finding.path, finding.line, finding.column, finding.category)) + return sorted( + findings, + key=lambda finding: ( + finding.path, + finding.line, + finding.column, + finding.category, + ), + ) -def filter_findings(findings: Iterable[BoundaryFinding], min_severity: str) -> list[BoundaryFinding]: +def filter_findings( + findings: Iterable[BoundaryFinding], + min_severity: str, +) -> list[BoundaryFinding]: threshold = SEVERITY_ORDER[min_severity] return [finding for finding in findings if SEVERITY_ORDER[finding.severity] >= threshold] +def _callable_reference(value: Any) -> str | None: + if value is None or not callable(value): + return None + unwrapped = inspect.unwrap(value) + module = getattr(unwrapped, "__module__", type(unwrapped).__module__) + qualname = getattr( + unwrapped, + "__qualname__", + getattr(unwrapped, "__name__", type(unwrapped).__qualname__), + ) + return f"{module}.{qualname}" + + +def _type_reference(value: object | type) -> str: + cls = value if isinstance(value, type) else type(value) + return f"{cls.__module__}.{cls.__qualname__}" + + +def _inspect_tool( + tool: object, + *, + configured_name: str | None = None, + source: str | None = None, +) -> RuntimeToolFinding: + from langchain_core.tools import BaseTool, StructuredTool + + tool_class = type(tool) + coroutine = getattr(tool, "coroutine", None) + sync_function = getattr(tool, "func", None) + if not callable(sync_function): + run_method = getattr(tool_class, "_run", None) + sync_function = run_method if run_method is not None and run_method is not BaseTool._run else None + + arun_method = getattr(tool_class, "_arun", None) + requires_default_executor = not callable(coroutine) and arun_method in { + BaseTool._arun, + StructuredTool._arun, + } + async_callable = coroutine + if not callable(async_callable) and not requires_default_executor: + async_callable = arun_method + + return RuntimeToolFinding( + configured_name=configured_name, + name=str(getattr(tool, "name", configured_name or tool_class.__name__)), + type=_type_reference(tool), + source_module=tool_class.__module__, + synchronous_function=_callable_reference(sync_function), + async_coroutine=_callable_reference(async_callable), + async_requires_default_executor=requires_default_executor, + boundary_kind=(ASYNCIO_DEFAULT_EXECUTOR if requires_default_executor else None), + source=source, + ) + + +def _inspect_model( + name: str, + model: object | type, + *, + source: str | None = None, +) -> RuntimeModelFinding: + from langchain_core.language_models.chat_models import BaseChatModel + + model_class = model if isinstance(model, type) else type(model) + agenerate_overridden = getattr(model_class, "_agenerate", None) is not BaseChatModel._agenerate + astream_overridden = getattr(model_class, "_astream", None) is not BaseChatModel._astream + inherits_fallback = not agenerate_overridden or not astream_overridden + return RuntimeModelFinding( + name=name, + type=_type_reference(model_class), + source_module=model_class.__module__, + agenerate_overridden=agenerate_overridden, + astream_overridden=astream_overridden, + inherits_base_chat_model_executor_fallback=inherits_fallback, + boundary_kind=ASYNCIO_DEFAULT_EXECUTOR if inherits_fallback else None, + source=source, + ) + + +def inspect_runtime_components( + *, + tools: Iterable[object] = (), + models: Iterable[tuple[str, object | type]] = (), +) -> RuntimeInventory: + """Inspect already-resolved components without invoking or constructing them.""" + return RuntimeInventory( + tools=[_inspect_tool(tool) for tool in tools], + models=[_inspect_model(name, model) for name, model in models], + unresolved=[], + ) + + +def inspect_configured_components( + config: object, + *, + resolve_tool: Callable[[str, type], object] | None = None, + resolve_model: Callable[[str, type], type] | None = None, +) -> RuntimeInventory: + """Resolve configured symbols and inspect them without external calls. + + Tool variables are imported because their concrete ``func``/``coroutine`` + attributes are runtime state. Model classes are imported but deliberately + not instantiated, so provider constructors and network clients never run. + Resolution failures are retained as unresolved dynamic boundaries. + """ + from langchain_core.language_models.chat_models import BaseChatModel + from langchain_core.tools import BaseTool + + if resolve_tool is None or resolve_model is None: + from deerflow.reflection import resolve_class, resolve_variable + + resolve_tool = resolve_tool or resolve_variable + resolve_model = resolve_model or resolve_class + + tool_findings: list[RuntimeToolFinding] = [] + model_findings: list[RuntimeModelFinding] = [] + unresolved: list[UnresolvedComponent] = [] + + for tool_config in getattr(config, "tools", ()): + configured_name = str(getattr(tool_config, "name", "")) + source = str(getattr(tool_config, "use", "")) + try: + tool = resolve_tool(source, BaseTool) + tool_findings.append( + _inspect_tool( + tool, + configured_name=configured_name, + source=source, + ) + ) + except Exception as exc: + unresolved.append( + UnresolvedComponent( + component_kind="tool", + name=configured_name, + source=source, + error=f"{type(exc).__name__}: {exc}", + ) + ) + + for model_config in getattr(config, "models", ()): + model_name = str(getattr(model_config, "name", "")) + source = str(getattr(model_config, "use", "")) + try: + model_class = resolve_model(source, BaseChatModel) + model_findings.append(_inspect_model(model_name, model_class, source=source)) + except Exception as exc: + unresolved.append( + UnresolvedComponent( + component_kind="model", + name=model_name, + source=source, + error=f"{type(exc).__name__}: {exc}", + ) + ) + + return RuntimeInventory( + tools=tool_findings, + models=model_findings, + unresolved=unresolved, + ) + + +def _inspect_config_file(config_path: Path) -> RuntimeInventory: + harness_path = REPO_ROOT / "backend" / "packages" / "harness" + if str(harness_path) not in sys.path: + sys.path.insert(0, str(harness_path)) + from deerflow.config.app_config import AppConfig + + config = AppConfig.from_file(str(config_path)) + return inspect_configured_components(config) + + +def build_summary( + findings: Sequence[BoundaryFinding], + runtime: RuntimeInventory | None = None, +) -> dict[str, object]: + by_boundary_kind = Counter(finding.boundary_kind for finding in findings) + by_category = Counter(finding.category for finding in findings) + return { + "static_findings": len(findings), + "by_boundary_kind": {kind: by_boundary_kind.get(kind, 0) for kind in BOUNDARY_KINDS if by_boundary_kind.get(kind, 0)}, + "by_category": dict(sorted(by_category.items())), + "runtime_tools": len(runtime.tools) if runtime else 0, + "runtime_models": len(runtime.models) if runtime else 0, + "runtime_unresolved": len(runtime.unresolved) if runtime else 0, + } + + +def build_inventory_payload( + findings: Sequence[BoundaryFinding], + runtime: RuntimeInventory | None = None, +) -> dict[str, object]: + return { + "schema_version": SCHEMA_VERSION, + "static_findings": [finding.to_dict() for finding in findings], + "runtime": runtime.to_dict() if runtime is not None else None, + "summary": build_summary(findings, runtime), + } + + +def format_summary( + findings: Sequence[BoundaryFinding], + runtime: RuntimeInventory | None = None, +) -> str: + summary = build_summary(findings, runtime) + lines = [f"Thread-boundary inventory: {summary['static_findings']} static findings"] + if runtime is not None: + lines[0] += f", {summary['runtime_tools']} configured tools, {summary['runtime_models']} configured models, {summary['runtime_unresolved']} unresolved" + boundary_counts = summary["by_boundary_kind"] + if boundary_counts: + lines.append("Execution domains:") + for kind in BOUNDARY_KINDS: + count = boundary_counts.get(kind) + if count: + lines.append(f" {kind}: {count}") + return "\n".join(lines) + + def format_text(findings: Sequence[BoundaryFinding]) -> str: if not findings: return "No async/thread boundary findings." lines: list[str] = [] for finding in findings: - lines.append(f"{finding.severity} {finding.category} {finding.path}:{finding.line}:{finding.column + 1} in {finding.function} async={str(finding.async_context).lower()}") + lines.append(f"{finding.severity} {finding.category} [{finding.boundary_kind}] {finding.path}:{finding.line}:{finding.column + 1} in {finding.function} async={str(finding.async_context).lower()}") lines.append(f" symbol: {finding.symbol}") lines.append(f" note: {finding.message}") if finding.code: @@ -476,7 +1061,7 @@ def format_text(findings: Sequence[BoundaryFinding]) -> str: def build_parser() -> argparse.ArgumentParser: - parser = argparse.ArgumentParser(description=("Detect async/thread boundary points for developer review. Findings are an inventory, not automatic bug decisions.")) + parser = argparse.ArgumentParser(description=("Inventory backend thread/executor/event-loop boundaries. Findings are evidence, not automatic bug decisions.")) parser.add_argument( "paths", nargs="*", @@ -485,15 +1070,25 @@ def build_parser() -> argparse.ArgumentParser: ) parser.add_argument( "--format", - choices=("text", "json"), - default="text", - help="Output format.", + choices=("summary", "text", "json"), + default="summary", + help="Console output format (default: concise summary).", ) parser.add_argument( "--min-severity", choices=tuple(SEVERITY_ORDER), default="INFO", - help="Only show findings at or above this severity.", + help="Only include findings at or above this severity.", + ) + parser.add_argument( + "--runtime-config", + type=Path, + help=("Inspect tools and model classes configured in this AppConfig YAML. Components are never invoked and models are never instantiated."), + ) + parser.add_argument( + "--json-output", + type=Path, + help="Also write the complete versioned inventory JSON to this path.", ) return parser @@ -503,9 +1098,23 @@ def main(argv: Sequence[str] | None = None) -> int: args = parser.parse_args(argv) paths = args.paths or list(DEFAULT_SCAN_PATHS) findings = filter_findings(scan_paths(paths), args.min_severity) + runtime = _inspect_config_file(args.runtime_config) if args.runtime_config is not None else None + payload = build_inventory_payload(findings, runtime) + + if args.json_output is not None: + args.json_output.parent.mkdir(parents=True, exist_ok=True) + args.json_output.write_text( + json.dumps(payload, indent=2, sort_keys=True) + "\n", + encoding="utf-8", + ) if args.format == "json": - print(json.dumps([finding.to_dict() for finding in findings], indent=2, sort_keys=True)) - else: + print(json.dumps(payload, indent=2, sort_keys=True)) + elif args.format == "text": print(format_text(findings)) + if runtime is not None: + print() + print(format_summary(findings, runtime)) + else: + print(format_summary(findings, runtime)) return 0 diff --git a/backend/tests/test_detect_thread_boundaries.py b/backend/tests/test_detect_thread_boundaries.py index cc48eac38..71b8a3a66 100644 --- a/backend/tests/test_detect_thread_boundaries.py +++ b/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): diff --git a/backend/tests/test_detector_repo_root.py b/backend/tests/test_detector_repo_root.py index 340345735..c1e8a9584 100644 --- a/backend/tests/test_detector_repo_root.py +++ b/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: