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

# Developer Guide

> Extend AgentCompass with new components.

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.

```python theme={null}
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:

```python theme={null}
from .my_benchmark import MyBenchmark  # noqa: F401
```

Then verify the component list and run a single task:

```bash theme={null}
uv run agentcompass list benchmark
```

```bash theme={null}
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.

```python theme={null}
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`:

```python theme={null}
from .my_harness import MyHarness  # noqa: F401
```

Then check discovery and a smoke run:

```bash theme={null}
uv run agentcompass list harness
```

```bash theme={null}
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.

```python theme={null}
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:

```python theme={null}
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:

```bash theme={null}
python -m compileall src/agentcompass/environments/my_environment.py
```

```bash theme={null}
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.

```python theme={null}
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:

```python theme={null}
# 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:

```bash theme={null}
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()`.

```python theme={null}
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:

```python theme={null}
# 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:

```bash theme={null}
uv run agentcompass list analyzer
```

Run it during evaluation:

```bash theme={null}
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:

```bash theme={null}
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.