diff --git a/backend/app/api/schedules.py b/backend/app/api/schedules.py
index a78bdf49..fe66843e 100644
--- a/backend/app/api/schedules.py
+++ b/backend/app/api/schedules.py
@@ -439,8 +439,9 @@ async def _fire_outs(session: AsyncSession, fires: list[ScheduleFire]) -> list[d
raw = f.inputs.get("event") if isinstance(f.inputs, dict) and f.schedule_id not in maintenance else None
if not isinstance(raw, dict):
return None
+ # windows / periods 只在几个过了宽限的窗口合成一条跳过时有：合了几个窗口、合掉的期次（上传表格）
return {key: raw.get(key) for key in ("event", "subject_id", "subject_name", "text", "merged", "first_at",
- "last_at", "depth", "sample", "real")}
+ "last_at", "depth", "sample", "real", "windows", "periods")}
compared = {f.id: f.compare_artifact for f in fires if f.compare_artifact and f.run_id}
compares = await auto_compare.briefs(list(compared.values()))
diff --git a/backend/app/db/models_triggers.py b/backend/app/db/models_triggers.py
index 399687d3..368eb36b 100644
--- a/backend/app/db/models_triggers.py
+++ b/backend/app/db/models_triggers.py
@@ -5,13 +5,20 @@
**为什么要一张「待派发事件」表，而不是在发生处直接发起运行。** 事件发生在别的事务里：启用上传表格的新版本、正式运行的
收尾、记下数据源的变化。在那里直接发起运行，发起失败要不要回滚那次启用？重复发生、服务重启时发起到一半又算什么？说不清。
-改成在发生处、和那件事**同一个事务**里写一行 TriggerEvent（发生了就一定有这一行，没发生就一定没有），调度循环一拍取走、
-按各定时任务的触发条件派发：发起运行的规矩（重叠跳过、正式运行只从发布版本发起、触发记录唯一）全部沿用定时任务那一套。
+改成在发生处写一行 TriggerEvent，调度循环一拍取走、按各定时任务的触发条件派发：发起运行的规矩（重叠跳过、正式运行只从
+发布版本发起、触发记录唯一）全部沿用定时任务那一套。
+
+什么时候和那件事同一个事务：table.version（切指针那一下）、workflow.finished（写终态那一下）是同一个事务——发生了就一定有
+这一行，没发生就一定没有。source.changed 不是：一次探查、剖析的变化分好几批落库，「这一批只触发一次」要等任务结束才知道，
+事件在任务结束后另开会话写（data/source_changes.notify_after_job）。中间进程死了、库被锁住写不进，变化记录在、事件没有：
+调度的一拍按任务号补写（scheduler/outbox.backfill_source_changes），dedupe_key（source.changed:<任务号>）唯一，补写和正常的
+那一次不会记成两条。
调度关着（AGENTLAB_SCHEDULER）时没人取，事件留在表里，接口照实说有几条在等。第 2 波多进程时，取走靠的是条件 UPDATE，
多个进程也只有一个取得到。
- TriggerEvent 一次事件：种类（table.version / workflow.finished / source.changed）、对象 id、给模板用的事件内容、触发链
- 的深度和经过的运行、发生时刻。status：pending 还没派发；routed 已经按触发条件派给了 matched 个定时任务（0 个也算派完）。
+ 的深度和经过的运行、发生时刻、去重键（只有 source.changed 用）。status：pending 还没派发；routed 已经按触发条件派给了
+ matched 个定时任务（0 个也算派完）。
- TriggerDelivery 一个事件派给一个定时任务、等着去抖：同一个定时任务窗口内（第一条事件发生之后 debounce_s 秒）的几条
合成一次触发，fire_id 是那次触发记录；status：pending 等着、fired 合进了某次触发（发起、跳过、未能发起都算）、dropped
定时任务在等待期间被停用或改了触发条件，不再派发。(定时任务, 事件) 唯一：同一个事件不会派给同一个任务两次。
@@ -61,6 +68,9 @@ class TriggerEvent(Base):
routed_at: Mapped[datetime | None] = mapped_column(UTCDateTime, default=None)
#: 派给了几个定时任务
matched: Mapped[int] = mapped_column(Integer, default=0)
+ #: 去重键：同一件事只记一条。source.changed 是 source.changed:<任务号>（任务结束时记、没记上的补写，两边撞上唯一约束）；
+ #: 别的种类为空（它们和那件事同一个事务，不会记两遍）
+ dedupe_key: Mapped[str | None] = mapped_column(String(80), default=None, unique=True)
created_at: Mapped[datetime] = mapped_column(UTCDateTime, default=utcnow)
diff --git a/backend/app/scheduler/outbox.py b/backend/app/scheduler/outbox.py
index c61b70b7..366051a5 100644
--- a/backend/app/scheduler/outbox.py
+++ b/backend/app/scheduler/outbox.py
@@ -3,18 +3,24 @@
**写（发生处）。** 三处，各自只加一两行调用这里：
- 上传表格按期累积的新一期启用（data/table_versions 的 publish_build / publish_snapshot，切指针的同一个事务）：stage_table_version；
-- 正式运行成功收尾（engine/runner 的 _finalize，写终态的同一个事务）：stage_run_finished；
-- 探查、剖析任务结束、这一批新记了「要处理」的变化（data/source_changes.notify_after_job）：emit_source_changed，一批一条。
+- 正式运行成功收尾（engine/runner 的 _finalize，写终态的同一个事务）：stage_run_finished；途经驳回的（W1-b 的驳回出口）带上
+ rejected_via，缺省不认它的任务不派；
+- 探查、剖析任务结束、这一批新记了「要处理」的变化（data/source_changes.notify_after_job）：emit_source_changed，一个任务一条。
+ 这一处不和变化记录同一个事务（一批只触发一次要等任务结束才知道），中间丢了的由每一拍的 backfill_source_changes 按任务号补写，
+ 去重键唯一、不会记成两条。
发生处只写一行 TriggerEvent，不查有没有任务在等、不发起运行：事务边界、重复与重启都在这边说清。
**取（调度循环的一拍，dispatch_due）。**
1. 派：pending 的事件逐条按条件 UPDATE 认领成 routed（多个进程也只有一个认领得到），同一个事务里给每个合条件、启用着的
事件触发任务写一行 TriggerDelivery。没人合条件的也认领掉（matched=0）。
-2. 发：每个有等待中的 delivery 的任务，窗口从最早那一条的发生时刻起算 debounce_s 秒；窗口过了就把落在窗口里的几条合成一次
- 触发：内容取最新的那一条，fire_at 就是窗口的结束时刻（同一个窗口算出来的永远是同一个时刻，触发记录 (任务, fire_at, kind)
- 唯一，重复派发、重启后补派都不会发起两次）。上一次的运行还没结束照 overlap=skip 记 skipped_overlap；触发链超过上限记未能
- 发起；派发得太晚（调度停过、超出补发宽限）记 skipped_misfire。之后把这几条标 fired。任务停用了、改了触发条件，等着的标
- dropped，不再派发。
+2. 发：每个有等待中的 delivery 的任务，按发生时刻把它们切成去抖窗口：窗口从第一条的发生时刻起算 debounce_s 秒，落在里面的
+ 几条合成一次触发，内容取最新的那一条，fire_at 是窗口的结束时刻（同一个窗口算出来的永远是同一个时刻，触发记录
+ (任务, fire_at, kind) 唯一，重复派发、重启后补派都不会发起两次）。上传表格按期次分窗口（triggers.window_key）：9 月、10 月
+ 连着启用各有各的触发，同一期的重传照旧合并。一拍里把到点的窗口处理完（最多 MAX_WINDOWS 个），不是一拍一个：以前积压时
+ 一拍只走一个窗口，后面的越等越久、最后全过了宽限。到点的窗口里已经过了补发宽限的（排在前面）合成**一条** skipped_misfire，
+ 写合并了几次事件、几个窗口、时间跨度——派给任务就已经晚了（调度停过）才说「调度未运行」，否则照实说积压；之后宽限内的
+ 照常发起：上一次的运行还没结束照 overlap=skip 记 skipped_overlap，触发链超过上限记未能发起。处理过的标 fired。任务停用了、
+ 改了触发条件，等着的标 dropped，不再派发。
调度关着（AGENTLAB_SCHEDULER）时没有这一拍：事件留在表里，接口照实说有几条在等（pending_for）。
**留多久。** 派完的事件、合进触发的 delivery 留 RETAIN_DAYS 天给排查，之后由这一拍顺手清掉（每小时最多一次）。
@@ -45,6 +51,12 @@ CLEAN_EVERY = timedelta(hours=1)
CHAIN_KEEP = 10
#: 一次触发里记几条合并进来的事件 id
MERGED_KEEP = 20
+#: 一个任务一拍最多处理几个到点的去抖窗口、读几条等着的事件（积压特别多时分几拍处理完，不让一拍拖太久）
+MAX_WINDOWS = 100
+DRAIN_BATCH = 5000
+#: source.changed 往回补写几天内结束的任务；补写最多多久一次（刚起的服务第一拍就补）
+BACKFILL_DAYS = 2
+BACKFILL_EVERY = timedelta(minutes=10)
# --------------------------------------------------------------------------
@@ -76,9 +88,12 @@ async def stage_table_version(session: Any, *, source_id: str, source_name: str,
period_end=period_end, activated_at=now, replaced=replaced, seq=seq)))
-def stage_run_finished(session: Any, run: Run) -> None:
+def stage_run_finished(session: Any, run: Run, rejected_via: Any = None) -> None:
"""正式运行成功结束：往收尾的会话里加一条 workflow.finished（不提交，随终态一起提交）。探索运行、没挂在工作流上的不算。
- 这次运行自己是事件触发的，事件带上它的触发深度和经过的运行，下游据此防环。"""
+ 这次运行自己是事件触发的，事件带上它的触发深度和经过的运行，下游据此防环。
+
+ rejected_via：这次运行途经了哪几个关卡的驳回（W1-b 的驳回出口：按驳回那条路走完，状态仍是 succeeded；runner 收尾时把
+ _rejected_via(state) 交进来），缺省 None 是没途经驳回。照发，事件里带上它：缺省不认途经驳回的任务不派（triggers.matches）。"""
if run.status != "succeeded" or run.run_class != "formal" or not run.workflow_id:
return
upstream = (run.origin or {}).get("trigger") if run.origin_kind == "schedule" and isinstance(run.origin, dict) else None
@@ -87,29 +102,92 @@ def stage_run_finished(session: Any, run: Run) -> None:
chain = [str(x) for x in (upstream.get("chain") or []) if x][-(CHAIN_KEEP - 1):] + [run.id]
session.add(TriggerEvent(
event="workflow.finished", subject_id=run.workflow_id, subject_name=(run.workflow_name or "")[:200],
- payload=triggers.run_payload(run), depth=depth, chain=chain, occurred_at=run.finished_at or utcnow()))
+ payload=triggers.run_payload(run, rejected_via), depth=depth, chain=chain,
+ occurred_at=run.finished_at or utcnow()))
-async def emit_source_changed(source_id: str, job_id: str, new: int) -> None:
- """一次探查、剖析任务结束、新记了 new 条「要处理」的变化：记一条 source.changed（自己开会话提交）。同一个任务只记一条：
- 「同一次探查 / 剖析产生的一批只触发一次」。出错只记日志：变化记录本身已经落库。"""
+def source_key(job_id: str) -> str:
+ """source.changed 的去重键：一个任务一条。"""
+ return f"source.changed:{job_id}"
+
+
+async def emit_source_changed(source_id: str, job_id: str, new: int, *, occurred_at: datetime | None = None) -> bool:
+ """一次探查、剖析任务结束、新记了 new 条「要处理」的变化：记一条 source.changed（自己开会话提交），返回记没记下。同一个任务
+ 只记一条（去重键唯一）：「同一次探查 / 剖析产生的一批只触发一次」，补写（backfill_source_changes）和这里撞上也只有一条。
+ 出错只记日志：变化记录本身已经落库，没记上的由补写接着记。"""
+ from sqlalchemy.exc import IntegrityError
+
from app.data import source_changes
from app.db.models import DataSource
try:
async with SessionLocal() as session:
+ if await session.scalar(select(TriggerEvent.id).where(TriggerEvent.dedupe_key == source_key(job_id))):
+ return False
source = await session.get(DataSource, source_id)
got = await source_changes.counts(session, source_id)
by = got.get("by_severity") or {}
name = source.name if source is not None else ""
session.add(TriggerEvent(
- event="source.changed", subject_id=source_id, subject_name=name[:200],
+ event="source.changed", subject_id=source_id, subject_name=name[:200], dedupe_key=source_key(job_id),
+ occurred_at=occurred_at or utcnow(),
payload={"source": name, "source_id": source_id, "job_id": job_id, "counts": {
"new": int(new), "action": int(by.get("action") or 0), "warning": int(by.get("warning") or 0),
"info": int(by.get("info") or 0), "pending": int(got.get("pending") or 0)}}))
- await session.commit()
+ try:
+ await session.commit()
+ except IntegrityError:
+ await session.rollback() # 同一个任务刚被别处记上（补写和收尾撞在一起）
+ return False
+ return True
except Exception: # noqa: BLE001
logger.exception("任务 %s 结束后记数据源变化的待派发事件时出错", job_id)
+ return False
+
+
+_backfilled_at: datetime | None = None
+
+
+async def backfill_source_changes(now: datetime | None = None) -> int:
+ """补写 source.changed：BACKFILL_DAYS 天内测到、还开着的「要处理」级变化，按任务号分组；任务已经结束（探查任务；剖析任务
+ 只算取值观察那几种，同 notify_after_job）、却没有那个任务的事件的，补一条（发生时刻取任务结束的时刻）。返回补了几条。
+ 在 Python 里按任务号分组，不用各库写法不同的 JSON 取值。"""
+ from app.data.catalog_profile_jobs import OBSERVING_SCOPES
+ from app.db.models import CatalogProfileJob
+ from app.db.models_sources import IntrospectJob, SourceChange
+
+ now = now or utcnow()
+ async with SessionLocal() as session:
+ rows = (await session.execute(select(SourceChange.source_id, SourceChange.detected_by).where(
+ SourceChange.severity == "action", SourceChange.status == "open",
+ SourceChange.detected_at >= now - timedelta(days=BACKFILL_DAYS)))).all()
+ jobs: dict[tuple[str, str, str], int] = {}
+ for source_id, detected in rows:
+ detected = detected if isinstance(detected, dict) else {}
+ job_id, kind = detected.get("job_id"), detected.get("job_kind")
+ if job_id and kind in ("introspect", "profile"):
+ key = (str(kind), str(job_id), source_id)
+ jobs[key] = jobs.get(key, 0) + 1
+ if not jobs:
+ return 0
+ have = set((await session.execute(select(TriggerEvent.dedupe_key).where(
+ TriggerEvent.dedupe_key.in_([source_key(j) for _, j, _ in jobs])))).scalars())
+ todo: list[tuple[str, str, int, datetime | None]] = []
+ for (kind, job_id, source_id), n in jobs.items():
+ if source_key(job_id) in have:
+ continue
+ job = await session.get(IntrospectJob if kind == "introspect" else CatalogProfileJob, job_id)
+ if job is None or job.status in ("queued", "running"):
+ continue # 还在进行：结束时自己会记
+ if kind == "profile" and job.scope not in OBSERVING_SCOPES:
+ continue
+ todo.append((source_id, job_id, n, job.finished_at))
+ made = 0
+ for source_id, job_id, n, finished in todo:
+ if await emit_source_changed(source_id, job_id, n, occurred_at=finished):
+ made += 1
+ logger.info("补写了任务 %s 的数据源变化事件（它结束时没记上）", job_id)
+ return made
# --------------------------------------------------------------------------
@@ -170,8 +248,16 @@ _cleaned_at: datetime | None = None
async def dispatch_due(now: datetime | None = None) -> list[ScheduleFire]:
- """完整的一拍：派新来的事件、发去抖窗口到了的触发、顺手清理。返回这一拍记下的触发。一个任务出错不影响别的。"""
+ """完整的一拍：补写没记上的 source.changed（每 BACKFILL_EVERY 一次，刚起的服务第一拍就补）、派新来的事件、发去抖窗口到了的
+ 触发、顺手清理。返回这一拍记下的触发。一个任务出错不影响别的。"""
+ global _backfilled_at
now = now or utcnow()
+ if _backfilled_at is None or now - _backfilled_at >= BACKFILL_EVERY:
+ _backfilled_at = now
+ try:
+ await backfill_source_changes(now)
+ except Exception: # noqa: BLE001 - 补写是兜底，出错不挡派发
+ logger.exception("补写数据源变化的待派发事件时出错")
await route(now)
out = await fire_waiting(now)
await _clean(now)
@@ -213,72 +299,144 @@ async def route(now: datetime | None = None) -> int:
async def fire_waiting(now: datetime | None = None) -> list[ScheduleFire]:
- """去抖窗口到了的任务各合成一次触发。"""
+ """有等着的事件的任务，各把到点的去抖窗口处理完。"""
now = now or utcnow()
async with SessionLocal() as session:
- waiting = (await session.execute(
- select(TriggerDelivery.schedule_id, func.min(TriggerDelivery.occurred_at))
- .where(TriggerDelivery.status == "pending").group_by(TriggerDelivery.schedule_id))).all()
+ waiting = list((await session.execute(
+ select(TriggerDelivery.schedule_id).where(TriggerDelivery.status == "pending").distinct())).scalars())
out: list[ScheduleFire] = []
- for schedule_id, first in waiting:
+ for schedule_id in waiting:
try:
- fire = await _fire_one(schedule_id, first, now)
- if fire is not None:
- out.append(fire)
+ out.extend(await _drain(schedule_id, now))
except Exception: # noqa: BLE001 - 一个任务出错不能拖住这一拍的其他任务
logger.exception("事件触发的定时任务 %s 这一拍处理失败", schedule_id)
return out
-async def _fire_one(schedule_id: str, first: datetime, now: datetime) -> ScheduleFire | None:
+@dataclass
+class _Window:
+ """一个去抖窗口：同一个分窗口的键（上传表格是期次）、落在里面的几条（delivery id, 事件, 派给任务的时刻）、结束的时刻。"""
+
+ key: str
+ start: datetime
+ due: datetime
+ items: list[tuple[str, TriggerEvent, datetime]] = field(default_factory=list)
+
+ @property
+ def events(self) -> list[TriggerEvent]:
+ return [ev for _, ev, _ in self.items]
+
+
+def windows(items: list[tuple[str, TriggerEvent, datetime]], debounce_s: int) -> list[_Window]:
+ """按发生时刻切成去抖窗口（items 已按发生时刻排好）：每个分窗口的键各自一串，窗口从第一条起算 debounce_s 秒；按结束时刻
+ 排。两个窗口结束在同一刻（不同期次同一刻启用）时后一个往后挪 1 微秒：fire_at 不撞触发记录的唯一约束，同样的输入永远切出
+ 同样的窗口。"""
+ from app.scheduler import service
+
+ span = timedelta(seconds=debounce_s)
+ current: dict[str, _Window] = {}
+ out: list[_Window] = []
+ for did, ev, routed in items:
+ at = service.as_utc(ev.occurred_at)
+ key = triggers.window_key(ev.event, ev.payload)
+ win = current.get(key)
+ if win is None or at > win.due:
+ win = current[key] = _Window(key, at, at + span)
+ out.append(win)
+ win.items.append((did, ev, routed))
+ out.sort(key=lambda w: (w.due, w.key))
+ for prev, cur in zip(out, out[1:]):
+ if cur.due <= prev.due:
+ cur.due = prev.due + timedelta(microseconds=1)
+ return out
+
+
+async def _drain(schedule_id: str, now: datetime) -> list[ScheduleFire]:
+ """一个任务这一拍到点的窗口：已过补发宽限的合成一条跳过，其余照常发起。"""
from app.scheduler import loop, service
async with SessionLocal() as session:
schedule = await session.get(Schedule, schedule_id)
if schedule is None:
- return None
+ return []
trigger, tz = schedule.trigger if isinstance(schedule.trigger, dict) else {}, schedule.timezone
if not schedule.enabled or schedule.trigger_kind != "event":
# 等着的时候被停用、改成了别的触发方式：停用是「别再触发了」，等着的不再派发
await _close(session, _pending(schedule_id), "dropped")
- return None
- due = service.as_utc(first) + timedelta(seconds=triggers.debounce_of(trigger))
- if now < due:
- return None
- window = (await session.execute(
- select(TriggerDelivery.id, TriggerEvent).join(TriggerEvent, TriggerEvent.id == TriggerDelivery.event_id)
- .where(TriggerDelivery.schedule_id == schedule_id, TriggerDelivery.status == "pending",
- TriggerDelivery.occurred_at <= due)
- .order_by(TriggerDelivery.occurred_at, TriggerEvent.created_at))).all()
+ return []
+ rows = (await session.execute(
+ select(TriggerDelivery.id, TriggerDelivery.created_at, TriggerEvent)
+ .join(TriggerEvent, TriggerEvent.id == TriggerDelivery.event_id)
+ .where(TriggerDelivery.schedule_id == schedule_id, TriggerDelivery.status == "pending")
+ .order_by(TriggerDelivery.occurred_at, TriggerEvent.created_at).limit(DRAIN_BATCH))).all()
# 等着的时候改了触发条件（换了对象、加了过滤）：已经不合条件的不再派发
- stale = [did for did, ev in window if not triggers.matches(trigger, ev.event, ev.subject_id, ev.payload)]
+ stale = [did for did, _, ev in rows if not triggers.matches(trigger, ev.event, ev.subject_id, ev.payload)]
if stale:
await _close(session, TriggerDelivery.id.in_(stale), "dropped")
- window = [(did, ev) for did, ev in window if did not in set(stale)]
- if not window:
- return None
+ live = [(did, ev, service.as_utc(routed)) for did, routed, ev in rows if did not in set(stale)]
+ due = [w for w in windows(live, triggers.debounce_of(trigger)) if w.due <= now][:MAX_WINDOWS]
+ if not due:
+ return []
grace = max(int(schedule.misfire_grace_s or 0), 0)
- events = [ev for _, ev in window]
- latest = events[-1]
- outputs = await _outputs_of(session, latest)
- fire_event = EventFire(
+ limit = max(loop.ON_TIME_S, grace)
+ missed = [w for w in due if (now - w.due).total_seconds() > limit]
+ timely = [w for w in due if w not in missed]
+ outputs = {id(w): await _outputs_of(session, w.events[-1]) for w in timely}
+ out: list[ScheduleFire] = []
+ if missed:
+ fire = await _skip_missed(schedule_id, missed, tz, grace)
+ if fire is not None:
+ out.append(fire)
+ await _close_window(missed, fire.id)
+ for w in timely:
+ fire = await service.fire_once(schedule_id, w.due, "event", event=_event_fire(w.events, tz, outputs[id(w)]))
+ if fire is not None:
+ out.append(fire)
+ await _close_window([w], fire.id)
+ return out
+
+
+def _event_fire(events: list[TriggerEvent], tz: str, outputs: dict[str, Any]) -> EventFire:
+ """几条合成一次的事件：内容取最新的那一条，深度取最深的那一条加 1。"""
+ latest = events[-1]
+ return EventFire(
event=latest.event, subject_id=latest.subject_id, subject_name=latest.subject_name,
values=triggers.values(latest.event, latest.payload, tz, outputs=outputs),
text=triggers.event_text(latest.event, latest.subject_name, latest.payload),
depth=max(int(ev.depth or 0) for ev in events) + 1, chain=[str(x) for x in latest.chain or []],
event_ids=[ev.id for ev in events], first_at=_iso(events[0].occurred_at), last_at=_iso(latest.occurred_at))
- lateness = (now - due).total_seconds()
- if lateness > max(loop.ON_TIME_S, grace):
- reason = (f"事件发生在 {service.local_text(events[0].occurred_at, tz)}，服务当时没有派发（调度未运行或已停止），"
- f"超出补发宽限（{service.duration_text(grace)}），没有发起")
- fire = await service.record_fire(schedule_id, due, "event", "skipped_misfire", reason,
- {"event": fire_event.brief()})
+
+
+async def _skip_missed(schedule_id: str, missed: list[_Window], tz: str, grace: int) -> ScheduleFire | None:
+ """已过补发宽限的几个窗口合成一条 skipped_misfire（fire_at 取最后一个的结束时刻）：合并了几次事件、几个窗口、时间跨度、
+ 合掉的期次。派给任务的时刻离事件发生已经超过宽限的（调度停过、没在派）才说「调度未运行」，否则照实说积压。"""
+ from app.scheduler import service
+
+ events = [ev for w in missed for ev in w.events]
+ events.sort(key=lambda ev: service.as_utc(ev.occurred_at))
+ late_routed = any((routed - service.as_utc(ev.occurred_at)).total_seconds() > grace
+ for w in missed for _, ev, routed in w.items)
+ first, last = service.local_text(events[0].occurred_at, tz), service.local_text(events[-1].occurred_at, tz)
+ span = first if first == last else f"{first} 至 {last}"
+ count = f"{len(events)} 次事件" + (f"（{len(missed)} 个去抖窗口）" if len(missed) > 1 else "")
+ if late_routed:
+ why = "服务当时没有派发（调度未运行或已停止）"
else:
- fire = await service.fire_once(schedule_id, due, "event", event=fire_event)
- if fire is not None:
- async with SessionLocal() as session:
- await _close(session, TriggerDelivery.id.in_([did for did, _ in window]), "fired", fire_id=fire.id)
- return fire
+ why = "事件积压，轮到派发时"
+ reason = f"{span} 的 {count}：{why}，已超出补发宽限（{service.duration_text(grace)}），没有发起；积压的事件不补"
+ brief = _event_fire(events, tz, {}).brief()
+ brief["windows"] = len(missed)
+ periods = [p for p in dict.fromkeys(str((ev.payload or {}).get("period") or "") for ev in events) if p]
+ if events[-1].event == "table.version" and periods:
+ brief["periods"] = periods
+ return await service.record_fire(schedule_id, max(w.due for w in missed), "event", "skipped_misfire", reason,
+ {"event": brief})
+
+
+async def _close_window(wins: list[_Window], fire_id: str) -> None:
+ async with SessionLocal() as session:
+ await _close(session, TriggerDelivery.id.in_([did for w in wins for did, _, _ in w.items]), "fired",
+ fire_id=fire_id)
def _pending(schedule_id: str) -> Any:
@@ -292,13 +450,13 @@ async def _close(session: Any, cond: Any, status: str, *, fire_id: str | None =
async def _outputs_of(session: Any, ev: TriggerEvent) -> dict[str, Any]:
- """workflow.finished 的出口字段：读那次运行的成果（运行已经结束、不会再变）。读不到为空。"""
+ """workflow.finished 的出口字段：读那次运行的成果（运行已经结束、不会再变），过滤同预览、立即运行（triggers.outputs_of）。
+ 读不到为空。"""
if ev.event != "workflow.finished":
return {}
run_id = (ev.payload or {}).get("run_id")
run = await session.get(Run, run_id) if run_id else None
- output = run.output if run is not None and isinstance(run.output, dict) else {}
- return {k: v for k, v in output.items() if not str(k).startswith("_")}
+ return triggers.outputs_of(run.output if run is not None else None)
def _iso(moment: datetime | None) -> str | None:
diff --git a/backend/app/scheduler/triggers.py b/backend/app/scheduler/triggers.py
index f8cc65fa..1a667cf3 100644
--- a/backend/app/scheduler/triggers.py
+++ b/backend/app/scheduler/triggers.py
@@ -8,11 +8,15 @@
- table.version：一张按期累积的上传表格启用了新的一期（data/table_versions 启用那一下）。可选只认新增的一期（period=new）：
重传、替换已有那一期（改错数）的不算；
- workflow.finished：一个工作流的正式运行成功结束。可选只认某个应用（app_id）发起的、某种来源（origin_kind）的；
- 下游的输入模板能引用这次运行的出口字段（{{ event.outputs.<字段> }}）；
+ 下游的输入模板能引用这次运行的出口字段（{{ event.outputs.<字段> }}）。两种成功结束缺省不认：途经驳回的（审批接了驳回出口、
+ 按驳回那条路走完，状态仍是 succeeded，事件里带 rejected_via）——下游会以为「A 成功出了结果」，其实那一期被驳回了，
+ include_rejected 时才认、模板里有 {{ event.rejected }}；应用补算的（app_backfill，升版后按历次输入补算，一连发好几条、
+ 各是某一期的历史输入），include_backfill 或只认这种来源时才认；
- source.changed：一个数据源出现「要处理」级的变化。按源去抖：同一次探查、剖析产生的一批只记一条事件（data/source_changes）。
-trigger 的形状：{event, 对象（source_id / workflow_id / source_id）, 过滤（见上）, debounce_s（缺省 60）}。同一个定时任务
-在 debounce_s 内的几次事件合成一次触发（scheduler/outbox）。
+trigger 的形状：{event, 对象（source_id / workflow_id / source_id）, 过滤（period；app_id、origin_kind、include_rejected、
+include_backfill）, debounce_s（缺省 60）}。同一个定时任务在 debounce_s 内的几次事件合成一次触发，上传表格按期次分窗口
+（window_key，scheduler/outbox）。
**防环。** workflow.finished 能把工作流串起来：A 跑完触发 B、B 跑完又触发 A 就停不下来。保存时查串联成不成环（find_cycle，
停用的任务也算：启用不该绕过这道检查）；运行时按 origin 里记的触发链深度兜底，超过 MAX_DEPTH 层不发起、照实记一次未能发起
@@ -56,7 +60,10 @@ ORIGIN_LABEL = {
}
#: trigger 里认得的键（多出来的拒收：拼错的过滤条件悄悄不生效，比报错更糟）。subject_name 是保存时记下的对象名（接口不给、
#: 带回来也不认，对象删掉之后说法里用它）
-_KEYS = {"event", "source_id", "workflow_id", "period", "app_id", "origin_kind", "debounce_s", "subject_name"}
+_KEYS = {"event", "source_id", "workflow_id", "period", "app_id", "origin_kind", "debounce_s", "subject_name",
+ "include_rejected", "include_backfill"}
+#: 应用补算发起的运行的来源（engine/origin.py）
+BACKFILL_ORIGIN = "app_backfill"
#: 示例事件里给不出真实值时写的占位说明
EXAMPLE_NOTE = "示例值"
@@ -118,6 +125,18 @@ def normalize(value: Any) -> dict[str, Any]:
labels = "、".join(ORIGIN_LABEL[o] for o in FORMAL_ORIGINS)
raise TriggerError(f"只认的来源需要是{labels}之一")
out[name] = text
+ for name in ("include_rejected", "include_backfill"):
+ raw = value.get(name)
+ if raw in (None, False):
+ continue
+ if event != "workflow.finished":
+ raise TriggerError("只有「上游工作流运行成功结束」能选也认途经驳回的、应用补算的")
+ if raw is not True:
+ raise TriggerError("也认途经驳回的、应用补算的需要是 true 或 false")
+ # 只认补算的：补算本来就认，不再另记
+ if name == "include_backfill" and out.get("origin_kind") == BACKFILL_ORIGIN:
+ continue
+ out[name] = True
debounce = value.get("debounce_s", DEFAULT_DEBOUNCE_S)
if isinstance(debounce, bool) or not isinstance(debounce, int) or not 0 <= debounce <= MAX_DEBOUNCE_S:
raise TriggerError(f"去抖需要是 0 到 {MAX_DEBOUNCE_S} 之间的整数秒数（最长 1 天）", "debounce_s")
@@ -149,6 +168,11 @@ def matches(trigger: dict[str, Any] | None, event: str, subject_id: str, payload
return False
if trigger.get("origin_kind") and payload.get("origin_kind") != trigger["origin_kind"]:
return False
+ if payload.get("rejected") and not trigger.get("include_rejected"):
+ return False
+ if payload.get("origin_kind") == BACKFILL_ORIGIN and not (
+ trigger.get("include_backfill") or trigger.get("origin_kind") == BACKFILL_ORIGIN):
+ return False
return True
@@ -282,6 +306,15 @@ def describe(trigger: dict[str, Any], subject_name: str | None, *, app_name: str
notes.append(f"只认应用「{app_name}」发起的" if app_name else "只认指定的应用发起的")
if trigger.get("origin_kind"):
notes.append(f"只认来源为「{ORIGIN_LABEL.get(trigger['origin_kind'], trigger['origin_kind'])}」的")
+ if event == "workflow.finished":
+ # 缺省不认的两种成功结束，照实写出来
+ kinds = [("途经驳回", bool(trigger.get("include_rejected")))]
+ if trigger.get("origin_kind") != BACKFILL_ORIGIN:
+ kinds.append(("应用补算", bool(trigger.get("include_backfill"))))
+ excluded = "、".join(k for k, on in kinds if not on)
+ included = "、".join(k for k, on in kinds if on)
+ notes += [f"{included}的也算"] if included else []
+ notes += [f"{excluded}的不算"] if excluded else []
debounce = debounce_of(trigger)
from app.scheduler.service import duration_text
@@ -300,7 +333,8 @@ def event_text(event: str, subject_name: str, payload: dict[str, Any] | None) ->
return f"上传表格{name}{what}{f'（{period}）' if period else ''}"
if event == "workflow.finished":
run = str(payload.get("run_id") or "")
- return f"工作流{name}的正式运行成功结束{f'（#{run[:6]}）' if run else ''}"
+ rejected = "（途经驳回）" if payload.get("rejected") else ""
+ return f"工作流{name}的正式运行成功结束{f'（#{run[:6]}）' if run else ''}{rejected}"
if event == "source.changed":
n = (payload.get("counts") or {}).get("new")
return f"数据源{name}出现要处理的变化{f'（新记 {n} 条）' if isinstance(n, int) and n else ''}"
@@ -326,6 +360,7 @@ VARIABLES: dict[str, tuple[tuple[str, str, str], ...]] = {
("workflow_id", "上游工作流的编号", "wf…"),
("run_id", "上游那次运行的编号", "run…"),
("finished_at", "上游运行结束的时间（带时区偏移的 ISO 时间）", "2026-10-08T09:30:00+08:00"),
+ ("rejected", "上游这次运行有没有途经驳回（true / false；缺省不认途经驳回的，勾了「也认」才会是 true）", "false"),
("outputs", "上游运行的出口字段（写 event.outputs.<字段名>）", "{…}"),
),
"source.changed": (
@@ -396,7 +431,7 @@ def values(event: str, payload: dict[str, Any] | None, tz: str, *, outputs: dict
if event == "workflow.finished":
return {"workflow": payload.get("workflow") or "", "workflow_id": payload.get("workflow_id") or "",
"run_id": payload.get("run_id") or "", "finished_at": _zoned(payload.get("finished_at"), tz),
- "outputs": dict(outputs or {})}
+ "rejected": "true" if payload.get("rejected") else "false", "outputs": dict(outputs or {})}
if event == "source.changed":
counts = payload.get("counts") if isinstance(payload.get("counts"), dict) else {}
return {"source": payload.get("source") or "", "source_id": payload.get("source_id") or "",
@@ -501,11 +536,34 @@ async def sample(session: Any, trigger: dict[str, Any]) -> Sample:
query = query.where(Run.app_id == trigger["app_id"])
if trigger.get("origin_kind"):
query = query.where(Run.origin_kind == trigger["origin_kind"])
- run = (await session.execute(query.order_by(Run.finished_at.desc()).limit(1))).scalars().first() if subject else None
- if run is None:
- return Sample(event, subject, name, {"workflow": name or EXAMPLE_NOTE, "workflow_id": subject, "run_id": "",
- "finished_at": ""}, real=False)
- return Sample(event, subject, name, run_payload(run), outputs=dict(run.output or {}))
+ candidates = list((await session.execute(query.order_by(Run.finished_at.desc()).limit(SAMPLE_SCAN)))
+ .scalars()) if subject else []
+ # 和真派发同一套条件（matches）：缺省不认的途经驳回、应用补算跳过，取合条件的最近一次
+ for run in candidates:
+ payload = run_payload(run, await _rejected_via_of(session, run.id))
+ if matches(trigger, event, subject, payload):
+ return Sample(event, subject, name, payload, outputs=outputs_of(run.output))
+ return Sample(event, subject, name, {"workflow": name or EXAMPLE_NOTE, "workflow_id": subject, "run_id": "",
+ "finished_at": "", "rejected": False}, real=False)
+
+
+#: 示例事件往回找几次成功的正式运行（找合条件的最近一次）
+SAMPLE_SCAN = 20
+
+
+async def _rejected_via_of(session: Any, run_id: str) -> list[Any] | None:
+ """这次运行途经的驳回：run.finished 事件里的 rejected_via（W1-b 的驳回出口写的），没有为 None。"""
+ from app.db.models import RunEvent
+
+ data = (await session.execute(select(RunEvent.data).where(RunEvent.run_id == run_id, RunEvent.type == "run.finished")
+ .order_by(RunEvent.seq.desc()).limit(1))).scalar()
+ via = data.get("rejected_via") if isinstance(data, dict) else None
+ return list(via) if isinstance(via, list) and via else None
+
+
+def outputs_of(output: Any) -> dict[str, Any]:
+ """给模板的出口字段（{{ event.outputs.* }}）：「_」开头的是系统写的（出具档位之类），不给。真派发、预览、立即运行共用。"""
+ return {k: v for k, v in output.items() if not str(k).startswith("_")} if isinstance(output, dict) else {}
def table_payload(name: str, *, snapshot_id: str, period_start: str | None, period_end: str | None,
@@ -516,11 +574,21 @@ def table_payload(name: str, *, snapshot_id: str, period_start: str | None, peri
"activated_at": _iso(activated_at), "replaced": bool(replaced), "seq": seq}
-def run_payload(run: Any) -> dict[str, Any]:
- """workflow.finished 的事件内容：不含出口字段（派发时按 run_id 读成果）。"""
+def run_payload(run: Any, rejected_via: Any = None) -> dict[str, Any]:
+ """workflow.finished 的事件内容：不含出口字段（派发时按 run_id 读成果）。rejected_via 是途经了哪几个关卡的驳回（W1-b 的
+ 驳回出口；runner 收尾时交进来），没途经驳回为 None。"""
+ via = (list(rejected_via) if isinstance(rejected_via, (list, tuple)) else [rejected_via]) if rejected_via else None
return {"workflow": run.workflow_name or "", "workflow_id": run.workflow_id or "", "run_id": run.id,
"finished_at": _iso(run.finished_at), "app_id": run.app_id, "origin_kind": run.origin_kind,
- "version": run.version}
+ "version": run.version, "rejected": bool(via), "rejected_via": via}
+
+
+def window_key(event: str, payload: dict[str, Any] | None) -> str:
+ """去抖时按什么分窗口：上传表格按期次分（9 月、10 月连着启用，各有各的触发，不把 9 月合掉），别的不分。"""
+ if event != "table.version":
+ return ""
+ payload = payload or {}
+ return str(payload.get("period") or f"{payload.get('period_start') or ''}~{payload.get('period_end') or ''}")
def _iso(moment: datetime | None) -> str:
diff --git a/backend/tests/test_schedule_events_api.py b/backend/tests/test_schedule_events_api.py
index fe86acb2..e07d7561 100644
--- a/backend/tests/test_schedule_events_api.py
+++ b/backend/tests/test_schedule_events_api.py
@@ -369,3 +369,21 @@ async def test_trigger_options(client):
assert flows[draft]["ok"] is False and "发布" in flows[draft]["reason"]
assert "hook" in [o["kind"] for o in out["origins"]] and out["max_depth"] == 5
assert out["default_debounce_s"] == 60
+
+
+async def test_include_rejected_and_backfill_through_the_api(client):
+ """复核 M1、C5：上游工作流的触发缺省不认途经驳回的、应用补算的，说法照实写；勾了「也认」的存下来、说法跟着变；
+ 别的事件带这两项拒收"""
+ a, a_name = await workflow("日报")
+ b, _ = await workflow("汇总", fields=[{"name": "rej"}])
+ plain = (await client.post("/api/schedules", json=body(b, {"event": "workflow.finished", "workflow_id": a},
+ input_template={"rej": "{{ event.rejected }}"}))).json()
+ assert "途经驳回、应用补算的不算" in plain["trigger_text"] and "include_rejected" not in plain["trigger"]
+ r = await client.patch(f"/api/schedules/{plain['id']}", json={
+ "trigger": {"event": "workflow.finished", "workflow_id": a, "include_rejected": True}, "version": plain["version"]})
+ assert r.status_code == 200, r.text
+ assert r.json()["trigger"]["include_rejected"] is True and "途经驳回的也算；应用补算的不算" in r.json()["trigger_text"]
+ src = await upload_table("门店销售")
+ bad = await client.post("/api/schedules", json=body(b, {"event": "table.version", "source_id": src,
+ "include_backfill": True}))
+ assert bad.status_code == 422 and bad.json()["field"] == "trigger"
diff --git a/backend/tests/test_schedule_events_review.py b/backend/tests/test_schedule_events_review.py
new file mode 100644
index 00000000..14eb4dc7
--- /dev/null
+++ b/backend/tests/test_schedule_events_review.py
@@ -0,0 +1,277 @@
+"""数据事件触发复核意见的修正（docs/design/orchestration-round/00-总方案.md §4；scheduler/outbox.py、triggers.py）。
+
+- M1 途经驳回：正式运行走驳回出口跑完（succeeded，run.finished 带 rejected_via）照发 workflow.finished，事件内容带 rejected_via、
+ 说法接「（途经驳回）」、模板有 {{ event.rejected }}；触发条件缺省只认没途经驳回的，include_rejected 时也认；
+- C1 去抖积压：一拍处理到没有到点的窗口为止；已过宽限的几个窗口合成一条跳过（合并了几条、时间跨度），「调度未运行」只在
+ 派发确实晚到（派给任务的时刻离事件发生超过宽限）时说，否则照实说积压；积压清完之后新来的事件照常触发；
+- C3 source.changed 按任务补写：同一个任务只记一条（dedupe），任务结束后没记上的（进程死了、库被锁）在调度的一拍按 job_id 补写；
+- C4 不同期次的 table.version 不合进同一个窗口：各期有各自的触发；同一期的重传照旧合并；
+- C5 应用补算（app_backfill）发的 workflow.finished 缺省不认，include_backfill 或只认补算时才认；
+- C6 预览、立即运行的示例事件和真派发同一套出口字段过滤（「_」开头的不给模板）。
+
+工作流、表格、数据源都是虚构的。
+"""
+from __future__ import annotations
+
+import uuid
+from datetime import timedelta
+
+from sqlalchemy import select
+
+from app.db.base import SessionLocal, utcnow
+from app.db.models import DataSource, Run, Schedule
+from app.db.models_sources import IntrospectJob, SourceChange
+from app.db.models_triggers import TriggerDelivery, TriggerEvent
+from app.scheduler import outbox, triggers
+from tests.test_schedule_events import ( # noqa: F401 （engine_up：每个测试清掉定时任务和事件）
+ deliveries, engine_up, event_schedule, events, fires, formal_run, later, put_event, settle, workflow,
+)
+
+
+# --------------------------------------------------------------------------
+# M1 途经驳回
+# --------------------------------------------------------------------------
+
+
+async def test_rejected_via_is_carried_and_excluded_by_default():
+ upstream = await workflow("门店日报")
+ plain = await event_schedule(await workflow("区域汇总"),
+ {"event": "workflow.finished", "workflow_id": upstream, "debounce_s": 0}, name="不认驳回")
+ also = await event_schedule(await workflow("驳回复核", fields=[{"name": "rej"}]),
+ {"event": "workflow.finished", "workflow_id": upstream, "debounce_s": 0,
+ "include_rejected": True}, template={"rej": "{{ event.rejected }}"}, name="也认驳回")
+ a = await formal_run(upstream)
+ async with SessionLocal() as session:
+ # runner 收尾时把这次运行途经的驳回交进来（W1-b 的 _rejected_via(state)）：这里直接调，覆盖那一下
+ await session.execute(TriggerEvent.__table__.delete().where(TriggerEvent.subject_id == upstream))
+ outbox.stage_run_finished(session, await session.get(Run, a.id),
+ rejected_via=[{"node_id": "gate", "label": "主管审核"}])
+ await session.commit()
+ [ev] = await events("workflow.finished", upstream)
+ assert ev.payload["rejected"] is True and ev.payload["rejected_via"] == [{"node_id": "gate", "label": "主管审核"}]
+ assert triggers.event_text(ev.event, ev.subject_name, ev.payload).endswith("（途经驳回）")
+ [fire] = await outbox.dispatch_due(later())
+ assert fire.schedule_id == also and fire.status == "started" and fire.inputs["inputs"] == {"rej": "true"}
+ assert fire.inputs["event"]["text"].endswith("（途经驳回）")
+ assert await fires(plain) == [] and await deliveries(plain) == []
+ await settle(fire.run_id)
+
+
+async def test_plain_runs_say_not_rejected():
+ upstream = await workflow("门店日报")
+ sid = await event_schedule(await workflow("区域汇总", fields=[{"name": "rej"}]),
+ {"event": "workflow.finished", "workflow_id": upstream, "debounce_s": 0},
+ template={"rej": "{{ event.rejected }}"})
+ await formal_run(upstream) # runner 收尾没交 rejected_via：没途经驳回
+ [ev] = await events("workflow.finished", upstream)
+ assert ev.payload["rejected"] is False and ev.payload["rejected_via"] is None
+ [fire] = await outbox.dispatch_due(later())
+ assert fire.schedule_id == sid and fire.inputs["inputs"] == {"rej": "false"}
+ assert "途经驳回" not in fire.inputs["event"]["text"]
+ await settle(fire.run_id)
+
+
+def test_rejected_and_backfill_options_are_described():
+ base = triggers.normalize({"event": "workflow.finished", "workflow_id": "wf"})
+ assert "（途经驳回、应用补算的不算；去抖 1 分钟）" in triggers.describe(base, "门店日报")
+ both = triggers.normalize({"event": "workflow.finished", "workflow_id": "wf", "include_rejected": True,
+ "include_backfill": True})
+ assert both["include_rejected"] is True and both["include_backfill"] is True
+ assert "（途经驳回、应用补算的也算；去抖 1 分钟）" in triggers.describe(both, "门店日报")
+ # 只认应用补算的：补算本来就要认，不再说「补算的不算」
+ only = triggers.normalize({"event": "workflow.finished", "workflow_id": "wf", "origin_kind": "app_backfill"})
+ text = triggers.describe(only, "门店日报")
+ assert "只认来源为「应用补算」的" in text and "应用补算的不算" not in text and "途经驳回的不算" in text
+ # 别的事件不收这两项
+ for key in ("include_rejected", "include_backfill"):
+ try:
+ triggers.normalize({"event": "table.version", "source_id": "x", key: True})
+ except triggers.TriggerError as e:
+ assert "上游工作流运行成功结束" in str(e)
+ else:
+ raise AssertionError(key)
+ assert "event.rejected" in triggers.names("workflow.finished")
+
+
+# --------------------------------------------------------------------------
+# C5 应用补算
+# --------------------------------------------------------------------------
+
+
+def test_backfill_runs_are_excluded_unless_asked():
+ payload = {"origin_kind": "app_backfill", "rejected": False}
+ base = triggers.normalize({"event": "workflow.finished", "workflow_id": "wf"})
+ assert not triggers.matches(base, "workflow.finished", "wf", payload)
+ assert triggers.matches(base, "workflow.finished", "wf", {**payload, "origin_kind": "launcher"})
+ assert triggers.matches({**base, "include_backfill": True}, "workflow.finished", "wf", payload)
+ assert triggers.matches({**base, "origin_kind": "app_backfill"}, "workflow.finished", "wf", payload)
+
+
+# --------------------------------------------------------------------------
+# C6 示例事件的出口字段
+# --------------------------------------------------------------------------
+
+
+async def test_sample_outputs_use_the_same_filter_as_dispatch():
+ upstream = await workflow("门店日报")
+ run = await formal_run(upstream)
+ async with SessionLocal() as session:
+ await session.execute(Run.__table__.update().where(Run.id == run.id).values(
+ output={"r": "日报完成", "_issuance": {"tier": "full"}}))
+ await session.commit()
+ got = await triggers.sample(session, {"event": "workflow.finished", "workflow_id": upstream})
+ assert got.real and got.outputs == {"r": "日报完成"}
+ assert triggers.outputs_of({"r": 1, "_x": 2}) == {"r": 1}
+
+
+async def test_sample_skips_rejected_and_backfill_runs_by_default():
+ upstream = await workflow("门店日报")
+ plain = await formal_run(upstream)
+ backfill = await formal_run(upstream, origin_kind="app_backfill")
+ async with SessionLocal() as session:
+ got = await triggers.sample(session, {"event": "workflow.finished", "workflow_id": upstream})
+ wide = await triggers.sample(session, {"event": "workflow.finished", "workflow_id": upstream,
+ "include_backfill": True})
+ assert got.payload["run_id"] == plain.id and wide.payload["run_id"] == backfill.id
+
+
+# --------------------------------------------------------------------------
+# C1 去抖积压
+# --------------------------------------------------------------------------
+
+
+async def _backlog(upstream: str, sid: str, *, n: int, gap_s: int, routed_late: bool) -> tuple:
+ """n 条相隔 gap_s 秒的事件，派给了任务、还没处理。routed_late：派给任务的时刻是现在（调度停过）；否则发生后 1 秒就派了
+ （调度在跑，只是没处理完）。"""
+ t0 = utcnow().replace(microsecond=0) - timedelta(seconds=gap_s * (n - 1) + 61)
+ ids = []
+ async with SessionLocal() as session:
+ for i in range(n):
+ at = t0 + timedelta(seconds=gap_s * i)
+ ev = TriggerEvent(event="workflow.finished", subject_id=upstream, subject_name="门店日报",
+ payload={"workflow_id": upstream, "run_id": f"run-{i}", "rejected": False},
+ occurred_at=at, created_at=at, status="routed", matched=1)
+ session.add(ev)
+ await session.flush()
+ session.add(TriggerDelivery(schedule_id=sid, event_id=ev.id, occurred_at=at,
+ created_at=utcnow() if routed_late else at + timedelta(seconds=1)))
+ ids.append(ev.id)
+ await session.commit()
+ return t0, ids
+
+
+async def test_backlog_is_drained_in_one_tick_and_missed_windows_merge_into_one_skip():
+ """12 条相隔 5 分钟、去抖 60 秒、宽限 600 秒：一拍处理完。过了宽限的 10 个窗口合成一条跳过（合并 10 次、写时间跨度），
+ 宽限内的两个照常发起（第二个照重叠规则可能跳过）；积压清完以后新来的事件照常触发"""
+ upstream = await workflow("门店日报")
+ sid = await event_schedule(await workflow("区域汇总", fields=[{"name": "run"}]),
+ {"event": "workflow.finished", "workflow_id": upstream, "debounce_s": 60}, grace=600,
+ template={"run": "{{ event.run_id }}"})
+ t0, ids = await _backlog(upstream, sid, n=12, gap_s=300, routed_late=True)
+ done = await outbox.dispatch_due(utcnow())
+ assert len(done) == 3, [(f.status, f.reason) for f in done]
+ skip, *rest = done
+ assert skip.status == "skipped_misfire" and skip.kind == "event"
+ assert skip.inputs["event"]["merged"] == 10 and skip.inputs["event"]["windows"] == 10
+ assert skip.inputs["event"]["event_ids"] == ids[:10]
+ assert "服务当时没有派发（调度未运行或已停止）" in skip.reason and " 至 " in skip.reason and "10 次事件" in skip.reason
+ assert [f.inputs["event"]["event_ids"] for f in rest] == [[ids[10]], [ids[11]]]
+ assert rest[0].status == "started" and rest[1].status in ("started", "skipped_overlap")
+ assert {d.status for d in await deliveries(sid)} == {"fired"}
+ for f in rest:
+ if f.run_id:
+ await settle(f.run_id)
+ # 积压清完：新来的一条照常触发
+ new = await put_event("workflow.finished", upstream, {"workflow_id": upstream, "run_id": "run-new"}, utcnow())
+ [fresh] = await outbox.dispatch_due(later(61))
+ assert fresh.status == "started" and fresh.inputs["event"]["event_ids"] == [new]
+ await settle(fresh.run_id)
+
+
+async def test_backlog_while_scheduler_ran_says_backlog_not_scheduler_off():
+ upstream = await workflow("门店日报")
+ sid = await event_schedule(await workflow("区域汇总"),
+ {"event": "workflow.finished", "workflow_id": upstream, "debounce_s": 60}, grace=600)
+ await _backlog(upstream, sid, n=4, gap_s=600, routed_late=False)
+ skip, *rest = await outbox.dispatch_due(utcnow())
+ assert skip.status == "skipped_misfire" and skip.inputs["event"]["merged"] == 3
+ assert "调度未运行" not in skip.reason and "积压" in skip.reason, skip.reason
+ for f in rest:
+ if f.run_id:
+ await settle(f.run_id)
+
+
+# --------------------------------------------------------------------------
+# C4 不同期次各自一个窗口
+# --------------------------------------------------------------------------
+
+
+async def test_different_periods_get_their_own_windows():
+ table = uuid.uuid4().hex
+ sid = await event_schedule(await workflow("月报", fields=[{"name": "month"}]),
+ {"event": "table.version", "source_id": table, "debounce_s": 60},
+ template={"month": "{{ event.period }}"})
+ t0 = utcnow().replace(microsecond=0)
+ payload = lambda p, s, e: {"table": "门店销售", "period": p, "period_start": s, "period_end": e, # noqa: E731
+ "version": f"snap-{p}", "replaced": False}
+ sep = await put_event("table.version", table, payload("2026-09", "2026-09-01", "2026-09-30"), t0, name="门店销售")
+ oct_ = await put_event("table.version", table, payload("2026-10", "2026-10-01", "2026-10-31"),
+ t0 + timedelta(seconds=30), name="门店销售")
+ again = await put_event("table.version", table, {**payload("2026-09", "2026-09-01", "2026-09-30"), "replaced": True},
+ t0 + timedelta(seconds=40), name="门店销售")
+ [first] = await outbox.dispatch_due(t0 + timedelta(seconds=61))
+ # 9 月一个窗口：9 月和 9 月的重传合在一起（内容取最新的那一条），10 月不合进来
+ assert first.inputs["event"]["event_ids"] == [sep, again] and first.inputs["inputs"] == {"month": "2026-09"}
+ await settle(first.run_id)
+ [second] = await outbox.dispatch_due(t0 + timedelta(seconds=91))
+ assert second.inputs["event"]["event_ids"] == [oct_] and second.inputs["inputs"] == {"month": "2026-10"}
+ assert second.fire_at == t0 + timedelta(seconds=90) and second.status == "started"
+ await settle(second.run_id)
+
+
+# --------------------------------------------------------------------------
+# C3 source.changed 按任务补写
+# --------------------------------------------------------------------------
+
+
+async def _source_with_changes(job_status: str) -> tuple[str, str]:
+ async with SessionLocal() as session:
+ source = DataSource(name=f"门店库-{uuid.uuid4().hex[:6]}", kind="sqlite", database=":memory:")
+ session.add(source)
+ await session.flush()
+ job = IntrospectJob(source_id=source.id, status=job_status,
+ finished_at=utcnow() if job_status != "running" else None)
+ session.add(job)
+ await session.flush()
+ for i in range(2):
+ session.add(SourceChange(source_id=source.id, source_name=source.name, table_name="orders",
+ column_name=f"c{i}", kind="column_removed", severity="action",
+ detected_by={"job_kind": "introspect", "job_id": job.id},
+ fingerprint=uuid.uuid4().hex, status="open"))
+ await session.commit()
+ return source.id, job.id
+
+
+async def test_source_changed_is_backfilled_once_per_job():
+ """任务结束了、变化记下了，那条事件却没记上（进程在中间死了）：调度的一拍按 job_id 补写，只补一次；之后再走到
+ notify_after_job 也不重复"""
+ source_id, job_id = await _source_with_changes("succeeded")
+ running_source, _ = await _source_with_changes("running")
+ assert await events("source.changed", source_id) == []
+ assert await outbox.backfill_source_changes(utcnow()) >= 1
+ [ev] = await events("source.changed", source_id)
+ assert ev.payload["job_id"] == job_id and ev.payload["counts"]["new"] == 2 and ev.dedupe_key == f"source.changed:{job_id}"
+ # 还在进行的任务不补：它结束时自己会记
+ assert await events("source.changed", running_source) == []
+ await outbox.backfill_source_changes(utcnow())
+ from app.data import source_changes
+
+ await source_changes.notify_after_job(source_id, job_id)
+ assert len(await events("source.changed", source_id)) == 1
+
+
+async def test_dispatch_tick_runs_the_backfill():
+ source_id, _ = await _source_with_changes("succeeded")
+ outbox._backfilled_at = None # 刚起的服务：第一拍就补
+ await outbox.dispatch_due(utcnow())
+ assert len(await events("source.changed", source_id)) == 1
diff --git a/frontend/src/lib/termsTrigger.ts b/frontend/src/lib/termsTrigger.ts
index 13f4cc1d..2a94ad32 100644
--- a/frontend/src/lib/termsTrigger.ts
+++ b/frontend/src/lib/termsTrigger.ts
@@ -55,6 +55,10 @@ export const TRIGGER_TEXT = {
originLabel: '只认来源',
originAny: '不限来源',
filterHint: '只在上游那次正式运行由这个应用、这种来源发起时触发',
+ includeRejected: '也认途经驳回的',
+ includeRejectedHint: '上游某个审批被驳回、按驳回那条路走完时，运行仍算成功结束。不勾时这类运行不触发；勾了以后，输入模板里的 {{ event.rejected }} 是 true',
+ includeBackfill: '也认应用补算的',
+ includeBackfillHint: '应用升版后按历次输入补算，会连着结束好几次正式运行，每次是某一期的历史输入。不勾时补算的运行不触发',
debounce: '去抖',
debounceHint: '这段时间内的几次事件合成一次触发，按最新的那一次渲染输入；上一次运行还没结束时仍按重叠策略跳过',
debounceOption: (span: string) => (span ? `${span}内合成一次` : '不去抖，每次事件都触发'),
@@ -86,6 +90,9 @@ export const TRIGGER_TEXT = {
merged: (n: number) => `合并了 ${n} 次事件`,
depth: (n: number) => `触发链第 ${n} 层`,
sample: '立即运行，按最近一次事件',
+ /** 几个过了补发宽限的去抖窗口合成一条跳过 */
+ windows: (n: number) => `${n} 个去抖窗口`,
+ periods: (list: string) => `期次 ${list}`,
sampleExample: '立即运行，事件为示例值',
// ---- 运行详情页头
diff --git a/frontend/src/pages/schedules/FireHistory.tsx b/frontend/src/pages/schedules/FireHistory.tsx
index 5f0378a9..480e18bc 100644
--- a/frontend/src/pages/schedules/FireHistory.tsx
+++ b/frontend/src/pages/schedules/FireHistory.tsx
@@ -199,6 +199,8 @@ function FireRow({ fire: f, schedule, now }: { fire: ScheduleFire; schedule: Sch
function EventFire({ event }: { event: ScheduleFireEvent }) {
const notes = [
event.merged > 1 ? TT.merged(event.merged) : null,
+ event.windows && event.windows > 1 ? TT.windows(event.windows) : null,
+ event.periods && event.periods.length > 1 ? TT.periods(event.periods.join('、')) : null,
event.depth > 1 ? TT.depth(event.depth) : null,
event.sample ? (event.real ? TT.sample : TT.sampleExample) : null,
].filter(Boolean)
diff --git a/frontend/src/pages/schedules/TriggerEditor.tsx b/frontend/src/pages/schedules/TriggerEditor.tsx
index a3216f61..d36f131b 100644
--- a/frontend/src/pages/schedules/TriggerEditor.tsx
+++ b/frontend/src/pages/schedules/TriggerEditor.tsx
@@ -6,7 +6,9 @@ import { errorMessage } from '../../lib/errors'
import { timezoneOptions } from '../../lib/freq'
import { TRIGGER_TEXT as TT } from '../../lib/termsTrigger'
import type { ScheduleEventKind, ScheduleTriggerKind, ScheduleTriggerOptions } from '../../types'
-import { DEBOUNCE_OPTIONS, EVENT_KINDS, TRIGGER_KINDS, graceLabel, switchEvent, type TriggerForm } from './model'
+import {
+ BACKFILL_ORIGIN, DEBOUNCE_OPTIONS, EVENT_KINDS, TRIGGER_KINDS, graceLabel, switchEvent, type TriggerForm,
+} from './model'
import { isUnsupported } from './parts'
/**
@@ -179,6 +181,21 @@ export function EventSection({ value, onChange, timezone, setTimezone, savedName
{TT.filterHint}
+ {/* 缺省不认的两种成功结束：途经驳回的、应用补算的（服务端 triggers.matches），要认的勾上 */}
+
+
+ {value.originKind !== BACKFILL_ORIGIN && (
+
+ )}
+
{TT.chainHint(options.data.max_depth)}
)}
diff --git a/frontend/src/pages/schedules/model.ts b/frontend/src/pages/schedules/model.ts
index 73acc350..8e922f84 100644
--- a/frontend/src/pages/schedules/model.ts
+++ b/frontend/src/pages/schedules/model.ts
@@ -43,9 +43,15 @@ export interface TriggerForm {
period: 'any' | 'new'
appId: string
originKind: string
+ /** 也认途经驳回的、也认应用补算的（只给上游工作流，缺省都不认） */
+ includeRejected: boolean
+ includeBackfill: boolean
debounceS: number
}
+/** 应用补算的来源：选了「只认应用补算」时，「也认应用补算的」不再适用 */
+export const BACKFILL_ORIGIN = 'app_backfill'
+
export const EVENT_KINDS: readonly ScheduleEventKind[] = ['table.version', 'workflow.finished', 'source.changed']
export const TRIGGER_KINDS: readonly ScheduleTriggerKind[] = ['cron', 'event']
/** 去抖的选项（秒）：不去抖、1 分钟（缺省）、5 分钟、15 分钟、1 小时 */
@@ -60,6 +66,8 @@ export function triggerFormOf(t: ScheduleTrigger | null | undefined): TriggerFor
period: t?.period === 'new' ? 'new' : 'any',
appId: t?.app_id ?? '',
originKind: t?.origin_kind ?? '',
+ includeRejected: t?.include_rejected === true,
+ includeBackfill: t?.include_backfill === true,
debounceS: typeof t?.debounce_s === 'number' ? t.debounce_s : DEFAULT_DEBOUNCE_S,
}
}
@@ -71,6 +79,8 @@ export function triggerBody(t: TriggerForm): ScheduleTrigger {
out.workflow_id = t.subjectId
if (t.appId) out.app_id = t.appId
if (t.originKind) out.origin_kind = t.originKind
+ if (t.includeRejected) out.include_rejected = true
+ if (t.includeBackfill && t.originKind !== BACKFILL_ORIGIN) out.include_backfill = true
} else {
out.source_id = t.subjectId
if (t.event === 'table.version' && t.period === 'new') out.period = 'new'
@@ -81,7 +91,8 @@ export function triggerBody(t: TriggerForm): ScheduleTrigger {
/** 换了事件种类：对象、过滤都不再适用，清掉；去抖留着 */
export function switchEvent(t: TriggerForm, event: ScheduleEventKind): TriggerForm {
if (t.event === event) return t
- return { event, subjectId: '', period: 'any', appId: '', originKind: '', debounceS: t.debounceS }
+ return { event, subjectId: '', period: 'any', appId: '', originKind: '', includeRejected: false, includeBackfill: false,
+ debounceS: t.debounceS }
}
export interface ScheduleForm {
diff --git a/frontend/src/types.ts b/frontend/src/types.ts
index 5fe469f3..b80e2ed2 100644
--- a/frontend/src/types.ts
+++ b/frontend/src/types.ts
@@ -4069,6 +4069,12 @@ export interface ScheduleTrigger {
/** 只认某个应用发起的、某种来源的运行，只给 workflow.finished */
app_id?: string
origin_kind?: string
+ /**
+ * 也认途经驳回的（上游走了某个审批的驳回出口、按驳回那条路跑完，状态仍是成功）、也认应用补算的；缺省都不认。只给
+ * workflow.finished，不写即 false
+ */
+ include_rejected?: boolean
+ include_backfill?: boolean
/** 去抖：同一个任务这么多秒内的几次事件合成一次 */
debounce_s: number
/** 保存时记下的对象名（对象删掉之后界面写它）；读出来才有，交上去不用带 */
@@ -4108,6 +4114,9 @@ export interface ScheduleFireEvent {
/** 立即运行时按示例事件发起的（不是真来了一次事件）；real 为假时是示例值 */
sample: boolean
real: boolean
+ /** 几个过了补发宽限的去抖窗口合成一条跳过时才有：合了几个窗口、合掉的期次（上传表格）；老服务端不给 */
+ windows?: number | null
+ periods?: string[] | null
}
export interface ScheduleFire {
diff --git a/scripts/check-schedules.mjs b/scripts/check-schedules.mjs
index b1e08b8e..b125ed57 100644
--- a/scripts/check-schedules.mjs
+++ b/scripts/check-schedules.mjs
@@ -326,12 +326,20 @@ await section('纯函数：事件触发的表单 ⇄ 请求体', async () => {
body.trigger_kind === 'event' && JSON.stringify(body.trigger) === JSON.stringify({ event: 'table.version', debounce_s: 300,
source_id: 'ds-sales', period: 'new' }) && body.cron === '' && body.name === '门店月报'
&& JSON.stringify(body.input_template) === '{"date":"{{ event.period }}"}', JSON.stringify(body))
- const wf = m.triggerBody({ event: 'workflow.finished', subjectId: 'wf-daily', period: 'new', appId: 'app-board', originKind: '', debounceS: 0 })
- check('上游工作流：写 workflow_id，带应用、不带期次和空的来源', JSON.stringify(wf) === JSON.stringify({ event: 'workflow.finished',
- debounce_s: 0, workflow_id: 'wf-daily', app_id: 'app-board' }), JSON.stringify(wf))
- const switched = m.switchEvent(ev.trigger, 'source.changed')
+ const wfForm = { event: 'workflow.finished', subjectId: 'wf-daily', period: 'new', appId: 'app-board', originKind: '',
+ includeRejected: false, includeBackfill: false, debounceS: 0 }
+ const wf = m.triggerBody(wfForm)
+ check('上游工作流：写 workflow_id，带应用、不带期次和空的来源，缺省不带「也认」两项', JSON.stringify(wf) === JSON.stringify({
+ event: 'workflow.finished', debounce_s: 0, workflow_id: 'wf-daily', app_id: 'app-board' }), JSON.stringify(wf))
+ const wide = m.triggerBody({ ...wfForm, includeRejected: true, includeBackfill: true })
+ check('勾了也认途经驳回的、应用补算的：各交 true', wide.include_rejected === true && wide.include_backfill === true, JSON.stringify(wide))
+ const onlyBackfill = m.triggerBody({ ...wfForm, originKind: 'app_backfill', includeBackfill: true })
+ check('只认应用补算的：不再另交「也认应用补算的」', onlyBackfill.origin_kind === 'app_backfill' && !('include_backfill' in onlyBackfill),
+ JSON.stringify(onlyBackfill))
+ check('表格的触发条件不带这两项', !('include_rejected' in m.triggerBody({ ...wfForm, event: 'table.version', includeRejected: true })))
+ const switched = m.switchEvent({ ...ev.trigger, includeRejected: true }, 'source.changed')
check('换事件种类：对象、过滤清掉，去抖留着', switched.subjectId === '' && switched.period === 'any' && switched.appId === ''
- && switched.originKind === '' && switched.debounceS === 300 && m.switchEvent(ev.trigger, 'table.version') === ev.trigger)
+ && switched.originKind === '' && switched.includeRejected === false && switched.debounceS === 300 && m.switchEvent(ev.trigger, 'table.version') === ev.trigger)
check('问题：事件触发没选对象时说要选什么，不查频率', m.formProblems({ ...ev, trigger: switched, freq: { ...m.DEFAULT_FREQ, preset: 'weekly',
weekdays: [] } }, null).trigger === '请选择数据源' && !m.formProblems({ ...ev, freq: { ...m.DEFAULT_FREQ, preset: 'weekly', weekdays: [] } },
null).freq)
@@ -625,7 +633,8 @@ const TRIGGER_OPTIONS = {
{ id: 'ds-shop', name: '门店库', ok: true, reason: null },
] },
],
- origins: [{ kind: 'launcher', label: '手动正式运行' }, { kind: 'schedule', label: '定时任务' }, { kind: 'hook', label: '入站钩子' }],
+ origins: [{ kind: 'launcher', label: '手动正式运行' }, { kind: 'schedule', label: '定时任务' },
+ { kind: 'app_backfill', label: '应用补算' }, { kind: 'hook', label: '入站钩子' }],
apps: [{ id: 'app-board', name: '门店看板' }],
default_debounce_s: 60, max_debounce_s: 86400, max_depth: 5,
}
@@ -672,6 +681,10 @@ function eventFires() {
event: { ...EVENT_BRIEF, merged: 0, sample: true, real: false } }),
fire('fe3', { ...none, kind: 'event', status: 'skipped_overlap', fire_at: iso(NOW - 48 * H), created_at: iso(NOW - 48 * H),
reason: '上一次运行 #a1b2c3 仍在进行中，本次触发已跳过', event: { ...EVENT_BRIEF, merged: 1, depth: 2 } }),
+ // 调度停过：三期的事件都过了补发宽限，合成一条跳过（几个窗口、合掉的期次照实写）
+ fire('fe4', { ...none, kind: 'event', status: 'skipped_misfire', fire_at: iso(NOW - 72 * H), created_at: iso(NOW - 70 * H),
+ reason: '2026-10-14 09:00 至 2026-10-14 09:20 的 3 次事件（3 个去抖窗口）：服务当时没有派发（调度未运行或已停止），已超出补发宽限（1 小时），没有发起；积压的事件不补',
+ event: { ...EVENT_BRIEF, merged: 3, windows: 3, periods: ['2026-07', '2026-08', '2026-09'] } }),
]
}
/** 由事件触发的那次运行：来源里记着 trigger（服务端发起时写下的那一句） */
@@ -2336,6 +2349,32 @@ await section('事件触发：新建的请求体', async () => {
check('没有页面报错、原生对话框', !errors.length, errors.join(' ｜ '))
})
+await section('事件触发：上游工作流的请求体（途经驳回、应用补算）', async () => {
+ const { page, sent, errors } = await open('/schedules/new')
+ await page.locator('[data-schedule-form]').waitFor()
+ await page.locator('[data-workflow-picker] input').click()
+ await page.locator('[data-workflow-option="wf-weekly"]').click()
+ await page.locator('[data-trigger-mode="event"]').click()
+ await page.locator('[data-trigger-event="workflow.finished"]').click()
+ await page.locator('[data-trigger-subject="workflow.finished"]').selectOption('wf-daily')
+ const rej = page.locator('[data-trigger-include-rejected]')
+ const back = page.locator('[data-trigger-include-backfill]')
+ check('上游工作流：有「也认途经驳回的」「也认应用补算的」，缺省都不勾', await rej.count() === 1 && await back.count() === 1
+ && !(await rej.isChecked()) && !(await back.isChecked()))
+ check('……说明照实写不勾时这类运行不触发', (await text(page, '[data-trigger-section]')).includes('不勾时这类运行不触发'))
+ await page.locator('[data-trigger-origin]').selectOption('app_backfill')
+ check('……只认应用补算时「也认应用补算的」不再出现', await back.count() === 0)
+ await page.locator('[data-trigger-origin]').selectOption('')
+ await rej.check()
+ await page.locator('[data-input-field="date"] input').fill('{{ event.rejected }}')
+ await page.locator('[data-schedule-save]').click()
+ await until(() => writes(sent, /^POST \/schedules$/).length === 1)
+ const body = writes(sent, /^POST \/schedules$/)[0]?.body
+ check('POST 请求体：勾了的「也认途经驳回的」交 true，没勾的不交', JSON.stringify(body?.trigger) === JSON.stringify({
+ event: 'workflow.finished', debounce_s: 60, workflow_id: 'wf-daily', include_rejected: true }), JSON.stringify(body?.trigger))
+ check('没有页面报错', !errors.length, errors.join(' ｜ '))
+})
+
await section('事件触发：列表、详情与触发历史', async () => {
const state = freshState({ schedules: [structuredClone(S1), structuredClone(S_EVENT), structuredClone(S_EVENT_OFF)], fires: eventFires() })
const list = await open('/schedules', { state })
@@ -2375,6 +2414,9 @@ await section('事件触发：列表、详情与触发历史', async () => {
&& fe2.includes('立即运行，事件为示例值') && !fe2.includes('由事件触发')
&& (await text(page, '[data-fire-id="fe2"]')).includes('还没有发生过这种事件'), fe2)
check('……上一次没结束跳过的写第几层', (await text(page, '[data-fire-id="fe3"] [data-fire-event]')).includes('触发链第 2 层'))
+ const fe4 = await text(page, '[data-fire-id="fe4"] [data-fire-event]')
+ check('……过了宽限合成一条跳过的：写合并了几次、几个窗口、合掉的期次', fe4.includes('合并了 3 次事件') && fe4.includes('3 个去抖窗口')
+ && fe4.includes('期次 2026-07、2026-08、2026-09'), fe4)
await shot(page, 'event-history-dark')
// 390px 窄屏、浅色：事件那一节（三种事件的按钮、对象下拉里带原因的长名字）不撑出横向滚动
const narrow = await open('/schedules/s-event', { state, width: 390, height: 844, theme: 'light' })