From 45fe01effe2a4d71d50b8539cbed673ff0f60397 Mon Sep 17 00:00:00 2001 From: ThearchyHelios Date: Tue, 6 Oct 2026 08:48:56 +0800 Subject: [PATCH] =?UTF-8?q?=E5=AE=9A=E6=97=B6=E4=BB=BB=E5=8A=A1=E8=87=AA?= =?UTF-8?q?=E5=8A=A8=E5=AF=B9=E6=AF=94=E5=A4=8D=E6=A0=B8=E6=95=B4=E6=94=B9?= =?UTF-8?q?=EF=BC=88=E5=90=8E=E7=AB=AF=EF=BC=89=EF=BC=9A=E5=8F=AA=E6=AF=94?= =?UTF-8?q?=E6=88=90=E5=8A=9F=E7=9A=84=E8=BF=90=E8=A1=8C=E3=80=81=E6=8C=89?= =?UTF-8?q?=E5=8F=91=E8=B5=B7=E5=85=88=E5=90=8E=E6=8C=91=E4=B8=8A=E4=B8=80?= =?UTF-8?q?=E6=AC=A1=E6=88=90=E5=8A=9F=E7=9A=84=E5=9F=BA=E5=87=86=E3=80=81?= =?UTF-8?q?=E5=92=8C=E4=B8=A4=E8=BE=B9=E7=9A=84=E5=B0=81=E5=AD=98=E7=BB=91?= =?UTF-8?q?=E5=AE=9A=EF=BC=8C=E5=AF=B9=E6=AF=94=E6=8C=AA=E5=88=B0=E8=AE=A4?= =?UTF-8?q?=E9=A2=86=E4=B9=8B=E5=89=8D=E5=9C=A8=E9=94=81=E5=A4=96=E5=81=9A?= =?UTF-8?q?=E5=B9=B6=E9=99=90=E6=97=B6=E9=99=90=E5=B9=B6=E5=8F=91=EF=BC=9B?= =?UTF-8?q?=E5=A4=B1=E8=B4=A5=E9=80=9A=E7=9F=A5=E7=9A=84=E5=A4=84=E7=90=86?= =?UTF-8?q?=E8=A1=8C=E4=B8=8D=E5=86=8D=E9=87=8D=E5=A4=8D=E5=8E=9F=E5=9B=A0?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 只比成功结束的运行，基准也只取成功的：失败、取消的运行指标不全，和上一次比只会得出「移除 N 个」，拿失败的那次当 基准每个指标都成了「新增」。基准改成自己查同一定时任务、同一工作流、比这次先发起的运行里成功并封存的最近一次， 不再用对比接口的 default_base（按结束时间挑：老的失败运行事后被接着跑完，会拿后一次触发当它的「上一次」）。 中间隔着失败、取消的运行时，通知写「与上一次成功的运行（#短编号，日期）相比」，Fire.compare 多一个 previous - 和封存绑定：运行的封存要盖住最后一条结束事件才算封存完（sealed_now）。runner 先写终态、再另起会话封存，失败后 接着跑完的运行在两步之间还带着失败时的清单哈希，以前这一拍会直接认领、按旧封存比、通知里写旧的清单哈希；现在 等新的封存，等不来就照常推送但不比、不写哈希。停在审批上的通知也不再写失败时留下的哈希。对比工件连同两边的清单 哈希一起存，触发历史发现任何一边不在终态或清单哈希变了就给 null；对比途中封存变了不记（条件写） - 时机和时限：锁里只挑出状态有变化的触发，认领挪到对比之后、锁外按条件 UPDATE 做。对比最多等 20 秒（含排队）， 同一时刻最多 2 个；超时只记警告，通知不带对比照常推送。停机时扫描在对比途中被取消，这次触发还没认领，下次照常 比、照常推（以前先认领、后比，这条通知会永远丢掉）。同一份封存已经比过的直接用记下的结果 - 失败通知：从没开始执行的，原因刚写过，处理行不再让人「查看原因」，任务在排期上写「下一次触发会照常发起」，停用 的写怎么重新发起；开始执行过的改成「请在 AgentLab 的「记录」中查看失败原因」，和挂起那句指向同一个导航入口 - 新增 test_scheduler_compare_guards.py（封存绑定、时限、取消、并发、复用），test_scheduler_compare.py 补真实缺 指标的失败运行、上一次失败、补跑老运行几种用例；去掉实现逐项验证过会变红 --- backend/app/api/schedules.py | 45 ++- backend/app/scheduler/__init__.py | 3 +- backend/app/scheduler/auto_compare.py | 201 ++++++++--- backend/app/scheduler/webhook.py | 161 ++++++--- backend/tests/test_scheduler_compare.py | 196 ++++++++--- .../tests/test_scheduler_compare_guards.py | 312 ++++++++++++++++++ backend/tests/test_scheduler_notice_rules.py | 28 +- backend/tests/test_scheduler_webhook.py | 2 +- 8 files changed, 766 insertions(+), 182 deletions(-) create mode 100644 backend/tests/test_scheduler_compare_guards.py diff --git a/backend/app/api/schedules.py b/backend/app/api/schedules.py index 7931aaf3..e45c2e0a 100644 --- a/backend/app/api/schedules.py +++ b/backend/app/api/schedules.py @@ -1,4 +1,5 @@ -"""定时任务：按周期自动发起工作流运行，结束、失败、停在审批时推送 Webhook，运行结束后和同一任务的上一次自动对比。 +"""定时任务：按周期自动发起工作流运行，结束、失败、停在审批时推送 Webhook，运行成功结束后和同一任务上一次成功的运行 +自动对比。 形状按前后端约定（Schedule、Fire 两种对象，见 _schedule_out、_fire_outs）。Webhook 地址和签名密钥加密存放，接口 只给主机和末 4 位（url_hint）、有没有密钥（has_secret）。请求体自己解析、自己校验：错误一律回中文 detail，表单 @@ -279,13 +280,30 @@ def _webhook_out(row: Schedule) -> dict[str, Any] | None: async def _fire_outs(session: AsyncSession, fires: list[ScheduleFire]) -> list[dict[str, Any]]: """Fire 对象：run_status 是运行现在的状态，notified 是这次触发的推送记录 [{event, status, at}]（按时间先后）， - compare 是运行结束后和同一任务上一次的自动对比 {base_run_id, summary}（summary 同对比接口的计数），没有为 null。 + compare 是运行成功结束后和同一任务上一次成功的运行的自动对比 {base_run_id, summary, previous}（summary 同对比接口 + 的计数，previous 说基准是不是紧挨着的上一次），没有为 null。 + + 对比和比的时候两边的封存绑定（auto_compare）：哪一边现在不在终态（被继续执行）、被删掉、清单哈希和比的时候不一样 + （重新封存过）都给 null——对比页按现在的封存现算，记下的计数和点进去看到的会对不上，或者干脆比不了。工件读不出来、 + 被改过的同样给 null。""" + compared = {f.id: f.compare_artifact for f in fires if f.compare_artifact and f.run_id} + compares = await auto_compare.briefs(list(compared.values())) + run_ids = {f.run_id for f in fires if f.run_id} | {brief["base_run_id"] for brief in compares.values()} + # 运行现在的状态和清单哈希：run_status 照写，对比按它核对两边的封存 + current = {run_id: (status, seal) for run_id, status, seal in (await session.execute( + select(Run.id, Run.status, Run.manifest_hash).where(Run.id.in_(run_ids)))).all()} if run_ids else {} + + def compare_of(f: ScheduleFire) -> dict[str, Any] | None: + brief = compares.get(compared.get(f.id) or "") + if brief is None or brief["target_run_id"] != f.run_id: + return None + for run_id, seal in ((brief["target_run_id"], brief["target_manifest"]), + (brief["base_run_id"], brief["base_manifest"])): + status, now_seal = current.get(run_id, (None, None)) + if status not in auto_compare.TERMINAL or now_seal != seal: + return None + return auto_compare.public(brief) - 运行又被继续执行、现在不在终态时 compare 也给 null：对比页只认结束并封存的运行，链过去只会被拒；再结束时 - 通知扫描会按新的封存重新比。基准那一次被删掉、被继续执行的，工件读不出来、被改过的，同样给 null。""" - run_ids = [f.run_id for f in fires if f.run_id] - statuses = dict((await session.execute(select(Run.id, Run.status).where(Run.id.in_(run_ids)))).all()) \ - if run_ids else {} notified: dict[str, list[dict[str, Any]]] = {} if fires: rows = (await session.execute( @@ -296,25 +314,16 @@ async def _fire_outs(session: AsyncSession, fires: list[ScheduleFire]) -> list[d notified.setdefault(d.fire_id, []).append({"event": d.event, "status": d.status, "at": _iso(d.created_at)}) # 正在发起的触发：库里是占位的「未能发起」（为了进程被强杀时照实说），进程还在、只是还没发起完，按已发起给 launching = {f.id for f in fires if service.launching(f.id)} - compared = {f.id: f.compare_artifact for f in fires - if f.compare_artifact and statuses.get(f.run_id) in auto_compare.TERMINAL} - compares = await auto_compare.briefs(list(compared.values())) - # 基准一侧同样要在终态：被删掉（强制删除封存的正式运行）、失败后被继续执行的，对比页比不了 - base_ids = list({brief["base_run_id"] for brief in compares.values()}) - if base_ids: - bases = dict((await session.execute(select(Run.id, Run.status).where(Run.id.in_(base_ids)))).all()) - compares = {artifact: brief for artifact, brief in compares.items() - if bases.get(brief["base_run_id"]) in auto_compare.TERMINAL} return [{ "id": f.id, "fire_at": _iso(f.fire_at), "kind": f.kind, "status": "started" if f.id in launching else f.status, "run_id": f.run_id, - "run_status": statuses.get(f.run_id) if f.run_id else None, + "run_status": current.get(f.run_id, (None, None))[0] if f.run_id else None, "reason": None if f.id in launching else f.reason, "notified": notified.get(f.id, []), - "compare": compares.get(compared[f.id]) if f.id in compared else None, + "compare": compare_of(f), "created_at": _iso(f.created_at), } for f in fires] diff --git a/backend/app/scheduler/__init__.py b/backend/app/scheduler/__init__.py index beac3ead..57b7c9d7 100644 --- a/backend/app/scheduler/__init__.py +++ b/backend/app/scheduler/__init__.py @@ -4,7 +4,8 @@ - service.py：输入模板、和工作流入口字段的对照、发起一次触发（调度循环和「立即运行」共用） - loop.py：调度循环，条件 UPDATE 认领到点的任务 - webhook.py：通知推送（通用 / 飞书 / 企业微信 / 钉钉） -- auto_compare.py：运行结束后和同一定时任务的上一次自动对比，结果记在触发记录上，通知里只带计数摘要和对比页链接 +- auto_compare.py：运行成功结束后和同一定时任务上一次成功的运行自动对比，结果记在触发记录上，通知里只带计数摘要和 + 对比页链接 服务启动时（main.py 的 lifespan）只在 scheduler_enabled() 时调 start，关闭时总是调 stop——stop 要能在没 start 过时安全返回。start 自己也看一次开关：拷来的数据目录（检查脚本的 --stacks、验收副本）不管从哪条路径起服务， diff --git a/backend/app/scheduler/auto_compare.py b/backend/app/scheduler/auto_compare.py index 90c15bc6..83a209b4 100644 --- a/backend/app/scheduler/auto_compare.py +++ b/backend/app/scheduler/auto_compare.py @@ -1,15 +1,24 @@ -"""定时运行结束后的自动对比：和同一定时任务的上一次运行比一次，结果存成工件，工件 id 记在触发记录上 +"""定时运行结束后的自动对比：和同一定时任务上一次成功的运行比一次，结果存成工件，工件 id 记在触发记录上 （ScheduleFire.compare_artifact）。 -对比本身调运行对比接口的现成函数（api/compare.py 的 default_base、compare_runs），这里只管什么时候比、跟谁比、结果 -放哪、怎么说给触发历史和通知听： +对比本身调运行对比接口的现成函数（api/compare.py 的 compare_runs，不改它），这里只管比哪些、跟谁比、什么时候比、 +结果放哪、怎么说给触发历史和通知听： -- 什么时候：通知扫描（webhook.notify_due）认领到运行进入终态（结束、失败、取消）的那一次之后、推送之前。和推送一样， - 同一个终态只处理一次；失败后被继续执行、又结束的，按新的封存重新比，覆盖旧的结果 -- 跟谁比：default_base 的「同一定时任务」那一档，即这个任务里更早结束、封存过的终态运行中最近的一次。它退回「同一 - 工作流、同一运行类别」时不比：那不是这个定时任务的上一次，别人手动改的输入、另一个任务的参数都会被算成变化 -- 结果：整份对比结果存成工件（kind=run_compare）。不挂在任何运行名下：它不是运行产出的，不该出现在运行的工件列表里。 - 没封存的不比（服务重启时还在排队、被改成失败的，永远不会封存）；比不了、出错只记日志，通知照常推送 +- 比哪些：成功结束、封存完的运行。失败、取消的运行指标不全（口径卡常常没执行），和上一次比只会得出「移除 N 个」， + 收到通知的人会以为报告删了指标 +- 跟谁比：同一定时任务、同一工作流、比这次先发起的运行里成功结束并封存的最近一次（_previous）。失败的那次不当基准： + 拿它当基准，每个指标都成了「新增」。按发起先后挑，不按结束时间：老的失败运行事后被接着跑完，结束得比后一次晚， + 它的上一次仍是它前面那次。紧挨着的上一次不是成功的（失败、取消、还没结束）时 previous 为 False，通知和触发历史 + 写明比的是哪一次 +- 和封存绑定：运行的封存要盖住它最后一条结束事件才算封存完（sealed_now）。runner 先写终态、再另起会话封存，失败后 + 接着跑完的运行在两步之间还带着失败时的清单哈希，拿它去比只比得到旧封存范围里的事件。结果里记下两边的清单哈希： + 哪一边后来变了，触发历史就不再给这份对比（api/schedules._fire_outs）；对比途中变了，结果不记 +- 什么时候：通知扫描（webhook.notify_due）挑出这次触发之后、认领之前，在锁外比，比完（或等过时限）再认领、推送。 + 比到一半服务停了，这次触发还没认领，下次启动照常比、照常推；同一份封存已经比过的，直接用记下的结果 +- 时限：一次对比最多等 WAIT_S 秒，同一时刻最多 SLOTS 个在算（读快照、重算都在线程里，不能一起占满接口请求也要用的 + 线程池）。超时、比不了（对比接口拒绝，如基准的封存核对不一致）、出错只记日志，通知照常推送、不带对比 +- 结果：整份对比结果连同两边的清单哈希存成工件（kind=run_compare）。不挂在任何运行名下：它不是运行产出的，不该出现 + 在运行的工件列表里 - 对外：触发历史（Fire.compare）和通知只给基准是哪一次、各类指标的个数，不给具体数字 """ from __future__ import annotations @@ -17,21 +26,29 @@ from __future__ import annotations import asyncio import logging import threading +import weakref from typing import Any from urllib.parse import urlencode from fastapi import HTTPException -from sqlalchemy import update +from sqlalchemy import func, select, update from app.core import artifact_store +from app.core.events import EventType from app.db.base import SessionLocal -from app.db.models import Run, ScheduleFire +from app.db.models import Run, RunEvent, ScheduleFire logger = logging.getLogger(__name__) KIND = "run_compare" -#: 终态，同对比接口（api/evidence._TERMINAL）：只有这几种状态、封存过的运行能参与对比 +#: 终态，同对比接口（api/evidence._TERMINAL）：触发历史里两边都要在终态，对比页才比得了 TERMINAL = ("succeeded", "failed", "cancelled") +#: 结束事件：封存要盖住这次运行的最后一条，才是这一次结束时封的 +_ENDS = (EventType.RUN_FINISHED.value, EventType.RUN_FAILED.value, EventType.RUN_CANCELLED.value) +#: 一次对比最多等多久（秒），排队等空位的时间也算在里面。等不到就不等了：通知不带对比照常推送 +WAIT_S = 20.0 +#: 同一时刻最多几个对比在算 +SLOTS = 2 #: 各类指标的叫法，和对比页一致（frontend/src/lib/terms.ts 的 RUN_COMPARE_TEXT.classLabel） CLASS_LABEL = { "data_only": "只数据变", @@ -56,6 +73,8 @@ NO_METRICS = "两次运行都没有指标" _BRIEFS: dict[str, dict[str, Any]] = {} _BRIEFS_MAX = 512 _BRIEFS_LOCK = threading.Lock() +#: 对比的空位，按事件循环分开（同 service.launch_lock）：测试每个用例一个新循环，跨循环用同一个信号量会报错 +_SLOTS: "weakref.WeakKeyDictionary[asyncio.AbstractEventLoop, asyncio.Semaphore]" = weakref.WeakKeyDictionary() def summary_text(summary: dict[str, Any]) -> str: @@ -73,68 +92,148 @@ def page_url(base_url: str, target: str, base: str) -> str: return f"{base_url}/runs/compare?{urlencode({'target': target, 'base': base})}" -async def compare_fire(fire_id: str, run_id: str) -> dict[str, Any] | None: - """这次触发发起的运行刚进入终态：和同一定时任务的上一次比一次，结果记进触发记录。返回给通知用的摘要 - {base_run_id, summary}；没有可比的、比不了时返回 None，触发记录上之前的结果一并清掉（那是上一次封存的）。 +async def sealed_now(session: Any, run: Run) -> bool: + """这次运行的封存是不是这一次结束时封的：有清单哈希，而且封存范围盖住了最后一条结束事件。 - 不抛异常：比不了（对比接口拒绝，如基准的封存核对不一致）记一条警告，意外的异常记堆栈，都不影响通知。""" - artifact: str | None = None - brief: dict[str, Any] | None = None + runner 先在一个事务里写终态和结束事件，再另起会话封存。失败后接着跑完（或接着跑之后停在审批上被放弃）的运行，在 + 两步之间还带着失败时的清单哈希，只看「有没有哈希」会把旧的封存当成这一次的。没有 manifest_seq 的是加这一列之前 + 封存的老运行，只能看有没有哈希。""" + if not run.manifest_hash: + return False + if run.manifest_seq is None: + return True + last = await session.scalar(select(func.max(RunEvent.seq)).where( + RunEvent.run_id == run.id, RunEvent.type.in_(_ENDS))) + return last is None or run.manifest_seq >= last + + +async def compare_fire(fire_id: str, run_id: str) -> dict[str, Any] | None: + """这次触发发起的运行成功结束、封存完了：和同一定时任务上一次成功的运行比一次，结果记进触发记录。返回给通知用的 + 摘要 {base_run_id, summary, previous, base_started_at, …}；不该比、没有可比的、比不了时返回 None。 + + 最多等 WAIT_S 秒（含排队）。超时、对比接口拒绝（基准的封存核对不一致等）记一条警告，意外的异常记堆栈，都不影响 + 通知。取消不吞：扫描被取消（停机）时这次触发还没认领，下次照常比。""" try: - artifact, brief = await _compare(run_id) + return await asyncio.wait_for(_compare(fire_id, run_id), timeout=WAIT_S) + except TimeoutError: + logger.warning("定时任务的运行 %s 自动对比超过 %g 秒仍未完成，通知不带对比照常推送", run_id, WAIT_S) except HTTPException as e: logger.warning("定时任务的运行 %s 自动对比未完成：%s", run_id, e.detail) except Exception: # noqa: BLE001 - 对比出错不能拖住通知，也不能让通知扫描停下 logger.exception("定时任务的运行 %s 自动对比出错", run_id) - await _store(fire_id, artifact) - return brief + return None -async def _compare(run_id: str) -> tuple[str | None, dict[str, Any] | None]: +async def _compare(fire_id: str, run_id: str) -> dict[str, Any] | None: # 现成的对比函数在用到时才导入：api/compare.py 连带证据、推演接口，调度模块被导入时不必一起加载 from app.api import compare as compare_api + # 挑基准、核对封存、看有没有比过都只是读库，不占对比的空位：前面排着慢的对比时，第一次运行这类没什么可比的 + # 不必跟着排队、等到超时 async with SessionLocal() as session: - run = await session.get(Run, run_id) - if run is None or run.status not in TERMINAL or not run.manifest_hash: - return None, None - found = await compare_api.default_base(run_id) - if found is None or found[1] != "same_schedule": - return None, None - base = found[0] - result = await compare_api.compare_runs(run_id, base.id) - brief = _brief(result) - if brief is None: - return None, None - artifact = await artifact_store.put_json(result, kind=KIND, meta={"target_run_id": run_id, "base_run_id": base.id}) + target = await session.get(Run, run_id) + stored = await session.scalar(select(ScheduleFire.compare_artifact).where(ScheduleFire.id == fire_id)) + if target is None or target.status != "succeeded" or not target.schedule_id or not target.workflow_id \ + or not await sealed_now(session, target): + return None + found = await _previous(session, target) + if found is None: + return None + base, previous = found + if not await sealed_now(session, base): + # 基准正好处在写完终态、还没封存的那一刻，或者封存没补上就停了机：拿它比只比得到旧封存里的事件 + logger.warning("作为基准的运行 %s 还没有封存到它的结束事件，运行 %s 这次不比", base.id, run_id) + return None + sides = ((target.id, target.manifest_hash), (base.id, base.manifest_hash)) + (target_id, target_seal), (base_id, base_seal) = sides + if stored: + # 同一份封存已经比过（比完还没认领就停了机）：直接用 + known = await asyncio.to_thread(brief_of, stored) + if known is not None and _sides_of(known) == sides: + return _copy(known) + async with _slots(): + result = await compare_api.compare_runs(target_id, base_id) + payload = {"target_manifest": target_seal, "base_manifest": base_seal, "previous": previous, "result": result} + brief = _brief(payload) + if brief is None or _sides_of(brief) != sides: + return None + artifact = await artifact_store.put_json(payload, kind=KIND, + meta={"target_run_id": target_id, "base_run_id": base_id}) + if not await _store(fire_id, artifact, sides): + logger.warning("运行 %s 或作为基准的运行 %s 在对比途中重新封存（或触发记录已删除），这次的对比结果不记", + target_id, base_id) + return None _remember(artifact, brief) - return artifact, _copy(brief) + return _copy(brief) -async def _store(fire_id: str, artifact: str | None) -> None: - try: - async with SessionLocal() as session: - await session.execute(update(ScheduleFire).where(ScheduleFire.id == fire_id) - .values(compare_artifact=artifact)) - await session.commit() - except Exception: # noqa: BLE001 - 任务刚被删掉之类：记不下不影响通知 - logger.warning("触发记录 %s 的自动对比结果没能保存", fire_id) +async def _previous(session: Any, target: Run) -> tuple[Run, bool] | None: + """(基准, 是不是紧挨着的上一次)：同一定时任务、同一工作流、比这次先发起的运行里成功结束并封存的最近一次。 - -def _brief(result: Any) -> dict[str, Any] | None: - """对比结果里给触发历史、通知的那一点：基准是哪一次、各类指标的个数（同对比接口的 summary）。""" - if not isinstance(result, dict): + 按发起先后挑（created_at），不按结束时间；同一工作流才比得了（任务改过工作流的，之前那个工作流的运行不算）。 + 中间隔着别的运行（失败、取消、还没结束的）时第二项为 False。""" + earlier = (Run.schedule_id == target.schedule_id, Run.workflow_id == target.workflow_id, Run.id != target.id, + Run.created_at < target.created_at) + base = (await session.execute( + select(Run).where(*earlier, Run.status == "succeeded", Run.manifest_hash.is_not(None)) + .order_by(Run.created_at.desc()).limit(1))).scalars().first() + if base is None: return None - base, summary = result.get("base"), result.get("summary") - if not isinstance(base, dict) or not isinstance(base.get("id"), str) or not isinstance(summary, dict): + between = await session.scalar(select(func.count()).select_from(Run).where( + *earlier, Run.created_at > base.created_at)) + return base, not between + + +def _slots() -> asyncio.Semaphore: + loop = asyncio.get_running_loop() + slots = _SLOTS.get(loop) + if slots is None: + slots = _SLOTS[loop] = asyncio.Semaphore(SLOTS) + return slots + + +async def _store(fire_id: str, artifact: str, sides: tuple[tuple[str, str | None], ...]) -> bool: + """把工件 id 记进触发记录。条件写：两边都还是成功结束、清单哈希还是比的时候那一份，才记。""" + same = [select(Run.id).where(Run.id == run_id, Run.status == "succeeded", Run.manifest_hash == seal).exists() + for run_id, seal in sides] + async with SessionLocal() as session: + result = await session.execute(update(ScheduleFire).where(ScheduleFire.id == fire_id, *same) + .values(compare_artifact=artifact)) + await session.commit() + return result.rowcount == 1 + + +def _sides_of(brief: dict[str, Any]) -> tuple[tuple[str, str | None], ...]: + return (brief["target_run_id"], brief["target_manifest"]), (brief["base_run_id"], brief["base_manifest"]) + + +def _brief(payload: Any) -> dict[str, Any] | None: + """对比工件里给触发历史、通知的那一点：两边是哪一次、比的时候两边的清单哈希、基准是不是紧挨着的上一次、基准 + 什么时候开始的、各类指标的个数（同对比接口的 summary）。形状不对的（这一版之前的工件、被改过的）返回 None。""" + if not isinstance(payload, dict) or not isinstance(payload.get("result"), dict): return None - return {"base_run_id": base["id"], + result = payload["result"] + target, base, summary = result.get("target"), result.get("base"), result.get("summary") + if not isinstance(target, dict) or not isinstance(base, dict) or not isinstance(summary, dict): + return None + keys = (target.get("id"), base.get("id"), payload.get("target_manifest"), payload.get("base_manifest")) + if not all(isinstance(key, str) and key for key in keys): + return None + started = base.get("started_at") + return {"target_run_id": keys[0], "base_run_id": keys[1], "target_manifest": keys[2], "base_manifest": keys[3], + "previous": payload.get("previous") is not False, + "base_started_at": started if isinstance(started, str) else None, "summary": {key: value for key, value in summary.items() if isinstance(value, int) and not isinstance(value, bool)}} +def public(brief: dict[str, Any]) -> dict[str, Any]: + """接口给前端的那一份（Fire.compare）：基准是哪一次、各类指标的个数、基准是不是紧挨着的上一次。""" + return {"base_run_id": brief["base_run_id"], "summary": dict(brief["summary"]), "previous": brief["previous"]} + + def _copy(brief: dict[str, Any]) -> dict[str, Any]: - return {"base_run_id": brief["base_run_id"], "summary": dict(brief["summary"])} + return {**brief, "summary": dict(brief["summary"])} def _remember(artifact: str, brief: dict[str, Any]) -> None: diff --git a/backend/app/scheduler/webhook.py b/backend/app/scheduler/webhook.py index e74e12b7..8d8927d4 100644 --- a/backend/app/scheduler/webhook.py +++ b/backend/app/scheduler/webhook.py @@ -9,8 +9,8 @@ Webhook 地址本身就是凭证（企业微信的 key、钉钉的 access_token 一个过滤器把推送期间的那几行换成主机名。 通知里只有任务名、状态、出具档位、缺口条数、运行链接、封存哈希（界面上叫清单哈希）前 12 位和起止时间，失败时 -另写失败节点的名字和去哪里查看（不许诺能继续运行）；运行结束后和同一任务上一次的对比（auto_compare）只写各类 -指标的个数和对比页链接。不放报告里的具体数字，也不放失败时的原始报错——报错里可能带着查询到的数据，Webhook +另写失败节点的名字和下一步（不许诺能继续运行）；运行成功结束后和同一任务上一次成功的运行的对比（auto_compare）只写 +各类指标的个数和对比页链接。不放报告里的具体数字，也不放失败时的原始报错——报错里可能带着查询到的数据，Webhook 推去的群不一定有权看。 三个平台出错时都回 HTTP 200、错误码放在响应体里，所以送没送达要看响应体：按地址的主机认平台，不看选的格式—— @@ -67,14 +67,17 @@ _STATUS_LABEL = {"succeeded": "已完成", "failed": "失败", "cancelled": "已 #: 挂起的运行要人去处理：不处理，这个任务之后的触发都按「上一次还没结束」跳过 SUSPENDED_REASON = "运行停在断点上、没有待审批，通常是服务重启所致" SUSPENDED_ACTION = "请在 AgentLab 的「记录」中继续运行或放弃这次运行；它结束之前，这个定时任务之后的触发都会跳过" -#: 失败的运行怎么处理：不许诺「修正后可继续运行」。没开始执行的没有断点；开始执行过的也可能没有（执行第一个节点前就 -#: 失败）或已不兼容（系统升级），能不能继续以记录页的提示为准 -FAILED_ACTION = "请在 AgentLab 的运行记录中查看原因" +#: 失败的运行怎么处理：不许诺「修正后可继续运行」。开始执行过的也可能没有断点（执行第一个节点前就失败）或已不兼容 +#: （系统升级），能不能继续以记录页的提示为准。「记录」是导航里那一页的名字，和挂起那句一致 +FAILED_ACTION = "请在 AgentLab 的「记录」中查看失败原因" #: 从没开始执行就失败的（服务重启时还在排队，被改成失败、没有开始时间）：没有可继续的进度。不照抄运行记录上的报错—— #: 通知一律不放原始报错，这一句是按「没有开始时间」认出来的，自己写 NOT_STARTED_REASON = "运行在开始执行前就已中止（通常是服务重启时仍在排队），没有可继续的进度" #: 失败的运行不挡之后的触发（挂起的才挡）：任务还在排期上时补这一句，免得人以为这个定时任务就此停了 NEXT_AS_USUAL = "下一次触发会照常发起" +#: 从没开始执行、任务又不在排期上（停用了）：原因刚写过，记录页上也没有更多可看的，下一步是重新发起。「立即运行」 +#: 是定时任务页上那个按钮的名字，停用的任务也能点 +RERUN_ACTION = "如需重新发起，请在 AgentLab 的「定时任务」中点击「立即运行」" # -------------------------------------------------------------------------- @@ -228,7 +231,8 @@ class Notice: suspended：运行停在断点上、没有待审批（通常是服务重启挂起的），按 failed 事件推。error_node：失败节点的名字。 next_fire：任务启用着、往后还会触发（失败通知据此补「下一次触发会照常发起」）。 - compare：和同一定时任务上一次的对比摘要 {base_run_id, summary}（只有个数），compare_url：对比页的链接。 + compare：和同一定时任务上一次成功的运行的对比摘要（auto_compare.compare_fire 的返回：基准是哪一次、只有个数的 + summary、基准是不是紧挨着的上一次、基准开始的时间），compare_url：对比页的链接。 fmt、signed 只给测试消息用：说明里只说这条消息证明得了的事。""" event: str @@ -293,7 +297,7 @@ def message(notice: Notice) -> tuple[str, list[tuple[str, str]], str | None]: lines.append(("出具档位", f"{ISSUANCE_LABEL.get(notice.tier, notice.tier)}{gaps}")) if notice.compare: # 只给各类指标的个数：具体的数在对比页上，跟着运行的权限走，不随消息推进群里 - lines.append(("与上一次相比", auto_compare.summary_text(notice.compare.get("summary") or {}))) + lines.append((_compare_head(notice), auto_compare.summary_text(notice.compare.get("summary") or {}))) if notice.pending_approvals: lines.append(("待审批", f"{notice.pending_approvals} 项，请到 AgentLab 处理；运行结束之前，" "这个定时任务之后的触发都会跳过")) @@ -302,12 +306,14 @@ def message(notice: Notice) -> tuple[str, list[tuple[str, str]], str | None]: elif notice.suspended: lines += [("原因", SUSPENDED_REASON), ("处理", SUSPENDED_ACTION)] elif notice.status == "failed": - # 不写原始报错（可能带着查询到的数据），写失败节点的名字和去哪里处理 + # 不写原始报错（可能带着查询到的数据），写失败节点的名字和下一步 if notice.error_node: lines.append(("失败节点", f"「{notice.error_node}」")) if notice.never_started: - lines.append(("原因", NOT_STARTED_REASON)) - lines.append(("处理", f"{FAILED_ACTION}；{NEXT_AS_USUAL}" if notice.next_fire else FAILED_ACTION)) + # 原因刚写过，也没有断点可续：不再让人去查看原因，说下一次照常，不在排期上的说怎么重新发起 + lines += [("原因", NOT_STARTED_REASON), ("处理", NEXT_AS_USUAL if notice.next_fire else RERUN_ACTION)] + else: + lines.append(("处理", f"{FAILED_ACTION}；{NEXT_AS_USUAL}" if notice.next_fire else FAILED_ACTION)) if notice.fire_at is not None: when = f"{_local(notice.fire_at, notice.timezone, seconds=False)}（{notice.timezone}）" # 立即运行不写「计划触发」：群里的人会以为报告按时出了 @@ -329,6 +335,27 @@ def message(notice: Notice) -> tuple[str, list[tuple[str, str]], str | None]: return title, lines, notice.run_url +def _compare_head(notice: Notice) -> str: + """对比那一行的项名。基准是紧挨着的上一次时写「与上一次相比」；不是（上一次失败了、取消了）时写明比的是哪一次： + 短编号（同界面，#前 6 位）和开始的日期（按定时任务的时区）。""" + compare = notice.compare or {} + if compare.get("previous", True): + return "与上一次相比" + when = _date_of(compare.get("base_started_at"), notice.timezone) + return f"与上一次成功的运行（#{str(compare.get('base_run_id') or '')[:6]}{f'，{when}' if when else ''}）相比" + + +def _date_of(text: Any, tz: str) -> str | None: + """ISO 时间（对比结果里运行的开始时间）按定时任务的时区写成日期，读不出来的返回 None。""" + try: + moment = datetime.fromisoformat(str(text)) + except ValueError: + return None + if moment.tzinfo is None: + moment = moment.replace(tzinfo=timezone.utc) + return (_local(moment, tz, seconds=False) or "")[:10] or None + + def _status_label(notice: Notice) -> str: if notice.status == "interrupted": return "等待审批" if notice.pending_approvals else "已挂起，可继续运行" @@ -406,9 +433,10 @@ def _generic(notice: Notice, now: datetime) -> dict[str, Any]: else f"{SUSPENDED_REASON}。{SUSPENDED_ACTION}" if notice.suspended else NOT_STARTED_REASON if notice.never_started else None), "error_node": notice.error_node, - # 对比只给基准是哪一次、各类指标的个数和对比页链接 + # 对比只给基准是哪一次（是不是紧挨着的上一次）、各类指标的个数和对比页链接 "compare": ({"base_run_id": notice.compare.get("base_run_id"), - "summary": dict(notice.compare.get("summary") or {}), "url": notice.compare_url} + "summary": dict(notice.compare.get("summary") or {}), + "previous": notice.compare.get("previous", True) is not False, "url": notice.compare_url} if notice.compare else None), "sent_at": _iso(now), } @@ -592,12 +620,16 @@ class _Job: @dataclass class _Due: - """认领到的一次状态变化接下来要做的事：运行进入终态时和上一次比一次（compare_run），订阅了这个事件时推一条 - （job）。两样都在锁外做：对比要读快照、重算，推送要等接收端，都不能挡着发起。""" + """挑出来的一次状态变化接下来要做的事：运行成功结束时和上一次成功的运行比一次（compare_run），按条件 UPDATE + 认领（把 notified_status 从 old 改成 state），认领到了、订阅了这个事件时推一条（job）。都在锁外做：对比要读快照、 + 重算，推送要等接收端，都不能挡着发起。pushed：认领到了、发出了推送（含失败的）。""" fire_id: str + old: str | None + state: str compare_run: str | None job: _Job | None + pushed: bool = False async def notify_due(now: datetime | None = None) -> int: @@ -607,8 +639,11 @@ async def notify_due(now: datetime | None = None) -> int: 推一次。没订阅这个事件、任务没配 Webhook 的也记下已处理——之后再配上 Webhook，不会把旧的补推一遍。 停在断点上的运行每一拍都要重新看（见 _state_of）：它的已处理状态带着第几段执行，和运行状态本身对不上。 - 运行进入终态的那一次，推送之前先和同一定时任务的上一次比一次（auto_compare），结束、失败的通知带上计数摘要和 - 对比页链接。没配 Webhook 的也比：触发历史里要用。比不了、出错不影响推送。 + 运行成功结束、封存完的那一次，认领之前先和同一定时任务上一次成功的运行比一次（auto_compare），结束的通知带上计数 + 摘要和对比页链接。没配 Webhook 的也比：触发历史里要用。对比在锁外、认领之前做，最多等 auto_compare.WAIT_S 秒， + 过了就不带对比推送：一次慢的对比最多让这一拍晚这么久，不会把之后所有的通知一直压着，也不拖着发起用的锁。比到一半 + 扫描被取消（停机只等通知扫描一个宽限），这次触发还没认领，下次照常比、照常推——以前先认领、后比，取消之后这条 + 通知再也不会推。比不了、出错不影响推送。 """ now = now or utcnow() dues: list[_Due] = [] @@ -617,7 +652,9 @@ async def notify_due(now: datetime | None = None) -> int: # 扫描窗口按触发记录的创建时间算，两种例外：运行记录窗口内有过变化（运行状态一变 updated_at 就跟着变），以及 # 停在审批上、有待审批的——以前只看触发记录，创建 30 天以后才停到审批上的运行收不到「等待审批」 waiting = select(Approval.id).where(Approval.run_id == Run.id, Approval.status == "pending").exists() - # 和发起共用一把锁：发起时先记一条「未能发起」再发起运行、成功了才改成 started，扫描不能撞在这中间 + # 和发起共用一把锁：发起时先记一条「未能发起」再发起运行、成功了才改成 started，扫描不能撞在这中间。锁里只挑出 + # 状态有变化的、拼好要推的消息；认领在锁外按条件 UPDATE 做，挑出来的是锁里看到的状态，和别的扫描撞上也只有一个 + # 认领得到 async with service.launch_lock(): async with SessionLocal() as session: rows = (await session.execute( @@ -636,39 +673,52 @@ async def notify_due(now: datetime | None = None) -> int: base_url = (await notify_settings(session))["public_base_url"] for fire, run in rows: - state, event, pending = await _state_of(session, fire, run, now) - if state is None or state == fire.notified_status \ - or not await _mark(fire.id, fire.notified_status, state): + seen = await _state_of(session, fire, run, now) + if seen.state is None or seen.state == fire.notified_status: continue - compare_run = run.id if run is not None and run.status in auto_compare.TERMINAL else None schedule = await session.get(Schedule, fire.schedule_id) url = decrypt(schedule.webhook_url) if schedule and schedule.webhook_url else None job = None - if event and url and schedule.webhook_format in FORMATS and event in (schedule.webhook_events or []): - notice = await _notice(session, schedule, fire, run, event, base_url, pending) + if seen.event and url and schedule.webhook_format in FORMATS \ + and seen.event in (schedule.webhook_events or []): + notice = await _notice(session, schedule, fire, run, seen, base_url) job = _Job(schedule.webhook_format, url, decrypt(schedule.webhook_secret), notice) - if compare_run or job: - dues.append(_Due(fire.id, compare_run, job)) + # 只比成功结束、封存完的（见 auto_compare）：失败、取消的指标不全，等不来新封存的比了也是旧的 + compare_run = run.id if run is not None and run.status == "succeeded" and seen.sealed else None + dues.append(_Due(fire.id, fire.notified_status, seen.state, compare_run, job)) if dues: await asyncio.gather(*(_follow_up(due, base_url) for due in dues)) - return sum(1 for due in dues if due.job is not None) + return sum(1 for due in dues if due.pushed) async def _follow_up(due: _Due, base_url: str | None) -> None: - """先比（需要的话），再推：对比摘要要写进这条通知。对比自己兜住了异常，出什么事都不挡推送。""" - if due.compare_run: - brief = await auto_compare.compare_fire(due.fire_id, due.compare_run) - if brief is not None and due.job is not None: - due.job.notice.compare = brief - if base_url: - due.job.notice.compare_url = auto_compare.page_url(base_url, due.compare_run, brief["base_run_id"]) - if due.job is not None: - await _deliver(due.job) + """先比（需要的话），再认领，认领到了再推：对比摘要要写进这条通知；认领放在对比之后，比到一半被取消时这次触发 + 还没认领。对比自己兜住了超时和异常，出什么事都不挡推送。""" + brief = await auto_compare.compare_fire(due.fire_id, due.compare_run) if due.compare_run else None + if not await _mark(due.fire_id, due.old, due.state) or due.job is None: + return + if brief is not None: + due.job.notice.compare = brief + if base_url: + due.job.notice.compare_url = auto_compare.page_url(base_url, due.compare_run, brief["base_run_id"]) + due.pushed = True + await _deliver(due.job) -async def _state_of(session: Any, fire: ScheduleFire, run: Run | None, - now: datetime) -> tuple[str | None, str | None, int]: - """(要记下的已处理状态, 要推送的事件, 待审批数)。还不能下结论（没封存、审批刚批完还没接着执行）时状态为 None。 +@dataclass +class _Seen: + """_state_of 的结论。state：要记下的已处理状态，还不能下结论时为 None；event：要推送的事件；pending：待审批数； + sealed：终态的运行封存完了（封存盖住了这一次的结束事件，见 auto_compare.sealed_now），只有这时才比、通知里才写 + 清单哈希。""" + + state: str | None + event: str | None = None + pending: int = 0 + sealed: bool = False + + +async def _state_of(session: Any, fire: ScheduleFire, run: Run | None, now: datetime) -> _Seen: + """这次触发现在该记成什么状态、推什么。还不能下结论（没封存、审批刚批完还没接着执行）时状态为 None。 停在断点上（interrupted）的分两种，各推一次： - 有待审批：停在审批，推 awaiting_approval。批完接着执行、又停在下一个审批点时要再推，所以已处理状态里带着这是 @@ -678,18 +728,21 @@ async def _state_of(session: Any, fire: ScheduleFire, run: Run | None, 审批批完到运行改回 running 之间，审批是 answered、还没有 resolved：那不是挂起，等下一拍。 """ if fire.status == "failed_to_start": - return "failed_to_start", "failed", 0 + return _Seen("failed_to_start", "failed") if run is None: - return None, None, 0 + return _Seen(None) if run.status in ("succeeded", "failed", "cancelled"): - # runner 先写终态和结束时间、再封存，两步之间被这一拍撞上就等。没有结束时间的（服务重启时还在排队、被改成 - # failed 的）永远不会封存：按它最后一次更新的时间等，以前按「现在」等，每一拍都等、永远不推 + # runner 先写终态和结束时间、再另起会话封存，两步之间被这一拍撞上就等。失败后接着跑完的运行这时还带着失败时 + # 的清单哈希，不能只看有没有哈希，要看封存盖没盖住这一次的结束事件。没有结束时间的（服务重启时还在排队、被改 + # 成 failed 的）永远不会封存：按它最后一次更新的时间等，以前按「现在」等，每一拍都等、永远不推。等过时限仍没 + # 封存完的照常推，只是不比、不写清单哈希 + sealed = await auto_compare.sealed_now(session, run) since = run.finished_at or run.updated_at or fire.created_at - if not run.manifest_hash and (now - service.as_utc(since)).total_seconds() < SEAL_WAIT_S: - return None, None, 0 - return run.status, _EVENT_OF[run.status], 0 + if not sealed and (now - service.as_utc(since)).total_seconds() < SEAL_WAIT_S: + return _Seen(None) + return _Seen(run.status, _EVENT_OF[run.status], sealed=sealed) if run.status != "interrupted": - return None, None, 0 + return _Seen(None) counts = dict((await session.execute( select(Approval.status, func.count()).where( Approval.run_id == run.id, Approval.status.in_(("pending", "answered"))) @@ -697,10 +750,10 @@ async def _state_of(session: Any, fire: ScheduleFire, run: Run | None, segment = await session.scalar(select(func.count()).select_from(RunEvent).where( RunEvent.run_id == run.id, RunEvent.type == EventType.RUN_STARTED.value)) or 0 if counts.get("pending"): - return f"interrupted:approval:{segment}", "awaiting_approval", counts["pending"] + return _Seen(f"interrupted:approval:{segment}", "awaiting_approval", counts["pending"]) if counts.get("answered"): - return None, None, 0 - return f"interrupted:held:{segment}", "failed", 0 + return _Seen(None) + return _Seen(f"interrupted:held:{segment}", "failed") async def _mark(fire_id: str, old: str | None, state: str) -> bool: @@ -712,17 +765,19 @@ async def _mark(fire_id: str, old: str | None, state: str) -> bool: return result.rowcount == 1 -async def _notice(session: Any, schedule: Schedule, fire: ScheduleFire, run: Run | None, event: str, - base_url: str | None, pending: int) -> Notice: +async def _notice(session: Any, schedule: Schedule, fire: ScheduleFire, run: Run | None, seen: _Seen, + base_url: str | None) -> Notice: tier, gaps = await _issuance_of(session, run) if run is not None else (None, None) + pending = seen.pending return Notice( - event=event, schedule_id=schedule.id, schedule_name=schedule.name, timezone=schedule.timezone, + event=seen.event or "", schedule_id=schedule.id, schedule_name=schedule.name, timezone=schedule.timezone, fire_id=fire.id, fire_at=fire.fire_at, fire_kind=fire.kind, run_id=run.id if run else None, status=run.status if run else fire.status, tier=tier, gaps=gaps, run_url=f"{base_url}/runs/{run.id}" if base_url and run else None, - manifest_hash=run.manifest_hash if run else None, + # 只写这一次结束时封的：停在审批上、封存还没补上的运行，身上可能还留着上一次失败时的清单哈希 + manifest_hash=run.manifest_hash if run is not None and seen.sealed else None, started_at=run.started_at if run else None, finished_at=run.finished_at if run else None, reason=fire.reason if fire.status == "failed_to_start" else None, diff --git a/backend/tests/test_scheduler_compare.py b/backend/tests/test_scheduler_compare.py index d7f79238..3c9af745 100644 --- a/backend/tests/test_scheduler_compare.py +++ b/backend/tests/test_scheduler_compare.py @@ -1,17 +1,20 @@ -"""定时运行结束后的自动对比（app/scheduler/auto_compare.py，由通知扫描 webhook.notify_due 在推送之前触发）： +"""定时运行结束后的自动对比（app/scheduler/auto_compare.py，由通知扫描 webhook.notify_due 在认领、推送之前触发）： -- 运行进入终态且已封存、同一定时任务有更早的已封存运行时，用运行对比的现成函数（api/compare.py 的 default_base、 - compare_runs）和它比一次，整份结果存成工件（kind=run_compare，不挂在运行名下、不进运行的工件列表），工件 id 记进 - 触发记录的 compare_artifact。基准是同一定时任务的上一次：同一工作流更近的手动运行、别的定时任务的运行都不算 -- 这个任务的第一次（只有同一工作流别的运行可比）、还没封存时不比；对比出错（基准的封存核对不一致、意外的异常）只记 - 日志，通知照常推送、不带对比 -- 接口：Fire 多一个 compare: {base_run_id, summary} 或 null，summary 同对比接口的计数；运行又被继续执行、不在终态 - 时为 null（对比页只认封存的运行） -- 通知：结束、失败的通知多一行「与上一次相比：只数据变 2 个、口径与数据都变 0 个、SQL 结构变 0 个」，配了对外访问 - 地址时附对比页链接（/runs/compare?target=…&base=…）；只给个数，报告、指标、查询里的具体数字一个不出现 +- 只比成功结束并封存的运行：失败、取消的指标不全，和上一次比只会得出「移除 N 个」，误导收到通知的人 +- 基准是同一定时任务、同一工作流、比这次先发起的运行里成功结束并封存的最近一次：同一工作流更近的手动运行、别的 + 定时任务的运行都不算；失败的那一次不当基准（拿它当基准每个指标都成了「新增」）；按发起先后挑，不按结束时间—— + 老的失败运行事后被接着跑完，结束得比后一次晚，它的上一次仍是它前面那次。紧挨着的上一次不是成功的时，通知和触发 + 历史写明比的是哪一次 +- 整份结果连同两边的清单哈希存成工件（kind=run_compare，不挂在运行名下、不进运行的工件列表），工件 id 记进触发 + 记录的 compare_artifact。这个任务的第一次（只有同一工作流别的运行可比）、还没封存时不比；对比出错（基准的封存核对 + 不一致、意外的异常）只记日志，通知照常推送、不带对比 +- 接口：Fire 多一个 compare: {base_run_id, summary, previous} 或 null，summary 同对比接口的计数，previous 说基准是不是 + 紧挨着的上一次；基准被删、被继续执行（不在终态）时为 null（对比页只认封存的运行） +- 通知：结束的通知多一行「与上一次相比：只数据变 2 个、口径与数据都变 0 个、SQL 结构变 0 个」，配了对外访问地址时附 + 对比页链接（/runs/compare?target=…&base=…）；只给个数，报告、指标、查询里的具体数字一个不出现 -两次周报运行用 tests/fixtures/compare_runs.py 的虚构运行（区域、金额、日期都是编的），按 runner 的算法封存，不真的 -执行；Webhook 用 httpx.MockTransport 接住，不连外网。 +封存绑定、时限、取消这几道保护另见 test_scheduler_compare_guards.py。两次周报运行用 tests/fixtures/compare_runs.py 的 +虚构运行（区域、金额、日期都是编的），按 runner 的算法封存，不真的执行；Webhook 用 httpx.MockTransport 接住，不连外网。 """ from __future__ import annotations @@ -32,7 +35,7 @@ from app.db.base import SessionLocal from app.db.models import Artifact, Run, RunEvent, Schedule, ScheduleFire, Setting, Workflow from app.main import app from app.scheduler import auto_compare, webhook -from tests.fixtures.compare_runs import NEXT_ROWS, Fake, week +from tests.fixtures.compare_runs import NEXT_ROWS, Fake, region_sql, week UTC = timezone.utc T0 = datetime(2026, 9, 7, 1, 0, tzinfo=UTC) @@ -111,8 +114,9 @@ async def new_schedule(workflow_id: str, *, fmt: str = "generic", url: str | Non async def persist(fake: Fake, workflow_id: str, *, finished: datetime, schedule_id: str | None = None, - status: str = "succeeded", seal: bool = True) -> str: - """把一次虚构运行落库：工件进工件库，事件进运行记录，按 runner 的算法封存（同 test_compare_api.py）。""" + status: str = "succeeded", seal: bool = True, seal_upto: int | None = None) -> str: + """把一次虚构运行落库：工件进工件库，事件进运行记录，按 runner 的算法封存（同 test_compare_api.py）。 + seal_upto 只封到第几条事件：失败时封存过、接着跑完又写下了终态，新的封存还没补上的那一刻。""" for content in fake.store.values(): await put_json(content, kind="compare_fixture") async with SessionLocal() as session: @@ -128,11 +132,39 @@ async def persist(fake: Fake, workflow_id: str, *, finished: datetime, schedule_ rows.append((seq, etype, node_id, data)) run.last_seq = len(rows) if seal: - run.manifest_hash, run.manifest_seq = manifest_hash(rows), rows[-1][0] + covered = [r for r in rows if seal_upto is None or r[0] <= seal_upto] + run.manifest_hash, run.manifest_seq = manifest_hash(covered), covered[-1][0] await session.commit() return run.id +async def reseal(run_id: str, fake: Fake, *, status: str, finished: datetime) -> None: + """一次运行被接着跑完（或者被重新封存）：事件换成 fake 的那一份，按新的事件封存，状态、结束时间一并改。""" + for content in fake.store.values(): + await put_json(content, kind="compare_fixture") + async with SessionLocal() as session: + await session.execute(delete(RunEvent).where(RunEvent.run_id == run_id)) + rows = [] + for seq, etype, node_id, data in fake.events: + session.add(RunEvent(run_id=run_id, seq=seq, type=etype, node_id=node_id, ts=float(seq), data=data)) + rows.append((seq, etype, node_id, data)) + run = await session.get(Run, run_id) + run.last_seq = len(rows) + run.manifest_hash, run.manifest_seq = manifest_hash(rows), rows[-1][0] + run.status, run.finished_at = status, finished + await session.commit() + + +def failed_week(run_id: str, rows=NEXT_ROWS) -> Fake: + """在 Agent 节点失败的一次周报：取数做完了，Agent 没交回结果，口径卡没有执行，一个指标都没有。""" + run = Fake(run_id) + run.tool("fetch", region_sql(), rows=rows) + run.add("node.started", "bot", {"node_type": "agent"}) + run.add("node.failed", "bot", {"error": "模型接入超时"}) + run.add("run.failed", None, {"error": "节点「区域分析」失败"}) + return run + + async def seal(run_id: str) -> None: async with SessionLocal() as session: run = await session.get(Run, run_id) @@ -206,8 +238,14 @@ async def test_finished_run_is_compared_with_the_previous_run_of_its_schedule(re from app.api.compare import compare_runs stored = load(artifact) - assert stored == json.loads(canonical_json(await compare_runs(target_id, base_id))) - assert stored["base"]["id"] == base_id and stored["summary"] == SUMMARY + # 整份对比结果，连同比的时候两边的清单哈希（封存变了，这份结果就不再作数） + assert stored["result"] == json.loads(canonical_json(await compare_runs(target_id, base_id))) + assert stored["result"]["base"]["id"] == base_id and stored["result"]["summary"] == SUMMARY + async with SessionLocal() as session: + runs = await session.scalars(select(Run).where(Run.id.in_((target_id, base_id)))) + seals = {r.id: r.manifest_hash for r in runs} + assert (stored["target_manifest"], stored["base_manifest"], stored["previous"]) == ( + seals[target_id], seals[base_id], True) async with SessionLocal() as session: rows = list((await session.execute(select(Artifact).where(Artifact.id == artifact))).scalars()) # 对比结果不是运行产出的：不挂在运行名下，不进这次运行的工件列表 @@ -215,7 +253,7 @@ async def test_finished_run_is_compared_with_the_previous_run_of_its_schedule(re assert artifact not in [a["id"] for a in (await client.get(f"/api/runs/{target_id}/artifacts")).json()] items = {f["id"]: f for f in (await client.get(f"/api/schedules/{sid}/fires")).json()["items"]} - assert items[fire_id]["compare"] == {"base_run_id": base_id, "summary": SUMMARY} + assert items[fire_id]["compare"] == {"base_run_id": base_id, "summary": SUMMARY, "previous": True} assert items[base_fire]["compare"] is None [listed] = [s for s in (await client.get("/api/schedules")).json()["items"] if s["id"] == sid] assert listed["last_fire"]["id"] == fire_id and listed["last_fire"]["compare"]["base_run_id"] == base_id @@ -223,7 +261,7 @@ async def test_finished_run_is_compared_with_the_previous_run_of_its_schedule(re [body] = recorder.bodies() link = f"{BASE_URL}/runs/compare?target={target_id}&base={base_id}" assert body["event"] == "finished" and LINE in body["text"] and f"查看对比：{link}" in body["text"] - assert body["compare"] == {"base_run_id": base_id, "summary": SUMMARY, "url": link} + assert body["compare"] == {"base_run_id": base_id, "summary": SUMMARY, "previous": True, "url": link} query = parse_qs(urlsplit(body["compare"]["url"]).query) assert query == {"target": [target_id], "base": [base_id]} assert no_numbers(recorder.requests[0].content.decode()) == [] @@ -231,25 +269,79 @@ async def test_finished_run_is_compared_with_the_previous_run_of_its_schedule(re assert await webhook.notify_due() == 0 and (await fire_row(fire_id)).compare_artifact == artifact -@pytest.mark.parametrize("fmt, url", [("feishu", FEISHU), ("generic", GENERIC)]) -async def test_failed_run_notice_carries_the_counts_too(recorder, fmt, url): - """失败的运行也封存：和上一次比一次，失败通知同样带计数摘要""" +@pytest.mark.parametrize("status, fmt, url", [("failed", "feishu", FEISHU), ("failed", "generic", GENERIC), + ("cancelled", "generic", GENERIC)]) +async def test_failed_or_cancelled_run_is_not_compared(recorder, client, status, fmt, url): + """失败、取消的运行指标不全（这次在 Agent 节点失败，口径卡没有执行）：和上一次比只会得出「移除 2 个」，收到通知 + 的人会以为报告删了两个指标。不比：触发记录上没有对比，失败通知不带计数那一行和对比页链接""" await public_address() - _, base_id, target_id, fire_id = await weekly_pair(target_status="failed", fmt=fmt, url=url) - assert await webhook.notify_due() == 1 - assert (await fire_row(fire_id)).compare_artifact - raw = recorder.requests[0].content.decode() - text = json.loads(raw)["content"]["text"] if fmt == "feishu" else json.loads(raw)["text"] - assert "运行失败" in text and LINE in text and f"base={base_id}" in text and f"target={target_id}" in text - assert no_numbers(raw) == [] - - -async def test_cancelled_run_is_compared_but_not_pushed(recorder, client): - """取消的运行不推送（有人手动停的），对比照样算：触发历史里一样能点「对比上一次」""" - sid, base_id, _, fire_id = await weekly_pair(target_status="cancelled") - assert await webhook.notify_due() == 0 and recorder.requests == [] + workflow = await new_workflow() + sid = await new_schedule(workflow, fmt=fmt, url=url) + base_id = await persist(week("r1"), workflow, finished=T0, schedule_id=sid) + await fire_for(sid, base_id, T0 - timedelta(minutes=2), notified="succeeded") + target_id = await persist(failed_week("r2"), workflow, finished=T0 + WEEK, schedule_id=sid, status=status) + fire_id = await fire_for(sid, target_id, T0 + WEEK - timedelta(minutes=2)) + # 取消的不推送（有人手动停的） + assert await webhook.notify_due() == (1 if status == "failed" else 0) + assert (await fire_row(fire_id)).compare_artifact is None items = {f["id"]: f for f in (await client.get(f"/api/schedules/{sid}/fires")).json()["items"]} - assert items[fire_id]["compare"]["base_run_id"] == base_id + assert items[fire_id]["compare"] is None + if status == "failed": + raw = recorder.requests[0].content.decode() + text = json.loads(raw)["content"]["text"] if fmt == "feishu" else json.loads(raw)["text"] + assert "运行失败" in text and f"{BASE_URL}/runs/{target_id}" in text + for absent in ("相比", "移除", "查看对比", "/runs/compare"): + assert absent not in text, text + if fmt == "generic": + assert json.loads(raw)["compare"] is None + + +async def test_after_a_failed_run_the_base_is_the_last_successful_one(recorder, client): + """上一次失败了：基准取再往前、成功的那一次——拿失败的那次当基准，每个指标都成了「新增」。紧挨着的上一次不是 + 它，通知和触发历史写明比的是哪一次（短编号和发起的日期，按定时任务的时区）""" + await public_address() + workflow = await new_workflow() + sid = await new_schedule(workflow) + first = await persist(week("r1"), workflow, finished=T0, schedule_id=sid) + await fire_for(sid, first, T0 - timedelta(minutes=2), notified="succeeded") + failed = await persist(failed_week("r2"), workflow, finished=T0 + WEEK, schedule_id=sid, status="failed") + await fire_for(sid, failed, T0 + WEEK - timedelta(minutes=2), notified="failed") + target_id = await persist(week("r3", NEXT_ROWS), workflow, finished=T0 + 2 * WEEK, schedule_id=sid) + fire_id = await fire_for(sid, target_id, T0 + 2 * WEEK - timedelta(minutes=2)) + + assert await webhook.notify_due() == 1 + stored = load((await fire_row(fire_id)).compare_artifact) + assert stored["result"]["base"]["id"] == first and stored["previous"] is False + [body] = recorder.bodies() + # 第一次发起于北京时间 2026-09-07 08:58 + line = f"与上一次成功的运行（#{first[:6]}，2026-09-07）相比：只数据变 2 个、口径与数据都变 0 个、SQL 结构变 0 个" + assert line in body["text"] and "新增" not in body["text"], body["text"] + assert f"base={first}" in body["text"] and failed not in body["text"] + assert (body["compare"]["base_run_id"], body["compare"]["previous"]) == (first, False) + items = {f["id"]: f for f in (await client.get(f"/api/schedules/{sid}/fires")).json()["items"]} + assert items[fire_id]["compare"] == {"base_run_id": first, "summary": SUMMARY, "previous": False} + + +async def test_base_follows_launch_order_not_finish_time(recorder): + """第 2 周失败、第 3 周照常成功之后，有人把第 2 周接着跑完：它结束得比第 3 周晚，但它的上一次仍是第 1 周。以前 + 按结束时间挑，拿后一次触发的第 3 周当基准，「与上一次相比」的变化方向是反的。第 3 周已经比过的那一份不受影响""" + workflow = await new_workflow() + sid = await new_schedule(workflow, url=None) + first = await persist(week("r1"), workflow, finished=T0, schedule_id=sid) + await fire_for(sid, first, T0 - timedelta(minutes=2), notified="succeeded") + second = await persist(failed_week("r2"), workflow, finished=T0 + WEEK, schedule_id=sid, status="failed") + second_fire = await fire_for(sid, second, T0 + WEEK - timedelta(minutes=2)) + third = await persist(week("r3", NEXT_ROWS), workflow, finished=T0 + 2 * WEEK, schedule_id=sid) + third_fire = await fire_for(sid, third, T0 + 2 * WEEK - timedelta(minutes=2)) + await webhook.notify_due() + third_artifact = (await fire_row(third_fire)).compare_artifact + assert load(third_artifact)["result"]["base"]["id"] == first + + await reseal(second, week("r2", NEXT_ROWS), status="succeeded", finished=T0 + 2 * WEEK + timedelta(days=1)) + await webhook.notify_due() + stored = load((await fire_row(second_fire)).compare_artifact) + assert stored["result"]["base"]["id"] == first and stored["previous"] is True + assert (await fire_row(third_fire)).compare_artifact == third_artifact async def test_first_run_of_a_schedule_has_no_compare(recorder, client): @@ -276,7 +368,7 @@ async def test_waits_for_the_seal_before_comparing(recorder): assert (await fire_row(fire_id)).compare_artifact is None and recorder.requests == [] await seal(target_id) assert await webhook.notify_due() == 1 - assert load((await fire_row(fire_id)).compare_artifact)["base"]["id"] == base_id + assert load((await fire_row(fire_id)).compare_artifact)["result"]["base"]["id"] == base_id assert LINE in recorder.bodies()[0]["text"] @@ -327,18 +419,28 @@ async def test_compare_failure_only_logs_and_the_notice_still_goes_out(recorder, assert "封存核对不一致" in caplog.text -async def test_compare_is_hidden_once_the_run_is_running_again(client): - """失败后被继续执行（不在终态）：对比页只认封存的运行，接口不给 compare；再次结束时重新比""" - sid, base_id, target_id, fire_id = await weekly_pair(target_status="failed", url=None) +async def test_a_failed_run_is_compared_once_it_is_continued_to_success(client): + """失败时不比；接着跑（不在终态）时也没有；跑完、成功并重新封存之后按新的封存比一次""" + workflow = await new_workflow() + sid = await new_schedule(workflow, url=None) + base_id = await persist(week("r1"), workflow, finished=T0, schedule_id=sid) + await fire_for(sid, base_id, T0 - timedelta(minutes=2), notified="succeeded") + target_id = await persist(failed_week("r2"), workflow, finished=T0 + WEEK, schedule_id=sid, status="failed") + fire_id = await fire_for(sid, target_id, T0 + WEEK - timedelta(minutes=2)) + + async def shown(): + items = (await client.get(f"/api/schedules/{sid}/fires")).json()["items"] + return {f["id"]: f for f in items}[fire_id] + await webhook.notify_due() - assert (await fire_row(fire_id)).compare_artifact + assert (await fire_row(fire_id)).compare_artifact is None and (await shown())["compare"] is None await set_run(target_id, status="running") - [item] = [f for f in (await client.get(f"/api/schedules/{sid}/fires")).json()["items"] if f["id"] == fire_id] - assert item["run_status"] == "running" and item["compare"] is None - await set_run(target_id, status="succeeded", finished_at=T0 + WEEK + timedelta(hours=1)) await webhook.notify_due() - [item] = [f for f in (await client.get(f"/api/schedules/{sid}/fires")).json()["items"] if f["id"] == fire_id] - assert item["compare"] == {"base_run_id": base_id, "summary": SUMMARY} + item = await shown() + assert item["run_status"] == "running" and item["compare"] is None + await reseal(target_id, week("r2", NEXT_ROWS), status="succeeded", finished=T0 + WEEK + timedelta(hours=1)) + await webhook.notify_due() + assert (await shown())["compare"] == {"base_run_id": base_id, "summary": SUMMARY, "previous": True} async def test_compare_is_hidden_once_the_base_run_is_gone_or_running(client): diff --git a/backend/tests/test_scheduler_compare_guards.py b/backend/tests/test_scheduler_compare_guards.py new file mode 100644 index 00000000..1e1754f7 --- /dev/null +++ b/backend/tests/test_scheduler_compare_guards.py @@ -0,0 +1,312 @@ +"""定时运行自动对比的几道保护（app/scheduler/auto_compare.py、webhook.notify_due）： + +- 和封存绑定：运行的封存要盖住它最后一条结束事件才算封存完。runner 先写终态、再另起会话封存，失败后接着跑完的运行 + 在两步之间还带着失败时的清单哈希——这时去比，只比得到旧封存范围里的事件，通知里的清单哈希也是旧的。等不来新的 + 封存（两步之间进程没了）时照常推送，不带对比、不写旧的哈希 +- 对比结果记下两边的清单哈希：任何一边后来变了，触发历史就不再给这份对比（悬停的计数和点进去的对比页对不上）； + 对比途中封存变了，结果不记 +- 时限：对比放在认领之前、锁外做，最多等 WAIT_S 秒，同一时刻最多 SLOTS 个。慢的对比不压住这条通知，也不压住别的 + 定时任务的通知；扫描在对比途中被取消（停机），这次触发还没认领，下次照常比、照常推。同一份封存已经比过的（比完 + 还没推送就停机了），下次直接用记下的结果 + +运行、任务名、金额都是虚构的（tests/fixtures/compare_runs.py），不真的执行；Webhook 用 httpx.MockTransport 接住。 +""" +from __future__ import annotations + +import asyncio +import logging +import time +from datetime import datetime, timedelta, timezone + +import httpx +import pytest +from httpx import ASGITransport, AsyncClient +from sqlalchemy import select + +from app import scheduler +from app.api import compare as compare_api +from app.core.artifact_store import load +from app.core.config import Settings +from app.db.base import SessionLocal +from app.db.models import Run, RunEvent +from app.main import app +from app.scheduler import auto_compare, webhook +from tests.fixtures.compare_runs import BASE_ROWS, NEXT_ROWS, Fake, default_metrics, region_sql, week +from tests.test_scheduler_compare import ( # noqa: F401 - clean、recorder 是夹具，按名字引入 + BASE_URL, LINE, T0, WEEK, Recorder, clean, fire_for, fire_row, new_schedule, new_workflow, persist, + public_address, recorder, reseal, seal, set_run, weekly_pair, +) + +UTC = timezone.utc + + +@pytest.fixture +async def client(): + async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as c: + yield c + + +@pytest.fixture +def slow_compare(monkeypatch): + """对比要 5 秒：真的对比之前先等着。返回原来的函数，用例里可以换回去""" + real = compare_api.compare_runs + + async def slow(*args, **kwargs): + await asyncio.sleep(5) + return await real(*args, **kwargs) + + monkeypatch.setattr(compare_api, "compare_runs", slow) + return real + + +def failed_then_continued(run_id: str, rows=NEXT_ROWS) -> tuple[Fake, int]: + """在 Agent 节点失败、之后从断点接着跑完的一次周报。返回 (运行, 失败时封存到第几条事件)。""" + run = Fake(run_id) + sql = region_sql() + fetch = run.tool("fetch", sql, rows=rows) + run.add("node.started", "bot", {"node_type": "agent"}) + run.add("node.failed", "bot", {"error": "模型接入超时"}) + run.add("run.failed", None, {"error": "节点「区域分析」失败"}) + failed_at = len(run.events) + run.add("run.resumed", None, {"from": ["bot"], "message": "从「区域分析」继续运行"}) + bot = run.agent("bot", sql, rows=rows) + run.card(default_metrics(fetch, bot, rows, rows)) + run.add("run.finished", None, {"output": {}}) + return run, failed_at + + +async def previous_week(*, url: str | None = None): + """一个定时任务和它上一次成功、已推送过的周报。返回 (工作流, 任务, 上一次)。""" + workflow = await new_workflow() + sid = await new_schedule(workflow, url=url) if url else await new_schedule(workflow) + base_id = await persist(week("r1"), workflow, finished=T0, schedule_id=sid) + await fire_for(sid, base_id, T0 - timedelta(minutes=2), notified="succeeded") + return workflow, sid, base_id + + +async def seal_of(run_id: str) -> str | None: + async with SessionLocal() as session: + return (await session.get(Run, run_id)).manifest_hash + + +# -------------------------------------------------------------------------- +# 和封存绑定 +# -------------------------------------------------------------------------- + + +async def test_waits_for_the_seal_of_this_ending_not_the_failed_one(recorder): + """失败后接着跑完：终态和结束时间已经写下，封存还是失败时那一份（只盖到 run.failed）。这一拍不比、不推，新的 + 封存补上之后再比，比的是完整的一次，通知里的清单哈希是新的""" + await public_address() + workflow, sid, base_id = await previous_week() + fake, failed_at = failed_then_continued("r2") + target_id = await persist(fake, workflow, finished=datetime.now(UTC), schedule_id=sid, seal_upto=failed_at) + fire_id = await fire_for(sid, target_id, T0 + WEEK - timedelta(minutes=2)) + stale = await seal_of(target_id) + + assert await webhook.notify_due() == 0 + assert recorder.requests == [] and (await fire_row(fire_id)).compare_artifact is None + await seal(target_id) + fresh = await seal_of(target_id) + assert fresh != stale + assert await webhook.notify_due() == 1 + stored = load((await fire_row(fire_id)).compare_artifact) + assert (stored["target_manifest"], stored["result"]["base"]["id"]) == (fresh, base_id) + [body] = recorder.bodies() + assert LINE in body["text"] and body["manifest_hash"] == fresh[:12] + + +async def test_a_seal_that_never_catches_up_is_not_used(recorder): + """新的封存一直没补上（写完终态、封存之前进程没了）：等过时限照常推送，但不比，也不写失败时那份旧的清单哈希""" + await public_address() + workflow, sid, _ = await previous_week() + fake, failed_at = failed_then_continued("r2") + target_id = await persist(fake, workflow, finished=datetime.now(UTC), schedule_id=sid, seal_upto=failed_at) + fire_id = await fire_for(sid, target_id, T0 + WEEK - timedelta(minutes=2)) + later = datetime.now(UTC) + timedelta(seconds=webhook.SEAL_WAIT_S + 5) + assert await webhook.notify_due(later) == 1 + [body] = recorder.bodies() + assert (body["event"], body["manifest_hash"], body["compare"]) == ("finished", None, None) + assert "清单哈希" not in body["text"] and "相比" not in body["text"] + assert (await fire_row(fire_id)).compare_artifact is None + + +async def test_approval_notice_does_not_carry_the_seal_of_an_earlier_failure(recorder): + """失败后接着跑、停在审批上：运行上还留着失败时的清单哈希，那不是这一次的封存，「等待审批」里不写""" + workflow, sid, _ = await previous_week() + fake, failed_at = failed_then_continued("r2") + target_id = await persist(fake, workflow, finished=T0 + WEEK, schedule_id=sid, status="interrupted", + seal_upto=failed_at) + await set_run(target_id, finished_at=None) + async with SessionLocal() as session: + from app.db.models import Approval + + session.add(Approval(run_id=target_id, node_id="card", title="审核周报口径", status="pending")) + await session.commit() + await fire_for(sid, target_id, T0 + WEEK - timedelta(minutes=2)) + assert await webhook.notify_due() == 1 + [body] = recorder.bodies() + assert (body["event"], body["manifest_hash"]) == ("awaiting_approval", None) + assert "清单哈希" not in body["text"] + + +@pytest.mark.parametrize("side", ["target", "base"]) +async def test_fire_history_drops_a_compare_whose_seal_changed(client, side): + """比过之后哪一边被重新封存（内容变了）：记下的计数已经不是对比页现场算出来的那样，触发历史不再给这份对比""" + sid, base_id, target_id, fire_id = await weekly_pair(url=None) + await webhook.notify_due() + + async def shown(): + items = (await client.get(f"/api/schedules/{sid}/fires")).json()["items"] + return {f["id"]: f for f in items}[fire_id]["compare"] + + assert (await shown())["base_run_id"] == base_id + # 换成和另一边一样的数：现场再比就是「未变」 + run_id, rows = (target_id, BASE_ROWS) if side == "target" else (base_id, NEXT_ROWS) + async with SessionLocal() as session: + finished = (await session.get(Run, run_id)).finished_at + await reseal(run_id, week("rx", rows), status="succeeded", finished=finished) + assert await shown() is None + # 对比页现场算出来的和记下的确实不一样 + assert (await compare_api.compare_runs(target_id, base_id))["summary"]["unchanged"] == 2 + + +async def test_result_is_dropped_when_a_seal_changes_mid_compare(recorder, monkeypatch): + """对比途中基准被重新封存：比出来的是旧封存的结果，不记进触发记录、不写进通知""" + await public_address() + _, base_id, _, fire_id = await weekly_pair() + real = compare_api.compare_runs + + async def resealing(run_id, other): + out = await real(run_id, other) + await set_run(base_id, manifest_hash="ab" * 32) + return out + + monkeypatch.setattr(compare_api, "compare_runs", resealing) + assert await webhook.notify_due() == 1 + assert (await fire_row(fire_id)).compare_artifact is None + [body] = recorder.bodies() + assert body["compare"] is None and "相比" not in body["text"] + + +# -------------------------------------------------------------------------- +# 时限、取消、并发 +# -------------------------------------------------------------------------- + + +async def test_a_slow_compare_is_cut_off_and_the_notice_goes_out(recorder, monkeypatch, caplog, slow_compare): + """对比超过时限：通知不等它，照常推送、不带对比，记一条警告""" + monkeypatch.setattr(auto_compare, "WAIT_S", 0.3, raising=False) + await public_address() + _, _, target_id, fire_id = await weekly_pair() + began = time.monotonic() + with caplog.at_level(logging.WARNING, logger="app.scheduler"): + assert await webhook.notify_due() == 1 + assert time.monotonic() - began < 2 + [body] = recorder.bodies() + assert (body["event"], body["compare"]) == ("finished", None) and "相比" not in body["text"] + assert f"{BASE_URL}/runs/{target_id}" in body["text"] + assert (await fire_row(fire_id)).compare_artifact is None + assert "超过" in caplog.text and target_id in caplog.text + + +async def test_scan_cancelled_mid_compare_pushes_next_time(recorder, monkeypatch, slow_compare): + """停机时通知扫描只等 1.5 秒就被取消：撞在对比途中的那次触发还没认领，下次启动照常比、照常推。以前先认领、后比， + 取消之后这条「运行结束」再也不会推""" + await public_address() + _, _, _, fire_id = await weekly_pair() + with pytest.raises(TimeoutError): + await asyncio.wait_for(webhook.notify_due(), timeout=0.5) + assert (await fire_row(fire_id)).notified_status is None and recorder.requests == [] + monkeypatch.setattr(compare_api, "compare_runs", slow_compare) + assert await webhook.notify_due() == 1 + [body] = recorder.bodies() + assert LINE in body["text"] + assert (await fire_row(fire_id)).notified_status == "succeeded" + + +async def test_a_slow_compare_does_not_hold_other_schedules(monkeypatch, slow_compare): + """任务 A 的对比慢：调度循环里通知扫描同一时刻只跑一份，以前 A 比多久，之后才结束的任务 B（第一次，不用比） + 的通知就要等多久。现在 A 的对比等过时限就不等了，B 的通知在下一拍推出""" + rec = Recorder() + monkeypatch.setattr(webhook, "_transport", httpx.MockTransport(rec)) + monkeypatch.setattr(webhook, "RETRY_DELAYS", (0, 0, 0)) + monkeypatch.setattr(auto_compare, "WAIT_S", 0.5, raising=False) + await weekly_pair() + on = Settings(_env_file=None, scheduler=True) + assert await scheduler.start(interval=0.05, config=on) is True + try: + await asyncio.sleep(0.2) + workflow = await new_workflow("仓库日报") + sid = await new_schedule(workflow, name="仓库日报 · 每天") + run_id = await persist(week("b1"), workflow, finished=T0, schedule_id=sid) + await fire_for(sid, run_id, T0 - timedelta(minutes=2)) + began = time.monotonic() + while not any(b.get("run_id") == run_id for b in rec.bodies()) and time.monotonic() - began < 8: + await asyncio.sleep(0.02) + waited = time.monotonic() - began + finally: + await scheduler.stop() + assert any(b.get("run_id") == run_id for b in rec.bodies()) + assert waited < 2, f"任务 B 的通知等了 {waited:.2f} 秒" + + +async def test_compares_take_turns(monkeypatch): + """同一拍里好几个任务都要比：同一时刻最多 SLOTS 个，不一起占满线程池（接口请求也要用）""" + monkeypatch.setattr(auto_compare, "SLOTS", 2, raising=False) + real = compare_api.compare_runs + active = peak = 0 + + async def counted(*args, **kwargs): + nonlocal active, peak + active += 1 + peak = max(peak, active) + try: + await asyncio.sleep(0.2) + return await real(*args, **kwargs) + finally: + active -= 1 + + monkeypatch.setattr(compare_api, "compare_runs", counted) + fires = [(await weekly_pair(url=None))[3] for _ in range(4)] + await webhook.notify_due() + assert peak == 2 + assert all([(await fire_row(f)).compare_artifact for f in fires]) + + +async def test_a_compare_already_made_for_this_seal_is_reused(recorder, monkeypatch): + """比完、还没认领就停机了，结果已经记在触发记录上：下次直接用，不再算一遍""" + await public_address() + _, _, target_id, fire_id = await weekly_pair() + assert await auto_compare.compare_fire(fire_id, target_id) is not None + artifact = (await fire_row(fire_id)).compare_artifact + assert artifact + calls = 0 + + async def counted(*args, **kwargs): + nonlocal calls + calls += 1 + raise RuntimeError("不该再算一遍") + + monkeypatch.setattr(compare_api, "compare_runs", counted) + auto_compare._BRIEFS.clear() + assert await webhook.notify_due() == 1 + assert calls == 0 and LINE in recorder.bodies()[0]["text"] + assert (await fire_row(fire_id)).compare_artifact == artifact + + +async def test_seal_covers_the_last_ending(): + """封存盖没盖住最后一条结束事件：老运行（没有 manifest_seq）只看有没有哈希""" + workflow = await new_workflow() + fake, failed_at = failed_then_continued("r9") + partial = await persist(fake, workflow, finished=T0, seal_upto=failed_at) + full = await persist(failed_then_continued("r8")[0], workflow, finished=T0) + legacy = await persist(week("r7"), workflow, finished=T0) + await set_run(legacy, manifest_seq=None) + unsealed = await persist(week("r6"), workflow, finished=T0, seal=False) + async with SessionLocal() as session: + got = {rid: await auto_compare.sealed_now(session, await session.get(Run, rid)) + for rid in (partial, full, legacy, unsealed)} + assert await session.scalar(select(RunEvent.seq).where( + RunEvent.run_id == partial, RunEvent.type == "run.finished")) > failed_at + assert got == {partial: False, full: True, legacy: True, unsealed: False} diff --git a/backend/tests/test_scheduler_notice_rules.py b/backend/tests/test_scheduler_notice_rules.py index dfc11af7..6d098db4 100644 --- a/backend/tests/test_scheduler_notice_rules.py +++ b/backend/tests/test_scheduler_notice_rules.py @@ -1,9 +1,10 @@ """定时任务通知（app/scheduler/webhook.py）的两条规则： - 失败通知的「处理」行不许诺能继续运行：服务重启时还在排队、从没开始执行的运行（被改成失败、没有开始时间）没有 - 断点，「修正后可继续运行」是错的。按有没有开始执行分开写——没开始的另写一行原因（没有可继续的进度），处理一律是 - 中性的「请在运行记录中查看原因」；任务还在排期上时补一句「下一次触发会照常发起」：失败的运行不挡之后的触发， - 和挂起的那种（之后的触发都会跳过）说清区别 + 断点，「修正后可继续运行」是错的。按有没有开始执行分开写——没开始的另写一行原因（没有可继续的进度），处理行不再 + 让人去「查看原因」（原因刚写过），任务还在排期上时只写「下一次触发会照常发起」，不在排期上时写怎么重新发起；开始 + 执行过的写「请在 AgentLab 的「记录」中查看失败原因」（和挂起那句一样指到导航里的「记录」），任务还在排期上时补 + 「下一次触发会照常发起」：失败的运行不挡之后的触发，和挂起的那种（之后的触发都会跳过）说清区别 - 扫描窗口（SCAN_DAYS）之外的触发：停在审批、有待审批的运行不受窗口限制——创建 30 天以后才停到审批上的照样推送 「等待审批」；运行记录在窗口内有变化的（等了一个多月的审批批完、跑完）照样推送结束。早就结束、记录多日没有变化 的旧触发仍不扫 @@ -27,6 +28,8 @@ from app.scheduler import webhook UTC = timezone.utc GENERIC = "https://hooks.example.com/agentlab/inbox-3e9d" NOT_STARTED = "运行在开始执行前就已中止（通常是服务重启时仍在排队），没有可继续的进度" +LOOK = "请在 AgentLab 的「记录」中查看失败原因" +RERUN = "如需重新发起，请在 AgentLab 的「定时任务」中点击「立即运行」" class Recorder: @@ -106,12 +109,15 @@ async def notified(fire_id: str) -> str | None: def test_a_run_that_never_started_is_not_promised_a_continue(): - """服务重启时还在排队的运行没有断点：不写「修正后可继续运行」，另写一行原因说没有可继续的进度""" + """服务重启时还在排队的运行没有断点：不写「修正后可继续运行」，另写一行原因说没有可继续的进度。处理行不再让人 + 去「查看原因」（原因刚写过）：还在排期上的只说下一次照常，不在排期上的说怎么重新发起""" _, lines, _ = webhook.message(notice(started_at=None, next_fire=True)) shown = dict(lines) assert shown["原因"] == NOT_STARTED - assert shown["处理"] == "请在 AgentLab 的运行记录中查看原因；下一次触发会照常发起" - assert not any("继续运行" in value for _, value in lines), lines + assert shown["处理"] == "下一次触发会照常发起" + assert not any("继续运行" in value or "查看" in value for _, value in lines), lines + _, lines, _ = webhook.message(notice(started_at=None, next_fire=False)) + assert dict(lines)["处理"] == RERUN def test_a_failed_run_gets_a_neutral_action_without_promising_a_continue(): @@ -120,14 +126,14 @@ def test_a_failed_run_gets_a_neutral_action_without_promising_a_continue(): _, lines, _ = webhook.message(notice(error_node="汇总库存", next_fire=True)) shown = dict(lines) assert shown["失败节点"] == "「汇总库存」" and "原因" not in shown - assert shown["处理"] == "请在 AgentLab 的运行记录中查看原因；下一次触发会照常发起" + assert shown["处理"] == f"{LOOK}；下一次触发会照常发起" assert "修正后可继续运行" not in webhook._text(notice(error_node="汇总库存")) def test_next_fire_is_only_promised_while_the_schedule_is_on(): """任务已暂停（或往后不再触发）时不说「下一次触发会照常发起」""" _, lines, _ = webhook.message(notice(next_fire=False)) - assert dict(lines)["处理"] == "请在 AgentLab 的运行记录中查看原因" + assert dict(lines)["处理"] == LOOK async def test_restart_failure_notice_end_to_end(recorder): @@ -140,12 +146,12 @@ async def test_restart_failure_notice_end_to_end(recorder): by_fire = {b["fire_id"]: b for b in recorder.bodies()} on, off = by_fire[on_fire], by_fire[off_fire] assert (on["event"], on["status"], on["reason"]) == ("failed", "failed", NOT_STARTED) - assert "处理：请在 AgentLab 的运行记录中查看原因；下一次触发会照常发起" in on["text"] - assert "处理：请在 AgentLab 的运行记录中查看原因\n" in off["text"] + "\n" and "照常发起" not in off["text"] + assert "处理：下一次触发会照常发起" in on["text"] + assert f"处理：{RERUN}" in off["text"] and "照常发起" not in off["text"] for body in (on, off): assert "继续运行" not in body["text"], body["text"] # 运行记录上的原始报错不进通知 - assert "请重新发起运行" not in body["text"] + assert "没有可恢复的进度" not in body["text"] # -------------------------------------------------------------------------- diff --git a/backend/tests/test_scheduler_webhook.py b/backend/tests/test_scheduler_webhook.py index ca32dca3..635a59ab 100644 --- a/backend/tests/test_scheduler_webhook.py +++ b/backend/tests/test_scheduler_webhook.py @@ -179,7 +179,7 @@ def test_failed_run_text_points_to_the_node_without_repeating_itself(): shown = dict(lines) assert title.endswith("运行失败") and shown["状态"] == "失败" # 处理只说去哪里看，不许诺能继续运行（没有断点的运行续不了，见 test_scheduler_notice_rules.py） - assert shown["失败节点"] == "「查询门店销售」" and shown["处理"] == "请在 AgentLab 的运行记录中查看原因" + assert shown["失败节点"] == "「查询门店销售」" and shown["处理"] == "请在 AgentLab 的「记录」中查看失败原因" # 不再写「原因：运行失败，详情请查看运行记录」这种同义反复 assert "原因" not in shown