refactor: isolate workflow queries

tests/fixtures/architecture/transaction-debt-baseline.json

@@ -1,12 +1,12 @@
 {
   "model_decorators": {
     "by_kind": {
-      "async_db_query": 18,
+      "async_db_query": 14,
       "async_db_update": 0,
-      "db_query": 20,
+      "db_query": 16,
       "db_update": 0
     },
-    "count": 38,
+    "count": 30,
     "methods": [
       {
         "decorator": "async_db_query",
@@ -157,46 +157,6 @@
         "decorator": "db_query",
         "file": "app/db/models/transferpending.py",
         "method": "TransferPending.list_all"
-      },
-      {
-        "decorator": "async_db_query",
-        "file": "app/db/models/workflow.py",
-        "method": "Workflow.async_get_by_name"
-      },
-      {
-        "decorator": "async_db_query",
-        "file": "app/db/models/workflow.py",
-        "method": "Workflow.async_get_enabled_workflows"
-      },
-      {
-        "decorator": "async_db_query",
-        "file": "app/db/models/workflow.py",
-        "method": "Workflow.async_get_event_triggered_workflows"
-      },
-      {
-        "decorator": "async_db_query",
-        "file": "app/db/models/workflow.py",
-        "method": "Workflow.async_get_timer_triggered_workflows"
-      },
-      {
-        "decorator": "db_query",
-        "file": "app/db/models/workflow.py",
-        "method": "Workflow.get_by_name"
-      },
-      {
-        "decorator": "db_query",
-        "file": "app/db/models/workflow.py",
-        "method": "Workflow.get_enabled_workflows"
-      },
-      {
-        "decorator": "db_query",
-        "file": "app/db/models/workflow.py",
-        "method": "Workflow.get_event_triggered_workflows"
-      },
-      {
-        "decorator": "db_query",
-        "file": "app/db/models/workflow.py",
-        "method": "Workflow.get_timer_triggered_workflows"
       }
     ]
   },

tests/test_architecture_contract_baseline.py

@@ -126,8 +126,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"] == 38
-    assert sum(baseline["model_decorators"]["by_kind"].values()) == 38
+    assert baseline["model_decorators"]["count"] == 30
+    assert sum(baseline["model_decorators"]["by_kind"].values()) == 30
     assert baseline["model_decorators"]["by_kind"]["db_update"] == 0
     assert baseline["model_decorators"]["by_kind"]["async_db_update"] == 0
     assert baseline["model_transaction_calls"] == {"count": 0, "calls": []}

tests/test_db_workflow_queries.py

@@ -9,8 +9,10 @@ import asyncio
 
 import pytest
 
+from app.db import decorators
 from app.db.models.workflow import Workflow
-from app.db.session import async_session_scope
+from app.db.oper.workflow import WorkflowOper
+from app.db.session import SessionFactory, async_session_scope
 
 
 @pytest.fixture(autouse=True)
@@ -58,6 +60,44 @@ def test_list_and_get_by_name_match_async_twins(db):
     assert sync_ids == async_ids
 
 
+def test_workflow_oper_reuses_explicit_query_sessions(db, monkeypatch):
+    """WorkflowOper 绑定显式会话后不得再创建兼容查询会话。"""
+    created = db.add(_flow("wf-explicit-session"))
+    monkeypatch.setattr(
+        decorators,
+        "ScopedSession",
+        lambda: (_ for _ in ()).throw(AssertionError("不应创建额外同步会话")),
+    )
+
+    assert WorkflowOper(db.session).get_by_name(created.name).id == created.id
+
+    async def check() -> None:
+        """验证异步 Oper 同样复用调用方会话。"""
+        async with async_session_scope() as session:
+            monkeypatch.setattr(
+                decorators,
+                "async_session_scope",
+                lambda: (_ for _ in ()).throw(AssertionError("不应创建额外异步会话")),
+            )
+            assert (await WorkflowOper(session).async_get_by_name(created.name)).id == created.id
+
+    asyncio.run(check())
+
+
+def test_workflow_model_legacy_queries_keep_no_session_abi(db, monkeypatch):
+    """旧插件直接调用 Workflow Model 时仍应按签名自动补入短会话。"""
+    created = db.add(_flow("wf-legacy-query"))
+    opened = []
+    monkeypatch.setattr(
+        decorators,
+        "ScopedSession",
+        lambda: (opened.append(True) or SessionFactory()),
+    )
+
+    assert Workflow.get_by_name(name=created.name).id == created.id
+    assert opened == [True]
+
+
 def test_enabled_workflows_exclude_paused(db):
     """
     启用列表排除暂停状态。