diff --git a/backend/app/api/settings.py b/backend/app/api/settings.py index a8259092..566fae74 100644 --- a/backend/app/api/settings.py +++ b/backend/app/api/settings.py @@ -562,6 +562,66 @@ async def verify_model_spec(session: AsyncSession, node_config: dict[str, Any] | return ModelSpec(provider=provider, model=model, max_tokens=JUDGE_MAX_TOKENS, thinking="off") +# -------------------------------------------------------------------------- +# 通知：定时任务推送 Webhook 时拼运行链接用的对外访问地址 +# -------------------------------------------------------------------------- + +#: 设置里 notify 这一组。public_base_url 是别人打开 AgentLab 用的地址（https://agentlab.example.com），通知里的 +#: 运行链接 = 它 + /runs/{运行 id}。服务端不知道自己在反向代理后面叫什么，只能由人填；留空的通知里不附链接 +NOTIFY_GROUP = "notify" +NOTIFY_DEFAULTS: dict[str, Any] = {"public_base_url": None} +NOTIFY_SETTING_LABEL = {"public_base_url": "对外访问地址"} + + +def _base_url_problem(value: Any) -> str | None: + if value is None or (isinstance(value, str) and not value.strip()): + return None + label = NOTIFY_SETTING_LABEL["public_base_url"] + hint = (f"「{label}」需要是 http:// 或 https:// 开头的完整地址（例如 https://agentlab.example.com），" + f"留空表示通知里不附运行链接，当前为「{value}」") + if not isinstance(value, str): + return hint + from urllib.parse import urlsplit + + try: + parts = urlsplit(value.strip()) + except ValueError: + return hint + if parts.scheme not in ("http", "https") or not parts.hostname or " " in value.strip(): + return hint + if parts.query or parts.fragment: + return f"「{label}」不能带 ? 或 # 之后的部分，运行链接会接在地址后面，当前为「{value}」" + return None + + +def notify_problem(value: Any) -> str | None: + """设置页写进来的 notify 这一组有什么问题，没问题返回 None。""" + if not isinstance(value, dict): + return f"「通知」设置格式有误，需要是一个对象，当前为「{value}」" + for key, item in value.items(): + if key not in NOTIFY_DEFAULTS: + return (f"「通知」设置中不支持「{key}」，可设置的项：" + + "、".join(f"「{NOTIFY_SETTING_LABEL[k]}」" for k in NOTIFY_DEFAULTS)) + if problem := _base_url_problem(item): + return problem + return None + + +def notify_values(stored: Any) -> dict[str, Any]: + """库里存的（或刚写进来的）notify 一组 → 补齐默认、规范化：去掉首尾空白和末尾的 /，空串当没填；存坏了的按没填。""" + stored = stored if isinstance(stored, dict) else {} + url = stored.get("public_base_url") + if not isinstance(url, str) or not url.strip() or _base_url_problem(url): + return {"public_base_url": None} + return {"public_base_url": url.strip().rstrip("/")} + + +async def notify_settings(session: AsyncSession) -> dict[str, Any]: + """设置里的 notify 一组，没存过的补默认。定时任务推送 Webhook 时从这里读。""" + row = await session.get(Setting, NOTIFY_GROUP) + return notify_values(row.value if row else None) + + # -------------------------------------------------------------------------- # 用户设置 # -------------------------------------------------------------------------- @@ -592,6 +652,8 @@ DEFAULT_SETTINGS: dict[str, Any] = { JUDGE_GROUP: dict(JUDGE_DEFAULTS), # 核查（双盲复算、反方审稿、审批质询）：核查模型、三项的开关和上限、每日金额上限。和裁判一组一样由用户定 VERIFY_GROUP: verify_values(None), + # 通知：定时任务的 Webhook 里拼运行链接用的对外访问地址 + NOTIFY_GROUP: notify_values(None), } @@ -647,6 +709,7 @@ async def get_settings(session: AsyncSession = Depends(get_session)) -> dict[str # 裁判、核查设置按引擎实际生效的样子给：存坏了的项显示默认，存了 null 的上限显示不限 merged[JUDGE_GROUP] = judge_values(stored.get(JUDGE_GROUP)) merged[VERIFY_GROUP] = verify_values(stored.get(VERIFY_GROUP)) + merged[NOTIFY_GROUP] = notify_values(stored.get(NOTIFY_GROUP)) for key, value in stored.items(): merged.setdefault(key, value) return merged @@ -667,6 +730,11 @@ async def put_settings( raise HTTPException(422, problem) if VERIFY_GROUP in payload.values and (problem := verify_problem(payload.values[VERIFY_GROUP])): raise HTTPException(422, problem) + if NOTIFY_GROUP in payload.values: + if problem := notify_problem(payload.values[NOTIFY_GROUP]): + raise HTTPException(422, problem) + # 存规范化之后的：末尾带 / 的地址拼链接会多出一个斜杠 + payload.values[NOTIFY_GROUP] = notify_values(payload.values[NOTIFY_GROUP]) for key, value in payload.values.items(): if key in (TRUST_KEY, "limits", JUDGE_SPEND, VERIFY_SPEND): # 信任三档只走 PUT /api/tools/trust：设置页整组回写时带着一份旧的， diff --git a/backend/app/db/models.py b/backend/app/db/models.py index 3d7cc860..0a250142 100644 --- a/backend/app/db/models.py +++ b/backend/app/db/models.py @@ -749,3 +749,97 @@ class ConversationTurn(Base, TimestampMixin): meta: Mapped[dict[str, Any] | None] = mapped_column(default=None) conversation: Mapped[Conversation] = relationship(back_populates="turns") + + +# -------------------------------------------------------------------------- +# 定时任务：按 cron 到点发起运行，结束、失败、停在审批时推送 Webhook（app/scheduler） +# -------------------------------------------------------------------------- + + +class Schedule(Base, TimestampMixin): + """一个定时任务。webhook_url、webhook_secret 落库前经 app.core.crypto 加密，接口只返回主机和末 4 位。 + + 不给 workflow_id 设外键：工作流删掉之后任务留着，到点照常发起、照实记一次未能发起（「工作流不存在」）， + 订阅了失败通知的人会收到——悄悄跟着删掉，每周的报告就无声无息地没了。 + """ + + __tablename__ = "schedules" + + id: Mapped[str] = mapped_column(String(32), primary_key=True, default=new_id) + name: Mapped[str] = mapped_column(String(200), default="") + workflow_id: Mapped[str] = mapped_column(String(32), index=True) + # 保存时的工作流名。工作流还在时接口给它现在的名字，删掉之后才用这一份 + workflow_name: Mapped[str] = mapped_column(String(200), default="") + cron: Mapped[str] = mapped_column(String(200)) + timezone: Mapped[str] = mapped_column(String(64)) + # formal：每次都从当前发布版本发起正式运行；exploratory：用工作流现在的画布 + run_class: Mapped[str] = mapped_column(String(20), default="formal") + # {输入字段名: 模板字符串}，到点按触发时间渲染（scheduler/service.py） + input_template: Mapped[dict[str, Any]] = mapped_column(default=dict) + enabled: Mapped[bool] = mapped_column(default=True) + # 上一次运行还没结束时怎么办。目前只有 skip：记一次跳过 + overlap: Mapped[str] = mapped_column(String(20), default="skip") + # 错过的触发（服务停机等）在这么多秒内补一次，超出就记跳过 + misfire_grace_s: Mapped[int] = mapped_column(Integer, default=3600) + # generic | feishu | wecom | dingtalk。没配 Webhook 时 url 为空 + webhook_format: Mapped[str | None] = mapped_column(String(20), default=None) + webhook_url: Mapped[str | None] = mapped_column(Text, default=None) # 密文 + webhook_secret: Mapped[str | None] = mapped_column(Text, default=None) # 密文 + webhook_events: Mapped[list[Any]] = mapped_column(default=list) + # 下一次触发的时刻（UTC）。停用时为空；调度循环按它认领，认领时推进到再下一次 + next_run_at: Mapped[datetime | None] = mapped_column(UTCDateTime, default=None, index=True) + created_by: Mapped[str | None] = mapped_column(String(100), default=None) + # 乐观锁：PATCH 带上读到的版本，对不上回 409。调度循环推进 next_run_at 不改它 + version: Mapped[int] = mapped_column(Integer, default=1) + + +class ScheduleFire(Base): + """一次触发：到点、立即运行、或者错过的补跑 / 跳过。(schedule_id, fire_at, kind) 唯一：两拍撞在一起时， + 后插入的那条撞上约束就知道这一次已经有人处理了，不会发起第二次。""" + + __tablename__ = "schedule_fires" + __table_args__ = ( + UniqueConstraint("schedule_id", "fire_at", "kind"), + Index("ix_schedule_fires_schedule_created", "schedule_id", "created_at"), + ) + + id: Mapped[str] = mapped_column(String(32), primary_key=True, default=new_id) + schedule_id: Mapped[str] = mapped_column(ForeignKey("schedules.id", ondelete="CASCADE")) + # 计划触发的时刻（UTC）。立即运行是点下去的那一秒；输入模板的日期变量都按它算 + fire_at: Mapped[datetime] = mapped_column(UTCDateTime) + # scheduled | manual | misfire + kind: Mapped[str] = mapped_column(String(20), default="scheduled") + # started | skipped_overlap | skipped_misfire | failed_to_start + status: Mapped[str] = mapped_column(String(30), default="started") + run_id: Mapped[str | None] = mapped_column(String(32), default=None, index=True) + reason: Mapped[str | None] = mapped_column(Text, default=None) + # 渲染后的输入，就是交给运行的那一份 + inputs: Mapped[dict[str, Any]] = mapped_column(default=dict) + # 已经按哪个运行状态处理过通知（succeeded / failed / cancelled / interrupted / failed_to_start）。 + # 运行状态变了、和它不同才再看要不要推送，同一状态只处理一次 + notified_status: Mapped[str | None] = mapped_column(String(30), default=None) + # 和同一任务上一次运行的比较结果（工件 id），后续的运行对比工作包写 + compare_artifact: Mapped[str | None] = mapped_column(String(64), default=None) + created_at: Mapped[datetime] = mapped_column(UTCDateTime, default=utcnow) + + +class WebhookDelivery(Base): + """一次 Webhook 推送（含重试）的结果。error 只写主机名和平台返回的错误码，不含地址、key、密钥。 + 「测试 Webhook」也记一条：fire_id、run_id 为空，event 是 test。""" + + __tablename__ = "webhook_deliveries" + + id: Mapped[str] = mapped_column(String(32), primary_key=True, default=new_id) + schedule_id: Mapped[str] = mapped_column(ForeignKey("schedules.id", ondelete="CASCADE"), index=True) + fire_id: Mapped[str | None] = mapped_column( + ForeignKey("schedule_fires.id", ondelete="CASCADE"), default=None, index=True + ) + run_id: Mapped[str | None] = mapped_column(String(32), default=None) + # finished | failed | awaiting_approval | test + event: Mapped[str] = mapped_column(String(30)) + # ok | failed + status: Mapped[str] = mapped_column(String(20)) + http_status: Mapped[int | None] = mapped_column(Integer, default=None) + attempts: Mapped[int] = mapped_column(Integer, default=0) + error: Mapped[str | None] = mapped_column(Text, default=None) + created_at: Mapped[datetime] = mapped_column(UTCDateTime, default=utcnow) diff --git a/backend/app/scheduler/__init__.py b/backend/app/scheduler/__init__.py index e6da00b8..aa1ef8b6 100644 --- a/backend/app/scheduler/__init__.py +++ b/backend/app/scheduler/__init__.py @@ -1,12 +1,22 @@ -"""定时任务的后台调度：到点按定时任务的设置发起工作流运行。 +"""定时任务的后台调度：到点按定时任务的设置发起工作流运行，结束、失败、停在审批时推送 Webhook。 -目前 start / stop 都是空的，调度循环由定时任务的工作包补上。服务启动时（main.py 的 lifespan） -只在 scheduler_enabled() 时调 start，关闭时总是调 stop——stop 要能在没 start 过时安全返回。 +- cron.py：五字段 cron 的解析、下次触发、中文描述 +- service.py：输入模板、和工作流入口字段的对照、发起一次触发（调度循环和「立即运行」共用） +- loop.py：调度循环，条件 UPDATE 认领到点的任务 +- webhook.py：通知推送（通用 / 飞书 / 企业微信 / 钉钉） + +服务启动时（main.py 的 lifespan）只在 scheduler_enabled() 时调 start，关闭时总是调 stop——stop 要能在没 start +过时安全返回。start 自己也看一次开关：拷来的数据目录（检查脚本的 --stacks、验收副本）不管从哪条路径起服务， +都不会替原库把定时任务再执行一遍。 """ from __future__ import annotations +import logging + from app.core.config import Settings, settings +logger = logging.getLogger(__name__) + def scheduler_enabled(config: Settings | None = None) -> bool: """调度循环开不开：AGENTLAB_SCHEDULER，默认开，0 / false / off 关（见 core/config.py）。 @@ -15,9 +25,25 @@ def scheduler_enabled(config: Settings | None = None) -> bool: return bool((config or settings).scheduler) -async def start() -> None: - """起调度循环。调度循环在后续工作包里实现，现在什么都不做。""" +async def start(*, interval: float | None = None, config: Settings | None = None) -> bool: + """起调度循环，返回起没起。开关关着时什么都不做；已经在跑时不起第二个。interval 只给测试用。""" + if not scheduler_enabled(config): + logger.info("定时任务调度已关闭（AGENTLAB_SCHEDULER），不会自动发起运行") + return False + from app.scheduler.loop import TICK_SECONDS, runner + + await runner.start(interval if interval is not None else TICK_SECONDS) + return True async def stop() -> None: - """停调度循环，等手上的那一轮收尾。现在什么都不做；没 start 过也要能安全调用。""" + """停调度循环，等手上的那一轮收尾。没 start 过也能安全调用。""" + from app.scheduler.loop import runner + + await runner.stop() + + +def running() -> bool: + from app.scheduler.loop import runner + + return runner.running() diff --git a/backend/app/scheduler/loop.py b/backend/app/scheduler/loop.py new file mode 100644 index 00000000..2eb7df6a --- /dev/null +++ b/backend/app/scheduler/loop.py @@ -0,0 +1,174 @@ +"""调度循环：每一拍认领到点的定时任务、发起运行，再顺便查已发起的运行要不要推送通知。 + +认领靠条件 UPDATE：UPDATE schedules SET next_run_at = 下一次 WHERE id = … AND next_run_at = 旧值。两拍撞在 +一起（或者以后多开了一个进程），只有一拍改得动这一行，另一拍拿到 0 行就知道这一次已经有人处理了。触发记录 +(任务, fire_at, kind) 唯一是第二道保险。 + +错过的触发（服务停机、进程卡住）：只看最近应触发的那一次——在宽限内补一次（kind=misfire），超出就记一条 +skipped_misfire；更早的积压只记一条跳过，不补。下一次一律推进到现在之后，不会醒来就把积压的一口气全发起。 +""" +from __future__ import annotations + +import asyncio +import contextlib +import logging +from datetime import datetime + +from sqlalchemy import select, update + +from app.db.base import SessionLocal, utcnow +from app.db.models import Schedule, ScheduleFire +from app.scheduler import cron, service + +logger = logging.getLogger(__name__) + +#: 一拍的间隔（秒）。测试里 start(interval=…) 调小 +TICK_SECONDS = 15.0 +#: 晚了这么多秒以内都算准点（kind=scheduled），和补发宽限无关：拍与拍之间的十几秒不该算错过 +ON_TIME_S = 60 +#: stop 等手上那一拍收尾最多等多久，超过就取消 +STOP_GRACE_S = 10.0 + + +async def fire_due(now: datetime | None = None) -> list[ScheduleFire]: + """处理所有到点的任务（enabled 且 next_run_at <= now），返回这一拍记下的触发。一个任务出错不影响别的。""" + now = now or utcnow() + async with SessionLocal() as session: + due = (await session.execute( + select(Schedule.id, Schedule.next_run_at) + .where(Schedule.enabled.is_(True), Schedule.next_run_at.is_not(None), Schedule.next_run_at <= now) + .order_by(Schedule.next_run_at) + )).all() + out: list[ScheduleFire] = [] + for schedule_id, claimed_at in due: + try: + out.extend(await _due_one(schedule_id, claimed_at, now)) + except Exception: # noqa: BLE001 - 一个任务出错不能拖住这一拍的其他任务，也不能让循环停下 + logger.exception("定时任务 %s 这一拍处理失败", schedule_id) + return out + + +async def _due_one(schedule_id: str, claimed_at: datetime, now: datetime) -> list[ScheduleFire]: + async with SessionLocal() as session: + row = await session.get(Schedule, schedule_id) + if row is None or not row.enabled or row.next_run_at != claimed_at: + return [] + cron_expr, tz, grace = row.cron, row.timezone, max(int(row.misfire_grace_s or 0), 0) + try: + spec, zone = cron.parse(cron_expr), cron.zone(tz) + # 最近应触发的那一次。库里的 next_run_at 理应就是一个触发时刻，万一不是（手改过库）就按它算 + latest = max(cron.last_due(spec, zone, now) or claimed_at, claimed_at) + upcoming = cron.next_fire(spec, zone, now) + except (cron.CronError, cron.TimezoneError) as e: + # 存进库里之后才坏掉的表达式（手改过库、时区库升级删了时区）：停止排期，在触发历史里照实说 + if not await _claim(schedule_id, claimed_at, None): + return [] + logger.warning("定时任务 %s 的 cron 或时区无法解析，已停止排期：%s", schedule_id, e) + fire = await service._record(schedule_id, claimed_at, "scheduled", "failed_to_start", + f"cron 表达式或时区无法解析，已停止排期：{e}", {}) + return [fire] if fire else [] + if not await _claim(schedule_id, claimed_at, upcoming): + return [] + + out: list[ScheduleFire | None] = [] + lateness = (now - latest).total_seconds() + if lateness <= max(ON_TIME_S, grace): + if latest > claimed_at: + # 最近这一次照常（或补一次），更早的积压只记一条，不补 + last_missed = cron.prev_fire(spec, zone, latest) or claimed_at + reason = f"服务未运行，错过了 {_span(claimed_at, last_missed, tz)} 的触发；积压的触发不补，只发起最近一次" + out.append(await service.record_skip(schedule_id, claimed_at, reason)) + kind = "scheduled" if lateness <= ON_TIME_S else "misfire" + out.append(await service.fire_once(schedule_id, latest, kind)) + else: + reason = (f"服务未运行，错过了 {_span(claimed_at, latest, tz)} 的触发，" + f"超出补发宽限（{service.duration_text(grace)}），没有发起") + out.append(await service.record_skip(schedule_id, latest, reason)) + return [fire for fire in out if fire is not None] + + +async def _claim(schedule_id: str, claimed_at: datetime, upcoming: datetime | None) -> bool: + """把 next_run_at 从 claimed_at 推进到 upcoming。改动了一行才算认领到。 + + updated_at 照旧：它说的是任务最近一次被人修改，推进排期不算。version 也不动——乐观锁防的是两个人同时改设置， + 调度推进 next_run_at 不该让正在编辑的人保存时撞上 409。""" + async with SessionLocal() as session: + result = await session.execute( + update(Schedule) + .where(Schedule.id == schedule_id, Schedule.enabled.is_(True), Schedule.next_run_at == claimed_at) + .values(next_run_at=upcoming, updated_at=Schedule.updated_at) + .execution_options(synchronize_session=False) + ) + await session.commit() + return result.rowcount == 1 + + +def _span(first: datetime, last: datetime, tz: str) -> str: + a, b = service.local_text(first, tz), service.local_text(last, tz) + return a if a == b else f"{a} 至 {b}" + + +async def tick(now: datetime | None = None) -> None: + """完整的一拍：发起到点的，再推送该推送的通知。测试和手动调用用；循环里两件事分开跑（见 _Runner._loop）。""" + from app.scheduler import webhook + + await fire_due(now) + await webhook.notify_due(now) + + +class _Runner: + """调度循环的任务。照 sandbox/manager.py 回收闲置会话的写法：一个 asyncio 任务，出错记日志接着下一拍。""" + + def __init__(self) -> None: + self.task: asyncio.Task[None] | None = None + #: 通知推送另起任务：Webhook 超时加重试最长要几十秒，不能拖着下一拍的发起。同一时刻只跑一份 + self.notify: asyncio.Task[None] | None = None + self._stopping: asyncio.Event | None = None + + def running(self) -> bool: + return self.task is not None and not self.task.done() + + async def start(self, interval: float = TICK_SECONDS) -> None: + if self.running(): + return + self._stopping = asyncio.Event() + self.task = asyncio.create_task(self._loop(interval, self._stopping), name="agentlab-scheduler") + logger.info("定时任务调度已启动，每 %s 秒检查一次", interval) + + async def _loop(self, interval: float, stopping: asyncio.Event) -> None: + from app.scheduler import webhook + + while not stopping.is_set(): + try: + await fire_due() + if self.notify is None or self.notify.done(): + self.notify = asyncio.create_task(webhook.notify_due(), name="agentlab-scheduler-notify") + except asyncio.CancelledError: + raise + except Exception: # noqa: BLE001 - 这一拍出错不该让循环停下 + logger.exception("定时任务调度这一拍出错，下一拍照常进行") + with contextlib.suppress(TimeoutError): + await asyncio.wait_for(stopping.wait(), timeout=interval) + + async def stop(self) -> None: + """让循环收完手上这一拍再停；等不到（STOP_GRACE_S）就取消。没 start 过直接返回。""" + tasks = [t for t in (self.task, self.notify) if t is not None] + self.task = self.notify = None + if self._stopping is not None: + self._stopping.set() + for task in tasks: + if task.done(): + continue + try: + await asyncio.wait_for(asyncio.shield(task), timeout=STOP_GRACE_S) + except TimeoutError: + task.cancel() + with contextlib.suppress(asyncio.CancelledError, Exception): + await task + except Exception: # noqa: BLE001 - 停的时候不再追究这一拍的错 + pass + if tasks: + logger.info("定时任务调度已停止") + + +runner = _Runner() diff --git a/backend/app/scheduler/service.py b/backend/app/scheduler/service.py new file mode 100644 index 00000000..1255381f --- /dev/null +++ b/backend/app/scheduler/service.py @@ -0,0 +1,318 @@ +"""定时任务的业务规则：输入模板、和工作流入口字段的对照、发起一次触发。 + +调度循环（loop.py）到点调 fire_once，接口的「立即运行」也调它：重叠跳过、渲染输入、记触发、发起运行这一套 +只有一份。发起本身走 engine/launch.py 的 launch_run，和 HTTP 发起是同一套规则——正式运行只能从当前发布版本 +发起，这一条不会在定时任务这边被绕过。 +""" +from __future__ import annotations + +import asyncio +import logging +import re +import weakref +from datetime import date, datetime, timedelta, timezone +from typing import Any + +from sqlalchemy import select, update +from sqlalchemy.exc import IntegrityError + +from app.db.base import SessionLocal, utcnow +from app.db.models import Run, Schedule, ScheduleFire +from app.engine.expressions import _FILTERS, render_template +from app.scheduler import cron + +logger = logging.getLogger(__name__) + +#: 变量清单的示例按这个时区算（GET /api/schedules/variables 没带时区时） +DEFAULT_TIMEZONE = "Asia/Shanghai" +RUN_CLASSES = ("formal", "exploratory") +#: 这几个状态的运行算「还没结束」：停在审批上、服务重启挂起的（interrupted）也算，它们随时会接着执行 +ACTIVE_RUN = ("queued", "running", "interrupted") + +#: 模板里能用的变量：(变量名, 说明)。说明写在变量清单接口里，和示例一起给界面 +VARIABLES: tuple[tuple[str, str], ...] = ( + ("today", "今天（YYYY-MM-DD）"), + ("yesterday", "昨天（YYYY-MM-DD）"), + ("this_week_start", "本周一（YYYY-MM-DD）"), + ("last_week_start", "上周一（YYYY-MM-DD）"), + ("last_week_end", "上周日（YYYY-MM-DD）"), + ("this_month_start", "本月 1 日（YYYY-MM-DD）"), + ("last_month_start", "上月 1 日（YYYY-MM-DD）"), + ("last_month_end", "上月最后一天（YYYY-MM-DD）"), + ("fire_time", "本次计划触发时间（带时区偏移的 ISO 时间）"), + ("last_fire_time", "上一次计划触发时间（带时区偏移的 ISO 时间）"), +) +_NAMES = tuple(f"schedule.{key}" for key, _ in VARIABLES) +_TEMPLATE_RE = re.compile(r"\{\{(.*?)\}\}", re.DOTALL) + +#: 触发记下来、运行还没发起时先写成这样：发起成功再改成 started。进程在这中间被强杀，留下的记录照实说没有发起 +INTERRUPTED_REASON = "服务在发起运行前停止，这次触发没有发起运行" +#: launch_run 抛了 LaunchRefused 以外的异常（库出错之类）。详细信息只进日志 +GENERIC_REASON = "发起运行时出错，详细信息见服务日志" + + +class TemplateError(ValueError): + """输入模板写错了。str() 是给人看的中文说明，指出是哪个输入。""" + + +# -------------------------------------------------------------------------- +# 模板变量 +# -------------------------------------------------------------------------- + + +def template_values(fire_at: datetime, tz: str, cron_expr: str | cron.CronSpec) -> dict[str, str]: + """按计划触发时间在任务时区算出的变量（键不带 schedule. 前缀）。日期写 YYYY-MM-DD，时间写带偏移的 ISO、只到秒。 + + 按 fire_at 算而不是按「现在」：补发晚了一小时、立即运行在零点前后点下去，「昨天」都是那次触发的昨天。 + last_fire_time 是按 cron 往前找的上一次计划触发，不看历史记录：同一个 (cron, 时区, fire_at) 永远渲染出同一份输入。 + """ + zone = cron.zone(tz) + local = fire_at.astimezone(zone) + today = local.date() + week_start = today - timedelta(days=today.weekday()) + month_start = today.replace(day=1) + last_month_end = month_start - timedelta(days=1) + previous = cron.prev_fire(cron_expr, zone, fire_at) + return { + "today": _day(today), + "yesterday": _day(today - timedelta(days=1)), + "this_week_start": _day(week_start), + "last_week_start": _day(week_start - timedelta(days=7)), + "last_week_end": _day(week_start - timedelta(days=1)), + "this_month_start": _day(month_start), + "last_month_start": _day(last_month_end.replace(day=1)), + "last_month_end": _day(last_month_end), + "fire_time": local.isoformat(timespec="seconds"), + "last_fire_time": previous.astimezone(zone).isoformat(timespec="seconds") if previous else "", + } + + +def _day(value: date) -> str: + return value.isoformat() + + +def render_inputs(template: dict[str, str], values: dict[str, str]) -> dict[str, str]: + """渲染输入模板。用项目通用的 {{ }} 渲染（engine/expressions.render_template），变量挂在 schedule 下。""" + ctx = {"schedule": values} + return {name: render_template(text, ctx) for name, text in template.items()} + + +def check_template(template: Any) -> dict[str, str]: + """校验输入模板，原样返回（None 当空模板）。写错抛 TemplateError。 + + 项目的模板渲染取不到值时渲染成空串、不报错——画布上的半成品图要能运行。定时任务没人盯着，拼错的变量名 + 会让每一次运行都拿到空的日期，所以保存和预览时就要说出来。 + """ + if template is None: + return {} + if not isinstance(template, dict): + raise TemplateError("输入模板需要是「字段名：模板」的对象，例如 stat_date 对应 {{ schedule.yesterday }}") + for name, text in template.items(): + if not isinstance(name, str) or not name.strip(): + raise TemplateError("输入模板里有空的字段名，请填写工作流的输入字段名") + if not isinstance(text, str): + raise TemplateError(f"输入「{name}」的值需要是文本（可以带 {{{{ schedule.yesterday }}}} 这样的变量），当前为「{text}」") + _check_text(name, text) + return dict(template) + + +def _check_text(name: str, text: str) -> None: + for match in _TEMPLATE_RE.finditer(text): + expr = match.group(1).strip() + if not expr: + raise TemplateError(f"输入「{name}」的模板里有空的 {{{{ }}}}，请在中间写变量名，例如 schedule.yesterday") + path, *filters = (part.strip() for part in expr.split("|")) + if path not in _NAMES: + raise TemplateError(f"输入「{name}」的模板里「{path}」不是可用的变量，可用的变量：{'、'.join(_NAMES)}") + for filt in filters: + if filt not in _FILTERS: + raise TemplateError(f"输入「{name}」的模板里「{filt}」不是可用的过滤器，可用的过滤器：{'、'.join(_FILTERS)}") + rest = _TEMPLATE_RE.sub("", text) + if "{{" in rest or "}}" in rest: + raise TemplateError(f"输入「{name}」的模板里 {{{{ 和 }}}} 没有成对出现，请检查是否少写了一半") + + +def variables(now: datetime | None = None, tz: str = DEFAULT_TIMEZONE) -> list[dict[str, str]]: + """变量清单：[{name, label, example}]。示例按 now（默认现在）这一刻触发算；上一次计划触发时间没有 cron + 可依，示例按每天同一时刻触发算。""" + moment = (now or utcnow()).replace(second=0, microsecond=0) + local = moment.astimezone(cron.zone(tz)) + daily = f"{local.minute} {local.hour} * * *" + values = template_values(moment, tz, daily) + return [{"name": f"schedule.{key}", "label": label, "example": values[key]} for key, label in VARIABLES] + + +# -------------------------------------------------------------------------- +# 和工作流入口字段对照 +# -------------------------------------------------------------------------- + + +def input_fields(graph: Any) -> list[dict[str, Any]]: + """工作流入口节点声明的字段 [{name, required, default}]，没声明（或图不成形）是空列表。""" + from app.engine.schema import GraphSpec, NodeType + + try: + spec = GraphSpec.model_validate(graph) + except Exception: # noqa: BLE001 - 图不成形交给发起时报 + return [] + out: list[dict[str, Any]] = [] + for node in spec.nodes: + if node.type != NodeType.INPUT: + continue + for field in node.config.get("fields") or []: + if isinstance(field, dict) and isinstance(field.get("name"), str) and field["name"].strip(): + out.append({"name": field["name"].strip(), "required": bool(field.get("required")), + "default": field.get("default")}) + return out + + +def check_fields(template: dict[str, str], fields: list[dict[str, Any]], workflow_name: str, *, + require: bool = True) -> None: + """入口声明了字段时，模板里的字段名要在里面；require 时必填且没有默认值的字段要有模板。不对抛 TemplateError。 + + 入口没声明字段的工作流，输入是开放的（运行时传什么键都收），不查。预览时用户还在填，只查多出来的名字。 + """ + if not fields: + return + names = [f["name"] for f in fields] + unknown = [name for name in template if name not in names] + if unknown: + raise TemplateError(f"输入「{unknown[0]}」不是工作流「{workflow_name}」入口声明的字段，可用的字段：{'、'.join(names)}") + if not require: + return + missing = [f["name"] for f in fields + if f["required"] and f["default"] in (None, "") and f["name"] not in template] + if missing: + raise TemplateError(f"工作流「{workflow_name}」的必填输入「{'、'.join(missing)}」没有填写，" + "到点发起的运行会因缺少输入而失败") + + +# -------------------------------------------------------------------------- +# 发起一次触发 +# -------------------------------------------------------------------------- + +#: 每个事件循环一把锁：发起（查重叠 → 记触发 → 发起运行 → 改记录）和通知扫描查库互斥。单进程部署，进程内 +#: 互斥就够；按事件循环分开是因为测试每个用例一个新循环，跨循环用同一把 asyncio.Lock 会报错 +_LOCKS: "weakref.WeakKeyDictionary[asyncio.AbstractEventLoop, asyncio.Lock]" = weakref.WeakKeyDictionary() + + +def launch_lock() -> asyncio.Lock: + loop = asyncio.get_running_loop() + lock = _LOCKS.get(loop) + if lock is None: + lock = _LOCKS[loop] = asyncio.Lock() + return lock + + +def actor_label(name: str) -> str: + """定时任务发起的运行记在谁名下。""" + return f"定时任务：{name}"[:100] + + +def next_run(cron_expr: str, tz: str, enabled: bool, now: datetime | None = None) -> datetime | None: + """下一次触发的时刻（UTC）；停用的任务没有下一次。""" + if not enabled: + return None + return cron.next_fire(cron_expr, tz, now or utcnow()) + + +async def fire_once(schedule_id: str, fire_at: datetime, kind: str, *, actor: str | None = None, + ) -> ScheduleFire | None: + """处理一次触发，返回记下的触发记录。同一 (任务, fire_at, kind) 已经有记录（两拍撞在一起、立即运行连点两下） + 时不再发起，返回已有的那一条；任务不存在返回 None。 + + 上一次运行还没结束就记 skipped_overlap；输入渲染不了、launch_run 拒绝或出错记 failed_to_start（拒绝写原话， + 其他异常写通用说明、详细信息进日志）；发起成功记 started 和运行 id。 + """ + async with launch_lock(): + async with SessionLocal() as session: + schedule = await session.get(Schedule, schedule_id) + if schedule is None: + return None + name, workflow_id, run_class = schedule.name, schedule.workflow_id, schedule.run_class + tz, cron_expr, template = schedule.timezone, schedule.cron, schedule.input_template + try: + inputs = render_inputs(check_template(template), template_values(fire_at, tz, cron_expr)) + except (TemplateError, cron.CronError, cron.TimezoneError) as e: + return await _record(schedule_id, fire_at, kind, "failed_to_start", f"输入模板无法渲染：{e}", {}) + busy = await _active_run(schedule_id) + if busy is not None: + reason = f"上一次运行（{busy[:8]}）还没有结束，按设置跳过这次触发" + return await _record(schedule_id, fire_at, kind, "skipped_overlap", reason, inputs) + fire = await _record(schedule_id, fire_at, kind, "failed_to_start", INTERRUPTED_REASON, inputs, fresh=True) + if fire is None: + return await _existing(schedule_id, fire_at, kind) + from app.engine.launch import LaunchRefused, launch_run + + try: + async with SessionLocal() as session: + run = await launch_run(session, workflow_id=workflow_id, run_class=run_class, input_payload=inputs, + actor=actor or actor_label(name), schedule_id=schedule_id) + except LaunchRefused as e: + return await _finish(fire.id, status="failed_to_start", reason=e.message) + except Exception: # noqa: BLE001 - 发起失败不能让调度循环停下；详细信息进日志，记录上写通用说明 + logger.exception("定时任务「%s」（%s）发起运行失败", name, schedule_id) + return await _finish(fire.id, status="failed_to_start", reason=GENERIC_REASON) + return await _finish(fire.id, status="started", reason=None, run_id=run.id) + + +async def record_skip(schedule_id: str, fire_at: datetime, reason: str) -> ScheduleFire | None: + """记一次错过、没有补发的触发（kind=misfire，status=skipped_misfire）。""" + return await _record(schedule_id, fire_at, "misfire", "skipped_misfire", reason, {}) + + +async def _active_run(schedule_id: str) -> str | None: + """这个任务发起的、还没结束的运行 id（最近的那一个），没有返回 None。""" + async with SessionLocal() as session: + return await session.scalar( + select(Run.id).where(Run.schedule_id == schedule_id, Run.status.in_(ACTIVE_RUN)) + .order_by(Run.created_at.desc()).limit(1)) + + +async def _record(schedule_id: str, fire_at: datetime, kind: str, status: str, reason: str | None, + inputs: dict[str, Any], *, fresh: bool = False) -> ScheduleFire | None: + """插入一条触发记录。撞上唯一约束（同一次触发已经记过）或任务已被删除（外键）时：fresh 返回 None，否则返回 + 已有的那一条（没有就是 None）。""" + fire = ScheduleFire(schedule_id=schedule_id, fire_at=fire_at, kind=kind, status=status, reason=reason, + inputs=inputs) + async with SessionLocal() as session: + session.add(fire) + try: + await session.commit() + except IntegrityError: + await session.rollback() + return None if fresh else await _existing(schedule_id, fire_at, kind) + await session.refresh(fire) + return fire + + +async def _existing(schedule_id: str, fire_at: datetime, kind: str) -> ScheduleFire | None: + async with SessionLocal() as session: + return await session.scalar(select(ScheduleFire).where( + ScheduleFire.schedule_id == schedule_id, ScheduleFire.fire_at == fire_at, ScheduleFire.kind == kind)) + + +async def _finish(fire_id: str, *, status: str, reason: str | None, run_id: str | None = None) -> ScheduleFire | None: + async with SessionLocal() as session: + await session.execute(update(ScheduleFire).where(ScheduleFire.id == fire_id) + .values(status=status, reason=reason, run_id=run_id)) + await session.commit() + return await session.get(ScheduleFire, fire_id) + + +def local_text(moment: datetime, tz: str) -> str: + """给人看的本地时间「2026-10-05 09:00」，写进跳过原因、通知里。""" + return moment.astimezone(cron.zone(tz)).strftime("%Y-%m-%d %H:%M") + + +def duration_text(seconds: int) -> str: + """补发宽限这类秒数的说法：3600 →「1 小时」，600 →「10 分钟」，90 →「90 秒」。""" + if seconds and seconds % 3600 == 0: + return f"{seconds // 3600} 小时" + if seconds and seconds % 60 == 0: + return f"{seconds // 60} 分钟" + return f"{seconds} 秒" + + +def as_utc(value: datetime) -> datetime: + return value.astimezone(timezone.utc) if value.tzinfo else value.replace(tzinfo=timezone.utc) diff --git a/backend/app/scheduler/webhook.py b/backend/app/scheduler/webhook.py new file mode 100644 index 00000000..390fb0b4 --- /dev/null +++ b/backend/app/scheduler/webhook.py @@ -0,0 +1,567 @@ +"""定时任务的通知：运行结束、失败、停在审批时按任务的设置推送 Webhook。 + +四种格式：通用 JSON、飞书（文本消息）、企业微信（markdown）、钉钉（markdown）。配了签名密钥时：通用格式在请求头 +里带 HMAC-SHA256 签名，飞书、钉钉按各自平台的算法签（飞书放在请求体，钉钉放在地址上）；企业微信的 key 本来就在 +地址里，没有另外的签名。 + +Webhook 地址本身就是凭证（企业微信的 key、钉钉的 access_token 都在地址里，飞书的在路径里），和签名密钥一样 +加密落库，日志、推送记录、接口返回里只出现主机名。httpx 每发一个请求都会在 INFO 级别记一行完整地址，这里装了 +一个过滤器把推送期间的那几行换成主机名。 + +通知里只有任务名、状态、出具档位、缺口条数、运行链接、封存哈希前 12 位和起止时间，不放报告里的具体数字，也不放 +失败时的原始报错——报错里可能带着查询到的数据，Webhook 推去的群不一定有权看。 +""" +from __future__ import annotations + +import asyncio +import base64 +import contextvars +import hashlib +import hmac +import json +import logging +from dataclasses import dataclass +from datetime import datetime, timedelta, timezone +from typing import Any +from urllib.parse import parse_qsl, quote_plus, urlsplit + +import httpx +from sqlalchemy import and_, func, or_, select, update + +from app.core.crypto import decrypt +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 + +logger = logging.getLogger(__name__) + +FORMATS = ("generic", "feishu", "wecom", "dingtalk") +EVENTS = ("finished", "failed", "awaiting_approval") +#: 单次请求的超时（秒） +TIMEOUT_S = 10.0 +#: 失败后再试 3 次，每次之前等这么久（秒）。4xx（429 除外）不重试：地址或参数错了，重试也一样 +RETRY_DELAYS: tuple[float, ...] = (1.0, 2.0, 4.0) +#: 只看这么多天内的触发：更早的运行状态不会再变，不用每一拍都扫 +SCAN_DAYS = 30 +#: 运行结束后等封存（清单哈希）最多等多久。runner 先写终态再封存，两步之间被这一拍撞上就等下一拍 +SEAL_WAIT_S = 60 +#: 测试用：换成 httpx.MockTransport 接住请求 +_transport: httpx.AsyncBaseTransport | None = None + +_HEADLINE = {"finished": "运行完成", "failed": "运行失败", "awaiting_approval": "等待审批", "test": "Webhook 测试"} +#: 运行状态在通知里的叫法 +_STATUS_LABEL = {"succeeded": "已完成", "failed": "失败", "interrupted": "停在人工审批", "cancelled": "已取消", + "failed_to_start": "未能发起"} + + +# -------------------------------------------------------------------------- +# 地址：校验、提示、脱敏 +# -------------------------------------------------------------------------- + + +def check_url(url: Any) -> str: + """校验 Webhook 地址，返回去掉首尾空白的地址。不对抛 ValueError（中文说明，不回显地址）。""" + text = url.strip() if isinstance(url, str) else "" + try: + parts = urlsplit(text) + except ValueError: + parts = None + if (not text or parts is None or parts.scheme not in ("http", "https") or not parts.hostname + or any(c.isspace() for c in text)): + raise ValueError("Webhook 地址需要是 http:// 或 https:// 开头的完整地址，例如飞书机器人的 " + "https://open.feishu.cn/open-apis/bot/v2/hook/…") + if len(text) > 2000: + raise ValueError("Webhook 地址过长（超过 2000 个字符），请检查是否粘贴了多余的内容") + return text + + +def host_of(url: str) -> str: + """主机名（带非默认端口）。日志、报错里只写它。""" + try: + parts = urlsplit(url) + host = parts.hostname or "?" + return f"{host}:{parts.port}" if parts.port else host + except ValueError: + return "?" + + +def url_hint(url: str) -> str: + """给界面看的地址提示：只露主机和末 4 位（open.feishu.cn/…ab12）。""" + return f"{host_of(url)}/…{url[-4:]}" + + +def _sensitive(url: str, secret: str | None) -> list[str]: + """一条地址里不能出现在日志、报错里的片段：整条地址、查询参数的值（key、access_token）、路径里的长片段（飞书 + 的令牌在路径里）、签名密钥。长的在前，免得先换掉短的、长的就认不出来了。""" + out = {url, secret or ""} + try: + parts = urlsplit(url) + out |= {value for _, value in parse_qsl(parts.query)} + out |= {seg for seg in parts.path.split("/") if len(seg) >= 12} + out.add(f"{parts.netloc}{parts.path}") + except ValueError: + pass + return sorted((s for s in out if len(s) >= 4), key=len, reverse=True) + + +def _scrub(text: str, url: str, secret: str | None) -> str: + for piece in _sensitive(url, secret): + text = text.replace(piece, "…") + return text + + +#: 正在推送哪条地址、用的哪个密钥。httpx 记请求日志时还在同一个任务里，过滤器据此只改推送的那几行 +_SENDING: contextvars.ContextVar[tuple[str, str | None] | None] = contextvars.ContextVar("webhook_sending", default=None) + + +class _RedactWebhookUrls(logging.Filter): + """把 httpx 记下的推送请求的地址换成主机名。别的请求（模型接入、数据源工具）原样放过。""" + + def filter(self, record: logging.LogRecord) -> bool: + sending = _SENDING.get() + if sending and isinstance(record.args, tuple): + url, secret = sending + record.args = tuple( + url_hint(str(arg)) if isinstance(arg, httpx.URL) + else _scrub(arg, url, secret) if isinstance(arg, str) else arg + for arg in record.args) + return True + + +def _install_filter() -> None: + target = logging.getLogger("httpx") + if not any(isinstance(f, _RedactWebhookUrls) for f in target.filters): + target.addFilter(_RedactWebhookUrls()) + + +_install_filter() + + +# -------------------------------------------------------------------------- +# 消息内容 +# -------------------------------------------------------------------------- + + +@dataclass +class Notice: + """一条通知要说的事。event 是 finished / failed / awaiting_approval / test；status 是运行状态，没能发起的是 + failed_to_start。没有的项为 None。""" + + event: str + schedule_id: str + schedule_name: str + timezone: str + fire_id: str | None = None + fire_at: datetime | None = None + fire_kind: str | None = None + run_id: str | None = None + status: str | None = None + tier: str | None = None + gaps: int | None = None + run_url: str | None = None + manifest_hash: str | None = None + started_at: datetime | None = None + finished_at: datetime | None = None + reason: str | None = None + pending_approvals: int | None = None + + +def _local(moment: datetime | None, tz: str, *, seconds: bool = True) -> str | None: + if moment is None: + return None + try: + from app.scheduler import cron + + local = moment.astimezone(cron.zone(tz)) + except ValueError: + local = moment.astimezone(timezone.utc) + return local.strftime("%Y-%m-%d %H:%M:%S" if seconds else "%Y-%m-%d %H:%M") + + +def message(notice: Notice) -> tuple[str, list[tuple[str, str]], str | None]: + """(标题, [(项, 内容)…], 运行链接)。三种文本格式都从这一份排出来。""" + if notice.event == "failed" and notice.status == "failed_to_start": + headline = "未能发起运行" + else: + headline = _HEADLINE.get(notice.event, notice.event) + title = f"定时任务「{notice.schedule_name}」{headline}" + lines: list[tuple[str, str]] = [] + if notice.event == "test": + lines.append(("说明", "这是一条测试消息，收到说明 Webhook 地址和签名配置正确")) + return title, lines, None + if notice.status: + lines.append(("状态", _STATUS_LABEL.get(notice.status, notice.status))) + 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.pending_approvals: + lines.append(("待审批", f"{notice.pending_approvals} 项，请到 AgentLab 处理")) + if notice.status == "failed_to_start" and notice.reason: + lines.append(("原因", notice.reason)) + elif notice.status == "failed": + lines.append(("原因", "运行失败，详情请查看运行记录")) + if notice.fire_at is not None: + lines.append(("计划触发", f"{_local(notice.fire_at, notice.timezone, seconds=False)}（{notice.timezone}）")) + if notice.started_at is not None: + lines.append(("开始", _local(notice.started_at, notice.timezone) or "")) + if notice.finished_at is not None: + lines.append(("结束", _local(notice.finished_at, notice.timezone) or "")) + if notice.manifest_hash: + lines.append(("封存哈希", notice.manifest_hash[:12])) + if notice.run_id: + lines.append(("运行编号", notice.run_id)) + return title, lines, notice.run_url + + +def _text(notice: Notice) -> str: + title, lines, link = message(notice) + out = [f"【AgentLab】{title}", *(f"{k}：{v}" for k, v in lines)] + if link: + out.append(f"查看运行：{link}") + return "\n".join(out) + + +def _wecom_markdown(notice: Notice) -> str: + title, lines, link = 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})") + return "\n".join(out) + + +def _dingtalk_markdown(notice: Notice) -> tuple[str, str]: + title, lines, link = message(notice) + out = [f"#### 【AgentLab】{title}", "", *(f"- {k}：{v}" for k, v in lines)] + if link: + out += ["", f"[查看运行]({link})"] + return title, "\n".join(out) + + +def _iso(moment: datetime | None) -> str | None: + return service.as_utc(moment).isoformat() if moment is not None else None + + +def _generic(notice: Notice, now: datetime) -> dict[str, Any]: + title, _, _ = message(notice) + return { + "source": "agentlab", + "event": notice.event, + "title": title, + "text": _text(notice), + "schedule_id": notice.schedule_id, + "schedule_name": notice.schedule_name, + "fire_id": notice.fire_id, + "fire_at": _iso(notice.fire_at), + "fire_kind": notice.fire_kind, + "run_id": notice.run_id, + "status": notice.status, + "tier": notice.tier, + "tier_label": ISSUANCE_LABEL.get(notice.tier) if notice.tier else None, + "gaps": notice.gaps, + "pending_approvals": notice.pending_approvals, + "run_url": notice.run_url, + "manifest_hash": notice.manifest_hash[:12] if notice.manifest_hash else None, + "started_at": _iso(notice.started_at), + "finished_at": _iso(notice.finished_at), + "reason": notice.reason if notice.status == "failed_to_start" else None, + "sent_at": _iso(now), + } + + +def build_request(fmt: str, url: str, secret: str | None, notice: Notice, *, + now: datetime | None = None) -> tuple[str, dict[str, str], bytes]: + """(请求地址, 请求头, 请求体)。签名里的时间戳按 now 取，每次重试重新算：飞书要求时间戳在一小时内。""" + now = now or utcnow() + headers = {"Content-Type": "application/json"} + if fmt == "feishu": + body: dict[str, Any] = {"msg_type": "text", "content": {"text": _text(notice)}} + if secret: + ts = str(int(now.timestamp())) + # 飞书自定义机器人的签名：以「timestamp + 换行 + 密钥」为键，对空串做 HMAC-SHA256，再 base64 + key = f"{ts}\n{secret}".encode() + body = {"timestamp": ts, "sign": base64.b64encode(hmac.new(key, b"", hashlib.sha256).digest()).decode(), + **body} + return url, headers, _json(body) + if fmt == "wecom": + return url, headers, _json({"msgtype": "markdown", "markdown": {"content": _wecom_markdown(notice)}}) + if fmt == "dingtalk": + title, text = _dingtalk_markdown(notice) + target = url + if secret: + # 钉钉自定义机器人的签名：毫秒时间戳，以密钥为键对「timestamp + 换行 + 密钥」做 HMAC-SHA256，base64 后 + # 再做 URL 编码，和 timestamp 一起加在地址上 + ts = str(int(now.timestamp() * 1000)) + digest = hmac.new(secret.encode(), f"{ts}\n{secret}".encode(), hashlib.sha256).digest() + sign = quote_plus(base64.b64encode(digest).decode()) + target = f"{url}{'&' if '?' in url else '?'}timestamp={ts}&sign={sign}" + return target, headers, _json({"msgtype": "markdown", "markdown": {"title": title, "text": text}}) + payload = _json(_generic(notice, now)) + if secret: + # 通用格式：接收端按 HMAC-SHA256(密钥, 「时间戳.请求体」) 验签，时间戳防重放 + ts = str(int(now.timestamp())) + mac = hmac.new(secret.encode(), f"{ts}.".encode() + payload, hashlib.sha256).hexdigest() + headers |= {"X-AgentLab-Timestamp": ts, "X-AgentLab-Signature": f"sha256={mac}"} + return url, headers, payload + + +def _json(value: dict[str, Any]) -> bytes: + return json.dumps(value, ensure_ascii=False, separators=(",", ":")).encode() + + +# -------------------------------------------------------------------------- +# 推送 +# -------------------------------------------------------------------------- + + +@dataclass +class Outcome: + """一次推送（含重试）的结果。error 是中文说明，只有主机名，不含地址和密钥。""" + + ok: bool + http_status: int | None + attempts: int + error: str | None = None + #: 这一次失败值不值得再试（只在单次尝试的结果上有意义） + retryable: bool = False + + +def _client() -> httpx.AsyncClient: + return httpx.AsyncClient(timeout=TIMEOUT_S, transport=_transport, follow_redirects=False) + + +async def send(fmt: str, url: str, secret: str | None, notice: Notice, *, retry: bool = True) -> Outcome: + """推送一条通知。retry 时失败再试 3 次（退避），4xx（429 除外）不重试。失败只记主机名。""" + host = host_of(url) + delays = RETRY_DELAYS if retry else () + last = Outcome(False, None, 0, "没有发出请求") + async with _client() as client: + for attempt in range(len(delays) + 1): + if attempt: + await asyncio.sleep(delays[attempt - 1]) + target, headers, body = build_request(fmt, url, secret, notice) + last = await _post_once(client, fmt, target, headers, body, url, secret) + last.attempts = attempt + 1 + if last.ok: + return last + logger.warning("定时任务「%s」的 Webhook（%s）第 %d 次推送失败：%s", + notice.schedule_name, host, attempt + 1, last.error) + if not last.retryable: + break + return last + + +async def _post_once(client: httpx.AsyncClient, fmt: str, target: str, headers: dict[str, str], body: bytes, + url: str, secret: str | None) -> Outcome: + host = host_of(url) + token = _SENDING.set((url, secret)) + try: + response = await client.post(target, content=body, headers=headers) + except httpx.TimeoutException: + return Outcome(False, None, 1, f"请求 {host} 超时（{TIMEOUT_S:g} 秒）", retryable=True) + except httpx.ConnectError: + return Outcome(False, None, 1, f"无法连接 {host}", retryable=True) + except httpx.HTTPError as e: + return Outcome(False, None, 1, f"请求 {host} 失败（{type(e).__name__}）", retryable=True) + finally: + _SENDING.reset(token) + status = response.status_code + reply = _reply_json(response) + if not 200 <= status < 300: + detail = _platform_message(reply) + text = f"{host} 返回 HTTP {status}" + (f"：{detail}" if detail else "") + return Outcome(False, status, 1, _scrub(text, url, secret)[:300], retryable=status == 429 or status >= 500) + if problem := _platform_error(fmt, reply): + return Outcome(False, status, 1, _scrub(f"{host} {problem}", url, secret)[:300], retryable=True) + return Outcome(True, status, 1) + + +def _reply_json(response: httpx.Response) -> dict[str, Any] | None: + try: + data = response.json() + except ValueError: + return None + return data if isinstance(data, dict) else None + + +def _platform_message(reply: dict[str, Any] | None) -> str: + if not reply: + return "" + text = reply.get("errmsg") or reply.get("msg") or reply.get("StatusMessage") or "" + return str(text)[:200] + + +def _platform_error(fmt: str, reply: dict[str, Any] | None) -> str | None: + """HTTP 200 里的平台错误码：飞书 code（老接口 StatusCode），企业微信、钉钉 errcode。0 是成功。 + + 说明里写主机名（调用方拼在前面），不写平台名：主机名已经说明是哪个平台，配错了格式时也不会张冠李戴。""" + if not reply or fmt == "generic": + return None + code = reply.get("code", reply.get("StatusCode")) if fmt == "feishu" else reply.get("errcode") + if code in (None, 0, "0"): + return None + detail = _platform_message(reply) + return f"返回错误码 {code}" + (f"：{detail}" if detail else "") + + +async def _record(schedule_id: str, fire_id: str | None, run_id: str | None, event: str, outcome: Outcome) -> None: + async with SessionLocal() as session: + session.add(WebhookDelivery(schedule_id=schedule_id, fire_id=fire_id, run_id=run_id, event=event, + status="ok" if outcome.ok else "failed", http_status=outcome.http_status, + attempts=outcome.attempts, error=outcome.error)) + try: + await session.commit() + except Exception: # noqa: BLE001 - 任务刚被删掉（外键）之类：推送记录记不下不影响别的 + logger.warning("定时任务 %s 的推送记录没能保存", schedule_id) + + +# -------------------------------------------------------------------------- +# 每一拍的通知扫描 +# -------------------------------------------------------------------------- + +#: 运行到了这些状态才可能要推送 +_WATCHED = ("succeeded", "failed", "cancelled", "interrupted") +#: 运行状态 → 事件。cancelled 是有人手动停的，不推 +_EVENT_OF = {"succeeded": "finished", "failed": "failed", "cancelled": None, "interrupted": "awaiting_approval"} + + +@dataclass +class _Job: + fmt: str + url: str + secret: str | None + notice: Notice + + +async def notify_due(now: datetime | None = None) -> int: + """查已发起的触发，运行状态有变化的按订阅推送一次。返回这一拍推送了几条（含失败的）。 + + 处理过的状态记在 notified_status 上，用条件 UPDATE 认领（WHERE notified_status = 旧值）：两拍撞在一起也只 + 推一次。没订阅这个事件、任务没配 Webhook 的也记下已处理——之后再配上 Webhook，不会把旧的补推一遍。 + """ + now = now or utcnow() + jobs: list[_Job] = [] + # 和发起共用一把锁：发起时先记一条「未能发起」再发起运行、成功了才改成 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_( + 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)), + )) + .order_by(ScheduleFire.created_at) + )).all() + base_url: str | None = None + if rows: + from app.api.settings import notify_settings + + 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 not await _mark(fire.id, fire.notified_status, state): + continue + 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) + + +async def _state_of(session: Any, fire: ScheduleFire, run: Run | None, + now: datetime) -> tuple[str | None, str | None, int]: + """(要记下的已处理状态, 要推送的事件, 待审批数)。还不能下结论（没封存、挂起但没有待审批）时状态为 None。""" + if fire.status == "failed_to_start": + return "failed_to_start", "failed", 0 + if run is None: + return None, None, 0 + if run.status in ("succeeded", "failed", "cancelled"): + finished = run.finished_at or now + if not run.manifest_hash and (now - service.as_utc(finished)).total_seconds() < SEAL_WAIT_S: + return None, None, 0 + return run.status, _EVENT_OF[run.status], 0 + if run.status == "interrupted": + # 服务重启挂起的运行也是 interrupted，但没人要审批：确有待审批才算停在审批 + pending = await session.scalar(select(func.count()).select_from(Approval).where( + Approval.run_id == run.id, Approval.status == "pending")) or 0 + return ("interrupted", "awaiting_approval", pending) if pending else (None, None, 0) + return None, None, 0 + + +async def _mark(fire_id: str, old: str | None, state: str) -> bool: + cond = ScheduleFire.notified_status.is_(None) if old is None else ScheduleFire.notified_status == old + async with SessionLocal() as session: + result = await session.execute(update(ScheduleFire).where(ScheduleFire.id == fire_id, cond) + .values(notified_status=state)) + await session.commit() + 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: + tier, gaps = await _issuance_of(session, run) if run is not None else (None, None) + return Notice( + event=event, 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, + 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, + pending_approvals=pending or None, + ) + + +async def _issuance_of(session: Any, run: Run) -> tuple[str | None, int | None]: + """出具档位和缺口条数：成果上的 _issuance，取不到时退回最后一条 issuance 事件（同 api/evidence.py 的读法）。""" + issuance = (run.output or {}).get("_issuance") if isinstance(run.output, dict) else None + if not isinstance(issuance, dict): + issuance = await session.scalar(select(RunEvent.data).where( + RunEvent.run_id == run.id, RunEvent.type == "issuance").order_by(RunEvent.seq.desc()).limit(1)) + if not isinstance(issuance, dict) or not issuance.get("tier"): + return None, None + gaps = issuance.get("gaps") + return str(issuance["tier"]), len(gaps) if isinstance(gaps, list) else None + + +async def _deliver(job: _Job) -> None: + try: + outcome = await send(job.fmt, job.url, job.secret, job.notice) + except Exception: # noqa: BLE001 - 拼消息、签名出错也不能让这一拍的其他推送跟着失败 + logger.exception("定时任务「%s」的 Webhook（%s）推送出错", job.notice.schedule_name, host_of(job.url)) + outcome = Outcome(False, None, 0, "推送时出错，详细信息见服务日志") + await _record(job.notice.schedule_id, job.notice.fire_id, job.notice.run_id, job.notice.event, outcome) + + +async def send_test(schedule_id: str) -> dict[str, Any]: + """按任务存的 Webhook 发一条测试消息（不重试），记一条 event=test 的推送记录。任务不存在抛 KeyError， + 没配 Webhook 抛 ValueError。返回 {ok, http_status, message}。""" + async with SessionLocal() as session: + schedule = await session.get(Schedule, schedule_id) + if schedule is None: + raise KeyError(schedule_id) + url = decrypt(schedule.webhook_url) if schedule.webhook_url else None + if not url or schedule.webhook_format not in FORMATS: + raise ValueError("这个定时任务没有配置 Webhook，请先填写地址并保存") + fmt, secret = schedule.webhook_format, decrypt(schedule.webhook_secret) + notice = Notice(event="test", schedule_id=schedule.id, schedule_name=schedule.name, + timezone=schedule.timezone) + outcome = await send(fmt, url, secret, notice, retry=False) + await _record(schedule_id, None, None, "test", outcome) + if outcome.ok: + return {"ok": True, "http_status": outcome.http_status, "message": f"测试消息已送达（HTTP {outcome.http_status}）"} + return {"ok": False, "http_status": outcome.http_status, "message": f"推送失败：{outcome.error}"} diff --git a/backend/tests/test_scheduler_flag.py b/backend/tests/test_scheduler_flag.py index c0ebb94c..ac72ce7d 100644 --- a/backend/tests/test_scheduler_flag.py +++ b/backend/tests/test_scheduler_flag.py @@ -38,7 +38,7 @@ def test_unrecognized_value_fails_at_startup(monkeypatch): async def test_start_and_stop_are_safe_noops(): - """调度循环还没实现：start / stop 什么都不做，没 start 过直接 stop 也不出错""" + """测试环境关着调度（conftest）：start 不起循环，没 start 过直接 stop 也不出错。开着时的循环见 test_scheduler_loop""" await scheduler.stop() await scheduler.start() await scheduler.stop() diff --git a/backend/tests/test_scheduler_inputs.py b/backend/tests/test_scheduler_inputs.py new file mode 100644 index 00000000..83f48b69 --- /dev/null +++ b/backend/tests/test_scheduler_inputs.py @@ -0,0 +1,182 @@ +"""定时任务的输入模板（app/scheduler/service.py）：日期变量按触发时间在任务时区算，模板写错就说出来。 + +- 变量都按 fire_at（计划触发时间）在任务时区算，不按「现在」：补跑晚了一小时，昨天还是计划那天的昨天 +- 跨月、跨年、跨时区（同一时刻在上海已经是周一，在纽约还是周日）；周从周一算起 +- schedule.last_fire_time 是上一次计划触发的时间（按 cron 往前找），和本次一样写成带偏移的 ISO 时间 +- 模板写错（未知变量、未知过滤器、{{ }} 不成对、空的 {{ }}、值不是文本）抛 TemplateError，中文说明哪个输入 +- 声明了入口字段的工作流：模板里多出来的字段名、缺的必填输入（没有默认值）都拦下 + +夹具的工作流、字段名都是虚构的。 +""" +from __future__ import annotations + +from datetime import datetime, timezone + +import pytest + +from app.scheduler import service +from app.scheduler.service import TemplateError + +UTC = timezone.utc + + +def values(fire_at: datetime, tz: str = "Asia/Shanghai", cron: str = "0 9 * * 1") -> dict[str, str]: + return service.template_values(fire_at, tz, cron) + + +# -------------------------------------------------------------------------- +# 变量按 fire_at 在任务时区算 +# -------------------------------------------------------------------------- + + +def test_monday_morning_in_shanghai(): + v = values(datetime(2026, 10, 5, 1, 0, tzinfo=UTC)) # 上海 2026-10-05（周一）09:00 + assert v == { + "today": "2026-10-05", + "yesterday": "2026-10-04", + "this_week_start": "2026-10-05", + "last_week_start": "2026-09-28", + "last_week_end": "2026-10-04", + "this_month_start": "2026-10-01", + "last_month_start": "2026-09-01", + "last_month_end": "2026-09-30", + "fire_time": "2026-10-05T09:00:00+08:00", + "last_fire_time": "2026-09-28T09:00:00+08:00", + } + + +def test_across_a_month_boundary(): + # 上海 2026-03-01（周日）00:30，UTC 还是 2 月 28 日 + v = values(datetime(2026, 2, 28, 16, 30, tzinfo=UTC), cron="30 0 * * *") + assert (v["today"], v["yesterday"]) == ("2026-03-01", "2026-02-28") + assert (v["this_week_start"], v["last_week_start"], v["last_week_end"]) == ("2026-02-23", "2026-02-16", "2026-02-22") + assert (v["this_month_start"], v["last_month_start"], v["last_month_end"]) == ("2026-03-01", "2026-02-01", "2026-02-28") + assert v["last_fire_time"] == "2026-02-28T00:30:00+08:00" + + +def test_across_a_year_boundary(): + v = values(datetime(2027, 1, 1, 0, 0, tzinfo=UTC), cron="0 8 1 * *") # 上海 2027-01-01（周五）08:00 + assert (v["today"], v["yesterday"]) == ("2027-01-01", "2026-12-31") + assert (v["this_week_start"], v["last_week_start"], v["last_week_end"]) == ("2026-12-28", "2026-12-21", "2026-12-27") + assert (v["this_month_start"], v["last_month_start"], v["last_month_end"]) == ("2027-01-01", "2026-12-01", "2026-12-31") + assert v["last_fire_time"] == "2026-12-01T08:00:00+08:00" + + +def test_leap_february_ends_on_the_29th(): + v = values(datetime(2028, 3, 1, 1, 0, tzinfo=UTC), cron="0 9 1 * *") + assert (v["last_month_start"], v["last_month_end"]) == ("2028-02-01", "2028-02-29") + + +def test_same_moment_in_two_timezones(): + """UTC 2026-10-04 20:00：上海已经是周一 04:00，纽约还是周日 16:00""" + moment = datetime(2026, 10, 4, 20, 0, tzinfo=UTC) + shanghai = values(moment, "Asia/Shanghai", "0 4 * * *") + new_york = values(moment, "America/New_York", "0 16 * * *") + assert (shanghai["today"], shanghai["this_week_start"]) == ("2026-10-05", "2026-10-05") + assert (new_york["today"], new_york["this_week_start"]) == ("2026-10-04", "2026-09-28") + assert shanghai["fire_time"] == "2026-10-05T04:00:00+08:00" + assert new_york["fire_time"] == "2026-10-04T16:00:00-04:00" + + +def test_fire_time_drops_sub_seconds(): + """立即运行的触发时间是点下去的那一刻，写进输入时只到秒""" + v = values(datetime(2026, 10, 5, 3, 4, 5, 678901, tzinfo=UTC)) + assert v["fire_time"] == "2026-10-05T11:04:05+08:00" + + +# -------------------------------------------------------------------------- +# 渲染与校验 +# -------------------------------------------------------------------------- + + +def test_render_inputs(): + v = values(datetime(2026, 10, 5, 1, 0, tzinfo=UTC)) + out = service.render_inputs({ + "stat_date": "{{ schedule.yesterday }}", + "week": "{{schedule.last_week_start}}~{{ schedule.last_week_end }}", + "region": "华东", + "tag": "{{ schedule.today | upper }}", + }, v) + assert out == {"stat_date": "2026-10-04", "week": "2026-09-28~2026-10-04", "region": "华东", "tag": "2026-10-05"} + + +def test_valid_templates_pass(): + assert service.check_template({"a": "{{ schedule.today }}", "b": "固定值", "c": "{{ schedule.fire_time | trim }}"}) == { + "a": "{{ schedule.today }}", "b": "固定值", "c": "{{ schedule.fire_time | trim }}"} + assert service.check_template(None) == {} + + +@pytest.mark.parametrize("template, fragments", [ + ({"stat_date": "{{ schedule.yesterdy }}"}, ["stat_date", "schedule.yesterdy", "schedule.yesterday"]), + ({"stat_date": "{{ input.day }}"}, ["stat_date", "input.day"]), + ({"stat_date": "{{ schedule.today"}, ["stat_date", "成对"]), + ({"stat_date": "schedule.today }}"}, ["stat_date", "成对"]), + ({"stat_date": "{{ }}"}, ["stat_date", "空的"]), + ({"stat_date": "{{ schedule.today | shout }}"}, ["stat_date", "shout", "过滤器"]), + ({"stat_date": 20261005}, ["stat_date", "文本"]), + ({"": "{{ schedule.today }}"}, ["字段名"]), + (["{{ schedule.today }}"], ["输入模板"]), +]) +def test_bad_templates_explain_in_chinese(template, fragments): + with pytest.raises(TemplateError) as caught: + service.check_template(template) + for fragment in fragments: + assert fragment in str(caught.value), str(caught.value) + + +def test_variable_catalog(): + items = service.variables(datetime(2026, 10, 5, 1, 0, tzinfo=UTC), "Asia/Shanghai") + names = [i["name"] for i in items] + assert names == [f"schedule.{key}" for key in ( + "today", "yesterday", "this_week_start", "last_week_start", "last_week_end", + "this_month_start", "last_month_start", "last_month_end", "fire_time", "last_fire_time")] + by_name = {i["name"]: i for i in items} + assert by_name["schedule.yesterday"] == {"name": "schedule.yesterday", "label": "昨天（YYYY-MM-DD）", + "example": "2026-10-04"} + assert all(i["label"] and i["example"] for i in items) + + +# -------------------------------------------------------------------------- +# 和工作流入口字段对照 +# -------------------------------------------------------------------------- + + +def graph(*fields: dict) -> dict: + return { + "nodes": [ + {"id": "start", "type": "input", "position": {"x": 0, "y": 0}, + "data": {"label": "输入", "config": {"fields": list(fields)}}}, + {"id": "out", "type": "output", "position": {"x": 300, "y": 0}, + "data": {"label": "成果", "config": {"fields": [{"name": "r", "value": "完成"}]}}}, + ], + "edges": [{"source": "start", "target": "out"}], + } + + +def test_fields_of_a_graph(): + g = graph({"name": "stat_date", "required": True}, {"name": "region", "default": "华东"}) + assert [f["name"] for f in service.input_fields(g)] == ["stat_date", "region"] + assert service.input_fields({"nodes": "坏掉的图"}) == [] + + +def test_unknown_field_names_are_refused(): + fields = service.input_fields(graph({"name": "stat_date"}, {"name": "region"})) + with pytest.raises(TemplateError) as caught: + service.check_fields({"stat_dat": "{{ schedule.yesterday }}"}, fields, "门店周报") + assert "stat_dat" in str(caught.value) and "门店周报" in str(caught.value) and "stat_date" in str(caught.value) + + +def test_missing_required_inputs_are_refused_only_when_asked(): + fields = service.input_fields(graph({"name": "stat_date", "required": True}, + {"name": "region", "required": True, "default": "华东"})) + with pytest.raises(TemplateError) as caught: + service.check_fields({}, fields, "门店周报") + assert "stat_date" in str(caught.value) and "必填" in str(caught.value) + # 预览时还在填，不查缺的；有默认值的必填项不算缺 + service.check_fields({}, fields, "门店周报", require=False) + service.check_fields({"stat_date": "{{ schedule.yesterday }}"}, fields, "门店周报") + + +def test_graph_without_declared_fields_accepts_any_name(): + """入口没声明字段时输入是开放的：运行时传什么键都收""" + service.check_fields({"anything": "x"}, service.input_fields(graph()), "门店周报") diff --git a/backend/tests/test_scheduler_loop.py b/backend/tests/test_scheduler_loop.py new file mode 100644 index 00000000..60e8c947 --- /dev/null +++ b/backend/tests/test_scheduler_loop.py @@ -0,0 +1,341 @@ +"""定时任务的调度循环（app/scheduler/loop.py、service.fire_once）。 + +- 到点只发起一次：条件 UPDATE 认领（WHERE next_run_at = 旧值）并推进到下一次，两拍并发也只有一拍认领得到； + 触发记录 (任务, fire_at, kind) 唯一，撞上约束的不再发起 +- 发起走 engine/launch.py 的 launch_run：记在「定时任务：<名字>」名下、带 schedule_id，输入按 fire_at 渲染； + 正式运行用当前发布版本，没有发布版本被拒，拒绝的原话记成 failed_to_start；别的异常记通用说明、详情进日志 +- 上一次的运行还没结束（排队、运行中、停在审批 / 挂起都算）记 skipped_overlap +- 错过的触发：最近一次在宽限内补一次（kind=misfire），更早的积压只记一条跳过；最近一次也超出宽限就记一条 + skipped_misfire，推进到未来的下一次，不补积压 +- start / stop：起一个 asyncio 任务按拍执行，stop 干净地停下；AGENTLAB_SCHEDULER 关着时 start 不起循环 + +工作流、字段、任务名都是虚构的。 +""" +from __future__ import annotations + +import asyncio +import logging +import uuid +from datetime import datetime, timedelta, timezone + +import pytest +from sqlalchemy import delete, func, select + +from app import scheduler +from app.core.config import Settings +from app.db.base import SessionLocal +from app.db.models import Run, Schedule, ScheduleFire, Workflow, WorkflowVersion +from app.engine.runner import run_manager +from app.scheduler import loop, service + +UTC = timezone.utc +#: 上海 2026-10-05（周一）09:00 +NINE = datetime(2026, 10, 5, 1, 0, tzinfo=UTC) + + +@pytest.fixture(autouse=True) +async def engine_up(): + await run_manager.setup() + async with SessionLocal() as session: + await session.execute(delete(Schedule)) + await session.commit() + yield + await run_manager.shutdown() + + +def node(nid, ntype, **config): + return {"id": nid, "type": ntype, "position": {"x": 0, "y": 0}, "data": {"label": nid, "config": config}} + + +def graph(text: str) -> dict: + return { + "nodes": [node("start", "input", fields=[{"name": "stat_date", "required": True}]), + node("out", "output", fields=[{"name": "r", "value": f"{text} {{{{ input.stat_date }}}}"}])], + "edges": [{"source": "start", "target": "out"}], + } + + +async def workflow(*, published: bool = True, canvas: str = "画布", snapshot: str = "发布版") -> str: + async with SessionLocal() as session: + wf = Workflow(name=f"门店周报-{uuid.uuid4().hex[:6]}", graph=graph(canvas), + status="published" if published else "draft", published_version=1 if published else None) + session.add(wf) + await session.flush() + session.add(WorkflowVersion(workflow_id=wf.id, version=1, graph=graph(snapshot))) + await session.commit() + return wf.id + + +async def schedule(workflow_id: str, *, cron: str = "0 9 * * *", next_run_at: datetime | None = NINE, + run_class: str = "formal", grace: int = 3600, enabled: bool = True, name: str = "周一早报") -> str: + async with SessionLocal() as session: + row = Schedule(name=name, workflow_id=workflow_id, cron=cron, timezone="Asia/Shanghai", run_class=run_class, + input_template={"stat_date": "{{ schedule.yesterday }}"}, enabled=enabled, + misfire_grace_s=grace, next_run_at=next_run_at) + session.add(row) + await session.commit() + return row.id + + +async def fires(schedule_id: str) -> list[ScheduleFire]: + async with SessionLocal() as session: + return list((await session.execute( + select(ScheduleFire).where(ScheduleFire.schedule_id == schedule_id).order_by(ScheduleFire.created_at) + )).scalars()) + + +async def runs_of(schedule_id: str) -> list[Run]: + async with SessionLocal() as session: + return list((await session.execute(select(Run).where(Run.schedule_id == schedule_id))).scalars()) + + +async def next_of(schedule_id: str) -> datetime | None: + async with SessionLocal() as session: + return (await session.get(Schedule, schedule_id)).next_run_at + + +async def settle(run_id: str) -> Run: + for _ in range(400): + async with SessionLocal() as session: + row = await session.get(Run, run_id) + if row.status in ("succeeded", "failed", "cancelled") and row.manifest_seq is not None: + await run_manager.wait_idle(run_id) + return row + await asyncio.sleep(0.03) + raise AssertionError(f"运行 {run_id} 停在 {row.status}") + + +# -------------------------------------------------------------------------- +# 到点发起 +# -------------------------------------------------------------------------- + + +async def test_due_schedule_launches_once_and_moves_on(): + sid = await schedule(await workflow()) + await loop.fire_due(NINE + timedelta(seconds=10)) + [fire] = await fires(sid) + assert (fire.kind, fire.status, fire.fire_at, fire.reason) == ("scheduled", "started", NINE, None) + assert fire.inputs == {"stat_date": "2026-10-04"} + run = await settle(fire.run_id) + assert run.schedule_id == sid and run.run_class == "formal" and run.status == "succeeded" + assert run.started_by == "定时任务：周一早报" + assert run.input == {"stat_date": "2026-10-04"} + # 正式运行用的是发布版本，不是画布 + assert run.version == 1 and run.output == {"r": "发布版 2026-10-04"} + assert await next_of(sid) == NINE + timedelta(days=1) + # 同一拍再来一次：已经推进到明天，不再发起 + await loop.fire_due(NINE + timedelta(seconds=20)) + assert len(await fires(sid)) == 1 and len(await runs_of(sid)) == 1 + + +async def test_exploratory_schedule_uses_the_canvas(): + sid = await schedule(await workflow(published=False), run_class="exploratory") + await loop.fire_due(NINE + timedelta(seconds=5)) + [fire] = await fires(sid) + run = await settle(fire.run_id) + assert run.run_class == "exploratory" and run.output == {"r": "画布 2026-10-04"} + + +async def test_two_concurrent_ticks_launch_only_once(): + sid = await schedule(await workflow()) + now = NINE + timedelta(seconds=3) + await asyncio.gather(loop.fire_due(now), loop.fire_due(now), loop.fire_due(now)) + assert len(await fires(sid)) == 1 + [run] = await runs_of(sid) + await settle(run.id) + + +async def test_concurrent_ticks_with_different_clocks_claim_once(): + """一拍准点、一拍晚了两分钟（会被当成补发、kind 不同、撞不上唯一约束）：只有认领到的那一拍处理""" + sid = await schedule(await workflow()) + await asyncio.gather(loop.fire_due(NINE + timedelta(seconds=3)), loop.fire_due(NINE + timedelta(minutes=2))) + [fire] = await fires(sid) + [run] = await runs_of(sid) + assert fire.run_id == run.id + await settle(run.id) + + +async def test_a_fire_already_recorded_is_not_launched_again(): + """认领之外的第二道保险：同一 (任务, fire_at, kind) 已有记录，fire_once 交回那一条，不再发起""" + sid = await schedule(await workflow()) + first = await service.fire_once(sid, NINE, "scheduled") + again = await service.fire_once(sid, NINE, "scheduled") + assert again.id == first.id and len(await fires(sid)) == 1 + [run] = await runs_of(sid) + await settle(run.id) + + +async def test_disabled_schedule_is_not_fired(): + sid = await schedule(await workflow(), enabled=False) + await loop.fire_due(NINE + timedelta(seconds=5)) + assert await fires(sid) == [] and await runs_of(sid) == [] + + +@pytest.mark.parametrize("status", ["queued", "running", "interrupted"]) +async def test_previous_run_still_going_means_skip(status): + wf = await workflow() + sid = await schedule(wf) + async with SessionLocal() as session: + busy = Run(workflow_id=wf, workflow_name="门店周报", status=status, schedule_id=sid, run_class="formal") + session.add(busy) + await session.commit() + await loop.fire_due(NINE + timedelta(seconds=5)) + [fire] = await fires(sid) + assert (fire.status, fire.run_id) == ("skipped_overlap", None) + assert busy.id[:8] in fire.reason and "跳过" in fire.reason + assert [r.id for r in await runs_of(sid)] == [busy.id] + assert await next_of(sid) == NINE + timedelta(days=1) + + +# -------------------------------------------------------------------------- +# 错过的触发 +# -------------------------------------------------------------------------- + + +async def test_missed_within_grace_is_caught_up_once_with_inputs_of_its_own_day(): + """服务停到 00:30（上海 10 月 6 日）才起来：补一次 9 点那一次，昨天还是 10 月 4 日""" + sid = await schedule(await workflow(), grace=86400) + late = NINE + timedelta(hours=15, minutes=30) + await loop.fire_due(late) + [fire] = await fires(sid) + assert (fire.kind, fire.status, fire.fire_at) == ("misfire", "started", NINE) + assert fire.inputs == {"stat_date": "2026-10-04"} + await settle(fire.run_id) + assert await next_of(sid) == NINE + timedelta(days=1) + + +async def test_backlog_is_skipped_and_only_the_latest_is_caught_up(): + """每小时一次，停了三个多小时：01–03 点的积压记一条跳过，04 点那次在宽限内补发""" + sid = await schedule(await workflow(), cron="0 * * * *", grace=3600) + await loop.fire_due(NINE + timedelta(hours=3, minutes=20)) + skipped, caught_up = await fires(sid) + assert (skipped.kind, skipped.status, skipped.fire_at, skipped.run_id) == ("misfire", "skipped_misfire", NINE, None) + assert "2026-10-05 09:00" in skipped.reason and "2026-10-05 11:00" in skipped.reason + assert (caught_up.kind, caught_up.status, caught_up.fire_at) == ("misfire", "started", NINE + timedelta(hours=3)) + assert len(await runs_of(sid)) == 1 + await settle(caught_up.run_id) + assert await next_of(sid) == NINE + timedelta(hours=4) + + +async def test_missed_beyond_grace_is_skipped_without_launching(): + sid = await schedule(await workflow(), grace=600) + await loop.fire_due(NINE + timedelta(hours=2)) + [fire] = await fires(sid) + assert (fire.kind, fire.status, fire.fire_at, fire.run_id) == ("misfire", "skipped_misfire", NINE, None) + assert "10 分钟" in fire.reason and "2026-10-05 09:00" in fire.reason + assert await runs_of(sid) == [] + assert await next_of(sid) == NINE + timedelta(days=1) + + +async def test_days_of_downtime_leave_one_skip_record(): + sid = await schedule(await workflow(), grace=600, next_run_at=NINE - timedelta(days=4)) + await loop.fire_due(NINE + timedelta(hours=2)) + [fire] = await fires(sid) + assert (fire.status, fire.fire_at) == ("skipped_misfire", NINE) + assert "2026-10-01 09:00" in fire.reason and "2026-10-05 09:00" in fire.reason + assert await runs_of(sid) == [] and await next_of(sid) == NINE + timedelta(days=1) + + +async def test_on_time_fire_ignores_a_small_grace(): + """宽限写 0 也不影响准点的触发：拍与拍之间的十几秒不算错过""" + sid = await schedule(await workflow(), grace=0) + await loop.fire_due(NINE + timedelta(seconds=14)) + [fire] = await fires(sid) + assert (fire.kind, fire.status) == ("scheduled", "started") + await settle(fire.run_id) + + +# -------------------------------------------------------------------------- +# 发起被拒、出错 +# -------------------------------------------------------------------------- + + +async def test_formal_schedule_without_published_version_records_the_refusal(): + wf = await workflow(published=False) + sid = await schedule(wf) + await loop.fire_due(NINE + timedelta(seconds=5)) + [fire] = await fires(sid) + async with SessionLocal() as session: + name = (await session.get(Workflow, wf)).name + assert (fire.status, fire.run_id) == ("failed_to_start", None) + assert fire.reason == f"「{name}」还没有发布版本，请先在编排页发布" + assert await runs_of(sid) == [] + assert await next_of(sid) == NINE + timedelta(days=1) + + +async def test_deleted_workflow_records_the_refusal(): + sid = await schedule(f"gone{uuid.uuid4().hex[:8]}") + await loop.fire_due(NINE + timedelta(seconds=5)) + [fire] = await fires(sid) + assert (fire.status, fire.reason) == ("failed_to_start", "工作流不存在，可能已被删除") + + +async def test_unexpected_errors_record_a_generic_reason(monkeypatch, caplog): + from app.engine import launch + + async def explode(*args, **kwargs): + raise RuntimeError("库连接断开：sqlite busy") + + monkeypatch.setattr(launch, "launch_run", explode) + sid = await schedule(await workflow()) + with caplog.at_level(logging.ERROR): + await loop.fire_due(NINE + timedelta(seconds=5)) + [fire] = await fires(sid) + assert (fire.status, fire.reason) == ("failed_to_start", service.GENERIC_REASON) + assert "库连接断开" in caplog.text + assert await next_of(sid) == NINE + timedelta(days=1) + + +async def test_broken_cron_in_the_database_stops_the_schedule(): + sid = await schedule(await workflow(), cron="0 25 * * *") + await loop.fire_due(NINE + timedelta(seconds=5)) + [fire] = await fires(sid) + assert fire.status == "failed_to_start" and "「时」" in fire.reason + assert await next_of(sid) is None + + +# -------------------------------------------------------------------------- +# start / stop +# -------------------------------------------------------------------------- + + +async def test_start_does_nothing_when_switched_off(): + """测试环境 AGENTLAB_SCHEDULER=0（conftest）：start 不起循环，stop 照样能调""" + assert await scheduler.start() is False + assert scheduler.running() is False + await scheduler.stop() + + +async def test_loop_ticks_until_stopped(): + sid = await schedule(await workflow(), next_run_at=datetime.now(UTC) - timedelta(seconds=1), cron="* * * * *") + on = Settings(_env_file=None, scheduler=True) + assert await scheduler.start(interval=0.05, config=on) is True + assert scheduler.running() + for _ in range(200): + if await fires(sid): + break + await asyncio.sleep(0.02) + await scheduler.stop() + assert scheduler.running() is False + [fire] = await fires(sid) + assert fire.status == "started" + await settle(fire.run_id) + # 停下之后不再有拍：把下一次拨回过去也不会再发起 + 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 + + +async def test_start_twice_keeps_one_loop(): + on = Settings(_env_file=None, scheduler=True) + await scheduler.start(interval=0.05, config=on) + first = loop.runner.task + await scheduler.start(interval=0.05, config=on) + assert loop.runner.task is first + await scheduler.stop() + assert first.done() diff --git a/backend/tests/test_scheduler_webhook.py b/backend/tests/test_scheduler_webhook.py new file mode 100644 index 00000000..2a90f8b4 --- /dev/null +++ b/backend/tests/test_scheduler_webhook.py @@ -0,0 +1,399 @@ +"""定时任务的 Webhook（app/scheduler/webhook.py）：运行结束、失败、停在审批时推送一次。 + +- 四种格式的请求体和签名：通用 JSON（配了密钥时 X-AgentLab-Signature = HMAC-SHA256(密钥, 时间戳.请求体)）、 + 飞书文本（timestamp + sign，sign 是以「时间戳\\n密钥」为键的 HMAC-SHA256）、企业微信 markdown（key 在地址里）、 + 钉钉 markdown（地址上加 timestamp 和 sign） +- 通知里有任务名、状态、出具档位、缺口条数、运行链接（对外访问地址 + /runs/{id}）、封存哈希前 12 位、开始结束 + 时间；不放报告里的具体数字，不放失败的原始报错 +- 每一拍查已发起的触发：运行状态变了、任务订阅了这个事件、这个状态还没处理过才推送；停在审批只在确有待审批时 + 算；结束的运行等封存好再推；没订阅、没配 Webhook 的也记下已处理，之后再配上不会补发旧的 +- 失败重试 3 次（4xx 不重试），每次推送记一条 WebhookDelivery；日志和错误里只有主机名，没有地址、key、 + access_token、密钥（httpx 自己的请求日志也一样） + +用 httpx.MockTransport 接住请求，不连外网。任务名、地址、密钥、数字都是虚构的。 +""" +from __future__ import annotations + +import base64 +import hashlib +import hmac +import json +import logging +import uuid +from datetime import datetime, timedelta, timezone +from urllib.parse import parse_qs, quote_plus, urlsplit + +import httpx +import pytest +from sqlalchemy import delete, select + +from app.core.crypto import encrypt +from app.db.base import SessionLocal +from app.db.models import Approval, Run, Schedule, ScheduleFire, Setting, WebhookDelivery +from app.scheduler import webhook + +UTC = timezone.utc +NOW = datetime(2026, 10, 5, 1, 2, 3, tzinfo=UTC) +FEISHU = "https://open.feishu.cn/open-apis/bot/v2/hook/0f1e2d3c-aaaa-bbbb-cccc-9d8e7f6a5b4c" +WECOM = "https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=6a7b8c9d-1111-2222-3333-444455556666" +DINGTALK = "https://oapi.dingtalk.com/robot/send?access_token=f00dfacecafebabe0123456789abcdef" +GENERIC = "https://hooks.example.com/agentlab/inbox-7f3a" +SECRET = "SECdemo-not-a-real-secret-42" + + +class Recorder: + """MockTransport 的处理函数：记下每个请求，按 replies 依次回应（用完了就一直回最后一个）。""" + + def __init__(self, *replies): + self.requests: list[httpx.Request] = [] + self.replies = list(replies) or [httpx.Response(200, json={"code": 0, "errcode": 0})] + + def __call__(self, request: httpx.Request) -> httpx.Response: + self.requests.append(request) + reply = self.replies[min(len(self.requests), len(self.replies)) - 1] + if isinstance(reply, Exception): + raise reply + return reply + + +@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="finished", 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="r0123456789abcdef", status="succeeded", tier="degraded", gaps=2, + run_url="https://agentlab.example.com/runs/r0123456789abcdef", + manifest_hash="a1b2c3d4e5f60718293a", started_at=datetime(2026, 10, 5, 1, 0, 3, tzinfo=UTC), + finished_at=datetime(2026, 10, 5, 1, 1, 10, tzinfo=UTC), reason=None, pending_approvals=None) + base.update(over) + return webhook.Notice(**base) + + +# -------------------------------------------------------------------------- +# 请求体和签名 +# -------------------------------------------------------------------------- + + +def test_generic_json_carries_the_agreed_fields(): + url, headers, body = webhook.build_request("generic", GENERIC, None, notice(), now=NOW) + data = json.loads(body) + assert url == GENERIC and headers["Content-Type"] == "application/json" + assert {k: data[k] for k in ("event", "schedule_name", "run_id", "status", "tier", "tier_label", "gaps", + "run_url", "manifest_hash")} == { + "event": "finished", "schedule_name": "门店周报", "run_id": "r0123456789abcdef", "status": "succeeded", + "tier": "degraded", "tier_label": "降档出具", "gaps": 2, + "run_url": "https://agentlab.example.com/runs/r0123456789abcdef", "manifest_hash": "a1b2c3d4e5f6"} + assert data["started_at"] == "2026-10-05T01:00:03+00:00" and data["finished_at"] == "2026-10-05T01:01:10+00:00" + assert "X-AgentLab-Signature" not in headers + + +def test_generic_signature_when_a_secret_is_set(): + _, headers, body = webhook.build_request("generic", GENERIC, SECRET, notice(), now=NOW) + ts = headers["X-AgentLab-Timestamp"] + assert ts == str(int(NOW.timestamp())) + expected = hmac.new(SECRET.encode(), f"{ts}.".encode() + body, hashlib.sha256).hexdigest() + assert headers["X-AgentLab-Signature"] == f"sha256={expected}" + + +def test_feishu_text_and_signature(): + url, _, body = webhook.build_request("feishu", FEISHU, SECRET, notice(), now=NOW) + data = json.loads(body) + assert url == FEISHU and data["msg_type"] == "text" + ts = data["timestamp"] + assert ts == str(int(NOW.timestamp())) + # 飞书的签名：以「timestamp + 换行 + 密钥」为键，对空串做 HMAC-SHA256，再 base64 + key = f"{ts}\n{SECRET}".encode() + assert data["sign"] == base64.b64encode(hmac.new(key, b"", hashlib.sha256).digest()).decode() + text = data["content"]["text"] + for fragment in ("门店周报", "降档出具", "缺口 2 条", "a1b2c3d4e5f6", "/runs/r0123456789abcdef", + "2026-10-05 09:00", "2026-10-05 09:00:03", "2026-10-05 09:01:10"): + assert fragment in text, text + + +def test_feishu_without_secret_has_no_sign(): + _, _, body = webhook.build_request("feishu", FEISHU, None, notice(), now=NOW) + assert "sign" not in json.loads(body) and "timestamp" not in json.loads(body) + + +def test_wecom_markdown_keeps_the_key_in_the_address(): + url, _, body = webhook.build_request("wecom", WECOM, None, notice(), now=NOW) + data = json.loads(body) + assert url == WECOM and data["msgtype"] == "markdown" + assert "门店周报" in data["markdown"]["content"] and "降档出具" in data["markdown"]["content"] + + +def test_dingtalk_markdown_and_signed_address(): + url, _, body = webhook.build_request("dingtalk", DINGTALK, SECRET, notice(), now=NOW) + data = json.loads(body) + assert data["msgtype"] == "markdown" and "门店周报" in data["markdown"]["title"] + assert "降档出具" in data["markdown"]["text"] + query = parse_qs(urlsplit(url).query) + ts = str(int(NOW.timestamp() * 1000)) + assert query["access_token"] == ["f00dfacecafebabe0123456789abcdef"] and query["timestamp"] == [ts] + sign = base64.b64encode(hmac.new(SECRET.encode(), f"{ts}\n{SECRET}".encode(), hashlib.sha256).digest()).decode() + assert query["sign"] == [sign] + assert f"sign={quote_plus(sign)}" in url + + +def test_awaiting_approval_and_failed_to_start_texts(): + waiting = webhook.message(notice(event="awaiting_approval", status="interrupted", tier=None, gaps=None, + pending_approvals=2, finished_at=None)) + assert "等待审批" in waiting[0] and any("2 项" in value for _, value in waiting[1]) + refused = webhook.message(notice(event="failed", status="failed_to_start", run_id=None, run_url=None, + tier=None, gaps=None, manifest_hash=None, started_at=None, finished_at=None, + reason="「门店周报」还没有发布版本，请先在编排页发布")) + assert "未能发起" in refused[0] and any("还没有发布版本" in value for _, value in refused[1]) + + +def test_url_hint_shows_only_host_and_last_four(): + assert webhook.url_hint(FEISHU) == "open.feishu.cn/…5b4c" + assert webhook.url_hint(WECOM) == "qyapi.weixin.qq.com/…6666" + assert webhook.url_hint("http://127.0.0.1:8794/hook") == "127.0.0.1:8794/…hook" + + +@pytest.mark.parametrize("url", ["", "open.feishu.cn/hook", "ftp://x.example.com/a", "https://", "https://a b.com/x"]) +def test_bad_addresses_are_refused(url): + with pytest.raises(ValueError) as caught: + webhook.check_url(url) + assert "Webhook 地址" in str(caught.value) + + +# -------------------------------------------------------------------------- +# 推送与重试 +# -------------------------------------------------------------------------- + + +async def test_success_on_first_try(recorder): + outcome = await webhook.send("feishu", FEISHU, SECRET, notice()) + assert (outcome.ok, outcome.http_status, outcome.attempts, outcome.error) == (True, 200, 1, None) + assert len(recorder.requests) == 1 and str(recorder.requests[0].url) == FEISHU + + +async def test_server_errors_are_retried(recorder): + recorder.replies = [httpx.Response(502), httpx.ReadTimeout("timed out"), httpx.Response(200, json={"code": 0})] + outcome = await webhook.send("feishu", FEISHU, None, notice()) + assert (outcome.ok, outcome.attempts) == (True, 3) + + +async def test_gives_up_after_three_retries(recorder): + recorder.replies = [httpx.Response(500)] + outcome = await webhook.send("generic", GENERIC, None, notice()) + assert (outcome.ok, outcome.http_status, outcome.attempts) == (False, 500, 4) + assert "HTTP 500" in outcome.error + + +async def test_client_errors_are_not_retried(recorder): + recorder.replies = [httpx.Response(404)] + outcome = await webhook.send("generic", GENERIC, None, notice()) + assert (outcome.ok, outcome.http_status, outcome.attempts) == (False, 404, 1) + + +async def test_platform_error_code_counts_as_failure(recorder): + recorder.replies = [httpx.Response(200, json={"code": 19021, "msg": "sign match fail or timestamp is not within one hour"})] + outcome = await webhook.send("feishu", FEISHU, SECRET, notice()) + assert not outcome.ok and outcome.error.startswith("open.feishu.cn 返回错误码 19021：sign match fail") + recorder.replies = [httpx.Response(200, json={"errcode": 310000, "errmsg": "sign not match"})] + assert "310000" in (await webhook.send("dingtalk", DINGTALK, SECRET, notice())).error + + +async def test_connection_errors_name_only_the_host(recorder): + recorder.replies = [httpx.ConnectError(f"cannot reach {WECOM}")] + outcome = await webhook.send("wecom", WECOM, None, notice()) + assert not outcome.ok and outcome.attempts == 4 + assert "qyapi.weixin.qq.com" in outcome.error and "6a7b8c9d" not in outcome.error + + +async def test_logs_never_contain_the_address_key_or_secret(recorder, caplog): + caplog.set_level(logging.DEBUG) + recorder.replies = [httpx.Response(500, json={"errmsg": f"bad token {DINGTALK}"}), httpx.Response(200, json={"errcode": 0})] + await webhook.send("dingtalk", DINGTALK, SECRET, notice()) + await webhook.send("feishu", FEISHU, SECRET, notice()) + text = caplog.text + assert "oapi.dingtalk.com" in text and "open.feishu.cn" in text # 主机名照常记 + for leaked in (DINGTALK, FEISHU, "f00dfacecafebabe0123456789abcdef", "0f1e2d3c-aaaa", SECRET, "access_token="): + assert leaked not in text, leaked + + +# -------------------------------------------------------------------------- +# 每一拍的通知扫描 +# -------------------------------------------------------------------------- + + +async def make(*, run_status: str | None = "succeeded", events=("finished", "failed", "awaiting_approval"), + fmt: str = "generic", url: str | None = GENERIC, secret: str | None = None, sealed: bool = True, + fire_status: str = "started", output: dict | None = None, reason: str | None = None): + """建一个配了 Webhook 的任务、一次触发和它发起的运行（运行不真的执行，状态直接写库）。""" + async with SessionLocal() as session: + sched = Schedule(name="门店周报", workflow_id="wf-demo", cron="0 9 * * 1", timezone="Asia/Shanghai", + webhook_format=fmt if url else None, webhook_url=encrypt(url) if url else None, + webhook_secret=encrypt(secret) if secret else None, webhook_events=list(events)) + session.add(sched) + await session.flush() + run = None + if run_status: + now = datetime.now(UTC) + run = Run(workflow_id=None, workflow_name="门店周报", status=run_status, schedule_id=sched.id, + run_class="formal", started_at=now - timedelta(minutes=1), + finished_at=now if run_status in ("succeeded", "failed", "cancelled") else None, + manifest_hash=("c0ffee" * 10) if sealed else None, + output=output if output is not None else { + "report": "本周销售额 98765.43 元，同比增长 12.5%", + "_issuance": {"tier": "formal", "gaps": []}}) + session.add(run) + await session.flush() + fire = ScheduleFire(schedule_id=sched.id, fire_at=datetime.now(UTC).replace(microsecond=0), + kind="scheduled", status=fire_status, run_id=run.id if run else None, reason=reason) + session.add(fire) + await session.commit() + return sched.id, fire.id, run.id if run else None + + +async def fire_row(fire_id: str) -> ScheduleFire: + async with SessionLocal() as session: + return await session.get(ScheduleFire, fire_id) + + +async def deliveries(fire_id: str) -> list[WebhookDelivery]: + async with SessionLocal() as session: + return list((await session.execute( + select(WebhookDelivery).where(WebhookDelivery.fire_id == fire_id).order_by(WebhookDelivery.created_at) + )).scalars()) + + +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 test_finished_run_is_pushed_once(recorder): + async with SessionLocal() as session: + session.add(Setting(key="notify", value={"public_base_url": "https://agentlab.example.com"})) + await session.commit() + _, fire_id, run_id = await make() + assert await webhook.notify_due() == 1 + [req] = recorder.requests + data = json.loads(req.content) + assert (data["event"], data["status"], data["tier"], data["gaps"]) == ("finished", "succeeded", "formal", 0) + assert data["run_url"] == f"https://agentlab.example.com/runs/{run_id}" + assert data["manifest_hash"] == ("c0ffee" * 2) + # 报告里的数字不进通知 + assert "98765.43" not in req.content.decode() and "12.5" not in req.content.decode() + [delivery] = await deliveries(fire_id) + assert (delivery.event, delivery.status, delivery.http_status, delivery.attempts, delivery.run_id) == ( + "finished", "ok", 200, 1, run_id) + assert (await fire_row(fire_id)).notified_status == "succeeded" + # 下一拍：同一状态不再推送 + assert await webhook.notify_due() == 0 + assert len(recorder.requests) == 1 + + +async def test_no_public_address_means_no_link(recorder): + await make() + await webhook.notify_due() + assert json.loads(recorder.requests[0].content)["run_url"] is None + + +async def test_failed_run_and_failed_to_start(recorder): + _, failed_fire, _ = await make(run_status="failed", output={}) + _, refused_fire, _ = await make(run_status=None, fire_status="failed_to_start", + reason="「门店周报」还没有发布版本，请先在编排页发布") + assert await webhook.notify_due() == 2 + bodies = sorted((json.loads(r.content) for r in recorder.requests), key=lambda b: b["status"]) + assert [(b["event"], b["status"]) for b in bodies] == [("failed", "failed"), ("failed", "failed_to_start")] + assert bodies[1]["run_id"] is None and "还没有发布版本" in bodies[1]["reason"] + assert (await fire_row(failed_fire)).notified_status == "failed" + assert (await fire_row(refused_fire)).notified_status == "failed_to_start" + + +async def test_unsubscribed_event_is_marked_but_not_pushed(recorder): + _, fire_id, _ = await make(events=("failed",)) + assert await webhook.notify_due() == 0 + assert recorder.requests == [] and await deliveries(fire_id) == [] + assert (await fire_row(fire_id)).notified_status == "succeeded" + + +async def test_without_webhook_old_fires_are_not_pushed_later(recorder): + sid, fire_id, _ = await make(url=None) + await webhook.notify_due() + assert (await fire_row(fire_id)).notified_status == "succeeded" + async with SessionLocal() as session: + await session.execute(Schedule.__table__.update().where(Schedule.id == sid).values( + webhook_format="generic", webhook_url=encrypt(GENERIC), webhook_events=["finished"])) + await session.commit() + assert await webhook.notify_due() == 0 and recorder.requests == [] + + +async def test_cancelled_run_is_not_pushed(recorder): + _, fire_id, _ = await make(run_status="cancelled") + await webhook.notify_due() + assert recorder.requests == [] and (await fire_row(fire_id)).notified_status == "cancelled" + + +async def test_waits_for_the_seal_before_pushing(recorder): + _, fire_id, run_id = await make(sealed=False) + assert await webhook.notify_due() == 0 and (await fire_row(fire_id)).notified_status is None + await set_run(run_id, manifest_hash="beef" * 16) + assert await webhook.notify_due() == 1 + assert json.loads(recorder.requests[0].content)["manifest_hash"] == "beef" * 3 + + +async def test_awaiting_approval_only_with_a_real_pending_approval(recorder): + """服务重启挂起的运行也是 interrupted，但没有待审批：不推。有了待审批推一次，批完跑完再推完成""" + _, fire_id, run_id = await make(run_status="interrupted") + assert await webhook.notify_due() == 0 and (await fire_row(fire_id)).notified_status is None + async with SessionLocal() as session: + session.add(Approval(run_id=run_id, node_id="gate", title="审核周报", status="pending")) + await session.commit() + assert await webhook.notify_due() == 1 + data = json.loads(recorder.requests[0].content) + assert (data["event"], data["status"], data["pending_approvals"]) == ("awaiting_approval", "interrupted", 1) + assert await webhook.notify_due() == 0 + await set_run(run_id, status="succeeded", finished_at=datetime.now(UTC), manifest_hash="ab" * 32) + assert await webhook.notify_due() == 1 + assert [json.loads(r.content)["event"] for r in recorder.requests] == ["awaiting_approval", "finished"] + assert [d.event for d in await deliveries(fire_id)] == ["awaiting_approval", "finished"] + + +async def test_failed_delivery_is_recorded_without_secrets(recorder): + recorder.replies = [httpx.Response(503, text=f"upstream {DINGTALK} down")] + _, fire_id, _ = await make(fmt="dingtalk", url=DINGTALK, secret=SECRET) + await webhook.notify_due() + [delivery] = await deliveries(fire_id) + assert (delivery.status, delivery.http_status, delivery.attempts) == ("failed", 503, 4) + assert "HTTP 503" in delivery.error + for leaked in (DINGTALK, "f00dfacecafebabe", SECRET): + assert leaked not in delivery.error + # 推送失败也算这个状态处理过了：不会每一拍都再试一遍 + assert (await fire_row(fire_id)).notified_status == "succeeded" + + +async def test_test_message(recorder): + sid, _, _ = await make(fmt="feishu", url=FEISHU, secret=SECRET, run_status=None, fire_status="skipped_overlap") + result = await webhook.send_test(sid) + assert result == {"ok": True, "http_status": 200, "message": "测试消息已送达（HTTP 200）"} + data = json.loads(recorder.requests[0].content) + assert "测试" in data["content"]["text"] and "门店周报" in data["content"]["text"] and data["sign"] + async with SessionLocal() as session: + [row] = (await session.execute(select(WebhookDelivery).where(WebhookDelivery.schedule_id == sid))).scalars() + assert (row.event, row.status, row.fire_id, row.attempts) == ("test", "ok", None, 1) + recorder.replies = [httpx.ConnectError("refused")] + failed = await webhook.send_test(sid) + assert failed["ok"] is False and failed["http_status"] is None and "open.feishu.cn" in failed["message"] + + diff --git a/backend/tests/test_settings_notify.py b/backend/tests/test_settings_notify.py new file mode 100644 index 00000000..5e3d4312 --- /dev/null +++ b/backend/tests/test_settings_notify.py @@ -0,0 +1,77 @@ +"""通知设置（api/settings.py 的 notify 一组）：定时任务的 Webhook 里拼运行链接用的对外访问地址。 + +- 默认没填（null），通知里就不附运行链接 +- 只收 http(s) 开头的完整地址；末尾的 / 去掉、首尾空白去掉，空串当没填；带 ? 或 # 的拒收 +- 写错整个请求回 422、detail 是中文，一项都不落库 +""" +from __future__ import annotations + +import pytest +from httpx import ASGITransport, AsyncClient +from sqlalchemy import delete + +from app.api.settings import NOTIFY_GROUP, notify_settings, notify_values +from app.db.base import SessionLocal +from app.db.models import Setting +from app.main import app + + +@pytest.fixture(autouse=True) +async def clean(): + async with SessionLocal() as session: + await session.execute(delete(Setting).where(Setting.key.in_([NOTIFY_GROUP, "run"]))) + await session.commit() + yield + + +def client() -> AsyncClient: + return AsyncClient(transport=ASGITransport(app=app), base_url="http://test") + + +async def test_default_is_no_link(): + async with client() as c: + body = (await c.get("/api/settings")).json() + assert body["notify"] == {"public_base_url": None} + async with SessionLocal() as session: + assert await notify_settings(session) == {"public_base_url": None} + + +@pytest.mark.parametrize("given, stored", [ + ("https://agentlab.example.com/", "https://agentlab.example.com"), + (" http://10.20.30.40:8080 ", "http://10.20.30.40:8080"), + ("https://intra.example.com/agentlab/", "https://intra.example.com/agentlab"), + ("", None), + (None, None), +]) +async def test_saved_and_normalized(given, stored): + async with client() as c: + put = await c.put("/api/settings", json={"values": {NOTIFY_GROUP: {"public_base_url": given}}}) + assert put.status_code == 200, put.text + assert put.json()["notify"] == {"public_base_url": stored} + assert (await c.get("/api/settings")).json()["notify"] == {"public_base_url": stored} + async with SessionLocal() as session: + assert await notify_settings(session) == {"public_base_url": stored} + + +@pytest.mark.parametrize("value, fragment", [ + ({"public_base_url": "agentlab.example.com"}, "http:// 或 https://"), + ({"public_base_url": "ftp://files.example.com"}, "http:// 或 https://"), + ({"public_base_url": "https://"}, "http:// 或 https://"), + ({"public_base_url": 8080}, "http:// 或 https://"), + ({"public_base_url": "https://agentlab.example.com/?from=bot"}, "? 或 #"), + ({"public_url": "https://agentlab.example.com"}, "不支持「public_url」"), + ("https://agentlab.example.com", "格式有误"), +]) +async def test_invalid_values_are_refused_in_chinese(value, fragment): + async with client() as c: + put = await c.put("/api/settings", json={"values": {NOTIFY_GROUP: value, "run": {"stream_tokens": False}}}) + assert put.status_code == 422 + assert fragment in put.json()["detail"] + assert "对外访问地址" in put.json()["detail"] or "通知" in put.json()["detail"] + # 先查后写：同一请求里别的组也没有落库 + assert (await c.get("/api/settings")).json()["run"]["stream_tokens"] is True + + +def test_broken_stored_value_reads_as_unset(): + assert notify_values({"public_base_url": "不是地址"}) == {"public_base_url": None} + assert notify_values("坏掉的") == {"public_base_url": None}