# 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` 是：

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

保存成字典后大概是：

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

---

## 3. 三个核心容器

`scheduled_jobs`：

```python
scheduled_jobs: dict[str, CronJob]
```

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

`cron_queue`：

```python
cron_queue: list[CronJob]
```

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

`_last_fired`：

```python
_last_fired: dict[str, str]
```

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

---

## 4. Cron 表达式

本节使用 5 字段 cron：

```text
分 时 月内日 月 周内日
```

例如：

```text
0 9 * * *
```

意思是每天 9:00。

支持：

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

---

## 5. `cron_scheduler_loop()`

这是一个后台守护线程。

流程：

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

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

---

## 6. `queue_processor_loop()`

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

为什么需要队列？

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

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

---

## 7. Agent Loop 中的变化

每轮开始先消费 cron 队列：

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

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

---

## 8. 本节课堂重点

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

核心结构是：

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