Code Implementation

Implement a Harness as the owner of one agent loop: validate compatibility, start its runtime, execute one PreparedTask, normalize a public RunResult, and release everything it created.

The tutorial adapter below returns a configured answer instead of calling a model. That makes registry and lifecycle smoke tests deterministic. Replace only its execution body when integrating a real SDK or CLI; keep the same public contracts.

Record the Upstream Contract

Record the official framework or CLI version, supported model protocols, configuration format, prompt flow, tool and workspace behavior, installation method, timeouts, termination rules, trajectory format, and credential handling. Pin the version when it affects commands, prompts, parsing, or reproducibility, and prefer the public SDK or CLI over private functions.

Create the Minimal File

The smallest complete integration needs one implementation file and one package export:

src/agentcompass/harnesses/
├── __init__.py
└── example_answer.py

Create src/agentcompass/harnesses/example_answer.py:

from __future__ import annotations

from dataclasses import dataclass
from typing import Any

from agentcompass.runtime import (
    HARNESSES,
    BaseHarness,
    EnvironmentSession,
    EnvironmentSpec,
    HarnessPlan,
    ModelSpec,
    PreparedTask,
    RunRequest,
    RunResult,
    TaskStatus,
)
from agentcompass.runtime.config import RuntimeHarnessConfig, config_field


@dataclass(slots=True)
class ExampleAnswerConfig(RuntimeHarnessConfig):
    """User-facing parameters for the tutorial Harness."""

    answer: str = config_field(
        default="Paris",
        description="Deterministic answer returned by the tutorial Harness.",
    )

    def __post_init__(self) -> None:
        self.answer = str(self.answer)


@dataclass(slots=True)
class ExampleAnswerPlan(HarnessPlan):
    """Resolved runtime state passed to every lifecycle method."""

    answer: str = "Paris"


@HARNESSES.register()
class ExampleAnswerHarness(BaseHarness):
    id = "example_answer"
    description = "Deterministic Harness used by the developer tutorial."
    config_class = ExampleAnswerConfig
    plan_class = ExampleAnswerPlan

    def supports(self, environment: EnvironmentSpec, model: ModelSpec) -> bool:
        _ = environment, model
        return True

    async def start_session(
        self,
        env: EnvironmentSession,
        req: RunRequest,
        plan: HarnessPlan,
    ) -> dict[str, Any]:
        _ = req
        self._require_plan(plan)
        return {"env": env}

    async def run_task(
        self,
        session: dict[str, Any],
        prepared: PreparedTask,
        req: RunRequest,
        plan: HarnessPlan,
    ) -> RunResult:
        _ = session, req
        harness_plan = self._require_plan(plan)
        return RunResult(
            task_id=prepared.task_id,
            status=TaskStatus.COMPLETED,
            category=prepared.category,
            final_answer=harness_plan.answer,
            ground_truth=prepared.ground_truth,
            metrics={"answer_characters": len(harness_plan.answer)},
        )

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

    @staticmethod
    def _require_plan(plan: HarnessPlan) -> ExampleAnswerPlan:
        if not isinstance(plan, ExampleAnswerPlan):
            raise TypeError("example_answer requires ExampleAnswerPlan")
        return plan

This covers the complete BaseHarness abstract surface: supports(), start_session(), and run_task(). It also shows close_session() explicitly even though the base class provides a no-op. BaseHarness.build_plan() copies matching config fields into plan_class, including the inherited inject_network_restriction_notice field.

Export and Inspect the Registration

Add the import to src/agentcompass/harnesses/__init__.py:

from .example_answer import ExampleAnswerHarness

Then inspect discovery and the generated config schema:

uv run agentcompass list harness
uv run agentcompass config docs harness example_answer

The first command should contain example_answer and its description. The second should show answer with default Paris and the inherited network-notice field. An absent ID indicates an import or registration failure; a present ID does not prove that a real upstream runtime can install or launch.

Run One Task

Use example_exact_match from the Harness-driven Benchmark tutorial:

uv run agentcompass run example_exact_match example_answer unused-model \
  --env host_process \
  --benchmark-params '{"sample_ids":["capital-france"]}' \
  --harness-params '{"answer":"Paris"}' \
  --task-concurrency 1 \
  --no-enable-analysis \
  --results-dir results-dev \
  --run-name harness-smoke

The command needs no endpoint because this tutorial Harness does not call req.model. The terminal result should complete with one selected task and report paths.run_info; its parent directory is the run directory. That directory should contain run_info.json, params.json, progress.json, progress.jsonl, logs/*.log, one details/*.json, and summary.md. The detail attempt should contain status: "completed", final_answer: "Paris", and correct: true after Benchmark evaluation.

For a real Harness, repeat the smoke with one bounded upstream task and its actual credentials. A registry check alone never exercises installation, launch, parsing, cleanup, or the model endpoint.

Replace the Tutorial Execution Body

Keep public parameters in RuntimeHarnessConfig so the CLI, Python SDK, config files, and generated documentation resolve the same fields. Put normalized runtime choices such as version, launch mode, install strategy, step limit, command timeout, and cost behavior in a typed HarnessPlan. Do not mutate RunRequest, inspect private Benchmark fields, or persist secrets in a plan.

Implement supports(environment, model) around capabilities rather than Benchmark IDs. Validate model protocol, shell and filesystem needs, endpoint forwarding, browser or GUI needs, workspace assumptions, credential location, and whether installation can work in the selected Environment. Reject unsupported combinations before Environment startup and never silently switch protocols, providers, install modes, or models.

The runtime calls the lifecycle in this order:

start_session under baseline network policy
  -> run_task under run network policy
  -> close_session under run network policy

Use start_session() for trusted installation, config generation, uploads, clients, or background services. run_task() executes exactly one PreparedTask and consumes only public fields such as prepared.input.prompt, messages, files, media, tools, and workspace. close_session() releases Harness-owned clients, processes, servers, temporary config, and background tasks after success, timeout, cancellation, or error; the runtime, not the Harness, closes the Environment.

Normalize Results Without Scoring

A Harness reports execution, not Benchmark correctness. Return the best available final_answer, requested files, ordered trajectory, token usage, timing, artifacts, and an accurate TaskStatus. Preserve timeout, refusal, invalid-output, termination, installation, launch, parsing, and model API errors.

Do not set Benchmark correct or score in the Harness, and do not turn evaluator failure into Harness failure. A process exit code of zero is not automatically a correct result. Ensure the public answer is assigned to RunResult.final_answer; a Benchmark must not need to recover it from a Harness-private artifact.

Support only reproducible installation strategies: a pinned preinstalled image, controlled installation before restricted execution, or an isolated driver-side optional extra. Do not assume every image has a package manager or compiler, and do not broaden the run network policy to make installation convenient.

Inject model, judge, search, and provider credentials through supported Environment or configuration mechanisms. Recursively redact them from commands, files, logs, trajectories, URLs, exceptions, dataclass representations, and persisted metadata.

Give each limit one owner: Harness command timeout limits the agent process, step limits bound the agent loop, model-client retries handle request transport, and runtime retries repeat a failed task attempt. Unknown model pricing must follow the Harness cost contract and should not terminate a run when the user explicitly selected an ignore-errors or disabled-cost mode.

Diagnose Failures by Stage

Symptom	Stage	First check
ID missing from list harness	Import and registration	harnesses/__init__.py, duplicate IDs, optional imports, and traceback
config docs omits a field	Config and plan	config_class, config_field(), matching plan_class field names, and __post_init__()
Run fails before Environment opens	Compatibility preflight	supports() protocol and capability checks
Environment opens, then setup fails	start_session()	Install command, baseline network policy, credentials, uploads, and launch logs
Attempt has run_error	run_task()	Exit code, timeout owner, response parsing, final-answer extraction, and trajectory conversion
Correct execution but Benchmark sees no answer	Result normalization	RunResult.final_answer rather than a private artifact or metric
Leaked process or client after cancellation	close_session()	Cleanup ownership and cancellation-safe finally paths
Answer exists but correct: false	Benchmark evaluation	Do not change the Harness status; inspect the evaluator contract and output format

For a compact real lifecycle and result adapter, read qwen3vl_gui.py. For an Environment-executed agent that installs, launches, parses a public final answer, and converts a trajectory, read naive_search_agent/harness.py.