refactor(schedules): register pi.token_refresh and auth.cleanup as scheduled jobs

Pi token refresh (10 min) and auth session cleanup (1 h) no longer run
their own setInterval; both register in the maintain group of the
schedules registry with the same default intervals.

Co-Authored-By: Claude Opus 5.5 <[email protected]>

apps/server/src/modules/auth-settings/index.ts

@@ -1,4 +1,5 @@
 import { sql } from "@umt/db";
+import { registerScheduledJob } from "../schedules/registry.ts";
 import type { ServerContext, ServerModule } from "../types.ts";
 import { registerAuthRoutes } from "./auth-routes.ts";
 import { ensureSetupToken } from "./auth-service.ts";
@@ -18,7 +19,7 @@ export interface AuthSettingsModuleOptions {
   engineSchema?: string;
   /** api 角色启动时若未初始化管理员则生成并打印 setup token(默认 true;测试关掉自己调 ensureSetupToken) */
   setupTokenOnStart?: boolean;
-  /** 过期会话/登录尝试清理间隔,默认 1 小时 */
+  /** 定时任务 auth.cleanup(过期会话/登录尝试清理)的默认周期,默认 1 小时;<=0 不登记 */
   cleanupIntervalMs?: number;
 }
 
@@ -30,7 +31,6 @@ export async function cleanupAuthTables(ctx: Pick<ServerContext, "db">): Promise
 }
 
 export function createAuthSettingsModule(opts: AuthSettingsModuleOptions = {}): ServerModule {
-  let timer: ReturnType<typeof setInterval> | null = null;
   return {
     name: "auth-settings",
     registerRoutes(app, ctx) {
@@ -44,14 +44,23 @@ export function createAuthSettingsModule(opts: AuthSettingsModuleOptions = {}):
       }
     },
     registerTasks() {},
+    /** 清理周期登记为定时任务 auth.cleanup(维护组,由 schedules 调度器驱动) */
     start(ctx) {
-      const run = () => cleanupAuthTables(ctx).catch((err) => ctx.logger.warn({ err }, "认证表清理失败"));
-      timer = setInterval(run, opts.cleanupIntervalMs ?? 3600_000);
-      void run();
-    },
-    stop() {
-      if (timer) clearInterval(timer);
-      timer = null;
+      const everyMs = opts.cleanupIntervalMs ?? 3600_000;
+      if (everyMs <= 0) return;
+      registerScheduledJob(ctx, {
+        key: "auth.cleanup",
+        defaults: {
+          name: "登录会话清理",
+          description: "删除 7 天前的登录尝试和 30 天前过期或吊销的会话",
+          group: "maintain",
+          schedule: { type: "interval", everyMs },
+        },
+        run: async () => {
+          await cleanupAuthTables(ctx);
+          return { summary: "已清理" };
+        },
+      });
     },
   };
 }

apps/server/src/modules/pi/index.ts

@@ -2,6 +2,7 @@ import { randomUUID } from "node:crypto";
 import type { FakeOAuthControl } from "@umt/agent";
 import { defineTask } from "@umt/tasks";
 import { z } from "zod";
+import { registerScheduledJob } from "../schedules/registry.ts";
 import type { ServerContext, ServerModule } from "../types.ts";
 import { type OAuthOptions, OAuthSessions } from "./oauth.ts";
 import { refreshPiTokens } from "./refresh.ts";
@@ -11,14 +12,14 @@ export { type RunUpstream, resolveRunUpstream } from "./service.ts";
 
 /**
  * W7-A Pi 模型上游:pi_upstream CRUD、能力表、测试、模型表、服务端订阅登录(挂起式 interaction)、
- * 定时任务 refreshPiTokens(OAuth 过期前刷新,行锁内写回)。runAgent 经 ./service.ts 的 resolveRunUpstream 取上游。
+ * 持久任务 refreshPiTokens(周期登记为定时任务 pi.token_refresh;OAuth 过期前刷新,行锁内写回)。runAgent 经 ./service.ts 的 resolveRunUpstream 取上游。
  * 用法见 docs/runbook/agent.md「上游与登录」。
  */
 export interface PiModuleOptions {
   /** provider = fake 的模拟 OAuth(测试注入) */
   fakeOAuth?: FakeOAuthControl;
   oauth?: Omit<OAuthOptions, "client">;
-  /** 刷新任务入队间隔(默认 10 分钟;<=0 关闭) */
+  /** 定时任务 pi.token_refresh 的默认周期(默认 10 分钟;<=0 不登记) */
   refreshIntervalMs?: number;
   /** 剩余有效期少于它就刷新(默认 30 分钟) */
   refreshWindowMs?: number;
@@ -37,7 +38,6 @@ export function createPiModule(
     }
     return s;
   };
-  let timer: ReturnType<typeof setInterval> | null = null;
   const live: OAuthSessions[] = [];
   const interval = opts.refreshIntervalMs ?? 10 * 60_000;
   const windowMs = opts.refreshWindowMs ?? 30 * 60_000;
@@ -67,25 +67,28 @@ export function createPiModule(
         }),
       );
     },
+    /** 刷新周期登记为定时任务 pi.token_refresh(维护组,由 schedules 调度器驱动),执行体只入队持久任务 */
     start(ctx) {
       if (interval <= 0) return;
-      const enqueue = async () => {
-        try {
-          await ctx.engine.enqueue(
+      registerScheduledJob(ctx, {
+        key: "pi.token_refresh",
+        defaults: {
+          name: "上游登录刷新",
+          description: "订阅登录即将过期的 Agent 上游,提前刷新 token",
+          group: "maintain",
+          schedule: { type: "interval", everyMs: interval },
+        },
+        run: async (c) => {
+          const r = await ctx.engine.enqueue(
             "refreshPiTokens",
             {},
-            { idempotencyKey: `refreshPiTokens:${Math.floor(Date.now() / interval)}`, traceId: randomUUID() },
+            { idempotencyKey: `refreshPiTokens:${c.slot}`, traceId: randomUUID() },
           );
-        } catch (err) {
-          ctx.logger.warn({ err: { message: (err as Error).message } }, "refreshPiTokens 入队失败");
-        }
-      };
-      void enqueue();
-      timer = setInterval(() => void enqueue(), interval);
+          return { summary: r.deduplicated ? "本周期已入队" : "已入队刷新" };
+        },
+      });
     },
     async stop() {
-      if (timer) clearInterval(timer);
-      timer = null;
       await Promise.all(live.splice(0).map((s) => s.shutdown()));
     },
   };

apps/server/src/modules/schedules/index.ts

@@ -12,9 +12,6 @@ export { registerScheduledJob, scheduleRegistry } from "./registry.ts";
  * 各模块 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;

apps/server/test/schedules/module-jobs.test.ts

@@ -0,0 +1,51 @@
+import { describe, expect, it } from "vitest";
+import { createAuthSettingsModule } from "../../src/modules/auth-settings/index.ts";
+import { createPiModule } from "../../src/modules/pi/index.ts";
+import { scheduleRegistry } from "../../src/modules/schedules/registry.ts";
+import type { ServerContext } from "../../src/modules/types.ts";
+
+// W8-A:pi token 刷新与认证表清理不再自建计时器,改为登记到定时任务注册表(维护组、默认开启、周期不变)。
+
+describe("模块周期任务登记", () => {
+  it("pi.token_refresh:10 分钟,执行体按时间槽幂等入队 refreshPiTokens", async () => {
+    const enqueued: { name: string; key: string }[] = [];
+    const ctx = {
+      engine: {
+        enqueue: (name: string, _i: unknown, o: { idempotencyKey: string }) => {
+          enqueued.push({ name, key: o.idempotencyKey });
+          return Promise.resolve({ deduplicated: false });
+        },
+      },
+    } as unknown as ServerContext;
+    const mod = createPiModule();
+    await mod.start?.(ctx);
+    const def = scheduleRegistry(ctx).get("pi.token_refresh");
+    expect(def?.defaults).toMatchObject({ group: "maintain", schedule: { type: "interval", everyMs: 600_000 } });
+    expect(def?.defaults.enabled).not.toBe(false);
+    await def?.run({
+      jobId: "pi.token_refresh",
+      runId: "r",
+      trigger: "schedule",
+      now: new Date(),
+      config: {},
+      slot: "42",
+    });
+    expect(enqueued).toEqual([{ name: "refreshPiTokens", key: "refreshPiTokens:42" }]);
+    await mod.stop?.();
+  });
+
+  it("refreshIntervalMs<=0 不登记", async () => {
+    const ctx = {} as ServerContext;
+    await createPiModule({ refreshIntervalMs: 0 }).start?.(ctx);
+    expect(scheduleRegistry(ctx).get("pi.token_refresh")).toBeUndefined();
+  });
+
+  it("auth.cleanup:1 小时,维护组", async () => {
+    const ctx = {} as ServerContext;
+    await createAuthSettingsModule({ setupTokenOnStart: false }).start?.(ctx);
+    expect(scheduleRegistry(ctx).get("auth.cleanup")?.defaults).toMatchObject({
+      group: "maintain",
+      schedule: { type: "interval", everyMs: 3_600_000 },
+    });
+  });
+});

docs/runbook/schedules.md

@@ -28,6 +28,8 @@
 | `library.reconcile` | 资产对账 | 整理 | 6 小时 | 入队 `reconcileLibrary`，幂等键是 `auto:<时间槽>` |
 | `cache.threshold` | 缓存阈值检查 | 维护 | 同 qB 同步 | 超过硬阈值时入队 `cleanupDownloadCache` |
 | `trash.purge` | 回收站清理 | 维护 | 24 小时 | 入队 `purgeTrash` |
+| `pi.token_refresh` | 上游登录刷新 | 维护 | 10 分钟 | 入队 `refreshPiTokens`,幂等键 `refreshPiTokens:<时间槽>`;对 30 分钟内过期的订阅登录上游刷新 token |
+| `auth.cleanup` | 登录会话清理 | 维护 | 1 小时 | 直接执行 `cleanupAuthTables`:删 7 天前的登录尝试、30 天前过期或吊销的会话 |
 | `maintain.backup` | 备份 | 维护 | 每天 03:30(占位) | 未配置。备份目前由宿主机上的 `scripts/backup.sh` 执行 |
 | `check.pending` | 待处理巡检 | Agent | 6 小时，默认关闭 | 只读。有未关闭事项时才会运行 |
 | `check.downloads` | 下载诊断 | Agent | 2 小时，默认关闭 | 默认只读。打开 `allowRecover` 后，在授权范围内重试或重新校验 |
@@ -37,8 +39,7 @@
 
 - 自动化总开关关闭，或处于只读对账模式时，编排类任务会记为 `skipped`，不执行。
 - 各项默认周期和改造前的计时器一致。测试时可以通过模块参数调短，比如 `syncIntervalMs`、`schedulerIntervalMs`。
-- auth-settings 的会话清理(每小时)目前还在模块里自建计时器，尚未登记。
-- Pi 的 token 刷新(`refreshPiTokens`)目前也还是自建计时器，待 W7-A 登记为 `pi.token_refresh`。
+- 所有周期任务都已登记，模块里不再有自建计时器(W8-A 把 `pi.token_refresh`、`auth.cleanup` 也迁了进来)。
 
 ## 调度语义