Add dynamic scene recovery for atomic actions

.agents/skills/add-atomic-action/SKILL.md

@@ -5,9 +5,11 @@ description: Add a new simulation atomic action or motion primitive to EmbodiCha
 
 # Add Atomic Action
 
-Add an action-owned goal and a side-effect-free `AtomicAction.plan()`
-implementation. Keep task-graph/MLLM logic, simulator stepping, controller I/O,
-and physical-effect commits outside the action.
+Add an action-owned goal and a side-effect-free `AtomicAction._plan()`
+implementation. The inherited public `plan()` entry point binds the current
+collision scene before calling the skill hook. Keep task-graph/MLLM logic,
+simulator stepping, controller I/O, and physical-effect commits outside the
+action.
 
 ## Read the current contracts
 
@@ -20,10 +22,12 @@ Inspect only the files relevant to the requested skill:
 | Role-to-resource binding | `embodichain/lab/sim/atomic_actions/bindings.py` |
 | Invocation policies | `embodichain/lab/sim/atomic_actions/policies.py` |
 | Robot/task/scene state | `embodichain/lab/sim/atomic_actions/state.py` |
+| Dynamic scene provider contract | `embodichain/lab/sim/atomic_actions/scene.py` |
 | Effects and plans | `embodichain/lab/sim/atomic_actions/effects.py`, `plans.py` |
 | Trajectory helpers | `embodichain/lab/sim/atomic_actions/trajectory.py` |
 | Reference implementations | `embodichain/lab/sim/atomic_actions/primitives/` |
 | Static compiler and execution session | `engine.py`, `execution.py` |
+| Controller-facing execution ports | `runner.py`, `sim_adapter.py` |
 
 The public contract is:
 
@@ -107,7 +111,7 @@ class Push(AtomicAction[PushGoal]):
         super().__init__(motion_generator, cfg or PushCfg())
         self.builder = TrajectoryBuilder(motion_generator)
 
-    def plan(
+    def _plan(
         self,
         invocation: ActionInvocation[PushGoal],
         context: PlanningContext,
@@ -146,6 +150,9 @@ class Push(AtomicAction[PushGoal]):
 Follow these invariants:
 
 - Call `require_goal()` before planning.
+- Implement `_plan()` rather than overriding the framework-owned public
+  `plan()` method; the latter injects the latest dynamic obstacle poses into a
+  copied planner policy.
 - Plan from `context.robot.qpos`, never an implicit live robot start state.
 - Return full-robot `(B, N, robot.dof)` motion as a tensor or
   `TimedTrajectory` with matching `env_ids`.
@@ -158,6 +165,9 @@ Follow these invariants:
   applies them only after verification.
 - Set `scene_dependencies` indirectly by using `SceneEntityPose` in the goal;
   `build_plan()` records them for dynamic invalidation.
+- Do not add dynamic-obstacle arguments to a skill. A `SceneProvider` declares
+  `collision_entity_ids`; supported planners receive those entity poses through
+  the framework-owned `plan()` entry point.
 
 ## 4. Register and invoke
 
@@ -186,8 +196,10 @@ invocation = ActionInvocation(
 compiled = engine.compile((invocation,))
 ```
 
-Use `engine.start(...).tick(...)` instead when dynamic scene updates or online
-error recovery are required.
+For dynamic scene updates or online error recovery, create a session with
+`engine.start(...)`, then connect it to observation, command, and clock ports
+through `ExecutionRunner`. Use non-blocking `runner.step()` in an existing event
+loop or `runner.run_until_blocked()` in a simple application.
 
 ## 5. Export and document
 
@@ -212,6 +224,8 @@ Add pure pytest tests under `tests/sim/atomic_actions/`. Cover:
 - side-effect-free context handling;
 - masked `StateDelta` application for task effects;
 - `SceneEntityPose` replanning when the action accepts a dynamic goal;
+- collision-world revision replanning when the action uses a dynamic-world
+  planner;
 - effect verification when the action declares a non-empty delta.
 
 Run focused tests, format changed Python files with the pinned Black version,
@@ -229,4 +243,5 @@ then use the `pre-commit-check` skill before committing.
 | Return an arm-only tensor | Embed into full robot DoF. |
 | Mutate held state after planning | Declare a `StateDelta`. |
 | Treat `plan_success` as physical success | Verify effects during execution. |
-| Step the simulator from the action | Emit plans; let the caller own execution. |
+| Step the simulator from the action | Emit plans; connect execution through `ExecutionRunner`. |
+| Override public `plan()` | Implement `_plan()` so scene binding cannot be bypassed. |

agent_context/MAP.yaml

@@ -435,6 +435,14 @@ topics:
       - ActionPlan
       - PlanningContext
       - ExecutionSession
+      - ExecutionRunner
+      - ObservationProvider
+      - CommandSink
+      - SimulationExecutionAdapter
+      - SceneProvider
+      - RigidObjectSceneProvider
+      - collision world revision
+      - dynamic obstacle
       - StateDelta
       - held_objects
       - ActionBinding
@@ -455,6 +463,9 @@ topics:
       - embodichain/lab/sim/atomic_actions/state.py
       - embodichain/lab/sim/atomic_actions/plans.py
       - embodichain/lab/sim/atomic_actions/execution.py
+      - embodichain/lab/sim/atomic_actions/runner.py
+      - embodichain/lab/sim/atomic_actions/scene.py
+      - embodichain/lab/sim/atomic_actions/sim_adapter.py
       - embodichain/lab/sim/atomic_actions/engine.py
       - embodichain/lab/sim/atomic_actions/trajectory.py
       - embodichain/lab/sim/atomic_actions/primitives/

agent_context/topics/atomic-actions/atomic-actions.md

@@ -8,6 +8,11 @@ Atomic actions are side-effect-free, environment-batched planners:
 plan = action.plan(invocation: ActionInvocation, context: PlanningContext)
 ```
 
+`plan()` is the framework-owned public template method. It binds collision
+entities from the current scene into a copied planner policy, then delegates to
+the skill-specific `_plan()` hook. New actions must implement `_plan()` and
+must not override `plan()`.
+
 There is no `ActionTarget`, `WorldState`, `ActionResult`, `execute()`, or
 `AtomicActionEngine.run()` compatibility surface.
 
@@ -46,13 +51,22 @@ snapshot every time the action plans. Its entity ID is recorded in
 
 ```python
 session = engine.start(invocations, initial_context)
-tick = session.tick(latest_context, effect_success=None)
+runner = ExecutionRunner(
+    session,
+    observation_provider,
+    command_sink,
+    clock=execution_clock,
+)
+result = runner.step(effect_success=None)
 ```
 
-An `ExecutionSession` emits at most one `JointCommand` per tick and monitors:
+`ExecutionSession` owns deterministic planning progress and recovery state. It
+emits at most one `JointCommand` per tick and monitors:
 
 - joint tracking error against the previous command;
 - translation/rotation drift of referenced scene entities;
+- per-environment collision-world revision changes for collision-sensitive
+  phases;
 - phase timeout;
 - planner and semantic-effect failure.
 
@@ -61,6 +75,60 @@ or exhausted failures are reported as structured `ExecutionEvent` objects. A
 non-empty `StateDelta` is not committed until the caller supplies an external
 `effect_success` mask.
 
+`ExecutionRunner` owns the controller-facing lifecycle around a session:
+
+- `ObservationProvider.observe(task_state)` supplies a fresh, monotonically
+  timestamped `PlanningContext` when a feedback cycle is due;
+- `CommandSink.send/hold/cancel` returns a `CommandAcknowledgement` with
+  `accepted`, `rejected`, or `timed_out` status;
+- `ExecutionClock` supplies monotonic time and backend waiting;
+- non-blocking `step()` dispatches only when the current command's
+  `hold_duration` has elapsed;
+- `run_until_blocked()` is a convenience loop that waits through the clock and
+  stops at a terminal state or an unhandled effect-verification boundary; the
+  runner remembers that boundary so a later verifier call can resume it;
+- cancellation, observation/session exceptions, and negative acknowledgements
+  enter a best-effort cancel-then-hold path.
+
+`TimedTrajectory.dt[:, i]` is the interval leading to sample `i`.
+`ExecutionSession` maps this to `JointCommand.hold_duration`; the final sample
+uses its own interval as a settling window before terminal validation. Batched
+execution currently advances at a synchronized barrier using the longest active
+row interval.
+
+`SimulationExecutionAdapter` implements observation, command, and clock ports
+for a `SimulationManager`/`Robot` pair. Its `sleep()` advances an integral
+number of physics steps, so simulation execution does not depend on wall time.
+Stable context IDs are correlation identifiers; the adapter maps command rows
+to simulation robot indices rather than using those IDs as array indices.
+Real-device adapters should implement the same protocols and enforce the passed
+acknowledgement timeout in their transport/controller layer.
+
+`SceneProvider.snapshot(timestamp=..., env_ids=...)` is the scene-observation
+boundary used by execution adapters. `SceneSnapshot.collision_entity_ids`
+identifies obstacle poses consumed by a planner, while
+`collision_world_revision` is either global or per environment. A newer
+revision invalidates only affected batch rows. `RigidObjectSceneProvider`
+tracks live simulation objects, filters sub-threshold pose noise, advances the
+general scene version, and maintains per-environment collision revisions.
+
+The public `AtomicAction.plan()` copies `MotionPolicy` and forwards collision
+entity poses through `BasePlanner.with_collision_world()`. Backends opt in via
+`supports_collision_world_updates`; cuRobo implements this bridge using
+`CuroboPlanOptions.dynamic_obstacle_poses`. Thus replanning uses the same scene
+snapshot that triggered invalidation without adding obstacle parameters to each
+skill. Add/remove/geometry mutations are not yet supported by this pose-update
+path; providers should revision only pose-updatable registered obstacles.
+
+The latest validated session context is retained for safe hold if the first
+live observation fails. Environment IDs must remain stable and ordered for the
+entire session; robot and scene timestamps and scene versions must be monotonic.
+
+Runnable closed-loop examples live under `scripts/tutorials/atomic_action/`:
+`tracking_error_recovery.py`, `moving_target_recovery.py`, and
+`dynamic_obstacle_recovery.py`. Each injects one disturbance, reports the
+structured invalidation/replan events, and requires terminal completion.
+
 ## Parameter ownership
 
 Goal dataclasses carry only semantic task intent. They do not carry robot part
@@ -90,10 +158,12 @@ lift distances, and grasp constraints.
 
 1. Define a frozen action-owned goal dataclass with `goal_kind`.
 2. Declare `skill_id`, `GoalType`, and required semantic roles on the action.
-3. Validate with `require_goal(invocation)`.
-4. Plan from `context.robot.qpos`; never read an implicit live start state.
-5. Return full-robot positions or a `TimedTrajectory` through `build_plan()`.
-6. Declare symbolic changes with `StateDelta`; do not mutate context or commit
+3. Implement `_plan()`; do not override the framework-owned `plan()` method.
+4. Validate with `require_goal(invocation)`.
+5. Plan from `context.robot.qpos`; never read an implicit live start state.
+6. Return full-robot positions or a `TimedTrajectory` through `build_plan()`.
+7. Declare symbolic changes with `StateDelta`; do not mutate context or commit
    physical effects during planning.
-7. Keep scene stepping, controller I/O, and task-graph/MLLM logic outside the
-   atomic action.
+8. Keep scene stepping, controller I/O, and task-graph/MLLM logic outside the
+   atomic action. Put execution-loop I/O behind the runner protocols rather than
+   calling a simulator or device from `plan()` or `ExecutionSession`.

agent_context/topics/motion-planning/motion-planning.md

@@ -113,6 +113,15 @@ differences require `"cuboid"` or `"mesh"` representation, registration in
 data and collision caches, so retain the shared default for identical rebased
 layouts.
 
+`BasePlanner.supports_collision_world_updates` and
+`with_collision_world(options, obstacle_poses=...)` form the generic per-plan
+dynamic-world bridge. The base implementation opts out and leaves options
+unchanged. `CuroboPlanner` opts in, clones the supplied pose tensors, and merges
+them into `CuroboPlanOptions.dynamic_obstacle_poses`. Atomic actions call this
+hook from their framework-owned `plan()` template when a `SceneSnapshot`
+declares collision entities; individual skills must not construct backend
+obstacle options themselves.
+
 ### MotionGenerator
 
 Unified interface for trajectory planning with optional pre-interpolation.
@@ -198,14 +207,17 @@ Helper: `PlanResult.is_all_success() -> bool` returns `True` only when every env
 1. Create a `BasePlanner` subclass with a `plan()` method decorated with `@validate_plan_options`.
 2. Create a `BasePlannerCfg` subclass with a unique `planner_type` string.
 3. Optionally create a `PlanOptions` subclass for planner-specific options.
-4. Register in `MotionGenerator._support_planner_dict`:
+4. For a planner that accepts live obstacles, set
+   `supports_collision_world_updates = True` and implement
+   `with_collision_world()` without mutating caller-owned reusable options.
+5. Register in `MotionGenerator._support_planner_dict`:
    ```python
    _support_planner_dict = {
        "toppra": (ToppraPlanner, ToppraPlannerCfg),
        "neural": (NeuralPlanner, NeuralPlannerCfg),
    }
    ```
-5. Export from `embodichain/lab/sim/planners/__init__.py`.
+6. Export from `embodichain/lab/sim/planners/__init__.py`.
 
 ### validate_plan_options decorator
 
@@ -232,3 +244,4 @@ The decorator checks that every `PlanState` in `target_states` shares the same l
 - **Constraint tolerance** — `is_satisfied_constraint` allows 10% velocity / 25% acceleration overshoot. Dense waypoint trajectories may appear to violate constraints but pass validation.
 - **Fork safety with GPU sim** — `ToppraPlannerCfg.mp_context=None` defaults to `spawn` on GPU to avoid fork-after-CUDA-init hazards. Force `fork` only when the sim device is CPU or you have verified it is safe.
 - **cuRobo shared-world mismatch** — World-frame poses may differ solely because replicated arenas are offset. Compare poses after robot-base rebasing: keep `multi_env=False` if they match, and enable it only when robot-relative layouts differ.
+- **Dynamic obstacles silently stale** — A planner participates in atomic-action collision revision recovery only when it declares `supports_collision_world_updates=True`; its hook must bind every `collision_entity_id` pose into the current planning attempt.

docs/source/api_reference/embodichain/embodichain.lab.sim.atomic_actions.rst

@@ -31,9 +31,23 @@ embodichain.lab.sim.atomic_actions
       AtomicAction
       AtomicActionEngine
       ExecutionSession
+      ExecutionRunner
+      ExecutionRunnerCfg
+      RunnerStep
+      RunnerStatus
+      ObservationProvider
+      CommandSink
+      CommandAcknowledgement
+      CommandAckStatus
+      CommandDispatch
+      CommandOperation
+      ExecutionClock
+      SimulationExecutionAdapter
       ExecutionTick
       JointCommand
       ExecutionEvent
+      ExecutionEventKind
+      ExecutionStatus
 
    .. rubric:: Built-in goals and actions
 
@@ -120,6 +134,46 @@ Engine and execution
 .. autoclass:: ExecutionSession
    :members:
 
+.. autoclass:: ExecutionRunner
+   :members:
+
+.. autoclass:: ExecutionRunnerCfg
+   :members:
+   :exclude-members: __init__, copy, replace, to_dict
+
+.. autoclass:: ObservationProvider
+   :members:
+
+.. autoclass:: CommandSink
+   :members:
+
+.. autoclass:: ExecutionClock
+   :members:
+
+.. autoclass:: MonotonicExecutionClock
+   :members:
+
+.. autoclass:: SimulationExecutionAdapter
+   :members:
+
+.. autoclass:: CommandAcknowledgement
+   :members:
+
+.. autoclass:: CommandAckStatus
+   :members:
+
+.. autoclass:: CommandDispatch
+   :members:
+
+.. autoclass:: CommandOperation
+   :members:
+
+.. autoclass:: RunnerStep
+   :members:
+
+.. autoclass:: RunnerStatus
+   :members:
+
 .. autoclass:: ExecutionTick
    :members:
 
@@ -129,6 +183,12 @@ Engine and execution
 .. autoclass:: ExecutionEvent
    :members:
 
+.. autoclass:: ExecutionEventKind
+   :members:
+
+.. autoclass:: ExecutionStatus
+   :members:
+
 Semantic objects and helpers
 ----------------------------