> This page is for version Release 3.0 (default).
> For other versions, use one of these documentation indexes:
> - Release 3.0 (default): https://docs.nvidia.com/isaaclab/3.0/llms.txt
> - Release 2.3: https://docs.nvidia.com/isaaclab/2.3/llms.txt
> - Release 2.2: https://docs.nvidia.com/isaaclab/2.2/llms.txt
> - Release 2.1: https://docs.nvidia.com/isaaclab/2.1/llms.txt

> For clean Markdown of any page, append .md to the page URL.
> For a complete documentation index, see https://docs.nvidia.com/isaaclab/llms.txt.
> For AI client integration (Claude Code, Cursor, etc.), connect to the MCP server at https://docs.nvidia.com/isaaclab/_mcp/server.

# Creating a Direct Workflow RL Environment

> **Info**
>
> This tutorial is the source of truth for the `isaaclab-randomizing-with-events` agent skill
> ([skills/user/domain-randomization-events/](https://github.com/isaac-sim/IsaacLab/blob/release/3.0.0/skills/user/domain-randomization-events/SKILL.md)).
> When you change this page, update the skill so agent guidance stays in sync. See
> [/source/developer-tools/agent_skills](/source/developer-tools/agent_skills).

In addition to the `envs.ManagerBasedRLEnv` class, which encourages the use of configuration classes
for more modular environments, the `~isaaclab.envs.DirectRLEnv` class allows for more direct control
in the scripting of environment.

Instead of using Manager classes for defining rewards and observations, the direct workflow tasks
implement the full reward and observation functions directly in the task script.
This allows for more control in the implementation of the methods, such as using pytorch jit
features, and provides a less abstracted framework that makes it easier to find the various
pieces of code.

In this tutorial, we will configure the cartpole environment using the direct workflow implementation to create a task
for balancing the pole upright. We will declare its scene and implement actions, resets, rewards, and observations.

## The Code

For this tutorial, we use the cartpole environment defined in `isaaclab_tasks.core.cartpole` module.

```python
# Copyright (c) 2022-2026, The Isaac Lab Project Developers (https://github.com/isaac-sim/IsaacLab/blob/main/CONTRIBUTORS.md).
# All rights reserved.
#
# SPDX-License-Identifier: BSD-3-Clause

"""Direct-workflow cartpole balancing environment."""

from __future__ import annotations

from collections.abc import Sequence
from typing import TYPE_CHECKING

import torch

from isaaclab.envs import DirectRLEnv
from isaaclab.utils.math import sample_uniform, wrap_to_pi

if TYPE_CHECKING:
    from .cartpole_direct_env_cfg import CartpoleEnvCfg

class CartpoleEnv(DirectRLEnv):
    """Cartpole balancing environment driven by proprioceptive (joint-state) observations."""

    cfg: CartpoleEnvCfg

    def __init__(self, cfg: CartpoleEnvCfg, render_mode: str | None = None, **kwargs):
        super().__init__(cfg, render_mode, **kwargs)

        self.cartpole = self.scene["cartpole"]
        cart_dof_idx, _ = self.cartpole.find_joints(self.cfg.cart_dof_name)
        pole_dof_idx, _ = self.cartpole.find_joints(self.cfg.pole_dof_name)
        # device indices avoid per-step host uploads
        self._cart_dof_idx = torch.tensor(cart_dof_idx, device=self.device)
        self._pole_dof_idx = torch.tensor(pole_dof_idx, device=self.device)
        self.action_scale = self.cfg.action_scale

        self.joint_pos = self.cartpole.data.joint_pos.torch
        self.joint_vel = self.cartpole.data.joint_vel.torch

    def _pre_physics_step(self, actions: torch.Tensor) -> None:
        self.actions = self.action_scale * actions

    def _apply_action(self) -> None:
        self.cartpole.set_joint_effort_target_index(target=self.actions, joint_ids=self._cart_dof_idx)

    def _get_observations(self) -> dict:
        joint_pos_rel = self.joint_pos - self.cartpole.data.default_joint_pos.torch
        joint_vel_rel = self.joint_vel - self.cartpole.data.default_joint_vel.torch
        obs = torch.cat(
            (
                joint_pos_rel[:, self._cart_dof_idx],
                joint_pos_rel[:, self._pole_dof_idx],
                joint_vel_rel[:, self._cart_dof_idx],
                joint_vel_rel[:, self._pole_dof_idx],
            ),
            dim=-1,
        )
        return {"policy": obs}

    def _get_rewards(self) -> torch.Tensor:
        return compute_rewards(
            self.cfg.rew_scale_alive,
            self.cfg.rew_scale_terminated,
            self.cfg.rew_scale_pole_pos,
            self.cfg.rew_scale_cart_vel,
            self.cfg.rew_scale_pole_vel,
            self.joint_pos[:, self._pole_dof_idx],
            self.joint_vel[:, self._pole_dof_idx],
            self.joint_vel[:, self._cart_dof_idx],
            self.reset_terminated,
            self.step_dt,
        )

    def _get_dones(self) -> tuple[torch.Tensor, torch.Tensor]:
        self.joint_pos = self.cartpole.data.joint_pos.torch
        self.joint_vel = self.cartpole.data.joint_vel.torch

        time_out = self.episode_length_buf >= self.max_episode_length
        out_of_bounds = torch.any(torch.abs(self.joint_pos[:, self._cart_dof_idx]) > self.cfg.max_cart_pos, dim=1)
        return out_of_bounds, time_out

    def _reset_idx(self, env_ids: Sequence[int] | None):
        if env_ids is None:
            env_ids = self.cartpole._ALL_INDICES

        # log the survival success rate before resetting (survived = timed out without terminating early)
        survived = self.reset_time_outs[env_ids].float()
        # no .item(): avoids a sync on every reset
        self.extras.setdefault("log", {})["Metrics/success_rate"] = survived.mean()

        super()._reset_idx(env_ids)

        joint_pos = self.cartpole.data.default_joint_pos.torch[env_ids].clone()
        joint_pos[:, self._cart_dof_idx] += sample_uniform(
            self.cfg.initial_cart_position_range[0],
            self.cfg.initial_cart_position_range[1],
            joint_pos[:, self._cart_dof_idx].shape,
            joint_pos.device,
        )
        joint_vel = self.cartpole.data.default_joint_vel.torch[env_ids].clone()
        joint_vel[:, self._cart_dof_idx] += sample_uniform(
            self.cfg.initial_cart_velocity_range[0],
            self.cfg.initial_cart_velocity_range[1],
            joint_vel[:, self._cart_dof_idx].shape,
            joint_vel.device,
        )
        joint_pos[:, self._pole_dof_idx] += sample_uniform(
            self.cfg.initial_pole_angle_range[0],
            self.cfg.initial_pole_angle_range[1],
            joint_pos[:, self._pole_dof_idx].shape,
            joint_pos.device,
        )
        joint_vel[:, self._pole_dof_idx] += sample_uniform(
            self.cfg.initial_pole_velocity_range[0],
            self.cfg.initial_pole_velocity_range[1],
            joint_vel[:, self._pole_dof_idx].shape,
            joint_vel.device,
        )

        # clamp the sampled state to the joint limits (matches the manager-based reset_joints_by_offset)
        joint_pos_limits = self.cartpole.data.soft_joint_pos_limits.torch[env_ids]
        joint_pos = joint_pos.clamp_(joint_pos_limits[..., 0], joint_pos_limits[..., 1])
        joint_vel_limits = self.cartpole.data.soft_joint_vel_limits.torch[env_ids]
        joint_vel = joint_vel.clamp_(-joint_vel_limits, joint_vel_limits)

        default_root_pose = self.cartpole.data.default_root_pose.torch[env_ids].clone()
        default_root_pose[:, :3] += self.scene.env_origins[env_ids]
        default_root_vel = self.cartpole.data.default_root_vel.torch[env_ids].clone()

        self.joint_pos[env_ids] = joint_pos
        self.joint_vel[env_ids] = joint_vel

        self.cartpole.write_root_pose_to_sim_index(root_pose=default_root_pose, env_ids=env_ids)
        self.cartpole.write_root_velocity_to_sim_index(root_velocity=default_root_vel, env_ids=env_ids)
        self.cartpole.write_joint_position_to_sim_index(position=joint_pos, env_ids=env_ids)
        self.cartpole.write_joint_velocity_to_sim_index(velocity=joint_vel, env_ids=env_ids)

@torch.jit.script
def compute_rewards(
    rew_scale_alive: float,
    rew_scale_terminated: float,
    rew_scale_pole_pos: float,
    rew_scale_cart_vel: float,
    rew_scale_pole_vel: float,
    pole_pos: torch.Tensor,
    pole_vel: torch.Tensor,
    cart_vel: torch.Tensor,
    reset_terminated: torch.Tensor,
    step_dt: float,
) -> torch.Tensor:
    pole_pos = wrap_to_pi(pole_pos)
    rew_alive = rew_scale_alive * (1.0 - reset_terminated.float())
    rew_termination = rew_scale_terminated * reset_terminated.float()
    rew_pole_pos = rew_scale_pole_pos * torch.sum(torch.square(pole_pos), dim=-1)
    rew_cart_vel = rew_scale_cart_vel * torch.sum(torch.abs(cart_vel), dim=-1)
    rew_pole_vel = rew_scale_pole_vel * torch.sum(torch.abs(pole_vel), dim=-1)
    total_reward = (rew_alive + rew_termination + rew_pole_pos + rew_cart_vel + rew_pole_vel) * step_dt
    return total_reward
```

## The Code Explained

Similar to the manager-based environments, a configuration class is defined for the task to hold settings
for the simulation parameters, the scene, the actors, and the task. With the direct workflow implementation,
the `envs.DirectRLEnvCfg` class is used as the base class for configurations.
Since the direct workflow implementation does not use Action and Observation managers, the task
config should define the number of actions and observations for the environment.

```python
@configclass
class CartpoleEnvCfg(DirectRLEnvCfg):
   ...
   action_space = 1
   observation_space = 4
   state_space = 0
```

The config class can also be used to define task-specific attributes, such as scaling for reward terms
and thresholds for reset conditions.

```python
@configclass
class CartpoleEnvCfg(DirectRLEnvCfg):
   ...
   # reset
   max_cart_pos = 3.0
   initial_pole_angle_range = [-0.25, 0.25]

   # reward scales
   rew_scale_alive = 1.0
   rew_scale_terminated = -2.0
   rew_scale_pole_pos = -1.0
   rew_scale_cart_vel = -0.01
   rew_scale_pole_vel = -0.005
```

When creating a new environment, the code should define a new class that inherits from `~isaaclab.envs.DirectRLEnv`.

```python
class CartpoleEnv(DirectRLEnv):
   cfg: CartpoleEnvCfg

   def __init__(self, cfg: CartpoleEnvCfg, render_mode: str | None = None, **kwargs):
     super().__init__(cfg, render_mode, **kwargs)
```

The class can also hold class variables that are accessible by all functions in the class,
including functions for applying actions, computing resets, rewards, and observations.

### Scene Creation

Direct and manager-based environments use the same declarative scene lifecycle. Define the
assets and sensors on an `~isaaclab.scene.InteractiveSceneCfg` subclass:

```python
@configclass
class CartpoleSceneCfg(InteractiveSceneCfg):
    """Cartpole assets constructed and cloned as one scene."""

    ground = AssetBaseCfg(prim_path="/World/ground", spawn=sim_utils.GroundPlaneCfg())
    cartpole: ArticulationCfg = replace(CARTPOLE_CFG, prim_path="{ENV_REGEX_NS}/Robot")
    light = AssetBaseCfg(
        prim_path="/World/Light",
        spawn=sim_utils.DistantLightCfg(intensity=2000.0),
        init_state=AssetBaseCfg.InitialStateCfg(rot=LIGHT_ORIENTATION),
    )
```

Assign that configuration to the task's `scene` field. `~isaaclab.envs.DirectRLEnv`
constructs the scene and completes its one clone lifecycle before the task constructor resumes.
Task code can then keep a convenient reference to a declared entity:

```python
self.cartpole = self.scene["cartpole"]
```

### Defining Rewards

Reward function should be defined in the `_get_rewards(self)` API, which returns the reward
buffer as a return value. Within this function, the task is free to implement the logic of
the reward function. In this example, we implement a Pytorch JIT function that computes
the various components of the reward function.

```python
def _get_rewards(self) -> torch.Tensor:
     total_reward = compute_rewards(
         self.cfg.rew_scale_alive,
         self.cfg.rew_scale_terminated,
         self.cfg.rew_scale_pole_pos,
         self.cfg.rew_scale_cart_vel,
         self.cfg.rew_scale_pole_vel,
         self.joint_pos[:, self._pole_dof_idx[0]],
         self.joint_vel[:, self._pole_dof_idx[0]],
         self.joint_pos[:, self._cart_dof_idx[0]],
         self.joint_vel[:, self._cart_dof_idx[0]],
         self.reset_terminated,
     )
     return total_reward

@torch.jit.script
def compute_rewards(
    rew_scale_alive: float,
    rew_scale_terminated: float,
    rew_scale_pole_pos: float,
    rew_scale_cart_vel: float,
    rew_scale_pole_vel: float,
    pole_pos: torch.Tensor,
    pole_vel: torch.Tensor,
    cart_pos: torch.Tensor,
    cart_vel: torch.Tensor,
    reset_terminated: torch.Tensor,
):
    rew_alive = rew_scale_alive * (1.0 - reset_terminated.float())
    rew_termination = rew_scale_terminated * reset_terminated.float()
    rew_pole_pos = rew_scale_pole_pos * torch.sum(torch.square(pole_pos), dim=-1)
    rew_cart_vel = rew_scale_cart_vel * torch.sum(torch.abs(cart_vel), dim=-1)
    rew_pole_vel = rew_scale_pole_vel * torch.sum(torch.abs(pole_vel), dim=-1)
    total_reward = rew_alive + rew_termination + rew_pole_pos + rew_cart_vel + rew_pole_vel
    return total_reward
```

### Defining Observations

The observation buffer should be computed in the `_get_observations(self)` function,
which constructs the observation buffer for the environment. At the end of this API,
a dictionary should be returned that contains `policy` as the key, and the full
observation buffer as the value. For asymmetric policies, the dictionary should also
include the key `critic` and the states buffer as the value.

```python
def _get_observations(self) -> dict:
    joint_pos_rel = self.joint_pos - self.cartpole.data.default_joint_pos.torch
    joint_vel_rel = self.joint_vel - self.cartpole.data.default_joint_vel.torch
    obs = torch.cat(
        (
            joint_pos_rel[:, self._cart_dof_idx],
            joint_pos_rel[:, self._pole_dof_idx],
            joint_vel_rel[:, self._cart_dof_idx],
            joint_vel_rel[:, self._pole_dof_idx],
        ),
        dim=-1,
    )
    return {"policy": obs}
```

### Computing Dones and Performing Resets

Populating the `dones` buffer should be done in the `_get_dones(self)` method.
This method is free to implement logic that computes which environments would need to be reset
and which environments have reached the episode length limit. Both results should be
returned by the `_get_dones(self)` function, in the form of a tuple of boolean tensors.

```python
def _get_dones(self) -> tuple[torch.Tensor, torch.Tensor]:
    self.joint_pos = self.cartpole.data.joint_pos.torch
    self.joint_vel = self.cartpole.data.joint_vel.torch

    time_out = self.episode_length_buf >= self.max_episode_length
    out_of_bounds = torch.any(torch.abs(self.joint_pos[:, self._cart_dof_idx]) > self.cfg.max_cart_pos, dim=1)
    return out_of_bounds, time_out
```

Once the indices for environments requiring reset have been computed, the `_reset_idx(self, env_ids)`
function performs the reset operations on those environments. Within this function, new states
for the environments requiring reset should be set directly into simulation.

```python
def _reset_idx(self, env_ids: Sequence[int] | None):
    if env_ids is None:
        env_ids = self.cartpole._ALL_INDICES

    # log the survival success rate before resetting (survived = timed out without terminating early)
    survived = self.reset_time_outs[env_ids].float()
    # no .item(): avoids a sync on every reset
    self.extras.setdefault("log", {})["Metrics/success_rate"] = survived.mean()

    super()._reset_idx(env_ids)

    joint_pos = self.cartpole.data.default_joint_pos.torch[env_ids].clone()
    joint_pos[:, self._cart_dof_idx] += sample_uniform(
        self.cfg.initial_cart_position_range[0],
        self.cfg.initial_cart_position_range[1],
        joint_pos[:, self._cart_dof_idx].shape,
        joint_pos.device,
    )
    joint_vel = self.cartpole.data.default_joint_vel.torch[env_ids].clone()
    joint_vel[:, self._cart_dof_idx] += sample_uniform(
        self.cfg.initial_cart_velocity_range[0],
        self.cfg.initial_cart_velocity_range[1],
        joint_vel[:, self._cart_dof_idx].shape,
        joint_vel.device,
    )
    joint_pos[:, self._pole_dof_idx] += sample_uniform(
        self.cfg.initial_pole_angle_range[0],
        self.cfg.initial_pole_angle_range[1],
        joint_pos[:, self._pole_dof_idx].shape,
        joint_pos.device,
    )
    joint_vel[:, self._pole_dof_idx] += sample_uniform(
        self.cfg.initial_pole_velocity_range[0],
        self.cfg.initial_pole_velocity_range[1],
        joint_vel[:, self._pole_dof_idx].shape,
        joint_vel.device,
    )

    # clamp the sampled state to the joint limits (matches the manager-based reset_joints_by_offset)
    joint_pos_limits = self.cartpole.data.soft_joint_pos_limits.torch[env_ids]
    joint_pos = joint_pos.clamp_(joint_pos_limits[..., 0], joint_pos_limits[..., 1])
    joint_vel_limits = self.cartpole.data.soft_joint_vel_limits.torch[env_ids]
    joint_vel = joint_vel.clamp_(-joint_vel_limits, joint_vel_limits)

    default_root_pose = self.cartpole.data.default_root_pose.torch[env_ids].clone()
    default_root_pose[:, :3] += self.scene.env_origins[env_ids]
    default_root_vel = self.cartpole.data.default_root_vel.torch[env_ids].clone()

    self.joint_pos[env_ids] = joint_pos
    self.joint_vel[env_ids] = joint_vel

    self.cartpole.write_root_pose_to_sim_index(root_pose=default_root_pose, env_ids=env_ids)
    self.cartpole.write_root_velocity_to_sim_index(root_velocity=default_root_vel, env_ids=env_ids)
    self.cartpole.write_joint_position_to_sim_index(position=joint_pos, env_ids=env_ids)
    self.cartpole.write_joint_velocity_to_sim_index(velocity=joint_vel, env_ids=env_ids)
```

### Applying Actions

There are two APIs that are designed for working with actions. The `_pre_physics_step(self, actions)` takes in actions
from the policy as an argument and is called once per RL step, prior to taking any physics steps. This function can
be used to process the actions buffer from the policy and cache the data in a class variable for the environment.

```python
def _pre_physics_step(self, actions: torch.Tensor) -> None:
    self.actions = self.action_scale * actions
```

The `_apply_action(self)` API is called `decimation` number of times for each RL step, prior to taking
each physics step. This provides more flexibility for environments where actions should be applied
for each physics step.

```python
def _apply_action(self) -> None:
    self.cartpole.set_joint_effort_target_index(target=self.actions, joint_ids=self._cart_dof_idx)
```

## The Code Execution

To run training for the direct workflow Cartpole environment, we can use the following command:

#### uv (Recommended)

```bash
uv run --extra rl-games isaaclab train --rl_library rl_games --task=Isaac-Cartpole-Direct
```

#### isaaclab.sh / isaaclab.bat

```bash
./isaaclab.sh train --rl_library rl_games --task=Isaac-Cartpole-Direct
```

![result of train.py](/isaaclab/_fern-img/3bd3eb489ede1537805972aad8f761c6ce19d2e854d29811619f356c681325bd.webp)

All direct workflow tasks have the suffix `-Direct` added to the task name to differentiate the implementation style.

## Domain Randomization

In the direct workflow, domain randomization configuration uses the `~isaaclab.utils.configclass` module
to specify a configuration class consisting of `~managers.EventTermCfg` variables.

Below is an example of a configuration class for domain randomization:

```python
@configclass
class EventCfg:
  robot_physics_material = EventTerm(
      func=mdp.randomize_rigid_body_material,
      mode="reset",
      params={
          "asset_cfg": SceneEntityCfg("robot", body_names=".*"),
          "static_friction_range": (0.7, 1.3),
          "dynamic_friction_range": (1.0, 1.0),
          "restitution_range": (1.0, 1.0),
          "num_buckets": 250,
      },
  )
  robot_joint_stiffness_and_damping = EventTerm(
      func=mdp.randomize_actuator_gains,
      mode="reset",
      params={
          "asset_cfg": SceneEntityCfg("robot", joint_names=".*"),
          "stiffness_distribution_params": (0.75, 1.5),
          "damping_distribution_params": (0.3, 3.0),
          "operation": "scale",
          "distribution": "log_uniform",
      },
  )
  reset_gravity = EventTerm(
      func=mdp.randomize_physics_scene_gravity,
      mode="interval",
      is_global_time=True,
      interval_range_s=(36.0, 36.0),  # time_s = num_steps * (decimation * dt)
      params={
          "gravity_distribution_params": ([0.0, 0.0, 0.0], [0.0, 0.0, 0.4]),
          "operation": "add",
          "distribution": "gaussian",
      },
  )
```

Each `EventTerm` object is of the `~managers.EventTermCfg` class and takes in a `func` parameter
for specifying the function to call during randomization, a `mode` parameter, which can be `startup`,
`reset` or `interval`. The `params` dictionary should provide the necessary arguments to the
function that is specified in the `func` parameter.
Functions specified as `func` for the `EventTerm` can be found in the `~envs.mdp.events` module.

Note that as part of the `"asset_cfg": SceneEntityCfg("robot", body_names=".*")` parameter, the name of
the actor `"robot"` is provided, along with the body or joint names specified as a regex expression,
which will be the actors and bodies/joints that will have randomization applied.

Once the `configclass` for the randomization terms have been set up, the class must be added
to the base config class for the task and be assigned to the variable `events`.

```python
@configclass
class MyTaskConfig:
  events: EventCfg = EventCfg()
```

### Action and Observation Noise

Actions and observation noise can also be added using the `~utils.configclass` module.
Action and observation noise configs must be added to the main task config using the
`action_noise_model` and `observation_noise_model` variables:

```python
@configclass
class MyTaskConfig:

    # at every time-step add gaussian noise + bias. The bias is a gaussian sampled at reset
    action_noise_model: NoiseModelWithAdditiveBiasCfg = NoiseModelWithAdditiveBiasCfg(
      noise_cfg=GaussianNoiseCfg(mean=0.0, std=0.05, operation="add"),
      bias_noise_cfg=GaussianNoiseCfg(mean=0.0, std=0.015, operation="abs"),
    )

    # at every time-step add gaussian noise + bias. The bias is a gaussian sampled at reset
    observation_noise_model: NoiseModelWithAdditiveBiasCfg = NoiseModelWithAdditiveBiasCfg(
      noise_cfg=GaussianNoiseCfg(mean=0.0, std=0.002, operation="add"),
      bias_noise_cfg=GaussianNoiseCfg(mean=0.0, std=0.0001, operation="abs"),
    )
```

`~.utils.noise.NoiseModelWithAdditiveBiasCfg` can be used to sample both uncorrelated noise
per step as well as correlated noise that is re-sampled at reset time.

The `noise_cfg` term specifies the Gaussian distribution that will be sampled at each
step for all environments. This noise will be added to the corresponding actions and
observations buffers at every step.

The `bias_noise_cfg` term specifies the Gaussian distribution for the correlated noise
that will be sampled at reset time for the environments being reset. The same noise
will be applied each step for the remaining of the episode for the environments and
resampled at the next reset.

If only per-step noise is desired, `~utils.noise.GaussianNoiseCfg` can be used
to specify an additive Gaussian distribution that adds the sampled noise to the input buffer.

```python
@configclass
class MyTaskConfig:
  action_noise_model: GaussianNoiseCfg = GaussianNoiseCfg(mean=0.0, std=0.05, operation="add")
```

In this tutorial, we learnt how to create a direct workflow task environment for reinforcement learning. We declared
the scene in configuration and extended the base environment with actions, dones, reset, reward, and observation
functions.

While it is possible to manually create an instance of `~isaaclab.envs.DirectRLEnv` class for a desired task,
this is not scalable as it requires specialized scripts for each task. Thus, we exploit the
`gymnasium.make` function to create the environment with the gym interface. We will learn how to do this
in the next tutorial.