代码实现

实现 Environment provider 时，应遵循共享会话契约。它需要根据解析后的执行计划构建类型化 provider 配置，为任务创建 Environment，提供命令与文件原语，并可靠释放自己创建的资源。下面的教程 provider 基于公开的本地进程会话实现，因此无需外部账号即可执行所有必要方法。接入远程 provider 时，应使用其官方 SDK 替换对应调用；不要把 provider 行为放入 Benchmark 或 Harness。

记录 provider 契约

以 provider 官方 SDK 和 API 文档为权威来源。需要记录身份验证与账号范围，互斥的镜像、快照或模板选择器，工作区持久化方式，CPU、内存、磁盘、GPU、放置策略与配额，启动与删除语义，可强制执行的网络模式，命令、传输、端点、取消与错误行为，以及异步和线程安全保证。

创建最小文件

先创建一个 provider 模块和一个软件包导出：

src/agentcompass/environments/
├── __init__.py
└── example_local.py

按如下方式实现 example_local.py：

from __future__ import annotations

import asyncio
from dataclasses import dataclass
from pathlib import Path
from typing import Any

from agentcompass.environments.host_process import HostProcessSession
from agentcompass.runtime import (
    ENVIRONMENTS,
    BaseEnvironment,
    EnvironmentSession,
    ExecResult,
    ExecutionPlan,
    NetworkMode,
    RunRequest,
)
from agentcompass.runtime.config import RuntimeEnvironmentConfig, config_field


class ExampleLocalSession(EnvironmentSession):
    """Complete session surface backed by the public local implementation."""

    def __init__(self, delegate: HostProcessSession) -> None:
        self._delegate = delegate

    async def exec(
        self,
        command: list[str] | str,
        *,
        shell: bool = False,
        cwd: str | None = None,
        env: dict[str, str] | None = None,
        required_env: dict[str, str] | None = None,
        timeout: float | None = None,
        detach: bool = False,
        flags: dict[str, Any] | None = None,
    ) -> ExecResult:
        return await self._delegate.exec(
            command,
            shell=shell,
            cwd=cwd,
            env=self._merge_exec_env(env, required_env=required_env),
            required_env=required_env,
            timeout=timeout,
            detach=detach,
            flags=flags,
        )

    async def upload(self, src: str, dst: str) -> None:
        await self._delegate.upload(src, dst)

    async def download(self, src: str, dst: str) -> None:
        await self._delegate.download(src, dst)

    async def write_text(self, path: str, content: str) -> None:
        await self._delegate.write_text(path, content)

    async def read_text(self, path: str) -> str:
        return await self._delegate.read_text(path)

    async def upload_dir(self, src: Path | str, dst: str) -> None:
        await self._delegate.upload_dir(src, dst)

    async def download_dir(self, src: str, dst: Path | str) -> None:
        await self._delegate.download_dir(src, dst)

    async def endpoint(self) -> str | None:
        return await self._delegate.endpoint()


@dataclass(slots=True)
class ExampleLocalConfig(RuntimeEnvironmentConfig):
    """Internal adapter defaults for the tutorial provider."""

    workdir: str = config_field(default=".", description="Default command directory.")

    def __post_init__(self) -> None:
        self.workdir = str(self.workdir or ".")


@ENVIRONMENTS.register()
class ExampleLocalEnvironment(BaseEnvironment):
    id = "example_local"
    description = "Local Environment wrapper used by the developer tutorial."
    config_class = ExampleLocalConfig
    setup_workdir_parameter = "workdir"
    supported_network_modes = frozenset({NetworkMode.PUBLIC})
    supports_dynamic_network_policy = False

    async def open(
        self,
        req: RunRequest,
        plan: ExecutionPlan,
    ) -> EnvironmentSession:
        config = self.build_config(req, plan)
        if not isinstance(config, ExampleLocalConfig):
            raise TypeError("example_local requires ExampleLocalConfig")

        workdir = Path(config.workdir).resolve()
        await asyncio.to_thread(workdir.mkdir, parents=True, exist_ok=True)
        delegate = HostProcessSession(workdir=str(workdir))
        return ExampleLocalSession(delegate)

    async def close(self, env: EnvironmentSession) -> None:
        _ = env

以上代码展示了 EnvironmentSession 的每个抽象方法和 BaseEnvironment 的两个抽象方法。这里没有独立的公开 EnvironmentPlan 类型：provider 使用经过 Recipe 调整的 ExecutionPlan，build_config(req, plan) 从 plan.environment.params 读取配置并验证解析后的网络阶段。这个包装器只用于练习契约。生产 provider 应调用自己的 SDK 并返回自己的 EnvironmentSession，不应依赖 HostProcessSession。公开 schema 由公共 EnvironmentSpec 字段和 config_class 声明的 provider 专属配置组成，CLI 展示、YAML/SDK 校验和适配器构造使用同一份定义。通用映射目标（setup_image_parameter、setup_workdir_parameter、setup_timeout_parameter、env_variables_param）自动从专属参数中排除；完全由 AgentCompass 推导的字段应删除，其他生成字段或内部适配器状态通过 config_field(public=False, ...) 标记。未声明的键仍然报错。专属配置保留在 EnvironmentSpec.params，构造适配器时先复制这些配置，再映射统一字段，不能为统一字段增加原生别名入口。

导出并检查注册

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

from .example_local import ExampleLocalEnvironment

然后检查注册表发现和当前配置结构：

uv run agentcompass list env
uv run agentcompass config docs env example_local

第一条命令的输出应包含 example_local；第二条命令应列出 setup.workdir 的默认值与描述。如果可选 provider SDK 可能缺失，只能在 __init__.py 中捕获并处理已明确记录的依赖缺失错误；不要吞掉无关异常或注册错误。

运行单个任务

使用配套的 Benchmark 与 Harness 教程组件，在没有外部凭证的情况下执行 provider 打开、会话构建与关闭：

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

命令应报告一个已完成任务并输出 paths.run_info，该路径的父目录就是本次运行目录。在 run_info.json 中，确认 request → environment → id 为 example_local，并且 resolved_execution_plans 的第 1 次任务尝试包含相同的 Environment ID；还应确认存在task 和 attempt 的 result.json 文件和 summary.md。.agentcompass/environment-smoke 目录可以证明 open() 使用了 provider 配置；它不是结果目录。远程 provider 还应接受会话级检查，包括执行一条列表形式命令、写入并读回 UTF-8 文本，以及上传并下载单个文件和目录。随后还要在 provider 控制台确认资源已经清理。注册表或本地模拟检查成功，不能证明远程生命周期正确，也不能证明网络策略已得到强制执行。

映射真实会话原语

按以下语义实现各方法：

方法	必要行为
`exec()`	列表形式命令不经过命令解释器；字符串命令只允许显式使用 `shell=True`；保留返回码、标准输出、标准错误和超时
`upload()` / `download()`	向活动 Environment 传入或取回一个文件
`write_text()` / `read_text()`	执行确定性 UTF-8 文本 I/O，并在路径缺失时给出清晰的错误信息
`upload_dir()` / `download_dir()`	传输完整目录树，不能静默改变请求根目录
`endpoint()`	支持服务时返回外部可访问端点，否则返回 `None`
`set_network_policy()`	只有支持动态切换时才应用新的可强制策略

将 provider 响应标准化为 ExecResult。命令的非零返回码应作为结果数据返回，而不是作为 provider 异常抛出；只有传输失败或 provider 本身无法执行操作时才抛出异常。超时和 provider 错误也必须保持可区分。如果 SDK 提供异步接口，应直接使用；只有 SDK 仅提供阻塞调用时，才需要显式隔离，避免在高任务并发下阻塞事件循环。

处理启动、关闭与部分启动清理

任务级镜像和启动要求通过 plan.environment.setup 传入，其类型为 EnvironmentSetup。用 setup_image_parameter 声明 registry 镜像参数名。不支持 registry 镜像的 provider 保持该参数未设置，并拒绝镜像要求。如果 SDK 还提供原生启动超时，通过 setup_timeout_parameter 声明参数名；build_config() 将最终通用字段映射到该原生参数；不能再把原生别名作为用户或 Recipe 输入。 EnvironmentSetup 位于 agentcompass.runtime.setup，还包含默认命令目录 workdir。provider 有原生对应参数时，通过 setup_workdir_parameter 声明；Docker、HostProcess 和 Modal 都将它映射到 adapter 的 workdir。分配完成后，BaseEnvironment.open() 会确保该目录存在，并设置返回 session 的 default_workdir。命令适配器通过 agentcompass.environments.utils.exec 中的 with_exec_workdir 解析 cwd，再传给 SDK。显式命令目录优先，未指定 workdir 时保留镜像或 provider 的默认行为。Environment 不负责选择任务 workspace：Benchmark 写入 PreparedTask.input.workspace。agentcompass.utils.workspace 中的通用函数 resolve_task_workspace 保留绝对 workspace 路径，将相对路径基于 Environment 的实际 pwd 解析，空值则使用该目录。Harness 使用同一个函数，不再另外分配任务目录；临时配置和日志独立隔离。OpenEvolve 的程序演化协议依赖 Benchmark 准备的材料，因此仍要求显式 workspace。根据实际执行能力声明 supported_operating_systems。EnvironmentSetup.os 是分配前校验的要求，不会转换镜像或新增 OS backend。现有容器 provider 只接受 Linux 要求；host_process 仅在 Linux 宿主机上接受 Linux。未声明 OS 时保留原有行为。支持任务环境变量时声明 supports_task_env，并在每条 provider 执行路径前调用 self._merge_exec_env(env, required_env=required_env)，包括 shell 和 detached 命令。按命令默认值、provider 变量、任务公共变量、当前阶段变量的顺序合并，用户显式配置优先于命令默认值。runtime 只在 Harness 初始化和执行时绑定 run 变量，只在评测器执行时绑定 evaluation 变量；其他生命周期操作使用公共变量。合并后补入缺失的 required_env，接受相同值，在调用 provider 前拒绝冲突，报错只列出变量名。session 包装层也必须转发 required_env。它是适配器内部约束，不是用户配置字段，其字面值不作宿主变量引用解析。BaseEnvironment 解析公共配置时不会修改宿主 os.environ，并为打开的 session 绑定公共变量。不要将解析后的值写回计划或打印到日志。通过 env_variables_param 指定 provider 原生的公共环境变量参数。只有这些公共变量能到达镜像 entrypoint 时，才声明 supports_startup_env；连接已有 sandbox 的 provider 应相应覆盖 can_inject_startup_env()。require_startup_env 要求公共变量具备该启动注入能力，运行和评测阶段专用变量仍然只用于命令。若执行包装器跨越另一层进程边界，也需要按名称转发声明的变量，不要把启动进程的全部环境变量传入远程 sandbox。公共产物实现通过 exec()、upload() 和 download()，配合 Linux 的 tar、mktemp、stat 及标准 shell 工具完成传输。文件传输不得截断；缺少工具或传输失败时必须报错，不能报告采集成功。采集和恢复使用独立的类型化字节数、条目数和时间限制，不占用 agent 或 verifier 的阶段 deadline。 BaseEnvironment 在全局创建速率限制的排队结束后，使用 build_timeout_seconds 限制 open()。这不会修改 sandbox 生命周期或单命令超时。取消必须在清理后继续向上传播：部分启动清理路径要处理 CancelledError，保留已知资源 ID，不要将取消转换为可重试的 setup error。取消后的清理可能需要额外时间。 open() 应先根据解析后的执行计划构建并验证 provider 配置，再解析互斥选择器，应用资源、工作区、标签和基线阶段网络策略，并在启动超时内创建 sandbox。只有 provider 报告资源可用后，才能构造会话。任何步骤失败时，都要先释放已经创建的部分资源，再向上抛出错误。 close() 只能停止或删除明确属于当前会话的资源。清理逻辑必须能处理部分启动；即使操作被取消或重复调用，也要保持幂等。绝不能通过宽泛的名称或未经验证的全局搜索来确定清理目标。根据实际的强制执行能力声明 supported_network_modes、supported_allowlist_entry_types、supports_network_target_ports 和 supports_dynamic_network_policy。无法强制某个模式、目标类型或端口限制时必须默认拒绝。不要声明仅靠提示词、环境变量或要求 agent 尽力遵守就能实现某项限制。支持动态策略切换的 provider 必须能够从基线策略切换到运行策略；复用 Environment 进行评测时，还要能从运行策略直接切换到评测策略。代理凭证、策略令牌、签名 URL 和临时端点都必须脱敏。正常关闭或启动失败后，还要移除临时网络配置和策略。相对远端路径必须在命令和文件操作中使用相同基准。在直接调用 SDK 的命令方法上添加 with_exec_workdir 装饰器；它调用 resolve_workdir，让相对 cwd 查询受命令预算约束，并将解析后的目录与剩余超时传给适配器。该装饰器应放在重试装饰器外层。原生命令超时处理仍负责部分输出和子进程清理；外层取消必须继续传播。远端上传目标、下载源以及文件读写路径使用 await self.resolve_path(path)。这些函数保留绝对路径及其涉及符号链接的路径分量，将相对路径基于 session 的实际默认 cwd 解析。如果传输工具会按字符串规则折叠 ..，必须先在 Environment 内解析受影响的路径再传输。不要在启动 host 上解析远端路径，也不要在 provider 默认目录未知时猜测为 /。Benchmark 应在准备材料前解析任务布局目录，并在执行与评测时复用这些绝对目录；输出文件声明中的相对路径以准备好的 workspace 为基准。基于命令遍历目录的适配器可使用 agentcompass.environments.utils.files 中的 download_directory。它保留相对目录层级和空目录，使用 NUL 分隔文件名，并在遍历失败时明确报错，不会静默生成不完整的目录树。它复制普通文件，不跟随源目录树内的符号链接，并拒绝会越出本地目标目录的路径。

保留配置与 Recipe 优先级

SDK 凭证和部署配置通过 provider 声明的专属参数传入；镜像、资源、网络策略和阶段环境变量仍由通用 EnvironmentSpec 管理。专属参数不会绕过统一字段的校验。 Environment 代码使用最终计划，Recipe 提供 Benchmark 专属默认值。二者都必须遵循以下优先级：

显式公共请求设置
  > Task setup 声明
  > Recipe 回退默认值

先确定优先级最高的选择器，再移除与它不兼容的字段。资源配置应逐字段合并：用户显式传入的值优先，未指定的字段可以继承任务提示。Recipe 应复制执行计划，只匹配范围明确的 Benchmark/provider 组合，并且绝不能调用 provider SDK。除了 BaseEnvironment 提供的进程级全局 provider 启动限流，还要遵守 provider SDK 的请求限制、账号配额和容量限制。日志应记录稳定的 sandbox ID、生命周期阶段、耗时、所选的非敏感镜像信息和便于处理的错误信息，但绝不能记录可能包含密钥的完整配置字典。

按阶段诊断失败

现象	阶段	首先检查
`list env` 中没有 ID	导入与注册	`environments/__init__.py`、可选依赖保护、重复 ID 和堆栈
`config docs` 缺少 provider 字段	配置结构	`config_class`、`config_field()`、单位、默认值和数据类验证
调用 provider API 前失败	计划与配置	Recipe 解析后的选择器、网络能力和 `build_config()`
`open()` 出错后留下孤儿资源	部分启动	每条失败分支是否保存资源 ID 并执行清理
命令非零返回变成异常	会话标准化	返回 `ExecResult`；只为传输或 provider 失败保留异常
文件出现在错误根目录	传输语义	相对路径解析和目录根保留
基线阶段成功但运行阶段失败	网络切换	声明的动态支持和真实 `set_network_policy()` 强制执行
取消后泄漏 sandbox	关闭生命周期	明确清理责任，并正确处理取消和重复清理
计划记录一种资源但 provider 创建了另一种	优先级	选择器胜出项和逐字段资源覆盖顺序

最简单的真实参考实现是 host_process.py。需要查看容器 provider 的镜像生命周期、命令执行、传输和可强制网络行为时，可对照 docker.py。

架构

扩展组件

参与贡献

记录 provider 契约

创建最小文件

导出并检查注册

运行单个任务

映射真实会话原语

处理启动、关闭与部分启动清理

保留配置与 Recipe 优先级

按阶段诊断失败

​记录 provider 契约

​创建最小文件

​导出并检查注册

​运行单个任务

​映射真实会话原语

​处理启动、关闭与部分启动清理

​保留配置与 Recipe 优先级

​按阶段诊断失败

记录 provider 契约

创建最小文件

导出并检查注册

运行单个任务

映射真实会话原语

处理启动、关闭与部分启动清理

保留配置与 Recipe 优先级

按阶段诊断失败