refactor: move workflow execution writes to uow

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 (

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()

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)

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(),

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()