Skip to content
Simulo

Cartpole Anomaly

Cartpole training with one deliberate fault: at a step you choose, environment 0 reports a NaN reward. DebugOnAnomaly watches the run, catches the bad reward the moment it appears, and writes a debug recording of the steps that led up to it. The fault is staged because real training blow-ups don’t happen on cue. The detection and the recording are the real platform code.

Save this file as cartpole_anomaly/train.py. It holds its own task, so it needs no task.py:

Show complete codeHide complete codecartpole_anomaly/train.py
cartpole_anomaly/train.py
from __future__ import annotations


import math
from typing import Any, Tuple


import simulo


# The cartpole robot - a validated, version-pinned global-catalog asset.
cartpole = simulo.Asset.from_registry("simulo/robot/cartpole:v1")


# Advanced: pick a different Simulo runtime with
# App("name", runtime=simulo.Runtime.from_registry("simulo/gpu-rl:2026.06"));
# see the Runtimes docs.
app = simulo.App("cartpole-anomaly")


with app.runtime.imports():
    import torch  # noqa: F401  (resolved only in execution mode, on the worker)


# This callback is declared on @app.job(callbacks=[...]) below and enabled
# automatically for the training run.
# window_steps bounds the pre-anomaly evidence buffer; max_captures=1 disarms
# the monitor after the first capture so a noisy run can never spam disk.
# video="side_cam" additionally buffers the side camera's frames for env 0
# (the watched env) so the debug MCAP shows what the sim LOOKED like in the
# window before the fault - at the documented cost of rendering during
# training (see the module docstring).
debug_cb = simulo.callbacks.DebugOnAnomaly(window_steps=64, check_every_n_steps=1, max_captures=1, video="side_cam")




class FaultInjectedCartpole(simulo.Task):
    """The cartpole balance task with a single-step NaN reward fault injector.


    Observation (4-dim): pole angle, pole angular velocity, cart position,
    cart velocity. Action (1-dim): scaled horizontal force on the cart.
    ``fault_step`` (a policy-step index; 0 disables) is set by the job body.
    """


    observation_dim = 4
    action_dim = 1


    episode_length_s = 5.0
    action_scale = 100.0  # [N]


    max_cart_pos = 3.0  # [m]
    initial_pole_angle_range = (-0.25, 0.25)  # fraction of pi [rad]


    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


    #: Policy step at which env 0's reward is poisoned with NaN (0 = never).
    fault_step = 0


    # Framework-injected at runtime by ``simulo.core.Task`` / ``LearningEnv``
    # (declared for the type checker only; PEP 563 strings, never evaluated).
    device: str
    max_episode_length: int
    episode_length_buf: torch.Tensor
    reset_terminated: torch.Tensor


    def build(self, scene: simulo.Scene) -> None:
        scene.add(simulo.Terrain.plane(name="ground"), at="/", per_environment=False)
        scene.add(
            simulo.Light.dome(name="light", intensity=2000.0, color=(0.75, 0.75, 0.75)),
            at="/",
            per_environment=False,
        )
        self.robot = simulo.Robot(asset=cartpole, initial_pose=simulo.Pose.identity())
        scene.add(self.robot, at="/World/Robot")


        # Add one non-replicated side camera for env 0. It supplies the
        # ``video="side_cam"`` evidence without adding a camera per environment.
        env0_slider = "/World/envs/env_0/Robot/slider" if scene.num_envs > 1 else "/World/Robot/slider"
        scene.add(
            simulo.Camera(
                width=640,
                height=480,
                data_types=["rgb"],
                update_period=1.0 / 30.0,  # 30 Hz render - ~1s of footage in a 64-step window
                offset=simulo.SensorOffset.look_at(pos=(-5.0, 0.0, 0.5), target=(0.0, 0.0, 0.5)),
                spawn=simulo.CameraSpawnConfig(
                    focal_length=18.0,  # wide enough to keep the cart's ±3 m travel in frame
                    focus_distance=400.0,
                    horizontal_aperture=20.955,
                    clipping_range=(0.1, 1.0e5),
                ),
            ),
            at=f"{env0_slider}/side_cam",
            per_environment=False,  # ONE render product total - never per-env replicated
        )


    def on_start(self, env: simulo.LearningEnv) -> None:
        self._cart_dof_idx = self.robot.find_joints("slider_to_cart")
        self._pole_dof_idx = self.robot.find_joints("cart_to_pole")
        self._policy_steps = 0


    def get_observations(self) -> torch.Tensor:
        pole_idx = self._pole_dof_idx[0]
        cart_idx = self._cart_dof_idx[0]
        joint_pos = self.robot.state.joint_positions
        joint_vel = self.robot.state.joint_velocities
        pole_pos = joint_pos[:, pole_idx].view(-1, 1)
        pole_vel = joint_vel[:, pole_idx].view(-1, 1)
        cart_pos = joint_pos[:, cart_idx].view(-1, 1)
        cart_vel = joint_vel[:, cart_idx].view(-1, 1)
        return torch.cat((pole_pos, pole_vel, cart_pos, cart_vel), dim=-1)


    def get_rewards(self) -> torch.Tensor:
        joint_pos = self.robot.state.joint_positions
        joint_vel = self.robot.state.joint_velocities
        pole_pos = joint_pos[:, self._pole_dof_idx[0]]
        pole_vel = joint_vel[:, self._pole_dof_idx[0]]
        cart_vel = joint_vel[:, self._cart_dof_idx[0]]
        terminated = self.reset_terminated.view(-1).float()


        reward = (
            self.rew_scale_alive * (1.0 - terminated)
            + self.rew_scale_terminated * terminated
            + self.rew_scale_pole_pos * torch.square(pole_pos.view(-1))
            + self.rew_scale_cart_vel * torch.abs(cart_vel.view(-1))
            + self.rew_scale_pole_vel * torch.abs(pole_vel.view(-1))
        )
        # Keep the (num_envs,) per-env reward contract even when num_envs == 1.
        reward = reward.view(-1)


        # --- THE DELIBERATE FAULT (illustrative; see module docstring) ------
        # One step, one env: a NaN reward, exactly what a broken reward kernel
        # or a physics blow-up feeding a reward term would produce.
        self._policy_steps += 1
        if self.fault_step and self._policy_steps == self.fault_step:
            reward[0] = float("nan")
        # ---------------------------------------------------------------------
        return reward


    def get_dones(self) -> Tuple[dict[str, torch.Tensor], torch.Tensor]:
        pole_idx = self._pole_dof_idx[0]
        cart_idx = self._cart_dof_idx[0]
        joint_pos = self.robot.state.joint_positions
        truncated = self.episode_length_buf >= self.max_episode_length - 1
        cart_out = torch.abs(joint_pos[:, cart_idx]) > self.max_cart_pos
        pole_fallen = torch.abs(joint_pos[:, pole_idx]) > math.pi / 2
        terminated = {"cart left the track": cart_out, "pole fell": pole_fallen}
        return terminated, truncated


    def apply_actions(self, actions: torch.Tensor) -> None:
        self.robot.set_joint_effort_target(self.action_scale * actions, joint_ids=self._cart_dof_idx)


    def reset_idx(self, env_ids: torch.Tensor) -> None:
        num_resets = len(env_ids)
        if num_resets == 0:
            return
        self.robot.reset(env_ids)
        pole_idx = self._pole_dof_idx[0]
        # Start from the default joint positions for the selected environments.
        joint_pos = self.robot.default_joint_positions[env_ids]
        random_angles = torch.empty(num_resets, device=self.device).uniform_(
            self.initial_pole_angle_range[0] * math.pi,
            self.initial_pole_angle_range[1] * math.pi,
        )
        joint_pos[:, pole_idx] += random_angles
        # Apply the randomized starting positions and default velocities together.
        joint_vel = self.robot.default_joint_velocities[env_ids]
        self.robot.set_joint_state(joint_pos, velocities=joint_vel, env_ids=env_ids)




@app.job(
    type="train",
    # Tier 1: T4 GPU, 16 GB VRAM. Run `simulo systems` for the full four-tier catalog.
    system=simulo.SystemType.TIER_1,
    timeout=1 * 60 * 60,
    callbacks=[debug_cb],
)
def train_with_fault(num_envs: int = 256, max_iterations: int = 12, fault_step: int = 185) -> dict[str, Any]:
    """Short PPO run with anomaly capture armed and a NaN fault injected.


    ``max_iterations=12`` is 192 policy steps (16 per PPO rollout); the default
    ``fault_step=185`` therefore lands inside the FINAL rollout window, so the
    visible training stays healthy end-to-end while the monitor still catches
    the poisoned reward the moment it appears.


    Returns the training stats. The bounded debug MCAP is available through
    the job's recordings and outputs.
    """
    task = FaultInjectedCartpole()
    task.fault_step = int(fault_step)


    env = simulo.LearningEnv(
        task=task,
        num_envs=num_envs,
        device="cuda",
        env_spacing=4.0,
        headless=True,
        seed=42,
        # Required by the video-evidence camera (debug_cb's video="side_cam"):
        # the simulator refuses to spawn Camera sensors unless camera
        # rendering is enabled, and this kwarg wires it through. THE
        # render-cost trade of video evidence lives on this line - remove it
        # together with the camera + video= to train render-free.
        enable_cameras=True,
    )
    trainer = simulo.RLTrainer(
        env=env,
        algorithm="PPO",
        device="cuda",
        seed=42,
        # The callback declared on @app.job(callbacks=[debug_cb]) above is
        # enabled automatically.
    )


    stats = trainer.train(max_iterations=max_iterations)


    trainer.close()
    env.close()


    return {"num_envs": num_envs, "fault_step": int(fault_step), **stats}
Show complete codeHide complete codecartpole_anomaly/eval.py
cartpole_anomaly/eval.py
from __future__ import annotations


import math


import simulo


from train import FaultInjectedCartpole, app as _training_app


app = simulo.App(
    _training_app.name, runtime=_training_app.runtime, mounts=_training_app.mounts
)
del _training_app




@app.job(type="eval", system=simulo.SystemType.TIER_1, timeout=60 * 60)
def evaluate(
    policy: simulo.PolicyCheckpoint,
    episodes: int = 100,
    num_envs: int = 100,
    actions: str = "best",
):
    return simulo.evaluate(
        FaultInjectedCartpole(),
        policy,
        episodes=episodes,
        num_envs=num_envs,
        actions=actions,
    )




@app.success
def balanced(task):
    """Count an episode when the pole remains upright until the time limit."""
    pole_angle = task.robot.state.joint_positions[
        :, task.robot.find_joints("cart_to_pole")[0]
    ]
    lasted = task.check("lasted to the time limit", task.survived())
    upright = task.check(
        "pole within 10 degrees of upright", pole_angle.abs() < math.radians(10)
    )
    return lasted & upright

Run it

Section titled “Run it”
Terminal window
simulo run cartpole_anomaly/train.py --num-envs 256 --max-iterations 12 --fault-step 185

These are also the job’s defaults. Each training iteration collects 16 steps, so 12 iterations run 192 steps and step 185 falls in the last one. The training you watch stays healthy until the very end.

A capture doesn’t stop training. The monitor records one capture, stops watching, and the job finishes with status completed. Then look at what it caught:

Terminal window
simulo jobs          # the OUTPUTS column shows: ⚠ 1 anomaly (non_finite_reward)
simulo result        # stderr reports: ⚠ 1 anomaly (non_finite_reward)
simulo recordings    # downloads recordings from your most recent job you submitted (train, play, preview or eval), including the capture
simulo outputs       # lists the same capture as an anomaly_capture output

The capture is an MCAP file. It holds the anomaly signal, the observations, rewards, and terminations of environment 0 for up to 64 steps before the fault, and video of those steps from a side camera.

Evaluate it

Section titled “Evaluate it”

The success rule counts an episode only when the pole lasts to the time limit and stays within 10 degrees of upright. Run simulo run cartpole_anomaly/eval.py --policy <policy-id>:best; see Evaluate a trained policy.

What to reuse

Section titled “What to reuse”
  • Declare the callback on the job with @app.job(callbacks=[debug_cb]). The platform attaches it when the job runs, so the job body never wires it into the trainer.
  • Keep window_steps and max_captures small so a noisy run can’t fill the disk with recordings.
  • Add video only when you need to see the scene. The camera turns rendering on for the whole run, which slows every step, and it needs enable_cameras=True on the environment. Drop both to capture numbers only.

See Callbacks and anomaly capture for every DebugOnAnomaly option and Recordings & live view for opening the MCAP file.