Atomic Skills

Atomic Skills#

Define a skill once and reuse it across robots and grippers by binding compatible motion and grasp resources. The skill expresses the operation and its effects; the embodiment supplies kinematics, tool geometry, TCPs and command meanings. Reuse requires a physically feasible task: reachable poses, joint limits, collision clearance, sufficient jaw opening and stable contact.

Atomic actions are typed, side-effect-free motion planners. The engine resolves a grounded ActionInvocation into a ResolvedActionRequest; an action combines that snapshot with the latest PlanningContext and returns an ActionPlan.

For the complete architecture and ownership model, see Atomic actions. For the capability matrix and visual demonstrations of every built-in skill, see Built-in atomic actions. Canonical scene identity and snapshot/provider setup are documented in Scene registry.

The contracts deliberately separate seven concerns:

a goal describes what should happen;
a SkillBindingContract declares action-local participant slots, endpoint capabilities, typed commands, and physical disjointness;
an engine-owned ActionBinding contains adapter-resolved EndpointBinding snapshots and immutable runtime targets for one call;
a ControlPartCommandProfile maps embodiment-specific meanings such as open, grasp, or ready to typed commands;
typed ActionOptions contain behavior that may vary for one skill call;
a MotionPolicy and RecoveryPolicy describe reusable planning and bounded-recovery choices;
a PlanningContext contains measured robot state, verified task state, and a versioned scene snapshot.

Slots such as primary or source name participants only within one skill. Each slot exposes skill-local endpoint protocols such as motion and grasp. There are no global arm, hand, mobile-base, or whole-body binding fields. A profile matches endpoint capabilities to generic robot resources and uses an endpoint adapter to create the runtime target.

For advanced direct-core use, joint-backed endpoint selections are concrete RobotCfg.control_parts keys, not joint, link, TCP-frame, or scene-object names. Build them through bind_control_parts():

binding = engine.bind_control_parts(
    "pick_up",
    {
        "primary": {
            "motion": "left_arm",
            "grasp": "left_hand",
        }
    },
)

The helper validates the installed skill contract, resolves joint indices and commands, and returns the engine-owned generic binding. Profile endpoint adapters may instead resolve locomotion, whole-body, or custom controller targets without changing ActionBinding.

The engine exclusively owns the MotionGenerator, shared trajectory builder, and control-part profiles. It creates and binds all built-in actions by default; callers select them by stable skill_id without a separate register() step. Put invocation-varying behavior in ActionInvocation.skill_options. register() remains available for custom implementations, and load_builtins=False creates an isolated or fully custom engine.

Trajectory timing is strict. A planner that returns positions must also return per-waypoint dt; PlanResult.duration is derived from it. A custom action must pass a complete TimedTrajectory to build_plan. The engine does not repair missing timing. Action-owned interpolation reads an explicit PlanningContext.control_dt supplied by the integration, normally BaseEnv.step_dt.

Choosing an engine entry point#

Application code normally uses one of three engine entry points:

API	Choose it when	Returns	State and observation behavior
`engine.plan()`	Planning or inspecting one action	`ActionPlan`	Reads one context and does not project a next context
`engine.compile()`	Planning a fixed sequence whose goals are known and whose plans retain joint trajectories	`CompiledTrajectory`	Propagates hypothetical qpos and expected effects, without observing execution
`engine.start()`	Executing from fresh observations with bounded recovery	`ExecutionSession`	`tick()` consumes measured context, emits commands, requests effect verification, and can replan

As a short rule: use plan for one action, compile for a static joint-trajectory sequence, and start followed by tick for observed execution and error recovery. None of these APIs steps the simulator directly. The application sends commands returned by an execution session and supplies new observations.

AtomicAction.plan(request, context) is different from engine.plan(). It is the framework-owned template method called by the engine, not an additional application execution entry point. Atomic-action authors implement the protected _plan() hook instead. Register custom action instances with engine.register() before using the same public planning entry points.

This extension contract is intentionally strict: a subclass that defines plan() raises TypeError at class definition. There is no legacy adapter; custom actions must rename that implementation to _plan() so the framework-owned collision-scene preparation cannot be bypassed.

Executable walkthroughs#

scripts/tutorials/atomic_action contains three teaching scripts: single_arm.py, dual_arm.py and embodiment_switching.py. Their robot, asset, TCP and recording helpers come from examples/sim/atomic_action. The complete runnable scene catalog, including recovery and coordinated manipulation, lives in that examples directory and is installed with the package. Browse it with:

embodichain run-example --list
embodichain run-example sim/atomic_action --help
embodichain run-example sim/atomic_action/coordinated_pickment --help

Single arm: PickUp followed by Place#

Run the single-arm walkthrough from a source checkout:

python scripts/tutorials/atomic_action/single_arm.py --headless --auto_play --device cpu

single_arm.py creates the same cube and calibrated robot as the runnable Place example. It then declares the hand’s semantic open and grasp commands and gives the engine an object-backed scene provider and grasp-pose generator. The primary participant binds to the concrete arm and eef control parts. A GraspGoal requests acquisition; a subsequent PlaceGoal supplies the desired tool-pose waypoints:

engine = create_simulation_atomic_action_engine(
    motion_generator=create_tutorial_motion_generator(robot, args.planner),
    scene_entities=(obj,),
    control_profiles={
        "eef": ControlPartCommandProfile.joint_positions(
            open=hand_open, grasp=hand_close
        )
    },
    grasp_pose_generators={
        "eef": create_parallel_jaw_grasp_pose_generator(
            n_sample=args.n_sample, force_refresh=args.force_reannotate
        )
    },
)
binding = {"primary": {"motion": "arm", "grasp": "eef"}}
policy = MotionPolicy(strategy="motion_gen", sample_count=120)

# -- Goals --
pick = engine.make_invocation(
    "pick_up",
    GraspGoal(semantics),
    control_parts=binding,
    motion_policy=policy,
    skill_options=PickUpOptions(
        pre_grasp_distance=0.15, lift_height=0.16, hand_interp_steps=12
    ),
)
place = engine.make_invocation(
    "place",
    PlaceGoal(
        broadcast_waypoint_pose_batch(
            scene.make_place_eef_poses(sim.device), robot.get_qpos().shape[0]
        )
    ),
    control_parts=binding,
    motion_policy=policy,
    skill_options=PlaceOptions(lift_height=0.10, hand_interp_steps=12),
)

The observed robot and scene state enter through an explicit control period. compile() projects the first action’s expected attachment into the next planning context, allowing Place to plan its held-object motion. It does not step physics or establish that the real grasp succeeded:

wait = prepare_tutorial_scene(sim, args, "Inspect the scene, then press Enter...")
context = engine.initial_context(control_dt=sim.sim_config.physics_dt)
compiled = engine.compile((pick, place), context)
if not bool(compiled.plan_success.all()):
    raise RuntimeError("The single-arm walkthrough could not plan both skills.")

The host subsequently passes compiled.trajectory to the shared replay_trajectory() helper. Its arrival intervals determine the simulation steps. The source script shows the complete setup, lift-boundary callback, terminal hold and run_tutorial() resource-cleanup boundary.

Two arms: one HandOver invocation#

The dual-arm walkthrough binds two motion/grasp pairs to one skill. Its default can fixture uses Robotiq 2F-140; --is_horizontal selects the stick and matching PGC-140-50 asset, TCP, grasp model and closing commands:

python scripts/tutorials/atomic_action/dual_arm.py --headless --auto_play --device cpu
python scripts/tutorials/atomic_action/dual_arm.py --is_horizontal --headless --auto_play --device cpu

dual_arm.py obtains physical setup and fixture calibration from the canonical HandOver example. It registers an open/grasp profile for each hand and supplies both grasp-pose generators to the engine. The application then constructs one goal and binds both participants:

target = torch.eye(4, dtype=torch.float32, device=sim.device)
target[:3, 3] = torch.tensor(scene.FINAL_OBJECT_XYZ, device=sim.device)
invocation = engine.make_invocation(
    "hand_over",
    HandOverGoal(semantics, target_pose=target),
    control_parts={
        "source": {"motion": "left_arm", "grasp": "left_eef"},
        "destination": {"motion": "right_arm", "grasp": "right_eef"},
    },
    motion_policy=MotionPolicy(strategy="motion_gen", sample_count=220),
    skill_options=HandOverOptions(
        pre_grasp_distance=0.08, lift_height=0.15, hand_interp_steps=10
    ),
)

source and destination are skill-local candidate participants. HandOver chooses the nearer pickup arm, plans pickup and receiving grasps, and coordinates hand closure, transfer, source release and final transport. The application does not split that transfer into unrelated single-arm calls. Both motion/hand resources must satisfy the skill’s disjointness contract.

As in the single-arm case, compile and playback remain separate. The same control_dt governs both arms. A successful compile reports planned feasibility; use measured effect verification when physical task success must be certified. The observed-execution and recovery sections below describe that alternative to offline compilation.

Reuse the same skills across robots and grippers#

embodiment_switching.py executes the same two hardware-independent recipes:

PickUp → Place: UR5 with PikaGripper or DH_AG95, Franka Panda with its native hand, and ARX5 X5A with its native hand.
HandOver → Place: CobotMagic, Tianji Marvin, and DexForce W1 v030 with PikaGripper or DH_AG95 on both sides. All four recordings use a front view.

Changing --robot or --gripper selects an embodiment configuration. The implementations of PickUp, HandOver and Place, and the tutorial’s pick_place() / handover() invocation code, stay the same. The recordings below show physical simulator playback with freely moving objects.

One PickUp → Place recipe, four embodiments.

One HandOver → Place recipe, four embodiments; front view.

Run from a source checkout:

# Default: PickUp → Place with UR5 + PikaGripper.
python scripts/tutorials/atomic_action/embodiment_switching.py
python scripts/tutorials/atomic_action/embodiment_switching.py --robot ur5 --gripper dh_ag95
python scripts/tutorials/atomic_action/embodiment_switching.py --robot franka
python scripts/tutorials/atomic_action/embodiment_switching.py --robot arx5

# HandOver defaults to W1 v030 + PikaGripper when --robot is omitted.
python scripts/tutorials/atomic_action/embodiment_switching.py --task handover --robot cobotmagic
python scripts/tutorials/atomic_action/embodiment_switching.py --task handover --robot tianji
python scripts/tutorials/atomic_action/embodiment_switching.py --task handover
python scripts/tutorials/atomic_action/embodiment_switching.py --task handover --gripper dh_ag95

Append --headless --renderer hybrid --auto_play --auto_play_video_path /tmp/demo.mp4 to record a run. Dual-arm camera poses are part of the embodiment fixture, so these commands reproduce the front view. --headless --renderer no-render runs the physical checks without recording. This walkthrough uses one environment.

What the abstraction preserves#

The application describes an object task, binds resources, compiles invocations, and replays their trajectory:

Goal: GraspGoal identifies the object. AssembleGoal describes the desired object pose relative to its destination support. Place derives each new TCP target from the established object_to_eef grasp relationship.
Binding: primary binds to the single arm and hand. HandOver binds source to the robot’s left arm/hand and destination to its right arm/hand, with arm_selection="bound".
Compile: the existing skills produce robot commands using the selected kinematics, TCPs, grasp geometry and semantic open / grasp commands.
Playback: the host dispatches commands and reads actual object and finger state to check acquisition, transfer, release and stable placement.

examples/sim/atomic_action/embodiments.py owns the replaceable physical configuration: robot assets and control parts, initial IK seeds and branch limits, task frames, gripper installation transforms, full TCPs, command units and directions, mimic relations, grasp models and drive parameters. Pika uses prismatic opening coordinates in metres; DH_AG95 uses rotational coordinates in radians and mimic followers. Both implement the same semantic commands.

Swapping either UR5 or W1’s gripper keeps that robot’s object, start, destination and task frame fixed. Different robots use reachable fixture heights while preserving the object-goal formulation and task recipe. ARX5 uses a 50 mm table, inherits the outward-elbow ready pose and enabled self-collision from embodichain_robots.ARX5Cfg, and selects an unflipped wrist branch for this fixture to keep the wrist clear of its forearm. The dual-arm beam rests on central supports that leave space beneath both grasp regions.

Franka’s ready TCP yaw aligns with the selected cube grasp, reducing the initial wrist turn and keeping placement on a compact elbow branch. CobotMagic retains its camera housings and source joint limits: a -35° task-frame yaw, lower work surface and nearer destination make same-facing equivalent grasps feasible. The two camera housings then remain separated through exchange and withdrawal. Both fixtures enable simulator self-collision. These are embodiment calibrations; the skill implementations and invocation recipes stay the same.

Check physical feasibility for every new robot/tool/scene combination. The recordings use the Default backend, trapezoidal planner and one environment. Their measured results are available as validation data. ARX5 also has a sampled collision-mesh clearance report for its planned and measured joint trajectories. The trapezoidal planner times the selected paths; collision avoidance requires a suitable path and planner. The recordings also include a Franka trajectory comparison and a CobotMagic camera/arm clearance report. The latter checks both collision meshes and visible housings at every planned command and measured playback state; these sampled checks have a finite scope. compile() projects attachments for planning, while the physical checks use fresh playback observations and return a nonzero exit code on failure.

Runnable demonstrations#

All per-skill and specialist demonstrations run from examples:

embodichain run-example sim/atomic_action --skill pickup --headless
embodichain run-example sim/atomic_action --skill place --planner toppra --headless
embodichain run-example sim/atomic_action/handover --is_horizontal --headless --auto_play
embodichain run-example sim/atomic_action/assemble --headless --auto_play
embodichain run-example sim/atomic_action/coordinated_placement --headless --auto_play
embodichain run-example sim/atomic_action/moving_target_recovery --headless --auto_play

The examples/sim/atomic_action/README.md lists fixed-scene selectors, standalone demonstrations, supported backends and recording options. The cube MoveHeldObject scene uses --skill move_held_object; the distinct paper-cup sequence is sim/atomic_action/move_held_object_cup.

sim/atomic_action/control_dt demonstrates two explicit command periods. It chooses a minimum sample count that permits the faster interpolation under its motion limits, checks that both geometric paths agree, then checks that the slower arrival intervals and duration scale with the requested period.

The motion_generator variable in the snippets below is a configured MotionGenerator; its robot, planner, device, cache, and collision world become the resources owned by the engine.

The teaching scripts and runnable examples use the shared scene setup, which creates exactly one global sun light (main_light) for the whole simulation. The sun uses a common downward direction and is not duplicated for each vectorized arena; tutorial-specific light positions are no longer needed.

Selecting a motion planner#

The teaching scripts and runnable examples accept the shared --planner switch. It selects the non-neural backend used to construct its MotionGenerator:

Value	Backend	Behavior and availability
`toppra`	`ToppraPlanner`	Time-optimal joint-space timing. Select this explicitly when comparing against the trapezoidal default.
`trapezoidal`	`TrapezoidalPlanner`	Deterministic trapezoidal joint-space timing and the default for atomic examples. Free Cartesian goals become sparse IK waypoints; explicit `is_linear` requests retain a dense Cartesian path. Required waypoints remain exact. Backend-neutral sample counts are minima, and control-grid retiming preserves native phase duration when combining arm and hand commands. Explicit `TrapezoidalPlanOptions.sample_interval` stays authoritative at the backend boundary.
`curobo`	`CuroboPlanner`	CUDA-backed Cartesian/joint planning with collision-world support when the tutorial supplies one. Select this explicitly when CUDA-backed collision-aware planning is needed.

NeuralPlanner is intentionally not a choice here: it requires a tutorial-specific ONNX model and frame configuration rather than being a drop-in backend for these examples. For example, the same pose tutorial can be run with each supported backend as follows:

Unless noted below, omitting --planner selects trapezoidal.

embodichain run-example sim/atomic_action/move_end_effector --device cpu
embodichain run-example sim/atomic_action/move_end_effector --device cpu --planner toppra
embodichain run-example sim/atomic_action/move_end_effector --device cuda --planner curobo

The selector does not override a skill’s explicit motion contract. In particular, dynamic_obstacle_recovery.py is deliberately cuRobo-only because it updates a live collision world; coordinated_pickment.py, coordinated_placement.py, and handover.py use dual-arm planning that currently rejects cuRobo. control_dt.py intentionally uses strategy="ik_interp" to compare control periods, so its planner choice is constructed for consistency but does not change that interpolation experiment. coordinated_pickment.py likewise retains its synchronized multi-arm IK/keyframe implementation; its selector validates the backend but does not replace those custom synchronized phases. The OpenDoor, Press, Slide and Twist examples select ik_interp to preserve their exact Cartesian/contact paths. Selecting a backend does not replace those constrained IK phases.

Control-part commands#

Hand qpos and named robot postures are robot knowledge rather than action configuration. Register them by concrete Robot.control_parts key when the engine is built:

from embodichain.lab.sim.atomic_actions import (
    AtomicActionEngine,
    ControlPartCommandProfile,
)

engine = AtomicActionEngine(
    motion_generator,
    control_profiles={
        "left_hand": ControlPartCommandProfile.joint_positions(
            open=left_open_qpos,
            grasp=left_grasp_qpos,
        ),
        "left_arm": ControlPartCommandProfile.joint_positions(
            ready=left_ready_qpos,
        ),
    },
)

PickUp, Place, and the other manipulation skills resolve open and grasp from their bound grasp endpoints. MoveJoints resolves a string target from primary.motion. Joint limits validate possible commands, but do not define their semantic meaning; supply calibrated robot commands in production.

Planning one action#

Use plan() when one registered action needs to be inspected, tested, or passed through application-owned orchestration:

from embodichain.lab.sim.atomic_actions import (
    ActionInvocation,
    EndEffectorPoseGoal,
    MotionPolicy,
)

binding = engine.bind_control_parts(
    "move_end_effector",
    {"primary": {"motion": "left_arm"}},
)
invocation = ActionInvocation(
    skill_id="move_end_effector",
    goal=EndEffectorPoseGoal(xpos=target_pose),
    binding=binding,
    motion_policy=MotionPolicy(sample_count=80),
)

plan = engine.plan(invocation, latest_context)
if plan.plan_success.all():
    command_frames = plan.commands.frames
    if plan.joint_trajectory is not None:
        trajectory = plan.joint_trajectory.positions
    diagnostics = plan.diagnostics
    segments = plan.segments

Named segments use half-open action-local waypoint ranges. For a compiled sequence, call compiled.segment(action_index, name) to get the corresponding range in concatenated-trajectory coordinates. This is preferable to repeating a primitive’s private sample-split formula in application or tutorial code.

The returned ActionPlan always owns a transport-neutral commands sequence. Joint-planned actions may also retain joint_trajectory for feedback, inspection, and static qpos projection. The plan describes only that invocation: expected effects are not committed, and plan does not produce a projected context for a following action. Use compile when the engine should propagate hypothetical state through a sequence.

Static compilation#

Use compile() when the scene is treated as fixed, all goals in a sequence are known during planning, and every action retains joint_trajectory:

from embodichain.lab.sim.atomic_actions import (
    ActionInvocation,
    AtomicActionEngine,
    EndEffectorPoseGoal,
    MotionPolicy,
)

engine = AtomicActionEngine(motion_generator)
binding = engine.bind_control_parts(
    "move_end_effector",
    {"primary": {"motion": "left_arm"}},
)
motion_policy = MotionPolicy(sample_count=80)

approach = ActionInvocation(
    skill_id="move_end_effector",
    goal=EndEffectorPoseGoal(xpos=approach_pose),
    binding=binding,
    motion_policy=motion_policy,
)
retreat = ActionInvocation(
    skill_id="move_end_effector",
    goal=EndEffectorPoseGoal(xpos=retreat_pose),
    binding=binding,
    motion_policy=motion_policy,
)

initial_context = engine.initial_context()
compiled = engine.compile((approach, retreat), initial_context)
if compiled.plan_success.all():
    trajectory = compiled.trajectory.positions
    approach_plan, retreat_plan = compiled.action_plans
    final_context = compiled.projected_context

compile never steps the simulator. It applies each plan’s expected StateDelta only to the returned projected_context so a following action can be planned against hypothetical state. Calling it with one invocation is valid, but plan is simpler when a projected context and sequence-shaped result are unnecessary.

This is intentionally an offline joint-trajectory projection API. It rejects a generic command plan without joint_trajectory; such plans remain valid for plan and closed-loop start/tick execution.

Do not compile across a point where later targets depend on physical execution. The coordinated-placement tutorial, for example, compiles both pick-ups, executes them, rebuilds held-object state from measured poses, and then compiles the placement stage. Use start when that observation and recovery loop should remain active throughout execution.

Dynamic goals and closed-loop execution#

Use SceneEntityPose when a goal must be resolved from the latest scene snapshot:

from embodichain.lab.sim.atomic_actions import (
    EndEffectorPoseGoal,
    RecoveryPolicy,
    SceneEntityPose,
    TrackingPolicy,
)

invocation = ActionInvocation(
    skill_id="move_end_effector",
    goal=EndEffectorPoseGoal(
        xpos=SceneEntityPose("moving_tray", relative_pose=tray_to_tcp)
    ),
    binding=engine.bind_control_parts(
        "move_end_effector",
        {"primary": {"motion": "left_arm"}},
    ),
    recovery_policy=RecoveryPolicy(
        max_replans=3,
        goal_translation_threshold=0.02,
    ),
    tracking_policy=TrackingPolicy.joint_position(
        in_flight_max_abs_error=0.05,
        terminal_max_abs_error=0.05,
    ),
)

from embodichain.lab.sim.atomic_actions import (
    EndpointCommandRouter,
    ExecutionRunner,
    SimulationExecutionAdapter,
    TaskState,
)
from embodichain.lab.task_program.semantics import SceneRegistry

registry = SceneRegistry.from_simulation(
    sim,
    rigid_objects={"moving_tray": moving_tray.uid},
)
scene_provider = registry.make_planning_scene_provider(
    motion_generator,
    batch_size=robot.num_instances,
)
adapter = SimulationExecutionAdapter(
    sim,
    robot,
    scene_provider=scene_provider,
)
task = TaskState.empty(robot.get_qpos().shape[0], robot.device)
initial_context = adapter.observe(task)
initial_eligible = determine_ready_rows(initial_context)
session = engine.start(
    (invocation,),
    initial_context,
    eligible_mask=initial_eligible,
)
router = EndpointCommandRouter((adapter,))
runner = ExecutionRunner(session, adapter, router, clock=adapter)
result = runner.run_until_blocked()

For a lightweight scene source that does not need environment correlation IDs, pass a scene_supplier(timestamp) callback instead. scene_provider and scene_supplier are mutually exclusive.

The session owns planning progress and bounded recovery. The runner owns the outer lifecycle: it requests fresh observations, schedules each RuntimeCommandFrame from its hold_duration, checks controller acknowledgements, and performs cancel-then-hold on failure. EndpointCommandRouter preflights the whole frame, groups endpoint commands by exact transport ID, and aggregates their acknowledgements. Unknown or incompatible transports are rejected before any partial dispatch. Safe stop cancels every armed runtime target, then asks its transport to hold from the latest observed context. The simulation adapter advances physics instead of sleeping in wall-clock time. ExecutionRunnerCfg contains runner-level transport and scheduling settings; it is not an atomic-action option and is not replaced by invocation revision.

For an application that already owns its event loop, call the non-blocking step() method. A step with is_waiting set has not consumed a new observation or effect result; use its wait_duration to schedule the next call.

eligible_mask is an owned initial cohort, not a one-tick filter. Eligibility can only shrink for the lifetime of the session and remains inactive across action barriers and replans. If an application later loses a row, deactivate it through the runner that owns scheduling:

changed = runner.deactivate_rows(
    lost_tracking_mask,
    reason="object tracking was lost",
)

The operation is idempotent and the next command actively neutralizes changed rows. Deactivating rows while an effect is pending narrows the request and changes its verification_id. Deactivating the last eligible row fails and terminates the session. Do not call session.deactivate_rows() directly while an ExecutionRunner owns the session because the runner must refresh its cached effect boundary.

The complete simulation example starts with a visible cube directly in front of the robot, then applies a short horizontal force pulse so physics and friction slide it sideways during one PickUp invocation whose GraspGoal.grasp_xpos is a SceneEntityPose. The session observes dynamic_goal_changed and replanned events, discards the entire stale approach/close/lift plan, and rebuilds it from the cube’s new location while the approach segment is active. After approach is dispatched, Pick stops monitoring that object dependency so contact-, close-, and lift-induced movement does not trigger a false dynamic-goal update. The replanned action closes the gripper, verifies the physical lift, and finishes while holding the cube. The original and regenerated goal axes remain visible for comparison:

embodichain run-example sim/atomic_action/moving_target_recovery --headless --auto_play --device cpu

For collision-aware execution, register each pose-updatable obstacle with SceneCollisionRole.DYNAMIC and configure the same canonical registry IDs as the planner’s dynamic obstacle names. Derive the cuRobo object mapping with registry.collision_geometry_by_id() and construct the runtime provider with registry.make_planning_scene_provider(motion_generator, batch_size=...). That one factory call checks that the registry’s complete STATIC ∪ DYNAMIC set exactly matches the planner’s complete collision world, then checks that the registry, provider, and planner dynamic subsets exactly match. It also checks planner capability and shared/per-environment world mode. One environment may infer a shared world; a multi-environment dynamic registry must choose SceneCollisionWorldMode.SHARED or PER_ENV explicitly. See Scene registry for the complete cuRobo mapping example.

The provider advances per-environment collision-world revisions when an obstacle moves; the session invalidates affected rows and the framework binds the latest poses before replanning. Pose thresholds use the last materially published pose as their baseline, so cumulative sub-threshold motion is eventually reported:

embodichain run-example sim/atomic_action/dynamic_obstacle_recovery --headless --auto_play

MotionPolicy.dynamic_collision_mode defaults to DynamicCollisionMode.AUTO. Use DynamicCollisionMode.REQUIRED when planning must fail unless the live collision world is available, or DynamicCollisionMode.OFF to ignore snapshot collision entities and collision-world revisions. These modes do not disable static-world or self-collision checks configured by the selected planner.

Recovery replans reuse one immutable invocation-revision snapshot. If an application intentionally changes the goal, options, policy, binding, or a control command while the action is active, submit a strictly newer revision:

revised = ActionInvocation(
    skill_id=invocation.skill_id,
    goal=updated_goal,
    binding=invocation.binding,
    motion_policy=invocation.motion_policy,
    recovery_policy=invocation.recovery_policy,
    skill_options=updated_options,
    control_overrides=updated_control_commands,
    invocation_id=invocation.invocation_id,
    revision=invocation.revision + 1,
)
runner.revise_current(revised)

The session replans from its latest context and emits an invocation_revised event. skill_id and invocation_id must still identify the active logical call, and the replacement must preserve the current non-empty runtime destination set and exact target address fingerprints. Use a new invocation when changing from an arm endpoint to a base, whole-body controller, or another controller. The runner keeps the current frame deadline, then observes fresh state and installs the revision at that due boundary. It rejects revision while a physical effect is awaiting verification; verify the effect first, or cancel and start a new invocation. A manually ticked session can call session.revise_current(revised, context=fresh_context) directly.

Every emitted command is authorized against the binding-owned target and physical claims. Non-empty plan frames and recovery replans keep a stable destination set. Transports must actively neutralize inactive batch rows for every addressed target; simply skipping those rows can leave a persistent controller command active.

Entities referenced through SceneEntityPose become automatic scene-motion dependencies. Object-centric skills may additionally declare an explicit ObjectSemantics.entity_id when they ground an object pose from the same scene snapshot; for example, PickUp automatically tracks that ID. The legacy ObjectSemantics.entity live-pose fallback is deprecated and does not create a scene dependency. An ActionPlan may bound dependency monitoring for every dependency with scene_dependency_end_segment or assign per-entity exclusive command-frame cutoffs with scene_dependency_monitor_until. PickUp uses the end of approach for its semantic object ID; joint tracking and collision-world revision checks are unaffected.

Task-state effects#

Pick, place, and coordinated skills declare attachment changes as a StateDelta. Planning does not commit those changes. During closed-loop execution, a non-empty effect requires a correlated per-environment verification result:

import torch

from embodichain.lab.sim.atomic_actions import EffectVerificationResult

def verify_effect(context, request):
    success_mask, failure_mask = verify_grasp_or_release(context, request.env_mask)
    return EffectVerificationResult(
        verification_id=request.verification_id,
        success_mask=success_mask,
        failure_mask=failure_mask,
        invalidation_mask=failure_mask,
        retry_mask=torch.zeros_like(failure_mask),
    )

result = runner.run_until_blocked(effect_verifier=verify_effect)

This prevents a successful trajectory plan from being mistaken for a successful physical grasp or release. The runner invokes this synchronous callback after a fresh due-cycle observation and feeds its result to the session in that same cycle. Returning all-false masks keeps the remaining rows unresolved. If verification is asynchronous, omit the callback; run_until_blocked returns at the verification boundary and the application can later resume from the current pending request:

request = runner.session.pending_effect
assert request is not None
success_mask, failure_mask = await_effect_observation(request.env_mask)
verified = EffectVerificationResult(
    verification_id=request.verification_id,
    success_mask=success_mask,
    failure_mask=failure_mask,
    invalidation_mask=failure_mask,
    retry_mask=torch.zeros_like(failure_mask),
)
resumed = runner.step(effect_result=verified)
if resumed.is_waiting:
    schedule_after(resumed.wait_duration)
    # This call did not consume ``verified``. Re-read the current request
    # and submit a result for that ID again at the due cycle.

Alternatively, call run_until_blocked(effect_verifier=...) again. Success and failure masks must be disjoint subsets of the request mask; rows in neither mask remain unresolved. A result must reuse the current request’s verification_id. Deactivation, partial resolution, or retry can replace the request, so re-read it before delayed submission and re-verify if its ID or mask changed. request.deadline uses the robot-observation timestamp domain; RecoveryPolicy.action_timeout covers both trajectory execution and the terminal effect wait. A result submitted after timeout cannot satisfy the new retry attempt because its old ID is invalid. The runner remembers the pending boundary even though the session emits its event only once. The durable state is tick.pending_effect (an EffectVerificationRequest), not the presence of that one-time event. invalidation_mask and retry_mask must both be subsets of failure_mask. Invalidation selects rows for the request’s core-owned, removal-only failure_invalidation delta; a verifier cannot publish arbitrary replacement state. Set a retry row only when replaying the same invocation remains physically valid. Other failed rows enter external recovery after selected invalidation. Unresolved evidence at the action deadline is reconciled fail-closed when covered verified state is still active.

Trajectory-segment effect gates#

An invocation may declare a PhaseEffectGateRequirement for a named, non-initial trajectory segment. The execution session then exposes a PhaseEffectGateRequest immediately before the first frame of that segment. Curated semantic calls install these automatically: Pick gates lift on destination attachment, Place gates retract on source detachment, and HandOver gates source release on destination attachment.

Supply phase_effect_gate_verifier(context, request) to runner.step() or runner.run_until_blocked(). It runs on a fresh due-cycle observation and returns a correlated PhaseEffectGateResult. If neither the success nor failure mask selects every remaining active row, the session keeps the whole cohort at the boundary and resends the command immediately before the gated segment. This preserves a close/open command and its physical preload; it is not an observed-position hold.

Gate success only permits the next command and does not update TaskState. The terminal effect verifier still owns the semantic commit. A contradictory row may consume the enclosing action’s retry budget; a row outside the result’s retry_mask requires external recovery. The gate shares the action timeout, and each consumed observation replaces its request ID. Without a gate verifier, run_until_blocked() returns the pending boundary for asynchronous handling.

Adding an action#

Define an action-owned frozen goal dataclass. Then define typed runtime options when needed, implement the protected _plan(request, context) hook, and declare the stable skill metadata. Do not override the inherited public plan() method because it binds the latest collision scene first. Legacy custom actions that implemented plan() must rename it to _plan(); defining plan() is rejected immediately.

Return scalar or per-environment planner success through build_plan. The framework normalizes the mask and holds failed rows at the observed qpos, so a new action should not reproduce that masking itself. build_plan accepts only a TimedTrajectory; an untimed position tensor raises immediately.

A minimal implementation looks like:

from dataclasses import dataclass
from typing import ClassVar

import torch

from embodichain.lab.sim.atomic_actions import (
    CARTESIAN_POSE_CAPABILITY,
    ActionOptions,
    ActionPlan,
    AtomicAction,
    JointPositionTarget,
    PlanningContext,
    ResolvedActionRequest,
    SkillBindingContract,
    SkillEndpointRequirement,
    SkillResourceSlot,
    TimedTrajectory,
)

@dataclass(frozen=True, slots=True)
class PushGoal:
    contact_pose: torch.Tensor

@dataclass(frozen=True, slots=True)
class PushOptions(ActionOptions):
    retreat_distance: float = 0.1

class Push(AtomicAction[PushGoal, PushOptions]):
    skill_id: ClassVar[str] = "push"
    GoalType: ClassVar[type] = PushGoal
    OptionsType: ClassVar[type] = PushOptions
    binding_contract: ClassVar[SkillBindingContract] = SkillBindingContract(
        slots=(
            SkillResourceSlot(
                slot_id="primary",
                endpoints=(
                    SkillEndpointRequirement(
                        endpoint_id="motion",
                        capabilities=frozenset(
                            {CARTESIAN_POSE_CAPABILITY}
                        ),
                    ),
                ),
            ),
        ),
    )

    def __init__(self, default_options: PushOptions | None = None) -> None:
        super().__init__(default_options)

    def _plan(
        self,
        request: ResolvedActionRequest[PushGoal, PushOptions],
        context: PlanningContext,
    ) -> ActionPlan:
        goal = self.require_goal(request)
        options = request.skill_options
        motion = request.binding.endpoint("primary", "motion")
        motion_target = motion.require_target(JointPositionTarget)
        # Plan from context.robot.qpos using motion_target.joint_ids and
        # produce full_robot_positions.
        # The joint helper lowers the result into RuntimeCommandFrame values
        # and retains the trajectory for joint-position feedback.
        trajectory = TimedTrajectory.from_uniform_step(
            full_robot_positions,
            env_ids=context.env_ids,
            step_dt=context.require_control_dt(),
        )
        return self.build_plan(
            request,
            context,
            success=success_mask,
            trajectory=trajectory,
        )

For a non-joint endpoint, define a typed RuntimeEndpointTarget and matching RuntimeCommandPayload, have the profile endpoint adapter produce that target, and call build_command_plan(commands=TimedCommandSequence(...)). Register the matching EndpointCommandTransport with the runner’s router. The skill contract, resource graph, binding, runner, and recovery model do not gain controller-specific fields.

Do not step simulation, mutate PlanningContext, commit StateDelta, or expose planner-specific configuration through the goal. See the in-repository add-atomic-action skill for the complete checklist.