learn-claude-code s12: 为 Agent 构建进程内 Cron 调度器——让 Agent Loop 按本地时刻自动开工【免费下载链接】learn-claude-codeBash is all you need - A nano claude code–like 「agent harness」, built from 0 to 1项目地址: https://gitcode.com/GitHub_Trending/an/learn-claude-code本篇技术指南围绕 learn-claude-code 教程的第 12 章s12_cron_scheduler展开完整覆盖其核心机制如何为 Agent Harness 增加schedule_cron、list_crons、cancel_cron三个工具通过一条调度线程 一条队列处理线程把未来某个时刻要执行的任务 prompt变成到期后自动投递给 Agent Loop 的一轮会话。读完后你将掌握 5 字段 cron 表达式的解析与校验实现、到期入队与空闲投递的线程协作模型、基于.scheduled_tasks.json的持久化边界以及 at-least-once 投递保证的完整实现并可对照 实现源码 与 测试用例 逐行验证。问题S11 解决了怎么跑但没解决什么时候开工s12 是课程s01 → ... → s10 → s11 → s12 → [s13](https://link.gitcode.com/i/496ace50312b7435cfaf183851324bbc) → ... → s17中的第 12 站。前一章 s11 处理的是命令启动之后的运行方式耗时的 Bash 命令可以放到后台执行但它既没有记录未来的工作应该在何时启动也不存在任何持续检查当前时刻的组件。对于每天早上 9 点跑测试每 30 分钟检查一次 CI 状态这类诉求仅靠现有的 Agent Loop用户必须等到每个时刻点再次亲手提交 prompt。要让 Harness 自行接管需要三件事把执行计划持久保存、时刻一到就把对应 prompt 放进待处理队列、并在 Agent 空闲时把队列内容交给 Agent Loop。这正是 s12 要补齐的能力。解决方案注册一个定时任务后的完整链路假设 Agent 注册了一个定时任务cron: 0 9 * * * prompt: run tests本地时间 09:00 时调度线程scheduler thread检测到该任务到期把[Scheduled] run tests放入cron_queue队列处理器queue processor等待 Agent 空闲后开启 Agent Loop 的一轮会话模型随后可调用 Bash 工具去跑测试。s12 的代码保留 s04 阶段的 5 个基础工具bash、read_file、write_file、edit_file、glob与 Hooks 机制新增schedule_cron、list_crons、cancel_cron三个工具不包含 s11 的后台命令机制因为本章投递的是开始新工作的 prompt而不是已在运行中命令的结果。tests/test_cron_scheduler.py 中的test_s12_keeps_the_s04_kernel_and_adds_three_cron_tools断言了这一点工具表恰好是上述 8 个且模块不再带有 s11 的background_tasks等属性。运行环境方面代码依赖 requirements.txt 中的anthropic0.25.0与python-dotenv1.0.0并在启动时读取MODEL_ID环境变量构造模型名可选通过ANTHROPIC_BASE_URL指向兼容端点见 code.py L39-L46。核心数据结构CronJob 保存什么dataclass class CronJob: id: str cron: str prompt: str recurring: bool durable: bool pending_delivery: bool False last_fired: str | None None对应 code.py L249-L257各字段职责cron决定任务何时到期prompt是投递给 Agent 的工作内容pending_delivery标记已到期但模型尚未接收的任务是恢复投递与失败回滚的关键状态last_fired记录上一次发火的分钟标记格式%Y-%m-%d %H:%M防止同一分钟内重复入队。模块级全局状态同样简单scheduled_jobs字典保存全部任务、cron_queue列表保存待投递任务、cron_lockthreading.RLock保护两者code.py L260-L262。任务 ID 由new_cron_id()用secrets.token_hex(4)生成前缀固定为cron_冲突时最多重试 100 次code.py L406-L411。5 字段 cron 表达式解析、匹配与校验s12 支持的表达式示例分 时 日 月 曜日 * * * * * 毎分 0 9 * * * 毎日 09:00 */5 * * * * 5 分ごと 0 9 * * 1-5 平日 09:00本章覆盖*、*/N、N、N-M、N,M,...五种形式。schedule_job()在保存前会调用validate_cron()拒绝字段数错误或取值越界的表达式。从源码看匹配逻辑位于 cron_matches()五个字段逐一比对其中周字段做了(moment.weekday() 1) % 7的换算使周日为 0、周六为 6与标准 crontab 惯例一致单日day-of-month与星期day-of-week采用经典 crontab 语义两者都是*则恒匹配只有一个为*时以另一个为准两者都指定时取或关系任一匹配即发火code.py L295-L301。校验逻辑位于 validate_cron()先检查必须恰好 5 个字段再按字段各自的合法范围递归校验_validate_cron_field()字段合法范围常见错误提示minute0–59minute: Value 60 is outside [0-59]hour0–23hour: Value 24 is outside [0-23]day-of-month1–31day-of-month: ...month1–12month: ...day-of-week0–60周日day-of-week: ...*/N的步长必须为正整数N-M要求起点不大于终点且两端在范围内逗号列表会递归校验每个子项。tests/test_cron_scheduler.py L96-L105 直接验证了这些行为validate_cron(0 24 * * *)返回含hour的错误、validate_cron(0 9 * *)返回Expected 5 fields且0 9 * * 1-5匹配 2026-08-10周一09:00 而不匹配 09:30。到期入队调度线程每秒读一次本地时间调度循环只有三行cron_scheduler_loop()每 1 秒等待stop_event超时后调用poll_due_jobs(datetime.now())。到期判定与入队poll_due_jobs()def poll_due_jobs(moment: datetime): minute_marker moment.strftime(%Y-%m-%d %H:%M) with cron_lock: for job in list(scheduled_jobs.values()): if job.pending_delivery or job.last_fired minute_marker: continue if cron_matches(job.cron, moment): _enqueue_due_job(job, minute_marker)_enqueue_due_job()的关键设计是先落盘、后入队code.py L461-L474先把pending_delivery与last_fired写入内存并尝试持久化持久化成功后才把任务加入cron_queue若写盘失败则把两个字段恢复到原值并抛出异常——绝不把一个只存在于内存中的投递暴露给队列处理器。这意味着durableTrue的任务其已到期状态要么已安全落盘要么根本没有进入队列。[tests/test_cron_scheduler.py L108-L128](https://link.gitcode.com/i/da65ba663a4939caae331214071ad48d#L108-L128)模拟save_durable_jobs抛OSError(disk full)断言新任务被回滚、不出现在scheduled_jobs中。空闲投递队列处理器不关心时钟只关心 Agent 是否忙queue_processor_loop()自身完全不检查时间code.py L706-L714def queue_processor_loop(stop_eventRUNTIME_STOP): while not stop_event.wait(0.2): if not has_cron_queue() or not agent_lock.acquire(blockingFalse): continue try: if has_cron_queue(): run_agent_turn_locked() finally: agent_lock.release()它每 0.2 秒轮询一次队列非空且能非阻塞地拿到agent_lock时才执行一轮agent_lock保证定时轮次与用户轮次不会同时改动会话。这一调度器只负责判断时刻、处理器负责执行的分工把 cron 匹配保持在毫秒级即使某个定时轮次跑了很久也不会阻塞后续到期检查该设计决策在 web/src/data/annotations/s12.json 中被显式记录A Queue Decouples Due Time from Execution。Agent Loop 侧的投递发生在 agent_loop() 入口fired consume_cron_queue() for job in fired: messages.append({role: user, content: f[Scheduled] {job.prompt}})每个到期任务被包装成一条新的 user message 加入当前会话。围绕这条投递链路有两段可靠性逻辑模型调用失败时回滚若client.messages.create抛异常代码会删除本轮新加入的[Scheduled]消息并调用restore_cron_jobs()把任务的pending_delivery置回True、重新入队code.py L646-L651 与 restore_cron_jobs()。tests/test_cron_scheduler.py L131-L154 验证了这一点模型离线时会话历史保持为空、任务仍在队列与任务表中等待下一次投递且不会留下重复消息。模型接收后确认模型首次成功响应后调用acknowledge_cron_jobs()code.py L498-L525recurringTrue的任务清除pending_delivery等待下一次匹配一次性任务直接从scheduled_jobs中移除。若确认时的写盘失败会恢复pending_delivery、把被移除的一次性任务放回任务表并保证任务回到队列——不丢任务。持久化边界.scheduled_tasks.json与 at-least-once模式保存位置进程重启后durableTrue.scheduled_tasks.json工作目录下见 code.py L44重新加载durableFalse内存消失写盘通过临时文件 os.replace()完成原子替换临时文件名字带 PID 与线程 ID 避免并发碰撞save_durable_jobs()。启动时 load_durable_jobs() 会对每条记录做严格校验cron 表达式必须重新通过validate_cron、ID 必须以cron_开头、prompt 非空单条非法记录只跳过并打印提示但如果整个文件损坏JSON 解析失败启动时会明确报错而不是静默忽略——[tests/test_cron_scheduler.py L178-L186](https://link.gitcode.com/i/da65ba663a4939caae331214071ad48d#L178-L186)写入{broken内容后断言输出包含could not load .scheduled_tasks.json。投递保证是at-least-once若模型已接收 prompt但确认状态尚未写盘时进程退出重启后同一任务可能被再次投递。因此定时 prompt 的措辞最好天然幂等例如检查 X 并汇报而不是给 X 发一封邮件。另外load_durable_jobs()会把pending_deliveryTrue的记录直接补入cron_queuecode.py L399-L401与写盘先行的设计互为镜像落盘了就意味着要负责投递。运行边界它不是什么原文档明确划定的边界均能在源码中得到印证调度器使用 Agent 进程的本地时间datetime.now()code.py L624-L626Agent 进程退出调度线程随之停止两个线程都是 daemon 线程start_runtime_threads()durable保住的是任务定义不是执行时机重启会加载已保存的任务但不会补跑停机期间错过的时刻定时轮次运行在队列处理器线程中凡是需要交互式许可的 tool call 会被直接拒绝request_permission()先检查threading.current_thread() is not threading.main_thread()非主线程直接返回Permission denied: scheduled turns cannot request interactive approvalcode.py L176-L185。[tests/test_cron_scheduler.py L157-L175](https://link.gitcode.com/i/da65ba663a4939caae331214071ad48d#L157-L175)用patch(builtins.input)断言子线程中触发权限请求绝不会读取键盘输入——避免定时轮次与主终端抢输入调度器与队列处理器线程只在 CLI 运行时启动__main__块中start_runtime_threads()code.py L753-L768仅import code.py不产生任何后台线程[tests/test_cron_scheduler.py L84-L93](https://link.gitcode.com/i/da65ba663a4939caae331214071ad48d#L84-L93)专门断言导入后runtime_started为假且线程枚举中不存在cron-scheduler、cron-queue-processor。如果需要 Agent 关闭期间也照常执行应改用系统级方案crontab、systemd timer 或外部调度器在触发时启动 Agent 进程。三个新工具的接口细节工具定义见 code.py L575-L597schedule_cron(cron, prompt, recurringTrue, durableTrue)必填cron与prompt成功返回Scheduled cron_xxxx: cron - prompt校验失败时返回Error: 原因如Error: Expected 5 fields, got 4list_crons()逐行输出id: cron - prompt 前 60 字符 [recurring|one-shot, durable|session]无任务时输出No cron jobs.run_list_crons()cancel_cron(job_id)从任务表与队列中同时移除durable 任务写盘失败时会把内存状态整体还原cancel_job()。系统提示词也相应扩展提示模型Use schedule_cron for work that should start at a future local timecode.py L48-L51。交互式演示场景见 web/src/data/scenarios/s12.json用户说每个工作日上午提醒我复盘未完成事项Agent 调用schedule_cron注册0 9 * * 1-5的任务后续由调度器与队列处理器接力完成投递。动手试试cd learn-claude-code python s12_cron_scheduler/code.py依次输入以下 promptSchedule run date every 2 minutes and keep it after restart.List all cron jobs.Cancel the cron job you just created.然后观察两处输出.scheduled_tasks.json的内容任务定义已落盘以及到点时出现的[Scheduled] run date消息与[cron] delivered cron_xxxx日志。注意分钟级任务测试期间需要保持 Agent 进程运行——进程一旦退出调度线程随之停止错过的分钟不会被补跑。小结与下一章s12 用一条 1 秒周期的调度线程、一条 0.2 秒周期的队列处理器线程、一把agent_lock和一个 JSON 文件为 Agent Harness 补齐了定时开工能力调度器只判时刻、队列解耦到期与执行、持久化写盘先行、投递 at-least-once。对照阅读入口是 s12_cron_scheduler/README.md另附 中文版 与 日文版。下一章 s13 Agent Teams 要解决的是另一维度的扩展调度器能按时开启 Agent Loop 的一轮会话但那一轮仍由单个 Agent 处理当任务需要多个模块并行调查、修改并汇总结果时Harness 还需要把任务分派给多个 Agent并通过 inbox 收集各自的执行结果入口在 s13_agent_teams/。【免费下载链接】learn-claude-codeBash is all you need - A nano claude code–like 「agent harness」, built from 0 to 1项目地址: https://gitcode.com/GitHub_Trending/an/learn-claude-code创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考