embodichain.lab.task_program.runtime

Provider-independent Semantic Call execution and parallel scheduling.

Runtime services execute lowered Semantic Calls and synchronize parallel branches. They do not own Gym episode lifecycle, env.step(), or trajectory recording.

ParallelSemanticExecutor	Run independent JIT semantic lanes on one synchronized command grid.
ParallelTimingPolicy	Strict environment-grid policy for one parallel barrier.
SemanticCallExecutor	JIT-ground and execute semantic calls through one runner per call.
SemanticExecutionResult	Immutable workflow snapshot returned by sync and step-wise execution.
SemanticExecutionStatus	Lifecycle state of one semantic workflow run.

Classes:

ParallelSemanticExecutor	Run independent JIT semantic lanes on one synchronized command grid.
ParallelTimingPolicy	Strict environment-grid policy for one parallel barrier.
SemanticCallExecutor	JIT-ground and execute semantic calls through one runner per call.
SemanticExecutionResult	Immutable workflow snapshot returned by sync and step-wise execution.
SemanticExecutionStatus	Lifecycle state of one semantic workflow run.

class embodichain.lab.task_program.runtime.ParallelSemanticExecutor

Run independent JIT semantic lanes on one synchronized command grid.

Schema v2 deliberately uses conservative barrier ownership: branches are not assigned disjoint environment-row partitions, so two branches that write the same symbolic key conflict for the complete started batch even when their observed value masks happen to be disjoint. A future schema may add explicit row partitioning before relaxing this invariant.

A lane completion hold is forwarded as an explicit grid action. Other lanes therefore receive deterministic hold-padding for that environment step; a merged frame generated in the same coordinator cycle is retained and dispatched only after the clock advances. Branch runners are not stepped while that retained frame is being dispatched. This keeps the physical order observed hold -> next command and limits every normal coordinator step to one action-producing transport operation.

Methods:

__init__(branches, command_sink, clock, ...)
cancel([reason])	Cancel every lane and forward one target-scoped transport cancel.
from_template(template_runtime, ...[, ...])	Analyze claims and derive independent lanes from one runtime.
start(*[, workflow_id, eligible_mask])	Start all lanes from the same verified barrier state.
step()	Advance one deterministic coordinator state-machine transition.

Attributes:

branch_claims	Return immutable statically analyzed claims in branch order.
clock	Return the exact clock shared by the coordinator and every lane.
result	Return an owned barrier snapshot.
runner_cfg	Return an owned copy of the coordinator transport policy.

__init__(branches, command_sink, clock, timing_policy, safety_validator, *, timeout_steps, failure_policy='fail_fast', runner_cfg=None)

property branch_claims: Mapping[str, ResourceClaim]

Return immutable statically analyzed claims in branch order.

cancel(reason='Parallel workflow cancelled by caller.')

Cancel every lane and forward one target-scoped transport cancel.

Return type:

ParallelSemanticExecutionResult

property clock: ExecutionClock

Return the exact clock shared by the coordinator and every lane.

classmethod from_template(template_runtime, branch_calls, command_sink, timing_policy, safety_validator, *, timeout_steps, failure_policy='fail_fast', runner_cfg=None, workflow_id='parallel_static_analysis', branch_paths=None)

Analyze claims and derive independent lanes from one runtime.

This factory deliberately accepts semantic calls instead of compiled Gym-program types. It keeps the simulation runtime independent of the higher-level configuration package while giving every frontend one canonical resource-conflict and lane-construction path.

Parameters:

• template_runtime (SemanticCallExecutor) – Idle runtime providing shared compiler and ports.

• branch_calls (Mapping[str, tuple[SemanticCallSpec, ...]]) – Ordered branch ID to semantic-call sequence mapping.

• command_sink (CommandSink) – The sole outbound merged command sink.

• timing_policy (ParallelTimingPolicy) – Exact shared environment grid.

• safety_validator (ParallelCommandSafetyValidator) – Required physical/collision safety gate for each synchronized outbound command.

• timeout_steps (int) – Maximum environment steps at the barrier.

• failure_policy (str) – Row-local barrier failure policy.

• runner_cfg (ExecutionRunnerCfg | None) – Shared command timeout, safe-stop, completion-hold, and minimum-cycle policy selected by the runtime preset.

• workflow_id (str) – Stable prefix for provider-free claim analysis.

• branch_paths (Mapping[str, tuple[str | int, ...]] | None) – Optional exact source path for every branch.

Return type:

ParallelSemanticExecutor

Returns:

A one-shot parallel runtime whose branches share no mutable runner state.

property result: ParallelSemanticExecutionResult

Return an owned barrier snapshot.

property runner_cfg: ExecutionRunnerCfg

Return an owned copy of the coordinator transport policy.

start(*, workflow_id='parallel_workflow', eligible_mask=None)

Start all lanes from the same verified barrier state.

Return type:

ParallelSemanticExecutionResult

step()

Advance one deterministic coordinator state-machine transition.

Return type:

ParallelSemanticExecutionResult

class embodichain.lab.task_program.runtime.ParallelTimingPolicy

Strict environment-grid policy for one parallel barrier.

Fractional frame durations are rejected. Padding repeats the last controller target, which is a deterministic position/tool hold; no interpolation is hidden inside the scheduler.

Methods:

__init__(step_dt[, tolerance])

__init__(step_dt, tolerance=1e-06)

class embodichain.lab.task_program.runtime.SemanticCallExecutor

JIT-ground and execute semantic calls through one runner per call.

Static workflow analysis occurs once in start(). Each call then gets a fresh observation, one grounded invocation, one execution session, and one ExecutionRunner. Verified task state and row eligibility cross call barriers; execution sessions never do.

Methods:

__init__(compiler, observation_provider, ...)
adopt_verified_task_state(task_state)	Install a verified state snapshot between independent workflows.
cancel([reason])	Cancel the active runner and inherit its cancel-then-hold behavior.
deactivate_rows(env_mask, *, reason)	Cancel selected rows while the remaining shared call keeps running.
fork(command_sink, *[, task_state])	Create an independent execution lane from the same runtime ports.
run(*calls[, workflow_id, eligible_mask, ...])	Synchronously execute an analyzed semantic-call prefix.
start(*calls[, workflow_id, eligible_mask, ...])	Analyze once and prepare the first call without blocking on motion.
step()	Advance the current call by at most one due runner cycle.
validate(calls, *[, workflow_id])	Analyze a workflow without executing it.

Attributes:

clock	Return the shared execution clock used by this runtime.
compiler	Return the installed semantic compiler.
engine	Return the atomic-action engine owned by this runtime.
observation_provider	Return the observation port used for just-in-time grounding.
result	Return an immutable snapshot of the current workflow.
scene_registry	Return the authoritative semantic scene registry.
status	Return the current workflow status.
task_state	Return an owned snapshot of persistent verified task state.

__init__(compiler, observation_provider, command_sink, evidence_collector, *, task_state=None, clock=None, runner_cfg=None)

adopt_verified_task_state(task_state)

Install a verified state snapshot between independent workflows.

Parallel coordinators use this explicit barrier operation after deterministically merging branch-local effects. Running workflows cannot replace their runner-owned state.

Return type:

SemanticExecutionResult

cancel(reason='Semantic workflow cancelled by caller.')

Cancel the active runner and inherit its cancel-then-hold behavior.

Return type:

SemanticExecutionResult

property clock: ExecutionClock

Return the shared execution clock used by this runtime.

Parallel coordinators use the same clock for every derived lane so a branch cannot advance independently of the environment step grid.

property compiler: SemanticCallCompiler

Return the installed semantic compiler.

deactivate_rows(env_mask, *, reason)

Cancel selected rows while the remaining shared call keeps running.

This is the row-local cancellation boundary used by a parallel fail-fast coordinator. The active runner remains the sole owner of controller neutralization and effect-request correlation.

Parameters:

• env_mask (Tensor) – Rows to remove permanently from this workflow.

• reason (str) – Human-readable cancellation reason.

Return type:

SemanticExecutionResult

Returns:

Updated immutable workflow result.

property engine: AtomicActionEngine

Return the atomic-action engine owned by this runtime.

fork(command_sink, *, task_state=None)

Create an independent execution lane from the same runtime ports.

The derived runtime shares the immutable compiler integration, observation/evidence providers, clock, and runner policy, but owns its workflow, runner, masks, and verified task state. Its command sink is supplied explicitly so a parallel coordinator can buffer commands until all lanes have reached the same environment tick.

Parameters:

• command_sink (CommandSink) – Lane-local command sink.

• task_state (TaskState | None) – Optional verified barrier state. The current owned task state is used when omitted.

Return type:

SemanticCallExecutor

Returns:

A new idle semantic runtime for one independent lane.

property observation_provider: ObservationProvider

Return the observation port used for just-in-time grounding.

property result: SemanticExecutionResult

Return an immutable snapshot of the current workflow.

run(*calls, workflow_id='semantic_workflow', eligible_mask=None, execution_prefix_length=None, max_steps=100000, on_step=None)

Synchronously execute an analyzed semantic-call prefix.

Return type:

SemanticExecutionResult

property scene_registry: SceneRegistry

Return the authoritative semantic scene registry.

start(*calls, workflow_id='semantic_workflow', eligible_mask=None, execution_prefix_length=None)

Analyze once and prepare the first call without blocking on motion.

Parameters:

• *calls (SemanticCallSpec | Iterable[SemanticCallSpec]) – Complete ordered semantic analysis window. Calls after the execution prefix participate in static look-ahead but are not grounded or executed by this run.

• workflow_id (str) – Stable workflow identifier used in diagnostics.

• eligible_mask (Tensor | None) – Optional row-local execution eligibility.

• execution_prefix_length (int | None) – Number of leading calls to execute. When omitted, the complete analysis window is executed.

Return type:

SemanticExecutionResult

Returns:

Immutable initial runtime result.

property status: SemanticExecutionStatus

Return the current workflow status.

step()

Advance the current call by at most one due runner cycle.

Return type:

SemanticExecutionResult

property task_state: TaskState

Return an owned snapshot of persistent verified task state.

validate(calls, *, workflow_id='semantic_workflow')

Analyze a workflow without executing it.

Return type:

object

class embodichain.lab.task_program.runtime.SemanticExecutionResult

Immutable workflow snapshot returned by sync and step-wise execution.

Methods:

__init__(status, workflow_id, ...[, events, ...])
require_all_succeeded()	Raise unless every environment completed the workflow successfully.
to_metadata()	Return a fresh deterministic JSON-safe workflow result.

Attributes:

terminal

Whether the workflow no longer accepts execution steps.

__init__(status, workflow_id, current_call_index, env_ids, success_mask, failure_mask, cancelled_mask, eligible_mask, task_state, events=(), calls=(), effects=(), workflow_recoveries=(), failures=(), wait_duration=0.0, message=None)

require_all_succeeded()

Raise unless every environment completed the workflow successfully.

This convenience assertion keeps application entry points concise while the canonical result continues to expose row-local masks.

Return type:

None

property terminal: bool

Whether the workflow no longer accepts execution steps.

to_metadata()

Return a fresh deterministic JSON-safe workflow result.

Core recovery remains represented by the ordered ExecutionEvent stream and each call’s plan-attempt history. Workflow re-acquisition additionally appears in workflow_recoveries. The returned object owns only Python scalars, lists, and dictionaries and can be serialized with json.dumps(..., allow_nan=False).

Return type:

dict[str, object]

class embodichain.lab.task_program.runtime.SemanticExecutionStatus

Lifecycle state of one semantic workflow run.

Methods:

__new__(value)

__new__(value)