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. `push_green_button`. See Environments and Tasks.
`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 for the cell's cameras, injected by the platform. Pass straight into your robot config.
`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 job's task (resolved by the orchestrator from the tasks table and sent with the job).
`reset(confirm=None)`	method	Restore the workspace to the state the task starts from. See Resetting between episodes.
`readings`	`Mapping[str, Reading]`	Live instrumentation channels on an automated cell, e.g. each BusyBox control's current position. Empty on an uninstrumented cell.
`instrument(robot)`	method	Hand the connected robot to the cell's environment so it can score and reset episodes itself. Returns the robot.
`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.instrument(robot)   # let an automated environment score and reset
ctx.cell.reset()

Install the environment package in your image

An environment scores and resets through the robot your container is driving, so its package has to be installed there. For a BusyBox cell that means:

RUN pip install armnet-busybox

You never import it — instrument() finds it through the armnet.environments entry point. Without it the job still runs, but every episode falls back to an operator judging and resetting it by hand, which on a 20-rollout eval is 20 trips to the FMS. ctx.cell.readings staying empty on an instrumented cell is the symptom.

Resetting between episodes¶

ctx.cell.reset() asks the cell's environment what restoring the scene takes, so the same call does the right thing everywhere. On a BusyBox cell a momentary button needs nothing beyond returning the arm to rest, a switch left flipped is put back by a motion plan, and a control with no taught plan waits for an operator. On a manual cell it always waits for an operator.

Instrumented motion-plan resets are verified against the panel state. A safe trajectory that misses is retried up to five total attempts, with a small alternating rail re-seat between attempts when the plan was taught at a fixed rail pose and a fresh bounded per-joint pose offset on each retry. Hardware/control failures still stop immediately, and five misses fall back to an operator.

Override that when your own job knows better:

ctx.cell.reset(confirm=True)    # insist on a human check
ctx.cell.reset(confirm=False)   # never block on a human

Call ctx.cell.instrument(robot) once after connecting if you want automated scoring and resets; without it the cell falls back to operator confirmation.

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)
    # Leaderboard recording state (see init_leaderboard). None until enabled.
    _leaderboard: 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_core.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 get_robot_telemetry(self, *, arm: str | None = None) -> dict[str, Any]:
        """Return the latest cached edge telemetry snapshot for an arm."""

        return self.cell.get_robot_telemetry(arm=arm)

    def init_leaderboard(
        self,
        policy_repo_id: str,
        *,
        revision: Optional[str] = None,
        model_type: Optional[str] = None,
        training_framework: str = "unknown",
        user: Optional[str] = None,
        source: str = "script",
        repo: Optional[str] = None,
        token: Optional[str] = None,
    ) -> None:
        """Start recording this job's rollouts to the shared Armnet leaderboard.

        Call once before your rollout loop, naming the policy you are
        evaluating. From then on every :meth:`Cell.rollout_end` records that
        rollout's *cell-scored* outcome (from the cell's automated completion
        monitor or an operator's verdict) — success counts are never
        self-reported by user code. Call :meth:`submit_results_to_leaderboard`
        once the loop finishes to publish the pooled result.

        Identity metadata (``policy_repo_id``, ``revision``, ``model_type``) is
        yours to declare; only the success counts are enforced from the cell.
        ``embodiment``, ``task`` and the cell id are taken from this context.

        Writing needs a HuggingFace token with write access to the leaderboard
        dataset. This reuses the container's ambient HF token (the one used to
        resolve the recorded dataset owner); no bespoke credential is
        provisioned. See ``armnet_runtime.leaderboard`` for the schema and the
        note on future server-side submission.
        """
        from armnet_runtime import leaderboard as _lb

        resolved_revision = revision or _lb.resolve_revision(policy_repo_id, token)
        resolved_model_type = model_type or _lb.resolve_model_type(
            policy_repo_id, resolved_revision, token
        )
        self._leaderboard = {
            "policy_repo_id": policy_repo_id,
            "revision": resolved_revision,
            "model_type": resolved_model_type,
            "training_framework": training_framework or "unknown",
            "user": user,
            "source": source,
            "repo": repo or _lb.LEADERBOARD_REPO,
            "token": token,
            "outcomes": [],
        }
        self.cell._rollout_outcome_sink = self._record_leaderboard_outcome
        logger.info(
            "leaderboard recording enabled for %s@%s (%s)",
            policy_repo_id,
            (resolved_revision or "unpinned")[:8],
            resolved_model_type,
        )

    def _record_leaderboard_outcome(self, status: "CompletionStatus") -> None:
        """Sink installed on the cell: append one rollout's cell-scored success."""
        if self._leaderboard is not None:
            self._leaderboard["outcomes"].append(bool(status.success))

    def submit_results_to_leaderboard(self) -> Optional[dict[str, Any]]:
        """Publish the recorded rollout results to the leaderboard (best-effort).

        Aggregates the per-rollout outcomes recorded since
        :meth:`init_leaderboard` into one pooled run and appends it to the
        dataset. Returns a summary dict, or ``None`` if recording wasn't
        enabled, no rollouts were recorded, or the upload failed. Never raises
        into the job — a leaderboard hiccup must not fail an otherwise good eval.
        """
        state = self._leaderboard
        if not state:
            logger.warning(
                "submit_results_to_leaderboard called without init_leaderboard; skipping"
            )
            return None
        outcomes = state["outcomes"]
        if not outcomes:
            logger.warning("no rollouts recorded for the leaderboard; skipping submission")
            return None
        n_rollouts = len(outcomes)
        n_success = sum(1 for s in outcomes if s)

        from armnet_runtime import leaderboard as _lb

        user = state["user"]
        if not user:
            try:
                from huggingface_hub import whoami

                user = whoami(token=state["token"]).get("name")
            except Exception:  # noqa: BLE001 - user attribution is best-effort
                user = None
        try:
            _lb.record_run(
                repo_id=state["policy_repo_id"],
                revision=state["revision"] or "unknown",
                n_rollouts=n_rollouts,
                n_success=n_success,
                model_type=state["model_type"],
                training_framework=state.get("training_framework", "unknown"),
                source=state["source"],
                cell_id=self.cell.cell_id,
                embodiment=self.embodiment,
                task=self.task,
                user=user,
                repo=state["repo"],
                token=state["token"],
            )
        except Exception:  # noqa: BLE001 - persistence must not fail the eval
            logger.warning("failed to submit results to the leaderboard", exc_info=True)
            return None
        self.report_progress(
            f"leaderboard: recorded {n_success}/{n_rollouts} for {state['policy_repo_id']}"
        )
        return {
            "repo_id": state["policy_repo_id"],
            "revision": state["revision"],
            "model_type": state["model_type"],
            "n_rollouts": n_rollouts,
            "n_success": n_success,
            "source": state["source"],
        }

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

    cell_id: Optional[str] = None
    """Stable cell identifier (e.g. ``cell-08``) from the cell config, used to
    scope leaderboard entries to the physical cell that produced them."""

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

    environment: Optional[str] = None
    """The kind of workcell this cell is set up as, e.g. ``"busybox"``."""

    task: Optional[Task] = None
    """Which of the environment's tasks this job is running.

    A cell is set up for one environment but runs any task within it, so this is
    what tells the environment which goal to watch for and how to reset.
    """

    environment_config: dict[str, Any] = field(default_factory=dict)
    """The environment's own settings, passed through undecoded.

    Only the package implementing the environment understands these. Keeping
    them opaque is what lets armnet-runtime, which is baked into every customer
    image, stay free of any environment's dependencies.
    """

    rails: tuple[RuntimeRail, ...] = ()
    """Measured rail calibrations from the cell JSON (empty if this cell has none)."""

    lightbox: Optional[RuntimeLightbox] = None
    """The cell's LED lightbox, or None on a cell that has none."""

    camera_mounts: dict[str, RuntimeCameraMount] = field(default_factory=dict)
    """Pan/tilt mounts keyed by camera name.

    Only cameras the cell config gives a mount block appear here, so a wrist
    camera riding on the arm is absent rather than present and immovable.
    """

    scene_calibration: Optional[SceneCalibration] = None
    """Manipulation-area snapshot selected when this cell process started."""

    device_config: Optional[CellDeviceConfig] = None
    """What this cell may vary and within what bounds.

    Derived from the same cell config the orchestrator publishes on the
    heartbeat, so a scenario sampled here and an override the client checked
    before submitting are held to the same envelope.
    """

    disable_automated_reset: bool = False
    """Put an operator back in the loop for every :meth:`reset`.

    Set from the job's ``disable_automated_reset`` argument. An instrumented cell
    normally restores the workspace itself where it can; this forces the manual
    confirmation back on for jobs where a human should be checking the scene.
    """

    # 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 per-endpoint connections for high-rate cached telemetry polling.
    _telemetry_conns: dict[str, Any] = field(
        default_factory=dict,
        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)
    # Most-recent completion outcome seen during the current rollout, plus an
    # optional sink (installed by Context.init_leaderboard) that records each
    # rollout's cell-scored outcome. Kept here so rollout_end can record the
    # authoritative outcome without user code ever passing a success value.
    _last_completion: Any = field(default=None, init=False, repr=False, compare=False)
    _rollout_outcome_sink: Any = field(default=None, init=False, repr=False, compare=False)
    # The live instrumentation session, opened by instrument(). None on a cell
    # whose environment has no instrumentation, which every method below treats
    # as "nothing known" rather than as an error.
    _instrumented: Any = field(default=None, init=False, repr=False, compare=False)

    def __post_init__(self) -> None:
        # Device handles act through this cell's connector, so they are given a
        # way back to it here rather than each carrying its own endpoint.
        for device in (*self.rails, self.lightbox, *self.camera_mounts.values()):
            if device is not None:
                device._bind(self)

    @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 get_robot_telemetry(self, *, arm: str | None = None) -> dict[str, Any]:
        """Return the edge's latest cached telemetry snapshot without bus I/O.

        Edge timestamps and sequence/config-generation values are returned
        unchanged. Missing caches, transport failures, and old edges that do
        not support this operation are represented as data so telemetry polling
        cannot fail the caller's control loop.
        """

        endpoint = self.robot_port
        if arm is not None and self.arms:
            runtime_arm = self.arms.get(arm)
            if runtime_arm is None:
                return _telemetry_unavailable(arm=arm, error=f"cell has no arm named {arm!r}")
            endpoint = runtime_arm.robot_port
        if not endpoint or not _looks_like_connector_endpoint(endpoint):
            return _telemetry_unavailable(arm=arm, error="robot connector is unavailable")

        request: dict[str, Any] = {"op": "get_robot_telemetry"}
        if arm is not None:
            request["arm"] = arm
        connection = self._telemetry_conns.get(endpoint)
        if connection is None:
            connection = _TeleopConnection(
                endpoint,
                timeout=_ROBOT_TELEMETRY_READ_TIMEOUT_S,
            )
            self._telemetry_conns[endpoint] = connection
        try:
            response = connection.request(request)
        except Exception as exc:  # noqa: BLE001 - telemetry is always best-effort
            return _telemetry_unavailable(arm=arm, error=str(exc))

        telemetry = response.get("telemetry")
        if response.get("ok") and isinstance(telemetry, dict):
            return telemetry
        error = str(response.get("error", "robot telemetry is unavailable"))
        if _is_unsupported_telemetry_response(response):
            return {
                "arm": arm,
                "telemetry_valid": False,
                "telemetry_error_code": 1,
                "unsupported": True,
                "error": error,
            }
        return _telemetry_unavailable(arm=arm, error=error)

    def prepare_calibration_dir(self) -> tuple[Optional[str], Optional[Path]]:
        """Resolve the single-arm calibration into a ``(robot_id, dir)`` LeRobot takes.

        A cell whose calibration is managed from the FMS is given the file
        itself rather than a directory, and LeRobot only accepts a directory it
        can find ``<robot_id>.json`` in, so the file is staged into one.
        """

        if self.calibration_file_path:
            path = self.calibration_file_path
            robot_id = self.robot_id or path.stem.removesuffix("_calib")
            calibration_dir = Path(tempfile.mkdtemp(prefix="armnet-calibration-"))
            shutil.copy2(path, calibration_dir / f"{robot_id}.json")
            return robot_id, calibration_dir
        return self.robot_id, self.calibration_dir

    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 instrument(self, robot: Any) -> Any:
        """Connect this cell's environment and return the robot to drive.

        Call once, right after building the robot, and use what comes back.
        Environments that record their own readings alongside yours hand back a
        wrapper; ones that don't hand back the robot untouched. Either way the
        observation a policy sees is unchanged.

        This is also what gives :meth:`reset`, :meth:`is_complete` and
        :meth:`readings` something to work with, so a job that skips it still
        runs — it just falls back to the operator for everything.
        """

        if self._instrumented is not None:
            return self._instrumented.instrument(robot)
        if not self.environment:
            return robot

        config = dict(self.environment_config)
        config.setdefault("cell_id", self.cell_id)
        config.setdefault("language_instruction", self.language_instruction)
        try:
            environment = environment_for(self.environment)
        except EnvironmentNotFound:
            # An image that does not ship this environment's package is a
            # normal thing to run: nobody writing their own job should have to
            # install ours to use a cell that happens to be instrumented. Say
            # so once and fall back to the operator, rather than failing a job
            # over a workspace reading it never asked for.
            self._report_progress(
                f"this image has no {self.environment!r} environment installed; "
                "the workspace will be reset and scored by the operator"
            )
            return robot
        session = environment.connect(
            config,
            task=self.task,
            report_progress=self._report_progress,
        )
        if session is None:
            return robot
        self._instrumented = session
        return session.instrument(robot)

    @property
    def instrumentation(self) -> Optional[InstrumentedCell]:
        """The live instrumentation session, or None if the cell has none."""

        return self._instrumented

    def readings(self) -> Mapping[str, Reading]:
        """Current value of every instrumented channel in the workspace.

        Empty when the cell has no instrumentation or none has been heard from
        yet. Values are advisory: nothing here should fail a rollout.
        """

        if self._instrumented is None:
            return {}
        return self._instrumented.readings()

    def attach_dataset(self, dataset_root: Any) -> None:
        """Record instrumentation readings beside a dataset being written.

        Readings are written as a sidecar, not folded into ``observation.state``,
        so a policy trained without the instrumentation sees the same features
        with it attached.
        """

        if self._instrumented is not None:
            self._instrumented.attach_dataset(dataset_root)

    def record_frame(self) -> None:
        """Record the instrumentation's view of the frame just captured."""

        if self._instrumented is not None:
            self._instrumented.record_frame()

    def commit_episode(self, episode_index: int) -> None:
        """Persist instrumentation readings for an episode being kept."""

        if self._instrumented is not None:
            self._instrumented.commit_episode(episode_index)

    def discard_episode(self) -> None:
        """Drop instrumentation readings for an episode being thrown away."""

        if self._instrumented is not None:
            self._instrumented.discard_episode()

    def close(self) -> None:
        """Release the instrumentation session."""

        if self._instrumented is not None:
            self._instrumented.close()
            self._instrumented = None

    def reset(self, *, confirm: Optional[bool] = None, unlock: bool = False) -> None:
        """Restore the workspace to the state this task starts from.

        The default asks the cell's environment what that takes. A BusyBox
        button springs back on its own, so nothing happens beyond returning the
        arm to rest; a switch left flipped is put back by a motion plan; a
        workspace nobody can restore automatically waits for an operator.

        Pass ``confirm=True`` to insist on a human regardless — worth doing when
        a job's own setup needs checking — or ``confirm=False`` to forbid one on
        an uninstrumented cell.

        Pass ``unlock=True`` after a command-safety trip (the edge rejected an
        unsafe goal and stopped accepting motion). The lock is released before
        the arm is returned to rest, so the same job can drive the next
        rollout. Other trips — a stalled joint, an over-current, a reset
        timeout — are not cleared this way.

        The job's ``disable_automated_reset`` argument forces the confirmation on
        for every reset, so a caller that passes nothing still gets an operator.

        An environment may temporarily stage configured rails for a reset motion
        plan. Before this method returns, each staged rail is restored directly
        to the position selected for this job's episodes.
        """

        if unlock:
            self._clear_safety_interlock()
        if self.disable_automated_reset and confirm is None:
            confirm = True
        if self._instrumented is not None:
            rail_staged = False
            staged_position: float | None = None

            def set_temporary_rail(pos: float) -> None:
                nonlocal rail_staged, staged_position
                first_staging_move = not rail_staged
                rail_staged = True
                staged_position = None
                target = max(0.0, min(1.0, float(pos)))
                self._set_rail_targets(
                    ((rail, target) for rail in self.rails),
                    # The environment may issue explicit retry jiggles after
                    # the first staging move. Those already guarantee travel;
                    # adding the short-move preload to each leg would turn a
                    # requested ±3% wiggle into a much larger excursion.
                    reset_approach=first_staging_move,
                )
                staged_position = target

            try:
                self._instrumented.reset_scene(
                    confirm=bool(confirm),
                    reset_cell=self._reset_cell,
                    set_rail=set_temporary_rail if self.rails else None,
                )
            except Exception as reset_error:
                if rail_staged:
                    restore_error: Exception | None = None
                    try:
                        self._restore_job_rails(staged_position)
                    except Exception as exc:
                        restore_error = exc
                    if restore_error is not None:
                        reset_error.add_note(
                            f"rail restoration also failed: {restore_error}"
                        )
                        # Keep the reset/motion failure as the raised exception
                        # while retaining the later safety-restore failure as
                        # its explicit cause. Clear the restore exception's
                        # implicit context to avoid a circular exception chain.
                        restore_error.__context__ = None
                        reset_error.__cause__ = restore_error
                        reset_error.__suppress_context__ = True
                raise
            if rail_staged:
                self._restore_job_rails(staged_position)
            return
        self._reset_cell(True if confirm is None else confirm)

    def set_rail(self, pos: float) -> None:
        """Move every configured rail to ``pos`` (0..1) and wait until settled.

        Use :meth:`RuntimeRail.set` instead to move one carriage of a bimanual
        cell, whose two rails may want different positions.
        """
        if not self.rails:
            raise RuntimeError("this cell has no rail calibration")
        if not self.robot_port or not _looks_like_connector_endpoint(self.robot_port):
            raise RuntimeError("robot connector is unavailable; cannot position rail")
        target = max(0.0, min(1.0, float(pos)))
        self._set_rail_targets((rail, target) for rail in self.rails)

    def _set_lightbox_brightness(
        self, lightbox: RuntimeLightbox, brightness: float
    ) -> None:
        """Back end for :meth:`RuntimeLightbox.set`."""
        if not self.robot_port or not _looks_like_connector_endpoint(self.robot_port):
            raise RuntimeError("robot connector is unavailable; cannot set the light")
        percent = max(0.0, min(100.0, float(brightness)))
        # The ESP takes an 8-bit duty, so a percentage has to land on one of 256
        # steps; rounding here means the value reported back is the one the
        # hardware was actually given.
        duty = round(percent * 255 / 100)
        response = _connector_request(
            self.robot_port,
            {
                "op": "lightbox_set",
                "url": lightbox.url,
                "b": duty,
                "f": lightbox.frequency,
                "fade": lightbox.fade,
            },
            read_timeout=_LIGHTBOX_READ_TIMEOUT_S,
        )
        if not response.get("ok"):
            raise RuntimeError(
                f"lightbox at {lightbox.url} could not be set to {percent:.0f}%: "
                f"{response.get('error', 'unknown error')}"
            )

    def _set_camera_mount(
        self, mount: RuntimeCameraMount, pan: float, tilt: float
    ) -> None:
        """Back end for :meth:`RuntimeCameraMount.set`."""
        if not self.robot_port or not _looks_like_connector_endpoint(self.robot_port):
            raise RuntimeError("robot connector is unavailable; cannot aim the mount")
        response = _connector_request(
            self.robot_port,
            {
                "op": "move_mount",
                "camera": mount.camera,
                "pan": float(pan),
                "tilt": float(tilt),
                "mount_config": mount.mount_config,
                # Aim without adopting: the calibrated pose stays the cell's, so
                # the next job-start reset returns the mount to where the cell
                # was commissioned rather than to wherever this job left it.
                "transient": True,
            },
            read_timeout=_MOUNT_READ_TIMEOUT_S,
        )
        if not response.get("ok"):
            raise RuntimeError(
                f"camera mount {mount.camera!r} could not be aimed to "
                f"pan={pan:.0f} tilt={tilt:.0f}: "
                f"{response.get('error', 'unknown error')}"
            )
        if not response.get("available", True):
            raise RuntimeError(
                f"camera mount {mount.camera!r} did not respond on its I2C bus; "
                "the scene did not change"
            )

    def _restore_job_rails(self, staged_position: float | None) -> None:
        """Restore rails moved for an environment plan to their episode poses."""
        targets = [
            (rail, rail.job_position)
            for rail in self.rails
            if staged_position is None
            or not math.isclose(staged_position, rail.job_position)
        ]
        if not targets:
            self._report_progress("Rail already at the job position; no restore needed")
            return
        for rail, target in targets:
            label = f" {rail.arm}" if rail.arm else ""
            self._report_progress(
                f"Restoring rail{label} to the job position ({target * 100:.0f}%)..."
            )
        if len(self.rails) == 1:
            self.set_rail(targets[0][1])
            return
        self._set_rail_targets(targets)

    def _set_rail_targets(
        self,
        targets: Iterable[tuple[RuntimeRail, float]],
        *,
        reset_approach: bool = False,
    ) -> None:
        """Move selected rails to per-rail targets using one polling path.

        ``reset_approach`` is private staging policy, not a general rail-control
        knob: a sensitive environment plan may request a fixed-side final leg,
        while ordinary job and variation moves retain their direct path.
        """
        if not self.robot_port or not _looks_like_connector_endpoint(self.robot_port):
            raise RuntimeError("robot connector is unavailable; cannot position rail")
        for rail, raw_target in targets:
            target = max(0.0, min(1.0, float(raw_target)))
            request: dict[str, Any] = {
                "op": "rail_set",
                "pos": target,
                "rail": {
                    "servo_id": rail.servo_id,
                    "total_rail_steps": rail.total_rail_steps,
                    "left_dir": rail.left_dir,
                    "home_end": rail.home_end,
                },
                # In-job reset: the cell already homed this rail once before the
                # container started, and nothing touches the carriage between the
                # arm's own moves within a job, so trust the recorded position
                # and skip the seek. The edge falls back to homing if that record
                # is not usable, so this never moves from a guess.
                "home": False,
            }
            if rail.arm is not None:
                request["arm"] = rail.arm
            if reset_approach:
                request["approach"] = {
                    "if_within": _RESET_RAIL_APPROACH_IF_WITHIN,
                    "distance": _RESET_RAIL_APPROACH_DISTANCE,
                    "side": "higher",
                }
            label = f" {rail.arm}" if rail.arm else ""
            response = _connector_request(
                self.robot_port,
                request,
                read_timeout=_RAIL_STATUS_READ_TIMEOUT_S,
            )
            if not response.get("ok"):
                raise RuntimeError(
                    f"rail{label} could not be set to {target * 100:.0f}%: "
                    f"{response.get('error', 'unknown error')}"
                )
            operation_id = response.get("operation_id")
            if operation_id is not None and not isinstance(operation_id, str):
                raise RuntimeError(
                    f"rail{label} returned an invalid operation id "
                    f"{operation_id!r}"
                )
            deadline = time.monotonic() + _RAIL_POLL_DEADLINE_S
            recovery_home_reported = False
            while True:
                status_request = {"op": "rail_status"}
                if operation_id is not None:
                    status_request["operation_id"] = operation_id
                status = _connector_request(
                    self.robot_port,
                    status_request,
                    read_timeout=_RAIL_STATUS_READ_TIMEOUT_S,
                )
                if not status.get("ok"):
                    raise RuntimeError(
                        f"rail{label} status failed: "
                        f"{status.get('error', 'unknown error')}"
                    )
                if (
                    operation_id is not None
                    and status.get("operation_id") != operation_id
                ):
                    raise RuntimeError(
                        f"rail{label} status was returned for the wrong operation"
                    )
                rail_status = status.get("rail") or {}
                state = rail_status.get("state")
                reason = rail_status.get("reason") or ""
                detail = rail_status.get("detail") or ""
                recovery_home = rail_status.get("recovery_home")
                if (
                    not recovery_home_reported
                    and isinstance(recovery_home, Mapping)
                ):
                    recovery_reason = str(
                        recovery_home.get("reason") or "unknown reason"
                    ).replace("_", " ")
                    recovery_end = str(recovery_home.get("end") or rail.home_end)
                    self._report_progress(
                        f"rail{label} position became unknown ({recovery_reason}); "
                        f"safety re-homing to the {recovery_end} end before continuing"
                    )
                    recovery_home_reported = True
                if state == "done":
                    break
                if state == "absent":
                    raise RuntimeError(
                        f"rail{label} servo is absent while positioning to "
                        f"{target * 100:.0f}%"
                    )
                if state in {"unreferenced", "error"}:
                    raise RuntimeError(
                        f"rail{label} position could not be established "
                        f"({state}: {reason or detail})"
                    )
                if time.monotonic() >= deadline:
                    raise RuntimeError(
                        f"rail{label} positioning to {target * 100:.0f}% timed out "
                        f"after {_RAIL_POLL_DEADLINE_S:.0f}s"
                    )
                time.sleep(_RAIL_POLL_INTERVAL_S)

    def _clear_safety_interlock(self) -> None:
        """Release a command-safety lock on the robot connector, if there is one."""

        if not self.robot_port or not _looks_like_connector_endpoint(self.robot_port):
            return
        response = _connector_request(self.robot_port, {"op": "clear_interlock"})
        if not response.get("ok"):
            raise RuntimeError(
                response.get("error", "could not clear the safety interlock")
            )

    def _reset_cell(self, confirm: bool = True) -> 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.

        ``confirm=False`` does step 1 and skips step 2, for a task where nothing
        in the workspace needs restoring. The arm still returns to rest — that
        is safety and a consistent start pose, not a workspace concern, so it is
        never skipped. This is the callback environments are handed so they can
        make that choice themselves.
        """

        self._report_progress(
            "Returning to rest..." if not confirm else "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"))

        if not confirm:
            return

        self._await_operator_reset()

    def confirm_scene_reset(self) -> None:
        """Ask the operator to stage the scene and confirm, moving nothing.

        :meth:`reset` always returns the arm to rest before any operator
        confirmation. A job that has already moved the arms to a
        policy-specific start pose wants the staging prompt alone: returning
        to rest would undo that pose, and ramping to it only after staging
        could sweep the arm through the objects the operator just placed.
        Call :meth:`reset` with ``confirm=False`` first, command the start
        pose, then call this.
        """

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

    def _await_operator_reset(self) -> None:
        # 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.

        Three judges, consulted in order of authority:

        1. A human. An operator hitting success or fail in the FMS during a
           live rollout ends the episode immediately with their verdict, which
           is how a dangerous rollout is stopped without stopping the job.
        2. The environment's instrumentation, which can see the goal directly —
           the switch is up, the buttons were pressed. It abstains when the
           goal is unmet or when nothing readable bears on it.
        3. The cell's automated completion monitor, a model watching the
           camera. Pass ``block=True`` for a final check that waits for the
           latest frames to be scored.

        ``status.scored_by`` names whichever decided. ``bool(status)`` is
        ``status.complete``.
        """
        status = self._query_completion(block=block)
        # Cache the latest outcome so rollout_end can record the authoritative,
        # cell-scored result for the leaderboard without trusting user input.
        self._last_completion = status
        return status

    def _query_completion(self, *, block: bool) -> CompletionStatus:
        reported = self._human_completion()
        if reported is not None:
            return CompletionStatus(True, reported, "operator")

        if self._instrumented is not None:
            status = self._instrumented.episode_status(final=block)
            if status.complete:
                return status

        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,
    ) -> list[str]:
        """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`.

        This also opens the instrumentation's scoring window and returns
        anything it finds already in the goal state — a switch the reset failed
        to put back. Such an episode still runs, but a verdict from the
        instrumentation would be unearned, so it abstains and the operator
        scores it. Most callers can ignore the return value; one that can offer
        the operator another go at the reset should use it.
        """
        # Reset the per-rollout completion cache so a stale outcome from the
        # previous rollout can't leak into this one's leaderboard record.
        self._last_completion = None
        problems = self._begin_instrumented_episode()
        for problem in problems:
            self._report_progress(f"bad reset: {problem}; the operator scores this episode")
        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)
        return problems

    def _begin_instrumented_episode(self) -> list[str]:
        if self._instrumented is None:
            return []
        return list(self._instrumented.begin_episode())

    def rollout_end(self, *, success: Optional[bool] = None, aborted: bool = False) -> None:
        """Tell the platform the current rollout has ended (hides FMS buttons).

        When leaderboard recording is active (see
        :meth:`Context.init_leaderboard`), this also records the rollout's
        outcome.

        Pass ``success`` when the eval runtime computed the episode's
        authoritative outcome itself — the common case being a BusyBox
        goal-state verdict, which is scored by the instrumented task box rather
        than reported back through :meth:`is_complete` (so it never reaches the
        cached completion). This is still an automated (box) or operator score,
        never a value the policy under test can supply. When omitted, the
        cell-scored result cached by the last :meth:`is_complete` call is
        recorded — an episode that never scored complete is a failure.

        Pass ``aborted=True`` when the rollout yielded no verdict at all, such as
        a camera dropping off the USB bus part-way through. The FMS buttons are
        hidden as usual but nothing is recorded: a hardware fault is not a policy
        failure, and scoring it as one would quietly drag down the number the
        leaderboard reports.
        """
        sink = self._rollout_outcome_sink
        if sink is not None and not aborted:
            if success is not None:
                outcome = CompletionStatus(complete=True, success=bool(success))
            else:
                outcome = self._last_completion or CompletionStatus(complete=False, success=False)
            try:
                sink(outcome)
            except Exception:  # noqa: BLE001 - recording must never break a rollout
                logger.warning("leaderboard rollout recording failed", exc_info=True)
        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_core.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(*, confirm: Optional[bool] = None, unlock: bool = False) -> None

Restore the workspace to the state this task starts from.

The default asks the cell's environment what that takes. A BusyBox button springs back on its own, so nothing happens beyond returning the arm to rest; a switch left flipped is put back by a motion plan; a workspace nobody can restore automatically waits for an operator.

Pass confirm=True to insist on a human regardless — worth doing when a job's own setup needs checking — or confirm=False to forbid one on an uninstrumented cell.

Pass unlock=True after a command-safety trip (the edge rejected an unsafe goal and stopped accepting motion). The lock is released before the arm is returned to rest, so the same job can drive the next rollout. Other trips — a stalled joint, an over-current, a reset timeout — are not cleared this way.

The job's disable_automated_reset argument forces the confirmation on for every reset, so a caller that passes nothing still gets an operator.

An environment may temporarily stage configured rails for a reset motion plan. Before this method returns, each staged rail is restored directly to the position selected for this job's episodes.

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

def reset(self, *, confirm: Optional[bool] = None, unlock: bool = False) -> None:
    """Restore the workspace to the state this task starts from.

    The default asks the cell's environment what that takes. A BusyBox
    button springs back on its own, so nothing happens beyond returning the
    arm to rest; a switch left flipped is put back by a motion plan; a
    workspace nobody can restore automatically waits for an operator.

    Pass ``confirm=True`` to insist on a human regardless — worth doing when
    a job's own setup needs checking — or ``confirm=False`` to forbid one on
    an uninstrumented cell.

    Pass ``unlock=True`` after a command-safety trip (the edge rejected an
    unsafe goal and stopped accepting motion). The lock is released before
    the arm is returned to rest, so the same job can drive the next
    rollout. Other trips — a stalled joint, an over-current, a reset
    timeout — are not cleared this way.

    The job's ``disable_automated_reset`` argument forces the confirmation on
    for every reset, so a caller that passes nothing still gets an operator.

    An environment may temporarily stage configured rails for a reset motion
    plan. Before this method returns, each staged rail is restored directly
    to the position selected for this job's episodes.
    """

    if unlock:
        self._clear_safety_interlock()
    if self.disable_automated_reset and confirm is None:
        confirm = True
    if self._instrumented is not None:
        rail_staged = False
        staged_position: float | None = None

        def set_temporary_rail(pos: float) -> None:
            nonlocal rail_staged, staged_position
            first_staging_move = not rail_staged
            rail_staged = True
            staged_position = None
            target = max(0.0, min(1.0, float(pos)))
            self._set_rail_targets(
                ((rail, target) for rail in self.rails),
                # The environment may issue explicit retry jiggles after
                # the first staging move. Those already guarantee travel;
                # adding the short-move preload to each leg would turn a
                # requested ±3% wiggle into a much larger excursion.
                reset_approach=first_staging_move,
            )
            staged_position = target

        try:
            self._instrumented.reset_scene(
                confirm=bool(confirm),
                reset_cell=self._reset_cell,
                set_rail=set_temporary_rail if self.rails else None,
            )
        except Exception as reset_error:
            if rail_staged:
                restore_error: Exception | None = None
                try:
                    self._restore_job_rails(staged_position)
                except Exception as exc:
                    restore_error = exc
                if restore_error is not None:
                    reset_error.add_note(
                        f"rail restoration also failed: {restore_error}"
                    )
                    # Keep the reset/motion failure as the raised exception
                    # while retaining the later safety-restore failure as
                    # its explicit cause. Clear the restore exception's
                    # implicit context to avoid a circular exception chain.
                    restore_error.__context__ = None
                    reset_error.__cause__ = restore_error
                    reset_error.__suppress_context__ = True
            raise
        if rail_staged:
            self._restore_job_rails(staged_position)
        return
    self._reset_cell(True if confirm is None else confirm)

is_complete ¶

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

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

Three judges, consulted in order of authority:

A human. An operator hitting success or fail in the FMS during a live rollout ends the episode immediately with their verdict, which is how a dangerous rollout is stopped without stopping the job.
The environment's instrumentation, which can see the goal directly — the switch is up, the buttons were pressed. It abstains when the goal is unmet or when nothing readable bears on it.
The cell's automated completion monitor, a model watching the camera. Pass block=True for a final check that waits for the latest frames to be scored.

status.scored_by names whichever decided. bool(status) is status.complete.

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.

    Three judges, consulted in order of authority:

    1. A human. An operator hitting success or fail in the FMS during a
       live rollout ends the episode immediately with their verdict, which
       is how a dangerous rollout is stopped without stopping the job.
    2. The environment's instrumentation, which can see the goal directly —
       the switch is up, the buttons were pressed. It abstains when the
       goal is unmet or when nothing readable bears on it.
    3. The cell's automated completion monitor, a model watching the
       camera. Pass ``block=True`` for a final check that waits for the
       latest frames to be scored.

    ``status.scored_by`` names whichever decided. ``bool(status)`` is
    ``status.complete``.
    """
    status = self._query_completion(block=block)
    # Cache the latest outcome so rollout_end can record the authoritative,
    # cell-scored result for the leaderboard without trusting user input.
    self._last_completion = status
    return status

rollout_begin ¶

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

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.

This also opens the instrumentation's scoring window and returns anything it finds already in the goal state — a switch the reset failed to put back. Such an episode still runs, but a verdict from the instrumentation would be unearned, so it abstains and the operator scores it. Most callers can ignore the return value; one that can offer the operator another go at the reset should use it.

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,
) -> list[str]:
    """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`.

    This also opens the instrumentation's scoring window and returns
    anything it finds already in the goal state — a switch the reset failed
    to put back. Such an episode still runs, but a verdict from the
    instrumentation would be unearned, so it abstains and the operator
    scores it. Most callers can ignore the return value; one that can offer
    the operator another go at the reset should use it.
    """
    # Reset the per-rollout completion cache so a stale outcome from the
    # previous rollout can't leak into this one's leaderboard record.
    self._last_completion = None
    problems = self._begin_instrumented_episode()
    for problem in problems:
        self._report_progress(f"bad reset: {problem}; the operator scores this episode")
    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)
    return problems

rollout_end ¶

rollout_end(*, success: Optional[bool] = None, aborted: bool = False) -> None

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

When leaderboard recording is active (see :meth:Context.init_leaderboard), this also records the rollout's outcome.

Pass success when the eval runtime computed the episode's authoritative outcome itself — the common case being a BusyBox goal-state verdict, which is scored by the instrumented task box rather than reported back through :meth:is_complete (so it never reaches the cached completion). This is still an automated (box) or operator score, never a value the policy under test can supply. When omitted, the cell-scored result cached by the last :meth:is_complete call is recorded — an episode that never scored complete is a failure.

Pass aborted=True when the rollout yielded no verdict at all, such as a camera dropping off the USB bus part-way through. The FMS buttons are hidden as usual but nothing is recorded: a hardware fault is not a policy failure, and scoring it as one would quietly drag down the number the leaderboard reports.

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

def rollout_end(self, *, success: Optional[bool] = None, aborted: bool = False) -> None:
    """Tell the platform the current rollout has ended (hides FMS buttons).

    When leaderboard recording is active (see
    :meth:`Context.init_leaderboard`), this also records the rollout's
    outcome.

    Pass ``success`` when the eval runtime computed the episode's
    authoritative outcome itself — the common case being a BusyBox
    goal-state verdict, which is scored by the instrumented task box rather
    than reported back through :meth:`is_complete` (so it never reaches the
    cached completion). This is still an automated (box) or operator score,
    never a value the policy under test can supply. When omitted, the
    cell-scored result cached by the last :meth:`is_complete` call is
    recorded — an episode that never scored complete is a failure.

    Pass ``aborted=True`` when the rollout yielded no verdict at all, such as
    a camera dropping off the USB bus part-way through. The FMS buttons are
    hidden as usual but nothing is recorded: a hardware fault is not a policy
    failure, and scoring it as one would quietly drag down the number the
    leaderboard reports.
    """
    sink = self._rollout_outcome_sink
    if sink is not None and not aborted:
        if success is not None:
            outcome = CompletionStatus(complete=True, success=bool(success))
        else:
            outcome = self._last_completion or CompletionStatus(complete=False, success=False)
        try:
            sink(outcome)
        except Exception:  # noqa: BLE001 - recording must never break a rollout
            logger.warning("leaderboard rollout recording failed", exc_info=True)
    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¶

Resetting between episodes¶

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` ¶