From 52c5c14f878762ace260cfbbd9d91ec1a94355ca Mon Sep 17 00:00:00 2001 From: jxxghp Date: Mon, 24 Aug 2026 04:59:45 +0800 Subject: [PATCH] refactor: unify monitor history port access --- app/application/history.py | 5 +++++ app/monitor/dispatcher.py | 4 ++-- docs/refactor/backend-architecture-next-stage.md | 10 +++++++++- docs/rules/05-architecture.md | 3 +++ tests/test_architecture_dependencies.py | 15 +++++++++++++++ tests/test_monitor_dispatcher_history.py | 4 ++-- tests/test_monitor_watchfiles.py | 4 ++-- 7 files changed, 38 insertions(+), 7 deletions(-) diff --git a/app/application/history.py b/app/application/history.py index 57b57e61b..af2fa9618 100644 --- a/app/application/history.py +++ b/app/application/history.py @@ -85,6 +85,11 @@ class TransferHistoryPort: return getattr(_get_transfer_history_writer(None), name) +def get_transfer_history_port() -> TransferHistoryWriter: + """返回启动组合根登记的整理历史数据端口实例。""" + return _get_transfer_history_writer(None) + + @dataclass(frozen=True, slots=True) class HistoryMutationResult: """描述历史记录维护操作是否成功及兼容提示。""" diff --git a/app/monitor/dispatcher.py b/app/monitor/dispatcher.py index 068f65241..b34837476 100644 --- a/app/monitor/dispatcher.py +++ b/app/monitor/dispatcher.py @@ -9,9 +9,9 @@ from app.runtime.cache import TTLCache from app.application.directory import DirectoryHelper from app.application.history import ( HistoryGateAction, - TransferHistoryPort as TransferHistoryOper, describe_history_gate, evaluate_history_gate, + get_transfer_history_port, is_skip_action, max_failed_retries, resolve_history, @@ -111,7 +111,7 @@ class TransferDispatcher: """ try: history = resolve_history(src_path, storage=storage, - transfer_history_oper=TransferHistoryOper()) + transfer_history_oper=get_transfer_history_port()) except Exception as err: logger.error(f"查询整理历史失败: {src_path} - {err}") return None diff --git a/docs/refactor/backend-architecture-next-stage.md b/docs/refactor/backend-architecture-next-stage.md index e4dd4c0c4..f33f9f5d9 100644 --- a/docs/refactor/backend-architecture-next-stage.md +++ b/docs/refactor/backend-architecture-next-stage.md @@ -6,7 +6,7 @@ > 审计范围：宿主后端；排除 `app/plugins/**` 运行时插件副本 > 规范优先级：`AGENTS.md` 与 `docs/rules/` 高于本文 > 相关文档：`docs/architecture-overview.md`、`docs/refactor/backend-architecture-governance.md`、`docs/refactor/backend-module-refactor-compatibility.md` -> 实施进度：阶段 0～6 的宿主架构能力已完成收口；API/Application 公共复杂度基线已清零，启动组合根的 SystemConfigOper 构造点已由 14 降至 1；API 进程内后台任务已完成首批统一登记，插件仓适配和 Outbox 外围扩展仍按风险切片推进。Model/Base 查询与写装饰器、legacy 隐式会话外壳均已清零，插件 SDK 也不再导出宿主 Model。2026-08-23 的长期整改阶段 0 已恢复宿主、启动性能、官方插件和 SDK 契约门禁的可信基线；阶段 1a 已补齐 TaskRegistry owner 零债务门禁和诚实的关停超时语义；阶段 1b1 已收口整理 worker、pending 回放、失败通知、进程内 AI 重试、插件监控与事件投递的生命周期所有权；2026-08-24 的阶段 2 已将 212 个已观察宿主模块方法的 legacy aggregation 清零，并补齐可执行 fanout 与下载器文件 DTO 边界；阶段 3 已将消息交互和远程命令的订阅删除统一到 Application/UoW/outbox，宿主不再调用裸线程统计入口；阶段 4 已统一七种消息渠道的宿主回环与后台执行边界；阶段 5 已补齐事件窗口聚合任务的生命周期所有权；阶段 6 已统一插件文件操作的取消完成语义；阶段 7 已统一插件协程补偿的终态等待；阶段 8 已统一宿主同步函数的异步线程池入口；阶段 9 已统一工作流运行时的宿主获取路径；阶段 10 已统一模块、插件与调度运行时的显式 getter 调用；阶段 11 已清除系统配置 getter 的 Oper 形别名；阶段 12 已完成工作流域的显式 Chain 数据端口迁移；阶段 13 已收口用户、交互与消息链的数据端口；阶段 14 已收口音乐订阅数据端口；阶段 15 已收口站点数据端口；阶段 16 已收口媒体服务器数据端口；阶段 17 已收口下载数据端口；阶段 18 已收口主订阅数据端口；阶段 19 已收口整理数据端口；阶段 20 已收口 Agent 数据端口。 +> 实施进度：阶段 0～6 的宿主架构能力已完成收口；API/Application 公共复杂度基线已清零，启动组合根的 SystemConfigOper 构造点已由 14 降至 1；API 进程内后台任务已完成首批统一登记，插件仓适配和 Outbox 外围扩展仍按风险切片推进。Model/Base 查询与写装饰器、legacy 隐式会话外壳均已清零，插件 SDK 也不再导出宿主 Model。2026-08-23 的长期整改阶段 0 已恢复宿主、启动性能、官方插件和 SDK 契约门禁的可信基线；阶段 1a 已补齐 TaskRegistry owner 零债务门禁和诚实的关停超时语义；阶段 1b1 已收口整理 worker、pending 回放、失败通知、进程内 AI 重试、插件监控与事件投递的生命周期所有权；2026-08-24 的阶段 2 已将 212 个已观察宿主模块方法的 legacy aggregation 清零，并补齐可执行 fanout 与下载器文件 DTO 边界；阶段 3 已将消息交互和远程命令的订阅删除统一到 Application/UoW/outbox，宿主不再调用裸线程统计入口；阶段 4 已统一七种消息渠道的宿主回环与后台执行边界；阶段 5 已补齐事件窗口聚合任务的生命周期所有权；阶段 6 已统一插件文件操作的取消完成语义；阶段 7 已统一插件协程补偿的终态等待；阶段 8 已统一宿主同步函数的异步线程池入口；阶段 9 已统一工作流运行时的宿主获取路径；阶段 10 已统一模块、插件与调度运行时的显式 getter 调用；阶段 11 已清除系统配置 getter 的 Oper 形别名；阶段 12 已完成工作流域的显式 Chain 数据端口迁移；阶段 13 已收口用户、交互与消息链的数据端口；阶段 14 已收口音乐订阅数据端口；阶段 15 已收口站点数据端口；阶段 16 已收口媒体服务器数据端口；阶段 17 已收口下载数据端口；阶段 18 已收口主订阅数据端口；阶段 19 已收口整理数据端口；阶段 20 已收口 Agent 数据端口；阶段 21 已收口监控历史端口。 ## 当前复核结论（2026-08-24） @@ -221,6 +221,14 @@ - 兼容边界不变：`AgentDataPorts` 注册表、十个旧 Port 代理类、DB Oper、Agent/工具公开参数和返回值、 权限判断以及 V2/V3 插件调用方式均未修改。 +### 长期整改阶段 21：监控历史端口收口（2026-08-24） + +- `TransferDispatcher` 原先把应用层兼容 `TransferHistoryPort` 再别名为 `TransferHistoryOper`；现在通过 + `get_transfer_history_port()` 直接取得组合根登记的整理历史端口。 +- 监控历史与文件事件测试改为替换命名 getter，架构门禁禁止监控分发器重新导入兼容 Facade。 +- 兼容边界不变：`TransferHistoryPort`、DB Oper、监控候选判定、历史查重、失败重试和整理触发语义均 + 未修改。 + ### 总体判断 当前架构总体合理，已经从跨层混合的遗留单体收敛为**边界清晰的模块化单体**： diff --git a/docs/rules/05-architecture.md b/docs/rules/05-architecture.md index c383e886c..4612a0063 100644 --- a/docs/rules/05-architecture.md +++ b/docs/rules/05-architecture.md @@ -140,6 +140,9 @@ Agent orchestration, memory and tool implementations follow the same rule via the named `get_agent_*_port()` functions from `app/application/agentdata.py`. The legacy Agent `*Port` proxy classes remain import-compatible boundaries and must not be reintroduced as Oper aliases in canonical Agent modules. +Monitor history checks use `get_transfer_history_port()` from +`app/application/history.py`; the constructible `TransferHistoryPort` facade is +retained only for compatibility and is not a canonical Oper substitute. ### Adapter boundaries diff --git a/tests/test_architecture_dependencies.py b/tests/test_architecture_dependencies.py index 6f19f4920..7a4b524c3 100644 --- a/tests/test_architecture_dependencies.py +++ b/tests/test_architecture_dependencies.py @@ -599,6 +599,21 @@ def test_agent_consumers_use_explicit_data_port_getters(): assert violations == [] +def test_monitor_dispatcher_uses_explicit_history_port_getter(): + """监控分发器不得把兼容 TransferHistoryPort 伪装成数据库 Oper。""" + path = APP_ROOT / "monitor" / "dispatcher.py" + tree = ast.parse(path.read_text(encoding="utf-8-sig"), filename=str(path)) + violations = [ + f"{path.relative_to(PROJECT_ROOT).as_posix()}:{node.lineno}" + for node in ast.walk(tree) + if isinstance(node, ast.ImportFrom) + and node.module == "app.application.history" + and any(alias.name == "TransferHistoryPort" for alias in node.names) + ] + + assert violations == [] + + def test_plugin_components_do_not_reexport_legacy_abi_names(): """新插件组件只提供 canonical 能力，不得复制旧 Helper、Manager 或 Oper 导出。""" violations: list[str] = [] diff --git a/tests/test_monitor_dispatcher_history.py b/tests/test_monitor_dispatcher_history.py index d18ed9e51..f6cbdbc1b 100644 --- a/tests/test_monitor_dispatcher_history.py +++ b/tests/test_monitor_dispatcher_history.py @@ -54,7 +54,7 @@ def _patch_history(monkeypatch, record=None, success_record=None) -> MagicMock: oper = MagicMock() oper.get_by_src.return_value = record oper.get_success_by_src.return_value = success_record - monkeypatch.setattr("app.monitor.dispatcher.TransferHistoryOper", MagicMock(return_value=oper)) + monkeypatch.setattr("app.monitor.dispatcher.get_transfer_history_port", MagicMock(return_value=oper)) return oper @@ -229,7 +229,7 @@ def test_history_query_error_registers_pending(monkeypatch): dispatcher = _build_dispatcher() oper = MagicMock() oper.get_by_src.side_effect = RuntimeError("数据库不可用") - monkeypatch.setattr("app.monitor.dispatcher.TransferHistoryOper", MagicMock(return_value=oper)) + monkeypatch.setattr("app.monitor.dispatcher.get_transfer_history_port", MagicMock(return_value=oper)) chain = _patch_chain(monkeypatch) assert dispatcher.handle_file(storage="local", event_path=Path("/downloads/a.mkv"), file_size=100) is False diff --git a/tests/test_monitor_watchfiles.py b/tests/test_monitor_watchfiles.py index 9819a4db1..86d3451a7 100644 --- a/tests/test_monitor_watchfiles.py +++ b/tests/test_monitor_watchfiles.py @@ -262,7 +262,7 @@ def test_handle_file_skips_transfer_when_history_exists(monkeypatch): transfer_chain = MagicMock() logger_info = MagicMock() logger_debug = MagicMock() - monkeypatch.setattr("app.monitor.dispatcher.TransferHistoryOper", FakeTransferHistoryOper) + monkeypatch.setattr("app.monitor.dispatcher.get_transfer_history_port", FakeTransferHistoryOper) monkeypatch.setattr("app.monitor.dispatcher.TransferChain", transfer_chain) monkeypatch.setattr("app.monitor.dispatcher.logger.info", logger_info) monkeypatch.setattr("app.monitor.dispatcher.logger.debug", logger_debug) @@ -300,7 +300,7 @@ def test_handle_file_invokes_transfer_when_history_missing(monkeypatch): transfer_chain_instance = MagicMock() transfer_chain = MagicMock(return_value=transfer_chain_instance) - monkeypatch.setattr("app.monitor.dispatcher.TransferHistoryOper", FakeTransferHistoryOper) + monkeypatch.setattr("app.monitor.dispatcher.get_transfer_history_port", FakeTransferHistoryOper) monkeypatch.setattr("app.monitor.dispatcher.TransferChain", transfer_chain) handled = dispatcher.handle_file(