refactor(architecture): unify event facts and policy

.github/workflows/test.yml

@@ -41,13 +41,20 @@ jobs:
       - name: Install dependencies
         run: uv sync --locked
 
-      - name: Check host architecture contracts
+      - name: Check event semantic policy
         run: |
           uv run --locked --no-sync pytest \
             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_event_facts.py \
+            tests/test_architecture_event_policy.py -q
+          uv run --locked --no-sync python \
+            scripts/architecture/event_policy.py
+
+      - name: Check host architecture snapshot
+        run: |
+          uv run --locked --no-sync pytest \
             tests/test_architecture_contract_baseline.py \
             tests/test_architecture_baseline_cli.py -q
           uv run --locked --no-sync python \

docs/architecture-optimization-checklist.md

@@ -2,7 +2,7 @@
 
 > 审计日期：2026-08-26
 >
-> 审计基线：`v3@9053db926d20`，与 `origin/v3` 一致
+> 首次审计历史快照：`v3@9053db926d20`；当前交付状态以路线图和生成 fixture 为准
 >
 > 文档性质：当前源码的差距清单与分阶段执行说明，不是历史重构结项账本
 
@@ -69,7 +69,7 @@ MoviePilot V3 已经形成较清晰的模块化单体：`foundation`、`domain`
 
 | 指标 | 当前值 | 解释 |
 |---|---:|---|
-| 宿主 Python 模块 / 内部依赖边 | 835 / 6,810 | `dependency-baseline.json` 当前快照 |
+| 宿主 Python 模块 / 内部依赖边 | 835 / 6,817 | `dependency-baseline.json` 当前快照 |
 | 非平凡 SCC | 2 | 新增 Chain 包根环；另一个是隔离的 29 模块 TMDB 移植包环 |
 | 跨层 DB 边界债务 | 0 | Application、Chain、API、Agent、Runtime、Workflow 到 DB 的受控债务均为零 |
 | Model/Oper 事务债务 | 0 | 自建 Session、自动事务装饰器、直接 commit/rollback 等基线均为零 |
@@ -78,8 +78,8 @@ MoviePilot V3 已经形成较清晰的模块化单体：`foundation`、`domain`
 | Python 源码量 | 约 271,400 行 | 60 个文件超过 1,000 行，14 个超过 2,000 行 |
 | 长方法 | 281 个超过 80 行 | 67 个超过 150 行，23 个超过 250 行；大量是私有方法 |
 | 全量 mypy 历史债务 | 11,983 / 601 文件 | strict frontier 当前只覆盖 41 个文件，且 ratchet 已新增 2 个错误 |
-| Ruff 历史诊断 | 973 | 低水位门禁通过，但规则集只覆盖 `E4/E7/E9/F/I` |
-| 覆盖率低水位 | Application 77.76%，Domain 79.24% | Chain、Runtime、Agent、Adapter、Startup 未进入包级覆盖率门禁 |
+| Ruff 历史诊断 | 972 | 低水位门禁通过，但规则集只覆盖 `E4/E7/E9/F/I` |
+| 覆盖率低水位 | Application 77.82%，Domain 79.24% | Chain、Runtime、Agent、Adapter、Startup 未进入包级覆盖率门禁 |
 
 ### 3.3 热点文件
 
@@ -160,15 +160,17 @@ MoviePilot V3 已经形成较清晰的模块化单体：`foundation`、`domain`
 - 架构总览此前仍记录 811 模块、6,572 条边和 1 个 SCC，已经落后于当前基线。
 - Event consumer 扫描曾把任意同名 `.register()` 调用当成事件注册；S0-L2.5 已改为证明
   canonical EventManager receiver，10 个动态误报归零并保留唯一 workflow 动态注册。
+- S0-L2.6 已将 producer/consumer 合并为逐调用事实源：99 个 producer（98 静态、1 动态）与
+  17 个 consumer（16 静态、1 动态）；consumer 由不可自动写入的精确人工 policy 管理。
 
 **目标与步骤**
 
-- [ ] 指定一个机器可读事实源，文档指标由 fixture 生成或只保留不易漂移的语义描述。
+- [x] 指定统一 Event 机器事实源；生成快照与人工 consumer policy 分离且互相不能覆盖。
 - [x] 修正 Oper 示例，分别展示宿主显式 UoW 与插件兼容 Facade，并以文档测试禁止回退。
 - [x] 明确 Application/Chain 不永久直连具体 Adapter；业务层拥有 Port，startup 注入实现。
 - [x] 让 SCC 规则、精确 policy 和文档声明一致；Chain 临时债务与 TMDB vendor containment 分开治理。
 - [x] Event 扫描只识别 EventManager 实例/别名和事件装饰器；未知 receiver 不再污染动态事实。
-- [ ] CI 分开报告“快照一致”与“语义规则通过”，禁止把前者表述为架构完全正确。
+- [x] CI 分开报告“Event 语义 policy”与“宿主快照一致”，禁止把后者表述为架构完全正确。
 
 **验收**
 
@@ -496,8 +498,7 @@ MoviePilot V3 已经形成较清晰的模块化单体：`foundation`、`domain`
 - 复杂度脚本只检查 API、Application、Chain 的公共入口；私有长方法、类/文件规模和圈复杂度不受控。
 - strict mypy 仅 41 个文件，高风险 lifecycle、Scheduler、Agent、Plugin Manager 多数不在 frontier。
 - coverage ratchet 只聚合 Application 和 Domain。
-- Task owner gate 不盘点原生并发原语；Event producer 别名/关键字识别和 consumer 人工 policy
-  尚未纳入统一事实源门禁。
+- Task owner gate 尚未盘点原生并发原语；Event producer/consumer 已纳入统一事实源和人工 policy。
 - Ruff 仅是有限规则集的历史低水位，不代表整体风格/正确性无债务。
 
 **目标与步骤**
@@ -508,7 +509,7 @@ MoviePilot V3 已经形成较清晰的模块化单体：`foundation`、`domain`
 - [ ] coverage 增加 Chain、Runtime、Agent、Startup 的高风险子包或关键文件组，不用低价值行数冲百分比。
 - [ ] 原生并发门禁按 ARCH-101/106 修正。
 - [x] Event consumer 扫描证明 canonical EventManager receiver，清除同名方法误报。
-- [ ] Event producer 识别和 consumer zero-growth policy 在 S0-L2.6 纳入统一事实源门禁。
+- [x] Event producer 别名/关键字/有限条件识别和 consumer exact policy 纳入统一事实源门禁。
 - [ ] Module Quality Scale 增加 capability -> required rules -> evidence tests 映射，避免“已登记”等同“已验证”。
 - [ ] 修改 CI 或门禁脚本时同时运行 `tests/test_architecture_ci.py` 和对应脚本单元测试。
 

docs/architecture-overview.md

@@ -7,7 +7,7 @@
 > [`docs/rules/04-design-patterns.md`](rules/04-design-patterns.md) 为准，本文与其保持一致；
 > 如出现差异，以规则文档为准。
 >
-> *Last Updated: 2026-08-24*
+> *Last Updated: 2026-08-27*
 
 ---
 
@@ -369,6 +369,13 @@ collector 只接受 canonical `eventmanager`、`EventManager()` 及其有限别
 `register`/`add_event_listener`；当前宿主有 16 个静态注册点，另保留 1 个由工作流配置驱动的
 真实动态注册。`app/plugins/**` 插件副本不进入宿主事实。
 
+生产者与消费者共用 `scripts/architecture/event_facts.py` 这一份逐调用事实源。当前宿主有 99 个
+生产调用，其中 98 个静态解析为 100 个事件引用，只有 `Command.send_plugin_event` 的插件事件类型
+保持动态；17 个消费注册中 16 个静态、1 个动态。生成的
+`runtime-contract-baseline.json` 保存 line-free 事实、数量和枚举索引；人工维护的
+`runtime-contract-policy.json` 只批准 consumer 的精确 fingerprint、owner 和理由，任何新增、替换、
+重复或陈旧项都会失败，`--write-host` 不会改写该 policy。
+
 ```python
 from app.sdk.events import snapshot_event_data
 
@@ -683,6 +690,9 @@ flowchart LR
   stream/vendor/diagnostic/control-plane 事实是精确 containment。每条初始边的指纹由测试独立冻结，
   bindings/uses 变化、分类互换、通配导入和初始边增长都会失败；债务删除时同步删除冻结项以禁止恢复，
   `--write-host` 不会改写人工 policy 或冻结上界。
+- `event_facts` 是生产者/消费者唯一收集源；运行快照记录 99 个生产调用和 17 个消费注册，
+  consumer 的 17 个唯一 fingerprint 另由只读人工 policy 精确准入。CI 将语义 policy 与生成快照
+  分成独立步骤，前者不能通过刷新后者绕过。
 - 任何所有权迁移必须同步更新：canonical 导入、`app/runtime/compat/manifest.py`、
   SDK 导出（若公开）、`docs/rules/05-architecture.md` 与上述架构测试。
 - 延迟导入不被接受为隐藏循环依赖的手段。
@@ -695,17 +705,18 @@ flowchart LR
 | 指标 | 当前值 |
 |---|---:|
 | Python 模块 | 835 |
-| 内部导入边 | 6,810 |
+| 内部导入边 | 6,817 |
 | 非平凡 SCC | 2（`ARCH-107` 临时 Chain 包根环；精确 containment 的 TMDB 移植包环） |
 | Direct egress | 66（12 条待迁移债务，54 条精确 containment） |
 | Module Contract V2 spec | 215（其中 214 个进入 `run_module` 观察面） |
 | Event Contract | 53 |
+| Event producer / consumer | 99（98 静态、1 动态）/ 17（16 静态、1 动态） |
 | Model/Oper 自动事务与自建 Session | 0 |
 | 组合根外 `SystemConfigOper()` | 0 |
 
-架构专项验证：`tests/test_architecture_dependencies.py` 与
-`tests/test_architecture_contract_baseline.py` 共同提供语义断言和快照比较；
-`scripts/architecture/baseline.py --check-host` 只验证快照一致，不能替代依赖合理性审查。
+架构专项验证分为两个 CI 投影：`Check event semantic policy` 先运行依赖、Adapter、出口和 Event
+语义门禁，`Check host architecture snapshot` 再执行快照测试及一次
+`scripts/architecture/baseline.py --check-host`。快照一致只说明事实未漂移，不能替代边界合理性审查。
 
 本总览与本轮架构治理的关系如下：
 

docs/architecture-refactor-roadmap.md

@@ -78,8 +78,8 @@ G-ARCH 只有在以下条件全部满足后才可完成：
 | 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 事实与政策 | `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 与快照一致性 |
+| S0-L2.5 Event consumer 识别 | `DELIVERED` | S0-L2.1 | `86157be2a`：consumer 只识别可静态证明的 EventManager 注册；全量 6,459 passed / 6 skipped，CI `33029645165`/`33029645254` 全绿，远端 `0/0` |
+| S0-L2.6 事实源与 CI 投影 | `VERIFIED` | S0-L2.2,S0-L2.4b,S0-L2.5 | 99/17 条逐调用事实、17 条 consumer policy 与 CI 分层本地通过；全量 6,481 passed / 6 skipped，待推送 CI |
 
 ### S1：可靠性、事务与数据合同
 
@@ -136,7 +136,7 @@ G-ARCH 只有在以下条件全部满足后才可完成：
 | S4-L2 Event strict contract | `PLANNED` | S0-L5,S1-L6 | 宿主事件输入/输出按风险 strict，诊断例外只属于第三方插件兼容 |
 | S4-L3 Complexity v2 | `PLANNED` | S3 | 私有方法、class/file、圈复杂度进入门禁；所有超限通过职责拆分归零 |
 | S4-L4 全量 mypy 清零 | `PLANNED` | S3,S4-L1,S4-L2 | `mypy-baseline.json` 归零并删除债务接受路径，全宿主 strict 类型通过 |
-| S4-L5 Ruff 治理债务清零 | `PLANNED` | S3 | 当前受控 973 条诊断归零，规则集扩展经过独立审查且新增诊断为零 |
+| S4-L5 Ruff 治理债务清零 | `PLANNED` | S3 | 当前受控 972 条诊断归零，规则集扩展经过独立审查且新增诊断为零 |
 | S4-L6 Coverage/并发/质量证据 | `PLANNED` | S3,S4-L1,S4-L2 | 高风险包纳入 coverage；raw concurrency 分类清零；Module Quality 有真实 evidence test |
 
 ### S5：Plugin、Agent、Domain、Startup 与最终收口
@@ -154,48 +154,49 @@ G-ARCH 只有在以下条件全部满足后才可完成：
 
 ## 4. 当前活动叶子
 
-### S0-L2.5 Event consumer 识别
+### S0-L2.6 事实源与 CI 投影
 
 **Status:** `VERIFIED`（本地验收完成，等待提交、推送和远端 CI 确认）
 
 **Outcome**
 
-把 Event consumer 从“末级方法名碰巧是 `register`/`add_event_listener`”收紧为可静态证明的
-canonical `EventManager` receiver。16 个静态注册点保持不变；10 个 selector、SDK hook、Oper、
-Model、TaskRegistry、`atexit` 和 EventManager 内部展开误报归零；配置驱动的 workflow 注册是唯一
-真实动态 consumer。
+统一 Event producer/consumer 的 AST 事实源，完整解析 positional/keyword 参数、别名、重绑定和
+有限条件表达式。生成快照保存逐调用 line-free 事实及 multiplicity；consumer 由独立人工 policy
+按 exact fingerprint set 准入，任何刷新快照的操作都不能自动接受新消费注册。
 
 **Ownership**
 
-- `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 投影。
-- 本路线图的叶子状态和交付记录。
+- `scripts/architecture/event_facts.py` 的统一 producer/consumer provenance collector。
+- `scripts/architecture/event_policy.py` 与 `runtime-contract-policy.json` 的只读人工 consumer policy。
+- `scripts/architecture/baseline.py` 的 runtime schema v3、迁移链、事实索引与 diagnostics。
+- producer/consumer、policy、baseline/CLI 和 CI 分层测试。
+- 架构规范、总览、优化清单与本路线图的单一事实说明。
 
 **Excluded**
 
 - 不修改生产 EventManager、事件 ABI、handler 执行顺序或插件消费者。
 - 不把 `app/plugins/**` 副本纳入宿主扫描。
-- producer 关键字/别名解析与 consumer 人工 zero-growth policy 留给 S0-L2.6 统一事实源叶；本叶只把
-  consumer 识别结果变为真实、完整且可测试的事实。
+- 不使用源码行号、通配符或自动写入 policy 接受新 consumer。
 
 **Acceptance**
 
 ```bash
 .venv/bin/python -m pytest \
-  tests/test_architecture_event_consumers.py \
+  tests/test_architecture_event_facts.py \
+  tests/test_architecture_event_policy.py \
   tests/test_architecture_dependencies.py \
   tests/test_architecture_contract_baseline.py \
   tests/test_architecture_baseline_cli.py \
   tests/test_architecture_ci.py -q
+.venv/bin/python scripts/architecture/event_policy.py
 .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/event_consumers.py \
-  tests/test_architecture_event_consumers.py \
+  scripts/architecture/event_facts.py \
+  scripts/architecture/event_policy.py \
+  tests/test_architecture_event_facts.py \
+  tests/test_architecture_event_policy.py \
   tests/test_architecture_dependencies.py \
   tests/test_architecture_contract_baseline.py \
   tests/test_architecture_baseline_cli.py \
@@ -205,5 +206,5 @@ git diff --check
 
 **Delivery**
 
-- 单一提交主题：以可证明的 EventManager provenance 替换同名方法扫描并清除全部动态误报。
+- 单一提交主题：统一 Event facts、锁定 consumer policy 并拆分 CI 语义/快照投影。
 - 推送 `origin/v3` 后确认提交祖先关系、远端 SHA 和 ahead/behind `0/0`。

docs/rules/05-architecture.md

@@ -569,16 +569,25 @@ 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
+Event producer and consumer facts share `scripts/architecture/event_facts.py` as
+their only collector. A send/register method name alone is not evidence: the
+receiver must resolve to the canonical `app.runtime.events.eventmanager`, an
+`EventManager` instance, or a proven injected Event publisher/manager port.
+Unknown same-name receivers are ignored. Positional and keyword event arguments,
+finite aliases and conditional enum choices must be resolved; only a proven
+receiver whose event value remains unknowable may produce a dynamic fact. The
 collector respects lexical shadowing and rebinds, distinguishes decorator
 application from obtaining a decorator factory, and never scans `app/plugins/**`
 as host code.
 
+The generated runtime baseline preserves every line-free call fact and its
+multiplicity. Consumer admission is separate and non-generated: every current
+consumer fingerprint, owner, classification and concrete reason must exactly
+match `runtime-contract-policy.json`. New, changed, duplicate, invalid and stale
+consumer identities fail independently, and `--write-host` never modifies the
+policy. The current host permits one dynamic consumer only: the configuration-
+driven workflow registration.
+
 ## Key File Locations
 
 | Path | Purpose |

scripts/architecture/baseline.py

@@ -9,7 +9,7 @@ import importlib.util
 import json
 import subprocess
 import sys
-from collections import defaultdict
+from collections import Counter, defaultdict
 from pathlib import Path
 from typing import Any, Optional
 
@@ -19,9 +19,9 @@ except ModuleNotFoundError:
     from egress import collect_direct_egress
 
 try:
-    from scripts.architecture.event_consumers import collect_event_consumers
+    from scripts.architecture.event_facts import collect_event_facts
 except ModuleNotFoundError:
-    from event_consumers import collect_event_consumers
+    from event_facts import collect_event_facts
 
 PROJECT_ROOT = Path(__file__).resolve().parents[2]
 APP_ROOT = PROJECT_ROOT / "app"
@@ -701,17 +701,6 @@ def collect_run_module_diagnostics() -> dict[str, Any]:
     }
 
 
-def _event_reference(node: ast.AST) -> str | None:
-    """从 AST 节点解析 EventType/ChainEventType 的静态成员引用。"""
-    if (
-        isinstance(node, ast.Attribute)
-        and isinstance(node.value, ast.Name)
-        and node.value.id in {"EventType", "ChainEventType"}
-    ):
-        return f"{node.value.id}.{node.attr}"
-    return None
-
-
 def _event_enum_members(enum_name: str) -> tuple[str, ...]:
     """从 schema 源码读取事件枚举成员，避免基线脚本导入宿主运行时。"""
     tree = parse_source(APP_ROOT / "schemas" / "types.py")
@@ -738,120 +727,91 @@ def _event_enum_members(enum_name: str) -> tuple[str, ...]:
     )
 
 
-def _collect_event_locations() -> tuple[
-    list[str],
-    dict[str, list[dict[str, Any]]],
-    dict[str, list[dict[str, Any]]],
-    list[dict[str, Any]],
-    list[dict[str, Any]],
-]:
-    """扫描事件枚举及其生产、消费位置，供语义和诊断视图复用。"""
+def collect_current_event_facts() -> dict[str, list[dict[str, Any]]]:
+    """收集排除插件副本的当前宿主 Event producer/consumer 事实。"""
     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)
-    dynamic_producers: list[dict[str, Any]] = []
-    consumers, dynamic_consumers = collect_event_consumers(
-        modules,
+    return collect_event_facts(
+        discover_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(
-                node.func,
-                ast.Attribute,
-            ):
-                continue
-            location = {"caller": module_name, "line": node.lineno}
-            if node.func.attr in {"send_event", "async_send_event"}:
-                reference = _event_reference(node.args[0]) if node.args else None
-                if reference:
-                    producers[reference].append(location)
-                else:
-                    dynamic_producers.append(location)
 
-    enum_names = [
-        *(f"EventType.{member}" for member in event_members),
-        *(f"ChainEventType.{member}" for member in chain_event_members),
-    ]
-    return (
-        enum_names,
-        producers,
-        consumers,
-        dynamic_producers,
-        dynamic_consumers,
+
+def _line_free_event_fact(fact: dict[str, Any]) -> dict[str, Any]:
+    """移除只用于诊断的源码行号，保留完整稳定事件身份。"""
+    return {key: value for key, value in fact.items() if key != "line"}
+
+
+def collect_event_fact_contract() -> dict[str, Any]:
+    """生成逐调用事实与按 fingerprint 索引的宿主事件契约。"""
+    event_names = sorted([
+        *(
+            f"EventType.{member}"
+            for member in _event_enum_members("EventType")
+        ),
+        *(
+            f"ChainEventType.{member}"
+            for member in _event_enum_members("ChainEventType")
+        ),
+    ])
+    current = collect_current_event_facts()
+    producers = sorted(
+        (_line_free_event_fact(fact) for fact in current["producers"]),
+        key=lambda fact: (fact["caller"], fact["qualname"], fact["fingerprint"]),
     )
-
-
-def collect_event_contracts() -> dict[str, Any]:
-    """收集不受源码行号变化影响的宿主事件语义契约。"""
-    (
-        enum_names,
-        producers,
-        consumers,
-        dynamic_producers,
-        dynamic_consumers,
-    ) = _collect_event_locations()
-    contracts = {
-        name: {
-            "producers": _aggregate_locations(
-                producers.get(name, []),
-                ("caller",),
+    consumers = sorted(
+        (_line_free_event_fact(fact) for fact in current["consumers"]),
+        key=lambda fact: (fact["caller"], fact["qualname"], fact["fingerprint"]),
+    )
+    all_facts = (*producers, *consumers)
+    event_index = {
+        event_name: {
+            "producer_fingerprints": sorted(
+                fact["fingerprint"]
+                for fact in producers
+                if event_name in fact["events"]
             ),
-            "consumers": _aggregate_locations(
-                consumers.get(name, []),
-                ("caller",),
+            "consumer_fingerprints": sorted(
+                fact["fingerprint"]
+                for fact in consumers
+                if event_name in fact["events"]
             ),
         }
-        for name in sorted(enum_names)
+        for event_name in event_names
     }
     return {
-        "event_count": len(contracts),
-        "producer_count": sum(len(items) for items in producers.values()),
-        "consumer_count": sum(len(items) for items in consumers.values()),
-        "events": contracts,
-        "dynamic_producers": _aggregate_locations(dynamic_producers, ("caller",)),
-        "dynamic_consumers": _aggregate_locations(dynamic_consumers, ("caller",)),
+        "event_count": len(event_names),
+        "producer_call_count": len(producers),
+        "static_producer_call_count": sum(
+            not fact["dynamic"] and not fact["invalid"] for fact in producers
+        ),
+        "dynamic_producer_count": sum(fact["dynamic"] for fact in producers),
+        "invalid_producer_count": sum(fact["invalid"] for fact in producers),
+        "producer_event_reference_count": sum(
+            len(fact["events"]) for fact in producers
+        ),
+        "consumer_registration_count": len(consumers),
+        "static_consumer_count": sum(
+            not fact["dynamic"] and not fact["invalid"] for fact in consumers
+        ),
+        "dynamic_consumer_count": sum(fact["dynamic"] for fact in consumers),
+        "invalid_consumer_count": sum(fact["invalid"] for fact in consumers),
+        "consumer_event_reference_count": sum(
+            len(fact["events"]) for fact in consumers
+        ),
+        "fact_count": len(all_facts),
+        "producers": producers,
+        "consumers": consumers,
+        "event_index": event_index,
     }
 
 
-def collect_event_diagnostics() -> dict[str, Any]:
-    """收集事件生产与消费的当前源码位置，仅用于人工诊断。"""
-    (
-        enum_names,
-        producers,
-        consumers,
-        dynamic_producers,
-        dynamic_consumers,
-    ) = _collect_event_locations()
-    return {
-        "events": {
-            name: {
-                "producers": sorted(
-                    producers.get(name, []),
-                    key=lambda item: (item["caller"], item["line"]),
-                ),
-                "consumers": sorted(
-                    consumers.get(name, []),
-                    key=lambda item: (item["caller"], item["line"]),
-                ),
-            }
-            for name in sorted(enum_names)
-        },
-        "dynamic_producers": sorted(
-            dynamic_producers,
-            key=lambda item: (item["caller"], item["line"]),
-        ),
-        "dynamic_consumers": sorted(
-            dynamic_consumers,
-            key=lambda item: (item["caller"], item["line"]),
-        ),
-    }
+def collect_event_fact_diagnostics() -> dict[str, list[dict[str, Any]]]:
+    """返回带源码行号的逐调用事件事实，仅用于人工诊断。"""
+    return collect_current_event_facts()
 
 
 def _sdk_all_names(tree: ast.Module, path: Path) -> tuple[str, ...]:
@@ -1021,10 +981,15 @@ def collect_compat_manifest() -> dict[str, Any]:
 def collect_runtime_baseline() -> dict[str, Any]:
     """生成模块调度、SDK 和兼容层公开契约基线。"""
     return {
-        "schema_version": 2,
+        "schema_version": 3,
+        "scope": {
+            "repository": "MoviePilot",
+            "roots": ["app"],
+            "excluded": ["app/plugins"],
+        },
         "run_module": collect_run_module_contracts(),
         "module_method_specs": collect_module_method_specs(),
-        "events": collect_event_contracts(),
+        "event_facts": collect_event_fact_contract(),
         "event_specs": collect_event_specs(),
         "sdk_exports": collect_sdk_exports(),
         "compat_manifest": collect_compat_manifest(),
@@ -1087,7 +1052,7 @@ def collect_runtime_diagnostics() -> dict[str, Any]:
     """生成带当前源码行号的运行契约诊断视图，不写入语义 fixture。"""
     return {
         "run_module": collect_run_module_diagnostics(),
-        "events": collect_event_diagnostics(),
+        "event_facts": collect_event_fact_diagnostics(),
     }
 
 
@@ -1295,10 +1260,8 @@ def _display_path(path: Path) -> Path:
         return path
 
 
-def _migrate_runtime_baseline(value: dict[str, Any]) -> dict[str, Any]:
-    """把包含源码行号的 v1 运行契约转换为稳定语义结构。"""
-    if value.get("schema_version") != 1:
-        return value
+def _migrate_runtime_v1_to_v2(value: dict[str, Any]) -> dict[str, Any]:
+    """把包含源码行号的 v1 运行契约转换为 v2 聚合语义。"""
     run_module = value["run_module"]
     events = value["events"]
     return {
@@ -1349,6 +1312,37 @@ def _migrate_runtime_baseline(value: dict[str, Any]) -> dict[str, Any]:
     }
 
 
+def _migrate_runtime_v2_to_v3(value: dict[str, Any]) -> dict[str, Any]:
+    """把 v2 聚合事件投影为显式待刷新的 v3 兼容视图。"""
+    return {
+        "schema_version": 3,
+        "scope": {
+            "repository": "MoviePilot",
+            "roots": ["app"],
+            "excluded": ["app/plugins"],
+        },
+        "run_module": value["run_module"],
+        "module_method_specs": value.get("module_method_specs", {}),
+        "event_facts": {
+            "migration_required": True,
+            "legacy_v2_projection": value["events"],
+        },
+        "event_specs": value.get("event_specs", {}),
+        "sdk_exports": value["sdk_exports"],
+        "compat_manifest": value["compat_manifest"],
+    }
+
+
+def _migrate_runtime_baseline(value: dict[str, Any]) -> dict[str, Any]:
+    """链式迁移旧运行契约，保证检查只报告语义变化而不崩溃。"""
+    migrated = value
+    if migrated.get("schema_version") == 1:
+        migrated = _migrate_runtime_v1_to_v2(migrated)
+    if migrated.get("schema_version") == 2:
+        migrated = _migrate_runtime_v2_to_v3(migrated)
+    return migrated
+
+
 def _migrate_plugin_baseline(value: dict[str, Any]) -> dict[str, Any]:
     """把来源信息混排的 v2 插件基线转换为 scope/provenance 结构。"""
     if value.get("schema_version") != 2:
@@ -1484,18 +1478,20 @@ def _compare_semantic_values(
             )
         return
     if isinstance(expected, list) and isinstance(actual, list):
-        expected_items = {
-            json.dumps(item, ensure_ascii=False, sort_keys=True): item
+        expected_keys = [
+            json.dumps(item, ensure_ascii=False, sort_keys=True)
             for item in expected
-        }
-        actual_items = {
-            json.dumps(item, ensure_ascii=False, sort_keys=True): item
+        ]
+        actual_keys = [
+            json.dumps(item, ensure_ascii=False, sort_keys=True)
             for item in actual
-        }
-        for key in sorted(expected_items.keys() - actual_items.keys()):
-            report["removed"].append({"path": path, "value": expected_items[key]})
-        for key in sorted(actual_items.keys() - expected_items.keys()):
-            report["added"].append({"path": path, "value": actual_items[key]})
+        ]
+        expected_counts = Counter(expected_keys)
+        actual_counts = Counter(actual_keys)
+        for key in sorted((expected_counts - actual_counts).elements()):
+            report["removed"].append({"path": path, "value": json.loads(key)})
+        for key in sorted((actual_counts - expected_counts).elements()):
+            report["added"].append({"path": path, "value": json.loads(key)})
         return
     if expected != actual:
         report["changed"].append(

scripts/architecture/event_facts.py

@@ -1,7 +1,9 @@
-"""静态收集可证明的宿主 EventManager consumer。"""
+"""静态收集可证明的宿主 EventManager producer 与 consumer 事实。"""
 
 import ast
-from collections import defaultdict
+import hashlib
+import json
+from collections.abc import Mapping
 from dataclasses import dataclass
 from pathlib import Path
 from typing import Any, Literal, TypeAlias
@@ -9,6 +11,7 @@ from typing import Any, Literal, TypeAlias
 _DEFAULT_IDENTITY = "<default>"
 _DYNAMIC_IDENTITY = "<dynamic>"
 _EVENT_MANAGER_METHODS = {"add_event_listener", "register"}
+_PRODUCER_METHODS = {"async_send_event", "send_event"}
 _COMPREHENSION_SCOPES = (
     ast.ListComp,
     ast.SetComp,
@@ -33,6 +36,8 @@ class _Symbol:
         "manager_class",
         "manager_factory",
         "manager_instance",
+        "publisher_instance",
+        "injected_owner",
         "type_checking",
     ]
     value: str = ""
@@ -45,6 +50,20 @@ class _EventSelection:
     events: tuple[str, ...]
     kind: Literal["member", "enum", "list"]
     dynamic: bool = False
+    invalid: bool = False
+
+
+@dataclass(frozen=True, slots=True)
+class _BoundEventMethod:
+    """记录已证明 receiver 的 EventManager 或 EventPublisher 绑定方法。"""
+
+    method: Literal[
+        "add_event_listener",
+        "register",
+        "send_event",
+        "async_send_event",
+    ]
+    receiver_kind: str
 
 
 @dataclass(frozen=True, slots=True)
@@ -53,6 +72,7 @@ class _DecoratorFactory:
 
     selection: _EventSelection
     priority: str
+    receiver_kind: str
 
 
 @dataclass(frozen=True, slots=True)
@@ -63,9 +83,12 @@ class _Registration:
     selection: _EventSelection
     handler: str
     priority: str
+    receiver_kind: str
 
 
-_ScopeValue: TypeAlias = _Symbol | _EventSelection | _DecoratorFactory | None
+_ScopeValue: TypeAlias = (
+    _Symbol | _EventSelection | _DecoratorFactory | _BoundEventMethod | None
+)
 
 
 def _expression_name(node: ast.AST | None) -> str:
@@ -98,15 +121,47 @@ def _priority_identity(node: ast.AST | None) -> str:
     return _DYNAMIC_IDENTITY
 
 
-def _location_sort_key(item: dict[str, Any]) -> tuple[str, int, str]:
-    """返回 consumer location 的稳定排序键。"""
+def fingerprint_event_fact(fact: Mapping[str, object]) -> str:
+    """计算排除诊断行号与已有摘要后的字段敏感 SHA256。"""
+    payload = {
+        key: value
+        for key, value in fact.items()
+        if key not in {"fingerprint", "line"}
+    }
+    encoded = json.dumps(
+        payload,
+        ensure_ascii=True,
+        separators=(",", ":"),
+        sort_keys=True,
+    ).encode("utf-8")
+    return hashlib.sha256(encoded).hexdigest()
+
+
+def _fact_sort_key(item: dict[str, Any]) -> tuple[str, str, int, str]:
+    """返回逐调用 Event fact 的确定排序键。"""
     return (
         str(item["caller"]),
+        str(item["qualname"]),
         int(item["line"]),
-        str(item["identity"]),
+        str(item["fingerprint"]),
     )
 
 
+def _annotation_names(node: ast.AST | None) -> set[str]:
+    """提取类型注解中的有限点分名称。"""
+    if node is None:
+        return set()
+    if isinstance(node, ast.Name):
+        return {node.id}
+    if isinstance(node, ast.Attribute):
+        return {_expression_name(node)}
+    return {
+        name
+        for child in ast.iter_child_nodes(node)
+        for name in _annotation_names(child)
+    }
+
+
 def _bound_names(target: ast.AST) -> set[str]:
     """返回赋值目标在当前 lexical scope 绑定的名称。"""
     if isinstance(target, ast.Name):
@@ -236,8 +291,221 @@ def _function_local_names(
     return local_names - global_names - nonlocal_names
 
 
-class _EventConsumerCollector(ast.NodeVisitor):
-    """以有限 lexical provenance 收集单个宿主模块的事件消费者。"""
+def _event_port_symbol(
+    annotation: ast.AST | None,
+    canonical_aliases: Mapping[str, str] | None = None,
+) -> _Symbol | None:
+    """把明确的 EventPublisher/EventManager 注解转换为 receiver provenance。"""
+    annotation_names = _annotation_names(annotation)
+    leaf_names = {name.rsplit(".", 1)[-1] for name in annotation_names}
+    if any(name.endswith("EventPublisher") for name in leaf_names):
+        return _Symbol("publisher_instance", "injected_event_publisher")
+    if any(name.endswith("EventManagerPort") for name in leaf_names):
+        return _Symbol("manager_instance", "injected_event_manager")
+    for name in annotation_names:
+        head, *tail = name.split(".")
+        canonical = ".".join(
+            ((canonical_aliases or {}).get(head, head), *tail)
+        )
+        if canonical == "app.runtime.events.EventManager":
+            return _Symbol("manager_instance", "injected_event_manager")
+    return None
+
+
+def _function_parameter_symbols(
+    node: ast.FunctionDef | ast.AsyncFunctionDef | ast.Lambda,
+    canonical_aliases: Mapping[str, str] | None = None,
+) -> dict[str, _Symbol]:
+    """收集函数参数上明确声明的 Event 发布端口。"""
+    arguments = (
+        *node.args.posonlyargs,
+        *node.args.args,
+        *node.args.kwonlyargs,
+    )
+    return {
+        argument.arg: symbol
+        for argument in arguments
+        if (
+            symbol := _event_port_symbol(
+                argument.annotation,
+                canonical_aliases,
+            )
+        ) is not None
+    }
+
+
+_InjectedFieldKey: TypeAlias = tuple[str, str]
+
+
+def _module_import_aliases(tree: ast.Module) -> dict[str, str]:
+    """收集解析类继承关系所需的模块级 import 别名。"""
+    aliases: dict[str, str] = {}
+    for node in tree.body:
+        if isinstance(node, ast.Import):
+            for item in node.names:
+                bound = item.asname or item.name.split(".", 1)[0]
+                aliases[bound] = item.name if item.asname else bound
+        elif isinstance(node, ast.ImportFrom) and not node.level and node.module:
+            for item in node.names:
+                if item.name != "*":
+                    aliases[item.asname or item.name] = f"{node.module}.{item.name}"
+    return aliases
+
+
+def _iter_classes(
+    statements: list[ast.stmt],
+    prefix: tuple[str, ...] = (),
+) -> list[tuple[str, ast.ClassDef]]:
+    """按 lexical qualname 返回语句中的类定义，不进入函数 scope。"""
+    classes: list[tuple[str, ast.ClassDef]] = []
+    for statement in statements:
+        if not isinstance(statement, ast.ClassDef):
+            continue
+        qualname = ".".join((*prefix, statement.name))
+        classes.append((qualname, statement))
+        classes.extend(_iter_classes(statement.body, (*prefix, statement.name)))
+    return classes
+
+
+def _canonical_class_name(
+    node: ast.expr,
+    *,
+    module_name: str,
+    aliases: dict[str, str],
+) -> str:
+    """把有限 Name/Attribute 基类表达式还原成 canonical 类名。"""
+    name = _expression_name(node)
+    if not name:
+        return ""
+    head, *tail = name.split(".")
+    if head in aliases:
+        return ".".join((aliases[head], *tail))
+    return f"{module_name}.{name}"
+
+
+def _discover_injected_event_fields(
+    trees: dict[str, ast.Module],
+) -> dict[_InjectedFieldKey, dict[str, _Symbol]]:
+    """按 owning class 发现构造注入字段，并沿已知继承关系传播。"""
+    classes = {
+        f"{module_name}.{qualname}": (module_name, qualname, node)
+        for module_name, tree in trees.items()
+        for qualname, node in _iter_classes(tree.body)
+    }
+    aliases = {
+        module_name: _module_import_aliases(tree)
+        for module_name, tree in trees.items()
+    }
+    fields: dict[str, dict[str, _Symbol]] = {
+        canonical: {} for canonical in classes
+    }
+    bases_by_class: dict[str, set[str]] = {
+        canonical: set() for canonical in classes
+    }
+    descendants_by_class: dict[str, set[str]] = {
+        canonical: set() for canonical in classes
+    }
+    for canonical, (module_name, _qualname, class_node) in classes.items():
+        for base in class_node.bases:
+            base_name = _canonical_class_name(
+                base,
+                module_name=module_name,
+                aliases=aliases[module_name],
+            )
+            if base_name in classes:
+                bases_by_class[canonical].add(base_name)
+                descendants_by_class[base_name].add(canonical)
+
+    for canonical, (module_name, qualname, class_node) in classes.items():
+        class_fields = fields[canonical]
+        for constructor in (
+            node
+            for node in class_node.body
+            if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef))
+            and node.name == "__init__"
+        ):
+            parameters = _function_parameter_symbols(
+                constructor,
+                aliases[module_name],
+            )
+            for candidate in constructor.body:
+                candidates = ast.walk(candidate) if not isinstance(
+                    candidate,
+                    (ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef),
+                ) else ()
+                for assignment in candidates:
+                    if isinstance(assignment, ast.Assign):
+                        pairs = (
+                            (target, assignment.value)
+                            for target in assignment.targets
+                        )
+                    elif isinstance(assignment, ast.AnnAssign) and assignment.value:
+                        pairs = ((assignment.target, assignment.value),)
+                    else:
+                        continue
+                    for target, value in pairs:
+                        if not (
+                            isinstance(target, ast.Attribute)
+                            and isinstance(target.value, ast.Name)
+                            and target.value.id == "self"
+                        ):
+                            continue
+                        symbol = (
+                            parameters.get(value.id)
+                            if isinstance(value, ast.Name)
+                            else None
+                        )
+                        if (
+                            symbol is None
+                            and module_name == "app.chain"
+                            and qualname == "ChainBase"
+                            and target.attr == "eventmanager"
+                            and isinstance(value, ast.Attribute)
+                            and value.attr == "event_manager"
+                        ):
+                            symbol = _Symbol(
+                                "manager_instance",
+                                "injected_event_manager",
+                            )
+                        if symbol is not None:
+                            class_fields[target.attr] = symbol
+
+    direct_fields = {
+        canonical: dict(class_fields)
+        for canonical, class_fields in fields.items()
+        if class_fields
+    }
+    for owner, owner_fields in direct_fields.items():
+        def reachable(
+            starts: set[str],
+            relations: dict[str, set[str]],
+        ) -> set[str]:
+            """返回给定继承方向上的传递闭包。"""
+            pending = list(starts)
+            reached: set[str] = set()
+            while pending:
+                current = pending.pop()
+                if current in reached:
+                    continue
+                reached.add(current)
+                pending.extend(relations[current])
+            return reached
+
+        descendants = reachable({owner}, descendants_by_class)
+        targets = set(descendants)
+        targets.update(reachable(descendants, bases_by_class))
+        for target in targets:
+            fields[target].update(owner_fields)
+
+    return {
+        (module_name, qualname): fields[canonical]
+        for canonical, (module_name, qualname, _node) in classes.items()
+        if fields[canonical]
+    }
+
+
+class _EventFactCollector(ast.NodeVisitor):
+    """以有限 lexical provenance 收集单个宿主模块的 Event 事实。"""
 
     def __init__(
         self,
@@ -246,18 +514,21 @@ class _EventConsumerCollector(ast.NodeVisitor):
         *,
         collect_facts: bool,
         module_final_scope: dict[str, _ScopeValue] | None = None,
+        injected_fields: dict[_InjectedFieldKey, dict[str, _Symbol]] | 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._injected_fields = injected_fields 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]] = []
+        self._class_qualnames: list[str] = []
+        self.producers: list[dict[str, Any]] = []
+        self.consumers: list[dict[str, Any]] = []
 
     def module_scope(self) -> dict[str, _ScopeValue]:
         """返回按模块执行顺序收敛后的符号状态。"""
@@ -304,22 +575,70 @@ class _EventConsumerCollector(ast.NodeVisitor):
         scope_kinds: list[str],
     ) -> dict[str, _ScopeValue]:
         """无事实副作用地计算一组语句执行后的最内层 scope。"""
-        discovery = _EventConsumerCollector(
+        discovery = _EventFactCollector(
             self._module_name,
             self._event_members,
             collect_facts=False,
             module_final_scope=self._module_final_scope,
+            injected_fields=self._injected_fields,
         )
         discovery._scopes = [dict(scope) for scope in scopes]
         discovery._scope_kinds = list(scope_kinds)
+        discovery._qualnames = list(self._qualnames)
+        discovery._class_qualnames = list(self._class_qualnames)
         for statement in statements:
             discovery.visit(statement)
         return dict(discovery._scopes[-1])
 
+    def _function_initial_scope(
+        self,
+        node: ast.FunctionDef | ast.AsyncFunctionDef | ast.Lambda,
+    ) -> dict[str, _ScopeValue]:
+        """构造函数调用期的局部符号，并保留明确的参数端口 provenance。"""
+        scope: dict[str, _ScopeValue] = {
+            name: None for name in _function_local_names(node)
+        }
+        canonical_aliases = {
+            name: "app.runtime.events.EventManager"
+            for name, value in self._module_final_scope.items()
+            if isinstance(value, _Symbol) and value.kind == "manager_class"
+        }
+        if (
+            isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef))
+            and node.name == "__init__"
+            and self._class_qualnames
+        ):
+            scope.update(_function_parameter_symbols(node, canonical_aliases))
+        positional = (*node.args.posonlyargs, *node.args.args)
+        if (
+            self._module_name == "app.runtime.events"
+            and self._qualnames == ["EventManager"]
+            and positional
+            and positional[0].arg == "self"
+        ):
+            scope["self"] = _Symbol("publisher_instance", "event_manager_self")
+        elif (
+            self._class_qualnames
+            and positional
+            and not any(
+                _expression_name(decorator) in {"classmethod", "staticmethod"}
+                for decorator in getattr(node, "decorator_list", ())
+            )
+            and (
+                self._module_name,
+                self._class_qualnames[-1],
+            ) in self._injected_fields
+        ):
+            scope[positional[0].arg] = _Symbol(
+                "injected_owner",
+                self._class_qualnames[-1],
+            )
+        return scope
+
     def _symbol_for_canonical(self, canonical: str) -> _ScopeValue:
         """把有限 canonical 路径转换为 collector symbol。"""
         if canonical == "app.runtime.events.eventmanager":
-            return _Symbol("manager_instance")
+            return _Symbol("manager_instance", "canonical_singleton")
         if canonical == "app.runtime.events.EventManager":
             return _Symbol("manager_class")
         if canonical == "typing.TYPE_CHECKING":
@@ -356,11 +675,29 @@ class _EventConsumerCollector(ast.NodeVisitor):
             return self._lookup(node.id)
         if isinstance(node, ast.Attribute):
             parent = self._resolve(node.value)
+            if isinstance(parent, _Symbol) and parent.kind == "injected_owner":
+                fields = self._injected_fields.get(
+                    (self._module_name, parent.value),
+                    {},
+                )
+                if node.attr in fields:
+                    return fields[node.attr]
             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, _Symbol) and parent.kind in {
+                "manager_instance",
+                "publisher_instance",
+            }:
+                methods = (
+                    _PRODUCER_METHODS | _EVENT_MANAGER_METHODS
+                    if parent.kind == "manager_instance"
+                    else _PRODUCER_METHODS
+                )
+                if node.attr in methods:
+                    return _BoundEventMethod(node.attr, parent.value)
             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, ()):
@@ -368,6 +705,7 @@ class _EventConsumerCollector(ast.NodeVisitor):
                         events=(f"{enum_name}.{node.attr}",),
                         kind="member",
                     )
+                return _EventSelection((), "member", invalid=True)
             return None
         if isinstance(node, ast.List):
             return self._resolve_event_selection(
@@ -375,18 +713,30 @@ class _EventConsumerCollector(ast.NodeVisitor):
                 allow_enum=True,
                 allow_list=True,
             )
+        if isinstance(node, ast.IfExp):
+            return self._resolve_event_selection(
+                node,
+                allow_enum=False,
+                allow_list=False,
+            )
         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")
+                receiver_kind = (
+                    "constructed_manager"
+                    if target.kind == "manager_class"
+                    else "existing_manager"
+                )
+                return _Symbol("manager_instance", receiver_kind)
             registration = self._registration(node)
             if registration and registration.method == "register":
                 return _DecoratorFactory(
                     selection=registration.selection,
                     priority=registration.priority,
+                    receiver_kind=registration.receiver_kind,
                 )
         return None
 
@@ -413,6 +763,7 @@ class _EventConsumerCollector(ast.NodeVisitor):
                 events=tuple(sorted({event for item in selections for event in item.events})),
                 kind="list",
                 dynamic=any(item.dynamic for item in selections),
+                invalid=any(item.invalid for item in selections),
             )
         if isinstance(node, ast.IfExp):
             branches = (
@@ -431,6 +782,7 @@ class _EventConsumerCollector(ast.NodeVisitor):
                 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),
+                invalid=any(item.invalid for item in branches),
             )
         resolved = self._resolve(node)
         if isinstance(resolved, _EventSelection):
@@ -468,14 +820,13 @@ class _EventConsumerCollector(ast.NodeVisitor):
 
     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":
+        bound_method = self._resolve(node.func)
+        if not (
+            isinstance(bound_method, _BoundEventMethod)
+            and bound_method.method in _EVENT_MANAGER_METHODS
+        ):
             return None
+        method = bound_method.method
         if method == "register":
             arguments = self._bind_call_arguments(
                 node,
@@ -504,6 +855,34 @@ class _EventConsumerCollector(ast.NodeVisitor):
             selection=selection,
             handler=_handler_identity(handler_node),
             priority=_priority_identity(arguments.get("priority")),
+            receiver_kind=bound_method.receiver_kind,
+        )
+
+    def _producer_call(
+        self,
+        node: ast.Call,
+    ) -> tuple[_BoundEventMethod, _EventSelection] | None:
+        """解析 receiver 与调用签名均可证明的 Event producer。"""
+        bound_method = self._resolve(node.func)
+        if not (
+            isinstance(bound_method, _BoundEventMethod)
+            and bound_method.method in _PRODUCER_METHODS
+        ):
+            return None
+        arguments = self._bind_call_arguments(
+            node,
+            ("etype", "data", "priority"),
+            frozenset({"etype"}),
+        )
+        if arguments is None:
+            return None
+        return (
+            bound_method,
+            self._resolve_event_selection(
+                arguments["etype"],
+                allow_enum=False,
+                allow_list=False,
+            ),
         )
 
     def _decorator_factory_application(
@@ -523,39 +902,84 @@ class _EventConsumerCollector(ast.NodeVisitor):
             return None
         return factory, arguments["f"]
 
-    def _record(
+    def _base_fact(
         self,
         node: ast.AST,
         selection: _EventSelection,
         *,
+        method: str,
+        receiver_kind: str,
+    ) -> dict[str, Any]:
+        """构造带 line-free fingerprint 的逐调用基础事实。"""
+        fact = {
+            "caller": self._module_name,
+            "line": node.lineno,
+            "qualname": ".".join(self._qualnames) or "<module>",
+            "method": method,
+            "receiver_kind": receiver_kind,
+            "events": list(selection.events),
+            "dynamic": selection.dynamic,
+            "invalid": selection.invalid,
+        }
+        fact["fingerprint"] = fingerprint_event_fact(fact)
+        return fact
+
+    def _record_consumer(
+        self,
+        node: ast.AST,
+        selection: _EventSelection,
+        *,
+        method: str,
+        receiver_kind: str,
         handler: str,
         priority: str,
         registration_kind: Literal["decorator", "listener"],
     ) -> None:
-        """写入静态事件位置，并为未知余项写入一条 dynamic 位置。"""
+        """写入一条 handler、kind 和 priority 完整的 consumer 事实。"""
         if not self._collect_facts:
             return
-        location = {
-            "caller": self._module_name,
-            "line": node.lineno,
+        fact = self._base_fact(
+            node,
+            selection,
+            method=method,
+            receiver_kind=receiver_kind,
+        )
+        fact.update({
             "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))
+        })
+        fact["fingerprint"] = fingerprint_event_fact(fact)
+        self.consumers.append(fact)
+
+    def _record_producer(
+        self,
+        node: ast.Call,
+        method: _BoundEventMethod,
+        selection: _EventSelection,
+    ) -> None:
+        """写入一条 receiver 已证明的 producer 事实。"""
+        if not self._collect_facts:
+            return
+        self.producers.append(
+            self._base_fact(
+                node,
+                selection,
+                method=method.method,
+                receiver_kind=method.receiver_kind,
+            )
+        )
 
     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(
+        self._record_consumer(
             decorator,
             factory.selection,
+            method="register",
+            receiver_kind=factory.receiver_kind,
             handler=handler,
             priority=factory.priority,
             registration_kind="decorator",
@@ -837,7 +1261,7 @@ class _EventConsumerCollector(ast.NodeVisitor):
         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)},
+            self._function_initial_scope(node),
         ]
         body_kinds = [
             "module",
@@ -867,12 +1291,14 @@ class _EventConsumerCollector(ast.NodeVisitor):
         for expression in (*node.bases, *(keyword.value for keyword in node.keywords)):
             self.visit(expression)
         self._qualnames.append(node.name)
+        self._class_qualnames.append(".".join(self._qualnames))
         self._scopes.append({})
         self._scope_kinds.append("class")
         for statement in node.body:
             self.visit(statement)
         self._scope_kinds.pop()
         self._scopes.pop()
+        self._class_qualnames.pop()
         self._qualnames.pop()
         self._set(node.name, None)
 
@@ -890,7 +1316,7 @@ class _EventConsumerCollector(ast.NodeVisitor):
         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)},
+            self._function_initial_scope(node),
         ]
         body_kinds = [
             "module",
@@ -911,73 +1337,86 @@ class _EventConsumerCollector(ast.NodeVisitor):
         self._scope_kinds = saved_kinds
 
     def visit_Call(self, node: ast.Call) -> None:
-        """记录直接 listener 调用或立即应用的 register decorator。"""
+        """记录 receiver 与调用签名均可证明的 producer/consumer。"""
+        producer = self._producer_call(node)
+        if producer is not None:
+            method, selection = producer
+            self._record_producer(node, method, selection)
+
         factory_application = self._decorator_factory_application(node)
         if factory_application:
             factory, handler_node = factory_application
-            self._record(
+            self._record_consumer(
                 node,
                 factory.selection,
+                method="register",
+                receiver_kind=factory.receiver_kind,
                 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",
-            )
+            if registration and registration.method == "add_event_listener":
+                self._record_consumer(
+                    node,
+                    registration.selection,
+                    method="add_event_listener",
+                    receiver_kind=registration.receiver_kind,
+                    handler=registration.handler,
+                    priority=registration.priority,
+                    registration_kind="listener",
+                )
         self.generic_visit(node)
 
 
-def collect_event_consumers(
+def collect_event_facts(
     modules: dict[str, Path],
     event_members: dict[str, tuple[str, ...]],
-) -> tuple[dict[str, list[dict[str, Any]]], list[dict[str, Any]]]:
+) -> dict[str, list[dict[str, Any]]]:
     """
-    收集宿主 EventManager 的静态和动态 consumer 位置
+    收集宿主 EventManager 的逐调用 producer 与 consumer 事实。
 
-    只有可追溯到 ``app.runtime.events.eventmanager`` 或 canonical
-    ``EventManager`` 实例的注册才进入事实；未知 receiver 直接忽略。
+    只有可追溯到 canonical EventManager、其内部 ``self`` 或明确注入事件端口的
+    调用才进入事实。插件模块与未知同名 receiver 始终忽略；每条事实携带排除行号
+    的字段敏感 SHA256，供基线稳定追踪。
 
     :param modules: 宿主模块名到 Python 源码路径的映射
     :param event_members: EventType/ChainEventType 到公开成员名的映射
-    :return: 静态事件位置映射与事件值未知的位置列表
+    :return: ``producers`` 与 ``consumers`` 两组稳定排序的逐调用事实
     """
-    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(
+    trees = {
+        module_name: ast.parse(
+            path.read_text(encoding="utf-8-sig"),
+            filename=str(path),
+        )
+        for module_name, path in sorted(modules.items())
+        if module_name != "app.plugins"
+        and not module_name.startswith("app.plugins.")
+    }
+    injected_fields = _discover_injected_event_fields(trees)
+    producers: list[dict[str, Any]] = []
+    consumers: list[dict[str, Any]] = []
+    for module_name, tree in sorted(trees.items()):
+        discovery = _EventFactCollector(
             module_name,
             event_members,
             collect_facts=False,
+            injected_fields=injected_fields,
         )
         discovery.visit(tree)
-        collector = _EventConsumerCollector(
+        collector = _EventFactCollector(
             module_name,
             event_members,
             collect_facts=True,
             module_final_scope=discovery.module_scope(),
+            injected_fields=injected_fields,
         )
         collector.visit(tree)
-        for event, locations in collector.static.items():
-            static[event].extend(locations)
-        dynamic.extend(collector.dynamic)
+        producers.extend(collector.producers)
+        consumers.extend(collector.consumers)
 
-    return (
-        {
-            event: sorted(locations, key=_location_sort_key)
-            for event, locations in sorted(static.items())
-        },
-        sorted(dynamic, key=_location_sort_key),
-    )
+    return {
+        "producers": sorted(producers, key=_fact_sort_key),
+        "consumers": sorted(consumers, key=_fact_sort_key),
+    }

scripts/architecture/event_policy.py

@@ -0,0 +1,563 @@
+#!/usr/bin/env python3
+"""校验人工审查的宿主 Event consumer 精确政策。"""
+
+import argparse
+import json
+import re
+import sys
+from collections.abc import Mapping, Sequence
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Any
+
+try:
+    from scripts.architecture.event_facts import fingerprint_event_fact
+except ModuleNotFoundError:
+    from event_facts import fingerprint_event_fact
+
+PROJECT_ROOT = Path(__file__).resolve().parents[2]
+DEFAULT_EVENT_POLICY_PATH = (
+    PROJECT_ROOT
+    / "tests"
+    / "fixtures"
+    / "architecture"
+    / "runtime-contract-policy.json"
+)
+EVENT_CONSUMER_POLICY_SCOPE = {
+    "root": "app",
+    "excluded": ["app.plugins"],
+    "receiver_contract": "canonical_event_manager_only",
+}
+EVENT_CONSUMER_FACT_FIELDS = (
+    "caller",
+    "qualname",
+    "method",
+    "receiver_kind",
+    "events",
+    "dynamic",
+    "invalid",
+    "handler",
+    "registration_kind",
+    "priority",
+    "fingerprint",
+)
+EVENT_CONSUMER_POLICY_FIELDS = (
+    *EVENT_CONSUMER_FACT_FIELDS,
+    "classification",
+    "owner",
+    "reason",
+)
+STATIC_CLASSIFICATION = "approved_static_registration"
+DYNAMIC_CLASSIFICATION = "approved_dynamic_exception"
+_FINGERPRINT_PATTERN = re.compile(r"[0-9a-f]{64}")
+_WILDCARD_CHARACTERS = frozenset("*?[")
+_PLACEHOLDER_REASONS = frozenset({"todo", "tbd"})
+
+
+@dataclass(frozen=True, slots=True)
+class EventPolicyViolation:
+    """描述一条可稳定排序和断言的 Event policy 违规。"""
+
+    code: str
+    fingerprint: str | None
+    detail: str
+
+
+@dataclass(frozen=True, slots=True)
+class EventPolicyCheck:
+    """汇总当前 consumer 事实、人工政策和全部校验结果。"""
+
+    actual_count: int
+    reviewed_count: int
+    static_count: int
+    dynamic_count: int
+    invalid_count: int
+    violations: tuple[EventPolicyViolation, ...]
+
+    @property
+    def ok(self) -> bool:
+        """返回当前事实是否完整匹配人工政策。"""
+        return not self.violations
+
+
+def _violation(
+    code: str,
+    detail: str,
+    fingerprint: object = None,
+) -> EventPolicyViolation:
+    """构造字段类型安全的违规对象。"""
+    return EventPolicyViolation(
+        code=code,
+        fingerprint=fingerprint if isinstance(fingerprint, str) else None,
+        detail=detail,
+    )
+
+
+def _sort_violations(
+    violations: list[EventPolicyViolation],
+) -> tuple[EventPolicyViolation, ...]:
+    """按错误码、指纹和说明输出确定顺序。"""
+    return tuple(
+        sorted(
+            violations,
+            key=lambda item: (
+                item.code,
+                item.fingerprint or "",
+                item.detail,
+            ),
+        )
+    )
+
+
+def _has_wildcard(value: object) -> bool:
+    """判断字符串或字符串列表是否包含 glob 通配符。"""
+    if isinstance(value, str):
+        return any(character in value for character in _WILDCARD_CHARACTERS)
+    if isinstance(value, list):
+        return any(_has_wildcard(item) for item in value)
+    return False
+
+
+def _fact_projection(value: Mapping[str, object]) -> dict[str, object]:
+    """提取 policy 与 collector 共用的完整 line-free consumer 字段。"""
+    return {
+        field: value[field]
+        for field in EVENT_CONSUMER_FACT_FIELDS
+        if field != "fingerprint" and field in value
+    }
+
+
+def _validate_fact_shape(
+    value: object,
+    *,
+    source: str,
+) -> list[EventPolicyViolation]:
+    """校验 collector fact 或 policy entry 的公共语义字段。"""
+    if not isinstance(value, Mapping):
+        return [_violation("invalid_entry", f"{source} 必须是对象")]
+
+    violations: list[EventPolicyViolation] = []
+    fingerprint = value.get("fingerprint")
+    missing = sorted(set(EVENT_CONSUMER_FACT_FIELDS) - set(value))
+    if missing:
+        violations.append(
+            _violation(
+                "invalid_entry",
+                f"{source} 缺少字段：{', '.join(missing)}",
+                fingerprint,
+            )
+        )
+        return violations
+
+    string_fields = (
+        "caller",
+        "qualname",
+        "method",
+        "receiver_kind",
+        "handler",
+        "registration_kind",
+        "priority",
+    )
+    if any(
+        not isinstance(value.get(field), str) or not str(value[field]).strip()
+        for field in string_fields
+    ):
+        violations.append(
+            _violation(
+                "invalid_entry",
+                f"{source} 的 consumer identity 字段必须是非空字符串",
+                fingerprint,
+            )
+        )
+
+    events = value.get("events")
+    if (
+        not isinstance(events, list)
+        or any(not isinstance(event, str) or not event for event in events)
+        or events != sorted(set(events))
+    ):
+        violations.append(
+            _violation(
+                "invalid_entry",
+                f"{source}.events 必须是排序、去重的字符串列表",
+                fingerprint,
+            )
+        )
+    if not isinstance(value.get("dynamic"), bool) or not isinstance(
+        value.get("invalid"), bool
+    ):
+        violations.append(
+            _violation(
+                "invalid_entry",
+                f"{source}.dynamic/invalid 必须是布尔值",
+                fingerprint,
+            )
+        )
+    if not isinstance(fingerprint, str) or not _FINGERPRINT_PATTERN.fullmatch(
+        fingerprint
+    ):
+        violations.append(
+            _violation(
+                "invalid_entry",
+                f"{source}.fingerprint 必须是 64 位小写 SHA256",
+                fingerprint,
+            )
+        )
+
+    wildcard_fields = (
+        "caller",
+        "qualname",
+        "method",
+        "receiver_kind",
+        "events",
+        "handler",
+        "registration_kind",
+        "priority",
+    )
+    wildcard_names = [
+        field for field in wildcard_fields if _has_wildcard(value.get(field))
+    ]
+    if wildcard_names:
+        violations.append(
+            _violation(
+                "wildcard_entry",
+                f"{source} 禁止通配字段：{', '.join(wildcard_names)}",
+                fingerprint,
+            )
+        )
+    caller = value.get("caller")
+    if isinstance(caller, str) and (
+        caller == "app.plugins" or caller.startswith("app.plugins.")
+    ):
+        violations.append(
+            _violation(
+                "plugin_scope_violation",
+                f"{source} 不得包含插件副本：{caller}",
+                fingerprint,
+            )
+        )
+    if value.get("invalid") is True:
+        violations.append(
+            _violation(
+                "invalid_consumer",
+                f"{source} 的事件选择包含非法成员，不得获得政策准入",
+                fingerprint,
+            )
+        )
+    return violations
+
+
+def _validate_policy_entry(
+    value: object,
+    *,
+    index: int,
+) -> list[EventPolicyViolation]:
+    """校验人工 entry 的精确字段、分类、owner 和理由。"""
+    source = f"policy.entries[{index}]"
+    violations = _validate_fact_shape(value, source=source)
+    if not isinstance(value, Mapping):
+        return violations
+
+    fingerprint = value.get("fingerprint")
+    if set(value) != set(EVENT_CONSUMER_POLICY_FIELDS):
+        violations.append(
+            _violation(
+                "invalid_entry",
+                f"{source} 必须精确包含规定字段",
+                fingerprint,
+            )
+        )
+    dynamic = value.get("dynamic")
+    expected_classification = (
+        DYNAMIC_CLASSIFICATION if dynamic is True else STATIC_CLASSIFICATION
+    )
+    if value.get("classification") != expected_classification:
+        violations.append(
+            _violation(
+                "classification_mismatch",
+                f"{source}.classification 应为 {expected_classification}",
+                fingerprint,
+            )
+        )
+    owner = value.get("owner")
+    if (
+        not isinstance(owner, str)
+        or not owner.strip()
+        or owner != value.get("caller")
+    ):
+        violations.append(
+            _violation(
+                "owner_mismatch",
+                f"{source}.owner 必须精确等于 caller",
+                fingerprint,
+            )
+        )
+    elif _has_wildcard(owner):
+        violations.append(
+            _violation(
+                "wildcard_entry",
+                f"{source}.owner 禁止通配符",
+                fingerprint,
+            )
+        )
+    reason = value.get("reason")
+    if (
+        not isinstance(reason, str)
+        or not reason.strip()
+        or reason.strip().lower() in _PLACEHOLDER_REASONS
+    ):
+        violations.append(
+            _violation(
+                "empty_reason",
+                f"{source}.reason 必须是非占位的具体理由",
+                fingerprint,
+            )
+        )
+    return violations
+
+
+def _fingerprint_errors(
+    value: Mapping[str, object],
+    *,
+    source: str,
+) -> list[EventPolicyViolation]:
+    """使用统一 Event fact 摘要函数验证 line-free identity。"""
+    fingerprint = value.get("fingerprint")
+    if not isinstance(fingerprint, str):
+        return []
+    expected = fingerprint_event_fact(_fact_projection(value))
+    if fingerprint == expected:
+        return []
+    return [
+        _violation(
+            "fingerprint_mismatch",
+            f"{source}.fingerprint 与完整 consumer identity 不匹配",
+            fingerprint,
+        )
+    ]
+
+
+def _duplicate_fingerprint_errors(
+    values: Sequence[Mapping[str, object]],
+    *,
+    code: str,
+    source: str,
+) -> list[EventPolicyViolation]:
+    """拒绝会被集合比较吞掉的重复 consumer identity。"""
+    fingerprints: list[str] = [
+        fingerprint
+        for value in values
+        if isinstance((fingerprint := value.get("fingerprint")), str)
+    ]
+    duplicates = sorted({item for item in fingerprints if fingerprints.count(item) > 1})
+    return [
+        _violation(code, f"{source} 存在重复 fingerprint", fingerprint)
+        for fingerprint in duplicates
+    ]
+
+
+def validate_event_consumer_policy(
+    policy: Mapping[str, object],
+    facts: Sequence[Mapping[str, object]],
+) -> EventPolicyCheck:
+    """按 exact fingerprint set 校验当前 consumer 事实与人工政策。"""
+    violations: list[EventPolicyViolation] = []
+    actual_facts = list(facts)
+    valid_actual: list[Mapping[str, object]] = []
+    for index, fact in enumerate(actual_facts):
+        fact_errors = _validate_fact_shape(fact, source=f"facts[{index}]")
+        violations.extend(fact_errors)
+        if not any(error.code == "invalid_entry" for error in fact_errors):
+            violations.extend(_fingerprint_errors(fact, source=f"facts[{index}]"))
+            valid_actual.append(fact)
+
+    if not isinstance(policy, Mapping) or set(policy) != {
+        "schema_version",
+        "scope",
+        "event_consumers",
+    }:
+        violations.append(
+            _violation("invalid_schema", "policy 顶层字段不符合 schema v1")
+        )
+    if isinstance(policy, Mapping) and policy.get("schema_version") != 1:
+        violations.append(
+            _violation("invalid_schema", "policy.schema_version 必须为 1")
+        )
+    if not isinstance(policy, Mapping) or policy.get("scope") != EVENT_CONSUMER_POLICY_SCOPE:
+        violations.append(
+            _violation("scope_mismatch", "policy.scope 必须精确匹配宿主扫描范围")
+        )
+
+    policy_section = policy.get("event_consumers") if isinstance(policy, Mapping) else None
+    if not isinstance(policy_section, Mapping) or set(policy_section) != {
+        "match_mode",
+        "entries",
+    }:
+        violations.append(
+            _violation("invalid_schema", "policy.event_consumers 字段不完整")
+        )
+        entries: list[object] = []
+    else:
+        if policy_section.get("match_mode") != "exact_fingerprint_set":
+            violations.append(
+                _violation(
+                    "invalid_schema",
+                    "policy.event_consumers.match_mode 必须为 exact_fingerprint_set",
+                )
+            )
+        raw_entries = policy_section.get("entries")
+        if not isinstance(raw_entries, list):
+            violations.append(
+                _violation("invalid_schema", "policy.event_consumers.entries 必须是列表")
+            )
+            entries = []
+        else:
+            entries = raw_entries
+
+    valid_entries: list[Mapping[str, object]] = []
+    for index, entry in enumerate(entries):
+        entry_errors = _validate_policy_entry(entry, index=index)
+        violations.extend(entry_errors)
+        if isinstance(entry, Mapping) and not any(
+            error.code == "invalid_entry" for error in entry_errors
+        ):
+            violations.extend(
+                _fingerprint_errors(entry, source=f"policy.entries[{index}]")
+            )
+            valid_entries.append(entry)
+
+    violations.extend(
+        _duplicate_fingerprint_errors(
+            valid_actual,
+            code="duplicate_fact_fingerprint",
+            source="collector facts",
+        )
+    )
+    violations.extend(
+        _duplicate_fingerprint_errors(
+            valid_entries,
+            code="duplicate_policy_fingerprint",
+            source="policy entries",
+        )
+    )
+
+    actual_by_fingerprint = {
+        str(fact["fingerprint"]): fact
+        for fact in valid_actual
+        if isinstance(fact.get("fingerprint"), str)
+    }
+    policy_by_fingerprint = {
+        str(entry["fingerprint"]): entry
+        for entry in valid_entries
+        if isinstance(entry.get("fingerprint"), str)
+    }
+    for fingerprint in sorted(actual_by_fingerprint.keys() - policy_by_fingerprint.keys()):
+        violations.append(
+            _violation(
+                "unreviewed_fact",
+                "当前 consumer 事实未经过人工政策审查",
+                fingerprint,
+            )
+        )
+    for fingerprint in sorted(policy_by_fingerprint.keys() - actual_by_fingerprint.keys()):
+        violations.append(
+            _violation(
+                "stale_policy",
+                "人工政策引用的 consumer 事实已经消失或被替换",
+                fingerprint,
+            )
+        )
+    for fingerprint in sorted(actual_by_fingerprint.keys() & policy_by_fingerprint.keys()):
+        if _fact_projection(actual_by_fingerprint[fingerprint]) != _fact_projection(
+            policy_by_fingerprint[fingerprint]
+        ):
+            violations.append(
+                _violation(
+                    "fingerprint_mismatch",
+                    "相同 fingerprint 对应的 consumer identity 不一致",
+                    fingerprint,
+                )
+            )
+
+    static_count = sum(fact.get("dynamic") is False for fact in actual_facts)
+    dynamic_count = sum(fact.get("dynamic") is True for fact in actual_facts)
+    invalid_count = sum(fact.get("invalid") is True for fact in actual_facts)
+    return EventPolicyCheck(
+        actual_count=len(actual_facts),
+        reviewed_count=len(entries),
+        static_count=static_count,
+        dynamic_count=dynamic_count,
+        invalid_count=invalid_count,
+        violations=_sort_violations(violations),
+    )
+
+
+def check_event_consumer_policy(
+    facts: Sequence[Mapping[str, object]],
+    policy_path: Path = DEFAULT_EVENT_POLICY_PATH,
+) -> EventPolicyCheck:
+    """读取人工 policy 文件并校验给定的当前 consumer facts。"""
+    try:
+        policy = json.loads(policy_path.read_text(encoding="utf-8"))
+    except (OSError, json.JSONDecodeError) as error:
+        facts_list = list(facts)
+        return EventPolicyCheck(
+            actual_count=len(facts_list),
+            reviewed_count=0,
+            static_count=sum(fact.get("dynamic") is False for fact in facts_list),
+            dynamic_count=sum(fact.get("dynamic") is True for fact in facts_list),
+            invalid_count=sum(fact.get("invalid") is True for fact in facts_list),
+            violations=(
+                _violation(
+                    "invalid_policy_file",
+                    f"无法读取 Event consumer policy：{error}",
+                ),
+            ),
+        )
+    if not isinstance(policy, Mapping):
+        policy = {}
+    return validate_event_consumer_policy(policy, facts)
+
+
+def parse_args(argv: Sequence[str] | None = None) -> argparse.Namespace:
+    """解析只读 Event policy 检查参数。"""
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument(
+        "--policy",
+        type=Path,
+        default=DEFAULT_EVENT_POLICY_PATH,
+        help="人工 Event consumer policy 路径",
+    )
+    return parser.parse_args(argv)
+
+
+def main(argv: Sequence[str] | None = None) -> int:
+    """收集当前事实并只读检查人工政策，不提供任何写入入口。"""
+    args = parse_args(argv)
+    try:
+        from scripts.architecture.baseline import collect_current_event_facts
+    except ImportError:
+        from baseline import collect_current_event_facts
+
+    current = collect_current_event_facts()
+    consumers: Any = current.get("consumers") if isinstance(current, Mapping) else None
+    if not isinstance(consumers, list):
+        print("当前 Event facts 缺少 consumers 列表", file=sys.stderr)
+        return 1
+    result = check_event_consumer_policy(consumers, args.policy)
+    if result.ok:
+        print(
+            "Event consumer policy 通过："
+            f"{result.actual_count} 条事实，"
+            f"{result.static_count} 条静态注册，"
+            f"{result.dynamic_count} 条动态例外"
+        )
+        return 0
+    for violation in result.violations:
+        suffix = f" [{violation.fingerprint}]" if violation.fingerprint else ""
+        print(f"{violation.code}{suffix}: {violation.detail}", file=sys.stderr)
+    return 1
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

tests/fixtures/architecture/runtime-contract-policy.json

@@ -0,0 +1,319 @@
+{
+  "schema_version": 1,
+  "scope": {
+    "root": "app",
+    "excluded": [
+      "app.plugins"
+    ],
+    "receiver_contract": "canonical_event_manager_only"
+  },
+  "event_consumers": {
+    "match_mode": "exact_fingerprint_set",
+    "entries": [
+      {
+        "caller": "app.chain.download",
+        "qualname": "DownloadChain",
+        "method": "register",
+        "receiver_kind": "canonical_singleton",
+        "events": [
+          "EventType.DownloadFileDeleted"
+        ],
+        "dynamic": false,
+        "invalid": false,
+        "handler": "DownloadChain.download_file_deleted",
+        "registration_kind": "decorator",
+        "priority": "<default>",
+        "fingerprint": "eb33287636f1bb928cb221c64dfb9663fb20b87ad699c41182a673f5af19ab27",
+        "classification": "approved_static_registration",
+        "owner": "app.chain.download",
+        "reason": "下载文件删除事件由 DownloadChain 处理关联状态清理。"
+      },
+      {
+        "caller": "app.chain.scraping",
+        "qualname": "ScrapingChain",
+        "method": "register",
+        "receiver_kind": "canonical_singleton",
+        "events": [
+          "EventType.MetadataScrape"
+        ],
+        "dynamic": false,
+        "invalid": false,
+        "handler": "ScrapingChain.scrape_metadata_event",
+        "registration_kind": "decorator",
+        "priority": "<default>",
+        "fingerprint": "b744f0a05247f5ad9871c180f3e9fd8952d382e94b6e1acad7234796bfea7b4c",
+        "classification": "approved_static_registration",
+        "owner": "app.chain.scraping",
+        "reason": "元数据刮削事件由 ScrapingChain 执行宿主刮削流程。"
+      },
+      {
+        "caller": "app.chain.search",
+        "qualname": "SearchChain",
+        "method": "register",
+        "receiver_kind": "canonical_singleton",
+        "events": [
+          "EventType.SiteDeleted"
+        ],
+        "dynamic": false,
+        "invalid": false,
+        "handler": "SearchChain.remove_site",
+        "registration_kind": "decorator",
+        "priority": "<default>",
+        "fingerprint": "5a7967a84e5716f1939530b1f5d561dbf97130d22330ad7c282d2924e0743fe5",
+        "classification": "approved_static_registration",
+        "owner": "app.chain.search",
+        "reason": "站点删除后由 SearchChain 清理搜索侧站点状态。"
+      },
+      {
+        "caller": "app.chain.site",
+        "qualname": "SiteChain",
+        "method": "register",
+        "receiver_kind": "canonical_singleton",
+        "events": [
+          "EventType.SiteUpdated"
+        ],
+        "dynamic": false,
+        "invalid": false,
+        "handler": "SiteChain.cache_site_icon",
+        "registration_kind": "decorator",
+        "priority": "<default>",
+        "fingerprint": "1200b7097150958ec2003a597237ab746defe80a865b0efef30596dc3b503598",
+        "classification": "approved_static_registration",
+        "owner": "app.chain.site",
+        "reason": "站点更新后由 SiteChain 刷新站点图标缓存。"
+      },
+      {
+        "caller": "app.chain.site",
+        "qualname": "SiteChain",
+        "method": "register",
+        "receiver_kind": "canonical_singleton",
+        "events": [
+          "EventType.SiteUpdated"
+        ],
+        "dynamic": false,
+        "invalid": false,
+        "handler": "SiteChain.clear_site_data",
+        "registration_kind": "decorator",
+        "priority": "<default>",
+        "fingerprint": "6ca5d8f1adb42c94670be1125435387ea8b9f3f20b4505337b2adc090a4b5714",
+        "classification": "approved_static_registration",
+        "owner": "app.chain.site",
+        "reason": "站点更新后由 SiteChain 清理已经失效的站点数据。"
+      },
+      {
+        "caller": "app.chain.site",
+        "qualname": "SiteChain",
+        "method": "register",
+        "receiver_kind": "canonical_singleton",
+        "events": [
+          "EventType.SiteUpdated"
+        ],
+        "dynamic": false,
+        "invalid": false,
+        "handler": "SiteChain.cache_site_userdata",
+        "registration_kind": "decorator",
+        "priority": "<default>",
+        "fingerprint": "c43b969848df947bc8e13d1777a08282cd45386867991786fba7f7ffe5027b47",
+        "classification": "approved_static_registration",
+        "owner": "app.chain.site",
+        "reason": "站点更新后由 SiteChain 刷新站点用户数据缓存。"
+      },
+      {
+        "caller": "app.chain.subscribe",
+        "qualname": "SubscribeChain",
+        "method": "register",
+        "receiver_kind": "canonical_singleton",
+        "events": [
+          "EventType.SiteDeleted"
+        ],
+        "dynamic": false,
+        "invalid": false,
+        "handler": "SubscribeChain.remove_site",
+        "registration_kind": "decorator",
+        "priority": "<default>",
+        "fingerprint": "f29dd0d5dd598e19281e2d1a577b98adc5a0c0722c569bb3e974cc6b984c0737",
+        "classification": "approved_static_registration",
+        "owner": "app.chain.subscribe",
+        "reason": "站点删除后由 SubscribeChain 清理订阅侧站点引用。"
+      },
+      {
+        "caller": "app.chain.subscribe",
+        "qualname": "SubscribeChain",
+        "method": "register",
+        "receiver_kind": "canonical_singleton",
+        "events": [
+          "EventType.ConfigChanged"
+        ],
+        "dynamic": false,
+        "invalid": false,
+        "handler": "SubscribeChain.reconcile_rule_group_references",
+        "registration_kind": "decorator",
+        "priority": "<default>",
+        "fingerprint": "5764fa4e793a602ce36195edf6ce15a16d4875910c0104589b1c5aa9b3173077",
+        "classification": "approved_static_registration",
+        "owner": "app.chain.subscribe",
+        "reason": "配置变更后由 SubscribeChain 对齐订阅规则组引用。"
+      },
+      {
+        "caller": "app.chain.workflow",
+        "qualname": "WorkflowChain",
+        "method": "register",
+        "receiver_kind": "canonical_singleton",
+        "events": [
+          "EventType.WorkflowExecute"
+        ],
+        "dynamic": false,
+        "invalid": false,
+        "handler": "WorkflowChain.event_process",
+        "registration_kind": "decorator",
+        "priority": "<default>",
+        "fingerprint": "50704edda70674af0932e769ddda40d2c21d50b134155c975d675035d7c933cd",
+        "classification": "approved_static_registration",
+        "owner": "app.chain.workflow",
+        "reason": "工作流执行事件由 WorkflowChain 进入标准执行编排。"
+      },
+      {
+        "caller": "app.command",
+        "qualname": "Command",
+        "method": "register",
+        "receiver_kind": "canonical_singleton",
+        "events": [
+          "EventType.CommandExcute"
+        ],
+        "dynamic": false,
+        "invalid": false,
+        "handler": "Command.command_event",
+        "registration_kind": "decorator",
+        "priority": "<default>",
+        "fingerprint": "64006f45841531463b1489576e1aee24ae1b50a2509a9773bc2149415f2c7d18",
+        "classification": "approved_static_registration",
+        "owner": "app.command",
+        "reason": "命令执行事件由 Command 分派到宿主命令处理入口。"
+      },
+      {
+        "caller": "app.command",
+        "qualname": "Command",
+        "method": "register",
+        "receiver_kind": "canonical_singleton",
+        "events": [
+          "EventType.ModuleReload"
+        ],
+        "dynamic": false,
+        "invalid": false,
+        "handler": "Command.module_reload_event",
+        "registration_kind": "decorator",
+        "priority": "<default>",
+        "fingerprint": "89480fbdb61c366e8c77e42fd3bdaa330e634579e7d85cf28e2475b1e3ba7ae6",
+        "classification": "approved_static_registration",
+        "owner": "app.command",
+        "reason": "模块重载事件由 Command 更新命令注册状态。"
+      },
+      {
+        "caller": "app.runtime.extensions.module_manager",
+        "qualname": "ModuleManager.__init__",
+        "method": "add_event_listener",
+        "receiver_kind": "canonical_singleton",
+        "events": [
+          "EventType.ConfigChanged"
+        ],
+        "dynamic": false,
+        "invalid": false,
+        "handler": "self.handle_config_changed",
+        "registration_kind": "listener",
+        "priority": "<default>",
+        "fingerprint": "37daee8e6104140ee28a6efebefd30f1f9ac6a7c44f9609199c1b4863b30c564",
+        "classification": "approved_static_registration",
+        "owner": "app.runtime.extensions.module_manager",
+        "reason": "模块管理器监听配置变化并重建受影响的模块运行状态。"
+      },
+      {
+        "caller": "app.runtime.reload",
+        "qualname": "ConfigReloadMixin.__init_subclass__",
+        "method": "add_event_listener",
+        "receiver_kind": "canonical_singleton",
+        "events": [
+          "EventType.ConfigChanged"
+        ],
+        "dynamic": false,
+        "invalid": false,
+        "handler": "handler",
+        "registration_kind": "listener",
+        "priority": "<default>",
+        "fingerprint": "c53927bd5752ec004ac403f6dc07fca7b253f0db152fc773643972d7c3bf5d34",
+        "classification": "approved_static_registration",
+        "owner": "app.runtime.reload",
+        "reason": "配置重载 Mixin 为声明式宿主类型安装统一刷新处理器。"
+      },
+      {
+        "caller": "app.scheduler",
+        "qualname": "Scheduler",
+        "method": "register",
+        "receiver_kind": "canonical_singleton",
+        "events": [
+          "EventType.PluginReload"
+        ],
+        "dynamic": false,
+        "invalid": false,
+        "handler": "Scheduler.on_plugin_reload",
+        "registration_kind": "decorator",
+        "priority": "<default>",
+        "fingerprint": "055c11260c15adabe65ddf1c9fd39000a7c1b7e255848e3c028e3985731e0778",
+        "classification": "approved_static_registration",
+        "owner": "app.scheduler",
+        "reason": "插件重载后由 Scheduler 对齐插件提供的定时任务。"
+      },
+      {
+        "caller": "app.startup.initializers.agent",
+        "qualname": "AgentInitializer.__init__",
+        "method": "add_event_listener",
+        "receiver_kind": "canonical_singleton",
+        "events": [
+          "EventType.ConfigChanged"
+        ],
+        "dynamic": false,
+        "invalid": false,
+        "handler": "_handle_agent_config_changed",
+        "registration_kind": "listener",
+        "priority": "<default>",
+        "fingerprint": "829bb1b5d09ee66c26f4f500a25e12a768f041a8bcc5dab8954453a186fe0909",
+        "classification": "approved_static_registration",
+        "owner": "app.startup.initializers.agent",
+        "reason": "Agent 初始化器监听配置变化并刷新 Agent 运行配置。"
+      },
+      {
+        "caller": "app.startup.initializers.modules",
+        "qualname": "init_modules",
+        "method": "add_event_listener",
+        "receiver_kind": "constructed_manager",
+        "events": [
+          "EventType.NoticeMessage"
+        ],
+        "dynamic": false,
+        "invalid": false,
+        "handler": "dispatch_web_agent_message_event",
+        "registration_kind": "listener",
+        "priority": "<default>",
+        "fingerprint": "59263eddd85e4e5e30606a9d0ab6d02ba6a9483bff7f262a3818495e2c4f6ef5",
+        "classification": "approved_static_registration",
+        "owner": "app.startup.initializers.modules",
+        "reason": "模块初始化阶段把通知消息桥接到 WebAgent 消息分发入口。"
+      },
+      {
+        "caller": "app.workflow",
+        "qualname": "WorkFlowManager.register_workflow_event",
+        "method": "add_event_listener",
+        "receiver_kind": "canonical_singleton",
+        "events": [],
+        "dynamic": true,
+        "invalid": false,
+        "handler": "self._handle_event",
+        "registration_kind": "listener",
+        "priority": "<default>",
+        "fingerprint": "042068d816db7e46ab4da6e96f8549af97b57bd9d75ba710cc1ff4fec7e5e188",
+        "classification": "approved_dynamic_exception",
+        "owner": "app.workflow",
+        "reason": "工作流配置在运行期决定事件类型，receiver 与 handler 仍可静态证明。"
+      }
+    ]
+  }
+}

tests/test_architecture_baseline_cli.py

@@ -6,6 +6,7 @@ from pathlib import Path
 import pytest
 
 from scripts.architecture import baseline as architecture_baseline
+from scripts.architecture.event_policy import DEFAULT_EVENT_POLICY_PATH
 from scripts.startup import performance as startup_performance
 
 
@@ -153,7 +154,7 @@ def test_architecture_report_only_supports_check_operations(capsys, tmp_path: Pa
 
 
 def test_runtime_semantics_ignore_line_changes_but_keep_call_count(tmp_path: Path):
-    """旧 fixture 的行号变化不影响门禁，重复调用次数仍属于语义。"""
+    """v1 行号变化经链式迁移不影响语义，重复调用次数仍被保留。"""
     baseline_path = tmp_path / "runtime-contract-baseline.json"
     old_value = {
         "schema_version": 1,
@@ -199,11 +200,110 @@ def test_runtime_semantics_ignore_line_changes_but_keep_call_count(tmp_path: Pat
     )
 
     assert old_semantic == moved_semantic
+    assert old_semantic["schema_version"] == 3
+    assert old_semantic["event_facts"]["migration_required"] is True
     assert old_semantic["run_module"]["methods"]["search"] == [
         {"caller": "app.chain.search", "mode": "sync", "count": 2}
     ]
 
 
+def test_runtime_v2_fixture_migrates_to_explicit_v3_review_projection(
+    tmp_path: Path,
+) -> None:
+    """v2 聚合事件只能迁移为待审查投影，不能伪造逐调用事实身份。"""
+    baseline_path = tmp_path / "runtime-contract-baseline.json"
+    value = {
+        "schema_version": 2,
+        "run_module": {"method_count": 0, "call_count": 0},
+        "events": {
+            "event_count": 1,
+            "producer_count": 1,
+            "consumer_count": 0,
+            "events": {},
+            "dynamic_producers": [],
+            "dynamic_consumers": [],
+        },
+        "event_specs": {"EventType.Alpha": {"schema_version": 1}},
+        "module_method_specs": {"search": {"mode": "sync"}},
+        "sdk_exports": {},
+        "compat_manifest": {},
+    }
+
+    migrated = architecture_baseline.semantic_baseline(baseline_path, value)
+
+    assert migrated["schema_version"] == 3
+    assert migrated["scope"]["excluded"] == ["app/plugins"]
+    assert migrated["event_facts"] == {
+        "migration_required": True,
+        "legacy_v2_projection": value["events"],
+    }
+    assert migrated["event_specs"] == value["event_specs"]
+    assert migrated["module_method_specs"] == value["module_method_specs"]
+
+
+def test_runtime_v2_check_and_report_fail_cleanly_until_explicit_refresh(
+    tmp_path: Path,
+) -> None:
+    """旧 v2 fixture 面对逐调用 v3 事实时应正常报告差异而非崩溃。"""
+    baseline_path = tmp_path / "runtime-contract-baseline.json"
+    old_value = {
+        "schema_version": 2,
+        "run_module": {},
+        "events": {
+            "event_count": 0,
+            "producer_count": 0,
+            "consumer_count": 0,
+            "events": {},
+            "dynamic_producers": [],
+            "dynamic_consumers": [],
+        },
+        "sdk_exports": {},
+        "compat_manifest": {},
+    }
+    baseline_path.write_text(json.dumps(old_value), encoding="utf-8")
+    actual = {
+        "schema_version": 3,
+        "scope": {
+            "repository": "MoviePilot",
+            "roots": ["app"],
+            "excluded": ["app/plugins"],
+        },
+        "run_module": {},
+        "module_method_specs": {},
+        "event_facts": {
+            "producer_call_count": 0,
+            "producers": [],
+            "consumers": [],
+        },
+        "event_specs": {},
+        "sdk_exports": {},
+        "compat_manifest": {},
+    }
+
+    assert not architecture_baseline.check_json(
+        baseline_path,
+        actual,
+        write_hint="--write-host",
+    )
+    report = architecture_baseline.build_comparison_report(baseline_path, actual)
+    assert report["semantic_match"] is False
+    assert report["added"] or report["removed"] or report["changed"]
+
+
+def test_comparison_report_preserves_duplicate_fact_multiplicity(tmp_path: Path) -> None:
+    """相同 producer fingerprint 的调用次数变化也必须出现在审查报告。"""
+    baseline_path = tmp_path / "other-baseline.json"
+    baseline_path.write_text('{"facts": [{"id": "same"}]}', encoding="utf-8")
+
+    report = architecture_baseline.build_comparison_report(
+        baseline_path,
+        {"facts": [{"id": "same"}, {"id": "same"}]},
+    )
+
+    assert report["semantic_match"] is False
+    assert report["added"] == [{"path": "$.facts", "value": {"id": "same"}}]
+
+
 def test_plugin_provenance_does_not_participate_in_semantic_gate(tmp_path: Path):
     """插件仓 revision 和源码摘要变化不应伪装成 ABI 变化。"""
     baseline_path = tmp_path / "official-plugin-baseline.json"
@@ -455,6 +555,7 @@ def test_architecture_write_host_only_updates_host_files(
     policy_path = tmp_path / "dependency-policy.json"
     plugin_path = tmp_path / "plugin.json"
     policy_path.write_text('{"manual": true}\n', encoding="utf-8")
+    event_policy_before = DEFAULT_EVENT_POLICY_PATH.read_bytes()
     monkeypatch.setattr(
         architecture_baseline,
         "DEPENDENCY_BASELINE_PATH",
@@ -513,6 +614,7 @@ def test_architecture_write_host_only_updates_host_files(
         "scope": "host-configuration"
     }
     assert json.loads(policy_path.read_text()) == {"manual": True}
+    assert DEFAULT_EVENT_POLICY_PATH.read_bytes() == event_policy_before
     assert not plugin_path.exists()
     output = capsys.readouterr().out
     assert "即将写入" in output
@@ -521,6 +623,7 @@ def test_architecture_write_host_only_updates_host_files(
     assert "transaction.json" in output
     assert "configuration.json" in output
     assert "dependency-policy.json" not in output
+    assert "runtime-contract-policy.json" not in output
 
 
 def test_architecture_write_plugins_only_updates_plugin_file(

tests/test_architecture_ci.py

@@ -27,15 +27,37 @@ def test_unit_test_workflow_has_independent_host_architecture_gate():
     """主仓 PR 与推送必须在全量分片外快速执行宿主架构门禁。"""
     workflow = _load_workflow("test.yml")
     commands = _step_commands(workflow, "architecture")
+    steps = workflow["jobs"]["architecture"]["steps"]
+    semantic_step = next(
+        step for step in steps if step.get("name") == "Check event semantic policy"
+    )
+    snapshot_step = next(
+        step for step in steps if step.get("name") == "Check host architecture snapshot"
+    )
 
     assert "pull_request" in workflow["on"]
     assert "push" in workflow["on"]
     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_event_facts.py" in semantic_step["run"]
+    assert "tests/test_architecture_event_policy.py" in semantic_step["run"]
+    assert "tests/test_architecture_dependencies.py" in semantic_step["run"]
+    assert "tests/test_architecture_adapter_imports.py" in semantic_step["run"]
+    assert "tests/test_architecture_egress.py" in semantic_step["run"]
+    assert "scripts/architecture/event_policy.py" in semantic_step["run"]
+    assert "scripts/architecture/baseline.py" not in semantic_step["run"]
+    assert not {
+        "tests/test_architecture_dependencies.py",
+        "tests/test_architecture_adapter_imports.py",
+        "tests/test_architecture_egress.py",
+        "tests/test_architecture_event_facts.py",
+        "tests/test_architecture_event_policy.py",
+    } & set(snapshot_step["run"].split())
+    assert steps.index(semantic_step) < steps.index(snapshot_step)
     assert "tests/test_architecture_contract_baseline.py" in commands
     assert "scripts/architecture/baseline.py --check-host" in commands
+    assert commands.count("scripts/architecture/baseline.py --check-host") == 1
     assert "scripts/architecture/ruff_ratchet.py" in commands
     assert "scripts/architecture/mypy_ratchet.py" in commands
     assert "scripts/architecture/ruff_ratchet.py --write" not in commands

tests/test_architecture_contract_baseline.py

@@ -11,33 +11,6 @@ PROJECT_ROOT = Path(__file__).parents[1]
 BASELINE_ROOT = PROJECT_ROOT / "tests" / "fixtures" / "architecture"
 
 
-def test_architecture_contract_baselines_match_current_source():
-    """宿主依赖图和公开运行契约变化必须显式刷新基线。"""
-    baseline_paths = (
-        BASELINE_ROOT / "dependency-baseline.json",
-        BASELINE_ROOT / "runtime-contract-baseline.json",
-        BASELINE_ROOT / "transaction-debt-baseline.json",
-        BASELINE_ROOT / "configuration-debt-baseline.json",
-    )
-    contents_before = {
-        path: path.read_bytes()
-        for path in baseline_paths
-    }
-    result = subprocess.run(
-        [sys.executable, "scripts/architecture/baseline.py", "--check-host"],
-        cwd=PROJECT_ROOT,
-        capture_output=True,
-        text=True,
-        check=False,
-    )
-
-    assert result.returncode == 0, result.stderr
-    assert {
-        path: path.read_bytes()
-        for path in baseline_paths
-    } == contents_before
-
-
 def test_official_plugin_baseline_records_external_source():
     """官方插件快照必须绑定独立仓提交，且不得引用宿主插件副本。"""
     baseline_path = BASELINE_ROOT / "official-plugin-baseline.json"
@@ -93,6 +66,69 @@ def test_dependency_baseline_records_nonempty_host_graph() -> None:
     assert all("line" not in entry for entry in direct_egress["entries"])
 
 
+def test_architecture_documents_match_generated_quality_metrics() -> None:
+    """高漂移量化指标必须与生成 fixture 同步，不能在多份文档中分叉。"""
+    dependency = json.loads(
+        (BASELINE_ROOT / "dependency-baseline.json").read_text(encoding="utf-8")
+    )
+    ruff = json.loads(
+        (BASELINE_ROOT / "ruff-baseline.json").read_text(encoding="utf-8")
+    )
+    coverage = json.loads(
+        (BASELINE_ROOT / "coverage-baseline.json").read_text(encoding="utf-8")
+    )
+    runtime = json.loads(
+        (BASELINE_ROOT / "runtime-contract-baseline.json").read_text(
+            encoding="utf-8"
+        )
+    )
+    overview = (PROJECT_ROOT / "docs" / "architecture-overview.md").read_text(
+        encoding="utf-8"
+    )
+    checklist = (
+        PROJECT_ROOT / "docs" / "architecture-optimization-checklist.md"
+    ).read_text(encoding="utf-8")
+    roadmap = (
+        PROJECT_ROOT / "docs" / "architecture-refactor-roadmap.md"
+    ).read_text(encoding="utf-8")
+    edge_count = f"{dependency['edge_count']:,}"
+    ruff_count = sum(
+        count
+        for diagnostics in ruff.values()
+        for count in diagnostics.values()
+    )
+    application_coverage = coverage["application"]["percent"]
+    domain_coverage = coverage["domain"]["percent"]
+    event_facts = runtime["event_facts"]
+
+    assert edge_count in overview
+    assert f"{dependency['module_count']} / {edge_count}" in checklist
+    assert f"Ruff 历史诊断 | {ruff_count}" in checklist
+    assert f"Application {application_coverage:.2f}%" in checklist
+    assert f"Domain {domain_coverage:.2f}%" in checklist
+    assert f"当前受控 {ruff_count} 条诊断归零" in roadmap
+    assert (
+        f"当前宿主有 {event_facts['producer_call_count']} 个\n"
+        f"生产调用，其中 {event_facts['static_producer_call_count']} 个静态解析为 "
+        f"{event_facts['producer_event_reference_count']} 个事件引用"
+    ) in overview
+    assert (
+        f"{event_facts['consumer_registration_count']} 个消费注册中 "
+        f"{event_facts['static_consumer_count']} 个静态、"
+        f"{event_facts['dynamic_consumer_count']} 个动态"
+    ) in overview
+    assert (
+        f"{event_facts['producer_call_count']} 个 producer（"
+        f"{event_facts['static_producer_call_count']} 静态、"
+        f"{event_facts['dynamic_producer_count']} 动态）"
+    ) in checklist
+    assert (
+        f"{event_facts['consumer_registration_count']} 个 consumer（"
+        f"{event_facts['static_consumer_count']} 静态、"
+        f"{event_facts['dynamic_consumer_count']} 动态）"
+    ) in checklist
+
+
 def test_official_discovery_plugins_explicitly_keep_host_page_envelope():
     """宿主探索页消费的官方插件 API 不得依赖动态路由隐式包装。"""
     baseline_path = BASELINE_ROOT / "official-plugin-baseline.json"
@@ -139,7 +175,12 @@ def test_runtime_contract_baseline_excludes_diagnostic_line_numbers():
     baseline_path = BASELINE_ROOT / "runtime-contract-baseline.json"
     baseline = json.loads(baseline_path.read_text(encoding="utf-8"))
 
-    assert baseline["schema_version"] == 2
+    assert baseline["schema_version"] == 3
+    assert baseline["scope"] == {
+        "repository": "MoviePilot",
+        "roots": ["app"],
+        "excluded": ["app/plugins"],
+    }
     assert '"line"' not in json.dumps(baseline)
 
 
@@ -383,24 +424,30 @@ for package_name, symbol_name in contracts:
 
 
 def test_event_contract_baseline_covers_every_public_event_enum() -> None:
-    """事件生产者/消费者快照必须覆盖全部广播和链式事件枚举。"""
+    """统一事件事实快照必须覆盖全部枚举和真实生产、消费调用。"""
     baseline_path = BASELINE_ROOT / "runtime-contract-baseline.json"
     baseline = json.loads(baseline_path.read_text(encoding="utf-8"))
-    events = baseline["events"]
+    events = baseline["event_facts"]
     expected = {
         *(f"EventType.{member.name}" for member in EventType),
         *(f"ChainEventType.{member.name}" for member in ChainEventType),
     }
 
-    assert set(events["events"]) == expected
+    assert set(events["event_index"]) == expected
     assert events["event_count"] == len(expected)
-    assert events["producer_count"] > 0
-    assert events["consumer_count"] == 16
-    assert events["dynamic_consumers"] == [
-        {"caller": "app.workflow", "count": 1}
-    ]
+    assert events["producer_call_count"] == 99
+    assert events["static_producer_call_count"] == 98
+    assert events["dynamic_producer_count"] == 1
+    assert events["invalid_producer_count"] == 0
+    assert events["producer_event_reference_count"] == 100
+    assert events["consumer_registration_count"] == 17
+    assert events["static_consumer_count"] == 16
+    assert events["dynamic_consumer_count"] == 1
+    assert events["invalid_consumer_count"] == 0
+    assert events["consumer_event_reference_count"] == 16
+    assert events["fact_count"] == 116
+    assert len({fact["fingerprint"] for fact in events["consumers"]}) == 17
     assert all(
-        not item["caller"].startswith("app.plugins")
-        for contract in events["events"].values()
-        for item in contract["consumers"]
+        not fact["caller"].startswith("app.plugins")
+        for fact in (*events["producers"], *events["consumers"])
     )

tests/test_architecture_dependencies.py

@@ -4,7 +4,7 @@ from functools import lru_cache
 from pathlib import Path
 
 from scripts.architecture.baseline import (
-    collect_event_contracts as _collect_event_contracts,
+    collect_current_event_facts as _collect_current_event_facts,
 )
 from scripts.architecture.baseline import (
     discover_modules as _discover_modules,
@@ -1288,13 +1288,8 @@ def test_application_services_do_not_resolve_event_manager_singleton():
 
 def test_http_endpoints_do_not_register_process_event_listeners():
     """HTTP 端点不得拥有进程级事件监听器，监听装配必须留在 startup。"""
-    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"])
+    facts = _collect_current_event_facts()
+    callers = {item["caller"] for item in facts["consumers"]}
 
     assert {caller for caller in callers if caller.startswith("app.api")} == set()
 

tests/test_architecture_event_facts.py

@@ -1,8 +1,12 @@
-"""Event consumer AST collector tests."""
+"""Unified Event producer and consumer AST collector tests."""
 
+from collections import defaultdict
 from pathlib import Path
 
-from scripts.architecture.event_consumers import collect_event_consumers
+from scripts.architecture.event_facts import (
+    collect_event_facts,
+    fingerprint_event_fact,
+)
 
 EVENT_MEMBERS = {
     "EventType": ("Alpha", "Beta", "Gamma"),
@@ -10,6 +14,18 @@ EVENT_MEMBERS = {
 }
 
 
+def _collect_facts(
+    tmp_path: Path,
+    source: str,
+    *,
+    module_name: str = "app.sample",
+) -> dict[str, list[dict]]:
+    """从单个临时宿主模块收集统一 Event 事实。"""
+    path = tmp_path / f"{module_name.replace('.', '_')}.py"
+    path.write_text(source, encoding="utf-8")
+    return collect_event_facts({module_name: path}, EVENT_MEMBERS)
+
+
 def _collect(
     tmp_path: Path,
     source: str,
@@ -19,7 +35,29 @@ def _collect(
     """从单个临时宿主模块收集 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)
+    facts = collect_event_facts({module_name: path}, EVENT_MEMBERS)
+    static: dict[str, list[dict]] = defaultdict(list)
+    dynamic: list[dict] = []
+    for fact in facts["consumers"]:
+        location = {
+            "caller": fact["caller"],
+            "line": fact["line"],
+            "handler": fact["handler"],
+            "identity": "|".join(
+                (
+                    fact["registration_kind"],
+                    fact["handler"],
+                    fact["priority"],
+                )
+            ),
+            "priority": fact["priority"],
+            "registration_kind": fact["registration_kind"],
+        }
+        for event in fact["events"]:
+            static[event].append(dict(location))
+        if fact["dynamic"]:
+            dynamic.append(dict(location))
+    return dict(static), dynamic
 
 
 def _stable_locations(locations: list[dict]) -> list[dict]:
@@ -627,7 +665,273 @@ def handler(event):
     host_path.write_text(source, encoding="utf-8")
     plugin_path.write_text(source.replace("Alpha", "Beta"), encoding="utf-8")
 
-    static, dynamic = collect_event_consumers(
+    facts = collect_event_facts(
+        {
+            "app.host": host_path,
+            "app.plugins.sample": plugin_path,
+        },
+        EVENT_MEMBERS,
+    )
+    static: dict[str, list[dict]] = defaultdict(list)
+    dynamic: list[dict] = []
+    for fact in facts["consumers"]:
+        for event in fact["events"]:
+            static[event].append(fact)
+        if fact["dynamic"]:
+            dynamic.append(fact)
+
+    assert dynamic == []
+    assert set(static) == {"EventType.Alpha"}
+
+
+def test_collect_event_facts_resolves_producer_receivers_and_bound_aliases(
+    tmp_path: Path,
+) -> None:
+    """Producer 仅接受 canonical receiver，并保留方法与 qualname。"""
+    facts = _collect_facts(
+        tmp_path,
+        '''
+from app.runtime.events import EventManager, eventmanager
+from app.schemas.types import ChainEventType, EventType
+import app.runtime.events as runtime_events
+import app.schemas.types as schema_types
+
+receiver = eventmanager
+emit = receiver.send_event
+emit(EventType.Alpha)
+EventManager().send_event(EventType.Gamma)
+EventManager.get_existing_instance().send_event(ChainEventType.Delta)
+
+async def publish():
+    await runtime_events.eventmanager.async_send_event(
+        etype=schema_types.EventType.Beta,
+    )
+''',
+    )
+
+    assert facts["consumers"] == []
+    assert len(facts["producers"]) == 4
+    assert {
+        (
+            fact["qualname"],
+            fact["method"],
+            fact["receiver_kind"],
+            tuple(fact["events"]),
+        )
+        for fact in facts["producers"]
+    } == {
+        ("<module>", "send_event", "canonical_singleton", ("EventType.Alpha",)),
+        ("<module>", "send_event", "constructed_manager", ("EventType.Gamma",)),
+        ("<module>", "send_event", "existing_manager", ("ChainEventType.Delta",)),
+        ("publish", "async_send_event", "canonical_singleton", ("EventType.Beta",)),
+    }
+    producer_keys = {
+        "caller",
+        "line",
+        "qualname",
+        "method",
+        "receiver_kind",
+        "events",
+        "dynamic",
+        "invalid",
+        "fingerprint",
+    }
+    assert all(set(fact) == producer_keys for fact in facts["producers"])
+    assert all(
+        fact["fingerprint"] == fingerprint_event_fact(fact)
+        for fact in facts["producers"]
+    )
+
+
+def test_collect_event_facts_scopes_injected_ports_to_owning_classes(
+    tmp_path: Path,
+) -> None:
+    """构造注入端口可沿继承使用，但不得证明无关类的同名字段。"""
+    facts = _collect_facts(
+        tmp_path,
+        '''
+from typing import Protocol
+from app.runtime.events import EventManager as CanonicalEventManager
+from app.schemas.types import EventType
+
+class SampleEventPublisher(Protocol):
+    def send_event(self, etype, data=None, priority=10): ...
+
+class PublisherMixin:
+    def emit(self):
+        receiver = self.publisher
+        bound = receiver.send_event
+        bound(EventType.Alpha)
+
+class GoodPublisher(PublisherMixin):
+    def __init__(self, publisher: SampleEventPublisher):
+        self.publisher = publisher
+
+class SiblingPublisher(PublisherMixin):
+    def emit_sibling(self):
+        self.publisher.send_event(EventType.Beta)
+
+class BadPublisher:
+    def emit(self):
+        self.publisher.send_event(EventType.Beta)
+
+def free_function(publisher: SampleEventPublisher):
+    publisher.send_event(EventType.Beta)
+
+class GoodManager:
+    def __init__(self, manager: CanonicalEventManager):
+        self.manager = manager
+
+    def emit(self):
+        self.manager.async_send_event(EventType.Gamma)
+
+class EventManager:
+    pass
+
+class FakeManagerOwner:
+    def __init__(self, manager: EventManager):
+        self.manager = manager
+
+    def emit(self):
+        self.manager.send_event(EventType.Beta)
+''',
+    )
+
+    assert facts["consumers"] == []
+    assert [fact["events"] for fact in facts["producers"]] == [
+        ["EventType.Gamma"],
+        ["EventType.Alpha"],
+    ]
+    assert {fact["receiver_kind"] for fact in facts["producers"]} == {
+        "injected_event_manager",
+        "injected_event_publisher",
+    }
+
+
+def test_collect_event_facts_separates_static_dynamic_and_invalid_producers(
+    tmp_path: Path,
+) -> None:
+    """有限 IfExp 展开，运行期值与伪 enum 成员分别标记。"""
+    facts = _collect_facts(
+        tmp_path,
+        '''
+from app.runtime.events import eventmanager
+from app.schemas.types import EventType
+
+selected = EventType.Alpha if enabled else EventType.Beta
+eventmanager.send_event(selected)
+eventmanager.send_event(EventType(name))
+eventmanager.send_event(EventType.Missing)
+unknown.send_event(EventType.Gamma)
+
+eventmanager.send_event()
+eventmanager.send_event(event_type=EventType.Alpha)
+eventmanager.send_event(EventType.Alpha, etype=EventType.Beta)
+eventmanager.send_event(EventType.Alpha, unknown=True)
+''',
+    )
+
+    producers = facts["producers"]
+    assert len(producers) == 3
+    assert [
+        (fact["events"], fact["dynamic"], fact["invalid"])
+        for fact in producers
+    ] == [
+        (["EventType.Alpha", "EventType.Beta"], False, False),
+        ([], True, False),
+        ([], False, True),
+    ]
+
+
+def test_collect_event_facts_accepts_only_event_manager_internal_producers(
+    tmp_path: Path,
+) -> None:
+    """EventManager 内部 self 可生产事件，但注册委托不重复计 consumer。"""
+    facts = _collect_facts(
+        tmp_path,
+        '''
+from app.schemas.types import EventType
+
+class EventManager:
+    def __init__(self):
+        self.send_event(EventType.Alpha)
+
+    def register(self, handler):
+        self.add_event_listener(EventType.Beta, handler)
+''',
+        module_name="app.runtime.events",
+    )
+
+    assert facts["consumers"] == []
+    assert len(facts["producers"]) == 1
+    assert facts["producers"][0]["receiver_kind"] == "event_manager_self"
+    assert facts["producers"][0]["events"] == ["EventType.Alpha"]
+
+
+def test_collect_event_facts_consumer_schema_and_fingerprint_are_stable(
+    tmp_path: Path,
+) -> None:
+    """Consumer 使用完整字段名，摘要排除 line 且覆盖全部治理字段。"""
+    source = '''
+from app.runtime.events import eventmanager
+from app.schemas.types import EventType
+
+@eventmanager.register(EventType.Alpha, priority=3)
+def handler(event):
+    pass
+'''
+    fact = _collect_facts(tmp_path, source)["consumers"][0]
+    assert set(fact) == {
+        "caller",
+        "line",
+        "qualname",
+        "method",
+        "receiver_kind",
+        "events",
+        "dynamic",
+        "invalid",
+        "fingerprint",
+        "handler",
+        "priority",
+        "registration_kind",
+    }
+    assert fact["registration_kind"] == "decorator"
+    moved = dict(fact, line=fact["line"] + 100, fingerprint="stale")
+    assert fingerprint_event_fact(moved) == fact["fingerprint"]
+    for key, value in {
+        "caller": "app.changed",
+        "qualname": "Changed.handler",
+        "method": "add_event_listener",
+        "receiver_kind": "constructed_manager",
+        "events": ["EventType.Beta"],
+        "dynamic": True,
+        "invalid": True,
+        "handler": "other",
+        "priority": "4",
+        "registration_kind": "listener",
+    }.items():
+        changed = dict(fact, **{key: value})
+        assert fingerprint_event_fact(changed) != fact["fingerprint"]
+
+
+def test_collect_event_facts_excludes_all_plugin_facts(tmp_path: Path) -> None:
+    """误传插件路径时 producer 与 consumer 都不得进入宿主事实。"""
+    host_path = tmp_path / "host_facts.py"
+    plugin_path = tmp_path / "plugin_facts.py"
+    source = '''
+from app.runtime.events import eventmanager
+from app.schemas.types import EventType
+
+eventmanager.send_event(EventType.Alpha)
+
+@eventmanager.register(EventType.Beta)
+def handler(event):
+    pass
+'''
+    host_path.write_text(source, encoding="utf-8")
+    plugin_path.write_text(source.replace("Alpha", "Gamma"), encoding="utf-8")
+
+    facts = collect_event_facts(
         {
             "app.host": host_path,
             "app.plugins.sample": plugin_path,
@@ -635,5 +939,38 @@ def handler(event):
         EVENT_MEMBERS,
     )
 
-    assert dynamic == []
-    assert set(static) == {"EventType.Alpha"}
+    assert len(facts["producers"]) == 1
+    assert facts["producers"][0]["caller"] == "app.host"
+    assert len(facts["consumers"]) == 1
+    assert facts["consumers"][0]["caller"] == "app.host"
+
+
+def test_collect_event_facts_matches_current_host_inventory() -> None:
+    """统一事实覆盖当前 99 个 producer 调用与 17 个 consumer 调用。"""
+    from scripts.architecture.baseline import (
+        _event_enum_members,
+        discover_modules,
+    )
+
+    facts = collect_event_facts(
+        discover_modules(),
+        {
+            enum_name: _event_enum_members(enum_name)
+            for enum_name in ("EventType", "ChainEventType")
+        },
+    )
+    producers = facts["producers"]
+    consumers = facts["consumers"]
+
+    assert len(producers) == 99
+    assert sum(not fact["dynamic"] and not fact["invalid"] for fact in producers) == 98
+    assert sum(fact["dynamic"] for fact in producers) == 1
+    assert sum(fact["invalid"] for fact in producers) == 0
+    assert sum(len(fact["events"]) for fact in producers) == 100
+    assert len(consumers) == 17
+    assert sum(not fact["dynamic"] and not fact["invalid"] for fact in consumers) == 16
+    assert sum(fact["dynamic"] for fact in consumers) == 1
+    assert sum(fact["invalid"] for fact in consumers) == 0
+    assert sum(len(fact["events"]) for fact in consumers) == 16
+    consumer_fingerprints = [fact["fingerprint"] for fact in consumers]
+    assert len(consumer_fingerprints) == len(set(consumer_fingerprints))

tests/test_architecture_event_policy.py

@@ -0,0 +1,323 @@
+"""宿主 Event consumer 人工政策测试。"""
+
+import json
+from copy import deepcopy
+from pathlib import Path
+from typing import Any, Callable
+
+import pytest
+
+from scripts.architecture.baseline import collect_current_event_facts
+from scripts.architecture.event_facts import fingerprint_event_fact
+from scripts.architecture.event_policy import (
+    DEFAULT_EVENT_POLICY_PATH,
+    DYNAMIC_CLASSIFICATION,
+    EVENT_CONSUMER_POLICY_SCOPE,
+    STATIC_CLASSIFICATION,
+    EventPolicyCheck,
+    check_event_consumer_policy,
+    parse_args,
+    validate_event_consumer_policy,
+)
+
+
+def _fact(
+    *,
+    caller: str = "app.sample",
+    events: list[str] | None = None,
+    dynamic: bool = False,
+    invalid: bool = False,
+    handler: str = "Sample.handle",
+    line: int = 10,
+) -> dict[str, Any]:
+    """构造一条与统一 collector 相同字段的 consumer fact。"""
+    fact: dict[str, Any] = {
+        "caller": caller,
+        "line": line,
+        "qualname": "Sample",
+        "method": "register",
+        "receiver_kind": "canonical_instance",
+        "events": events if events is not None else ["EventType.Alpha"],
+        "dynamic": dynamic,
+        "invalid": invalid,
+        "handler": handler,
+        "registration_kind": "decorator",
+        "priority": "<default>",
+    }
+    fact["fingerprint"] = fingerprint_event_fact(fact)
+    return fact
+
+
+def _entry(
+    fact: dict[str, Any],
+    *,
+    classification: str | None = None,
+    owner: str | None = None,
+    reason: str = "宿主显式注册并拥有该事件处理器。",
+) -> dict[str, Any]:
+    """把 collector fact 转换为完整 line-free policy entry。"""
+    entry = {key: value for key, value in fact.items() if key != "line"}
+    entry.update({
+        "classification": classification
+        or (
+            DYNAMIC_CLASSIFICATION
+            if fact["dynamic"] is True
+            else STATIC_CLASSIFICATION
+        ),
+        "owner": owner or str(fact["caller"]),
+        "reason": reason,
+    })
+    return entry
+
+
+def _policy(entries: list[dict[str, Any]]) -> dict[str, Any]:
+    """构造 exact fingerprint set policy。"""
+    return {
+        "schema_version": 1,
+        "scope": dict(EVENT_CONSUMER_POLICY_SCOPE),
+        "event_consumers": {
+            "match_mode": "exact_fingerprint_set",
+            "entries": entries,
+        },
+    }
+
+
+def _codes(result: EventPolicyCheck) -> list[str]:
+    """返回校验结果中的稳定错误码列表。"""
+    return [violation.code for violation in result.violations]
+
+
+def test_event_policy_accepts_exact_static_and_dynamic_facts() -> None:
+    """静态注册和明确动态例外应按完整 fingerprint 精确匹配。"""
+    static = _fact()
+    dynamic = _fact(
+        caller="app.workflow",
+        events=[],
+        dynamic=True,
+        handler="self._handle_event",
+        line=20,
+    )
+
+    result = validate_event_consumer_policy(
+        _policy([_entry(static), _entry(dynamic)]),
+        [static, dynamic],
+    )
+
+    assert result.ok
+    assert result.actual_count == result.reviewed_count == 2
+    assert result.static_count == 1
+    assert result.dynamic_count == 1
+    assert result.invalid_count == 0
+
+
+def test_current_event_consumer_policy_matches_exact_reviewed_set() -> None:
+    """当前 17 条宿主 consumer 必须逐条匹配人工政策且没有非法成员。"""
+    facts = collect_current_event_facts()["consumers"]
+    policy = json.loads(DEFAULT_EVENT_POLICY_PATH.read_text(encoding="utf-8"))
+    entries = policy["event_consumers"]["entries"]
+
+    result = check_event_consumer_policy(facts)
+
+    assert result.ok
+    assert result.actual_count == result.reviewed_count == 17
+    assert result.static_count == 16
+    assert result.dynamic_count == 1
+    assert result.invalid_count == 0
+    assert len({fact["fingerprint"] for fact in facts}) == 17
+    assert {entry["fingerprint"] for entry in entries} == {
+        fact["fingerprint"] for fact in facts
+    }
+    assert sum(
+        entry["classification"] == STATIC_CLASSIFICATION for entry in entries
+    ) == 16
+    dynamic_entries = [
+        entry
+        for entry in entries
+        if entry["classification"] == DYNAMIC_CLASSIFICATION
+    ]
+    assert dynamic_entries == [{
+        "caller": "app.workflow",
+        "qualname": "WorkFlowManager.register_workflow_event",
+        "method": "add_event_listener",
+        "receiver_kind": "canonical_singleton",
+        "events": [],
+        "dynamic": True,
+        "invalid": False,
+        "handler": "self._handle_event",
+        "registration_kind": "listener",
+        "priority": "<default>",
+        "fingerprint": (
+            "042068d816db7e46ab4da6e96f8549af97b57bd9d75ba710cc1ff4fec7e5e188"
+        ),
+        "classification": DYNAMIC_CLASSIFICATION,
+        "owner": "app.workflow",
+        "reason": "工作流配置在运行期决定事件类型，receiver 与 handler 仍可静态证明。",
+    }]
+    assert all("line" not in entry for entry in entries)
+
+
+def test_event_policy_reuses_line_free_event_fact_fingerprint() -> None:
+    """源码行移动不改变摘要，任一 line-free 语义字段变化都改变摘要。"""
+    fact = _fact()
+    moved = {**fact, "line": 999}
+    assert fingerprint_event_fact(fact) == fingerprint_event_fact(moved)
+
+    for field, value in {
+        "caller": "app.other",
+        "qualname": "Other",
+        "method": "add_event_listener",
+        "receiver_kind": "constructed_instance",
+        "events": ["EventType.Beta"],
+        "dynamic": True,
+        "invalid": True,
+        "handler": "Other.handle",
+        "registration_kind": "listener",
+        "priority": "10",
+    }.items():
+        changed = {**fact, field: value}
+        assert fingerprint_event_fact(changed) != fact["fingerprint"], field
+
+
+def test_event_policy_rejects_add_remove_and_same_count_replacement() -> None:
+    """新增、陈旧和同计数替换都不能通过刷新事实集合隐藏。"""
+    original = _fact()
+    added = _fact(handler="Sample.added", line=20)
+    policy = _policy([_entry(original)])
+
+    assert _codes(validate_event_consumer_policy(policy, [original])) == []
+    assert "unreviewed_fact" in _codes(
+        validate_event_consumer_policy(policy, [original, added])
+    )
+    assert _codes(validate_event_consumer_policy(policy, [])) == ["stale_policy"]
+    assert _codes(validate_event_consumer_policy(policy, [added])) == [
+        "stale_policy",
+        "unreviewed_fact",
+    ]
+
+
+def test_event_policy_rejects_classification_and_owner_swaps() -> None:
+    """事实仍相同时也不得互换静态/动态分类或责任 owner。"""
+    fact = _fact()
+    classification_swap = _entry(fact, classification=DYNAMIC_CLASSIFICATION)
+    owner_swap = _entry(fact, owner="app.other")
+
+    assert "classification_mismatch" in _codes(
+        validate_event_consumer_policy(_policy([classification_swap]), [fact])
+    )
+    assert "owner_mismatch" in _codes(
+        validate_event_consumer_policy(_policy([owner_swap]), [fact])
+    )
+
+
+def test_event_policy_rejects_invalid_schema_wildcards_and_reasons() -> None:
+    """人工 policy 必须保持精确 scope、字段和可审查理由。"""
+    def schema_drift(policy: dict[str, Any]) -> None:
+        policy["schema_version"] = 2
+
+    def scope_drift(policy: dict[str, Any]) -> None:
+        policy["scope"]["root"] = "app.chain"
+
+    def wildcard_handler(policy: dict[str, Any]) -> None:
+        policy["event_consumers"]["entries"][0]["handler"] = "Sample.*"
+
+    def empty_reason(policy: dict[str, Any]) -> None:
+        policy["event_consumers"]["entries"][0]["reason"] = " "
+
+    def placeholder_reason(policy: dict[str, Any]) -> None:
+        policy["event_consumers"]["entries"][0]["reason"] = "TODO"
+
+    mutations: tuple[tuple[Callable[[dict[str, Any]], None], str], ...] = (
+        (schema_drift, "invalid_schema"),
+        (scope_drift, "scope_mismatch"),
+        (wildcard_handler, "wildcard_entry"),
+        (empty_reason, "empty_reason"),
+        (placeholder_reason, "empty_reason"),
+    )
+    fact = _fact()
+    for mutate, expected_code in mutations:
+        policy = _policy([_entry(fact)])
+        mutate(policy)
+        assert expected_code in _codes(
+            validate_event_consumer_policy(policy, [fact])
+        )
+
+
+def test_event_policy_rejects_duplicate_policy_and_fact_fingerprints() -> None:
+    """重复 identity 不能被 exact set 的字典或集合投影吞掉。"""
+    fact = _fact()
+    entry = _entry(fact)
+
+    policy_result = validate_event_consumer_policy(
+        _policy([entry, deepcopy(entry)]),
+        [fact],
+    )
+    fact_result = validate_event_consumer_policy(
+        _policy([entry]),
+        [fact, deepcopy(fact)],
+    )
+
+    assert "duplicate_policy_fingerprint" in _codes(policy_result)
+    assert "duplicate_fact_fingerprint" in _codes(fact_result)
+
+
+def test_event_policy_rejects_invalid_and_plugin_consumers() -> None:
+    """非法事件成员与 app.plugins 副本均不得写入人工准入。"""
+    invalid = _fact(invalid=True)
+    plugin = _fact(caller="app.plugins.demo")
+
+    invalid_result = validate_event_consumer_policy(
+        _policy([_entry(invalid)]),
+        [invalid],
+    )
+    plugin_result = validate_event_consumer_policy(
+        _policy([_entry(plugin)]),
+        [plugin],
+    )
+
+    assert "invalid_consumer" in _codes(invalid_result)
+    assert "plugin_scope_violation" in _codes(plugin_result)
+
+
+def test_event_policy_rejects_tampered_fingerprint() -> None:
+    """policy 不得保留旧摘要同时篡改 receiver、method 或其他事实字段。"""
+    fact = _fact()
+    entry = _entry(fact)
+    entry["receiver_kind"] = "constructed_instance"
+
+    result = validate_event_consumer_policy(_policy([entry]), [fact])
+
+    assert "fingerprint_mismatch" in _codes(result)
+
+
+def test_event_policy_file_check_reports_invalid_json(tmp_path: Path) -> None:
+    """只读文件入口把缺失或损坏 policy 报告为结构化失败。"""
+    path = tmp_path / "runtime-contract-policy.json"
+    path.write_text("{", encoding="utf-8")
+
+    result = check_event_consumer_policy([_fact()], path)
+
+    assert not result.ok
+    assert _codes(result) == ["invalid_policy_file"]
+
+
+def test_event_policy_cli_has_no_write_mode() -> None:
+    """人工 policy CLI 永远不暴露自动写入入口。"""
+    with pytest.raises(SystemExit) as error:
+        parse_args(["--write"])
+
+    assert error.value.code == 2
+
+
+def test_event_policy_violation_order_is_deterministic() -> None:
+    """CI 输出必须按错误码和 fingerprint 稳定排序。"""
+    first = _fact(handler="Sample.first")
+    second = _fact(handler="Sample.second")
+
+    result = validate_event_consumer_policy(_policy([]), [second, first])
+
+    assert result.violations == tuple(
+        sorted(
+            result.violations,
+            key=lambda item: (item.code, item.fingerprint or "", item.detail),
+        )
+    )