mirror of
https://github.com/ThearchyHelios/AgentLab.git
synced 2026-11-15 00:11:59 +00:00
定时任务自动对比复核整改（后端）：只比成功的运行、按发起先后挑上一次成功的基准、和两边的封存绑定，对比挪到认领之前在锁外做并限时限并发；失败通知的处理行不再重复原因
- 只比成功结束的运行，基准也只取成功的：失败、取消的运行指标不全，和上一次比只会得出「移除 N 个」，拿失败的那次当 基准每个指标都成了「新增」。基准改成自己查同一定时任务、同一工作流、比这次先发起的运行里成功并封存的最近一次， 不再用对比接口的 default_base（按结束时间挑：老的失败运行事后被接着跑完，会拿后一次触发当它的「上一次」）。 中间隔着失败、取消的运行时，通知写「与上一次成功的运行（#短编号，日期）相比」，Fire.compare 多一个 previous - 和封存绑定：运行的封存要盖住最后一条结束事件才算封存完（sealed_now）。runner 先写终态、再另起会话封存，失败后 接着跑完的运行在两步之间还带着失败时的清单哈希，以前这一拍会直接认领、按旧封存比、通知里写旧的清单哈希；现在 等新的封存，等不来就照常推送但不比、不写哈希。停在审批上的通知也不再写失败时留下的哈希。对比工件连同两边的清单 哈希一起存，触发历史发现任何一边不在终态或清单哈希变了就给 null；对比途中封存变了不记（条件写） - 时机和时限：锁里只挑出状态有变化的触发，认领挪到对比之后、锁外按条件 UPDATE 做。对比最多等 20 秒（含排队）， 同一时刻最多 2 个；超时只记警告，通知不带对比照常推送。停机时扫描在对比途中被取消，这次触发还没认领，下次照常 比、照常推（以前先认领、后比，这条通知会永远丢掉）。同一份封存已经比过的直接用记下的结果 - 失败通知：从没开始执行的，原因刚写过，处理行不再让人「查看原因」，任务在排期上写「下一次触发会照常发起」，停用 的写怎么重新发起；开始执行过的改成「请在 AgentLab 的「记录」中查看失败原因」，和挂起那句指向同一个导航入口 - 新增 test_scheduler_compare_guards.py（封存绑定、时限、取消、并发、复用），test_scheduler_compare.py 补真实缺 指标的失败运行、上一次失败、补跑老运行几种用例；去掉实现逐项验证过会变红
This commit is contained in:
ThearchyHelios
@@ -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]
|
||||
|
||||
|
||||
@@ -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、验收副本）不管从哪条路径起服务，
|
||||
|
||||
@@ -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:
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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):
|
||||
|
||||
@@ -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}
|
||||
@@ -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"]
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------
|
||||
|
||||
@@ -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
|
||||
|
||||
|
||||
Reference in New Issue
Block a user