工具调用失败别急着重试：给 Harness 加一层上下文注入式错误恢复

# 最小复现：工具永远失败，看模型会重试几次
import json

def run_loop(agent_step, tools, user_task, max_steps=12):
    messages = [{"role": "user", "content": user_task}]
    for i in range(max_steps):
        plan = agent_step(messages)              # 模型决定下一步
        if not plan.get("tool"):
            return plan.get("content"), i        # 给出最终答复
        name, args = plan["tool"], plan["args"]
        try:
            result = tools[name](**args)
            messages.append({"role": "tool", "name": name, "content": str(result)})
        except Exception as e:
            # 问题就在这里：只回传一句裸报错
            messages.append({"role": "tool", "name": name, "content": f"ERROR: {e}"})
    return None, max_steps

# 观察点：同一 (工具, 参数) 组合被重复调用的次数

替换掉那句 ERROR: {e}，改为注入下面这个结构。三段缺一不可：

[Error Recovery Context]
Operation: edit_file("src/config.ts", ...)
Error: File not found: src/config.ts
Previous attempts (this session):
  edit_file("src/config.ts", ...) -> File not found
  read_file("src/config.ts")      -> File not found
Recovery suggestions:
  Verify the path exists (use list_directory or find_file)
  Check for typos in directory or filename
  The file may have been moved or renamed earlier in this session

SuggestionCatalog = {
    "FileNotFoundError": [
        "确认路径存在：先调 list_directory 或 find_file",
        "检查目录名或文件名是否拼错",
        "该文件可能在本会话中被移动或重命名过",
    ],
    "PermissionError": [
        "检查凭证是否过期或作用域不足",
        "确认沙箱是否限制了对该目录的写权限",
    ],
    "rate_limit": [
        "退避后重试，不要立刻重发",
        "考虑换用批量接口或降低并发",
    ],
}

class FailureLog:
    # 按 (操作类型, 目标) 记录本会话内的失败，供注入时引用
    def __init__(self):
        self.by_key = {}

    def key(self, name, args):
        target = args.get("path") or args.get("url") or args.get("table") or ""
        return (name, str(target))

    def record(self, name, args, error):
        k = self.key(name, args)
        self.by_key.setdefault(k, []).append({"args": args, "error": str(error)})

    def history(self, name, args, limit=5):
        return self.by_key.get(self.key(name, args), [])[-limit:]

def build_error_context(name, args, error, log):
    lines = ["[Error Recovery Context]"]
    lines.append(f'Operation: {name}({json.dumps(args, ensure_ascii=False)})')
    lines.append(f'Error: {error}')                     # 原文保留
    hist = log.history(name, args)
    if hist:
        lines.append("Previous attempts (this session):")
        for h in hist:
            lines.append(f'  {name}({json.dumps(h["args"], ensure_ascii=False)}) -> {h["error"]}')
    sug = SuggestionCatalog.get(type(error).__name__, [])
    if sug:
        lines.append("Recovery suggestions:")
        for s in sug:
            lines.append(f'  {s}')
    return "
".join(lines)

# 注入：错误结果与结构化块一起进消息列表
log.record(name, args, e)
messages.append({"role": "tool", "name": name, "content": str(e)})
messages.append({"role": "tool", "name": name,
                 "content": build_error_context(name, args, e, log)})

注入式恢复和循环检测解决的是失败生命周期的不同阶段：前者在失败瞬间起作用，靠信息阻断循环形成；后者在循环已经形成后把它掐断。两者要一起用。

MAX_REPEAT = 3

def guard_repeat(log, name, args):
    n = len(log.history(name, args))
    if n >= MAX_REPEAT:
        return {
            "status": "error",
            "code": "LOOP_DETECTED",
            "message": f'已连续 {n} 次在同一操作上失败，停止重试，请换策略或升级给人工',
        }
    return None

# 对照组：裸报错回传    实验组：注入结构化错误块
# 固定随机种子与任务集，记录三个数：
#   1) 坏循环率        = 失败超过 2 次的 (工具, 目标) 占比
#   2) 平均步数        = 完成任务所用工具调用次数
#   3) 输入 token 总量 = 含所有重试的开销

# 期望结果：坏循环率下降，步数下降；token 总量要单独看——
# 因为注入本身也增加 token，近上下文上限的会话可能反而变差