Learn-Claude-Code | 笔记 | Concurrency | s14_new Cron Scheduler
目录
写在前面
learn-claude-code 项目目前有两条教程线，一条是之前的 s12 章节的，另一条是最近更新的 s20 章节的，大家如果刚学习这个项目的话推荐直接看最新的 s20 章节的教程即可，由于博主之前学习过 s12 章节的内容，因此打算把 s20 章节中新增章节的内容给补充学习，内容重复的章节博主这边就跳过了。
下面是旧版到新版的对应关系：
| Legacy 12-lesson track | Current 20-lesson track | Topic |
|---|---|---|
| old s01 | new s01 | Agent Loop |
| old s02 | new s02 | Tool Use |
| old s03 | new s05 | TodoWrite |
| old s04 | new s06 | Subagent |
| old s05 | new s07 | Skill Loading |
| old s06 | new s08 | Context Compact |
| old s07 | new s12 | Task System |
| old s08 | new s13 | Background Tasks |
| old s09 | new s15 | Agent Teams |
| old s10 | new s16 | Team Protocols |
| old s11 | new s17 | Autonomous Agents |
| old s12 | new s18 | Worktree Isolation |
| new only | s03, s04, s09, s10, s11, s14, s19, s20 | Permission, Hooks, Memory, System Prompt, Error Recovery, Cron, MCP, Comprehensive Agent |
从上表中我们可以看出我们需要补充的内容包括 s03（已补充）、s04（已补充）、s09（已补充）、s10（已补充）、s11（已补充）、s14、s19 以及 s20 八个章节的内容。
新版学习路径如下：
主线：能动手 → 能做复杂任务 → 能记住和恢复 → 能长期运行 → 能协作 → 能扩展并合体
前言
在上篇文章 Learn-Claude-Code | 笔记 | Planning & Coordination | s11_new Error Recovery 中，我们介绍了开源项目 learn-claude-code 新版第十一个章节 s11_new: Error Recovery 的内容，这篇文章我们继续跟着教程文档来学习规划与拆解相关内容，记录下个人学习笔记，和大家一起分享交流😄
Note：本篇文章主要学习记录 新版教程 第四部分 Concurrency 中 s14: Cron Scheduler 章节的内容。
github：https://github.com/shareAI-lab/learn-claude-code
reference：https://chatgpt.com/
1. s14: Cron Scheduler
在前面的章节中，我们已经逐步把一个最小 Agent Loop 扩展成了越来越完整的工程系统。s01 让模型进入循环；s02 让模型可以调用工具；s03 引入权限控制；s04 用 Hooks 把外部规则注入运行过程；s08 解决上下文压缩；s09 加入长期记忆；s10 把 system prompt 变成运行时组装的产物；s11 增加错误恢复能力；s12 让任务可以被创建、领取和完成；s13 则进一步让一些耗时操作可以在后台线程中执行。
但是到 s13 为止，Agent 的所有行为仍然有一个共同前提：必须先有人发起一轮对话。也就是说，哪怕 Agent 已经可以后台执行任务，也仍然是 “用户说一句，它动一下”。如果我们希望 Agent 能够每天早上 9 点自动检查 PR、每隔 30 分钟自动查看 CI 状态、每周自动生成一次报告，那么仅靠用户输入就不够了。此时需要的是一种新的触发来源：时间。
这就是 s14 Cron Scheduler 要解决的问题。它并不是重新设计 Agent Loop，而是在已有 Agent Loop 的外侧增加一个独立的调度层：当时间匹配某个 cron 表达式时，调度器把对应任务放入队列；当 Agent 空闲时，队列处理器再把这个任务交给正常的 agent_loop 执行。这样一来，Agent 的行为不再只能由用户实时输入触发，也可以由预先注册的时间表触发。
2. 问题
s13 的后台任务解决的是 “一个操作太慢怎么办”。比如模型让工具执行 build、test、install 这类耗时命令，如果一直阻塞主循环，用户体验就会很差。因此 s13 把慢任务丢到后台线程里执行，然后通过通知机制把结果带回主循环。
但 s13 仍然没有解决另一个问题：任务什么时候开始？
在 s13 中，任务的开始依赖用户或模型当前这一轮的工具调用。也就是说，后台能力只是把 “已经被触发的任务” 放到后台执行，而不是主动在未来某个时间点触发任务。对于 “每天早上跑一次测试”、“每 5 分钟检查一次状态”、“工作日 9 点提醒我处理 PR” 这类需求，Agent 需要一个独立于用户输入的时间驱动机制。
所以 s14 的核心不是 “让工具跑得更快”，而是 “让 Agent 可以按时间表生产新的工作”。这也是它和 s13 最关键的区别：s13 关注执行方式，s14 关注触发方式。
3. 解决方案
先看这一节的总览图：
从图中可以看到，s14 并没有推翻前面的主流程。原来的主线依然是：messages → prompt + cache → LLM try/except → TOOL_DISPATCH → tool_results → messages。
也就是说，模型调用、工具分发、工具结果回填、下一轮循环，这些东西都还在。s14 新增的是左下角那条独立的调度链路：cron_scheduler_loop → cron_queue → consume cron_queue → agent_loop。
这条链路的关键点在于：调度线程和 Agent Loop 是分开的。调度线程只负责看时间是否到了；到了以后，它并不直接调用模型，也不直接执行工具，而是把任务放入 cron_queue。真正消费这个任务的，仍然是原来的 agent_loop。
这就形成了一个很清晰的工程分层：调度器只生产 “待执行任务”，队列只负责缓冲任务，队列处理器只负责在 Agent 空闲时唤醒执行，Agent Loop 则仍然按照原来的方式处理用户消息、模型响应和工具调用。这样做的好处是，s14 不需要破坏已有 Agent Loop，只是在外部增加一个 “定时注入点”。
s14 的实现可以拆成四层来理解：
第一层是 Scheduler。它是一个独立的 daemon 线程，每秒钟醒来一次，遍历当前注册的 cron 任务，判断当前时间是否匹配任务的 cron 表达式。如果匹配，就认为这个任务 “到点了”。
第二层是 Queue。调度器不会直接执行任务，而是把到点的任务追加到 cron_queue 中。这个队列是调度线程和 Agent Loop 之间的缓冲区，它让 “时间触发” 和 “实际执行” 解耦。
第三层是 Queue Processor。它会周期性检查 cron_queue 是否有任务。如果队列非空，并且当前 Agent 没有在执行其他任务，它就会拿到 agent_lock，然后启动一轮 agent_loop。
第四层是 Consumer。真正进入 agent_loop 后，代码会调用 consume_cron_queue() 把已触发的 cron 任务取出来，然后把它们转换成普通的 user message，例如：
[Scheduled] review open PR every weekday
从这个角度看，cron 任务最后并不是一种特殊执行流，而是被包装成了一条普通消息，交给已有 Agent Loop 继续处理。这一点非常重要，因为它意味着定时任务并不需要一套新的模型调用逻辑，也不需要一套新的工具调用逻辑。它只是换了一个消息来源。
4. Cron Scheduler 流程图分析
Web 教程中的 6 张 Cron Scheduler 示例图，用非常直观的方式展示了一个定时任务从创建到触发再到执行完成的过程。
第一张图中，用户还只是提出一个普通需求：希望 “review open PR every weekday”。此时 Schedule book 里还没有真正保存任务，Due queue 也是空的，Agent inbox 只是显示 agent loop 可用。这对应的是定时任务创建前的状态：用户把一个普通 prompt 转换成了一个可重复执行的意图。
第二张图中，Schedule book 里出现了 0 9 * * 1-5，说明任务已经被转换成 cron 表达式并保存下来。这里强调的是 “任务卡片” 的持久化：任务不再只是当前对话里的一句话，而是进入了调度系统，可以在未来继续触发。
第三张图中，时间走到了 09:00，Due queue 的 watcher 进入 running 状态。这表示调度线程正在观察时间变化。Agent 此时不需要用户继续输入，调度线程自己会判断当前时间是否命中 cron 表达式。
第四张图中，Due queue 中出现了 due copy。这说明 cron 表达式匹配成功后，调度器并没有直接执行任务，而是把任务复制了一份放入待执行队列。这个 “copy” 很重要，因为原始 schedule card 还要保留，以便下次时间匹配时继续触发。
第五张图中，Agent inbox 收到了一个 agent turn。这说明 queue processor 已经把 due copy 交给 Agent Loop。此时定时任务终于进入了熟悉的主循环：它会像普通用户请求一样经过 system prompt、LLM、tool dispatch 和 tool result。
第六张图中，Agent turn 已经完成，结果被记录下来，Due queue 被清空，而 Schedule book 中的原始任务仍然保留。这说明一次触发结束后，调度系统不会删除 recurring 任务，而是继续等待下一次匹配。
这 6 张图连起来看，其实就是 s14 的完整执行链路：用户提出周期任务 → 生成 cron 表达式 → 保存到 schedule book → 调度线程观察时间 → 时间匹配后写入 due queue → queue processor 交给 agent_loop → agent_loop 正常执行 → 结果回写，周期任务继续保留。
完整动画演示如下图所示：
5. 工作原理（代码分析）
接下来我们就来看看具体代码是如何实现的：
1. CronJob：把 “未来要做的事” 结构化保存下来
s14 首先定义了一个人 CronJob 数据结构，用来描述一个可调度任务。
DURABLE_PATH = WORKDIR / ".scheduled_tasks.json"
@dataclass
class CronJob:
id: str
cron: str # "0 9 * * *"
prompt: str # message to inject when fired
recurring: bool # True = recurring, False = one-shot
durable: bool # True = persist to disk
scheduled_jobs: dict[str, CronJob] = {}
cron_queue: list[CronJob] = []
cron_lock = threading.Lock()
agent_lock = threading.Lock()
_last_fired: dict[str, str] = {} # job_id → "YYYY-MM-DD HH:MM"
这里最重要的是 cron 和 prompt 两个字段。cron 决定任务什么时候触发，例如 0 9 * * 1-5 表示工作日早上 9 点；prompt 则是任务触发后要重新送进 Agent Loop 的用户请求。
scheduled_jobs 保存所有已注册的定时任务，cron_queue 保存已经到时间、但还没交给 Agent 执行的任务。这里特意引入了两个锁：cron_lock 用来保护调度任务和队列，agent_lock 用来保证同一时间只有一个 Agent Loop 在运行，避免用户输入和定时任务同时抢占同一个会话。
_last_fired 则是一个非常关键的小细节。因为调度线程是每秒检查一次，如果某个任务在 09:00 这一分钟内一直匹配 cron 表达式，就可能被重复触发几十次。因此代码用 YYYY-MM-DD HH:MM 作为分钟级标记，保证同一个任务在同一分钟只触发一次。
2. cron 表达式匹配：判断当前时间是否命中任务
接下来是 cron 表达式的解析与匹配。s14 使用的是标准的五段式 cron 表达式：
分钟 小时 日 月 星期
也就是总览图中提到的：
* * * * *
对应代码如下：
def _cron_field_matches(field: str, value: int) -> bool:
"""Match a single cron field against a value."""
if field == "*":
return True
if field.startswith("*/"):
step = int(field[2:])
return step > 0 and value % step == 0
if "," in field:
return any(_cron_field_matches(f.strip(), value)
for f in field.split(","))
if "-" in field:
lo, hi = field.split("-", 1)
return int(lo) <= value <= int(hi)
return value == int(field)
这一段负责判断单个字段是否匹配当前时间值。比如 * 表示任意值都匹配，*/5 表示每 5 个单位触发一次，1-5 表示一个范围，1,3,5 表示多个离散值。
在这个基础上，cron_matches() 会把五个字段组合起来，判断一个完整的 cron 表达式是否命中当前时间。
def cron_matches(cron_expr: str, dt: datetime) -> bool:
"""Check if a 5-field cron expression matches the given datetime.
Standard cron semantics: DOM and DOW use OR when both are constrained."""
fields = cron_expr.strip().split()
if len(fields) != 5:
return False
minute, hour, dom, month, dow = fields
dow_val = (dt.weekday() + 1) % 7 # Python Monday=0 → cron Sunday=0
m = _cron_field_matches(minute, dt.minute)
h = _cron_field_matches(hour, dt.hour)
dom_ok = _cron_field_matches(dom, dt.day)
month_ok = _cron_field_matches(month, dt.month)
dow_ok = _cron_field_matches(dow, dow_val)
if not (m and h and month_ok):
return False
dom_unconstrained = dom == "*"
dow_unconstrained = dow == "*"
if dom_unconstrained and dow_unconstrained:
return True
if dom_unconstrained:
return dow_ok
if dow_unconstrained:
return dom_ok
return dom_ok or dow_ok
这里还有一个容易忽略的点：Python 的 weekday() 中周一是 0，而 cron 语义中通常周日是 0。因此代码用：
dow_val = (dt.weekday() + 1) % 7
把 Python 的星期值转换成 cron 的星期值。
另外，day-of-month 和 day-of-week 在标准 cron 语义中，当两者都被约束时通常采用 OR 逻辑。代码里最后的：
return dom_ok or dow_ok
就是在模拟这个行为。
3. validate_cron：创建任务前先校验表达式
为了避免 LLM 生成错误的 cron 表达式，s14 还增加了校验逻辑。
def validate_cron(cron_expr: str) -> str | None:
"""Validate a cron expression. Returns error message or None."""
fields = cron_expr.strip().split()
if len(fields) != 5:
return f"Expected 5 fields, got {len(fields)}"
bounds = [(0, 59), (0, 23), (1, 31), (1, 12), (0, 6)]
names = ["minute", "hour", "day-of-month", "month", "day-of-week"]
for i, (field, (lo, hi), name) in enumerate(zip(fields, bounds, names)):
err = _validate_cron_field(field, lo, hi)
if err:
return f"{name}: {err}"
return None
这一步很重要，因为 cron 任务不是一次性的普通工具调用，而是会在未来自动触发。如果表达式本身就是错的，那么错误会被延迟到未来发生，排查成本会更高。因此 s14 在注册任务时就先检查字段数量、数值范围、步长、区间等内容。
4. 持久化：durable 任务写入 .scheduled_tasks.json
s14 的另一个新增点是定时任务可以持久化。只要 durable=True，任务就会保存到工作目录下的 .scheduled_tasks.json。
def save_durable_jobs():
"""Persist durable jobs to .scheduled_tasks.json."""
durable = [asdict(j) for j in scheduled_jobs.values() if j.durable]
DURABLE_PATH.write_text(json.dumps(durable, indent=2))
def load_durable_jobs():
"""Load durable jobs from disk on startup."""
if not DURABLE_PATH.exists():
return
try:
jobs = json.loads(DURABLE_PATH.read_text())
for j in jobs:
job = CronJob(**j)
err = validate_cron(job.cron)
if err:
print(f" \033[31m[cron] skipping invalid job {job.id}: {err}\033[0m")
continue
scheduled_jobs[job.id] = job
valid = [j for j in jobs if j["id"] in scheduled_jobs]
if valid:
print(f" \033[35m[cron] loaded {len(valid)} durable job(s)\033[0m")
except Exception:
pass
这里体现了两类任务的区别：一类是 durable=True 的持久化任务，进程重启后还能恢复；另一种是 durable=False 的 session-only 任务，只存在于当前进程内存中，程序结束后就消失。
这也解释了教程图里的那句话：这里不是操作系统级别的 crontab，而是 Agent Harness 自己维护的一套轻量级调度系统。所以进程关闭后，调度线程本身会停止；只有 durable 任务的定义会保存在磁盘上，等下次程序启动时再加载回来。
5. schedule_job / cancel_job：注册与取消定时任务
LLM 真正调用 schedule_cron 工具时，底层会走到 schedule_job()。
def schedule_job(cron: str, prompt: str, recurring: bool = True,
durable: bool = True) -> CronJob | str:
"""Register a new cron job. Returns CronJob or error string."""
err = validate_cron(cron)
if err:
return err
job = CronJob(
id=f"cron_{random.randint(0, 999999):06d}",
cron=cron, prompt=prompt,
recurring=recurring, durable=durable,
)
with cron_lock:
scheduled_jobs[job.id] = job
if durable:
save_durable_jobs()
print(f" \033[35m[cron register] {job.id} '{cron}' → {prompt[:40]}\033[0m")
return job
这段代码完成了三个动作：先校验 cron 表达式，再创建 CronJob，最后把它注册到 scheduled_jobs。如果任务是持久化任务，就顺手写入 .scheduled_tasks.json。
取消任务的逻辑则比较直接：
def cancel_job(job_id: str) -> str:
"""Cancel a cron job."""
with cron_lock:
job = scheduled_jobs.pop(job_id, None)
if not job:
return f"Job {job_id} not found"
if job.durable:
save_durable_jobs()
print(f" \033[31m[cron cancel] {job_id}\033[0m")
return f"Cancelled {job_id}"
这里同样要注意持久化文件的同步。如果取消的是 durable 任务，那么内存里的 scheduled_jobs 改了之后，磁盘上的 .scheduled_tasks.json 也要同步更新，否则下次重启又会把已经取消的任务加载回来。
6. cron_scheduler_loop：独立 daemon 线程负责 “到点触发”
s14 最核心的新增逻辑就是 cron_scheduler_loop()。
def cron_scheduler_loop():
"""Independent daemon thread: poll every 1s, fire matching jobs.
Individual job errors are caught to prevent one bad job from
killing the entire scheduler thread."""
while True:
time.sleep(1)
now = datetime.now()
minute_marker = now.strftime("%Y-%m-%d %H:%M")
with cron_lock:
for job in list(scheduled_jobs.values()):
try:
if cron_matches(job.cron, now):
if _last_fired.get(job.id) != minute_marker:
cron_queue.append(job)
_last_fired[job.id] = minute_marker
print(f" \033[35m[cron fire] {job.id} → "
f"{job.prompt[:40]}\033[0m")
if not job.recurring:
scheduled_jobs.pop(job.id, None)
if job.durable:
save_durable_jobs()
except Exception as e:
print(f" \033[31m[cron error] {job.id}: {e}\033[0m")
这一段可以看作 s14 的 “时钟”。它每秒醒来一次，拿当前时间和所有 scheduled_jobs 做匹配。如果某个任务命中，就把它追加到 cron_queue。
这里的设计非常清晰：调度线程不直接调用 LLM，也不直接执行工具，而只是把 “到期任务” 放入队列。这样做的好处是调度系统和 Agent Loop 解耦了。调度线程只负责发现 “该做什么”，真正 “怎么做” 仍然交给原来的 Agent Loop。
对于一次性任务，任务会在触发后删除它：
if not job.recurring:
scheduled_jobs.pop(job.id, None)
if job.durable:
save_durable_jobs()
所以 recurring 任务会一直保留，下一次命中时间继续触发；one-shot 任务触发一次就自动消失。
程序启动后，会先加载持久化任务，然后启动调度线程：
load_durable_jobs()
threading.Thread(target=cron_scheduler_loop, daemon=True).start()
print(" \033[35m[cron] scheduler thread started\033[0m")
这里的 daemon=True 表示这是一个后台守护线程。主进程还在，它就继续跑；主进程退出，它也会一起结束。
7. cron 工具：把调度能力暴露给 LLM
为了让模型能够使用这套调度能力，s14 在工具列表中新增了三个工具：schedule_cron、list_crons、cancel_cron。
def run_schedule_cron(cron: str, prompt: str,
recurring: bool = True, durable: bool = True) -> str:
result = schedule_job(cron, prompt, recurring, durable)
if isinstance(result, str):
return f"Error: {result}"
return f"Scheduled {result.id}: '{cron}' → {prompt}"
def run_list_crons() -> str:
with cron_lock:
jobs = list(scheduled_jobs.values())
if not jobs:
return "No cron jobs. Use schedule_cron to add one."
lines = []
for j in jobs:
tag = "recurring" if j.recurring else "one-shot"
dur = "durable" if j.durable else "session"
lines.append(f" {j.id}: '{j.cron}' → {j.prompt[:40]} "
f"[{tag}, {dur}]")
return "\n".join(lines)
def run_cancel_cron(job_id: str) -> str:
return cancel_job(job_id)
对应的工具定义如下：
{"name": "schedule_cron",
"description": "Schedule a cron job. cron is 5-field: min hour dom month dow.",
"input_schema": {"type": "object",
"properties": {
"cron": {"type": "string",
"description": "5-field cron expression"},
"prompt": {"type": "string",
"description": "Message to inject when fired"},
"recurring": {"type": "boolean",
"description": "True=recurring, False=one-shot"},
"durable": {"type": "boolean",
"description": "True=persist to disk"}},
"required": ["cron", "prompt"]}},
{"name": "list_crons",
"description": "List all registered cron jobs.",
"input_schema": {"type": "object", "properties": {},
"required": []}},
{"name": "cancel_cron",
"description": "Cancel a cron job by ID.",
"input_schema": {"type": "object",
"properties": {"job_id": {"type": "string"}},
"required": ["job_id"]}},
这一步把调度能力变成了 Agent 可以调用的工具。也就是说，用户可以用自然语言说：“每天早上 9 点提醒我检查 PR”，模型会把这个意图转换成一次 schedule_cron 工具调用，Harness 再把它注册成一个真正的定时任务。
从工程角度看，这一步非常关键。因为 LLM 本身并不会真的 “等待到明天早上 9 点”，它只是生成一个工具调用；真正的等待、触发和重新注入，全部由 Harness 负责。
8. execute_tool：统一分发普通工具、任务工具和 cron 工具
s14 也修改了工具分发逻辑，把 cron 相关工具接入到统一的 execute_tool() 中。
def execute_tool(block) -> str:
"""Execute a tool call block, return output."""
handler = {
"bash": run_bash, "read_file": run_read, "write_file": run_write,
"create_task": run_create_task, "list_tasks": run_list_tasks,
"get_task": run_get_task, "claim_task": run_claim_task,
"complete_task": run_complete_task,
"schedule_cron": run_schedule_cron, "list_crons": run_list_crons,
"cancel_cron": run_cancel_cron,
}.get(block.name)
if handler:
return handler(**block.input)
return f"Unknown tool: {block.name}"
这说明 s14 并没有破坏原来的工具调用路径。无论是 bash、read_file、write_file，还是 s12 的 task 工具、s14 的 cron 工具，最终都走同一个工具分发入口。
这也是教程一直强调的 Harness Engineering 思路：Agent 能力不是一次性堆在模型里，而是通过一层层可组合的外部机制接入进来。s14 的 cron scheduler，就是在已有工具系统上继续增加一种新的工具类型。
9. agent_loop：消费 cron_queue，把定时任务重新注入 messages
有了调度线程之后，还需要有人消费 cron_queue。这一步发生在 agent_loop() 的开头。
def agent_loop(messages: list, context: dict) -> dict:
system = get_system_prompt(context)
while True:
# Layer 4: consume fired cron jobs → inject as messages
fired = consume_cron_queue()
for job in fired:
messages.append({"role": "user",
"content": f"[Scheduled] {job.prompt}"})
print(f" \033[35m[inject cron] {job.prompt[:50]}\033[0m")
这里是 s14 与前面章节连接最紧密的地方。定时任务到点之后，并不是绕开 Agent Loop 直接执行，而是被转换成一条新的 user message：
{"role": "user", "content": f"[Scheduled] {job.prompt}"}
这意味着定时任务最终仍然是一次普通的 Agent 回合。它依然会经过 system prompt、LLM 推理、工具调用、tool_result 回填、下一轮循环这些路径。
这就是图中 “consume cron_queue → messages → prompt + cache → LLM → tool dispatch” 的含义。Cron Scheduler 只是多了一个消息注入入口，并没有改变 Agent Loop 的主干结构。
10. queue_processor_loop：Agent 空闲时自动交付定时任务
如果只有 agent_loop() 开头消费队列，还不够。因为用户不一定刚好在任务触发后输入新问题。如果没有新的用户输入，Agent Loop 就不会被调用，cron_queue 里的任务就会一直堆着。
所以 s14 又加了一个 queue_processor_loop()，专门负责在 Agent 空闲时自动启动一轮 Agent Loop。
def queue_processor_loop():
"""Auto-deliver fired cron jobs when the agent is idle."""
global session_context
while True:
time.sleep(0.2)
if not has_cron_queue():
continue
if not agent_lock.acquire(blocking=False):
continue
try:
if not has_cron_queue():
continue
print("\n \033[35m[queue processor] delivering scheduled work\033[0m")
run_agent_turn_locked()
finally:
agent_lock.release()
这一段代码解决的是 “到点之后谁来唤醒 Agent” 的问题。它每 0.2 秒检查一次 cron_queue，如果发现有任务等待执行，并且当前 Agent 没有被用户输入占用，就获取 agent_lock，然后调用：
run_agent_turn_locked()
注意这里没有传入新的 user_query，因为定时任务已经在 cron_queue 中了。真正进入 agent_loop() 后，开头的 consume_cron_queue() 会把这些任务注入到 messages 里。
这就是 s14 相比普通工具调用最特殊的地方：普通工具调用必须由用户输入触发，而 cron 任务可以由后台线程触发，再由 queue processor 自动送进 Agent Loop。
11. run_agent_turn_locked：用户输入和定时任务共用同一条执行路径
s14 并没有给定时任务单独写一套执行流程，而是让用户输入和定时任务最终都走 run_agent_turn_locked()。
def run_agent_turn_locked(user_query: str | None = None):
"""Run one agent turn. Caller must hold agent_lock."""
global session_context
if user_query is not None:
session_history.append({"role": "user", "content": user_query})
session_context = agent_loop(session_history, session_context)
session_context = update_context(session_context, session_history)
print_latest_assistant_text(session_history)
print()
如果是用户主动输入，那么 user_query 不为空，代码会先把用户消息追加到 session_history。如果是定时任务触发，那么 user_query 是 None，它不会追加普通用户输入，而是进入 agent_loop() 后消费 cron_queue。
这就形成了统一结构：用户输入是一种消息来源，cron_queue 也是一种消息来源。只要进入 Agent Loop，后面的流程就完全一致。
12. main：启动 queue processor，并用 agent_lock 包住用户输入
最后，在程序入口处，s14 启动了 queue processor 线程，并且在处理用户输入时使用 agent_lock。
if __name__ == "__main__":
print("s14: cron scheduler")
print("Enter a question, press Enter to send. Type q to quit.\n")
threading.Thread(target=queue_processor_loop, daemon=True).start()
print(" \033[35m[queue processor] started\033[0m")
while True:
try:
query = input("\033[36ms14 >> \033[0m")
except (EOFError, KeyboardInterrupt):
break
if query.strip().lower() in ("q", "exit", ""):
break
with agent_lock:
run_agent_turn_locked(query)
这里的 agent_lock 非常重要。因为现在 Agent Loop 有两个入口：一个是用户在终端输入问题，另一个是 queue processor 自动交付定时任务。如果没有锁，就可能出现用户请求还没跑完，后台定时任务又启动一轮 Agent Loop，导致 session_history 和 session_context 被并发修改。
所以 s14 的并发模型可以概括为三条线：主线程负责接收用户输入，cron scheduler 线程负责到点触发任务，queue processor 线程负责在 Agent 空闲时把任务送进 Agent Loop。它们之间通过 cron_queue 和 agent_lock 协调。
博主在给定下面的提示词情况下：
Schedule a task to print the current date every 2 minutes
想通过调试看看整个过程发生了什么，我们来具体分析下：
Loop 1
第一次循环模型响应结果
第一次循环 schedule job 工具执行
第一次循环工具执行结果
从第一次循环可以看到，用户输入的是一个自然语言调度需求：
Schedule a task to print the current date every 2 minutes
模型并没有直接执行 date 命令，而是先把这个需求转换成一次 schedule_cron 工具调用：
ToolUseBlock(
name="schedule_cron",
input={
"cron": "*/2 * * * *",
"prompt": "Print the current date. Run: date",
"recurring": True
}
)
这里体现出 s14 的关键变化：用户说的是 “每 2 分钟打印一次当前日期”，但真正落到 Harness 层时，它被拆成了两个部分。第一部分是 cron 字段，用 */2 * * * * 表示每 2 分钟触发一次；第二部分是 prompt 字段，也就是未来触发时要重新注入 Agent Loop 的任务内容。
随后 schedule_job() 会创建一个新的 CronJob 对象：
CronJob(
id="cron_960034",
cron="*/2 * * * *",
prompt="Print the current date. Run: date",
recurring=True,
durable=True
)
并把它写入 scheduled_jobs：
scheduled_jobs[job.id] = job
调试中可以看到，任务被注册后返回了：
Scheduled cron_960034: */2 * * * * -> Print the current date. Run: date
这说明第一次循环完成的并不是 “执行定时任务”，而是 “创建定时任务”。换句话说，Loop 1 只是把一个未来要执行的动作登记进调度系统，真正的执行要等 cron_scheduler_loop 后续命中时间后再触发。
Loop 2
第二次循环消费已触发的定时任务
第二次循环模型响应结果
第二次循环已经不是用户手动输入触发的普通对话，而是后台队列触发的自动执行。从调试结果可以看到，cron_scheduler_loop 已经命中了 cron_960034，并将它放入 cron_queue。随后 queue_processor_loop 发现队列中有任务，于是自动调用 Agent Loop。
进入 agent_loop() 后，代码首先执行：
fired = consume_cron_queue()
for job in fired:
messages.append({
"role": "user",
"content": f"[Scheduled] {job.prompt}"
})
调试中可以看到：
fired = [
CronJob(
id="cron_960034",
cron="*/2 * * * *",
prompt="Print the current date. Run: date",
recurring=True,
durable=True
)
]
随后 messages[-1] 变成了：
{
"role": "user",
"content": "[Scheduled] Print the current date. Run: date"
}
这一步非常关键。它说明 Cron Scheduler 并不是绕过 Agent Loop 去执行命令，而是把 “到期任务” 重新包装成一条普通的 user message。只不过这条消息不是用户现场输入的，而是由调度线程在命中时间后注入进来的。
接着模型看到 [Scheduled] Print the current date. Run: date 后，开始正常推理，并生成 bash 工具调用：
ToolUseBlock(
name="bash",
input={"command": "date"}
)
所以 Loop 2 展示的是 s14 的完整自动触发链路：cron 命中 → cron_queue → consume_cron_queue() → 注入 [Scheduled] message → LLM 生成 bash 工具调用
这也印证了前面代码分析中的结论：Cron Scheduler 只是新增了一个 “时间驱动的消息入口”，真正执行任务时仍然回到原来的 Agent Loop。
Loop 3
第三次循环模型响应结果
第三次循环工具执行结果
第三次循环展示的是第一次定时任务真正执行后的工具结果。模型在上一轮根据 [Scheduled] 消息生成了：
bash({"command": "date"})
工具执行后，results 中出现了标准 tool_result：
{
"type": "tool_result",
"tool_use_id": "call_00_LpyfpqRtU2bWKhZfBWsI6131",
"content": "Fri Jun 5 08:54:19 AM CST 2026"
}
这说明 date 命令已经真实执行，并把当前系统时间返回给 Agent Loop。也就是说，从用户角度看，这个任务已经在指定的 2 分钟周期内自动运行了一次；从代码角度看，它本质上仍然是普通的工具调用结果回填：
messages.append({
"role": "user",
"content": results
})
然后模型在下一轮根据这个工具结果生成自然语言回复，告诉用户当前时间。这里没有任何特殊的 “定时任务执行分支”，仍然是前面所有章节一直沿用的 tool_use → handler → tool_result → messages → LLM，唯一不同的是，这次 tool_use 的源头不是用户实时输入，而是 cron_queue 注入的 [Scheduled] 消息。
Loop 4
第四次循环模型响应结果
第四次循环工具执行结果
第四次循环展示了 recurring 定时任务的第二次自动触发。由于这个任务的 cron 表达式是：
*/2 * * * *
并且 recurring=True，所以它不会在第一次触发后被删除，而是会继续保留在 scheduled_jobs 中，等待下一个匹配的分钟再次触发。
调试中可以看到，新的 fired 仍然是同一个任务：
CronJob(
id="cron_960034",
cron="*/2 * * * *",
prompt="Print the current date. Run: date",
recurring=True,
durable=True
)
它再次被注入为：
{
"role": "user",
"content": "[Scheduled] Print the current date. Run: date"
}
模型再次理解为需要执行 date 命令，于是继续生成：
ToolUseBlock(
name="bash",
input={"command": "date"}
)
工具执行结果也再次返回了新的当前时间：
Fri Jun 5 08:56:09 AM CST 2026
这说明 recurring 定时任务的周期性触发逻辑是有效的。更重要的是，调试结果也证明了 _last_fired 的作用：同一个任务不会在同一分钟里重复疯狂触发，而是在下一个匹配分钟才再次进入 cron_queue。
所以 Loop 4 体现的是 s14 的完整循环能力：第一次命中 → 注入 → 执行 → 返回结果，下一次命中 → 再次注入 → 再次执行 → 返回新结果。
这正是 Cron Scheduler 相比普通工具调用新增的能力：它让同一个 prompt 可以跨时间反复进入 Agent Loop，而不是只在当前用户输入中执行一次。
本次调试过程中的完整输出如下：
从完整输出可以看到，程序启动后先启动了两条后台线程：
[cron] scheduler thread started
[queue processor] started
其中 cron_scheduler_loop 负责按时间检查任务，queue_processor_loop 负责在任务到期后把它送回 Agent Loop。用户输入调度需求后，模型先调用 schedule_cron 注册任务，Harness 将其保存为：
cron_960034: */2 * * * * -> Print the current date. Run: date
随后，当时间命中 cron 表达式时，系统打印出：
[cron fire] cron_960034 -> Print the current date. Run: date
这表示调度线程已经发现任务到期，并将其放入 cron_queue。紧接着出现：
[queue processor] delivering scheduled work
[inject cron] Print the current date. Run: date
这说明队列处理线程接管了这个到期任务，并把它作为 [Scheduled] ... 消息注入到 Agent Loop 中。之后模型像处理普通用户请求一样生成 bash 工具调用，执行 date，最终返回当前系统时间。
因此，通过本次调试我们知道：s14 的 Cron Scheduler 并没有让模型本身具备 “定时等待” 的能力，而是在 Harness 层新增了一个外部时钟、一个待执行队列和一个自动投递线程。定时任务到点后，会被重新转换成普通消息进入 Agent Loop。这样，原来的 messages → LLM → tool_use → tool_result 主流程没有被破坏，但 Agent 已经具备了 “未来自动执行任务” 的能力。
OK，以上就是 s14 Cron Scheduler 工作原理的完整分析了。
那大家感兴趣的话可以试试下面这些 prompt 感受下定时调度引入后的一些变化：
1. Schedule a task to print the current date every 2 minutes
2. List all cron jobs
3. Create a one-shot reminder in 1 minute to check the build status
4. Cancel the recurring job and verify with list_crons
6. 相对 s13 的变更
| 组件 | 之前 (s13) | 之后 (s14) |
|---|---|---|
| 触发方式 | 用户手动触发 | 调度线程自动入队 |
| 新类型 | — | CronJob dataclass (id, cron, prompt, recurring, durable) |
| 新函数 | — | cron_matches, validate_cron, schedule_job, cancel_job, cron_scheduler_loop, queue_processor_loop |
| 新存储 | — | .scheduled_tasks.json (durable) + 内存 (session-only) |
| 线程 | 后台执行线程 | + 调度线程 (daemon, 1s 轮询) + queue processor 线程 |
| 队列 | background_results | + cron_queue (调度线程写, queue processor 交付, agent_loop 消费) |
| 工具 | 8 (s12/s13) | + schedule_cron, list_crons, cancel_cron (11) |
7. 小结
s14 这一节我们学习的是 Agent 工程中的 “定时触发层”。如果说前面的 Agent Loop 主要处理的是用户实时输入，那么 s14 开始让 Agent 拥有一种新的工作来源：时间。通过 cron 表达式，用户可以把一个普通请求变成可重复触发的计划任务；通过 CronJob，系统可以把这个计划任务保存成结构化状态；通过 cron_scheduler_loop，后台线程可以独立观察时间；通过 cron_queue，调度线程可以把已触发任务安全地交给主循环；最后通过 agent_loop，这些定时任务仍然按照普通消息的方式被模型理解和执行。
这一节最重要的工程思想是：调度和执行必须解耦。调度器只负责判断 “什么时候该做”，不负责 “具体怎么做”；队列只负责缓存已触发任务，不负责模型推理；Agent Loop 只负责消费任务，不负责轮询时间。正是因为这几层职责清晰分开，s14 才能在不破坏原有 Agent Loop 的情况下，为系统增加按时间自动行动的能力。
同时，s14 也提醒我们区分 durable 和真正的系统级调度。.scheduled_tasks.json 只能保证任务定义跨重启保留，但不能保证进程关闭期间任务仍然执行。教学版的 cron scheduler 是进程内调度器，它适合解释 Agent 内部如何处理定时任务；如果要做真正生产环境中的长期调度，还需要结合系统级定时器、服务守护、锁机制和多实例协调。
OK，以上就是本期想要分享的全部内容了。
结语
本篇文章我们围绕 s14 Cron Scheduler 这一节，完整梳理了 Agent 在已有循环和后台任务机制之上，如何引入时间触发的调度能力，使系统可以在没有用户输入的情况下，按照预定计划自动执行任务。
相比 s13 的后台任务，s14 的核心价值不在于执行速度或并行能力，而在于触发源从用户输入扩展到时间。通过
CronJob数据结构、cron_scheduler_loop和queue_processor_loop，系统把定时任务和 Agent Loop 解耦：调度线程负责判定任务到期，队列缓存已触发任务，Agent Loop 负责消费并执行。这种分层设计保证了职责清晰，增强了系统可维护性和可扩展性。从工程视角来看，s14 的意义在于将 “未来任务” 纳入系统可控范围。每个 cron 任务可以是持久化的（durable）或者仅存在于会话内（session-only），通过队列安全交付到 Agent Loop，任务执行仍遵循原有消息、prompt、tool_use、tool_result 的流程。系统既不会破坏已有逻辑，又可以新增定时触发能力，这体现了典型的 Harness Engineering 思路：职责分明、层次清晰、可复用。
进一步来看，s14 完成了 Agent 系统在时间维度上的一次质变：Agent 不再完全依赖实时用户输入，而是可以周期性、自主地执行任务，形成 “时间驱动的消息入口”。Recurring 任务保证了任务周期性，_last_fired 控制重复触发，保证系统在多线程环境下仍然安全稳定。
总而言之，Cron Scheduler 的引入让 Agent 系统首次具备真正的 “长期计划能力”，为后续更复杂的调度、跨任务协作和长期运行场景奠定了基础。它不仅是定时触发的实现，更体现了整个系统解耦触发与执行、保持 Agent Loop 稳定与可扩展的工程设计理念。
下篇文章我们将来学习新版教程 s19 MCP Plugin 章节的内容，敬请期待🤗
参考
更多推荐
14
0- 0
已为社区贡献9条内容
所有评论(0)