OpenClaw Cron 系统设计：实现 AI Agent 自主定时任务

比如你想让任务每小时整点执行，而不是从现在开始每小时执行，就可以设置一个整点时间戳作为锚点。计算公式是：

nextRun = anchor + Math.ceil((now - anchor) / everyMs) * everyMs 

cron：表达式调度

{
  kind: "cron",
  expr: "0 9 * * 1-5",
  tz: "Asia/Shanghai"
}

标准 cron 格式，还支持时区设置。上面这个表达式的意思是：工作日每天早 9 点（上海时间）。

cron 表达式的格式：秒 分 时 日 月 周

OpenClaw 使用 croniter 库解析表达式。这个库在 Python 生态里很成熟，处理各种边界情况（比如闰年、夏令时）都很稳定。

两种执行模式：Main vs Isolated

这是 Cron 系统最有趣的设计。同样是定时任务，执行方式完全不同：

Main Session：注入主会话

{
  sessionTarget: "main",
  payload: {
    kind: "systemEvent",
    text: "每日提醒：检查邮件"
  },
  wakeMode: "now"
}

任务不是执行，而是注入。系统把消息塞进主会话的消息队列，就像有人在对话框里发了一条系统消息。

这种模式适合：

• 简单提醒（不需要 Agent 做复杂操作）
• 需要用户看到并响应的任务
• 依赖现有上下文的任务

wakeMode 参数控制是否立即触发 Agent 心跳：

• now：立刻触发，Agent 马上处理这条消息
• next-heartbeat：等下次自然心跳时再处理

Isolated Session：独立会话执行

{
  sessionTarget: "isolated",
  payload: {
    kind: "agentTurn",
    message: "检查服务器健康状态并生成报告",
    model: "claude-3-5-sonnet",
    timeoutSeconds: 300,
    deliver: true,
    channel: "telegram",
    to: "@user123"
  }
}

这才是真正的后台执行。系统会启动一个全新的 Agent 会话，专门执行这个任务。执行完后，把结果汇报回主会话。

这种模式适合：

• 复杂任务（需要多轮思考和工具调用）
• 耗时任务（用户不想等）
• 不需要用户介入的任务

关键参数解释：

定时器核心逻辑

OpenClaw 的定时器实现有几个精巧的细节：

单一定时器模式

const MAX_TIMEOUT_MS = 2**31 - 1; // JS setTimeout 最大值
export function armTimer(state: CronServiceState) {
  // 1. 清除旧定时器
  if (state.timer) clearTimeout(state.timer);
  state.timer = null;

  // 2. 找到最近的待执行时间
  const nextAt = nextWakeAtMs(state);
  if (!nextAt) return;

  // 3. 设置新定时器（注意延迟上限）
  const delay = Math.max(nextAt - state.deps.nowMs(), 0);
  const clampedDelay = Math.min(delay, MAX_TIMEOUT_MS);
  state.timer = setTimeout(() => {
    void onTimer(state);
  }, clampedDelay);
  state.timer.unref?.(); // 允许进程在定时器未触发时退出
}

几个要点：

1. 单一定时器：永远只有一个活跃的定时器，指向最近的任务。这避免了定时器泛滥。
2. 延迟上限处理：JavaScript 的 setTimeout 最大只支持约 24.8 天（2^31-1 毫秒）。如果任务在更远的未来，先设一个最大延迟，到时候再重新计算。
3. unref 调用：timer.unref() 让这个定时器不阻止 Node.js 进程退出。如果用户关闭了应用，不会因为还有待执行的定时任务而卡住。

并发控制

export async function onTimer(state: CronServiceState) {
  if (state.running) return; // 防止并发执行
  state.running = true;
  try {
    await locked(state, async () => {
      await ensureLoaded(state);
      await runDueJobs(state);
      await persist(state);
      armTimer(state);
    });
  } finally {
    state.running = false;
  }
}

用一个简单的 running 标志位防止重入。如果定时器触发时上一次执行还没结束，直接跳过。

任务执行流程

让我们跟踪一个完整的执行流程：

阶段 1：Timer 触发

定时器到期，onTimer() 被调用。

阶段 2：筛选到期任务

const due = jobs.filter(j => 
  j.enabled && 
  j.state.runningAtMs === null && // 没有在执行中
  j.state.nextRunAtMs !== null && 
  nowMs >= j.state.nextRunAtMs // 已经到期
);

注意 runningAtMs 检查——如果一个任务正在执行（比如上次还没跑完），不会重复触发。

阶段 3：执行任务

根据 sessionTarget 分叉：

Main 路径：

// 注入系统事件
state.deps.enqueueSystemEvent(text, { agentId: job.agentId });
// 如果 wakeMode 是 now，立即触发心跳
if (job.wakeMode === "now") {
  const result = await state.deps.runHeartbeatOnce({ reason: `cron:${job.id}` });
}

Isolated 路径：

// 启动独立 Agent 会话
const res = await state.deps.runIsolatedAgentJob({ job, message: job.payload.message });
// 把结果汇报到主会话
state.deps.enqueueSystemEvent(`${prefix}: ${res.summary}`, { agentId: job.agentId });

阶段 4：更新状态

job.state.lastRunAtMs = startedAt;
job.state.lastStatus = status; // ok | error | skipped
job.state.lastDurationMs = endedAt - startedAt; // 计算下次执行时间
if (job.schedule.kind === "at" && status === "ok") {
  job.enabled = false; // 一次性任务完成后禁用
} else if (job.enabled) {
  job.state.nextRunAtMs = computeNextRunAtMs(job.schedule, nowMs);
}

阶段 5：持久化 + 重新调度

保存任务状态到存储，然后 armTimer() 重新设置下一个定时器。

任务状态机

一个 Cron Job 的状态流转：

┌─────────────────────────────────────────────────────────────┐
│                                                             │
│  ┌──────────┐ add() ┌──────────┐                          │
│  │ 创建     │ ─────>│ enabled  │ <────┐                   │
│  └──────────┘       └────┬─────┘      │                   │
│                         │             │                   │
│                         │ 到期触发    │                   │
│                         │             │                   │
│                         v             │                   │
│                  ┌──────────┐         │                   │
│                  │ running  │         │                   │
│                  └────┬─────┘         │                   │
│                       │               │                   │
│          ┌────────────┼───────────────┐                   │
│          │            │               │                   │
│          v            v               v                   │
│   ┌──────────┐ ┌──────────┐ ┌──────────┐                 │
│   │ ok       │ │ error    │ │ skipped  │                 │
│   └────┬─────┘ └────┬─────┘ └────┬─────┘                 │
│        │           │           │                         │
│        └───────────┴───────────┘                         │
│              │                                           │
│      ┌───────┴───────┐                                   │
│      │               │                                   │
│      v               v                                   │
│ ┌──────────┐   ┌──────────┐                             │
│ │ disabled │   │ 等待下次执行 │                           │
│ └──────────┘   └──────────┘                             │
│                                                             │
└─────────────────────────────────────────────────────────────┘

关键状态说明：

设计要点总结

1. 单一定时器 vs 多定时器

OpenClaw 选择了单一定时器模式。为什么？

多定时器的问题：

• 内存占用：每个任务一个定时器，1000 个任务就是 1000 个定时器
• 精度问题：大量定时器可能导致事件循环拥堵
• 难以管理：取消、更新操作复杂

单一定时器的优势：

• 内存恒定：永远只有一个活跃定时器
• 逻辑清晰：所有调度逻辑集中在 armTimer()
• 易于调试：只需关注一个定时器的行为

2. Main vs Isolated 的权衡

什么时候用 Main？

• 只是提醒用户做某事
• 需要用户确认或响应
• 任务依赖当前对话上下文

什么时候用 Isolated？

• 任务可能耗时较长
• 不需要用户介入
• 需要干净的执行环境

3. 结果汇报机制

Isolated 任务执行完后，通过 enqueueSystemEvent() 把结果注入主会话。用户会看到类似：

Cron: 服务器健康检查完成，所有服务正常运行

这个 Cron: 前缀是可配置的（postToMainPrefix 字段）。

Python 复现建议

如果你想用 Python 实现类似的 Cron 系统，核心依赖是：

pip install croniter # cron 表达式解析

关键实现点：

1. 调度计算

from croniter import croniter
from zoneinfo import ZoneInfo

def compute_next_run(schedule, now_ms):
    if schedule.kind == "cron":
        tz = ZoneInfo(schedule.tz) if schedule.tz else timezone.utc
        base_time = datetime.fromtimestamp(now_ms / 1000, tz=tz)
        cron = croniter(schedule.expr, base_time)
        next_time = cron.get_next(datetime)
        return int(next_time.timestamp() * 1000)

2. 异步定时器

async def arm_timer(self):
    if self.timer_task:
        self.timer_task.cancel()
    next_at = self.next_wake_at_ms()
    if next_at is None:
        return
    delay = max(0, (next_at - self.now_ms()) / 1000)
    self.timer_task = asyncio.create_task(self.timer_tick(delay))

3. 依赖注入

class CronService:
    def __init__(
        self,
        on_system_event: Callable[[str], None],
        run_agent_turn: Callable[[Job, str], dict],
        run_heartbeat: Callable[[str], dict],
    ):
        self.on_system_event = on_system_event
        self.run_agent_turn = run_agent_turn
        self.run_heartbeat = run_heartbeat

把执行 Agent、触发心跳等操作作为依赖注入，让 Cron 模块可以独立测试。

思考

这套设计解决了什么问题？

传统的定时任务系统（比如 crontab、APScheduler）只管到点执行。但 Agent 场景下，执行本身是个复杂的过程——需要上下文、需要推理、需要调用工具、还需要汇报结果。

OpenClaw 的 Cron 系统把这些都考虑进去了：

• 上下文隔离：Isolated 模式避免污染主对话
• 结果回传：执行完自动汇报
• 灵活调度：三种调度类型覆盖常见场景

还有什么可以改进？

1. 任务依赖：当前任务之间是独立的。如果 A 任务失败了，B 任务是否还执行？没有依赖图的概念。
2. 重试机制：任务失败后没有自动重试。对于网络请求类任务，这可能是个问题。
3. 优先级调度：所有任务平等。如果同一时刻有多个任务到期，按什么顺序执行？
4. 分布式支持：单机单定时器的设计，在分布式场景下需要改造。

实际应用场景

这套 Cron 系统特别适合：

• 个人助理 Agent：每日提醒、定期汇总、自动检查
• 监控 Agent：定时巡检、异常告警
• 内容 Agent：定时抓取、自动发布

想象一下：你对 AI 说每天晚上 10 点帮我总结一下今天的邮件，它就真的每天 10 点自动干活，干完了发个总结给你。这才是真正有用的 AI 助手。