Runtime Context¶

Context is the most important object in armnet-runtime. Every @main-decorated runtime function receives one:

from armnet_runtime import Context, main


@main
def run(ctx: Context) -> dict:
    ctx.report_progress("starting")
    return {"job_id": ctx.job_id}

Context Fields¶

Field	Type	Description
`job_id`	`str`	Platform job ID for the current execution.
`embodiment`	`str`	Robot embodiment requested by the job, e.g. `lerobot/so-101` or `lerobot/arx5`.
`task`	`str`	Task routed to the cell, e.g. `assemble_block_tower`.
`args`	`dict[str, Any]`	JSON arguments supplied by `execute(..., args={...})`.
`cell`	`Cell`	Handle for robot-cell services such as reset, completion, robot port, calibration, and language instruction.
`camera_configs`	`dict[str, Any]`	LeRobot camera configs injected by the cell from `robot_cell.json`.
`cache_home`	`Path \| None`	Per-user cache directory. `HF_HOME` points inside this directory. Cache data may be reused but is not a durable API.
`volume`	`Volume`	Per-user persistent volume mount. Use this for checkpoints, artifacts, and files that should survive across runs.
`secrets`	`dict[str, str]`	Secret values injected into the job, keyed by environment variable name. Values are also available as process env vars.
`timeout_seconds`	`int \| None`	Wall-clock timeout requested for the container.

Methods¶

`ctx.report_progress(message)`¶

Emits a progress line to stdout with a armnet marker. The cell streams this line back to the client alongside normal container logs.

ctx.report_progress("connected to robot")

`ctx.log_rerun_data(observation=None, action=None, compress_images=True)`¶

Streams observation/action data to a Rerun viewer running on the client. 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.

The cell container has no viewer, so this serializes a protobuf packet, 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 (compress_images=False sends raw RGB).

obs = robot.get_observation()
action = policy.select_action(obs)
robot.send_action(action)
ctx.log_rerun_data(observation=obs, action=action)

On the client side, start a viewer and opt into streaming:

from armnet_client import execute, init_rerun

init_rerun("my-eval", spawn=True)           # requires the `viz` extra
execute(image=img, embodiment=..., task=..., args={"use_rerun": True}, use_rerun=True)

Cell¶

ctx.cell describes the physical robot cell the job is running on.

Field / Method	Type	Description
`robot_port`	`str \| None`	Value to pass into LeRobot robot configs. In remote containers this is a connector endpoint, not the host serial/device path.
`robot_id`	`str \| None`	Stable robot ID used by LeRobot calibration lookup.
`calibration_dir`	`Path \| None`	Calibration directory visible inside the container.
`calibration_file_path`	`Path \| None`	Exact calibration file path visible inside the container.
`arms`	`dict[str, RuntimeArm]`	Named arms for a multi-arm cell, keyed by names such as `left` and `right`. Empty for single-arm cells.
`is_bimanual`	`bool`	True when the cell exposes both `left` and `right` arms.
`arm(name)`	method	Return one named `RuntimeArm`, including its connector endpoint and calibration path.
`prepare_bimanual_calibration_dir()`	method	Copy per-arm calibration files into LeRobot's expected `<robot_id>_left.json` / `<robot_id>_right.json` layout and return a `BimanualCalibrationLayout` containing `robot_id` and `calibration_dir`.
`language_instruction`	`str \| None`	Natural-language instruction for the cell's assigned task (set from the FMS/database, forwarded to the cell when it starts).
`reset(message=None)`	method	Blocks until the operator confirms the workspace has been reset.
`is_complete(block=False)`	method	Returns a `CompletionStatus(complete, success)`. A human operator's success/fail verdict (reported from the FMS during a rollout) wins and ends the episode immediately; otherwise the automated completion monitor is consulted (where `success == complete`). `bool(status)` is `status.complete`, so `if ctx.cell.is_complete():` still works.
`rollout_begin(index=None, total=None, outcome_controls=True)` / `rollout_end()`	method	Announce the start/end of a rollout/episode in the job's loop. Pass `index` (1-based) and `total` so the FMS shows loop progress ("rollout N / M") for evals and data collection. With `outcome_controls=True` (default, for policy evals) the FMS also shows operator Success/Fail buttons; hitting one ends the rollout with that verdict (and triggers a workspace reset) without stopping the job. Pass `outcome_controls=False` for progress-only loops such as teleop data collection. Best-effort signalling.

Example:

cfg = SO101FollowerConfig(
    port=ctx.cell.robot_port,
    id=ctx.cell.robot_id,
    calibration_dir=ctx.cell.calibration_dir,
    cameras=ctx.camera_configs,
)

robot = SO101Follower(cfg)
robot.connect(calibrate=False)
ctx.cell.reset("Reset the workspace, then press Enter.")

For bimanual SO-101 cells, use the named arms and LeRobot's BiSOFollower:

from lerobot.robots.bi_so_follower import BiSOFollower, BiSOFollowerConfig
from lerobot.robots.so_follower import SO101FollowerConfig

layout = ctx.cell.prepare_bimanual_calibration_dir()
robot = BiSOFollower(
    BiSOFollowerConfig(
        id=layout.robot_id,
        calibration_dir=layout.calibration_dir,
        left_arm_config=SO101FollowerConfig(port=ctx.cell.arm("left").robot_port),
        right_arm_config=SO101FollowerConfig(port=ctx.cell.arm("right").robot_port),
    )
)

Volume¶

ctx.volume gives runtime code access to the user volume mounted into the container.

checkpoint_dir = ctx.volume.path("openpi/checkpoints/my-checkpoint")
config_text = ctx.volume.read_text("configs/eval.json")
ctx.volume.write_text("outputs/status.txt", "done")

Volume paths must be relative and cannot contain ...

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,
        )

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)

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,
    )

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)

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:

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.
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):

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.
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")

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

Runtime Context¶

Context Fields¶

Methods¶

ctx.report_progress(message)¶

ctx.log_rerun_data(observation=None, action=None, compress_images=True)¶

Cell¶

Volume¶

Context dataclass ¶

report_progress ¶

log_rerun_data ¶

Cell dataclass ¶

arm ¶

prepare_bimanual_calibration_dir ¶

reset ¶

is_complete ¶

rollout_begin ¶

rollout_end ¶

Volume dataclass ¶

root class-attribute instance-attribute ¶

path ¶

read_bytes ¶

read_text ¶

write_bytes ¶

write_text ¶

`ctx.report_progress(message)`¶

`ctx.log_rerun_data(observation=None, action=None, compress_images=True)`¶

Context `dataclass` ¶

Cell `dataclass` ¶

Volume `dataclass` ¶

root `class-attribute` `instance-attribute` ¶