vla-evaluation-harness: Skill for Claude Code

Add Benchmark

Integrate a new simulation benchmark into vla-eval. Benchmarks run inside Docker containers and communicate with model servers over WebSocket + msgpack.

1. Gather requirements

Ask the user for (if not already provided):

• Benchmark name (e.g. maniskill3)
• Simulation framework (e.g. MuJoCo, SAPIEN, PyBullet, Isaac Sim)
• Key pip dependencies needed inside Docker
• Observation format — cameras, resolution, whether to include proprioceptive state
• Action space — dimension, format (e.g. 7-DoF delta EEF + gripper)
• Success condition — how to detect task completion
• Max steps per episode

2. Create the benchmark module

Create src/vla_eval/benchmarks/<name>/:

src/vla_eval/benchmarks/<name>/
├── __init__.py      # empty
├── benchmark.py     # main implementation
└── utils.py         # optional helpers

Subclass StepBenchmark from vla_eval.benchmarks.base and implement the required methods:

from typing import Any

import numpy as np

from vla_eval.benchmarks.base import StepBenchmark, StepResult
from vla_eval.specs import DimSpec
from vla_eval.types import Action, EpisodeResult, Observation, Task


class MyBenchmark(StepBenchmark):
    def __init__(self, **kwargs: Any) -> None:
        super().__init__()
        # Accept benchmark-specific params from config YAML `params:` section.
        # Lazily import heavy deps (MuJoCo, SAPIEN) — NOT at module level,
        # because the registry resolves the class without loading the sim.
        ...

    # --- Required methods (6) ---

    def get_tasks(self) -> list[Task]:
        # Return list of task dicts. Each MUST have a "name" key.
        # May include "suite" for task filtering.
        ...

    def reset(self, task: Task) -> Any:
        # Reset env for task. Store env on self. Return initial raw observation.
        # task dict has "episode_idx" (int) injected by the orchestrator.
        ...

    def step(self, action: Action) -> StepResult:
        # action dict has "actions" key (np.ndarray from model server).
        # Return StepResult(obs, reward, done, info).
        ...

    def make_obs(self, raw_obs: Any, task: Task) -> Observation:
        # Convert raw env observation to dict for model server.
        # Convention:
        #   {"images": {"cam_name": np.ndarray HWC uint8},
        #    "task_description": str}
        # Optionally add "state": np.ndarray for proprioception.
        ...

    def get_step_result(self, step_result: StepResult) -> EpisodeResult:
        # Extract episode result from the final StepResult.
        # Must return at least {"success": bool}.
        ...

    # --- Optional overrides ---

    def check_done(self, step_result: StepResult) -> bool:
        # Default: step_result.done. Override for custom termination logic.
        return step_result.done

    def get_action_spec(self) -> dict[str, DimSpec]:
        # Declare the action format this benchmark's env consumes.
        # The orchestrator compares this against the model server's spec
        # and warns on mismatches — catching convention bugs early.
        ...

    def get_observation_spec(self) -> dict[str, DimSpec]:
        # Declare the observation format this benchmark produces.
        ...

    def get_metric_keys(self) -> dict[str, str]:
        # Declare which metrics from get_step_result() to aggregate.
        # Default: {"success": "mean"} (= success rate).
        # Aggregation options: "mean", "sum", "max", "min".
        return {"success": "mean"}

    def get_metadata(self) -> dict[str, Any]:
        # Return {"max_steps": N} for benchmark default.
        return {}

    def cleanup(self) -> None:
        # Release resources (envs, renderers). Called at end of evaluation.
        ...