使用 cron 表达式或间隔时间，调度并自动化周期性的 Agent 任务

定时任务（Scheduled Tasks）

ProductReady 内置一个生产就绪的定时任务系统，用于自动化周期性的 AI Agent 工作流。非常适合：

⏰ 周期自动化（小时报表、每日同步）
🔄 健康检查与监控
📊 定时数据处理
🤖 自动创建 Agent 任务

生产就绪：使用 PostgreSQL FOR UPDATE SKIP LOCKED（行级锁）实现多实例安全，并支持可选 Redis 缓存！

包含内容

✅ 数据库表结构 - 定时任务（cron / interval / once）
✅ Cron 解析器 - 轻量 cron 表达式解析（无依赖）
✅ tRPC API - 完整 CRUD + 校验
✅ 仪表板 UI - 用户友好的定时任务创建表单
✅ Redis 缓存（可选）- 大规模部署性能优化
✅ K8s CronJob - 可直接部署的 Kubernetes 清单
✅ 多实例安全 - PostgreSQL 行级锁防重入

架构概览

┌─────────────────┐     ┌─────────────────┐     ┌─────────────────┐
│ Cron Trigger    │────▶│  /api/scheduler │────▶│  PostgreSQL     │
│ (every 1 min)   │     │     /tick       │     │  FOR UPDATE     │
└─────────────────┘     └─────────────────┘     │  SKIP LOCKED    │
                               │                └─────────────────┘
                               │
                        ┌──────▼──────┐
                        │   Redis     │
                        │   (cache)   │
                        └─────────────┘

关键设计决策

外部触发器：每分钟触发一次（K8s CronJob、GitHub Actions、VM cron 等）
PostgreSQL 锁：FOR UPDATE SKIP LOCKED 确保同一任务只被一个实例处理
Redis 缓存（可选）：减少高频场景下的数据库查询
幂等处理：tick 可安全重试

快速开始

启用 scheduled_tasks 数据表

执行数据库 push 以创建所需表：

pnpm db:push

会创建两张表：

scheduled_tasks - 任务配置与调度信息
scheduled_task_runs - 执行历史

创建你的第一个定时任务

使用 tRPC API 或仪表板 UI：

访问 /agent/schedule 并点击"新建"：

基本信息：输入调度名称和可选描述
调度类型：选择其中一种：
- 间隔：简单的周期执行（例如：每 60 分钟）
- Cron：使用 cron 表达式的高级调度（例如：工作日上午 9 点）
- 一次性：在特定日期/时间运行一次
任务模板：配置触发时要创建的 Agent 任务
高级选项：可选设置最大运行次数限制

UI 功能包括：

实时 cron 表达式验证
人类可读的调度描述
常见 cron 预设（每小时、每天、工作日等）
验证错误的可视化反馈

// Using tRPC
const task = await trpc.scheduledTasks.create.mutate({
  name: "Daily Report",
  description: "Generate daily summary report",
  taskTemplate: {
    title: "Daily Summary Report",
    prompt: "Generate a summary of today's activities",
    priority: "medium",
  },
  scheduleType: "cron",
  cronExpression: "0 9 * * *", // Every day at 9:00
  isEnabled: true,
});

部署 K8s CronJob

在集群中应用清单：

kubectl apply -f k8s/scheduler-cronjob.yaml

本地开发也可以手动触发：

curl -X POST http://localhost:3000/api/scheduler/tick \
  -H "Authorization: Bearer your-cron-secret"

本地开发：Auto-Tick（仅限 Dev 环境）

生产环境中，定时任务由外部 CronJob（K8s、GitHub Actions 等）调用 POST /api/scheduler/tick 来触发。但在本地开发时，通常没有 CronJob 在运行。

为此，ProductReady 内置了一个仅限开发环境的 auto-tick 机制，在 pnpm dev 启动时自动在后台处理到期任务。

仅限开发环境！ Auto-tick 在生产环境（NODE_ENV=production）下永远不会启用。生产部署必须使用外部触发器（K8s CronJob、GitHub Actions、服务器 cron 等）。

工作原理

文件 src/server/scheduler/auto-tick-dev.ts 通过 instrumentation.node.ts 在服务启动时加载：

检查 NODE_ENV — 如果是 production，直接退出（不执行）
启动 setInterval 循环（开发环境默认每 10 秒）
每次 tick 调用 processDueTasks() — 与 /api/scheduler/tick 端点使用的是同一个函数

pnpm dev  →  instrumentation.node.ts  →  auto-tick-dev.ts
                                              │
                                              ▼
                                     setInterval(10s)
                                              │
                                              ▼
                                     processDueTasks()

环境变量

变量	默认值	说明
`ENABLE_SCHEDULER_AUTO_TICK`	`true`（dev）/ `false`（prod）	显式启用或禁用 auto-tick
`SCHEDULER_TICK_INTERVAL_MS`	`10000`（dev）/ `60000`（prod）	tick 间隔毫秒数（最小 5000）

文件索引

文件	作用
`src/server/scheduler/auto-tick-dev.ts`	仅限开发环境的后台 tick 循环
`instrumentation.node.ts`	启动时导入并启动 auto-tick

文件命名为 auto-tick-dev.ts（而非 auto-tick.ts），以明确表示这是开发便利工具，不属于生产执行路径。

非 K8s 的部署方式

所有部署方式本质都一样：周期性调用 scheduler tick。

触发端点：POST /api/scheduler/tick（需要 Authorization: Bearer $CRON_SECRET）
健康端点：GET /api/scheduler/tick（仅返回状态）

Vercel Cron Jobs 可以用 vercel.json 配置：

{
  "$schema": "https://openapi.vercel.sh/vercel.json",
  "crons": [
    {
      "path": "/api/scheduler/tick",
      "schedule": "* * * * *"
    }
  ]
}

Vercel Cron Jobs 会对配置的 path 发起 HTTP GET 请求。在 ProductReady 中，GET /api/scheduler/tick 是健康检查，不会处理到期任务。

如果要在 Vercel 上真正跑定时任务：

使用一个能发送 POST + Authorization 头 的外部调度器（见 GitHub Actions 方案），或
另行实现一个 GET 触发的 cron endpoint，并配套合适的安全保护策略。

用 GitHub Actions 的 cron 去调用你的部署地址（对 Vercel / Fly.io / Render / 自建 VM 等都适用）：

# .github/workflows/scheduler-tick.yml
name: Scheduler Tick

on:
  schedule:
    - cron: "* * * * *" # every minute (UTC)
  workflow_dispatch: {}

jobs:
  tick:
    runs-on: ubuntu-latest
    steps:
      - name: Call scheduler tick
        run: |
          curl -X POST "${{ secrets.SCHEDULER_URL }}/api/scheduler/tick" \
            -H "Authorization: Bearer ${{ secrets.CRON_SECRET }}" \
            -H "Content-Type: application/json" \
            --fail --silent --show-error

需要设置仓库 Secrets：

SCHEDULER_URL（例如：https://your-app.vercel.app）
CRON_SECRET

在传统 VM（Ubuntu/Debian 等）上用 cron 调用 tick：

# Run every minute
* * * * * curl -X POST "https://your-app.example.com/api/scheduler/tick" \
  -H "Authorization: Bearer $CRON_SECRET" \
  -H "Content-Type: application/json" \
  --fail --silent --show-error

调度类型

ProductReady 支持三种调度类型：

标准 5 字段 cron 表达式：

{
  scheduleType: "cron",
  cronExpression: "0 9 * * 1-5", // Weekdays at 9:00
}

常见模式：

* * * * * - 每分钟
0 * * * * - 每小时整点
0 9 * * * - 每天 9:00
0 9 * * 1-5 - 工作日 9:00
0 0 1 * * - 每月 1 号 0:00
*/15 * * * * - 每 15 分钟

按分钟间隔执行：

{
  scheduleType: "interval",
  intervalMinutes: 30, // Every 30 minutes
}

适合无需固定时间点、只要求周期性的任务。

在某个时间点执行一次：

{
  scheduleType: "once",
  scheduledAt: new Date("2024-12-25T00:00:00Z"),
}

执行后会自动禁用。

数据库表结构

scheduled_tasks

// src/db/schema/scheduled-tasks.ts
export const scheduledTasks = pgTable("scheduled_tasks", {
  id: text("id").primaryKey().$defaultFn(() => generateId("sch")),
  name: varchar("name", { length: 255 }).notNull(),
  description: text("description"),

  // Task template (what to create when triggered)
  taskTemplate: jsonb("task_template").$type<TaskTemplate>().notNull(),

  // Schedule configuration
  scheduleType: varchar("schedule_type", { length: 20 }).notNull(), // cron | interval | once
  cronExpression: varchar("cron_expression", { length: 100 }),
  intervalMinutes: integer("interval_minutes"),
  scheduledAt: timestamp("scheduled_at"),
  timezone: varchar("timezone", { length: 50 }).default("UTC"),

  // Execution limits
  maxRuns: integer("max_runs"), // null = unlimited
  isEnabled: boolean("is_enabled").notNull().default(true),

  // Tracking
  nextRunAt: timestamp("next_run_at"),
  lastRunAt: timestamp("last_run_at"),
  runCount: integer("run_count").notNull().default(0),

  // Relations
  spaceId: text("space_id").notNull().references(() => spaces.id),
  createdBy: text("created_by").notNull().references(() => users.id),
  createdAt: timestamp("created_at").defaultNow().notNull(),
  updatedAt: timestamp("updated_at"),
});

TaskTemplate 接口

interface TaskTemplate {
  title: string;           // Required: Task title
  prompt?: string;         // Optional: AI prompt
  initialParts?: unknown[];// Optional: Initial context
  priority?: "low" | "medium" | "high" | "urgent";
}

tRPC API 参考

scheduledTasks.list

列出当前 space 的定时任务。

const result = await trpc.scheduledTasks.list.query({
  limit: 20,
  offset: 0,
  isEnabled: true, // Optional filter
});
// Returns: { tasks: ScheduledTask[], total: number }

scheduledTasks.create

创建一个新的定时任务。

const task = await trpc.scheduledTasks.create.mutate({
  name: "Weekly Digest",
  taskTemplate: { title: "Generate Weekly Digest" },
  scheduleType: "cron",
  cronExpression: "0 9 * * 1", // Every Monday 9:00
});

scheduledTasks.update

更新已有任务。

await trpc.scheduledTasks.update.mutate({
  id: "sch_abc123",
  name: "Updated Name",
  cronExpression: "0 10 * * 1", // Change to 10:00
});

scheduledTasks.enable / disable

启用/禁用：

await trpc.scheduledTasks.enable.mutate({ id: "sch_abc123" });
await trpc.scheduledTasks.disable.mutate({ id: "sch_abc123" });

scheduledTasks.validateCron

保存前校验 cron 表达式：

const result = await trpc.scheduledTasks.validateCron.query({
  expression: "0 9 * * *",
});
// Returns: { isValid: true, description: "Daily at 09:00" }

scheduledTasks.stats

获取处理器统计：

const stats = await trpc.scheduledTasks.stats.query();
// Returns: { totalTasks, enabledTasks, dueTasksCount, recentRuns }

处理器：它是如何工作的

调度处理器（src/lib/scheduler/processor.ts）是核心执行引擎：

1. 全局锁获取

const lockAcquired = await acquireLock("scheduler:tick", 55);
if (!lockAcquired) {
  console.log("Another tick is in progress, skipping");
  return;
}

用于防止多个实例同时执行 tick。

2. 使用行锁拉取到期任务

const dueTasks = await db.execute(sql`
  SELECT id, name, task_template, ...
  FROM scheduled_tasks
  WHERE is_enabled = true 
    AND next_run_at <= NOW()
    AND (max_runs IS NULL OR run_count < max_runs)
  ORDER BY next_run_at ASC
  LIMIT 10
  FOR UPDATE SKIP LOCKED
`);

FOR UPDATE - 锁定选中的行
SKIP LOCKED - 跳过其他实例已锁住的行

3. 逐个处理任务

对每个到期任务：

根据 taskTemplate 创建新的 agent_task
在 scheduled_task_runs 记录执行结果
计算并更新 nextRunAt
runCount + 1
如启用 Redis，则更新缓存

4. 自动禁用

满足以下条件会自动禁用：

scheduleType === 'once' 执行一次后
设置了 maxRuns 且 runCount >= maxRuns

K8s 部署

CronJob 清单

# k8s/scheduler-cronjob.yaml
apiVersion: batch/v1
kind: CronJob
metadata:
  name: productready-scheduler
spec:
  schedule: "* * * * *"  # Every minute
  concurrencyPolicy: Forbid
  successfulJobsHistoryLimit: 3
  failedJobsHistoryLimit: 3
  jobTemplate:
    spec:
      template:
        spec:
          containers:
          - name: scheduler
            image: curlimages/curl:latest
            command:
            - /bin/sh
            - -c
            - |
              curl -X POST "$INTERNAL_URL/api/scheduler/tick" \
                -H "Authorization: Bearer $CRON_SECRET" \
                -H "Content-Type: application/json" \
                --fail --silent --show-error
            env:
            - name: INTERNAL_URL
              valueFrom:
                configMapKeyRef:
                  name: productready-config
                  key: INTERNAL_URL
            - name: CRON_SECRET
              valueFrom:
                secretKeyRef:
                  name: productready-secrets
                  key: CRON_SECRET
          restartPolicy: OnFailure

环境变量

变量	必需	说明
`CRON_SECRET`	是	API 鉴权 Bearer token
`INTERNAL_URL`	是	内网服务地址（例：`http://productready:3000`）
`REDIS_URL`	否	Redis URL（用于缓存）

Redis 缓存（可选）

Redis 缓存用于提升大规模部署性能：

缓存 next run time（ZSET）以加速查询
分布式锁防止重复处理
可降级：即使无 Redis 也能工作，只是更慢

配置

# Set environment variable
REDIS_URL=redis://localhost:6379

缓存操作

// Add task to cache
await cacheTaskNextRun(taskId, nextRunAt);

// Remove from cache (when disabled/deleted)
await removeFromCache(taskId);

// Acquire distributed lock
const acquired = await acquireLock("scheduler:tick", 55);

// Release lock
await releaseLock("scheduler:tick");

Redis 是可选的。scheduler 仅用 PostgreSQL 也能正常运行，但高并发场景下 Redis 能显著提升性能。

测试

集成测试在 tests/scheduler.test.ts：

# Run scheduler tests
pnpm vitest run tests/scheduler.test.ts

# Run all tests
pnpm test

测试类别

Cron Parser 单元测试 - cron 表达式解析
Router 集成测试 - tRPC CRUD
Redis 缓存测试 - 缓存读写与锁

前置条件

PostgreSQL 运行中（make db）
Redis 运行中（make db）- 可选
数据库种子（pnpm db:seed）

监控

健康检查端点

curl http://localhost:3000/api/scheduler/tick \
  -H "Authorization: Bearer $CRON_SECRET"
# GET returns health status, POST triggers processing

Stats API

const stats = await trpc.scheduledTasks.stats.query();
// {
//   totalTasks: 10,
//   enabledTasks: 8,
//   dueTasksCount: 2,
//   recentRuns: { total: 50, succeeded: 48, failed: 2 }
// }

日志

关注以 [Scheduler] 开头的日志：

[Scheduler] Processing 3 due tasks
[Scheduler] Task sch_abc123 executed successfully
[Scheduler] Task sch_def456 failed: timeout
[Scheduler Redis] Connected

最佳实践

1. 使用有意义的名称

// ✅ Good
name: "Daily Sales Report - 9 AM"
name: "Hourly Health Check"

// ❌ Bad
name: "Task 1"
name: "Cron job"

2. 对一次性/有限次任务设置 maxRuns

{
  scheduleType: "interval",
  intervalMinutes: 5,
  maxRuns: 12, // Run 12 times then stop
}

3. 在 taskTemplate 里包含必要上下文

taskTemplate: {
  title: "Weekly Report - Week 42",
  prompt: "Generate sales report for the past week",
  initialParts: [
    { type: "context", data: { weekNumber: 42 } }
  ],
}

4. 监控执行历史

用 scheduledTasks.runs 查询执行历史：

const history = await trpc.scheduledTasks.runs.query({
  scheduledTaskId: "sch_abc123",
  limit: 20,
});

故障排除

任务没有运行

检查 isEnabled：SELECT is_enabled FROM scheduled_tasks WHERE id = ?
检查 nextRunAt：是否在过去？
检查 maxRuns：是否已到上限？
检查 CronJob 日志：kubectl logs -l app=productready-scheduler

重复执行

如果任务被多次执行：

确认 FOR UPDATE SKIP LOCKED 正常生效
检查 Redis 锁获取日志
确保只有一个 Cron 触发器在跑（例如：kubectl get cronjobs）

性能问题

启用 Redis 缓存
降低 processor 批处理大小
添加索引：CREATE INDEX idx_scheduled_tasks_next_run ON scheduled_tasks(next_run_at) WHERE is_enabled = true

仪表板 UI

创建定时任务

访问 /agent/schedule 来访问定时任务仪表板：

查看所有任务：查看所有定时任务的列表，包括状态、调度和运行历史
新建：点击"新建"打开创建对话框
启用/禁用：使用开关切换任务的开启/关闭
删除：通过下拉菜单删除任务

UI 功能

调度类型：

间隔：简单的周期执行（例如："每 60 分钟"）
Cron：使用 cron 表达式的高级调度
- 常见预设：每小时、每天、工作日上午 9 点等
- 自定义表达式，带实时验证
- 自动显示人类可读的描述
一次性：在特定日期/时间调度

表单验证：

实时 cron 表达式验证
必填字段指示
可视化错误反馈
智能提示（例如：间隔为 60 分钟时显示 "≈ 1 小时"）

任务配置：

任务标题和可选提示
优先级（低/中/高/紧急）
最大运行次数限制（可选）
调度名称和描述

UI 组件

UI 构建使用：

CreateScheduledTaskDialog - 主创建表单组件
ScheduledTasksView - 列表视图，带启用/禁用/删除操作
使用 trpc.scheduledTasks.validateCron 进行实时验证
Toast 通知用于成功/错误反馈

文件索引

文件	作用
`src/db/schema/scheduled-tasks.ts`	任务配置表
`src/db/schema/scheduled-task-runs.ts`	执行历史表
`src/lib/scheduler/cron-parser.ts`	cron 表达式解析
`src/lib/scheduler/redis-cache.ts`	Redis 缓存操作
`src/lib/scheduler/processor.ts`	核心处理器
`src/app/api/scheduler/tick/route.ts`	触发/健康端点
`src/server/scheduler/auto-tick-dev.ts`	仅限开发环境的后台 auto-tick（本地 `pnpm dev` 专用）
`src/server/routers/scheduled-tasks.ts`	tRPC router
`src/components/agent/create-scheduled-task-dialog.tsx`	创建表单 UI
`src/components/agent/scheduled-tasks-view.tsx`	列表视图 UI
`k8s/scheduler-cronjob.yaml`	Kubernetes 部署
`tests/scheduler.test.ts`	集成测试

定时任务（Scheduled Tasks）

On this page