> ## Documentation Index
> Fetch the complete documentation index at: https://agent-compass.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
        self.default_workspace_root = delegate.default_workspace_root

    async def exec(
        self,
        command: list[str] | str,
        *,
        shell: bool = False,
        cwd: str | None = None,
        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=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):
    """User-facing parameters for the tutorial provider."""

    workspace: str = config_field(
        default=".agentcompass/example-local",
        description="Host directory used as the task workspace.",
    )
    default_workspace_root: str = config_field(
        default="workspace/",
        description="Default relative workspace exposed to Harnesses.",
    )

    def __post_init__(self) -> None:
        self.workspace = str(self.workspace or ".agentcompass/example-local")
        self.default_workspace_root = str(self.default_workspace_root or "workspace/")


@ENVIRONMENTS.register()
class ExampleLocalEnvironment(BaseEnvironment):
    id = "example_local"
    description = "Local Environment wrapper used by the developer tutorial."
    config_class = ExampleLocalConfig
    default_workspace_root = "workspace/"
    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")

        workspace = Path(config.workspace).expanduser().resolve()
        await asyncio.to_thread(workspace.mkdir, parents=True, exist_ok=True)
        self.default_workspace_root = config.default_workspace_root
        delegate = HostProcessSession(
            workspace=str(workspace),
            default_workspace_root=config.default_workspace_root,
        )
        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`。

## 导出并检查注册

在 `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`；第二条命令应列出 `workspace` 和 `default_workspace_root` 的默认值与描述。如果可选 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 '{"workspace":".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；还应确认存在一个 `details/*.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 仅提供阻塞调用时，才需要显式隔离，避免在高任务并发下阻塞事件循环。

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

`open()` 应先根据解析后的执行计划构建并验证 provider 配置，再解析互斥选择器，应用资源、工作区、标签和基线阶段网络策略，并在启动超时内创建 sandbox。只有 provider 报告资源可用后，才能构造会话。任何步骤失败时，都要先释放已经创建的部分资源，再向上抛出错误。

`close()` 只能停止或删除明确属于当前会话的资源。清理逻辑必须能处理部分启动；即使操作被取消或重复调用，也要保持幂等。绝不能通过宽泛的名称或未经验证的全局搜索来确定清理目标。

根据实际的强制执行能力声明 `supported_network_modes`、`supported_allowlist_entry_types`、`supports_network_target_ports` 和 `supports_dynamic_network_policy`。无法强制某个模式、目标类型或端口限制时必须默认拒绝。不要声明仅靠提示词、环境变量或要求 agent 尽力遵守就能实现某项限制。

支持动态策略切换的 provider 必须能够从基线策略切换到运行策略；复用 Environment 进行评测时，还要能从运行策略直接切换到评测策略。代理凭证、策略令牌、签名 URL 和临时端点都必须脱敏。正常关闭或启动失败后，还要移除临时网络配置和策略。

## 保留配置与 Recipe 优先级

为公开 provider 的每项设置定义类型化配置字段。身份验证、sandbox 来源、生命周期超时、资源、工作区和 provider 元数据必须分开配置，并明确单位、默认值和验证规则；互斥字段同时出现时，应报告清楚的错误。凭证不得进入日志或持久化计划。

Environment 代码使用最终计划，Recipe 提供 Benchmark 专属默认值。二者都必须遵循以下优先级：

```text theme={"system"}
显式的 provider 原生选择器
  > 显式 Environment 镜像
  > 任务元数据
  > 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)。