diff --git a/agent_context/topics/atomic-actions/atomic-actions.md b/agent_context/topics/atomic-actions/atomic-actions.md index 70d1fb642..e9adf2285 100644 --- a/agent_context/topics/atomic-actions/atomic-actions.md +++ b/agent_context/topics/atomic-actions/atomic-actions.md @@ -442,15 +442,27 @@ asynchronous integrations instead pass `effect_result` explicitly on a due `step()` call. ```python +import torch + request = tick.pending_effect effect_result = EffectVerificationResult( verification_id=request.verification_id, success_mask=observed_success, failure_mask=observed_failure, + invalidation_mask=observed_failure, + retry_mask=torch.zeros_like(observed_failure), ) result = runner.step(effect_result=effect_result) ``` +Both failure-policy masks must be subsets of `failure_mask`. +`invalidation_mask` selects rows on which the core applies the request-owned, +removal-only `failure_invalidation` delta; it does not let the verifier inject +state. `retry_mask` is reserved for rows whose physical preconditions still +make replay of the same invocation valid. Other failed rows require external +recovery. Unresolved evidence at the action deadline is reconciled fail-closed +when the pending effect covers active verified state. + The semantic layer keeps physical observation separate from symbolic effect commit. `SkillPolicyPreset.effect_monitors` maps exact semantic call IDs to versioned, bounded-declarative `EffectMonitorRef` values. Omitting the mapping diff --git a/docs/design/declarative_expert_program_plan.md b/docs/design/declarative_expert_program_plan.md index 2cfb53d98..13f4cd66d 100644 --- a/docs/design/declarative_expert_program_plan.md +++ b/docs/design/declarative_expert_program_plan.md @@ -603,16 +603,22 @@ the affected row's assumed relation, and enter bounded recovery instead of repairing the scene. The runtime now exposes the active named motion phase, observes phase-scoped held-object invariants from fresh physical evidence, and applies removal-only ``StateDelta`` reconciliation to failed rows before any -retry or recovery hand-off. ``Pick`` can use the existing bounded action retry; -``Place`` and ``HandOver`` currently emit a typed ``RECOVERY_REQUIRED`` boundary -because replaying the same invocation after its required relation was removed -would be invalid. A workflow-level re-acquisition policy, blocking acquisition -gates, per-expectation terminal failure reconciliation, and fail-closed -reconciliation for evidence that remains unresolved at the action deadline -remain explicit design decisions rather than implicit scene repair. For -handover, success transfers the verified relation from source to destination -while the destination remains physically closed. Releasing the destination is -a separate ``Place`` or ``Release`` semantic call. +retry or recovery hand-off. The monitor publishes one current-observation +outcome per physical expectation, including the stronger proof that every +clause reached its inverse band. ``Pick`` can use the existing bounded action +retry. ``Place`` retries only when that complete inverse proof shows the source +is still attached; otherwise it invalidates the relation and emits a typed +``RECOVERY_REQUIRED`` boundary. ``HandOver`` always hands terminal failure to +workflow recovery, retaining the source relation only when complete inverse +evidence proves it is still attached. A verifier selects row-local retry versus +external recovery, but the core owns the removal-only invalidation delta and +applies it before either path. Evidence that remains unresolved at the action +deadline is reconciled fail-closed: any active verified state covered by the +pending effect is removed before external recovery. Workflow-level +re-acquisition and blocking acquisition gates remain open; neither may repair +the scene implicitly. For handover, success transfers the verified relation +from source to destination while the destination remains physically closed. +Releasing the destination is a separate ``Place`` or ``Release`` semantic call. The first pure-dynamics rollout uses the staged **B** continuation policy. The standard simulation factory lowers both trajectory ``control_dt`` and runner @@ -1226,9 +1232,11 @@ implemented. Physical simulation acceptance is partial: Open Drawer and one cube Pick/Place/settle/validator cycle have completed. The embodiment-owned dual-UR5/PGI HandOver slice now completes Pick, transfer, terminal physical-effect verification, settling, and target validation through real -contact dynamics; blocking acquisition gates, workflow-level re-acquisition, -per-expectation terminal reconciliation, fault-injection coverage, and the full -repeated-cube run remain validation or design work. +contact dynamics. Per-expectation terminal outcomes, core-owned failure +invalidation, row-local retry/recovery decisions, and fail-closed deadline +reconciliation are implemented. Blocking acquisition gates, workflow-level +re-acquisition, fault-injection coverage, and the full repeated-cube run remain +validation or design work. Deliverables: @@ -1485,10 +1493,11 @@ The design is complete when all of the following hold: - [ ] Physical held-object loss is observed as effect failure, invalidates the affected symbolic relation, and exercises bounded recovery rather than being hidden by a simulator-side attachment. The phase-aware observation, - row-local invalidation, bounded Pick retry, and typed recovery boundary - are implemented; blocking acquisition, per-expectation terminal - reconciliation, workflow-level re-acquisition, and real-simulation fault - injection remain open. + row-local core-owned invalidation, per-expectation terminal + reconciliation, fail-closed deadline handling, bounded Pick/retained-Place + retry, and typed recovery boundary are implemented; blocking acquisition, + workflow-level re-acquisition, and real-simulation fault injection remain + open. - [x] Repeated sub-threshold motion eventually publishes the correct scene revision. - [x] Custom actions have a documented and tested intentional hard-break diff --git a/docs/source/overview/sim/atomic_actions/index.md b/docs/source/overview/sim/atomic_actions/index.md index d1e5fe56d..c59bcbabf 100644 --- a/docs/source/overview/sim/atomic_actions/index.md +++ b/docs/source/overview/sim/atomic_actions/index.md @@ -775,6 +775,8 @@ At the terminal waypoint, an `ExecutionSession` requests an external, correlated per-environment result before committing a non-empty effect: ```python +import torch + from embodichain.lab.sim.atomic_actions import EffectVerificationResult tick = session.tick(latest_context) @@ -785,6 +787,8 @@ if tick.pending_effect is not None: verification_id=request.verification_id, success_mask=success_mask, failure_mask=failure_mask, + invalidation_mask=failure_mask, + retry_mask=torch.zeros_like(failure_mask), ) tick = session.tick(latest_context, effect_result=effect_result) ``` @@ -797,6 +801,14 @@ 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. +Every result also classifies failed rows with `invalidation_mask` and +`retry_mask`, both subsets of `failure_mask`. Invalidation applies the +request's core-owned removal-only `failure_invalidation`; the verifier cannot +inject replacement state. Retry is valid only when the same invocation's +physical preconditions remain satisfied. Failed rows outside `retry_mask` +enter external recovery, and unresolved evidence at the action deadline removes +covered active verified state before recovery. + `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 diff --git a/docs/source/tutorial/atomic_actions.rst b/docs/source/tutorial/atomic_actions.rst index adceebbed..93907fd7e 100644 --- a/docs/source/tutorial/atomic_actions.rst +++ b/docs/source/tutorial/atomic_actions.rst @@ -489,6 +489,8 @@ correlated per-environment verification result: .. code-block:: python + import torch + from embodichain.lab.sim.atomic_actions import EffectVerificationResult def verify_effect(context, request): @@ -497,6 +499,8 @@ correlated per-environment verification result: 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) @@ -518,6 +522,8 @@ can later resume from the *current* pending request: 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: @@ -536,7 +542,13 @@ 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. +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. Adding an action ---------------- diff --git a/embodichain/lab/sim/atomic_actions/__init__.py b/embodichain/lab/sim/atomic_actions/__init__.py index 681092706..733097151 100644 --- a/embodichain/lab/sim/atomic_actions/__init__.py +++ b/embodichain/lab/sim/atomic_actions/__init__.py @@ -59,6 +59,7 @@ unregister_action, ) from .execution import ( + EffectExpectationResult, EffectVerificationRequest, EffectVerificationResult, ExecutionEvent, @@ -262,6 +263,7 @@ "EndpointCommandRouter", "EndpointCommandTransport", "EntityState", + "EffectExpectationResult", "EffectVerificationRequest", "EffectVerificationRequirement", "EffectVerificationResult", diff --git a/embodichain/lab/sim/atomic_actions/execution.py b/embodichain/lab/sim/atomic_actions/execution.py index 1c251ef54..6b463dfe5 100644 --- a/embodichain/lab/sim/atomic_actions/execution.py +++ b/embodichain/lab/sim/atomic_actions/execution.py @@ -18,7 +18,7 @@ from __future__ import annotations -from dataclasses import dataclass +from dataclasses import dataclass, field from enum import Enum import math from typing import TYPE_CHECKING @@ -229,6 +229,8 @@ class EffectVerificationRequest: 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. + ``failure_invalidation`` is a core-owned removal-only delta; verification + results may select failed rows on which to apply it but cannot replace it. """ verification_id: int @@ -243,6 +245,7 @@ class EffectVerificationRequest: env_mask: torch.Tensor expected_effects: StateDelta effect_verification: EffectVerificationRequirement | None = None + failure_invalidation: StateDelta = field(default_factory=StateDelta) def __post_init__(self) -> None: if type(self.verification_id) is not int or self.verification_id < 0: @@ -290,8 +293,32 @@ def __post_init__(self) -> None: "Effect verification requires expected symbolic effects or an " "explicit physical-effect requirement." ) + if not isinstance(self.failure_invalidation, StateDelta): + raise TypeError("failure_invalidation must be a StateDelta.") + if ( + any( + value is not None + for value in self.failure_invalidation.held_object_updates.values() + ) + or any( + value is not None + for value in self.failure_invalidation.coordinated_held_object_updates.values() + ) + or any( + value is not None + for value in self.failure_invalidation.articulation_joint_updates.values() + ) + ): + raise ValueError( + "failure_invalidation may only remove previously verified state." + ) object.__setattr__(self, "env_mask", self.env_mask.clone()) object.__setattr__(self, "expected_effects", self.expected_effects.snapshot()) + object.__setattr__( + self, + "failure_invalidation", + self.failure_invalidation.snapshot(), + ) object.__setattr__( self, "effect_verification", @@ -317,6 +344,74 @@ def snapshot(self) -> EffectVerificationRequest: env_mask=self.env_mask, expected_effects=self.expected_effects, effect_verification=self.effect_verification, + failure_invalidation=self.failure_invalidation, + ) + + +@dataclass(frozen=True, slots=True, eq=False) +class EffectExpectationResult: + """Current-observation outcome for one physical state expectation. + + ``inverse_satisfied_mask`` is stronger than contradiction: every clause + must have reached its explicit inverse band for the monitor's complete + hysteresis window. It may therefore be used to retain a pre-existing + relation during failure reconciliation, while a single contradictory + clause may not. + """ + + expectation_id: str + satisfied_mask: torch.Tensor + contradicted_mask: torch.Tensor + inverse_satisfied_mask: torch.Tensor + + def __post_init__(self) -> None: + if ( + type(self.expectation_id) is not str + or not self.expectation_id + or self.expectation_id != self.expectation_id.strip() + ): + raise ValueError( + "expectation_id must be a non-empty string without outer whitespace." + ) + for name in ( + "satisfied_mask", + "contradicted_mask", + "inverse_satisfied_mask", + ): + value = getattr(self, name) + if not isinstance(value, torch.Tensor): + raise TypeError(f"{name} must be a torch.Tensor.") + if value.dtype != torch.bool or value.dim() != 1: + raise ValueError(f"{name} must be a one-dimensional bool tensor.") + masks = ( + self.satisfied_mask, + self.contradicted_mask, + self.inverse_satisfied_mask, + ) + if any(mask.shape != masks[0].shape for mask in masks[1:]): + raise ValueError("Expectation-result masks must have equal shapes.") + if any(mask.device != masks[0].device for mask in masks[1:]): + raise ValueError("Expectation-result masks must use the same device.") + if (self.satisfied_mask & self.contradicted_mask).any(): + raise ValueError("satisfied_mask and contradicted_mask must not overlap.") + if (self.inverse_satisfied_mask & ~self.contradicted_mask).any(): + raise ValueError( + "inverse_satisfied_mask must be a subset of contradicted_mask." + ) + for name in ( + "satisfied_mask", + "contradicted_mask", + "inverse_satisfied_mask", + ): + object.__setattr__(self, name, getattr(self, name).clone()) + + def snapshot(self) -> EffectExpectationResult: + """Return an independently owned expectation outcome.""" + return EffectExpectationResult( + expectation_id=self.expectation_id, + satisfied_mask=self.satisfied_mask, + contradicted_mask=self.contradicted_mask, + inverse_satisfied_mask=self.inverse_satisfied_mask, ) @@ -324,32 +419,96 @@ def snapshot(self) -> EffectVerificationRequest: class EffectVerificationResult: """Correlated per-environment update for one effect boundary. - Rows absent from both masks remain unresolved. This lets one shared batch - barrier commit verified rows while other rows continue observing the same - physical effect. + Rows absent from both ``success_mask`` and ``failure_mask`` remain + unresolved. ``invalidation_mask`` and ``retry_mask`` classify only failed + rows: the former selects the request's core-owned removal delta, while the + latter authorizes replay of the same invocation. Failed rows outside the + retry mask require external recovery. This lets one shared batch barrier + commit verified rows while other rows continue observing the same physical + effect. """ verification_id: int success_mask: torch.Tensor failure_mask: torch.Tensor + invalidation_mask: torch.Tensor + retry_mask: torch.Tensor + expectation_results: tuple[EffectExpectationResult, ...] = () def __post_init__(self) -> None: if type(self.verification_id) is not int or self.verification_id < 0: raise ValueError("verification_id must be a non-negative integer.") - for name in ("success_mask", "failure_mask"): + for name in ( + "success_mask", + "failure_mask", + "invalidation_mask", + "retry_mask", + ): value = getattr(self, name) if not isinstance(value, torch.Tensor): raise TypeError(f"{name} must be a torch.Tensor.") if value.dtype != torch.bool or value.dim() != 1: raise ValueError(f"{name} must be a one-dimensional bool tensor.") - if self.success_mask.shape != self.failure_mask.shape: - raise ValueError("success_mask and failure_mask must have equal shapes.") - if self.success_mask.device != self.failure_mask.device: - raise ValueError("success_mask and failure_mask must use the same device.") + masks = ( + self.success_mask, + self.failure_mask, + self.invalidation_mask, + self.retry_mask, + ) + if any(mask.shape != masks[0].shape for mask in masks[1:]): + raise ValueError("Effect-result masks must have equal shapes.") + if any(mask.device != masks[0].device for mask in masks[1:]): + raise ValueError("Effect-result masks must use the same device.") if (self.success_mask & self.failure_mask).any(): raise ValueError("success_mask and failure_mask must not overlap.") - object.__setattr__(self, "success_mask", self.success_mask.clone()) - object.__setattr__(self, "failure_mask", self.failure_mask.clone()) + if (self.invalidation_mask & ~self.failure_mask).any(): + raise ValueError("invalidation_mask must be a subset of failure_mask.") + if (self.retry_mask & ~self.failure_mask).any(): + raise ValueError("retry_mask must be a subset of failure_mask.") + expectation_results = tuple(self.expectation_results) + if not all( + type(value) is EffectExpectationResult for value in expectation_results + ): + raise TypeError( + "expectation_results must contain exact EffectExpectationResult values." + ) + expectation_ids = [value.expectation_id for value in expectation_results] + if len(set(expectation_ids)) != len(expectation_ids): + raise ValueError("Effect expectation-result IDs must be unique.") + if expectation_results: + expected_success = torch.ones_like(self.success_mask) + expected_failure = torch.zeros_like(self.failure_mask) + for value in expectation_results: + if value.satisfied_mask.shape != self.success_mask.shape: + raise ValueError( + "Expectation and aggregate result masks must have equal shapes." + ) + if value.satisfied_mask.device != self.success_mask.device: + raise ValueError( + "Expectation and aggregate result masks must use the same device." + ) + expected_success &= value.satisfied_mask + expected_failure |= value.contradicted_mask + if not torch.equal(self.success_mask, expected_success): + raise ValueError( + "success_mask must equal the conjunction of expectation results." + ) + if not torch.equal(self.failure_mask, expected_failure): + raise ValueError( + "failure_mask must equal the union of expectation results." + ) + for name in ( + "success_mask", + "failure_mask", + "invalidation_mask", + "retry_mask", + ): + object.__setattr__(self, name, getattr(self, name).clone()) + object.__setattr__( + self, + "expectation_results", + tuple(value.snapshot() for value in expectation_results), + ) @dataclass(frozen=True, slots=True, eq=False) @@ -1062,12 +1221,36 @@ def tick( self._pending_effect.env_mask & self._pending & self._plan.plan_success ) if self._action_timed_out(self._plan, execution_mask): + pending_request = self._pending_effect timed_out = execution_mask.clone() known_failures = self._effect_failures.clone() planning_failed = self._pending & ~self._plan.plan_success - retry_mask = timed_out | known_failures | planning_failed + invalidation_presence = self._failure_invalidation_presence_mask( + pending_request.failure_invalidation + ) + external_recovery = timed_out & invalidation_presence + retry_mask = ( + (timed_out & ~external_recovery) | known_failures | planning_failed + ) + self._apply_effect_failure_invalidation( + pending_request.failure_invalidation, + timed_out, + ) self._pending_effect = None self._effect_failures.zero_() + if external_recovery.any(): + self._eligible &= ~external_recovery + self._pending &= ~external_recovery + self._last_command_mask &= ~external_recovery + events.append( + self._event( + ExecutionEventKind.RECOVERY_REQUIRED, + external_recovery, + "Effect evidence remained unresolved at the action " + "deadline, so previously verified state was " + "invalidated before external recovery.", + ) + ) if known_failures.any(): events.append( self._event( @@ -1758,6 +1941,8 @@ def _finish_action( ) return None, active_targets, events else: + assert self._pending_effect is not None + pending_request = self._pending_effect success_input = self._normalize_mask( effect_result.success_mask, "effect_result.success_mask", @@ -1766,17 +1951,67 @@ def _finish_action( effect_result.failure_mask, "effect_result.failure_mask", ) + invalidation_input = self._normalize_mask( + effect_result.invalidation_mask, + "effect_result.invalidation_mask", + ) + retry_input = self._normalize_mask( + effect_result.retry_mask, + "effect_result.retry_mask", + ) reported = success_input | failure_input if (reported & ~execution_mask).any(): raise ValueError( "Effect verification masks must be subsets of the pending " "effect request env_mask." ) + for outcome in effect_result.expectation_results: + for name in ( + "satisfied_mask", + "contradicted_mask", + "inverse_satisfied_mask", + ): + outcome_mask = self._normalize_mask( + getattr(outcome, name), + f"effect_result.expectation_results.{name}", + ) + if (outcome_mask & ~execution_mask).any(): + raise ValueError( + "Effect expectation-result masks must be subsets " + "of the pending effect request env_mask." + ) verified = execution_mask & success_input failed_effect = execution_mask & failure_input unresolved = execution_mask & ~reported made_progress = bool(reported.any().item()) - self._effect_failures |= failed_effect + invalidated = failed_effect & invalidation_input + retryable_failure = failed_effect & retry_input + external_recovery = failed_effect & ~retry_input + self._apply_effect_failure_invalidation( + pending_request.failure_invalidation, + invalidated, + ) + self._effect_failures |= retryable_failure + if external_recovery.any(): + self._eligible &= ~external_recovery + self._pending &= ~external_recovery + self._effect_failures &= ~external_recovery + self._last_command_mask &= ~external_recovery + events.extend( + ( + self._event( + ExecutionEventKind.EFFECT_VERIFICATION_FAILED, + external_recovery, + "Required physical effects were contradicted.", + ), + self._event( + ExecutionEventKind.RECOVERY_REQUIRED, + external_recovery, + "The reconciled effect failure cannot safely replay " + "the current invocation.", + ), + ) + ) if not unresolved.any(): self._pending_effect = None @@ -1796,8 +2031,12 @@ def _finish_action( if made_progress: self._pending_effect = self._effect_verification_request(unresolved) return None, active_targets, events - retry_mask = self._effect_failures | planning_failed - if retry_mask.any(): + terminal_event = self._update_terminal_status() + if terminal_event is not None: + events.append(terminal_event) + return None, active_targets, events + retry_candidates = self._effect_failures | planning_failed + if retry_candidates.any(): effect_failure_mask = self._effect_failures.clone() self._effect_failures.zero_() reason = ( @@ -1806,7 +2045,7 @@ def _finish_action( else ExecutionEventKind.ACTION_PLANNING_FAILED ) reason_mask = ( - effect_failure_mask if effect_failure_mask.any() else retry_mask + effect_failure_mask if effect_failure_mask.any() else retry_candidates ) if effect_failure_mask.any() and planning_failed.any(): events.append( @@ -1818,7 +2057,7 @@ def _finish_action( ) events.extend( self._attempt_action_retry( - retry_mask, + retry_candidates, reason, "Planning or expected-effect verification failed.", reason_mask=reason_mask, @@ -2328,8 +2567,69 @@ def _effect_verification_request( env_mask=env_mask, expected_effects=self._plan.expected_effects, effect_verification=self._plan.effect_verification, + failure_invalidation=self._effect_failure_invalidation(), + ) + + def _effect_failure_invalidation(self) -> StateDelta: + """Build the core-owned fail-closed state removal for this effect.""" + assert self._plan is not None + expected = self._plan.expected_effects + held_keys = set(expected.held_object_updates) + coordinated_keys = set(expected.coordinated_held_object_updates) + coordinated_keys.update( + resources + for resources in self._task_state.coordinated_held_objects + if not set(resources).isdisjoint(held_keys) + ) + return StateDelta( + held_object_updates={key: None for key in held_keys}, + coordinated_held_object_updates={ + resources: None for resources in coordinated_keys + }, + articulation_joint_updates={ + key: None for key in expected.articulation_joint_updates + }, + ) + + def _apply_effect_failure_invalidation( + self, + state_invalidation: StateDelta, + env_mask: torch.Tensor, + ) -> None: + """Apply a request-owned failure delta and refresh planning context.""" + if not env_mask.any() or state_invalidation.is_empty: + return + self._task_state = state_invalidation.apply(self._task_state, env_mask) + self._context = PlanningContext( + robot=self._context.robot, + task=self._task_state, + scene=self._context.scene, + env_ids=self._context.env_ids, ) + def _failure_invalidation_presence_mask( + self, + state_invalidation: StateDelta, + ) -> torch.Tensor: + """Return rows whose verified state would actually be removed.""" + present = torch.zeros_like(self._eligible) + for key in state_invalidation.held_object_updates: + value = self._task_state.held_objects.get(key) + if value is not None: + assert value.env_mask is not None + present |= value.env_mask.to(present.device) + for key in state_invalidation.coordinated_held_object_updates: + value = self._task_state.coordinated_held_objects.get(key) + if value is not None: + assert value.env_mask is not None + present |= value.env_mask.to(present.device) + for key in state_invalidation.articulation_joint_updates: + value = self._task_state.articulation_joints.get(key) + if value is not None: + assert value.env_mask is not None + present |= value.env_mask.to(present.device) + return present + def _event( self, kind: ExecutionEventKind, @@ -2403,6 +2703,7 @@ def _tick_result( __all__ = [ + "EffectExpectationResult", "EffectVerificationRequest", "EffectVerificationResult", "ExecutionEvent", diff --git a/embodichain/lab/sim/skills/runtime.py b/embodichain/lab/sim/skills/runtime.py index b9d6bd96a..095ce5853 100644 --- a/embodichain/lab/sim/skills/runtime.py +++ b/embodichain/lab/sim/skills/runtime.py @@ -31,6 +31,7 @@ from ..atomic_actions.engine import AtomicActionEngine from ..atomic_actions.effects import StateDelta from ..atomic_actions.execution import ( + EffectExpectationResult, EffectVerificationRequest, EffectVerificationResult, ExecutionEvent, @@ -57,7 +58,7 @@ TrackingMetricCfg, TrackingPolicy, ) -from .calls import SemanticCallSpec +from .calls import HandOver, Pick, Place, SemanticCallSpec from .compiler import ( GroundedHeldObjectGuard, HeldObjectGuardBaseline, @@ -66,6 +67,7 @@ from .effects import ( BinaryEffectEvidenceBatch, EffectEvidenceBatch, + EffectExpectationDecision, EffectMonitor, EffectMonitorDecision, EffectMonitorRef, @@ -955,6 +957,7 @@ class SkillEffectTrace: timestamp: float success_mask: torch.Tensor failure_mask: torch.Tensor + expectation_decisions: tuple[EffectExpectationDecision, ...] effect_spec: SemanticEffectSpec monitor_id: str monitor_revision: str | None @@ -1003,6 +1006,54 @@ def __post_init__(self) -> None: raise ValueError("Effect trace masks must not overlap.") if not isinstance(self.effect_spec, SemanticEffectSpec): raise TypeError("effect_spec must be a SemanticEffectSpec.") + expectation_decisions = tuple(self.expectation_decisions) + if not all( + type(value) is EffectExpectationDecision for value in expectation_decisions + ): + raise TypeError( + "expectation_decisions must contain exact " + "EffectExpectationDecision values." + ) + for value in expectation_decisions: + if value.satisfied_mask.shape != self.success_mask.shape: + raise ValueError( + "Expectation and aggregate trace masks must have equal shapes." + ) + if value.satisfied_mask.device != self.success_mask.device: + raise ValueError( + "Expectation and aggregate trace masks must share a device." + ) + physical_ids = tuple( + expectation.expectation_id + for expectation in self.effect_spec.state_expectations + if any( + clause.expectation_id == expectation.expectation_id + for clause in self.effect_spec.clauses + ) + ) + outcome_ids = tuple(value.expectation_id for value in expectation_decisions) + if outcome_ids != physical_ids: + raise ValueError( + "Effect trace must contain one ordered outcome for every " + f"physical expectation; expected={physical_ids}, " + f"got={outcome_ids}." + ) + if expectation_decisions: + expected_success = torch.ones_like(self.success_mask) + expected_failure = torch.zeros_like(self.failure_mask) + for value in expectation_decisions: + expected_success &= value.satisfied_mask + expected_failure |= value.contradicted_mask + if not torch.equal(self.success_mask, expected_success): + raise ValueError( + "success_mask must equal the conjunction of expectation " + "trace outcomes." + ) + if not torch.equal(self.failure_mask, expected_failure): + raise ValueError( + "failure_mask must equal the union of expectation trace " + "outcomes." + ) if type(self.monitor_id) is not str or not self.monitor_id: raise ValueError("monitor_id must be a non-empty string.") if self.monitor_revision is not None and ( @@ -1026,6 +1077,11 @@ def __post_init__(self) -> None: evidence[evidence_id] = batch.snapshot() object.__setattr__(self, "success_mask", self.success_mask.clone()) object.__setattr__(self, "failure_mask", self.failure_mask.clone()) + object.__setattr__( + self, + "expectation_decisions", + tuple(value.snapshot() for value in expectation_decisions), + ) object.__setattr__(self, "effect_spec", self.effect_spec.snapshot()) object.__setattr__( self, @@ -1048,6 +1104,7 @@ def snapshot(self) -> SkillEffectTrace: timestamp=self.timestamp, success_mask=self.success_mask, failure_mask=self.failure_mask, + expectation_decisions=self.expectation_decisions, effect_spec=self.effect_spec, monitor_id=self.monitor_id, monitor_revision=self.monitor_revision, @@ -1080,6 +1137,17 @@ def to_metadata(self) -> dict[str, object]: "decision": { "success_mask": _metadata_value(self.success_mask), "failure_mask": _metadata_value(self.failure_mask), + "expectations": [ + { + "expectation_id": value.expectation_id, + "satisfied_mask": _metadata_value(value.satisfied_mask), + "contradicted_mask": _metadata_value(value.contradicted_mask), + "inverse_satisfied_mask": _metadata_value( + value.inverse_satisfied_mask + ), + } + for value in self.expectation_decisions + ], }, } metadata["boundary"] = {"kind": self.boundary_kind} @@ -2147,12 +2215,94 @@ def _effect_verifier( spec=spec, monitor=monitor, ) + expectation_decisions = self._validated_expectation_decisions( + spec, + decision, + ) + invalidation_mask, retry_mask = self._terminal_failure_policy( + grounded, + decision.failure_mask, + expectation_decisions, + ) return EffectVerificationResult( verification_id=request.verification_id, success_mask=decision.success_mask, failure_mask=decision.failure_mask, + invalidation_mask=invalidation_mask, + retry_mask=retry_mask, + expectation_results=tuple( + EffectExpectationResult( + expectation_id=value.expectation_id, + satisfied_mask=value.satisfied_mask, + contradicted_mask=value.contradicted_mask, + inverse_satisfied_mask=value.inverse_satisfied_mask, + ) + for value in expectation_decisions + ), ) + @staticmethod + def _validated_expectation_decisions( + spec: SemanticEffectSpec, + decision: EffectMonitorDecision, + ) -> tuple[EffectExpectationDecision, ...]: + """Require one current-observation outcome per physical expectation.""" + physical_ids = tuple( + expectation.expectation_id + for expectation in spec.state_expectations + if any( + clause.expectation_id == expectation.expectation_id + for clause in spec.clauses + ) + ) + outcomes = tuple(decision.expectation_decisions) + if not outcomes and len(physical_ids) == 1: + outcomes = ( + EffectExpectationDecision( + expectation_id=physical_ids[0], + satisfied_mask=decision.success_mask, + contradicted_mask=decision.failure_mask, + inverse_satisfied_mask=torch.zeros_like(decision.failure_mask), + ), + ) + outcome_ids = tuple(value.expectation_id for value in outcomes) + if outcome_ids != physical_ids: + raise ValueError( + "Effect monitor must return one ordered outcome for every " + f"physical expectation; expected={physical_ids}, got={outcome_ids}." + ) + return outcomes + + @staticmethod + def _terminal_failure_policy( + grounded: object, + failure_mask: torch.Tensor, + expectation_decisions: tuple[EffectExpectationDecision, ...], + ) -> tuple[torch.Tensor, torch.Tensor]: + """Select fail-closed invalidation and safe local retry rows.""" + call = getattr(getattr(grounded, "analyzed", None), "call", None) + invalidation = failure_mask.clone() + retry = failure_mask.clone() + if type(call) is Pick: + return invalidation, retry + if type(call) is Place: + source = next( + value + for value in expectation_decisions + if value.expectation_id == "source" + ) + retained = failure_mask & source.inverse_satisfied_mask + return failure_mask & ~retained, retained + if type(call) is HandOver: + source = next( + value + for value in expectation_decisions + if value.expectation_id == "source" + ) + retained = failure_mask & source.inverse_satisfied_mask + return failure_mask & ~retained, torch.zeros_like(failure_mask) + return invalidation, retry + def _held_object_guard_verifier( self, context: PlanningContext, @@ -2310,7 +2460,16 @@ def _observe_effect_monitor( observation_revision=observation_revision, env_ids=selected_env_ids, ) - decision = monitor.observe(request, evidence) + observed = monitor.observe(request, evidence) + expectation_decisions = self._validated_expectation_decisions( + spec, + observed, + ) + decision = EffectMonitorDecision( + success_mask=observed.success_mask, + failure_mask=observed.failure_mask, + expectation_decisions=expectation_decisions, + ) analyzed = getattr(grounded, "analyzed", None) monitor_ref = getattr(analyzed, "effect_monitor_ref", None) if monitor_ref is not None and not isinstance(monitor_ref, EffectMonitorRef): @@ -2333,6 +2492,7 @@ def _observe_effect_monitor( timestamp=context.robot.timestamp, success_mask=decision.success_mask, failure_mask=decision.failure_mask, + expectation_decisions=decision.expectation_decisions, effect_spec=spec, monitor_id=monitor_id, monitor_revision=monitor_revision, diff --git a/scripts/tutorials/atomic_action/moving_target_recovery.py b/scripts/tutorials/atomic_action/moving_target_recovery.py index 4c76f5968..fb34bb829 100644 --- a/scripts/tutorials/atomic_action/moving_target_recovery.py +++ b/scripts/tutorials/atomic_action/moving_target_recovery.py @@ -432,6 +432,8 @@ def verify_pickup_effect( verification_id=request.verification_id, success_mask=verified_success, failure_mask=request.env_mask & ~success, + invalidation_mask=request.env_mask & ~success, + retry_mask=request.env_mask & ~success, ) recording_started = start_auto_play_recording( diff --git a/tests/sim/atomic_actions/test_engine_per_env.py b/tests/sim/atomic_actions/test_engine_per_env.py index 85bf987f5..3386af1c7 100644 --- a/tests/sim/atomic_actions/test_engine_per_env.py +++ b/tests/sim/atomic_actions/test_engine_per_env.py @@ -42,6 +42,7 @@ EndpointTrackingChannelBinding, EndpointTrackingFeedbackAddress, EntityState, + EffectExpectationResult, ExecutionEventKind, ExecutionSession, ExecutionStatus, @@ -83,6 +84,30 @@ from embodichain.lab.sim.planners import PlanOptions +def _effect_result( + verification_id: int, + success_mask: torch.Tensor, + failure_mask: torch.Tensor, + *, + invalidation_mask: torch.Tensor | None = None, + retry_mask: torch.Tensor | None = None, + expectation_results: tuple[EffectExpectationResult, ...] = (), +) -> EffectVerificationResult: + """Build an explicit terminal decision with legacy retry semantics.""" + return EffectVerificationResult( + verification_id=verification_id, + success_mask=success_mask, + failure_mask=failure_mask, + invalidation_mask=( + torch.zeros_like(failure_mask) + if invalidation_mask is None + else invalidation_mask + ), + retry_mask=failure_mask if retry_mask is None else retry_mask, + expectation_results=expectation_results, + ) + + class DynamicAction(AtomicAction[EndEffectorPoseGoal, ActionOptions]): """Test action whose terminal joint command follows a scene entity x pose.""" @@ -692,6 +717,7 @@ def _effect_session( action_timeout: float = 30.0, eligible_mask: torch.Tensor | None = None, action: DynamicAction | None = None, + task_state: TaskState | None = None, ) -> tuple[ExecutionSession, ExecutionTick]: """Advance a test effect action to its verification boundary.""" engine, _ = _engine(batch_size=batch_size) @@ -711,9 +737,12 @@ def _effect_session( ) qpos = tuple(0.0 for _ in range(batch_size)) target = tuple(0.2 for _ in range(batch_size)) + initial_context = _context(0.0, qpos, target, 0) + if task_state is not None: + initial_context = replace(initial_context, task=task_state) session = engine.start( (invocation,), - _context(0.0, qpos, target, 0), + initial_context, eligible_mask=eligible_mask, ) session.tick(_context(0.0, qpos, target, 0)) @@ -1988,7 +2017,7 @@ def test_explicit_verification_with_empty_delta_preserves_task_state() -> None: completed = session.tick( _context(0.21, 0.2, 0.2, 0), - effect_result=EffectVerificationResult( + effect_result=_effect_result( request.verification_id, success_mask=torch.tensor([True]), failure_mask=torch.tensor([False]), @@ -2013,7 +2042,7 @@ def test_explicit_verification_keeps_partial_and_retry_row_lifecycle() -> None: retry = session.tick( _context(0.21, (0.2, 0.2), (0.2, 0.2), 0), - effect_result=EffectVerificationResult( + effect_result=_effect_result( first_request.verification_id, success_mask=torch.tensor([True, False]), failure_mask=torch.tensor([False, True]), @@ -2043,7 +2072,7 @@ def test_explicit_verification_keeps_partial_and_retry_row_lifecycle() -> None: completed = session.tick( _context(0.25, (0.2, 0.2), (0.2, 0.2), 0), - effect_result=EffectVerificationResult( + effect_result=_effect_result( second_request.verification_id, success_mask=torch.tensor([False, True]), failure_mask=torch.tensor([False, False]), @@ -2068,7 +2097,7 @@ def test_explicit_verification_partial_success_shrinks_request_without_state_del partial = session.tick( _context(0.21, (0.2, 0.2), (0.2, 0.2), 0), - effect_result=EffectVerificationResult( + effect_result=_effect_result( first_request.verification_id, success_mask=torch.tensor([True, False]), failure_mask=torch.tensor([False, False]), @@ -2087,7 +2116,7 @@ def test_explicit_verification_partial_success_shrinks_request_without_state_del completed = session.tick( _context(0.22, (0.2, 0.2), (0.2, 0.2), 0), - effect_result=EffectVerificationResult( + effect_result=_effect_result( second_request.verification_id, success_mask=torch.tensor([False, True]), failure_mask=torch.tensor([False, False]), @@ -2143,7 +2172,7 @@ def test_nonempty_effect_is_committed_only_after_external_verification() -> None still_waiting = session.tick(_context(0.25, 0.2, 0.2, 0)) completed = session.tick( _context(0.3, 0.2, 0.2, 0), - effect_result=EffectVerificationResult( + effect_result=_effect_result( waiting.pending_effect.verification_id, torch.tensor([True]), torch.tensor([False]), @@ -2186,7 +2215,7 @@ def test_initially_ineligible_rows_never_receive_effects() -> None: completed = session.tick( _context(0.21, (0.2, 0.2), (0.2, 0.2), 0), - effect_result=EffectVerificationResult( + effect_result=_effect_result( request.verification_id, success_mask=torch.tensor([True, False]), failure_mask=torch.tensor([False, False]), @@ -2207,7 +2236,7 @@ def test_partial_effect_success_commits_resolved_rows_and_shrinks_request() -> N no_progress = session.tick( _context(0.205, (0.2, 0.2), (0.2, 0.2), 0), - effect_result=EffectVerificationResult( + effect_result=_effect_result( first_request.verification_id, success_mask=torch.tensor([False, False]), failure_mask=torch.tensor([False, False]), @@ -2218,7 +2247,7 @@ def test_partial_effect_success_commits_resolved_rows_and_shrinks_request() -> N partial = session.tick( _context(0.21, (0.2, 0.2), (0.2, 0.2), 0), - effect_result=EffectVerificationResult( + effect_result=_effect_result( first_request.verification_id, success_mask=torch.tensor([True, False]), failure_mask=torch.tensor([False, False]), @@ -2241,7 +2270,7 @@ def test_partial_effect_success_commits_resolved_rows_and_shrinks_request() -> N with pytest.raises(ValueError, match="verification_id"): session.tick( _context(0.22, (0.2, 0.2), (0.2, 0.2), 0), - effect_result=EffectVerificationResult( + effect_result=_effect_result( first_request.verification_id, success_mask=torch.tensor([False, True]), failure_mask=torch.tensor([False, False]), @@ -2251,7 +2280,7 @@ def test_partial_effect_success_commits_resolved_rows_and_shrinks_request() -> N current_request = partial.pending_effect completed = session.tick( _context(0.23, (0.2, 0.2), (0.2, 0.2), 0), - effect_result=EffectVerificationResult( + effect_result=_effect_result( current_request.verification_id, success_mask=torch.tensor([False, True]), failure_mask=torch.tensor([False, False]), @@ -2267,17 +2296,45 @@ def test_partial_effect_success_commits_resolved_rows_and_shrinks_request() -> N def test_effect_result_masks_are_owned_disjoint_and_request_scoped() -> None: success = torch.tensor([True, False]) failure = torch.tensor([False, True]) - result = EffectVerificationResult(0, success, failure) + result = _effect_result(0, success, failure) success.fill_(False) failure.fill_(False) assert result.success_mask.tolist() == [True, False] assert result.failure_mask.tolist() == [False, True] with pytest.raises(ValueError, match="must not overlap"): - EffectVerificationResult( + _effect_result( + 0, + torch.tensor([True, False]), + torch.tensor([True, False]), + ) + with pytest.raises(ValueError, match="invalidation_mask must be a subset"): + _effect_result( 0, + torch.tensor([False, False]), torch.tensor([True, False]), + invalidation_mask=torch.tensor([False, True]), + ) + with pytest.raises(ValueError, match="retry_mask must be a subset"): + _effect_result( + 0, + torch.tensor([False, False]), torch.tensor([True, False]), + retry_mask=torch.tensor([False, True]), + ) + with pytest.raises(ValueError, match="conjunction"): + _effect_result( + 0, + torch.tensor([False, False]), + torch.tensor([False, True]), + expectation_results=( + EffectExpectationResult( + expectation_id="destination", + satisfied_mask=torch.tensor([True, False]), + contradicted_mask=torch.tensor([False, True]), + inverse_satisfied_mask=torch.tensor([False, False]), + ), + ), ) session, waiting = _effect_session(batch_size=2) @@ -2300,7 +2357,7 @@ def test_effect_result_masks_are_owned_disjoint_and_request_scoped() -> None: partial = session.tick( _context(0.21, (0.2, 0.2), (0.2, 0.2), 0), - effect_result=EffectVerificationResult( + effect_result=_effect_result( preserved.verification_id, success_mask=torch.tensor([True, False]), failure_mask=torch.tensor([False, False]), @@ -2315,7 +2372,7 @@ def test_effect_result_masks_are_owned_disjoint_and_request_scoped() -> None: with pytest.raises(ValueError, match="subsets"): session.tick( _context(0.22, (0.2, 0.2), (0.2, 0.2), 0), - effect_result=EffectVerificationResult( + effect_result=_effect_result( current.verification_id, success_mask=torch.tensor([True, False]), failure_mask=torch.tensor([False, False]), @@ -2366,7 +2423,7 @@ def test_partial_effect_failure_waits_for_unresolved_rows_then_retries_failure() partial = session.tick( _context(0.21, (0.2, 0.2), (0.2, 0.2), 0), - effect_result=EffectVerificationResult( + effect_result=_effect_result( request.verification_id, success_mask=torch.tensor([False, False]), failure_mask=torch.tensor([True, False]), @@ -2382,7 +2439,7 @@ def test_partial_effect_failure_waits_for_unresolved_rows_then_retries_failure() unresolved_request = partial.pending_effect resolved = session.tick( _context(0.22, (0.2, 0.2), (0.2, 0.2), 0), - effect_result=EffectVerificationResult( + effect_result=_effect_result( unresolved_request.verification_id, success_mask=torch.tensor([False, True]), failure_mask=torch.tensor([False, False]), @@ -2410,6 +2467,95 @@ def test_partial_effect_failure_waits_for_unresolved_rows_then_retries_failure() assert retry_command.command.active_mask.tolist() == [True, False] +def test_effect_failure_applies_request_owned_invalidation_before_recovery() -> None: + initial = _with_held_object(_context(0.0, (0.0, 0.0), (0.2, 0.2), 0)).task + session, waiting = _effect_session( + batch_size=2, + max_action_retries=1, + task_state=initial, + ) + request = waiting.pending_effect + assert request is not None + assert request.failure_invalidation.held_object_updates == {"arm": None} + + terminal = session.tick( + _context(0.21, (0.2, 0.2), (0.2, 0.2), 0), + effect_result=_effect_result( + request.verification_id, + success_mask=torch.tensor([False, True]), + failure_mask=torch.tensor([True, False]), + invalidation_mask=torch.tensor([True, False]), + retry_mask=torch.tensor([False, False]), + ), + ) + + held = terminal.task_state.get_held_object("arm") + assert held is not None and held.env_mask is not None + assert held.env_mask.tolist() == [False, True] + assert terminal.eligible_mask.tolist() == [False, True] + assert any( + event.kind is ExecutionEventKind.RECOVERY_REQUIRED + and event.env_mask.tolist() == [True, False] + for event in terminal.events + ) + assert not any( + event.kind is ExecutionEventKind.ACTION_RETRY for event in terminal.events + ) + + +def test_inverse_proof_can_preserve_state_while_failure_requires_recovery() -> None: + initial = _with_held_object(_context(0.0, 0.0, 0.2, 0)).task + session, waiting = _effect_session(task_state=initial) + request = waiting.pending_effect + assert request is not None + failure = torch.tensor([True]) + + terminal = session.tick( + _context(0.21, 0.2, 0.2, 0), + effect_result=_effect_result( + request.verification_id, + success_mask=torch.tensor([False]), + failure_mask=failure, + invalidation_mask=torch.tensor([False]), + retry_mask=torch.tensor([False]), + expectation_results=( + EffectExpectationResult( + expectation_id="source", + satisfied_mask=torch.tensor([False]), + contradicted_mask=failure, + inverse_satisfied_mask=failure, + ), + ), + ), + ) + + held = terminal.task_state.get_held_object("arm") + assert held is not None and held.env_mask is not None and held.env_mask.all() + assert terminal.status is ExecutionStatus.FAILED + assert any( + event.kind is ExecutionEventKind.RECOVERY_REQUIRED for event in terminal.events + ) + + +def test_unresolved_effect_timeout_invalidates_active_state_fail_closed() -> None: + initial = _with_held_object(_context(0.0, 0.0, 0.2, 0)).task + session, waiting = _effect_session( + action_timeout=0.25, + max_action_retries=1, + task_state=initial, + ) + assert waiting.pending_effect is not None + + terminal = session.tick(_context(0.26, 0.2, 0.2, 0)) + + assert terminal.status is ExecutionStatus.FAILED + assert terminal.task_state.get_held_object("arm") is None + kinds = {event.kind for event in terminal.events} + assert ExecutionEventKind.EFFECT_VERIFICATION_TIMEOUT in kinds + assert ExecutionEventKind.RECOVERY_REQUIRED in kinds + assert ExecutionEventKind.ACTION_RETRY not in kinds + + def test_effect_failure_exhaustion_advances_completed_rows_without_empty_request() -> ( None ): @@ -2419,7 +2565,7 @@ def test_effect_failure_exhaustion_advances_completed_rows_without_empty_request terminal = session.tick( _context(0.21, (0.2, 0.2), (0.2, 0.2), 0), - effect_result=EffectVerificationResult( + effect_result=_effect_result( request.verification_id, success_mask=torch.tensor([True, False]), failure_mask=torch.tensor([False, True]), @@ -2449,7 +2595,7 @@ def test_deactivating_last_unresolved_effect_row_advances_barrier() -> None: assert request is not None partial = session.tick( _context(0.21, (0.2, 0.2), (0.2, 0.2), 0), - effect_result=EffectVerificationResult( + effect_result=_effect_result( request.verification_id, success_mask=torch.tensor([True, False]), failure_mask=torch.tensor([False, False]), @@ -2508,7 +2654,7 @@ def test_effect_request_deadline_is_stable_and_accepts_result_at_boundary() -> N completed = session.tick( _context(0.25, 0.2, 0.2, 0), - effect_result=EffectVerificationResult( + effect_result=_effect_result( request.verification_id, success_mask=torch.tensor([True]), failure_mask=torch.tensor([False]), @@ -2540,7 +2686,7 @@ def test_session_revision_cannot_abandon_pending_effect_verification() -> None: assert session.effect_verification_pending is True completed = session.tick( _context(0.3, 0.2, 0.2, 0), - effect_result=EffectVerificationResult( + effect_result=_effect_result( waiting.pending_effect.verification_id, torch.tensor([True]), torch.tensor([False]), @@ -2569,7 +2715,7 @@ def test_effect_failure_does_not_commit_and_exhausts_retry_budget() -> None: assert waiting.pending_effect is not None failed = session.tick( _context(0.3, 0.2, 0.2, 0), - effect_result=EffectVerificationResult( + effect_result=_effect_result( waiting.pending_effect.verification_id, torch.tensor([False]), torch.tensor([True]), @@ -2606,7 +2752,7 @@ def test_pending_effect_timeout_exhausts_without_committing_late_result() -> Non timed_out = session.tick( _context(0.3, 0.2, 0.2, 0), - effect_result=EffectVerificationResult( + effect_result=_effect_result( waiting.pending_effect.verification_id, torch.tensor([True]), torch.tensor([False]), @@ -2631,7 +2777,7 @@ def test_effect_timeout_exhaustion_advances_rows_already_verified() -> None: assert request is not None partial = session.tick( _context(0.21, (0.2, 0.2), (0.2, 0.2), 0), - effect_result=EffectVerificationResult( + effect_result=_effect_result( request.verification_id, success_mask=torch.tensor([True, False]), failure_mask=torch.tensor([False, False]), @@ -2702,7 +2848,7 @@ def test_deferred_effect_failure_charges_concurrent_planning_failures() -> None: assert request is not None partial = session.tick( _context(0.21, (0.2, 0.2, 0.2), (0.2, 0.2, 0.2), 0), - effect_result=EffectVerificationResult( + effect_result=_effect_result( request.verification_id, success_mask=torch.tensor([False, False, False]), failure_mask=torch.tensor([True, False, False]), @@ -2770,7 +2916,7 @@ def test_effect_retry_invalidates_previous_verification_id() -> None: with pytest.raises(ValueError, match="verification_id"): session.tick( _context(0.55, 0.2, 0.2, 0), - effect_result=EffectVerificationResult( + effect_result=_effect_result( old_id, torch.tensor([True]), torch.tensor([False]), diff --git a/tests/sim/atomic_actions/test_runner.py b/tests/sim/atomic_actions/test_runner.py index c83216699..b10f273f5 100644 --- a/tests/sim/atomic_actions/test_runner.py +++ b/tests/sim/atomic_actions/test_runner.py @@ -84,6 +84,28 @@ TARGET_POSITION = 1.0 +def _effect_result( + verification_id: int, + success_mask: torch.Tensor, + failure_mask: torch.Tensor, + *, + invalidation_mask: torch.Tensor | None = None, + retry_mask: torch.Tensor | None = None, +) -> EffectVerificationResult: + """Build an explicit terminal decision with legacy retry semantics.""" + return EffectVerificationResult( + verification_id=verification_id, + success_mask=success_mask, + failure_mask=failure_mask, + invalidation_mask=( + torch.zeros_like(failure_mask) + if invalidation_mask is None + else invalidation_mask + ), + retry_mask=failure_mask if retry_mask is None else retry_mask, + ) + + class FakeClock: """Deterministic clock used by non-blocking runner tests.""" @@ -404,7 +426,7 @@ def _successful_effect_result( request: EffectVerificationRequest, ) -> EffectVerificationResult: """Correlate a successful result with the pending effect boundary.""" - return EffectVerificationResult( + return _effect_result( verification_id=request.verification_id, success_mask=torch.ones( context.batch_size, @@ -424,7 +446,7 @@ def _unresolved_effect_result( request: EffectVerificationRequest, ) -> EffectVerificationResult: """Keep every row pending at the current effect boundary.""" - return EffectVerificationResult( + return _effect_result( verification_id=request.verification_id, success_mask=torch.zeros( context.batch_size, @@ -1093,7 +1115,7 @@ def test_effect_verifier_is_not_called_after_deadline_and_session_retries() -> N def test_effect_result_and_effect_verifier_are_mutually_exclusive() -> None: runner, _, _, sink, action = _make_runner(with_effect=True) - result = EffectVerificationResult( + result = _effect_result( verification_id=0, success_mask=torch.tensor([True]), failure_mask=torch.tensor([False]), @@ -1153,7 +1175,7 @@ def report_no_progress( request: EffectVerificationRequest, ) -> EffectVerificationResult: observed_requests.append((request.verification_id, request.attempt_generation)) - return EffectVerificationResult( + return _effect_result( 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), @@ -1189,7 +1211,7 @@ def verify_in_two_updates( None if held is None or held.env_mask is None else held.env_mask.tolist() ) if request.env_mask.tolist() == [True, True]: - return EffectVerificationResult( + return _effect_result( verification_id=request.verification_id, success_mask=torch.tensor([True, False]), failure_mask=torch.tensor([False, False]), @@ -1198,7 +1220,7 @@ def verify_in_two_updates( 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( + return _effect_result( verification_id=request.verification_id, success_mask=torch.tensor([False, True]), failure_mask=torch.tensor([False, False]), @@ -1311,7 +1333,7 @@ def verify_remaining( context: PlanningContext, request: EffectVerificationRequest, ) -> EffectVerificationResult: - return EffectVerificationResult( + return _effect_result( verification_id=request.verification_id, success_mask=torch.tensor([True, False]), failure_mask=torch.tensor([False, False]), @@ -1331,7 +1353,7 @@ def mismatched_effect_result( context: PlanningContext, request: EffectVerificationRequest, ) -> EffectVerificationResult: - return EffectVerificationResult( + return _effect_result( 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), diff --git a/tests/sim/skills/test_runtime.py b/tests/sim/skills/test_runtime.py index 147b78ddb..0f912ac38 100644 --- a/tests/sim/skills/test_runtime.py +++ b/tests/sim/skills/test_runtime.py @@ -60,7 +60,7 @@ TrackingProjectorRef, ) from embodichain.lab.sim.atomic_actions.tracking import TrackingPolicy -from embodichain.lab.sim.skills.calls import RegisteredSemanticCall +from embodichain.lab.sim.skills.calls import HandOver, Place, RegisteredSemanticCall from embodichain.lab.sim.skills.compiler import ( GroundedHeldObjectGuard, HeldObjectGuardBaseline, @@ -73,6 +73,7 @@ ControlPartEvidenceAddress, EffectEvidenceBatch, EffectEvidenceSourceRef, + EffectExpectationDecision, EffectMonitor, EffectMonitorDecision, HeldObjectRelation, @@ -91,7 +92,7 @@ from embodichain.lab.sim.skills.parallel import ParallelTimingPolicy from embodichain.lab.sim.skills.parallel_runtime import ParallelSkillRuntime from embodichain.lab.sim.skills.profiles import ResourceClaim -from embodichain.lab.sim.skills.scene import SceneRegistry +from embodichain.lab.sim.skills.scene import SceneObjectRef, SceneRegistry BATCH_SIZE = 2 @@ -213,6 +214,7 @@ def observe( return EffectMonitorDecision( self._decision.success_mask, self._decision.failure_mask, + self._decision.expectation_decisions, ) @@ -614,6 +616,88 @@ def test_nonblocking_step_routes_effect_feedback_through_collector() -> None: assert system.compiler.monitors[0].requests[0].verification_id == 0 +def test_runtime_preserves_per_expectation_effect_outcomes_in_trace() -> None: + expectation = EffectExpectationDecision( + expectation_id="joint_target", + satisfied_mask=_mask(True, True), + contradicted_mask=_mask(False, False), + inverse_satisfied_mask=_mask(False, False), + ) + system = _system( + ( + EffectMonitorDecision( + _mask(True, True), + _mask(False, False), + (expectation,), + ), + ) + ) + + result = system.runtime.run(_call("expectation_trace")) + + assert result.status is SkillStatus.COMPLETED + assert len(result.effects) == 1 + recorded = result.effects[0].expectation_decisions + assert len(recorded) == 1 + assert recorded[0].expectation_id == "joint_target" + assert result.to_metadata()["effects"][0]["decision"]["expectations"] == [ + { + "expectation_id": "joint_target", + "satisfied_mask": [True, True], + "contradicted_mask": [False, False], + "inverse_satisfied_mask": [False, False], + } + ] + + +@pytest.mark.parametrize( + ("call", "expected_invalidation", "expected_retry"), + ( + ( + Place( + object=SceneObjectRef("cube"), + inside=SceneObjectRef("bin"), + ), + _mask(False, True), + _mask(True, False), + ), + ( + HandOver(object=SceneObjectRef("cube")), + _mask(False, True), + _mask(False, False), + ), + ), +) +def test_terminal_failure_policy_only_retains_strongly_proven_source_attachment( + call: Place | HandOver, + expected_invalidation: torch.Tensor, + expected_retry: torch.Tensor, +) -> None: + failure = _mask(True, True) + source = EffectExpectationDecision( + expectation_id="source", + satisfied_mask=_mask(False, False), + contradicted_mask=failure, + inverse_satisfied_mask=_mask(True, False), + ) + destination = EffectExpectationDecision( + expectation_id="destination", + satisfied_mask=_mask(False, False), + contradicted_mask=failure, + inverse_satisfied_mask=_mask(False, False), + ) + grounded = SimpleNamespace(analyzed=SimpleNamespace(call=call)) + + invalidation, retry = SkillRuntime._terminal_failure_policy( + grounded, + failure, + (source, destination), + ) + + assert torch.equal(invalidation, expected_invalidation) + assert torch.equal(retry, expected_retry) + + def test_in_flight_guard_collects_live_evidence_and_builds_loss_reconciliation() -> ( None ):