定时运行结束后自动和同一定时任务的上一次对比：结果存成工件记在触发记录上，触发历史和通知只给计数摘要与对比页链接

- 新增 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

@@ -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]
 

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、验收副本）不管从哪条路径起服务，

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)

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'<font color="{color}">{value}</font>' 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,

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")

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)