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:

src/agentcompass/environments/
├── __init__.py
└── example_local.py

Implement example_local.py as follows:

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:

from .example_local import ExampleLocalEnvironment

Then inspect registry discovery and the live config schema:

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:

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:

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. For image lifecycle, command execution, transfer, and enforceable network behavior in a container provider, compare docker.py.