Rust

The rlmesh crate serves environments, wraps policies as models, and drives evaluations from Rust. Environments and policies are traits you implement; keyword options are plain structs with Default; every fallible call returns rlmesh::Result<T>. Each blocking call has an _async twin for async code.

Install

The crate needs Rust 1.99 or newer and builds its gRPC stack from source, so the first build takes a few minutes.

cargo add rlmesh --git https://github.com/ArenaX-Labs/rlmesh

A build from source stamps a development workflow edition and only connects to binaries built from the identical source tree. To talk to another build, such as the Python package from PyPI, set RLMESH_RELEASE_BUILD=1 for the build:

RLMESH_RELEASE_BUILD=1 cargo build --release

Most authoring code starts with the prelude. It brings in the authoring traits (Env, VectorEnv, EnvFactory, Policy), the value types (Value, InfoDict, Kwargs, Metadata), Space, and Error / Result. The spaces module stays out of the prelude because spaces::Box would shadow std::boxed::Box.

use rlmesh::prelude::*;

Quickstart

Two programs: one serves an environment, the other connects a model to it and runs three episodes. They talk over gRPC, so either side could be replaced by a Python, C, or C++ program.

Serve an environment

Implement Env in Gymnasium shape and hand it to an EnvServer. reset returns (observation, info) and step returns (observation, reward, terminated, truncated, info).

use rlmesh::prelude::*;
use rlmesh::spaces::{Discrete, DiscreteOptions};
use rlmesh::{EnvServer, EnvServerOptions};

/// A three-step counter: the observation is the step index, every action
/// earns 1.0, and the episode terminates after the third step.
struct CounterEnv {
    observation_space: Space,
    action_space: Space,
    step: i64,
}

impl Env for CounterEnv {
    fn observation_space(&self) -> Space {
        self.observation_space.clone()
    }

    fn action_space(&self) -> Space {
        self.action_space.clone()
    }

    fn id(&self) -> Option<String> {
        Some("CounterEnv-v0".to_string())
    }

    fn reset(
        &mut self,
        _seed: Option<i64>,
        _options: Option<&InfoDict>,
    ) -> rlmesh::Result<(Value, InfoDict)> {
        self.step = 0;
        Ok((Value::Int(0), InfoDict::new()))
    }

    fn step(&mut self, _action: Option<Value>) -> rlmesh::Result<(Value, f64, bool, bool, InfoDict)> {
        self.step += 1;
        Ok((Value::Int(self.step), 1.0, self.step >= 3, false, InfoDict::new()))
    }
}

fn main() -> rlmesh::Result<()> {
    let env = CounterEnv {
        observation_space: Discrete::new(4, DiscreteOptions::default())?.into(),
        action_space: Discrete::new(2, DiscreteOptions::default())?.into(),
        step: 0,
    };
    // The server binds here, so the resolved address is known before serving.
    let server = EnvServer::new(
        env,
        EnvServerOptions {
            address: Some("tcp://127.0.0.1:5555".into()),
            ..Default::default()
        },
    )?;
    println!("serving on {}", server.address());
    // Blocks until Ctrl-C or a shutdown; the server then drains and closes the env.
    match server.serve() {
        Err(rlmesh::Error::Cancelled) => Ok(()),
        other => other,
    }
}

serve() keeps the calling thread. Use start() to serve in the background, then wait(timeout) and shutdown() to stop it.

Run a model against it

Implement Policy, wrap it in a Model, and call run with the environment’s address:

use rlmesh::prelude::*;
use rlmesh::{Model, ModelOptions, RunOptions};

/// Always takes action 0.
struct ConstantPolicy;

impl Policy for ConstantPolicy {
    fn predict(&mut self, _observation: Value) -> rlmesh::Result<Value> {
        Ok(Value::Int(0))
    }
}

fn main() -> rlmesh::Result<()> {
    let model = Model::new(ConstantPolicy, ModelOptions::default())?;
    let result = model.run(
        "tcp://127.0.0.1:5555",
        RunOptions {
            episodes: Some(3),
            seeds: Some(vec![0, 1, 2]),
            ..Default::default()
        },
    )?;
    println!("{result}");
    println!(
        "{} episodes, {} steps, mean reward {:.2}",
        result.num_episodes(),
        result.total_steps(),
        result.mean_reward()
    );
    model.close()
}

RunResult(episodes=3, mean_reward=3.000, total_steps=9)
3 episodes, 9 steps, mean reward 3.00

The repository ships the same pair as runnable examples: cargo run -p rlmesh --example serve_env and cargo run -p rlmesh --example run_model.

Conventions

Options are structs. Every call that takes optional settings takes an options struct whose Default holds the defaults: EnvServerOptions, RunOptions, SessionOptions, ModelOptions, ModelServeOptions, and so on. Always finish a struct literal with ..Default::default(). New fields can arrive in a minor release, and code that spells out every field would stop compiling. The client options also have builder methods:

let options = RemoteEnvOptions::default()
    .address("tcp://127.0.0.1:5555")
    .request_timeout(Duration::from_secs(30));

Optional values are Option. A seed is Option<i64>, a step limit is Option<u64>, and a timeout is Option<Duration>. None means “not set”, which is distinct from Some(0).

Blocking calls have _async twins. run / run_async, Session::step / step_async, RemoteEnv::new / new_async, EnvServer::serve / serve_async, and so on. The blocking form refuses to run inside an async task (it returns the RuntimeError class instead of stalling your executor), so call the _async twin there:

#[tokio::main]
async fn main() -> rlmesh::Result<()> {
    let model = Model::new(|_observation: Value| Ok(Value::Int(0)), ModelOptions::default())?;
    let result = model
        .run_async("tcp://127.0.0.1:5555", RunOptions::default())
        .await?;
    println!("{result}");
    model.close_async().await
}

Dropping a run_async future cancels the run.

Borrowing follows ownership. A run or session borrows the model and the environment you pass it. It never closes a model you supplied, and it closes the environment only when you set close_env: true. Handles release their resources on Drop; call close() when you want to see a cleanup error.

Language extras live in rlmesh::ext. Helpers with no counterpart in the other languages are kept out of the main namespace, so the main types read the same everywhere. The most useful ones:

Environments

An environment is any type that implements Env. Only the spaces, reset, and step are required. render, close, render_mode, metadata, and id have defaults. The server reads the spaces and attributes once, when it is built, and each lane runs on its own thread.

• Fail a request by returning an error. Error::environment_recoverable(message) fails just that request and keeps the client’s session; any other error ends the session.
• Batched simulators implement VectorEnv and serve with EnvServer::vector. To serve several independent Env instances as one vector endpoint, use EnvServer::lanes.
• Families of environments implement EnvFactory (tag parameters, variants, a make method) and serve with EnvFactoryExt::serve.
• Serving options in EnvServerOptions cover the address (or host / port / path / transport), adapter tags, timeouts and the workflow edition through options: Some(ServeOptions { .. }), and close_env_on_shutdown.

use rlmesh::{Error, Result, Value};

fn check_action(action: Option<Value>) -> Result<i64> {
    let Some(action) = action else {
        // Fails this request; the client's session stays usable.
        return Err(Error::environment_recoverable("step needs an action"));
    };
    action.as_int()
}

See Serve an Environment for addresses, authentication, health checks, and supervision.

Spaces and values

A Space is a space description plus its sampler. spaces::Box, Discrete, and Dict build the common kinds and convert into Space with .into(). Every space can be seeded, sampled, and asked whether it contains a value.

Data crossing the boundary is a Value: an enum of None, Bool, Int, Float, Text, Bytes, Tensor, List, Tuple, and Dict. A Discrete action is a Value::Int; a Box observation is a Value::Tensor. Metadata is an insertion-ordered map of Values, and InfoDict and Kwargs are aliases of it. The as_* accessors return the TypeError class for the wrong kind. Values sent across the boundary must match their space’s structure: a Dict value with a missing or extra key fails with Error::InvalidArgument before it is sent ($.position: missing key 'x' names the path), and a received value with one is rejected before it reaches your code.

use rlmesh::spaces::{Box, BoxOptions, Dict, Discrete, DiscreteOptions, SpaceSpec};
use rlmesh::{DType, Metadata, Tensor, Value};

let position = Box::new(-1.0, 1.0, [3], BoxOptions::default())?;
let mode = Discrete::new(4, DiscreteOptions { start: 1, ..Default::default() })?;
let observation = Dict::new([
    ("position", SpaceSpec::from(&position)),
    ("mode", SpaceSpec::from(&mode)),
])?;
observation.seed(Some(7));
let sample = observation.sample()?;
assert!(observation.contains(&sample));

// The same structure built by hand: a float32[3] tensor and an int.
let xyz: [f32; 3] = [0.0, 0.5, -0.5];
let bytes: Vec<u8> = xyz.iter().flat_map(|v| v.to_ne_bytes()).collect();
let mut dict = Metadata::new();
dict.insert("position", Value::Tensor(Tensor::new(&bytes, &[3], DType::Float32)?));
dict.insert("mode", Value::Int(2));
assert!(observation.contains(&Value::Dict(dict)));

Tensor::new copies the buffer. ext::TensorExt::from_vec adopts a Vec<u8> without copying, and Tensor exchanges CPU data with other frameworks through DLPack. See Spaces and Values for how each space kind encodes.

Models and policies

A Policy maps observations to actions. Write predict for one observation, or any of the other prediction methods (predict_batch, predict_chunk, predict_chunk_batch) and list them in const CORNERS. A method that needs the episode’s identity or per-episode state is listed in const CONTEXT too and written as its _with_context twin. A policy may also implement load (weights from a config), on_episode_end, and close.

A closure is a policy too, which is handy for baselines:

let model = Model::new(|_observation: Value| Ok(Value::Int(0)), ModelOptions::default())?;

ModelOptions carries the adapter spec, episode-end and close hooks, and trust_entrypoints. Model::from_config::<P>(kwargs) builds a policy with Default and passes the config to load.

The context gives each prediction the episode id, its seed, a per-predict seed for reproducible sampling, and an EpisodeState store that is dropped when the episode ends:

use rlmesh::prelude::*;
use rlmesh::{Model, ModelOptions, ModelServeOptions};

/// Counts its predicts per episode in that episode's state.
struct Counter;

impl Policy for Counter {
    const CONTEXT: Corners = Corners::PREDICT;

    fn predict_with_context(
        &mut self,
        _observation: Value,
        context: &mut PredictContext<'_>,
    ) -> rlmesh::Result<Value> {
        *context.state.get_or_insert_with("predicts", || 0u64)? += 1;
        Ok(Value::Int(0))
    }
}

fn main() -> rlmesh::Result<()> {
    let model = Model::new(Counter, ModelOptions::default())?;
    // Serve the model; blocks until cancelled.
    model.serve("tcp://127.0.0.1:50051", ModelServeOptions::default())
}

See Build a Model for prediction corners, batching, and chunking.

Runs and sessions

run (or Model::run) plays whole episodes and returns a RunResult with one EpisodeResult per episode plus timing telemetry. The environment can be an address string, an &EnvServer, a &mut RemoteEnv, or a local &mut env, which runs in-process behind a loopback server. The model can be a &Model, a &RemoteModel, or RANDOM_SAMPLE.

RunOptions sets the episode budget (episodes, seeds), truncation (max_episode_steps, max_episode), hooks, instruction, close_env, execution_horizon, workflow_edition, and cancel. With neither episodes nor seeds, a run plays one episode; with only seeds, one episode per seed.

To observe a run, implement RunHooks; every method has a default:

use rlmesh::{EpisodeResult, Model, ModelOptions, RunHooks, RunOptions, StepEvent, Value};

#[derive(Default)]
struct Progress {
    steps: u64,
}

impl RunHooks for Progress {
    fn on_step(&mut self, _event: &StepEvent) -> rlmesh::Result<()> {
        self.steps += 1;
        Ok(())
    }

    fn on_episode_end(&mut self, result: &EpisodeResult) -> rlmesh::Result<()> {
        println!("episode {}: reward {} in {} steps", result.index, result.reward, result.steps);
        Ok(())
    }
}

fn main() -> rlmesh::Result<()> {
    let model = Model::new(|_observation: Value| Ok(Value::Int(0)), ModelOptions::default())?;
    let mut progress = Progress::default();
    model.run(
        "tcp://127.0.0.1:5555",
        RunOptions { episodes: Some(2), hooks: Some(&mut progress), ..Default::default() },
    )?;
    println!("{} steps observed", progress.steps);
    Ok(())
}

A Session gives you the loop itself. It connects lazily, applies the resolved adapter, and ends the previous episode on each reset:

use rlmesh::prelude::*;
use rlmesh::{Model, ModelOptions, RemoteEnv, RemoteEnvOptions, SessionOptions};

fn main() -> rlmesh::Result<()> {
    let model = Model::new(|_observation: Value| Ok(Value::Int(0)), ModelOptions::default())?;
    let mut env = RemoteEnv::new(RemoteEnvOptions::default().address("tcp://127.0.0.1:5555"))?;

    let mut session = model.session(&mut env, SessionOptions::default())?;
    let (mut observation, _info) = session.reset(Some(0), None)?;
    loop {
        let action = session.predict(observation)?;
        let (next, reward, terminated, truncated, _info) = session.step(action)?;
        println!("reward {reward}");
        observation = next;
        if terminated || truncated {
            break;
        }
    }
    session.close()
}

Session::reset takes the seed and an optional trial index. A session can also run episodes with SessionRunOptions, read the EnvContract, and read observation roles and frame sources for visualization. See Run an Evaluation.

Remote clients

RemoteEnv and RemoteVectorEnv connect and handshake when constructed, so an unreachable address fails immediately. RemoteModel only validates its address; it dials when a session first needs it.

let mut env = RemoteEnv::new(
    RemoteEnvOptions::default()
        .address("tcp://127.0.0.1:5555")
        .connect_timeout(Duration::from_secs(5))
        .request_timeout(Duration::from_secs(30)),
)?;
let (_observation, _info) = env.reset(Some(42), None)?;
let (_observation, reward, terminated, truncated, _info) = env.step(Value::Int(1))?;
env.close()?;

let remote = RemoteModel::new(RemoteModelOptions::default().address("tcp://127.0.0.1:50051"))?;
let result = rlmesh::run(&remote, "tcp://127.0.0.1:5555", RunOptions::default())?;

A timeout of zero means no timeout, the same as leaving it unset. I/O on a closed client fails with the RuntimeError class. See Connect a Client.

Errors

Every error is an rlmesh::Error. Each variant corresponds to one error class, and err.class() returns the exact ErrorClass, whose is_subclass_of follows the class hierarchy (for example, ConnectionError is a kind of OSError). Display gives the user-facing message.

use rlmesh::{Error, ErrorClass, RemoteEnv, RemoteEnvOptions};

match RemoteEnv::new(RemoteEnvOptions::default().address("tcp://127.0.0.1:5999")) {
    Ok(_) => println!("connected"),
    Err(Error::Connection(message)) => eprintln!("endpoint unreachable: {message}"),
    Err(Error::Timeout(after)) => eprintln!("timed out after {after:?}"),
    Err(err) if err.class().is_subclass_of(ErrorClass::RLMeshException) => {
        eprintln!("{}: {err} (recoverable: {})", err.class().name(), err.is_recoverable())
    }
    Err(err) => eprintln!("{err}"),
}

Error is #[non_exhaustive], so always keep a fallback arm. Library warnings (such as a missing workflow edition) go to stderr by default; set_warning_handler routes them elsewhere.

Adapters

Adapters let an environment and a model that were written independently agree on observation and action layouts. The environment publishes EnvTags (pass them as EnvServerOptions::tags, or attach them with adapters::tag), the model declares a ModelSpec (ModelOptions::spec), and runs and sessions resolve the pair when they connect:

use rlmesh::adapters::{
    ACTION_DELTA_POS, Action, Actuator, EEF_POS, EnvTags, IMAGE_PRIMARY, Image, ImageTag,
    InputNode, ModelSpec, ObsNode, State, StateTag,
};

let tags = EnvTags::new(
    ObsNode::dict([
        ("camera", ImageTag::new(IMAGE_PRIMARY).into()),
        ("eef_pos", StateTag::new(EEF_POS).into()),
    ]),
    Action::new([Actuator { role: Some(ACTION_DELTA_POS.into()), dim: 3, ..Default::default() }]),
)?;

let spec = ModelSpec::new(
    InputNode::dict([
        ("image", Image::new(IMAGE_PRIMARY).into()),
        ("state", State::new(EEF_POS).into()),
    ]),
    Action::new([Actuator { role: Some(ACTION_DELTA_POS.into()), dim: 3, ..Default::default() }]),
)?;
let model = Model::new(
    |_observation: Value| Ok(Value::None),
    ModelOptions { spec: Some(spec.into()), ..Default::default() },
)?;

Use NO_ADAPTER as the spec for a model that adapts its own observations. adapters::resolve and Adapter expose the resolution and per-step transforms directly. The Adapters guide explains roles and matching.

View and record

View streams frames live to a terminal or an HTTP page; set it on SessionOptions::view (or on RunOptions::view with a remote model or RANDOM_SAMPLE). View::parse("http:8008") accepts the short forms term, http:PORT, and both.

Recorder collects results into a portable result bundle. add records a finished run; capture returns hooks that also record video during a run.

let mut recorder = Recorder::new(RecorderOptions::default())?;
recorder.add(&result, "constant", "CounterEnv-v0", RecorderAddOptions::default())?;
let path = recorder.export_("results", RecorderExportOptions::default())?;
recorder.close()?;

Reference

• Rust API reference: every type, trait, and function, by module.
• rlmesh, rlmesh::env, rlmesh::spaces, rlmesh::adapters, rlmesh::recorder, and rlmesh::ext.
• Native Languages compares the Rust, C, and C++ APIs.