fix(workflow): route async cache deletion through database worker

app/api/dependencies/workflow.py

@@ -1,6 +1,6 @@
 """工作流领域的请求级 command/query 依赖。"""
 
-from typing import Any, cast
+from typing import cast
 
 from fastapi import Depends
 from sqlalchemy.ext.asyncio import AsyncSession
@@ -10,6 +10,7 @@ from app.adapters.external.server import MoviePilotServerHelper
 from app.api.context import get_async_session, get_host_runtime, get_sync_session
 from app.application.scheduling import Scheduler
 from app.application.workflow import (
+    WorkflowCachePort,
     WorkflowDefinitionCommand,
     WorkflowMutationCommand,
     WorkflowQueryService,
@@ -26,6 +27,7 @@ def get_workflow_mutation_command(
     """组装请求级工作流写用例和提交后的调度副作用。"""
     scheduler = Scheduler()
     workflow_manager = WorkFlowManager()
+    system_config = cast(WorkflowCachePort, runtime.workflow.system_config())
     return WorkflowMutationCommand(
         repository=runtime.workflow.repository(db),
         unit_of_work=runtime.persistence.sync_transaction(db),
@@ -35,9 +37,9 @@ def get_workflow_mutation_command(
         remove_event=workflow_manager.remove_workflow_event,
         refresh_event=workflow_manager.update_workflow_event,
         stop_running=global_vars.stop_workflow,
-        delete_cache=lambda workflow_id: cast(
+        delete_cache=lambda workflow_id: system_config.delete(
-            Any, runtime.workflow.system_config()
+            f"WorkflowCache-{workflow_id}"
-        ).delete(f"WorkflowCache-{workflow_id}"),
+        ),
     )
 
 
@@ -46,13 +48,14 @@ def get_workflow_definition_command(
     runtime: HostRuntime = Depends(get_host_runtime),
 ) -> WorkflowDefinitionCommand:
     """组装工作流创建、复用和重置的异步写用例。"""
+    system_config = cast(WorkflowCachePort, runtime.workflow.system_config())
     return WorkflowDefinitionCommand(
         repository=runtime.workflow.repository(db),
         unit_of_work=runtime.persistence.async_transaction(db),
         stop_running=global_vars.stop_workflow,
-        delete_cache=lambda workflow_id: cast(
+        async_delete_cache=lambda workflow_id: system_config.async_delete(
-            Any, runtime.workflow.system_config()
+            f"WorkflowCache-{workflow_id}"
-        ).delete(f"WorkflowCache-{workflow_id}"),
+        ),
         report_fork=MoviePilotServerHelper.async_workflow_fork_by_id,
     )
 

app/application/configuration.py

@@ -318,6 +318,12 @@ class SystemConfigService:
         result = await self._async_executor.run(partial(self._writer.set, key, value))
         return cast(bool | None, result)
 
+    async def async_delete(self, key: Any) -> Any:
+        """异步删除配置，并等待数据库提交或回滚完成。"""
+        if self._async_executor is None:
+            raise RuntimeError("系统配置异步数据库执行端口尚未配置")
+        return await self._async_executor.run(partial(self._writer.delete, key))
+
     def delete(self, key: Any) -> Any:
         """删除配置。"""
         return self._writer.delete(key)

app/application/workflow.py

@@ -29,6 +29,18 @@ class AsyncWorkflowQueryRepository(Protocol):
         ...
 
 
+class WorkflowCachePort(Protocol):
+    """工作流重置所需的同步与异步配置缓存端口。"""
+
+    def delete(self, key: Any) -> Any:
+        """删除配置缓存。"""
+        ...
+
+    async def async_delete(self, key: Any) -> Any:
+        """通过异步数据库执行端口删除配置缓存。"""
+        ...
+
+
 class WorkflowQueryService:
     """提供工作流列表和详情查询，隔离 API 与数据库会话。"""
 
@@ -364,14 +376,14 @@ class WorkflowDefinitionCommand:
         repository: AsyncWorkflowDefinitionRepository,
         unit_of_work: AsyncUnitOfWork,
         stop_running: Callable[[int], None],
-        delete_cache: Callable[[int], None],
+        async_delete_cache: Callable[[int], Awaitable[Any]],
         report_fork: Optional[Callable[[int], Awaitable[object]]] = None,
     ) -> None:
         """保存异步事务和提交后运行时副作用端口。"""
         self._repository = repository
         self._unit_of_work = unit_of_work
         self._stop_running = stop_running
-        self._delete_cache = delete_cache
+        self._async_delete_cache = async_delete_cache
         self._report_fork = report_fork
 
     async def create(self, payload: Mapping[str, Any]) -> WorkflowMutationResult:
@@ -448,7 +460,7 @@ class WorkflowDefinitionCommand:
         await self._repository.stage_reset(workflow_id, reset_count=True)
         await self._commit()
         self._stop_running(workflow_id)
-        self._delete_cache(workflow_id)
+        await self._async_delete_cache(workflow_id)
         return WorkflowMutationResult(True)
 
     async def _commit(self) -> None:

app/startup/context.py

@@ -1,6 +1,6 @@
 """宿主启动阶段构建的类型化运行时上下文。"""
 
-from collections.abc import AsyncGenerator, Generator
+from collections.abc import AsyncGenerator, Callable, Generator
 from dataclasses import dataclass
 from typing import Protocol
 
@@ -16,6 +16,7 @@ from app.application.subscription.mutation import (
     SubscriptionHistoryMutationRepository,
     SubscriptionMutationRepository,
 )
+from app.application.workflow import WorkflowCachePort
 
 
 class AgentChatRepositoryFactory(Protocol):
@@ -162,7 +163,7 @@ class WorkflowRuntime:
     """工作流定义、状态与缓存操作所需的数据工厂。"""
 
     repository: RepositoryFactory
-    system_config: StandaloneRepositoryFactory
+    system_config: Callable[[], WorkflowCachePort]
 
 
 @dataclass(frozen=True, slots=True)

app/startup/modules_initializer.py

@@ -671,7 +671,7 @@ async def init_modules() -> HostRuntime:
         ),
         workflow=WorkflowRuntime(
             repository=WorkflowOper,
-            system_config=SystemConfigOper,
+            system_config=get_configured_system_config,
         ),
         configuration=runtime_configuration,
         settings=runtime_settings,

tests/fixtures/architecture/dependency-baseline.json

@@ -13,8 +13,8 @@
     "runtime_to_db": [],
     "workflow_to_db": []
   },
-  "edge_count": 6434,
+  "edge_count": 6435,
-  "edge_sha256": "5c47cae41f5d90e1030757d0cb47db8eed373424dda5db17c900271e1d03a9c8",
+  "edge_sha256": "ed079837faf943d744ba7c6dee9172449bc204a2068a89db75e030e0be12fbb2",
   "edges": [
     "app -> app.runtime",
     "app -> app.runtime.compat",
@@ -5979,6 +5979,7 @@
     "app.startup.context -> app.application.subscription.delete",
     "app.startup.context -> app.application.subscription.identity",
     "app.startup.context -> app.application.subscription.mutation",
+    "app.startup.context -> app.application.workflow",
     "app.startup.database -> app.adapters",
     "app.startup.database -> app.adapters.system",
     "app.startup.database -> app.adapters.system.backup",

tests/test_agent_lifecycle.py

@@ -139,7 +139,10 @@ async def test_agent_initialization_failure_does_not_stop_module_startup(
     monkeypatch.setattr(modules_initializer, "check_auth", check_auth)
 
     try:
-        await modules_initializer.init_modules()
+        runtime = await modules_initializer.init_modules()
+        assert runtime.workflow.system_config() is (
+            modules_initializer.get_configured_system_config()
+        )
     finally:
         await modules_initializer.stop_database_worker()
 

tests/test_configuration_ports.py

@@ -88,6 +88,7 @@ def test_system_config_service_supports_separate_reader_and_writer() -> None:
     reader.get.return_value = "old"
     writer = MagicMock()
     writer.set.return_value = True
+    writer.delete.return_value = True
     service = SystemConfigService(
         reader=reader,
         writer=writer,
@@ -98,13 +99,17 @@ def test_system_config_service_supports_separate_reader_and_writer() -> None:
     assert service.set("key", "new") is True
     assert asyncio.run(service.async_set("key", "new")) is True
     service.delete("key")
+    assert asyncio.run(service.async_delete("key")) is True
 
     reader.get.assert_called_once_with("key")
     assert writer.set.call_args_list == [
         (("key", "new"), {}),
         (("key", "new"), {}),
     ]
-    writer.delete.assert_called_once_with("key")
+    assert writer.delete.call_args_list == [
+        (("key",), {}),
+        (("key",), {}),
+    ]
 
 
 def test_user_configuration_service_supports_sync_and_async_writes() -> None:

tests/test_workflow_mutation_command.py

@@ -192,7 +192,7 @@ def _definition_command(*, existing=None, commit_error=None, report_fork=None):
         "repository": repository,
         "unit_of_work": unit_of_work,
         "stop_running": Mock(),
-        "delete_cache": Mock(),
+        "async_delete_cache": AsyncMock(),
         "report_fork": report_fork or AsyncMock(),
     }
     return WorkflowDefinitionCommand(**dependencies), dependencies
@@ -279,7 +279,7 @@ async def test_reset_commit_failure_does_not_stop_runtime_or_delete_cache():
 
     dependencies["unit_of_work"].rollback.assert_awaited_once_with()
     dependencies["stop_running"].assert_not_called()
-    dependencies["delete_cache"].assert_not_called()
+    dependencies["async_delete_cache"].assert_not_awaited()
 
 
 @pytest.mark.asyncio
@@ -289,9 +289,13 @@ async def test_reset_commits_before_runtime_cleanup():
     command, dependencies = _definition_command(existing=_workflow())
     dependencies["unit_of_work"].commit.side_effect = lambda: calls.append("commit")
     dependencies["stop_running"].side_effect = lambda _id: calls.append("stop")
-    dependencies["delete_cache"].side_effect = lambda _id: calls.append("cache")
+    async def delete_cache(_id):
+        calls.append("cache")
+
+    dependencies["async_delete_cache"].side_effect = delete_cache
 
     result = await command.reset(7)
 
     assert result.success is True
     assert calls == ["commit", "stop", "cache"]
+    dependencies["async_delete_cache"].assert_awaited_once_with(7)