Benchmark-Driven Execution

Use HarnessFreeBenchmark when the Benchmark must run its upstream-defined inference or interaction protocol. This can be a response-generation workflow, multi-role interaction with a user simulator, or a domain state machine. Keep scoring in evaluate() even when the official runner packages generation and evaluation as one pipeline.

Choose the execution owner by the evaluation contract. Reuse a Harness when its agent loop preserves that contract; keep a dedicated upstream runner in the Benchmark when adapting it to a generic loop would change the protocol, termination conditions, or trajectory semantics. A workflow’s stages alone do not determine whether it needs an external Harness.

How It Differs from an Ordinary Benchmark

HarnessFreeBenchmark inherits every BaseBenchmark contract and additionally requires run_task():

prepare_task
  → Benchmark.run_task
  → artifact_collect commands (when declared and enabled)
  → download_artifacts (runtime)
  → evaluate

Use none as the Harness ID in the run command:

uv run agentcompass run example_interactive none MODEL \
  --env docker

Here, none means there is no external Harness. It does not require evaluation_environment_mode to be none; Benchmark-driven execution can still use reuse or fresh evaluation.

Implement run_task()

The following excerpt shows only what differs from BaseBenchmark. Assume prepare_task() has written the upstream runner request into the task Environment and stored its public paths under official_runner in PreparedTask.metadata:

import json

from agentcompass.runtime import (
    EnvironmentSession,
    ExecutionPlan,
    HarnessFreeBenchmark,
    PreparedTask,
    RunRequest,
    RunResult,
    ExecutionIssue,
    TaskSpec,
    TaskStatus,
)
from agentcompass.runtime.metrics import make_metric_contract


class ExampleInteractiveBenchmark(HarnessFreeBenchmark):
    evaluation_environment_mode = "reuse"
    metric_contract = make_metric_contract(
        primary="score",
        scalar=("score",),
        labels={"score": "Reward"},
    )

    async def run_task(
        self,
        task: TaskSpec,
        prepared: PreparedTask,
        req: RunRequest,
        plan: ExecutionPlan,
        env: EnvironmentSession | None = None,
    ) -> RunResult:
        _ = req, plan
        if env is None:
            raise RuntimeError(
                "example_interactive requires an EnvironmentSession"
            )

        runner = prepared.metadata["official_runner"]
        execution = await env.exec(
            [
                "python3",
                "-m",
                "example_interactive.runner",
                "run",
                "--request",
                runner["request_path"],
                "--output",
                runner["result_path"],
            ],
            timeout=runner["timeout_seconds"],
        )
        if execution.returncode != 0 or execution.timed_out:
            detail = (execution.stderr or execution.stdout).strip()
            return RunResult(
                task_id=task.task_id,
                category=task.category,
                status=TaskStatus.RUN_ERROR,
                issues=[ExecutionIssue("error" if execution.timed_out else "fatal", "run", "runner_failed", detail or "official runner failed")],
            )

        payload = json.loads(
            await env.read_text(runner["result_path"])
        )
        return RunResult(
            task_id=task.task_id,
            category=task.category,
            status=TaskStatus.COMPLETED,
            final_answer=payload.get("final_answer"),
            artifacts={"official_result": payload},
        )

This excerpt cannot run by itself because data loading, task preparation, and the upstream runner depend on the actual protocol. It demonstrates the ownership boundary: the runtime manages the Environment lifecycle and error flow, while run_task() runs the Benchmark-specific loop and translates its output into RunResult.

Keep Execution and Scoring Separate

Even if the upstream runner already produces a reward, let evaluate() translate it into the final verdict instead of scoring inside run_task(). This preserves the runtime’s distinction between execution and evaluation failures:

from dataclasses import replace


async def evaluate(
    self,
    task: TaskSpec,
    prepared: PreparedTask,
    result: RunResult,
    req: RunRequest,
    plan: ExecutionPlan,
    env: EnvironmentSession | None = None,
) -> RunResult:
    _ = task, prepared, req, plan
    if any(issue.severity == "fatal" for issue in result.issues):
        return replace(result, metrics={})
    if env is None:
        raise RuntimeError("evaluation requires the task Environment")

    payload = result.artifacts["official_result"]
    reward = float(payload["reward"])
    return replace(
        result,
        metrics={"score": reward},
    )

If another command must run to obtain the reward, execute the verifier through the env passed to evaluate() and classify a verifier crash as EVAL_ERROR. See Evaluation Modes and Artifacts for the Environment lifecycle and Results and Aggregation for status composition.

Responsibility Boundaries

• prepare_task() places the upstream runner, data, and request material in the Environment; repeating it during a retry must not corrupt task state.
• run_task() owns the Benchmark-specific inference or interaction protocol, Model-call orchestration, and trajectory conversion, but it must not create or close the Environment itself.
• Declare submission preparation commands through TaskSpec.artifact_collect; without commands, runtime collects existing files directly.
• evaluate() interprets runner output or invokes the official verifier while preserving any existing execution error.
• Pass credentials for the Model, simulator Model, or judge Model through existing Model-configuration boundaries; they must never enter results, logs, or publishable metadata.

For a complete production implementation, inspect TauBenchBenchmark. It uses the none Harness, starts the upstream interaction runner in run_task(), and evaluates in the same task Environment through reuse mode.