feat(contracts,db): scheduled_job registry tables and schedules API contracts

- 0028_scheduled_jobs: scheduled_job (stable key, group, kind task|agent_check,
  enabled/schedule with defaults + customized flag, config, last_* / next_run_at /
  run_requested_at / running_since, counters) and scheduled_job_run history
- contracts/schedules.ts: ScheduleSpec (interval | 5-field cron), AgentCheckConfig
  (goal template, capability allowlist, budget, trigger condition, allowRecover,
  reportAsDecision), board / run views
- Api.schedules: list, create (agent_check only), update, remove, run-now, skip,
  runs; all admin + console session only

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

packages/contracts/src/api.ts

@@ -14,6 +14,7 @@ import * as ops from "./ops.ts";
 import * as organization from "./organization.ts";
 import * as pi from "./pi.ts";
 import * as policy from "./policy.ts";
+import * as schedules from "./schedules.ts";
 import { GlobalSettings, GlobalSettingsUpdate } from "./settings.ts";
 import * as sources from "./sources.ts";
 
@@ -1561,6 +1562,74 @@ export const Api = {
       response: z.object({ report: z.record(z.string(), z.unknown()) }),
     }),
   },
+
+  // ── 定时任务看板(W7-C):统一注册表 + Agent 巡检;全部只接受控制台会话 ──
+  schedules: {
+    list: ep({
+      method: "get",
+      path: "/schedules",
+      scope: "admin",
+      auth: "session",
+      summary: "定时任务看板:全部登记的周期任务与 Agent 巡检、统计、调度器是否在线",
+      response: schedules.ScheduleBoard,
+    }),
+    create: ep({
+      method: "post",
+      path: "/schedules",
+      scope: "admin",
+      auth: "session",
+      summary: "新建自定义 Agent 巡检(目标模板 + 能力白名单 + 预算 + 间隔)",
+      body: schedules.AgentCheckCreate,
+      response: schedules.ScheduledJob,
+    }),
+    update: ep({
+      method: "patch",
+      path: "/schedules/{id}",
+      scope: "admin",
+      auth: "session",
+      summary: "修改开关 / 间隔或 cron / 参数;reset 恢复默认开关与调度",
+      params: schedules.ScheduleIdParam,
+      body: schedules.ScheduleUpdate,
+      response: schedules.ScheduledJob,
+    }),
+    remove: ep({
+      method: "delete",
+      path: "/schedules/{id}",
+      scope: "admin",
+      auth: "session",
+      summary: "删除自定义 Agent 巡检(内置任务不可删)",
+      params: schedules.ScheduleIdParam,
+      response: OkResult,
+    }),
+    runNow: ep({
+      method: "post",
+      path: "/schedules/{id}/run-now",
+      scope: "admin",
+      auth: "session",
+      summary: "立即执行一次(不改下次运行时间;已在排队或运行中则合并)",
+      params: schedules.ScheduleIdParam,
+      response: schedules.ScheduleRunNowResult,
+    }),
+    skip: ep({
+      method: "post",
+      path: "/schedules/{id}/skip",
+      scope: "admin",
+      auth: "session",
+      summary: "跳过下一次运行(下次运行时间顺延一个周期)",
+      params: schedules.ScheduleIdParam,
+      response: schedules.ScheduledJob,
+    }),
+    runs: ep({
+      method: "get",
+      path: "/schedules/{id}/runs",
+      scope: "admin",
+      auth: "session",
+      summary: "最近运行历史(新→旧)",
+      params: schedules.ScheduleIdParam,
+      query: schedules.ScheduleRunsQuery,
+      response: z.object({ items: z.array(schedules.ScheduledJobRun) }),
+    }),
+  },
 } as const;
 
 /** 用参数填充 `{name}` 路径占位,返回带 API_PREFIX 的完整路径 */
@@ -1582,3 +1651,4 @@ export type EndpointBody<E extends EndpointDef> = E["body"] extends z.ZodType ?
 export type EndpointResponse<E extends EndpointDef> = z.infer<E["response"]>;
 
 export * from "./files.ts";
+export * from "./schedules.ts";

packages/contracts/src/schedules.ts

@@ -0,0 +1,167 @@
+import { z } from "zod";
+import { AgentBudget } from "./agent.ts";
+import { KnownCapabilityId } from "./capability.ts";
+import { Timestamp } from "./common.ts";
+
+// 定时任务统一注册表(W7-C,docs/runbook/schedules.md):
+// 各模块的周期逻辑登记为 scheduled_job,由 io worker 上唯一的调度器(PG 咨询锁)按 next_run_at 触发;
+// 看板 /console/schedules 展示与调整(开关、间隔/cron、立即执行、跳过本次、历史)。
+// kind=agent_check 是"Agent 巡检":到点按目标模板入队 runAgent(能力白名单 + 预算)。
+
+/** 看板分组:采集 / 下载 / 整理 / 维护 / Agent 巡检 */
+export const ScheduleGroup = z.enum(["collect", "download", "organize", "maintain", "agent"]);
+export type ScheduleGroup = z.infer<typeof ScheduleGroup>;
+
+export const ScheduleKind = z.enum(["task", "agent_check"]);
+export type ScheduleKind = z.infer<typeof ScheduleKind>;
+
+export const ScheduleRunStatus = z.enum(["ok", "failed", "skipped", "running"]);
+export type ScheduleRunStatus = z.infer<typeof ScheduleRunStatus>;
+
+export const ScheduleTrigger = z.enum(["schedule", "manual", "skip"]);
+export type ScheduleTrigger = z.infer<typeof ScheduleTrigger>;
+
+/** 用户可设的间隔下限(内置默认值可以更短,如测试) */
+export const MIN_INTERVAL_MS = 10_000;
+export const MAX_INTERVAL_MS = 30 * 24 * 3600_000;
+
+/** 5 段 cron(分 时 日 月 周),支持 * , - / */
+export const CronExpr = z
+  .string()
+  .trim()
+  .regex(/^(\S+\s+){4}\S+$/, "cron 需要 5 段:分 时 日 月 周");
+
+export const ScheduleSpec = z.discriminatedUnion("type", [
+  z.object({ type: z.literal("interval"), everyMs: z.number().int().positive() }),
+  z.object({ type: z.literal("cron"), cron: CronExpr }),
+]);
+export type ScheduleSpec = z.infer<typeof ScheduleSpec>;
+
+/** 用户提交的调度:间隔有上下限 */
+export const ScheduleSpecInput = z.discriminatedUnion("type", [
+  z.object({ type: z.literal("interval"), everyMs: z.number().int().min(MIN_INTERVAL_MS).max(MAX_INTERVAL_MS) }),
+  z.object({ type: z.literal("cron"), cron: CronExpr }),
+]);
+export type ScheduleSpecInput = z.infer<typeof ScheduleSpecInput>;
+
+/**
+ * 巡检触发条件:到点先按业务表判断,不满足记 skipped,不唤醒模型。
+ * always 无条件;open_decisions 有未关闭待处理;stalled_downloads 有无进展/失败的下载;
+ * subscriptions_active 有追踪中的订阅。
+ */
+export const AgentCheckTrigger = z.enum(["always", "open_decisions", "stalled_downloads", "subscriptions_active"]);
+export type AgentCheckTrigger = z.infer<typeof AgentCheckTrigger>;
+
+export const AgentCheckConfig = z.object({
+  /** 目标模板:{{now}} {{date}} {{openDecisions}} {{stalledDownloads}} {{activeSubscriptions}} 会被替换 */
+  goalTemplate: z.string().trim().min(4).max(4000),
+  /** 能力白名单(空 = 只读类全部) */
+  capabilityAllowlist: z.array(KnownCapabilityId).max(40).default([]),
+  budget: AgentBudget.default({}),
+  triggerCondition: AgentCheckTrigger.default("always"),
+  /** 允许授权内恢复(task.retry / download.recheck):仍需有效持续授权,否则降级只读 */
+  allowRecover: z.boolean().default(false),
+  /** 结果写成一条待处理(info)事项 */
+  reportAsDecision: z.boolean().default(false),
+});
+export type AgentCheckConfig = z.infer<typeof AgentCheckConfig>;
+export type AgentCheckConfigInput = z.input<typeof AgentCheckConfig>;
+
+export const ScheduledJob = z.object({
+  id: z.string().min(1),
+  name: z.string().min(1),
+  description: z.string().nullable(),
+  group: ScheduleGroup,
+  kind: ScheduleKind,
+  /** 内置(代码登记)还是用户新建的自定义巡检 */
+  builtin: z.boolean(),
+  /** false = 占位(如备份未配置),不会执行 */
+  available: z.boolean(),
+  enabled: z.boolean(),
+  schedule: ScheduleSpec,
+  defaultSchedule: ScheduleSpec,
+  /** 用户改过开关或调度(之后模块登记不再覆盖) */
+  customized: z.boolean(),
+  config: z.record(z.string(), z.unknown()),
+  lastRunAt: Timestamp.nullable(),
+  lastStatus: ScheduleRunStatus.nullable(),
+  lastDurationMs: z.number().int().nonnegative().nullable(),
+  lastError: z.string().nullable(),
+  nextRunAt: Timestamp.nullable(),
+  runRequestedAt: Timestamp.nullable(),
+  runningSince: Timestamp.nullable(),
+  runCount: z.number().int().nonnegative(),
+  failCount: z.number().int().nonnegative(),
+  createdAt: Timestamp,
+  updatedAt: Timestamp,
+});
+export type ScheduledJob = z.infer<typeof ScheduledJob>;
+
+export const ScheduledJobRun = z.object({
+  id: z.uuid(),
+  jobId: z.string(),
+  trigger: ScheduleTrigger,
+  status: ScheduleRunStatus,
+  startedAt: Timestamp,
+  finishedAt: Timestamp.nullable(),
+  durationMs: z.number().int().nonnegative().nullable(),
+  error: z.string().nullable(),
+  summary: z.string().nullable(),
+  agentRunId: z.uuid().nullable(),
+  /** 巡检入队的 agent_run 结束后回填其终态 */
+  agentStatus: z.string().nullable(),
+  createdAt: Timestamp,
+});
+export type ScheduledJobRun = z.infer<typeof ScheduledJobRun>;
+
+export const ScheduleBoard = z.object({
+  items: z.array(ScheduledJob),
+  stats: z.object({
+    total: z.number().int().nonnegative(),
+    enabled: z.number().int().nonnegative(),
+    failed: z.number().int().nonnegative(),
+    running: z.number().int().nonnegative(),
+  }),
+  /** 有没有调度器持有锁(io worker 在线) */
+  schedulerOnline: z.boolean(),
+});
+export type ScheduleBoard = z.infer<typeof ScheduleBoard>;
+
+export const ScheduleIdParam = z.object({ id: z.string().min(1).max(200) });
+
+export const ScheduleUpdate = z
+  .object({
+    enabled: z.boolean().optional(),
+    schedule: ScheduleSpecInput.optional(),
+    /** agent_check 为 AgentCheckConfig(整体替换);task 为任务参数(浅合并) */
+    config: z.record(z.string(), z.unknown()).optional(),
+    /** 恢复默认开关与调度 */
+    reset: z.boolean().optional(),
+    /** 只有自定义巡检可改名称与说明 */
+    name: z.string().trim().min(1).max(60).optional(),
+    description: z.string().trim().max(200).optional(),
+  })
+  .refine((v) => Object.keys(v).length > 0, "至少修改一项");
+export type ScheduleUpdate = z.infer<typeof ScheduleUpdate>;
+
+export const AgentCheckCreate = z.object({
+  name: z.string().trim().min(1).max(60),
+  description: z.string().trim().max(200).optional(),
+  enabled: z.boolean().default(false),
+  schedule: ScheduleSpecInput,
+  config: AgentCheckConfig,
+});
+export type AgentCheckCreate = z.input<typeof AgentCheckCreate>;
+
+export const ScheduleRunNowResult = z.object({
+  job: ScheduledJob,
+  /** false:已在排队或运行中,本次请求被合并 */
+  queued: z.boolean(),
+});
+export type ScheduleRunNowResult = z.infer<typeof ScheduleRunNowResult>;
+
+export const ScheduleRunsQuery = z.object({ limit: z.coerce.number().int().min(1).max(100).default(30) });
+
+/** SSE:调度器每次开始/结束一次运行、用户修改后发出 */
+export const SCHEDULE_CHANGED_EVENT = "schedule.changed";
+export const ScheduleChangedPayload = z.object({ jobId: z.string(), status: ScheduleRunStatus.nullable() });

packages/contracts/test/db-sync.test.ts

@@ -100,6 +100,9 @@ const ROWS: Record<string, { row: Obj; dbOnly?: string[]; apiOnly?: string[] }>
   SlotDeferral: { row: C.SlotDeferral },
   SlotUnit: { row: C.SlotUnit },
   SubscriptionHistory: { row: C.SubscriptionHistory },
+  // W7-C 定时任务(0028):board_group 在 API 里叫 group;default_enabled 只用于"恢复默认"
+  ScheduledJob: { row: C.ScheduledJob, dbOnly: ["boardGroup", "defaultEnabled"], apiOnly: ["group"] },
+  ScheduledJobRun: { row: C.ScheduledJobRun },
   // W7-A Pi 上游(0027):secret_ref 不出 API
   PiUpstream: { row: C.PiUpstreamRow, dbOnly: ["secretRef"] },
 };
@@ -194,6 +197,11 @@ const ENUMS: Record<string, z.ZodEnum> = {
   ReconcileRunMode: C.ReconcileMode,
   ReconcileRunStatus: C.ReconcileStatus,
   SecretPurpose: C.SecretPurpose,
+  ScheduledJobBoardGroup: C.ScheduleGroup,
+  ScheduledJobKind: C.ScheduleKind,
+  ScheduledJobLastStatus: C.ScheduleRunStatus,
+  ScheduledJobRunTrigger: C.ScheduleTrigger,
+  ScheduledJobRunStatus: C.ScheduleRunStatus,
   PiUpstreamAuthKind: C.PiAuthKind,
   PiUpstreamStatus: C.PiUpstreamStatus,
   PiUpstreamThinkingLevel: C.PiThinkingLevel,

packages/db/migrations/0028_scheduled_jobs.ts

@@ -0,0 +1,73 @@
+import type { Kysely } from "kysely";
+import { dropTables, execDdl, oneOf } from "../src/migration-helpers.ts";
+
+// W7-C 定时任务统一注册表(docs/runbook/schedules.md):
+// - scheduled_job:稳定 key 为主键(如 feeds.poll);模块启动时登记默认值(customized 后不覆盖 enabled/schedule);
+//   io worker 上持有咨询锁的调度器按 next_run_at / run_requested_at 触发,结果写回 last_* 与计数。
+//   kind=agent_check 为 Agent 巡检(config 存目标模板/能力白名单/预算/触发条件),builtin=false 为用户自定义。
+// - scheduled_job_run:运行历史,每个任务只保留最近若干条(调度器写入时裁剪);巡检入队的 agent_run 结束后
+//   调度器回填 agent_status 与 summary。
+
+const GROUPS = ["collect", "download", "organize", "maintain", "agent"] as const;
+const KINDS = ["task", "agent_check"] as const;
+const STATUSES = ["ok", "failed", "skipped", "running"] as const;
+const TRIGGERS = ["schedule", "manual", "skip"] as const;
+
+// biome-ignore lint/suspicious/noExplicitAny: 迁移不依赖应用表类型
+export async function up(db: Kysely<any>): Promise<void> {
+  await execDdl(
+    db,
+    `
+create table scheduled_job (
+  id text primary key,
+  name text not null,
+  description text,
+  board_group text not null ${oneOf("board_group", GROUPS)},
+  kind text not null ${oneOf("kind", KINDS)},
+  builtin boolean not null default true,
+  available boolean not null default true,
+  enabled boolean not null,
+  default_enabled boolean not null,
+  schedule jsonb not null,
+  default_schedule jsonb not null,
+  customized boolean not null default false,
+  config jsonb not null default '{}'::jsonb,
+  last_run_at timestamptz,
+  last_status text ${oneOf("last_status", STATUSES)},
+  last_duration_ms integer,
+  last_error text,
+  next_run_at timestamptz,
+  run_requested_at timestamptz,
+  running_since timestamptz,
+  run_count integer not null default 0,
+  fail_count integer not null default 0,
+  created_at timestamptz not null default now(),
+  updated_at timestamptz not null default now()
+);
+create trigger scheduled_job_touch_updated_at before update on scheduled_job
+  for each row execute function umt_touch_updated_at();
+create index scheduled_job_due_idx on scheduled_job (next_run_at) where enabled;
+
+create table scheduled_job_run (
+  id uuid primary key default gen_random_uuid(),
+  job_id text not null references scheduled_job (id) on delete cascade,
+  trigger text not null ${oneOf("trigger", TRIGGERS)},
+  status text not null ${oneOf("status", STATUSES)},
+  started_at timestamptz not null default now(),
+  finished_at timestamptz,
+  duration_ms integer,
+  error text,
+  summary text,
+  agent_run_id uuid references agent_run (id) on delete set null,
+  agent_status text,
+  created_at timestamptz not null default now()
+);
+create index scheduled_job_run_job_idx on scheduled_job_run (job_id, started_at desc);
+`,
+  );
+}
+
+// biome-ignore lint/suspicious/noExplicitAny: 同上
+export async function down(db: Kysely<any>): Promise<void> {
+  await execDdl(db, dropTables(["scheduled_job_run", "scheduled_job"]));
+}

packages/db/src/types.ts

@@ -202,6 +202,11 @@ export type ResolutionAttemptOutcome =
   | "manual"
   | "skipped_locked";
 export type ResolutionAttemptActorType = "user" | "agent" | "system" | "api_key";
+export type ScheduledJobBoardGroup = "collect" | "download" | "organize" | "maintain" | "agent";
+export type ScheduledJobKind = "task" | "agent_check";
+export type ScheduledJobLastStatus = "ok" | "failed" | "skipped" | "running";
+export type ScheduledJobRunTrigger = "schedule" | "manual" | "skip";
+export type ScheduledJobRunStatus = "ok" | "failed" | "skipped" | "running";
 export type SecretPurpose =
   | "site_cookie"
   | "site_passkey"
@@ -1097,6 +1102,48 @@ export interface ResolutionAttemptTable {
   updated_at: GeneratedTimestamptz;
 }
 
+export interface ScheduledJobTable {
+  id: string;
+  name: string;
+  description: string | null;
+  board_group: ScheduledJobBoardGroup;
+  kind: ScheduledJobKind;
+  builtin: Generated<boolean>;
+  available: Generated<boolean>;
+  enabled: boolean;
+  default_enabled: boolean;
+  schedule: Json;
+  default_schedule: Json;
+  customized: Generated<boolean>;
+  config: Generated<Json>;
+  last_run_at: Timestamptz | null;
+  last_status: ScheduledJobLastStatus | null;
+  last_duration_ms: number | null;
+  last_error: string | null;
+  next_run_at: Timestamptz | null;
+  run_requested_at: Timestamptz | null;
+  running_since: Timestamptz | null;
+  run_count: Generated<number>;
+  fail_count: Generated<number>;
+  created_at: GeneratedTimestamptz;
+  updated_at: GeneratedTimestamptz;
+}
+
+export interface ScheduledJobRunTable {
+  id: Generated<string>;
+  job_id: string;
+  trigger: ScheduledJobRunTrigger;
+  status: ScheduledJobRunStatus;
+  started_at: GeneratedTimestamptz;
+  finished_at: Timestamptz | null;
+  duration_ms: number | null;
+  error: string | null;
+  summary: string | null;
+  agent_run_id: string | null;
+  agent_status: string | null;
+  created_at: GeneratedTimestamptz;
+}
+
 export interface SecretTable {
   id: Generated<string>;
   name: string;
@@ -1427,6 +1474,8 @@ export interface DB {
   preference_tag: PreferenceTagTable;
   reconcile_run: ReconcileRunTable;
   resolution_attempt: ResolutionAttemptTable;
+  scheduled_job: ScheduledJobTable;
+  scheduled_job_run: ScheduledJobRunTable;
   secret: SecretTable;
   server_mapping: ServerMappingTable;
   settings: SettingsTable;
@@ -1589,6 +1638,12 @@ export type ReconcileRunUpdate = Updateable<ReconcileRunTable>;
 export type ResolutionAttemptRow = Selectable<ResolutionAttemptTable>;
 export type NewResolutionAttempt = Insertable<ResolutionAttemptTable>;
 export type ResolutionAttemptUpdate = Updateable<ResolutionAttemptTable>;
+export type ScheduledJobRow = Selectable<ScheduledJobTable>;
+export type NewScheduledJob = Insertable<ScheduledJobTable>;
+export type ScheduledJobUpdate = Updateable<ScheduledJobTable>;
+export type ScheduledJobRunRow = Selectable<ScheduledJobRunTable>;
+export type NewScheduledJobRun = Insertable<ScheduledJobRunTable>;
+export type ScheduledJobRunUpdate = Updateable<ScheduledJobRunTable>;
 export type SecretRow = Selectable<SecretTable>;
 export type NewSecret = Insertable<SecretTable>;
 export type SecretUpdate = Updateable<SecretTable>;