From f0f1b99ead921f65022ab78ba0e17a9cd2f2882b Mon Sep 17 00:00:00 2001 From: ThearchyHelios Date: Mon, 5 Oct 2026 18:45:59 +0800 Subject: [PATCH] =?UTF-8?q?=E5=AE=9A=E6=97=B6=E8=BF=90=E8=A1=8C=E7=BB=93?= =?UTF-8?q?=E6=9D=9F=E5=90=8E=E8=87=AA=E5=8A=A8=E5=92=8C=E5=90=8C=E4=B8=80?= =?UTF-8?q?=E5=AE=9A=E6=97=B6=E4=BB=BB=E5=8A=A1=E7=9A=84=E4=B8=8A=E4=B8=80?= =?UTF-8?q?=E6=AC=A1=E5=AF=B9=E6=AF=94=EF=BC=9A=E7=BB=93=E6=9E=9C=E5=AD=98?= =?UTF-8?q?=E6=88=90=E5=B7=A5=E4=BB=B6=E8=AE=B0=E5=9C=A8=E8=A7=A6=E5=8F=91?= =?UTF-8?q?=E8=AE=B0=E5=BD=95=E4=B8=8A=EF=BC=8C=E8=A7=A6=E5=8F=91=E5=8E=86?= =?UTF-8?q?=E5=8F=B2=E5=92=8C=E9=80=9A=E7=9F=A5=E5=8F=AA=E7=BB=99=E8=AE=A1?= =?UTF-8?q?=E6=95=B0=E6=91=98=E8=A6=81=E4=B8=8E=E5=AF=B9=E6=AF=94=E9=A1=B5?= =?UTF-8?q?=E9=93=BE=E6=8E=A5?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 新增 app/scheduler/auto_compare.py：通知扫描认领到运行进入终态（结束、失败、取消）的那一次之后、推送之前， 调运行对比接口的现成函数（default_base、compare_runs，不改它们）比一次。基准只取 default_base 的「同一定时 任务」那一档，退回「同一工作流」的不比——那不是这个任务的上一次。没封存的不比；比不了（基准封存核对不一致等） 记警告、意外异常记堆栈，都不影响通知。整份结果存成 run_compare 工件，不挂在运行名下（不进运行的工件列表）， 工件 id 写进 ScheduleFire.compare_artifact；再次进入终态（失败后继续执行）时按新的封存重新比、覆盖旧结果 - 接口：Fire 多一个 compare: {base_run_id, summary}（summary 同对比接口的计数）或 null。运行现在不在终态 （被继续执行）、工件读不出来时为 null；工件摘要按 id 缓存（内容寻址，不会过期），读盘在线程里 - 通知：结束、失败的通知多一行「与上一次相比：只数据变 3 个、口径与数据都变 1 个、SQL 结构变 0 个」（三类总写， 其余各类有才写，两边都没有指标时不写一串 0 个），配了对外访问地址时附对比页链接（/runs/compare?target=…&base=…）， 四种格式都带；通用格式多一个 compare 字段。只给个数，不放任何具体数字 - 没配 Webhook 的任务也比：触发历史里要用 --- backend/app/api/schedules.py | 14 +- backend/app/scheduler/__init__.py | 1 + backend/app/scheduler/auto_compare.py | 178 +++++++++++ backend/app/scheduler/webhook.py | 85 +++-- backend/tests/test_scheduler_compare.py | 395 ++++++++++++++++++++++++ backend/tests/test_schedules_api.py | 2 +- 6 files changed, 649 insertions(+), 26 deletions(-) create mode 100644 backend/app/scheduler/auto_compare.py create mode 100644 backend/tests/test_scheduler_compare.py diff --git a/backend/app/api/schedules.py b/backend/app/api/schedules.py index 9664d6c4..b96d57b8 100644 --- a/backend/app/api/schedules.py +++ b/backend/app/api/schedules.py @@ -1,4 +1,4 @@ -"""定时任务：按周期自动发起工作流运行，结束、失败、停在审批时推送 Webhook。 +"""定时任务：按周期自动发起工作流运行，结束、失败、停在审批时推送 Webhook，运行结束后和同一任务的上一次自动对比。 形状按前后端约定（Schedule、Fire 两种对象，见 _schedule_out、_fire_outs）。Webhook 地址和签名密钥加密存放，接口 只给主机和末 4 位（url_hint）、有没有密钥（has_secret）。请求体自己解析、自己校验：错误一律回中文 detail，表单 @@ -18,7 +18,7 @@ from app.api.runs import actor_of from app.core.crypto import decrypt, encrypt from app.db.base import get_session, utcnow from app.db.models import Run, Schedule, ScheduleFire, WebhookDelivery, Workflow, WorkflowVersion -from app.scheduler import cron, service, webhook +from app.scheduler import auto_compare, cron, service, webhook router = APIRouter(prefix="/api/schedules", tags=["schedules"]) @@ -278,7 +278,11 @@ 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}]（按时间先后）。""" + """Fire 对象：run_status 是运行现在的状态，notified 是这次触发的推送记录 [{event, status, at}]（按时间先后）， + compare 是运行结束后和同一任务上一次的自动对比 {base_run_id, summary}（summary 同对比接口的计数），没有为 null。 + + 运行又被继续执行、现在不在终态时 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 {} @@ -292,6 +296,9 @@ 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())) return [{ "id": f.id, "fire_at": _iso(f.fire_at), @@ -301,6 +308,7 @@ async def _fire_outs(session: AsyncSession, fires: list[ScheduleFire]) -> list[d "run_status": statuses.get(f.run_id) 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, "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 aa1ef8b6..beac3ead 100644 --- a/backend/app/scheduler/__init__.py +++ b/backend/app/scheduler/__init__.py @@ -4,6 +4,7 @@ - service.py：输入模板、和工作流入口字段的对照、发起一次触发（调度循环和「立即运行」共用） - loop.py：调度循环，条件 UPDATE 认领到点的任务 - webhook.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 new file mode 100644 index 00000000..dc464b3f --- /dev/null +++ b/backend/app/scheduler/auto_compare.py @@ -0,0 +1,178 @@ +"""定时运行结束后的自动对比：和同一定时任务的上一次运行比一次，结果存成工件，工件 id 记在触发记录上 +（ScheduleFire.compare_artifact）。 + +对比本身调运行对比接口的现成函数（api/compare.py 的 default_base、compare_runs），这里只管什么时候比、跟谁比、结果 +放哪、怎么说给触发历史和通知听： + +- 什么时候：通知扫描（webhook.notify_due）认领到运行进入终态（结束、失败、取消）的那一次之后、推送之前。和推送一样， + 同一个终态只处理一次；失败后被继续执行、又结束的，按新的封存重新比，覆盖旧的结果 +- 跟谁比：default_base 的「同一定时任务」那一档，即这个任务里更早结束、封存过的终态运行中最近的一次。它退回「同一 + 工作流、同一运行类别」时不比：那不是这个定时任务的上一次，别人手动改的输入、另一个任务的参数都会被算成变化 +- 结果：整份对比结果存成工件（kind=run_compare）。不挂在任何运行名下：它不是运行产出的，不该出现在运行的工件列表里。 + 没封存的不比（服务重启时还在排队、被改成失败的，永远不会封存）；比不了、出错只记日志，通知照常推送 +- 对外：触发历史（Fire.compare）和通知只给基准是哪一次、各类指标的个数，不给具体数字 +""" +from __future__ import annotations + +import asyncio +import logging +import threading +from typing import Any +from urllib.parse import urlencode + +from fastapi import HTTPException +from sqlalchemy import update + +from app.core import artifact_store +from app.db.base import SessionLocal +from app.db.models import Run, ScheduleFire + +logger = logging.getLogger(__name__) + +KIND = "run_compare" +#: 终态，同对比接口（api/evidence._TERMINAL）：只有这几种状态、封存过的运行能参与对比 +TERMINAL = ("succeeded", "failed", "cancelled") +#: 各类指标的叫法，和对比页一致（frontend/src/lib/terms.ts 的 RUN_COMPARE_TEXT.classLabel） +CLASS_LABEL = { + "data_only": "只数据变", + "caliber_and_data": "口径与数据都变", + "sql_changed": "SQL 结构变", + "params_only": "只参数变", + "added": "新增", + "removed": "移除", + "incomparable": "无法比较", + "unchanged": "未变", +} +#: 变化三分法的三类，0 个也写：「SQL 结构变 0 个」本身就是要告诉人的事 +CORE = ("data_only", "caliber_and_data", "sql_changed") +#: 其余各类有才写，次序同对比页的摘要 +OTHERS = ("params_only", "added", "removed", "incomparable", "unchanged") +#: 两边都没有指标时的说法：不写一串「0 个」，免得读成「什么都没变」 +NO_METRICS = "两次运行都没有指标" + +#: 读过的对比工件：工件 id → 摘要。工件按内容寻址、内容不会变，记下来不会过期；触发历史、任务列表每次打开都要读， +#: 不必每次都去盘上读一份几十 KB 的 JSON。只记读成功的，满了丢最早的。读盘在线程里、记下新算的在事件循环里， +#: 两边都动这个字典，加一把线程锁 +_BRIEFS: dict[str, dict[str, Any]] = {} +_BRIEFS_MAX = 512 +_BRIEFS_LOCK = threading.Lock() + + +def summary_text(summary: dict[str, Any]) -> str: + """计数摘要：「只数据变 3 个、口径与数据都变 1 个、SQL 结构变 0 个」，其余各类有才接在后面。只有个数。""" + counts = {key: value for key, value in summary.items() if isinstance(value, int) and not isinstance(value, bool)} + if not any(counts.values()): + return NO_METRICS + parts = [f"{CLASS_LABEL[key]} {counts.get(key, 0)} 个" for key in CORE] + parts += [f"{CLASS_LABEL[key]} {counts[key]} 个" for key in OTHERS if counts.get(key)] + return "、".join(parts) + + +def page_url(base_url: str, target: str, base: str) -> str: + """对比页的地址（对外访问地址 + /runs/compare?target=…&base=…），和界面上运行详情跳过去的是同一个页面。""" + 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，触发记录上之前的结果一并清掉（那是上一次封存的）。 + + 不抛异常：比不了（对比接口拒绝，如基准的封存核对不一致）记一条警告，意外的异常记堆栈，都不影响通知。""" + artifact: str | None = None + brief: dict[str, Any] | None = None + try: + artifact, brief = await _compare(run_id) + 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 + + +async def _compare(run_id: str) -> tuple[str | None, 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}) + _remember(artifact, brief) + return artifact, _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) + + +def _brief(result: Any) -> dict[str, Any] | None: + """对比结果里给触发历史、通知的那一点：基准是哪一次、各类指标的个数（同对比接口的 summary）。""" + if not isinstance(result, dict): + 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): + return None + return {"base_run_id": base["id"], + "summary": {key: value for key, value in summary.items() + if isinstance(value, int) and not isinstance(value, bool)}} + + +def _copy(brief: dict[str, Any]) -> dict[str, Any]: + return {"base_run_id": brief["base_run_id"], "summary": dict(brief["summary"])} + + +def _remember(artifact: str, brief: dict[str, Any]) -> None: + with _BRIEFS_LOCK: + if artifact not in _BRIEFS and len(_BRIEFS) >= _BRIEFS_MAX: + _BRIEFS.pop(next(iter(_BRIEFS))) + _BRIEFS[artifact] = brief + + +def brief_of(artifact: str) -> dict[str, Any] | None: + """读一份对比工件的摘要。文件不在、内容和哈希对不上（另记一条警告）、形状不对的返回 None，不抛异常。""" + with _BRIEFS_LOCK: + known = _BRIEFS.get(artifact) + if known is not None: + return _copy(known) + try: + brief = _brief(artifact_store.load(artifact)) + except ValueError: + logger.warning("对比工件 %s… 内容与哈希不符，触发历史中不显示这次对比", artifact[:12]) + return None + if brief is None: + return None + _remember(artifact, brief) + return _copy(brief) + + +async def briefs(artifacts: list[str]) -> dict[str, dict[str, Any]]: + """几份对比工件的摘要：工件 id → 摘要，读不出来的不在里面。读盘在线程里做。""" + wanted = list(dict.fromkeys(a for a in artifacts if a)) + if not wanted: + return {} + + def read() -> dict[str, dict[str, Any]]: + out = {} + for artifact in wanted: + brief = brief_of(artifact) + if brief is not None: + out[artifact] = brief + return out + + return await asyncio.to_thread(read) diff --git a/backend/app/scheduler/webhook.py b/backend/app/scheduler/webhook.py index c7b9f2b1..e74e12b7 100644 --- a/backend/app/scheduler/webhook.py +++ b/backend/app/scheduler/webhook.py @@ -9,8 +9,9 @@ Webhook 地址本身就是凭证（企业微信的 key、钉钉的 access_token 一个过滤器把推送期间的那几行换成主机名。 通知里只有任务名、状态、出具档位、缺口条数、运行链接、封存哈希（界面上叫清单哈希）前 12 位和起止时间，失败时 -另写失败节点的名字和去哪里查看（不许诺能继续运行）。不放报告里的具体数字，也不放失败时的原始报错——报错里可能带着 -查询到的数据，Webhook 推去的群不一定有权看。 +另写失败节点的名字和去哪里查看（不许诺能继续运行）；运行结束后和同一任务上一次的对比（auto_compare）只写各类 +指标的个数和对比页链接。不放报告里的具体数字，也不放失败时的原始报错——报错里可能带着查询到的数据，Webhook +推去的群不一定有权看。 三个平台出错时都回 HTTP 200、错误码放在响应体里，所以送没送达要看响应体：按地址的主机认平台，不看选的格式—— 格式选错了（飞书的地址选了通用格式）正是最需要认出来的情形。群机器人的地址只认自己平台的格式，保存时就拦下 @@ -39,7 +40,7 @@ from app.core.events import EventType from app.db.base import SessionLocal, utcnow from app.db.models import Approval, Run, RunEvent, Schedule, ScheduleFire, WebhookDelivery from app.engine.labels import ISSUANCE_LABEL -from app.scheduler import service +from app.scheduler import auto_compare, service logger = logging.getLogger(__name__) @@ -227,6 +228,7 @@ class Notice: suspended：运行停在断点上、没有待审批（通常是服务重启挂起的），按 failed 事件推。error_node：失败节点的名字。 next_fire：任务启用着、往后还会触发（失败通知据此补「下一次触发会照常发起」）。 + compare：和同一定时任务上一次的对比摘要 {base_run_id, summary}（只有个数），compare_url：对比页的链接。 fmt、signed 只给测试消息用：说明里只说这条消息证明得了的事。""" event: str @@ -249,6 +251,8 @@ class Notice: suspended: bool = False error_node: str | None = None next_fire: bool = False + compare: dict[str, Any] | None = None + compare_url: str | None = None fmt: str | None = None signed: bool = False @@ -287,6 +291,9 @@ def message(notice: Notice) -> tuple[str, list[tuple[str, str]], str | None]: if notice.tier: gaps = f"（缺口 {notice.gaps} 条）" if notice.gaps is not None else "" 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 {}))) if notice.pending_approvals: lines.append(("待审批", f"{notice.pending_approvals} 项，请到 AgentLab 处理；运行结束之前，" "这个定时任务之后的触发都会跳过")) @@ -338,31 +345,34 @@ def _test_text(notice: Notice) -> str: return "这是一条测试消息。收到即表示 Webhook 地址配置正确" +def _links(notice: Notice) -> list[tuple[str, str]]: + """消息末尾的链接：运行，有对比时再加对比页。没设对外访问地址时都没有。""" + return [(label, url) for label, url in (("查看运行", notice.run_url), ("查看对比", notice.compare_url)) if url] + + def _text(notice: Notice) -> str: - title, lines, link = message(notice) + title, lines, _ = message(notice) out = [f"【AgentLab】{title}", *(f"{k}：{v}" for k, v in lines)] - if link: - out.append(f"查看运行：{link}") + out += [f"{label}：{url}" for label, url in _links(notice)] return "\n".join(out) def _wecom_markdown(notice: Notice) -> str: - title, lines, link = message(notice) + title, lines, _ = message(notice) color = "info" if notice.event in ("finished", "test") else "warning" out = [f"**【AgentLab】{title}**"] for key, value in lines: shown = f'{value}' if key == "状态" else value out.append(f"> {key}：{shown}") - if link: - out.append(f"[查看运行]({link})") + out += [f"[{label}]({url})" for label, url in _links(notice)] return "\n".join(out) def _dingtalk_markdown(notice: Notice) -> tuple[str, str]: - title, lines, link = message(notice) + title, lines, _ = message(notice) out = [f"#### 【AgentLab】{title}", "", *(f"- {k}：{v}" for k, v in lines)] - if link: - out += ["", f"[查看运行]({link})"] + for label, url in _links(notice): + out += ["", f"[{label}]({url})"] return title, "\n".join(out) @@ -396,6 +406,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} + if notice.compare else None), "sent_at": _iso(now), } @@ -576,15 +590,29 @@ class _Job: notice: Notice +@dataclass +class _Due: + """认领到的一次状态变化接下来要做的事：运行进入终态时和上一次比一次（compare_run），订阅了这个事件时推一条 + （job）。两样都在锁外做：对比要读快照、重算，推送要等接收端，都不能挡着发起。""" + + fire_id: str + compare_run: str | None + job: _Job | None + + async def notify_due(now: datetime | None = None) -> int: """查已发起的触发，运行状态有变化的按订阅推送一次。返回这一拍推送了几条（含失败的）。 处理过的状态记在 notified_status 上，用条件 UPDATE 认领（WHERE notified_status = 旧值）：两拍撞在一起也只 推一次。没订阅这个事件、任务没配 Webhook 的也记下已处理——之后再配上 Webhook，不会把旧的补推一遍。 停在断点上的运行每一拍都要重新看（见 _state_of）：它的已处理状态带着第几段执行，和运行状态本身对不上。 + + 运行进入终态的那一次，推送之前先和同一定时任务的上一次比一次（auto_compare），结束、失败的通知带上计数摘要和 + 对比页链接。没配 Webhook 的也比：触发历史里要用。比不了、出错不影响推送。 """ now = now or utcnow() - jobs: list[_Job] = [] + dues: list[_Due] = [] + base_url: str | None = None cutoff = now - timedelta(days=SCAN_DAYS) # 扫描窗口按触发记录的创建时间算，两种例外：运行记录窗口内有过变化（运行状态一变 updated_at 就跟着变），以及 # 停在审批上、有待审批的——以前只看触发记录，创建 30 天以后才停到审批上的运行收不到「等待审批」 @@ -603,7 +631,6 @@ async def notify_due(now: datetime | None = None) -> int: )) .order_by(ScheduleFire.created_at) )).all() - base_url: str | None = None if rows: from app.api.settings import notify_settings @@ -613,16 +640,30 @@ async def notify_due(now: datetime | None = None) -> int: if state is None or state == fire.notified_status \ or not await _mark(fire.id, fire.notified_status, state): 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 - if not (event and url and schedule.webhook_format in FORMATS - and event in (schedule.webhook_events or [])): - continue - notice = await _notice(session, schedule, fire, run, event, base_url, pending) - jobs.append(_Job(schedule.webhook_format, url, decrypt(schedule.webhook_secret), notice)) - if jobs: - await asyncio.gather(*(_deliver(job) for job in jobs)) - return len(jobs) + 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) + job = _Job(schedule.webhook_format, url, decrypt(schedule.webhook_secret), notice) + if compare_run or job: + dues.append(_Due(fire.id, 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) + + +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) async def _state_of(session: Any, fire: ScheduleFire, run: Run | None, diff --git a/backend/tests/test_scheduler_compare.py b/backend/tests/test_scheduler_compare.py new file mode 100644 index 00000000..480bfb53 --- /dev/null +++ b/backend/tests/test_scheduler_compare.py @@ -0,0 +1,395 @@ +"""定时运行结束后的自动对比（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=…）；只给个数，报告、指标、查询里的具体数字一个不出现 + +两次周报运行用 tests/fixtures/compare_runs.py 的虚构运行（区域、金额、日期都是编的），按 runner 的算法封存，不真的 +执行；Webhook 用 httpx.MockTransport 接住，不连外网。 +""" +from __future__ import annotations + +import json +import logging +from datetime import datetime, timedelta, timezone +from urllib.parse import parse_qs, urlsplit + +import httpx +import pytest +from httpx import ASGITransport, AsyncClient +from sqlalchemy import delete, select + +from app.core.artifact_store import canonical_json, load, manifest_hash, put_json +from app.core.crypto import encrypt +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 + +UTC = timezone.utc +T0 = datetime(2026, 9, 7, 1, 0, tzinfo=UTC) +WEEK = timedelta(days=7) +GENERIC = "https://hooks.example.com/agentlab/inbox-5a7c" +FEISHU = "https://open.feishu.cn/open-apis/bot/v2/hook/1a2b3c4d-aaaa-bbbb-cccc-0d9e8f7a6b5c" +BASE_URL = "https://agentlab.example.com" +#: 两次虚构周报（BASE_ROWS → NEXT_ROWS）之间两个指标都只是数据变了 +SUMMARY = {"unchanged": 0, "data_only": 2, "caliber_and_data": 0, "sql_changed": 0, "params_only": 0, "added": 0, + "removed": 0, "incomparable": 0} +LINE = "与上一次相比：只数据变 2 个、口径与数据都变 0 个、SQL 结构变 0 个" +#: 两次运行查到、算出的数（金额、退款、合计、变化量），通知里一个都不能有 +NUMBERS = ("1200.5", "1,200.5", "1500.5", "1,500.5", "2600.5", "2,600.5", "2800.5", "2,800.5", "300.0", "800.0", + "700.0", "600.0") + + +class Recorder: + def __init__(self) -> None: + self.requests: list[httpx.Request] = [] + + def __call__(self, request: httpx.Request) -> httpx.Response: + self.requests.append(request) + return httpx.Response(200, json={"code": 0, "errcode": 0}) + + def bodies(self) -> list[dict]: + return [json.loads(r.content) for r in self.requests] + + +@pytest.fixture +def recorder(monkeypatch): + rec = Recorder() + monkeypatch.setattr(webhook, "_transport", httpx.MockTransport(rec)) + monkeypatch.setattr(webhook, "RETRY_DELAYS", (0, 0, 0)) + return rec + + +@pytest.fixture(autouse=True) +async def clean(): + async with SessionLocal() as session: + await session.execute(delete(Schedule)) + await session.execute(delete(Setting).where(Setting.key == "notify")) + await session.commit() + yield + + +@pytest.fixture +async def client(): + async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as c: + yield c + + +async def public_address() -> None: + async with SessionLocal() as session: + session.add(Setting(key="notify", value={"public_base_url": BASE_URL})) + await session.commit() + + +async def new_workflow(name: str = "区域周报") -> str: + async with SessionLocal() as session: + workflow = Workflow(name=name) + session.add(workflow) + await session.commit() + return workflow.id + + +async def new_schedule(workflow_id: str, *, fmt: str = "generic", url: str | None = GENERIC, + name: str = "区域周报 · 每周") -> str: + async with SessionLocal() as session: + row = Schedule(name=name, workflow_id=workflow_id, cron="0 9 * * 1", timezone="Asia/Shanghai", + enabled=True, next_run_at=datetime.now(UTC) + timedelta(days=3), + webhook_format=fmt if url else None, webhook_url=encrypt(url) if url else None, + webhook_events=["finished", "failed", "awaiting_approval"]) + session.add(row) + await session.commit() + return row.id + + +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）。""" + for content in fake.store.values(): + await put_json(content, kind="compare_fixture") + async with SessionLocal() as session: + run = Run(workflow_id=workflow_id, workflow_name="区域周报", status=status, graph=fake.graph, + input=dict(fake.inputs), run_class="formal", version=fake.version, schedule_id=schedule_id, + started_at=finished - timedelta(minutes=2), finished_at=finished, agent_limits={}, + started_by="定时任务：区域周报 · 每周" if schedule_id else "周报编辑") + session.add(run) + await session.flush() + 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.last_seq = len(rows) + if seal: + run.manifest_hash, run.manifest_seq = manifest_hash(rows), rows[-1][0] + await session.commit() + return run.id + + +async def seal(run_id: str) -> None: + async with SessionLocal() as session: + run = await session.get(Run, run_id) + rows = [tuple(r) for r in await session.execute( + select(RunEvent.seq, RunEvent.type, RunEvent.node_id, RunEvent.data) + .where(RunEvent.run_id == run_id).order_by(RunEvent.seq))] + run.manifest_hash, run.manifest_seq = manifest_hash(rows), rows[-1][0] + await session.commit() + + +async def fire_for(schedule_id: str, run_id: str, at: datetime, *, notified: str | None = None) -> str: + async with SessionLocal() as session: + fire = ScheduleFire(schedule_id=schedule_id, fire_at=at, kind="scheduled", status="started", run_id=run_id, + notified_status=notified) + session.add(fire) + await session.commit() + return fire.id + + +async def fire_row(fire_id: str) -> ScheduleFire: + async with SessionLocal() as session: + return await session.get(ScheduleFire, fire_id) + + +async def set_run(run_id: str, **values) -> None: + async with SessionLocal() as session: + await session.execute(Run.__table__.update().where(Run.id == run_id).values(**values)) + await session.commit() + + +async def weekly_pair(*, target_status: str = "succeeded", target_seal: bool = True, + target_finished: datetime | None = None, fmt: str = "generic", url: str | None = GENERIC): + """同一个定时任务的两次周报：上一次（已推送过）和这一次（还没处理），这次的数变了。返回 (任务, 上一次, 这一次, + 这一次的触发)。""" + 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(week("r2", NEXT_ROWS), workflow, finished=target_finished or T0 + WEEK, + schedule_id=sid, status=target_status, seal=target_seal) + fire_id = await fire_for(sid, target_id, T0 + WEEK - timedelta(minutes=2)) + return sid, base_id, target_id, fire_id + + +def no_numbers(raw: str) -> list[str]: + return [n for n in NUMBERS if n in raw] + + +# -------------------------------------------------------------------------- +# 什么时候比、跟谁比、存在哪 +# -------------------------------------------------------------------------- + + +async def test_finished_run_is_compared_with_the_previous_run_of_its_schedule(recorder, client): + await public_address() + workflow = await new_workflow() + sid = await new_schedule(workflow) + other = await new_schedule(workflow, url=None, name="区域周报 · 补跑") + base_id = await persist(week("r1"), workflow, finished=T0, schedule_id=sid) + base_fire = await fire_for(sid, base_id, T0 - timedelta(minutes=2), notified="succeeded") + # 同一工作流、比上一次更近的两次：手动发起的、别的定时任务发起的。都不是「同一定时任务的上一次」 + await persist(week("m1"), workflow, finished=T0 + timedelta(days=3)) + await persist(week("o1"), workflow, finished=T0 + timedelta(days=5), schedule_id=other) + target_id = await persist(week("r2", NEXT_ROWS), workflow, finished=T0 + WEEK, schedule_id=sid) + fire_id = await fire_for(sid, target_id, T0 + WEEK - timedelta(minutes=2)) + + assert await webhook.notify_due() == 1 + + artifact = (await fire_row(fire_id)).compare_artifact + assert artifact + 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 + async with SessionLocal() as session: + rows = list((await session.execute(select(Artifact).where(Artifact.id == artifact))).scalars()) + # 对比结果不是运行产出的：不挂在运行名下，不进这次运行的工件列表 + assert [(a.kind, a.run_id) for a in rows] == [("run_compare", "")] + 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[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 + + [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} + query = parse_qs(urlsplit(body["compare"]["url"]).query) + assert query == {"target": [target_id], "base": [base_id]} + assert no_numbers(recorder.requests[0].content.decode()) == [] + # 下一拍：同一状态不再比、不再推 + 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): + """失败的运行也封存：和上一次比一次，失败通知同样带计数摘要""" + 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 == [] + 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 + + +async def test_first_run_of_a_schedule_has_no_compare(recorder, client): + """这个任务的第一次：同一工作流有更早的手动运行，对比接口会退回「同一工作流」那一档，那不是「上一次」，不比""" + await public_address() + workflow = await new_workflow() + sid = await new_schedule(workflow) + await persist(week("m1"), workflow, finished=T0) + target_id = await persist(week("r2", NEXT_ROWS), workflow, finished=T0 + WEEK, schedule_id=sid) + fire_id = await fire_for(sid, target_id, T0 + WEEK - timedelta(minutes=2)) + assert await webhook.notify_due() == 1 + assert (await fire_row(fire_id)).compare_artifact is None + [item] = (await client.get(f"/api/schedules/{sid}/fires")).json()["items"] + assert item["compare"] is None + [body] = recorder.bodies() + assert body["compare"] is None and "与上一次相比" not in body["text"] and "查看对比" not in body["text"] + assert f"查看运行：{BASE_URL}/runs/{target_id}" in body["text"] + + +async def test_waits_for_the_seal_before_comparing(recorder): + """runner 先写终态、再封存：两步之间被这一拍撞上就不比、不推，封存好了下一拍再比""" + _, base_id, target_id, fire_id = await weekly_pair(target_seal=False, target_finished=datetime.now(UTC)) + assert await webhook.notify_due() == 0 + 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 LINE in recorder.bodies()[0]["text"] + + +async def test_no_public_address_means_counts_without_a_link(recorder): + _, _, _, fire_id = await weekly_pair() + assert await webhook.notify_due() == 1 + [body] = recorder.bodies() + assert LINE in body["text"] and "查看对比" not in body["text"] and body["compare"]["url"] is None + assert (await fire_row(fire_id)).compare_artifact + + +# -------------------------------------------------------------------------- +# 比不了、出错 +# -------------------------------------------------------------------------- + + +async def break_seal(run_id: str) -> None: + async with SessionLocal() as session: + event = (await session.execute(select(RunEvent).where( + RunEvent.run_id == run_id, RunEvent.type == "tool.end"))).scalars().first() + event.data = {**event.data, "duration_ms": 999} + await session.commit() + + +async def explode(*args, **kwargs): + raise RuntimeError("快照读取失败：磁盘错误") + + +@pytest.mark.parametrize("failure", ["broken_seal", "unexpected"]) +async def test_compare_failure_only_logs_and_the_notice_still_goes_out(recorder, caplog, monkeypatch, failure): + """基准的封存核对不一致（对比接口回 409）、对比中途抛了意外的异常：只记日志，通知照常推送、不带对比""" + await public_address() + _, base_id, target_id, fire_id = await weekly_pair() + if failure == "broken_seal": + await break_seal(base_id) + else: + from app.api import compare as compare_api + + monkeypatch.setattr(compare_api, "compare_runs", explode) + with caplog.at_level(logging.INFO, logger="app.scheduler"): + assert await webhook.notify_due() == 1 + assert (await fire_row(fire_id)).compare_artifact is None + [body] = recorder.bodies() + assert (body["event"], body["run_id"], body["compare"]) == ("finished", target_id, None) + assert "与上一次相比" not in body["text"] and f"{BASE_URL}/runs/{target_id}" in body["text"] + assert "自动对比" in caplog.text and target_id in caplog.text + if failure == "broken_seal": + 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) + await webhook.notify_due() + assert (await fire_row(fire_id)).compare_artifact + 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} + + +async def test_a_missing_artifact_reads_as_no_compare(client): + """工件文件不在了、被改过：接口给 null，不报错""" + sid, _, _, fire_id = await weekly_pair(url=None) + async with SessionLocal() as session: + await session.execute(ScheduleFire.__table__.update().where(ScheduleFire.id == fire_id) + .values(compare_artifact="0" * 64)) + await session.commit() + resp = await client.get(f"/api/schedules/{sid}/fires") + assert resp.status_code == 200 + assert {f["id"]: f for f in resp.json()["items"]}[fire_id]["compare"] is None + + +# -------------------------------------------------------------------------- +# 计数摘要的写法 +# -------------------------------------------------------------------------- + + +def counts(**over) -> dict[str, int]: + return {**{k: 0 for k in SUMMARY}, **over} + + +def test_summary_text_always_names_the_three_kinds_of_change(): + assert auto_compare.summary_text(counts(data_only=3, caliber_and_data=1)) == \ + "只数据变 3 个、口径与数据都变 1 个、SQL 结构变 0 个" + # 其余各类有才写，次序同对比页 + assert auto_compare.summary_text(counts(sql_changed=1, params_only=4, added=1, removed=2, incomparable=1, + unchanged=5)) == \ + "只数据变 0 个、口径与数据都变 0 个、SQL 结构变 1 个、只参数变 4 个、新增 1 个、移除 2 个、无法比较 1 个、未变 5 个" + # 两边都没有指标：不写一串 0 个，免得读成「什么都没变」 + assert auto_compare.summary_text(counts()) == "两次运行都没有指标" + + +@pytest.mark.parametrize("fmt, url, link_text", [ + ("generic", GENERIC, "查看对比：{link}"), + ("feishu", FEISHU, "查看对比：{link}"), + ("wecom", "https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=0a1b2c3d-1111-2222-3333-444455556666", + "[查看对比]({link})"), + ("dingtalk", "https://oapi.dingtalk.com/robot/send?access_token=0123456789abcdef0123456789abcdef", + "[查看对比]({link})"), +]) +def test_every_format_carries_the_counts_and_the_compare_link(fmt, url, link_text): + link = f"{BASE_URL}/runs/compare?target=r2&base=r1" + notice = webhook.Notice(event="finished", schedule_id="s1", schedule_name="区域周报", timezone="Asia/Shanghai", + run_id="r2", status="succeeded", run_url=f"{BASE_URL}/runs/r2", + compare={"base_run_id": "r1", "summary": counts(data_only=3, caliber_and_data=1)}, + compare_url=link) + _, _, body = webhook.build_request(fmt, url, None, notice) + # 请求体是 ensure_ascii=False 的 JSON：中文、链接原样出现在里面 + text = body.decode() + assert "与上一次相比：只数据变 3 个、口径与数据都变 1 个、SQL 结构变 0 个" in text + assert link_text.format(link=link) in text + # 运行链接照旧在前 + assert text.index("/runs/r2") < text.index("/runs/compare") diff --git a/backend/tests/test_schedules_api.py b/backend/tests/test_schedules_api.py index fe4eb26d..cec1eeca 100644 --- a/backend/tests/test_schedules_api.py +++ b/backend/tests/test_schedules_api.py @@ -40,7 +40,7 @@ SECRET = "SECdemo-api-not-real-7" SCHEDULE_KEYS = {"id", "name", "workflow_id", "workflow_name", "cron", "timezone", "run_class", "input_template", "enabled", "overlap", "misfire_grace_s", "webhook", "next_run_at", "last_fire", "description_text", "created_by", "created_at", "updated_at", "version"} -FIRE_KEYS = {"id", "fire_at", "kind", "status", "run_id", "run_status", "reason", "notified", "created_at"} +FIRE_KEYS = {"id", "fire_at", "kind", "status", "run_id", "run_status", "reason", "notified", "compare", "created_at"} @pytest.fixture(autouse=True)