1. 从 s13 到 s14为什么 Agent 需要一个 Cron Scheduler如果你跟着 learn-claude-code 的教程一路走到 s13会发现 Agent 已经能后台跑任务了模型调用 build、test、install 这类耗时命令时不再阻塞主循环而是丢到后台线程执行再通过通知机制把结果带回。但 s13 有一个隐含前提——任务必须由用户或模型在当前这一轮对话里触发。换句话说后台能力只是把「已经被触发的任务」放到后台执行而不是主动在未来某个时间点触发任务。这就是 s14 Cron Scheduler 要解决的问题。它要的不是「让工具跑得更快」而是「让 Agent 可以按时间表生产新的工作」。每天早上 9 点自动检查 PR、每 30 分钟自动查看 CI 状态、每周自动生成一次报告——这些需求靠用户实时输入是做不到的必须引入一个新的触发源时间。s14 的做法很克制它没有推翻前面的主流程而是在已有 Agent Loop 外侧加了一个独立的调度层。当时间匹配某个 cron 表达式时调度器把对应任务放入队列当 Agent 空闲时队列处理器再把这个任务交给正常的 agent_loop 执行。这样一来Agent 的行为不再只能由用户实时输入触发也可以由预先注册的时间表触发。这篇笔记聚焦 Claude Code 并发场景下 s14_new Cron Scheduler 的配置落地从 settings.json 骨架到任务触发链路给出可复制的配置片段与验证动作帮你在本地跑通定时调度并观察并发行为。适合已经跑过 s01 到 s13、想补齐 Concurrency 这一部分内容的读者。2. 前置准备TaoToken 接入与本地环境在动手改配置之前先把模型接入这一层准备好。s14 的调度链路本身不依赖特定模型服务但你要验证「定时任务触发后模型能正常推理并调用工具」就需要一个稳定的 API 入口。我这边用的是 TaoToken它的接口兼容主流 SDK 调用方式配置成本低。官网地址是 https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content API 端点是 https://taotoken.net/api 。注意 API 地址不带 UTM 参数直接写 https://taotoken.net/api 即可。你需要先拿到 API Key。进入控制台创建密钥https://taotoken.net/console?utm_sourcetaotoken_aicg_blog_endutm_contentconsoleutm_campaignrewrite 然后在 API Keys 页面生成https://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_contentapi-keysutm_campaignrewrite 。生成后把 Key 写进环境变量不要硬编码进代码。export TAOTOKEN_API_KEYsk-你的密钥 export TAOTOKEN_BASE_URLhttps://taotoken.net/api本地环境方面确认 Python 版本在 3.10 以上因为 s14 用到了str | None这种联合类型注解。另外确认工作目录可写因为 durable 任务会写入.scheduled_tasks.json。python --version # Python 3.10.x 或更高 ls -la # 确认当前目录有写权限如果你还没跑过前面的章节建议先跑通 s13 的后台任务再进入 s14。因为 s14 的 queue processor 和 s13 的后台线程在并发模型上是叠加关系跳过 s13 直接看 s14 容易在锁的层面卡住。3. settings.json 骨架与 CronJob 数据结构s14 的配置骨架分两部分一部分是运行时的数据结构定义一部分是持久化文件。先看数据结构。from dataclasses import dataclass, asdict from pathlib import Path import threading WORKDIR Path.cwd() 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作为分钟级标记保证同一个任务在同一分钟只触发一次。持久化文件.scheduled_tasks.json的结构就是 CronJob 的序列化结果[ { id: cron_960034, cron: */2 * * * *, prompt: Print the current date. Run: date, recurring: true, durable: true } ]如果你希望任务在进程重启后还能恢复就把durable设为true如果只是当前会话临时用设为false即可进程结束就消失。4. 可复制配置cron 匹配、校验与调度线程这一节给出可以直接复制运行的配置片段。先看 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 表达式是否命中当前时间。from datetime import datetime 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 Monday0 - cron Sunday0 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做转换。另外day-of-month 和 day-of-week 在标准 cron 语义中当两者都被约束时通常采用 OR 逻辑代码最后的return dom_ok or dow_ok就是在模拟这个行为。接下来是校验逻辑。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 fExpected 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然后是调度线程本体。这是 s14 最核心的新增逻辑。import time def cron_scheduler_loop(): Independent daemon thread: poll every 1s, fire matching jobs. 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[cron fire] {job.id} - {job.prompt[:40]}) if not job.recurring: scheduled_jobs.pop(job.id, None) if job.durable: save_durable_jobs() except Exception as e: print(f[cron error] {job.id}: {e})这一段可以看作 s14 的「时钟」。它每秒醒来一次拿当前时间和所有scheduled_jobs做匹配。如果某个任务命中就把它追加到cron_queue。注意调度线程不直接调用 LLM也不直接执行工具只是把「到期任务」放入队列。这样调度系统和 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(targetcron_scheduler_loop, daemonTrue).start() print([cron] scheduler thread started)这里的daemonTrue表示这是一个后台守护线程。主进程还在它就继续跑主进程退出它也会一起结束。5. 验证请求从注册到触发的完整链路配置写好后怎么验证它真的跑通了我试过用下面这个 prompt 来观察整个链路Schedule a task to print the current date every 2 minutes模型不会直接执行date命令而是先把这个需求转换成一次schedule_cron工具调用ToolUseBlock( nameschedule_cron, input{ cron: */2 * * * *, prompt: Print the current date. Run: date, recurring: True } )随后schedule_job()会创建一个新的 CronJob 对象并写入scheduled_jobs。调试中可以看到任务被注册后返回了Scheduled cron_960034: */2 * * * * - Print the current date. Run: date这说明第一次循环完成的是「创建定时任务」而不是「执行定时任务」。真正的执行要等cron_scheduler_loop后续命中时间后再触发。当时间命中 cron 表达式时系统打印出[cron fire] cron_960034 - Print the current date. Run: date紧接着出现[queue processor] delivering scheduled work [inject cron] Print the current date. Run: date这说明队列处理线程接管了这个到期任务并把它作为[Scheduled] ...消息注入到 Agent Loop 中。之后模型像处理普通用户请求一样生成 bash 工具调用执行date最终返回当前系统时间。进入agent_loop()后代码首先执行fired consume_cron_queue() for job in fired: messages.append({ role: user, content: f[Scheduled] {job.prompt} })调试中可以看到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( namebash, input{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 分钟周期内自动运行了一次从代码角度看它本质上仍然是普通的工具调用结果回填。由于这个任务的 cron 表达式是*/2 * * * *并且recurringTrue所以它不会在第一次触发后被删除而是会继续保留在scheduled_jobs中等待下一个匹配的分钟再次触发。第二次触发时新的 fired 仍然是同一个任务再次被注入为[Scheduled] ...消息模型再次执行date返回新的时间。这证明了 recurring 定时任务的周期性触发逻辑是有效的。更重要的是调试结果也证明了_last_fired的作用同一个任务不会在同一分钟里重复疯狂触发而是在下一个匹配分钟才再次进入cron_queue。6. 本篇常见错排查6.1 任务注册成功但从不触发最常见的原因是 cron 表达式字段数不对。s14 只接受标准五段式分 时 日 月 周。如果你写了六段带秒validate_cron会直接返回Expected 5 fields, got 6。检查你注册时返回的消息如果以Error:开头说明校验没过。另一个原因是时区。datetime.now()取的是本地时间如果你的机器时区和预期不符任务会在错误的时间触发。验证方法是在调度线程里临时打印now确认它和你预期的时间一致。6.2 同一分钟内任务被触发多次如果你去掉了_last_fired的判断或者minute_marker的格式写错就会出现这个问题。minute_marker必须是YYYY-MM-DD HH:MM这种分钟级精度不能带秒。带秒的话每次循环 marker 都不同判断就失效了。6.3 用户输入和定时任务同时执行导致会话错乱这是并发场景下最容易踩的坑。s14 用agent_lock保证同一时间只有一个 Agent Loop 在跑。如果你在main里处理用户输入时没有加锁with agent_lock: run_agent_turn_locked(query)那么用户请求还没跑完queue processor 又启动了一轮 Agent Loopsession_history和session_context就会被并发修改。表现是消息顺序错乱、上下文丢失、甚至报 KeyError。6.4 durable 任务取消后重启又回来了取消 durable 任务时除了从scheduled_jobs里 pop还必须同步更新磁盘文件def cancel_job(job_id: str) - str: with cron_lock: job scheduled_jobs.pop(job_id, None) if not job: return fJob {job_id} not found if job.durable: save_durable_jobs() return fCancelled {job_id}如果漏了save_durable_jobs()内存里取消了但.scheduled_tasks.json里还在下次启动load_durable_jobs()又会把它加载回来。6.5 调度线程被单个坏任务拖死cron_scheduler_loop里对每个 job 的匹配都包了 try/except。如果你自己改代码时去掉了这个 try某个 job 的 cron 表达式解析抛异常整个调度线程就会退出之后所有任务都不再触发。表现是「第一个任务正常后面全部静默」。排查方法是看有没有[cron error]输出。7. 语义一致的 CTA 与后续学习路径如果你在排障过程中发现是模型接入层的问题比如请求超时、鉴权失败、返回格式异常优先去检查 API Key 和接入配置。API Keys 管理页面在 https://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_contentapi-keysutm_campaignrewrite 接入文档在 https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_contentdocutm_campaignrewrite 。这两个页面能覆盖大部分接入类报错。如果你想先验证模型本身在定时任务场景下的推理表现比如让它把自然语言调度需求转成 cron 表达式可以直接在模型对话页面测试https://taotoken.net/chat?utm_sourcetaotoken_aicg_blog_endutm_contentmodel-chatutm_campaignrewrite 。输入类似「每个工作日早上 9 点检查 PR」这样的需求看它生成的 cron 表达式是否符合预期再决定要不要写进schedule_cron。如果你打算把 s14 的调度能力用到长期编码或 Agent 工作流里比如让 Agent 每天自动跑测试、定期检查依赖更新那更适合用 Coding Plan 来管理调用配额和并发https://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_contentcoding-planutm_campaignrewrite 。定时任务的特点是低频但长期配额规划比单次调用更重要。回到 s14 本身这一节最值得记住的工程思想是调度和执行必须解耦。调度器只负责判断「什么时候该做」不负责「具体怎么做」队列只负责缓存已触发任务不负责模型推理Agent Loop 只负责消费任务不负责轮询时间。正是因为这几层职责清晰分开s14 才能在不破坏原有 Agent Loop 的情况下为系统增加按时间自动行动的能力。同时也要清楚 durable 和真正系统级调度的区别。.scheduled_tasks.json只能保证任务定义跨重启保留但不能保证进程关闭期间任务仍然执行。教学版的 cron scheduler 是进程内调度器它适合解释 Agent 内部如何处理定时任务如果要做生产环境中的长期调度还需要结合系统级定时器、服务守护、锁机制和多实例协调。下一篇会继续看 s19 MCP Plugin 章节把插件扩展这一块补上。