diff --git a/docs/refactor/backend-architecture-next-stage.md b/docs/refactor/backend-architecture-next-stage.md index ee17870ed..f91ce0b32 100644 --- a/docs/refactor/backend-architecture-next-stage.md +++ b/docs/refactor/backend-architecture-next-stage.md @@ -966,19 +966,15 @@ scanner，strict 清单扩大到 26 个源文件；已登记范围保持零错 **实施记录（2026-08-21）**： -- 新增 `scripts/architecture/async_blocking.py`，AST 扫描 `app/api`、`app/agent`、`app/application` 的 - async 函数，覆盖直接 `open`/Path 读写遍历、`requests`、`time.sleep`、同步 `subprocess` 和目录遍历。 -- scanner 通过局部类型流识别 `aiofiles` 与 `anyio.AsyncPath`，不会把正确异步 I/O 记成债务；baseline - 只允许调用减少/删除，新增或次数增长均在 CI architecture job 失败。 -- Web Agent 上传已从 `Path.open/write/unlink` 改为 `aiofiles` 写入和统一 `run_in_threadpool` 清理；当前仅保留 - ActivityLog 为保证 `O_EXCL` 原子创建使用的一处 `os.open` 精确债务，不泛化豁免整个文件或目录。 +- `scripts/architecture/async_blocking.py` 扫描 canonical 主程序目录和顶层运行入口中的 async 函数，覆盖 + 同步 HTTP、Oper、Path、`shutil`、`subprocess`、`os`、`time.sleep` 与 `open`。 +- scanner 按 import 来源、局部别名、互斥分支和嵌套函数定义点解析符号；`AsyncRequestUtils`、 + `anyio.Path`、延迟回调及受控 worker 内执行的同步函数不记为 async 直接阻塞。函数和 lambda 的默认值、 + decorator 等定义时表达式仍在所在 async 执行体中检查。 +- baseline 只允许调用减少或删除，新增调用及次数增长均使 CI architecture job 失败；当前记录 10 条已确认 + 存量，包括 8 条文件元数据访问、1 条目录删除和 1 条同步 Oper 读取，由后续数据库与文件 adapter 叶迁移。 - pytest 全局启用 `asyncio_debug`，专项测试验证实际 loop debug 状态；AST ratchet 与 46 个 Agent 流式回归 通过。同步第三方 Module 仍由 dispatcher 的 `app.runtime.execution.run_in_threadpool` 兼容。 -- 2026-08-22 扫描范围扩大到 `app/chain`、`app/modules`、`app/startup` 与 `app/scheduler.py`；扩大后未发现 - 新存量，仍只保留 ActivityLog 的一处原子 `os.open` 精确债务，并由测试锁定扫描根目录。 -- 扫描进一步覆盖 `adapters/db/doctor/domain/foundation/monitor/runtime/schemas/workflow` 及 CLI、Command、 - Factory、Main 顶层入口，明确排除插件源码和 SDK。AsyncPath 条件派生识别已修正；Release zip 解压读取和 - ActivityLog `O_EXCL` 独占创建移入线程池，扩围后 async 阻塞 baseline 从 1 降为 0。 ## 6. 推荐执行队列 diff --git a/scripts/architecture/async_blocking.py b/scripts/architecture/async_blocking.py index d66d5a8f1..7de5f5660 100644 --- a/scripts/architecture/async_blocking.py +++ b/scripts/architecture/async_blocking.py @@ -6,7 +6,8 @@ import argparse import ast import json from collections import Counter -from collections.abc import Iterator +from collections.abc import Iterator, Sequence +from dataclasses import dataclass from pathlib import Path PROJECT_ROOT = Path(__file__).resolve().parents[2] @@ -33,159 +34,545 @@ SCAN_ROOTS = ( "app/main.py", "app/scheduler.py", ) -BLOCKING_EXACT = { - "open", - "time.sleep", - "subprocess.call", - "subprocess.check_call", - "subprocess.check_output", - "subprocess.Popen", - "subprocess.run", - "os.listdir", - "os.scandir", - "os.walk", - "requests.delete", - "requests.get", - "requests.head", - "requests.patch", - "requests.post", - "requests.put", - "requests.request", + +_SYNC_HTTP_METHODS = { + "delete_res", + "get", + "get_json", + "get_res", + "get_stream", + "post", + "post_json", + "post_res", + "put", + "put_res", + "request", + "response_manager", } -BLOCKING_ATTRIBUTES = { +_REQUESTS_METHODS = { + "delete", + "get", + "head", + "patch", + "post", + "put", + "request", +} +_PATH_IO_METHODS = { + "exists", "glob", + "is_dir", + "is_file", "iterdir", + "mkdir", "open", "read_bytes", "read_text", + "rename", "rglob", + "stat", + "unlink", "write_bytes", "write_text", } +_SHUTIL_METHODS = { + "copy", + "copyfile", + "copytree", + "make_archive", + "move", + "rmtree", + "unpack_archive", + "which", +} +_SUBPROCESS_METHODS = { + "Popen", + "call", + "check_call", + "check_output", + "run", +} +_OS_IO_METHODS = {"listdir", "scandir", "walk"} +_SYSTEM_CONFIG_MEMORY_READS = { + "app.db.oper.SystemConfigOper.all", + "app.db.oper.SystemConfigOper.get", + "app.db.oper.systemconfig.SystemConfigOper.all", + "app.db.oper.systemconfig.SystemConfigOper.get", +} -def _call_name(node: ast.Call) -> str: - """把简单名称和属性调用还原为点分文本。""" - parts = [] - target: ast.expr = node.func - while isinstance(target, ast.Attribute): - parts.append(target.attr) - target = target.value - if isinstance(target, ast.Name): - parts.append(target.id) - return ".".join(reversed(parts)) +@dataclass(frozen=True) +class _Binding: + """描述由明确 import 或局部赋值建立的符号来源。""" + + family: str + qualified_name: str + kind: str = "module" -def _root_name(expression: ast.expr) -> str | None: - """返回属性调用最左侧的变量名。""" - while isinstance(expression, ast.Attribute): - expression = expression.value - return expression.id if isinstance(expression, ast.Name) else None +_UNKNOWN_BINDING = _Binding("unknown", "", "unknown") +_BLOCKING_FAMILIES = { + "os", + "requests", + "shutil", + "subprocess", + "sync_http", + "sync_oper", + "sync_path", + "time", +} -class _AsyncPathCollector(ast.NodeVisitor): - """用局部数据流识别 anyio AsyncPath 变量及其派生值。""" +def _binding_for_qualified(qualified_name: str) -> _Binding: + """按稳定模块路径识别门禁关心的符号族。""" + if qualified_name == "app.adapters.network.http.RequestUtils": + return _Binding("sync_http", qualified_name, "class") + if qualified_name == "app.adapters.network.http.AsyncRequestUtils": + return _Binding("async_http", qualified_name, "class") + if qualified_name in {"requests.Session", "requests.sessions.Session"}: + return _Binding("sync_http", qualified_name, "class") + if qualified_name == "pathlib.Path": + return _Binding("sync_path", qualified_name, "class") + if qualified_name == "anyio.Path": + return _Binding("async_path", qualified_name, "class") + if qualified_name.startswith("app.db.oper.") and qualified_name.endswith("Oper"): + return _Binding("sync_oper", qualified_name, "class") + if qualified_name.startswith("requests."): + return _Binding("requests", qualified_name, "callable") + if qualified_name.startswith("shutil."): + return _Binding("shutil", qualified_name, "callable") + if qualified_name.startswith("subprocess."): + return _Binding("subprocess", qualified_name, "callable") + if qualified_name.startswith("os."): + return _Binding("os", qualified_name, "callable") + if qualified_name.startswith("time."): + return _Binding("time", qualified_name, "callable") + return _Binding("module", qualified_name) - def __init__(self, function: ast.AsyncFunctionDef) -> None: - """从参数注解初始化 AsyncPath 变量集合。""" - self.paths = { - argument.arg - for argument in (*function.args.posonlyargs, *function.args.args) - if argument.annotation and "AsyncPath" in ast.unparse(argument.annotation) - } - self.path_collections: set[str] = set() + +def _resolve_binding( + expression: ast.expr, + bindings: dict[str, _Binding], +) -> _Binding | None: + """解析明确 import、构造、属性访问和简单路径派生。""" + if isinstance(expression, ast.Name): + return bindings.get(expression.id) + if isinstance(expression, ast.Attribute): + base = _resolve_binding(expression.value, bindings) + if not base or base.kind in {"collection", "unknown"}: + return None + qualified_name = f"{base.qualified_name}.{expression.attr}" + if base.kind == "module": + return _binding_for_qualified(qualified_name) + return _Binding(base.family, qualified_name, "callable") + if isinstance(expression, ast.Call): + target = _resolve_binding(expression.func, bindings) + if target and target.kind == "class": + return _Binding(target.family, target.qualified_name, "instance") + return None + if isinstance(expression, ast.BinOp) and isinstance(expression.op, ast.Div): + left = _resolve_binding(expression.left, bindings) + if left and left.family in {"async_path", "sync_path"}: + return left + if isinstance(expression, ast.IfExp): + return _merge_binding_options( + "", + ( + _resolve_binding(expression.body, bindings), + _resolve_binding(expression.orelse, bindings), + ), + ) + return None + + +def _merge_binding_options( + name: str, + options: Sequence[_Binding | None], +) -> _Binding | None: + """控制流合流时保留任一分支可能进入的同步阻塞类型。""" + if options and all(option == options[0] for option in options): + return options[0] + blocking = sorted( + ( + option + for option in options + if option and option.family in _BLOCKING_FAMILIES + ), + key=lambda item: (item.family, item.qualified_name, item.kind), + ) + if blocking: + return blocking[0] + if name == "open" and any(option is None for option in options): + return None + return _UNKNOWN_BINDING + + +def _merge_binding_states( + states: Sequence[dict[str, _Binding]], +) -> dict[str, _Binding]: + """合并互斥控制流的局部符号表。""" + names = set().union(*(state.keys() for state in states)) + merged: dict[str, _Binding] = {} + for name in names: + binding = _merge_binding_options(name, tuple(state.get(name) for state in states)) + if binding: + merged[name] = binding + return merged + + +def _load_oper_methods(root: Path) -> dict[str, set[str]]: + """从 Oper 源码建立同步方法索引，避免按方法名猜测数据库调用。""" + methods: dict[str, set[str]] = {} + oper_root = root / "app/db/oper" + if not oper_root.is_dir(): + return methods + for path in sorted(oper_root.glob("*.py")): + tree = ast.parse(path.read_text(encoding="utf-8"), filename=str(path)) + module = ".".join(path.relative_to(root).with_suffix("").parts) + for node in tree.body: + if not isinstance(node, ast.ClassDef) or not node.name.endswith("Oper"): + continue + qualified_name = f"{module}.{node.name}" + sync_methods = { + child.name + for child in node.body + if isinstance(child, ast.FunctionDef) + and not child.name.startswith("__") + } + methods[qualified_name] = sync_methods + methods[f"app.db.oper.{node.name}"] = sync_methods + return methods + + +class _ImportCollector(ast.NodeVisitor): + """收集模块级 import，作为每个 async 函数的初始符号表。""" + + def __init__(self) -> None: + self.bindings: dict[str, _Binding] = {} + + def visit_Import(self, node: ast.Import) -> None: + for alias in node.names: + local_name = alias.asname or alias.name.split(".", 1)[0] + qualified_name = alias.name if alias.asname else local_name + self.bindings[local_name] = _binding_for_qualified(qualified_name) + + def visit_ImportFrom(self, node: ast.ImportFrom) -> None: + if node.level or not node.module: + return + for alias in node.names: + if alias.name == "*": + continue + local_name = alias.asname or alias.name + qualified_name = f"{node.module}.{alias.name}" + self.bindings[local_name] = _binding_for_qualified(qualified_name) + + def _bind_target(self, target: ast.expr, binding: _Binding | None) -> None: + if isinstance(target, ast.Name): + self.bindings[target.id] = binding or _UNKNOWN_BINDING + elif isinstance(target, (ast.Tuple, ast.List)): + for element in target.elts: + self._bind_target(element, None) def visit_Assign(self, node: ast.Assign) -> None: - """识别 AsyncPath 构造和已知路径的 `/` 派生赋值。""" - is_async_path = self._is_async_path_expression(node.value) - if is_async_path: - for target in node.targets: - if isinstance(target, ast.Name): - self.paths.add(target.id) - self.generic_visit(node) - - def _is_async_path_expression(self, expression: ast.expr) -> bool: - """识别 AsyncPath 构造及任意层级的 `/` 路径派生表达式。""" - if isinstance(expression, ast.Call): - return _call_name(expression).endswith("AsyncPath") - if isinstance(expression, ast.Name): - return expression.id in self.paths - if isinstance(expression, ast.BinOp): - return self._is_async_path_expression(expression.left) - if isinstance(expression, ast.IfExp): - return ( - self._is_async_path_expression(expression.body) - and self._is_async_path_expression(expression.orelse) - ) - return False + binding = _resolve_binding(node.value, self.bindings) + for target in node.targets: + self._bind_target(target, binding) def visit_AnnAssign(self, node: ast.AnnAssign) -> None: - """识别 `list[AsyncPath]` 等路径集合。""" - if isinstance(node.target, ast.Name): - annotation = ast.unparse(node.annotation) - if "AsyncPath" in annotation: - if "list" in annotation or "List" in annotation: - self.path_collections.add(node.target.id) - else: - self.paths.add(node.target.id) - self.generic_visit(node) - - def visit_AsyncFor(self, node: ast.AsyncFor) -> None: - """AsyncPath.iterdir 产出的元素仍是 AsyncPath。""" - if isinstance(node.target, ast.Name) and isinstance(node.iter, ast.Call): - receiver = _root_name(node.iter.func) - if receiver in self.paths: - self.paths.add(node.target.id) - self.generic_visit(node) - - def visit_For(self, node: ast.For) -> None: - """从 `list[AsyncPath]` 迭代得到的元素仍是 AsyncPath。""" - if ( - isinstance(node.target, ast.Name) - and isinstance(node.iter, ast.Name) - and node.iter.id in self.path_collections - ): - self.paths.add(node.target.id) - self.generic_visit(node) + binding = _resolve_binding(node.value, self.bindings) if node.value else None + self._bind_target(node.target, binding) def visit_FunctionDef(self, node: ast.FunctionDef) -> None: - """不分析嵌套同步函数。""" + return def visit_AsyncFunctionDef(self, node: ast.AsyncFunctionDef) -> None: - """不分析嵌套异步函数。""" + return + + def visit_ClassDef(self, node: ast.ClassDef) -> None: + return class _AsyncCallVisitor(ast.NodeVisitor): - """只收集一个 async 函数本体中的阻塞调用，不进入嵌套函数定义。""" + """用局部符号传播识别一个 async 函数直接执行的阻塞调用。""" - def __init__(self, async_paths: set[str]) -> None: - """初始化违规计数器和异步文件对象白名单。""" + def __init__( + self, + function: ast.FunctionDef | ast.AsyncFunctionDef, + module_bindings: dict[str, _Binding], + oper_methods: dict[str, set[str]], + *, + record_calls: bool, + ) -> None: self.calls: Counter[str] = Counter() - self._async_paths = async_paths + self.nested_functions: list[ + tuple[ast.FunctionDef | ast.AsyncFunctionDef, dict[str, _Binding]] + ] = [] + self._bindings = dict(module_bindings) + self._oper_methods = oper_methods + self._record_calls = record_calls + self._bind_arguments(function) + + def _bind_arguments( + self, + function: ast.FunctionDef | ast.AsyncFunctionDef, + ) -> None: + arguments = ( + *function.args.posonlyargs, + *function.args.args, + *function.args.kwonlyargs, + ) + for argument in arguments: + binding = self._annotation_binding(argument.annotation) + if binding: + self._bindings[argument.arg] = binding + else: + self._bindings[argument.arg] = _UNKNOWN_BINDING + + def _annotation_binding(self, annotation: ast.expr | None) -> _Binding | None: + if annotation is None: + return None + if isinstance(annotation, ast.Subscript): + container = ast.unparse(annotation.value).rsplit(".", 1)[-1] + elements = ( + annotation.slice.elts + if isinstance(annotation.slice, ast.Tuple) + else (annotation.slice,) + ) + for element in elements: + binding = self._annotation_binding(element) + if not binding: + continue + if container in {"list", "List", "Sequence", "set", "tuple"}: + return _Binding( + binding.family, + binding.qualified_name, + "collection", + ) + if container in {"Annotated", "Optional", "Union"}: + return binding + return None + if isinstance(annotation, ast.BinOp) and isinstance(annotation.op, ast.BitOr): + return self._annotation_binding(annotation.left) or self._annotation_binding( + annotation.right + ) + binding = _resolve_binding(annotation, self._bindings) + if binding and binding.kind == "class": + return _Binding(binding.family, binding.qualified_name, "instance") + return None + + def _resolve(self, expression: ast.expr) -> _Binding | None: + return _resolve_binding(expression, self._bindings) + + @staticmethod + def _call_label(binding: _Binding) -> str: + parts = binding.qualified_name.split(".") + if binding.family == "sync_oper" and len(parts) >= 2: + return ".".join(parts[-2:]) + if binding.family in {"async_path", "sync_path"} and parts: + return f"Path.{parts[-1]}" + if binding.family == "sync_http" and len(parts) >= 2: + return ".".join(parts[-2:]) + if binding.family in {"os", "requests", "shutil", "subprocess", "time"}: + return ".".join(parts[-2:]) + return binding.qualified_name + + def _record_call(self, node: ast.Call, binding: _Binding | None) -> None: + if not self._record_calls: + return + if not binding: + if isinstance(node.func, ast.Name) and node.func.id == "open": + self.calls["open"] += 1 + return + method = binding.qualified_name.rsplit(".", 1)[-1] + blocked = False + if binding.family == "sync_http": + blocked = binding.kind == "callable" and method in _SYNC_HTTP_METHODS + elif binding.family == "requests": + blocked = method in _REQUESTS_METHODS + elif binding.family == "sync_path": + blocked = binding.kind == "callable" and method in _PATH_IO_METHODS + elif binding.family == "shutil": + blocked = method in _SHUTIL_METHODS + elif binding.family == "subprocess": + blocked = method in _SUBPROCESS_METHODS + elif binding.family == "os": + blocked = method in _OS_IO_METHODS + elif binding.family == "time": + blocked = method == "sleep" + elif binding.family == "sync_oper" and binding.kind == "callable": + class_name, method_name = binding.qualified_name.rsplit(".", 1) + blocked = ( + method_name in self._oper_methods.get(class_name, set()) + and binding.qualified_name not in _SYSTEM_CONFIG_MEMORY_READS + ) + if blocked: + self.calls[self._call_label(binding)] += 1 + + def _bind_target(self, target: ast.expr, binding: _Binding | None) -> None: + if isinstance(target, ast.Name): + self._bindings[target.id] = binding or _UNKNOWN_BINDING + elif isinstance(target, (ast.Tuple, ast.List)): + for element in target.elts: + self._bind_target(element, None) + + def visit_Import(self, node: ast.Import) -> None: + collector = _ImportCollector() + collector.visit_Import(node) + self._bindings.update(collector.bindings) + + def visit_ImportFrom(self, node: ast.ImportFrom) -> None: + collector = _ImportCollector() + collector.visit_ImportFrom(node) + self._bindings.update(collector.bindings) + + def visit_Assign(self, node: ast.Assign) -> None: + self.visit(node.value) + binding = self._resolve(node.value) + for target in node.targets: + self._bind_target(target, binding) + + def visit_AnnAssign(self, node: ast.AnnAssign) -> None: + if node.value: + self.visit(node.value) + binding = self._resolve(node.value) if node.value else None + binding = binding or self._annotation_binding(node.annotation) + self._bind_target(node.target, binding) + + def visit_AugAssign(self, node: ast.AugAssign) -> None: + self.visit(node.value) + self._bind_target(node.target, None) + + def _visit_branch( + self, + statements: Sequence[ast.stmt], + initial: dict[str, _Binding], + ) -> dict[str, _Binding]: + saved = self._bindings + self._bindings = dict(initial) + for statement in statements: + self.visit(statement) + result = self._bindings + self._bindings = saved + return result + + def visit_If(self, node: ast.If) -> None: + self.visit(node.test) + initial = dict(self._bindings) + body = self._visit_branch(node.body, initial) + alternate = self._visit_branch(node.orelse, initial) if node.orelse else initial + self._bindings = _merge_binding_states((body, alternate)) + + def visit_For(self, node: ast.For | ast.AsyncFor) -> None: + self.visit(node.iter) + initial = dict(self._bindings) + collection = self._resolve(node.iter) + element = ( + _Binding(collection.family, collection.qualified_name, "instance") + if collection and collection.kind == "collection" + else None + ) + self._bind_target(node.target, element) + for statement in node.body: + self.visit(statement) + iterated = dict(self._bindings) + completed = self._visit_branch(node.orelse, iterated) + self._bindings = _merge_binding_states((initial, completed)) + + def visit_AsyncFor(self, node: ast.AsyncFor) -> None: + self.visit_For(node) + + def _visit_comprehension( + self, + generators: Sequence[ast.comprehension], + outputs: Sequence[ast.expr], + ) -> None: + saved = self._bindings + self._bindings = dict(saved) + for generator in generators: + self.visit(generator.iter) + collection = self._resolve(generator.iter) + element = ( + _Binding(collection.family, collection.qualified_name, "instance") + if collection and collection.kind == "collection" + else None + ) + self._bind_target(generator.target, element) + for condition in generator.ifs: + self.visit(condition) + for output in outputs: + self.visit(output) + self._bindings = saved + + def visit_ListComp(self, node: ast.ListComp) -> None: + self._visit_comprehension(node.generators, (node.elt,)) + + def visit_SetComp(self, node: ast.SetComp) -> None: + self._visit_comprehension(node.generators, (node.elt,)) + + def visit_GeneratorExp(self, node: ast.GeneratorExp) -> None: + self._visit_comprehension(node.generators, (node.elt,)) + + def visit_DictComp(self, node: ast.DictComp) -> None: + self._visit_comprehension(node.generators, (node.key, node.value)) + + def visit_NamedExpr(self, node: ast.NamedExpr) -> None: + self.visit(node.value) + self._bind_target(node.target, self._resolve(node.value)) def visit_Call(self, node: ast.Call) -> None: - """记录命中阻塞名单的调用并继续遍历参数表达式。""" - name = _call_name(node) - attribute = name.rsplit(".", 1)[-1] - receiver = _root_name(node.func) - async_safe = name.startswith("aiofiles.") or receiver in self._async_paths - if not async_safe and ( - name in BLOCKING_EXACT or attribute in BLOCKING_ATTRIBUTES - ): - self.calls[name or attribute] += 1 + binding = self._resolve(node.func) + self._record_call(node, binding) + if isinstance(node.func, ast.Lambda): + self._visit_lambda_defaults(node.func) + for argument in node.args: + self.visit(argument) + for keyword in node.keywords: + self.visit(keyword.value) + self.visit(node.func.body) + return self.generic_visit(node) + def _visit_definition_expressions( + self, + node: ast.FunctionDef | ast.AsyncFunctionDef, + ) -> None: + for decorator in node.decorator_list: + self.visit(decorator) + for default in node.args.defaults: + self.visit(default) + for keyword_default in node.args.kw_defaults: + if keyword_default: + self.visit(keyword_default) + + def _visit_nested_function( + self, + node: ast.FunctionDef | ast.AsyncFunctionDef, + ) -> None: + self._visit_definition_expressions(node) + self.nested_functions.append((node, dict(self._bindings))) + self._bindings[node.name] = _UNKNOWN_BINDING + def visit_FunctionDef(self, node: ast.FunctionDef) -> None: - """嵌套同步函数不属于外层 async 的直接执行体。""" + self._visit_nested_function(node) def visit_AsyncFunctionDef(self, node: ast.AsyncFunctionDef) -> None: - """嵌套异步函数由模块级收集器单独治理。""" + self._visit_nested_function(node) + + def _visit_lambda_defaults(self, node: ast.Lambda) -> None: + for default in node.args.defaults: + self.visit(default) + for keyword_default in node.args.kw_defaults: + if keyword_default: + self.visit(keyword_default) + + def visit_Lambda(self, node: ast.Lambda) -> None: + self._visit_lambda_defaults(node) -def _async_functions( +def _root_functions( tree: ast.Module, ) -> Iterator[tuple[str, ast.AsyncFunctionDef]]: - """产出模块顶层及类直接拥有的 async 函数限定名。""" + """产出模块顶层和类直接拥有的 async 入口。""" for node in tree.body: if isinstance(node, ast.AsyncFunctionDef): yield node.name, node @@ -195,24 +582,63 @@ def _async_functions( yield f"{node.name}.{method.name}", method -def collect_async_blocking(root: Path = PROJECT_ROOT) -> dict[str, int]: +def _scan_paths(root: Path, scan_roots: Sequence[str | Path]) -> Iterator[Path]: + """按稳定顺序产出存在的扫描目标。""" + for scan_root in scan_roots: + target = root / scan_root + if target.is_file(): + yield target + elif target.is_dir(): + yield from sorted(target.rglob("*.py")) + + +def collect_async_blocking( + root: Path = PROJECT_ROOT, + scan_roots: Sequence[str | Path] = SCAN_ROOTS, +) -> dict[str, int]: """扫描关键目录并以文件、函数、调用名聚合存量次数。""" debt: Counter[str] = Counter() - for scan_root in SCAN_ROOTS: - target = root / scan_root - paths = [target] if target.is_file() else sorted(target.rglob("*.py")) - for path in paths: - tree = ast.parse(path.read_text(encoding="utf-8"), filename=str(path)) - relative = path.relative_to(root).as_posix() - for qualname, function in _async_functions(tree): - collector = _AsyncPathCollector(function) - for statement in function.body: - collector.visit(statement) - visitor = _AsyncCallVisitor(collector.paths) - for statement in function.body: - visitor.visit(statement) + oper_methods = _load_oper_methods(root) + for path in _scan_paths(root, scan_roots): + tree = ast.parse(path.read_text(encoding="utf-8"), filename=str(path)) + imports = _ImportCollector() + for statement in tree.body: + imports.visit(statement) + relative = path.relative_to(root).as_posix() + pending: list[ + tuple[ + str, + ast.FunctionDef | ast.AsyncFunctionDef, + dict[str, _Binding], + ] + ] = [ + (qualname, function, imports.bindings) + for qualname, function in _root_functions(tree) + ] + pending_index = 0 + while pending_index < len(pending): + qualname, function, lexical_bindings = pending[pending_index] + pending_index += 1 + is_async = isinstance(function, ast.AsyncFunctionDef) + visitor = _AsyncCallVisitor( + function, + lexical_bindings, + oper_methods, + record_calls=is_async, + ) + for statement in function.body: + visitor.visit(statement) + if is_async: for call_name, count in visitor.calls.items(): debt[f"{relative}:{qualname}:{call_name}"] += count + pending.extend( + ( + f"{qualname}.{nested.name}", + nested, + bindings, + ) + for nested, bindings in visitor.nested_functions + ) return dict(sorted(debt.items())) diff --git a/tests/fixtures/architecture/async-blocking-baseline.json b/tests/fixtures/architecture/async-blocking-baseline.json index 0967ef424..22e985d53 100644 --- a/tests/fixtures/architecture/async-blocking-baseline.json +++ b/tests/fixtures/architecture/async-blocking-baseline.json @@ -1 +1,12 @@ -{} +{ + "app/agent/tools/impl/_plugin_tool_utils.py:uninstall_plugin_runtime:shutil.rmtree": 1, + "app/agent/tools/impl/scrape_metadata.py:ScrapeMetadataTool.run:Path.exists": 1, + "app/agent/tools/impl/scrape_metadata.py:ScrapeMetadataTool.run:Path.is_dir": 1, + "app/chain/media.py:MediaChain._async_music_album_dir_fallback:Path.exists": 1, + "app/chain/media.py:MediaChain._async_music_album_dir_fallback:Path.is_file": 1, + "app/chain/media.py:MediaChain.async_recognize_music_album_directory:Path.is_dir": 1, + "app/modules/acoustid/__init__.py:AcoustIdModule.async_identify_music_by_fingerprint:Path.is_file": 1, + "app/modules/discord/discord.py:Discord._send_file:Path.exists": 1, + "app/modules/discord/discord.py:Discord._send_file:Path.is_file": 1, + "app/scheduler.py:Scheduler.execute_agent_task:AgentTaskOper.get": 1 +} diff --git a/tests/test_async_blocking_gate.py b/tests/test_async_blocking_gate.py index 2a1d468ef..bfbc10a46 100644 --- a/tests/test_async_blocking_gate.py +++ b/tests/test_async_blocking_gate.py @@ -1,10 +1,33 @@ """async 阻塞调用 ratchet 与 debug 模式测试。""" import asyncio +import textwrap +from pathlib import Path import pytest -from scripts.architecture.async_blocking import SCAN_ROOTS, compare_async_blocking +from scripts.architecture.async_blocking import ( + SCAN_ROOTS, + collect_async_blocking, + compare_async_blocking, +) + + +def _scan_source( + tmp_path: Path, + source: str, + *, + oper_sources: dict[str, str] | None = None, +) -> dict[str, int]: + """构造最小仓库并通过公开扫描入口验证源码，而非测试内部 AST 细节。""" + api_root = tmp_path / "app/api" + api_root.mkdir(parents=True) + (api_root / "sample.py").write_text(textwrap.dedent(source), encoding="utf-8") + for filename, oper_source in (oper_sources or {}).items(): + oper_path = tmp_path / "app/db/oper" / filename + oper_path.parent.mkdir(parents=True, exist_ok=True) + oper_path.write_text(textwrap.dedent(oper_source), encoding="utf-8") + return collect_async_blocking(tmp_path, scan_roots=("app/api",)) def test_async_blocking_scan_covers_runtime_entrypoints() -> None: @@ -51,6 +74,276 @@ def test_async_blocking_ratchet_allows_removal_and_rejects_growth() -> None: assert any("新增" in problem for problem in problems) +@pytest.mark.parametrize( + ("source", "expected"), + [ + ( + """ + from app.adapters.network.http import RequestUtils as RU + + async def load(): + client = RU() + alias = client + return alias.get_res("https://example.com") + """, + {"app/api/sample.py:load:RequestUtils.get_res": 1}, + ), + ( + """ + import app.adapters.network.http as http + + async def submit(): + return http.RequestUtils().post_res("https://example.com") + """, + {"app/api/sample.py:submit:RequestUtils.post_res": 1}, + ), + ( + """ + from pathlib import Path as P + + async def inspect_file(): + path = P("payload") / "item.json" + return path.exists() + """, + {"app/api/sample.py:inspect_file:Path.exists": 1}, + ), + ( + """ + import shutil as files + from subprocess import run as run_process + + async def cleanup(): + files.rmtree("payload") + run_process(["true"]) + """, + { + "app/api/sample.py:cleanup:shutil.rmtree": 1, + "app/api/sample.py:cleanup:subprocess.run": 1, + }, + ), + ( + """ + import os as operating_system + import time as clock + from requests import Session as HttpSession + + async def legacy_io(): + open("payload") + operating_system.listdir(".") + clock.sleep(0.1) + HttpSession().get("https://example.com") + """, + { + "app/api/sample.py:legacy_io:Session.get": 1, + "app/api/sample.py:legacy_io:open": 1, + "app/api/sample.py:legacy_io:os.listdir": 1, + "app/api/sample.py:legacy_io:time.sleep": 1, + }, + ), + ( + """ + from pathlib import Path + + async def outer(): + async def inner(): + return Path("payload").read_text() + return await inner() + """, + {"app/api/sample.py:outer.inner:Path.read_text": 1}, + ), + ( + """ + from pathlib import Path + + async def inspect_files(files: list[Path]): + for file in files: + if file.is_file(): + return file + return None + """, + {"app/api/sample.py:inspect_files:Path.is_file": 1}, + ), + ], +) +def test_async_blocking_scan_resolves_imports_aliases_and_nested_async( + tmp_path: Path, + source: str, + expected: dict[str, int], +) -> None: + """别名、简单局部传播和嵌套 async 都必须进入真实扫描结果。""" + assert _scan_source(tmp_path, source) == expected + + +def test_async_blocking_scan_uses_oper_source_method_kinds(tmp_path: Path) -> None: + """Oper 仅按源码中真实同步方法报告，不依赖方法名前缀猜测。""" + source = """ + from app.db.oper import SiteOper + from app.db.oper.site import SiteOper as SO + + async def load(oper: SO): + oper.list() + return await oper.get_by_id(1) + + async def load_from_facade(): + return SiteOper().list() + """ + oper_sources = { + "site.py": """ + class SiteOper: + def list(self): + return [] + + async def get_by_id(self, site_id): + return None + """, + } + + assert _scan_source(tmp_path, source, oper_sources=oper_sources) == { + "app/api/sample.py:load:SiteOper.list": 1, + "app/api/sample.py:load_from_facade:SiteOper.list": 1, + } + + +def test_async_blocking_scan_exempts_async_apis_and_memory_reads( + tmp_path: Path, +) -> None: + """异步实现、受控 worker 与内存配置读取不得成为阻塞债务。""" + source = """ + import asyncio + import anyio + import subprocess + from anyio import Path as AsyncPath + from app.agent.tools.base import run_agent_blocking + from app.adapters.network.http import AsyncRequestUtils + from app.db.oper.systemconfig import SystemConfigOper + from app.runtime.execution import run_in_threadpool as runtime_worker + from fastapi.concurrency import run_in_threadpool + + async def load(config: SystemConfigOper): + await AsyncRequestUtils().get_res("https://example.com") + await AsyncPath("payload").exists() + await run_in_threadpool(lambda: subprocess.run(["true"])) + await runtime_worker(lambda: subprocess.run(["true"])) + await run_agent_blocking("plugin", lambda: subprocess.run(["true"])) + await asyncio.to_thread(lambda: subprocess.run(["true"])) + await anyio.to_thread.run_sync(lambda: subprocess.run(["true"])) + deferred = lambda: subprocess.run(["true"]) + assert deferred + return config.get("key") + + def ordinary_sync(): + subprocess.run(["true"]) + """ + oper_sources = { + "systemconfig.py": """ + class SystemConfigOper: + def get(self, key): + return key + """, + } + + assert _scan_source(tmp_path, source, oper_sources=oper_sources) == {} + + +def test_async_blocking_scan_checks_worker_arguments_evaluated_on_loop( + tmp_path: Path, +) -> None: + """worker 调用前同步求值的普通参数仍在事件循环中执行。""" + source = """ + import asyncio + from pathlib import Path + + async def load(): + await asyncio.to_thread(print, Path("payload").read_text()) + """ + + assert _scan_source(tmp_path, source) == { + "app/api/sample.py:load:Path.read_text": 1, + } + + +def test_async_blocking_scan_merges_branch_bindings_conservatively( + tmp_path: Path, +) -> None: + """任一互斥分支可能产生同步对象时，合流调用仍属于阻塞风险。""" + source = """ + from anyio import Path as AsyncPath + from pathlib import Path + + async def load(use_sync: bool): + if use_sync: + target = Path("payload") + else: + target = AsyncPath("payload") + return target.read_text() + """ + + assert _scan_source(tmp_path, source) == { + "app/api/sample.py:load:Path.read_text": 1, + } + + +def test_nested_async_inherits_bindings_from_definition_scope(tmp_path: Path) -> None: + """嵌套 async 使用定义点已有的局部 import 和别名。""" + source = """ + async def outer(): + from pathlib import Path as LocalPath + alias = LocalPath + + async def inner(): + return alias("payload").read_text() + + return await inner() + """ + + assert _scan_source(tmp_path, source) == { + "app/api/sample.py:outer.inner:Path.read_text": 1, + } + + +def test_definition_time_expressions_remain_in_async_execution_body( + tmp_path: Path, +) -> None: + """延迟函数体不扫描，但默认值和 decorator 在定义时立即求值。""" + source = """ + import asyncio + from pathlib import Path + + def register(value): + return lambda function: function + + async def outer(): + await asyncio.to_thread( + lambda value=Path("lambda").read_text(): value + ) + + @register(Path("decorator").read_text()) + async def inner(value=Path("default").read_text()): + return value + + return await inner() + """ + + assert _scan_source(tmp_path, source) == { + "app/api/sample.py:outer:Path.read_text": 3, + } + + +def test_local_shadowing_does_not_reuse_import_or_builtin_bindings( + tmp_path: Path, +) -> None: + """参数和 comprehension target 会遮蔽同名 builtin 或导入符号。""" + source = """ + from pathlib import Path + + async def invoke(open, items): + open() + return [Path.exists() for Path in items] + """ + + assert _scan_source(tmp_path, source) == {} + + @pytest.mark.asyncio async def test_asyncio_debug_is_enabled_for_async_tests() -> None: """专项异步测试必须启用慢 callback 和阻塞诊断所需的 debug 模式。"""