> ## Documentation Index
> Fetch the complete documentation index at: https://opencompass-docs-preview-pr-335-0.mintlify.site/llms.txt
> Use this file to discover all available pages before exploring further.

# 代码实现

实现 Environment provider 时，应遵循共享会话契约。它需要根据解析后的执行计划构建类型化 provider 配置，为任务创建 Environment，提供命令与文件原语，并可靠释放自己创建的资源。

下面的教程 provider 基于公开的本地进程会话实现，因此无需外部账号即可执行所有必要方法。接入远程 provider 时，应使用其官方 SDK 替换对应调用；不要把 provider 行为放入 Benchmark 或 Harness。

## 记录 provider 契约

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

## 创建最小文件

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

```text theme={"system"}
src/agentcompass/environments/
├── __init__.py
└── example_local.py
```

按如下方式实现 `example_local.py`：

```python theme={"system"}
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` 中添加导入：

```python theme={"system"}
from .example_local import ExampleLocalEnvironment
```

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

```bash theme={"system"}
uv run agentcompass list env
uv run agentcompass config docs env example_local
```

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

## 运行单个任务

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

```bash theme={"system"}
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 专属默认值。二者都必须遵循以下优先级：

```text theme={"system"}
显式公共请求设置
  > 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`](https://github.com/open-compass/AgentCompass/blob/main/src/agentcompass/environments/host_process.py)。需要查看容器 provider 的镜像生命周期、命令执行、传输和可强制网络行为时，可对照 [`docker.py`](https://github.com/open-compass/AgentCompass/blob/main/src/agentcompass/environments/docker.py)。


This documentation is built and hosted on [Mintlify](https://mintlify.com), a developer documentation platform.