Cartpole Anomaly
Inject one non-finite reward into Cartpole training and let DebugOnAnomaly
capture the signals and video leading up to it. The fault is deliberate; the
detection, bounded evidence window, and MCAP output are the real runtime path.
Save this complete app as cartpole_anomaly/app.py:
Show complete codecartpole_anomaly/app.py
from __future__ import annotations
import mathfrom 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)
# ONE callback object, declared on @app.job(callbacks=[...]) below — recorded# in the manifest and honored automatically by the platform runner; the job# body never wires it into RLTrainer by hand.# 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")
# The anomaly monitor's video evidence source (debug_cb's # ``video="side_cam"``): a static side-view camera on the cartpole's # fixed ``slider`` (rail) link, watching ONLY env 0 — the same # framing cartpole_eval's rollout camera uses, but a SINGLE render # product regardless of ``num_envs``. # # A corrected-bug note: an earlier version attached this camera via # ``self.robot.add_sensor(camera, attach_to=...)`` BEFORE # ``scene.add(self.robot, ...)``. That inherits the ROBOT's # ``per_environment=True`` (the default), so ``Scene._add_robot`` registers # the sensor once PER ENV — at the walkthrough's documented # ``--num-envs 256`` that is 256 640x480 render products, which # exhausts the Vulkan descriptor-set pool on real GPUs (observed on # an RTX 3090: `Unable to allocate descriptor sets` / `HydraEngine:: # render failed`, job wedges at simulator init). ``Robot.add_sensor`` has # no per-env-index option, so instead this camera is a STANDALONE # scene item (``scene.add(camera, ..., per_environment=False)``), added # AFTER the robot (no sensor-enumeration-timing constraint applies to # a standalone item). Its ``at=`` path targets the ALREADY-CLONED # link of ENV 0's ROBOT specifically — the simulator's env clone # namespace is the stable, version-pinned constant # ``/World/envs/env_{N}`` # once ``num_envs > 1``; below that this task's own scene path # (``/World/Robot``) is used verbatim (no env cloning happens for a # single env). Either way this gives ONE camera, parented to a REAL # existing robot link, framing it exactly as before. The anomaly # monitor's camera lookup (``AnomalyMonitor._resolve_video_camera``) # walks ``scene.get_items()`` by type + name regardless of # replication, so it finds this standalone camera unmodified. # # ``simulo.Camera`` / ``simulo.SensorOffset`` / ``simulo.CameraSpawnConfig`` # are on the torch-free ``simulo.*`` surface (the one-import rule's second # half) — the same lazy, mode-aware resolution ``simulo.Robot`` / ``simulo.Scene`` # use, so no import is needed and this stays torch-free at discovery whether or # not ``build()`` ever runs (it never does at submit). 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") # robot.state is the supported, typed way to read live state (robot.internals # is the unstable engine escape hatch — see the Scene, Robot & World docs). self._joint_pos = self.robot.state.joint_positions self._joint_vel = self.robot.state.joint_velocities self._policy_steps = 0
def get_observations(self) -> torch.Tensor: pole_idx = self._pole_dof_idx[0] cart_idx = self._cart_dof_idx[0] pole_pos = self._joint_pos[:, pole_idx].view(-1, 1) pole_vel = self._joint_vel[:, pole_idx].view(-1, 1) cart_pos = self._joint_pos[:, cart_idx].view(-1, 1) cart_vel = self._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: pole_pos = self._joint_pos[:, self._pole_dof_idx[0]] pole_vel = self._joint_vel[:, self._pole_dof_idx[0]] cart_vel = self._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[torch.Tensor, torch.Tensor]: self._joint_pos = self.robot.state.joint_positions self._joint_vel = self.robot.state.joint_velocities pole_idx = self._pole_dof_idx[0] cart_idx = self._cart_dof_idx[0] truncated = self.episode_length_buf >= self.max_episode_length - 1 cart_out = torch.abs(self._joint_pos[:, cart_idx]) > self.max_cart_pos pole_fallen = torch.abs(self._joint_pos[:, pole_idx]) > math.pi / 2 terminated = cart_out | 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] # robot.state has no default-joint-value equivalent, so this stays on the # internals escape hatch (there is nothing unstable about reading it here, # just no supported, typed name for it yet). joint_pos = self.robot.internals.default_joint_pos[env_ids].clone() 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 # set_joint_state writes both positions and velocities through the engine's # own command path (Articulation.write_joint_state_to_sim -> # write_joint_{position,velocity}_to_sim), which updates robot.state's # backing buffers in place AND pushes to the physics view in the same call. # self._joint_pos / self._joint_vel are the SAME objects as those buffers # (the hold-safety contract on core/robot.py), so they are already current # after this call — no separate write into either tensor is needed, and # robot.state's contract is never write into a member's tensor directly. joint_vel = self.robot.internals.default_joint_vel[env_ids] self.robot.set_joint_state(joint_pos, velocities=joint_vel, env_ids=env_ids)
@app.job( # 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, including ``anomaly_captures`` — the bounded debug MCAP path(s) the monitor wrote. """ task = FaultInjectedCartpole() task.fault_step = int(fault_step)
env = simulo.LearningEnv( task=task, num_envs=num_envs, device="cuda", dt=1.0 / 120.0, physics_steps_per_action=2, 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, # No debug_on_anomaly= here: the platform runner honors the callback # declared on @app.job(callbacks=[debug_cb]) above on its own — see # the module docstring. )
stats = trainer.train(max_iterations=max_iterations)
trainer.close() env.close()
return {"num_envs": num_envs, "fault_step": int(fault_step), **stats}Run the app
Section titled “Run the app”simulo run cartpole_anomaly/app.py --num-envs 256 --max-iterations 12 --fault-step 185At the selected step, environment 0 emits a NaN reward. The monitor writes one
bounded debug recording, then disarms. The job completes and returns the capture
path in anomaly_captures.
What to reuse
Section titled “What to reuse”- Declare
DebugOnAnomalyon the job so the callback is recorded in the app manifest and attached automatically at execution. - Keep
window_stepsandmax_capturesbounded so debugging cannot grow without limit. - Add
videoonly when visual context is worth enabling camera rendering.
See Callbacks for callback options and Recordings & Live Viewstream for working with the captured MCAP.