Jean's Blog

一个专注软件测试开发技术的个人博客

0%

Harness Engineering实战--构建自己的Harness Agent

发表于 分类于
学习Harness Engineering学习笔记第四讲

前面我们已经逐一分析了Naive Agent中的八大故障点，接下来，我们将基于这些故障点，亲手编写一个Harness Agent。通过这一实践过程，不仅能够加深对工程掌控力的理解，也能真正将理论落地。目前我们聚焦于这八个典型故障，后续随着系统演进，可能还会发现更多新的故障类型，届时我们会在后续章节中陆续补充。

代码地址：

配置环境搭建

开发环境：

  • python >= 3.13
  • python-dotenv >= 1.2.2
  • opena >= 2.52.0

大模型：deepseek-v4-flash

代码结构及说明

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
agent-evolve/
├── .env                            # DeepSeek/OpenAI-compatible 配置
├── demo.py                         # 主工程运行入口
├── harness_agent/
│   ├── __init__.py                 # 公共 API
│   ├── agent.py                    # 唯一 Agent 主循环
│   ├── budget.py                   # Token / 费用硬预算
│   ├── checkpoint.py               # 原子快照与断点恢复
│   ├── config.py                   # 统一配置
│   ├── context.py                  # 上下文管理
│   ├── core.py                     # 函数式兼容入口
│   ├── evaluator.py                # Generator-Evaluator
│   ├── hooks.py                    # 生命周期事件与 Gate
│   ├── permission.py               # 权限策略
│   ├── planner.py                  # Feature List
│   ├── progress.py                 # JSONL 进度与审计
│   ├── subagent.py                 # 子 Agent 委托
│   ├── tools.py                    # 工具注册、校验与沙箱
│   └── verifier.py                 # 独立 pytest 验证
└── tests/
    ├── test_harness_agent.py       # 八大故障治理测试
    ├── test_harness_extensions.py  # 扩展机制测试
    └── test_verifier_real.py       # 真实 pytest 子进程测试
机制编号 机制名称 主要解决的故障模式 当前实现文件 当前工程实现说明
① Agent Loop ① 循环失控 / ③ Prompt 漂移 harness_agent/agent.py、harness_agent/core.py HarnessAgent.run() 是唯一主循环；max_steps 硬终止；System Prompt 初始化后保持不变；core.py 仅提供函数式兼容入口，不维护第二套循环
② Tool Use ④ Tool 错误吞没 / ⑥ 参数缺口 harness_agent/tools.py、harness_agent/agent.py 自动生成 Function Calling Schema；函数签名绑定；基础类型检查；未知参数拦截；统一 ToolResult 将错误类型和原因回写模型
③ Progress Tracking ⑤ 状态过程不可见 / 缺少审计 harness_agent/progress.py、harness_agent/hooks.py 通过 Hook 记录 Session、轮次、工具、验证、Checkpoint 和终止事件；使用 JSONL 便于人类审计和程序分析；不承担精确恢复职责
④ Context Management ② Context 溢出 / ③ 上下文基线漂移 harness_agent/context.py、harness_agent/agent.py Token 估算；保留固定 System Prompt 和原始 Goal；摘要较早历史；将 Assistant Tool Calls 与 Tool Results 作为完整消息块，避免协议被拆断
⑤ Feature List ② 长任务上下文膨胀（源头防控）/ ⑤ 任务进度丢失 harness_agent/planner.py、demo.py PlanTracker 提供 list/update/clear；支持 replace/merge；状态属于单个会话，不使用示例中的模块全局 TODOS
⑥ Verification Loop ⑦ 模型自判完成 / ④ 下游错误未发现 harness_agent/verifier.py、harness_agent/agent.py 模型无工具调用后仍不能直接结束；必须通过独立 pytest Verifier；失败结果回写模型继续修复；使用 sys.executable 保证解释器一致
⑦ Subagents ② 父上下文膨胀 / ⑧ 子任务失控 harness_agent/subagent.py、harness_agent/tools.py 子 Agent 使用独立消息历史；工具白名单；不复用父 Agent Checkpoint；限制递归深度、最大轮次和 Token；返回结构化子任务结果
⑧ Generator-Evaluator ⑦ 缺少自动化候选评审 harness_agent/evaluator.py 对多个候选按 Rubric 低温度评分；校验分数范围、数量和最佳下标；解析失败显式返回错误；可接入统一 BudgetGuard
— 以上为八个核心机制；以下为四个扩展机制 — — —
⑨ Permission Gate ⑥ 权限缺口 harness_agent/permission.py、harness_agent/tools.py、harness_agent/agent.py pre_tool_use Gate 在执行前否决操作；危险命令正则、工具拒绝列表、Fail-Closed；文件工具另外使用 Workspace 路径沙箱和符号链接解析后检查
⑩ Hooks 贯穿全部机制的事件基础设施 harness_agent/hooks.py、harness_agent/agent.py 支持 Session、Iteration、Tool、Verification、Checkpoint、Stop 生命周期；普通 Hook 失败隔离；安全 Gate 异常时默认拒绝
⑪ Token Budget ⑧ 成本失控 harness_agent/budget.py、harness_agent/agent.py、harness_agent/evaluator.py、harness_agent/subagent.py 模型调用前检查预计输入；调用后记录真实或估算 Token；支持美元费用上限；Evaluator 可共享 BudgetGuard；子 Agent 设置独立硬上限
⑫ Checkpoint Recovery ⑤ 状态丢失 harness_agent/checkpoint.py、harness_agent/agent.py 原子 JSON 快照；保存 Messages、轮次、预算和验证失败次数；支持 resume=True；恢复时校验任务和 System Prompt，成功后清理旧快照

八大故障点治理示例代码

下面的代码分别展示 Naive Agent 中八类故障的典型表现，以及当前 Harness 的解决方式。
示例默认已经准备好以下基础对象：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
from harness_agent import (
    HarnessAgent,
    HarnessConfig,
    HookManager,
    PermissionGate,
    create_workspace_tools,
    pytest_verifier,
)

config = HarnessConfig.from_env()
tools = create_workspace_tools(
    config.workspace,
    timeout=config.tool_timeout_seconds,
)
verifier = pytest_verifier(
    config.workspace,
    "tests",
    timeout=config.tool_timeout_seconds,
)

故障点①：循环失控

故障代码

1
2
3
4
# 没有最大轮次；如果模型一直返回 tool_calls，循环不会结束。
while True:
    response = client.chat.completions.create(...)
    execute_tool_calls(response)

Harness 解决方式

1
2
3
4
5
6
7
8
9
10
11
12
config.max_steps = 10

agent = HarnessAgent(
    config=config,
    tools=tools,
    verifier=verifier,
)
result = agent.run("检查并修复测试")

if result["status"] == "max_steps":
    print("达到最大轮次，已安全停止")
    print("已执行轮次：", result["steps"])

HarnessAgent.run() 的主循环为：

1
2
while steps < self.config.max_steps:
    ...

达到上限后返回结构化状态，而不是继续消耗 API：

1
2
3
4
5
{
  "status": "max_steps",
  "answer": "达到最大执行轮次，任务尚未通过外部验证",
  "steps": 10
}

故障点②：Context 无限膨胀

故障代码

1
2
3
# 每轮都追加 Assistant 和 Tool 消息，但从不清理历史。
messages.append(assistant_message)
messages.append(tool_message)

Harness 解决方式

Agent 每次调用模型前都会执行上下文压缩：

1
2
3
4
5
messages = compress_messages(
    messages,
    config.context_token_limit,
    config.keep_recent_blocks,
)

也可以单独使用上下文模块：

1
2
3
4
5
6
7
8
9
10
11
12
from harness_agent.context import compress_messages, estimate_tokens

before_tokens = estimate_tokens(messages)
compressed_messages = compress_messages(
    messages,
    token_limit=12_000,
    keep_recent_blocks=6,
)
after_tokens = estimate_tokens(compressed_messages)

print("压缩前：", before_tokens)
print("压缩后：", after_tokens)

压缩过程会固定保留：

1
2
3
4
System Prompt
原始 User Goal
较早历史摘要
最近若干完整交互消息块

Assistant Tool Calls 和对应 Tool Results 会被视为一个不可拆分的消息块：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
[
    {
        "role": "assistant",
        "tool_calls": [
            {
                "id": "call-1",
                "function": {
                    "name": "read_file",
                    "arguments": "{\"path\": \"demo.py\"}",
                },
            }
        ],
    },
    {
        "role": "tool",
        "tool_call_id": "call-1",
        "content": "...",
    },
]

因此不会因为压缩而产生缺失 tool_call_id 的协议错误。

故障点③：System Prompt 漂移

故障代码

1
2
3
4
5
while True:
    # 每轮修改 System Prompt，使模型行为基线不断变化。
    messages[0]["content"] = (
        f"你是编程助手。[session_iter={iteration}]"
    )

Harness 解决方式

System Prompt 只在创建 Agent 时设置一次：

1
2
3
4
5
6
7
8
9
10
11
12
FIXED_SYSTEM_PROMPT = """你是一个严谨的编程智能体。
只能在配置的工作区中操作。
必须通过外部验证后才能声称任务完成。"""

agent = HarnessAgent(
    config=config,
    tools=tools,
    verifier=verifier,
    system_prompt=FIXED_SYSTEM_PROMPT,
)

result = agent.run("修复测试")

主循环只读取 self.system_prompt，不会在每轮修改它。恢复 Checkpoint 时还会校验：

1
2
3
4
5
if (
    state.get("task") != task
    or state.get("system_prompt") != self.system_prompt
):
    raise ValueError("checkpoint 与当前任务或系统提示词不匹配")

这能避免使用其他任务或其他 Prompt 的旧状态继续运行。

故障点④：Tool 错误被吞没

故障代码

1
2
3
4
5
try:
    result = tool(**arguments)
except Exception:
    # 模型无法区分空文件、权限错误和工具异常。
    result = ""

Harness 解决方式

所有工具统一返回 ToolResult：

1
2
3
4
5
6
7
8
9
10
11
12
from harness_agent import create_workspace_tools

registry = create_workspace_tools(config.workspace)
result = registry.execute(
    "read_file",
    {"path": "not-exists.txt"},
)

print(result.ok)
print(result.error_type)
print(result.error)
print(result.to_json())

预期结果类似：

1
2
3
4
5
6
{
  "ok": false,
  "value": null,
  "error": "[Errno 2] No such file or directory: 'not-exists.txt'",
  "error_type": "FileNotFoundError"
}

模型可以根据 error_type 选择正确动作，例如：

  • FileNotFoundError：重新确认路径；
  • InvalidArguments：修正工具参数；
  • PermissionDenied：停止危险操作；
  • TimeoutExpired：缩小任务范围或调整策略。

工具名和参数也会在调用前进行校验：

1
2
3
4
5
6
7
8
unknown = registry.execute("unknown_tool", {})
wrong_args = registry.execute(
    "read_file",
    {"path": "README.md", "unexpected": True},
)

assert unknown.error_type == "UnknownTool"
assert wrong_args.error_type == "InvalidArguments"

故障点⑤：状态丢失，无法断点续跑

故障代码

1
2
3
4
5
6
7
messages = [
    {"role": "system", "content": system_prompt},
    {"role": "user", "content": task},
]

# 进程退出后 messages、steps 和 token 全部丢失。
run_loop(messages)

Harness 解决方式

配置 Checkpoint 路径：

1
2
3
4
5
6
7
from pathlib import Path

config.checkpoint_path = (
    Path(config.workspace)
    / ".harness"
    / "checkpoint.json"
)

首次运行：

1
2
3
4
5
6
7
8
agent = HarnessAgent(
    config=config,
    tools=tools,
    verifier=verifier,
)
result = agent.run("完成一个多轮代码任务")

print(result["status"])

如果任务因为最大轮次、预算或异常而中断，Checkpoint 会保存：

1
2
3
4
5
6
任务描述
固定 System Prompt
完整 Messages
已执行轮次
输入和输出 Token
外部验证失败次数

恢复同一个任务：

1
2
3
4
5
6
7
resumed_result = agent.run(
    "完成一个多轮代码任务",
    resume=True,
)

print(resumed_result["steps"])
print(resumed_result["status"])

Checkpoint 使用临时文件和原子替换：

1
2
temporary_path.write_text(checkpoint_json, encoding="utf-8")
os.replace(temporary_path, checkpoint_path)

因此进程在写入期间退出时，不会轻易留下半个 JSON 文件。

故障点⑥：缺少参数校验、权限门禁和路径沙箱

故障代码

1
2
3
4
5
def write_file(path: str, content: str) -> str:
    # 模型可以传入任意绝对路径或 ../ 路径。
    with open(path, "w", encoding="utf-8") as file:
        file.write(content)
    return "done"

Harness 解决方式一：Workspace 路径沙箱

1
2
3
4
5
6
7
8
9
10
11
12
registry = create_workspace_tools(config.workspace)

escaped = registry.execute(
    "write_file",
    {
        "path": "../outside.txt",
        "content": "不允许写出工作区",
    },
)

assert escaped.ok is False
assert escaped.error_type == "PermissionError"

路径会先执行 resolve()，然后检查真实路径是否仍位于 Workspace 内：

1
2
candidate = (workspace / requested_path).resolve()
candidate.relative_to(workspace.resolve())

该检查同时覆盖：

  • ../ 路径逃逸；
  • 工作区外绝对路径；
  • 已存在符号链接指向工作区外部。

Harness 解决方式二：Permission Gate

1
2
3
4
5
6
7
8
9
10
11
12
hooks = HookManager()
gate = PermissionGate()
gate.register_to(hooks)

allowed, reason = hooks.trigger_gate(
    "pre_tool_use",
    "run_bash",
    {"command": "rm -rf /"},
)

assert allowed is False
print(reason)

将 Hooks 接入 Agent：

1
2
3
4
5
6
agent = HarnessAgent(
    config=config,
    tools=tools,
    verifier=verifier,
    hooks=hooks,
)

实际工具执行顺序为：

1
2
3
4
5
6
7
JSON 解析
→ pre_tool_use Gate
→ PermissionGate
→ 函数签名绑定
→ 基础类型校验
→ Workspace 路径检查
→ 实际执行

如果 Permission Handler 自身异常，Gate 会采用 Fail-Closed：

1
2
allowed = False
reason = "权限 Hook 执行失败"

故障点⑦：模型自行判断任务完成

故障代码

1
2
3
if not message.tool_calls:
    # 模型没有调用工具，不代表代码和测试真的正确。
    return message.content

Harness 解决方式

创建独立 pytest Verifier：

1
2
3
4
5
6
7
8
9
10
11
12
verifier = pytest_verifier(
    config.workspace,
    test_path="tests",
    timeout=60,
)

agent = HarnessAgent(
    config=config,
    tools=tools,
    verifier=verifier,
)
result = agent.run("修复所有失败测试")

完成判定变为：

flowchart LR
    A["模型声称完成"] --> B["执行独立 pytest"]
    B --> C{"pytest 是否通过?"}
    C -- 否 --> D["把失败输出反馈给模型"]
    D --> E["模型继续修复"]
    E --> B
    C -- 是 --> F["返回 completed"]

验证失败时，Harness 会追加新的 User 消息：

1
2
3
4
5
6
7
{
    "role": "user",
    "content": (
        "外部验证失败，请根据以下客观结果继续修复，"
        "验证通过前不要声称任务完成：..."
    ),
}

只有 Verifier 返回：

1
2
3
4
5
VerificationResult(
    passed=True,
    summary="pytest 验证通过",
    details={"exit_code": 0},
)

Agent 才返回：

1
2
3
4
5
6
7
{
  "status": "completed",
  "verification": {
    "passed": true,
    "summary": "pytest 验证通过"
  }
}

故障点⑧：Token 和费用成本失控

故障代码

1
2
3
while True:
    # 没有轮次、Token 或费用上限。
    response = client.chat.completions.create(...)

Harness 解决方式

配置 Token 硬上限：

1
2
config.max_steps = 20
config.max_total_tokens = 50_000

可选配置费用上限：

1
2
3
4
# 以下价格仅为配置示例，请按实际模型价格填写。
config.max_cost_usd = 0.50
config.input_cost_per_1k = 0.001
config.output_cost_per_1k = 0.002

Agent 在模型调用前检查预计输入：

1
2
3
budget.check_before_call(
    estimate_tokens(messages)
)

模型返回后再记录真实用量：

1
2
3
4
budget.record(
    input_tokens=response.usage.prompt_tokens,
    output_tokens=response.usage.completion_tokens,
)

如果供应商没有返回 usage，Harness 仍会执行本地估算，不能借此绕过预算。

超出预算时不会继续调用模型：

1
2
3
4
5
result = agent.run("执行受预算控制的任务")

if result["status"] == "budget_exceeded":
    print(result["budget"])
    print(result["errors"])

返回示例：

1
2
3
4
5
6
7
8
9
10
11
{
  "status": "budget_exceeded",
  "budget": {
    "input_tokens": 42000,
    "output_tokens": 9000,
    "total_tokens": 51000,
    "cost_usd": 0.06,
    "max_total_tokens": 50000,
    "max_cost_usd": 0.5
  }
}

子 Agent 同样可以设置独立限制：

1
2
3
4
5
6
7
8
9
10
11
from harness_agent import delegate

subtask_result = delegate(
    "分析测试覆盖缺口",
    config=config,
    tools=tools,
    allowed_tools=["read_file"],
    max_steps=5,
    max_total_tokens=10_000,
    max_depth=2,
)

八大故障治理完整组合示例

下面的代码将循环、上下文、工具、安全、进度、验证、预算和恢复机制组合到同一个 Agent 中：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
from harness_agent import (
    HarnessAgent,
    HarnessConfig,
    HookManager,
    PermissionGate,
    PlanTracker,
    ProgressTracker,
    create_workspace_tools,
    pytest_verifier,
)

# 1. 加载并设置硬限制
config = HarnessConfig.from_env()
config.max_steps = 20
config.max_total_tokens = 100_000
config.context_token_limit = 12_000
config.keep_recent_blocks = 6
config.verification_attempts = 2

# 2. 创建 Workspace 沙箱工具
tools = create_workspace_tools(
    config.workspace,
    timeout=config.tool_timeout_seconds,
)

# 3. 注册会话级 Feature List
planner = PlanTracker()
tools.register(
    planner.tool,
    name="todo",
    description="查询、更新或清空当前任务计划",
)

# 4. 创建生命周期 Hooks
hooks = HookManager()

# 5. 注册权限门禁
permission_gate = PermissionGate()
permission_gate.register_to(hooks)

# 6. 注册结构化进度日志
progress = ProgressTracker(
    config.workspace / ".harness" / "progress.jsonl"
)
progress.register_to(hooks)

# 7. 创建独立 pytest 验证器
verifier = pytest_verifier(
    config.workspace,
    test_path="tests",
    timeout=config.tool_timeout_seconds,
)

# 8. 创建并运行 Harness Agent
agent = HarnessAgent(
    config=config,
    tools=tools,
    verifier=verifier,
    hooks=hooks,
)

result = agent.run(
    "先制定计划，再检查当前工程测试；如果失败则修复，"
    "直到外部 pytest 验证通过。",
    resume=False,
)

print("状态：", result["status"])
print("轮次：", result["steps"])
print("预算：", result["budget"])
print("验证：", result["verification"])
print("错误：", result["errors"])

如果任务之前中断，可以使用完全相同的任务描述恢复：

1
2
3
4
5
result = agent.run(
    "先制定计划，再检查当前工程测试；如果失败则修复，"
    "直到外部 pytest 验证通过。",
    resume=True,
)

该组合对应关系如下：

故障点 生效机制 关键配置或对象
① 循环失控 Agent Loop config.max_steps
② Context 溢出 Context Management context_token_limit、keep_recent_blocks
③ Prompt 漂移 Stable System Prompt HarnessAgent.system_prompt
④ Tool 错误吞没 Tool Use ToolRegistry、ToolResult
⑤ 状态丢失 Checkpoint Recovery、Progress checkpoint_path、ProgressTracker
⑥ 权限缺口 Permission Gate、Workspace Sandbox PermissionGate、create_workspace_tools
⑦ 模型自判完成 Verification Loop pytest_verifier
⑧ 成本失控 Token Budget max_total_tokens、max_cost_usd

Agent主循环

flowchart TD
    Start["启动任务"] --> Init["初始化消息、预算和错误状态"]
    Init --> Resume{"resume=True 且存在 Checkpoint?"}
    Resume -- 是 --> Load["恢复 Messages、轮次、预算和验证次数"]
    Resume -- 否 --> Session["触发 session_start"]
    Load --> Session

    Session --> Limit{"steps < max_steps?"}
    Limit -- 否 --> Max["返回 max_steps"]
    Limit -- 是 --> Pre["触发 pre_iteration"]
    Pre --> Compress["压缩过长上下文"]
    Compress --> PreBudget["调用前 Token 预算检查"]
    PreBudget --> LLM["调用大模型"]
    LLM --> Usage["记录真实或估算 Token"]
    Usage --> ToolDecision{"存在 Tool Calls?"}

    ToolDecision -- 是 --> Parse["解析 JSON 参数"]
    Parse --> Gate["触发 pre_tool_use Gate"]
    Gate --> Allowed{"权限是否放行?"}
    Allowed -- 否 --> Denied["生成 PermissionDenied"]
    Allowed -- 是 --> Execute["签名校验并执行工具"]
    Denied --> ToolResult["结构化 ToolResult 回写"]
    Execute --> ToolResult
    ToolResult --> Progress["触发 post_tool_use"]
    Progress --> Save["原子保存 Checkpoint"]
    Save --> Limit

    ToolDecision -- 否 --> Verify{"配置了 Verifier?"}
    Verify -- 否 --> Complete["返回 completed"]
    Verify -- 是 --> External["执行独立外部验证"]
    External --> Passed{"验证通过?"}
    Passed -- 是 --> Clear["清理 Checkpoint"]
    Clear --> Complete
    Passed -- 否 --> Attempts{"超过验证重试次数?"}
    Attempts -- 是 --> Failed["返回 verification_failed"]
    Attempts -- 否 --> Feedback["把客观失败结果反馈给模型"]
    Feedback --> Save

终止状态

状态 含义 是否保留 Checkpoint
completed 模型完成且外部验证通过，或配置允许不验证 验证成功时清理
max_steps 达到最大执行轮次 是
budget_exceeded Token 或费用超过硬预算 是
verification_failed 达到外部验证失败次数上限 是
error 未预期异常 是

代码测试及执行结果

测试代码，需要修复的

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
# test_calculator.py
"""计算器模块（包含故意制造的bug）"""

def add(a, b):
    # Bug：本该 a + b，错误写成 a - b
    return a - b

def mul(a, b):
    return a * b


# pytest 测试用例
def test_add():
    assert add(2, 3) == 5

def test_mul():
    assert mul(3, 4) == 12

demo.py

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
# Project ：agent-evolve
# File    ：demo.py
# Author  ：jinglv
# Date    ：2026/8/4
# Software：PyCharm
"""
Harness Agent 真实模型调用演示入口。
"""
import json

from dotenv import load_dotenv

from harness_agent.agent import HarnessAgent
from harness_agent.config import HarnessConfig
from harness_agent.hooks import HookManager
from harness_agent.permission import PermissionGate
from harness_agent.planner import PlanTracker
from harness_agent.progress import ProgressTracker
from harness_agent.tools import create_workspace_tools
from harness_agent.verifier import pytest_verifier


# ====================== 演示入口 ======================
def main() -> None:
    """加载 DeepSeek 配置并执行计算器测试检查任务。"""
    # 从工程根目录 .env 加载 DeepSeek API 配置。
    load_dotenv()
    config = HarnessConfig.from_env()

    # 工具和验证器使用同一个 workspace 与超时配置，避免边界不一致。
    tools = create_workspace_tools(
        config.workspace,
        timeout=config.tool_timeout_seconds,
    )

    # Feature List 使用会话级 PlanTracker，不使用模块全局 TODO。
    planner = PlanTracker()
    tools.register(
        planner.tool,
        name="todo",
        description="查询、更新或清空当前任务计划",
    )

    # Hooks 串联权限门禁和结构化 Progress Tracking。
    hooks = HookManager()
    PermissionGate().register_to(hooks)
    ProgressTracker(
        config.workspace / ".harness" / "progress.jsonl"
    ).register_to(hooks)

    verifier = pytest_verifier(
        config.workspace,
        "tests/test_calculator.py",
        timeout=config.tool_timeout_seconds,
    )

    agent = HarnessAgent(
        config,
        tools,
        verifier=verifier,
        hooks=hooks,
    )
    result = agent.run(
        "检查 tests/test_calculator.py；若测试失败则修复实现，"
        "并运行 pytest 完成外部验证。"
    )
    print(json.dumps(result))


if __name__ == "__main__":
    main()

执行结果

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
{
    "status": "completed",
    "answer": "已完成。\n\n- 检查 `tests/test_calculator.py`：`add` 函数存在故意 bug，写成了 `a - b`。\n- 修复：改为 `return a + b`。\n- 运行 pytest：**21 passed**，外部验证通过。",
    "steps": 4,
    "budget": {
        "input_tokens": 4722,
        "output_tokens": 428,
        "total_tokens": 5150,
        "cost_usd": 0.0,
        "max_total_tokens": 100000,
        "max_cost_usd": null
    },
    "errors": [],
    "verification": {
        "passed": true,
        "summary": "pytest 验证通过",
        "details": {
            "passed": true,
            "exit_code": 0,
            "stdout": "..                                                                       [100%]\n2 passed in 0.00s\n",
            "stderr": ""
        }
    },
    "messages": [
        {
            "role": "system",
            "content": "You are a careful coding agent operating inside a constrained workspace.\nUse tools when evidence is needed. Treat tool errors as actionable data.\nDo not claim completion until the external verifier reports success.\nKeep changes minimal, preserve existing behavior, and stop when the task is verified."
        },
        {
            "role": "user",
            "content": "检查 tests/test_calculator.py；若测试失败则修复实现，并运行 pytest 完成外部验证。"
        },
        {
            "role": "assistant",
            "content": "",
            "tool_calls": [
                {
                    "id": "call_00_FjpjZj6ZXFIPJoPMHjhR7968",
                    "type": "function",
                    "function": {
                        "name": "read_file",
                        "arguments": "{\"path\": \"tests/test_calculator.py\"}"
                    }
                },
                {
                    "id": "call_01_F1Tg08q6Xg3ZvDp0aI6R3114",
                    "type": "function",
                    "function": {
                        "name": "run_pytest",
                        "arguments": "{}"
                    }
                }
            ]
        },
        {
            "role": "tool",
            "tool_call_id": "call_00_FjpjZj6ZXFIPJoPMHjhR7968",
            "content": "{\"ok\": true, \"value\": \"# Project ：agent-evolve\\n# File    ：test_calculator.py\\n# Author  ：jinglv\\n# Date    ：2026/8/5 15:03\\n# Software：PyCharm\\n\\\"\\\"\\\"计算器模块（包含故意制造的bug）\\\"\\\"\\\"\\n\\n\\ndef add(a, b):\\n    # Bug：本该 a + b，错误写成 a - b\\n    return a - b\\n\\n\\ndef mul(a, b):\\n    return a * b\\n\\n\\n# pytest 测试用例\\ndef test_add():\\n    assert add(2, 3) == 5\\n\\n\\ndef test_mul():\\n    assert mul(3, 4) == 12\\n\", \"error\": null, \"error_type\": null}"
        },
        {
            "role": "tool",
            "tool_call_id": "call_01_F1Tg08q6Xg3ZvDp0aI6R3114",
            "content": "{\"ok\": true, \"value\": {\"passed\": false, \"exit_code\": 1, \"stdout\": \"..F..................                                                    [100%]\\n=================================== FAILURES ===================================\\n___________________________________ test_add ___________________________________\\ntests/test_calculator.py:20: in test_add\\n    assert add(2, 3) == 5\\nE   assert -1 == 5\\nE    +  where -1 = add(2, 3)\\n=========================== short test summary info ============================\\nFAILED tests/test_calculator.py::test_add - assert -1 == 5\\n1 failed, 20 passed in 0.41s\\n\", \"stderr\": \"\"}, \"error\": null, \"error_type\": null}"
        },
        {
            "role": "assistant",
            "content": "The test file contains a deliberate bug: `add` returns `a - b` instead of `a + b`. Let me fix it.",
            "tool_calls": [
                {
                    "id": "call_00_aHQbw5q9xXawjFBZVJ994351",
                    "type": "function",
                    "function": {
                        "name": "write_file",
                        "arguments": "{\"path\": \"tests/test_calculator.py\", \"content\": \"# Project ：agent-evolve\\n# File    ：test_calculator.py\\n# Author  ：jinglv\\n# Date    ：2026/8/5 15:03\\n# Software：PyCharm\\n\\\"\\\"\\\"计算器模块（包含故意制造的bug）\\\"\\\"\\\"\\n\\n\\ndef add(a, b):\\n    return a + b\\n\\n\\ndef mul(a, b):\\n    return a * b\\n\\n\\n# pytest 测试用例\\ndef test_add():\\n    assert add(2, 3) == 5\\n\\n\\ndef test_mul():\\n    assert mul(3, 4) == 12\\n\"}"
                    }
                }
            ]
        },
        {
            "role": "tool",
            "tool_call_id": "call_00_aHQbw5q9xXawjFBZVJ994351",
            "content": "{\"ok\": true, \"value\": {\"path\": \"tests/test_calculator.py\", \"bytes\": 358}, \"error\": null, \"error_type\": null}"
        },
        {
            "role": "assistant",
            "content": "",
            "tool_calls": [
                {
                    "id": "call_00_ET_DIfIvth1pGYgqUhAiHnP9895",
                    "type": "function",
                    "function": {
                        "name": "run_pytest",
                        "arguments": "{}"
                    }
                }
            ]
        },
        {
            "role": "tool",
            "tool_call_id": "call_00_ET_DIfIvth1pGYgqUhAiHnP9895",
            "content": "{\"ok\": true, \"value\": {\"passed\": true, \"exit_code\": 0, \"stdout\": \".....................                                                    [100%]\\n21 passed in 0.40s\\n\", \"stderr\": \"\"}, \"error\": null, \"error_type\": null}"
        },
        {
            "role": "assistant",
            "content": "已完成。\n\n- 检查 `tests/test_calculator.py`：`add` 函数存在故意 bug，写成了 `a - b`。\n- 修复：改为 `return a + b`。\n- 运行 pytest：**21 passed**，外部验证通过。"
        }
    ]
}