代码实现

Harness 负责执行一次完整的 agent 循环：验证兼容性、启动所需 runtime、执行一个 PreparedTask、生成标准化的公开 RunResult，并释放自己创建的所有资源。

下面的教程适配器直接返回配置的答案，不会调用 Model，因此注册与生命周期冒烟测试的结果是确定的。接入真实 SDK 或 CLI 时，只需替换执行主体，并继续使用相同的公开契约。

​

记录上游契约

记录官方框架或 CLI 版本、支持的 Model 协议、配置格式、提示词流程、工具与工作区行为、安装方式、超时、终止规则、轨迹格式和凭证处理。如果版本差异会影响命令、提示词、解析或可复现性，必须固定版本。优先使用公开 SDK 或 CLI，不要依赖私有函数。

​

创建最小文件

最小完整集成需要一个实现文件和一个软件包导出：

src/agentcompass/harnesses/
├── __init__.py
└── example_answer.py

创建 src/agentcompass/harnesses/example_answer.py：

from __future__ import annotations

from dataclasses import dataclass
from typing import Any

from agentcompass.runtime import (
    HARNESSES,
    BaseHarness,
    EnvironmentSession,
    EnvironmentSpec,
    HarnessPlan,
    ModelSpec,
    PreparedTask,
    RunRequest,
    RunResult,
    TaskStatus,
)
from agentcompass.runtime.config import RuntimeHarnessConfig, config_field


@dataclass(slots=True)
class ExampleAnswerConfig(RuntimeHarnessConfig):
    """User-facing parameters for the tutorial Harness."""

    answer: str = config_field(
        default="Paris",
        description="Deterministic answer returned by the tutorial Harness.",
    )

    def __post_init__(self) -> None:
        self.answer = str(self.answer)


@dataclass(slots=True)
class ExampleAnswerPlan(HarnessPlan):
    """Resolved runtime state passed to every lifecycle method."""

    answer: str = "Paris"


@HARNESSES.register()
class ExampleAnswerHarness(BaseHarness):
    id = "example_answer"
    description = "Deterministic Harness used by the developer tutorial."
    config_class = ExampleAnswerConfig
    plan_class = ExampleAnswerPlan

    def supports(self, environment: EnvironmentSpec, model: ModelSpec) -> bool:
        _ = environment, model
        return True

    async def start_session(
        self,
        env: EnvironmentSession,
        req: RunRequest,
        plan: HarnessPlan,
    ) -> dict[str, Any]:
        _ = req
        self._require_plan(plan)
        return {"env": env}

    async def execute_task(
        self,
        session: dict[str, Any],
        prepared: PreparedTask,
        req: RunRequest,
        plan: HarnessPlan,
    ) -> RunResult:
        _ = session, req
        harness_plan = self._require_plan(plan)
        return RunResult(
            task_id=prepared.task_id,
            status=TaskStatus.COMPLETED,
            category=prepared.category,
            final_answer=harness_plan.answer,
            telemetry={"answer_characters": len(harness_plan.answer)},
        )

    async def close_session(self, session: dict[str, Any]) -> None:
        _ = session

    @staticmethod
    def _require_plan(plan: HarnessPlan) -> ExampleAnswerPlan:
        if not isinstance(plan, ExampleAnswerPlan):
            raise TypeError("example_answer requires ExampleAnswerPlan")
        return plan

以上代码实现了 supports()、start_session() 和执行钩子 execute_task()。不需要额外回收时，继承的 collect_result() 原样返回传入的结果。虽然基类已经提供空操作实现，示例仍显式展示了 close_session()。BaseHarness.build_plan() 会把名称匹配的配置字段复制到 plan_class，其中也包括继承的 inject_network_restriction_notice 字段。

​

导出并检查注册

在 src/agentcompass/harnesses/__init__.py 中添加导入：

from .example_answer import ExampleAnswerHarness

然后检查组件发现和自动生成的配置文档：

uv run agentcompass list harness
uv run agentcompass config docs harness example_answer

第一条命令的输出应包含 example_answer 及其描述；第二条命令应显示默认值为 Paris 的 answer 和继承的网络提示字段。找不到 ID 说明导入或注册失败；能够找到 ID 也不代表真实上游 runtime 已经可以安装和启动。

​

运行单个任务

使用 Benchmark 的 Harness 驱动教程中的 example_exact_match：

uv run agentcompass run example_exact_match example_answer unused-model \
  --env host_process \
  --benchmark-params '{"sample_ids":["capital-france"]}' \
  --harness-params '{"answer":"Paris"}' \
  --task-concurrency 1 \
  --no-enable-analysis \
  --results-dir results-dev \
  --run-name harness-smoke

这个教程 Harness 不调用 req.model，因此命令不需要 Model 端点。命令应完成选定任务并输出 paths.run_info，该路径的父目录就是本次运行目录。

运行目录应包含 run_info.json、params.json、progress.json、progress.jsonl、run.log、task 和 attempt 的 result.json 文件、metrics.json 和 summary.md。经过 Benchmark 评测后，详情记录中的任务 attempt 应包含 status: "completed"、final_answer: "Paris"、metrics.correct: true 和 meta.harness.telemetry.answer_characters: 5。

真实 Harness 还需要使用一个范围受控的上游任务和实际凭证重复冒烟测试。只检查注册表不会验证安装、启动、解析、清理或 Model 端点调用。

​

替换教程执行主体

公开参数应放入 RuntimeHarnessConfig，确保 CLI、Python SDK、配置文件和自动生成文档使用相同字段。版本、启动模式、安装策略、步骤上限、命令超时和成本行为等标准化 runtime 选项，应放入类型化 HarnessPlan。不要修改 RunRequest、读取私有 Benchmark 字段或在计划中持久化密钥。

supports(environment, model) 应根据实际能力判断兼容性，而不是根据 Benchmark ID。检查内容包括 Model 协议、终端和文件系统要求、端点转发、浏览器或 GUI 能力、工作区假设、凭证位置，以及所选 Environment 能否完成安装。应尽可能在 Environment 启动前拒绝不支持的组合，绝不能静默切换协议、provider、安装模式或 Model。

runtime 按以下顺序调用生命周期：

在基线网络策略下执行 start_session
  -> 在运行网络策略和执行 deadline 下调用 execute_task
  -> 在运行网络策略和独立回收预算下调用 collect_result
  -> 在运行网络策略下执行 close_session

start_session() 用于执行可信安装、生成配置、上传文件、创建客户端或启动后台服务。execute_task() 每次只执行一个 PreparedTask，并且只能使用 prepared.input.prompt、messages、files、media、tools 和 workspace 等公开字段。

无论任务成功、超时、取消还是出错，close_session() 都要释放 Harness 负责的客户端、进程、服务器、临时配置和后台任务。Environment 由 runtime 关闭，不属于 Harness 的清理范围。

​

执行结束后的结果回收

正常结束、阶段超时或执行报错后，runtime 都会在关闭 Harness session 之前调用 collect_result(session, prepared, req, plan, result)。覆盖此钩子以读取已保存的输出、轨迹或候选文件，并在等待 agent 执行之前将回收所需的状态写入 session。钩子只回收现有输出，不得恢复 agent 循环、调用 Model 或生成新候选。返回 RunResult；即使回收成功，runtime 也会保留原始执行错误。

execution.harness_result_timeout_seconds 单独限制回收，默认 60 秒，必须为有限正数；执行和评测倍率不影响该预算。回收失败会单独记录，不覆盖原始执行错误。调用方主动取消时直接传播取消并进入清理，不再启动回收。

仍然实现 run_task() 的旧扩展可通过默认 execute_task() 桥接继续运行，但要独立于执行协程回收输出，需要实现 collect_result()。新扩展应实现 execute_task()；直接调用基类 run_task() 会组合执行与回收，不附带 runtime 的阶段 watchdog。

Harness 回退预算使用类属性 default_run_timeout_seconds 声明。如果旧 HarnessPlan.timeout 提供非空默认值，却未显式声明类属性，规划阶段会报错并提示迁移。显式声明 None 表示不提供 Harness 默认值。最终公共预算仅由 Planner 映射到原生 plan.timeout，不要再开放另一个用户 timeout 入口。

​

标准化结果，但不评分

Harness 只报告执行结果，不判断 Benchmark 正确性。它应返回当前能够获取的最完整 final_answer，并保留请求文件、有序轨迹、词元用量、耗时、产物和对应的 TaskStatus。超时、拒绝、无效输出、终止、安装、启动、解析和 Model API 错误都应保留原始语义。

不要在 Harness 中向 RunResult.metrics 写入 Benchmark 观测，也不要把评测器失败转换成 Harness 失败。Harness 诊断应写入 RunResult.telemetry。进程退出码为零不等于结果正确。公开答案必须写入 RunResult.final_answer；Benchmark 不应从 Harness 私有产物中恢复答案。

只支持能够复现的安装策略：固定的预安装镜像、受限执行前的受控安装，或隔离的驱动侧可选依赖。不要假设每个镜像都有软件包管理器或编译器，也不要为了安装方便而放宽运行阶段的网络策略。

通过受支持的 Environment 或配置机制注入 Model、评测器、搜索服务和 provider 的凭证，并对命令、文件、日志、轨迹、URL、异常、数据类表示和持久化元数据进行递归脱敏。

每种限制只能由一层负责：公共执行 deadline 限制 agent 阶段，命令超时限制单次工具命令，步骤上限约束 agent 循环，Model 客户端重试处理请求传输，runtime 重试则重新执行失败的任务尝试。遇到未知 Model 定价时，必须遵循 Harness 的成本契约；如果用户明确选择忽略错误或禁用成本模式，不应因此终止运行。

agent 变量通过 Environment 阶段作用域传递：run_env_variables 在 start_session() 和 execute_task() 中生效，在 collect_result() 和 close_session() 前退出。嵌套原生工具配置应使用 env.get_task_env_variables()，或通过 env._merge_exec_env() 合并自身默认值。不要新增 Harness 的 env 配置或 plan 字段。原生协议约束通过 env.exec(..., required_env={...}) 声明，共享合并层补入缺失值并在启动命令前拒绝冲突。可覆盖的默认值放在 env 中。内部模块路径优先使用 runner 参数或 Python 启动代码，保留用户的 PYTHONPATH。自定义 Environment session 必须实现并转发 required_env 关键字参数。

​

按阶段诊断失败

现象	阶段	首先检查
list harness 中没有 ID	导入与注册	harnesses/__init__.py、重复 ID、可选导入和堆栈
config docs 缺少字段	配置与计划	config_class、config_field()、plan_class 中同名字段和 __post_init__()
Environment 打开前失败	兼容性预检	supports() 的协议与能力检查
Environment 已打开，随后准备失败	start_session()	安装命令、基线阶段的网络策略、凭证、上传内容和启动日志
任务尝试出现 run_error	execute_task() / collect_result()	退出码、超时归属、响应解析、最终答案提取和轨迹转换
执行正确但 Benchmark 看不到答案	结果标准化	应写入 RunResult.final_answer，而不是私有产物或指标
取消后仍残留进程或客户端	close_session()	清理责任和能在取消时执行的 finally 清理路径
已有答案但 metrics.correct: false	Benchmark 评测	不要修改 Harness 状态；检查评测器契约与输出格式

简洁的真实生命周期与结果适配器可参考 qwen3vl_gui.py。需要查看如何通过 host_process 在 AgentCompass 进程内直接运行 agent、收集最终答案并转换轨迹时，可参考 naive_search_agent/harness.py。

Harness 配置 dataclass 定义允许的参数 key。构造配置时会拒绝未知 key；Recipe 所需参数必须声明为配置字段，或通过类型化的计划处理。

内部 POSIX 搜索路径的前置或追加统一使用 agentcompass.utils.command 中的 command_with_path_updates。它在公共和阶段变量合并后，在目标环境内扩展继承值，正确引用路径和参数，并通过 exec 保持 provider 对进程的控制。不要读取宿主 PATH 来构造 sandbox 命令，也不要启动会重置已解析环境变量的 login shell。上传的模块、脚本或 Python console script 统一使用 agentcompass.utils.command 中的启动器，仅向当前解释器的 sys.path 添加源码路径，保留用户的 PYTHONPATH。脚本启动器保留脚本所在目录的导入语义，移除 python -c 隐式加入的工作目录；用户在 PYTHONPATH 中显式声明的工作目录仍然生效。这些工具属于内部实现，不新增 CLI 字段。不要为任务修改宿主 os.environ；HTTP 客户端直接配置，子进程设置通过 Environment 传递。

from agentcompass.utils.command import command_with_path_updates, python_module_command

await env.exec(command_with_path_updates(
    ["bash", "/tests/test.sh"], prepend={"PATH": ["/opt/verifier/bin"]},
))
await env.exec(python_module_command(
    "python3", "runner", "--task", task_path, source_path="/opt/shipped_source",
))