Robots and workers

harness_policy.core.robot

JSON/HTTP transport to a worker in a separate Python environment.

This is the framework’s protocol, NOT the native RoboDojo/XPolicyLab protocol. An integration must implement Environment on the worker side.

RemoteEnvironment

class RemoteEnvironment
    __init__(endpoint: str, timeout: float=120.0)

A worker reached over HTTP (POST /rpc), with the bearer token from $HARNESS_WORKER_TOKEN when it is set. A worker error is raised as a RuntimeError, and no request is retried: a motion that timed out may have executed.

tools

def tools() -> list[dict]

The worker’s tool schemas. Checks its protocol version and takes its name; it also serves as the check that a worker answers.

reset

def reset(task: str, seed: int) -> Observation

Start an episode of task with seed and return its first observation.

execute

def execute(action: Action, episode_id: str, step: int) -> Transition

Execute action on the observation (episode_id, step); the worker rejects a stale pair. Returns the transition to the next observation.

call

def call(name: str, **arguments)

Invoke an environment-specific worker method.

close

def close() -> None

Nothing: the worker belongs to the process that launched it.

harness_policy.core.worker

A serial worker keeps all simulator operations on the serving thread.

WorkerAPI

class WorkerAPI
    __init__(environment, contract=None)

The worker’s RPC methods over one environment: describe, reset, execute and call. It tracks the current observation so a stale or replayed action is never executed, and passes results through the observation contract when one is given.

dispatch

def dispatch(method: str, arguments: dict)

The result of one RPC: describe (protocol version, name, tools, methods), reset, execute (only on the current episode and step) or call (one of the environment’s own methods). An unknown method is an error.

create_server

def create_server(environment, host='127.0.0.1', port=8765, token=None, contract=None)

An HTTP server (not yet serving) that answers POST /rpc with WorkerAPI, one request at a time. With a token (required on a non-loopback host) requests must carry it as a bearer token; errors come back as {“error”: …} with status 400.

harness_policy.core.contract

Observation contracts: what a robot connection lets a policy see.

A simulator worker can report more than an evaluation host hands a policy. The XPolicyLab contract is what RoboDojo’s official client gives (checked 2026-10-08, RoboDojo cb2a37d): camera colors, measured arm joints, link6 poses, the instruction, and for each gripper only the value it was last commanded; no camera calibration, no depth, no step limit. A worker served with a contract (harness serve --contract xpolicylab --embodiment arx_x5) reports only that, so development runs see what an evaluated policy will:

• grippers: the last commanded value instead of the measured opening;
• cameras: intrinsics and camera_to_world from the embodiment’s calibration (the head camera fixed in the world, wrist cameras fixed on link6), not the simulator’s;
• the step limit is left out of the profile and the state;
• the task is not named: a system under this contract has no task descriptions (config/systems.py), the instruction is all it is told.

The profile says which contract applies (observation_contract, gripper_feedback), so skills that need a measured gripper can check differently. Standard library only.

pose_matrix

def pose_matrix(pose)

The 4x4 homogeneous transform (nested lists) of a pose [x, y, z, qw, qx, qy, qz]; the quaternion is normalized first.

ContractFilter

class ContractFilter
    __init__(contract: str, embodiment: dict)

Applied by the worker server to everything its environment returns.

active

def active() -> bool

Whether the filter changes anything (every contract but native).

apply

def apply(method: str, arguments: dict, result)

What worker method method returned, as the contract shows it (unchanged under native); resets and executed ticks update the last commanded gripper values. States report commanded grippers and no step limit, camera views the embodiment’s calibration, and the profile names the contract; other methods pass unchanged.

harness_policy.embodiments

Embodiments: what a robot or a computer is, independent of how a run reaches it.

embodiments/<name>/embodiment.toml holds the facts a policy may know about the embodiment without asking a simulator. Its kind (KINDS, default robot) decides the session that drives it (runtime/episode.py), the skills that run on it (a skill’s embodiments) and the keys it must have.

A robot’s file holds joint groups and limits, arm mounts, the fingertip offset, gripper widths and readings, control rate, planner limits, camera calibration (head camera fixed in the world, wrist cameras fixed on link6), the action keys of XPolicyLab’s protocol, and the shield’s defaults. A simulator worker reports the same facts live in its profile; the XPolicyLab connector and contract-restricted workers take them from here. A computer’s (desktop) file holds its screen and the rate of its input steps.

embodiment_path

def embodiment_path(name) -> Path

The embodiment file of a name (embodiments/<name>/embodiment.toml) or a path; an unknown name is an error.

embodiment_kind

def embodiment_kind(data: dict) -> str

The kind of an embodiment tree or a worker profile (default robot); an unknown kind is an error.

load_embodiment

def load_embodiment(name) -> dict

An embodiment’s tree (a name or a path, see embodiment_path); it must have the keys of its kind (KINDS): a robot’s name, control_hz, groups, arm_bases and cameras.

profile

def profile(embodiment: dict) -> dict

The worker profile a robot embodiment implies (what a simulator worker reports live), for connections without a simulator.