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

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

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

## 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 `workspace` and `default_workspace_root` 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 '{"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
```

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 one `details/*.json` 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

During `open()`, build and validate provider config from the resolved plan; resolve mutually exclusive selectors; apply resources, workspace, 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.

## 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 provider-native selector
  > explicit environment image
  > task metadata
  > 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           | `config_class`, `config_field()`, units, defaults, and dataclass validation        |
| 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).