# ----------------------------------------------------------------------------
# Copyright (c) 2021-2026 DexForce Technology Co., Ltd.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
# ----------------------------------------------------------------------------
"""Immutable results and trace metadata produced by Task Program execution."""
from __future__ import annotations
from collections.abc import Callable, Iterable, Iterator, Mapping, Sequence
from dataclasses import dataclass, fields, is_dataclass, replace
from enum import Enum
import math
from types import MappingProxyType
from typing import Generic, TypeVar
import torch
from embodichain.lab.sim.atomic_actions.bindings import EndpointBinding
from embodichain.lab.sim.atomic_actions.execution import (
ExecutionEvent,
ExecutionPlanAttempt,
)
from embodichain.lab.sim.atomic_actions.plans import PlanningFailure, TrajectorySegment
from embodichain.lab.sim.atomic_actions.policies import MotionPolicy, RecoveryPolicy
from embodichain.lab.sim.atomic_actions.runner import RunnerStatus
from embodichain.lab.sim.atomic_actions.state import TaskState
from embodichain.lab.sim.atomic_actions.tracking import (
FeedbackTerminalAcceptance,
TimedTrackingSequence,
TrackingMetricCfg,
TrackingPolicy,
)
from embodichain.lab.task_program.semantics.effects import (
BinaryEffectEvidenceBatch,
EffectEvidenceBatch,
EffectExpectationDecision,
JointStateEvidenceBatch,
PoseRelationEvidenceBatch,
ScalarEffectEvidenceBatch,
SemanticEffectSpec,
)
from embodichain.lab.task_program.semantics.integration import SemanticDiagnostic
def _snapshot_task_state(state: TaskState) -> TaskState:
"""Return a tensor-owning snapshot of verified symbolic state."""
return TaskState(
batch_size=state.batch_size,
device=state.device,
held_objects=state.held_objects,
coordinated_held_objects=state.coordinated_held_objects,
articulation_joints=state.articulation_joints,
)
def _snapshot_event(event: ExecutionEvent) -> ExecutionEvent:
"""Return an independently owned execution event."""
return ExecutionEvent(
kind=event.kind,
timestamp=event.timestamp,
skill_id=event.skill_id,
invocation_id=event.invocation_id,
invocation_revision=event.invocation_revision,
invocation_index=event.invocation_index,
env_mask=event.env_mask,
message=event.message,
segment_name=event.segment_name,
failure_code=event.failure_code,
retryable=event.retryable,
)
def _metadata_value(value: object, *, depth: int = 0) -> object:
"""Convert supported runtime diagnostics to deterministic JSON-safe data."""
if depth > 16:
return {"type": f"{type(value).__module__}.{type(value).__qualname__}"}
if value is None or type(value) in (bool, int, str):
return value
if type(value) is float:
return value if math.isfinite(value) else None
if isinstance(value, Enum):
return value.value
if isinstance(value, torch.Tensor):
return _metadata_value(value.detach().cpu().tolist(), depth=depth + 1)
if isinstance(value, torch.device):
return str(value)
if isinstance(value, type):
return {"__type__": f"{value.__module__}.{value.__qualname__}"}
if isinstance(value, Mapping):
items = sorted(value.items(), key=lambda item: str(item[0]))
if all(type(key) is str and key and key == key.strip() for key, _ in items):
return {
key: _metadata_value(nested, depth=depth + 1) for key, nested in items
}
return {
"__entries__": [
{
"key": _metadata_value(key, depth=depth + 1),
"value": _metadata_value(nested, depth=depth + 1),
}
for key, nested in items
]
}
if isinstance(value, (tuple, list)):
return [_metadata_value(nested, depth=depth + 1) for nested in value]
if isinstance(value, (set, frozenset)):
return [
_metadata_value(nested, depth=depth + 1)
for nested in sorted(value, key=str)
]
return {"type": f"{type(value).__module__}.{type(value).__qualname__}"}
def _freeze_metadata_value(value: object) -> object:
"""Recursively freeze already JSON-safe metadata for immutable traces."""
if isinstance(value, dict):
return MappingProxyType(
{key: _freeze_metadata_value(nested) for key, nested in value.items()}
)
if isinstance(value, list):
return tuple(_freeze_metadata_value(nested) for nested in value)
return value
def _snapshot_metadata_mapping(value: Mapping[str, object]) -> Mapping[str, object]:
"""Own one JSON-safe string-keyed metadata mapping."""
if not isinstance(value, Mapping):
raise TypeError("metadata must be a mapping.")
normalized = _metadata_value(value)
if not isinstance(normalized, dict):
raise TypeError("metadata normalization must produce a dict.")
return MappingProxyType(normalized)
def _event_to_metadata(event: ExecutionEvent) -> dict[str, object]:
"""Serialize one execution/recovery event without exposing tensors."""
metadata: dict[str, object] = {
"kind": event.kind.value,
"timestamp": _metadata_value(event.timestamp),
"skill_id": event.skill_id,
"invocation_id": event.invocation_id,
"invocation_revision": event.invocation_revision,
"invocation_index": event.invocation_index,
"env_mask": _metadata_value(event.env_mask),
"message": event.message,
}
if event.segment_name is not None:
metadata["segment_name"] = event.segment_name
if event.failure_code is not None:
metadata["failure_code"] = event.failure_code
metadata["retryable"] = event.retryable
return metadata
def task_state_to_metadata(state: TaskState) -> dict[str, object]:
"""Return verified symbolic task state as deterministic JSON-safe data."""
if not isinstance(state, TaskState):
raise TypeError("state must be a TaskState.")
held = []
for resource_id, value in sorted(state.held_objects.items()):
held.append(
{
"resource_id": resource_id,
"object_id": value.semantics.entity_id,
"object_label": value.semantics.label,
"object_to_eef": _metadata_value(value.object_to_eef),
"grasp_xpos": _metadata_value(value.grasp_xpos),
"active_mask": _metadata_value(value.env_mask),
}
)
coordinated = []
for resource_ids, value in sorted(state.coordinated_held_objects.items()):
coordinated.append(
{
"resource_ids": list(resource_ids),
"object_id": value.semantics.entity_id,
"object_label": value.semantics.label,
"left_object_to_eef": _metadata_value(value.left_object_to_eef),
"right_object_to_eef": _metadata_value(value.right_object_to_eef),
"left_grasp_xpos": _metadata_value(value.left_grasp_xpos),
"right_grasp_xpos": _metadata_value(value.right_grasp_xpos),
"active_mask": _metadata_value(value.env_mask),
}
)
articulations = []
for (articulation_id, joint_id), value in sorted(state.articulation_joints.items()):
articulations.append(
{
"articulation_id": articulation_id,
"joint_id": joint_id,
"position": _metadata_value(value.position),
"active_mask": _metadata_value(value.env_mask),
}
)
return {
"batch_size": state.batch_size,
"device": str(state.device),
"held_objects": held,
"coordinated_held_objects": coordinated,
"articulation_joints": articulations,
}
[docs]
class SemanticExecutionStatus(str, Enum):
"""Lifecycle state of one semantic workflow run."""
IDLE = "idle"
RUNNING = "running"
COMPLETED = "completed"
FAILED = "failed"
CANCELLED = "cancelled"
class SkillWorkflowRecoveryRole(str, Enum):
"""Role of one real semantic call inside workflow recovery."""
RETRY_RETAINED = "retry_retained"
REACQUIRE = "reacquire"
RETRY_REACQUIRED = "retry_reacquired"
@dataclass(frozen=True, slots=True)
class SkillEndpointTrackingChannelTrace:
"""Stable provider and projector route for one endpoint feedback channel."""
channel_id: str
provider_id: str
provider_revision: str
projector_id: str
projector_revision: str
feedback_address_type: str
address_fingerprint: object
route_fingerprint: object
def __post_init__(self) -> None:
for name in (
"channel_id",
"provider_id",
"provider_revision",
"projector_id",
"projector_revision",
"feedback_address_type",
):
if type(getattr(self, name)) is not str or not getattr(self, name):
raise ValueError(f"{name} must be a non-empty string.")
object.__setattr__(
self,
"address_fingerprint",
_freeze_metadata_value(_metadata_value(self.address_fingerprint)),
)
object.__setattr__(
self,
"route_fingerprint",
_freeze_metadata_value(_metadata_value(self.route_fingerprint)),
)
def to_metadata(self) -> dict[str, object]:
"""Return the exact immutable tracking route without live objects."""
return {
"channel_id": self.channel_id,
"feedback_source": {
"provider_id": self.provider_id,
"revision": self.provider_revision,
"address_type": self.feedback_address_type,
"address_fingerprint": _metadata_value(self.address_fingerprint),
},
"projector": {
"projector_id": self.projector_id,
"revision": self.projector_revision,
},
"route_fingerprint": _metadata_value(self.route_fingerprint),
}
@dataclass(frozen=True, slots=True)
class SkillEndpointBindingTrace:
"""JSON-safe typed projection of one resolved execution endpoint."""
slot_id: str
endpoint_id: str
resource_id: str
adapter_id: str
transport_id: str
target_id: str
target_type: str
task_state_key: str
capabilities: tuple[str, ...]
command_ids: tuple[str, ...]
tracking_channels: tuple[SkillEndpointTrackingChannelTrace, ...]
claim_tokens: tuple[str, ...]
joint_ids: tuple[int, ...]
def __post_init__(self) -> None:
for name in (
"slot_id",
"endpoint_id",
"resource_id",
"adapter_id",
"transport_id",
"target_id",
"target_type",
"task_state_key",
):
value = getattr(self, name)
if type(value) is not str or not value:
raise ValueError(f"{name} must be a non-empty string.")
for name in ("capabilities", "command_ids", "claim_tokens"):
values = tuple(getattr(self, name))
if tuple(sorted(set(values))) != values or not all(
type(value) is str and value for value in values
):
raise ValueError(f"{name} must contain sorted unique identifiers.")
object.__setattr__(self, name, values)
tracking_channels = tuple(self.tracking_channels)
if not all(
type(value) is SkillEndpointTrackingChannelTrace
for value in tracking_channels
):
raise TypeError(
"tracking_channels must contain exact "
"SkillEndpointTrackingChannelTrace values."
)
channel_ids = tuple(value.channel_id for value in tracking_channels)
if tuple(sorted(set(channel_ids))) != channel_ids:
raise ValueError(
"tracking_channels must use sorted unique channel identifiers."
)
object.__setattr__(self, "tracking_channels", tracking_channels)
joint_ids = tuple(self.joint_ids)
if len(set(joint_ids)) != len(joint_ids) or not all(
type(value) is int and value >= 0 for value in joint_ids
):
raise ValueError("joint_ids must contain unique non-negative integers.")
object.__setattr__(self, "joint_ids", joint_ids)
@classmethod
def from_binding(cls, binding: EndpointBinding) -> SkillEndpointBindingTrace:
"""Project one owned endpoint binding without retaining its target."""
if not isinstance(binding, EndpointBinding):
raise TypeError("binding must be an EndpointBinding.")
target = binding.target
return cls(
slot_id=binding.slot_id,
endpoint_id=binding.endpoint_id,
resource_id=binding.resource_id,
adapter_id=binding.adapter_id,
transport_id=target.transport_id,
target_id=target.target_id,
target_type=f"{type(target).__module__}.{type(target).__qualname__}",
task_state_key=binding.task_state_key,
capabilities=tuple(sorted(binding.capabilities)),
command_ids=tuple(sorted(binding.commands)),
tracking_channels=tuple(
SkillEndpointTrackingChannelTrace(
channel_id=channel_id,
provider_id=channel.source.provider_id,
provider_revision=channel.source.revision,
projector_id=channel.projector.projector_id,
projector_revision=channel.projector.revision,
feedback_address_type=(
f"{type(channel.source.address).__module__}."
f"{type(channel.source.address).__qualname__}"
),
address_fingerprint=(channel.source.address.address_fingerprint),
route_fingerprint=channel.route_fingerprint,
)
for channel_id, channel in sorted(binding.tracking_channels.items())
),
claim_tokens=tuple(sorted(binding.claim_tokens)),
joint_ids=binding.joint_ids,
)
def to_metadata(self) -> dict[str, object]:
"""Return stable endpoint, resource, adapter, and transport metadata."""
return {
"slot_id": self.slot_id,
"endpoint_id": self.endpoint_id,
"resource_id": self.resource_id,
"adapter_id": self.adapter_id,
"transport_id": self.transport_id,
"target_id": self.target_id,
"target_type": self.target_type,
"task_state_key": self.task_state_key,
"capabilities": list(self.capabilities),
"command_ids": list(self.command_ids),
"tracking_channels": [
channel.to_metadata() for channel in self.tracking_channels
],
"claim_tokens": list(self.claim_tokens),
"joint_ids": list(self.joint_ids),
}
def _motion_policy_to_metadata(policy: MotionPolicy) -> dict[str, object]:
"""Serialize one owned core motion policy without retaining planner objects."""
plan_options = policy.plan_opts
options_metadata: object = None
if plan_options is not None:
values = (
plan_options.to_dict()
if callable(getattr(plan_options, "to_dict", None))
else None
)
options_metadata = {
"type": f"{type(plan_options).__module__}.{type(plan_options).__qualname__}",
"values": _metadata_value(values),
}
return {
"strategy": policy.strategy,
"sample_count": policy.sample_count,
"velocity_limit": _metadata_value(policy.velocity_limit),
"acceleration_limit": _metadata_value(policy.acceleration_limit),
"dynamic_collision_mode": policy.dynamic_collision_mode.value,
"plan_options": options_metadata,
}
def _recovery_policy_to_metadata(policy: RecoveryPolicy) -> dict[str, object]:
"""Serialize all bounded-recovery settings."""
return {
"max_replans": policy.max_replans,
"max_action_retries": policy.max_action_retries,
"goal_translation_threshold": _metadata_value(
policy.goal_translation_threshold
),
"goal_rotation_threshold": _metadata_value(policy.goal_rotation_threshold),
"action_timeout": _metadata_value(policy.action_timeout),
}
def _tracking_metric_to_metadata(metric: TrackingMetricCfg) -> dict[str, object]:
"""Serialize one exact typed metric and its unit-preserving tolerances."""
parameters = (
{
value.name: _metadata_value(getattr(metric, value.name))
for value in fields(metric)
}
if is_dataclass(metric)
else {}
)
return {
"metric_id": metric.metric_id,
"revision": metric.revision,
"channel_id": metric.channel_id,
"type": f"{type(metric).__module__}.{type(metric).__qualname__}",
"parameters": parameters,
}
def _tracking_policy_to_metadata(policy: TrackingPolicy) -> dict[str, object]:
"""Serialize independent in-flight and terminal tracking contracts."""
in_flight = policy.in_flight
terminal = policy.terminal
return {
"in_flight": (
None
if in_flight is None
else {
"metrics": [
_tracking_metric_to_metadata(metric) for metric in in_flight.metrics
],
"consecutive_violations": in_flight.consecutive_violations,
"grace_period": _metadata_value(in_flight.grace_period),
}
),
"terminal": (
{
"mode": "feedback",
"metrics": [
_tracking_metric_to_metadata(metric) for metric in terminal.metrics
],
"settle_timeout": _metadata_value(terminal.settle_timeout),
"consecutive_acceptances": terminal.consecutive_acceptances,
}
if isinstance(terminal, FeedbackTerminalAcceptance)
else {
"mode": "timed",
"settle_duration": _metadata_value(terminal.settle_duration),
}
),
}
def _tracking_sequence_to_metadata(
sequence: TimedTrackingSequence | None,
) -> dict[str, object] | None:
"""Serialize the provider/projector shape of one plan-owned contract."""
if sequence is None:
return None
first_frame = None if not sequence.frames else sequence.frames[0]
return {
"env_ids": _metadata_value(sequence.env_ids),
"frame_count": sequence.frame_count,
"setpoints": [
{
"endpoint": list(setpoint.endpoint_key),
"channel_id": setpoint.binding.channel_id,
"state_type": (
f"{type(setpoint.desired).__module__}."
f"{type(setpoint.desired).__qualname__}"
),
"feedback_source": {
"provider_id": setpoint.binding.source.provider_id,
"revision": setpoint.binding.source.revision,
"address_fingerprint": _metadata_value(
setpoint.binding.source.address.address_fingerprint
),
},
"projector": {
"projector_id": setpoint.binding.projector.projector_id,
"revision": setpoint.binding.projector.revision,
},
"route_fingerprint": _metadata_value(
setpoint.binding.route_fingerprint
),
}
for setpoint in (() if first_frame is None else first_frame.setpoints)
],
}
@dataclass(frozen=True, slots=True)
class ResolvedCorePolicyTrace:
"""Resolved preset, core policies, and execution binding for one plan."""
profile_id: str
preset_id: str
motion_policy: MotionPolicy
tracking_policy: TrackingPolicy
recovery_policy: RecoveryPolicy
endpoints: tuple[SkillEndpointBindingTrace, ...]
def __post_init__(self) -> None:
for name in ("profile_id", "preset_id"):
value = getattr(self, name)
if type(value) is not str or not value:
raise ValueError(f"{name} must be a non-empty string.")
if not isinstance(self.motion_policy, MotionPolicy):
raise TypeError("motion_policy must be a MotionPolicy.")
if not isinstance(self.tracking_policy, TrackingPolicy):
raise TypeError("tracking_policy must be a TrackingPolicy.")
if not isinstance(self.recovery_policy, RecoveryPolicy):
raise TypeError("recovery_policy must be a RecoveryPolicy.")
endpoints = tuple(self.endpoints)
if not all(type(value) is SkillEndpointBindingTrace for value in endpoints):
raise TypeError(
"endpoints must contain exact SkillEndpointBindingTrace values."
)
keys = tuple((value.slot_id, value.endpoint_id) for value in endpoints)
if len(set(keys)) != len(keys):
raise ValueError("endpoints must use unique slot/endpoint keys.")
object.__setattr__(self, "motion_policy", replace(self.motion_policy))
object.__setattr__(self, "tracking_policy", self.tracking_policy.snapshot())
object.__setattr__(self, "recovery_policy", replace(self.recovery_policy))
object.__setattr__(self, "endpoints", endpoints)
@classmethod
def from_resolved_binding(
cls,
*,
profile_id: str,
preset_id: str,
motion_policy: MotionPolicy,
tracking_policy: TrackingPolicy,
recovery_policy: RecoveryPolicy,
endpoints: Iterable[EndpointBinding],
) -> ResolvedCorePolicyTrace:
"""Project one resolved preset and action binding to a trace."""
return cls(
profile_id=profile_id,
preset_id=preset_id,
motion_policy=motion_policy,
tracking_policy=tracking_policy,
recovery_policy=recovery_policy,
endpoints=tuple(
SkillEndpointBindingTrace.from_binding(endpoint)
for endpoint in endpoints
),
)
def snapshot(self) -> ResolvedCorePolicyTrace:
"""Return an independently owned core-policy and binding trace."""
return ResolvedCorePolicyTrace(
profile_id=self.profile_id,
preset_id=self.preset_id,
motion_policy=self.motion_policy,
tracking_policy=self.tracking_policy,
recovery_policy=self.recovery_policy,
endpoints=self.endpoints,
)
def to_metadata(self) -> dict[str, object]:
"""Return deterministic policy and endpoint-binding metadata."""
return {
"profile_id": self.profile_id,
"preset": {"preset_id": self.preset_id},
"motion_policy": _motion_policy_to_metadata(self.motion_policy),
"tracking_policy": _tracking_policy_to_metadata(self.tracking_policy),
"recovery_policy": _recovery_policy_to_metadata(self.recovery_policy),
"endpoints": [endpoint.to_metadata() for endpoint in self.endpoints],
}
@dataclass(frozen=True, slots=True, eq=False)
class SkillPlanAttemptTrace:
"""Compact, typed trace of one installed action-plan generation.
``scene_dependency_monitor_until`` preserves the plan's per-entity exclusive
waypoint cutoff: an entity is monitored only while the current waypoint index
is smaller than its configured value.
"""
attempt_generation: int
trigger: str
planned_at: float
invocation_index: int
planned_mask: torch.Tensor
action_retry_counts: tuple[int, ...]
replan_counts: tuple[int, ...]
skill_id: str
invocation_id: str | None
invocation_revision: int
plan_success_mask: torch.Tensor
command_frame_count: int
trajectory_segments: tuple[TrajectorySegment, ...]
planned_scene_version: int
planned_collision_world_revision: tuple[int, ...]
scene_dependencies: tuple[str, ...]
scene_dependency_monitor_until: Mapping[str, int]
collision_world_sensitive: bool
replannable: bool
tracking_policy: TrackingPolicy
tracking: TimedTrackingSequence | None
effect_verification_kind: str | None
resolved_core_policy: ResolvedCorePolicyTrace
planner_backend: str
planner_messages: tuple[str, ...]
planner_metadata: Mapping[str, object]
planner_failure: PlanningFailure | None
def __post_init__(self) -> None:
if type(self.attempt_generation) is not int or self.attempt_generation < 0:
raise ValueError("attempt_generation must be non-negative.")
if type(self.trigger) is not str or not self.trigger:
raise ValueError("trigger must be a non-empty string.")
if not math.isfinite(self.planned_at) or self.planned_at < 0.0:
raise ValueError("planned_at must be finite and non-negative.")
if type(self.invocation_index) is not int or self.invocation_index < 0:
raise ValueError("invocation_index must be non-negative.")
for name in ("planned_mask", "plan_success_mask"):
value = getattr(self, name)
if (
not isinstance(value, torch.Tensor)
or value.dtype != torch.bool
or value.dim() != 1
):
raise ValueError(f"{name} must be a one-dimensional bool tensor.")
if self.planned_mask.shape != self.plan_success_mask.shape:
raise ValueError("Plan-attempt masks must have equal shapes.")
if self.planned_mask.device != self.plan_success_mask.device:
raise ValueError("Plan-attempt masks must share a device.")
batch_size = int(self.planned_mask.numel())
retries = tuple(self.action_retry_counts)
replans = tuple(self.replan_counts)
if len(retries) != batch_size or len(replans) != batch_size:
raise ValueError("Recovery counters must contain one value per row.")
if any(type(value) is not int or value < 0 for value in (*retries, *replans)):
raise ValueError("Recovery counters must be non-negative integers.")
if type(self.skill_id) is not str or not self.skill_id:
raise ValueError("skill_id must be a non-empty string.")
if self.invocation_id is not None and (
type(self.invocation_id) is not str or not self.invocation_id
):
raise ValueError("invocation_id must be a non-empty string or None.")
if type(self.invocation_revision) is not int or self.invocation_revision < 0:
raise ValueError("invocation_revision must be non-negative.")
if type(self.command_frame_count) is not int or self.command_frame_count < 0:
raise ValueError("command_frame_count must be non-negative.")
segments = tuple(self.trajectory_segments)
if not all(type(value) is TrajectorySegment for value in segments):
raise TypeError(
"trajectory_segments must contain TrajectorySegment values."
)
if (
type(self.planned_scene_version) is not int
or self.planned_scene_version < 0
):
raise ValueError("planned_scene_version must be non-negative.")
collision_revisions = tuple(self.planned_collision_world_revision)
if len(collision_revisions) != batch_size or any(
type(value) is not int or value < 0 for value in collision_revisions
):
raise ValueError(
"planned_collision_world_revision must contain one non-negative "
"integer per row."
)
dependencies = tuple(self.scene_dependencies)
if len(set(dependencies)) != len(dependencies) or not all(
type(value) is str and value for value in dependencies
):
raise ValueError("scene_dependencies must contain unique identifiers.")
if not isinstance(self.scene_dependency_monitor_until, Mapping):
raise TypeError("scene_dependency_monitor_until must be a mapping.")
monitor_until = dict(self.scene_dependency_monitor_until)
if not set(monitor_until).issubset(dependencies):
raise ValueError(
"scene_dependency_monitor_until keys must be scene dependencies."
)
for entity_id, waypoint_index in monitor_until.items():
if (
type(entity_id) is not str
or not entity_id
or type(waypoint_index) is not int
or not 0 <= waypoint_index <= self.command_frame_count
):
raise ValueError(
"scene_dependency_monitor_until must map non-empty entity IDs "
"to waypoint indices within the command sequence."
)
if type(self.collision_world_sensitive) is not bool:
raise TypeError("collision_world_sensitive must be a bool.")
if type(self.replannable) is not bool:
raise TypeError("replannable must be a bool.")
if not isinstance(self.tracking_policy, TrackingPolicy):
raise TypeError("tracking_policy must be a TrackingPolicy.")
if self.tracking is not None and not isinstance(
self.tracking, TimedTrackingSequence
):
raise TypeError("tracking must be a TimedTrackingSequence or None.")
if self.effect_verification_kind is not None and (
type(self.effect_verification_kind) is not str
or not self.effect_verification_kind
):
raise ValueError("effect_verification_kind must be non-empty or None.")
if type(self.resolved_core_policy) is not ResolvedCorePolicyTrace:
raise TypeError(
"resolved_core_policy must be exactly ResolvedCorePolicyTrace."
)
if type(self.planner_backend) is not str or not self.planner_backend:
raise ValueError("planner_backend must be a non-empty string.")
if (
self.planner_failure is not None
and type(self.planner_failure) is not PlanningFailure
):
raise TypeError("planner_failure must be PlanningFailure or None.")
messages = tuple(self.planner_messages)
if not all(type(value) is str for value in messages):
raise TypeError("planner_messages must contain strings.")
object.__setattr__(self, "planned_mask", self.planned_mask.clone())
object.__setattr__(self, "plan_success_mask", self.plan_success_mask.clone())
object.__setattr__(self, "action_retry_counts", retries)
object.__setattr__(self, "replan_counts", replans)
object.__setattr__(self, "trajectory_segments", segments)
object.__setattr__(
self,
"planned_collision_world_revision",
collision_revisions,
)
object.__setattr__(self, "scene_dependencies", dependencies)
object.__setattr__(
self,
"scene_dependency_monitor_until",
MappingProxyType(monitor_until),
)
object.__setattr__(self, "tracking_policy", self.tracking_policy.snapshot())
if self.tracking is not None:
object.__setattr__(self, "tracking", self.tracking.snapshot())
object.__setattr__(
self,
"resolved_core_policy",
self.resolved_core_policy.snapshot(),
)
object.__setattr__(self, "planner_messages", messages)
if self.planner_failure is not None:
object.__setattr__(
self,
"planner_failure",
PlanningFailure(
self.planner_failure.code,
self.planner_failure.retryable,
),
)
object.__setattr__(
self,
"planner_metadata",
_snapshot_metadata_mapping(self.planner_metadata),
)
@classmethod
def from_execution_attempt(
cls,
attempt: ExecutionPlanAttempt,
*,
profile_id: str,
preset_id: str,
) -> SkillPlanAttemptTrace:
"""Project one session-owned plan attempt to compact trace metadata."""
if not isinstance(attempt, ExecutionPlanAttempt):
raise TypeError("attempt must be an ExecutionPlanAttempt.")
plan = attempt.plan
request = attempt.request
return cls(
attempt_generation=attempt.attempt_generation,
trigger=attempt.event_kind.value,
planned_at=attempt.planned_at,
invocation_index=attempt.invocation_index,
planned_mask=attempt.planned_mask,
action_retry_counts=attempt.action_retry_counts,
replan_counts=attempt.replan_counts,
skill_id=plan.skill_id,
invocation_id=plan.invocation_id,
invocation_revision=plan.invocation_revision,
plan_success_mask=plan.plan_success,
command_frame_count=plan.commands.frame_count,
trajectory_segments=plan.segments,
planned_scene_version=plan.planned_scene_version,
planned_collision_world_revision=plan.planned_collision_world_revision,
scene_dependencies=plan.scene_dependencies,
scene_dependency_monitor_until=plan.scene_dependency_monitor_until,
collision_world_sensitive=plan.collision_world_sensitive,
replannable=plan.replannable,
tracking_policy=plan.tracking_policy,
tracking=plan.tracking,
effect_verification_kind=(
None
if plan.effect_verification is None
else plan.effect_verification.kind
),
resolved_core_policy=ResolvedCorePolicyTrace.from_resolved_binding(
profile_id=profile_id,
preset_id=preset_id,
motion_policy=request.motion_policy,
tracking_policy=request.tracking_policy,
recovery_policy=request.recovery_policy,
endpoints=request.binding.endpoints,
),
planner_backend=plan.diagnostics.backend,
planner_messages=plan.diagnostics.messages,
planner_metadata=plan.diagnostics.metadata,
planner_failure=plan.diagnostics.failure,
)
def snapshot(self) -> SkillPlanAttemptTrace:
"""Return an independently owned compact plan-attempt trace."""
return SkillPlanAttemptTrace(
attempt_generation=self.attempt_generation,
trigger=self.trigger,
planned_at=self.planned_at,
invocation_index=self.invocation_index,
planned_mask=self.planned_mask,
action_retry_counts=self.action_retry_counts,
replan_counts=self.replan_counts,
skill_id=self.skill_id,
invocation_id=self.invocation_id,
invocation_revision=self.invocation_revision,
plan_success_mask=self.plan_success_mask,
command_frame_count=self.command_frame_count,
trajectory_segments=self.trajectory_segments,
planned_scene_version=self.planned_scene_version,
planned_collision_world_revision=self.planned_collision_world_revision,
scene_dependencies=self.scene_dependencies,
scene_dependency_monitor_until=self.scene_dependency_monitor_until,
collision_world_sensitive=self.collision_world_sensitive,
replannable=self.replannable,
tracking_policy=self.tracking_policy,
tracking=self.tracking,
effect_verification_kind=self.effect_verification_kind,
resolved_core_policy=self.resolved_core_policy,
planner_backend=self.planner_backend,
planner_messages=self.planner_messages,
planner_metadata=self.planner_metadata,
planner_failure=self.planner_failure,
)
def to_metadata(self) -> dict[str, object]:
"""Return one plan generation as deterministic JSON-safe data."""
planner_diagnostics: dict[str, object] = {
"backend": self.planner_backend,
"messages": list(self.planner_messages),
"metadata": _metadata_value(self.planner_metadata),
}
if self.planner_failure is not None:
planner_diagnostics["failure"] = {
"code": self.planner_failure.code,
"retryable": self.planner_failure.retryable,
}
return {
"attempt_generation": self.attempt_generation,
"trigger": self.trigger,
"planned_at": self.planned_at,
"invocation_index": self.invocation_index,
"planned_mask": _metadata_value(self.planned_mask),
"recovery_counters": {
"action_retries": list(self.action_retry_counts),
"replans": list(self.replan_counts),
},
"skill_id": self.skill_id,
"invocation_id": self.invocation_id,
"invocation_revision": self.invocation_revision,
"plan_success_mask": _metadata_value(self.plan_success_mask),
"command_frame_count": self.command_frame_count,
"trajectory_segments": [
{
"name": segment.name,
"start": segment.start,
"stop": segment.stop,
"waypoint_count": segment.waypoint_count,
}
for segment in self.trajectory_segments
],
"planned_scene_version": self.planned_scene_version,
"planned_collision_world_revision": list(
self.planned_collision_world_revision
),
"scene_dependencies": list(self.scene_dependencies),
"scene_dependency_monitor_until": {
entity_id: self.scene_dependency_monitor_until[entity_id]
for entity_id in sorted(self.scene_dependency_monitor_until)
},
"collision_world_sensitive": self.collision_world_sensitive,
"replannable": self.replannable,
"tracking_policy": _tracking_policy_to_metadata(self.tracking_policy),
"tracking_contract": _tracking_sequence_to_metadata(self.tracking),
"effect_verification_kind": self.effect_verification_kind,
"resolved_core_policy": self.resolved_core_policy.to_metadata(),
"planner_diagnostics": planner_diagnostics,
}
@dataclass(frozen=True, slots=True, eq=False)
class SkillEffectTrace:
"""One monitor decision correlated with an atomic verification boundary."""
call_index: int
verification_id: int
observation_revision: int
timestamp: float
success_mask: torch.Tensor
failure_mask: torch.Tensor
expectation_decisions: tuple[EffectExpectationDecision, ...]
effect_spec: SemanticEffectSpec
monitor_id: str
monitor_revision: str | None
configured_monitor_params: Mapping[str, object]
resolved_monitor_params: Mapping[str, object]
evidence: Mapping[str, EffectEvidenceBatch]
boundary_kind: str = "terminal"
guard_id: str | None = None
gate_id: str | None = None
segment_name: str | None = None
def __post_init__(self) -> None:
if type(self.call_index) is not int or self.call_index < 0:
raise ValueError("call_index must be a non-negative integer.")
if type(self.verification_id) is not int or self.verification_id < 0:
raise ValueError("verification_id must be a non-negative integer.")
if type(self.observation_revision) is not int or self.observation_revision < 0:
raise ValueError("observation_revision must be non-negative.")
if self.boundary_kind not in {
"terminal",
"in_flight_guard",
"phase_effect_gate",
}:
raise ValueError(
"boundary_kind must be 'terminal', 'in_flight_guard', or "
"'phase_effect_gate'."
)
for name in ("guard_id", "gate_id", "segment_name"):
value = getattr(self, name)
if value is not None and (type(value) is not str or not value):
raise ValueError(f"{name} must be a non-empty string or None.")
if self.boundary_kind == "terminal":
if (
self.guard_id is not None
or self.gate_id is not None
or self.segment_name is not None
):
raise ValueError(
"Terminal effect traces cannot declare segment-boundary metadata."
)
elif self.boundary_kind == "in_flight_guard" and (
self.guard_id is None
or self.gate_id is not None
or self.segment_name is None
):
raise ValueError(
"In-flight guard traces require only guard_id and segment_name."
)
elif self.boundary_kind == "phase_effect_gate" and (
self.gate_id is None
or self.guard_id is not None
or self.segment_name is None
):
raise ValueError(
"Phase-effect gate traces require only gate_id and segment_name."
)
if not math.isfinite(self.timestamp) or self.timestamp < 0.0:
raise ValueError("timestamp must be finite and non-negative.")
for name in ("success_mask", "failure_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("Effect trace masks must have equal shapes.")
if self.success_mask.device != self.failure_mask.device:
raise ValueError("Effect trace masks must share a device.")
if (self.success_mask & self.failure_mask).any():
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 (
type(self.monitor_revision) is not str or not self.monitor_revision
):
raise ValueError("monitor_revision must be non-empty or None.")
evidence_types = (
PoseRelationEvidenceBatch,
BinaryEffectEvidenceBatch,
ScalarEffectEvidenceBatch,
JointStateEvidenceBatch,
)
evidence: dict[str, EffectEvidenceBatch] = {}
for evidence_id, batch in self.evidence.items():
if type(evidence_id) is not str or not evidence_id:
raise ValueError("evidence keys must be non-empty strings.")
if type(batch) not in evidence_types:
raise TypeError("evidence values must be exact evidence batches.")
if batch.evidence_id != evidence_id:
raise ValueError("evidence keys must match batch evidence_id values.")
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,
"configured_monitor_params",
_snapshot_metadata_mapping(self.configured_monitor_params),
)
object.__setattr__(
self,
"resolved_monitor_params",
_snapshot_metadata_mapping(self.resolved_monitor_params),
)
object.__setattr__(self, "evidence", MappingProxyType(evidence))
def snapshot(self) -> SkillEffectTrace:
"""Return an independently owned trace."""
return SkillEffectTrace(
call_index=self.call_index,
verification_id=self.verification_id,
observation_revision=self.observation_revision,
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,
configured_monitor_params=self.configured_monitor_params,
resolved_monitor_params=self.resolved_monitor_params,
evidence=self.evidence,
boundary_kind=self.boundary_kind,
guard_id=self.guard_id,
gate_id=self.gate_id,
segment_name=self.segment_name,
)
def to_metadata(self) -> dict[str, object]:
"""Return monitor contract, evidence, thresholds, and decision metadata."""
metadata = {
"call_index": self.call_index,
"verification_id": self.verification_id,
"observation_revision": self.observation_revision,
"timestamp": self.timestamp,
"effect_spec": self.effect_spec.to_metadata(),
"monitor": {
"monitor_id": self.monitor_id,
"revision": self.monitor_revision,
"configured_params": _metadata_value(self.configured_monitor_params),
"resolved_params": _metadata_value(self.resolved_monitor_params),
},
"evidence": {
evidence_id: batch.to_metadata()
for evidence_id, batch in sorted(self.evidence.items())
},
"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}
if self.boundary_kind == "in_flight_guard":
metadata["boundary"].update(
{
"guard_id": self.guard_id,
"segment_name": self.segment_name,
}
)
elif self.boundary_kind == "phase_effect_gate":
metadata["boundary"].update(
{
"gate_id": self.gate_id,
"segment_name": self.segment_name,
}
)
return metadata
@dataclass(frozen=True, slots=True, eq=False)
class SkillFailure:
"""Per-environment semantic workflow failure."""
call_index: int
semantic_id: str
env_mask: torch.Tensor
message: str
code: str = "semantic_call_failed"
phase: str = "execution"
diagnostic: SemanticDiagnostic | None = None
def __post_init__(self) -> None:
if type(self.call_index) is not int or self.call_index < 0:
raise ValueError("call_index must be a non-negative integer.")
if type(self.semantic_id) is not str or not self.semantic_id:
raise ValueError("semantic_id must be a non-empty string.")
if not isinstance(self.env_mask, torch.Tensor):
raise TypeError("env_mask must be a torch.Tensor.")
if self.env_mask.dtype != torch.bool or self.env_mask.dim() != 1:
raise ValueError("env_mask must be a one-dimensional bool tensor.")
if type(self.message) is not str or not self.message:
raise ValueError("message must be a non-empty string.")
if (
type(self.code) is not str
or not self.code
or self.code != self.code.strip()
):
raise ValueError(
"code must be a non-empty string without outer whitespace."
)
if type(self.phase) is not str:
raise TypeError("phase must be a string.")
if self.phase not in {"preparation", "execution", "runtime"}:
raise ValueError(
"phase must be one of 'preparation', 'execution', or 'runtime'."
)
if self.diagnostic is not None and not isinstance(
self.diagnostic,
SemanticDiagnostic,
):
raise TypeError("diagnostic must be a SemanticDiagnostic or None.")
object.__setattr__(self, "env_mask", self.env_mask.clone())
def snapshot(self) -> SkillFailure:
"""Return an independently owned failure."""
return SkillFailure(
call_index=self.call_index,
semantic_id=self.semantic_id,
env_mask=self.env_mask,
message=self.message,
code=self.code,
phase=self.phase,
diagnostic=self.diagnostic,
)
def to_metadata(self) -> dict[str, object]:
"""Return one row-local failure as JSON-safe data."""
return {
"call_index": self.call_index,
"semantic_id": self.semantic_id,
"env_mask": _metadata_value(self.env_mask),
"message": self.message,
"code": self.code,
"phase": self.phase,
"diagnostic": (
None
if self.diagnostic is None
else {
"code": self.diagnostic.code,
"path": _metadata_value(self.diagnostic.path),
"rendered_path": self.diagnostic.rendered_path,
"message": self.diagnostic.message,
"candidates": _metadata_value(self.diagnostic.candidates),
}
),
}
@dataclass(frozen=True, slots=True, eq=False)
class SkillCallTrace:
"""Terminal trace for exactly one semantic call and execution session."""
call_index: int
semantic_id: str
call_metadata: Mapping[str, object]
skill_id: str
invocation_id: str | None
invocation_revision: int
status: RunnerStatus
entered_mask: torch.Tensor
completed_mask: torch.Tensor
failed_mask: torch.Tensor
command_count: int
resolved_core_policy: ResolvedCorePolicyTrace
plan_attempts: tuple[SkillPlanAttemptTrace, ...]
events: tuple[ExecutionEvent, ...] = ()
effects: tuple[SkillEffectTrace, ...] = ()
def __post_init__(self) -> None:
if type(self.call_index) is not int or self.call_index < 0:
raise ValueError("call_index must be a non-negative integer.")
for name in ("semantic_id", "skill_id"):
value = getattr(self, name)
if type(value) is not str or not value:
raise ValueError(f"{name} must be a non-empty string.")
normalized_call = _snapshot_metadata_mapping(self.call_metadata)
if normalized_call.get("semantic_id") != self.semantic_id:
raise ValueError("call_metadata semantic_id must match semantic_id.")
if self.invocation_id is not None and (
type(self.invocation_id) is not str or not self.invocation_id
):
raise ValueError("invocation_id must be a non-empty string or None.")
if self.invocation_revision < 0:
raise ValueError("invocation_revision must be non-negative.")
if not isinstance(self.status, RunnerStatus):
raise TypeError("status must be a RunnerStatus.")
if self.status is RunnerStatus.RUNNING:
raise ValueError("A terminal call trace cannot have running status.")
for name in ("entered_mask", "completed_mask", "failed_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 not (
self.entered_mask.shape
== self.completed_mask.shape
== self.failed_mask.shape
):
raise ValueError("Call trace masks must have equal shapes.")
if not (
self.entered_mask.device
== self.completed_mask.device
== self.failed_mask.device
):
raise ValueError("Call trace masks must share a device.")
if (self.completed_mask & ~self.entered_mask).any():
raise ValueError("completed_mask must be a subset of entered_mask.")
if (self.failed_mask & ~self.entered_mask).any():
raise ValueError("failed_mask must be a subset of entered_mask.")
if (self.completed_mask & self.failed_mask).any():
raise ValueError("completed_mask and failed_mask must not overlap.")
if type(self.command_count) is not int or self.command_count < 0:
raise ValueError("command_count must be non-negative.")
if type(self.resolved_core_policy) is not ResolvedCorePolicyTrace:
raise TypeError(
"resolved_core_policy must be exactly ResolvedCorePolicyTrace."
)
attempts = tuple(self.plan_attempts)
if not all(type(attempt) is SkillPlanAttemptTrace for attempt in attempts):
raise TypeError("plan_attempts must contain SkillPlanAttemptTrace values.")
if attempts:
generations = tuple(attempt.attempt_generation for attempt in attempts)
if generations != tuple(
range(generations[0], generations[0] + len(attempts))
):
raise ValueError(
"plan_attempts must use contiguous ordered generations."
)
if attempts[-1].skill_id != self.skill_id:
raise ValueError("The active plan-attempt skill must match skill_id.")
elif self.status is not RunnerStatus.FAILED or self.command_count != 0:
raise ValueError(
"Only a preparation failure with no commands may omit plan_attempts."
)
object.__setattr__(self, "entered_mask", self.entered_mask.clone())
object.__setattr__(self, "completed_mask", self.completed_mask.clone())
object.__setattr__(self, "failed_mask", self.failed_mask.clone())
object.__setattr__(self, "call_metadata", normalized_call)
object.__setattr__(
self,
"resolved_core_policy",
self.resolved_core_policy.snapshot(),
)
object.__setattr__(
self,
"plan_attempts",
tuple(attempt.snapshot() for attempt in attempts),
)
object.__setattr__(
self,
"events",
tuple(_snapshot_event(event) for event in self.events),
)
object.__setattr__(
self,
"effects",
tuple(effect.snapshot() for effect in self.effects),
)
def snapshot(self) -> SkillCallTrace:
"""Return an independently owned call trace."""
return SkillCallTrace(
call_index=self.call_index,
semantic_id=self.semantic_id,
call_metadata=self.call_metadata,
skill_id=self.skill_id,
invocation_id=self.invocation_id,
invocation_revision=self.invocation_revision,
status=self.status,
entered_mask=self.entered_mask,
completed_mask=self.completed_mask,
failed_mask=self.failed_mask,
command_count=self.command_count,
resolved_core_policy=self.resolved_core_policy,
plan_attempts=self.plan_attempts,
events=self.events,
effects=self.effects,
)
@property
def active_plan(self) -> SkillPlanAttemptTrace:
"""Return the final installed plan generation as an owned trace."""
if not self.plan_attempts:
raise RuntimeError("This call failed before an action plan was installed.")
return self.plan_attempts[-1].snapshot()
def to_metadata(self) -> dict[str, object]:
"""Return one semantic call, recovery history, and effects as JSON-safe data."""
attempts = [attempt.to_metadata() for attempt in self.plan_attempts]
return {
"call_index": self.call_index,
"semantic_id": self.semantic_id,
"call": _metadata_value(self.call_metadata),
"skill_id": self.skill_id,
"invocation_id": self.invocation_id,
"invocation_revision": self.invocation_revision,
"status": self.status.value,
"masks": {
"entered": _metadata_value(self.entered_mask),
"completed": _metadata_value(self.completed_mask),
"failed": _metadata_value(self.failed_mask),
},
"command_count": self.command_count,
"active_plan_attempt_generation": (
None
if not self.plan_attempts
else self.plan_attempts[-1].attempt_generation
),
"resolved_core_policy": self.resolved_core_policy.to_metadata(),
"plan_attempts": attempts,
"events": [_event_to_metadata(event) for event in self.events],
"effects": [effect.to_metadata() for effect in self.effects],
}
@dataclass(frozen=True, slots=True, eq=False)
class SkillWorkflowRecoveryTrace:
"""One auditable real semantic call within bounded workflow recovery.
Args:
recovery_id: Monotonic runtime-local trace identifier.
trigger_call_index: Original workflow call held at the shared barrier.
trigger_semantic_id: Semantic ID of the original failed call.
attempt_index: One-based per-row recovery-cycle index.
max_recovery_attempts: Configured per-row recovery-cycle budget.
role: Whether this call retries, re-acquires, or retries after pickup.
source_resource_id: Resolved robot resource that owns the source object.
source_task_state_key: Verified held-object state key for that resource.
entered_mask: Rows that entered this real recovery call.
completed_mask: Entered rows that completed this call.
failed_mask: Entered rows that failed this call.
call: Nested semantic-call trace, or ``None`` when preparation failed.
message: Optional terminal or preparation diagnostic.
"""
recovery_id: int
trigger_call_index: int
trigger_semantic_id: str
attempt_index: int
max_recovery_attempts: int
role: SkillWorkflowRecoveryRole
source_resource_id: str
source_task_state_key: str
entered_mask: torch.Tensor
completed_mask: torch.Tensor
failed_mask: torch.Tensor
call: SkillCallTrace | None
message: str | None = None
def __post_init__(self) -> None:
if type(self.recovery_id) is not int or self.recovery_id < 0:
raise ValueError("recovery_id must be a non-negative integer.")
if type(self.trigger_call_index) is not int or self.trigger_call_index < 0:
raise ValueError("trigger_call_index must be non-negative.")
for name in (
"trigger_semantic_id",
"source_resource_id",
"source_task_state_key",
):
value = getattr(self, name)
if type(value) is not str or not value:
raise ValueError(f"{name} must be a non-empty string.")
if type(self.attempt_index) is not int or self.attempt_index <= 0:
raise ValueError("attempt_index must be a positive integer.")
if (
type(self.max_recovery_attempts) is not int
or self.max_recovery_attempts <= 0
or self.attempt_index > self.max_recovery_attempts
):
raise ValueError(
"max_recovery_attempts must cover the positive attempt_index."
)
if not isinstance(self.role, SkillWorkflowRecoveryRole):
raise TypeError("role must be a SkillWorkflowRecoveryRole.")
for name in ("entered_mask", "completed_mask", "failed_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 not (
self.entered_mask.shape
== self.completed_mask.shape
== self.failed_mask.shape
) or not (
self.entered_mask.device
== self.completed_mask.device
== self.failed_mask.device
):
raise ValueError("Workflow-recovery masks must share shape and device.")
if (self.completed_mask & self.failed_mask).any():
raise ValueError("Recovery completion and failure masks cannot overlap.")
if ((self.completed_mask | self.failed_mask) & ~self.entered_mask).any():
raise ValueError("Recovery outcomes must be subsets of entered_mask.")
if self.call is not None:
if type(self.call) is not SkillCallTrace:
raise TypeError("call must be a SkillCallTrace or None.")
if not torch.equal(self.call.entered_mask, self.entered_mask):
raise ValueError("Recovery call entered_mask must match the trace.")
if not torch.equal(self.call.completed_mask, self.completed_mask):
raise ValueError("Recovery call completed_mask must match the trace.")
if not torch.equal(self.call.failed_mask, self.failed_mask):
raise ValueError("Recovery call failed_mask must match the trace.")
elif not torch.equal(self.failed_mask, self.entered_mask):
raise ValueError(
"A recovery preparation failure must fail every entered row."
)
if self.message is not None and (
type(self.message) is not str or not self.message
):
raise ValueError("message must be a non-empty string or None.")
for name in ("entered_mask", "completed_mask", "failed_mask"):
object.__setattr__(self, name, getattr(self, name).clone())
if self.call is not None:
object.__setattr__(self, "call", self.call.snapshot())
def snapshot(self) -> SkillWorkflowRecoveryTrace:
"""Return an independently owned workflow-recovery trace."""
return SkillWorkflowRecoveryTrace(
recovery_id=self.recovery_id,
trigger_call_index=self.trigger_call_index,
trigger_semantic_id=self.trigger_semantic_id,
attempt_index=self.attempt_index,
max_recovery_attempts=self.max_recovery_attempts,
role=self.role,
source_resource_id=self.source_resource_id,
source_task_state_key=self.source_task_state_key,
entered_mask=self.entered_mask,
completed_mask=self.completed_mask,
failed_mask=self.failed_mask,
call=self.call,
message=self.message,
)
def to_metadata(self) -> dict[str, object]:
"""Return deterministic JSON-safe workflow-recovery metadata."""
return {
"recovery_id": self.recovery_id,
"trigger_call_index": self.trigger_call_index,
"trigger_semantic_id": self.trigger_semantic_id,
"attempt_index": self.attempt_index,
"max_recovery_attempts": self.max_recovery_attempts,
"role": self.role.value,
"source_resource_id": self.source_resource_id,
"source_task_state_key": self.source_task_state_key,
"masks": {
"entered": _metadata_value(self.entered_mask),
"completed": _metadata_value(self.completed_mask),
"failed": _metadata_value(self.failed_mask),
},
"call": None if self.call is None else self.call.to_metadata(),
"message": self.message,
}
_Trace = TypeVar("_Trace")
_HISTORY_FIELDS = frozenset(
{"events", "calls", "effects", "workflow_recoveries", "failures"}
)
@dataclass(frozen=True, slots=True)
class _HistoryEntry(Generic[_Trace]):
value: _Trace
previous: _HistoryEntry[_Trace] | None
length: int
class _HistoryJournal(Sequence[_Trace]):
"""Append-only owned entries linked only toward the historical prefix."""
def __init__(self) -> None:
self._tail: _HistoryEntry[_Trace] | None = None
def __len__(self) -> int:
return 0 if self._tail is None else self._tail.length
def __iter__(self) -> Iterator[_Trace]:
values: list[_Trace] = []
entry = self._tail
while entry is not None:
values.append(entry.value)
entry = entry.previous
return reversed(values)
def __getitem__(self, index: int | slice) -> _Trace | tuple[_Trace, ...]:
if isinstance(index, slice):
start, stop, step = index.indices(len(self))
if step != 1:
return tuple(self)[index]
values: list[_Trace] = []
entry = self._tail
while entry is not None and entry.length > start:
if entry.length <= stop:
values.append(entry.value)
entry = entry.previous
return tuple(reversed(values))
if index < 0:
index += len(self)
if not 0 <= index < len(self):
raise IndexError(index)
entry = self._tail
while entry.length != index + 1:
entry = entry.previous
return entry.value
def append(self, value: _Trace) -> None:
self._tail = _HistoryEntry(value, self._tail, len(self) + 1)
def extend(self, values: Iterable[_Trace]) -> None:
for value in values:
self.append(value)
def snapshot(
self, copy_entry: Callable[[_Trace], _Trace]
) -> _HistorySnapshot[_Trace]:
return _HistorySnapshot(self._tail, copy_entry)
@dataclass(frozen=True, slots=True)
class _HistorySnapshot(Generic[_Trace]):
"""Capture an owned immutable journal prefix without copying its history.
Only the runtime constructs these views. Journal entries already own their
inputs and are never changed after append. Old tails cannot retain future
entries, including after the runtime starts a new workflow.
Public access materializes independent copies, preserving result ownership.
"""
tail: _HistoryEntry[_Trace] | None
copy_entry: Callable[[_Trace], _Trace]
def materialize(self) -> tuple[_Trace, ...]:
values: list[_Trace] = []
entry = self.tail
while entry is not None:
values.append(self.copy_entry(entry.value))
entry = entry.previous
return tuple(reversed(values))
[docs]
@dataclass(frozen=True, slots=True, eq=False)
class SemanticExecutionResult:
"""Immutable workflow snapshot returned by sync and step-wise execution."""
status: SemanticExecutionStatus
workflow_id: str | None
current_call_index: int | None
env_ids: torch.Tensor
success_mask: torch.Tensor
failure_mask: torch.Tensor
cancelled_mask: torch.Tensor
eligible_mask: torch.Tensor
task_state: TaskState
events: tuple[ExecutionEvent, ...] = ()
calls: tuple[SkillCallTrace, ...] = ()
effects: tuple[SkillEffectTrace, ...] = ()
workflow_recoveries: tuple[SkillWorkflowRecoveryTrace, ...] = ()
failures: tuple[SkillFailure, ...] = ()
wait_duration: float = 0.0
message: str | None = None
def __getattribute__(self, name: str) -> object:
value = object.__getattribute__(self, name)
if name in _HISTORY_FIELDS and type(value) is _HistorySnapshot:
value = value.materialize()
object.__setattr__(self, name, value)
return value
def __post_init__(self) -> None:
if not isinstance(self.status, SemanticExecutionStatus):
raise TypeError("status must be a SemanticExecutionStatus.")
if self.workflow_id is not None and (
type(self.workflow_id) is not str or not self.workflow_id
):
raise ValueError("workflow_id must be a non-empty string or None.")
if self.current_call_index is not None and (
type(self.current_call_index) is not int or self.current_call_index < 0
):
raise ValueError("current_call_index must be non-negative or None.")
if not isinstance(self.env_ids, torch.Tensor):
raise TypeError("env_ids must be a torch.Tensor.")
if self.env_ids.dtype != torch.long or self.env_ids.dim() != 1:
raise ValueError("env_ids must be a one-dimensional torch.long tensor.")
if self.env_ids.numel() == 0:
raise ValueError("env_ids must contain at least one environment.")
if torch.unique(self.env_ids).numel() != self.env_ids.numel():
raise ValueError("env_ids must be unique.")
batch_size = int(self.env_ids.numel())
for name in (
"success_mask",
"failure_mask",
"cancelled_mask",
"eligible_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.shape != (batch_size,):
raise ValueError(f"{name} must be bool with shape ({batch_size},).")
if value.device != self.env_ids.device:
raise ValueError(f"{name} and env_ids must share a device.")
if (self.success_mask & (self.failure_mask | self.cancelled_mask)).any():
raise ValueError("Successful rows cannot also fail or be cancelled.")
if (self.failure_mask & self.cancelled_mask).any():
raise ValueError("Failed and cancelled masks must not overlap.")
if (self.eligible_mask & (self.failure_mask | self.cancelled_mask)).any():
raise ValueError("Eligible rows cannot also fail or be cancelled.")
if (self.success_mask & ~self.eligible_mask).any():
raise ValueError("success_mask must be a subset of eligible_mask.")
if not isinstance(self.task_state, TaskState):
raise TypeError("task_state must be a TaskState.")
if self.task_state.batch_size != batch_size:
raise ValueError("task_state batch size must match env_ids.")
if self.task_state.device != self.env_ids.device:
raise ValueError("task_state and env_ids must share a device.")
if not math.isfinite(self.wait_duration) or self.wait_duration < 0.0:
raise ValueError("wait_duration must be finite and non-negative.")
if self.message is not None and type(self.message) is not str:
raise TypeError("message must be a string or None.")
recoveries = object.__getattribute__(self, "workflow_recoveries")
if type(recoveries) is not _HistorySnapshot and not all(
type(recovery) is SkillWorkflowRecoveryTrace for recovery in recoveries
):
raise TypeError(
"workflow_recoveries must contain SkillWorkflowRecoveryTrace values."
)
object.__setattr__(self, "env_ids", self.env_ids.clone())
for name in (
"success_mask",
"failure_mask",
"cancelled_mask",
"eligible_mask",
):
object.__setattr__(self, name, getattr(self, name).clone())
object.__setattr__(self, "task_state", _snapshot_task_state(self.task_state))
for name, copy_entry in (
("events", _snapshot_event),
("calls", SkillCallTrace.snapshot),
("effects", SkillEffectTrace.snapshot),
("workflow_recoveries", SkillWorkflowRecoveryTrace.snapshot),
("failures", SkillFailure.snapshot),
):
history = object.__getattribute__(self, name)
if type(history) is not _HistorySnapshot:
object.__setattr__(
self, name, tuple(copy_entry(entry) for entry in history)
)
@property
def terminal(self) -> bool:
"""Whether the workflow no longer accepts execution steps."""
return self.status in {
SemanticExecutionStatus.COMPLETED,
SemanticExecutionStatus.FAILED,
SemanticExecutionStatus.CANCELLED,
}
[docs]
def require_all_succeeded(self) -> None:
"""Raise unless every environment completed the workflow successfully.
This convenience assertion keeps application entry points concise while
the canonical result continues to expose row-local masks.
"""
if self.status is not SemanticExecutionStatus.COMPLETED or not bool(
self.success_mask.all()
):
raise RuntimeError(
"Semantic workflow did not succeed for every environment: "
f"status={self.status.value}, "
f"failed={self.failure_mask.detach().cpu().tolist()}, "
f"cancelled={self.cancelled_mask.detach().cpu().tolist()}."
)
__all__: list[str] = []