learn-claude-code s06 深度解析：用嵌套 Agent Loop 为子任务创建隔离上下文（Subagent）

本文基于 learn-claude-code 仓库第 06 章 s06_subagent/README.md 及其可运行实现 s06_subagent/code.py 展开，讲解 Subagent（子代理）这一 harness 委托机制：父 Agent 通过一个名为 task 的同步工具，在干净的 messages[] 中运行一个嵌套 agent loop，只把最终文本作为单条 tool_result 带回父会话。读完你可以理解"消息隔离"与"进程隔离"的边界差异，掌握 run_subagent 的完整实现要点，并知道如何在本仓库中运行、验证该机制。

为什么需要 Subagent：上下文污染问题

在 s06_subagent/README.md 中，作者先给出一个典型场景：Agent 正在修一个 bug，为了追踪调用链读了很多文件，每一次工具调用及其结果都会滞留在父级 messages[] 里。当调用链弄清楚之后，这些中间细节大多不再需要，却仍然持续占用上下文窗口、稀释后续推理的信号密度。

仓库主 README.md 对 harness 工程师的职责划分里也印证了这一点：上下文管理（Manage context）是核心工作之一——"Subagents keep focused work in a separate message list"（子代理把聚焦的工作保留在独立的消息列表中）。s06 要解决的就是这个问题：给子任务一份全新的 messages[]，让它的最终文本作为一条 tool_result 返回——这正是仓库课程列表中 s06 的 motto："Give a subtask fresh messages[]; its final text returns as one tool result"。

从 harness 分层角度看，README 把 s06 归入 "Delegation（委托）" 层：在一个独立的对话上下文中运行一个聚焦任务。注意这里的关键词是对话上下文，而不是进程或文件系统。

核心边界：消息隔离，而非进程隔离

s06 的解决方案可以用一句话概括：同步调用 task，在父循环内运行一个嵌套 agent loop，使用全新的 messages[]；循环结束时，最终文本成为父会话中的 tool result。

README 明确指出一个常被误解的边界：

This is message isolation, not process or filesystem isolation. Parent and subagent run in the same Python process and share WORKDIR, so writes and commands still affect the same workspace.

即这是消息隔离，不是进程隔离，也不是文件系统隔离。父 Agent 与子代理运行在同一个 Python 进程里，共享同一个 WORKDIR，因此子代理的写文件和 shell 命令依然作用于同一工作区。子代理拥有五个基础工具但没有 task 工具，且它的工具调用走与父级完全相同的权限与生命周期 hooks。

[ s06_subagent/code.py ] 中的实现也完全对应这一边界（见 s06_subagent/code.py#L41-L52）：

WORKDIR = Path.cwd()
client = Anthropic(base_url=os.getenv("ANTHROPIC_BASE_URL"))
MODEL = os.environ["MODEL_ID"]

SYSTEM = (
    f"You are a coding agent at {WORKDIR}. "
    "Use task for focused exploration or a self-contained subtask."
)
SUB_SYSTEM = (
    f"You are a coding agent at {WORKDIR}. "
    "Complete the given task, then return a concise final answer."
)

值得注意的两个 system prompt 设计细节：

1. 父级 SYSTEM 主动引导委托："Use task for focused exploration or a self-contained subtask."——通过 system prompt 教会模型何时应该把探索性、自包含的子任务外包出去，而不是自己埋头读一堆文件；
2. SUB_SYSTEM 强调"完成后返回简洁的最终答案"："Complete the given task, then return a concise final answer."——子代理的产出物被约束为一段可被父级直接消费的总结文本，而不是冗长的过程叙述。

README 用一个决策表总结了五个关键设计选择，这也是理解 s06 全部实现细节的总纲：

决策项	选择	原因
对话上下文	全新的 messages[]	父级历史不复制进子代理
执行环境	同一进程、同一 WORKDIR	文件系统变更对两个循环都可见
返回值	只有最终文本	子代理的工具调用与结果不会拷贝进父级 messages
委托深度	SUB_TOOLS 中没有 task	本章只允许一层委托，禁止递归派生
工具策略	共享 Hooks	父级与子代理使用相同的权限检查

下面结合 s06_subagent/code.py 的完整源码逐层展开。

基础工具层：五个共享工具

父级与子代理共享同一组基础工具（s06_subagent/code.py#L113-L132）：

BASE_TOOLS = [
    {"name": "bash", "description": "Run a shell command.",
     "input_schema": {"type": "object", "properties": {"command": {"type": "string"}}, "required": ["command"]}},
    {"name": "read_file", "description": "Read file contents.",
     "input_schema": {"type": "object", "properties": {"path": {"type": "string"}, "limit": {"type": "integer"}}, "required": ["path"]}},
    {"name": "write_file", "description": "Write content to a file.",
     "input_schema": {"type": "object", "properties": {"path": {"type": "string"}, "content": {"type": "string"}}, "required": ["path", "content"]}},
    {"name": "edit_file", "description": "Replace exact text in a file once.",
     "input_schema": {"type": "object", "properties": {"path": {"type": "string"}, "old_text": {"type": "string"}, "new_text": {"type": "string"}}, "required": ["path", "old_text", "new_text"]}},
    {"name": "glob", "description": "Find files matching a glob pattern.",
     "input_schema": {"type": "object", "properties": {"pattern": {"type": "string"}}, "required": ["pattern"]}},
]

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

各实现有几个值得记录的防御性细节：

• run_bash：subprocess.run(command, shell=True, cwd=WORKDIR, ..., timeout=120)，stdout 与 stderr 合并后截断到 50000 字符，超时返回 "Error: Timeout (120s)"；
• run_read：支持可选 limit 参数，超出部分以 ... (N more lines) 提示，避免一次性把大文件灌进上下文；
• run_write / run_edit：均通过 (WORKDIR / path).resolve() 解析路径，run_edit 要求 old_text 精确匹配且只替换一次（text.replace(old_text, new_text, 1)），匹配不到时返回 "Error: text not found in {path}"；
• run_glob：在 root_dir=WORKDIR 下匹配，并额外校验 (WORKDIR / match).resolve().is_relative_to(WORKDIR)，过滤掉符号链接等逃逸出工作区的结果。

这组工具与父级完全共享，是"同进程、同工作区"边界的直接体现——子代理 write_file 写出的文件，父级随后用同一个 WORKDIR 就能看到。

run_subagent：嵌套 Agent Loop 的完整实现

这是 s06 新增的核心代码（s06_subagent/code.py#L257-L293）：

SUB_TOOLS = list(BASE_TOOLS)      # no task tool
SUB_HANDLERS = dict(BASE_HANDLERS)

def extract_text(content) -> str:
    if not isinstance(content, list):
        return str(content)
    return "\n".join(
        getattr(block, "text", "")
        for block in content
        if getattr(block, "type", None) == "text"
    )

def run_subagent(prompt: str) -> str:
    print("\n\033[35m[Subagent started]\033[0m")
    messages = [{"role": "user", "content": prompt}]

    for _ in range(30):
        response = client.messages.create(
            model=MODEL,
            system=SUB_SYSTEM,
            messages=messages,
            tools=SUB_TOOLS,
            max_tokens=8000,
        )
        messages.append({"role": "assistant", "content": response.content})

        if response.stop_reason != "tool_use":
            force = trigger_hooks("Stop", messages)
            if force:
                messages.append({"role": "user", "content": force})
                continue
            print("\033[35m[Subagent done]\033[0m")
            return extract_text(response.content) or "(no summary)"

        results = []
        for block in response.content:
            if block.type != "tool_use":
                continue
            output = execute_tool(block, SUB_HANDLERS)
            print(f"  \033[90m[sub] {block.name}: {output[:100]}\033[0m")
            results.append({
                "type": "tool_result",
                "tool_use_id": block.id,
                "content": output,
            })
        messages.append({"role": "user", "content": results})

    print("\033[35m[Subagent stopped]\033[0m")
    return "Subagent stopped after 30 turns without a final answer."

对照 README 中的教学版本，真实源码多出了两处值得关注的细节：

1. 消息隔离的起点：messages = [{"role": "user", "content": prompt}]——子代理的第一条消息就是任务 prompt 本身，父级的全部历史（可能包含几十条工具调用结果）完全不进入这份列表。这就是 "fresh messages[]" 的全部含义：不是清空，而是从未复制过。

2. 30 轮安全上限：与父级 while True 的无界循环不同，子代理用 for _ in range(30) 设置硬性轮次上限。若 30 轮内模型始终在 tool_use 而没有给出最终文本，函数返回 "Subagent stopped after 30 turns without a final answer." 这条错误文案本身会作为 tool_result 回到父级——父 Agent 能感知到"子任务没跑完"并据此决策（换一种方式重试、或亲自接手）。

3. Stop hook 在子循环中同样生效：当 stop_reason != "tool_use" 时，先执行 trigger_hooks("Stop", messages)；若某个 Stop hook 返回非空值（force 消息），会以新的 user 消息注入并 continue，强制子代理继续工作。README 的简版代码省略了这一段，但真实实现里父级与子级在生命周期 hook 上的行为是完全一致的——对应决策表中 "Tool policy: Shared Hooks" 一行。

4. 终端可观测性：子代理的每次工具调用以 [sub] {tool}: {output[:100]} 缩进打印（output 截断到 100 字符仅用于显示，完整 output 仍写入 messages），配合 [Subagent started] / [Subagent done] 标记，让你能在终端里实时看到"子上下文"的活动，而父级 messages[] 里只有那一条 tool_result。

5. 返回值提取：extract_text 只拼接 type == "text" 的块；若为空则返回兜底文案 "(no summary)"，保证 task 工具永远返回一个可用的字符串。

task 工具注册：父级视角的零特殊分支

子循环定义好之后，把它接入父级只需两行（s06_subagent/code.py#L296-L307）：

TASK_TOOL = {
    "name": "task",
    "description": "Run a subagent with fresh conversation context and return its final text.",
    "input_schema": {
        "type": "object",
        "properties": {"prompt": {"type": "string", "minLength": 1}},
        "required": ["prompt"],
    },
}

TOOLS = [*BASE_TOOLS, TASK_TOOL]
TOOL_HANDLERS = {**BASE_HANDLERS, "task": run_subagent}

父级 agent loop（s06_subagent/code.py#L312-L340）没有任何针对 task 的特殊分支：

def agent_loop(messages: list):
    while True:
        response = client.messages.create(
            model=MODEL,
            system=SYSTEM,
            messages=messages,
            tools=TOOLS,
            max_tokens=8000,
        )
        messages.append({"role": "assistant", "content": response.content})

        if response.stop_reason != "tool_use":
            force = trigger_hooks("Stop", messages)
            if force:
                messages.append({"role": "user", "content": force})
                continue
            return

        results = []
        for block in response.content:
            if block.type != "tool_use":
                continue
            output = execute_tool(block, TOOL_HANDLERS)
            results.append({
                "type": "tool_result",
                "tool_use_id": block.id,
                "content": output,
            })
        messages.append({"role": "user", "content": results})

对模型来说，task 就是一个普通的同步工具：模型发出 tool_use（name=task，input 含 prompt），父循环像调度 bash 一样调度它，不同的是 handler 是一个完整的 agent loop。从父级 messages[] 的角度看，一次 task 调用的"耗时"内部可能发生了几十次模型请求与工具执行，但落进父级历史的消息只有一对：assistant 的 tool_use 块 + user 的一条 tool_result（内容为子代理最终文本）。这就是"中间对话不回流"在消息结构上的准确形态。

共享 Hooks：子代理不享受任何权限豁免

s06 的 task 工具之所以安全，关键在于子代理的工具执行走的是与父级同一个 execute_tool 函数（s06_subagent/code.py#L226-L238）：

def execute_tool(block, handlers: dict) -> str:
    blocked = trigger_hooks("PreToolUse", block)
    if blocked:
        return str(blocked)

    handler = 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)
    return str(output)

run_subagent 内部对每个 tool_use 块调用 execute_tool(block, SUB_HANDLERS)，与父循环调用 execute_tool(block, TOOL_HANDLERS) 完全同构。hooks 的注册（s06_subagent/code.py#L219-L223）：

register_hook("UserPromptSubmit", context_inject_hook)
register_hook("PreToolUse", permission_hook)
register_hook("PreToolUse", log_hook)
register_hook("PostToolUse", large_output_hook)
register_hook("Stop", summary_hook)

其中 permission_hook（s06_subagent/code.py#L156-L180）承担真正的安全边界：

• bash 拒绝清单：DENY_LIST = ["rm -rf /", "sudo", "shutdown", "reboot", "mkfs", "dd if="]，命中即返回 "Permission denied by deny list"，不询问直接拦截；
• 破坏性命令询问：DESTRUCTIVE = ["rm ", "> /etc/", "chmod 777"]，命中则终端交互式询问 Allow? [y/N]，拒绝时返回 "Permission denied by user"；
• 工作区逃逸检查：对 read_file / write_file / edit_file，若 (WORKDIR / path).resolve().is_relative_to(WORKDIR) 为 False（例如 ../../etc/passwd 或绝对路径），同样触发交互式询问。

由于 hooks 是全局列表（HOOKS = {"UserPromptSubmit": [], "PreToolUse": [], "PostToolUse": [], "Stop": []}），子代理的每一次工具调用都会触发同样的 permission_hook、log_hook 与 large_output_hook（output 超过 100000 字符时警告）。也就是说：子代理在能力上被"降权"（少了 task 工具），在约束上与父级完全平权。这与"消息隔离而非进程隔离"的边界一致——同一个进程里共享同一套信任策略。

运行与验证：Try It

仓库根 README.md 给出的安装流程适用于本章：

git clone <本仓库>
cd learn-claude-code
pip install -r requirements.txt
cp .env.example .env   # configure ANTHROPIC_API_KEY

requirements.txt 声明了三个依赖：anthropic>=0.25.0、python-dotenv>=1.0.0、pyyaml>=6.0。此外 s06_subagent/code.py 在启动时通过 load_dotenv(override=True) 读取环境变量，并强制要求 MODEL_ID（MODEL = os.environ["MODEL_ID"]，缺失会直接 KeyError）；ANTHROPIC_BASE_URL 为可选项，设置后会作为 Anthropic(base_url=...) 传入，同时代码会 pop 掉 ANTHROPIC_AUTH_TOKEN 以避免认证头冲突。

运行本章：

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

进入交互终端（提示符 s06 >> ，输入 q 退出）。README 推荐三条验证 prompt：

1. Use a subtask to find what testing framework this project uses——子 Agent 读文件探索，主 Agent 只收到结论；
2. Delegate: read all .py files in agents/ and summarize what each one does——典型的"高 I/O、低保留价值"任务，最适合演示上下文隔离收益；
3. Use a task to create s06_subagent/example/string_tools.py with a slugify(text: str) function, then verify it from the parent agent——让子代理写文件、父级再亲自验证，直接印证"共享 WORKDIR"这一边界：子代理的写操作对父级可见。

观察要点（对应 README 的 What to watch for）：

• 终端是否出现 [Subagent started] / [Subagent done] 标记；
• 子代理的工具调用是否以缩进的 [sub] ... 形式打印，且父级历史中不出现这些中间结果；
• 父级是否只拿着 task 返回的最终文本继续推进。

除人工验证外，仓库提供了自动化测试 tests/test_s06_subagent.py，用 mock 客户端把上述行为固化为断言，共三个用例：

测试	验证内容
test_s06_is_kernel_plus_task	BASE_TOOLS 恰为五个基础工具；父级 TOOLS = 基础工具 + task；子级 SUB_TOOLS 恰为基础工具（无 task）；TASK_TOOL 的 required 字段为 ["prompt"]
test_subagent_starts_with_fresh_messages_and_returns_final_text	模拟两轮响应（先 read_file 的 tool_use，再 end_turn 文本），断言子代理首轮请求的 messages 恰好是 [{role: user, content: prompt}]、工具集合为五个基础工具、返回值即最终文本
test_subagent_file_tools_keep_the_kernel_permission_boundary	让子代理 write_file 一个工作区外的路径，mock 用户输入 n，断言结果为 "Permission denied by user" 且文件未被创建——即权限边界在子代理中同样生效

该测试通过 importlib 以临时目录加载 s06_subagent/code.py，并伪造 anthropic / dotenv 模块、设置 MODEL_ID=test-model，因此可以在无真实 API key 的环境下运行（依赖 pytest）。

横向对照：同一机制在不同章节中的形态

从源码结构看，s06 的 Subagent 实现并非孤立存在，仓库中至少有三处可对照的实现，有助于理解该机制的演进方向：

1. 遗留版本 agents/s04_subagent.py：老 12 章课程中的 Subagent 课（仓库 README.md 的迁移映射表中 "old s04 → new s06"）。它展示了"无 hooks 版"的嵌套循环——危险命令用模块内硬编码的 dangerous 列表拦截，工具执行不走 execute_tool 而是内联在循环里；task 工具额外带一个可选 description 字段，父级循环对 task 做了显式特判分支（if block.name == "task"）并打印 > task (desc): prompt。相比之下，s06 版本把"派发子代理"收敛成了 handler 表中的一个普通条目（TOOL_HANDLERS = {**BASE_HANDLERS, "task": run_subagent}），父循环零特判，权限检查统一交由 hooks——这是 s03/s04 引入 Permission 与 Hooks 之后的架构红利。

2. 集成版 s15_integrated_harness/code.py#L1769-L1859 的 spawn_subagent：作为 17 章累积式运行时的一部分，它保留了 s06 的全部骨架（30 轮上限、max_tokens=8000、SUB_SYSTEM 委托 prompt），有两处可见的演进：一是 SUB_SYSTEM 追加了 "Do not spawn more agents."，把"禁止递归委托"从"不给 task 工具"的结构保证，显式补了一道 prompt 层约束；二是改用 has_tool_use(content) 判断是否继续循环，并在 30 轮耗尽后反向扫描 messages 寻找最后一个非空文本作为总结返回，而不是直接返回"没答完"的错误文案——从源码结构看，这是对子代理"边想边答但没停"这一边缘情况更宽容的收尾策略。

3. s07 的衔接：s06_subagent/README.md 的 What's Next 部分指出——不同任务需要不同知识（改前端组件要 React 约定、写 SQL 要表结构），把全部知识塞进 system prompt 会撑爆上下文，因此下一章 s07_skill_loading/README.md 转向按需加载知识：先列技能清单，需要时才展开。Subagent 解决"上下文装不下过程"，Skill Loading 解决"system prompt 装不下知识"，二者共同构成 harness 的上下文管理支柱。

小结：一条可复用的 Subagent 设计清单

把 s06 的要点浓缩为设计一个 Subagent 机制时的检查清单：

1. 隔离只做到消息层：子代理用全新 messages = [{"role": "user", "content": prompt}] 起步，父级历史零复制；进程、文件系统、环境变量全部共享，因此子代理的产出（写文件、改配置）天然对父级可见；
2. 返回值收敛为最终文本：用 extract_text 只取 text 块，空文本兜底 "(no summary)"；中间工具调用与结果留在子循环内、随子循环销毁；
3. 限制委托深度：SUB_TOOLS = list(BASE_TOOLS) 天然不含 task，结构上杜绝递归派生（集成版再叠加 prompt 层 "Do not spawn more agents."）；
4. 设置轮次上限：for _ in range(30) 保证嵌套循环必然终止，超时文案作为 tool_result 回到父级供其决策；
5. 权限策略不降级：子代理工具执行复用同一 execute_tool + hooks 链，deny list、破坏性命令询问、工作区边界检查与父级完全一致；
6. 把 task 注册为普通 handler：TOOL_HANDLERS = {**BASE_HANDLERS, "task": run_subagent}，父 agent loop 不需要为"派发子代理"写任何特殊分支。

这一章的价值在于它用最少的代码量（核心增量约 40 行）演示了 harness 工程中"委托"的完整形态：模型自己决定何时 task、写什么样的 prompt，harness 负责提供隔离的上下文、共享的执行环境与统一的信任边界。后续在 s15_integrated_harness/code.py 中，该机制与 TodoWrite、上下文压缩、hooks 等一起汇入同一个 agent loop，构成完整的累积式运行时。