coding-agent/s14_cron_scheduler/代码讲解.md

S14 Cron Scheduler 代码讲解

这一节只讲相对 S13 新增的内容：

让 Agent 可以被定时任务唤醒，而不是只能等用户输入。

---

1. 本节新增内容

• CronJob dataclass：定时任务数据模型。
• scheduled_jobs：内存中的定时任务表。
• cron_queue：已触发、待投递的任务队列。
• _cron_field_matches()：匹配单个 cron 字段。
• cron_matches()：判断某个时间是否命中 cron 表达式。
• validate_cron()：校验 cron 表达式是否合法。
• save_durable_jobs() / load_durable_jobs()：持久化定时任务。
• schedule_job() / cancel_job()：注册和取消任务。
• cron_scheduler_loop()：后台调度线程。
• queue_processor_loop()：队列处理器，自动唤醒 Agent。
• 3 个工具：schedule_cron、list_crons、cancel_cron。

---

2. CronJob 数据结构

CronJob 是：

@dataclass
class CronJob:
    id: str
    cron: str
    prompt: str
    recurring: bool
    durable: bool

保存成字典后大概是：

{
    "id": "cron_123456",
    "cron": "0 9 * * *",
    "prompt": "每天早上检查项目状态",
    "recurring": True,
    "durable": True
}

---

3. 三个核心容器

scheduled_jobs：

scheduled_jobs: dict[str, CronJob]

保存所有已注册的 cron 任务。

cron_queue：

cron_queue: list[CronJob]

保存已经触发、但还没交给 Agent 处理的任务。

_last_fired：

_last_fired: dict[str, str]

用来防止同一分钟内重复触发同一个任务。

---

4. Cron 表达式

本节使用 5 字段 cron：

分 时 月内日 月 周内日

例如：

0 9 * * *

意思是每天 9:00。

支持：

*      任意值
*/5    每 5 个单位
1,2,3  多个值
1-5    范围

---

5. cron_scheduler_loop()

这是一个后台守护线程。

流程：

每 1 秒醒来一次
拿当前时间 datetime.now()
遍历 scheduled_jobs
如果 cron_matches(job.cron, now):
    把 job 放进 cron_queue

它只负责“发现任务到点了”，不直接调用模型。

---

6. queue_processor_loop()

队列处理器负责把 cron_queue 里的任务交给 Agent。

为什么需要队列？

为了让调度线程和 Agent 执行解耦：

调度器只负责触发
队列负责暂存
Agent 空闲时再消费

---

7. Agent Loop 中的变化

每轮开始先消费 cron 队列：

fired = consume_cron_queue()
for job in fired:
    messages.append({
        "role": "user",
        "content": f"[Scheduled] {job.prompt}"
    })

也就是说，定时任务最终会变成一条用户消息。

---

8. 本节课堂重点

Cron Scheduler 让 Agent 从“被动响应用户”变成“能被时间触发”。

核心结构是：

后台调度线程 -> cron_queue -> 队列处理器 -> agent_loop