事件触发的规模：定时任务列表里每个事件触发任务「有几条在等」一次查齐（outbox.pending_for_many：待派发数一条 SQL 按任务分组算，窗口要看的事件一次读出），不再每行查一次；每一拍派发只读触发条件的事件种类、对象对得上的任务（SQLAlchemy 的通用 JSON 访问），保存时查环一层查一次、只顺着「上游工作流运行成功结束」的边找；选对象（trigger-options）按事件、关键字查、一页最多 100 个、more 说还有没有，当前选中的总在里面，编辑页一页没列完时多一个按名称查找。顺带修：等着的时候任务改了条件、最早那条不再合条件时，「什么时候合成触发」按剩下的最早一条算（列表上的时刻和真发起的一致）；一个窗口的结束时刻和这个任务已有的另一次触发撞在同一刻（别的期次早一拍发了）时往后挪，那几条事件照常发起，不再被记到已有的触发上

backend/app/api/schedules.py

@@ -516,9 +516,9 @@ async def _schedule_outs(session: AsyncSession, rows: list[Schedule]) -> list[di
         from app import scheduler
 
         on = scheduler.scheduler_enabled()
-        for r in rows:
-            if _event_kind(r):
-                states[r.id] = {**await outbox.pending_for(session, r), "scheduler_on": on}
+        # 一次查齐（outbox.pending_for_many），不再每行各查一次
+        waiting = await outbox.pending_for_many(session, [r for r in rows if _event_kind(r)])
+        states = {sid: {**state, "scheduler_on": on} for sid, state in waiting.items()}
 
     def target(r: Schedule) -> dict[str, Any]:
         kind, target_id = service.target_of(r)
@@ -592,36 +592,70 @@ async def list_variables(tz: str | None = Query(default=None, alias="timezone"),
     return service.variables(tz=zone, event=event)
 
 
+#: 选对象时一页最多给几个（按名字排；更多的按关键字查）
+TRIGGER_OPTIONS_LIMIT = 100
+
+
 @router.get("/trigger-options")
-async def trigger_options(session: AsyncSession = Depends(get_session)) -> dict[str, Any]:
+async def trigger_options(event: str | None = None, q: str | None = None, selected: str | None = None,
+                          limit: int = Query(default=TRIGGER_OPTIONS_LIMIT, ge=1, le=500),
+                          session: AsyncSession = Depends(get_session)) -> dict[str, Any]:
     """新建、修改事件触发任务时能选的：每种事件的对象（能不能选、为什么不能，由这里判，界面不自己判）、「只认某种来源」的
     来源、「只认某个应用」的应用、去抖的缺省值、触发链的上限。
 
-    {events: [{event, label, subject_label, objects: [{id, name, ok, reason}]}], origins: [{kind, label}],
+    对象按名字排、一页最多 limit 个，more 说后面还有没有：数据源、工作流成百上千时不一次全给，界面按关键字（q，名字里含它）
+    再查。event 给了只查这一种事件的对象；selected 是当前选中的那一个（修改时），总在里面，不被分页挤掉、不显示成「已不存在」。
+
+    {events: [{event, label, subject_label, objects: [{id, name, ok, reason}], more}], origins: [{kind, label}],
      apps: [{id, name}], default_debounce_s, max_debounce_s, max_depth}"""
     from app.db.models import App, DataSource
 
-    sources = list((await session.execute(select(DataSource).order_by(DataSource.name))).scalars())
-    flows = list((await session.execute(select(Workflow).where(Workflow.is_template.is_(False))
-                                        .order_by(Workflow.name))).scalars())
-    tables = []
-    for src in sources:
-        if (src.origin or "manual") != "upload":
-            continue
-        ok = await triggers.accumulating(session, src)
-        tables.append({"id": src.id, "name": src.name, "ok": ok,
-                       "reason": None if ok else "每次上传替换整张表，没有「新的一期」：要用这个事件请在导入设置里改为按期累积"})
-    objects = {
-        "table.version": tables,
-        "workflow.finished": [{"id": w.id, "name": w.name, "ok": bool(w.published_version),
-                               "reason": None if w.published_version else "还没有发布版本：只有正式运行成功结束才算事件"}
-                              for w in flows],
-        "source.changed": [{"id": src.id, "name": src.name, "ok": True, "reason": None} for src in sources],
-    }
+    if event is not None and event not in triggers.EVENTS:
+        raise _invalid(f"不认识的事件种类：{event}", "event")
+    wanted = [event] if event else list(triggers.EVENTS)
+    words = (q or "").strip()
+
+    async def page(model: Any, *conds: Any) -> tuple[list[Any], bool]:
+        query = select(model).where(*conds)
+        if words:
+            query = query.where(model.name.contains(words, autoescape=True))
+        rows = list((await session.execute(query.order_by(model.name, model.id).limit(limit + 1))).scalars())
+        more = len(rows) > limit
+        rows = rows[:limit]
+        if selected and all(r.id != selected for r in rows):
+            if (picked := (await session.execute(select(model).where(model.id == selected, *conds))).scalar_one_or_none()):
+                rows.append(picked)
+        return rows, more
+
+    async def tables() -> tuple[list[dict[str, Any]], bool]:
+        # 只有上传表格有「新的一期」：按来源筛在库里做（老行没写来源的是手工接的源）
+        rows, more = await page(DataSource, DataSource.origin == "upload")
+        out = []
+        for src in rows:
+            ok = await triggers.accumulating(session, src)
+            out.append({"id": src.id, "name": src.name, "ok": ok,
+                        "reason": None if ok else "每次上传替换整张表，没有「新的一期」：要用这个事件请在导入设置里改为按期累积"})
+        return out, more
+
+    async def flows() -> tuple[list[dict[str, Any]], bool]:
+        rows, more = await page(Workflow, Workflow.is_template.is_(False))
+        return [{"id": w.id, "name": w.name, "ok": bool(w.published_version),
+                 "reason": None if w.published_version else "还没有发布版本：只有正式运行成功结束才算事件"}
+                for w in rows], more
+
+    async def sources() -> tuple[list[dict[str, Any]], bool]:
+        rows, more = await page(DataSource)
+        return [{"id": src.id, "name": src.name, "ok": True, "reason": None} for src in rows], more
+
+    makers = {"table.version": tables, "workflow.finished": flows, "source.changed": sources}
+    entries = []
+    for e in wanted:
+        objects, more = await makers[e]()
+        entries.append({"event": e, "label": triggers.EVENT_LABEL[e], "subject_label": triggers.SUBJECT_LABEL[e],
+                        "objects": objects, "more": more})
     apps = (await session.execute(select(App.id, App.name).where(App.status != "archived").order_by(App.name))).all()
     return {
-        "events": [{"event": e, "label": triggers.EVENT_LABEL[e], "subject_label": triggers.SUBJECT_LABEL[e],
-                    "objects": objects[e]} for e in triggers.EVENTS],
+        "events": entries,
         "origins": [{"kind": k, "label": triggers.ORIGIN_LABEL[k]} for k in triggers.FORMAL_ORIGINS],
         "apps": [{"id": i, "name": n} for i, n in apps],
         "default_debounce_s": triggers.DEFAULT_DEBOUNCE_S, "max_debounce_s": triggers.MAX_DEBOUNCE_S,

backend/app/scheduler/outbox.py

@@ -273,12 +273,10 @@ async def route(now: datetime | None = None) -> int:
                                              .limit(ROUTE_BATCH))).scalars())
         if not events:
             return 0
-        rows = (await session.execute(select(Schedule.id, Schedule.trigger, Schedule.target_kind).where(
-            Schedule.trigger_kind == "event", Schedule.enabled.is_(True)))).all()
-    listeners = [(sid, trigger) for sid, trigger, kind in rows if (kind or "workflow") == "workflow"]
+        listening = await listeners(session, events)
     routed = 0
     for ev in events:
-        matched = [sid for sid, trigger in listeners if triggers.matches(trigger, ev.event, ev.subject_id, ev.payload)]
+        matched = [sid for sid, trigger in listening if triggers.matches(trigger, ev.event, ev.subject_id, ev.payload)]
         try:
             async with SessionLocal() as session:
                 # 条件 UPDATE 认领：两拍撞在一起（以后多开了进程）只有一拍改得动这一行
@@ -298,6 +296,28 @@ async def route(now: datetime | None = None) -> int:
     return routed
 
 
+def _listening_to(pairs: Any) -> Any:
+    """触发条件监听着这几对（事件种类, 对象）的事件触发任务的 WHERE 条件：按触发条件里的事件种类和对象筛（SQLAlchemy 的通用
+    JSON 访问，各库都能用），不把全部事件触发任务读出来。pairs 是 {事件种类: {对象 id…}}。"""
+    from sqlalchemy import and_, false, or_
+
+    conds = [and_(Schedule.trigger["event"].as_string() == kind,
+                  Schedule.trigger[triggers.SUBJECT_KEY[kind]].as_string().in_(sorted(subjects)))
+             for kind, subjects in pairs.items() if kind in triggers.SUBJECT_KEY and subjects]
+    return or_(*conds) if conds else false()
+
+
+async def listeners(session: Any, events: list[TriggerEvent]) -> list[tuple[str, dict[str, Any]]]:
+    """这几条事件可能派给的任务：启用着的、指向工作流的事件触发任务里，触发条件的事件种类和对象对得上的 [(任务 id, 触发条件)]。
+    过滤条件（应用、来源、期次……）由 triggers.matches 逐条再判。"""
+    pairs: dict[str, set[str]] = {}
+    for ev in events:
+        pairs.setdefault(ev.event, set()).add(ev.subject_id)
+    rows = (await session.execute(select(Schedule.id, Schedule.trigger, Schedule.target_kind).where(
+        Schedule.trigger_kind == "event", Schedule.enabled.is_(True), _listening_to(pairs)))).all()
+    return [(sid, trigger) for sid, trigger, kind in rows if (kind or "workflow") == "workflow"]
+
+
 async def fire_waiting(now: datetime | None = None) -> list[ScheduleFire]:
     """有等着的事件的任务，各把到点的去抖窗口处理完。"""
     now = now or utcnow()
@@ -382,6 +402,7 @@ async def _drain(schedule_id: str, now: datetime) -> list[ScheduleFire]:
         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}
+    await _clear_of_earlier_fires(schedule_id, timely)
     out: list[ScheduleFire] = []
     if missed:
         fire = await _skip_missed(schedule_id, missed, tz, grace)
@@ -396,6 +417,32 @@ async def _drain(schedule_id: str, now: datetime) -> list[ScheduleFire]:
     return out
 
 
+async def _clear_of_earlier_fires(schedule_id: str, wins: list[_Window]) -> None:
+    """要发起的窗口的结束时刻（fire_at）和这个任务已经有的另一次事件触发撞在同一刻（别的期次的窗口早一拍发了，结束时刻相同）
+    时往后挪 1 微秒，直到不撞：service.fire_once 按 (任务, fire_at, kind) 认「已经处理过」，撞上了就交回已有的那一条，这个
+    窗口的几条事件会被记到别人的触发上、不再发起。已有的那次触发里有这个窗口的事件（两拍撞在一起处理了同一个窗口）的不挪：
+    那就是同一次，交回已有的那一条是对的。"""
+    from app.scheduler import service
+
+    if not wins:
+        return
+    lo, hi = min(w.due for w in wins), max(w.due for w in wins) + timedelta(seconds=1)
+    async with SessionLocal() as session:
+        rows = (await session.execute(select(ScheduleFire.fire_at, ScheduleFire.inputs).where(
+            ScheduleFire.schedule_id == schedule_id, ScheduleFire.kind == "event",
+            ScheduleFire.fire_at >= lo, ScheduleFire.fire_at <= hi))).all()
+    taken: dict[datetime, set[str]] = {}
+    for at, inputs in rows:
+        brief = (inputs or {}).get("event") if isinstance(inputs, dict) else None
+        ids = brief.get("event_ids") if isinstance(brief, dict) else None
+        taken.setdefault(service.as_utc(at), set()).update(str(i) for i in ids or [])
+    for w in sorted(wins, key=lambda w: w.due):
+        mine = {ev.id for ev in w.events}
+        while w.due in taken and not taken[w.due] & mine:
+            w.due += timedelta(microseconds=1)
+        taken.setdefault(w.due, set()).update(mine)
+
+
 def _event_fire(events: list[TriggerEvent], tz: str, outputs: dict[str, Any]) -> EventFire:
     """几条合成一次的事件：内容取最新的那一条，深度取最深的那一条加 1。"""
     latest = events[-1]
@@ -491,17 +538,57 @@ async def _clean(now: datetime) -> None:
 
 
 async def pending_for(session: Any, schedule: Schedule) -> dict[str, Any]:
-    """一个事件触发任务现在有几条事件在等：events 还没派发的（调度关着时一直留在这里）、waiting 已经派给它、在去抖窗口里的；
-    due_at 是窗口结束、会合成一次触发的时刻（没有在等的为空）。"""
-    trigger = schedule.trigger if isinstance(schedule.trigger, dict) else {}
-    subject, event = triggers.subject_of(trigger), trigger.get("event") or ""
-    rows = (await session.execute(select(TriggerEvent.payload).where(
-        TriggerEvent.status == "pending", TriggerEvent.event == event, TriggerEvent.subject_id == subject)
-        .limit(ROUTE_BATCH))).scalars() if subject else []
-    unrouted = sum(1 for payload in rows if triggers.matches(trigger, event, subject, payload))
-    waiting, first = (await session.execute(select(func.count(), func.min(TriggerDelivery.occurred_at)).where(
-        TriggerDelivery.schedule_id == schedule.id, TriggerDelivery.status == "pending"))).one()
+    """一个事件触发任务现在有几条事件在等（见 pending_for_many）。"""
+    return (await pending_for_many(session, [schedule]))[schedule.id]
+
+
+async def pending_for_many(session: Any, schedules: list[Schedule]) -> dict[str, dict[str, Any]]:
+    """几个事件触发任务各有几条事件在等，一次查齐（定时任务列表每行都要，不再每行各查一次）：任务 id → {events, waiting, due_at}。
+
+    events 是还没派发的（调度关着时一直留在这里）；waiting 是已经派给它、在去抖窗口里的；due_at 是最早那个窗口结束、会合成
+    一次触发的时刻（没有在等的为空）。等着的时候任务改了条件、已经不合条件的（下一拍会标 dropped、不再派发）不算：窗口按剩下的
+    最早一条起算，和派发时（_drain）切的窗口一样。待派发数按任务分组一条 SQL 算；窗口要看事件内容，等着的那几条一次读出来。"""
     from app.scheduler import service
 
-    due = service.as_utc(first) + timedelta(seconds=triggers.debounce_of(trigger)) if first is not None else None
-    return {"events": unrouted, "waiting": int(waiting or 0), "due_at": due.isoformat() if due else None}
+    rows = [s for s in schedules if s is not None]
+    out = {s.id: {"events": 0, "waiting": 0, "due_at": None} for s in rows}
+    if not rows:
+        return out
+    trig = {s.id: s.trigger if isinstance(s.trigger, dict) else {} for s in rows}
+    ids = list(out)
+    counts = dict((await session.execute(select(TriggerDelivery.schedule_id, func.count()).where(
+        TriggerDelivery.schedule_id.in_(ids), TriggerDelivery.status == "pending")
+        .group_by(TriggerDelivery.schedule_id))).all())
+    waiting = (await session.execute(
+        select(TriggerDelivery.schedule_id, TriggerDelivery.id, TriggerDelivery.created_at, TriggerEvent)
+        .join(TriggerEvent, TriggerEvent.id == TriggerDelivery.event_id)
+        .where(TriggerDelivery.schedule_id.in_([i for i in ids if counts.get(i)]), TriggerDelivery.status == "pending")
+        .order_by(TriggerDelivery.occurred_at, TriggerEvent.created_at).limit(DRAIN_BATCH))).all() if counts else []
+    by: dict[str, list[tuple[str, TriggerEvent, datetime]]] = {}
+    stale: dict[str, int] = {}
+    for sid, did, routed, ev in waiting:
+        if triggers.matches(trig[sid], ev.event, ev.subject_id, ev.payload):
+            by.setdefault(sid, []).append((did, ev, service.as_utc(routed)))
+        else:
+            stale[sid] = stale.get(sid, 0) + 1
+    for sid in ids:
+        out[sid]["waiting"] = max(int(counts.get(sid) or 0) - stale.get(sid, 0), 0)
+        wins = windows(by.get(sid, []), triggers.debounce_of(trig[sid]))
+        if wins:
+            out[sid]["due_at"] = min(w.due for w in wins).isoformat()
+    pairs: dict[str, set[str]] = {}
+    for sid in ids:
+        if subject := triggers.subject_of(trig[sid]):
+            pairs.setdefault(str(trig[sid].get("event") or ""), set()).add(subject)
+    if pairs:
+        from sqlalchemy import and_, or_
+
+        cond = or_(*[and_(TriggerEvent.event == kind, TriggerEvent.subject_id.in_(sorted(subs)))
+                     for kind, subs in pairs.items()])
+        unrouted = (await session.execute(select(TriggerEvent.event, TriggerEvent.subject_id, TriggerEvent.payload)
+                                          .where(TriggerEvent.status == "pending", cond).limit(ROUTE_BATCH))).all()
+        for kind, subject, payload in unrouted:
+            for sid in ids:
+                if triggers.matches(trig[sid], kind, subject, payload):
+                    out[sid]["events"] += 1
+    return out

backend/app/scheduler/triggers.py

@@ -248,29 +248,40 @@ async def find_cycle(session: Any, *, schedule_id: str | None, trigger: dict[str
     start, target = subject_of(trigger), target_workflow_id
     if start == target:
         return [start, start]
-    edges: dict[str, set[str]] = {}
-    rows = (await session.execute(select(Schedule.id, Schedule.trigger, Schedule.workflow_id, Schedule.target_kind)
-                                  .where(Schedule.trigger_kind == "event"))).all()
-    for sid, other, workflow_id, kind in rows:
-        if sid == schedule_id or (kind or "workflow") != "workflow" or not isinstance(other, dict):
-            continue
-        if other.get("event") == "workflow.finished" and subject_of(other) and workflow_id:
-            edges.setdefault(subject_of(other), set()).add(workflow_id)
-    # 从 B 出发找回 A 的路（广度优先，取最短的一条说给人听）
-    parent: dict[str, str | None] = {target: None}
-    queue = [target]
-    while queue:
-        node = queue.pop(0)
-        for nxt in sorted(edges.get(node, ())):
-            if nxt in parent:
+
+    async def edges_from(nodes: list[str]) -> dict[str, set[str]]:
+        """这几个工作流跑完会触发哪些工作流：只读「上游工作流运行成功结束」、对象是它们的那几个任务（按触发条件里的事件种类和
+        对象筛，SQLAlchemy 的通用 JSON 访问），不把全部事件触发任务读出来。"""
+        rows = (await session.execute(select(Schedule.id, Schedule.trigger, Schedule.workflow_id, Schedule.target_kind)
+                                      .where(Schedule.trigger_kind == "event",
+                                             Schedule.trigger["event"].as_string() == "workflow.finished",
+                                             Schedule.trigger["workflow_id"].as_string().in_(nodes)))).all()
+        out: dict[str, set[str]] = {}
+        for sid, other, workflow_id, kind in rows:
+            if sid == schedule_id or (kind or "workflow") != "workflow" or not isinstance(other, dict):
                 continue
-            parent[nxt] = node
-            if nxt == start:
-                back = [start]                  # A ← … ← B，倒着记
-                while back[-1] != target:
-                    back.append(parent[back[-1]] or target)
-                return [start, *reversed(back)]
-            queue.append(nxt)
+            if other.get("event") == "workflow.finished" and subject_of(other) and workflow_id:
+                out.setdefault(subject_of(other), set()).add(workflow_id)
+        return out
+
+    # 从 B 出发找回 A 的路（广度优先，一层查一次，取最短的一条说给人听）
+    parent: dict[str, str | None] = {target: None}
+    frontier = [target]
+    while frontier:
+        edges = await edges_from(frontier)
+        nxt_level: list[str] = []
+        for node in frontier:
+            for nxt in sorted(edges.get(node, ())):
+                if nxt in parent:
+                    continue
+                parent[nxt] = node
+                if nxt == start:
+                    back = [start]                  # A ← … ← B，倒着记
+                    while back[-1] != target:
+                        back.append(parent[back[-1]] or target)
+                    return [start, *reversed(back)]
+                nxt_level.append(nxt)
+        frontier = nxt_level
     return None
 
 

backend/tests/test_schedule_events_scale.py

@@ -0,0 +1,171 @@
+"""事件触发的规模（scheduler/outbox.py、scheduler/triggers.py、api/schedules.py）：
+
+- 定时任务列表里每个事件触发任务「有几条在等」一次查齐（outbox.pending_for_many：一条 SQL 按任务分组算待派发数），不再
+  每行各查一次；
+- 每一拍派发只读监听着这几种事件、这几个对象的任务（按触发条件里的事件种类、对象筛，用 SQLAlchemy 的通用 JSON 访问），
+  不再把全部事件触发任务读出来；保存时查环只顺着「上游工作流运行成功结束」那几条边往下找；
+- 选对象（trigger-options）按事件、关键字查、分页：返回 more 说还有没有，当前选中的那一个总在里面；
+- 去抖窗口最早那一条因为任务改了条件被标 dropped 时，「什么时候合成触发」按剩下的最早一条算；一个窗口的结束时刻和这个任务
+  已经有的另一次触发撞在同一刻（别的期次、早一拍发的）时往后挪，那几条事件照常发起，不并进已有的触发。
+工作流、表格、数据源都是虚构的。
+"""
+from __future__ import annotations
+
+from contextlib import contextmanager
+from datetime import timedelta
+
+import pytest
+from httpx import ASGITransport, AsyncClient
+from sqlalchemy import event as sa_event
+from sqlalchemy import select
+
+from app.db.base import SessionLocal, engine, utcnow
+from app.db.models import Schedule, ScheduleFire, Workflow
+from app.db.models_triggers import TriggerDelivery
+from app.scheduler import outbox
+from tests.test_schedule_events import (  # noqa: F401 - engine_up 是本模块也要的 autouse 夹具
+    deliveries, engine_up, event_schedule, fires, put_event, settle, workflow,
+)
+
+
+@contextmanager
+def counted():
+    """数这一段里发了几条 SQL。"""
+    seen: list[str] = []
+
+    def hook(conn, cursor, statement, *_a):  # noqa: ANN001
+        seen.append(statement)
+
+    target = engine.sync_engine
+    sa_event.listen(target, "before_cursor_execute", hook)
+    try:
+        yield seen
+    finally:
+        sa_event.remove(target, "before_cursor_execute", hook)
+
+
+@pytest.fixture
+async def client():
+    from app.main import app
+
+    async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as c:
+        yield c
+
+
+async def _many(n: int) -> tuple[str, list[str]]:
+    upstream = await workflow("门店日报")
+    downstream = await workflow("区域汇总")
+    ids = [await event_schedule(downstream, {"event": "workflow.finished", "workflow_id": upstream, "debounce_s": 600},
+                                name=f"汇总{i}") for i in range(n)]
+    return upstream, ids
+
+
+async def test_waiting_counts_for_many_schedules_in_one_go():
+    upstream, ids = await _many(3)
+    t0 = utcnow().replace(microsecond=0)
+    for i in range(2):
+        await put_event("workflow.finished", upstream, {"workflow_id": upstream, "run_id": f"r{i}"}, t0 + timedelta(seconds=i))
+    await outbox.route(t0 + timedelta(seconds=5))
+    await put_event("workflow.finished", upstream, {"workflow_id": upstream, "run_id": "r9"}, t0 + timedelta(seconds=9))
+    async with SessionLocal() as session:
+        rows = list((await session.execute(select(Schedule).where(Schedule.id.in_(ids)))).scalars())
+        one = {r.id: await outbox.pending_for(session, r) for r in rows}
+        with counted() as sql:
+            many = await outbox.pending_for_many(session, rows)
+    assert many == one
+    assert all(v == {"events": 1, "waiting": 2, "due_at": (t0 + timedelta(seconds=600)).isoformat()} for v in many.values())
+    assert len(sql) <= 3, sql
+
+
+async def test_the_schedule_list_does_not_query_once_per_row(client):
+    await _many(2)
+    with counted() as few:
+        assert (await client.get("/api/schedules")).status_code == 200
+    await _many(5)
+    with counted() as more:
+        assert (await client.get("/api/schedules")).status_code == 200
+    assert len(more) <= len(few), (len(few), len(more))
+
+
+async def test_a_tick_reads_only_the_schedules_listening_to_the_pending_events():
+    upstream, ids = await _many(2)
+    other = await workflow("别的日报")
+    elsewhere = await event_schedule(await workflow("别的汇总"), {"event": "workflow.finished", "workflow_id": other})
+    await put_event("workflow.finished", upstream, {"workflow_id": upstream, "run_id": "r1"}, utcnow())
+    async with SessionLocal() as session:
+        from app.db.models_triggers import TriggerEvent
+
+        pending = list((await session.execute(select(TriggerEvent).where(TriggerEvent.status == "pending"))).scalars())
+        listeners = await outbox.listeners(session, pending)
+    assert sorted(sid for sid, _ in listeners) == sorted(ids) and elsewhere not in {sid for sid, _ in listeners}
+
+
+async def test_trigger_options_search_and_page(client):
+    names = [f"检索样例{c}-{i}" for c in "甲乙" for i in range(3)]
+    async with SessionLocal() as session:
+        rows = [Workflow(name=n, graph={"nodes": [], "edges": []}) for n in names]
+        session.add_all(rows)
+        await session.commit()
+        chosen = next(r.id for r in rows if r.name == "检索样例乙-2")
+    r = await client.get("/api/schedules/trigger-options", params={"event": "workflow.finished", "q": "检索样例甲",
+                                                                   "limit": 2})
+    assert r.status_code == 200, r.text
+    [entry] = r.json()["events"]
+    assert entry["event"] == "workflow.finished" and entry["more"] is True
+    assert [o["name"] for o in entry["objects"]] == ["检索样例甲-0", "检索样例甲-1"]
+    r = await client.get("/api/schedules/trigger-options", params={"event": "workflow.finished", "q": "检索样例甲",
+                                                                   "limit": 2, "selected": chosen})
+    [entry] = r.json()["events"]
+    assert chosen in [o["id"] for o in entry["objects"]], "当前选中的那一个总在里面（不被分页挤掉、不显示成已不存在）"
+    # 不带参数：照旧给全部种类，每种最多一页
+    body = (await client.get("/api/schedules/trigger-options")).json()
+    assert {e["event"] for e in body["events"]} == {"table.version", "workflow.finished", "source.changed"}
+    assert all("more" in e for e in body["events"])
+
+
+async def test_a_dropped_earliest_event_no_longer_sets_the_window_end():
+    """窗口最早那一条因为任务改了条件被标 dropped：什么时候合成触发按剩下的最早一条算（列表上的 due_at、真发起的 fire_at 一致）。"""
+    upstream = await workflow("门店日报")
+    downstream = await workflow("区域汇总")
+    sid = await event_schedule(downstream, {"event": "workflow.finished", "workflow_id": upstream, "debounce_s": 60})
+    t0 = utcnow().replace(microsecond=0)
+    await put_event("workflow.finished", upstream, {"workflow_id": upstream, "run_id": "r1", "origin_kind": "launcher"}, t0)
+    e2 = await put_event("workflow.finished", upstream, {"workflow_id": upstream, "run_id": "r2",
+                                                         "origin_kind": "app_entry"}, t0 + timedelta(seconds=30))
+    await outbox.route(t0 + timedelta(seconds=31))
+    async with SessionLocal() as session:
+        await session.execute(Schedule.__table__.update().where(Schedule.id == sid).values(
+            trigger={"event": "workflow.finished", "workflow_id": upstream, "debounce_s": 60, "origin_kind": "app_entry"}))
+        await session.commit()
+        state = await outbox.pending_for(session, await session.get(Schedule, sid))
+    assert state["due_at"] == (t0 + timedelta(seconds=90)).isoformat(), state
+    assert await outbox.dispatch_due(t0 + timedelta(seconds=61)) == []
+    [fire] = await outbox.dispatch_due(t0 + timedelta(seconds=91))
+    assert fire.fire_at == t0 + timedelta(seconds=90) and fire.inputs["event"]["event_ids"] == [e2]
+    await settle(fire.run_id)
+
+
+async def test_a_window_ending_where_another_fire_already_is_still_fires():
+    """一个窗口的结束时刻和这个任务已有的另一次触发撞在同一刻（别的期次早一拍发了）：往后挪，这几条照常发起，
+    不并进已有的那次触发、不被它吞掉。"""
+    source = "src-虚构"
+    downstream = await workflow("月报")
+    sid = await event_schedule(downstream, {"event": "table.version", "source_id": source, "debounce_s": 60})
+    t0 = utcnow().replace(microsecond=0)
+    aug = await put_event("table.version", source, {"period": "2026-08", "table": "日报"}, t0)
+    [first] = await outbox.dispatch_due(t0 + timedelta(seconds=61))
+    assert first.fire_at == t0 + timedelta(seconds=60) and first.inputs["event"]["event_ids"] == [aug]
+    await settle(first.run_id)
+    # 另一期的事件发生在同一刻，晚到了一拍（比如补写的）：窗口结束同样是 t0+60
+    sep = await put_event("table.version", source, {"period": "2026-09", "table": "日报"}, t0)
+    [second] = await outbox.dispatch_due(t0 + timedelta(seconds=62))
+    assert second.id != first.id and second.inputs["event"]["event_ids"] == [sep]
+    assert second.fire_at > first.fire_at
+    states = {d.event_id: (d.status, d.fire_id) for d in await deliveries(sid)}
+    assert states[sep] == ("fired", second.id)
+    await settle(second.run_id)
+    async with SessionLocal() as session:
+        assert len(list((await session.execute(select(ScheduleFire).where(ScheduleFire.schedule_id == sid))).scalars())) == 2
+    assert len(await fires(sid)) == 2
+    assert [d.status for d in await deliveries(sid)] == ["fired", "fired"]
+    assert TriggerDelivery is not None

frontend/src/api/schedules.ts

@@ -36,8 +36,12 @@ export const schedules = {
    */
   eventVariables: (event: ScheduleEventKind, opts?: RequestOptions) =>
     get<ScheduleVariable[]>(`/schedules/variables${qs({ trigger_kind: 'event', event })}`, opts),
-  /** 新建、修改事件触发任务时能选的对象（能不能选、为什么由服务端判）、来源、应用。老服务端没有，回 404 */
-  triggerOptions: (opts?: RequestOptions) => get<ScheduleTriggerOptions>('/schedules/trigger-options', opts),
+  /**
+   * 新建、修改事件触发任务时能选的对象（能不能选、为什么由服务端判）、来源、应用。老服务端没有，回 404。对象一页有上限
+   * （more 为真时没给完）：按事件种类、名字里的关键字再查；selected 是当前选中的那一个，总在结果里
+   */
+  triggerOptions: (find?: { event?: ScheduleEventKind; q?: string; selected?: string }, opts?: RequestOptions) =>
+    get<ScheduleTriggerOptions>(`/schedules/trigger-options${qs(find)}`, opts),
   /** 立即运行一次：记一条手动触发，上一次还没结束时照样按重叠策略跳过 */
   run: (id: string) => post<ScheduleRunResult>(`${at(id)}/run`),
   /** 触发历史，新的在前。before 是上一页给的 next_before */

frontend/src/lib/termsTrigger.ts

@@ -43,6 +43,10 @@ export const TRIGGER_TEXT = {
   /** 不能选的对象：名字后面接服务端给的原因 */
   subjectNotOk: (name: string, reason: string) => `${name}（${reason}）`,
   subjectGone: (name: string) => `${name}（已不存在）`,
+  /** 对象太多、一页没列完：按名字查找 */
+  subjectSearch: (label: string) => `按名称查找${label}`,
+  subjectSearchHint: '只列出了一部分，输入名称中的字查找',
+  subjectSearchNone: '没有名称中含这些字的',
   optionsLoading: '正在读取可选的对象…',
   optionsFailed: '无法读取可选的对象',
   optionsUnsupported: '服务端暂不支持事件触发，请升级服务端后再使用',

frontend/src/pages/schedules/TriggerEditor.tsx

@@ -73,6 +73,28 @@ function useTriggerOptions(): OptionsState {
   return state
 }
 
+/**
+ * 对象太多、一页没列完（服务端的 more）时按名字查：输入停一下再查，结果只换这一种事件的对象；当前选中的那一个总带上
+ * （selected），不被分页挤掉、不显示成已不存在
+ */
+function useObjectSearch(kind: ScheduleEventKind, selected: string, enabled: boolean) {
+  const [q, setQ] = useState('')
+  const [found, setFound] = useState<{ key: string; objects: ScheduleTriggerOptions['events'][number]['objects'] } | null>(null)
+  const words = q.trim()
+  const key = `${kind}\u0000${words}`
+  useEffect(() => {
+    if (!enabled || !words) return
+    const controller = new AbortController()
+    const timer = setTimeout(() => {
+      api.schedules.triggerOptions({ event: kind, q: words, selected: selected || undefined }, { signal: controller.signal })
+        .then((data) => setFound({ key, objects: data.events.find((e) => e.event === kind)?.objects ?? [] }))
+        .catch(() => {})
+    }, 300)
+    return () => { clearTimeout(timer); controller.abort() }
+  }, [enabled, kind, words, selected, key])
+  return { q, setQ, objects: words && found?.key === key ? found.objects : null }
+}
+
 export function EventSection({ value, onChange, timezone, setTimezone, savedName, error, debounceError }: {
   value: TriggerForm
   onChange: (t: TriggerForm) => void
@@ -91,7 +113,9 @@ export function EventSection({ value, onChange, timezone, setTimezone, savedName
   const kind = value.event
   const label = TT.subject[kind]
   const entry = options.status === 'ok' ? options.data.events.find((e) => e.event === kind) : undefined
-  const objects = entry?.objects ?? []
+  const paged = !!entry?.more
+  const search = useObjectSearch(kind, value.subjectId, paged)
+  const objects = search.objects ?? entry?.objects ?? []
   const listed = objects.some((o) => o.id === value.subjectId)
 
   let picker
@@ -112,7 +136,10 @@ export function EventSection({ value, onChange, timezone, setTimezone, savedName
               aria-describedby={error ? 'schedule-trigger-subject-error' : undefined}
               onChange={(e) => set({ subjectId: e.target.value })}>
         <option value="">{objects.length ? TT.subjectPlaceholder(label) : TT.subjectEmpty[kind]}</option>
-        {value.subjectId && !listed && <option value={value.subjectId}>{TT.subjectGone(savedName || value.subjectId)}</option>}
+        {/* 一页没列完时不在列表里不等于不存在：写保存时记下的名字，不标「已不存在」 */}
+        {value.subjectId && !listed && (
+          <option value={value.subjectId}>{paged ? savedName || value.subjectId : TT.subjectGone(savedName || value.subjectId)}</option>
+        )}
         {objects.map((o) => (
           <option key={o.id} value={o.id} disabled={!o.ok && o.id !== value.subjectId} data-trigger-object-ok={String(o.ok)}>
             {o.ok || !o.reason ? o.name : TT.subjectNotOk(o.name, o.reason)}
@@ -140,7 +167,17 @@ export function EventSection({ value, onChange, timezone, setTimezone, savedName
       </div>
 
       <Field label={label} htmlFor="schedule-trigger-subject" required>
+        {paged && (
+          <input type="search" className="field mb-1.5" value={search.q} data-trigger-subject-search=""
+                 placeholder={TT.subjectSearch(label)} aria-label={TT.subjectSearch(label)}
+                 aria-describedby="schedule-trigger-subject-search-hint" onChange={(e) => search.setQ(e.target.value)} />
+        )}
         {picker}
+        {paged && (
+          <p id="schedule-trigger-subject-search-hint" className="mt-1 text-2xs leading-relaxed text-faint">
+            {search.objects && !search.objects.length ? TT.subjectSearchNone : TT.subjectSearchHint}
+          </p>
+        )}
       </Field>
       {error && (
         <div id="schedule-trigger-subject-error" className="-mt-1.5 text-2xs leading-relaxed text-[var(--err)]" role="alert"

frontend/src/types.ts

@@ -4172,6 +4172,8 @@ export interface ScheduleTriggerOptions {
     label: string
     subject_label: string
     objects: { id: string; name: string; ok: boolean; reason: string | null }[]
+    /** 一页没给完（对象按名字排、一页有上限）：界面按关键字再查。老服务端不给，当作给完了 */
+    more?: boolean
   }[]
   origins: { kind: string; label: string }[]
   apps: { id: string; name: string }[]

scripts/check-schedules.mjs

@@ -2349,6 +2349,41 @@ await section('事件触发：新建的请求体', async () => {
   check('没有页面报错、原生对话框', !errors.length, errors.join(' ｜ '))
 })
 
+await section('事件触发：对象太多时按名称查找', async () => {
+  // 服务端一页没给完（more）：选择框上面多一个查找框，输入后按事件种类、关键字再查，结果只换这一种事件的对象
+  const many = { ...TRIGGER_OPTIONS, events: TRIGGER_OPTIONS.events.map((e) => (e.event === 'workflow.finished'
+    ? { ...e, more: true } : e)) }
+  const queries = []
+  const override = (method, path, _body, url) => {
+    if (method !== 'GET' || path !== '/schedules/trigger-options') return null
+    const q = url.searchParams.get('q')
+    if (!q) return { json: many }
+    queries.push({ event: url.searchParams.get('event'), q })
+    return { json: { ...many, events: [{ event: 'workflow.finished', label: '上游工作流运行成功结束', subject_label: '工作流',
+      more: false, objects: [{ id: 'wf-far', name: `${q}周报（第 301 个）`, ok: true, reason: null }] }] } }
+  }
+  const { page, errors } = await open('/schedules/new', { state: { ...freshState(), override } })
+  await page.locator('[data-schedule-form]').waitFor()
+  await page.locator('[data-trigger-mode="event"]').click()
+  await page.locator('[data-trigger-event="workflow.finished"]').click()
+  const search = page.locator('[data-trigger-subject-search]')
+  await search.waitFor()
+  check('一页没列完：选择框上面有按名称查找', await search.isVisible())
+  check('上传表格一页列完了：没有查找框', await (async () => {
+    await page.locator('[data-trigger-event="table.version"]').click()
+    const none = await page.locator('[data-trigger-subject-search]').count() === 0
+    await page.locator('[data-trigger-event="workflow.finished"]').click()
+    return none
+  })())
+  await page.locator('[data-trigger-subject-search]').fill('华东')
+  const far = page.locator('[data-trigger-subject="workflow.finished"] option[value="wf-far"]')
+  await far.waitFor({ state: 'attached', timeout: 5000 }).catch(() => {})
+  check('按关键字查到的对象换进选择框', await far.count() === 1, String(await far.count()))
+  check('查的是这一种事件、带着关键字', queries.some((x) => x.event === 'workflow.finished' && x.q === '华东'),
+    JSON.stringify(queries))
+  check('没有未捕获的运行时错误', errors.length === 0, errors[0] ?? '')
+})
+
 await section('事件触发：上游工作流的请求体（途经驳回、应用补算）', async () => {
   const { page, sent, errors } = await open('/schedules/new')
   await page.locator('[data-schedule-form]').waitFor()