diff --git a/backend/app/scheduler/webhook.py b/backend/app/scheduler/webhook.py index 85907b66..c7b9f2b1 100644 --- a/backend/app/scheduler/webhook.py +++ b/backend/app/scheduler/webhook.py @@ -9,8 +9,8 @@ Webhook 地址本身就是凭证（企业微信的 key、钉钉的 access_token 一个过滤器把推送期间的那几行换成主机名。 通知里只有任务名、状态、出具档位、缺口条数、运行链接、封存哈希（界面上叫清单哈希）前 12 位和起止时间，失败时 -另写失败节点的名字。不放报告里的具体数字，也不放失败时的原始报错——报错里可能带着查询到的数据，Webhook 推去的群 -不一定有权看。 +另写失败节点的名字和去哪里查看（不许诺能继续运行）。不放报告里的具体数字，也不放失败时的原始报错——报错里可能带着 +查询到的数据，Webhook 推去的群不一定有权看。 三个平台出错时都回 HTTP 200、错误码放在响应体里，所以送没送达要看响应体：按地址的主机认平台，不看选的格式—— 格式选错了（飞书的地址选了通用格式）正是最需要认出来的情形。群机器人的地址只认自己平台的格式，保存时就拦下 @@ -52,7 +52,8 @@ EVENT_LABEL = {"finished": "运行结束", "failed": "运行失败", "awaiting_a TIMEOUT_S = 10.0 #: 失败后再试 3 次，每次之前等这么久（秒）。4xx（429 除外）不重试：地址或参数错了，重试也一样 RETRY_DELAYS: tuple[float, ...] = (1.0, 2.0, 4.0) -#: 只看这么多天内的触发：更早的运行状态不会再变，不用每一拍都扫 +#: 只看这么多天内的触发：更早结束的运行状态不会再变，不用每一拍都扫。例外见 notify_due：停在审批上、有待审批的不受 +#: 这个限制，运行记录这些天内有过变化的也照扫——审批可以等上一个多月，批完接着执行、停到下一个审批点或跑完都要推 SCAN_DAYS = 30 #: 运行结束后等封存（清单哈希）最多等多久。runner 先写终态再封存，两步之间被这一拍撞上就等下一拍 SEAL_WAIT_S = 60 @@ -65,6 +66,14 @@ _STATUS_LABEL = {"succeeded": "已完成", "failed": "失败", "cancelled": "已 #: 挂起的运行要人去处理：不处理，这个任务之后的触发都按「上一次还没结束」跳过 SUSPENDED_REASON = "运行停在断点上、没有待审批，通常是服务重启所致" SUSPENDED_ACTION = "请在 AgentLab 的「记录」中继续运行或放弃这次运行；它结束之前，这个定时任务之后的触发都会跳过" +#: 失败的运行怎么处理：不许诺「修正后可继续运行」。没开始执行的没有断点；开始执行过的也可能没有（执行第一个节点前就 +#: 失败）或已不兼容（系统升级），能不能继续以记录页的提示为准 +FAILED_ACTION = "请在 AgentLab 的运行记录中查看原因" +#: 从没开始执行就失败的（服务重启时还在排队，被改成失败、没有开始时间）：没有可继续的进度。不照抄运行记录上的报错—— +#: 通知一律不放原始报错，这一句是按「没有开始时间」认出来的，自己写 +NOT_STARTED_REASON = "运行在开始执行前就已中止（通常是服务重启时仍在排队），没有可继续的进度" +#: 失败的运行不挡之后的触发（挂起的才挡）：任务还在排期上时补这一句，免得人以为这个定时任务就此停了 +NEXT_AS_USUAL = "下一次触发会照常发起" # -------------------------------------------------------------------------- @@ -217,6 +226,7 @@ class Notice: failed_to_start。没有的项为 None。 suspended：运行停在断点上、没有待审批（通常是服务重启挂起的），按 failed 事件推。error_node：失败节点的名字。 + next_fire：任务启用着、往后还会触发（失败通知据此补「下一次触发会照常发起」）。 fmt、signed 只给测试消息用：说明里只说这条消息证明得了的事。""" event: str @@ -238,9 +248,15 @@ class Notice: pending_approvals: int | None = None suspended: bool = False error_node: str | None = None + next_fire: bool = False fmt: str | None = None signed: bool = False + @property + def never_started(self) -> bool: + """失败的运行从没开始执行（没有开始时间）：服务重启时还在排队、被改成失败的就是这种，没有断点可续。""" + return self.status == "failed" and self.run_id is not None and self.started_at is None + def _local(moment: datetime | None, tz: str, *, seconds: bool = True) -> str | None: if moment is None: @@ -282,7 +298,9 @@ def message(notice: Notice) -> tuple[str, list[tuple[str, str]], str | None]: # 不写原始报错（可能带着查询到的数据），写失败节点的名字和去哪里处理 if notice.error_node: lines.append(("失败节点", f"「{notice.error_node}」")) - lines.append(("处理", "请在 AgentLab 的运行记录中查看报错，修正后可继续运行")) + if notice.never_started: + lines.append(("原因", NOT_STARTED_REASON)) + 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}）" # 立即运行不写「计划触发」：群里的人会以为报告按时出了 @@ -375,7 +393,8 @@ def _generic(notice: Notice, now: datetime) -> dict[str, Any]: "started_at": _iso(notice.started_at), "finished_at": _iso(notice.finished_at), "reason": (notice.reason if notice.status == "failed_to_start" - else f"{SUSPENDED_REASON}。{SUSPENDED_ACTION}" if notice.suspended else None), + else f"{SUSPENDED_REASON}。{SUSPENDED_ACTION}" if notice.suspended + else NOT_STARTED_REASON if notice.never_started else None), "error_node": notice.error_node, "sent_at": _iso(now), } @@ -566,13 +585,18 @@ async def notify_due(now: datetime | None = None) -> int: """ now = now or utcnow() jobs: list[_Job] = [] + cutoff = now - timedelta(days=SCAN_DAYS) + # 扫描窗口按触发记录的创建时间算，两种例外：运行记录窗口内有过变化（运行状态一变 updated_at 就跟着变），以及 + # 停在审批上、有待审批的——以前只看触发记录，创建 30 天以后才停到审批上的运行收不到「等待审批」 + waiting = select(Approval.id).where(Approval.run_id == Run.id, Approval.status == "pending").exists() # 和发起共用一把锁：发起时先记一条「未能发起」再发起运行、成功了才改成 started，扫描不能撞在这中间 async with service.launch_lock(): async with SessionLocal() as session: rows = (await session.execute( select(ScheduleFire, Run) .outerjoin(Run, Run.id == ScheduleFire.run_id) - .where(ScheduleFire.created_at >= now - timedelta(days=SCAN_DAYS), or_( + .where(or_(ScheduleFire.created_at >= cutoff, Run.updated_at >= cutoff, + and_(Run.status == "interrupted", waiting)), or_( and_(ScheduleFire.status == "failed_to_start", ScheduleFire.notified_status.is_(None)), and_(ScheduleFire.status == "started", Run.status.in_(_WATCHED), or_(ScheduleFire.notified_status.is_(None), ScheduleFire.notified_status != Run.status)), @@ -664,6 +688,7 @@ async def _notice(session: Any, schedule: Schedule, fire: ScheduleFire, run: Run pending_approvals=pending or None, suspended=run is not None and run.status == "interrupted" and not pending, error_node=_failed_node(run) if run is not None and run.status == "failed" else None, + next_fire=bool(schedule.enabled and schedule.next_run_at is not None), ) diff --git a/backend/tests/test_scheduler_loop.py b/backend/tests/test_scheduler_loop.py index 94dab830..91ee6d90 100644 --- a/backend/tests/test_scheduler_loop.py +++ b/backend/tests/test_scheduler_loop.py @@ -369,9 +369,20 @@ async def test_start_does_nothing_when_switched_off(): await scheduler.stop() -async def test_loop_ticks_until_stopped(): - # 下一次放在整分上（和 cron 对齐）：放在「现在 - 1 秒」的话，这一秒里恰好跨过整分时会多记一条错过的跳过 +async def fire_count(schedule_id: str) -> int: + async with SessionLocal() as session: + return await session.scalar(select(func.count()).select_from(ScheduleFire) + .where(ScheduleFire.schedule_id == schedule_id)) + + +@pytest.mark.parametrize("crossed", [False, True], ids=["on_the_minute", "minute_crossed"]) +async def test_loop_ticks_until_stopped(crossed): + """循环按拍执行、stop 之后不再有拍。只看「发起了」和停下前后的条数差，不看总条数：建任务和第一拍之间恰好跨过 + 整分时，第一拍会先记一条错过的跳过、再发起最近这一次——以前按总条数断言，跨整分时偶发失败。crossed 把下一次 + 放在上一个整分，必然走到这一种""" due = cron.last_due("* * * * *", "UTC", datetime.now(UTC)) + if crossed: + due -= timedelta(minutes=1) sid = await schedule(await workflow(), next_run_at=due, cron="* * * * *") on = Settings(_env_file=None, scheduler=True) assert await scheduler.start(interval=0.05, config=on) is True @@ -382,18 +393,20 @@ async def test_loop_ticks_until_stopped(): await asyncio.sleep(0.02) await scheduler.stop() assert scheduler.running() is False - # 只认发起的那一条：建任务和第一拍之间跨过整分时，前面还会有一条错过的跳过，那不是这里要测的 - [fire] = [f for f in await fires(sid) if f.status == "started"] - await settle(fire.run_id) - # 停下之后不再有拍：把下一次拨回过去也不会再发起 + started = [f for f in await fires(sid) if f.status == "started"] + assert started + if crossed: + assert [f.status for f in await fires(sid)][0] == "skipped_misfire" + for fire in started: + await settle(fire.run_id) + # 停下之后不再有拍：把下一次拨回过去，过一会儿条数一条不多 + before = await fire_count(sid) async with SessionLocal() as session: await session.execute(Schedule.__table__.update().where(Schedule.id == sid) .values(next_run_at=datetime.now(UTC) - timedelta(seconds=1))) await session.commit() await asyncio.sleep(0.2) - async with SessionLocal() as session: - assert await session.scalar(select(func.count()).select_from(ScheduleFire) - .where(ScheduleFire.schedule_id == sid)) == 1 + assert await fire_count(sid) == before async def test_start_twice_keeps_one_loop(): diff --git a/backend/tests/test_scheduler_notice_rules.py b/backend/tests/test_scheduler_notice_rules.py new file mode 100644 index 00000000..dfc11af7 --- /dev/null +++ b/backend/tests/test_scheduler_notice_rules.py @@ -0,0 +1,188 @@ +"""定时任务通知（app/scheduler/webhook.py）的两条规则： + +- 失败通知的「处理」行不许诺能继续运行：服务重启时还在排队、从没开始执行的运行（被改成失败、没有开始时间）没有 + 断点，「修正后可继续运行」是错的。按有没有开始执行分开写——没开始的另写一行原因（没有可继续的进度），处理一律是 + 中性的「请在运行记录中查看原因」；任务还在排期上时补一句「下一次触发会照常发起」：失败的运行不挡之后的触发， + 和挂起的那种（之后的触发都会跳过）说清区别 +- 扫描窗口（SCAN_DAYS）之外的触发：停在审批、有待审批的运行不受窗口限制——创建 30 天以后才停到审批上的照样推送 + 「等待审批」；运行记录在窗口内有变化的（等了一个多月的审批批完、跑完）照样推送结束。早就结束、记录多日没有变化 + 的旧触发仍不扫 + +用 httpx.MockTransport 接住请求，不连外网。任务名、地址、数字都是虚构的。 +""" +from __future__ import annotations + +import json +from datetime import datetime, timedelta, timezone + +import httpx +import pytest +from sqlalchemy import delete + +from app.core.crypto import encrypt +from app.db.base import SessionLocal +from app.db.models import Approval, Run, RunEvent, Schedule, ScheduleFire, Setting +from app.scheduler import webhook + +UTC = timezone.utc +GENERIC = "https://hooks.example.com/agentlab/inbox-3e9d" +NOT_STARTED = "运行在开始执行前就已中止（通常是服务重启时仍在排队），没有可继续的进度" + + +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={}) + + 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 + + +def notice(**over) -> webhook.Notice: + base = dict(event="failed", schedule_id="s1", schedule_name="仓库日报", timezone="Asia/Shanghai", + fire_id="f1", fire_at=datetime(2026, 10, 5, 1, 0, tzinfo=UTC), fire_kind="scheduled", + run_id="r9876543210fedcba", status="failed", run_url=None, manifest_hash=None, + started_at=datetime(2026, 10, 5, 1, 0, 2, tzinfo=UTC), finished_at=None) + base.update(over) + return webhook.Notice(**base) + + +async def make(*, run_status: str, started: bool = True, fire_age: timedelta = timedelta(0), + run_age: timedelta = timedelta(0), enabled: bool = True, sealed: bool = True) -> tuple[str, str, str]: + """一个配了通用 Webhook 的任务、一次触发和它发起的运行（不真的执行，状态直接写库）。fire_age、run_age 是触发 + 记录、运行记录最后一次更新离现在多久；started=False 是从没开始执行的（服务重启时还在排队）。""" + now = datetime.now(UTC) + async with SessionLocal() as session: + sched = Schedule(name="仓库日报", workflow_id="wf-demo", cron="0 9 * * *", timezone="Asia/Shanghai", + enabled=enabled, next_run_at=now + timedelta(hours=3) if enabled else None, + webhook_format="generic", webhook_url=encrypt(GENERIC), + webhook_events=["finished", "failed", "awaiting_approval"]) + session.add(sched) + await session.flush() + ended = run_status in ("succeeded", "failed", "cancelled") and started + touched = now - run_age + run = Run(workflow_id=None, workflow_name="仓库日报", status=run_status, schedule_id=sched.id, + run_class="formal", started_at=touched - timedelta(minutes=1) if started else None, + finished_at=touched if ended else None, manifest_hash=("d00d" * 16) if sealed and ended else None, + created_at=now - fire_age, updated_at=touched, + error="服务重启时这次运行仍在排队，没有可恢复的进度，请重新发起运行" if not started else None) + session.add(run) + await session.flush() + fire = ScheduleFire(schedule_id=sched.id, fire_at=(now - fire_age).replace(microsecond=0), kind="scheduled", + status="started", run_id=run.id, created_at=now - fire_age) + session.add(fire) + await session.commit() + return sched.id, fire.id, run.id + + +async def notified(fire_id: str) -> str | None: + async with SessionLocal() as session: + return (await session.get(ScheduleFire, fire_id)).notified_status + + +# -------------------------------------------------------------------------- +# 失败通知的「处理」行 +# -------------------------------------------------------------------------- + + +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 + + +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 "修正后可继续运行" 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 的运行记录中查看原因" + + +async def test_restart_failure_notice_end_to_end(recorder): + """服务重启时还在排队、被改成失败的运行：不会封存，等过封存时限后推一次失败，原因、处理按上面写；通用格式的 + reason 字段也写这句原因。任务停用时不补「下一次触发会照常发起」""" + _, on_fire, _ = await make(run_status="failed", started=False, sealed=False) + _, off_fire, _ = await make(run_status="failed", started=False, sealed=False, enabled=False) + later = datetime.now(UTC) + timedelta(seconds=webhook.SEAL_WAIT_S + 5) + assert await webhook.notify_due(later) == 2 + 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"] + for body in (on, off): + assert "继续运行" not in body["text"], body["text"] + # 运行记录上的原始报错不进通知 + assert "请重新发起运行" not in body["text"] + + +# -------------------------------------------------------------------------- +# 扫描窗口之外 +# -------------------------------------------------------------------------- + + +async def test_old_fire_stopped_at_an_approval_is_still_pushed(recorder): + """触发是 40 天前的，运行刚停到审批上（或者停下时服务没在推送）：有待审批的运行不受 30 天的窗口限制。 + 运行记录本身也很久没有变化，照样推「等待审批」""" + _, fire_id, run_id = await make(run_status="interrupted", fire_age=timedelta(days=40), run_age=timedelta(days=35)) + async with SessionLocal() as session: + session.add(RunEvent(run_id=run_id, seq=1, type="run.started", ts=1.0, data={})) + session.add(Approval(run_id=run_id, node_id="gate", title="审核日报", status="pending")) + await session.commit() + assert await webhook.notify_due() == 1 + [body] = recorder.bodies() + assert (body["event"], body["pending_approvals"]) == ("awaiting_approval", 1) + assert await notified(fire_id) == "interrupted:approval:1" + assert await webhook.notify_due() == 0 + + +async def test_old_fire_whose_run_just_finished_is_pushed(recorder): + """等了一个多月的审批批完、刚跑完：运行记录在窗口内有变化，推「运行结束」""" + _, fire_id, _ = await make(run_status="succeeded", fire_age=timedelta(days=45)) + assert await webhook.notify_due() == 1 + assert recorder.bodies()[0]["event"] == "finished" and await notified(fire_id) == "succeeded" + + +async def test_old_fire_whose_run_ended_long_ago_is_not_scanned(recorder): + """早就结束、多日没有变化的旧触发仍不扫：窗口本来就是为了不每一拍把全部历史翻一遍""" + _, fire_id, _ = await make(run_status="succeeded", fire_age=timedelta(days=45), run_age=timedelta(days=40)) + assert await webhook.notify_due() == 0 + assert recorder.requests == [] and await notified(fire_id) is None + + +async def test_old_suspended_run_without_approval_is_not_rescanned_forever(recorder): + """挂起（没有待审批）、多日没有变化的旧触发不在例外里：例外只给停在审批上的""" + _, fire_id, _ = await make(run_status="interrupted", fire_age=timedelta(days=40), run_age=timedelta(days=40)) + assert await webhook.notify_due() == 0 and await notified(fire_id) is None diff --git a/backend/tests/test_scheduler_webhook.py b/backend/tests/test_scheduler_webhook.py index 98c41404..ca32dca3 100644 --- a/backend/tests/test_scheduler_webhook.py +++ b/backend/tests/test_scheduler_webhook.py @@ -178,7 +178,8 @@ def test_failed_run_text_points_to_the_node_without_repeating_itself(): error_node="查询门店销售")) shown = dict(lines) assert title.endswith("运行失败") and shown["状态"] == "失败" - assert shown["失败节点"] == "「查询门店销售」" and "继续运行" in shown["处理"] + # 处理只说去哪里看，不许诺能继续运行（没有断点的运行续不了，见 test_scheduler_notice_rules.py） + assert shown["失败节点"] == "「查询门店销售」" and shown["处理"] == "请在 AgentLab 的运行记录中查看原因" # 不再写「原因：运行失败，详情请查看运行记录」这种同义反复 assert "原因" not in shown