""" 验证模块归属、兼容导入路径及分层依赖的架构约束。 """ import ast import json from functools import lru_cache from pathlib import Path from app.runtime.compat.manifest import MODULE_ALIASES, SYMBOL_ALIASES from scripts.architecture.baseline import ( discover_modules as _discover_modules, ) from scripts.architecture.baseline import ( resolve_imports as _resolve_imports, ) 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" DEPENDENCY_POLICY_PATH = PROJECT_ROOT / "tests" / "fixtures" / "architecture" / "dependency-policy.json" LEGACY_ROOTS = ("app.core", "app.helper", "app.utils") LEGACY_MODULES = {"app.log"} IMPLEMENTATION_ROOTS = ( "app.agent.skills", "app.adapters", "app.application", "app.db.adapters", "app.domain", "app.foundation", "app.runtime", ) CYCLE_ROOTS = (*IMPLEMENTATION_ROOTS, "app.runtime.compat", "app.sdk") RETIRED_CANONICAL_ROOTS = ( "compat", "extensions", "infrastructure", "integrations", "messaging", "platform", "security", "services", ) RETIRED_CANONICAL_FILES = ( "app/infrastructure/package_installer.py", "app/infrastructure/resource_updater.py", "app/infrastructure/rust_accel.py", "app/messaging/agent_bridge.py", "app/platform/config_reload.py", "app/platform/rate_limit.py", "app/platform/thread_pool.py", "app/services/filter_rules.py", "app/services/transfer_history.py", "app/extensions/module_loader.py", "app/extensions/plugin_market.py", "app/extensions/plugin_repository.py", "app/infrastructure/http.py", "app/integrations/rss.py", "app/security/two_factor.py", "app/infrastructure/gc.py", "app/infrastructure/web.py", "app/security/url_safety.py", "app/domain/mediaserver.py", "app/domain/nfo.py", "app/domain/string.py", "app/log.py", "app/foundation/diagnostics.py", "app/infrastructure/log.py", "app/startup/diagnostics_initializer.py", "app/startup/log_initializer.py", "app/messaging/notification.py", "app/messaging/webpush.py", "app/foundation/jieba.py", "app/foundation/module.py", "app/foundation/object.py", "app/foundation/structures.py", "app/foundation/zhconv.py", "app/runtime/runtime.py", "app/runtime/dependencies.py", "app/runtime/native_dependencies.py", "app/agent/runtime_loader.py", "app/agent/llm/server_tools.py", "app/agent/middleware/patch_tool_calls.py", "app/agent/middleware/runtime_config.py", "app/agent/middleware/tool_selection.py", "app/agent/policy/secret_fields.py", "app/agent/prompt/transfer_redo.py", "app/api/openai_utils.py", "app/api/router_specs.py", "app/adapters/external/wechat_crypt.py", "app/modules/_base/media_auxiliary.py", "app/modules/indexer/parser/ipt_project.py", "app/modules/musicbrainz/music_cache.py", "app/modules/themoviedb/tmdb_cache.py", "app/modules/thetvdb/tvdb_v4_official.py", "app/runtime/compat/resource_imports.py", "app/runtime/extensions/host_module_adapter.py", "app/runtime/extensions/module_manager.py", "app/runtime/extensions/plugin_manager.py", "app/runtime/extensions/service_config.py", "app/testing/network_guard.py", "app/chain/media.py", "app/adapters/network/rss.py", "app/adapters/network/sites.pyi", "app/application/plugins.py", "app/application/subscribe.py", "app/application/torrent.py", "app/application/torrent_cache.py", "app/application/chain/durable_events.py", "app/application/transfer.py", "app/application/transfer_execution.py", "app/db/adapters/transfer.py", "app/db/adapters/transfer_execution.py", "app/runtime/extensions/managed_resource_adapter.py", "app/runtime/managed_resources.py", "app/startup/agent_initializer.py", "app/startup/cache_initializer.py", "app/startup/chain_events.py", "app/startup/command_initializer.py", "app/startup/configuration.py", "app/startup/context.py", "app/startup/database.py", "app/startup/database_initializer.py", "app/startup/domain_initializer.py", "app/startup/download_failure.py", "app/startup/managed_resources_initializer.py", "app/startup/initializers/managed_resources.py", "app/startup/modules_initializer.py", "app/startup/monitor_initializer.py", "app/startup/outbox.py", "app/startup/plugins_initializer.py", "app/startup/routers_initializer.py", "app/startup/scheduler_initializer.py", "app/startup/site.py", "app/startup/subscription.py", "app/startup/transaction.py", "app/startup/transfer_initializer.py", "app/startup/workflow.py", "app/startup/workflow_initializer.py", ) HOST_MODULE_PACKAGE_EXPORTS = { "filemanager": {"FileManagerModule"}, "qqbot": {"QQBotModule"}, "telegram": {"TelegramModule"}, "trimemedia": {"TrimeMediaModule"}, "ugreen": {"UgreenModule"}, } PLUGIN_COMPONENT_ROOTS = ( "app/adapters/external/plugin", "app/adapters/system/plugin", "app/application/plugin", "app/runtime/extensions/plugin", ) PLUGIN_LEGACY_ABI_NAMES = { "MoviePilotServerHelper", "PluginHelper", "PluginManager", } FORBIDDEN_IMPORT_PREFIXES = { "app.foundation": ( "app.adapters", "app.application", "app.db", "app.domain", "app.runtime", "app.sdk", ), "app.domain": ( "app.adapters", "app.application", "app.db", "app.runtime", "app.sdk", ), "app.adapters": ( "app.application", "app.runtime.compat", "app.runtime.extensions", "app.sdk", ), "app.runtime": ( "app.adapters", "app.application", "app.sdk", ), "app.application": ( "app.runtime.extensions", "app.runtime.compat", "app.sdk", ), "app.api": ( "app.runtime.extensions.plugin.manager", "app.runtime.extensions.module.manager", "app.scheduler", ), "app.agent": ( "app.runtime.extensions.plugin.manager", "app.runtime.extensions.module.manager", ), "app.chain": ( "app.runtime.extensions.plugin.manager", "app.runtime.extensions.module.manager", "app.runtime.extensions.module.dispatcher", ), "app.workflow": ( "app.runtime.extensions.plugin.manager", "app.runtime.extensions.module.manager", ), } def _legacy_imports(path: Path) -> set[str]: """ 提取源码中的静态和常量动态旧路径导入。 """ tree = _parse_source(path) imports: set[str] = set() for node in _walk_module(tree): candidates: list[str] = [] if isinstance(node, ast.Import): candidates.extend(alias.name for alias in node.names) elif isinstance(node, ast.ImportFrom) and node.module: candidates.append(node.module) elif isinstance(node, ast.Call) and node.args: argument = node.args[0] if isinstance(argument, ast.Constant) and isinstance(argument.value, str): candidates.append(argument.value) imports.update( candidate for candidate in candidates if candidate in LEGACY_MODULES or candidate.startswith(LEGACY_ROOTS) ) return imports def _attribute_parts(node: ast.Attribute) -> list[str]: """ 将静态属性访问还原为从根名称开始的完整路径片段。 """ parts = [node.attr] value = node.value while isinstance(value, ast.Attribute): parts.append(value.attr) value = value.value if not isinstance(value, ast.Name): return [] parts.append(value.id) return list(reversed(parts)) def _compat_symbol_references(tree: ast.AST) -> set[tuple[int, str]]: """ 收集显式导入或静态属性访问命中的兼容符号。 """ compatibility_symbols = { (module_name, symbol_name) for module_name, symbols in SYMBOL_ALIASES.items() for symbol_name in symbols } module_bindings: dict[str, str] = {} references: set[tuple[int, str]] = set() 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] module_bindings[binding] = alias.name if alias.asname else binding elif isinstance(node, ast.ImportFrom) and node.module: for alias in node.names: if (node.module, alias.name) in compatibility_symbols: references.add((node.lineno, f"{node.module}.{alias.name}")) continue binding = alias.asname or alias.name module_bindings[binding] = f"{node.module}.{alias.name}" for node in _walk_module(tree): if not isinstance(node, ast.Attribute): continue parts = _attribute_parts(node) if len(parts) < 2: continue resolved_root = module_bindings.get(parts[0], parts[0]) resolved = [*resolved_root.split("."), *parts[1:]] module_name = ".".join(resolved[:-1]) symbol_name = resolved[-1] if (module_name, symbol_name) in compatibility_symbols: references.add((node.lineno, f"{module_name}.{symbol_name}")) return references def _class_annotations(path: Path, class_name: str) -> dict[str, str]: """ 返回指定类的源码级字段注解。 """ 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) for node in class_node.body if isinstance(node, ast.AnnAssign) and isinstance(node.target, ast.Name) } def test_legacy_roots_contain_no_python_sources(): """ 旧目录只能作为运行时虚拟包存在，仓库中不得重新出现源码。 """ leftovers = sorted( str(path.relative_to(PROJECT_ROOT)) for root_name in ("core", "helper", "utils") for path in (APP_ROOT / root_name).rglob("*.py") ) assert leftovers == [] def test_legacy_source_directories_do_not_exist(): """ core/helper/utils 物理目录应完全退役，旧导入只由虚拟兼容包解析。 """ leftovers = [root_name for root_name in ("core", "helper", "utils") if (APP_ROOT / root_name).exists()] assert leftovers == [] def test_retired_canonical_filenames_do_not_return(): """ 能力包应使用包内语境明确的短文件名，避免再次出现冗余角色后缀。 """ leftovers = [relative_path for relative_path in RETIRED_CANONICAL_FILES if (PROJECT_ROOT / relative_path).exists()] assert leftovers == [] def test_agent_manager_uses_focused_owner_modules_and_precise_facade(): """ AgentManager 必须保持薄门面，编排器不得重新聚合 Manager 实现。 """ owner_paths = { "manager": APP_ROOT / "agent" / "manager.py", "session": APP_ROOT / "agent" / "session.py", "lifecycle": APP_ROOT / "agent" / "lifecycle.py", "tasks": APP_ROOT / "agent" / "tasks.py", "orchestrator": APP_ROOT / "agent" / "orchestrator.py", } assert all(path.is_file() for path in owner_paths.values()) manager_tree = ast.parse(owner_paths["manager"].read_text(encoding="utf-8-sig")) manager_class = next( node for node in manager_tree.body if isinstance(node, ast.ClassDef) and node.name == "AgentManager" ) manager_methods = { node.name for node in manager_class.body if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)) } assert manager_methods == {"__init__"} assert [ast.unparse(base) for base in manager_class.bases] == ["AgentTaskOwner"] orchestrator_tree = ast.parse(owner_paths["orchestrator"].read_text(encoding="utf-8-sig")) orchestrator_classes = {node.name for node in orchestrator_tree.body if isinstance(node, ast.ClassDef)} orchestrator_assignments = { target.id for node in orchestrator_tree.body if isinstance(node, ast.Assign) for target in node.targets if isinstance(target, ast.Name) } assert "AgentManager" not in orchestrator_classes assert "_MessageTask" not in orchestrator_classes assert "agent_manager" not in orchestrator_assignments startup_source = (APP_ROOT / "startup" / "initializers" / "agent.py").read_text(encoding="utf-8-sig") assert "from app.agent.manager import AgentManager" in startup_source assert "from app.agent.orchestrator import AgentManager" not in startup_source def test_agent_graph_resources_have_one_owner() -> None: """ 图资源只归 graph 所有，执行对象不得重新建立平行资源集合或私有兼容导出。 """ graph_path = APP_ROOT / "agent" / "graph.py" assert graph_path.is_file() graph_tree = _parse_source(graph_path) assert {node.name for node in graph_tree.body if isinstance(node, ast.ClassDef)} == { "GraphResources", "CompiledAgentBundle", "AgentGraphCache", } orchestrator_path = APP_ROOT / "agent" / "orchestrator.py" orchestrator = orchestrator_path.read_text(encoding="utf-8") for retired in ("_compiled_agent_bundle", "_retired_llm_models", "self._subagent_middlewares", "def _cache_agent", "def _release_llm_models", "class _CompiledAgentBundle"): assert retired not in orchestrator assert "from app.agent.graph import AgentGraphCache, CompiledAgentBundle" in orchestrator assert not any( isinstance(node, ast.ImportFrom) and node.module == "app.agent.orchestrator" for node in ast.walk(graph_tree) ) assert "app/agent/graph.py" in (PROJECT_ROOT / "docs/rules/05-architecture.md").read_text(encoding="utf-8") def test_mcp_schema_projection_has_one_pure_owner() -> None: """ MCP schema 投影使用单词 owner，且不引入运行时或端点依赖。 """ owner = APP_ROOT / "agent" / "tools" / "projection.py" assert owner.is_file() assert not (owner.parent / "schema_budget.py").exists() imports = [ node.module or "" for node in ast.walk(_parse_source(owner)) if isinstance(node, ast.ImportFrom) ] assert not any(name == "app" or name.startswith("app.") for name in imports) def test_runtime_dependencies_use_same_named_single_word_package() -> None: """ 运行依赖能力必须使用同名包，且宿主直接依赖单一职责子模块。 """ package = APP_ROOT / "runtime" / "dependencies" assert {path.name for path in package.glob("*.py")} == { "__init__.py", "native.py", "profile.py", } init_tree = ast.parse( (package / "__init__.py").read_text(encoding="utf-8"), filename=str(package / "__init__.py"), ) assert ast.get_docstring(init_tree) assert len(init_tree.body) == 1 violations: list[str] = [] for path in APP_ROOT.rglob("*.py"): relative = path.relative_to(APP_ROOT) if relative.parts[0] == "plugins": continue 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): violations.extend( f"{relative}:{node.lineno}:package-root" for alias in node.names if alias.name == "app.runtime.dependencies" ) assert violations == [] def test_domain_classification_is_a_pure_direct_import_package() -> None: """ 分类领域包根仅承载文档，且实现不得依赖配置、持久化或具体来源。 """ package = APP_ROOT / "domain" / "classification" assert {path.name for path in package.glob("*.py")} == { "__init__.py", "conditions.py", "evaluator.py", "facts.py", "fields.py", "sources.py", "validation.py", "vocabulary.py", } init_path = package / "__init__.py" init_tree = ast.parse(init_path.read_text(encoding="utf-8"), filename=str(init_path)) assert ast.get_docstring(init_tree) assert len(init_tree.body) == 1 forbidden_prefixes = ( "app.adapters", "app.application", "app.core.config", "app.db", "app.modules", "app.runtime", "app.sdk", ) violations: list[str] = [] for path in package.glob("*.py"): 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) elif isinstance(node, ast.Import): imported_modules.extend(alias.name for alias in node.names) violations.extend( f"{path.name}:{node.lineno}:{module}" for module in imported_modules if module.startswith(forbidden_prefixes) ) assert violations == [] root_imports: list[str] = [] for path in APP_ROOT.rglob("*.py"): if path.is_relative_to(APP_ROOT / "plugins"): continue 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): root_imports.extend( f"{path.relative_to(PROJECT_ROOT).as_posix()}:{node.lineno}" for alias in node.names if alias.name == "app.domain.classification" ) assert root_imports == [] def test_application_classification_uses_same_named_package() -> None: """ 分类应用能力必须归入同名包，包根不得重复导出内部实现。 """ package = APP_ROOT / "application" / "classification" assert {path.name for path in package.glob("*.py")} == { "__init__.py", "analysis.py", "catalog.py", "configuration.py", "contract.py", "enrichment.py", "execution.py", "legacy.py", "migration.py", "compiler.py", "projection.py", "reference.py", "runtime.py", } init_path = package / "__init__.py" init_tree = ast.parse(init_path.read_text(encoding="utf-8"), filename=str(init_path)) assert ast.get_docstring(init_tree) assert len(init_tree.body) == 1 def test_site_application_owners_use_single_word_module_names() -> None: """ 站点流控等用例必须留在站点包中，新增子模块保持单词命名。 """ package = APP_ROOT / "application" / "site" assert (package / "throttle.py").is_file() assert all(path.stem.isalpha() for path in package.glob("*.py") if path.stem != "__init__") assert not (APP_ROOT / "application" / "site_throttle.py").exists() def test_application_torrent_uses_same_named_single_word_package() -> None: """ 种子应用能力必须归入同名包，包根不得重复导出内部实现。 """ package = APP_ROOT / "application" / "torrent" assert {path.name for path in package.glob("*.py")} == { "__init__.py", "cache.py", "download.py", } init_tree = ast.parse( (package / "__init__.py").read_text(encoding="utf-8"), filename=str(package / "__init__.py"), ) assert ast.get_docstring(init_tree) assert len(init_tree.body) == 1 root_imports: list[str] = [] for root in (APP_ROOT, PROJECT_ROOT / "database"): for path in root.rglob("*.py"): if path.is_relative_to(APP_ROOT / "plugins"): continue 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 == [] def test_host_module_package_roots_only_export_capability_entrypoints() -> None: """ 宿主模块包根只公开 capability entrypoint，内部 owner 必须直达子模块。 """ base_path = APP_ROOT / "modules" / "_base" / "__init__.py" base_tree = ast.parse(base_path.read_text(encoding="utf-8"), filename=str(base_path)) assert ast.get_docstring(base_tree) assert len(base_tree.body) == 1 root_imports: list[str] = [] for path in (APP_ROOT / "modules").rglob("*.py"): if path.is_relative_to(APP_ROOT / "plugins"): continue 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 = _parse_source(path) assignments = { target.id: ast.literal_eval(node.value) for node in tree.body if isinstance(node, ast.Assign) for target in node.targets if isinstance(target, ast.Name) and target.id in {"_EXPORTS", "__all__"} } assert set(assignments["_EXPORTS"]) == expected_exports assert set(assignments["__all__"]) == expected_exports def test_domain_media_projection_uses_single_word_owner_package() -> None: """ 媒体来源投影必须由单词 owner 完整承接，canonical 模型只保留薄委托。 """ package = APP_ROOT / "domain" / "projection" assert {path.name for path in package.glob("*.py")} == { "__init__.py", "anilist.py", "bangumi.py", "douban.py", "mapping.py", "tmdb.py", } init_tree = ast.parse( (package / "__init__.py").read_text(encoding="utf-8"), filename=str(package / "__init__.py"), ) assert ast.get_docstring(init_tree) assert len(init_tree.body) == 1 context_path = APP_ROOT / "domain" / "context.py" context_tree = ast.parse( context_path.read_text(encoding="utf-8-sig"), filename=str(context_path), ) media_class = next( node for node in context_tree.body if isinstance(node, ast.ClassDef) and node.name == "MediaInfo" ) methods = { node.name: node for node in media_class.body if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)) } for name in ( "set_tmdb_info", "set_douban_info", "set_bangumi_info", "set_anilist_info", ): method = methods[name] executable = [ node for node in method.body if not ( isinstance(node, ast.Expr) and isinstance(node.value, ast.Constant) and isinstance(node.value.value, str) ) ] assert len(executable) == 1 assert isinstance(executable[0], ast.Expr) assert isinstance(executable[0].value, ast.Call) schema_source = (APP_ROOT / "schemas" / "context.py").read_text(encoding="utf-8-sig") assert "app.domain.projection" not in schema_source assert not any(name in schema_source for name in methods if name.startswith("set_") and name.endswith("_info")) forbidden_calls: list[str] = [] for path in APP_ROOT.rglob("*.py"): relative = path.relative_to(APP_ROOT) if relative.parts[0] == "plugins" or path == context_path: continue 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 { "get_bangumi_media_type", "get_anilist_media_type", "_anilist_date", "_anilist_chinese_title", }: forbidden_calls.append(f"{relative}:{node.lineno}:{node.attr}") assert forbidden_calls == [] def test_workflow_query_contract_returns_only_typed_snapshots(): """ 工作流正式查询端口不得退化为 Any 或 ORM 返回值。 """ path = APP_ROOT / "application" / "workflow.py" tree = _parse_source(path) query_classes = { node.name: node for node in tree.body if isinstance(node, ast.ClassDef) and node.name in {"WorkflowQueryRepository", "WorkflowQueryService"} } methods = [ node for query_class in query_classes.values() for node in query_class.body if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)) and not node.name.startswith("__") and node.returns is not None ] assert set(query_classes) == {"WorkflowQueryRepository", "WorkflowQueryService"} assert methods for method in methods: annotation = ast.unparse(method.returns) assert "Any" not in annotation if method.name in {"count", "async_count"}: assert annotation == "int" else: assert "WorkflowSnapshot" in annotation def test_workflow_query_consumers_do_not_reach_raw_oper(): """ API、Agent、共享服务和运行时管理器只消费统一快照查询服务。 """ consumer_paths = ( "app/api/dependencies/workflow.py", "app/application/server/share.py", "app/workflow/__init__.py", ) violations = {} for relative_path in consumer_paths: source = (PROJECT_ROOT / relative_path).read_text(encoding="utf-8") forbidden = { name for name in ( "WorkflowOper", "get_agent_workflow_port", "get_chain_workflow_port", ) if name in source } if forbidden: violations[relative_path] = sorted(forbidden) assert violations == {} def test_workflow_query_adapter_owns_projection_sessions(): """ 唯一查询适配器必须在自有同步和异步 Session 内投影快照。 """ path = APP_ROOT / "db" / "adapters" / "workflow.py" source = path.read_text(encoding="utf-8") assert "class TransactionalWorkflowQueryRepository" in source assert "session.close()" in source assert "async with self._async_session() as session" in source assert "_project_workflow(record)" in source def test_workflow_execution_chain_uses_single_application_owned_port(): """ 工作流 Chain 写端必须只使用 Application owner 的唯一配置入口。 """ contract_path = APP_ROOT / "application" / "workflow.py" contract_tree = ast.parse( contract_path.read_text(encoding="utf-8"), filename=str(contract_path), ) contract = next( node for node in contract_tree.body if isinstance(node, ast.ClassDef) and node.name == "WorkflowExecutionPort" ) methods = { node.name: ast.unparse(node.returns) for node in contract.body if isinstance(node, ast.FunctionDef) and node.returns is not None } assert methods == { "start": "bool", "success": "bool", "fail": "bool", "step": "bool", "reset": "bool", } chain_source = (APP_ROOT / "chain" / "workflow.py").read_text(encoding="utf-8") startup_source = (APP_ROOT / "startup" / "initializers" / "modules.py").read_text(encoding="utf-8") composition_source = (APP_ROOT / "startup" / "composition" / "database.py").read_text(encoding="utf-8") assert chain_source.count("get_configured_workflow_execution()") == 1 assert "get_chain_workflow_port" not in chain_source assert "configure_workflow_execution(TransactionalWorkflowExecutionService(SessionFactory))" in composition_source assert "configure_workflow_execution_composition()" in startup_source assert "workflow=lambda:" not in startup_source def test_chain_runtime_context_owns_typed_repository_instances(): """ Chain 数据能力必须作为明确类型的实例字段进入运行时上下文。 """ annotations = _class_annotations( APP_ROOT / "application" / "chain" / "context.py", "ChainRuntimeContext", ) expected = { "site_repository": "SiteRepository", "subscription_repository": "SubscriptionRepository", "download_history_repository": "DownloadHistoryRepository", "transfer_history_repository": "TransferHistoryRepository", "transfer_admission_repository": "TransferAdmissionRepository", "transfer_execution_repository": "TransferExecutionRepository", "media_server_repository": "MediaServerRepository", "download_failure_repository": "DownloadFailureRepository", "user_repository": "ChainUserRepository", } assert {name: annotations[name] for name in expected} == expected chain_base_source = (APP_ROOT / "chain" / "base.py").read_text(encoding="utf-8-sig") for field_name in expected: assert f"self.{field_name} = context.{field_name}" in chain_base_source download_source = "\n".join( path.read_text(encoding="utf-8") for path in sorted((APP_ROOT / "chain" / "download").glob("*.py")) ) mediaserver_source = (APP_ROOT / "chain" / "mediaserver.py").read_text(encoding="utf-8") application_source = (APP_ROOT / "application" / "mediaserver.py").read_text(encoding="utf-8") startup_source = (APP_ROOT / "startup" / "composition" / "chain.py").read_text(encoding="utf-8") assert "DownloadFailure = Any" not in download_source assert "dboper" not in mediaserver_source assert "async def async_get_item_id(" in application_source assert "TransactionalMediaServerRepository(SessionFactory)" in startup_source def test_startup_composes_typed_chain_and_agent_data_contexts(): """ 组合根必须一次性构造 Chain 与 Agent 的类型化数据上下文。 """ context_specs = { "ChainRuntimeContext": ( APP_ROOT / "application" / "chain" / "context.py", APP_ROOT / "startup" / "composition" / "chain.py", { "site_repository", "subscription_repository", "subscription_mutation_scope", "sync_subscription_mutation_scope", "subscription_delete_scope", "sync_subscription_delete_scope", "subscription_completion_scope", "download_history_repository", "transfer_history_repository", "transfer_admission_repository", "transfer_execution_repository", "media_server_repository", "download_failure_repository", "user_repository", }, ), "AgentDataContext": ( APP_ROOT / "application" / "agent.py", APP_ROOT / "startup" / "composition" / "agent.py", { "chat", "chat_persistence", "tasks", "users", "sites", "subscriptions", "subscription_mutation_scope", "subscription_delete_scope", "subscription_history", "transfer_history", "transfer_execution", "download_history", "plugin_data", }, ), } for class_name, (context_path, composition_path, expected_fields) in context_specs.items(): annotations = _class_annotations(context_path, class_name) assert expected_fields <= annotations.keys() assert all( "Any" not in annotations[field_name] and "Oper" not in annotations[field_name] and "Callable" not in annotations[field_name] for field_name in expected_fields ) composition_tree = ast.parse( composition_path.read_text(encoding="utf-8-sig"), filename=str(composition_path), ) constructors = [ node for node in ast.walk(composition_tree) if isinstance(node, ast.Call) and isinstance(node.func, ast.Name) and node.func.id == class_name ] assert len(constructors) == 1 values = {keyword.arg: keyword.value for keyword in constructors[0].keywords if keyword.arg is not None} assert expected_fields <= values.keys() assert all(not isinstance(values[field_name], ast.Lambda) for field_name in expected_fields) def test_download_history_ports_are_typed_detached_and_canonically_injected(): """ 下载历史宿主调用面只能消费冻结快照和显式事务 adapter。 """ history_path = APP_ROOT / "application" / "history" / "__init__.py" history_tree = ast.parse( history_path.read_text(encoding="utf-8"), filename=str(history_path), ) classes = {node.name: node for node in history_tree.body if isinstance(node, ast.ClassDef)} for class_name in ( "DownloadHistorySnapshot", "DownloadFileSnapshot", "DownloadHistoryWrite", "DownloadFileWrite", ): decorator = next( item for item in classes[class_name].decorator_list if isinstance(item, ast.Call) and isinstance(item.func, ast.Name) and item.func.id == "dataclass" ) keywords = {item.arg: ast.literal_eval(item.value) for item in decorator.keywords} assert keywords == {"frozen": True, "slots": True} for class_name in ("DownloadHistoryQueryPort", "DownloadHistoryWritePort"): annotations = [ ast.unparse(node.returns) for node in classes[class_name].body if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)) and node.returns is not None ] assert annotations assert all("Any" not in annotation for annotation in annotations) chain_annotations = _class_annotations( APP_ROOT / "application" / "chain" / "context.py", "ChainRuntimeContext", ) agent_annotations = _class_annotations( APP_ROOT / "application" / "agent.py", "AgentDataContext", ) assert chain_annotations["download_history_repository"] == "DownloadHistoryRepository" assert agent_annotations["download_history"] == "DownloadHistoryRepository" consumer_paths = ( *sorted((APP_ROOT / "chain" / "download").glob("*.py")), *sorted((APP_ROOT / "chain" / "transfer").glob("*.py")), APP_ROOT / "application" / "transfer" / "workflow.py", ) for path in consumer_paths: source = path.read_text(encoding="utf-8") assert "app.db.oper.downloadhistory" not in source assert "DownloadHistory = Any" not in source assert "DownloadFiles = Any" not in source startup_source = (APP_ROOT / "startup" / "initializers" / "modules.py").read_text(encoding="utf-8") runtime_source = (APP_ROOT / "startup" / "composition" / "runtime.py").read_text(encoding="utf-8") assert "TransactionalDownloadHistoryRepository(" not in startup_source assert "TransactionalDownloadHistoryRepository(" in runtime_source assert "SessionDownloadHistoryRepository" not in startup_source assert "SessionDownloadHistoryRepository" in runtime_source assert "DownloadHistoryOper" not in startup_source adapter_source = (APP_ROOT / "db" / "adapters" / "history" / "download.py").read_text(encoding="utf-8") assert "class TransactionalDownloadHistoryRepository" in adapter_source assert "class SessionDownloadHistoryRepository" in adapter_source assert "_project_history" in adapter_source assert "SqlAlchemyUnitOfWork" in adapter_source legacy_source = (APP_ROOT / "sdk" / "_legacy" / "transfer.py").read_text(encoding="utf-8") assert "download_history: Optional[Any]" in legacy_source def test_user_configuration_uses_typed_transactional_adapter(): """ 用户配置宿主入口只消费类型化端口，旧 Oper 写入口仅承担兼容 ABI。 """ application_path = APP_ROOT / "application" / "security" / "userconfig.py" application_source = application_path.read_text(encoding="utf-8") application_tree = ast.parse(application_source, filename=str(application_path)) repository = next( node for node in application_tree.body if isinstance(node, ast.ClassDef) and node.name == "UserConfigurationRepository" ) returns = { node.name: ast.unparse(node.returns) for node in repository.body if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)) and node.returns is not None } assert returns == { "get": "JsonData", "set": "None", "publish_rename": "None", "publish_delete": "None", } assert "Any" not in application_source composition_source = (APP_ROOT / "startup" / "composition" / "configuration.py").read_text(encoding="utf-8") assert "TransactionalUserConfigurationRepository(SessionFactory)" in composition_source assert "UserConfigOper" not in composition_source adapter_path = APP_ROOT / "db" / "adapters" / "configuration.py" adapter_source = adapter_path.read_text(encoding="utf-8") assert "class TransactionalUserConfigurationRepository" in adapter_source assert "SqlAlchemyUnitOfWork" in adapter_source assert "Any" not in adapter_source oper_source = (APP_ROOT / "db" / "oper" / "userconfig.py").read_text(encoding="utf-8") assert "def stage_set(" in oper_source assert "def set(" in oper_source assert "Any" not in oper_source def test_modules_initializer_delegates_configuration_and_database_composition(): """ 模块初始化器只编排配置与数据库组合 API，不再内联其构造细节。 """ initializer_source = (APP_ROOT / "startup" / "initializers" / "modules.py").read_text(encoding="utf-8") configuration_source = (APP_ROOT / "startup" / "composition" / "configuration.py").read_text(encoding="utf-8") database_source = (APP_ROOT / "startup" / "composition" / "database.py").read_text(encoding="utf-8") runtime_source = (APP_ROOT / "startup" / "composition" / "runtime.py").read_text(encoding="utf-8") for call in ( "start_database_runtime()", "stop_database_runtime", "compose_configuration(", "compose_database_services(", "publish_database_services(", "publish_configuration(", "reset_database_services", "reset_configuration", "configure_database()", ): assert call in initializer_source for detail in ( "DatabaseWorker()", "TransactionalWriteRunner(", "PluginPersistenceService(", "SqlAlchemyDataQueryAdapter(", "RuntimeConfiguration(", "RuntimeSettingsService(", "TransactionalUserConfigurationRepository(", ): assert detail not in initializer_source assert "stop_database_worker" not in initializer_source assert "RuntimeConfiguration(" in configuration_source assert "RuntimeSettingsService(" in configuration_source assert "TransactionalUserConfigurationRepository(" in configuration_source assert "DatabaseWorker()" in database_source assert "TransactionalWriteRunner(" in database_source assert "PluginPersistenceService(" in database_source assert "SqlAlchemyDataQueryAdapter(" in database_source assert "reset_transaction_runners()" in database_source assert "configure_api_data_runtime" not in initializer_source assert "configure_api_data_runtime" not in database_source assert "configure_api_data_runtime" in runtime_source def test_modules_initializer_delegates_network_composition(): """ 模块初始化器只调用网络组合 API，不再内联具体 Adapter 组装。 """ initializer_source = (APP_ROOT / "startup" / "initializers" / "modules.py").read_text(encoding="utf-8") network_source = (APP_ROOT / "startup" / "composition" / "network.py").read_text(encoding="utf-8") assert "configure_application_network_ports()" in initializer_source for detail in ( "_NetworkTestTransportAdapter", "_ImageTransportAdapter", "_InternalAddressAdapter", "_MessageIngressAdapter", "NetworkTestService(", "configure_image_ports(", "configure_message_ingress_port(", ): assert detail not in initializer_source assert detail in network_source def test_modules_initializer_delegates_outbox_composition(): """ 模块初始化器只发布 Outbox dispatcher，不再内联 handler 与存储装配。 """ initializer_source = (APP_ROOT / "startup" / "initializers" / "modules.py").read_text(encoding="utf-8") composition_source = (APP_ROOT / "startup" / "composition" / "outbox.py").read_text(encoding="utf-8") package_source = (APP_ROOT / "startup" / "composition" / "__init__.py").read_text(encoding="utf-8") assert "configure_outbox_dispatcher(build_outbox_dispatcher, submit=ThreadHelper().submit)" in initializer_source for retired_name in ("_build_outbox_handlers", "_build_outbox_dispatcher"): assert retired_name not in initializer_source for detail in ( "def build_outbox_handlers()", "def build_outbox_dispatcher()", "validate_durable_event_handlers(handlers)", "SqlAlchemyOutboxDispatchStore(SessionFactory)", ): assert detail in composition_source assert "build_outbox_handlers" not in package_source assert "build_outbox_dispatcher" not in package_source def test_modules_initializer_delegates_server_service_composition(): """ 中心服务对象图只由单词型 composition owner 构造，initializer 仅保留调用顺序。 """ initializer_source = (APP_ROOT / "startup" / "initializers" / "modules.py").read_text(encoding="utf-8") composition_path = APP_ROOT / "startup" / "composition" / "server.py" composition_source = composition_path.read_text(encoding="utf-8") architecture_source = (PROJECT_ROOT / "docs" / "rules" / "05-architecture.md").read_text(encoding="utf-8") package_tree = ast.parse((APP_ROOT / "startup" / "composition" / "__init__.py").read_text(encoding="utf-8")) assert composition_path.is_file() assert not list(composition_path.parent.glob("server_*.py")) assert "from app.startup.composition.server import configure_server_services" in (initializer_source) assert "configure_server_services(workflow_query, runtime_dependencies.subscription)" in initializer_source assert "`startup/composition/server.py` owns MoviePilot Server" in (architecture_source) for detail in ( "ServerReportService(", "ServerSharingService(", "configure_server_application_services(", ): assert detail not in initializer_source assert detail in composition_source assert not any( isinstance(node, (ast.Import, ast.ImportFrom, ast.FunctionDef, ast.ClassDef)) for node in package_tree.body ) def test_modules_initializer_delegates_agent_composition(): """ 模块初始化器只编排 Agent composition，不再内联数据与任务构造。 """ initializer_source = (APP_ROOT / "startup" / "initializers" / "modules.py").read_text(encoding="utf-8") agent_source = (APP_ROOT / "startup" / "composition" / "agent.py").read_text(encoding="utf-8") runtime_source = (APP_ROOT / "startup" / "composition" / "runtime.py").read_text(encoding="utf-8") assert "compose_agent(" in initializer_source assert "publish_agent_services(" in initializer_source assert initializer_source.index("await start_database_runtime()") < initializer_source.index("compose_agent(") for detail in ( "AgentDataContext(", "AgentChatPersistenceService(", "AgentChatService(", "TransactionalAgentTaskRepository(", "AgentTaskExecutionService(", "configure_agent_chat_service(", "configure_agent_chat_persistence(", "configure_agent_task_execution(", ): assert detail not in initializer_source assert detail in agent_source assert "AgentChatRuntime(" not in initializer_source assert "AgentChatRuntime(" not in agent_source assert "AgentChatRuntime(" in runtime_source assert "runtime.worker.snapshot().capacity" in agent_source assert "startup.initializers" not in agent_source def test_runtime_composition_is_the_sole_host_and_domain_runtime_owner(): """ HostRuntime、领域 Runtime 与 ApiData 投影只能由单词型 runtime owner 构造。 """ initializer_source = (APP_ROOT / "startup" / "initializers" / "modules.py").read_text(encoding="utf-8") runtime_path = APP_ROOT / "startup" / "composition" / "runtime.py" runtime_source = runtime_path.read_text(encoding="utf-8") database_source = (APP_ROOT / "startup" / "composition" / "database.py").read_text(encoding="utf-8") package_tree = ast.parse((APP_ROOT / "startup" / "composition" / "__init__.py").read_text(encoding="utf-8")) assert runtime_path.is_file() assert not list(runtime_path.parent.glob("runtime_*.py")) assert "compose_runtime_dependencies()" in initializer_source assert "compose_runtime(" in initializer_source assert "RuntimeInputs(" in initializer_source assert "publish_runtime(runtime_composition)" in initializer_source assert "HostRuntime(" not in initializer_source assert "ApiDataPorts(" not in initializer_source assert "ApiDataPorts(" not in database_source for detail in ( "class RuntimeDependencies:", "class RuntimeInputs:", "class RuntimeComposition:", "HostRuntime(", "AgentChatRuntime(", "AuthenticationRuntime(", "PersistenceRuntime(", "MessagingRuntime(", "HistoryRuntime(", "SiteRuntime(", "SubscriptionRuntime(", "WorkflowRuntime(", "ApiDataPorts(", ): assert detail in runtime_source constructors = [] for path in APP_ROOT.rglob("*.py"): if path.is_relative_to(APP_ROOT / "plugins"): continue tree = _parse_source(path) if any( isinstance(node, ast.Call) and isinstance(node.func, ast.Name) and node.func.id == "HostRuntime" for node in _walk_module(tree) ): constructors.append(path.relative_to(APP_ROOT).as_posix()) assert constructors == ["startup/composition/runtime.py"] context_tree = ast.parse((APP_ROOT / "startup" / "composition" / "context.py").read_text(encoding="utf-8")) host_runtime = next( node for node in context_tree.body if isinstance(node, ast.ClassDef) and node.name == "HostRuntime" ) task_field = next( node for node in host_runtime.body if isinstance(node, ast.AnnAssign) and isinstance(node.target, ast.Name) and node.target.id == "tasks" ) assert task_field.value is None assert ( _class_annotations( APP_ROOT / "startup" / "composition" / "context.py", "AuthenticationRuntime", )["passkey_repository"] == "RepositoryFactory" ) assert not any( isinstance(node, (ast.Import, ast.ImportFrom, ast.FunctionDef, ast.ClassDef)) for node in package_tree.body ) def test_modules_initializer_delegates_chain_composition(): """ Chain 上下文、壁纸与旧整理命令只由单词型 composition owner 装配。 """ initializer_source = (APP_ROOT / "startup" / "initializers" / "modules.py").read_text(encoding="utf-8") composition_path = APP_ROOT / "startup" / "composition" / "chain.py" composition_source = composition_path.read_text(encoding="utf-8") composition_tree = ast.parse(composition_source, filename=str(composition_path)) package_tree = ast.parse((APP_ROOT / "startup" / "composition" / "__init__.py").read_text(encoding="utf-8")) architecture_source = (PROJECT_ROOT / "docs" / "rules" / "05-architecture.md").read_text(encoding="utf-8") runtime_source = (APP_ROOT / "startup" / "composition" / "runtime.py").read_text(encoding="utf-8") assert composition_path.is_file() assert not list(composition_path.parent.glob("chain_*.py")) assert "from app.startup.composition.chain import (" in initializer_source assert initializer_source.index("configure_wallpaper_services()") < ( initializer_source.index("configure_chain_runtime_context(") ) for retired_name in ( "_execute_legacy_transfer_command", "_build_chain_runtime_context", ): assert retired_name not in initializer_source for detail in ( "ChainRuntimeContext(", "TransactionalTransferAdmissionRepository(SessionFactory)", "TransactionalChainDurableEventWriter(SessionFactory)", "def execute_legacy_transfer_command(", "def build_chain_runtime_context(", "def configure_wallpaper_services(", "def configure_chain_runtime_context(", ): assert detail not in initializer_source assert detail in composition_source assert "TransactionalTransferExecutionRepository(" not in initializer_source assert runtime_source.count("TransactionalTransferExecutionRepository(") == 1 assert "dependencies=runtime_dependencies" in initializer_source assert "transfer_execution_repository=dependencies.transfer_execution" in composition_source assert "TransactionalTransferExecutionRepository" not in composition_source transfer_function = next( node for node in composition_tree.body if isinstance(node, ast.FunctionDef) and node.name == "execute_legacy_transfer_command" ) assert any( isinstance(node, ast.ImportFrom) and node.module == "app.chain.transfer.facade" for node in ast.walk(transfer_function) ) assert not any( isinstance(node, ast.ImportFrom) and node.module == "app.chain.transfer.facade" for node in composition_tree.body ) assert "`startup/composition/chain.py` owns" in architecture_source assert not any( isinstance(node, (ast.Import, ast.ImportFrom, ast.FunctionDef, ast.ClassDef)) for node in package_tree.body ) def test_canonical_workflow_oper_has_no_legacy_writer_or_duplicate_exports(): """ 工作流旧写入口只能存在于 SDK Legacy facade。 """ oper_path = APP_ROOT / "db" / "oper" / "workflow.py" oper_tree = ast.parse( oper_path.read_text(encoding="utf-8"), filename=str(oper_path), ) oper_class = next(node for node in oper_tree.body if isinstance(node, ast.ClassDef) and node.name == "WorkflowOper") method_names = {node.name for node in oper_class.body if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef))} assert {"start", "success", "fail", "step", "reset"}.isdisjoint(method_names) assert "legacy" not in oper_path.read_text(encoding="utf-8").lower() package_source = (APP_ROOT / "db" / "oper" / "__init__.py").read_text(encoding="utf-8") assert '"WorkflowOper"' not in package_source def test_agent_data_ports_do_not_duplicate_workflow_query_capability(): """ Agent 显式数据上下文不得重复拥有工作流查询能力。 """ annotations = _class_annotations( APP_ROOT / "application" / "agent.py", "AgentDataContext", ) assert "workflow" not in annotations assert "workflow_query" not in annotations def test_host_uses_canonical_workflow_manager_name(): """ 宿主代码不得继续定义或导入旧 WorkFlowManager 拼写。 """ violations = [] for path in APP_ROOT.rglob("*.py"): relative_path = path.relative_to(APP_ROOT) if relative_path.parts[:2] == ("runtime", "compat"): continue if "WorkFlowManager" in path.read_text(encoding="utf-8"): violations.append(str(path.relative_to(PROJECT_ROOT))) assert violations == [] def test_plugin_route_refresh_is_imported_from_its_application_owner(): """ 插件路由刷新只有 Application 一个拥有者，端点之间不得互相取用它。 """ owner = "app.application.plugin.routes" violations: dict[str, str] = {} for path in APP_ROOT.rglob("*.py"): relative = path.relative_to(APP_ROOT) if relative.parts[0] == "plugins": continue tree = _parse_source(path) for node in _walk_module(tree): if not isinstance(node, ast.ImportFrom) or node.level: continue if node.module == owner: continue if any(alias.name == "register_plugin_api" for alias in node.names): violations[str(relative)] = node.module or "" assert violations == {} def test_startup_root_contains_only_composition_packages(): """ 组合根顶层只保留稳定分区，禁止再次堆叠扁平实现文件。 """ startup_root = APP_ROOT / "startup" root_modules = sorted(path.name for path in startup_root.glob("*.py")) python_packages = sorted(path.name for path in startup_root.iterdir() if path.is_dir() and any(path.rglob("*.py"))) assert root_modules == ["__init__.py"] assert python_packages == ["composition", "initializers", "lifecycle"] def test_retired_canonical_roots_contain_no_python_sources(): """ 已收敛的新增目录不得再次以顶级 Python 包形式出现。 """ leftovers = sorted( str(path.relative_to(PROJECT_ROOT)) for root_name in RETIRED_CANONICAL_ROOTS for path in (APP_ROOT / root_name).rglob("*.py") ) assert leftovers == [] def test_host_code_does_not_import_legacy_roots(): """ 除插件和兼容层外，宿主代码必须使用 canonical 路径。 """ violations: dict[str, set[str]] = {} for path in APP_ROOT.rglob("*.py"): relative = path.relative_to(APP_ROOT) if relative.parts[0] == "plugins" or relative.parts[:2] == ( "runtime", "compat", ): continue imports = _legacy_imports(path) if imports: violations[str(relative)] = imports assert violations == {} def test_compat_symbol_scanner_covers_static_import_shapes() -> None: """ 兼容符号扫描必须覆盖显式导入、模块别名和完整属性链。 """ tree = ast.parse( """ from app.schemas import TransferTask import app.schemas as schema_alias schema_alias.TransferQueue import app.schemas app.schemas.TransferTask from app.application import transfer as transfer_package transfer_package.TransferQueue """ ) assert _compat_symbol_references(tree) == { (2, "app.schemas.TransferTask"), (4, "app.schemas.TransferQueue"), (6, "app.schemas.TransferTask"), (8, "app.application.transfer.TransferQueue"), } def test_host_code_does_not_use_compat_symbol_aliases() -> None: """ 宿主必须导入 canonical 符号，不得反向消费插件兼容覆盖。 """ violations: list[str] = [] for path in APP_ROOT.rglob("*.py"): relative = path.relative_to(APP_ROOT) if ( relative.parts[0] == "plugins" or relative.parts[:2] == ("runtime", "compat") or relative.parts[:2] == ("sdk", "_legacy") ): continue tree = _parse_source(path) violations.extend( f"{relative.as_posix()}:{line}:{symbol}" for line, symbol in sorted(_compat_symbol_references(tree)) ) assert violations == [] def test_host_code_uses_explicit_runtime_facade_getters(): """ 宿主消费者必须显式调用 getter，不得把兼容 Facade 当作新代码入口。 """ forbidden_imports = { "app.application.module": {"ModuleManager"}, "app.application.scheduling": {"Scheduler"}, } violations: list[str] = [] for path in APP_ROOT.rglob("*.py"): relative = path.relative_to(APP_ROOT) if relative.parts[0] == "plugins" or relative.parts[:2] == ( "runtime", "compat", ): continue 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": violations.append(f"{relative.as_posix()}:{node.lineno}:SystemConfigOper") continue if not isinstance(node, ast.ImportFrom) or not node.module: continue forbidden_names = forbidden_imports.get(node.module, set()) for alias in node.names: class_shaped_plugin_getter = ( node.module == "app.application.plugin.runtime" and alias.name == "get_plugin_manager" and alias.asname is not None ) class_shaped_config_getter = ( node.module == "app.application.configuration" and alias.name == "get_configured_system_config" and alias.asname is not None ) if alias.name in forbidden_names or class_shaped_plugin_getter or class_shaped_config_getter: imported_name = alias.asname or alias.name violations.append(f"{relative.as_posix()}:{node.lineno}:{imported_name}") assert violations == [] def test_user_chain_and_agent_ports_are_typed_and_orm_free(): """ 用户 Chain、Agent 与宿主模块只能消费 Application 用户端口。 """ chain_annotations = _class_annotations( APP_ROOT / "application" / "chain" / "context.py", "ChainRuntimeContext", ) agent_annotations = _class_annotations( APP_ROOT / "application" / "agent.py", "AgentDataContext", ) assert chain_annotations["user_repository"] == "ChainUserRepository" assert agent_annotations["users"] == "ChainUserRepository" production_paths = [ APP_ROOT / "chain" / "user.py", APP_ROOT / "chain" / "interaction.py", APP_ROOT / "chain" / "_messaging.py", APP_ROOT / "agent" / "orchestrator.py", APP_ROOT / "modules" / "feishu" / "feishu.py", ] violations: list[str] = [] for path in production_paths: 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": violations.append(f"{path.relative_to(PROJECT_ROOT)}:{node.lineno}:import") assert violations == [] def test_startup_injects_user_adapter_instead_of_raw_oper(): """ 启动组合根不得把无会话 UserOper 注入宿主查询调用面。 """ initializer_source = (APP_ROOT / "startup" / "initializers" / "modules.py").read_text(encoding="utf-8-sig") database_source = (APP_ROOT / "startup" / "composition" / "database.py").read_text(encoding="utf-8-sig") chain_source = (APP_ROOT / "startup" / "composition" / "chain.py").read_text(encoding="utf-8-sig") runtime_source = (APP_ROOT / "startup" / "composition" / "runtime.py").read_text(encoding="utf-8-sig") security_source = (APP_ROOT / "startup" / "composition" / "security.py").read_text(encoding="utf-8-sig") source = initializer_source + database_source + chain_source + runtime_source + security_source assert "TransactionalUserRepository" in source assert "SqlAlchemyUserRepository" in source assert "build_transactional_user_repository" not in initializer_source assert "build_transactional_user_repository" in chain_source assert "from app.db.oper.user import UserOper" not in source assert "user=lambda: UserOper()" not in source def test_transfer_pending_oper_import_is_confined_to_database_boundary(): """ 宿主仅允许事务适配器和兼容导出直接导入整理待处理 Oper。 """ allowed_paths = { "app/db/adapters/transfer/admission.py", "app/db/adapters/chain.py", "app/db/adapters/transfer/execution.py", "app/db/oper/__init__.py", } violations: list[str] = [] for path in APP_ROOT.rglob("*.py"): relative = path.relative_to(PROJECT_ROOT).as_posix() if relative.startswith("app/plugins/") or relative in allowed_paths: continue 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}") elif isinstance(node, ast.ImportFrom) and ( node.module == "app.db.oper.transferpending" or ( node.module == "app.db.oper" and any(alias.name in {"transferpending", "TransferPendingOper"} for alias in node.names) ) ): violations.append(f"{relative}:{node.lineno}") assert violations == [] def test_startup_injects_transactional_transfer_admission_repository(): """ 启动组合根必须向 Chain 注入事务型整理准入仓储。 """ path = APP_ROOT / "startup" / "composition" / "chain.py" 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 _walk_module(tree) ) chain_context_calls = [ node for node in _walk_module(tree) if isinstance(node, ast.Call) and isinstance(node.func, ast.Name) and node.func.id == "ChainRuntimeContext" ] assert imports_repository is True assert len(chain_context_calls) == 1 keywords = {keyword.arg: keyword.value for keyword in chain_context_calls[0].keywords if keyword.arg is not None} admission = keywords["transfer_admission_repository"] assert isinstance(admission, ast.Call) assert isinstance(admission.func, ast.Name) assert admission.func.id == "TransactionalTransferAdmissionRepository" def test_transfer_admission_chain_context_field_is_typed(): """ 整理准入仓储必须以明确 Protocol 注入 Chain 上下文。 """ annotations = _class_annotations( APP_ROOT / "application" / "chain" / "context.py", "ChainRuntimeContext", ) assert annotations["transfer_admission_repository"] == "TransferAdmissionRepository" def test_scheduler_does_not_depend_on_database_implementation(): """ Scheduler 只能消费应用端口，不得重新直连 app.db 实现。 """ dependencies = set().union( *( module_dependencies for module_name, module_dependencies in _build_module_graph().items() if module_name == "app.scheduler" or module_name.startswith("app.scheduler.") ) ) assert { dependency for dependency in dependencies if dependency == "app.db" or dependency.startswith("app.db.") } == set() def test_monitor_dispatcher_uses_explicit_history_port_getter(): """ 监控分发器不得把兼容 TransferHistoryPort 伪装成数据库 Oper。 """ path = APP_ROOT / "monitor" / "dispatcher.py" tree = _parse_source(path) violations = [ f"{path.relative_to(PROJECT_ROOT).as_posix()}:{node.lineno}" 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) ] assert violations == [] def test_canonical_service_config_consumers_use_application_directory(): """ Chain、API、Scheduler 与 Agent 不得绕过命名应用目录读取服务配置。 """ paths = [ APP_ROOT / "chain" / "_messaging.py", APP_ROOT / "chain" / "mediaserver.py", APP_ROOT / "api" / "endpoints" / "message.py", *(APP_ROOT / "scheduler").glob("*.py"), APP_ROOT / "agent" / "llm" / "capability.py", APP_ROOT / "agent" / "tools" / "base.py", APP_ROOT / "api" / "endpoints" / "mediaserver.py", ] violations: list[str] = [] for path in paths: 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 ( path.name == "mediaserver.py" and isinstance(node, ast.ImportFrom) and node.module == "app.application.mediaserver" and any(alias.name == "MediaServerHelper" for alias in node.names) ): violations.append(f"{path.relative_to(PROJECT_ROOT).as_posix()}:{node.lineno}:MediaServerHelper") assert violations == [] def test_plugin_components_do_not_reexport_legacy_abi_names(): """ 插件组件只在 manager owner 定义 PluginManager，不得复制旧 ABI。 """ violations: list[str] = [] manager_path = APP_ROOT / "runtime" / "extensions" / "plugin" / "manager.py" for root in PLUGIN_COMPONENT_ROOTS: for path in (PROJECT_ROOT / root).rglob("*.py"): 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: violations.append(f"{path.relative_to(PROJECT_ROOT)}:{node.name}") elif isinstance(node, ast.ClassDef): is_canonical_manager = path == manager_path and node.name == "PluginManager" if not is_canonical_manager and ( node.name in PLUGIN_LEGACY_ABI_NAMES or node.name.endswith("Oper") ): violations.append(f"{path.relative_to(PROJECT_ROOT)}:{node.name}") elif isinstance(node, ast.ImportFrom): for alias in node.names: is_legacy_name = alias.name in PLUGIN_LEGACY_ABI_NAMES or alias.name.endswith("Oper") is_private = bool(alias.asname and alias.asname.startswith("_")) if is_legacy_name and not is_private: violations.append(f"{path.relative_to(PROJECT_ROOT)}:{alias.name}") elif isinstance(node, (ast.Assign, ast.AnnAssign)): targets = node.targets if isinstance(node, ast.Assign) else [node.target] names = {target.id for target in targets if isinstance(target, ast.Name)} forbidden = { name for name in names if name == "__all__" or name in PLUGIN_LEGACY_ABI_NAMES or name.endswith("Oper") } violations.extend(f"{path.relative_to(PROJECT_ROOT)}:{name}" for name in sorted(forbidden)) assert violations == [] def test_host_code_uses_precise_schema_modules(): """ 宿主不得重新依赖 schema 聚合入口或星号导出。 """ violations: list[str] = [] for path in APP_ROOT.rglob("*.py"): relative = path.relative_to(APP_ROOT) if relative.parts[0] in {"plugins", "schemas"}: continue 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)) break if isinstance(node, ast.ImportFrom) and node.module == "app.schemas": violations.append(str(relative)) break assert violations == [] def test_database_internals_do_not_import_db_facades(): """ DB 子模块必须依赖具体实现文件，不得回流到包级兼容入口。 """ violations: list[str] = [] for path in (APP_ROOT / "db").rglob("*.py"): if path.name == "__init__.py": continue tree = _parse_source(path) if any( 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 == [] def test_database_opers_use_dboper_transaction_dispatchers(): """ Oper 不得绕过 DbOper 的统一 Session 类型分派直接调用事务 runner。 """ runner_names = {"run_sync_transaction", "run_async_transaction"} violations: list[str] = [] for path in (APP_ROOT / "db" / "oper").rglob("*.py"): 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 if imported: relative = path.relative_to(PROJECT_ROOT).as_posix() violations.append(f"{relative}:{node.lineno}:{','.join(sorted(imported))}") assert violations == [] def test_models_and_base_require_explicit_database_sessions(): """ Model/Base 不得装饰事务，且所有 db 参数必须由调用方显式传入。 """ decorator_names = { "db_query", "db_update", "async_db_query", "async_db_update", "legacy_db_query", "legacy_db_update", "legacy_async_db_query", "legacy_async_db_update", } violations: list[str] = [] paths = [APP_ROOT / "db" / "base.py"] paths.extend((APP_ROOT / "db" / "models").rglob("*.py")) for path in paths: tree = _parse_source(path) relative = str(path.relative_to(PROJECT_ROOT)) 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") if path.name == "base.py": base_class = next(node for node in tree.body if isinstance(node, ast.ClassDef) and node.name == "Base") nodes = list(ast.walk(base_class)) for node in nodes: if not isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)): continue for decorator in node.decorator_list: name = ( decorator.id if isinstance(decorator, ast.Name) else decorator.attr if isinstance(decorator, ast.Attribute) else None ) if name in decorator_names: violations.append(f"{relative}:{node.lineno}:@{name}") arguments = [*node.args.posonlyargs, *node.args.args] defaults = [None] * (len(arguments) - len(node.args.defaults)) + list(node.args.defaults) for argument, default in zip(arguments, defaults): if argument.arg != "db": continue annotation = ast.unparse(argument.annotation) if argument.annotation else "" if default is not None or "None" in annotation: violations.append(f"{relative}:{node.lineno}:{node.name}:optional-db") assert violations == [] def test_plugin_sdk_does_not_import_or_export_host_models(): """ 插件 SDK 不得暴露 ORM Model，只有精确旧 ABI 门面可在内部访问。 """ internal_compat_imports = { ("app/sdk/_legacy/transferpending.py", "app.db.models.transferpending"), } violations: list[str] = [] for path in (APP_ROOT / "sdk").rglob("*.py"): relative = path.relative_to(PROJECT_ROOT).as_posix() tree = _parse_source(path) for node in _walk_module(tree): if ( isinstance(node, ast.ImportFrom) and node.module and (node.module == "app.db.models" or node.module.startswith("app.db.models.")) and (relative, node.module) not in internal_compat_imports ): violations.append(f"{path.relative_to(PROJECT_ROOT)}:{node.lineno}:{node.module}") assert violations == [] def test_entry_layers_do_not_import_database_implementations(): """ API、应用、编排、Agent、监控、模块和 Runtime 只能经端口访问持久化。 """ graph = _build_module_graph() layer_roots = ( "app.api", "app.application", "app.agent", "app.chain", "app.monitor", "app.modules", "app.runtime", "app.workflow", "app.adapters", ) violations = { source: sorted(dependency for dependency in dependencies if dependency.startswith("app.db")) for source, dependencies in graph.items() if source.startswith(layer_roots) and any(dependency.startswith("app.db") for dependency in dependencies) } assert violations == {} def test_migrated_modules_are_not_in_import_cycles(): """ 任何 canonical 迁移模块都不得进入完整应用依赖图的环。 """ modules = _discover_modules() known_modules = set(modules) graph = {name: _resolve_imports(name, path, known_modules) for name, path in modules.items()} relevant_cycles = [ sorted(component) for component in _strongly_connected_components(graph) if any(name.startswith(CYCLE_ROOTS) for name in component) ] assert relevant_cycles == [] def test_canonical_layers_do_not_depend_on_sdk_or_compat(): """ canonical 实现层不得反向依赖面向插件的 SDK 或兼容层。 """ violations: dict[str, set[str]] = {} modules = _discover_modules() known_modules = set(modules) for module_name, path in modules.items(): if not module_name.startswith(IMPLEMENTATION_ROOTS): continue if module_name.startswith("app.runtime.compat"): continue dependencies = _resolve_imports(module_name, path, known_modules) forbidden = { dependency for dependency in dependencies if dependency.startswith(("app.sdk", "app.runtime.compat")) } if forbidden: violations[module_name] = forbidden assert violations == {} def test_capability_packages_do_not_import_forbidden_upper_layers(): """ 新能力包只能依赖明确允许的下层或同层协作包。 """ modules = _discover_modules() known_modules = set(modules) violations: dict[str, set[str]] = {} for module_name, path in modules.items(): source_root = next( (root for root in FORBIDDEN_IMPORT_PREFIXES if module_name == root or module_name.startswith(f"{root}.")), None, ) if not source_root: continue dependencies = _resolve_imports(module_name, path, known_modules) forbidden = { dependency for dependency in dependencies if dependency.startswith(FORBIDDEN_IMPORT_PREFIXES[source_root]) } if forbidden: violations[module_name] = forbidden assert violations == {} 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 = _parse_source(path) forbidden: set[str] = set() for node in _walk_module(tree): candidates: list[str] = [] if isinstance(node, ast.Import): candidates.extend(alias.name for alias in node.names) elif isinstance(node, ast.ImportFrom) and node.module: candidates.append(node.module) forbidden.update( candidate for candidate in candidates if candidate.startswith(("fastapi", "starlette", "app.api", "app.adapters.web")) ) if forbidden: violations[str(path.relative_to(PROJECT_ROOT))] = forbidden assert violations == {} def test_site_domain_uses_foundation_dom_boundary(): """ 站点领域规则应依赖 DOM 原语，不得重新耦合聚合字符串工具。 """ modules = _discover_modules() dependencies = _resolve_imports( "app.domain.site", modules["app.domain.site"], set(modules), ) assert "app.foundation.dom" in dependencies assert "app.domain.string" not in dependencies def test_host_code_does_not_use_string_utils_facade(): """ 聚合 StringUtils 只服务插件兼容，宿主实现必须使用拆分后的能力。 """ violations: list[str] = [] for path in APP_ROOT.rglob("*.py"): relative = path.relative_to(APP_ROOT) if relative.parts[0] in {"plugins", "sdk"}: continue 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 == [] def test_host_code_does_not_import_chain_plugin_abi_roots() -> None: """ Subscribe/Transfer 包根只服务插件 ABI，宿主必须直接导入规范 Facade。 """ compatibility_roots = {"app.chain.subscribe", "app.chain.transfer"} violations: list[str] = [] for path in APP_ROOT.rglob("*.py"): relative = path.relative_to(APP_ROOT) if relative.parts[0] == "plugins": continue 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): violations.extend( f"{relative}:{node.lineno}:{alias.name}" for alias in node.names if alias.name in compatibility_roots ) assert violations == [] def test_runtime_log_is_a_dependency_leaf(): """ 底层可引用运行时日志，但日志模块本身不得反向导入应用模块。 """ modules = _discover_modules() dependencies = _resolve_imports( "app.runtime.log", modules["app.runtime.log"], set(modules), ) assert {dependency for dependency in dependencies if dependency.startswith("app.")} == set() def test_foundation_does_not_emit_runtime_logs(): """ 基础机制不打印或初始化日志系统，运行期诊断由上层调用方负责。 """ violations: list[str] = [] for path in (APP_ROOT / "foundation").rglob("*.py"): 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 if isinstance(node, ast.ImportFrom) and node.module in { "logging", "app.runtime.log", }: violations.append(str(path.relative_to(PROJECT_ROOT))) break if isinstance(node, ast.Call) and isinstance(node.func, ast.Name) and node.func.id == "print": violations.append(str(path.relative_to(PROJECT_ROOT))) break assert violations == [] def test_cache_contract_does_not_import_concrete_adapters(): """ 运行时缓存契约和内存机制不得反向导入具体缓存适配器。 """ modules = _discover_modules() dependencies = _resolve_imports( "app.runtime.cache", modules["app.runtime.cache"], set(modules), ) assert {dependency for dependency in dependencies if dependency.startswith("app.adapters.cache")} == set() def test_passkey_application_does_not_select_cache_backend(): """ PassKey 用例只消费原子缓存端口，不得识别 Redis 或后端类型。 """ modules = _discover_modules() path = modules["app.application.security.passkey"] dependencies = _resolve_imports( "app.application.security.passkey", path, set(modules), ) source = path.read_text(encoding="utf-8-sig") assert {dependency for dependency in dependencies if dependency.startswith("app.adapters.cache")} == set() assert "RedisHelper" not in source assert ".is_redis(" not in source def test_startup_explicitly_configures_passkey_challenge_cache(): """ PassKey challenge 缓存必须由启动组合根显式装配。 """ path = APP_ROOT / "startup" / "composition" / "security.py" 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 _walk_module(tree) ) assert configured is True def test_resource_adapter_does_not_restart_process(): """ 资源下载安装适配器不得反向调用进程重启能力。 """ modules = _discover_modules() dependencies = _resolve_imports( "app.adapters.system.resource", modules["app.adapters.system.resource"], set(modules), ) assert "app.runtime.state" not in dependencies def test_modules_do_not_import_other_modules_or_chain(): """ 模块之间以及模块对链层的直接依赖被禁止，跨模块编排归链层。 `app.modules._base` 是模块共享样板基类包（模块发现会跳过），不视为业务模块。 """ modules = _discover_modules() known_modules = set(modules) violations: dict[str, set[str]] = {} for module_name, path in modules.items(): if not module_name.startswith("app.modules."): continue own_package = module_name.split(".")[2] dependencies = _resolve_imports(module_name, path, known_modules) forbidden = { dependency for dependency in dependencies if dependency.startswith("app.chain") or (dependency.startswith("app.modules.") and dependency.split(".")[2] not in (own_package, "_base")) } if forbidden: violations[module_name] = forbidden assert violations == {} def test_entrypoints_do_not_import_module_internals(): """ 入口层不得穿透导入具体模块实现，应经由链层或应用服务。 """ modules = _discover_modules() known_modules = set(modules) entrypoint_roots = ("app.api", "app.agent", "app.monitor", "app.workflow", "app.doctor") violations: dict[str, set[str]] = {} for module_name, path in modules.items(): if not module_name.startswith(entrypoint_roots): continue dependencies = _resolve_imports(module_name, path, known_modules) forbidden = {dependency for dependency in dependencies if dependency.startswith("app.modules.")} if forbidden: violations[module_name] = forbidden assert violations == {} 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 = _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) elif isinstance(node, ast.ImportFrom) and node.module and node.level == 0: names.append(node.module) if any(name.split(".")[0] in forbidden_sdks for name in names): violations.append(str(path.relative_to(PROJECT_ROOT))) break assert violations == [] def test_chain_does_not_import_module_internals(): """ 链层与模块只能通过 run_module 方法名契约互联，禁止直接导入模块实现。 模块内容必须封闭在模块内部，链层不显式指定具体模块的类、异常或常量， 这样模块才是可插拔的。 """ modules = _discover_modules() known_modules = set(modules) violations: dict[str, set[str]] = {} for module_name, path in modules.items(): if not module_name.startswith("app.chain"): continue dependencies = _resolve_imports(module_name, path, known_modules) forbidden = {dependency for dependency in dependencies if dependency.startswith("app.modules.")} if forbidden: violations[module_name] = forbidden assert violations == {} @lru_cache(maxsize=1) def _build_module_graph() -> dict[str, set[str]]: """ 构建非插件模块的完整静态依赖图，供既有包治理断言复用。 纯静态 AST 分析无副作用，结果可安全缓存；多断言共享一次解析。 """ modules = _discover_modules() known_modules = set(modules) return {name: _resolve_imports(name, path, known_modules) for name, path in modules.items()} def _load_dependency_policy() -> dict: """ 读取人工审查的依赖语义 policy；生成基线不得改写该文件。 """ return json.loads(DEPENDENCY_POLICY_PATH.read_text(encoding="utf-8")) def _scc_policy_violations( graph: dict[str, set[str]], entries: list[dict], ) -> tuple[list[list[str]], list[list[str]]]: """ 返回未审查 SCC 与已经失效但未清理的 policy SCC。 """ actual = {tuple(component) for component in _strongly_connected_components(graph)} reviewed = {tuple(entry["modules"]) for entry in entries} return ( [list(component) for component in sorted(actual - reviewed)], [list(component) for component in sorted(reviewed - actual)], ) def test_complete_host_sccs_match_reviewed_policy() -> None: """ 完整宿主图的每个 SCC 都必须精确匹配人工 policy，且不得保留陈旧项。 """ policy = _load_dependency_policy() entries = policy["allowed_sccs"] assert policy["schema_version"] == 3 assert policy["scope"] == { "dependency_kind": "static_runtime_imports", "excluded_roots": ["app/plugins"], "include_parent_package_initialization": True, "root": "app", } assert {entry["classification"] for entry in entries} == {"contained_vendor"} assert len({entry["id"] for entry in entries}) == len(entries) assert all(entry["modules"] == sorted(set(entry["modules"])) for entry in entries) assert all(entry["reason"] and entry["tracking"] for entry in entries) all_members = [member for entry in entries for member in entry["modules"]] assert len(all_members) == len(set(all_members)) vendor = next(entry for entry in entries if entry["classification"] == "contained_vendor") assert vendor["id"] == "themoviedb-vendored-package" assert vendor["tracking"] == "replace-or-upgrade-vendored-package" assert len(vendor["modules"]) == 29 assert all(member.startswith("app.modules.themoviedb") for member in vendor["modules"]) unreviewed, stale = _scc_policy_violations(_build_module_graph(), entries) assert unreviewed == [] assert stale == [] def test_scc_policy_rejects_unreviewed_cycles_in_every_host_root() -> None: """ API、Chain、Module 与 Startup 内的新环都必须落入未审查集合。 """ for root in ("app.api", "app.chain", "app.modules", "app.startup"): first = f"{root}.first" second = f"{root}.second" unreviewed, stale = _scc_policy_violations( {first: {second}, second: {first}}, [], ) assert unreviewed == [[first, second]] assert stale == [] def test_scc_policy_rejects_changed_or_stale_membership() -> None: """ 已知 SCC 缩小、扩大或消失后必须同步审查 policy，不能静默放行。 """ entries = [{"modules": ["app.chain.first", "app.chain.second", "app.chain.third"]}] unreviewed, stale = _scc_policy_violations( { "app.chain.first": {"app.chain.second"}, "app.chain.second": {"app.chain.first"}, "app.chain.third": set(), }, entries, ) assert unreviewed == [["app.chain.first", "app.chain.second"]] assert stale == [["app.chain.first", "app.chain.second", "app.chain.third"]] def test_contained_vendor_scc_may_have_one_way_outbound_dependency() -> None: """ vendor SCC 的普通单向向下依赖不应把包外模块误纳入 containment。 """ first = "app.modules.themoviedb.first" second = "app.modules.themoviedb.second" domain = "app.domain.media" entries = [{"modules": [first, second]}] unreviewed, stale = _scc_policy_violations( {first: {second, domain}, second: {first}, domain: set()}, entries, ) assert unreviewed == [] assert stale == [] def test_host_dependency_graph_excludes_plugin_copies() -> None: """ `app/plugins/**` 是运行时副本，不能进入宿主模块或 SCC 图。 """ assert all(not module.startswith("app.plugins") for module in _discover_modules()) def test_chain_does_not_import_agent_implementation(): """ 编排层不得反向依赖 Agent 实现，跨域编排经 application 门面。 """ violations: dict[str, set[str]] = {} for module_name, dependencies in _build_module_graph().items(): if not module_name.startswith("app.chain"): continue forbidden = {dependency for dependency in dependencies if dependency.startswith("app.agent")} if forbidden: violations[module_name] = forbidden assert violations == {} def test_agent_application_facade_does_not_import_agent_implementation(): """ Agent application 门面只能接收组合根注入，不能反向解析具体实现。 """ dependencies = _build_module_graph()["app.application.agent"] assert {dependency for dependency in dependencies if dependency.startswith("app.agent")} == set() def test_host_consumers_get_agent_manager_through_application_facade(): """ 宿主消费者不得绕过 Application 门面直接定位 Agent manager。 """ forbidden = {"get_agent_manager", "get_running_agent_manager"} violations: dict[str, set[str]] = {} for module_name, path in _discover_modules().items(): if module_name.startswith(("app.agent", "app.startup")): continue tree = _parse_source(path) imported = { alias.name 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 } if imported: violations[module_name] = imported assert violations == {} 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 = _parse_source(path) assert all(isinstance(node, ast.Expr) for node in tree.body), relative_path assert set(SYMBOL_ALIASES["app.agent"]) == { "AgentChain", "AgentManager", "HEARTBEAT_SESSION_PREFIX", "MoviePilotAgent", "ReplyMode", "UNSUPPORTED_IMAGE_INPUT_MESSAGE", "agent_manager", } assert set(SYMBOL_ALIASES["app.agent.llm"]) == {"LLMHelper"} helper_alias = MODULE_ALIASES["app.helper.llm"] assert helper_alias.target == "app.agent.llm.helper" assert helper_alias.replacement == "app.agent.llm.helper" assert not helper_alias.is_package 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 = _parse_source(path) imported = { f"{node.module}.{alias.name}" 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 } if imported: violations[module_name] = imported assert violations == {} def test_host_consumers_resolve_llm_provider_runtime_through_gateway(): """ 宿主不得绕过 gateway 或 LLM 公共导出穿透 provider 实现。 """ violations: dict[str, set[str]] = {} graph = _build_module_graph() for module_name, path in _discover_modules().items(): if module_name.startswith(("app.agent", "app.startup")): continue tree = _parse_source(path) imported = { alias.name 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" } forbidden = set(imported) if "app.agent.llm.provider" in graph[module_name]: forbidden.add("app.agent.llm.provider") if forbidden: violations[module_name] = forbidden assert violations == {} def test_host_consumers_use_agent_audio_capability_application_port(): """ 宿主消费者不得绕过 Application 门面直连 Agent 音频实现。 """ violations: dict[str, set[str]] = {} for module_name, path in _discover_modules().items(): if module_name.startswith(("app.agent", "app.startup")): continue tree = _parse_source(path) imported = { alias.name 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" } if imported: violations[module_name] = imported assert violations == {} def test_application_services_do_not_resolve_event_manager_singleton(): """ Application 服务必须接收事件端口，不得自行定位进程级事件单例。 """ violations = { module_name: dependencies & {"app.runtime.events"} for module_name, dependencies in _build_module_graph().items() if module_name.startswith("app.application") and "app.runtime.events" in dependencies } assert violations == {} def test_http_endpoints_do_not_register_process_event_listeners(): """ HTTP 端点不得拥有进程级事件监听器，监听装配必须留在 startup。 """ 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() def test_agent_tools_do_not_import_entrypoint_internals(): """ Agent 工具不得穿透导入 HTTP 端点、调度器与命令注册表内部实现。 工具对进程级状态的读写必须收敛到 application 门面， 否则 agent 层与入口层互相穿透会形成不可测试的循环。 """ violations: dict[str, set[str]] = {} for module_name, dependencies in _build_module_graph().items(): if not module_name.startswith("app.agent.tools"): continue forbidden = { dependency for dependency in dependencies if dependency.startswith(("app.api", "app.scheduler", "app.command")) } if forbidden: violations[module_name] = forbidden assert violations == {} def test_modules_read_deployment_settings_through_runtime_port(): """ 宿主 Module 不得绕过 runtime 配置端口直接依赖 Settings 实例。 """ violations: list[str] = [] for path in (APP_ROOT / "modules").rglob("*.py"): 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): violations.append(path.relative_to(PROJECT_ROOT).as_posix()) break assert violations == [] def test_runtime_implementation_does_not_use_legacy_settings_proxy(): """ runtime 实现只能使用只读配置端口，不得重新引入迁移期代理对象。 """ violations: list[str] = [] for path in (APP_ROOT / "runtime").rglob("*.py"): if path == APP_ROOT / "runtime" / "settings.py": continue tree = _parse_source(path) for node in _walk_module(tree): if not isinstance(node, ast.ImportFrom): continue if node.module != "app.runtime.settings": continue if any(alias.name.lower().endswith("compat") for alias in node.names): violations.append(path.relative_to(PROJECT_ROOT).as_posix()) break assert violations == [] def test_deprecated_settings_proxy_imports_are_zero(): """ 宿主代码不得导入已删除的 Settings 兼容代理。 """ limits = { "adapters": 0, "agent": 0, "application": 0, "cli.py": 0, "doctor": 0, "factory.py": 0, "main.py": 0, "modules": 0, "startup": 0, } counts: dict[str, int] = {} for path in APP_ROOT.rglob("*.py"): if path == APP_ROOT / "runtime" / "settings.py": continue if path.is_relative_to(APP_ROOT / "plugins"): continue 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 _walk_module(tree) ) if not imports_compat: continue relative = path.relative_to(APP_ROOT) group = relative.parts[0] if len(relative.parts) > 1 else relative.as_posix() counts[group] = counts.get(group, 0) + 1 unexpected = set(counts) - set(limits) exceeded = {group: count for group, count in counts.items() if group in limits and count > limits[group]} assert unexpected == set() assert exceeded == {} def test_global_settings_imports_stay_within_compatibility_baseline(): """ 真实 Settings 对象只能保留在已知迁移点和插件 SDK，不得产生新宿主调用。 """ allowed = { "app/sdk/config.py", "app/startup/initializers/modules.py", # 共用测试引导是插件仓单测的组合根，与 tests/conftest.py 一样按测试 settings 登记运行时配置快照。 "app/testing/bootstrap.py", } imports: set[str] = set() for path in APP_ROOT.rglob("*.py"): if path.is_relative_to(APP_ROOT / "plugins"): continue 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 _walk_module(tree) ): imports.add(path.relative_to(PROJECT_ROOT).as_posix()) assert imports <= allowed def test_api_does_not_import_factory(): """ 装配器（factory）只允许 app.main 使用，HTTP 端点不得回引。 """ violations: dict[str, set[str]] = {} for module_name, dependencies in _build_module_graph().items(): if not module_name.startswith("app.api"): continue forbidden = {dependency for dependency in dependencies if dependency.startswith("app.factory")} if forbidden: violations[module_name] = forbidden assert violations == {} def test_web_agent_non_transport_state_stays_in_application(): """ Agent 端点不得重新拥有附件、音频、投影或会话写入实现。 """ endpoint_path = APP_ROOT / "api" / "endpoints" / "agent.py" application_path = APP_ROOT / "application" / "messaging" / "agent.py" endpoint_tree = ast.parse( endpoint_path.read_text(encoding="utf-8-sig"), filename=str(endpoint_path), ) application_tree = ast.parse( application_path.read_text(encoding="utf-8-sig"), filename=str(application_path), ) endpoint_owners = { node.name for node in endpoint_tree.body if isinstance(node, (ast.ClassDef, ast.FunctionDef, ast.AsyncFunctionDef)) } application_owners = { node.name for node in application_tree.body if isinstance(node, (ast.ClassDef, ast.FunctionDef, ast.AsyncFunctionDef)) } required = { "WebAgentEventPublisher", "apply_web_agent_display_event", "build_web_agent_command_items", "build_web_agent_input_attachments", "build_web_agent_message_events_async", "build_web_agent_session_id_async", "build_web_agent_stream", "collect_web_agent_traditional_events", "get_web_agent_registered_file", "prepare_web_agent_audio_attachment_path_async", "save_web_agent_display_snapshot", "transcribe_web_agent_audio_input", } forbidden_endpoint_imports = { "app.application.commands", "app.application.messaging.router", "app.chain.message", "app.runtime.execution", "app.runtime.stop", } endpoint_imports = { node.module for node in ast.walk(endpoint_tree) if isinstance(node, ast.ImportFrom) and node.module } assert required <= application_owners assert required.isdisjoint(endpoint_owners) assert endpoint_imports.isdisjoint(forbidden_endpoint_imports) def test_plugin_market_use_cases_stay_in_application(): """ Plugin 端点不得重新组合市场 Adapter 或拥有目录投影实现。 """ endpoint_path = APP_ROOT / "api" / "endpoints" / "plugin.py" endpoint_tree = ast.parse( endpoint_path.read_text(encoding="utf-8-sig"), filename=str(endpoint_path), ) forbidden_imports = { "app.adapters.external.market", "app.adapters.external.server", } imported = {node.module for node in ast.walk(endpoint_tree) if isinstance(node, ast.ImportFrom) and node.module} endpoint_owners = { node.name for node in endpoint_tree.body if isinstance(node, (ast.ClassDef, ast.FunctionDef, ast.AsyncFunctionDef)) } assert forbidden_imports.isdisjoint(imported) assert { "_get_market_plugin_from_repo", "_get_plugin_history_detail", "_installed_plugins_with_declared_metadata", "_prepare_update_candidates", }.isdisjoint(endpoint_owners) def test_system_nettest_is_owned_by_application_service(): """ System API 不得重新拥有网络目标目录、安全准入或具体传输组装。 """ endpoint_path = APP_ROOT / "api" / "endpoints" / "system.py" endpoint_tree = ast.parse( endpoint_path.read_text(encoding="utf-8-sig"), filename=str(endpoint_path), ) retired_endpoint_symbols = { "_build_nettest_rules", "_close_nettest_response", "_get_nettest_rule", "_is_allowed_nettest_redirect", "_match_nettest_prefix", "_validate_nettest_url", } endpoint_functions = { node.name for node in endpoint_tree.body if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)) } assert endpoint_functions.isdisjoint(retired_endpoint_symbols) assert not any( isinstance(node, ast.Call) and isinstance(node.func, ast.Name) and node.func.id == "NetworkTestService" for node in ast.walk(endpoint_tree) ) application_path = APP_ROOT / "application" / "network.py" application_tree = ast.parse( application_path.read_text(encoding="utf-8-sig"), filename=str(application_path), ) concrete_adapter_imports = { node.module for node in ast.walk(application_tree) if isinstance(node, ast.ImportFrom) and node.module and node.module.startswith("app.adapters") } concrete_adapter_imports.update( alias.name for node in ast.walk(application_tree) if isinstance(node, ast.Import) for alias in node.names if alias.name.startswith("app.adapters") ) assert concrete_adapter_imports == set() def test_plugin_market_construction_is_owned_by_startup_composition(): """ 插件 initializer 只能消费组合根创建的市场依赖，禁止隐式构造第二套 owner。 """ initializer_path = APP_ROOT / "startup" / "initializers" / "plugins.py" initializer_tree = ast.parse( initializer_path.read_text(encoding="utf-8-sig"), filename=str(initializer_path), ) initializer_imports = { alias.name for node in ast.walk(initializer_tree) if isinstance(node, ast.ImportFrom) for alias in node.names } assert "compose_plugin_market" in initializer_imports forbidden_names = { "PluginMarketTransport", "PluginPackageSourceClient", "PluginPackageManager", "PluginRuntimeHealth", "PluginDependencyInstaller", } assert forbidden_names.isdisjoint(initializer_imports) assert not any( isinstance(node, ast.Call) and isinstance(node.func, ast.Name) and node.func.id in forbidden_names for node in ast.walk(initializer_tree) ) composition_path = APP_ROOT / "startup" / "composition" / "plugin.py" composition_tree = ast.parse( composition_path.read_text(encoding="utf-8-sig"), filename=str(composition_path), ) composition_calls = [ node.func.id for node in ast.walk(composition_tree) if isinstance(node, ast.Call) and isinstance(node.func, ast.Name) and node.func.id in forbidden_names ] assert composition_calls.count("PluginMarketTransport") == 1 assert composition_calls.count("PluginPackageSourceClient") == 1 assert composition_calls.count("PluginPackageManager") == 1 assert composition_calls.count("PluginRuntimeHealth") == 1 assert composition_calls.count("PluginDependencyInstaller") == 1 def test_plugin_market_transport_is_process_shared_across_compat_and_composition(): """ 兼容门面和组合根不得物化两套插件市场传输 owner。 """ from app.adapters.external.market import _plugin_market_transport from app.adapters.external.plugin.client import PluginMarketTransport assert PluginMarketTransport.get_existing_instance() is _plugin_market_transport def test_plugin_runtime_owners_are_shared_with_compat_facade(): """ 兼容门面必须消费组合根发布的健康检查与依赖安装 owner。 """ from app.adapters.external.market import ( _plugin_dependency_owner, _plugin_runtime_health_owner, configure_plugin_runtime_owners, reset_plugin_runtime_owners, ) health = object() dependency = object() configure_plugin_runtime_owners( # 只验证依赖所有者的对象身份，用占位对象隔离真实资源构造。 health=health, # type: ignore[arg-type] # 只验证依赖所有者的对象身份，用占位对象隔离真实资源构造。 dependency=dependency, # type: ignore[arg-type] ) try: assert _plugin_runtime_health_owner() is health assert _plugin_dependency_owner() is dependency finally: reset_plugin_runtime_owners() def test_domain_initializer_delegates_construction_to_composition(): """ 领域 initializer 只负责生命周期调用，不得直接构造 Adapter 或领域服务。 """ initializer_path = APP_ROOT / "startup" / "initializers" / "domain.py" initializer_tree = ast.parse( initializer_path.read_text(encoding="utf-8-sig"), filename=str(initializer_path), ) imported_modules = { node.module for node in ast.walk(initializer_tree) if isinstance(node, ast.ImportFrom) and node.module } assert imported_modules == {"app.startup.composition.domain"} calls = [ast.unparse(node.func) for node in ast.walk(initializer_tree) if isinstance(node, ast.Call)] assert calls == ["compose_domain_dependencies"] composition_path = APP_ROOT / "startup" / "composition" / "domain.py" composition_tree = ast.parse( composition_path.read_text(encoding="utf-8-sig"), filename=str(composition_path), ) composition_imports = { node.module for node in ast.walk(composition_tree) if isinstance(node, ast.ImportFrom) and node.module } assert "app.adapters.network.resolver" in composition_imports assert "app.adapters.system.host" in composition_imports assert "app.application.recognition" in composition_imports def test_network_initializer_delegates_construction_to_composition(): """ 网络 initializer 只保留生命周期入口，不得重新拥有具体技术适配器。 """ initializer_path = APP_ROOT / "startup" / "initializers" / "network.py" initializer_tree = ast.parse( initializer_path.read_text(encoding="utf-8-sig"), filename=str(initializer_path), ) imported_modules = { node.module for node in ast.walk(initializer_tree) if isinstance(node, ast.ImportFrom) and node.module } assert imported_modules == {"app.startup.composition.network"} calls = {ast.unparse(node.func) for node in ast.walk(initializer_tree) if isinstance(node, ast.Call)} assert calls == { "configure_chain_network_composition", "reset_chain_network_composition", } composition_path = APP_ROOT / "startup" / "composition" / "network.py" composition_tree = ast.parse( composition_path.read_text(encoding="utf-8-sig"), filename=str(composition_path), ) composition_imports = { node.module for node in ast.walk(composition_tree) if isinstance(node, ast.ImportFrom) and node.module } assert "app.adapters.network.http" in composition_imports assert "app.adapters.system.host" in composition_imports assert "app.chain.download.ports" in composition_imports def test_site_initializer_delegates_construction_to_composition(): """ 站点 initializer 只保留生命周期入口，不得重新拥有具体技术适配器。 """ initializer_path = APP_ROOT / "startup" / "initializers" / "site.py" initializer_tree = ast.parse( initializer_path.read_text(encoding="utf-8-sig"), filename=str(initializer_path), ) imported_modules = { node.module for node in ast.walk(initializer_tree) if isinstance(node, ast.ImportFrom) and node.module } assert imported_modules == {"app.startup.composition.site"} calls = {ast.unparse(node.func) for node in ast.walk(initializer_tree) if isinstance(node, ast.Call)} assert calls == { "configure_site_access_composition", "reset_site_access_composition", } composition_path = APP_ROOT / "startup" / "composition" / "site.py" composition_tree = ast.parse( composition_path.read_text(encoding="utf-8-sig"), filename=str(composition_path), ) composition_imports = { node.module for node in ast.walk(composition_tree) if isinstance(node, ast.ImportFrom) and node.module } assert "app.adapters.external.cookiecloud" in composition_imports assert "app.adapters.network.browser" in composition_imports assert "app.application.rss" in composition_imports def test_chain_initializer_delegates_construction_to_composition(): """ Chain initializer 只保留生命周期入口，不得重新拥有外部技术适配器。 """ initializer_path = APP_ROOT / "startup" / "initializers" / "chain.py" initializer_tree = ast.parse( initializer_path.read_text(encoding="utf-8-sig"), filename=str(initializer_path), ) imported_modules = { node.module for node in ast.walk(initializer_tree) if isinstance(node, ast.ImportFrom) and node.module } assert imported_modules == {"app.startup.composition.chain"} calls = {ast.unparse(node.func) for node in ast.walk(initializer_tree) if isinstance(node, ast.Call)} assert calls == { "configure_chain_port_composition", "reset_chain_port_composition", } composition_path = APP_ROOT / "startup" / "composition" / "chain.py" composition_tree = ast.parse( composition_path.read_text(encoding="utf-8-sig"), filename=str(composition_path), ) composition_imports = { node.module for node in ast.walk(composition_tree) if isinstance(node, ast.ImportFrom) and node.module } assert "app.adapters.external.server" in composition_imports assert "app.adapters.system.host" in composition_imports assert "app.chain._recognition" in composition_imports def test_cache_initializer_delegates_construction_to_composition(): """ 缓存 initializer 只保留启动入口，不得直接登记具体缓存 Adapter。 """ initializer_path = APP_ROOT / "startup" / "initializers" / "cache.py" initializer_tree = ast.parse( initializer_path.read_text(encoding="utf-8-sig"), filename=str(initializer_path), ) imported_modules = { node.module for node in ast.walk(initializer_tree) if isinstance(node, ast.ImportFrom) and node.module } assert imported_modules == {"app.startup.composition.cache"} calls = [ast.unparse(node.func) for node in ast.walk(initializer_tree) if isinstance(node, ast.Call)] assert calls == ["configure_cache_composition"] def test_resource_initializer_delegates_construction_to_composition(): """ 资源 initializer 只保留生命周期入口，不得构造托管资源 Adapter。 """ initializer_path = APP_ROOT / "startup" / "initializers" / "resources.py" initializer_tree = ast.parse( initializer_path.read_text(encoding="utf-8-sig"), filename=str(initializer_path), ) imported_modules = { node.module for node in ast.walk(initializer_tree) if isinstance(node, ast.ImportFrom) and node.module } assert imported_modules == { "app.runtime.capabilities.runtime", "app.startup.composition.resource", } constructors = {ast.unparse(node.func) for node in ast.walk(initializer_tree) if isinstance(node, ast.Call)} assert constructors == { "configure_managed_resource_composition", "stop_managed_resource_composition", "reset_managed_resource_composition", } def test_modules_initializer_consumes_doh_and_workflow_composition(): """ 模块 initializer 不得直接构造 DoH 或事务型工作流执行 Adapter。 """ initializer_path = APP_ROOT / "startup" / "initializers" / "modules.py" initializer_tree = ast.parse( initializer_path.read_text(encoding="utf-8-sig"), filename=str(initializer_path), ) imported_modules = { node.module for node in ast.walk(initializer_tree) if isinstance(node, ast.ImportFrom) and node.module } assert "app.adapters.network.doh" not in imported_modules assert "app.db.adapters.workflow" not in imported_modules calls = {ast.unparse(node.func) for node in ast.walk(initializer_tree) if isinstance(node, ast.Call)} assert "DohHelper" not in calls assert "TransactionalWorkflowExecutionService" not in calls assert "configure_doh_composition" in calls assert "configure_workflow_execution_composition" in calls def test_system_api_business_orchestration_is_owned_by_application_service(): """ System API 只保留传输映射，不得重新拥有日志、设置或更新实现。 """ endpoint_path = APP_ROOT / "api" / "endpoints" / "system.py" endpoint_tree = ast.parse( endpoint_path.read_text(encoding="utf-8-sig"), filename=str(endpoint_path), ) imported = {node.module for node in ast.walk(endpoint_tree) if isinstance(node, ast.ImportFrom) and node.module} direct_adapters = { module for module in imported if module.startswith("app.adapters") and module != "app.adapters.web.security.access" } endpoint_functions = { node.name for node in endpoint_tree.body if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)) } retired_helpers = { "_collect_named_log_files", "_validate_database_backup_config", "_validate_llm_server_tool_config", "_is_allowed_plugin_market_wiki_url", } assert not direct_adapters assert endpoint_functions.isdisjoint(retired_helpers) assert not imported.intersection( { "app.application.plugin.runtime", "app.runtime.events", "app.runtime.state", } ) application_path = APP_ROOT / "application" / "system.py" application_tree = ast.parse( application_path.read_text(encoding="utf-8-sig"), filename=str(application_path), ) assert not any( isinstance(node, ast.ImportFrom) and node.module and node.module.startswith("app.adapters") for node in ast.walk(application_tree) ) composition_source = (APP_ROOT / "startup" / "composition" / "system.py").read_text(encoding="utf-8-sig") assert "def compose_system_service(" in composition_source assert "system=compose_system_service(" in (APP_ROOT / "startup" / "composition" / "runtime.py").read_text( encoding="utf-8-sig" ) PROCESS_LEVEL_ROOTS = ( "app.api", "app.chain", "app.agent", "app.scheduler", "app.command", "app.monitor", "app.startup", "app.factory", ) def test_process_level_packages_are_not_mutually_cyclic(): """ 进程级根包之间不得形成跨包强连通分量。 允许的环只存在于：单一包内部（modules 模块内、db 内、schemas 包内、 agent 子域内、doctor 内）。跨根包的环意味着入口层、编排层与 Agent 层 互相穿透，破坏可插拔性与可测试性。 """ graph = _build_module_graph() components = _strongly_connected_components(graph) violations: list[list[str]] = [] for component in components: roots = {name.split(".")[1] for name in component if name.startswith("app.") and name.count(".") >= 1} involved = {root for root in roots if f"app.{root}" in PROCESS_LEVEL_ROOTS} if len(involved) > 1: violations.append(sorted(component)) assert violations == []