refactor: unify workflow runtime boundary

app/api/dependencies/workflow.py

@@ -14,9 +14,9 @@ from app.application.workflow import (
     WorkflowDefinitionCommand,
     WorkflowMutationCommand,
     WorkflowQueryService,
+    get_workflow_manager,
 )
 from app.runtime.config import global_vars
-from app.workflow import WorkFlowManager
 from app.startup.composition.context import HostRuntime
 
 
@@ -26,7 +26,7 @@ def get_workflow_mutation_command(
 ) -> WorkflowMutationCommand:
     """组装请求级工作流写用例和提交后的调度副作用。"""
     scheduler = Scheduler()
-    workflow_manager = WorkFlowManager()
+    workflow_manager = get_workflow_manager()
     system_config = cast(WorkflowCachePort, runtime.workflow.system_config())
     return WorkflowMutationCommand(
         repository=runtime.workflow.repository(db),

app/api/endpoints/workflow.py

@@ -13,10 +13,10 @@ from app.application.workflow import (
     WorkflowDefinitionCommand,
     WorkflowMutationCommand,
     WorkflowQueryService,
+    get_workflow_manager,
 )
 from app.chain.workflow import WorkflowChain
 from app.application.plugin.runtime import get_plugin_manager as PluginManager
-from app.workflow import WorkFlowManager
 from app.api.dependencies.auth import (
     get_current_active_manage_user,
     get_current_active_manage_user_async,
@@ -78,7 +78,7 @@ async def list_actions(_: Any = Depends(get_current_active_manage_user_async)) -
     """
     获取所有动作
     """
-    return WorkFlowManager().list_actions()
+    return get_workflow_manager().list_actions()
 
 
 @router.get(

app/application/workflow.py

@@ -17,6 +17,56 @@ SUPPORTED_WORKFLOW_TRIGGERS = {
 }
 
 
+class WorkflowRuntime(Protocol):
+    """声明宿主入口与 Chain 消费的工作流运行时能力。"""
+
+    def execute(self, *args: Any, **kwargs: Any) -> Any:
+        """执行单个工作流动作，参数与 concrete 管理器保持一致。"""
+        ...
+
+    def list_actions(self) -> list[dict[str, Any]]:
+        """返回当前运行时登记的工作流动作定义。"""
+        ...
+
+    def load_workflow_events(self, workflow_id: Optional[int] = None) -> None:
+        """加载全部或指定工作流的事件触发器。"""
+        ...
+
+    def remove_workflow_event(
+            self,
+            workflow_id: Optional[int] = None,
+            event_type_str: Optional[str] = None,
+    ) -> None:
+        """移除全部或指定工作流的事件触发器。"""
+        ...
+
+    def update_workflow_event(self, workflow: Any) -> None:
+        """按最新定义刷新工作流事件触发器。"""
+        ...
+
+
+WorkflowRuntimeProvider = Callable[[], WorkflowRuntime]
+
+
+def _unconfigured_workflow_runtime() -> WorkflowRuntime:
+    """拒绝在启动组合根装配前隐式创建工作流管理器。"""
+    raise RuntimeError("工作流运行时尚未由启动组合根装配")
+
+
+_workflow_runtime_provider: WorkflowRuntimeProvider = _unconfigured_workflow_runtime
+
+
+def configure_workflow_runtime(provider: WorkflowRuntimeProvider) -> None:
+    """由启动组合根登记工作流运行时实例提供器。"""
+    global _workflow_runtime_provider
+    _workflow_runtime_provider = provider
+
+
+def get_workflow_manager() -> WorkflowRuntime:
+    """返回组合根提供的工作流运行时，避免消费者直接定位 Singleton。"""
+    return _workflow_runtime_provider()
+
+
 class AsyncWorkflowQueryRepository(Protocol):
     """工作流查询用例需要的异步读取端口。"""
 

app/chain/workflow.py

@@ -17,6 +17,7 @@ from pydantic import BaseModel
 from app.chain import ChainBase
 from app.runtime.config import global_vars
 from app.runtime.events import Event, eventmanager
+from app.application.workflow import get_workflow_manager
 from app.application.chain.data import WorkflowPortProxy as WorkflowOper
 from app.runtime.log import logger
 from app.schemas.workflow import ActionContext
@@ -25,7 +26,6 @@ from app.schemas.workflow import Action
 from app.schemas.workflow import ActionExecution
 from app.schemas.workflow import ActionResult
 from app.schemas.types import EventType
-from app.workflow import WorkFlowManager
 
 ARTIFACT_FIELDS = {"torrents", "medias", "fileitems", "downloads", "sites", "subscribes"}
 DEFAULT_WORKFLOW_MAX_WORKERS = 4
@@ -162,7 +162,7 @@ class WorkflowExecutor:
         self.flow_failed = set()
 
         # 工作流管理器
-        self.workflowmanager = WorkFlowManager()
+        self.workflowmanager = get_workflow_manager()
         # 线程安全队列
         self.queue = deque()
         self.queued_actions = set()

app/startup/initializers/workflow.py

@@ -1,6 +1,11 @@
+from app.application.workflow import configure_workflow_runtime
 from app.workflow import WorkFlowManager
 
 
+# 启动模块是 concrete WorkFlowManager 的唯一宿主装配边界。
+configure_workflow_runtime(lambda: WorkFlowManager())
+
+
 def init_workflow():
     """
     初始化工作流

docs/refactor/backend-architecture-next-stage.md

@@ -6,7 +6,7 @@
 > 审计范围：宿主后端；排除 `app/plugins/**` 运行时插件副本
 > 规范优先级：`AGENTS.md` 与 `docs/rules/` 高于本文
 > 相关文档：`docs/architecture-overview.md`、`docs/refactor/backend-architecture-governance.md`、`docs/refactor/backend-module-refactor-compatibility.md`
-> 实施进度：阶段 0～6 的宿主架构能力已完成收口；API/Application 公共复杂度基线已清零，启动组合根的 SystemConfigOper 构造点已由 14 降至 1；API 进程内后台任务已完成首批统一登记，插件仓适配和 Outbox 外围扩展仍按风险切片推进。Model/Base 查询与写装饰器、legacy 隐式会话外壳均已清零，插件 SDK 也不再导出宿主 Model。2026-08-23 的长期整改阶段 0 已恢复宿主、启动性能、官方插件和 SDK 契约门禁的可信基线；阶段 1a 已补齐 TaskRegistry owner 零债务门禁和诚实的关停超时语义；阶段 1b1 已收口整理 worker、pending 回放、失败通知、进程内 AI 重试、插件监控与事件投递的生命周期所有权；2026-08-24 的阶段 2 已将 212 个已观察宿主模块方法的 legacy aggregation 清零，并补齐可执行 fanout 与下载器文件 DTO 边界；阶段 3 已将消息交互和远程命令的订阅删除统一到 Application/UoW/outbox，宿主不再调用裸线程统计入口；阶段 4 已统一七种消息渠道的宿主回环与后台执行边界；阶段 5 已补齐事件窗口聚合任务的生命周期所有权；阶段 6 已统一插件文件操作的取消完成语义；阶段 7 已统一插件协程补偿的终态等待；阶段 8 已统一宿主同步函数的异步线程池入口。
+> 实施进度：阶段 0～6 的宿主架构能力已完成收口；API/Application 公共复杂度基线已清零，启动组合根的 SystemConfigOper 构造点已由 14 降至 1；API 进程内后台任务已完成首批统一登记，插件仓适配和 Outbox 外围扩展仍按风险切片推进。Model/Base 查询与写装饰器、legacy 隐式会话外壳均已清零，插件 SDK 也不再导出宿主 Model。2026-08-23 的长期整改阶段 0 已恢复宿主、启动性能、官方插件和 SDK 契约门禁的可信基线；阶段 1a 已补齐 TaskRegistry owner 零债务门禁和诚实的关停超时语义；阶段 1b1 已收口整理 worker、pending 回放、失败通知、进程内 AI 重试、插件监控与事件投递的生命周期所有权；2026-08-24 的阶段 2 已将 212 个已观察宿主模块方法的 legacy aggregation 清零，并补齐可执行 fanout 与下载器文件 DTO 边界；阶段 3 已将消息交互和远程命令的订阅删除统一到 Application/UoW/outbox，宿主不再调用裸线程统计入口；阶段 4 已统一七种消息渠道的宿主回环与后台执行边界；阶段 5 已补齐事件窗口聚合任务的生命周期所有权；阶段 6 已统一插件文件操作的取消完成语义；阶段 7 已统一插件协程补偿的终态等待；阶段 8 已统一宿主同步函数的异步线程池入口；阶段 9 已统一工作流运行时的宿主获取路径。
 
 ## 当前复核结论（2026-08-24）
 
@@ -106,6 +106,17 @@
 - 兼容边界不变：七个渠道类、模块方法、配置字段、消息 payload、同步/异步 SDK 回调方式和
   `/api/v1/message` HTTP 合同均未改；没有修改插件仓或向 SDK/Compat 新增宿主内部入口。
 
+### 长期整改阶段 9：工作流运行时获取路径统一（2026-08-24）
+
+- API 列表端点、请求级工作流写用例装配和 `WorkflowChain` 原先分别直接构造
+  `app.workflow.WorkFlowManager` Singleton；现在统一通过 `app.application.workflow` 的 typed runtime
+  provider 获取启动组合根登记的同一实例，不再由消费者自行定位 concrete 管理器。
+- 架构门禁禁止 `app.workflow/**` 实现包和 `app.startup/initializers/workflow.py` 以外的宿主模块直接依赖
+  `app.workflow`。依赖图总边数保持 `6544`：移除 3 条 API/Chain concrete 边，同时新增 Chain 的
+  Application 端口边和 startup 的两条装配边；重点禁止边与自有 SCC 均未增长。
+- 兼容边界不变：`app.workflow.WorkFlowManager` 的类路径、Singleton identity、公开方法、事件监听和
+  action 加载保持原样，旧插件仍可直接使用 concrete 类；本阶段只收口 canonical 宿主消费者。
+
 ### 总体判断
 
 当前架构总体合理，已经从跨层混合的遗留单体收敛为**边界清晰的模块化单体**：
@@ -162,13 +173,16 @@
    配置治理基线进一步区分债务与批准边界：canonical 未批准 Settings 直连和非组合根
    `SystemConfigOper()` 构造均为 `0`；`app/db/base.py`、`engine.py`、`session.py` 的启动前数据库基础设施
    读取及 startup 唯一 Oper 构造点以带理由的固定边界登记。四个集合都只能减少，不能新增或换位置。
+   工作流运行时也已按相同模式收口：API、请求依赖与 Chain 只依赖
+   `app.application.workflow.get_workflow_manager()`；只有工作流实现包和 startup 组合根可直接依赖
+   concrete `WorkFlowManager`，架构测试拒绝宿主重新引入第二条实例获取路径。
 
 ### P2：中长期可演进性债务
 
 - **大型职责域仍偏重。** 代表性热点包括 `app/chain/subscribe.py`（约 `4141` 行）、`app/chain/transfer.py`（约 `2944` 行）、`app/agent/orchestrator.py`（约 `3540` 行）、`app/agent/llm/provider.py`（约 `3529` 行）、`app/adapters/external/market.py`（约 `3139` 行）和 `app/api/endpoints/agent.py`（约 `2489` 行）。复杂度 ratchet 只保证不超过当前基线，不代表这些文件已经易维护。只有在行为快照、调用命中和事务边界明确后，才值得按用例拆分。
-- **类型门禁覆盖面不足。** `mypy.ini` strict 文件清单目前约 `37` 个文件，Agent、Chain、Module、Adapter 大量代码仍依赖动态类型。应从模块契约、生命周期、Repository/Port 和关键 Chain 返回值开始扩展，而不是直接开启全仓 strict。
+- **类型门禁覆盖面不足。** `mypy.ini` strict 文件清单目前为 `39` 个文件，Agent、Chain、Module、Adapter 大量代码仍依赖动态类型。应从模块契约、生命周期、Repository/Port 和关键 Chain 返回值开始扩展，而不是直接开启全仓 strict。
 - **Pylint 仍是增量硬门禁。** `.github/workflows/pylint.yml` 对改动 Python 文件执行硬检查，但全仓报告使用 `|| true` 仅作 advisory。该策略适合存量迁移，却没有形成全仓质量趋势约束；应增加按目录和新增问题数的 ratchet。
-- **测试风格存在历史混用。** 当前约 `499` 个测试文件，仍有约 `70` 个 `unittest.TestCase` 文件。它不是生产架构缺陷，但会增加 fixture、状态隔离和异步测试迁移成本，应在触碰相关模块时渐进迁移。
+- **测试风格存在历史混用。** 当前有 `527` 个测试文件，仍有 `70` 个 `unittest.TestCase` 文件。它不是生产架构缺陷，但会增加 fixture、状态隔离和异步测试迁移成本，应在触碰相关模块时渐进迁移。
 - **跨仓治理链路尚未完全闭环。** 前端已有 lint、typecheck、分片 Vitest 和构建门禁；插件仓有 V1/V2/V3 索引及版本/依赖检查；资源和 Rust 仓有独立构建发布链路。但插件 CI 本地复核因插件仓环境缺少主仓依赖 `httpx2` 无法完成收集，说明“插件仓测试环境与主仓锁定依赖”的可复现性仍需加强。资源构建通过 PR 同步到 `MoviePilot-Resources`，Rust 发布后自动向主仓发依赖 bump PR，链路合理但仍是多仓异步发布，需保留版本 provenance 和回滚点。
 
 ### 已解决、不应重复治理的问题
@@ -219,7 +233,7 @@ MoviePilot V3 当前不是“目录混乱、必须推倒重来”的状态。第
 5. **模块与事件契约登记均已完成。**当前 212 个模块 spec 的宿主观察面已无 legacy aggregation，53 个事件全部绑定 typed payload，可见性、投递等级、错误行为和敏感字段均有基线，legacy event payload 为 `0`。后续重点是保持新增能力 ratchet、观察未知第三方 fallback 命中，以及 6 个 durable-required 事件的真实持久投递，不是重复创建契约或事件 DTO。
 6. **后台副作用缺少统一可靠性定义。**事件队列、APScheduler、FastAPI BackgroundTasks 和线程池任务的丢失、重试、幂等、关停语义各不相同；数据库提交与事件/上报之间仍有进程崩溃窗口。
 7. **核心关联与健康边界已落地，指标导出仍未收口。**HTTP/SSE correlation ID 已传播到线程池、事件、工作流、子进程、外部请求和日志；`/health/live`、`/health/ready` 已由部署入口消费，事件/数据库队列深度及模块/事件耗时使用低基数指标登记。当前缺口是稳定 exporter、运维查询面和跨进程聚合，而不是重新实现 request ID 或健康路由。
-8. **质量门禁已具备增量硬约束，但覆盖面仍需扩大。**push/PR 对变更 Python 文件执行 Pylint，CI 同时运行 host architecture、37 个 strict mypy 文件、复杂度、async 阻塞和 task owner ratchet；全仓 Pylint 仍是 advisory，strict 类型和复杂度拆分仍应随业务切片渐进扩展。
+8. **质量门禁已具备增量硬约束，但覆盖面仍需扩大。**push/PR 对变更 Python 文件执行 Pylint，CI 同时运行 host architecture、39 个 strict mypy 文件、复杂度、async 阻塞和 task owner ratchet；全仓 Pylint 仍是 advisory，strict 类型和复杂度拆分仍应随业务切片渐进扩展。
 
 建议保持**模块化单体**，按以下顺序治理：
 

docs/rules/05-architecture.md

@@ -425,6 +425,7 @@ policy. `app/db` therefore has no dependency on `app/domain`.
 | `chain -> agent implementation` | Forbidden; chains reach Agent runtime only through `app/application/agent.py`; `app/startup/initializers/agent.py` registers lightweight providers at import time, and implementations are materialized only when the capability is enabled or first used |
 | `agent.tools -> api / scheduler / command` | Forbidden; tools use `app/application/plugin/routes.py`, `plugin/folders.py`, `scheduling.py` and `commands.py` application services |
 | `api -> factory` | Forbidden; the FastAPI route adapter is injected into `app/application/plugin/routes.py` by the composition root after creation |
+| `api / chain -> app.workflow` | Forbidden; workflow consumers use `app/application/workflow.py`, while only `app/workflow/**` and `app/startup/initializers/workflow.py` access the concrete runtime |
 | `application -> domain / runtime contract` | Allowed |
 | `application -> DB / Oper / concrete adapter` | Forbidden; define a Protocol in Application and inject an implementation |
 | `db.adapters -> application persistence Protocol / db.oper / UoW` | Allowed; this is dependency inversion, not an upper-layer use-case call |
@@ -446,6 +447,7 @@ policy. `app/db` therefore has no dependency on `app/domain`.
 | `app/application/subscription/write.py` | Subscription media translation and sync/async write-port orchestration |
 | `app/application/scheduling.py` | Runtime scheduler facade for Agent tools and endpoints; `Scheduler` class registered by `app/startup/initializers/scheduler.py` |
 | `app/application/commands.py` | Command registry facade for Agent tools and endpoints; `Command` class registered by `app/startup/initializers/command.py` |
+| `app/application/workflow.py` | Workflow use cases plus the runtime port consumed by API and Chain; `WorkFlowManager` is registered by `app/startup/initializers/workflow.py` |
 | `app/db/adapters/` | SQLAlchemy repository/UoW implementations for Application-owned persistence Protocols |
 | `app/startup/composition/` | HostRuntime, configuration snapshots and cross-layer adapter wiring |
 | `app/startup/initializers/` | Domain-scoped initialization and shutdown hooks |

tests/conftest.py

@@ -145,7 +145,13 @@ def configure_plugin_system_services():
     configure_module_runtime(lambda: ModuleManager())
     from app.application.site.query import SiteQueryService, configure_site_query_service
     from app.application.site.health import SiteHealthService, configure_site_health_service
-    from app.application.workflow import WorkflowQueryService, configure_workflow_query
+    from app.application.workflow import (
+        WorkflowQueryService,
+        configure_workflow_query,
+        configure_workflow_runtime,
+    )
+    from app.workflow import WorkFlowManager
+    configure_workflow_runtime(lambda: WorkFlowManager())
     from app.application.agentdata import configure_agent_data_ports
     from app.db.oper.agentchat import AgentChatOper
     from app.db.oper.downloadfailure import DownloadFailureOper

tests/fixtures/architecture/dependency-baseline.json

@@ -14,7 +14,7 @@
     "workflow_to_db": []
   },
   "edge_count": 6544,
-  "edge_sha256": "44812359c634235d327fdc499bc0ccf93e24923fba24bed4a29fc14adc26bf84",
+  "edge_sha256": "562471ed16680ca403b6cdb7661b20b1200c53b75da5e162efe4e7400cde174b",
   "edges": [
     "app -> app.runtime",
     "app -> app.runtime.compat",
@@ -1637,7 +1637,6 @@
     "app.api.dependencies.workflow -> app.startup",
     "app.api.dependencies.workflow -> app.startup.composition",
     "app.api.dependencies.workflow -> app.startup.composition.context",
-    "app.api.dependencies.workflow -> app.workflow",
     "app.api.deps -> app.api",
     "app.api.deps -> app.api.dependencies",
     "app.api.deps -> app.api.dependencies.agent",
@@ -2417,7 +2416,6 @@
     "app.api.endpoints.workflow -> app.schemas.response",
     "app.api.endpoints.workflow -> app.schemas.types",
     "app.api.endpoints.workflow -> app.schemas.workflow",
-    "app.api.endpoints.workflow -> app.workflow",
     "app.api.response -> app.schemas",
     "app.api.response -> app.schemas.common",
     "app.api.response -> app.schemas.response",
@@ -3480,6 +3478,7 @@
     "app.chain.workflow -> app.application",
     "app.chain.workflow -> app.application.chain",
     "app.chain.workflow -> app.application.chain.data",
+    "app.chain.workflow -> app.application.workflow",
     "app.chain.workflow -> app.chain",
     "app.chain.workflow -> app.runtime",
     "app.chain.workflow -> app.runtime.config",
@@ -3488,7 +3487,6 @@
     "app.chain.workflow -> app.schemas",
     "app.chain.workflow -> app.schemas.types",
     "app.chain.workflow -> app.schemas.workflow",
-    "app.chain.workflow -> app.workflow",
     "app.cli -> app.application",
     "app.cli -> app.application.backup",
     "app.cli -> app.doctor",
@@ -6336,6 +6334,8 @@
     "app.startup.initializers.scheduler -> app.scheduler",
     "app.startup.initializers.transfer -> app.chain",
     "app.startup.initializers.transfer -> app.chain.transfer",
+    "app.startup.initializers.workflow -> app.application",
+    "app.startup.initializers.workflow -> app.application.workflow",
     "app.startup.initializers.workflow -> app.workflow",
     "app.startup.lifecycle -> app.adapters",
     "app.startup.lifecycle -> app.adapters.external",

tests/test_architecture_dependencies.py

@@ -926,6 +926,20 @@ def test_runtime_consumers_use_command_application_facade():
     assert violations == {}
 
 
+def test_runtime_consumers_use_workflow_application_facade():
+    """WorkFlowManager concrete 实现只允许 startup 组合根直接依赖。"""
+    allowed = {"app.startup.initializers.workflow"}
+    violations = {
+        module_name: dependencies & {"app.workflow"}
+        for module_name, dependencies in _build_module_graph().items()
+        if not module_name.startswith("app.workflow")
+        and module_name not in allowed
+        and "app.workflow" in dependencies
+    }
+
+    assert violations == {}
+
+
 def test_modules_read_deployment_settings_through_runtime_port():
     """宿主 Module 不得绕过 runtime 配置端口直接依赖 Settings 实例。"""
     violations: list[str] = []

tests/test_workflow_execution.py

@@ -142,7 +142,7 @@ def test_workflow_executor_preserves_trigger_context(monkeypatch):
         [],
         results={"A": run_action},
     )
-    monkeypatch.setattr(workflow_module, "WorkFlowManager", lambda: fake_manager)
+    monkeypatch.setattr(workflow_module, "get_workflow_manager", lambda: fake_manager)
     monkeypatch.setattr(
         workflow_module.global_vars,
         "workflow_resume",
@@ -189,7 +189,7 @@ def test_workflow_executor_resumes_downstream_nodes(monkeypatch):
         context=_encoded_context(ActionContext()),
     )
 
-    monkeypatch.setattr(workflow_module, "WorkFlowManager", lambda: fake_manager)
+    monkeypatch.setattr(workflow_module, "get_workflow_manager", lambda: fake_manager)
     monkeypatch.setattr(workflow_module.global_vars, "workflow_resume", lambda workflow_id: None)
     monkeypatch.setattr(workflow_module.global_vars, "is_workflow_stopped", lambda workflow_id: False)
 
@@ -214,7 +214,7 @@ def test_workflow_executor_restores_structured_context(monkeypatch):
         },
     )
 
-    monkeypatch.setattr(workflow_module, "WorkFlowManager", lambda: fake_manager)
+    monkeypatch.setattr(workflow_module, "get_workflow_manager", lambda: fake_manager)
     monkeypatch.setattr(workflow_module.global_vars, "workflow_resume", lambda workflow_id: None)
     monkeypatch.setattr(workflow_module.global_vars, "is_workflow_stopped", lambda workflow_id: False)
 
@@ -232,7 +232,7 @@ def test_workflow_executor_reports_incremental_progress(monkeypatch):
     progresses = []
     fake_manager = _FakeWorkflowManager(calls)
 
-    monkeypatch.setattr(workflow_module, "WorkFlowManager", lambda: fake_manager)
+    monkeypatch.setattr(workflow_module, "get_workflow_manager", lambda: fake_manager)
     monkeypatch.setattr(workflow_module.global_vars, "workflow_resume", lambda workflow_id: None)
     monkeypatch.setattr(workflow_module.global_vars, "is_workflow_stopped", lambda workflow_id: False)
 
@@ -272,7 +272,7 @@ def test_workflow_executor_skips_false_condition_branch(monkeypatch):
         ],
     )
 
-    monkeypatch.setattr(workflow_module, "WorkFlowManager", lambda: fake_manager)
+    monkeypatch.setattr(workflow_module, "get_workflow_manager", lambda: fake_manager)
     monkeypatch.setattr(workflow_module.global_vars, "workflow_resume", lambda workflow_id: None)
     monkeypatch.setattr(workflow_module.global_vars, "is_workflow_stopped", lambda workflow_id: False)
 
@@ -325,7 +325,7 @@ def test_workflow_executor_all_success_join_waits_parallel_branches(monkeypatch)
         ],
     )
 
-    monkeypatch.setattr(workflow_module, "WorkFlowManager", lambda: fake_manager)
+    monkeypatch.setattr(workflow_module, "get_workflow_manager", lambda: fake_manager)
     monkeypatch.setattr(workflow_module.global_vars, "workflow_resume", lambda workflow_id: None)
     monkeypatch.setattr(workflow_module.global_vars, "is_workflow_stopped", lambda workflow_id: False)
 
@@ -367,7 +367,7 @@ def test_workflow_executor_any_success_join_runs_after_available_branch(monkeypa
         ],
     )
 
-    monkeypatch.setattr(workflow_module, "WorkFlowManager", lambda: fake_manager)
+    monkeypatch.setattr(workflow_module, "get_workflow_manager", lambda: fake_manager)
     monkeypatch.setattr(workflow_module.global_vars, "workflow_resume", lambda workflow_id: None)
     monkeypatch.setattr(workflow_module.global_vars, "is_workflow_stopped", lambda workflow_id: False)
 
@@ -399,7 +399,7 @@ def test_workflow_executor_all_done_join_can_continue_after_failure(monkeypatch)
         ],
     )
 
-    monkeypatch.setattr(workflow_module, "WorkFlowManager", lambda: fake_manager)
+    monkeypatch.setattr(workflow_module, "get_workflow_manager", lambda: fake_manager)
     monkeypatch.setattr(workflow_module.global_vars, "workflow_resume", lambda workflow_id: None)
     monkeypatch.setattr(workflow_module.global_vars, "is_workflow_stopped", lambda workflow_id: False)
 
@@ -438,7 +438,7 @@ def test_workflow_executor_exclusive_branch_uses_first_matching_flow(monkeypatch
         ],
     )
 
-    monkeypatch.setattr(workflow_module, "WorkFlowManager", lambda: fake_manager)
+    monkeypatch.setattr(workflow_module, "get_workflow_manager", lambda: fake_manager)
     monkeypatch.setattr(workflow_module.global_vars, "workflow_resume", lambda workflow_id: None)
     monkeypatch.setattr(workflow_module.global_vars, "is_workflow_stopped", lambda workflow_id: False)
 
@@ -475,7 +475,7 @@ def test_workflow_executor_passes_declared_inputs(monkeypatch):
         ],
     )
 
-    monkeypatch.setattr(workflow_module, "WorkFlowManager", lambda: fake_manager)
+    monkeypatch.setattr(workflow_module, "get_workflow_manager", lambda: fake_manager)
     monkeypatch.setattr(workflow_module.global_vars, "workflow_resume", lambda workflow_id: None)
     monkeypatch.setattr(workflow_module.global_vars, "is_workflow_stopped", lambda workflow_id: False)
 
@@ -516,7 +516,7 @@ def test_workflow_executor_uses_contract_inputs(monkeypatch):
         ],
     )
 
-    monkeypatch.setattr(workflow_module, "WorkFlowManager", lambda: fake_manager)
+    monkeypatch.setattr(workflow_module, "get_workflow_manager", lambda: fake_manager)
     monkeypatch.setattr(workflow_module.global_vars, "workflow_resume", lambda workflow_id: None)
     monkeypatch.setattr(workflow_module.global_vars, "is_workflow_stopped", lambda workflow_id: False)
 
@@ -543,7 +543,7 @@ def test_workflow_executor_persists_structured_state(monkeypatch):
         }
     )
 
-    monkeypatch.setattr(workflow_module, "WorkFlowManager", lambda: fake_manager)
+    monkeypatch.setattr(workflow_module, "get_workflow_manager", lambda: fake_manager)
     monkeypatch.setattr(workflow_module.global_vars, "workflow_resume", lambda workflow_id: None)
     monkeypatch.setattr(workflow_module.global_vars, "is_workflow_stopped", lambda workflow_id: False)
 
@@ -576,7 +576,7 @@ def test_workflow_executor_restores_outputs_from_execution_state(monkeypatch):
         ],
     )
 
-    monkeypatch.setattr(workflow_module, "WorkFlowManager", lambda: fake_manager)
+    monkeypatch.setattr(workflow_module, "get_workflow_manager", lambda: fake_manager)
     monkeypatch.setattr(workflow_module.global_vars, "workflow_resume", lambda workflow_id: None)
     monkeypatch.setattr(workflow_module.global_vars, "is_workflow_stopped", lambda workflow_id: False)
 
@@ -603,7 +603,7 @@ def test_workflow_executor_keeps_execution_state_dict_for_non_json_leaf(monkeypa
         }
     )
 
-    monkeypatch.setattr(workflow_module, "WorkFlowManager", lambda: fake_manager)
+    monkeypatch.setattr(workflow_module, "get_workflow_manager", lambda: fake_manager)
     monkeypatch.setattr(workflow_module.global_vars, "workflow_resume", lambda workflow_id: None)
     monkeypatch.setattr(workflow_module.global_vars, "is_workflow_stopped", lambda workflow_id: False)
 
@@ -634,7 +634,7 @@ def test_workflow_chain_process_serializes_circular_context(monkeypatch):
     )
     fake_oper = _FakeWorkflowOper(workflow)
 
-    monkeypatch.setattr(workflow_module, "WorkFlowManager", lambda: fake_manager)
+    monkeypatch.setattr(workflow_module, "get_workflow_manager", lambda: fake_manager)
     monkeypatch.setattr(workflow_module, "WorkflowOper", lambda: fake_oper)
     monkeypatch.setattr(workflow_module.global_vars, "workflow_resume", lambda workflow_id: None)
     monkeypatch.setattr(workflow_module.global_vars, "is_workflow_stopped", lambda workflow_id: False)
@@ -683,7 +683,7 @@ def test_workflow_executor_concurrency_key_serializes_parallel_nodes(monkeypatch
         execution_config={"max_workers": 2},
     )
 
-    monkeypatch.setattr(workflow_module, "WorkFlowManager", lambda: fake_manager)
+    monkeypatch.setattr(workflow_module, "get_workflow_manager", lambda: fake_manager)
     monkeypatch.setattr(workflow_module.global_vars, "workflow_resume", lambda workflow_id: None)
     monkeypatch.setattr(workflow_module.global_vars, "is_workflow_stopped", lambda workflow_id: False)
 
@@ -721,7 +721,7 @@ def test_workflow_executor_filter_action_replaces_artifact_outputs(monkeypatch):
         ],
     )
 
-    monkeypatch.setattr(workflow_module, "WorkFlowManager", lambda: fake_manager)
+    monkeypatch.setattr(workflow_module, "get_workflow_manager", lambda: fake_manager)
     monkeypatch.setattr(workflow_module.global_vars, "workflow_resume", lambda workflow_id: None)
     monkeypatch.setattr(workflow_module.global_vars, "is_workflow_stopped", lambda workflow_id: False)
 
@@ -772,7 +772,7 @@ def test_workflow_executor_filter_action_replaces_with_empty_outputs(monkeypatch
         ],
     )
 
-    monkeypatch.setattr(workflow_module, "WorkFlowManager", lambda: fake_manager)
+    monkeypatch.setattr(workflow_module, "get_workflow_manager", lambda: fake_manager)
     monkeypatch.setattr(workflow_module.global_vars, "workflow_resume", lambda workflow_id: None)
     monkeypatch.setattr(workflow_module.global_vars, "is_workflow_stopped", lambda workflow_id: False)
 
@@ -789,7 +789,7 @@ def test_workflow_executor_stop_is_not_success(monkeypatch):
     calls = []
     fake_manager = _FakeWorkflowManager(calls)
 
-    monkeypatch.setattr(workflow_module, "WorkFlowManager", lambda: fake_manager)
+    monkeypatch.setattr(workflow_module, "get_workflow_manager", lambda: fake_manager)
     monkeypatch.setattr(workflow_module.global_vars, "workflow_resume", lambda workflow_id: None)
     monkeypatch.setattr(workflow_module.global_vars, "is_workflow_stopped", lambda workflow_id: True)
 

tests/test_workflow_runtime_facade.py

@@ -0,0 +1,29 @@
+"""工作流运行时 Application Facade 回归。"""
+
+import pytest
+
+from app.application import workflow as workflow_application
+
+
+def test_workflow_runtime_facade_preserves_registered_identity(monkeypatch) -> None:
+    """Application Facade 必须返回组合根登记的同一个运行时对象。"""
+    runtime = object()
+    monkeypatch.setattr(
+        workflow_application,
+        "_workflow_runtime_provider",
+        lambda: runtime,
+    )
+
+    assert workflow_application.get_workflow_manager() is runtime
+
+
+def test_workflow_runtime_facade_fails_before_composition(monkeypatch) -> None:
+    """未装配时不得隐式创建第二个 WorkFlowManager Singleton。"""
+    monkeypatch.setattr(
+        workflow_application,
+        "_workflow_runtime_provider",
+        workflow_application._unconfigured_workflow_runtime,
+    )
+
+    with pytest.raises(RuntimeError, match="工作流运行时尚未由启动组合根装配"):
+        workflow_application.get_workflow_manager()