第 27 章：迁移现有 Benchmark 与开发 Adapter
团队已经维护一套代码生成 Benchmark：上游给出 JSONL，评测脚本把模型补全拼到函数签名之后，再运行隐藏测试。现在要让一个能读文件、执行命令的 Agent 在 Harbor 中完成同一批题。最容易犯的错误，是把每行 JSON 变成一个 Task 目录，然后把 Harbor 的平均 Reward 命名为原指标。目录确实生成了，实验语义却已经改变。
Adapter（适配器）不是文件格式转换脚本，而是一份可审计的语义迁移：它要锁定上游数据，说明每个字段去了哪里，保留或明确修改执行协议，防止答案与测试泄露，并证明新指标与旧指标在什么条件下才可比较。
本章使用 OpenAI 官方 HumanEval 仓库做案例。书中不搬运题目，只用官方文件的固定提交做转换探针。读完本章，你应当能够：
- 把外部 schema 映射为 Harbor 的 Task、Dataset、Verifier、Reward 与 Metric；
- 写出带摘要校验、ID 冲突检查、路径防护和许可门禁的批量转换器；
- 用 Oracle、错误实现和投机实现审计转换结果；
- 识别“生成成功”“Oracle 通过”与“原 Benchmark 结果复现”之间的差距；
- 制定可发布、可升级、可回滚的 Adapter 维护契约。
27.1 先写迁移契约，不要先写复制循环
HumanEval 的固定数据记录包含 task_id、prompt、entry_point、canonical_solution 和 test；官方 harness 把 prompt + completion + test + check(entry_point) 组成程序并执行。1 这决定了我们的映射不能只看字段名称。
| 上游对象 | Harbor 落点 | 本章约定 | 语义变化 |
|---|---|---|---|
task_id | Task 目录、[task].name、迁移清单 | HumanEval/7 映射成 he-007 | 不把公开 ID 放进 Agent 可见指令 |
prompt | environment/candidate.py | 容器启动时已有待补全函数 | 从文本补全变成文件编辑 |
canonical_solution | solution/reference.py | 仅供 Oracle | 不复制进 Agent 镜像 |
test、entry_point | tests/problem.json、evaluate.py | Verifier 阶段才挂载 | 仍执行上游断言，但运行环境改变 |
| 单个 completion 是否通过 | Trial Reward | 0 或 1 | 一次 Trial 对应一个候选实现 |
上游 pass@k | 独立结果后处理 | 按 task 分组后使用原估计器 | 不能用 Harbor mean 冒充 |
| 全部记录 | 本地 Dataset 目录 | 一题一个 Task | Registry Dataset 另需 digest 清单 |
Harbor v0.18.0 的 TaskConfig 读取 task.toml，其中 [task].name 必须符合 org/name；DatasetManifest 的 Registry 引用则要求每个 Task 带 sha256:<64 hex> digest。2 因此，Adapter 的直接产物可以先是一组本地 Task；准备发布时再初始化 dataset.toml，让 Harbor 计算包摘要。不要在转换器里虚构 Registry digest。
映射契约最好拆成三层。第一层是表示层：字符编码、换行、字段类型、ID 和目录名怎样规范化。第二层是执行层：上游 completion 在哪里落盘，谁能读取测试，怎样处理异常和超时。第三层是统计层：一个样本对应一个 Trial 还是一个 Task，多次采样怎样分组，缺失结果是否进入分母。只要任一层发生变化，就应在 README 与 parity 记录中写明，而不能藏在模板替换代码里。
还要区分五种 ID。source_id 是上游稳定键；local_id 是安全目录名；[task].name 是 Harbor 包名；Trial ID 是一次运行实例；内容 digest 标识包内容。把这五者都压缩成 he-007 会失去审计能力，把原始 ID 全部暴露在 Agent 容器里又会增加查表捷径。本章把完整映射只放在宿主侧的 migration-manifest.json，Agent 可见文件只保留完成任务所需的信息。
所谓“批量”也不是在循环外加进度条。转换器必须在落盘前完成全局检查，因为重复源 ID 和规范化后的碰撞只有看到全集才能发现。筛选 --task-ids 时，未知 ID 应报错而不是静默忽略；--limit 0 应拒绝空 Dataset；重跑时应比较所有权清单而不是对任意同名目录执行 rmtree。这些规则使错误可见，也使自动化作业能够可靠失败。
本章的验收标准是：固定输入生成固定目录；164 条官方记录全部通过 schema 和 TaskConfig 校验；164 个参考解都能通过对应上游断言；非法 schema、重复 ID、规范化碰撞、路径穿越、摘要错误、未知筛选 ID、非法覆盖和未接受许可都必须失败关闭。这里不运行模型，也不声称复现任何 HumanEval 分数。
27.2 固定来源、许可与内容，而不只是 URL
本书 Harbor 侧固定为 v0.18.0、提交 527d50deb63a5d279e8c20593c18a2cbc7f61f9e。案例上游另固定为 HumanEval 提交 6d43fb980f9fee3c892a914eda09951f772ad10d。该提交根目录附带 MIT License；其条件包括在软件或其重要部分中保留版权与许可文本。3 这是一条工程门禁，不是法律意见：组织仍应核对再分发范围、第三方材料和所在地要求，并在发布包中保留来源提交、许可证副本和 NOTICE。
本章实际核验的 data/HumanEval.jsonl.gz 有 164 条记录，压缩文件 SHA-256 为 b796127e635a67f93fb35c04f4cb03cf06f38c8072ee7cee8833d7bee06979ef。这一数字来自对固定提交中二进制文件的本地逐字节计算；官方仓库是数据来源。4 固定提交防止 Git 内容漂移，文件摘要防止下载缓存、代理或人工拷贝发生静默变化，两者缺一不可。
在 /private/tmp 或自己的工作目录运行：
git clone https://github.com/openai/human-eval.git human-eval
git -C human-eval checkout 6d43fb980f9fee3c892a914eda09951f772ad10d
git -C human-eval rev-parse HEAD
shasum -a 256 human-eval/data/HumanEval.jsonl.gz
uv run --python 3.12 python - <<'PY'
import gzip, json
from pathlib import Path
path = Path("human-eval/data/HumanEval.jsonl.gz")
with gzip.open(path, "rt", encoding="utf-8") as stream:
rows = [json.loads(line) for line in stream if line.strip()]
print({"records": len(rows), "keys": sorted(rows[0])})
PY
本章探针得到的最后一行是：
{'records': 164, 'keys': ['canonical_solution', 'entry_point', 'prompt', 'task_id', 'test']}
这不是下载器的职责。生产 Adapter 应让 CI 或数据准备作业把固定文件放入内容寻址缓存，例如 cache/sha256/b796.../HumanEval.jsonl.gz；转换器只接受本地路径并校验摘要。这样，网络重试不会和任务生成混在一起，离线复验也不会意外访问新版本。
缓存命中不能只看文件名或 HTTP ETag。缓存索引至少记录来源 URL、固定 Git 提交、获取时间、字节数、SHA-256、许可证文件摘要和获取工具版本；读取时仍重新计算内容摘要。下载到临时文件、校验成功后再 rename 到内容寻址路径，能避免中断下载被当成有效缓存。若上游使用 Git LFS、认证 URL 或会重定向的下载端点，还应把最终解析的对象 ID 写进证据清单。
许可审查也不能简化成“仓库页面显示 MIT”。应回答：根许可证是否覆盖数据目录；题面是否包含第三方文本；是否允许修改与再分发；是否要求署名、NOTICE 或相同许可；能否把标准解与隐藏测试公开；组织是否只发布生成脚本而不再分发原数据。本章选择“不搬运题目、让读者从官方固定提交取得数据”，降低再分发面，但没有消除使用数据本身的合规责任。
27.3 从 Harbor 的 Adapter 模板起步
Harbor v0.18.0 把主命令注册为单数 harbor adapter，复数形式只是隐藏的兼容别名。init 支持名称、adapter ID、来源 URL 和许可证参数；向导使用 uv init --package 建立 src/<package>/ 布局，并生成 adapter.py、main.py、Task 模板、元数据和 parity 文件。5
在 Harbor v0.18.0 源码根目录执行：
uv run --frozen harbor adapter init human-eval-mini \
--name HumanEval \
--description "Pinned HumanEval migration example" \
--source-url https://github.com/openai/human-eval \
--license MIT
生成目录的关键部分是：
adapters/human-eval-mini/
├── README.md
├── adapter_metadata.json
├── parity_experiment.json
├── pyproject.toml
└── src/human_eval_mini/
├── adapter.py
├── main.py
└── task-template/
├── task.toml
├── instruction.md
├── environment/Dockerfile
├── solution/solve.sh
└── tests/test.sh
模板要求 main.py 保留 --output-dir、--limit、--overwrite 和 --task-ids，并调用 Adapter 的 run()。结构审查还会检查包名、入口点、错误处理、模板占位符、Oracle 和泄露面。6
注意：向导调用本机
uv init，所以新建pyproject.toml的requires-python可能反映本机 uv 的默认版本，而不是 Harbor 的>=3.12基线；应人工改回兼容范围并在 CI 校验。另一个 v0.18.0 细节是，向导结束语打印了不存在的harbor adapters validate，但该版本实际提供的是harbor adapter review。源码中的命令注册与结束语可以直接看到这处不一致。7
正确的纯结构审查命令是：
uv run --frozen harbor adapter review \
--path adapters/human-eval-mini \
--skip-ai \
--output adapter-review-report.md
27.4 一个失败关闭的批量转换器
下面的清单可先保存为 adapters/human-eval-mini/src/human_eval_mini/adapter_probe.py。正式接入模板时，让 HumanEvalAdapter.run() 调用 convert()，并由 main.py 传入相同参数。示例刻意不自动下载数据；--accept-license MIT 只是显式门禁，不能替代许可证审查。
from __future__ import annotations
import argparse
import gzip
import hashlib
import json
import keyword
import os
import re
import shutil
import tempfile
from dataclasses import dataclass
from pathlib import Path
from typing import Iterable
SOURCE_COMMIT = "6d43fb980f9fee3c892a914eda09951f772ad10d"
SOURCE_SHA256 = "b796127e635a67f93fb35c04f4cb03cf06f38c8072ee7cee8833d7bee06979ef"
REQUIRED = ("task_id", "prompt", "entry_point", "canonical_solution", "test")
SOURCE_ID = re.compile(r"^HumanEval/([0-9]+)$")
@dataclass(frozen=True)
class Problem:
source_id: str
local_id: str
prompt: str
entry_point: str
canonical_solution: str
test: str
record_sha256: str
def sha256_bytes(data: bytes) -> str:
return hashlib.sha256(data).hexdigest()
def canonical_digest(record: dict[str, object]) -> str:
payload = json.dumps(
record, sort_keys=True, ensure_ascii=False, separators=(",", ":")
).encode()
return sha256_bytes(payload)
def _rows(path: Path) -> Iterable[tuple[int, dict[str, object]]]:
opener = gzip.open if path.suffix == ".gz" else open
with opener(path, "rt", encoding="utf-8") as stream:
for line_no, line in enumerate(stream, 1):
if not line.strip():
continue
value = json.loads(line)
if not isinstance(value, dict):
raise ValueError(f"line {line_no}: JSON value must be an object")
yield line_no, value
def load_problems(path: Path) -> list[Problem]:
actual = sha256_bytes(path.read_bytes())
if actual != SOURCE_SHA256:
raise ValueError(f"source digest mismatch: expected {SOURCE_SHA256}, got {actual}")
problems: list[Problem] = []
source_ids: set[str] = set()
local_ids: dict[str, str] = {}
for line_no, record in _rows(path):
missing = [key for key in REQUIRED if key not in record]
if missing:
raise ValueError(f"line {line_no}: missing fields {missing}")
if any(not isinstance(record[key], str) for key in REQUIRED):
raise ValueError(f"line {line_no}: required fields must be strings")
source_id = str(record["task_id"])
match = SOURCE_ID.fullmatch(source_id)
if match is None:
raise ValueError(f"line {line_no}: invalid task_id {source_id!r}")
if source_id in source_ids:
raise ValueError(f"line {line_no}: duplicate source id {source_id!r}")
source_ids.add(source_id)
local_id = f"he-{int(match.group(1)):03d}"
collision_key = local_id.casefold()
if collision_key in local_ids:
raise ValueError(
f"line {line_no}: local id collision {source_id!r} and "
f"{local_ids[collision_key]!r} -> {local_id!r}"
)
local_ids[collision_key] = source_id
entry_point = str(record["entry_point"])
if not entry_point.isidentifier() or keyword.iskeyword(entry_point):
raise ValueError(f"line {line_no}: invalid entry_point {entry_point!r}")
prompt = str(record["prompt"])
solution = str(record["canonical_solution"])
test = str(record["test"])
if not prompt.strip() or not solution.strip() or not test.strip():
raise ValueError(f"line {line_no}: prompt, solution, and test must be non-empty")
compile(prompt + solution + "\n" + test + f"\ncheck({entry_point})\n", source_id, "exec")
problems.append(
Problem(
source_id, f"he-{int(match.group(1)):03d}", prompt,
entry_point, solution, test, canonical_digest(record)
)
)
if not problems:
raise ValueError("source contains no records")
return problems
def safe_child(root: Path, name: str) -> Path:
if not re.fullmatch(r"[a-z0-9][a-z0-9-]*", name):
raise ValueError(f"unsafe local id {name!r}")
child = (root / name).resolve()
if child.parent != root.resolve():
raise ValueError(f"path escapes output root: {name!r}")
return child
def write_task(root: Path, problem: Problem) -> None:
task = safe_child(root, problem.local_id)
for relative in ("environment", "solution", "tests"):
(task / relative).mkdir(parents=True, exist_ok=False)
(task / "task.toml").write_text(f'''schema_version = "1.3"
[task]
name = "example/human-eval-mini__{problem.local_id}"
description = "Complete one Python function"
authors = [{{ name = "OpenAI" }}]
keywords = ["python", "functional-correctness", "adapter-example"]
[metadata]
source_commit = "{SOURCE_COMMIT}"
record_sha256 = "{problem.record_sha256}"
[agent]
timeout_sec = 600.0
user = "agent"
network_mode = "no-network"
[verifier]
timeout_sec = 30.0
user = "root"
network_mode = "no-network"
[environment]
build_timeout_sec = 300.0
network_mode = "no-network"
cpus = 1
memory_mb = 512
''', encoding="utf-8")
(task / "instruction.md").write_text(
"Complete the function in `/app/candidate.py`. Preserve its public signature. "
"Do not add network or third-party dependencies.\n", encoding="utf-8"
)
(task / "environment" / "candidate.py").write_text(problem.prompt, encoding="utf-8")
(task / "environment" / "Dockerfile").write_text(
"FROM python:3.12.11-slim-bookworm\n"
"RUN useradd --create-home --uid 10001 agent\n"
"WORKDIR /app\nCOPY candidate.py /app/candidate.py\n"
"RUN chown agent:agent /app/candidate.py && chmod 0644 /app/candidate.py\n",
encoding="utf-8",
)
(task / "solution" / "reference.py").write_text(
problem.prompt + problem.canonical_solution + "\n", encoding="utf-8"
)
(task / "solution" / "solve.sh").write_text(
"#!/usr/bin/env bash\nset -euo pipefail\n"
"cp /solution/reference.py /app/candidate.py\n", encoding="utf-8"
)
os.chmod(task / "solution" / "solve.sh", 0o755)
(task / "tests" / "problem.json").write_text(json.dumps(
{"entry_point": problem.entry_point, "test": problem.test},
ensure_ascii=False, sort_keys=True
) + "\n", encoding="utf-8")
(task / "tests" / "evaluate.py").write_text('''import json
from pathlib import Path
candidate = Path("/app/candidate.py").read_text(encoding="utf-8")
problem = json.loads(Path("/tests/problem.json").read_text(encoding="utf-8"))
namespace = {"__name__": "candidate"}
exec(compile(candidate, "/app/candidate.py", "exec"), namespace)
exec(compile(problem["test"], "/tests/hidden_checks.py", "exec"), namespace)
entry = namespace.get(problem["entry_point"])
check = namespace.get("check")
if not callable(entry) or not callable(check):
raise AssertionError("candidate entry point or check function is missing")
check(entry)
''', encoding="utf-8")
(task / "tests" / "test.sh").write_text('''#!/usr/bin/env bash
set -u
mkdir -p /logs/verifier
if timeout 10s /usr/local/bin/python /tests/evaluate.py; then
reward=1
else
reward=0
fi
printf '%s\\n' "$reward" > /logs/verifier/reward.txt
exit 0
''', encoding="utf-8")
os.chmod(task / "tests" / "test.sh", 0o755)
def select_problems(problems, task_ids, limit):
if limit is not None and limit < 0:
raise ValueError("limit must be non-negative")
selected = problems
if task_ids:
wanted = set(task_ids)
known = {p.source_id for p in problems} | {p.local_id for p in problems}
unknown = sorted(wanted - known)
if unknown:
raise ValueError(f"unknown task ids: {unknown}")
selected = [p for p in problems if p.source_id in wanted or p.local_id in wanted]
selected = selected if limit is None else selected[:limit]
if not selected:
raise ValueError("selection contains no records")
return selected
def convert(source, output, accept_license, *,
task_ids=None, limit=None, overwrite=False):
if accept_license != "MIT":
raise ValueError("license gate not accepted: pass --accept-license MIT")
problems = select_problems(load_problems(source), task_ids, limit)
output = output.resolve()
output.parent.mkdir(parents=True, exist_ok=True)
if output.exists() and not overwrite:
raise FileExistsError(f"refusing to overwrite existing output: {output}")
if output.exists():
try:
owner = json.loads((output / "migration-manifest.json").read_text())
except (OSError, json.JSONDecodeError) as exc:
raise ValueError("overwrite refused: no valid ownership manifest") from exc
if owner.get("adapter") != "human-eval-mini":
raise ValueError("overwrite refused: output belongs to another producer")
stage = Path(tempfile.mkdtemp(prefix=f".{output.name}.", dir=output.parent))
backup = None
try:
for problem in problems:
write_task(stage, problem)
manifest = {
"adapter": "human-eval-mini", "license": "MIT",
"source_commit": SOURCE_COMMIT, "source_sha256": SOURCE_SHA256,
"task_count": len(problems),
"tasks": [{"source_id": p.source_id, "local_id": p.local_id,
"record_sha256": p.record_sha256} for p in problems],
}
(stage / "migration-manifest.json").write_text(
json.dumps(manifest, ensure_ascii=False, indent=2, sort_keys=True) + "\n"
)
if output.exists():
backup = Path(tempfile.mkdtemp(prefix=f".{output.name}.backup.",
dir=output.parent))
backup.rmdir()
os.replace(output, backup)
try:
os.replace(stage, output)
except BaseException:
if backup is not None and backup.exists() and not output.exists():
os.replace(backup, output)
raise
if backup is not None:
shutil.rmtree(backup)
return manifest
except BaseException:
shutil.rmtree(stage, ignore_errors=True)
raise
def main():
parser = argparse.ArgumentParser()
parser.add_argument("--source", type=Path, required=True)
parser.add_argument("--output-dir", type=Path, required=True)
parser.add_argument("--accept-license", required=True)
parser.add_argument("--limit", type=int)
parser.add_argument("--task-ids", nargs="+")
parser.add_argument("--overwrite", action="store_true")
args = parser.parse_args()
manifest = convert(
args.source, args.output_dir, args.accept_license,
task_ids=args.task_ids, limit=args.limit, overwrite=args.overwrite
)
print(json.dumps({"task_count": manifest["task_count"],
"source_sha256": manifest["source_sha256"]}))
if __name__ == "__main__":
main()
这段代码有四个值得保留的性质。
第一，先验证、后写盘。全部记录先经过字段、类型、Python 标识符、编译和碰撞检查；任何一条失败都不会留下半套 Dataset。第二，ID 转换是白名单，而不是简单 replace("/", "-")；HumanEval/0 与 HumanEval/000 会映射到同一目录，转换器会在写文件前报告碰撞。第三，每条记录和整个压缩文件分别有摘要，维护者可以区分“上游包变了”和“单条规范化结果变了”。第四，覆盖只允许作用于带本 Adapter 所有权清单的目录，并通过同目录 rename 交换；它不会递归删除一个来源未知的路径。
转换命令没有提供覆盖 SOURCE_SHA256 的参数。若允许操作者传入修改后文件的摘要，却继续把 SOURCE_COMMIT 写成固定提交，清单内部会出现无法证明的来源组合。升级上游时应在一次受审查的 Adapter 版本变更中同时更新提交、文件摘要和回归 fixture，而不是在运行时绕过锁。
compile() 只检查组合程序能否解析，不会执行上游测试，也不会证明 entry_point 与断言一致。真正的 Oracle 探针要执行参考实现；真正的容器验收还要经过 Harbor 的 Solution、Verifier 和 Reward 路径。反过来，生成阶段绝不能为了“验证”而执行任意下载内容：解析和执行应是两个明确的安全域，后者只在资源受限的验证环境中进行。
示例的 Dockerfile 固定了 Python 补丁版本，但没有固定 OCI manifest digest，因此它仍不是完全不可变的环境。生产发布应针对实际架构记录镜像 digest，并验证 Registry 没有被同名标签替换。书稿没有执行 Docker，因此必须把基础镜像能否解析并运行 timeout 列入镜像烟雾测试。若镜像缺少该命令，Verifier 会写入 0，这应归类为基础设施失败，而不是模型失败。
record_sha256 采用对 JSON 对象按 key 排序、无多余空白的规范化序列化。它能忽略源文件的键顺序和空格变化，适合定位记录语义载荷；source_sha256 则对压缩文件原始字节计算，任何 gzip 元数据变化也会改变。两个摘要回答不同问题，不能相互替代。维护报告应同时给出“包字节变化”和“规范化记录变化”。
执行转换：
cd /path/to/harbor-v0.18.0
uv run --frozen python adapters/human-eval-mini/src/human_eval_mini/adapter_probe.py \
--source /private/tmp/human-eval/data/HumanEval.jsonl.gz \
--output-dir /private/tmp/human-eval-harbor \
--accept-license MIT
本章实际输出为：
{"task_count": 164, "source_sha256": "b796127e635a67f93fb35c04f4cb03cf06f38c8072ee7cee8833d7bee06979ef"}
27.5 从 Task 目录组成 Dataset
转换器不生成假的 digest。让 Harbor v0.18.0 扫描已经存在的 Task 子目录：
uv run --frozen harbor dataset init example/human-eval-mini \
--output-dir /private/tmp/human-eval-harbor \
--description "Pinned HumanEval adapter"
uv run --frozen harbor sync /private/tmp/human-eval-harbor
dataset init 会扫描输出目录中的 TaskConfig 并加入 manifest；sync 对本地 Task 重新计算内容摘要。发布命令也会在发布 Dataset 前自动执行同步，而且默认可见性是 private；使用 --public 时还会询问是否把所含 Task 一起公开。8
这里有一个常见顺序错误：先初始化空 Dataset，再让 Adapter 用 --overwrite 替换整个输出目录，会把 dataset.toml 一起删除。本章先原子生成 Task，后运行 dataset init。后续增量更新时，不应覆盖 Dataset 根目录；应把 Adapter 的 Task staging 与最终 Dataset manifest 更新拆开，先比对任务集合，再用 Harbor 的 add/remove/sync 流程更新引用。这样，删除一条上游记录会在 manifest diff 中明确出现。
Dataset 不是 Task 的共同测试脚本。每个 HumanEval Task 的隐藏断言属于该 Task；Dataset 只组织成员与聚合。若加入自定义 metric.py，它必须随 Dataset 文件 digest 发布。将上游测试塞进 Dataset 级脚本会破坏 Task 的独立可运行性，也容易让本地路径运行和 Registry 运行得到不同结果。
先用 Oracle 做单题烟雾测试，再做全量可解性检查：
uv run --frozen harbor trial start \
-p /private/tmp/human-eval-harbor/he-000
uv run --frozen harbor run \
-p /private/tmp/human-eval-harbor \
-a oracle \
--job-name human-eval-mini-oracle
这些命令需要可用的 Docker 环境。本章没有执行 Docker Job；实际执行的是更窄的转换器探针：164 个 task.toml 全部通过 v0.18.0 TaskConfig.model_validate_toml()，并在宿主 Python 中把每个 reference.py 与对应的 test 组合执行，164 个 Oracle 均通过。这证明字段映射和参考解可用，不证明容器、Agent 安装、并发调度或 Registry 发布成功。
27.6 Reward 不是 pass@k：指标语义对齐
HumanEval 官方实现对每题采样 n 个 completion，其中 c 个正确，并使用
[ \widehat{\mathrm{pass@k}} = 1 - \frac{\binom{n-c}{k}}{\binom{n}{k}} ]
再对题目取平均；当任一题的样本数小于 k 时，官方脚本不报告该 pass@k。9 Harbor v0.18.0 的内置 Mean 则把 Trial reward 字典逐项求算术平均，并把 None 当成 0。10 两者只在一个受限情形下重合：每题恰好一个候选、每题等权、二元 Reward、失败计入规则一致，此时 Dataset 的平均 Reward 可解释为一次采样的经验通过率。
如果每题运行 20 次，直接对 3280 个 Reward 求平均仍是“候选级通过比例”，不是 pass@10。正确流程是：
- 保存
source_id、Trial ID、样本序号、Reward 与基础设施状态； - 将超时、Verifier 错误和未生成结果按预注册规则处理；
- 按
source_id分组，得到每题的n与c； - 调用上游固定提交中的
estimate_pass_at_k，并报告实际可计算的k； - 同时保留 Harbor 原生 mean，使用不同指标名，例如
trial_success_mean与humaneval_pass_at_10。
举一个教学性构造：某题有 20 个候选，其中 2 个通过。候选级平均值是 2/20=0.1；“20 次里至少一个通过”的已观测指示量是 1；官方 pass@10 估计器使用 1-C(18,10)/C(20,10)。三者分别回答“随机抽一个候选是否通过”“这一批是否出现过正确候选”和“从 20 个候选中无放回取 10 个至少一个正确的估计概率”。把它们都叫成功率会让实验不可比较。
基础设施失败政策尤其重要。Harbor Mean 会把 None 当 0，但上游脚本期待每道题都被尝试，并把候选执行结果归为 passed、timed out 或 failed。适配后可以选择把镜像拉取失败排除并单独报告，也可以保守计 0；无论选择哪一种，都要在实验前注册，并在原 harness 一侧应用相同规则。事后根据分数好坏挑政策是选择性报告。
警告：给 Harbor 的
mean改显示名称，不能改变它的数学语义。指标对齐需要相同的采样单位、分组键、失败政策和估计器，而不只是相同的百分号。
此外，本章让 Agent 编辑完整文件，而上游让模型返回“prompt 之后的 completion”；Agent 还拥有终端、600 秒时间和文件上下文。这些变化可能影响结果。因此 parity 实验至少要固定同一批题、同一模型、同一采样数、同一温度、相同失败计入和同一上游估计器，再分别运行原 harness 与 Harbor。即使分数接近，也只能说在该实验配置下观察到一致性，不能由一次比较证明一般等价。
27.7 Oracle、Verifier 与 gaming 审计
官方 README 明确警告 harness 会运行不可信的模型生成代码，应放在可靠安全沙箱中；固定提交的执行器也注明其 reliability_guard 不是安全沙箱。11 Harbor 的容器只是隔离层，不应被描述为绝对安全边界。
本章 Task 采取了几项最低限度措施：Agent 使用非 root 用户；Agent 与 Verifier 均禁网；测试和参考解没有 COPY 进 Agent 镜像；Verifier 以 root 运行绝对路径的 Python；test.sh 每次都覆盖 /logs/verifier/reward.txt，不会相信 Agent 预先写入的分数。Harbor 自带 Adapter 审查也以“Agent 看不到 ground truth，且不能影响 pass/fail”为核心不变量，并检查测试挂载、奖励覆盖和 benchmark 身份泄露。12
这些措施仍未消除所有攻击。候选 Python 与隐藏断言最终在同一解释器 namespace 中执行；恶意候选可以检查运行时、修改全局对象、耗尽资源或利用容器漏洞。若 Benchmark 面向对抗性 Agent，应把函数调用改造成受限 IPC 协议，在独立 Verifier 容器中以更低权限运行候选，使用只读根文件系统、进程/内存/系统调用限制，并把每个测试放入独立一次性进程。任何这样的变化都必须进入 parity 说明，因为它可能改变超时和异常语义。
回归套件至少包含以下攻击矩阵：
| 类别 | 输入或行为 | 期望 |
|---|---|---|
| 非法 schema | 缺少 test、字段不是字符串、语法错误 | 转换失败，无输出目录 |
| 重复与碰撞 | 重复 HumanEval/0；再加入 HumanEval/000 | 分别报告重复和本地 ID 碰撞 |
| 路径穿越 | ../escape、绝对路径、反斜杠 | ID 白名单拒绝，根目录外无文件 |
| 摘要漂移 | 修改一个字节但保留旧 SHA | 读取记录前失败 |
| Oracle | prompt + canonical_solution | Reward 应为 1 |
| 错误实现 | 恒定返回值、异常、超时 | Reward 应为 0 |
| 奖励投机 | 候选尝试写 reward、删除测试或改解释器 | 非 root Agent 失败；Verifier 仍重写 reward |
| 身份泄露 | 扫描 Agent 镜像 | 不出现 solution/、tests/、上游 ID 或来源 URL |
| 指标漂移 | 每题多次 Trial 后直接求 mean | 审查必须阻止把结果标为 pass@k |
测试要覆盖“失败之后的文件系统”，而不仅断言抛出异常。路径穿越用例应记录输出根的父目录快照，失败后比较没有新文件；原子替换用例可在 os.replace(stage, output) 前注入异常，确认旧 Dataset 被恢复；覆盖用例应把另一个 Adapter 的所有权清单放进目标目录，确认其中哨兵文件未被删除。这样的测试能验证安全性质，而不是只验证错误消息。
Oracle 回归应再配三类负样本：语法正确但结果错误、导入时异常、永不返回。第一类验证断言，第二类验证异常路径，第三类验证超时与资源回收。再加入一个试图写 /logs/verifier/reward.txt 的候选，确认非 root Agent 无法写，并确认 test.sh 无论成功失败都覆盖 reward。若投机样本拿到 1，先修 Verifier，不要把它当作 Agent 能力。
注意：HumanEval 本身是公开数据。即使移除
HumanEval/0和仓库 URL，模型仍可能从函数内容识别已知题目；“隐藏 benchmark 名称”只能减少直接查表捷径，不能证明没有训练数据污染。
27.8 发布与长期维护
发布是一道人工门，而不是转换命令的尾声。Harbor v0.18.0 的 harbor publish 能识别 Task 或 Dataset 目录、先发布 Task，再发布 Dataset，并以内容哈希展示版本；它不会替你证明许可证、Oracle、指标 parity 或数据污染。13
建议把 Adapter 的维护状态拆成五个独立版本：
harbor_version：本章为v0.18.0和精确提交；upstream_commit：本章为6d43fb...；source_sha256：原始二进制内容；adapter_version：转换规则、模板与安全策略的版本；dataset_digest：Harbor 发布后得到的 Task/Dataset 内容摘要。
每次升级只改变一个轴。上游数据变更时，先输出新增、删除、记录摘要变化和 ID 映射差异；Harbor 升级时，重新跑 TaskConfig、Adapter review、Oracle 与错误答案；模板或 Verifier 变更时，把旧、新 Dataset digest 都保留，并重新做 parity。不要把同名 latest 当作复现实验的依据。
长期维护需要一张兼容性矩阵，而不是一条“支持最新版”。每行至少列出 Harbor 提交、Adapter 提交、上游提交、源摘要、镜像 digest、Task 数、排除项、Oracle 状态、负样本状态、parity 状态和发布日期。出现上游更正时，旧 Dataset 不应原地消失；标为 deprecated，说明已知问题并指向新 digest。研究结果仍可引用旧版本，工程用户则能明确迁移。
监控也应围绕契约。定期作业可以检查上游 HEAD 是否变化，但不能自动升级；它只创建候选更新报告。报告先比较许可证、schema、记录数和摘要，再生成 staging Dataset，最后运行确定性回归。只有维护者审阅差异并完成 parity，才能更新公开标签。把“发现新提交”和“发布新 Benchmark”拆开，能防止上游一次无关 README 修改触发成本高昂的全量评测。
正式公开前应满足：来源与许可复核完成；所有 Task 的原始 ID 到本地 ID 映射唯一；Oracle 全量通过；负样本和 gaming 回归通过；镜像与依赖固定；原、适配后指标使用同一估计器；parity_experiment.json 记录真实命令、Agent/模型版本、样本数、均值与不确定性；README 清楚写出修改、排除项和限制。没有完成真实 parity 时，相应字段应保留为未执行，而不是填入教学数字。
27.9 本章小结
- Adapter 是 schema、执行协议、安全边界和指标语义的共同迁移，不是目录生成器。
- 来源提交、文件摘要、记录摘要与许可证记录共同构成可追溯输入。
- ID 白名单、碰撞预检、路径约束、原子写入和所有权覆盖门禁应在批量生成前完成。
- Oracle 通过只证明参考解在适配环境可解；转换成功与结果复现是三个不同结论。
- HumanEval
pass@k需要按题分组的上游估计器，Harbor 的平均 Reward 不能自动替代。 - 发布前必须同时审计答案泄露、Verifier gaming、不可信代码隔离、许可和 parity。
27.10 练习
- 为转换器补充 pytest：构造
HumanEval/0与HumanEval/000，验证输出目录尚未创建时就报告碰撞；再构造../escape，验证根目录之外没有新增文件。 - 在不改变
test断言的前提下，把候选执行移到独立子进程；记录超时、异常和 Reward 语义相对本章实现发生的变化。 - 为每题生成 20 个教学性二元结果，分别计算
trial_success_mean、朴素“至少一次成功比例”和官方pass@10估计值，解释三者为何不同。 - 设计一次 parity 实验计划，明确原 harness 与 Harbor 的模型、温度、样本数、时间限制、失败计入和置信区间；不要实际调用付费模型。
- 模拟上游文件只修改一条记录，生成记录摘要差异报告，并决定 Adapter、Dataset 与 parity 中哪些版本需要更新。