ci: share architecture scans and stop rerunning gate tests in coverage shards

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

.github/workflows/test.yml

@@ -125,22 +125,18 @@ jobs:
       fail-fast: false
       matrix:
         include:
-          - shard: '1/8'
+          - shard: '1/6'
             index: '1'
-          - shard: '2/8'
+          - shard: '2/6'
             index: '2'
-          - shard: '3/8'
+          - shard: '3/6'
             index: '3'
-          - shard: '4/8'
+          - shard: '4/6'
             index: '4'
-          - shard: '5/8'
+          - shard: '5/6'
             index: '5'
-          - shard: '6/8'
+          - shard: '6/6'
             index: '6'
-          - shard: '7/8'
-            index: '7'
-          - shard: '8/8'
-            index: '8'
 
     steps:
       - name: Checkout code
@@ -160,7 +156,7 @@ jobs:
 
       - name: Generate coverage data
         timeout-minutes: 10
-        run: uv run --locked --no-sync python -m coverage run --parallel-mode tests/run.py --shard "${{ matrix.shard }}"
+        run: uv run --locked --no-sync python -m coverage run --parallel-mode tests/run.py --exclude-architecture-gate --shard "${{ matrix.shard }}"
 
       - name: Upload coverage data
         uses: actions/upload-artifact@v7

docs/testing.md

@@ -15,13 +15,16 @@ uv run --locked --no-sync python tests/run.py --serial              # 串行全
 uv run --locked --no-sync python tests/run.py --shard 1/4           # 只跑指定分片，供 CI 复用
 ```
 
-`tests/run.py` 的 runner 参数只有 `--serial` 和 `--shard N/TOTAL`；其余参数保持原顺序
+`tests/run.py` 的 runner 参数只有 `--serial`、`--shard N/TOTAL` 和 `--exclude-architecture-gate`；其余参数保持原顺序
 透传给 pytest，例如 `python tests/run.py -q --maxfail=1`。分片使用受版本控制的
 `tests/fixtures/durations.json` 慢文件耗时估算：按耗时降序，逐个分给预计总耗时最短的分片；
 耗时相同时按路径、分片编号决定归属，片内仍按路径排序。新增或未记录文件使用 1 秒权重，
 不依赖本机缓存，确保同一代码版本在本地与 CI 的文件集合和顺序一致。
 耗时数据来自成功 Ubuntu/Python 3.14 Coverage 日志的逐文件进度时间戳（含续行），
 只记录至少 3 秒的文件并向上取整；后续发现慢文件偏移时可依据成功 CI 日志更新。
+`--exclude-architecture-gate` 跳过 `ARCHITECTURE_GATE_TESTS` 列出的架构测试文件，这些文件由 CI 的
+`architecture` job 在同一进程内运行并共享全量扫描结果（`tests/architecture_cache.py`），覆盖率分片不再重复执行；
+本地默认全量仍包含它们。
 这些估算仅用于调度，不构成性能或覆盖率基线。
 
 - 不再使用 `python -m unittest discover`：它不导入 `tests` 包、收不到纯函数用例，且绕过 `conftest.py` 的隔离。
@@ -183,9 +186,9 @@ def test_recognize_prefers_explicit_identity(sample_meta, monkeypatch):
 - **合并检查复用**：单测/架构与 Pylint 工作流各自保留 PR、push 和手动入口。PR 完整通过全部门禁后，末尾 `CI proof (<github.sha>)` job 记录实际验证的模拟合并提交。合并 push 只在同一工作流的最新 PR 运行完整成功、证明成功、代码树完全相同、模拟合并父提交分别等于 push 前的目标分支和 PR head 时跳过重复门禁。直接 push、强制 push、基线变化、旧工作流缺少证明、失败/未完成运行或 API 异常都执行全量；不依赖提交消息，也不把 PR 的 head SHA 当作测试的合并 SHA。标准 merge/squash/rebase 仅在上述证据一致时复用。GitHub 构建与发布工作流保持独立。
 - **去重脚本验证**：`node --test .github/scripts/reuse.test.mjs` 离线覆盖成功复用及保守回退，两个检查工作流均在判定前执行。复用 job 仅持有 contents/actions/pull-requests 读取权限；没有修改分支保护设置。
 
-- **门禁**：`.github/workflows/test.yml` 在指向 `v3` 的 `pull_request` / `push` 及手动触发时，从 `uv.lock` 同步环境。独立 `architecture` job 先运行宿主依赖、运行契约和基线 CLI 快速门禁；全量测试通过 `coverage run --parallel-mode tests/run.py --shard N/8` 分到 8 个 job，一次执行同时验证单测并采集覆盖率。每个分片都有独立进程和临时 `CONFIG_DIR`，不共用 SQLite 或进程级状态，由单一报告 job 合并后检查 Application 与 Domain 的固定 80% 基线。
+- **门禁**：`.github/workflows/test.yml` 在指向 `v3` 的 `pull_request` / `push` 及手动触发时，从 `uv.lock` 同步环境。独立 `architecture` job 先运行宿主依赖、运行契约和基线 CLI 快速门禁；其余测试通过 `coverage run --parallel-mode tests/run.py --exclude-architecture-gate --shard N/6` 分到 6 个 job，一次执行同时验证单测并采集覆盖率；门禁已运行的架构测试不在分片中重复执行。每个分片都有独立进程和临时 `CONFIG_DIR`，不共用 SQLite 或进程级状态，由单一报告 job 合并后检查 Application 与 Domain 的固定 80% 基线。
 - **跨仓观察**：`.github/workflows/architecture-observe.yml` 每周或手工检出官方插件仓最新 `main`，使用 `--check-plugins` 比较公开导入、Hook 和动态 API 契约。它只上传 `official-plugin-architecture-report.json`，不会自动刷新 fixture；语义变化必须人工审查后显式执行 `--write-plugins`。
 - **静态检查**：`.github/workflows/pylint.yml` 对指向 `v3` 的 PR、推送和手工触发运行 Pylint。PR/推送改动到的 Python 文件是硬门禁；`app/` 全量扫描保留为建议性 JSON 构建工件，存量告警不会掩盖或阻塞本次增量治理。
 - **PR 本地验证**：按上文「验证范围与维护者安排」选择 contributor 默认检查或已记录的维护者安排，统一处理证据复用与失败归属；需要断点、输出顺序或测试污染诊断时使用 `--serial`。确认受影响路径与零真实出站，准确标注验证范围；本地执行安排不改变 CI 的全量验证与复用合同。
-- **覆盖率门禁**：`Unit Tests with Coverage` jobs 会在 `v3` 的 PR、push 和手工触发中通过 `tests/run.py --shard N/8` 并行采集覆盖率数据，`Coverage Report` 再合并全部分片并只读检查 Application 与 Domain 是否达到 Ubuntu/Python 3.14 canonical 的固定 80% 行覆盖率基线，同时上传 JSON / XML 工件。低于 80% 会阻塞；达到或超过 80% 不要求同步运行时语句计数。macOS 本地报告只用于诊断，不直接作为可提交基线。
+- **覆盖率门禁**：`Unit Tests with Coverage` jobs 会在 `v3` 的 PR、push 和手工触发中通过 `tests/run.py --exclude-architecture-gate --shard N/6` 并行采集覆盖率数据，`Coverage Report` 再合并全部分片并只读检查 Application 与 Domain 是否达到 Ubuntu/Python 3.14 canonical 的固定 80% 行覆盖率基线，同时上传 JSON / XML 工件。低于 80% 会阻塞；达到或超过 80% 不要求同步运行时语句计数。macOS 本地报告只用于诊断，不直接作为可提交基线。
 - 复现 CI 使用 `uv sync --locked`；主程序运行依赖位于 `[project].dependencies`，pytest 与覆盖率工具位于默认 `dev` 依赖组。

tests/architecture_cache.py

@@ -0,0 +1,47 @@
+"""架构测试共享的全量扫描结果：同一 pytest 进程内每类扫描只执行一次。
+
+门禁把多个架构测试文件放在同一进程运行，这些文件会对同一份源码重复做相同的全量收集。
+缓存只存放原始结果，每次返回深拷贝，测试修改返回值不会影响其他用例。
+"""
+
+from __future__ import annotations
+
+import ast
+import copy
+import functools
+from pathlib import Path
+from typing import Any
+
+from scripts.architecture.baseline import collect_current_event_facts, collect_dependency_baseline
+
+
+@functools.cache
+def _dependency_baseline() -> dict[str, Any]:
+    return collect_dependency_baseline()
+
+
+@functools.cache
+def _current_event_facts() -> dict[str, list[dict[str, Any]]]:
+    return collect_current_event_facts()
+
+
+def dependency_baseline() -> dict[str, Any]:
+    """返回当前宿主依赖基线的独立副本。"""
+    return copy.deepcopy(_dependency_baseline())
+
+
+def current_event_facts() -> dict[str, list[dict[str, Any]]]:
+    """返回当前宿主事件事实的独立副本。"""
+    return copy.deepcopy(_current_event_facts())
+
+
+@functools.cache
+def parse_module(path: Path) -> ast.Module:
+    """解析并缓存源码 AST；调用方只读取，不得修改返回的节点。"""
+    return ast.parse(path.read_text(encoding="utf-8-sig"), filename=str(path))
+
+
+@functools.cache
+def walk_module(tree: ast.Module) -> tuple[ast.AST, ...]:
+    """按 ``ast.walk`` 顺序返回模块全部节点，供多个测试重复遍历同一只读 AST。"""
+    return tuple(ast.walk(tree))

tests/fixtures/durations.json

@@ -1,37 +1,39 @@
 {
-  "source": "https://github.com/jxxghp/MoviePilot/actions/runs/34326739404",
-  "description": "Ubuntu/Python 3.14 coverage 日志中逐文件进度时间戳的耗时估算，含续行；仅记录至少 3 秒的文件并向上取整，其余文件使用 1 秒权重。用于分片均衡，不是性能或覆盖率基线。",
+  "source": "https://github.com/jxxghp/MoviePilot/actions/runs/36409216765",
+  "description": "Ubuntu/Python 3.14 coverage 日志中逐文件进度时间戳的耗时估算，含续行；仅记录至少 3 秒的文件并向上取整，其余文件使用 1 秒权重。不含架构门禁 job 运行的文件；用于分片均衡，不是性能或覆盖率基线。",
   "files": {
-    "test_agent_api_lazy_imports.py": 15,
-    "test_agent_api_surface_audit.py": 5,
-    "test_agent_lazy_runtime_boundary.py": 7,
-    "test_agent_tool_result_policy.py": 4,
-    "test_api_response.py": 29,
-    "test_architecture_adapter_imports.py": 67,
-    "test_architecture_baseline_cli.py": 7,
-    "test_architecture_contract_baseline.py": 6,
-    "test_architecture_dependencies.py": 95,
-    "test_architecture_egress.py": 67,
-    "test_architecture_event_facts.py": 17,
-    "test_architecture_event_policy.py": 16,
-    "test_bangumi_media_type.py": 4,
-    "test_chain_base_boundary.py": 9,
-    "test_data_cleanup_chain.py": 7,
-    "test_duplicate_code.py": 6,
-    "test_execute_command_tool.py": 4,
-    "test_host_runtime_context.py": 4,
+    "test_module_manager_capability_adapter.py": 38,
+    "test_agent_evaluation_runtime.py": 37,
+    "test_api_response.py": 22,
+    "test_router_aggregation.py": 20,
+    "test_agent_api_lazy_imports.py": 16,
+    "test_chain_base_boundary.py": 14,
+    "test_agent_lazy_runtime_boundary.py": 8,
+    "test_agent_evaluation_server.py": 7,
+    "test_service_locator_gate.py": 7,
+    "test_subscription_governance_scale.py": 7,
+    "test_transfer_history_architecture.py": 7,
+    "test_subscription_search_queue.py": 6,
+    "test_subscription_search_governance.py": 6,
+    "test_download_architecture.py": 6,
+    "test_duplicate_code.py": 5,
+    "test_optional_sdk_lazy_import.py": 5,
+    "test_workflow_mutation_command.py": 5,
+    "test_chain_layering.py": 5,
+    "test_agent_subagent_permissions.py": 5,
     "test_legacy_import_compat.py": 5,
-    "test_llm_facade_governance.py": 4,
-    "test_main_direct_execution.py": 4,
-    "test_media_auxiliary.py": 4,
-    "test_module_manager_capability_adapter.py": 36,
-    "test_monitor_mount_isolation.py": 5,
-    "test_music_matching.py": 6,
-    "test_router_aggregation.py": 10,
-    "test_service_locator_gate.py": 5,
-    "test_subscription_governance_scale.py": 6,
-    "test_subscription_search_governance.py": 4,
+    "test_monitor_mount_isolation.py": 4,
+    "test_data_locator_architecture.py": 4,
+    "test_workflow_invoke_plugin.py": 4,
+    "test_agent_api_surface_audit.py": 4,
+    "test_scheduler_architecture.py": 4,
+    "test_runtime_execution.py": 4,
+    "test_zspace_sync_items.py": 4,
+    "test_docker_bootstrap.py": 4,
     "test_task_ownership_gate.py": 4,
-    "test_transfer_history_architecture.py": 6
+    "test_host_runtime_context.py": 4,
+    "test_workflow_runtime_facade.py": 4,
+    "test_agent_tool_result_policy.py": 4,
+    "test_main_direct_execution.py": 4
   }
 }

tests/run.py

@@ -15,11 +15,22 @@ RUNNER_PATH = Path(__file__).resolve()
 DEFAULT_SHARD_COUNT = 4
 DURATIONS_PATH = TESTS_DIR / "fixtures" / "durations.json"
 DEFAULT_TEST_DURATION = 1.0
+# CI 架构门禁 job 独立运行的测试文件；分片使用 --exclude-architecture-gate 时不再重复执行
+ARCHITECTURE_GATE_TESTS = (
+    "test_architecture_dependencies.py",
+    "test_architecture_adapter_imports.py",
+    "test_architecture_egress.py",
+    "test_architecture_event_facts.py",
+    "test_architecture_event_policy.py",
+    "test_architecture_contract_baseline.py",
+    "test_architecture_baseline_cli.py",
+)
 
 
-def collect_test_files() -> list[Path]:
-    """按稳定路径顺序返回根测试目录中的全部测试文件。"""
-    return sorted(TESTS_DIR.glob("test_*.py"))
+def collect_test_files(exclude_architecture_gate: bool = False) -> list[Path]:
+    """按稳定路径顺序返回根测试目录中的测试文件，可排除已由架构门禁运行的文件。"""
+    excluded = set(ARCHITECTURE_GATE_TESTS) if exclude_architecture_gate else set()
+    return sorted(path for path in TESTS_DIR.glob("test_*.py") if path.name not in excluded)
 
 
 def load_test_durations() -> dict[str, float]:
@@ -77,9 +88,16 @@ def parse_args(argv: Sequence[str]) -> tuple[argparse.Namespace, list[str]]:
         metavar="N/TOTAL",
         help="只运行指定文件分片；CI 使用同一参数启动独立 job。",
     )
+    parser.add_argument(
+        "--exclude-architecture-gate",
+        action="store_true",
+        help="跳过 CI 架构门禁已运行的测试文件，供覆盖率分片使用。",
+    )
     args, pytest_args = parser.parse_known_args(argv)
     if args.serial and args.shard is not None:
         parser.error("--serial 不能与 --shard 同时使用")
+    if args.serial and args.exclude_architecture_gate:
+        parser.error("--serial 不能与 --exclude-architecture-gate 同时使用")
     return args, pytest_args
 
 
@@ -89,20 +107,23 @@ def run_pytest(paths: Sequence[Path], pytest_args: Sequence[str]) -> int:
 
 
 def _worker_command(
-    shard_index: int, shard_count: int, pytest_args: Sequence[str]
+    shard_index: int, shard_count: int, pytest_args: Sequence[str],
+    exclude_architecture_gate: bool = False,
 ) -> list[str]:
-    """构造与 CI 完全相同的单分片 worker 命令。"""
+    """构造与 CI 完全相同的单分片 worker 命令；排除选项必须一并传递，否则分片划分不一致。"""
     return [
         sys.executable,
         str(RUNNER_PATH),
         "--shard",
         f"{shard_index}/{shard_count}",
+        *(["--exclude-architecture-gate"] if exclude_architecture_gate else []),
         *pytest_args,
     ]
 
 
 def run_parallel_shards(
-    shards: Sequence[Sequence[Path]], pytest_args: Sequence[str]
+    shards: Sequence[Sequence[Path]], pytest_args: Sequence[str],
+    exclude_architecture_gate: bool = False,
 ) -> int:
     """启动独立 pytest 进程并等待全部文件分片结束。"""
     shard_count = len(shards)
@@ -116,7 +137,9 @@ def run_parallel_shards(
         )
         processes.append((
             shard_index,
-            subprocess.Popen(_worker_command(shard_index, shard_count, pytest_args)),
+            subprocess.Popen(_worker_command(
+                shard_index, shard_count, pytest_args, exclude_architecture_gate,
+            )),
         ))
 
     exit_code = 0
@@ -146,7 +169,7 @@ def main(argv: Sequence[str] | None = None) -> int:
     if args.serial:
         return run_pytest([TESTS_DIR], pytest_args)
 
-    test_files = collect_test_files()
+    test_files = collect_test_files(args.exclude_architecture_gate)
     if not test_files:
         print(f"未在 {TESTS_DIR} 找到 test_*.py", file=sys.stderr)
         return 2
@@ -164,7 +187,7 @@ def main(argv: Sequence[str] | None = None) -> int:
         return run_pytest(selected, pytest_args)
 
     shards = split_test_files(test_files, DEFAULT_SHARD_COUNT)
-    return run_parallel_shards(shards, pytest_args)
+    return run_parallel_shards(shards, pytest_args, args.exclude_architecture_gate)
 
 
 if __name__ == "__main__":

tests/test_architecture_adapter_imports.py

@@ -5,9 +5,9 @@ from collections import Counter
 from pathlib import Path
 
 from scripts.architecture.baseline import (
-    collect_dependency_baseline,
     collect_direct_adapter_imports,
 )
+from tests.architecture_cache import dependency_baseline
 
 PROJECT_ROOT = Path(__file__).parents[1]
 DEPENDENCY_POLICY_PATH = (
@@ -208,7 +208,7 @@ def test_direct_adapter_collector_handles_relative_imports_and_scope(
 
 def test_current_direct_adapter_imports_are_stable_generated_facts() -> None:
     """生成事实必须有稳定排序、自洽统计，且不保存符号或行号。"""
-    contract = collect_dependency_baseline()["direct_adapter_imports"]
+    contract = dependency_baseline()["direct_adapter_imports"]
     edges = contract["edges"]
     sources = sorted({edge["source"] for edge in edges})
     targets = sorted({edge["target"] for edge in edges})
@@ -237,7 +237,7 @@ def test_current_direct_adapter_imports_are_stable_generated_facts() -> None:
 def test_current_direct_adapter_imports_match_temporary_debt_policy() -> None:
     """现存直连必须逐条绑定冻结 owner，并允许债务集合只减不增。"""
     policy = json.loads(DEPENDENCY_POLICY_PATH.read_text(encoding="utf-8"))
-    contract = collect_dependency_baseline()["direct_adapter_imports"]
+    contract = dependency_baseline()["direct_adapter_imports"]
     adapter_policy = policy["direct_adapter_imports"]
     entries = adapter_policy["entries"]
 

tests/test_architecture_ci.py

@@ -5,6 +5,8 @@ from pathlib import Path
 
 from ruamel.yaml import YAML
 
+from tests.run import ARCHITECTURE_GATE_TESTS
+
 PROJECT_ROOT = Path(__file__).parents[1]
 WORKFLOW_ROOT = PROJECT_ROOT / ".github" / "workflows"
 
@@ -64,6 +66,13 @@ def test_unit_test_workflow_has_independent_host_architecture_gate():
     assert "scripts/architecture/mypy_ratchet.py --write" not in commands
     assert "scripts/architecture/service_locator.py" in commands
     assert "scripts/startup/performance.py --check --repeat 3" in commands
+    # 覆盖率分片跳过的文件必须恰好由门禁运行，否则会有测试在 CI 中无人执行
+    gate_test_files = {
+        token.removeprefix("tests/")
+        for token in commands.split()
+        if token.startswith("tests/test_") and token.endswith(".py")
+    }
+    assert gate_test_files == set(ARCHITECTURE_GATE_TESTS)
 
 
 def test_official_plugin_observation_is_scheduled_and_never_writes_fixture():
@@ -98,17 +107,8 @@ def test_coverage_jobs_parallelize_data_and_keep_one_global_ratchet() -> None:
     assert shard_job["name"] == "Unit Tests with Coverage (${{ matrix.shard }})"
     assert shard_job["timeout-minutes"] == 15
     shard_matrix = shard_job["strategy"]["matrix"]["include"]
-    assert [item["shard"] for item in shard_matrix] == [
-        "1/8",
-        "2/8",
-        "3/8",
-        "4/8",
-        "5/8",
-        "6/8",
-        "7/8",
-        "8/8",
-    ]
-    assert [item["index"] for item in shard_matrix] == [str(index) for index in range(1, 9)]
+    assert [item["shard"] for item in shard_matrix] == [f"{index}/6" for index in range(1, 7)]
+    assert [item["index"] for item in shard_matrix] == [str(index) for index in range(1, 7)]
     assert report_job["needs"] == "coverage-shard"
     assert report_job["runs-on"] == "ubuntu-latest"
     assert report_job["timeout-minutes"] == 10
@@ -121,7 +121,10 @@ def test_coverage_jobs_parallelize_data_and_keep_one_global_ratchet() -> None:
         step for step in shard_steps if step.get("name") == "Generate coverage data"
     )
     assert generate_step["timeout-minutes"] == 10
-    assert "coverage run --parallel-mode tests/run.py --shard" in generate_step["run"]
+    assert (
+        "coverage run --parallel-mode tests/run.py --exclude-architecture-gate --shard"
+        in generate_step["run"]
+    )
     upload_data_step = next(
         step for step in shard_steps if step.get("name") == "Upload coverage data"
     )

tests/test_architecture_dependencies.py

@@ -4,9 +4,6 @@ from functools import lru_cache
 from pathlib import Path
 
 from app.runtime.compat.manifest import MODULE_ALIASES, SYMBOL_ALIASES
-from scripts.architecture.baseline import (
-    collect_current_event_facts as _collect_current_event_facts,
-)
 from scripts.architecture.baseline import (
     discover_modules as _discover_modules,
 )
@@ -16,6 +13,11 @@ from scripts.architecture.baseline import (
 from scripts.architecture.baseline import (
     strongly_connected_components as _strongly_connected_components,
 )
+from tests.architecture_cache import (
+    current_event_facts as _collect_current_event_facts,
+)
+from tests.architecture_cache import parse_module as _parse_source
+from tests.architecture_cache import walk_module as _walk_module
 
 PROJECT_ROOT = Path(__file__).parents[1]
 APP_ROOT = PROJECT_ROOT / "app"
@@ -212,9 +214,9 @@ FORBIDDEN_IMPORT_PREFIXES = {
 
 def _legacy_imports(path: Path) -> set[str]:
     """提取源码中的静态和常量动态旧路径导入。"""
-    tree = ast.parse(path.read_text(encoding="utf-8-sig"), filename=str(path))
+    tree = _parse_source(path)
     imports: set[str] = set()
-    for node in ast.walk(tree):
+    for node in _walk_module(tree):
         candidates: list[str] = []
         if isinstance(node, ast.Import):
             candidates.extend(alias.name for alias in node.names)
@@ -251,7 +253,7 @@ def _compat_symbol_references(tree: ast.AST) -> set[tuple[int, str]]:
     module_bindings: dict[str, str] = {}
     references: set[tuple[int, str]] = set()
 
-    for node in ast.walk(tree):
+    for node in _walk_module(tree):
         if isinstance(node, ast.Import):
             for alias in node.names:
                 binding = alias.asname or alias.name.split(".", maxsplit=1)[0]
@@ -264,7 +266,7 @@ def _compat_symbol_references(tree: ast.AST) -> set[tuple[int, str]]:
                 binding = alias.asname or alias.name
                 module_bindings[binding] = f"{node.module}.{alias.name}"
 
-    for node in ast.walk(tree):
+    for node in _walk_module(tree):
         if not isinstance(node, ast.Attribute):
             continue
         parts = _attribute_parts(node)
@@ -282,7 +284,7 @@ def _compat_symbol_references(tree: ast.AST) -> set[tuple[int, str]]:
 
 def _class_annotations(path: Path, class_name: str) -> dict[str, str]:
     """返回指定类的源码级字段注解。"""
-    tree = ast.parse(path.read_text(encoding="utf-8-sig"), filename=str(path))
+    tree = _parse_source(path)
     class_node = next(node for node in tree.body if isinstance(node, ast.ClassDef) and node.name == class_name)
     return {
         node.target.id: ast.unparse(node.annotation)
@@ -373,8 +375,8 @@ def test_runtime_dependencies_use_same_named_single_word_package() -> None:
         relative = path.relative_to(APP_ROOT)
         if relative.parts[0] == "plugins":
             continue
-        tree = ast.parse(path.read_text(encoding="utf-8-sig"), filename=str(path))
-        for node in ast.walk(tree):
+        tree = _parse_source(path)
+        for node in _walk_module(tree):
             if isinstance(node, ast.ImportFrom) and node.module == "app.runtime.dependencies":
                 violations.append(f"{relative}:{node.lineno}:from-package-root")
             elif isinstance(node, ast.Import):
@@ -416,8 +418,8 @@ def test_domain_classification_is_a_pure_direct_import_package() -> None:
     )
     violations: list[str] = []
     for path in package.glob("*.py"):
-        tree = ast.parse(path.read_text(encoding="utf-8-sig"), filename=str(path))
-        for node in ast.walk(tree):
+        tree = _parse_source(path)
+        for node in _walk_module(tree):
             imported_modules: list[str] = []
             if isinstance(node, ast.ImportFrom) and node.module:
                 imported_modules.append(node.module)
@@ -434,8 +436,8 @@ def test_domain_classification_is_a_pure_direct_import_package() -> None:
     for path in APP_ROOT.rglob("*.py"):
         if path.is_relative_to(APP_ROOT / "plugins"):
             continue
-        tree = ast.parse(path.read_text(encoding="utf-8-sig"), filename=str(path))
-        for node in ast.walk(tree):
+        tree = _parse_source(path)
+        for node in _walk_module(tree):
             if isinstance(node, ast.ImportFrom) and node.module == "app.domain.classification":
                 root_imports.append(f"{path.relative_to(PROJECT_ROOT).as_posix()}:{node.lineno}")
             elif isinstance(node, ast.Import):
@@ -492,8 +494,8 @@ def test_application_torrent_uses_same_named_single_word_package() -> None:
         for path in root.rglob("*.py"):
             if path.is_relative_to(APP_ROOT / "plugins"):
                 continue
-            tree = ast.parse(path.read_text(encoding="utf-8-sig"), filename=str(path))
-            for node in ast.walk(tree):
+            tree = _parse_source(path)
+            for node in _walk_module(tree):
                 if isinstance(node, ast.ImportFrom) and node.module == "app.application.torrent":
                     root_imports.append(f"{path.relative_to(PROJECT_ROOT).as_posix()}:{node.lineno}")
     assert root_imports == []
@@ -510,15 +512,15 @@ def test_host_module_package_roots_only_export_capability_entrypoints() -> None:
     for path in (APP_ROOT / "modules").rglob("*.py"):
         if path.is_relative_to(APP_ROOT / "plugins"):
             continue
-        tree = ast.parse(path.read_text(encoding="utf-8-sig"), filename=str(path))
-        for node in ast.walk(tree):
+        tree = _parse_source(path)
+        for node in _walk_module(tree):
             if isinstance(node, ast.ImportFrom) and node.module == "app.modules._base":
                 root_imports.append(f"{path.relative_to(PROJECT_ROOT).as_posix()}:{node.lineno}")
     assert root_imports == []
 
     for package_name, expected_exports in HOST_MODULE_PACKAGE_EXPORTS.items():
         path = APP_ROOT / "modules" / package_name / "__init__.py"
-        tree = ast.parse(path.read_text(encoding="utf-8"), filename=str(path))
+        tree = _parse_source(path)
         assignments = {
             target.id: ast.literal_eval(node.value)
             for node in tree.body
@@ -588,8 +590,8 @@ def test_domain_media_projection_uses_single_word_owner_package() -> None:
         relative = path.relative_to(APP_ROOT)
         if relative.parts[0] == "plugins" or path == context_path:
             continue
-        tree = ast.parse(path.read_text(encoding="utf-8-sig"), filename=str(path))
-        for node in ast.walk(tree):
+        tree = _parse_source(path)
+        for node in _walk_module(tree):
             if not isinstance(node, ast.Attribute) or not isinstance(node.value, ast.Name):
                 continue
             if node.value.id == "MediaInfo" and node.attr in {
@@ -605,7 +607,7 @@ def test_domain_media_projection_uses_single_word_owner_package() -> None:
 def test_workflow_query_contract_returns_only_typed_snapshots():
     """工作流正式查询端口不得退化为 Any 或 ORM 返回值。"""
     path = APP_ROOT / "application" / "workflow.py"
-    tree = ast.parse(path.read_text(encoding="utf-8"), filename=str(path))
+    tree = _parse_source(path)
     query_classes = {
         node.name: node
         for node in tree.body
@@ -1085,10 +1087,10 @@ def test_runtime_composition_is_the_sole_host_and_domain_runtime_owner():
     for path in APP_ROOT.rglob("*.py"):
         if path.is_relative_to(APP_ROOT / "plugins"):
             continue
-        tree = ast.parse(path.read_text(encoding="utf-8"), filename=str(path))
+        tree = _parse_source(path)
         if any(
             isinstance(node, ast.Call) and isinstance(node.func, ast.Name) and node.func.id == "HostRuntime"
-            for node in ast.walk(tree)
+            for node in _walk_module(tree)
         ):
             constructors.append(path.relative_to(APP_ROOT).as_posix())
     assert constructors == ["startup/composition/runtime.py"]
@@ -1218,8 +1220,8 @@ def test_plugin_route_refresh_is_imported_from_its_application_owner():
         relative = path.relative_to(APP_ROOT)
         if relative.parts[0] == "plugins":
             continue
-        tree = ast.parse(path.read_text(encoding="utf-8"))
-        for node in ast.walk(tree):
+        tree = _parse_source(path)
+        for node in _walk_module(tree):
             if not isinstance(node, ast.ImportFrom) or node.level:
                 continue
             if node.module == owner:
@@ -1299,7 +1301,7 @@ def test_host_code_does_not_use_compat_symbol_aliases() -> None:
             or relative.parts[:2] == ("sdk", "_legacy")
         ):
             continue
-        tree = ast.parse(path.read_text(encoding="utf-8-sig"), filename=str(path))
+        tree = _parse_source(path)
         violations.extend(
             f"{relative.as_posix()}:{line}:{symbol}" for line, symbol in sorted(_compat_symbol_references(tree))
         )
@@ -1321,8 +1323,8 @@ def test_host_code_uses_explicit_runtime_facade_getters():
             "compat",
         ):
             continue
-        tree = ast.parse(path.read_text(encoding="utf-8-sig"), filename=str(path))
-        for node in ast.walk(tree):
+        tree = _parse_source(path)
+        for node in _walk_module(tree):
             if isinstance(node, ast.Assign) and isinstance(node.value, ast.Name):
                 target_names = {target.id for target in node.targets if isinstance(target, ast.Name)}
                 if "SystemConfigOper" in target_names and node.value.id == "get_configured_system_config":
@@ -1371,8 +1373,8 @@ def test_user_chain_and_agent_ports_are_typed_and_orm_free():
     ]
     violations: list[str] = []
     for path in production_paths:
-        tree = ast.parse(path.read_text(encoding="utf-8-sig"), filename=str(path))
-        for node in ast.walk(tree):
+        tree = _parse_source(path)
+        for node in _walk_module(tree):
             if isinstance(node, ast.ClassDef) and node.name == "UserOper":
                 violations.append(f"{path.relative_to(PROJECT_ROOT)}:{node.lineno}:class")
             if isinstance(node, ast.ImportFrom) and node.module == "app.db.oper.user":
@@ -1411,8 +1413,8 @@ def test_transfer_pending_oper_import_is_confined_to_database_boundary():
         relative = path.relative_to(PROJECT_ROOT).as_posix()
         if relative.startswith("app/plugins/") or relative in allowed_paths:
             continue
-        tree = ast.parse(path.read_text(encoding="utf-8-sig"), filename=str(path))
-        for node in ast.walk(tree):
+        tree = _parse_source(path)
+        for node in _walk_module(tree):
             if isinstance(node, ast.Import):
                 if any(alias.name == "app.db.oper.transferpending" for alias in node.names):
                     violations.append(f"{relative}:{node.lineno}")
@@ -1431,16 +1433,16 @@ def test_transfer_pending_oper_import_is_confined_to_database_boundary():
 def test_startup_injects_transactional_transfer_admission_repository():
     """启动组合根必须向 Chain 注入事务型整理准入仓储。"""
     path = APP_ROOT / "startup" / "composition" / "chain.py"
-    tree = ast.parse(path.read_text(encoding="utf-8-sig"), filename=str(path))
+    tree = _parse_source(path)
     imports_repository = any(
         isinstance(node, ast.ImportFrom)
         and node.module == "app.db.adapters.transfer.admission"
         and any(alias.name == "TransactionalTransferAdmissionRepository" for alias in node.names)
-        for node in ast.walk(tree)
+        for node in _walk_module(tree)
     )
     chain_context_calls = [
         node
-        for node in ast.walk(tree)
+        for node in _walk_module(tree)
         if isinstance(node, ast.Call) and isinstance(node.func, ast.Name) and node.func.id == "ChainRuntimeContext"
     ]
 
@@ -1480,10 +1482,10 @@ def test_scheduler_does_not_depend_on_database_implementation():
 def test_monitor_dispatcher_uses_explicit_history_port_getter():
     """监控分发器不得把兼容 TransferHistoryPort 伪装成数据库 Oper。"""
     path = APP_ROOT / "monitor" / "dispatcher.py"
-    tree = ast.parse(path.read_text(encoding="utf-8-sig"), filename=str(path))
+    tree = _parse_source(path)
     violations = [
         f"{path.relative_to(PROJECT_ROOT).as_posix()}:{node.lineno}"
-        for node in ast.walk(tree)
+        for node in _walk_module(tree)
         if isinstance(node, ast.ImportFrom)
         and node.module == "app.application.history"
         and any(alias.name == "TransferHistoryPort" for alias in node.names)
@@ -1505,8 +1507,8 @@ def test_canonical_service_config_consumers_use_application_directory():
     ]
     violations: list[str] = []
     for path in paths:
-        tree = ast.parse(path.read_text(encoding="utf-8-sig"), filename=str(path))
-        for node in ast.walk(tree):
+        tree = _parse_source(path)
+        for node in _walk_module(tree):
             if isinstance(node, ast.ImportFrom) and node.module == "app.runtime.extensions.service":
                 violations.append(f"{path.relative_to(PROJECT_ROOT).as_posix()}:{node.lineno}")
             if (
@@ -1526,7 +1528,7 @@ def test_plugin_components_do_not_reexport_legacy_abi_names():
     manager_path = APP_ROOT / "runtime" / "extensions" / "plugin" / "manager.py"
     for root in PLUGIN_COMPONENT_ROOTS:
         for path in (PROJECT_ROOT / root).rglob("*.py"):
-            tree = ast.parse(path.read_text(encoding="utf-8-sig"), filename=str(path))
+            tree = _parse_source(path)
             for node in tree.body:
                 if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
                     if node.name == "__getattr__" or node.name in PLUGIN_LEGACY_ABI_NAMES:
@@ -1562,8 +1564,8 @@ def test_host_code_uses_precise_schema_modules():
         relative = path.relative_to(APP_ROOT)
         if relative.parts[0] in {"plugins", "schemas"}:
             continue
-        tree = ast.parse(path.read_text(encoding="utf-8-sig"), filename=str(path))
-        for node in ast.walk(tree):
+        tree = _parse_source(path)
+        for node in _walk_module(tree):
             if isinstance(node, ast.ImportFrom) and node.module == "app":
                 if any(alias.name == "schemas" for alias in node.names):
                     violations.append(str(relative))
@@ -1580,9 +1582,9 @@ def test_database_internals_do_not_import_db_facades():
     for path in (APP_ROOT / "db").rglob("*.py"):
         if path.name == "__init__.py":
             continue
-        tree = ast.parse(path.read_text(encoding="utf-8-sig"), filename=str(path))
+        tree = _parse_source(path)
         if any(
-            isinstance(node, ast.ImportFrom) and node.module in {"app.db", "app.db.models"} for node in ast.walk(tree)
+            isinstance(node, ast.ImportFrom) and node.module in {"app.db", "app.db.models"} for node in _walk_module(tree)
         ):
             violations.append(str(path.relative_to(PROJECT_ROOT)))
     assert violations == []
@@ -1593,8 +1595,8 @@ def test_database_opers_use_dboper_transaction_dispatchers():
     runner_names = {"run_sync_transaction", "run_async_transaction"}
     violations: list[str] = []
     for path in (APP_ROOT / "db" / "oper").rglob("*.py"):
-        tree = ast.parse(path.read_text(encoding="utf-8-sig"), filename=str(path))
-        for node in ast.walk(tree):
+        tree = _parse_source(path)
+        for node in _walk_module(tree):
             if not isinstance(node, ast.ImportFrom) or node.module != "app.db.uow":
                 continue
             imported = {alias.name for alias in node.names} & runner_names
@@ -1621,9 +1623,9 @@ def test_models_and_base_require_explicit_database_sessions():
     paths = [APP_ROOT / "db" / "base.py"]
     paths.extend((APP_ROOT / "db" / "models").rglob("*.py"))
     for path in paths:
-        tree = ast.parse(path.read_text(encoding="utf-8-sig"), filename=str(path))
+        tree = _parse_source(path)
         relative = str(path.relative_to(PROJECT_ROOT))
-        nodes = list(ast.walk(tree))
+        nodes = list(_walk_module(tree))
         for node in nodes:
             if isinstance(node, ast.ImportFrom) and node.module == "app.db.decorators":
                 violations.append(f"{relative}:{node.lineno}:decorator-import")
@@ -1662,8 +1664,8 @@ def test_plugin_sdk_does_not_import_or_export_host_models():
     violations: list[str] = []
     for path in (APP_ROOT / "sdk").rglob("*.py"):
         relative = path.relative_to(PROJECT_ROOT).as_posix()
-        tree = ast.parse(path.read_text(encoding="utf-8-sig"), filename=str(path))
-        for node in ast.walk(tree):
+        tree = _parse_source(path)
+        for node in _walk_module(tree):
             if (
                 isinstance(node, ast.ImportFrom)
                 and node.module
@@ -1753,9 +1755,9 @@ def test_application_does_not_import_transport_frameworks():
     """应用层不得依赖 FastAPI、Starlette 或宿主 HTTP 适配器。"""
     violations: dict[str, set[str]] = {}
     for path in (APP_ROOT / "application").rglob("*.py"):
-        tree = ast.parse(path.read_text(encoding="utf-8-sig"), filename=str(path))
+        tree = _parse_source(path)
         forbidden: set[str] = set()
-        for node in ast.walk(tree):
+        for node in _walk_module(tree):
             candidates: list[str] = []
             if isinstance(node, ast.Import):
                 candidates.extend(alias.name for alias in node.names)
@@ -1790,8 +1792,8 @@ def test_host_code_does_not_use_string_utils_facade():
         relative = path.relative_to(APP_ROOT)
         if relative.parts[0] in {"plugins", "sdk"}:
             continue
-        tree = ast.parse(path.read_text(encoding="utf-8-sig"), filename=str(path))
-        if any(isinstance(node, ast.Name) and node.id == "StringUtils" for node in ast.walk(tree)):
+        tree = _parse_source(path)
+        if any(isinstance(node, ast.Name) and node.id == "StringUtils" for node in _walk_module(tree)):
             violations.append(str(relative))
     assert violations == []
 
@@ -1804,8 +1806,8 @@ def test_host_code_does_not_import_chain_plugin_abi_roots() -> None:
         relative = path.relative_to(APP_ROOT)
         if relative.parts[0] == "plugins":
             continue
-        tree = ast.parse(path.read_text(encoding="utf-8-sig"), filename=str(path))
-        for node in ast.walk(tree):
+        tree = _parse_source(path)
+        for node in _walk_module(tree):
             if isinstance(node, ast.ImportFrom) and node.module in compatibility_roots:
                 violations.append(f"{relative}:{node.lineno}:{node.module}")
             elif isinstance(node, ast.Import):
@@ -1833,8 +1835,8 @@ def test_foundation_does_not_emit_runtime_logs():
     """基础机制不打印或初始化日志系统，运行期诊断由上层调用方负责。"""
     violations: list[str] = []
     for path in (APP_ROOT / "foundation").rglob("*.py"):
-        tree = ast.parse(path.read_text(encoding="utf-8-sig"), filename=str(path))
-        for node in ast.walk(tree):
+        tree = _parse_source(path)
+        for node in _walk_module(tree):
             if isinstance(node, ast.Import) and any(alias.name == "logging" for alias in node.names):
                 violations.append(str(path.relative_to(PROJECT_ROOT)))
                 break
@@ -1880,12 +1882,12 @@ def test_passkey_application_does_not_select_cache_backend():
 def test_startup_explicitly_configures_passkey_challenge_cache():
     """PassKey challenge 缓存必须由启动组合根显式装配。"""
     path = APP_ROOT / "startup" / "composition" / "security.py"
-    tree = ast.parse(path.read_text(encoding="utf-8-sig"), filename=str(path))
+    tree = _parse_source(path)
     configured = any(
         isinstance(node, ast.Call)
         and isinstance(node.func, ast.Name)
         and node.func.id == "configure_passkey_challenge_cache"
-        for node in ast.walk(tree)
+        for node in _walk_module(tree)
     )
 
     assert configured is True
@@ -1947,8 +1949,8 @@ def test_chain_does_not_import_downloader_sdks():
     forbidden_sdks = {"qbittorrentapi", "transmission_rpc"}
     violations: list[str] = []
     for path in (APP_ROOT / "chain").rglob("*.py"):
-        tree = ast.parse(path.read_text(encoding="utf-8-sig"), filename=str(path))
-        for node in ast.walk(tree):
+        tree = _parse_source(path)
+        for node in _walk_module(tree):
             names: list[str] = []
             if isinstance(node, ast.Import):
                 names.extend(alias.name for alias in node.names)
@@ -2110,10 +2112,10 @@ def test_host_consumers_get_agent_manager_through_application_facade():
     for module_name, path in _discover_modules().items():
         if module_name.startswith(("app.agent", "app.startup")):
             continue
-        tree = ast.parse(path.read_text(encoding="utf-8"), filename=str(path))
+        tree = _parse_source(path)
         imported = {
             alias.name
-            for node in ast.walk(tree)
+            for node in _walk_module(tree)
             if isinstance(node, ast.ImportFrom) and node.module == "app.agent.loader"
             for alias in node.names
             if alias.name in forbidden
@@ -2127,7 +2129,7 @@ def test_agent_package_roots_do_not_duplicate_implementation_exports():
     """Agent 与 LLM 包根不得实现动态转发，旧符号只能由精确 Compat 路由承接。"""
     for relative_path in ("app/agent/__init__.py", "app/agent/llm/__init__.py"):
         path = PROJECT_ROOT / relative_path
-        tree = ast.parse(path.read_text(encoding="utf-8"), filename=str(path))
+        tree = _parse_source(path)
         assert all(isinstance(node, ast.Expr) for node in tree.body), relative_path
 
     assert set(SYMBOL_ALIASES["app.agent"]) == {
@@ -2150,10 +2152,10 @@ def test_host_code_imports_agent_and_llm_symbols_from_owner_modules():
     """宿主不得消费 Agent 包根兼容符号，避免包根再次成为第二公开面。"""
     violations: dict[str, set[str]] = {}
     for module_name, path in _discover_modules().items():
-        tree = ast.parse(path.read_text(encoding="utf-8"), filename=str(path))
+        tree = _parse_source(path)
         imported = {
             f"{node.module}.{alias.name}"
-            for node in ast.walk(tree)
+            for node in _walk_module(tree)
             if isinstance(node, ast.ImportFrom) and node.module in {"app.agent", "app.agent.llm"}
             for alias in node.names
         }
@@ -2169,10 +2171,10 @@ def test_host_consumers_resolve_llm_provider_runtime_through_gateway():
     for module_name, path in _discover_modules().items():
         if module_name.startswith(("app.agent", "app.startup")):
             continue
-        tree = ast.parse(path.read_text(encoding="utf-8"), filename=str(path))
+        tree = _parse_source(path)
         imported = {
             alias.name
-            for node in ast.walk(tree)
+            for node in _walk_module(tree)
             if isinstance(node, ast.ImportFrom) and node.module in {"app.agent.llm", "app.agent.llm.provider"}
             for alias in node.names
             if alias.name == "LLMProviderManager"
@@ -2191,10 +2193,10 @@ def test_host_consumers_use_agent_audio_capability_application_port():
     for module_name, path in _discover_modules().items():
         if module_name.startswith(("app.agent", "app.startup")):
             continue
-        tree = ast.parse(path.read_text(encoding="utf-8"), filename=str(path))
+        tree = _parse_source(path)
         imported = {
             alias.name
-            for node in ast.walk(tree)
+            for node in _walk_module(tree)
             if isinstance(node, ast.ImportFrom) and node.module in {"app.agent.llm", "app.agent.llm.capability"}
             for alias in node.names
             if alias.name == "AgentCapabilityManager"
@@ -2246,8 +2248,8 @@ def test_modules_read_deployment_settings_through_runtime_port():
     """宿主 Module 不得绕过 runtime 配置端口直接依赖 Settings 实例。"""
     violations: list[str] = []
     for path in (APP_ROOT / "modules").rglob("*.py"):
-        tree = ast.parse(path.read_text(encoding="utf-8-sig"), filename=str(path))
-        for node in ast.walk(tree):
+        tree = _parse_source(path)
+        for node in _walk_module(tree):
             if not isinstance(node, ast.ImportFrom) or node.module != "app.runtime.config":
                 continue
             if any(alias.name == "settings" for alias in node.names):
@@ -2263,8 +2265,8 @@ def test_runtime_implementation_does_not_use_legacy_settings_proxy():
     for path in (APP_ROOT / "runtime").rglob("*.py"):
         if path == APP_ROOT / "runtime" / "settings.py":
             continue
-        tree = ast.parse(path.read_text(encoding="utf-8-sig"), filename=str(path))
-        for node in ast.walk(tree):
+        tree = _parse_source(path)
+        for node in _walk_module(tree):
             if not isinstance(node, ast.ImportFrom):
                 continue
             if node.module != "app.runtime.settings":
@@ -2295,12 +2297,12 @@ def test_deprecated_settings_proxy_imports_are_zero():
             continue
         if path.is_relative_to(APP_ROOT / "plugins"):
             continue
-        tree = ast.parse(path.read_text(encoding="utf-8-sig"), filename=str(path))
+        tree = _parse_source(path)
         imports_compat = any(
             isinstance(node, ast.ImportFrom)
             and node.module == "app.runtime.settings"
             and any(alias.name.lower().endswith("compat") for alias in node.names)
-            for node in ast.walk(tree)
+            for node in _walk_module(tree)
         )
         if not imports_compat:
             continue
@@ -2324,12 +2326,12 @@ def test_global_settings_imports_stay_within_compatibility_baseline():
     for path in APP_ROOT.rglob("*.py"):
         if path.is_relative_to(APP_ROOT / "plugins"):
             continue
-        tree = ast.parse(path.read_text(encoding="utf-8-sig"), filename=str(path))
+        tree = _parse_source(path)
         if any(
             isinstance(node, ast.ImportFrom)
             and node.module == "app.runtime.config"
             and any(alias.name == "settings" for alias in node.names)
-            for node in ast.walk(tree)
+            for node in _walk_module(tree)
         ):
             imports.add(path.relative_to(PROJECT_ROOT).as_posix())
 

tests/test_architecture_egress.py

@@ -3,8 +3,8 @@
 import json
 from pathlib import Path
 
-from scripts.architecture.baseline import collect_dependency_baseline
 from scripts.architecture.egress import collect_direct_egress
+from tests.architecture_cache import dependency_baseline
 
 PROJECT_ROOT = Path(__file__).parents[1]
 DEPENDENCY_POLICY_PATH = (
@@ -672,7 +672,7 @@ class Service:
 
 def test_current_egress_facts_are_complete_and_self_consistent() -> None:
     """当前宿主 egress identity 必须完整、可收缩、排序且统计自洽。"""
-    value = collect_dependency_baseline()["direct_egress"]
+    value = dependency_baseline()["direct_egress"]
     entries = value["entries"]
     application_entries = [
         entry
@@ -722,7 +722,7 @@ def test_current_egress_facts_are_complete_and_self_consistent() -> None:
 
 def test_current_egress_facts_match_exact_policy() -> None:
     """现存事实必须逐条分类，且 registry/scope 不得与 collector 漂移。"""
-    facts = collect_dependency_baseline()["direct_egress"]
+    facts = dependency_baseline()["direct_egress"]
     policy = json.loads(DEPENDENCY_POLICY_PATH.read_text(encoding="utf-8"))
     egress_policy = policy["direct_egress"]
     groups = egress_policy["groups"]

tests/test_architecture_event_facts.py

@@ -7,6 +7,7 @@ from scripts.architecture.event_facts import (
     collect_event_facts,
     fingerprint_event_fact,
 )
+from tests.architecture_cache import current_event_facts
 
 EVENT_MEMBERS = {
     "EventType": ("Alpha", "Beta", "Gamma"),
@@ -1135,15 +1136,7 @@ def handler(event):
 
 def test_collect_event_facts_matches_current_host_inventory() -> None:
     """统一事实覆盖当前 86 个 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")},
-    )
+    facts = current_event_facts()
     producers = facts["producers"]
     consumers = facts["consumers"]
 

tests/test_architecture_event_policy.py

@@ -7,7 +7,6 @@ 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,
@@ -19,6 +18,7 @@ from scripts.architecture.event_policy import (
     parse_args,
     validate_event_consumer_policy,
 )
+from tests.architecture_cache import current_event_facts
 
 
 def _fact(
@@ -112,7 +112,7 @@ def test_event_policy_accepts_exact_static_and_dynamic_facts() -> None:
 
 def test_current_event_consumer_policy_matches_exact_reviewed_set() -> None:
     """当前 17 条宿主 consumer 必须逐条匹配人工政策且没有非法成员。"""
-    facts = collect_current_event_facts()["consumers"]
+    facts = current_event_facts()["consumers"]
     policy = json.loads(DEFAULT_EVENT_POLICY_PATH.read_text(encoding="utf-8"))
     entries = policy["event_consumers"]["entries"]
 

tests/test_test_runner.py

@@ -53,11 +53,11 @@ def test_split_test_files_rejects_nonpositive_shard_count() -> None:
 
 
 def test_recorded_durations_balance_the_current_ci_suite() -> None:
-    """慢文件权重必须有效，并将当前全集的八分片预计耗时维持均衡。"""
+    """慢文件权重必须有效，并将 CI 分片实际运行的文件集的六分片预计耗时维持均衡。"""
     durations = test_runner.load_test_durations()
     assert all(value > 0 for value in durations.values())
-    test_files = test_runner.collect_test_files()
-    shards = test_runner.split_test_files(test_files, shard_count=8)
+    test_files = test_runner.collect_test_files(exclude_architecture_gate=True)
+    shards = test_runner.split_test_files(test_files, shard_count=6)
     totals = [sum(durations.get(path.name, 1.0) for path in shard) for shard in shards]
 
     assert sorted(path for shard in shards for path in shard) == test_files
@@ -71,9 +71,9 @@ def test_main_defaults_to_four_parallel_shards(monkeypatch) -> None:
     """无 runner 参数时应并行执行四个独立 pytest 文件分片。"""
     test_files = _test_files(10)
     captured = {}
-    monkeypatch.setattr(test_runner, "collect_test_files", lambda: test_files)
+    monkeypatch.setattr(test_runner, "collect_test_files", lambda *_: test_files)
 
-    def fake_run_parallel(shards, pytest_args):
+    def fake_run_parallel(shards, pytest_args, _exclude_architecture_gate=False):
         """记录默认并行入口接收的分片与 pytest 参数。"""
         captured["shards"] = shards
         captured["pytest_args"] = pytest_args
@@ -90,7 +90,7 @@ def test_main_runs_requested_ci_shard_in_current_process(monkeypatch) -> None:
     """CI 指定分片时只运行该分片，并继续透传 pytest 参数。"""
     test_files = _test_files(10)
     captured = {}
-    monkeypatch.setattr(test_runner, "collect_test_files", lambda: test_files)
+    monkeypatch.setattr(test_runner, "collect_test_files", lambda *_: test_files)
 
     def fake_run_pytest(paths, pytest_args):
         """记录 pytest 入口接收的文件与透传参数。"""
@@ -122,6 +122,24 @@ def test_main_serial_preserves_legacy_full_suite_entry(monkeypatch) -> None:
     assert captured["pytest_args"] == ["-q", "--maxfail=1"]
 
 
+def test_architecture_gate_files_are_excluded_only_on_request() -> None:
+    """默认全量包含门禁文件；排除后剩余文件与门禁文件恰好构成全集。"""
+    full = test_runner.collect_test_files()
+    remaining = test_runner.collect_test_files(exclude_architecture_gate=True)
+    gate = [path for path in full if path.name in test_runner.ARCHITECTURE_GATE_TESTS]
+
+    assert sorted(path.name for path in gate) == sorted(test_runner.ARCHITECTURE_GATE_TESTS)
+    assert sorted([*remaining, *gate]) == full
+
+
+def test_parallel_workers_inherit_architecture_gate_exclusion() -> None:
+    """并行 worker 必须收到同一排除选项，否则各自划分的分片会遗漏或重复文件。"""
+    command = test_runner._worker_command(2, 4, ["-q"], exclude_architecture_gate=True)
+
+    assert command[2:] == ["--shard", "2/4", "--exclude-architecture-gate", "-q"]
+    assert "--exclude-architecture-gate" not in test_runner._worker_command(2, 4, ["-q"])
+
+
 @pytest.mark.parametrize("value", ["0/4", "5/4", "1/0", "invalid"])
 def test_invalid_shard_values_are_rejected(value: str) -> None:
     """分片参数必须使用有效的一基 N/TOTAL 范围。"""
@@ -134,6 +152,6 @@ def test_workflow_uses_the_shared_runner_contract() -> None:
     workflow = WORKFLOW.read_text(encoding="utf-8")
 
     assert 'run: uv run --locked --no-sync python tests/run.py --shard' not in workflow
-    assert "python -m coverage run --parallel-mode tests/run.py --shard" in workflow
+    assert "python -m coverage run --parallel-mode tests/run.py --exclude-architecture-gate --shard" in workflow
     assert "mapfile" not in workflow
     assert "SHARD_INDEX" not in workflow