learn-claude-code s05 TodoWrite：用「先列计划再执行」的任务列表解决 Agent 长任务跑偏问题

在 learn-claude-code 这个从零构建 nano claude code 式 agent harness 的课程仓库中，s05 一课专注于 harness 层的「规划」机制：通过给 Agent 增加一个 todo_write 工具和 reminder 计数器，让它在动手之前先列出步骤、执行过程中持续更新状态，从而降低长任务中「做着做着忘了目标」的概率。读完全文，你将掌握 TodoManager 的状态校验与渲染实现、todo_write 的工具注册与分发方式、rounds_since_todo reminder 注入机制的源码细节，并能在本地跑通该课的可运行示例并观察 Agent 的规划行为。

问题：没有计划的 Agent 做着做着就偏了

给 Agent 一个复杂任务：「把所有 Python 文件改成 snake_case 命名，然后跑测试，修好失败。」

Agent 开始干活，改了 3 个文件，跑了个测试，发现 2 个失败，开始修。修着修着，它忘了最初是「改成 snake_case」，测试失败把注意力全吸走了。

对话越长越严重：工具结果不断填满上下文，系统提示的影响力被稀释。一个 10 步重构，做完 1-3 步就开始即兴发挥，因为 4-10 步已经被挤出注意力了。

这正是 s05 要解决的核心问题。课程口号是「An agent without a plan drifts」——没有计划的 agent 走哪算哪，先列步骤再动手，长任务更不容易漏项。

解决方案：保留 s04 全部机制，新增 todo_write 与 reminder 计数器

s05 建立在 s01-s04 的累积之上：保留工具分发（s02）、权限检查（s03）和 Hooks（s04），只新增两样东西——todo_write 工具与 rounds_since_todo reminder 计数器。todo_write 只更新计划状态，实际工作仍由原有 5 个工具（bash、read_file、write_file、edit_file、glob）完成。

新工具仍走既有的 TOOL_HANDLERS[block.name] 分发路径，不改变主循环结构。连续三个工具调用轮次没有使用 todo_write 时，harness 会把 <reminder> 文本追加到第三轮的工具结果批次中。

在课程 17 课的进度体系中，s05 属于「处理复杂工作」阶段（与 s06 Subagent、s08 Context Compact 并列），位于 课程总览 的 s01 → s02 → s03 → s04 → s05 → s06 → ... 序列中。注意仓库当前存在新旧两条教程轨道：根目录 s01_*~s17_* 是现行 17 课轨道，docs/ 与 agents/ 目录是旧的 12 课过渡轨道，其中旧 s03 对应的就是本讲的 TodoWrite 主题，跨轨道引用时不要混用章节号。

核心实现解析（源码级）

s05 的完整实现位于 s05_todo_write/code.py，约 350 行，可直接运行。下面按模块展开。

SYSTEM 提示词：加入「先计划再执行」的引导

s05 相对 s04 的第一个改动在系统提示词（code.py#L48-L53）：

# s05 change: SYSTEM prompt adds planning guidance
SYSTEM = (
    f"You are a coding agent at {WORKDIR}. "
    "Before starting any multi-step task, use todo_write to plan your steps. "
    "Update status as you go."
)

提示词明确给出两条行为指令：多步任务开始前先用 todo_write 列步骤；执行过程中随手更新状态。这是「软约束」，真正保证行为的是后面的工具定义与 reminder 机制。

TodoManager：持有状态、校验更新、渲染进度

TodoManager 是 s05 的核心数据结构（code.py#L110-L166），它持有内存中的任务列表，负责校验更新，并把渲染结果返回给模型：

class TodoManager:
    def __init__(self):
        self.items: list[dict] = []

    def update(self, todos: list | str) -> str:
        if isinstance(todos, str):
            try:
                todos = json.loads(todos)
            except json.JSONDecodeError:
                try:
                    todos = ast.literal_eval(todos)
                except (SyntaxError, ValueError) as e:
                    raise ValueError("todos must be a list or JSON array string") from e

        if not isinstance(todos, list):
            raise ValueError("todos must be a list")
        if len(todos) > 20:
            raise ValueError("Max 20 todos allowed")

        validated = []
        in_progress_count = 0
        for index, todo in enumerate(todos):
            if not isinstance(todo, dict):
                raise ValueError(f"todos[{index}] must be an object")
            content = str(todo.get("content", "")).strip()
            status = str(todo.get("status", "pending")).lower()
            if not content:
                raise ValueError(f"todos[{index}] requires content")
            if status not in ("pending", "in_progress", "completed"):
                raise ValueError(f"todos[{index}] has invalid status '{status}'")
            if status == "in_progress":
                in_progress_count += 1
            validated.append({"content": content, "status": status})

        if in_progress_count > 1:
            raise ValueError("Only one todo can be in_progress at a time")

        self.items = validated
        return self.render()

    def render(self) -> str:
        if not self.items:
            return "No todos."
        lines = []
        for todo in self.items:
            marker = {"pending": "[ ]", "in_progress": "[>]", "completed": "[x]"}[todo["status"]]
            lines.append(f"{marker} {todo['content']}")
        done = sum(todo["status"] == "completed" for todo in self.items)
        lines.append(f"\n({done}/{len(self.items)} completed)")
        return "\n".join(lines)

从源码结构看，update() 的校验规则与文档描述一一对应，可以归纳为一条参数约束表：

约束项	规则	违反时的错误信息
项数上限	一次更新最多 20 项	Max 20 todos allowed
content	每项必须有非空 content	todos[i] requires content
status 取值	仅限 pending / in_progress / completed（默认 pending，大小写不敏感）	todos[i] has invalid status '...'
并发约束	同一时间只能有一个 in_progress	Only one todo can be in_progress at a time
原子性	校验全部通过后才替换 self.items	校验失败时旧状态保持不变

原子替换是一个容易被忽略但很重要的细节：任何一条校验失败都会抛出 ValueError，self.items 保持原值不变。测试用例 test_rejects_invalid_updates_without_replacing_state 专门验证了这一点——先写入一条合法任务，再分别提交空 content、双 in_progress、21 项三种非法更新，断言每次返回 Error: 且 TODO.items 仍是原来那条 {"content": "keep this", "status": "pending"}。这避免了模型一次拼坏参数就把已有计划整个冲掉。

字符串输入的双路解析：update() 接受 list | str。若传入字符串，先尝试 json.loads，失败后回退到 ast.literal_eval，两路都不行才报错。整个过程不使用 eval，这是有意的安全设计——ast.literal_eval 只能解析字面量，无法执行 __import__ 之类的表达式。对应测试 test_issue_340_does_not_eval_string_inputs 传入字符串 __import__('pathlib').Path(...).write_text('bad')，断言结果以 Error: 开头且目标标记文件并未被创建；另外两条测试（L72、L87）分别验证 JSON 数组字符串与 Python 列表表示字符串（单引号风格）都能被正常接受。

渲染格式：render() 把三态映射为 [ ]（pending）、[>]（in progress）、[x]（completed），结尾附 (done/total completed) 进度统计。这份渲染文本会作为 tool_result 返回给模型，也同步打印到终端。

run_todo_write：工具入口与终端展示

模块级单例 TODO = TodoManager() 与入口函数（code.py#L169-L178）：

TODO = TodoManager()

def run_todo_write(todos: list | str) -> str:
    try:
        output = TODO.update(todos)
    except ValueError as e:
        return f"Error: {e}"
    print(f"\n\033[33m## Current Tasks\033[0m\n{output}")
    return output

注意两点：校验失败被捕获为 Error: ... 字符串返回给模型（而不是抛出异常终止循环），让模型有机会自己修正参数；终端输出带黄色 ## Current Tasks 标题，便于人工旁观 Agent 的计划状态。

工具注册：加入 TOOLS 与 TOOL_HANDLERS

工具定义和其他 5 个工具一起加入 dispatch map（code.py#L180-L199）：

TOOLS = [
    {"name": "bash", ...},
    {"name": "read_file", ...},
    {"name": "write_file", ...},
    {"name": "edit_file", ...},
    {"name": "glob", ...},
    # s05: new tool
    {"name": "todo_write", "description": "Create and manage a task list for your current coding session.",
     "input_schema": {"type": "object", "properties": {"todos": {
         "type": "array", "maxItems": 20,
         "items": {"type": "object", "properties": {
             "content": {"type": "string", "minLength": 1},
             "status": {"type": "string", "enum": ["pending", "in_progress", "completed"]},
         }, "required": ["content", "status"]}}}, "required": ["todos"]}},
]

TOOL_HANDLERS = {
    "bash": run_bash, "read_file": run_read, "write_file": run_write,
    "edit_file": run_edit, "glob": run_glob, "todo_write": run_todo_write,
}

从源码结构看，schema 层的 maxItems: 20、minLength: 1、enum 与 TodoManager.update() 的运行时校验是同一套规则的两种表达：前者约束模型生成的参数形状，后者兜底校验实际到达的值。新增工具完全不需要改主循环——这正是 s02 确立的「加工具 = 加一个 handler」原则的延续。

Reminder 计数器：连续三轮不更新计划就注入提醒

主循环中的关键改动（code.py#L278-L326）：

def agent_loop(messages: list):
    rounds_since_todo = 0
    while True:
        response = client.messages.create(
            model=MODEL, system=SYSTEM, messages=messages,
            tools=TOOLS, max_tokens=8000,
        )
        messages.append({"role": "assistant", "content": response.content})
        ...
        results = []
        used_todo = False
        for block in response.content:
            if block.type != "tool_use":
                continue

            blocked = trigger_hooks("PreToolUse", block)
            if blocked:
                results.append({"type": "tool_result", "tool_use_id": block.id,
                                "content": str(blocked)})
                continue

            handler = TOOL_HANDLERS.get(block.name)
            try:
                output = handler(**block.input) if handler else f"Unknown: {block.name}"
            except Exception as e:
                output = f"Error: {e}"

            trigger_hooks("PostToolUse", block, output)

            if block.name == "todo_write":
                used_todo = True

            results.append({"type": "tool_result", "tool_use_id": block.id,
                            "content": str(output)})

        rounds_since_todo = 0 if used_todo else rounds_since_todo + 1
        if rounds_since_todo >= 3:
            results.append({"type": "text",
                           "text": "<reminder>Update your todos.</reminder>"})
            rounds_since_todo = 0

        messages.append({"role": "user", "content": results})

机制拆解：

1. 按轮计数，而非按调用计数：rounds_since_todo 在每轮 assistant 回复处理完毕后结算——本轮出现过 todo_write 调用就归零，否则加一。注意计数发生在工具执行（含被权限 hook 拦截）之后、结果批次写回 messages 之前。
2. reminder 的注入位置：达到阈值（>= 3）时，reminder 以 {"type": "text"} 块追加到第三轮的工具结果批次末尾，与同批次的 tool_result 一起作为一条 user 消息发回模型。这种「贴着工具结果注入」的方式比单独发一条 user 消息更符合对话结构，也不打断 stop 逻辑。
3. 注入后清零：rounds_since_todo = 0，避免下一轮重复注入。
4. s04 的 hooks 原样保留：PreToolUse 的权限检查与日志、PostToolUse 的大输出警告、Stop 的会话统计等回调（code.py#L202-L273）都还在原位置，reminder 逻辑没有侵入任何 hook。

行为验证来自 test_appends_one_reminder_to_the_third_tool_result_batch：测试用 fake client 连续返回 3 轮只调 glob 的 tool_use 回复加 1 轮 end_turn，断言前两个结果批次中没有 text 块，而第三个批次恰好包含一条 {"type": "text", "text": "<reminder>Update your todos.</reminder>"}。

Agent 收到任务后的典型流程

结合 SYSTEM 提示词与 reminder 机制，模型侧的典型行为序列是：

1. 先调 todo_write 列出所有步骤（全 pending）；
2. 做一个步骤，把它改成 in_progress；
3. 做完改成 completed；
4. 看下一个 pending，继续；
5. 若中途连续 3 轮忘了更新，会在第 3 轮工具结果里收到 <reminder>Update your todos.</reminder>，被拉回计划轨道。

关键洞察：todo_write 不给 Agent 增加任何执行能力——它改不了文件、跑不了命令。它增加的是规划能力：把「接下来要做什么」从模型注意力中易失的部分，变成 harness 里一份结构化、可校验、始终可见的状态。

相对 s04 的变更

组件	之前 (s04)	之后 (s05)
工具数量	5 (bash, read, write, edit, glob)	6 (+todo_write)
规划能力	无	带状态的 TODO 列表 + reminder
SYSTEM 提示	通用提示	加入「先计划再执行」引导
循环	工具分发与 Hooks	保留分发路径，加入 rounds_since_todo 和 reminder 注入

运行与验证

环境要求

依赖见 requirements.txt：

anthropic>=0.25.0
python-dotenv>=1.0.0
pyyaml>=6.0

运行前需要设置 MODEL_ID 环境变量（代码通过 MODEL = os.environ["MODEL_ID"] 读取，缺失会直接抛错）；可选设置 ANTHROPIC_BASE_URL 指向兼容端点。

试一下

cd learn-claude-code
python s05_todo_write/code.py

启动后进入交互式 REPL（提示符 s05 >>，输入 q 退出），试试这些 prompt：

1. Refactor s05_todo_write/example/hello.py: add type hints, docstrings, and a main guard（期望先列 3 步再执行——example/hello.py 当前就是一个没有类型标注、没有 docstring、没有 main guard 的 6 行小脚本，正好是三步重构的靶子）
2. Create a Python package under s05_todo_write/example/demo_pkg with __init__.py, utils.py, and tests/test_utils.py
3. Review Python files under s05_todo_write/example and fix any style issues

观察重点：第一次工具调用是不是 todo_write？TODO 列了几步？执行过程中状态有没有从 pending 变成 in_progress / completed？终端里会看到黄色的 ## Current Tasks 区块随每次 todo_write 更新。

离线跑测试

不想消耗 API 配额的话，可以直接跑针对本讲机制的单元测试（tests/test_todo_write_string_input.py），它用 fake anthropic/dotenv 模块加载 s05 代码，不发起真实请求：

python -m unittest tests.test_todo_write_string_input

该测试文件同时覆盖 s15 集成 harness 中的同名工具（两个模块共用同一套字符串输入兼容逻辑），用例包括：字符串输入双路解析、拒绝 eval、非法更新不破坏既有状态、以及 reminder 精确注入到第三个结果批次的行为断言。此外 tests/test_chapter_readmes.py 保证 17 个章节目录的三语 README 导航一致且 code.py 均可在 Python 3.11 下编译。

局限与下一步：为什么 TODO 列表还不够

todo_write 解决的是「注意力被稀释」，但解决不了「任务本身太大」。比如「重构整个认证模块」这种任务，它本身就是几十个小任务的集合，放在同一个对话里会被上下文淹没——TODO 列表最多 20 项，且没有持久化，会话结束状态即丢失。

这正是 s06 的切入点：s06 Subagent 把大任务拆成子任务，每个子任务派一个独立 Agent，给子任务全新的 messages[]，其最终文本作为一条 tool_result 返回。子 Agent 有自己的干净上下文，不会互相污染。规划（s05）与隔离（s06）配合，才构成复杂任务的完整处理策略；后续 s08 Context Compact 再解决长上下文的空间问题。