refactor(atomic-actions): preserve per-environment runtime lifecycle

agent_context/MAP.yaml

@@ -467,6 +467,10 @@ topics:
       - PlanningContext
       - ExecutionSession
       - EffectVerificationRequest
+      - EffectVerificationResult
+      - eligible_mask
+      - deactivate_rows
+      - effect verification deadline
       - ExecutionRunner
       - ObservationProvider
       - CommandSink

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

@@ -378,7 +378,7 @@ runner = ExecutionRunner(
     command_sink,
     clock=execution_clock,
 )
-result = runner.step(effect_success=None)
+result = runner.step(effect_result=None)
 ```
 
 `ExecutionSession` owns deterministic planning progress and recovery state. It
@@ -404,15 +404,43 @@ active targets so the caller can still hold them. The session monitors:
 - action-attempt timeout;
 - planner and semantic-effect failure.
 
-It replans from the latest observation within per-environment budgets. The
-budgets and eligibility masks are row-local, while the action waypoint cursor
-is batch-synchronized: one allowed replan regenerates the active cohort and
-restarts its action trajectory without charging unaffected rows. Unknown
-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. While verification is outstanding,
-`ExecutionTick.pending_effect` retains a typed `EffectVerificationRequest` on
-every tick; `EFFECT_VERIFICATION_REQUIRED` is only the one-time audit event.
+It replans from the latest observation within per-environment budgets. Pass an
+owned boolean `eligible_mask` to `engine.start()` when a previous semantic call
+has already deactivated rows. Eligibility can only shrink; use
+`runner.deactivate_rows(mask, reason=...)` while a runner owns scheduling so its
+cached effect request stays correlated. The budgets, verified task state, and
+eligibility masks are row-local, while the action waypoint cursor and call
+barrier are batch-synchronized. One allowed replan regenerates the still-pending
+cohort without charging unaffected rows. Exhausted rows hold and never become
+eligible again.
+
+A non-empty `StateDelta` is not committed until the caller supplies a
+correlated `EffectVerificationResult`. Its disjoint `success_mask` and
+`failure_mask` must be subsets of the current request mask; requested rows in
+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`
+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.
+
+```python
+request = tick.pending_effect
+effect_result = EffectVerificationResult(
+    verification_id=request.verification_id,
+    success_mask=observed_success,
+    failure_mask=observed_failure,
+)
+result = runner.step(effect_result=effect_result)
+```
+
+Cause events (`ACTION_PLANNING_FAILED`, `EFFECT_VERIFICATION_FAILED`, and
+`EFFECT_VERIFICATION_TIMEOUT`) are distinct from the `ACTION_RETRY` recovery
+event. `SESSION_COMPLETED` and `SESSION_FAILED` are distinct terminal events.
 
 Recovery replans reuse the current immutable `ResolvedActionRequest`, including
 its owned goal snapshot. Mutable goal values are copied, while simulator-backed

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

@@ -427,9 +427,11 @@ an older custom action by renaming its implementation to `_plan()`.
 | `AtomicAction.plan(request, context)` | `AtomicActionEngine` | Binds the current collision scene into a copied policy, then delegates to `_plan()` |
 | `AtomicAction._plan(request, context)` | Atomic-action implementer | Consumes the prepared immutable `ResolvedActionRequest` and returns an `ActionPlan` |
 | `engine.plan_action(action, invocation, context)` | Extension or isolated test | Temporarily binds and plans an unregistered action instance; built-in parameter variants should use invocation `skill_options` instead |
+| `engine.start(invocations, context, eligible_mask=...)` | Runtime orchestrator | Starts a session whose owned row cohort can only shrink across action barriers and recovery |
 | `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.step(effect_success=...)` | Non-blocking controller integration | Observes and routes a `RuntimeCommandFrame` only when it is due |
+| `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.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 |
 
@@ -607,6 +609,16 @@ unknown or incompatible transport cannot cause partial dispatch. Cancellation,
 observation/session exceptions, and negative acknowledgements enter a
 best-effort cancel-then-hold path for every armed runtime target.
 
+Pass an owned `eligible_mask` to `engine.start()` when only a subset of rows may
+enter the invocation sequence. This cohort is sticky: eligibility can only
+shrink across action barriers and replans. Later failures outside the atomic
+runtime should call `runner.deactivate_rows(mask, reason=...)`; the operation is
+idempotent, the next command neutralizes changed rows, and removing the final
+eligible row fails and terminates the session. When effect verification is
+pending, deactivation narrows the request and assigns a new
+`verification_id`. Do not mutate `session` directly while its runner owns
+scheduling, because the runner must refresh its cached effect boundary.
+
 The engine authorizes every emitted command against the immutable target and
 physical claims in the resolved binding. A command cannot address an unbound
 destination, substitute target metadata, or overlap another endpoint's joints
@@ -757,20 +769,39 @@ environment row. Pick, place, handover, and coordinated skills also return an
 uncommitted `StateDelta` describing the attachment state expected after
 execution.
 
-At the terminal waypoint, an `ExecutionSession` requests an external
-per-environment verification mask before committing a non-empty effect:
+At the terminal waypoint, an `ExecutionSession` requests an external,
+correlated per-environment result before committing a non-empty effect:
 
 ```python
+from embodichain.lab.sim.atomic_actions import EffectVerificationResult
+
 tick = session.tick(latest_context)
 if tick.pending_effect is not None:
-    effect_success = verify_grasp_or_release()
-    tick = session.tick(latest_context, effect_success=effect_success)
+    request = tick.pending_effect
+    success_mask, failure_mask = verify_grasp_or_release(request.env_mask)
+    effect_result = EffectVerificationResult(
+        verification_id=request.verification_id,
+        success_mask=success_mask,
+        failure_mask=failure_mask,
+    )
+    tick = session.tick(latest_context, effect_result=effect_result)
 ```
 
 This prevents a collision-free or well-tracked command plan from being
 misreported as a successful grasp, release, or handover. The typed
 `EffectVerificationRequest` persists on subsequent ticks while waiting;
-`EFFECT_VERIFICATION_REQUIRED` remains a one-time observability event.
+`EFFECT_VERIFICATION_REQUIRED` remains a one-time observability event. Success
+and failure masks are disjoint subsets of the request mask; omitted request rows
+remain unresolved. Request IDs change after mask shrinkage or whole-action
+retry, so a delayed result cannot commit a newer attempt.
+
+`request.deadline` is expressed in the robot-observation timestamp domain.
+`RecoveryPolicy.action_timeout` covers both trajectory execution and the
+terminal effect wait; a retry invalidates the old request ID. With
+`ExecutionRunner.step()`, a call made before the next due cycle does not consume
+its `effect_result`: schedule another call using `wait_duration`, re-read the
+current request, and submit a result for that current ID. Partial resolution and
+row deactivation can also replace the request before the delayed result arrives.
 
 ## Action Agent integration
 

docs/source/tutorial/atomic_actions.rst

@@ -332,7 +332,12 @@ must be resolved from the latest scene snapshot:
    )
    task = TaskState.empty(robot.get_qpos().shape[0], robot.device)
    initial_context = adapter.observe(task)
-   session = engine.start((invocation,), initial_context)
+   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()
@@ -359,6 +364,25 @@ For an application that already owns its event loop, call the non-blocking
 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:
+
+.. code-block:: python
+
+   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
@@ -451,24 +475,58 @@ Task-state effects
 
 Pick, place, handover, and coordinated skills declare attachment changes as a
 :class:`~embodichain.lab.sim.atomic_actions.StateDelta`. Planning does not commit
-those changes. During closed-loop execution, a non-empty effect requires an
-external per-environment verification mask:
+those changes. During closed-loop execution, a non-empty effect requires a
+correlated per-environment verification result:
 
 .. code-block:: python
 
+   from embodichain.lab.sim.atomic_actions import EffectVerificationResult
+
    def verify_effect(context, tick):
-       return verify_grasp_or_release(context)
+       request = tick.pending_effect
+       assert request is not None
+       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,
+       )
 
    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;
 ``run_until_blocked`` returns at the verification boundary and the application
-can later resume with ``runner.step(effect_success=verified)`` when the next
-cycle is due, or call ``run_until_blocked(effect_verifier=...)`` again. 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.
+can later resume from the *current* pending request:
+
+.. code-block:: python
+
+   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,
+   )
+   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.
 
 Adding an action
 ----------------

embodichain/lab/sim/atomic_actions/__init__.py

@@ -58,6 +58,7 @@
 )
 from .execution import (
     EffectVerificationRequest,
+    EffectVerificationResult,
     ExecutionEvent,
     ExecutionEventKind,
     ExecutionSession,
@@ -202,6 +203,7 @@
     "EndpointCommandTransport",
     "EntityState",
     "EffectVerificationRequest",
+    "EffectVerificationResult",
     "EffectVerifier",
     "ExecutionClock",
     "ExecutionFeedbackMode",