From ea365ddf16f364e9061183931480dc21e3cef451 Mon Sep 17 00:00:00 2001
From: jxxghp <jxxghp@gmail.com>
Date: Fri, 16 Oct 2026 23:15:37 +0800
Subject: [PATCH] refactor(types): govern workflow oper session contracts

---
 app/db/oper/workflow.py           |  9 +++++++--
 mypy.ini                          |  1 +
 tests/test_db_workflow_queries.py | 17 +++++++++++++++++
 3 files changed, 25 insertions(+), 2 deletions(-)

diff --git a/app/db/oper/workflow.py b/app/db/oper/workflow.py
index e87e02f53..5ea8478f4 100644
--- a/app/db/oper/workflow.py
+++ b/app/db/oper/workflow.py
@@ -2,6 +2,7 @@ from typing import Any, List, Mapping, Optional, Tuple
 
 from sqlalchemy import delete as sqlalchemy_delete
 from sqlalchemy import func, or_, select
+from sqlalchemy.ext.asyncio import AsyncSession
 from sqlalchemy.orm import Session
 
 from app.db.base import DbOper
@@ -47,12 +48,12 @@ class WorkflowOper(DbOper):
     工作流管理
     """
 
-    def add(self, **kwargs) -> Tuple[bool, str]:
+    def add(self, **kwargs: Any) -> Tuple[bool, str]:
         """
         新增工作流
         """
         wf = Workflow(**kwargs)
-        if not self.get_by_name(kwargs.get("name")):
+        if not self.get_by_name(wf.name):
             self._stage_create(wf)
             return True, "新增工作流成功"
         return False, "工作流已存在"
@@ -87,6 +88,8 @@ class WorkflowOper(DbOper):
 
     def stage_delete(self, workflow_id: int) -> None:
         """暂存工作流删除，由请求级 UnitOfWork 统一提交。"""
+        if not isinstance(self._db, Session):
+            raise RuntimeError("工作流暂存写入需要调用方提供同步 Session")
         self._db.execute(
             sqlalchemy_delete(Workflow).where(Workflow.id == workflow_id)
         )
@@ -189,6 +192,8 @@ class WorkflowOper(DbOper):
 
     async def stage_create(self, payload: Mapping[str, Any]) -> Workflow:
         """暂存新工作流，不在操作器内提交事务。"""
+        if not isinstance(self._db, AsyncSession):
+            raise RuntimeError("工作流暂存写入需要调用方提供异步 Session")
         workflow = Workflow(**dict(payload))
         self._db.add(workflow)
         await self._db.flush()
diff --git a/mypy.ini b/mypy.ini
index 4e9e659ef..47c5d8f6c 100644
--- a/mypy.ini
+++ b/mypy.ini
@@ -116,6 +116,7 @@ files =
     app/db/oper/user.py,
     app/db/oper/systemconfig.py,
     app/db/oper/site.py,
+    app/db/oper/workflow.py,
     app/db/models/agentchat.py,
     app/db/models/agenttask.py,
     app/db/models/agenttaskrun.py,
diff --git a/tests/test_db_workflow_queries.py b/tests/test_db_workflow_queries.py
index 80f4a77a2..94d6892a0 100644
--- a/tests/test_db_workflow_queries.py
+++ b/tests/test_db_workflow_queries.py
@@ -94,6 +94,23 @@ def test_workflow_oper_reuses_explicit_query_sessions(db, monkeypatch):
     asyncio.run(check())
 
 
+def test_workflow_staged_writes_require_matching_caller_sessions(db) -> None:
+    """暂存入口拒绝缺失或错配会话，不能静默丢失删除或另开事务。"""
+    with pytest.raises(RuntimeError, match="同步 Session"):
+        WorkflowOper().stage_delete(1)
+
+    async def check() -> None:
+        """分别覆盖同步删除和异步创建的会话类型边界。"""
+        async with async_session_scope() as session:
+            with pytest.raises(RuntimeError, match="同步 Session"):
+                WorkflowOper(session).stage_delete(1)
+        for oper in (WorkflowOper(), WorkflowOper(db.session)):
+            with pytest.raises(RuntimeError, match="异步 Session"):
+                await oper.stage_create({"name": "wrong-session"})
+
+    asyncio.run(check())
+
+
 def test_query_repository_returns_detached_deep_copied_snapshot(db):
     """查询仓储必须在关闭短 Session 前投影，且 JSON 不与 ORM 记录共享。"""
     workflow = _flow("wf-snapshot")