diff --git a/app/application/chain/data.py b/app/application/chain/data.py index 92c92ea75..5b736c707 100644 --- a/app/application/chain/data.py +++ b/app/application/chain/data.py @@ -12,8 +12,10 @@ from typing import Any, Optional from app.application.transfer.execution import TransferExecutionRepository from app.application.transfer.workflow import TransferAdmissionRepository +from app.application.workflow import WorkflowExecutionPort OperFactory = Callable[[], Any] +WorkflowExecutionPortFactory = Callable[[], WorkflowExecutionPort] TransferAdmissionRepositoryFactory = Callable[[], TransferAdmissionRepository] TransferExecutionRepositoryFactory = Callable[[], TransferExecutionRepository] @@ -24,7 +26,7 @@ class ChainDataPorts: site: OperFactory subscribe: OperFactory - workflow: OperFactory + workflow: WorkflowExecutionPortFactory download_history: OperFactory transfer_history: OperFactory transfer_pending: TransferAdmissionRepositoryFactory @@ -103,25 +105,33 @@ class UserPortProxy(_ChainDataPortProxy): _ports: Optional[ChainDataPorts] = None -def configure_chain_data_ports(**factories: OperFactory) -> None: - """由启动组合根登记 Chain 的数据端口实现。""" - required = { - "site", - "subscribe", - "workflow", - "download_history", - "transfer_history", - "transfer_pending", - "transfer_execution", - "media_server", - "download_failure", - "user", - } - missing = sorted(required - factories.keys()) - if missing: - raise ValueError(f"Chain 数据端口缺少实现: {', '.join(missing)}") +def configure_chain_data_ports( + *, + site: OperFactory, + subscribe: OperFactory, + workflow: WorkflowExecutionPortFactory, + download_history: OperFactory, + transfer_history: OperFactory, + transfer_pending: TransferAdmissionRepositoryFactory, + transfer_execution: TransferExecutionRepositoryFactory, + media_server: OperFactory, + download_failure: OperFactory, + user: OperFactory, +) -> None: + """由启动组合根登记显式命名的 Chain 数据端口实现。""" global _ports - _ports = ChainDataPorts(**{name: factories[name] for name in required}) + _ports = ChainDataPorts( + site=site, + subscribe=subscribe, + workflow=workflow, + download_history=download_history, + transfer_history=transfer_history, + transfer_pending=transfer_pending, + transfer_execution=transfer_execution, + media_server=media_server, + download_failure=download_failure, + user=user, + ) def get_chain_data_ports() -> ChainDataPorts: @@ -141,8 +151,8 @@ def get_chain_subscribe_port() -> Any: return get_chain_data_ports().subscribe() -def get_chain_workflow_port() -> Any: - """创建工作流数据端口实例。""" +def get_chain_workflow_port() -> WorkflowExecutionPort: + """返回类型化的工作流执行状态事务端口。""" return get_chain_data_ports().workflow() diff --git a/app/application/workflow.py b/app/application/workflow.py index 0e2aea90f..c39ed0a6d 100644 --- a/app/application/workflow.py +++ b/app/application/workflow.py @@ -240,6 +240,56 @@ class UnitOfWork(Protocol): ... +class WorkflowExecutionPort(Protocol): + """工作流 Chain 提交执行状态所需的类型化事务端口。""" + + def start(self, workflow_id: int) -> bool: + """提交工作流运行中状态。""" + ... + + def success( + self, + workflow_id: int, + result: Optional[str] = None, + ) -> bool: + """提交工作流成功状态。""" + ... + + def fail(self, workflow_id: int, result: str) -> bool: + """提交工作流失败状态。""" + ... + + def step( + self, + workflow_id: int, + action_id: str, + context: dict[str, Any], + execution_state: Optional[dict[str, Any]] = None, + ) -> bool: + """提交工作流动作进度。""" + ... + + def reset(self, workflow_id: int, reset_count: bool = False) -> bool: + """提交工作流执行状态重置。""" + ... + + +_configured_workflow_execution: Optional[WorkflowExecutionPort] = None + + +def configure_workflow_execution(service: WorkflowExecutionPort) -> None: + """由启动组合根登记唯一工作流执行状态事务服务。""" + global _configured_workflow_execution + _configured_workflow_execution = service + + +def get_configured_workflow_execution() -> WorkflowExecutionPort: + """返回启动阶段登记的工作流执行状态事务服务。""" + if _configured_workflow_execution is None: + raise RuntimeError("工作流执行状态事务服务尚未配置") + return _configured_workflow_execution + + class WorkflowExecutionRepository(Protocol): """工作流执行状态写入所需的最小暂存端口。""" diff --git a/app/chain/workflow.py b/app/chain/workflow.py index dd4193ffb..02198b61c 100644 --- a/app/chain/workflow.py +++ b/app/chain/workflow.py @@ -1286,11 +1286,11 @@ class WorkflowChain(ChainBase): :param from_begin: 是否从头开始，默认为True :param progress_callback: 定时服务进度更新回调 """ - workflowoper = get_chain_workflow_port() + workflow_execution = get_chain_workflow_port() # 重置工作流 if from_begin: - workflowoper.reset(workflow_id) + workflow_execution.reset(workflow_id) # 查询工作流数据 workflow = get_configured_workflow_query().get_sync(workflow_id) @@ -1311,9 +1311,10 @@ class WorkflowChain(ChainBase): completed: bool, ) -> None: """保存动作上下文和结构化执行状态。""" - get_chain_workflow_port().step( + persisted_action_id = (action.id or "") if completed else "" + workflow_execution.step( workflow_id, - action_id=action.id if completed else "", + action_id=persisted_action_id, context=_serialize_workflow_context(context), execution_state=_serialize_workflow_value(execution_state) ) @@ -1348,7 +1349,7 @@ class WorkflowChain(ChainBase): logger.warning("工作流服务正在停止，拒绝执行 %s", workflow.name) return False, executor.errmsg try: - workflowoper.start(workflow_id) + workflow_execution.start(workflow_id) except Exception: executor.abort_before_execute() raise @@ -1360,10 +1361,10 @@ class WorkflowChain(ChainBase): if not executor.success or executor.has_failure: logger.info(f"工作流 {workflow.name} 执行失败：{executor.errmsg}") - workflowoper.fail(workflow_id, result=executor.errmsg) + workflow_execution.fail(workflow_id, result=executor.errmsg) return False, executor.errmsg logger.info(f"工作流 {workflow.name} 执行完成") - workflowoper.success(workflow_id) + workflow_execution.success(workflow_id) if progress_callback: progress_callback(value=100, text=f"工作流 {workflow.name} 执行完成") return True, "" diff --git a/app/db/oper/__init__.py b/app/db/oper/__init__.py index 07794a1cf..b72cb87ac 100644 --- a/app/db/oper/__init__.py +++ b/app/db/oper/__init__.py @@ -36,7 +36,6 @@ if TYPE_CHECKING: from app.db.oper.transfersettlementreceipt import TransferSettlementReceiptOper from app.db.oper.user import UserOper from app.db.oper.userconfig import UserConfigOper - from app.db.oper.workflow import WorkflowOper # 类名 -> 所在子模块。子模块名即实体名，与 app/db/models 对齐。 _OPER_MODULES = { @@ -56,7 +55,6 @@ _OPER_MODULES = { "TransferSettlementReceiptOper": "transfersettlementreceipt", "UserConfigOper": "userconfig", "UserOper": "user", - "WorkflowOper": "workflow", } @@ -99,5 +97,4 @@ __all__ = [ "TransferSettlementReceiptOper", "UserConfigOper", "UserOper", - "WorkflowOper", ] diff --git a/app/db/oper/workflow.py b/app/db/oper/workflow.py index 51334a411..8c2a6725b 100644 --- a/app/db/oper/workflow.py +++ b/app/db/oper/workflow.py @@ -1,4 +1,4 @@ -from typing import List, Mapping, Tuple, Optional, Any, Protocol +from typing import Any, List, Mapping, Optional, Tuple from sqlalchemy import delete as sqlalchemy_delete from sqlalchemy.orm import Session @@ -7,56 +7,6 @@ from app.db.base import DbOper from app.db.models.workflow import Workflow -class WorkflowLegacyWriter(Protocol): - """无显式 Session 的旧 Oper 写入口所需事务服务。""" - - def start(self, workflow_id: int) -> bool: - """提交工作流运行中状态。""" - ... - - def success( - self, - workflow_id: int, - result: Optional[str] = None, - ) -> bool: - """提交工作流成功状态。""" - ... - - def fail(self, workflow_id: int, result: str) -> bool: - """提交工作流失败状态。""" - ... - - def step( - self, - workflow_id: int, - action_id: str, - context: dict[str, Any], - execution_state: Optional[dict[str, Any]] = None, - ) -> bool: - """提交工作流动作进度。""" - ... - - def reset(self, workflow_id: int, reset_count: bool = False) -> bool: - """提交工作流执行状态重置。""" - ... - - -_legacy_writer: Optional[WorkflowLegacyWriter] = None - - -def configure_workflow_legacy_writer(writer: WorkflowLegacyWriter) -> None: - """由启动组合根为旧的无 Session Oper 写入口注入事务服务。""" - global _legacy_writer - _legacy_writer = writer - - -def _get_workflow_legacy_writer() -> WorkflowLegacyWriter: - """返回已装配的兼容事务服务，避免 Oper 自行创建会话。""" - if _legacy_writer is None: - raise RuntimeError("工作流兼容写服务尚未配置") - return _legacy_writer - - class WorkflowOper(DbOper): """ 工作流管理 @@ -193,67 +143,24 @@ class WorkflowOper(DbOper): workflow.run_count = 0 return workflow - def start(self, wid: int) -> bool: - """ - 启动 - """ - if self._db is None: - return _get_workflow_legacy_writer().start(wid) - return self.stage_start(wid) - def stage_start(self, wid: int) -> bool: """在调用方持有的会话中暂存运行中状态。""" if not isinstance(self._db, Session): raise RuntimeError("工作流暂存写入需要调用方提供同步 Session") return Workflow.start(self._db, wid) - def success(self, wid: int, result: Optional[str] = None) -> bool: - """ - 成功 - """ - if self._db is None: - return _get_workflow_legacy_writer().success(wid, result) - return self.stage_success(wid, result) - def stage_success(self, wid: int, result: Optional[str] = None) -> bool: """在调用方持有的会话中暂存成功状态。""" if not isinstance(self._db, Session): raise RuntimeError("工作流暂存写入需要调用方提供同步 Session") return Workflow.success(self._db, wid, result) - def fail(self, wid: int, result: str) -> bool: - """ - 失败 - """ - if self._db is None: - return _get_workflow_legacy_writer().fail(wid, result) - return self.stage_fail(wid, result) - def stage_fail(self, wid: int, result: str) -> bool: """在调用方持有的会话中暂存失败状态。""" if not isinstance(self._db, Session): raise RuntimeError("工作流暂存写入需要调用方提供同步 Session") return Workflow.fail(self._db, wid, result) - def step( - self, - wid: int, - action_id: str, - context: dict[str, Any], - execution_state: Optional[dict[str, Any]] = None, - ) -> bool: - """ - 步进 - """ - if self._db is None: - return _get_workflow_legacy_writer().step( - wid, - action_id, - context, - execution_state, - ) - return self.stage_step(wid, action_id, context, execution_state) - def stage_step( self, wid: int, @@ -272,14 +179,6 @@ class WorkflowOper(DbOper): execution_state ) - def reset(self, wid: int, reset_count: bool = False) -> bool: - """ - 重置 - """ - if self._db is None: - return _get_workflow_legacy_writer().reset(wid, reset_count) - return self.stage_execution_reset(wid, reset_count) - def stage_execution_reset( self, wid: int, diff --git a/app/runtime/compat/manifest.py b/app/runtime/compat/manifest.py index 01a4791cd..549779d17 100644 --- a/app/runtime/compat/manifest.py +++ b/app/runtime/compat/manifest.py @@ -171,10 +171,10 @@ MODULE_ALIASES: Dict[str, ModuleAlias] = { owner="db", ), "app.db.workflow_oper": ModuleAlias( - target="app.db.oper.workflow", - replacement="app.db.oper.workflow", + target="app.sdk._legacy.workflow", + replacement="app.application.workflow.WorkflowExecutionPort", introduced="v3.0.0", - owner="db", + owner="sdk", ), "app.utils.crypto": ModuleAlias( target="app.foundation.crypto", @@ -774,6 +774,13 @@ _MESSAGE_NOTIFICATION_SYMBOL_ALIASES: Dict[str, SymbolAlias] = { } SYMBOL_ALIASES: Dict[str, Dict[str, SymbolAlias]] = { + "app.db.oper": { + "WorkflowOper": SymbolAlias( + target_module="app.sdk._legacy.workflow", + target_name="WorkflowOper", + replacement="app.application.workflow.WorkflowExecutionPort", + ), + }, "app.workflow": { "WorkFlowManager": SymbolAlias( target_module="app.workflow", diff --git a/app/sdk/_legacy/workflow.py b/app/sdk/_legacy/workflow.py new file mode 100644 index 000000000..e47e43dc1 --- /dev/null +++ b/app/sdk/_legacy/workflow.py @@ -0,0 +1,54 @@ +"""保留旧工作流 Oper 的无 Session 执行状态写入契约。""" + +from typing import Any, Optional + +from app.application.workflow import get_configured_workflow_execution +from app.db.oper.workflow import WorkflowOper as CanonicalWorkflowOper + + +class WorkflowOper(CanonicalWorkflowOper): + """继承显式 Session 查询，并保留旧执行状态写入方法。""" + + def start(self, wid: int) -> bool: + """按旧签名提交工作流运行中状态。""" + if self._db is None: + return get_configured_workflow_execution().start(wid) + return self.stage_start(wid) + + def success(self, wid: int, result: Optional[str] = None) -> bool: + """按旧签名提交工作流成功状态。""" + if self._db is None: + return get_configured_workflow_execution().success(wid, result) + return self.stage_success(wid, result) + + def fail(self, wid: int, result: str) -> bool: + """按旧签名提交工作流失败状态。""" + if self._db is None: + return get_configured_workflow_execution().fail(wid, result) + return self.stage_fail(wid, result) + + def step( + self, + wid: int, + action_id: str, + context: dict[str, Any], + execution_state: Optional[dict[str, Any]] = None, + ) -> bool: + """按旧签名提交工作流动作进度。""" + if self._db is None: + return get_configured_workflow_execution().step( + wid, + action_id, + context, + execution_state, + ) + return self.stage_step(wid, action_id, context, execution_state) + + def reset(self, wid: int, reset_count: bool = False) -> bool: + """按旧签名提交工作流执行状态重置。""" + if self._db is None: + return get_configured_workflow_execution().reset(wid, reset_count) + return self.stage_execution_reset(wid, reset_count) + + +__all__ = ["WorkflowOper"] diff --git a/app/startup/initializers/modules.py b/app/startup/initializers/modules.py index bf5d74509..34756f504 100644 --- a/app/startup/initializers/modules.py +++ b/app/startup/initializers/modules.py @@ -102,7 +102,11 @@ from app.application.service import configure_service_directory from app.application.site.health import SiteHealthService, configure_site_health_service from app.application.site.query import SiteQueryService, configure_site_query_service from app.application.subscription.write import configure_subscribe_writer -from app.application.workflow import WorkflowQueryService, configure_workflow_query +from app.application.workflow import ( + WorkflowQueryService, + configure_workflow_execution, + configure_workflow_query, +) from app.command import CommandChain from app.db.adapters.chain import TransactionalChainDurableEventWriter from app.db.adapters.download import TransactionalDownloadFailureRepository @@ -132,7 +136,7 @@ from app.db.oper.systemconfig import SystemConfigOper from app.db.oper.transferhistory import TransferHistoryOper from app.db.oper.user import UserOper from app.db.oper.userconfig import UserConfigOper -from app.db.oper.workflow import WorkflowOper, configure_workflow_legacy_writer +from app.db.oper.workflow import WorkflowOper from app.db.session import ( SessionFactory, async_session_scope, @@ -861,14 +865,14 @@ async def init_modules() -> HostRuntime: configure_api_data_runtime(api_data) configure_runtime_data_providers(workflow_query) workflow_execution = TransactionalWorkflowExecutionService(SessionFactory) - configure_workflow_legacy_writer(workflow_execution) + configure_workflow_execution(workflow_execution) configure_chain_data_ports( site=lambda: TransactionalSiteRepository( sync_session=SessionFactory, async_session=async_session_scope, ), subscribe=lambda: SubscribeOper(), - workflow=lambda: WorkflowOper(), + workflow=lambda: workflow_execution, download_history=lambda: DownloadHistoryOper(), transfer_history=lambda: TransferHistoryOper(), transfer_pending=lambda: TransactionalTransferAdmissionRepository( diff --git a/docs/architecture-optimization-checklist.md b/docs/architecture-optimization-checklist.md index 8aa4ceb55..f87161c3a 100644 --- a/docs/architecture-optimization-checklist.md +++ b/docs/architecture-optimization-checklist.md @@ -69,7 +69,7 @@ MoviePilot V3 已经形成较清晰的模块化单体：`foundation`、`domain` | 指标 | 当前值 | 解释 | |---|---:|---| -| 宿主 Python 模块 / 内部依赖边 | 844 / 6,901 | `dependency-baseline.json` 当前快照 | +| 宿主 Python 模块 / 内部依赖边 | 845 / 6,907 | `dependency-baseline.json` 当前快照 | | 非平凡 SCC | 2 | 新增 Chain 包根环；另一个是隔离的 29 模块 TMDB 移植包环 | | 跨层 DB 边界债务 | 0 | Application、Chain、API、Agent、Runtime、Workflow 到 DB 的受控债务均为零 | | Model/Oper 事务债务 | 0 | 自建 Session、自动事务装饰器、直接 commit/rollback 等基线均为零 | @@ -78,7 +78,7 @@ MoviePilot V3 已经形成较清晰的模块化单体：`foundation`、`domain` | Python 源码量 | 约 271,400 行 | 60 个文件超过 1,000 行，14 个超过 2,000 行 | | 长方法 | 281 个超过 80 行 | 67 个超过 150 行，23 个超过 250 行；大量是私有方法 | | 全量 mypy 历史债务 | 11,809 / 596 文件 | strict frontier 当前覆盖 41 个文件，本批迁移路径的类型债务已清零 | -| Ruff 历史诊断 | 881 | 低水位门禁通过，但规则集只覆盖 `E4/E7/E9/F/I` | +| Ruff 历史诊断 | 879 | 低水位门禁通过，但规则集只覆盖 `E4/E7/E9/F/I` | | 覆盖率低水位 | Application 78.78%，Domain 79.29% | Chain、Runtime、Agent、Adapter、Startup 未进入包级覆盖率门禁 | ### 3.3 热点文件 @@ -287,6 +287,8 @@ MoviePilot V3 已经形成较清晰的模块化单体：`foundation`、`domain` 读取方法仍可在调用方 Session 中查询。 - [x] Workflow 查询 Port 在 adapter Session 内映射为冻结 DTO/Projection，API、Agent、Chain、Scheduler、 Workflow runtime 和中心服务分享均不接收 ORM。 +- [x] Workflow 执行写端由 Chain 直连 `WorkflowExecutionPort` 和短 Session/UoW 事务服务；canonical + `WorkflowOper` 只保留显式 Session query/stage，旧无 Session 五方法只存在于 SDK Legacy/Compat。 - [ ] `ChainDataPorts`/`AgentDataPorts` 可暂时保留为兼容聚合器，但字段必须显式、可类型检查。 - [ ] 以一个业务纵切面迁移并验证后，再迁移下一组，禁止一次替换所有 Oper。 - [ ] 增加 AST 门禁，禁止向 `ChainDataPorts`、`AgentDataPorts` 和新的 canonical use-case service diff --git a/docs/architecture-overview.md b/docs/architecture-overview.md index f505fc971..a344f12f5 100644 --- a/docs/architecture-overview.md +++ b/docs/architecture-overview.md @@ -704,8 +704,8 @@ flowchart LR | 指标 | 当前值 | |---|---:| -| Python 模块 | 844 | -| 内部导入边 | 6,901 | +| Python 模块 | 845 | +| 内部导入边 | 6,907 | | 非平凡 SCC | 2（`ARCH-107` 临时 Chain 包根环；精确 containment 的 TMDB 移植包环） | | Direct egress | 66（12 条待迁移债务，54 条精确 containment） | | Module Contract V2 spec | 217（其中 215 个进入 `run_module` 观察面） | diff --git a/docs/architecture-refactor-roadmap.md b/docs/architecture-refactor-roadmap.md index afc9d564f..55d8ba928 100644 --- a/docs/architecture-refactor-roadmap.md +++ b/docs/architecture-refactor-roadmap.md @@ -98,7 +98,15 @@ canonical 主程序；兼容只经统一 Compat/SDK 门面提供。 | S1-L1.4 幂等执行与终态结算 | `VERIFIED` | S1-L1.3 | 文件操作、历史提交和 checkpoint 可重放；唯一 retry owner 生效，未知外部结果进入 `manual_review`，仅完整终态删除 pending | | S1-L1.5 E3 全链收口 | `DELIVERED` | S1-L1.4 | `e9de149db`、`a2e249f20`：崩溃矩阵、3.0.17 升降级、重复回放和插件 ABI 验收完整；旧 fail-open、重复状态与兼容层外旧入口删除；Unit Tests `33092427327`、Pylint `33092427348` 全绿，ARCH-102 债务归零 | | S1-L2 Workflow typed query | `DELIVERED` | S0 | `b4f873654`、`a01a35bcb`：Workflow Application Port 不返回 `Any`/ORM，Session 内投影冻结 DTO，正式调用方全部切换；Unit Tests `33098869736`、Pylint `33098869837` 全绿，覆盖率低水位提升至 Application `78.78%` | -| S1-L3 Chain/Agent typed data ports | `PLANNED` | S1-L2 | `ChainDataPorts`/`AgentDataPorts` 的 raw Oper/`Any` factory 全部清零，兼容调用进入 Legacy 层 | +| S1-L3 Chain/Agent typed data ports | `ACTIVE` | S1-L2 | `ChainDataPorts`/`AgentDataPorts` 的 raw Oper/`Any` factory 全部清零，兼容调用进入 Legacy 层 | +| S1-L3.1 Workflow typed execution | `VERIFIED` | S1-L2 | Chain 直连类型化事务服务且单次执行只取一个 port；canonical Oper 删除旧 writer/无 Session 写方法，旧 ABI 只在 `_legacy/workflow.py` 与 Compat overlay | +| S1-L3.2 Chain registry/DI | `PLANNED` | S1-L3.1 | 显式类型化 factory，删除 PortProxy 与失效的双重注入，构造器注入真实控制调用 | +| S1-L3.3 DownloadFailure/MediaServer | `PLANNED` | S1-L3.2 | 两组窄 DTO/Port/adapter 清零 raw Oper，不跨远端 I/O 持有 Session | +| S1-L3.4 User | `PLANNED` | S1-L3.3 | 认证、偏好与渠道绑定投影冻结快照，User Chain/Agent 不接收 ORM | +| S1-L3.5 History | `PLANNED` | S1-L3.4 | Download/Transfer history 统一 typed query/mutation，删除下载历史双事务 fail-open | +| S1-L3.6 Site | `PLANNED` | S1-L3.5 | 复用 Site query/health，补齐同步 typed command，Session 内完成 DTO 投影 | +| S1-L3.7 Subscription | `PLANNED` | S1-L3.6 | Chain/Workflow/interaction 全部消费 typed query/command；完成后进入 S1-L4 原子事务收口 | +| S1-L3.8 Agent/Transfer locator gate | `PLANNED` | S1-L3.7 | 删除 AgentDataPorts 与 Chain locator 跨层泄漏，AST 门禁确认 canonical 无 raw getter/Oper/Any | | S1-L4 Subscription mutation UoW | `PLANNED` | S1-L3 | Subscription mutation 不跨 Session 传 ORM，正式写路径一个 UoW，旧自动事务入口退出 canonical 路径 | | S1-L5 站点/规则引用原子清理 | `PLANNED` | S1-L4 | SystemConfig+Subscribe 同事务更新，commit 后快照原子发布，并发/故障注入无部分状态 | | S1-L6 Outbox 完成语义 | `PLANNED` | S0 | claim 竞争双发清零；业务提交与 effect pending 可区分；stager/store 分离；handler 幂等与崩溃测试完整 | @@ -144,7 +152,7 @@ canonical 主程序；兼容只经统一 Compat/SDK 门面提供。 | S4-L2 Event strict contract | `PLANNED` | S0-L2.6,S1-L6 | 宿主事件输入/输出按风险 strict，诊断例外只属于第三方插件兼容 | | S4-L3 Complexity v2 | `PLANNED` | S3 | 私有方法、class/file、圈复杂度进入门禁；所有超限通过职责拆分归零 | | S4-L4 全量 mypy 清零 | `PLANNED` | S3,S4-L1,S4-L2 | `mypy-baseline.json` 归零并删除债务接受路径，全宿主 strict 类型通过 | -| S4-L5 Ruff 治理债务清零 | `PLANNED` | S3 | 当前受控 881 条诊断归零，规则集扩展经过独立审查且新增诊断为零 | +| S4-L5 Ruff 治理债务清零 | `PLANNED` | S3 | 当前受控 879 条诊断归零，规则集扩展经过独立审查且新增诊断为零 | | S4-L6 Coverage/并发/质量证据 | `PLANNED` | S3,S4-L1,S4-L2 | 高风险包纳入 coverage；raw concurrency 分类清零；Module Quality 有真实 evidence test | ### S5：Plugin、Agent、Domain、Startup 与最终收口 diff --git a/docs/rules/05-architecture.md b/docs/rules/05-architecture.md index e37dc6722..f1d965d70 100644 --- a/docs/rules/05-architecture.md +++ b/docs/rules/05-architecture.md @@ -233,6 +233,11 @@ manager 停机先封口新执行并向活动 owner 发送本地取消，再有 嵌套 JSON 深拷贝。API、Agent、Chain、Scheduler、`WorkflowManager` 和中心服务分享不得读取 raw `WorkflowOper` 或把 ORM 带出 Session。旧 `WorkFlowManager` 拼写只由 Compat 符号覆盖承接，不进入 canonical 模块定义或 `__all__`。 +工作流执行状态写入统一依赖 `app.application.workflow.WorkflowExecutionPort`；Chain 在一次执行中只获取 +一个事务端口，并由 `TransactionalWorkflowExecutionService` 为每次状态写入持有短 Session/UoW。 +canonical `app.db.oper.workflow.WorkflowOper` 只提供显式 Session 的 query/stage 方法；旧无 Session +`start/success/fail/step/reset` 仅由 `app.sdk._legacy.workflow` 和精确 Compat 映射承接，且不进入 +`app.db.oper.__all__`。 协程环境文件日志属于有界 E1 观测能力，只允许单一队列 writer；队列满时不得再以无界 executor 形成第二条异步写入路径。日志关闭必须有限等待 writer 与文件处理器，未收敛时 `LoggerManager` 保留原 owner 并让 lifespan 以关闭失败结束，不得先清空引用或用无界 `join()` 掩盖失败。 diff --git a/tests/conftest.py b/tests/conftest.py index 0d498972a..a77c3a651 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -211,6 +211,7 @@ def configure_plugin_system_services(): from app.application.site.query import SiteQueryService, configure_site_query_service from app.application.workflow import ( WorkflowQueryService, + configure_workflow_execution, configure_workflow_query, configure_workflow_runtime, ) @@ -243,7 +244,7 @@ def configure_plugin_system_services(): from app.db.oper.subscribehistory import SubscribeHistoryOper from app.db.oper.transferhistory import TransferHistoryOper from app.db.oper.user import UserOper - from app.db.oper.workflow import WorkflowOper, configure_workflow_legacy_writer + from app.db.oper.workflow import WorkflowOper def create_sync_session() -> Session: """为无显式会话的 Oper 测试入口创建独占同步 Session。""" @@ -258,9 +259,8 @@ def configure_plugin_system_services(): async_=transaction_runner.async_, ) - configure_workflow_legacy_writer( - TransactionalWorkflowExecutionService(SessionFactory) - ) + workflow_execution = TransactionalWorkflowExecutionService(SessionFactory) + configure_workflow_execution(workflow_execution) configure_api_data_ports( sync_session=get_db, @@ -304,7 +304,7 @@ def configure_plugin_system_services(): configure_chain_data_ports( site=site_repository, subscribe=lambda: SubscribeOper(), - workflow=lambda: WorkflowOper(), + workflow=lambda: workflow_execution, download_history=lambda: DownloadHistoryOper(), transfer_history=lambda: TransferHistoryOper(), transfer_pending=lambda: TransactionalTransferAdmissionRepository( diff --git a/tests/fixtures/architecture/dependency-baseline.json b/tests/fixtures/architecture/dependency-baseline.json index 2fac978e0..ac1404872 100644 --- a/tests/fixtures/architecture/dependency-baseline.json +++ b/tests/fixtures/architecture/dependency-baseline.json @@ -1441,8 +1441,8 @@ "runtime_only": true } }, - "edge_count": 6901, - "edge_sha256": "860590c25bd889096c9faa04f35ad3e3312e2c28ab415e57dce1d669d490e0a1", + "edge_count": 6907, + "edge_sha256": "8b3bb489cfc9573e377d0ee21337633b842b9969fc69fe99b86f547613b86a2a", "edges": [ "app -> app.runtime", "app -> app.runtime.compat", @@ -3988,6 +3988,7 @@ "app.application.chain.data -> app.application.transfer", "app.application.chain.data -> app.application.transfer.execution", "app.application.chain.data -> app.application.transfer.workflow", + "app.application.chain.data -> app.application.workflow", "app.application.chain.events -> app.application", "app.application.chain.events -> app.application.history", "app.application.chain.events -> app.application.transfer", @@ -7700,6 +7701,11 @@ "app.sdk._legacy.user -> app.db", "app.sdk._legacy.user -> app.db.oper", "app.sdk._legacy.user -> app.db.oper.user", + "app.sdk._legacy.workflow -> app.application", + "app.sdk._legacy.workflow -> app.application.workflow", + "app.sdk._legacy.workflow -> app.db", + "app.sdk._legacy.workflow -> app.db.oper", + "app.sdk._legacy.workflow -> app.db.oper.workflow", "app.sdk.browser -> app.adapters", "app.sdk.browser -> app.adapters.network", "app.sdk.browser -> app.adapters.network.browser", @@ -8346,7 +8352,7 @@ "app.workflow.actions.transfer_file -> app.workflow", "app.workflow.actions.transfer_file -> app.workflow.actions" ], - "module_count": 844, + "module_count": 845, "modules": [ "app", "app.adapters", @@ -9136,6 +9142,7 @@ "app.sdk._legacy.transfer", "app.sdk._legacy.transferpending", "app.sdk._legacy.user", + "app.sdk._legacy.workflow", "app.sdk.browser", "app.sdk.cache", "app.sdk.config", diff --git a/tests/fixtures/architecture/ruff-baseline.json b/tests/fixtures/architecture/ruff-baseline.json index d0311fd57..4290e7dad 100644 --- a/tests/fixtures/architecture/ruff-baseline.json +++ b/tests/fixtures/architecture/ruff-baseline.json @@ -479,9 +479,6 @@ "app/db/oper/userconfig.py": { "I001": 1 }, - "app/db/oper/workflow.py": { - "I001": 1 - }, "app/db/session.py": { "I001": 1 }, @@ -1263,9 +1260,6 @@ "tests/test_db_lazy_engine.py": { "I001": 1 }, - "tests/test_db_oper_layer.py": { - "I001": 1 - }, "tests/test_db_oper_layer_extra.py": { "I001": 1 }, diff --git a/tests/fixtures/architecture/runtime-contract-baseline.json b/tests/fixtures/architecture/runtime-contract-baseline.json index b42dfb21c..76aa6bbbb 100644 --- a/tests/fixtures/architecture/runtime-contract-baseline.json +++ b/tests/fixtures/architecture/runtime-contract-baseline.json @@ -284,9 +284,9 @@ "app.db.workflow_oper": { "introduced": "v3.0.0", "is_package": false, - "owner": "db", - "replacement": "app.db.oper.workflow", - "target": "app.db.oper.workflow" + "owner": "sdk", + "replacement": "app.application.workflow.WorkflowExecutionPort", + "target": "app.sdk._legacy.workflow" }, "app.domain.string": { "introduced": "v3.0.0", @@ -926,6 +926,13 @@ "target_name": "MediaInteractionChain" } }, + "app.db.oper": { + "WorkflowOper": { + "replacement": "app.application.workflow.WorkflowExecutionPort", + "target_module": "app.sdk._legacy.workflow", + "target_name": "WorkflowOper" + } + }, "app.domain.media": { "MEDIA_SOURCE_ALIASES": { "replacement": "app.schemas.media.MEDIA_SOURCE_ALIASES", diff --git a/tests/test_architecture_dependencies.py b/tests/test_architecture_dependencies.py index 9e9d7c9c4..aa0026160 100644 --- a/tests/test_architecture_dependencies.py +++ b/tests/test_architecture_dependencies.py @@ -347,6 +347,94 @@ def test_workflow_query_adapter_owns_projection_sessions(): assert "_project_workflow(record)" in source +def test_workflow_execution_chain_uses_typed_transaction_port(): + """工作流 Chain 写端不得再经过 raw Oper 或重复获取全局端口。""" + 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", + } + + data_path = APP_ROOT / "application" / "chain" / "data.py" + data_tree = ast.parse( + data_path.read_text(encoding="utf-8"), + filename=str(data_path), + ) + data_class = next( + node + for node in data_tree.body + if isinstance(node, ast.ClassDef) and node.name == "ChainDataPorts" + ) + workflow_field = next( + node + for node in data_class.body + if isinstance(node, ast.AnnAssign) + and isinstance(node.target, ast.Name) + and node.target.id == "workflow" + ) + workflow_getter = next( + node + for node in data_tree.body + if isinstance(node, ast.FunctionDef) + and node.name == "get_chain_workflow_port" + ) + assert ast.unparse(workflow_field.annotation) == "WorkflowExecutionPortFactory" + assert ast.unparse(workflow_getter.returns) == "WorkflowExecutionPort" + + 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") + assert chain_source.count("get_chain_workflow_port()") == 1 + assert "workflow=lambda: workflow_execution" in startup_source + assert "workflow=lambda: WorkflowOper()" not in startup_source + + +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 数据聚合器不得重新暴露无类型工作流读取入口。""" source = (APP_ROOT / "application" / "agentdata.py").read_text( diff --git a/tests/test_db_oper_layer.py b/tests/test_db_oper_layer.py index 5010b4d1d..674019740 100644 --- a/tests/test_db_oper_layer.py +++ b/tests/test_db_oper_layer.py @@ -6,12 +6,11 @@ Oper 层大多是模型方法的薄封装，但薄封装恰恰是最容易出错 验证 Oper 的对外契约，而不是验证它调了哪个模型方法。 """ import asyncio +import importlib from unittest.mock import Mock import pytest -from app.db.oper.downloadhistory import DownloadHistoryOper -from app.db.oper.mediaserver import MediaServerOper from app.db.models.downloadhistory import DownloadFiles, DownloadHistory from app.db.models.mediaserver import MediaServerItem from app.db.models.plugindata import PluginData @@ -22,6 +21,8 @@ from app.db.models.siteuserdata import SiteUserData from app.db.models.user import User from app.db.models.userconfig import UserConfig from app.db.models.workflow import Workflow +from app.db.oper.downloadhistory import DownloadHistoryOper +from app.db.oper.mediaserver import MediaServerOper from app.db.oper.plugindata import PluginDataOper from app.db.oper.site import SiteOper from app.db.oper.user import UserOper @@ -374,7 +375,7 @@ def test_workflow_oper_add_rejects_duplicate_name(db): assert oper.add(**_workflow_kwargs("op-wf")) == (False, "工作流已存在") -def test_workflow_oper_exposes_lists_and_lifecycle(db): +def test_workflow_oper_exposes_lists_and_staged_lifecycle(db): """ 列表入口与生命周期方法都应透传到模型并落库。 """ @@ -387,23 +388,24 @@ def test_workflow_oper_exposes_lists_and_lifecycle(db): assert {w.name for w in oper.list_enabled()} >= {"op-wf-life"} assert {w.name for w in oper.get_timer_triggered_workflows()} >= {"op-wf-life"} - oper.start(flow.id) + oper.stage_start(flow.id) assert oper.get(flow.id).state == "R" - oper.step(flow.id, "a1", {"n": 1}) + oper.stage_step(flow.id, "a1", {"n": 1}) assert oper.get(flow.id).current_action == "a1" - oper.success(flow.id, "完成") + oper.stage_success(flow.id, "完成") assert oper.get(flow.id).state == "S" - oper.fail(flow.id, "出错") + oper.stage_fail(flow.id, "出错") assert oper.get(flow.id).state == "F" - oper.reset(flow.id, reset_count=True) + oper.stage_execution_reset(flow.id, reset_count=True) assert (oper.get(flow.id).state, oper.get(flow.id).run_count) == ("W", 0) def test_workflow_oper_no_session_uses_configured_uow_writer(db): """旧的无 Session Oper 写入口仍可用，但事务由组合根服务持有。""" flow = db.add(Workflow(**_workflow_kwargs("op-wf-legacy"))) + legacy = importlib.import_module("app.db.workflow_oper") - assert WorkflowOper().start(flow.id) is True + assert legacy.WorkflowOper().start(flow.id) is True db.session.expire_all() assert WorkflowOper(db=db.session).get(flow.id).state == "R" diff --git a/tests/test_legacy_db_behavior_compat.py b/tests/test_legacy_db_behavior_compat.py index ccbedf474..a257d28af 100644 --- a/tests/test_legacy_db_behavior_compat.py +++ b/tests/test_legacy_db_behavior_compat.py @@ -7,6 +7,62 @@ from app.db.models.transferhistory import TransferHistory from app.schemas.file import FileItem +def test_legacy_workflow_writes_delegate_to_configured_execution_port(monkeypatch): + """旧 WorkflowOper 无 Session 写入必须完整委托类型化事务端口。""" + legacy = importlib.import_module("app.db.workflow_oper") + calls = [] + + class ExecutionPort: + """记录五种旧工作流写入调用。""" + + def start(self, workflow_id): + """记录启动。""" + calls.append(("start", workflow_id)) + return True + + def success(self, workflow_id, result=None): + """记录成功。""" + calls.append(("success", workflow_id, result)) + return True + + def fail(self, workflow_id, result): + """记录失败。""" + calls.append(("fail", workflow_id, result)) + return True + + def step(self, workflow_id, action_id, context, execution_state=None): + """记录步骤。""" + calls.append( + ("step", workflow_id, action_id, context, execution_state) + ) + return True + + def reset(self, workflow_id, reset_count=False): + """记录重置。""" + calls.append(("reset", workflow_id, reset_count)) + return True + + monkeypatch.setattr( + legacy, + "get_configured_workflow_execution", + lambda: ExecutionPort(), + ) + oper = legacy.WorkflowOper() + + assert oper.start(7) is True + assert oper.success(7, "done") is True + assert oper.fail(7, "failed") is True + assert oper.step(7, "A", {"value": 1}, {"runtime": {}}) is True + assert oper.reset(7, reset_count=True) is True + assert calls == [ + ("start", 7), + ("success", 7, "done"), + ("fail", 7, "failed"), + ("step", 7, "A", {"value": 1}, {"runtime": {}}), + ("reset", 7, True), + ] + + def test_legacy_subscribe_add_delegates_to_application_service(monkeypatch): """旧 SubscribeOper.add 应保留 mediainfo 写入签名。""" legacy = importlib.import_module("app.db.subscribe_oper") diff --git a/tests/test_legacy_import_compat.py b/tests/test_legacy_import_compat.py index 3c05b33ef..8f588a47e 100644 --- a/tests/test_legacy_import_compat.py +++ b/tests/test_legacy_import_compat.py @@ -314,6 +314,19 @@ def test_db_refactor_legacy_modules_are_all_registered(): assert expected <= set(MODULE_ALIASES) +def test_workflow_oper_compatibility_is_only_exposed_by_overlay(): + """旧工作流写入口只由 Legacy facade 和精确符号映射提供。""" + legacy = importlib.import_module("app.db.workflow_oper") + canonical = importlib.import_module("app.db.oper.workflow") + oper_package = importlib.import_module("app.db.oper") + + assert MODULE_ALIASES["app.db.workflow_oper"].target == "app.sdk._legacy.workflow" + assert issubclass(legacy.WorkflowOper, canonical.WorkflowOper) + assert legacy.WorkflowOper is not canonical.WorkflowOper + assert oper_package.WorkflowOper is legacy.WorkflowOper + assert "WorkflowOper" not in oper_package.__all__ + + def test_split_user_oper_facade_exports_data_and_auth_contracts(): """旧 user_oper 同时提供 UserOper 与八个认证依赖。""" legacy = importlib.import_module("app.db.user_oper") @@ -469,6 +482,7 @@ def test_plugin_scan_reports_moved_symbol_import(tmp_path: Path): def test_symbol_alias_manifest_covers_all_moved_public_symbols(): """符号级映射清单应覆盖媒体身份、整理工作项、刮削拆分与消息/通知命名统一的旧入口。""" + assert set(SYMBOL_ALIASES["app.db.oper"]) == {"WorkflowOper"} assert set(SYMBOL_ALIASES["app.workflow"]) == {"WorkFlowManager"} assert set(SYMBOL_ALIASES["app.domain.media"]) == { "MEDIA_SOURCE_ALIASES", diff --git a/tests/test_workflow_execution.py b/tests/test_workflow_execution.py index c65b06259..36fcc1480 100644 --- a/tests/test_workflow_execution.py +++ b/tests/test_workflow_execution.py @@ -635,9 +635,15 @@ def test_workflow_chain_process_serializes_circular_context(monkeypatch): flows=[{"id": "flow-end", "source": "A", "target": "END", "animated": True}], ) fake_oper = _FakeWorkflowOper(workflow) + port_calls = [] + + def get_execution_port(): + """记录单次执行获取事务端口的次数。""" + port_calls.append(True) + return fake_oper monkeypatch.setattr(workflow_module, "get_workflow_manager", lambda: fake_manager) - monkeypatch.setattr(workflow_module, "get_chain_workflow_port", lambda: fake_oper) + monkeypatch.setattr(workflow_module, "get_chain_workflow_port", get_execution_port) monkeypatch.setattr( workflow_module, "get_configured_workflow_query", @@ -650,6 +656,7 @@ def test_workflow_chain_process_serializes_circular_context(monkeypatch): assert success is True assert message == "" + assert port_calls == [True] assert fake_oper.succeeded is True saved_workflow_context = fake_oper.steps[-1]["context"]["workflow_context"] saved_self = saved_workflow_context["self"]