第 26 章：扩展 Agent、Environment 与运行后端
一个云端 Trial 在“创建沙箱”请求发出后恰好收到取消信号。服务端已经创建资源，客户端却还没来得及把返回对象写入 self._sandbox。如果 stop() 只看这个字段，它会得出“没有资源可清理”的结论；评测显示为已取消，账单中的沙箱却继续存在。
这不是抽象的并发难题。Harbor v0.18.0 的 Daytona 实现专门屏蔽创建请求免受外层取消；若外层已取消，它会短暂等待创建任务返回，以便取得资源句柄再进入清理路径。1 一个可靠的 Provider，价值不只在于“能执行命令”，还在于失败、超时和取消时仍能回答三个问题：资源是否已创建、句柄是否已记录、重复清理是否安全。
本章锁定 Harbor Framework v0.18.0、提交 527d50deb63a5d279e8c20593c18a2cbc7f61f9e。我们先划清 Agent 与 Environment 的公共契约，再比较 Docker、E2B 和 Daytona 的实现差异，最后编写一个可由 Factory 导入的契约测试替身。替身不会执行宿主 shell，也不提供隔离，因此不能用来运行真实 Benchmark；它的用途是让 Provider 作者在接入云 SDK 之前验证 Harbor 侧的生命周期、文件、环境变量、能力拒绝和错误语义。
读完本章，你应该能够：
- 判断一个自定义执行器应继承
BaseAgent还是BaseInstalledAgent； - 实现
BaseEnvironment的命令、文件和生命周期契约，并通过import_path接入 Factory； - 解释 Docker、E2B、Daytona 在认证、快照、网络、资源、Compose 与删除行为上的差异；
- 在创建失败、命令超时、取消和重复
stop()时避免泄漏资源； - 用契约测试证明一个 Provider 会拒绝自己无法兑现的能力。
26.1 两条扩展轴，不是一层插件
Agent 与 Environment 是正交的两条扩展轴。Agent 决定“如何完成任务”：安装什么工具、怎样把 instruction 交给模型或 CLI、怎样记录 token、费用和 trajectory。Environment 决定“在哪里执行”：怎样创建隔离资源、传文件、启动进程、应用用户与环境变量、限制网络，最后停止或删除资源。
BaseAgent 的公共抽象面很小：name()、version()、setup() 和 run()；run() 接受 instruction、Environment 与可变的 AgentContext。populate_context_post_run() 是可选的宿主侧回填点。2 BaseEnvironment 的必需面则包含 type()、定义校验、start()、stop(delete)、四个上传/下载方法和 exec()；目录过滤、healthcheck、主 service 路由等能力建立在这些原语之上。3
这带来第一条设计规则：不要让 Agent 直接 import 某个云 SDK。否则 Agent 会和 Daytona/E2B 绑定，无法再在 Docker 中复现；也不要让 Environment 知道特定 Agent 的 prompt 或日志格式。二者只通过 BaseEnvironment 的命令、文件和上下文作用域协作。
本章将接口按稳定性分为三层：
| 层 | 例子 | 扩展代码怎样使用 |
|---|---|---|
| 公共抽象契约 | BaseAgent.run()、BaseEnvironment.exec()、EnvironmentCapabilities | 实现并写契约测试 |
| 公共配置/入口 | AgentConfig.import_path、EnvironmentConfig.import_path、CLI --agent/--env | 作为部署入口 |
| 内部实现 | _ENVIRONMENT_REGISTRY、_apply_network_policy()、Provider 私有 strategy | 只用于理解锁定版本；升级时重审 |
第三方 Environment 不必把名字加入 EnvironmentType。v0.18.0 的 type() 明确允许外部子类返回任意字符串，而 Factory 的 import-path 分支会动态导入类并传入标准构造参数。内置 Provider 才使用枚举与惰性 registry。4 这意味着修改 _ENVIRONMENT_REGISTRY 不是普通项目接入所需步骤；那会变成维护 Harbor fork。
26.2 扩展 Agent：直接 API 还是 Installed Agent
选择基类的依据不是“Agent 是否复杂”，而是进程边界。
- 若 Python 类自己调用模型 SDK、维护对话并使用
environment.exec()驱动工具，继承BaseAgent。 - 若核心程序已是容器内 CLI，需要安装、构造 flags、注入环境变量并把非零退出映射为异常，继承
BaseInstalledAgent。 - 若只有一个 shell 脚本，也不要把它伪装成 Provider；它仍是 Agent 的执行策略。
Agent Factory 对内置名称和自定义 module.path:ClassName 采用不同分支；AgentConfig.env 会先解析，再作为 extra_env 传给 Agent。5 BaseInstalledAgent 进一步提供 CliFlag、EnvVar、error pattern、root/agent 用户命令包装、可选版本探测和统一的 setup。它的 _exec() 会在命令前加 set -o pipefail，非零退出再按 stdout/stderr 分类；默认 error patterns 覆盖限流、额度、API 过载和常见网络错误。6
下面的最小 Agent 假设镜像已经把一个经过供应链验证的二进制放在 /opt/diagnostic-cli/bin/diagnostic-cli。install() 只把它安装到 PATH，不在 Trial 中从浮动 URL 下载。这个选择把依赖来源固定在环境镜像中，也让 build 与 run 的证据边界清楚。
from __future__ import annotations
import shlex
from typing import override
from harbor.agents.installed.base import (
BaseInstalledAgent,
CliFlag,
ErrorPattern,
NonZeroAgentExitCodeError,
)
from harbor.environments.base import BaseEnvironment
from harbor.models.agent.context import AgentContext
class DiagnosticAuthenticationError(NonZeroAgentExitCodeError):
pass
class DiagnosticCliAgent(BaseInstalledAgent):
CLI_FLAGS = [
CliFlag("max_steps", cli="--max-steps", type="int", default=20),
]
ERROR_PATTERNS = [
ErrorPattern(r"AUTH_DENIED", DiagnosticAuthenticationError),
*BaseInstalledAgent.ERROR_PATTERNS,
]
@staticmethod
@override
def name() -> str:
return "diagnostic-cli"
@override
def get_version_command(self) -> str:
return "diagnostic-cli --version"
@override
async def install(self, environment: BaseEnvironment) -> None:
await self.exec_as_root(
environment,
"test -x /opt/diagnostic-cli/bin/diagnostic-cli && "
"install -m 0755 /opt/diagnostic-cli/bin/diagnostic-cli "
"/usr/local/bin/diagnostic-cli",
)
@override
async def run(
self,
instruction: str,
environment: BaseEnvironment,
context: AgentContext,
) -> None:
prompt = shlex.quote(self.render_instruction(instruction))
command = f"diagnostic-cli {self.build_cli_flags()} --prompt {prompt}"
result = await self.exec_as_agent(
environment,
command,
timeout_sec=900,
)
context.metadata = {
"exit_code": result.return_code,
"stdout_tail": (result.stdout or "")[-2000:],
}
这里有四个容易漏掉的工程点。
第一，instruction 必须作为一个 shell 参数引用，不能把未转义文本拼入命令。第二，系统级安装显式使用 root，运行阶段使用 Trial 设置的默认 Agent 用户；BaseEnvironment._resolve_user() 的规则是显式 user 优先，否则回落到 default_user。7 第三，Agent 只记录自己确实得到的 metadata；如果 CLI 没有提供 token 或费用，不能填入估算值冒充计量。第四，error pattern 的异常应保留 NonZeroAgentExitCodeError 语义，使上层既能按具体原因分组，又不会把失败退出当作成功结果。示例的 stdout_tail 只是诊断字段；生产实现应先删除 token、key 和用户数据，再决定是否落盘。
注意：
BaseInstalledAgent._exec()的set -o pipefail是 POSIX/Bash 语义，而且BaseAgent.SUPPORTS_WINDOWS默认是False。不要仅把SUPPORTS_WINDOWS=True就宣称兼容 Windows；install()、flags、引用规则、日志路径和底层 CLI 都必须有 Windows 证据。
26.3 BaseEnvironment 的真实契约
“实现了七个抽象方法”只是语法通过。一个可用 Provider 还要守住构造、运行与清理三段契约。
构造阶段：先拒绝，后计费
BaseEnvironment.__init__() 接收 task 环境定义、资源 override、持久环境变量、mount、网络策略和可能的 phase policies。它会解析 override 与 task env，然后依次校验定义、资源模式、GPU、TPU、网络和 Windows 支持。8 因而 capabilities 不是展示用元数据，而是创建远端资源之前的门禁。
EnvironmentCapabilities 分开表达 GPU、TPU、断网、allowlist 各类条目、动态网络切换、Windows、mounted 与 Docker Compose。EnvironmentResourceCapabilities 又把 CPU/内存的 limit 和 request 分开。9 生产 Provider 应遵循保守声明：只声明已实现并有测试的能力。把 disable_internet=True 写上去，却在 SDK 不支持时悄悄放行，比直接拒绝任务更危险。基类对 no-network、allowlist 及条目类型的要求就是“准确执行或拒绝”；动态切换还要求实现 _apply_network_policy()。10
运行阶段：统一结果，不统一传输技术
exec() 的形状统一为 command、cwd、env、timeout 和 user，返回 ExecResult(stdout, stderr, return_code)。11 但它没有规定 Provider 必须用 Docker exec、SSH 还是云 SDK。实现仍需明确：
- 非零退出是普通
ExecResult，还是 Provider 异常； - timeout 会终止远端进程，还是只停止本地等待；
- HTTP 重试是否可能重复有副作用的命令；
- stdout/stderr 是否可分别取得；
env、持久 env 与 Agent scoped env 的优先级是否落实。
Harbor 基类的合并顺序是 persistent < per-exec < scoped；scoped env 用 ContextVar 隔离 asyncio task，避免并发 Trial 互相泄漏。7 Provider 若绕开 _merge_env()，AgentConfig 的环境覆盖就会失效。E2B 的实现只在“能证明命令尚未送达”的连接建立错误或 429 上重试 dispatch；拿到 command handle 后不再重发，以避免重复副作用。12 这是一种 Provider 特定策略，不是所有云 SDK 自动提供的保证。
文件契约同样不能只测“一个文本文件”。至少覆盖空文件、二进制、嵌套空目录、覆盖已有目标、缺失源、权限以及路径穿越。upload_dir(source, target) 在团队内部还必须约定是复制目录本身还是目录内容；Harbor 的 Agent 与 Artifact 调用方依赖具体 Provider 对同一接口保持一致。
清理阶段：句柄先落地，stop 可重入
推荐把生命周期写成下面的状态机，而不是一个布尔值：
NEW → CREATING → RUNNING → STOPPING → STOPPED
│ │ │
└──────────┴──────────┴──→ cleanup(delete=True)
一旦服务端可能创建资源，就要尽早记录可删除的 id；创建被取消时，要么用 idempotency key 查询，要么像 Daytona 那样 shield 创建并等待句柄。stop(delete=True) 应允许在 NEW、部分创建、已经删除和第二次调用时执行。认证失败也不能吞掉：Daytona 在 delete 被认证/授权拒绝时退化为 stop，并明确警告沙箱仍留在账户中。13 “本地对象已清空”不等于“远端资源已删除”。
26.4 三个后端，三套能力边界
下表只描述 Harbor v0.18.0 中这三个实现，不代表其他 Provider。
| 维度 | Docker | E2B | Daytona |
|---|---|---|---|
| preflight | 检查 docker 与 daemon | 检查 E2B_API_KEY 是否存在 | 接受 DAYTONA_API_KEY，或 JWT + organization id |
| 环境构建 | Compose build 或 prebuilt image | content-hash 命名的 E2B Template | direct image/Dockerfile、显式/自动 snapshot，或 DinD Compose |
| 资源语义 | CPU/内存 limit | CPU/内存 request | CPU/内存 request；另有 GPU 映射 |
| 文件/进程 | Docker Compose 与 tar/copy | E2B files/commands SDK | Daytona fs/process；Compose 时经 DinD strategy |
delete=False | down，但不带删除 local image/volume 参数 | 仍 kill；实现说明其沙箱是 ephemeral | direct 模式保留沙箱并丢弃本地句柄 |
| Compose | 支持；sidecar 有 service 操作 | 不声明支持 | 支持，但进入 DinD 后能力会变化 |
| 网络 | Linux 且启动时启用 egress sidecar 后可断网/allowlist/动态切换；Windows 拒绝非 public | 支持断网、部分 allowlist 条目与动态切换 | direct 支持断网、部分 allowlist 与动态切换；DinD 不声明 allowlist/动态切换 |
Docker 的 capabilities 是实例相关的：只有本次启动策略需要且平台能启用 egress control 时，断网、allowlist 和动态切换才为真；它同时声明 mounted、Compose、Windows，资源策略则是 CPU/内存 hard limit。14 启动会写临时 Compose overlay、按需构建、清理同 session 的旧容器再 up --detach --wait；停止在 finally 中删除这些临时文件。15 Docker timeout 路径会先 terminate，五秒仍未退出再 kill；这说明“本地超时”与“进程已消失”在该实现里被显式连接。16
E2B preflight 只检查 E2B_API_KEY 是否存在；E2B 官方也把这个环境变量列为 API key 的默认来源。17 Harbor 用环境内容哈希的一部分组成 template alias，缺失或 force build 时构建 Template，再创建一天 timeout 的 AsyncSandbox；stop() 即使收到 delete=False 也会 kill，并在 finally 中清空句柄。18 它声明 hostname、wildcard hostname 与 IPv4 literal allowlist，但不声明 IPv6 与 CIDR；不要把“E2B 支持 allowlist”简化成“任意 IP 规则都可用”。19
Daytona 的分支最多。它在构造阶段根据 docker-compose.yaml 或额外 overlay 选择 direct 或 DinD strategy；direct 可以使用显式 snapshot、content-hash 自动 snapshot、prebuilt image 或 Dockerfile。20 自动 snapshot 对同名创建加 asyncio lock，处理 ACTIVE/PENDING/ERROR，并在等待十分钟后抛出 TimeoutError；显式 ERROR snapshot 失败，自动 ERROR snapshot 则尝试删除后重建。21 Daytona 官方文档也把 snapshot 作为创建沙箱的基础对象，并由 SDK 暴露 snapshot service。22
快照在这里是构建/启动复用机制，不是 Trial 结果存储。Artifact、trajectory 和 Result 仍要回传到 Harbor 的 Job 目录；不能因为使用 snapshot 就省略 download_* 和 stop() 的测试。
警告：本章没有连接 E2B 或 Daytona 账户，没有创建云沙箱，也没有测价格、启动速度或稳定性。表格是锁定源码的接口事实，不是云性能排名。
26.5 一个不说谎的 Provider 契约替身
下面的 ContractTestEnvironment 完整继承 v0.18.0 的抽象类。它把“远端文件系统”映射到 Trial 目录下的临时根，但不执行任意命令，只识别测试命令 read、exit 和 sleep-forever。因此它可以测试 Harbor 调用约定，却不可能误伤宿主，更不能被当成沙箱。
from __future__ import annotations
import asyncio
import shutil
import shlex
from dataclasses import dataclass
from pathlib import Path, PurePosixPath
from typing import override
from harbor.environments.base import BaseEnvironment, ExecResult
from harbor.environments.capabilities import (
EnvironmentCapabilities,
EnvironmentResourceCapabilities,
)
@dataclass(frozen=True)
class ExecCall:
command: str
cwd: str | None
env: dict[str, str] | None
timeout_sec: int | None
user: str | int | None
class ContractTestEnvironment(BaseEnvironment):
"""A deterministic contract test double; it is not an isolation backend."""
def __init__(self, *args, fail_start: bool = False, **kwargs) -> None:
self._fail_start = fail_start
self._started = False
self._root: Path | None = None
self.calls: list[ExecCall] = []
super().__init__(*args, **kwargs)
@staticmethod
@override
def type() -> str:
return "contract-test"
@classmethod
@override
def resource_capabilities(cls) -> EnvironmentResourceCapabilities:
return EnvironmentResourceCapabilities()
@property
@override
def capabilities(self) -> EnvironmentCapabilities:
return EnvironmentCapabilities()
@override
def _validate_definition(self) -> None:
if not self.environment_dir.is_dir():
raise FileNotFoundError(self.environment_dir)
def _require_started(self) -> Path:
if not self._started or self._root is None:
raise RuntimeError("environment is not started")
return self._root
def _remote_path(self, remote: str) -> Path:
root = self._require_started().resolve()
relative = (
PurePosixPath(remote).relative_to("/")
if remote.startswith("/")
else PurePosixPath(remote)
)
candidate = (root / Path(*relative.parts)).resolve()
if not candidate.is_relative_to(root):
raise ValueError(f"path escapes sandbox root: {remote}")
return candidate
@override
async def start(self, force_build: bool) -> None:
if self._started:
return
root = self.trial_paths.trial_dir / "contract-sandbox"
root.mkdir(parents=True, exist_ok=True)
self._root = root
try:
if self._fail_start:
raise RuntimeError("injected start failure")
self._started = True
except BaseException:
shutil.rmtree(root, ignore_errors=True)
self._root = None
self._started = False
raise
@override
async def stop(self, delete: bool) -> None:
self._started = False
if delete and self._root is not None:
shutil.rmtree(self._root, ignore_errors=True)
self._root = None
@override
async def upload_file(self, source_path: Path | str, target_path: str) -> None:
target = self._remote_path(target_path)
target.parent.mkdir(parents=True, exist_ok=True)
shutil.copy2(source_path, target)
@override
async def upload_dir(self, source_dir: Path | str, target_dir: str) -> None:
source = Path(source_dir)
if not source.is_dir():
raise FileNotFoundError(source)
target = self._remote_path(target_dir)
target.mkdir(parents=True, exist_ok=True)
shutil.copytree(source, target, dirs_exist_ok=True)
@override
async def download_file(self, source_path: str, target_path: Path | str) -> None:
source = self._remote_path(source_path)
target = Path(target_path)
target.parent.mkdir(parents=True, exist_ok=True)
shutil.copy2(source, target)
@override
async def download_dir(self, source_dir: str, target_dir: Path | str) -> None:
shutil.copytree(self._remote_path(source_dir), target_dir, dirs_exist_ok=True)
@override
async def is_dir(self, path: str, user: str | int | None = None) -> bool:
return self._remote_path(path).is_dir()
@override
async def is_file(self, path: str, user: str | int | None = None) -> bool:
return self._remote_path(path).is_file()
async def _sleep(self, timeout_sec: int | None) -> None:
try:
if timeout_sec is None:
await asyncio.sleep(3600)
else:
await asyncio.wait_for(asyncio.sleep(3600), timeout=timeout_sec)
except TimeoutError as exc:
raise RuntimeError(f"command timed out after {timeout_sec} seconds") from exc
@override
async def exec(
self,
command: str,
cwd: str | None = None,
env: dict[str, str] | None = None,
timeout_sec: int | None = None,
user: str | int | None = None,
) -> ExecResult:
self._require_started()
merged_env = self._merge_env(env)
effective_user = self._resolve_user(user)
self.calls.append(
ExecCall(command, cwd, merged_env, timeout_sec, effective_user)
)
parts = shlex.split(command)
if parts and parts[0] == "sleep-forever":
await self._sleep(timeout_sec)
if len(parts) == 2 and parts[0] == "read":
output = self._remote_path(parts[1]).read_text()
result = ExecResult(stdout=output, stderr=None, return_code=0)
elif len(parts) == 2 and parts[0] == "exit":
result = ExecResult(
stdout=None,
stderr="injected failure",
return_code=int(parts[1]),
)
else:
result = ExecResult(stdout="ok\n", stderr=None, return_code=0)
callback = self._output_callback()
if callback is not None and result.stdout:
await callback(result.stdout, "stdout")
return result
这个替身有意把所有 capability 保持为 false。它不支持 GPU、Windows、Compose、资源 limit/request、断网或 allowlist；只要 Task 请求其中某项，基类就应在 start() 前拒绝。路径解析也把目标限制在 fake root 内，避免 ../../ 逃逸。
Factory 的公开接入方式是 import path。CLI --env/-e 同时接受内置枚举值与 module.path:ClassName；旧的 --environment-import-path 在 v0.18.0 已是隐藏的 deprecated flag。额外构造参数通过 --ek key=value 进入 EnvironmentConfig.kwargs。23
对于这个 test double，不要直接运行完整 Benchmark。应把上面的类保存为 contract_env.py，Agent 保存为 diag_installed.py，然后用下面的测试先验证契约：
from __future__ import annotations
from pathlib import Path
from unittest.mock import AsyncMock, MagicMock
import pytest
from contract_env import ContractTestEnvironment
from diag_installed import DiagnosticAuthenticationError, DiagnosticCliAgent
from harbor.agents.factory import AgentFactory
from harbor.environments.base import ExecResult
from harbor.environments.factory import EnvironmentFactory
from harbor.models.agent.context import AgentContext
from harbor.models.task.config import EnvironmentConfig, NetworkMode, NetworkPolicy
from harbor.models.trial.config import AgentConfig
from harbor.models.trial.config import EnvironmentConfig as TrialEnvironmentConfig
from harbor.models.trial.paths import TrialPaths
def make_env(tmp_path: Path, **kwargs) -> ContractTestEnvironment:
environment_dir = tmp_path / "environment"
environment_dir.mkdir(exist_ok=True)
trial_paths = TrialPaths(tmp_path / "trial")
trial_paths.mkdir()
config = TrialEnvironmentConfig(
import_path="contract_env:ContractTestEnvironment",
env={"BASE": "persistent", "OVERRIDE": "persistent"},
kwargs=kwargs,
)
env = EnvironmentFactory.create_environment_from_config(
config,
environment_dir=environment_dir,
environment_name="contract",
session_id="trial__env",
trial_paths=trial_paths,
task_env_config=EnvironmentConfig(),
)
assert isinstance(env, ContractTestEnvironment)
return env
@pytest.mark.asyncio
async def test_factory_exec_transfer_and_idempotent_cleanup(tmp_path: Path) -> None:
env = make_env(tmp_path)
await env.start(force_build=False)
source = tmp_path / "source.txt"
source.write_text("evidence\n")
await env.upload_file(source, "/work/evidence.txt")
target = tmp_path / "downloaded.txt"
await env.download_file("/work/evidence.txt", target)
assert target.read_text() == "evidence\n"
env.default_user = "agent"
with env.scoped_exec_env({"OVERRIDE": "scoped"}):
result = await env.exec(
"read /work/evidence.txt",
env={"ONCE": "1", "OVERRIDE": "per-exec"},
)
assert result.stdout == "evidence\n"
assert env.calls[-1].user == "agent"
assert env.calls[-1].env == {
"BASE": "persistent",
"OVERRIDE": "scoped",
"ONCE": "1",
}
root = env._root
await env.stop(delete=False)
assert root is not None and root.exists()
await env.stop(delete=True)
await env.stop(delete=True)
assert root is not None and not root.exists()
@pytest.mark.asyncio
async def test_timeout_and_nonzero_are_distinct(tmp_path: Path) -> None:
env = make_env(tmp_path)
await env.start(False)
with pytest.raises(RuntimeError, match="timed out"):
await env.exec("sleep-forever", timeout_sec=0)
result = await env.exec("exit 7")
assert result.return_code == 7
await env.stop(True)
@pytest.mark.asyncio
async def test_start_failure_rolls_back_partial_resource(tmp_path: Path) -> None:
env = make_env(tmp_path, fail_start=True)
expected_root = env.trial_paths.trial_dir / "contract-sandbox"
with pytest.raises(RuntimeError, match="injected start failure"):
await env.start(False)
assert not expected_root.exists()
await env.stop(True)
def test_capability_rejection_happens_before_start(tmp_path: Path) -> None:
environment_dir = tmp_path / "environment"
environment_dir.mkdir()
trial_paths = TrialPaths(tmp_path / "trial")
trial_paths.mkdir()
with pytest.raises(ValueError, match="no-network"):
ContractTestEnvironment(
environment_dir=environment_dir,
environment_name="contract",
session_id="trial__env",
trial_paths=trial_paths,
task_env_config=EnvironmentConfig(),
network_policy=NetworkPolicy(network_mode=NetworkMode.NO_NETWORK),
)
def test_agent_factory_and_error_mapping(tmp_path: Path) -> None:
agent = AgentFactory.create_agent_from_config(
AgentConfig(
import_path="diag_installed:DiagnosticCliAgent",
kwargs={"max_steps": 3},
),
logs_dir=tmp_path / "agent",
)
assert isinstance(agent, DiagnosticCliAgent)
assert agent.build_cli_flags() == "--max-steps 3"
error = agent._classify_exec_error(
"diagnostic-cli",
ExecResult(stdout=None, stderr="AUTH_DENIED", return_code=2),
)
assert isinstance(error, DiagnosticAuthenticationError)
@pytest.mark.asyncio
async def test_agent_run_populates_context(tmp_path: Path) -> None:
agent = DiagnosticCliAgent(logs_dir=tmp_path / "agent", max_steps=4)
env = MagicMock()
env.exec = AsyncMock(
return_value=ExecResult(stdout="done\n", stderr=None, return_code=0)
)
context = AgentContext()
await agent.run("inspect config", env, context)
assert context.metadata == {"exit_code": 0, "stdout_tail": "done\n"}
assert "--max-steps 4" in env.exec.await_args.kwargs["command"]
从 Harbor v0.18.0 源码根目录运行：
cd /path/to/harbor-v0.18.0
uv run --frozen pytest -c pyproject.toml -q /path/to/test_contract.py
uv run --frozen ruff check /path/to/contract_env.py /path/to/diag_installed.py
uv run --frozen ty check /path/to/contract_env.py /path/to/diag_installed.py
本章实际验证版本还增加了三项：sleep-forever 必须变成 timeout 异常而不是非零退出；exit 7 必须保留 return_code=7；Agent Factory 必须能导入 DiagnosticCliAgent 并把 AUTH_DENIED 映射为具体异常。六项章节测试与 Harbor 针对 BaseEnvironment、Docker、E2B、Daytona、preflight、Agent Factory 和 Installed Agent 的相关测试合并运行，共 470 项，结果为 469 passed、1 skipped；跳过项来自 Harbor 既有 Docker 单元测试条件，不是章节示例跳过。
不要只看通过数：把契约变成门禁矩阵
一个 Provider 的测试报告如果只有“start、exec、stop 通过”，仍然无法说明它适合批量评测。测试矩阵至少要沿“调用阶段 × 失败类型 × 是否已产生远端副作用”展开。
| 阶段 | 必测输入 | 必须观察的证据 |
|---|---|---|
| Factory | import path 正确/错误、未知 kwarg、能力不匹配 | 类类型、构造异常、尚未调用云 create |
| preflight | 无凭据、变量为空、测试凭据存在 | 明确退出；日志不含 secret；无远端资源 |
| start | 首次、并发同定义、force build、创建中取消 | build/cache 选择、id/label、部分资源回收 |
| exec | 成功、非零、timeout、dispatch 前/后断网 | ExecResult 或异常类型、command id、进程终止证据 |
| transfer | 空/大/二进制文件、嵌套目录、缺失源 | checksum、目录相对路径、临时归档清理 |
| stop | delete 两种值、NotFound、Auth denied、重复调用 | 控制面反查、保留原因、第二次调用无新副作用 |
Factory 测试要在任何云 SDK mock 之前运行，因为它能发现两类便宜但致命的错误。其一，capabilities 访问了一个在 super().__init__() 之后才赋值的字段；基类构造期校验网络或 GPU 时就会触发属性错误。其二，Provider 在构造函数里吞掉未知 kwargs，导致用户拼错配置仍继续创建收费资源。对第二类错误，生产实现可以显式列出支持的关键字，或对剩余值做校验；不能因为基类接受 **kwargs 就默认忽略一切。
认证还要区分“Provider 控制面凭据”和“Task/Agent 运行时凭据”。E2B_API_KEY、DAYTONA_API_KEY 用于宿主侧创建与删除沙箱，不应因为 Agent 也运行在沙箱里就自动塞进 task_env_config.env。后者会经过 Provider 的 persistent env 合并，可能被容器内命令读取。只有任务确实需要调用同一控制面、并且威胁模型允许时，才通过明确的 Agent env 或 Provider secret 机制授予最小权限。Daytona 的 secrets 又是该 Provider direct 模式的特性，源码明确拒绝把它用于 DinD/Compose；不能把这套注入方式写成 BaseEnvironment 的通用能力。20
并发创建测试不能只断言“最终有一个可用资源”。要记录每次 create 的 idempotency key 或内容哈希、远端返回 id、局部状态转移和 delete 次数。两个 Trial 同时看到 cache miss 时，正确结果可能是一个构建、另一个等待，也可能是平台接受两个独立 sandbox；取决于 Provider 的缓存层。Daytona 的自动 snapshot 在进程内按 snapshot name 加锁，E2B 对 template alias 的存在检查与构建采用自己的重试路径；这两个实现不能推导出“跨进程全局只构建一次”。2118
命令失败则要保留三分法。程序正常启动后返回 7，应保留 return_code=7，让 Installed Agent 再按输出分类；命令超过预算，Provider 应终止或追踪远端进程并抛出清楚的 timeout；HTTP 在 dispatch 后断开时，不能自动再发同一命令。把三者都包装成 RuntimeError("exec failed") 会丢掉 retry 决策所需的信息。反过来，直接把云 SDK 的庞大异常层次暴露给 Trial 也会让调用方耦合 Provider；更稳妥的做法是在保留 raise ... from exc 因果链的同时，映射少量稳定的 Harbor/Provider 异常。
文件传输门禁应比较内容哈希，而不只是文件名存在。目录测试同时保存相对路径集合，才能发现“把 source 目录本身复制进去”和“只复制 source 内容”的一层偏差。若实现借助远端 tar，成功、下载失败和取消三个路径都要删除临时归档；删除失败可告警，但不能覆盖原始 transfer 异常。Harbor 基类的带 exclusion 下载就是在 finally 语义下尝试清理 transfer 文件，这可以作为扩展实现的参考，而不是要求所有 Provider 必须使用 tar。3
最后，为云端测试建立资源账本。每个测试在 create 成功后立即登记 provider、resource id、session id、context id、创建时间和预期删除时间；测试结束用控制面 list/get 反查，再从账本移除。测试进程崩溃时，独立 janitor 按标签和 TTL 清理。由此可以推断：stop() 的单元测试证明代码发出了删除请求，控制面反查才证明账户里没有遗留资源。两者缺一不可，但后者属于集成测试，本章没有执行。
26.6 从替身迁移到真实 Provider
把 test double 替换为云 SDK 时，按下面顺序推进，而不是先让 happy path 跑通再补清理：
- 认证与 preflight：只检查不会产生费用的必要条件；不要把 key 打进日志。若 preflight 只验证“变量存在”，正文和错误信息就不要宣称“凭据有效”。
- 能力表：先全 false，每实现并测试一种能力再打开。Compose 能力同时意味着 sidecar 的 exec/copy/stop 契约，不能只表示“能解析 YAML”。9
- 创建状态机：为每个远端 create 设计 idempotency key、句柄落盘或取消恢复策略。
session_id适合作为可读标签，context_id才是耐久关联标识；不要假设可读名称全局唯一。8 - 命令语义：分别测试成功、非零、timeout、取消和 transport error。只重试能证明未 dispatch 的请求；有副作用命令不能盲目 replay。
- 文件语义：二进制与目录树双向传输，检查权限、符号链接、覆盖与缺失源；云 SDK 单文件 API 与 tar 流不能表现出两种不同目录语义。
- 网络与资源：验证 Provider 实际收到的 request/limit 和规则；“SDK 接受参数”不是“平台已执行策略”的充分证据。
- 清理审计：
start()每个失败点后调用stop(True)；再调用第二次。远端 NotFound 应作为已删除处理，Auth/Permission 应记录为未删除告警。
生产验收不能只依赖 mock。先在隔离测试账户运行一个最小 Task，记录创建出的 resource id；注入 setup 失败、Agent timeout 和进程取消；最后从 Provider 控制面或 list API 反查 resource id 是否仍存在。涉及 delete=False 的调试运行必须有 TTL 或单独的清理作业。这个验收会产生真实云资源与潜在费用，因此本章没有替读者执行。
26.7 失败模式与排查顺序
“配置写了 no-network，Agent 仍能联网”
先看 Provider 的 capabilities，再看本次实例的模式。Docker 的网络能力取决于 Linux egress sidecar 是否在启动前启用；Daytona direct 与 DinD 的 allowlist 能力不同。若 Provider 不支持，正确行为是在创建前拒绝，而不是降级为 public。随后检查 phase policy 是否调用了 set_network_policy()，以及 _apply_network_policy() 是否真的更新远端规则。
“timeout 了，命令却继续修改文件”
区分 SDK request timeout、Harbor command timeout 和远端进程 lifetime。Docker 的实现会 terminate/kill 本地 compose 子进程；云 Provider 必须核对 SDK 的 timeout 是否也终止 sandbox 内进程。若不能证明，就应保存 command/session id，在 timeout handler 中显式 cancel，并用无副作用探针验证进程消失。
“stop 成功返回，但资源还在”
检查 delete 的语义和日志，不要只看 Python 没抛异常。E2B 忽略保留意图并 kill；Daytona direct 在 delete=False 时保留；Daytona delete 遇到权限问题会改为 stop 并警告仍在账户。1813 为每个 Provider 写“list by label/id”的清理后断言，而不是跨 Provider 统一假定。
“自定义类能 import，Job 创建时才崩”
依次检查构造签名是否接受 Factory 传来的标准参数与 **kwargs，Provider 特有状态是否在 super().__init__() 之前初始化，capabilities 是否在基类构造校验时就可访问，以及 EnvironmentConfig.kwargs 的名字/类型是否准确。Factory 会把 override、env、extra compose 与自定义 kwargs 合并后传入；参数冲突会在远端资源创建前暴露。4
26.8 本章小结
- Agent 负责任务策略，Environment 负责执行边界；不要让任一方直接吞并另一方职责。
- Installed Agent 的重点是安装、用户、flags、凭据传递与错误映射，不只是拼一条命令。
capabilities是拒绝不安全配置的可执行契约；不确定就保持 false。- Docker、E2B、Daytona 的资源、网络、快照、Compose 与删除语义各不相同，不能用“云 Provider”一词抹平。
- Provider 的完成标准包括创建取消、timeout、部分失败、重复 stop 和清理后反查。
- 契约替身可以证明 Harbor 集成面，但不能证明真实隔离、云性能、账单或平台策略执行。
26.9 练习
- 给
ContractTestEnvironment增加目录上传/下载测试，覆盖空目录、二进制文件和../路径穿越；解释每个失败应由哪一层抛出。 - 为
DiagnosticCliAgent增加一个布尔CliFlag与一个EnvVar，测试 kwargs、宿主环境 fallback 和 default 的优先级；测试值非法时在构造阶段失败。 - 把 test double 的
resource_capabilities()改成只支持 CPU request，分别构造ResourceMode.REQUEST、LIMIT与GUARANTEE，记录基类拒绝结果。 - 选择一个真实云 Provider，在测试账户中设计取消注入点与清理后 list 断言。只写实验计划和预算上限，不实际创建资源。
- 比较 E2B 和 Daytona 对 allowlist 条目类型的声明，为 hostname、IPv4 literal、IPv4 CIDR、IPv6 literal 各写一个构造测试，禁止根据一个 Provider 的结果推断另一个。