embodichain.lab.task_program.integrations

Explicit provider, environment, and simulation Task Program integrations.

AntipodalGraspAffordanceBinding	Build one antipodal grasp affordance from a selected rigid-object mesh.
ConfiguredHandOverPoseProvider	Resolve a hand-over delivery target from immutable configuration.
ContainerAffordanceBinding	Declare one exact object target frame inside a container parent.
ControlPartCommandPreset	Named one-dimensional joint commands for one exact control part.
ControlPartEvidenceProviderDeclaration	Provider-free identity of one control-part evidence factory.
ControlPartEvidenceProviderFactory	Registration-owned factory for one live control-part evidence provider.
ControlPartEndpointBinding	Profile endpoint backed by one explicit robot control part.
ControlPartResourceBinding	Joint-backed robot resource containing control-part endpoints.
CuroboParallelCommandSafetyValidator	Validate the exact synchronized joint segment before transport dispatch.
CuroboParallelSafetyValidatorFactory	Create exact-sample collision gates for one aggregate control part.
EndpointAdapterDeclaration	Provider-free declaration of one exact endpoint adapter.
TaskProgramAdapterFactory	Bind one initialized environment to its Task Program adapter.
TaskProgramEnvironmentAdapter	Compile and run Task Programs through explicit environment factories.
TaskProgramEnvironmentFactory	Environment-owned factories for one explicit semantic integration.
TaskProgramIntegrationCatalog	Provider-free integration directory owned by one task registration.
TaskProgramRuntimeAssembly	Auditable result of one fresh environment runtime assembly.
IntegrationFingerprintMismatch	Raised when a live integration no longer matches its registration.
ParallelCommandSafetyValidatorFactory	Registration-owned factory for one authoritative live safety gate.
ParallelSafetyDeclaration	Provider-free identity and transport coverage of one safety factory.
PlanningObservationPort	Combined observation and full-qpos port required by the Gym runtime.
RegisteredSemanticLowererDeclaration	Provider-free identity of one registered semantic lowerer factory.
RegisteredSemanticLowererFactory	Registration-owned factory for one fresh live semantic lowerer.
RuntimeTransportDeclaration	Provider-free declaration of one ordered runtime transport encoder.
SimulationArticulationBinding	Explicit binding for one simulation articulation.
SimulationArticulationLinkBinding	Explicit canonical link backed by one native articulation link.
SimulationTaskProgramAdapterFactory	Create one standard simulation adapter after an environment is initialized.
SimulationTaskProgramFactory	Build every live Task Program component from one task registration.
SimulationTaskProgramRegistration	Exact immutable task-owned simulation integration registration.
SimulationRigidObjectBinding	Explicit binding for one simulation rigid object.
SimulationRobotSkillProfileBinding	Build a profile from typed resources with strict native validation.
SimulationSceneBinding	Build one authoritative registry from explicit simulation bindings.
SimulationSegmentPolicyPort	Execute built-in segment policies against explicitly bound simulation data.
StandardExtensionDeclarations	Cross-checked provider-free declarations for the standard factory.
SupportSurfaceAffordanceBinding	Declare one exact object target frame on a support parent.
VersionedKey	Built-in immutable sequence.
create_simulation_task_program_adapter(...)	Create a complete production adapter from one standard Gym environment.
default_simulation_settle_presets()	Return independently owned built-in post-policy presets.

Classes:

AntipodalGraspAffordanceBinding	Build one antipodal grasp affordance from a selected rigid-object mesh.
ConfiguredHandOverPoseProvider	Resolve a hand-over delivery target from immutable configuration.
ContainerAffordanceBinding	Declare one exact object target frame inside a container parent.
ControlPartCommandPreset	Named one-dimensional joint commands for one exact control part.
ControlPartEndpointBinding	Profile endpoint backed by one explicit robot control part.
ControlPartEvidenceProviderDeclaration	Provider-free identity of one control-part evidence factory.
ControlPartEvidenceProviderFactory	Registration-owned factory for one live control-part evidence provider.
ControlPartResourceBinding	Joint-backed robot resource containing control-part endpoints.
CuroboParallelCommandSafetyValidator	Validate the exact synchronized joint segment before transport dispatch.
CuroboParallelSafetyValidatorFactory	Create exact-sample collision gates for one aggregate control part.
EndpointAdapterDeclaration	Provider-free declaration of one exact endpoint adapter.
ParallelCommandSafetyValidatorFactory	Registration-owned factory for one authoritative live safety gate.
ParallelSafetyDeclaration	Provider-free identity and transport coverage of one safety factory.
PlanningObservationPort	Combined observation and full-qpos port required by the Gym runtime.
RegisteredSemanticLowererDeclaration	Provider-free identity of one registered semantic lowerer factory.
RegisteredSemanticLowererFactory	Registration-owned factory for one fresh live semantic lowerer.
RuntimeTransportDeclaration	Provider-free declaration of one ordered runtime transport encoder.
SimulationArticulationBinding	Explicit binding for one simulation articulation.
SimulationArticulationLinkBinding	Explicit canonical link backed by one native articulation link.
SimulationRigidObjectBinding	Explicit binding for one simulation rigid object.
SimulationRobotSkillProfileBinding	Build a profile from typed resources with strict native validation.
SimulationSceneBinding	Build one authoritative registry from explicit simulation bindings.
SimulationSegmentPolicyPort	Execute built-in segment policies against explicitly bound simulation data.
SimulationTaskProgramAdapterFactory	Create one standard simulation adapter after an environment is initialized.
SimulationTaskProgramFactory	Build every live Task Program component from one task registration.
SimulationTaskProgramRegistration	Exact immutable task-owned simulation integration registration.
StandardExtensionDeclarations	Cross-checked provider-free declarations for the standard factory.
SupportSurfaceAffordanceBinding	Declare one exact object target frame on a support parent.
TaskProgramAdapterFactory	Bind one initialized environment to its Task Program adapter.
TaskProgramEnvironmentAdapter	Compile and run Task Programs through explicit environment factories.
TaskProgramEnvironmentFactory	Environment-owned factories for one explicit semantic integration.
TaskProgramIntegrationCatalog	Provider-free integration directory owned by one task registration.
TaskProgramRuntimeAssembly	Auditable result of one fresh environment runtime assembly.

Exceptions:

IntegrationFingerprintMismatch

Raised when a live integration no longer matches its registration.

Functions:

create_simulation_task_program_adapter(...)	Create a complete production adapter from one standard Gym environment.
default_simulation_settle_presets()	Return independently owned built-in post-policy presets.

class embodichain.lab.task_program.integrations.AntipodalGraspAffordanceBinding

Build one antipodal grasp affordance from a selected rigid-object mesh.

internal_axis upgrades the payload to AxisAlignAffordance while preserving the same antipodal grasp capability. The axis is expressed in the rigid object’s local frame and can be reused by held-object skills such as Pour.

Methods:

__init__(entity_id, object_id, native_name, ...)

__init__(entity_id, object_id, native_name, revision, aliases=(), relative_pose=(1.0, 0.0, 0.0, 0.0, 0.0, 1.0, 0.0, 0.0, 0.0, 0.0, 1.0, 0.0, 0.0, 0.0, 0.0, 1.0), mesh_env_id=0, internal_axis=None)

class embodichain.lab.task_program.integrations.ConfiguredHandOverPoseProvider

Resolve a hand-over delivery target from immutable configuration.

The provider carries object-space poses rather than arm trajectories. The shared semantic compiler and atomic HandOver implementation remain responsible for grasp selection, IK, motion generation, transfer, release, and delivery. Keeping the numeric declaration as tuple fields also makes the provider suitable for task-registration catalog fingerprinting.

Parameters:

• final_position (tuple[float, float, float]) – World-frame object delivery position.

• final_quaternion_wxyz (tuple[float, float, float, float]) – World-frame object delivery orientation.

Methods:

__init__(final_position, final_quaternion_wxyz)
resolve(call, *, context, bound)	Return an independently owned object-space delivery target.

__init__(final_position, final_quaternion_wxyz)

resolve(call, *, context, bound)

Return an independently owned object-space delivery target.

Parameters:

• call (HandOver) – Canonical hand-over semantic call.

• context (PlanningContext) – Latest immutable planning observation.

• bound (BoundSemanticCall) – Engine/profile-bound hand-over call.

Return type:

HandOverPoseTargets

Returns:

Configured final object-space target.

class embodichain.lab.task_program.integrations.ContainerAffordanceBinding

Declare one exact object target frame inside a container parent.

Parameters:

• entity_id (str) – Canonical ID of the placement affordance.

• parent_id (str) – Canonical object, articulation, or link parent ID.

• native_name (str) – Stable native name of this target frame.

• aliases (tuple[str, ...]) – Optional non-authoritative lookup aliases.

• object_target_pose (tuple[float, ...]) – Desired object pose relative to the parent.

• minimum_confidence (float) – Minimum parent/affordance observation confidence.

• is_default (bool) – Whether this is the parent’s default Place(inside=...) frame.

Methods:

__init__(entity_id, parent_id, native_name)

__init__(entity_id, parent_id, native_name, aliases=(), object_target_pose=(1.0, 0.0, 0.0, 0.0, 0.0, 1.0, 0.0, 0.0, 0.0, 0.0, 1.0, 0.0, 0.0, 0.0, 0.0, 1.0), minimum_confidence=0.0, is_default=False)

class embodichain.lab.task_program.integrations.ControlPartCommandPreset

Named one-dimensional joint commands for one exact control part.

Methods:

__init__(preset_id, control_part, commands)
build(*, control_dof)	Build a command profile after validating the native control width.
declare()	Build a provider-free command profile from declared tuple widths.

__init__(preset_id, control_part, commands)

build(*, control_dof)

Build a command profile after validating the native control width.

Return type:

ControlPartCommandProfile

declare()

Build a provider-free command profile from declared tuple widths.

Return type:

ControlPartCommandProfile

class embodichain.lab.task_program.integrations.ControlPartEndpointBinding

Profile endpoint backed by one explicit robot control part.

Methods:

__init__(endpoint_id, control_part, capabilities)
build(robot)	Build a joint-backed endpoint after native control-part validation.
declare()	Return the endpoint contract without reading a robot.

__init__(endpoint_id, control_part, capabilities, command_preset=None)

build(robot)

Build a joint-backed endpoint after native control-part validation.

Return type:

ResourceEndpoint

declare()

Return the endpoint contract without reading a robot.

Return type:

ResourceEndpoint

class embodichain.lab.task_program.integrations.ControlPartEvidenceProviderDeclaration

Provider-free identity of one control-part evidence factory.

Parameters:

• factory_type (type[object]) – Exact immutable factory implementation type.

• provider_id (str) – Stable effect-evidence provider identifier.

• revision (str) – Exact provider contract revision.

Methods:

__init__(factory_type, provider_id, revision)

__init__(factory_type, provider_id, revision)

class embodichain.lab.task_program.integrations.ControlPartEvidenceProviderFactory

Registration-owned factory for one live control-part evidence provider.

Methods:

__init__(*args, **kwargs)
create(*, simulation, robot, scene_registry, ...)	Create one live provider bound to the exact assembled runtime.

__init__(*args, **kwargs)

create(*, simulation, robot, scene_registry, engine, scene_provider)

Create one live provider bound to the exact assembled runtime.

Parameters:

• simulation (object) – Simulation that owns the selected robot and sensors.

• robot (object) – Exact robot selected by the environment factory.

• scene_registry (SceneRegistry) – Exact live semantic scene registry.

• engine (AtomicActionEngine) – Atomic-action engine assembled for the robot.

• scene_provider (SceneProvider) – Shared synchronized live scene provider.

Return type:

EffectEvidenceProvider

Returns:

Fresh effect-evidence provider for the assembled runtime.

class embodichain.lab.task_program.integrations.ControlPartResourceBinding

Joint-backed robot resource containing control-part endpoints.

Methods:

__init__(resource_id[, endpoints, members])
build(robot)	Build a resource containing strictly validated control-part endpoints.
declare()	Return the resource graph without reading native control parts.

__init__(resource_id, endpoints=(), members=())

build(robot)

Build a resource containing strictly validated control-part endpoints.

Return type:

RobotResource

declare()

Return the resource graph without reading native control parts.

Return type:

RobotResource

class embodichain.lab.task_program.integrations.CuroboParallelCommandSafetyValidator

Validate the exact synchronized joint segment before transport dispatch.

Parameters:

• robot (object) – Live robot supplying measured joint state and control-part IDs.

• motion_generator (MotionGenerator) – Runtime motion generator backed by exact cuRobo.

• scene_registry (SceneRegistry) – Authoritative live collision-scene registry.

• validation_control_part (str) – Aggregate control part for merged commands.

• max_joint_step (float) – Maximum displacement between collision samples.

• max_interpolation_samples (int) – Fail-closed sample-count upper bound.

Methods:

__init__(*, robot, motion_generator, ...)
validate(*, branch_frames, merged_frame)	Reject a merged command whose exact interpolated segment collides.

__init__(*, robot, motion_generator, scene_registry, validation_control_part, max_joint_step, max_interpolation_samples)

validate(*, branch_frames, merged_frame)

Reject a merged command whose exact interpolated segment collides.

Return type:

None

class embodichain.lab.task_program.integrations.CuroboParallelSafetyValidatorFactory

Create exact-sample collision gates for one aggregate control part.

validation_control_part must contain every joint that any parallel branch can command. A common dual-arm example is "dual_arm". The cuRobo model for that part remains the authoritative bounds, self-collision, and world-collision model.

Parameters:

• validation_control_part (str) – Aggregate robot control part containing every joint that a parallel lane may command.

• max_joint_step (float) – Maximum absolute joint displacement between collision samples in radians or the joint’s native linear unit.

• max_interpolation_samples (int) – Fail-closed upper bound on samples per frame.

Methods:

__init__(validation_control_part[, ...])
create(*, simulation, robot, scene_registry, ...)	Create one fresh validator bound to the assembled live runtime.

__init__(validation_control_part, max_joint_step=0.025, max_interpolation_samples=256)

create(*, simulation, robot, scene_registry, engine)

Create one fresh validator bound to the assembled live runtime.

Return type:

CuroboParallelCommandSafetyValidator

class embodichain.lab.task_program.integrations.EndpointAdapterDeclaration

Provider-free declaration of one exact endpoint adapter.

Methods:

__init__(endpoint_type, adapter_type, ...)

__init__(endpoint_type, adapter_type, adapter_id, runtime_transport_ids, runtime_target_types, tracking_feedback_source_keys, tracking_projector_keys, effect_evidence_source_keys)

exception embodichain.lab.task_program.integrations.IntegrationFingerprintMismatch

Raised when a live integration no longer matches its registration.

class embodichain.lab.task_program.integrations.ParallelCommandSafetyValidatorFactory

Registration-owned factory for one authoritative live safety gate.

Methods:

__init__(*args, **kwargs)
create(*, simulation, robot, scene_registry, ...)	Create one live gate bound to the exact assembled runtime.

__init__(*args, **kwargs)

create(*, simulation, robot, scene_registry, engine)

Create one live gate bound to the exact assembled runtime.

Return type:

ParallelCommandSafetyValidator

class embodichain.lab.task_program.integrations.ParallelSafetyDeclaration

Provider-free identity and transport coverage of one safety factory.

Methods:

__init__(factory_type, validator_id, ...)

__init__(factory_type, validator_id, revision, supported_transport_ids)

class embodichain.lab.task_program.integrations.PlanningObservationPort

Combined observation and full-qpos port required by the Gym runtime.

class embodichain.lab.task_program.integrations.RegisteredSemanticLowererDeclaration

Provider-free identity of one registered semantic lowerer factory.

Parameters:

• factory_type (type[object]) – Exact immutable factory implementation type.

• call_id (str) – Registered semantic-call ID owned by the factory.

• revision (str) – Exact factory contract revision.

Methods:

__init__(factory_type, call_id, revision)

__init__(factory_type, call_id, revision)

class embodichain.lab.task_program.integrations.RegisteredSemanticLowererFactory

Registration-owned factory for one fresh live semantic lowerer.

Methods:

__init__(*args, **kwargs)
create(*, simulation, robot, scene_registry, ...)	Create a lowerer bound to the exact assembled simulation runtime.

__init__(*args, **kwargs)

create(*, simulation, robot, scene_registry, engine)

Create a lowerer bound to the exact assembled simulation runtime.

Parameters:

• simulation (object) – Simulation that owns the selected robot and entities.

• robot (object) – Exact robot selected by the environment factory.

• scene_registry (SceneRegistry) – Exact live semantic scene registry.

• engine (AtomicActionEngine) – Atomic-action engine assembled for the robot.

Return type:

RegisteredSemanticLowerer

Returns:

A fresh registered semantic lowerer matching the declared call.

class embodichain.lab.task_program.integrations.RuntimeTransportDeclaration

Provider-free declaration of one ordered runtime transport encoder.

Methods:

__init__(transport_type, transport_id, ...)

__init__(transport_type, transport_id, target_types, payload_types)

class embodichain.lab.task_program.integrations.SimulationArticulationBinding

Explicit binding for one simulation articulation.

Methods:

__init__(entity_id, simulation_uid[, ...])

__init__(entity_id, simulation_uid, aliases=(), dynamics=SceneDynamics.UNKNOWN, collision_role=SceneCollisionRole.NONE, semantic_type=None, geometry_provider=None)

class embodichain.lab.task_program.integrations.SimulationArticulationLinkBinding

Explicit canonical link backed by one native articulation link.

Methods:

__init__(entity_id, articulation_id, ...[, ...])

__init__(entity_id, articulation_id, native_link_name, aliases=(), dynamics=SceneDynamics.UNKNOWN, semantic_type=None)

class embodichain.lab.task_program.integrations.SimulationRigidObjectBinding

Explicit binding for one simulation rigid object.

Methods:

__init__(entity_id, simulation_uid[, ...])

__init__(entity_id, simulation_uid, aliases=(), dynamics=SceneDynamics.UNKNOWN, collision_role=SceneCollisionRole.NONE, semantic_type=None, default_grasp_affordance=None, geometry_provider=None)

class embodichain.lab.task_program.integrations.SimulationRobotSkillProfileBinding

Build a profile from typed resources with strict native validation.

Methods:

__init__(profile_id, resources[, ...])
build(robot)	Build the existing profile after validating every typed resource.
declare()	Project the complete provider-free robot skill profile.

__init__(profile_id, resources, command_presets=(), defaults=<factory>, presets=(), default_preset=None, skill_presets=<factory>, grounding_providers=<factory>)

build(robot)

Build the existing profile after validating every typed resource.

Parameters:

robot (Robot) – Live robot selected by the simulation factory.

Return type:

RobotSkillProfile

Returns:

Reusable, engine-independent robot skill profile.

declare()

Project the complete provider-free robot skill profile.

Return type:

RobotSkillProfile

class embodichain.lab.task_program.integrations.SimulationSceneBinding

Build one authoritative registry from explicit simulation bindings.

Methods:

__init__(registry_id[, rigid_objects, ...])
build(simulation)	Build the existing authoritative scene registry.
declare()	Project the complete provider-free scene declaration.

__init__(registry_id, rigid_objects=(), articulations=(), links=(), antipodal_grasps=(), support_surfaces=(), containers=(), collision_world_mode=None)

build(simulation)

Build the existing authoritative scene registry.

Parameters:

simulation (SimulationManager) – Live simulation used only for explicitly named lookups.

Return type:

SceneRegistry

Returns:

Immutable registry with typed roots, links, and affordances.

declare()

Project the complete provider-free scene declaration.

Return type:

SceneManifest

class embodichain.lab.task_program.integrations.SimulationSegmentPolicyPort

Execute built-in segment policies against explicitly bound simulation data.

Parameters:

• simulation (SimulationManager) – Live simulation used only for UIDs declared in scene_binding.

• robot (Robot) – Live robot used to produce full target-qpos holds while the post-policy observes settling.

• scene_binding (SimulationSceneBinding) – Exact canonical-to-native scene declaration.

• settle_presets (Mapping[str, DynamicSettleMonitorCfg] | None) – Named settling policies. None installs the shared rigid_object and articulation presets.

• env_ids (Tensor | None) – Optional stable logical row IDs. They describe correlation, not simulator row indices; simulator rows remain ordered exactly as returned by the robot and bound entities.

The same instance implements both SegmentPostPolicyPort and SegmentValidatorPort. Unknown policy types, presets, canonical IDs, or native entities fail before an action is emitted.

Methods:

__init__(simulation, robot, scene_binding, *)
actions(policy, *, segment, active_mask)	Yield full target-qpos hold actions until rows settle or time out.
post_policy_metadata(policy, *, segment)	Return the latest JSON-safe settling trace for one policy.
post_policy_result(policy, *, segment)	Return the latest independently owned per-row settling result.
validate(validator, *, segment)	Observe one explicitly bound entity against its validator contract.
validate_policy(policy, *, segment)	Validate one post-policy against static bindings without observation.
validate_validator(validator, *, segment)	Validate one validator against static bindings without observation.
validator_metadata(validator, *, segment)	Return an owned JSON-safe trace for one completed validator.

Attributes:

settle_preset_ids

Return installed post-policy preset IDs in declaration order.

__init__(simulation, robot, scene_binding, *, settle_presets=None, env_ids=None)

actions(policy, *, segment, active_mask)

Yield full target-qpos hold actions until rows settle or time out.

Parameters:

• policy (Any) – Exact compiled wait_stable policy.

• segment (Any) – Exact segment that owns policy.

• active_mask (Tensor) – Rows that remain eligible after runtime execution and preceding post-policies. Inactive rows are held safely but do not participate in settling, timeout, or success results.

Yields:

Fresh full target-qpos hold commands consumed by ordinary env.step(). Reading drive targets instead of measured joint positions preserves contact preload in position-controlled tools. Rows inactive when the policy starts use fresh measured qpos holds.

Return type:

Iterator[Tensor]

Timeout is a normal row-local result boundary. Timed-out rows are exposed through post_policy_result() and post_policy_metadata(); no batch-level exception is raised.

post_policy_metadata(policy, *, segment)

Return the latest JSON-safe settling trace for one policy.

The trace is available after the policy generator has started. A terminal trace has status "settled", "timed_out", or "skipped" when no rows remain active; an early demo interruption intentionally retains the latest "running" snapshot for diagnosis.

Return type:

Mapping[str, object]

post_policy_result(policy, *, segment)

Return the latest independently owned per-row settling result.

Return type:

Tensor

property settle_preset_ids: tuple[str, ...]

Return installed post-policy preset IDs in declaration order.

validate(validator, *, segment)

Observe one explicitly bound entity against its validator contract.

Parameters:

• validator (Any) – Exact compiled segment validator.

• segment (Any) – Exact segment that owns validator.

Return type:

Tensor

Returns:

Boolean tensor with one result per simulation row.

validate_policy(policy, *, segment)

Validate one post-policy against static bindings without observation.

This method reads only the compiled declaration, installed preset table, and entities resolved when the port was constructed. It never samples velocity or qpos and never emits a controller action.

Return type:

None

validate_validator(validator, *, segment)

Validate one validator against static bindings without observation.

Return type:

None

validator_metadata(validator, *, segment)

Return an owned JSON-safe trace for one completed validator.

Return type:

Mapping[str, object]

class embodichain.lab.task_program.integrations.SimulationTaskProgramAdapterFactory

Create one standard simulation adapter after an environment is initialized.

Parameters:

• registration (SimulationTaskProgramRegistration) – Immutable provider-free task integration.

• grasp_pose_generator_factories (Mapping[str, Callable[[], GraspPoseGenerator]] | None) – Zero-argument factories keyed by runtime grasp endpoint target ID. Each environment receives fresh services.

Methods:

__init__(registration, *[, ...])
create_adapter(environment)	Create fresh grasp services and bind the initialized environment.

Attributes:

registration

Return the exact static registration owned by this factory.

__init__(registration, *, grasp_pose_generator_factories=None)

create_adapter(environment)

Create fresh grasp services and bind the initialized environment.

Return type:

TaskProgramEnvironmentAdapter

property registration: SimulationTaskProgramRegistration

Return the exact static registration owned by this factory.

class embodichain.lab.task_program.integrations.SimulationTaskProgramFactory

Build every live Task Program component from one task registration.

Parameters:

• simulation (SimulationManager) – Exact live simulation that owns robot and scene UIDs.

• robot (Robot) – Exact robot selected for planning and evidence acquisition.

• registration (SimulationTaskProgramRegistration) – Frozen scene, profile, and extension declarations.

• step_dt (float) – Authoritative Gym control cadence.

• planner_cfg (BasePlannerCfg | None) – Explicit planner configuration. None selects TOPPRA for robot.uid.

• motion_generator_factory (Callable[[], MotionGenerator] | None) – Optional fresh-generator factory. It is mutually exclusive with planner_cfg and intended for custom planners and isolated tests.

• grasp_pose_generators (Mapping[str, GraspPoseGenerator] | None) – Standalone grasp-pose services keyed by grasp endpoint target ID, normally a robot control-part name.

• translation_threshold (float) – Material scene translation threshold.

• rotation_threshold (float) – Material scene rotation threshold.

The Gym cadence is attached to each live PlanningContext. Motion policy presets describe behavior; they do not own the environment’s command period.

Methods:

__init__(simulation, robot, registration, *, ...)
create_adapter()	Create the exact Gym adapter with shared simulation policy ports.
create_atomic_action_engine(profile)	Create a fresh engine around the selected planner and exact profile.
create_effect_evidence_providers(*, ...)	Create built-in control-part and articulation evidence providers.
create_parallel_command_safety_validator(*, ...)	Create the registration-owned validator for this runtime assembly.
create_planning_observation_provider(*, ...)	Create one planning port and planner-validated shared scene provider.
create_registered_semantic_lowerers(*, ...)	Create fresh registration-owned lowerers for this runtime assembly.
create_robot_skill_profile()	Build the robot profile used to lower calls to Atomic Skills.
create_scene_registry()	Build one fresh authoritative registry from the task registration.
from_environment(environment, *, registration)	Create a registered factory from the standard Gym environment surface.
registration_owned_segment_policy_ports()	Return registration-owned post-policy and validator ports.

Attributes:

robot_profile_id	Return the exact configured robot-profile ID.
scene_registry_id	Return the exact configured scene-registry ID.
segment_policy_port	Return the shared simulation post-policy and validator port.
step_dt	Return the authoritative Gym control cadence.
task_program_registration	Return the exact standard registration owned by this factory.

__init__(simulation, robot, registration, *, step_dt, planner_cfg=None, motion_generator_factory=None, grasp_pose_generators=None, translation_threshold=0.0001, rotation_threshold=0.001)

create_adapter()

Create the exact Gym adapter with shared simulation policy ports.

Return type:

TaskProgramEnvironmentAdapter

create_atomic_action_engine(profile)

Create a fresh engine around the selected planner and exact profile.

Return type:

AtomicActionEngine

create_effect_evidence_providers(*, scene_registry, engine, observation_provider)

Create built-in control-part and articulation evidence providers.

Return type:

Iterable[EffectEvidenceProvider]

create_parallel_command_safety_validator(*, scene_registry, engine, observation_provider)

Create the registration-owned validator for this runtime assembly.

Return type:

ParallelCommandSafetyValidator

create_planning_observation_provider(*, scene_registry, engine, clock)

Create one planning port and planner-validated shared scene provider.

Return type:

PlanningObservationPort

create_registered_semantic_lowerers(*, scene_registry, engine)

Create fresh registration-owned lowerers for this runtime assembly.

Return type:

tuple[RegisteredSemanticLowerer, ...]

create_robot_skill_profile()

Build the robot profile used to lower calls to Atomic Skills.

Return type:

RobotSkillProfile

create_scene_registry()

Build one fresh authoritative registry from the task registration.

Return type:

SceneRegistry

classmethod from_environment(environment, *, registration, planner_cfg=None, motion_generator_factory=None, grasp_pose_generators=None, translation_threshold=0.0001, rotation_threshold=0.001)

Create a registered factory from the standard Gym environment surface.

Return type:

SimulationTaskProgramFactory

registration_owned_segment_policy_ports()

Return registration-owned post-policy and validator ports.

Return type:

tuple[SimulationSegmentPolicyPort, SimulationSegmentPolicyPort]

property robot_profile_id: str

Return the exact configured robot-profile ID.

property scene_registry_id: str

Return the exact configured scene-registry ID.

property segment_policy_port: SimulationSegmentPolicyPort

Return the shared simulation post-policy and validator port.

property step_dt: float

Return the authoritative Gym control cadence.

property task_program_registration: SimulationTaskProgramRegistration

Return the exact standard registration owned by this factory.

class embodichain.lab.task_program.integrations.SimulationTaskProgramRegistration

Exact immutable task-owned simulation integration registration.

Methods:

__init__(scene_binding, robot_profile_binding)
assert_unchanged()	Reject nested declaration drift before live component creation.
create_control_part_evidence_provider(*, ...)	Create the registration-owned live control-part evidence provider.
create_parallel_safety_validator(*, ...)	Create and strictly validate the registration-owned live safety gate.
create_registered_semantic_lowerers(*, ...)	Create fresh live lowerers declared by this registration.
validate_bound_profile(bound_profile)	Validate resolved semantic endpoints against this registration.
validate_engine(engine)	Validate live atomic skills against this registration.
validate_robot_profile(profile)	Validate a live profile against its registered declaration.
validate_scene_registry(registry)	Validate a live registry against the registered scene declaration.

Attributes:

endpoint_adapter_map	Return custom live adapters keyed by their exact endpoint type.
fingerprint	Return the canonical registration fingerprint.

__init__(scene_binding, robot_profile_binding, call_catalog=<factory>, settle_presets=<factory>, relation_grounders=(), handover_pose_providers=(), endpoint_adapters=(), runtime_transports=(), parallel_safety_factory=None, control_part_evidence_factory=None, registered_semantic_lowerer_factories=())

assert_unchanged()

Reject nested declaration drift before live component creation.

Return type:

None

create_control_part_evidence_provider(*, simulation, robot, scene_registry, engine, scene_provider)

Create the registration-owned live control-part evidence provider.

Parameters:

• simulation (object) – Simulation that owns the selected robot and sensors.

• robot (object) – Exact robot selected by the environment factory.

• scene_registry (SceneRegistry) – Exact live semantic scene registry.

• engine (AtomicActionEngine) – Atomic-action engine assembled for the robot.

• scene_provider (SceneProvider) – Shared synchronized live scene provider.

Return type:

EffectEvidenceProvider | None

Returns:

Fresh registered provider, or None when no factory is declared.

create_parallel_safety_validator(*, simulation, robot, scene_registry, engine)

Create and strictly validate the registration-owned live safety gate.

Return type:

ParallelCommandSafetyValidator | None

create_registered_semantic_lowerers(*, simulation, robot, scene_registry, engine)

Create fresh live lowerers declared by this registration.

Return type:

tuple[RegisteredSemanticLowerer, ...]

property endpoint_adapter_map: Mapping[type[ResourceEndpoint], ResourceEndpointAdapter]

Return custom live adapters keyed by their exact endpoint type.

property fingerprint: str

Return the canonical registration fingerprint.

validate_bound_profile(bound_profile)

Validate resolved semantic endpoints against this registration.

Return type:

None

validate_engine(engine)

Validate live atomic skills against this registration.

Return type:

None

validate_robot_profile(profile)

Validate a live profile against its registered declaration.

Return type:

None

validate_scene_registry(registry)

Validate a live registry against the registered scene declaration.

Return type:

None

class embodichain.lab.task_program.integrations.StandardExtensionDeclarations

Cross-checked provider-free declarations for the standard factory.

Methods:

__init__(endpoint_adapters, ...[, ...])

__init__(endpoint_adapters, runtime_transports, parallel_safety, control_part_evidence, registered_semantic_lowerers=<factory>)

class embodichain.lab.task_program.integrations.SupportSurfaceAffordanceBinding

Declare one exact object target frame on a support parent.

Parameters:

• entity_id (str) – Canonical ID of the placement affordance.

• parent_id (str) – Canonical object, articulation, or link parent ID.

• native_name (str) – Stable native name of this target frame.

• aliases (tuple[str, ...]) – Optional non-authoritative lookup aliases.

• object_target_pose (tuple[float, ...]) – Desired object pose relative to the parent.

• minimum_confidence (float) – Minimum parent/affordance observation confidence.

• is_default (bool) – Whether this is the parent’s default Place(on=...) frame.

Methods:

__init__(entity_id, parent_id, native_name)

__init__(entity_id, parent_id, native_name, aliases=(), object_target_pose=(1.0, 0.0, 0.0, 0.0, 0.0, 1.0, 0.0, 0.0, 0.0, 0.0, 1.0, 0.0, 0.0, 0.0, 0.0, 1.0), minimum_confidence=0.0, is_default=False)

class embodichain.lab.task_program.integrations.TaskProgramAdapterFactory

Bind one initialized environment to its Task Program adapter.

Methods:

__init__(*args, **kwargs)
create_adapter(environment)	Create the adapter after the environment has built its live scene.

__init__(*args, **kwargs)

create_adapter(environment)

Create the adapter after the environment has built its live scene.

Return type:

TaskProgramEnvironmentAdapter

class embodichain.lab.task_program.integrations.TaskProgramEnvironmentAdapter

Compile and run Task Programs through explicit environment factories.

Parameters:

• factory (TaskProgramEnvironmentFactory) – Environment-owned live-provider and engine factory.

• step_dt (float) – Authoritative Gym control cadence in seconds.

• integration_catalog (TaskProgramIntegrationCatalog | None) – Optional immutable task-registration catalog used for provider-free compilation.

• registration (SimulationTaskProgramRegistration | None) – Optional exact standard task registration. When present, every compiler/runtime extension comes exclusively from it.

• call_catalog (SemanticCallCatalog | None) – Optional immutable semantic call catalog. The built-in catalog is used when omitted.

• endpoint_adapters (Mapping[type[ResourceEndpoint], ResourceEndpointAdapter] | None) – Optional custom robot endpoint adapters.

• registered_lowerers (Iterable[RegisteredSemanticLowerer]) – Explicit lowerers for registered semantic calls.

• relation_grounders (Iterable[RelationTargetGrounder]) – Explicit relation-target grounding providers.

• handover_pose_providers (Iterable[HandOverPoseProvider]) – Explicit embodiment hand-over providers.

• effect_monitor_registry (EffectMonitorRegistry | None) – Optional exact-version monitor registry.

• runtime_transports (Iterable[RuntimeTransportActionEncoder]) – Additional runtime-command-to-Gym encoders.

• runner_cfg (ExecutionRunnerCfg | None) – Optional execution-runner policy.

• post_policy_port (SegmentPostPolicyPort | None) – Optional environment post-policy executor.

• validator_port (SegmentValidatorPort | None) – Optional environment segment validator.

• parallel_safety_validator (ParallelCommandSafetyValidator | None) – Optional authoritative parallel safety gate.

A call to compile() snapshots only scene identities. A call to assemble_runtime() creates a fresh live runtime, which makes reset and episode ownership explicit and avoids retaining providers in compiled data.

Methods:

__init__(factory, *, step_dt[, ...])
assemble_runtime(integration)	Create a fresh fully connected semantic runtime.
compile(program)	Compile one program after exact integration-selection validation.
create_bridge(program)	Create a fresh Gym bridge for one provider-free compiled program.

Attributes:

robot_profile_id	Return the exact robot profile ID accepted by this adapter.
scene_registry_id	Return the exact scene integration ID accepted by this adapter.
step_dt	Return the authoritative environment-step cadence.

__init__(factory, *, step_dt, integration_catalog=None, registration=None, call_catalog=None, endpoint_adapters=None, registered_lowerers=(), relation_grounders=(), handover_pose_providers=(), effect_monitor_registry=None, runtime_transports=(), runner_cfg=None, post_policy_port=None, validator_port=None, parallel_safety_validator=None)

assemble_runtime(integration)

Create a fresh fully connected semantic runtime.

Parameters:

integration (TaskProgramIntegrationCfg) – Exact scene, profile, and runtime-preset selection.

Return type:

TaskProgramRuntimeAssembly

Returns:

Owned assembly containing every validated runtime boundary.

compile(program)

Compile one program after exact integration-selection validation.

Parameters:

program (TaskProgramCfg) – Strict declarative program configuration.

Return type:

CompiledTaskProgram

Returns:

Provider-free lazily expanded compiled program.

create_bridge(program)

Create a fresh Gym bridge for one provider-free compiled program.

Parameters:

program (CompiledTaskProgram) – Program compiled for this adapter’s exact integration IDs.

Return type:

TaskProgramDemoBridge

Returns:

Lazy bridge sharing one newly assembled runtime, clock, and sink.

property robot_profile_id: str

Return the exact robot profile ID accepted by this adapter.

Returns:

Stable robot-profile identifier.

property scene_registry_id: str

Return the exact scene integration ID accepted by this adapter.

Returns:

Stable scene-registry identifier.

property step_dt: float

Return the authoritative environment-step cadence.

Returns:

Positive control step duration in seconds.

class embodichain.lab.task_program.integrations.TaskProgramEnvironmentFactory

Environment-owned factories for one explicit semantic integration.

Implementations normally live in reusable robot/task integration modules, not in individual task motion planners. Every method is passed the exact objects selected earlier in the assembly so a factory cannot silently bind a different scene, robot profile, or engine.

Methods:

__init__(*args, **kwargs)
create_atomic_action_engine(profile)	Create an engine for exactly profile and its motion backend.
create_effect_evidence_providers(*, ...)	Create exact-version providers used by semantic effect monitors.
create_planning_observation_provider(*, ...)	Create fresh planning observations and aligned full-qpos reads.
create_robot_skill_profile()	Create the authoritative declarative robot skill profile.
create_scene_registry()	Create the authoritative explicitly registered live scene.

Attributes:

robot_profile_id	Return the configuration ID selecting this robot profile.
scene_registry_id	Return the configuration ID selecting this scene declaration.

__init__(*args, **kwargs)

create_atomic_action_engine(profile)

Create an engine for exactly profile and its motion backend.

Parameters:

profile (RobotSkillProfile) – Profile selected and validated by the adapter.

Return type:

AtomicActionEngine

Returns:

Atomic engine connected to the environment’s robot and planner.

create_effect_evidence_providers(*, scene_registry, engine, observation_provider)

Create exact-version providers used by semantic effect monitors.

Parameters:

• scene_registry (SceneRegistry) – Exact registry selected for this runtime.

• engine (AtomicActionEngine) – Exact atomic engine selected for this runtime.

• observation_provider (PlanningObservationPort) – Shared planning observation provider.

Return type:

Iterable[EffectEvidenceProvider]

Returns:

Explicit provider set; an empty iterable is permitted.

create_planning_observation_provider(*, scene_registry, engine, clock)

Create fresh planning observations and aligned full-qpos reads.

Parameters:

• scene_registry (SceneRegistry) – Exact registry selected for this runtime.

• engine (AtomicActionEngine) – Exact atomic engine selected for this runtime.

• clock (EnvironmentStepClock) – Shared environment-step execution clock.

Return type:

PlanningObservationPort

Returns:

Combined planning-observation and qpos provider.

create_robot_skill_profile()

Create the authoritative declarative robot skill profile.

Return type:

RobotSkillProfile

Returns:

Profile whose ID matches robot_profile_id.

create_scene_registry()

Create the authoritative explicitly registered live scene.

Return type:

SceneRegistry

Returns:

Fresh registry containing only explicitly selected entities.

property robot_profile_id: str

Return the configuration ID selecting this robot profile.

Returns:

Stable robot-profile identifier.

property scene_registry_id: str

Return the configuration ID selecting this scene declaration.

Returns:

Stable scene-registry identifier.

class embodichain.lab.task_program.integrations.TaskProgramIntegrationCatalog

Provider-free integration directory owned by one task registration.

Methods:

__init__(scene_registry_id, ...)
preflight(program)	Compile and statically link every expanded semantic call.
validate_bound_endpoint_extensions(bound_profile)	Match every live resolved endpoint to its fingerprinted declaration.
validate_engine(engine)	Require the live engine to expose every statically selected skill.
validate_integration(integration, *, path)	Validate exact scene, profile, and runtime-preset selection.
validate_post_policy(policy, *, path)	Validate one registered post-policy kind and named preset.
validate_scene_reference(reference, *, role, ...)	Validate one typed scene reference against its declared role.
validate_semantic_call(call, *, path)	Validate that the semantic call is present in the integration catalog.
validate_validator(validator, *, path)	Validate one registered segment-validator kind.

__init__(scene_registry_id, robot_profile_id, scene, robot_profile, call_catalog, relation_grounder_keys, settle_preset_ids, endpoint_adapter_declarations, runtime_transport_declarations, parallel_safety_declaration, control_part_evidence_declaration, registered_semantic_lowerer_declarations, fingerprint, _required_skills)

preflight(program)

Compile and statically link every expanded semantic call.

Return type:

CompiledTaskProgram

validate_bound_endpoint_extensions(bound_profile)

Match every live resolved endpoint to its fingerprinted declaration.

Return type:

None

validate_engine(engine)

Require the live engine to expose every statically selected skill.

Return type:

None

validate_integration(integration, *, path)

Validate exact scene, profile, and runtime-preset selection.

Return type:

None

validate_post_policy(policy, *, path)

Validate one registered post-policy kind and named preset.

Return type:

None

validate_scene_reference(reference, *, role, path)

Validate one typed scene reference against its declared role.

Return type:

None

validate_semantic_call(call, *, path)

Validate that the semantic call is present in the integration catalog.

Return type:

None

validate_validator(validator, *, path)

Validate one registered segment-validator kind.

Return type:

None

class embodichain.lab.task_program.integrations.TaskProgramRuntimeAssembly

Auditable result of one fresh environment runtime assembly.

integration

Owned integration-selection snapshot.

scene_registry

Authoritative live scene registry.

robot_profile

Declarative robot resource profile.

manifest

Static scene/profile/call integration manifest.

engine

Bound atomic action engine.

compiler

Bound Semantic Call compiler.

observation_provider

Shared planning and full-qpos provider.

evidence_collector

Exact-version semantic evidence collector.

clock

Shared environment-step clock.

command_encoder

Runtime-frame to Gym-action encoder.

command_sink

Buffered Gym command sink.

runner_cfg

Runner policy selected by the integration runtime preset.

parallel_safety_validator

Optional fresh registration-owned safety gate.

runtime

Nonblocking Semantic Call runtime.

Methods:

__init__(integration, scene_registry, ...)

__init__(integration, scene_registry, robot_profile, manifest, engine, compiler, observation_provider, evidence_collector, clock, command_encoder, command_sink, runner_cfg, parallel_safety_validator, runtime)

embodichain.lab.task_program.integrations.create_simulation_task_program_adapter(environment, *, registration, planner_cfg=None, motion_generator_factory=None, grasp_pose_generators=None, translation_threshold=0.0001, rotation_threshold=0.001)

Create a complete production adapter from one standard Gym environment.

This is the intended task-side one-line integration. Standard tasks pass one immutable registration that owns their pre-simulation catalog and every runtime extension.

Parameters:

• environment (SimulationTaskProgramEnvironment) – Standard Gym simulation environment exposing sim, robot, and step_dt.

• registration (SimulationTaskProgramRegistration) – Required immutable task registration.

• planner_cfg (BasePlannerCfg | None) – Optional planner configuration owned by the factory.

• motion_generator_factory (Callable[[], MotionGenerator] | None) – Optional factory for one fresh motion generator.

• grasp_pose_generators (Mapping[str, GraspPoseGenerator] | None) – Standalone grasp-pose services keyed by grasp endpoint target ID.

• translation_threshold (float) – Scene translation revision threshold.

• rotation_threshold (float) – Scene rotation revision threshold.

Return type:

TaskProgramEnvironmentAdapter

Returns:

Complete production Task Program environment adapter.

embodichain.lab.task_program.integrations.default_simulation_settle_presets()

Return independently owned built-in post-policy presets.

Return type:

Mapping[str, DynamicSettleMonitorCfg]

Environment assembly

The provider-independent language and compiler live in embodichain.lab.task_program. This package owns explicit environment, registration, and simulation assembly. Gym stepping remains in embodichain.lab.gym.envs.task_program.

class embodichain.lab.task_program.integrations.TaskProgramEnvironmentAdapter

Compile and run Task Programs through explicit environment factories.

Parameters:

• factory (TaskProgramEnvironmentFactory) – Environment-owned live-provider and engine factory.

• step_dt (float) – Authoritative Gym control cadence in seconds.

• integration_catalog (TaskProgramIntegrationCatalog | None) – Optional immutable task-registration catalog used for provider-free compilation.

• registration (SimulationTaskProgramRegistration | None) – Optional exact standard task registration. When present, every compiler/runtime extension comes exclusively from it.

• call_catalog (SemanticCallCatalog | None) – Optional immutable semantic call catalog. The built-in catalog is used when omitted.

• endpoint_adapters (Mapping[type[ResourceEndpoint], ResourceEndpointAdapter] | None) – Optional custom robot endpoint adapters.

• registered_lowerers (Iterable[RegisteredSemanticLowerer]) – Explicit lowerers for registered semantic calls.

• relation_grounders (Iterable[RelationTargetGrounder]) – Explicit relation-target grounding providers.

• handover_pose_providers (Iterable[HandOverPoseProvider]) – Explicit embodiment hand-over providers.

• effect_monitor_registry (EffectMonitorRegistry | None) – Optional exact-version monitor registry.

• runtime_transports (Iterable[RuntimeTransportActionEncoder]) – Additional runtime-command-to-Gym encoders.

• runner_cfg (ExecutionRunnerCfg | None) – Optional execution-runner policy.

• post_policy_port (SegmentPostPolicyPort | None) – Optional environment post-policy executor.

• validator_port (SegmentValidatorPort | None) – Optional environment segment validator.

• parallel_safety_validator (ParallelCommandSafetyValidator | None) – Optional authoritative parallel safety gate.

A call to compile() snapshots only scene identities. A call to assemble_runtime() creates a fresh live runtime, which makes reset and episode ownership explicit and avoids retaining providers in compiled data.

Methods:

__init__(factory, *, step_dt[, ...])
assemble_runtime(integration)	Create a fresh fully connected semantic runtime.
compile(program)	Compile one program after exact integration-selection validation.
create_bridge(program)	Create a fresh Gym bridge for one provider-free compiled program.

Attributes:

robot_profile_id	Return the exact robot profile ID accepted by this adapter.
scene_registry_id	Return the exact scene integration ID accepted by this adapter.
step_dt	Return the authoritative environment-step cadence.

__init__(factory, *, step_dt, integration_catalog=None, registration=None, call_catalog=None, endpoint_adapters=None, registered_lowerers=(), relation_grounders=(), handover_pose_providers=(), effect_monitor_registry=None, runtime_transports=(), runner_cfg=None, post_policy_port=None, validator_port=None, parallel_safety_validator=None)

assemble_runtime(integration)

Create a fresh fully connected semantic runtime.

Parameters:

integration (TaskProgramIntegrationCfg) – Exact scene, profile, and runtime-preset selection.

Return type:

TaskProgramRuntimeAssembly

Returns:

Owned assembly containing every validated runtime boundary.

compile(program)

Compile one program after exact integration-selection validation.

Parameters:

program (TaskProgramCfg) – Strict declarative program configuration.

Return type:

CompiledTaskProgram

Returns:

Provider-free lazily expanded compiled program.

create_bridge(program)

Create a fresh Gym bridge for one provider-free compiled program.

Parameters:

program (CompiledTaskProgram) – Program compiled for this adapter’s exact integration IDs.

Return type:

TaskProgramDemoBridge

Returns:

Lazy bridge sharing one newly assembled runtime, clock, and sink.

property robot_profile_id: str

Return the exact robot profile ID accepted by this adapter.

Returns:

Stable robot-profile identifier.

property scene_registry_id: str

Return the exact scene integration ID accepted by this adapter.

Returns:

Stable scene-registry identifier.

property step_dt: float

Return the authoritative environment-step cadence.

Returns:

Positive control step duration in seconds.

class embodichain.lab.task_program.integrations.TaskProgramAdapterFactory

Bind one initialized environment to its Task Program adapter.

Methods:

__init__(*args, **kwargs)
create_adapter(environment)	Create the adapter after the environment has built its live scene.

__init__(*args, **kwargs)

create_adapter(environment)

Create the adapter after the environment has built its live scene.

Return type:

TaskProgramEnvironmentAdapter

class embodichain.lab.task_program.integrations.TaskProgramEnvironmentFactory

Environment-owned factories for one explicit semantic integration.

Implementations normally live in reusable robot/task integration modules, not in individual task motion planners. Every method is passed the exact objects selected earlier in the assembly so a factory cannot silently bind a different scene, robot profile, or engine.

Methods:

__init__(*args, **kwargs)
__new__(**kwargs)

__init__(*args, **kwargs)

__new__(**kwargs)

class embodichain.lab.task_program.integrations.TaskProgramRuntimeAssembly

Auditable result of one fresh environment runtime assembly.

integration

Owned integration-selection snapshot.

scene_registry

Authoritative live scene registry.

robot_profile

Declarative robot resource profile.

manifest

Static scene/profile/call integration manifest.

engine

Bound atomic action engine.

compiler

Bound Semantic Call compiler.

observation_provider

Shared planning and full-qpos provider.

evidence_collector

Exact-version semantic evidence collector.

clock

Shared environment-step clock.

command_encoder

Runtime-frame to Gym-action encoder.

command_sink

Buffered Gym command sink.

runner_cfg

Runner policy selected by the integration runtime preset.

parallel_safety_validator

Optional fresh registration-owned safety gate.

runtime

Nonblocking Semantic Call runtime.

Methods:

__init__(integration, scene_registry, ...)

__init__(integration, scene_registry, robot_profile, manifest, engine, compiler, observation_provider, evidence_collector, clock, command_encoder, command_sink, runner_cfg, parallel_safety_validator, runtime)

class embodichain.lab.task_program.integrations.PlanningObservationPort

Combined observation and full-qpos port required by the Gym runtime.

Methods:

__init__(*args, **kwargs)
__new__(**kwargs)

__init__(*args, **kwargs)

__new__(**kwargs)

Registration and extensions

class embodichain.lab.task_program.integrations.TaskProgramIntegrationCatalog

Provider-free integration directory owned by one task registration.

Methods:

__init__(scene_registry_id, ...)
preflight(program)	Compile and statically link every expanded semantic call.
validate_bound_endpoint_extensions(bound_profile)	Match every live resolved endpoint to its fingerprinted declaration.
validate_engine(engine)	Require the live engine to expose every statically selected skill.
validate_integration(integration, *, path)	Validate exact scene, profile, and runtime-preset selection.
validate_post_policy(policy, *, path)	Validate one registered post-policy kind and named preset.
validate_scene_reference(reference, *, role, ...)	Validate one typed scene reference against its declared role.
validate_semantic_call(call, *, path)	Validate that the semantic call is present in the integration catalog.
validate_validator(validator, *, path)	Validate one registered segment-validator kind.

__init__(scene_registry_id, robot_profile_id, scene, robot_profile, call_catalog, relation_grounder_keys, settle_preset_ids, endpoint_adapter_declarations, runtime_transport_declarations, parallel_safety_declaration, control_part_evidence_declaration, registered_semantic_lowerer_declarations, fingerprint, _required_skills)

preflight(program)

Compile and statically link every expanded semantic call.

Return type:

CompiledTaskProgram

validate_bound_endpoint_extensions(bound_profile)

Match every live resolved endpoint to its fingerprinted declaration.

Return type:

None

validate_engine(engine)

Require the live engine to expose every statically selected skill.

Return type:

None

validate_integration(integration, *, path)

Validate exact scene, profile, and runtime-preset selection.

Return type:

None

validate_post_policy(policy, *, path)

Validate one registered post-policy kind and named preset.

Return type:

None

validate_scene_reference(reference, *, role, path)

Validate one typed scene reference against its declared role.

Return type:

None

validate_semantic_call(call, *, path)

Validate that the semantic call is present in the integration catalog.

Return type:

None

validate_validator(validator, *, path)

Validate one registered segment-validator kind.

Return type:

None

class embodichain.lab.task_program.integrations.SimulationTaskProgramRegistration

Exact immutable task-owned simulation integration registration.

Methods:

__init__(scene_binding, robot_profile_binding)
assert_unchanged()	Reject nested declaration drift before live component creation.
create_control_part_evidence_provider(*, ...)	Create the registration-owned live control-part evidence provider.
create_parallel_safety_validator(*, ...)	Create and strictly validate the registration-owned live safety gate.
create_registered_semantic_lowerers(*, ...)	Create fresh live lowerers declared by this registration.
validate_bound_profile(bound_profile)	Validate resolved semantic endpoints against this registration.
validate_engine(engine)	Validate live atomic skills against this registration.
validate_robot_profile(profile)	Validate a live profile against its registered declaration.
validate_scene_registry(registry)	Validate a live registry against the registered scene declaration.

Attributes:

endpoint_adapter_map	Return custom live adapters keyed by their exact endpoint type.
fingerprint	Return the canonical registration fingerprint.

__init__(scene_binding, robot_profile_binding, call_catalog=<factory>, settle_presets=<factory>, relation_grounders=(), handover_pose_providers=(), endpoint_adapters=(), runtime_transports=(), parallel_safety_factory=None, control_part_evidence_factory=None, registered_semantic_lowerer_factories=())

assert_unchanged()

Reject nested declaration drift before live component creation.

Return type:

None

create_control_part_evidence_provider(*, simulation, robot, scene_registry, engine, scene_provider)

Create the registration-owned live control-part evidence provider.

Parameters:

• simulation (object) – Simulation that owns the selected robot and sensors.

• robot (object) – Exact robot selected by the environment factory.

• scene_registry (SceneRegistry) – Exact live semantic scene registry.

• engine (AtomicActionEngine) – Atomic-action engine assembled for the robot.

• scene_provider (SceneProvider) – Shared synchronized live scene provider.

Return type:

EffectEvidenceProvider | None

Returns:

Fresh registered provider, or None when no factory is declared.

create_parallel_safety_validator(*, simulation, robot, scene_registry, engine)

Create and strictly validate the registration-owned live safety gate.

Return type:

ParallelCommandSafetyValidator | None

create_registered_semantic_lowerers(*, simulation, robot, scene_registry, engine)

Create fresh live lowerers declared by this registration.

Return type:

tuple[RegisteredSemanticLowerer, ...]

property endpoint_adapter_map: Mapping[type[ResourceEndpoint], ResourceEndpointAdapter]

Return custom live adapters keyed by their exact endpoint type.

property fingerprint: str

Return the canonical registration fingerprint.

validate_bound_profile(bound_profile)

Validate resolved semantic endpoints against this registration.

Return type:

None

validate_engine(engine)

Validate live atomic skills against this registration.

Return type:

None

validate_robot_profile(profile)

Validate a live profile against its registered declaration.

Return type:

None

validate_scene_registry(registry)

Validate a live registry against the registered scene declaration.

Return type:

None

class embodichain.lab.task_program.integrations.IntegrationFingerprintMismatch

Raised when a live integration no longer matches its registration.

Methods:

__init__(*args, **kwargs)
__new__(**kwargs)

__init__(*args, **kwargs)

__new__(**kwargs)

class embodichain.lab.task_program.integrations.ControlPartEvidenceProviderDeclaration

Provider-free identity of one control-part evidence factory.

Parameters:

• factory_type (type[object]) – Exact immutable factory implementation type.

• provider_id (str) – Stable effect-evidence provider identifier.

• revision (str) – Exact provider contract revision.

Methods:

__init__(factory_type, provider_id, revision)

__init__(factory_type, provider_id, revision)

class embodichain.lab.task_program.integrations.ControlPartEvidenceProviderFactory

Registration-owned factory for one live control-part evidence provider.

Methods:

__init__(*args, **kwargs)
__new__(**kwargs)

__init__(*args, **kwargs)

__new__(**kwargs)

class embodichain.lab.task_program.integrations.RegisteredSemanticLowererDeclaration

Provider-free identity of one registered semantic lowerer factory.

Parameters:

• factory_type (type[object]) – Exact immutable factory implementation type.

• call_id (str) – Registered semantic-call ID owned by the factory.

• revision (str) – Exact factory contract revision.

Methods:

__init__(factory_type, call_id, revision)

__init__(factory_type, call_id, revision)

class embodichain.lab.task_program.integrations.RegisteredSemanticLowererFactory

Registration-owned factory for one fresh live semantic lowerer.

Methods:

__init__(*args, **kwargs)
__new__(**kwargs)

__init__(*args, **kwargs)

__new__(**kwargs)

class embodichain.lab.task_program.integrations.EndpointAdapterDeclaration

Provider-free declaration of one exact endpoint adapter.

Methods:

__init__(endpoint_type, adapter_type, ...)

__init__(endpoint_type, adapter_type, adapter_id, runtime_transport_ids, runtime_target_types, tracking_feedback_source_keys, tracking_projector_keys, effect_evidence_source_keys)

class embodichain.lab.task_program.integrations.RuntimeTransportDeclaration

Provider-free declaration of one ordered runtime transport encoder.

Methods:

__init__(transport_type, transport_id, ...)

__init__(transport_type, transport_id, target_types, payload_types)

class embodichain.lab.task_program.integrations.ParallelSafetyDeclaration

Provider-free identity and transport coverage of one safety factory.

Methods:

__init__(factory_type, validator_id, ...)

__init__(factory_type, validator_id, revision, supported_transport_ids)

class embodichain.lab.task_program.integrations.ParallelCommandSafetyValidatorFactory

Registration-owned factory for one authoritative live safety gate.

Methods:

__init__(*args, **kwargs)
__new__(**kwargs)

__init__(*args, **kwargs)

__new__(**kwargs)

class embodichain.lab.task_program.integrations.StandardExtensionDeclarations

Cross-checked provider-free declarations for the standard factory.

Methods:

__init__(endpoint_adapters, ...[, ...])

__init__(endpoint_adapters, runtime_transports, parallel_safety, control_part_evidence, registered_semantic_lowerers=<factory>)

embodichain.lab.task_program.integrations.VersionedKey(*args, **kwargs)

Built-in immutable sequence.

If no argument is given, the constructor returns an empty tuple. If iterable is specified the tuple is initialized from iterable’s items.

If the argument is a tuple, the return value is the same object.

alias of tuple[str, str]

embodichain.lab.task_program.integrations.default_simulation_settle_presets()

Return independently owned built-in post-policy presets.

Return type:

Mapping[str, DynamicSettleMonitorCfg]

Simulation bindings

class embodichain.lab.task_program.integrations.SimulationSceneBinding

Build one authoritative registry from explicit simulation bindings.

Methods:

__init__(registry_id[, rigid_objects, ...])
build(simulation)	Build the existing authoritative scene registry.
declare()	Project the complete provider-free scene declaration.

__init__(registry_id, rigid_objects=(), articulations=(), links=(), antipodal_grasps=(), support_surfaces=(), containers=(), collision_world_mode=None)

build(simulation)

Build the existing authoritative scene registry.

Parameters:

simulation (SimulationManager) – Live simulation used only for explicitly named lookups.

Return type:

SceneRegistry

Returns:

Immutable registry with typed roots, links, and affordances.

declare()

Project the complete provider-free scene declaration.

Return type:

SceneManifest

class embodichain.lab.task_program.integrations.SimulationRigidObjectBinding

Explicit binding for one simulation rigid object.

Methods:

__init__(entity_id, simulation_uid[, ...])

__init__(entity_id, simulation_uid, aliases=(), dynamics=SceneDynamics.UNKNOWN, collision_role=SceneCollisionRole.NONE, semantic_type=None, default_grasp_affordance=None, geometry_provider=None)

class embodichain.lab.task_program.integrations.SimulationArticulationBinding

Explicit binding for one simulation articulation.

Methods:

__init__(entity_id, simulation_uid[, ...])

__init__(entity_id, simulation_uid, aliases=(), dynamics=SceneDynamics.UNKNOWN, collision_role=SceneCollisionRole.NONE, semantic_type=None, geometry_provider=None)

class embodichain.lab.task_program.integrations.SimulationArticulationLinkBinding

Explicit canonical link backed by one native articulation link.

Methods:

__init__(entity_id, articulation_id, ...[, ...])

__init__(entity_id, articulation_id, native_link_name, aliases=(), dynamics=SceneDynamics.UNKNOWN, semantic_type=None)

class embodichain.lab.task_program.integrations.AntipodalGraspAffordanceBinding

Build one antipodal grasp affordance from a selected rigid-object mesh.

internal_axis upgrades the payload to AxisAlignAffordance while preserving the same antipodal grasp capability. The axis is expressed in the rigid object’s local frame and can be reused by held-object skills such as Pour.

Methods:

__init__(entity_id, object_id, native_name, ...)

__init__(entity_id, object_id, native_name, revision, aliases=(), relative_pose=(1.0, 0.0, 0.0, 0.0, 0.0, 1.0, 0.0, 0.0, 0.0, 0.0, 1.0, 0.0, 0.0, 0.0, 0.0, 1.0), mesh_env_id=0, internal_axis=None)

class embodichain.lab.task_program.integrations.ContainerAffordanceBinding

Declare one exact object target frame inside a container parent.

Parameters:

• entity_id (str) – Canonical ID of the placement affordance.

• parent_id (str) – Canonical object, articulation, or link parent ID.

• native_name (str) – Stable native name of this target frame.

• aliases (tuple[str, ...]) – Optional non-authoritative lookup aliases.

• object_target_pose (tuple[float, ...]) – Desired object pose relative to the parent.

• minimum_confidence (float) – Minimum parent/affordance observation confidence.

• is_default (bool) – Whether this is the parent’s default Place(inside=...) frame.

Methods:

__init__(entity_id, parent_id, native_name)

__init__(entity_id, parent_id, native_name, aliases=(), object_target_pose=(1.0, 0.0, 0.0, 0.0, 0.0, 1.0, 0.0, 0.0, 0.0, 0.0, 1.0, 0.0, 0.0, 0.0, 0.0, 1.0), minimum_confidence=0.0, is_default=False)

class embodichain.lab.task_program.integrations.SupportSurfaceAffordanceBinding

Declare one exact object target frame on a support parent.

Parameters:

• entity_id (str) – Canonical ID of the placement affordance.

• parent_id (str) – Canonical object, articulation, or link parent ID.

• native_name (str) – Stable native name of this target frame.

• aliases (tuple[str, ...]) – Optional non-authoritative lookup aliases.

• object_target_pose (tuple[float, ...]) – Desired object pose relative to the parent.

• minimum_confidence (float) – Minimum parent/affordance observation confidence.

• is_default (bool) – Whether this is the parent’s default Place(on=...) frame.

Methods:

__init__(entity_id, parent_id, native_name)

__init__(entity_id, parent_id, native_name, aliases=(), object_target_pose=(1.0, 0.0, 0.0, 0.0, 0.0, 1.0, 0.0, 0.0, 0.0, 0.0, 1.0, 0.0, 0.0, 0.0, 0.0, 1.0), minimum_confidence=0.0, is_default=False)

class embodichain.lab.task_program.integrations.ControlPartCommandPreset

Named one-dimensional joint commands for one exact control part.

Methods:

__init__(preset_id, control_part, commands)
build(*, control_dof)	Build a command profile after validating the native control width.
declare()	Build a provider-free command profile from declared tuple widths.

__init__(preset_id, control_part, commands)

build(*, control_dof)

Build a command profile after validating the native control width.

Return type:

ControlPartCommandProfile

declare()

Build a provider-free command profile from declared tuple widths.

Return type:

ControlPartCommandProfile

class embodichain.lab.task_program.integrations.ControlPartEndpointBinding

Profile endpoint backed by one explicit robot control part.

Methods:

__init__(endpoint_id, control_part, capabilities)
build(robot)	Build a joint-backed endpoint after native control-part validation.
declare()	Return the endpoint contract without reading a robot.

__init__(endpoint_id, control_part, capabilities, command_preset=None)

build(robot)

Build a joint-backed endpoint after native control-part validation.

Return type:

ResourceEndpoint

declare()

Return the endpoint contract without reading a robot.

Return type:

ResourceEndpoint

class embodichain.lab.task_program.integrations.ControlPartResourceBinding

Joint-backed robot resource containing control-part endpoints.

Methods:

__init__(resource_id[, endpoints, members])
build(robot)	Build a resource containing strictly validated control-part endpoints.
declare()	Return the resource graph without reading native control parts.

__init__(resource_id, endpoints=(), members=())

build(robot)

Build a resource containing strictly validated control-part endpoints.

Return type:

RobotResource

declare()

Return the resource graph without reading native control parts.

Return type:

RobotResource

class embodichain.lab.task_program.integrations.SimulationRobotSkillProfileBinding

Build a profile from typed resources with strict native validation.

Methods:

__init__(profile_id, resources[, ...])
build(robot)	Build the existing profile after validating every typed resource.
declare()	Project the complete provider-free robot skill profile.

__init__(profile_id, resources, command_presets=(), defaults=<factory>, presets=(), default_preset=None, skill_presets=<factory>, grounding_providers=<factory>)

build(robot)

Build the existing profile after validating every typed resource.

Parameters:

robot (Robot) – Live robot selected by the simulation factory.

Return type:

RobotSkillProfile

Returns:

Reusable, engine-independent robot skill profile.

declare()

Project the complete provider-free robot skill profile.

Return type:

RobotSkillProfile

class embodichain.lab.task_program.integrations.SimulationTaskProgramFactory

Build every live Task Program component from one task registration.

Parameters:

• simulation (SimulationManager) – Exact live simulation that owns robot and scene UIDs.

• robot (Robot) – Exact robot selected for planning and evidence acquisition.

• registration (SimulationTaskProgramRegistration) – Frozen scene, profile, and extension declarations.

• step_dt (float) – Authoritative Gym control cadence.

• planner_cfg (BasePlannerCfg | None) – Explicit planner configuration. None selects TOPPRA for robot.uid.

• motion_generator_factory (Callable[[], MotionGenerator] | None) – Optional fresh-generator factory. It is mutually exclusive with planner_cfg and intended for custom planners and isolated tests.

• grasp_pose_generators (Mapping[str, GraspPoseGenerator] | None) – Standalone grasp-pose services keyed by grasp endpoint target ID, normally a robot control-part name.

• translation_threshold (float) – Material scene translation threshold.

• rotation_threshold (float) – Material scene rotation threshold.

The Gym cadence is attached to each live PlanningContext. Motion policy presets describe behavior; they do not own the environment’s command period.

Methods:

__init__(simulation, robot, registration, *, ...)
create_adapter()	Create the exact Gym adapter with shared simulation policy ports.
create_atomic_action_engine(profile)	Create a fresh engine around the selected planner and exact profile.
create_effect_evidence_providers(*, ...)	Create built-in control-part and articulation evidence providers.
create_parallel_command_safety_validator(*, ...)	Create the registration-owned validator for this runtime assembly.
create_planning_observation_provider(*, ...)	Create one planning port and planner-validated shared scene provider.
create_registered_semantic_lowerers(*, ...)	Create fresh registration-owned lowerers for this runtime assembly.
create_robot_skill_profile()	Build the robot profile used to lower calls to Atomic Skills.
create_scene_registry()	Build one fresh authoritative registry from the task registration.
from_environment(environment, *, registration)	Create a registered factory from the standard Gym environment surface.
registration_owned_segment_policy_ports()	Return registration-owned post-policy and validator ports.

Attributes:

robot_profile_id	Return the exact configured robot-profile ID.
scene_registry_id	Return the exact configured scene-registry ID.
segment_policy_port	Return the shared simulation post-policy and validator port.
step_dt	Return the authoritative Gym control cadence.
task_program_registration	Return the exact standard registration owned by this factory.

__init__(simulation, robot, registration, *, step_dt, planner_cfg=None, motion_generator_factory=None, grasp_pose_generators=None, translation_threshold=0.0001, rotation_threshold=0.001)

create_adapter()

Create the exact Gym adapter with shared simulation policy ports.

Return type:

TaskProgramEnvironmentAdapter

create_atomic_action_engine(profile)

Create a fresh engine around the selected planner and exact profile.

Return type:

AtomicActionEngine

create_effect_evidence_providers(*, scene_registry, engine, observation_provider)

Create built-in control-part and articulation evidence providers.

Return type:

Iterable[EffectEvidenceProvider]

create_parallel_command_safety_validator(*, scene_registry, engine, observation_provider)

Create the registration-owned validator for this runtime assembly.

Return type:

ParallelCommandSafetyValidator

create_planning_observation_provider(*, scene_registry, engine, clock)

Create one planning port and planner-validated shared scene provider.

Return type:

PlanningObservationPort

create_registered_semantic_lowerers(*, scene_registry, engine)

Create fresh registration-owned lowerers for this runtime assembly.

Return type:

tuple[RegisteredSemanticLowerer, ...]

create_robot_skill_profile()

Build the robot profile used to lower calls to Atomic Skills.

Return type:

RobotSkillProfile

create_scene_registry()

Build one fresh authoritative registry from the task registration.

Return type:

SceneRegistry

classmethod from_environment(environment, *, registration, planner_cfg=None, motion_generator_factory=None, grasp_pose_generators=None, translation_threshold=0.0001, rotation_threshold=0.001)

Create a registered factory from the standard Gym environment surface.

Return type:

SimulationTaskProgramFactory

registration_owned_segment_policy_ports()

Return registration-owned post-policy and validator ports.

Return type:

tuple[SimulationSegmentPolicyPort, SimulationSegmentPolicyPort]

property robot_profile_id: str

Return the exact configured robot-profile ID.

property scene_registry_id: str

Return the exact configured scene-registry ID.

property segment_policy_port: SimulationSegmentPolicyPort

Return the shared simulation post-policy and validator port.

property step_dt: float

Return the authoritative Gym control cadence.

property task_program_registration: SimulationTaskProgramRegistration

Return the exact standard registration owned by this factory.

class embodichain.lab.task_program.integrations.SimulationTaskProgramAdapterFactory

Create one standard simulation adapter after an environment is initialized.

Parameters:

• registration (SimulationTaskProgramRegistration) – Immutable provider-free task integration.

• grasp_pose_generator_factories (Mapping[str, Callable[[], GraspPoseGenerator]] | None) – Zero-argument factories keyed by runtime grasp endpoint target ID. Each environment receives fresh services.

Methods:

__init__(registration, *[, ...])
create_adapter(environment)	Create fresh grasp services and bind the initialized environment.

Attributes:

registration

Return the exact static registration owned by this factory.

__init__(registration, *, grasp_pose_generator_factories=None)

create_adapter(environment)

Create fresh grasp services and bind the initialized environment.

Return type:

TaskProgramEnvironmentAdapter

property registration: SimulationTaskProgramRegistration

Return the exact static registration owned by this factory.

class embodichain.lab.task_program.integrations.ConfiguredHandOverPoseProvider

Resolve a hand-over delivery target from immutable configuration.

The provider carries object-space poses rather than arm trajectories. The shared semantic compiler and atomic HandOver implementation remain responsible for grasp selection, IK, motion generation, transfer, release, and delivery. Keeping the numeric declaration as tuple fields also makes the provider suitable for task-registration catalog fingerprinting.

Parameters:

• final_position (tuple[float, float, float]) – World-frame object delivery position.

• final_quaternion_wxyz (tuple[float, float, float, float]) – World-frame object delivery orientation.

Methods:

__init__(final_position, final_quaternion_wxyz)
resolve(call, *, context, bound)	Return an independently owned object-space delivery target.

__init__(final_position, final_quaternion_wxyz)

resolve(call, *, context, bound)

Return an independently owned object-space delivery target.

Parameters:

• call (HandOver) – Canonical hand-over semantic call.

• context (PlanningContext) – Latest immutable planning observation.

• bound (BoundSemanticCall) – Engine/profile-bound hand-over call.

Return type:

HandOverPoseTargets

Returns:

Configured final object-space target.

class embodichain.lab.task_program.integrations.CuroboParallelCommandSafetyValidator

Validate the exact synchronized joint segment before transport dispatch.

Parameters:

• robot (object) – Live robot supplying measured joint state and control-part IDs.

• motion_generator (MotionGenerator) – Runtime motion generator backed by exact cuRobo.

• scene_registry (SceneRegistry) – Authoritative live collision-scene registry.

• validation_control_part (str) – Aggregate control part for merged commands.

• max_joint_step (float) – Maximum displacement between collision samples.

• max_interpolation_samples (int) – Fail-closed sample-count upper bound.

Methods:

__init__(*, robot, motion_generator, ...)
validate(*, branch_frames, merged_frame)	Reject a merged command whose exact interpolated segment collides.

__init__(*, robot, motion_generator, scene_registry, validation_control_part, max_joint_step, max_interpolation_samples)

validate(*, branch_frames, merged_frame)

Reject a merged command whose exact interpolated segment collides.

Return type:

None

class embodichain.lab.task_program.integrations.CuroboParallelSafetyValidatorFactory

Create exact-sample collision gates for one aggregate control part.

validation_control_part must contain every joint that any parallel branch can command. A common dual-arm example is "dual_arm". The cuRobo model for that part remains the authoritative bounds, self-collision, and world-collision model.

Parameters:

• validation_control_part (str) – Aggregate robot control part containing every joint that a parallel lane may command.

• max_joint_step (float) – Maximum absolute joint displacement between collision samples in radians or the joint’s native linear unit.

• max_interpolation_samples (int) – Fail-closed upper bound on samples per frame.

Methods:

__init__(validation_control_part[, ...])
create(*, simulation, robot, scene_registry, ...)	Create one fresh validator bound to the assembled live runtime.

__init__(validation_control_part, max_joint_step=0.025, max_interpolation_samples=256)

create(*, simulation, robot, scene_registry, engine)

Create one fresh validator bound to the assembled live runtime.

Return type:

CuroboParallelCommandSafetyValidator

class embodichain.lab.task_program.integrations.SimulationSegmentPolicyPort

Execute built-in segment policies against explicitly bound simulation data.

Parameters:

• simulation (SimulationManager) – Live simulation used only for UIDs declared in scene_binding.

• robot (Robot) – Live robot used to produce full target-qpos holds while the post-policy observes settling.

• scene_binding (SimulationSceneBinding) – Exact canonical-to-native scene declaration.

• settle_presets (Mapping[str, DynamicSettleMonitorCfg] | None) – Named settling policies. None installs the shared rigid_object and articulation presets.

• env_ids (Tensor | None) – Optional stable logical row IDs. They describe correlation, not simulator row indices; simulator rows remain ordered exactly as returned by the robot and bound entities.

The same instance implements both SegmentPostPolicyPort and SegmentValidatorPort. Unknown policy types, presets, canonical IDs, or native entities fail before an action is emitted.

Methods:

__init__(simulation, robot, scene_binding, *)
actions(policy, *, segment, active_mask)	Yield full target-qpos hold actions until rows settle or time out.
post_policy_metadata(policy, *, segment)	Return the latest JSON-safe settling trace for one policy.
post_policy_result(policy, *, segment)	Return the latest independently owned per-row settling result.
validate(validator, *, segment)	Observe one explicitly bound entity against its validator contract.
validate_policy(policy, *, segment)	Validate one post-policy against static bindings without observation.
validate_validator(validator, *, segment)	Validate one validator against static bindings without observation.
validator_metadata(validator, *, segment)	Return an owned JSON-safe trace for one completed validator.

Attributes:

settle_preset_ids

Return installed post-policy preset IDs in declaration order.

__init__(simulation, robot, scene_binding, *, settle_presets=None, env_ids=None)

actions(policy, *, segment, active_mask)

Yield full target-qpos hold actions until rows settle or time out.

Parameters:

• policy (Any) – Exact compiled wait_stable policy.

• segment (Any) – Exact segment that owns policy.

• active_mask (Tensor) – Rows that remain eligible after runtime execution and preceding post-policies. Inactive rows are held safely but do not participate in settling, timeout, or success results.

Yields:

Fresh full target-qpos hold commands consumed by ordinary env.step(). Reading drive targets instead of measured joint positions preserves contact preload in position-controlled tools. Rows inactive when the policy starts use fresh measured qpos holds.

Return type:

Iterator[Tensor]

Timeout is a normal row-local result boundary. Timed-out rows are exposed through post_policy_result() and post_policy_metadata(); no batch-level exception is raised.

post_policy_metadata(policy, *, segment)

Return the latest JSON-safe settling trace for one policy.

The trace is available after the policy generator has started. A terminal trace has status "settled", "timed_out", or "skipped" when no rows remain active; an early demo interruption intentionally retains the latest "running" snapshot for diagnosis.

Return type:

Mapping[str, object]

post_policy_result(policy, *, segment)

Return the latest independently owned per-row settling result.

Return type:

Tensor

property settle_preset_ids: tuple[str, ...]

Return installed post-policy preset IDs in declaration order.

validate(validator, *, segment)

Observe one explicitly bound entity against its validator contract.

Parameters:

• validator (Any) – Exact compiled segment validator.

• segment (Any) – Exact segment that owns validator.

Return type:

Tensor

Returns:

Boolean tensor with one result per simulation row.

validate_policy(policy, *, segment)

Validate one post-policy against static bindings without observation.

This method reads only the compiled declaration, installed preset table, and entities resolved when the port was constructed. It never samples velocity or qpos and never emits a controller action.

Return type:

None

validate_validator(validator, *, segment)

Validate one validator against static bindings without observation.

Return type:

None

validator_metadata(validator, *, segment)

Return an owned JSON-safe trace for one completed validator.

Return type:

Mapping[str, object]

embodichain.lab.task_program.integrations.create_simulation_task_program_adapter(environment, *, registration, planner_cfg=None, motion_generator_factory=None, grasp_pose_generators=None, translation_threshold=0.0001, rotation_threshold=0.001)

Create a complete production adapter from one standard Gym environment.

This is the intended task-side one-line integration. Standard tasks pass one immutable registration that owns their pre-simulation catalog and every runtime extension.

Parameters:

• environment (SimulationTaskProgramEnvironment) – Standard Gym simulation environment exposing sim, robot, and step_dt.

• registration (SimulationTaskProgramRegistration) – Required immutable task registration.

• planner_cfg (BasePlannerCfg | None) – Optional planner configuration owned by the factory.

• motion_generator_factory (Callable[[], MotionGenerator] | None) – Optional factory for one fresh motion generator.

• grasp_pose_generators (Mapping[str, GraspPoseGenerator] | None) – Standalone grasp-pose services keyed by grasp endpoint target ID.

• translation_threshold (float) – Scene translation revision threshold.

• rotation_threshold (float) – Scene rotation revision threshold.

Return type:

TaskProgramEnvironmentAdapter

Returns:

Complete production Task Program environment adapter.

Implementation module exports

CuroboParallelCommandSafetyValidator	Validate the exact synchronized joint segment before transport dispatch.
CuroboParallelSafetyValidatorFactory	Create exact-sample collision gates for one aggregate control part.

TaskProgramIntegrationCatalog	Provider-free integration directory owned by one task registration.
IntegrationFingerprintMismatch	Raised when a live integration no longer matches its registration.
SimulationTaskProgramRegistration	Exact immutable task-owned simulation integration registration.

ControlPartEvidenceProviderDeclaration	Provider-free identity of one control-part evidence factory.
ControlPartEvidenceProviderFactory	Registration-owned factory for one live control-part evidence provider.
EndpointAdapterDeclaration	Provider-free declaration of one exact endpoint adapter.
ParallelCommandSafetyValidatorFactory	Registration-owned factory for one authoritative live safety gate.
ParallelSafetyDeclaration	Provider-free identity and transport coverage of one safety factory.
RegisteredSemanticLowererDeclaration	Provider-free identity of one registered semantic lowerer factory.
RegisteredSemanticLowererFactory	Registration-owned factory for one fresh live semantic lowerer.
RuntimeTransportDeclaration	Provider-free declaration of one ordered runtime transport encoder.
StandardExtensionDeclarations	Cross-checked provider-free declarations for the standard factory.
VersionedKey	Exact (provider_or_projector_id, revision) registry key.
build_standard_extension_declarations(*, ...)	Cross-check standard-runtime extensions against one exact profile.
declare_control_part_evidence_factory(factory)	Read one control-part evidence factory's exact static identity.
declare_endpoint_adapter(adapter)	Read one endpoint adapter's exact static extension contract.
declare_parallel_safety_factory(factory)	Read one safety factory's exact static identity and coverage.
declare_registered_semantic_lowerer_factory(factory)	Read one semantic lowerer factory's exact static identity.
declare_runtime_transport(transport)	Read one runtime encoder's exact static target/payload contract.
validate_immutable_extension_declaration(...)	Accept only a deeply immutable frozen dataclass or stateless instance.

embodichain.lab.task_program.integrations.extensions.build_standard_extension_declarations(*, profile, endpoint_adapters, runtime_transports, parallel_safety_factory, control_part_evidence_factory=None, registered_semantic_lowerer_factories=())

Cross-check standard-runtime extensions against one exact profile.

The built-in control-part adapter and joint-position transport cannot be overridden. They are installed first only when the profile uses a ControlPartEndpoint; a pure-custom profile contains only its custom declarations. Custom adapters and transports must cover exactly the endpoint types and transport IDs used by the registered profile; unused declarations fail closed.

Return type:

StandardExtensionDeclarations

embodichain.lab.task_program.integrations.extensions.declare_control_part_evidence_factory(factory)

Read one control-part evidence factory’s exact static identity.

Parameters:

factory (ControlPartEvidenceProviderFactory) – Immutable registration-owned factory declaration.

Return type:

ControlPartEvidenceProviderDeclaration

Returns:

Provider-free exact factory identity.

embodichain.lab.task_program.integrations.extensions.declare_endpoint_adapter(adapter)

Read one endpoint adapter’s exact static extension contract.

Return type:

EndpointAdapterDeclaration

embodichain.lab.task_program.integrations.extensions.declare_parallel_safety_factory(factory)

Read one safety factory’s exact static identity and coverage.

Return type:

ParallelSafetyDeclaration

embodichain.lab.task_program.integrations.extensions.declare_registered_semantic_lowerer_factory(factory)

Read one semantic lowerer factory’s exact static identity.

Parameters:

factory (RegisteredSemanticLowererFactory) – Immutable registration-owned lowerer factory.

Return type:

RegisteredSemanticLowererDeclaration

Returns:

Provider-free call, revision, and factory-type declaration.

Raises:

• TypeError – If the factory has no callable creator or is mutable.

• ValueError – If its call ID or revision is not an exact identifier.

embodichain.lab.task_program.integrations.extensions.declare_runtime_transport(transport)

Read one runtime encoder’s exact static target/payload contract.

Return type:

RuntimeTransportDeclaration

embodichain.lab.task_program.integrations.extensions.validate_immutable_extension_declaration(value, *, field_name)

Accept only a deeply immutable frozen dataclass or stateless instance.

Frozen dataclass fields may contain only immutable scalar values, types, enums with immutable values, exact tuples, exact frozensets, and recursively frozen dataclasses. Mutable leaves such as mappings, lists, sets, bytearrays, and tensors are rejected because registration-owned live extensions are shared with an assembled runtime. A non-dataclass extension must not have instance or slot state at all.

Return type:

None