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

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

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:

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