"""定时任务：按周期自动发起工作流运行，结束、失败、停在审批时推送 Webhook，运行成功结束后和同一任务上一次成功的运行
自动对比。

**目标种类**（docs/design/next-round/00-总方案.md X4：定时只有一套）：target_kind 是 workflow（缺省，字段和行为一字不变）、
eval_suite（跑一次评测题集）或 source_maintenance（维护一个数据源）。非工作流目标写 target_id，不写 workflow_id、输入模板、
运行类别和 Webhook：核对目标交给各自的模块（scheduler/service.check_target → app/eval/scheduled、app/data/source_maintenance）；
它们的通知走通知渠道的订阅（定时评测有退步、定时评测没有评完……），不走定时任务的 Webhook——Webhook 的三种事件（运行
结束、运行失败、等待审批）说的是这一次发起的那次运行，非工作流目标没有那一次运行，配上了也收不到，所以照实拒收。接口返回带
target_kind、target_id、target_name（目标现在的名字，不在了用保存时记下的那一份）。

**触发方式**（docs/design/orchestration-round/00-总方案.md §4，一套定时机制）：trigger_kind 是 cron（缺省，字段和行为一字不变）
或 event（由数据事件触发，规则在 scheduler/triggers.py、派发在 scheduler/outbox.py）。event 时请求体写 trigger（事件种类、对象、
过滤、去抖），目标只能是工作流，cron 存空串、下一次触发为空；核对对象、输入模板里的 {{ event.* }}、workflow.finished 串成的环
（A→B→A 拒收，说清是哪几个）都在保存时做。触发方式建好之后不能改。Webhook 照旧可配。接口返回另带 trigger_kind、trigger、
trigger_text（给人看的触发说明，事件触发时也是 description_text）、trigger_state（几条事件在等、去抖到什么时候、调度开没开）。

形状按前后端约定（Schedule、Fire 两种对象，见 _schedule_out、_fire_outs）。Webhook 地址和签名密钥加密存放，接口
只给主机和末 4 位（url_hint）、有没有密钥（has_secret）。请求体自己解析、自己校验：错误一律回中文 detail，表单
据 field 把报错落到那一格（422 {detail, code, field}，同 api/coded.py）。后台的调度循环在 app/scheduler 里。
"""
from __future__ import annotations

import re
from datetime import datetime, timedelta
from typing import Any

from fastapi import APIRouter, Depends, Header, HTTPException, Query, Request, Response
from sqlalchemy import and_, delete, func, select, update
from sqlalchemy.ext.asyncio import AsyncSession

from app.api.coded import CodedHTTPException
from app.api.runs import actor_of
from app.core.crypto import decrypt, encrypt
from app.db.base import get_session, utcnow
from app.db.models import Run, Schedule, ScheduleFire, WebhookDelivery, Workflow, WorkflowVersion
from app.scheduler import auto_compare, cron, outbox, service, triggers, webhook

router = APIRouter(prefix="/api/schedules", tags=["schedules"])

#: 请求不合规（422）的机器码。field 是出错的那一项：name、target_kind、target_id、workflow_id、cron、timezone、run_class、
#: input_template、enabled、overlap、misfire_grace_s、webhook、version、limit、before、trigger_kind、trigger、debounce_s
SCHEDULE_INVALID = "schedule_invalid"
#: 非工作流目标配了 Webhook（见模块说明）
NO_WEBHOOK = ("目标是「{kind}」的定时任务不推 Webhook：Webhook 的事件说的是定时发起的那次运行。要收这类定时的通知，"
              "请在「设置 → 通知」的渠道里订阅对应的事件")
_GONE = "定时任务不存在，可能已被删除"
#: 错过的触发在多少秒内补发：默认 1 小时，最长 7 天
DEFAULT_GRACE_S = 3600
MAX_GRACE_S = 7 * 24 * 3600
#: 预览最多给几次
MAX_PREVIEW = 20


def _invalid(detail: str, field: str | None = None) -> CodedHTTPException:
    return CodedHTTPException(422, detail, SCHEDULE_INVALID, field=field)


def _iso(moment: datetime | None) -> str | None:
    return service.as_utc(moment).isoformat() if moment is not None else None


#: 查询串把「+」解成空格之后，时间末尾的时区偏移：「…03.376320 00:00」
_SPACED_OFFSET = re.compile(r"(.*\d) (\d{2}:?\d{2})")


def _cursor(moment: datetime) -> str:
    """触发历史的翻页游标：UTC 写成 Z 结尾、精确到微秒。写成 +00:00 时照接口说明原样拼进地址，「+」在查询串里会被
    解成空格，回 422（2026-10 验收）；Z 结尾的放进地址不用编码，和运行列表的游标一个写法。"""
    return service.as_utc(moment).isoformat().replace("+00:00", "Z")


def _cursor_time(before: str) -> datetime:
    """before → 时间：Z、+00:00 都认，也认「+」被查询串解成空格的「 00:00」（老的游标照原样拼进地址时就是这样）。
    不是时间的照样抛 ValueError。"""
    text = before.strip()
    try:
        return datetime.fromisoformat(text.replace("Z", "+00:00"))
    except ValueError:
        spaced = _SPACED_OFFSET.fullmatch(text)
        if spaced is None:
            raise
        return datetime.fromisoformat(f"{spaced[1]}+{spaced[2]}")


async def _body(request: Request) -> dict[str, Any]:
    try:
        data = await request.json()
    except ValueError:
        data = None
    if not isinstance(data, dict):
        raise _invalid("请求内容需要是 JSON 对象")
    return data


# --------------------------------------------------------------------------
# 逐项校验：返回规范化之后的值，不对抛 422（带 field）
# --------------------------------------------------------------------------


def _name(value: Any) -> str:
    text = value.strip() if isinstance(value, str) else ""
    if not text:
        raise _invalid("请填写定时任务的名称", "name")
    if len(text) > 200:
        raise _invalid("定时任务的名称不能超过 200 个字符", "name")
    return text


def _workflow_id(value: Any) -> str:
    if not isinstance(value, str) or not value.strip():
        raise _invalid("请选择要定时运行的工作流", "workflow_id")
    return value.strip()


def _target_kind(value: Any) -> str:
    if value not in service.TARGET_KINDS:
        labels = "、".join(service.TARGET_LABEL[k] for k in service.TARGET_KINDS)
        raise _invalid(f"定时任务的目标需要是{labels}之一{_current(value)}", "target_kind")
    return value


def _target_id(value: Any, kind: str) -> str:
    if not isinstance(value, str) or not value.strip():
        raise _invalid(f"请选择要定时的{service.TARGET_LABEL[kind]}", "target_id")
    return value.strip()


def _target_extras(body: dict[str, Any], kind: str) -> None:
    """非工作流目标不写工作流、输入模板、Webhook（见模块说明）；运行类别带了也只认合法的写法，不存。"""
    label = service.TARGET_LABEL[kind]
    if not _blank(body.get("workflow_id")):
        raise _invalid(f"目标是「{label}」的定时任务不指定工作流", "workflow_id")
    if body.get("input_template") not in (None, {}):
        raise _invalid(f"目标是「{label}」的定时任务没有输入模板", "input_template")
    hook = body.get("webhook")
    if isinstance(hook, dict) and not _blank(hook.get("url")):
        raise _invalid(NO_WEBHOOK.format(kind=label), "webhook")
    if hook is not None and not isinstance(hook, dict):
        raise _invalid("Webhook 设置需要是一个对象，或者为空表示不推送", "webhook")
    if "run_class" in body:
        _run_class(body["run_class"])


def _cron(value: Any) -> str:
    try:
        return cron.parse(value if isinstance(value, str) else "").expr
    except cron.CronError as e:
        raise _invalid(str(e), "cron") from e


def _timezone(value: Any) -> str:
    try:
        return cron.timezone_name(value if isinstance(value, str) else "")
    except cron.TimezoneError as e:
        raise _invalid(str(e), "timezone") from e


def _run_class(value: Any) -> str:
    if value not in service.RUN_CLASSES:
        raise _invalid(f"运行类别需要选择正式运行或探索运行{_current(value)}", "run_class")
    return value


def _template(value: Any, event: str | None = None) -> dict[str, str]:
    try:
        return service.check_template(value, event=event)
    except service.TemplateError as e:
        raise _invalid(str(e), "input_template") from e


#: 触发方式：按 cron 到点、由数据事件触发
TRIGGER_KINDS = ("cron", "event")


def _trigger_kind(value: Any) -> str:
    if value not in TRIGGER_KINDS:
        raise _invalid(f"触发方式需要是定时或事件触发{_current(value)}", "trigger_kind")
    return value


def _trigger(value: Any) -> dict[str, Any]:
    try:
        return triggers.normalize(value)
    except triggers.TriggerError as e:
        raise _invalid(str(e), e.field) from e


async def _check_trigger(session: AsyncSession, trigger: dict[str, Any], *, schedule_id: str | None,
                         target_workflow_id: str) -> triggers.Checked:
    """核对触发条件里的对象，再查 workflow.finished 串没串成环。不对抛 422（field=trigger）。"""
    try:
        checked = await triggers.check(session, trigger)
    except triggers.TriggerError as e:
        raise _invalid(str(e), e.field) from e
    cycle = await triggers.find_cycle(session, schedule_id=schedule_id, trigger=trigger,
                                      target_workflow_id=target_workflow_id)
    if cycle:
        names = dict((await session.execute(select(Workflow.id, Workflow.name).where(Workflow.id.in_(set(cycle))))).all())
        raise _invalid(triggers.cycle_text([names.get(w, w) for w in cycle]), "trigger")
    return checked


def _current(value: Any) -> str:
    return service.current_value(value)


def _enabled(value: Any) -> bool:
    if not isinstance(value, bool):
        raise _invalid(f"启用状态需要是 true 或 false{_current(value)}", "enabled")
    return value


def _overlap(value: Any) -> str:
    if value != "skip":
        raise _invalid(f"重叠处理目前只支持「跳过」：上一次运行还没结束时跳过这次触发{_current(value)}", "overlap")
    return value


def _grace(value: Any) -> int:
    if isinstance(value, bool) or not isinstance(value, int) or not 0 <= value <= MAX_GRACE_S:
        raise _invalid(f"补发宽限需要是 0 到 {MAX_GRACE_S} 之间的整数秒数（最长 7 天）{_current(value)}",
                       "misfire_grace_s")
    return value


def _format(value: Any) -> str:
    if value not in webhook.FORMATS:
        labels = "、".join(webhook.FORMAT_LABEL[f] for f in webhook.FORMATS)
        raise _invalid(f"Webhook 格式需要是{labels}之一{_current(value)}", "webhook")
    return value


def _events(value: Any) -> list[str]:
    labels = "、".join(webhook.EVENT_LABEL[e] for e in webhook.EVENTS)
    if not isinstance(value, list):
        raise _invalid(f"Webhook 的通知事件需要从{labels}中选择", "webhook")
    unknown = [e for e in value if e not in webhook.EVENTS]
    if unknown:
        first = unknown[0]
        named = f"「{first}」" if isinstance(first, str) and first.strip() else "中有一项"
        raise _invalid(f"Webhook 的通知事件{named}不可选，只能从{labels}中选择", "webhook")
    return [e for e in webhook.EVENTS if e in value]


def _url(value: Any) -> str:
    try:
        return webhook.check_url(value)
    except ValueError as e:
        raise _invalid(str(e), "webhook") from e


def _secret(value: Any) -> str | None:
    if value is None or (isinstance(value, str) and not value.strip()):
        return None
    if not isinstance(value, str):
        raise _invalid("Webhook 签名密钥需要是文本", "webhook")
    return value.strip()


def _blank(value: Any) -> bool:
    return value is None or (isinstance(value, str) and not value.strip())


def _matched_format(value: Any, url: str) -> str:
    """校验格式，并要求和地址对得上：群机器人的地址只认自己平台的格式（见 webhook.bot_format）。"""
    fmt = _format(value)
    if problem := webhook.format_problem(fmt, url):
        raise _invalid(problem, "webhook")
    return fmt


def _new_webhook(value: Any) -> dict[str, Any] | None:
    """创建时的 webhook：null 或 url 为空表示不配；否则校验格式、地址、密钥、事件（事件省略表示三种都要）。
    格式省略时按地址认：群机器人的地址用它平台的格式，别的用通用格式。"""
    if value is None:
        return None
    if not isinstance(value, dict):
        raise _invalid("Webhook 设置需要是一个对象，或者为空表示不推送", "webhook")
    if _blank(value.get("url")):
        return None
    url = _url(value.get("url"))
    given = value.get("format")
    fmt = _matched_format(given, url) if given is not None else webhook.bot_format(url) or "generic"
    return {"format": fmt, "url": url, "secret": _secret(value.get("secret")),
            "events": _events(value.get("events", list(webhook.EVENTS)))}


def _webhook_columns(hook: dict[str, Any] | None) -> dict[str, Any]:
    if hook is None:
        return {"webhook_format": None, "webhook_url": None, "webhook_secret": None, "webhook_events": []}
    return {"webhook_format": hook["format"], "webhook_url": encrypt(hook["url"]),
            "webhook_secret": encrypt(hook["secret"]) if hook["secret"] else None, "webhook_events": hook["events"]}


def _patched_webhook(row: Schedule, value: Any) -> dict[str, Any]:
    """PATCH 的 webhook → 要写的列。null 或 url 写空串：删除整个 Webhook；url 省略：地址不变；secret 省略：密钥不变、
    写空串：去掉密钥。format 省略：换了地址的按新地址认（群机器人用它平台的格式），没换的不变。

    地址或格式真的变了才查两者对不对得上：库里已有对不上的设置（这条校验之前存的）时，整份表单回写只为停用、
    改名，不能因为 Webhook 被拦下。"""
    if value is None:
        return _webhook_columns(None)
    if not isinstance(value, dict):
        raise _invalid("Webhook 设置需要是一个对象，或者为空表示不推送", "webhook")
    if "url" in value and _blank(value.get("url")):
        return _webhook_columns(None)
    has_url = bool(row.webhook_url)
    if "url" not in value and not has_url:
        raise _invalid("请填写 Webhook 地址", "webhook")
    url = _url(value.get("url")) if "url" in value else None
    if value.get("format") is not None:
        fmt = _format(value["format"])
    elif url is not None:
        fmt = webhook.bot_format(url) or (row.webhook_format if has_url else "generic")
    else:
        fmt = row.webhook_format
    if url is not None or fmt != row.webhook_format:
        target = url or decrypt(row.webhook_url)
        if target and (problem := webhook.format_problem(fmt, target)):
            raise _invalid(problem, "webhook")
    out: dict[str, Any] = {
        "webhook_format": _format(fmt),
        "webhook_events": _events(value.get("events", list(row.webhook_events or []) if has_url
                                            else list(webhook.EVENTS))),
    }
    if url is not None:
        out["webhook_url"] = encrypt(url)
    if "secret" in value:
        secret = _secret(value.get("secret"))
        out["webhook_secret"] = encrypt(secret) if secret else None
    return out


async def _check_workflow(session: AsyncSession, workflow_id: str, run_class: str, template: dict[str, str]) -> Workflow:
    """工作流要在；正式运行要有发布版本；模板里的字段名对得上入口字段，必填项都有（按这次要用的那张图查）。"""
    workflow = await session.get(Workflow, workflow_id)
    if workflow is None:
        raise _invalid("工作流不存在，可能已被删除", "workflow_id")
    graph: Any = workflow.graph
    if run_class == "formal":
        if not workflow.published_version:
            raise _invalid(f"「{workflow.name}」还没有发布版本，正式运行只能从发布版本发起。"
                           "请先在编排页发布，或改为探索运行", "run_class")
        graph = await _published_graph(session, workflow) or graph
    try:
        service.check_fields(template, service.input_fields(graph), workflow.name)
    except service.TemplateError as e:
        raise _invalid(str(e), "input_template") from e
    return workflow


async def _published_graph(session: AsyncSession, workflow: Workflow) -> Any:
    return await session.scalar(select(WorkflowVersion.graph).where(
        WorkflowVersion.workflow_id == workflow.id, WorkflowVersion.version == workflow.published_version))


# --------------------------------------------------------------------------
# 对外的形状
# --------------------------------------------------------------------------


def _description(row: Schedule) -> str:
    try:
        return cron.describe(row.cron, row.timezone)
    except cron.CronError:
        return f"按 cron 表达式 {row.cron}（{row.timezone}）"


def _event_kind(row: Schedule) -> bool:
    return (getattr(row, "trigger_kind", None) or "cron") == "event"


async def _trigger_names(session: AsyncSession, rows: list[Schedule]) -> dict[str, str]:
    """事件触发任务的对象、过滤里的应用现在叫什么：{id: 名字}。不在了的不在里面（说法里用保存时的名字）。"""
    from app.db.models import App, DataSource

    wanted: dict[str, set[str]] = {"source": set(), "workflow": set(), "app": set()}
    for r in rows:
        trigger = r.trigger if _event_kind(r) and isinstance(r.trigger, dict) else None
        if not trigger:
            continue
        subject = triggers.subject_of(trigger)
        wanted["workflow" if trigger.get("event") == "workflow.finished" else "source"].add(subject)
        if trigger.get("app_id"):
            wanted["app"].add(trigger["app_id"])
    out: dict[str, str] = {}
    for model, ids in ((DataSource, wanted["source"]), (Workflow, wanted["workflow"]), (App, wanted["app"])):
        if ids:
            out |= dict((await session.execute(select(model.id, model.name).where(model.id.in_(ids)))).all())
    return out


def _trigger_text(row: Schedule, names: dict[str, str]) -> str:
    trigger = row.trigger if isinstance(row.trigger, dict) else {}
    subject = triggers.subject_of(trigger)
    stored = (trigger.get("subject_name") if isinstance(trigger.get("subject_name"), str) else None)
    return triggers.describe(trigger, names.get(subject) or stored, app_name=names.get(trigger.get("app_id") or ""))


def _webhook_out(row: Schedule) -> dict[str, Any] | None:
    if not row.webhook_url:
        return None
    plain = decrypt(row.webhook_url)
    return {
        "format": row.webhook_format,
        # 解不开（加密密钥换过）时不猜：照实说，请人重新填
        "url_hint": webhook.url_hint(plain) if plain else "（地址无法解密，请重新填写）",
        "has_secret": bool(row.webhook_secret),
        "events": [e for e in webhook.EVENTS if e in (row.webhook_events or [])],
    }


async def _fire_outs(session: AsyncSession, fires: list[ScheduleFire]) -> list[dict[str, Any]]:
    """Fire 对象：run_status 是运行现在的状态，notified 是这次触发的推送记录 [{event, status, at}]（按时间先后），
    compare 是运行成功结束后和同一任务上一次成功的运行的自动对比 {base_run_id, summary, previous}（summary 同对比接口
    的计数，previous 说基准是不是紧挨着的上一次），没有为 null。

    对比和比的时候两边的封存绑定（auto_compare）：哪一边现在不在终态（被继续执行）、被删掉、清单哈希和比的时候不一样
    （重新封存过）都给 null——对比页按现在的封存现算，记下的计数和点进去看到的会对不上，或者干脆比不了。工件读不出来、
    被改过的同样给 null。基准已经不是现在挑出来的那一次（auto_compare.outdated：中间隔着的运行事后接着运行成功了）也
    给 null：运行详情里同名的入口比的是另一次。

    inputs 只在目标是数据源维护时给：{steps, jobs: [{step, kind, id, tables, scope?}], skipped: [{step, reason}]}——这一次
    建了哪几个探查、剖析任务，哪几步没执行和原因（app/data/source_maintenance 记在触发上的那一份）。维护没有运行，触发历史、
    「上次触发」和立即运行的答复靠它写出建了什么，不写「未发起运行」。别的目标给 null：工作流触发的 inputs 是按模板渲染的
    工作流输入，触发历史不列。"""
    maintenance = set((await session.execute(select(Schedule.id).where(
        Schedule.id.in_({f.schedule_id for f in fires}), Schedule.target_kind == "source_maintenance"))).scalars()) \
        if fires else set()

    def inputs_of(f: ScheduleFire) -> dict[str, Any] | None:
        if f.schedule_id not in maintenance:
            return None
        raw = f.inputs if isinstance(f.inputs, dict) else {}
        return {key: raw[key] for key in ("steps", "jobs", "skipped") if key in raw}

    def event_of(f: ScheduleFire) -> dict[str, Any] | None:
        """事件触发的那几次（派发的、事件触发任务的立即运行）：事件的说明（scheduler/outbox.EventFire.brief），没有为 null。"""
        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", "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()))
    stale = await auto_compare.outdated(session, compares)
    run_ids = {f.run_id for f in fires if f.run_id} | {brief["base_run_id"] for brief in compares.values()}
    # 运行现在的状态和清单哈希：run_status 照写，对比按它核对两边的封存
    current = {run_id: (status, seal) for run_id, status, seal in (await session.execute(
        select(Run.id, Run.status, Run.manifest_hash).where(Run.id.in_(run_ids)))).all()} if run_ids else {}

    def compare_of(f: ScheduleFire) -> dict[str, Any] | None:
        artifact = compared.get(f.id) or ""
        brief = compares.get(artifact)
        if brief is None or brief["target_run_id"] != f.run_id or artifact in stale:
            return None
        for run_id, seal in ((brief["target_run_id"], brief["target_manifest"]),
                             (brief["base_run_id"], brief["base_manifest"])):
            status, now_seal = current.get(run_id, (None, None))
            if status not in auto_compare.TERMINAL or now_seal != seal:
                return None
        return auto_compare.public(brief)

    notified: dict[str, list[dict[str, Any]]] = {}
    if fires:
        rows = (await session.execute(
            select(WebhookDelivery).where(WebhookDelivery.fire_id.in_([f.id for f in fires]))
            .order_by(WebhookDelivery.created_at, WebhookDelivery.id)
        )).scalars()
        for d in rows:
            notified.setdefault(d.fire_id, []).append({"event": d.event, "status": d.status, "at": _iso(d.created_at)})
    # 正在发起的触发：库里是占位的「未能发起」（为了进程被强杀时照实说），进程还在、只是还没发起完，按已发起给
    launching = {f.id for f in fires if service.launching(f.id, f)}
    return [{
        "id": f.id,
        "fire_at": _iso(f.fire_at),
        "kind": f.kind,
        "status": "started" if f.id in launching else f.status,
        "run_id": f.run_id,
        "run_status": current.get(f.run_id, (None, None))[0] if f.run_id else None,
        # 目标是评测题集时这一次触发建的那次评测（没有运行 id）
        "eval_run_id": f.eval_run_id,
        "inputs": inputs_of(f),
        # 事件触发的那几次：事件的说明（种类、对象、给人看的一句、合并了几次、触发深度、是不是立即运行的示例）
        "event": event_of(f),
        "reason": None if f.id in launching else f.reason,
        "notified": notified.get(f.id, []),
        "compare": compare_of(f),
        "created_at": _iso(f.created_at),
    } for f in fires]


async def _schedule_outs(session: AsyncSession, rows: list[Schedule]) -> list[dict[str, Any]]:
    ids = [r.id for r in rows]
    names: dict[str, str] = {}
    targets: dict[str, str | None] = {}
    last: dict[str, dict[str, Any]] = {}
    if rows:
        names = dict((await session.execute(
            select(Workflow.id, Workflow.name).where(Workflow.id.in_({r.workflow_id for r in rows if r.workflow_id})))).all())
        for r in rows:
            kind, target_id = service.target_of(r)
            if kind != "workflow":
                targets[r.id] = await service.target_name(session, kind, target_id)
        newest = (select(ScheduleFire.schedule_id, func.max(ScheduleFire.created_at).label("at"))
                  .where(ScheduleFire.schedule_id.in_(ids)).group_by(ScheduleFire.schedule_id).subquery())
        fires = list((await session.execute(select(ScheduleFire).join(newest, and_(
            ScheduleFire.schedule_id == newest.c.schedule_id, ScheduleFire.created_at == newest.c.at)))).scalars())
        for out, fire in zip(await _fire_outs(session, fires), fires):
            last.setdefault(fire.schedule_id, out)
    trigger_names = await _trigger_names(session, rows) if rows else {}
    states: dict[str, dict[str, Any]] = {}
    if any(_event_kind(r) for r in rows):
        from app import scheduler

        on = scheduler.scheduler_enabled()
        # 一次查齐（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)
        name = names.get(r.workflow_id) if kind == "workflow" else targets.get(r.id)
        # 目标还在就给它现在的名字；删掉之后用保存时记下的那一份
        return {"target_kind": kind, "target_id": target_id, "target_label": service.TARGET_LABEL.get(kind, kind),
                "target_name": name or r.workflow_name, "target_exists": name is not None}

    def trigger(r: Schedule) -> dict[str, Any]:
        """触发方式：cron 的 trigger、trigger_text、trigger_state 为 null；事件触发的给规范化后的触发条件、说法、在等的事件。"""
        if not _event_kind(r):
            return {"trigger_kind": "cron", "trigger": None, "trigger_text": None, "trigger_state": None}
        # subject_name 是保存时记下的对象名（对象删掉之后界面写它）；带回来时不认（triggers.normalize）
        stored = dict(r.trigger or {}) if isinstance(r.trigger, dict) else {}
        return {"trigger_kind": "event", "trigger": stored, "trigger_text": _trigger_text(r, trigger_names),
                "trigger_state": states.get(r.id)}

    return [{
        "id": r.id,
        "name": r.name,
        **target(r),
        **trigger(r),
        "workflow_id": r.workflow_id,
        # 工作流还在就给它现在的名字；删掉之后用保存时记下的那一份。非工作流目标这一项是目标的名字（同 target_name）
        "workflow_name": names.get(r.workflow_id) or targets.get(r.id) or r.workflow_name,
        "cron": r.cron,
        "timezone": r.timezone,
        "run_class": r.run_class,
        "input_template": dict(r.input_template or {}),
        "enabled": r.enabled,
        "overlap": r.overlap,
        "misfire_grace_s": r.misfire_grace_s,
        "webhook": _webhook_out(r),
        "next_run_at": _iso(r.next_run_at),
        "last_fire": last.get(r.id),
        "description_text": _trigger_text(r, trigger_names) if _event_kind(r) else _description(r),
        "created_by": r.created_by,
        "created_at": _iso(r.created_at),
        "updated_at": _iso(r.updated_at),
        "version": r.version,
    } for r in rows]


async def _schedule_out(session: AsyncSession, row: Schedule) -> dict[str, Any]:
    return (await _schedule_outs(session, [row]))[0]


async def _get(session: AsyncSession, schedule_id: str) -> Schedule:
    row = await session.get(Schedule, schedule_id)
    if row is None:
        raise HTTPException(404, _GONE)
    return row


# --------------------------------------------------------------------------
# 固定路径（要排在 /{schedule_id} 前面）
# --------------------------------------------------------------------------


@router.get("/variables")
async def list_variables(tz: str | None = Query(default=None, alias="timezone"), trigger_kind: str | None = None,
                         event: str | None = None) -> list[dict[str, str]]:
    """输入模板能用的变量：[{name, label, example, note}]。示例按现在这一刻、timezone（默认 Asia/Shanghai）触发算；
    note 是补充说明（ISO 周的跨年规则），没有的是空串。trigger_kind=event 时按 event（事件种类）另给 {{ event.* }}、去掉
    上一次计划触发时间。"""
    zone = _timezone(tz) if tz is not None else service.DEFAULT_TIMEZONE
    if trigger_kind is None or _trigger_kind(trigger_kind) == "cron":
        return service.variables(tz=zone)
    if event not in triggers.EVENTS:
        raise _invalid("请指明事件种类：上传表格出了新的一期、上游工作流运行成功结束、数据源有要处理的变化", "trigger")
    return service.variables(tz=zone, event=event)


#: 选对象时一页最多给几个（按名字排；更多的按关键字查）
TRIGGER_OPTIONS_LIMIT = 100


@router.get("/trigger-options")
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]:
    """新建、修改事件触发任务时能选的：每种事件的对象（能不能选、为什么不能，由这里判，界面不自己判）、「只认某种来源」的
    来源、「只认某个应用」的应用、去抖的缺省值、触发链的上限。

    对象按名字排、一页最多 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

    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:
            # 名字里含这几个字，不分大小写：两种库同一个说法（db/portable.contains_ci，PostgreSQL 上按库的区域设置折大小写）
            from app.db.portable import contains_ci

            query = query.where(contains_ci(model.name, words))
        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, App.id))).all()
    return {
        "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,
        "max_depth": triggers.MAX_DEPTH,
    }


@router.post("/preview")
async def preview(request: Request, session: AsyncSession = Depends(get_session)) -> dict[str, Any]:
    """{cron, timezone, input_template?, workflow_id?, n?=5} → {description_text, next: [ISO（UTC）…], inputs}。

    inputs 是按下一次触发渲染的输入。带了 workflow_id 时对照入口字段的名字；必填项没填不在这里报——用户还在填。

    trigger_kind=event 时带 trigger、不带 cron：description_text 是触发说明，next 为空，inputs 按一个示例事件渲染（对象最近
    一次真实的事件，没有时是示例值），另给 event {text, real}。
    """
    body = await _body(request)
    if _trigger_kind(body.get("trigger_kind", "cron")) == "event":
        return await _preview_event(session, body)
    cron_expr, tz = _cron(body.get("cron")), _timezone(body.get("timezone"))
    template = _template(body.get("input_template"))
    n = body.get("n", 5)
    n = min(max(n, 1), MAX_PREVIEW) if isinstance(n, int) and not isinstance(n, bool) else 5
    await _preview_fields(session, body, template)
    upcoming = cron.preview(cron_expr, tz, n)
    first = upcoming[0] if upcoming else utcnow()
    return {
        "description_text": cron.describe(cron_expr, tz),
        "next": [_iso(m) for m in upcoming],
        "inputs": service.render_inputs(template, service.template_values(first, tz, cron_expr)),
    }


async def _preview_fields(session: AsyncSession, body: dict[str, Any], template: dict[str, str]) -> None:
    """预览时带了 workflow_id：对照入口字段的名字（发布版本和画布的字段都认，还不知道按哪一类运行）。"""
    workflow_id = body.get("workflow_id")
    if not (isinstance(workflow_id, str) and workflow_id.strip()):
        return
    workflow = await session.get(Workflow, workflow_id.strip())
    if workflow is None:
        return
    published = await _published_graph(session, workflow) if workflow.published_version else None
    fields = service.input_fields(workflow.graph) + service.input_fields(published)
    try:
        service.check_fields(template, fields, workflow.name, require=False)
    except service.TemplateError as e:
        raise _invalid(str(e), "input_template") from e


async def _preview_event(session: AsyncSession, body: dict[str, Any]) -> dict[str, Any]:
    if not _blank(body.get("cron")):
        raise _invalid(EVENT_NO_CRON, "cron")
    tz = _timezone(body.get("timezone", service.DEFAULT_TIMEZONE))
    trigger = _trigger(body.get("trigger"))
    template = _template(body.get("input_template"), trigger["event"])
    await _preview_fields(session, body, template)
    try:
        checked = await triggers.check(session, trigger)
    except triggers.TriggerError as e:
        raise _invalid(str(e), e.field) from e
    sample = await triggers.sample(session, trigger)
    fire = outbox.sample_fire(sample, tz)
    now = utcnow().replace(microsecond=0)
    return {
        "description_text": triggers.describe(trigger, checked.subject_name, app_name=checked.app_name),
        "next": [],
        "inputs": service.render_inputs(template, service.template_values(now, tz, ""), event=fire.values),
        "event": {"text": fire.text, "real": sample.real},
    }


# --------------------------------------------------------------------------
# 增删改查
# --------------------------------------------------------------------------


@router.get("")
async def list_schedules(session: AsyncSession = Depends(get_session)) -> dict[str, Any]:
    rows = list((await session.execute(select(Schedule).order_by(Schedule.created_at.desc(), Schedule.id.desc()))).scalars())
    return {"items": await _schedule_outs(session, rows)}


@router.post("", status_code=201)
async def create_schedule(request: Request, session: AsyncSession = Depends(get_session),
                          x_actor: str | None = Header(default=None)) -> dict[str, Any]:
    row = await create_schedule_row(session, await _body(request), actor=actor_of(x_actor))
    return await _schedule_out(session, row)


def schedule_values(body: dict[str, Any]) -> dict[str, Any]:
    """建定时任务的请求体 → 规范化之后的列（不含工作流名、下一次触发），逐项校验，不对抛 422（带 field）。只读，
    不碰库：发布单的预检用它「只检查」，建任务的接口和发布单建任务都先过它。

    target_kind 缺省 workflow：交回的键和以前一字不差。非工作流目标交回 target_kind、target_id，运行类别存探索（不用它）、
    输入模板为空、没有 Webhook。

    trigger_kind 缺省 cron：同上一字不差。event 时目标只能是工作流，不写 cron（存空串），时区缺省 Asia/Shanghai（日期变量按它算），
    另交回 trigger_kind、trigger（规范化之后的触发条件），输入模板按事件种类核对。"""
    if _trigger_kind(body.get("trigger_kind", "cron")) == "event":
        return _event_values(body)
    kind = _target_kind(body.get("target_kind", "workflow"))
    if kind != "workflow":
        _target_extras(body, kind)
        return {"name": _name(body.get("name")), "target_kind": kind, "target_id": _target_id(body.get("target_id"), kind),
                "cron": _cron(body.get("cron")), "timezone": _timezone(body.get("timezone")),
                "run_class": "exploratory", "input_template": {}, "enabled": _enabled(body.get("enabled", True)),
                "overlap": _overlap(body.get("overlap", "skip")),
                "misfire_grace_s": _grace(body.get("misfire_grace_s", DEFAULT_GRACE_S)), "webhook": None}
    values = {"name": _name(body.get("name")), "workflow_id": _workflow_id(body.get("workflow_id")),
              "cron": _cron(body.get("cron")), "timezone": _timezone(body.get("timezone")),
              "run_class": _run_class(body.get("run_class", "formal")),
              "input_template": _template(body.get("input_template")),
              "enabled": _enabled(body.get("enabled", True)), "overlap": _overlap(body.get("overlap", "skip")),
              "misfire_grace_s": _grace(body.get("misfire_grace_s", DEFAULT_GRACE_S))}
    return {**values, "webhook": _new_webhook(body.get("webhook"))}


#: 事件触发的任务不写 cron
EVENT_NO_CRON = "事件触发的定时任务不写 cron 表达式：它在事件发生时触发，不按时间"
#: 事件触发只能指向工作流
EVENT_WORKFLOW_ONLY = "事件触发的定时任务只能指向工作流：评测题集、数据源维护请用定时触发"


def _event_values(body: dict[str, Any]) -> dict[str, Any]:
    """事件触发任务的请求体 → 规范化之后的列（同 schedule_values，不碰库）。"""
    if _target_kind(body.get("target_kind", "workflow")) != "workflow":
        raise _invalid(EVENT_WORKFLOW_ONLY, "target_kind")
    if not _blank(body.get("cron")):
        raise _invalid(EVENT_NO_CRON, "cron")
    trigger = _trigger(body.get("trigger"))
    values = {"name": _name(body.get("name")), "workflow_id": _workflow_id(body.get("workflow_id")), "cron": "",
              "timezone": _timezone(body.get("timezone", service.DEFAULT_TIMEZONE)),
              "run_class": _run_class(body.get("run_class", "formal")),
              "input_template": _template(body.get("input_template"), trigger["event"]),
              "enabled": _enabled(body.get("enabled", True)), "overlap": _overlap(body.get("overlap", "skip")),
              "misfire_grace_s": _grace(body.get("misfire_grace_s", DEFAULT_GRACE_S)),
              "trigger_kind": "event", "trigger": trigger}
    return {**values, "webhook": _new_webhook(body.get("webhook"))}


def template_problem(template: dict[str, str], graph: Any, workflow_name: str) -> str | None:
    """输入模板对不上这张图的入口字段（字段名、必填项）：说明；对得上为 None。发布单的预检按将要发布的图查。"""
    try:
        service.check_fields(template, service.input_fields(graph), workflow_name)
    except service.TemplateError as e:
        return str(e)
    return None


async def _check_target(session: AsyncSession, kind: str, target_id: str) -> str:
    """非工作流目标：交给它的模块核对（存在、可用），交回目标现在的名字。不对抛 422（field=target_id）。"""
    if problem := await service.check_target(session, kind, target_id):
        raise _invalid(problem, "target_id")
    return await service.target_name(session, kind, target_id) or ""


async def create_schedule_row(session: AsyncSession, body: dict[str, Any], *, actor: str | None) -> Schedule:
    """校验并建一个定时任务（提交）。建任务的接口和发布单（停用着建，应用发布后再启用）共用。"""
    values = schedule_values(body)
    hook = values.pop("webhook")
    if values.get("target_kind", "workflow") != "workflow":
        kind, target_id = values.pop("target_kind"), values.pop("target_id")
        name = await _check_target(session, kind, target_id)
        # 目标的几列只经 scheduler/service 的一处封装写（非工作流目标：workflow_id 存空串）；workflow_name 记保存时的目标名
        values |= service.target_columns(kind, target_id)
        row = Schedule(**values, workflow_name=name,
                       next_run_at=service.next_run(values["cron"], values["timezone"], values["enabled"]),
                       created_by=actor, **_webhook_columns(None))
        session.add(row)
        await session.commit()
        await session.refresh(row)
        return row
    workflow = await _check_workflow(session, values["workflow_id"], values["run_class"], values["input_template"])
    next_run_at = None
    if values.get("trigger_kind") == "event":
        # 事件触发：核对对象、查环；记下对象现在的名字（对象删掉之后说法里用它）。没有下一次「到点」
        checked = await _check_trigger(session, values["trigger"], schedule_id=None,
                                       target_workflow_id=values["workflow_id"])
        values["trigger"] = {**values["trigger"], "subject_name": checked.subject_name}
    else:
        next_run_at = service.next_run(values["cron"], values["timezone"], values["enabled"])
    # 目标的几列只经 scheduler/service 的一处封装写（工作流目标：target_id 同 workflow_id）
    values |= service.target_columns("workflow", values["workflow_id"])
    row = Schedule(**values, workflow_name=workflow.name, next_run_at=next_run_at,
                   created_by=actor, **_webhook_columns(hook))
    session.add(row)
    await session.commit()
    await session.refresh(row)
    return row


@router.get("/{schedule_id}")
async def get_schedule(schedule_id: str, session: AsyncSession = Depends(get_session)) -> dict[str, Any]:
    return await _schedule_out(session, await _get(session, schedule_id))


@router.patch("/{schedule_id}")
async def update_schedule(schedule_id: str, request: Request,
                          session: AsyncSession = Depends(get_session)) -> dict[str, Any]:
    """改任意可改的项，带上读到的 version；对不上回 409。只读的项（id、next_run_at、last_fire……）带了也不理。

    看的是值变没变，不是键在不在：编辑表单把整份字段带回来是正常用法。
    - 工作流、运行类别、输入模板真改了才按改后的组合重新校验。只停用、改名不查工作流——工作流删了、撤了发布，
      也要能先把任务停下来。
    - cron、时区、启用状态真改了才重算下一次触发（见 _next_after_edit）。没改的不动它：到点了、这一拍还没来时
      保存一次，这一次照常发起，不会被推到下一次、连记录都没有就丢了。
    """
    body = await _body(request)
    row = await _get(session, schedule_id)
    version = body.get("version")
    if isinstance(version, bool) or not isinstance(version, int):
        raise _invalid("保存时需要带上读到的版本号 version，请刷新后重试", "version")
    if version != row.version:
        raise HTTPException(409, f"这个定时任务已被修改（当前是第 {row.version} 版，提交的是第 {version} 版），"
                                 "请刷新后再保存")
    values: dict[str, Any] = {}
    kind, current_target = service.target_of(row)
    if "target_kind" in body and _target_kind(body["target_kind"]) != kind:
        raise _invalid("定时任务的目标种类建好之后不能改：请新建一个定时任务", "target_kind")
    event = _event_kind(row)
    if "trigger_kind" in body and _trigger_kind(body["trigger_kind"]) != ("event" if event else "cron"):
        raise _invalid("定时任务的触发方式建好之后不能改：请新建一个定时任务", "trigger_kind")
    if kind != "workflow":
        return await _update_target_schedule(session, row, body, kind, current_target, version)
    if event:
        return await _update_event_schedule(session, row, body, version)
    if "name" in body:
        values["name"] = _name(body["name"])
    if "workflow_id" in body:
        # 目标的几列一起改（scheduler/service.target_columns）：只改 workflow_id 的话 target_id 还指着原来的工作流
        values |= service.target_columns("workflow", _workflow_id(body["workflow_id"]))
    if "cron" in body:
        values["cron"] = _cron(body["cron"])
    if "timezone" in body:
        values["timezone"] = _timezone(body["timezone"])
    if "run_class" in body:
        values["run_class"] = _run_class(body["run_class"])
    if "input_template" in body:
        values["input_template"] = _template(body["input_template"])
    if "enabled" in body:
        values["enabled"] = _enabled(body["enabled"])
    if "overlap" in body:
        values["overlap"] = _overlap(body["overlap"])
    if "misfire_grace_s" in body:
        values["misfire_grace_s"] = _grace(body["misfire_grace_s"])
    if "webhook" in body:
        values |= _patched_webhook(row, body["webhook"])

    def merged(key: str) -> Any:
        return values.get(key, getattr(row, key))

    def changed(key: str) -> bool:
        if key not in values:
            return False
        stored = getattr(row, key)
        return values[key] != (dict(stored or {}) if key == "input_template" else stored)

    if changed("workflow_id") or changed("run_class") or changed("input_template"):
        workflow = await _check_workflow(session, merged("workflow_id"), merged("run_class"), merged("input_template"))
        values["workflow_name"] = workflow.name
    # cron、时区在上面已经规范化（字段之间一个空格、时区的标准写法），和库里存的同一种写法，直接比
    if changed("cron") or changed("timezone") or changed("enabled"):
        values["next_run_at"] = _next_after_edit(row, merged("cron"), merged("timezone"), merged("enabled"))
    # 条件 UPDATE 才是真的乐观锁：两个人同时读到第 3 版、同时保存，只有先到的那个改得动
    result = await session.execute(
        update(Schedule).where(Schedule.id == schedule_id, Schedule.version == version)
        .values(**values, version=version + 1).execution_options(synchronize_session=False))
    if result.rowcount != 1:
        await session.rollback()
        raise HTTPException(409, "这个定时任务刚被别人修改过，请刷新后再保存")
    await session.commit()
    session.expire_all()
    return await _schedule_out(session, await _get(session, schedule_id))


async def _update_event_schedule(session: AsyncSession, row: Schedule, body: dict[str, Any],
                                 version: int) -> dict[str, Any]:
    """改一个事件触发的定时任务：名称、目标工作流、运行类别、输入模板、触发条件、时区、启用、重叠、补发宽限、Webhook。cron 不适用
    （整份表单带回来的空串不算）；下一次触发一直为空。看值变没变（同 update_schedule）：触发条件、目标真改了才重新核对对象、查环，
    模板、事件种类、目标、运行类别真改了才按改后的组合核对模板和入口字段。"""
    if "cron" in body and not _blank(body["cron"]):
        raise _invalid(EVENT_NO_CRON, "cron")
    values: dict[str, Any] = {}
    stored = dict(row.trigger or {}) if isinstance(row.trigger, dict) else {}
    current = {k: v for k, v in stored.items() if k != "subject_name"}
    trigger = _trigger(body["trigger"]) if "trigger" in body else current
    if "name" in body:
        values["name"] = _name(body["name"])
    if "workflow_id" in body:
        values |= service.target_columns("workflow", _workflow_id(body["workflow_id"]))
    for key, check in (("timezone", _timezone), ("run_class", _run_class), ("enabled", _enabled),
                       ("overlap", _overlap), ("misfire_grace_s", _grace)):
        if key in body:
            values[key] = check(body[key])
    if "input_template" in body:
        values["input_template"] = _template(body["input_template"], trigger.get("event"))
    if "webhook" in body:
        values |= _patched_webhook(row, body["webhook"])

    def merged(key: str) -> Any:
        return values.get(key, getattr(row, key))

    def changed(key: str) -> bool:
        if key not in values:
            return False
        return values[key] != (dict(getattr(row, key) or {}) if key == "input_template" else getattr(row, key))

    trigger_changed = trigger != current
    if trigger_changed or changed("workflow_id"):
        checked = await _check_trigger(session, trigger, schedule_id=row.id, target_workflow_id=merged("workflow_id"))
        values["trigger"] = {**trigger, "subject_name": checked.subject_name}
    if trigger_changed and "input_template" not in values:
        # 换了事件种类：原来的模板里的 {{ event.* }} 不一定还认
        values["input_template"] = _template(dict(row.input_template or {}), trigger.get("event"))
    if changed("workflow_id") or changed("run_class") or changed("input_template") or trigger_changed:
        workflow = await _check_workflow(session, merged("workflow_id"), merged("run_class"), merged("input_template"))
        values["workflow_name"] = workflow.name
    result = await session.execute(
        update(Schedule).where(Schedule.id == row.id, Schedule.version == version)
        .values(**values, version=version + 1).execution_options(synchronize_session=False))
    if result.rowcount != 1:
        await session.rollback()
        raise HTTPException(409, "这个定时任务刚被别人修改过，请刷新后再保存")
    schedule_id = row.id
    await session.commit()
    session.expire_all()
    return await _schedule_out(session, await _get(session, schedule_id))


async def _update_target_schedule(session: AsyncSession, row: Schedule, body: dict[str, Any], kind: str,
                                  current_target: str, version: int) -> dict[str, Any]:
    """改一个非工作流目标的定时任务：名称、目标（同一种）、cron、时区、启用、重叠、补发宽限。工作流、输入模板、运行类别、
    Webhook 不适用（带了照实拒；整份表单带回来的空值不算）。目标真改了才重新核对。"""
    probe = {k: body[k] for k in ("workflow_id", "input_template", "webhook", "run_class") if k in body}
    if probe.get("webhook") is None:
        probe.pop("webhook", None)
    _target_extras(probe, kind)
    values: dict[str, Any] = {}
    if "name" in body:
        values["name"] = _name(body["name"])
    if "target_id" in body:
        target_id = _target_id(body["target_id"], kind)
        if target_id != current_target:
            values["workflow_name"] = await _check_target(session, kind, target_id)
            values |= service.target_columns(kind, target_id)
    for key, check in (("cron", _cron), ("timezone", _timezone), ("enabled", _enabled), ("overlap", _overlap),
                       ("misfire_grace_s", _grace)):
        if key in body:
            values[key] = check(body[key])
    if any(k in values and values[k] != getattr(row, k) for k in ("cron", "timezone", "enabled")):
        values["next_run_at"] = _next_after_edit(row, values.get("cron", row.cron), values.get("timezone", row.timezone),
                                                 values.get("enabled", row.enabled))
    result = await session.execute(
        update(Schedule).where(Schedule.id == row.id, Schedule.version == version)
        .values(**values, version=version + 1).execution_options(synchronize_session=False))
    if result.rowcount != 1:
        await session.rollback()
        raise HTTPException(409, "这个定时任务刚被别人修改过，请刷新后再保存")
    schedule_id = row.id
    await session.commit()
    session.expire_all()
    return await _schedule_out(session, await _get(session, schedule_id))


async def retime(session: AsyncSession, row: Schedule, body: dict[str, Any]) -> Schedule:
    """按新的设置改一个已有的定时任务（提交）：名称、cron、时区、输入模板，校验同建任务；启用状态不动。cron、时区真改了才
    按 PATCH 同一个规则重算下一次触发。发布单给已经发布过的蓝图再发一版时用它沿用原来的任务，不另建一个（两个任务会
    在同一时刻各发起一次）。"""
    values = schedule_values({**body, "workflow_id": row.workflow_id, "enabled": row.enabled})
    workflow = await _check_workflow(session, row.workflow_id, row.run_class, values["input_template"])
    changed = {"name": values["name"], "cron": values["cron"], "timezone": values["timezone"],
               "input_template": values["input_template"], "workflow_name": workflow.name}
    if values["cron"] != row.cron or values["timezone"] != row.timezone:
        changed["next_run_at"] = _next_after_edit(row, values["cron"], values["timezone"], row.enabled)
    await session.execute(update(Schedule).where(Schedule.id == row.id)
                          .values(**changed, version=row.version + 1).execution_options(synchronize_session=False))
    await session.commit()
    await session.refresh(row)
    return row


async def set_enabled(session: AsyncSession, row: Schedule, enabled: bool) -> Schedule:
    """启用、停用一个定时任务（提交），下一次触发按 PATCH 同一个规则算，版本号加一。发布单在应用发布之后用它启用
    先停用着建的任务；已经是这个状态的不动。"""
    if row.enabled == enabled:
        return row
    await session.execute(
        update(Schedule).where(Schedule.id == row.id)
        .values(enabled=enabled,
                next_run_at=None if _event_kind(row) else _next_after_edit(row, row.cron, row.timezone, enabled),
                version=row.version + 1).execution_options(synchronize_session=False))
    await session.commit()
    # 只重读这一行：调用方（发布单）同一个会话里还拿着别的对象，不能整个会话过期
    await session.refresh(row)
    return row


def _next_after_edit(row: Schedule, cron_expr: str, tz: str, enabled: bool) -> datetime | None:
    """改了 cron、时区或启用状态之后的下一次触发。

    停用：没有下一次。刚启用：从现在算，不补停用期间错过的。启用着改了 cron、时区：原来那一次已经到点、这一拍还没来
    的，从那一刻接着按新的表达式算——新表达式也覆盖那一刻（每天 → 工作日）就还发起这一次，只属于旧表达式的不再发起。
    """
    if not enabled:
        return None
    now = utcnow()
    due = service.as_utc(row.next_run_at) if row.enabled and row.next_run_at is not None else None
    if due is not None and due <= now:
        return cron.next_fire(cron_expr, tz, due - timedelta(microseconds=1))
    return service.next_run(cron_expr, tz, True, now)


@router.delete("/{schedule_id}", status_code=204)
async def delete_schedule(schedule_id: str, session: AsyncSession = Depends(get_session)) -> Response:
    """删除任务，连同它的触发历史和推送记录。它发起过的运行留着（运行上的来源 id 也留着）：那是追溯凭证。"""
    await _get(session, schedule_id)
    # 外键是 ON DELETE CASCADE；这里照样按顺序删一遍，不依赖连接上开没开外键约束
    await session.execute(delete(WebhookDelivery).where(WebhookDelivery.schedule_id == schedule_id))
    await session.execute(delete(ScheduleFire).where(ScheduleFire.schedule_id == schedule_id))
    await session.execute(delete(Schedule).where(Schedule.id == schedule_id))
    await session.commit()
    return Response(status_code=204)


# --------------------------------------------------------------------------
# 立即运行、触发历史、测试 Webhook
# --------------------------------------------------------------------------


@router.post("/{schedule_id}/run")
async def run_now(schedule_id: str, session: AsyncSession = Depends(get_session),
                  x_actor: str | None = Header(default=None)) -> dict[str, Any]:
    """立即触发一次（kind=manual），停用的任务也可以。规则和到点触发相同：上一次运行没结束就跳过，正式运行只从当前
    发布版本发起。运行记在点按钮的人名下，没有署名时记在「定时任务：<名字>」名下。返回 {fire, run_id}；目标是评测题集的，
    发起的是一次评测，fire.eval_run_id 是它，run_id 为空。

    触发时间取到秒：连点两下落在同一秒的，交回同一条触发记录，不会发起两次。

    事件触发的任务按示例事件发起（scheduler/triggers.sample：对象最近一次真实的事件）：运行来源里记 trigger（sample=true），
    触发链从这里起算。对象还没发生过这种事件、模板又用到 {{ event.* }} 的照实记一次未能发起。"""
    row = await _get(session, schedule_id)
    event = None
    if _event_kind(row) and isinstance(row.trigger, dict):
        event = outbox.sample_fire(await triggers.sample(session, row.trigger), row.timezone)
    fire = await service.fire_once(schedule_id, utcnow().replace(microsecond=0), "manual", actor=actor_of(x_actor),
                                   event=event)
    if fire is None:
        raise HTTPException(404, _GONE)
    return {"fire": (await _fire_outs(session, [fire]))[0], "run_id": fire.run_id}


@router.get("/{schedule_id}/fires")
async def list_fires(schedule_id: str, limit: str | None = None, before: str | None = None,
                     session: AsyncSession = Depends(get_session)) -> dict[str, Any]:
    """触发历史，新的在前：{items: [Fire], next_before}。翻页把 next_before 原样作为 before 传回来（Z 结尾，拼进地址
    不用编码）；没有更多时为空。"""
    await _get(session, schedule_id)
    size = 20
    if limit is not None:
        if not limit.strip().isdecimal() or not 1 <= int(limit) <= 100:
            raise _invalid(f"每页条数需要是 1 到 100 之间的整数，当前为「{limit}」", "limit")
        size = int(limit)
    stmt = select(ScheduleFire).where(ScheduleFire.schedule_id == schedule_id)
    if before is not None:
        try:
            cursor = _cursor_time(before)
        except ValueError as e:
            raise _invalid(f"翻页位置需要是接口返回的 next_before，当前为「{before}」", "before") from e
        stmt = stmt.where(ScheduleFire.created_at < service.as_utc(cursor))
    rows = list((await session.execute(
        stmt.order_by(ScheduleFire.created_at.desc(), ScheduleFire.id.desc()).limit(size + 1))).scalars())
    items = rows[:size]
    return {"items": await _fire_outs(session, items),
            "next_before": _cursor(items[-1].created_at) if len(rows) > size else None}


@router.post("/{schedule_id}/webhook/test")
async def test_webhook(schedule_id: str, session: AsyncSession = Depends(get_session)) -> dict[str, Any]:
    """按保存的 Webhook 发一条测试消息（不重试）：{ok, http_status, message}。失败的说明里只有主机名。"""
    await _get(session, schedule_id)
    try:
        return await webhook.send_test(schedule_id)
    except KeyError as e:
        raise HTTPException(404, _GONE) from e
    except ValueError as e:
        raise _invalid(str(e), "webhook") from e