diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml index 343a5a425..64f1ef815 100644 --- a/.github/workflows/test.yml +++ b/.github/workflows/test.yml @@ -47,6 +47,7 @@ jobs: tests/test_architecture_dependencies.py \ tests/test_architecture_adapter_imports.py \ tests/test_architecture_egress.py \ + tests/test_architecture_event_consumers.py \ tests/test_architecture_contract_baseline.py \ tests/test_architecture_baseline_cli.py -q uv run --locked --no-sync python \ diff --git a/docs/architecture-optimization-checklist.md b/docs/architecture-optimization-checklist.md index 033ab7a20..a4e12375a 100644 --- a/docs/architecture-optimization-checklist.md +++ b/docs/architecture-optimization-checklist.md @@ -158,7 +158,8 @@ MoviePilot V3 已经形成较清晰的模块化单体：`foundation`、`domain` S0-L2.4 已统一为 Application-owned Port + startup 注入，并将现有直连全部列为临时债务。 - 审计时完整 SCC 只进入生成快照，语义测试只覆盖特定根；S0-L2.2 已增加完整宿主 SCC policy 门禁。 - 架构总览此前仍记录 811 模块、6,572 条边和 1 个 SCC，已经落后于当前基线。 -- Event consumer 扫描把任意名为 `.register()` 的调用都当成事件注册，存在明确误报。 +- Event consumer 扫描曾把任意同名 `.register()` 调用当成事件注册；S0-L2.5 已改为证明 + canonical EventManager receiver，10 个动态误报归零并保留唯一 workflow 动态注册。 **目标与步骤** @@ -166,7 +167,7 @@ MoviePilot V3 已经形成较清晰的模块化单体：`foundation`、`domain` - [x] 修正 Oper 示例，分别展示宿主显式 UoW 与插件兼容 Facade，并以文档测试禁止回退。 - [x] 明确 Application/Chain 不永久直连具体 Adapter；业务层拥有 Port，startup 注入实现。 - [x] 让 SCC 规则、精确 policy 和文档声明一致；Chain 临时债务与 TMDB vendor containment 分开治理。 -- [ ] Event 扫描只识别 EventManager 实例/别名和事件装饰器。 +- [x] Event 扫描只识别 EventManager 实例/别名和事件装饰器；未知 receiver 不再污染动态事实。 - [ ] CI 分开报告“快照一致”与“语义规则通过”，禁止把前者表述为架构完全正确。 **验收** @@ -495,7 +496,8 @@ MoviePilot V3 已经形成较清晰的模块化单体：`foundation`、`domain` - 复杂度脚本只检查 API、Application、Chain 的公共入口；私有长方法、类/文件规模和圈复杂度不受控。 - strict mypy 仅 41 个文件，高风险 lifecycle、Scheduler、Agent、Plugin Manager 多数不在 frontier。 - coverage ratchet 只聚合 Application 和 Domain。 -- Task owner gate 不盘点原生并发原语；Event consumer 扫描存在误报。 +- Task owner gate 不盘点原生并发原语；Event producer 别名/关键字识别和 consumer 人工 policy + 尚未纳入统一事实源门禁。 - Ruff 仅是有限规则集的历史低水位，不代表整体风格/正确性无债务。 **目标与步骤** @@ -504,7 +506,9 @@ MoviePilot V3 已经形成较清晰的模块化单体：`foundation`、`domain` - [ ] strict mypy frontier 按 `runtime extensions -> startup -> workflow/scheduler -> messaging -> Agent` 扩大；每批必须先清零再加入配置。 - [ ] coverage 增加 Chain、Runtime、Agent、Startup 的高风险子包或关键文件组，不用低价值行数冲百分比。 -- [ ] 原生并发门禁和 Event 扫描按 ARCH-101/106 修正。 +- [ ] 原生并发门禁按 ARCH-101/106 修正。 +- [x] Event consumer 扫描证明 canonical EventManager receiver，清除同名方法误报。 +- [ ] Event producer 识别和 consumer zero-growth policy 在 S0-L2.6 纳入统一事实源门禁。 - [ ] Module Quality Scale 增加 capability -> required rules -> evidence tests 映射，避免“已登记”等同“已验证”。 - [ ] 修改 CI 或门禁脚本时同时运行 `tests/test_architecture_ci.py` 和对应脚本单元测试。 diff --git a/docs/architecture-overview.md b/docs/architecture-overview.md index 57b934043..8ea9ed6f5 100644 --- a/docs/architecture-overview.md +++ b/docs/architecture-overview.md @@ -364,6 +364,11 @@ flowchart TB `errors` 分别用于处理未知自定义事件和兼容期校验失败。快照修改不会回写原事件，如需拦截、取消 或替换链式结果，插件仍应修改原始 `event.event_data`。 +架构基线中的 consumer 表示宿主源码中可静态证明的注册点，不是运行时 listener 实例数。 +collector 只接受 canonical `eventmanager`、`EventManager()` 及其有限别名，忽略其他对象的同名 +`register`/`add_event_listener`；当前宿主有 16 个静态注册点，另保留 1 个由工作流配置驱动的 +真实动态注册。`app/plugins/**` 插件副本不进入宿主事实。 + ```python from app.sdk.events import snapshot_event_data diff --git a/docs/architecture-refactor-roadmap.md b/docs/architecture-refactor-roadmap.md index 9ccec73f9..272a7799a 100644 --- a/docs/architecture-refactor-roadmap.md +++ b/docs/architecture-refactor-roadmap.md @@ -77,8 +77,8 @@ G-ARCH 只有在以下条件全部满足后才可完成： | S0-L2.2 完整宿主 SCC policy | `DELIVERED` | S0-L2.1 | `a884ab5c2`：完整宿主 SCC 精确 policy 生效，远端 `0/0` | | S0-L2.3 Adapter 直连事实 | `DELIVERED` | S0-L2.1 | `e1483e85d`：锁定 28 条原始 Adapter import 事实，远端 `0/0` | | S0-L2.4 Adapter zero-growth | `DELIVERED` | S0-L2.3 | `2553226f3`：冻结 28 条直连及 owner，收缩/新增/stale policy 门禁生效，远端 `0/0` | -| S0-L2.4b HTTP/Egress 事实与政策 | `VERIFIED` | S0-L2.4 | 66 条事实、12 条债务和 54 条精确例外已冻结；collector、policy、全量测试与本地质量门禁通过，待推送后确认远端 CI | -| S0-L2.5 Event consumer 识别 | `PLANNED` | S0-L2.1 | consumer 只识别可静态证明的 EventManager 注册，动态误报归零 | +| S0-L2.4b HTTP/Egress 事实与政策 | `DELIVERED` | S0-L2.4 | `47f0de745`、`43d52a35b`、`8d602149f`：冻结 66 条出口事实，消除 CI 类型/覆盖率漂移；远端全绿且 `0/0` | +| S0-L2.5 Event consumer 识别 | `VERIFIED` | S0-L2.1 | consumer 只识别可静态证明的 EventManager 注册，10 个同名方法动态误报归零；本地 6,459 passed / 6 skipped，待推送 CI | | S0-L2.6 事实源与 CI 投影 | `PLANNED` | S0-L2.2,S0-L2.4b,S0-L2.5 | fixture/policy/overview 职责固定，CI 分开报告语义 policy 与快照一致性 | ### S1：可靠性、事务与数据合同 @@ -154,53 +154,56 @@ G-ARCH 只有在以下条件全部满足后才可完成： ## 4. 当前活动叶子 -### S0-L2.4b HTTP/Egress 事实与政策 +### S0-L2.5 Event consumer 识别 **Status:** `VERIFIED`（本地验收完成，等待提交、推送和远端 CI 确认） **Outcome** -扫描宿主 raw transport、network SDK 和库名扫描会漏掉的协议操作，事实保留 import provenance、 -稳定 callable/operation，不保存行号。普通 HTTP/Session bridge 是待迁移债务；canonical transport、 -SDK、streaming、contained vendor 和 local control-plane 只允许精确 containment。 +把 Event consumer 从“末级方法名碰巧是 `register`/`add_event_listener`”收紧为可静态证明的 +canonical `EventManager` receiver。16 个静态注册点保持不变；10 个 selector、SDK hook、Oper、 +Model、TaskRegistry、`atexit` 和 EventManager 内部展开误报归零；配置驱动的 workflow 注册是唯一 +真实动态 consumer。 **Ownership** -- `scripts/architecture/` 的 direct egress AST collector 与 registry。 -- `tests/fixtures/architecture/dependency-baseline.json` 的生成事实和 - `dependency-policy.json` 的人工分类。 -- `tests/test_architecture_egress.py` 的 collector、当前事实、policy 和 zero-growth 门禁。 -- HTTP 规范、架构总览、优化清单与快速架构 CI 投影。 +- `scripts/architecture/event_consumers.py` 的 receiver/event provenance collector。 +- `scripts/architecture/baseline.py` 的 runtime event contract 集成与 diagnostics。 +- `tests/fixtures/architecture/runtime-contract-baseline.json` 的生成事实。 +- `tests/test_architecture_event_consumers.py` 的 alias、shadow、rebind、decorator 与动态边界测试。 +- API listener 所有权规则、架构规范、优化清单与快速架构 CI 投影。 - 本路线图的叶子状态和交付记录。 **Excluded** -- 不在本叶迁移生产 HTTP 调用；本叶完成完整事实、分类和不可增长门禁。 -- 不修改 Adapter 直连 policy、完整依赖图/SCC/digest、运行时代码或 `app/plugins/**`。 -- 未登记 registry 的任意第三方包不能被猜测为网络 SDK；新增 SDK 必须显式扩展 registry 与 policy。 +- 不修改生产 EventManager、事件 ABI、handler 执行顺序或插件消费者。 +- 不把 `app/plugins/**` 副本纳入宿主扫描。 +- producer 关键字/别名解析与 consumer 人工 zero-growth policy 留给 S0-L2.6 统一事实源叶；本叶只把 + consumer 识别结果变为真实、完整且可测试的事实。 **Acceptance** ```bash .venv/bin/python -m pytest \ - tests/test_architecture_egress.py \ + tests/test_architecture_event_consumers.py \ + tests/test_architecture_dependencies.py \ tests/test_architecture_contract_baseline.py \ tests/test_architecture_baseline_cli.py \ - tests/test_architecture_ci.py \ - tests/test_plugin_identity_transitions.py -q -.venv/bin/python scripts/architecture/baseline.py --check-host + tests/test_architecture_ci.py -q +.venv/bin/python scripts/architecture/baseline.py --check-host --diagnostics .venv/bin/python scripts/architecture/ruff_ratchet.py .venv/bin/python scripts/architecture/mypy_ratchet.py .venv/bin/pylint scripts/architecture/baseline.py \ - scripts/architecture/egress.py \ - tests/test_architecture_egress.py \ + scripts/architecture/event_consumers.py \ + tests/test_architecture_event_consumers.py \ + tests/test_architecture_dependencies.py \ tests/test_architecture_contract_baseline.py \ - tests/test_architecture_ci.py \ - tests/test_plugin_identity_transitions.py + tests/test_architecture_baseline_cli.py \ + tests/test_architecture_ci.py git diff --check ``` **Delivery** -- 单一提交主题：建立并冻结 direct egress 事实与精确政策。 +- 单一提交主题：以可证明的 EventManager provenance 替换同名方法扫描并清除全部动态误报。 - 推送 `origin/v3` 后确认提交祖先关系、远端 SHA 和 ahead/behind `0/0`。 diff --git a/docs/rules/05-architecture.md b/docs/rules/05-architecture.md index f403dec8d..0db57e9f0 100644 --- a/docs/rules/05-architecture.md +++ b/docs/rules/05-architecture.md @@ -569,6 +569,16 @@ removed in the same reviewed change so that it cannot return. | `compat -> canonical implementation at module import time` | Forbidden | | Any import that creates a module-level cycle | Forbidden; the complete host graph must match the exact reviewed SCC policy, and temporary debt must have a removal owner | +Event consumer facts require statically proven ownership. A `register` or +`add_event_listener` method name alone is not evidence: the receiver must resolve +to the canonical `app.runtime.events.eventmanager`, an `EventManager` instance, +or a finite alias of either. Unknown receivers are ignored. A proven manager with +an event value that cannot be resolved is recorded as a dynamic consumer; the +current host permits only the configuration-driven workflow registration. The +collector respects lexical shadowing and rebinds, distinguishes decorator +application from obtaining a decorator factory, and never scans `app/plugins/**` +as host code. + ## Key File Locations | Path | Purpose | diff --git a/scripts/architecture/baseline.py b/scripts/architecture/baseline.py index 33d386ee8..34081008d 100644 --- a/scripts/architecture/baseline.py +++ b/scripts/architecture/baseline.py @@ -18,6 +18,11 @@ try: except ModuleNotFoundError: from egress import collect_direct_egress +try: + from scripts.architecture.event_consumers import collect_event_consumers +except ModuleNotFoundError: + from event_consumers import collect_event_consumers + PROJECT_ROOT = Path(__file__).resolve().parents[2] APP_ROOT = PROJECT_ROOT / "app" BASELINE_ROOT = PROJECT_ROOT / "tests" / "fixtures" / "architecture" @@ -743,12 +748,18 @@ def _collect_event_locations() -> tuple[ """扫描事件枚举及其生产、消费位置，供语义和诊断视图复用。""" event_members = _event_enum_members("EventType") chain_event_members = _event_enum_members("ChainEventType") + modules = discover_modules() producers: dict[str, list[dict[str, Any]]] = defaultdict(list) - consumers: dict[str, list[dict[str, Any]]] = defaultdict(list) dynamic_producers: list[dict[str, Any]] = [] - dynamic_consumers: list[dict[str, Any]] = [] - for module_name, path in discover_modules().items(): + consumers, dynamic_consumers = collect_event_consumers( + modules, + { + "EventType": event_members, + "ChainEventType": chain_event_members, + }, + ) + for module_name, path in modules.items(): tree = parse_source(path) for node in ast.walk(tree): if not isinstance(node, ast.Call) or not isinstance( @@ -763,37 +774,6 @@ def _collect_event_locations() -> tuple[ producers[reference].append(location) else: dynamic_producers.append(location) - continue - if node.func.attr not in {"register", "add_event_listener"}: - continue - references: list[str] = [] - if node.args: - target = node.args[0] - if reference := _event_reference(target): - references.append(reference) - elif isinstance(target, (ast.List, ast.Tuple)): - references.extend( - reference - for item in target.elts - if (reference := _event_reference(item)) - ) - elif ( - isinstance(target, ast.Name) - and target.id in {"EventType", "ChainEventType"} - ): - enum_members = ( - event_members - if target.id == "EventType" - else chain_event_members - ) - references.extend( - f"{target.id}.{member}" for member in enum_members - ) - if references: - for reference in references: - consumers[reference].append(location) - else: - dynamic_consumers.append(location) enum_names = [ *(f"EventType.{member}" for member in event_members), diff --git a/scripts/architecture/event_consumers.py b/scripts/architecture/event_consumers.py new file mode 100644 index 000000000..534a1862b --- /dev/null +++ b/scripts/architecture/event_consumers.py @@ -0,0 +1,983 @@ +"""静态收集可证明的宿主 EventManager consumer。""" + +import ast +from collections import defaultdict +from dataclasses import dataclass +from pathlib import Path +from typing import Any, Literal, TypeAlias + +_DEFAULT_IDENTITY = "" +_DYNAMIC_IDENTITY = "" +_EVENT_MANAGER_METHODS = {"add_event_listener", "register"} +_COMPREHENSION_SCOPES = ( + ast.ListComp, + ast.SetComp, + ast.DictComp, + ast.GeneratorExp, +) +_FUNCTION_SCOPES = ( + ast.FunctionDef, + ast.AsyncFunctionDef, + ast.Lambda, + ast.ClassDef, + *_COMPREHENSION_SCOPES, +) + + +@dataclass(frozen=True, slots=True) +class _Symbol: + """记录 canonical module、EventManager 类或实例的静态来源。""" + + kind: Literal[ + "module", + "manager_class", + "manager_factory", + "manager_instance", + "type_checking", + ] + value: str = "" + + +@dataclass(frozen=True, slots=True) +class _EventSelection: + """记录一次注册可静态确定的事件集合及未知余项。""" + + events: tuple[str, ...] + kind: Literal["member", "enum", "list"] + dynamic: bool = False + + +@dataclass(frozen=True, slots=True) +class _DecoratorFactory: + """记录尚未应用到 handler 的 EventManager.register 返回值。""" + + selection: _EventSelection + priority: str + + +@dataclass(frozen=True, slots=True) +class _Registration: + """记录已证明 receiver 后解析出的注册调用。""" + + method: Literal["add_event_listener", "register"] + selection: _EventSelection + handler: str + priority: str + + +_ScopeValue: TypeAlias = _Symbol | _EventSelection | _DecoratorFactory | None + + +def _expression_name(node: ast.AST | None) -> str: + """返回 Name/Attribute 表达式的稳定点分名称。""" + if isinstance(node, ast.Name): + return node.id + if isinstance(node, ast.Attribute): + prefix = _expression_name(node.value) + return ".".join(part for part in (prefix, node.attr) if part) + return "" + + +def _handler_identity(node: ast.AST | None) -> str: + """返回不含源码位置的 handler identity。""" + if name := _expression_name(node): + return name + if isinstance(node, ast.Lambda): + return "" + return _DYNAMIC_IDENTITY + + +def _priority_identity(node: ast.AST | None) -> str: + """返回不含源码位置的 priority identity。""" + if node is None: + return _DEFAULT_IDENTITY + if isinstance(node, ast.Constant): + return repr(node.value) + if name := _expression_name(node): + return name + return _DYNAMIC_IDENTITY + + +def _location_sort_key(item: dict[str, Any]) -> tuple[str, int, str]: + """返回 consumer location 的稳定排序键。""" + return ( + str(item["caller"]), + int(item["line"]), + str(item["identity"]), + ) + + +def _bound_names(target: ast.AST) -> set[str]: + """返回赋值目标在当前 lexical scope 绑定的名称。""" + if isinstance(target, ast.Name): + return {target.id} + if isinstance(target, (ast.List, ast.Tuple)): + return { + name + for element in target.elts + for name in _bound_names(element) + } + if isinstance(target, ast.Starred): + return _bound_names(target.value) + return set() + + +def _pattern_bound_names(pattern: ast.pattern) -> set[str]: + """返回结构化匹配 pattern 捕获到当前 scope 的名称。""" + if isinstance(pattern, ast.MatchAs): + names = _pattern_bound_names(pattern.pattern) if pattern.pattern else set() + if pattern.name: + names.add(pattern.name) + return names + if isinstance(pattern, ast.MatchStar): + return {pattern.name} if pattern.name else set() + if isinstance(pattern, ast.MatchSequence): + return { + name + for child in pattern.patterns + for name in _pattern_bound_names(child) + } + if isinstance(pattern, ast.MatchMapping): + names = { + name + for child in pattern.patterns + for name in _pattern_bound_names(child) + } + if pattern.rest: + names.add(pattern.rest) + return names + if isinstance(pattern, ast.MatchClass): + return { + name + for child in (*pattern.patterns, *pattern.kwd_patterns) + for name in _pattern_bound_names(child) + } + if isinstance(pattern, ast.MatchOr): + return { + name + for child in pattern.patterns + for name in _pattern_bound_names(child) + } + return set() + + +def _function_local_names( + node: ast.FunctionDef | ast.AsyncFunctionDef | ast.Lambda, +) -> set[str]: + """按 Python lexical scope 预收集函数局部绑定，避免回退到同名全局。""" + parents: dict[ast.AST, ast.AST] = {} + for parent in ast.walk(node): + for child in ast.iter_child_nodes(parent): + parents[child] = parent + + def belongs_to_function( + candidate: ast.AST, + *, + cross_comprehensions: bool = False, + ) -> bool: + """判断节点是否属于当前函数而非嵌套 scope。""" + parent = parents.get(candidate) + while parent is not None and parent is not node: + if isinstance(parent, _FUNCTION_SCOPES): + if not ( + cross_comprehensions + and isinstance(parent, _COMPREHENSION_SCOPES) + ): + return False + parent = parents.get(parent) + return parent is node + + local_names = { + argument.arg + for argument in ( + *node.args.posonlyargs, + *node.args.args, + *node.args.kwonlyargs, + ) + } + if node.args.vararg: + local_names.add(node.args.vararg.arg) + if node.args.kwarg: + local_names.add(node.args.kwarg.arg) + global_names: set[str] = set() + nonlocal_names: set[str] = set() + for candidate in ast.walk(node): + if candidate is node or not belongs_to_function(candidate): + continue + if isinstance(candidate, ast.Name) and isinstance( + candidate.ctx, + (ast.Store, ast.Del), + ): + local_names.add(candidate.id) + elif isinstance(candidate, (ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef)): + local_names.add(candidate.name) + elif isinstance(candidate, (ast.Import, ast.ImportFrom)): + for alias in candidate.names: + if alias.name != "*": + local_names.add(alias.asname or alias.name.split(".", 1)[0]) + elif isinstance(candidate, ast.ExceptHandler) and candidate.name: + local_names.add(candidate.name) + elif isinstance(candidate, ast.Match): + local_names.update( + name + for case in candidate.cases + for name in _pattern_bound_names(case.pattern) + ) + elif isinstance(candidate, ast.Global): + global_names.update(candidate.names) + elif isinstance(candidate, ast.Nonlocal): + nonlocal_names.update(candidate.names) + for candidate in ast.walk(node): + if isinstance(candidate, ast.NamedExpr) and belongs_to_function( + candidate, + cross_comprehensions=True, + ): + local_names.update(_bound_names(candidate.target)) + return local_names - global_names - nonlocal_names + + +class _EventConsumerCollector(ast.NodeVisitor): + """以有限 lexical provenance 收集单个宿主模块的事件消费者。""" + + def __init__( + self, + module_name: str, + event_members: dict[str, tuple[str, ...]], + *, + collect_facts: bool, + module_final_scope: dict[str, _ScopeValue] | None = None, + ) -> None: + """初始化模块、事件枚举、收集模式和最终模块绑定。""" + self._module_name = module_name + self._event_members = event_members + self._collect_facts = collect_facts + self._module_final_scope = module_final_scope or {} + self._scopes: list[dict[str, _ScopeValue]] = [{}] + self._scope_kinds = ["module"] + self._function_final_scopes: list[dict[str, _ScopeValue]] = [] + self._qualnames: list[str] = [] + self.static: dict[str, list[dict[str, Any]]] = defaultdict(list) + self.dynamic: list[dict[str, Any]] = [] + + def module_scope(self) -> dict[str, _ScopeValue]: + """返回按模块执行顺序收敛后的符号状态。""" + return dict(self._scopes[0]) + + def _lookup(self, name: str) -> _ScopeValue: + """从内向外解析 lexical binding。""" + skip_class_scope = self._scope_kinds[-1] == "comprehension" + for scope, kind in reversed(list(zip(self._scopes, self._scope_kinds))): + if skip_class_scope and kind == "class": + continue + if name in scope: + return scope[name] + return None + + def _set(self, name: str, value: _ScopeValue) -> None: + """在当前 lexical scope 写入或清空 binding。""" + self._scopes[-1][name] = value + + @staticmethod + def _merge_scope_states( + states: list[list[dict[str, _ScopeValue]]], + ) -> list[dict[str, _ScopeValue]]: + """仅保留所有控制流路径一致的静态 provenance。""" + return [ + { + name: ( + scopes[0].get(name) + if all( + scope.get(name) == scopes[0].get(name) + for scope in scopes[1:] + ) + else None + ) + for name in set().union(*(scope.keys() for scope in scopes)) + } + for scopes in zip(*states) + ] + + def _discover_scope_after( + self, + statements: list[ast.stmt], + scopes: list[dict[str, _ScopeValue]], + scope_kinds: list[str], + ) -> dict[str, _ScopeValue]: + """无事实副作用地计算一组语句执行后的最内层 scope。""" + discovery = _EventConsumerCollector( + self._module_name, + self._event_members, + collect_facts=False, + module_final_scope=self._module_final_scope, + ) + discovery._scopes = [dict(scope) for scope in scopes] + discovery._scope_kinds = list(scope_kinds) + for statement in statements: + discovery.visit(statement) + return dict(discovery._scopes[-1]) + + def _symbol_for_canonical(self, canonical: str) -> _ScopeValue: + """把有限 canonical 路径转换为 collector symbol。""" + if canonical == "app.runtime.events.eventmanager": + return _Symbol("manager_instance") + if canonical == "app.runtime.events.EventManager": + return _Symbol("manager_class") + if canonical == "typing.TYPE_CHECKING": + return _Symbol("type_checking") + if canonical in { + "app", + "app.runtime", + "app.runtime.events", + "app.schemas", + "app.schemas.types", + "typing", + }: + return _Symbol("module", canonical) + for enum_name, members in self._event_members.items(): + if canonical == f"app.schemas.types.{enum_name}": + return _EventSelection( + events=tuple(f"{enum_name}.{member}" for member in members), + kind="enum", + ) + return None + + def _runtime_type_checking_value(self, test: ast.expr) -> bool | None: + """仅对可证明的 typing.TYPE_CHECKING 返回确定运行期分支。""" + negate = isinstance(test, ast.UnaryOp) and isinstance(test.op, ast.Not) + target = test.operand if negate else test + resolved = self._resolve(target) + if isinstance(resolved, _Symbol) and resolved.kind == "type_checking": + return negate + return None + + def _resolve(self, node: ast.AST) -> _ScopeValue: + """解析有限 import/module/赋值别名，不跨函数推断返回值。""" + if isinstance(node, ast.Name): + return self._lookup(node.id) + if isinstance(node, ast.Attribute): + parent = self._resolve(node.value) + if isinstance(parent, _Symbol) and parent.kind == "module": + return self._symbol_for_canonical(f"{parent.value}.{node.attr}") + if isinstance(parent, _Symbol) and parent.kind == "manager_class": + if node.attr == "get_existing_instance": + return _Symbol("manager_factory") + if isinstance(parent, _EventSelection) and parent.kind == "enum": + enum_name = parent.events[0].split(".", 1)[0] if parent.events else "" + if node.attr in self._event_members.get(enum_name, ()): + return _EventSelection( + events=(f"{enum_name}.{node.attr}",), + kind="member", + ) + return None + if isinstance(node, ast.List): + return self._resolve_event_selection( + node, + allow_enum=True, + allow_list=True, + ) + if isinstance(node, ast.Call): + target = self._resolve(node.func) + if isinstance(target, _Symbol) and target.kind in { + "manager_class", + "manager_factory", + }: + return _Symbol("manager_instance") + registration = self._registration(node) + if registration and registration.method == "register": + return _DecoratorFactory( + selection=registration.selection, + priority=registration.priority, + ) + return None + + def _resolve_event_selection( + self, + node: ast.AST, + *, + allow_enum: bool, + allow_list: bool, + ) -> _EventSelection: + """解析单个事件、enum 类或 register 接受的静态 list。""" + if isinstance(node, ast.List): + if not allow_list: + return _EventSelection((), "list", dynamic=True) + selections = [ + self._resolve_event_selection( + item, + allow_enum=True, + allow_list=False, + ) + for item in node.elts + ] + return _EventSelection( + events=tuple(sorted({event for item in selections for event in item.events})), + kind="list", + dynamic=any(item.dynamic for item in selections), + ) + if isinstance(node, ast.IfExp): + branches = ( + self._resolve_event_selection( + node.body, + allow_enum=allow_enum, + allow_list=allow_list, + ), + self._resolve_event_selection( + node.orelse, + allow_enum=allow_enum, + allow_list=allow_list, + ), + ) + return _EventSelection( + events=tuple(sorted({event for item in branches for event in item.events})), + kind="list" if allow_list else "member", + dynamic=any(item.dynamic for item in branches), + ) + resolved = self._resolve(node) + if isinstance(resolved, _EventSelection): + if resolved.kind == "member": + return resolved + if resolved.kind == "enum" and allow_enum: + return resolved + if resolved.kind == "list" and allow_list: + return resolved + return _EventSelection((), "member", dynamic=True) + + @staticmethod + def _bind_call_arguments( + node: ast.Call, + parameter_names: tuple[str, ...], + required_names: frozenset[str], + ) -> dict[str, ast.AST] | None: + """按真实 Python 调用规则绑定有限签名，拒绝未知或重复参数。""" + if len(node.args) > len(parameter_names) or any( + isinstance(argument, ast.Starred) for argument in node.args + ): + return None + arguments = dict(zip(parameter_names, node.args)) + for keyword in node.keywords: + if ( + keyword.arg is None + or keyword.arg not in parameter_names + or keyword.arg in arguments + ): + return None + arguments[keyword.arg] = keyword.value + if not required_names.issubset(arguments): + return None + return arguments + + def _registration(self, node: ast.Call) -> _Registration | None: + """仅解析 receiver 已证明为 canonical EventManager 实例的注册。""" + if not isinstance(node.func, ast.Attribute): + return None + method = node.func.attr + if method not in _EVENT_MANAGER_METHODS: + return None + receiver = self._resolve(node.func.value) + if not isinstance(receiver, _Symbol) or receiver.kind != "manager_instance": + return None + if method == "register": + arguments = self._bind_call_arguments( + node, + ("etype", "priority"), + frozenset({"etype"}), + ) + event_name = "etype" + handler_node = None + else: + arguments = self._bind_call_arguments( + node, + ("event_type", "handler", "priority"), + frozenset({"event_type", "handler"}), + ) + event_name = "event_type" + handler_node = arguments.get("handler") if arguments else None + if arguments is None: + return None + selection = self._resolve_event_selection( + arguments[event_name], + allow_enum=method == "register", + allow_list=method == "register", + ) + return _Registration( + method=method, + selection=selection, + handler=_handler_identity(handler_node), + priority=_priority_identity(arguments.get("priority")), + ) + + def _decorator_factory_application( + self, + node: ast.Call, + ) -> tuple[_DecoratorFactory, ast.AST] | None: + """解析合法的 decorator(f) 或 decorator(f=...) 立即应用。""" + factory = self._resolve(node.func) + if not isinstance(factory, _DecoratorFactory): + return None + arguments = self._bind_call_arguments( + node, + ("f",), + frozenset({"f"}), + ) + if arguments is None: + return None + return factory, arguments["f"] + + def _record( + self, + node: ast.AST, + selection: _EventSelection, + *, + handler: str, + priority: str, + registration_kind: Literal["decorator", "listener"], + ) -> None: + """写入静态事件位置，并为未知余项写入一条 dynamic 位置。""" + if not self._collect_facts: + return + location = { + "caller": self._module_name, + "line": node.lineno, + "handler": handler, + "priority": priority, + "registration_kind": registration_kind, + "identity": f"{registration_kind}|{handler}|{priority}", + } + for event in selection.events: + self.static[event].append(dict(location)) + if selection.dynamic: + self.dynamic.append(dict(location)) + + def _record_decorator(self, decorator: ast.expr, handler: str) -> bool: + """记录直接或简单赋值别名形式的 register decorator。""" + factory = self._resolve(decorator) + if not isinstance(factory, _DecoratorFactory): + return False + self._record( + decorator, + factory.selection, + handler=handler, + priority=factory.priority, + registration_kind="decorator", + ) + return True + + def _visit_definition_decorators( + self, + decorators: list[ast.expr], + handler: str, + ) -> None: + """按定义期 scope 访问装饰器，Event register 只在实际应用时记账。""" + for decorator in decorators: + if not self._record_decorator(decorator, handler): + self.visit(decorator) + + def visit_Import(self, node: ast.Import) -> None: + """发布模块 import 及其别名。""" + for alias in node.names: + bound_name = alias.asname or alias.name.split(".", 1)[0] + canonical = alias.name if alias.asname else bound_name + self._set(bound_name, self._symbol_for_canonical(canonical)) + + def visit_ImportFrom(self, node: ast.ImportFrom) -> None: + """发布 canonical from-import 及其别名。""" + if node.level or not node.module: + for alias in node.names: + if alias.name != "*": + self._set(alias.asname or alias.name, None) + return + for alias in node.names: + if alias.name == "*": + continue + self._set( + alias.asname or alias.name, + self._symbol_for_canonical(f"{node.module}.{alias.name}"), + ) + + def visit_Assign(self, node: ast.Assign) -> None: + """按执行顺序传播或清空简单赋值别名。""" + value = self._resolve(node.value) + self.visit(node.value) + for target in node.targets: + for name in _bound_names(target): + self._set(name, value) + + def visit_AnnAssign(self, node: ast.AnnAssign) -> None: + """传播带注解且有值的简单赋值别名。""" + value = self._resolve(node.value) if node.value is not None else None + if node.value is not None: + self.visit(node.value) + for name in _bound_names(node.target): + self._set(name, value) + + def visit_AugAssign(self, node: ast.AugAssign) -> None: + """增量赋值使目标 provenance 失效。""" + self.visit(node.value) + for name in _bound_names(node.target): + self._set(name, None) + + def visit_NamedExpr(self, node: ast.NamedExpr) -> None: + """传播海象表达式的简单别名。""" + value = self._resolve(node.value) + self.visit(node.value) + scope_index = len(self._scopes) - 1 + while self._scope_kinds[scope_index] == "comprehension": + scope_index -= 1 + for name in _bound_names(node.target): + self._scopes[scope_index][name] = value + + def visit_Delete(self, node: ast.Delete) -> None: + """删除名称后清空其 provenance。""" + for target in node.targets: + for name in _bound_names(target): + self._set(name, None) + + def visit_If(self, node: ast.If) -> None: + """精确执行 TYPE_CHECKING 分支，其余条件保守合并 binding。""" + self.visit(node.test) + runtime_value = self._runtime_type_checking_value(node.test) + if runtime_value is not None: + statements = node.body if runtime_value else node.orelse + for statement in statements: + self.visit(statement) + return + + original = [dict(scope) for scope in self._scopes] + for statement in node.body: + self.visit(statement) + body_scopes = [dict(scope) for scope in self._scopes] + self._scopes = [dict(scope) for scope in original] + for statement in node.orelse: + self.visit(statement) + else_scopes = [dict(scope) for scope in self._scopes] + self._scopes = self._merge_scope_states([body_scopes, else_scopes]) + + def visit_For(self, node: ast.For) -> None: + """让循环目标遮蔽旧绑定，并保守合并零次与迭代路径。""" + self._visit_for(node) + + def visit_AsyncFor(self, node: ast.AsyncFor) -> None: + """按同步循环相同规则处理异步循环目标。""" + self._visit_for(node) + + def _visit_for(self, node: ast.For | ast.AsyncFor) -> None: + """实现同步和异步循环共享的 provenance 分析。""" + self.visit(node.iter) + original = [dict(scope) for scope in self._scopes] + for name in _bound_names(node.target): + self._set(name, None) + for statement in node.body: + self.visit(statement) + body_scopes = [dict(scope) for scope in self._scopes] + loop_exit = self._merge_scope_states([original, body_scopes]) + self._scopes = [dict(scope) for scope in loop_exit] + for statement in node.orelse: + self.visit(statement) + else_scopes = [dict(scope) for scope in self._scopes] + self._scopes = self._merge_scope_states([loop_exit, else_scopes]) + + def visit_While(self, node: ast.While) -> None: + """保守合并 while 的零次、迭代和 else 路径。""" + self.visit(node.test) + original = [dict(scope) for scope in self._scopes] + for statement in node.body: + self.visit(statement) + body_scopes = [dict(scope) for scope in self._scopes] + loop_exit = self._merge_scope_states([original, body_scopes]) + self._scopes = [dict(scope) for scope in loop_exit] + for statement in node.orelse: + self.visit(statement) + else_scopes = [dict(scope) for scope in self._scopes] + self._scopes = self._merge_scope_states([loop_exit, else_scopes]) + + def visit_With(self, node: ast.With) -> None: + """按进入顺序分析 context，并清空 with 目标的旧 provenance。""" + self._visit_with(node) + + def visit_AsyncWith(self, node: ast.AsyncWith) -> None: + """按同步 with 相同规则处理异步上下文目标。""" + self._visit_with(node) + + def _visit_with(self, node: ast.With | ast.AsyncWith) -> None: + """实现同步和异步上下文管理器共享的绑定分析。""" + for item in node.items: + self.visit(item.context_expr) + if item.optional_vars is not None: + for name in _bound_names(item.optional_vars): + self._set(name, None) + for statement in node.body: + self.visit(statement) + + def visit_Match(self, node: ast.Match) -> None: + """隔离 match case 捕获名称并保守合并所有匹配路径。""" + self.visit(node.subject) + original = [dict(scope) for scope in self._scopes] + states = [original] + for case in node.cases: + self._scopes = [dict(scope) for scope in original] + self.visit(case.pattern) + for name in _pattern_bound_names(case.pattern): + self._set(name, None) + if case.guard is not None: + self.visit(case.guard) + for statement in case.body: + self.visit(statement) + states.append([dict(scope) for scope in self._scopes]) + self._scopes = self._merge_scope_states(states) + + def visit_Try(self, node: ast.Try) -> None: + """隔离异常处理路径，并让异常别名在 handler 内外失效。""" + self._visit_try(node) + + def visit_TryStar(self, node: ast.TryStar) -> None: + """按普通 try 相同规则处理异常组分支。""" + self._visit_try(node) + + def _visit_try(self, node: ast.Try | ast.TryStar) -> None: + """实现 try 与 try-star 共享的保守控制流合并。""" + original = [dict(scope) for scope in self._scopes] + exception_states = [original] + for statement in node.body: + exception_states.append([dict(scope) for scope in self._scopes]) + self.visit(statement) + exception_states.append([dict(scope) for scope in self._scopes]) + exception_entry = self._merge_scope_states(exception_states) + for statement in node.orelse: + self.visit(statement) + continuing_states = [[dict(scope) for scope in self._scopes]] + for handler in node.handlers: + self._scopes = [dict(scope) for scope in exception_entry] + if handler.type is not None: + self.visit(handler.type) + if handler.name: + self._set(handler.name, None) + for statement in handler.body: + self.visit(statement) + if handler.name: + self._set(handler.name, None) + continuing_states.append([dict(scope) for scope in self._scopes]) + self._scopes = self._merge_scope_states( + [*continuing_states, exception_entry] + if node.finalbody + else continuing_states + ) + for statement in node.finalbody: + self.visit(statement) + + def _visit_comprehension( + self, + generators: list[ast.comprehension], + expressions: tuple[ast.expr, ...], + ) -> None: + """在独立 lexical scope 内分析推导式目标和表达式。""" + first, *remaining = generators + self.visit(first.iter) + outer_original = [dict(scope) for scope in self._scopes] + self._scopes.append({}) + self._scope_kinds.append("comprehension") + for name in _bound_names(first.target): + self._set(name, None) + for condition in first.ifs: + self.visit(condition) + for generator in remaining: + self.visit(generator.iter) + for name in _bound_names(generator.target): + self._set(name, None) + for condition in generator.ifs: + self.visit(condition) + for expression in expressions: + self.visit(expression) + self._scope_kinds.pop() + self._scopes.pop() + outer_after = [dict(scope) for scope in self._scopes] + self._scopes = self._merge_scope_states([outer_original, outer_after]) + + def visit_ListComp(self, node: ast.ListComp) -> None: + """在独立 scope 内分析列表推导式。""" + self._visit_comprehension(node.generators, (node.elt,)) + + def visit_SetComp(self, node: ast.SetComp) -> None: + """在独立 scope 内分析集合推导式。""" + self._visit_comprehension(node.generators, (node.elt,)) + + def visit_GeneratorExp(self, node: ast.GeneratorExp) -> None: + """在独立 scope 内分析生成器表达式。""" + self._visit_comprehension(node.generators, (node.elt,)) + + def visit_DictComp(self, node: ast.DictComp) -> None: + """在独立 scope 内分析字典推导式。""" + self._visit_comprehension(node.generators, (node.key, node.value)) + + def visit_FunctionDef(self, node: ast.FunctionDef) -> None: + """按定义期装饰器和调用期 lexical scope 分别分析函数。""" + self._visit_function(node) + + def visit_AsyncFunctionDef(self, node: ast.AsyncFunctionDef) -> None: + """按普通函数相同规则分析异步函数。""" + self._visit_function(node) + + def _visit_function( + self, + node: ast.FunctionDef | ast.AsyncFunctionDef, + ) -> None: + """实现同步和异步函数共享的 scope 分析。""" + if not self._collect_facts: + self._set(node.name, None) + return + handler = ".".join((*self._qualnames, node.name)) + self._visit_definition_decorators(node.decorator_list, handler) + for expression in ( + *node.args.defaults, + *(default for default in node.args.kw_defaults if default is not None), + ): + self.visit(expression) + + saved_scopes = self._scopes + saved_kinds = self._scope_kinds + body_scopes = [ + dict(self._module_final_scope), + *(dict(scope) for scope in self._function_final_scopes), + {name: None for name in _function_local_names(node)}, + ] + body_kinds = [ + "module", + *("function" for _ in self._function_final_scopes), + "function", + ] + final_scope = self._discover_scope_after(node.body, body_scopes, body_kinds) + self._scopes = body_scopes + self._scope_kinds = body_kinds + self._function_final_scopes.append(final_scope) + self._qualnames.append(node.name) + for statement in node.body: + self.visit(statement) + self._qualnames.pop() + self._function_final_scopes.pop() + self._scopes = saved_scopes + self._scope_kinds = saved_kinds + self._set(node.name, None) + + def visit_ClassDef(self, node: ast.ClassDef) -> None: + """在类定义期 scope 记录类装饰器和方法装饰器。""" + if not self._collect_facts: + self._set(node.name, None) + return + handler = ".".join((*self._qualnames, node.name)) + self._visit_definition_decorators(node.decorator_list, handler) + for expression in (*node.bases, *(keyword.value for keyword in node.keywords)): + self.visit(expression) + self._qualnames.append(node.name) + self._scopes.append({}) + self._scope_kinds.append("class") + for statement in node.body: + self.visit(statement) + self._scope_kinds.pop() + self._scopes.pop() + self._qualnames.pop() + self._set(node.name, None) + + def visit_Lambda(self, node: ast.Lambda) -> None: + """让 lambda 参数遮蔽同名模块 alias。""" + if not self._collect_facts: + return + for expression in ( + *node.args.defaults, + *(default for default in node.args.kw_defaults if default is not None), + ): + self.visit(expression) + saved_scopes = self._scopes + saved_kinds = self._scope_kinds + body_scopes = [ + dict(self._module_final_scope), + *(dict(scope) for scope in self._function_final_scopes), + {name: None for name in _function_local_names(node)}, + ] + body_kinds = [ + "module", + *("function" for _ in self._function_final_scopes), + "function", + ] + final_scope = self._discover_scope_after( + [ast.Expr(value=node.body)], + body_scopes, + body_kinds, + ) + self._scopes = body_scopes + self._scope_kinds = body_kinds + self._function_final_scopes.append(final_scope) + self.visit(node.body) + self._function_final_scopes.pop() + self._scopes = saved_scopes + self._scope_kinds = saved_kinds + + def visit_Call(self, node: ast.Call) -> None: + """记录直接 listener 调用或立即应用的 register decorator。""" + factory_application = self._decorator_factory_application(node) + if factory_application: + factory, handler_node = factory_application + self._record( + node, + factory.selection, + handler=_handler_identity(handler_node), + priority=factory.priority, + registration_kind="decorator", + ) + else: + registration = self._registration(node) + if not registration or registration.method != "add_event_listener": + self.generic_visit(node) + return + self._record( + node, + registration.selection, + handler=registration.handler, + priority=registration.priority, + registration_kind="listener", + ) + self.generic_visit(node) + + +def collect_event_consumers( + modules: dict[str, Path], + event_members: dict[str, tuple[str, ...]], +) -> tuple[dict[str, list[dict[str, Any]]], list[dict[str, Any]]]: + """ + 收集宿主 EventManager 的静态和动态 consumer 位置 + + 只有可追溯到 ``app.runtime.events.eventmanager`` 或 canonical + ``EventManager`` 实例的注册才进入事实；未知 receiver 直接忽略。 + + :param modules: 宿主模块名到 Python 源码路径的映射 + :param event_members: EventType/ChainEventType 到公开成员名的映射 + :return: 静态事件位置映射与事件值未知的位置列表 + """ + static: dict[str, list[dict[str, Any]]] = defaultdict(list) + dynamic: list[dict[str, Any]] = [] + for module_name, path in sorted(modules.items()): + if module_name == "app.plugins" or module_name.startswith("app.plugins."): + continue + tree = ast.parse(path.read_text(encoding="utf-8-sig"), filename=str(path)) + discovery = _EventConsumerCollector( + module_name, + event_members, + collect_facts=False, + ) + discovery.visit(tree) + collector = _EventConsumerCollector( + module_name, + event_members, + collect_facts=True, + module_final_scope=discovery.module_scope(), + ) + collector.visit(tree) + for event, locations in collector.static.items(): + static[event].extend(locations) + dynamic.extend(collector.dynamic) + + return ( + { + event: sorted(locations, key=_location_sort_key) + for event, locations in sorted(static.items()) + }, + sorted(dynamic, key=_location_sort_key), + ) diff --git a/tests/fixtures/architecture/runtime-contract-baseline.json b/tests/fixtures/architecture/runtime-contract-baseline.json index dc31298f9..71b7a02a6 100644 --- a/tests/fixtures/architecture/runtime-contract-baseline.json +++ b/tests/fixtures/architecture/runtime-contract-baseline.json @@ -1967,38 +1967,6 @@ "events": { "consumer_count": 16, "dynamic_consumers": [ - { - "caller": "app.adapters.system.fsproxy", - "count": 1 - }, - { - "caller": "app.agent.llm.provider", - "count": 1 - }, - { - "caller": "app.chain.transfer", - "count": 1 - }, - { - "caller": "app.db.oper.transferpending", - "count": 1 - }, - { - "caller": "app.runtime.events", - "count": 3 - }, - { - "caller": "app.runtime.tasks", - "count": 1 - }, - { - "caller": "app.startup.lifecycle", - "count": 1 - }, - { - "caller": "app.testing.bootstrap", - "count": 1 - }, { "caller": "app.workflow", "count": 1 diff --git a/tests/test_architecture_ci.py b/tests/test_architecture_ci.py index 43146266b..1e07a33c9 100644 --- a/tests/test_architecture_ci.py +++ b/tests/test_architecture_ci.py @@ -33,6 +33,7 @@ def test_unit_test_workflow_has_independent_host_architecture_gate(): assert "tests/test_architecture_dependencies.py" in commands assert "tests/test_architecture_adapter_imports.py" in commands assert "tests/test_architecture_egress.py" in commands + assert "tests/test_architecture_event_consumers.py" in commands assert "tests/test_architecture_contract_baseline.py" in commands assert "scripts/architecture/baseline.py --check-host" in commands assert "scripts/architecture/ruff_ratchet.py" in commands diff --git a/tests/test_architecture_contract_baseline.py b/tests/test_architecture_contract_baseline.py index 49d40db5e..22bb23e82 100644 --- a/tests/test_architecture_contract_baseline.py +++ b/tests/test_architecture_contract_baseline.py @@ -395,4 +395,12 @@ def test_event_contract_baseline_covers_every_public_event_enum() -> None: assert set(events["events"]) == expected assert events["event_count"] == len(expected) assert events["producer_count"] > 0 - assert events["consumer_count"] > 0 + assert events["consumer_count"] == 16 + assert events["dynamic_consumers"] == [ + {"caller": "app.workflow", "count": 1} + ] + assert all( + not item["caller"].startswith("app.plugins") + for contract in events["events"].values() + for item in contract["consumers"] + ) diff --git a/tests/test_architecture_dependencies.py b/tests/test_architecture_dependencies.py index e290fe8af..bbf82758d 100644 --- a/tests/test_architecture_dependencies.py +++ b/tests/test_architecture_dependencies.py @@ -3,6 +3,9 @@ import json from functools import lru_cache from pathlib import Path +from scripts.architecture.baseline import ( + collect_event_contracts as _collect_event_contracts, +) from scripts.architecture.baseline import ( discover_modules as _discover_modules, ) @@ -1285,21 +1288,15 @@ def test_application_services_do_not_resolve_event_manager_singleton(): def test_http_endpoints_do_not_register_process_event_listeners(): """HTTP 端点不得拥有进程级事件监听器，监听装配必须留在 startup。""" - violations: dict[str, set[str]] = {} - for module_name, path in _discover_modules().items(): - if not module_name.startswith("app.api"): - continue - tree = ast.parse(path.read_text(encoding="utf-8"), filename=str(path)) - calls = { - node.func.attr - for node in ast.walk(tree) - if isinstance(node, ast.Call) - and isinstance(node.func, ast.Attribute) - and node.func.attr in {"register", "add_event_listener"} - } - if calls: - violations[module_name] = calls - assert violations == {} + contracts = _collect_event_contracts() + callers = { + item["caller"] + for event in contracts["events"].values() + for item in event["consumers"] + } + callers.update(item["caller"] for item in contracts["dynamic_consumers"]) + + assert {caller for caller in callers if caller.startswith("app.api")} == set() def test_agent_tools_do_not_import_entrypoint_internals(): diff --git a/tests/test_architecture_event_consumers.py b/tests/test_architecture_event_consumers.py new file mode 100644 index 000000000..d8b5eae5d --- /dev/null +++ b/tests/test_architecture_event_consumers.py @@ -0,0 +1,639 @@ +"""Event consumer AST collector tests.""" + +from pathlib import Path + +from scripts.architecture.event_consumers import collect_event_consumers + +EVENT_MEMBERS = { + "EventType": ("Alpha", "Beta", "Gamma"), + "ChainEventType": ("Delta", "Epsilon"), +} + + +def _collect( + tmp_path: Path, + source: str, + *, + module_name: str = "app.sample", +) -> tuple[dict[str, list[dict]], list[dict]]: + """从单个临时宿主模块收集 Event consumer。""" + path = tmp_path / f"{module_name.replace('.', '_')}.py" + path.write_text(source, encoding="utf-8") + return collect_event_consumers({module_name: path}, EVENT_MEMBERS) + + +def _stable_locations(locations: list[dict]) -> list[dict]: + """移除仅用于诊断的行号，比较稳定 consumer identity。""" + return [{key: value for key, value in item.items() if key != "line"} for item in locations] + + +def test_collect_event_consumers_resolves_canonical_instances_and_aliases( + tmp_path: Path, +) -> None: + """Canonical singleton、构造器、模块和简单赋值别名都应可证明。""" + static, dynamic = _collect( + tmp_path, + ''' +from app.runtime.events import EventManager as Bus, eventmanager as global_bus +from app.schemas.types import EventType as Events, ChainEventType as ChainEvents +import app.runtime.events as runtime_events +import app.schemas.types as schema_types + +alias = global_bus +constructed = Bus() +existing = Bus.get_existing_instance() + +def listener(event): + pass + +@alias.register(Events.Alpha, priority=3) +def decorated(event): + pass + +@global_bus.register(etype=ChainEvents.Epsilon) +def keyword_decorated(event): + pass + +constructed.add_event_listener( + event_type=ChainEvents.Delta, + handler=listener, + priority=7, +) +runtime_events.eventmanager.add_event_listener( + schema_types.EventType.Beta, + listener, +) +existing.add_event_listener(Events.Gamma, listener, 11) +''', + ) + + assert dynamic == [] + assert _stable_locations(static["EventType.Alpha"]) == [ + { + "caller": "app.sample", + "handler": "decorated", + "identity": "decorator|decorated|3", + "priority": "3", + "registration_kind": "decorator", + } + ] + assert _stable_locations(static["EventType.Beta"]) == [ + { + "caller": "app.sample", + "handler": "listener", + "identity": "listener|listener|", + "priority": "", + "registration_kind": "listener", + } + ] + assert static["EventType.Gamma"][0]["priority"] == "11" + assert static["ChainEventType.Delta"][0]["priority"] == "7" + assert static["ChainEventType.Epsilon"][0]["handler"] == "keyword_decorated" + + +def test_collect_event_consumers_only_counts_applied_register_decorators( + tmp_path: Path, +) -> None: + """register 只有装饰声明或立即应用到 handler 时才形成 consumer。""" + static, dynamic = _collect( + tmp_path, + ''' +from app.runtime.events import eventmanager +from app.schemas.types import EventType + +unused = eventmanager.register(EventType.Alpha) +used = eventmanager.register(EventType.Beta, priority=4) + +@used +def via_alias(event): + pass + +def immediate(event): + pass + +eventmanager.register(EventType.Gamma, priority=8)(immediate) +''', + ) + + assert dynamic == [] + assert "EventType.Alpha" not in static + assert static["EventType.Beta"][0]["handler"] == "via_alias" + assert static["EventType.Beta"][0]["registration_kind"] == "decorator" + assert static["EventType.Gamma"][0]["handler"] == "immediate" + assert static["EventType.Gamma"][0]["registration_kind"] == "decorator" + + +def test_collect_event_consumers_expands_enum_and_partial_dynamic_lists( + tmp_path: Path, +) -> None: + """register 的 enum/list 静态成员和未知成员必须分别保留。""" + static, dynamic = _collect( + tmp_path, + ''' +from app.runtime.events import eventmanager +from app.schemas.types import EventType, ChainEventType + +runtime_event = EventType("runtime") +selected = [EventType.Alpha, runtime_event, ChainEventType.Delta] + +@eventmanager.register(EventType) +def all_broadcast(event): + pass + +@eventmanager.register(selected, priority=5) +def selected_events(event): + pass +''', + ) + + assert set(static) == { + "ChainEventType.Delta", + "EventType.Alpha", + "EventType.Beta", + "EventType.Gamma", + } + assert [item["handler"] for item in static["EventType.Alpha"]] == [ + "all_broadcast", + "selected_events", + ] + assert _stable_locations(dynamic) == [ + { + "caller": "app.sample", + "handler": "selected_events", + "identity": "decorator|selected_events|5", + "priority": "5", + "registration_kind": "decorator", + } + ] + + +def test_collect_event_consumers_ignores_unknown_receivers_and_internal_delegation( + tmp_path: Path, +) -> None: + """同名 API 和 EventManager 实现内部转调都不是独立 consumer 声明。""" + static, dynamic = _collect( + tmp_path, + ''' +from app.schemas.types import EventType + +registry.register(EventType.Alpha) +registry.add_event_listener(EventType.Beta, handler) + +class EventManager: + def register(self, event): + self.add_event_listener(event, handler) +''', + module_name="app.runtime.events", + ) + + assert static == {} + assert dynamic == [] + + +def test_collect_event_consumers_respects_shadowing_and_rebinding( + tmp_path: Path, +) -> None: + """参数遮蔽、局部重绑和模块最终重绑不得伪造 manager provenance。""" + static, dynamic = _collect( + tmp_path, + ''' +from app.runtime.events import eventmanager as bus +from app.schemas.types import EventType + +@bus.register(EventType.Alpha) +def before_rebind(event): + pass + +def listener(event): + pass + +def shadow(bus): + bus.add_event_listener(EventType.Beta, listener) + +def local_alias(): + local = original_bus + local.add_event_listener(EventType.Beta, listener) + local = object() + local.add_event_listener(EventType.Gamma, listener) + +def final_global_binding(): + bus.add_event_listener(EventType.Gamma, listener) + +original_bus = bus +bus = object() +''', + ) + + assert dynamic == [] + assert set(static) == {"EventType.Alpha", "EventType.Beta"} + assert static["EventType.Alpha"][0]["handler"] == "before_rebind" + assert static["EventType.Beta"][0]["handler"] == "listener" + + +def test_collect_event_consumers_handles_type_checking_runtime_else( + tmp_path: Path, +) -> None: + """TYPE_CHECKING body 只用于类型，else 才是运行期 provenance。""" + static, dynamic = _collect( + tmp_path, + ''' +from typing import TYPE_CHECKING +from app.schemas.types import EventType + +if TYPE_CHECKING: + from app.runtime.events import eventmanager as type_bus +else: + from app.runtime.events import eventmanager as runtime_bus + +def listener(event): + pass + +type_bus.add_event_listener(EventType.Alpha, listener) +runtime_bus.add_event_listener(EventType.Beta, listener) +''', + ) + + assert dynamic == [] + assert set(static) == {"EventType.Beta"} + + +def test_collect_event_consumers_requires_canonical_type_checking_provenance( + tmp_path: Path, +) -> None: + """只有未遮蔽的 typing.TYPE_CHECKING 才能裁剪运行期分支。""" + static, dynamic = _collect( + tmp_path, + ''' +from typing import TYPE_CHECKING as TC +from app.runtime.events import eventmanager +from app.schemas.types import EventType + +def type_only(event): + pass + +def runtime_only(event): + pass + +def shadowed_only(event): + pass + +def rebound_only(event): + pass + +if TC: + eventmanager.add_event_listener(EventType.Alpha, type_only) +else: + eventmanager.add_event_listener(EventType.Beta, runtime_only) + +def shadowed(TC): + if TC: + eventmanager.add_event_listener(EventType.Gamma, shadowed_only) + +TYPE_CHECKING = True +if TYPE_CHECKING: + eventmanager.add_event_listener(EventType.Alpha, rebound_only) +''', + ) + + assert dynamic == [] + assert set(static) == { + "EventType.Alpha", + "EventType.Beta", + "EventType.Gamma", + } + assert [item["handler"] for item in static["EventType.Alpha"]] == [ + "rebound_only" + ] + assert static["EventType.Beta"][0]["handler"] == "runtime_only" + assert static["EventType.Gamma"][0]["handler"] == "shadowed_only" + + +def test_collect_event_consumers_respects_compound_scope_targets( + tmp_path: Path, +) -> None: + """循环、异常和推导式目标必须遮蔽同名 EventManager alias。""" + static, dynamic = _collect( + tmp_path, + ''' +from app.runtime.events import eventmanager as bus +from app.schemas.types import EventType + +def listener(event): + pass + +def scoped(values): + local = bus + for local in values: + local.add_event_listener(EventType.Alpha, listener) + local.add_event_listener(EventType.Beta, listener) + + try: + raise RuntimeError + except RuntimeError as bus: + bus.add_event_listener(EventType.Alpha, listener) + + [bus.add_event_listener(EventType.Beta, listener) for bus in values] + +bus.add_event_listener(EventType.Gamma, listener) +''', + ) + + assert dynamic == [] + assert set(static) == {"EventType.Gamma"} + + +def test_collect_event_consumers_merges_try_exception_entry_states( + tmp_path: Path, +) -> None: + """异常 handler 必须从 try 内所有可能抛出点的合并状态开始。""" + static, dynamic = _collect( + tmp_path, + ''' +from app.runtime.events import eventmanager +from app.schemas.types import EventType + +class Other: + def add_event_listener(self, *args): + pass + +def listener(event): + pass + +bus = eventmanager +try: + bus = Other() + raise RuntimeError +except RuntimeError: + bus.add_event_listener(EventType.Alpha, listener) + +bus = eventmanager +try: + raise RuntimeError +except RuntimeError: + bus.add_event_listener(EventType.Beta, listener) +''', + ) + + assert dynamic == [] + assert set(static) == {"EventType.Beta"} + + +def test_collect_event_consumers_models_while_with_match_and_delete_bindings( + tmp_path: Path, +) -> None: + """复合语句目标和 del 必须遵循 Python 的绑定及路径语义。""" + static, dynamic = _collect( + tmp_path, + ''' +from contextlib import nullcontext +from app.runtime.events import eventmanager as bus +from app.schemas.types import EventType + +global_bus = bus + +class Other: + def add_event_listener(self, *args): + pass + +def listener(event): + pass + +def scoped(context): + bus = Other() + while False: + bus = global_bus + bus.add_event_listener(EventType.Alpha, listener) + + bus = global_bus + with nullcontext(Other()) as bus: + bus.add_event_listener(EventType.Alpha, listener) + bus.add_event_listener(EventType.Alpha, listener) + + bus = global_bus + match Other(): + case bus: + bus.add_event_listener(EventType.Alpha, listener) + + bus = global_bus + while context: + bus.add_event_listener(EventType.Beta, listener) + break + with nullcontext(): + bus.add_event_listener(EventType.Gamma, listener) + +async def async_scoped(context): + bus = global_bus + async with context as bus: + bus.add_event_listener(EventType.Alpha, listener) + +def deleted_name_is_local(): + global_bus.add_event_listener(EventType.Beta, listener) + bus.add_event_listener(EventType.Alpha, listener) + del bus +''', + ) + + assert dynamic == [] + assert set(static) == {"EventType.Beta", "EventType.Gamma"} + assert len(static["EventType.Beta"]) == 2 + + +def test_collect_event_consumers_writes_walrus_to_containing_scope( + tmp_path: Path, +) -> None: + """推导式海象绑定属于包含函数，普通外层 alias 仍可读取。""" + static, dynamic = _collect( + tmp_path, + ''' +from app.runtime.events import eventmanager as bus +from app.schemas.types import EventType + +global_bus = bus + +def listener(event): + pass + +def rebound(values): + [(bus := object()) for value in values] + bus.add_event_listener(EventType.Alpha, listener) + +def retained(values): + bus = global_bus + [bus.add_event_listener(EventType.Beta, listener) for value in values] + +def eager_iterable(): + [value for value in ((bus := global_bus),)] + bus.add_event_listener(EventType.Gamma, listener) +''', + ) + + assert dynamic == [] + assert set(static) == {"EventType.Beta", "EventType.Gamma"} + + +def test_collect_event_consumers_uses_late_bound_function_closures( + tmp_path: Path, +) -> None: + """嵌套函数和 lambda 必须读取包含函数的最终 cell provenance。""" + static, dynamic = _collect( + tmp_path, + ''' +from app.runtime.events import eventmanager as global_bus +from app.schemas.types import EventType + +def listener(event): + pass + +def rebound(): + bus = global_bus + def nested(): + bus.add_event_listener(EventType.Alpha, listener) + callback = lambda: bus.add_event_listener(EventType.Alpha, listener) + bus = object() + return nested, callback + +def retained(): + bus = global_bus + def nested(): + bus.add_event_listener(EventType.Beta, listener) + callback = lambda: bus.add_event_listener(EventType.Gamma, listener) + return nested, callback +''', + ) + + assert dynamic == [] + assert set(static) == {"EventType.Beta", "EventType.Gamma"} + + +def test_collect_event_consumers_validates_decorator_factory_application( + tmp_path: Path, +) -> None: + """保存后的 factory 仅在合法单参数应用时形成 consumer。""" + static, dynamic = _collect( + tmp_path, + ''' +from app.runtime.events import eventmanager +from app.schemas.types import EventType + +def positional(event): + pass + +def keyword(event): + pass + +factory = eventmanager.register(EventType.Beta, priority=4) +factory(positional) +eventmanager.register(EventType.Gamma)(f=keyword) + +unused = eventmanager.register(EventType.Alpha) +unused() +unused(handler=positional) +unused(positional, keyword) +unused(positional, f=keyword) +''', + ) + + assert dynamic == [] + assert set(static) == {"EventType.Beta", "EventType.Gamma"} + assert static["EventType.Beta"][0]["handler"] == "positional" + assert static["EventType.Gamma"][0]["handler"] == "keyword" + + +def test_collect_event_consumers_validates_real_event_manager_signatures( + tmp_path: Path, +) -> None: + """两种注册 API 只接受各自真实且完整的参数绑定。""" + static, dynamic = _collect( + tmp_path, + ''' +from app.runtime.events import eventmanager +from app.schemas.types import EventType + +def listener(event): + pass + +eventmanager.add_event_listener( + event_type=EventType.Alpha, + handler=listener, + priority=3, +) + +@eventmanager.register(etype=EventType.Beta, priority=4) +def decorated(event): + pass + +eventmanager.add_event_listener(etype=EventType.Gamma, handler=listener) +eventmanager.add_event_listener(EventType.Gamma) +eventmanager.add_event_listener( + EventType.Gamma, + listener, + handler=listener, +) +eventmanager.add_event_listener(EventType.Gamma, listener, unknown=True) + +@eventmanager.register(event_type=EventType.Gamma) +def wrong_keyword(event): + pass + +@eventmanager.register() +def missing_event(event): + pass +''', + ) + + assert dynamic == [] + assert set(static) == {"EventType.Alpha", "EventType.Beta"} + assert static["EventType.Alpha"][0]["priority"] == "3" + assert static["EventType.Beta"][0]["priority"] == "4" + + +def test_collect_event_consumers_keeps_only_proven_dynamic_registrations( + tmp_path: Path, +) -> None: + """只有 receiver 已证明而事件值未知时才进入 dynamic facts。""" + static, dynamic = _collect( + tmp_path, + ''' +from app.runtime.events import eventmanager +from app.schemas.types import EventType + +def register_runtime(name, handler): + event_type = EventType(name) + eventmanager.add_event_listener(event_type=event_type, handler=handler) + unrelated.add_event_listener(event_type, handler) + unrelated.register(event_type)(handler) +''', + ) + + assert static == {} + assert len(dynamic) == 1 + assert dynamic[0]["caller"] == "app.sample" + assert dynamic[0]["handler"] == "handler" + assert dynamic[0]["registration_kind"] == "listener" + + +def test_collect_event_consumers_excludes_plugin_modules(tmp_path: Path) -> None: + """即使调用方误传插件模块，collector 也不得读取其注册事实。""" + host_path = tmp_path / "host.py" + plugin_path = tmp_path / "plugin.py" + source = ''' +from app.runtime.events import eventmanager +from app.schemas.types import EventType + +@eventmanager.register(EventType.Alpha) +def handler(event): + pass +''' + host_path.write_text(source, encoding="utf-8") + plugin_path.write_text(source.replace("Alpha", "Beta"), encoding="utf-8") + + static, dynamic = collect_event_consumers( + { + "app.host": host_path, + "app.plugins.sample": plugin_path, + }, + EVENT_MEMBERS, + ) + + assert dynamic == [] + assert set(static) == {"EventType.Alpha"}