Skip to content

armnet-runtime API Reference

armnet_runtime

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

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

Typical use::

from armnet_runtime import main, Context

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

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

Embodiment module-attribute 

Embodiment = str

Task module-attribute 

Task = str

TerminalStatus module-attribute 

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

BIMANUAL_SO101_EMBODIMENT module-attribute 

BIMANUAL_SO101_EMBODIMENT = 'lerobot/bimanual_so101'

SO101_EMBODIMENT module-attribute 

SO101_EMBODIMENT = 'lerobot/so-101'

Job

Bases: BaseModel

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

Source code in core/src/armnet_core/models.py 
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
class Job(BaseModel):
    """The orchestrator's view of a job (response body of GET /jobs/{id})."""

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

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

id class-attribute instance-attribute 

id: str = Field(default_factory=_new_job_id)

spec instance-attribute 

spec: JobSpec

status class-attribute instance-attribute 

status: JobStatus = JobStatus.SUBMITTED

created_at class-attribute instance-attribute 

created_at: datetime = Field(default_factory=_utcnow)

updated_at class-attribute instance-attribute 

updated_at: datetime = Field(default_factory=_utcnow)

cell_id class-attribute instance-attribute 

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

result class-attribute instance-attribute 

result: Optional[JobResult] = None

dispatched class-attribute instance-attribute 

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

is_terminal 

is_terminal() -> bool
Source code in core/src/armnet_core/models.py 
584
585
def is_terminal(self) -> bool:
    return self.status in TerminalStatus

JobResult

Bases: BaseModel

Terminal result published by a cell on the results subject.

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

Source code in core/src/armnet_core/models.py 
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
class JobResult(BaseModel):
    """Terminal result published by a cell on the results subject.

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

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

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

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

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

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

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

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

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

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

status class-attribute instance-attribute 

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

exit_code class-attribute instance-attribute 

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

stdout class-attribute instance-attribute 

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

stderr class-attribute instance-attribute 

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

error class-attribute instance-attribute 

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

traceback class-attribute instance-attribute 

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

return_value class-attribute instance-attribute 

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

started_at class-attribute instance-attribute 

started_at: Optional[datetime] = None

finished_at class-attribute instance-attribute 

finished_at: Optional[datetime] = None

raise_for_status 

raise_for_status() -> None

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

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

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

For successful results this is a no-op.

Source code in core/src/armnet_core/models.py 
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
def raise_for_status(self) -> None:
    """Raise :class:`RemoteExecutionError` iff this result isn't ``SUCCEEDED``.

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

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

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

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

JobSpec

Bases: BaseModel

The fields a client supplies when creating a job.

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

Source code in core/src/armnet_core/models.py 
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
class JobSpec(BaseModel):
    """The fields a client supplies when creating a job.

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

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

image class-attribute instance-attribute 

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

args class-attribute instance-attribute 

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

embodiment class-attribute instance-attribute 

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

task class-attribute instance-attribute 

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

timeout_seconds class-attribute instance-attribute 

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

secrets class-attribute instance-attribute 

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

detach class-attribute instance-attribute 

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

username class-attribute instance-attribute 

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

JobStatus

Bases: str, Enum

Job lifecycle states.

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

Source code in core/src/armnet_core/models.py 
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
class JobStatus(str, Enum):
    """Job lifecycle states.

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

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

SUBMITTED class-attribute instance-attribute 

SUBMITTED = 'submitted'

QUEUED class-attribute instance-attribute 

QUEUED = 'queued'

DISPATCHED class-attribute instance-attribute 

DISPATCHED = 'dispatched'

RUNNING class-attribute instance-attribute 

RUNNING = 'running'

SUCCEEDED class-attribute instance-attribute 

SUCCEEDED = 'succeeded'

FAILED class-attribute instance-attribute 

FAILED = 'failed'

TIMEOUT class-attribute instance-attribute 

TIMEOUT = 'timeout'

CANCELLED class-attribute instance-attribute 

CANCELLED = 'cancelled'

RemoteExecutionError

Bases: RuntimeError

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

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

Source code in core/src/armnet_core/models.py 
546
547
548
549
550
551
552
553
554
555
556
557
558
class RemoteExecutionError(RuntimeError):
    """Raised by :meth:`JobResult.raise_for_status` for non-succeeded results.

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

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

result instance-attribute 

result = result

Cell dataclass

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

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

Source code in runtime/src/armnet_runtime/context.py 
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
@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.
    """

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

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

        if self._instrumented is not None:
            self._instrumented.reset_scene(
                confirm=bool(confirm),
                reset_cell=self._reset_cell,
            )
            return
        self._reset_cell(True if confirm is None else confirm)

    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

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

        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_runtime.markers import PROGRESS_MARKER
        print(f"{PROGRESS_MARKER} {message}", flush=True)
        time.sleep(0.01)

robot_port class-attribute instance-attribute 

robot_port: Optional[str] = None

Robot port value to pass into LeRobot robot configs.

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

robot_id class-attribute instance-attribute 

robot_id: Optional[str] = None

Stable robot id used by LeRobot to find calibration data.

cell_id class-attribute instance-attribute 

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 class-attribute instance-attribute 

calibration_dir: Optional[Path] = None

Calibration store path visible inside the customer container.

calibration_file_path class-attribute instance-attribute 

calibration_file_path: Optional[Path] = None

Exact LeRobot calibration file path visible inside the customer container.

language_instruction class-attribute instance-attribute 

language_instruction: Optional[str] = None

Task instruction provided by the cell.

local_control_endpoint class-attribute instance-attribute 

local_control_endpoint: Optional[str] = None

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

operator_call_endpoint class-attribute instance-attribute 

operator_call_endpoint: Optional[str] = None

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

is_local_container class-attribute instance-attribute 

is_local_container: bool = False

True when running a Docker image locally for development.

safety_limit class-attribute instance-attribute 

safety_limit: Optional[float] = None

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

arms class-attribute instance-attribute 

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

Named arms for bimanual/multi-arm cells.

environment class-attribute instance-attribute 

environment: Optional[str] = None

The kind of workcell this cell is set up as, e.g. "busybox".

task class-attribute instance-attribute 

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 class-attribute instance-attribute 

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.

is_bimanual property 

is_bimanual: bool

instrumentation property 

instrumentation: Optional[InstrumentedCell]

The live instrumentation session, or None if the cell has none.

arm 

arm(name: str) -> RuntimeArm
Source code in runtime/src/armnet_runtime/context.py 
223
224
225
226
227
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

get_robot_telemetry 

get_robot_telemetry(*, 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.

Source code in runtime/src/armnet_runtime/context.py 
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
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)

prepare_calibration_dir 

prepare_calibration_dir() -> 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.

Source code in runtime/src/armnet_runtime/context.py 
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
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

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 
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
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)

instrument 

instrument(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.

Source code in runtime/src/armnet_runtime/context.py 
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
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)

readings 

readings() -> 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.

Source code in runtime/src/armnet_runtime/context.py 
378
379
380
381
382
383
384
385
386
387
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()

attach_dataset 

attach_dataset(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.

Source code in runtime/src/armnet_runtime/context.py 
389
390
391
392
393
394
395
396
397
398
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)

record_frame 

record_frame() -> None

Record the instrumentation's view of the frame just captured.

Source code in runtime/src/armnet_runtime/context.py 
400
401
402
403
404
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()

commit_episode 

commit_episode(episode_index: int) -> None

Persist instrumentation readings for an episode being kept.

Source code in runtime/src/armnet_runtime/context.py 
406
407
408
409
410
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)

discard_episode 

discard_episode() -> None

Drop instrumentation readings for an episode being thrown away.

Source code in runtime/src/armnet_runtime/context.py 
412
413
414
415
416
def discard_episode(self) -> None:
    """Drop instrumentation readings for an episode being thrown away."""

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

close 

close() -> None

Release the instrumentation session.

Source code in runtime/src/armnet_runtime/context.py 
418
419
420
421
422
423
def close(self) -> None:
    """Release the instrumentation session."""

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

reset 

reset(*, confirm: Optional[bool] = None) -> 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.

Source code in runtime/src/armnet_runtime/context.py 
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
def reset(self, *, confirm: Optional[bool] = None) -> 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.
    """

    if self._instrumented is not None:
        self._instrumented.reset_scene(
            confirm=bool(confirm),
            reset_cell=self._reset_cell,
        )
        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:

  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.

Source code in runtime/src/armnet_runtime/context.py 
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
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 
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
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 
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
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")

is_shutting_down 

is_shutting_down() -> bool

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

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

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

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

Source code in runtime/src/armnet_runtime/context.py 
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
def is_shutting_down(self) -> bool:
    """Return True once the cell has entered the job's post-timeout grace window.

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

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

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

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

should_stop 

should_stop() -> bool

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

Source code in runtime/src/armnet_runtime/context.py 
700
701
702
703
704
705
706
707
708
709
def should_stop(self) -> bool:
    """Return True when local/remote control asks user code to stop safely."""

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

get_teleop_action 

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

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

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

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

Source code in runtime/src/armnet_runtime/context.py 
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
def get_teleop_action(self) -> Optional[dict[str, float]]:
    """Return the freshest remote-teleoperation action for this job, or None.

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

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

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

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

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

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

get_teleop_event 

get_teleop_event() -> Optional[str]

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

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

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

Source code in runtime/src/armnet_runtime/context.py 
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
def get_teleop_event(self) -> Optional[str]:
    """Return the next pending recording-control event, or None.

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

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

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

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

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

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

CompletionStatus

Bases: NamedTuple

Whether an episode has finished, how it went, and who decided.

bool(status) is status.complete, so a caller that only cares whether to keep going can use it directly.

scored_by names the judge — an environment's own name when its instrumentation decided, "operator" for a human verdict, "monitor" for the automated completion model. Recorded with results so a success rate can be read knowing what produced it.

Source code in core/src/armnet_core/environment.py 
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
class CompletionStatus(NamedTuple):
    """Whether an episode has finished, how it went, and who decided.

    ``bool(status)`` is ``status.complete``, so a caller that only cares
    whether to keep going can use it directly.

    ``scored_by`` names the judge — an environment's own name when its
    instrumentation decided, ``"operator"`` for a human verdict, ``"monitor"``
    for the automated completion model. Recorded with results so a success rate
    can be read knowing what produced it.
    """

    complete: bool
    success: bool
    scored_by: Optional[str] = None

    def __bool__(self) -> bool:
        return self.complete

complete instance-attribute 

complete: bool

success instance-attribute 

success: bool

scored_by class-attribute instance-attribute 

scored_by: Optional[str] = None

Context dataclass

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

Source code in runtime/src/armnet_runtime/context.py 
 955
 956
 957
 958
 959
 960
 961
 962
 963
 964
 965
 966
 967
 968
 969
 970
 971
 972
 973
 974
 975
 976
 977
 978
 979
 980
 981
 982
 983
 984
 985
 986
 987
 988
 989
 990
 991
 992
 993
 994
 995
 996
 997
 998
 999
1000
1001
1002
1003
1004
1005
1006
1007
1008
1009
1010
1011
1012
1013
1014
1015
1016
1017
1018
1019
1020
1021
1022
1023
1024
1025
1026
1027
1028
1029
1030
1031
1032
1033
1034
1035
1036
1037
1038
1039
1040
1041
1042
1043
1044
1045
1046
1047
1048
1049
1050
1051
1052
1053
1054
1055
1056
1057
1058
1059
1060
1061
1062
1063
1064
1065
1066
1067
1068
1069
1070
1071
1072
1073
1074
1075
1076
1077
1078
1079
1080
1081
1082
1083
1084
1085
1086
1087
1088
1089
1090
1091
1092
1093
1094
1095
1096
1097
1098
1099
1100
1101
1102
1103
1104
1105
1106
1107
1108
1109
1110
1111
1112
1113
1114
1115
1116
1117
1118
1119
1120
1121
1122
1123
1124
1125
1126
1127
1128
1129
1130
1131
1132
1133
1134
1135
1136
1137
1138
1139
1140
1141
1142
1143
1144
1145
1146
1147
1148
1149
1150
1151
1152
1153
1154
1155
1156
1157
1158
1159
1160
1161
1162
1163
1164
1165
1166
1167
1168
1169
1170
1171
1172
1173
1174
@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_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 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,
        )

job_id instance-attribute 

job_id: str

embodiment instance-attribute 

embodiment: Embodiment

task instance-attribute 

task: Task

args class-attribute instance-attribute 

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

cell class-attribute instance-attribute 

cell: Cell = field(default_factory=Cell)

camera_configs class-attribute instance-attribute 

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

cache_home class-attribute instance-attribute 

cache_home: Optional[Path] = None

volume class-attribute instance-attribute 

volume: Volume = field(default_factory=Volume)

secrets class-attribute instance-attribute 

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

timeout_seconds class-attribute instance-attribute 

timeout_seconds: Optional[int] = None

report_progress 

report_progress(message: str) -> None

Surface a progress message back to the platform.

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

Source code in runtime/src/armnet_runtime/context.py 
975
976
977
978
979
980
981
982
983
984
985
986
987
988
def report_progress(self, message: str) -> None:
    """Surface a progress message back to the platform.

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

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

is_shutting_down 

is_shutting_down() -> bool

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

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

Source code in runtime/src/armnet_runtime/context.py 
990
991
992
993
994
995
996
997
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()

get_robot_telemetry 

get_robot_telemetry(*, arm: str | None = None) -> dict[str, Any]

Return the latest cached edge telemetry snapshot for an arm.

Source code in runtime/src/armnet_runtime/context.py 
 999
1000
1001
1002
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)

init_leaderboard 

init_leaderboard(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.

Source code in runtime/src/armnet_runtime/context.py 
1004
1005
1006
1007
1008
1009
1010
1011
1012
1013
1014
1015
1016
1017
1018
1019
1020
1021
1022
1023
1024
1025
1026
1027
1028
1029
1030
1031
1032
1033
1034
1035
1036
1037
1038
1039
1040
1041
1042
1043
1044
1045
1046
1047
1048
1049
1050
1051
1052
1053
1054
1055
1056
1057
1058
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,
    )

submit_results_to_leaderboard 

submit_results_to_leaderboard() -> 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.

Source code in runtime/src/armnet_runtime/context.py 
1065
1066
1067
1068
1069
1070
1071
1072
1073
1074
1075
1076
1077
1078
1079
1080
1081
1082
1083
1084
1085
1086
1087
1088
1089
1090
1091
1092
1093
1094
1095
1096
1097
1098
1099
1100
1101
1102
1103
1104
1105
1106
1107
1108
1109
1110
1111
1112
1113
1114
1115
1116
1117
1118
1119
1120
1121
1122
1123
1124
1125
1126
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"],
    }

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 
1128
1129
1130
1131
1132
1133
1134
1135
1136
1137
1138
1139
1140
1141
1142
1143
1144
1145
1146
1147
1148
1149
1150
1151
1152
1153
1154
1155
1156
1157
1158
1159
1160
1161
1162
1163
1164
1165
1166
1167
1168
1169
1170
1171
1172
1173
1174
def log_rerun_data(
    self,
    observation: dict[str, Any] | None = None,
    action: dict[str, Any] | None = None,
    *,
    compress_images: bool = True,
    jpeg_quality: int = 75,
) -> None:
    """Stream observation/action data to a Rerun viewer on the client.

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

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

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

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

    if not observation and not action:
        return

    from armnet_runtime.rerun import RerunStreamer

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

ResetTimeoutException

Bases: RuntimeError

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

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

try:
    ctx.cell.reset()
except ResetTimeoutException:
    dataset.push_to_hub()  # save what we collected
    raise
Source code in runtime/src/armnet_runtime/context.py 
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
class ResetTimeoutException(RuntimeError):
    """Raised by ``ctx.cell.reset()`` when a manual reset is not actioned in time.

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

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

MainRegistrationError

Bases: RuntimeError

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

Source code in runtime/src/armnet_runtime/decorator.py 
31
32
class MainRegistrationError(RuntimeError):
    """Raised when ``@main`` is used incorrectly (multiple times, etc.)."""

require_so101_embodiment 

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

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

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

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

Source code in runtime/src/armnet_runtime/context.py 
1181
1182
1183
1184
1185
1186
1187
1188
1189
1190
1191
1192
1193
1194
1195
1196
1197
1198
1199
1200
1201
1202
1203
1204
1205
1206
def require_so101_embodiment(ctx: "Context", runtime_name: str) -> bool:
    """Validate the job's embodiment is a (single or bimanual) SO-101.

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

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

main 

main(fn: EntryPoint) -> EntryPoint

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

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

Source code in runtime/src/armnet_runtime/decorator.py 
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
def main(fn: EntryPoint) -> EntryPoint:
    """Decorator: mark ``fn`` as the script's entry point.

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

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

armnet_runtime.context

Job context surfaced to @main-decorated functions.

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

logger module-attribute 

logger = logging.getLogger(__name__)

SO101_EMBODIMENT module-attribute 

SO101_EMBODIMENT = 'lerobot/so-101'

BIMANUAL_SO101_EMBODIMENT module-attribute 

BIMANUAL_SO101_EMBODIMENT = 'lerobot/bimanual_so101'

ResetTimeoutException

Bases: RuntimeError

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

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

try:
    ctx.cell.reset()
except ResetTimeoutException:
    dataset.push_to_hub()  # save what we collected
    raise
Source code in runtime/src/armnet_runtime/context.py 
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
class ResetTimeoutException(RuntimeError):
    """Raised by ``ctx.cell.reset()`` when a manual reset is not actioned in time.

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

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

Volume dataclass

User volume mounted into the runtime container.

Source code in runtime/src/armnet_runtime/context.py 
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
@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 
77
78
79
80
81
82
83
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 
85
86
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 
88
89
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 
91
92
93
94
95
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 
 97
 98
 99
100
101
def write_text(self, relative_path: str | Path, data: str) -> Path:
    path = self.path(relative_path)
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(data)
    return path

RuntimeArm dataclass

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

Source code in runtime/src/armnet_runtime/context.py 
104
105
106
107
108
109
110
111
112
113
@dataclass(frozen=True)
class RuntimeArm:
    """Runtime-facing handle for one named robot arm in a multi-arm cell."""

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

name instance-attribute 

name: str

robot_port instance-attribute 

robot_port: str

robot_id class-attribute instance-attribute 

robot_id: Optional[str] = None

calibration_dir class-attribute instance-attribute 

calibration_dir: Optional[Path] = None

calibration_file_path class-attribute instance-attribute 

calibration_file_path: Optional[Path] = None

safety_limit class-attribute instance-attribute 

safety_limit: Optional[float] = None

BimanualCalibrationLayout dataclass

Temporary calibration layout matching LeRobot's bimanual id convention.

Source code in runtime/src/armnet_runtime/context.py 
116
117
118
119
120
121
@dataclass(frozen=True)
class BimanualCalibrationLayout:
    """Temporary calibration layout matching LeRobot's bimanual id convention."""

    robot_id: str
    calibration_dir: Path

robot_id instance-attribute 

robot_id: str

calibration_dir instance-attribute 

calibration_dir: Path

Cell dataclass

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

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

Source code in runtime/src/armnet_runtime/context.py 
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
@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.
    """

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

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

        if self._instrumented is not None:
            self._instrumented.reset_scene(
                confirm=bool(confirm),
                reset_cell=self._reset_cell,
            )
            return
        self._reset_cell(True if confirm is None else confirm)

    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

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

        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_runtime.markers import PROGRESS_MARKER
        print(f"{PROGRESS_MARKER} {message}", flush=True)
        time.sleep(0.01)

robot_port class-attribute instance-attribute 

robot_port: Optional[str] = None

Robot port value to pass into LeRobot robot configs.

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

robot_id class-attribute instance-attribute 

robot_id: Optional[str] = None

Stable robot id used by LeRobot to find calibration data.

cell_id class-attribute instance-attribute 

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 class-attribute instance-attribute 

calibration_dir: Optional[Path] = None

Calibration store path visible inside the customer container.

calibration_file_path class-attribute instance-attribute 

calibration_file_path: Optional[Path] = None

Exact LeRobot calibration file path visible inside the customer container.

language_instruction class-attribute instance-attribute 

language_instruction: Optional[str] = None

Task instruction provided by the cell.

local_control_endpoint class-attribute instance-attribute 

local_control_endpoint: Optional[str] = None

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

operator_call_endpoint class-attribute instance-attribute 

operator_call_endpoint: Optional[str] = None

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

is_local_container class-attribute instance-attribute 

is_local_container: bool = False

True when running a Docker image locally for development.

safety_limit class-attribute instance-attribute 

safety_limit: Optional[float] = None

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

arms class-attribute instance-attribute 

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

Named arms for bimanual/multi-arm cells.

environment class-attribute instance-attribute 

environment: Optional[str] = None

The kind of workcell this cell is set up as, e.g. "busybox".

task class-attribute instance-attribute 

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 class-attribute instance-attribute 

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.

is_bimanual property 

is_bimanual: bool

instrumentation property 

instrumentation: Optional[InstrumentedCell]

The live instrumentation session, or None if the cell has none.

arm 

arm(name: str) -> RuntimeArm
Source code in runtime/src/armnet_runtime/context.py 
223
224
225
226
227
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

get_robot_telemetry 

get_robot_telemetry(*, 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.

Source code in runtime/src/armnet_runtime/context.py 
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
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)

prepare_calibration_dir 

prepare_calibration_dir() -> 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.

Source code in runtime/src/armnet_runtime/context.py 
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
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

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 
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
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)

instrument 

instrument(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.

Source code in runtime/src/armnet_runtime/context.py 
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
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)

readings 

readings() -> 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.

Source code in runtime/src/armnet_runtime/context.py 
378
379
380
381
382
383
384
385
386
387
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()

attach_dataset 

attach_dataset(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.

Source code in runtime/src/armnet_runtime/context.py 
389
390
391
392
393
394
395
396
397
398
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)

record_frame 

record_frame() -> None

Record the instrumentation's view of the frame just captured.

Source code in runtime/src/armnet_runtime/context.py 
400
401
402
403
404
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()

commit_episode 

commit_episode(episode_index: int) -> None

Persist instrumentation readings for an episode being kept.

Source code in runtime/src/armnet_runtime/context.py 
406
407
408
409
410
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)

discard_episode 

discard_episode() -> None

Drop instrumentation readings for an episode being thrown away.

Source code in runtime/src/armnet_runtime/context.py 
412
413
414
415
416
def discard_episode(self) -> None:
    """Drop instrumentation readings for an episode being thrown away."""

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

close 

close() -> None

Release the instrumentation session.

Source code in runtime/src/armnet_runtime/context.py 
418
419
420
421
422
423
def close(self) -> None:
    """Release the instrumentation session."""

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

reset 

reset(*, confirm: Optional[bool] = None) -> 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.

Source code in runtime/src/armnet_runtime/context.py 
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
def reset(self, *, confirm: Optional[bool] = None) -> 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.
    """

    if self._instrumented is not None:
        self._instrumented.reset_scene(
            confirm=bool(confirm),
            reset_cell=self._reset_cell,
        )
        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:

  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.

Source code in runtime/src/armnet_runtime/context.py 
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
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 
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
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 
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
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")

is_shutting_down 

is_shutting_down() -> bool

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

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

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

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

Source code in runtime/src/armnet_runtime/context.py 
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
def is_shutting_down(self) -> bool:
    """Return True once the cell has entered the job's post-timeout grace window.

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

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

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

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

should_stop 

should_stop() -> bool

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

Source code in runtime/src/armnet_runtime/context.py 
700
701
702
703
704
705
706
707
708
709
def should_stop(self) -> bool:
    """Return True when local/remote control asks user code to stop safely."""

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

get_teleop_action 

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

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

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

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

Source code in runtime/src/armnet_runtime/context.py 
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
def get_teleop_action(self) -> Optional[dict[str, float]]:
    """Return the freshest remote-teleoperation action for this job, or None.

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

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

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

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

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

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

get_teleop_event 

get_teleop_event() -> Optional[str]

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

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

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

Source code in runtime/src/armnet_runtime/context.py 
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
def get_teleop_event(self) -> Optional[str]:
    """Return the next pending recording-control event, or None.

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

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

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

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

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

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

Context dataclass

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

Source code in runtime/src/armnet_runtime/context.py 
 955
 956
 957
 958
 959
 960
 961
 962
 963
 964
 965
 966
 967
 968
 969
 970
 971
 972
 973
 974
 975
 976
 977
 978
 979
 980
 981
 982
 983
 984
 985
 986
 987
 988
 989
 990
 991
 992
 993
 994
 995
 996
 997
 998
 999
1000
1001
1002
1003
1004
1005
1006
1007
1008
1009
1010
1011
1012
1013
1014
1015
1016
1017
1018
1019
1020
1021
1022
1023
1024
1025
1026
1027
1028
1029
1030
1031
1032
1033
1034
1035
1036
1037
1038
1039
1040
1041
1042
1043
1044
1045
1046
1047
1048
1049
1050
1051
1052
1053
1054
1055
1056
1057
1058
1059
1060
1061
1062
1063
1064
1065
1066
1067
1068
1069
1070
1071
1072
1073
1074
1075
1076
1077
1078
1079
1080
1081
1082
1083
1084
1085
1086
1087
1088
1089
1090
1091
1092
1093
1094
1095
1096
1097
1098
1099
1100
1101
1102
1103
1104
1105
1106
1107
1108
1109
1110
1111
1112
1113
1114
1115
1116
1117
1118
1119
1120
1121
1122
1123
1124
1125
1126
1127
1128
1129
1130
1131
1132
1133
1134
1135
1136
1137
1138
1139
1140
1141
1142
1143
1144
1145
1146
1147
1148
1149
1150
1151
1152
1153
1154
1155
1156
1157
1158
1159
1160
1161
1162
1163
1164
1165
1166
1167
1168
1169
1170
1171
1172
1173
1174
@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_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 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,
        )

job_id instance-attribute 

job_id: str

embodiment instance-attribute 

embodiment: Embodiment

task instance-attribute 

task: Task

args class-attribute instance-attribute 

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

cell class-attribute instance-attribute 

cell: Cell = field(default_factory=Cell)

camera_configs class-attribute instance-attribute 

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

cache_home class-attribute instance-attribute 

cache_home: Optional[Path] = None

volume class-attribute instance-attribute 

volume: Volume = field(default_factory=Volume)

secrets class-attribute instance-attribute 

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

timeout_seconds class-attribute instance-attribute 

timeout_seconds: Optional[int] = None

report_progress 

report_progress(message: str) -> None

Surface a progress message back to the platform.

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

Source code in runtime/src/armnet_runtime/context.py 
975
976
977
978
979
980
981
982
983
984
985
986
987
988
def report_progress(self, message: str) -> None:
    """Surface a progress message back to the platform.

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

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

is_shutting_down 

is_shutting_down() -> bool

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

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

Source code in runtime/src/armnet_runtime/context.py 
990
991
992
993
994
995
996
997
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()

get_robot_telemetry 

get_robot_telemetry(*, arm: str | None = None) -> dict[str, Any]

Return the latest cached edge telemetry snapshot for an arm.

Source code in runtime/src/armnet_runtime/context.py 
 999
1000
1001
1002
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)

init_leaderboard 

init_leaderboard(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.

Source code in runtime/src/armnet_runtime/context.py 
1004
1005
1006
1007
1008
1009
1010
1011
1012
1013
1014
1015
1016
1017
1018
1019
1020
1021
1022
1023
1024
1025
1026
1027
1028
1029
1030
1031
1032
1033
1034
1035
1036
1037
1038
1039
1040
1041
1042
1043
1044
1045
1046
1047
1048
1049
1050
1051
1052
1053
1054
1055
1056
1057
1058
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,
    )

submit_results_to_leaderboard 

submit_results_to_leaderboard() -> 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.

Source code in runtime/src/armnet_runtime/context.py 
1065
1066
1067
1068
1069
1070
1071
1072
1073
1074
1075
1076
1077
1078
1079
1080
1081
1082
1083
1084
1085
1086
1087
1088
1089
1090
1091
1092
1093
1094
1095
1096
1097
1098
1099
1100
1101
1102
1103
1104
1105
1106
1107
1108
1109
1110
1111
1112
1113
1114
1115
1116
1117
1118
1119
1120
1121
1122
1123
1124
1125
1126
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"],
    }

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 
1128
1129
1130
1131
1132
1133
1134
1135
1136
1137
1138
1139
1140
1141
1142
1143
1144
1145
1146
1147
1148
1149
1150
1151
1152
1153
1154
1155
1156
1157
1158
1159
1160
1161
1162
1163
1164
1165
1166
1167
1168
1169
1170
1171
1172
1173
1174
def log_rerun_data(
    self,
    observation: dict[str, Any] | None = None,
    action: dict[str, Any] | None = None,
    *,
    compress_images: bool = True,
    jpeg_quality: int = 75,
) -> None:
    """Stream observation/action data to a Rerun viewer on the client.

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

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

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

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

    if not observation and not action:
        return

    from armnet_runtime.rerun import RerunStreamer

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

require_so101_embodiment 

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

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

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

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

Source code in runtime/src/armnet_runtime/context.py 
1181
1182
1183
1184
1185
1186
1187
1188
1189
1190
1191
1192
1193
1194
1195
1196
1197
1198
1199
1200
1201
1202
1203
1204
1205
1206
def require_so101_embodiment(ctx: "Context", runtime_name: str) -> bool:
    """Validate the job's embodiment is a (single or bimanual) SO-101.

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

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

build_context 

build_context() -> Context

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

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

Source code in runtime/src/armnet_runtime/context.py 
1209
1210
1211
1212
1213
1214
1215
1216
1217
1218
1219
1220
1221
1222
1223
1224
1225
1226
1227
1228
1229
1230
1231
1232
1233
1234
1235
1236
1237
1238
1239
1240
1241
1242
1243
1244
1245
1246
1247
1248
1249
1250
1251
1252
1253
1254
1255
1256
1257
1258
1259
1260
1261
1262
1263
1264
1265
1266
1267
1268
1269
1270
1271
1272
1273
1274
1275
1276
1277
1278
1279
1280
1281
1282
1283
1284
1285
1286
1287
1288
def build_context() -> Context:
    """Read the cell-injected env vars and construct a :class:`Context`.

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

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

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

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

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

armnet_runtime.decorator

The @main decorator and its module-level registry.

Customer code looks like::

from armnet_runtime import main, Context

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

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

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

EntryPoint module-attribute 

EntryPoint = Callable[..., Any]

MainRegistrationError

Bases: RuntimeError

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

Source code in runtime/src/armnet_runtime/decorator.py 
31
32
class MainRegistrationError(RuntimeError):
    """Raised when ``@main`` is used incorrectly (multiple times, etc.)."""

main 

main(fn: EntryPoint) -> EntryPoint

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

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

Source code in runtime/src/armnet_runtime/decorator.py 
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
def main(fn: EntryPoint) -> EntryPoint:
    """Decorator: mark ``fn`` as the script's entry point.

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

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

registered_main 

registered_main() -> Optional[EntryPoint]

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

Source code in runtime/src/armnet_runtime/decorator.py 
56
57
58
59
def registered_main() -> Optional[EntryPoint]:
    """Return the function previously registered with ``@main``, if any."""

    return _registered

armnet_runtime.cli

armnet-runtime console script.

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

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

and this script:

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

EXIT_OK module-attribute 

EXIT_OK = 0

EXIT_USER_RAISED module-attribute 

EXIT_USER_RAISED = 1

EXIT_USAGE module-attribute 

EXIT_USAGE = 2

EXIT_NO_MAIN module-attribute 

EXIT_NO_MAIN = 3

EXIT_USER_IMPORT_FAILED module-attribute 

EXIT_USER_IMPORT_FAILED = 4

EXIT_BAD_CONTEXT module-attribute 

EXIT_BAD_CONTEXT = 5

run 

run(argv: Sequence[str] | None = None) -> int
Source code in runtime/src/armnet_runtime/cli.py 
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
def run(argv: Sequence[str] | None = None) -> int:
    args = _parse_args(sys.argv[1:] if argv is None else list(argv))
    _install_timestamped_streams()

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

    install_lerobot_remote_mode()

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

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

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

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

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

armnet_runtime.env

Env var keys the cell injects into customer containers.

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

JOB_ID module-attribute 

JOB_ID = 'ARMNET_JOB_ID'

EMBODIMENT module-attribute 

EMBODIMENT = 'ARMNET_EMBODIMENT'

TASK module-attribute 

TASK = 'ARMNET_TASK'

ENVIRONMENT module-attribute 

ENVIRONMENT = 'ARMNET_ENVIRONMENT'

TIMEOUT_SECONDS module-attribute 

TIMEOUT_SECONDS = 'ARMNET_TIMEOUT_SECONDS'

ARGS module-attribute 

ARGS = 'ARMNET_ARGS'

CELL_CONFIG module-attribute 

CELL_CONFIG = 'ARMNET_CELL_CONFIG'

CELL_SOCKET module-attribute 

CELL_SOCKET = 'ARMNET_CELL_SOCKET'

OPERATOR_CALL_ENDPOINT module-attribute 

OPERATOR_CALL_ENDPOINT = 'ARMNET_OPERATOR_CALL_ENDPOINT'

LOCAL_CONTROL_ENDPOINT module-attribute 

LOCAL_CONTROL_ENDPOINT = 'ARMNET_LOCAL_CONTROL_ENDPOINT'

LOCAL_CONTAINER module-attribute 

LOCAL_CONTAINER = 'ARMNET_LOCAL_CONTAINER'

ROBOT_ID module-attribute 

ROBOT_ID = 'ARMNET_ROBOT_ID'

CALIBRATION_DIR module-attribute 

CALIBRATION_DIR = 'ARMNET_CALIBRATION_DIR'

CALIBRATION_FILE_PATH module-attribute 

CALIBRATION_FILE_PATH = 'ARMNET_CALIBRATION_FILE_PATH'

CACHE_HOME module-attribute 

CACHE_HOME = 'REMOTEROBOT_CACHE_HOME'

VOLUME_HOME module-attribute 

VOLUME_HOME = 'REMOTEROBOT_VOLUME_HOME'

HF_HOME module-attribute 

HF_HOME = 'HF_HOME'

SECRETS module-attribute 

SECRETS = 'ARMNET_SECRETS'

armnet_runtime.lerobot

LeRobot integration helpers for armnet-runtime.

RemoteARX5Arm

Proxy for arx5_common.ARX5Arm.

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

Source code in runtime/src/armnet_runtime/lerobot/remote_arx5.py 
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
class RemoteARX5Arm:
    """Proxy for `arx5_common.ARX5Arm`.

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

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

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

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

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

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

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

        return method

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

RemoteOpenCVCamera

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

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

Source code in runtime/src/armnet_runtime/lerobot/remote_camera.py 
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
class RemoteOpenCVCamera:
    """Drop-in-ish proxy for LeRobot's OpenCVCamera.

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

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

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

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

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

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

    def _method_proxy(self, name: str):
        def method(*args: Any, **kwargs: Any) -> Any:
            response = self._request({
                "op": "call",
                "object_id": self._object_id,
                "method": name,
                "args": jsonable(args),
                "kwargs": jsonable(kwargs),
            })
            # Track connection state so is_connected can be served locally.
            if name == "connect":
                self._set_connected(True)
            elif name == "disconnect":
                self._set_connected(False)
            if response.get("context_manager"):
                return RemoteContextManager(
                    conn=self._conn,
                    object_id=response["object_id"],
                )
            value = response.get("value")
            # Only edges new enough to stamp their frames report an age; older
            # ones simply omit it.
            if isinstance(value, dict) and "age_ms" in value:
                CALL_STATS.record_frame_age(self._conn.label, float(value["age_ms"]))
            return decode(value)

        return method

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

RemoteFeetechMotorsBus

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

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

Source code in runtime/src/armnet_runtime/lerobot/remote_feetech.py 
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
class RemoteFeetechMotorsBus:
    """Drop-in-ish proxy for LeRobot's FeetechMotorsBus.

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

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

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

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

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

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

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

        return method

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

install_lerobot_remote_mode 

install_lerobot_remote_mode() -> None

Replace known LeRobot Feetech bus classes with remote proxies.

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

Source code in runtime/src/armnet_runtime/lerobot/import_swap.py 
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
def install_lerobot_remote_mode() -> None:
    """Replace known LeRobot Feetech bus classes with remote proxies.

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

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

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

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

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

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

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