代码实现

实现 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
        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 中添加导入：

from .example_local import ExampleLocalEnvironment

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

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 打开、会话构建与关闭：

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 专属默认值。二者都必须遵循以下优先级：

显式的 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。需要查看容器 provider 的镜像生命周期、命令执行、传输和可强制网络行为时，可对照 docker.py。