refactor(atomic-actions): verify effects on due observations

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

@@ -420,14 +420,24 @@ correlated `EffectVerificationResult`. Its disjoint `success_mask` and
 neither mask remain unresolved. Partial successes commit immediately while
 unresolved rows keep the barrier pending. `EffectVerificationRequest` carries a
 monotonic `verification_id`, stable `requested_at`/`deadline` values in the
-robot-observation timestamp domain, and an owned effect snapshot. Mask shrinkage
-creates a new ID without extending the deadline; whole-action retry creates a
-new attempt. Results for an old ID are rejected. `RecoveryPolicy.action_timeout`
+robot-observation timestamp domain, a session-local `attempt_generation`, and
+an owned effect snapshot. Mask shrinkage creates a new ID without extending the
+deadline or changing the generation; installing a replacement plan increments
+the generation. Results for an old ID are rejected. `RecoveryPolicy.action_timeout`
 covers the trajectory and terminal effect wait together, and only timestamps
 strictly greater than the deadline time out. While verification is outstanding,
 `ExecutionTick.pending_effect` retains the request on every tick;
 `EFFECT_VERIFICATION_REQUIRED` is only the one-time audit event.
 
+For synchronous verification, pass `effect_verifier(context, request)` to
+`runner.step()` or `run_until_blocked()`. The runner calls it after the fresh
+due-cycle observation and supplies its result to `session.tick()` in that same
+cycle. It does not call the verifier when the observation timestamp is already
+past the request deadline. A verifier must return an exact
+`EffectVerificationResult`; all-false masks mean unresolved. External
+asynchronous integrations instead pass `effect_result` explicitly on a due
+`step()` call.
+
 ```python
 request = tick.pending_effect
 effect_result = EffectVerificationResult(

docs/source/overview/sim/atomic_actions/index.md

@@ -431,7 +431,7 @@ an older custom action by renaming its implementation to `_plan()`.
 | `session.revise_current(invocation)` | Manually ticked runtime orchestrator | Replaces the active logical call with a newer same-destination revision and replans from the latest observed context |
 | `runner.revise_current(invocation)` | Runner-driven runtime orchestrator or Action Agent | Snapshots a revision, preserves the current frame deadline, then replans from a fresh due-time observation |
 | `runner.deactivate_rows(mask, reason=...)` | Runner-driven runtime orchestrator | Permanently removes rows and refreshes the runner's cached effect request; prefer it over direct session mutation |
-| `runner.step(effect_result=...)` | Non-blocking controller integration | Observes and routes a `RuntimeCommandFrame` only when it is due |
+| `runner.step(effect_result=..., effect_verifier=...)` | Non-blocking controller integration | Observes only when due; accepts either an asynchronous correlated result or a synchronous verifier, never both |
 | `runner.run_until_blocked(...)` | Simple blocking application or tutorial | Advances the injected clock until terminal or external effect verification is required |
 | `runner.cancel(reason)` | Explicit safe stop | Requests controller cancellation followed by an observed-position hold |
 

docs/source/tutorial/atomic_actions.rst

@@ -482,9 +482,7 @@ correlated per-environment verification result:
 
    from embodichain.lab.sim.atomic_actions import EffectVerificationResult
 
-   def verify_effect(context, tick):
-       request = tick.pending_effect
-       assert request is not None
+   def verify_effect(context, request):
        success_mask, failure_mask = verify_grasp_or_release(context, request.env_mask)
        return EffectVerificationResult(
            verification_id=request.verification_id,
@@ -495,7 +493,10 @@ correlated per-environment verification result:
    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. If verification is asynchronous, omit the callback;
+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:
 

embodichain/lab/sim/atomic_actions/execution.py

@@ -108,14 +108,17 @@ class EffectVerificationRequest:
 
     ``requested_at`` and ``deadline`` use the same timestamp domain as
     :class:`RobotObservation`. Request-mask shrinkage retains both values;
-    only a whole-action retry starts a new attempt deadline.
+    only a newly installed plan starts a new attempt deadline.
+    ``attempt_generation`` is session-local and remains stable when partial
+    resolution or row deactivation replaces only the request ID.
     """
 
     verification_id: int
     skill_id: str
     invocation_id: str | None
     invocation_revision: int
     invocation_index: int
+    attempt_generation: int
     terminal_segment: str | None
     requested_at: float
     deadline: float
@@ -135,6 +138,8 @@ def __post_init__(self) -> None:
             raise ValueError("invocation_revision must be non-negative.")
         if self.invocation_index < 0:
             raise ValueError("invocation_index must be non-negative.")
+        if type(self.attempt_generation) is not int or self.attempt_generation < 0:
+            raise ValueError("attempt_generation must be a non-negative integer.")
         if self.terminal_segment is not None and (
             not isinstance(self.terminal_segment, str) or not self.terminal_segment
         ):
@@ -166,6 +171,7 @@ def snapshot(self) -> EffectVerificationRequest:
             invocation_id=self.invocation_id,
             invocation_revision=self.invocation_revision,
             invocation_index=self.invocation_index,
+            attempt_generation=self.attempt_generation,
             terminal_segment=self.terminal_segment,
             requested_at=self.requested_at,
             deadline=self.deadline,
@@ -304,6 +310,7 @@ def __init__(
         ] = {}
         self._planned_scene = context.scene
         self._action_started_at = context.robot.timestamp
+        self._attempt_generation = -1
         self._last_joint_command: torch.Tensor | None = None
         self._last_joint_ids: tuple[int, ...] = ()
         self._last_command_mask = torch.zeros(
@@ -887,6 +894,7 @@ def _install_plan(
         ):
             self._active_targets = replacement_targets
         self._plan = plan
+        self._attempt_generation += 1
         self._waypoint_index = 0
         self._planned_scene = context.scene
         self._action_started_at = context.robot.timestamp
@@ -1468,6 +1476,7 @@ def _effect_verification_request(
             invocation_id=request.invocation_id,
             invocation_revision=request.revision,
             invocation_index=self._invocation_index,
+            attempt_generation=self._attempt_generation,
             terminal_segment=(
                 self._plan.segments[-1].name if self._plan.segments else None
             ),

embodichain/lab/sim/atomic_actions/runner.py

@@ -31,6 +31,7 @@
 
 from .bindings import RuntimeEndpointTarget
 from .execution import (
+    EffectVerificationRequest,
     EffectVerificationResult,
     ExecutionSession,
     ExecutionStatus,
@@ -293,10 +294,10 @@ def is_waiting(self) -> bool:
 
 
 EffectVerifier = Callable[
-    [PlanningContext, ExecutionTick],
-    EffectVerificationResult | None,
+    [PlanningContext, EffectVerificationRequest],
+    EffectVerificationResult,
 ]
-"""Callback that verifies a pending semantic effect for each environment."""
+"""Synchronous verifier called on a fresh due-cycle observation."""
 
 RunnerStepCallback = Callable[[RunnerStep], None]
 """Optional observer called after every blocking runner-loop iteration."""
@@ -464,18 +465,30 @@ def step(
         self,
         *,
         effect_result: EffectVerificationResult | None = None,
+        effect_verifier: EffectVerifier | None = None,
     ) -> RunnerStep:
         """Perform one due observation/session/controller update without sleeping.
 
         Args:
             effect_result: Optional correlated effect result. If this call
                 occurs before the next cycle is due, it is not consumed and
                 must be supplied again on a later call.
+            effect_verifier: Optional synchronous verifier for the current
+                pending request. It runs after a fresh due-cycle observation
+                and before the session consumes the result. It is not called
+                after the request deadline. Mutually exclusive with
+                ``effect_result``.
 
         Returns:
             Runner status, optional session tick, controller acknowledgements,
             and time remaining before another update is due.
         """
+        if effect_result is not None and effect_verifier is not None:
+            raise ValueError(
+                "effect_result and effect_verifier are mutually exclusive."
+            )
+        if effect_verifier is not None and not callable(effect_verifier):
+            raise TypeError("effect_verifier must be callable or None.")
         now = self._clock_now()
         if self._status is not RunnerStatus.RUNNING:
             return self._result(timestamp=now)
@@ -499,6 +512,25 @@ def step(
             )
         self._last_context = context
 
+        pending_effect = self._session.pending_effect
+        if (
+            effect_verifier is not None
+            and pending_effect is not None
+            and context.robot.timestamp <= pending_effect.deadline
+        ):
+            try:
+                effect_result = effect_verifier(context, pending_effect)
+                if type(effect_result) is not EffectVerificationResult:
+                    raise TypeError(
+                        "EffectVerifier must return exactly "
+                        "EffectVerificationResult."
+                    )
+            except Exception as exc:
+                return self._fail(
+                    f"Effect verifier failed: {type(exc).__name__}: {exc}",
+                    context=context,
+                )
+
         try:
             if self._pending_revision is not None:
                 self._session._install_prepared_revision(
@@ -659,9 +691,10 @@ def run_until_blocked(
         """Run with clock-driven waiting until terminal or effect verification blocks.
 
         Args:
-            effect_verifier: Optional callback used after an
-                ``effect_verification_required`` event. Without one, the method
-                returns the running step so the caller can verify externally.
+            effect_verifier: Optional synchronous callback used on fresh
+                due-cycle observations while effect verification is pending.
+                Without one, the method returns the running boundary so the
+                caller can verify externally.
             on_step: Optional callback for tracing or tutorial visualization.
             max_steps: Hard bound on loop iterations.
 
@@ -670,7 +703,6 @@ def run_until_blocked(
         """
         if max_steps <= 0:
             raise ValueError("max_steps must be greater than zero.")
-        effect_result: EffectVerificationResult | None = None
         now = self._clock_now()
         last_result = self._result(
             timestamp=now,
@@ -681,9 +713,7 @@ def run_until_blocked(
         if self.effect_verification_pending and effect_verifier is None:
             return last_result
         for _ in range(max_steps):
-            result = self.step(effect_result=effect_result)
-            if result.tick is not None:
-                effect_result = None
+            result = self.step(effect_verifier=effect_verifier)
             if on_step is not None:
                 try:
                     on_step(result)
@@ -700,20 +730,8 @@ def run_until_blocked(
             verification_required = (
                 result.tick is not None and result.tick.pending_effect is not None
             )
-            if verification_required:
-                if effect_verifier is None or result.context is None:
-                    return result
-                try:
-                    effect_result = effect_verifier(result.context, result.tick)
-                except Exception as exc:
-                    return self._fail(
-                        f"Effect verifier failed: {type(exc).__name__}: {exc}",
-                        context=result.context,
-                        tick=result.tick,
-                        dispatches=list(result.dispatches),
-                    )
-                if effect_result is None:
-                    return result
+            if verification_required and effect_verifier is None:
+                return result
             if result.wait_duration > 0.0:
                 try:
                     self._clock.sleep(result.wait_duration)

scripts/tutorials/atomic_action/moving_target_recovery.py

@@ -36,10 +36,11 @@
     AtomicActionEngine,
     ControlPartCommandProfile,
     EntityState,
+    EffectVerificationRequest,
+    EffectVerificationResult,
     ExecutionEventKind,
     ExecutionRunner,
     ExecutionRunnerCfg,
-    ExecutionTick,
     GraspGoal,
     MotionPolicy,
     ObjectSemantics,
@@ -406,8 +407,8 @@ def on_step(step: RunnerStep) -> None:
 
     def verify_pickup_effect(
         _context: PlanningContext,
-        _: ExecutionTick,
-    ) -> torch.Tensor:
+        request: EffectVerificationRequest,
+    ) -> EffectVerificationResult:
         """Verify that the cube rose with, and remains near, the end effector."""
         cube_position = target.get_local_pose(to_matrix=True)[:, :3, 3]
         eef_position = robot.compute_fk(
@@ -426,7 +427,12 @@ def verify_pickup_effect(
             f"cube-to-EEF={held_distance.detach().cpu().tolist()} m, "
             f"success={success.detach().cpu().tolist()}."
         )
-        return success
+        verified_success = request.env_mask & success
+        return EffectVerificationResult(
+            verification_id=request.verification_id,
+            success_mask=verified_success,
+            failure_mask=request.env_mask & ~success,
+        )
 
     recording_started = start_auto_play_recording(
         sim,

tests/sim/atomic_actions/test_engine_per_env.py

@@ -1406,6 +1406,7 @@ def test_partial_effect_success_commits_resolved_rows_and_shrinks_request() -> N
     assert partial.pending_effect is not None
     assert partial.pending_effect.env_mask.tolist() == [False, True]
     assert partial.pending_effect.verification_id != first_request.verification_id
+    assert partial.pending_effect.attempt_generation == first_request.attempt_generation
     assert partial.pending_effect.requested_at == first_request.requested_at
     assert partial.pending_effect.deadline == first_request.deadline
     assert not any(
@@ -1929,6 +1930,7 @@ def test_effect_retry_invalidates_previous_verification_id() -> None:
     assert first_wait.pending_effect is not None
     old_id = first_wait.pending_effect.verification_id
     old_deadline = first_wait.pending_effect.deadline
+    old_generation = first_wait.pending_effect.attempt_generation
 
     retry = session.tick(_context(0.3, 0.2, 0.2, 0))
     assert retry.command is not None
@@ -1937,6 +1939,7 @@ def test_effect_retry_invalidates_previous_verification_id() -> None:
     second_wait = session.tick(_context(0.5, 0.2, 0.2, 0))
     assert second_wait.pending_effect is not None
     assert second_wait.pending_effect.verification_id != old_id
+    assert second_wait.pending_effect.attempt_generation == old_generation + 1
     assert second_wait.pending_effect.deadline > old_deadline
 
     with pytest.raises(ValueError, match="verification_id"):
@@ -1950,6 +1953,30 @@ def test_effect_retry_invalidates_previous_verification_id() -> None:
         )
 
 
+def test_effect_request_generation_advances_after_tracking_replan() -> None:
+    engine, _ = _engine()
+    effect = EffectAction()
+    engine.register(effect)
+    base = _invocation(engine)
+    invocation = ActionInvocation(
+        skill_id=effect.skill_id,
+        goal=base.goal,
+        binding=base.binding,
+        motion_policy=base.motion_policy,
+        recovery_policy=base.recovery_policy,
+    )
+    session = engine.start((invocation,), _context(0.0, 0.0, 0.2, 0))
+    session.tick(_context(0.0, 0.0, 0.2, 0))
+
+    replanned = session.tick(_context(0.1, 1.0, 0.2, 0))
+    session.tick(_context(0.2, 1.0, 0.2, 0))
+    waiting = session.tick(_context(0.3, 0.2, 0.2, 0))
+
+    assert any(event.kind is ExecutionEventKind.REPLANNED for event in replanned.events)
+    assert waiting.pending_effect is not None
+    assert waiting.pending_effect.attempt_generation == 1
+
+
 def test_failed_effect_plan_retries_without_requesting_effect_verification() -> None:
     engine, _ = _engine()
     engine.register(FailedEffectAction())