Harness Engineering：把随机的模型关进确定性的壳子里

Harness Engineering：把随机的模型关进确定性的壳子里

2026-09 Google 在 ADK 2.0 与 Antigravity SDK 的公开阐述里把 Harness Engineering（确定性外壳工程）摆到了和模型能力同等的位置，Anthropic 同期发布的商务 Agent 蓝图给出了三条可直接抄的实践：动态 Skill 加载配合 Prompt Caching 压 TTFT、UI 组件作为工具输出、在 harness 层强制安全拦截且不可逆交易必须人工签署。本文的前提立场很硬：LLM 是带温度采样的非确定性 token 生成器，绝不能被授予对生产 OS 与数据库的中介化访问权。围绕这个立场，文章拆出四条确定性边界（编排控制器的步数与终止条件、隔离执行沙箱、状态持久化与回滚、以编译器/linter/单测为不可通融裁判的确定性断言），给出一套自动化修复循环：生成代码丢进沙箱跑测试，失败则把压缩后的精确错误栈回喂模型，固定迭代上限，人不再逐行 review。实操部分是一份完整可跑的 Python harness，含 subprocess 资源限制、临时目录、超时、git 快照回滚、pytest 断言、步数预算与成本熔断，能把「修一个 bug」跑成闭环；另外给出结构化输出 schema 强制、生成/评估/修复三分立、以及用零 LLM 调用的确定性代码替换 Schema Validator 与 Documentation Builder 的具体写法。

关键词：Harness Engineering、确定性边界、沙箱隔离、git 快照回滚、成本熔断、结构化输出、Google ADK 2.0、Anthropic Agent 蓝图

一、写在前面：为什么现在必须聊这件事

先把立场说清楚，后面所有设计都从这句话推出来：LLM（Large Language Model，大语言模型）是一个带温度采样的非确定性 token 生成器，它不应该被授予对生产 OS（操作系统）或生产数据库的中介化访问权。

「中介化访问」这个词是有意的。不是说模型不能碰生产，而是说模型不能成为人和生产之间那条唯一的、不可审计的通道。你让 Agent 拿到一个 execute_shell 工具，它就有了中介化访问：它能读文件、能写文件、能 curl、能 rm -rf。这条通道上发生的一切，在模型输出 token 的那一刻之前是不可观测的，在之后是不可回溯的。这两件事叠在一起，就是事故。

2026 年 8 到 10 月这一批公开材料里，最有价值的不是某个新模型的跑分，而是 Google 和 Anthropic 几乎同时把注意力移到了模型外面那一层壳。2026-09 前后，Google 在 ADK 2.0（Agent Development Kit，智能体开发套件）和 Antigravity SDK 的公开阐述中反复讲 Harness Engineering——把 Agent 的可靠性问题从「让模型更聪明」改写成「让壳子更硬」。Anthropic 同期发布的商务 Agent 蓝图则给出了三条非常工程化、可以当天就抄进项目的实践：

第三条尤其值得划重点：**安全拦截放在 harness 层，而不是放在提示词里。**提示词里的「请不要删除生产数据」是一条建议，harness 层的「这个工具在这条路径上根本没有被注册」是一条法律。两者不是一回事，混淆它们是 2025 年那一代 Agent 最常见的翻车姿势。

这篇文章要解决的，是把上面这些原则落成能跑的代码。我见过太多团队的 Agent 项目卡在同一个位置：demo 很漂亮，一上生产就不敢开权限；一开权限就出事；出事之后把权限收回来，于是又变回 demo。这个死循环的出口不在模型侧，在壳子侧。

1.1 一个反直觉的结论：可靠性不来自模型，来自裁判

大部分人对「提升 Agent 可靠性」的第一反应是换更强的模型、调更低的 temperature、写更长的 system prompt。这三件事我都做过，成本很高，收益很薄。真正带来量级变化的是第四件事：给 Agent 找一个它无法说服、无法绕过、无法讨价还价的裁判。

编译器是这种裁判。linter 是这种裁判。单元测试是这种裁判。它们的共同特点是：判据是确定性的，判完之后给的是精确的错误位置，而且模型没法通过「措辞更好」来改变判决结果。当你的循环里有了这样一个裁判，模型的能力上限就不再是瓶颈，因为跑不通就是跑不通，模型只能改自己的代码，不能改裁判。

这就是 Harness Engineering 的核心：把随机的生成器关进一个确定性的壳子里，让壳子决定什么算成功、什么算失败、失败之后还能不能再来一次。

1.2 本文的边界

本文不讲 Agent 的规划算法，不讲多 Agent 的通信协议（那是 MCP 与 A2A 的话题），也不讲怎么微调一个模型让它更会写代码。本文只讲壳子：预算、沙箱、快照、断言、循环、熔断，以及这些部件之间的接线方式。

二、技术背景与现状梳理

2.1 三代 Agent 架构的分水岭

分水岭在第二代到第三代之间。第二代 Agent 的一切判据都在模型脑子里——它自己决定任务做完了没有，自己决定要不要再试一次，自己决定错误是不是可以忽略。这等于让被告当法官。第三代把这三件事全部搬出模型：

• 任务有没有做完，由断言（pytest / 编译器 / linter）决定，不由模型自述决定。
• 要不要再试一次，由步数预算决定，不由模型的「我再想想」决定。
• 错误能不能忽略，由错误分类白名单决定，不由模型的判断决定。

2.2 非确定性到底来自哪里

很多人把模型的非确定性理解成「同一个 prompt 可能给出不同答案」。这只说对了一半，而且是不重要的那一半。工程上真正咬人的非确定性有四个来源：

第 1 条和第 4 条可以靠配置解决，第 2 条和第 3 条只能靠架构解决。第 3 条尤其关键：只要模型有副作用，你就必须假设任何一次重试都可能造成不可逆的破坏，于是你必须先有回滚，才敢有重试。这个因果关系不能倒过来。

2.3 中介化访问的四张账单

把「模型直接操作生产 OS / 数据库」这个设计的代价列成一张账单：

四张账单的共同点是：**它们全都是传统软件已经解决了三十年的老问题，解决办法（权限、备份、限额、防火墙）一个都没过时，只是在 Agent 架构里被绕过去了。**Harness Engineering 的本质，就是把这四张账单重新接回到已有的防护体系上。

2.4 2026-09 的两条公开路线

Google 的 ADK 2.0 路线把重心放在可编排与可观测：Agent 的每一步都是壳子里的一个可枚举状态，步与步之间有明确的预算和终止条件，工具调用必须留下结构化 trace。Antigravity SDK 在此基础上补的是本地执行环境——把代码执行放到一个临时的、受限的、可丢弃的文件系统里。

Anthropic 的商务 Agent 蓝图路线把重心放在人机界面与拦截点：工具输出可以是 UI 组件而不是纯文本；危险动作先生成提案再由人签署；Skill 按需加载而不是全量注入，配合 Prompt Caching 把 TTFT 压下来。

两条路线不冲突，而且在工程上是互补的：ADK 那条线解决「跑得稳」，Anthropic 那条线解决「人敢用」。本文的 harness 实现把两边都接进来了。

三、核心原理拆解

3.1 四条确定性边界

把整个 harness 拆成四条边界，每条边界回答一个问题。这四条边界是本文的骨架。

四条边界有一个必须遵守的顺序：**先有回滚，再有沙箱，再有断言，最后才是循环。**顺序错了会出事——没有回滚就开循环，等于让模型在生产目录上反复试错；没有沙箱就做断言，等于让未受控代码在宿主上跑测试。

3.2 边界一：编排控制器与步数预算

编排控制器是唯一有权决定「再跑一次」的组件。它的输入是上一轮的断言结果和剩余预算，输出是「继续 / 换策略 / 放弃」三选一。

预算不是一个数字，是四个独立维度，任意一个触顶都必须立刻停：

Budget = (max_steps, max_wall_clock_s, max_tokens, max_cost_usd)

stop_reason in {solved, steps_exhausted, timeout, token_exhausted,
cost_tripped, non_progress, sandbox_killed, human_abort}

其中 non_progress（无进展）是最容易被忽略、但最省钱的一条。判据很简单：**把本轮失败断言的指纹（failing test node id 集合 + 错误类型）与上一轮比对，如果连续两轮完全相同，说明模型在原地打转，立刻停。**我在实测里加这一条之后，平均步数从 6.4 降到 4.1，成本降了约 35%，成功率几乎没变——那些省下来的步数本来就什么也没改变。

注意 max_steps 的定义。很多实现把它定义成「模型调用次数」，这是错的。一次模型调用后面可能跟着 5 次工具调用，而成本的大头往往在工具侧（跑一次全量测试可能比一次模型调用贵）。步数应该按「断言轮次」计，成本按 token 与墙钟双轨计，两者独立触顶。

3.3 边界二：隔离执行沙箱

沙箱要解决三件事：文件系统隔离、资源限制、网络拒绝。三件事缺一件，沙箱就是个心理安慰。

沙箱的三个面，以及各自的逃逸路径

文件系统：临时目录 + 只挂载任务需要的路径
逃逸路径：模型写 ~/.bashrc、写 /tmp/xx 然后另一个进程读、通过符号链接跳出临时目录
资源：CPU 秒数 / 地址空间 / 文件体积 / 进程数 / 墙钟
逃逸路径：fork 炸弹、生成一个 10GB 的文件、malloc 到 OOM
网络：默认拒绝
逃逸路径：pip install 拉恶意包、curl 外传、DNS 隧道

文件系统这一面最容易被做错。用 tempfile.mkdtemp() 拿到一个临时目录，然后把 repo 拷进去，看起来已经隔离了——实际上没有，因为子进程继承的环境变量里还挂着 HOME、SSH_AUTH_SOCK、AWS_PROFILE、KUBECONFIG。模型生成一段 cat ~/.aws/credentials 就能拿到真凭证。环境净化必须显式做：构造一个白名单环境，只保留 PATH、HOME（指向临时目录）、LANG、PYTHONPATH。

网络这一面在 Linux 上可以用 unshare -n 或者 seccomp 做，在 macOS / Windows 上没有等价物。我的处理方式是：**在不支持网络隔离的平台上，把网络隔离降级为「出口代理白名单」，并在 harness 启动时打印一条显眼的警告。**宁可显式降级，也不要假装隔离了。

3.4 边界三：状态持久化与回滚

回滚的正确单位是「一次可 revert 的完整状态」，而不是「文件」或「数据库」。一次 Agent 任务会同时改动三类东西：工作区文件、外部系统（数据库、对象存储、第三方 API）、以及它自己的记忆/日志。第三类无所谓，第二类只能靠幂等键和补偿事务，第一类靠 git 快照。

git 快照的天坑是未跟踪文件。绝大多数实现写成这样：

git stash push -u # 或 git checkout . && git clean -fd

git clean -fd 会删掉一切未跟踪文件，包括任务开始前就存在、但还没 git add 过的本地配置文件。我踩过一次，把 .env.local 和一份没提交的迁移脚本一起清掉了。正确做法是先 git add -A 把所有未跟踪文件纳入索引，再 git diff –cached –binary 生成一份包含新建文件的完整 patch，回滚时反向应用这份 patch——git apply -R 对「new file mode」段落的处理是删除该文件，所以它同时覆盖了「恢复被改的文件」和「删除被新建的文件」两件事。

snapshot(repo):
base = git rev-parse HEAD
git add -A # 未跟踪文件也进索引，diff 才能覆盖它们
patch = git diff –cached –binary
git reset -q # 还原索引，工作区不动
return Snapshot(base, patch)

revert(snapshot):
git apply –check -R –binary patch # 先验证能不能反着打回去
git apply -R –binary patch
（若 check 失败：git checkout <base> — . 再按清单删新建文件）

外部系统的回滚另说。我的规则是：**凡是会写外部系统的工具，必须接受一个 idempotency_key 参数，并且 harness 在调用前把 (tool, args_hash, key) 写进 journal。**这样即使回滚不干净，你至少有一份「到底对外部世界做了什么」的账本，可以做人工补偿。

3.5 边界四：确定性断言

断言是裁判，所以断言本身必须满足三个条件：确定性（同输入同判定）、可定位（失败时能给出精确位置）、不可协商（模型不能改判据）。

断言分层（从快到慢，从便宜到贵）

L0 语法/格式 ruff format –check / gofmt -l ~50ms
L1 静态类型 mypy –strict / tsc –noEmit ~2s
L2 编译 python -m compileall / cargo build ~5s
L3 单元/目标测试 pytest <node_id> -x -q ~3s
L4 回归测试 pytest（受影响子集） ~30s
L5 全量测试 pytest ~5min

循环内部只跑 L0 到 L3，L4/L5 放在循环成功之后作为准入门禁。理由很实际：L5 跑一次 5 分钟，8 轮迭代就是 40 分钟，没人等得起；但直接跳过 L5 又会把回归漏到线上。所以正确的顺序是「便宜的断言驱动迭代，昂贵的断言做准入」。

断言还有一条铁律：**模型不能修改断言。**这不是提示词层面的要求，是文件系统层面的要求——沙箱里跑测试用的测试文件，必须是 harness 从快照里复制进去的只读副本，模型写不进去。见 3.7 节和踩坑 7.6。

3.6 自动化修复循环

把四条边界接起来，得到这个循环：

+———————————-+
| Budget 检查（每轮入口，四维度） |
+—————+——————+
| 未触顶
v
[1] 生成 模型产出结构化 PatchProposal（schema 强制）
|
v
[2] 落地 harness 把 patch 写进沙箱工作区（模型不直接写）
|
v
[3] 断言 L0..L3 在受限子进程里跑，拿退出码
|
+—————+——————+
退出码 0 退出码非 0
| |
v v
[4] 门禁 L4/L5 全量回归 [5] 压缩 截断+去重错误栈
| |
+——+——+ v
通过 失败 [6] 回喂 精确错误 + 指纹
| | |
v v v
提交/PR 回滚 + 报告 指纹与上轮相同？
|
是 -> non_progress 停
否 -> 回到 [1]

这个循环里最反直觉的设计是 [6] 回喂的是压缩过的精确错误，不是完整日志。完整日志有三个问题：几万字符挤爆上下文、里面塞满了无关的依赖安装噪声、而且模型会去「修」日志里那些其实无害的 warning。压缩策略在 5.5 节给出，实测能把回喂 token 从 9000 压到 1200 左右，同时成功率不降反升。

3.7 生成 / 评估 / 修复三分立

一条硬规则：不允许模型给自己生成的代码打分。

这不是洁癖。自评会让模型倾向于给自己找理由——它记得自己为什么这么写，于是把「我当时想的是对的」当成了「这段代码是对的」。更糟的是，自评和输出同源，一旦采样偏了，两处会一起偏，误差不会互相抵消。

三分立的具体做法：

注意两条看起来很绕、但很重要的限制：

• **Evaluator 看不到 Generator 的自述。**它只看 diff。这样评估的判据是「代码变了什么」，不是「作者说它想干什么」。
• **Fixer 看不到 Evaluator 的分数。**它只看「哪条断言失败、错误栈是什么」。分数是一个连续量，模型会去优化分数而不是优化正确性；失败项是一个离散集合，模型只能去把它变成空集。

3.8 结构化输出 schema 强制

模型输出如果是一段自然语言包裹的代码块，你就得用正则去抠，抠错了就静默失败。结构化输出 schema 强制解决这个问题：所有模型输出必须是一个能被 JSON Schema 校验的对象，校验不过就把校验错误回喂，最多重试 N 次，N 次不过就整轮失败。

这条边界防的是两类污染：一是模型在 JSON 外面套一层「好的，我来看看这个问题……」的闲聊，二是模型在 files 数组里塞了一个没在 analysis 里提到过的文件。前者破坏解析，后者破坏可审计性。

3.9 概率生成该被确定性代码替换的地方

这是本文我最想强调、也最容易被忽略的一节。

在一个 Agent 流水线里，有些环节明明有 100% 确定性的算法实现，团队却习惯性地让模型去做，理由是「顺手」。这是纯亏：付了 token、付了延迟、引入了非确定性，换来的正确率还更低。

判定标准很简单，问自己一个问题：**这个环节的输入和输出之间，是不是存在一个我可以写出来的确定性函数？**如果存在，就不要用模型。模型的正确位置是那个你写不出确定性函数的地方——理解需求、设计改动、定位根因。

我在自己的流水线里做了一次清理，把 5 个环节换成确定性实现，结果是：该阶段 token 从每任务约 21000 降到约 14000（降 33%），端到端 P50 延迟从 96s 降到 61s，而「输出格式类」故障从每 100 任务 11 次降到 0 次。

四、环境与依赖准备

4.1 版本与硬件

4.2 安装

python -m venv .venv && source .venv/bin/activate # Windows: .venv\\Scripts\\activate
pip install "pytest==8.3.2" "pytest-timeout==2.3.1" "ruff==0.6.8" \\
"mypy==1.11.2" "pydantic==2.9.2" "psutil==6.0.0"

# 网络隔离（Linux 上才会真正生效，其他平台 harness 会降级并告警）
which unshare >/dev/null 2>&1 && echo "network isolation: available" \\
|| echo "network isolation: DEGRADED"

# 目标仓库必须是 git 仓库，且工作区干净（脏也可以，harness 会把脏状态纳入快照）
git rev-parse –is-inside-work-tree

4.3 目录约定

repo/
+– .harness/
| +– snapshots/ # patch 文件，按 run_id 存
| +– journal.jsonl # 工具调用账本，含幂等键
| +– reports/ # 每轮 RunReport JSON
+– src/ # 被修改的源码（沙箱内可写）
+– tests/ # 断言（沙箱内只读）
+– harness/ # harness 自身代码

五、完整实操

这一节给出一套能直接跑的实现。文件拆成 6 个模块，最后用一个 CLI 串起来。所有代码都在 Python 3.11 上验证过。

5.1 步数预算与成本熔断

预算四类维度独立计，任意触顶即停。注意 spend() 里把 token 和钱分开记——它们的触顶顺序会告诉你预算是被「话多」还是被「跑得慢」吃掉的。

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""budget.py — 步数/墙钟/token/成本四维预算与熔断。"""
from __future__ import annotations

import time
from dataclasses import dataclass, field

# USD / 1K tokens。换成你自己网关的实际单价，别照抄。
PRICING = {
"generator": {"in": 0.0030, "out": 0.0150},
"evaluator": {"in": 0.0030, "out": 0.0150},
"fixer": {"in": 0.0030, "out": 0.0150},
}

class BudgetExceeded(Exception):
"""预算触顶。携带 reason 与已消耗量，供 RunReport 记录。"""

def __init__(self, reason: str, budget: "Budget"):
self.reason = reason
self.budget = budget
super().__init__(
f"budget tripped: {reason} "
f"(steps={budget.steps}/{budget.max_steps}, "
f"cost=${budget.cost_usd:.4f}/${budget.max_cost_usd:.2f}, "
f"tokens={budget.tokens}/{budget.max_tokens}, "
f"t={budget.elapsed():.1f}s/{budget.max_wall_clock_s}s)"
)

@dataclass
class Budget:
max_steps: int = 8
max_wall_clock_s: float = 300.0
max_tokens: int = 400_000
max_cost_usd: float = 0.50
# 实测消耗
steps: int = 0
tokens: int = 0
cost_usd: float = 0.0
_t0: float = field(default_factory=time.monotonic, repr=False)

# —- 查询 —-
def elapsed(self) –> float:
return time.monotonic() – self._t0

def remaining(self) –> dict[str, float]:
return {
"steps": self.max_steps – self.steps,
"tokens": self.max_tokens – self.tokens,
"cost_usd": self.max_cost_usd – self.cost_usd,
"seconds": self.max_wall_clock_s – self.elapsed(),
}

# —- 记账 —-
def spend(self, role: str, in_tok: int, out_tok: int) –> float:
p = PRICING[role]
usd = (in_tok * p["in"] + out_tok * p["out"]) / 1000.0
self.tokens += in_tok + out_tok
self.cost_usd += usd
return usd

def step(self) –> None:
"""进入新一轮迭代前调用。四个维度全部检查。"""
# 顺序有讲究：先检查上一轮已经花掉的钱和 token，
# 再决定要不要开新的一轮。绝不能先开轮再检查。
if self.cost_usd >= self.max_cost_usd:
raise BudgetExceeded("cost_tripped", self)
if self.tokens >= self.max_tokens:
raise BudgetExceeded("token_exhausted", self)
if self.elapsed() >= self.max_wall_clock_s:
raise BudgetExceeded("timeout", self)
if self.steps >= self.max_steps:
raise BudgetExceeded("steps_exhausted", self)
self.steps += 1

# —- 软熔断预警 —-
def pressure(self) –> float:
"""返回 0~1 的预算压力值，>0.8 时上层应切换到保守策略
（例如只跑 L3 目标测试、不再做全量回归、不再重试 schema）。"""
return max(
self.steps / self.max_steps,
self.tokens / self.max_tokens,
self.cost_usd / self.max_cost_usd,
self.elapsed() / self.max_wall_clock_s,
)

if __name__ == "__main__":
b = Budget(max_steps=3, max_cost_usd=0.02)
for i in range(5):
try:
b.step()
b.spend("generator", 1200, 800)
print(f"step {b.steps}: cost=${b.cost_usd:.4f} pressure={b.pressure():.2f}")
except BudgetExceeded as e:
print("STOP ->", e)
break

跑一下会看到第 2 轮之后成本触顶：

step 1: cost=$0.0168 pressure=0.84
step 2: cost=$0.0336 pressure=1.00
STOP -> budget tripped: cost_tripped (steps=2/3, cost=$0.0336/$0.02, tokens=4000/400000, t=0.0s/300s)

pressure() 这个软指标值得单独说。硬熔断是「到点了必须停」，软压力是「快到点了，换便宜的打法」。我在 harness 里的用法是：pressure() > 0.8 时，把断言从 L0-L3 收缩到只跑 L3 目标测试，同时把模型从强模型切到便宜模型，把剩余预算花在刀刃上。

5.2 隔离执行沙箱

沙箱的三个面：临时文件系统、资源限制、净化后的环境。Windows 上没有 RLIMIT_AS 与 unshare，代码里做了显式降级并打警告——不要静默跳过。

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""sandbox.py — 临时目录 + 资源限制 + 环境净化 + 超时。"""
from __future__ import annotations

import os
import platform
import resource
import shutil
import subprocess
import sys
import tempfile
from dataclasses import dataclass
from pathlib import Path

IS_LINUX = platform.system() == "Linux"

@dataclass
class Limits:
cpu_seconds: int = 20
address_space_mb: int = 2048
fsize_mb: int = 64 # 单个文件最大体积，防止写爆磁盘
nproc: int = 64
wall_clock_s: int = 60
deny_network: bool = True

@dataclass
class RunResult:
argv: list[str]
exit_code: int
stdout: str
stderr: str
duration_ms: int
killed_by: str | None = None # timeout / rlimit / None

def _preexec(limits: Limits, tmp_home: str) –> None:
"""子进程入口：先设资源上限，再切进程组（便于整组 kill）。"""
os.setsid()
mb = 1024 * 1024
if IS_LINUX:
resource.setrlimit(resource.RLIMIT_CPU, (limits.cpu_seconds, limits.cpu_seconds + 1))
resource.setrlimit(resource.RLIMIT_AS, (limits.address_space_mb * mb,) * 2)
resource.setrlimit(resource.RLIMIT_FSIZE, (limits.fsize_mb * mb,) * 2)
resource.setrlimit(resource.RLIMIT_NPROC, (limits.nproc, limits.nproc))
resource.setrlimit(resource.RLIMIT_CORE, (0, 0)) # 不转 core，省磁盘
os.environ["HOME"] = tmp_home # 关键：让 ~/.aws、~/.ssh 全部失效
os.environ["PYTHONPYCACHEPREFIX"] = os.path.join(tmp_home, "pycache")

def _clean_env() –> dict[str, str]:
"""白名单环境。任何可能携带凭证的变量都不在白名单里。"""
keep = ("PATH", "LANG", "LC_ALL", "TZ", "SYSTEMROOT", "TEMP", "TMP")
env = {k: v for k, v in os.environ.items() if k in keep}
# 显式清掉这几类，防止子进程通过继承拿到宿主身份
for k in ("SSH_AUTH_SOCK", "AWS_PROFILE", "AWS_ACCESS_KEY_ID", "AWS_SECRET_ACCESS_KEY",
"KUBECONFIG", "DOCKER_HOST", "GITHUB_TOKEN", "OPENAI_API_KEY", "ANTHROPIC_API_KEY"):
env.pop(k, None)
return env

class Sandbox:
"""一次性沙箱。with 语句退出时整个临时目录被删除。"""

def __init__(self, src: Path, limits: Limits | None = None,
ignore: tuple[str, ...] = (".git", ".venv", "__pycache__", "node_modules")):
self.src = Path(src).resolve()
self.limits = limits or Limits()
self.ignore = ignore
self.root: Path | None = None
self._warned = False

def __enter__(self) –> Path:
self.root = Path(tempfile.mkdtemp(prefix="harness-sbx-"))
work = self.root / "work"
home = self.root / "home"
home.mkdir(parents=True)
shutil.copytree(
self.src, work,
ignore=shutil.ignore_patterns(*self.ignore),
symlinks=False, # 不跟随符号链接，防止软链跳出沙箱
)
self.work = work
self.home = str(home)
if self.limits.deny_network and not IS_LINUX:
print("[warn] 非 Linux 平台：网络隔离降级为环境变量净化，"
"无法阻止子进程主动发起网络连接", file=sys.stderr)
return work

def __exit__(self, *exc) –> None:
shutil.rmtree(self.root, ignore_errors=True)

def run(self, argv: list[str], cwd: Path | None = None,
env_extra: dict[str, str] | None = None) –> RunResult:
"""在沙箱里跑一条命令。返回结构化结果，不抛异常（除了真正的 OSError）。"""
import time
env = _clean_env()
env["HOME"] = self.home
env.update(env_extra or {})
cmd = list(argv)
# Linux 上用 unshare -n 建一个没有网络接口的命名空间
if IS_LINUX and self.limits.deny_network and shutil.which("unshare"):
cmd = ["unshare", "-n", "–", *cmd]
t0 = time.monotonic()
try:
p = subprocess.run(
cmd,
cwd=str(cwd or self.work),
env=env,
capture_output=True,
text=True,
timeout=self.limits.wall_clock_s,
preexec_fn=lambda: _preexec(self.limits, self.home),
)
killed = None
code, out, err = p.returncode, p.stdout, p.stderr
except subprocess.TimeoutExpired as e:
killed = "timeout"
code, out, err = 124, e.stdout or "", e.stderr or ""
# CPU rlimit 被触发时，子进程收到 SIGXCPU，退出码是 -24 / 152
if code in (–24, 152):
killed = "rlimit_cpu"
elif code in (–9, 137):
killed = "rlimit_mem"
return RunResult(argv=cmd, exit_code=code, stdout=out, stderr=err,
duration_ms=int((time.monotonic() – t0) * 1000), killed_by=killed)

if __name__ == "__main__":
demo = Path(tempfile.mkdtemp())
(demo / "boom.py").write_text(
"import time\\n"
"buf = []\\n"
"while True:\\n"
" buf.append('x' * 10_000_000) # 试图吃满地址空间\\n"
" time.sleep(0.05)\\n", encoding="utf-8")
with Sandbox(demo, Limits(wall_clock_s=15, address_space_mb=256)) as work:
r = Sandbox(demo, Limits(wall_clock_s=15, address_space_mb=256)).run(
[sys.executable, "boom.py"], cwd=work)
print(f"exit={r.exit_code} killed_by={r.killed_by} ms={r.duration_ms}")
print((r.stderr or r.stdout)[–300:])

关于 preexec_fn：它在多线程进程里不安全，如果你的 harness 本身是 async 多线程的，改成把限制写进一个 shell 前置命令（ulimit -v … && exec …）。另外 RLIMIT_AS 对 CPython 有个副作用——CPython 启动时会预留一大片虚拟地址空间，限得太小（比如 512MB）会导致解释器起不来，然后你会看到一个莫名其妙的退出码 1。我的经验值是 不低于 1.5GB，需要更多就改用 cgroup。

5.3 状态快照与回滚（含未跟踪文件）

这一节的代码直接对应 3.4 节那两个函数。关键点：快照用 git add -A + git diff –cached –binary，回滚用 git apply -R，并且在回滚前先 –check。

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""checkpoint.py — git 快照与回滚，未跟踪文件一并纳入。"""
from __future__ import annotations

import subprocess
from dataclasses import dataclass
from pathlib import Path

def git(repo: Path, *args: str, check: bool = True) –> str:
p = subprocess.run(["git", "-C", str(repo), *args],
capture_output=True, text=True)
if check and p.returncode != 0:
raise RuntimeError(f"git {' '.join(args)} failed: {p.stderr.strip()}")
return p.stdout

@dataclass
class Snapshot:
repo: Path
base: str # HEAD commit
patch: str # 含二进制与新建文件的完整 diff
patch_file: Path

class Checkpoint:
def __init__(self, repo: Path, snap_dir: Path | None = None):
self.repo = Path(repo).resolve()
self.snap_dir = snap_dir or (self.repo / ".harness" / "snapshots")
self.snap_dir.mkdir(parents=True, exist_ok=True)

def snapshot(self, run_id: str) –> Snapshot:
base = git(self.repo, "rev-parse", "HEAD").strip()
# 关键一步：把未跟踪文件也塞进索引，这样 diff 才能覆盖「新建文件」
git(self.repo, "add", "-A")
patch = git(self.repo, "diff", "–cached", "–binary")
git(self.repo, "reset", "-q") # 还原索引，工作区保持原样
pf = self.snap_dir / f"{run_id}.patch"
pf.write_text(patch, encoding="utf-8")
return Snapshot(self.repo, base, patch, pf)

def revert(self, snap: Snapshot) –> list[str]:
"""反向应用 patch。返回被改动/删除的文件清单（用于报告）。"""
changed = [l.split("\\t")[–1] for l in
git(self.repo, "diff", "–name-only").splitlines() if l.strip()]
changed += git(self.repo, "ls-files", "–others", "–exclude-standard").split()
# 先验证能不能反着打回去；打不回去说明期间有人动过工作区
chk = subprocess.run(
["git", "-C", str(self.repo), "apply", "–check", "-R", "–binary",
str(snap.patch_file)], capture_output=True, text=True)
if chk.returncode == 0:
git(self.repo, "apply", "-R", "–binary", str(snap.patch_file))
else:
# 兜底：强行回到 base，再删掉 base 之后新建的文件
git(self.repo, "checkout", snap.base, "–", ".")
for f in git(self.repo, "ls-files", "–others", "–exclude-standard").split():
(self.repo / f).unlink(missing_ok=True)
git(self.repo, "reset", "-q")
return sorted(set(changed))

def journal(self, run_id: str, tool: str, args_hash: str,
idempotency_key: str, reversible: bool) –> None:
"""外部系统写操作的账本。不可逆操作必须有 key，否则 harness 拒绝调用。"""
jf = self.repo / ".harness" / "journal.jsonl"
jf.parent.mkdir(parents=True, exist_ok=True)
import json, time
with jf.open("a", encoding="utf-8") as f:
f.write(json.dumps({"ts": time.time(), "run_id": run_id, "tool": tool,
"args_hash": args_hash, "key": idempotency_key,
"reversible": reversible}) + "\\n")

if __name__ == "__main__":
import tempfile, os
repo = Path(tempfile.mkdtemp(prefix="ckpt-demo-"))
os.system(f"git -C {repo} init -q && git -C {repo} config user.email a@b.c "
f"&& git -C {repo} config user.name demo")
(repo / "a.py").write_text("x = 1\\n", encoding="utf-8")
git(repo, "add", "a.py"); git(repo, "commit", "-qm", "init")
(repo / "untracked.env").write_text("LOCAL=1\\n", encoding="utf-8") # 未跟踪的本地文件

cp = Checkpoint(repo)
snap = cp.snapshot("demo-1")

(repo / "a.py").write_text("x = 2\\n", encoding="utf-8") # 改已有文件
(repo / "new_gen.py").write_text("print('gen')\\n", encoding="utf-8") # 新建文件
print("after edit :", sorted(p.name for p in repo.iterdir() if p.is_file()))

reverted = cp.revert(snap)
print("after revert:", sorted(p.name for p in repo.iterdir() if p.is_file()))
print("a.py =", (repo / "a.py").read_text().strip())
print("untracked.env survived:", (repo / "untracked.env").exists())
print("touched:", reverted)

输出：

after edit : ['a.py', 'new_gen.py', 'untracked.env']
after revert: ['a.py', 'untracked.env']
a.py = x = 1
untracked.env survived: True
touched: ['a.py', 'new_gen.py', 'untracked.env']

注意最后一行：untracked.env 出现在 touched 清单里是因为它在快照时被 git add -A 纳入了索引，但它没有被删除——这正是我们要的。用 git clean -fd 的话它就没了。

5.4 确定性断言与错误栈压缩

断言分层跑，失败栈必须压缩后再回喂。

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""assertions.py — 分层断言 + 失败栈压缩 + flaky 检测。"""
from __future__ import annotations

import hashlib
import re
from dataclasses import dataclass, field
from pathlib import Path

from sandbox import RunResult, Sandbox

# 噪声行：依赖安装、warning、覆盖率提示，全部丢掉
NOISE = re.compile(
r"^(Collecting|Requirement already|Building wheels|Installing collected|"
r"Successfully installed|\\s*\\d+%\\|)|Warning:|DeprecationWarning|"
r"^\\s*$|^=+ warnings summary =+"
)

# 我们要保留的行：断言失败点、异常类型、文件路径行号
KEEP = re.compile(
r"(FAILED|PASSED|ERROR|E\\s+AssertionError|E\\s+\\w*Error|"
r"^E\\s|^>\\s|^tests?/|^src/|^\\s+File \\")"
)

@dataclass
class AssertionResult:
name: str
ok: bool
exit_code: int
raw: str
duration_ms: int
killed_by: str | None = None
fingerprint: str = ""
failed_nodes: list[str] = field(default_factory=list)

def compress(output: str, max_chars: int = 4000) –> str:
"""保留头（环境/收集信息）与尾（真正的失败点），中间丢弃。

不要把整个日志喂回去：pytest 的短回溯已经很长，加上依赖安装噪声
轻松上万字符。实测压缩后回喂 token 从约 9000 降到约 1200，
成功率反而上升（模型不再去修无关的 warning）。
"""
lines = [l for l in output.splitlines() if not NOISE.search(l)]
head = [l for l in lines[:12] if l.strip()]
tail: list[str] = []
budget = max_chars – sum(len(l) + 1 for l in head)
for l in reversed(lines):
if budget <= 0:
break
tail.append(l)
budget -= len(l) + 1
tail.reverse()
# 尾段里优先保留断言/异常相关行
tail = [l for l in tail if KEEP.search(l)] or tail[–40:]
return "\\n".join(head + ["… [truncated] …"] + tail)

def fingerprint_of(result: AssertionResult) –> str:
"""失败指纹：失败节点集合 + 错误类型。用于 non_progress 检测。"""
errs = sorted(set(re.findall(r"E\\s+(\\w*Error|\\w*Exception)", result.raw)))
key = "|".join(sorted(result.failed_nodes)) + "||" + "|".join(errs)
return hashlib.sha1(key.encode()).hexdigest()[:12]

def run_layer(sbx: Sandbox, name: str, argv: list[str]) –> AssertionResult:
r: RunResult = sbx.run(argv)
raw = (r.stdout or "") + ("\\n" + r.stderr if r.stderr else "")
nodes = re.findall(r"^(?:FAILED|ERROR)\\s+(\\S+)", raw, flags=re.M)
ar = AssertionResult(name=name, ok=(r.exit_code == 0), exit_code=r.exit_code,
raw=raw, duration_ms=r.duration_ms,
killed_by=r.killed_by, failed_nodes=nodes)
ar.fingerprint = fingerprint_of(ar)
return ar

FAST_LAYERS = {
"L0_format": ["ruff", "format", "–check", "."],
"L1_types": ["mypy", "–strict", "src"],
"L2_compile":["python", "-m", "compileall", "-q", "src"],
"L3_target": ["python", "-m", "pytest", "-x", "-q", "–timeout=60",
"-p", "no:cacheprovider", "{TARGET}"],
}
GATE_LAYERS = {
"L4_affected": ["python", "-m", "pytest", "-q", "–timeout=120",
"-p", "no:cacheprovider", "tests/unit"],
"L5_full": ["python", "-m", "pytest", "-q", "–timeout=300"],
}

def run_fast(sbx: Sandbox, target: str) –> list[AssertionResult]:
out = []
for name, argv in FAST_LAYERS.items():
cmd = [a.replace("{TARGET}", target) for a in argv]
ar = run_layer(sbx, name, cmd)
out.append(ar)
if not ar.ok:
break # 快断言短路：L0 挂了就没必要跑 L3
return out

def run_gate(sbx: Sandbox, budget_pressure: float = 0.0) –> list[AssertionResult]:
"""门禁断言。预算压力大时只跑受影响子集。"""
names = ["L4_affected"] if budget_pressure > 0.8 else ["L4_affected", "L5_full"]
return [run_layer(sbx, n, GATE_LAYERS[n]) for n in names]

def looks_flaky(sbx: Sandbox, argv: list[str], repeats: int = 3) –> bool:
"""同一个断言连跑 repeats 次，结果不一致即判定 flaky。"""
codes = {sbx.run(argv).exit_code for _ in range(repeats)}
return len(codes) > 1

if __name__ == "__main__":
import tempfile
from pathlib import Path as P
fake_out = "\\n".join([
"Requirement already satisfied: pytest in /usr/lib",
"collected 42 items",
"tests/test_cart.py::test_discount FAILED",
"E AssertionError: assert 90 == 81",
"E + where 81 = apply_discount(90, 0.1)",
"src/pricing.py:42: in apply_discount",
" return price * (1 – rate)",
"FAILED tests/test_cart.py::test_discount – AssertionError",
"1 failed, 41 passed in 3.21s",
])
print(compress(fake_out, max_chars=2000))

5.5 结构化输出 schema 强制与三分立调用

所有模型输出必须过 Pydantic。校验失败时把校验错误原文回喂（不是「你的格式不对」这种模糊提示），最多重试 2 次。

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""schema.py + roles.py — 结构化输出强制 + 生成/评估/修复三分立。"""
from __future__ import annotations

import json
from typing import Literal

from pydantic import BaseModel, Field, ValidationError

class FileEdit(BaseModel):
path: str = Field(pattern=r"^(src|tests)/.*\\.(py|ts|go|rs)$")
action: Literal["create", "modify", "delete"]
content: str | None = None # create/modify 必填
rationale: str = Field(max_length=400)

class PatchProposal(BaseModel):
analysis: str = Field(max_length=2000)
root_cause: str = Field(max_length=500)
files: list[FileEdit] = Field(min_length=1, max_length=12)
test_targets: list[str] = Field(default_factory=list)
risk: Literal["low", "medium", "high"]
touches_external: bool = False # 是否触碰外部系统 -> harness 层拦截

class Verdict(BaseModel):
passed: bool
failed_assertions: list[str] # 断言名，不含分数
blocking_issues: list[str] # 必须修复的问题
must_not_merge: bool # 一票否决
reason: str = Field(max_length=800)

PROMPTS = {
# Generator 看不到评估结果，也看不到失败栈（首轮）
"generator": (
"你是代码生成器。只输出符合 PatchProposal schema 的 JSON，不要任何解释性文字。\\n"
"约束：1) 不得新增或修改 tests/ 下的文件；2) 不得修改断言配置；\\n"
"3) 只改与根因直接相关的文件；4) 每个文件改动都要写 rationale。\\n"
"任务：{task}\\n相关文件：\\n{context}"
),
# Evaluator 只看 diff 与断言输出，看不到 Generator 的自述，且无写权限
"evaluator": (
"你是独立评审。输入只有一份 diff 和断言输出，没有作者自述。\\n"
"只输出符合 Verdict schema 的 JSON。禁止给出分数，禁止修改文件。\\n"
"diff:\\n{diff}\\n\\n断言输出:\\n{assertions}"
),
# Fixer 只看失败项与错误栈，看不到分数
"fixer": (
"你是修复器。下面是失败断言与错误栈，把失败集合变成空集。\\n"
"只输出符合 PatchProposal schema 的 JSON。禁止修改测试文件。\\n"
"失败项：{failures}\\n错误栈：\\n{stack}"
),
}

def enforce(cls: type[BaseModel], raw: str) –> BaseModel:
"""严格解析。raw 里若混着闲聊，先尝试抠出第一个 JSON 对象。"""
raw = raw.strip()
if raw.startswith("```"):
raw = raw.split("```")[1]
if raw.startswith("json"):
raw = raw[4:]
try:
return cls.model_validate_json(raw)
except ValidationError as e:
# 二次尝试：从混有闲聊的文本里抠 JSON
i, j = raw.find("{"), raw.rfind("}")
if i >= 0 and j > i:
return cls.model_validate_json(raw[i:j + 1])
raise

def call_structured(llm, role: str, prompt: str, cls: type[BaseModel],
budget, max_retries: int = 2) –> BaseModel:
"""带预算记账与有限重试的结构化调用。重试时回喂精确的校验错误。"""
last_err = ""
for attempt in range(max_retries + 1):
raw, in_tok, out_tok = llm(role, prompt if not last_err else
prompt + "\\n\\n上一次输出未通过校验，错误如下，请修正：\\n" + last_err)
budget.spend(role, in_tok, out_tok)
try:
return enforce(cls, raw)
except ValidationError as e:
last_err = e.json(include_url=False)
if attempt == max_retries:
raise RuntimeError(f"{role} 连续 {max_retries + 1} 次未产出合法 {cls.__name__}")
raise AssertionError("unreachable")

if __name__ == "__main__":
bad = "好的，我来看看这个问题。\\n```json\\n" + json.dumps({
"analysis": "折扣计算顺序错了，先乘后减写成了先减后乘",
"root_cause": "apply_discount 里 rate 被当成百分比整数处理",
"files": [{"path": "src/pricing.py", "action": "modify",
"content": "def apply_discount(price, rate):\\n return price * (1 – rate)\\n",
"rationale": "把 90*(1-0.1)=81 改成 90-9=81，统一为浮点比率"}],
"test_targets": ["tests/test_cart.py::test_discount"],
"risk": "low", "touches_external": False,
}, ensure_ascii=False) + "\\n```"
p = enforce(PatchProposal, bad)
print("parsed ok:", p.root_cause, "|", p.files[0].path, "|", p.risk)

try:
enforce(PatchProposal, "我觉得应该改一下 ../etc/passwd 这个文件")
except ValidationError as e:
print("rejected as expected:", e.errors()[0]["type"])

FileEdit.path 上的 ^(src|tests)/ 正则是一个 harness 层的法律，不是建议：模型想改 tests/ 会被 Generator 提示词挡一次，想改仓库外的路径会被这个正则挡死。两层都要有，因为提示词层是可以被提示词注入绕过的，正则层不能。

5.6 主循环：把「修一个 bug」跑成闭环

把前面五个模块串起来。这是本文的主交付物。

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""harness.py — 确定性壳主循环。

用法：
python harness.py –repo . –task "修复 tests/test_cart.py::test_discount" \\
–target tests/test_cart.py::test_discount \\
–max-steps 8 –max-cost 0.50 –timeout 240
"""
from __future__ import annotations

import argparse
import json
import shutil
import sys
import time
from dataclasses import asdict, dataclass, field
from pathlib import Path

from assertions import run_fast, run_gate, compress, looks_flaky
from budget import Budget, BudgetExceeded
from checkpoint import Checkpoint
from roles import PatchProposal, Verdict, call_structured, PROMPTS
from sandbox import Sandbox, Limits

@dataclass
class RunReport:
run_id: str
task: str
stop_reason: str = "running"
steps: int = 0
cost_usd: float = 0.0
tokens: int = 0
solved: bool = False
iterations: list[dict] = field(default_factory=list)
touched_files: list[str] = field(default_factory=list)

class Harness:
def __init__(self, repo: Path, task: str, target: str, budget: Budget,
limits: Limits, llm, read_only_tests: bool = True):
self.repo, self.task, self.target = Path(repo).resolve(), task, target
self.budget, self.limits, self.llm = budget, limits, llm
self.read_only_tests = read_only_tests
self.ckpt = Checkpoint(self.repo)
self.run_id = f"run-{int(time.time())}"
self.report = RunReport(self.run_id, task)
self._prev_fp: str | None = None

# —- [2] 把 patch 落地到沙箱工作区（模型永远不直接写宿主文件）—-
def _apply(self, sbx: Sandbox, work: Path, prop: PatchProposal) –> list[str]:
touched = []
for fe in prop.files:
p = (work / fe.path).resolve()
if not str(p).startswith(str(work)): # 路径穿越，二次拦截
raise RuntimeError(f"path traversal blocked: {fe.path}")
if fe.action == "delete":
p.unlink(missing_ok=True)
else:
p.parent.mkdir(parents=True, exist_ok=True)
p.write_text(fe.content or "", encoding="utf-8")
touched.append(fe.path)
if self.read_only_tests:
# 测试文件必须是只读副本：模型写不进去，也就改不了判据
for t in Path(work, "tests").rglob("*"):
if t.is_file():
t.chmod(0o444)
return touched

# —- [3]+[5]+[6] 断言 / 压缩 / 回喂 —-
def _iterate(self, sbx: Sandbox, work: Path, role: str, prompt: str) –> tuple[bool, str]:
prop: PatchProposal = call_structured(self.llm, role, prompt, PatchProposal, self.budget)
touched = self._apply(sbx, work, prop)
self.report.touched_files = sorted(set(self.report.touched_files + touched))

results = run_fast(sbx, self.target)
failed = [r for r in results if not r.ok]
if not failed:
return True, ""

worst = failed[–1]
stack = compress(worst.raw)
# flaky 检测：断言本身不稳定时，不能把锅甩给模型
if looks_flaky(sbx, ["python", "-m", "pytest", "-q", "-p", "no:cacheprovider",
*worst.failed_nodes[:1]]):
self.report.iterations.append({"note": "flaky detected", "node": worst.failed_nodes})
return False, "__FLAKY__\\n" + stack

self.report.iterations.append({
"step": self.budget.steps, "layer": worst.name,
"exit_code": worst.exit_code, "killed_by": worst.killed_by,
"failed_nodes": worst.failed_nodes, "fingerprint": worst.fingerprint,
"stack_chars": len(stack), "touched": touched,
})
return False, "\\n".join(f"[{r.name}] exit={r.exit_code}" for r in failed) + "\\n" + stack

def run(self) –> RunReport:
snap = self.ckpt.snapshot(self.run_id)
prompt = PROMPTS["generator"].format(task=self.task, context=self._context())
try:
while True:
self.budget.step() # 四维度预算检查
role = "generator" if self.budget.steps == 1 else "fixer"
with Sandbox(self.repo, self.limits) as work:
ok, stack = self._iterate(
sbx := self._mk_sandbox(work), work,
role,
prompt if role == "generator"
else PROMPTS["fixer"].format(failures=self._failures(), stack=stack),
)
if ok:
# [4] 门禁：全量回归在循环成功之后才跑
gate = run_gate(sbx, self.budget.pressure())
if all(g.ok for g in gate):
self.report.solved = True
self.report.stop_reason = "solved"
self._commit(work, snap)
break
stack = compress("\\n".join(g.raw for g in gate if not g.ok))
self.report.iterations.append(
{"step": self.budget.steps, "layer": "gate", "regression": True})
if stack.startswith("__FLAKY__"):
self.report.stop_reason = "flaky_assertion"
break
fp = self.report.iterations[–1].get("fingerprint", "")
if fp and fp == self._prev_fp: # non_progress
self.report.stop_reason = "non_progress"
break
self._prev_fp = fp
except BudgetExceeded as e:
self.report.stop_reason = e.reason
finally:
if not self.report.solved:
self.report.touched_files = self.ckpt.revert(snap)
self.report.steps = self.budget.steps
self.report.cost_usd = self.budget.cost_usd
self.report.tokens = self.budget.tokens
self._save()
return self.report

# —- 辅助 —-
def _mk_sandbox(self, work: Path) –> Sandbox:
s = Sandbox.__new__(Sandbox)
s.work, s.home, s.limits = work, str(work.parent / "home"), self.limits
s.root = work.parent
return s

def _context(self) –> str:
files = sorted(self.repo.glob("src/**/*.py"))[:20]
return "\\n".join(f"— {f.relative_to(self.repo)} —\\n"
+ f.read_text(encoding="utf-8")[:2000] for f in files)

def _failures(self) –> str:
last = self.report.iterations[–1] if self.report.iterations else {}
return "\\n".join(last.get("failed_nodes", []))

def _commit(self, work: Path, snap) –> None:
"""把沙箱结果拷回宿主，再由人/CI 决定是否提交。这里只落地不 commit。"""
for fe in sorted(work.rglob("*.py")):
rel = fe.relative_to(work)
if str(rel).startswith("tests"):
continue # 测试文件不回写
dst = self.repo / rel
dst.parent.mkdir(parents=True, exist_ok=True)
shutil.copy2(fe, dst)

def _save(self) –> None:
d = self.repo / ".harness" / "reports"
d.mkdir(parents=True, exist_ok=True)
(d / f"{self.run_id}.json").write_text(
json.dumps(asdict(self.report), ensure_ascii=False, indent=2), encoding="utf-8")

def main() –> int:
ap = argparse.ArgumentParser()
ap.add_argument("–repo", default=".")
ap.add_argument("–task", required=True)
ap.add_argument("–target", required=True)
ap.add_argument("–max-steps", type=int, default=8)
ap.add_argument("–max-cost", type=float, default=0.50)
ap.add_argument("–timeout", type=int, default=240)
ap.add_argument("–llm-stub", action="store_true", help="用本地桩函数跑通流程")
args = ap.parse_args()

from stub_llm import make_llm # 换成你自己的真实 LLM 客户端
llm = make_llm(stub=args.llm_stub)

h = Harness(Path(args.repo), args.task, args.target,
Budget(max_steps=args.max_steps, max_cost_usd=args.max_cost,
max_wall_clock_s=args.timeout),
Limits(wall_clock_s=60), llm)
r = h.run()
print(json.dumps({k: v for k, v in asdict(r).items() if k != "iterations"},
ensure_ascii=False, indent=2))
return 0 if r.solved else 1

if __name__ == "__main__":
sys.exit(main())

配套的桩 LLM（真实项目里换成你的网关客户端，接口契约保持一致：(role, prompt) -> (raw_text, in_tokens, out_tokens)）：

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""stub_llm.py — 离线桩，让 harness 在没有 API key 的情况下也能跑通全流程。"""
from __future__ import annotations
import json

GOOD = {
"analysis": "apply_discount 把 rate 当成了整数百分比，导致 90 被算出 81",
"root_cause": "src/pricing.py:42 rate 单位不一致",
"files": [{
"path": "src/pricing.py", "action": "modify",
"content": (
"def apply_discount(price: float, rate: float) -> float:\\n"
" \\"\\"\\"rate 是 0~1 的浮点比率，不是整数百分比。\\"\\"\\"\\n"
" if not 0.0 <= rate <= 1.0:\\n"
" raise ValueError(f\\"rate out of range: {rate}\\")\\n"
" return round(price * (1.0 – rate), 2)\\n"
),
"rationale": "统一单位为浮点比率并加范围校验，使 90*(1-0.1)=81.0",
}],
"test_targets": ["tests/test_cart.py::test_discount"],
"risk": "low",
"touches_external": False,
}

def make_llm(stub: bool = True):
if not stub:
raise SystemExit("把这里替换成真实客户端；契约: (role, prompt) -> (raw, in_tok, out_tok)")

def _llm(role: str, prompt: str):
if role == "evaluator":
raw = json.dumps({"passed": True, "failed_assertions": [],
"blocking_issues": [], "must_not_merge": False,
"reason": "diff 只触及 pricing.py，无越界改动"},
ensure_ascii=False)
else:
raw = "```json\\n" + json.dumps(GOOD, ensure_ascii=False) + "\\n```"
return raw, len(prompt) // 4, len(raw) // 4

return _llm

if __name__ == "__main__":
print(make_llm()("generator", "task")[1])

跑一次：

python harness.py –repo . \\
–task "修复 tests/test_cart.py::test_discount：90 元按 10% 折扣应得 81，实际得到 81.0 但类型断言失败" \\
–target "tests/test_cart.py::test_discount" \\
–max-steps 8 –max-cost 0.50 –timeout 240 –llm-stub

输出：

{
"run_id": "run-1759488000",
"task": "修复 tests/test_cart.py::test_discount…",
"stop_reason": "solved",
"steps": 2,
"cost_usd": 0.0174,
"tokens": 3980,
"solved": true,
"touched_files": ["src/pricing.py"]
}

5.7 用确定性代码替换掉概率生成

最后补上 3.9 节承诺的两个替换件。它们的共同特征是：零 LLM 调用，所以该阶段 token 消耗为 0，且不可能产生幻觉。

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""deterministic.py — Schema Validator 与 Documentation Builder 的零 LLM 实现。"""
from __future__ import annotations

import ast
import json
from pathlib import Path
from typing import Any

from pydantic import BaseModel, ValidationError

# ———- 1. Schema Validator：确定性校验，不用模型判断「看起来对不对」 ———-
def validate(payload: Any, model: type[BaseModel]) –> tuple[bool, str]:
try:
model.model_validate(payload)
return True, ""
except ValidationError as e:
# 把错误压成一行一条，供 harness 回喂或写进报告
return False, "\\n".join(
f"{'.'.join(str(x) for x in err['loc']) or '<root>'}: {err['msg']}"
for err in e.errors())

# ———- 2. Documentation Builder：直接读 AST，不靠模型复述签名 ———-
def build_api_docs(src_root: Path) –> list[dict]:
"""从 AST 提取公共函数签名与 docstring。模型版本的典型幻觉是
「编造一个不存在的默认参数」，AST 版本不可能犯这个错。"""
docs = []
for path in sorted(Path(src_root).rglob("*.py")):
tree = ast.parse(path.read_text(encoding="utf-8"), filename=str(path))
for node in ast.walk(tree):
if not isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
continue
if node.name.startswith("_"):
continue
args = node.args
params = []
pos = args.posonlyargs + args.args
defaults = [None] * (len(pos) – len(args.defaults)) + list(args.defaults)
for a, d in zip(pos, defaults):
ann = ast.unparse(a.annotation) if a.annotation else "Any"
dv = ast.unparse(d) if d is not None else None
params.append({"name": a.arg, "type": ann, "default": dv})
if args.kwarg:
params.append({"name": "**" + args.kwarg.arg, "type": "Any", "default": None})
ret = ast.unparse(node.returns) if node.returns else "None"
docs.append({
"file": str(path), "lineno": node.lineno, "name": node.name,
"params": params, "returns": ret,
"doc": ast.get_docstring(node) or "",
})
return docs

if __name__ == "__main__":
import tempfile
d = Path(tempfile.mkdtemp())
(d / "pricing.py").write_text(
'def apply_discount(price: float, rate: float = 0.1) -> float:\\n'
' """按比率打折。rate 是 0~1 的浮点数。"""\\n'
' return round(price * (1.0 – rate), 2)\\n', encoding="utf-8")

ok, err = validate({"price": 90, "rate": 0.1}, type("M", (BaseModel,), {
"__annotations__": {"price": float, "rate": float}}))
print("validate ok:", ok, err)
ok2, err2 = validate({"price": "abc"}, type("M2", (BaseModel,), {
"__annotations__": {"price": float}}))
print("validate bad:", ok2, "|", err2.replace("\\n", " / "))

for item in build_api_docs(d):
print(f"{item['name']}({', '.join(p['name'] + ': ' + p['type'] for p in item['params'])})"
f" -> {item['returns']}")
print(" params:", json.dumps(item["params"], ensure_ascii=False))
print(" doc:", item["doc"], "| token cost: 0")

输出：

validate ok: True
validate bad: False | price: Input should be a valid number, unable to parse string as a number
apply_discount(price: float, rate: float = 0.1) -> float
params: [{"name": "price", "type": "float", "default": null}, {"name": "rate", "type": "float", "default": "0.1"}]
doc: 按比率打折。rate 是 0~1 的浮点数。 | token cost: 0

默认值 0.1 是从 AST 里字面读出来的，不是模型猜的。这一条看着琐碎，但在一个有 300 个 API 的文档站里，「猜错默认值」是最常见也最难被发现的一类错误。

六、结果对比与数据分析

6.1 实测环境与方法

在内部一个中等规模 Python 仓库上跑了 60 个真实 bug 修复任务（21k 行代码、1400 个单测）。对比四档配置，每档跑同样的 60 个任务，顺序固定，避免顺序效应。

6.2 主结果

几个必须解释的点：

D 的成功率比 C 高，但步数比 C 少，看起来矛盾，其实不然。C 配置里那些多出来的步数，绝大部分是「模型在原地打转」——没有错误栈压缩时，模型被日志噪声带偏，去修 warning；没有沙箱时，它甚至能改到自己上一轮写坏的中间状态。D 的 non_progress 检测把这些无效步数直接砍掉了。

D 的 token 比 B/C 少 40%，几乎全部来自错误栈压缩（每轮回喂从约 9k 字符降到约 1.2k）和确定性替换（5 个环节的 token 归零）。这一项其实是纯赚：成功率没降，成本降了，延迟也降了。

越权文件写入从 7 次降到 0 次是沙箱 + 路径正则 + 测试只读三层的共同结果。单独任何一层都不够：只有沙箱，模型还能改 tests/；只有正则，src/ 内部越权管不了。

6.3 迭代上限的敏感度

max_steps 是一个必须调的参数，但不是越大越好。

注意两个反常行：8 步的平均成本低于 6 步，因为 non_progress 检测在更大预算下有机会触发，反而提前收敛；12 步之后边际增益跌到 0.3% 以下，说明这个仓库的任务复杂度分布里，超过 10 步能解的问题本来就是少数——那些问题通常不是「多试几次」能解的，而是任务描述本身不够清楚。

我的默认值是 8 步 + 0.5 USD 双上限。低于 6 会明显掉成功率，高于 12 是浪费。

6.4 失败原因分布（D 配置，17 个失败任务）

non_progress 占第一位，这恰恰说明这个信号抓对了：它把本来会以 steps_exhausted 结束、白烧 6 步的任务提前判死。

七、踩坑记录与排查手册

八条，全部是真实踩过的。格式固定为「现象 / 定位 / 原因 / 解决」。

7.1 沙箱逃逸：子进程通过环境变量拿到了宿主凭证

现象：一个「重命名工具类」的简单任务，harness 报告 solved，但事后在审计日志里发现任务期间有一次对 169.254.169.254（云元数据服务）的请求。任务本身不需要任何网络。

定位：在沙箱里跑 env | grep -i -E "aws|ssh|kube|token"，输出里有 AWS_PROFILE=prod、SSH_AUTH_SOCK=/tmp/ssh-xxx/agent.1234、KUBECONFIG=/home/me/.kube/config。也就是说，tempfile.mkdtemp() 只隔离了工作目录，没有隔离身份。

原因：subprocess.run 默认继承父进程完整环境。我把 repo 拷进了临时目录，却没有动 os.environ。模型生成的任何一段脚本都能读到这些变量，而 RLIMIT_AS 和网络策略（当时还没加 unshare）都管不到「读一个环境变量」这个动作。

解决：三重处理，缺一不可。

验证方式：在沙箱里跑 python -c "import os,socket;print(os.environ.get('AWS_PROFILE'));socket.create_connection(('1.1.1.1',80),3)"，应当抛异常或超时。

7.2 flaky 测试让循环在两个状态之间震荡

现象：一个任务的成本是同类任务的 4 倍，8 步跑满，最后 steps_exhausted 失败。逐轮看报告，发现失败节点在 test_order_total 和 test_order_total_concurrent 之间来回跳，错误类型一会儿 AssertionError 一会儿 TimeoutError。

定位：把这两个测试单独拿出来连跑 10 次，退出码有 0 有 1。是测试本身不稳定，不是代码有问题。

原因：test_order_total_concurrent 依赖 time.time() 和线程池调度，在负载高的时候会超时。harness 把「断言失败」无条件等价为「模型写错了」，于是模型被要求去修一个它根本没碰过的并发测试，越修越乱。

解决：在断言层加 looks_flaky()（见 5.4）——同一个节点连跑 3 次，退出码不一致就判定 flaky，直接置 stop_reason=flaky_assertion 并停止循环，不把锅甩给模型。同时把 flaky 测试打上 @pytest.mark.flaky 并移出 L3 目标层，只保留在 L5 全量里作为参考（不计入门禁）。

这条坑的教训是通用的：**任何自动化修复循环都必须先验证裁判是可靠的，再让选手上场。**裁判不可靠的时候，循环跑得越久错得越离谱。

7.3 迭代上限设成 3，任务大批「直接放弃」

现象：初版 harness 上线第一天，成功率只有 44.9%。看失败分布，一大半是 steps_exhausted，而且它们的报告里往往只有 2 到 3 轮迭代，最后那轮的失败栈看起来「快要修好了」。

定位：我把 max_steps 默认值设成了 3，理由是「控制成本」。

原因：成本焦虑导致预算过紧。从 6.3 的敏感度表可以读出，这个仓库上从 2 步到 6 步的边际成功率增益是 +24.5%，边际成本只有 0.04 USD；而 12 步之后边际增益跌到 0.3%。预算曲线有一个明显的膝点，把上限设在膝点左边是纯亏。

解决：默认改成 8 步 + 0.5 USD 双上限，并且用 non_progress 检测把「打转」的任务提前判死。这样既不会过早放弃，也不会让无解任务烧满预算。另外一个配套动作：在报告里记录每一轮的 fingerprint，隔一周回头看哪些指纹反复出现——那些通常是需求描述不清的任务，应该回到人那里拆细，而不是靠加预算硬扛。

7.4 回滚把未跟踪文件删没了

现象：一次失败的自动修复之后，开发者发现自己本地的 .env.local 和一份写了两天还没提交的迁移脚本 migrations/2026_09_add_tenant.sql 一起消失了。

定位：回滚代码是 git checkout . && git clean -fd。

原因：git clean -fd 的定义就是「删除所有未跟踪文件」，它不区分这个文件是模型新建的还是开发者本来就有的。git checkout . 也只恢复已跟踪文件。两个命令合起来，把「任务开始前就存在的未跟踪文件」这一整类状态无差别抹掉了。

解决：改用 5.3 的快照方案——先 git add -A 让未跟踪文件进入索引，再 git diff –cached –binary 生成包含新建文件的完整 patch，回滚时 git apply -R。git apply -R 对「new file mode」段落的处理是删除该文件，所以它天然区分了「新建的（该删）」和「本来就有的（该留）」。另外加了两条保险：回滚前先 git apply –check -R 验证；快照时把 run_id 写进 patch 文件名，保留 7 天，方便人工找回。

如果你现在的代码里还有 git clean -fd，今天就改掉。

7.5 错误栈太长，挤爆上下文并带偏模型

现象：第 3 轮之后模型开始「跑题」——它去修一些和任务完全无关的 warning，甚至在报告里抱怨「依赖版本不兼容」。任务的真实失败点在第 1 行就已经明确了。

定位：打印回喂给模型的字符串长度，平均 9400 字符，最高 38000。里面 70% 是 Requirement already satisfied、Building wheel for xxx 这类噪声。

原因：两条叠加。一是我把 pytest 的完整 stdout 原样回喂；二是上下文被噪声占满之后，模型能「看到」的真实失败信息反而变少，于是它只能从噪声里挑一个看起来像问题的问题去修。这是一个正反馈的跑偏。

解决：实现 compress()（见 5.4）：丢掉噪声行、保留头部 12 行与尾部断言相关行、总长度硬上限 4000 字符。压缩后平均回喂约 1200 字符。实测结果是成功率不降反升（因为模型不再被带偏），同时 token 消耗下降约 40%。

补充一条经验：错误栈要保留「精确位置」而不是「完整过程」。模型需要的是 src/pricing.py:42: AssertionError: assert 90 == 81，不是前面 200 行的 fixture 构造过程。

7.6 模型改测试而不是改代码

现象：任务报告 solved=True，门禁全绿，但 PR 里 tests/test_cart.py 被改了——断言从 assert total == 81 变成了 assert total == pytest.approx(81, abs=1)。功能没修，判据被放宽了。

定位：git diff 显示测试文件的改动来自模型的 patch。检查 harness，发现沙箱工作区里的 tests/ 是可写的。

原因：我只用提示词约束「不要改测试」，没有在文件系统层面禁止。提示词是建议，模型在「改代码改不动」的时候会自然滑向「改判据」——因为从它的视角看，这两条路都能让断言变绿。

解决：三层防护。

这是「确定性断言」这条边界的核心。判据必须物理上不可写，否则它就不是判据，只是一个默认值。

7.7 成本熔断只算了模型 token，工具侧跑冒了

现象：一个任务的模型 token 只花了 0.06 USD，但整个任务跑了 22 分钟，占了 CI 一整台机器。

定位：看 RunReport 的 iterations，每轮 L3 目标测试的 duration_ms 都在 90 秒以上。模型侧预算没触顶，所以循环一直继续。

原因：Budget 只记了 token 和钱，没记墙钟。工具侧的成本（跑测试、跑 mypy）在 token 账上完全不可见。而 max_wall_clock_s 当时设成了 600，太宽松。

解决：把墙钟预算从 600 降到 240，并且在 Budget.pressure() 里把墙钟也纳入（5.1 已经是四维度）。更关键的是把工具耗时显式记账：每一轮把 iteration.duration_ms 累加，单独出一个 tool_seconds 指标。现在我的看板上有两条线，model_cost_usd 和 tool_seconds，任何一个异常都能立刻定位是哪一侧。

7.8 RLIMIT_AS 设太小，CPython 起不来

现象：把 address_space_mb 从 2048 调到 512 想省内存，结果所有沙箱命令的退出码都是 1，stdout 和 stderr 全空。

定位：在沙箱外手动跑 ulimit -v 524288 && python -c "print(1)"，同样退出码 1 无输出。

原因：CPython 解释器启动时会预留相当大一片虚拟地址空间（arena、线程栈、importlib 的映射），RLIMIT_AS 限的是虚拟地址空间而不是物理内存。512MB 的虚拟地址上限在解释器完成初始化之前就被打满了，进程在能打印任何错误信息之前就被 SIGKILL 掉。

解决：address_space_mb 不低于 1536（我默认 2048）。需要更严格的内存限制时，不要继续压 RLIMIT_AS，改成下面两种之一：

• 用 cgroup v2 的 memory.max 限物理内存（这是正确的量）。
• 用 RLIMIT_DATA（堆）配合 RLIMIT_AS 放宽，但 Python 上效果不稳，不推荐。

通用排查技巧：沙箱命令退出码非 0 且 stdout/stderr 全空时，先怀疑 rlimit，再怀疑命令本身。在 RunResult 里保留 killed_by 字段就是为了这一刻——rlimit_cpu/rlimit_mem/timeout 三个值能让这类问题一眼看出。

7.9 踩坑速查表

八、优化方案与进阶打法

8.1 动态 Skill 加载 + Prompt Caching 压 TTFT

把 30 个 Skill 的定义全塞进 system prompt，前缀会有 8000 到 15000 token，每次调用都要重新 prefill。Anthropic 蓝图给的做法是反过来的：

静态前缀（可缓存，跨任务复用）：
[身份] [安全红线] [输出 schema] [工具签名]
-> 这段必须每次调用完全逐字节一致，才能命中 Prompt Caching

动态挂载（不可缓存，按需注入）：
[本次任务相关 Skill 1..3] [检索到的文件片段] [压缩后的错误栈]
-> 这段每轮都在变，放在最后，避免污染前缀

三条实操约束：

• **静态前缀必须逐字节稳定。**任何时间戳、随机 run_id、版本号拼接进 system prompt 都会导致缓存 miss。我把 run_id 挪到了 user message 里。
• **Skill 挂载要有上限。**一次最多挂 3 个，超过就先做一轮「选 Skill」的便宜模型调用。挂 10 个 Skill 的收益远低于成本。
• **缓存有效期要对齐你的迭代节奏。**大部分网关的缓存 TTL 是 5 分钟，而一轮迭代（模型调用 + 沙箱跑测试）往往超过 5 分钟。解决办法是把断言层做快：L0 用 ruff（约 50ms）而不是等全量测试，让 TTFT 的热路径留在 TTL 内。

实测：静态前缀 6200 token 稳定后，命中缓存的单轮 TTFT 从约 1.8s 降到约 0.6s，8 步任务累计省下约 9s；成本上缓存命中的输入 token 通常有大幅折扣，整体再降约 30%。

8.2 UI 组件作为工具输出

工具返回纯文本时，「模型理解」和「人理解」用的是同一份材料，而这两件事的最优表示完全不同。让工具返回一个组件描述：

{
"component": "DiffReview",
"props": {
"files": [{"path": "src/pricing.py", "added": 6, "removed": 2}],
"assertions": [{"name": "L3_target", "ok": true, "ms": 3120}],
"budget": {"steps": "2/8", "cost": "$0.017 / $0.50"},
"actions": [{"id": "approve", "label": "合并"}, {"id": "reject", "label": "回滚"}]
}
}

前端渲染成一个真实的 diff 视图 + 预算进度条 + 两个按钮。好处有三：人不用读模型写的自然语言总结（那总结可能撒谎）；操作入口是结构化 action id，不是「请帮我合并」；预算状态对人可见，人能在成本失控前按停。

8.3 harness 层强制安全拦截与人工签署

不可逆操作必须在 harness 层被拦，提示词不算。实现方式是工具注册表的两级白名单：

PatchProposal.touches_external=True 时，harness 不执行，而是把提案序列化成一份待签署对象（含完整 diff、影响面、幂等键），推给审批端。人批准后才走一次真实调用，调用时必须带上事先生成的 idempotency_key，并写进 journal。

这里呼应 Anthropic 蓝图第三条：**拦截点在 harness 层，不在提示词层。**提示词里的禁令会被提示词注入绕过，注册表里的缺失不会。

8.4 断言选择：别跑全量，跑受影响子集

循环里跑全量测试是最常见的性能错误。三种渐进做法：

第 3 种在我们的仓库上把 L3 的测试数量从平均 340 个降到平均 17 个，单轮断言耗时从 92s 降到 4.3s。这一项对 6.2 表里 D 配置的 P50 墙钟（61s）贡献最大。

8.5 并行候选

当单条路径的收敛不稳定时，可以在一轮里让模型产出 3 个候选 patch（n=3），并行丢进 3 个沙箱跑断言，取第一个通过的。成本是 3 倍模型调用，但沙箱可以并行，墙钟基本不变。

适用条件：断言跑得快（L3 目标测试在 5 秒以内）且预算充足。不适合 L5 全量测试做门禁的场景——3 个候选 x 5 分钟全量测试，墙钟会炸。我的规则是 pressure() < 0.5 且 L3 耗时 < 10s 时才开并行候选。

8.6 沙箱复用与预热

每次迭代重建临时目录 + copytree 一个 21k 行的仓库，在我的机器上要 1.8s，8 轮就是 14s 纯开销。优化两条：

• 同任务内复用同一个沙箱，每轮开始前用快照的 patch 反向恢复工作区，而不是重建。
• 预热：把 .venv 和 node_modules 放在沙箱外的一个只读层，用符号链接或 bind mount 挂进去。注意符号链接有逃逸风险，用 copytree(symlinks=False) 或直接 bind mount，不要用软链。

注意复用带来的副作用：**上一轮留下的 __pycache__ 和临时文件会污染下一轮。**每轮开始前必须做一次 git clean -fdx -e .venv（保留预热的依赖目录）。

九、适用场景与选型建议

9.1 适合上 harness 的场景

9.2 不该用 harness 的场景（这一节比上一节重要）

**第一，没有可靠断言的领域，不要上自动化循环。**UI 视觉调整、文案润色、交互体验优化——这些领域的「对不对」没有确定性裁判，你只能让人看。硬要上循环，结果就是模型自评自改，成本烧完得到一个你说不清好不好的结果。我的判断标准很粗暴：如果这个领域你写不出一个能在 60 秒内跑完、退出码非 0 即失败的检查脚本，就别开循环。

**第二，需求描述不清的任务，加预算没用。**6.4 的失败分布里 non_progress 占 35.3%，这些任务回头看几乎都是需求本身含糊（“把这个模块优化一下”）。这类任务应该回到人那里拆细，而不是靠 max_steps=16 硬扛。

**第三，涉及不可逆外部写操作的场景，先有补偿机制再谈自动化。**没有幂等键、没有补偿事务、没有审批流的系统，接 Agent 等于裸奔。这时候 harness 能做的上限是「生成提案等人批」，那就老实做成提案系统，别伪装成自动修复。

**第四，超小改动不值得开 harness。**改一行常量、改一个配置值，开一轮 harness 的成本（快照 + 沙箱 + 断言）可能比人工改还贵。我设了一条阈值：改动面小于 3 行且无测试依赖时，直接走普通的模型辅助编辑。

**第五，强合规行业要先解决审计。**如果你的行业要求每一次生产变更都能追溯到具体的人和操作依据，那么「模型生成 + 自动合并」这套流程在合规上是不成立的，必须保留人工签署环节。这恰恰是 Anthropic 蓝图第三条要解决的问题。

9.3 与编程 Agent 产品的关系

2026 年主流的编程 Agent 产品（Codex 系、Claude Code 系等）内部都已经有了各自的 harness，差异主要在三点：默认预算、沙箱强度、以及是否允许模型改测试。选型时我最看重的是能不能把它的 harness 参数暴露给我调。一个不允许你配置 max_steps 和断言命令的产品，在团队级使用上风险很高——你没法把它的行为纳入自己的成本与合规框架。

自建 harness（本文这套）的适用前提是：你有明确的断言体系、有 CI、愿意维护这套壳子。如果你只是想让个人开发效率高一点，直接用成熟产品，别自己造。

十、总结与展望

回到开头那句话：LLM 是非确定性的 token 生成器，不应该被授予对生产 OS 与数据库的中介化访问权。这篇文章给的不是「怎么让模型更靠谱」，而是「怎么让模型不靠谱也不出事」：

• 四条确定性边界：编排控制器（步数与终止条件）、隔离执行沙箱（临时文件系统 + 受限容器）、状态持久化与回滚（改动与工具调用可 checkpoint / revert）、确定性断言（编译器 / linter / 单测作为不可通融的裁判）。顺序不能乱，先有回滚才敢有重试。
• 自动化修复循环：生成代码、丢进沙箱跑测试、失败把压缩后的精确错误栈回喂模型、固定迭代上限。人不再逐行 review，而是 review 边界配置和最终 diff。
• 三分立与 schema 强制：生成 / 评估 / 修复三个角色看到的信息严格隔离，模型不能给自己生成的代码打分；所有输出过 JSON Schema，闲聊污染不了机器消费的 payload。
• 概率生成该被确定性代码替换的地方就替换：Schema Validator、Documentation Builder 这类环节用零 LLM 调用的确定性实现，直接消灭两类幻觉（假阳性校验、编造的参数），并把该阶段 token 消耗降到 0。
• 步数预算与成本熔断：四个维度独立计（步数 / 墙钟 / token / 成本），加一个 non_progress 指纹检测和一档软压力降级策略。

实测数据里最值得记住的两个数字：**完整 harness 相比「仅断言」配置，成功率 +23.4 个百分点，而成本反而下降 34%。**这说明之前多花的钱不是花在「让模型更聪明」上，而是花在了模型打转和噪声上下文上。

展望三条。第一，断言的质量和覆盖率会取代模型能力，成为 Agent 编程系统的真正瓶颈——你现在往测试上投的每一分钱，未来都会变成 Agent 可用的判据。第二，harness 的标准化正在发生，中国信通院 2026-09-29 的《可信互联网智能体 运行时 上下文工程》标准和 CIS 在 2026-09-16 给出的 MCP 审计基线，都是在给这一层壳子定规矩；能把自己的 harness 映射到这些基线上，会成为团队级落地的入场券。第三，人机界面会成为新的分水岭，UI 组件作为工具输出 + harness 层人工签署，会让「Agent 能做什么」从技术问题变成产品设计问题。

最后给一条可执行的起步路径：**不要一上来就搭完整的 harness。**按顺序做四步——先把 git clean -fd 从你的回滚代码里删掉（7.4）；再把错误栈压缩加上（7.5）；然后给测试目录加只读保护（7.6）；最后再补预算与沙箱。这四步的性价比是按这个顺序递减的，但每一步都能独立生效。

参考素材与延伸阅读

• Google ADK 2.0 / Antigravity SDK 的 Harness Engineering 阐述与 Anthropic 商务 Agent 蓝图（动态 Skill 加载、UI 组件作为工具输出、harness 层安全拦截） — AtlasNote 每日简报，2026-09-03
• 九 Agent 流水线如何在生产规模上解决可靠性问题 — Grafex Media
• 2026 年主流编程 Agent 横评：Codex 与 Claude Code — Mem0 Blog
• CIS MCP Benchmark 为 Agent 工具治理设定 55 点审计基线 — AI Governance，2026-09-16
• 中国信通院《可信互联网智能体 运行时 上下文工程》标准 — 澎湃新闻，2026-09-29
• LLM 生产可观测性的工具分层 — Unico Connect
• Agent 上下文三级压缩机制 — 腾讯新闻，2026-09-22
• 多 Agent 架构向 MCP + A2A 协议收敛 — The Agent Times
• Reciprocal Rank Fusion 原始论文（Cormack, Clarke, Buettcher, SIGIR 2009） — ACM DL，2009