From a68fcea13b57b5b1336963c7a76983246eddb302 Mon Sep 17 00:00:00 2001 From: root Date: Mon, 28 Sep 2026 02:32:33 +0800 Subject: [PATCH] feat(server): unified scheduled-job registry and scheduler; module timers moved onto it - modules/schedules: registerScheduledJob({key, defaults, run}) registry; defaults are synced into scheduled_job without overriding user-changed enabled/schedule - scheduler runs only on the io worker, one instance via pg_try_advisory_lock held on a dedicated connection; due = enabled and next_run_at reached, or run requested; jobs run in parallel, never re-entrant; interval next = finish + period, cron next match; run-now does not move next_run_at and merges duplicate requests; skip advances one period; history trimmed to the last 50 runs; interrupted runs marked failed on takeover - agent_check jobs: trigger condition checked first, goal template rendered, runAgent enqueued via startRun(trigger=schedule) with allowlist and budget, one run per time slot; allowRecover uses the autoInvestigate grant check; results harvested back into the run history and optionally reported as an info decision; 3 built-in checks, off by default - sources-identity (feeds.poll, sites.snapshot, identity.review), acquisition-organize (downloads.sync, cache.threshold, downloads.orchestrate, slots.reevaluate, subscriptions.search, subscriptions.metadata, library.reconcile) and file-manager (trash.purge) register jobs instead of running their own setInterval; default intervals unchanged; backup registered as an unconfigured placeholder - endpoints under /schedules (admin, console session only) TODO: pi refreshPiTokens and auth-settings session cleanup still own their timers. Co-Authored-By: Claude Opus 5.5 --- .../__tests__/background-loop.test.ts | 6 +- .../src/modules/acquisition-organize/index.ts | 194 +++++++---- apps/server/src/modules/file-manager/index.ts | 44 ++- .../src/modules/schedules/agent-checks.ts | 162 +++++++++ apps/server/src/modules/schedules/cron.ts | 89 +++++ apps/server/src/modules/schedules/index.ts | 61 ++++ apps/server/src/modules/schedules/registry.ts | 99 ++++++ apps/server/src/modules/schedules/routes.ts | 141 ++++++++ .../server/src/modules/schedules/scheduler.ts | 326 ++++++++++++++++++ apps/server/src/modules/schedules/store.ts | 322 +++++++++++++++++ .../src/modules/sources-identity/index.ts | 34 +- .../src/modules/sources-identity/scheduler.ts | 42 ++- apps/server/test/api-registry.test.ts | 8 + apps/server/test/e2e/fixtures/modules.ts | 3 + apps/server/test/modules.test.ts | 3 +- apps/server/test/schedules/schedules.test.ts | 325 +++++++++++++++++ 16 files changed, 1732 insertions(+), 127 deletions(-) create mode 100644 apps/server/src/modules/schedules/agent-checks.ts create mode 100644 apps/server/src/modules/schedules/cron.ts create mode 100644 apps/server/src/modules/schedules/index.ts create mode 100644 apps/server/src/modules/schedules/registry.ts create mode 100644 apps/server/src/modules/schedules/routes.ts create mode 100644 apps/server/src/modules/schedules/scheduler.ts create mode 100644 apps/server/src/modules/schedules/store.ts create mode 100644 apps/server/test/schedules/schedules.test.ts diff --git a/apps/server/src/modules/acquisition-organize/__tests__/background-loop.test.ts b/apps/server/src/modules/acquisition-organize/__tests__/background-loop.test.ts index 743cb7cd..c3cc6efc 100644 --- a/apps/server/src/modules/acquisition-organize/__tests__/background-loop.test.ts +++ b/apps/server/src/modules/acquisition-organize/__tests__/background-loop.test.ts @@ -1,6 +1,7 @@ import { sql } from "@umt/db"; import { createMockQb, createTempTree, type MockQb, makeTorrent, type TempTree } from "@umt/testkit"; import { afterAll, beforeAll, describe, expect, it } from "vitest"; +import { createSchedulesModule } from "../../schedules/index.ts"; import { evaluateAcquisition } from "../services/acquisition.ts"; import { submitDownload, syncDownloads } from "../services/downloads.ts"; import { executeRecord, planImport } from "../services/organize.ts"; @@ -15,7 +16,7 @@ import { seedSlot, } from "./harness.ts"; -// W4-D P0:后台循环只编排。补入库在巡检里只入队 verifyAndImport;一个 50MB 的入库(复制 + sha256 + 慢 ffprobe) +// W4-D P0:后台循环只编排(W7-C 起由 schedules 调度器驱动模块登记的定时任务)。补入库在巡检里只入队 verifyAndImport;一个 50MB 的入库(复制 + sha256 + 慢 ffprobe) // 进行中时,qB 同步定时器仍按周期打到 mock qB(用 mock qB 的请求时间戳断言)。 const hasPg = await pgAvailable(); @@ -31,6 +32,7 @@ describe.skipIf(!hasPg)("acquisition-organize 后台循环不被入库阻塞", ( const disk = fakeDisk(10 * 1024 ** 3); const torrents = new Map(); let base: Awaited>; + const schedules = createSchedulesModule({ builtinChecks: false, scheduler: { tickMs: 50, lockRetryMs: 100 } }); beforeAll(async () => { qb = await createMockQb({ addVisibleAfterPolls: 0 }); @@ -53,6 +55,7 @@ describe.skipIf(!hasPg)("acquisition-organize 后台循环不被入库阻塞", ( }, 60_000); afterAll(async () => { + await schedules.stop?.(); await h?.module.stop?.(); await h?.close(); await qb?.close(); @@ -82,6 +85,7 @@ describe.skipIf(!hasPg)("acquisition-organize 后台循环不被入库阻塞", ( await sql`update download_task set completed_at = now() - interval '10 minutes' where id = ${dtId}`.execute(h.db); await h.module.start?.(h.ctx); + await schedules.start?.(h.ctx); const deadline = Date.now() + 10_000; while (!h.engine.enqueued.some((e) => e.task === "verifyAndImport") && Date.now() < deadline) { await new Promise((r) => setTimeout(r, 50)); diff --git a/apps/server/src/modules/acquisition-organize/index.ts b/apps/server/src/modules/acquisition-organize/index.ts index f02ae782..a500d279 100644 --- a/apps/server/src/modules/acquisition-organize/index.ts +++ b/apps/server/src/modules/acquisition-organize/index.ts @@ -1,5 +1,6 @@ import { type EventPayload, EventType } from "@umt/contracts"; import { sql } from "@umt/db"; +import { registerScheduledJob } from "../schedules/registry.ts"; import type { ServerContext, ServerModule } from "../types.ts"; import { registerAoCapabilities } from "./capabilities.ts"; import { registerAoTasks } from "./handlers.ts"; @@ -31,49 +32,30 @@ const CLEANUP_SLOT_MS = 5 * 60_000; /** * 模块工厂:测试注入 mock qB / 固定磁盘用量 / 假种子来源;正式装配用 acquisitionOrganizeModule。 * - * 后台只做编排,两个独立定时器互不阻塞: - * sync —— qB 同步 + 缓存阈值暂停/恢复(syncDownloads),只有轻量 HTTP 与 DB 写; - * plan —— 阈值判断后入队 cleanupDownloadCache、巡检补投 submitDownload / verifyAndImport、按周期入队 reconcileLibrary。 - * 复制/删种/全量 stat 这些重活都进任务引擎(租约 + attemptToken),不在循环里 await。 + * 后台只做编排,start() 把各项周期登记到定时任务注册表(schedules 模块的调度器驱动,不同任务并行、同一任务不重入): + * downloads.sync —— qB 同步 + 缓存阈值暂停/恢复(syncDownloads),只有轻量 HTTP 与 DB 写; + * cache.threshold —— 超硬阈值入队 cleanupDownloadCache; + * downloads.orchestrate —— 巡检补投 submitDownload / verifyAndImport; + * slots.reevaluate / subscriptions.search / subscriptions.metadata —— Slot 到期重评、订阅周期搜索、总集数刷新; + * library.reconcile —— 按周期入队 reconcileLibrary。 + * 复制/删种/全量 stat 这些重活都进任务引擎(租约 + attemptToken),不在调度里 await。 */ export function createAcquisitionOrganizeModule(overrides: AoOverrides = {}): ServerModule & { deps(): AoDeps } { let deps: AoDeps | null = null; - const timers = new Set>(); - let stopped = false; - let lastReconcileSlot: string | null = null; // register* 阶段只保存 ctx;端口在首次使用时组装(ctx 其它字段此时才完整) const lazy = (ctx: ServerContext): AoDeps => { deps ??= { ctx, ports: createPorts(ctx, overrides) }; return deps; }; - /** 编排一轮:阈值 → 入队清理;补投提交/入库;周期入队资产对账 */ - const planTick = async (d: AoDeps) => { - const blocked = await automationBlock(d); - if (!blocked) { - const cache = await d.ctx.settings.get("cache"); - const over = (await loadVolumes(d)).some((v) => (usedPercent(v) ?? 0) >= cache.hardThresholdPercent); - if (over) await enqueueThresholdCleanup(d); - await sweepReservedPlans(d); - await sweepPendingImports(d); - // W5-C:Slot 到期重评(忙时延后的发布 / 定时)、订阅周期主动搜索、追踪剧集的总集数刷新 - await sweepDueSlots(d); - await sweepSubscriptionSearch(d); - await sweepMetadataRefresh(d); - } - const every = d.ports.options.reconcileIntervalMs; - if (every > 0) { - const slot = timeSlot(d.ports.now(), every); - if (slot !== lastReconcileSlot) { - await enqueueTracked(d, { - kind: "reconcileLibrary", - idempotencyKey: `auto:${slot}`, - input: {}, - }); - lastReconcileSlot = slot; - } - } - }; + /** 编排类任务:自动化总开关关闭或只读对账时跳过 */ + const guarded = + (d: AoDeps, fn: (d: AoDeps) => Promise) => async (): Promise<{ status?: "skipped"; summary: string }> => { + const blocked = await automationBlock(d); + if (blocked) + return { status: "skipped", summary: blocked === "automation_disabled" ? "自动化已关闭" : "只读对账模式" }; + return { summary: await fn(d) }; + }; const module: ServerModule & { deps(): AoDeps } = { name: "acquisition-organize", @@ -113,49 +95,114 @@ export function createAcquisitionOrganizeModule(overrides: AoOverrides = {}): Se async start(ctx) { const d = lazy(ctx); if (!d.ports.options.backgroundLoops) return; - stopped = false; - const intervalMs = async () => { - const cache = await ctx.settings.get("cache").catch(() => null); - return d.ports.options.syncIntervalMs ?? (cache?.checkIntervalSec ?? 60) * 1000; - }; - /** 自调度循环:上一轮结束后再排下一轮,单个循环内不重入;两个循环互不等待 */ - const loop = (name: string, fn: (d: AoDeps) => Promise, period: () => Promise) => { - const schedule = async () => { - if (stopped) return; - const t = setTimeout( - async () => { - timers.delete(t); - try { - await fn(d); - } catch (err) { - ctx.logger.error( - { err: { message: errMessage(err) }, loop: name }, - "acquisition-organize 后台轮询失败", - ); - } - void schedule(); - }, - await period(), - ); - timers.add(t); - if (stopped) clearTimeout(t); - }; - void schedule(); - }; - loop("sync", syncDownloads, intervalMs); - loop("plan", planTick, async () => d.ports.options.planIntervalMs ?? (await intervalMs())); - }, - stop() { - stopped = true; - for (const t of timers) clearTimeout(t); - timers.clear(); + const cache = await ctx.settings.get("cache").catch(() => null); + const syncMs = d.ports.options.syncIntervalMs ?? (cache?.checkIntervalSec ?? 60) * 1000; + const planMs = d.ports.options.planIntervalMs ?? syncMs; + const every = (everyMs: number) => ({ type: "interval" as const, everyMs }); + + registerScheduledJob(ctx, { + key: "downloads.sync", + defaults: { + name: "qB 同步", + description: "同步下载器任务状态,按缓存阈值暂停或恢复", + group: "download", + schedule: every(syncMs), + }, + run: async () => { + const st = await syncDownloads(d); + if (st.instances > 0 && st.unreachable.length === st.instances) { + throw new Error(`下载器都连不上:${st.unreachable.join("、")}`); + } + return { + summary: + `${st.instances} 个下载器,更新 ${st.updated},完成 ${st.completed.length}` + + (st.unreachable.length ? `,连不上 ${st.unreachable.length}` : ""), + }; + }, + }); + registerScheduledJob(ctx, { + key: "cache.threshold", + defaults: { + name: "缓存阈值检查", + description: "下载缓存超过硬阈值时入队清理", + group: "maintain", + schedule: every(planMs), + }, + run: guarded(d, async (d) => { + const hard = (await d.ctx.settings.get("cache")).hardThresholdPercent; + const over = (await loadVolumes(d)).some((v) => (usedPercent(v) ?? 0) >= hard); + if (!over) return "未超阈值"; + return (await enqueueThresholdCleanup(d)) ? "已入队清理" : "清理进行中"; + }), + }); + registerScheduledJob(ctx, { + key: "downloads.orchestrate", + defaults: { + name: "补提交与补入库", + description: "补投已预留未提交的下载、已完成未入库的任务", + group: "download", + schedule: every(planMs), + }, + run: guarded(d, async (d) => { + const submitted = await sweepReservedPlans(d); + const imports = await sweepPendingImports(d); + return `补提交 ${submitted},补入库 ${imports}`; + }), + }); + registerScheduledJob(ctx, { + key: "slots.reevaluate", + defaults: { + name: "到期重评", + description: "到了重评时间的收录位重新评估(忙时延后的发布、定时)", + group: "download", + schedule: every(planMs), + }, + run: guarded(d, async (d) => `${await sweepDueSlots(d)} 项`), + }); + registerScheduledJob(ctx, { + key: "subscriptions.search", + defaults: { + name: "订阅周期搜索", + description: "到期的订阅主动搜索站点", + group: "collect", + schedule: every(planMs), + }, + run: guarded(d, async (d) => `${await sweepSubscriptionSearch(d)} 项`), + }); + registerScheduledJob(ctx, { + key: "subscriptions.metadata", + defaults: { + name: "总集数刷新", + description: "追踪中的剧集刷新总集数", + group: "organize", + schedule: every(planMs), + }, + run: guarded(d, async (d) => `${await sweepMetadataRefresh(d)} 项`), + }); + const reconcileMs = d.ports.options.reconcileIntervalMs; + if (reconcileMs > 0) { + registerScheduledJob(ctx, { + key: "library.reconcile", + defaults: { + name: "资产对账", + description: "核对媒体库文件与资产记录", + group: "organize", + schedule: every(reconcileMs), + }, + run: async (c) => { + // 同一周期时间槽只入队一次;立即执行用独立槽 + await enqueueTracked(d, { kind: "reconcileLibrary", idempotencyKey: `auto:${c.slot}`, input: {} }); + return { summary: "已入队对账" }; + }, + }); + } }, }; return module; } -/** 超阈值:按时间槽入队一次 cleanupDownloadCache;上一次清理仍在进行则跳过 */ -async function enqueueThresholdCleanup(d: AoDeps): Promise { +/** 超阈值:按时间槽入队一次 cleanupDownloadCache;上一次清理仍在进行则跳过(返回 false) */ +async function enqueueThresholdCleanup(d: AoDeps): Promise { const busy = await d.ctx.db .selectFrom("task") .select("id") @@ -163,12 +210,13 @@ async function enqueueThresholdCleanup(d: AoDeps): Promise { .where("status", "in", ["pending", "running"]) .where("updated_at", ">", new Date(d.ports.now().getTime() - 30 * 60_000)) .executeTakeFirst(); - if (busy) return; + if (busy) return false; await enqueueTracked(d, { kind: "cleanupDownloadCache", idempotencyKey: `threshold:${timeSlot(d.ports.now(), CLEANUP_SLOT_MS)}`, input: {}, }); + return true; } /** 下载已完成、没有整理记录(入库事件被总开关延后等)的纳管任务:入队 verifyAndImport(业务行 import:) */ diff --git a/apps/server/src/modules/file-manager/index.ts b/apps/server/src/modules/file-manager/index.ts index 0129aa41..44d0c637 100644 --- a/apps/server/src/modules/file-manager/index.ts +++ b/apps/server/src/modules/file-manager/index.ts @@ -3,9 +3,9 @@ import { defineTask } from "@umt/tasks"; import { z } from "zod"; import type { AoOverrides } from "../acquisition-organize/index.ts"; import { createPorts } from "../acquisition-organize/ports.ts"; -import { timeSlot } from "../acquisition-organize/services/common.ts"; +import { registerScheduledJob } from "../schedules/registry.ts"; import type { ServerContext, ServerModule } from "../types.ts"; -import { defaultOptions, errMessage, type FmDeps, type FmOptions } from "./common.ts"; +import { defaultOptions, type FmDeps, type FmOptions } from "./common.ts"; import { registerFileRoutes } from "./routes.ts"; import { purgeTrash } from "./trash.ts"; @@ -13,7 +13,7 @@ import { purgeTrash } from "./trash.ts"; * W5-A 文件管理(/files,docs/plan/06-file-manager-spec.md):只浏览已登记根(库根 + 下载器路径映射), * 行内业务状态、改名同步资产与路径绑定修订、移动/复制两段式 planId、整理走 planImport、回收站 + 引用检查、识别/探测/刮削。 * 复用 acquisition-organize 的服务导出(planImport / 路径守卫 / Move 授权 / H&R),不改其文件。 - * 持久任务 purgeTrash:每日清理过期回收站项(FILES_TRASH_RETENTION_DAYS,默认 7 天)。 + * 持久任务 purgeTrash:每日清理过期回收站项(FILES_TRASH_RETENTION_DAYS,默认 7 天);周期登记为定时任务 trash.purge。 */ export interface FmOverrides { ao?: AoOverrides; @@ -22,7 +22,6 @@ export interface FmOverrides { export function createFileManagerModule(overrides: FmOverrides = {}): ServerModule & { deps(): FmDeps } { let deps: FmDeps | null = null; - let timer: ReturnType | null = null; const lazy = (ctx: ServerContext): FmDeps => { if (!deps) { const { options: aoOptions, ...aoRest } = overrides.ao ?? {}; @@ -34,20 +33,6 @@ export function createFileManagerModule(overrides: FmOverrides = {}): ServerModu } return deps; }; - const enqueuePurge = async (d: FmDeps) => { - try { - await d.ctx.engine.enqueue( - "purgeTrash", - {}, - { - idempotencyKey: `purgeTrash:${timeSlot(d.ao.ports.now(), d.options.purgeIntervalMs)}`, - traceId: randomUUID(), - }, - ); - } catch (err) { - d.ctx.logger.warn({ err: { message: errMessage(err) } }, "回收站清理入队失败"); - } - }; return { name: "file-manager", deps() { @@ -75,12 +60,23 @@ export function createFileManagerModule(overrides: FmOverrides = {}): ServerModu start(ctx) { const d = lazy(ctx); if (!d.options.backgroundLoops || d.options.purgeIntervalMs <= 0) return; - void enqueuePurge(d); - timer = setInterval(() => void enqueuePurge(d), Math.min(d.options.purgeIntervalMs, 3600_000)); - }, - stop() { - if (timer) clearInterval(timer); - timer = null; + registerScheduledJob(ctx, { + key: "trash.purge", + defaults: { + name: "回收站清理", + description: "清除过期的回收站项", + group: "maintain", + schedule: { type: "interval", everyMs: d.options.purgeIntervalMs }, + }, + run: async (c) => { + const r = await d.ctx.engine.enqueue( + "purgeTrash", + {}, + { idempotencyKey: `purgeTrash:${c.slot}`, traceId: randomUUID() }, + ); + return { summary: r.deduplicated ? "本周期已入队" : "已入队清理" }; + }, + }); }, }; } diff --git a/apps/server/src/modules/schedules/agent-checks.ts b/apps/server/src/modules/schedules/agent-checks.ts new file mode 100644 index 00000000..dbf66428 --- /dev/null +++ b/apps/server/src/modules/schedules/agent-checks.ts @@ -0,0 +1,162 @@ +import { + AgentCheckConfig, + type AgentCheckConfigInput, + CAPABILITY_CATALOG, + type KnownCapabilityId, +} from "@umt/contracts"; +import { sql } from "@umt/db"; +import { recoveryGrants } from "../agent/investigate.ts"; +import { INVESTIGATE_CAPABILITIES, PLAN_ONLY_KINDS, RECOVER_CAPABILITIES } from "../agent/runner.ts"; +import { startRun } from "../agent/service.ts"; +import type { ServerContext } from "../types.ts"; +import type { JobRunContext, JobRunResult, ScheduledJobDef } from "./registry.ts"; + +// Agent 巡检(kind = agent_check):到点先按业务表判断触发条件,满足才渲染目标模板、入队 runAgent +// (trigger=schedule,能力白名单 + 预算;同一时间槽 dedupe_key 相同只建一个 run)。 +// allowRecover 走与 autoInvestigate 相同的授权路径:有覆盖恢复动作的有效持续授权才用 execute 模式, +// 否则降级只读;实际每次调用仍由 authorize 按授权判定。run 结束后调度器回填结果(见 scheduler.harvest)。 + +export const HOUR = 3600_000; + +/** 内置巡检模板(默认关闭) */ +export const BUILTIN_AGENT_CHECKS: { + key: string; + name: string; + description: string; + everyMs: number; + config: AgentCheckConfigInput; +}[] = [ + { + key: "check.pending", + name: "待处理巡检", + description: "检查待处理事项并给出建议(只读)", + everyMs: 6 * HOUR, + config: { + goalTemplate: + "巡检待处理中心({{now}}):当前有 {{openDecisions}} 条未关闭事项。逐条查看证据,按紧急程度排序," + + "给出每条的处理建议与理由;只读,不执行任何变更。", + capabilityAllowlist: [...INVESTIGATE_CAPABILITIES], + budget: { maxToolCalls: 20, maxDurationSec: 300 }, + triggerCondition: "open_decisions", + }, + }, + { + key: "check.downloads", + name: "下载诊断", + description: "检查无进展或失败的下载并诊断;开启恢复后在授权内重试或重新校验", + everyMs: 2 * HOUR, + config: { + goalTemplate: + "巡检下载({{now}}):有 {{stalledDownloads}} 个下载无进展或失败。用只读能力查明原因(站点、做种、磁盘、任务错误)," + + "给出诊断与建议;若本次允许恢复,只在授权内做重试或强制重新校验,每个任务最多一次,不换源、不删除。", + capabilityAllowlist: [...INVESTIGATE_CAPABILITIES], + budget: { maxToolCalls: 24, maxDurationSec: 420 }, + triggerCondition: "stalled_downloads", + allowRecover: false, + }, + }, + { + key: "check.weekly", + name: "订阅与站点周报", + description: "检查订阅缺集与站点健康,汇总写入待处理(只读)", + everyMs: 7 * 24 * HOUR, + config: { + goalTemplate: + "生成本周巡检周报({{date}}):追踪中的订阅 {{activeSubscriptions}} 个。列出缺集的订阅与原因推测、" + + "健康异常的站点与 Feed,给出本周建议;只读,最后用一段不超过 300 字的中文汇总。", + capabilityAllowlist: [...INVESTIGATE_CAPABILITIES], + budget: { maxToolCalls: 30, maxDurationSec: 600 }, + triggerCondition: "subscriptions_active", + reportAsDecision: true, + }, + }, +]; + +interface Facts { + openDecisions: number; + stalledDownloads: number; + activeSubscriptions: number; +} + +async function facts(ctx: ServerContext): Promise { + const r = await sql` + select + (select count(*)::int from decision where status in ('open', 'investigating', 'awaiting_user')) as "openDecisions", + (select count(*)::int from download_task where removed_at is null and status in ('stalled', 'failed')) as "stalledDownloads", + (select count(*)::int from subscription where status = 'active') as "activeSubscriptions"`.execute(ctx.db); + return r.rows[0] ?? { openDecisions: 0, stalledDownloads: 0, activeSubscriptions: 0 }; +} + +export function renderGoal(template: string, now: Date, f: Facts): string { + const vars: Record = { + now: now.toISOString(), + date: now.toISOString().slice(0, 10), + openDecisions: String(f.openDecisions), + stalledDownloads: String(f.stalledDownloads), + activeSubscriptions: String(f.activeSubscriptions), + }; + return template.replace(/\{\{\s*(\w+)\s*\}\}/g, (m, k: string) => vars[k] ?? m); +} + +const isPlanOnly = (id: string) => { + const d = CAPABILITY_CATALOG[id as KnownCapabilityId]; + return !!d && PLAN_ONLY_KINDS.includes(d.kind); +}; + +/** agent_check 的执行体(内置与自定义共用) */ +export async function runAgentCheck(ctx: ServerContext, c: JobRunContext): Promise { + const parsed = AgentCheckConfig.safeParse(c.config); + if (!parsed.success) throw new Error(`巡检参数无效:${parsed.error.issues[0]?.message ?? "未知"}`); + const cfg = parsed.data; + const f = await facts(ctx); + const unmet = + (cfg.triggerCondition === "open_decisions" && f.openDecisions === 0) || + (cfg.triggerCondition === "stalled_downloads" && f.stalledDownloads === 0) || + (cfg.triggerCondition === "subscriptions_active" && f.activeSubscriptions === 0); + // 手动执行不看触发条件 + if (unmet && c.trigger === "schedule") return { status: "skipped", summary: "触发条件不满足,未唤醒 Agent" }; + + let mode: "execute" | "plan_only" = "plan_only"; + let allowlist = cfg.capabilityAllowlist.filter(isPlanOnly); + let note = ""; + if (cfg.allowRecover) { + const grants = await recoveryGrants(ctx, ["task.retry", "download.recheck"]); + if (grants.length > 0) { + mode = "execute"; + allowlist = [...new Set([...(allowlist.length ? allowlist : INVESTIGATE_CAPABILITIES), ...RECOVER_CAPABILITIES])]; + } else note = "(没有覆盖恢复动作的有效授权,已降级只读)"; + } + const goal = renderGoal(cfg.goalTemplate, c.now, f); + const dedupeKey = c.trigger === "manual" ? `schedule:${c.jobId}:manual:${c.runId}` : `schedule:${c.jobId}:${c.slot}`; + const { run, created } = await startRun(ctx, { + goal, + mode, + capabilityAllowlist: allowlist, + trigger: "schedule", + createdBy: `schedule:${c.jobId}`, + dedupeKey, + budget: cfg.budget, + }); + return { + summary: `${created ? "已入队" : "本时段已有"} Agent 运行(${mode === "execute" ? "授权内恢复" : "只读"})${note}`, + agentRunId: run.id, + }; +} + +/** 内置巡检的注册定义(默认关闭) */ +export function builtinAgentCheckDefs(ctx: ServerContext): ScheduledJobDef[] { + return BUILTIN_AGENT_CHECKS.map((t) => ({ + key: t.key, + kind: "agent_check" as const, + defaults: { + name: t.name, + description: t.description, + group: "agent" as const, + schedule: { type: "interval" as const, everyMs: t.everyMs }, + enabled: false, + config: AgentCheckConfig.parse(t.config) as unknown as Record, + firstRun: "after_interval" as const, + }, + run: (c) => runAgentCheck(ctx, c), + })); +} diff --git a/apps/server/src/modules/schedules/cron.ts b/apps/server/src/modules/schedules/cron.ts new file mode 100644 index 00000000..04a1d80c --- /dev/null +++ b/apps/server/src/modules/schedules/cron.ts @@ -0,0 +1,89 @@ +// 最小 5 段 cron(分 时 日 月 周),按服务器本地时区计算;支持 `*`、`a,b`、`a-b`、`*/n`、`a-b/n`。 +// 周:0–7(0 与 7 都是周日)。日与周同时受限时按 vixie cron 语义取"或"。 + +export interface CronFields { + minute: Set; + hour: Set; + day: Set; + month: Set; + weekday: Set; + dayAny: boolean; + weekdayAny: boolean; +} + +const RANGES: [number, number][] = [ + [0, 59], + [0, 23], + [1, 31], + [1, 12], + [0, 7], +]; + +function parseField(src: string, [lo, hi]: [number, number], label: string): Set { + const out = new Set(); + for (const part of src.split(",")) { + const m = /^(\*|(\d+)(?:-(\d+))?)(?:\/(\d+))?$/.exec(part); + if (!m) throw new Error(`cron ${label} 段无效:${part}`); + let a = lo; + let b = hi; + if (m[1] !== "*") { + a = Number(m[2]); + b = m[3] !== undefined ? Number(m[3]) : m[4] !== undefined ? hi : a; + } + const step = m[4] !== undefined ? Number(m[4]) : 1; + if (a < lo || b > hi || a > b || step < 1) throw new Error(`cron ${label} 段越界:${part}`); + for (let v = a; v <= b; v += step) out.add(v); + } + return out; +} + +export function parseCron(expr: string): CronFields { + const parts = expr.trim().split(/\s+/); + if (parts.length !== 5) throw new Error("cron 需要 5 段:分 时 日 月 周"); + const [mi, h, d, mo, w] = parts as [string, string, string, string, string]; + const weekday = parseField(w, RANGES[4]!, "周"); + if (weekday.has(7)) weekday.add(0); + return { + minute: parseField(mi, RANGES[0]!, "分"), + hour: parseField(h, RANGES[1]!, "时"), + day: parseField(d, RANGES[2]!, "日"), + month: parseField(mo, RANGES[3]!, "月"), + weekday, + dayAny: d === "*", + weekdayAny: w === "*", + }; +} + +/** 严格晚于 after 的下一次触发时间(最多向后找约 4 年,找不到抛错) */ +export function nextCronAfter(expr: string | CronFields, after: Date): Date { + const f = typeof expr === "string" ? parseCron(expr) : expr; + const t = new Date(after.getTime()); + t.setSeconds(0, 0); + t.setMinutes(t.getMinutes() + 1); + const limit = after.getTime() + 4 * 366 * 24 * 3600_000; + while (t.getTime() <= limit) { + if (!f.month.has(t.getMonth() + 1)) { + t.setMonth(t.getMonth() + 1, 1); + t.setHours(0, 0, 0, 0); + continue; + } + const dayOk = f.day.has(t.getDate()); + const wdOk = f.weekday.has(t.getDay()); + const dateOk = f.dayAny && f.weekdayAny ? true : f.dayAny ? wdOk : f.weekdayAny ? dayOk : dayOk || wdOk; + if (!dateOk) { + t.setDate(t.getDate() + 1); + t.setHours(0, 0, 0, 0); + continue; + } + if (!f.hour.has(t.getHours())) { + t.setHours(t.getHours() + 1, 0, 0, 0); + continue; + } + if (!f.minute.has(t.getMinutes())) { + t.setMinutes(t.getMinutes() + 1, 0, 0); + continue; + } + return t; + } + throw new Error(`cron 在 4 年内没有触发时间:${typeof expr === "string" ? expr : "(parsed)"}`); +} diff --git a/apps/server/src/modules/schedules/index.ts b/apps/server/src/modules/schedules/index.ts new file mode 100644 index 00000000..200cdb5a --- /dev/null +++ b/apps/server/src/modules/schedules/index.ts @@ -0,0 +1,61 @@ +import type { ServerContext, ServerModule } from "../types.ts"; +import { builtinAgentCheckDefs } from "./agent-checks.ts"; +import { registerScheduledJob, scheduleRegistry } from "./registry.ts"; +import { registerScheduleRoutes } from "./routes.ts"; +import { Scheduler, type SchedulerOptions } from "./scheduler.ts"; + +export type { JobRunContext, JobRunResult, ScheduledJobDef, ScheduledJobDefaults } from "./registry.ts"; +export { registerScheduledJob, scheduleRegistry } from "./registry.ts"; + +/** + * W7-C 定时任务统一注册表 + 看板(docs/runbook/schedules.md): + * 各模块 start() 用 registerScheduledJob 登记周期任务;本模块在 io worker 上起唯一调度器(PG 咨询锁)按表驱动, + * 并登记内置 Agent 巡检模板(默认关闭)与备份占位。看板 API 见 routes.ts(全部只接受控制台会话)。 + * 须放在 SERVER_MODULES 末尾:start 时其它模块已登记完(之后再登记的也会在下一轮同步)。 + * + * TODO(W7-A):pi 模块的 refreshPiTokens 目前在 pi/index.ts 自建 setInterval, + * 待其改为 registerScheduledJob({ key: "pi.token_refresh", defaults: { group: "maintain", ... } })。 + */ +export interface SchedulesModuleOptions { + scheduler?: SchedulerOptions; + /** 缺省 true;测试可关掉内置巡检 */ + builtinChecks?: boolean; +} + +export function createSchedulesModule(opts: SchedulesModuleOptions = {}): ServerModule & { + scheduler(): Scheduler | null; +} { + let scheduler: Scheduler | null = null; + return { + name: "schedules", + scheduler: () => scheduler, + registerRoutes(app, ctx) { + registerScheduleRoutes(app, ctx, () => scheduleRegistry(ctx)); + }, + registerTasks() {}, + start(ctx: ServerContext) { + if (opts.builtinChecks !== false) for (const def of builtinAgentCheckDefs(ctx)) registerScheduledJob(ctx, def); + // 备份目前由宿主机 scripts/backup.sh(cron / 手动)执行,这里只占位,看板上显示"未配置" + registerScheduledJob(ctx, { + key: "maintain.backup", + defaults: { + name: "备份", + description: "数据库与数据目录备份;目前由 scripts/backup.sh 在宿主机执行,未接入调度", + group: "maintain", + schedule: { type: "cron", cron: "30 3 * * *" }, + enabled: false, + available: false, + }, + run: async () => ({ status: "skipped", summary: "未配置" }), + }); + scheduler = new Scheduler(ctx, scheduleRegistry(ctx), opts.scheduler); + scheduler.start(); + }, + async stop() { + await scheduler?.stop(); + scheduler = null; + }, + }; +} + +export const schedulesModule = createSchedulesModule(); diff --git a/apps/server/src/modules/schedules/registry.ts b/apps/server/src/modules/schedules/registry.ts new file mode 100644 index 00000000..9f409a9b --- /dev/null +++ b/apps/server/src/modules/schedules/registry.ts @@ -0,0 +1,99 @@ +import type { ScheduleGroup, ScheduleKind, ScheduleSpec, ScheduleTrigger } from "@umt/contracts"; +import type { ServerContext } from "../types.ts"; + +// 定时任务注册表(进程内):模块在 start() 里登记 {key, defaults, run},schedules 模块的调度器 +// 把定义同步进 scheduled_job 表(不覆盖用户改过的开关/调度),按 next_run_at 调用 run。 +// 各模块只 import 本文件(不依赖调度器实现);没装 schedules 模块时登记只是一张内存表,不会执行。 + +export interface ScheduledJobDefaults { + name: string; + description?: string; + group: ScheduleGroup; + schedule: ScheduleSpec; + /** 缺省 true */ + enabled?: boolean; + /** 任务参数(看板可改;task 浅合并,agent_check 整体替换) */ + config?: Record; + /** false = 占位(如备份未配置),不执行、不可开启 */ + available?: boolean; + /** 首次登记时何时第一次运行:now(缺省,尽快)/ after_interval(一个周期后) */ + firstRun?: "now" | "after_interval"; +} + +export interface JobRunContext { + jobId: string; + /** scheduled_job_run.id */ + runId: string; + trigger: Exclude; + now: Date; + /** 当前任务参数(表里的 config) */ + config: Record; + /** + * 时间槽:按调度周期取整的稳定串(interval:floor(now/everyMs);cron:本次计划时间)。 + * 入队幂等键用它拼,多实例 / 重复触发同槽只入队一次;立即执行用 manual:。 + */ + slot: string; +} + +export interface JobRunResult { + /** 缺省 ok;业务侧主动放弃(如只读对账模式)返回 skipped */ + status?: "ok" | "skipped"; + summary?: string | null; + agentRunId?: string | null; +} + +export type JobRunFn = (c: JobRunContext) => Promise; + +export interface ScheduledJobDef { + /** 稳定 key(表主键),如 feeds.poll */ + key: string; + kind?: ScheduleKind; + defaults: ScheduledJobDefaults; + run: JobRunFn; +} + +export interface ScheduleRegistry { + /** 每次登记 +1,调度器据此判断是否需要重新同步到表 */ + readonly version: number; + get(key: string): ScheduledJobDef | undefined; + list(): ScheduledJobDef[]; + register(def: ScheduledJobDef): void; + unregister(key: string): void; +} + +class MemoryRegistry implements ScheduleRegistry { + version = 0; + private defs = new Map(); + get(key: string) { + return this.defs.get(key); + } + list() { + return [...this.defs.values()]; + } + register(def: ScheduledJobDef) { + if (!/^[a-z][a-z0-9_.:-]{1,120}$/.test(def.key)) throw new Error(`定时任务 key 不合法:${def.key}`); + this.defs.set(def.key, def); + this.version += 1; + } + unregister(key: string) { + if (this.defs.delete(key)) this.version += 1; + } +} + +const registries = new WeakMap(); + +/** 取 ctx 对应的注册表(同一 ServerContext 共用一张) */ +export function scheduleRegistry(ctx: Pick): ScheduleRegistry { + const k = ctx as object; + let r = registries.get(k); + if (!r) { + r = new MemoryRegistry(); + registries.set(k, r); + } + return r; +} + +/** 模块 start() 里调用:登记一个周期任务(同 key 重复登记以最后一次为准) */ +export function registerScheduledJob(ctx: Pick, def: ScheduledJobDef): void { + scheduleRegistry(ctx).register(def); +} diff --git a/apps/server/src/modules/schedules/routes.ts b/apps/server/src/modules/schedules/routes.ts new file mode 100644 index 00000000..53e6e267 --- /dev/null +++ b/apps/server/src/modules/schedules/routes.ts @@ -0,0 +1,141 @@ +import { createRoute, type RouteConfig } from "@hono/zod-openapi"; +import { + type Actor, + type AgentCheckCreate, + Api, + apiResponse, + type EndpointDef, + SCHEDULE_CHANGED_EVENT, + type ScheduleUpdate, +} from "@umt/contracts"; +import type { z } from "zod"; +import { requireAdminSession, requireScope } from "../../http/auth.ts"; +import type { ModuleApp, ServerContext } from "../types.ts"; +import type { ScheduleRegistry } from "./registry.ts"; +import { + board, + createAgentCheck, + deleteJob, + jobView, + listRuns, + requestRun, + skipNext, + syncDefinitions, + updateJob, +} from "./store.ts"; + +type Infer = T extends z.ZodType ? z.infer : Record; +interface Input { + params: Infer; + query: Infer; + body: Infer; + actor: Actor | undefined; +} + +/** 合同 EndpointDef → OpenAPI 路由 + requireScope + 控制台会话;出参按合同 parse */ +function route(app: ModuleApp, def: D, handler: (i: Input) => Promise) { + const request: RouteConfig["request"] = {}; + if (def.params) request.params = def.params as never; + if (def.query) request.query = def.query as never; + if (def.body) request.body = { content: { "application/json": { schema: def.body as never } }, required: true }; + const r = createRoute({ + method: def.method, + path: def.path, + summary: def.summary, + tags: ["schedules"], + middleware: [requireScope(def.scope), ...(def.auth === "session" ? [requireAdminSession()] : [])] as never, + request, + responses: { + 200: { description: def.summary, content: { "application/json": { schema: apiResponse(def.response) } } }, + }, + }); + // biome-ignore lint/suspicious/noExplicitAny: 路由泛型在封装里擦除,入参已由 zod-openapi 校验 + app.openapi(r, (async (c: any) => { + const input = { + params: def.params ? c.req.valid("param") : {}, + query: def.query ? c.req.valid("query") : {}, + body: def.body ? c.req.valid("json") : {}, + actor: c.get("actor"), + } as Input; + const data = def.response.parse(await handler(input)); + return c.json({ ok: true, data }, 200); + // biome-ignore lint/suspicious/noExplicitAny: 同上 + }) as any); +} + +async function audit(ctx: ServerContext, actor: Actor | undefined, action: string, id: string, payload = {}) { + try { + await ctx.db + .insertInto("audit_event") + .values({ + actor_type: actor?.type === "api_key" ? "api_key" : actor ? "user" : "system", + actor_id: actor?.id ?? null, + action, + target_type: "scheduled_job", + target_id: id, + payload: JSON.stringify(payload), + }) + .execute(); + } catch (err) { + ctx.logger.warn({ err, action }, "审计写入失败"); + } +} + +export function registerScheduleRoutes(app: ModuleApp, ctx: ServerContext, registry: () => ScheduleRegistry) { + const changed = (jobId: string) => void ctx.events.publish(SCHEDULE_CHANGED_EVENT, { jobId, status: null }); + // api 与 worker 同进程(role all)时,worker 还没同步过也能先看到内置定义 + let synced = -1; + const ensureSynced = async () => { + const r = registry(); + if (r.version === synced || r.list().length === 0) return; + const v = r.version; + await syncDefinitions(ctx.db, r.list(), new Date()); + synced = v; + }; + + route(app, Api.schedules.list, async () => { + await ensureSynced(); + return await board(ctx.db); + }); + + route(app, Api.schedules.create, async ({ body, actor }) => { + const row = await createAgentCheck(ctx.db, body as AgentCheckCreate, new Date()); + await audit(ctx, actor, "schedule.create", row.id, { name: row.name }); + changed(row.id); + return jobView(row); + }); + + route(app, Api.schedules.update, async ({ params, body, actor }) => { + const row = await updateJob(ctx.db, params.id, body as ScheduleUpdate, new Date()); + await audit(ctx, actor, "schedule.update", row.id, body as Record); + changed(row.id); + return jobView(row); + }); + + route(app, Api.schedules.remove, async ({ params, actor }) => { + await deleteJob(ctx.db, params.id); + await audit(ctx, actor, "schedule.delete", params.id); + changed(params.id); + return { ok: true as const }; + }); + + route(app, Api.schedules.runNow, async ({ params, actor }) => { + const { row, queued } = await requestRun(ctx.db, params.id); + if (queued) { + await audit(ctx, actor, "schedule.run_now", row.id); + changed(row.id); + } + return { job: jobView(row), queued }; + }); + + route(app, Api.schedules.skip, async ({ params, actor }) => { + const row = await skipNext(ctx.db, params.id, new Date()); + await audit(ctx, actor, "schedule.skip", row.id, { nextRunAt: row.next_run_at?.toISOString() ?? null }); + changed(row.id); + return jobView(row); + }); + + route(app, Api.schedules.runs, async ({ params, query }) => ({ + items: await listRuns(ctx.db, params.id, query.limit), + })); +} diff --git a/apps/server/src/modules/schedules/scheduler.ts b/apps/server/src/modules/schedules/scheduler.ts new file mode 100644 index 00000000..f96a3496 --- /dev/null +++ b/apps/server/src/modules/schedules/scheduler.ts @@ -0,0 +1,326 @@ +import { SCHEDULE_CHANGED_EVENT } from "@umt/contracts"; +import { type DB, type Kysely, type ScheduledJobRow, sql } from "@umt/db"; +import type { Logger } from "pino"; +import type { EventBusPort } from "../../http/sse.ts"; +import type { ServerContext } from "../types.ts"; +import { runAgentCheck } from "./agent-checks.ts"; +import type { JobRunFn, JobRunResult, ScheduleRegistry } from "./registry.ts"; +import { computeNext, LOCK_NS, parseSpec, slotOf, syncDefinitions, trimRuns } from "./store.ts"; + +// 调度器:只在 io worker(或 all)运行;多个实例用 pg_try_advisory_lock 选出唯一执行者(锁在一条专用连接上, +// 连接断开即释放,其它实例下一轮接手)。执行者每 tickMs 查一次到期(enabled 且 next_run_at 已到)或被请求 +// 立即执行(run_requested_at)的任务,逐个异步执行(不同任务并行,同一任务不重入),结果写回表与运行历史。 +// interval 任务的下次时间 = 本次结束 + 周期(与原来的自调度循环一致);cron 任务取下一个匹配时间。 +// 立即执行不改下次运行时间。 + +export interface SchedulerOptions { + /** 查到期任务的间隔(缺省 1s) */ + tickMs?: number; + /** 抢锁失败后的重试间隔(缺省 5s) */ + lockRetryMs?: number; + /** 回填巡检 Agent 结果的间隔(缺省 15s) */ + harvestMs?: number; + /** 停止时等待进行中任务的上限(缺省 10s) */ + stopWaitMs?: number; + now?: () => Date; +} + +const TERMINAL_AGENT = ["succeeded", "failed", "cancelled", "budget_exceeded"]; + +const errText = (err: unknown) => (err instanceof Error ? err.message : String(err)).slice(0, 2000); + +export class Scheduler { + private stopped = false; + private wake: (() => void) | null = null; + private loop: Promise | null = null; + private inflight = new Map>(); + private syncedVersion = -1; + private lastHarvest = 0; + leader = false; + private readonly db: Kysely; + private readonly log: Logger; + private readonly events: EventBusPort; + private readonly now: () => Date; + + constructor( + private readonly ctx: ServerContext, + private readonly registry: ScheduleRegistry, + private readonly opts: SchedulerOptions = {}, + ) { + this.db = ctx.db; + this.log = ctx.logger.child({ component: "scheduler" }); + this.events = ctx.events; + this.now = opts.now ?? (() => new Date()); + } + + start(): void { + if (this.loop) return; + this.stopped = false; + this.loop = this.leaderLoop(); + } + + async stop(): Promise { + this.stopped = true; + this.wake?.(); + await this.loop; + this.loop = null; + const pending = [...this.inflight.values()]; + if (pending.length) { + await Promise.race([Promise.allSettled(pending), this.sleep(this.opts.stopWaitMs ?? 10_000, true)]); + } + } + + /** 可被 stop 打断的等待 */ + private sleep(ms: number, ignoreStop = false): Promise { + return new Promise((resolve) => { + const t = setTimeout(done, ms); + function done() { + clearTimeout(t); + resolve(); + } + if (!ignoreStop) this.wake = done; + }); + } + + private async leaderLoop(): Promise { + const retry = this.opts.lockRetryMs ?? 5000; + while (!this.stopped) { + try { + await this.db.connection().execute(async (conn) => { + while (!this.stopped) { + const r = await sql<{ ok: boolean }>` + select pg_try_advisory_lock(${LOCK_NS}::int4, hashtext(current_schema())) as ok`.execute(conn); + if (r.rows[0]?.ok) break; + await this.sleep(retry); + } + if (this.stopped) return; + this.leader = true; + this.log.info("调度器取得执行权"); + try { + await this.recoverInterrupted(); + while (!this.stopped) { + await this.tick(); + await this.sleep(this.opts.tickMs ?? 1000); + // 探活:专用连接断了锁也就没了,抛错后重新抢锁 + await sql`select 1`.execute(conn); + } + } finally { + this.leader = false; + await sql`select pg_advisory_unlock(${LOCK_NS}::int4, hashtext(current_schema()))` + .execute(conn) + .catch(() => undefined); + } + }); + } catch (err) { + this.leader = false; + this.log.warn({ err: { message: errText(err) } }, "调度器循环出错,稍后重试"); + if (!this.stopped) await this.sleep(retry); + } + } + } + + /** 上一个执行者中断时留下的 running 状态:标为失败 */ + private async recoverInterrupted(): Promise { + const msg = "调度器重启,上次运行中断"; + await this.db + .updateTable("scheduled_job_run") + .set({ status: "failed", error: msg, finished_at: this.now() }) + .where("status", "=", "running") + .execute(); + await this.db + .updateTable("scheduled_job") + .set({ running_since: null, last_status: "failed", last_error: msg }) + .where("running_since", "is not", null) + .execute(); + } + + /** 一轮:同步定义 → 找到期任务 → 异步执行;定期回填巡检结果。测试可直接调用 */ + async tick(): Promise { + const now = this.now(); + if (this.registry.version !== this.syncedVersion) { + const v = this.registry.version; + await syncDefinitions(this.db, this.registry.list(), now); + this.syncedVersion = v; + } + const busy = [...this.inflight.keys()]; + const due = await this.db + .selectFrom("scheduled_job") + .selectAll() + .where("available", "=", true) + .where((eb) => + eb.or([ + eb("run_requested_at", "is not", null), + eb.and([eb("enabled", "=", true), eb("next_run_at", "<=", now)]), + ]), + ) + .$if(busy.length > 0, (q) => q.where("id", "not in", busy)) + .orderBy("next_run_at") + .limit(50) + .execute(); + const started: string[] = []; + for (const job of due) { + const fn = this.runFor(job); + if (!fn) continue; + const p = this.execute(job, fn).finally(() => this.inflight.delete(job.id)); + this.inflight.set(job.id, p); + started.push(job.id); + } + if (Date.now() - this.lastHarvest >= (this.opts.harvestMs ?? 15_000)) { + this.lastHarvest = Date.now(); + await this.harvest().catch((err) => this.log.warn({ err: { message: errText(err) } }, "巡检结果回填失败")); + } + return started; + } + + /** 等进行中的任务全部结束(测试用) */ + async idle(): Promise { + while (this.inflight.size) await Promise.allSettled([...this.inflight.values()]); + } + + private runFor(job: ScheduledJobRow): JobRunFn | null { + const def = this.registry.get(job.id); + if (def) return def.run; + // 自定义巡检不在代码里登记,共用巡检执行体 + if (job.kind === "agent_check") return (c) => runAgentCheck(this.ctx, c); + return null; + } + + private publish(jobId: string, status: string | null) { + void this.events.publish(SCHEDULE_CHANGED_EVENT, { jobId, status }); + } + + private async execute(job: ScheduledJobRow, fn: JobRunFn): Promise { + const manual = job.run_requested_at !== null; + const spec = parseSpec(job.schedule); + const startedAt = this.now(); + const t0 = performance.now(); + // 认领:清掉请求标志、标运行中(条件更新,别的执行者抢先认领过就放弃) + const claimed = await this.db + .updateTable("scheduled_job") + .set({ running_since: startedAt, last_status: "running", run_requested_at: null }) + .where("id", "=", job.id) + .where("running_since", "is", null) + .returning("id") + .executeTakeFirst(); + if (!claimed) return; + const run = await this.db + .insertInto("scheduled_job_run") + .values({ job_id: job.id, trigger: manual ? "manual" : "schedule", status: "running", started_at: startedAt }) + .returning("id") + .executeTakeFirstOrThrow(); + if (manual || job.last_status !== "ok") this.publish(job.id, "running"); + + let status: "ok" | "skipped" | "failed" = "ok"; + let error: string | null = null; + let result: JobRunResult | undefined; + try { + result = + (await fn({ + jobId: job.id, + runId: run.id, + trigger: manual ? "manual" : "schedule", + now: startedAt, + config: (job.config ?? {}) as Record, + slot: manual ? `manual:${run.id}` : slotOf(spec, startedAt, job.next_run_at), + })) ?? undefined; + status = result?.status ?? "ok"; + } catch (err) { + status = "failed"; + error = errText(err); + this.log.error({ err: { message: error }, job: job.id }, "定时任务执行失败"); + } + const finishedAt = this.now(); + const durationMs = Math.max(0, Math.round(performance.now() - t0)); + // 计划触发(或从未排过)才推进下次时间;立即执行保持原计划 + const advance = !manual || job.next_run_at === null || job.next_run_at <= startedAt; + try { + await this.db.transaction().execute(async (trx) => { + const cur = await trx + .selectFrom("scheduled_job") + .select(["schedule"]) + .where("id", "=", job.id) + .executeTakeFirst(); + if (!cur) return; // 运行期间被删除 + const liveSpec = parseSpec(cur.schedule); + await trx + .updateTable("scheduled_job") + .set({ + running_since: null, + last_run_at: startedAt, + last_status: status, + last_duration_ms: durationMs, + last_error: error, + run_count: sql`run_count + 1`, + ...(status === "failed" ? { fail_count: sql`fail_count + 1` } : {}), + ...(advance + ? { next_run_at: computeNext(liveSpec, liveSpec.type === "interval" ? finishedAt : startedAt) } + : {}), + }) + .where("id", "=", job.id) + .execute(); + await trx + .updateTable("scheduled_job_run") + .set({ + status, + finished_at: finishedAt, + duration_ms: durationMs, + error, + summary: result?.summary ?? null, + agent_run_id: result?.agentRunId ?? null, + }) + .where("id", "=", run.id) + .execute(); + await trimRuns(trx, job.id); + }); + } catch (err) { + this.log.error({ err: { message: errText(err) }, job: job.id }, "定时任务结果写回失败"); + } + if (manual || status !== job.last_status || status === "failed") this.publish(job.id, status); + } + + /** 巡检入队的 agent_run 结束后:回填终态与摘要;配置了 reportAsDecision 的写一条 info 待处理 */ + async harvest(): Promise { + const rows = await sql<{ + id: string; + job_id: string; + job_name: string; + config: Record; + agent_run_id: string; + status: string; + summary: string | null; + }>` + select r.id, r.job_id, j.name as job_name, j.config, r.agent_run_id, a.status, a.summary + from scheduled_job_run r + join scheduled_job j on j.id = r.job_id + join agent_run a on a.id = r.agent_run_id + where r.agent_status is null and a.status = any(${TERMINAL_AGENT}) + limit 50`.execute(this.db); + for (const r of rows.rows) { + const summary = (r.summary ?? "").trim() || `Agent 运行结束:${r.status}`; + await this.db + .updateTable("scheduled_job_run") + .set({ agent_status: r.status, summary: summary.slice(0, 4000) }) + .where("id", "=", r.id) + .execute(); + if (r.config?.reportAsDecision === true && r.status === "succeeded") { + const ins = await sql<{ id: string }>` + insert into decision (kind, severity, title, summary, subject_type, subject_id, dedupe_key, evidence, agent_run_id) + values ('other', 'info', ${`巡检报告:${r.job_name}`}, ${summary.slice(0, 4000)}, 'agent_run', ${r.agent_run_id}, + ${`schedule-report:${r.id}`}, ${JSON.stringify({ scheduledJobId: r.job_id, scheduledRunId: r.id })}::jsonb, + ${r.agent_run_id}) + on conflict do nothing + returning id`.execute(this.db); + const d = ins.rows[0]; + if (d) + await this.ctx.outbox.emit( + this.db, + "decision.created", + { decisionId: d.id, kind: "other", severity: "info", title: `巡检报告:${r.job_name}` }, + { idempotencyKey: `decision.created:${d.id}` }, + ); + } + this.publish(r.job_id, null); + } + return rows.rows.length; + } +} diff --git a/apps/server/src/modules/schedules/store.ts b/apps/server/src/modules/schedules/store.ts new file mode 100644 index 00000000..c5271f4b --- /dev/null +++ b/apps/server/src/modules/schedules/store.ts @@ -0,0 +1,322 @@ +import { randomUUID } from "node:crypto"; +import { + AgentCheckConfig, + type AgentCheckCreate, + type ScheduleBoard, + type ScheduledJob, + type ScheduledJobRun, + ScheduleSpec, + type ScheduleUpdate, +} from "@umt/contracts"; +import { type DB, type Kysely, type ScheduledJobRow, type ScheduledJobRunRow, sql } from "@umt/db"; +import { ApiException } from "../../http/errors.ts"; +import { nextCronAfter, parseCron } from "./cron.ts"; +import type { ScheduledJobDef } from "./registry.ts"; + +// scheduled_job / scheduled_job_run 读写:定义同步(不覆盖用户改动)、视图映射、看板操作。 + +/** 咨询锁命名空间('umts');第二个键是 hashtext(current_schema()),测试 schema 之间互不影响 */ +export const LOCK_NS = 0x756d7473; +/** 每个任务保留的运行历史条数 */ +export const RUN_HISTORY_KEEP = 50; + +const iso = (d: Date | null) => (d ? d.toISOString() : null); + +export function parseSpec(v: unknown): ScheduleSpec { + return ScheduleSpec.parse(v); +} + +/** from 之后的下一次计划时间 */ +export function computeNext(spec: ScheduleSpec, from: Date): Date { + return spec.type === "interval" ? new Date(from.getTime() + spec.everyMs) : nextCronAfter(spec.cron, from); +} + +/** 时间槽:interval 按周期取整;cron 用本次计划时间 */ +export function slotOf(spec: ScheduleSpec, now: Date, planned: Date | null): string { + if (spec.type === "interval") return String(Math.floor(now.getTime() / spec.everyMs)); + return String(Math.floor((planned ?? now).getTime() / 60_000)); +} + +export function validateSpec(spec: ScheduleSpec): void { + if (spec.type === "cron") { + try { + parseCron(spec.cron); + } catch (err) { + throw new ApiException("validation_failed", (err as Error).message); + } + } +} + +export function jobView(r: ScheduledJobRow): ScheduledJob { + return { + id: r.id, + name: r.name, + description: r.description, + group: r.board_group, + kind: r.kind, + builtin: r.builtin, + available: r.available, + enabled: r.enabled, + schedule: parseSpec(r.schedule), + defaultSchedule: parseSpec(r.default_schedule), + customized: r.customized, + config: (r.config ?? {}) as Record, + lastRunAt: iso(r.last_run_at), + lastStatus: r.last_status, + lastDurationMs: r.last_duration_ms, + lastError: r.last_error, + nextRunAt: iso(r.next_run_at), + runRequestedAt: iso(r.run_requested_at), + runningSince: iso(r.running_since), + runCount: r.run_count, + failCount: r.fail_count, + createdAt: r.created_at.toISOString(), + updatedAt: r.updated_at.toISOString(), + }; +} + +export function runView(r: ScheduledJobRunRow): ScheduledJobRun { + return { + id: r.id, + jobId: r.job_id, + trigger: r.trigger, + status: r.status, + startedAt: r.started_at.toISOString(), + finishedAt: iso(r.finished_at), + durationMs: r.duration_ms, + error: r.error, + summary: r.summary, + agentRunId: r.agent_run_id, + agentStatus: r.agent_status, + createdAt: r.created_at.toISOString(), + }; +} + +/** + * 把代码登记的定义同步进表:新行按默认值插入;已有行刷新名称/分组/默认值, + * 用户改过(customized)的 enabled/schedule 保留;config 以表内值为准、补上新增的默认键。 + * 未改过且默认调度变了的行,下次运行改为立即(随后按新周期)。 + * 同步时代码里已不再登记的内置 task 行标为不可用(不会执行,看板显示"未登记")。 + */ +export async function syncDefinitions(db: Kysely, defs: readonly ScheduledJobDef[], now: Date): Promise { + for (const d of defs) { + const spec = parseSpec(d.defaults.schedule); + const enabled = d.defaults.enabled ?? true; + const first = d.defaults.firstRun === "after_interval" ? computeNext(spec, now) : now; + await sql` + insert into scheduled_job (id, name, description, board_group, kind, builtin, available, enabled, default_enabled, + schedule, default_schedule, config, next_run_at) + values (${d.key}, ${d.defaults.name}, ${d.defaults.description ?? null}, ${d.defaults.group}, ${d.kind ?? "task"}, + true, ${d.defaults.available ?? true}, ${enabled}, ${enabled}, + ${JSON.stringify(spec)}::jsonb, ${JSON.stringify(spec)}::jsonb, ${JSON.stringify(d.defaults.config ?? {})}::jsonb, + ${first}) + on conflict (id) do update set + name = excluded.name, + description = excluded.description, + board_group = excluded.board_group, + kind = excluded.kind, + builtin = true, + available = excluded.available, + default_enabled = excluded.default_enabled, + default_schedule = excluded.default_schedule, + enabled = case when scheduled_job.customized then scheduled_job.enabled else excluded.enabled end, + schedule = case when scheduled_job.customized then scheduled_job.schedule else excluded.schedule end, + config = excluded.config || scheduled_job.config, + next_run_at = case + when not scheduled_job.customized and scheduled_job.schedule is distinct from excluded.schedule then ${now} + else coalesce(scheduled_job.next_run_at, excluded.next_run_at) end`.execute(db); + } + const keys = defs.map((d) => d.key); + await db + .updateTable("scheduled_job") + .set({ available: false }) + .where("builtin", "=", true) + .where("kind", "=", "task") + .where("available", "=", true) + .$if(keys.length > 0, (q) => q.where("id", "not in", keys)) + .execute(); +} + +export async function getJob(db: Kysely, id: string): Promise { + const row = await db.selectFrom("scheduled_job").selectAll().where("id", "=", id).executeTakeFirst(); + if (!row) throw new ApiException("not_found", `定时任务不存在:${id}`); + return row; +} + +export async function schedulerOnline(db: Kysely): Promise { + const r = await sql<{ on: boolean }>` + select exists( + select 1 from pg_locks + where locktype = 'advisory' and objsubid = 2 and granted + and classid = ${LOCK_NS}::int4::oid and objid = hashtext(current_schema())::oid + ) as on`.execute(db); + return r.rows[0]?.on ?? false; +} + +export async function board(db: Kysely): Promise { + const rows = await db + .selectFrom("scheduled_job") + .selectAll() + .orderBy("board_group") + .orderBy("builtin", "desc") + .orderBy("created_at") + .orderBy("id") + .execute(); + const items = rows.map(jobView); + return { + items, + stats: { + total: items.length, + enabled: items.filter((j) => j.enabled && j.available).length, + failed: items.filter((j) => j.lastStatus === "failed").length, + running: items.filter((j) => j.runningSince !== null).length, + }, + schedulerOnline: await schedulerOnline(db), + }; +} + +export async function updateJob( + db: Kysely, + id: string, + patch: ScheduleUpdate, + now: Date, +): Promise { + const row = await getJob(db, id); + const set: Record = {}; + if (patch.reset) { + const spec = parseSpec(row.default_schedule); + Object.assign(set, { + enabled: row.default_enabled, + schedule: JSON.stringify(spec), + customized: false, + next_run_at: computeNext(spec, now), + }); + } + if (patch.schedule) { + validateSpec(patch.schedule); + Object.assign(set, { + schedule: JSON.stringify(patch.schedule), + customized: true, + next_run_at: computeNext(patch.schedule, now), + }); + } + if (patch.enabled !== undefined) { + if (patch.enabled && !row.available) throw new ApiException("conflict", "该任务未配置,不能开启"); + set.enabled = patch.enabled; + set.customized = true; + if (patch.enabled && !row.enabled && !set.next_run_at) { + set.next_run_at = computeNext(parseSpec(set.schedule ? JSON.parse(set.schedule as string) : row.schedule), now); + } + } + if (patch.name !== undefined || patch.description !== undefined) { + if (row.builtin) throw new ApiException("conflict", "内置任务不能改名"); + if (patch.name !== undefined) set.name = patch.name; + if (patch.description !== undefined) set.description = patch.description || null; + } + if (patch.config) { + if (row.kind === "agent_check") { + const parsed = AgentCheckConfig.safeParse(patch.config); + if (!parsed.success) throw new ApiException("validation_failed", parsed.error.issues[0]?.message ?? "参数无效"); + set.config = JSON.stringify(parsed.data); + } else { + set.config = JSON.stringify({ ...((row.config ?? {}) as object), ...patch.config }); + } + } + return await db.updateTable("scheduled_job").set(set).where("id", "=", id).returningAll().executeTakeFirstOrThrow(); +} + +/** 立即执行:置 run_requested_at(已在排队或运行中则合并,不重复) */ +export async function requestRun(db: Kysely, id: string): Promise<{ row: ScheduledJobRow; queued: boolean }> { + const row = await getJob(db, id); + if (!row.available) throw new ApiException("conflict", "该任务未配置,不能执行"); + const upd = await db + .updateTable("scheduled_job") + .set({ run_requested_at: sql`now()` }) + .where("id", "=", id) + .where("run_requested_at", "is", null) + .where("running_since", "is", null) + .returningAll() + .executeTakeFirst(); + return upd ? { row: upd, queued: true } : { row, queued: false }; +} + +/** 跳过本次:下次运行顺延一个周期,记一条 skip 历史 */ +export async function skipNext(db: Kysely, id: string, now: Date): Promise { + const row = await getJob(db, id); + const spec = parseSpec(row.schedule); + const base = row.next_run_at && row.next_run_at > now ? row.next_run_at : now; + const next = computeNext(spec, base); + return await db.transaction().execute(async (trx) => { + const upd = await trx + .updateTable("scheduled_job") + .set({ next_run_at: next }) + .where("id", "=", id) + .returningAll() + .executeTakeFirstOrThrow(); + await trx + .insertInto("scheduled_job_run") + .values({ + job_id: id, + trigger: "skip", + status: "skipped", + started_at: now, + finished_at: now, + duration_ms: 0, + summary: `跳过 ${base.toISOString()},下次 ${next.toISOString()}`, + }) + .execute(); + await trimRuns(trx, id); + return upd; + }); +} + +export async function trimRuns(db: Kysely, jobId: string, keep = RUN_HISTORY_KEEP): Promise { + await sql` + delete from scheduled_job_run where job_id = ${jobId} and id not in ( + select id from scheduled_job_run where job_id = ${jobId} order by started_at desc limit ${keep})`.execute(db); +} + +export async function listRuns(db: Kysely, id: string, limit: number): Promise { + await getJob(db, id); + const rows = await db + .selectFrom("scheduled_job_run") + .selectAll() + .where("job_id", "=", id) + .orderBy("started_at", "desc") + .limit(limit) + .execute(); + return rows.map(runView); +} + +export async function createAgentCheck(db: Kysely, body: AgentCheckCreate, now: Date): Promise { + const spec = parseSpec(body.schedule); + validateSpec(spec); + const config = AgentCheckConfig.parse(body.config); + const enabled = body.enabled ?? false; + return await db + .insertInto("scheduled_job") + .values({ + id: `check.custom.${randomUUID().slice(0, 8)}`, + name: body.name, + description: body.description ?? null, + board_group: "agent", + kind: "agent_check", + builtin: false, + available: true, + enabled, + default_enabled: enabled, + schedule: JSON.stringify(spec), + default_schedule: JSON.stringify(spec), + customized: true, + config: JSON.stringify(config), + next_run_at: computeNext(spec, now), + }) + .returningAll() + .executeTakeFirstOrThrow(); +} + +export async function deleteJob(db: Kysely, id: string): Promise { + const row = await getJob(db, id); + if (row.builtin || row.kind !== "agent_check") throw new ApiException("conflict", "内置任务不能删除,可以停用"); + await db.deleteFrom("scheduled_job").where("id", "=", id).execute(); +} diff --git a/apps/server/src/modules/sources-identity/index.ts b/apps/server/src/modules/sources-identity/index.ts index 1dc56727..fdf08323 100644 --- a/apps/server/src/modules/sources-identity/index.ts +++ b/apps/server/src/modules/sources-identity/index.ts @@ -1,3 +1,4 @@ +import { registerScheduledJob } from "../schedules/registry.ts"; import type { ServerContext, ServerModule } from "../types.ts"; import { registerSourcesCapabilities } from "./capabilities.ts"; import { decisionRoutes } from "./routes/decisions.ts"; @@ -5,7 +6,7 @@ import { identityRoutes } from "./routes/identity.ts"; import { sourcesRoutes } from "./routes/sources.ts"; import { workbenchRoutes } from "./routes/workbench.ts"; import { type RuntimeOverrides, SourcesRuntime } from "./runtime.ts"; -import { schedulerTick } from "./scheduler.ts"; +import { readOnlyReconcile, tickFeeds, tickReviews, tickSnapshots } from "./scheduler.ts"; import { registerSourcesTasks, sourcesOutboxRoutes } from "./tasks.ts"; /** @@ -25,8 +26,6 @@ export function createSourcesIdentityModule(overrides: RuntimeOverrides = {}): S } return rt; }; - let timer: ReturnType | null = null; - let ticking: Promise | null = null; return { name: "sources-identity", @@ -45,23 +44,22 @@ export function createSourcesIdentityModule(overrides: RuntimeOverrides = {}): S registerCapabilities(registry, ctx) { registerSourcesCapabilities(registry, () => runtime(ctx)); }, + /** 周期扫描登记到定时任务注册表(schedules 模块的调度器驱动);周期是总扫描频率 */ start(ctx) { const rt = runtime(ctx); - const tick = () => { - if (ticking) return; - ticking = schedulerTick(rt) - .catch((err) => ctx.logger.warn({ err }, "sources-identity 调度失败")) - .finally(() => { - ticking = null; - }); - }; - timer = setInterval(tick, overrides.schedulerIntervalMs ?? 30_000); - tick(); - }, - async stop() { - if (timer) clearInterval(timer); - timer = null; - await ticking; + const everyMs = overrides.schedulerIntervalMs ?? 30_000; + const job = (key: string, name: string, description: string, fn: (rt: SourcesRuntime) => Promise) => + registerScheduledJob(ctx, { + key, + defaults: { name, description, group: "collect", schedule: { type: "interval", everyMs } }, + run: async () => { + if (await readOnlyReconcile(rt)) return { status: "skipped", summary: "只读对账模式" }; + return { summary: `${await fn(rt)} 项` }; + }, + }); + job("feeds.poll", "Feed 轮询", "扫描到期的 Feed 并入队抓取(每个 Feed 仍按自己的轮询间隔)", tickFeeds); + job("sites.snapshot", "站点快照", "扫描到周期的站点并入队快照", tickSnapshots); + job("identity.review", "待识别复查", "到期的待识别条目重新识别", tickReviews); }, }; } diff --git a/apps/server/src/modules/sources-identity/scheduler.ts b/apps/server/src/modules/sources-identity/scheduler.ts index 84566c26..03dea559 100644 --- a/apps/server/src/modules/sources-identity/scheduler.ts +++ b/apps/server/src/modules/sources-identity/scheduler.ts @@ -4,6 +4,8 @@ import type { SourcesRuntime } from "./runtime.ts"; // worker 后台调度:按 Feed 轮询策略发 feed.due(outbox → ingestFeed)、按站点快照周期入队 snapshotSite、 // 按待识别池 next_review_at 入队复查。幂等键按时间槽生成,多 worker 并发也不会重复。 +// 三项分别登记为定时任务(feeds.poll / sites.snapshot / identity.review,由 schedules 模块的调度器驱动, +// 这里的周期只是总扫描频率,每个 Feed / 站点仍按自己的策略判断是否到期);schedulerTick 依次跑三项(测试用)。 export interface TickResult { feedsDue: number; @@ -14,14 +16,24 @@ export interface TickResult { const REVIEW_LEASE_MS = 10 * 60_000; -export async function schedulerTick(rt: SourcesRuntime): Promise { - const { ctx } = rt; - const automation = await ctx.settings.get("automation"); - // 恢复后只读对账模式:禁止自动写入 - if (automation.readOnlyReconcile) return { feedsDue: 0, snapshotsDue: 0, reviewsDue: 0, skipped: "read_only" }; - const now = rt.now(); +/** 恢复后只读对账模式:禁止自动写入 */ +export async function readOnlyReconcile(rt: SourcesRuntime): Promise { + return (await rt.ctx.settings.get("automation")).readOnlyReconcile; +} - // Feed:到期(next_fetch_at 空或已到)且不在退避期、站点(若有)开启 +export async function schedulerTick(rt: SourcesRuntime): Promise { + if (await readOnlyReconcile(rt)) return { feedsDue: 0, snapshotsDue: 0, reviewsDue: 0, skipped: "read_only" }; + return { + feedsDue: await tickFeeds(rt), + snapshotsDue: await tickSnapshots(rt), + reviewsDue: await tickReviews(rt), + }; +} + +/** Feed:到期(next_fetch_at 空或已到)且不在退避期、站点(若有)开启 → feed.due */ +export async function tickFeeds(rt: SourcesRuntime): Promise { + const { ctx } = rt; + const now = rt.now(); const feedsDue = await ctx.db.transaction().execute(async (trx) => { const due = await sql<{ id: string; poll_interval_sec: number }>` select f.id, f.poll_interval_sec from feed_source f left join site s on s.id = f.site_id @@ -47,8 +59,13 @@ export async function schedulerTick(rt: SourcesRuntime): Promise { } return due.rows.length; }); + return feedsDue; +} - // 站点快照:开启快照且到周期 +/** 站点快照:开启快照且到周期 → snapshotSite */ +export async function tickSnapshots(rt: SourcesRuntime): Promise { + const { ctx } = rt; + const now = rt.now(); const sites = await sql<{ id: string; snapshot_interval_sec: number }>` select id, snapshot_interval_sec from site where enabled and snapshot_enabled and snapshot_interval_sec is not null @@ -63,8 +80,13 @@ export async function schedulerTick(rt: SourcesRuntime): Promise { { idempotencyKey: `snapshotSite:${s.id}:${slot}`, traceId: randomUUID() }, ); } + return sites.rows.length; +} - // 待识别池复查:到期的先顺延一个租约,避免下个 tick 重复入队 +/** 待识别池复查:到期的先顺延一个租约,避免下个 tick 重复入队 */ +export async function tickReviews(rt: SourcesRuntime): Promise { + const { ctx } = rt; + const now = rt.now(); const reviews = await ctx.db.transaction().execute(async (trx) => { const rows = await sql<{ id: string; source_release_id: string; attempts: number }>` select id, source_release_id, attempts from identity_candidate @@ -86,5 +108,5 @@ export async function schedulerTick(rt: SourcesRuntime): Promise { { idempotencyKey: `review:${r.id}:${r.attempts}`, traceId: randomUUID() }, ); } - return { feedsDue, snapshotsDue: sites.rows.length, reviewsDue: reviews.length }; + return reviews.length; } diff --git a/apps/server/test/api-registry.test.ts b/apps/server/test/api-registry.test.ts index 6fbdbe58..13c40632 100644 --- a/apps/server/test/api-registry.test.ts +++ b/apps/server/test/api-registry.test.ts @@ -75,6 +75,14 @@ describe("Api 注册表鉴权标注", () => { "POST /pi/upstreams/{id}/oauth/start", "POST /pi/upstreams/{id}/oauth/complete", "POST /pi/upstreams/{id}/oauth/logout", + // W7-C 定时任务看板:全部只接受控制台会话 + "GET /schedules", + "POST /schedules", + "PATCH /schedules/{id}", + "DELETE /schedules/{id}", + "POST /schedules/{id}/run-now", + "POST /schedules/{id}/skip", + "GET /schedules/{id}/runs", ].sort(), ); for (const e of listEndpoints().filter((x) => x.auth === "session")) expect(e.scope).toBe("admin"); diff --git a/apps/server/test/e2e/fixtures/modules.ts b/apps/server/test/e2e/fixtures/modules.ts index ea6ca07e..20f98dfc 100644 --- a/apps/server/test/e2e/fixtures/modules.ts +++ b/apps/server/test/e2e/fixtures/modules.ts @@ -5,6 +5,7 @@ import { makeTorrent } from "@umt/testkit"; import { createAcquisitionOrganizeModule } from "../../../src/modules/acquisition-organize/index.ts"; import { createAgentModule } from "../../../src/modules/agent/index.ts"; import { authSettingsModule } from "../../../src/modules/auth-settings/index.ts"; +import { createSchedulesModule } from "../../../src/modules/schedules/index.ts"; import { createSourcesIdentityModule } from "../../../src/modules/sources-identity/index.ts"; import type { ServerModule } from "../../../src/modules/types.ts"; @@ -64,5 +65,7 @@ export function buildModules(o: E2EModuleOptions): ServerModule[] { ...(o.fakeScripts ? { fakeScripts: o.fakeScripts } : {}), goalPollSec: 1, }), + // 定时任务调度器(驱动上面模块登记的周期任务;须在末尾) + createSchedulesModule({ builtinChecks: false, scheduler: { tickMs: 100, lockRetryMs: 200 } }), ]; } diff --git a/apps/server/test/modules.test.ts b/apps/server/test/modules.test.ts index 806da2f3..91d95fad 100644 --- a/apps/server/test/modules.test.ts +++ b/apps/server/test/modules.test.ts @@ -43,7 +43,7 @@ const pingModule: ServerModule = { }; describe("ServerModule 骨架", () => { - it("清单里是四个 W2 模块 + 文件管理 + Pi 上游,名字唯一", () => { + it("清单里是四个 W2 模块 + 文件管理 + Pi 上游 + 定时任务(末尾),名字唯一", () => { expect(SERVER_MODULES.map((m) => m.name)).toEqual([ "auth-settings", "sources-identity", @@ -51,6 +51,7 @@ describe("ServerModule 骨架", () => { "agent", "file-manager", "pi", + "schedules", ]); }); diff --git a/apps/server/test/schedules/schedules.test.ts b/apps/server/test/schedules/schedules.test.ts new file mode 100644 index 00000000..4e7beb2b --- /dev/null +++ b/apps/server/test/schedules/schedules.test.ts @@ -0,0 +1,325 @@ +import { randomUUID } from "node:crypto"; +import { sql } from "@umt/db"; +import { afterAll, beforeAll, describe, expect, it } from "vitest"; +import { ensureSetupToken } from "../../src/modules/auth-settings/auth-service.ts"; +import { nextCronAfter } from "../../src/modules/schedules/cron.ts"; +import { createSchedulesModule } from "../../src/modules/schedules/index.ts"; +import { + registerScheduledJob, + type ScheduledJobDef, + type ScheduleRegistry, + scheduleRegistry, +} from "../../src/modules/schedules/registry.ts"; +import { Scheduler } from "../../src/modules/schedules/scheduler.ts"; +import type { ServerContext } from "../../src/modules/types.ts"; +import { Browser, bearer, createHarness, type Harness, pgAvailable } from "../auth-settings/harness.ts"; +import { DEV_KEY, json } from "../helpers.ts"; + +// W7-C 定时任务注册表:登记不覆盖用户改动、调度单实例锁、立即执行幂等、失败记录、跳过本次、 +// agent_check 入队 runAgent 与结果回填、端点鉴权(只接受控制台会话)。 + +const hasPg = await pgAvailable(); +const PASSWORD = "correct horse battery"; + +describe("cron", () => { + it("按 5 段表达式取下一次时间", () => { + const base = new Date(2026, 8, 28, 10, 7, 30); // 周一 + expect(nextCronAfter("*/15 * * * *", base)).toEqual(new Date(2026, 8, 28, 10, 15, 0)); + expect(nextCronAfter("30 3 * * *", base)).toEqual(new Date(2026, 8, 29, 3, 30, 0)); + expect(nextCronAfter("0 9 * * 0", base)).toEqual(new Date(2026, 9, 4, 9, 0, 0)); + expect(nextCronAfter("0 0 1 1 *", base)).toEqual(new Date(2027, 0, 1, 0, 0, 0)); + expect(() => nextCronAfter("61 * * * *", base)).toThrow(); + }); +}); + +interface Enq { + task: string; + input: unknown; + key: string; +} + +describe.skipIf(!hasPg)("定时任务注册表(真实 PG)", () => { + let h: Harness; + let admin: Browser; + const enqueued: Enq[] = []; + const mod = createSchedulesModule(); + let ctx: ServerContext; + let reg: ScheduleRegistry; + let sched: Scheduler; + const calls = new Map(); + + const api = async (method: string, path: string, body?: unknown) => { + const res = await admin.req(method, path, body); + return { status: res.status, body: await json(res) }; + }; + const job = async (id: string) => + await h.db.selectFrom("scheduled_job").selectAll().where("id", "=", id).executeTakeFirstOrThrow(); + const counting = (key: string, extra: Partial = {}): ScheduledJobDef => ({ + key, + defaults: { name: key, group: "maintain", schedule: { type: "interval", everyMs: 60_000 } }, + run: async () => { + calls.set(key, (calls.get(key) ?? 0) + 1); + return { summary: "done" }; + }, + ...extra, + }); + + beforeAll(async () => { + h = await createHarness({ extraModules: [mod] }); + ctx = h.ctx; + // runAgent 入队用的假引擎(同键去重) + const keys = new Set(); + (ctx as { engine: ServerContext["engine"] }).engine = { + ...ctx.engine, + async enqueue(task: string, input: unknown, opts: { idempotencyKey: string }) { + const dup = keys.has(opts.idempotencyKey); + if (!dup) { + keys.add(opts.idempotencyKey); + enqueued.push({ task, input, key: opts.idempotencyKey }); + } + return { runId: randomUUID(), deduplicated: dup }; + }, + } as ServerContext["engine"]; + reg = scheduleRegistry(ctx); + sched = new Scheduler(ctx, reg, { tickMs: 3_600_000, lockRetryMs: 50, harvestMs: 0 }); + const token = await ensureSetupToken(ctx); + admin = new Browser(h.app); + expect( + (await admin.req("POST", "/auth/setup", { setupToken: token, username: "admin", password: PASSWORD })).status, + ).toBe(200); + }, 60_000); + + afterAll(async () => { + await sched?.stop(); + await h?.close(); + }); + + it("登记默认值;用户改过的开关与调度不被再次登记覆盖;reset 恢复默认", async () => { + registerScheduledJob(ctx, counting("test.keep")); + await sched.tick(); + await sched.idle(); + let row = await job("test.keep"); + expect(row).toMatchObject({ enabled: true, customized: false, schedule: { type: "interval", everyMs: 60_000 } }); + + const r = await api("PATCH", "/schedules/test.keep", { + enabled: false, + schedule: { type: "interval", everyMs: 120_000 }, + }); + expect(r.status).toBe(200); + expect(r.body.data).toMatchObject({ enabled: false, customized: true }); + + // 模块再次启动,默认值变了 + registerScheduledJob( + ctx, + counting("test.keep", { + defaults: { name: "改名了", group: "download", schedule: { type: "interval", everyMs: 30_000 } }, + }), + ); + await sched.tick(); + row = await job("test.keep"); + expect(row).toMatchObject({ + name: "改名了", + board_group: "download", + enabled: false, + schedule: { type: "interval", everyMs: 120_000 }, + default_schedule: { type: "interval", everyMs: 30_000 }, + }); + + const reset = await api("PATCH", "/schedules/test.keep", { reset: true }); + expect(reset.body.data).toMatchObject({ + enabled: true, + customized: false, + schedule: { type: "interval", everyMs: 30_000 }, + }); + + // 间隔下限与 cron 校验 + expect((await api("PATCH", "/schedules/test.keep", { schedule: { type: "interval", everyMs: 500 } })).status).toBe( + 422, + ); + expect( + (await api("PATCH", "/schedules/test.keep", { schedule: { type: "cron", cron: "99 * * * *" } })).status, + ).toBe(422); + const cron = await api("PATCH", "/schedules/test.keep", { schedule: { type: "cron", cron: "0 4 * * *" } }); + expect(cron.status).toBe(200); + expect(new Date(cron.body.data.nextRunAt).getHours()).toBe(4); + }); + + it("立即执行幂等(合并重复请求)且不改下次运行时间", async () => { + registerScheduledJob( + ctx, + counting("test.now", { defaults: { ...counting("x").defaults, firstRun: "after_interval" } }), + ); + await sched.tick(); + const before = await job("test.now"); + expect(before.next_run_at!.getTime()).toBeGreaterThan(Date.now()); + const a = await api("POST", "/schedules/test.now/run-now"); + const b = await api("POST", "/schedules/test.now/run-now"); + expect(a.body.data.queued).toBe(true); + expect(b.body.data.queued).toBe(false); + const started = await sched.tick(); + expect(started).toContain("test.now"); + await sched.idle(); + expect(await sched.tick()).not.toContain("test.now"); + await sched.idle(); + expect(calls.get("test.now")).toBe(1); + const after = await job("test.now"); + expect(after.next_run_at).toEqual(before.next_run_at); + expect(after).toMatchObject({ last_status: "ok", run_count: 1, run_requested_at: null, running_since: null }); + const runs = await api("GET", "/schedules/test.now/runs"); + expect(runs.body.data.items).toHaveLength(1); + expect(runs.body.data.items[0]).toMatchObject({ trigger: "manual", status: "ok", summary: "done" }); + }); + + it("失败写入 last_error / fail_count 与历史;跳过本次顺延一个周期", async () => { + registerScheduledJob(ctx, { + key: "test.fail", + defaults: { name: "会失败", group: "maintain", schedule: { type: "interval", everyMs: 60_000 } }, + run: async () => { + throw new Error("qB 连不上"); + }, + }); + await sched.tick(); + await sched.idle(); + const row = await job("test.fail"); + expect(row).toMatchObject({ last_status: "failed", last_error: "qB 连不上", fail_count: 1, run_count: 1 }); + // interval:下次 = 结束 + 周期 + expect(row.next_run_at!.getTime() - Date.now()).toBeGreaterThan(50_000); + + const skip = await api("POST", "/schedules/test.fail/skip"); + expect(skip.status).toBe(200); + expect(new Date(skip.body.data.nextRunAt).getTime() - row.next_run_at!.getTime()).toBe(60_000); + const runs = (await api("GET", "/schedules/test.fail/runs")).body.data.items; + expect(runs.map((r: { trigger: string; status: string }) => `${r.trigger}:${r.status}`)).toEqual([ + "skip:skipped", + "schedule:failed", + ]); + expect(runs[1].error).toBe("qB 连不上"); + }); + + it("单实例锁:两个调度器只有一个执行;执行者停止后另一个接手", async () => { + const regA = scheduleRegistry({ db: {} } as ServerContext); + const regB = scheduleRegistry({ db: {} } as ServerContext); + const ran: string[] = []; + for (const [name, r] of [ + ["A", regA], + ["B", regB], + ] as const) { + r.register({ + key: "test.lock", + defaults: { name: "锁", group: "maintain", schedule: { type: "interval", everyMs: 1 } }, + run: async () => { + ran.push(name); + }, + }); + } + const a = new Scheduler(ctx, regA, { tickMs: 20, lockRetryMs: 30 }); + const b = new Scheduler(ctx, regB, { tickMs: 20, lockRetryMs: 30 }); + // 表里其它测试的任务不要被 a/b 执行:只登记了 test.lock,其余 task 行没有执行体会被跳过 + a.start(); + await new Promise((r) => setTimeout(r, 150)); + b.start(); + await new Promise((r) => setTimeout(r, 400)); + expect(a.leader).toBe(true); + expect(b.leader).toBe(false); + expect(ran.length).toBeGreaterThan(2); + expect(new Set(ran)).toEqual(new Set(["A"])); + const board = await api("GET", "/schedules"); + expect(board.body.data.schedulerOnline).toBe(true); + + await a.stop(); + await new Promise((r) => setTimeout(r, 400)); + expect(b.leader).toBe(true); + expect(ran).toContain("B"); + await b.stop(); + await sql`update scheduled_job set enabled = false where id = 'test.lock'`.execute(h.db); + }); + + it("agent_check:新建巡检 → 立即执行入队 runAgent;结束后回填结果并写入待处理", async () => { + const created = await api("POST", "/schedules", { + name: "自定义巡检", + schedule: { type: "interval", everyMs: 3_600_000 }, + config: { + goalTemplate: "巡检 {{date}}:待处理 {{openDecisions}} 条", + capabilityAllowlist: ["task.inspect", "inventory.query", "download.submit"], + budget: { maxToolCalls: 5 }, + reportAsDecision: true, + }, + }); + expect(created.status).toBe(200); + const id = created.body.data.id as string; + expect(created.body.data).toMatchObject({ kind: "agent_check", builtin: false, group: "agent", enabled: false }); + + expect((await api("POST", `/schedules/${id}/run-now`)).body.data.queued).toBe(true); + await sched.tick(); + await sched.idle(); + const [run] = (await api("GET", `/schedules/${id}/runs`)).body.data.items; + expect(run).toMatchObject({ trigger: "manual", status: "ok" }); + expect(run.agentRunId).toBeTruthy(); + const ar = await h.db + .selectFrom("agent_run") + .selectAll() + .where("id", "=", run.agentRunId) + .executeTakeFirstOrThrow(); + expect(ar).toMatchObject({ trigger: "schedule", mode: "plan_only", created_by: `schedule:${id}` }); + expect(ar.goal).toMatch(/^巡检 \d{4}-\d{2}-\d{2}:待处理 0 条$/); + // 只读巡检:execute 类能力(download.submit)被剔除 + expect((ar.scope as { capabilities: string[] }).capabilities).toEqual(["task.inspect", "inventory.query"]); + expect(ar.budget).toMatchObject({ maxToolCalls: 5 }); + expect(enqueued.filter((e) => e.task === "runAgent")).toHaveLength(1); + + // runAgent 结束 → 回填 + await h.db + .updateTable("agent_run") + .set({ status: "succeeded", summary: "一切正常" }) + .where("id", "=", ar.id) + .execute(); + expect(await sched.harvest()).toBe(1); + const [after] = (await api("GET", `/schedules/${id}/runs`)).body.data.items; + expect(after).toMatchObject({ agentStatus: "succeeded", summary: "一切正常" }); + const d = await h.db.selectFrom("decision").selectAll().where("agent_run_id", "=", ar.id).executeTakeFirstOrThrow(); + expect(d).toMatchObject({ kind: "other", severity: "info", summary: "一切正常" }); + expect(await sched.harvest()).toBe(0); + + // 自定义可改名、可删;内置不可删 + expect((await api("PATCH", `/schedules/${id}`, { name: "夜间巡检" })).body.data.name).toBe("夜间巡检"); + expect((await api("DELETE", "/schedules/test.keep")).status).toBe(409); + expect((await api("DELETE", `/schedules/${id}`)).status).toBe(200); + expect((await api("GET", `/schedules/${id}/runs`)).status).toBe(404); + }); + + it("内置巡检默认关闭;触发条件不满足时计划触发记 skipped,不唤醒 Agent", async () => { + await mod.start?.(ctx); + await mod.stop?.(); + await sched.tick(); + const board = (await api("GET", "/schedules")).body.data; + const checks = board.items.filter((j: { kind: string }) => j.kind === "agent_check"); + expect(checks.map((j: { id: string }) => j.id).sort()).toEqual([ + "check.downloads", + "check.pending", + "check.weekly", + ]); + expect(checks.every((j: { enabled: boolean }) => !j.enabled)).toBe(true); + const backup = board.items.find((j: { id: string }) => j.id === "maintain.backup"); + expect(backup).toMatchObject({ available: false, enabled: false }); + expect((await api("PATCH", "/schedules/maintain.backup", { enabled: true })).status).toBe(409); + + // 上一个用例写入的巡检报告算未关闭事项,先关掉 + await sql`update decision set status = 'dismissed'`.execute(h.db); + await api("PATCH", "/schedules/check.pending", { enabled: true }); + await sql`update scheduled_job set next_run_at = now() - interval '1 minute' where id = 'check.pending'`.execute( + h.db, + ); + const before = enqueued.length; + await sched.tick(); + await sched.idle(); + const row = await job("check.pending"); + expect(row.last_status).toBe("skipped"); + expect(enqueued.length).toBe(before); + await api("PATCH", "/schedules/check.pending", { enabled: false }); + }); + + it("端点只接受控制台会话", async () => { + const res = await bearer(h.app, DEV_KEY)("GET", "/schedules"); + expect(res.status).toBe(403); + }); +});