State Machines for Robot Behavior

Tutorial

Note

These tutorials replace the deprecated Isaac Cortex behavior-programming examples. They demonstrate the same kinds of reactive manipulation tasks (Franka block stacking, UR10 bin palletizing) using application-owned Python libraries and the native Omniverse Behavior Tree runtime instead of a custom decision framework. For release-level guidance, see Cortex behavior workflow migration. See Migrating from Cortex at the bottom of this page for a mapping between Cortex concepts and the patterns used here.

Prerequisites

• Review the Controllers and the RobotState tutorial to understand RobotState, BaseController, and controller composition.

Running the examples

All command line examples are written relative to the Isaac Sim root and use the Isaac Sim Python launcher. On Linux, use ./python.sh. On Windows, use python.bat instead.

The standalone Python examples for these tutorials are packaged under standalone_examples/tutorials/state_machine/.

When to reach for a state machine

When you first write a controller for a multi-step task, a few if/elif branches are usually enough: check what the robot just finished, then decide what to do next. But as the task grows – more steps, more recovery cases – that branching piles up inside one class, and the code gets hard to read and harder to extend.

A state machine is the usual answer. It models the robot as being in exactly one named state at a time, with explicit rules for which states transition to which others. Each state and transition gets its own labeled spot instead of another branch in a growing chain. Reach for one when:

• the task has more than a few ordered steps (approach, grasp, lift, transport, place, …), or

• an event should interrupt the current action sequence, such as re-grasping a dropped object or replanning around a moved obstacle.

This page is in two parts:

1. A state-machine-free controller (franka_pick_place_ifelse.py). One controller class owns the whole task. It works, but it shows why growing if/elif logic becomes hard to maintain.

2. Separating task logic from control. The Python Franka examples share one controller stack; the UR10 example uses its own controllers. Four approaches organize the task logic differently.

   • A library-free FSM (franka_pick_place_fsm.py). PickPlacePhase names each phase, PickPlaceState stores one transfer’s data, and FrankaStackingFSM owns the task decisions and transitions. Best when the task is a fixed sequence of phases and you want no extra dependencies.

   • A behavior tree with py_trees (franka_pick_place_py_trees.py). Each motion or gripper action is a reusable leaf Behaviour; recovery branches sit at the top of the tree so they automatically interrupt normal flow. Best when several recovery branches must preempt each other in a clear priority order.

   • A USD-native behavior tree with Omniverse Behavior Tree (franka_pick_place.json). Registered Python nodes provide robot operations; JSON composes them into behavior that runs from the USD stage and is inspectable in the Behavior Tree UI.

   • A declarative FSM with transitions (ur10_palletizing_transitions.py). A UR10 picks bins off a conveyor; some arrive upside-down, so the robot either goes straight to the pallet or detours through a flip jig. That fork lives as one row in a transition table instead of an if/else in step code. Best when the graph is small but a few edges branch on world state.

A state-machine-free controller

The 4-cube stacker (red -> blue -> green -> yellow), written without a separate state-machine abstraction. FrankaStackingIfElseController owns both task decisions and robot control: it advances a phase with if/elif logic, drives the arm controller, and merges arm and finger commands. Task context is spread across instance attributes such as _color, _phase, _pickup_position, and _aborted.

forward(): decide, react, and drive the arm and hand.

    def forward(
        self,
        estimated_state: mg.RobotState,
        setpoint_state: mg.RobotState | None,
        t: float,
        *,
        observation: Observation,
        **kwargs: object,
    ) -> mg.RobotState | None:
        """Decide, react, and drive the arm and hand -- all in one method."""
        target_reached = self._target_reached(observation)
        gripper_reached = self._gripper_reached(observation)

        # Decide what to work on, react to the world, and advance the phase.
        if (self._color is None or self._phase >= len(_PHASES)) and not self._select_next_cube(observation):
            self.done = True
            self._color = None
        elif self._cube_was_dropped(observation):
            print(f"[drop] {self._color} fell out of gripper, re-decide")
            self._color = None
        else:
            self._restart_pickup_if_cube_moved(observation)
            self._advance_if_ready(observation, target_reached, gripper_reached)

        # Turn the current phase into an arm target and a gripper command, run
        # the arm controller, and merge the finger command into its output.
        phase = _PHASES[self._phase] if self._is_active() else "release"
        self._target = self._compute_target(observation) if self._is_active() else None
        self._gripper_open = phase not in _CLOSED_PHASES
        setpoint = None if self._target is None else self._arm_setpoint(self._target)
        arm_state = self._arm_controller.forward(estimated_state, setpoint, t, **kwargs)
        gripper_state = self._gripper_command(self._gripper_open)
        out_state = mg.RobotState()
        for state in (arm_state, gripper_state):
            out_state = mg.combine_robot_states(out_state, state)
            if out_state is None:
                return None
        return out_state

The controller does three jobs every tick:

1. Grade the previous command: did the arm reach its target, and did the gripper reach its width?

2. Decide and react: choose the next cube, recover from moved or dropped cubes, and advance the phase.

3. Drive the robot: turn the current phase into an arm target and a gripper width, run the arm controller, and merge the finger command into its output.

This works, but it scales poorly:

• Task context is spread across attributes instead of grouped in a state object.

• Each new phase or recovery case adds another branch.

This controller is a useful starting point, but new behavior keeps growing the same class. The next section separates task logic from robot control before introducing the state-machine examples.

Run it

./python.sh standalone_examples/tutorials/state_machine/franka_pick_place_ifelse.py

Press Play, then drag a cube mid-stack to trigger the reactive checks.

Separating control from task logic

A structured design separates small controllers that move the robot from the task policy that selects each phase. The controllers do not make task decisions, and the task policy does not access robot APIs.

The Franka examples build the control side once, then pair it with either the library-free FSM or the behavior tree shown below.

Small, single-purpose controllers

GripperController commands the two Franka finger joints to one fixed width. The task policy decides when to open or close:

GripperController: command the hand to one width.

class GripperController(mg.BaseController):
    """Command the Franka finger joints to one fixed position."""

    def __init__(self, *, joint_space: list[str], finger_joint_names: tuple[str, ...], target_position: float) -> None:
        self._joint_space = joint_space
        self._finger_names = [name for name in finger_joint_names if name in joint_space]
        self._positions = wp.array([target_position] * len(self._finger_names), dtype=wp.float32)

    def reset(
        self, estimated_state: mg.RobotState, setpoint_state: mg.RobotState | None, t: float, **kwargs: object
    ) -> bool:
        return True

    def forward(
        self, estimated_state: mg.RobotState, setpoint_state: mg.RobotState | None, t: float, **kwargs: object
    ) -> mg.RobotState:
        return mg.RobotState(
            joints=mg.JointState.from_name(
                robot_joint_space=self._joint_space,
                positions=(self._finger_names, self._positions),
            )
        )

The arm controller uses the mg.BaseController interface; these examples use the cuMotion RmpFlowController.

Two CombinedController instances pair arm motion with open and closed gripper commands.

Pair the arm with an open or closed gripper.

        # Pair the arm controller with an open or closed gripper. Each pair is a
        # single controller that moves the arm and holds the hand at one width.
        open_gripper = GripperController(
            joint_space=joint_space,
            finger_joint_names=_FINGER_JOINTS,
            target_position=config.open_position,
        )
        closed_gripper = GripperController(
            joint_space=joint_space,
            finger_joint_names=_FINGER_JOINTS,
            target_position=config.closed_position,
        )
        open_and_move = mg.CombinedController([arm_controller, open_gripper])
        close_and_move = mg.CombinedController([arm_controller, closed_gripper])

Selecting controllers with a SelectableController

SelectableController stores controllers by phase key:

Build the container: one controller per phase.

        # Map every phase to the pair it needs. The container swaps between them
        # by enum key and resets each controller the first time it becomes active.
        self._controllers = mg.SelectableController(
            controller_options={
                PickPlacePhase.PRE_GRASP: open_and_move,
                PickPlacePhase.APPROACH: open_and_move,
                PickPlacePhase.GRASP: close_and_move,
                PickPlacePhase.LIFT: close_and_move,
                PickPlacePhase.TRANSPORT: close_and_move,
                PickPlacePhase.LOWER: close_and_move,
                PickPlacePhase.RELEASE: open_and_move,
                PickPlacePhase.RETRACT: open_and_move,
            },
            initial_controller_selection=PickPlacePhase.PRE_GRASP,
        )

Each tick, PickPlaceController passes the policy’s phase to the policy-independent selector, which runs the mapped controller:

The one line that couples task logic to control.

        # The policy chose a phase; the container maps it to the matching
        # controller and runs it. This is the only line that couples the
        # task decision to the low-level controllers.
        self._controllers.set_next_controller(phase)
        setpoint = None if self._target is None else self._setpoint(self._target)
        return self._controllers.forward(estimated_state, setpoint, t, **kwargs)

The Franka FSM and behavior-tree examples share FrankaStackingSceneIO and PickPlaceController; only the task policy differs. The UR10 example uses the same separation with its own controller sequences.

Library-free FSM

Same 4-cube stacker demo as above, now with a state machine. The integer phase counter is replaced by named PickPlacePhase enum values. The PickPlaceState dataclass stores the active cube, destination, current phase, timing, and cached snapshots that recovery checks need. FrankaStackingFSM owns this state and returns a PickPlaceDecision. It has no scene or robot-controller access. PickPlaceController – the controller built above – evaluates completion of its previous command, runs the FSM, and hands the resulting phase to the container.

PickPlaceState: active-transfer data.

@dataclass
class PickPlaceState:
    """State for one active cube transfer."""

    color: str
    destination: np.ndarray
    phase: PickPlacePhase
    phase_started: float
    settle_started: float | None = None
    pickup_position: np.ndarray | None = None
    cube_reference_xy: np.ndarray | None = None
    aborted: bool = False

    @property
    def holding(self) -> bool:
        return self.phase in {PickPlacePhase.LIFT, PickPlacePhase.TRANSPORT, PickPlacePhase.LOWER}

The eight enum values are PRE_GRASP, APPROACH, GRASP, LIFT, TRANSPORT, LOWER, RELEASE, and RETRACT. The controller uses the live cube pose while the robot approaches, then switches to the frozen pickup_position snapshot after the grasp.

Eight-phase pick-and-place ring. Drop detection clears the active transfer while the robot holds a cube; pickup resync returns to PRE_GRASP if the cube moves during the approach.

The reactive checks live at the top of forward(); they run before the regular phase dispatch:

forward(): per-step reactive checks.

    def forward(self, observation: Observation, *, target_reached: bool, gripper_reached: bool) -> PickPlaceDecision:
        """Advance once from an observation and controller feedback."""
        if self._active is None:
            self._active = self._select_next_cube(observation)
            if self._active is None:
                return PickPlaceDecision(done=True)

        if self._cube_was_dropped(observation):
            print(f"[drop] {self._active.color} fell out of gripper, re-decide")
            self._active = None
            return PickPlaceDecision()

        self._restart_pickup_if_cube_moved(observation)
        self._advance_if_ready(observation, target_reached, gripper_reached)
        return self._decision()

Two per-tick reactive checks run here:

• Drop detection: if we’re holding a cube and the cube has drifted > 10 cm from the EE link, the gripper lost it. Clear the active task, let the next forward() re-decide. The controller emits a safe hold/open command.

• Pickup resync: if we’re approaching a cube that’s still on the ground (no pickup_position yet) and the cube has been pushed > 5 cm from where we initially saw it, return to PRE_GRASP so the descent re-aims from a clean start.

A third reactive check, stack-validity check, runs on each named phase advance rather than every tick:

_advance: phase-boundary stack-health check.

    def _advance(self, observation: Observation, reason: str = "ok") -> None:
        active = self._active
        assert active is not None
        print(f"  [{active.color}] {active.phase.name.lower()} done ({reason})")
        if active.phase is PickPlacePhase.RETRACT:
            self._active = None
            return

        active.phase = self._NEXT_PHASE[active.phase]
        active.phase_started = observation.time
        active.settle_started = None
        if active.phase is PickPlacePhase.GRASP:
            active.pickup_position = observation.cube_positions[active.color].copy()
        if active.holding and not active.aborted and not self._stack_below_is_valid(active.color, observation):
            print(f"[abort] stack invalid below {active.color}, returning to spawn")
            active.destination = self._config.spawn_positions[active.color].copy()
            active.aborted = True

On each phase advance, if we’re holding a cube and any cube below it in the stack order has been displaced:

1. Retarget the destination to that cube’s spawn so the held cube is dropped safely.

2. After the active transfer finishes, _select_next_cube() picks up the displaced cube and re-stacks.

When RETRACT completes, the FSM clears PickPlaceState so the next forward() selects the next cube.

Run it

./python.sh standalone_examples/tutorials/state_machine/franka_pick_place_fsm.py

Press Play in the viewport. Drag a cube while the robot is mid-stack to see the stack recovery path return the held cube before re-stacking.

Behavior tree with py_trees

Same 4-cube stacker demo and FrankaStackingSceneIO backend, implemented as a py_trees behavior tree. Key concepts:

• Tree of nodes. Each leaf Behaviour returns RUNNING, SUCCESS, or FAILURE on every tick.

• Composite nodes (Sequence, Selector) decide which child to run next from those return values.

• Blackboard: a shared key-value store any node can read or write, used here for pick_target, holding, and pickup_position.

Two reusable leaf Behaviour classes cover the demo:

• Phase – all motion and gripper phases.

• Action – blackboard updates.

Leaf Behaviour classes for motion and gripper actions.

class Phase(py_trees.behaviour.Behaviour):
    """Run one phase using the current observation and prior-command feedback."""

    def __init__(self, context: _Context, phase: PickPlacePhase, *, spawn: bool = False, drift: bool = False) -> None:
        super().__init__(phase.name.title())
        self.context = context
        self.phase = phase
        self.spawn = spawn
        self.drift = drift
        self.bb = py_trees.blackboard.Client(name=self.name)
        for key in ("pick_target", "holding", "pickup_position"):
            self.bb.register_key(key, access=py_trees.common.Access.READ)
            self.bb.register_key(key, access=py_trees.common.Access.WRITE)

    def initialise(self) -> None:
        """Reset phase timing."""
        self._started = self.context.observation.time
        self._settle_started: float | None = None
        self._anchor_xy: np.ndarray | None = None
        self._first_tick = True
        if self.phase is PickPlacePhase.GRASP:
            color = self.bb.pick_target
            self.bb.pickup_position = self.context.observation.cube_positions[color].copy()

    def update(self) -> py_trees.common.Status:
        """Emit the phase decision and report progress."""
        observation = self.context.observation
        color = self.bb.holding if self.spawn else self.bb.pick_target
        self._publish_decision(color)
        if self._pickup_target_drifted(observation, color):
            print(f"  [{self.name}] target drifted, restarting pickup")
            return S.FAILURE
        return S.SUCCESS if self._phase_finished(observation) else S.RUNNING

    def _publish_decision(self, color: str) -> None:
        destination = (
            self.context.config.spawn_positions[color] if self.spawn else self.context.config.stack_positions[color]
        )
        pickup = self.bb.pickup_position if self.bb.pickup_position is not None else None
        self.context.decision = PickPlaceDecision(
            phase=self.phase,
            color=color,
            destination=destination,
            pickup_position=pickup,
        )

    def _pickup_target_drifted(self, observation: Observation, color: str) -> bool:
        if not self.drift:
            return False
        current_xy = observation.cube_positions[color][:2]
        if self._anchor_xy is None:
            self._anchor_xy = current_xy.copy()
            return False
        return bool(np.linalg.norm(current_xy - self._anchor_xy) > self.context.config.cube_move_tolerance)

    def _phase_finished(self, observation: Observation) -> bool:
        elapsed = max(0.0, observation.time - self._started)
        if self.phase is PickPlacePhase.GRASP:
            return elapsed >= self.context.config.minimum_duration
        if self.phase is PickPlacePhase.RELEASE:
            return elapsed >= self.context.config.minimum_duration and self.context.gripper_reached

        if not self._first_tick and self.context.target_reached:
            if self._settle_started is None:
                self._settle_started = observation.time
        else:
            self._settle_started = None
        self._first_tick = False
        settled = (
            self._settle_started is not None
            and observation.time - self._settle_started >= self.context.config.settle_duration
        )
        short = self.phase in {
            PickPlacePhase.APPROACH,
            PickPlacePhase.LIFT,
            PickPlacePhase.TRANSPORT,
            PickPlacePhase.LOWER,
        }
        timeout = self.context.config.short_phase_timeout if short else self.context.config.phase_timeout
        if elapsed >= self.context.config.minimum_duration and settled:
            return True
        if elapsed >= timeout:
            print(f"  [{self.name}] timeout, advancing")
            return True
        return False


class Action(py_trees.behaviour.Behaviour):
    """Run one blackboard action."""

    def __init__(self, name: str, action: Callable[[], None]) -> None:
        super().__init__(name)
        self._action = action

    def update(self) -> py_trees.common.Status:
        """Run the action."""
        self._action()
        return S.SUCCESS

Each pick-and-place action is one leaf node in the tree. The composition lives in build_tree:

build_tree: root Selector, recovery branches, normal flow.

def build_tree(context: _Context) -> py_trees.trees.BehaviourTree:
    """Build the reactive stacking tree."""
    bb = py_trees.blackboard.Client(name="stacking")
    for key in ("pick_target", "holding", "pickup_position"):
        bb.register_key(key, access=py_trees.common.Access.READ)
        bb.register_key(key, access=py_trees.common.Access.WRITE)
    bb.pick_target, bb.holding, bb.pickup_position = "", None, None

    phase = lambda value, **kwargs: Phase(context, value, **kwargs)
    action = lambda name, fn: Action(name, fn)

    def clear_dropped() -> None:
        bb.holding = None
        context.decision = PickPlaceDecision()

    pickup = py_trees.composites.Sequence(
        "Pickup",
        memory=True,
        children=[phase(PickPlacePhase.PRE_GRASP), phase(PickPlacePhase.APPROACH, drift=True)],
    )
    place = py_trees.composites.Sequence(
        "Place",
        memory=True,
        children=[
            phase(PickPlacePhase.LIFT),
            phase(PickPlacePhase.TRANSPORT),
            phase(PickPlacePhase.LOWER),
            phase(PickPlacePhase.RELEASE),
            phase(PickPlacePhase.RETRACT),
            action("ClearHolding", lambda: setattr(bb, "holding", None)),
        ],
    )
    normal = py_trees.composites.Sequence(
        "Normal",
        memory=True,
        children=[
            SelectNextCube(context, bb),
            pickup,
            phase(PickPlacePhase.GRASP),
            action("MarkHolding", lambda: setattr(bb, "holding", bb.pick_target)),
            place,
        ],
    )
    drop_recovery = py_trees.composites.Sequence(
        "DropRecovery",
        memory=True,
        children=[
            CubeDropped(context, bb),
            action("ClearDropped", clear_dropped),
        ],
    )
    stack_recovery = py_trees.composites.Sequence(
        "StackRecovery",
        memory=True,
        children=[
            StackBelowBroken(context, bb),
            phase(PickPlacePhase.LIFT, spawn=True),
            phase(PickPlacePhase.TRANSPORT, spawn=True),
            phase(PickPlacePhase.LOWER, spawn=True),
            phase(PickPlacePhase.RELEASE, spawn=True),
            phase(PickPlacePhase.RETRACT, spawn=True),
            action("ClearHolding", lambda: setattr(bb, "holding", None)),
        ],
    )
    root = py_trees.composites.Selector(
        "Root",
        memory=False,
        children=[AllStacked(context, bb), drop_recovery, stack_recovery, normal],
    )
    tree = py_trees.trees.BehaviourTree(root)
    tree.setup()
    return tree

The tree’s shape is the standard reactive-tree pattern:

• Root is a Selector(memory=False). A Selector tries its children left-to-right and stops at the first one that returns RUNNING or SUCCESS; memory=False means it restarts from the leftmost child every tick instead of resuming where it left off. Recovery branches placed left of normal flow therefore win automatically the moment they activate.

• Sequence nodes (used inside the recovery branches and the Pickup, Place, and Normal subtrees) run their children left-to-right and fail as soon as one child fails. One FAILURE aborts the whole subtree, which lets us turn any leaf’s failure into a clean restart.

• AllStacked is the leftmost terminal condition.

• Drop recovery is the highest-priority recovery branch: CubeDropped? -> ClearDropped. The tree controller emits the safe hold/open command.

• Stack recovery sits between drop recovery and normal flow: StackBelowBroken? -> drive to spawn, release.

• Normal flow is the rightmost branch: SelectNextCube writes the blackboard target, then Pickup, Grasp, and Place run the transfer.

Each Behaviour declares which blackboard keys it reads and writes, so the data flow is explicit.

Cube-drift handling uses Phase(drift=True) as a normal-flow restart:

1. The cube moves past the drift threshold – leaf returns FAILURE.

2. Enclosing Sequence propagates the failure up.

3. Next tick the root Selector re-enters the normal branch from scratch, restarting the pickup.

Behavior tree for the Franka stacker. Recovery branches sit left of normal flow so they preempt automatically when their conditions fire.

Run it

./python.sh standalone_examples/tutorials/state_machine/franka_pick_place_py_trees.py

The console prints the unicode tree every 60 frames; watch the [*] status indicators move as branches activate.

Behavior tree with Omniverse Behavior Tree

Omniverse Behavior Tree provides a Kit-native runtime, JSON descriptors, typed ports and blackboards, USD attachment, and a visual editor. In this example, two Frankas dismantle a six-cube pyramid and each build a three-cube stack. Python nodes handle robot actions and coordination, while one reusable JSON descriptor defines the task.

See the Behavior Tree User Guide for authoring and debugging, and the API Reference for custom node libraries.

This example highlights five useful patterns:

• Custom node library. Python nodes use cuMotion RMPflow for motion and provide gripper control, cube reservation, task checks, and recovery. The @node decorator declares each node and its typed ports. The on_tick method returns RUNNING, SUCCESS, or FAILURE.

• One shared descriptor. franka_pick_place.json composes those nodes with Selector, Sequence, Repeat, and Timeout. Both Frankas run separate instances of this same file.

• Local and shared blackboards. Local variables hold each robot’s current cube, target paths, and recovery keys. /BehaviorTreeBlackboard holds cube reservations, the center-workspace lock, and placement counts shared by both robots. Motion and check ports bind to local paths in JSON; coordination nodes access shared state directly.

• Reactive recovery. Motion, gripper, and verification nodes set a per-robot flag when a cube moves or drops. A higher-priority branch uses CheckBlackboard in lower-priority abort mode to wait for the cube to settle, stage above its new pose, and retry the same reservation. Recovery retries nonconverging targets, logs a warning every five seconds, and resumes pickup when the cube becomes reachable.

• USD-managed execution. BehaviorTreeAPI on /World/franka_left and /World/franka_right links the JSON and shared blackboard. Pressing Play creates and ticks both instances; the Behavior Tree window shows their live status.

FrankaAcquireCube: reserve shared work and populate local state.

@node(
    type="FrankaAcquireCube",
    doc="Atomically reserve the next pyramid cube and publish robot-local motion targets.",
    keywords=["franka", "coordination", "blackboard", "reservation"],
)
class FrankaAcquireCube(_FrankaNodeBase):
    """Reserve one shared cube and configure target Xforms for this robot."""

    def on_tick(self, context: IBehaviorActionNodeContext) -> NodeStatus:
        """Acquire the center-zone lock and reserve the next available cube."""
        if self._resolve_robot() is None:
            return self._status_while_resolving()

        shared = context.get_blackboard()
        local = _get_local_blackboard(context)
        if local is None:
            carb.log_error("[FrankaAcquireCube] Tree-local blackboard is unavailable")
            return NodeStatus.FAILURE

        robot_id = _get_robot_id(self.get_prim_path())
        current_cube = _get_local_string(local, "current_cube")
        center_owner = str(_blackboard_get(shared, "center_owner", "") or "")
        if center_owner and center_owner != robot_id:
            return NodeStatus.RUNNING

        selected_new_cube = not current_cube
        if selected_new_cube:
            cube_paths = _PYRAMID_CUBE_PATHS
            if robot_id == "right":
                cube_paths = tuple(_PYRAMID_CUBE_PATHS[index] for index in (0, 2, 1, 5, 4, 3))
            for cube_path in cube_paths:
                state_key = f"cube_state:{cube_path}"
                if _blackboard_get(shared, state_key, "available") == "available":
                    current_cube = cube_path
                    break
        if not current_cube:
            return NodeStatus.FAILURE

        placed_count = int(_blackboard_get(shared, f"placed_count:{robot_id}", 0))
        try:
            cube_position = RigidPrim(current_cube).get_world_poses()[0].numpy()[0]
            stack_y = -0.28 if robot_id == "left" else 0.28
            goal_position = np.asarray([0.46, stack_y, 0.025 + 0.05 * placed_count], dtype=np.float32)
            target_root = f"/World/Targets/{robot_id}"
            target_positions = {
                "approach_target": cube_position + np.asarray([0.0, 0.0, 0.20]),
                "grasp_target": cube_position + np.asarray([0.0, 0.0, 0.09]),
                "lift_target": np.asarray([cube_position[0], cube_position[1], 0.35]),
                "place_approach_target": goal_position + np.asarray([0.0, 0.0, 0.325]),
                "place_target": goal_position + np.asarray([0.0, 0.0, 0.125]),
                "retract_target": goal_position + np.asarray([0.0, 0.0, 0.325]),
                "object_goal": goal_position,
            }
            target_paths = {name: f"{target_root}/{name}" for name in target_positions}
            for name, position in target_positions.items():
                XformPrim(target_paths[name]).set_world_poses(positions=position)
        except (IndexError, RuntimeError, ValueError) as exc:
            carb.log_error(f"[FrankaAcquireCube] Failed to prepare targets for '{current_cube}': {exc}")
            return NodeStatus.FAILURE

        if selected_new_cube:
            shared[f"cube_state:{current_cube}"] = f"reserved:{robot_id}"
            shared["center_owner"] = robot_id
            local["current_cube"] = current_cube
        local["recovery_key"] = f"recovery_requested:{robot_id}"
        local["completion_key"] = f"stack_complete:{robot_id}"
        if local["completion_key"] not in shared:
            shared[local["completion_key"]] = False
        for name, target_path in target_paths.items():
            local[name] = target_path
        _set_pickup_reference(local, cube_position)
        shared[local["recovery_key"]] = False
        return NodeStatus.SUCCESS

FrankaRecoverCube: move back above a moved or dropped cube.

@node(
    type="FrankaRecoverCube",
    doc="Restage above a cube that moved during pickup or fell from the gripper.",
    keywords=["franka", "manipulation", "recovery", "reactive"],
    ports=[
        value_port("movement_tolerance", 0.008),
        value_port("staging_height", 0.20),
        value_port("position_tolerance", 0.025),
        value_port("stable_duration", 0.20),
        value_port("warning_interval", 5.0),
        value_port("end_effector_yaw_degrees", 90.0),
    ],
)
class FrankaRecoverCube(_FrankaNodeBase):
    """Reactive recovery branch for moved or dropped reserved cubes."""

    def __init__(self) -> None:
        super().__init__()
        self._last_cube_position: np.ndarray | None = None
        self._stable_elapsed = 0.0
        self._warning_elapsed = 0.0
        self._motion_started = False

    def on_init(self, context: IBehaviorActionNodeContext) -> None:
        """Initialize recovery stability tracking."""
        super().on_init(context)
        self._last_cube_position = None
        self._stable_elapsed = 0.0
        self._warning_elapsed = 0.0
        self._motion_started = False

    def on_reset(self, reason: NodeResetReason) -> None:
        """Release robot state and reset recovery stability tracking."""
        super().on_reset(reason)
        self._last_cube_position = None
        self._stable_elapsed = 0.0
        self._warning_elapsed = 0.0
        self._motion_started = False

    def on_tick(self, context: IBehaviorActionNodeContext) -> NodeStatus:
        """Preempt the transfer and restage above the cube when recovery is needed."""
        robot = self._resolve_robot()
        if robot is None:
            return self._status_while_resolving()

        shared = context.get_blackboard()
        local = _get_local_blackboard(context)
        current_cube = _get_local_string(local, "current_cube")
        if local is None or not current_cube:
            return NodeStatus.FAILURE

        try:
            cube_position = RigidPrim(current_cube).get_world_poses()[0].numpy()[0]
        except (IndexError, RuntimeError, ValueError) as exc:
            carb.log_error(f"[FrankaRecoverCube] Failed to read recovery state for '{current_cube}': {exc}")
            return NodeStatus.FAILURE

        movement_tolerance = float(context.get_input("movement_tolerance"))
        recovery_key = _get_local_string(local, "recovery_key")
        recovery_requested = bool(_blackboard_get(shared, recovery_key, False)) if recovery_key else False
        if not recovery_requested:
            return NodeStatus.FAILURE

        robot_id = _get_robot_id(self.get_prim_path())
        center_owner = str(_blackboard_get(shared, "center_owner", "") or "")
        if center_owner and center_owner != robot_id:
            return NodeStatus.RUNNING

        warning_interval = float(context.get_input("warning_interval"))
        if warning_interval <= 0.0:
            carb.log_error("[FrankaRecoverCube] 'warning_interval' must be positive")
            return NodeStatus.FAILURE
        shared["center_owner"] = robot_id
        shared[f"cube_state:{current_cube}"] = f"reserved:{robot_id}"
        delta_time = context.get_delta_time()
        try:
            robot.open_gripper()
            if self._last_cube_position is None:
                self._last_cube_position = cube_position.copy()
                self._stable_elapsed = 0.0
            elif float(np.linalg.norm(cube_position - self._last_cube_position)) > movement_tolerance * 0.5:
                self._stable_elapsed = 0.0
                self._last_cube_position = cube_position.copy()
                self._motion_started = False
            else:
                self._stable_elapsed += delta_time

            _update_pick_targets(local, cube_position)
            staging_position = cube_position + np.asarray(
                [0.0, 0.0, float(context.get_input("staging_height"))], dtype=np.float32
            )
            XformPrim(str(local["approach_target"])).set_world_poses(positions=staging_position)
            orientation = _get_downward_orientation(robot, float(context.get_input("end_effector_yaw_degrees")))
            robot.move_to_pose(
                position=staging_position,
                orientation=orientation,
                delta_time=delta_time,
                reset=not self._motion_started,
            )
            self._motion_started = True
            current_position, _ = robot.get_tool_pose()
            position_error = float(np.linalg.norm(current_position[0] - staging_position))
        except (IndexError, RuntimeError, ValueError) as exc:
            carb.log_error(f"[FrankaRecoverCube] Failed to restage above '{current_cube}': {exc}")
            return NodeStatus.FAILURE

        stable_duration = float(context.get_input("stable_duration"))
        position_tolerance = float(context.get_input("position_tolerance"))
        if position_error > position_tolerance:
            self._warning_elapsed += delta_time
            if self._warning_elapsed >= warning_interval:
                carb.log_warn(
                    f"[FrankaRecoverCube] Motion to '{current_cube}' has not converged for "
                    f"'{robot_id}' after another {warning_interval:.2f} seconds; still trying"
                )
                self._warning_elapsed %= warning_interval
            return NodeStatus.RUNNING
        self._warning_elapsed = 0.0
        if self._stable_elapsed < stable_duration:
            return NodeStatus.RUNNING

        shared[recovery_key] = False
        self._last_cube_position = None
        self._stable_elapsed = 0.0
        self._warning_elapsed = 0.0
        self._motion_started = False
        return NodeStatus.FAILURE

The gripper uses physical contact, not a fixed joint, with a 90-degree wrist yaw for clearance. After pickup, FrankaCalibratePlaceTargets adjusts the place target from the measured cube pose. The center lock is released after lift, so one robot can transport while the other starts its next pickup.

The sample stage contains:

• /World/Pyramid/*: the six source cubes.

• /World/Targets/left/* and /World/Targets/right/*: target Xforms for each robot.

• /BehaviorTreeBlackboard: shared coordination state.

Run it

1. Launch the Isaac Sim full application.

2. In the Content Browser, open [Isaac Sim Assets Path]/Isaac/Samples/BehaviorTree/FrankaPickPlace/franka_pick_place.usd. [Isaac Sim Assets Path] is the path to the Isaac Sim Assets.

3. Open Window > Behavior > Behavior Tree to inspect the descriptor. Use Target Prim to switch between /World/franka_left and /World/franka_right.

4. Press Play. The robots reserve and move cubes concurrently until each stack contains three cubes. Both trees finish with SUCCESS; the timeline continues playing.

5. Reload the USD to run the demo again from its initial state.

The sample asset folder also contains franka_pick_place.json. If you copy the USD elsewhere, copy the JSON with it or update the USD’s omni:behavior:tree:descriptorFile asset path.

Declarative FSM

Scenario: a UR10 picks small KLT bins off a moving conveyor and stacks them onto a 3x3 single-layer pallet (9 positions). Bins spawn with a 50% chance of being upside-down; the FSM routes them through a flip-station jig before placing.

This demo uses the cuMotion RmpFlowController for end-effector target tracking. Each loop reads a PalletizerObservation, runs a PalletizerController, then applies its PalletizerCommand through BinStackingContext. The controller runs PalletizerFSM internally; the FSM returns PalletizerDecision values and never accesses prims, the articulation, the gripper API, or RMPflow directly.

How transitions works:

• The FSM is one Python list of {trigger, source, dest} rows.

• The library generates one method per trigger name on the model (PalletizerFSM). Calling self.pick_done() fires that transition.

• Two rows can share a trigger. A guard predicate (conditions= or unless=) picks the destination, keeping the routing in the table instead of an if/else in your step code.

5 states + 6 transitions (two pairs share a trigger).

    _STATES = ["idle", "picking", "flipping", "placing", "done"]

    _TRANSITIONS = [
        {"trigger": "go_pick", "source": "idle", "dest": "picking"},
        {"trigger": "pick_done", "source": "picking", "dest": "flipping", "conditions": "needs_flip"},
        {"trigger": "pick_done", "source": "picking", "dest": "placing", "unless": "needs_flip"},
        {"trigger": "flip_done", "source": "flipping", "dest": "picking"},
        {
            "trigger": "place_done",
            "source": "placing",
            "dest": "done",
            "conditions": "placement_completes_stack",
        },
        {
            "trigger": "place_done",
            "source": "placing",
            "dest": "idle",
            "unless": "placement_completes_stack",
        },
    ]

The two shared-trigger pairs:

• pick_done from picking goes to flipping when the guard needs_flip returns True (bin came off the conveyor upside-down), otherwise to placing.

• place_done from placing goes to done when the guard placement_completes_stack returns True, otherwise back to idle for the next bin.

model=self attaches the generated trigger methods directly onto the FSM instance. transitions then picks up the following by name from self:

• Guard predicates: needs_flip, placement_completes_stack.

Define a method with the right name and it gets called at the right time.

UR10 palletizing FSM. Edges labeled by trigger; the two pairs that share a trigger (pick_done and place_done) branch on a guard predicate.

A state’s work (moving the arm, opening the gripper, waiting) is broken into a list of small sequential substeps. PalletizerController builds and runs the list selected by PalletizerFSM. This is the same separation as the Franka examples – the FSM chooses a state, and a controller turns it into motion – with a state-local substep list in place of a shared container:

• _SubStateResult.done advances the controller to the next substep.

• List exhausted -> the controller reports completion to the FSM.

• The FSM fires the matching trigger (for example, pick_done), and Machine follows the conditional edge to the next state.

_load_sequence_for_state: controller sequences selected by the FSM.

    def _load_sequence_for_state(self, state: str, observation: PalletizerObservation) -> None:
        if state == "picking":
            pickup_wait = (
                _Wait(wait_time=1.0) if self._picking_after_flip else _Wait(wait_time=1.0, pause_conveyor_on_done=True)
            )
            lift = (
                TimedLift(height=0.3, duration=0.1)
                if observation.active_bin_needs_flip
                else _LiftToClearance(height=0.5)
            )
            self._sequence = [
                pickup_wait,
                ReachToPick(),
                _Wait(wait_time=1.0),
                CloseSuctionGripper(),
                lift,
            ]
        elif state == "flipping":
            self._sequence = [
                OrientToHome(),
                MoveToFlipStation(),
                OpenSuctionGripper(),
                _Wait(wait_time=0.25),
                ReleaseFlipStationBin(),
                OrientToHome(),
            ]
        elif state == "placing":
            self._sequence = [
                ReachToPlace(),
                _Wait(wait_time=0.5),
                OpenSuctionGripper(),
                TimedLift(height=0.1, duration=0.25),
                OrientToHome(),
            ]
        else:
            self._sequence = []
        self._sequence_index = 0
        if self._sequence:
            self._sequence[0].enter(observation)

All eight substeps are demo-local classes:

• ReachToPick / ReachToPlace – IK pose targets with approach corridors.

• MoveToFlipStation – fixed pose.

• OrientToHome – joint-space reset (see note below).

• CloseSuctionGripper / OpenSuctionGripper – gripper.

• _Wait – fixed delay.

• TimedLift – fixed-duration vertical lift.

The flipping and placing controller sequences compose the same primitives in different orders.

Run it

./python.sh standalone_examples/tutorials/state_machine/ur10_palletizing_transitions.py

Choosing your approach

Start without a separate state-machine abstraction. If a few if/elif branches stay easy to read, that is all you need. Reach for a state machine once new steps and recovery cases make one controller class hard to maintain. From there, the right structure depends on how the logic branches:

• Library-free FSM for a fixed sequence of phases with a small, fixed set of reactive checks. Use an enum for named phases and a dataclass for active-transfer data so the per-tick code stays flat.

• py_trees when several recovery cases need to preempt normal flow in a clear priority order. The tree’s left-to-right shape encodes the priority.

• Omniverse Behavior Tree when the tree should be a reusable JSON/USD asset, run with the timeline, and be inspectable in the Behavior Tree UI. Robot operations still live in registered Python node classes.

• transitions when the state graph is small but branchy. Routing decisions live in the table instead of nested if/else in your step code.

These aren’t mutually exclusive: a production system might use transitions for the high-level task graph and a behavior tree (or library-free phase code) inside each state.

py_trees and Omniverse Behavior Tree are parallel authoring choices, not drop-in API replacements for each other or for Cortex. py_trees is Python-native and application-owned. Omniverse Behavior Tree is Kit-native, descriptor-driven, and attached to USD prims.

Migrating from Cortex

The deprecated Isaac Cortex framework solved a similar problem with a custom decision-network DSL. If you’re porting Cortex code to one of the libraries above, the following mapping is approximate but covers the common cases.

Cortex concept	py_trees equivalent	transitions equivalent
DfDecider	Selector(memory=False)	The Machine itself
DfStateMachineDecider	Sequence(memory=True) of leaf Behaviour nodes	Machine state plus a controller-owned substep sequence
DfState	Behaviour subclass	Machine state or state-local substep
Logical state monitors	Condition leaf Behaviour nodes	Guard predicates (needs_flip, etc.) or per-tick checks
DfNetwork.context	Blackboard keys	self attributes (model=self)
ObstacleMonitor	Custom leaf Behaviour wrapping the toggle	State-local controller command flags

For Omniverse Behavior Tree, map DfAction and DfState to registered action nodes, logical-state monitors to condition nodes evaluated before the action sequence, DfNetwork.context to typed ports or a blackboard, and an atomic DfStateMachineDecider chain to a Sequence subtree with explicit RUNNING, SUCCESS, and FAILURE results. This is a conceptual port; there is no drop-in Cortex replacement.

Approach corridors and posture biasing

Approach corridors (Cortex approach_params) bent the EE’s path so it slid in along a chosen axis instead of cutting straight through obstacles. In the new demos:

• Franka demos: dropped; use simple scene geometry, a looser convergence threshold, and a settle buffer.

• Transitions demo: kept. MotionCommand.approach_params and shift_for_approach are inlined directly from Cortex.

Posture biasing (Cortex posture_config) softly biased redundant-DOF arms toward a preferred joint posture. None of the demos use this directly:

• Franka demos: 7-DOF, but do not set explicit posture targets; they only tune cuMotion’s c-space metric.

• Transitions demo: 6-DOF (no redundancy); uses an explicit joint_target substep (OrientToHome) when a specific posture is needed.

The original Cortex tutorials remain in the documentation as deprecated reference material:

• Isaac Cortex: Overview

• Walkthrough: Franka Block Stacking

• Walkthrough: UR10 Bin Stacking

The block-stacking and bin-stacking Cortex tutorials map most directly to Library-free FSM / Behavior tree with py_trees / Behavior tree with Omniverse Behavior Tree and Declarative FSM respectively.

Back to Gallery View