armnet-runtime API Reference

armnet_runtime

armnet runtime SDK — used inside customer containers on a cell.

Customer-facing surface: a @main decorator and a :class:Context. The container's entrypoint is the armnet-runtime console script (installed by this package); it loads the user's file, finds the @main-decorated function, builds the context, and calls it.

Typical use::

from armnet_runtime import main, Context

@main
def run(ctx: Context):
    seed = ctx.args.get("seed", 0)
    ctx.report_progress("starting")
    ...
    return {"success_rate": 1.0}

Wire types (Embodiment, Task, JobSpec, JobResult, ...) are re-exported from :mod:armnet_core for ergonomics, so customer code never needs to know about the core package directly.

Embodiment module-attribute

Embodiment = str

Task module-attribute

Task = str

TerminalStatus module-attribute

TerminalStatus = frozenset({JobStatus.SUCCEEDED, JobStatus.FAILED, JobStatus.TIMEOUT, JobStatus.CANCELLED})

BIMANUAL_SO101_EMBODIMENT module-attribute

BIMANUAL_SO101_EMBODIMENT = 'lerobot/bimanual_so101'

SO101_EMBODIMENT module-attribute

SO101_EMBODIMENT = 'lerobot/so-101'

Job

Bases: BaseModel

The orchestrator's view of a job (response body of GET /jobs/{id}).

Source code in core/src/armnet_core/models.py

class Job(BaseModel):
    """The orchestrator's view of a job (response body of GET /jobs/{id})."""

    id: str = Field(default_factory=_new_job_id)
    spec: JobSpec
    status: JobStatus = JobStatus.SUBMITTED
    created_at: datetime = Field(default_factory=_utcnow)
    updated_at: datetime = Field(default_factory=_utcnow)
    cell_id: Optional[str] = Field(
        default=None,
        description="ID of the cell that picked up the job, set on dispatch.",
    )
    result: Optional[JobResult] = None
    dispatched: Optional[bool] = Field(
        default=None,
        description=(
            "Only set on the POST /jobs response: True if the job was dispatched "
            "to a cell immediately, False if it was accepted but queued (no "
            "matching cell was free). None on all other responses. Clients can use "
            "this to decide whether to wait for logs/result or just report 'queued'."
        ),
    )

    def is_terminal(self) -> bool:
        return self.status in TerminalStatus

id class-attribute instance-attribute

id: str = Field(default_factory=_new_job_id)

spec instance-attribute

spec: JobSpec

status class-attribute instance-attribute

status: JobStatus = JobStatus.SUBMITTED

created_at class-attribute instance-attribute

created_at: datetime = Field(default_factory=_utcnow)

updated_at class-attribute instance-attribute

updated_at: datetime = Field(default_factory=_utcnow)

cell_id class-attribute instance-attribute

cell_id: Optional[str] = Field(default=None, description='ID of the cell that picked up the job, set on dispatch.')

result class-attribute instance-attribute

result: Optional[JobResult] = None

dispatched class-attribute instance-attribute

dispatched: Optional[bool] = Field(default=None, description="Only set on the POST /jobs response: True if the job was dispatched to a cell immediately, False if it was accepted but queued (no matching cell was free). None on all other responses. Clients can use this to decide whether to wait for logs/result or just report 'queued'.")

is_terminal

is_terminal() -> bool

Source code in core/src/armnet_core/models.py

def is_terminal(self) -> bool:
    return self.status in TerminalStatus

JobResult

Bases: BaseModel

Terminal result published by a cell on the results subject.

Also returned (embedded in :class:Job) by GET /jobs/{id} once the job is in a terminal state.

Source code in core/src/armnet_core/models.py

class JobResult(BaseModel):
    """Terminal result published by a cell on the results subject.

    Also returned (embedded in :class:`Job`) by ``GET /jobs/{id}`` once the
    job is in a terminal state.
    """

    status: JobStatus = Field(
        ...,
        description="One of the terminal statuses (succeeded/failed/timeout/cancelled).",
    )
    exit_code: Optional[int] = Field(
        default=None,
        description="Container process exit code if the container ran to completion.",
    )
    stdout: str = Field(default="", description="Captured container stdout.")
    stderr: str = Field(default="", description="Captured container stderr.")
    error: Optional[str] = Field(
        default=None,
        description="Short, infra-side reason this job didn't run user code "
        "to completion: image pull failure, docker error, timeout, etc. "
        "Mutually exclusive with `traceback` in practice (one is a platform "
        "failure, the other is a user-code failure).",
    )
    traceback: Optional[str] = Field(
        default=None,
        description="Python traceback from the customer's `@main`-decorated "
        "function if it raised. Extracted by the cell from the "
        "`[armnet:traceback]:json` marker line in stdout. Capped at "
        "~64 KiB on the cell side; the full untruncated traceback is also "
        "in `stderr` for power users. None for successful jobs and for "
        "infra-side failures (those go in `error` instead).",
    )
    return_value: Optional[Any] = Field(
        default=None,
        description="Value returned by the customer's `@main`-decorated "
        "function. Extracted by the cell from the marker line that the "
        "`armnet-runtime` entrypoint prints to stdout. None if the "
        "function returned None or did not run to completion.",
    )
    started_at: Optional[datetime] = None
    finished_at: Optional[datetime] = None

    def raise_for_status(self) -> None:
        """Raise :class:`RemoteExecutionError` iff this result isn't ``SUCCEEDED``.

        The httpx-style "opt-in raising" pattern. Use it when you'd rather
        bail than branch on ``result.status``::

            result = execute(...)
            result.raise_for_status()
            do_thing(result.return_value)

        For successful results this is a no-op.
        """

        if self.status != JobStatus.SUCCEEDED:
            raise RemoteExecutionError(self)

    def __str__(self) -> str:
        """Human-readable rendering. Uses indentation to make tracebacks scannable.

        ``print(result)`` is intended to be the one-liner that tells you
        what happened. ``repr(result)`` (pydantic's default) still shows
        every field for debugging.
        """

        lines: list[str] = []
        head = f"JobResult(status={self.status.value}"
        if self.exit_code is not None:
            head += f", exit_code={self.exit_code}"
        head += ")"
        lines.append(head)
        if self.return_value is not None:
            lines.append(f"  return_value: {self.return_value!r}")
        if self.error:
            lines.append(f"  error: {self.error}")
        if self.traceback:
            lines.append("  traceback (from @main):")
            for tb_line in self.traceback.splitlines():
                lines.append(f"    {tb_line}")
        return "\n".join(lines)

status class-attribute instance-attribute

status: JobStatus = Field(..., description='One of the terminal statuses (succeeded/failed/timeout/cancelled).')

exit_code class-attribute instance-attribute

exit_code: Optional[int] = Field(default=None, description='Container process exit code if the container ran to completion.')

stdout class-attribute instance-attribute

stdout: str = Field(default='', description='Captured container stdout.')

stderr class-attribute instance-attribute

stderr: str = Field(default='', description='Captured container stderr.')

error class-attribute instance-attribute

error: Optional[str] = Field(default=None, description="Short, infra-side reason this job didn't run user code to completion: image pull failure, docker error, timeout, etc. Mutually exclusive with `traceback` in practice (one is a platform failure, the other is a user-code failure).")

traceback class-attribute instance-attribute

traceback: Optional[str] = Field(default=None, description="Python traceback from the customer's `@main`-decorated function if it raised. Extracted by the cell from the `[armnet:traceback]:json` marker line in stdout. Capped at ~64 KiB on the cell side; the full untruncated traceback is also in `stderr` for power users. None for successful jobs and for infra-side failures (those go in `error` instead).")

return_value class-attribute instance-attribute

return_value: Optional[Any] = Field(default=None, description="Value returned by the customer's `@main`-decorated function. Extracted by the cell from the marker line that the `armnet-runtime` entrypoint prints to stdout. None if the function returned None or did not run to completion.")

started_at class-attribute instance-attribute

started_at: Optional[datetime] = None

finished_at class-attribute instance-attribute

finished_at: Optional[datetime] = None

raise_for_status

raise_for_status() -> None

Raise :class:RemoteExecutionError iff this result isn't SUCCEEDED.

The httpx-style "opt-in raising" pattern. Use it when you'd rather bail than branch on result.status::

result = execute(...)
result.raise_for_status()
do_thing(result.return_value)

For successful results this is a no-op.

Source code in core/src/armnet_core/models.py

def raise_for_status(self) -> None:
    """Raise :class:`RemoteExecutionError` iff this result isn't ``SUCCEEDED``.

    The httpx-style "opt-in raising" pattern. Use it when you'd rather
    bail than branch on ``result.status``::

        result = execute(...)
        result.raise_for_status()
        do_thing(result.return_value)

    For successful results this is a no-op.
    """

    if self.status != JobStatus.SUCCEEDED:
        raise RemoteExecutionError(self)

JobSpec

Bases: BaseModel

The fields a client supplies when creating a job.

This is the body of POST /jobs. The orchestrator wraps it into a :class:Job (assigning id, status, timestamps) before persisting.

Source code in core/src/armnet_core/models.py

class JobSpec(BaseModel):
    """The fields a client supplies when creating a job.

    This is the body of ``POST /jobs``. The orchestrator wraps it into a
    :class:`Job` (assigning ``id``, ``status``, timestamps) before
    persisting.
    """

    image: str = Field(
        ...,
        description="Fully-qualified container image reference, e.g. "
        "`ghcr.io/my-org/my-image:tag` or `my-image:latest` for local M0.5.",
    )
    args: dict[str, Any] = Field(
        default_factory=dict,
        description="Keyword arguments passed to the customer's @main-decorated "
        "function as `ctx.args`. JSON-encoded into the `ARMNET_ARGS` env "
        "var by the cell; decoded by the `armnet-runtime` entrypoint "
        "before calling user code. Must be JSON-serialisable.",
    )
    embodiment: Embodiment = Field(
        ...,
        description="Required robot embodiment. The orchestrator routes the "
        "job onto the NATS subject for this embodiment+task pair, where the "
        "matching cell picks it up.",
    )
    task: Optional[Task] = Field(
        default=None,
        description="Optional task. When set, only cells configured for this "
        "(embodiment, task) pair run the job. When omitted, any cell of the "
        "embodiment may pick it up regardless of its configured task.",
    )
    timeout_seconds: int = Field(
        default=120,
        ge=1,
        description="Wall-clock cap on container execution.",
    )
    secrets: dict[str, str] = Field(
        default_factory=dict,
        description=(
            "Mapping of environment variable name to user secret name. "
            "Example: {'HF_TOKEN': 'huggingface-token'} resolves the "
            "authenticated user's secret and injects it as HF_TOKEN."
        ),
    )
    detach: bool = Field(
        default=False,
        description=(
            "If false, losing the client log WebSocket requests graceful job "
            "cancellation. If true, the job keeps running after client disconnect."
        ),
    )
    username: Optional[str] = Field(
        default=None,
        description=(
            "Authenticated armnet username. Set by the orchestrator from "
            "the API key; clients should not rely on supplied values being preserved."
        ),
    )

image class-attribute instance-attribute

image: str = Field(..., description='Fully-qualified container image reference, e.g. `ghcr.io/my-org/my-image:tag` or `my-image:latest` for local M0.5.')

args class-attribute instance-attribute

args: dict[str, Any] = Field(default_factory=dict, description="Keyword arguments passed to the customer's @main-decorated function as `ctx.args`. JSON-encoded into the `ARMNET_ARGS` env var by the cell; decoded by the `armnet-runtime` entrypoint before calling user code. Must be JSON-serialisable.")

embodiment class-attribute instance-attribute

embodiment: Embodiment = Field(..., description='Required robot embodiment. The orchestrator routes the job onto the NATS subject for this embodiment+task pair, where the matching cell picks it up.')

task class-attribute instance-attribute

task: Optional[Task] = Field(default=None, description='Optional task. When set, only cells configured for this (embodiment, task) pair run the job. When omitted, any cell of the embodiment may pick it up regardless of its configured task.')

timeout_seconds class-attribute instance-attribute

timeout_seconds: int = Field(default=120, ge=1, description='Wall-clock cap on container execution.')

secrets class-attribute instance-attribute

secrets: dict[str, str] = Field(default_factory=dict, description="Mapping of environment variable name to user secret name. Example: {'HF_TOKEN': 'huggingface-token'} resolves the authenticated user's secret and injects it as HF_TOKEN.")

detach class-attribute instance-attribute

detach: bool = Field(default=False, description='If false, losing the client log WebSocket requests graceful job cancellation. If true, the job keeps running after client disconnect.')

username class-attribute instance-attribute

username: Optional[str] = Field(default=None, description='Authenticated armnet username. Set by the orchestrator from the API key; clients should not rely on supplied values being preserved.')

JobStatus

Bases: str, Enum

Job lifecycle states.

Mirrors the design doc §3.5 state machine. A job is created SUBMITTED; if the target cell isn't available it becomes QUEUED (run when the cell next comes online — the scheduler is a later phase), otherwise it is DISPATCHED to a cell, then RUNNING, then a terminal state.

Source code in core/src/armnet_core/models.py

class JobStatus(str, Enum):
    """Job lifecycle states.

    Mirrors the design doc §3.5 state machine. A job is created ``SUBMITTED``;
    if the target cell isn't available it becomes ``QUEUED`` (run when the cell
    next comes online — the
    scheduler is a later phase), otherwise it is ``DISPATCHED`` to a cell, then
    ``RUNNING``, then a terminal state.
    """

    SUBMITTED = "submitted"
    QUEUED = "queued"
    DISPATCHED = "dispatched"
    RUNNING = "running"
    SUCCEEDED = "succeeded"
    FAILED = "failed"
    TIMEOUT = "timeout"
    CANCELLED = "cancelled"

SUBMITTED class-attribute instance-attribute

SUBMITTED = 'submitted'

QUEUED class-attribute instance-attribute

QUEUED = 'queued'

DISPATCHED class-attribute instance-attribute

DISPATCHED = 'dispatched'

RUNNING class-attribute instance-attribute

RUNNING = 'running'

SUCCEEDED class-attribute instance-attribute

SUCCEEDED = 'succeeded'

FAILED class-attribute instance-attribute

FAILED = 'failed'

TIMEOUT class-attribute instance-attribute

TIMEOUT = 'timeout'

CANCELLED class-attribute instance-attribute

CANCELLED = 'cancelled'

RemoteExecutionError

Bases: RuntimeError

Raised by :meth:JobResult.raise_for_status for non-succeeded results.

The exception's __str__ includes the underlying status, any platform-side error, and the user's traceback (if any), so an unhandled raise prints all the diagnostic context an operator needs. The original :class:JobResult is available as :attr:result for structured access.

Source code in core/src/armnet_core/models.py

class RemoteExecutionError(RuntimeError):
    """Raised by :meth:`JobResult.raise_for_status` for non-succeeded results.

    The exception's ``__str__`` includes the underlying status, any
    platform-side ``error``, and the user's ``traceback`` (if any), so an
    unhandled raise prints all the diagnostic context an operator needs.
    The original :class:`JobResult` is available as :attr:`result` for
    structured access.
    """

    def __init__(self, result: "JobResult") -> None:
        self.result = result
        super().__init__(str(result))

result instance-attribute

result = result

Cell dataclass

Handle to the physical cell the user code is running on.

M0.5 stub: there is no real cell yet, so robot_port is always None and :meth:reset is a no-op. The shape is fixed now so the spec example compiles end-to-end and so M2/M3 can fill in the implementation without touching customer-facing imports.

Source code in runtime/src/armnet_runtime/context.py

@dataclass
class Cell:
    """Handle to the physical cell the user code is running on.

    M0.5 stub: there is no real cell yet, so ``robot_port`` is always ``None``
    and :meth:`reset` is a no-op. The shape is fixed now so the spec
    example compiles end-to-end and so M2/M3 can fill in the
    implementation without touching customer-facing imports.
    """

    robot_port: Optional[str] = None
    """Robot port value to pass into LeRobot robot configs.

    In container-backed remote execution this is the connector endpoint, not
    the host's physical serial path. The SDK's import-system swap routes that
    endpoint through the cell-side connector, which then opens the real robot
    port configured on the cell host.
    """

    robot_id: Optional[str] = None
    """Stable robot id used by LeRobot to find calibration data."""

    calibration_dir: Optional[Path] = None
    """Calibration store path visible inside the customer container."""

    calibration_file_path: Optional[Path] = None
    """Exact LeRobot calibration file path visible inside the customer container."""

    language_instruction: Optional[str] = None
    """Task instruction provided by the cell."""

    local_control_endpoint: Optional[str] = None
    """Developer local-container control endpoint for keyboard-driven state."""

    operator_call_endpoint: Optional[str] = None
    """Operator-call endpoint served by the cell program for human-in-the-loop
    calls (manual reset confirmation). Distinct from ``robot_port``, which is the
    robot/bus connector (potentially a headless edge device)."""

    is_local_container: bool = False
    """True when running a Docker image locally for development."""
    safety_limit: Optional[float] = None
    """Relative action safety limit exposed by the cell, if applicable."""

    arms: dict[str, RuntimeArm] = field(default_factory=dict)
    """Named arms for bimanual/multi-arm cells."""

    # Reused connection + log throttle for teleop polling (see get_teleop_action).
    # Not part of the constructor or the public/comparable surface.
    _teleop_conn: Any = field(default=None, init=False, repr=False, compare=False)
    _teleop_last_error_log: float = field(default=0.0, init=False, repr=False, compare=False)
    # Reused connection for polling the human-reported rollout outcome served by
    # the cell program's operator-call endpoint (see is_complete).
    _completion_conn: Any = field(default=None, init=False, repr=False, compare=False)
    _completion_last_error_log: float = field(default=0.0, init=False, repr=False, compare=False)

    @property
    def is_bimanual(self) -> bool:
        return {"left", "right"}.issubset(self.arms)

    def arm(self, name: str) -> RuntimeArm:
        try:
            return self.arms[name]
        except KeyError as exc:
            raise KeyError(f"cell has no arm named {name!r}") from exc

    def prepare_bimanual_calibration_dir(self) -> BimanualCalibrationLayout:
        """Create a temp calibration dir using LeRobot's `<base>_<arm>.json` names.

        Each arm's source calibration is resolved (in order) from its own
        ``calibration_file_path``, its own ``calibration_dir`` keyed by the arm's
        ``robot_id``, or—when the arm declares neither—the **cell-level**
        ``calibration_dir`` keyed by the arm's ``robot_id`` (``<robot_id>.json``).
        This mirrors how the cell's per-arm health check resolves calibration
        (``arm.calibration_dir or cell.calibration_dir``), so a config that only
        sets a top-level ``calibration_dir`` (per-arm ``robot_id`` only) works.
        """

        if not self.is_bimanual:
            raise RuntimeError("bimanual calibration requires left and right arms")
        robot_id = self.robot_id
        if not robot_id:
            raise RuntimeError("bimanual calibration requires ctx.cell.robot_id")
        calibration_dir = Path(tempfile.mkdtemp(prefix="armnet-bimanual-calibration-"))
        for arm_name in ("left", "right"):
            arm = self.arm(arm_name)
            source = arm.calibration_file_path
            if source is None:
                cal_dir = arm.calibration_dir or self.calibration_dir
                if cal_dir is not None and arm.robot_id:
                    source = Path(cal_dir) / f"{arm.robot_id}.json"
            if source is None or not Path(source).is_file():
                raise RuntimeError(
                    f"no calibration file found for {arm_name} arm "
                    f"(robot_id={arm.robot_id!r}); looked for "
                    f"{source if source is not None else '<unresolved>'}. Set the "
                    "cell-level calibration_dir (with per-arm robot_id) or each "
                    "arm's calibration_dir/calibration_file_path."
                )
            shutil.copy2(source, calibration_dir / f"{robot_id}_{arm_name}.json")
        return BimanualCalibrationLayout(robot_id=robot_id, calibration_dir=calibration_dir)

    def reset(self) -> None:
        """Return the robot to rest, then block until the operator confirms.

        Two concerns, two endpoints:

        1. Returning the arm to its rest position is a low-level bus operation,
           so it is sent to the robot connector (``robot_port``), which may be a
           headless edge device.
        2. Operator confirmation is a human-in-the-loop concern, so it is sent
           to the ``operator_call_endpoint`` served by the ``armnet-cell``
           process, whose stdin is the operator's terminal.

        The operator-facing prompt is owned by the cell, not by job code: job
        code only signals *that* a reset point has been reached.
        """

        self._report_progress("Waiting for workspace reset...")

        # 1. Safety: return the arm to rest via the robot connector, if present.
        if self.robot_port and _looks_like_connector_endpoint(self.robot_port):
            response = _connector_request(self.robot_port, {"op": "return_to_rest"})
            if not response.get("ok"):
                raise RuntimeError(response.get("error", "robot return-to-rest failed"))

        # 2. Operator confirmation on the cell-served operator-call endpoint
        # (fallback to the dev local-control endpoint).
        request = {"op": "reset", "request": {"kind": "manual"}}
        operator_endpoint = self.operator_call_endpoint or self.local_control_endpoint
        if operator_endpoint:
            response = _connector_request(operator_endpoint, request)
            if not response.get("ok"):
                error = response.get("error", "operator reset confirmation failed")
                if response.get("error_type") == "ResetTimeoutException":
                    raise ResetTimeoutException(error)
                raise RuntimeError(error)
            return

        # No operator endpoint attached (degenerate in-process dev run): block on
        # the local terminal with a standard prompt owned by the runtime.
        input("Reset the cell workspace, then press Enter. ")

    def is_complete(self, *, block: bool = False) -> CompletionStatus:
        """Return whether the current episode is complete, and its success.

        Returns a :class:`CompletionStatus` ``(complete, success)``:

        1. A human-reported outcome (an operator hitting success/fail in the
           FMS during a live rollout) takes precedence and ends the episode
           immediately, with ``success`` set to the operator's choice. This is
           how an operator stops a dangerous rollout without stopping the job.
        2. Otherwise the cell's automated completion monitor is consulted; a
           task scored complete is reported as a success (``success ==
           complete``). Pass ``block=True`` for a final episode check that waits
           for the cell to score the latest cached frames before returning.

        ``bool(status)`` is ``status.complete`` for backwards compatibility.
        """

        # 1. Human-reported outcome (operator/FMS) wins and ends the episode now.
        reported = self._human_completion()
        if reported is not None:
            return CompletionStatus(complete=True, success=reported)

        # 2. Automated completion scoring via the local-dev or cell connector.
        request = {"op": "is_complete", "block": block}
        if self.local_control_endpoint:
            response = _connector_request(self.local_control_endpoint, request)
            if not response.get("ok"):
                raise RuntimeError(response.get("error", "local completion check failed"))
            return _completion_from_response(response)
        if self.robot_port and _looks_like_connector_endpoint(self.robot_port):
            response = _connector_request(self.robot_port, request)
            if not response.get("ok"):
                raise RuntimeError(response.get("error", "cell completion check failed"))
            return _completion_from_response(response)
        return CompletionStatus(complete=False, success=False)

    def _human_completion(self) -> Optional[bool]:
        """Return the operator's reported success/fail, or None if none pending.

        Polls the cell program's operator-call endpoint (where FMS rollout
        commands land). A wedged/slow channel must never stall the control
        loop, so this mirrors get_teleop_action: bounded read, throttled error
        logging, drop the connection on error, and treat failures as "no
        outcome" so the episode simply continues.
        """

        endpoint = self.operator_call_endpoint or self.local_control_endpoint
        if not endpoint:
            return None

        if self._completion_conn is None or self._completion_conn.endpoint != endpoint:
            if self._completion_conn is not None:
                self._completion_conn.close()
            self._completion_conn = _TeleopConnection(endpoint, timeout=_TELEOP_READ_TIMEOUT_S)

        try:
            response = self._completion_conn.request({"op": "get_completion"})
        except Exception as exc:  # noqa: BLE001
            now = time.monotonic()
            if now - self._completion_last_error_log >= _TELEOP_ERROR_LOG_INTERVAL_S:
                self._completion_last_error_log = now
                logger.warning(
                    "completion read from %s failed (treating as not complete): %r",
                    endpoint,
                    exc,
                )
            return None

        if not response.get("ok") or not response.get("reported"):
            return None
        return bool(response.get("success", False))

    def rollout_begin(
        self,
        *,
        index: Optional[int] = None,
        total: Optional[int] = None,
        outcome_controls: bool = True,
    ) -> None:
        """Tell the platform a rollout/episode in this job's loop has started.

        The cell publishes this to the FMS, which shows the loop progress
        ("rollout N / M") for the live job. Pass ``index`` (1-based) and, when
        known, ``total`` so operators see how far along the loop is.

        ``outcome_controls`` controls whether the FMS also shows operator
        success/fail buttons: keep the default ``True`` for policy evals; pass
        ``False`` for progress-only loops such as teleop data collection, where
        a human verdict doesn't apply. Best-effort: a failed notification never
        breaks the rollout. Pair with :meth:`rollout_end`.
        """
        payload: dict[str, Any] = {"outcome_controls": bool(outcome_controls)}
        if index is not None:
            payload["index"] = int(index)
        if total is not None:
            payload["total"] = int(total)
        self._rollout_signal("rollout_begin", **payload)

    def rollout_end(self) -> None:
        """Tell the platform the current rollout has ended (hides FMS buttons)."""
        self._rollout_signal("rollout_end")

    def _rollout_signal(self, op: str, **payload: Any) -> None:
        endpoint = self.operator_call_endpoint or self.local_control_endpoint
        if not endpoint:
            return
        try:
            _connector_request(endpoint, {"op": op, **payload})
        except Exception:  # noqa: BLE001 - rollout signalling is best-effort
            logger.warning("rollout signal %s to %s failed", op, endpoint, exc_info=True)

    def is_shutting_down(self) -> bool:
        """Return True once the cell has entered the job's post-timeout grace window.

        When a job exceeds its ``timeout_seconds`` the cell does not kill the
        container straight away: it trips the robot interlock (so any further
        robot-bus calls fail) and opens a short *grace window* during which this
        returns True, before force-killing the container. Poll it in your loop
        and break out to finalize gracefully — e.g. save/push a dataset — instead
        of being killed mid-write::

            for episode in range(n):
                if ctx.cell.is_shutting_down():
                    break  # finalize below
                ...

        Resilient by design: returns False when no cell/operator endpoint is
        attached or the status can't be read, so it never stalls or crashes the
        control loop.
        """

        endpoint = self.operator_call_endpoint or self.local_control_endpoint
        if not endpoint:
            return False
        try:
            response = _connector_request(
                endpoint, {"op": "shutdown_status"}, read_timeout=_TELEOP_READ_TIMEOUT_S
            )
        except Exception:  # noqa: BLE001 - never let a status poll break the loop
            return False
        return bool(response.get("ok") and response.get("shutting_down"))

    def should_stop(self) -> bool:
        """Return True when local/remote control asks user code to stop safely."""

        request = {"op": "should_stop"}
        if self.local_control_endpoint:
            response = _connector_request(self.local_control_endpoint, request)
            if not response.get("ok"):
                raise RuntimeError(response.get("error", "local stop check failed"))
            return bool(response.get("stop", False))
        return False

    def get_teleop_action(self) -> Optional[dict[str, float]]:
        """Return the freshest remote-teleoperation action for this job, or None.

        The client samples a local leader arm and pushes actions to the cell,
        which keeps only the most recent one (older messages are dropped). This
        reads that most-recent-value register over the operator-call endpoint.

        Returns ``None`` when no teleop has been received yet (or no operator
        endpoint is attached), so a control loop can hold position until the
        operator starts driving. The returned dict is keyed for LeRobot's
        ``send_action`` (e.g. ``{"shoulder_pan.pos": 12.3, ...}``).
        """

        endpoint = self.operator_call_endpoint or self.local_control_endpoint
        if not endpoint:
            return None

        if self._teleop_conn is None or self._teleop_conn.endpoint != endpoint:
            if self._teleop_conn is not None:
                self._teleop_conn.close()
            self._teleop_conn = _TeleopConnection(endpoint, timeout=_TELEOP_READ_TIMEOUT_S)

        try:
            response = self._teleop_conn.request({"op": "get_teleop"})
        except Exception as exc:  # noqa: BLE001
            # A wedged/slow teleop channel must not stall or crash the control
            # loop: log (throttled) so a recurrence is diagnosable, drop the
            # connection (already done in request()) so we reconnect next tick,
            # and hold position by returning None.
            now = time.monotonic()
            if now - self._teleop_last_error_log >= _TELEOP_ERROR_LOG_INTERVAL_S:
                self._teleop_last_error_log = now
                logger.warning(
                    "teleop read from %s failed (holding position; will reconnect): %r",
                    endpoint,
                    exc,
                )
            return None

        if not response.get("ok"):
            logger.warning("teleop read returned error: %s", response.get("error"))
            return None
        action = response.get("action")
        if not action:
            return None
        return {str(key): float(value) for key, value in action.items()}

    def get_teleop_event(self) -> Optional[str]:
        """Return the next pending recording-control event, or None.

        While teleoperating, the client can send discrete recording-control
        events alongside the action stream — LeRobot's standard dataset
        recording shortcuts: ``"next_episode"`` (Right Arrow: save the episode
        and move on), ``"rerecord_episode"`` (Left Arrow: discard and redo) and
        ``"stop_recording"`` (Esc: end the session). The cell queues them in
        arrival order; each call pops at most one.

        Like :meth:`get_teleop_action`, a wedged channel never stalls the
        control loop: errors log (throttled), drop the connection so the next
        call reconnects, and return None.
        """

        endpoint = self.operator_call_endpoint or self.local_control_endpoint
        if not endpoint:
            return None

        if self._teleop_conn is None or self._teleop_conn.endpoint != endpoint:
            if self._teleop_conn is not None:
                self._teleop_conn.close()
            self._teleop_conn = _TeleopConnection(endpoint, timeout=_TELEOP_READ_TIMEOUT_S)

        try:
            response = self._teleop_conn.request({"op": "get_teleop_event"})
        except Exception as exc:  # noqa: BLE001
            now = time.monotonic()
            if now - self._teleop_last_error_log >= _TELEOP_ERROR_LOG_INTERVAL_S:
                self._teleop_last_error_log = now
                logger.warning(
                    "teleop event read from %s failed (will reconnect): %r",
                    endpoint,
                    exc,
                )
            return None

        if not response.get("ok"):
            logger.warning("teleop event read returned error: %s", response.get("error"))
            return None
        event = response.get("event")
        return str(event) if event else None

    def _report_progress(self, message: str) -> None:
        """Surface a progress message back to the platform.

        M0.5: prints to stdout with a discoverable marker so the cell's
        captured stdout shows progress in order with other prints. M1+
        will also publish a NATS message so the orchestrator can stream
        progress back to the client without waiting for the job to
        terminate.
        """

        # Imported locally to avoid pulling markers into the public API
        # surface of `Context`.
        from armnet_runtime.markers import PROGRESS_MARKER
        print(f"{PROGRESS_MARKER} {message}", flush=True)
        time.sleep(0.01)

robot_port class-attribute instance-attribute

robot_port: Optional[str] = None

Robot port value to pass into LeRobot robot configs.

In container-backed remote execution this is the connector endpoint, not the host's physical serial path. The SDK's import-system swap routes that endpoint through the cell-side connector, which then opens the real robot port configured on the cell host.

robot_id class-attribute instance-attribute

robot_id: Optional[str] = None

Stable robot id used by LeRobot to find calibration data.

calibration_dir class-attribute instance-attribute

calibration_dir: Optional[Path] = None

Calibration store path visible inside the customer container.

calibration_file_path class-attribute instance-attribute

calibration_file_path: Optional[Path] = None

Exact LeRobot calibration file path visible inside the customer container.

language_instruction class-attribute instance-attribute

language_instruction: Optional[str] = None

Task instruction provided by the cell.

local_control_endpoint class-attribute instance-attribute

local_control_endpoint: Optional[str] = None

Developer local-container control endpoint for keyboard-driven state.

operator_call_endpoint class-attribute instance-attribute

operator_call_endpoint: Optional[str] = None

Operator-call endpoint served by the cell program for human-in-the-loop calls (manual reset confirmation). Distinct from robot_port, which is the robot/bus connector (potentially a headless edge device).

is_local_container class-attribute instance-attribute

is_local_container: bool = False

True when running a Docker image locally for development.

safety_limit class-attribute instance-attribute

safety_limit: Optional[float] = None

Relative action safety limit exposed by the cell, if applicable.

arms class-attribute instance-attribute

arms: dict[str, RuntimeArm] = field(default_factory=dict)

Named arms for bimanual/multi-arm cells.

is_bimanual property

is_bimanual: bool

arm

arm(name: str) -> RuntimeArm

Source code in runtime/src/armnet_runtime/context.py

def arm(self, name: str) -> RuntimeArm:
    try:
        return self.arms[name]
    except KeyError as exc:
        raise KeyError(f"cell has no arm named {name!r}") from exc

prepare_bimanual_calibration_dir

prepare_bimanual_calibration_dir() -> BimanualCalibrationLayout

Create a temp calibration dir using LeRobot's <base>_<arm>.json names.

Each arm's source calibration is resolved (in order) from its own calibration_file_path, its own calibration_dir keyed by the arm's robot_id, or—when the arm declares neither—the cell-level calibration_dir keyed by the arm's robot_id (<robot_id>.json). This mirrors how the cell's per-arm health check resolves calibration (arm.calibration_dir or cell.calibration_dir), so a config that only sets a top-level calibration_dir (per-arm robot_id only) works.

Source code in runtime/src/armnet_runtime/context.py

def prepare_bimanual_calibration_dir(self) -> BimanualCalibrationLayout:
    """Create a temp calibration dir using LeRobot's `<base>_<arm>.json` names.

    Each arm's source calibration is resolved (in order) from its own
    ``calibration_file_path``, its own ``calibration_dir`` keyed by the arm's
    ``robot_id``, or—when the arm declares neither—the **cell-level**
    ``calibration_dir`` keyed by the arm's ``robot_id`` (``<robot_id>.json``).
    This mirrors how the cell's per-arm health check resolves calibration
    (``arm.calibration_dir or cell.calibration_dir``), so a config that only
    sets a top-level ``calibration_dir`` (per-arm ``robot_id`` only) works.
    """

    if not self.is_bimanual:
        raise RuntimeError("bimanual calibration requires left and right arms")
    robot_id = self.robot_id
    if not robot_id:
        raise RuntimeError("bimanual calibration requires ctx.cell.robot_id")
    calibration_dir = Path(tempfile.mkdtemp(prefix="armnet-bimanual-calibration-"))
    for arm_name in ("left", "right"):
        arm = self.arm(arm_name)
        source = arm.calibration_file_path
        if source is None:
            cal_dir = arm.calibration_dir or self.calibration_dir
            if cal_dir is not None and arm.robot_id:
                source = Path(cal_dir) / f"{arm.robot_id}.json"
        if source is None or not Path(source).is_file():
            raise RuntimeError(
                f"no calibration file found for {arm_name} arm "
                f"(robot_id={arm.robot_id!r}); looked for "
                f"{source if source is not None else '<unresolved>'}. Set the "
                "cell-level calibration_dir (with per-arm robot_id) or each "
                "arm's calibration_dir/calibration_file_path."
            )
        shutil.copy2(source, calibration_dir / f"{robot_id}_{arm_name}.json")
    return BimanualCalibrationLayout(robot_id=robot_id, calibration_dir=calibration_dir)

reset

reset() -> None

Return the robot to rest, then block until the operator confirms.

Two concerns, two endpoints:

1. Returning the arm to its rest position is a low-level bus operation, so it is sent to the robot connector (robot_port), which may be a headless edge device.
2. Operator confirmation is a human-in-the-loop concern, so it is sent to the operator_call_endpoint served by the armnet-cell process, whose stdin is the operator's terminal.

The operator-facing prompt is owned by the cell, not by job code: job code only signals that a reset point has been reached.

Source code in runtime/src/armnet_runtime/context.py

def reset(self) -> None:
    """Return the robot to rest, then block until the operator confirms.

    Two concerns, two endpoints:

    1. Returning the arm to its rest position is a low-level bus operation,
       so it is sent to the robot connector (``robot_port``), which may be a
       headless edge device.
    2. Operator confirmation is a human-in-the-loop concern, so it is sent
       to the ``operator_call_endpoint`` served by the ``armnet-cell``
       process, whose stdin is the operator's terminal.

    The operator-facing prompt is owned by the cell, not by job code: job
    code only signals *that* a reset point has been reached.
    """

    self._report_progress("Waiting for workspace reset...")

    # 1. Safety: return the arm to rest via the robot connector, if present.
    if self.robot_port and _looks_like_connector_endpoint(self.robot_port):
        response = _connector_request(self.robot_port, {"op": "return_to_rest"})
        if not response.get("ok"):
            raise RuntimeError(response.get("error", "robot return-to-rest failed"))

    # 2. Operator confirmation on the cell-served operator-call endpoint
    # (fallback to the dev local-control endpoint).
    request = {"op": "reset", "request": {"kind": "manual"}}
    operator_endpoint = self.operator_call_endpoint or self.local_control_endpoint
    if operator_endpoint:
        response = _connector_request(operator_endpoint, request)
        if not response.get("ok"):
            error = response.get("error", "operator reset confirmation failed")
            if response.get("error_type") == "ResetTimeoutException":
                raise ResetTimeoutException(error)
            raise RuntimeError(error)
        return

    # No operator endpoint attached (degenerate in-process dev run): block on
    # the local terminal with a standard prompt owned by the runtime.
    input("Reset the cell workspace, then press Enter. ")

is_complete

is_complete(*, block: bool = False) -> CompletionStatus

Return whether the current episode is complete, and its success.

Returns a :class:CompletionStatus (complete, success):

1. A human-reported outcome (an operator hitting success/fail in the FMS during a live rollout) takes precedence and ends the episode immediately, with success set to the operator's choice. This is how an operator stops a dangerous rollout without stopping the job.
2. Otherwise the cell's automated completion monitor is consulted; a task scored complete is reported as a success (success == complete). Pass block=True for a final episode check that waits for the cell to score the latest cached frames before returning.

bool(status) is status.complete for backwards compatibility.

Source code in runtime/src/armnet_runtime/context.py

def is_complete(self, *, block: bool = False) -> CompletionStatus:
    """Return whether the current episode is complete, and its success.

    Returns a :class:`CompletionStatus` ``(complete, success)``:

    1. A human-reported outcome (an operator hitting success/fail in the
       FMS during a live rollout) takes precedence and ends the episode
       immediately, with ``success`` set to the operator's choice. This is
       how an operator stops a dangerous rollout without stopping the job.
    2. Otherwise the cell's automated completion monitor is consulted; a
       task scored complete is reported as a success (``success ==
       complete``). Pass ``block=True`` for a final episode check that waits
       for the cell to score the latest cached frames before returning.

    ``bool(status)`` is ``status.complete`` for backwards compatibility.
    """

    # 1. Human-reported outcome (operator/FMS) wins and ends the episode now.
    reported = self._human_completion()
    if reported is not None:
        return CompletionStatus(complete=True, success=reported)

    # 2. Automated completion scoring via the local-dev or cell connector.
    request = {"op": "is_complete", "block": block}
    if self.local_control_endpoint:
        response = _connector_request(self.local_control_endpoint, request)
        if not response.get("ok"):
            raise RuntimeError(response.get("error", "local completion check failed"))
        return _completion_from_response(response)
    if self.robot_port and _looks_like_connector_endpoint(self.robot_port):
        response = _connector_request(self.robot_port, request)
        if not response.get("ok"):
            raise RuntimeError(response.get("error", "cell completion check failed"))
        return _completion_from_response(response)
    return CompletionStatus(complete=False, success=False)

rollout_begin

rollout_begin(*, index: Optional[int] = None, total: Optional[int] = None, outcome_controls: bool = True) -> None

Tell the platform a rollout/episode in this job's loop has started.

The cell publishes this to the FMS, which shows the loop progress ("rollout N / M") for the live job. Pass index (1-based) and, when known, total so operators see how far along the loop is.

outcome_controls controls whether the FMS also shows operator success/fail buttons: keep the default True for policy evals; pass False for progress-only loops such as teleop data collection, where a human verdict doesn't apply. Best-effort: a failed notification never breaks the rollout. Pair with :meth:rollout_end.

Source code in runtime/src/armnet_runtime/context.py

def rollout_begin(
    self,
    *,
    index: Optional[int] = None,
    total: Optional[int] = None,
    outcome_controls: bool = True,
) -> None:
    """Tell the platform a rollout/episode in this job's loop has started.

    The cell publishes this to the FMS, which shows the loop progress
    ("rollout N / M") for the live job. Pass ``index`` (1-based) and, when
    known, ``total`` so operators see how far along the loop is.

    ``outcome_controls`` controls whether the FMS also shows operator
    success/fail buttons: keep the default ``True`` for policy evals; pass
    ``False`` for progress-only loops such as teleop data collection, where
    a human verdict doesn't apply. Best-effort: a failed notification never
    breaks the rollout. Pair with :meth:`rollout_end`.
    """
    payload: dict[str, Any] = {"outcome_controls": bool(outcome_controls)}
    if index is not None:
        payload["index"] = int(index)
    if total is not None:
        payload["total"] = int(total)
    self._rollout_signal("rollout_begin", **payload)

rollout_end

rollout_end() -> None

Tell the platform the current rollout has ended (hides FMS buttons).

Source code in runtime/src/armnet_runtime/context.py

def rollout_end(self) -> None:
    """Tell the platform the current rollout has ended (hides FMS buttons)."""
    self._rollout_signal("rollout_end")

is_shutting_down

is_shutting_down() -> bool

Return True once the cell has entered the job's post-timeout grace window.

When a job exceeds its timeout_seconds the cell does not kill the container straight away: it trips the robot interlock (so any further robot-bus calls fail) and opens a short grace window during which this returns True, before force-killing the container. Poll it in your loop and break out to finalize gracefully — e.g. save/push a dataset — instead of being killed mid-write::

for episode in range(n):
    if ctx.cell.is_shutting_down():
        break  # finalize below
    ...

Resilient by design: returns False when no cell/operator endpoint is attached or the status can't be read, so it never stalls or crashes the control loop.

Source code in runtime/src/armnet_runtime/context.py

def is_shutting_down(self) -> bool:
    """Return True once the cell has entered the job's post-timeout grace window.

    When a job exceeds its ``timeout_seconds`` the cell does not kill the
    container straight away: it trips the robot interlock (so any further
    robot-bus calls fail) and opens a short *grace window* during which this
    returns True, before force-killing the container. Poll it in your loop
    and break out to finalize gracefully — e.g. save/push a dataset — instead
    of being killed mid-write::

        for episode in range(n):
            if ctx.cell.is_shutting_down():
                break  # finalize below
            ...

    Resilient by design: returns False when no cell/operator endpoint is
    attached or the status can't be read, so it never stalls or crashes the
    control loop.
    """

    endpoint = self.operator_call_endpoint or self.local_control_endpoint
    if not endpoint:
        return False
    try:
        response = _connector_request(
            endpoint, {"op": "shutdown_status"}, read_timeout=_TELEOP_READ_TIMEOUT_S
        )
    except Exception:  # noqa: BLE001 - never let a status poll break the loop
        return False
    return bool(response.get("ok") and response.get("shutting_down"))

should_stop

should_stop() -> bool

Return True when local/remote control asks user code to stop safely.

Source code in runtime/src/armnet_runtime/context.py

def should_stop(self) -> bool:
    """Return True when local/remote control asks user code to stop safely."""

    request = {"op": "should_stop"}
    if self.local_control_endpoint:
        response = _connector_request(self.local_control_endpoint, request)
        if not response.get("ok"):
            raise RuntimeError(response.get("error", "local stop check failed"))
        return bool(response.get("stop", False))
    return False

get_teleop_action

get_teleop_action() -> Optional[dict[str, float]]

Return the freshest remote-teleoperation action for this job, or None.

The client samples a local leader arm and pushes actions to the cell, which keeps only the most recent one (older messages are dropped). This reads that most-recent-value register over the operator-call endpoint.

Returns None when no teleop has been received yet (or no operator endpoint is attached), so a control loop can hold position until the operator starts driving. The returned dict is keyed for LeRobot's send_action (e.g. {"shoulder_pan.pos": 12.3, ...}).

Source code in runtime/src/armnet_runtime/context.py

def get_teleop_action(self) -> Optional[dict[str, float]]:
    """Return the freshest remote-teleoperation action for this job, or None.

    The client samples a local leader arm and pushes actions to the cell,
    which keeps only the most recent one (older messages are dropped). This
    reads that most-recent-value register over the operator-call endpoint.

    Returns ``None`` when no teleop has been received yet (or no operator
    endpoint is attached), so a control loop can hold position until the
    operator starts driving. The returned dict is keyed for LeRobot's
    ``send_action`` (e.g. ``{"shoulder_pan.pos": 12.3, ...}``).
    """

    endpoint = self.operator_call_endpoint or self.local_control_endpoint
    if not endpoint:
        return None

    if self._teleop_conn is None or self._teleop_conn.endpoint != endpoint:
        if self._teleop_conn is not None:
            self._teleop_conn.close()
        self._teleop_conn = _TeleopConnection(endpoint, timeout=_TELEOP_READ_TIMEOUT_S)

    try:
        response = self._teleop_conn.request({"op": "get_teleop"})
    except Exception as exc:  # noqa: BLE001
        # A wedged/slow teleop channel must not stall or crash the control
        # loop: log (throttled) so a recurrence is diagnosable, drop the
        # connection (already done in request()) so we reconnect next tick,
        # and hold position by returning None.
        now = time.monotonic()
        if now - self._teleop_last_error_log >= _TELEOP_ERROR_LOG_INTERVAL_S:
            self._teleop_last_error_log = now
            logger.warning(
                "teleop read from %s failed (holding position; will reconnect): %r",
                endpoint,
                exc,
            )
        return None

    if not response.get("ok"):
        logger.warning("teleop read returned error: %s", response.get("error"))
        return None
    action = response.get("action")
    if not action:
        return None
    return {str(key): float(value) for key, value in action.items()}

get_teleop_event

get_teleop_event() -> Optional[str]

Return the next pending recording-control event, or None.

While teleoperating, the client can send discrete recording-control events alongside the action stream — LeRobot's standard dataset recording shortcuts: "next_episode" (Right Arrow: save the episode and move on), "rerecord_episode" (Left Arrow: discard and redo) and "stop_recording" (Esc: end the session). The cell queues them in arrival order; each call pops at most one.

Like :meth:get_teleop_action, a wedged channel never stalls the control loop: errors log (throttled), drop the connection so the next call reconnects, and return None.

Source code in runtime/src/armnet_runtime/context.py

def get_teleop_event(self) -> Optional[str]:
    """Return the next pending recording-control event, or None.

    While teleoperating, the client can send discrete recording-control
    events alongside the action stream — LeRobot's standard dataset
    recording shortcuts: ``"next_episode"`` (Right Arrow: save the episode
    and move on), ``"rerecord_episode"`` (Left Arrow: discard and redo) and
    ``"stop_recording"`` (Esc: end the session). The cell queues them in
    arrival order; each call pops at most one.

    Like :meth:`get_teleop_action`, a wedged channel never stalls the
    control loop: errors log (throttled), drop the connection so the next
    call reconnects, and return None.
    """

    endpoint = self.operator_call_endpoint or self.local_control_endpoint
    if not endpoint:
        return None

    if self._teleop_conn is None or self._teleop_conn.endpoint != endpoint:
        if self._teleop_conn is not None:
            self._teleop_conn.close()
        self._teleop_conn = _TeleopConnection(endpoint, timeout=_TELEOP_READ_TIMEOUT_S)

    try:
        response = self._teleop_conn.request({"op": "get_teleop_event"})
    except Exception as exc:  # noqa: BLE001
        now = time.monotonic()
        if now - self._teleop_last_error_log >= _TELEOP_ERROR_LOG_INTERVAL_S:
            self._teleop_last_error_log = now
            logger.warning(
                "teleop event read from %s failed (will reconnect): %r",
                endpoint,
                exc,
            )
        return None

    if not response.get("ok"):
        logger.warning("teleop event read returned error: %s", response.get("error"))
        return None
    event = response.get("event")
    return str(event) if event else None

CompletionStatus

Bases: NamedTuple

Result of :meth:Cell.is_complete: whether the episode is over and how.

complete is True once the current rollout/episode should end (the task was scored complete by the automated monitor, or a human operator reported an outcome). success is the verdict — for automated completion it equals complete (a task scored complete is a success); for a human-reported outcome it is the operator's success/fail choice, so an operator can end a rollout as a failure (e.g. the robot behaved dangerously).

Backwards-compatible truthiness: bool(status) is status.complete, so existing if ctx.cell.is_complete(): ... callers keep working, while new code can unpack complete, success = ctx.cell.is_complete().

Source code in runtime/src/armnet_runtime/context.py

class CompletionStatus(NamedTuple):
    """Result of :meth:`Cell.is_complete`: whether the episode is over and how.

    ``complete`` is True once the current rollout/episode should end (the task
    was scored complete by the automated monitor, or a human operator reported
    an outcome). ``success`` is the verdict — for automated completion it equals
    ``complete`` (a task scored complete is a success); for a human-reported
    outcome it is the operator's success/fail choice, so an operator can end a
    rollout as a *failure* (e.g. the robot behaved dangerously).

    Backwards-compatible truthiness: ``bool(status)`` is ``status.complete``, so
    existing ``if ctx.cell.is_complete(): ...`` callers keep working, while new
    code can unpack ``complete, success = ctx.cell.is_complete()``.
    """

    complete: bool
    success: bool

    def __bool__(self) -> bool:  # noqa: D401 - truthy iff the episode is complete
        return self.complete

complete instance-attribute

complete: bool

success instance-attribute

success: bool

Context dataclass

Everything a @main-decorated function needs from the platform.

Source code in runtime/src/armnet_runtime/context.py

@dataclass
class Context:
    """Everything a ``@main``-decorated function needs from the platform."""

    job_id: str
    embodiment: Embodiment
    task: Task
    args: dict[str, Any] = field(default_factory=dict)
    cell: Cell = field(default_factory=Cell)
    camera_configs: dict[str, Any] = field(default_factory=dict)
    cache_home: Optional[Path] = None
    volume: Volume = field(default_factory=Volume)
    secrets: dict[str, str] = field(default_factory=dict)
    timeout_seconds: Optional[int] = None
    # Lazily created background Rerun streamer (see log_rerun_data). Not part of
    # the constructor or the public/comparable surface.
    _rerun_streamer: Any = field(default=None, init=False, repr=False, compare=False)

    def report_progress(self, message: str) -> None:
        """Surface a progress message back to the platform.

        M0.5: prints to stdout with a discoverable marker so the cell's
        captured stdout shows progress in order with other prints. M1+
        will also publish a NATS message so the orchestrator can stream
        progress back to the client without waiting for the job to
        terminate.
        """

        # Imported locally to avoid pulling markers into the public API
        # surface of `Context`.
        from armnet_runtime.markers import PROGRESS_MARKER
        print(f"{PROGRESS_MARKER} {message}", flush=True)

    def is_shutting_down(self) -> bool:
        """Whether the cell has entered the job's post-timeout grace window.

        Convenience delegate for :meth:`Cell.is_shutting_down`. Poll it in long
        loops and break out to finalize gracefully before the cell kills the
        container.
        """
        return self.cell.is_shutting_down()

    def log_rerun_data(
        self,
        observation: dict[str, Any] | None = None,
        action: dict[str, Any] | None = None,
        *,
        compress_images: bool = True,
        jpeg_quality: int = 75,
    ) -> None:
        """Stream observation/action data to a Rerun viewer on the client.

        Mirrors LeRobot's ``log_rerun_data``: scalars are logged as Rerun
        scalars, image-like arrays as images, and other arrays as per-element
        scalars. Keys are namespaced with ``observation.`` / ``action.`` when
        not already.

        Unlike the LeRobot helper, this does not call ``rr.log`` in-process
        (the cell container has no viewer). Instead it serializes a protobuf
        packet and emits it on stdout behind a marker; the cell republishes it
        on ``logs.<job_id>.rerun`` and the client's orchestrate script replays
        it into the viewer it started with ``rr.init(...)``.

        Images are JPEG-compressed by default to keep the NATS stream light;
        set ``compress_images=False`` to send raw RGB. opencv is required for
        compression and numpy for any array handling; both are imported lazily.

        Non-blocking: the snapshot is handed to a background worker thread that
        does the encoding and stdout write, so the calling control loop never
        stalls on visualization. The worker's queue is bounded and drops the
        oldest pending frame under backpressure (tune with
        ``ARMNET_RERUN_QUEUE_MAXSIZE``), so a slow consumer sheds frames
        rather than slowing the robot loop.
        """

        if not observation and not action:
            return

        from armnet_runtime.rerun import RerunStreamer

        if self._rerun_streamer is None:
            self._rerun_streamer = RerunStreamer(self.job_id)
            self._rerun_streamer.start()
        self._rerun_streamer.submit(
            observation,
            action,
            compress_images=compress_images,
            jpeg_quality=jpeg_quality,
        )

job_id instance-attribute

job_id: str

embodiment instance-attribute

embodiment: Embodiment

task instance-attribute

task: Task

args class-attribute instance-attribute

args: dict[str, Any] = field(default_factory=dict)

cell class-attribute instance-attribute

cell: Cell = field(default_factory=Cell)

camera_configs class-attribute instance-attribute

camera_configs: dict[str, Any] = field(default_factory=dict)

cache_home class-attribute instance-attribute

cache_home: Optional[Path] = None

volume class-attribute instance-attribute

volume: Volume = field(default_factory=Volume)

secrets class-attribute instance-attribute

secrets: dict[str, str] = field(default_factory=dict)

timeout_seconds class-attribute instance-attribute

timeout_seconds: Optional[int] = None

report_progress

report_progress(message: str) -> None

Surface a progress message back to the platform.

M0.5: prints to stdout with a discoverable marker so the cell's captured stdout shows progress in order with other prints. M1+ will also publish a NATS message so the orchestrator can stream progress back to the client without waiting for the job to terminate.

Source code in runtime/src/armnet_runtime/context.py

def report_progress(self, message: str) -> None:
    """Surface a progress message back to the platform.

    M0.5: prints to stdout with a discoverable marker so the cell's
    captured stdout shows progress in order with other prints. M1+
    will also publish a NATS message so the orchestrator can stream
    progress back to the client without waiting for the job to
    terminate.
    """

    # Imported locally to avoid pulling markers into the public API
    # surface of `Context`.
    from armnet_runtime.markers import PROGRESS_MARKER
    print(f"{PROGRESS_MARKER} {message}", flush=True)

is_shutting_down

is_shutting_down() -> bool

Whether the cell has entered the job's post-timeout grace window.

Convenience delegate for :meth:Cell.is_shutting_down. Poll it in long loops and break out to finalize gracefully before the cell kills the container.

Source code in runtime/src/armnet_runtime/context.py

def is_shutting_down(self) -> bool:
    """Whether the cell has entered the job's post-timeout grace window.

    Convenience delegate for :meth:`Cell.is_shutting_down`. Poll it in long
    loops and break out to finalize gracefully before the cell kills the
    container.
    """
    return self.cell.is_shutting_down()

log_rerun_data

log_rerun_data(observation: dict[str, Any] | None = None, action: dict[str, Any] | None = None, *, compress_images: bool = True, jpeg_quality: int = 75) -> None

Stream observation/action data to a Rerun viewer on the client.

Mirrors LeRobot's log_rerun_data: scalars are logged as Rerun scalars, image-like arrays as images, and other arrays as per-element scalars. Keys are namespaced with observation. / action. when not already.

Unlike the LeRobot helper, this does not call rr.log in-process (the cell container has no viewer). Instead it serializes a protobuf packet and emits it on stdout behind a marker; the cell republishes it on logs.<job_id>.rerun and the client's orchestrate script replays it into the viewer it started with rr.init(...).

Images are JPEG-compressed by default to keep the NATS stream light; set compress_images=False to send raw RGB. opencv is required for compression and numpy for any array handling; both are imported lazily.

Non-blocking: the snapshot is handed to a background worker thread that does the encoding and stdout write, so the calling control loop never stalls on visualization. The worker's queue is bounded and drops the oldest pending frame under backpressure (tune with ARMNET_RERUN_QUEUE_MAXSIZE), so a slow consumer sheds frames rather than slowing the robot loop.

Source code in runtime/src/armnet_runtime/context.py

def log_rerun_data(
    self,
    observation: dict[str, Any] | None = None,
    action: dict[str, Any] | None = None,
    *,
    compress_images: bool = True,
    jpeg_quality: int = 75,
) -> None:
    """Stream observation/action data to a Rerun viewer on the client.

    Mirrors LeRobot's ``log_rerun_data``: scalars are logged as Rerun
    scalars, image-like arrays as images, and other arrays as per-element
    scalars. Keys are namespaced with ``observation.`` / ``action.`` when
    not already.

    Unlike the LeRobot helper, this does not call ``rr.log`` in-process
    (the cell container has no viewer). Instead it serializes a protobuf
    packet and emits it on stdout behind a marker; the cell republishes it
    on ``logs.<job_id>.rerun`` and the client's orchestrate script replays
    it into the viewer it started with ``rr.init(...)``.

    Images are JPEG-compressed by default to keep the NATS stream light;
    set ``compress_images=False`` to send raw RGB. opencv is required for
    compression and numpy for any array handling; both are imported lazily.

    Non-blocking: the snapshot is handed to a background worker thread that
    does the encoding and stdout write, so the calling control loop never
    stalls on visualization. The worker's queue is bounded and drops the
    oldest pending frame under backpressure (tune with
    ``ARMNET_RERUN_QUEUE_MAXSIZE``), so a slow consumer sheds frames
    rather than slowing the robot loop.
    """

    if not observation and not action:
        return

    from armnet_runtime.rerun import RerunStreamer

    if self._rerun_streamer is None:
        self._rerun_streamer = RerunStreamer(self.job_id)
        self._rerun_streamer.start()
    self._rerun_streamer.submit(
        observation,
        action,
        compress_images=compress_images,
        jpeg_quality=jpeg_quality,
    )

ResetTimeoutException

Bases: RuntimeError

Raised by ctx.cell.reset() when a manual reset is not actioned in time.

The cell waits for an operator to confirm the reset (via the Fleet Management System). If no confirmation arrives within the cell's reset timeout, the cell trips its safety interlock (no further robot commands are allowed, as with a safety violation) and this exception is raised into the job code. Catch it to shut down gracefully — e.g. commit a dataset that was being recorded — before letting the job fail::

try:
    ctx.cell.reset()
except ResetTimeoutException:
    dataset.push_to_hub()  # save what we collected
    raise

Source code in runtime/src/armnet_runtime/context.py

class ResetTimeoutException(RuntimeError):
    """Raised by ``ctx.cell.reset()`` when a manual reset is not actioned in time.

    The cell waits for an operator to confirm the reset (via the Fleet
    Management System). If no confirmation arrives within the cell's reset
    timeout, the cell trips its safety interlock (no further robot commands are
    allowed, as with a safety violation) and this exception is raised into the
    job code. Catch it to shut down gracefully — e.g. commit a dataset that was
    being recorded — before letting the job fail::

        try:
            ctx.cell.reset()
        except ResetTimeoutException:
            dataset.push_to_hub()  # save what we collected
            raise
    """

MainRegistrationError

Bases: RuntimeError

Raised when @main is used incorrectly (multiple times, etc.).

Source code in runtime/src/armnet_runtime/decorator.py

class MainRegistrationError(RuntimeError):
    """Raised when ``@main`` is used incorrectly (multiple times, etc.)."""

require_so101_embodiment

require_so101_embodiment(ctx: 'Context', runtime_name: str) -> bool

Validate the job's embodiment is a (single or bimanual) SO-101.

The embodiment is the source of truth for how many arms the robot has — lerobot/so-101 is a single arm, lerobot/bimanual_so101 is two — and the orchestrator only routes a job to a cell of the matching embodiment. Returns True for the bimanual embodiment (so the caller builds a two-arm robot), False for single-arm.

Raises :class:NotImplementedError for any other embodiment, and :class:RuntimeError if the embodiment's arm count disagrees with the cell's actual wiring (ctx.cell.is_bimanual) — a misrouted or misconfigured cell.

Source code in runtime/src/armnet_runtime/context.py

def require_so101_embodiment(ctx: "Context", runtime_name: str) -> bool:
    """Validate the job's embodiment is a (single or bimanual) SO-101.

    The embodiment is the source of truth for how many arms the robot has —
    ``lerobot/so-101`` is a single arm, ``lerobot/bimanual_so101`` is two — and
    the orchestrator only routes a job to a cell of the matching embodiment.
    Returns ``True`` for the bimanual embodiment (so the caller builds a two-arm
    robot), ``False`` for single-arm.

    Raises :class:`NotImplementedError` for any other embodiment, and
    :class:`RuntimeError` if the embodiment's arm count disagrees with the cell's
    actual wiring (``ctx.cell.is_bimanual``) — a misrouted or misconfigured cell.
    """
    if ctx.embodiment not in (SO101_EMBODIMENT, BIMANUAL_SO101_EMBODIMENT):
        raise NotImplementedError(
            f"{runtime_name} supports {SO101_EMBODIMENT!r} and "
            f"{BIMANUAL_SO101_EMBODIMENT!r}, got {ctx.embodiment!r}"
        )
    expect_bimanual = ctx.embodiment == BIMANUAL_SO101_EMBODIMENT
    if expect_bimanual != ctx.cell.is_bimanual:
        raise RuntimeError(
            f"embodiment {ctx.embodiment!r} expects "
            f"{'two arms (left+right)' if expect_bimanual else 'a single arm'}, "
            f"but the cell exposes arms={sorted(ctx.cell.arms)}"
        )
    return expect_bimanual

main

main(fn: EntryPoint) -> EntryPoint

Decorator: mark fn as the script's entry point.

The function is invoked with a :class:~armnet_runtime.Context by the armnet-runtime entrypoint. It may return any JSON-serialisable value; the value becomes :attr:~armnet_core.JobResult.return_value.

Source code in runtime/src/armnet_runtime/decorator.py

def main(fn: EntryPoint) -> EntryPoint:
    """Decorator: mark ``fn`` as the script's entry point.

    The function is invoked with a :class:`~armnet_runtime.Context` by
    the ``armnet-runtime`` entrypoint. It may return any
    JSON-serialisable value; the value becomes
    :attr:`~armnet_core.JobResult.return_value`.
    """

    global _registered
    if _registered is not None:
        raise MainRegistrationError(
            "armnet: multiple @main-decorated functions found "
            f"(already registered: {_registered.__module__}.{_registered.__qualname__}; "
            f"new: {fn.__module__}.{fn.__qualname__}). Only one entry point "
            "per container is supported."
        )
    _registered = fn
    return fn

armnet_runtime.context

Job context surfaced to @main-decorated functions.

The cell program injects job env vars plus a JSON-encoded cell config when it starts the container; :func:build_context reads them and constructs the :class:Context object that the armnet-runtime entrypoint passes to the user's @main function.

logger module-attribute

logger = logging.getLogger(__name__)

SO101_EMBODIMENT module-attribute

SO101_EMBODIMENT = 'lerobot/so-101'

BIMANUAL_SO101_EMBODIMENT module-attribute

BIMANUAL_SO101_EMBODIMENT = 'lerobot/bimanual_so101'

ResetTimeoutException

Bases: RuntimeError

Raised by ctx.cell.reset() when a manual reset is not actioned in time.

The cell waits for an operator to confirm the reset (via the Fleet Management System). If no confirmation arrives within the cell's reset timeout, the cell trips its safety interlock (no further robot commands are allowed, as with a safety violation) and this exception is raised into the job code. Catch it to shut down gracefully — e.g. commit a dataset that was being recorded — before letting the job fail::

try:
    ctx.cell.reset()
except ResetTimeoutException:
    dataset.push_to_hub()  # save what we collected
    raise

Source code in runtime/src/armnet_runtime/context.py

class ResetTimeoutException(RuntimeError):
    """Raised by ``ctx.cell.reset()`` when a manual reset is not actioned in time.

    The cell waits for an operator to confirm the reset (via the Fleet
    Management System). If no confirmation arrives within the cell's reset
    timeout, the cell trips its safety interlock (no further robot commands are
    allowed, as with a safety violation) and this exception is raised into the
    job code. Catch it to shut down gracefully — e.g. commit a dataset that was
    being recorded — before letting the job fail::

        try:
            ctx.cell.reset()
        except ResetTimeoutException:
            dataset.push_to_hub()  # save what we collected
            raise
    """

CompletionStatus

Bases: NamedTuple

Result of :meth:Cell.is_complete: whether the episode is over and how.

complete is True once the current rollout/episode should end (the task was scored complete by the automated monitor, or a human operator reported an outcome). success is the verdict — for automated completion it equals complete (a task scored complete is a success); for a human-reported outcome it is the operator's success/fail choice, so an operator can end a rollout as a failure (e.g. the robot behaved dangerously).

Backwards-compatible truthiness: bool(status) is status.complete, so existing if ctx.cell.is_complete(): ... callers keep working, while new code can unpack complete, success = ctx.cell.is_complete().

Source code in runtime/src/armnet_runtime/context.py

class CompletionStatus(NamedTuple):
    """Result of :meth:`Cell.is_complete`: whether the episode is over and how.

    ``complete`` is True once the current rollout/episode should end (the task
    was scored complete by the automated monitor, or a human operator reported
    an outcome). ``success`` is the verdict — for automated completion it equals
    ``complete`` (a task scored complete is a success); for a human-reported
    outcome it is the operator's success/fail choice, so an operator can end a
    rollout as a *failure* (e.g. the robot behaved dangerously).

    Backwards-compatible truthiness: ``bool(status)`` is ``status.complete``, so
    existing ``if ctx.cell.is_complete(): ...`` callers keep working, while new
    code can unpack ``complete, success = ctx.cell.is_complete()``.
    """

    complete: bool
    success: bool

    def __bool__(self) -> bool:  # noqa: D401 - truthy iff the episode is complete
        return self.complete

complete instance-attribute

complete: bool

success instance-attribute

success: bool

Volume dataclass

User volume mounted into the runtime container.

Source code in runtime/src/armnet_runtime/context.py

@dataclass
class Volume:
    """User volume mounted into the runtime container."""

    root: Optional[Path] = None

    def path(self, relative_path: str | Path) -> Path:
        if self.root is None:
            raise RuntimeError("armnet volume is not mounted in this context")
        rel = Path(relative_path)
        if rel.is_absolute() or ".." in rel.parts:
            raise ValueError("volume path must be relative and must not contain '..'")
        return self.root / rel

    def read_bytes(self, relative_path: str | Path) -> bytes:
        return self.path(relative_path).read_bytes()

    def read_text(self, relative_path: str | Path) -> str:
        return self.path(relative_path).read_text()

    def write_bytes(self, relative_path: str | Path, data: bytes) -> Path:
        path = self.path(relative_path)
        path.parent.mkdir(parents=True, exist_ok=True)
        path.write_bytes(data)
        return path

    def write_text(self, relative_path: str | Path, data: str) -> Path:
        path = self.path(relative_path)
        path.parent.mkdir(parents=True, exist_ok=True)
        path.write_text(data)
        return path

root class-attribute instance-attribute

root: Optional[Path] = None

path

path(relative_path: str | Path) -> Path

Source code in runtime/src/armnet_runtime/context.py

def path(self, relative_path: str | Path) -> Path:
    if self.root is None:
        raise RuntimeError("armnet volume is not mounted in this context")
    rel = Path(relative_path)
    if rel.is_absolute() or ".." in rel.parts:
        raise ValueError("volume path must be relative and must not contain '..'")
    return self.root / rel

read_bytes

read_bytes(relative_path: str | Path) -> bytes

Source code in runtime/src/armnet_runtime/context.py

def read_bytes(self, relative_path: str | Path) -> bytes:
    return self.path(relative_path).read_bytes()

read_text

read_text(relative_path: str | Path) -> str

Source code in runtime/src/armnet_runtime/context.py

def read_text(self, relative_path: str | Path) -> str:
    return self.path(relative_path).read_text()

write_bytes

write_bytes(relative_path: str | Path, data: bytes) -> Path

Source code in runtime/src/armnet_runtime/context.py

def write_bytes(self, relative_path: str | Path, data: bytes) -> Path:
    path = self.path(relative_path)
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_bytes(data)
    return path

write_text

write_text(relative_path: str | Path, data: str) -> Path

Source code in runtime/src/armnet_runtime/context.py

def write_text(self, relative_path: str | Path, data: str) -> Path:
    path = self.path(relative_path)
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(data)
    return path

RuntimeArm dataclass

Runtime-facing handle for one named robot arm in a multi-arm cell.

Source code in runtime/src/armnet_runtime/context.py

@dataclass(frozen=True)
class RuntimeArm:
    """Runtime-facing handle for one named robot arm in a multi-arm cell."""

    name: str
    robot_port: str
    robot_id: Optional[str] = None
    calibration_dir: Optional[Path] = None
    calibration_file_path: Optional[Path] = None
    safety_limit: Optional[float] = None

name instance-attribute

name: str

robot_port instance-attribute

robot_port: str

robot_id class-attribute instance-attribute

robot_id: Optional[str] = None

calibration_dir class-attribute instance-attribute

calibration_dir: Optional[Path] = None

calibration_file_path class-attribute instance-attribute

calibration_file_path: Optional[Path] = None

safety_limit class-attribute instance-attribute

safety_limit: Optional[float] = None

BimanualCalibrationLayout dataclass

Temporary calibration layout matching LeRobot's bimanual id convention.

Source code in runtime/src/armnet_runtime/context.py

@dataclass(frozen=True)
class BimanualCalibrationLayout:
    """Temporary calibration layout matching LeRobot's bimanual id convention."""

    robot_id: str
    calibration_dir: Path

robot_id instance-attribute

robot_id: str

calibration_dir instance-attribute

calibration_dir: Path

Cell dataclass

Handle to the physical cell the user code is running on.

M0.5 stub: there is no real cell yet, so robot_port is always None and :meth:reset is a no-op. The shape is fixed now so the spec example compiles end-to-end and so M2/M3 can fill in the implementation without touching customer-facing imports.

Source code in runtime/src/armnet_runtime/context.py

@dataclass
class Cell:
    """Handle to the physical cell the user code is running on.

    M0.5 stub: there is no real cell yet, so ``robot_port`` is always ``None``
    and :meth:`reset` is a no-op. The shape is fixed now so the spec
    example compiles end-to-end and so M2/M3 can fill in the
    implementation without touching customer-facing imports.
    """

    robot_port: Optional[str] = None
    """Robot port value to pass into LeRobot robot configs.

    In container-backed remote execution this is the connector endpoint, not
    the host's physical serial path. The SDK's import-system swap routes that
    endpoint through the cell-side connector, which then opens the real robot
    port configured on the cell host.
    """

    robot_id: Optional[str] = None
    """Stable robot id used by LeRobot to find calibration data."""

    calibration_dir: Optional[Path] = None
    """Calibration store path visible inside the customer container."""

    calibration_file_path: Optional[Path] = None
    """Exact LeRobot calibration file path visible inside the customer container."""

    language_instruction: Optional[str] = None
    """Task instruction provided by the cell."""

    local_control_endpoint: Optional[str] = None
    """Developer local-container control endpoint for keyboard-driven state."""

    operator_call_endpoint: Optional[str] = None
    """Operator-call endpoint served by the cell program for human-in-the-loop
    calls (manual reset confirmation). Distinct from ``robot_port``, which is the
    robot/bus connector (potentially a headless edge device)."""

    is_local_container: bool = False
    """True when running a Docker image locally for development."""
    safety_limit: Optional[float] = None
    """Relative action safety limit exposed by the cell, if applicable."""

    arms: dict[str, RuntimeArm] = field(default_factory=dict)
    """Named arms for bimanual/multi-arm cells."""

    # Reused connection + log throttle for teleop polling (see get_teleop_action).
    # Not part of the constructor or the public/comparable surface.
    _teleop_conn: Any = field(default=None, init=False, repr=False, compare=False)
    _teleop_last_error_log: float = field(default=0.0, init=False, repr=False, compare=False)
    # Reused connection for polling the human-reported rollout outcome served by
    # the cell program's operator-call endpoint (see is_complete).
    _completion_conn: Any = field(default=None, init=False, repr=False, compare=False)
    _completion_last_error_log: float = field(default=0.0, init=False, repr=False, compare=False)

    @property
    def is_bimanual(self) -> bool:
        return {"left", "right"}.issubset(self.arms)

    def arm(self, name: str) -> RuntimeArm:
        try:
            return self.arms[name]
        except KeyError as exc:
            raise KeyError(f"cell has no arm named {name!r}") from exc

    def prepare_bimanual_calibration_dir(self) -> BimanualCalibrationLayout:
        """Create a temp calibration dir using LeRobot's `<base>_<arm>.json` names.

        Each arm's source calibration is resolved (in order) from its own
        ``calibration_file_path``, its own ``calibration_dir`` keyed by the arm's
        ``robot_id``, or—when the arm declares neither—the **cell-level**
        ``calibration_dir`` keyed by the arm's ``robot_id`` (``<robot_id>.json``).
        This mirrors how the cell's per-arm health check resolves calibration
        (``arm.calibration_dir or cell.calibration_dir``), so a config that only
        sets a top-level ``calibration_dir`` (per-arm ``robot_id`` only) works.
        """

        if not self.is_bimanual:
            raise RuntimeError("bimanual calibration requires left and right arms")
        robot_id = self.robot_id
        if not robot_id:
            raise RuntimeError("bimanual calibration requires ctx.cell.robot_id")
        calibration_dir = Path(tempfile.mkdtemp(prefix="armnet-bimanual-calibration-"))
        for arm_name in ("left", "right"):
            arm = self.arm(arm_name)
            source = arm.calibration_file_path
            if source is None:
                cal_dir = arm.calibration_dir or self.calibration_dir
                if cal_dir is not None and arm.robot_id:
                    source = Path(cal_dir) / f"{arm.robot_id}.json"
            if source is None or not Path(source).is_file():
                raise RuntimeError(
                    f"no calibration file found for {arm_name} arm "
                    f"(robot_id={arm.robot_id!r}); looked for "
                    f"{source if source is not None else '<unresolved>'}. Set the "
                    "cell-level calibration_dir (with per-arm robot_id) or each "
                    "arm's calibration_dir/calibration_file_path."
                )
            shutil.copy2(source, calibration_dir / f"{robot_id}_{arm_name}.json")
        return BimanualCalibrationLayout(robot_id=robot_id, calibration_dir=calibration_dir)

    def reset(self) -> None:
        """Return the robot to rest, then block until the operator confirms.

        Two concerns, two endpoints:

        1. Returning the arm to its rest position is a low-level bus operation,
           so it is sent to the robot connector (``robot_port``), which may be a
           headless edge device.
        2. Operator confirmation is a human-in-the-loop concern, so it is sent
           to the ``operator_call_endpoint`` served by the ``armnet-cell``
           process, whose stdin is the operator's terminal.

        The operator-facing prompt is owned by the cell, not by job code: job
        code only signals *that* a reset point has been reached.
        """

        self._report_progress("Waiting for workspace reset...")

        # 1. Safety: return the arm to rest via the robot connector, if present.
        if self.robot_port and _looks_like_connector_endpoint(self.robot_port):
            response = _connector_request(self.robot_port, {"op": "return_to_rest"})
            if not response.get("ok"):
                raise RuntimeError(response.get("error", "robot return-to-rest failed"))

        # 2. Operator confirmation on the cell-served operator-call endpoint
        # (fallback to the dev local-control endpoint).
        request = {"op": "reset", "request": {"kind": "manual"}}
        operator_endpoint = self.operator_call_endpoint or self.local_control_endpoint
        if operator_endpoint:
            response = _connector_request(operator_endpoint, request)
            if not response.get("ok"):
                error = response.get("error", "operator reset confirmation failed")
                if response.get("error_type") == "ResetTimeoutException":
                    raise ResetTimeoutException(error)
                raise RuntimeError(error)
            return

        # No operator endpoint attached (degenerate in-process dev run): block on
        # the local terminal with a standard prompt owned by the runtime.
        input("Reset the cell workspace, then press Enter. ")

    def is_complete(self, *, block: bool = False) -> CompletionStatus:
        """Return whether the current episode is complete, and its success.

        Returns a :class:`CompletionStatus` ``(complete, success)``:

        1. A human-reported outcome (an operator hitting success/fail in the
           FMS during a live rollout) takes precedence and ends the episode
           immediately, with ``success`` set to the operator's choice. This is
           how an operator stops a dangerous rollout without stopping the job.
        2. Otherwise the cell's automated completion monitor is consulted; a
           task scored complete is reported as a success (``success ==
           complete``). Pass ``block=True`` for a final episode check that waits
           for the cell to score the latest cached frames before returning.

        ``bool(status)`` is ``status.complete`` for backwards compatibility.
        """

        # 1. Human-reported outcome (operator/FMS) wins and ends the episode now.
        reported = self._human_completion()
        if reported is not None:
            return CompletionStatus(complete=True, success=reported)

        # 2. Automated completion scoring via the local-dev or cell connector.
        request = {"op": "is_complete", "block": block}
        if self.local_control_endpoint:
            response = _connector_request(self.local_control_endpoint, request)
            if not response.get("ok"):
                raise RuntimeError(response.get("error", "local completion check failed"))
            return _completion_from_response(response)
        if self.robot_port and _looks_like_connector_endpoint(self.robot_port):
            response = _connector_request(self.robot_port, request)
            if not response.get("ok"):
                raise RuntimeError(response.get("error", "cell completion check failed"))
            return _completion_from_response(response)
        return CompletionStatus(complete=False, success=False)

    def _human_completion(self) -> Optional[bool]:
        """Return the operator's reported success/fail, or None if none pending.

        Polls the cell program's operator-call endpoint (where FMS rollout
        commands land). A wedged/slow channel must never stall the control
        loop, so this mirrors get_teleop_action: bounded read, throttled error
        logging, drop the connection on error, and treat failures as "no
        outcome" so the episode simply continues.
        """

        endpoint = self.operator_call_endpoint or self.local_control_endpoint
        if not endpoint:
            return None

        if self._completion_conn is None or self._completion_conn.endpoint != endpoint:
            if self._completion_conn is not None:
                self._completion_conn.close()
            self._completion_conn = _TeleopConnection(endpoint, timeout=_TELEOP_READ_TIMEOUT_S)

        try:
            response = self._completion_conn.request({"op": "get_completion"})
        except Exception as exc:  # noqa: BLE001
            now = time.monotonic()
            if now - self._completion_last_error_log >= _TELEOP_ERROR_LOG_INTERVAL_S:
                self._completion_last_error_log = now
                logger.warning(
                    "completion read from %s failed (treating as not complete): %r",
                    endpoint,
                    exc,
                )
            return None

        if not response.get("ok") or not response.get("reported"):
            return None
        return bool(response.get("success", False))

    def rollout_begin(
        self,
        *,
        index: Optional[int] = None,
        total: Optional[int] = None,
        outcome_controls: bool = True,
    ) -> None:
        """Tell the platform a rollout/episode in this job's loop has started.

        The cell publishes this to the FMS, which shows the loop progress
        ("rollout N / M") for the live job. Pass ``index`` (1-based) and, when
        known, ``total`` so operators see how far along the loop is.

        ``outcome_controls`` controls whether the FMS also shows operator
        success/fail buttons: keep the default ``True`` for policy evals; pass
        ``False`` for progress-only loops such as teleop data collection, where
        a human verdict doesn't apply. Best-effort: a failed notification never
        breaks the rollout. Pair with :meth:`rollout_end`.
        """
        payload: dict[str, Any] = {"outcome_controls": bool(outcome_controls)}
        if index is not None:
            payload["index"] = int(index)
        if total is not None:
            payload["total"] = int(total)
        self._rollout_signal("rollout_begin", **payload)

    def rollout_end(self) -> None:
        """Tell the platform the current rollout has ended (hides FMS buttons)."""
        self._rollout_signal("rollout_end")

    def _rollout_signal(self, op: str, **payload: Any) -> None:
        endpoint = self.operator_call_endpoint or self.local_control_endpoint
        if not endpoint:
            return
        try:
            _connector_request(endpoint, {"op": op, **payload})
        except Exception:  # noqa: BLE001 - rollout signalling is best-effort
            logger.warning("rollout signal %s to %s failed", op, endpoint, exc_info=True)

    def is_shutting_down(self) -> bool:
        """Return True once the cell has entered the job's post-timeout grace window.

        When a job exceeds its ``timeout_seconds`` the cell does not kill the
        container straight away: it trips the robot interlock (so any further
        robot-bus calls fail) and opens a short *grace window* during which this
        returns True, before force-killing the container. Poll it in your loop
        and break out to finalize gracefully — e.g. save/push a dataset — instead
        of being killed mid-write::

            for episode in range(n):
                if ctx.cell.is_shutting_down():
                    break  # finalize below
                ...

        Resilient by design: returns False when no cell/operator endpoint is
        attached or the status can't be read, so it never stalls or crashes the
        control loop.
        """

        endpoint = self.operator_call_endpoint or self.local_control_endpoint
        if not endpoint:
            return False
        try:
            response = _connector_request(
                endpoint, {"op": "shutdown_status"}, read_timeout=_TELEOP_READ_TIMEOUT_S
            )
        except Exception:  # noqa: BLE001 - never let a status poll break the loop
            return False
        return bool(response.get("ok") and response.get("shutting_down"))

    def should_stop(self) -> bool:
        """Return True when local/remote control asks user code to stop safely."""

        request = {"op": "should_stop"}
        if self.local_control_endpoint:
            response = _connector_request(self.local_control_endpoint, request)
            if not response.get("ok"):
                raise RuntimeError(response.get("error", "local stop check failed"))
            return bool(response.get("stop", False))
        return False

    def get_teleop_action(self) -> Optional[dict[str, float]]:
        """Return the freshest remote-teleoperation action for this job, or None.

        The client samples a local leader arm and pushes actions to the cell,
        which keeps only the most recent one (older messages are dropped). This
        reads that most-recent-value register over the operator-call endpoint.

        Returns ``None`` when no teleop has been received yet (or no operator
        endpoint is attached), so a control loop can hold position until the
        operator starts driving. The returned dict is keyed for LeRobot's
        ``send_action`` (e.g. ``{"shoulder_pan.pos": 12.3, ...}``).
        """

        endpoint = self.operator_call_endpoint or self.local_control_endpoint
        if not endpoint:
            return None

        if self._teleop_conn is None or self._teleop_conn.endpoint != endpoint:
            if self._teleop_conn is not None:
                self._teleop_conn.close()
            self._teleop_conn = _TeleopConnection(endpoint, timeout=_TELEOP_READ_TIMEOUT_S)

        try:
            response = self._teleop_conn.request({"op": "get_teleop"})
        except Exception as exc:  # noqa: BLE001
            # A wedged/slow teleop channel must not stall or crash the control
            # loop: log (throttled) so a recurrence is diagnosable, drop the
            # connection (already done in request()) so we reconnect next tick,
            # and hold position by returning None.
            now = time.monotonic()
            if now - self._teleop_last_error_log >= _TELEOP_ERROR_LOG_INTERVAL_S:
                self._teleop_last_error_log = now
                logger.warning(
                    "teleop read from %s failed (holding position; will reconnect): %r",
                    endpoint,
                    exc,
                )
            return None

        if not response.get("ok"):
            logger.warning("teleop read returned error: %s", response.get("error"))
            return None
        action = response.get("action")
        if not action:
            return None
        return {str(key): float(value) for key, value in action.items()}

    def get_teleop_event(self) -> Optional[str]:
        """Return the next pending recording-control event, or None.

        While teleoperating, the client can send discrete recording-control
        events alongside the action stream — LeRobot's standard dataset
        recording shortcuts: ``"next_episode"`` (Right Arrow: save the episode
        and move on), ``"rerecord_episode"`` (Left Arrow: discard and redo) and
        ``"stop_recording"`` (Esc: end the session). The cell queues them in
        arrival order; each call pops at most one.

        Like :meth:`get_teleop_action`, a wedged channel never stalls the
        control loop: errors log (throttled), drop the connection so the next
        call reconnects, and return None.
        """

        endpoint = self.operator_call_endpoint or self.local_control_endpoint
        if not endpoint:
            return None

        if self._teleop_conn is None or self._teleop_conn.endpoint != endpoint:
            if self._teleop_conn is not None:
                self._teleop_conn.close()
            self._teleop_conn = _TeleopConnection(endpoint, timeout=_TELEOP_READ_TIMEOUT_S)

        try:
            response = self._teleop_conn.request({"op": "get_teleop_event"})
        except Exception as exc:  # noqa: BLE001
            now = time.monotonic()
            if now - self._teleop_last_error_log >= _TELEOP_ERROR_LOG_INTERVAL_S:
                self._teleop_last_error_log = now
                logger.warning(
                    "teleop event read from %s failed (will reconnect): %r",
                    endpoint,
                    exc,
                )
            return None

        if not response.get("ok"):
            logger.warning("teleop event read returned error: %s", response.get("error"))
            return None
        event = response.get("event")
        return str(event) if event else None

    def _report_progress(self, message: str) -> None:
        """Surface a progress message back to the platform.

        M0.5: prints to stdout with a discoverable marker so the cell's
        captured stdout shows progress in order with other prints. M1+
        will also publish a NATS message so the orchestrator can stream
        progress back to the client without waiting for the job to
        terminate.
        """

        # Imported locally to avoid pulling markers into the public API
        # surface of `Context`.
        from armnet_runtime.markers import PROGRESS_MARKER
        print(f"{PROGRESS_MARKER} {message}", flush=True)
        time.sleep(0.01)

robot_port class-attribute instance-attribute

robot_port: Optional[str] = None

Robot port value to pass into LeRobot robot configs.

In container-backed remote execution this is the connector endpoint, not the host's physical serial path. The SDK's import-system swap routes that endpoint through the cell-side connector, which then opens the real robot port configured on the cell host.

robot_id class-attribute instance-attribute

robot_id: Optional[str] = None

Stable robot id used by LeRobot to find calibration data.

calibration_dir class-attribute instance-attribute

calibration_dir: Optional[Path] = None

Calibration store path visible inside the customer container.

calibration_file_path class-attribute instance-attribute

calibration_file_path: Optional[Path] = None

Exact LeRobot calibration file path visible inside the customer container.

language_instruction class-attribute instance-attribute

language_instruction: Optional[str] = None

Task instruction provided by the cell.

local_control_endpoint class-attribute instance-attribute

local_control_endpoint: Optional[str] = None

Developer local-container control endpoint for keyboard-driven state.

operator_call_endpoint class-attribute instance-attribute

operator_call_endpoint: Optional[str] = None

Operator-call endpoint served by the cell program for human-in-the-loop calls (manual reset confirmation). Distinct from robot_port, which is the robot/bus connector (potentially a headless edge device).

is_local_container class-attribute instance-attribute

is_local_container: bool = False

True when running a Docker image locally for development.

safety_limit class-attribute instance-attribute

safety_limit: Optional[float] = None

Relative action safety limit exposed by the cell, if applicable.

arms class-attribute instance-attribute

arms: dict[str, RuntimeArm] = field(default_factory=dict)

Named arms for bimanual/multi-arm cells.

is_bimanual property

is_bimanual: bool

arm

arm(name: str) -> RuntimeArm

Source code in runtime/src/armnet_runtime/context.py

def arm(self, name: str) -> RuntimeArm:
    try:
        return self.arms[name]
    except KeyError as exc:
        raise KeyError(f"cell has no arm named {name!r}") from exc

prepare_bimanual_calibration_dir

prepare_bimanual_calibration_dir() -> BimanualCalibrationLayout

Create a temp calibration dir using LeRobot's <base>_<arm>.json names.

Each arm's source calibration is resolved (in order) from its own calibration_file_path, its own calibration_dir keyed by the arm's robot_id, or—when the arm declares neither—the cell-level calibration_dir keyed by the arm's robot_id (<robot_id>.json). This mirrors how the cell's per-arm health check resolves calibration (arm.calibration_dir or cell.calibration_dir), so a config that only sets a top-level calibration_dir (per-arm robot_id only) works.

Source code in runtime/src/armnet_runtime/context.py

def prepare_bimanual_calibration_dir(self) -> BimanualCalibrationLayout:
    """Create a temp calibration dir using LeRobot's `<base>_<arm>.json` names.

    Each arm's source calibration is resolved (in order) from its own
    ``calibration_file_path``, its own ``calibration_dir`` keyed by the arm's
    ``robot_id``, or—when the arm declares neither—the **cell-level**
    ``calibration_dir`` keyed by the arm's ``robot_id`` (``<robot_id>.json``).
    This mirrors how the cell's per-arm health check resolves calibration
    (``arm.calibration_dir or cell.calibration_dir``), so a config that only
    sets a top-level ``calibration_dir`` (per-arm ``robot_id`` only) works.
    """

    if not self.is_bimanual:
        raise RuntimeError("bimanual calibration requires left and right arms")
    robot_id = self.robot_id
    if not robot_id:
        raise RuntimeError("bimanual calibration requires ctx.cell.robot_id")
    calibration_dir = Path(tempfile.mkdtemp(prefix="armnet-bimanual-calibration-"))
    for arm_name in ("left", "right"):
        arm = self.arm(arm_name)
        source = arm.calibration_file_path
        if source is None:
            cal_dir = arm.calibration_dir or self.calibration_dir
            if cal_dir is not None and arm.robot_id:
                source = Path(cal_dir) / f"{arm.robot_id}.json"
        if source is None or not Path(source).is_file():
            raise RuntimeError(
                f"no calibration file found for {arm_name} arm "
                f"(robot_id={arm.robot_id!r}); looked for "
                f"{source if source is not None else '<unresolved>'}. Set the "
                "cell-level calibration_dir (with per-arm robot_id) or each "
                "arm's calibration_dir/calibration_file_path."
            )
        shutil.copy2(source, calibration_dir / f"{robot_id}_{arm_name}.json")
    return BimanualCalibrationLayout(robot_id=robot_id, calibration_dir=calibration_dir)

reset

reset() -> None

Return the robot to rest, then block until the operator confirms.

Two concerns, two endpoints:

1. Returning the arm to its rest position is a low-level bus operation, so it is sent to the robot connector (robot_port), which may be a headless edge device.
2. Operator confirmation is a human-in-the-loop concern, so it is sent to the operator_call_endpoint served by the armnet-cell process, whose stdin is the operator's terminal.

The operator-facing prompt is owned by the cell, not by job code: job code only signals that a reset point has been reached.

Source code in runtime/src/armnet_runtime/context.py

def reset(self) -> None:
    """Return the robot to rest, then block until the operator confirms.

    Two concerns, two endpoints:

    1. Returning the arm to its rest position is a low-level bus operation,
       so it is sent to the robot connector (``robot_port``), which may be a
       headless edge device.
    2. Operator confirmation is a human-in-the-loop concern, so it is sent
       to the ``operator_call_endpoint`` served by the ``armnet-cell``
       process, whose stdin is the operator's terminal.

    The operator-facing prompt is owned by the cell, not by job code: job
    code only signals *that* a reset point has been reached.
    """

    self._report_progress("Waiting for workspace reset...")

    # 1. Safety: return the arm to rest via the robot connector, if present.
    if self.robot_port and _looks_like_connector_endpoint(self.robot_port):
        response = _connector_request(self.robot_port, {"op": "return_to_rest"})
        if not response.get("ok"):
            raise RuntimeError(response.get("error", "robot return-to-rest failed"))

    # 2. Operator confirmation on the cell-served operator-call endpoint
    # (fallback to the dev local-control endpoint).
    request = {"op": "reset", "request": {"kind": "manual"}}
    operator_endpoint = self.operator_call_endpoint or self.local_control_endpoint
    if operator_endpoint:
        response = _connector_request(operator_endpoint, request)
        if not response.get("ok"):
            error = response.get("error", "operator reset confirmation failed")
            if response.get("error_type") == "ResetTimeoutException":
                raise ResetTimeoutException(error)
            raise RuntimeError(error)
        return

    # No operator endpoint attached (degenerate in-process dev run): block on
    # the local terminal with a standard prompt owned by the runtime.
    input("Reset the cell workspace, then press Enter. ")

is_complete

is_complete(*, block: bool = False) -> CompletionStatus

Return whether the current episode is complete, and its success.

Returns a :class:CompletionStatus (complete, success):

1. A human-reported outcome (an operator hitting success/fail in the FMS during a live rollout) takes precedence and ends the episode immediately, with success set to the operator's choice. This is how an operator stops a dangerous rollout without stopping the job.
2. Otherwise the cell's automated completion monitor is consulted; a task scored complete is reported as a success (success == complete). Pass block=True for a final episode check that waits for the cell to score the latest cached frames before returning.

bool(status) is status.complete for backwards compatibility.

Source code in runtime/src/armnet_runtime/context.py

def is_complete(self, *, block: bool = False) -> CompletionStatus:
    """Return whether the current episode is complete, and its success.

    Returns a :class:`CompletionStatus` ``(complete, success)``:

    1. A human-reported outcome (an operator hitting success/fail in the
       FMS during a live rollout) takes precedence and ends the episode
       immediately, with ``success`` set to the operator's choice. This is
       how an operator stops a dangerous rollout without stopping the job.
    2. Otherwise the cell's automated completion monitor is consulted; a
       task scored complete is reported as a success (``success ==
       complete``). Pass ``block=True`` for a final episode check that waits
       for the cell to score the latest cached frames before returning.

    ``bool(status)`` is ``status.complete`` for backwards compatibility.
    """

    # 1. Human-reported outcome (operator/FMS) wins and ends the episode now.
    reported = self._human_completion()
    if reported is not None:
        return CompletionStatus(complete=True, success=reported)

    # 2. Automated completion scoring via the local-dev or cell connector.
    request = {"op": "is_complete", "block": block}
    if self.local_control_endpoint:
        response = _connector_request(self.local_control_endpoint, request)
        if not response.get("ok"):
            raise RuntimeError(response.get("error", "local completion check failed"))
        return _completion_from_response(response)
    if self.robot_port and _looks_like_connector_endpoint(self.robot_port):
        response = _connector_request(self.robot_port, request)
        if not response.get("ok"):
            raise RuntimeError(response.get("error", "cell completion check failed"))
        return _completion_from_response(response)
    return CompletionStatus(complete=False, success=False)

rollout_begin

rollout_begin(*, index: Optional[int] = None, total: Optional[int] = None, outcome_controls: bool = True) -> None

Tell the platform a rollout/episode in this job's loop has started.

The cell publishes this to the FMS, which shows the loop progress ("rollout N / M") for the live job. Pass index (1-based) and, when known, total so operators see how far along the loop is.

outcome_controls controls whether the FMS also shows operator success/fail buttons: keep the default True for policy evals; pass False for progress-only loops such as teleop data collection, where a human verdict doesn't apply. Best-effort: a failed notification never breaks the rollout. Pair with :meth:rollout_end.

Source code in runtime/src/armnet_runtime/context.py

def rollout_begin(
    self,
    *,
    index: Optional[int] = None,
    total: Optional[int] = None,
    outcome_controls: bool = True,
) -> None:
    """Tell the platform a rollout/episode in this job's loop has started.

    The cell publishes this to the FMS, which shows the loop progress
    ("rollout N / M") for the live job. Pass ``index`` (1-based) and, when
    known, ``total`` so operators see how far along the loop is.

    ``outcome_controls`` controls whether the FMS also shows operator
    success/fail buttons: keep the default ``True`` for policy evals; pass
    ``False`` for progress-only loops such as teleop data collection, where
    a human verdict doesn't apply. Best-effort: a failed notification never
    breaks the rollout. Pair with :meth:`rollout_end`.
    """
    payload: dict[str, Any] = {"outcome_controls": bool(outcome_controls)}
    if index is not None:
        payload["index"] = int(index)
    if total is not None:
        payload["total"] = int(total)
    self._rollout_signal("rollout_begin", **payload)

rollout_end

rollout_end() -> None

Tell the platform the current rollout has ended (hides FMS buttons).

Source code in runtime/src/armnet_runtime/context.py

def rollout_end(self) -> None:
    """Tell the platform the current rollout has ended (hides FMS buttons)."""
    self._rollout_signal("rollout_end")

is_shutting_down

is_shutting_down() -> bool

Return True once the cell has entered the job's post-timeout grace window.

When a job exceeds its timeout_seconds the cell does not kill the container straight away: it trips the robot interlock (so any further robot-bus calls fail) and opens a short grace window during which this returns True, before force-killing the container. Poll it in your loop and break out to finalize gracefully — e.g. save/push a dataset — instead of being killed mid-write::

for episode in range(n):
    if ctx.cell.is_shutting_down():
        break  # finalize below
    ...

Resilient by design: returns False when no cell/operator endpoint is attached or the status can't be read, so it never stalls or crashes the control loop.

Source code in runtime/src/armnet_runtime/context.py

def is_shutting_down(self) -> bool:
    """Return True once the cell has entered the job's post-timeout grace window.

    When a job exceeds its ``timeout_seconds`` the cell does not kill the
    container straight away: it trips the robot interlock (so any further
    robot-bus calls fail) and opens a short *grace window* during which this
    returns True, before force-killing the container. Poll it in your loop
    and break out to finalize gracefully — e.g. save/push a dataset — instead
    of being killed mid-write::

        for episode in range(n):
            if ctx.cell.is_shutting_down():
                break  # finalize below
            ...

    Resilient by design: returns False when no cell/operator endpoint is
    attached or the status can't be read, so it never stalls or crashes the
    control loop.
    """

    endpoint = self.operator_call_endpoint or self.local_control_endpoint
    if not endpoint:
        return False
    try:
        response = _connector_request(
            endpoint, {"op": "shutdown_status"}, read_timeout=_TELEOP_READ_TIMEOUT_S
        )
    except Exception:  # noqa: BLE001 - never let a status poll break the loop
        return False
    return bool(response.get("ok") and response.get("shutting_down"))

should_stop

should_stop() -> bool

Return True when local/remote control asks user code to stop safely.

Source code in runtime/src/armnet_runtime/context.py

def should_stop(self) -> bool:
    """Return True when local/remote control asks user code to stop safely."""

    request = {"op": "should_stop"}
    if self.local_control_endpoint:
        response = _connector_request(self.local_control_endpoint, request)
        if not response.get("ok"):
            raise RuntimeError(response.get("error", "local stop check failed"))
        return bool(response.get("stop", False))
    return False

get_teleop_action

get_teleop_action() -> Optional[dict[str, float]]

Return the freshest remote-teleoperation action for this job, or None.

The client samples a local leader arm and pushes actions to the cell, which keeps only the most recent one (older messages are dropped). This reads that most-recent-value register over the operator-call endpoint.

Returns None when no teleop has been received yet (or no operator endpoint is attached), so a control loop can hold position until the operator starts driving. The returned dict is keyed for LeRobot's send_action (e.g. {"shoulder_pan.pos": 12.3, ...}).

Source code in runtime/src/armnet_runtime/context.py

def get_teleop_action(self) -> Optional[dict[str, float]]:
    """Return the freshest remote-teleoperation action for this job, or None.

    The client samples a local leader arm and pushes actions to the cell,
    which keeps only the most recent one (older messages are dropped). This
    reads that most-recent-value register over the operator-call endpoint.

    Returns ``None`` when no teleop has been received yet (or no operator
    endpoint is attached), so a control loop can hold position until the
    operator starts driving. The returned dict is keyed for LeRobot's
    ``send_action`` (e.g. ``{"shoulder_pan.pos": 12.3, ...}``).
    """

    endpoint = self.operator_call_endpoint or self.local_control_endpoint
    if not endpoint:
        return None

    if self._teleop_conn is None or self._teleop_conn.endpoint != endpoint:
        if self._teleop_conn is not None:
            self._teleop_conn.close()
        self._teleop_conn = _TeleopConnection(endpoint, timeout=_TELEOP_READ_TIMEOUT_S)

    try:
        response = self._teleop_conn.request({"op": "get_teleop"})
    except Exception as exc:  # noqa: BLE001
        # A wedged/slow teleop channel must not stall or crash the control
        # loop: log (throttled) so a recurrence is diagnosable, drop the
        # connection (already done in request()) so we reconnect next tick,
        # and hold position by returning None.
        now = time.monotonic()
        if now - self._teleop_last_error_log >= _TELEOP_ERROR_LOG_INTERVAL_S:
            self._teleop_last_error_log = now
            logger.warning(
                "teleop read from %s failed (holding position; will reconnect): %r",
                endpoint,
                exc,
            )
        return None

    if not response.get("ok"):
        logger.warning("teleop read returned error: %s", response.get("error"))
        return None
    action = response.get("action")
    if not action:
        return None
    return {str(key): float(value) for key, value in action.items()}

get_teleop_event

get_teleop_event() -> Optional[str]

Return the next pending recording-control event, or None.

While teleoperating, the client can send discrete recording-control events alongside the action stream — LeRobot's standard dataset recording shortcuts: "next_episode" (Right Arrow: save the episode and move on), "rerecord_episode" (Left Arrow: discard and redo) and "stop_recording" (Esc: end the session). The cell queues them in arrival order; each call pops at most one.

Like :meth:get_teleop_action, a wedged channel never stalls the control loop: errors log (throttled), drop the connection so the next call reconnects, and return None.

Source code in runtime/src/armnet_runtime/context.py

def get_teleop_event(self) -> Optional[str]:
    """Return the next pending recording-control event, or None.

    While teleoperating, the client can send discrete recording-control
    events alongside the action stream — LeRobot's standard dataset
    recording shortcuts: ``"next_episode"`` (Right Arrow: save the episode
    and move on), ``"rerecord_episode"`` (Left Arrow: discard and redo) and
    ``"stop_recording"`` (Esc: end the session). The cell queues them in
    arrival order; each call pops at most one.

    Like :meth:`get_teleop_action`, a wedged channel never stalls the
    control loop: errors log (throttled), drop the connection so the next
    call reconnects, and return None.
    """

    endpoint = self.operator_call_endpoint or self.local_control_endpoint
    if not endpoint:
        return None

    if self._teleop_conn is None or self._teleop_conn.endpoint != endpoint:
        if self._teleop_conn is not None:
            self._teleop_conn.close()
        self._teleop_conn = _TeleopConnection(endpoint, timeout=_TELEOP_READ_TIMEOUT_S)

    try:
        response = self._teleop_conn.request({"op": "get_teleop_event"})
    except Exception as exc:  # noqa: BLE001
        now = time.monotonic()
        if now - self._teleop_last_error_log >= _TELEOP_ERROR_LOG_INTERVAL_S:
            self._teleop_last_error_log = now
            logger.warning(
                "teleop event read from %s failed (will reconnect): %r",
                endpoint,
                exc,
            )
        return None

    if not response.get("ok"):
        logger.warning("teleop event read returned error: %s", response.get("error"))
        return None
    event = response.get("event")
    return str(event) if event else None

Context dataclass

Everything a @main-decorated function needs from the platform.

Source code in runtime/src/armnet_runtime/context.py

@dataclass
class Context:
    """Everything a ``@main``-decorated function needs from the platform."""

    job_id: str
    embodiment: Embodiment
    task: Task
    args: dict[str, Any] = field(default_factory=dict)
    cell: Cell = field(default_factory=Cell)
    camera_configs: dict[str, Any] = field(default_factory=dict)
    cache_home: Optional[Path] = None
    volume: Volume = field(default_factory=Volume)
    secrets: dict[str, str] = field(default_factory=dict)
    timeout_seconds: Optional[int] = None
    # Lazily created background Rerun streamer (see log_rerun_data). Not part of
    # the constructor or the public/comparable surface.
    _rerun_streamer: Any = field(default=None, init=False, repr=False, compare=False)

    def report_progress(self, message: str) -> None:
        """Surface a progress message back to the platform.

        M0.5: prints to stdout with a discoverable marker so the cell's
        captured stdout shows progress in order with other prints. M1+
        will also publish a NATS message so the orchestrator can stream
        progress back to the client without waiting for the job to
        terminate.
        """

        # Imported locally to avoid pulling markers into the public API
        # surface of `Context`.
        from armnet_runtime.markers import PROGRESS_MARKER
        print(f"{PROGRESS_MARKER} {message}", flush=True)

    def is_shutting_down(self) -> bool:
        """Whether the cell has entered the job's post-timeout grace window.

        Convenience delegate for :meth:`Cell.is_shutting_down`. Poll it in long
        loops and break out to finalize gracefully before the cell kills the
        container.
        """
        return self.cell.is_shutting_down()

    def log_rerun_data(
        self,
        observation: dict[str, Any] | None = None,
        action: dict[str, Any] | None = None,
        *,
        compress_images: bool = True,
        jpeg_quality: int = 75,
    ) -> None:
        """Stream observation/action data to a Rerun viewer on the client.

        Mirrors LeRobot's ``log_rerun_data``: scalars are logged as Rerun
        scalars, image-like arrays as images, and other arrays as per-element
        scalars. Keys are namespaced with ``observation.`` / ``action.`` when
        not already.

        Unlike the LeRobot helper, this does not call ``rr.log`` in-process
        (the cell container has no viewer). Instead it serializes a protobuf
        packet and emits it on stdout behind a marker; the cell republishes it
        on ``logs.<job_id>.rerun`` and the client's orchestrate script replays
        it into the viewer it started with ``rr.init(...)``.

        Images are JPEG-compressed by default to keep the NATS stream light;
        set ``compress_images=False`` to send raw RGB. opencv is required for
        compression and numpy for any array handling; both are imported lazily.

        Non-blocking: the snapshot is handed to a background worker thread that
        does the encoding and stdout write, so the calling control loop never
        stalls on visualization. The worker's queue is bounded and drops the
        oldest pending frame under backpressure (tune with
        ``ARMNET_RERUN_QUEUE_MAXSIZE``), so a slow consumer sheds frames
        rather than slowing the robot loop.
        """

        if not observation and not action:
            return

        from armnet_runtime.rerun import RerunStreamer

        if self._rerun_streamer is None:
            self._rerun_streamer = RerunStreamer(self.job_id)
            self._rerun_streamer.start()
        self._rerun_streamer.submit(
            observation,
            action,
            compress_images=compress_images,
            jpeg_quality=jpeg_quality,
        )

job_id instance-attribute

job_id: str

embodiment instance-attribute

embodiment: Embodiment

task instance-attribute

task: Task

args class-attribute instance-attribute

args: dict[str, Any] = field(default_factory=dict)

cell class-attribute instance-attribute

cell: Cell = field(default_factory=Cell)

camera_configs class-attribute instance-attribute

camera_configs: dict[str, Any] = field(default_factory=dict)

cache_home class-attribute instance-attribute

cache_home: Optional[Path] = None

volume class-attribute instance-attribute

volume: Volume = field(default_factory=Volume)

secrets class-attribute instance-attribute

secrets: dict[str, str] = field(default_factory=dict)

timeout_seconds class-attribute instance-attribute

timeout_seconds: Optional[int] = None

report_progress

report_progress(message: str) -> None

Surface a progress message back to the platform.

M0.5: prints to stdout with a discoverable marker so the cell's captured stdout shows progress in order with other prints. M1+ will also publish a NATS message so the orchestrator can stream progress back to the client without waiting for the job to terminate.

Source code in runtime/src/armnet_runtime/context.py

def report_progress(self, message: str) -> None:
    """Surface a progress message back to the platform.

    M0.5: prints to stdout with a discoverable marker so the cell's
    captured stdout shows progress in order with other prints. M1+
    will also publish a NATS message so the orchestrator can stream
    progress back to the client without waiting for the job to
    terminate.
    """

    # Imported locally to avoid pulling markers into the public API
    # surface of `Context`.
    from armnet_runtime.markers import PROGRESS_MARKER
    print(f"{PROGRESS_MARKER} {message}", flush=True)

is_shutting_down

is_shutting_down() -> bool

Whether the cell has entered the job's post-timeout grace window.

Convenience delegate for :meth:Cell.is_shutting_down. Poll it in long loops and break out to finalize gracefully before the cell kills the container.

Source code in runtime/src/armnet_runtime/context.py

def is_shutting_down(self) -> bool:
    """Whether the cell has entered the job's post-timeout grace window.

    Convenience delegate for :meth:`Cell.is_shutting_down`. Poll it in long
    loops and break out to finalize gracefully before the cell kills the
    container.
    """
    return self.cell.is_shutting_down()

log_rerun_data

log_rerun_data(observation: dict[str, Any] | None = None, action: dict[str, Any] | None = None, *, compress_images: bool = True, jpeg_quality: int = 75) -> None

Stream observation/action data to a Rerun viewer on the client.

Mirrors LeRobot's log_rerun_data: scalars are logged as Rerun scalars, image-like arrays as images, and other arrays as per-element scalars. Keys are namespaced with observation. / action. when not already.

Unlike the LeRobot helper, this does not call rr.log in-process (the cell container has no viewer). Instead it serializes a protobuf packet and emits it on stdout behind a marker; the cell republishes it on logs.<job_id>.rerun and the client's orchestrate script replays it into the viewer it started with rr.init(...).

Images are JPEG-compressed by default to keep the NATS stream light; set compress_images=False to send raw RGB. opencv is required for compression and numpy for any array handling; both are imported lazily.

Non-blocking: the snapshot is handed to a background worker thread that does the encoding and stdout write, so the calling control loop never stalls on visualization. The worker's queue is bounded and drops the oldest pending frame under backpressure (tune with ARMNET_RERUN_QUEUE_MAXSIZE), so a slow consumer sheds frames rather than slowing the robot loop.

Source code in runtime/src/armnet_runtime/context.py

def log_rerun_data(
    self,
    observation: dict[str, Any] | None = None,
    action: dict[str, Any] | None = None,
    *,
    compress_images: bool = True,
    jpeg_quality: int = 75,
) -> None:
    """Stream observation/action data to a Rerun viewer on the client.

    Mirrors LeRobot's ``log_rerun_data``: scalars are logged as Rerun
    scalars, image-like arrays as images, and other arrays as per-element
    scalars. Keys are namespaced with ``observation.`` / ``action.`` when
    not already.

    Unlike the LeRobot helper, this does not call ``rr.log`` in-process
    (the cell container has no viewer). Instead it serializes a protobuf
    packet and emits it on stdout behind a marker; the cell republishes it
    on ``logs.<job_id>.rerun`` and the client's orchestrate script replays
    it into the viewer it started with ``rr.init(...)``.

    Images are JPEG-compressed by default to keep the NATS stream light;
    set ``compress_images=False`` to send raw RGB. opencv is required for
    compression and numpy for any array handling; both are imported lazily.

    Non-blocking: the snapshot is handed to a background worker thread that
    does the encoding and stdout write, so the calling control loop never
    stalls on visualization. The worker's queue is bounded and drops the
    oldest pending frame under backpressure (tune with
    ``ARMNET_RERUN_QUEUE_MAXSIZE``), so a slow consumer sheds frames
    rather than slowing the robot loop.
    """

    if not observation and not action:
        return

    from armnet_runtime.rerun import RerunStreamer

    if self._rerun_streamer is None:
        self._rerun_streamer = RerunStreamer(self.job_id)
        self._rerun_streamer.start()
    self._rerun_streamer.submit(
        observation,
        action,
        compress_images=compress_images,
        jpeg_quality=jpeg_quality,
    )

require_so101_embodiment

require_so101_embodiment(ctx: 'Context', runtime_name: str) -> bool

Validate the job's embodiment is a (single or bimanual) SO-101.

The embodiment is the source of truth for how many arms the robot has — lerobot/so-101 is a single arm, lerobot/bimanual_so101 is two — and the orchestrator only routes a job to a cell of the matching embodiment. Returns True for the bimanual embodiment (so the caller builds a two-arm robot), False for single-arm.

Raises :class:NotImplementedError for any other embodiment, and :class:RuntimeError if the embodiment's arm count disagrees with the cell's actual wiring (ctx.cell.is_bimanual) — a misrouted or misconfigured cell.

Source code in runtime/src/armnet_runtime/context.py

def require_so101_embodiment(ctx: "Context", runtime_name: str) -> bool:
    """Validate the job's embodiment is a (single or bimanual) SO-101.

    The embodiment is the source of truth for how many arms the robot has —
    ``lerobot/so-101`` is a single arm, ``lerobot/bimanual_so101`` is two — and
    the orchestrator only routes a job to a cell of the matching embodiment.
    Returns ``True`` for the bimanual embodiment (so the caller builds a two-arm
    robot), ``False`` for single-arm.

    Raises :class:`NotImplementedError` for any other embodiment, and
    :class:`RuntimeError` if the embodiment's arm count disagrees with the cell's
    actual wiring (``ctx.cell.is_bimanual``) — a misrouted or misconfigured cell.
    """
    if ctx.embodiment not in (SO101_EMBODIMENT, BIMANUAL_SO101_EMBODIMENT):
        raise NotImplementedError(
            f"{runtime_name} supports {SO101_EMBODIMENT!r} and "
            f"{BIMANUAL_SO101_EMBODIMENT!r}, got {ctx.embodiment!r}"
        )
    expect_bimanual = ctx.embodiment == BIMANUAL_SO101_EMBODIMENT
    if expect_bimanual != ctx.cell.is_bimanual:
        raise RuntimeError(
            f"embodiment {ctx.embodiment!r} expects "
            f"{'two arms (left+right)' if expect_bimanual else 'a single arm'}, "
            f"but the cell exposes arms={sorted(ctx.cell.arms)}"
        )
    return expect_bimanual

build_context

build_context() -> Context

Read the cell-injected env vars and construct a :class:Context.

Called by the armnet-runtime entrypoint before invoking the user's @main function. Raises :class:RuntimeError with a clear message if any required env var is missing — that indicates the program is being run outside a armnet cell.

Source code in runtime/src/armnet_runtime/context.py

def build_context() -> Context:
    """Read the cell-injected env vars and construct a :class:`Context`.

    Called by the ``armnet-runtime`` entrypoint before invoking the
    user's ``@main`` function. Raises :class:`RuntimeError` with a clear
    message if any required env var is missing \u2014 that indicates the
    program is being run outside a armnet cell.
    """

    job_id = os.environ.get(env_keys.JOB_ID)
    embodiment_raw = os.environ.get(env_keys.EMBODIMENT)
    task_raw = os.environ.get(env_keys.TASK)
    if not job_id or not embodiment_raw or not task_raw:
        raise RuntimeError(
            "armnet runtime env vars not set; this program is meant to "
            "be executed inside a armnet cell container. Missing one of: "
            f"{env_keys.JOB_ID}, {env_keys.EMBODIMENT}, {env_keys.TASK}."
        )

    timeout_raw = os.environ.get(env_keys.TIMEOUT_SECONDS)
    timeout = int(timeout_raw) if timeout_raw and timeout_raw.isdigit() else None

    args_raw = os.environ.get(env_keys.ARGS, "{}")
    try:
        args = json.loads(args_raw)
    except json.JSONDecodeError as exc:
        raise RuntimeError(
            f"{env_keys.ARGS} env var is not valid JSON: {exc}"
        ) from exc
    if not isinstance(args, dict):
        raise RuntimeError(
            f"{env_keys.ARGS} must decode to a JSON object (dict); got {type(args).__name__}."
        )

    cell_config = _load_cell_config()
    return Context(
        job_id=job_id,
        embodiment=embodiment_raw,
        task=task_raw,
        args=args,
        cell=Cell(
            robot_port=(
                os.environ.get(env_keys.CELL_SOCKET)
                or (cell_config.robot_connector_endpoint if cell_config else None)
                or (cell_config.connector_socket_path if cell_config else None)
                or (cell_config.robot_port if cell_config else None)
            ),
            operator_call_endpoint=(
                os.environ.get(env_keys.OPERATOR_CALL_ENDPOINT)
                or (cell_config.operator_call_endpoint if cell_config else None)
            ),
            robot_id=os.environ.get(env_keys.ROBOT_ID)
            or (cell_config.robot_id if cell_config else None),
            calibration_dir=_path_from_env_or_config(
                env_keys.CALIBRATION_DIR,
                cell_config.calibration_dir if cell_config else None,
            ),
            calibration_file_path=_path_from_env_or_config(
                env_keys.CALIBRATION_FILE_PATH,
                cell_config.calibration_file_path if cell_config else None,
            ),
            language_instruction=cell_config.language_instruction if cell_config else None,
            local_control_endpoint=os.environ.get(env_keys.LOCAL_CONTROL_ENDPOINT),
            is_local_container=bool(os.environ.get(env_keys.LOCAL_CONTAINER)),
            safety_limit=_cell_safety_limit(cell_config, embodiment_raw),
            arms=_build_runtime_arms(cell_config),
        ),
        camera_configs=_build_camera_configs(
            cell_config.camera_configs if cell_config else {}
        ),
        cache_home=_path_from_env_or_config(env_keys.CACHE_HOME, None),
        volume=Volume(root=_path_from_env_or_config(env_keys.VOLUME_HOME, None)),
        secrets=_load_secrets(),
        timeout_seconds=timeout,
    )

armnet_runtime.decorator

The @main decorator and its module-level registry.

Customer code looks like::

from armnet_runtime import main, Context

@main
def run(ctx: Context):
    ...
    return {"success_rate": 1.0}

The decorator records run as the script's entry point. The armnet-runtime console script loads the script (which executes the decorator as a side effect) and then calls :func:registered_main to get the function to invoke.

Single entry point per script — multiple @main-decorated functions in the same file is almost certainly a bug, so the decorator raises rather than silently overwriting the previous registration.

EntryPoint module-attribute

EntryPoint = Callable[..., Any]

MainRegistrationError

Bases: RuntimeError

Raised when @main is used incorrectly (multiple times, etc.).

Source code in runtime/src/armnet_runtime/decorator.py

class MainRegistrationError(RuntimeError):
    """Raised when ``@main`` is used incorrectly (multiple times, etc.)."""

main

main(fn: EntryPoint) -> EntryPoint

Decorator: mark fn as the script's entry point.

The function is invoked with a :class:~armnet_runtime.Context by the armnet-runtime entrypoint. It may return any JSON-serialisable value; the value becomes :attr:~armnet_core.JobResult.return_value.

Source code in runtime/src/armnet_runtime/decorator.py

def main(fn: EntryPoint) -> EntryPoint:
    """Decorator: mark ``fn`` as the script's entry point.

    The function is invoked with a :class:`~armnet_runtime.Context` by
    the ``armnet-runtime`` entrypoint. It may return any
    JSON-serialisable value; the value becomes
    :attr:`~armnet_core.JobResult.return_value`.
    """

    global _registered
    if _registered is not None:
        raise MainRegistrationError(
            "armnet: multiple @main-decorated functions found "
            f"(already registered: {_registered.__module__}.{_registered.__qualname__}; "
            f"new: {fn.__module__}.{fn.__qualname__}). Only one entry point "
            "per container is supported."
        )
    _registered = fn
    return fn

registered_main

registered_main() -> Optional[EntryPoint]

Return the function previously registered with @main, if any.

Source code in runtime/src/armnet_runtime/decorator.py

def registered_main() -> Optional[EntryPoint]:
    """Return the function previously registered with ``@main``, if any."""

    return _registered

armnet_runtime.cli

armnet-runtime console script.

This is the container entrypoint for any image built on top of the armnet runtime SDK. The container's CMD looks like::

CMD ["armnet-runtime", "/app/hello.py"]

and this script:

1. (M3+) Performs the LeRobot import-system swap so customer code transparently routes hardware operations through the safety-aware robot connector. Currently a marker; nothing is swapped yet.
2. Adds the user script's directory to sys.path so it can import sibling modules.
3. Imports the user script — which executes the @main decorator as a side effect, registering the entry-point function.
4. Builds a :class:~armnet_runtime.Context from cell-injected env vars.
5. Calls the registered function with the context.
6. Prints the function's return value with the :data:~armnet_runtime.markers.RESULT_MARKER_JSON prefix; the cell extracts that and surfaces it as :attr:~armnet_core.JobResult.return_value.
7. Maps exceptions to a non-zero exit so the cell records the job as FAILED; the traceback ends up in the captured stderr.

EXIT_OK module-attribute

EXIT_OK = 0

EXIT_USER_RAISED module-attribute

EXIT_USER_RAISED = 1

EXIT_USAGE module-attribute

EXIT_USAGE = 2

EXIT_NO_MAIN module-attribute

EXIT_NO_MAIN = 3

EXIT_USER_IMPORT_FAILED module-attribute

EXIT_USER_IMPORT_FAILED = 4

EXIT_BAD_CONTEXT module-attribute

EXIT_BAD_CONTEXT = 5

run

run(argv: Sequence[str] | None = None) -> int

Source code in runtime/src/armnet_runtime/cli.py

def run(argv: Sequence[str] | None = None) -> int:
    args = _parse_args(sys.argv[1:] if argv is None else list(argv))
    _install_timestamped_streams()

    # Install guarded LeRobot import replacement before loading user code so
    # `from lerobot...` inside the script sees the remote Feetech proxy.
    from armnet_runtime.lerobot import install_lerobot_remote_mode

    install_lerobot_remote_mode()

    try:
        _load_user_script(args.script)
    except Exception:
        print("armnet-runtime: failed to import user script:", file=sys.stderr)
        traceback.print_exc()
        # An import failure is still a Python traceback in user code (think
        # syntax error in `hello.py`); surface it the same way as a runtime
        # raise so the cell lifts it onto JobResult.traceback.
        _print_traceback_marker(traceback.format_exc())
        return EXIT_USER_IMPORT_FAILED

    fn = registered_main()
    if fn is None:
        print(
            f"armnet-runtime: no @main-decorated function found in {args.script}. "
            "Decorate exactly one function with `@main` to mark it as the entry point.",
            file=sys.stderr,
        )
        return EXIT_NO_MAIN

    try:
        ctx = build_context()
    except Exception as exc:  # noqa: BLE001
        print(f"armnet-runtime: failed to build context: {exc}", file=sys.stderr)
        return EXIT_BAD_CONTEXT

    try:
        result = fn(ctx)
    except SystemExit:
        # Customer code called sys.exit explicitly \u2014 let that propagate.
        raise
    except Exception:
        # Stderr: full, human-readable for `docker logs` users.
        print(
            f"armnet-runtime: @main function raised; "
            f"job {ctx.job_id} will be marked FAILED.",
            file=sys.stderr,
        )
        traceback.print_exc()
        # Stdout: structured marker the cell parses into JobResult.traceback.
        _print_traceback_marker(traceback.format_exc())
        return EXIT_USER_RAISED

    if result is not None:
        _print_result(result)
    return EXIT_OK

armnet_runtime.env

Env var keys the cell injects into customer containers.

Single source of truth shared by the cell (which sets them) and the runtime SDK (which reads them).

JOB_ID module-attribute

JOB_ID = 'ARMNET_JOB_ID'

EMBODIMENT module-attribute

EMBODIMENT = 'ARMNET_EMBODIMENT'

TASK module-attribute

TASK = 'ARMNET_TASK'

TIMEOUT_SECONDS module-attribute

TIMEOUT_SECONDS = 'ARMNET_TIMEOUT_SECONDS'

ARGS module-attribute

ARGS = 'ARMNET_ARGS'

CELL_CONFIG module-attribute

CELL_CONFIG = 'ARMNET_CELL_CONFIG'

CELL_SOCKET module-attribute

CELL_SOCKET = 'ARMNET_CELL_SOCKET'

OPERATOR_CALL_ENDPOINT module-attribute

OPERATOR_CALL_ENDPOINT = 'ARMNET_OPERATOR_CALL_ENDPOINT'

LOCAL_CONTROL_ENDPOINT module-attribute

LOCAL_CONTROL_ENDPOINT = 'ARMNET_LOCAL_CONTROL_ENDPOINT'

LOCAL_CONTAINER module-attribute

LOCAL_CONTAINER = 'ARMNET_LOCAL_CONTAINER'

ROBOT_ID module-attribute

ROBOT_ID = 'ARMNET_ROBOT_ID'

CALIBRATION_DIR module-attribute

CALIBRATION_DIR = 'ARMNET_CALIBRATION_DIR'

CALIBRATION_FILE_PATH module-attribute

CALIBRATION_FILE_PATH = 'ARMNET_CALIBRATION_FILE_PATH'

CACHE_HOME module-attribute

CACHE_HOME = 'REMOTEROBOT_CACHE_HOME'

VOLUME_HOME module-attribute

VOLUME_HOME = 'REMOTEROBOT_VOLUME_HOME'

HF_HOME module-attribute

HF_HOME = 'HF_HOME'

SECRETS module-attribute

SECRETS = 'ARMNET_SECRETS'

armnet_runtime.lerobot

LeRobot integration helpers for armnet-runtime.

RemoteARX5Arm

Proxy for arx5_common.ARX5Arm.

The user container constructs this class through import replacement, while the cell-side connector owns the real arx5_interface backed arm.

Source code in runtime/src/armnet_runtime/lerobot/remote_arx5.py

class RemoteARX5Arm:
    """Proxy for `arx5_common.ARX5Arm`.

    The user container constructs this class through import replacement, while
    the cell-side connector owns the real `arx5_interface` backed arm.
    """

    def __init__(self, *args: Any, **kwargs: Any) -> None:
        object.__setattr__(self, "_socket_path", _connector_endpoint())
        object.__setattr__(self, "_object_id", None)
        object.__setattr__(self, "_method_cache", set())
        object.__setattr__(self, "_missing_cache", set())
        object.__setattr__(self, "_connected", None)
        object.__setattr__(self, "_connected_at", 0.0)
        object.__setattr__(self, "_conn", PersistentConnection(self._socket_path))
        logger.info("RemoteARX5Arm connecting to socket %r", self._socket_path)
        object.__setattr__(self, "_object_id", self._request({
            "op": "create",
            "class": "ARX5Arm",
            "args": jsonable(args),
            "kwargs": jsonable(kwargs),
        })["object_id"])

    def _set_connected(self, value: bool) -> None:
        object.__setattr__(self, "_connected", value)
        object.__setattr__(self, "_connected_at", time.monotonic())

    def __del__(self) -> None:
        conn = getattr(self, "_conn", None)
        if conn is not None:
            conn.close()

    def __getattr__(self, name: str) -> Any:
        # Serve the heavily-polled is_connected property from a short-lived
        # local cache; the TTL bounds staleness so an edge-driven disconnect is
        # re-checked within IS_CONNECTED_TTL_S (state set via connect()/
        # disconnect(), observed below).
        if (
            name == "is_connected"
            and self._connected is not None
            and (time.monotonic() - self._connected_at) < IS_CONNECTED_TTL_S
        ):
            return self._connected
        # Method-ness is immutable, so cache resolved method names to skip the
        # getattr round-trip that would otherwise precede every method call.
        if name in self._method_cache:
            return self._method_proxy(name)
        # A name the edge arm genuinely lacks stays absent for the object's
        # lifetime, so remember it and raise AttributeError locally so that
        # ``getattr(obj, name, default)`` falls back to its default rather than
        # treating the miss as a (truthy) method proxy.
        if name in self._missing_cache:
            raise AttributeError(name)
        try:
            response = self._request({"op": "getattr", "object_id": self._object_id, "name": name})
            if response.get("callable"):
                self._method_cache.add(name)
                return self._method_proxy(name)
            value = decode(response["value"])
            if name == "is_connected":
                self._set_connected(bool(value))
            return value
        except RemoteAttributeError:
            # RemoteAttributeError subclasses AttributeError; negative-cache and
            # propagate so getattr-with-default works. Real methods take the
            # ``callable`` branch above, so this only fires for genuine misses.
            self._missing_cache.add(name)
            raise

    def __setattr__(self, name: str, value: Any) -> None:
        if name.startswith("_"):
            object.__setattr__(self, name, value)
            return
        self._request({
            "op": "setattr",
            "object_id": self._object_id,
            "name": name,
            "value": jsonable(value),
        })

    def _method_proxy(self, name: str):
        def method(*args: Any, **kwargs: Any) -> Any:
            response = self._request({
                "op": "call",
                "object_id": self._object_id,
                "method": name,
                "args": jsonable(args),
                "kwargs": jsonable(kwargs),
            })
            # Track connection state so is_connected can be served locally.
            if name == "connect":
                self._set_connected(True)
            elif name == "disconnect":
                self._set_connected(False)
            if response.get("context_manager"):
                return RemoteContextManager(
                    conn=self._conn,
                    object_id=response["object_id"],
                )
            return decode(response.get("value"))

        return method

    def _request(self, payload: dict[str, Any]) -> dict[str, Any]:
        logger.debug("ARX5 request op=%r path=%r", payload.get("op"), self._socket_path)
        return check_response(self._conn.request(payload), error_message="remote ARX5 call failed")

RemoteOpenCVCamera

Drop-in-ish proxy for LeRobot's OpenCVCamera.

The real camera stays on the cell host. The customer container holds this proxy and forwards camera lifecycle/read calls through the connector.

Source code in runtime/src/armnet_runtime/lerobot/remote_camera.py

class RemoteOpenCVCamera:
    """Drop-in-ish proxy for LeRobot's OpenCVCamera.

    The real camera stays on the cell host. The customer container holds this
    proxy and forwards camera lifecycle/read calls through the connector.
    """

    def __init__(self, *args: Any, **kwargs: Any) -> None:
        object.__setattr__(self, "_socket_path", _connector_endpoint())
        object.__setattr__(self, "_object_id", None)
        object.__setattr__(self, "_method_cache", set())
        object.__setattr__(self, "_missing_cache", set())
        object.__setattr__(self, "_connected", None)
        object.__setattr__(self, "_connected_at", 0.0)
        object.__setattr__(self, "_conn", PersistentConnection(self._socket_path))
        logger.info("RemoteOpenCVCamera connecting to socket %r", self._socket_path)
        object.__setattr__(self, "_object_id", self._request({
            "op": "create",
            "class": "OpenCVCamera",
            "args": jsonable(args),
            "kwargs": jsonable(kwargs),
        })["object_id"])

    def _set_connected(self, value: bool) -> None:
        object.__setattr__(self, "_connected", value)
        object.__setattr__(self, "_connected_at", time.monotonic())

    def __del__(self) -> None:
        conn = getattr(self, "_conn", None)
        if conn is not None:
            conn.close()

    def __getattr__(self, name: str) -> Any:
        # Serve the heavily-polled is_connected property from a short-lived
        # local cache; the TTL bounds staleness so an edge-driven disconnect is
        # re-checked within IS_CONNECTED_TTL_S (state set via connect()/
        # disconnect(), observed below).
        if (
            name == "is_connected"
            and self._connected is not None
            and (time.monotonic() - self._connected_at) < IS_CONNECTED_TTL_S
        ):
            return self._connected
        # Method-ness is immutable, so cache resolved method names to skip the
        # getattr round-trip that would otherwise precede every method call.
        if name in self._method_cache:
            return self._method_proxy(name)
        # A name the edge camera genuinely lacks stays absent for the object's
        # lifetime, so remember it and raise AttributeError locally. This lets
        # ``getattr(cam, name, default)`` fall back to its default — lerobot
        # 0.6.0's ``get_observation()`` probes ``getattr(cam, "use_depth", False)``
        # every frame; without this the miss would look like a (truthy) method
        # and trigger a bogus ``read_latest_depth()`` call.
        if name in self._missing_cache:
            raise AttributeError(name)
        try:
            response = self._request({"op": "getattr", "object_id": self._object_id, "name": name})
            if response.get("callable"):
                self._method_cache.add(name)
                return self._method_proxy(name)
            value = decode(response["value"])
            if name == "is_connected":
                self._set_connected(bool(value))
            return value
        except RemoteAttributeError:
            # RemoteAttributeError subclasses AttributeError; negative-cache and
            # propagate so getattr-with-default works. Real methods take the
            # ``callable`` branch above, so this only fires for genuine misses.
            self._missing_cache.add(name)
            raise

    def __setattr__(self, name: str, value: Any) -> None:
        if name.startswith("_"):
            object.__setattr__(self, name, value)
            return
        self._request({
            "op": "setattr",
            "object_id": self._object_id,
            "name": name,
            "value": jsonable(value),
        })

    def _method_proxy(self, name: str):
        def method(*args: Any, **kwargs: Any) -> Any:
            response = self._request({
                "op": "call",
                "object_id": self._object_id,
                "method": name,
                "args": jsonable(args),
                "kwargs": jsonable(kwargs),
            })
            # Track connection state so is_connected can be served locally.
            if name == "connect":
                self._set_connected(True)
            elif name == "disconnect":
                self._set_connected(False)
            if response.get("context_manager"):
                return RemoteContextManager(
                    conn=self._conn,
                    object_id=response["object_id"],
                )
            return decode(response.get("value"))

        return method

    def _request(self, payload: dict[str, Any]) -> dict[str, Any]:
        logger.debug("camera request op=%r path=%r", payload.get("op"), self._socket_path)
        return check_response(self._conn.request(payload), error_message="remote camera call failed")

RemoteFeetechMotorsBus

Drop-in-ish proxy for LeRobot's FeetechMotorsBus.

The connector creates the real bus object on the cell host; this class forwards method calls and simple attribute gets/sets over JSON-lines. It is intentionally generic for M2 so we don't need to perfectly mirror LeRobot's evolving bus API upfront.

Source code in runtime/src/armnet_runtime/lerobot/remote_feetech.py

class RemoteFeetechMotorsBus:
    """Drop-in-ish proxy for LeRobot's FeetechMotorsBus.

    The connector creates the real bus object on the cell host; this class
    forwards method calls and simple attribute gets/sets over JSON-lines.
    It is intentionally generic for M2 so we don't need to perfectly mirror
    LeRobot's evolving bus API upfront.
    """

    def __init__(self, *args: Any, **kwargs: Any) -> None:
        object.__setattr__(self, "_socket_path", _socket_path_from_args(args, kwargs))
        object.__setattr__(self, "_object_id", None)
        object.__setattr__(self, "_method_cache", set())
        object.__setattr__(self, "_missing_cache", set())
        object.__setattr__(self, "_connected", None)
        object.__setattr__(self, "_connected_at", 0.0)
        object.__setattr__(self, "_conn", PersistentConnection(self._socket_path))
        logger.info("RemoteFeetechMotorsBus connecting to socket %r", self._socket_path)
        object.__setattr__(self, "_object_id", self._request({
            "op": "create",
            "class": "FeetechMotorsBus",
            "args": jsonable(args),
            "kwargs": jsonable(kwargs),
        })["object_id"])

    def _set_connected(self, value: bool) -> None:
        object.__setattr__(self, "_connected", value)
        object.__setattr__(self, "_connected_at", time.monotonic())

    def __del__(self) -> None:
        conn = getattr(self, "_conn", None)
        if conn is not None:
            conn.close()

    def __getattr__(self, name: str) -> Any:
        # Serve the heavily-polled is_connected property from a short-lived
        # local cache. State changes via connect()/disconnect() (observed in
        # _method_proxy); the TTL bounds staleness so an edge-driven disconnect
        # is re-checked within IS_CONNECTED_TTL_S rather than lying forever.
        if (
            name == "is_connected"
            and self._connected is not None
            and (time.monotonic() - self._connected_at) < IS_CONNECTED_TTL_S
        ):
            return self._connected
        # Method-ness is immutable for an object's lifetime, so cache resolved
        # method names to skip the getattr round-trip that would otherwise
        # precede every single method call.
        if name in self._method_cache:
            return self._method_proxy(name)
        # A name the edge object genuinely lacks stays absent for the object's
        # lifetime, so remember it and raise AttributeError locally. This is what
        # lets ``getattr(obj, name, default)`` fall back to its default — e.g.
        # lerobot 0.6.0 probes ``getattr(cam, "use_depth", False)`` on every
        # ``get_observation()``; without this the miss would masquerade as a
        # (truthy) method and trigger a bogus ``read_latest_depth()`` call.
        if name in self._missing_cache:
            raise AttributeError(name)
        # Otherwise try simple attribute access. Methods surface via the edge's
        # ``callable`` flag below; a genuine miss propagates as AttributeError.
        try:
            response = self._request({"op": "getattr", "object_id": self._object_id, "name": name})
            if response.get("callable"):
                self._method_cache.add(name)
                return self._method_proxy(name)
            value = decode(response["value"])
            if name == "is_connected":
                self._set_connected(bool(value))
            return value
        except RemoteAttributeError:
            # RemoteAttributeError subclasses AttributeError; negative-cache the
            # name and let it propagate so getattr-with-default works and Python
            # attribute semantics hold. (Real methods never reach here — they
            # take the ``callable`` branch above.)
            self._missing_cache.add(name)
            raise

    def __setattr__(self, name: str, value: Any) -> None:
        if name.startswith("_"):
            object.__setattr__(self, name, value)
            return
        self._request({
            "op": "setattr",
            "object_id": self._object_id,
            "name": name,
            "value": jsonable(value),
        })

    def _method_proxy(self, name: str):
        def method(*args: Any, **kwargs: Any) -> Any:
            response = self._request({
                "op": "call",
                "object_id": self._object_id,
                "method": name,
                "args": jsonable(args),
                "kwargs": jsonable(kwargs),
            })
            # Track connection state so is_connected can be served locally.
            if name == "connect":
                self._set_connected(True)
            elif name == "disconnect":
                self._set_connected(False)
            if response.get("context_manager"):
                return RemoteContextManager(
                    conn=self._conn,
                    object_id=response["object_id"],
                )
            return decode(response.get("value"))

        return method

    def _request(self, payload: dict[str, Any]) -> dict[str, Any]:
        logger.debug("Feetech request op=%r path=%r", payload.get("op"), self._socket_path)
        return check_response(self._conn.request(payload), error_message="remote call failed")

install_lerobot_remote_mode

install_lerobot_remote_mode() -> None

Replace known LeRobot Feetech bus classes with remote proxies.

This is best-effort and guarded. If LeRobot is not installed, nothing happens. If LeRobot is installed but the Feetech module moved, we log and continue; the user's script will then fail with its normal ImportError, which is useful signal while we refine supported versions.

Source code in runtime/src/armnet_runtime/lerobot/import_swap.py

def install_lerobot_remote_mode() -> None:
    """Replace known LeRobot Feetech bus classes with remote proxies.

    This is best-effort and guarded. If LeRobot is not installed, nothing
    happens. If LeRobot is installed but the Feetech module moved, we log and
    continue; the user's script will then fail with its normal ImportError,
    which is useful signal while we refine supported versions.
    """

    if os.environ.get(env_keys.LOCAL_CONTAINER):
        logger.info("skipping armnet LeRobot remote mode for local-container direct hardware run")
        return

    try:
        importlib.import_module("lerobot")
    except ImportError:
        return

    patched: list[str] = []
    for module_name in _FEETECH_MODULE_CANDIDATES:
        try:
            module = importlib.import_module(module_name)
        except ImportError:
            continue
        for attr in ("FeetechMotorsBus",):
            if hasattr(module, attr):
                setattr(module, attr, RemoteFeetechMotorsBus)
                patched.append(f"{module_name}.{attr}")

    for module_name in _OPENCV_CAMERA_MODULE_CANDIDATES:
        try:
            module = importlib.import_module(module_name)
        except ImportError:
            continue
        for attr in ("OpenCVCamera",):
            if hasattr(module, attr):
                setattr(module, attr, RemoteOpenCVCamera)
                patched.append(f"{module_name}.{attr}")

    for module_name in _ARX5_MODULE_CANDIDATES:
        try:
            module = importlib.import_module(module_name)
        except ImportError:
            continue
        if hasattr(module, "ARX5Arm"):
            setattr(module, "ARX5Arm", RemoteARX5Arm)
            patched.append(f"{module_name}.ARX5Arm")

    if patched:
        logger.info("installed armnet LeRobot remote mode: patched %s", patched)
    else:
        logger.warning(
            "LeRobot is installed, but no known Feetech bus, OpenCV camera, or ARX5 arm class was found to patch; "
            "remote robot access may fail. Checked: %s",
            _FEETECH_MODULE_CANDIDATES + _OPENCV_CAMERA_MODULE_CANDIDATES + _ARX5_MODULE_CANDIDATES,
        )