From 1317f99a8db208b925c5a4fd0d371a2c495d8200 Mon Sep 17 00:00:00 2001 From: yuecideng Date: Tue, 11 Aug 2026 03:50:13 +0800 Subject: [PATCH] refactor(atomic-actions): verify effects on due observations --- .../topics/atomic-actions/atomic-actions.md | 16 +- .../overview/sim/atomic_actions/index.md | 2 +- docs/source/tutorial/atomic_actions.rst | 9 +- .../lab/sim/atomic_actions/execution.py | 11 +- embodichain/lab/sim/atomic_actions/runner.py | 66 +++--- .../atomic_action/moving_target_recovery.py | 14 +- .../sim/atomic_actions/test_engine_per_env.py | 27 +++ tests/sim/atomic_actions/test_runner.py | 199 ++++++++++++++++-- 8 files changed, 285 insertions(+), 59 deletions(-) diff --git a/agent_context/topics/atomic-actions/atomic-actions.md b/agent_context/topics/atomic-actions/atomic-actions.md index de796f9fb..dfcd1dbb1 100644 --- a/agent_context/topics/atomic-actions/atomic-actions.md +++ b/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( diff --git a/docs/source/overview/sim/atomic_actions/index.md b/docs/source/overview/sim/atomic_actions/index.md index 56afd9a57..e033a7937 100644 --- a/docs/source/overview/sim/atomic_actions/index.md +++ b/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 | diff --git a/docs/source/tutorial/atomic_actions.rst b/docs/source/tutorial/atomic_actions.rst index 181472035..b451fe9ce 100644 --- a/docs/source/tutorial/atomic_actions.rst +++ b/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: diff --git a/embodichain/lab/sim/atomic_actions/execution.py b/embodichain/lab/sim/atomic_actions/execution.py index d44f24505..ba21c669c 100644 --- a/embodichain/lab/sim/atomic_actions/execution.py +++ b/embodichain/lab/sim/atomic_actions/execution.py @@ -108,7 +108,9 @@ 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 @@ -116,6 +118,7 @@ class EffectVerificationRequest: 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 ), diff --git a/embodichain/lab/sim/atomic_actions/runner.py b/embodichain/lab/sim/atomic_actions/runner.py index 65043e374..8dac661a6 100644 --- a/embodichain/lab/sim/atomic_actions/runner.py +++ b/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,6 +465,7 @@ def step( self, *, effect_result: EffectVerificationResult | None = None, + effect_verifier: EffectVerifier | None = None, ) -> RunnerStep: """Perform one due observation/session/controller update without sleeping. @@ -471,11 +473,22 @@ def step( 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) diff --git a/scripts/tutorials/atomic_action/moving_target_recovery.py b/scripts/tutorials/atomic_action/moving_target_recovery.py index db9da7b13..4c76f5968 100644 --- a/scripts/tutorials/atomic_action/moving_target_recovery.py +++ b/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, diff --git a/tests/sim/atomic_actions/test_engine_per_env.py b/tests/sim/atomic_actions/test_engine_per_env.py index 141813b98..eb2dcf6cb 100644 --- a/tests/sim/atomic_actions/test_engine_per_env.py +++ b/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()) diff --git a/tests/sim/atomic_actions/test_runner.py b/tests/sim/atomic_actions/test_runner.py index ed9432387..58dda7380 100644 --- a/tests/sim/atomic_actions/test_runner.py +++ b/tests/sim/atomic_actions/test_runner.py @@ -37,11 +37,11 @@ CommandAckStatus, CommandOperation, EndEffectorPoseGoal, + EffectVerificationRequest, EffectVerificationResult, ExecutionEventKind, ExecutionRunner, ExecutionRunnerCfg, - ExecutionTick, HeldObjectState, JOINT_POSITION_CAPABILITY, JointPositionPayload, @@ -323,13 +323,11 @@ def _make_runner( def _successful_effect_result( context: PlanningContext, - tick: ExecutionTick, + request: EffectVerificationRequest, ) -> EffectVerificationResult: """Correlate a successful result with the pending effect boundary.""" - pending_effect = tick.pending_effect - assert pending_effect is not None return EffectVerificationResult( - verification_id=pending_effect.verification_id, + verification_id=request.verification_id, success_mask=torch.ones( context.batch_size, dtype=torch.bool, @@ -683,10 +681,10 @@ def test_resumed_effect_verifier_uses_a_fresh_observation() -> None: def record_fresh_context( context: PlanningContext, - tick: ExecutionTick, + request: EffectVerificationRequest, ) -> EffectVerificationResult: observed_at.append(context.robot.timestamp) - return _successful_effect_result(context, tick) + return _successful_effect_result(context, request) completed = runner.run_until_blocked(effect_verifier=record_fresh_context) @@ -695,32 +693,191 @@ def record_fresh_context( assert observed_at[0] > blocked_at +def test_due_effect_verifier_consumes_fresh_observation_in_the_same_step() -> None: + runner, clock, _, _, _ = _make_runner(with_effect=True) + blocked = runner.run_until_blocked() + assert blocked.context is not None + assert blocked.tick is not None and blocked.tick.pending_effect is not None + blocked_at = blocked.context.robot.timestamp + clock.advance(MINIMUM_CYCLE_TIME) + observed_at: list[float] = [] + + def verify_fresh_observation( + context: PlanningContext, + request: EffectVerificationRequest, + ) -> EffectVerificationResult: + observed_at.append(context.robot.timestamp) + return _successful_effect_result(context, request) + + completed = runner.step(effect_verifier=verify_fresh_observation) + + assert completed.status is RunnerStatus.COMPLETED + assert completed.tick is not None and completed.tick.pending_effect is None + assert completed.tick.task_state.get_held_object("arm") is not None + assert completed.context is not None + assert observed_at == [completed.context.robot.timestamp] + assert observed_at[0] > blocked_at + + +def test_effect_verifier_runs_and_succeeds_at_the_request_deadline() -> None: + runner, clock, _, _, _ = _make_runner( + with_effect=True, + action_timeout=2.0, + ) + blocked = runner.run_until_blocked() + assert blocked.tick is not None and blocked.tick.pending_effect is not None + request = blocked.tick.pending_effect + clock.advance(request.deadline - clock.now()) + observed_at: list[float] = [] + + def verify_at_deadline( + context: PlanningContext, + current_request: EffectVerificationRequest, + ) -> EffectVerificationResult: + observed_at.append(context.robot.timestamp) + return _successful_effect_result(context, current_request) + + completed = runner.step(effect_verifier=verify_at_deadline) + + assert completed.status is RunnerStatus.COMPLETED + assert observed_at == pytest.approx([request.deadline]) + + +def test_effect_verifier_is_not_called_after_deadline_and_session_retries() -> None: + runner, clock, _, _, action = _make_runner( + with_effect=True, + max_action_retries=1, + action_timeout=2.0, + ) + blocked = runner.run_until_blocked() + assert blocked.tick is not None and blocked.tick.pending_effect is not None + request = blocked.tick.pending_effect + plan_count = action.plan_count + clock.advance(request.deadline - clock.now() + MINIMUM_CYCLE_TIME) + verifier = Mock() + + retry = runner.step(effect_verifier=verifier) + + verifier.assert_not_called() + assert retry.status is RunnerStatus.RUNNING + assert retry.tick is not None and retry.tick.command is not None + assert action.plan_count == plan_count + 1 + assert { + ExecutionEventKind.EFFECT_VERIFICATION_TIMEOUT, + ExecutionEventKind.ACTION_RETRY, + ExecutionEventKind.REPLANNED, + }.issubset({event.kind for event in retry.tick.events}) + + +def test_effect_result_and_effect_verifier_are_mutually_exclusive() -> None: + runner, _, _, sink, action = _make_runner(with_effect=True) + result = EffectVerificationResult( + verification_id=0, + success_mask=torch.tensor([True]), + failure_mask=torch.tensor([False]), + ) + + with pytest.raises(ValueError, match="mutually exclusive"): + runner.step( + effect_result=result, + effect_verifier=_successful_effect_result, + ) + + assert action.plan_count == 1 + assert sink.sent == [] + + +@pytest.mark.parametrize( + "invalid_result", + [None, True], + ids=["none", "wrong-type"], +) +def test_effect_verifier_invalid_result_fails_with_cancel_then_hold( + invalid_result: object | None, +) -> None: + runner, clock, _, sink, _ = _make_runner(with_effect=True) + blocked = runner.run_until_blocked() + assert blocked.tick is not None and blocked.tick.pending_effect is not None + clock.advance(MINIMUM_CYCLE_TIME) + + def invalid_verifier( + context: PlanningContext, + request: EffectVerificationRequest, + ) -> object | None: + del context, request + return invalid_result + + failed = runner.step(effect_verifier=invalid_verifier) + + assert failed.status is RunnerStatus.FAILED + assert [item.operation for item in failed.dispatches] == [ + CommandOperation.CANCEL, + CommandOperation.HOLD, + ] + assert sink.cancel_count == 1 + assert failed.message is not None + assert "must return exactly EffectVerificationResult" in failed.message + + +def test_all_false_effect_updates_keep_polling_the_same_request() -> None: + runner, clock, _, sink, _ = _make_runner(with_effect=True) + blocked = runner.run_until_blocked() + assert blocked.tick is not None and blocked.tick.pending_effect is not None + initial_request = blocked.tick.pending_effect + observed_requests: list[tuple[int, int]] = [] + + def report_no_progress( + context: PlanningContext, + request: EffectVerificationRequest, + ) -> EffectVerificationResult: + observed_requests.append((request.verification_id, request.attempt_generation)) + return EffectVerificationResult( + verification_id=request.verification_id, + success_mask=torch.zeros(context.batch_size, dtype=torch.bool), + failure_mask=torch.zeros(context.batch_size, dtype=torch.bool), + ) + + clock.advance(MINIMUM_CYCLE_TIME) + first_poll = runner.step(effect_verifier=report_no_progress) + clock.advance(MINIMUM_CYCLE_TIME) + second_poll = runner.step(effect_verifier=report_no_progress) + + assert first_poll.status is RunnerStatus.RUNNING + assert second_poll.status is RunnerStatus.RUNNING + assert first_poll.tick is not None and first_poll.tick.pending_effect is not None + assert second_poll.tick is not None and second_poll.tick.pending_effect is not None + assert observed_requests == [ + (initial_request.verification_id, initial_request.attempt_generation), + (initial_request.verification_id, initial_request.attempt_generation), + ] + assert sink.cancel_count == 0 + assert second_poll.tick.task_state.get_held_object("arm") is None + + def test_partial_effect_verifier_receives_the_committed_task_state() -> None: runner, _, _, _, _ = _make_runner(with_effect=True, batch_size=2) observations: list[list[bool] | None] = [] def verify_in_two_updates( context: PlanningContext, - tick: ExecutionTick, + request: EffectVerificationRequest, ) -> EffectVerificationResult: - pending_effect = tick.pending_effect - assert pending_effect is not None held = context.task.get_held_object("arm") observations.append( None if held is None or held.env_mask is None else held.env_mask.tolist() ) - if pending_effect.env_mask.tolist() == [True, True]: + if request.env_mask.tolist() == [True, True]: return EffectVerificationResult( - verification_id=pending_effect.verification_id, + verification_id=request.verification_id, success_mask=torch.tensor([True, False]), failure_mask=torch.tensor([False, False]), ) - assert pending_effect.env_mask.tolist() == [False, True] + assert request.env_mask.tolist() == [False, True] assert held is not None and held.env_mask is not None assert held.env_mask.tolist() == [True, False] assert torch.equal(context.task.held_objects["arm"].env_mask, held.env_mask) return EffectVerificationResult( - verification_id=pending_effect.verification_id, + verification_id=request.verification_id, success_mask=torch.tensor([False, True]), failure_mask=torch.tensor([False, False]), ) @@ -786,6 +943,7 @@ def test_runner_deactivation_refreshes_cached_effect_request() -> None: blocked = runner.run_until_blocked() assert blocked.tick is not None and blocked.tick.pending_effect is not None old_id = blocked.tick.pending_effect.verification_id + old_generation = blocked.tick.pending_effect.attempt_generation changed = runner.deactivate_rows( torch.tensor([False, True]), @@ -797,15 +955,14 @@ def test_runner_deactivation_refreshes_cached_effect_request() -> None: assert refreshed.tick is not None and refreshed.tick.pending_effect is not None assert refreshed.tick.pending_effect.env_mask.tolist() == [True, False] assert refreshed.tick.pending_effect.verification_id != old_id + assert refreshed.tick.pending_effect.attempt_generation == old_generation def verify_remaining( context: PlanningContext, - tick: ExecutionTick, + request: EffectVerificationRequest, ) -> EffectVerificationResult: - pending_effect = tick.pending_effect - assert pending_effect is not None return EffectVerificationResult( - verification_id=pending_effect.verification_id, + verification_id=request.verification_id, success_mask=torch.tensor([True, False]), failure_mask=torch.tensor([False, False]), ) @@ -822,12 +979,10 @@ def test_blocking_runner_fails_safely_for_a_mismatched_effect_result() -> None: def mismatched_effect_result( context: PlanningContext, - tick: ExecutionTick, + request: EffectVerificationRequest, ) -> EffectVerificationResult: - pending_effect = tick.pending_effect - assert pending_effect is not None return EffectVerificationResult( - verification_id=pending_effect.verification_id + 1, + verification_id=request.verification_id + 1, success_mask=torch.ones(context.batch_size, dtype=torch.bool), failure_mask=torch.zeros(context.batch_size, dtype=torch.bool), )