Developer Guide - AgentCompass Documentation

This guide explains how to extend AgentCompass without crossing runtime boundaries.

Add a Benchmark

A benchmark owns the dataset contract, task preparation, evaluation logic, and aggregate metrics. It should not own model-provider calls, agent loops, sandbox lifecycle, or provider-specific image selection. Add a benchmark when you are introducing a new task family or a new scoring protocol. If the change only adapts an existing benchmark to a new sandbox image or workspace layout, add a recipe instead.

Benchmark File

Create a module under src/agentcompass/benchmarks/. Most benchmarks define a RuntimeBenchmarkConfig subclass, an optional BenchmarkPlan, and a BaseBenchmark subclass.

from dataclasses import dataclass

from agentcompass.benchmarks.config import RuntimeBenchmarkConfig
from agentcompass.runtime.base import BaseBenchmark, EnvironmentSession
from agentcompass.runtime.models import (
    BenchmarkPlan,
    EnvironmentSpec,
    ExecutionPlan,
    PreparedTask,
    RunRequest,
    RunResult,
    TaskInput,
    TaskOutput,
    TaskSpec,
    TaskStatus,
)
from agentcompass.runtime.registry import BENCHMARKS


@dataclass(slots=True)
class MyBenchmarkConfig(RuntimeBenchmarkConfig):
    dataset_path: str = "data/my_benchmark/tasks.jsonl"


@dataclass(slots=True)
class MyBenchmarkPlan(BenchmarkPlan):
    workspace_root: str = "workspace/"


@BENCHMARKS.register()
class MyBenchmark(BaseBenchmark):
    id = "my_benchmark"
    description = "My Benchmark: one-sentence task and scoring description (paper or project URL)."
    config_class = MyBenchmarkConfig
    evaluation_environment_mode = "none"

    def load_tasks(self, req: RunRequest) -> list[TaskSpec]:
        config = self.build_config(req)
        return [
            TaskSpec(
                task_id="sample-1",
                question="...",
                category="default",
                ground_truth="...",
                metadata={"source": config.dataset_path},
            )
        ]

    def build_plan(self, task: TaskSpec, req: RunRequest, environment: EnvironmentSpec) -> MyBenchmarkPlan:
        return MyBenchmarkPlan()

    async def prepare_task(
        self,
        task: TaskSpec,
        env: EnvironmentSession,
        req: RunRequest,
        plan: MyBenchmarkPlan,
    ) -> PreparedTask:
        return PreparedTask(
            task_id=task.task_id,
            category=task.category,
            ground_truth=task.ground_truth,
            input=TaskInput(prompt=task.question, workspace=plan.workspace_root),
            output=TaskOutput(),
            metadata=dict(task.metadata),
        )

    async def evaluate(
        self,
        task: TaskSpec,
        prepared: PreparedTask,
        result: RunResult,
        req: RunRequest,
        plan: ExecutionPlan,
        env: EnvironmentSession | None = None,
    ) -> RunResult:
        result.correct = result.final_answer == prepared.ground_truth
        result.score = 1.0 if result.correct else 0.0
        result.status = TaskStatus.COMPLETED if not result.error else TaskStatus.EVAL_ERROR
        return result

Benchmark Responsibilities

Hook	Purpose
`id`	Stable CLI and config id. It is also used by recipes, analyzers, result paths, and docs.
`description`	User-facing description exported by `agentcompass list dump`. Benchmarks and harnesses must define a non-empty description.
`config_class`	Parses `--benchmark-params` and config defaults. Extend `RuntimeBenchmarkConfig` when the benchmark needs dataset paths, versions, splits, or sampling controls.
`load_tasks()`	Loads raw tasks and returns `TaskSpec` objects. Keep this deterministic and fail early on missing datasets.
`select_tasks()`	Optional override for custom sampling. The base class already supports `sample_ids`.
`build_plan()`	Produces benchmark-side execution state for one task. Recipes may rewrite this plan later.
`prepare_task()`	Converts a `TaskSpec` into the benchmark-to-harness `PreparedTask` contract. Upload files or write prompts here when needed.
`evaluate()`	Scores a harness `RunResult` and returns the normalized final `RunResult`.
`aggregate_metrics()`	Optional aggregate metric override. The default path aggregates binary correctness.

Use TaskSpec.metadata for source records, image names, docker metadata, expected files, and benchmark-specific IDs. Use PreparedTask.input and PreparedTask.output for the actual harness contract.

Register And Verify

Export the benchmark in src/agentcompass/benchmarks/__init__.py so registry import side effects can discover it:

from .my_benchmark import MyBenchmark  # noqa: F401

Then verify the component list and run a single task:

uv run agentcompass list benchmark

agentcompass run \
  my_benchmark \
  openai_chat \
  your-model \
  --env <env-provider> \
  --benchmark-params '{"sample_ids":["sample-1"]}' \
  --model-base-url "$MODEL_BASE_URL" \
  --model-api-key "$MODEL_API_KEY"

Add a Harness

A harness owns the agent or model execution loop. It consumes PreparedTask, RunRequest.model, and an EnvironmentSession, then returns a normalized RunResult. It should not load datasets, score benchmark correctness, or hard-code provider-specific sandbox images. Add a harness when you are integrating a new agent framework, coding assistant, GUI agent, direct model-call path, or tool-use strategy.

Harness File

Create a module under src/agentcompass/harnesses/. A typical harness defines a user-facing config, a runtime plan, and a BaseHarness subclass.

from dataclasses import dataclass
from typing import Any

from agentcompass.runtime.base import BaseHarness, EnvironmentSession
from agentcompass.runtime.component_config import RuntimeHarnessConfig
from agentcompass.runtime.models import (
    EnvironmentSpec,
    HarnessPlan,
    ModelSpec,
    PreparedTask,
    RunRequest,
    RunResult,
    TaskStatus,
)
from agentcompass.runtime.registry import HARNESSES


@dataclass(slots=True)
class MyHarnessConfig(RuntimeHarnessConfig):
    max_steps: int = 30


@dataclass(slots=True)
class MyHarnessPlan(HarnessPlan):
    max_steps: int = 30


@HARNESSES.register()
class MyHarness(BaseHarness):
    id = "my_harness"
    description = "Runs MyAgent against text, file, or workspace tasks (official website: https://...)."
    config_class = MyHarnessConfig
    plan_class = MyHarnessPlan

    def supports(self, environment: EnvironmentSpec, model: ModelSpec) -> bool:
        return environment.id in {"host_process", "docker", "daytona", "modal"}

    async def start_session(self, env: EnvironmentSession, req: RunRequest, plan: MyHarnessPlan) -> dict[str, Any]:
        return {"env": env, "model": req.model.id, "max_steps": plan.max_steps}

    async def run_task(
        self,
        session: dict[str, Any],
        prepared: PreparedTask,
        req: RunRequest,
        plan: MyHarnessPlan,
    ) -> RunResult:
        return RunResult(
            task_id=prepared.task_id,
            category=prepared.category,
            status=TaskStatus.COMPLETED,
            final_answer="...",
            ground_truth=prepared.ground_truth,
            artifacts={"workspace": prepared.input.workspace},
        )

    async def close_session(self, session: dict[str, Any]) -> None:
        return None

Harness Responsibilities

Hook	Purpose
`id` / `description`	Stable CLI id and exported description. Include the agent/framework website when available.
`supports()`	Validates the environment and model protocol combination before execution starts. Raise a clear error for unsupported protocols.
`config_class`	Parses `--harness-params` and config defaults. Keep public knobs here, not in ad hoc metadata.
`plan_class`	Runtime execution state derived from config. Recipes may inspect or rewrite the final execution plan.
`start_session()`	Initializes clients, agent config files, persistent processes, or per-run state.
`run_task()`	Executes exactly one `PreparedTask` and returns `RunResult`. Preserve trajectory, artifacts, usage, and errors whenever possible.
`close_session()`	Releases external processes, clients, temporary sessions, or remote framework state.

Use prepared.input.workspace, prepared.input.files, prepared.input.media, prepared.input.tools, and prepared.input.messages instead of reaching back into benchmark internals. A harness should be reusable across benchmarks that emit the same prepared-task contract.

Register And Verify

Export the harness in src/agentcompass/harnesses/__init__.py:

from .my_harness import MyHarness  # noqa: F401

Then check discovery and a smoke run:

uv run agentcompass list harness

agentcompass run \
  scicode \
  my_harness \
  your-model \
  --env <env-provider> \
  --benchmark-params '{"sample_ids":["<task-id>"]}' \
  --model-base-url "$MODEL_BASE_URL" \
  --model-api-key "$MODEL_API_KEY"

Add an Environment

An environment provider owns execution primitives: commands, file transfer, text IO, directory sync, endpoint discovery, and teardown. It should not know benchmark scoring rules or agent behavior. Add an environment when AgentCompass needs to run the same benchmark/harness matrix on a new local process, container runtime, cloud sandbox, or remote execution provider.

Environment File

Create a module under src/agentcompass/environments/. Each provider usually has a session class, config dataclass, and BaseEnvironment implementation.

from dataclasses import dataclass
from pathlib import Path
from typing import Any

from agentcompass.runtime.base import BaseEnvironment, EnvironmentSession
from agentcompass.runtime.component_config import RuntimeEnvironmentConfig
from agentcompass.runtime.models import ExecResult, ExecutionPlan, RunRequest
from agentcompass.runtime.registry import ENVIRONMENTS


class MyEnvironmentSession(EnvironmentSession):
    default_workspace_root = "/workspace/"

    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:
        command = self._validate_exec_command(command, shell=shell)
        # Map AgentCompass exec semantics onto the provider SDK here.
        return ExecResult(returncode=0, stdout="", stderr="")

    async def upload(self, src: str, dst: str) -> None: ...
    async def download(self, src: str, dst: str) -> None: ...
    async def write_text(self, path: str, content: str) -> None: ...
    async def read_text(self, path: str) -> str: ...
    async def upload_dir(self, src: Path | str, dst: str) -> None: ...
    async def download_dir(self, src: str, dst: Path | str) -> None: ...
    async def endpoint(self) -> str | None: ...


@dataclass(slots=True)
class MyEnvironmentConfig(RuntimeEnvironmentConfig):
    image: str = "python:3.12-slim"
    default_workspace_root: str = "/workspace/"


@ENVIRONMENTS.register()
class MyEnvironment(BaseEnvironment):
    id = "my_environment"
    config_class = MyEnvironmentConfig
    default_workspace_root = "/workspace/"

    async def open(self, req: RunRequest, plan: ExecutionPlan) -> MyEnvironmentSession:
        config = self.build_config(req, plan)
        return MyEnvironmentSession()

    async def close(self, env: EnvironmentSession) -> None:
        return None

Environment Responsibilities

Surface	Purpose
`EnvironmentSession.exec()`	Implements subprocess-like execution. Respect `shell`, `cwd`, `env`, `timeout`, `detach`, and return `ExecResult` with stdout, stderr, returncode, and timeout status.
File primitives	Implement `upload`, `download`, `write_text`, `read_text`, `upload_dir`, and `download_dir` so benchmark and harness code can stay provider-neutral.
`endpoint()`	Returns a reachable URL when the provider exposes one, otherwise `None`.
`RuntimeEnvironmentConfig`	Documents public provider params such as image, resources, region, credentials, or workspace root.
`BaseEnvironment.open()`	Builds the provider session from the recipe-adjusted `ExecutionPlan`. Always use `plan.environment.params`, not only the original request.
`BaseEnvironment.close()`	Terminates or detaches remote resources deterministically. It should tolerate partially initialized sessions.

Provider defaults belong in config/defaults.yaml when they are useful for normal users. Benchmark-specific image, snapshot, workspace, or resource mapping belongs in recipes.

Register And Verify

Export the environment from src/agentcompass/environments/__init__.py. Guard optional SDK imports if the dependency is not always installed:

try:
    from .my_environment import MyEnvironment  # noqa: F401
except ModuleNotFoundError as exc:
    if exc.name != "my_provider_sdk":
        raise

Then compile the provider and run a minimal command-oriented benchmark or harness smoke test:

python -m compileall src/agentcompass/environments/my_environment.py

agentcompass run \
  terminal_bench_2 \
  terminus2 \
  your-model \
  --env my_environment \
  --benchmark-params '{"sample_ids":["<task-id>"]}' \
  --model-base-url "$MODEL_BASE_URL" \
  --model-api-key "$MODEL_API_KEY"

Add a Recipe

A recipe is a compatibility layer between one benchmark family and one environment provider. It rewrites the per-task ExecutionPlan before the sandbox starts, usually to select images, workspaces, resource requests, snapshots, environment variables, or evaluation layout. Add a recipe when task metadata already contains provider-relevant information and users should not have to pass it manually on the CLI.

Recipe File

Create a module under src/agentcompass/recipes/<benchmark_family>/. Keep one provider recipe per file when the provider behavior is meaningfully different.

from copy import deepcopy

from agentcompass.runtime.base import BaseRecipe
from agentcompass.runtime.models import ExecutionPlan, RunRequest, TaskSpec
from agentcompass.runtime.registry import RECIPES


@RECIPES.register()
class MyBenchmarkModalRecipe(BaseRecipe):
    id = "my_benchmark_modal"
    enabled_by_default = True

    def matches(self, req: RunRequest, task: TaskSpec, plan: ExecutionPlan) -> bool:
        return req.benchmark.id == "my_benchmark" and req.environment.id == "modal"

    def apply(self, plan: ExecutionPlan, req: RunRequest, task: TaskSpec) -> ExecutionPlan:
        _ = req
        updated = deepcopy(plan)
        params = dict(updated.environment.params)

        user_image = str(params.get("image") or "").strip()
        task_image = str(task.metadata.get("docker_image") or "").strip()
        if not user_image and not task_image:
            raise ValueError(f"{self.id} requires image or task.metadata.docker_image")

        params.setdefault("image", task_image)
        params.setdefault("default_workspace_root", "/workspace")
        updated.environment.params = params
        return updated

Recipe Responsibilities

Rule	Why it matters
Match narrowly	Check benchmark id, environment id, and required task metadata. Avoid recipes that fire for unrelated tasks.
Preserve user overrides	Use `setdefault()` or explicit compatibility checks when users pass image, resources, workspace, or credentials themselves.
Fail before sandbox startup	If a heavy benchmark requires a task image and none is available, raise a clear `ValueError` in `apply()`.
Rewrite plans, not configs	Mutate a copied `ExecutionPlan`; do not mutate the original `RunRequest`.
Keep execution out	Recipes should not run commands, create sessions, evaluate results, or call model APIs.
Share helpers by benchmark family	Put shared layout helpers in `recipes/<benchmark_family>/common.py` when multiple providers need the same workspace mapping.

Recipes are applied automatically when execution.enabled_recipes is empty. Users can restrict recipe selection with --enabled-recipes.

Register And Verify

Export the recipe from the package initializer:

# src/agentcompass/recipes/my_benchmark/__init__.py
from .modal import MyBenchmarkModalRecipe  # noqa: F401

# src/agentcompass/recipes/__init__.py
from .my_benchmark import MyBenchmarkModalRecipe  # noqa: F401

Run a one-sample task and confirm the logs contain Recipe matched with the new recipe id:

agentcompass run \
  my_benchmark \
  my_harness \
  your-model \
  --env <env-provider> \
  --benchmark-params '{"sample_ids":["sample-1"]}' \
  --model-base-url "$MODEL_BASE_URL" \
  --model-api-key "$MODEL_API_KEY"

Add an Analyzer

Analyzers run after a task produces a RunResult. They inspect trajectories, metrics, errors, latency, model output, or tool calls, then attach an AnalysisResult under analysis_result.<AnalyzerId> in each details JSON. Aggregate fields are rendered into analysis_summary.md and analysis_summary.json. Add an analyzer when the logic is post-execution diagnosis or statistics. Do not put analyzer logic in a benchmark or harness if it can be rerun later with agentcompass analysis.

Analyzer File

Create a new file under src/agentcompass/analyzers/basic/, or create a dedicated sub-package for a larger analyzer family. Each analyzer must subclass BaseAnalyzer, register with @ANALYZERS.register(), and implement analysis().

from agentcompass.runtime.base import BaseAnalyzer
from agentcompass.runtime.models import AnalysisResult, AnalyzerCategory, RunResult
from agentcompass.runtime.registry import ANALYZERS


@ANALYZERS.register()
class MyAnalyzer(BaseAnalyzer):
    id = "MyAnalyzer"
    description = "Detects a custom trajectory pattern."
    category = AnalyzerCategory.BEHAVIOR
    datasets = []
    data_requirements = []
    conf = {"only_incorrect": False, "threshold": 0.0}
    distribution_fields = {
        "total_steps": "numeric_stats",
        "pattern_types": "value_counts",
    }

    async def analysis(self, task, prepared, result: RunResult, req, plan) -> AnalysisResult:
        if result is None or result.trajectory is None:
            return AnalysisResult(
                task_id=task.task_id,
                is_badcase=None,
                error="no trajectory available",
            )

        total_steps = len(result.trajectory.steps or [])

        return AnalysisResult(
            task_id=task.task_id,
            is_badcase=False,
            score=None,
            details={
                "total_steps": total_steps,
                "pattern_types": [],
            },
        )

Analyzer Attributes

Attribute	Purpose
`id`	Stable analyzer id. Used by the registry, CLI selection, details JSON, and summaries.
`description`	User-facing description exported by `agentcompass list dump`.
`category`	Summary grouping. Use `AnalyzerCategory.ERROR`, `EFFICIENCY`, `BEHAVIOR`, `ABILITY`, `BASIC_BADCASE`, or `ENV_FRAMEWORK_ERROR`.
`datasets`	Benchmark ids this analyzer supports. Empty list means all benchmarks.
`data_requirements`	JSONPath requirements checked before analysis. Missing fields skip the sample.
`conf`	Analyzer config merged from `execution.analysis_params.<AnalyzerId>`. Common keys include `only_incorrect` and `threshold`.
`distribution_fields`	Details fields to aggregate across tasks. Values are `numeric_stats` or `value_counts`.
`base_analyzer` / `priority`	Override mechanism for benchmark-specific analyzer variants. Higher priority wins within the same family.

AnalysisResult.is_badcase should be True for detected badcases, False for explicitly clean samples, and None for statistics-only analyzers that should not affect badcase ratios.

Aggregate Fields

Declare every cross-task summary field in distribution_fields:

Method	Use for	Summary output
`numeric_stats`	Numeric values or numeric lists.	count, min, mean, percentiles, and max.
`value_counts`	Strings or string lists.	frequency counts and ratios.

Only keys returned inside AnalysisResult.details can be aggregated.

Register And Verify

Export the analyzer from package initializers so registry import side effects can see it:

# src/agentcompass/analyzers/basic/__init__.py
from agentcompass.analyzers.basic.my_analyzer import MyAnalyzer  # noqa: F401

# src/agentcompass/analyzers/__init__.py
from agentcompass.analyzers.basic import MyAnalyzer  # noqa: F401

Then verify it appears in the component list:

uv run agentcompass list analyzer

Run it during evaluation:

agentcompass run \
  terminal_bench_2 \
  terminus2 \
  your-model \
  --env <env-provider> \
  --benchmark-params '{"sample_ids":["<task-id>"]}' \
  --model-base-url "$MODEL_BASE_URL" \
  --model-api-key "$MODEL_API_KEY" \
  --enable-analysis \
  --analysis-params '{"analyzers":["MyAnalyzer"]}'

Or rerun it on existing results:

agentcompass analysis --input results/<benchmark>/<model>/<run_id> --analysis-params '{"analyzers":["MyAnalyzer"]}' --task_concurrency 8

Developer Notes

Keep analyzer code read-only with respect to task execution. It should inspect persisted results, not mutate benchmark behavior.
Use datasets for benchmark-specific analyzers instead of branching on unrelated task metadata.
Use base_analyzer and priority when a benchmark-specific analyzer should override a generic analyzer.
Return a descriptive error instead of raising when a sample cannot be analyzed because optional trajectory fields are missing.

​Add a Benchmark

​Benchmark File

​Benchmark Responsibilities

​Register And Verify

​Add a Harness

​Harness File

​Harness Responsibilities

​Register And Verify

​Add an Environment

​Environment File

​Environment Responsibilities

​Register And Verify

​Add a Recipe

​Recipe File

​Recipe Responsibilities

​Register And Verify

​Add an Analyzer

​Analyzer File

​Analyzer Attributes

​Aggregate Fields

​Register And Verify

​Developer Notes

Add a Benchmark

Benchmark File

Benchmark Responsibilities

Register And Verify

Add a Harness

Harness File

Harness Responsibilities

Register And Verify

Add an Environment

Environment File

Environment Responsibilities

Register And Verify

Add a Recipe

Recipe File

Recipe Responsibilities

Register And Verify

Add an Analyzer

Analyzer File

Analyzer Attributes

Aggregate Fields

Register And Verify

Developer Notes