fix(architecture): harden complexity and concurrency gates

scripts/architecture/complexity.py

@@ -47,46 +47,36 @@ def _iter_python_files(root: Path, roots: tuple[str, ...]) -> Iterable[Path]:
             yield candidate
 
 
-def _walk_functions(
+def _walk_owned_nodes(
     nodes: Iterable[ast.AST],
     prefix: str = "",
-) -> Iterable[tuple[str, ast.FunctionDef | ast.AsyncFunctionDef]]:
-    """递归收集函数、方法和嵌套编排器，避免只检查公开入口。"""
+) -> Iterable[tuple[str, ast.AST]]:
+    """通过完整 AST 子节点遍历收集类、方法和任意控制流中的嵌套 owner。"""
     for node in nodes:
+        if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef)):
+            qualname = f"{prefix}.{node.name}" if prefix else node.name
+            yield qualname, node
+            yield from _walk_owned_nodes(ast.iter_child_nodes(node), qualname)
+        else:
+            yield from _walk_owned_nodes(ast.iter_child_nodes(node), prefix)
+
+
+def _walk_functions(
+    nodes: Iterable[ast.AST],
+) -> Iterable[tuple[str, ast.FunctionDef | ast.AsyncFunctionDef]]:
+    """收集含 dunder、Match、TryStar 在内的全部函数与方法 owner。"""
+    for qualname, node in _walk_owned_nodes(nodes):
         if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
-            qualname = f"{prefix}.{node.name}" if prefix else node.name
-            if not node.name.startswith("__"):
-                yield qualname, node
-            yield from _walk_functions(node.body, qualname)
-        elif isinstance(node, ast.ClassDef):
-            qualname = f"{prefix}.{node.name}" if prefix else node.name
-            yield from _walk_functions(node.body, qualname)
-        elif isinstance(node, ast.If):
-            yield from _walk_functions(node.body, prefix)
-            yield from _walk_functions(node.orelse, prefix)
-        elif isinstance(node, (ast.For, ast.AsyncFor, ast.While, ast.With, ast.AsyncWith)):
-            yield from _walk_functions(node.body, prefix)
-            yield from _walk_functions(node.orelse, prefix) if isinstance(node, (ast.For, ast.AsyncFor)) else ()
-        elif isinstance(node, ast.Try):
-            yield from _walk_functions(node.body, prefix)
-            yield from _walk_functions(node.orelse, prefix)
-            yield from _walk_functions(node.finalbody, prefix)
-            for handler in node.handlers:
-                yield from _walk_functions(handler.body, prefix)
+            yield qualname, node
 
 
 def _walk_classes(
     nodes: Iterable[ast.AST],
-    prefix: str = "",
 ) -> Iterable[tuple[str, ast.ClassDef]]:
-    """递归收集类体长度，覆盖 Scheduler 和私有 owner。"""
-    for node in nodes:
+    """收集任意控制流和函数内部的类，并保留完整嵌套 owner。"""
+    for qualname, node in _walk_owned_nodes(nodes):
         if isinstance(node, ast.ClassDef):
-            qualname = f"{prefix}.{node.name}" if prefix else node.name
             yield qualname, node
-            yield from _walk_classes(node.body, qualname)
-        elif isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
-            yield from _walk_classes(node.body, prefix)
 
 
 def collect_complexity_v2(root: Path = PROJECT_ROOT) -> dict[str, dict[str, int]]:
@@ -99,12 +89,12 @@ def collect_complexity_v2(root: Path = PROJECT_ROOT) -> dict[str, dict[str, int]
         if len(source_lines) > 1000:
             report["file"][relative] = len(source_lines)
         tree = ast.parse("\n".join(source_lines), filename=str(path))
-        for qualname, node in _walk_functions(tree.body):
-            line_count = (node.end_lineno or node.lineno) - node.lineno + 1
+        for qualname, method_node in _walk_functions(tree.body):
+            line_count = (method_node.end_lineno or method_node.lineno) - method_node.lineno + 1
             if line_count > 150:
                 report["method"][f"{relative}:{qualname}"] = line_count
-        for qualname, node in _walk_classes(tree.body):
-            line_count = (node.end_lineno or node.lineno) - node.lineno + 1
+        for qualname, class_node in _walk_classes(tree.body):
+            line_count = (class_node.end_lineno or class_node.lineno) - class_node.lineno + 1
             if line_count > 500:
                 report["class"][f"{relative}:{qualname}"] = line_count
     return report

scripts/architecture/concurrency.py

@@ -7,58 +7,73 @@ import ast
 import json
 from dataclasses import dataclass
 from pathlib import Path
-from typing import Iterable
+from typing import Iterable, TypedDict, cast
 
 PROJECT_ROOT = Path(__file__).resolve().parents[2]
 DEFAULT_BASELINE = PROJECT_ROOT / "tests/fixtures/architecture/concurrency-baseline.json"
-SCAN_ROOTS = (
-    "app/adapters",
-    "app/api",
-    "app/agent",
-    "app/application",
-    "app/chain",
-    "app/command.py",
-    "app/db",
-    "app/domain",
-    "app/foundation",
-    "app/main.py",
-    "app/monitor",
-    "app/modules",
-    "app/runtime",
-    "app/scheduler",
-    "app/startup",
-    "app/workflow",
-)
+SCAN_ROOTS = ("app",)
+EXCLUDED_ROOTS = ("app/plugins", "app/sdk", "app/runtime/compat")
 
-_CALL_SUFFIXES = frozenset({
-    "create_task", "ensure_future", "to_thread", "run_in_executor", "run_in_threadpool",
+_RECORDED_CALLS = frozenset({
+    "app.runtime.execution.run_in_threadpool",
+    "app.runtime.execution.run_in_threadpool_to_completion",
+    "asyncio.TaskGroup",
+    "asyncio.create_task",
+    "asyncio.ensure_future",
+    "asyncio.to_thread",
+    "concurrent.futures.ProcessPoolExecutor",
+    "concurrent.futures.ThreadPoolExecutor",
+    "multiprocessing.Process",
+    "starlette.concurrency.run_in_threadpool",
+    "threading.Thread",
+    "threading.Timer",
 })
-_CONSTRUCTOR_NAMES = frozenset({
-    "Thread", "Timer", "ThreadPoolExecutor", "ProcessPoolExecutor", "Process",
+_LOOP_FACTORIES = frozenset({
+    "asyncio.get_event_loop",
+    "asyncio.get_running_loop",
+    "asyncio.new_event_loop",
 })
+_LOOP_TYPES = frozenset({"asyncio.AbstractEventLoop", "asyncio.BaseEventLoop"})
+_EXECUTOR_TYPES = frozenset({
+    "concurrent.futures.Executor",
+    "concurrent.futures.ProcessPoolExecutor",
+    "concurrent.futures.ThreadPoolExecutor",
+})
+_EVENT_LOOP_KIND = "asyncio.AbstractEventLoop"
+_TASK_GROUP_KIND = "asyncio.TaskGroup"
+_EXECUTOR_KIND = "concurrent.futures.Executor"
+_RECEIVER_CALLS = {
+    (_EVENT_LOOP_KIND, "create_task"): "asyncio.AbstractEventLoop.create_task",
+    (_EVENT_LOOP_KIND, "run_in_executor"): "asyncio.AbstractEventLoop.run_in_executor",
+    (_TASK_GROUP_KIND, "create_task"): "asyncio.TaskGroup.create_task",
+    (_EXECUTOR_KIND, "map"): "concurrent.futures.Executor.map",
+    (_EXECUTOR_KIND, "submit"): "concurrent.futures.Executor.submit",
+}
+
+
+class ConcurrencyRecord(TypedDict):
+    """描述一种原生并发调用在文件内的 owner 计数。"""
+
+    target: str
+    owners: dict[str, int]
 
 
 @dataclass(frozen=True, slots=True)
 class ConcurrencyFact:
-    """描述一个原生并发调用及其静态 owner。"""
+    """描述一个已确认来源的原生并发调用及其静态 owner。"""
 
     path: str
-    line: int
     target: str
     owner: str
 
     @property
     def key(self) -> str:
-        """返回跨运行稳定的事实键。"""
-        return f"{self.path}:{self.line}:{self.target}"
-
-    def as_dict(self) -> dict[str, str]:
-        """转换为基线文件中的稳定记录。"""
-        return {"owner": self.owner, "target": self.target}
+        """返回不依赖物理行号的事实键。"""
+        return f"{self.path}:{self.target}"
 
 
 def _iter_python_files(root: Path, scan_roots: Iterable[str]) -> Iterable[Path]:
-    """遍历宿主源码并排除插件副本、SDK 和兼容实现。"""
+    """遍历完整宿主源码，并只排除有独立治理契约的兼容表面。"""
     for scan_root in scan_roots:
         path = root / scan_root
         if path.is_file():
@@ -71,63 +86,203 @@ def _iter_python_files(root: Path, scan_roots: Iterable[str]) -> Iterable[Path]:
             relative = candidate.relative_to(root).as_posix()
             if any(
                 relative == excluded or relative.startswith(f"{excluded}/")
-                for excluded in ("app/plugins", "app/sdk", "app/runtime/compat")
+                for excluded in EXCLUDED_ROOTS
             ):
                 continue
             yield candidate
 
 
-def _target_name(node: ast.Call) -> str | None:
-    """识别约定的原生并发 API，避免按普通方法名误报。"""
-    if not isinstance(node.func, (ast.Name, ast.Attribute)):
-        return None
-    target = ast.unparse(node.func)
-    terminal = target.rsplit(".", 1)[-1]
-    if terminal in _CALL_SUFFIXES or terminal in _CONSTRUCTOR_NAMES:
-        return terminal
-    return None
-
-
 class _ConcurrencyVisitor(ast.NodeVisitor):
-    """收集函数、类和模块级并发调用的词法 owner。"""
+    """按 canonical import、局部来源和词法 owner 收集并发调用。"""
 
     def __init__(self, relative_path: str) -> None:
-        """初始化文件位置和 owner 栈。"""
+        """初始化文件位置、owner 栈和逐作用域名称绑定。"""
         self._relative_path = relative_path
         self._owners: list[str] = []
+        self._class_owners: list[str] = []
+        self._bindings: list[dict[str, str | None]] = [{}]
+        self._attribute_bindings: dict[tuple[str, str], str] = {}
         self.facts: list[ConcurrencyFact] = []
 
     @property
     def _owner(self) -> str:
         """返回最具体的可定位 owner。"""
-        return ".".join(self._owners) if self._owners else self._relative_path
+        return ".".join(self._owners) if self._owners else "<module>"
+
+    @property
+    def _class_owner(self) -> str:
+        """返回当前类路径，用于区分不同类的实例属性。"""
+        return ".".join(self._class_owners)
+
+    def _bind_name(self, name: str, origin: str | None) -> None:
+        """在当前作用域绑定名称；None 表示显式遮蔽外层导入。"""
+        self._bindings[-1][name] = origin
+
+    def _lookup_name(self, name: str) -> str | None:
+        """按 Python 词法作用域顺序解析名称来源。"""
+        for scope in reversed(self._bindings):
+            if name in scope:
+                return scope[name]
+        return None
+
+    def _attribute_key(self, node: ast.Attribute) -> tuple[str, str]:
+        """为实例属性生成带类 owner 的稳定绑定键。"""
+        return self._class_owner, ast.unparse(node)
+
+    def _resolve_expression(self, node: ast.AST) -> str | None:
+        """解析名称、属性或构造结果的 canonical 来源。"""
+        if isinstance(node, ast.Name):
+            return self._lookup_name(node.id)
+        if isinstance(node, ast.Attribute):
+            bound = self._attribute_bindings.get(self._attribute_key(node))
+            if bound:
+                return bound
+            base = self._resolve_expression(node.value)
+            return f"{base}.{node.attr}" if base else None
+        if isinstance(node, ast.Call):
+            target = self._resolve_expression(node.func)
+            if target in _LOOP_FACTORIES:
+                return _EVENT_LOOP_KIND
+            if target == _TASK_GROUP_KIND:
+                return _TASK_GROUP_KIND
+            if target in _EXECUTOR_TYPES:
+                return _EXECUTOR_KIND
+        return None
+
+    def _resolve_call_target(self, node: ast.Call) -> str | None:
+        """仅在调用来源可证明时返回受治理的 canonical target。"""
+        direct = self._resolve_expression(node.func)
+        if direct in _RECORDED_CALLS:
+            return direct
+        if isinstance(node.func, ast.Attribute):
+            receiver = self._resolve_expression(node.func.value)
+            return _RECEIVER_CALLS.get((receiver or "", node.func.attr))
+        return None
+
+    def _bind_target(self, target: ast.expr, origin: str | None) -> None:
+        """把赋值目标关联到可证明来源，并显式遮蔽未知名称。"""
+        if isinstance(target, ast.Name):
+            self._bind_name(target.id, origin)
+        elif isinstance(target, ast.Attribute):
+            key = self._attribute_key(target)
+            if origin:
+                self._attribute_bindings[key] = origin
+            else:
+                self._attribute_bindings.pop(key, None)
+        elif isinstance(target, (ast.Tuple, ast.List)):
+            for item in target.elts:
+                self._bind_target(item, None)
+
+    def _bind_annotated_arguments(self, node: ast.FunctionDef | ast.AsyncFunctionDef) -> None:
+        """根据参数注解标记事件循环、TaskGroup 和 Executor 来源。"""
+        arguments = [*node.args.posonlyargs, *node.args.args, *node.args.kwonlyargs]
+        if node.args.vararg:
+            arguments.append(node.args.vararg)
+        if node.args.kwarg:
+            arguments.append(node.args.kwarg)
+        for argument in arguments:
+            if argument.annotation is None:
+                continue
+            origins = {
+                origin
+                for item in ast.walk(argument.annotation)
+                if (origin := self._resolve_expression(item))
+            }
+            if origins & _LOOP_TYPES:
+                self._bind_name(argument.arg, _EVENT_LOOP_KIND)
+            elif _TASK_GROUP_KIND in origins:
+                self._bind_name(argument.arg, _TASK_GROUP_KIND)
+            elif origins & _EXECUTOR_TYPES:
+                self._bind_name(argument.arg, _EXECUTOR_KIND)
+
+    def _visit_function(self, node: ast.FunctionDef | ast.AsyncFunctionDef) -> None:
+        """进入函数 owner 和局部作用域后扫描全部子节点。"""
+        self._bind_name(node.name, None)
+        self._owners.append(node.name)
+        self._bindings.append({})
+        self._bind_annotated_arguments(node)
+        self.generic_visit(node)
+        self._bindings.pop()
+        self._owners.pop()
+
+    def visit_Import(self, node: ast.Import) -> None:
+        """登记模块导入及其别名。"""
+        for alias in node.names:
+            bound_name = alias.asname or alias.name.split(".", 1)[0]
+            origin = alias.name if alias.asname else bound_name
+            self._bind_name(bound_name, origin)
+
+    def visit_ImportFrom(self, node: ast.ImportFrom) -> None:
+        """登记直接导入符号及其别名。"""
+        if not node.module:
+            return
+        for alias in node.names:
+            if alias.name == "*":
+                continue
+            self._bind_name(alias.asname or alias.name, f"{node.module}.{alias.name}")
 
     def visit_ClassDef(self, node: ast.ClassDef) -> None:
-        """把类名纳入 owner 路径后继续扫描方法体。"""
+        """把类名纳入 owner 路径并隔离类体名称绑定。"""
+        self._bind_name(node.name, None)
         self._owners.append(node.name)
+        self._class_owners.append(node.name)
+        self._bindings.append({})
         self.generic_visit(node)
+        self._bindings.pop()
+        self._class_owners.pop()
         self._owners.pop()
 
     def visit_FunctionDef(self, node: ast.FunctionDef) -> None:
-        """把同步函数纳入 owner 路径后扫描其嵌套并发。"""
-        self._owners.append(node.name)
-        self.generic_visit(node)
-        self._owners.pop()
+        """扫描同步函数及其嵌套 owner。"""
+        self._visit_function(node)
 
     def visit_AsyncFunctionDef(self, node: ast.AsyncFunctionDef) -> None:
-        """把异步函数纳入 owner 路径后扫描其嵌套并发。"""
-        self._owners.append(node.name)
-        self.generic_visit(node)
-        self._owners.pop()
+        """扫描异步函数及其嵌套 owner。"""
+        self._visit_function(node)
+
+    def visit_Assign(self, node: ast.Assign) -> None:
+        """记录赋值右侧调用并传播可证明的运行时来源。"""
+        self.visit(node.value)
+        origin = self._resolve_expression(node.value)
+        for target in node.targets:
+            self._bind_target(target, origin)
+
+    def visit_AnnAssign(self, node: ast.AnnAssign) -> None:
+        """处理带注解赋值并传播右侧来源。"""
+        self.visit(node.annotation)
+        if node.value:
+            self.visit(node.value)
+        self._bind_target(node.target, self._resolve_expression(node.value) if node.value else None)
+
+    def visit_NamedExpr(self, node: ast.NamedExpr) -> None:
+        """处理海象表达式中的来源传播。"""
+        self.visit(node.value)
+        self._bind_target(node.target, self._resolve_expression(node.value))
+
+    def _visit_with(self, node: ast.With | ast.AsyncWith) -> None:
+        """先解析上下文管理器，再让其绑定在语句体内可见。"""
+        for item in node.items:
+            self.visit(item.context_expr)
+            if item.optional_vars:
+                self._bind_target(item.optional_vars, self._resolve_expression(item.context_expr))
+        for statement in node.body:
+            self.visit(statement)
+
+    def visit_With(self, node: ast.With) -> None:
+        """扫描同步上下文管理器及其绑定。"""
+        self._visit_with(node)
+
+    def visit_AsyncWith(self, node: ast.AsyncWith) -> None:
+        """扫描异步上下文管理器及其绑定。"""
+        self._visit_with(node)
 
     def visit_Call(self, node: ast.Call) -> None:
-        """记录已识别原语并继续访问参数中的嵌套调用。"""
-        target = _target_name(node)
+        """记录来源明确的原语并继续访问参数中的嵌套调用。"""
+        target = self._resolve_call_target(node)
         if target:
             self.facts.append(
                 ConcurrencyFact(
                     path=self._relative_path,
-                    line=node.lineno,
                     target=target,
                     owner=self._owner,
                 )
@@ -139,36 +294,54 @@ def collect_concurrency(
     root: Path = PROJECT_ROOT,
     *,
     scan_roots: Iterable[str] = SCAN_ROOTS,
-) -> dict[str, dict[str, str]]:
-    """收集全部宿主原生并发调用及静态 owner。"""
-    facts: dict[str, dict[str, str]] = {}
+) -> dict[str, ConcurrencyRecord]:
+    """按文件、canonical target 和 owner 计数收集原生并发事实。"""
+    records: dict[str, ConcurrencyRecord] = {}
     for path in _iter_python_files(root, scan_roots):
         relative = path.relative_to(root).as_posix()
         tree = ast.parse(path.read_text(encoding="utf-8-sig"), filename=str(path))
         visitor = _ConcurrencyVisitor(relative)
         visitor.visit(tree)
         for fact in visitor.facts:
-            facts[fact.key] = fact.as_dict()
-    return dict(sorted(facts.items()))
+            record = records.setdefault(fact.key, {"target": fact.target, "owners": {}})
+            record["owners"][fact.owner] = record["owners"].get(fact.owner, 0) + 1
+    return dict(sorted(records.items()))
 
 
 def compare_concurrency(
-    baseline: dict[str, dict[str, str]],
-    current: dict[str, dict[str, str]],
+    baseline: dict[str, ConcurrencyRecord],
+    current: dict[str, ConcurrencyRecord],
 ) -> list[str]:
-    """拒绝新增原语、owner 丢失或事实 owner 漂移。"""
+    """拒绝新增/增长，并要求事实减少时同步下调低水位基线。"""
     problems: list[str] = []
-    for key, record in current.items():
-        if not record.get("owner"):
-            problems.append(f"并发事实缺少 owner：{key}")
-            continue
+    for key in sorted(baseline.keys() | current.keys()):
         previous = baseline.get(key)
-        if previous is None:
-            problems.append(f"新增原生并发事实：{key}")
-        elif previous.get("owner") != record.get("owner"):
-            problems.append(
-                f"并发 owner 漂移：{key}={record.get('owner')}（基线为 {previous.get('owner')}）"
-            )
+        record = current.get(key)
+        if previous is None and record is not None:
+            problems.append(f"新增原生并发事实：{key} owners={record['owners']}")
+            continue
+        if record is None and previous is not None:
+            problems.append(f"并发低水位已下降，请刷新基线：{key}")
+            continue
+        if previous is None or record is None:
+            continue
+        if record.get("target") != previous.get("target"):
+            problems.append(f"并发 target 漂移：{key}")
+        previous_owners = previous.get("owners", {})
+        current_owners = record.get("owners", {})
+        for owner in sorted(previous_owners.keys() | current_owners.keys()):
+            old_count = previous_owners.get(owner, 0)
+            new_count = current_owners.get(owner, 0)
+            if not owner:
+                problems.append(f"并发事实缺少 owner：{key}")
+            elif new_count > old_count:
+                problems.append(
+                    f"并发调用新增或增长：{key} owner={owner} {new_count}>{old_count}"
+                )
+            elif new_count < old_count:
+                problems.append(
+                    f"并发低水位已下降，请刷新基线：{key} owner={owner} {new_count}<{old_count}"
+                )
     return problems
 
 
@@ -187,7 +360,10 @@ def main() -> int:
         )
         print(f"已写入 {args.baseline.relative_to(PROJECT_ROOT)}")
         return 0
-    baseline = json.loads(args.baseline.read_text(encoding="utf-8"))
+    baseline = cast(
+        dict[str, ConcurrencyRecord],
+        json.loads(args.baseline.read_text(encoding="utf-8")),
+    )
     problems = compare_concurrency(baseline, current)
     if problems:
         print("\n".join(problems))

tests/fixtures/architecture/complexity-v2-baseline.json

@@ -58,8 +58,10 @@
     "app/application/messaging/skill.py:SkillInteractionHandler._handle_text_interaction": 296,
     "app/application/messaging/subscribe.py:SubscribeInteractionHandler._handle_text_interaction": 205,
     "app/application/network.py:NetworkTestService._build_rules": 206,
+    "app/application/plugin/install.py:PluginInstallCommand.__execute_locked": 277,
     "app/application/rss.py:RssHelper._parse_impl": 213,
     "app/application/security/cookie.py:CookieHelper._get_site_cookie_ua_impl": 241,
+    "app/application/security/cookie.py:CookieHelper._get_site_cookie_ua_impl.__page_handler": 214,
     "app/chain/download/batch.py:DownloadBatchOwner._execute_batch_download": 478,
     "app/chain/download/submission.py:DownloadSubmissionOwner._execute_download_single": 176,
     "app/chain/message.py:MessageChain._handle_ai_message": 202,

tests/test_complexity_gate.py

@@ -1,6 +1,17 @@
 """复杂度只降不增 ratchet 测试。"""
 
-from scripts.architecture.complexity import compare_complexity, compare_complexity_v2
+from pathlib import Path
+
+from scripts.architecture.complexity import (
+    collect_complexity_v2,
+    compare_complexity,
+    compare_complexity_v2,
+)
+
+
+def _assignments(indent: str, count: int) -> str:
+    """生成足够长的临时语句体，精确触发复杂度阈值。"""
+    return "\n".join(f"{indent}value_{index} = {index}" for index in range(count))
 
 
 def test_complexity_ratchet_allows_removal_and_reduction() -> None:
@@ -54,3 +65,58 @@ def test_complexity_v2_ratchet_checks_categories_outside_public_rules() -> None:
     assert any("method: 既有超限增长" in problem for problem in problems)
     assert any("class: 新增超限" in problem for problem in problems)
     assert any("file: 新增超限" in problem for problem in problems)
+
+
+def test_complexity_v2_walks_dunder_match_trystar_and_nested_class_owners(
+    tmp_path: Path,
+) -> None:
+    """完整 AST 遍历必须覆盖 dunder、Match、TryStar 和函数内嵌套类。"""
+    scheduler = tmp_path / "app/scheduler"
+    scheduler.mkdir(parents=True)
+    (scheduler / "dunder.py").write_text(
+        "class Runner:\n"
+        "    def __hidden(self):\n"
+        f"{_assignments('        ', 151)}\n",
+        encoding="utf-8",
+    )
+    (scheduler / "control_flow.py").write_text(
+        "def dispatch(value):\n"
+        "    match value:\n"
+        "        case 1:\n"
+        "            def matched():\n"
+        f"{_assignments('                ', 151)}\n"
+        "            return matched()\n\n"
+        "def guarded():\n"
+        "    try:\n"
+        "        raise ExceptionGroup('failure', [ValueError()])\n"
+        "    except* ValueError:\n"
+        "        def recovered():\n"
+        f"{_assignments('            ', 151)}\n"
+        "        return recovered()\n",
+        encoding="utf-8",
+    )
+    nested_classes = (
+        "class Outer:\n"
+        "    def first(self):\n"
+        "        class Local:\n"
+        f"{_assignments('            ', 501)}\n\n"
+        "    def second(self):\n"
+        "        class Local:\n"
+        f"{_assignments('            ', 501)}\n"
+    )
+    (scheduler / "nested_classes.py").write_text(nested_classes, encoding="utf-8")
+    (scheduler / "conditional.py").write_text(
+        "if enabled:\n"
+        "    class Conditional:\n"
+        f"{_assignments('        ', 501)}\n",
+        encoding="utf-8",
+    )
+
+    report = collect_complexity_v2(tmp_path)
+
+    assert "app/scheduler/dunder.py:Runner.__hidden" in report["method"]
+    assert "app/scheduler/control_flow.py:dispatch.matched" in report["method"]
+    assert "app/scheduler/control_flow.py:guarded.recovered" in report["method"]
+    assert "app/scheduler/conditional.py:Conditional" in report["class"]
+    assert "app/scheduler/nested_classes.py:Outer.first.Local" in report["class"]
+    assert "app/scheduler/nested_classes.py:Outer.second.Local" in report["class"]

tests/test_concurrency_gate.py

@@ -6,57 +6,159 @@ from pathlib import Path
 from scripts.architecture.concurrency import collect_concurrency, compare_concurrency
 
 
-def test_concurrency_gate_tracks_nested_owners_and_known_primitives(tmp_path: Path) -> None:
-    """嵌套函数、类方法和常见原生并发入口都必须被枚举。"""
-    source_path = tmp_path / "app/scheduler/sample.py"
-    source_path.parent.mkdir(parents=True)
-    source_path.write_text(
-        textwrap.dedent(
-            """
-            import asyncio
-            import threading
-            from concurrent.futures import ThreadPoolExecutor
+def _write_source(root: Path, relative: str, source: str) -> None:
+    """写入临时宿主源码，供 AST collector 做结构性验证。"""
+    source_path = root / relative
+    source_path.parent.mkdir(parents=True, exist_ok=True)
+    source_path.write_text(textwrap.dedent(source), encoding="utf-8")
 
-            class Runner:
-                async def run(self):
-                    asyncio.create_task(self.child())
-                    threading.Timer(1, self.child).start()
-                    with ThreadPoolExecutor() as executor:
-                        executor.submit(self.child)
 
-                async def child(self):
-                    await asyncio.to_thread(lambda: None)
-            """
-        ),
-        encoding="utf-8",
+def test_concurrency_gate_resolves_aliases_and_runtime_sources(tmp_path: Path) -> None:
+    """canonical 别名、TaskGroup、事件循环和 Executor 来源必须完整枚举。"""
+    _write_source(
+        tmp_path,
+        "app/doctor/sample.py",
+        """
+        import asyncio as aio
+        import concurrent.futures as cf
+        from asyncio import AbstractEventLoop, TaskGroup as TG, create_task as spawn
+        from concurrent.futures import ThreadPoolExecutor as Pool
+        from threading import Thread as WorkerThread
+
+        async def run(job, typed_loop: AbstractEventLoop):
+            loop = aio.get_running_loop()
+            spawn(job())
+            aio.create_task(job())
+            loop.create_task(job())
+            typed_loop.run_in_executor(None, job)
+            async with TG() as group:
+                group.create_task(job())
+            with cf.ThreadPoolExecutor() as executor:
+                executor.submit(job)
+            with Pool() as executor:
+                executor.map(job, [])
+            WorkerThread(target=job)
+        """,
     )
 
-    facts = collect_concurrency(tmp_path, scan_roots=("app/scheduler",))
-    assert facts["app/scheduler/sample.py:8:create_task"]["owner"] == "Runner.run"
-    assert facts["app/scheduler/sample.py:9:Timer"]["owner"] == "Runner.run"
-    assert facts["app/scheduler/sample.py:10:ThreadPoolExecutor"]["owner"] == "Runner.run"
-    assert facts["app/scheduler/sample.py:14:to_thread"]["owner"] == "Runner.child"
+    facts = collect_concurrency(tmp_path)
+
+    assert facts["app/doctor/sample.py:asyncio.create_task"]["owners"] == {"run": 2}
+    assert facts["app/doctor/sample.py:asyncio.AbstractEventLoop.create_task"]["owners"] == {
+        "run": 1
+    }
+    assert facts["app/doctor/sample.py:asyncio.AbstractEventLoop.run_in_executor"][
+        "owners"
+    ] == {"run": 1}
+    assert facts["app/doctor/sample.py:asyncio.TaskGroup"]["owners"] == {"run": 1}
+    assert facts["app/doctor/sample.py:asyncio.TaskGroup.create_task"]["owners"] == {
+        "run": 1
+    }
+    assert facts["app/doctor/sample.py:concurrent.futures.ThreadPoolExecutor"][
+        "owners"
+    ] == {"run": 2}
+    assert facts["app/doctor/sample.py:concurrent.futures.Executor.submit"]["owners"] == {
+        "run": 1
+    }
+    assert facts["app/doctor/sample.py:concurrent.futures.Executor.map"]["owners"] == {
+        "run": 1
+    }
+    assert facts["app/doctor/sample.py:threading.Thread"]["owners"] == {"run": 1}
 
 
-def test_concurrency_ratchet_rejects_new_facts_and_owner_drift() -> None:
-    """新增并发调用和 owner 漂移必须阻断 ratchet。"""
+def test_concurrency_gate_excludes_unproven_same_named_calls(tmp_path: Path) -> None:
+    """普通同名函数和对象方法没有 canonical 来源时不得误报。"""
+    _write_source(
+        tmp_path,
+        "app/testing/fakes.py",
+        """
+        def create_task():
+            return None
+
+        def run_in_threadpool():
+            return None
+
+        def run(obj):
+            create_task()
+            run_in_threadpool()
+            obj.create_task()
+            obj.submit()
+        """,
+    )
+
+    assert collect_concurrency(tmp_path) == {}
+
+
+def test_concurrency_facts_ignore_line_drift_and_preserve_counts(tmp_path: Path) -> None:
+    """空行漂移不得改变事实键，同一行多个调用必须保留精确计数。"""
+    relative = "app/runtime/sample.py"
+    _write_source(
+        tmp_path,
+        relative,
+        """
+        import asyncio
+        async def run(first, second):
+            asyncio.create_task(first()); asyncio.create_task(second())
+        """,
+    )
+    baseline = collect_concurrency(tmp_path)
+    _write_source(
+        tmp_path,
+        relative,
+        """
+
+
+        import asyncio
+        async def run(first, second):
+            asyncio.create_task(first()); asyncio.create_task(second())
+        """,
+    )
+
+    current = collect_concurrency(tmp_path)
+    assert current == baseline
+    assert current["app/runtime/sample.py:asyncio.create_task"]["owners"] == {"run": 2}
+
+
+def test_concurrency_gate_scans_complete_app_and_keeps_exact_exclusions(
+    tmp_path: Path,
+) -> None:
+    """默认扫描必须覆盖完整 app，同时排除 Plugin、SDK 和 Compat 表面。"""
+    source = """
+        import asyncio
+        asyncio.create_task(job())
+    """
+    _write_source(tmp_path, "app/doctor/check.py", source)
+    _write_source(tmp_path, "app/plugins/copy.py", source)
+    _write_source(tmp_path, "app/sdk/facade.py", source)
+    _write_source(tmp_path, "app/runtime/compat/legacy.py", source)
+
+    facts = collect_concurrency(tmp_path)
+    assert list(facts) == ["app/doctor/check.py:asyncio.create_task"]
+    assert facts["app/doctor/check.py:asyncio.create_task"]["owners"] == {"<module>": 1}
+
+
+def test_concurrency_ratchet_rejects_growth_and_stale_low_water() -> None:
+    """新增 owner、调用增长和未刷新的下降事实都必须阻断 ratchet。"""
     baseline = {
-        "app/runtime/tasks.py:10:create_task": {
-            "owner": "Runtime.run",
-            "target": "create_task",
+        "app/runtime/tasks.py:asyncio.create_task": {
+            "target": "asyncio.create_task",
+            "owners": {"Runtime.run": 2, "Runtime.old": 1},
         },
     }
     current = {
-        "app/runtime/tasks.py:10:create_task": {
-            "owner": "Runtime.changed",
-            "target": "create_task",
+        "app/runtime/tasks.py:asyncio.create_task": {
+            "target": "asyncio.create_task",
+            "owners": {"Runtime.run": 3, "Runtime.new": 1},
         },
-        "app/api/system.py:20:Thread": {
-            "owner": "system",
-            "target": "Thread",
+        "app/api/system.py:threading.Thread": {
+            "target": "threading.Thread",
+            "owners": {"system": 1},
         },
     }
 
     problems = compare_concurrency(baseline, current)
-    assert any("owner 漂移" in problem for problem in problems)
+
     assert any("新增原生并发事实" in problem for problem in problems)
+    assert any("Runtime.run 3>2" in problem for problem in problems)
+    assert any("Runtime.new 1>0" in problem for problem in problems)
+    assert any("Runtime.old 0<1" in problem for problem in problems)