From f8f70d8e0993032f3eded0005a2aa28928b223a9 Mon Sep 17 00:00:00 2001 From: root Date: Mon, 28 Sep 2026 02:32:22 +0800 Subject: [PATCH] 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 --- packages/contracts/src/api.ts | 70 ++++++++ packages/contracts/src/schedules.ts | 167 ++++++++++++++++++ packages/contracts/test/db-sync.test.ts | 8 + packages/db/migrations/0028_scheduled_jobs.ts | 73 ++++++++ packages/db/src/types.ts | 55 ++++++ 5 files changed, 373 insertions(+) create mode 100644 packages/contracts/src/schedules.ts create mode 100644 packages/db/migrations/0028_scheduled_jobs.ts diff --git a/packages/contracts/src/api.ts b/packages/contracts/src/api.ts index 2f58bed5..154a405b 100644 --- a/packages/contracts/src/api.ts +++ b/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["body"] extends z.ZodType ? export type EndpointResponse = z.infer; export * from "./files.ts"; +export * from "./schedules.ts"; diff --git a/packages/contracts/src/schedules.ts b/packages/contracts/src/schedules.ts new file mode 100644 index 00000000..7728957c --- /dev/null +++ b/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; + +export const ScheduleKind = z.enum(["task", "agent_check"]); +export type ScheduleKind = z.infer; + +export const ScheduleRunStatus = z.enum(["ok", "failed", "skipped", "running"]); +export type ScheduleRunStatus = z.infer; + +export const ScheduleTrigger = z.enum(["schedule", "manual", "skip"]); +export type ScheduleTrigger = z.infer; + +/** 用户可设的间隔下限(内置默认值可以更短,如测试) */ +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; + +/** 用户提交的调度:间隔有上下限 */ +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; + +/** + * 巡检触发条件:到点先按业务表判断,不满足记 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; + +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; +export type AgentCheckConfigInput = z.input; + +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; + +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; + +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; + +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; + +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; + +export const ScheduleRunNowResult = z.object({ + job: ScheduledJob, + /** false:已在排队或运行中,本次请求被合并 */ + queued: z.boolean(), +}); +export type ScheduleRunNowResult = z.infer; + +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() }); diff --git a/packages/contracts/test/db-sync.test.ts b/packages/contracts/test/db-sync.test.ts index df42b090..c7a1ec66 100644 --- a/packages/contracts/test/db-sync.test.ts +++ b/packages/contracts/test/db-sync.test.ts @@ -100,6 +100,9 @@ const ROWS: Record 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 = { 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, diff --git a/packages/db/migrations/0028_scheduled_jobs.ts b/packages/db/migrations/0028_scheduled_jobs.ts new file mode 100644 index 00000000..2b52a95e --- /dev/null +++ b/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): Promise { + 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): Promise { + await execDdl(db, dropTables(["scheduled_job_run", "scheduled_job"])); +} diff --git a/packages/db/src/types.ts b/packages/db/src/types.ts index 3b0a44ec..a3133a0f 100644 --- a/packages/db/src/types.ts +++ b/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; + available: Generated; + enabled: boolean; + default_enabled: boolean; + schedule: Json; + default_schedule: Json; + customized: Generated; + config: Generated; + 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; + fail_count: Generated; + created_at: GeneratedTimestamptz; + updated_at: GeneratedTimestamptz; +} + +export interface ScheduledJobRunTable { + id: Generated; + 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; 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; export type ResolutionAttemptRow = Selectable; export type NewResolutionAttempt = Insertable; export type ResolutionAttemptUpdate = Updateable; +export type ScheduledJobRow = Selectable; +export type NewScheduledJob = Insertable; +export type ScheduledJobUpdate = Updateable; +export type ScheduledJobRunRow = Selectable; +export type NewScheduledJobRun = Insertable; +export type ScheduledJobRunUpdate = Updateable; export type SecretRow = Selectable; export type NewSecret = Insertable; export type SecretUpdate = Updateable;