From e61b5168ad94b0291ebde3f4cff815aca52334dd Mon Sep 17 00:00:00 2001 From: jxxghp Date: Sat, 22 Aug 2026 12:51:10 +0800 Subject: [PATCH] refactor: move workflow execution writes to uow --- app/application/workflow.py | 116 +++++++++++++++++- app/db/models/workflow.py | 8 +- app/db/oper/workflow.py | 105 +++++++++++++++- app/startup/modules_initializer.py | 5 +- app/startup/workflow.py | 71 +++++++++++ docs/architecture-overview.md | 2 +- .../backend-architecture-next-stage.md | 7 ++ docs/rules/10-data-and-persistent.md | 2 +- mypy.ini | 2 + tests/conftest.py | 7 +- .../architecture/dependency-baseline.json | 14 ++- .../transaction-debt-baseline.json | 34 +---- tests/test_architecture_contract_baseline.py | 4 +- tests/test_db_oper_layer.py | 10 ++ tests/test_db_workflow_queries.py | 2 + tests/test_type_gate.py | 4 +- tests/test_workflow_mutation_command.py | 48 ++++++++ 17 files changed, 389 insertions(+), 52 deletions(-) create mode 100644 app/startup/workflow.py diff --git a/app/application/workflow.py b/app/application/workflow.py index b7879b850..580902b73 100644 --- a/app/application/workflow.py +++ b/app/application/workflow.py @@ -4,7 +4,7 @@ from dataclasses import dataclass import json from collections.abc import Awaitable from datetime import datetime -from typing import Any, Callable, Mapping, Optional, Protocol +from typing import Any, Callable, Mapping, Optional, Protocol, TypeVar WORKFLOW_TRIGGER_TIMER = "timer" @@ -101,6 +101,117 @@ class UnitOfWork(Protocol): ... +class WorkflowExecutionRepository(Protocol): + """工作流执行状态写入所需的最小暂存端口。""" + + def stage_start(self, workflow_id: int) -> bool: + """暂存运行中状态。""" + ... + + def stage_success( + self, + workflow_id: int, + result: Optional[str] = None, + ) -> bool: + """暂存成功状态和执行次数。""" + ... + + def stage_fail(self, workflow_id: int, result: str) -> bool: + """暂存失败状态和错误信息。""" + ... + + def stage_step( + self, + workflow_id: int, + action_id: str, + context: dict[str, Any], + execution_state: Optional[dict[str, Any]] = None, + ) -> bool: + """暂存动作进度和执行上下文。""" + ... + + def stage_execution_reset( + self, + workflow_id: int, + reset_count: bool = False, + ) -> bool: + """暂存执行状态重置。""" + ... + + +_ExecutionResult = TypeVar("_ExecutionResult") + + +class WorkflowExecutionCommand: + """在一个显式 UnitOfWork 中提交单次工作流执行状态变更。""" + + def __init__( + self, + *, + repository: WorkflowExecutionRepository, + unit_of_work: UnitOfWork, + ) -> None: + """保存工作流执行仓储和事务端口。""" + self._repository = repository + self._unit_of_work = unit_of_work + + def start(self, workflow_id: int) -> bool: + """提交工作流运行中状态。""" + return self._commit(lambda: self._repository.stage_start(workflow_id)) + + def success( + self, + workflow_id: int, + result: Optional[str] = None, + ) -> bool: + """提交工作流成功状态。""" + return self._commit( + lambda: self._repository.stage_success(workflow_id, result) + ) + + def fail(self, workflow_id: int, result: str) -> bool: + """提交工作流失败状态。""" + return self._commit( + lambda: self._repository.stage_fail(workflow_id, result) + ) + + def step( + self, + workflow_id: int, + action_id: str, + context: dict[str, Any], + execution_state: Optional[dict[str, Any]] = None, + ) -> bool: + """提交工作流动作进度。""" + return self._commit( + lambda: self._repository.stage_step( + workflow_id, + action_id, + context, + execution_state, + ) + ) + + def reset(self, workflow_id: int, reset_count: bool = False) -> bool: + """提交工作流执行状态重置。""" + return self._commit( + lambda: self._repository.stage_execution_reset( + workflow_id, + reset_count, + ) + ) + + def _commit(self, operation: Callable[[], _ExecutionResult]) -> _ExecutionResult: + """提交暂存操作；失败时回滚并原样传播异常。""" + try: + result = operation() + self._unit_of_work.commit() + return result + except Exception: + self._unit_of_work.rollback() + raise + + class WorkflowMutationCommand: """协调工作流状态、定义、调度和事件注册变更。""" @@ -175,6 +286,9 @@ class WorkflowMutationCommand: values["trigger_type"] = WORKFLOW_TRIGGER_TIMER updated = self._repository.stage_update(workflow_id, values) + if not updated: + self._unit_of_work.rollback() + return WorkflowMutationResult(False, "工作流不存在") self._commit() self._remove_timer(updated) if ( diff --git a/app/db/models/workflow.py b/app/db/models/workflow.py index 4abc3fbc1..90561893e 100644 --- a/app/db/models/workflow.py +++ b/app/db/models/workflow.py @@ -7,7 +7,7 @@ from sqlalchemy.orm import Mapped, mapped_column from sqlalchemy.ext.asyncio import AsyncSession from app.db.base import Base, get_id_column -from app.db.decorators import db_query, db_update, async_db_query, async_db_update +from app.db.decorators import db_query, async_db_query, async_db_update class Workflow(Base): @@ -130,7 +130,6 @@ class Workflow(Base): return result.scalars().first() @classmethod - @db_update def update_state(cls, db, wid: int, state: str): db.execute(update(cls).where(cls.id == wid).values(state=state)) return True @@ -142,7 +141,6 @@ class Workflow(Base): return True @classmethod - @db_update def start(cls, db, wid: int): db.execute(update(cls).where(cls.id == wid).values(state='R')) return True @@ -154,7 +152,6 @@ class Workflow(Base): return True @classmethod - @db_update def fail(cls, db, wid: int, result: str): db.execute(update(cls).where( and_(cls.id == wid, cls.state != "P") @@ -178,7 +175,6 @@ class Workflow(Base): return True @classmethod - @db_update def success(cls, db, wid: int, result: Optional[str] = None): db.execute(update(cls).where( and_(cls.id == wid, cls.state != "P") @@ -204,7 +200,6 @@ class Workflow(Base): return True @classmethod - @db_update def reset(cls, db, wid: int, reset_count: Optional[bool] = False): db.execute(update(cls).where(cls.id == wid).values( state='W', @@ -230,7 +225,6 @@ class Workflow(Base): return True @classmethod - @db_update def update_current_action(cls, db, wid: int, action_id: str, context: dict, execution_state: Optional[dict] = None): workflow = db.execute(select(cls).where(cls.id == wid)).scalars().first() diff --git a/app/db/oper/workflow.py b/app/db/oper/workflow.py index e2439912b..53ef649c4 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 +from typing import List, Mapping, Tuple, Optional, Any, Protocol from sqlalchemy import delete as sqlalchemy_delete @@ -6,6 +6,56 @@ 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): """ 工作流管理 @@ -132,24 +182,65 @@ class WorkflowOper(DbOper): """ 启动 """ + if self._db is None: + return _get_workflow_legacy_writer().start(wid) + return self.stage_start(wid) + + def stage_start(self, wid: int) -> bool: + """在调用方持有的会话中暂存运行中状态。""" 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: + """在调用方持有的会话中暂存成功状态。""" 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: + """在调用方持有的会话中暂存失败状态。""" return Workflow.fail(self._db, wid, result) - def step(self, wid: int, action_id: str, context: dict, execution_state: Optional[dict] = None) -> bool: + 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, + action_id: str, + context: dict[str, Any], + execution_state: Optional[dict[str, Any]] = None, + ) -> bool: + """在调用方持有的会话中暂存动作进度。""" return Workflow.update_current_action( self._db, wid, @@ -162,4 +253,14 @@ class WorkflowOper(DbOper): """ 重置 """ + 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, + reset_count: bool = False, + ) -> bool: + """在调用方持有的会话中暂存执行状态重置。""" return Workflow.reset(self._db, wid, reset_count=reset_count) diff --git a/app/startup/modules_initializer.py b/app/startup/modules_initializer.py index ff07f4bbf..b7ab312ae 100644 --- a/app/startup/modules_initializer.py +++ b/app/startup/modules_initializer.py @@ -96,7 +96,7 @@ from app.db.oper.message import MessageOper from app.db.oper.subscribehistory import SubscribeHistoryOper from app.db.oper.plugindata import PluginDataOper from app.db.oper.systemconfig import SystemConfigOper -from app.db.oper.workflow import WorkflowOper +from app.db.oper.workflow import WorkflowOper, configure_workflow_legacy_writer from app.command import CommandChain from app.schemas.message import Message from app.schemas.message import MessageType @@ -113,6 +113,7 @@ from app.startup.subscription import ( ) from app.startup.chain_events import TransactionalChainDurableEventWriter from app.startup.download_failure import TransactionalDownloadFailureRepository +from app.startup.workflow import TransactionalWorkflowExecutionService from app.startup.context import AgentChatRuntime, HostRuntime, SubscriptionRuntime from app.adapters.web.security.access import set_superuser_token_payload_provider from app.application.security.auth import build_superuser_token_payload @@ -570,6 +571,8 @@ async def init_modules() -> HostRuntime: configure_runtime_configuration(host_runtime.configuration) configure_api_data_runtime(host_runtime.compatibility_api_data) configure_runtime_data_providers() + workflow_execution = TransactionalWorkflowExecutionService(SessionFactory) + configure_workflow_legacy_writer(workflow_execution) configure_chain_data_ports( site=lambda: SiteOper(), subscribe=lambda: SubscribeOper(), diff --git a/app/startup/workflow.py b/app/startup/workflow.py new file mode 100644 index 000000000..5f69b6dc2 --- /dev/null +++ b/app/startup/workflow.py @@ -0,0 +1,71 @@ +"""工作流执行状态事务适配器。""" + +from collections.abc import Callable +from typing import Any, TypeVar + +from sqlalchemy.orm import Session + +from app.application.workflow import WorkflowExecutionCommand +from app.db.oper.workflow import WorkflowOper +from app.db.uow import SqlAlchemyUnitOfWork + + +_Result = TypeVar("_Result") + + +class TransactionalWorkflowExecutionService: + """为每次工作流执行状态写入创建独立短会话和 UnitOfWork。""" + + def __init__(self, session_factory: Callable[[], Session]) -> None: + """保存由启动组合根提供的同步 Session 工厂。""" + self._session_factory = session_factory + + def start(self, workflow_id: int) -> bool: + """以独立事务提交运行中状态。""" + return self._run(lambda command: command.start(workflow_id)) + + def success(self, workflow_id: int, result: str | None = None) -> bool: + """以独立事务提交成功状态。""" + return self._run(lambda command: command.success(workflow_id, result)) + + def fail(self, workflow_id: int, result: str) -> bool: + """以独立事务提交失败状态。""" + return self._run(lambda command: command.fail(workflow_id, result)) + + def step( + self, + workflow_id: int, + action_id: str, + context: dict[str, Any], + execution_state: dict[str, Any] | None = None, + ) -> bool: + """以独立事务提交动作进度。""" + return self._run( + lambda command: command.step( + workflow_id, + action_id, + context, + execution_state, + ) + ) + + def reset(self, workflow_id: int, reset_count: bool = False) -> bool: + """以独立事务提交执行状态重置。""" + return self._run( + lambda command: command.reset(workflow_id, reset_count) + ) + + def _run( + self, + operation: Callable[[WorkflowExecutionCommand], _Result], + ) -> _Result: + """创建短会话并把提交/回滚交给 Application command。""" + session = self._session_factory() + try: + command = WorkflowExecutionCommand( + repository=WorkflowOper(db=session), + unit_of_work=SqlAlchemyUnitOfWork(session), + ) + return operation(command) + finally: + session.close() diff --git a/docs/architecture-overview.md b/docs/architecture-overview.md index 3503fe267..d06da798e 100644 --- a/docs/architecture-overview.md +++ b/docs/architecture-overview.md @@ -370,7 +370,7 @@ flowchart LR 成功后执行。订阅新增样板由 `startup/subscription.py` 创建独占 Session， `application/subscription/write.py` 决定事务与 post-commit 边界，`SubscribeOper.stage_add()` 只查重、`add` 和 `flush`。旧 SDK 显式构造的无会话 Oper 暂留兼容自动短会话，不得被新代码复用。 - `transaction-debt-baseline.json` 将存量 178 个 Model 事务装饰器冻结为只降不增低水位。 + `transaction-debt-baseline.json` 将存量 168 个 Model 事务装饰器冻结为只降不增低水位。 - 站点、历史、工作流、Agent 会话删除和插件数据重置已经形成同构事务切片；对应 Application Command/Service 持有 UoW，Oper 的 `stage_*` 方法只修改当前会话。插件数据重置从 `startup/plugins_initializer.py` 创建独占会话，插件直接使用 `PluginDataOper` 的旧 ABI 仅作兼容。 diff --git a/docs/refactor/backend-architecture-next-stage.md b/docs/refactor/backend-architecture-next-stage.md index 03caa5d61..c1273822a 100644 --- a/docs/refactor/backend-architecture-next-stage.md +++ b/docs/refactor/backend-architecture-next-stage.md @@ -772,6 +772,10 @@ ADR 必须逐个映射当前 Event、BackgroundTasks、Scheduler job、Agent tas - 下载失败冷却切片继续迁移到 `TransactionalDownloadFailureRepository`：Chain 每次读写使用独立短会话， 写成功由显式 `SqlAlchemyUnitOfWork` commit，异常 rollback；`DownloadFailure` 查询和记录方法不再拥有 自动会话/提交装饰器。Model decorator 基线继续从 176 降到 174，Oper 内显式 commit/rollback 仍为 0。 +- Workflow 执行状态切片新增 `WorkflowExecutionCommand` 与短会话事务适配器；运行中、动作进度、成功、 + 失败和重置均由 Application command 显式 commit/rollback。`WorkflowOper()` 的旧方法名、参数和返回值 + 继续可用，无 Session 调用委托组合根服务，显式 Session 调用只暂存；同步 Model 自动提交装饰器移除 6 个， + 事务低水位从 174 降到 168，Oper 仍不创建 Session、也不直接 commit/rollback。 **禁止**：本阶段不引入 Celery、Kafka、RabbitMQ 等新基础设施。 @@ -886,6 +890,9 @@ OTel 初始化只能位于 Startup/Adapter；Domain/Application 只依赖 no-op- **扩展实施记录（2026-08-22）**：mypy 目标运行时更新到 Python 3.14，严格清单扩大到 20 个源文件； 新增纳管配置快照和下载失败事务适配器，仍保持零错误、无全局 ignore。 +Workflow 执行状态 UoW 切片将 `app/application/workflow.py` 与 `app/startup/workflow.py` 纳入 strict 清单， +治理范围扩大到 22 个源文件；事务命令、仓储 Protocol 和短会话适配器保持零错误。 + #### ARCH-271：复杂度和端点预算 ratchet **目标**：阻止大方法继续增长，并让拆分对应真实阶段，而不是机械 helper 化。 diff --git a/docs/rules/10-data-and-persistent.md b/docs/rules/10-data-and-persistent.md index 674063524..2a5a98684 100644 --- a/docs/rules/10-data-and-persistent.md +++ b/docs/rules/10-data-and-persistent.md @@ -84,7 +84,7 @@ Oper classes accept and return persistence values. Turning a `MediaInfo` or ### Transaction ownership ratchet - `tests/fixtures/architecture/transaction-debt-baseline.json` records the - existing Model transaction decorators. The current 178 legacy decorators are + existing Model transaction decorators. The current 168 legacy decorators are migration debt: they may decrease but must never increase or move to a new Model method. - New Model methods must not use `db_query`, `db_update`, `async_db_query`, or diff --git a/mypy.ini b/mypy.ini index 7c00f9e1b..17d67989d 100644 --- a/mypy.ini +++ b/mypy.ini @@ -17,6 +17,7 @@ files = app/runtime/extensions/module/contracts.py, app/application/outbox.py, app/application/configuration.py, + app/application/workflow.py, app/application/chain/context.py, app/application/chain/durable_events.py, app/application/subscription/delete.py, @@ -25,5 +26,6 @@ files = app/startup/context.py, app/startup/chain_events.py, app/startup/download_failure.py, + app/startup/workflow.py, app/api/context.py, app/api/dependencies/subscription.py diff --git a/tests/conftest.py b/tests/conftest.py index 144449dff..a90ef0655 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -108,10 +108,15 @@ def configure_plugin_system_services(): from app.db.oper.transferhistory import TransferHistoryOper from app.db.oper.transferpending import TransferPendingOper from app.db.oper.user import UserOper - from app.db.oper.workflow import WorkflowOper + from app.db.oper.workflow import WorkflowOper, configure_workflow_legacy_writer from app.db.oper.message import MessageOper from app.db.oper.passkey import PassKeyOper from app.startup.subscription import TransactionalSubscribeWriter + from app.startup.workflow import TransactionalWorkflowExecutionService + + configure_workflow_legacy_writer( + TransactionalWorkflowExecutionService(SessionFactory) + ) configure_api_data_ports( sync_session=get_db, diff --git a/tests/fixtures/architecture/dependency-baseline.json b/tests/fixtures/architecture/dependency-baseline.json index d573a70f1..a90584564 100644 --- a/tests/fixtures/architecture/dependency-baseline.json +++ b/tests/fixtures/architecture/dependency-baseline.json @@ -13,8 +13,8 @@ "runtime_to_db": [], "workflow_to_db": [] }, - "edge_count": 6360, - "edge_sha256": "09c923d8b167a889e22829c320e05f8801c62400f672b1386d6524c55b064564", + "edge_count": 6367, + "edge_sha256": "62c413857cd12dbaf4d31e4fad4c6b6c80f7dc747d0d5a74ecb61c44b4996822", "edges": [ "app -> app.runtime", "app -> app.runtime.compat", @@ -6120,6 +6120,7 @@ "app.startup.modules_initializer -> app.startup.managed_resources_initializer", "app.startup.modules_initializer -> app.startup.outbox", "app.startup.modules_initializer -> app.startup.subscription", + "app.startup.modules_initializer -> app.startup.workflow", "app.startup.monitor_initializer -> app.monitor", "app.startup.outbox -> app.application", "app.startup.outbox -> app.application.outbox", @@ -6200,6 +6201,12 @@ "app.startup.subscription -> app.startup.outbox", "app.startup.transfer_initializer -> app.chain", "app.startup.transfer_initializer -> app.chain.transfer", + "app.startup.workflow -> app.application", + "app.startup.workflow -> app.application.workflow", + "app.startup.workflow -> app.db", + "app.startup.workflow -> app.db.oper", + "app.startup.workflow -> app.db.oper.workflow", + "app.startup.workflow -> app.db.uow", "app.startup.workflow_initializer -> app.workflow", "app.testing -> app.testing.stub", "app.testing.bootstrap -> app.application", @@ -6377,7 +6384,7 @@ "app.workflow.actions.transfer_file -> app.workflow", "app.workflow.actions.transfer_file -> app.workflow.actions" ], - "module_count": 789, + "module_count": 790, "modules": [ "app", "app.adapters", @@ -7146,6 +7153,7 @@ "app.startup.scheduler_initializer", "app.startup.subscription", "app.startup.transfer_initializer", + "app.startup.workflow", "app.startup.workflow_initializer", "app.testing", "app.testing.bootstrap", diff --git a/tests/fixtures/architecture/transaction-debt-baseline.json b/tests/fixtures/architecture/transaction-debt-baseline.json index 9182e7bc7..b95014416 100644 --- a/tests/fixtures/architecture/transaction-debt-baseline.json +++ b/tests/fixtures/architecture/transaction-debt-baseline.json @@ -4,9 +4,9 @@ "async_db_query": 49, "async_db_update": 12, "db_query": 74, - "db_update": 39 + "db_update": 33 }, - "count": 174, + "count": 168, "methods": [ { "decorator": "async_db_query", @@ -828,11 +828,6 @@ "file": "app/db/models/workflow.py", "method": "Workflow.async_update_state" }, - { - "decorator": "db_update", - "file": "app/db/models/workflow.py", - "method": "Workflow.fail" - }, { "decorator": "db_query", "file": "app/db/models/workflow.py", @@ -852,31 +847,6 @@ "decorator": "db_query", "file": "app/db/models/workflow.py", "method": "Workflow.get_timer_triggered_workflows" - }, - { - "decorator": "db_update", - "file": "app/db/models/workflow.py", - "method": "Workflow.reset" - }, - { - "decorator": "db_update", - "file": "app/db/models/workflow.py", - "method": "Workflow.start" - }, - { - "decorator": "db_update", - "file": "app/db/models/workflow.py", - "method": "Workflow.success" - }, - { - "decorator": "db_update", - "file": "app/db/models/workflow.py", - "method": "Workflow.update_current_action" - }, - { - "decorator": "db_update", - "file": "app/db/models/workflow.py", - "method": "Workflow.update_state" } ] }, diff --git a/tests/test_architecture_contract_baseline.py b/tests/test_architecture_contract_baseline.py index 8746ac28b..3959b577c 100644 --- a/tests/test_architecture_contract_baseline.py +++ b/tests/test_architecture_contract_baseline.py @@ -125,8 +125,8 @@ def test_transaction_debt_baseline_is_a_model_and_oper_ratchet() -> None: baseline = json.loads(baseline_path.read_text(encoding="utf-8")) assert baseline["schema_version"] == 1 - assert baseline["model_decorators"]["count"] == 174 - assert sum(baseline["model_decorators"]["by_kind"].values()) == 174 + assert baseline["model_decorators"]["count"] == 168 + assert sum(baseline["model_decorators"]["by_kind"].values()) == 168 assert baseline["model_transaction_calls"] == {"count": 0, "calls": []} assert baseline["model_session_factories"] == {"count": 0, "calls": []} assert baseline["oper_transaction_calls"] == {"count": 0, "calls": []} diff --git a/tests/test_db_oper_layer.py b/tests/test_db_oper_layer.py index 09596c738..ef1342643 100644 --- a/tests/test_db_oper_layer.py +++ b/tests/test_db_oper_layer.py @@ -357,6 +357,16 @@ def test_workflow_oper_exposes_lists_and_lifecycle(db): 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"))) + + assert WorkflowOper().start(flow.id) is True + + db.session.expire_all() + assert WorkflowOper(db=db.session).get(flow.id).state == "R" + + def test_workflow_oper_event_list_and_async_accessors(db): """ 事件触发列表与异步访问器同样可用。 diff --git a/tests/test_db_workflow_queries.py b/tests/test_db_workflow_queries.py index 6e1877be2..0981e57a8 100644 --- a/tests/test_db_workflow_queries.py +++ b/tests/test_db_workflow_queries.py @@ -228,6 +228,8 @@ def test_update_current_action_matches_async_twin(db): for action in ("a1", "a2", "a1"): Workflow.update_current_action(db.session, sync_flow.id, action, {}) + # 同步 Model 方法只暂存 SQL；由测试持有的事务边界先提交，避免与异步会话争锁。 + db.session.commit() asyncio.run(Workflow.async_update_current_action( wid=async_flow.id, action_id=action, context={})) diff --git a/tests/test_type_gate.py b/tests/test_type_gate.py index 7cf6e3b3f..35a33a0bb 100644 --- a/tests/test_type_gate.py +++ b/tests/test_type_gate.py @@ -25,8 +25,10 @@ def test_mypy_gate_has_explicit_strict_scope_without_global_ignore() -> None: assert "app/runtime/extensions/module/contracts.py" in governed_files assert "app/startup/context.py" in governed_files assert "app/startup/download_failure.py" in governed_files + assert "app/startup/workflow.py" in governed_files + assert "app/application/workflow.py" in governed_files assert "app/api/context.py" in governed_files - assert len(governed_files) >= 20 + assert len(governed_files) >= 22 assert any(path.startswith("app/domain/") for path in governed_files) assert "ignore_errors" not in MYPY_CONFIG.read_text(encoding="utf-8") diff --git a/tests/test_workflow_mutation_command.py b/tests/test_workflow_mutation_command.py index cba2fc5ac..03f5589c7 100644 --- a/tests/test_workflow_mutation_command.py +++ b/tests/test_workflow_mutation_command.py @@ -5,6 +5,7 @@ import pytest from app.application.workflow import ( WorkflowDefinitionCommand, + WorkflowExecutionCommand, WorkflowMutationCommand, WorkflowQueryService, ) @@ -44,6 +45,53 @@ def _command(workflow=None, commit_error=None): return WorkflowMutationCommand(**dependencies), dependencies +def _execution_command(commit_error=None): + """构造可观察的工作流执行状态事务命令。""" + repository = Mock() + repository.stage_start = Mock(return_value=True) + repository.stage_success = Mock(return_value=True) + repository.stage_fail = Mock(return_value=True) + repository.stage_step = Mock(return_value=True) + repository.stage_execution_reset = Mock(return_value=True) + unit_of_work = Mock() + unit_of_work.commit = Mock(side_effect=commit_error) + unit_of_work.rollback = Mock() + return WorkflowExecutionCommand( + repository=repository, + unit_of_work=unit_of_work, + ), repository, unit_of_work + + +def test_execution_step_is_staged_before_unit_of_work_commit(): + """工作流进度写入必须由应用命令暂存后统一提交。""" + command, repository, unit_of_work = _execution_command() + + result = command.step(7, "action-1", {"value": 1}, {"runtime": {}}) + + assert result is True + repository.stage_step.assert_called_once_with( + 7, + "action-1", + {"value": 1}, + {"runtime": {}}, + ) + unit_of_work.commit.assert_called_once_with() + unit_of_work.rollback.assert_not_called() + + +def test_execution_commit_failure_rolls_back(): + """执行状态提交失败时必须回滚并保留原始异常。""" + command, repository, unit_of_work = _execution_command( + RuntimeError("commit failed") + ) + + with pytest.raises(RuntimeError, match="commit failed"): + command.fail(7, "failed") + + repository.stage_fail.assert_called_once_with(7, "failed") + unit_of_work.rollback.assert_called_once_with() + + @pytest.mark.asyncio async def test_workflow_query_service_delegates_list_and_get_to_repository(): """工作流查询服务只调用读取端口，不持有数据库会话或事务。"""