S07 的 agent 会随对话增长积累大量消息历史，最终触发 API 的 prompt_too_long 错误。S08 在每次 LLM 调用前插入四层渐进式压缩，原则：便宜的先做，贵的后做。
CONTEXT_LIMIT / KEEP_RECENT / PERSIST_THRESHOLD 常量compact 工具（主动触发 L4）agent_loop：每轮前运行压缩流水线TRANSCRIPT_DIR / TOOL_RESULTS_DIR 路径常量import time（transcript 文件名用时间戳）import yaml（S08 不需要 YAML 前置解析）messages[]
↓
L3 budget ─→ L1 snip ─→ L2 micro ─→ [token > threshold?]
├─ No → LLM
└─ Yes → L4 summary
↓
LLM call
[prompt_too_long?]
└─ Yes → reactive compact
S07 只有 WORKDIR / SKILLS_DIR。S08 新增三个路径常量和三个数值阈值常量，以及 import time。
| import ast, json, os, subprocess, time | 相对 S07 新增 time。用于 write_transcript 中生成时间戳文件名（int(time.time())）。S08 移除了 S07 的 import yaml（S08 的 _parse_frontmatter 用手写解析替代了 yaml.safe_load）。 |
| TRANSCRIPT_DIR = WORKDIR / ".transcripts" | S08 新增路径。L4 autoCompact 和 Emergency reactive 在压缩前先把完整对话历史写入 .transcripts/ 目录，防止压缩后信息丢失。路径以 . 开头（约定隐藏目录）。 |
| TOOL_RESULTS_DIR = WORKDIR / ".task_outputs" / "tool-results" | S08 新增路径。L3 toolResultBudget 将超过 PERSIST_THRESHOLD 的工具输出持久化到此目录，消息中只保留预览和文件路径引用。 |
| CONTEXT_LIMIT = 50000 | 模块级常量；int 字面量；50000 字符约等于 12k tokens，用于决定是否触发 L4 LLM 摘要。estimate_size(msgs) > CONTEXT_LIMIT 时执行 L4。 |
| KEEP_RECENT = 3 | L2 microCompact 保留最近 N 条工具结果不压缩（最新的结果 LLM 仍需完整参考）。值为 3：保留最近三次工具调用的结果。 |
| PERSIST_THRESHOLD = 30000 | L3 触发阈值：单条工具结果超过 30000 字符时才持久化到磁盘（约 7k tokens）。低于此值的小结果不写磁盘，避免 I/O 开销。 |
压缩时需要：(a) 快速估算 token 使用量；(b) 判断消息类型（含工具调用？含工具结果？），以便在截断时保持 tool_use/tool_result 配对完整（Anthropic API 要求二者必须成对出现）。
| def estimate_size(msgs): return len(str(msgs)) | 单行函数；str(msgs) 将列表序列化为字符串；len() 计算字符数。时间复杂度 O(n)（遍历整个列表）。注释说明：便宜但足够好——用于决定是否触发 L4，不需要精确 token 计数。 |
| def _block_type(block): | 下划线前缀表示模块私有（约定，非语言层面强制）。统一处理两种 block 格式。 |
| return block.get("type") if isinstance(block, dict) else getattr(block, "type", None) | 三元表达式：API 响应里 block 有时是 dict（手动构造的消息），有时是 SDK 对象（有 .type 属性）。isinstance(block, dict)：运行时类型检查；block.get("type")：dict 安全访问；getattr(block, "type", None)：SDK 对象安全属性访问，避免 AttributeError。 |
| def _message_has_tool_use(msg): | 判断 assistant 消息是否包含工具调用。用于 L1 snip_compact：截断 head 边界时，若 head 末尾是工具调用消息，需要把对应的 tool_result 也保留进来。 |
| if msg.get("role") != "assistant": | 快速失败：只有 assistant 消息才可能有 tool_use block，非 assistant 直接返回 False。 |
| content = msg.get("content") | 安全取 content 字段（可能为 None 或 str，不一定是 list）。 |
| if not isinstance(content, list): | 非 list content（如字符串 content）不可能包含 tool_use block。 |
| return any(_block_type(block) == "tool_use" for block in content) | any(generator)：短路求值——找到第一个 tool_use 即返回 True，不遍历剩余 blocks（性能优化）。 |
| def _is_tool_result_message(msg): | 判断 user 消息是否是工具结果返回消息。用于 L1 snip_compact 的边界调整，保证不在 tool_use/tool_result 配对中间截断。 |
| if msg.get("role") != "user": | tool_result 只出现在 user 消息（Anthropic API 协议）。 |
| return any(isinstance(block, dict) and block.get("type") == "tool_result" for block in content) | 注意：这里用 isinstance(block, dict) + block.get("type")，而非 _block_type()。原因：tool_result 消息是手动构造的 dict（非 SDK 对象），此处可以直接检查 dict。any() 同样短路求值。 |
# assistant 发出的工具调用消息（两种 block 类型并存）
msg = {
"role": "assistant",
"content": [
{"type": "text", "text": "我先看看目录结构。"}, # text block
{ # tool_use block
"type": "tool_use",
"id": "toolu_abc123",
"name": "bash",
"input": {"command": "ls -la"}
}
]
}
# _message_has_tool_use(msg) → True（content 中存在 type=="tool_use" 的 block）
# user 返回工具结果消息（对应上面的 tool_use）
msg = {
"role": "user",
"content": [
{
"type": "tool_result",
"tool_use_id": "toolu_abc123", # 与 tool_use.id 对应
"content": "total 8\ndrwxr-xr-x scraper.py\n..."
}
]
}
# _is_tool_result_message(msg) → True
当消息总数超过 max_messages=50 时，保留 head（最早 3 条）和 tail（最新 47 条），删除中间部分，插入占位符 [snipped N messages]。截断时会检查边界是否落在 tool_use/tool_result 配对中间，若是则调整边界以保持配对完整。
| def snip_compact(messages, max_messages=50): | 参数默认值 50；调用时可覆盖。函数纯净：不修改 messages 原地，返回新列表（切片操作产生新 list）。 |
| if len(messages) <= max_messages: return messages | 快速路径：消息数未超限时直接返回原列表（无拷贝开销）。 |
| keep_head, keep_tail = 3, max_messages - 3 | 元组解包：一行赋值两个变量。保留头部 3 条（初始上下文）和尾部 47 条（最近对话）。 |
| head_end, tail_start = keep_head, len(messages) - keep_tail | head_end：头部保留到索引 3（不含）。tail_start：尾部从 len(messages)-47 开始。 |
| if head_end > 0 and _message_has_tool_use(messages[head_end - 1]): | 检查 head 末尾消息是否是工具调用——如果是，需要把对应的 tool_result 也包含进 head，否则会产生孤立的 tool_use（API 报错）。 |
| while head_end < len(messages) and _is_tool_result_message(messages[head_end]): | while 循环向右扩展 head_end：只要下一条是 tool_result 消息就继续纳入 head。head_end < len(messages) 防止越界。 |
| head_end += 1 | 每次扩展一条消息。 |
| if (tail_start > 0 and tail_start < len(messages) | 检查 tail 起始消息是否是 tool_result——若是，需要把前面对应的 tool_use 也纳入 tail，保持配对。 |
| and _is_tool_result_message(messages[tail_start]) | tail 第一条是 tool_result 消息。 |
| and _message_has_tool_use(messages[tail_start - 1])): | tail 前一条（被截断的）是 tool_use 消息。两个条件都满足才需要调整。 |
| tail_start -= 1 | 向左扩展 tail_start：把 tool_use 消息纳入 tail。 |
| if head_end >= tail_start: | 安全检查：若 head 和 tail 重叠（边界调整后），则不压缩（返回原始列表）。 |
| return messages | 不压缩时返回原列表（非拷贝）。 |
| snipped = tail_start - head_end | 计算被删除的消息数量，用于占位符提示。 |
| return messages[:head_end] + [{"role": "user", "content": f"[snipped {snipped} messages]"}] + messages[tail_start:] | 列表拼接：三段拼接。messages[:head_end] 取头部；中间插入单条 user 占位符消息；messages[tail_start:] 取尾部。每次切片都生成新列表，整体是新列表（不修改原 messages）。 |
原始：[msg0, msg1, ..., msg59] （60 条，超过 max_messages=50）
keep_head = 3, keep_tail = 47
head_end = 3 → 保留 [msg0, msg1, msg2]
tail_start = 60 - 47 = 13 → 保留 [msg13, ..., msg59]
若 msg2 是 tool_use 且 msg3 是 tool_result：
head_end 扩展至 4 → 保留 [msg0, msg1, msg2, msg3]
snipped = 13 - 4 = 9
结果：
[msg0, msg1, msg2, msg3,
{"role":"user","content":"[snipped 9 messages]"},
msg13, msg14, ..., msg59]
收集全部对话中的 tool_result block，对除最新 KEEP_RECENT=3 条外的所有旧结果，若其 content 超过 120 字符，则替换为占位符字符串。注意：直接修改 block dict（in-place），无需重建消息列表。
| def collect_tool_results(messages): | 辅助函数：遍历所有 user 消息，收集所有 tool_result block 的引用（包含消息索引 mi 和 block 索引 bi）。 |
| blocks = [] | 结果列表，每个元素是 (消息索引, block索引, block对象) 的元组。 |
| for mi, msg in enumerate(messages): | enumerate()：同时获取索引 mi 和元素 msg。 |
| if msg.get("role") != "user" or not isinstance(msg.get("content"), list): continue | 跳过非 user 消息和非 list content：tool_result 只可能在 user 消息的 list content 中。 |
| for bi, block in enumerate(msg["content"]): | 遍历消息的每个 content block（bi 为 block 索引，当前未使用但保留以备扩展）。 |
| if isinstance(block, dict) and block.get("type") == "tool_result": | 仅收集 tool_result 类型的 dict block。 |
| blocks.append((mi, bi, block)) | 关键：block 是 dict 的引用，不是拷贝。Python dict 是可变对象，后续对 block["content"] 的修改会直接反映到原 messages 中的对应位置（in-place 修改）。 |
| return blocks | 返回 (mi, bi, block_ref) 元组列表，按消息顺序排列。 |
| def micro_compact(messages): | L2 主函数：利用 collect_tool_results 获取引用后，直接修改旧 block 内容。 |
| tool_results = collect_tool_results(messages) | 获取所有 tool_result block 的引用列表（按时间顺序）。 |
| if len(tool_results) <= KEEP_RECENT: return messages | 快速路径：结果总数不超过 KEEP_RECENT（3）时无需压缩。 |
| for _, _, block in tool_results[:-KEEP_RECENT]: | 列表切片 [:-KEEP_RECENT]：取除最后 3 条外的所有"旧"结果。_, _：Python 约定，不使用 mi 和 bi 索引（only block 引用需要）。 |
| if len(block.get("content", "")) > 120: | 只压缩超过 120 字符的结果（短结果保留原文，压缩短文本意义不大且会丢失信息）。block.get("content", "")：安全取值，无 content 字段时用空字符串。 |
| block["content"] = "[Earlier tool result compacted. Re-run if needed.]" | 直接修改 dict 引用（in-place）：由于 block 是原 messages 中 dict 的引用，此赋值直接修改了 messages 中的原始数据。不需要重建消息列表，O(1) 操作。占位符文字建议 LLM 在需要时重新执行工具。 |
| return messages | 返回原 messages 列表（in-place 修改后，引用未变）。 |
# 假设 messages 中有 5 条 tool_result
tool_results = collect_tool_results(messages)
# tool_results = [(0,0,blockA), (2,0,blockB), (4,0,blockC), (6,0,blockD), (8,0,blockE)]
# ↑旧 ↑旧 ↑最新3条保留
# tool_results[:-3] = [(0,0,blockA), (2,0,blockB)]
for _, _, block in tool_results[:-3]:
if len(block.get("content","")) > 120:
block["content"] = "[Earlier tool result compacted. Re-run if needed.]"
# ↑ block 是 dict 引用，此修改直接写入 messages[0]["content"][0]["content"]
# messages 中 blockA 和 blockB 的 content 已被替换，无需重建列表
专门处理最新一条消息中的超大工具结果（如读取了一个 100KB 文件）。将超过 PERSIST_THRESHOLD 的内容写入磁盘，消息中只保留路径引用 + 前 2000 字符的预览。执行顺序在 L1/L2 之前（先处理最大的，再做其他压缩）。
| def persist_large_output(tool_use_id, output): | 辅助函数：将单条工具输出写入磁盘，返回替换后的内容字符串（含文件路径和预览）。 |
| if len(output) <= PERSIST_THRESHOLD: return output | 快速路径：未超阈值直接返回原始输出（无磁盘 I/O）。 |
| TOOL_RESULTS_DIR.mkdir(parents=True, exist_ok=True) | parents=True：自动创建所有中间父目录（.task_outputs/ 不存在时也会创建）；exist_ok=True：幂等——目录已存在不报错。 |
| path = TOOL_RESULTS_DIR / f"{tool_use_id}.txt" | 以 tool_use_id（如 toolu_abc123）命名文件，确保唯一性（Anthropic API 保证每个 tool_use id 全局唯一）。 |
| if not path.exists(): path.write_text(output) | 幂等写入：文件已存在（同一 tool_use_id 被再次 persist）则跳过，避免重复写磁盘。Path.write_text()：写入字符串到文件，UTF-8 编码。 |
| return f"<persisted-output>\nFull output: {path}\nPreview:\n{output[:2000]}\n</persisted-output>" | 返回替换内容：XML 标签包裹、完整路径引用（LLM 可用 bash 工具读取）、前 2000 字符预览（供 LLM 快速判断内容类型）。f-string 多行（反斜杠连接）。 |
| def tool_result_budget(messages, max_bytes=200_000): | L3 主函数。200_000：Python 数字字面量允许下划线分隔（PEP 515），等价于 200000，提高可读性。只处理 messages 最后一条（messages[-1]）——最新工具结果最可能超大。 |
| last = messages[-1] if messages else None | 安全处理空列表：messages[-1] 在空列表上会 IndexError，三元表达式避免此错误。 |
| if not last or last.get("role") != "user" or not isinstance(last.get("content"), list): return messages | 三个快速失败条件：(1) last 为 None/falsy；(2) 非 user 消息；(3) content 非 list——以上任一成立则跳过 L3。 |
| blocks = [(i, b) for i, b in enumerate(last["content"]) if isinstance(b, dict) and b.get("type") == "tool_result"] | 列表推导：从最新消息收集所有 tool_result block（带索引）。 |
| total = sum(len(str(b.get("content", ""))) for _, b in blocks) | 计算最新消息中所有 tool_result 内容的总字节数。str(b.get("content",""))：content 可能是 list 或字符串，str() 统一转换。 |
| if total <= max_bytes: return messages | 总量未超 max_bytes 时直接返回（无需持久化）。 |
| ranked = sorted(blocks, key=lambda p: len(str(p[1].get("content", ""))), reverse=True) | 贪心策略：按 content 长度降序排列——先处理最大的 block，减少持久化次数。lambda p: ...：匿名函数取元组第二个元素（block）的内容长度；reverse=True：降序。 |
| for _, block in ranked: | 按大小从大到小遍历 block。 |
| if total <= max_bytes: break | 总量降到阈值以下则停止（不必持久化所有超大 block）。 |
| content = str(block.get("content", "")) | 获取 content 字符串。 |
| if len(content) <= PERSIST_THRESHOLD: continue | 只持久化超过 PERSIST_THRESHOLD 的 block（小 block 不值得写磁盘）。 |
| tid = block.get("tool_use_id", "unknown") | 获取工具调用 ID，用作文件名（唯一性保证）。"unknown" 是防御性默认值。 |
| block["content"] = persist_large_output(tid, content) | in-place 修改：直接替换 block 的 content 为持久化后的引用字符串。等号左侧修改 dict 值，不创建新 dict。 |
| total = sum(len(str(b.get("content", ""))) for _, b in blocks) | 重新计算总量（因为刚才修改了 block content）。循环条件 total <= max_bytes 下次迭代时检查新值。 |
| return messages | in-place 修改后返回原列表引用。 |
当 L1/L2/L3 之后 estimate_size(messages) > CONTEXT_LIMIT 时，调用 LLM 生成对话摘要，将整个 messages 历史替换为单条摘要消息。压缩前先将完整对话写入 transcript 文件（防止信息丢失）。
| def write_transcript(messages): | 将对话历史写入 .transcripts/ 目录的 JSONL 文件。L4 和 reactive compact 调用压缩前都先执行此函数。 |
| TRANSCRIPT_DIR.mkdir(parents=True, exist_ok=True) | 幂等创建目录（首次调用时创建 .transcripts/ 文件夹）。 |
| path = TRANSCRIPT_DIR / f"transcript_{int(time.time())}.jsonl" | int(time.time())：Unix 时间戳取整（秒级），作为唯一文件名。JSONL 格式便于逐行解析。 |
| with path.open("w") as f: | Path.open("w")：写入模式打开（覆盖，若存在）。with：上下文管理器，保证文件被正确关闭（即使发生异常）。 |
| for msg in messages: f.write(json.dumps(msg, default=str) + "\n") | json.dumps(msg, default=str)：default=str 参数处理不可序列化的对象（如 SDK Block 对象）——遇到无法 JSON 序列化的对象时，调用 str() 转为字符串。每条消息写一行（JSONL 格式）。 |
| return path | 返回 transcript 文件路径（agent_loop 打印路径供用户查看）。 |
| def summarize_history(messages): | 核心 LLM 调用：将对话历史 JSON 化后作为 prompt，请求 LLM 生成结构化摘要。 |
| conversation = json.dumps(messages, default=str)[:80000] | 序列化后截取前 80000 字符（约 20k tokens），防止摘要本身的 prompt 也超长。[:80000] 是字符串切片（O(n) 但无副作用）。 |
| prompt = ("Summarize this coding-agent conversation so work can continue.\n" | 括号内多行字符串隐式拼接（Python 特性）。 |
| "Preserve: 1. current goal, 2. key findings/decisions, 3. files read/changed, " | 明确告诉 LLM 摘要需要保留的五类信息，确保 agent 压缩后仍能继续工作。 |
| "4. remaining work, 5. user constraints.\nBe compact but concrete.\n\n" + conversation) | 字符串拼接 conversation（对话 JSON），作为 prompt 最后部分。 |
| response = client.messages.create(model=MODEL, messages=[{"role": "user", "content": prompt}], max_tokens=2000) | 单独的 LLM 调用（无 tools、无 system prompt）：简化调用，仅用于摘要生成。max_tokens=2000：摘要不超过 2000 tokens（约 1500 汉字）。 |
| return "\n".join( | 从 response 提取文本：遍历 content block 列表。 |
| getattr(block, "text", "") | getattr(block, "text", "")：安全属性访问，非 text block 返回 ""。 |
| for block in response.content | 生成器表达式（配合外层 "\n".join()）。 |
| if getattr(block, "type", None) == "text").strip() or "(empty summary)" | 过滤条件：只取 text block。.strip()：去除首尾空白；or "(empty summary)"：若结果为空字符串（falsy），返回默认值（防御性处理）。 |
| def compact_history(messages): | L4 主函数：组合 write_transcript + summarize_history，返回单条摘要消息列表。 |
| transcript_path = write_transcript(messages) | 先保存，再压缩（防止信息丢失）。 |
| print(f"[transcript saved: {transcript_path}]") | 告知用户 transcript 文件路径，方便事后查阅或调试。 |
| summary = summarize_history(messages) | 调用 LLM 生成摘要（1 API 调用）。 |
| return [{"role": "user", "content": f"[Compacted]\n\n{summary}"}] | 返回单条 user 消息列表：整个 messages 历史被替换为这一条。[Compacted] 前缀让 LLM 知道这是压缩后的摘要。返回新列表（不修改原 messages）。 |
当 API 返回 prompt_too_long 或 too many tokens 错误时触发（即 L1~L4 流水线之后仍然超长）。与 compact_history 类似，但额外保留最后 5 条消息（保证最近的 tool_use/result 配对不丢失）。
| def reactive_compact(messages): | Emergency compact：API 报错后调用。比 L4 更激进——强制执行，无 token 阈值判断。 |
| transcript = write_transcript(messages) | 先保存完整历史（变量 transcript 赋值但未使用，仅为副作用——写文件）。 |
| summary = summarize_history(messages) | 调用 LLM 生成摘要（与 L4 相同逻辑）。 |
| tail_start = max(0, len(messages) - 5) | max(0, ...)：防止负数索引（消息总数少于 5 时 tail_start 为 0，保留全部）。保留最后 5 条是经验值——足够包含最近的 tool_use/result 配对。 |
| if (tail_start > 0 and tail_start < len(messages) | 与 L1 snip_compact 相同的 tail 边界调整逻辑：防止在 tool_use/tool_result 配对中间截断。 |
| and _is_tool_result_message(messages[tail_start]) | tail 第一条是 tool_result。 |
| and _message_has_tool_use(messages[tail_start - 1])): | 其前一条是 tool_use（配对存在）。 |
| tail_start -= 1 | 向左扩展 tail，纳入 tool_use 消息。 |
| return [{"role": "user", "content": f"[Reactive compact]\n\n{summary}"}, *messages[tail_start:]] | 列表解包 *messages[tail_start:]：* 将切片列表展开插入外层列表（Python 3.5+ unpacking generalization）。结果：摘要消息 + 最近 5 条消息。与 L4 的区别：L4 只返回摘要，reactive 同时保留最近消息（API 正在等待工具结果，不能丢）。 |
# compact_history（L4）—— 主动压缩，token 超阈值时触发
return [{"role": "user", "content": f"[Compacted]\n\n{summary}"}]
# → 仅保留摘要，丢弃所有历史消息
# reactive_compact（Emergency）—— 被动压缩，API 报错后触发
return [{"role": "user", "content": f"[Reactive compact]\n\n{summary}"}, *messages[tail_start:]]
# → 摘要 + 最近 5 条消息（保留当前工具调用上下文）
MAX_REACTIVE_RETRIES = 1 # 重试上限：只允许 reactive compact 一次，防止无限循环
S08 在 TOOLS 列表中新增 compact 工具，允许 LLM 主动请求压缩对话历史（不等待 token 超限触发）。这是 S07 完全没有的新工具。
| {"name": "compact", "description": "Summarize earlier conversation to free context space.", | 工具名 compact，描述文字帮助 LLM 理解何时使用（当感知到对话过长时主动调用）。 |
| "input_schema": {"type": "object", "properties": {"focus": {"type": "string"}}}}, | focus 参数为可选字符串：LLM 可指定摘要重点（如 "代码变更"）。当前 compact_history 实现未使用 focus 参数（教学版简化），但 schema 预留接口。required 字段缺省，即无必填参数。 |
S07 的 agent_loop 直接调用 LLM。S08 在每轮 LLM 调用前插入三层预处理器，并用 try/except 捕获 prompt_too_long 错误触发 reactive compact。同时新增对 compact 工具的特殊处理逻辑。
| MAX_REACTIVE_RETRIES = 1 | 模块级常量：reactive compact 最多重试 1 次。防止 reactive compact 之后仍然超长导致无限递归。 |
| def agent_loop(messages: list): | S08 简化了函数签名：移除了 S07 的 rounds_since_todo global 变量逻辑（S08 不再有 todo nag 提醒功能）。 |
| reactive_retries = 0 | 函数级局部变量（非全局）：跟踪本次 agent_loop 调用的 reactive compact 次数。每次用户输入开始一次新的 agent_loop 调用，计数器从 0 开始。 |
| # s08 change: three preprocessors (0 API calls, cheap first) | 注释标注这是 S08 新增的核心逻辑。 |
| messages[:] = tool_result_budget(messages) # L3: persist large results first | messages[:] 切片赋值：in-place 替换列表内容（不改变 messages 变量本身的引用，只修改列表对象内容）。等价于 messages.clear(); messages.extend(...)。这样外层调用者持有的 messages 引用依然有效。顺序：L3 先（最大限度减小 messages 体积，为后续 L1/L2 减负）。 |
| messages[:] = snip_compact(messages) # L1: trim middle | L1 第二步：截断中间消息（cheap，O(n)）。 |
| messages[:] = micro_compact(messages) # L2: old result placeholders | L2 第三步：替换旧工具结果（cheap，O(n)）。 |
| if estimate_size(messages) > CONTEXT_LIMIT: | 三层 cheap 压缩后仍超过 CONTEXT_LIMIT（50000 字符）才触发 L4 LLM 摘要。 |
| print("[auto compact]") | 用户可见日志，提示正在执行 L4 自动压缩（会消耗 1 API 调用，有延迟）。 |
| messages[:] = compact_history(messages) | L4 执行：写 transcript + LLM 摘要 + 替换 messages 内容。 |
| try: | S08 新增 try/except：捕获 LLM API 调用中的 token 超长错误。S07 无此保护。 |
| response = client.messages.create(model=MODEL, system=SYSTEM, messages=messages, tools=TOOLS, max_tokens=8000) | 正常 LLM 调用（与 S07 相同）。 |
| reactive_retries = 0 # reset on successful API call | 成功调用后重置重试计数：下次若再次超长可以再重试一次（而不是因为本次的计数已满而拒绝）。 |
| except Exception as e: | 捕获所有异常——prompt_too_long 是 Anthropic SDK 抛出的异常，其类型取决于 SDK 版本（可能是 BadRequestError 或自定义异常）。用 str(e) 检查错误消息更可靠。 |
| if ("prompt_too_long" in str(e).lower() or "too many tokens" in str(e).lower()) and reactive_retries < MAX_REACTIVE_RETRIES: | str(e).lower()：转小写后检查关键词，兼容不同 SDK 版本的错误消息格式。and reactive_retries < MAX_REACTIVE_RETRIES：限制重试次数（防止无限循环）。 |
| print("[reactive compact]") | 用户可见日志：触发 Emergency 压缩。 |
| messages[:] = reactive_compact(messages) | Emergency 压缩：切片赋值更新 messages 内容。 |
| reactive_retries += 1 | 增加重试计数（最多到 MAX_REACTIVE_RETRIES=1）。 |
| continue | continue 跳回 while True 顶部：重新执行整个压缩流水线（包括 L1/L2/L3/L4 判断），然后再次尝试 LLM 调用。 |
| raise | 非 token 错误（或已超过重试次数）：重新抛出原始异常，让调用者处理（main loop 的 except 会打印错误）。 |
| # s08: compact tool triggers compact_history, not a no-op string | 注释说明：compact 工具有特殊处理逻辑（不通过 TOOL_HANDLERS 路由，而是在 agent_loop 内部直接处理）。 |
| if block.name == "compact": | 特殊判断：compact 工具不在 TOOL_HANDLERS 中，需要在循环里单独处理。 |
| messages[:] = compact_history(messages) | 主动触发 L4 compact（与自动 L4 相同逻辑）。 |
| results.append({"type": "tool_result", "tool_use_id": block.id, | 构造 tool_result 响应（API 要求每个 tool_use 都要有对应的 tool_result）。 |
| "content": "[Compacted. Conversation history has been summarized.]"}) | 告知 LLM 压缩已完成——LLM 可据此继续对话（了解上下文已变化）。 |
| messages.append({"role": "user", "content": results}) | 将 compact 结果追加为 user 消息。 |
| break # end current turn, start fresh with compacted context | break 退出 for block 循环：compact 后不再处理本轮其他工具调用——立刻开始新的一轮（with compacted context）。注释解释了设计意图。 |
| else: | for...else 语法：for 循环正常完成（未被 break 中断，即 compact 工具未被调用）时执行 else 块。 |
| # normal path: no compact was called | 注释说明：正常路径（无 compact 工具调用）。 |
| messages.append({"role": "user", "content": results}) | 正常路径：追加工具结果后 continue 进入下一轮 while 循环。 |
| continue | 显式 continue：进入 while True 下一轮（含新的压缩流水线判断）。 |
| # compact was called: results already appended above | 注释：compact 路径下 results 已在 break 前 append，这里不需要重复 append。 |
| continue | compact 路径的显式 continue（for...else 的 break 落地后执行此行）：继续 while 循环（with compacted history）。 |
while True:
┌─ 压缩流水线（每轮前执行）
│ L3: tool_result_budget(messages)
│ L1: snip_compact(messages)
│ L2: micro_compact(messages)
│ if estimate_size > CONTEXT_LIMIT:
│ messages[:] = compact_history(messages) # L4
│
├─ try:
│ response = client.messages.create(...)
│ reactive_retries = 0 # 成功则重置
│ except prompt_too_long and retries < 1:
│ messages[:] = reactive_compact(messages)
│ reactive_retries += 1
│ continue # ← 重新进入 while 顶部
│ except other:
│ raise
│
├─ messages.append(assistant response)
├─ if stop_reason != "tool_use": return
│
└─ for block in response.content:
if block.name == "compact":
messages[:] = compact_history(messages)
results.append(tool_result)
messages.append(user: results)
break # ← 退出 for 循环
else:
... (normal tool handling)
results.append(...)
else: # for 正常完成（无 compact）
messages.append(user: results)
continue
continue # compact 路径的 continue