GithubBackup/MoviePilot
1
0
Fork 0
mirror of https://github.com/jxxghp/MoviePilot.git synced 2026-08-30 20:54:32 +08:00
Code Issues Packages Projects Releases Wiki Activity

refactor(workflow): isolate legacy execution writes

Browse Source
This commit is contained in:
jxxghp
2026-08-28 02:10:40 +08:00
parent 9a7e87dbe0
commit 17d8be2af2
21 changed files with 385 additions and 173 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/application/chain/data.py
+31 -21
View File
@@ -12,8 +12,10 @@ from typing import Any, Optional
from app.application.transfer.execution import TransferExecutionRepository
from app.application.transfer.workflow import TransferAdmissionRepository
from app.application.workflow import WorkflowExecutionPort
OperFactory = Callable[[], Any]
WorkflowExecutionPortFactory = Callable[[], WorkflowExecutionPort]
TransferAdmissionRepositoryFactory = Callable[[], TransferAdmissionRepository]
TransferExecutionRepositoryFactory = Callable[[], TransferExecutionRepository]
@@ -24,7 +26,7 @@ class ChainDataPorts:
site: OperFactory
subscribe: OperFactory
workflow: OperFactory
workflow: WorkflowExecutionPortFactory
download_history: OperFactory
transfer_history: OperFactory
transfer_pending: TransferAdmissionRepositoryFactory
@@ -103,25 +105,33 @@ class UserPortProxy(_ChainDataPortProxy):
_ports: Optional[ChainDataPorts] = None
def configure_chain_data_ports(**factories: OperFactory) -> None:
"""由启动组合根登记 Chain 的数据端口实现。"""
required = {
"site",
"subscribe",
"workflow",
"download_history",
"transfer_history",
"transfer_pending",
"transfer_execution",
"media_server",
"download_failure",
"user",
}
missing = sorted(required - factories.keys())
if missing:
raise ValueError(f"Chain 数据端口缺少实现: {', '.join(missing)}")
def configure_chain_data_ports(
*,
site: OperFactory,
subscribe: OperFactory,
workflow: WorkflowExecutionPortFactory,
download_history: OperFactory,
transfer_history: OperFactory,
transfer_pending: TransferAdmissionRepositoryFactory,
transfer_execution: TransferExecutionRepositoryFactory,
media_server: OperFactory,
download_failure: OperFactory,
user: OperFactory,
) -> None:
"""由启动组合根登记显式命名的 Chain 数据端口实现。"""
global _ports
_ports = ChainDataPorts(**{name: factories[name] for name in required})
_ports = ChainDataPorts(
site=site,
subscribe=subscribe,
workflow=workflow,
download_history=download_history,
transfer_history=transfer_history,
transfer_pending=transfer_pending,
transfer_execution=transfer_execution,
media_server=media_server,
download_failure=download_failure,
user=user,
)
def get_chain_data_ports() -> ChainDataPorts:
@@ -141,8 +151,8 @@ def get_chain_subscribe_port() -> Any:
return get_chain_data_ports().subscribe()
def get_chain_workflow_port() -> Any:
"""创建工作流数据端口实例。"""
def get_chain_workflow_port() -> WorkflowExecutionPort:
"""返回类型化的工作流执行状态事务端口。"""
return get_chain_data_ports().workflow()
app/application/workflow.py
+50
View File
@@ -240,6 +240,56 @@ class UnitOfWork(Protocol):
...
class WorkflowExecutionPort(Protocol):
"""工作流 Chain 提交执行状态所需的类型化事务端口。"""
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:
"""提交工作流执行状态重置。"""
...
_configured_workflow_execution: Optional[WorkflowExecutionPort] = None
def configure_workflow_execution(service: WorkflowExecutionPort) -> None:
"""由启动组合根登记唯一工作流执行状态事务服务。"""
global _configured_workflow_execution
_configured_workflow_execution = service
def get_configured_workflow_execution() -> WorkflowExecutionPort:
"""返回启动阶段登记的工作流执行状态事务服务。"""
if _configured_workflow_execution is None:
raise RuntimeError("工作流执行状态事务服务尚未配置")
return _configured_workflow_execution
class WorkflowExecutionRepository(Protocol):
"""工作流执行状态写入所需的最小暂存端口。"""
app/chain/workflow.py
+8 -7
View File
@@ -1286,11 +1286,11 @@ class WorkflowChain(ChainBase):
:param from_begin: 是否从头开始，默认为True
:param progress_callback: 定时服务进度更新回调
"""
workflowoper = get_chain_workflow_port()
workflow_execution = get_chain_workflow_port()
# 重置工作流
if from_begin:
workflowoper.reset(workflow_id)
workflow_execution.reset(workflow_id)
# 查询工作流数据
workflow = get_configured_workflow_query().get_sync(workflow_id)
@@ -1311,9 +1311,10 @@ class WorkflowChain(ChainBase):
completed: bool,
) -> None:
"""保存动作上下文和结构化执行状态。"""
get_chain_workflow_port().step(
persisted_action_id = (action.id or "") if completed else ""
workflow_execution.step(
workflow_id,
action_id=action.id if completed else "",
action_id=persisted_action_id,
context=_serialize_workflow_context(context),
execution_state=_serialize_workflow_value(execution_state)
)
@@ -1348,7 +1349,7 @@ class WorkflowChain(ChainBase):
logger.warning("工作流服务正在停止，拒绝执行 %s", workflow.name)
return False, executor.errmsg
try:
workflowoper.start(workflow_id)
workflow_execution.start(workflow_id)
except Exception:
executor.abort_before_execute()
raise
@@ -1360,10 +1361,10 @@ class WorkflowChain(ChainBase):
if not executor.success or executor.has_failure:
logger.info(f"工作流 {workflow.name} 执行失败：{executor.errmsg}")
workflowoper.fail(workflow_id, result=executor.errmsg)
workflow_execution.fail(workflow_id, result=executor.errmsg)
return False, executor.errmsg
logger.info(f"工作流 {workflow.name} 执行完成")
workflowoper.success(workflow_id)
workflow_execution.success(workflow_id)
if progress_callback:
progress_callback(value=100, text=f"工作流 {workflow.name} 执行完成")
return True, ""
app/db/oper/__init__.py
-3
View File
@@ -36,7 +36,6 @@ if TYPE_CHECKING:
from app.db.oper.transfersettlementreceipt import TransferSettlementReceiptOper
from app.db.oper.user import UserOper
from app.db.oper.userconfig import UserConfigOper
from app.db.oper.workflow import WorkflowOper
# 类名 -> 所在子模块。子模块名即实体名，与 app/db/models 对齐。
_OPER_MODULES = {
@@ -56,7 +55,6 @@ _OPER_MODULES = {
"TransferSettlementReceiptOper": "transfersettlementreceipt",
"UserConfigOper": "userconfig",
"UserOper": "user",
"WorkflowOper": "workflow",
}
@@ -99,5 +97,4 @@ __all__ = [
"TransferSettlementReceiptOper",
"UserConfigOper",
"UserOper",
"WorkflowOper",
]
app/db/oper/workflow.py
+1 -102
View File
@@ -1,4 +1,4 @@
from typing import List, Mapping, Tuple, Optional, Any, Protocol
from typing import Any, List, Mapping, Optional, Tuple
from sqlalchemy import delete as sqlalchemy_delete
from sqlalchemy.orm import Session
@@ -7,56 +7,6 @@ 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):
"""
工作流管理
@@ -193,67 +143,24 @@ class WorkflowOper(DbOper):
workflow.run_count = 0
return workflow
def start(self, wid: int) -> bool:
"""
启动
"""
if self._db is None:
return _get_workflow_legacy_writer().start(wid)
return self.stage_start(wid)
def stage_start(self, wid: int) -> bool:
"""在调用方持有的会话中暂存运行中状态。"""
if not isinstance(self._db, Session):
raise RuntimeError("工作流暂存写入需要调用方提供同步 Session")
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:
"""在调用方持有的会话中暂存成功状态。"""
if not isinstance(self._db, Session):
raise RuntimeError("工作流暂存写入需要调用方提供同步 Session")
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:
"""在调用方持有的会话中暂存失败状态。"""
if not isinstance(self._db, Session):
raise RuntimeError("工作流暂存写入需要调用方提供同步 Session")
return Workflow.fail(self._db, wid, result)
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,
@@ -272,14 +179,6 @@ class WorkflowOper(DbOper):
execution_state
)
def reset(self, wid: int, reset_count: bool = False) -> bool:
"""
重置
"""
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,
app/runtime/compat/manifest.py
+10 -3
View File
@@ -171,10 +171,10 @@ MODULE_ALIASES: Dict[str, ModuleAlias] = {
owner="db",
),
"app.db.workflow_oper": ModuleAlias(
target="app.db.oper.workflow",
replacement="app.db.oper.workflow",
target="app.sdk._legacy.workflow",
replacement="app.application.workflow.WorkflowExecutionPort",
introduced="v3.0.0",
owner="db",
owner="sdk",
),
"app.utils.crypto": ModuleAlias(
target="app.foundation.crypto",
@@ -774,6 +774,13 @@ _MESSAGE_NOTIFICATION_SYMBOL_ALIASES: Dict[str, SymbolAlias] = {
}
SYMBOL_ALIASES: Dict[str, Dict[str, SymbolAlias]] = {
"app.db.oper": {
"WorkflowOper": SymbolAlias(
target_module="app.sdk._legacy.workflow",
target_name="WorkflowOper",
replacement="app.application.workflow.WorkflowExecutionPort",
),
},
"app.workflow": {
"WorkFlowManager": SymbolAlias(
target_module="app.workflow",
app/sdk/_legacy/workflow.py
+54
View File
@@ -0,0 +1,54 @@
"""保留旧工作流 Oper 的无 Session 执行状态写入契约。"""
from typing import Any, Optional
from app.application.workflow import get_configured_workflow_execution
from app.db.oper.workflow import WorkflowOper as CanonicalWorkflowOper
class WorkflowOper(CanonicalWorkflowOper):
"""继承显式 Session 查询，并保留旧执行状态写入方法。"""
def start(self, wid: int) -> bool:
"""按旧签名提交工作流运行中状态。"""
if self._db is None:
return get_configured_workflow_execution().start(wid)
return self.stage_start(wid)
def success(self, wid: int, result: Optional[str] = None) -> bool:
"""按旧签名提交工作流成功状态。"""
if self._db is None:
return get_configured_workflow_execution().success(wid, result)
return self.stage_success(wid, result)
def fail(self, wid: int, result: str) -> bool:
"""按旧签名提交工作流失败状态。"""
if self._db is None:
return get_configured_workflow_execution().fail(wid, result)
return self.stage_fail(wid, result)
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_configured_workflow_execution().step(
wid,
action_id,
context,
execution_state,
)
return self.stage_step(wid, action_id, context, execution_state)
def reset(self, wid: int, reset_count: bool = False) -> bool:
"""按旧签名提交工作流执行状态重置。"""
if self._db is None:
return get_configured_workflow_execution().reset(wid, reset_count)
return self.stage_execution_reset(wid, reset_count)
__all__ = ["WorkflowOper"]
app/startup/initializers/modules.py
+8 -4
View File
@@ -102,7 +102,11 @@ from app.application.service import configure_service_directory
from app.application.site.health import SiteHealthService, configure_site_health_service
from app.application.site.query import SiteQueryService, configure_site_query_service
from app.application.subscription.write import configure_subscribe_writer
from app.application.workflow import WorkflowQueryService, configure_workflow_query
from app.application.workflow import (
WorkflowQueryService,
configure_workflow_execution,
configure_workflow_query,
)
from app.command import CommandChain
from app.db.adapters.chain import TransactionalChainDurableEventWriter
from app.db.adapters.download import TransactionalDownloadFailureRepository
@@ -132,7 +136,7 @@ from app.db.oper.systemconfig import SystemConfigOper
from app.db.oper.transferhistory import TransferHistoryOper
from app.db.oper.user import UserOper
from app.db.oper.userconfig import UserConfigOper
from app.db.oper.workflow import WorkflowOper, configure_workflow_legacy_writer
from app.db.oper.workflow import WorkflowOper
from app.db.session import (
SessionFactory,
async_session_scope,
@@ -861,14 +865,14 @@ async def init_modules() -> HostRuntime:
configure_api_data_runtime(api_data)
configure_runtime_data_providers(workflow_query)
workflow_execution = TransactionalWorkflowExecutionService(SessionFactory)
configure_workflow_legacy_writer(workflow_execution)
configure_workflow_execution(workflow_execution)
configure_chain_data_ports(
site=lambda: TransactionalSiteRepository(
sync_session=SessionFactory,
async_session=async_session_scope,
),
subscribe=lambda: SubscribeOper(),
workflow=lambda: WorkflowOper(),
workflow=lambda: workflow_execution,
download_history=lambda: DownloadHistoryOper(),
transfer_history=lambda: TransferHistoryOper(),
transfer_pending=lambda: TransactionalTransferAdmissionRepository(
docs/architecture-optimization-checklist.md
+4 -2
View File
@@ -69,7 +69,7 @@ MoviePilot V3 已经形成较清晰的模块化单体：`foundation`、`domain`
| 指标 | 当前值 | 解释 |
|---|---:|---|
| 宿主 Python 模块 / 内部依赖边 | 844 / 6,901 | `dependency-baseline.json` 当前快照 |
| 宿主 Python 模块 / 内部依赖边 | 845 / 6,907 | `dependency-baseline.json` 当前快照 |
| 非平凡 SCC | 2 | 新增 Chain 包根环；另一个是隔离的 29 模块 TMDB 移植包环 |
| 跨层 DB 边界债务 | 0 | Application、Chain、API、Agent、Runtime、Workflow 到 DB 的受控债务均为零 |
| Model/Oper 事务债务 | 0 | 自建 Session、自动事务装饰器、直接 commit/rollback 等基线均为零 |
@@ -78,7 +78,7 @@ MoviePilot V3 已经形成较清晰的模块化单体：`foundation`、`domain`
| Python 源码量 | 约 271,400 行 | 60 个文件超过 1,000 行，14 个超过 2,000 行 |
| 长方法 | 281 个超过 80 行 | 67 个超过 150 行，23 个超过 250 行；大量是私有方法 |
| 全量 mypy 历史债务 | 11,809 / 596 文件 | strict frontier 当前覆盖 41 个文件，本批迁移路径的类型债务已清零 |
| Ruff 历史诊断 | 881 | 低水位门禁通过，但规则集只覆盖 `E4/E7/E9/F/I` |
| Ruff 历史诊断 | 879 | 低水位门禁通过，但规则集只覆盖 `E4/E7/E9/F/I` |
| 覆盖率低水位 | Application 78.78%，Domain 79.29% | Chain、Runtime、Agent、Adapter、Startup 未进入包级覆盖率门禁 |
### 3.3 热点文件
@@ -287,6 +287,8 @@ MoviePilot V3 已经形成较清晰的模块化单体：`foundation`、`domain`
读取方法仍可在调用方 Session 中查询。
- [x] Workflow 查询 Port 在 adapter Session 内映射为冻结 DTO/Projection，API、Agent、Chain、Scheduler、
Workflow runtime 和中心服务分享均不接收 ORM。
- [x] Workflow 执行写端由 Chain 直连 `WorkflowExecutionPort` 和短 Session/UoW 事务服务；canonical
`WorkflowOper` 只保留显式 Session query/stage，旧无 Session 五方法只存在于 SDK Legacy/Compat。
- [ ] `ChainDataPorts`/`AgentDataPorts` 可暂时保留为兼容聚合器，但字段必须显式、可类型检查。
- [ ] 以一个业务纵切面迁移并验证后，再迁移下一组，禁止一次替换所有 Oper。
- [ ] 增加 AST 门禁，禁止向 `ChainDataPorts`、`AgentDataPorts` 和新的 canonical use-case service
docs/architecture-overview.md
+2 -2
View File
@@ -704,8 +704,8 @@ flowchart LR
| 指标 | 当前值 |
|---|---:|
| Python 模块 | 844 |
| 内部导入边 | 6,901 |
| Python 模块 | 845 |
| 内部导入边 | 6,907 |
| 非平凡 SCC | 2（`ARCH-107` 临时 Chain 包根环；精确 containment 的 TMDB 移植包环） |
| Direct egress | 66（12 条待迁移债务，54 条精确 containment） |
| Module Contract V2 spec | 217（其中 215 个进入 `run_module` 观察面） |
docs/architecture-refactor-roadmap.md
+10 -2
View File
@@ -98,7 +98,15 @@ canonical 主程序；兼容只经统一 Compat/SDK 门面提供。
| S1-L1.4 幂等执行与终态结算 | `VERIFIED` | S1-L1.3 | 文件操作、历史提交和 checkpoint 可重放；唯一 retry owner 生效，未知外部结果进入 `manual_review`，仅完整终态删除 pending |
| S1-L1.5 E3 全链收口 | `DELIVERED` | S1-L1.4 | `e9de149db`、`a2e249f20`：崩溃矩阵、3.0.17 升降级、重复回放和插件 ABI 验收完整；旧 fail-open、重复状态与兼容层外旧入口删除；Unit Tests `33092427327`、Pylint `33092427348` 全绿，ARCH-102 债务归零 |
| S1-L2 Workflow typed query | `DELIVERED` | S0 | `b4f873654`、`a01a35bcb`：Workflow Application Port 不返回 `Any`/ORM，Session 内投影冻结 DTO，正式调用方全部切换；Unit Tests `33098869736`、Pylint `33098869837` 全绿，覆盖率低水位提升至 Application `78.78%` |
| S1-L3 Chain/Agent typed data ports | `PLANNED` | S1-L2 | `ChainDataPorts`/`AgentDataPorts` 的 raw Oper/`Any` factory 全部清零，兼容调用进入 Legacy 层 |
| S1-L3 Chain/Agent typed data ports | `ACTIVE` | S1-L2 | `ChainDataPorts`/`AgentDataPorts` 的 raw Oper/`Any` factory 全部清零，兼容调用进入 Legacy 层 |
| S1-L3.1 Workflow typed execution | `VERIFIED` | S1-L2 | Chain 直连类型化事务服务且单次执行只取一个 port；canonical Oper 删除旧 writer/无 Session 写方法，旧 ABI 只在 `_legacy/workflow.py` 与 Compat overlay |
| S1-L3.2 Chain registry/DI | `PLANNED` | S1-L3.1 | 显式类型化 factory，删除 PortProxy 与失效的双重注入，构造器注入真实控制调用 |
| S1-L3.3 DownloadFailure/MediaServer | `PLANNED` | S1-L3.2 | 两组窄 DTO/Port/adapter 清零 raw Oper，不跨远端 I/O 持有 Session |
| S1-L3.4 User | `PLANNED` | S1-L3.3 | 认证、偏好与渠道绑定投影冻结快照，User Chain/Agent 不接收 ORM |
| S1-L3.5 History | `PLANNED` | S1-L3.4 | Download/Transfer history 统一 typed query/mutation，删除下载历史双事务 fail-open |
| S1-L3.6 Site | `PLANNED` | S1-L3.5 | 复用 Site query/health，补齐同步 typed command，Session 内完成 DTO 投影 |
| S1-L3.7 Subscription | `PLANNED` | S1-L3.6 | Chain/Workflow/interaction 全部消费 typed query/command；完成后进入 S1-L4 原子事务收口 |
| S1-L3.8 Agent/Transfer locator gate | `PLANNED` | S1-L3.7 | 删除 AgentDataPorts 与 Chain locator 跨层泄漏，AST 门禁确认 canonical 无 raw getter/Oper/Any |
| S1-L4 Subscription mutation UoW | `PLANNED` | S1-L3 | Subscription mutation 不跨 Session 传 ORM，正式写路径一个 UoW，旧自动事务入口退出 canonical 路径 |
| S1-L5 站点/规则引用原子清理 | `PLANNED` | S1-L4 | SystemConfig+Subscribe 同事务更新，commit 后快照原子发布，并发/故障注入无部分状态 |
| S1-L6 Outbox 完成语义 | `PLANNED` | S0 | claim 竞争双发清零；业务提交与 effect pending 可区分；stager/store 分离；handler 幂等与崩溃测试完整 |
@@ -144,7 +152,7 @@ canonical 主程序；兼容只经统一 Compat/SDK 门面提供。
| S4-L2 Event strict contract | `PLANNED` | S0-L2.6,S1-L6 | 宿主事件输入/输出按风险 strict，诊断例外只属于第三方插件兼容 |
| S4-L3 Complexity v2 | `PLANNED` | S3 | 私有方法、class/file、圈复杂度进入门禁；所有超限通过职责拆分归零 |
| S4-L4 全量 mypy 清零 | `PLANNED` | S3,S4-L1,S4-L2 | `mypy-baseline.json` 归零并删除债务接受路径，全宿主 strict 类型通过 |
| S4-L5 Ruff 治理债务清零 | `PLANNED` | S3 | 当前受控 881 条诊断归零，规则集扩展经过独立审查且新增诊断为零 |
| S4-L5 Ruff 治理债务清零 | `PLANNED` | S3 | 当前受控 879 条诊断归零，规则集扩展经过独立审查且新增诊断为零 |
| S4-L6 Coverage/并发/质量证据 | `PLANNED` | S3,S4-L1,S4-L2 | 高风险包纳入 coverage；raw concurrency 分类清零；Module Quality 有真实 evidence test |
### S5：Plugin、Agent、Domain、Startup 与最终收口
docs/rules/05-architecture.md
+5
View File
@@ -233,6 +233,11 @@ manager 停机先封口新执行并向活动 owner 发送本地取消，再有
嵌套 JSON 深拷贝。API、Agent、Chain、Scheduler、`WorkflowManager` 和中心服务分享不得读取 raw
`WorkflowOper` 或把 ORM 带出 Session。旧 `WorkFlowManager` 拼写只由 Compat 符号覆盖承接，不进入
canonical 模块定义或 `__all__`。
工作流执行状态写入统一依赖 `app.application.workflow.WorkflowExecutionPort`；Chain 在一次执行中只获取
一个事务端口，并由 `TransactionalWorkflowExecutionService` 为每次状态写入持有短 Session/UoW。
canonical `app.db.oper.workflow.WorkflowOper` 只提供显式 Session 的 query/stage 方法；旧无 Session
`start/success/fail/step/reset` 仅由 `app.sdk._legacy.workflow` 和精确 Compat 映射承接，且不进入
`app.db.oper.__all__`。
协程环境文件日志属于有界 E1 观测能力，只允许单一队列 writer；队列满时不得再以无界 executor
形成第二条异步写入路径。日志关闭必须有限等待 writer 与文件处理器，未收敛时 `LoggerManager`
保留原 owner 并让 lifespan 以关闭失败结束，不得先清空引用或用无界 `join()` 掩盖失败。
tests/conftest.py
+5 -5
View File
@@ -211,6 +211,7 @@ def configure_plugin_system_services():
from app.application.site.query import SiteQueryService, configure_site_query_service
from app.application.workflow import (
WorkflowQueryService,
configure_workflow_execution,
configure_workflow_query,
configure_workflow_runtime,
)
@@ -243,7 +244,7 @@ def configure_plugin_system_services():
from app.db.oper.subscribehistory import SubscribeHistoryOper
from app.db.oper.transferhistory import TransferHistoryOper
from app.db.oper.user import UserOper
from app.db.oper.workflow import WorkflowOper, configure_workflow_legacy_writer
from app.db.oper.workflow import WorkflowOper
def create_sync_session() -> Session:
"""为无显式会话的 Oper 测试入口创建独占同步 Session。"""
@@ -258,9 +259,8 @@ def configure_plugin_system_services():
async_=transaction_runner.async_,
)
configure_workflow_legacy_writer(
TransactionalWorkflowExecutionService(SessionFactory)
)
workflow_execution = TransactionalWorkflowExecutionService(SessionFactory)
configure_workflow_execution(workflow_execution)
configure_api_data_ports(
sync_session=get_db,
@@ -304,7 +304,7 @@ def configure_plugin_system_services():
configure_chain_data_ports(
site=site_repository,
subscribe=lambda: SubscribeOper(),
workflow=lambda: WorkflowOper(),
workflow=lambda: workflow_execution,
download_history=lambda: DownloadHistoryOper(),
transfer_history=lambda: TransferHistoryOper(),
transfer_pending=lambda: TransactionalTransferAdmissionRepository(
tests/fixtures/architecture/dependency-baseline.json
Vendored
+10 -3
View File
@@ -1441,8 +1441,8 @@
"runtime_only": true
}
},
"edge_count": 6901,
"edge_sha256": "860590c25bd889096c9faa04f35ad3e3312e2c28ab415e57dce1d669d490e0a1",
"edge_count": 6907,
"edge_sha256": "8b3bb489cfc9573e377d0ee21337633b842b9969fc69fe99b86f547613b86a2a",
"edges": [
"app -> app.runtime",
"app -> app.runtime.compat",
@@ -3988,6 +3988,7 @@
"app.application.chain.data -> app.application.transfer",
"app.application.chain.data -> app.application.transfer.execution",
"app.application.chain.data -> app.application.transfer.workflow",
"app.application.chain.data -> app.application.workflow",
"app.application.chain.events -> app.application",
"app.application.chain.events -> app.application.history",
"app.application.chain.events -> app.application.transfer",
@@ -7700,6 +7701,11 @@
"app.sdk._legacy.user -> app.db",
"app.sdk._legacy.user -> app.db.oper",
"app.sdk._legacy.user -> app.db.oper.user",
"app.sdk._legacy.workflow -> app.application",
"app.sdk._legacy.workflow -> app.application.workflow",
"app.sdk._legacy.workflow -> app.db",
"app.sdk._legacy.workflow -> app.db.oper",
"app.sdk._legacy.workflow -> app.db.oper.workflow",
"app.sdk.browser -> app.adapters",
"app.sdk.browser -> app.adapters.network",
"app.sdk.browser -> app.adapters.network.browser",
@@ -8346,7 +8352,7 @@
"app.workflow.actions.transfer_file -> app.workflow",
"app.workflow.actions.transfer_file -> app.workflow.actions"
],
"module_count": 844,
"module_count": 845,
"modules": [
"app",
"app.adapters",
@@ -9136,6 +9142,7 @@
"app.sdk._legacy.transfer",
"app.sdk._legacy.transferpending",
"app.sdk._legacy.user",
"app.sdk._legacy.workflow",
"app.sdk.browser",
"app.sdk.cache",
"app.sdk.config",
tests/fixtures/architecture/ruff-baseline.json
Vendored
-6
View File
@@ -479,9 +479,6 @@
"app/db/oper/userconfig.py": {
"I001": 1
},
"app/db/oper/workflow.py": {
"I001": 1
},
"app/db/session.py": {
"I001": 1
},
@@ -1263,9 +1260,6 @@
"tests/test_db_lazy_engine.py": {
"I001": 1
},
"tests/test_db_oper_layer.py": {
"I001": 1
},
"tests/test_db_oper_layer_extra.py": {
"I001": 1
},
tests/fixtures/architecture/runtime-contract-baseline.json
Vendored
+10 -3
View File
@@ -284,9 +284,9 @@
"app.db.workflow_oper": {
"introduced": "v3.0.0",
"is_package": false,
"owner": "db",
"replacement": "app.db.oper.workflow",
"target": "app.db.oper.workflow"
"owner": "sdk",
"replacement": "app.application.workflow.WorkflowExecutionPort",
"target": "app.sdk._legacy.workflow"
},
"app.domain.string": {
"introduced": "v3.0.0",
@@ -926,6 +926,13 @@
"target_name": "MediaInteractionChain"
}
},
"app.db.oper": {
"WorkflowOper": {
"replacement": "app.application.workflow.WorkflowExecutionPort",
"target_module": "app.sdk._legacy.workflow",
"target_name": "WorkflowOper"
}
},
"app.domain.media": {
"MEDIA_SOURCE_ALIASES": {
"replacement": "app.schemas.media.MEDIA_SOURCE_ALIASES",
tests/test_architecture_dependencies.py
+88
View File
@@ -347,6 +347,94 @@ def test_workflow_query_adapter_owns_projection_sessions():
assert "_project_workflow(record)" in source
def test_workflow_execution_chain_uses_typed_transaction_port():
"""工作流 Chain 写端不得再经过 raw Oper 或重复获取全局端口。"""
contract_path = APP_ROOT / "application" / "workflow.py"
contract_tree = ast.parse(
contract_path.read_text(encoding="utf-8"),
filename=str(contract_path),
)
contract = next(
node
for node in contract_tree.body
if isinstance(node, ast.ClassDef)
and node.name == "WorkflowExecutionPort"
)
methods = {
node.name: ast.unparse(node.returns)
for node in contract.body
if isinstance(node, ast.FunctionDef)
and node.returns is not None
}
assert methods == {
"start": "bool",
"success": "bool",
"fail": "bool",
"step": "bool",
"reset": "bool",
}
data_path = APP_ROOT / "application" / "chain" / "data.py"
data_tree = ast.parse(
data_path.read_text(encoding="utf-8"),
filename=str(data_path),
)
data_class = next(
node
for node in data_tree.body
if isinstance(node, ast.ClassDef) and node.name == "ChainDataPorts"
)
workflow_field = next(
node
for node in data_class.body
if isinstance(node, ast.AnnAssign)
and isinstance(node.target, ast.Name)
and node.target.id == "workflow"
)
workflow_getter = next(
node
for node in data_tree.body
if isinstance(node, ast.FunctionDef)
and node.name == "get_chain_workflow_port"
)
assert ast.unparse(workflow_field.annotation) == "WorkflowExecutionPortFactory"
assert ast.unparse(workflow_getter.returns) == "WorkflowExecutionPort"
chain_source = (APP_ROOT / "chain" / "workflow.py").read_text(encoding="utf-8")
startup_source = (
APP_ROOT / "startup" / "initializers" / "modules.py"
).read_text(encoding="utf-8")
assert chain_source.count("get_chain_workflow_port()") == 1
assert "workflow=lambda: workflow_execution" in startup_source
assert "workflow=lambda: WorkflowOper()" not in startup_source
def test_canonical_workflow_oper_has_no_legacy_writer_or_duplicate_exports():
"""工作流旧写入口只能存在于 SDK Legacy facade。"""
oper_path = APP_ROOT / "db" / "oper" / "workflow.py"
oper_tree = ast.parse(
oper_path.read_text(encoding="utf-8"),
filename=str(oper_path),
)
oper_class = next(
node
for node in oper_tree.body
if isinstance(node, ast.ClassDef) and node.name == "WorkflowOper"
)
method_names = {
node.name
for node in oper_class.body
if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef))
}
assert {"start", "success", "fail", "step", "reset"}.isdisjoint(method_names)
assert "legacy" not in oper_path.read_text(encoding="utf-8").lower()
package_source = (APP_ROOT / "db" / "oper" / "__init__.py").read_text(
encoding="utf-8"
)
assert '"WorkflowOper"' not in package_source
def test_agent_data_ports_do_not_duplicate_workflow_query_capability():
"""Agent 数据聚合器不得重新暴露无类型工作流读取入口。"""
source = (APP_ROOT / "application" / "agentdata.py").read_text(
tests/test_db_oper_layer.py
+11 -9
View File
@@ -6,12 +6,11 @@ Oper 层大多是模型方法的薄封装，但薄封装恰恰是最容易出错
验证 Oper 的对外契约，而不是验证它调了哪个模型方法。
"""
import asyncio
import importlib
from unittest.mock import Mock
import pytest
from app.db.oper.downloadhistory import DownloadHistoryOper
from app.db.oper.mediaserver import MediaServerOper
from app.db.models.downloadhistory import DownloadFiles, DownloadHistory
from app.db.models.mediaserver import MediaServerItem
from app.db.models.plugindata import PluginData
@@ -22,6 +21,8 @@ from app.db.models.siteuserdata import SiteUserData
from app.db.models.user import User
from app.db.models.userconfig import UserConfig
from app.db.models.workflow import Workflow
from app.db.oper.downloadhistory import DownloadHistoryOper
from app.db.oper.mediaserver import MediaServerOper
from app.db.oper.plugindata import PluginDataOper
from app.db.oper.site import SiteOper
from app.db.oper.user import UserOper
@@ -374,7 +375,7 @@ def test_workflow_oper_add_rejects_duplicate_name(db):
assert oper.add(**_workflow_kwargs("op-wf")) == (False, "工作流已存在")
def test_workflow_oper_exposes_lists_and_lifecycle(db):
def test_workflow_oper_exposes_lists_and_staged_lifecycle(db):
"""
列表入口与生命周期方法都应透传到模型并落库。
"""
@@ -387,23 +388,24 @@ def test_workflow_oper_exposes_lists_and_lifecycle(db):
assert {w.name for w in oper.list_enabled()} >= {"op-wf-life"}
assert {w.name for w in oper.get_timer_triggered_workflows()} >= {"op-wf-life"}
oper.start(flow.id)
oper.stage_start(flow.id)
assert oper.get(flow.id).state == "R"
oper.step(flow.id, "a1", {"n": 1})
oper.stage_step(flow.id, "a1", {"n": 1})
assert oper.get(flow.id).current_action == "a1"
oper.success(flow.id, "完成")
oper.stage_success(flow.id, "完成")
assert oper.get(flow.id).state == "S"
oper.fail(flow.id, "出错")
oper.stage_fail(flow.id, "出错")
assert oper.get(flow.id).state == "F"
oper.reset(flow.id, reset_count=True)
oper.stage_execution_reset(flow.id, reset_count=True)
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")))
legacy = importlib.import_module("app.db.workflow_oper")
assert WorkflowOper().start(flow.id) is True
assert legacy.WorkflowOper().start(flow.id) is True
db.session.expire_all()
assert WorkflowOper(db=db.session).get(flow.id).state == "R"
tests/test_legacy_db_behavior_compat.py
+56
View File
@@ -7,6 +7,62 @@ from app.db.models.transferhistory import TransferHistory
from app.schemas.file import FileItem
def test_legacy_workflow_writes_delegate_to_configured_execution_port(monkeypatch):
"""旧 WorkflowOper 无 Session 写入必须完整委托类型化事务端口。"""
legacy = importlib.import_module("app.db.workflow_oper")
calls = []
class ExecutionPort:
"""记录五种旧工作流写入调用。"""
def start(self, workflow_id):
"""记录启动。"""
calls.append(("start", workflow_id))
return True
def success(self, workflow_id, result=None):
"""记录成功。"""
calls.append(("success", workflow_id, result))
return True
def fail(self, workflow_id, result):
"""记录失败。"""
calls.append(("fail", workflow_id, result))
return True
def step(self, workflow_id, action_id, context, execution_state=None):
"""记录步骤。"""
calls.append(
("step", workflow_id, action_id, context, execution_state)
)
return True
def reset(self, workflow_id, reset_count=False):
"""记录重置。"""
calls.append(("reset", workflow_id, reset_count))
return True
monkeypatch.setattr(
legacy,
"get_configured_workflow_execution",
lambda: ExecutionPort(),
)
oper = legacy.WorkflowOper()
assert oper.start(7) is True
assert oper.success(7, "done") is True
assert oper.fail(7, "failed") is True
assert oper.step(7, "A", {"value": 1}, {"runtime": {}}) is True
assert oper.reset(7, reset_count=True) is True
assert calls == [
("start", 7),
("success", 7, "done"),
("fail", 7, "failed"),
("step", 7, "A", {"value": 1}, {"runtime": {}}),
("reset", 7, True),
]
def test_legacy_subscribe_add_delegates_to_application_service(monkeypatch):
"""旧 SubscribeOper.add 应保留 mediainfo 写入签名。"""
legacy = importlib.import_module("app.db.subscribe_oper")
tests/test_legacy_import_compat.py
+14
View File
@@ -314,6 +314,19 @@ def test_db_refactor_legacy_modules_are_all_registered():
assert expected <= set(MODULE_ALIASES)
def test_workflow_oper_compatibility_is_only_exposed_by_overlay():
"""旧工作流写入口只由 Legacy facade 和精确符号映射提供。"""
legacy = importlib.import_module("app.db.workflow_oper")
canonical = importlib.import_module("app.db.oper.workflow")
oper_package = importlib.import_module("app.db.oper")
assert MODULE_ALIASES["app.db.workflow_oper"].target == "app.sdk._legacy.workflow"
assert issubclass(legacy.WorkflowOper, canonical.WorkflowOper)
assert legacy.WorkflowOper is not canonical.WorkflowOper
assert oper_package.WorkflowOper is legacy.WorkflowOper
assert "WorkflowOper" not in oper_package.__all__
def test_split_user_oper_facade_exports_data_and_auth_contracts():
"""旧 user_oper 同时提供 UserOper 与八个认证依赖。"""
legacy = importlib.import_module("app.db.user_oper")
@@ -469,6 +482,7 @@ def test_plugin_scan_reports_moved_symbol_import(tmp_path: Path):
def test_symbol_alias_manifest_covers_all_moved_public_symbols():
"""符号级映射清单应覆盖媒体身份、整理工作项、刮削拆分与消息/通知命名统一的旧入口。"""
assert set(SYMBOL_ALIASES["app.db.oper"]) == {"WorkflowOper"}
assert set(SYMBOL_ALIASES["app.workflow"]) == {"WorkFlowManager"}
assert set(SYMBOL_ALIASES["app.domain.media"]) == {
"MEDIA_SOURCE_ALIASES",
tests/test_workflow_execution.py
+8 -1
View File
@@ -635,9 +635,15 @@ def test_workflow_chain_process_serializes_circular_context(monkeypatch):
flows=[{"id": "flow-end", "source": "A", "target": "END", "animated": True}],
)
fake_oper = _FakeWorkflowOper(workflow)
port_calls = []
def get_execution_port():
"""记录单次执行获取事务端口的次数。"""
port_calls.append(True)
return fake_oper
monkeypatch.setattr(workflow_module, "get_workflow_manager", lambda: fake_manager)
monkeypatch.setattr(workflow_module, "get_chain_workflow_port", lambda: fake_oper)
monkeypatch.setattr(workflow_module, "get_chain_workflow_port", get_execution_port)
monkeypatch.setattr(
workflow_module,
"get_configured_workflow_query",
@@ -650,6 +656,7 @@ def test_workflow_chain_process_serializes_circular_context(monkeypatch):
assert success is True
assert message == ""
assert port_calls == [True]
assert fake_oper.succeeded is True
saved_workflow_context = fake_oper.steps[-1]["context"]["workflow_context"]
saved_self = saved_workflow_context["self"]