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 <[email protected]>

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<string, Uint8Array>();
   let base: Awaited<ReturnType<typeof seedBase>>;
+  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));

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<ReturnType<typeof setTimeout>>();
-  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<string>) => 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<unknown>, period: () => Promise<number>) => {
-        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<void> {
+/** 超阈值:按时间槽入队一次 cleanupDownloadCache;上一次清理仍在进行则跳过(返回 false) */
+async function enqueueThresholdCleanup(d: AoDeps): Promise<boolean> {
   const busy = await d.ctx.db
     .selectFrom("task")
     .select("id")
@@ -163,12 +210,13 @@ async function enqueueThresholdCleanup(d: AoDeps): Promise<void> {
     .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:<id>) */