> ## 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.

# Code Implementation

Implement an Environment provider through the shared session contract: build typed provider config from the resolved plan, create one task Environment, expose command and file primitives, and release the exact resource you created.

The tutorial provider below wraps the public local-process session so every required method is executable without an external account. Replace each delegation with the official provider SDK when building a remote integration; do not add provider behavior to a Benchmark or Harness.

## Record the Provider Contract

Use the provider's official SDK and API documentation as the source of truth. Record authentication and account scope; mutually exclusive image, snapshot, or template selectors; workspace persistence; CPU, memory, disk, GPU, placement, and quotas; startup and deletion semantics; enforceable network modes; command, transfer, endpoint, cancellation, and error behavior; and async or thread-safety guarantees.

## Create the Minimal File

Start with one provider module and one package export:

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

Implement `example_local.py` as follows:

```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
```

This shows every abstract `EnvironmentSession` method and both abstract `BaseEnvironment` methods. There is no separate public `EnvironmentPlan` type: providers consume the Recipe-adjusted `ExecutionPlan`, and `build_config(req, plan)` reads `plan.environment.params` while validating the resolved network phases.

The wrapper is only a contract exercise. A production provider should call its own SDK and return its own `EnvironmentSession`; it should not depend on `HostProcessSession`.

The public schema combines shared `EnvironmentSpec` fields with declared provider-specific options from `config_class`. CLI discovery, YAML/SDK validation, and adapter construction use the same schema. Shared mapping targets (`setup_image_parameter`, `setup_workdir_parameter`, `setup_timeout_parameter`, and `env_variables_param`) are excluded from provider options automatically. Remove fields that are wholly derived by AgentCompass; mark other generated or internal adapter state with `config_field(public=False, ...)`. Unrecognized keys still fail. Provider options remain in `EnvironmentSpec.params` and are copied into adapter configuration before the shared fields are mapped; never add a native alias for a shared setting.

## Export and Inspect the Registration

Add the import to `src/agentcompass/environments/__init__.py`:

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

Then inspect registry discovery and the live config schema:

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

The first command should contain `example_local`. The second should list `setup.workdir` with their defaults and descriptions. If importing an optional provider SDK can fail, guard only its documented missing dependency in `__init__.py`; do not swallow unrelated exceptions or registration errors.

## Run One Task

Use the companion Benchmark and Harness tutorial components to exercise provider open, session construction, and close without external credentials:

```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
```

The terminal result should report one completed task and `paths.run_info`; its parent directory is the run directory. In `run_info.json`, confirm that `request` → `environment` → `id` is `example_local` and that `resolved_execution_plans` contains the same Environment ID for attempt `1`. Also confirm that task and attempt `result.json` files file and `summary.md` exist. The `.agentcompass/environment-smoke` directory confirms `open()` used the provider config; it is not a result directory.

For a remote provider, add one session-level check that runs a list-form command, writes and reads UTF-8 text, uploads and downloads one file and one directory, and verifies cleanup in the provider console. A successful registry or local mock check does not prove remote lifecycle or network enforcement.

## Map the Real Session Primitives

Implement the methods with these semantics:

| Method | Required behavior |
| - | - |
| `exec()` | Run list-form commands without a shell, or string commands only with `shell=True`; preserve return code, stdout, stderr, and timeout |
| `upload()` / `download()` | Transfer one file to or from the active Environment |
| `write_text()` / `read_text()` | Perform deterministic UTF-8 text I/O with actionable missing-path errors |
| `upload_dir()` / `download_dir()` | Transfer complete directory trees without silently changing the requested root |
| `endpoint()` | Return an externally reachable endpoint when supported, otherwise `None` |
| `set_network_policy()` | Apply a new enforceable policy only when dynamic switching is supported |

Normalize provider responses into `ExecResult`. A command's nonzero return code is data, not a provider exception; raise only when transport or provider execution itself fails. Preserve timeout versus provider-error meaning. Use async SDK methods when available, and explicitly isolate blocking calls so high task concurrency does not block the event loop.

## Own Open, Close, and Partial Cleanup

Task-level image and startup requirements arrive in `plan.environment.setup`, an `EnvironmentSetup` value. Declare `setup_image_parameter` for a registry-image parameter. Providers without registry-image support leave the parameter unset and reject image requirements. If an SDK also accepts a native startup deadline, name it with `setup_timeout_parameter`; `build_config()` maps the resolved shared field to that native parameter. Native aliases must not be accepted as user or Recipe input.

`EnvironmentSetup` lives in `agentcompass.runtime.setup` and also carries `workdir`, the default command directory. Declare `setup_workdir_parameter` when the provider has a native equivalent; Docker, HostProcess, and Modal all map it to their adapter `workdir`. After allocation, `BaseEnvironment.open()` ensures this directory exists and sets the returned session's `default_workdir`. Command adapters use `with_exec_workdir` from `agentcompass.environments.utils.exec` to resolve `cwd` before passing it to the SDK. Explicit command directories take precedence, and an unspecified workdir preserves the image or provider default. Environments do not select task workspaces: Benchmarks set `PreparedTask.input.workspace`. The shared `resolve_task_workspace` helper in `agentcompass.utils.workspace` preserves absolute workspace paths, resolves relative paths against the Environment's actual `pwd`, and uses that directory for an empty workspace. Harnesses use the same helper without allocating a replacement task directory; temporary configuration and logs are isolated separately. OpenEvolve still requires an explicit workspace because its program-evolution protocol needs benchmark materials.

Declare `supported_operating_systems` from the provider's real execution capability. `EnvironmentSetup.os` is a requirement checked before allocation, not an image conversion or a new OS backend. Current container providers accept Linux requirements only; `host_process` accepts Linux only on a Linux host. An unspecified OS preserves existing behavior.

For task environment bindings, declare `supports_task_env` and call `self._merge_exec_env(env, required_env=required_env)` before every provider execution path, including shell and detached commands. Merge command defaults, provider bindings, common task bindings, then active phase bindings; explicit user bindings always take precedence over command defaults. The runtime scopes run bindings to Harness setup/execution and evaluation bindings to the evaluator; other lifecycle operations use common bindings. The merge then injects missing `required_env` values, accepts identical values, and rejects conflicts before invoking the provider, naming only the conflicting keys. Forward `required_env` through session wrappers as well. These are internal adapter requirements, not user configuration fields. Do not resolve their literal values as host references. `BaseEnvironment` resolves public bindings without modifying host `os.environ` and binds common values to the opened session. Do not serialize resolved values back into plans or log them.

Set `env_variables_param` to the provider-native common environment mapping, and advertise `supports_startup_env` only if those common bindings can reach the image entrypoint. Providers attaching to an existing sandbox should override `can_inject_startup_env()` accordingly. `require_startup_env` requires this capability for common bindings; run-only and evaluation-only bindings remain command-scoped. An execution wrapper that creates another process boundary must forward declared variable names as well. Do not forward the launcher's entire environment to a remote sandbox.

The shared artifact implementation uses `exec()`, `upload()`, and `download()` with Linux `tar`, `mktemp`, `stat`, and standard shell primitives. Implement file transfers without truncation. Unsupported tools and failed transfers must propagate errors instead of reporting successful collection. Collection/restore has its own typed byte, entry-count, and time limits, separate from agent and verifier deadlines.

`BaseEnvironment` enforces `build_timeout_seconds` around `open()`, after the global rate-limit queue. This does not change sandbox lifetime or command timeouts. Cancellation must propagate after cleanup: include `CancelledError` in partial-start cleanup paths, preserve known resource IDs, and never convert cancellation into a retryable setup error. Cleanup may take additional time after cancellation.

During `open()`, build and validate provider config from the resolved plan; resolve mutually exclusive selectors; apply resources, the default command directory, labels, and baseline network policy; create the sandbox within the startup timeout; and construct a session only after the provider reports a usable state. If any step fails, release every partially created resource before propagating the error.

During `close()`, stop or delete the exact resource owned by that session. Make cleanup safe after partial startup and sufficiently idempotent for cancellation or repeated error handling. Never discover cleanup targets through broad names or unvalidated global searches.

Declare `supported_network_modes`, `supported_allowlist_entry_types`, `supports_network_target_ports`, and `supports_dynamic_network_policy` from real enforcement capability. Fail closed when a mode, target type, or port restriction cannot be enforced. Do not advertise restrictions implemented only by prompts, environment variables, or best-effort agent instructions.

Dynamic providers must switch from baseline to run policy and, for reused evaluation, directly from run to evaluation policy. Protect and redact proxy credentials, policy tokens, signed URLs, and generated endpoints, and remove temporary networks or policies after normal close and startup failure.

Relative remote paths must use the same base for commands and file operations. Decorate the SDK-facing command method with `with_exec_workdir`; it calls `resolve_workdir`, limits relative-cwd lookup to the command budget, and passes the resolved directory and remaining timeout to the adapter. Apply it outside retry decorators. Native command timeout handling still owns partial output and subprocess cleanup; external cancellation must propagate. Use `await self.resolve_path(path)` for remote upload destinations, download sources, and file reads/writes. The helpers preserve absolute paths, including symlink-sensitive components, and anchor relative paths to the session's actual default cwd. If the transfer tool lexically normalizes `..`, resolve the affected path inside the environment before transferring it. Do not resolve remote paths on the launch host or guess `/` when the provider's default is unknown. Benchmarks should resolve task layout directories before preparing materials and reuse those absolute directories for execution and evaluation; relative output-file specifications are relative to the prepared workspace.

Command-backed directory downloads can use `download_directory` from `agentcompass.environments.utils.files`. It preserves relative hierarchy and empty directories, uses NUL-delimited filenames, and reports listing failures instead of silently producing an incomplete tree. It copies regular files without following symlinks inside the source tree and rejects paths that would escape the local destination.

## Preserve Config and Recipe Precedence

Define one typed config field for every public provider setting. Keep authentication, sandbox source, lifecycle timeouts, resources, workspace, and provider metadata distinct; use clear units, defaults, validation, and mutual-exclusion errors. Credentials must not enter logs or persisted plans.

Environment code consumes the final plan while Recipes supply Benchmark-specific defaults. Both layers preserve:

```text theme={"system"}
explicit shared request settings
  > task setup declarations
  > recipe fallback
```

Resolve the winning selector before removing incompatible fields. Apply resources field by field so explicit user values win while unspecified fields can inherit task hints. A Recipe copies the plan, stays narrow to a Benchmark/provider pair, and never calls the provider SDK.

Respect the process-global provider-open limiter applied by `BaseEnvironment`, plus the provider's SDK request limits, account quotas, and capacity. Log stable sandbox IDs, lifecycle phases, elapsed time, selected non-secret images, and actionable errors; never log full config dictionaries that may contain secrets.

## Diagnose Failures by Stage

| Symptom | Stage | First check |
| - | - | - |
| ID missing from `list env` | Import and registration | `environments/__init__.py`, optional dependency guard, duplicate ID, and traceback |
| `config docs` misses a provider field | Config schema | Common `EnvironmentSpec` / `EnvironmentSetup` fields, provider config fields, and mapping declarations |
| Failure before provider API call | Planning and config | Recipe-resolved selectors, network capabilities, and `build_config()` |
| Orphan after an `open()` error | Partial startup | Resource ID capture and cleanup on every failure branch |
| Nonzero command becomes an exception | Session normalization | Return `ExecResult`; reserve exceptions for transport or provider failure |
| File appears under the wrong root | Transfer semantics | Relative-path resolution and directory-root preservation |
| Baseline succeeds but run phase fails | Network transition | Declared dynamic support and actual `set_network_policy()` enforcement |
| Cancellation leaks a sandbox | Close lifecycle | Exact ownership, cancellation handling, and idempotent cleanup |
| Plan shows one resource but provider created another | Precedence | Selector winner and per-field resource overlay order |

The simplest real reference is [`host_process.py`](https://github.com/open-compass/AgentCompass/blob/main/src/agentcompass/environments/host_process.py). For image lifecycle, command execution, transfer, and enforceable network behavior in a container provider, compare [`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.